oikos/internal/adapters/postgres/repositories.go

package db

import (
	"context"
	"fmt"
	"encoding/json"
	"errors"
	"strings"
	"sync"
	"time"

	"github.com/dtoro/oikos/internal/core/app"
	"github.com/dtoro/oikos/internal/core/domain"
	"github.com/dtoro/oikos/internal/core/ports"
	"github.com/dtoro/oikos/internal/adapters/postgres/sqlcgen"
	"github.com/dtoro/oikos/internal/observability"
	"github.com/dtoro/oikos/internal/ontology"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
)

// EntityRepo implements ports.EntityRepository over the postgres pool.
// Command methods run everything in their input inside one transaction
// (ADR 0016 §3.6): entity write, derived checks, idempotency record, audit,
// event.
type EntityRepo struct {
	pool *Pool
}

var _ ports.EntityRepository = (*EntityRepo)(nil)

// NewEntityRepo builds the entity repository.
func NewEntityRepo(pool *Pool) *EntityRepo { return &EntityRepo{pool: pool} }


// mustUUID converts a domain.UUID (string alias) to uuid.UUID. The domain
// layer guarantees UUID-shaped strings; parse failures are programming
// errors and panic loudly rather than half-failing a transaction.
func mustUUID(id domain.UUID) uuid.UUID {
	u, err := uuid.Parse(string(id))
	if err != nil {
		panic(fmt.Sprintf("invalid entity UUID %q", string(id)))
	}
	return u
}

const entityFullCols = `id, slug, type, name, state, attributes, maintenance_until, version, created_at, updated_at`

type rowScanner interface{ Scan(dest ...any) error }

func scanDomainEntity(row rowScanner) (domain.Entity, 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
	if err := row.Scan(&id, &slug, &typ, &name, &state, &attrs, &maint, &version, &createdAt, &updatedAt); err != nil {
		return domain.Entity{}, err
	}
	d := 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 {
		d.State = *state
	}
	if len(attrs) > 0 {
		var m map[string]any
		if json.Unmarshal(attrs, &m) == nil {
			d.Attributes = m
		}
	}
	return d, nil
}

func mapRowErr(err error) error {
	if errors.Is(err, pgx.ErrNoRows) {
		return domain.ErrNotFound
	}
	return err
}

// Get returns the entity by ID.
func (r *EntityRepo) Get(ctx context.Context, id domain.UUID) (domain.Entity, error) {
	uid, err := uuid.Parse(string(id))
	if err != nil {
		return domain.Entity{}, err
	}
	e, err := scanDomainEntity(r.pool.QueryRow(ctx,
		`SELECT `+entityFullCols+` FROM entities WHERE id = $1`, uid))
	return e, mapRowErr(err)
}

// BySlug returns the entity by slug.
func (r *EntityRepo) BySlug(ctx context.Context, slug string) (domain.Entity, error) {
	e, err := scanDomainEntity(r.pool.QueryRow(ctx,
		`SELECT `+entityFullCols+` FROM entities WHERE slug = $1`, slug))
	return e, mapRowErr(err)
}

// List returns entities filtered by type (including descendant types),
// state, domain, layer, and a slug/name substring, keyset-paginated by
// slug. The second return is the next cursor ("" when exhausted).
func (r *EntityRepo) List(ctx context.Context, f ports.EntityFilters) ([]domain.Entity, string, error) {
	limit := f.Limit
	if limit <= 0 {
		limit = 50
	}
	// NULL-able filter args: an absent filter must bind SQL NULL, not "".
	nullable := func(s string) any {
		if s == "" {
			return nil
		}
		return s
	}

	// Type filter includes descendants via the parent hierarchy (R3-1).
	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 `+entityFullCols+`
		FROM entities e
		JOIN entity_types et ON et.name = e.type
		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 := []domain.Entity{}
	for rows.Next() {
		e, err := scanDomainEntity(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].Slug
	}
	return items, next, nil
}

// Search matches slug/name substrings.
func (r *EntityRepo) Search(ctx context.Context, q string, limit int) ([]domain.Entity, error) {
	items, _, err := r.List(ctx, ports.EntityFilters{Q: q, Limit: limit})
	return items, err
}

// writeSideEffects writes audit entries and the event inside the open
// transaction.
func writeSideEffects(ctx context.Context, tx pgx.Tx, entityID domain.UUID, audits []ports.AuditEntry, event *ports.Event) error {
	q := sqlcgen.New(tx)
	eid := mustUUID(entityID)
	for _, a := range audits {
		if err := observability.Audit(ctx, q, a.ActorType, a.ActorLabel, a.Action, &eid,
			a.Method, a.Path, a.CorrelationID, nil, a.Details); err != nil {
			return err
		}
	}
	if event != nil {
		if err := observability.Event(ctx, q, event.Type, &eid, event.Severity, event.Source,
			event.CorrelationID, event.Data); err != nil {
			return err
		}
	}
	return nil
}

