From 9e3783734eb371064497c873de70ad8473eb64cd Mon Sep 17 00:00:00 2001 From: dtoro Date: Sat, 15 Aug 2026 23:38:21 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20Phase=203b=20=E2=80=94=20EntityService,?= =?UTF-8?q?=20postgres=20EntityRepository,=20converged=20mutations?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- VERSION | 2 +- .../adapters/postgres/integration_test.go | 2 +- internal/adapters/postgres/repositories.go | 506 ++++++++++++++++++ internal/core/app/entities.go | 262 +++++++++ internal/core/ports/entities.go | 88 ++- internal/core/ports/events.go | 14 +- internal/core/ports/portstest/fakes.go | 70 ++- internal/httpapi/entity_mutations.go | 312 ++++------- internal/httpapi/server.go | 11 +- internal/mcp/create_entity_test.go | 4 +- internal/mcp/entity_tools.go | 151 ++---- internal/mcp/secrets_tools_test.go | 2 +- internal/mcp/server.go | 9 +- internal/mcp/server_test.go | 2 +- internal/mcp/tools.go | 4 +- 15 files changed, 1071 insertions(+), 368 deletions(-) create mode 100644 internal/adapters/postgres/repositories.go create mode 100644 internal/core/app/entities.go diff --git a/VERSION b/VERSION index eb771362..be386c9e 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.32.4 +0.33.0 diff --git a/internal/adapters/postgres/integration_test.go b/internal/adapters/postgres/integration_test.go index 07353700..2ed6eedd 100644 --- 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. diff --git a/internal/adapters/postgres/repositories.go b/internal/adapters/postgres/repositories.go new file mode 100644 index 00000000..c7cad321 --- /dev/null +++ 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 +} diff --git a/internal/core/app/entities.go b/internal/core/app/entities.go new file mode 100644 index 00000000..4067e28e --- /dev/null +++ b/internal/core/app/entities.go @@ -0,0 +1,262 @@ +package app + +import ( + "context" + "encoding/json" + "time" + + "github.com/dtoro/oikos/internal/core/domain" + "github.com/dtoro/oikos/internal/core/ports" + "github.com/google/uuid" +) + +// EntityService owns the entity aggregate's use-cases: create with derived +// checks, update with optimistic versioning and check regeneration, +// lifecycle transitions. Validation against the (cached) ontology happens +// here, in the pure half; transactional check-then-act invariants +// (version, declared transitions, preconditions) run in the repository +// (plan §3.6). +type EntityService struct { + entities ports.EntityRepository + onto ports.OntologyStore +} + +// NewEntityService wires the service. +func NewEntityService(entities ports.EntityRepository, onto ports.OntologyStore) *EntityService { + return &EntityService{entities: entities, onto: onto} +} + +// CreateEntityCmd is one entity creation. Actor identifies the calling +// surface for the audit trail ("operator: