decouling of the AI indexer from dlockss and extracting the code to a different repo (cid-indexer-ipfs)

Author: ihlec

cmd/dlockss-monitor/main.go

@@ -53,9 +53,7 @@ func main() {
 	if geoDBPath == "" {
 		geoDBPath = os.Getenv("DLOCKSS_MONITOR_GEOIP_DB")
 	}
-	saiaAPIKey := os.Getenv("SAIA_API_KEY")
-
-	m := monitor.NewMonitor(cfg, geoDBPath, saiaAPIKey)
+	m := monitor.NewMonitor(cfg, geoDBPath)
 	defer m.Close()
 
 	h, err := monitor.StartLibP2P(ctx, m)

docs/DLOCKSS_PROTOCOL.md

@@ -247,14 +247,13 @@ services:
     depends_on: [ipfs]
     volumes:
       - ipfs-data:/ipfs-repo:ro
-      - dlockss-data:/data
+      - ./dlockss-files:/data
     environment:
       DLOCKSS_IPFS_CONFIG: /ipfs-repo/config
       DLOCKSS_IPFS_NODE: /dns4/ipfs/tcp/5001
       DLOCKSS_DATA_DIR: /data/ingest
 volumes:
   ipfs-data:
-  dlockss-data:
 ```
 
 **Legacy migration:** If a `dlockss.key` exists in the working directory but not at the configured identity path (e.g. after upgrading), it is automatically copied to the new location so the node retains its Peer ID.

internal/config/config.go

@@ -61,6 +61,21 @@ func getEnvBool(key string, defaultValue bool) bool {
 	return defaultValue
 }
 
+func getEnvStringSlice(key string) []string {
+	if value := os.Getenv(key); value != "" {
+		parts := strings.Split(value, ",")
+		var result []string
+		for _, p := range parts {
+			p = strings.TrimSpace(p)
+			if p != "" {
+				result = append(result, p)
+			}
+		}
+		return result
+	}
+	return nil
+}
+
 func clusterStorePath(dataDir string) string {
 	if p := os.Getenv("DLOCKSS_CLUSTER_STORE"); p != "" {
 		return p
@@ -79,10 +94,15 @@ func nodeNamePath(dataDir string) string {
 	return filepath.Join(filepath.Dir(dataDir), "node_name")
 }
 
+// DefaultTopicName is the archive topic when none is configured.
+const DefaultTopicName = "creative-commons"
+
 // Config holds all runtime configuration for a D-LOCKSS node.
 type Config struct {
 	DiscoveryServiceTag            string
 	PubsubTopicPrefix              string
+	TopicName                      string
+	IngestAllowlist                []string
 	FileWatchFolder                string
 	ClusterStorePath               string
 	MinReplication                 int
@@ -160,6 +180,8 @@ func DefaultConfig() *Config {
 	return &Config{
 		DiscoveryServiceTag:            "dlockss-prod",
 		PubsubTopicPrefix:              DefaultPubsubVersion,
+		TopicName:                      DefaultTopicName,
+		IngestAllowlist:                nil,
 		FileWatchFolder:                dataDir,
 		ClusterStorePath:               filepath.Join(filepath.Dir(dataDir), "cluster_store"),
 		MinReplication:                 5,
@@ -238,6 +260,8 @@ func LoadFromEnv() *Config {
 	return &Config{
 		DiscoveryServiceTag:            getEnvString("DLOCKSS_DISCOVERY_TAG", "dlockss-prod"),
 		PubsubTopicPrefix:              getEnvString("DLOCKSS_PUBSUB_TOPIC_PREFIX", DefaultPubsubVersion),
+		TopicName:                      getEnvString("DLOCKSS_TOPIC_NAME", DefaultTopicName),
+		IngestAllowlist:                getEnvStringSlice("DLOCKSS_INGEST_ALLOWLIST"),
 		FileWatchFolder:                dataDir,
 		ClusterStorePath:               clusterStorePath(dataDir),
 		MinReplication:                 getEnvInt("DLOCKSS_MIN_REPLICATION", 5),
@@ -372,9 +396,15 @@ func (c *Config) ValidatePathSafetyCheck() string {
 func (c *Config) Log() {
 	c.Validate()
 
+	ingestMode := "open"
+	if len(c.IngestAllowlist) > 0 {
+		ingestMode = fmt.Sprintf("allowlist (%d peers)", len(c.IngestAllowlist))
+	}
 	slog.Info("config: network",
 		"discovery_tag", c.DiscoveryServiceTag,
 		"pubsub_prefix", c.PubsubTopicPrefix,
+		"topic_name", c.TopicName,
+		"ingest_mode", ingestMode,
 		"ipfs_node", c.IPFSNodeAddress,
 		"api_port", c.APIPort,
 		"bootstrap_timeout", c.BootstrapTimeout,

internal/fileops/fileops.go

@@ -31,6 +31,7 @@ type ShardIdentity interface {
 	GetShardInfo() (string, int)
 	AnnouncePinned(manifestCID string)
 	AmIResponsibleFor(key string) bool
+	IsLocalNodeIngestor() bool
 }
 
 // ShardPublisher handles ingest publishing and cluster pinning within the node's own shard.

internal/fileops/fileops_process.go

@@ -45,6 +45,11 @@ func (fp *FileProcessor) validateFilePath(path string) bool {
 func (fp *FileProcessor) processNewFile(path string) {
 	slog.Info("processing file", "path", path)
 
+	if !fp.shardMgr.IsLocalNodeIngestor() {
+		slog.Warn("node is not an authorized ingestor, skipping file", "path", path)
+		return
+	}
+
 	if !fp.validateFilePath(path) {
 		slog.Warn("file validation failed", "path", path)
 		return

internal/fileops/fileops_test.go

@@ -51,8 +51,8 @@ func (m *mockIPFS) IsPinned(context.Context, cid.Cid) (bool, error)   { return f
 func (m *mockIPFS) GetFileSize(context.Context, cid.Cid) (uint64, error) {
 	return 0, nil
 }
-func (m *mockIPFS) GetPeerID(context.Context) (string, error)         { return "test-peer", nil }
-func (m *mockIPFS) SwarmConnect(context.Context, []string) error       { return nil }
+func (m *mockIPFS) GetPeerID(context.Context) (string, error)    { return "test-peer", nil }
+func (m *mockIPFS) SwarmConnect(context.Context, []string) error { return nil }
 
 // ---------------------------------------------------------------------------
 // Mock: ShardCoordinator (ShardIdentity + ShardPublisher + CustodialInjector)
@@ -64,15 +64,16 @@ type mockShardCoordinator struct {
 	shardDepth  int
 	responsible bool
 
-	mu           sync.Mutex
-	announced    []string
-	pinned       []cid.Cid
-	publishedTo  []string
+	mu          sync.Mutex
+	announced   []string
+	pinned      []cid.Cid
+	publishedTo []string
 }
 
-func (m *mockShardCoordinator) PeerID() peer.ID                  { return m.peerID }
-func (m *mockShardCoordinator) GetShardInfo() (string, int)      { return m.shardID, m.shardDepth }
-func (m *mockShardCoordinator) AmIResponsibleFor(string) bool    { return m.responsible }
+func (m *mockShardCoordinator) PeerID() peer.ID               { return m.peerID }
+func (m *mockShardCoordinator) GetShardInfo() (string, int)   { return m.shardID, m.shardDepth }
+func (m *mockShardCoordinator) AmIResponsibleFor(string) bool { return m.responsible }
+func (m *mockShardCoordinator) IsLocalNodeIngestor() bool     { return true }
 
 func (m *mockShardCoordinator) AnnouncePinned(manifestCID string) {
 	m.mu.Lock()
@@ -96,11 +97,11 @@ func (m *mockShardCoordinator) PublishIngestMessageToCurrentAndChildIfSplit(data
 func (m *mockShardCoordinator) ResolveTargetShardForCustodial(nominal, _ string) string {
 	return nominal
 }
-func (m *mockShardCoordinator) JoinShardAsObserver(string) bool             { return true }
-func (m *mockShardCoordinator) LeaveShardAsObserver(string)                 {}
+func (m *mockShardCoordinator) JoinShardAsObserver(string) bool                     { return true }
+func (m *mockShardCoordinator) LeaveShardAsObserver(string)                         {}
 func (m *mockShardCoordinator) EnsureClusterForShard(context.Context, string) error { return nil }
 func (m *mockShardCoordinator) PinToShard(context.Context, string, cid.Cid) error   { return nil }
-func (m *mockShardCoordinator) PublishToShardCBOR([]byte, string)           {}
+func (m *mockShardCoordinator) PublishToShardCBOR([]byte, string)                   {}
 
 // ---------------------------------------------------------------------------
 // Mock: StorageTracker