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

package postgres

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"

	ppconnector "knowledge-governance/internal/connectors/pp"
	"knowledge-governance/internal/governance"
)

var (
	ErrSourceConnectionNotScannable = errors.New("source connection is not scannable")
	ErrSourceScanAlreadyActive      = errors.New("source scan already active")
	ErrSourceScanStateConflict      = errors.New("source scan state conflict")
	ErrSourceScanNotFound           = errors.New("source scan not found")
)

const triggerPPSourceScanSQL = `
WITH target AS (
    SELECT id, source_type
    FROM source_connections
    WHERE id = $1 AND source_type = 'pp'
), active AS (
    SELECT 1
    FROM source_scans
    WHERE source_connection_id = $1
      AND state IN ('pending', 'running')
), created_scan AS (
    INSERT INTO source_scans (
        id, source_connection_id, source_type, state, created_at, updated_at
    )
    SELECT $2, target.id, target.source_type, 'pending', $5, $5
    FROM target
    WHERE NOT EXISTS (SELECT 1 FROM active)
    RETURNING id, source_connection_id, source_type, state, created_at
), created_job AS (
    INSERT INTO jobs (
        id, type, payload, state, priority, available_at,
        max_attempts, idempotency_key, created_at, updated_at
    )
    SELECT $4, 'source.scan.pp', $3::jsonb, 'pending', 20, $5,
           3, 'source-scan:' || created_scan.id, $5, $5
    FROM created_scan
    RETURNING id
)
SELECT created_scan.id, created_scan.source_connection_id,
       created_scan.source_type, created_scan.state, created_scan.created_at
FROM created_scan CROSS JOIN created_job
`

const inspectPPSourceScanBlockSQL = `
SELECT connection.source_type,
       EXISTS (
           SELECT 1 FROM source_scans AS scan
           WHERE scan.source_connection_id = connection.id
             AND scan.state IN ('pending', 'running')
       )
FROM source_connections AS connection
WHERE connection.id = $1
`

const startSourceScanSQL = `
WITH transitioned AS (
    UPDATE source_scans
    SET state = 'running',
        started_at = COALESCE(started_at, $2),
        updated_at = $2
    WHERE id = $1 AND state IN ('pending', 'running')
    RETURNING id, source_connection_id, source_type, state, created_at
), already_completed AS (
    SELECT id, source_connection_id, source_type, state, created_at
    FROM source_scans
    WHERE id = $1 AND state = 'completed'
)
SELECT id, source_connection_id, source_type, state, created_at
FROM transitioned
UNION ALL
SELECT id, source_connection_id, source_type, state, created_at
FROM already_completed
LIMIT 1
`

const completeSourceScanSQL = `
WITH completed AS (
    UPDATE source_scans
    SET state = 'completed', snapshot_sha256 = $2,
        observed_items = $3, unchanged_items = $4,
        added_items = $5, updated_items = $6,
        over_limit_items = $7, missing_items = $8,
        planned_uploads = $9, error_code = '',
        completed_at = $10, updated_at = $10
    WHERE id = $1 AND state = 'running'
    RETURNING source_connection_id
), inserted_changes AS (
    INSERT INTO source_scan_changes (
        id, source_scan_id, source_connection_id, stable_external_id,
        source_item_id, change_type, previous_revision,
        observed_revision, observed_size_bytes, created_at
    )
    SELECT change.id, $1, completed.source_connection_id,
           change.stable_external_id, item.id, change.change_type,
           change.previous_revision, change.observed_revision,
           change.observed_size_bytes, $10
    FROM completed
    CROSS JOIN jsonb_to_recordset($11::jsonb) AS change(
        id TEXT, stable_external_id TEXT, change_type TEXT,
        previous_revision TEXT, observed_revision TEXT,
        observed_size_bytes BIGINT
    )
    LEFT JOIN source_items AS item
      ON item.source_connection_id = completed.source_connection_id
     AND item.stable_external_id = change.stable_external_id
    ON CONFLICT (source_scan_id, stable_external_id) DO NOTHING
    RETURNING id
), healthy AS (
    UPDATE source_connections AS connection
    SET last_health = jsonb_build_object(
            'status', 'healthy', 'checked_at', $10, 'scan_id', $1
        ),
        updated_at = $10
    FROM completed
    WHERE connection.id = completed.source_connection_id
    RETURNING connection.id
)
SELECT (SELECT COUNT(*)::INTEGER FROM completed),
       (SELECT COUNT(*)::INTEGER FROM inserted_changes),
       (SELECT COUNT(*)::INTEGER FROM healthy)
`