func appTargetOf(e domain.Entity) app.CheckTarget {
	attrs, _ := json.Marshal(e.Attributes)
	return app.CheckTarget{ID: string(e.ID), Slug: e.Slug, Type: e.Type, Name: e.Name, Attrs: attrs}
}

func writeDerivedChecks(ctx context.Context, tx pgx.Tx, e domain.Entity, checks []ports.DerivedCheck) (int, error) {
	if _, err := tx.Exec(ctx,
		`INSERT INTO entity_status (entity_id, health, updated_at)
		 VALUES ($1, 'unknown', now())
		 ON CONFLICT (entity_id) DO NOTHING`, mustUUID(e.ID)); err != nil {
		return 0, err
	}
	created := 0
	t := appTargetOf(e)
	for i, dc := range checks {
		ok, err := writeCheck(ctx, tx, t, i, app.CheckDef{Kind: dc.Kind, Config: dc.Config, IntervalS: dc.IntervalS})
		if err != nil {
			return created, err
		}
		if ok {
			created++
		}
	}
	return created, nil
}

// rederiveChecks re-runs derivation for an entity inside the open tx, with
// the graph host fallback active (a service inherits its container's
// address once the hosting edge exists).
func rederiveChecks(ctx context.Context, tx pgx.Tx, e domain.Entity, tree *ontology.TypeTree) (int, error) {
	t := appTargetOf(e)
	defs, _ := app.Derive(tree, t, func() map[string]any {
		attrs, err := hostViaGraph(ctx, tx, t.ID)
		if err != nil {
			return nil
		}
		return attrs
	})
	derived := make([]ports.DerivedCheck, len(defs))
	for i, d := range defs {
		derived[i] = ports.DerivedCheck{Kind: d.Kind, Config: d.Config, IntervalS: d.IntervalS}
	}
	return writeDerivedChecks(ctx, tx, e, derived)
}

// Create inserts the entity with its derived checks, idempotency record,
// audit, and event — one transaction.
func (r *EntityRepo) Create(ctx context.Context, in ports.EntityCreateInput) (domain.Entity, error) {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return domain.Entity{}, err
	}
	defer func() { _ = tx.Rollback(ctx) }()

	e := in.Entity
	attrsJSON, _ := json.Marshal(e.Attributes)
	if e.Attributes == nil {
		attrsJSON = []byte("{}")
	}
	var state *string
	if e.State != "" {
		state = &e.State
	}

	created, err := scanDomainEntity(tx.QueryRow(ctx, `
		INSERT INTO entities (id, slug, type, name, state, attributes)
		VALUES ($1, $2, $3, $4, $5, $6)
		RETURNING `+entityFullCols,
		mustUUID(e.ID), e.Slug, e.Type, e.Name, state, attrsJSON))
	if err != nil {
		if strings.Contains(err.Error(), "unique") || strings.Contains(err.Error(), "duplicate") {
			return domain.Entity{}, errors.Join(domain.ErrAlreadyExists, err)
		}
		return domain.Entity{}, err
	}

	if in.Idempotency != nil {
		var body []byte
		if in.Idempotency.RenderBody != nil {
			body = in.Idempotency.RenderBody(created)
		}
		code := int32(201)
		if err := sqlcgen.New(tx).PutIdempotentResponse(ctx, sqlcgen.PutIdempotentResponseParams{
			Actor: in.Idempotency.Actor, Key: in.Idempotency.Key,
			RequestHash: in.Idempotency.RequestHash, ResponseCode: &code, ResponseBody: body,
		}); err != nil {
			return domain.Entity{}, err
		}
	}

	if _, err := writeDerivedChecks(ctx, tx, created, in.DerivedChecks); err != nil {
		return domain.Entity{}, err
	}
	if err := writeSideEffects(ctx, tx, created.ID, in.Audit, in.Event); err != nil {
		return domain.Entity{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return domain.Entity{}, err
	}
	return created, nil
}

