Plan: Security Audit Subscriber

-- services/canopy-security/migrations/20260326000000_create_security_tables.sql

-- Every event captured from canopy.events, parsed into structured audit fields.
CREATE TABLE audit_events (
    id UUID PRIMARY KEY,
    event_id UUID NOT NULL,             -- original EventEnvelope.id
    event_type TEXT NOT NULL,           -- e.g., "determination.completed"
    source_service TEXT NOT NULL,       -- e.g., "canopy-eligibility"
    action TEXT NOT NULL,               -- e.g., "create", "read", "update", "delete", "determine"
    resource_type TEXT NOT NULL,        -- e.g., "person", "household", "determination"
    resource_id TEXT,                   -- ID of the affected resource (may be null for bulk ops)
    user_id TEXT,                       -- user who triggered the action (from JWT claims in payload)
    user_role TEXT,                     -- role of the user at time of action
    ip_address TEXT,                    -- source IP if available in payload
    metadata JSONB NOT NULL DEFAULT '{}',  -- additional parsed fields
    event_timestamp TIMESTAMPTZ NOT NULL,  -- original EventEnvelope.timestamp
    received_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Breach detection alerts generated by detection rules.
CREATE TABLE breach_alerts (
    id UUID PRIMARY KEY,
    rule_name TEXT NOT NULL,            -- e.g., "failed_auth", "bulk_access"
    severity TEXT NOT NULL,             -- critical, high, medium, low
    description TEXT NOT NULL,
    user_id TEXT,                       -- user involved, if applicable
    source_service TEXT,
    evidence JSONB NOT NULL DEFAULT '{}',  -- event IDs and details that triggered the alert
    status TEXT NOT NULL DEFAULT 'open',   -- open, investigating, resolved, false_positive
    resolved_by TEXT,
    resolved_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- NIST SP 800-53 control mapping.
-- Maps Canopy audit events to NIST controls for compliance reporting.
CREATE TABLE nist_control_mappings (
    id UUID PRIMARY KEY,
    control_id TEXT NOT NULL,           -- e.g., "AU-2", "AC-6", "SI-4"
    control_name TEXT NOT NULL,         -- e.g., "Audit Events"
    control_family TEXT NOT NULL,       -- e.g., "Audit and Accountability"
    event_types TEXT[] NOT NULL,        -- event types that satisfy this control
    description TEXT NOT NULL,
    implementation_status TEXT NOT NULL DEFAULT 'planned',  -- planned, partial, implemented
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Detection rules table (configurable, not hardcoded).
CREATE TABLE detection_rules (
    id UUID PRIMARY KEY,
    rule_name TEXT NOT NULL UNIQUE,
    rule_type TEXT NOT NULL,            -- failed_auth, bulk_access, privilege_escalation, after_hours_access
    threshold INTEGER NOT NULL,
    window_minutes INTEGER NOT NULL,
    severity TEXT NOT NULL,             -- low, medium, high, critical
    enabled BOOLEAN NOT NULL DEFAULT true,
    active BOOLEAN NOT NULL DEFAULT true,
    notify_webhook TEXT,                -- optional webhook URL for alert notifications
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Archive table for aged audit events.
CREATE TABLE audit_events_archive (
    LIKE audit_events INCLUDING ALL
);

-- Indexes
CREATE INDEX idx_audit_events_event_type ON audit_events(event_type);
CREATE INDEX idx_audit_events_source_service ON audit_events(source_service);
CREATE INDEX idx_audit_events_action ON audit_events(action);
CREATE INDEX idx_audit_events_user_id ON audit_events(user_id);
CREATE INDEX idx_audit_events_resource_type ON audit_events(resource_type);
CREATE INDEX idx_audit_events_event_timestamp ON audit_events(event_timestamp);
CREATE INDEX idx_audit_events_metadata ON audit_events USING GIN(metadata);
CREATE INDEX idx_breach_alerts_rule_name ON breach_alerts(rule_name);
CREATE INDEX idx_breach_alerts_status ON breach_alerts(status);
CREATE INDEX idx_breach_alerts_severity ON breach_alerts(severity);
CREATE INDEX idx_nist_control_mappings_control_id ON nist_control_mappings(control_id);
CREATE INDEX idx_detection_rules_rule_type ON detection_rules(rule_type);

-- services/canopy-security/migrations/20260326000001_seed_nist_controls.sql

INSERT INTO nist_control_mappings (id, control_id, control_name, control_family, event_types, description, implementation_status) VALUES
(gen_random_uuid(), 'AU-2',  'Audit Events',                    'Audit and Accountability', ARRAY['*'], 'All events captured via wildcard subscriber', 'implemented'),
(gen_random_uuid(), 'AU-3',  'Content of Audit Records',        'Audit and Accountability', ARRAY['*'], 'Event parsing extracts action, resource, user, timestamp', 'implemented'),
(gen_random_uuid(), 'AU-6',  'Audit Review, Analysis, Reporting','Audit and Accountability', ARRAY['*'], 'Breach detection rules analyze audit stream', 'implemented'),
(gen_random_uuid(), 'AU-9',  'Protection of Audit Information', 'Audit and Accountability', ARRAY['*'], 'Audit events stored in dedicated security database', 'implemented'),
(gen_random_uuid(), 'AU-11', 'Audit Record Retention',          'Audit and Accountability', ARRAY['*'], 'Archive management with configurable retention', 'implemented'),
(gen_random_uuid(), 'AC-2',  'Account Management',              'Access Control',           ARRAY['auth.login', 'auth.logout', 'auth.failed'], 'Authentication events tracked', 'planned'),
(gen_random_uuid(), 'AC-6',  'Least Privilege',                 'Access Control',           ARRAY['auth.role_changed', 'auth.permission_granted'], 'Privilege changes tracked', 'planned'),
(gen_random_uuid(), 'AC-7',  'Unsuccessful Login Attempts',     'Access Control',           ARRAY['auth.failed'], 'Failed auth triggers breach detection', 'implemented'),
(gen_random_uuid(), 'SI-4',  'Information System Monitoring',   'System and Info Integrity', ARRAY['*'], 'Continuous monitoring via event stream', 'implemented'),
(gen_random_uuid(), 'IR-4',  'Incident Handling',               'Incident Response',        ARRAY['breach.*'], 'Breach alerts created and tracked', 'implemented'),
(gen_random_uuid(), 'IR-5',  'Incident Monitoring',             'Incident Response',        ARRAY['breach.*'], 'Breach alert status tracking', 'implemented');

-- services/canopy-security/migrations/20260326000002_seed_detection_rules.sql

INSERT INTO detection_rules (id, rule_name, rule_type, threshold, window_minutes, severity) VALUES
(gen_random_uuid(), 'Failed Authentication',    'failed_auth',           5,   10, 'high'),
(gen_random_uuid(), 'Bulk Data Access',         'bulk_access',         100,    5, 'medium'),
(gen_random_uuid(), 'Privilege Escalation',     'privilege_escalation',  1,   60, 'critical'),
(gen_random_uuid(), 'After Hours Access',       'after_hours_access',    1, 1440, 'low');

/// Parsed audit fields extracted from an EventEnvelope.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ParsedAuditEvent {
    pub event_id: Uuid,
    pub event_type: String,
    pub source_service: String,
    pub action: String,
    pub resource_type: String,
    pub resource_id: Option<String>,
    pub user_id: Option<String>,
    pub user_role: Option<String>,
    pub ip_address: Option<String>,
    pub metadata: serde_json::Value,
    pub event_timestamp: DateTime<Utc>,
}

/// Parse an EventEnvelope into structured audit fields.
pub fn parse_event(envelope: &EventEnvelope) -> ParsedAuditEvent {
    let (action, resource_type) = parse_event_type(&envelope.event_type);
    let user_id = extract_string_field(
        &envelope.payload,
        &["user_id", "created_by", "updated_by", "requested_by",
          "accessed_by", "approved_by", "archived_by"],
    );
    let resource_id = extract_string_field(
        &envelope.payload,
        &["id", "resource_id", "person_id", "household_id",
          "application_id", "determination_id"],
    );
    let user_role = extract_string_field(
        &envelope.payload,
        &["role", "user_role"],
    );
    let ip_address = extract_string_field(
        &envelope.payload,
        &["ip_address", "source_ip"],
    );

    ParsedAuditEvent {
        event_id: envelope.id,
        event_type: envelope.event_type.clone(),
        source_service: envelope.source_service.clone(),
        action,
        resource_type,
        resource_id,
        user_id,
        user_role,
        ip_address,
        metadata: envelope.payload.clone(),
        event_timestamp: envelope.timestamp,
    }
}

/// Parse event_type string into (action, resource_type).
/// Explicit mapping for known event types; fallback splits on first dot.
///
/// Ported from d:/code/craig/services/craig-security/src/event_parsing.rs
/// which maps CRAIG-specific event types. Canopy uses these mappings:
pub fn parse_event_type(event_type: &str) -> (String, String) {
    match event_type {
        // Person lifecycle
        "person.created" => ("create".into(), "person".into()),
        "person.updated" => ("update".into(), "person".into()),
        "person.deleted" => ("delete".into(), "person".into()),
        // Household lifecycle
        "household.created" => ("create".into(), "household".into()),
        "household.updated" => ("update".into(), "household".into()),
        "household.member_added" => ("add_member".into(), "household".into()),
        "household.member_removed" => ("remove_member".into(), "household".into()),
        // Application lifecycle
        "application.submitted" => ("submit".into(), "application".into()),
        "application.updated" => ("update".into(), "application".into()),
        "application.withdrawn" => ("withdraw".into(), "application".into()),
        // Determination lifecycle
        "determination.completed" => ("determine".into(), "determination".into()),
        "determination.signed" => ("sign".into(), "determination".into()),
        "determination.verified" => ("verify".into(), "determination".into()),
        // Program-specific determinations
        "snap.determined" => ("determine".into(), "snap_determination".into()),
        "tanf.determined" => ("determine".into(), "tanf_determination".into()),
        "medicaid.determined" => ("determine".into(), "medicaid_determination".into()),
        "chip.determined" => ("determine".into(), "chip_determination".into()),
        // Rules engine
        "rules.evaluated" => ("evaluate".into(), "rule_set".into()),
        "rules.published" => ("publish".into(), "rule_set".into()),
        // Work requirements (TANF)
        "tanf.work_requirement_updated" => ("update".into(), "work_requirement".into()),
        "tanf.time_limit_warning" => ("warn".into(), "time_limit".into()),
        // Auth events
        "auth.login" => ("login".into(), "auth".into()),
        "auth.logout" => ("logout".into(), "auth".into()),
        "auth.failed" => ("fail".into(), "auth".into()),
        "auth.role_changed" => ("change_role".into(), "auth".into()),
        "auth.permission_granted" => ("grant_permission".into(), "auth".into()),
        // Verification
        "verification.requested" => ("request".into(), "verification".into()),
        "verification.completed" => ("complete".into(), "verification".into()),
        // Notices
        "notice.generated" => ("generate".into(), "notice".into()),
        "notice.sent" => ("send".into(), "notice".into()),
        // Security (own events -- avoid infinite loop by not re-inserting)
        ev if ev.starts_with("security.") => ("system".into(), "security".into()),
        // Fallback: split on first dot
        other => {
            match other.split_once('.') {
                Some((resource, action)) => (action.to_string(), resource.to_string()),
                None => ("unknown".to_string(), other.to_string()),
            }
        }
    }
}

/// Extract the first matching string field from a JSON payload.
/// Checks field names in order and returns the first match found.
fn extract_string_field(payload: &serde_json::Value, field_names: &[&str]) -> Option<String> {
    for name in field_names {
        if let Some(serde_json::Value::String(v)) = payload.get(name) {
            return Some(v.clone());
        }
        // Also check UUID-typed fields (some payloads store IDs as UUIDs, not strings)
        if let Some(v) = payload.get(name) {
            if let Some(s) = v.as_str() {
                return Some(s.to_string());
            }
        }
    }
    None
}

/// Detection rule loaded from the detection_rules database table.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct DetectionRule {
    pub id: Uuid,
    pub rule_name: String,
    pub rule_type: String,      // failed_auth, bulk_access, privilege_escalation, after_hours_access
    pub threshold: i32,
    pub window_minutes: i32,
    pub severity: String,       // low, medium, high, critical
    pub enabled: bool,
    pub active: bool,
    pub notify_webhook: Option<String>,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
}

/// Alert generated when a detection rule threshold is exceeded.
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
pub struct SecurityAlert {
    pub id: Uuid,
    pub rule_name: String,
    pub severity: String,
    pub description: String,
    pub user_id: Option<String>,
    pub source_service: Option<String>,
    pub evidence: serde_json::Value,
    pub status: String,
    pub resolved_by: Option<String>,
    pub resolved_at: Option<DateTime<Utc>>,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
}

/// Run all enabled detection rules against the audit log.
/// Returns newly created alerts.
///
/// Ported from d:/code/craig/services/craig-security/src/detection.rs
pub async fn run_detection_scan(
    pool: &PgPool,
    publisher: &Publisher,
) -> Result<Vec<SecurityAlert>, sqlx::Error> {
    let rules = list_enabled_detection_rules(pool).await?;
    let mut alerts = Vec::new();

    for rule in rules.iter().filter(|r| r.enabled && r.active) {
        let count = match rule.rule_type.as_str() {
            "failed_auth" => count_failed_auth(pool, rule.window_minutes).await?,
            "bulk_access" => count_bulk_access(pool, rule.window_minutes).await?,
            "privilege_escalation" => count_privilege_escalation(pool, rule.window_minutes).await?,
            "after_hours_access" => count_after_hours_access(pool, rule.window_minutes).await?,
            _ => 0,
        };

        if count >= rule.threshold as i64 {
            let evidence = serde_json::json!({
                "count": count,
                "window_minutes": rule.window_minutes,
                "threshold": rule.threshold,
                "scanned_at": Utc::now(),
            });
            let alert = create_alert(
                pool,
                &rule.rule_name,
                &rule.severity,
                &format!(
                    "{}: {} occurrences in {} minutes (threshold: {})",
                    rule.rule_name, count, rule.window_minutes, rule.threshold
                ),
                None, // user_id extracted per-rule below
                None, // source_service
                &evidence,
            )
            .await?;

            // Publish security alert event
            let _ = publish_alert_created(publisher, alert.id, &rule.rule_name, &rule.severity).await;

            alerts.push(alert);
        }
    }

    Ok(alerts)
}

/// List all enabled detection rules from the database.
async fn list_enabled_detection_rules(pool: &PgPool) -> Result<Vec<DetectionRule>, sqlx::Error> {
    sqlx::query_as::<_, DetectionRule>(
        "SELECT * FROM detection_rules WHERE enabled = true AND active = true"
    )
    .fetch_all(pool)
    .await
}

/// Count failed auth events in the detection window.
async fn count_failed_auth(pool: &PgPool, window_minutes: i32) -> Result<i64, sqlx::Error> {
    let cutoff = Utc::now() - chrono::Duration::minutes(window_minutes as i64);
    let (count,): (i64,) = sqlx::query_as(
        "SELECT COUNT(*) FROM audit_events
         WHERE event_type = 'auth.failed' AND event_timestamp >= $1"
    )
    .bind(cutoff)
    .fetch_one(pool)
    .await?;
    Ok(count)
}

/// Count bulk data access (>N reads from same user in window).
async fn count_bulk_access(pool: &PgPool, window_minutes: i32) -> Result<i64, sqlx::Error> {
    let cutoff = Utc::now() - chrono::Duration::minutes(window_minutes as i64);
    let (count,): (i64,) = sqlx::query_as(
        "SELECT COALESCE(MAX(user_count), 0) FROM (
            SELECT COUNT(*) AS user_count
            FROM audit_events
            WHERE action IN ('read', 'created') AND event_timestamp >= $1 AND user_id IS NOT NULL
            GROUP BY user_id
        ) sub"
    )
    .bind(cutoff)
    .fetch_one(pool)
    .await?;
    Ok(count)
}

/// Count privilege escalation events in the detection window.
async fn count_privilege_escalation(pool: &PgPool, window_minutes: i32) -> Result<i64, sqlx::Error> {
    let cutoff = Utc::now() - chrono::Duration::minutes(window_minutes as i64);
    let (count,): (i64,) = sqlx::query_as(
        "SELECT COUNT(*) FROM audit_events
         WHERE event_type = 'auth.role_changed' AND event_timestamp >= $1"
    )
    .bind(cutoff)
    .fetch_one(pool)
    .await?;
    Ok(count)
}

/// Count after-hours access events (events outside 06:00-22:00 ET from non-service accounts).
async fn count_after_hours_access(pool: &PgPool, window_minutes: i32) -> Result<i64, sqlx::Error> {
    let cutoff = Utc::now() - chrono::Duration::minutes(window_minutes as i64);
    let (count,): (i64,) = sqlx::query_as(
        "SELECT COUNT(*) FROM audit_events
         WHERE event_timestamp >= $1
           AND user_id IS NOT NULL
           AND user_id NOT LIKE 'service-%'
           AND (EXTRACT(HOUR FROM event_timestamp AT TIME ZONE 'America/New_York') < 6
                OR EXTRACT(HOUR FROM event_timestamp AT TIME ZONE 'America/New_York') >= 22)"
    )
    .bind(cutoff)
    .fetch_one(pool)
    .await?;
    Ok(count)
}

/// Insert a breach alert into the database.
async fn create_alert(
    pool: &PgPool,
    rule_name: &str,
    severity: &str,
    description: &str,
    user_id: Option<&str>,
    source_service: Option<&str>,
    evidence: &serde_json::Value,
) -> Result<SecurityAlert, sqlx::Error> {
    sqlx::query_as::<_, SecurityAlert>(
        "INSERT INTO breach_alerts (id, rule_name, severity, description, user_id, source_service, evidence)
         VALUES ($1, $2, $3, $4, $5, $6, $7)
         RETURNING *"
    )
    .bind(Uuid::now_v7())
    .bind(rule_name)
    .bind(severity)
    .bind(description)
    .bind(user_id)
    .bind(source_service)
    .bind(evidence)
    .fetch_one(pool)
    .await
}

/// Publish a security alert event to canopy.events.
async fn publish_alert_created(
    publisher: &Publisher,
    alert_id: Uuid,
    rule_name: &str,
    severity: &str,
) -> Result<(), lapin::Error> {
    let payload = serde_json::json!({
        "alert_id": alert_id,
        "rule_name": rule_name,
        "severity": severity,
        "created_at": Utc::now(),
    });
    let envelope = EventEnvelope::new("canopy-security", "security.alert_created", payload);
    publisher.publish(&envelope).await
}

// In main.rs, after bootstrap:
let db_for_handler = boot.db.clone();
let publisher_for_handler = publisher.clone();

boot.subscriber
    .subscribe(
        "canopy-security.audit",
        &["#"],  // wildcard -- captures ALL events on canopy.events
        move |envelope: EventEnvelope| {
            let db = db_for_handler.clone();
            let publisher = publisher_for_handler.clone();
            async move {
                handle_inbound_event(&db, &publisher, envelope).await
            }
        },
    )
    .await
    .context("failed to start wildcard event subscriber")?;

/// Handle inbound events from ALL Canopy services -- populates the audit log.
/// Ported from d:/code/craig/services/craig-security/src/main.rs (lines 80-120).
async fn handle_inbound_event(
    db: &DbPool,
    publisher: &Publisher,
    envelope: EventEnvelope,
) -> anyhow::Result<()> {
    // Step 1: Parse the event into structured audit fields
    let parsed = event_parsing::parse_event(&envelope);

    // Step 2: Persist the parsed audit event
    store::audit::insert_audit_event(db, &parsed).await?;

    // Step 3: Run breach detection scan
    // Detection rules are evaluated after each event insert
    // This is acceptable because detection queries are simple count queries against indexed columns
    let alerts = detection::run_detection_scan(db.inner(), publisher).await?;

    if !alerts.is_empty() {
        tracing::warn!(
            alert_count = alerts.len(),
            "breach detection alerts triggered"
        );
    }

    Ok(())
}

-- services/canopy-security/migrations/20260326000000_create_security_tables.sql

CREATE TABLE audit_events (
    id UUID PRIMARY KEY,
    event_id UUID NOT NULL,
    event_type TEXT NOT NULL,
    source_service TEXT NOT NULL,
    action TEXT NOT NULL,
    resource_type TEXT NOT NULL,
    resource_id TEXT,
    user_id TEXT,
    user_role TEXT,
    ip_address TEXT,
    metadata JSONB NOT NULL DEFAULT '{}',
    event_timestamp TIMESTAMPTZ NOT NULL,
    received_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE breach_alerts (
    id UUID PRIMARY KEY,
    rule_name TEXT NOT NULL,
    severity TEXT NOT NULL,
    description TEXT NOT NULL,
    user_id TEXT,
    source_service TEXT,
    evidence JSONB NOT NULL DEFAULT '{}',
    status TEXT NOT NULL DEFAULT 'open',
    resolved_by TEXT,
    resolved_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE nist_control_mappings (
    id UUID PRIMARY KEY,
    control_id TEXT NOT NULL,
    control_name TEXT NOT NULL,
    control_family TEXT NOT NULL,
    event_types TEXT[] NOT NULL,
    description TEXT NOT NULL,
    implementation_status TEXT NOT NULL DEFAULT 'planned',
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE detection_rules (
    id UUID PRIMARY KEY,
    rule_name TEXT NOT NULL UNIQUE,
    rule_type TEXT NOT NULL,
    threshold INTEGER NOT NULL,
    window_minutes INTEGER NOT NULL,
    severity TEXT NOT NULL,
    enabled BOOLEAN NOT NULL DEFAULT true,
    active BOOLEAN NOT NULL DEFAULT true,
    notify_webhook TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE audit_events_archive (
    LIKE audit_events INCLUDING ALL
);

CREATE INDEX idx_audit_events_event_type ON audit_events(event_type);
CREATE INDEX idx_audit_events_source_service ON audit_events(source_service);
CREATE INDEX idx_audit_events_action ON audit_events(action);
CREATE INDEX idx_audit_events_user_id ON audit_events(user_id);
CREATE INDEX idx_audit_events_resource_type ON audit_events(resource_type);
CREATE INDEX idx_audit_events_event_timestamp ON audit_events(event_timestamp);
CREATE INDEX idx_audit_events_metadata ON audit_events USING GIN(metadata);
CREATE INDEX idx_breach_alerts_rule_name ON breach_alerts(rule_name);
CREATE INDEX idx_breach_alerts_status ON breach_alerts(status);
CREATE INDEX idx_breach_alerts_severity ON breach_alerts(severity);
CREATE INDEX idx_nist_control_mappings_control_id ON nist_control_mappings(control_id);
CREATE INDEX idx_detection_rules_rule_type ON detection_rules(rule_type);

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let (
        settings,
        BootstrapResult {
            db, auth, publisher, subscriber, _telemetry, mq_health,
        },
    ) = bootstrap("CANOPY_SECURITY", "canopy-security").await?;

    // 1. Run migrations
    db.run_migrations(&sqlx::migrate!())
        .await
        .context("failed to run database migrations")?;

    // 2. Wire wildcard subscriber for audit logging
    let db_for_sub = db.clone();
    let publisher_for_sub = publisher.clone();
    let _sub_handle = subscriber
        .subscribe(
            "canopy-security.audit",
            &["#"],
            move |envelope: EventEnvelope| {
                let db = db_for_sub.clone();
                let pub_clone = publisher_for_sub.clone();
                async move { handle_inbound_event(&db, &pub_clone, envelope).await }
            },
        )
        .await
        .context("failed to start wildcard event subscriber")?;

    // 3. Build Axum router
    let state = AppState { db, auth };
    let service_routes = api::routes(publisher);
    let router = ApiServer::router(state, service_routes, server_opts, Some(api::ApiDoc::openapi()))
        .layer(Extension(mq_health));

    // 4. Start HTTP server
    ApiServer::serve(router, settings.port, shutdown_signal()).await?;
    Ok(())
}

/// Move audit events older than retention_days to audit_events_archive.
/// Default retention: 365 days for active table; archived records retained 7 years.
pub async fn archive_old_events(
    db: &DbPool,
    retention_days: u32,
) -> Result<u64, sqlx::Error> {
    let cutoff = Utc::now() - chrono::Duration::days(retention_days as i64);

    let mut tx = db.inner().begin().await?;

    let archived = sqlx::query(
        "WITH moved AS (
            DELETE FROM audit_events
            WHERE event_timestamp < $1
            RETURNING *
        )
        INSERT INTO audit_events_archive
        SELECT * FROM moved"
    )
    .bind(cutoff)
    .execute(&mut *tx)
    .await?;

    tx.commit().await?;

    Ok(archived.rows_affected())
}

/// Purge archived records older than purge_years.
pub async fn purge_archived_events(
    db: &DbPool,
    purge_years: u32,
) -> Result<u64, sqlx::Error> {
    let cutoff = Utc::now() - chrono::Duration::days(purge_years as i64 * 365);

    let purged = sqlx::query(
        "DELETE FROM audit_events_archive WHERE event_timestamp < $1"
    )
    .bind(cutoff)
    .execute(db.inner())
    .await?;

    Ok(purged.rows_affected())
}

// services/canopy-security/src/api/audit.rs

/// GET /v1/audit/events
/// Requires security_auditor or security_admin role.
pub async fn list_audit_events(
    claims: Extension<Claims>,
    Query(params): Query<AuditEventFilter>,
    State(state): State<AppState>,
) -> Result<Json<AuditEventPage>, ApiError> {
    claims.require_any_role(&["security_auditor", "security_admin"])?;
    let page = store::audit::query_audit_events(&state.db, params).await.map_err(ApiError::internal)?;
    Ok(Json(page))
}

/// GET /v1/audit/events/{id}
pub async fn get_audit_event(
    claims: Extension<Claims>,
    Path(id): Path<Uuid>,
    State(state): State<AppState>,
) -> Result<Json<AuditEventRow>, ApiError> {
    claims.require_any_role(&["security_auditor", "security_admin"])?;
    match store::audit::get_audit_event(&state.db, id).await.map_err(ApiError::internal)? {
        Some(event) => Ok(Json(event)),
        None => Err(ApiError::not_found("audit_event", id)),
    }
}

/// GET /v1/audit/summary
/// Returns events per service per day, active alerts by severity, NIST coverage.
#[derive(Debug, Serialize)]
pub struct AuditSummary {
    pub events_per_service: Vec<ServiceEventCount>,
    pub active_alerts_by_severity: Vec<SeverityCount>,
    pub nist_coverage: NistCoverage,
}

#[derive(Debug, Serialize, sqlx::FromRow)]
pub struct ServiceEventCount {
    pub source_service: String,
    pub event_count: i64,
}

#[derive(Debug, Serialize, sqlx::FromRow)]
pub struct SeverityCount {
    pub severity: String,
    pub count: i64,
}

#[derive(Debug, Serialize)]
pub struct NistCoverage {
    pub total_controls: i64,
    pub implemented: i64,
    pub partial: i64,
    pub planned: i64,
    pub coverage_percentage: f64,
}

pub async fn audit_summary(
    claims: Extension<Claims>,
    State(state): State<AppState>,
) -> Result<Json<AuditSummary>, ApiError> {
    claims.require_any_role(&["security_auditor", "security_admin"])?;
    // Build summary from aggregate queries
    // ...
    todo!()
}

// services/canopy-security/src/api/alerts.rs

/// GET /v1/audit/alerts
pub async fn list_alerts(
    claims: Extension<Claims>,
    Query(params): Query<AlertFilter>,
    State(state): State<AppState>,
) -> Result<Json<AlertPage>, ApiError> {
    claims.require_any_role(&["security_auditor", "security_admin"])?;
    let page = store::alerts::query_breach_alerts(&state.db, params).await.map_err(ApiError::internal)?;
    Ok(Json(page))
}

/// GET /v1/audit/alerts/{id}
pub async fn get_alert(
    claims: Extension<Claims>,
    Path(id): Path<Uuid>,
    State(state): State<AppState>,
) -> Result<Json<SecurityAlert>, ApiError> {
    claims.require_any_role(&["security_auditor", "security_admin"])?;
    match store::alerts::get_breach_alert(&state.db, id).await.map_err(ApiError::internal)? {
        Some(alert) => Ok(Json(alert)),
        None => Err(ApiError::not_found("breach_alert", id)),
    }
}

/// PATCH /v1/audit/alerts/{id}
/// Requires security_admin role.
pub async fn update_alert(
    claims: Extension<Claims>,
    Path(id): Path<Uuid>,
    State(state): State<AppState>,
    Json(update): Json<AlertStatusUpdate>,
) -> Result<Json<SecurityAlert>, ApiError> {
    claims.require_role("security_admin")?;
    let alert = store::alerts::update_alert_status(&state.db, id, &update)
        .await
        .map_err(ApiError::internal)?;
    Ok(Json(alert))
}

/// GET /v1/audit/nist-controls
pub async fn list_nist_controls(
    claims: Extension<Claims>,
    State(state): State<AppState>,
) -> Result<Json<Vec<NistControlMapping>>, ApiError> {
    claims.require_any_role(&["security_auditor", "security_admin"])?;
    let controls = store::nist::list_nist_controls(&state.db).await.map_err(ApiError::internal)?;
    Ok(Json(controls))
}

{
  "items": [
    {
      "id": "019513a0-0001-7000-8000-000000000001",
      "event_id": "019513a0-0001-7000-8000-000000000099",
      "event_type": "tanf.determined",
      "source_service": "canopy-tanf",
      "action": "determine",
      "resource_type": "tanf_determination",
      "resource_id": "019513a0-0001-7000-8000-000000000050",
      "user_id": "caseworker-jane@agency.gov",
      "user_role": "caseworker",
      "ip_address": "10.0.1.42",
      "metadata": {"application_id": "...", "household_id": "...", "status": "approved"},
      "event_timestamp": "2025-11-15T14:22:33Z",
      "received_at": "2025-11-15T14:22:33.050Z",
      "created_at": "2025-11-15T14:22:33.050Z"
    }
  ],
  "total": 1,
  "page": 0,
  "page_size": 5
}