oikos/internal/adapters/postgres/integration_test.go

package db

// Integration tests against a real TimescaleDB. Guarded by
// OIKOS_TEST_DATABASE_URL — skipped when unset. Run with:
//
//   docker compose up -d postgres
//   OIKOS_TEST_DATABASE_URL="postgres://oikos:oikos_dev@localhost:5432/oikos?sslmode=disable" go test ./internal/db/
//
// or `make test-db`. Each run creates a throwaway database and drops it.

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"math/rand"
	"os"
	"strings"
	"testing"

	"github.com/dtoro/oikos/internal/core/domain"
	"github.com/jackc/pgx/v5"
	"gopkg.in/yaml.v3"
)

func seedsDir() string { return "../../seeds" }

// newTestPool creates a throwaway database (dropped on cleanup), runs all
// migrations, and returns a pool connected to it.
func newTestPool(t *testing.T) *Pool {
	t.Helper()
	baseURL := os.Getenv("OIKOS_TEST_DATABASE_URL")
	if baseURL == "" {
		t.Skip("OIKOS_TEST_DATABASE_URL not set — skipping integration test")
	}
	ctx := context.Background()

	admin, err := pgx.Connect(ctx, baseURL)
	if err != nil {
		t.Fatalf("connect admin: %v", err)
	}
	dbName := fmt.Sprintf("oikos_test_%08x", rand.Int63())
	if _, err := admin.Exec(ctx, "CREATE DATABASE "+dbName); err != nil {
		admin.Close(ctx)
		t.Fatalf("create test db: %v", err)
	}
	admin.Close(ctx)

	testURL := swapDatabase(baseURL, dbName)
	pool, err := New(ctx, testURL)
	if err != nil {
		t.Fatalf("connect test db: %v", err)
	}
	t.Cleanup(func() {
		pool.Close()
		admin, err := pgx.Connect(ctx, baseURL)
		if err == nil {
			admin.Exec(ctx, "DROP DATABASE IF EXISTS "+dbName+" WITH (FORCE)")
			admin.Close(ctx)
		}
	})

	if err := pool.Migrate(ctx); err != nil {
		t.Fatalf("migrate: %v", err)
	}
	return pool
}

// swapDatabase replaces the database name in a postgres URL.
func swapDatabase(url, db string) string {
	// postgres://user:pass@host:port/dbname?params
	qi := strings.Index(url, "?")
	params := ""
	base := url
	if qi >= 0 {
		base, params = url[:qi], url[qi:]
	}
	si := strings.LastIndex(base, "/")
	return base[:si+1] + db + params
}

func seedAll(t *testing.T, pool *Pool, dir string) {
	t.Helper()
	for _, f := range []string{"ontology.yaml", "inventory.yaml", "policy.yaml"} {
		content, err := os.ReadFile(dir + "/" + f)
		if err != nil {
			t.Fatalf("read %s: %v", f, err)
		}
		ingestSeedContent(t, pool, f, content)
	}
}

func ingestSeedContent(t *testing.T, pool *Pool, name string, content []byte) {
	t.Helper()
	ctx := context.Background()
	err := pool.SeedIngest(ctx, name, content,
		func(ctx context.Context, tx pgx.Tx, data map[string]any) error {
			var err error
			switch name {
			case "ontology.yaml":
				_, err = IngestOntologySeed(ctx, tx, data)
			case "inventory.yaml":
				_, err = IngestInventorySeed(ctx, tx, data)
			case "policy.yaml":
				_, err = IngestPolicySeed(ctx, tx, data)
			}
			return err
		})
	if err != nil {
		t.Fatalf("ingest %s: %v", name, err)
	}
}

func count(t *testing.T, pool *Pool, query string) int {
	t.Helper()
	var n int
	if err := pool.QueryRow(context.Background(), query).Scan(&n); err != nil {
		t.Fatalf("count %q: %v", query, err)
	}
	return n
}

func TestMigrateIdempotent(t *testing.T) {
	pool := newTestPool(t)
	// second run must be a clean no-op
	if err := pool.Migrate(context.Background()); err != nil {
		t.Fatalf("second migrate: %v", err)
	}
}

