Repository: ipfs/kubo

Issue:
/cmd/ipfswatch `unknown datastore type: flatfs`
#### Version information:
<!-- Output From `ipfs version --all`

Please check dist.ipfs.io for a newer version of go-ipfs and update if necessary. Report back if the problem persists.

If you can't run `ipfs version --all` or that command fails, include as much information as you can: IPFS version, computer architecture (e.g., Intel x86 64bit), operating system, etc. -->
```
$ ipfs version --all
go-ipfs version: 0.4.22-
Repo version: 7
System version: amd64/linux
Golang version: go1.12.8
```

#### Description:
<!-- This is where you get to tell us what went wrong. When doing so, please make sure to include *all* relevant information.

Please try to include:
  * What you were doing when you experienced the bug.
  * Any error messages you saw, *where* you saw them, and what you believe may have caused them (if you have any ideas).
  * When possible, steps to reliably produce the bug.
-->

ipfswatch doesn't recognize the datastore type `flatfs`

```
~/ $ go get -u github.com/ipfs/go-ipfs/cmd/ipfswatch
~/ $ cd $(mktemp -d)
/tmp/tmp.9emc06hc3k $ export IPFS_PATH=$PWD/.ipfs
/tmp/tmp.9emc06hc3k $ ipfs init
initializing IPFS node at /tmp/tmp.9emc06hc3k/.ipfs
generating 2048-bit RSA keypair...done
peer identity: QmQPHBp7hgnUAuhjLafAxpu8npKLq5hNhPtJycragQtTRh
to get started, enter:

        ipfs cat /ipfs/QmS4ustL54uo8FzR9455qaxZwuMiUhyvMcX9Ba8nUH4uVv/readme

/tmp/tmp.9emc06hc3k $ mkdir watchDir
/tmp/tmp.9emc06hc3k $ cd watchDir 
tmp.9emc06hc3k/watchDir $ ipfswatch 
2019/10/22 10:27:23 running IPFSWatch on '.' using repo at '/tmp/tmp.9emc06hc3k/.ipfs'...
2019/10/22 10:27:23 .
2019/10/22 10:27:23 unknown datastore type: flatfs
```


Discussion:
@rudolph9: Associated ipfs config:

