knowledge-governance/internal/platform/postgres/source_connections.go

package postgres

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

	"knowledge-governance/internal/governance"
)

const listSourceConnectionsSQL = `
SELECT
    c.id,
    c.source_type,
    c.name,
    c.enabled,
    c.schedule <> '{}'::jsonb AS schedule_configured,
    CASE
        WHEN c.last_health->>'status' IN ('healthy', 'degraded', 'unhealthy')
            THEN c.last_health->>'status'
        ELSE 'unknown'
    END AS health_status,
    COUNT(i.id)::int AS item_count,
    c.updated_at,
    c.next_refresh_at,
    scan.id,
    scan.state,
    scan.observed_items,
    scan.unchanged_items,
    scan.added_items,
    scan.updated_items,
    scan.over_limit_items,
    scan.missing_items,
    scan.planned_uploads,
    scan.error_code,
    scan.created_at,
    scan.completed_at
FROM source_connections c
LEFT JOIN source_items i ON i.source_connection_id = c.id
LEFT JOIN LATERAL (
    SELECT id, state, observed_items, unchanged_items, added_items,
           updated_items, over_limit_items, missing_items, planned_uploads,
           error_code, created_at, completed_at
    FROM source_scans
    WHERE source_connection_id = c.id
    ORDER BY created_at DESC, id DESC
    LIMIT 1
) scan ON TRUE
WHERE c.archived_at IS NULL
GROUP BY c.id, c.source_type, c.name, c.enabled, c.schedule, c.last_health, c.updated_at, c.next_refresh_at,
         scan.id, scan.state, scan.observed_items, scan.unchanged_items,
         scan.added_items, scan.updated_items, scan.over_limit_items,
         scan.missing_items, scan.planned_uploads, scan.error_code,
         scan.created_at, scan.completed_at
ORDER BY c.source_type, c.name, c.id
`

func (s *Store) ListSourceConnections(ctx context.Context) (governance.SourceConnectionPage, error) {
	rows, err := s.db.Query(ctx, listSourceConnectionsSQL)
	if err != nil {
		return governance.SourceConnectionPage{}, fmt.Errorf("list source connections: %w", err)
	}
	defer rows.Close()

	page := governance.SourceConnectionPage{Items: make([]governance.SourceConnectionView, 0)}
	for rows.Next() {
		var item governance.SourceConnectionView
		var sourceType, healthStatus string
		var scanID, scanState, scanError *string
		var observed, unchanged, added, updated, overLimit, missing, planned *int
		var scanCreatedAt, scanCompletedAt *time.Time
		if err := rows.Scan(
			&item.ID,
			&sourceType,
			&item.Name,
			&item.Enabled,
			&item.ScheduleConfigured,
			&healthStatus,
			&item.ItemCount,
			&item.UpdatedAt,
			&item.NextRefreshAt,
			&scanID,
			&scanState,
			&observed,
			&unchanged,
			&added,
			&updated,
			&overLimit,
			&missing,
			&planned,
			&scanError,
			&scanCreatedAt,
			&scanCompletedAt,
		); err != nil {
			return governance.SourceConnectionPage{}, fmt.Errorf("scan source connection: %w", err)
		}
		item.SourceType = governance.SourceType(sourceType)
		item.HealthStatus = governance.SourceHealthStatus(healthStatus)
		if !item.SourceType.Valid() || !item.HealthStatus.Valid() || item.ItemCount < 0 {
			return governance.SourceConnectionPage{}, fmt.Errorf("invalid source connection projection")
		}
		if scanID != nil {
			if scanState == nil || observed == nil || unchanged == nil || added == nil || updated == nil ||
				overLimit == nil || missing == nil || planned == nil || scanError == nil || scanCreatedAt == nil {
				return governance.SourceConnectionPage{}, fmt.Errorf("incomplete source scan projection")
			}
			state := governance.SourceScanState(*scanState)
			if !state.Valid() || *observed < 0 || *unchanged < 0 || *added < 0 || *updated < 0 || *overLimit < 0 || *missing < 0 || *planned < 0 {
				return governance.SourceConnectionPage{}, fmt.Errorf("invalid source scan projection")
			}
			item.LastScan = &governance.SourceScanSummary{
				ID: *scanID, State: state, ObservedItems: *observed, UnchangedItems: *unchanged,
				AddedItems: *added, UpdatedItems: *updated, OverLimitItems: *overLimit,
				MissingItems: *missing, PlannedUploads: *planned, ErrorCode: *scanError,
				CreatedAt: *scanCreatedAt, CompletedAt: scanCompletedAt,
			}
		}
		page.Items = append(page.Items, item)
		page.Summary.Total++
		if item.Enabled {
			page.Summary.Enabled++
		} else {
			page.Summary.Disabled++
		}
		switch item.HealthStatus {
		case governance.SourceHealthHealthy:
			page.Summary.Healthy++
		case governance.SourceHealthDegraded:
			page.Summary.Degraded++
		case governance.SourceHealthUnhealthy:
			page.Summary.Unhealthy++
		case governance.SourceHealthUnknown:
			page.Summary.Unknown++
		}
	}
	if err := rows.Err(); err != nil {
		return governance.SourceConnectionPage{}, fmt.Errorf("iterate source connections: %w", err)
	}
	return page, nil
}