const failSourceScanSQL = `
WITH failed AS (
    UPDATE source_scans
    SET state = 'failed', error_code = $2,
        completed_at = $3, updated_at = $3
    WHERE id = $1 AND state = 'running'
    RETURNING source_connection_id
), unhealthy AS (
    UPDATE source_connections AS connection
    SET last_health = jsonb_build_object(
            'status', 'unhealthy', 'checked_at', $3,
            'scan_id', $1, 'error_code', $2
        ),
        updated_at = $3
    FROM failed
    WHERE connection.id = failed.source_connection_id
    RETURNING connection.id
)
SELECT (SELECT COUNT(*)::INTEGER FROM failed),
       (SELECT COUNT(*)::INTEGER FROM unhealthy)
`

func (s *Store) TriggerPPSourceScan(ctx context.Context, connectionID string, now time.Time) (governance.SourceScan, error) {
	if strings.TrimSpace(connectionID) == "" || len(connectionID) > 64 || now.IsZero() {
		return governance.SourceScan{}, ErrSourceConnectionNotScannable
	}
	scanID, err := newOpaqueID("scan-")
	if err != nil {
		return governance.SourceScan{}, err
	}
	jobID, err := newOpaqueID("job-")
	if err != nil {
		return governance.SourceScan{}, err
	}
	payload, err := json.Marshal(struct {
		ScanID string `json:"scan_id"`
	}{ScanID: scanID})
	if err != nil {
		return governance.SourceScan{}, errors.New("encode source scan job")
	}
	scan, err := scanSourceScan(s.db.QueryRow(ctx, triggerPPSourceScanSQL, connectionID, scanID, payload, jobID, now))
	if errors.Is(err, pgx.ErrNoRows) {
		var sourceType string
		var active bool
		inspectErr := s.db.QueryRow(ctx, inspectPPSourceScanBlockSQL, connectionID).Scan(&sourceType, &active)
		if inspectErr != nil && !errors.Is(inspectErr, pgx.ErrNoRows) {
			return governance.SourceScan{}, fmt.Errorf("inspect blocked PP source scan: %w", inspectErr)
		}
		if inspectErr == nil && sourceType == string(governance.SourceTypePP) && active {
			return governance.SourceScan{}, ErrSourceScanAlreadyActive
		}
		return governance.SourceScan{}, ErrSourceConnectionNotScannable
	}
	var pgErr *pgconn.PgError
	if errors.As(err, &pgErr) && pgErr.Code == "23505" {
		return governance.SourceScan{}, ErrSourceScanAlreadyActive
	}
	if err != nil {
		return governance.SourceScan{}, fmt.Errorf("trigger PP source scan: %w", err)
	}
	return scan, nil
}

func (s *Store) ListSourceScans(ctx context.Context, connectionID string, limit, offset int) (governance.SourceScanPage, error) {
	rows, err := s.db.Query(ctx, `
SELECT id, state, observed_items, unchanged_items, added_items,
       updated_items, over_limit_items, missing_items, planned_uploads,
       error_code, created_at, completed_at, COUNT(*) OVER()
FROM source_scans
WHERE source_connection_id = $1
ORDER BY created_at DESC, id DESC
LIMIT $2 OFFSET $3`, connectionID, limit, offset)
	if err != nil {
		return governance.SourceScanPage{}, fmt.Errorf("query source scans: %w", err)
	}
	defer rows.Close()
	page := governance.SourceScanPage{Items: []governance.SourceScanSummaryView{}, Limit: limit, Offset: offset}
	for rows.Next() {
		var item governance.SourceScanSummaryView
		var state string
		if err := rows.Scan(&item.ID, &state, &item.ObservedItems, &item.UnchangedItems, &item.AddedItems,
			&item.UpdatedItems, &item.OverLimitItems, &item.MissingItems, &item.PlannedUploads,
			&item.ErrorCode, &item.CreatedAt, &item.CompletedAt, &page.Total); err != nil {
			return governance.SourceScanPage{}, fmt.Errorf("scan source scan summary: %w", err)
		}
		item.State = governance.SourceScanState(state)
		if !item.State.Valid() {
			return governance.SourceScanPage{}, errors.New("invalid source scan summary state")
		}
		page.Items = append(page.Items, item)
	}
	if err := rows.Err(); err != nil {
		return governance.SourceScanPage{}, fmt.Errorf("iterate source scans: %w", err)
	}
	return page, nil
}

