feat: support mongo connector

coco.yml

@@ -319,6 +319,11 @@ connector:
     interval: 10s
     queue:
       name: indexing_documents
+  mongodb:  
+    enabled: true  
+    interval: 30s  
+    queue:  
+      name: indexing_documents 
   notion:
     enabled: true
     interval: 10s

docker/init-mongo.js

@@ -0,0 +1,41 @@
+// MongoDB initialization script for testing  
+db = db.getSiblingDB('coco_test');  
+
+// Create test user  
+db.createUser({  
+  user: 'coco_test',  
+  pwd: 'test_password',  
+  roles: [  
+    {  
+      role: 'readWrite',  
+      db: 'coco_test'  
+    }  
+  ]  
+});  
+
+// Create test collections with sample data  
+db.articles.insertMany([  
+  {  
+    title: "Sample Article 1",  
+    content: "This is sample content for testing",  
+    category: "Technology",  
+    tags: ["mongodb", "database"],  
+    url: "https://example.com/article1",  
+    updated_at: new Date(),  
+    status: "published"  
+  },  
+  {  
+    title: "Sample Article 2",   
+    content: "Another sample content for testing",  
+    category: "Programming",  
+    tags: ["go", "backend"],  
+    url: "https://example.com/article2",  
+    updated_at: new Date(),  
+    status: "draft"  
+  }  
+]);  
+
+// Create indexes for better performance  
+db.articles.createIndex({ "updated_at": 1 });  
+db.articles.createIndex({ "status": 1 });  
+db.articles.createIndex({ "category": 1 });

docker/mongodb-test.yml

@@ -0,0 +1,36 @@
+version: '3.8'  
+
+services:  
+  mongodb:  
+    image: mongo:7.0  
+    container_name: coco-mongodb-test  
+    ports:  
+      - "27017:27017"  
+    environment:  
+      MONGO_INITDB_ROOT_USERNAME: admin  
+      MONGO_INITDB_ROOT_PASSWORD: password  
+      MONGO_INITDB_DATABASE: coco_test  
+    volumes:  
+      - mongodb_data:/data/db  
+      - ./init-mongo.js:/docker-entrypoint-initdb.d/init-mongo.js:ro  
+    networks:  
+      - coco-test  
+
+  mongodb-replica:  
+    image: mongo:7.0  
+    container_name: coco-mongodb-replica-test  
+    ports:  
+      - "27018:27017"  
+    command: mongod --replSet rs0 --bind_ip_all  
+    volumes:  
+      - mongodb_replica_data:/data/db  
+    networks:  
+      - coco-test  
+
+volumes:  
+  mongodb_data:  
+  mongodb_replica_data:  
+
+networks:  
+  coco-test:  
+    driver: bridge

examples/mongodb.yml

@@ -0,0 +1,22 @@
+# MongoDB Connector Default Configuration  
+mongodb:  
+  # Default connection settings  
+  default_timeout: "30s"  
+  default_batch_size: 1000  
+  default_max_pool_size: 10  
+
+  # Default sync settings  
+  default_sync_strategy: "full"  
+
+  # Performance tuning  
+  max_concurrent_collections: 5  
+  memory_gc_interval: 10000  
+
+  # Retry settings  
+  connection_retry_attempts: 3  
+  connection_retry_delay: "30s"  
+
+  # Logging  
+  log_level: "info"  
+  log_slow_queries: true  
+  slow_query_threshold: "5s"

plugins/connectors/mongodb/config.go

@@ -0,0 +1,108 @@
+/* Copyright © INFINI LTD. All rights reserved.
+ * Web: https://infinilabs.com
+ * Email: hello#infini.ltd */
+
+package mongodb
+
+import (
+	"fmt"
+	"time"
+)
+
+// Config defines the configuration for the MongoDB connector
+type Config struct {
+	// Connection configuration
+	ConnectionURI string `config:"connection_uri"`
+	Host          string `config:"host"`
+	Port          int    `config:"port"`
+	Username      string `config:"username"`
+	Password      string `config:"password"`
+	Database      string `config:"database"`
+	AuthDatabase  string `config:"auth_database"`
+
+	// Replica set and sharding configuration
+	ReplicaSet     string `config:"replica_set"`
+	ReadPreference string `config:"read_preference"`
+
+	// TLS/SSL configuration
+	EnableTLS   bool   `config:"enable_tls"`
+	TLSCAFile   string `config:"tls_ca_file"`
+	TLSCertFile string `config:"tls_cert_file"`
+	TLSKeyFile  string `config:"tls_key_file"`
+	TLSInsecure bool   `config:"tls_insecure"`
+
+	// Data filtering configuration
+	Collections []CollectionConfig `config:"collections"`
+
+	// Performance optimization configuration
+	BatchSize   int    `config:"batch_size"`
+	Timeout     string `config:"timeout"`
+	MaxPoolSize int    `config:"max_pool_size"`
+
+	// Sync strategy
+	SyncStrategy   string    `config:"sync_strategy"`
+	TimestampField string    `config:"timestamp_field"`
+	LastSyncTime   time.Time `config:"last_sync_time"`
+}
+
+type CollectionConfig struct {
+	Name           string                 `config:"name"`
+	Filter         map[string]interface{} `config:"filter"`
+	Fields         []string               `config:"fields"`
+	TitleField     string                 `config:"title_field"`
+	ContentField   string                 `config:"content_field"`
+	CategoryField  string                 `config:"category_field"`
+	TagsField      string                 `config:"tags_field"`
+	URLField       string                 `config:"url_field"`
+	TimestampField string                 `config:"timestamp_field"`
+}
+
+func (p *Plugin) setDefaultConfig(config *Config) {
+	if config.BatchSize <= 0 {
+		config.BatchSize = 1000
+	}
+	if config.MaxPoolSize <= 0 {
+		config.MaxPoolSize = 10
+	}
+	if config.Timeout == "" {
+		config.Timeout = "30s"
+	}
+	if config.SyncStrategy == "" {
+		config.SyncStrategy = "full"
+	}
+}
+
+func (p *Plugin) validateConfig(config *Config) error {
+	if config.ConnectionURI == "" {
+		if config.Host == "" {
+			return fmt.Errorf("either connection_uri or host must be specified")
+		}
+		if config.Database == "" {
+			return fmt.Errorf("database must be specified")
+		}
+	}
+
+	if len(config.Collections) == 0 {
+		return fmt.Errorf("at least one collection must be configured")
+	}
+
+	for i, coll := range config.Collections {
+		if coll.Name == "" {
+			return fmt.Errorf("collection[%d].name is required", i)
+		}
+	}
+
+	if config.BatchSize < 0 {
+		return fmt.Errorf("batch_size must be positive")
+	}
+
+	if config.MaxPoolSize < 0 {
+		return fmt.Errorf("max_pool_size must be positive")
+	}
+
+	if config.SyncStrategy != "" && config.SyncStrategy != "full" && config.SyncStrategy != "incremental" {
+		return fmt.Errorf("sync_strategy must be 'full' or 'incremental'")
+	}
+
+	return nil
+}