```
{
  "Identity": {
    "PeerID": "QmQPHBp7hgnUAuhjLafAxpu8npKLq5hNhPtJycragQtTRh",
    "PrivKey": <PRIVATE_KEY>
  },
  "Datastore": {
    "StorageMax": "10GB",
    "StorageGCWatermark": 90,
    "GCPeriod": "1h",
    "Spec": {
      "mounts": [
        {
          "child": {
            "path": "blocks",
            "shardFunc": "/repo/flatfs/shard/v1/next-to-last/2",
            "sync": true,
            "type": "flatfs"
          },
          "mountpoint": "/blocks",
          "prefix": "flatfs.datastore",
          "type": "measure"
        },
        {
          "child": {
            "compression": "none",
            "path": "datastore",
            "type": "levelds"
          },
          "mountpoint": "/",
          "prefix": "leveldb.datastore",
          "type": "measure"
        }
      ],
      "type": "mount"
    },
    "HashOnRead": false,
    "BloomFilterSize": 0
  },
  "Addresses": {
    "Swarm": [
      "/ip4/0.0.0.0/tcp/4001",
      "/ip6/::/tcp/4001"
    ],
    "Announce": [],
    "NoAnnounce": [],
    "API": "/ip4/127.0.0.1/tcp/5001",
    "Gateway": "/ip4/127.0.0.1/tcp/8080"
  },
  "Mounts": {
    "IPFS": "/ipfs",
    "IPNS": "/ipns",
    "FuseAllowOther": false
  },
  "Discovery": {
    "MDNS": {
      "Enabled": true,
      "Interval": 10
    }
  },
  "Routing": {
    "Type": "dht"
  },
  "Ipns": {
    "RepublishPeriod": "",
    "RecordLifetime": "",
    "ResolveCacheSize": 128
  },
  "Bootstrap": [
    "/dnsaddr/bootstrap.libp2p.io/ipfs/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
    "/dnsaddr/bootstrap.libp2p.io/ipfs/QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa",
    "/dnsaddr/bootstrap.libp2p.io/ipfs/QmbLHAnMoJPWSCR5Zhtx6BHJX9KiKNN6tpvbUcqanj75Nb",
    "/dnsaddr/bootstrap.libp2p.io/ipfs/QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt",
    "/ip4/104.131.131.82/tcp/4001/ipfs/QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ",
    "/ip4/104.236.179.241/tcp/4001/ipfs/QmSoLPppuBtQSGwKDZT2M73ULpjvfd3aZ6ha4oFGL1KrGM",
    "/ip4/128.199.219.111/tcp/4001/ipfs/QmSoLSafTMBsPKadTEgaXctDQVcqN88CNLHXMkTNwMKPnu",
    "/ip4/104.236.76.40/tcp/4001/ipfs/QmSoLV4Bbm51jM9C4gDYZQ9Cy3U6aXMJDAbzgu2fzaDs64",
    "/ip4/178.62.158.247/tcp/4001/ipfs/QmSoLer265NRgSp2LA3dPaeykiS1J6DifTC88f5uVQKNAd",
    "/ip6/2604:a880:1:20::203:d001/tcp/4001/ipfs/QmSoLPppuBtQSGwKDZT2M73ULpjvfd3aZ6ha4oFGL1KrGM",
    "/ip6/2400:6180:0:d0::151:6001/tcp/4001/ipfs/QmSoLSafTMBsPKadTEgaXctDQVcqN88CNLHXMkTNwMKPnu",
    "/ip6/2604:a880:800:10::4a:5001/tcp/4001/ipfs/QmSoLV4Bbm51jM9C4gDYZQ9Cy3U6aXMJDAbzgu2fzaDs64",
    "/ip6/2a03:b0c0:0:1010::23:1001/tcp/4001/ipfs/QmSoLer265NRgSp2LA3dPaeykiS1J6DifTC88f5uVQKNAd"
  ],
  "Gateway": {
    "HTTPHeaders": {
      "Access-Control-Allow-Headers": [
        "X-Requested-With",
        "Range",
        "User-Agent"
      ],
      "Access-Control-Allow-Methods": [
        "GET"
      ],
      "Access-Control-Allow-Origin": [
        "*"
      ]
    },
    "RootRedirect": "",
    "Writable": false,
    "PathPrefixes": [],
    "APICommands": [],
    "NoFetch": false
  },
  "API": {
    "HTTPHeaders": {}
  },
  "Swarm": {
    "AddrFilters": null,
    "DisableBandwidthMetrics": false,
    "DisableNatPortMap": false,
    "DisableRelay": false,
    "EnableRelayHop": false,
    "EnableAutoRelay": false,
    "EnableAutoNATService": false,
    "ConnMgr": {
      "Type": "basic",
      "LowWater": 600,
      "HighWater": 900,
      "GracePeriod": "20s"
    }
  },
  "Pubsub": {
    "Router": "",
    "DisableSigning": false,
    "StrictSignatureVerification": false
  },
  "Reprovider": {
    "Interval": "12h",
    "Strategy": "all"
  },
  "Experimental": {
    "FilestoreEnabled": false,
    "UrlstoreEnabled": false,
    "ShardingEnabled": false,
    "Libp2pStreamMounting": false,
    "P2pHttpProxy": false,
    "QUIC": false,
    "PreferTLS": false
  }
}
```@gammazero: The problem is that ipfswatch does not recognize datastores that are loaded as plugins. This is fixed by #11078.

