func (s *Store) WriteDataURI(bidID auction.BidID, payloadCid, uri string, size uint64) (string, error) {
	duri, err := datauri.NewURI(payloadCid, uri)
	if err != nil {
		return "", fmt.Errorf("parsing data uri: %w", err)
	}
	f, err := os.Create(s.dealDataFilePathFor(bidID, payloadCid))
	if err != nil {
		return "", fmt.Errorf("opening file for deal data: %v", err)
	}
	defer func() {
		if err := f.Close(); err != nil {
			log.Errorf("closing data file: %v", err)
		}
	}()
	if _, err := f.Seek(0, 0); err != nil {
		return "", fmt.Errorf("seeking file to the beginning: %v", err)
	}
	log.Debugf("fetching %s with timeout of %v", uri, s.dealDataFetchTimeout)
	ctx, cancel := context.WithTimeout(s.ctx, s.dealDataFetchTimeout)
	defer cancel()
	if err := duri.Write(ctx, f); err != nil {
		return "", fmt.Errorf("writing data uri %s: %w", uri, err)
	}
	return f.Name(), nil
}
WriteDataURI writes the uri resource to the configured deal data directory.
package store
import (
	"bytes"
	"context"
	"encoding/gob"
	"errors"
	"fmt"
	"io/ioutil"
	"os"
	"path"
	"path/filepath"
	"strings"
	"sync"
	"time"
	"github.com/ipfs/go-cid"
	ds "github.com/ipfs/go-datastore"
	dsq "github.com/ipfs/go-datastore/query"
	"github.com/libp2p/go-libp2p-core/peer"
	"github.com/oklog/ulid/v2"
	"github.com/textileio/bidbot/lib/auction"
	"github.com/textileio/bidbot/lib/datauri"
	"github.com/textileio/bidbot/lib/dshelper/txndswrap"
	"github.com/textileio/bidbot/lib/filclient"
	"github.com/textileio/bidbot/service/limiter"
	"github.com/textileio/bidbot/service/lotusclient"
	dsextensions "github.com/textileio/go-datastore-extensions"
	golog "github.com/textileio/go-log/v2"
	"golang.org/x/sync/semaphore"
)
const (
	gcInterval = time.Hour
	defaultListLimit = 10
	maxListLimit = 1000
)
var (
	log = golog.Logger("bidbot/store")
	DataURIFetchStartDelay = time.Second * 10
	MaxDataURIFetchConcurrency = 3
	ErrBidNotFound = errors.New("bid not found")
	dsPrefix = ds.NewKey("/bids")
	dsQueuedPrefix = ds.NewKey("/data_queued")
	dsFetchingPrefix = ds.NewKey("/data_fetching")
)
type ProgressReporter interface {
	StartFetching(bidID auction.BidID, attempts uint32)
	ErrorFetching(bidID auction.BidID, attempts uint32, err error)
	StartImporting(bidID auction.BidID, attempts uint32)
	EndImporting(bidID auction.BidID, attempts uint32, err error)
	Finalized(bidID auction.BidID)
	Errored(bidID auction.BidID, errrorCause string)
}
type nullProgressReporter struct{}
func (pr nullProgressReporter) StartFetching(bidID auction.BidID, attempts uint32)            {}
func (pr nullProgressReporter) ErrorFetching(bidID auction.BidID, attempts uint32, err error) {}
func (pr nullProgressReporter) StartImporting(bidID auction.BidID, attempts uint32)           {}
func (pr nullProgressReporter) EndImporting(bidID auction.BidID, attempts uint32, err error)  {}
func (pr nullProgressReporter) Finalized(bidID auction.BidID)                                 {}
func (pr nullProgressReporter) Errored(bidID auction.BidID, errorCause string)                {}
type Bid struct {
	ID                   auction.BidID
	AuctionID            auction.ID
	AuctioneerID         peer.ID
	PayloadCid           cid.Cid
	DealSize             uint64
	DealDuration         uint64
	ClientAddress        string
	Sources              auction.Sources
	Status               BidStatus
	AskPrice             int64  
	VerifiedAskPrice     int64  
	StartEpoch           uint64
	FastRetrieval        bool
	ProposalCid          cid.Cid
	DataURIFetchAttempts uint32
	CreatedAt            time.Time
	UpdatedAt            time.Time
	ErrorCause           string
}
type BidStatus int
const (
	BidStatusUnspecified BidStatus = iota
	BidStatusSubmitted
	BidStatusAwaitingProposal
	BidStatusQueuedData
	BidStatusFetchingData
	BidStatusFinalized
	BidStatusErrored
)
var bidStatusStrings = map[BidStatus]string{
	BidStatusUnspecified:      "unspecified",
	BidStatusSubmitted:        "submitted",
	BidStatusAwaitingProposal: "awaiting_proposal",
	BidStatusQueuedData:       "queued_data",
	BidStatusFetchingData:     "fetching_data",
	BidStatusFinalized:        "finalized",
	BidStatusErrored:          "errored",
}
var bidStatusByString map[string]BidStatus
func init() {
	bidStatusByString = make(map[string]BidStatus)
	for b, s := range bidStatusStrings {
		bidStatusByString[s] = b
	}
}
func (as BidStatus) String() string {
	if s, exists := bidStatusStrings[as]; exists {
		return s
	}
	return "invalid"
}
func BidStatusByString(s string) (BidStatus, error) {
	if bs, exists := bidStatusByString[s]; exists {
		return bs, nil
	}
	return -1, errors.New("invalid bid status")
}
type Store struct {
	store        txndswrap.TxnDatastore
	fc           filclient.FilClient
	lc           lotusclient.LotusClient
	bytesLimiter limiter.Limiter
	jobCh  chan *Bid
	tickCh chan struct{}
	dealDataDirectory     string
	dealDataFetchAttempts uint32
	dealDataFetchTimeout  time.Duration
	dealProgressReporter  ProgressReporter
	ctx    context.Context
	cancel context.CancelFunc
	wg         sync.WaitGroup
	semImports *semaphore.Weighted  
}
func NewStore(
	store txndswrap.TxnDatastore,
	fc filclient.FilClient,
	lc lotusclient.LotusClient,
	dealDataDirectory string,
	dealDataFetchAttempts uint32,
	dealDataFetchTimeout time.Duration,
	dealProgressReporter ProgressReporter,
	discardOrphanDealsAfter time.Duration,
	bytesLimiter limiter.Limiter,
	concurrentImports int,
) (*Store, error) {
	ctx, cancel := context.WithCancel(context.Background())
	s := &Store{
		store:                 store,
		fc:                    fc,
		lc:                    lc,
		bytesLimiter:          bytesLimiter,
		jobCh:                 make(chan *Bid, MaxDataURIFetchConcurrency),
		tickCh:                make(chan struct{}, MaxDataURIFetchConcurrency),
		dealDataDirectory:     dealDataDirectory,
		dealDataFetchAttempts: dealDataFetchAttempts,
		dealDataFetchTimeout:  dealDataFetchTimeout,
		dealProgressReporter:  dealProgressReporter,
		ctx:                   ctx,
		cancel:                cancel,
	}
	if concurrentImports > 0 {
		s.semImports = semaphore.NewWeighted(int64(concurrentImports))
	}
	if s.dealProgressReporter == nil {
		s.dealProgressReporter = nullProgressReporter{}
	}
	if err := s.HealthCheck(); err != nil {
		return nil, fmt.Errorf("fails health check: %w", err)
	}
	s.wg.Add(MaxDataURIFetchConcurrency)
	for i := 0; i < MaxDataURIFetchConcurrency; i++ {
		go s.fetchWorker(i + 1)
	}
	if err := s.getOrphaned(); err != nil {
		return nil, fmt.Errorf("getting orphaned jobs: %w", err)
	}
	go s.startFetching()
	go s.periodicalGC(discardOrphanDealsAfter)
	return s, nil
}
func (s *Store) Close() error {
	s.cancel()
	s.wg.Wait()
	return nil
}
func (s *Store) SaveBid(bid Bid) error {
	if err := validate(bid); err != nil {
		return fmt.Errorf("invalid bid data: %s", err)
	}
	bid.CreatedAt = time.Now()
	if err := s.saveAndTransitionStatus(nil, &bid, BidStatusSubmitted); err != nil {
		return fmt.Errorf("saving bid: %v", err)
	}
	log.Infof("saved bid %s", bid.ID)
	return nil
}
func validate(b Bid) error {
	if b.ID == "" {
		return errors.New("id is empty")
	}
	if b.AuctionID == "" {
		return errors.New("auction id is empty")
	}
	if err := b.AuctioneerID.Validate(); err != nil {
		return fmt.Errorf("auctioneer id is not a valid peer id: %v", err)
	}
	if b.DealSize == 0 {
		return errors.New("deal size must be greater than zero")
	}
	if b.DealDuration == 0 {
		return errors.New("deal duration must be greater than zero")
	}
	if b.Status != BidStatusUnspecified {
		return errors.New("invalid initial bid status")
	}
	if b.AskPrice < 0 {
		return errors.New("ask price must be greater than or equal to zero")
	}
	if b.VerifiedAskPrice < 0 {
		return errors.New("verified ask price must be greater than or equal to zero")
	}
	if b.StartEpoch <= 0 {
		return errors.New("start epoch must be greater than zero")
	}
	if b.ProposalCid.Defined() {
		return errors.New("initial proposal cid cannot be defined")
	}
	if b.DataURIFetchAttempts != 0 {
		return errors.New("initial data uri fetch attempts must be zero")
	}
	if !b.CreatedAt.IsZero() {
		return errors.New("initial created at must be zero")
	}
	if !b.UpdatedAt.IsZero() {
		return errors.New("initial updated at must be zero")
	}
	if b.ErrorCause != "" {
		return errors.New("initial error cause must be empty")
	}
	return nil
}
func (s *Store) GetBid(id auction.BidID) (*Bid, error) {
	b, err := getBid(s.store, id)
	if err != nil {
		return nil, err
	}
	return b, nil
}
func getBid(reader ds.Read, id auction.BidID) (*Bid, error) {
	val, err := reader.Get(dsPrefix.ChildString(string(id)))
	if errors.Is(err, ds.ErrNotFound) {
		return nil, ErrBidNotFound
	} else if err != nil {
		return nil, fmt.Errorf("getting key: %v", err)
	}
	r, err := decode(val)
	if err != nil {
		return nil, fmt.Errorf("decoding value: %v", err)
	}
	return r, nil
}
func (s *Store) SetAwaitingProposalCid(id auction.BidID, sources auction.Sources) error {
	if err := sources.Validate(); err != nil {
		return err
	}
	txn, err := s.store.NewTransaction(false)
	if err != nil {
		return fmt.Errorf("creating txn: %v", err)
	}
	defer txn.Discard()
	b, err := getBid(txn, id)
	if err != nil {
		return err
	}
	if b.Status == BidStatusAwaitingProposal {
		log.Infof("bid %s already in '%s', duplicated message?", b.ID, b.Status)
		return nil
	}
	if b.Status != BidStatusSubmitted {
		return fmt.Errorf("expect bid to have status '%s', got '%s'", BidStatusSubmitted, b.Status)
	}
	b.Sources = sources
	if err := s.saveAndTransitionStatus(txn, b, BidStatusAwaitingProposal); err != nil {
		return fmt.Errorf("updating bid: %v", err)
	}
	if err := txn.Commit(); err != nil {
		return fmt.Errorf("committing txn: %v", err)
	}
	log.Infof("set awaiting proposal cid for bid %s", b.ID)
	return nil
}
func (s *Store) SetProposalCid(id auction.BidID, pcid cid.Cid) error {
	if !pcid.Defined() {
		return errors.New("proposal cid must be defined")
	}
	txn, err := s.store.NewTransaction(false)
	if err != nil {
		return fmt.Errorf("creating txn: %v", err)
	}
	defer txn.Discard()
	b, err := getBid(txn, id)
	if err != nil {
		return err
	}
	if b.Status > BidStatusAwaitingProposal {
		log.Infof("bid %s already in '%s', duplicated message?", b.ID, b.Status)
		return nil
	}
	if b.Status != BidStatusAwaitingProposal {
		return fmt.Errorf("expect bid to have status '%s', got '%s'", BidStatusAwaitingProposal, b.Status)
	}
	b.ProposalCid = pcid
	if err := s.enqueueDataURI(txn, b); err != nil {
		return fmt.Errorf("enqueueing data uri: %v", err)
	}
	if err := txn.Commit(); err != nil {
		return fmt.Errorf("committing txn: %v", err)
	}
	log.Infof("set proposal cid for bid %s; enqueued sources %s for fetch", b.ID, b.Sources)
	return nil
}
type Query struct {
	Offset string
	Order  Order
	Limit  int
}
func (q Query) setDefaults() Query {
	if q.Limit == -1 {
		q.Limit = maxListLimit
	} else if q.Limit <= 0 {
		q.Limit = defaultListLimit
	} else if q.Limit > maxListLimit {
		q.Limit = maxListLimit
	}
	return q
}
type Order int
const (
	OrderDescending Order = iota
	OrderAscending
)
func (s *Store) ListBids(query Query) ([]*Bid, error) {
	query = query.setDefaults()
	var (
		order dsq.Order
		seek  string
		limit = query.Limit
	)
	if len(query.Offset) != 0 {
		seek = dsPrefix.ChildString(query.Offset).String()
		limit++
	}
	switch query.Order {
	case OrderDescending:
		order = dsq.OrderByKeyDescending{}
		if len(seek) == 0 {
			seek = dsPrefix.ChildString(
				strings.ToLower(ulid.MustNew(ulid.MaxTime(), nil).String())).String()
		}
	case OrderAscending:
		order = dsq.OrderByKey{}
	}
	results, err := s.store.QueryExtended(dsextensions.QueryExt{
		Query: dsq.Query{
			Prefix: dsPrefix.String(),
			Orders: []dsq.Order{order},
			Limit:  limit,
		},
		SeekPrefix: seek,
	})
	if err != nil {
		return nil, fmt.Errorf("querying requests: %v", err)
	}
	defer func() {
		if err := results.Close(); err != nil {
			log.Errorf("closing results: %v", err)
		}
	}()
	var list []*Bid
	for res := range results.Next() {
		if res.Error != nil {
			return nil, fmt.Errorf("getting next result: %v", res.Error)
		}
		b, err := decode(res.Value)
		if err != nil {
			return nil, fmt.Errorf("decoding value: %v", err)
		}
		list = append(list, b)
	}
	if len(query.Offset) != 0 && len(list) > 0 {
		list = list[1:]
	}
	return list, nil
}
func (s *Store) WriteDealData(b *Bid) (string, error) {
	if b.Sources.CARURL != nil {
		return s.WriteDataURI(b.ID, b.PayloadCid.String(), b.Sources.CARURL.URL.String(), b.DealSize)
	}
	return "", errors.New("not implemented")
}