func (s *Store) GetSourceScanDetail(ctx context.Context, connectionID, scanID string) (governance.SourceScanDetailView, error) {
	var detail governance.SourceScanDetailView
	var state string
	err := s.db.QueryRow(ctx, `
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 = $1 AND id = $2`, connectionID, scanID).Scan(
		&detail.ID, &state, &detail.ObservedItems, &detail.UnchangedItems, &detail.AddedItems,
		&detail.UpdatedItems, &detail.OverLimitItems, &detail.MissingItems, &detail.PlannedUploads,
		&detail.ErrorCode, &detail.CreatedAt, &detail.CompletedAt,
	)
	if errors.Is(err, pgx.ErrNoRows) {
		return governance.SourceScanDetailView{}, ErrSourceScanNotFound
	}
	if err != nil {
		return governance.SourceScanDetailView{}, fmt.Errorf("query source scan detail: %w", err)
	}
	detail.State = governance.SourceScanState(state)
	if !detail.State.Valid() {
		return governance.SourceScanDetailView{}, errors.New("invalid source scan detail state")
	}
	rows, err := s.db.Query(ctx, `
SELECT id, source_item_id, change_type, observed_size_bytes, created_at
FROM source_scan_changes
WHERE source_connection_id = $1 AND source_scan_id = $2
ORDER BY created_at, id`, connectionID, scanID)
	if err != nil {
		return governance.SourceScanDetailView{}, fmt.Errorf("query source scan changes: %w", err)
	}
	defer rows.Close()
	detail.Changes = []governance.SourceScanChangeView{}
	for rows.Next() {
		var change governance.SourceScanChangeView
		var changeType string
		if err := rows.Scan(&change.ID, &change.SourceItemID, &changeType, &change.ObservedSizeBytes, &change.CreatedAt); err != nil {
			return governance.SourceScanDetailView{}, fmt.Errorf("scan source scan change: %w", err)
		}
		change.ChangeType = governance.SourceScanChangeType(changeType)
		switch change.ChangeType {
		case governance.SourceScanAdded, governance.SourceScanUpdated, governance.SourceScanUnavailable:
		default:
			return governance.SourceScanDetailView{}, errors.New("invalid source scan change type")
		}
		detail.Changes = append(detail.Changes, change)
	}
	if err := rows.Err(); err != nil {
		return governance.SourceScanDetailView{}, fmt.Errorf("iterate source scan changes: %w", err)
	}
	return detail, nil
}

func (s *Store) StartSourceScan(ctx context.Context, scanID string, now time.Time) (governance.SourceScan, error) {
	scan, err := scanSourceScan(s.db.QueryRow(ctx, startSourceScanSQL, scanID, now))
	if errors.Is(err, pgx.ErrNoRows) {
		return governance.SourceScan{}, ErrSourceScanStateConflict
	}
	if err != nil {
		return governance.SourceScan{}, fmt.Errorf("start source scan: %w", err)
	}
	return scan, nil
}

func (s *Store) LoadPPCurrentItems(ctx context.Context, connectionID string) (map[string]ppconnector.CurrentItem, error) {
	rows, err := s.db.Query(ctx, `
SELECT stable_external_id, revision
FROM source_items
WHERE source_connection_id = $1
ORDER BY stable_external_id`, connectionID)
	if err != nil {
		return nil, fmt.Errorf("query current PP items: %w", err)
	}
	defer rows.Close()
	items := make(map[string]ppconnector.CurrentItem)
	for rows.Next() {
		var sourceID string
		var item ppconnector.CurrentItem
		if err := rows.Scan(&sourceID, &item.Revision); err != nil {
			return nil, fmt.Errorf("scan current PP item: %w", err)
		}
		items[sourceID] = item
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate current PP items: %w", err)
	}
	return items, nil
}