Solution patch:
diff --git a/cmd/ipfswatch/main.go b/cmd/ipfswatch/main.go
index 79ea24224..3ba5dd3d9 100644
--- a/cmd/ipfswatch/main.go
+++ b/cmd/ipfswatch/main.go
@@ -9,6 +9,7 @@ import (
 	"os"
 	"os/signal"
 	"path/filepath"
+	"slices"
 	"syscall"
 
 	commands "github.com/ipfs/kubo/commands"
@@ -17,6 +18,11 @@ import (
 	coreapi "github.com/ipfs/kubo/core/coreapi"
 	corehttp "github.com/ipfs/kubo/core/corehttp"
 	"github.com/ipfs/kubo/misc/fsutil"
+	"github.com/ipfs/kubo/plugin"
+	pluginbadgerds "github.com/ipfs/kubo/plugin/plugins/badgerds"
+	pluginflatfs "github.com/ipfs/kubo/plugin/plugins/flatfs"
+	pluginlevelds "github.com/ipfs/kubo/plugin/plugins/levelds"
+	pluginpebbleds "github.com/ipfs/kubo/plugin/plugins/pebbleds"
 	fsrepo "github.com/ipfs/kubo/repo/fsrepo"
 
 	fsnotify "github.com/fsnotify/fsnotify"
@@ -60,6 +66,18 @@ func main() {
 	}
 }
 
+func loadDatastorePlugins(plugins []plugin.Plugin) error {
+	for _, pl := range plugins {
+		if pl, ok := pl.(plugin.PluginDatastore); ok {
+			err := fsrepo.AddDatastoreConfigHandler(pl.DatastoreTypeName(), pl.DatastoreConfigParser())
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 func run(ipfsPath, watchPath string) error {
 	log.Printf("running IPFSWatch on '%s' using repo at '%s'...", watchPath, ipfsPath)
 
@@ -77,6 +95,15 @@ func run(ipfsPath, watchPath string) error {
 		return err
 	}
 
+	if err = loadDatastorePlugins(slices.Concat(
+		pluginbadgerds.Plugins,
+		pluginflatfs.Plugins,
+		pluginlevelds.Plugins,
+		pluginpebbleds.Plugins,
+	)); err != nil {
+		return err
+	}
+
 	r, err := fsrepo.Open(ipfsPath)
 	if err != nil {
 		// TODO handle case: daemon running



Test patch:
diff --git a/test/cli/ipfswatch_test.go b/test/cli/ipfswatch_test.go
new file mode 100644
index 000000000..07e73a7ec
--- /dev/null
+++ b/test/cli/ipfswatch_test.go
@@ -0,0 +1,162 @@
+//go:build !plan9
+
+package cli
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"regexp"
+	"testing"
+	"time"
+
+	"github.com/ipfs/kubo/config"
+	"github.com/ipfs/kubo/test/cli/harness"
+	"github.com/stretchr/testify/require"
+)
+
+func TestIPFSWatch(t *testing.T) {
+	t.Parallel()
+
+	// Build ipfswatch binary once before running parallel subtests.
+	// This avoids race conditions and duplicate builds.
+	h := harness.NewT(t)
+	repoRoot := filepath.Dir(filepath.Dir(filepath.Dir(h.IPFSBin)))
+	ipfswatchBin := filepath.Join(repoRoot, "cmd", "ipfswatch", "ipfswatch")
+
+	if _, err := os.Stat(ipfswatchBin); os.IsNotExist(err) {
+		// -C changes to repo root so go.mod is found
+		cmd := exec.Command("go", "build", "-C", repoRoot, "-o", ipfswatchBin, "./cmd/ipfswatch")
+		out, err := cmd.CombinedOutput()
+		require.NoError(t, err, "failed to build ipfswatch: %s", string(out))
+	}
+
+	t.Run("ipfswatch adds watched files to IPFS", func(t *testing.T) {
+		t.Parallel()
+		h := harness.NewT(t)
+		node := h.NewNode().Init()
+
+		// Create a temp directory to watch
+		watchDir := filepath.Join(h.Dir, "watch")
+		err := os.MkdirAll(watchDir, 0o755)
+		require.NoError(t, err)
+
+		// Start ipfswatch in background
+		result := node.Runner.Run(harness.RunRequest{
+			Path:    ipfswatchBin,
+			Args:    []string{"--repo", node.Dir, "--path", watchDir},
+			RunFunc: harness.RunFuncStart,
+		})
+		require.NoError(t, result.Err, "ipfswatch should start without error")
+		defer func() {
+			if result.Cmd.Process != nil {
+				_ = result.Cmd.Process.Kill()
+				_, _ = result.Cmd.Process.Wait()
+			}
+		}()
+
+		// Wait for ipfswatch to initialize
+		time.Sleep(2 * time.Second)
+
+		// Check for startup errors
+		stderrStr := result.Stderr.String()
+		require.NotContains(t, stderrStr, "unknown datastore type", "ipfswatch should recognize datastore plugins")
+
+		// Create a test file with unique content based on timestamp
+		testContent := fmt.Sprintf("ipfswatch test content generated at %s", time.Now().Format(time.RFC3339Nano))
+		testFile := filepath.Join(watchDir, "test.txt")
+		err = os.WriteFile(testFile, []byte(testContent), 0o644)
+		require.NoError(t, err)
+
+		// Wait for ipfswatch to process the file and extract CID from log
+		// Log format: "added %s... key: %s"
+		cidPattern := regexp.MustCompile(`added .*/test\.txt\.\.\. key: (\S+)`)
+		var cid string
+		deadline := time.Now().Add(10 * time.Second)
+		for time.Now().Before(deadline) {
+			stderrStr = result.Stderr.String()
+			if matches := cidPattern.FindStringSubmatch(stderrStr); len(matches) > 1 {
+				cid = matches[1]
+				break
+			}
+			time.Sleep(100 * time.Millisecond)
+		}
+		require.NotEmpty(t, cid, "ipfswatch should have added test.txt and logged the CID, got stderr: %s", stderrStr)
+
+		// Kill ipfswatch to release the repo lock
+		if result.Cmd.Process != nil {
+			_ = result.Cmd.Process.Kill()
+			_, _ = result.Cmd.Process.Wait()
+		}
+
+		// Verify the content matches by reading it back via ipfs cat
+		catRes := node.RunIPFS("cat", "--offline", cid)
+		require.Equal(t, 0, catRes.Cmd.ProcessState.ExitCode(),
+			"ipfs cat should succeed, cid=%s, stderr: %s", cid, catRes.Stderr.String())
+		require.Equal(t, testContent, catRes.Stdout.String(),
+			"content read from IPFS should match what was written")
+	})
+
+	t.Run("ipfswatch loads datastore plugins for pebbleds", func(t *testing.T) {
+		t.Parallel()
+		h := harness.NewT(t)
+		node := h.NewNode().Init()
+
+		// Configure pebbleds as the datastore
+		node.UpdateConfig(func(cfg *config.Config) {
+			cfg.Datastore.Spec = map[string]interface{}{
+				"type": "mount",
+				"mounts": []interface{}{
+					map[string]interface{}{
+						"mountpoint": "/blocks",
+						"path":       "blocks",
+						"prefix":     "flatfs.datastore",
+						"shardFunc":  "/repo/flatfs/shard/v1/next-to-last/2",
+						"sync":       true,
+						"type":       "flatfs",
+					},
+					map[string]interface{}{
+						"mountpoint": "/",
+						"path":       "datastore",
+						"prefix":     "pebble.datastore",
+						"type":       "pebbleds",
+					},
+				},
+			}
+		})
+
+		// Re-initialize datastore directory for pebbleds
+		// (the repo was initialized with levelds, need to remove it)
+		dsPath := filepath.Join(node.Dir, "datastore")
+		err := os.RemoveAll(dsPath)
+		require.NoError(t, err)
+		err = os.MkdirAll(dsPath, 0o755)
+		require.NoError(t, err)
+
+		// Create a temp directory to watch
+		watchDir := filepath.Join(h.Dir, "watch")
+		err = os.MkdirAll(watchDir, 0o755)
+		require.NoError(t, err)
+
+		// Start ipfswatch in background
+		result := node.Runner.Run(harness.RunRequest{
+			Path:    ipfswatchBin,
+			Args:    []string{"--repo", node.Dir, "--path", watchDir},
+			RunFunc: harness.RunFuncStart,
+		})
+		require.NoError(t, result.Err, "ipfswatch should start without error")
+		defer func() {
+			if result.Cmd.Process != nil {
+				_ = result.Cmd.Process.Kill()
+				_, _ = result.Cmd.Process.Wait()
+			}
+		}()
+
+		// Wait for ipfswatch to initialize and check for errors
+		time.Sleep(3 * time.Second)
+
+		stderrStr := result.Stderr.String()
+		require.NotContains(t, stderrStr, "unknown datastore type", "ipfswatch should recognize pebbleds datastore plugin")
+	})
+}