func TestSeedIngestIdempotentAndNoDuplicateEdges(t *testing.T) {
	pool := newTestPool(t)
	seedAll(t, pool, seedsDir())

	entities := count(t, pool, "SELECT count(*) FROM entities")
	edges := count(t, pool, "SELECT count(*) FROM relationships WHERE valid_to IS NULL")
	if entities == 0 || edges == 0 {
		t.Fatalf("seed produced empty graph: %d entities, %d edges", entities, edges)
	}

	// Same content → hash no-op
	seedAll(t, pool, seedsDir())
	if got := count(t, pool, "SELECT count(*) FROM relationships WHERE valid_to IS NULL"); got != edges {
		t.Errorf("unchanged re-seed altered edges: %d → %d", edges, got)
	}

	// Changed content (hash differs) → full re-ingest must NOT duplicate edges
	// (regression: the old upsert conflicted on valid_from and duplicated all
	// 144 edges on every re-ingest)
	content, err := os.ReadFile(seedsDir() + "/inventory.yaml")
	if err != nil {
		t.Fatal(err)
	}
	touched := append(content, []byte("\n# touched for hash change\n")...)
	ingestSeedContent(t, pool, "inventory.yaml", touched)

	if got := count(t, pool, "SELECT count(*) FROM relationships WHERE valid_to IS NULL"); got != edges {
		t.Errorf("touched re-seed duplicated edges: %d → %d", edges, got)
	}
	if dup := count(t, pool, `SELECT count(*) FROM (
			SELECT source_id, target_id, type FROM relationships
			WHERE valid_to IS NULL GROUP BY 1,2,3 HAVING count(*) > 1) d`); dup != 0 {
		t.Errorf("%d duplicated current edges", dup)
	}
	if got := count(t, pool, "SELECT count(*) FROM entities"); got != entities {
		t.Errorf("touched re-seed altered entity count: %d → %d", entities, got)
	}
}

// Regression: insertOneEntityType read tMap["attribute_schema"], but
// seeds/ontology.yaml spells the key `attributes:`. The mismatch marshalled a
// nil into the JSON literal `null` for every one of the 60 types, so no
// attribute schema was ever ingested — the API and `oikos export` returned
// null across the board, silently, for the life of the project.
func TestSeedIngestsAttributeSchemas(t *testing.T) {
	pool := newTestPool(t)
	seedAll(t, pool, seedsDir())

	if n := count(t, pool,
		`SELECT count(*) FROM entity_types WHERE attribute_schema = 'null'::jsonb`); n != 0 {
		t.Errorf("%d entity types stored the JSON literal null instead of a schema or SQL NULL", n)
	}

	if n := count(t, pool,
		`SELECT count(*) FROM entity_types WHERE jsonb_typeof(attribute_schema) = 'object'`); n == 0 {
		t.Fatal("no entity type ingested an attribute schema")
	}

	// A type declaring `attributes:` must round-trip its properties.
	if n := count(t, pool, `SELECT count(*) FROM entity_types
		WHERE name = 'lxc' AND attribute_schema #>> '{properties,pve_id,type}' = 'integer'`); n != 1 {
		t.Error("lxc.attribute_schema lost its declared pve_id property")
	}

	// A type declaring none stores SQL NULL, not a JSON null.
	if n := count(t, pool, `SELECT count(*) FROM entity_types
		WHERE name = 'sensor' AND attribute_schema IS NULL`); n != 1 {
		t.Error("a type declaring no attributes should store SQL NULL")
	}
}

