Plan: Platform Stabilization Phase 2 — Concurrency & Atomicity Hardening

-- services/<svc>/migrations/<ts>_idempotency_atomic_claim.sql
-- Adds the claim-state machine to idempotency_responses.
-- Additive: existing rows from Step 10 are back-populated as 'succeeded'
-- because the only way they exist on this table is the legacy code path
-- which only INSERTed on a successful response.
ALTER TABLE idempotency_responses
    ADD COLUMN status        TEXT NOT NULL DEFAULT 'succeeded'
        CHECK (status IN ('processing', 'succeeded', 'failed')),
    ADD COLUMN started_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
    ADD COLUMN finished_at   TIMESTAMPTZ,
    -- Per-row claim TTL: a 'processing' row older than this may be
    -- recovered (the original requester died mid-handler). 30s ceiling
    -- bounds loser-poll wait + claim-recovery without bouncing to 409.
    ADD COLUMN claim_expires_at TIMESTAMPTZ NOT NULL
        DEFAULT (now() + INTERVAL '30 seconds'),
    -- Loosen NOT NULL constraints on response columns so we can persist
    -- the row at claim time before the handler has produced a response.
    ALTER COLUMN status_code   DROP NOT NULL,
    ALTER COLUMN headers       DROP NOT NULL,
    ALTER COLUMN body_fingerprint DROP NOT NULL;
    -- body, body_too_large already nullable.

CREATE INDEX idx_idempotency_responses_processing
    ON idempotency_responses (claim_expires_at)
    WHERE status = 'processing';

/// Outcome of attempting to claim a cache slot.
enum ClaimOutcome {
    /// We are the winner; run the handler then call `finalize_*`.
    Claimed,
    /// Another request reached the slot first. Carries the row.
    Existing(CachedRow),
}

async fn try_claim(
    pool: &PgPool,
    cache_key: &str,
    method: &str,
    path: &str,
    user_sub: uuid::Uuid,
    body_fp: &str,
) -> Result<ClaimOutcome, sqlx::Error>;

/// Finalize a winning claim with a successful handler response.
async fn finalize_succeeded(
    pool: &PgPool,
    cache_key: &str,
    status: u16,
    headers: serde_json::Value,
    body_bytes: Bytes,
) -> Result<Bytes, sqlx::Error>;

/// Finalize a winning claim where the handler errored. Marks 'failed'
/// so a retry from the *same* idempotency-key may re-claim.
async fn finalize_failed(
    pool: &PgPool,
    cache_key: &str,
    error_detail: &str,
) -> Result<(), sqlx::Error>;

/// Loser-side wait: poll the slot until status flips to 'succeeded' or
/// the per-request ceiling expires. Returns the replay response on
/// success, or None on ceiling timeout (caller emits 409 Retry-After).
async fn await_winner(
    pool: &PgPool,
    cache_key: &str,
    body_fp: &str,
    ceiling: Duration,
) -> Result<Option<Response>, sqlx::Error>;

INSERT INTO idempotency_responses
    (cache_key, method, path, user_sub, body_fingerprint, status, started_at)
VALUES ($1, $2, $3, $4, $5, 'processing', now())
ON CONFLICT (cache_key) DO NOTHING
RETURNING cache_key

UPDATE idempotency_responses
   SET status            = 'processing',
       started_at        = now(),
       claim_expires_at  = now() + INTERVAL '30 seconds',
       finished_at       = NULL,
       status_code       = NULL,
       headers           = NULL,
       body              = NULL,
       body_too_large    = false,
       body_fingerprint  = $2
 WHERE cache_key = $1
   AND status IN ('failed')
RETURNING cache_key

SELECT id, envelope FROM event_outbox
 WHERE published_at IS NULL
 ORDER BY created_at
 LIMIT $1
 FOR UPDATE SKIP LOCKED

pub async fn drain_once(&self) -> Result<usize, sqlx::Error> {
    let mut tx = self.pool.begin().await?;

    let rows: Vec<(Uuid, serde_json::Value)> = sqlx::query_as(
        r#"SELECT id, envelope FROM event_outbox
           WHERE published_at IS NULL
           ORDER BY created_at
           LIMIT $1
           FOR UPDATE SKIP LOCKED"#,
    )
    .bind(BATCH_SIZE)
    .fetch_all(&mut *tx)
    .await?;

    if rows.is_empty() {
        return Ok(0);
    }

    let mut published = 0usize;
    for (id, envelope_json) in rows {
        // … rehydrate envelope (existing logic) …
        match self.publisher.publish(&envelope).await {
            Ok(()) => {
                sqlx::query(
                    r#"UPDATE event_outbox
                          SET published_at = now(),
                              last_error = NULL
                        WHERE id = $1"#,
                )
                .bind(id)
                .execute(&mut *tx)
                .await?;
                published += 1;
            }
            Err(e) => {
                let msg = e.to_string();
                sqlx::query(
                    r#"UPDATE event_outbox
                          SET attempts = attempts + 1,
                              last_error = $2
                        WHERE id = $1"#,
                )
                .bind(id)
                .bind(&msg)
                .execute(&mut *tx)
                .await?;
                warn!(outbox_id = %id, error = %msg, "outbox publish failed");
            }
        }
    }

    tx.commit().await?;
    Ok(published)
}

