feat(identity): add audit system migration and provisioner reconciliation (#2179)

* feat(identity): add audit_log and audit_outbox migration

Identity service was missing audit infrastructure, causing identity
mutations to produce no audit trail. Add migration matching the party
service pattern: audit_log for permanent records, audit_outbox for
async processing. Tables use unqualified names so per-tenant
search_path routing places them in each org schema.

Without this migration, audit hooks targeting meridian_identity have
no destination tables and existing tenant schemas have no audit
trail at all. Companion change in the provisioner reconciles this
schema into already-provisioned tenants.

* feat(tenant): reconcile migrations on provisioning worker startup

The ReconcileMigrations endpoint already applies new migration files to
existing tenant schemas, but was only reachable via a manual gRPC call.
This left the gap where a deploy adding new migration files (such as the
identity audit_log/audit_outbox tables) would fix new tenants but leave
existing tenants without the schema additions.

Wire the worker's Start path to invoke ReconcileMigrations(nil) once after
crash recovery and before the polling loop begins. Errors are logged but
never block startup - reconciliation is best-effort, identical to the
existing crash recovery pass.

Tests use a controlled mock that tracks reconciliation calls to verify
the wiring runs exactly once on startup, plus a positive-path test that
confirms reconciliation does not block subsequent provisioning of pending
tenants. The MockProvisioner gains a ReconciliationCalls slice so other
tests can assert on the tenantID argument shape.

* fix(identity): align audit_log schema with audit infrastructure

The original migration used changed_at, matching the historical party
audit table at 20251217. Every other service had to retrofit that with
20260323 alignment migrations because the runtime audit code (GORM
AuditLog model, Kafka consumer buildAuditLogFromEvent, outbox processor,
audit-worker TenantAuditWriter) all read and write created_at, plus
event_id, schema_name, correlation_id, causation_id, and idempotency_key.

Identity is brand new so it can start with the aligned schema and skip
the retrofit chain. Concretely:

- Rename changed_at to created_at (with default now()) and update the
  index name. Without this, INSERT/SELECT from the audit consumer and
  the outbox worker would fail at runtime against the identity schema.
- Add event_id (with unique index) plus schema_name, correlation_id,
  causation_id, and idempotency_key columns expected by the audit-worker
  TenantAuditWriter idempotent insert path.
- Add INITIAL_IMPORT to the operation CHECK enum to match the proto
  AuditOperation surface in shared/platform/audit/consumer.go.

Also clear ReconciliationCalls in MockProvisioner.Reset() so the slice
behaves like the other call-tracking slices and doesn't leak state
across tests that share a mock instance.

Atlas hash regenerated; atlas migrate validate passes.

---------

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

Author: bjcoombs

services/identity/migrations/20260425000001_audit_system.sql

@@ -0,0 +1,91 @@
+-- Identity Service Audit System
+-- Static audit table creation (CockroachDB compatible)
+-- Audit logging handled at application level via GORM hooks
+-- See ADR-0009 for rationale
+-- Uses unqualified table names (relies on per-tenant schema routing via search_path)
+--
+-- Schema is aligned with shared/platform/audit (created_at, not changed_at) and the
+-- audit-worker TenantAuditWriter (event_id, schema_name, correlation_id, causation_id,
+-- idempotency_key) from the start. Other services accreted these columns through
+-- follow-up alignment migrations (services/party/migrations/20260323*); identity
+-- starts aligned to avoid the same retrofit.
+
+-- Create audit_log table (unqualified, singular)
+CREATE TABLE IF NOT EXISTS audit_log (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+
+    -- What changed
+    table_name VARCHAR(100) NOT NULL,
+    operation VARCHAR(10) NOT NULL CHECK (operation IN ('INSERT', 'UPDATE', 'DELETE', 'INITIAL_IMPORT')),
+
+    -- Record identification
+    record_id UUID NOT NULL,
+
+    -- Change metadata
+    -- created_at matches the GORM AuditLog model in shared/platform/audit/hooks.go
+    -- and the consumer/worker INSERT/SELECT paths.
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    changed_by VARCHAR(100),
+
+    -- Change details (TEXT to match GORM model - contains JSON strings)
+    old_values TEXT,
+    new_values TEXT,
+
+    -- Additional context
+    transaction_id VARCHAR(100),
+    client_ip VARCHAR(45),
+    user_agent TEXT,
+
+    -- Idempotency and tenant-aware writer fields used by the audit-worker
+    -- TenantAuditWriter (services/audit-worker/adapters/persistence/tenant_audit_writer.go).
+    -- event_id supports ON CONFLICT idempotent inserts.
+    event_id VARCHAR(100),
+    schema_name VARCHAR(100),
+    correlation_id VARCHAR(255),
+    causation_id VARCHAR(255),
+    idempotency_key VARCHAR(255)
+);
+
+-- Indexes for efficient audit queries
+CREATE INDEX IF NOT EXISTS idx_audit_log_table_name ON audit_log(table_name);
+CREATE INDEX IF NOT EXISTS idx_audit_log_record_id ON audit_log(record_id);
+CREATE INDEX IF NOT EXISTS idx_audit_log_created_at ON audit_log(created_at);
+CREATE INDEX IF NOT EXISTS idx_audit_log_changed_by ON audit_log(changed_by);
+CREATE INDEX IF NOT EXISTS idx_audit_log_operation ON audit_log(operation);
+-- Unique index on event_id supports ON CONFLICT for idempotent writes from
+-- the Kafka consumer path. CockroachDB treats NULLs as distinct in UNIQUE
+-- indexes, so outbox-path rows that leave event_id NULL coexist with
+-- Kafka-path rows that populate it (matches services/party/migrations/20260323*).
+CREATE UNIQUE INDEX IF NOT EXISTS idx_audit_log_event_id ON audit_log(event_id);
+
+-- Create audit_outbox table for async processing (unqualified, singular)
+-- GORM hooks write to outbox, background worker moves to audit_log
+CREATE TABLE IF NOT EXISTS audit_outbox (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+
+    -- What changed
+    table_name VARCHAR(100) NOT NULL,
+    operation VARCHAR(10) NOT NULL CHECK (operation IN ('INSERT', 'UPDATE', 'DELETE', 'INITIAL_IMPORT')),
+
+    -- Record identification
+    record_id UUID NOT NULL,
+
+    -- Change details (TEXT to match GORM model - contains JSON strings)
+    old_values TEXT,
+    new_values TEXT,
+
+    -- Processing status (includes 'completed' for successful processing)
+    status VARCHAR(20) NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'processing', 'completed', 'failed')),
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    retry_count INT NOT NULL DEFAULT 0,
+    last_error TEXT,
+
+    -- Additional context
+    changed_by VARCHAR(100),
+    transaction_id VARCHAR(100),
+    client_ip VARCHAR(45),
+    user_agent TEXT
+);
+
+-- Index for worker to efficiently find pending entries
+CREATE INDEX IF NOT EXISTS idx_audit_outbox_status_created ON audit_outbox(status, created_at);

