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

package postgres_test

import (
	"context"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

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

	postgresstore "knowledge-governance/internal/platform/postgres"
	"knowledge-governance/internal/workflows/sourcescan"
)

func TestPPSourceScanPersistentIntegration(t *testing.T) {
	databaseURL := isolatedExternalIntegrationDatabaseURL(t, os.Getenv("TEST_DATABASE_URL"))
	if databaseURL == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	if err := postgresstore.ApplyMigrations(ctx, pool); err != nil {
		t.Fatalf("ApplyMigrations() error = %v", err)
	}

	sourceID := strings.Repeat("a", 64)
	addedSourceID := strings.Repeat("c", 64)
	revision := strings.Repeat("b", 64)
	addedRevision := strings.Repeat("d", 64)
	if _, err := pool.Exec(ctx, `
INSERT INTO source_connections (
    id, source_type, name, base_url, secret_ref, enabled, schedule, last_health
) VALUES ('conn-pp', 'pp', 'PP fixture', '', '', FALSE, '{}'::jsonb, '{}'::jsonb)`,
	); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `
INSERT INTO source_items (
    id, source_connection_id, stable_external_id, canonical_uri_hash,
    title, item_type, revision, visibility, visibility_origin, source_change
) VALUES ('item-pp', 'conn-pp', $1::varchar, $1::char(64), '', 'document', $2,
          'external', 'pp_public_tag', 'unchanged')`, sourceID, revision); err != nil {
		t.Fatal(err)
	}

	root := t.TempDir()
	ledgerPath := filepath.Join(root, "state.jsonl")
	manifestPath := filepath.Join(root, "manifest.json")
	ledger := `{"kb_id":"kb-1","source_url_hash":"` + sourceID + `","source_ids":["pp-1"],"bytes":10,"sha256":"` + revision + `","status":"completed","knowledge_id":"knowledge-1"}` + "\n" +
		`{"kb_id":"kb-1","source_url_hash":"` + addedSourceID + `","source_ids":["pp-2"],"bytes":11,"sha256":"` + addedRevision + `","status":"completed","knowledge_id":"knowledge-2"}`
	if err := os.WriteFile(ledgerPath, []byte(ledger), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(manifestPath, []byte(`{"results":[]}`), 0o600); err != nil {
		t.Fatal(err)
	}
	scanner, err := sourcescan.NewFilePPScanner(sourcescan.FilePPScannerConfig{
		AllowedRoot: root, LedgerPath: ledgerPath, ManifestPath: manifestPath,
	})
	if err != nil {
		t.Fatal(err)
	}
	store := postgresstore.NewStore(pool)
	now := time.Date(2026, 9, 6, 6, 0, 0, 0, time.UTC)
	scan, err := store.TriggerPPSourceScan(ctx, "conn-pp", now)
	if err != nil {
		t.Fatalf("TriggerPPSourceScan() error = %v", err)
	}
	job, err := store.Claim(ctx, "integration-worker", []string{"source.scan.pp"}, now, time.Minute)
	if err != nil || job == nil {
		t.Fatalf("Claim() job=%#v error=%v", job, err)
	}
	handler, err := sourcescan.NewPPHandler(sourcescan.PPHandlerConfig{Store: store, Scanner: scanner, Now: func() time.Time { return now.Add(time.Second) }})
	if err != nil {
		t.Fatal(err)
	}
	if err := handler.Handle(ctx, *job); err != nil {
		t.Fatalf("Handle() error = %v", err)
	}
	if err := store.Complete(ctx, job.ID, "integration-worker", now.Add(2*time.Second)); err != nil {
		t.Fatalf("Complete() error = %v", err)
	}

	var state, health, jobState string
	var observed, unchanged, overLimit, changes int
	if err := pool.QueryRow(ctx, `
SELECT scan.state, scan.observed_items, scan.unchanged_items,
       scan.over_limit_items,
       (SELECT COUNT(*)::INTEGER FROM source_scan_changes WHERE source_scan_id = scan.id),
       connection.last_health->>'status'
FROM source_scans AS scan
JOIN source_connections AS connection ON connection.id = scan.source_connection_id
WHERE scan.id = $1`, scan.ID).Scan(&state, &observed, &unchanged, &overLimit, &changes, &health); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT state FROM jobs WHERE id = $1`, job.ID).Scan(&jobState); err != nil {
		t.Fatal(err)
	}
	if state != "completed" || observed != 2 || unchanged != 1 || overLimit != 0 || changes != 1 || health != "healthy" || jobState != "completed" {
		t.Fatalf("state=%s observed=%d unchanged=%d overLimit=%d changes=%d health=%s job=%s", state, observed, unchanged, overLimit, changes, health, jobState)
	}
	page, err := store.ListSourceScans(ctx, "conn-pp", 20, 0)
	if err != nil || page.Total != 1 || len(page.Items) != 1 || page.Items[0].AddedItems != 1 {
		t.Fatalf("ListSourceScans() page=%#v error=%v", page, err)
	}
	detail, err := store.GetSourceScanDetail(ctx, "conn-pp", scan.ID)
	if err != nil || len(detail.Changes) != 1 || detail.Changes[0].ChangeType != "added" || detail.Changes[0].SourceItemID != nil || detail.Changes[0].ObservedSizeBytes == nil || *detail.Changes[0].ObservedSizeBytes != 11 {
		t.Fatalf("GetSourceScanDetail() detail=%#v error=%v", detail, err)
	}
}