┌───────────────────────────────────────────────────────────────────────────┐
│                          数据源导入架构                                     │
│                                                                           │
│   ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐   │
│   │   飞书    │  │  Notion  │  │Confluence│  │  GitHub  │  │  其他...  │   │
│   │  (Wiki)  │  │          │  │          │  │          │  │          │   │
│   └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘   │
│        │ API         │ API         │ API         │ API         │          │
│   ─────┼─────────────┼────────────┼─────────────┼─────────────┼──────    │
│        ▼             ▼            ▼             ▼             ▼          │
│   ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐  │
│   │  Feishu  │  │  Notion  │  │Confluence│  │  GitHub  │  │   ...    │  │
│   │Connector │  │Connector │  │Connector │  │Connector │  │Connector │  │
│   └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘  │
│        │             │             │              │             │   连接  │
│   ─────┴─────────────┴─────────────┴──────────────┴─────────────┘   器层  │
│                      │                                                    │
│                      ▼                                                    │
│   ┌──────────────────────────────────────────────┐                        │
│   │         ConnectorRegistry                    │   连接器注册表          │
│   │   · 按类型查找连接器实例                        │                       │
│   │   · 连接器元数据（名称、认证方式、能力）          │                       │
│   └──────────────────┬───────────────────────────┘                        │
│                      │                                                    │
│   ───────────────────┼────────────────────────────────────────────────    │
│                      ▼                                                    │
│   ┌──────────────────────────────────────────────┐                        │
│   │       DataSourceService (业务逻辑层)          │                        │
│   │                                              │                        │
│   │  ┌────────────────────────────────────────┐  │                        │
│   │  │ · 数据源 CRUD                          │  │                        │
│   │  │ · 连接验证 (ValidateConnection)        │  │                        │
│   │  │ · 资源发现 (ListAvailableResources)    │  │                        │
│   │  │ · 手动同步 (ManualSync)                │  │                        │
│   │  │ · 暂停/恢复                            │  │                        │
│   │  │ · 同步执行 (ProcessSync)               │  │                        │
│   │  │ · 内容入库 (ingestItem)                │  │                        │
│   │  └────────────────────────────────────────┘  │                        │
│   └──────────────────┬───────────────────────────┘                        │
│                      │                                                    │
│   ───────────────────┼────────────────────────────────────────────────    │
│                      ▼                                                    │
│   ┌──────────────────────────────────┐  ┌────────────────────────────┐    │
│   │      Scheduler (Cron 调度器)     │  │    Task Queue (asynq)     │    │
│   │  · robfig/cron（含秒）           │  │  · 异步同步任务            │    │
│   │  · DB 加载活跃数据源             │  │  · Redis / Lite 模式      │    │
│   │  · 任务去重 (TaskID + Running)   │  │  · TypeDataSourceSync     │    │
│   └──────────────────────────────────┘  └────────────────────────────┘    │
│                      │                                                    │
│   ───────────────────┼────────────────────────────────────────────────    │
│                      ▼                                                    │
│   ┌──────────────────────────────────────────────────────┐                │
│   │            WeKnora Core (知识入库管道)                │                │
│   │  KnowledgeService · KnowledgeBaseService             │                │
│   │  文档解析 → 分块 → 向量化 → 索引                      │                │
│   └──────────────────────────────────────────────────────┘                │
└───────────────────────────────────────────────────────────────────────────┘

CREATE TABLE data_sources (
    id                      VARCHAR(36) PRIMARY KEY,
    tenant_id               BIGINT NOT NULL,
    knowledge_base_id       VARCHAR(36) NOT NULL,
    name                    VARCHAR(255) NOT NULL DEFAULT '',
    type                    VARCHAR(50) NOT NULL,          -- 连接器类型
    config                  JSONB NOT NULL DEFAULT '{}',   -- 加密的凭证和配置
    sync_schedule           VARCHAR(100) DEFAULT '',       -- Cron 表达式（6 段，含秒）
    sync_mode               VARCHAR(20) DEFAULT 'incremental',  -- 'incremental' | 'full'
    status                  VARCHAR(20) DEFAULT 'active',  -- 'active' | 'paused' | 'error' | 'deleted'
    conflict_strategy       VARCHAR(20) DEFAULT 'overwrite',    -- 'overwrite' | 'skip'
    sync_deletions          BOOLEAN DEFAULT true,
    last_sync_at            TIMESTAMPTZ,
    last_sync_cursor        JSONB,
    last_sync_result        JSONB,
    error_message           TEXT DEFAULT '',
    sync_log_retention_days INTEGER DEFAULT 30,
    created_at              TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at              TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    deleted_at              TIMESTAMPTZ
);