func (s *Store) CompleteSourceScan(ctx context.Context, scanID string, result governance.SourceScanResult, now time.Time) error {
	if !validSourceScanResult(result) || strings.TrimSpace(scanID) == "" || now.IsZero() {
		return ErrSourceScanStateConflict
	}
	type encodedChange struct {
		ID                string                          `json:"id"`
		StableExternalID  string                          `json:"stable_external_id"`
		ChangeType        governance.SourceScanChangeType `json:"change_type"`
		PreviousRevision  string                          `json:"previous_revision"`
		ObservedRevision  string                          `json:"observed_revision"`
		ObservedSizeBytes *int64                          `json:"observed_size_bytes"`
	}
	changes := make([]encodedChange, 0, len(result.Changes))
	for _, change := range result.Changes {
		changes = append(changes, encodedChange{
			ID:               deterministicGovernanceID("chg-", scanID, change.StableExternalID),
			StableExternalID: change.StableExternalID, ChangeType: change.ChangeType,
			PreviousRevision: change.PreviousRevision, ObservedRevision: change.ObservedRevision,
			ObservedSizeBytes: change.ObservedSizeBytes,
		})
	}
	payload, err := json.Marshal(changes)
	if err != nil {
		return errors.New("encode source scan changes")
	}
	var completed, inserted, health int
	err = s.db.QueryRow(ctx, completeSourceScanSQL,
		scanID, result.SnapshotSHA256, result.ObservedItems, result.UnchangedItems,
		result.AddedItems, result.UpdatedItems, result.OverLimitItems,
		result.MissingItems, result.PlannedUploads, now, payload,
	).Scan(&completed, &inserted, &health)
	if err != nil {
		return fmt.Errorf("complete source scan: %w", err)
	}
	if completed != 1 || inserted != len(changes) || health != 1 {
		return ErrSourceScanStateConflict
	}
	return nil
}

func (s *Store) FailSourceScan(ctx context.Context, scanID, code string, now time.Time) error {
	if strings.TrimSpace(scanID) == "" || now.IsZero() || code != "pp_snapshot_invalid" && code != "source_type_unsupported" {
		return ErrSourceScanStateConflict
	}
	var failed, health int
	if err := s.db.QueryRow(ctx, failSourceScanSQL, scanID, code, now).Scan(&failed, &health); err != nil {
		return fmt.Errorf("fail source scan: %w", err)
	}
	if failed != 1 || health != 1 {
		return ErrSourceScanStateConflict
	}
	return nil
}

func validSourceScanResult(result governance.SourceScanResult) bool {
	if !governance.ValidSHA256(result.SnapshotSHA256) || result.ObservedItems < 0 || result.UnchangedItems < 0 ||
		result.AddedItems < 0 || result.UpdatedItems < 0 || result.OverLimitItems < 0 || result.MissingItems < 0 ||
		result.PlannedUploads < 0 || result.ObservedItems != result.UnchangedItems+result.AddedItems+result.UpdatedItems+result.OverLimitItems ||
		result.PlannedUploads != result.UpdatedItems || len(result.Changes) != result.AddedItems+result.UpdatedItems+result.MissingItems {
		return false
	}
	seen := make(map[string]struct{}, len(result.Changes))
	for _, change := range result.Changes {
		if change.StableExternalID == "" || len(change.StableExternalID) > 512 || len(change.PreviousRevision) > 512 || len(change.ObservedRevision) > 512 {
			return false
		}
		if _, exists := seen[change.StableExternalID]; exists {
			return false
		}
		seen[change.StableExternalID] = struct{}{}
		switch change.ChangeType {
		case governance.SourceScanAdded:
			if change.PreviousRevision != "" || change.ObservedRevision == "" || change.ObservedSizeBytes == nil {
				return false
			}
		case governance.SourceScanUpdated:
			if change.PreviousRevision == "" || change.ObservedRevision == "" || change.ObservedSizeBytes == nil {
				return false
			}
		case governance.SourceScanUnavailable:
			if change.PreviousRevision == "" || change.ObservedRevision != "" || change.ObservedSizeBytes != nil {
				return false
			}
		default:
			return false
		}
		if change.ObservedSizeBytes != nil && *change.ObservedSizeBytes < 0 {
			return false
		}
	}
	return true
}

func scanSourceScan(row rowScanner) (governance.SourceScan, error) {
	var scan governance.SourceScan
	var sourceType, state string
	if err := row.Scan(&scan.ID, &scan.SourceConnectionID, &sourceType, &state, &scan.CreatedAt); err != nil {
		return governance.SourceScan{}, err
	}
	scan.SourceType = governance.SourceType(sourceType)
	scan.State = governance.SourceScanState(state)
	if !scan.SourceType.Valid() || !scan.State.Valid() {
		return governance.SourceScan{}, errors.New("invalid source scan row")
	}
	return scan, nil
}

func newOpaqueID(prefix string) (string, error) {
	var value [16]byte
	if _, err := rand.Read(value[:]); err != nil {
		return "", fmt.Errorf("generate opaque ID: %w", err)
	}
	return prefix + hex.EncodeToString(value[:]), nil
}