fluid-core/internal/evidence/sqlstore.go

package evidence

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/tegwick/fluid-core/internal/contract"
)

// SQLStore is the relational evidence store.
//
// One schema serves both backends (ADR-0004): SQLite for development, CI and
// single-node deployments, PostgreSQL for anything shared. Only portable SQL is
// used, so the statements CI exercises are the statements production runs.
type SQLStore struct {
	db      *sql.DB
	dialect dialect
}

type dialect int

const (
	dialectSQLite dialect = iota
	dialectPostgres
)

// schemaStatements builds the DDL for a dialect.
//
// The immutability of fluid_events is enforced at the database, not by
// convention in Go. Blueprint invariant 8 (every promotion is auditable) is
// only as strong as the weakest writer, and a trigger binds every one of them.
func schemaStatements(d dialect) []string {
	stmts := []string{
		`CREATE TABLE IF NOT EXISTS fluid_events (
			id           TEXT PRIMARY KEY,
			occurred_at  TEXT NOT NULL,
			entity_type  TEXT NOT NULL,
			entity_id    TEXT NOT NULL,
			event_type   TEXT NOT NULL,
			body         TEXT NOT NULL
		)`,
		`CREATE INDEX IF NOT EXISTS fluid_events_entity
			ON fluid_events (entity_type, entity_id, occurred_at)`,
		`CREATE INDEX IF NOT EXISTS fluid_events_time
			ON fluid_events (occurred_at)`,

		`CREATE TABLE IF NOT EXISTS fluid_telemetry (
			id            TEXT PRIMARY KEY,
			occurred_at   TEXT NOT NULL,
			interface_id  TEXT NOT NULL,
			kind          TEXT NOT NULL,
			revision      TEXT,
			experiment    TEXT,
			cohort        TEXT,
			body          TEXT NOT NULL
		)`,
		`CREATE INDEX IF NOT EXISTS fluid_telemetry_window
			ON fluid_telemetry (interface_id, occurred_at)`,
		`CREATE INDEX IF NOT EXISTS fluid_telemetry_revision
			ON fluid_telemetry (revision, occurred_at)`,

		// Records are derived summary state, rebuildable from events.
		`CREATE TABLE IF NOT EXISTS fluid_records (
			kind       TEXT NOT NULL,
			id         TEXT NOT NULL,
			updated_at TEXT NOT NULL,
			body       TEXT NOT NULL,
			PRIMARY KEY (kind, id)
		)`,
	}

	switch d {
	case dialectSQLite:
		stmts = append(stmts,
			`CREATE TRIGGER IF NOT EXISTS fluid_events_no_update
				BEFORE UPDATE ON fluid_events
				BEGIN SELECT RAISE(ABORT, 'fluid_events is append-only'); END`,
			`CREATE TRIGGER IF NOT EXISTS fluid_events_no_delete
				BEFORE DELETE ON fluid_events
				BEGIN SELECT RAISE(ABORT, 'fluid_events is append-only'); END`,
		)
	case dialectPostgres:
		stmts = append(stmts,
			`CREATE OR REPLACE FUNCTION fluid_events_append_only()
				RETURNS trigger AS $$
				BEGIN RAISE EXCEPTION 'fluid_events is append-only'; END;
				$$ LANGUAGE plpgsql`,
			`DROP TRIGGER IF EXISTS fluid_events_no_change ON fluid_events`,
			`CREATE TRIGGER fluid_events_no_change
				BEFORE UPDATE OR DELETE ON fluid_events
				FOR EACH ROW EXECUTE FUNCTION fluid_events_append_only()`,
		)
	}
	return stmts
}

// OpenSQLite opens (and migrates) a SQLite-backed store.
func OpenSQLite(ctx context.Context, path string) (*SQLStore, error) {
	db, err := sql.Open("sqlite", path)
	if err != nil {
		return nil, fmt.Errorf("open sqlite %q: %w", path, err)
	}
	// SQLite serializes writers; more than one connection buys contention.
	db.SetMaxOpenConns(1)

	for _, pragma := range []string{
		"PRAGMA journal_mode = WAL",
		"PRAGMA foreign_keys = ON",
		"PRAGMA busy_timeout = 5000",
	} {
		if _, err := db.ExecContext(ctx, pragma); err != nil {
			_ = db.Close()
			return nil, fmt.Errorf("%s: %w", pragma, err)
		}
	}

	s := &SQLStore{db: db, dialect: dialectSQLite}
	if err := s.migrate(ctx); err != nil {
		_ = db.Close()
		return nil, err
	}
	return s, nil
}

