From 1b046836399ab5282bb1d30f7b0da3f4b418e4c7 Mon Sep 17 00:00:00 2001 From: dtoro Date: Tue, 7 Jul 2026 08:11:01 +0200 Subject: [PATCH] phase 1 review fixes: dedup edges, real export, validation, tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of aa2ca0a found and fixed: - re-ingest duplicated ALL edges (upsert conflicted on valid_from=now(), never fired) — migration 007 dedupes + partial unique index on current edges; upsert now targets it. Regression-tested. - export was a stub that overwrote seeds/*.yaml with 11-byte "version: 1" files — implemented real deterministic export (ontology/inventory/policy, cognition-layer excluded); round-trip is byte-stable (tested) - DB password leaked in startup logs (slog JSON bypasses String()) — Config now implements slog.LogValuer; regression-tested - docker-compose had literal '***' as DB password — env-interpolated - uuid.New() (v4) → uuid.NewV7() per ADR-0005 - no ontology validation on ingest — internal/ontology TypeTree: abstract instantiation rejected, relationship endpoints hierarchy-validated, cardinality enforced in-transaction, lifecycle states checked, default state applied (Phase 1 gate items, R3-1) - getOrCreateEntityID swallowed non-ErrNoRows errors - migration runner now holds a session advisory lock on one connection - Makefile: hardcoded /opt/homebrew/bin/go → go; test-db target Tests: 4 unit suites + 7 integration tests (env-guarded, throwaway DB per run): migrate idempotent, seed idempotent + no dup edges, abstract/edge/ cardinality rejection, blast_radius cycle termination, export round-trip. Co-Authored-By: Claude Fable 5 --- Makefile | 9 +- docker-compose.yml | 8 +- internal/config/config.go | 30 +- internal/config/config_test.go | 42 +++ internal/db/export.go | 300 ++++++++++++++++ internal/db/integration_test.go | 338 ++++++++++++++++++ internal/db/pool.go | 26 +- internal/db/seed.go | 121 ++++--- internal/db/splitsql_test.go | 53 +++ internal/db/typetree.go | 143 ++++++++ internal/ontology/validate.go | 112 ++++++ internal/ontology/validate_test.go | 120 +++++++ .../007_relationships_current_unique.up.sql | 18 + 13 files changed, 1250 insertions(+), 70 deletions(-) create mode 100644 internal/config/config_test.go create mode 100644 internal/db/export.go create mode 100644 internal/db/integration_test.go create mode 100644 internal/db/splitsql_test.go create mode 100644 internal/db/typetree.go create mode 100644 internal/ontology/validate.go create mode 100644 internal/ontology/validate_test.go create mode 100644 migrations/007_relationships_current_unique.up.sql diff --git a/Makefile b/Makefile index abad76d..2aa1d71 100644 --- a/Makefile +++ b/Makefile @@ -1,7 +1,7 @@ .PHONY: build test lint generate dev migrate seed export clean BINARY := oikos -GO := /opt/homebrew/bin/go +GO ?= go build: $(GO) build -o $(BINARY) -tags timetzdata ./cmd/oikos @@ -9,6 +9,13 @@ build: test: $(GO) test -race -cover ./... +# Integration tests against the compose Postgres (starts it if needed) +test-db: + docker compose up -d postgres + @sleep 3 + OIKOS_TEST_DATABASE_URL="postgres://oikos:$${OIKOS_DB_PASSWORD:-oikos_dev}@localhost:5432/oikos?sslmode=disable" \ + $(GO) test -race -count=1 ./internal/db/ + lint: $(GO) vet ./... @command -v golangci-lint >/dev/null 2>&1 && golangci-lint run || echo "golangci-lint not installed, skipping" diff --git a/docker-compose.yml b/docker-compose.yml index 05b5158..d62f9fa 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -8,7 +8,7 @@ services: environment: POSTGRES_DB: oikos POSTGRES_USER: oikos - POSTGRES_PASSWORD: oikos_dev + POSTGRES_PASSWORD: ${OIKOS_DB_PASSWORD:-oikos_dev} ports: - "5432:5432" volumes: @@ -28,7 +28,7 @@ services: postgres: condition: service_healthy environment: - OIKOS_DATABASE_URL: postgres://oikos:***@postgres:5432/oikos?sslmode=disable + OIKOS_DATABASE_URL: postgres://oikos:${OIKOS_DB_PASSWORD:-oikos_dev}@postgres:5432/oikos?sslmode=disable command: ["migrate"] restart: "no" @@ -41,7 +41,7 @@ services: migrate: condition: service_completed_successfully environment: - OIKOS_DATABASE_URL: postgres://oikos:***@postgres:5432/oikos?sslmode=disable + OIKOS_DATABASE_URL: postgres://oikos:${OIKOS_DB_PASSWORD:-oikos_dev}@postgres:5432/oikos?sslmode=disable OIKOS_SEEDS_DIR: /app/seeds command: ["seed"] restart: "no" @@ -56,7 +56,7 @@ services: seed: condition: service_completed_successfully environment: - OIKOS_DATABASE_URL: postgres://oikos:***@postgres:5432/oikos?sslmode=disable + OIKOS_DATABASE_URL: postgres://oikos:${OIKOS_DB_PASSWORD:-oikos_dev}@postgres:5432/oikos?sslmode=disable OIKOS_API_LISTEN: ":8090" OIKOS_ENV: dev OIKOS_DEBUG: "true" diff --git a/internal/config/config.go b/internal/config/config.go index f6819da..1d132af 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -2,6 +2,7 @@ package config import ( "fmt" + "log/slog" "os" "strings" ) @@ -64,18 +65,41 @@ func FromEnv() Config { return c } -// String returns a human-safe representation (secrets redacted). -func (c Config) String() string { +// redactedDBURL masks credentials in a postgres:// URL. +func (c Config) redactedDBURL() string { dbURL := c.DatabaseURL if i := strings.Index(dbURL, "@"); i >= 0 { if j := strings.Index(dbURL, "://"); j >= 0 && j < i { dbURL = dbURL[:j+3] + "***" + dbURL[i:] } } + return dbURL +} + +// String returns a human-safe representation (secrets redacted). +func (c Config) String() string { token := "" if c.MCPBearerToken != "" { token = "***" } return fmt.Sprintf("Config{DB=%s Listen=%s Env=%s Debug=%v MCPToken=%s SeedsDir=%s}", - dbURL, c.APIListen, c.APIEnv, c.Debug, token, c.SeedsDir) + c.redactedDBURL(), c.APIListen, c.APIEnv, c.Debug, token, c.SeedsDir) +} + +// LogValue implements slog.LogValuer so structured handlers (JSON) never +// serialize raw secrets — without this, slog marshals struct fields +// directly and String() is bypassed. +func (c Config) LogValue() slog.Value { + token := "" + if c.MCPBearerToken != "" { + token = "***" + } + return slog.GroupValue( + slog.String("db", c.redactedDBURL()), + slog.String("listen", c.APIListen), + slog.String("env", c.APIEnv), + slog.Bool("debug", c.Debug), + slog.String("mcp_token", token), + slog.String("seeds_dir", c.SeedsDir), + ) } diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..4bf8335 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,42 @@ +package config + +import ( + "bytes" + "log/slog" + "strings" + "testing" +) + +func secretConfig() Config { + c := Default() + c.DatabaseURL = "postgres://oikos:supersecretpw@localhost:5432/oikos" + c.MCPBearerToken = "supersecrettoken" + return c +} + +func TestStringRedactsSecrets(t *testing.T) { + s := secretConfig().String() + for _, leak := range []string{"supersecretpw", "supersecrettoken"} { + if strings.Contains(s, leak) { + t.Errorf("String() leaks %q: %s", leak, s) + } + } +} + +// TestSlogJSONRedactsSecrets guards the bug where slog's JSON handler +// serialized Config struct fields directly, bypassing String() and leaking +// the DB password into logs. +func TestSlogJSONRedactsSecrets(t *testing.T) { + var buf bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&buf, nil)) + logger.Info("starting", "config", secretConfig()) + out := buf.String() + for _, leak := range []string{"supersecretpw", "supersecrettoken"} { + if strings.Contains(out, leak) { + t.Errorf("slog JSON output leaks %q: %s", leak, out) + } + } + if !strings.Contains(out, "***") { + t.Errorf("expected redaction marker in log output: %s", out) + } +} diff --git a/internal/db/export.go b/internal/db/export.go new file mode 100644 index 0000000..594c221 --- /dev/null +++ b/internal/db/export.go @@ -0,0 +1,300 @@ +package db + +import ( + "context" + "encoding/json" + "fmt" + + "gopkg.in/yaml.v3" +) + +// ExportToYAML regenerates the three seed YAMLs from the DB (DR / version +// control, plan D6). Output is deterministic: maps marshal with sorted keys +// (yaml.v3 default for map[string]any) and lists are ordered by slug/name, +// so export → ingest → export is byte-stable. +// +// Cognition-layer entities (signals, executions, patterns, …) are runtime +// state, not inventory — they are excluded from the export. +func ExportToYAML(ctx context.Context, pool *Pool) (map[string][]byte, error) { + result := make(map[string][]byte) + + for name, fn := range map[string]func(context.Context, *Pool) (map[string]any, error){ + "ontology.yaml": exportOntology, + "inventory.yaml": exportInventory, + "policy.yaml": exportPolicy, + } { + doc, err := fn(ctx, pool) + if err != nil { + return nil, fmt.Errorf("export %s: %w", name, err) + } + out, err := yaml.Marshal(doc) + if err != nil { + return nil, fmt.Errorf("marshal %s: %w", name, err) + } + result[name] = out + } + return result, nil +} + +func exportOntology(ctx context.Context, pool *Pool) (map[string]any, error) { + lifecycles := map[string]any{} + rows, err := pool.Query(ctx, + `SELECT id, states, default_state, terminal_states, transitions + FROM lifecycle_defs ORDER BY id`) + if err != nil { + return nil, err + } + for rows.Next() { + var id, def string + var states, terminal []string + var transitionsJSON []byte + if err := rows.Scan(&id, &states, &def, &terminal, &transitionsJSON); err != nil { + rows.Close() + return nil, err + } + var transitions map[string]any + if err := json.Unmarshal(transitionsJSON, &transitions); err != nil { + rows.Close() + return nil, fmt.Errorf("lifecycle %s transitions: %w", id, err) + } + lifecycles[id] = map[string]any{ + "states": states, + "default_state": def, + "terminal_states": terminal, + "transitions": transitions, + } + } + rows.Close() + if rows.Err() != nil { + return nil, rows.Err() + } + + entityTypes := map[string]any{} + rows, err = pool.Query(ctx, + `SELECT name, COALESCE(parent_type,''), is_abstract, domain, layer, + COALESCE(description,''), COALESCE(lifecycle_id,''), attribute_schema + FROM entity_types ORDER BY name`) + if err != nil { + return nil, err + } + for rows.Next() { + var name, parent, dom, layer, desc, lc string + var isAbstract bool + var schemaJSON []byte + if err := rows.Scan(&name, &parent, &isAbstract, &dom, &layer, &desc, &lc, &schemaJSON); err != nil { + rows.Close() + return nil, err + } + et := map[string]any{"domain": dom, "layer": layer} + if parent != "" { + et["parent"] = parent + } + if isAbstract { + et["abstract"] = true + } + if desc != "" { + et["description"] = desc + } + if lc != "" { + et["lifecycle"] = lc + } + if len(schemaJSON) > 0 && string(schemaJSON) != "null" { + var schema map[string]any + if err := json.Unmarshal(schemaJSON, &schema); err == nil && schema != nil { + et["attributes"] = schema + } + } + entityTypes[name] = et + } + rows.Close() + if rows.Err() != nil { + return nil, rows.Err() + } + + relTypes := map[string]any{} + rows, err = pool.Query(ctx, + `SELECT name, COALESCE(inverse,''), source_type, target_type, cardinality, + COALESCE(description,'') + FROM relationship_types ORDER BY name`) + if err != nil { + return nil, err + } + for rows.Next() { + var name, inverse, src, tgt, card, desc string + if err := rows.Scan(&name, &inverse, &src, &tgt, &card, &desc); err != nil { + rows.Close() + return nil, err + } + rt := map[string]any{"source": src, "target": tgt, "cardinality": card} + if inverse != "" { + rt["inverse"] = inverse + } + if desc != "" { + rt["description"] = desc + } + relTypes[name] = rt + } + rows.Close() + if rows.Err() != nil { + return nil, rows.Err() + } + + return map[string]any{ + "version": 1, + "lifecycles": lifecycles, + "entity_types": entityTypes, + "relationship_types": relTypes, + }, nil +} + +func exportInventory(ctx context.Context, pool *Pool) (map[string]any, error) { + var entities []any + rows, err := pool.Query(ctx, ` + SELECT e.slug, e.type, e.name, COALESCE(e.state,''), e.attributes + FROM entities e + JOIN entity_types et ON et.name = e.type + WHERE et.layer <> 'cognition' + ORDER BY e.slug`) + if err != nil { + return nil, err + } + for rows.Next() { + var slug, typ, name, state string + var attrsJSON []byte + if err := rows.Scan(&slug, &typ, &name, &state, &attrsJSON); err != nil { + rows.Close() + return nil, err + } + e := map[string]any{"slug": slug, "type": typ, "name": name} + if state != "" { + e["state"] = state + } + var attrs map[string]any + if err := json.Unmarshal(attrsJSON, &attrs); err == nil && len(attrs) > 0 { + e["attributes"] = attrs + } + entities = append(entities, e) + } + rows.Close() + if rows.Err() != nil { + return nil, rows.Err() + } + + var rels []any + rows, err = pool.Query(ctx, ` + SELECT se.slug, te.slug, r.type, r.attributes + FROM relationships r + JOIN entities se ON se.id = r.source_id + JOIN entities te ON te.id = r.target_id + WHERE r.valid_to IS NULL + ORDER BY r.type, se.slug, te.slug`) + if err != nil { + return nil, err + } + for rows.Next() { + var src, tgt, typ string + var attrsJSON []byte + if err := rows.Scan(&src, &tgt, &typ, &attrsJSON); err != nil { + rows.Close() + return nil, err + } + rel := map[string]any{"source": src, "target": tgt, "type": typ} + var attrs map[string]any + if len(attrsJSON) > 0 && json.Unmarshal(attrsJSON, &attrs) == nil && len(attrs) > 0 { + rel["attributes"] = attrs + } + rels = append(rels, rel) + } + rows.Close() + if rows.Err() != nil { + return nil, rows.Err() + } + + return map[string]any{ + "version": 1, + "entities": entities, + "relationships": rels, + }, nil +} + +func exportPolicy(ctx context.Context, pool *Pool) (map[string]any, error) { + riskClasses := map[string]any{} + rows, err := pool.Query(ctx, + `SELECT name, COALESCE(description,''), approval_required, autonomy_allowed + FROM risk_classes ORDER BY name`) + if err != nil { + return nil, err + } + for rows.Next() { + var name, desc, approval string + var autonomy bool + if err := rows.Scan(&name, &desc, &approval, &autonomy); err != nil { + rows.Close() + return nil, err + } + rc := map[string]any{"approval_required": approval, "autonomy_allowed": autonomy} + if desc != "" { + rc["description"] = desc + } + riskClasses[name] = rc + } + rows.Close() + if rows.Err() != nil { + return nil, rows.Err() + } + + var rules []any + rows, err = pool.Query(ctx, ` + SELECT COALESCE(ar.entity_type,''), ar.action, ar.risk_class, + ar.autonomy_level, COALESCE(se.slug,'') + FROM approval_rules ar + LEFT JOIN entities se ON se.id = ar.scope_entity + ORDER BY COALESCE(ar.entity_type,''), ar.action, COALESCE(se.slug,'')`) + if err != nil { + return nil, err + } + for rows.Next() { + var et, action, rc, al, scope string + if err := rows.Scan(&et, &action, &rc, &al, &scope); err != nil { + rows.Close() + return nil, err + } + rule := map[string]any{"action": action, "risk_class": rc, "autonomy_level": al} + if et != "" { + rule["entity_type"] = et + } + if scope != "" { + rule["scope_entity"] = scope + } + rules = append(rules, rule) + } + rows.Close() + if rows.Err() != nil { + return nil, rows.Err() + } + + settings := map[string]any{} + rows, err = pool.Query(ctx, `SELECT key, value FROM autonomy_settings ORDER BY key`) + if err != nil { + return nil, err + } + for rows.Next() { + var k, v string + if err := rows.Scan(&k, &v); err != nil { + rows.Close() + return nil, err + } + settings[k] = v + } + rows.Close() + if rows.Err() != nil { + return nil, rows.Err() + } + + return map[string]any{ + "version": 1, + "risk_classes": riskClasses, + "approval_rules": rules, + "autonomy_settings": settings, + }, nil +} diff --git a/internal/db/integration_test.go b/internal/db/integration_test.go new file mode 100644 index 0000000..7a103bf --- /dev/null +++ b/internal/db/integration_test.go @@ -0,0 +1,338 @@ +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/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) + } +} + +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 + 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:caddy": 1, "service:authentik": 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) + } + } + if len(got) != len(want) { + t.Errorf("blast_radius returned %d nodes, want %d: %v", len(got), len(want), got) + } +} + +// 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"])) + } +} diff --git a/internal/db/pool.go b/internal/db/pool.go index 3ce64fd..1efd316 100644 --- a/internal/db/pool.go +++ b/internal/db/pool.go @@ -39,11 +39,27 @@ func New(ctx context.Context, databaseURL string) (*Pool, error) { return &Pool{pool}, nil } +// migrationLockKey is the advisory-lock key serializing migration runs — +// two concurrent `oikos migrate` invocations must not interleave DDL. +const migrationLockKey = 0x01c05e5 + // Migrate runs all embedded forward migrations in order. -// Uses a schema_migrations table to track applied versions. +// Uses a schema_migrations table to track applied versions. The whole run +// happens on one connection holding a session advisory lock. func (p *Pool) Migrate(ctx context.Context) error { + conn, err := p.Acquire(ctx) + if err != nil { + return fmt.Errorf("acquire migration conn: %w", err) + } + defer conn.Release() + + if _, err := conn.Exec(ctx, "SELECT pg_advisory_lock($1)", migrationLockKey); err != nil { + return fmt.Errorf("acquire migration lock: %w", err) + } + defer conn.Exec(context.WithoutCancel(ctx), "SELECT pg_advisory_unlock($1)", migrationLockKey) + // Create tracking table if not exists - _, err := p.Exec(ctx, ` + _, err = conn.Exec(ctx, ` CREATE TABLE IF NOT EXISTS schema_migrations ( version INT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now() @@ -76,7 +92,7 @@ func (p *Pool) Migrate(ctx context.Context) error { // Check if already applied var applied int - err := p.QueryRow(ctx, + err := conn.QueryRow(ctx, "SELECT COUNT(*) FROM schema_migrations WHERE version = $1", version).Scan(&applied) if err != nil { return fmt.Errorf("check migration %d: %w", version, err) @@ -99,12 +115,12 @@ func (p *Pool) Migrate(ctx context.Context) error { if stmt == "" { continue } - _, err := p.Exec(ctx, stmt) + _, err := conn.Exec(ctx, stmt) if err != nil { return fmt.Errorf("exec migration %s stmt %d: %w", fname, i+1, err) } } - _, err = p.Exec(ctx, "INSERT INTO schema_migrations (version) VALUES ($1)", version) + _, err = conn.Exec(ctx, "INSERT INTO schema_migrations (version) VALUES ($1)", version) if err != nil { return fmt.Errorf("record migration %d: %w", version, err) } diff --git a/internal/db/seed.go b/internal/db/seed.go index 00892af..3b75b48 100644 --- a/internal/db/seed.go +++ b/internal/db/seed.go @@ -3,13 +3,11 @@ package db import ( "context" "encoding/json" + "errors" "fmt" - "time" - "github.com/dtoro/oikos/internal/domain" "github.com/google/uuid" "github.com/jackc/pgx/v5" - "gopkg.in/yaml.v3" ) // SeedResult holds counts from a seed ingest operation. @@ -83,11 +81,20 @@ func IngestOntologySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (*S } // IngestInventorySeed ingests seeds/inventory.yaml into the DB. +// Every entity and edge is validated against the ontology (abstract types +// rejected, lifecycle states checked, relationship endpoints hierarchy- +// validated, cardinality enforced) — a violating seed rolls back atomically. func IngestInventorySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (*SeedResult, error) { r := &SeedResult{} + tree, err := LoadTypeTree(ctx, tx) + if err != nil { + return nil, fmt.Errorf("load type tree: %w", err) + } + // Entities entities, _ := data["entities"].([]any) + entityTypes := make(map[string]string) // slug -> type, for edge validation for _, raw := range entities { eMap, ok := raw.(map[string]any) if !ok { @@ -99,7 +106,14 @@ func IngestInventorySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (* state, _ := eMap["state"].(string) attrs := eMap["attributes"] - // Generate UUIDv7 for new entities, or find existing by slug + if err := tree.ValidateEntity(typeName, state); err != nil { + return nil, fmt.Errorf("entity %s: %w", slug, err) + } + if state == "" { + state = tree.DefaultState(typeName) + } + entityTypes[slug] = typeName + entityID, err := getOrCreateEntityID(ctx, tx, slug) if err != nil { return nil, fmt.Errorf("entity %s: %w", slug, err) @@ -118,7 +132,8 @@ func IngestInventorySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (* r.Entities++ } - // Relationships + // Relationships — upsert against the current-edge partial unique index + // (migration 007) so re-ingest never duplicates edges. rels, _ := data["relationships"].([]any) for _, raw := range rels { relMap, ok := raw.(map[string]any) @@ -139,12 +154,32 @@ func IngestInventorySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (* return nil, fmt.Errorf("rel to %s: %w", target, err) } + srcType := entityTypes[source] + tgtType := entityTypes[target] + if srcType == "" || tgtType == "" { // entity pre-existing in DB, not in this seed + if srcType == "" { + srcType, err = getEntityTypeBySlug(ctx, tx, source) + if err != nil { + return nil, err + } + } + if tgtType == "" { + tgtType, err = getEntityTypeBySlug(ctx, tx, target) + if err != nil { + return nil, err + } + } + } + if err := tree.ValidateEdge(relType, srcType, tgtType); err != nil { + return nil, fmt.Errorf("rel %s→%s: %w", source, target, err) + } + attrsBytes, _ := json.Marshal(attrs) _, err = tx.Exec(ctx, `INSERT INTO relationships (source_id, target_id, type, attributes, valid_from, valid_to) VALUES ($1, $2, $3, $4, now(), NULL) - ON CONFLICT (source_id, target_id, type, valid_from) - DO UPDATE SET attributes = $4`, + ON CONFLICT (source_id, target_id, type) WHERE valid_to IS NULL + DO UPDATE SET attributes = EXCLUDED.attributes`, sourceID, targetID, relType, string(attrsBytes)) if err != nil { return nil, fmt.Errorf("rel %s→%s %s: %w", source, target, relType, err) @@ -152,6 +187,10 @@ func IngestInventorySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (* r.Relationships++ } + if err := ValidateCardinality(ctx, tx); err != nil { + return nil, err + } + return r, nil } @@ -231,15 +270,29 @@ func IngestPolicySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (*See // --- Helpers --- -// getOrCreateEntityID returns the UUID for a slug, generating a new UUIDv7 if not found. +// getOrCreateEntityID returns the UUID for a slug, generating a new +// time-ordered UUIDv7 if the slug doesn't exist yet (ADR-0005). func getOrCreateEntityID(ctx context.Context, tx pgx.Tx, slug string) (uuid.UUID, error) { var id uuid.UUID err := tx.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", slug).Scan(&id) - if err == nil { + switch { + case err == nil: return id, nil + case errors.Is(err, pgx.ErrNoRows): + return uuid.NewV7() + default: + return uuid.Nil, fmt.Errorf("lookup slug %s: %w", slug, err) } - // Generate a time-ordered UUID (using uuid.New for now; UUIDv7 in production) - return uuid.New(), nil +} + +// getEntityTypeBySlug resolves a slug to its entity type name. +func getEntityTypeBySlug(ctx context.Context, tx pgx.Tx, slug string) (string, error) { + var t string + err := tx.QueryRow(ctx, "SELECT type FROM entities WHERE slug = $1", slug).Scan(&t) + if err != nil { + return "", fmt.Errorf("resolve type of %s: %w", slug, err) + } + return t, nil } // getEntityIDBySlug resolves a slug to its UUID. @@ -338,49 +391,3 @@ func keysOf(m map[string]map[string]any) []string { return keys } -// ExportToYAML regenerates the three seed YAMLs from the DB (for DR / version control, D6). -func ExportToYAML(ctx context.Context, pool *Pool) (map[string][]byte, error) { - result := make(map[string][]byte) - - // Export ontology - onto, err := exportOntology(ctx, pool) - if err != nil { - return nil, fmt.Errorf("export ontology: %w", err) - } - result["ontology.yaml"], _ = yaml.Marshal(onto) - - // Export inventory - inv, err := exportInventory(ctx, pool) - if err != nil { - return nil, fmt.Errorf("export inventory: %w", err) - } - result["inventory.yaml"], _ = yaml.Marshal(inv) - - // Export policy - pol, err := exportPolicy(ctx, pool) - if err != nil { - return nil, fmt.Errorf("export policy: %w", err) - } - result["policy.yaml"], _ = yaml.Marshal(pol) - - return result, nil -} - -func exportOntology(ctx context.Context, pool *Pool) (map[string]any, error) { - // TODO: implement full export from DB - return map[string]any{"version": 1}, nil -} - -func exportInventory(ctx context.Context, pool *Pool) (map[string]any, error) { - // TODO: implement full export from DB - return map[string]any{"version": 1}, nil -} - -func exportPolicy(ctx context.Context, pool *Pool) (map[string]any, error) { - // TODO: implement full export from DB - return map[string]any{"version": 1}, nil -} - -// Unused import suppression for domain (will be needed when we add more logic) -var _ = domain.Entity{} -var _ = time.Now diff --git a/internal/db/splitsql_test.go b/internal/db/splitsql_test.go new file mode 100644 index 0000000..ecf4e07 --- /dev/null +++ b/internal/db/splitsql_test.go @@ -0,0 +1,53 @@ +package db + +import ( + "strings" + "testing" +) + +func nonEmpty(stmts []string) []string { + var out []string + for _, s := range stmts { + if strings.TrimSpace(s) != "" { + out = append(out, s) + } + } + return out +} + +func TestSplitSQLBasic(t *testing.T) { + stmts := nonEmpty(splitSQL("CREATE TABLE a (id int); CREATE TABLE b (id int);")) + if len(stmts) != 2 { + t.Fatalf("got %d statements, want 2: %#v", len(stmts), stmts) + } +} + +func TestSplitSQLDollarQuotedFunction(t *testing.T) { + sql := `CREATE FUNCTION f() RETURNS int AS $$ + SELECT 1; SELECT 2; +$$ LANGUAGE sql; +CREATE TABLE t (id int);` + stmts := nonEmpty(splitSQL(sql)) + if len(stmts) != 2 { + t.Fatalf("got %d statements, want 2: %#v", len(stmts), stmts) + } + if !strings.Contains(stmts[0], "SELECT 1; SELECT 2;") { + t.Errorf("dollar-quoted body was split: %q", stmts[0]) + } +} + +func TestSplitSQLTaggedDollarQuote(t *testing.T) { + sql := `DO $body$ BEGIN PERFORM 1; END $body$;SELECT 1;` + stmts := nonEmpty(splitSQL(sql)) + if len(stmts) != 2 { + t.Fatalf("got %d statements, want 2: %#v", len(stmts), stmts) + } +} + +func TestSplitSQLSemicolonInComment(t *testing.T) { + sql := "-- comment with ; semicolon\nCREATE TABLE t (id int); -- trailing; note\nSELECT 1;" + stmts := nonEmpty(splitSQL(sql)) + if len(stmts) != 2 { + t.Fatalf("got %d statements, want 2: %#v", len(stmts), stmts) + } +} diff --git a/internal/db/typetree.go b/internal/db/typetree.go new file mode 100644 index 0000000..c848ef1 --- /dev/null +++ b/internal/db/typetree.go @@ -0,0 +1,143 @@ +package db + +import ( + "context" + "fmt" + + "github.com/dtoro/oikos/internal/ontology" + "github.com/jackc/pgx/v5" +) + +// LoadTypeTree loads the ontology meta-schema from the DB for validation. +// Called within the same transaction as an ingest so it sees just-ingested +// types. +func LoadTypeTree(ctx context.Context, tx pgx.Tx) (*ontology.TypeTree, error) { + t := &ontology.TypeTree{ + Types: map[string]ontology.TypeInfo{}, + RelTypes: map[string]ontology.RelTypeInfo{}, + Lifecycles: map[string]ontology.LifecycleInfo{}, + } + + rows, err := tx.Query(ctx, + `SELECT name, COALESCE(parent_type,''), is_abstract, COALESCE(lifecycle_id,'') + FROM entity_types`) + if err != nil { + return nil, fmt.Errorf("load entity_types: %w", err) + } + for rows.Next() { + var name string + var info ontology.TypeInfo + if err := rows.Scan(&name, &info.Parent, &info.IsAbstract, &info.LifecycleID); err != nil { + rows.Close() + return nil, err + } + t.Types[name] = info + } + rows.Close() + if rows.Err() != nil { + return nil, rows.Err() + } + + rows, err = tx.Query(ctx, + `SELECT name, source_type, target_type, cardinality FROM relationship_types`) + if err != nil { + return nil, fmt.Errorf("load relationship_types: %w", err) + } + for rows.Next() { + var name string + var info ontology.RelTypeInfo + if err := rows.Scan(&name, &info.SourceType, &info.TargetType, &info.Cardinality); err != nil { + rows.Close() + return nil, err + } + t.RelTypes[name] = info + } + rows.Close() + if rows.Err() != nil { + return nil, rows.Err() + } + + rows, err = tx.Query(ctx, + `SELECT id, states, default_state FROM lifecycle_defs`) + if err != nil { + return nil, fmt.Errorf("load lifecycle_defs: %w", err) + } + for rows.Next() { + var id, def string + var states []string + if err := rows.Scan(&id, &states, &def); err != nil { + rows.Close() + return nil, err + } + set := make(map[string]bool, len(states)) + for _, s := range states { + set[s] = true + } + t.Lifecycles[id] = ontology.LifecycleInfo{States: set, DefaultState: def} + } + rows.Close() + if rows.Err() != nil { + return nil, rows.Err() + } + + return t, nil +} + +// ValidateCardinality checks all CURRENT edges against their relationship +// type's declared cardinality. Runs inside the ingest transaction so a +// violating seed rolls back atomically. +func ValidateCardinality(ctx context.Context, tx pgx.Tx) error { + // one-to-one / many-to-one: a source may have at most one outgoing + // current edge of the type. + rows, err := tx.Query(ctx, ` + SELECT r.type, se.slug, count(*) + FROM relationships r + JOIN relationship_types rt ON rt.name = r.type + JOIN entities se ON se.id = r.source_id + WHERE r.valid_to IS NULL AND rt.cardinality IN ('one-to-one','many-to-one') + GROUP BY r.type, se.slug HAVING count(*) > 1`) + if err != nil { + return err + } + violations, err := collectViolations(rows, "source") + if err != nil { + return err + } + + // one-to-one / one-to-many: a target may have at most one incoming + // current edge of the type. + rows, err = tx.Query(ctx, ` + SELECT r.type, te.slug, count(*) + FROM relationships r + JOIN relationship_types rt ON rt.name = r.type + JOIN entities te ON te.id = r.target_id + WHERE r.valid_to IS NULL AND rt.cardinality IN ('one-to-one','one-to-many') + GROUP BY r.type, te.slug HAVING count(*) > 1`) + if err != nil { + return err + } + v2, err := collectViolations(rows, "target") + if err != nil { + return err + } + violations = append(violations, v2...) + + if len(violations) > 0 { + return fmt.Errorf("cardinality violations: %v", violations) + } + return nil +} + +func collectViolations(rows pgx.Rows, side string) ([]string, error) { + defer rows.Close() + var out []string + for rows.Next() { + var relType, slug string + var n int + if err := rows.Scan(&relType, &slug, &n); err != nil { + return nil, err + } + out = append(out, fmt.Sprintf("%s %s=%s (%d edges)", relType, side, slug, n)) + } + return out, rows.Err() +} diff --git a/internal/ontology/validate.go b/internal/ontology/validate.go new file mode 100644 index 0000000..4c88d87 --- /dev/null +++ b/internal/ontology/validate.go @@ -0,0 +1,112 @@ +// Package ontology implements the meta-schema logic: the entity-type +// hierarchy (is-a with abstract types), relationship endpoint validation, +// cardinality enforcement, and lifecycle state checks. Both the seed +// ingest and the API mutation paths validate through this package so the +// graph can never violate the ontology (plan R3-1). +package ontology + +import ( + "fmt" + + "github.com/dtoro/oikos/internal/domain" +) + +// TypeInfo is the subset of an entity type the validator needs. +type TypeInfo struct { + Parent string + IsAbstract bool + LifecycleID string +} + +// RelTypeInfo is the subset of a relationship type the validator needs. +type RelTypeInfo struct { + SourceType string + TargetType string + Cardinality string +} + +// LifecycleInfo is the subset of a lifecycle the validator needs. +type LifecycleInfo struct { + States map[string]bool + DefaultState string +} + +// TypeTree holds the loaded ontology meta-schema for validation. +type TypeTree struct { + Types map[string]TypeInfo + RelTypes map[string]RelTypeInfo + Lifecycles map[string]LifecycleInfo +} + +// IsA reports whether typ is target or a descendant of it. +func (t *TypeTree) IsA(typ, target string) bool { + seen := map[string]bool{} + for cur := typ; cur != ""; cur = t.Types[cur].Parent { + if cur == target { + return true + } + if seen[cur] { + return false // cycle guard — ingest rejects cycles, belt and braces + } + seen[cur] = true + if _, ok := t.Types[cur]; !ok { + return false + } + } + return false +} + +// ValidateEntity checks that typ exists, is not abstract, and that state +// (if set) is legal for the type's lifecycle. +func (t *TypeTree) ValidateEntity(typ, state string) error { + info, ok := t.Types[typ] + if !ok { + return fmt.Errorf("%w: entity type %q", domain.ErrNotFound, typ) + } + if info.IsAbstract { + return fmt.Errorf("%w: %q", domain.ErrAbstractType, typ) + } + if state == "" { + return nil + } + if info.LifecycleID == "" { + return fmt.Errorf("%w: type %q has no lifecycle but state %q given", + domain.ErrInvalidTransition, typ, state) + } + lc, ok := t.Lifecycles[info.LifecycleID] + if !ok { + return fmt.Errorf("%w: lifecycle %q", domain.ErrNotFound, info.LifecycleID) + } + if !lc.States[state] { + return fmt.Errorf("%w: state %q not in lifecycle %q", + domain.ErrInvalidTransition, state, info.LifecycleID) + } + return nil +} + +// ValidateEdge checks that relType exists and that the endpoint entity +// types are the declared source/target types or descendants of them. +func (t *TypeTree) ValidateEdge(relType, sourceEntityType, targetEntityType string) error { + rt, ok := t.RelTypes[relType] + if !ok { + return fmt.Errorf("%w: relationship type %q", domain.ErrNotFound, relType) + } + if !t.IsA(sourceEntityType, rt.SourceType) { + return fmt.Errorf("%w: %s source %q is not a %q", + domain.ErrInvalidEdge, relType, sourceEntityType, rt.SourceType) + } + if !t.IsA(targetEntityType, rt.TargetType) { + return fmt.Errorf("%w: %s target %q is not a %q", + domain.ErrInvalidEdge, relType, targetEntityType, rt.TargetType) + } + return nil +} + +// DefaultState returns the default lifecycle state for a type ("" if none). +func (t *TypeTree) DefaultState(typ string) string { + info, ok := t.Types[typ] + if !ok || info.LifecycleID == "" { + return "" + } + return t.Lifecycles[info.LifecycleID].DefaultState +} diff --git a/internal/ontology/validate_test.go b/internal/ontology/validate_test.go new file mode 100644 index 0000000..b8af035 --- /dev/null +++ b/internal/ontology/validate_test.go @@ -0,0 +1,120 @@ +package ontology + +import ( + "errors" + "testing" + + "github.com/dtoro/oikos/internal/domain" +) + +func fixtureTree() *TypeTree { + return &TypeTree{ + Types: map[string]TypeInfo{ + "entity": {IsAbstract: true}, + "compute-entity": {Parent: "entity", IsAbstract: true}, + "machine": {Parent: "compute-entity", IsAbstract: true}, + "proxmox-host": {Parent: "machine", LifecycleID: "infrastructure"}, + "lxc": {Parent: "compute-entity", LifecycleID: "infrastructure"}, + "service": {Parent: "entity", LifecycleID: "infrastructure"}, + "document": {Parent: "entity"}, // no lifecycle + }, + RelTypes: map[string]RelTypeInfo{ + "hosts": {SourceType: "machine", TargetType: "compute-entity", Cardinality: "one-to-many"}, + "provides": {SourceType: "compute-entity", TargetType: "service", Cardinality: "one-to-many"}, + "depends-on": {SourceType: "service", TargetType: "service", Cardinality: "many-to-many"}, + "documents": {SourceType: "document", TargetType: "entity", Cardinality: "many-to-one"}, + }, + Lifecycles: map[string]LifecycleInfo{ + "infrastructure": { + States: map[string]bool{"planned": true, "active": true, "destroyed": true}, + DefaultState: "active", + }, + }, + } +} + +func TestIsAWalksHierarchy(t *testing.T) { + tree := fixtureTree() + cases := []struct { + typ, target string + want bool + }{ + {"proxmox-host", "machine", true}, + {"proxmox-host", "compute-entity", true}, + {"proxmox-host", "entity", true}, + {"proxmox-host", "proxmox-host", true}, + {"lxc", "machine", false}, + {"service", "compute-entity", false}, + {"nonexistent", "entity", false}, + } + for _, c := range cases { + if got := tree.IsA(c.typ, c.target); got != c.want { + t.Errorf("IsA(%q, %q) = %v, want %v", c.typ, c.target, got, c.want) + } + } +} + +func TestValidateEntityRejectsAbstract(t *testing.T) { + tree := fixtureTree() + for _, abstract := range []string{"entity", "compute-entity", "machine"} { + if err := tree.ValidateEntity(abstract, ""); !errors.Is(err, domain.ErrAbstractType) { + t.Errorf("ValidateEntity(%q) = %v, want ErrAbstractType", abstract, err) + } + } + if err := tree.ValidateEntity("lxc", "active"); err != nil { + t.Errorf("ValidateEntity(lxc, active) = %v, want nil", err) + } +} + +func TestValidateEntityStates(t *testing.T) { + tree := fixtureTree() + if err := tree.ValidateEntity("lxc", "flying"); !errors.Is(err, domain.ErrInvalidTransition) { + t.Errorf("bad state = %v, want ErrInvalidTransition", err) + } + // state on a type without a lifecycle is rejected + if err := tree.ValidateEntity("document", "active"); !errors.Is(err, domain.ErrInvalidTransition) { + t.Errorf("state without lifecycle = %v, want ErrInvalidTransition", err) + } + // unknown type + if err := tree.ValidateEntity("ghost", ""); !errors.Is(err, domain.ErrNotFound) { + t.Errorf("unknown type = %v, want ErrNotFound", err) + } +} + +func TestValidateEdgeHonorsInheritance(t *testing.T) { + tree := fixtureTree() + // proxmox-host is-a machine; lxc is-a compute-entity → valid + if err := tree.ValidateEdge("hosts", "proxmox-host", "lxc"); err != nil { + t.Errorf("hosts(proxmox-host→lxc) = %v, want nil", err) + } + // abstract endpoint declared, concrete descendant offered → valid + if err := tree.ValidateEdge("provides", "lxc", "service"); err != nil { + t.Errorf("provides(lxc→service) = %v, want nil", err) + } + // documents targets the root abstract 'entity' → anything is valid + if err := tree.ValidateEdge("documents", "document", "proxmox-host"); err != nil { + t.Errorf("documents(document→proxmox-host) = %v, want nil", err) + } + // service is not a machine → invalid source + if err := tree.ValidateEdge("hosts", "service", "lxc"); !errors.Is(err, domain.ErrInvalidEdge) { + t.Errorf("hosts(service→lxc) = %v, want ErrInvalidEdge", err) + } + // lxc is not a service → invalid target + if err := tree.ValidateEdge("depends-on", "service", "lxc"); !errors.Is(err, domain.ErrInvalidEdge) { + t.Errorf("depends-on(service→lxc) = %v, want ErrInvalidEdge", err) + } + // unknown relationship type + if err := tree.ValidateEdge("teleports", "lxc", "service"); !errors.Is(err, domain.ErrNotFound) { + t.Errorf("unknown rel type = %v, want ErrNotFound", err) + } +} + +func TestDefaultState(t *testing.T) { + tree := fixtureTree() + if got := tree.DefaultState("lxc"); got != "active" { + t.Errorf("DefaultState(lxc) = %q, want active", got) + } + if got := tree.DefaultState("document"); got != "" { + t.Errorf("DefaultState(document) = %q, want empty", got) + } +} diff --git a/migrations/007_relationships_current_unique.up.sql b/migrations/007_relationships_current_unique.up.sql new file mode 100644 index 0000000..f044cdc --- /dev/null +++ b/migrations/007_relationships_current_unique.up.sql @@ -0,0 +1,18 @@ +-- Migration 007: one current edge per (source, target, type). +-- The seed upsert previously conflicted on (source_id, target_id, type, +-- valid_from) — valid_from is now() at insert, so the conflict never fired +-- and every re-ingest duplicated all current edges. Dedupe (keep earliest +-- valid_from), then enforce uniqueness on current edges with a partial +-- unique index the upsert can target. + +DELETE FROM relationships r +USING relationships keep +WHERE r.valid_to IS NULL AND keep.valid_to IS NULL + AND r.source_id = keep.source_id + AND r.target_id = keep.target_id + AND r.type = keep.type + AND r.valid_from > keep.valid_from; + +CREATE UNIQUE INDEX IF NOT EXISTS uq_rel_current + ON relationships(source_id, target_id, type) + WHERE valid_to IS NULL;