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

package postgres

import (
	"context"
	"errors"
	"strings"
	"testing"
	"time"

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

	"knowledge-governance/internal/governance"
)

func TestListSourceConnectionsReturnsSafeOperationalProjection(t *testing.T) {
	now := time.Date(2026, 9, 6, 1, 0, 0, 0, time.UTC)
	horizon := now.Add(time.Hour)
	rows := &sourceConnectionRows{items: []sourceConnectionRow{
		{id: "conn-pp", sourceType: "pp", name: "PP历史公开知识", enabled: false, scheduleConfigured: false, health: "healthy", itemCount: 613, updatedAt: now, lastScan: true},
		{id: "conn-manual", sourceType: "manual", name: "本地手工验收", enabled: false, scheduleConfigured: false, health: "unknown", itemCount: 5, updatedAt: now},
		{id: "conn-alist", sourceType: "alist", name: "AList只读源", enabled: true, scheduleConfigured: true, health: "healthy", itemCount: 0, updatedAt: now, nextRefreshAt: &horizon},
	}}
	db := &recordingDB{rows: rows}

	page, err := NewStore(db).ListSourceConnections(context.Background())
	if err != nil {
		t.Fatalf("ListSourceConnections() error = %v", err)
	}
	if len(page.Items) != 3 || page.Items[0].ID != "conn-pp" || page.Items[0].ItemCount != 613 || page.Items[2].HealthStatus != governance.SourceHealthHealthy {
		t.Fatalf("items = %#v", page.Items)
	}
	if page.Items[0].LastScan == nil || page.Items[0].LastScan.State != governance.SourceScanCompleted || page.Items[0].LastScan.ObservedItems != 613 || page.Items[0].LastScan.OverLimitItems != 54 {
		t.Fatalf("last scan = %#v", page.Items[0].LastScan)
	}
	// V1: only the scheduled source exposes a next-refresh horizon.
	if page.Items[0].NextRefreshAt != nil || page.Items[1].NextRefreshAt != nil {
		t.Fatalf("unscheduled sources leak horizons: %v %v", page.Items[0].NextRefreshAt, page.Items[1].NextRefreshAt)
	}
	if page.Items[2].NextRefreshAt == nil || !page.Items[2].NextRefreshAt.Equal(horizon) {
		t.Fatalf("alist horizon = %v want %v", page.Items[2].NextRefreshAt, horizon)
	}
	wantSummary := governance.SourceConnectionSummary{Total: 3, Enabled: 1, Disabled: 2, Healthy: 2, Unknown: 1}
	if page.Summary != wantSummary {
		t.Fatalf("summary = %#v, want %#v", page.Summary, wantSummary)
	}
	query := strings.ToLower(db.querySQL)
	for _, required := range []string{"count(i.id)", "schedule <> '{}'::jsonb", "last_health", "group by", "left join lateral", "from source_scans", "limit 1", "c.next_refresh_at"} {
		if !strings.Contains(query, required) {
			t.Errorf("source connection query missing %q", required)
		}
	}
	for _, forbidden := range []string{"base_url", "secret_ref"} {
		if strings.Contains(query, forbidden) {
			t.Errorf("source connection query selected sensitive field %q", forbidden)
		}
	}
}

type sourceConnectionRow struct {
	id                 string
	sourceType         string
	name               string
	enabled            bool
	scheduleConfigured bool
	health             string
	itemCount          int
	updatedAt          time.Time
	nextRefreshAt      *time.Time
	lastScan           bool
}

func (row sourceConnectionRow) Scan(dest ...any) error {
	*(dest[0].(*string)) = row.id
	*(dest[1].(*string)) = row.sourceType
	*(dest[2].(*string)) = row.name
	*(dest[3].(*bool)) = row.enabled
	*(dest[4].(*bool)) = row.scheduleConfigured
	*(dest[5].(*string)) = row.health
	*(dest[6].(*int)) = row.itemCount
	*(dest[7].(*time.Time)) = row.updatedAt
	// V1: next_refresh_at pointer, present only for scheduled sources.
	*(dest[8].(**time.Time)) = row.nextRefreshAt
	if !row.lastScan {
		for index := 9; index <= 20; index++ {
			switch pointer := dest[index].(type) {
			case **string:
				*pointer = nil
			case **int:
				*pointer = nil
			case **time.Time:
				*pointer = nil
			}
		}
		return nil
	}
	id, state, errorCode := "scan-1", "completed", ""
	observed, unchanged, added, updated, overLimit, missing, planned := 613, 559, 0, 0, 54, 0, 0
	createdAt, completedAt := row.updatedAt.Add(-time.Second), row.updatedAt
	*(dest[9].(**string)) = &id
	*(dest[10].(**string)) = &state
	for index, value := range []*int{&observed, &unchanged, &added, &updated, &overLimit, &missing, &planned} {
		*(dest[11+index].(**int)) = value
	}
	*(dest[18].(**string)) = &errorCode
	*(dest[19].(**time.Time)) = &createdAt
	*(dest[20].(**time.Time)) = &completedAt
	return nil
}

type sourceConnectionRows struct {
	items []sourceConnectionRow
	index int
}

func (rows *sourceConnectionRows) Close()                                       {}
func (rows *sourceConnectionRows) Err() error                                   { return nil }
func (rows *sourceConnectionRows) CommandTag() pgconn.CommandTag                { return pgconn.CommandTag{} }
func (rows *sourceConnectionRows) FieldDescriptions() []pgconn.FieldDescription { return nil }
func (rows *sourceConnectionRows) Next() bool {
	if rows.index >= len(rows.items) {
		return false
	}
	rows.index++
	return true
}
func (rows *sourceConnectionRows) Scan(dest ...any) error {
	return rows.items[rows.index-1].Scan(dest...)
}
func (rows *sourceConnectionRows) Values() ([]any, error) { return nil, errors.New("not implemented") }
func (rows *sourceConnectionRows) RawValues() [][]byte    { return nil }
func (rows *sourceConnectionRows) Conn() *pgx.Conn        { return nil }