lib

store.rs (109822B)

---

 #![forbid(unsafe_code)]
 
 use crate::RadrootsOutboxError;
 use crate::migrations::{OUTBOX_MIGRATION_DOWN, OUTBOX_MIGRATION_UP};
 use crate::model::{
     RadrootsOutboxClaimedEvent, RadrootsOutboxEnqueueReceipt, RadrootsOutboxEnqueueStatus,
     RadrootsOutboxEventRecord, RadrootsOutboxEventState, RadrootsOutboxEventStoreIngestReceipt,
     RadrootsOutboxOperationInput, RadrootsOutboxOperationRecord, RadrootsOutboxOperationStatus,
     RadrootsOutboxRelayStatus, RadrootsOutboxRelayStatusRecord, RadrootsOutboxSignedOperationInput,
     RadrootsOutboxStatusSummary,
 };
 use radroots_event_store::{RadrootsEventIngest, RadrootsEventStore};
 use radroots_events::RadrootsNostrEvent;
 use radroots_events::draft::{
     RadrootsFrozenEventDraft, RadrootsSignedNostrEvent, validate_signed_nostr_event_matches_draft,
 };
 use serde::Serialize;
 use sha2::{Digest, Sha256};
 use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions, SqliteQueryResult};
 use sqlx::{Row, SqlitePool};
 use std::path::Path;
 use std::str::FromStr;
 
 #[derive(Clone)]
 pub struct RadrootsOutbox {
     pool: SqlitePool,
 }
 
 impl RadrootsOutbox {
     pub async fn open_memory() -> Result<Self, RadrootsOutboxError> {
         let options = SqliteConnectOptions::from_str("sqlite::memory:")?;
         let pool = SqlitePoolOptions::new()
             .max_connections(1)
             .connect_with(options)
             .await?;
         configure_connection(&pool, false).await?;
         apply_up(&pool).await?;
         Ok(Self { pool })
     }
 
     pub async fn open_file(path: impl AsRef<Path>) -> Result<Self, RadrootsOutboxError> {
         let options = SqliteConnectOptions::new()
             .filename(path)
             .create_if_missing(true);
         let pool = SqlitePoolOptions::new()
             .max_connections(1)
             .connect_with(options)
             .await?;
         configure_connection(&pool, true).await?;
         apply_up(&pool).await?;
         Ok(Self { pool })
     }
 
     pub fn pool(&self) -> &SqlitePool {
         &self.pool
     }
 
     pub async fn migrate_down(&self) -> Result<(), RadrootsOutboxError> {
         apply_down(&self.pool).await
     }
 
     pub async fn pragma_foreign_keys(&self) -> Result<i64, RadrootsOutboxError> {
         query_i64(&self.pool, "PRAGMA foreign_keys").await
     }
 
     pub async fn pragma_busy_timeout(&self) -> Result<i64, RadrootsOutboxError> {
         query_i64(&self.pool, "PRAGMA busy_timeout").await
     }
 
     pub async fn pragma_journal_mode(&self) -> Result<String, RadrootsOutboxError> {
         query_string(&self.pool, "PRAGMA journal_mode").await
     }
 
     pub async fn status_summary(
         &self,
         now_ms: i64,
     ) -> Result<RadrootsOutboxStatusSummary, RadrootsOutboxError> {
         let row = sqlx::query(
             "SELECT COUNT(*) AS total_events, COALESCE(SUM(CASE WHEN state IN ('draft_queued', 'signing', 'signed', 'publishing') THEN 1 ELSE 0 END), 0) AS pending_events, COALESCE(SUM(CASE WHEN state IN ('sign_retryable', 'publish_retryable') THEN 1 ELSE 0 END), 0) AS retryable_events, COALESCE(SUM(CASE WHEN state IN ('published', 'failed_terminal', 'cancelled') THEN 1 ELSE 0 END), 0) AS terminal_events, COALESCE(SUM(CASE WHEN state = 'failed_terminal' THEN 1 ELSE 0 END), 0) AS failed_terminal_events, COALESCE(SUM(CASE WHEN state = 'publishing' THEN 1 ELSE 0 END), 0) AS publishing_events FROM outbox_event",
         )
         .fetch_one(&self.pool)
         .await?;
         let ready_signed_events = sqlx::query(
             "SELECT COUNT(*) FROM outbox_event WHERE state IN ('signed', 'publish_retryable') AND signed_event_json IS NOT NULL AND next_attempt_after_ms <= ? AND (claim_token IS NULL OR claim_expires_at_ms <= ?)",
         )
         .bind(now_ms)
         .bind(now_ms)
         .fetch_one(&self.pool)
         .await?
         .try_get(0)?;
         let last_attempt_at_ms =
             sqlx::query("SELECT MAX(last_attempt_at_ms) FROM outbox_event_relay_status")
                 .fetch_one(&self.pool)
                 .await?
                 .try_get(0)?;
         let last_error = sqlx::query(
             "SELECT last_error FROM outbox_event WHERE last_error IS NOT NULL ORDER BY updated_at_ms DESC, outbox_event_id DESC LIMIT 1",
         )
         .fetch_optional(&self.pool)
         .await?
         .map(|row| row.try_get("last_error"))
         .transpose()?;
         Ok(RadrootsOutboxStatusSummary {
             total_events: row.try_get("total_events")?,
             pending_events: row.try_get("pending_events")?,
             retryable_events: row.try_get("retryable_events")?,
             terminal_events: row.try_get("terminal_events")?,
             failed_terminal_events: row.try_get("failed_terminal_events")?,
             ready_signed_events,
             publishing_events: row.try_get("publishing_events")?,
             last_attempt_at_ms,
             last_error,
         })
     }
 
     pub async fn enqueue_operation(
         &self,
         input: RadrootsOutboxOperationInput,
     ) -> Result<RadrootsOutboxEnqueueReceipt, RadrootsOutboxError> {
         let target_relays = ordered_unique_relays(input.target_relays);
         if target_relays.is_empty() && !input.allow_empty_target_relays {
             return Err(RadrootsOutboxError::EmptyTargetRelays);
         }
         let digest_relays = digest_relays(target_relays.as_slice());
         let digest = idempotency_digest(
             input.operation_kind.as_str(),
             input.draft.expected_pubkey.as_str(),
             &input.draft,
             &digest_relays,
         )?;
         let accepted_quorum = target_relays.len() as i64;
         let mut tx = self.pool.begin().await?;
 
         if let Some(idempotency_key) = input.idempotency_key.as_deref()
             && let Some(existing) = existing_idempotent_operation(
                 &mut tx,
                 input.operation_kind.as_str(),
                 input.draft.expected_pubkey.as_str(),
                 idempotency_key,
             )
             .await?
         {
             if existing.idempotency_digest != digest {
                 return Err(RadrootsOutboxError::IdempotencyConflict {
                     operation_kind: input.operation_kind,
                     expected_pubkey: input.draft.expected_pubkey,
                     idempotency_key: idempotency_key.to_owned(),
                     existing_digest: existing.idempotency_digest,
                     new_digest: digest,
                 });
             }
             tx.commit().await?;
             return Ok(RadrootsOutboxEnqueueReceipt {
                 status: RadrootsOutboxEnqueueStatus::Existing,
                 operation_id: existing.operation_id,
                 outbox_event_id: existing.outbox_event_id,
                 expected_event_id: existing.event_id,
                 idempotency_digest: digest,
             });
         }
 
         let operation = sqlx::query(
             "INSERT INTO outbox_operation(operation_kind, expected_pubkey, idempotency_key, idempotency_digest, status, created_at_ms, updated_at_ms) VALUES (?, ?, ?, ?, ?, ?, ?)",
         )
         .bind(input.operation_kind.as_str())
         .bind(input.draft.expected_pubkey.as_str())
         .bind(input.idempotency_key.as_deref())
         .bind(digest.as_str())
         .bind(RadrootsOutboxOperationStatus::Queued.as_str())
         .bind(input.created_at_ms)
         .bind(input.created_at_ms)
         .execute(&mut *tx)
         .await?;
         let operation_id = operation.last_insert_rowid();
         let draft_json = serde_json::to_string(&input.draft)?;
         let event = sqlx::query(
             "INSERT INTO outbox_event(operation_id, event_id, expected_pubkey, draft_json, state, accepted_quorum, attempt_count, next_attempt_after_ms, event_store_ingested, event_store_inserted, created_at_ms, updated_at_ms) VALUES (?, ?, ?, ?, ?, ?, 0, ?, 0, 0, ?, ?)",
         )
         .bind(operation_id)
         .bind(input.draft.expected_event_id.as_str())
         .bind(input.draft.expected_pubkey.as_str())
         .bind(draft_json.as_str())
         .bind(RadrootsOutboxEventState::DraftQueued.as_str())
         .bind(accepted_quorum)
         .bind(input.created_at_ms)
         .bind(input.created_at_ms)
         .bind(input.created_at_ms)
         .execute(&mut *tx)
         .await?;
         let outbox_event_id = event.last_insert_rowid();
 
         for relay_url in target_relays {
             sqlx::query(
                 "INSERT INTO outbox_event_relay_status(outbox_event_id, relay_url, status, attempt_count) VALUES (?, ?, ?, 0)",
             )
             .bind(outbox_event_id)
             .bind(relay_url.as_str())
             .bind(RadrootsOutboxRelayStatus::Pending.as_str())
             .execute(&mut *tx)
             .await?;
         }
 
         tx.commit().await?;
         Ok(RadrootsOutboxEnqueueReceipt {
             status: RadrootsOutboxEnqueueStatus::Inserted,
             operation_id,
             outbox_event_id,
             expected_event_id: input.draft.expected_event_id,
             idempotency_digest: digest,
         })
     }
 
     pub async fn enqueue_signed_operation(
         &self,
         input: RadrootsOutboxSignedOperationInput,
     ) -> Result<RadrootsOutboxEnqueueReceipt, RadrootsOutboxError> {
         validate_signed_nostr_event_matches_draft(&input.signed_event, &input.draft)?;
         let target_relays = ordered_unique_relays(input.target_relays);
         if target_relays.is_empty() && !input.allow_empty_target_relays {
             return Err(RadrootsOutboxError::EmptyTargetRelays);
         }
         let digest_relays = digest_relays(target_relays.as_slice());
         let digest = idempotency_digest(
             input.operation_kind.as_str(),
             input.draft.expected_pubkey.as_str(),
             &input.draft,
             &digest_relays,
         )?;
         let accepted_quorum = target_relays.len() as i64;
         let mut tx = self.pool.begin().await?;
 
         if let Some(idempotency_key) = input.idempotency_key.as_deref()
             && let Some(existing) = existing_idempotent_operation(
                 &mut tx,
                 input.operation_kind.as_str(),
                 input.draft.expected_pubkey.as_str(),
                 idempotency_key,
             )
             .await?
         {
             if existing.idempotency_digest != digest {
                 return Err(RadrootsOutboxError::IdempotencyConflict {
                     operation_kind: input.operation_kind,
                     expected_pubkey: input.draft.expected_pubkey,
                     idempotency_key: idempotency_key.to_owned(),
                     existing_digest: existing.idempotency_digest,
                     new_digest: digest,
                 });
             }
             tx.commit().await?;
             return Ok(RadrootsOutboxEnqueueReceipt {
                 status: RadrootsOutboxEnqueueStatus::Existing,
                 operation_id: existing.operation_id,
                 outbox_event_id: existing.outbox_event_id,
                 expected_event_id: existing.event_id,
                 idempotency_digest: digest,
             });
         }
 
         let operation = sqlx::query(
             "INSERT INTO outbox_operation(operation_kind, expected_pubkey, idempotency_key, idempotency_digest, status, created_at_ms, updated_at_ms) VALUES (?, ?, ?, ?, ?, ?, ?)",
         )
         .bind(input.operation_kind.as_str())
         .bind(input.draft.expected_pubkey.as_str())
         .bind(input.idempotency_key.as_deref())
         .bind(digest.as_str())
         .bind(RadrootsOutboxOperationStatus::Queued.as_str())
         .bind(input.created_at_ms)
         .bind(input.created_at_ms)
         .execute(&mut *tx)
         .await?;
         let operation_id = operation.last_insert_rowid();
         let draft_json = serde_json::to_string(&input.draft)?;
         let signed_event_json = serde_json::to_string(&input.signed_event)?;
         let event = sqlx::query(
             "INSERT INTO outbox_event(operation_id, event_id, expected_pubkey, draft_json, signed_event_json, raw_event_json, state, accepted_quorum, attempt_count, next_attempt_after_ms, event_store_ingested, event_store_inserted, event_store_ingested_at_ms, created_at_ms, updated_at_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, ?, 1, ?, ?, ?, ?)",
         )
         .bind(operation_id)
         .bind(input.draft.expected_event_id.as_str())
         .bind(input.draft.expected_pubkey.as_str())
         .bind(draft_json.as_str())
         .bind(signed_event_json.as_str())
         .bind(input.signed_event.raw_json.as_str())
         .bind(RadrootsOutboxEventState::Signed.as_str())
         .bind(accepted_quorum)
         .bind(input.created_at_ms)
         .bind(bool_i64(input.event_store_inserted))
         .bind(input.event_store_ingested_at_ms)
         .bind(input.created_at_ms)
         .bind(input.created_at_ms)
         .execute(&mut *tx)
         .await?;
         let outbox_event_id = event.last_insert_rowid();
 
         for relay_url in target_relays {
             sqlx::query(
                 "INSERT INTO outbox_event_relay_status(outbox_event_id, relay_url, status, attempt_count) VALUES (?, ?, ?, 0)",
             )
             .bind(outbox_event_id)
             .bind(relay_url.as_str())
             .bind(RadrootsOutboxRelayStatus::Pending.as_str())
             .execute(&mut *tx)
             .await?;
         }
 
         tx.commit().await?;
         Ok(RadrootsOutboxEnqueueReceipt {
             status: RadrootsOutboxEnqueueStatus::Inserted,
             operation_id,
             outbox_event_id,
             expected_event_id: input.draft.expected_event_id,
             idempotency_digest: digest,
         })
     }
 
     pub async fn get_operation(
         &self,
         operation_id: i64,
     ) -> Result<Option<RadrootsOutboxOperationRecord>, RadrootsOutboxError> {
         let row = sqlx::query(
             "SELECT operation_id, operation_kind, expected_pubkey, idempotency_key, idempotency_digest, status, created_at_ms, updated_at_ms FROM outbox_operation WHERE operation_id = ?",
         )
         .bind(operation_id)
         .fetch_optional(&self.pool)
         .await?;
         row.map(operation_from_row).transpose()
     }
 
     pub async fn get_event(
         &self,
         outbox_event_id: i64,
     ) -> Result<Option<RadrootsOutboxEventRecord>, RadrootsOutboxError> {
         let row = sqlx::query(
             "SELECT outbox_event_id, operation_id, event_id, expected_pubkey, draft_json, signed_event_json, raw_event_json, state, accepted_quorum, attempt_count, claim_token, claim_owner, claim_expires_at_ms, next_attempt_after_ms, last_error, event_store_ingested, event_store_inserted, event_store_ingested_at_ms, created_at_ms, updated_at_ms FROM outbox_event WHERE outbox_event_id = ?",
         )
         .bind(outbox_event_id)
         .fetch_optional(&self.pool)
         .await?;
         row.map(event_from_row).transpose()
     }
 
     pub async fn relay_statuses(
         &self,
         outbox_event_id: i64,
     ) -> Result<Vec<RadrootsOutboxRelayStatusRecord>, RadrootsOutboxError> {
         relay_statuses_for(&self.pool, outbox_event_id).await
     }
 
     pub async fn claim_next_ready_event(
         &self,
         claim_owner: impl AsRef<str>,
         claim_token: impl AsRef<str>,
         claim_expires_at_ms: i64,
         now_ms: i64,
     ) -> Result<Option<RadrootsOutboxClaimedEvent>, RadrootsOutboxError> {
         let mut tx = self.pool.begin().await?;
         let row = sqlx::query(
             "SELECT outbox_event_id, state, signed_event_json FROM outbox_event WHERE state IN ('draft_queued', 'sign_retryable', 'signed', 'publish_retryable') AND next_attempt_after_ms <= ? AND (claim_token IS NULL OR claim_expires_at_ms <= ?) ORDER BY created_at_ms, outbox_event_id LIMIT 1",
         )
         .bind(now_ms)
         .bind(now_ms)
         .fetch_optional(&mut *tx)
         .await?;
         let Some(row) = row else {
             tx.commit().await?;
             return Ok(None);
         };
         let outbox_event_id: i64 = row.try_get("outbox_event_id")?;
         let state = RadrootsOutboxEventState::parse(row.try_get::<String, _>("state")?.as_str())?;
         let signed_event_json: Option<String> = row.try_get("signed_event_json")?;
         let claimed_state = match (state, signed_event_json.as_ref()) {
             (
                 RadrootsOutboxEventState::DraftQueued | RadrootsOutboxEventState::SignRetryable,
                 None,
             ) => RadrootsOutboxEventState::Signing,
             _ => RadrootsOutboxEventState::Publishing,
         };
         let changed = sqlx::query(
             "UPDATE outbox_event SET state = ?, claim_token = ?, claim_owner = ?, claim_expires_at_ms = ?, attempt_count = attempt_count + 1, updated_at_ms = ? WHERE outbox_event_id = ? AND (claim_token IS NULL OR claim_expires_at_ms <= ?)",
         )
         .bind(claimed_state.as_str())
         .bind(claim_token.as_ref())
         .bind(claim_owner.as_ref())
         .bind(claim_expires_at_ms)
         .bind(now_ms)
         .bind(outbox_event_id)
         .bind(now_ms)
         .execute(&mut *tx)
         .await?;
         if changed.rows_affected() == 0 {
             tx.commit().await?;
             return Ok(None);
         }
         let record = event_by_id_tx(&mut tx, outbox_event_id).await?;
         let target_relays = relay_urls_for_tx(&mut tx, outbox_event_id).await?;
         tx.commit().await?;
         Ok(Some(RadrootsOutboxClaimedEvent {
             outbox_event_id: record.outbox_event_id,
             operation_id: record.operation_id,
             expected_event_id: record.event_id,
             attempt_count: record.attempt_count,
             state: claimed_state,
             claim_token: claim_token.as_ref().to_owned(),
             draft: record.draft,
             signed_event: record.signed_event,
             target_relays,
         }))
     }
 
     pub async fn claim_next_ready_signed_event(
         &self,
         claim_owner: impl AsRef<str>,
         claim_token: impl AsRef<str>,
         claim_expires_at_ms: i64,
         now_ms: i64,
     ) -> Result<Option<RadrootsOutboxClaimedEvent>, RadrootsOutboxError> {
         let mut tx = self.pool.begin().await?;
         let row = sqlx::query(
             "SELECT outbox_event_id FROM outbox_event WHERE state IN ('signed', 'publish_retryable') AND signed_event_json IS NOT NULL AND next_attempt_after_ms <= ? AND (claim_token IS NULL OR claim_expires_at_ms <= ?) ORDER BY created_at_ms, outbox_event_id LIMIT 1",
         )
         .bind(now_ms)
         .bind(now_ms)
         .fetch_optional(&mut *tx)
         .await?;
         let Some(row) = row else {
             tx.commit().await?;
             return Ok(None);
         };
         let outbox_event_id: i64 = row.try_get("outbox_event_id")?;
         let changed = sqlx::query(
             "UPDATE outbox_event SET state = ?, claim_token = ?, claim_owner = ?, claim_expires_at_ms = ?, attempt_count = attempt_count + 1, updated_at_ms = ? WHERE outbox_event_id = ? AND state IN ('signed', 'publish_retryable') AND signed_event_json IS NOT NULL AND (claim_token IS NULL OR claim_expires_at_ms <= ?)",
         )
         .bind(RadrootsOutboxEventState::Publishing.as_str())
         .bind(claim_token.as_ref())
         .bind(claim_owner.as_ref())
         .bind(claim_expires_at_ms)
         .bind(now_ms)
         .bind(outbox_event_id)
         .bind(now_ms)
         .execute(&mut *tx)
         .await?;
         if changed.rows_affected() == 0 {
             tx.commit().await?;
             return Ok(None);
         }
         let record = event_by_id_tx(&mut tx, outbox_event_id).await?;
         let target_relays = relay_urls_for_tx(&mut tx, outbox_event_id).await?;
         tx.commit().await?;
         Ok(Some(RadrootsOutboxClaimedEvent {
             outbox_event_id: record.outbox_event_id,
             operation_id: record.operation_id,
             expected_event_id: record.event_id,
             attempt_count: record.attempt_count,
             state: RadrootsOutboxEventState::Publishing,
             claim_token: claim_token.as_ref().to_owned(),
             draft: record.draft,
             signed_event: record.signed_event,
             target_relays,
         }))
     }
 
     pub async fn complete_signing(
         &self,
         outbox_event_id: i64,
         claim_token: &str,
         signed_event: RadrootsSignedNostrEvent,
         now_ms: i64,
     ) -> Result<RadrootsSignedNostrEvent, RadrootsOutboxError> {
         let record = self.claimed_event(outbox_event_id, claim_token).await?;
         if signed_event.id != record.event_id {
             return Err(RadrootsOutboxError::SignedEventIdMismatch {
                 expected_event_id: record.event_id,
                 actual_event_id: signed_event.id,
             });
         }
         let signed_event_json = serde_json::to_string(&signed_event)?;
         let changed = sqlx::query(
             "UPDATE outbox_event SET signed_event_json = ?, raw_event_json = ?, state = ?, last_error = NULL, updated_at_ms = ? WHERE outbox_event_id = ? AND claim_token = ?",
         )
         .bind(signed_event_json.as_str())
         .bind(signed_event.raw_json.as_str())
         .bind(RadrootsOutboxEventState::Signed.as_str())
         .bind(now_ms)
         .bind(outbox_event_id)
         .bind(claim_token)
         .execute(&self.pool)
         .await?;
         self.ensure_claimed_update(outbox_event_id, claim_token, changed)
             .await?;
         Ok(signed_event)
     }
 
     pub async fn mark_sign_retryable(
         &self,
         outbox_event_id: i64,
         claim_token: &str,
         error: impl AsRef<str>,
         next_attempt_after_ms: i64,
         now_ms: i64,
     ) -> Result<(), RadrootsOutboxError> {
         let changed = sqlx::query(
             "UPDATE outbox_event SET state = ?, claim_token = NULL, claim_owner = NULL, claim_expires_at_ms = NULL, last_error = ?, next_attempt_after_ms = ?, updated_at_ms = ? WHERE outbox_event_id = ? AND claim_token = ?",
         )
         .bind(RadrootsOutboxEventState::SignRetryable.as_str())
         .bind(error.as_ref())
         .bind(next_attempt_after_ms)
         .bind(now_ms)
         .bind(outbox_event_id)
         .bind(claim_token)
         .execute(&self.pool)
         .await?;
         self.ensure_claimed_update(outbox_event_id, claim_token, changed)
             .await?;
         Ok(())
     }
 
     pub async fn mark_publish_retryable(
         &self,
         outbox_event_id: i64,
         claim_token: &str,
         error: impl AsRef<str>,
         next_attempt_after_ms: i64,
         now_ms: i64,
     ) -> Result<(), RadrootsOutboxError> {
         let changed = sqlx::query(
             "UPDATE outbox_event SET state = ?, claim_token = NULL, claim_owner = NULL, claim_expires_at_ms = NULL, last_error = ?, next_attempt_after_ms = ?, updated_at_ms = ? WHERE outbox_event_id = ? AND claim_token = ?",
         )
         .bind(RadrootsOutboxEventState::PublishRetryable.as_str())
         .bind(error.as_ref())
         .bind(next_attempt_after_ms)
         .bind(now_ms)
         .bind(outbox_event_id)
         .bind(claim_token)
         .execute(&self.pool)
         .await?;
         self.ensure_claimed_update(outbox_event_id, claim_token, changed)
             .await?;
         Ok(())
     }
 
     pub async fn recover_expired_claims(&self, now_ms: i64) -> Result<u64, RadrootsOutboxError> {
         let changed = sqlx::query(
             "UPDATE outbox_event SET state = CASE WHEN state = 'signing' AND signed_event_json IS NULL THEN 'sign_retryable' WHEN state = 'signing' AND signed_event_json IS NOT NULL THEN 'signed' WHEN state = 'publishing' THEN 'publish_retryable' ELSE state END, claim_token = NULL, claim_owner = NULL, claim_expires_at_ms = NULL, updated_at_ms = ? WHERE claim_token IS NOT NULL AND claim_expires_at_ms <= ? AND state IN ('signing', 'signed', 'publishing')",
         )
         .bind(now_ms)
         .bind(now_ms)
         .execute(&self.pool)
         .await?;
         Ok(changed.rows_affected())
     }
 
     pub async fn ingest_signed_event_local(
         &self,
         event_store: &RadrootsEventStore,
         outbox_event_id: i64,
         claim_token: &str,
         observed_at_ms: i64,
     ) -> Result<RadrootsOutboxEventStoreIngestReceipt, RadrootsOutboxError> {
         let record = self.claimed_event(outbox_event_id, claim_token).await?;
         if record.event_store_ingested {
             return Ok(RadrootsOutboxEventStoreIngestReceipt {
                 outbox_event_id,
                 event_id: record.event_id,
                 already_ingested: true,
                 event_store_inserted: false,
             });
         }
         let signed_event = record
             .signed_event
             .ok_or(RadrootsOutboxError::MissingSignedEvent(outbox_event_id))?;
         let event = event_from_signed(&signed_event);
         let ingest = RadrootsEventIngest::new(event, observed_at_ms)
             .with_raw_json(signed_event.raw_json.clone());
         let receipt = event_store.ingest_event(ingest).await?;
         let changed = sqlx::query(
             "UPDATE outbox_event SET event_store_ingested = 1, event_store_inserted = ?, event_store_ingested_at_ms = ?, state = ?, updated_at_ms = ? WHERE outbox_event_id = ? AND claim_token = ?",
         )
         .bind(bool_i64(receipt.inserted))
         .bind(observed_at_ms)
         .bind(RadrootsOutboxEventState::Publishing.as_str())
         .bind(observed_at_ms)
         .bind(outbox_event_id)
         .bind(claim_token)
         .execute(&self.pool)
         .await?;
         self.ensure_claimed_update(outbox_event_id, claim_token, changed)
             .await?;
         Ok(RadrootsOutboxEventStoreIngestReceipt {
             outbox_event_id,
             event_id: receipt.event_id,
             already_ingested: false,
             event_store_inserted: receipt.inserted,
         })
     }
 
     pub async fn mark_relay_accepted(
         &self,
         outbox_event_id: i64,
         claim_token: &str,
         relay_url: &str,
         acknowledged_at_ms: i64,
     ) -> Result<(), RadrootsOutboxError> {
         let changed = sqlx::query(
             "UPDATE outbox_event_relay_status SET status = ?, attempt_count = attempt_count + 1, last_attempt_at_ms = ?, acknowledged_at_ms = ?, last_error = NULL WHERE outbox_event_id = ? AND relay_url = ? AND EXISTS (SELECT 1 FROM outbox_event WHERE outbox_event_id = ? AND claim_token = ?)",
         )
         .bind(RadrootsOutboxRelayStatus::Accepted.as_str())
         .bind(acknowledged_at_ms)
         .bind(acknowledged_at_ms)
         .bind(outbox_event_id)
         .bind(relay_url)
         .bind(outbox_event_id)
         .bind(claim_token)
         .execute(&self.pool)
         .await?;
         self.ensure_claimed_update(outbox_event_id, claim_token, changed)
             .await?;
         Ok(())
     }
 
     pub async fn set_publish_quorum(
         &self,
         outbox_event_id: i64,
         claim_token: &str,
         accepted_quorum: i64,
         now_ms: i64,
     ) -> Result<(), RadrootsOutboxError> {
         let changed = sqlx::query(
             "UPDATE outbox_event SET accepted_quorum = ?, updated_at_ms = ? WHERE outbox_event_id = ? AND claim_token = ?",
         )
         .bind(accepted_quorum)
         .bind(now_ms)
         .bind(outbox_event_id)
         .bind(claim_token)
         .execute(&self.pool)
         .await?;
         self.ensure_claimed_update(outbox_event_id, claim_token, changed)
             .await?;
         Ok(())
     }
 
     pub async fn mark_relay_failed_retryable(
         &self,
         outbox_event_id: i64,
         claim_token: &str,
         relay_url: &str,
         error: &str,
         attempted_at_ms: i64,
     ) -> Result<(), RadrootsOutboxError> {
         self.mark_relay_failed(
             outbox_event_id,
             claim_token,
             relay_url,
             RadrootsOutboxRelayStatus::FailedRetryable,
             error,
             attempted_at_ms,
         )
         .await
     }
 
     pub async fn mark_relay_failed_terminal(
         &self,
         outbox_event_id: i64,
         claim_token: &str,
         relay_url: &str,
         error: &str,
         attempted_at_ms: i64,
     ) -> Result<(), RadrootsOutboxError> {
         self.mark_relay_failed(
             outbox_event_id,
             claim_token,
             relay_url,
             RadrootsOutboxRelayStatus::FailedTerminal,
             error,
             attempted_at_ms,
         )
         .await
     }
 
     pub async fn complete_publish_attempt(
         &self,
         outbox_event_id: i64,
         claim_token: &str,
         retryable_error: impl AsRef<str>,
         terminal_error: impl AsRef<str>,
         next_attempt_after_ms: i64,
         now_ms: i64,
     ) -> Result<RadrootsOutboxEventState, RadrootsOutboxError> {
         let mut tx = self.pool.begin().await?;
         let row = claimed_event_identity_tx(&mut tx, outbox_event_id, claim_token).await?;
         let operation_id = row.operation_id;
         let accepted_quorum = row.accepted_quorum;
         let accepted_count: i64 = sqlx::query(
             "SELECT COUNT(*) FROM outbox_event_relay_status WHERE outbox_event_id = ? AND status = ?",
         )
         .bind(outbox_event_id)
         .bind(RadrootsOutboxRelayStatus::Accepted.as_str())
         .fetch_one(&mut *tx)
         .await?
         .try_get(0)?;
         let retryable_count: i64 = sqlx::query(
             "SELECT COUNT(*) FROM outbox_event_relay_status WHERE outbox_event_id = ? AND status = ?",
         )
         .bind(outbox_event_id)
         .bind(RadrootsOutboxRelayStatus::FailedRetryable.as_str())
         .fetch_one(&mut *tx)
         .await?
         .try_get(0)?;
         let pending_count: i64 = sqlx::query(
             "SELECT COUNT(*) FROM outbox_event_relay_status WHERE outbox_event_id = ? AND status = ?",
         )
         .bind(outbox_event_id)
         .bind(RadrootsOutboxRelayStatus::Pending.as_str())
         .fetch_one(&mut *tx)
         .await?
         .try_get(0)?;
 
         let (event_state, operation_status, last_error, next_attempt_after_ms) =
             if accepted_count >= accepted_quorum {
                 (
                     RadrootsOutboxEventState::Published,
                     Some(RadrootsOutboxOperationStatus::Complete),
                     None,
                     now_ms,
                 )
             } else if retryable_count > 0 || pending_count > 0 {
                 (
                     RadrootsOutboxEventState::PublishRetryable,
                     None,
                     Some(retryable_error.as_ref()),
                     next_attempt_after_ms,
                 )
             } else {
                 (
                     RadrootsOutboxEventState::FailedTerminal,
                     Some(RadrootsOutboxOperationStatus::FailedTerminal),
                     Some(terminal_error.as_ref()),
                     now_ms,
                 )
             };
 
         let changed = sqlx::query(
             "UPDATE outbox_event SET state = ?, claim_token = NULL, claim_owner = NULL, claim_expires_at_ms = NULL, last_error = ?, next_attempt_after_ms = ?, updated_at_ms = ? WHERE outbox_event_id = ? AND claim_token = ?",
         )
         .bind(event_state.as_str())
         .bind(last_error)
         .bind(next_attempt_after_ms)
         .bind(now_ms)
         .bind(outbox_event_id)
         .bind(claim_token)
         .execute(&mut *tx)
         .await?;
         if changed.rows_affected() == 0 {
             return Err(RadrootsOutboxError::ClaimTokenMismatch { outbox_event_id });
         }
 
         if let Some(operation_status) = operation_status {
             sqlx::query(
                 "UPDATE outbox_operation SET status = ?, updated_at_ms = ? WHERE operation_id = ?",
             )
             .bind(operation_status.as_str())
             .bind(now_ms)
             .bind(operation_id)
             .execute(&mut *tx)
             .await?;
         }
 
         tx.commit().await?;
         Ok(event_state)
     }
 
     pub async fn mark_publish_failed_terminal(
         &self,
         outbox_event_id: i64,
         claim_token: &str,
         error: impl AsRef<str>,
         now_ms: i64,
     ) -> Result<(), RadrootsOutboxError> {
         self.finish_claimed_event(
             outbox_event_id,
             claim_token,
             RadrootsOutboxEventState::FailedTerminal,
             RadrootsOutboxOperationStatus::FailedTerminal,
             Some(error.as_ref()),
             now_ms,
         )
         .await
     }
 
     pub async fn cancel_claimed_event(
         &self,
         outbox_event_id: i64,
         claim_token: &str,
         now_ms: i64,
     ) -> Result<(), RadrootsOutboxError> {
         self.finish_claimed_event(
             outbox_event_id,
             claim_token,
             RadrootsOutboxEventState::Cancelled,
             RadrootsOutboxOperationStatus::Cancelled,
             None,
             now_ms,
         )
         .await
     }
 
     async fn claimed_event(
         &self,
         outbox_event_id: i64,
         claim_token: &str,
     ) -> Result<RadrootsOutboxEventRecord, RadrootsOutboxError> {
         self.ensure_claim_token(outbox_event_id, claim_token)
             .await?;
         self.get_event(outbox_event_id)
             .await?
             .ok_or(RadrootsOutboxError::EventNotFound(outbox_event_id))
     }
 
     async fn ensure_claim_token(
         &self,
         outbox_event_id: i64,
         claim_token: &str,
     ) -> Result<(), RadrootsOutboxError> {
         let row = sqlx::query("SELECT claim_token FROM outbox_event WHERE outbox_event_id = ?")
             .bind(outbox_event_id)
             .fetch_optional(&self.pool)
             .await?;
         let Some(row) = row else {
             return Err(RadrootsOutboxError::EventNotFound(outbox_event_id));
         };
         let stored: Option<String> = row.try_get("claim_token")?;
         if stored.as_deref() != Some(claim_token) {
             return Err(RadrootsOutboxError::ClaimTokenMismatch { outbox_event_id });
         }
         Ok(())
     }
 
     async fn ensure_claimed_update(
         &self,
         outbox_event_id: i64,
         claim_token: &str,
         changed: SqliteQueryResult,
     ) -> Result<(), RadrootsOutboxError> {
         if changed.rows_affected() > 0 {
             return Ok(());
         }
         self.ensure_claim_token(outbox_event_id, claim_token)
             .await?;
         Err(RadrootsOutboxError::ClaimTokenMismatch { outbox_event_id })
     }
 
     async fn finish_claimed_event(
         &self,
         outbox_event_id: i64,
         claim_token: &str,
         event_state: RadrootsOutboxEventState,
         operation_status: RadrootsOutboxOperationStatus,
         last_error: Option<&str>,
         now_ms: i64,
     ) -> Result<(), RadrootsOutboxError> {
         let mut tx = self.pool.begin().await?;
         let row = claimed_event_identity_tx(&mut tx, outbox_event_id, claim_token).await?;
         let changed = sqlx::query(
             "UPDATE outbox_event SET state = ?, claim_token = NULL, claim_owner = NULL, claim_expires_at_ms = NULL, last_error = ?, next_attempt_after_ms = ?, updated_at_ms = ? WHERE outbox_event_id = ? AND claim_token = ?",
         )
         .bind(event_state.as_str())
         .bind(last_error)
         .bind(now_ms)
         .bind(now_ms)
         .bind(outbox_event_id)
         .bind(claim_token)
         .execute(&mut *tx)
         .await?;
         if changed.rows_affected() == 0 {
             return Err(RadrootsOutboxError::ClaimTokenMismatch { outbox_event_id });
         }
 
         sqlx::query(
             "UPDATE outbox_operation SET status = ?, updated_at_ms = ? WHERE operation_id = ?",
         )
         .bind(operation_status.as_str())
         .bind(now_ms)
         .bind(row.operation_id)
         .execute(&mut *tx)
         .await?;
 
         tx.commit().await?;
         Ok(())
     }
 
     async fn mark_relay_failed(
         &self,
         outbox_event_id: i64,
         claim_token: &str,
         relay_url: &str,
         status: RadrootsOutboxRelayStatus,
         error: &str,
         attempted_at_ms: i64,
     ) -> Result<(), RadrootsOutboxError> {
         let changed = sqlx::query(
             "UPDATE outbox_event_relay_status SET status = ?, attempt_count = attempt_count + 1, last_attempt_at_ms = ?, acknowledged_at_ms = NULL, last_error = ? WHERE outbox_event_id = ? AND relay_url = ? AND EXISTS (SELECT 1 FROM outbox_event WHERE outbox_event_id = ? AND claim_token = ?)",
         )
         .bind(status.as_str())
         .bind(attempted_at_ms)
         .bind(error)
         .bind(outbox_event_id)
         .bind(relay_url)
         .bind(outbox_event_id)
         .bind(claim_token)
         .execute(&self.pool)
         .await?;
         self.ensure_claimed_update(outbox_event_id, claim_token, changed)
             .await?;
         Ok(())
     }
 }
 
 struct ExistingOperation {
     operation_id: i64,
     outbox_event_id: i64,
     event_id: String,
     idempotency_digest: String,
 }
 
 struct ClaimedEventIdentity {
     operation_id: i64,
     accepted_quorum: i64,
 }
 
 async fn configure_connection(
     pool: &SqlitePool,
     file_backed: bool,
 ) -> Result<(), RadrootsOutboxError> {
     sqlx::query("PRAGMA foreign_keys = ON")
         .execute(pool)
         .await?;
     sqlx::query("PRAGMA busy_timeout = 5000")
         .execute(pool)
         .await?;
     if file_backed {
         sqlx::query("PRAGMA journal_mode = WAL")
             .execute(pool)
             .await?;
     }
     Ok(())
 }
 
 #[cfg_attr(coverage_nightly, coverage(off))]
 async fn apply_up(pool: &SqlitePool) -> Result<(), RadrootsOutboxError> {
     sqlx::raw_sql(OUTBOX_MIGRATION_UP).execute(pool).await?;
     Ok(())
 }
 
 #[cfg_attr(coverage_nightly, coverage(off))]
 async fn apply_down(pool: &SqlitePool) -> Result<(), RadrootsOutboxError> {
     sqlx::raw_sql(OUTBOX_MIGRATION_DOWN).execute(pool).await?;
     Ok(())
 }
 
 #[cfg_attr(coverage_nightly, coverage(off))]
 async fn query_i64(pool: &SqlitePool, sql: &str) -> Result<i64, RadrootsOutboxError> {
     let row = sqlx::query(sql).fetch_one(pool).await?;
     Ok(row.try_get(0)?)
 }
 
 #[cfg_attr(coverage_nightly, coverage(off))]
 async fn query_string(pool: &SqlitePool, sql: &str) -> Result<String, RadrootsOutboxError> {
     let row = sqlx::query(sql).fetch_one(pool).await?;
     Ok(row.try_get(0)?)
 }
 
 #[cfg_attr(coverage_nightly, coverage(off))]
 async fn existing_idempotent_operation(
     tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
     operation_kind: &str,
     expected_pubkey: &str,
     idempotency_key: &str,
 ) -> Result<Option<ExistingOperation>, RadrootsOutboxError> {
     let row = sqlx::query(
         "SELECT o.operation_id, o.idempotency_digest, e.outbox_event_id, e.event_id FROM outbox_operation o JOIN outbox_event e ON e.operation_id = o.operation_id WHERE o.operation_kind = ? AND o.expected_pubkey = ? AND o.idempotency_key = ? ORDER BY e.outbox_event_id LIMIT 1",
     )
     .bind(operation_kind)
     .bind(expected_pubkey)
     .bind(idempotency_key)
     .fetch_optional(&mut **tx)
     .await?;
     row.map(|row| {
         Ok(ExistingOperation {
             operation_id: row.try_get("operation_id")?,
             outbox_event_id: row.try_get("outbox_event_id")?,
             event_id: row.try_get("event_id")?,
             idempotency_digest: row.try_get("idempotency_digest")?,
         })
     })
     .transpose()
 }
 
 #[cfg_attr(coverage_nightly, coverage(off))]
 async fn event_by_id_tx(
     tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
     outbox_event_id: i64,
 ) -> Result<RadrootsOutboxEventRecord, RadrootsOutboxError> {
     let row = sqlx::query(
         "SELECT outbox_event_id, operation_id, event_id, expected_pubkey, draft_json, signed_event_json, raw_event_json, state, accepted_quorum, attempt_count, claim_token, claim_owner, claim_expires_at_ms, next_attempt_after_ms, last_error, event_store_ingested, event_store_inserted, event_store_ingested_at_ms, created_at_ms, updated_at_ms FROM outbox_event WHERE outbox_event_id = ?",
     )
     .bind(outbox_event_id)
     .fetch_one(&mut **tx)
     .await?;
     event_from_row(row)
 }
 
 async fn claimed_event_identity_tx(
     tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
     outbox_event_id: i64,
     claim_token: &str,
 ) -> Result<ClaimedEventIdentity, RadrootsOutboxError> {
     let row =
         sqlx::query("SELECT operation_id, accepted_quorum, claim_token FROM outbox_event WHERE outbox_event_id = ?")
             .bind(outbox_event_id)
             .fetch_optional(&mut **tx)
             .await?;
     let Some(row) = row else {
         return Err(RadrootsOutboxError::EventNotFound(outbox_event_id));
     };
     let stored: Option<String> = row.try_get("claim_token")?;
     if stored.as_deref() != Some(claim_token) {
         return Err(RadrootsOutboxError::ClaimTokenMismatch { outbox_event_id });
     }
     Ok(ClaimedEventIdentity {
         operation_id: row.try_get("operation_id")?,
         accepted_quorum: row.try_get("accepted_quorum")?,
     })
 }
 
 #[cfg_attr(coverage_nightly, coverage(off))]
 async fn relay_urls_for_tx(
     tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
     outbox_event_id: i64,
 ) -> Result<Vec<String>, RadrootsOutboxError> {
     let rows = sqlx::query(
         "SELECT relay_url FROM outbox_event_relay_status WHERE outbox_event_id = ? ORDER BY rowid",
     )
     .bind(outbox_event_id)
     .fetch_all(&mut **tx)
     .await?;
     rows.into_iter()
         .map(|row| row.try_get("relay_url").map_err(Into::into))
         .collect()
 }
 
 #[cfg_attr(coverage_nightly, coverage(off))]
 async fn relay_statuses_for(
     pool: &SqlitePool,
     outbox_event_id: i64,
 ) -> Result<Vec<RadrootsOutboxRelayStatusRecord>, RadrootsOutboxError> {
     let rows = sqlx::query(
         "SELECT outbox_event_id, relay_url, status, attempt_count, last_attempt_at_ms, acknowledged_at_ms, last_error FROM outbox_event_relay_status WHERE outbox_event_id = ? ORDER BY rowid",
     )
     .bind(outbox_event_id)
     .fetch_all(pool)
     .await?;
     rows.into_iter().map(relay_status_from_row).collect()
 }
 
 #[cfg_attr(coverage_nightly, coverage(off))]
 fn operation_from_row(
     row: sqlx::sqlite::SqliteRow,
 ) -> Result<RadrootsOutboxOperationRecord, RadrootsOutboxError> {
     let status =
         RadrootsOutboxOperationStatus::parse(row.try_get::<String, _>("status")?.as_str())?;
     Ok(RadrootsOutboxOperationRecord {
         operation_id: row.try_get("operation_id")?,
         operation_kind: row.try_get("operation_kind")?,
         expected_pubkey: row.try_get("expected_pubkey")?,
         idempotency_key: row.try_get("idempotency_key")?,
         idempotency_digest: row.try_get("idempotency_digest")?,
         status,
         created_at_ms: row.try_get("created_at_ms")?,
         updated_at_ms: row.try_get("updated_at_ms")?,
     })
 }
 
 #[cfg_attr(coverage_nightly, coverage(off))]
 fn event_from_row(
     row: sqlx::sqlite::SqliteRow,
 ) -> Result<RadrootsOutboxEventRecord, RadrootsOutboxError> {
     let draft: RadrootsFrozenEventDraft =
         serde_json::from_str(row.try_get::<String, _>("draft_json")?.as_str())?;
     let signed_event = row
         .try_get::<Option<String>, _>("signed_event_json")?
         .map(|json| serde_json::from_str(json.as_str()))
         .transpose()?;
     let state = RadrootsOutboxEventState::parse(row.try_get::<String, _>("state")?.as_str())?;
     Ok(RadrootsOutboxEventRecord {
         outbox_event_id: row.try_get("outbox_event_id")?,
         operation_id: row.try_get("operation_id")?,
         event_id: row.try_get("event_id")?,
         expected_pubkey: row.try_get("expected_pubkey")?,
         draft,
         signed_event,
         raw_event_json: row.try_get("raw_event_json")?,
         state,
         accepted_quorum: row.try_get("accepted_quorum")?,
         attempt_count: row.try_get("attempt_count")?,
         claim_token: row.try_get("claim_token")?,
         claim_owner: row.try_get("claim_owner")?,
         claim_expires_at_ms: row.try_get("claim_expires_at_ms")?,
         next_attempt_after_ms: row.try_get("next_attempt_after_ms")?,
         last_error: row.try_get("last_error")?,
         event_store_ingested: row.try_get::<i64, _>("event_store_ingested")? != 0,
         event_store_inserted: row.try_get::<i64, _>("event_store_inserted")? != 0,
         event_store_ingested_at_ms: row.try_get("event_store_ingested_at_ms")?,
         created_at_ms: row.try_get("created_at_ms")?,
         updated_at_ms: row.try_get("updated_at_ms")?,
     })
 }
 
 #[cfg_attr(coverage_nightly, coverage(off))]
 fn relay_status_from_row(
     row: sqlx::sqlite::SqliteRow,
 ) -> Result<RadrootsOutboxRelayStatusRecord, RadrootsOutboxError> {
     let status = RadrootsOutboxRelayStatus::parse(row.try_get::<String, _>("status")?.as_str())?;
     Ok(RadrootsOutboxRelayStatusRecord {
         outbox_event_id: row.try_get("outbox_event_id")?,
         relay_url: row.try_get("relay_url")?,
         status,
         attempt_count: row.try_get("attempt_count")?,
         last_attempt_at_ms: row.try_get("last_attempt_at_ms")?,
         acknowledged_at_ms: row.try_get("acknowledged_at_ms")?,
         last_error: row.try_get("last_error")?,
     })
 }
 
 fn event_from_signed(signed_event: &RadrootsSignedNostrEvent) -> RadrootsNostrEvent {
     RadrootsNostrEvent {
         id: signed_event.id.clone(),
         author: signed_event.pubkey.clone(),
         created_at: signed_event.created_at,
         kind: signed_event.kind,
         tags: signed_event.tags.clone(),
         content: signed_event.content.clone(),
         sig: signed_event.sig.clone(),
     }
 }
 
 fn ordered_unique_relays(relays: Vec<String>) -> Vec<String> {
     let mut out = Vec::new();
     for relay in relays {
         if !out.iter().any(|existing| existing == &relay) {
             out.push(relay);
         }
     }
     out
 }
 
 fn digest_relays(relays: &[String]) -> Vec<String> {
     let mut out = relays.to_vec();
     out.sort();
     out.dedup();
     out
 }
 
 #[derive(Serialize)]
 struct DigestInput<'a> {
     operation_kind: &'a str,
     expected_pubkey: &'a str,
     draft: &'a RadrootsFrozenEventDraft,
     target_relays: &'a [String],
 }
 
 fn idempotency_digest(
     operation_kind: &str,
     expected_pubkey: &str,
     draft: &RadrootsFrozenEventDraft,
     target_relays: &[String],
 ) -> Result<String, RadrootsOutboxError> {
     let input = DigestInput {
         operation_kind,
         expected_pubkey,
         draft,
         target_relays,
     };
     let bytes = serde_json::to_vec(&input)?;
     Ok(hex::encode(Sha256::digest(bytes)))
 }
 
 fn bool_i64(value: bool) -> i64 {
     if value { 1 } else { 0 }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
     use radroots_events::kinds::KIND_POST;
     use radroots_nostr::prelude::{
         RadrootsNostrKeys, RadrootsNostrSecretKey, radroots_nostr_sign_frozen_draft,
     };
 
     const FIXTURE_ALICE_SECRET_KEY_HEX: &str =
         "10c5304d6c9ae3a1a16f7860f1cc8f5e3a76225a2663b3a989a0d775919b7df5";
     const FIXTURE_ALICE_PUBLIC_KEY_HEX: &str =
         "585591529da0bab31b3b1b1f986611cf5f435dca84f978c89ee8a40cca7103df";
     const RELAY_PRIMARY_WSS: &str = "wss://relay.example.com";
     const RELAY_SECONDARY_WSS: &str = "wss://relay-2.example.com";
 
     fn hex_64(character: char) -> String {
         std::iter::repeat_n(character, 64).collect()
     }
 
     fn post_draft(expected_pubkey: &str, content: &str) -> RadrootsFrozenEventDraft {
         RadrootsFrozenEventDraft::new(
             "radroots.social.post.v1",
             KIND_POST,
             1_700_000_000,
             vec![vec!["t".to_owned(), "soil".to_owned()]],
             content,
             expected_pubkey,
         )
         .expect("post draft")
     }
 
     fn operation_input(
         draft: RadrootsFrozenEventDraft,
         created_at_ms: i64,
     ) -> RadrootsOutboxOperationInput {
         RadrootsOutboxOperationInput::new(
             "publish_post",
             draft,
             vec![
                 RELAY_PRIMARY_WSS.to_owned(),
                 RELAY_SECONDARY_WSS.to_owned(),
                 RELAY_PRIMARY_WSS.to_owned(),
             ],
             created_at_ms,
         )
     }
 
     fn signed_operation_input(
         draft: RadrootsFrozenEventDraft,
         signed_event: RadrootsSignedNostrEvent,
         created_at_ms: i64,
     ) -> RadrootsOutboxSignedOperationInput {
         RadrootsOutboxSignedOperationInput::new(
             "publish_post",
             draft,
             signed_event,
             vec![
                 RELAY_PRIMARY_WSS.to_owned(),
                 RELAY_SECONDARY_WSS.to_owned(),
                 RELAY_PRIMARY_WSS.to_owned(),
             ],
             true,
             created_at_ms + 7,
             created_at_ms,
         )
     }
 
     fn fixture_keys() -> RadrootsNostrKeys {
         let secret_key =
             RadrootsNostrSecretKey::from_hex(FIXTURE_ALICE_SECRET_KEY_HEX).expect("secret key");
         RadrootsNostrKeys::new(secret_key)
     }
 
     async fn enqueue_signed_fixture(
         outbox: &RadrootsOutbox,
     ) -> (RadrootsOutboxEnqueueReceipt, RadrootsOutboxClaimedEvent) {
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "hello");
         let receipt = outbox
             .enqueue_operation(operation_input(draft, 1_000))
             .await
             .expect("enqueue");
         let claimed = outbox
             .claim_next_ready_event("worker-a", "claim-a", 2_000, 1_000)
             .await
             .expect("claim")
             .expect("claimed event");
         (receipt, claimed)
     }
 
     async fn complete_claimed_signing(
         outbox: &RadrootsOutbox,
         claimed: &RadrootsOutboxClaimedEvent,
         keys: &RadrootsNostrKeys,
         now_ms: i64,
     ) -> RadrootsSignedNostrEvent {
         if let Some(signed_event) = claimed.signed_event.clone() {
             return signed_event;
         }
         let signed_event =
             radroots_nostr_sign_frozen_draft(keys, &claimed.draft).expect("signed event");
         outbox
             .complete_signing(
                 claimed.outbox_event_id,
                 claimed.claim_token.as_str(),
                 signed_event,
                 now_ms,
             )
             .await
             .expect("complete signing")
     }
 
     async fn table_count(outbox: &RadrootsOutbox, table_name: &str) -> i64 {
         let sql = format!("SELECT COUNT(*) FROM {table_name}");
         sqlx::query_scalar(sql.as_str())
             .fetch_one(outbox.pool())
             .await
             .expect("table count")
     }
 
     #[tokio::test]
     async fn migration_applies_pragmas_and_migrates_down() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
 
         assert_eq!(outbox.pragma_foreign_keys().await.expect("foreign keys"), 1);
         assert_eq!(
             outbox.pragma_busy_timeout().await.expect("busy timeout"),
             5_000
         );
         assert_eq!(
             outbox.pragma_journal_mode().await.expect("journal mode"),
             "memory"
         );
 
         let row = sqlx::query(
             "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'outbox_event'",
         )
         .fetch_optional(outbox.pool())
         .await
         .expect("table query");
         assert!(row.is_some());
 
         outbox.migrate_down().await.expect("migrate down");
         let row = sqlx::query(
             "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'outbox_event'",
         )
         .fetch_optional(outbox.pool())
         .await
         .expect("table query");
         assert!(row.is_none());
     }
 
     #[tokio::test]
     async fn file_outbox_reopens_existing_schema() {
         let tempdir = tempfile::tempdir().expect("tempdir");
         let path = tempdir.path().join("outbox.sqlite");
 
         let first = RadrootsOutbox::open_file(&path).await.expect("first");
         assert_eq!(first.pragma_foreign_keys().await.expect("foreign keys"), 1);
         drop(first);
 
         let second = RadrootsOutbox::open_file(&path).await.expect("second");
         assert_eq!(second.pragma_foreign_keys().await.expect("foreign keys"), 1);
     }
 
     #[tokio::test]
     async fn status_summary_counts_ready_publishing_retryable_and_terminal_work() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
 
         let empty = outbox.status_summary(1_000).await.expect("empty status");
         assert_eq!(empty.total_events, 0);
         assert_eq!(empty.pending_events, 0);
         assert_eq!(empty.retryable_events, 0);
         assert_eq!(empty.terminal_events, 0);
         assert_eq!(empty.failed_terminal_events, 0);
         assert_eq!(empty.ready_signed_events, 0);
         assert_eq!(empty.publishing_events, 0);
         assert_eq!(empty.last_attempt_at_ms, None);
         assert_eq!(empty.last_error, None);
 
         let keys = fixture_keys();
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "ready");
         let signed_event = radroots_nostr_sign_frozen_draft(&keys, &draft).expect("signed event");
         let receipt = outbox
             .enqueue_signed_operation(signed_operation_input(draft, signed_event, 1_000))
             .await
             .expect("signed enqueue");
         let ready = outbox.status_summary(1_000).await.expect("ready status");
         assert_eq!(ready.total_events, 1);
         assert_eq!(ready.pending_events, 1);
         assert_eq!(ready.retryable_events, 0);
         assert_eq!(ready.terminal_events, 0);
         assert_eq!(ready.ready_signed_events, 1);
 
         let claimed = outbox
             .claim_next_ready_signed_event("publisher", "claim-a", 2_500, 2_000)
             .await
             .expect("claim")
             .expect("claimed");
         let publishing = outbox
             .status_summary(2_000)
             .await
             .expect("publishing status");
         assert_eq!(publishing.pending_events, 1);
         assert_eq!(publishing.publishing_events, 1);
         assert_eq!(publishing.ready_signed_events, 0);
 
         outbox
             .mark_relay_failed_retryable(
                 receipt.outbox_event_id,
                 claimed.claim_token.as_str(),
                 RELAY_PRIMARY_WSS,
                 "timeout: relay unavailable",
                 2_100,
             )
             .await
             .expect("relay failed");
         outbox
             .mark_publish_retryable(
                 receipt.outbox_event_id,
                 claimed.claim_token.as_str(),
                 "relay publish incomplete",
                 3_000,
                 2_200,
             )
             .await
             .expect("retryable");
 
         let retry_wait = outbox
             .status_summary(2_900)
             .await
             .expect("retry wait status");
         assert_eq!(retry_wait.pending_events, 0);
         assert_eq!(retry_wait.retryable_events, 1);
         assert_eq!(retry_wait.terminal_events, 0);
         assert_eq!(retry_wait.failed_terminal_events, 0);
         assert_eq!(retry_wait.ready_signed_events, 0);
         assert_eq!(retry_wait.last_attempt_at_ms, Some(2_100));
         assert_eq!(
             retry_wait.last_error.as_deref(),
             Some("relay publish incomplete")
         );
 
         let retry_ready = outbox
             .status_summary(3_000)
             .await
             .expect("retry ready status");
         assert_eq!(retry_ready.ready_signed_events, 1);
     }
 
     #[test]
     fn terminal_and_cancelled_event_states_round_trip() {
         assert_eq!(bool_i64(true), 1);
         assert_eq!(bool_i64(false), 0);
         assert_eq!(
             RadrootsOutboxOperationStatus::parse("failed_terminal").expect("operation status"),
             RadrootsOutboxOperationStatus::FailedTerminal
         );
         assert_eq!(
             RadrootsOutboxOperationStatus::FailedTerminal.as_str(),
             "failed_terminal"
         );
         assert_eq!(
             RadrootsOutboxOperationStatus::parse("cancelled").expect("operation status"),
             RadrootsOutboxOperationStatus::Cancelled
         );
         assert_eq!(
             RadrootsOutboxOperationStatus::Cancelled.as_str(),
             "cancelled"
         );
         assert_eq!(
             RadrootsOutboxEventState::parse("failed_terminal").expect("event state"),
             RadrootsOutboxEventState::FailedTerminal
         );
         assert!(RadrootsOutboxEventState::FailedTerminal.is_terminal());
         assert_eq!(
             RadrootsOutboxEventState::parse("cancelled").expect("event state"),
             RadrootsOutboxEventState::Cancelled
         );
         assert!(RadrootsOutboxEventState::Cancelled.is_terminal());
         assert!(RadrootsOutboxEventState::Published.is_terminal());
         assert!(!RadrootsOutboxEventState::PublishRetryable.is_terminal());
     }
 
     #[tokio::test]
     async fn enqueue_idempotency_is_scoped_by_kind_pubkey_and_digest() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let first_draft = post_draft(hex_64('a').as_str(), "hello");
 
         let first = outbox
             .enqueue_operation(operation_input(first_draft.clone(), 1_000))
             .await
             .expect("first enqueue");
         let second = outbox
             .enqueue_operation(operation_input(first_draft.clone(), 1_001))
             .await
             .expect("second enqueue");
 
         assert_eq!(first.status, RadrootsOutboxEnqueueStatus::Inserted);
         assert_eq!(second.status, RadrootsOutboxEnqueueStatus::Inserted);
         assert_ne!(first.operation_id, second.operation_id);
         assert_ne!(first.outbox_event_id, second.outbox_event_id);
 
         let keyed_first = outbox
             .enqueue_operation(
                 operation_input(first_draft.clone(), 1_002).with_idempotency_key("idem-a"),
             )
             .await
             .expect("keyed first");
         let keyed_second = outbox
             .enqueue_operation(
                 operation_input(first_draft.clone(), 1_003).with_idempotency_key("idem-a"),
             )
             .await
             .expect("keyed second");
 
         assert_eq!(keyed_first.status, RadrootsOutboxEnqueueStatus::Inserted);
         assert_eq!(keyed_second.status, RadrootsOutboxEnqueueStatus::Existing);
         assert_eq!(keyed_first.operation_id, keyed_second.operation_id);
         assert_eq!(keyed_first.outbox_event_id, keyed_second.outbox_event_id);
         assert_eq!(
             keyed_first.idempotency_digest,
             keyed_second.idempotency_digest
         );
 
         let conflict = outbox
             .enqueue_operation(
                 operation_input(post_draft(hex_64('a').as_str(), "changed"), 1_004)
                     .with_idempotency_key("idem-a"),
             )
             .await
             .expect_err("conflict");
         assert!(matches!(
             conflict,
             RadrootsOutboxError::IdempotencyConflict { .. }
         ));
 
         let other_kind = outbox
             .enqueue_operation(
                 RadrootsOutboxOperationInput::new(
                     "publish_post_reply",
                     first_draft.clone(),
                     vec![RELAY_PRIMARY_WSS.to_owned()],
                     1_005,
                 )
                 .with_idempotency_key("idem-a"),
             )
             .await
             .expect("other kind");
         assert_eq!(other_kind.status, RadrootsOutboxEnqueueStatus::Inserted);
 
         let other_pubkey = outbox
             .enqueue_operation(
                 operation_input(post_draft(hex_64('b').as_str(), "hello"), 1_006)
                     .with_idempotency_key("idem-a"),
             )
             .await
             .expect("other pubkey");
         assert_eq!(other_pubkey.status, RadrootsOutboxEnqueueStatus::Inserted);
     }
 
     #[tokio::test]
     async fn enqueue_idempotency_digest_sorts_relays_but_publish_order_is_preserved() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(hex_64('a').as_str(), "hello");
         let first = outbox
             .enqueue_operation(
                 RadrootsOutboxOperationInput::new(
                     "publish_post",
                     draft.clone(),
                     vec![
                         RELAY_SECONDARY_WSS.to_owned(),
                         RELAY_PRIMARY_WSS.to_owned(),
                         RELAY_SECONDARY_WSS.to_owned(),
                     ],
                     1_000,
                 )
                 .with_idempotency_key("idem-relay-order"),
             )
             .await
             .expect("first enqueue");
         let second = outbox
             .enqueue_operation(
                 RadrootsOutboxOperationInput::new(
                     "publish_post",
                     draft,
                     vec![RELAY_PRIMARY_WSS.to_owned(), RELAY_SECONDARY_WSS.to_owned()],
                     1_001,
                 )
                 .with_idempotency_key("idem-relay-order"),
             )
             .await
             .expect("second enqueue");
 
         assert_eq!(second.status, RadrootsOutboxEnqueueStatus::Existing);
         assert_eq!(first.idempotency_digest, second.idempotency_digest);
 
         let claimed = outbox
             .claim_next_ready_event("worker-a", "claim-a", 2_000, 1_000)
             .await
             .expect("claim")
             .expect("claimed event");
         assert_eq!(
             claimed.target_relays,
             vec![RELAY_SECONDARY_WSS.to_owned(), RELAY_PRIMARY_WSS.to_owned()]
         );
     }
 
     #[tokio::test]
     async fn enqueue_rejects_empty_target_relays_before_persistence() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(hex_64('a').as_str(), "hello");
 
         let err = outbox
             .enqueue_operation(
                 RadrootsOutboxOperationInput::new("publish_post", draft, Vec::new(), 1_000)
                     .with_idempotency_key("empty-relays"),
             )
             .await
             .expect_err("empty relays");
 
         assert!(matches!(err, RadrootsOutboxError::EmptyTargetRelays));
         assert_eq!(table_count(&outbox, "outbox_operation").await, 0);
         assert_eq!(table_count(&outbox, "outbox_event").await, 0);
         assert_eq!(table_count(&outbox, "outbox_event_relay_status").await, 0);
     }
 
     #[tokio::test]
     async fn enqueue_signed_rejects_empty_target_relays_before_persistence() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "signed empty");
         let signed_event =
             radroots_nostr_sign_frozen_draft(&fixture_keys(), &draft).expect("signed event");
 
         let err = outbox
             .enqueue_signed_operation(RadrootsOutboxSignedOperationInput::new(
                 "publish_post",
                 draft,
                 signed_event,
                 Vec::new(),
                 false,
                 1_007,
                 1_000,
             ))
             .await
             .expect_err("empty relays");
 
         assert!(matches!(err, RadrootsOutboxError::EmptyTargetRelays));
         assert_eq!(table_count(&outbox, "outbox_operation").await, 0);
         assert_eq!(table_count(&outbox, "outbox_event").await, 0);
         assert_eq!(table_count(&outbox, "outbox_event_relay_status").await, 0);
     }
 
     #[tokio::test]
     async fn enqueue_allows_explicitly_delegated_empty_target_relays() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(hex_64('a').as_str(), "delegated empty");
 
         let receipt = outbox
             .enqueue_operation(
                 RadrootsOutboxOperationInput::new("publish_post", draft, Vec::new(), 1_000)
                     .allow_empty_target_relays(),
             )
             .await
             .expect("delegated empty relays");
 
         let event = outbox
             .get_event(receipt.outbox_event_id)
             .await
             .expect("event")
             .expect("event");
         assert_eq!(event.accepted_quorum, 0);
         assert_eq!(
             outbox
                 .relay_statuses(receipt.outbox_event_id)
                 .await
                 .expect("relay statuses"),
             Vec::new()
         );
 
         let claimed = outbox
             .claim_next_ready_event("worker-a", "claim-a", 2_000, 1_000)
             .await
             .expect("claim")
             .expect("claim");
         assert_eq!(claimed.target_relays, Vec::<String>::new());
     }
 
     #[tokio::test]
     async fn enqueue_signed_allows_explicitly_delegated_empty_target_relays() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "signed delegated empty");
         let signed_event =
             radroots_nostr_sign_frozen_draft(&fixture_keys(), &draft).expect("signed event");
 
         let receipt = outbox
             .enqueue_signed_operation(
                 RadrootsOutboxSignedOperationInput::new(
                     "publish_post",
                     draft,
                     signed_event.clone(),
                     Vec::new(),
                     false,
                     1_007,
                     1_000,
                 )
                 .allow_empty_target_relays(),
             )
             .await
             .expect("delegated signed empty relays");
 
         let event = outbox
             .get_event(receipt.outbox_event_id)
             .await
             .expect("event")
             .expect("event");
         assert_eq!(event.accepted_quorum, 0);
         assert_eq!(event.signed_event, Some(signed_event.clone()));
         assert_eq!(
             outbox
                 .relay_statuses(receipt.outbox_event_id)
                 .await
                 .expect("relay statuses"),
             Vec::new()
         );
 
         let claimed = outbox
             .claim_next_ready_signed_event("publisher-a", "claim-a", 2_000, 1_000)
             .await
             .expect("claim")
             .expect("claim");
         assert_eq!(claimed.signed_event, Some(signed_event));
         assert_eq!(claimed.target_relays, Vec::<String>::new());
     }
 
     #[tokio::test]
     async fn claim_next_ready_event_returns_none_when_no_work_is_ready() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
 
         assert!(
             outbox
                 .claim_next_ready_event("worker-a", "claim-a", 2_000, 1_000)
                 .await
                 .expect("claim")
                 .is_none()
         );
         assert!(
             outbox
                 .claim_next_ready_signed_event("publisher-a", "claim-b", 2_000, 1_000)
                 .await
                 .expect("claim signed")
                 .is_none()
         );
     }
 
     #[tokio::test]
     async fn enqueue_accepts_single_and_multiple_target_relays() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let single_draft = post_draft(hex_64('a').as_str(), "single");
 
         let single = outbox
             .enqueue_operation(RadrootsOutboxOperationInput::new(
                 "publish_post",
                 single_draft,
                 vec![RELAY_PRIMARY_WSS.to_owned()],
                 1_000,
             ))
             .await
             .expect("single relay");
         let single_event = outbox
             .get_event(single.outbox_event_id)
             .await
             .expect("single event")
             .expect("single event");
         assert_eq!(single_event.accepted_quorum, 1);
         assert_eq!(
             outbox
                 .relay_statuses(single.outbox_event_id)
                 .await
                 .expect("single relay statuses")
                 .len(),
             1
         );
 
         let multi_draft = post_draft(hex_64('b').as_str(), "multi");
         let multi = outbox
             .enqueue_operation(RadrootsOutboxOperationInput::new(
                 "publish_post",
                 multi_draft,
                 vec![
                     RELAY_PRIMARY_WSS.to_owned(),
                     RELAY_SECONDARY_WSS.to_owned(),
                     RELAY_PRIMARY_WSS.to_owned(),
                 ],
                 1_100,
             ))
             .await
             .expect("multiple relays");
         let multi_event = outbox
             .get_event(multi.outbox_event_id)
             .await
             .expect("multi event")
             .expect("multi event");
         assert_eq!(multi_event.accepted_quorum, 2);
         assert_eq!(
             outbox
                 .relay_statuses(multi.outbox_event_id)
                 .await
                 .expect("multi relay statuses")
                 .len(),
             2
         );
 
         let zero_quorum_count: i64 =
             sqlx::query_scalar("SELECT COUNT(*) FROM outbox_event WHERE accepted_quorum = 0")
                 .fetch_one(outbox.pool())
                 .await
                 .expect("zero quorum count");
         assert_eq!(zero_quorum_count, 0);
     }
 
     #[tokio::test]
     async fn enqueue_signed_operation_stores_pushable_signed_event_without_claim() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "signed");
         let signed_event =
             radroots_nostr_sign_frozen_draft(&fixture_keys(), &draft).expect("signed event");
         let expected_raw_json = signed_event.raw_json.clone();
 
         let receipt = outbox
             .enqueue_signed_operation(
                 signed_operation_input(draft.clone(), signed_event.clone(), 1_000)
                     .with_idempotency_key("signed-a"),
             )
             .await
             .expect("signed enqueue");
 
         assert_eq!(receipt.status, RadrootsOutboxEnqueueStatus::Inserted);
         assert_eq!(receipt.expected_event_id, draft.expected_event_id);
 
         let event = outbox
             .get_event(receipt.outbox_event_id)
             .await
             .expect("event")
             .expect("event");
         assert_eq!(event.state, RadrootsOutboxEventState::Signed);
         assert_eq!(event.signed_event, Some(signed_event.clone()));
         assert_eq!(event.raw_event_json, Some(expected_raw_json));
         assert!(event.event_store_ingested);
         assert!(event.event_store_inserted);
         assert_eq!(event.event_store_ingested_at_ms, Some(1_007));
         assert_eq!(event.claim_token, None);
         assert_eq!(event.claim_owner, None);
         assert_eq!(event.claim_expires_at_ms, None);
 
         let statuses = outbox
             .relay_statuses(receipt.outbox_event_id)
             .await
             .expect("statuses");
         assert_eq!(statuses.len(), 2);
         assert!(
             statuses
                 .iter()
                 .all(|status| status.status == RadrootsOutboxRelayStatus::Pending)
         );
 
         let claimed = outbox
             .claim_next_ready_event("publisher-a", "claim-a", 2_000, 1_000)
             .await
             .expect("claim")
             .expect("claimed");
         assert_eq!(claimed.state, RadrootsOutboxEventState::Publishing);
         assert_eq!(claimed.signed_event, Some(signed_event));
         assert_eq!(
             claimed.target_relays,
             vec![RELAY_PRIMARY_WSS.to_owned(), RELAY_SECONDARY_WSS.to_owned()]
         );
     }
 
     #[tokio::test]
     async fn claim_next_ready_signed_event_skips_unsigned_work() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let unsigned = outbox
             .enqueue_operation(operation_input(
                 post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "unsigned"),
                 900,
             ))
             .await
             .expect("unsigned enqueue");
         let signed_draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "signed-only");
         let signed_event =
             radroots_nostr_sign_frozen_draft(&fixture_keys(), &signed_draft).expect("signed event");
         let signed = outbox
             .enqueue_signed_operation(signed_operation_input(signed_draft, signed_event, 1_000))
             .await
             .expect("signed enqueue");
 
         let claimed = outbox
             .claim_next_ready_signed_event("publisher-a", "claim-a", 2_000, 1_000)
             .await
             .expect("claim")
             .expect("claimed");
         assert_eq!(claimed.outbox_event_id, signed.outbox_event_id);
         assert_eq!(claimed.state, RadrootsOutboxEventState::Publishing);
         assert!(claimed.signed_event.is_some());
 
         let unsigned_event = outbox
             .get_event(unsigned.outbox_event_id)
             .await
             .expect("unsigned event")
             .expect("unsigned event");
         assert_eq!(unsigned_event.state, RadrootsOutboxEventState::DraftQueued);
         assert!(unsigned_event.claim_token.is_none());
 
         let signing_claim = outbox
             .claim_next_ready_event("signer-a", "sign-a", 2_100, 1_100)
             .await
             .expect("sign claim")
             .expect("sign claim");
         assert_eq!(signing_claim.outbox_event_id, unsigned.outbox_event_id);
         assert_eq!(signing_claim.state, RadrootsOutboxEventState::Signing);
         assert!(signing_claim.signed_event.is_none());
     }
 
     #[tokio::test]
     async fn enqueue_signed_operation_idempotency_reuses_existing_signed_record() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "idem-signed");
         let signed_event =
             radroots_nostr_sign_frozen_draft(&fixture_keys(), &draft).expect("signed event");
 
         let first = outbox
             .enqueue_signed_operation(
                 signed_operation_input(draft.clone(), signed_event.clone(), 1_000)
                     .with_idempotency_key("signed-idem"),
             )
             .await
             .expect("first");
         let second = outbox
             .enqueue_signed_operation(
                 signed_operation_input(draft.clone(), signed_event, 1_100)
                     .with_idempotency_key("signed-idem"),
             )
             .await
             .expect("second");
 
         assert_eq!(first.status, RadrootsOutboxEnqueueStatus::Inserted);
         assert_eq!(second.status, RadrootsOutboxEnqueueStatus::Existing);
         assert_eq!(first.operation_id, second.operation_id);
         assert_eq!(first.outbox_event_id, second.outbox_event_id);
         assert_eq!(table_count(&outbox, "outbox_operation").await, 1);
         assert_eq!(table_count(&outbox, "outbox_event").await, 1);
 
         let changed_draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "changed");
         let changed_signed =
             radroots_nostr_sign_frozen_draft(&fixture_keys(), &changed_draft).expect("signed");
         let conflict = outbox
             .enqueue_signed_operation(
                 signed_operation_input(changed_draft, changed_signed, 1_200)
                     .with_idempotency_key("signed-idem"),
             )
             .await
             .expect_err("conflict");
         assert!(matches!(
             conflict,
             RadrootsOutboxError::IdempotencyConflict { .. }
         ));
     }
 
     #[tokio::test]
     async fn enqueue_signed_operation_rejects_mismatched_signed_event() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "trusted");
         let other_draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "other");
         let signed_event =
             radroots_nostr_sign_frozen_draft(&fixture_keys(), &other_draft).expect("signed event");
 
         let error = outbox
             .enqueue_signed_operation(signed_operation_input(draft, signed_event, 1_000))
             .await
             .expect_err("mismatch");
 
         assert!(matches!(
             error,
             RadrootsOutboxError::SignedEventDraftMismatch(_)
         ));
         assert_eq!(table_count(&outbox, "outbox_operation").await, 0);
         assert_eq!(table_count(&outbox, "outbox_event").await, 0);
     }
 
     #[tokio::test]
     async fn enqueue_signed_operation_rejects_event_id_mismatch() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "bad-id");
         let mut signed_event =
             radroots_nostr_sign_frozen_draft(&fixture_keys(), &draft).expect("signed event");
         signed_event.id = hex_64('f');
 
         let error = outbox
             .enqueue_signed_operation(signed_operation_input(draft, signed_event, 1_000))
             .await
             .expect_err("mismatch");
 
         assert!(matches!(
             error,
             RadrootsOutboxError::SignedEventDraftMismatch(_)
         ));
         assert_eq!(table_count(&outbox, "outbox_operation").await, 0);
         assert_eq!(table_count(&outbox, "outbox_event").await, 0);
     }
 
     #[tokio::test]
     async fn complete_signing_rejects_signed_event_id_mismatch() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "claimed draft");
         let receipt = outbox
             .enqueue_operation(operation_input(draft, 1_000))
             .await
             .expect("enqueue");
         let claimed = outbox
             .claim_next_ready_event("worker-a", "claim-a", 2_000, 1_000)
             .await
             .expect("claim")
             .expect("claim");
         let other_draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "other draft");
         let signed_event =
             radroots_nostr_sign_frozen_draft(&fixture_keys(), &other_draft).expect("signed event");
 
         let err = outbox
             .complete_signing(
                 receipt.outbox_event_id,
                 claimed.claim_token.as_str(),
                 signed_event.clone(),
                 1_100,
             )
             .await
             .expect_err("event id mismatch");
 
         assert_eq!(
             err.to_string(),
             format!(
                 "Signed event ID mismatch: expected {}, got {}",
                 receipt.expected_event_id, signed_event.id
             )
         );
     }
 
     #[tokio::test]
     async fn claim_token_guards_updates_and_expired_signing_claim_recovers() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(hex_64('a').as_str(), "hello");
         let receipt = outbox
             .enqueue_operation(operation_input(draft, 1_000))
             .await
             .expect("enqueue");
 
         let claimed = outbox
             .claim_next_ready_event("worker-a", "claim-a", 1_100, 1_000)
             .await
             .expect("claim")
             .expect("claimed event");
         assert_eq!(claimed.state, RadrootsOutboxEventState::Signing);
         assert_eq!(claimed.attempt_count, 1);
         assert_eq!(
             claimed.target_relays,
             vec![RELAY_PRIMARY_WSS.to_owned(), RELAY_SECONDARY_WSS.to_owned()]
         );
 
         let unavailable = outbox
             .claim_next_ready_event("worker-b", "claim-b", 1_100, 1_050)
             .await
             .expect("claim");
         assert!(unavailable.is_none());
 
         let wrong_token = outbox
             .mark_sign_retryable(
                 receipt.outbox_event_id,
                 "claim-b",
                 "sign failed",
                 1_200,
                 1_100,
             )
             .await
             .expect_err("wrong token");
         assert!(matches!(
             wrong_token,
             RadrootsOutboxError::ClaimTokenMismatch { .. }
         ));
 
         let recovered = outbox.recover_expired_claims(1_101).await.expect("recover");
         assert_eq!(recovered, 1);
 
         let event = outbox
             .get_event(receipt.outbox_event_id)
             .await
             .expect("event")
             .expect("event");
         assert_eq!(event.state, RadrootsOutboxEventState::SignRetryable);
         assert_eq!(event.attempt_count, 1);
         assert!(event.claim_token.is_none());
 
         let reclaimed = outbox
             .claim_next_ready_event("worker-b", "claim-b", 1_400, 1_200)
             .await
             .expect("claim")
             .expect("reclaimed");
         assert_eq!(reclaimed.state, RadrootsOutboxEventState::Signing);
         assert_eq!(reclaimed.attempt_count, 2);
     }
 
     #[tokio::test]
     async fn claimed_mutations_reject_stale_tokens_without_state_changes() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let event_store = RadrootsEventStore::open_memory()
             .await
             .expect("event store");
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "hello");
         let receipt = outbox
             .enqueue_operation(operation_input(draft, 1_000))
             .await
             .expect("enqueue");
 
         let first_claim = outbox
             .claim_next_ready_event("worker-a", "claim-a", 1_100, 1_000)
             .await
             .expect("claim")
             .expect("claim");
         let signed = complete_claimed_signing(&outbox, &first_claim, &fixture_keys(), 1_050).await;
         outbox.recover_expired_claims(1_101).await.expect("recover");
         let second_claim = outbox
             .claim_next_ready_event("worker-b", "claim-b", 1_500, 1_200)
             .await
             .expect("claim")
             .expect("claim");
         assert_eq!(second_claim.state, RadrootsOutboxEventState::Publishing);
 
         let retry_with_stale_token = outbox
             .mark_publish_retryable(
                 receipt.outbox_event_id,
                 "claim-a",
                 "stale retry",
                 1_600,
                 1_300,
             )
             .await
             .expect_err("stale retry token");
         assert!(matches!(
             retry_with_stale_token,
             RadrootsOutboxError::ClaimTokenMismatch { .. }
         ));
 
         let relay_with_stale_token = outbox
             .mark_relay_accepted(receipt.outbox_event_id, "claim-a", RELAY_PRIMARY_WSS, 1_300)
             .await
             .expect_err("stale relay token");
         assert!(matches!(
             relay_with_stale_token,
             RadrootsOutboxError::ClaimTokenMismatch { .. }
         ));
 
         let ingest_with_current_token = outbox
             .ingest_signed_event_local(&event_store, receipt.outbox_event_id, "claim-b", 1_350)
             .await
             .expect("current ingest");
         assert_eq!(ingest_with_current_token.event_id, signed.id);
 
         let event = outbox
             .get_event(receipt.outbox_event_id)
             .await
             .expect("event")
             .expect("event");
         assert_eq!(event.state, RadrootsOutboxEventState::Publishing);
         assert_eq!(event.claim_token.as_deref(), Some("claim-b"));
 
         let statuses = outbox
             .relay_statuses(receipt.outbox_event_id)
             .await
             .expect("statuses");
         assert!(
             statuses
                 .iter()
                 .all(|status| status.status == RadrootsOutboxRelayStatus::Pending)
         );
     }
 
     #[tokio::test]
     async fn claimed_update_paths_report_missing_events_and_wrong_tokens() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
 
         let missing = outbox
             .mark_sign_retryable(999, "missing-claim", "missing", 1_200, 1_100)
             .await
             .expect_err("missing event");
         assert!(matches!(missing, RadrootsOutboxError::EventNotFound(999)));
         let missing_publish = outbox
             .complete_publish_attempt(999, "missing-claim", "retryable", "terminal", 1_300, 1_200)
             .await
             .expect_err("missing publish event");
         assert!(matches!(
             missing_publish,
             RadrootsOutboxError::EventNotFound(999)
         ));
 
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "wrong token");
         let receipt = outbox
             .enqueue_operation(operation_input(draft, 1_000))
             .await
             .expect("enqueue");
         outbox
             .claim_next_ready_event("worker-a", "claim-a", 2_000, 1_000)
             .await
             .expect("claim")
             .expect("claim");
 
         let wrong_token = outbox
             .complete_publish_attempt(
                 receipt.outbox_event_id,
                 "claim-b",
                 "retryable",
                 "terminal",
                 2_500,
                 2_100,
             )
             .await
             .expect_err("wrong token");
         assert!(matches!(
             wrong_token,
             RadrootsOutboxError::ClaimTokenMismatch { .. }
         ));
     }
 
     #[tokio::test]
     async fn sign_retryable_update_succeeds_and_reports_ignored_update_with_current_token() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "retryable success");
         let receipt = outbox
             .enqueue_operation(operation_input(draft, 1_000))
             .await
             .expect("enqueue");
         outbox
             .claim_next_ready_event("worker-a", "claim-a", 2_000, 1_000)
             .await
             .expect("claim")
             .expect("claim");
 
         outbox
             .mark_sign_retryable(receipt.outbox_event_id, "claim-a", "retry", 1_500, 1_100)
             .await
             .expect("mark retryable");
         let event = outbox
             .get_event(receipt.outbox_event_id)
             .await
             .expect("event")
             .expect("event");
         assert_eq!(event.state, RadrootsOutboxEventState::SignRetryable);
 
         let ignored_outbox = RadrootsOutbox::open_memory().await.expect("ignored open");
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "ignored retryable");
         let ignored_receipt = ignored_outbox
             .enqueue_operation(operation_input(draft, 2_000))
             .await
             .expect("enqueue ignored");
         ignored_outbox
             .claim_next_ready_event("worker-b", "claim-b", 3_000, 2_000)
             .await
             .expect("claim ignored")
             .expect("claim ignored");
         sqlx::query(
             "CREATE TEMP TRIGGER ignore_sign_retry_update BEFORE UPDATE OF state ON outbox_event WHEN NEW.state = 'sign_retryable' BEGIN SELECT RAISE(IGNORE); END",
         )
         .execute(ignored_outbox.pool())
         .await
         .expect("retry trigger");
         let ignored = ignored_outbox
             .mark_sign_retryable(
                 ignored_receipt.outbox_event_id,
                 "claim-b",
                 "ignored retry",
                 2_500,
                 2_100,
             )
             .await
             .expect_err("ignored retryable update");
         assert!(matches!(
             ignored,
             RadrootsOutboxError::ClaimTokenMismatch { .. }
         ));
     }
 
     #[tokio::test]
     async fn ignored_sqlite_updates_preserve_race_guards() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "ignored claim");
         outbox
             .enqueue_operation(operation_input(draft, 1_000))
             .await
             .expect("enqueue");
         let signed_draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "ignored signed claim");
         let signed_event =
             radroots_nostr_sign_frozen_draft(&fixture_keys(), &signed_draft).expect("signed event");
         outbox
             .enqueue_signed_operation(signed_operation_input(signed_draft, signed_event, 1_001))
             .await
             .expect("signed enqueue");
         sqlx::query(
             "CREATE TEMP TRIGGER ignore_claim_update BEFORE UPDATE OF claim_token ON outbox_event WHEN NEW.claim_token IN ('blocked-claim', 'blocked-signed') BEGIN SELECT RAISE(IGNORE); END",
         )
         .execute(outbox.pool())
         .await
         .expect("claim trigger");
 
         assert!(
             outbox
                 .claim_next_ready_event("worker-a", "blocked-claim", 2_000, 1_000)
                 .await
                 .expect("claim")
                 .is_none()
         );
         assert!(
             outbox
                 .claim_next_ready_signed_event("publisher-a", "blocked-signed", 2_000, 1_001)
                 .await
                 .expect("claim signed")
                 .is_none()
         );
 
         let publish_outbox = RadrootsOutbox::open_memory().await.expect("publish open");
         let publish_draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "ignored publish");
         let publish_signed = radroots_nostr_sign_frozen_draft(&fixture_keys(), &publish_draft)
             .expect("signed event");
         let publish_receipt = publish_outbox
             .enqueue_signed_operation(signed_operation_input(publish_draft, publish_signed, 1_100))
             .await
             .expect("publish enqueue");
         let publish_claim = publish_outbox
             .claim_next_ready_signed_event("publisher-b", "publish-claim", 3_000, 1_100)
             .await
             .expect("claim publish")
             .expect("claim publish");
         publish_outbox
             .mark_relay_accepted(
                 publish_receipt.outbox_event_id,
                 publish_claim.claim_token.as_str(),
                 RELAY_PRIMARY_WSS,
                 1_150,
             )
             .await
             .expect("primary accepted");
         publish_outbox
             .mark_relay_accepted(
                 publish_receipt.outbox_event_id,
                 publish_claim.claim_token.as_str(),
                 RELAY_SECONDARY_WSS,
                 1_160,
             )
             .await
             .expect("secondary accepted");
         sqlx::query(
             "CREATE TEMP TRIGGER ignore_published_update BEFORE UPDATE OF state ON outbox_event WHEN NEW.state = 'published' BEGIN SELECT RAISE(IGNORE); END",
         )
         .execute(publish_outbox.pool())
         .await
         .expect("publish trigger");
         let ignored_publish = publish_outbox
             .complete_publish_attempt(
                 publish_receipt.outbox_event_id,
                 publish_claim.claim_token.as_str(),
                 "retryable",
                 "terminal",
                 2_500,
                 1_200,
             )
             .await
             .expect_err("ignored publish update");
         assert!(matches!(
             ignored_publish,
             RadrootsOutboxError::ClaimTokenMismatch { .. }
         ));
 
         let cancel_outbox = RadrootsOutbox::open_memory().await.expect("cancel open");
         let cancel_draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "ignored cancel");
         let cancel_receipt = cancel_outbox
             .enqueue_operation(operation_input(cancel_draft, 1_200))
             .await
             .expect("cancel enqueue");
         let cancel_claim = cancel_outbox
             .claim_next_ready_event("worker-b", "cancel-claim", 3_000, 1_200)
             .await
             .expect("cancel claim")
             .expect("cancel claim");
         sqlx::query(
             "CREATE TEMP TRIGGER ignore_cancel_update BEFORE UPDATE OF state ON outbox_event WHEN NEW.state = 'cancelled' BEGIN SELECT RAISE(IGNORE); END",
         )
         .execute(cancel_outbox.pool())
         .await
         .expect("cancel trigger");
         let ignored_cancel = cancel_outbox
             .cancel_claimed_event(
                 cancel_receipt.outbox_event_id,
                 cancel_claim.claim_token.as_str(),
                 1_300,
             )
             .await
             .expect_err("ignored cancel update");
         assert!(matches!(
             ignored_cancel,
             RadrootsOutboxError::ClaimTokenMismatch { .. }
         ));
     }
 
     #[tokio::test]
     async fn signed_events_are_reused_after_claim_recovery() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let (receipt, claimed) = enqueue_signed_fixture(&outbox).await;
         let keys = fixture_keys();
 
         let signed = complete_claimed_signing(&outbox, &claimed, &keys, 1_100).await;
         assert_eq!(signed.id, receipt.expected_event_id);
 
         let recovered = outbox.recover_expired_claims(2_001).await.expect("recover");
         assert_eq!(recovered, 1);
 
         let publish_claim = outbox
             .claim_next_ready_event("publisher-a", "claim-b", 3_000, 2_100)
             .await
             .expect("claim")
             .expect("publish claim");
         assert_eq!(publish_claim.state, RadrootsOutboxEventState::Publishing);
         assert_eq!(publish_claim.signed_event.as_ref(), Some(&signed));
 
         let reused = complete_claimed_signing(&outbox, &publish_claim, &keys, 2_200).await;
         assert_eq!(reused, signed);
 
         let event = outbox
             .get_event(receipt.outbox_event_id)
             .await
             .expect("event")
             .expect("event");
         assert_eq!(event.state, RadrootsOutboxEventState::Publishing);
         assert_eq!(event.signed_event.as_ref(), Some(&signed));
     }
 
     #[tokio::test]
     async fn local_signed_event_ingest_is_idempotent_without_relay_observation() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let event_store = RadrootsEventStore::open_memory()
             .await
             .expect("event store");
         let (receipt, claimed) = enqueue_signed_fixture(&outbox).await;
         let keys = fixture_keys();
         let signed = complete_claimed_signing(&outbox, &claimed, &keys, 1_100).await;
 
         let first = outbox
             .ingest_signed_event_local(&event_store, receipt.outbox_event_id, "claim-a", 1_200)
             .await
             .expect("first ingest");
         assert_eq!(first.outbox_event_id, receipt.outbox_event_id);
         assert_eq!(first.event_id, signed.id);
         assert!(!first.already_ingested);
         assert!(first.event_store_inserted);
 
         let stored = event_store
             .get_event(signed.id.as_str())
             .await
             .expect("stored event");
         assert!(stored.is_some());
 
         let observations = event_store
             .observations_for_event(signed.id.as_str())
             .await
             .expect("observations");
         assert!(observations.is_empty());
 
         let second = outbox
             .ingest_signed_event_local(&event_store, receipt.outbox_event_id, "claim-a", 1_300)
             .await
             .expect("second ingest");
         assert!(second.already_ingested);
         assert!(!second.event_store_inserted);
 
         let event = outbox
             .get_event(receipt.outbox_event_id)
             .await
             .expect("event")
             .expect("event");
         assert_eq!(event.state, RadrootsOutboxEventState::Publishing);
         assert!(event.event_store_ingested);
         assert!(event.event_store_inserted);
         assert_eq!(event.event_store_ingested_at_ms, Some(1_200));
 
         let recovered = outbox.recover_expired_claims(2_001).await.expect("recover");
         assert_eq!(recovered, 1);
 
         let event = outbox
             .get_event(receipt.outbox_event_id)
             .await
             .expect("event")
             .expect("event");
         assert_eq!(event.state, RadrootsOutboxEventState::PublishRetryable);
         assert!(event.claim_token.is_none());
 
         let reclaimed = outbox
             .claim_next_ready_event("publisher-a", "claim-b", 3_000, 2_100)
             .await
             .expect("claim")
             .expect("publish claim");
         assert_eq!(reclaimed.state, RadrootsOutboxEventState::Publishing);
         assert_eq!(reclaimed.signed_event.as_ref(), Some(&signed));
     }
 
     #[tokio::test]
     async fn terminal_and_cancelled_claimed_events_are_not_reclaimable() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let terminal_draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "terminal");
         let terminal_receipt = outbox
             .enqueue_operation(operation_input(terminal_draft, 1_000))
             .await
             .expect("enqueue");
         outbox
             .claim_next_ready_event("worker-a", "claim-a", 2_000, 1_000)
             .await
             .expect("claim")
             .expect("claim");
         outbox
             .mark_publish_failed_terminal(
                 terminal_receipt.outbox_event_id,
                 "claim-a",
                 "terminal failure",
                 1_100,
             )
             .await
             .expect("terminal");
 
         let terminal_event = outbox
             .get_event(terminal_receipt.outbox_event_id)
             .await
             .expect("event")
             .expect("event");
         assert_eq!(
             terminal_event.state,
             RadrootsOutboxEventState::FailedTerminal
         );
         assert!(terminal_event.state.is_terminal());
         assert!(terminal_event.claim_token.is_none());
         let terminal_operation = outbox
             .get_operation(terminal_receipt.operation_id)
             .await
             .expect("operation")
             .expect("operation");
         assert_eq!(
             terminal_operation.status,
             RadrootsOutboxOperationStatus::FailedTerminal
         );
         assert!(
             outbox
                 .claim_next_ready_event("worker-b", "claim-b", 2_000, 1_200)
                 .await
                 .expect("claim")
                 .is_none()
         );
 
         let cancelled_draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "cancelled");
         let cancelled_receipt = outbox
             .enqueue_operation(operation_input(cancelled_draft, 2_000))
             .await
             .expect("enqueue");
         outbox
             .claim_next_ready_event("worker-c", "claim-c", 3_000, 2_000)
             .await
             .expect("claim")
             .expect("claim");
         outbox
             .cancel_claimed_event(cancelled_receipt.outbox_event_id, "claim-c", 2_100)
             .await
             .expect("cancel");
         let cancelled_event = outbox
             .get_event(cancelled_receipt.outbox_event_id)
             .await
             .expect("event")
             .expect("event");
         assert_eq!(cancelled_event.state, RadrootsOutboxEventState::Cancelled);
         assert!(cancelled_event.state.is_terminal());
         assert!(cancelled_event.claim_token.is_none());
         let cancelled_operation = outbox
             .get_operation(cancelled_receipt.operation_id)
             .await
             .expect("operation")
             .expect("operation");
         assert_eq!(
             cancelled_operation.status,
             RadrootsOutboxOperationStatus::Cancelled
         );
         assert!(
             outbox
                 .claim_next_ready_event("worker-d", "claim-d", 3_000, 2_200)
                 .await
                 .expect("claim")
                 .is_none()
         );
     }
 
     #[tokio::test]
     async fn terminal_publish_attempt_fails_operation_when_quorum_cannot_be_met() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "terminal quorum");
         let receipt = outbox
             .enqueue_operation(RadrootsOutboxOperationInput::new(
                 "publish_post",
                 draft,
                 vec![
                     RELAY_PRIMARY_WSS.to_owned(),
                     RELAY_SECONDARY_WSS.to_owned(),
                     "wss://relay-3.example.com".to_owned(),
                 ],
                 1_000,
             ))
             .await
             .expect("enqueue");
         let sign_claim = outbox
             .claim_next_ready_event("signer", "sign-a", 2_000, 1_000)
             .await
             .expect("claim")
             .expect("claim");
         complete_claimed_signing(&outbox, &sign_claim, &fixture_keys(), 1_100).await;
         outbox.recover_expired_claims(2_001).await.expect("recover");
         outbox
             .claim_next_ready_event("publisher", "publish-a", 3_000, 2_100)
             .await
             .expect("claim")
             .expect("claim");
         outbox
             .set_publish_quorum(receipt.outbox_event_id, "publish-a", 3, 2_200)
             .await
             .expect("quorum");
         outbox
             .mark_relay_accepted(
                 receipt.outbox_event_id,
                 "publish-a",
                 RELAY_PRIMARY_WSS,
                 2_250,
             )
             .await
             .expect("accepted");
         outbox
             .mark_relay_failed_terminal(
                 receipt.outbox_event_id,
                 "publish-a",
                 RELAY_SECONDARY_WSS,
                 "restricted: denied",
                 2_260,
             )
             .await
             .expect("terminal");
         outbox
             .mark_relay_failed_terminal(
                 receipt.outbox_event_id,
                 "publish-a",
                 "wss://relay-3.example.com",
                 "blocked: denied",
                 2_270,
             )
             .await
             .expect("terminal");
 
         let state = outbox
             .complete_publish_attempt(
                 receipt.outbox_event_id,
                 "publish-a",
                 "retryable",
                 "terminal",
                 2_500,
                 2_300,
             )
             .await
             .expect("complete attempt");
         assert_eq!(state, RadrootsOutboxEventState::FailedTerminal);
         let event = outbox
             .get_event(receipt.outbox_event_id)
             .await
             .expect("event")
             .expect("event");
         assert_eq!(event.state, RadrootsOutboxEventState::FailedTerminal);
         assert!(event.claim_token.is_none());
         assert_eq!(event.last_error.as_deref(), Some("terminal"));
         let operation = outbox
             .get_operation(receipt.operation_id)
             .await
             .expect("operation")
             .expect("operation");
         assert_eq!(
             operation.status,
             RadrootsOutboxOperationStatus::FailedTerminal
         );
         assert!(
             outbox
                 .claim_next_ready_event("publisher", "publish-b", 4_000, 2_400)
                 .await
                 .expect("claim")
                 .is_none()
         );
     }
 
     #[tokio::test]
     async fn publish_attempt_marks_event_published_after_acceptance_quorum() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "published quorum");
         let signed_event =
             radroots_nostr_sign_frozen_draft(&fixture_keys(), &draft).expect("signed event");
         let receipt = outbox
             .enqueue_signed_operation(signed_operation_input(draft, signed_event, 1_000))
             .await
             .expect("enqueue");
         let claimed = outbox
             .claim_next_ready_signed_event("publisher", "publish-a", 3_000, 1_000)
             .await
             .expect("claim")
             .expect("claim");
 
         outbox
             .mark_relay_accepted(
                 receipt.outbox_event_id,
                 claimed.claim_token.as_str(),
                 RELAY_PRIMARY_WSS,
                 1_100,
             )
             .await
             .expect("primary accepted");
         outbox
             .mark_relay_accepted(
                 receipt.outbox_event_id,
                 claimed.claim_token.as_str(),
                 RELAY_SECONDARY_WSS,
                 1_110,
             )
             .await
             .expect("secondary accepted");
 
         let state = outbox
             .complete_publish_attempt(
                 receipt.outbox_event_id,
                 claimed.claim_token.as_str(),
                 "retryable",
                 "terminal",
                 2_500,
                 1_200,
             )
             .await
             .expect("complete attempt");
         let event = outbox
             .get_event(receipt.outbox_event_id)
             .await
             .expect("event")
             .expect("event");
         let operation = outbox
             .get_operation(receipt.operation_id)
             .await
             .expect("operation")
             .expect("operation");
 
         assert_eq!(state, RadrootsOutboxEventState::Published);
         assert_eq!(event.state, RadrootsOutboxEventState::Published);
         assert_eq!(event.next_attempt_after_ms, 1_200);
         assert_eq!(operation.status, RadrootsOutboxOperationStatus::Complete);
     }
 
     #[tokio::test]
     async fn publish_attempt_remains_retryable_when_relay_work_remains() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "retryable quorum");
         let signed_event =
             radroots_nostr_sign_frozen_draft(&fixture_keys(), &draft).expect("signed event");
         let receipt = outbox
             .enqueue_signed_operation(signed_operation_input(draft, signed_event, 1_000))
             .await
             .expect("enqueue");
         let claimed = outbox
             .claim_next_ready_signed_event("publisher", "publish-a", 3_000, 1_000)
             .await
             .expect("claim")
             .expect("claim");
 
         outbox
             .mark_relay_failed_retryable(
                 receipt.outbox_event_id,
                 claimed.claim_token.as_str(),
                 RELAY_PRIMARY_WSS,
                 "timeout",
                 1_100,
             )
             .await
             .expect("retryable relay");
 
         let state = outbox
             .complete_publish_attempt(
                 receipt.outbox_event_id,
                 claimed.claim_token.as_str(),
                 "retryable error",
                 "terminal",
                 2_500,
                 1_200,
             )
             .await
             .expect("complete attempt");
         let event = outbox
             .get_event(receipt.outbox_event_id)
             .await
             .expect("event")
             .expect("event");
         let operation = outbox
             .get_operation(receipt.operation_id)
             .await
             .expect("operation")
             .expect("operation");
 
         assert_eq!(state, RadrootsOutboxEventState::PublishRetryable);
         assert_eq!(event.state, RadrootsOutboxEventState::PublishRetryable);
         assert_eq!(event.last_error.as_deref(), Some("retryable error"));
         assert_eq!(event.next_attempt_after_ms, 2_500);
         assert_eq!(operation.status, RadrootsOutboxOperationStatus::Queued);
     }
 
     #[tokio::test]
     async fn publish_attempt_remains_retryable_when_pending_relay_work_remains() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let draft = post_draft(FIXTURE_ALICE_PUBLIC_KEY_HEX, "pending quorum");
         let signed_event =
             radroots_nostr_sign_frozen_draft(&fixture_keys(), &draft).expect("signed event");
         let receipt = outbox
             .enqueue_signed_operation(signed_operation_input(draft, signed_event, 1_000))
             .await
             .expect("enqueue");
         let claimed = outbox
             .claim_next_ready_signed_event("publisher", "publish-a", 3_000, 1_000)
             .await
             .expect("claim")
             .expect("claim");
 
         outbox
             .mark_relay_accepted(
                 receipt.outbox_event_id,
                 claimed.claim_token.as_str(),
                 RELAY_PRIMARY_WSS,
                 1_100,
             )
             .await
             .expect("accepted relay");
 
         let state = outbox
             .complete_publish_attempt(
                 receipt.outbox_event_id,
                 claimed.claim_token.as_str(),
                 "pending relay",
                 "terminal",
                 2_500,
                 1_200,
             )
             .await
             .expect("complete attempt");
         let event = outbox
             .get_event(receipt.outbox_event_id)
             .await
             .expect("event")
             .expect("event");
 
         assert_eq!(state, RadrootsOutboxEventState::PublishRetryable);
         assert_eq!(event.state, RadrootsOutboxEventState::PublishRetryable);
         assert_eq!(event.last_error.as_deref(), Some("pending relay"));
         assert_eq!(event.next_attempt_after_ms, 2_500);
     }
 
     #[tokio::test]
     async fn smoke_outbox_claim_cancel_cycles_complete_one_thousand_events() {
         let outbox = RadrootsOutbox::open_memory().await.expect("open");
         let mut receipts = Vec::new();
         for index in 0..1_000 {
             let draft = post_draft(
                 FIXTURE_ALICE_PUBLIC_KEY_HEX,
                 format!("claim-cycle-{index}").as_str(),
             );
             let receipt = outbox
                 .enqueue_operation(operation_input(draft, 1_000 + index))
                 .await
                 .expect("enqueue");
             receipts.push(receipt);
         }
 
         for index in 0..1_000 {
             let claim_token = format!("claim-{index}");
             let claimed = outbox
                 .claim_next_ready_event(
                     "smoke-worker",
                     claim_token.as_str(),
                     10_000 + index,
                     2_000 + index,
                 )
                 .await
                 .expect("claim")
                 .expect("claimed");
             outbox
                 .cancel_claimed_event(claimed.outbox_event_id, claim_token.as_str(), 3_000 + index)
                 .await
                 .expect("cancel");
         }
 
         for receipt in receipts {
             let event = outbox
                 .get_event(receipt.outbox_event_id)
                 .await
                 .expect("event")
                 .expect("event");
             assert_eq!(event.state, RadrootsOutboxEventState::Cancelled);
             assert!(event.claim_token.is_none());
             let operation = outbox
                 .get_operation(receipt.operation_id)
                 .await
                 .expect("operation")
                 .expect("operation");
             assert_eq!(operation.status, RadrootsOutboxOperationStatus::Cancelled);
         }
 
         assert!(
             outbox
                 .claim_next_ready_event("smoke-worker", "claim-final", 20_000, 20_000)
                 .await
                 .expect("claim")
                 .is_none()
         );
     }
 }