{
  "type": "feishu",
  "credentials": {
    "app_id": "cli_xxx",
    "app_secret": "xxx"
  },
  "resource_ids": ["space_id_1", "space_id_2"],
  "settings": {}
}

CREATE TABLE sync_logs (
    id              VARCHAR(36) PRIMARY KEY,
    data_source_id  VARCHAR(36) NOT NULL REFERENCES data_sources(id) ON DELETE CASCADE,
    tenant_id       BIGINT NOT NULL,
    status          VARCHAR(20) NOT NULL,    -- 'running' | 'success' | 'partial' | 'failed' | 'canceled'
    started_at      TIMESTAMPTZ NOT NULL,
    finished_at     TIMESTAMPTZ,
    items_total     INTEGER DEFAULT 0,
    items_created   INTEGER DEFAULT 0,
    items_updated   INTEGER DEFAULT 0,
    items_deleted   INTEGER DEFAULT 0,
    items_skipped   INTEGER DEFAULT 0,
    items_failed    INTEGER DEFAULT 0,
    error_message   TEXT DEFAULT '',
    result          JSONB,                   -- 详细同步结果
    created_at      TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

[
  {
    "type": "feishu",
    "name": "飞书",
    "description": "Import documents from Feishu/Lark Wiki spaces",
    "icon": "feishu",
    "priority": 1,
    "auth_type": "oauth2",
    "capabilities": ["incremental", "deletion_sync"]
  }
]

{
  "type": "feishu",
  "credentials": {
    "app_id": "cli_xxx",
    "app_secret": "xxx"
  }
}

type SyncCursor struct {
    LastSyncTime    *time.Time             // 上次同步时间
    ConnectorCursor map[string]interface{} // 连接器自定义状态
    LastSchemaHash  string                 // Schema 变更检测
}

type FetchedItem struct {
    ExternalID       string            // 外部平台的文档唯一标识
    Title            string            // 文档标题
    Content          []byte            // 文档内容（二进制）
    ContentType      string            // MIME 类型
    FileName         string            // 文件名（含扩展名）
    URL              string            // 原始 URL
    UpdatedAt        *time.Time        // 最后更新时间
    Metadata         map[string]string // 元数据（来源、作者等）
    IsDeleted        bool              // 标记为已删除（增量同步）
    SourceResourceID string            // 所属资源 ID
}

type Resource struct {
    ExternalID  string            // 外部 ID
    Name        string            // 名称
    Type        string            // 类型（如 wiki_space）
    Description string            // 描述
    URL         string            // 外部链接
    ModifiedAt  *time.Time        // 最后修改时间
    ParentID    string            // 父节点 ID（支持树形结构）
    Metadata    map[string]string // 附加信息
}

手动触发 / Cron 定时触发
        │
        ▼
┌─ 创建 SyncLog (status: running) ──────────────────┐
│  入队异步任务 (asynq: TypeDataSourceSync)           │
│  · 手动同步 → 队列: default                        │
│  · 定时同步 → 队列: low, TaskID 去重               │
└──────────────────────────┬────────────────────────┘
                           │
                           ▼
┌─ ProcessSync (Worker 异步执行) ───────────────────┐
│  1. 解析任务体 (DataSourceSyncPayload)             │
│  2. 加载数据源配置                                  │
│  3. 解密凭证 → 获取 Connector                      │
│  4. 设置租户上下文                                  │
│  5. 创建/获取自动标签 (连接器名·数据源名)            │
│  6. 判断同步模式:                                   │
│     ├─ ForceFull 或 SyncModeFull → FetchAll        │
│     └─ SyncModeIncremental → FetchIncremental      │
│  7. 遍历 FetchedItem 列表:                         │
│     └─ ingestItem → KnowledgeService               │
│  8. 更新 SyncLog (计数、状态、耗时)                 │
│  9. 更新 DataSource (游标、最后同步时间、结果)       │
└──────────────────────────────────────────────────┘

FetchedItem
    │
    ├─ IsDeleted = true?
    │     └─ 按 external_id 查找已有知识 → 软删除
    │
    ├─ 有文件内容 (Content)?
    │     └─ CreateKnowledgeFromFile
    │        · 按 external_id 查找已有知识
    │        · 已存在 → 删除旧版本 → 重建
    │        · 不存在 → 新建
    │        · metadata: external_id, source_resource_id, datasource_id
    │        · channel: 连接器类型 (如 "feishu")
    │        · 自动关联标签
    │
    └─ 仅有 URL (无 Content)?
          └─ CreateKnowledgeFromURL
             · 同上查重逻辑

创建数据源 (前端向导)
        │
        ▼
┌─ DataSourceService.Create ────────────────────┐
│  1. 验证知识库存在                              │
│  2. 获取对应 Connector                         │
│  3. 验证配置和凭证 (connector.Validate)         │
│  4. 加密凭证写库                               │
│  5. 若有 SyncSchedule → 注册 Cron 任务         │
└───────────────────────────────────────────────┘

暂停数据源:
  status → paused, 移除 Cron 任务

恢复数据源:
  status → active, 重新注册 Cron 任务

删除数据源:
  deleted_at = NOW(), 移除 Cron 任务, 级联删除 sync_logs

连接错误:
  ValidateConnection 失败 → status → error, 记录 error_message
  ValidateConnection 成功且原为 error → status → active, 清空 error_message

type Connector interface {
    Type() string
    Validate(ctx context.Context, config *types.DataSourceConfig) error
    ListResources(ctx context.Context, config *types.DataSourceConfig) ([]types.Resource, error)
    FetchAll(ctx context.Context, config *types.DataSourceConfig, resourceIDs []string) ([]types.FetchedItem, error)
    FetchIncremental(ctx context.Context, config *types.DataSourceConfig, cursor *types.SyncCursor) ([]types.FetchedItem, *types.SyncCursor, error)
}

type DataSourceService interface {
    CreateDataSource(ctx context.Context, ds *types.DataSource) (*types.DataSource, error)
    GetDataSource(ctx context.Context, id string) (*types.DataSource, error)
    ListDataSources(ctx context.Context, knowledgeBaseID string) ([]*types.DataSource, error)
    UpdateDataSource(ctx context.Context, ds *types.DataSource) (*types.DataSource, error)
    DeleteDataSource(ctx context.Context, id string) error
    ValidateConnection(ctx context.Context, dataSourceID string) error
    ValidateCredentials(ctx context.Context, connectorType string, credentials map[string]interface{}) error
    ListAvailableResources(ctx context.Context, dataSourceID string) ([]types.Resource, error)
    ManualSync(ctx context.Context, dataSourceID string) (*types.SyncLog, error)
    PauseDataSource(ctx context.Context, dataSourceID string) error
    ResumeDataSource(ctx context.Context, dataSourceID string) error
    GetSyncLogs(ctx context.Context, dataSourceID string, limit, offset int) ([]*types.SyncLog, error)
    GetSyncLog(ctx context.Context, syncLogID string) (*types.SyncLog, error)
    ProcessSync(ctx context.Context, task *asynq.Task) error
}

type ConnectorMetadata struct {
    Type         string   `json:"type"`
    Name         string   `json:"name"`
    Description  string   `json:"description"`
    Icon         string   `json:"icon"`
    Priority     int      `json:"priority"`
    AuthType     string   `json:"auth_type"`
    Capabilities []string `json:"capabilities"`
}

App ID + App Secret
        │
        ▼
  POST /open-apis/auth/v3/tenant_access_token/internal
        │
        ▼
  获取 Tenant Access Token（有效期 2 小时）
        │
        ▼
  Token 缓存，提前 5 分钟刷新

GET /open-apis/wiki/v2/spaces (分页, page_size=50)

对每个选中的 Wiki Space:
  1. 递归列举空间下所有节点 (ListAllWikiNodesRecursive)
     └─ GET /open-apis/wiki/v2/spaces/{space_id}/nodes (parent 为空 → 顶层)
     └─ 对 has_child=true 的节点递归
  2. 对每个节点:
     └─ fetchNodeContent → 判断文档类型 → 导出/下载

1. 加载上次游标 (feishuCursor)
   └─ SpaceNodeTimes[space_id][node_token] = 上次记录的编辑时间

2. 对每个 Space 递归列举所有节点（全树遍历）

3. 对比变更:
   ├─ 新节点（游标中不存在）→ 拉取内容
   ├─ obj_edit_time 变更（优先）或 node_edit_time 变更 → 拉取内容
   ├─ 编辑时间未变 → 跳过
   └─ 游标中存在但本次遍历不存在 → 标记为 IsDeleted

4. 返回新游标（含本轮每个节点的最新编辑时间）

1. CreateExportTask → 创建导出任务
2. 轮询 GetExportTaskStatus（间隔 2 秒，最长约 60 秒）
3. DownloadExportFile → 用 file_token 下载导出文件
4. 清理文件名 (sanitizeFileName) + 补全扩展名

DownloadDriveFile → GET /drive/v1/files/{token}/download

秒 分 时 日 月 星期

服务启动
    │
    ▼
Scheduler.Start(ctx)
    │
    ├─ 1. 从 DB 加载所有活跃数据源 (FindActive)
    │     条件: status = active, sync_schedule != '', deleted_at IS NULL
    │
    ├─ 2. 为每个数据源注册 Cron Entry
    │
    └─ 3. 启动 Cron Runner

Cron 触发时:
    ├─ 检查是否有运行中的同步 (HasRunningSync)
    │     是 → 跳过本次触发
    │
    ├─ 创建 SyncLog (status: running)
    │
    └─ 入队 asynq 任务
          队列: low
          TaskID: dssync:<dsID>:<UTC minute>（去重）

package myplatform

type Connector struct {
    // 连接器配置
}

func NewConnector() *Connector {
    return &Connector{}
}

func (c *Connector) Type() string {
    return types.ConnectorTypeMyPlatform
}

func (c *Connector) Validate(ctx context.Context, config *types.DataSourceConfig) error {
    // 验证凭证：尝试获取 Token 或调用 API
}

func (c *Connector) ListResources(ctx context.Context, config *types.DataSourceConfig) ([]types.Resource, error) {
    // 列出可选资源（工作区、空间、文件夹等）
}

func (c *Connector) FetchAll(ctx context.Context, config *types.DataSourceConfig, resourceIDs []string) ([]types.FetchedItem, error) {
    // 全量拉取所有文档
}

func (c *Connector) FetchIncremental(ctx context.Context, config *types.DataSourceConfig, cursor *types.SyncCursor) ([]types.FetchedItem, *types.SyncCursor, error) {
    // 增量拉取变更（可复用 FetchAll 逻辑 + 编辑时间对比）
}

internal/datasource/connector/myplatform/
├── types.go        # 平台 API 类型定义
├── client.go       # API 客户端（Token 管理、HTTP 调用）
└── connector.go    # Connector 接口实现

func initConnectorRegistry() *datasource.ConnectorRegistry {
    registry := datasource.NewConnectorRegistry()
    registry.Register(feishuConnector.NewConnector())
    registry.Register(myplatform.NewConnector()) // 新增
    return registry
}

types.ConnectorTypeMyPlatform: {
    Type:         types.ConnectorTypeMyPlatform,
    Name:         "My Platform",
    Description:  "Import documents from My Platform",
    Icon:         "myplatform",
    Priority:     10,
    AuthType:     "api_key",
    Capabilities: []string{"incremental"},
},

const ConnectorTypeMyPlatform = "myplatform"