// V1 per-source scheduler. One statement claims every enabled, due connection
// (FOR UPDATE OF c SKIP LOCKED so parallel dispatchers never share a row),
// skips sources whose manual or scheduled refresh job is still pending/running,
// enqueues one source.catalog.refresh per remaining source, and pushes each
// claimed source's next_refresh_at out by the interval configured on its
// ACTIVE revision. Enqueue and horizon-advance commit atomically in this single
// statement: a crash before commit rolls both back, and two racing dispatcher
// ticks serialize on the row lock, so the same window cannot double-enqueue.
// The deterministic idempotency key (scheduled-refresh:<id>:<window>) is a
// second belt. Disabling a source stops it from ever being claimed again while
// every historical scan, snapshot and job stays intact. Sources without a
// configured interval keep next_refresh_at NULL and are never due (manual-only).
const claimDueSourceRefreshesSQL = `
WITH due AS (
    SELECT c.id, c.name, c.source_type, c.next_refresh_at,
           COALESCE((r.schedule->>'refresh_interval_seconds')::INTEGER, 0) AS interval_seconds
    FROM source_connections AS c
    LEFT JOIN source_connection_revisions AS r
        ON r.id = c.current_revision_id AND r.state = 'active'
    WHERE c.enabled AND c.source_type <> 'manual' AND c.archived_at IS NULL
      AND c.next_refresh_at IS NOT NULL AND c.next_refresh_at <= $1
    ORDER BY c.next_refresh_at, c.id
    FOR UPDATE OF c SKIP LOCKED
),
enqueueable AS (
    SELECT d.* FROM due AS d
    WHERE NOT EXISTS (
        SELECT 1 FROM jobs
        WHERE jobs.type = 'source.catalog.refresh'
          AND jobs.state IN ('pending', 'running')
          AND jobs.payload->>'connection_id' = d.id
    )
),
inserted AS (
    INSERT INTO jobs (
        id, type, payload, state, priority, available_at,
        max_attempts, idempotency_key, created_at, updated_at
    )
    SELECT 'job-' || replace(gen_random_uuid()::text, '-', ''), 'source.catalog.refresh',
           jsonb_build_object(
               'connection_id', e.id, 'source_type', e.source_type, 'source_name', e.name,
               'stage', 'queued', 'progress', 0, 'message', '计划刷新已排队', 'processed', 0, 'total', 0),
           'pending', 20, $1, 3,
           'scheduled-refresh:' || e.id || ':' || FLOOR(EXTRACT(EPOCH FROM e.next_refresh_at))::BIGINT::TEXT,
           $1, $1
    FROM enqueueable AS e
    ON CONFLICT (idempotency_key) DO NOTHING
    RETURNING id
),
advanced AS (
    UPDATE source_connections AS c
    SET next_refresh_at = $1 + make_interval(secs => GREATEST(d.interval_seconds, 300)),
        last_scheduled_at = $1
    FROM due AS d
    WHERE c.id = d.id
    RETURNING c.id
)
SELECT (SELECT COUNT(*)::INTEGER FROM advanced), (SELECT COUNT(*)::INTEGER FROM inserted)
`

