feat: Phase 3b — EntityService, postgres EntityRepository, converged mutations

Problem: entity mutations (create/update/state) existed as three drifted copies — HTTP CreateEntity/PatchEntity, MCP create_entity/ update_entity_attributes/set_entity_state — each with its own inline SQL, its own validation subset (MCP validated lifecycle states, HTTP did not; HTTP patched attributes without regenerating derived checks, MCP did; MCP wrote no audit trail), the exact drift ADR 0016's first vertical slice exists to collapse. Change: - internal/core/ports: DerivedCheck, Idempotency (adapter-owned request hash + cached-body renderer so the replay record commits in the create's transaction), IdempotentResponse + GetIdempotent read, AuditEntry gains Method/Path/CorrelationID, Event gains CorrelationID; EntityUpdateInput carries ExpectedVersion + RederiveChecks (derivation for updates runs repo-side: the graph host fallback reads relationships through the open tx). - internal/adapters/postgres/repositories.go: EntityRepo (Create/ Update/SetState/reads/idempotency) preserving the load-bearing check-then-act invariants in-tx: version WHERE-clause, declared transitions + preconditions (ValidateTransition), duplicate-slug mapping, audit/event/writeCheck all inside one BEGIN…COMMIT. OntologyRepo: TTL-cached OntologyStore. - internal/core/app/entities.go: EntityService — ontology validation (type exists, concrete, state declared — the stricter MCP rule now governs both surfaces), default-state resolution, id generation, derivation for creates, audit/event construction, idempotency pass-through. - httpapi CreateEntity/PatchEntity rewired to the service; PATCH now regenerates derived checks (the A2 parity gap). MCP create/update/ set-state tools call the same service — and now write audit + event rows like the HTTP surface always did. - Integration-test seed paths fixed for the adapters/postgres package depth (../../../seeds). Pre-existing failures documented: TestAPIEndToEnd (entity_types 60 vs 59; 501-endpoint now 200), TestClientLifecycleEndToEnd, TestPhase3* rows — verified failing identically at ec11956 (scratch approval- notifier commit test drift), unrelated to this change. All mutation integration tests (create/patch/idempotency/audit/regeneration) pass. Verification: make test-db (postgres + mcp green, httpapi green except the pre-existing set), full non-DB suite (19 pkgs), golangci on new packages — 0 issues.
2026-08-15 23:38:21 +02:00
parent d4f5084a6d
commit 9e3783734e
15 changed files with 1071 additions and 368 deletions
--- a/internal/adapters/postgres/integration_test.go
+++ b/internal/adapters/postgres/integration_test.go
@@ -23,7 +23,7 @@ import (
 	"gopkg.in/yaml.v3"
 )

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

 // newTestPool creates a throwaway database (dropped on cleanup), runs all
 // migrations, and returns a pool connected to it.