// Update applies name/state/attributes/maintenance atomically with an
// optimistic-version check, re-deriving default checks when asked. The
// declared-transition + precondition validation runs inside the
// transaction (check-then-act, plan §3.6).
func (r *EntityRepo) Update(ctx context.Context, in ports.EntityUpdateInput) (domain.Entity, error) {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return domain.Entity{}, err
	}
	defer func() { _ = tx.Rollback(ctx) }()

	current, err := scanDomainEntity(tx.QueryRow(ctx,
		`SELECT `+entityFullCols+` FROM entities WHERE id = $1`, mustUUID(in.Entity.ID)))
	if err != nil {
		return domain.Entity{}, mapRowErr(err)
	}

	if in.ExpectedVersion > 0 && current.Version != in.ExpectedVersion {
		return domain.Entity{}, domain.ErrConflict
	}

	e := in.Entity
	// Fields the caller left zero keep their current values.
	if e.Name == "" {
		e.Name = current.Name
	}
	if e.Slug == "" {
		e.Slug = current.Slug
	}
	if e.Type == "" {
		e.Type = current.Type
	}

	// Lifecycle validation when state changes (declared transition +
	// preconditions, in-tx).
	if e.State != "" && e.State != current.State {
		if err := ValidateTransition(ctx, tx, mustUUID(current.ID), current.Type, current.State, e.State); err != nil {
			if errors.Is(err, ErrTransitionInvalid) {
				return domain.Entity{}, errors.Join(domain.ErrInvalidTransition, err)
			}
			return domain.Entity{}, err
		}
	} else if e.State == "" {
		e.State = current.State
	}

	if e.Attributes == nil {
		e.Attributes = current.Attributes
	}
	setMaintenance := e.MaintenanceUntil != nil
	maint := e.MaintenanceUntil
	if !setMaintenance {
		maint = current.MaintenanceUntil
	}

	attrsJSON, _ := json.Marshal(e.Attributes)
	if e.Attributes == nil {
		attrsJSON = []byte("{}")
	}
	var state *string
	if e.State != "" {
		state = &e.State
	}

	updated, err := scanDomainEntity(tx.QueryRow(ctx, `
		UPDATE entities
		SET name = $2, state = $3, attributes = $4, maintenance_until = $5, version = version + 1, updated_at = now()
		WHERE id = $1 AND version = $6
		RETURNING `+entityFullCols,
		mustUUID(e.ID), e.Name, state, attrsJSON, maint, current.Version))
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return domain.Entity{}, domain.ErrConflict
		}
		return domain.Entity{}, err
	}

	if in.RederiveChecks {
		tree, err := LoadTypeTree(ctx, tx)
		if err != nil {
			return domain.Entity{}, err
		}
		if _, err := rederiveChecks(ctx, tx, updated, tree); err != nil {
			return domain.Entity{}, err
		}
	}

	if err := writeSideEffects(ctx, tx, updated.ID, in.Audit, in.Event); err != nil {
		return domain.Entity{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return domain.Entity{}, err
	}
	return updated, nil
}

// SetState transitions an entity's lifecycle state. Declared-transition
// and precondition validation run in-tx; a stale From is refused.
func (r *EntityRepo) SetState(ctx context.Context, in ports.EntityTransitionInput) (domain.Entity, error) {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return domain.Entity{}, err
	}
	defer func() { _ = tx.Rollback(ctx) }()

	current, err := scanDomainEntity(tx.QueryRow(ctx,
		`SELECT `+entityFullCols+` FROM entities WHERE slug = $1`, in.Slug))
	if err != nil {
		return domain.Entity{}, mapRowErr(err)
	}
	if current.State != in.From {
		return domain.Entity{}, domain.ErrConflict
	}
	if err := ValidateTransition(ctx, tx, mustUUID(current.ID), current.Type, in.From, in.To); err != nil {
		if errors.Is(err, ErrTransitionInvalid) {
			return domain.Entity{}, errors.Join(domain.ErrInvalidTransition, err)
		}
		return domain.Entity{}, err
	}
	if _, err := tx.Exec(ctx,
		`UPDATE entities SET state = $2, version = version + 1, updated_at = now() WHERE id = $1`,
		mustUUID(current.ID), in.To); err != nil {
		return domain.Entity{}, err
	}

	after := current
	after.State = in.To
	if err := writeSideEffects(ctx, tx, after.ID, in.Audit, in.Event); err != nil {
		return domain.Entity{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return domain.Entity{}, err
	}
	return after, nil
}

// GetIdempotent returns the cached response for (actor, key) or
// domain.ErrNotFound.
func (r *EntityRepo) GetIdempotent(ctx context.Context, actor, key string) (ports.IdempotentResponse, error) {
	cached, err := sqlcgen.New(r.pool).GetIdempotentResponse(ctx, sqlcgen.GetIdempotentResponseParams{Actor: actor, Key: key})
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return ports.IdempotentResponse{}, domain.ErrNotFound
		}
		return ports.IdempotentResponse{}, err
	}
	code := 0
	if cached.ResponseCode != nil {
		code = int(*cached.ResponseCode)
	}
	return ports.IdempotentResponse{
		RequestHash:   cached.RequestHash,
		ResponseCode:  code,
		ResponseBody:  cached.ResponseBody,
	}, nil
}

// RelRepo implements ports.RelationshipRepository over the postgres pool.
type RelRepo struct {
	pool *Pool
}

var _ ports.RelationshipRepository = (*RelRepo)(nil)

