From c9975d60a574d10f3e44cbd8418d88a73b9bd508 Mon Sep 17 00:00:00 2001 From: dtoro Date: Tue, 7 Jul 2026 08:49:59 +0200 Subject: [PATCH] phase 2 (part 2): sqlc queries, audit/event helpers, event NOTIFY trigger - sqlc.yaml + internal/db/queries/*.sql: typed queries for entities, relationships, ontology, operations (signals, events, audit, idempotency, entity_status) - internal/db/sqlcgen/: generated Go from sqlc (pgx/v5) - internal/observability/record.go: Audit() and Event() helpers that write in the caller's transaction (SG10). actorLabel is interim text identity in detail JSON until OIDC resolution lands; actor_id column exists but is not yet populated - migrations/008: post-commit pg_notify trigger on events table for SSE fan-out (SG8/SG10) --- internal/db/queries/entities.sql | 52 ++++ internal/db/queries/ontology.sql | 13 + internal/db/queries/operations.sql | 56 ++++ internal/db/queries/relationships.sql | 33 +++ internal/db/sqlcgen/db.go | 32 ++ internal/db/sqlcgen/entities.sql.go | 254 ++++++++++++++++ internal/db/sqlcgen/models.go | 329 +++++++++++++++++++++ internal/db/sqlcgen/ontology.sql.go | 130 +++++++++ internal/db/sqlcgen/operations.sql.go | 357 +++++++++++++++++++++++ internal/db/sqlcgen/relationships.sql.go | 165 +++++++++++ internal/observability/record.go | 65 +++++ migrations/008_event_notify.up.sql | 24 ++ sqlc.yaml | 39 +++ 13 files changed, 1549 insertions(+) create mode 100644 internal/db/queries/entities.sql create mode 100644 internal/db/queries/ontology.sql create mode 100644 internal/db/queries/operations.sql create mode 100644 internal/db/queries/relationships.sql create mode 100644 internal/db/sqlcgen/db.go create mode 100644 internal/db/sqlcgen/entities.sql.go create mode 100644 internal/db/sqlcgen/models.go create mode 100644 internal/db/sqlcgen/ontology.sql.go create mode 100644 internal/db/sqlcgen/operations.sql.go create mode 100644 internal/db/sqlcgen/relationships.sql.go create mode 100644 internal/observability/record.go create mode 100644 migrations/008_event_notify.up.sql create mode 100644 sqlc.yaml diff --git a/internal/db/queries/entities.sql b/internal/db/queries/entities.sql new file mode 100644 index 0000000..9481653 --- /dev/null +++ b/internal/db/queries/entities.sql @@ -0,0 +1,52 @@ +-- Entity read + mutation queries (API paths). Aliased `e` throughout to +-- avoid ambiguity with joined tables. + +-- name: GetEntityByID :one +SELECT e.* FROM entities e WHERE e.id = $1; + +-- name: GetEntityBySlug :one +SELECT e.* FROM entities e WHERE e.slug = $1; + +-- name: ListEntities :many +WITH RECURSIVE tt AS ( + SELECT name FROM entity_types WHERE sqlc.narg('type')::text IS NULL OR name = sqlc.narg('type') + UNION + SELECT et.name FROM entity_types et JOIN tt ON et.parent_type = tt.name + WHERE sqlc.narg('type')::text IS NOT NULL +) +SELECT e.* FROM entities e +JOIN entity_types et ON et.name = e.type +WHERE e.type IN (SELECT name FROM tt) + AND (sqlc.narg('state')::text IS NULL OR e.state = sqlc.narg('state')) + AND (sqlc.narg('domain')::text IS NULL OR et.domain = sqlc.narg('domain')) + AND (sqlc.narg('layer')::text IS NULL OR et.layer = sqlc.narg('layer')) + AND (sqlc.narg('q')::text IS NULL + OR e.slug ILIKE '%'||sqlc.narg('q')||'%' + OR e.name ILIKE '%'||sqlc.narg('q')||'%') + AND (sqlc.narg('cursor')::text IS NULL OR e.slug > sqlc.narg('cursor')) +ORDER BY e.slug +LIMIT sqlc.arg('lim'); + +-- name: ListEntitiesCapped :many +SELECT e.* FROM entities e ORDER BY e.slug LIMIT $1; + +-- name: InsertEntity :one +INSERT INTO entities (id, slug, type, name, state, attributes) +VALUES ($1, $2, $3, $4, $5, $6) +RETURNING *; + +-- name: UpdateEntity :one +UPDATE entities SET + name = COALESCE(sqlc.narg('name'), name), + state = COALESCE(sqlc.narg('state'), state), + attributes = COALESCE(sqlc.narg('attributes'), attributes), + maintenance_until = CASE WHEN sqlc.arg('set_maintenance')::bool + THEN sqlc.narg('maintenance_until') ELSE maintenance_until END, + version = version + 1, + updated_at = now() +WHERE id = sqlc.arg('id') AND version = sqlc.arg('version') +RETURNING *; + +-- blast_radius(): the recursive-CTE traversal function's TABLE return type +-- is opaque to sqlc's analyzer — that one query stays hand-written pgx in +-- internal/httpapi (see impl.go). diff --git a/internal/db/queries/ontology.sql b/internal/db/queries/ontology.sql new file mode 100644 index 0000000..fe05e2a --- /dev/null +++ b/internal/db/queries/ontology.sql @@ -0,0 +1,13 @@ +-- name: ListEntityTypes :many +SELECT * FROM entity_types ORDER BY name; + +-- name: ListRelationshipTypes :many +SELECT * FROM relationship_types ORDER BY name; + +-- name: ListLifecycleDefs :many +SELECT * FROM lifecycle_defs ORDER BY id; + +-- name: GetLifecycleForType :one +SELECT ld.* FROM lifecycle_defs ld +JOIN entity_types et ON et.lifecycle_id = ld.id +WHERE et.name = $1; diff --git a/internal/db/queries/operations.sql b/internal/db/queries/operations.sql new file mode 100644 index 0000000..5bc41d7 --- /dev/null +++ b/internal/db/queries/operations.sql @@ -0,0 +1,56 @@ +-- name: ListSignals :many +SELECT sig.entity_id, se.slug, sig.kind, sig.severity, sig.state, + te.slug AS target_slug, sig.check_id, 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 (sqlc.narg('state')::text IS NULL OR sig.state = sqlc.narg('state')) + AND (sqlc.narg('severity')::text IS NULL OR sig.severity = sqlc.narg('severity')) + AND (sqlc.narg('target')::text IS NULL OR te.slug = sqlc.narg('target')) + AND (sqlc.narg('kind')::text IS NULL OR sig.kind = sqlc.narg('kind')) + AND (sqlc.narg('cursor')::text IS NULL OR se.slug > sqlc.narg('cursor')) +ORDER BY se.slug +LIMIT sqlc.arg('lim'); + +-- name: ListEntityStatus :many +SELECT e.slug, e.type, st.health, st.last_check_at +FROM entity_status st JOIN entities e ON e.id = st.entity_id +ORDER BY e.slug; + +-- name: GetIdempotentResponse :one +SELECT response_code, response_body, request_hash FROM idempotency_keys +WHERE actor = $1 AND key = $2; + +-- name: PutIdempotentResponse :exec +INSERT INTO idempotency_keys (actor, key, request_hash, response_code, response_body) +VALUES ($1, $2, $3, $4, $5) +ON CONFLICT (actor, key) DO NOTHING; + +-- name: InsertAuditEntry :exec +INSERT INTO audit_log (actor_type, actor_id, action, entity_id, method, path, + status_code, detail, source_ip, correlation_id) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10); + +-- name: InsertEvent :one +INSERT INTO events (type, entity_id, severity, source, data, correlation_id) +VALUES ($1, $2, $3, $4, $5, $6) +RETURNING id, ts; + +-- name: ListEvents :many +SELECT id, ts, type, entity_id, severity, source, data, correlation_id +FROM events +WHERE (sqlc.narg('type')::text IS NULL OR type = sqlc.narg('type')) + AND (sqlc.narg('entity_id')::uuid IS NULL OR entity_id = sqlc.narg('entity_id')) + AND (sqlc.narg('severity')::text IS NULL OR severity = sqlc.narg('severity')) + AND (sqlc.narg('correlation_id')::text IS NULL OR correlation_id = sqlc.narg('correlation_id')) + AND (sqlc.narg('from_ts')::timestamptz IS NULL OR ts >= sqlc.narg('from_ts')) + AND (sqlc.narg('to_ts')::timestamptz IS NULL OR ts <= sqlc.narg('to_ts')) + AND (sqlc.narg('before_id')::bigint IS NULL OR id < sqlc.narg('before_id')) +ORDER BY id DESC +LIMIT sqlc.arg('lim'); + +-- name: ListEventsAfter :many +SELECT id, ts, type, entity_id, severity, source, data, correlation_id +FROM events WHERE id > $1 ORDER BY id ASC LIMIT $2; diff --git a/internal/db/queries/relationships.sql b/internal/db/queries/relationships.sql new file mode 100644 index 0000000..6da6a3a --- /dev/null +++ b/internal/db/queries/relationships.sql @@ -0,0 +1,33 @@ +-- name: ListEntityRelations :many +SELECT se.slug AS source_slug, te.slug AS target_slug, r.type, r.attributes, + r.valid_from, r.valid_to +FROM relationships r +JOIN entities se ON se.id = r.source_id +JOIN entities te ON te.id = r.target_id +WHERE r.valid_to IS NULL + AND ((sqlc.arg('direction')::text IN ('out','both') AND r.source_id = sqlc.arg('id')) + OR (sqlc.arg('direction')::text IN ('in','both') AND r.target_id = sqlc.arg('id'))) + AND (sqlc.narg('rel_type')::text IS NULL OR r.type = sqlc.narg('rel_type')) +ORDER BY r.type, se.slug, te.slug; + +-- name: ListGraphEdges :many +SELECT se.slug AS source_slug, te.slug AS target_slug, r.type, r.attributes, + r.valid_from, r.valid_to +FROM relationships r +JOIN entities se ON se.id = r.source_id +JOIN entities te ON te.id = r.target_id +WHERE r.valid_to IS NULL + AND r.source_id = ANY(sqlc.arg('ids')::uuid[]) + AND r.target_id = ANY(sqlc.arg('ids')::uuid[]) + AND (sqlc.narg('rel_types')::text[] IS NULL OR r.type = ANY(sqlc.narg('rel_types')::text[])) +ORDER BY r.type, se.slug, te.slug; + +-- name: UpsertCurrentRelationship :exec +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) WHERE valid_to IS NULL +DO UPDATE SET attributes = EXCLUDED.attributes; + +-- name: EndCurrentRelationship :execrows +UPDATE relationships SET valid_to = now() +WHERE source_id = $1 AND target_id = $2 AND type = $3 AND valid_to IS NULL; diff --git a/internal/db/sqlcgen/db.go b/internal/db/sqlcgen/db.go new file mode 100644 index 0000000..860e66d --- /dev/null +++ b/internal/db/sqlcgen/db.go @@ -0,0 +1,32 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.29.0 + +package sqlcgen + +import ( + "context" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" +) + +type DBTX interface { + Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error) + Query(context.Context, string, ...interface{}) (pgx.Rows, error) + QueryRow(context.Context, string, ...interface{}) pgx.Row +} + +func New(db DBTX) *Queries { + return &Queries{db: db} +} + +type Queries struct { + db DBTX +} + +func (q *Queries) WithTx(tx pgx.Tx) *Queries { + return &Queries{ + db: tx, + } +} diff --git a/internal/db/sqlcgen/entities.sql.go b/internal/db/sqlcgen/entities.sql.go new file mode 100644 index 0000000..23f889a --- /dev/null +++ b/internal/db/sqlcgen/entities.sql.go @@ -0,0 +1,254 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.29.0 +// source: entities.sql + +package sqlcgen + +import ( + "context" + "time" + + "github.com/google/uuid" +) + +const getEntityByID = `-- name: GetEntityByID :one + +SELECT e.id, e.slug, e.type, e.name, e.state, e.attributes, e.maintenance_until, e.version, e.created_at, e.updated_at FROM entities e WHERE e.id = $1 +` + +// Entity read + mutation queries (API paths). Aliased `e` throughout to +// avoid ambiguity with joined tables. +func (q *Queries) GetEntityByID(ctx context.Context, id uuid.UUID) (Entity, error) { + row := q.db.QueryRow(ctx, getEntityByID, id) + var i Entity + err := row.Scan( + &i.ID, + &i.Slug, + &i.Type, + &i.Name, + &i.State, + &i.Attributes, + &i.MaintenanceUntil, + &i.Version, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const getEntityBySlug = `-- name: GetEntityBySlug :one +SELECT e.id, e.slug, e.type, e.name, e.state, e.attributes, e.maintenance_until, e.version, e.created_at, e.updated_at FROM entities e WHERE e.slug = $1 +` + +func (q *Queries) GetEntityBySlug(ctx context.Context, slug string) (Entity, error) { + row := q.db.QueryRow(ctx, getEntityBySlug, slug) + var i Entity + err := row.Scan( + &i.ID, + &i.Slug, + &i.Type, + &i.Name, + &i.State, + &i.Attributes, + &i.MaintenanceUntil, + &i.Version, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const insertEntity = `-- name: InsertEntity :one +INSERT INTO entities (id, slug, type, name, state, attributes) +VALUES ($1, $2, $3, $4, $5, $6) +RETURNING id, slug, type, name, state, attributes, maintenance_until, version, created_at, updated_at +` + +type InsertEntityParams struct { + ID uuid.UUID + Slug string + Type string + Name string + State *string + Attributes []byte +} + +func (q *Queries) InsertEntity(ctx context.Context, arg InsertEntityParams) (Entity, error) { + row := q.db.QueryRow(ctx, insertEntity, + arg.ID, + arg.Slug, + arg.Type, + arg.Name, + arg.State, + arg.Attributes, + ) + var i Entity + err := row.Scan( + &i.ID, + &i.Slug, + &i.Type, + &i.Name, + &i.State, + &i.Attributes, + &i.MaintenanceUntil, + &i.Version, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const listEntities = `-- name: ListEntities :many +WITH RECURSIVE tt AS ( + SELECT name FROM entity_types WHERE $7::text IS NULL OR name = $7 + UNION + SELECT et.name FROM entity_types et JOIN tt ON et.parent_type = tt.name + WHERE $7::text IS NOT NULL +) +SELECT e.id, e.slug, e.type, e.name, e.state, e.attributes, e.maintenance_until, e.version, e.created_at, e.updated_at FROM entities e +JOIN entity_types et ON et.name = e.type +WHERE e.type IN (SELECT name FROM tt) + AND ($1::text IS NULL OR e.state = $1) + AND ($2::text IS NULL OR et.domain = $2) + AND ($3::text IS NULL OR et.layer = $3) + AND ($4::text IS NULL + OR e.slug ILIKE '%'||$4||'%' + OR e.name ILIKE '%'||$4||'%') + AND ($5::text IS NULL OR e.slug > $5) +ORDER BY e.slug +LIMIT $6 +` + +type ListEntitiesParams struct { + State *string + Domain *string + Layer *string + Q *string + Cursor *string + Lim int32 + Type *string +} + +func (q *Queries) ListEntities(ctx context.Context, arg ListEntitiesParams) ([]Entity, error) { + rows, err := q.db.Query(ctx, listEntities, + arg.State, + arg.Domain, + arg.Layer, + arg.Q, + arg.Cursor, + arg.Lim, + arg.Type, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []Entity + for rows.Next() { + var i Entity + if err := rows.Scan( + &i.ID, + &i.Slug, + &i.Type, + &i.Name, + &i.State, + &i.Attributes, + &i.MaintenanceUntil, + &i.Version, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listEntitiesCapped = `-- name: ListEntitiesCapped :many +SELECT e.id, e.slug, e.type, e.name, e.state, e.attributes, e.maintenance_until, e.version, e.created_at, e.updated_at FROM entities e ORDER BY e.slug LIMIT $1 +` + +func (q *Queries) ListEntitiesCapped(ctx context.Context, limit int32) ([]Entity, error) { + rows, err := q.db.Query(ctx, listEntitiesCapped, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []Entity + for rows.Next() { + var i Entity + if err := rows.Scan( + &i.ID, + &i.Slug, + &i.Type, + &i.Name, + &i.State, + &i.Attributes, + &i.MaintenanceUntil, + &i.Version, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const updateEntity = `-- name: UpdateEntity :one +UPDATE entities SET + name = COALESCE($1, name), + state = COALESCE($2, state), + attributes = COALESCE($3, attributes), + maintenance_until = CASE WHEN $4::bool + THEN $5 ELSE maintenance_until END, + version = version + 1, + updated_at = now() +WHERE id = $6 AND version = $7 +RETURNING id, slug, type, name, state, attributes, maintenance_until, version, created_at, updated_at +` + +type UpdateEntityParams struct { + Name *string + State *string + Attributes []byte + SetMaintenance bool + MaintenanceUntil *time.Time + ID uuid.UUID + Version int32 +} + +func (q *Queries) UpdateEntity(ctx context.Context, arg UpdateEntityParams) (Entity, error) { + row := q.db.QueryRow(ctx, updateEntity, + arg.Name, + arg.State, + arg.Attributes, + arg.SetMaintenance, + arg.MaintenanceUntil, + arg.ID, + arg.Version, + ) + var i Entity + err := row.Scan( + &i.ID, + &i.Slug, + &i.Type, + &i.Name, + &i.State, + &i.Attributes, + &i.MaintenanceUntil, + &i.Version, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} diff --git a/internal/db/sqlcgen/models.go b/internal/db/sqlcgen/models.go new file mode 100644 index 0000000..4aed687 --- /dev/null +++ b/internal/db/sqlcgen/models.go @@ -0,0 +1,329 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.29.0 + +package sqlcgen + +import ( + "time" + + "github.com/google/uuid" +) + +type AgentActivity struct { + ID int64 + Ts time.Time + AgentID uuid.UUID + SessionID *string + ActivityType string + ToolName *string + EntityID *uuid.UUID + InputSummary *string + OutputSummary *string + DurationMs *int32 + TokenCount *int32 + Success *bool + CorrelationID *string +} + +type Approval struct { + EntityID uuid.UUID + SubjectEntityID *uuid.UUID + Action string + RiskClass string + Kind string + Payload []byte + Status string + TokenHash *string + ExpiresAt time.Time + DecidedAt *time.Time + DecidedBy *uuid.UUID + CreatedAt time.Time +} + +type ApprovalRule struct { + ID uuid.UUID + EntityType *string + Action string + RiskClass string + AutonomyLevel string + ScopeEntity *uuid.UUID + Version int32 + UpdatedAt time.Time +} + +type AuditLog struct { + ID int64 + Ts time.Time + ActorType string + ActorID *uuid.UUID + Action string + EntityID *uuid.UUID + Method *string + Path *string + StatusCode *int32 + Detail []byte + SourceIp *string + CorrelationID *string +} + +type AutonomySetting struct { + Key string + Value string + Version int32 + UpdatedAt time.Time +} + +type CheckDef struct { + EntityID uuid.UUID + TargetID *uuid.UUID + TargetType *string + Kind string + Config []byte + IntervalS int32 + TimeoutS int32 + Zone *string + Enabled bool + UpdatedAt time.Time +} + +type Classification struct { + EntityID uuid.UUID + SignalEntityID *uuid.UUID + TargetEntityID *uuid.UUID + Action string + RecommendedAction []byte + RiskClass string + Route string + BlastRadius []uuid.UUID + PatternConfidence *float32 + SkillID *uuid.UUID + AutonomyCheck *string + Reasoning []byte + CorrelationID string + CreatedAt time.Time +} + +type Entity struct { + ID uuid.UUID + Slug string + Type string + Name string + State *string + Attributes []byte + MaintenanceUntil *time.Time + Version int32 + CreatedAt time.Time + UpdatedAt time.Time +} + +type EntityStatus struct { + EntityID uuid.UUID + Health string + LastCheckAt *time.Time + Details []byte + UpdatedAt time.Time +} + +type EntityType struct { + Name string + ParentType *string + IsAbstract bool + Domain string + Layer string + Description *string + LifecycleID *string + AttributeSchema []byte + SchemaVersion int32 + Status string + CreatedAt time.Time + UpdatedAt time.Time +} + +type Event struct { + ID int64 + Ts time.Time + Type string + EntityID *uuid.UUID + Severity string + Source string + Data []byte + CorrelationID *string +} + +type Execution struct { + EntityID uuid.UUID + ClassificationID *uuid.UUID + SignalEntityID *uuid.UUID + TargetEntityID *uuid.UUID + Action string + RiskClass string + ApprovalID *uuid.UUID + AgentID *uuid.UUID + SkillID *uuid.UUID + SkillVersion *int32 + Status string + Result []byte + DurationMs *int32 + Verified bool + CorrelationID string + StartedAt *time.Time + CompletedAt *time.Time + CreatedAt time.Time +} + +type Feedback struct { + EntityID uuid.UUID + ExecutionID uuid.UUID + Outcome string + Observation *string + Lesson *string + UnexpectedSideEffects []string + Tags []string + CreatedAt time.Time +} + +type IdempotencyKey struct { + Key string + Actor string + RequestHash string + ResponseCode *int32 + ResponseBody []byte + CreatedAt time.Time +} + +type Ledger struct { + Ts time.Time + ExecutionID uuid.UUID + TargetEntityID *uuid.UUID + Action string + RiskClass string + Status string + Verified bool + Route *string + Reasoning []byte + ApprovalStatus *string + DecidedBy *uuid.UUID + AgentID *uuid.UUID + CorrelationID string +} + +type LifecycleDef struct { + ID string + States []string + DefaultState string + TerminalStates []string + Transitions []byte + CreatedAt time.Time +} + +type MetricRollups1d struct { + Bucket interface{} + EntityID uuid.UUID + Metric string + AvgValue float64 + MinValue interface{} + MaxValue interface{} + SampleCount int64 +} + +type MetricRollups1h struct { + Bucket interface{} + EntityID uuid.UUID + Metric string + AvgValue float64 + MinValue interface{} + MaxValue interface{} + SampleCount int64 +} + +type MetricSample struct { + Ts time.Time + EntityID uuid.UUID + Metric string + Value float64 + Tags []byte +} + +type Pattern struct { + EntityID uuid.UUID + AppliesType string + Action string + Pattern string + Confidence float32 + EvidenceCount int32 + SuccessCount int32 + FailureCount int32 + Status string + Quarantined bool + Version int32 + LastValidatedAt *time.Time + CreatedAt time.Time +} + +type Relationship struct { + SourceID uuid.UUID + TargetID uuid.UUID + Type string + Attributes []byte + ValidFrom time.Time + ValidTo *time.Time +} + +type RelationshipType struct { + Name string + Inverse *string + SourceType string + TargetType string + Cardinality string + Description *string + CreatedAt time.Time +} + +type RiskClass struct { + Name string + Description *string + ApprovalRequired string + AutonomyAllowed bool +} + +type SeedVersion struct { + File string + ContentHash string + AppliedAt time.Time +} + +type Signal struct { + EntityID uuid.UUID + Kind string + Severity string + TargetEntityID *uuid.UUID + CheckID *uuid.UUID + Evidence *string + LikelyCause *string + State string + OccurrenceCount int32 + FirstSeenAt time.Time + LastSeenAt time.Time + FlapCount int32 + HoldDownUntil *time.Time + MuteUntil *time.Time + CreatedAt time.Time + UpdatedAt time.Time +} + +type Skill struct { + EntityID uuid.UUID + Version int32 + Name string + Procedure []byte + AppliesType *string + Action string + PatternIds []uuid.UUID + Status string + SuccessRate *float32 + ChangedBy *uuid.UUID + ChangeReason *string + LastUsedAt *time.Time + CreatedAt time.Time +} diff --git a/internal/db/sqlcgen/ontology.sql.go b/internal/db/sqlcgen/ontology.sql.go new file mode 100644 index 0000000..5a3acf8 --- /dev/null +++ b/internal/db/sqlcgen/ontology.sql.go @@ -0,0 +1,130 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.29.0 +// source: ontology.sql + +package sqlcgen + +import ( + "context" +) + +const getLifecycleForType = `-- name: GetLifecycleForType :one +SELECT ld.id, ld.states, ld.default_state, ld.terminal_states, ld.transitions, ld.created_at FROM lifecycle_defs ld +JOIN entity_types et ON et.lifecycle_id = ld.id +WHERE et.name = $1 +` + +func (q *Queries) GetLifecycleForType(ctx context.Context, name string) (LifecycleDef, error) { + row := q.db.QueryRow(ctx, getLifecycleForType, name) + var i LifecycleDef + err := row.Scan( + &i.ID, + &i.States, + &i.DefaultState, + &i.TerminalStates, + &i.Transitions, + &i.CreatedAt, + ) + return i, err +} + +const listEntityTypes = `-- name: ListEntityTypes :many +SELECT name, parent_type, is_abstract, domain, layer, description, lifecycle_id, attribute_schema, schema_version, status, created_at, updated_at FROM entity_types ORDER BY name +` + +func (q *Queries) ListEntityTypes(ctx context.Context) ([]EntityType, error) { + rows, err := q.db.Query(ctx, listEntityTypes) + if err != nil { + return nil, err + } + defer rows.Close() + var items []EntityType + for rows.Next() { + var i EntityType + if err := rows.Scan( + &i.Name, + &i.ParentType, + &i.IsAbstract, + &i.Domain, + &i.Layer, + &i.Description, + &i.LifecycleID, + &i.AttributeSchema, + &i.SchemaVersion, + &i.Status, + &i.CreatedAt, + &i.UpdatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listLifecycleDefs = `-- name: ListLifecycleDefs :many +SELECT id, states, default_state, terminal_states, transitions, created_at FROM lifecycle_defs ORDER BY id +` + +func (q *Queries) ListLifecycleDefs(ctx context.Context) ([]LifecycleDef, error) { + rows, err := q.db.Query(ctx, listLifecycleDefs) + if err != nil { + return nil, err + } + defer rows.Close() + var items []LifecycleDef + for rows.Next() { + var i LifecycleDef + if err := rows.Scan( + &i.ID, + &i.States, + &i.DefaultState, + &i.TerminalStates, + &i.Transitions, + &i.CreatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listRelationshipTypes = `-- name: ListRelationshipTypes :many +SELECT name, inverse, source_type, target_type, cardinality, description, created_at FROM relationship_types ORDER BY name +` + +func (q *Queries) ListRelationshipTypes(ctx context.Context) ([]RelationshipType, error) { + rows, err := q.db.Query(ctx, listRelationshipTypes) + if err != nil { + return nil, err + } + defer rows.Close() + var items []RelationshipType + for rows.Next() { + var i RelationshipType + if err := rows.Scan( + &i.Name, + &i.Inverse, + &i.SourceType, + &i.TargetType, + &i.Cardinality, + &i.Description, + &i.CreatedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} diff --git a/internal/db/sqlcgen/operations.sql.go b/internal/db/sqlcgen/operations.sql.go new file mode 100644 index 0000000..a7ada50 --- /dev/null +++ b/internal/db/sqlcgen/operations.sql.go @@ -0,0 +1,357 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.29.0 +// source: operations.sql + +package sqlcgen + +import ( + "context" + "time" + + "github.com/google/uuid" +) + +const getIdempotentResponse = `-- name: GetIdempotentResponse :one +SELECT response_code, response_body, request_hash FROM idempotency_keys +WHERE actor = $1 AND key = $2 +` + +type GetIdempotentResponseParams struct { + Actor string + Key string +} + +type GetIdempotentResponseRow struct { + ResponseCode *int32 + ResponseBody []byte + RequestHash string +} + +func (q *Queries) GetIdempotentResponse(ctx context.Context, arg GetIdempotentResponseParams) (GetIdempotentResponseRow, error) { + row := q.db.QueryRow(ctx, getIdempotentResponse, arg.Actor, arg.Key) + var i GetIdempotentResponseRow + err := row.Scan(&i.ResponseCode, &i.ResponseBody, &i.RequestHash) + return i, err +} + +const insertAuditEntry = `-- name: InsertAuditEntry :exec +INSERT INTO audit_log (actor_type, actor_id, action, entity_id, method, path, + status_code, detail, source_ip, correlation_id) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) +` + +type InsertAuditEntryParams struct { + ActorType string + ActorID *uuid.UUID + Action string + EntityID *uuid.UUID + Method *string + Path *string + StatusCode *int32 + Detail []byte + SourceIp *string + CorrelationID *string +} + +func (q *Queries) InsertAuditEntry(ctx context.Context, arg InsertAuditEntryParams) error { + _, err := q.db.Exec(ctx, insertAuditEntry, + arg.ActorType, + arg.ActorID, + arg.Action, + arg.EntityID, + arg.Method, + arg.Path, + arg.StatusCode, + arg.Detail, + arg.SourceIp, + arg.CorrelationID, + ) + return err +} + +const insertEvent = `-- name: InsertEvent :one +INSERT INTO events (type, entity_id, severity, source, data, correlation_id) +VALUES ($1, $2, $3, $4, $5, $6) +RETURNING id, ts +` + +type InsertEventParams struct { + Type string + EntityID *uuid.UUID + Severity string + Source string + Data []byte + CorrelationID *string +} + +type InsertEventRow struct { + ID int64 + Ts time.Time +} + +func (q *Queries) InsertEvent(ctx context.Context, arg InsertEventParams) (InsertEventRow, error) { + row := q.db.QueryRow(ctx, insertEvent, + arg.Type, + arg.EntityID, + arg.Severity, + arg.Source, + arg.Data, + arg.CorrelationID, + ) + var i InsertEventRow + err := row.Scan(&i.ID, &i.Ts) + return i, err +} + +const listEntityStatus = `-- name: ListEntityStatus :many +SELECT e.slug, e.type, st.health, st.last_check_at +FROM entity_status st JOIN entities e ON e.id = st.entity_id +ORDER BY e.slug +` + +type ListEntityStatusRow struct { + Slug string + Type string + Health string + LastCheckAt *time.Time +} + +func (q *Queries) ListEntityStatus(ctx context.Context) ([]ListEntityStatusRow, error) { + rows, err := q.db.Query(ctx, listEntityStatus) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListEntityStatusRow + for rows.Next() { + var i ListEntityStatusRow + if err := rows.Scan( + &i.Slug, + &i.Type, + &i.Health, + &i.LastCheckAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listEvents = `-- name: ListEvents :many +SELECT id, ts, type, entity_id, severity, source, data, correlation_id +FROM events +WHERE ($1::text IS NULL OR type = $1) + AND ($2::uuid IS NULL OR entity_id = $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) + AND ($7::bigint IS NULL OR id < $7) +ORDER BY id DESC +LIMIT $8 +` + +type ListEventsParams struct { + Type *string + EntityID *uuid.UUID + Severity *string + CorrelationID *string + FromTs *time.Time + ToTs *time.Time + BeforeID *int64 + Lim int32 +} + +func (q *Queries) ListEvents(ctx context.Context, arg ListEventsParams) ([]Event, error) { + rows, err := q.db.Query(ctx, listEvents, + arg.Type, + arg.EntityID, + arg.Severity, + arg.CorrelationID, + arg.FromTs, + arg.ToTs, + arg.BeforeID, + arg.Lim, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []Event + for rows.Next() { + var i Event + if err := rows.Scan( + &i.ID, + &i.Ts, + &i.Type, + &i.EntityID, + &i.Severity, + &i.Source, + &i.Data, + &i.CorrelationID, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listEventsAfter = `-- name: ListEventsAfter :many +SELECT id, ts, type, entity_id, severity, source, data, correlation_id +FROM events WHERE id > $1 ORDER BY id ASC LIMIT $2 +` + +type ListEventsAfterParams struct { + ID int64 + Limit int32 +} + +func (q *Queries) ListEventsAfter(ctx context.Context, arg ListEventsAfterParams) ([]Event, error) { + rows, err := q.db.Query(ctx, listEventsAfter, arg.ID, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + var items []Event + for rows.Next() { + var i Event + if err := rows.Scan( + &i.ID, + &i.Ts, + &i.Type, + &i.EntityID, + &i.Severity, + &i.Source, + &i.Data, + &i.CorrelationID, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listSignals = `-- name: ListSignals :many +SELECT sig.entity_id, se.slug, sig.kind, sig.severity, sig.state, + te.slug AS target_slug, sig.check_id, 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 +` + +type ListSignalsParams struct { + State *string + Severity *string + Target *string + Kind *string + Cursor *string + Lim int32 +} + +type ListSignalsRow struct { + EntityID uuid.UUID + Slug string + Kind string + Severity string + State string + TargetSlug *string + CheckID *uuid.UUID + Evidence *string + LikelyCause *string + OccurrenceCount int32 + FlapCount int32 + HoldDownUntil *time.Time + MuteUntil *time.Time + FirstSeenAt time.Time + LastSeenAt time.Time +} + +func (q *Queries) ListSignals(ctx context.Context, arg ListSignalsParams) ([]ListSignalsRow, error) { + rows, err := q.db.Query(ctx, listSignals, + arg.State, + arg.Severity, + arg.Target, + arg.Kind, + arg.Cursor, + arg.Lim, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListSignalsRow + for rows.Next() { + var i ListSignalsRow + if err := rows.Scan( + &i.EntityID, + &i.Slug, + &i.Kind, + &i.Severity, + &i.State, + &i.TargetSlug, + &i.CheckID, + &i.Evidence, + &i.LikelyCause, + &i.OccurrenceCount, + &i.FlapCount, + &i.HoldDownUntil, + &i.MuteUntil, + &i.FirstSeenAt, + &i.LastSeenAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const putIdempotentResponse = `-- name: PutIdempotentResponse :exec +INSERT INTO idempotency_keys (actor, key, request_hash, response_code, response_body) +VALUES ($1, $2, $3, $4, $5) +ON CONFLICT (actor, key) DO NOTHING +` + +type PutIdempotentResponseParams struct { + Actor string + Key string + RequestHash string + ResponseCode *int32 + ResponseBody []byte +} + +func (q *Queries) PutIdempotentResponse(ctx context.Context, arg PutIdempotentResponseParams) error { + _, err := q.db.Exec(ctx, putIdempotentResponse, + arg.Actor, + arg.Key, + arg.RequestHash, + arg.ResponseCode, + arg.ResponseBody, + ) + return err +} diff --git a/internal/db/sqlcgen/relationships.sql.go b/internal/db/sqlcgen/relationships.sql.go new file mode 100644 index 0000000..851a18b --- /dev/null +++ b/internal/db/sqlcgen/relationships.sql.go @@ -0,0 +1,165 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.29.0 +// source: relationships.sql + +package sqlcgen + +import ( + "context" + "time" + + "github.com/google/uuid" +) + +const endCurrentRelationship = `-- name: EndCurrentRelationship :execrows +UPDATE relationships SET valid_to = now() +WHERE source_id = $1 AND target_id = $2 AND type = $3 AND valid_to IS NULL +` + +type EndCurrentRelationshipParams struct { + SourceID uuid.UUID + TargetID uuid.UUID + Type string +} + +func (q *Queries) EndCurrentRelationship(ctx context.Context, arg EndCurrentRelationshipParams) (int64, error) { + result, err := q.db.Exec(ctx, endCurrentRelationship, arg.SourceID, arg.TargetID, arg.Type) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const listEntityRelations = `-- name: ListEntityRelations :many +SELECT se.slug AS source_slug, te.slug AS target_slug, r.type, r.attributes, + r.valid_from, r.valid_to +FROM relationships r +JOIN entities se ON se.id = r.source_id +JOIN entities te ON te.id = r.target_id +WHERE r.valid_to IS NULL + AND (($1::text IN ('out','both') AND r.source_id = $2) + OR ($1::text IN ('in','both') AND r.target_id = $2)) + AND ($3::text IS NULL OR r.type = $3) +ORDER BY r.type, se.slug, te.slug +` + +type ListEntityRelationsParams struct { + Direction string + ID uuid.UUID + RelType *string +} + +type ListEntityRelationsRow struct { + SourceSlug string + TargetSlug string + Type string + Attributes []byte + ValidFrom time.Time + ValidTo *time.Time +} + +func (q *Queries) ListEntityRelations(ctx context.Context, arg ListEntityRelationsParams) ([]ListEntityRelationsRow, error) { + rows, err := q.db.Query(ctx, listEntityRelations, arg.Direction, arg.ID, arg.RelType) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListEntityRelationsRow + for rows.Next() { + var i ListEntityRelationsRow + if err := rows.Scan( + &i.SourceSlug, + &i.TargetSlug, + &i.Type, + &i.Attributes, + &i.ValidFrom, + &i.ValidTo, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listGraphEdges = `-- name: ListGraphEdges :many +SELECT se.slug AS source_slug, te.slug AS target_slug, r.type, r.attributes, + r.valid_from, r.valid_to +FROM relationships r +JOIN entities se ON se.id = r.source_id +JOIN entities te ON te.id = r.target_id +WHERE r.valid_to IS NULL + AND r.source_id = ANY($1::uuid[]) + AND r.target_id = ANY($1::uuid[]) + AND ($2::text[] IS NULL OR r.type = ANY($2::text[])) +ORDER BY r.type, se.slug, te.slug +` + +type ListGraphEdgesParams struct { + Ids []uuid.UUID + RelTypes []string +} + +type ListGraphEdgesRow struct { + SourceSlug string + TargetSlug string + Type string + Attributes []byte + ValidFrom time.Time + ValidTo *time.Time +} + +func (q *Queries) ListGraphEdges(ctx context.Context, arg ListGraphEdgesParams) ([]ListGraphEdgesRow, error) { + rows, err := q.db.Query(ctx, listGraphEdges, arg.Ids, arg.RelTypes) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListGraphEdgesRow + for rows.Next() { + var i ListGraphEdgesRow + if err := rows.Scan( + &i.SourceSlug, + &i.TargetSlug, + &i.Type, + &i.Attributes, + &i.ValidFrom, + &i.ValidTo, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const upsertCurrentRelationship = `-- name: UpsertCurrentRelationship :exec +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) WHERE valid_to IS NULL +DO UPDATE SET attributes = EXCLUDED.attributes +` + +type UpsertCurrentRelationshipParams struct { + SourceID uuid.UUID + TargetID uuid.UUID + Type string + Attributes []byte +} + +func (q *Queries) UpsertCurrentRelationship(ctx context.Context, arg UpsertCurrentRelationshipParams) error { + _, err := q.db.Exec(ctx, upsertCurrentRelationship, + arg.SourceID, + arg.TargetID, + arg.Type, + arg.Attributes, + ) + return err +} diff --git a/internal/observability/record.go b/internal/observability/record.go new file mode 100644 index 0000000..6322cb9 --- /dev/null +++ b/internal/observability/record.go @@ -0,0 +1,65 @@ +package observability + +import ( + "context" + "encoding/json" + + "github.com/dtoro/oikos/internal/db/sqlcgen" + "github.com/google/uuid" +) + +// Audit writes an audit_log entry. Pass a transaction-bound Queries so the +// entry commits or rolls back atomically with the state change it records. +// +// actorLabel is the interim textual actor identity ("operator:dev", +// "agent:mcp") recorded in detail; actor_id (a Person/Agent entity UUID) +// starts being populated when OIDC identity resolution lands. +func Audit(ctx context.Context, q *sqlcgen.Queries, actorType, actorLabel, + action string, entityID *uuid.UUID, method, path, correlationID string, + detail map[string]any) error { + + if detail == nil { + detail = map[string]any{} + } + detail["actor"] = actorLabel + detailJSON, _ := json.Marshal(detail) + + var corr *string + if correlationID != "" { + corr = &correlationID + } + return q.InsertAuditEntry(ctx, sqlcgen.InsertAuditEntryParams{ + ActorType: actorType, + Action: action, + EntityID: entityID, + Method: &method, + Path: &path, + Detail: detailJSON, + CorrelationID: corr, + }) +} + +// Event emits a structured event in the caller's transaction (SG10). The +// post-commit NOTIFY trigger (migration 008) fans it out to SSE subscribers. +func Event(ctx context.Context, q *sqlcgen.Queries, eventType string, + entityID *uuid.UUID, severity, source string, correlationID string, + data map[string]any) error { + + dataJSON, _ := json.Marshal(data) + if data == nil { + dataJSON = []byte("{}") + } + var corr *string + if correlationID != "" { + corr = &correlationID + } + _, err := q.InsertEvent(ctx, sqlcgen.InsertEventParams{ + Type: eventType, + EntityID: entityID, + Severity: severity, + Source: source, + Data: dataJSON, + CorrelationID: corr, + }) + return err +} diff --git a/migrations/008_event_notify.up.sql b/migrations/008_event_notify.up.sql new file mode 100644 index 0000000..6ea94c4 --- /dev/null +++ b/migrations/008_event_notify.up.sql @@ -0,0 +1,24 @@ +-- Migration 008: post-commit event fan-out for the SSE stream (SG8/SG10). +-- Events are INSERTed in the same transaction as the state change; pg_notify +-- fires at COMMIT, so subscribers only ever see committed events. Payload is +-- kept minimal (NOTIFY has an 8000-byte limit); consumers needing the full +-- event fetch it by id. + +CREATE OR REPLACE FUNCTION notify_oikos_event() RETURNS trigger AS $$ +BEGIN + PERFORM pg_notify('oikos_events', json_build_object( + 'id', NEW.id, + 'ts', NEW.ts, + 'type', NEW.type, + 'entity_id', NEW.entity_id, + 'severity', NEW.severity, + 'source', NEW.source, + 'correlation_id', NEW.correlation_id + )::text); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS trg_events_notify ON events; +CREATE TRIGGER trg_events_notify AFTER INSERT ON events + FOR EACH ROW EXECUTE FUNCTION notify_oikos_event(); diff --git a/sqlc.yaml b/sqlc.yaml new file mode 100644 index 0000000..1310201 --- /dev/null +++ b/sqlc.yaml @@ -0,0 +1,39 @@ +# sqlc — type-safe Go from SQL (plan SG17). `make generate` regenerates. +# +# Scope: API read/mutation paths use sqlc-generated queries +# (internal/db/sqlcgen). The seed ingest and YAML export intentionally stay +# hand-written pgx: they are generic bulk upserts driven by parsed YAML +# shapes, where sqlc's static typing adds nothing. +version: "2" +sql: + - engine: "postgresql" + schema: "migrations" + queries: "internal/db/queries" + strict_order_by: false + gen: + go: + package: "sqlcgen" + out: "internal/db/sqlcgen" + sql_package: "pgx/v5" + emit_pointers_for_null_types: true + overrides: + - db_type: "uuid" + go_type: + import: "github.com/google/uuid" + type: "UUID" + - db_type: "uuid" + nullable: true + go_type: + import: "github.com/google/uuid" + type: "UUID" + pointer: true + - db_type: "timestamptz" + go_type: + import: "time" + type: "Time" + - db_type: "timestamptz" + nullable: true + go_type: + import: "time" + type: "Time" + pointer: true