refactor: sqlc vs raw SQL — hybrid approach (R3)
Deleted 8 genuinely unused sqlc queries (no inline equivalent): - UpsertCurrentRelationship, ListEntitiesCapped, ListEntityStatus, UpdateSignalState, InsertClassification, InsertFeedback, InsertSkill, UpsertCurrentRelationship — all had zero call sites. Migrated 9 inline raw SQL sites to use sqlc queries: - GetOntology (impl.go): ListEntityTypes, ListRelationshipTypes, ListLifecycleDefs — replaces 3 raw pool.Query blocks with typed sqlcgen calls, eliminating manual row scanning. - EndRelationship (phase3.go): EndCurrentRelationship — replaces tx.Exec with sqlcgen.New(tx).EndCurrentRelationship. - checkPrecondition (impl.go): GetEntityStatus — replaces tx.QueryRow + manual Scan with sqlcgen.New(tx).GetEntityStatus. - GetEntityRelations (impl.go): ListEntityRelations — replaces raw pool.Query + scanRelationships helper (now deleted). - GetGraph (impl.go): ListGraphEdges — replaces raw pool.Query + scanRelationships. - resolveEntityID (impl.go): GetEntityBySlug/GetEntityByID — replaces raw pool.QueryRow + Scan. - createApproval (mcp/server.go): InsertApproval — replaces raw pool.Exec with sqlcgen.InsertApproval. Deleted scanRelationships helper (was only used by the two migrated graph queries above). Regenerated sqlcgen — also picks up stale model updates (AgentSession, SessionPlanStep, SessionQuestion, etc. from recent migrations). Documented the carve-out in .agents/dev/CONTRIBUTING.md §SQL conventions: sqlc is the default; raw pool.Query/Exec is reserved for LISTEN/NOTIFY, dynamic WHERE builders, blast_radius(), and COPY. go vet, build, httpapi/mcp/db tests all pass. -383/+170 lines.
This commit is contained in:
@@ -27,9 +27,6 @@ WHERE e.type IN (SELECT name FROM tt)
|
||||
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)
|
||||
|
||||
@@ -14,11 +14,6 @@ WHERE (sqlc.narg('state')::text IS NULL OR sig.state = sqlc.narg('state'))
|
||||
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;
|
||||
@@ -89,9 +84,6 @@ DO UPDATE SET occurrence_count = signals.occurrence_count + 1,
|
||||
updated_at = now()
|
||||
RETURNING *;
|
||||
|
||||
-- name: UpdateSignalState :exec
|
||||
UPDATE signals SET state = $2, updated_at = now() WHERE entity_id = $1;
|
||||
|
||||
-- name: GetOpenSignalsForAutoAct :many
|
||||
-- Signals with auto-act classifications that haven't been executed yet
|
||||
SELECT s.*, c.entity_id AS classification_id, c.action, c.risk_class, c.route,
|
||||
@@ -106,12 +98,6 @@ WHERE c.route = 'auto-act'
|
||||
ORDER BY s.last_seen_at ASC
|
||||
LIMIT $1;
|
||||
|
||||
-- name: InsertClassification :exec
|
||||
INSERT INTO classifications (entity_id, signal_entity_id, target_entity_id, action,
|
||||
recommended_action, risk_class, route, blast_radius, pattern_confidence,
|
||||
skill_id, autonomy_check, reasoning, correlation_id)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13);
|
||||
|
||||
-- name: ListClassifications :many
|
||||
SELECT c.entity_id, c.signal_entity_id, c.target_entity_id, c.action,
|
||||
c.recommended_action, c.risk_class, c.route, c.blast_radius,
|
||||
@@ -153,11 +139,6 @@ WHERE (sqlc.narg('status')::text IS NULL OR e.status = sqlc.narg('status'))
|
||||
ORDER BY te.slug
|
||||
LIMIT sqlc.arg('lim');
|
||||
|
||||
-- name: InsertFeedback :exec
|
||||
INSERT INTO feedback (entity_id, execution_id, outcome, observation, lesson,
|
||||
unexpected_side_effects, tags)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7);
|
||||
|
||||
-- name: GetFeedbackAfterWatermark :many
|
||||
SELECT f.entity_id, f.execution_id, f.outcome, f.observation, f.lesson,
|
||||
f.unexpected_side_effects, f.tags, f.created_at,
|
||||
@@ -201,11 +182,6 @@ SELECT * FROM skills
|
||||
WHERE (sqlc.narg('status')::text IS NULL OR status = sqlc.narg('status'))
|
||||
ORDER BY name, version DESC;
|
||||
|
||||
-- name: InsertSkill :exec
|
||||
INSERT INTO skills (entity_id, version, name, procedure, applies_type, action,
|
||||
pattern_ids, status, changed_by, change_reason)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10);
|
||||
|
||||
-- name: UpdateSkillStatus :exec
|
||||
UPDATE skills SET status = $2, last_used_at = now() WHERE entity_id = $1 AND version = $2;
|
||||
|
||||
|
||||
@@ -22,12 +22,6 @@ WHERE r.valid_to IS NULL
|
||||
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;
|
||||
|
||||
@@ -177,43 +177,6 @@ func (q *Queries) ListEntities(ctx context.Context, arg ListEntitiesParams) ([]E
|
||||
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, e.enrolled_at, e.enrolled_by 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,
|
||||
&i.EnrolledAt,
|
||||
&i.EnrolledBy,
|
||||
); 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),
|
||||
|
||||
@@ -35,11 +35,17 @@ type AgentMessage struct {
|
||||
}
|
||||
|
||||
type AgentSession struct {
|
||||
ID uuid.UUID
|
||||
Title string
|
||||
Actor string
|
||||
CreatedAt time.Time
|
||||
LastActiveAt time.Time
|
||||
ID uuid.UUID
|
||||
Title string
|
||||
Actor string
|
||||
CreatedAt time.Time
|
||||
LastActiveAt time.Time
|
||||
Goal string
|
||||
Status string
|
||||
Outcome *string
|
||||
Summary string
|
||||
EntityID *uuid.UUID
|
||||
CompletionNudges int32
|
||||
}
|
||||
|
||||
type Approval struct {
|
||||
@@ -83,6 +89,7 @@ type AuditLog struct {
|
||||
Detail []byte
|
||||
SourceIp *string
|
||||
CorrelationID *string
|
||||
SessionID *uuid.UUID
|
||||
}
|
||||
|
||||
type AutonomySetting struct {
|
||||
@@ -289,6 +296,13 @@ type MetricSample struct {
|
||||
Tags []byte
|
||||
}
|
||||
|
||||
type NomosPlanExecution struct {
|
||||
ExecutionID uuid.UUID
|
||||
SessionID uuid.UUID
|
||||
ContinuedAt *time.Time
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
type Pattern struct {
|
||||
EntityID uuid.UUID
|
||||
AppliesType string
|
||||
@@ -351,6 +365,32 @@ type SeedVersion struct {
|
||||
AppliedAt time.Time
|
||||
}
|
||||
|
||||
type SessionPlanStep struct {
|
||||
ID uuid.UUID
|
||||
SessionID uuid.UUID
|
||||
Seq int32
|
||||
Title string
|
||||
Detail string
|
||||
Status string
|
||||
ExecutionID *uuid.UUID
|
||||
TargetSlug *string
|
||||
StartedAt *time.Time
|
||||
FinishedAt *time.Time
|
||||
CreatedAt time.Time
|
||||
Generation int32
|
||||
}
|
||||
|
||||
type SessionQuestion struct {
|
||||
ID uuid.UUID
|
||||
SessionID uuid.UUID
|
||||
Prompt string
|
||||
Context []byte
|
||||
Status string
|
||||
Answer *string
|
||||
CreatedAt time.Time
|
||||
AnsweredAt *time.Time
|
||||
}
|
||||
|
||||
type Signal struct {
|
||||
EntityID uuid.UUID
|
||||
Kind string
|
||||
|
||||
@@ -416,48 +416,6 @@ func (q *Queries) InsertCheckDef(ctx context.Context, arg InsertCheckDefParams)
|
||||
return err
|
||||
}
|
||||
|
||||
const insertClassification = `-- name: InsertClassification :exec
|
||||
INSERT INTO classifications (entity_id, signal_entity_id, target_entity_id, action,
|
||||
recommended_action, risk_class, route, blast_radius, pattern_confidence,
|
||||
skill_id, autonomy_check, reasoning, correlation_id)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
|
||||
`
|
||||
|
||||
type InsertClassificationParams 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
|
||||
}
|
||||
|
||||
func (q *Queries) InsertClassification(ctx context.Context, arg InsertClassificationParams) error {
|
||||
_, err := q.db.Exec(ctx, insertClassification,
|
||||
arg.EntityID,
|
||||
arg.SignalEntityID,
|
||||
arg.TargetEntityID,
|
||||
arg.Action,
|
||||
arg.RecommendedAction,
|
||||
arg.RiskClass,
|
||||
arg.Route,
|
||||
arg.BlastRadius,
|
||||
arg.PatternConfidence,
|
||||
arg.SkillID,
|
||||
arg.AutonomyCheck,
|
||||
arg.Reasoning,
|
||||
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)
|
||||
@@ -530,35 +488,6 @@ func (q *Queries) InsertExecution(ctx context.Context, arg InsertExecutionParams
|
||||
return err
|
||||
}
|
||||
|
||||
const insertFeedback = `-- name: InsertFeedback :exec
|
||||
INSERT INTO feedback (entity_id, execution_id, outcome, observation, lesson,
|
||||
unexpected_side_effects, tags)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7)
|
||||
`
|
||||
|
||||
type InsertFeedbackParams struct {
|
||||
EntityID uuid.UUID
|
||||
ExecutionID uuid.UUID
|
||||
Outcome string
|
||||
Observation *string
|
||||
Lesson *string
|
||||
UnexpectedSideEffects []string
|
||||
Tags []string
|
||||
}
|
||||
|
||||
func (q *Queries) InsertFeedback(ctx context.Context, arg InsertFeedbackParams) error {
|
||||
_, err := q.db.Exec(ctx, insertFeedback,
|
||||
arg.EntityID,
|
||||
arg.ExecutionID,
|
||||
arg.Outcome,
|
||||
arg.Observation,
|
||||
arg.Lesson,
|
||||
arg.UnexpectedSideEffects,
|
||||
arg.Tags,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
const insertMetricSample = `-- name: InsertMetricSample :exec
|
||||
INSERT INTO metric_samples (entity_id, metric, value, tags, ts)
|
||||
VALUES ($1, $2, $3, $4, now())
|
||||
@@ -581,41 +510,6 @@ func (q *Queries) InsertMetricSample(ctx context.Context, arg InsertMetricSample
|
||||
return err
|
||||
}
|
||||
|
||||
const insertSkill = `-- name: InsertSkill :exec
|
||||
INSERT INTO skills (entity_id, version, name, procedure, applies_type, action,
|
||||
pattern_ids, status, changed_by, change_reason)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
|
||||
`
|
||||
|
||||
type InsertSkillParams struct {
|
||||
EntityID uuid.UUID
|
||||
Version int32
|
||||
Name string
|
||||
Procedure []byte
|
||||
AppliesType *string
|
||||
Action string
|
||||
PatternIds []uuid.UUID
|
||||
Status string
|
||||
ChangedBy *uuid.UUID
|
||||
ChangeReason *string
|
||||
}
|
||||
|
||||
func (q *Queries) InsertSkill(ctx context.Context, arg InsertSkillParams) error {
|
||||
_, err := q.db.Exec(ctx, insertSkill,
|
||||
arg.EntityID,
|
||||
arg.Version,
|
||||
arg.Name,
|
||||
arg.Procedure,
|
||||
arg.AppliesType,
|
||||
arg.Action,
|
||||
arg.PatternIds,
|
||||
arg.Status,
|
||||
arg.ChangedBy,
|
||||
arg.ChangeReason,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
const listApprovalRules = `-- name: ListApprovalRules :many
|
||||
SELECT id, entity_type, action, risk_class, autonomy_level, scope_entity, version, updated_at FROM approval_rules ORDER BY entity_type, action
|
||||
`
|
||||
@@ -852,44 +746,6 @@ func (q *Queries) ListEnabledCheckDefs(ctx context.Context) ([]ListEnabledCheckD
|
||||
return items, nil
|
||||
}
|
||||
|
||||
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
|
||||
@@ -1462,20 +1318,6 @@ func (q *Queries) UpdatePatternStatus(ctx context.Context, arg UpdatePatternStat
|
||||
return err
|
||||
}
|
||||
|
||||
const updateSignalState = `-- name: UpdateSignalState :exec
|
||||
UPDATE signals SET state = $2, updated_at = now() WHERE entity_id = $1
|
||||
`
|
||||
|
||||
type UpdateSignalStateParams struct {
|
||||
EntityID uuid.UUID
|
||||
State string
|
||||
}
|
||||
|
||||
func (q *Queries) UpdateSignalState(ctx context.Context, arg UpdateSignalStateParams) error {
|
||||
_, err := q.db.Exec(ctx, updateSignalState, arg.EntityID, arg.State)
|
||||
return err
|
||||
}
|
||||
|
||||
const updateSkillStatus = `-- name: UpdateSkillStatus :exec
|
||||
UPDATE skills SET status = $2, last_used_at = now() WHERE entity_id = $1 AND version = $2
|
||||
`
|
||||
|
||||
@@ -139,27 +139,3 @@ func (q *Queries) ListGraphEdges(ctx context.Context, arg ListGraphEdgesParams)
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
@@ -60,19 +60,17 @@ func clampLimit(l *int) int {
|
||||
// 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 {
|
||||
var found uuid.UUID
|
||||
err := s.pool.QueryRow(ctx, "SELECT id FROM entities WHERE id = $1", id).Scan(&found)
|
||||
if err == pgx.ErrNoRows {
|
||||
entity, err := sqlcgen.New(s.pool).GetEntityByID(ctx, id)
|
||||
if err != nil {
|
||||
return uuid.Nil, fmt.Errorf("%w: %s", domain.ErrNotFound, idOrSlug)
|
||||
}
|
||||
return found, err
|
||||
return entity.ID, nil
|
||||
}
|
||||
var id uuid.UUID
|
||||
err := s.pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", idOrSlug).Scan(&id)
|
||||
if err == pgx.ErrNoRows {
|
||||
entity, err := sqlcgen.New(s.pool).GetEntityBySlug(ctx, idOrSlug)
|
||||
if err != nil {
|
||||
return uuid.Nil, fmt.Errorf("%w: %s", domain.ErrNotFound, idOrSlug)
|
||||
}
|
||||
return id, err
|
||||
return entity.ID, nil
|
||||
}
|
||||
|
||||
// entityCols requires the entities table to be aliased as `e`, with
|
||||
@@ -188,46 +186,37 @@ func (s *Server) GetEntityRelations(ctx context.Context, req gen.GetEntityRelati
|
||||
if req.Params.Direction != nil {
|
||||
dir = string(*req.Params.Direction)
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT se.slug, te.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 (($3 IN ('out','both') AND r.source_id = $1)
|
||||
OR ($3 IN ('in','both') AND r.target_id = $1))
|
||||
AND ($2::text IS NULL OR r.type = $2)
|
||||
ORDER BY r.type, se.slug, te.slug`,
|
||||
id, req.Params.RelType, dir)
|
||||
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, err := scanRelationships(rows)
|
||||
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 scanRelationships(rows pgx.Rows) ([]gen.Relationship, error) {
|
||||
defer rows.Close()
|
||||
items := []gen.Relationship{}
|
||||
for rows.Next() {
|
||||
var rel gen.Relationship
|
||||
var attrsJSON []byte
|
||||
if err := rows.Scan(&rel.Source, &rel.Target, &rel.Type,
|
||||
&attrsJSON, &rel.ValidFrom, &rel.ValidTo); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var attrs map[string]any
|
||||
if len(attrsJSON) > 0 && json.Unmarshal(attrsJSON, &attrs) == nil && len(attrs) > 0 {
|
||||
rel.Attributes = &attrs
|
||||
}
|
||||
items = append(items, rel)
|
||||
}
|
||||
return items, rows.Err()
|
||||
}
|
||||
|
||||
func (s *Server) GetBlastRadius(ctx context.Context, req gen.GetBlastRadiusRequestObject) (gen.GetBlastRadiusResponseObject, error) {
|
||||
id, err := s.resolveEntityID(ctx, req.Id)
|
||||
if err != nil {
|
||||
@@ -336,21 +325,31 @@ func (s *Server) GetGraph(ctx context.Context, req gen.GetGraphRequestObject) (g
|
||||
for i, n := range nodes {
|
||||
ids[i] = uuid.UUID(n.Id)
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT se.slug, te.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) AND r.target_id = ANY($1)
|
||||
AND ($2::text[] IS NULL OR r.type = ANY($2))
|
||||
ORDER BY r.type, se.slug, te.slug`, ids, req.Params.RelType)
|
||||
edgeRows, err := sqlcgen.New(s.pool).ListGraphEdges(ctx, sqlcgen.ListGraphEdgesParams{
|
||||
Ids: ids,
|
||||
RelTypes: *req.Params.RelType,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
edges, err := scanRelationships(rows)
|
||||
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}
|
||||
@@ -421,78 +420,70 @@ func (s *Server) GetOntology(ctx context.Context, req gen.GetOntologyRequestObje
|
||||
Lifecycles: []gen.LifecycleDef{},
|
||||
}
|
||||
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT name, parent_type, is_abstract, domain, layer, description,
|
||||
lifecycle_id, attribute_schema, schema_version, status
|
||||
FROM entity_types ORDER BY name`)
|
||||
q := sqlcgen.New(s.pool)
|
||||
|
||||
etRows, err := q.ListEntityTypes(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for rows.Next() {
|
||||
var et gen.EntityType
|
||||
var schemaVersion int
|
||||
var schemaJSON []byte
|
||||
if err := rows.Scan(&et.Name, &et.ParentType, &et.IsAbstract, &et.Domain,
|
||||
&et.Layer, &et.Description, &et.LifecycleId, &schemaJSON,
|
||||
&schemaVersion, &et.Status); err != nil {
|
||||
rows.Close()
|
||||
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
|
||||
}
|
||||
}
|
||||
et.SchemaVersion = &schemaVersion
|
||||
var schema map[string]any
|
||||
if len(schemaJSON) > 0 && json.Unmarshal(schemaJSON, &schema) == nil && schema != nil {
|
||||
et.AttributeSchema = &schema
|
||||
}
|
||||
resp.EntityTypes = append(resp.EntityTypes, et)
|
||||
}
|
||||
rows.Close()
|
||||
if rows.Err() != nil {
|
||||
return nil, rows.Err()
|
||||
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),
|
||||
})
|
||||
}
|
||||
|
||||
rows, err = s.pool.Query(ctx, `
|
||||
SELECT name, inverse, source_type, target_type, cardinality, description
|
||||
FROM relationship_types ORDER BY name`)
|
||||
rtRows, err := q.ListRelationshipTypes(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for rows.Next() {
|
||||
var rt gen.RelationshipType
|
||||
if err := rows.Scan(&rt.Name, &rt.Inverse, &rt.SourceType, &rt.TargetType,
|
||||
&rt.Cardinality, &rt.Description); err != nil {
|
||||
rows.Close()
|
||||
return nil, err
|
||||
}
|
||||
resp.RelationshipTypes = append(resp.RelationshipTypes, rt)
|
||||
}
|
||||
rows.Close()
|
||||
if rows.Err() != nil {
|
||||
return nil, rows.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,
|
||||
})
|
||||
}
|
||||
|
||||
rows, err = s.pool.Query(ctx, `
|
||||
SELECT id, states, default_state, terminal_states, transitions
|
||||
FROM lifecycle_defs ORDER BY id`)
|
||||
lcRows, err := q.ListLifecycleDefs(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for rows.Next() {
|
||||
var lc gen.LifecycleDef
|
||||
var terminal []string
|
||||
var transJSON []byte
|
||||
if err := rows.Scan(&lc.Id, &lc.States, &lc.DefaultState, &terminal, &transJSON); err != nil {
|
||||
rows.Close()
|
||||
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)
|
||||
}
|
||||
lc.TerminalStates = &terminal
|
||||
if err := json.Unmarshal(transJSON, &lc.Transitions); err != nil {
|
||||
rows.Close()
|
||||
return nil, fmt.Errorf("lifecycle %s transitions: %w", lc.Id, err)
|
||||
}
|
||||
resp.Lifecycles = append(resp.Lifecycles, lc)
|
||||
resp.Lifecycles = append(resp.Lifecycles, gen.LifecycleDef{
|
||||
Id: lc.ID,
|
||||
States: lc.States,
|
||||
DefaultState: lc.DefaultState,
|
||||
TerminalStates: &terminal,
|
||||
Transitions: transitions,
|
||||
})
|
||||
}
|
||||
rows.Close()
|
||||
return resp, rows.Err()
|
||||
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// ─── Signals ──────────────────────────────────────────────────────────
|
||||
@@ -1605,10 +1596,9 @@ func checkPrecondition(ctx context.Context, tx pgx.Tx, entityID uuid.UUID, entit
|
||||
}
|
||||
}
|
||||
case "health-check-answering":
|
||||
var health string
|
||||
err := tx.QueryRow(ctx, "SELECT health FROM entity_status WHERE entity_id = $1", entityID).Scan(&health)
|
||||
if err != nil || health == "unknown" || health == "down" {
|
||||
return fmt.Errorf("health check not answering (status: %s)", health)
|
||||
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
|
||||
|
||||
@@ -2177,15 +2177,15 @@ func (s *Server) EndRelationship(ctx context.Context, req gen.EndRelationshipReq
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
|
||||
result, err := tx.Exec(ctx, `
|
||||
UPDATE relationships
|
||||
SET valid_to = now()
|
||||
WHERE source_id = $1 AND target_id = $2 AND type = $3 AND valid_to IS NULL`,
|
||||
sourceID, targetID, req.Params.RelType)
|
||||
result, err := sqlcgen.New(tx).EndCurrentRelationship(ctx, sqlcgen.EndCurrentRelationshipParams{
|
||||
SourceID: sourceID,
|
||||
TargetID: targetID,
|
||||
Type: req.Params.RelType,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if result.RowsAffected() == 0 {
|
||||
if result == 0 {
|
||||
return nil, fmt.Errorf("%w: active relationship %s:%s:%s",
|
||||
domain.ErrNotFound, req.Params.Source, req.Params.RelType, req.Params.Target)
|
||||
}
|
||||
|
||||
@@ -1672,12 +1672,16 @@ func createApproval(ctx context.Context, pool *db.Pool, execID, targetID uuid.UU
|
||||
// a fresh UUID here had no matching entities row, so the INSERT silently
|
||||
// failed, orphaning the execution and never alerting the operator. One
|
||||
// execution maps to at most one approval, so the 1:1 identity holds.
|
||||
if _, err := pool.Exec(ctx, `
|
||||
INSERT INTO approvals (entity_id, subject_entity_id, action, risk_class,
|
||||
kind, payload, status, expires_at, created_at)
|
||||
VALUES ($1, $2, $3, $4, 'execution', $5::jsonb, 'pending',
|
||||
now() + interval '1 hour', now())`,
|
||||
execID, targetID, action, riskClass, string(payload)); err != nil {
|
||||
if err := sqlcgen.New(pool).InsertApproval(ctx, sqlcgen.InsertApprovalParams{
|
||||
EntityID: execID,
|
||||
SubjectEntityID: &targetID,
|
||||
Action: action,
|
||||
RiskClass: riskClass,
|
||||
Kind: "execution",
|
||||
Payload: payload,
|
||||
TokenHash: nil,
|
||||
ExpiresAt: time.Now().Add(time.Hour),
|
||||
}); err != nil {
|
||||
slog.Error("createApproval: insert approval", "error", err, "execution", execID)
|
||||
return
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user