knowledge-governance/internal/workflows/sourcescan/handler_test.go

package sourcescan

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

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

func TestFilePPScannerReadsValidatedSnapshotUnderAllowedRoot(t *testing.T) {
	root := t.TempDir()
	source := strings.Repeat("a", 64)
	sha := strings.Repeat("b", 64)
	ledgerPath := filepath.Join(root, "state.jsonl")
	manifestPath := filepath.Join(root, "manifest.json")
	ledger := `{"kb_id":"kb-1","source_url_hash":"` + source + `","source_ids":["pp-1"],"bytes":10,"sha256":"` + sha + `","status":"completed","knowledge_id":"knowledge-1"}`
	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 := NewFilePPScanner(FilePPScannerConfig{AllowedRoot: root, LedgerPath: ledgerPath, ManifestPath: manifestPath})
	if err != nil {
		t.Fatalf("NewFilePPScanner() error = %v", err)
	}
	snapshot, fingerprint, err := scanner.Scan(context.Background())
	if err != nil {
		t.Fatalf("Scan() error = %v", err)
	}
	if len(snapshot.Items) != 1 || snapshot.Items[0].SourceURLHash != source {
		t.Fatalf("snapshot = %#v", snapshot)
	}
	if !governance.ValidSHA256(fingerprint) {
		t.Fatalf("fingerprint = %q", fingerprint)
	}
}

