lotus/paychmgr/store.go

package paychmgr

import (
	"bytes"
	"errors"
	"fmt"
	"strings"
	"sync"

	"github.com/filecoin-project/specs-actors/actors/builtin/paych"
	"github.com/ipfs/go-datastore"
	"github.com/ipfs/go-datastore/namespace"
	dsq "github.com/ipfs/go-datastore/query"
	"golang.org/x/xerrors"

	"github.com/filecoin-project/go-address"
	cborrpc "github.com/filecoin-project/go-cbor-util"

	"github.com/filecoin-project/lotus/node/modules/dtypes"
)

var ErrChannelNotTracked = errors.New("channel not tracked")

type Store struct {
	lk sync.Mutex // TODO: this can be split per paych

	ds datastore.Batching
}

func NewStore(ds dtypes.MetadataDS) *Store {
	ds = namespace.Wrap(ds, datastore.NewKey("/paych/"))
	return &Store{
		ds: ds,
	}
}

const (
	DirInbound  = 1
	DirOutbound = 2
)

type VoucherInfo struct {
	Voucher *paych.SignedVoucher
	Proof   []byte
}

type ChannelInfo struct {
	Channel address.Address
	Control address.Address
	Target  address.Address

	Direction uint64
	Vouchers  []*VoucherInfo
	NextLane  int64
}

func dskeyForChannel(addr address.Address) datastore.Key {
	return datastore.NewKey(addr.String())
}

func (ps *Store) putChannelInfo(ci *ChannelInfo) error {
	k := dskeyForChannel(ci.Channel)

	b, err := cborrpc.Dump(ci)
	if err != nil {
		return err
	}

	return ps.ds.Put(k, b)
}

func (ps *Store) getChannelInfo(addr address.Address) (*ChannelInfo, error) {
	k := dskeyForChannel(addr)

	b, err := ps.ds.Get(k)
	if err == datastore.ErrNotFound {
		return nil, ErrChannelNotTracked
	}
	if err != nil {
		return nil, err
	}

	var ci ChannelInfo
	if err := ci.UnmarshalCBOR(bytes.NewReader(b)); err != nil {
		return nil, err
	}

	return &ci, nil
}

func (ps *Store) TrackChannel(ch *ChannelInfo) error {
	ps.lk.Lock()
	defer ps.lk.Unlock()

	return ps.trackChannel(ch)
}

func (ps *Store) trackChannel(ch *ChannelInfo) error {
	_, err := ps.getChannelInfo(ch.Channel)
	switch err {
	default:
		return err
	case nil:
		return fmt.Errorf("already tracking channel: %s", ch.Channel)
	case ErrChannelNotTracked:
		return ps.putChannelInfo(ch)
	}
}

func (ps *Store) ListChannels() ([]address.Address, error) {
	ps.lk.Lock()
	defer ps.lk.Unlock()

	res, err := ps.ds.Query(dsq.Query{KeysOnly: true})
	if err != nil {
		return nil, err
	}
	defer res.Close()

	var out []address.Address
	for {
		res, ok := res.NextSync()
		if !ok {
			break
		}

		if res.Error != nil {
			return nil, err
		}

		addr, err := address.NewFromString(strings.TrimPrefix(res.Key, "/"))
		if err != nil {
			return nil, xerrors.Errorf("failed reading paych key (%q) from datastore: %w", res.Key, err)
		}

		out = append(out, addr)
	}

	return out, nil
}

func (ps *Store) findChan(filter func(*ChannelInfo) bool) (address.Address, error) {
	res, err := ps.ds.Query(dsq.Query{})
	if err != nil {
		return address.Undef, err
	}
	defer res.Close()

	var ci ChannelInfo

	for {
		res, ok := res.NextSync()
		if !ok {
			break
		}

		if res.Error != nil {
			return address.Undef, err
		}

		if err := ci.UnmarshalCBOR(bytes.NewReader(res.Value)); err != nil {
			return address.Undef, err
		}

		if !filter(&ci) {
			continue
		}

		addr, err := address.NewFromString(strings.TrimPrefix(res.Key, "/"))
		if err != nil {
			return address.Undef, xerrors.Errorf("failed reading paych key (%q) from datastore: %w", res.Key, err)
		}

		return addr, nil
	}

	return address.Undef, nil
}

func (ps *Store) AllocateLane(ch address.Address) (int64, error) {
	ps.lk.Lock()
	defer ps.lk.Unlock()

	ci, err := ps.getChannelInfo(ch)
	if err != nil {
		return 0, err
	}

	out := ci.NextLane
	ci.NextLane++

	return out, ps.putChannelInfo(ci)
}

func (ps *Store) VouchersForPaych(ch address.Address) ([]*VoucherInfo, error) {
	ci, err := ps.getChannelInfo(ch)
	if err != nil {
		return nil, err
	}

	return ci.Vouchers, nil
}