func (s *Store) ClaimDueSourceRefreshes(ctx context.Context, now time.Time) (claimed int, enqueued int, err error) {
	if now.IsZero() {
		return 0, 0, errors.New("claim due source refreshes requires a time")
	}
	if err := s.db.QueryRow(ctx, claimDueSourceRefreshesSQL, now.UTC()).Scan(&claimed, &enqueued); err != nil {
		return 0, 0, fmt.Errorf("claim due source refreshes: %w", err)
	}
	return claimed, enqueued, nil
}

// NextRefreshAtFor derives the scheduler horizon from the effective schedule;
// it is the ONLY place next_refresh_at values are computed from user input.
func NextRefreshAtFor(enabled bool, scheduleRaw []byte, from time.Time) (*time.Time, error) {
	if !enabled {
		return nil, nil
	}
	interval, err := governance.ParseSourceSchedule(scheduleRaw)
	if err != nil {
		return nil, err
	}
	if interval <= 0 {
		return nil, nil
	}
	next := from.UTC().Add(interval)
	return &next, nil
}

// ArchiveSourceConnection 把来源连接从平台上「去掉入口」：置 archived_at 之后它会
// 从数据源配置、来源与目录、计划刷新与扫描里全部消失，但行、目录快照、发布记录与
// 审计原样保留 —— 恢复（RestoreSourceConnection 或重新登记同一来源）即可原地复活。
//
// 幂等：已归档时不改写归档时间与归档人，只返回成功（重复点击不漂移语义）。同时停用并
// 清空 next_refresh_at：即使有人手工把 archived_at 改回来，也不会带着旧调度立刻跑起来。
func (s *Store) ArchiveSourceConnection(ctx context.Context, connectionID, actor string, now time.Time) error {
	if strings.TrimSpace(connectionID) == "" || now.IsZero() {
		return errors.New("archive source connection: id and time are required")
	}
	if strings.TrimSpace(actor) == "" {
		return errors.New("archive source connection: actor is required")
	}
	tag, err := s.db.Exec(ctx, `
UPDATE source_connections
SET archived_at = COALESCE(archived_at, $2),
    archived_by = CASE WHEN archived_at IS NULL THEN $3 ELSE archived_by END,
    enabled = FALSE,
    next_refresh_at = NULL,
    updated_at = $2,
    updated_by = $3
WHERE id = $1`, connectionID, now.UTC(), actor)
	if err != nil {
		return fmt.Errorf("archive source connection: %w", err)
	}
	if tag.RowsAffected() == 0 {
		return ErrSourceConnectionNotFound
	}
	return nil
}

// RestoreSourceConnection 撤销归档：来源重新出现在列表里，但**保持停用**，需要管理员
// 再显式启用（归档时被清掉的调度不会悄悄回来）。幂等：未归档时视为成功；只有连接真的
// 不存在才报 ErrSourceConnectionNotFound。
func (s *Store) RestoreSourceConnection(ctx context.Context, connectionID, actor string, now time.Time) error {
	if strings.TrimSpace(connectionID) == "" || now.IsZero() {
		return errors.New("restore source connection: id and time are required")
	}
	if strings.TrimSpace(actor) == "" {
		return errors.New("restore source connection: actor is required")
	}
	tag, err := s.db.Exec(ctx, `
UPDATE source_connections
SET archived_at = NULL, archived_by = '', updated_at = $2, updated_by = $3
WHERE id = $1 AND archived_at IS NOT NULL`, connectionID, now.UTC(), actor)
	if err != nil {
		return fmt.Errorf("restore source connection: %w", err)
	}
	if tag.RowsAffected() > 0 {
		return nil
	}
	var exists bool
	if err := s.db.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM source_connections WHERE id = $1)`, connectionID).Scan(&exists); err != nil {
		return fmt.Errorf("restore source connection: %w", err)
	}
	if !exists {
		return ErrSourceConnectionNotFound
	}
	return nil
}