kami
9e6b995538
Finding 5 and the "remove" list of REVIEW-2026-07-30.md.
The tests were prose, not verification. TestContract_ChangesSinceAlwaysZero
documented a bug commit 74b19e0 had already fixed and could only t.Logf.
TestContract_TimeoutGoroutineNotCancelled was a bare t.Log, and is now
falsified by the preceding commit anyway.
TestContract_DestructiveCapabilityDisabledByDefault re-implemented
`risk != "destructive"` inside the test and asserted on its own local
variable — it would have passed if the handler were deleted. The real
derivation is now domain.EnabledForRisk and is tested against production code.
Also removed: the systemd provider (no systemctl in the distroless image and
nothing ever registered it), Server.ListenUnix, Registry.List,
DiscoveredTools, Capability.IsDestructive, the ExecutionDenied status,
EventCapabilityUnavailable, ResolveRequest/ResolveResult (superseded by
nexusclient), Engine.emitEvent, and eight unused Err values.
Three items on the review's list were kept, having turned out to be wrong:
ifString is still used by BuildCapabilities; EventExecutionDenied is asserted
on by a real test; and schema_migrations was NOT dropped. migrate() applies
migrations by slice index and writes user_version = index + 1, so removing an
element renumbers every later migration and any database past that point would
permanently skip one it had not yet applied. The live database is far behind
HEAD, so that is a data hazard rather than a cleanup. The CREATE TABLE is now
a no-op holding its slot, with a comment saying why the slot must stay.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Uea55zaiWuEByEDC4UBSdd
595 lines
20 KiB
Go
595 lines
20 KiB
Go
package storage
|
|
|
|
import (
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/kami/hexis/internal/domain"
|
|
_ "modernc.org/sqlite"
|
|
)
|
|
|
|
type Store struct {
|
|
mu sync.RWMutex
|
|
db *sql.DB
|
|
path string
|
|
}
|
|
|
|
func Open(path string) (*Store, error) {
|
|
dir := filepath.Dir(path)
|
|
if err := os.MkdirAll(dir, 0755); err != nil {
|
|
return nil, fmt.Errorf("create directory: %w", err)
|
|
}
|
|
|
|
db, err := sql.Open("sqlite", path+"?_pragma=journal_mode(WAL)&_pragma=foreign_keys(1)")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("open database: %w", err)
|
|
}
|
|
|
|
db.SetMaxOpenConns(1)
|
|
|
|
store := &Store{db: db, path: path}
|
|
if err := store.migrate(); err != nil {
|
|
return nil, fmt.Errorf("migrate: %w", err)
|
|
}
|
|
return store, nil
|
|
}
|
|
|
|
func (s *Store) Close() error {
|
|
return s.db.Close()
|
|
}
|
|
|
|
func (s *Store) DB() *sql.DB {
|
|
return s.db
|
|
}
|
|
|
|
func (s *Store) migrate() error {
|
|
tx, err := s.db.Begin()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
var v int
|
|
err = tx.QueryRow("PRAGMA user_version").Scan(&v)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if v < len(migrations) {
|
|
for i, m := range migrations[v:] {
|
|
if _, err := tx.Exec(m); err != nil {
|
|
return fmt.Errorf("migration %d: %w", v+i+1, err)
|
|
}
|
|
if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version = %d", v+i+1)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
var migrations = []string{
|
|
`CREATE TABLE IF NOT EXISTS capabilities (
|
|
id TEXT PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
description TEXT NOT NULL DEFAULT '',
|
|
target_types TEXT NOT NULL DEFAULT '[]',
|
|
target_entity_id TEXT NOT NULL DEFAULT '',
|
|
provider TEXT NOT NULL,
|
|
operation TEXT NOT NULL,
|
|
risk TEXT NOT NULL DEFAULT '',
|
|
read_only INTEGER NOT NULL DEFAULT 1,
|
|
expected_side_effects TEXT NOT NULL DEFAULT '',
|
|
attributes TEXT NOT NULL DEFAULT '{}',
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
version INTEGER NOT NULL DEFAULT 0
|
|
)`,
|
|
`CREATE TABLE IF NOT EXISTS executions (
|
|
id TEXT PRIMARY KEY,
|
|
capability_id TEXT NOT NULL,
|
|
target_entity_id TEXT NOT NULL,
|
|
entity_version INTEGER NOT NULL DEFAULT 0,
|
|
arguments TEXT NOT NULL DEFAULT '{}',
|
|
requested_by TEXT NOT NULL DEFAULT '{}',
|
|
origin TEXT NOT NULL DEFAULT '{}',
|
|
idempotency_key TEXT UNIQUE,
|
|
status TEXT NOT NULL DEFAULT 'started',
|
|
result TEXT NOT NULL DEFAULT '{}',
|
|
error TEXT NOT NULL DEFAULT '',
|
|
resolution_evidence TEXT NOT NULL DEFAULT '[]',
|
|
correlation_id TEXT NOT NULL DEFAULT '',
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
)`,
|
|
`CREATE TABLE IF NOT EXISTS hexis_events (
|
|
id TEXT PRIMARY KEY,
|
|
sequence INTEGER NOT NULL,
|
|
type TEXT NOT NULL,
|
|
timestamp TEXT NOT NULL,
|
|
actor TEXT NOT NULL DEFAULT '',
|
|
correlation_id TEXT NOT NULL DEFAULT '',
|
|
causation_id TEXT NOT NULL DEFAULT '',
|
|
payload TEXT NOT NULL DEFAULT '{}'
|
|
)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_hevents_sequence ON hexis_events(sequence)`,
|
|
// Retired: schema_migrations was never read — PRAGMA user_version is the
|
|
// real mechanism. Migrations are applied by slice index, so this slot must
|
|
// keep its position or every later migration would be renumbered and
|
|
// skipped on already-migrated databases. Left as a no-op instead.
|
|
`SELECT 1`,
|
|
`ALTER TABLE capabilities ADD COLUMN requires_confirmation INTEGER NOT NULL DEFAULT 0`,
|
|
`ALTER TABLE capabilities ADD COLUMN enabled INTEGER NOT NULL DEFAULT 1`,
|
|
`ALTER TABLE capabilities ADD COLUMN timeout_seconds INTEGER NOT NULL DEFAULT 30`,
|
|
`ALTER TABLE executions ADD COLUMN confirmation_id TEXT NOT NULL DEFAULT ''`,
|
|
`CREATE TABLE IF NOT EXISTS confirmations (
|
|
id TEXT PRIMARY KEY,
|
|
capability_id TEXT NOT NULL,
|
|
capability_version INTEGER NOT NULL,
|
|
target_entity_id TEXT NOT NULL,
|
|
args_normalized TEXT NOT NULL DEFAULT '{}',
|
|
args_hash TEXT NOT NULL,
|
|
requester TEXT NOT NULL DEFAULT '',
|
|
created_at TEXT NOT NULL,
|
|
expires_at TEXT NOT NULL,
|
|
state TEXT NOT NULL DEFAULT 'pending'
|
|
)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_executions_inflight ON executions(capability_id, target_entity_id, status)`,
|
|
// Audit columns missing from executions (spec §4.1) plus a partial UNIQUE
|
|
// index enforcing one in-flight execution per (capability, target).
|
|
// Pre-existing duplicate in-flight rows would make the index creation fail,
|
|
// so they are resolved to 'unknown' first — that is the spec's outcome for
|
|
// an execution whose side effect can no longer be determined.
|
|
`ALTER TABLE executions ADD COLUMN causation_id TEXT NOT NULL DEFAULT '';
|
|
ALTER TABLE executions ADD COLUMN capability_version INTEGER NOT NULL DEFAULT 0;
|
|
UPDATE executions SET status = 'unknown'
|
|
WHERE status = 'started'
|
|
AND id NOT IN (
|
|
SELECT MIN(id) FROM executions WHERE status = 'started'
|
|
GROUP BY capability_id, target_entity_id
|
|
);
|
|
DROP INDEX IF EXISTS idx_executions_inflight;
|
|
CREATE UNIQUE INDEX IF NOT EXISTS idx_executions_inflight
|
|
ON executions(capability_id, target_entity_id) WHERE status = 'started';`,
|
|
// Supports the entity_id filter on GET /api/v1/executions (spec §4.5). The
|
|
// `since` cursor is the row's implicit rowid, which cannot appear in an
|
|
// index (SQLite indexes it as the btree key already), so ordering and
|
|
// range-scanning by rowid needs no index of its own.
|
|
`CREATE INDEX IF NOT EXISTS idx_executions_entity ON executions(target_entity_id)`,
|
|
}
|
|
|
|
const timeFmt = "2006-01-02T15:04:05.999999999Z07:00"
|
|
|
|
func formatTime(t time.Time) string {
|
|
return t.UTC().Format(timeFmt)
|
|
}
|
|
|
|
func parseTime(s string) time.Time {
|
|
t, err := time.Parse(timeFmt, s)
|
|
if err != nil {
|
|
t, err = time.Parse(time.RFC3339, s)
|
|
if err != nil {
|
|
return time.Time{}
|
|
}
|
|
}
|
|
return t
|
|
}
|
|
|
|
// Capability operations
|
|
|
|
func (s *Store) CreateCapability(c *domain.Capability) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
targetTypes, _ := json.Marshal(c.TargetTypes)
|
|
attrs, _ := json.Marshal(c.Attributes)
|
|
|
|
_, err := s.db.Exec(
|
|
`INSERT INTO capabilities (id, name, description, target_types, target_entity_id, provider, operation, risk, read_only, expected_side_effects, requires_confirmation, enabled, timeout_seconds, attributes, created_at, updated_at, version)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
|
c.ID, c.Name, c.Description, string(targetTypes), c.TargetEntityID, c.Provider, c.Operation, c.Risk, boolInt(c.ReadOnly), c.ExpectedSideEffects, boolInt(c.RequiresConfirmation), boolInt(c.Enabled), c.TimeoutSeconds, string(attrs), formatTime(c.CreatedAt), formatTime(c.UpdatedAt), c.Version,
|
|
)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) GetCapability(id string) (*domain.Capability, error) {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
|
|
row := s.db.QueryRow(
|
|
`SELECT id, name, description, target_types, COALESCE(target_entity_id,''), provider, operation, risk, read_only, COALESCE(expected_side_effects,''), requires_confirmation, enabled, timeout_seconds, attributes, created_at, updated_at, version
|
|
FROM capabilities WHERE id = ?`, id,
|
|
)
|
|
c := &domain.Capability{}
|
|
var targetTypes, attrs, createdAt, updatedAt string
|
|
err := row.Scan(&c.ID, &c.Name, &c.Description, &targetTypes, &c.TargetEntityID, &c.Provider, &c.Operation, &c.Risk, &c.ReadOnly, &c.ExpectedSideEffects, &c.RequiresConfirmation, &c.Enabled, &c.TimeoutSeconds, &attrs, &createdAt, &updatedAt, &c.Version)
|
|
if err == sql.ErrNoRows {
|
|
return nil, domain.ErrCapabilityNotFound
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
json.Unmarshal([]byte(targetTypes), &c.TargetTypes)
|
|
json.Unmarshal([]byte(attrs), &c.Attributes)
|
|
c.CreatedAt = parseTime(createdAt)
|
|
c.UpdatedAt = parseTime(updatedAt)
|
|
if c.TargetTypes == nil {
|
|
c.TargetTypes = []string{}
|
|
}
|
|
if c.Attributes == nil {
|
|
c.Attributes = map[string]any{}
|
|
}
|
|
return c, nil
|
|
}
|
|
|
|
func (s *Store) UpdateCapability(c *domain.Capability) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
targetTypes, _ := json.Marshal(c.TargetTypes)
|
|
attrs, _ := json.Marshal(c.Attributes)
|
|
|
|
res, err := s.db.Exec(
|
|
`UPDATE capabilities SET name=?, description=?, target_types=?, target_entity_id=?, provider=?, operation=?, risk=?, read_only=?, expected_side_effects=?, requires_confirmation=?, enabled=?, timeout_seconds=?, attributes=?, updated_at=?, version=version+1
|
|
WHERE id=? AND version=?`,
|
|
c.Name, c.Description, string(targetTypes), c.TargetEntityID, c.Provider, c.Operation, c.Risk, boolInt(c.ReadOnly), c.ExpectedSideEffects, boolInt(c.RequiresConfirmation), boolInt(c.Enabled), c.TimeoutSeconds, string(attrs), formatTime(c.UpdatedAt), c.ID, c.Version,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
if n == 0 {
|
|
return domain.ErrConflict
|
|
}
|
|
c.Version++
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) ListCapabilities(entityID string) ([]*domain.Capability, error) {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
|
|
query := `SELECT id, name, description, target_types, COALESCE(target_entity_id,''), provider, operation, risk, read_only, COALESCE(expected_side_effects,''), requires_confirmation, enabled, timeout_seconds, attributes, created_at, updated_at, version
|
|
FROM capabilities`
|
|
args := []any{}
|
|
if entityID != "" {
|
|
query += " WHERE target_entity_id = ?"
|
|
args = append(args, entityID)
|
|
}
|
|
query += " ORDER BY name ASC"
|
|
|
|
rows, err := s.db.Query(query, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var result []*domain.Capability
|
|
for rows.Next() {
|
|
c := &domain.Capability{}
|
|
var targetTypes, attrs, createdAt, updatedAt string
|
|
if err := rows.Scan(&c.ID, &c.Name, &c.Description, &targetTypes, &c.TargetEntityID, &c.Provider, &c.Operation, &c.Risk, &c.ReadOnly, &c.ExpectedSideEffects, &c.RequiresConfirmation, &c.Enabled, &c.TimeoutSeconds, &attrs, &createdAt, &updatedAt, &c.Version); err != nil {
|
|
return nil, err
|
|
}
|
|
json.Unmarshal([]byte(targetTypes), &c.TargetTypes)
|
|
json.Unmarshal([]byte(attrs), &c.Attributes)
|
|
c.CreatedAt = parseTime(createdAt)
|
|
c.UpdatedAt = parseTime(updatedAt)
|
|
if c.TargetTypes == nil {
|
|
c.TargetTypes = []string{}
|
|
}
|
|
if c.Attributes == nil {
|
|
c.Attributes = map[string]any{}
|
|
}
|
|
result = append(result, c)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Store) DeleteCapability(id string) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
_, err := s.db.Exec(`DELETE FROM capabilities WHERE id = ?`, id)
|
|
return err
|
|
}
|
|
|
|
// Execution operations
|
|
|
|
func (s *Store) CreateExecution(e *domain.Execution) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
args, _ := json.Marshal(e.Arguments)
|
|
reqBy, _ := json.Marshal(e.RequestedBy)
|
|
origin, _ := json.Marshal(e.Origin)
|
|
result, _ := json.Marshal(e.Result)
|
|
evidence, _ := json.Marshal(e.ResolutionEvidence)
|
|
|
|
_, err := s.db.Exec(
|
|
`INSERT INTO executions (id, capability_id, capability_version, target_entity_id, entity_version, arguments, requested_by, origin, idempotency_key, confirmation_id, status, result, error, resolution_evidence, correlation_id, causation_id, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
|
e.ID, e.CapabilityID, e.CapabilityVersion, e.TargetEntityID, e.EntityVersion, string(args), string(reqBy), string(origin), nullString(e.IdempotencyKey), e.ConfirmationID, string(e.Status), string(result), e.Error, string(evidence), e.CorrelationID, e.CausationID, formatTime(e.CreatedAt), formatTime(e.UpdatedAt),
|
|
)
|
|
if err != nil {
|
|
if strings.Contains(err.Error(), "UNIQUE") {
|
|
// The partial unique index on (capability_id, target_entity_id)
|
|
// WHERE status='started' is the in-flight guard, not a replay.
|
|
if strings.Contains(err.Error(), "idempotency_key") {
|
|
return domain.ErrIdempotencyReplay
|
|
}
|
|
if strings.Contains(err.Error(), "capability_id") || strings.Contains(err.Error(), "target_entity_id") {
|
|
return domain.ErrExecutionInFlight
|
|
}
|
|
return domain.ErrIdempotencyReplay
|
|
}
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) GetExecution(id string) (*domain.Execution, error) {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
|
|
row := s.db.QueryRow(
|
|
`SELECT id, capability_id, capability_version, target_entity_id, entity_version, arguments, requested_by, origin, COALESCE(idempotency_key,''), COALESCE(confirmation_id,''), status, result, COALESCE(error,''), resolution_evidence, COALESCE(correlation_id,''), COALESCE(causation_id,''), created_at, updated_at
|
|
FROM executions WHERE id = ?`, id,
|
|
)
|
|
e := &domain.Execution{}
|
|
var args, reqBy, origin, idempKey, confID, status, result, errStr, evidence, corrID, causID, createdAt, updatedAt string
|
|
err := row.Scan(&e.ID, &e.CapabilityID, &e.CapabilityVersion, &e.TargetEntityID, &e.EntityVersion, &args, &reqBy, &origin, &idempKey, &confID, &status, &result, &errStr, &evidence, &corrID, &causID, &createdAt, &updatedAt)
|
|
if err == sql.ErrNoRows {
|
|
return nil, domain.ErrExecutionNotFound
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
json.Unmarshal([]byte(args), &e.Arguments)
|
|
json.Unmarshal([]byte(reqBy), &e.RequestedBy)
|
|
json.Unmarshal([]byte(origin), &e.Origin)
|
|
json.Unmarshal([]byte(result), &e.Result)
|
|
json.Unmarshal([]byte(evidence), &e.ResolutionEvidence)
|
|
e.IdempotencyKey = idempKey
|
|
e.ConfirmationID = confID
|
|
e.Status = domain.ExecutionStatus(status)
|
|
e.Error = errStr
|
|
e.CorrelationID = corrID
|
|
e.CausationID = causID
|
|
e.CreatedAt = parseTime(createdAt)
|
|
e.UpdatedAt = parseTime(updatedAt)
|
|
if e.Arguments == nil {
|
|
e.Arguments = map[string]any{}
|
|
}
|
|
if e.RequestedBy == nil {
|
|
e.RequestedBy = map[string]string{}
|
|
}
|
|
if e.Origin == nil {
|
|
e.Origin = map[string]string{}
|
|
}
|
|
if e.Result == nil {
|
|
e.Result = map[string]any{}
|
|
}
|
|
return e, nil
|
|
}
|
|
|
|
func (s *Store) UpdateExecution(e *domain.Execution) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
result, _ := json.Marshal(e.Result)
|
|
|
|
_, err := s.db.Exec(
|
|
`UPDATE executions SET status=?, result=?, error=?, updated_at=? WHERE id=?`,
|
|
string(e.Status), string(result), e.Error, formatTime(e.UpdatedAt), e.ID,
|
|
)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) GetExecutionByIdempotencyKey(key string) (*domain.Execution, error) {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
|
|
row := s.db.QueryRow(
|
|
`SELECT id, capability_id, capability_version, target_entity_id, entity_version, arguments, requested_by, origin, COALESCE(idempotency_key,''), COALESCE(confirmation_id,''), status, result, COALESCE(error,''), resolution_evidence, COALESCE(correlation_id,''), COALESCE(causation_id,''), created_at, updated_at
|
|
FROM executions WHERE idempotency_key = ?`, key,
|
|
)
|
|
e := &domain.Execution{}
|
|
var args, reqBy, origin, idempKey, confID, status, result, errStr, evidence, corrID, causID, createdAt, updatedAt string
|
|
err := row.Scan(&e.ID, &e.CapabilityID, &e.CapabilityVersion, &e.TargetEntityID, &e.EntityVersion, &args, &reqBy, &origin, &idempKey, &confID, &status, &result, &errStr, &evidence, &corrID, &causID, &createdAt, &updatedAt)
|
|
if err == sql.ErrNoRows {
|
|
return nil, domain.ErrExecutionNotFound
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
json.Unmarshal([]byte(args), &e.Arguments)
|
|
json.Unmarshal([]byte(reqBy), &e.RequestedBy)
|
|
json.Unmarshal([]byte(origin), &e.Origin)
|
|
json.Unmarshal([]byte(result), &e.Result)
|
|
json.Unmarshal([]byte(evidence), &e.ResolutionEvidence)
|
|
e.IdempotencyKey = idempKey
|
|
e.ConfirmationID = confID
|
|
e.Status = domain.ExecutionStatus(status)
|
|
e.Error = errStr
|
|
e.CorrelationID = corrID
|
|
e.CausationID = causID
|
|
e.CreatedAt = parseTime(createdAt)
|
|
e.UpdatedAt = parseTime(updatedAt)
|
|
return e, nil
|
|
}
|
|
|
|
func (s *Store) GetInFlightExecution(capabilityID, targetEntityID string) (*domain.Execution, error) {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
|
|
row := s.db.QueryRow(
|
|
`SELECT id FROM executions WHERE capability_id = ? AND target_entity_id = ? AND status = ? LIMIT 1`,
|
|
capabilityID, targetEntityID, string(domain.ExecutionStarted),
|
|
)
|
|
var id string
|
|
err := row.Scan(&id)
|
|
if err == sql.ErrNoRows {
|
|
return nil, domain.ErrExecutionNotFound
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return s.getExecutionLocked(id)
|
|
}
|
|
|
|
// getExecutionLocked reads an execution without acquiring s.mu; callers must
|
|
// already hold it (read or write).
|
|
func (s *Store) getExecutionLocked(id string) (*domain.Execution, error) {
|
|
row := s.db.QueryRow(
|
|
`SELECT id, capability_id, capability_version, target_entity_id, entity_version, arguments, requested_by, origin, COALESCE(idempotency_key,''), COALESCE(confirmation_id,''), status, result, COALESCE(error,''), resolution_evidence, COALESCE(correlation_id,''), COALESCE(causation_id,''), created_at, updated_at
|
|
FROM executions WHERE id = ?`, id,
|
|
)
|
|
e := &domain.Execution{}
|
|
var args, reqBy, origin, idempKey, confID, status, result, errStr, evidence, corrID, causID, createdAt, updatedAt string
|
|
err := row.Scan(&e.ID, &e.CapabilityID, &e.CapabilityVersion, &e.TargetEntityID, &e.EntityVersion, &args, &reqBy, &origin, &idempKey, &confID, &status, &result, &errStr, &evidence, &corrID, &causID, &createdAt, &updatedAt)
|
|
if err == sql.ErrNoRows {
|
|
return nil, domain.ErrExecutionNotFound
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
json.Unmarshal([]byte(args), &e.Arguments)
|
|
json.Unmarshal([]byte(reqBy), &e.RequestedBy)
|
|
json.Unmarshal([]byte(origin), &e.Origin)
|
|
json.Unmarshal([]byte(result), &e.Result)
|
|
json.Unmarshal([]byte(evidence), &e.ResolutionEvidence)
|
|
e.IdempotencyKey = idempKey
|
|
e.Status = domain.ExecutionStatus(status)
|
|
e.Error = errStr
|
|
e.CorrelationID = corrID
|
|
e.CreatedAt = parseTime(createdAt)
|
|
e.UpdatedAt = parseTime(updatedAt)
|
|
return e, nil
|
|
}
|
|
|
|
// Confirmation operations
|
|
|
|
func (s *Store) CreateConfirmation(c *domain.Confirmation) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
_, err := s.db.Exec(
|
|
`INSERT INTO confirmations (id, capability_id, capability_version, target_entity_id, args_normalized, args_hash, requester, created_at, expires_at, state)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
|
c.ID, c.CapabilityID, c.CapabilityVersion, c.TargetEntityID, c.ArgsNormalized, c.ArgsHash, c.Requester, formatTime(c.CreatedAt), formatTime(c.ExpiresAt), string(c.State),
|
|
)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) GetConfirmation(id string) (*domain.Confirmation, error) {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
|
|
row := s.db.QueryRow(
|
|
`SELECT id, capability_id, capability_version, target_entity_id, args_normalized, args_hash, requester, created_at, expires_at, state
|
|
FROM confirmations WHERE id = ?`, id,
|
|
)
|
|
c := &domain.Confirmation{}
|
|
var createdAt, expiresAt, state string
|
|
err := row.Scan(&c.ID, &c.CapabilityID, &c.CapabilityVersion, &c.TargetEntityID, &c.ArgsNormalized, &c.ArgsHash, &c.Requester, &createdAt, &expiresAt, &state)
|
|
if err == sql.ErrNoRows {
|
|
return nil, domain.ErrConfirmationNotFound
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
c.CreatedAt = parseTime(createdAt)
|
|
c.ExpiresAt = parseTime(expiresAt)
|
|
c.State = domain.ConfirmationState(state)
|
|
return c, nil
|
|
}
|
|
|
|
func (s *Store) UpdateConfirmationState(id string, state domain.ConfirmationState) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
_, err := s.db.Exec(`UPDATE confirmations SET state = ? WHERE id = ?`, string(state), id)
|
|
return err
|
|
}
|
|
|
|
// Event operations
|
|
|
|
func (s *Store) AppendEvent(evt *domain.Event) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
var maxSeq sql.NullInt64
|
|
s.db.QueryRow(`SELECT MAX(sequence) FROM hexis_events`).Scan(&maxSeq)
|
|
evt.Sequence = maxSeq.Int64 + 1
|
|
|
|
payload, _ := json.Marshal(evt.Payload)
|
|
|
|
_, err := s.db.Exec(
|
|
`INSERT INTO hexis_events (id, sequence, type, timestamp, actor, correlation_id, causation_id, payload)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
|
|
evt.ID, evt.Sequence, string(evt.Type), formatTime(evt.Timestamp), evt.Actor, evt.CorrelationID, evt.CausationID, string(payload),
|
|
)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) EventsAfter(seq int64, limit int) ([]*domain.Event, error) {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
|
|
rows, err := s.db.Query(
|
|
`SELECT id, sequence, type, timestamp, COALESCE(actor,''), COALESCE(correlation_id,''), COALESCE(causation_id,''), payload
|
|
FROM hexis_events WHERE sequence > ? ORDER BY sequence LIMIT ?`, seq, limit,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var result []*domain.Event
|
|
for rows.Next() {
|
|
e := &domain.Event{}
|
|
var typ, payload, timestamp string
|
|
if err := rows.Scan(&e.ID, &e.Sequence, &typ, ×tamp, &e.Actor, &e.CorrelationID, &e.CausationID, &payload); err != nil {
|
|
return nil, err
|
|
}
|
|
e.Type = domain.HexisEventType(typ)
|
|
e.Timestamp = parseTime(timestamp)
|
|
json.Unmarshal([]byte(payload), &e.Payload)
|
|
result = append(result, e)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Store) LatestSequence() (int64, error) {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
|
|
var seq sql.NullInt64
|
|
s.db.QueryRow(`SELECT MAX(sequence) FROM hexis_events`).Scan(&seq)
|
|
return seq.Int64, nil
|
|
}
|
|
|
|
func nullString(s string) interface{} {
|
|
if s == "" {
|
|
return nil
|
|
}
|
|
return s
|
|
}
|
|
|
|
func boolInt(b bool) int {
|
|
if b {
|
|
return 1
|
|
}
|
|
return 0
|
|
}
|
|
|
|
var _ Interface = (*Store)(nil)
|