package pg import ( "context" "database/sql" "encoding/json" "errors" "fmt" "strings" "sync" "time" "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/hooks" "github.com/nextlevelbuilder/goclaw/internal/store" ) // PGHookStore implements hooks.HookStore backed by PostgreSQL. type PGHookStore struct { db *sql.DB cacheMu sync.Mutex cache map[string]pgHookCacheEntry // keyed by resolveKey } type pgHookCacheEntry struct { result []hooks.HookConfig maxVersion int expiresAt time.Time } const hookCacheTTL = 5 * time.Second // NewPGHookStore returns a PGHookStore backed by the given *sql.DB. func NewPGHookStore(db *sql.DB) *PGHookStore { return &PGHookStore{ db: db, cache: make(map[string]pgHookCacheEntry), } } // ─── Create ───────────────────────────────────────────────────────────────── func (s *PGHookStore) Create(ctx context.Context, cfg hooks.HookConfig) (uuid.UUID, error) { // Honor a caller-provided fixed ID when non-nil (H9 fix) — the Phase 04 // builtin seeder needs idempotent UPSERTs keyed by UUIDv5, and tests need // deterministic IDs. Fall back to UUIDv7 only when the caller did not // supply one. id := cfg.ID if id == uuid.Nil { id = uuid.Must(uuid.NewV7()) } now := time.Now().UTC() cfgJSON, err := json.Marshal(cfg.Config) if err != nil { return uuid.Nil, fmt.Errorf("marshal config: %w", err) } metaJSON, err := json.Marshal(cfg.Metadata) if err != nil { return uuid.Nil, fmt.Errorf("marshal metadata: %w", err) } tid := cfg.TenantID if tid == uuid.Nil { tid = tenantIDForInsert(ctx) } var matcher, ifExpr, name *string if cfg.Matcher != "" { matcher = &cfg.Matcher } if cfg.IfExpr != "" { ifExpr = &cfg.IfExpr } if cfg.Name != "" { name = &cfg.Name } tx, err := s.db.BeginTx(ctx, nil) if err != nil { return uuid.Nil, fmt.Errorf("begin transaction: %w", err) } defer tx.Rollback() //nolint:errcheck _, err = tx.ExecContext(ctx, ` INSERT INTO hooks (id, tenant_id, scope, event, handler_type, config, matcher, if_expr, timeout_ms, on_timeout, priority, enabled, version, source, metadata, name, created_by, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,1,$13,$14,$15,$16,$17,$17)`, id, tid, string(cfg.Scope), string(cfg.Event), string(cfg.HandlerType), cfgJSON, matcher, ifExpr, cfg.TimeoutMS, string(cfg.OnTimeout), cfg.Priority, cfg.Enabled, string(cfg.Source), metaJSON, name, cfg.CreatedBy, now, ) if err != nil { return uuid.Nil, fmt.Errorf("insert hook: %w", err) } // Bridge deprecated AgentID → AgentIDs for junction insertion. agentIDs := cfg.AgentIDs if len(agentIDs) == 0 && cfg.AgentID != nil && *cfg.AgentID != uuid.Nil { agentIDs = []uuid.UUID{*cfg.AgentID} } for _, aid := range agentIDs { if _, err := tx.ExecContext(ctx, `INSERT INTO hook_agents (hook_id, agent_id) VALUES ($1, $2) ON CONFLICT DO NOTHING`, id, aid); err != nil { return uuid.Nil, fmt.Errorf("insert hook agent link: %w", err) } } if err := tx.Commit(); err != nil { return uuid.Nil, fmt.Errorf("commit hook create: %w", err) } s.invalidateCache() return id, nil } // ─── GetByID ───────────────────────────────────────────────────────────────── func (s *PGHookStore) GetByID(ctx context.Context, id uuid.UUID) (*hooks.HookConfig, error) { q := ` SELECT id, tenant_id, scope, event, handler_type, config, matcher, if_expr, timeout_ms, on_timeout, priority, enabled, version, source, metadata, name, created_by, created_at, updated_at FROM hooks WHERE id = $1` args := []any{id} // Tenant-scope guard: non-master callers only see their own rows + globals. // Global hooks use tenant_id = SentinelTenantID (store.MasterTenantID). if !store.IsMasterScope(ctx) { tid := store.TenantIDFromContext(ctx) if tid == uuid.Nil { return nil, fmt.Errorf("tenant_id required for non-master scope") } q += " AND (tenant_id = $2 OR tenant_id = $3)" args = append(args, tid, store.MasterTenantID) } row := s.db.QueryRowContext(ctx, q, args...) cfg, err := scanHookPGRow(row) if errors.Is(err, sql.ErrNoRows) { return nil, nil } if err != nil { return nil, fmt.Errorf("get hook by id: %w", err) } if aids, err := s.GetHookAgents(ctx, cfg.ID); err == nil { cfg.AgentIDs = aids } return cfg, nil } // ─── List ──────────────────────────────────────────────────────────────────── func (s *PGHookStore) List(ctx context.Context, filter hooks.ListFilter) ([]hooks.HookConfig, error) { q := `SELECT id, tenant_id, scope, event, handler_type, config, matcher, if_expr, timeout_ms, on_timeout, priority, enabled, version, source, metadata, name, created_by, created_at, updated_at FROM hooks WHERE 1=1` var args []any n := 1 if !store.IsMasterScope(ctx) { tid := store.TenantIDFromContext(ctx) if tid == uuid.Nil { return nil, fmt.Errorf("tenant_id required for non-master scope") } // Return tenant-specific + global hooks. q += fmt.Sprintf(" AND (tenant_id = $%d OR tenant_id = $%d)", n, n+1) args = append(args, tid, store.MasterTenantID) n += 2 } else if filter.TenantID != nil { q += fmt.Sprintf(" AND tenant_id = $%d", n) args = append(args, *filter.TenantID) n++ } if filter.AgentID != nil { q += fmt.Sprintf(" AND id IN (SELECT hook_id FROM hook_agents WHERE agent_id = $%d)", n) args = append(args, *filter.AgentID) n++ } if filter.Event != nil { q += fmt.Sprintf(" AND event = $%d", n) args = append(args, string(*filter.Event)) n++ } if filter.Scope != nil { q += fmt.Sprintf(" AND scope = $%d", n) args = append(args, string(*filter.Scope)) n++ } if filter.Enabled != nil { q += fmt.Sprintf(" AND enabled = $%d", n) args = append(args, *filter.Enabled) n++ } _ = n q += " ORDER BY priority DESC, created_at ASC" rows, err := s.db.QueryContext(ctx, q, args...) if err != nil { return nil, fmt.Errorf("list hooks: %w", err) } defer rows.Close() var result []hooks.HookConfig for rows.Next() { cfg, err := scanHookPGRow(rows) if err != nil { return nil, err } result = append(result, *cfg) } if err := rows.Err(); err != nil { return nil, err } // Batch-populate AgentIDs from junction table. for i := range result { if aids, err := s.GetHookAgents(ctx, result[i].ID); err == nil { result[i].AgentIDs = aids } } return result, nil } // ─── Update ────────────────────────────────────────────────────────────────── func (s *PGHookStore) Update(ctx context.Context, id uuid.UUID, updates map[string]any) error { if _, ok := updates["version"]; ok { return fmt.Errorf("callers must not include 'version' in update map") } // Builtin-row protection (Phase 04). Rows with source='builtin' are canonical // embed-seeded artifacts: UI/user writes may only toggle `enabled`. The // builtin seeder itself passes through via hooks.WithSeedBypass ctx. // Fail-closed: a transient GetByID error must not let a non-enabled patch // slip through against a builtin row. if !hooks.IsSeedBypass(ctx) { current, err := s.GetByID(ctx, id) if err != nil { return fmt.Errorf("builtin guard: %w", err) } if current != nil && current.Source == hooks.SourceBuiltin { for k := range updates { if k != "enabled" { return hooks.ErrBuiltinReadOnly } } } } // Handle agent_ids separately — routes to junction table, not a column. if raw, ok := updates["agent_ids"]; ok { delete(updates, "agent_ids") ids, err := parseAgentIDsFromAny(raw) if err != nil { return fmt.Errorf("invalid agent_ids: %w", err) } if err := s.SetHookAgents(ctx, id, ids); err != nil { return err } } // Marshal map/slice values to JSON for JSONB columns. for k, v := range updates { switch k { case "config", "metadata": b, err := json.Marshal(v) if err != nil { return fmt.Errorf("marshal %s: %w", k, err) } updates[k] = json.RawMessage(b) } } // Build SET clause with version bump. var setClauses []string var args []any i := 1 for col, val := range updates { if !validColumnName.MatchString(col) { return fmt.Errorf("invalid column name: %q", col) } setClauses = append(setClauses, fmt.Sprintf("%s = $%d", col, i)) args = append(args, val) i++ } // Always bump version and updated_at atomically. setClauses = append(setClauses, fmt.Sprintf("version = version + 1, updated_at = $%d", i)) args = append(args, time.Now().UTC()) i++ args = append(args, id) q := fmt.Sprintf("UPDATE hooks SET %s WHERE id = $%d", strings.Join(setClauses, ", "), i) i++ if !store.IsMasterScope(ctx) { tid := store.TenantIDFromContext(ctx) if tid == uuid.Nil { return fmt.Errorf("tenant_id required for non-master scope") } q += fmt.Sprintf(" AND tenant_id = $%d", i) args = append(args, tid) } res, err := s.db.ExecContext(ctx, q, args...) if err != nil { return fmt.Errorf("update hook: %w", err) } if n, _ := res.RowsAffected(); n == 0 { return fmt.Errorf("hook not found: %s", id) } s.invalidateCache() return nil } // ─── Delete ────────────────────────────────────────────────────────────────── func (s *PGHookStore) Delete(ctx context.Context, id uuid.UUID) error { // Builtin-row protection (Phase 04): reject unless caller marked seed-bypass. // Fail-closed on GetByID errors so we don't accidentally delete a builtin. if !hooks.IsSeedBypass(ctx) { current, err := s.GetByID(ctx, id) if err != nil { return fmt.Errorf("builtin guard: %w", err) } if current != nil && current.Source == hooks.SourceBuiltin { return hooks.ErrBuiltinReadOnly } } q := "DELETE FROM hooks WHERE id = $1" args := []any{id} if !store.IsMasterScope(ctx) { tid := store.TenantIDFromContext(ctx) if tid == uuid.Nil { return fmt.Errorf("tenant_id required for non-master scope") } q += " AND tenant_id = $2" args = append(args, tid) } res, err := s.db.ExecContext(ctx, q, args...) if err != nil { return fmt.Errorf("delete hook: %w", err) } if n, _ := res.RowsAffected(); n == 0 { return fmt.Errorf("hook not found: %s", id) } s.invalidateCache() return nil } // ─── ResolveForEvent ───────────────────────────────────────────────────────── func (s *PGHookStore) ResolveForEvent(ctx context.Context, event hooks.Event) ([]hooks.HookConfig, error) { tenantID := event.TenantID agentID := event.AgentID hookEvent := event.HookEvent // Check max version in DB to validate cache freshness. var maxVersion int err := s.db.QueryRowContext(ctx, `SELECT COALESCE(MAX(version),0) FROM hooks WHERE enabled = TRUE AND event = $1 AND (tenant_id = $2 OR tenant_id = $3) AND ( scope IN ('global', 'tenant') OR (scope = 'agent' AND EXISTS ( SELECT 1 FROM hook_agents aha WHERE aha.hook_id = hooks.id AND aha.agent_id = $4 )) )`, string(hookEvent), tenantID, store.MasterTenantID, agentID, ).Scan(&maxVersion) if err != nil { return nil, fmt.Errorf("resolve version check: %w", err) } key := hookResolveKey(tenantID, agentID, hookEvent) s.cacheMu.Lock() entry, ok := s.cache[key] s.cacheMu.Unlock() if ok && time.Now().Before(entry.expiresAt) && entry.maxVersion == maxVersion { return entry.result, nil } rows, err := s.db.QueryContext(ctx, ` SELECT id, tenant_id, scope, event, handler_type, config, matcher, if_expr, timeout_ms, on_timeout, priority, enabled, version, source, metadata, name, created_by, created_at, updated_at FROM hooks WHERE enabled = TRUE AND event = $1 AND (tenant_id = $2 OR tenant_id = $3) AND ( scope IN ('global', 'tenant') OR (scope = 'agent' AND EXISTS ( SELECT 1 FROM hook_agents aha WHERE aha.hook_id = hooks.id AND aha.agent_id = $4 )) ) ORDER BY priority DESC, created_at ASC`, string(hookEvent), tenantID, store.MasterTenantID, agentID, ) if err != nil { return nil, fmt.Errorf("resolve hooks: %w", err) } defer rows.Close() var result []hooks.HookConfig for rows.Next() { cfg, err := scanHookPGRow(rows) if err != nil { return nil, err } result = append(result, *cfg) } if err := rows.Err(); err != nil { return nil, err } s.cacheMu.Lock() s.cache[key] = pgHookCacheEntry{ result: result, maxVersion: maxVersion, expiresAt: time.Now().Add(hookCacheTTL), } s.cacheMu.Unlock() return result, nil } // ─── WriteExecution ────────────────────────────────────────────────────────── func (s *PGHookStore) WriteExecution(ctx context.Context, exec hooks.HookExecution) error { metaJSON, err := json.Marshal(exec.Metadata) if err != nil { return fmt.Errorf("marshal exec metadata: %w", err) } now := exec.CreatedAt if now.IsZero() { now = time.Now().UTC() } var dedupKey, sessionID, inputHash, errStr *string if exec.DedupKey != "" { dedupKey = &exec.DedupKey } if exec.SessionID != "" { sessionID = &exec.SessionID } if exec.InputHash != "" { inputHash = &exec.InputHash } if exec.Error != "" { errStr = &exec.Error } _, err = s.db.ExecContext(ctx, ` INSERT INTO hook_executions (id, hook_id, tenant_id, session_id, event, input_hash, decision, duration_ms, retry, dedup_key, error, error_detail, metadata, created_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14) ON CONFLICT (dedup_key) WHERE dedup_key IS NOT NULL DO NOTHING`, exec.ID, exec.HookID, exec.TenantID, sessionID, string(exec.Event), inputHash, string(exec.Decision), exec.DurationMS, exec.Retry, dedupKey, errStr, exec.ErrorDetail, metaJSON, now, ) if err != nil { return fmt.Errorf("write execution: %w", err) } return nil } // parseAgentIDsFromAny converts the raw JSON-decoded value (typically // []interface{} of string UUIDs) into []uuid.UUID. func parseAgentIDsFromAny(raw any) ([]uuid.UUID, error) { switch v := raw.(type) { case []any: var ids []uuid.UUID for _, item := range v { s, ok := item.(string) if !ok { continue } id, err := uuid.Parse(s) if err != nil { return nil, fmt.Errorf("invalid agent_id %q: %w", s, err) } ids = append(ids, id) } return ids, nil case []uuid.UUID: return v, nil default: return nil, fmt.Errorf("agent_ids: unsupported type %T", raw) } } // ─── N:M junction: hook_agents ──────────────────────────────────────── func (s *PGHookStore) SetHookAgents(ctx context.Context, hookID uuid.UUID, agentIDs []uuid.UUID) error { tx, err := s.db.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() if _, err := tx.ExecContext(ctx, "DELETE FROM hook_agents WHERE hook_id = $1", hookID); err != nil { return fmt.Errorf("clear hook agents: %w", err) } for _, aid := range agentIDs { if _, err := tx.ExecContext(ctx, "INSERT INTO hook_agents (hook_id, agent_id) VALUES ($1, $2) ON CONFLICT DO NOTHING", hookID, aid); err != nil { return fmt.Errorf("insert hook agent: %w", err) } } s.invalidateCache() return tx.Commit() } func (s *PGHookStore) GetHookAgents(ctx context.Context, hookID uuid.UUID) ([]uuid.UUID, error) { rows, err := s.db.QueryContext(ctx, "SELECT agent_id FROM hook_agents WHERE hook_id = $1", hookID) if err != nil { return nil, fmt.Errorf("get hook agents: %w", err) } defer rows.Close() var ids []uuid.UUID for rows.Next() { var id uuid.UUID if err := rows.Scan(&id); err != nil { return nil, err } ids = append(ids, id) } return ids, rows.Err() } // ─── cache helpers ─────────────────────────────────────────────────────────── func (s *PGHookStore) invalidateCache() { s.cacheMu.Lock() s.cache = make(map[string]pgHookCacheEntry) s.cacheMu.Unlock() } func hookResolveKey(tenantID, agentID uuid.UUID, event hooks.HookEvent) string { return tenantID.String() + "|" + agentID.String() + "|" + string(event) } // ─── scan helper ───────────────────────────────────────────────────────────── type pgRowScanner interface { Scan(dest ...any) error } func scanHookPGRow(row pgRowScanner) (*hooks.HookConfig, error) { var ( cfg hooks.HookConfig createdBy *uuid.UUID scope, event string handlerType, onTimeout string source string matcher, ifExpr, name sql.NullString cfgJSON, metaJSON []byte ) err := row.Scan( &cfg.ID, &cfg.TenantID, &scope, &event, &handlerType, &cfgJSON, &matcher, &ifExpr, &cfg.TimeoutMS, &onTimeout, &cfg.Priority, &cfg.Enabled, &cfg.Version, &source, &metaJSON, &name, &createdBy, &cfg.CreatedAt, &cfg.UpdatedAt, ) if err != nil { return nil, err } cfg.Scope = hooks.Scope(scope) cfg.Event = hooks.HookEvent(event) cfg.HandlerType = hooks.HandlerType(handlerType) cfg.OnTimeout = hooks.Decision(onTimeout) cfg.Source = source cfg.CreatedBy = createdBy if matcher.Valid { cfg.Matcher = matcher.String } if ifExpr.Valid { cfg.IfExpr = ifExpr.String } if name.Valid { cfg.Name = name.String } if len(cfgJSON) > 0 { _ = json.Unmarshal(cfgJSON, &cfg.Config) } if cfg.Config == nil { cfg.Config = map[string]any{} } if len(metaJSON) > 0 { _ = json.Unmarshal(metaJSON, &cfg.Metadata) } if cfg.Metadata == nil { cfg.Metadata = map[string]any{} } return &cfg, nil }