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

package postgres

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"

	"knowledge-governance/internal/governance"
)

// V1 source settings revisions: every edit becomes a validated revision;
// activation is the ONLY way it takes effect and swaps atomically with the
// previous active revision, syncing the runtime snapshot row that the
// scheduler, FKs and existing readers use. next_refresh_at is always derived
// server-side from the active schedule — clients never write it directly.

var ErrInvalidSourceSettings = errors.New("invalid source settings")

// ErrSourceConnectionNotFound marks an unknown connection or revision.
var ErrSourceConnectionNotFound = errors.New("source connection not found")

const sourceSettingsScheduleFloorSeconds = 300

type SourceSettingsDraft struct {
	SourceType string          `json:"source_type"`
	Name       string          `json:"name"`
	BaseURL    string          `json:"base_url"`
	SecretRef  string          `json:"secret_ref"`
	Enabled    bool            `json:"enabled"`
	Schedule   json.RawMessage `json:"schedule"`
	Operator   string          `json:"operator"`
}

func (d SourceSettingsDraft) validate() error {
	switch d.SourceType {
	case "pp", "alist", "dingtalk", "manual":
	case "":
		return fmt.Errorf("%w: source type is required", ErrInvalidSourceSettings)
	default:
		return fmt.Errorf("%w: unknown source type", ErrInvalidSourceSettings)
	}
	if d.Name != strings.TrimSpace(d.Name) || d.Name == "" || len(d.Name) > 255 {
		return fmt.Errorf("%w: name must be trimmed and at most 255 characters", ErrInvalidSourceSettings)
	}
	if len(d.BaseURL) > 2048 || strings.ContainsAny(d.BaseURL, "\x00\r\n\t") {
		return fmt.Errorf("%w: invalid base URL", ErrInvalidSourceSettings)
	}
	if len(d.SecretRef) > 255 || strings.ContainsAny(d.SecretRef, "\x00\r\n\t") {
		return fmt.Errorf("%w: invalid secret reference", ErrInvalidSourceSettings)
	}
	if d.Operator == "" || d.Operator != strings.TrimSpace(d.Operator) || len(d.Operator) > 128 {
		return fmt.Errorf("%w: operator is required", ErrInvalidSourceSettings)
	}
	if _, err := parseScheduleForValidation(d.Schedule); err != nil {
		return err
	}
	return nil
}

func parseScheduleForValidation(raw json.RawMessage) (governance.SourceSchedule, error) {
	if len(raw) == 0 || string(raw) == "" {
		return governance.SourceSchedule{}, fmt.Errorf("%w: schedule is required", ErrInvalidSourceSettings)
	}
	interval, err := governance.ParseSourceSchedule(raw)
	if err != nil {
		return governance.SourceSchedule{}, fmt.Errorf("%w: %v", ErrInvalidSourceSettings, err)
	}
	var decoded governance.SourceSchedule
	if err := json.Unmarshal(raw, &decoded); err != nil {
		return governance.SourceSchedule{}, fmt.Errorf("%w: schedule must be an object", ErrInvalidSourceSettings)
	}
	_ = interval
	return decoded, nil
}

// CreateSourceSettingsRevision stores a validated draft for an existing
// connection without touching the active snapshot.
func (s *Store) CreateSourceSettingsRevision(ctx context.Context, connectionID string, draft SourceSettingsDraft, now time.Time) (string, error) {
	if !validShortID(connectionID) || now.IsZero() {
		return "", ErrInvalidSourceSettings
	}
	if err := draft.validate(); err != nil {
		return "", err
	}
	return insertSourceSettingsRevisionTx(ctx, s.db, connectionID, draft, now)
}

