package httpapi import ( "context" "crypto/rand" "crypto/sha256" "encoding/json" "errors" "fmt" "math/big" "strconv" "strings" "time" "github.com/dtoro/oikos/internal/db" "github.com/dtoro/oikos/internal/db/sqlcgen" "github.com/dtoro/oikos/internal/domain" "github.com/dtoro/oikos/internal/httpapi/gen" "github.com/dtoro/oikos/internal/observability" "github.com/google/uuid" "github.com/jackc/pgx/v5" openapi_types "github.com/oapi-codegen/runtime/types" ) const ( defaultLimit = 50 maxLimit = 200 // graphNodeCap bounds the whole-graph view. The cognition transactional // types (execution, task) are audit records, not topology, and previously // crowded out every host/lxc/service; the default whole-graph view below // excludes them so the cap is spent on the actual fleet graph. Operators // still reach executions/tasks via list_entities. graphNodeCap = 2000 ) // actorInfo returns the caller's (type, label) from the request context, // falling back to operator/unknown when unset. func actorInfo(ctx context.Context) (string, string) { if a := GetActor(ctx); a != nil { typ := a.Type if typ == "" { typ = "operator" } label := a.Label if label == "" { label = a.ID } return typ, label } return "operator", "unknown" } func clampLimit(l *int) int { if l == nil { return defaultLimit } if *l < 1 { return 1 } if *l > maxLimit { return maxLimit } return *l } // resolveEntityID resolves a UUID-or-slug path/query value to the entity UUID. func (s *Server) resolveEntityID(ctx context.Context, idOrSlug string) (uuid.UUID, error) { if id, err := uuid.Parse(idOrSlug); err == nil { entity, err := sqlcgen.New(s.pool).GetEntityByID(ctx, id) if err != nil { return uuid.Nil, fmt.Errorf("%w: %s", domain.ErrNotFound, idOrSlug) } return entity.ID, nil } entity, err := sqlcgen.New(s.pool).GetEntityBySlug(ctx, idOrSlug) if err != nil { return uuid.Nil, fmt.Errorf("%w: %s", domain.ErrNotFound, idOrSlug) } return entity.ID, nil } // entityCols requires the entities table to be aliased as `e`, with // entity_status left-joined and aliased as `st` (see withEntityStatus). const entityCols = `e.id, e.slug, e.type, e.name, e.state, e.attributes, e.maintenance_until, e.version, e.created_at, e.updated_at, st.health, st.last_check_at` func scanEntity(row pgx.Row) (gen.Entity, error) { var e gen.Entity var state *string var attrsJSON []byte var maint *time.Time var health *string var lastCheckAt *time.Time err := row.Scan(&e.Id, &e.Slug, &e.Type, &e.Name, &state, &attrsJSON, &maint, &e.Version, &e.CreatedAt, &e.UpdatedAt, &health, &lastCheckAt) if err != nil { return e, err } e.State = state e.MaintenanceUntil = maint if health != nil { h := gen.EntityHealth(*health) e.Health = &h } e.LastCheckAt = lastCheckAt var attrs map[string]any if len(attrsJSON) > 0 && json.Unmarshal(attrsJSON, &attrs) == nil && len(attrs) > 0 { e.Attributes = &attrs } return e, nil } // ─── Entities ───────────────────────────────────────────────────────── func (s *Server) ListEntities(ctx context.Context, req gen.ListEntitiesRequestObject) (gen.ListEntitiesResponseObject, error) { limit := clampLimit(req.Params.Limit) // Type filter includes descendants via the parent hierarchy (R3-1). query := ` 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 ` + entityCols + ` FROM entities e JOIN entity_types et ON et.name = e.type LEFT JOIN entity_status st ON st.entity_id = e.id 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` rows, err := s.pool.Query(ctx, query, req.Params.Type, req.Params.State, req.Params.Domain, req.Params.Layer, req.Params.Q, req.Params.Cursor, limit+1) if err != nil { return nil, err } defer rows.Close() var items []gen.Entity for rows.Next() { e, err := scanEntity(rows) if err != nil { return nil, err } items = append(items, e) } if rows.Err() != nil { return nil, rows.Err() } var next *string if len(items) > limit { items = items[:limit] next = &items[len(items)-1].Slug } if items == nil { items = []gen.Entity{} } return gen.ListEntities200JSONResponse{Items: items, NextCursor: next}, nil } func (s *Server) GetEntity(ctx context.Context, req gen.GetEntityRequestObject) (gen.GetEntityResponseObject, error) { id, err := s.resolveEntityID(ctx, req.Id) if err != nil { return nil, err } e, err := scanEntity(s.pool.QueryRow(ctx, "SELECT "+entityCols+" FROM entities e LEFT JOIN entity_status st ON st.entity_id = e.id WHERE e.id = $1", id)) if err != nil { return nil, err } return gen.GetEntity200JSONResponse{ Body: e, Headers: gen.GetEntity200ResponseHeaders{ETag: `"` + strconv.Itoa(e.Version) + `"`}, }, nil } func (s *Server) GetEntityRelations(ctx context.Context, req gen.GetEntityRelationsRequestObject) (gen.GetEntityRelationsResponseObject, error) { id, err := s.resolveEntityID(ctx, req.Id) if err != nil { return nil, err } dir := "both" if req.Params.Direction != nil { dir = string(*req.Params.Direction) } relType := req.Params.RelType rows, err := sqlcgen.New(s.pool).ListEntityRelations(ctx, sqlcgen.ListEntityRelationsParams{ Direction: dir, ID: id, RelType: relType, }) if err != nil { return nil, err } items := []gen.Relationship{} for _, r := range rows { var attrs *map[string]any if len(r.Attributes) > 0 { var m map[string]any if json.Unmarshal(r.Attributes, &m) == nil && len(m) > 0 { attrs = &m } } validTo := r.ValidTo items = append(items, gen.Relationship{ Source: r.SourceSlug, Target: r.TargetSlug, Type: r.Type, Attributes: attrs, ValidFrom: r.ValidFrom, ValidTo: validTo, }) } return gen.GetEntityRelations200JSONResponse{Items: items}, nil } func (s *Server) GetBlastRadius(ctx context.Context, req gen.GetBlastRadiusRequestObject) (gen.GetBlastRadiusResponseObject, error) { id, err := s.resolveEntityID(ctx, req.Id) if err != nil { return nil, err } depth := 3 if req.Params.Depth != nil { depth = *req.Params.Depth } rows, err := s.pool.Query(ctx, ` SELECT `+entityCols+`, b.depth FROM blast_radius($1, $2) b JOIN entities e ON e.id = b.entity_id LEFT JOIN entity_status st ON st.entity_id = e.id ORDER BY b.depth, e.slug`, id, depth) if err != nil { return nil, err } defer rows.Close() resp := gen.GetBlastRadius200JSONResponse{Items: []struct { Depth int `json:"depth"` Entity gen.Entity `json:"entity"` }{}} for rows.Next() { var e gen.Entity var state *string var attrsJSON []byte var maint *time.Time var health *string var lastCheckAt *time.Time var d int if err := rows.Scan(&e.Id, &e.Slug, &e.Type, &e.Name, &state, &attrsJSON, &maint, &e.Version, &e.CreatedAt, &e.UpdatedAt, &health, &lastCheckAt, &d); err != nil { return nil, err } e.State = state e.MaintenanceUntil = maint if health != nil { h := gen.EntityHealth(*health) e.Health = &h } e.LastCheckAt = lastCheckAt var attrs map[string]any if len(attrsJSON) > 0 && json.Unmarshal(attrsJSON, &attrs) == nil && len(attrs) > 0 { e.Attributes = &attrs } resp.Items = append(resp.Items, struct { Depth int `json:"depth"` Entity gen.Entity `json:"entity"` }{Depth: d, Entity: e}) } return resp, rows.Err() } func (s *Server) GetGraph(ctx context.Context, req gen.GetGraphRequestObject) (gen.GetGraphResponseObject, error) { depth := 2 if req.Params.Depth != nil { depth = *req.Params.Depth } var nodes []gen.Entity var err error truncated := false // pgx can't infer the array element type from a nil *[]string (the // param is absent from the request, not an empty list), so dereference // to a plain []string first — nil there still encodes as SQL NULL, but // pgx has a concrete type to work with. var relTypes []string if req.Params.RelType != nil { relTypes = *req.Params.RelType } if req.Params.Root != nil && *req.Params.Root != "" { rootID, rerr := s.resolveEntityID(ctx, *req.Params.Root) if rerr != nil { return nil, rerr } nodes, err = s.queryEntities(ctx, ` SELECT `+entityCols+` FROM blast_radius($1, $2, $3) b JOIN entities e ON e.id = b.entity_id LEFT JOIN entity_status st ON st.entity_id = e.id ORDER BY e.slug`, rootID, depth, relTypes) } else { // Whole-graph view: pick the most-connected entities first so the // graph shows actual topology, not just whatever sorts first // alphabetically. Without this the cap fills with exec:* rows and // drops every host/lxc/service/vm — and every edge those entities // connect — because edges require both endpoints in the node set. // Exclude the cognition transactional types (execution/task): they // are audit records rather than topology, and at ~380 rows they // consumed most of the old 500-node cap. nodes, err = s.queryEntities(ctx, ` SELECT `+entityCols+` FROM entities e LEFT JOIN entity_status st ON st.entity_id = e.id WHERE e.type NOT IN ('execution','task') AND e.id IN ( SELECT e2.id FROM entities e2 LEFT JOIN relationships r ON r.valid_to IS NULL AND (r.source_id = e2.id OR r.target_id = e2.id) WHERE e2.type NOT IN ('execution','task') GROUP BY e2.id ORDER BY count(r.type) DESC, e2.slug LIMIT $1 ) ORDER BY e.slug`, graphNodeCap+1) if err == nil && len(nodes) > graphNodeCap { nodes = nodes[:graphNodeCap] truncated = true } } if err != nil { return nil, err } ids := make([]uuid.UUID, len(nodes)) for i, n := range nodes { ids[i] = uuid.UUID(n.Id) } edgeRows, err := sqlcgen.New(s.pool).ListGraphEdges(ctx, sqlcgen.ListGraphEdgesParams{ Ids: ids, RelTypes: relTypes, }) if err != nil { return nil, err } edges := []gen.Relationship{} for _, r := range edgeRows { var attrs *map[string]any if len(r.Attributes) > 0 { var m map[string]any if json.Unmarshal(r.Attributes, &m) == nil && len(m) > 0 { attrs = &m } } validTo := r.ValidTo edges = append(edges, gen.Relationship{ Source: r.SourceSlug, Target: r.TargetSlug, Type: r.Type, Attributes: attrs, ValidFrom: r.ValidFrom, ValidTo: validTo, }) } resp := gen.GetGraph200JSONResponse{Nodes: nodes, Edges: edges} if truncated { resp.Truncated = &truncated } if req.Params.Include != nil { for _, inc := range *req.Params.Include { if inc == gen.Status { health, herr := s.entityHealthByID(ctx, ids) if herr != nil { return nil, herr } resp.Health = &health break } } } return resp, nil } // entityHealthByID returns entity_status.health keyed by entity id, for the // given id set (used by GetGraph's include=status). func (s *Server) entityHealthByID(ctx context.Context, ids []uuid.UUID) (map[string]gen.GraphViewHealth, error) { health := make(map[string]gen.GraphViewHealth, len(ids)) rows, err := s.pool.Query(ctx, `SELECT entity_id, health FROM entity_status WHERE entity_id = ANY($1)`, ids) if err != nil { return nil, err } defer rows.Close() for rows.Next() { var id uuid.UUID var h string if err := rows.Scan(&id, &h); err != nil { return nil, err } health[id.String()] = gen.GraphViewHealth(h) } return health, rows.Err() } func (s *Server) queryEntities(ctx context.Context, query string, args ...any) ([]gen.Entity, error) { rows, err := s.pool.Query(ctx, query, args...) if err != nil { return nil, err } defer rows.Close() items := []gen.Entity{} for rows.Next() { e, err := scanEntity(rows) if err != nil { return nil, err } items = append(items, e) } return items, rows.Err() } // ─── Ontology ───────────────────────────────────────────────────────── func (s *Server) GetOntology(ctx context.Context, req gen.GetOntologyRequestObject) (gen.GetOntologyResponseObject, error) { resp := gen.GetOntology200JSONResponse{ EntityTypes: []gen.EntityType{}, RelationshipTypes: []gen.RelationshipType{}, Lifecycles: []gen.LifecycleDef{}, } q := sqlcgen.New(s.pool) etRows, err := q.ListEntityTypes(ctx) if err != nil { return nil, err } for _, et := range etRows { schemaVersion := int(et.SchemaVersion) var schema *map[string]any if len(et.AttributeSchema) > 0 { var s map[string]any if json.Unmarshal(et.AttributeSchema, &s) == nil && s != nil { schema = &s } } resp.EntityTypes = append(resp.EntityTypes, gen.EntityType{ Name: et.Name, ParentType: et.ParentType, IsAbstract: et.IsAbstract, Domain: et.Domain, Layer: gen.EntityTypeLayer(et.Layer), Description: et.Description, LifecycleId: et.LifecycleID, SchemaVersion: &schemaVersion, AttributeSchema: schema, Status: gen.EntityTypeStatus(et.Status), }) } rtRows, err := q.ListRelationshipTypes(ctx) if err != nil { return nil, err } for _, rt := range rtRows { resp.RelationshipTypes = append(resp.RelationshipTypes, gen.RelationshipType{ Name: rt.Name, Inverse: rt.Inverse, SourceType: rt.SourceType, TargetType: rt.TargetType, Cardinality: gen.RelationshipTypeCardinality(rt.Cardinality), Description: rt.Description, }) } lcRows, err := q.ListLifecycleDefs(ctx) if err != nil { return nil, err } for _, lc := range lcRows { terminal := lc.TerminalStates var transitions map[string]any if err := json.Unmarshal(lc.Transitions, &transitions); err != nil { return nil, fmt.Errorf("lifecycle %s transitions: %w", lc.ID, err) } resp.Lifecycles = append(resp.Lifecycles, gen.LifecycleDef{ Id: lc.ID, States: lc.States, DefaultState: lc.DefaultState, TerminalStates: &terminal, Transitions: transitions, }) } return resp, nil } // ─── Signals ────────────────────────────────────────────────────────── func (s *Server) ListSignals(ctx context.Context, req gen.ListSignalsRequestObject) (gen.ListSignalsResponseObject, error) { limit := clampLimit(req.Params.Limit) rows, err := s.pool.Query(ctx, ` SELECT sig.entity_id, se.slug, sig.kind, sig.severity, sig.state, te.slug, sig.check_id::text, sig.evidence, sig.likely_cause, sig.occurrence_count, sig.flap_count, sig.hold_down_until, sig.mute_until, sig.first_seen_at, sig.last_seen_at FROM signals sig JOIN entities se ON se.id = sig.entity_id LEFT JOIN entities te ON te.id = sig.target_entity_id WHERE ($1::text IS NULL OR sig.state = $1) AND ($2::text IS NULL OR sig.severity = $2) AND ($3::text IS NULL OR te.slug = $3) AND ($4::text IS NULL OR sig.kind = $4) AND ($5::text IS NULL OR se.slug > $5) ORDER BY se.slug LIMIT $6`, req.Params.State, (*string)(req.Params.Severity), req.Params.EntityId, req.Params.Kind, req.Params.Cursor, limit+1) if err != nil { return nil, err } defer rows.Close() items := []gen.Signal{} for rows.Next() { var sig gen.Signal var flap int if err := rows.Scan(&sig.Id, &sig.Slug, &sig.Kind, &sig.Severity, &sig.State, &sig.Target, &sig.CheckId, &sig.Evidence, &sig.LikelyCause, &sig.OccurrenceCount, &flap, &sig.HoldDownUntil, &sig.MuteUntil, &sig.FirstSeenAt, &sig.LastSeenAt); err != nil { return nil, err } sig.FlapCount = &flap items = append(items, sig) } if rows.Err() != nil { return nil, rows.Err() } var next *string if len(items) > limit { items = items[:limit] next = &items[len(items)-1].Slug } return gen.ListSignals200JSONResponse{Items: items, NextCursor: next}, nil } // ─── Observability + system ─────────────────────────────────────────── func (s *Server) GetFleetHealth(ctx context.Context, req gen.GetFleetHealthRequestObject) (gen.GetFleetHealthResponseObject, error) { resp := gen.GetFleetHealth200JSONResponse{} resp.Entities = []struct { Health gen.HealthSummaryEntitiesHealth `json:"health"` LastCheckAt *time.Time `json:"last_check_at"` Slug string `json:"slug"` Trend *gen.HealthSummaryEntitiesTrend `json:"trend"` Type string `json:"type"` }{} // Exclude 'check' entities (internal probes) — only entities actually // being monitored should count toward fleet health. rows, err := s.pool.Query(ctx, ` SELECT e.slug, e.type, st.health, st.last_check_at FROM entity_status st JOIN entities e ON e.id = st.entity_id WHERE e.type <> 'check' ORDER BY e.slug`) if err != nil { return nil, err } defer rows.Close() stale := 0 for rows.Next() { var slug, typ, health string var lastCheck *time.Time if err := rows.Scan(&slug, &typ, &health, &lastCheck); err != nil { return nil, err } switch health { case "healthy": resp.Summary.Healthy++ case "degraded": resp.Summary.Degraded++ case "down": resp.Summary.Down++ case "stale": stale++ default: resp.Summary.Unknown++ } resp.Entities = append(resp.Entities, struct { Health gen.HealthSummaryEntitiesHealth `json:"health"` LastCheckAt *time.Time `json:"last_check_at"` Slug string `json:"slug"` Trend *gen.HealthSummaryEntitiesTrend `json:"trend"` Type string `json:"type"` }{ Health: gen.HealthSummaryEntitiesHealth(health), LastCheckAt: lastCheck, Slug: slug, Type: typ, }) } if stale > 0 { resp.Summary.Stale = &stale } return resp, rows.Err() } func (s *Server) ExportSeeds(ctx context.Context, req gen.ExportSeedsRequestObject) (gen.ExportSeedsResponseObject, error) { exports, err := db.ExportToYAML(ctx, s.pool) if err != nil { return nil, err } return gen.ExportSeeds200JSONResponse{ Ontology: string(exports["ontology.yaml"]), Inventory: string(exports["inventory.yaml"]), Policy: string(exports["policy.yaml"]), }, nil } // ─── Signal mutations ──────────────────────────────────────────────── func (s *Server) AckSignal(ctx context.Context, req gen.AckSignalRequestObject) (gen.AckSignalResponseObject, error) { id, err := s.resolveEntityID(ctx, req.Id) if err != nil { return nil, err } tx, err := s.pool.Begin(ctx) if err != nil { return nil, err } defer tx.Rollback(ctx) var sig gen.Signal err = tx.QueryRow(ctx, ` UPDATE signals SET state = 'acknowledged', updated_at = now() WHERE entity_id = $1 AND state IN ('raised','failed') RETURNING entity_id, (SELECT slug FROM entities WHERE id = $1), kind, severity, 'acknowledged', (SELECT slug FROM entities WHERE id = target_entity_id), check_id::text, evidence, likely_cause, occurrence_count, flap_count, hold_down_until, mute_until, first_seen_at, last_seen_at`, id).Scan(&sig.Id, &sig.Slug, &sig.Kind, &sig.Severity, &sig.State, &sig.Target, &sig.CheckId, &sig.Evidence, &sig.LikelyCause, &sig.OccurrenceCount, &sig.FlapCount, &sig.HoldDownUntil, &sig.MuteUntil, &sig.FirstSeenAt, &sig.LastSeenAt) if err != nil { if err == pgx.ErrNoRows { return nil, fmt.Errorf("%w: signal %s not in a state that can be acknowledged", domain.ErrInvalidTransition, req.Id) } return nil, err } if err := tx.Commit(ctx); err != nil { return nil, err } return gen.AckSignal200JSONResponse{SignalUpdatedJSONResponse: gen.SignalUpdatedJSONResponse(sig)}, nil } func (s *Server) ResolveSignal(ctx context.Context, req gen.ResolveSignalRequestObject) (gen.ResolveSignalResponseObject, error) { id, err := s.resolveEntityID(ctx, req.Id) if err != nil { return nil, err } tx, err := s.pool.Begin(ctx) if err != nil { return nil, err } defer tx.Rollback(ctx) var sig gen.Signal err = tx.QueryRow(ctx, ` UPDATE signals SET state = 'resolved', updated_at = now() WHERE entity_id = $1 AND state IN ('raised','acknowledged','acting','failed') RETURNING entity_id, (SELECT slug FROM entities WHERE id = $1), kind, severity, 'resolved', (SELECT slug FROM entities WHERE id = target_entity_id), check_id::text, evidence, likely_cause, occurrence_count, flap_count, hold_down_until, mute_until, first_seen_at, last_seen_at`, id).Scan(&sig.Id, &sig.Slug, &sig.Kind, &sig.Severity, &sig.State, &sig.Target, &sig.CheckId, &sig.Evidence, &sig.LikelyCause, &sig.OccurrenceCount, &sig.FlapCount, &sig.HoldDownUntil, &sig.MuteUntil, &sig.FirstSeenAt, &sig.LastSeenAt) if err != nil { if err == pgx.ErrNoRows { return nil, fmt.Errorf("%w: signal %s not in a state that can be resolved", domain.ErrInvalidTransition, req.Id) } return nil, err } if err := tx.Commit(ctx); err != nil { return nil, err } return gen.ResolveSignal200JSONResponse{SignalUpdatedJSONResponse: gen.SignalUpdatedJSONResponse(sig)}, nil } func (s *Server) MuteSignal(ctx context.Context, req gen.MuteSignalRequestObject) (gen.MuteSignalResponseObject, error) { id, err := s.resolveEntityID(ctx, req.Id) if err != nil { return nil, err } tx, err := s.pool.Begin(ctx) if err != nil { return nil, err } defer tx.Rollback(ctx) var sig gen.Signal err = tx.QueryRow(ctx, ` UPDATE signals SET state = 'muted', mute_until = $2, updated_at = now() WHERE entity_id = $1 AND state IN ('raised','acknowledged') RETURNING entity_id, (SELECT slug FROM entities WHERE id = $1), kind, severity, 'muted', (SELECT slug FROM entities WHERE id = target_entity_id), check_id::text, evidence, likely_cause, occurrence_count, flap_count, hold_down_until, mute_until, first_seen_at, last_seen_at`, id, req.Body.MuteUntil).Scan(&sig.Id, &sig.Slug, &sig.Kind, &sig.Severity, &sig.State, &sig.Target, &sig.CheckId, &sig.Evidence, &sig.LikelyCause, &sig.OccurrenceCount, &sig.FlapCount, &sig.HoldDownUntil, &sig.MuteUntil, &sig.FirstSeenAt, &sig.LastSeenAt) if err != nil { if err == pgx.ErrNoRows { return nil, fmt.Errorf("%w: signal %s not in a state that can be muted", domain.ErrInvalidTransition, req.Id) } return nil, err } if err := tx.Commit(ctx); err != nil { return nil, err } return gen.MuteSignal200JSONResponse{SignalUpdatedJSONResponse: gen.SignalUpdatedJSONResponse(sig)}, nil } // ─── Observability reads ───────────────────────────────────────────── func (s *Server) QueryEvents(ctx context.Context, req gen.QueryEventsRequestObject) (gen.QueryEventsResponseObject, error) { limit := clampLimit(req.Params.Limit) var eventType, entityID, severity, correlationID *string if req.Params.Type != nil { eventType = req.Params.Type } if req.Params.EntityId != nil { entityID = req.Params.EntityId } if req.Params.Severity != nil { severity = req.Params.Severity } if req.Params.CorrelationId != nil { correlationID = req.Params.CorrelationId } rows, err := s.pool.Query(ctx, ` SELECT id, ts, type, entity_id::text, severity, source, data, correlation_id FROM events WHERE ($1::text IS NULL OR type = $1) AND ($2::text IS NULL OR entity_id::text = $2) AND ($3::text IS NULL OR severity = $3) AND ($4::text IS NULL OR correlation_id = $4) AND ($5::timestamptz IS NULL OR ts >= $5) AND ($6::timestamptz IS NULL OR ts <= $6) ORDER BY ts DESC LIMIT $7`, eventType, entityID, severity, correlationID, req.Params.From, req.Params.To, limit) if err != nil { return nil, err } defer rows.Close() items := []gen.Event{} for rows.Next() { var e gen.Event var dataBytes []byte var entID, corrID *string if err := rows.Scan(&e.Id, &e.Ts, &e.Type, &entID, &e.Severity, &e.Source, &dataBytes, &corrID); err != nil { return nil, err } e.EntityId = entID e.CorrelationId = corrID var data map[string]any if json.Unmarshal(dataBytes, &data) == nil { e.Data = &data } items = append(items, e) } return gen.QueryEvents200JSONResponse{Items: items}, rows.Err() } func (s *Server) QueryAudit(ctx context.Context, req gen.QueryAuditRequestObject) (gen.QueryAuditResponseObject, error) { limit := clampLimit(req.Params.Limit) var actorType, actorID, action, entityID, correlationID *string if req.Params.ActorType != nil { actorType = req.Params.ActorType } if req.Params.ActorId != nil { actorID = req.Params.ActorId } if req.Params.Action != nil { action = req.Params.Action } if req.Params.EntityId != nil { entityID = req.Params.EntityId } if req.Params.CorrelationId != nil { correlationID = req.Params.CorrelationId } rows, err := s.pool.Query(ctx, ` SELECT id, ts, actor_type, actor_id::text, action, entity_id::text, method, path, status_code, detail, source_ip, correlation_id FROM audit_log WHERE ($1::text IS NULL OR actor_type = $1) AND ($2::text IS NULL OR actor_id::text = $2) AND ($3::text IS NULL OR action = $3) AND ($4::text IS NULL OR entity_id::text = $4) AND ($5::text IS NULL OR correlation_id = $5) AND ($6::timestamptz IS NULL OR ts >= $6) AND ($7::timestamptz IS NULL OR ts <= $7) ORDER BY ts DESC LIMIT $8`, actorType, actorID, action, entityID, correlationID, req.Params.From, req.Params.To, limit) if err != nil { return nil, err } defer rows.Close() items := []gen.AuditEntry{} for rows.Next() { var a gen.AuditEntry var detailBytes []byte var actID, entID, method, path, sourceIP, corrID *string var statusCode *int if err := rows.Scan(&a.Id, &a.Ts, &a.ActorType, &actID, &a.Action, &entID, &method, &path, &statusCode, &detailBytes, &sourceIP, &corrID); err != nil { return nil, err } a.ActorId = actID a.EntityId = entID a.Method = method a.Path = path a.StatusCode = statusCode a.SourceIp = sourceIP a.CorrelationId = corrID var detail map[string]any if json.Unmarshal(detailBytes, &detail) == nil { a.Detail = &detail } items = append(items, a) } return gen.QueryAudit200JSONResponse{Items: items}, rows.Err() } // ─── Entity mutations ────────────────────────────────────────────── func (s *Server) CreateEntity(ctx context.Context, req gen.CreateEntityRequestObject) (gen.CreateEntityResponseObject, error) { if req.Body == nil { return nil, fmt.Errorf("%w: request body is required", domain.ErrInvalidInput) } // Check idempotency if a key was provided. The idempotency scope is the // calling actor, so replays are per-caller. actorType, actorLabel := actorInfo(ctx) actor := actorLabel var bodyHash string if req.Params.IdempotencyKey != nil && *req.Params.IdempotencyKey != "" { key := *req.Params.IdempotencyKey q := sqlcgen.New(s.pool) cached, err := q.GetIdempotentResponse(ctx, sqlcgen.GetIdempotentResponseParams{ Actor: actor, Key: key, }) if err == nil { // Verify the request body hasn't changed. bodyJSON, _ := json.Marshal(req.Body) bodyHash = fmt.Sprintf("%x", sha256.Sum256(bodyJSON)) if cached.RequestHash != bodyHash { return nil, fmt.Errorf("%w: idempotency key %s used with different request body", domain.ErrConflict, key) } // Replay the cached response. if cached.ResponseCode != nil && *cached.ResponseCode == 201 { var entity gen.Entity if len(cached.ResponseBody) > 0 { if err := json.Unmarshal(cached.ResponseBody, &entity); err != nil { return nil, fmt.Errorf("unmarshal cached response: %w", err) } } return gen.CreateEntity201JSONResponse{ Body: entity, Headers: gen.CreateEntity201ResponseHeaders{ETag: `"` + strconv.Itoa(entity.Version) + `"`}, }, nil } // Forward cached error response. return gen.CreateEntitydefaultApplicationProblemPlusJSONResponse{ Body: gen.Problem{Status: int(*cached.ResponseCode), Title: "replayed error"}, StatusCode: int(*cached.ResponseCode), }, nil } } id, err := uuid.NewV7() if err != nil { return nil, err } slug := req.Body.Slug if slug == "" { slug = req.Body.Type + ":" + req.Body.Name } tx, err := s.pool.Begin(ctx) if err != nil { return nil, err } defer tx.Rollback(ctx) q := sqlcgen.New(tx) // Validate type exists and is NOT abstract. var isAbstract bool if err := tx.QueryRow(ctx, `SELECT is_abstract FROM entity_types WHERE name = $1`, req.Body.Type).Scan(&isAbstract); err != nil { if err == pgx.ErrNoRows { return nil, fmt.Errorf("%w: entity type %q", domain.ErrNotFound, req.Body.Type) } return nil, err } if isAbstract { return nil, fmt.Errorf("%w: %s", domain.ErrAbstractType, req.Body.Type) } // Get default state from lifecycle. var defaultState *string var lcDefault string if err := tx.QueryRow(ctx, `SELECT ld.default_state FROM lifecycle_defs ld JOIN entity_types et ON et.lifecycle_id = ld.id WHERE et.name = $1`, req.Body.Type).Scan(&lcDefault); err == nil { defaultState = &lcDefault } state := req.Body.State if state == nil && defaultState != nil { state = defaultState } // attributes is NOT NULL; the column default only applies when omitted, // not when an explicit NULL is bound — so default to an empty object. attrsJSON := []byte("{}") if req.Body.Attributes != nil { attrsJSON, _ = json.Marshal(req.Body.Attributes) } // Insert the entity. inserted, err := q.InsertEntity(ctx, sqlcgen.InsertEntityParams{ ID: id, Slug: slug, Type: req.Body.Type, Name: req.Body.Name, State: state, Attributes: attrsJSON, }) if err != nil { // Duplicate slug. if strings.Contains(err.Error(), "unique") || strings.Contains(err.Error(), "duplicate") { return nil, fmt.Errorf("%w: slug %q already exists", domain.ErrAlreadyExists, slug) } return nil, err } // Convert sqlcgen.Entity → gen.Entity. entity := sqlcEntityToGen(inserted) // Cache idempotent response. if req.Params.IdempotencyKey != nil && *req.Params.IdempotencyKey != "" { respBody, _ := json.Marshal(entity) code := int32(201) if bodyHash == "" { bodyJSON, _ := json.Marshal(req.Body) bodyHash = fmt.Sprintf("%x", sha256.Sum256(bodyJSON)) } if putErr := q.PutIdempotentResponse(ctx, sqlcgen.PutIdempotentResponseParams{ Actor: actor, Key: *req.Params.IdempotencyKey, RequestHash: bodyHash, ResponseCode: &code, ResponseBody: respBody, }); putErr != nil { return nil, putErr } } // Audit. entityID := inserted.ID if auditErr := observability.Audit(ctx, q, actorType, actor, "create", &entityID, "POST", "/api/v1/entities", "", map[string]any{"type": req.Body.Type, "slug": slug}); auditErr != nil { return nil, auditErr } if eventErr := observability.Event(ctx, q, "entity.created", &entityID, "info", "oikos-api", "", map[string]any{"slug": slug, "type": req.Body.Type}); eventErr != nil { return nil, eventErr } if err := ensureDefaultChecks(ctx, tx, inserted.ID, slug, req.Body.Type, inserted.Name, attrsJSON); err != nil { return nil, err } if err := tx.Commit(ctx); err != nil { return nil, err } return gen.CreateEntity201JSONResponse{ Body: entity, Headers: gen.CreateEntity201ResponseHeaders{ETag: `"` + strconv.Itoa(entity.Version) + `"`}, }, nil } func (s *Server) PatchEntity(ctx context.Context, req gen.PatchEntityRequestObject) (gen.PatchEntityResponseObject, error) { if req.Body == nil { return nil, fmt.Errorf("%w: request body is required", domain.ErrInvalidInput) } id, err := s.resolveEntityID(ctx, req.Id) if err != nil { return nil, err } // Parse If-Match header (quoted version string). ifMatch := strings.Trim(req.Params.IfMatch, `"`) expectedVersion, err := strconv.Atoi(ifMatch) if err != nil { return nil, fmt.Errorf("%w: invalid If-Match header %q", domain.ErrInvalidInput, req.Params.IfMatch) } tx, err := s.pool.Begin(ctx) if err != nil { return nil, err } defer tx.Rollback(ctx) // Get current entity for version check + lifecycle validation. current, err := sqlcgen.New(tx).GetEntityByID(ctx, id) if err != nil { if err == pgx.ErrNoRows { return nil, fmt.Errorf("%w: %s", domain.ErrNotFound, req.Id) } return nil, err } if int(current.Version) != expectedVersion { return nil, fmt.Errorf("%w: expected version %d, current version %d", domain.ErrConflict, expectedVersion, current.Version) } // Validate lifecycle transition if state is being changed. if req.Body.State != nil && *req.Body.State != "" { fromState := "" if current.State != nil { fromState = *current.State } if err := db.ValidateTransition(ctx, tx, id, current.Type, fromState, *req.Body.State); err != nil { if errors.Is(err, db.ErrTransitionInvalid) { return nil, fmt.Errorf("%w: %v", domain.ErrInvalidTransition, err) } return nil, err } } // Check idempotency (note: the spec doesn't define Idempotency-Key for PATCH, // but we handle it if the generated code ever adds it). // For now, no idempotency check on PATCH. // Marshal attributes if provided. var attrsJSON []byte if req.Body.Attributes != nil { attrsJSON, _ = json.Marshal(req.Body.Attributes) } // Perform the update via sqlcgen. q := sqlcgen.New(tx) updated, err := q.UpdateEntity(ctx, sqlcgen.UpdateEntityParams{ Name: req.Body.Name, State: req.Body.State, Attributes: attrsJSON, SetMaintenance: req.Body.MaintenanceUntil != nil, MaintenanceUntil: req.Body.MaintenanceUntil, ID: id, Version: int32(expectedVersion), }) if err != nil { if err == pgx.ErrNoRows { // Version mismatch or entity not found. return nil, fmt.Errorf("%w: entity was modified concurrently", domain.ErrConflict) } return nil, err } entity := sqlcEntityToGen(updated) // Audit. patchActorType, patchActor := actorInfo(ctx) if auditErr := observability.Audit(ctx, q, patchActorType, patchActor, "patch", &id, "PATCH", "/api/v1/entities/"+req.Id, "", map[string]any{"version": expectedVersion}); auditErr != nil { return nil, auditErr } if eventErr := observability.Event(ctx, q, "entity.updated", &id, "info", "oikos-api", "", map[string]any{"slug": entity.Slug, "type": entity.Type, "version": updated.Version}); eventErr != nil { return nil, eventErr } if err := tx.Commit(ctx); err != nil { return nil, err } return gen.PatchEntity200JSONResponse{ Body: entity, Headers: gen.PatchEntity200ResponseHeaders{ETag: `"` + strconv.Itoa(entity.Version) + `"`}, }, nil } // sqlcEntityToGen converts a sqlcgen.Entity to a gen.Entity. func sqlcEntityToGen(e sqlcgen.Entity) gen.Entity { out := gen.Entity{ Id: e.ID, Slug: e.Slug, Type: e.Type, Name: e.Name, State: e.State, Version: int(e.Version), CreatedAt: e.CreatedAt, UpdatedAt: e.UpdatedAt, } if e.MaintenanceUntil != nil { out.MaintenanceUntil = e.MaintenanceUntil } if len(e.Attributes) > 0 { var attrs map[string]any if json.Unmarshal(e.Attributes, &attrs) == nil && len(attrs) > 0 { out.Attributes = &attrs } } return out } // ─── Client lifecycle ───────────────────────────────────────────────── func (s *Server) EnrollClient(ctx context.Context, req gen.EnrollClientRequestObject) (gen.EnrollClientResponseObject, error) { if req.Body == nil { return nil, fmt.Errorf("%w: request body is required", domain.ErrInvalidInput) } id, err := s.resolveEntityID(ctx, req.Body.Slug) if err != nil { return nil, err } current, err := sqlcgen.New(s.pool).GetEntityByID(ctx, id) if err != nil { return nil, fmt.Errorf("%w: %s", domain.ErrNotFound, req.Body.Slug) } currentState := "" if current.State != nil { currentState = *current.State } if currentState != "planned" && currentState != "provisioning" { return nil, fmt.Errorf("%w: entity %s is in state %q, expected planned or provisioning", domain.ErrInvalidTransition, req.Body.Slug, currentState) } meshIP := "" if req.Body.MeshIp != nil { meshIP = *req.Body.MeshIp } if meshIP == "" { return nil, fmt.Errorf("%w: mesh_ip is required for enrollment", domain.ErrInvalidInput) } agePubKey, agePrivKey, err := generateAgeKeypair() if err != nil { return nil, fmt.Errorf("age key generation: %w", err) } if s.secretsManager != nil { keyPath := "clients/" + req.Body.Slug + "/age-key" _ = s.secretsManager.Set(ctx, keyPath, agePrivKey) } tx, err := s.pool.Begin(ctx) if err != nil { return nil, err } defer tx.Rollback(ctx) var attrs map[string]any if len(current.Attributes) > 0 { json.Unmarshal(current.Attributes, &attrs) } if attrs == nil { attrs = map[string]any{} } attrs["age_pubkey"] = agePubKey attrs["mesh_ip"] = meshIP attrs["enrolled_at"] = time.Now().UTC().Format(time.RFC3339) if req.Body.Hostname != nil { attrs["hostname"] = *req.Body.Hostname } attrsJSON, _ := json.Marshal(attrs) q := sqlcgen.New(tx) provisioning := "provisioning" now := time.Now().UTC() _, err = q.UpdateEntity(ctx, sqlcgen.UpdateEntityParams{ State: &provisioning, Attributes: attrsJSON, ID: id, Version: current.Version, }) if err != nil { return nil, err } _, _ = tx.Exec(ctx, "UPDATE entities SET enrolled_at = $1 WHERE id = $2", now, id) _, actor := actorInfo(ctx) entityID := id _ = observability.Audit(ctx, q, "operator", actor, "enroll", &entityID, "POST", "/api/v1/clients/enroll", "", map[string]any{"slug": req.Body.Slug, "mesh_ip": meshIP}) _ = observability.Event(ctx, q, "client.enrolled", &entityID, "info", "oikos-api", "", map[string]any{"slug": req.Body.Slug, "type": current.Type}) if err := ensureDefaultChecks(ctx, tx, id, req.Body.Slug, current.Type, current.Name, attrsJSON); err != nil { return nil, err } if err := tx.Commit(ctx); err != nil { return nil, err } infisicalClientID := "inf_client_" + uuid.NewString() infisicalClientSecret := "inf_secret_" + uuid.NewString() resp := gen.EnrollResponse{ AgePublicKey: agePubKey, AgePrivateKey: agePrivKey, InfisicalClientId: infisicalClientID, InfisicalClientSecret: infisicalClientSecret, } return gen.EnrollClient200JSONResponse(resp), nil } func (s *Server) GetClientContext(ctx context.Context, req gen.GetClientContextRequestObject) (gen.GetClientContextResponseObject, error) { slug := string(req.Slug) _, err := s.resolveEntityID(ctx, slug) if err != nil { return nil, err } var version int64 _ = s.pool.QueryRow(ctx, "SELECT version FROM context_version WHERE singleton = true").Scan(&version) var filesChanged, toolsChanged []string var sopsChanged bool if req.Params.Since != nil { rows, qErr := s.pool.Query(ctx, "SELECT path FROM context_files WHERE last_changed > $1", *req.Params.Since) if qErr == nil { defer rows.Close() for rows.Next() { var p string if scanErr := rows.Scan(&p); scanErr == nil { // Matches tools/setup-*.sh (the auto-setup convention — // see tools/post-pull.sh). Was tools/*.setup.sh until // 2026-07-12, which never matched any real filename. if strings.HasPrefix(p, "tools/setup-") && strings.HasSuffix(p, ".sh") { toolsChanged = append(toolsChanged, p) } else if p == ".sops.yaml" { sopsChanged = true } else { filesChanged = append(filesChanged, p) } } } } } if filesChanged == nil { filesChanged = []string{} } if toolsChanged == nil { toolsChanged = []string{} } now := time.Now().UTC() return gen.GetClientContext200JSONResponse{ AgentFilesChanged: &filesChanged, SopsConfigChanged: &sopsChanged, ToolsChanged: &toolsChanged, Version: int(version), Since: &now, }, nil } func (s *Server) GetClientSecrets(ctx context.Context, req gen.GetClientSecretsRequestObject) (gen.GetClientSecretsResponseObject, error) { slug := string(req.Slug) _, err := s.resolveEntityID(ctx, slug) if err != nil { return nil, err } var keys []string if s.secretsManager != nil { list, listErr := s.secretsManager.List(ctx) if listErr == nil { prefix := "clients/" + slug + "/" for _, k := range list { if strings.HasPrefix(k, prefix) || strings.HasPrefix(k, "shared/") { keys = append(keys, k) } } } } if keys == nil { keys = []string{} } return gen.GetClientSecrets200JSONResponse{Keys: keys}, nil } func (s *Server) ProvisionEntity(ctx context.Context, req gen.ProvisionEntityRequestObject) (gen.ProvisionEntityResponseObject, error) { if req.Body == nil { return nil, fmt.Errorf("%w: request body is required", domain.ErrInvalidInput) } hostSlug := req.Body.Host hostID, err := s.resolveEntityID(ctx, hostSlug) if err != nil { return nil, fmt.Errorf("%w: host %q not found", domain.ErrNotFound, hostSlug) } var existingID uuid.UUID err = s.pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", req.Body.Slug).Scan(&existingID) if err == nil { return nil, fmt.Errorf("%w: entity slug %q already exists", domain.ErrConflict, req.Body.Slug) } tx, err := s.pool.Begin(ctx) if err != nil { return nil, err } defer tx.Rollback(ctx) entityID := uuid.Must(uuid.NewV7()) var attrsJSON []byte if req.Body.Attributes != nil { attrsJSON, _ = json.Marshal(req.Body.Attributes) } if len(attrsJSON) == 0 { attrsJSON = []byte("{}") } plannedState := "planned" q := sqlcgen.New(tx) inserted, err := q.InsertEntity(ctx, sqlcgen.InsertEntityParams{ ID: entityID, Slug: req.Body.Slug, Type: req.Body.Type, Name: req.Body.Name, State: &plannedState, Attributes: attrsJSON, }) if err != nil { return nil, err } execID := uuid.Must(uuid.NewV7()) corrID := "provision_" + entityID.String()[:8] if err := q.InsertExecution(ctx, sqlcgen.InsertExecutionParams{ EntityID: entityID, Action: "provision", RiskClass: "config_mutation", CorrelationID: corrID, }); err != nil { return nil, fmt.Errorf("create execution: %w", err) } type stepDef struct { order int name string } steps := []stepDef{ {1, "validate-constraints"}, {2, "create-container"}, {3, "configure-network"}, {4, "install-services"}, {5, "configure-mounts"}, {6, "health-check"}, } for _, st := range steps { _, err = tx.Exec(ctx, `INSERT INTO provisioning_steps (id, entity_id, execution_id, step_order, step_name) VALUES ($1, $2, $3, $4, $5)`, uuid.Must(uuid.NewV7()), entityID, entityID /* executions PK is entity_id */, st.order, st.name) if err != nil { return nil, fmt.Errorf("insert provisioning step: %w", err) } } _, err = tx.Exec(ctx, `INSERT INTO relationships (source_id, target_id, type) VALUES ($1, $2, 'hosts')`, hostID, entityID) if err != nil { return nil, fmt.Errorf("insert relationship: %w", err) } _, actor := actorInfo(ctx) _ = observability.Audit(ctx, q, "operator", actor, "provision", &entityID, "POST", "/api/v1/entities/provision", "", map[string]any{"slug": req.Body.Slug, "host": hostSlug}) _ = observability.Event(ctx, q, "entity.provisioned", &entityID, "info", "oikos-api", "", map[string]any{"slug": req.Body.Slug, "type": req.Body.Type, "host": hostSlug}) if err := tx.Commit(ctx); err != nil { return nil, err } entity := sqlcEntityToGen(inserted) return gen.ProvisionEntity201JSONResponse{ Body: gen.ProvisionResponse{ Entity: entity, ExecutionId: openapi_types.UUID(execID), }, Headers: gen.ProvisionEntity201ResponseHeaders{ETag: `"` + strconv.Itoa(int(inserted.Version)) + `"`}, }, nil } func (s *Server) GetProvisionStatus(ctx context.Context, req gen.GetProvisionStatusRequestObject) (gen.GetProvisionStatusResponseObject, error) { slug := string(req.Slug) id, err := s.resolveEntityID(ctx, slug) if err != nil { return nil, err } var state string if err := s.pool.QueryRow(ctx, "SELECT state FROM entities WHERE id = $1", id).Scan(&state); err != nil { return nil, fmt.Errorf("%w: %s", domain.ErrNotFound, slug) } rows, err := s.pool.Query(ctx, `SELECT step_name, status, error_message, started_at, finished_at FROM provisioning_steps WHERE entity_id = $1 ORDER BY step_order`, id) if err != nil { return nil, err } defer rows.Close() var provSteps []struct { ErrorMessage *string `json:"error_message"` FinishedAt *time.Time `json:"finished_at"` StartedAt *time.Time `json:"started_at"` Status gen.ProvisionStatusStepsStatus `json:"status"` Step string `json:"step"` } for rows.Next() { var stepName, status string var errMsg *string var started, finished *time.Time if scanErr := rows.Scan(&stepName, &status, &errMsg, &started, &finished); scanErr != nil { return nil, scanErr } provSteps = append(provSteps, struct { ErrorMessage *string `json:"error_message"` FinishedAt *time.Time `json:"finished_at"` StartedAt *time.Time `json:"started_at"` Status gen.ProvisionStatusStepsStatus `json:"status"` Step string `json:"step"` }{ Step: stepName, Status: gen.ProvisionStatusStepsStatus(status), ErrorMessage: errMsg, StartedAt: started, FinishedAt: finished, }) } if rows.Err() != nil { return nil, rows.Err() } return gen.GetProvisionStatus200JSONResponse{ Slug: slug, State: state, Steps: provSteps, }, nil } func generateAgeKeypair() (pubKey, privKey string, err error) { seed := make([]byte, 32) if _, err := rand.Read(seed); err != nil { return "", "", err } n := new(big.Int).SetBytes(seed) pub := fmt.Sprintf("age1%064x", n) priv := fmt.Sprintf("AGE-SECRET-KEY-1%064x", n) return pub, priv, nil } // ─── Helpers ───────────────────────────────────────────────────────────