package db import ( "context" "encoding/json" "time" "github.com/dtoro/oikos/internal/core/domain" "github.com/dtoro/oikos/internal/core/ports" "github.com/google/uuid" "github.com/jackc/pgx/v5" ) // EntityReader implements ports.ReadModels over the postgres pool. type EntityReader struct { pool *Pool } var _ ports.ReadModels = (*EntityReader)(nil) // NewEntityReader builds the read-models service. func NewEntityReader(pool *Pool) *EntityReader { return &EntityReader{pool: pool} } const entityHealthCols = `e.id, e.slug, e.type, e.name, e.state, e.attributes, e.maintenance_until, e.version, e.created_at, e.updated_at, COALESCE(st.health, 'unknown'), st.last_check_at` type rowScanner2 interface{ Scan(dest ...any) error } func scanWithHealth(row rowScanner2) (ports.EntityWithHealth, error) { var id uuid.UUID var slug, typ, name string var state *string var attrs []byte var maint *time.Time var version int32 var createdAt, updatedAt time.Time var health string var lastCheck *time.Time if err := row.Scan(&id, &slug, &typ, &name, &state, &attrs, &maint, &version, &createdAt, &updatedAt, &health, &lastCheck); err != nil { return ports.EntityWithHealth{}, err } e := domain.Entity{ ID: domain.UUID(id.String()), Slug: slug, Type: typ, Name: name, Version: int(version), CreatedAt: createdAt, UpdatedAt: updatedAt, MaintenanceUntil: maint, } if state != nil { e.State = *state } if len(attrs) > 0 { var m map[string]any if json.Unmarshal(attrs, &m) == nil { e.Attributes = m } } return ports.EntityWithHealth{Entity: e, Health: health, LastCheckAt: lastCheck}, nil } func (r *EntityReader) ListEntities(ctx context.Context, f ports.EntityFilters) ([]ports.EntityWithHealth, string, error) { limit := f.Limit if limit <= 0 { limit = 50 } nullable := func(s string) any { if s == "" { return nil } return s } rows, err := r.pool.Query(ctx, ` 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 `+entityHealthCols+` 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`, nullable(f.Type), nullable(f.State), nullable(f.Domain), nullable(f.Layer), nullable(f.Q), nullable(f.Cursor), limit+1) if err != nil { return nil, "", err } defer rows.Close() items := []ports.EntityWithHealth{} for rows.Next() { e, err := scanWithHealth(rows) if err != nil { return nil, "", err } items = append(items, e) } if rows.Err() != nil { return nil, "", rows.Err() } next := "" if len(items) > limit { items = items[:limit] next = items[len(items)-1].Entity.Slug } return items, next, nil } func (r *EntityReader) GetEntity(ctx context.Context, id domain.UUID) (ports.EntityWithHealth, error) { e, err := scanWithHealth(r.pool.QueryRow(ctx, `SELECT `+entityHealthCols+` FROM entities e LEFT JOIN entity_status st ON st.entity_id = e.id WHERE e.id = $1`, mustUUID(id))) if err != nil { return ports.EntityWithHealth{}, mapRowErr(err) } return e, nil } func (r *EntityReader) GetEntityBySlug(ctx context.Context, slug string) (ports.EntityWithHealth, error) { e, err := scanWithHealth(r.pool.QueryRow(ctx, `SELECT `+entityHealthCols+` FROM entities e LEFT JOIN entity_status st ON st.entity_id = e.id WHERE e.slug = $1`, slug)) if err != nil { return ports.EntityWithHealth{}, mapRowErr(err) } return e, nil } func (r *EntityReader) GetEntityRelations(ctx context.Context, entityID domain.UUID, direction, relType string) ([]domain.Relationship, error) { eid := mustUUID(entityID) switch direction { case "outbound": return r.queryRels(ctx, `SELECT source_id, target_id, type, attributes, valid_from, valid_to FROM relationships WHERE valid_to IS NULL AND source_id = $1`, eid, relType) case "inbound": return r.queryRels(ctx, `SELECT source_id, target_id, type, attributes, valid_from, valid_to FROM relationships WHERE valid_to IS NULL AND target_id = $1`, eid, relType) default: return r.queryRels(ctx, `SELECT source_id, target_id, type, attributes, valid_from, valid_to FROM relationships WHERE valid_to IS NULL AND (source_id = $1 OR target_id = $1)`, eid, relType) } } func (r *EntityReader) queryRels(ctx context.Context, query string, eid uuid.UUID, relType string) ([]domain.Relationship, error) { var args []any args = append(args, eid) if relType != "" { query += ` AND type = $2` args = append(args, relType) } else { query += ` ORDER BY type` } rows, err := r.pool.Query(ctx, query, args...) if err != nil { return nil, err } defer rows.Close() return scanEdges(rows) } func (r *EntityReader) GetBlastRadius(ctx context.Context, entityID domain.UUID, depth int) ([]ports.EntityWithHealth, error) { rows, err := r.pool.Query(ctx, ` SELECT `+entityHealthCols+`, 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`, mustUUID(entityID), depth) if err != nil { return nil, err } defer rows.Close() var items []ports.EntityWithHealth for rows.Next() { var d int e, err := scanWithHealth(rows) if err != nil { return nil, err } _ = d items = append(items, e) } return items, rows.Err() } func (r *EntityReader) GetGraph(ctx context.Context, depth int, root *domain.UUID, relTypes []string, cap int) ([]ports.EntityWithHealth, []domain.Relationship, bool, error) { truncated := false var nodes []ports.EntityWithHealth var err error if root != nil { nodes, err = r.blastRadiusNodes(ctx, mustUUID(*root), depth, relTypes) } else { nodes, err = r.topologyNodes(ctx, cap) if err == nil && len(nodes) > cap { nodes = nodes[:cap] truncated = true } } if err != nil { return nil, nil, false, err } ids := make([]uuid.UUID, len(nodes)) for i, n := range nodes { uid, _ := uuid.Parse(string(n.Entity.ID)) ids[i] = uid } edges, err := r.listGraphEdges(ctx, ids, relTypes) if err != nil { return nil, nil, false, err } return nodes, edges, truncated, nil } func (r *EntityReader) blastRadiusNodes(ctx context.Context, rootID uuid.UUID, depth int, relTypes []string) ([]ports.EntityWithHealth, error) { query := ` SELECT ` + entityHealthCols + ` 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` var relTypesArg any if len(relTypes) > 0 { relTypesArg = relTypes } rows, err := r.pool.Query(ctx, query, rootID, depth, relTypesArg) if err != nil { return nil, err } defer rows.Close() var items []ports.EntityWithHealth for rows.Next() { e, err := scanWithHealth(rows) if err != nil { return nil, err } items = append(items, e) } return items, rows.Err() } func (r *EntityReader) topologyNodes(ctx context.Context, cap int) ([]ports.EntityWithHealth, error) { rows, err := r.pool.Query(ctx, ` SELECT `+entityHealthCols+` FROM entities e LEFT JOIN entity_status st ON st.entity_id = e.id WHERE e.type NOT IN ('execution','task') AND e.id IN ( SELECT e2.id FROM entities e2 LEFT JOIN relationships r ON r.valid_to IS NULL AND (r.source_id = e2.id OR r.target_id = e2.id) WHERE e2.type NOT IN ('execution','task') GROUP BY e2.id ORDER BY count(r.type) DESC, e2.slug LIMIT $1 ) ORDER BY e.slug`, cap+1) if err != nil { return nil, err } defer rows.Close() var items []ports.EntityWithHealth for rows.Next() { e, err := scanWithHealth(rows) if err != nil { return nil, err } items = append(items, e) } return items, rows.Err() } func (r *EntityReader) listGraphEdges(ctx context.Context, ids []uuid.UUID, relTypes []string) ([]domain.Relationship, error) { query := `SELECT r.source_id, r.target_id, r.type, r.attributes, r.valid_from, r.valid_to FROM relationships r WHERE r.valid_to IS NULL AND (r.source_id = ANY($1) OR r.target_id = ANY($1))` var args []any args = append(args, ids) if len(relTypes) > 0 { query += ` AND r.type = ANY($2)` args = append(args, relTypes) } else { query += ` ORDER BY r.type` } rows, err := r.pool.Query(ctx, query, args...) if err != nil { return nil, err } defer rows.Close() return scanEdges(rows) } func (r *EntityReader) ListEntityTypes(ctx context.Context) ([]domain.EntityType, error) { rows, err := r.pool.Query(ctx, `SELECT name, COALESCE(parent_type,''), is_abstract, COALESCE(domain,''), COALESCE(layer,''), COALESCE(description,''), COALESCE(lifecycle_id,''), '{}'::jsonb, COALESCE(schema_version,0), COALESCE(status,'') FROM entity_types ORDER BY name`) if err != nil { return nil, err } defer rows.Close() var items []domain.EntityType for rows.Next() { var t domain.EntityType if err := rows.Scan(&t.Name, &t.ParentType, &t.IsAbstract, &t.Domain, &t.Layer, &t.Description, &t.LifecycleID, &t.AttributeSchema, &t.SchemaVersion, &t.Status); err != nil { return nil, err } items = append(items, t) } return items, rows.Err() } func scanEdges(rows pgx.Rows) ([]domain.Relationship, error) { var items []domain.Relationship for rows.Next() { var src, tgt uuid.UUID var rType string var attrs []byte var vf time.Time var vt *time.Time if err := rows.Scan(&src, &tgt, &rType, &attrs, &vf, &vt); err != nil { return nil, err } rel := domain.Relationship{ SourceID: domain.UUID(src.String()), TargetID: domain.UUID(tgt.String()), Type: rType, Attributes: map[string]any{}, ValidFrom: vf, ValidTo: vt, } if len(attrs) > 0 { var m map[string]any if json.Unmarshal(attrs, &m) == nil { rel.Attributes = m } } items = append(items, rel) } return items, rows.Err() }