oikos/internal/adapters/postgres/readmodels.go

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()
}