pinconv.go

// Package pinconv converts pins between the dag-based ldpinner and the
// datastore-based dspinner.  Once conversion is complete, the pins from the
// source pinner are removed.
package pinconv

import (
	"context"
	"fmt"

	cid "gitlab.dms3.io/dms3/go-cid"
	ds "gitlab.dms3.io/dms3/go-datastore"
	dms3pinner "gitlab.dms3.io/dms3/go-dms3-pinner"
	"gitlab.dms3.io/dms3/go-dms3-pinner/dspinner"
	"gitlab.dms3.io/dms3/go-dms3-pinner/ldpinner"
	ld "gitlab.dms3.io/dms3/go-ld-format"
)

// ConvertPinsFromLDToDS converts pins stored in mdag based storage to pins
// stores in the datastore.  Returns a dspinner loaded with the converted pins,
// and a count of the recursive and direct pins converted.
//
// After pins are stored in datastore, the root pin key is deleted to unlink
// the pin data in the DAGService.
func ConvertPinsFromLDToDS(ctx context.Context, dstore ds.Datastore, dserv ld.DAGService, internal ld.DAGService) (dms3pinner.Pinner, int, error) {
	const ldPinPath = "/local/pins"

	dsPinner, err := dspinner.New(ctx, dstore, dserv)
	if err != nil {
		return nil, 0, err
	}

	var convCount int
	keyChan := make(chan cid.Cid)

	go func() {
		err = ldpinner.LoadKeys(ctx, dstore, dserv, internal, true, keyChan)
		close(keyChan)
	}()
	for key := range keyChan {
		dsPinner.PinWithMode(key, dms3pinner.Recursive)
		convCount++
	}
	if err != nil {
		return nil, 0, fmt.Errorf("cannot load recursive keys: %s", err)
	}

	keyChan = make(chan cid.Cid)
	go func() {
		err = ldpinner.LoadKeys(ctx, dstore, dserv, internal, false, keyChan)
		close(keyChan)
	}()
	for key := range keyChan {
		dsPinner.PinWithMode(key, dms3pinner.Direct)
		convCount++
	}
	if err != nil {
		return nil, 0, fmt.Errorf("cannot load direct keys: %s", err)
	}

	err = dsPinner.Flush(ctx)
	if err != nil {
		return nil, 0, err
	}

	// Delete root mdag key from datastore to remove old pin storage.
	ldPinDatastoreKey := ds.NewKey(ldPinPath)
	if err = dstore.Delete(ldPinDatastoreKey); err != nil {
		return nil, 0, fmt.Errorf("cannot delete old pin state: %v", err)
	}
	if err = dstore.Sync(ldPinDatastoreKey); err != nil {
		return nil, 0, fmt.Errorf("cannot sync old pin state: %v", err)
	}

	return dsPinner, convCount, nil
}

// ConvertPinsFromDSToLD converts the pins stored in the datastore by
// dspinner, into pins stored in the given internal DAGService by ldpinner.
// Returns an ldpinner loaded with the converted pins, and a count of the
// recursive and direct pins converted.
//
// After the pins are stored in the DAGService, the pins and their indexes are
// removed from the dspinner.
func ConvertPinsFromDSToLD(ctx context.Context, dstore ds.Datastore, dserv ld.DAGService, internal ld.DAGService) (dms3pinner.Pinner, int, error) {
	dsPinner, err := dspinner.New(ctx, dstore, dserv)
	if err != nil {
		return nil, 0, err
	}

	ldPinner, err := ldpinner.New(dstore, dserv, internal)
	if err != nil {
		return nil, 0, err
	}

	cids, err := dsPinner.RecursiveKeys(ctx)
	if err != nil {
		return nil, 0, err
	}
	for i := range cids {
		ldPinner.PinWithMode(cids[i], dms3pinner.Recursive)
		dsPinner.RemovePinWithMode(cids[i], dms3pinner.Recursive)
	}
	convCount := len(cids)

	cids, err = dsPinner.DirectKeys(ctx)
	if err != nil {
		return nil, 0, err
	}
	for i := range cids {
		ldPinner.PinWithMode(cids[i], dms3pinner.Direct)
		dsPinner.RemovePinWithMode(cids[i], dms3pinner.Direct)
	}
	convCount += len(cids)

	// Save the ldpinner pins
	err = ldPinner.Flush(ctx)
	if err != nil {
		return nil, 0, err
	}

	err = dsPinner.Flush(ctx)
	if err != nil {
		return nil, 0, err
	}

	return ldPinner, convCount, nil
}