// monitoring_spec drives which entities coverageSweep may flag as unmonitored,
// so the three states have to survive ingest distinctly: SQL NULL (undeclared,
// resolved from an ancestor or the layer default), '[]' (explicitly
// unmonitorable), and a non-empty array (the kinds the type warrants).
func TestSeedIngestsMonitoringSpec(t *testing.T) {
	pool := newTestPool(t)
	seedAll(t, pool, seedsDir())

	cases := []struct {
		typ, where, desc string
	}{
		{"service", `monitoring_spec = '["http","process"]'::jsonb`, "declared kinds"},
		{"machine", `monitoring_spec = '["ping","resource","updates"]'::jsonb`, "declared on an abstract type"},
		{"site", `monitoring_spec = '[]'::jsonb`, "explicitly unmonitorable"},
		{"lxc", `monitoring_spec IS NULL`, "inherits from container, so its own column is NULL"},
	}
	for _, c := range cases {
		if n := count(t, pool, fmt.Sprintf(
			`SELECT count(*) FROM entity_types WHERE name = '%s' AND %s`, c.typ, c.where)); n != 1 {
			t.Errorf("%s (%s): monitoring_spec did not match %s", c.typ, c.desc, c.where)
		}
	}
}

func TestAbstractTypeRejected(t *testing.T) {
	pool := newTestPool(t)
	seedAll(t, pool, seedsDir())

	bad := []byte(`
version: 1
entities:
  - {slug: "machine:ghost", type: machine, name: ghost}
`)
	ctx := context.Background()
	err := pool.SeedIngest(ctx, "inventory.yaml", bad,
		func(ctx context.Context, tx pgx.Tx, data map[string]any) error {
			_, err := IngestInventorySeed(ctx, tx, data)
			return err
		})
	if !errors.Is(err, domain.ErrAbstractType) {
		t.Errorf("abstract instantiation = %v, want ErrAbstractType", err)
	}
}

func TestEdgeEndpointValidation(t *testing.T) {
	pool := newTestPool(t)
	seedAll(t, pool, seedsDir())

	// routes-to requires source ingress-route; a service source must fail
	bad := []byte(`
version: 1
relationships:
  - {source: "service:gitea", target: "service:caddy", type: routes-to}
`)
	ctx := context.Background()
	err := pool.SeedIngest(ctx, "inventory.yaml", bad,
		func(ctx context.Context, tx pgx.Tx, data map[string]any) error {
			_, err := IngestInventorySeed(ctx, tx, data)
			return err
		})
	if !errors.Is(err, domain.ErrInvalidEdge) {
		t.Errorf("bad edge = %v, want ErrInvalidEdge", err)
	}

	// hosts from a proxmox-host (is-a machine) to an lxc (is-a compute-entity)
	// must PASS via hierarchy walk — already covered by the seed itself, but
	// assert an explicit one for clarity
	good := []byte(`
version: 1
relationships:
  - {source: "host:strong", target: "lxc:jellyfin", type: hosts}
`)
	err = pool.SeedIngest(ctx, "inventory.yaml", good,
		func(ctx context.Context, tx pgx.Tx, data map[string]any) error {
			_, err := IngestInventorySeed(ctx, tx, data)
			return err
		})
	if err != nil {
		t.Errorf("valid inherited edge rejected: %v", err)
	}
}

func TestCardinalityEnforced(t *testing.T) {
	pool := newTestPool(t)
	seedAll(t, pool, seedsDir())

	// routes-to is many-to-one: one ingress route cannot point at two services
	bad := []byte(`
version: 1
relationships:
  - {source: "ingress:git.hubris.network", target: "service:jellyfin", type: routes-to}
`)
	ctx := context.Background()
	err := pool.SeedIngest(ctx, "inventory.yaml", bad,
		func(ctx context.Context, tx pgx.Tx, data map[string]any) error {
			_, err := IngestInventorySeed(ctx, tx, data)
			return err
		})
	if err == nil || !strings.Contains(err.Error(), "cardinality") {
		t.Errorf("cardinality violation = %v, want cardinality error", err)
	}
}