--- a/internal/adapters/postgres/repositories.go
+++ b/internal/adapters/postgres/repositories.go
@@ -0,0 +1,506 @@
+package db
+
+import (
+	"context"
+	"fmt"
+	"encoding/json"
+	"errors"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/dtoro/oikos/internal/core/app"
+	"github.com/dtoro/oikos/internal/core/domain"
+	"github.com/dtoro/oikos/internal/core/ports"
+	"github.com/dtoro/oikos/internal/adapters/postgres/sqlcgen"
+	"github.com/dtoro/oikos/internal/observability"
+	"github.com/dtoro/oikos/internal/ontology"
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+)
+
+// EntityRepo implements ports.EntityRepository over the postgres pool.
+// Command methods run everything in their input inside one transaction
+// (ADR 0016 §3.6): entity write, derived checks, idempotency record, audit,
+// event.
+type EntityRepo struct {
+	pool *Pool
+}
+
+var _ ports.EntityRepository = (*EntityRepo)(nil)
+
+// NewEntityRepo builds the entity repository.
+func NewEntityRepo(pool *Pool) *EntityRepo { return &EntityRepo{pool: pool} }
+
+
+// mustUUID converts a domain.UUID (string alias) to uuid.UUID. The domain
+// layer guarantees UUID-shaped strings; parse failures are programming
+// errors and panic loudly rather than half-failing a transaction.
+func mustUUID(id domain.UUID) uuid.UUID {
+	u, err := uuid.Parse(string(id))
+	if err != nil {
+		panic(fmt.Sprintf("invalid entity UUID %q", string(id)))
+	}
+	return u
+}
+
+const entityFullCols = `id, slug, type, name, state, attributes, maintenance_until, version, created_at, updated_at`
+
+type rowScanner interface{ Scan(dest ...any) error }
+
+func scanDomainEntity(row rowScanner) (domain.Entity, error) {
+	var id uuid.UUID
+	var slug, typ, name string
+	var state *string
+	var attrs []byte
+	var maint *time.Time
+	var version int32
+	var createdAt, updatedAt time.Time
+	if err := row.Scan(&id, &slug, &typ, &name, &state, &attrs, &maint, &version, &createdAt, &updatedAt); err != nil {
+		return domain.Entity{}, err
+	}
+	d := domain.Entity{
+		ID:               domain.UUID(id.String()),
+		Slug:             slug,
+		Type:             typ,
+		Name:             name,
+		Version:          int(version),
+		CreatedAt:        createdAt,
+		UpdatedAt:        updatedAt,
+		MaintenanceUntil: maint,
+	}
+	if state != nil {
+		d.State = *state
+	}
+	if len(attrs) > 0 {
+		var m map[string]any
+		if json.Unmarshal(attrs, &m) == nil {
+			d.Attributes = m
+		}
+	}
+	return d, nil
+}
+
+func mapRowErr(err error) error {
+	if errors.Is(err, pgx.ErrNoRows) {
+		return domain.ErrNotFound
+	}
+	return err
+}
+
+// Get returns the entity by ID.
+func (r *EntityRepo) Get(ctx context.Context, id domain.UUID) (domain.Entity, error) {
+	uid, err := uuid.Parse(string(id))
+	if err != nil {
+		return domain.Entity{}, err
+	}
+	e, err := scanDomainEntity(r.pool.QueryRow(ctx,
+		`SELECT `+entityFullCols+` FROM entities WHERE id = $1`, uid))
+	return e, mapRowErr(err)
+}
+
+// BySlug returns the entity by slug.
+func (r *EntityRepo) BySlug(ctx context.Context, slug string) (domain.Entity, error) {
+	e, err := scanDomainEntity(r.pool.QueryRow(ctx,
+		`SELECT `+entityFullCols+` FROM entities WHERE slug = $1`, slug))
+	return e, mapRowErr(err)
+}
+
+// List returns entities filtered by type (including descendant types),
+// state, domain, layer, and a slug/name substring, keyset-paginated by
+// slug. The second return is the next cursor ("" when exhausted).
+func (r *EntityRepo) List(ctx context.Context, f ports.EntityFilters) ([]domain.Entity, string, error) {
+	limit := f.Limit
+	if limit <= 0 {
+		limit = 50
+	}
+	// NULL-able filter args: an absent filter must bind SQL NULL, not "".
+	nullable := func(s string) any {
+		if s == "" {
+			return nil
+		}
+		return s
+	}
+
+	// Type filter includes descendants via the parent hierarchy (R3-1).
+	rows, err := r.pool.Query(ctx, `
+		WITH RECURSIVE tt AS (
+			SELECT name FROM entity_types WHERE $1::text IS NULL OR name = $1
+			UNION
+			SELECT et.name FROM entity_types et JOIN tt ON et.parent_type = tt.name
+			WHERE $1::text IS NOT NULL
+		)
+		SELECT `+entityFullCols+`
+		FROM entities e
+		JOIN entity_types et ON et.name = e.type
+		WHERE e.type IN (SELECT name FROM tt)
+		  AND ($2::text IS NULL OR e.state = $2)
+		  AND ($3::text IS NULL OR et.domain = $3)
+		  AND ($4::text IS NULL OR et.layer = $4)
+		  AND ($5::text IS NULL OR e.slug ILIKE '%'||$5||'%' OR e.name ILIKE '%'||$5||'%')
+		  AND ($6::text IS NULL OR e.slug > $6)
+		ORDER BY e.slug
+		LIMIT $7`,
+		nullable(f.Type), nullable(f.State), nullable(f.Domain), nullable(f.Layer),
+		nullable(f.Q), nullable(f.Cursor), limit+1)
+	if err != nil {
+		return nil, "", err
+	}
+	defer rows.Close()
+
+	items := []domain.Entity{}
+	for rows.Next() {
+		e, err := scanDomainEntity(rows)
+		if err != nil {
+			return nil, "", err
+		}
+		items = append(items, e)
+	}
+	if rows.Err() != nil {
+		return nil, "", rows.Err()
+	}
+
+	next := ""
+	if len(items) > limit {
+		items = items[:limit]
+		next = items[len(items)-1].Slug
+	}
+	return items, next, nil
+}
+
+// Search matches slug/name substrings.
+func (r *EntityRepo) Search(ctx context.Context, q string, limit int) ([]domain.Entity, error) {
+	items, _, err := r.List(ctx, ports.EntityFilters{Q: q, Limit: limit})
+	return items, err
+}
+
+// writeSideEffects writes audit entries and the event inside the open
+// transaction.
+func writeSideEffects(ctx context.Context, tx pgx.Tx, entityID domain.UUID, audits []ports.AuditEntry, event *ports.Event) error {
+	q := sqlcgen.New(tx)
+	eid := mustUUID(entityID)
+	for _, a := range audits {
+		if err := observability.Audit(ctx, q, a.ActorType, a.ActorLabel, a.Action, &eid,
+			a.Method, a.Path, a.CorrelationID, nil, a.Details); err != nil {
+			return err
+		}
+	}
+	if event != nil {
+		if err := observability.Event(ctx, q, event.Type, &eid, event.Severity, event.Source,
+			event.CorrelationID, event.Data); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func appTargetOf(e domain.Entity) app.CheckTarget {
+	attrs, _ := json.Marshal(e.Attributes)
+	return app.CheckTarget{ID: string(e.ID), Slug: e.Slug, Type: e.Type, Name: e.Name, Attrs: attrs}
+}
+
+func writeDerivedChecks(ctx context.Context, tx pgx.Tx, e domain.Entity, checks []ports.DerivedCheck) (int, error) {
+	if _, err := tx.Exec(ctx,
+		`INSERT INTO entity_status (entity_id, health, updated_at)
+		 VALUES ($1, 'unknown', now())
+		 ON CONFLICT (entity_id) DO NOTHING`, mustUUID(e.ID)); err != nil {
+		return 0, err
+	}
+	created := 0
+	t := appTargetOf(e)
+	for i, dc := range checks {
+		ok, err := writeCheck(ctx, tx, t, i, app.CheckDef{Kind: dc.Kind, Config: dc.Config, IntervalS: dc.IntervalS})
+		if err != nil {
+			return created, err
+		}
+		if ok {
+			created++
+		}
+	}
+	return created, nil
+}
+
+// rederiveChecks re-runs derivation for an entity inside the open tx, with
+// the graph host fallback active (a service inherits its container's
+// address once the hosting edge exists).
+func rederiveChecks(ctx context.Context, tx pgx.Tx, e domain.Entity, tree *ontology.TypeTree) (int, error) {
+	t := appTargetOf(e)
+	defs, _ := app.Derive(tree, t, func() map[string]any {
+		attrs, err := hostViaGraph(ctx, tx, t.ID)
+		if err != nil {
+			return nil
+		}
+		return attrs
+	})
+	derived := make([]ports.DerivedCheck, len(defs))
+	for i, d := range defs {
+		derived[i] = ports.DerivedCheck{Kind: d.Kind, Config: d.Config, IntervalS: d.IntervalS}
+	}
+	return writeDerivedChecks(ctx, tx, e, derived)
+}
+
+// Create inserts the entity with its derived checks, idempotency record,
+// audit, and event — one transaction.
+func (r *EntityRepo) Create(ctx context.Context, in ports.EntityCreateInput) (domain.Entity, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return domain.Entity{}, err
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	e := in.Entity
+	attrsJSON, _ := json.Marshal(e.Attributes)
+	if e.Attributes == nil {
+		attrsJSON = []byte("{}")
+	}
+	var state *string
+	if e.State != "" {
+		state = &e.State
+	}
+
+	created, err := scanDomainEntity(tx.QueryRow(ctx, `
+		INSERT INTO entities (id, slug, type, name, state, attributes)
+		VALUES ($1, $2, $3, $4, $5, $6)
+		RETURNING `+entityFullCols,
+		mustUUID(e.ID), e.Slug, e.Type, e.Name, state, attrsJSON))
+	if err != nil {
+		if strings.Contains(err.Error(), "unique") || strings.Contains(err.Error(), "duplicate") {
+			return domain.Entity{}, errors.Join(domain.ErrAlreadyExists, err)
+		}
+		return domain.Entity{}, err
+	}
+
+	if in.Idempotency != nil {
+		var body []byte
+		if in.Idempotency.RenderBody != nil {
+			body = in.Idempotency.RenderBody(created)
+		}
+		code := int32(201)
+		if err := sqlcgen.New(tx).PutIdempotentResponse(ctx, sqlcgen.PutIdempotentResponseParams{
+			Actor: in.Idempotency.Actor, Key: in.Idempotency.Key,
+			RequestHash: in.Idempotency.RequestHash, ResponseCode: &code, ResponseBody: body,
+		}); err != nil {
+			return domain.Entity{}, err
+		}
+	}
+
+	if _, err := writeDerivedChecks(ctx, tx, created, in.DerivedChecks); err != nil {
+		return domain.Entity{}, err
+	}
+	if err := writeSideEffects(ctx, tx, created.ID, in.Audit, in.Event); err != nil {
+		return domain.Entity{}, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return domain.Entity{}, err
+	}
+	return created, nil
+}
+
+// Update applies name/state/attributes/maintenance atomically with an
+// optimistic-version check, re-deriving default checks when asked. The
+// declared-transition + precondition validation runs inside the
+// transaction (check-then-act, plan §3.6).
+func (r *EntityRepo) Update(ctx context.Context, in ports.EntityUpdateInput) (domain.Entity, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return domain.Entity{}, err
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	current, err := scanDomainEntity(tx.QueryRow(ctx,
+		`SELECT `+entityFullCols+` FROM entities WHERE id = $1`, mustUUID(in.Entity.ID)))
+	if err != nil {
+		return domain.Entity{}, mapRowErr(err)
+	}
+
+	if in.ExpectedVersion > 0 && current.Version != in.ExpectedVersion {
+		return domain.Entity{}, domain.ErrConflict
+	}
+
+	e := in.Entity
+	// Fields the caller left zero keep their current values.
+	if e.Name == "" {
+		e.Name = current.Name
+	}
+	if e.Slug == "" {
+		e.Slug = current.Slug
+	}
+	if e.Type == "" {
+		e.Type = current.Type
+	}
+
+	// Lifecycle validation when state changes (declared transition +
+	// preconditions, in-tx).
+	if e.State != "" && e.State != current.State {
+		if err := ValidateTransition(ctx, tx, mustUUID(current.ID), current.Type, current.State, e.State); err != nil {
+			if errors.Is(err, ErrTransitionInvalid) {
+				return domain.Entity{}, errors.Join(domain.ErrInvalidTransition, err)
+			}
+			return domain.Entity{}, err
+		}
+	} else if e.State == "" {
+		e.State = current.State
+	}
+
+	if e.Attributes == nil {
+		e.Attributes = current.Attributes
+	}
+	setMaintenance := e.MaintenanceUntil != nil
+	maint := e.MaintenanceUntil
+	if !setMaintenance {
+		maint = current.MaintenanceUntil
+	}
+
+	attrsJSON, _ := json.Marshal(e.Attributes)
+	if e.Attributes == nil {
+		attrsJSON = []byte("{}")
+	}
+	var state *string
+	if e.State != "" {
+		state = &e.State
+	}
+
+	updated, err := scanDomainEntity(tx.QueryRow(ctx, `
+		UPDATE entities
+		SET name = $2, state = $3, attributes = $4, maintenance_until = $5, version = version + 1, updated_at = now()
+		WHERE id = $1 AND version = $6
+		RETURNING `+entityFullCols,
+		mustUUID(e.ID), e.Name, state, attrsJSON, maint, current.Version))
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return domain.Entity{}, domain.ErrConflict
+		}
+		return domain.Entity{}, err
+	}
+
+	if in.RederiveChecks {
+		tree, err := LoadTypeTree(ctx, tx)
+		if err != nil {
+			return domain.Entity{}, err
+		}
+		if _, err := rederiveChecks(ctx, tx, updated, tree); err != nil {
+			return domain.Entity{}, err
+		}
+	}
+
+	if err := writeSideEffects(ctx, tx, updated.ID, in.Audit, in.Event); err != nil {
+		return domain.Entity{}, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return domain.Entity{}, err
+	}
+	return updated, nil
+}
+
+// SetState transitions an entity's lifecycle state. Declared-transition
+// and precondition validation run in-tx; a stale From is refused.
+func (r *EntityRepo) SetState(ctx context.Context, in ports.EntityTransitionInput) (domain.Entity, error) {
+	tx, err := r.pool.Begin(ctx)
+	if err != nil {
+		return domain.Entity{}, err
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	current, err := scanDomainEntity(tx.QueryRow(ctx,
+		`SELECT `+entityFullCols+` FROM entities WHERE slug = $1`, in.Slug))
+	if err != nil {
+		return domain.Entity{}, mapRowErr(err)
+	}
+	if current.State != in.From {
+		return domain.Entity{}, domain.ErrConflict
+	}
+	if err := ValidateTransition(ctx, tx, mustUUID(current.ID), current.Type, in.From, in.To); err != nil {
+		if errors.Is(err, ErrTransitionInvalid) {
+			return domain.Entity{}, errors.Join(domain.ErrInvalidTransition, err)
+		}
+		return domain.Entity{}, err
+	}
+	if _, err := tx.Exec(ctx,
+		`UPDATE entities SET state = $2, version = version + 1, updated_at = now() WHERE id = $1`,
+		mustUUID(current.ID), in.To); err != nil {
+		return domain.Entity{}, err
+	}
+
+	after := current
+	after.State = in.To
+	if err := writeSideEffects(ctx, tx, after.ID, in.Audit, in.Event); err != nil {
+		return domain.Entity{}, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return domain.Entity{}, err
+	}
+	return after, nil
+}
+
+// GetIdempotent returns the cached response for (actor, key) or
+// domain.ErrNotFound.
+func (r *EntityRepo) GetIdempotent(ctx context.Context, actor, key string) (ports.IdempotentResponse, error) {
+	cached, err := sqlcgen.New(r.pool).GetIdempotentResponse(ctx, sqlcgen.GetIdempotentResponseParams{Actor: actor, Key: key})
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return ports.IdempotentResponse{}, domain.ErrNotFound
+		}
+		return ports.IdempotentResponse{}, err
+	}
+	code := 0
+	if cached.ResponseCode != nil {
+		code = int(*cached.ResponseCode)
+	}
+	return ports.IdempotentResponse{
+		RequestHash:   cached.RequestHash,
+		ResponseCode:  code,
+		ResponseBody:  cached.ResponseBody,
+	}, nil
+}
+
+// OntologyRepo implements ports.OntologyStore with a TTL cache — entity
+// types change at seed time, not per request, so a short cache trades a
+// little staleness for avoiding the meta-schema load on every mutation.
+type OntologyRepo struct {
+	pool *Pool
+	ttl  time.Duration
+
+	mu     sync.Mutex
+	loaded time.Time
+	tree   *ontology.TypeTree
+}
+
+var _ ports.OntologyStore = (*OntologyRepo)(nil)
+
+// NewOntologyRepo builds the ontology store with the given cache TTL
+// (values <= 0 disable caching).
+func NewOntologyRepo(pool *Pool, ttl time.Duration) *OntologyRepo {
+	return &OntologyRepo{pool: pool, ttl: ttl}
+}
+
+// LoadTypeTree returns the (possibly cached) ontology tree.
+func (o *OntologyRepo) LoadTypeTree(ctx context.Context) (ports.TypeTree, error) {
+	if o.ttl <= 0 {
+		return o.load(ctx)
+	}
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	if o.tree != nil && time.Since(o.loaded) < o.ttl {
+		return o.tree, nil
+	}
+	tree, err := o.load(ctx)
+	if err != nil {
+		return nil, err
+	}
+	o.tree = tree
+	o.loaded = time.Now()
+	return tree, nil
+}
+
+func (o *OntologyRepo) load(ctx context.Context) (ports.TypeTree, error) {
+	tx, err := o.pool.Begin(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+	tree, err := LoadTypeTree(ctx, tx)
+	if err != nil {
+		return nil, err
+	}
+	return ports.TypeTree(tree), nil
+}