-- services/<svc>/migrations/<ts>_inbox_retry_semantics.sql
-- Adds error-tracking + cap fields to event_inbox.
ALTER TABLE event_inbox
    ADD COLUMN error_count  INT NOT NULL DEFAULT 0,
    ADD COLUMN last_error   TEXT,
    ADD COLUMN failed_at    TIMESTAMPTZ,
    ADD COLUMN attempt_started_at TIMESTAMPTZ;

CREATE INDEX idx_event_inbox_in_flight
    ON event_inbox (received_at)
    WHERE processed_at IS NULL AND failed_at IS NULL;

pub const INBOX_MAX_RETRIES: i32 = 5;
const INBOX_BACKOFF_BASE: Duration = Duration::from_secs(1);
const INBOX_BACKOFF_CAP: Duration = Duration::from_secs(60);

pub async fn handle_idempotently<F, Fut>(
    db: &PgPool,
    publisher: &Publisher,        // ← NEW: needed for DLX surface
    envelope: EventEnvelope,
    handler: F,
) -> anyhow::Result<()>
where
    F: Fn(EventEnvelope) -> Fut + Send + Sync,    // Fn (not FnOnce) — may run twice
    Fut: std::future::Future<Output = anyhow::Result<()>> + Send;

impl Publisher {
    pub async fn publish_dlx(
        &self,
        envelope: &EventEnvelope,
        original_queue: &str,
        retry_count: i32,
        last_error: &str,
    ) -> Result<(), lapin::Error>;
}

let object_id = craig_common::id::new_id();
let file_size = data.len() as i64;
let object_key = format!("cases/contacts/{case_id}/{contact_id}/{object_id}/{safe_name}");

// Phase 1 — metadata + event in tx, blob NOT yet written.
let mut tx = app.db.inner().begin().await.map_err(ApiError::internal)?;
let attachment = store::contact_attachments::create_attachment(
    &mut *tx,
    store::contact_attachments::CreateAttachmentParams {
        contact_id, attachment_type, file_name: safe_name.clone(),
        content_type: content_type.clone(), file_size,
        object_key: object_key.clone(),
        uploaded_by: claims.sub.clone(),
        worker_name: worker_name.clone(),
        // object_status defaults to 'pending' in SQL
    },
)
.await
.map_err(ApiError::internal)?;
events::publish_contact_attachment_uploaded(
    &mut tx, attachment.id, contact_id, &attachment.file_name, attachment.file_size,
)
.await
.map_err(ApiError::internal)?;
tx.commit().await.map_err(ApiError::internal)?;

// Phase 2 — blob write + status promotion. Compensate on failure.
match object_store.put(&object_key, data).await {
    Ok(()) => {
        store::contact_attachments::mark_object_present(app.db.inner(), attachment.id)
            .await
            .map_err(ApiError::internal)?;
    }
    Err(blob_err) => {
        tracing::warn!(
            attachment_id = %attachment.id,
            error = %blob_err,
            "blob write failed post-commit; compensating",
        );
        let mut tx = app.db.inner().begin().await.map_err(ApiError::internal)?;
        store::contact_attachments::delete_attachment_pending(&mut *tx, attachment.id)
            .await
            .map_err(ApiError::internal)?;
        events::publish_contact_attachment_upload_failed(
            &mut tx, attachment.id, contact_id, &format!("{blob_err}"),
        )
        .await
        .map_err(ApiError::internal)?;
        tx.commit().await.map_err(ApiError::internal)?;
        return Err(ApiError::from(blob_err));
    }
}
Ok(Json(attachment))

-- services/craig-cases/migrations/<TS>_attachment_object_status.sql
ALTER TABLE contact_attachments
    ADD COLUMN object_status TEXT NOT NULL DEFAULT 'present'
        CHECK (object_status IN ('pending', 'present', 'failed'));
ALTER TABLE contact_attachments
    ALTER COLUMN object_status DROP DEFAULT;

