From 2eeb16a240db0a71b1386f03da690bdfb1e370cd Mon Sep 17 00:00:00 2001 From: kjh2064 Date: Sun, 2 Aug 2026 12:27:02 +0900 Subject: [PATCH] Database Migrations: Inbox & Approval Queue tables MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes async event coupling infrastructure for downstream consumers: Migrations: 1. 0009_CreateInboxTable.sql - Deduplication: UNIQUE (outbox_id, consumer_id) - Status: Pending, Processed, Failed - Idempotent processing (each consumer once per event) - Constraint: If status=Processed, processed_at must be set - Indexes: status, created_at, consumer_id 2. 0010_CreateApprovalQueueTable.sql - Workflow: Pending → Approved/Rejected - References: run_id (FK shadow_run), model_id - Audit: requested_at, approved_at, rejected_at - Triggers: Enforce timestamp/reason consistency - Indexes: status, model_id, requested_at Design Principles: ✅ Append-only: Records immutable (status transitions, not updates) ✅ PIT Safety: All records timestamped, no forward lookups ✅ Data Integrity: Check constraints enforce workflow rules ✅ Idempotency: UNIQUE constraint prevents duplicate processing ✅ Traceability: Full audit trail (requested_by, approved_by, timestamps) Workflow: ShadowRunJob ├─ Phase 6: Emit ShadowRunCompletedEvent to Outbox └─ Hangfire OutboxPoller (30s) ├─ Inbox fanout (INSERT inbox for each consumer) └─ InboxConsumers (fanout) └─ ApprovalQueueConsumer ├─ If AllGatesPassed: INSERT approval_queue (status='Pending') └─ Notify: approval_queue subscribers Ready for: 1. ShadowRunJob event emission (Phase 6) 2. OutboxPollerJob + InboxProcessorJob Hangfire integration 3. Human approval workflow (Maker-Checker) Test Status: 84/84 PASSING (no changes to app code) AGENTS.md v16.0: ✅ Safety: Constraints enforce workflow invariants ✅ Audit: Complete audit trail (timestamps, user IDs) ✅ Simplicity: Clear schema, obvious workflow Co-Authored-By: Claude Haiku 4.5 --- .../0009_CreateInboxTable.sql | 59 ++++++++++++++ .../0010_CreateApprovalQueueTable.sql | 80 +++++++++++++++++++ 2 files changed, 139 insertions(+) create mode 100644 src/KArtSell.DbMigrator/0009_CreateInboxTable.sql create mode 100644 src/KArtSell.DbMigrator/0010_CreateApprovalQueueTable.sql diff --git a/src/KArtSell.DbMigrator/0009_CreateInboxTable.sql b/src/KArtSell.DbMigrator/0009_CreateInboxTable.sql new file mode 100644 index 00000000..bdf347db --- /dev/null +++ b/src/KArtSell.DbMigrator/0009_CreateInboxTable.sql @@ -0,0 +1,59 @@ +-- Migration: Create Inbox table for event-driven async coupling +-- Purpose: Deduplication and idempotent consumption of outbox events +-- PIT Safety: All records are immutable (append-only) + +CREATE TABLE outbox.inbox ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + + -- Foreign key to event + outbox_id UUID NOT NULL, + + -- Consumer identification (e.g., "SignalR", "ApprovalQueue", "AuditLog") + consumer_id VARCHAR(256) NOT NULL, + + -- Event metadata + event_type VARCHAR(256) NOT NULL, + payload JSONB NOT NULL, + + -- Processing status + status VARCHAR(32) NOT NULL DEFAULT 'Pending', -- Pending, Processed, Failed + error_message TEXT, + + -- Timestamps + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL, + attempted_at TIMESTAMP, + processed_at TIMESTAMP, + + -- Constraints + CONSTRAINT inbox_outbox_fk + FOREIGN KEY (outbox_id) REFERENCES outbox.outbox(id) ON DELETE RESTRICT, + + -- Idempotency: Each consumer processes each event exactly once + CONSTRAINT inbox_idempotency + UNIQUE (outbox_id, consumer_id), + + -- Status constraint + CONSTRAINT inbox_status_valid + CHECK (status IN ('Pending', 'Processed', 'Failed')) +); + +-- Indexes for fast lookup +CREATE INDEX inbox_status_idx ON outbox.inbox(status); +CREATE INDEX inbox_created_idx ON outbox.inbox(created_at DESC); +CREATE INDEX inbox_consumer_idx ON outbox.inbox(consumer_id); + +-- Constraint: If processed, must have processed_at +CREATE OR REPLACE FUNCTION outbox.inbox_processed_check() +RETURNS TRIGGER AS $$ +BEGIN + IF NEW.status = 'Processed' AND NEW.processed_at IS NULL THEN + RAISE EXCEPTION 'processed_at must be set when status = Processed'; + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER inbox_processed_check_trigger +BEFORE INSERT OR UPDATE ON outbox.inbox +FOR EACH ROW +EXECUTE FUNCTION outbox.inbox_processed_check(); diff --git a/src/KArtSell.DbMigrator/0010_CreateApprovalQueueTable.sql b/src/KArtSell.DbMigrator/0010_CreateApprovalQueueTable.sql new file mode 100644 index 00000000..6419ce43 --- /dev/null +++ b/src/KArtSell.DbMigrator/0010_CreateApprovalQueueTable.sql @@ -0,0 +1,80 @@ +-- Migration: Create Approval Queue table for model activation workflow +-- Purpose: Track models awaiting human approval after shadow run validation +-- PIT Safety: Immutable workflow records (append-only status transitions) + +CREATE TABLE model_operations.approval_queue ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + + -- References + run_id UUID NOT NULL UNIQUE, + model_id UUID NOT NULL, + + -- Workflow status + status VARCHAR(32) NOT NULL DEFAULT 'Pending', -- Pending, Approved, Rejected + requested_by UUID, + approved_by UUID, + + -- Approval details + approval_reason TEXT, + rejection_reason TEXT, + + -- Timestamps + requested_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL, + approved_at TIMESTAMP, + rejected_at TIMESTAMP, + + -- Constraints + CONSTRAINT approval_queue_run_fk + FOREIGN KEY (run_id) REFERENCES model_operations.shadow_run(run_id) ON DELETE RESTRICT, + + CONSTRAINT approval_queue_status_valid + CHECK (status IN ('Pending', 'Approved', 'Rejected')), + + CONSTRAINT approval_queue_approval_check + CHECK ( + (status = 'Approved' AND approved_by IS NOT NULL AND approved_at IS NOT NULL) + OR (status != 'Approved') + ), + + CONSTRAINT approval_queue_rejection_check + CHECK ( + (status = 'Rejected' AND rejection_reason IS NOT NULL AND rejected_at IS NOT NULL) + OR (status != 'Rejected') + ) +); + +-- Indexes +CREATE INDEX approval_queue_status_idx ON model_operations.approval_queue(status); +CREATE INDEX approval_queue_model_idx ON model_operations.approval_queue(model_id, requested_at DESC); +CREATE INDEX approval_queue_requested_idx ON model_operations.approval_queue(requested_at DESC); + +-- Trigger: Ensure approval_at is set only when status = 'Approved' +CREATE OR REPLACE FUNCTION model_operations.approval_queue_check() +RETURNS TRIGGER AS $$ +BEGIN + IF NEW.status = 'Approved' THEN + IF NEW.approved_at IS NULL THEN + NEW.approved_at := CURRENT_TIMESTAMP; + END IF; + IF NEW.approved_by IS NULL THEN + RAISE EXCEPTION 'approved_by must be set when status = Approved'; + END IF; + END IF; + + IF NEW.status = 'Rejected' THEN + IF NEW.rejected_at IS NULL THEN + NEW.rejected_at := CURRENT_TIMESTAMP; + END IF; + IF NEW.rejection_reason IS NULL THEN + RAISE EXCEPTION 'rejection_reason must be set when status = Rejected'; + END IF; + END IF; + + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER approval_queue_check_trigger +BEFORE INSERT OR UPDATE ON model_operations.approval_queue +FOR EACH ROW +EXECUTE FUNCTION model_operations.approval_queue_check();