func TestFilePPScannerRejectsManifestItemWithoutLocalPath(t *testing.T) {
	root := t.TempDir()
	source := strings.Repeat("a", 64)
	sha := strings.Repeat("b", 64)
	ledgerPath := filepath.Join(root, "state.jsonl")
	manifestPath := filepath.Join(root, "manifest.json")
	ledger := `{"kb_id":"kb-1","source_url_hash":"` + source + `","source_ids":["pp-1"],"bytes":10,"sha256":"` + sha + `","status":"completed","knowledge_id":"knowledge-1"}`
	if err := os.WriteFile(ledgerPath, []byte(ledger), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(manifestPath, []byte(`{"results":[{"local_path":""}]}`), 0o600); err != nil {
		t.Fatal(err)
	}
	scanner, err := NewFilePPScanner(FilePPScannerConfig{AllowedRoot: root, LedgerPath: ledgerPath, ManifestPath: manifestPath})
	if err != nil {
		t.Fatal(err)
	}
	if _, _, err := scanner.Scan(context.Background()); err == nil || !strings.Contains(err.Error(), "local_path") {
		t.Fatalf("Scan() error = %v, want missing local_path error", err)
	}
}

func TestFilePPScannerAllowsMissingArtifactForOverLimitRecord(t *testing.T) {
	root := t.TempDir()
	source := strings.Repeat("a", 64)
	sha := strings.Repeat("b", 64)
	ledgerPath := filepath.Join(root, "state.jsonl")
	manifestPath := filepath.Join(root, "manifest.json")
	ledger := `{"kb_id":"kb-1","source_url_hash":"` + source + `","source_ids":["pp-1"],"bytes":10,"sha256":"` + sha + `","status":"completed","knowledge_id":"knowledge-1"}`
	manifest := `{"results":[{"url":"https://source.invalid/large","local_path":"` + filepath.Join(root, "not-downloaded.bin") + `","bytes":52428801,"status":"over_limit","source_ids":["pp-large"]}]}`
	if err := os.WriteFile(ledgerPath, []byte(ledger), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(manifestPath, []byte(manifest), 0o600); err != nil {
		t.Fatal(err)
	}
	scanner, err := NewFilePPScanner(FilePPScannerConfig{AllowedRoot: root, LedgerPath: ledgerPath, ManifestPath: manifestPath})
	if err != nil {
		t.Fatal(err)
	}
	snapshot, _, err := scanner.Scan(context.Background())
	if err != nil {
		t.Fatalf("Scan() error = %v", err)
	}
	if len(snapshot.Items) != 2 || snapshot.Items[1].Status != ppconnector.StatusOverLimit {
		t.Fatalf("snapshot = %#v", snapshot)
	}
}

func TestPPHandlerCompletesAggregateReadOnlyReconciliation(t *testing.T) {
	now := time.Date(2026, 9, 6, 3, 0, 0, 0, time.UTC)
	sameSource := strings.Repeat("a", 64)
	updatedSource := strings.Repeat("b", 64)
	overLimitSource := strings.Repeat("c", 64)
	missingSource := strings.Repeat("d", 64)
	addedSource := strings.Repeat("e", 64)
	sameSHA := strings.Repeat("1", 64)
	newSHA := strings.Repeat("2", 64)
	oldSHA := strings.Repeat("3", 64)
	addedSHA := strings.Repeat("5", 64)

	store := &fakePPScanStore{
		scan: governance.SourceScan{ID: "scan-1", SourceConnectionID: "conn-pp", SourceType: governance.SourceTypePP, State: governance.SourceScanPending},
		current: map[string]ppconnector.CurrentItem{
			sameSource:    {Revision: sameSHA},
			updatedSource: {Revision: oldSHA},
			missingSource: {Revision: strings.Repeat("4", 64)},
		},
	}
	scanner := &fakePPScanner{snapshot: ppconnector.Snapshot{KBID: "kb-1", Items: []ppconnector.Item{
		{KBID: "kb-1", SourceURLHash: sameSource, SourceIDs: []string{"pp-1"}, Bytes: 10, SHA256: sameSHA, Status: ppconnector.StatusCompleted, KnowledgeID: "knowledge-1"},
		{KBID: "kb-1", SourceURLHash: updatedSource, SourceIDs: []string{"pp-2"}, Bytes: 20, SHA256: newSHA, Status: ppconnector.StatusCompleted, KnowledgeID: "knowledge-2"},
		{KBID: "kb-1", SourceURLHash: overLimitSource, SourceIDs: []string{"pp-3"}, Bytes: ppconnector.LegacyMaxBytes + 1, Status: ppconnector.StatusOverLimit},
		{KBID: "kb-1", SourceURLHash: addedSource, SourceIDs: []string{"pp-4"}, Bytes: 30, SHA256: addedSHA, Status: ppconnector.StatusCompleted, KnowledgeID: "knowledge-4"},
	}}, snapshotSHA: strings.Repeat("f", 64)}

	handler, err := NewPPHandler(PPHandlerConfig{Store: store, Scanner: scanner, Now: func() time.Time { return now }})
	if err != nil {
		t.Fatalf("NewPPHandler() error = %v", err)
	}
	job := jobs.Job{Type: "source.scan.pp", Payload: []byte(`{"scan_id":"scan-1"}`)}
	if err := handler.Handle(context.Background(), job); err != nil {
		t.Fatalf("Handle() error = %v", err)
	}

	want := governance.SourceScanResult{
		SnapshotSHA256: strings.Repeat("f", 64), ObservedItems: 4, UnchangedItems: 1, AddedItems: 1,
		UpdatedItems: 1, OverLimitItems: 1, MissingItems: 1, PlannedUploads: 1,
		Changes: []governance.SourceScanChange{
			{StableExternalID: updatedSource, ChangeType: governance.SourceScanUpdated, PreviousRevision: oldSHA, ObservedRevision: newSHA, ObservedSizeBytes: int64Pointer(20)},
			{StableExternalID: addedSource, ChangeType: governance.SourceScanAdded, ObservedRevision: addedSHA, ObservedSizeBytes: int64Pointer(30)},
			{StableExternalID: missingSource, ChangeType: governance.SourceScanUnavailable, PreviousRevision: strings.Repeat("4", 64)},
		},
	}
	if store.completedScanID != "scan-1" || !reflect.DeepEqual(store.completed, want) {
		t.Fatalf("completed scan = %q %#v, want %#v", store.completedScanID, store.completed, want)
	}
	if store.completedAt != now || store.failedCode != "" {
		t.Fatalf("completedAt=%v failedCode=%q", store.completedAt, store.failedCode)
	}
}

func TestPPHandlerNoOpsWhenReclaimedScanAlreadyCompleted(t *testing.T) {
	store := &fakePPScanStore{scan: governance.SourceScan{
		ID: "scan-completed", SourceConnectionID: "conn-pp", SourceType: governance.SourceTypePP, State: governance.SourceScanCompleted,
	}}
	scanner := &fakePPScanner{}
	handler, err := NewPPHandler(PPHandlerConfig{Store: store, Scanner: scanner})
	if err != nil {
		t.Fatal(err)
	}
	if err := handler.Handle(context.Background(), jobs.Job{Payload: []byte(`{"scan_id":"scan-completed"}`)}); err != nil {
		t.Fatalf("Handle() error = %v", err)
	}
	if scanner.calls != 0 || store.completedScanID != "" || store.failedCode != "" {
		t.Fatalf("scanner calls=%d completed=%q failed=%q", scanner.calls, store.completedScanID, store.failedCode)
	}
}

func int64Pointer(value int64) *int64 { return &value }

type fakePPScanner struct {
	snapshot    ppconnector.Snapshot
	snapshotSHA string
	err         error
	calls       int
}

func (s *fakePPScanner) Scan(context.Context) (ppconnector.Snapshot, string, error) {
	s.calls++
	return s.snapshot, s.snapshotSHA, s.err
}

type fakePPScanStore struct {
	scan            governance.SourceScan
	current         map[string]ppconnector.CurrentItem
	completedScanID string
	completed       governance.SourceScanResult
	completedAt     time.Time
	failedCode      string
}

func (s *fakePPScanStore) StartSourceScan(context.Context, string, time.Time) (governance.SourceScan, error) {
	if s.scan.State == governance.SourceScanPending {
		s.scan.State = governance.SourceScanRunning
	}
	return s.scan, nil
}

func (s *fakePPScanStore) LoadPPCurrentItems(context.Context, string) (map[string]ppconnector.CurrentItem, error) {
	return s.current, nil
}

func (s *fakePPScanStore) CompleteSourceScan(_ context.Context, scanID string, result governance.SourceScanResult, now time.Time) error {
	s.completedScanID, s.completed, s.completedAt = scanID, result, now
	return nil
}

func (s *fakePPScanStore) FailSourceScan(_ context.Context, _ string, code string, _ time.Time) error {
	s.failedCode = code
	return nil
}