package db import ( "context" "encoding/json" "fmt" "github.com/dtoro/oikos/internal/core/app" "github.com/dtoro/oikos/internal/ontology" "github.com/google/uuid" "github.com/jackc/pgx/v5" ) // EnsureEntityChecks derives an entity's default check_defs from the // monitoring spec of its type (resolving per-entity `monitoring` overrides). // // This is the single shared hook that keeps the check graph in sync with // entity mutations. Both the HTTP create/patch handlers and the MCP // entity-mutation tools (create_entity, update_entity_attributes) call it so // that flipping an entity's `monitoring` attribute regenerates checks // regardless of which surface made the change — previously only the HTTP // path ran check derivation, so entities mutated via MCP silently produced no // checks (see plans/2026-08-03-session-review-haos-monitoring-capability-gaps.md, A2). func EnsureEntityChecks(ctx context.Context, tx pgx.Tx, id uuid.UUID, slug, entityType, name string, attrs []byte) (app.DeriveResult, error) { tree, err := LoadTypeTree(ctx, tx) if err != nil { return app.DeriveResult{}, err } res, err := EnsureChecks(ctx, tx, tree, app.CheckTarget{ ID: id.String(), Slug: slug, Type: entityType, Name: name, Attrs: attrs, }) if err != nil { return res, err } app.LogDeriveResult(slug, entityType, res) return res, nil } // EnsureChecks writes the derived check_defs for one entity, idempotently. // Derivation is pure core logic (app.Derive); this function owns the // entity_status row, the graph host fallback, and the upserts. func EnsureChecks(ctx context.Context, tx pgx.Tx, tree *ontology.TypeTree, t app.CheckTarget) (app.DeriveResult, error) { var res app.DeriveResult if _, err := tx.Exec(ctx, `INSERT INTO entity_status (entity_id, health, updated_at) VALUES ($1, 'unknown', now()) ON CONFLICT (entity_id) DO NOTHING`, t.ID); err != nil { return res, fmt.Errorf("entity_status %s: %w", t.Slug, err) } defs, dres := app.Derive(tree, t, func() map[string]any { attrs, err := hostViaGraph(ctx, tx, t.ID) if err != nil { return nil } return attrs }) res.Skipped, res.Undeclared = dres.Skipped, dres.Undeclared for i, def := range defs { created, err := writeCheck(ctx, tx, t, i, def) if err != nil { return res, fmt.Errorf("check %s/%s: %w", t.Slug, def.Kind, err) } if created { res.Created++ } } return res, nil } // writeCheck upserts one check_def and its backing check entity. // // The entity upsert MUST return the row's id. The previous version generated // a fresh uuid, inserted ON CONFLICT (slug) DO NOTHING, then wrote a // check_defs row referencing that uuid. On any re-seed the slug already // existed, the entity insert became a no-op, and the check_defs insert // violated its foreign key — which aborted the whole ingest transaction and // made every subsequent statement fail with 25P02. Because the errors were // discarded, the only visible symptom was an unrelated failure much later. func writeCheck(ctx context.Context, tx pgx.Tx, t app.CheckTarget, idx int, def app.CheckDef) (bool, error) { // The full target slug, not a truncation of it. shortSlug() took the last // 8 characters, so all 21 ingress routes collapsed to ".network" and // generated one identical check slug — they overwrote each other and 20 // of them ended up with no check at all. It also collided service:jellyfin // with lxc:jellyfin. Entity slugs are unique; use them. checkSlug := fmt.Sprintf("check:%s:%s:%d", def.Kind, t.Slug, idx) newID, err := uuid.NewV7() if err != nil { newID = uuid.New() } var checkID uuid.UUID err = tx.QueryRow(ctx, `INSERT INTO entities (id, slug, type, name, state, attributes, version, created_at, updated_at) VALUES ($1, $2, 'check', $2, 'active', '{}', 1, now(), now()) ON CONFLICT (slug) DO UPDATE SET updated_at = now() RETURNING id`, newID, checkSlug).Scan(&checkID) if err != nil { return false, fmt.Errorf("upsert check entity %s: %w", checkSlug, err) } configJSON, err := json.Marshal(def.Config) if err != nil { return false, err } // Config is derived from the seed, so the seed wins on re-ingest and // attribute changes propagate. `enabled` is deliberately left alone: it // is operational state an operator may have toggled. // last_run_at is seeded to a random point inside the interval so checks // created together do not stay in lockstep. Every check the seed creates // would otherwise come due in the same instant forever: ~165 probes // landing at once each minute rather than spread across it. Deliberately // absent from the DO UPDATE below — a re-seed must not reset the schedule // and re-herd everything. tag, err := tx.Exec(ctx, `INSERT INTO check_defs (entity_id, target_id, target_type, kind, config, interval_s, timeout_s, enabled, last_run_at) VALUES ($1, $2, $6, $3, $4, $5, 30, true, now() - make_interval(secs => random() * $5::int)) ON CONFLICT (entity_id) DO UPDATE SET target_id = EXCLUDED.target_id, target_type = EXCLUDED.target_type, kind = EXCLUDED.kind, config = EXCLUDED.config, interval_s = EXCLUDED.interval_s, updated_at = now()`, checkID, t.ID, def.Kind, configJSON, def.IntervalS, t.Type) if err != nil { return false, fmt.Errorf("upsert check_def %s: %w", checkSlug, err) } return tag.RowsAffected() > 0, nil } // hostViaGraph returns the attributes of the entity that hosts or provides // this one, so a service can inherit its container's address. func hostViaGraph(ctx context.Context, tx pgx.Tx, entityID string) (map[string]any, error) { rows, err := tx.Query(ctx, ` SELECT e.attributes FROM relationships r JOIN entities e ON e.id = r.source_id WHERE r.target_id = $1 AND r.valid_to IS NULL -- backs-up-to points from the thing being backed up TO the target, -- so walking it backwards finds the machine that writes the backups -- — which is the only place a freshness check can run. AND r.type IN ('provides', 'hosts', 'runs-on', 'backs-up-to') ORDER BY CASE r.type WHEN 'provides' THEN 0 WHEN 'runs-on' THEN 1 WHEN 'backs-up-to' THEN 2 ELSE 3 END`, entityID) if err != nil { return nil, err } defer rows.Close() for rows.Next() { var raw []byte if err := rows.Scan(&raw); err != nil { return nil, err } var attrs map[string]any if json.Unmarshal(raw, &attrs) != nil { continue } if resolveGraphHost(attrs) != "" { return attrs, nil } } return nil, rows.Err() } // resolveGraphHost mirrors app's address resolution for graph-walk results. // It re-implements the small pure helper rather than exporting internals of // the core package: the shapes it accepts are exactly the seed attribute // shapes hostViaGraph can return. func resolveGraphHost(attrs map[string]any) string { if attrs == nil { return "" } if ip, ok := attrs["lan_ip"].(string); ok && ip != "" { return ip } if ip, ok := attrs["public_ipv4"].(string); ok && ip != "" { return ip } if mesh, ok := attrs["mesh"].(map[string]any); ok { if nb, ok := mesh["netbird"].(map[string]any); ok { if ip, ok := nb["ip"].(string); ok && ip != "" { return ip } if fqdn, ok := nb["fqdn"].(string); ok && fqdn != "" { return fqdn } } } if ip, ok := attrs["mesh_ip"].(string); ok && ip != "" { return ip } for _, key := range []string{"host", "address", "public_host"} { if v, ok := attrs[key].(string); ok && v != "" { return v } } return "" }