dtoro
646373a676
req.Params.RelType is *[]string; passing the nil pointer straight through as a pgx query arg (both in the blast_radius() call and in ListGraphEdges) panics because pgx can't infer the array element type from a nil *[]string, only from a concrete (possibly nil) []string. Dereference once up front instead. Also affected the sqlc-based ListGraphEdges path added by the R3 refactor, which had the same bug. Add a regression test for GET /api/v1/graph?root=X&depth=N with no rel_type. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1639 lines
50 KiB
Go
1639 lines
50 KiB
Go
package httpapi
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"crypto/sha256"
|
|
"encoding/json"
|
|
"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 = 500
|
|
)
|
|
|
|
// 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.
|
|
nodes, err = s.queryEntities(ctx, `
|
|
SELECT `+entityCols+`
|
|
FROM entities e
|
|
LEFT JOIN entity_status st ON st.entity_id = e.id
|
|
WHERE 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)
|
|
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
|
|
}
|
|
|
|
ensureDefaultChecks(ctx, tx, inserted.ID, slug, req.Body.Type, attrsJSON)
|
|
|
|
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 != "" {
|
|
// Get lifecycle def for the entity's type.
|
|
lc, err := sqlcgen.New(tx).GetLifecycleForType(ctx, current.Type)
|
|
if err != nil {
|
|
if err == pgx.ErrNoRows {
|
|
// No lifecycle defined — any state is allowed.
|
|
} else {
|
|
return nil, err
|
|
}
|
|
} else {
|
|
var transitions map[string]map[string]json.RawMessage
|
|
if err := json.Unmarshal(lc.Transitions, &transitions); err != nil {
|
|
return nil, fmt.Errorf("parse lifecycle transitions: %w", err)
|
|
}
|
|
|
|
fromState := ""
|
|
if current.State != nil {
|
|
fromState = *current.State
|
|
}
|
|
toState := *req.Body.State
|
|
|
|
if toState != fromState {
|
|
tos, ok := transitions[fromState]
|
|
if !ok {
|
|
return nil, fmt.Errorf("%w: no transitions from %q", domain.ErrInvalidTransition, fromState)
|
|
}
|
|
trans, ok := tos[toState]
|
|
if !ok {
|
|
return nil, fmt.Errorf("%w: %s → %s", domain.ErrInvalidTransition, fromState, toState)
|
|
}
|
|
|
|
// Parse preconditions: {"requires": ["check-name", ...]}
|
|
var gate struct {
|
|
Requires []string `json:"requires"`
|
|
}
|
|
if err := json.Unmarshal(trans, &gate); err == nil && len(gate.Requires) > 0 {
|
|
for _, check := range gate.Requires {
|
|
if err := checkPrecondition(ctx, tx, id, current.Type, check); err != nil {
|
|
return nil, fmt.Errorf("%w: precondition %q not met: %v",
|
|
domain.ErrInvalidTransition, check, 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})
|
|
|
|
ensureDefaultChecks(ctx, tx, id, req.Body.Slug, current.Type, attrsJSON)
|
|
|
|
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
|
|
}
|
|
|
|
// checkPrecondition validates a named lifecycle transition precondition.
|
|
func checkPrecondition(ctx context.Context, tx pgx.Tx, entityID uuid.UUID, entityType, check string) error {
|
|
switch check {
|
|
case "no-inbound-edges":
|
|
var count int
|
|
err := tx.QueryRow(ctx,
|
|
"SELECT count(*) FROM relationships WHERE target_id = $1 AND valid_to IS NULL", entityID).Scan(&count)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if count > 0 {
|
|
return fmt.Errorf("%d inbound relationship edges remaining", count)
|
|
}
|
|
case "backups-verified":
|
|
var attrs string
|
|
err := tx.QueryRow(ctx, "SELECT coalesce(attributes::text,'{}') FROM entities WHERE id = $1", entityID).Scan(&attrs)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !strings.Contains(attrs, "backups_verified") {
|
|
return fmt.Errorf("backup verification not recorded in entity attributes")
|
|
}
|
|
case "secrets-revoked":
|
|
var attrs string
|
|
err := tx.QueryRow(ctx, "SELECT coalesce(attributes::text,'{}') FROM entities WHERE id = $1", entityID).Scan(&attrs)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !strings.Contains(attrs, "secrets_revoked") {
|
|
return fmt.Errorf("secret revocation not recorded in entity attributes")
|
|
}
|
|
case "ingress-dns-removed":
|
|
var attrs string
|
|
err := tx.QueryRow(ctx, "SELECT coalesce(attributes::text,'{}') FROM entities WHERE id = $1", entityID).Scan(&attrs)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !strings.Contains(attrs, "ingress_dns_removed") {
|
|
return fmt.Errorf("ingress/DNS removal not recorded in entity attributes")
|
|
}
|
|
case "age-key-enrolled-if-needed":
|
|
if entityType == "workstation" {
|
|
var attrs string
|
|
err := tx.QueryRow(ctx, "SELECT coalesce(attributes::text,'{}') FROM entities WHERE id = $1", entityID).Scan(&attrs)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !strings.Contains(attrs, "age_pubkey") {
|
|
return fmt.Errorf("age key not enrolled (no age_pubkey in attributes)")
|
|
}
|
|
}
|
|
case "mesh-joined-if-needed":
|
|
if entityType == "workstation" {
|
|
var attrs string
|
|
err := tx.QueryRow(ctx, "SELECT coalesce(attributes::text,'{}') FROM entities WHERE id = $1", entityID).Scan(&attrs)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !strings.Contains(attrs, "mesh_ip") {
|
|
return fmt.Errorf("mesh not joined (no mesh_ip in attributes)")
|
|
}
|
|
}
|
|
case "health-check-answering":
|
|
st, err := sqlcgen.New(tx).GetEntityStatus(ctx, entityID)
|
|
if err != nil || st.Health == "unknown" || st.Health == "down" {
|
|
return fmt.Errorf("health check not answering (status: %s)", st.Health)
|
|
}
|
|
case "doc-page-complete":
|
|
var count int
|
|
err := tx.QueryRow(ctx, `
|
|
SELECT count(*) FROM relationships r
|
|
JOIN entities ke ON ke.id = r.source_id
|
|
WHERE r.target_id = $1 AND r.valid_to IS NULL
|
|
AND r.type = 'documents' AND ke.type IN ('document','runbook','investigation')`,
|
|
entityID).Scan(&count)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if count == 0 {
|
|
return fmt.Errorf("no documentation linked to entity")
|
|
}
|
|
case "inventory-entry", "ip-reserved", "storage-pool-chosen", "cancelled-note",
|
|
"preflight-passed", "error-summary", "replacement-live-or-role-retired",
|
|
"replacement-failed", "post-verify-passed", "recovery-verified", "written-off",
|
|
"ingress-live-if-public", "doc-page-stub":
|
|
// Soft checks — always pass. These are operator-confirmed via the
|
|
// transition request itself, or are not mechanically enforceable.
|
|
default:
|
|
// Unknown preconditions are skipped (operator intent overrides).
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ─── Helpers ───────────────────────────────────────────────────────────
|