func TestBlastRadiusTerminatesOnCycles(t *testing.T) {
	pool := newTestPool(t)
	seedAll(t, pool, seedsDir())

	// Build a dependency cycle: gitea → caddy → authentik → gitea.
	//
	// `depends-on` is declared blast_direction: backward — "A depends-on B"
	// means B failing breaks A — so the blast radius of gitea walks the edges
	// BACKWARDS: whoever depends on gitea is affected first. That is authentik
	// (1 hop), then caddy which depends on authentik (2 hops).
	//
	// This test previously asserted caddy=1, authentik=2, which is the same
	// cycle walked the wrong way round: blast_radius used to follow every edge
	// source→target regardless of what the edge means, so it answered "what
	// does gitea depend on" while being named for the opposite question.
	cycle := []byte(`
version: 1
relationships:
  - {source: "service:gitea", target: "service:caddy", type: depends-on}
  - {source: "service:caddy", target: "service:authentik", type: depends-on}
  - {source: "service:authentik", target: "service:gitea", type: depends-on}
`)
	ctx := context.Background()
	err := pool.SeedIngest(ctx, "inventory.yaml", cycle,
		func(ctx context.Context, tx pgx.Tx, data map[string]any) error {
			_, err := IngestInventorySeed(ctx, tx, data)
			return err
		})
	if err != nil {
		t.Fatalf("cycle ingest: %v", err)
	}

	rows, err := pool.Query(ctx, `
		SELECT e.slug, b.depth
		FROM blast_radius((SELECT id FROM entities WHERE slug='service:gitea'), 5,
		                  ARRAY['depends-on']) b
		JOIN entities e ON e.id = b.entity_id ORDER BY b.depth`)
	if err != nil {
		t.Fatalf("blast_radius: %v", err)
	}
	defer rows.Close()
	got := map[string]int{}
	for rows.Next() {
		var slug string
		var depth int
		if err := rows.Scan(&slug, &depth); err != nil {
			t.Fatal(err)
		}
		got[slug] = depth
	}
	want := map[string]int{"service:gitea": 0, "service:authentik": 1, "service:caddy": 2}
	for slug, depth := range want {
		if got[slug] != depth {
			t.Errorf("blast_radius[%s] = %d, want %d (full: %v)", slug, got[slug], depth, got)
		}
	}
	// Deliberately not an exact node count. Walking the right way round also
	// surfaces the real seed's own dependents of gitea (homelab-mcp and what
	// depends on it), which are correct answers — the old exact-count
	// assertion only held because the forward walk found nothing real.
	// What matters here is that the cycle terminates rather than recursing.
	if len(got) > 20 {
		t.Errorf("blast_radius did not terminate sensibly: %d nodes: %v", len(got), got)
	}
	for slug, depth := range got {
		if depth > 5 {
			t.Errorf("blast_radius[%s] = %d, beyond the max_depth bound", slug, depth)
		}
	}
}

// TestExportRoundTripStable: export → ingest into a fresh DB → export again
// must yield byte-identical YAML (the canonical-form fixpoint, plan D6).
func TestExportRoundTripStable(t *testing.T) {
	pool := newTestPool(t)
	seedAll(t, pool, seedsDir())
	ctx := context.Background()

	export1, err := ExportToYAML(ctx, pool)
	if err != nil {
		t.Fatalf("export 1: %v", err)
	}
	for name, content := range export1 {
		var doc map[string]any
		if err := yaml.Unmarshal(content, &doc); err != nil {
			t.Fatalf("export %s is not valid YAML: %v", name, err)
		}
	}

	pool2 := newTestPool(t)
	for _, name := range []string{"ontology.yaml", "inventory.yaml", "policy.yaml"} {
		ingestSeedContent(t, pool2, name, export1[name])
	}
	export2, err := ExportToYAML(ctx, pool2)
	if err != nil {
		t.Fatalf("export 2: %v", err)
	}
	for _, name := range []string{"ontology.yaml", "inventory.yaml", "policy.yaml"} {
		if !bytes.Equal(export1[name], export2[name]) {
			t.Errorf("%s round-trip not byte-stable (len %d vs %d)",
				name, len(export1[name]), len(export2[name]))
		}
	}

	// sanity: exported inventory carries the real graph, not a stub
	// (regression: export used to write 11-byte "version: 1" stubs)
	if len(export1["inventory.yaml"]) < 1000 {
		t.Errorf("inventory export suspiciously small: %d bytes", len(export1["inventory.yaml"]))
	}
}