// NewSQLStore wraps an already-open database, for Postgres or for tests.
func NewSQLStore(ctx context.Context, db *sql.DB, postgres bool) (*SQLStore, error) {
	d := dialectSQLite
	if postgres {
		d = dialectPostgres
	}
	s := &SQLStore{db: db, dialect: d}
	if err := s.migrate(ctx); err != nil {
		return nil, err
	}
	return s, nil
}

func (s *SQLStore) migrate(ctx context.Context) error {
	for _, stmt := range schemaStatements(s.dialect) {
		if _, err := s.db.ExecContext(ctx, stmt); err != nil {
			return fmt.Errorf("migrate: %w\nstatement: %s", err, stmt)
		}
	}
	return nil
}

// arg renders the nth placeholder for the dialect.
func (s *SQLStore) arg(n int) string {
	if s.dialect == dialectPostgres {
		return fmt.Sprintf("$%d", n)
	}
	return "?"
}

// rfc3339 normalizes timestamps so lexical ordering matches chronological
// ordering in both backends.
func rfc3339(t time.Time) string { return t.UTC().Format(time.RFC3339Nano) }

// AppendEvent records a lifecycle transition.
func (s *SQLStore) AppendEvent(ctx context.Context, ev contract.FluidEvent) error {
	if ev.ID == "" {
		return errors.New("event has no id")
	}
	body, err := json.Marshal(ev)
	if err != nil {
		return fmt.Errorf("marshal event %s: %w", ev.ID, err)
	}

	q := fmt.Sprintf(
		`INSERT INTO fluid_events (id, occurred_at, entity_type, entity_id, event_type, body)
		 VALUES (%s, %s, %s, %s, %s, %s)`,
		s.arg(1), s.arg(2), s.arg(3), s.arg(4), s.arg(5), s.arg(6))

	_, err = s.db.ExecContext(ctx, q,
		string(ev.ID), rfc3339(ev.OccurredAt), string(ev.EntityType),
		ev.EntityID, ev.EventType, string(body))
	if err != nil {
		return fmt.Errorf("append event %s: %w", ev.ID, err)
	}
	return nil
}

// Events returns matching events in occurrence order.
func (s *SQLStore) Events(ctx context.Context, f EventFilter) ([]contract.FluidEvent, error) {
	var (
		where []string
		args  []any
	)
	add := func(clause string, v any) {
		args = append(args, v)
		where = append(where, fmt.Sprintf(clause, s.arg(len(args))))
	}

	if f.EntityType != "" {
		add("entity_type = %s", string(f.EntityType))
	}
	if f.EntityID != "" {
		add("entity_id = %s", f.EntityID)
	}
	if f.EventType != "" {
		add("event_type = %s", f.EventType)
	}
	if !f.Since.IsZero() {
		add("occurred_at >= %s", rfc3339(f.Since))
	}
	if !f.Until.IsZero() {
		add("occurred_at <= %s", rfc3339(f.Until))
	}

	q := "SELECT body FROM fluid_events"
	if len(where) > 0 {
		q += " WHERE " + strings.Join(where, " AND ")
	}
	// id breaks ties so that two events in the same instant still order stably.
	q += " ORDER BY occurred_at, id"
	if f.Limit > 0 {
		q += fmt.Sprintf(" LIMIT %d", f.Limit)
	}

	rows, err := s.db.QueryContext(ctx, q, args...)
	if err != nil {
		return nil, fmt.Errorf("query events: %w", err)
	}
	defer rows.Close()

	var out []contract.FluidEvent
	for rows.Next() {
		var body string
		if err := rows.Scan(&body); err != nil {
			return nil, err
		}
		var ev contract.FluidEvent
		if err := json.Unmarshal([]byte(body), &ev); err != nil {
			return nil, fmt.Errorf("decode stored event: %w", err)
		}
		out = append(out, ev)
	}
	return out, rows.Err()
}

// WriteTelemetry records a normalized interaction event.
func (s *SQLStore) WriteTelemetry(ctx context.Context, ev contract.FluidTelemetry) error {
	body, err := json.Marshal(ev)
	if err != nil {
		return fmt.Errorf("marshal telemetry %s: %w", ev.ID, err)
	}

	var revision, experiment, cohort any
	if ev.Revision != nil {
		revision = string(*ev.Revision)
	}
	if ev.Experiment != nil {
		experiment = string(*ev.Experiment)
	}
	if ev.Cohort != nil {
		cohort = string(*ev.Cohort)
	}

	q := fmt.Sprintf(
		`INSERT INTO fluid_telemetry (id, occurred_at, interface_id, kind, revision, experiment, cohort, body)
		 VALUES (%s, %s, %s, %s, %s, %s, %s, %s)`,
		s.arg(1), s.arg(2), s.arg(3), s.arg(4), s.arg(5), s.arg(6), s.arg(7), s.arg(8))

	_, err = s.db.ExecContext(ctx, q,
		ev.ID, rfc3339(ev.OccurredAt), string(ev.InterfaceID), string(ev.Kind),
		revision, experiment, cohort, string(body))
	if err != nil {
		return fmt.Errorf("write telemetry %s: %w", ev.ID, err)
	}
	return nil
}