CREATE INDEX idx_contact_attachments_pending
    ON contact_attachments (created_at)
    WHERE object_status = 'pending';

pub struct AttachmentScanner {
    pool: PgPool,
    store: Store,
    grace: Duration,           // default 1 hour
    interval: Duration,        // default 5 min
}

impl AttachmentScanner {
    pub fn spawn(self) -> tokio::task::JoinHandle<()> { /* ... */ }

    /// One scan pass per registered table. For each pending row older than `grace`:
    ///   - object_store.head(&object_key) → 200 ⇒ UPDATE object_status = 'present'
    ///   - head 404 ⇒ DELETE row + publish *.upload_failed
    ///   - head error ⇒ leave row, log, retry next pass
    async fn scan_table(&self, table: &str) -> Result<ScanReport, ScanError> { /* ... */ }
}

-- services/craig-exchange/migrations/<TS>_exchange_send_jobs.sql
CREATE TABLE exchange_send_jobs (
    id              UUID PRIMARY KEY DEFAULT uuidv7(),
    transaction_id  UUID NOT NULL REFERENCES exchange_transactions(id) ON DELETE CASCADE,
    partner_id      UUID NOT NULL,
    correlation_id  UUID NOT NULL DEFAULT uuidv7(),
    payload         JSONB NOT NULL,
    status          TEXT NOT NULL DEFAULT 'pending'
                    CHECK (status IN ('pending', 'in_flight', 'sent', 'failed')),
    attempts        INT NOT NULL DEFAULT 0,
    last_error      TEXT,
    next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    completed_at    TIMESTAMPTZ
);
CREATE INDEX idx_exchange_send_jobs_drainable
    ON exchange_send_jobs (next_attempt_at)
    WHERE status = 'pending';
CREATE UNIQUE INDEX idx_exchange_send_jobs_correlation
    ON exchange_send_jobs (correlation_id);

// services/craig-exchange/src/send_worker.rs
pub const POLL_INTERVAL: Duration = Duration::from_secs(1);
pub const MAX_CONCURRENCY: usize = 8;
pub const MAX_ATTEMPTS: i32 = 6;            // 1 immediate + 5 retries → ~32 min total
pub fn backoff(attempts: i32) -> Duration {
    match attempts { 1 => 5, 2 => 30, 3 => 120, 4 => 300, 5 => 900, _ => 1800 }
        .pipe(Duration::from_secs)
}

pub struct ExchangeSendWorker {
    pool: PgPool,
    publisher: Publisher,
    client: reqwest::Client,
}

impl ExchangeSendWorker {
    pub fn new(pool: PgPool, publisher: Publisher, client: reqwest::Client) -> Self;
    pub fn spawn(self) -> tokio::task::JoinHandle<()>;
    async fn drain_once(&self) -> Result<usize, SendWorkerError>;
    async fn dispatch(&self, job: SendJob) -> Result<(), SendWorkerError>;
}

pub async fn get_placement_for_update<'e, E>(
    executor: E,
    id: Uuid,
) -> Result<Option<Placement>, sqlx::Error>
where
    E: sqlx::Executor<'e, Database = sqlx::Postgres>,
{
    sqlx::query_as::<_, Placement>("SELECT * FROM placements WHERE id = $1 FOR UPDATE")
        .bind(id)
        .fetch_optional(executor)
        .await
}

let mut tx = app.db.begin().await.map_err(ApiError::internal)?;

let current = store::placements::get_placement_for_update(&mut *tx, id)
    .await
    .map_err(ApiError::internal)?
    .ok_or_else(|| ApiError::not_found("placement", id))?;

if let Some(ref new_status) = body.status {
    transitions::validate_placement_transition(&current.status, new_status)
        .map_err(ApiError::bad_request)?;
}

if body.status.as_deref() == Some("ended")
    && let Some(home_id) = current.foster_home_id
{
    let _home = store::foster_homes::get_foster_home_for_update(&mut *tx, home_id)
        .await
        .map_err(ApiError::internal)?
        .ok_or_else(|| ApiError::not_found("foster home", home_id))?;
}

let placement = store::placements::update_placement(&mut *tx, id, /*...*/)
    .await
    .map_err(ApiError::internal)?
    .ok_or_else(|| ApiError::not_found("placement", id))?;

if body.status.as_deref() == Some("ended")
    && let Some(home_id) = current.foster_home_id
{
    store::foster_homes::decrement_occupancy(&mut *tx, home_id)
        .await
        .map_err(ApiError::internal)?;
}

if body.status.as_deref() == Some("ended") {
    let end_reason = placement.end_reason.as_deref().unwrap_or("unspecified");
    events::publish_placement_ended(&mut tx, placement.id, end_reason)
        .await
        .map_err(ApiError::internal)?;
}