services/identity/migrations/atlas.sum

@@ -1,4 +1,4 @@
-h1:HbreHZ7DfYjRRTYEAvjljVJTg+KizbMvh1A0wMA0WcU=
+h1:c25TZhcUPvpzfYUa0V6OPPg1/HQbd7kkUtN8519OZdM=
 20260302000001_initial.sql h1:SsQN4cGTrm/6qs12GD9YXoSO6QyKTIpNZrM+4rRBWJA=
 20260326000001_add_tenant_id.sql h1:+/NF9LBhT7EjKVqu5gxY1B+vJ+YDM9acG8lLSXGJEWs=
 20260326000002_add_tenant_id_indexes.sql h1:Ov5Wfsl+MTIrtNdnqdh9p5qGpaeBW/wBPXILN4JvuvM=
@@ -8,3 +8,4 @@ h1:HbreHZ7DfYjRRTYEAvjljVJTg+KizbMvh1A0wMA0WcU=
 20260327000004_password_reset_tokens.sql h1:KbLBrcNV5fCjyj04r6YNETLkNXg1dwSUGL4ZHk1pqqQ=
 20260327000005_drop_role_constraint.sql h1:d4mxJFQHkcJEtDEZ+ztR+U1j7Z47eud+KorVVr50TUQ=
 20260327000006_add_role_constraint.sql h1:WAiUQF0nKzDyrc506KuT1RGEkVeWbMEIfPYd/c+gfuM=