// ActivateSourceSettingsRevision atomically promotes a validated revision:
// retire the previous active, sync the runtime snapshot (name/base_url/
// secret_ref/enabled/schedule), and DERIVE next_refresh_at from the schedule.
// Disabling clears the horizon immediately: no new scheduled scans or jobs.
func (s *Store) ActivateSourceSettingsRevision(ctx context.Context, connectionID, revisionID string, now time.Time) error {
	if !validShortID(connectionID) || !validShortID(revisionID) || now.IsZero() {
		return ErrInvalidSourceSettings
	}
	beginner, ok := s.db.(interface {
		Begin(context.Context) (pgx.Tx, error)
	})
	if !ok {
		return errors.New("source settings transactions are unavailable")
	}
	tx, err := beginner.Begin(ctx)
	if err != nil {
		return fmt.Errorf("begin source settings activation: %w", err)
	}
	defer func() { _ = tx.Rollback(ctx) }()
	if err := activateSourceSettingsRevisionTx(ctx, tx, connectionID, revisionID, now); err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func activateSourceSettingsRevisionTx(ctx context.Context, tx pgx.Tx, connectionID, revisionID string, now time.Time) error {
	var (
		state, sourceType, name, baseURL, secretRef, revisionAuthor string
		enabled                                                     bool
		schedule                                                    []byte
	)
	err := tx.QueryRow(ctx, `
SELECT state, source_type, name, base_url, secret_ref, enabled, schedule::text, created_by
FROM source_connection_revisions
WHERE id = $1 AND connection_id = $2 FOR UPDATE`, revisionID, connectionID).
		Scan(&state, &sourceType, &name, &baseURL, &secretRef, &enabled, &schedule, &revisionAuthor)
	if errors.Is(err, pgx.ErrNoRows) {
		return ErrSourceConnectionNotFound
	}
	if err != nil {
		return fmt.Errorf("load source settings revision: %w", err)
	}
	if state == "active" {
		return nil // replay: already effective
	}
	if state != "validated" {
		return fmt.Errorf("%w: only validated revisions can be activated", ErrInvalidSourceSettings)
	}
	interval, err := governance.ParseSourceSchedule(schedule)
	if err != nil {
		return fmt.Errorf("%w: %v", ErrInvalidSourceSettings, err)
	}
	// manual sources are import-only: no live refresh exists, so they can never
	// carry a schedule horizon (the dispatcher would enqueue jobs that fail).
	if sourceType == "manual" && interval > 0 {
		return fmt.Errorf("%w: manual sources cannot be scheduled", ErrInvalidSourceSettings)
	}
	var nextRefresh any
	if enabled && interval > 0 {
		nextRefresh = now.UTC().Add(interval)
	}
	// Guard against a second activation racing in while we hold this row:
	// retiring the current active happens under the connection row lock.
	var lockID string
	if err := tx.QueryRow(ctx, `SELECT id FROM source_connections WHERE id = $1 FOR UPDATE`, connectionID).Scan(&lockID); err != nil {
		return fmt.Errorf("lock source connection: %w", err)
	}
	if _, err := tx.Exec(ctx, `
UPDATE source_connection_revisions
SET state = 'retired', retired_at = $3
WHERE connection_id = $1 AND state = 'active' AND id <> $2`, connectionID, revisionID, now.UTC()); err != nil {
		return fmt.Errorf("retire previous revision: %w", err)
	}
	if _, err := tx.Exec(ctx, `
UPDATE source_connection_revisions
SET state = 'active', activated_at = $2
WHERE id = $1 AND state = 'validated'`, revisionID, now.UTC()); err != nil {
		return fmt.Errorf("activate revision: %w", err)
	}
	tag, err := tx.Exec(ctx, `
UPDATE source_connections
SET name = $2, base_url = $3, secret_ref = $4, enabled = $5, schedule = $6::jsonb,
    next_refresh_at = $7, current_revision_id = $8, updated_by = $9, updated_at = $10
WHERE id = $1`, connectionID, name, baseURL, secretRef, enabled, string(schedule), nextRefresh, revisionID,
		revisionAuthor, now.UTC())
	if err != nil {
		return fmt.Errorf("sync source connection snapshot: %w", err)
	}
	if tag.RowsAffected() != 1 {
		return ErrSourceConnectionNotFound
	}
	return nil
}

// ListSourceConnectionRevisions returns safe history (never raw secrets —
// secret_ref is a reference, already safe to display to operators).
func (s *Store) ListSourceConnectionRevisions(ctx context.Context, connectionID string) ([]governance.SourceConnectionRevision, error) {
	rows, err := s.db.Query(ctx, `
SELECT id, revision_number, state, name, base_url, secret_ref, enabled,
       COALESCE(schedule->>'refresh_interval_seconds','') AS interval,
       created_by, created_at, validated_at, activated_at, retired_at
FROM source_connection_revisions
WHERE connection_id = $1
ORDER BY revision_number DESC`, connectionID)
	if err != nil {
		return nil, fmt.Errorf("list source connection revisions: %w", err)
	}
	defer rows.Close()
	items := []governance.SourceConnectionRevision{}
	for rows.Next() {
		var item governance.SourceConnectionRevision
		var interval string
		var validatedAt, activatedAt, retiredAt *time.Time
		if err := rows.Scan(&item.ID, &item.RevisionNumber, &item.State, &item.Name, &item.BaseURL, &item.SecretRef,
			&item.Enabled, &interval, &item.CreatedBy, &item.CreatedAt, &validatedAt, &activatedAt, &retiredAt); err != nil {
			return nil, fmt.Errorf("scan source connection revision: %w", err)
		}
		item.RefreshIntervalSeconds = interval
		item.ValidatedAt = validatedAt
		item.ActivatedAt = activatedAt
		item.RetiredAt = retiredAt
		items = append(items, item)
	}
	return items, rows.Err()
}

// C4 idempotent write paths for source settings (V1).

// CreateSourceSettingsRevisionWithOperation stores the validated draft once per
// operation; a replay returns the originally created revision id.
func (s *Store) CreateSourceSettingsRevisionWithOperation(ctx context.Context, connectionID string, draft SourceSettingsDraft, operationID string, now time.Time) (string, bool, error) {
	fingerprint, err := operationFingerprint(struct {
		ConnectionID string              `json:"connection_id"`
		Draft        SourceSettingsDraft `json:"draft"`
	}{ConnectionID: connectionID, Draft: draft})
	if err != nil {
		return "", false, err
	}
	resultJSON, replayed, err := s.executeOperation(ctx, "source.settings.create", operationID, fingerprint, now, func(tx pgx.Tx) (json.RawMessage, error) {
		revisionID, err := insertSourceSettingsRevisionTx(ctx, tx, connectionID, draft, now)
		if err != nil {
			return nil, err
		}
		return json.Marshal(revisionID)
	})
	if err != nil {
		return "", false, err
	}
	var revisionID string
	if err := json.Unmarshal(resultJSON, &revisionID); err != nil {
		return "", false, fmt.Errorf("decode stored source settings revision: %w", err)
	}
	return revisionID, replayed, nil
}

// ActivateSourceSettingsRevisionWithOperation activates the validated revision
// exactly once per operation; replays report "replayed" without re-syncing.
func (s *Store) ActivateSourceSettingsRevisionWithOperation(ctx context.Context, connectionID, revisionID, operationID string, now time.Time) (bool, error) {
	fingerprint, err := operationFingerprint(struct {
		ConnectionID string `json:"connection_id"`
		RevisionID   string `json:"revision_id"`
	}{ConnectionID: connectionID, RevisionID: revisionID})
	if err != nil {
		return false, err
	}
	_, replayed, err := s.executeOperation(ctx, "source.settings.activate", operationID, fingerprint, now, func(tx pgx.Tx) (json.RawMessage, error) {
		if err := activateSourceSettingsRevisionTx(ctx, tx, connectionID, revisionID, now); err != nil {
			return nil, err
		}
		return json.RawMessage(`{}`), nil
	})
	return replayed, err
}

type revisionRowQueryer interface {
	QueryRow(context.Context, string, ...any) pgx.Row
}

func insertSourceSettingsRevisionTx(ctx context.Context, db revisionRowQueryer, connectionID string, draft SourceSettingsDraft, now time.Time) (string, error) {
	if err := draft.validate(); err != nil {
		return "", err
	}
	revisionID, err := newOpaqueID("scr-")
	if err != nil {
		return "", err
	}
	err = db.QueryRow(ctx, `
WITH base AS (
    SELECT source_type, name, base_url, secret_ref FROM source_connections WHERE id = $1
),
inherited AS (
    SELECT base.source_type,
           base.name AS current_name, base.base_url, base.secret_ref
    FROM base
),
next_number AS (
    SELECT COALESCE(MAX(revision_number), 0) + 1 AS number
    FROM source_connection_revisions WHERE connection_id = $1
)
INSERT INTO source_connection_revisions (
    id, connection_id, revision_number, state, source_type, name, base_url,
    secret_ref, enabled, schedule, created_by, validated_at
)
SELECT $2, $1, next_number.number, 'validated', inherited.source_type,
       $3,
       CASE WHEN $4 = '' THEN inherited.base_url ELSE $4 END,
       CASE WHEN $5 = '' THEN inherited.secret_ref ELSE $5 END,
       $6, $7::jsonb, $8, $9
FROM inherited, next_number
RETURNING id`, connectionID, revisionID, draft.Name, draft.BaseURL, draft.SecretRef, draft.Enabled, string(draft.Schedule), draft.Operator, now.UTC()).Scan(&revisionID)
	if errors.Is(err, pgx.ErrNoRows) {
		return "", ErrSourceConnectionNotFound
	}
	if err != nil {
		return "", fmt.Errorf("create source settings revision: %w", err)
	}
	return revisionID, nil
}