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

package postgres

import (
	"context"
	"errors"
	"fmt"
	"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
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.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
}