ipfs: handle migrations gracefully

if repo needs migrations, fetch them and run them just like upstream ipfs daemon does

until now, we simply Fatal'ed out of the situation

data/ipfs/init.go

@@ -2,6 +2,7 @@ package ipfs
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -16,7 +17,10 @@ import (
 	ipfscore "github.com/ipfs/kubo/core"
 	ipfsapi "github.com/ipfs/kubo/core/coreapi"
 	"github.com/ipfs/kubo/plugin/loader"
+	"github.com/ipfs/kubo/repo"
 	"github.com/ipfs/kubo/repo/fsrepo"
+	"github.com/ipfs/kubo/repo/fsrepo/migrations"
+	"github.com/ipfs/kubo/repo/fsrepo/migrations/ipfsfetcher"
 	"github.com/multiformats/go-multicodec"
 	"github.com/multiformats/go-multihash"
 
@@ -73,6 +77,10 @@ func initRepository() error {
 func startNode() (*ipfscore.IpfsNode, coreiface.CoreAPI, error) {
 	log.Infow("starting IPFS node", "config", ConfigRoot)
 	r, err := fsrepo.Open(ConfigRoot)
+	if errors.Is(err, fsrepo.ErrNeedMigration) {
+		log.Warn("Found outdated ipfs repo, migrations need to be run.")
+		r, err = runMigrationsAndOpen(ConfigRoot)
+	}
 	if err != nil {
 		return nil, nil, err
 	}
@@ -99,6 +107,58 @@ func startNode() (*ipfscore.IpfsNode, coreiface.CoreAPI, error) {
 	return node, api, nil
 }
 
+// runMigrationsAndOpen fetches and applies migrations just like upstream kubo does
+// and returns fsrepo.Open(ConfigRoot)
+func runMigrationsAndOpen(ConfigRoot string) (repo.Repo, error) {
+	// Read Migration section of IPFS config
+	migrationCfg, err := migrations.ReadMigrationConfig(ConfigRoot, "")
+	if err != nil {
+		return nil, err
+	}
+
+	// Define function to create IPFS fetcher.  Do not supply an
+	// already-constructed IPFS fetcher, because this may be expensive and
+	// not needed according to migration config. Instead, supply a function
+	// to construct the particular IPFS fetcher implementation used here,
+	// which is called only if an IPFS fetcher is needed.
+	newIpfsFetcher := func(distPath string) migrations.Fetcher {
+		return ipfsfetcher.NewIpfsFetcher(distPath, 0, &ConfigRoot, "")
+	}
+
+	// Fetch migrations from current distribution, or location from environ
+	fetchDistPath := migrations.GetDistPathEnv(migrations.CurrentIpfsDist)
+
+	// Create fetchers according to migrationCfg.DownloadSources
+	fetcher, err := migrations.GetMigrationFetcher(migrationCfg.DownloadSources,
+		fetchDistPath, newIpfsFetcher)
+	if err != nil {
+		return nil, err
+	}
+	defer fetcher.Close()
+
+	if migrationCfg.Keep == "cache" || migrationCfg.Keep == "pin" {
+		// Create temp directory to store downloaded migration archives
+		migrations.DownloadDirectory, err = os.MkdirTemp("", "migrations")
+		if err != nil {
+			return nil, err
+		}
+		// Defer cleanup of download directory so that it gets cleaned up
+		// if daemon returns early due to error
+		defer func() {
+			if migrations.DownloadDirectory != "" {
+				_ = os.RemoveAll(migrations.DownloadDirectory)
+			}
+		}()
+	}
+
+	err = migrations.RunMigration(context.TODO(), fetcher, fsrepo.RepoVersion, ConfigRoot, false)
+	if err != nil {
+		return nil, fmt.Errorf("migrations of ipfs-repo failed: %w", err)
+	}
+
+	return fsrepo.Open(ConfigRoot)
+}
+
 // CmdCtx returns a commands.Context for the given node and repo path.
 func cmdCtx(node *ipfscore.IpfsNode, repoPath string) commands.Context {
 	return commands.Context{