package blockstore import ( "bytes" "context" "io" "github.com/ipfs/boxo/path" blocks "github.com/ipfs/go-block-format" "github.com/ipfs/go-cid" "github.com/multiformats/go-multiaddr" "github.com/multiformats/go-multihash" "golang.org/x/xerrors" rpc "github.com/filecoin-project/kubo-api-client" iface "github.com/filecoin-project/kubo-api-client/coreiface" "github.com/filecoin-project/kubo-api-client/coreiface/options" ) type IPFSBlockstore struct { ctx context.Context api, offlineAPI iface.CoreAPI } var _ BasicBlockstore = (*IPFSBlockstore)(nil) func NewLocalIPFSBlockstore(ctx context.Context, onlineMode bool) (Blockstore, error) { localApi, err := rpc.NewLocalApi() if err != nil { return nil, xerrors.Errorf("getting local ipfs api: %w", err) } api, err := localApi.WithOptions(options.Api.Offline(!onlineMode)) if err != nil { return nil, xerrors.Errorf("setting offline mode: %s", err) } offlineAPI := api if onlineMode { offlineAPI, err = localApi.WithOptions(options.Api.Offline(true)) if err != nil { return nil, xerrors.Errorf("applying offline mode: %s", err) } } bs := &IPFSBlockstore{ ctx: ctx, api: api, offlineAPI: offlineAPI, } return Adapt(bs), nil } func NewRemoteIPFSBlockstore(ctx context.Context, maddr multiaddr.Multiaddr, onlineMode bool) (Blockstore, error) { httpApi, err := rpc.NewApi(maddr) if err != nil { return nil, xerrors.Errorf("setting remote ipfs api: %w", err) } api, err := httpApi.WithOptions(options.Api.Offline(!onlineMode)) if err != nil { return nil, xerrors.Errorf("applying offline mode: %s", err) } offlineAPI := api if onlineMode { offlineAPI, err = httpApi.WithOptions(options.Api.Offline(true)) if err != nil { return nil, xerrors.Errorf("applying offline mode: %s", err) } } bs := &IPFSBlockstore{ ctx: ctx, api: api, offlineAPI: offlineAPI, } return Adapt(bs), nil } func (i *IPFSBlockstore) DeleteBlock(ctx context.Context, cid cid.Cid) error { return xerrors.Errorf("not supported") } func (i *IPFSBlockstore) Has(ctx context.Context, cid cid.Cid) (bool, error) { _, err := i.offlineAPI.Block().Stat(ctx, path.FromCid(cid)) if err != nil { // The underlying client is running in Offline mode. // Stat() will fail with an err if the block isn't in the // blockstore. If that's the case, return false without // an error since that's the original intention of this method. if err.Error() == "blockservice: key not found" { return false, nil } return false, xerrors.Errorf("getting ipfs block: %w", err) } return true, nil } func (i *IPFSBlockstore) Get(ctx context.Context, cid cid.Cid) (blocks.Block, error) { rd, err := i.api.Block().Get(ctx, path.FromCid(cid)) if err != nil { return nil, xerrors.Errorf("getting ipfs block: %w", err) } data, err := io.ReadAll(rd) if err != nil { return nil, err } return blocks.NewBlockWithCid(data, cid) } func (i *IPFSBlockstore) GetSize(ctx context.Context, cid cid.Cid) (int, error) { st, err := i.api.Block().Stat(ctx, path.FromCid(cid)) if err != nil { return 0, xerrors.Errorf("getting ipfs block: %w", err) } return st.Size(), nil } func (i *IPFSBlockstore) Put(ctx context.Context, block blocks.Block) error { mhd, err := multihash.Decode(block.Cid().Hash()) if err != nil { return err } _, err = i.api.Block().Put(ctx, bytes.NewReader(block.RawData()), options.Block.Hash(mhd.Code, mhd.Length), options.Block.Format(multihash.Codes[block.Cid().Type()])) return err } func (i *IPFSBlockstore) PutMany(ctx context.Context, blocks []blocks.Block) error { // TODO: could be done in parallel for _, block := range blocks { if err := i.Put(ctx, block); err != nil { return err } } return nil } func (i *IPFSBlockstore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) { return nil, xerrors.Errorf("not supported") } func (i *IPFSBlockstore) HashOnRead(enabled bool) { return // TODO: We could technically support this, but.. }