tx.commit().await.map_err(ApiError::internal)?;

pub async fn fetch_page<T: serde::de::DeserializeOwned>(
    api: &ApiClient, base_url: &str, path: &str, token: &str,
) -> Result<PageResponse<T>, ApiError> {
    let v = api.get(base_url, path, token).await
        .map_err(|e| ApiError::upstream(path, e))?;
    serde_json::from_value(v)
        .map_err(|e| ApiError::upstream_deserialize(path, e))
}

pub async fn fetch_page_or_empty<T: serde::de::DeserializeOwned + Default>(
    api: &ApiClient, base_url: &str, path: &str, token: &str,
) -> PageResponse<T> {
    fetch_page(api, base_url, path, token).await.unwrap_or_else(|e| {
        tracing::warn!(path, error = %e, "fetch_page_or_empty: returning default");
        PageResponse::default()
    })
}

pub fn validate_upload(
    content_type: &str,
    body: &[u8],
    validation: &UploadValidation<'_>,
) -> Result<(), StoreError>;

pub fn content_disposition_attachment(filename: &str) -> String;

#[serde(default)]
pub require_captcha: bool,

if self.require_captcha && self.captcha_secret == "disabled" {
    return Err(
        "require_captcha=true but captcha_secret is 'disabled' — \
         set CRAIG_INTAKE__CAPTCHA_SECRET to a real secret"
            .into(),
    );
}

let mut tx = app.db.inner().begin().await.map_err(ApiError::internal)?;
let set = store::rule_sets::create_in_tx(&mut tx, /* params */).await?;
let envelope = EventEnvelope::new(
    "craig-rules",
    "rules.cache_invalidated",
    serde_json::json!({"instance_id": engine.instance_id()}),
);
craig_mq::stage_event(&mut tx, &envelope).await
    .map_err(ApiError::internal)?;
tx.commit().await.map_err(ApiError::internal)?;
engine.insert_decision(set.name.clone(), decision).await;

if self.cors_origins.trim() == "*" {
    tracing::warn!(
        "CORS origins set to wildcard '*' — acceptable for public intake \
         (forms embedded on partner sites), but verify this matches the \
         deployment intent."
    );
}

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum EncryptionMode {
    #[default]
    Required,
    Optional,
}

// in ServiceSettings:
#[serde(default)]
pub encryption_mode: EncryptionMode,

let encryptor = match (
    settings.encryption_mode,
    FieldEncryptor::from_env("CRAIG_FIELD_ENCRYPTION_KEY"),
) {
    (EncryptionMode::Required, Ok(enc)) => {
        info!("field encryption enabled (mode=required)");
        Some(enc)
    }
    (EncryptionMode::Required, Err(e)) => {
        anyhow::bail!(
            "encryption_mode=required but CRAIG_FIELD_ENCRYPTION_KEY load failed: {e}"
        );
    }
    (EncryptionMode::Optional, Ok(enc)) => {
        info!("field encryption enabled (mode=optional)");
        Some(enc)
    }
    (EncryptionMode::Optional, Err(_)) => {
        info!("field encryption disabled (mode=optional, no key set)");
        None
    }
};

pub fn encrypt_field(
    encryptor: Option<&FieldEncryptor>,
    mode: EncryptionMode,
    value: &str,
) -> Result<String, ApiError> {
    match (encryptor, mode) {
        (Some(enc), _) => enc.encrypt_str(value).map_err(...),
        (None, EncryptionMode::Required) => Err(ApiError::internal(
            "encryption_mode=required but no encryptor available — startup invariant violated",
        )),
        (None, EncryptionMode::Optional) => Ok(value.to_string()),
    }
}

.layer(SetResponseHeaderLayer::overriding(
    HeaderName::from_static("strict-transport-security"),
    HeaderValue::from_static("max-age=31536000; includeSubDomains"),
))

use std::time::Duration;

pub const DEFAULT_PARTNER_RATE_LIMIT_RPM: u32 = 60;
pub const TOKEN_REFRESH_LEAD_TIME: Duration = Duration::from_secs(30);
// ... add as audit progresses

if let Some(doc) = api_doc {
    // Swagger UI + OpenAPI spec are intentionally public (Kerckhoffs):
    // every protected endpoint still requires a valid Bearer JWT through
    // the auth_middleware on /v1. Discoverable schemas do not weaken auth.
    // Decision: 2026-05-03 (per platform-stabilization-2 D10.6).
    router = router.merge(SwaggerUi::new("/swagger-ui").url("/api-doc/openapi.json", doc));
}