func NewRelRepo(pool *Pool) *RelRepo { return &RelRepo{pool: pool} }

func (r *RelRepo) Create(ctx context.Context, input ports.RelationshipCreateInput) (domain.Relationship, error) {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return domain.Relationship{}, err
	}
	defer func() { _ = tx.Rollback(ctx) }()

	attrsJSON := []byte("{}")
	if len(input.Relationship.Attributes) > 0 {
		attrsJSON, _ = json.Marshal(input.Relationship.Attributes)
	}

	_, err = tx.Exec(ctx, `
		INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
		VALUES ($1, $2, $3, $4, now())`,
		mustUUID(input.Relationship.SourceID), mustUUID(input.Relationship.TargetID),
		input.Relationship.Type, attrsJSON)
	if err != nil {
		if strings.Contains(err.Error(), "unique") || strings.Contains(err.Error(), "duplicate") {
			return domain.Relationship{}, errors.Join(domain.ErrAlreadyExists, err)
		}
		return domain.Relationship{}, err
	}

	if err := writeSideEffects(ctx, tx, input.Relationship.SourceID, input.Audit, input.Event); err != nil {
		return domain.Relationship{}, err
	}
	if err := tx.Commit(ctx); err != nil {
		return domain.Relationship{}, err
	}
	input.Relationship.ValidFrom = time.Now()
	return input.Relationship, nil
}

func (r *RelRepo) End(ctx context.Context, source, target domain.UUID, relType string) error {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(ctx) }()

	result, err := sqlcgen.New(tx).EndCurrentRelationship(ctx, sqlcgen.EndCurrentRelationshipParams{
		SourceID: mustUUID(source), TargetID: mustUUID(target), Type: relType,
	})
	if err != nil {
		return err
	}
	if result == 0 {
		return domain.ErrNotFound
	}
	return tx.Commit(ctx)
}

func (r *RelRepo) ListFor(ctx context.Context, entityID domain.UUID, direction string) ([]domain.Relationship, error) {
	eid := mustUUID(entityID)
	switch direction {
	case "outbound":
		return queryRelsBySource(r.pool, eid)
	case "inbound":
		return queryRelsByTarget(r.pool, eid)
	default:
		return queryRelsBoth(r.pool, eid)
	}
}

func queryRelsBySource(pool *Pool, eid uuid.UUID) ([]domain.Relationship, error) {
	rows, err := pool.Query(context.Background(),
		`SELECT source_id, target_id, type, attributes, valid_from, valid_to
		 FROM relationships WHERE valid_to IS NULL AND source_id = $1 ORDER BY type`, eid)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	return scanEdges(rows)
}

func queryRelsByTarget(pool *Pool, eid uuid.UUID) ([]domain.Relationship, error) {
	rows, err := pool.Query(context.Background(),
		`SELECT source_id, target_id, type, attributes, valid_from, valid_to
		 FROM relationships WHERE valid_to IS NULL AND target_id = $1 ORDER BY type`, eid)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	return scanEdges(rows)
}

func queryRelsBoth(pool *Pool, eid uuid.UUID) ([]domain.Relationship, error) {
	rows, err := pool.Query(context.Background(),
		`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) ORDER BY type`, eid)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	return scanEdges(rows)
}

// OntologyRepo implements ports.OntologyStore with a TTL cache — entity
// types change at seed time, not per request, so a short cache trades a
// little staleness for avoiding the meta-schema load on every mutation.
type OntologyRepo struct {
	pool *Pool
	ttl  time.Duration

	mu     sync.Mutex
	loaded time.Time
	tree   *ontology.TypeTree
}

var _ ports.OntologyStore = (*OntologyRepo)(nil)

// NewOntologyRepo builds the ontology store with the given cache TTL
// (values <= 0 disable caching).
func NewOntologyRepo(pool *Pool, ttl time.Duration) *OntologyRepo {
	return &OntologyRepo{pool: pool, ttl: ttl}
}

// LoadTypeTree returns the (possibly cached) ontology tree.
func (o *OntologyRepo) LoadTypeTree(ctx context.Context) (ports.TypeTree, error) {
	if o.ttl <= 0 {
		return o.load(ctx)
	}
	o.mu.Lock()
	defer o.mu.Unlock()
	if o.tree != nil && time.Since(o.loaded) < o.ttl {
		return o.tree, nil
	}
	tree, err := o.load(ctx)
	if err != nil {
		return nil, err
	}
	o.tree = tree
	o.loaded = time.Now()
	return tree, nil
}

func (o *OntologyRepo) load(ctx context.Context) (ports.TypeTree, error) {
	tx, err := o.pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	tree, err := LoadTypeTree(ctx, tx)
	if err != nil {
		return nil, err
	}
	return ports.TypeTree(tree), nil
}