// Telemetry returns matching telemetry in occurrence order.
func (s *SQLStore) Telemetry(ctx context.Context, f TelemetryFilter) ([]contract.FluidTelemetry, error) {
	var (
		where []string
		args  []any
	)
	add := func(clause string, v any) {
		args = append(args, v)
		where = append(where, fmt.Sprintf(clause, s.arg(len(args))))
	}

	if f.InterfaceID != "" {
		add("interface_id = %s", string(f.InterfaceID))
	}
	if f.Revision != "" {
		add("revision = %s", string(f.Revision))
	}
	if f.Experiment != "" {
		add("experiment = %s", string(f.Experiment))
	}
	if f.Cohort != "" {
		add("cohort = %s", string(f.Cohort))
	}
	if f.Kind != "" {
		add("kind = %s", string(f.Kind))
	}
	if !f.Since.IsZero() {
		add("occurred_at >= %s", rfc3339(f.Since))
	}
	if !f.Until.IsZero() {
		add("occurred_at <= %s", rfc3339(f.Until))
	}

	q := "SELECT body FROM fluid_telemetry"
	if len(where) > 0 {
		q += " WHERE " + strings.Join(where, " AND ")
	}
	q += " ORDER BY occurred_at, id"
	if f.Limit > 0 {
		q += fmt.Sprintf(" LIMIT %d", f.Limit)
	}

	rows, err := s.db.QueryContext(ctx, q, args...)
	if err != nil {
		return nil, fmt.Errorf("query telemetry: %w", err)
	}
	defer rows.Close()

	var out []contract.FluidTelemetry
	for rows.Next() {
		var body string
		if err := rows.Scan(&body); err != nil {
			return nil, err
		}
		var ev contract.FluidTelemetry
		if err := json.Unmarshal([]byte(body), &ev); err != nil {
			return nil, fmt.Errorf("decode stored telemetry: %w", err)
		}
		out = append(out, ev)
	}
	return out, rows.Err()
}

// PutRecord stores or supersedes a derived record.
func (s *SQLStore) PutRecord(ctx context.Context, kind contract.EntityKind, id string, body []byte) error {
	if id == "" {
		return errors.New("record has no id")
	}
	if !json.Valid(body) {
		return fmt.Errorf("record %s/%s: body is not valid JSON", kind, id)
	}

	q := fmt.Sprintf(
		`INSERT INTO fluid_records (kind, id, updated_at, body) VALUES (%s, %s, %s, %s)
		 ON CONFLICT (kind, id) DO UPDATE SET updated_at = excluded.updated_at, body = excluded.body`,
		s.arg(1), s.arg(2), s.arg(3), s.arg(4))

	_, err := s.db.ExecContext(ctx, q, string(kind), id, rfc3339(time.Now()), string(body))
	if err != nil {
		return fmt.Errorf("put record %s/%s: %w", kind, id, err)
	}
	return nil
}

// Record returns one stored record.
func (s *SQLStore) Record(ctx context.Context, kind contract.EntityKind, id string) ([]byte, error) {
	q := fmt.Sprintf(`SELECT body FROM fluid_records WHERE kind = %s AND id = %s`, s.arg(1), s.arg(2))

	var body string
	err := s.db.QueryRowContext(ctx, q, string(kind), id).Scan(&body)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, fmt.Errorf("%w: %s %s", ErrNotFound, kind, id)
	}
	if err != nil {
		return nil, err
	}
	return []byte(body), nil
}

// Records returns every record of a kind.
func (s *SQLStore) Records(ctx context.Context, kind contract.EntityKind) (map[string][]byte, error) {
	q := fmt.Sprintf(`SELECT id, body FROM fluid_records WHERE kind = %s ORDER BY id`, s.arg(1))

	rows, err := s.db.QueryContext(ctx, q, string(kind))
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	out := map[string][]byte{}
	for rows.Next() {
		var id, body string
		if err := rows.Scan(&id, &body); err != nil {
			return nil, err
		}
		out[id] = []byte(body)
	}
	return out, rows.Err()
}

// Close releases the database.
func (s *SQLStore) Close() error { return s.db.Close() }