feat: Add database schema for operational gateway (#1304)

* feat: Add database schema for operational gateway

Creates initial CockroachDB migrations for the operational gateway service:

- provider_connections: stores provider endpoint configuration including auth
  config (JSONB), retry policy, rate limits, health status, and circuit
  breaker state. Composite primary key (tenant_id, connection_id).
- instructions: outbox-pattern table for outbound directives with full
  lifecycle status tracking (PENDING → DISPATCHING → DELIVERED →
  ACKNOWLEDGED / FAILED / EXPIRED / CANCELLED). Partial index on
  (status, priority, scheduled_at) covers the dispatch worker hot path.
- instruction_attempts: per-attempt audit log recording response codes,
  error messages, and duration. FK to instructions.

Adds atlas.hcl configuration for local, ci, and production environments
using the operational_gateway schema loader.

* feat: Register operational-gateway in ServiceDatabases migration map

Adds the operational-gateway service to ServiceDatabases in the
migration runner so that migration discovery does not fail with
"unknown service: no database mapping" during startup and E2E tests.

Maps operational-gateway → meridian_operational_gateway database with
dedicated meridian_operational_gateway_user credentials, following the
same pattern as all other services.

* fix: Address schema review feedback in operational gateway migration

- Use UUID type for connection_id on provider_connections (proto defines
  it as uuid; VARCHAR(64) was imprecise and inconsistent with other services)
- Change priority column from VARCHAR to SMALLINT (LOW=1, NORMAL=2, HIGH=3,
  CRITICAL=4) to ensure correct numeric ordering in the outbox dispatch index;
  VARCHAR DESC produced NORMAL>LOW>HIGH>CRITICAL which was inverted
- Drop status from outbox index key columns; the partial WHERE clause already
  filters to PENDING/RETRYING so status as a leading key column was redundant
- Add composite FK (tenant_id, provider_connection_id) → provider_connections
  to prevent cross-tenant orphaned instructions caught only at dispatch time
- Add CHECK (attempt_count >= 0) and CHECK (max_attempts >= 1) to enforce
  valid retry counters at the DB layer
- Make idx_instruction_attempts_instruction UNIQUE to prevent duplicate
  attempt ordinals that would corrupt retry/audit semantics
- Regenerate atlas.sum for updated migration checksum

---------

Co-authored-by: Ben Coombs <bjcoombs@users.noreply.github.com>

Author: bjcoombs

internal/migrations/runner.go

@@ -77,6 +77,7 @@ var ServiceDatabases = map[string]ServiceDatabase{
 	"reconciliation":       {Database: "meridian_reconciliation", User: "meridian_reconciliation_user", Password: ""},
 	"forecasting":          {Database: "meridian_forecasting", User: "meridian_forecasting_user", Password: ""},
 	"reference-data":       {Database: "meridian_reference_data", User: "meridian_reference_data_user", Password: ""},
+	"operational-gateway":  {Database: "meridian_operational_gateway", User: "meridian_operational_gateway_user", Password: ""},
 }
 
 // serviceMigration holds a single migration file for a service.

services/operational-gateway/atlas/atlas.hcl

@@ -0,0 +1,72 @@
+// Atlas configuration for Operational Gateway Service
+// BIAN Service Domain: Operational Gateway
+// Manages outbound instructions to external providers and provider connection configurations.
+// Uses database-per-service architecture with unqualified table names.
+
+data "external_schema" "gorm" {
+  program = [
+    "go",
+    "run",
+    "-mod=mod",
+    "./utilities/atlas-loader",
+    "--schema=operational_gateway"
+  ]
+}
+
+env "local" {
+  // Service-specific migration directory
+  migration {
+    dir = "file://services/operational-gateway/migrations"
+  }
+
+  // Dev database - uses default public schema
+  dev = "docker://postgres/16/dev"
+
+  // Source schema from GORM models via external loader
+  src = data.external_schema.gorm.url
+
+  // Lint configuration to catch dangerous changes
+  lint {
+    destructive {
+      error = true
+    }
+    data_depend {
+      error = true
+    }
+    incompatible {
+      error = true
+    }
+  }
+}
+
+env "ci" {
+  migration {
+    dir = "file://services/operational-gateway/migrations"
+  }
+
+  dev = "docker://postgres/16/dev"
+
+  src = data.external_schema.gorm.url
+
+  lint {
+    destructive {
+      error = true
+    }
+    data_depend {
+      error = true
+    }
+    incompatible {
+      error = true
+    }
+  }
+}
+
+env "production" {
+  // Production environment - apply only, never diff
+  // URL points to service-specific database (meridian_operational_gateway)
+  url = getenv("DATABASE_URL")
+
+  migration {
+    dir = "file://services/operational-gateway/migrations"
+  }
+}

services/operational-gateway/migrations/20260301000001_operational_gateway_initial.sql

@@ -0,0 +1,118 @@
+-- Operational Gateway Service Schema
+-- Manages outbound instructions to external providers and provider connection configurations.
+-- Uses unqualified table names (relies on database-per-service architecture).
+
+-- provider_connections stores the configuration for connecting to external provider endpoints.
+-- Connections are reused across multiple instructions of matching instruction_types.
+CREATE TABLE provider_connections (
+    tenant_id UUID NOT NULL,
+    -- connection_id is a UUID identifier matching the proto field definition.
+    connection_id UUID NOT NULL,
+    provider_name VARCHAR(255) NOT NULL,
+    provider_type VARCHAR(128) NOT NULL,
+    -- protocol: HTTPS=1, GRPC=2, WEBHOOK=3, MQTT=4, AMQP=5
+    protocol VARCHAR(20) NOT NULL CHECK (protocol IN ('HTTPS', 'GRPC', 'WEBHOOK', 'MQTT', 'AMQP')),
+    base_url VARCHAR(2048) NOT NULL,
+    -- auth_config stores the serialised authentication configuration (ApiKeyAuth, BasicAuth, OAuth2Auth, HMACAuth, MTLSAuth).
+    -- Exactly one auth variant is populated; the auth_type discriminator field identifies which.
+    auth_config JSONB NOT NULL,
+    -- retry_policy stores the RetryPolicy fields: max_attempts, initial_backoff_seconds, max_backoff_seconds, backoff_multiplier.
+    retry_policy JSONB NULL,
+    -- rate_limit_config stores the RateLimit fields: requests_per_second, burst_size.
+    rate_limit_config JSONB NULL,
+    -- health_status: UNSPECIFIED=0, HEALTHY=1, DEGRADED=2, UNHEALTHY=3
+    health_status VARCHAR(20) NOT NULL DEFAULT 'UNSPECIFIED' CHECK (health_status IN ('UNSPECIFIED', 'HEALTHY', 'DEGRADED', 'UNHEALTHY')),
+    last_health_check_at TIMESTAMPTZ NULL,
+    -- circuit_state tracks the circuit-breaker state for this connection.
+    -- CLOSED (normal), OPEN (blocking dispatch), HALF_OPEN (probing recovery).
+    circuit_state VARCHAR(20) NOT NULL DEFAULT 'CLOSED' CHECK (circuit_state IN ('CLOSED', 'OPEN', 'HALF_OPEN')),
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    PRIMARY KEY (tenant_id, connection_id)
+);
+
+-- Index for listing connections per tenant (common read path).
+CREATE INDEX idx_provider_connections_tenant ON provider_connections (tenant_id);
+
+-- Index for health-based filtering (worker queries unhealthy connections for health checks).
+CREATE INDEX idx_provider_connections_health ON provider_connections (tenant_id, health_status);
+
+-- instructions is the outbox table for outbound operational directives sent to external providers.
+-- Each instruction follows the outbox pattern: written atomically with the originating saga step,
+-- then picked up by the dispatch worker for async delivery.
+CREATE TABLE instructions (
+    id UUID NOT NULL DEFAULT gen_random_uuid(),
+    tenant_id UUID NOT NULL,
+    -- instruction_type identifies the category of operation (e.g. "payment.initiate", "account.freeze").
+    instruction_type VARCHAR(255) NOT NULL,
+    -- provider_connection_id references the connection used to dispatch this instruction.
+    -- FK enforced as composite (tenant_id, provider_connection_id) to prevent cross-tenant orphans.
+    provider_connection_id UUID NOT NULL,
+    -- correlation_id links all events across services for a single user request.
+    correlation_id VARCHAR(255) NULL,
+    -- causation_id identifies the event or command that caused this instruction to be created.
+    causation_id VARCHAR(255) NULL,
+    -- payload is the instruction-specific data serialised as JSON (google.protobuf.Struct).
+    payload JSONB NOT NULL,
+    -- metadata stores additional key-value pairs for routing, filtering, or audit purposes.
+    metadata JSONB NULL,
+    -- priority uses SMALLINT to allow correct numeric ordering in the dispatch index.
+    -- LOW=1, NORMAL=2, HIGH=3, CRITICAL=4 (matching the Priority proto enum values).
+    priority SMALLINT NOT NULL DEFAULT 2 CHECK (priority IN (1, 2, 3, 4)),
+    -- status follows the InstructionStatus state machine defined in the proto.
+    status VARCHAR(20) NOT NULL DEFAULT 'PENDING' CHECK (status IN ('PENDING', 'DISPATCHING', 'DELIVERED', 'ACKNOWLEDGED', 'RETRYING', 'FAILED', 'EXPIRED', 'CANCELLED')),
+    scheduled_at TIMESTAMPTZ NULL,
+    expires_at TIMESTAMPTZ NULL,
+    attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
+    max_attempts INTEGER NOT NULL DEFAULT 3 CHECK (max_attempts >= 1),
+    next_retry_at TIMESTAMPTZ NULL,
+    -- idempotency_key ensures exactly-once dispatch of each instruction.
+    idempotency_key VARCHAR(255) NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    PRIMARY KEY (id),
+    FOREIGN KEY (tenant_id, provider_connection_id) REFERENCES provider_connections (tenant_id, connection_id)
+);
+
+-- Outbox polling index: dispatch worker polls for PENDING/RETRYING instructions ordered by priority DESC
+-- (CRITICAL=4 first) then by scheduled_at ASC. Partial index limits scan to actionable rows only.
+-- Using numeric priority (SMALLINT) ensures correct ordering: 4 > 3 > 2 > 1.
+CREATE INDEX idx_instructions_outbox ON instructions (priority DESC, scheduled_at ASC)
+    WHERE status IN ('PENDING', 'RETRYING');
+
+-- Tenant + type index: used by list and routing queries.
+CREATE INDEX idx_instructions_tenant_type ON instructions (tenant_id, instruction_type);
+
+-- Correlation index: used for distributed tracing lookups.
+CREATE INDEX idx_instructions_correlation ON instructions (tenant_id, correlation_id)
+    WHERE correlation_id IS NOT NULL;
+
+-- Idempotency index: enforces exactly-once dispatch per tenant.
+CREATE UNIQUE INDEX idx_instructions_idempotency ON instructions (tenant_id, idempotency_key);
+
+-- instruction_attempts records the outcome of each individual dispatch attempt.
+-- Stored separately to avoid unbounded row growth on the instructions table.
+CREATE TABLE instruction_attempts (
+    id UUID NOT NULL DEFAULT gen_random_uuid(),
+    -- instruction_id references the parent instruction.
+    instruction_id UUID NOT NULL,
+    -- attempt_number is the 1-based ordinal of this attempt.
+    -- Uniqueness constraint prevents duplicate ordinals from corrupting retry/audit semantics.
+    attempt_number INTEGER NOT NULL CHECK (attempt_number >= 1),
+    dispatched_at TIMESTAMPTZ NOT NULL,
+    completed_at TIMESTAMPTZ NULL,
+    -- response_status_code is the HTTP or gRPC status code returned by the provider (0 if no response).
+    response_status_code INTEGER NULL CHECK (response_status_code IS NULL OR (response_status_code >= 0 AND response_status_code <= 599)),
+    -- response_body_preview is the first 1KB of the provider response body for diagnostics.
+    response_body_preview VARCHAR(1024) NULL,
+    -- error_message describes the error if this attempt failed.
+    error_message TEXT NULL,
+    -- duration_ms is how long the dispatch attempt took in milliseconds.
+    duration_ms BIGINT NULL CHECK (duration_ms IS NULL OR duration_ms >= 0),
+    PRIMARY KEY (id),
+    FOREIGN KEY (instruction_id) REFERENCES instructions (id)
+);
+
+-- Unique index on (instruction_id, attempt_number) ensures ordinals cannot be duplicated,
+-- while also serving as the primary lookup path for attempt retrieval.
+CREATE UNIQUE INDEX idx_instruction_attempts_instruction ON instruction_attempts (instruction_id, attempt_number);

services/operational-gateway/migrations/atlas.sum

@@ -0,0 +1,2 @@
+h1:vg1f6HFO/8K8367MMxUfk8zXiPMf/669AYQlGx+ahz8=
+20260301000001_operational_gateway_initial.sql h1:0k52bq50GAdIbOVr4HTX9BEzFkbbL/RA4EQGSI0HnnU=