+20260425000001_audit_system.sql h1:1LjVA/FeCTwjkuQqroCA03dFlmClLKxd/1VAwJjFduo=

services/tenant/provisioner/mock_provisioner.go

@@ -42,6 +42,10 @@ type MockProvisioner struct {
 	// PurgeCalls tracks calls to PurgeSchemas for verification
 	PurgeCalls []tenant.TenantID
 
+	// ReconciliationCalls tracks calls to ReconcileMigrations for verification.
+	// Each entry is the tenantID argument; nil entries indicate "all tenants" reconciliation.
+	ReconciliationCalls []*tenant.TenantID
+
 	// DataRetentionPeriod for testing retention enforcement
 	DataRetentionPeriod time.Duration
 
@@ -63,6 +67,7 @@ func NewMockProvisioner(services []ServiceConfig) *MockProvisioner {
 		ProvisioningCalls:     make([]tenant.TenantID, 0),
 		DeprovisioningCalls:   make([]tenant.TenantID, 0),
 		PurgeCalls:            make([]tenant.TenantID, 0),
+		ReconciliationCalls:   make([]*tenant.TenantID, 0),
 		DataRetentionPeriod:   0, // No retention period by default for testing
 	}
 }
@@ -271,6 +276,7 @@ func (m *MockProvisioner) Reset() {
 	m.ProvisioningCalls = make([]tenant.TenantID, 0)
 	m.DeprovisioningCalls = make([]tenant.TenantID, 0)
 	m.PurgeCalls = make([]tenant.TenantID, 0)
+	m.ReconciliationCalls = make([]*tenant.TenantID, 0)
 	m.FailProvisioningFor = make(map[string]error)
 	m.FailDeprovisioningFor = make(map[string]error)
 	m.FailPurgeFor = make(map[string]error)
@@ -288,11 +294,21 @@ func (m *MockProvisioner) SetStatus(status *ProvisioningStatus) {
 
 // ReconcileMigrations simulates reconciling migrations for existing tenants.
 // In the mock, this is a no-op that returns success. Tests can verify calls
-// via ReconciliationCalls if needed.
+// via ReconciliationCalls.
 func (m *MockProvisioner) ReconcileMigrations(_ context.Context, tenantID *tenant.TenantID) (int, []string) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
+	// Record the call for test verification. Copy the tenant ID so tests can
+	// distinguish between separate "single tenant" calls without the caller's
+	// pointer aliasing affecting recorded history.
+	if tenantID != nil {
+		copied := *tenantID
+		m.ReconciliationCalls = append(m.ReconciliationCalls, &copied)
+	} else {
+		m.ReconciliationCalls = append(m.ReconciliationCalls, nil)
+	}
+
 	if tenantID != nil {
 		// Single tenant reconciliation
 		if status, exists := m.statuses[tenantID.String()]; exists && status.State == StateActive {

services/tenant/worker/provisioning_worker.go

@@ -150,8 +150,16 @@ func applyConfigDefaults(config Config) Config {
 // It runs until ctx is cancelled or Stop() is called.
 // The method blocks and should be run in a separate goroutine.
 //
-// On startup, it recovers any tenants that were stuck in PROVISIONING status
-// from a previous worker crash. This ensures they get re-queued for provisioning.
+// On startup, it performs two best-effort passes before entering the polling loop:
+//  1. Recover any tenants stuck in PROVISIONING status from a previous worker crash,
+//     re-queuing them for provisioning.
+//  2. Reconcile migrations against all active tenants, applying any new migration
+//     files that were added since each tenant was originally provisioned. This
+//     allows new schema additions (e.g. audit tables) to roll out to existing
+//     tenants on the next deploy without requiring a manual gRPC trigger.
+//
+// Both passes log errors but never block the worker from starting - a failed
+// reconciliation should not stop the worker from processing pending tenants.
 func (w *ProvisioningWorker) Start(ctx context.Context) {
 	// Recover any tenants stuck in PROVISIONING status from previous worker crash.
 	// This is best-effort - we log errors but continue starting the worker.
@@ -162,6 +170,12 @@ func (w *ProvisioningWorker) Start(ctx context.Context) {
 		w.logger.Info("startup recovery completed", "recovered_count", recoveredCount)
 	}
 
+	// Reconcile migrations across all active tenants. This applies new migration
+	// files (e.g. the identity audit_log/audit_outbox tables) to schemas that
+	// were provisioned before those migrations existed. Best-effort - per-tenant
+	// errors are logged but do not prevent the worker from starting.
+	w.reconcileMigrationsOnStartup(ctx)
+
 	ticker := time.NewTicker(w.pollInterval)
 	defer ticker.Stop()
 
@@ -597,6 +611,36 @@ var permanentPatterns = []string{
 	"deprovisioned",  // Tenant is deprovisioned
 }
 
+// reconcileMigrationsOnStartup invokes the provisioner's ReconcileMigrations against
+// all active tenants. This applies any new migration files that were added since the
+// tenant was originally provisioned.
+//
+// Best-effort: per-tenant errors are logged but never propagated. The method panics
+// recovery so a misbehaving provisioner cannot prevent worker startup.
+func (w *ProvisioningWorker) reconcileMigrationsOnStartup(ctx context.Context) {
+	defer func() {
+		if r := recover(); r != nil {
+			w.logger.Error("panic during startup migration reconciliation",
+				"panic", r)
+		}
+	}()
+
+	w.logger.Info("starting migration reconciliation for active tenants")
+
+	reconciledCount, errs := w.provisioner.ReconcileMigrations(ctx, nil)
+
+	if len(errs) > 0 {
+		w.logger.Warn("startup migration reconciliation completed with errors",
+			"reconciled_count", reconciledCount,
+			"error_count", len(errs),
+			"errors", errs)
+		return
+	}
+
+	w.logger.Info("startup migration reconciliation completed",
+		"reconciled_count", reconciledCount)
+}
+
 // RecoverStuckTenants resets tenants that have been stuck in PROVISIONING status
 // for longer than the specified threshold back to PROVISIONING_PENDING status.
 // This allows them to be picked up and re-provisioned on the next polling cycle.

services/tenant/worker/provisioning_worker_test.go

@@ -778,10 +778,11 @@ var (
 
 // ControlledMockProvisioner is a mock that allows controlled failure/success sequences.
 type ControlledMockProvisioner struct {
-	mu              sync.Mutex
-	calls           []tenant.TenantID
-	failureSequence []error // nil = success, error = fail with this error
-	callIndex       int
+	mu                  sync.Mutex
+	calls               []tenant.TenantID
+	failureSequence     []error // nil = success, error = fail with this error
+	callIndex           int
+	reconciliationCalls int // Tracks ReconcileMigrations invocations
 }
 
 func (m *ControlledMockProvisioner) ProvisionSchemas(_ context.Context, tenantID tenant.TenantID) error {
@@ -813,9 +814,20 @@ func (m *ControlledMockProvisioner) GetProvisioningStatus(_ context.Context, _ t
 }
 
 func (m *ControlledMockProvisioner) ReconcileMigrations(_ context.Context, _ *tenant.TenantID) (int, []string) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.reconciliationCalls++
 	return 0, nil
 }
 
+// GetReconciliationCallCount returns the number of times ReconcileMigrations
+// has been invoked. Used to verify worker startup wiring.
+func (m *ControlledMockProvisioner) GetReconciliationCallCount() int {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.reconciliationCalls
+}
+
 func (m *ControlledMockProvisioner) GetRequiredSchemas() []string {
 	return []string{"party", "current-account"}
 }
@@ -1993,3 +2005,106 @@ func TestNewProvisioningWorker_RecoveryThresholdCustom(t *testing.T) {
 	require.NotNil(t, worker)
 	assert.Equal(t, customThreshold, worker.recoveryThreshold, "recoveryThreshold should be set to custom value")
 }
+
+// =============================================================================
+// Startup Reconciliation Tests
+// =============================================================================
+
+// TestProvisioningWorker_StartupReconciliation verifies that worker.Start triggers
+// migration reconciliation against all active tenants. This ensures new migrations
+// (e.g., the identity audit_log/audit_outbox tables) are applied to existing tenant
+// schemas without requiring a manual gRPC trigger.
+func TestProvisioningWorker_StartupReconciliation(t *testing.T) {
+	_, repo := setupTestDB(t)
+
+	mockProv := &ControlledMockProvisioner{}
+	logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug}))
+	config := Config{
+		PollInterval:      100 * time.Millisecond,
+		RecoveryThreshold: 5 * time.Minute,
+	}
+	worker, err := NewProvisioningWorker(repo, mockProv, config, logger)
+	require.NoError(t, err)
+
+	workerCtx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		worker.Start(workerCtx)
+		close(done)
+	}()
+
+	// Reconciliation must run during Start, before the polling loop. Wait briefly
+	// for the call to land - the operation is synchronous within Start so this
+	// poll is checking ordering, not racing against rate-limited work.
+	err = await.AtMost(2 * time.Second).PollInterval(10 * time.Millisecond).Until(func() bool {
+		return mockProv.GetReconciliationCallCount() >= 1
+	})
+	require.NoError(t, err, "Start should invoke ReconcileMigrations exactly once")
+
+	cancel()
+	worker.Stop()
+	<-done
+
+	// Reconciliation should have run exactly once - on startup, not per poll.
+	assert.Equal(t, 1, mockProv.GetReconciliationCallCount(),
+		"reconciliation should run only on startup, not on every poll cycle")
+}
+
+// TestProvisioningWorker_StartupReconciliation_NonFatalErrors verifies that
+// reconciliation errors are logged but do NOT prevent the worker from starting.
+// A failing reconciliation is best-effort: it should not stop the worker from
+// processing pending tenants.
+func TestProvisioningWorker_StartupReconciliation_NonFatalErrors(t *testing.T) {
+	_, repo := setupTestDB(t)
+	ctx := context.Background()
+
+	// Create a pending tenant so we can verify the worker still processes it
+	// after reconciliation.
+	pendingTenant := &domain.Tenant{
+		ID:              tenant.MustNewTenantID("post_reconcile_tenant"),
+		DisplayName:     "Post Reconcile",
+		SettlementAsset: "GBP",
+		Status:          domain.StatusProvisioningPending,
+		CreatedAt:       time.Now(),
+		Version:         1,
+	}
+	require.NoError(t, repo.Create(ctx, pendingTenant))
+
+	// Use a real MockProvisioner so reconciliation returns valid (empty) results
+	// while ProvisionSchemas succeeds for the pending tenant.
+	mockProv := provisioner.NewMockProvisioner(nil)
+
+	var logBuffer safeBuffer
+	logger := slog.New(slog.NewTextHandler(&logBuffer, &slog.HandlerOptions{Level: slog.LevelDebug}))
+	config := Config{
+		PollInterval:      50 * time.Millisecond,
+		RecoveryThreshold: 5 * time.Minute,
+	}
+	worker, err := NewProvisioningWorker(repo, mockProv, config, logger)
+	require.NoError(t, err)
+
+	workerCtx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		worker.Start(workerCtx)
+		close(done)
+	}()
+
+	// The pending tenant should still be provisioned despite the reconciliation
+	// pass running first - proving startup isn't blocked by reconciliation.
+	err = await.AtMost(3 * time.Second).PollInterval(20 * time.Millisecond).Until(func() bool {
+		t, _ := repo.GetByID(ctx, pendingTenant.ID)
+		return t != nil && t.Status == domain.StatusActive
+	})
+	require.NoError(t, err, "pending tenant should be provisioned after startup reconciliation")
+
+	cancel()
+	worker.Stop()
+	<-done
+
+	// Reconciliation must have been called.
+	assert.Len(t, mockProv.ReconciliationCalls, 1,
+		"reconciliation should have been called once on startup")
+	assert.Nil(t, mockProv.ReconciliationCalls[0],
+		"startup reconciliation targets all tenants (nil tenantID)")
+}