Implement authorized live event replay

crates/remora-api/src/lib.rs

@@ -1,7 +1,12 @@
+mod migrate;
 mod public;
 mod worker;
 
-pub use public::{PublicApiConfig, PublicApiServices, PublicApiState, configure_public_api};
+pub use migrate::{migrate, migrator};
+pub use public::{
+    PublicApiConfig, PublicApiServices, PublicApiState, configure_public_api,
+    prune_live_event_history,
+};
 pub use worker::{
     AuthenticatedWorker, WorkerApiState, authenticate_rustls_worker, configure_worker_api,
 };

crates/remora-api/src/migrate.rs

@@ -0,0 +1,16 @@
+use remora_durable::DurableError;
+use sqlx::PgPool;
+
+const MIGRATION_TABLE: &str = "_remora_api_migrations";
+
+pub async fn migrate(pool: &PgPool) -> Result<(), DurableError> {
+    let mut migrator = sqlx::migrate!("./migrations");
+    migrator.dangerous_set_table_name(MIGRATION_TABLE);
+    migrator.run(pool).await.map_err(DurableError::from)
+}
+
+pub fn migrator() -> sqlx::migrate::Migrator {
+    let mut migrator = sqlx::migrate!("./migrations");
+    migrator.dangerous_set_table_name(MIGRATION_TABLE);
+    migrator
+}

crates/remora-api/src/public.rs

@@ -5,7 +5,7 @@ use actix_web::{
     http::{StatusCode, header},
     web,
 };
-use remora_admission::{AdmissionDecision, AdmissionError, AdmissionService};
+use remora_admission::{AdmissionDecision, AdmissionService};
 use remora_attempt::{AttemptService, CommittedResultRecord, ProgressSnapshot};
 use remora_contracts::{
     ApiErrorCodeDto, ApiErrorDto, CredentialMaterialDto, DateTimeDto, JobSubmissionDto,
@@ -39,6 +39,8 @@ use serde_json::{Value, json};
 use sqlx::{PgPool, types::Json};
 use time::{OffsetDateTime, format_description::well_known::Rfc3339};
 
+mod events;
+
 const DEFAULT_LIMIT: u32 = 50;
 const MAX_LIMIT: u32 = 200;
 
@@ -46,6 +48,9 @@ const MAX_LIMIT: u32 = 200;
 pub struct PublicApiConfig {
     pub cursor_secret: Vec<u8>,
     pub member_access_ttl: Duration,
+    pub event_retention: Duration,
+    pub event_poll_interval: Duration,
+    pub event_heartbeat_interval: Duration,
 }
 
 #[derive(Clone)]
@@ -60,6 +65,9 @@ pub struct PublicApiState {
     results: ResultInventory,
     cursors: CursorCodec,
     member_access_ttl: Duration,
+    event_retention: Duration,
+    event_poll_interval: Duration,
+    event_heartbeat_interval: Duration,
 }
 
 pub struct PublicApiServices {
@@ -82,6 +90,14 @@ impl PublicApiState {
                 "member access expiry must be positive".into(),
             ));
         }
+        if config.event_retention.is_zero()
+            || config.event_poll_interval.is_zero()
+            || config.event_heartbeat_interval.is_zero()
+        {
+            return Err(DurableError::Config(
+                "event retention and intervals must be positive".into(),
+            ));
+        }
         Ok(Self {
             results: ResultInventory::new(pool.clone()),
             cursors: CursorCodec::new(config.cursor_secret)?,
@@ -93,6 +109,9 @@ impl PublicApiState {
             deletion: services.deletion,
             targets: services.targets,
             member_access_ttl: config.member_access_ttl,
+            event_retention: config.event_retention,
+            event_poll_interval: config.event_poll_interval,
+            event_heartbeat_interval: config.event_heartbeat_interval,
         })
     }
 
@@ -173,10 +192,18 @@ pub fn configure_public_api(config: &mut web::ServiceConfig) {
             .route(
                 "/deletions/{operation_id}",
                 web::get().to(get_deletion_operation),
-            ),
+            )
+            .route("/events", web::get().to(events::connect)),
     );
 }
 
+pub async fn prune_live_event_history(
+    pool: &PgPool,
+    before: OffsetDateTime,
+) -> Result<u64, DurableError> {
+    events::prune_history(pool, before).await
+}
+
 #[derive(Deserialize)]
 struct PageQuery {
     cursor: Option<String>,
@@ -198,17 +225,69 @@ async fn submit_job(
 ) -> Result<HttpResponse, PublicApiError> {
     let principal = state.principal(&http).await?;
     let key = idempotency_key(&http)?;
-    let outcome = state
-        .admission
-        .submit(&principal, key, request.into_inner())
-        .await
-        .map_err(PublicApiError::from_admission)?;
-    let accepted = match outcome.decision {
-        AdmissionDecision::Accepted(accepted) => accepted,
-        AdmissionDecision::Rejected { .. } => return Err(PublicApiError::invalid_request()),
+    let request = request.into_inner();
+    let request_fingerprint =
+        fingerprint(&request).map_err(|_| PublicApiError::invalid_request())?;
+    let admission = state.admission.clone();
+    let state_for_projection = state.get_ref().clone();
+    let principal_for_effect = principal.clone();
+    let outcome = execute_mutation(
+        &state.pool,
+        MutationScope {
+            subject: &principal.id,
+            operation: "job.submit",
+            key,
+        },
+        "submit_job",
+        &[],
+        request_fingerprint,
+        move |tx| {
+            Box::pin(async move {
+                match admission
+                    .submit_in_tx(tx, &principal_for_effect, request)
+                    .await?
+                {
+                    AdmissionDecision::Accepted(accepted) => {
+                        let projection = load_authorized_job_in_tx(
+                            tx,
+                            &state_for_projection,
+                            &principal_for_effect,
+                            &accepted.job_id,
+                            false,
+                        )
+                        .await
+                        .map_err(|error| DurableError::Effect(error.to_string()))?;
+                        Ok((MutationStatus::Accepted, projection))
+                    }
+                    AdmissionDecision::Rejected { .. } => Ok((
+                        MutationStatus::Rejected,
+                        public_error_body(ApiErrorCodeDto::InvalidRequest, "invalid request"),
+                    )),
+                }
+            })
+        },
+    )
+    .await?;
+    if outcome.disposition == MutationDisposition::Conflict {
+        return Err(PublicApiError::idempotency_conflict());
+    }
+    if outcome.status == MutationStatus::Accepted {
+        let job_id = outcome
+            .response
+            .get("id")
+            .and_then(Value::as_str)
+            .ok_or_else(PublicApiError::internal)?;
+        let job_id = parse_id(job_id.to_owned())?;
+        authorize_job(&state, &principal, &job_id, ManagementOperation::Read).await?;
+    }
+    let response = if outcome.status == MutationStatus::Rejected {
+        HttpResponse::BadRequest()
+            .content_type("application/problem+json")
+            .json(outcome.response)
+    } else {
+        HttpResponse::Accepted().json(outcome.response)
     };
-    let job = load_authorized_job(&state, &principal, &accepted.job_id, false).await?;
-    Ok(with_idempotency_expiry(HttpResponse::Accepted().json(job)))
+    Ok(with_idempotency_expiry(response))
 }
 
 async fn get_job(
@@ -241,6 +320,7 @@ async fn list_jobs(
         .map(|token| state.cursors.bind(token, principal.id.as_str(), filter))
         .transpose()
         .map_err(PublicApiError::from_cursor)?;
+    let event_cursor = event_cursor(&state, &principal).await?;
     let mut rows = load_job_page_rows(
         &state.pool,
         &principal,
@@ -251,7 +331,6 @@ async fn list_jobs(
     .await?;
     let has_more = rows.len() > limit as usize;
     rows.truncate(limit as usize);
-    let event_cursor = event_cursor(&state, &principal).await?;
     let mut items = Vec::with_capacity(rows.len());
     for row in &rows {
         items.push(job_row_projection(
@@ -1073,6 +1152,7 @@ async fn load_authorized_job(
     job_id: &OpaqueId,
     include_deleted: bool,
 ) -> Result<Value, PublicApiError> {
+    let cursor = event_cursor(state, principal).await?;
     let row = load_job_row(&state.pool, job_id)
         .await?
         .ok_or_else(PublicApiError::not_found)?;
@@ -1080,7 +1160,6 @@ async fn load_authorized_job(
         return Err(PublicApiError::not_found());
     }
     authorize_row(principal, &row, ManagementOperation::Read)?;
-    let cursor = event_cursor(state, principal).await?;
     let progress = current_progress(state, &row).await?;
     job_row_projection(&row, progress, &cursor)
 }
@@ -1654,11 +1733,16 @@ async fn event_cursor(
     principal: &Principal,
 ) -> Result<String, PublicApiError> {
     let high_water = sqlx::query_as::<_, (i64, OffsetDateTime)>(
-        "SELECT id, created_at FROM outbox_messages ORDER BY id DESC LIMIT 1",
+        "SELECT GREATEST(
+                    COALESCE((SELECT MAX(id) FROM outbox_messages), 0),
+                    history.floor_outbox_id
+                ),
+                now()
+         FROM live_event_history_state history
+         WHERE history.singleton",
     )
-    .fetch_optional(&state.pool)
-    .await?
-    .unwrap_or((0, OffsetDateTime::UNIX_EPOCH));
+    .fetch_one(&state.pool)
+    .await?;
     encode_event_cursor(state, principal, high_water)
 }
 
@@ -1668,11 +1752,16 @@ async fn event_cursor_in_tx(
     principal: &Principal,
 ) -> Result<String, PublicApiError> {
     let high_water = sqlx::query_as::<_, (i64, OffsetDateTime)>(
-        "SELECT id, created_at FROM outbox_messages ORDER BY id DESC LIMIT 1",
+        "SELECT GREATEST(
+                    COALESCE((SELECT MAX(id) FROM outbox_messages), 0),
+                    history.floor_outbox_id
+                ),
+                now()
+         FROM live_event_history_state history
+         WHERE history.singleton",
     )
-    .fetch_optional(tx.as_mut())
-    .await?
-    .unwrap_or((0, OffsetDateTime::UNIX_EPOCH));
+    .fetch_one(tx.as_mut())
+    .await?;
     encode_event_cursor(state, principal, high_water)
 }
 
@@ -1721,6 +1810,7 @@ fn authorization_snapshot(principal: &Principal) -> Value {
         })
         .collect::<Vec<_>>();
     scopes.sort_unstable();
+    scopes.dedup();
     json!({
         "administrator": principal.administrator,
         "scopes": scopes
@@ -1875,15 +1965,6 @@ impl PublicApiError {
         }
     }
 
-    fn from_admission(error: AdmissionError) -> Self {
-        match error {
-            AdmissionError::IdempotencyConflict => Self::idempotency_conflict(),
-            AdmissionError::Unauthorized => Self::forbidden(),
-            AdmissionError::Invalid(_) => Self::invalid_request(),
-            _ => Self::unavailable(),
-        }
-    }
-
     fn from_credential(error: CredentialError) -> Self {
         match error {
             CredentialError::NotFound => Self::not_found(),
@@ -1988,6 +2069,7 @@ mod tests {
         };
         let first = principal(false, vec![scope("b"), scope("a")]);
         let reordered = principal(false, vec![scope("a"), scope("b")]);
+        let duplicated = principal(false, vec![scope("a"), scope("b"), scope("a")]);
         let expanded = principal(false, vec![scope("a"), scope("b"), scope("c")]);
         let administrator = principal(true, vec![scope("a"), scope("b")]);
 
@@ -1995,6 +2077,10 @@ mod tests {
             authorization_snapshot(&first),
             authorization_snapshot(&reordered)
         );
+        assert_eq!(
+            authorization_snapshot(&first),
+            authorization_snapshot(&duplicated)
+        );
         assert_ne!(
             authorization_snapshot(&first),
             authorization_snapshot(&expanded)

crates/remora-api/src/public/events.rs

@@ -0,0 +1,860 @@
+use actix_web::{HttpRequest, HttpResponse, web};
+use futures_util::StreamExt as _;
+use remora_contracts::{
+    CursorDto, DateTimeDto, DeletionChangedDto, DeletionStateDto, EventAttemptDto, EventFrameDto,
+    EventProgressDto, HeartbeatDto, JobChangedDto, NullableDto, PositiveIntegerDto,
+    ResyncReasonDto, ResyncRequiredDto, SafeIntegerDto, SchemaVersionDto,
+};
+use remora_domain::{JobLifecycle, JobPhase, OpaqueId};
+use remora_durable::{DurableError, KeysetCursor};
+use remora_identity::{ManagementOperation, Principal};
+use serde::Deserialize;
+use serde_json::Value;
+use sqlx::{FromRow, PgPool};
+use time::{OffsetDateTime, format_description::well_known::Rfc3339};
+
+use super::{
+    JobRow, PublicApiError, PublicApiState, authorization_snapshot, authorize_row,
+    current_progress, encode_event_cursor, event_cursor, load_job_row, parse_id, request_id,
+};
+
+const EVENT_TOPICS: [&str; 2] = ["job.updated", "deletion.updated"];
+const EVENT_BATCH: i64 = 100;
+
+#[derive(Deserialize)]
+pub(super) struct EventQuery {
+    cursor: Option<String>,
+}
+
+pub(super) async fn connect(
+    state: web::Data<PublicApiState>,
+    request: HttpRequest,
+    body: web::Payload,
+    query: web::Query<EventQuery>,
+) -> Result<HttpResponse, PublicApiError> {
+    let principal = state.principal(&request).await?;
+    let now = OffsetDateTime::now_utc();
+    let retention =
+        time::Duration::try_from(state.event_retention).map_err(|_| PublicApiError::internal())?;
+    prune_history(&state.pool, now - retention).await?;
+    let start = classify_cursor(&state, &principal, query.cursor.as_deref(), now).await?;
+    let (response, session, messages) =
+        actix_ws::handle(&request, body).map_err(|_| PublicApiError::invalid_request())?;
+    let state = state.get_ref().clone();
+    actix_web::rt::spawn(async move {
+        match start {
+            EventStart::Replay(after_id) => {
+                run_connection(state, principal, after_id, session, messages).await;
+            }
+            EventStart::Resync(reason) => {
+                send_resync(state, principal, reason, session).await;
+            }
+        }
+    });
+    Ok(response)
+}
+
+pub(super) async fn prune_history(
+    pool: &PgPool,
+    before: OffsetDateTime,
+) -> Result<u64, DurableError> {
+    let mut tx = pool.begin().await?;
+    let pruned: Vec<i64> = sqlx::query_scalar(
+        "DELETE FROM outbox_messages
+         WHERE topic = ANY($1) AND created_at < $2
+         RETURNING id",
+    )
+    .bind(&EVENT_TOPICS[..])
+    .bind(before)
+    .fetch_all(tx.as_mut())
+    .await?;
+    if let Some(floor) = pruned.iter().max() {
+        sqlx::query(
+            "UPDATE live_event_history_state
+             SET floor_outbox_id = GREATEST(floor_outbox_id, $1)
+             WHERE singleton",
+        )
+        .bind(floor)
+        .execute(tx.as_mut())
+        .await?;
+    }
+    tx.commit().await?;
+    Ok(pruned.len() as u64)
+}
+
+enum EventStart {
+    Replay(i64),
+    Resync(ResyncReasonDto),
+}
+
+async fn classify_cursor(
+    state: &PublicApiState,
+    principal: &Principal,
+    token: Option<&str>,
+    now: OffsetDateTime,
+) -> Result<EventStart, PublicApiError> {
+    let Some(token) = token else {
+        return Ok(EventStart::Resync(ResyncReasonDto::CursorAbsent));
+    };
+    let after_id =
+        match decode_event_cursor(&state.cursors, state.event_retention, principal, token, now)? {
+            Ok(after_id) => after_id,
+            Err(reason) => return Ok(EventStart::Resync(reason)),
+        };
+    let (floor, maximum): (i64, i64) = sqlx::query_as(
+        "SELECT history.floor_outbox_id,
+                GREATEST(
+                    COALESCE((SELECT MAX(id) FROM outbox_messages), 0),
+                    history.floor_outbox_id
+                )
+         FROM live_event_history_state history
+         WHERE history.singleton",
+    )
+    .fetch_one(&state.pool)
+    .await?;
+    if after_id < floor || after_id > maximum {
+        return Ok(EventStart::Resync(ResyncReasonDto::HistoryUnavailable));
+    }
+    Ok(EventStart::Replay(after_id))
+}
+
+fn decode_event_cursor(
+    cursors: &remora_durable::CursorCodec,
+    retention: std::time::Duration,
+    principal: &Principal,
+    token: &str,
+    now: OffsetDateTime,
+) -> Result<Result<i64, ResyncReasonDto>, PublicApiError> {
+    let cursor = match cursors.decode(token) {
+        Ok(cursor) => cursor,
+        Err(_) => return Ok(Err(ResyncReasonDto::CursorInvalid)),
+    };
+    let Some(after_id) = event_position(&cursor) else {
+        return Ok(Err(ResyncReasonDto::CursorInvalid));
+    };
+    let expected = remora_durable::CursorCodec::filter_fingerprint(&serde_json::json!({
+        "resource": "events",
+        "authorization": authorization_snapshot(principal)
+    }))?;
+    if cursor.subject != principal.id.as_str() || cursor.filter_fingerprint != expected {
+        return Ok(Err(ResyncReasonDto::AuthorizationChanged));
+    }
+    let retention = time::Duration::try_from(retention).map_err(|_| PublicApiError::internal())?;
+    if cursor.high_water.created_at + retention < now {
+        return Ok(Err(ResyncReasonDto::CursorExpired));
+    }
+    Ok(Ok(after_id))
+}
+
+fn event_position(cursor: &KeysetCursor) -> Option<i64> {
+    let high_water = cursor
+        .high_water
+        .id
+        .strip_prefix("event-")?
+        .parse::<i64>()
+        .ok()?;
+    let after = cursor.after.as_ref()?;
+    let after_id = after.id.strip_prefix("event-")?.parse::<i64>().ok()?;
+    (high_water == after_id && after.created_at == cursor.high_water.created_at).then_some(after_id)
+}
+
+async fn send_resync(
+    state: PublicApiState,
+    principal: Principal,
+    reason: ResyncReasonDto,
+    mut session: actix_ws::Session,
+) {
+    let Ok(cursor) = event_cursor(&state, &principal).await else {
+        let _ = session.close(None).await;
+        return;
+    };
+    let Ok(frame) = resync_frame(reason, 1, &cursor, OffsetDateTime::now_utc()) else {
+        let _ = session.close(None).await;
+        return;
+    };
+    let _ = send_frame(&mut session, &frame).await;
+    let _ = session.close(None).await;
+}
+
+async fn run_connection(
+    state: PublicApiState,
+    principal: Principal,
+    mut after_id: i64,
+    mut session: actix_ws::Session,
+    mut messages: actix_ws::MessageStream,
+) {
+    let mut sequence = 0_u64;
+    let mut poll = tokio::time::interval(state.event_poll_interval);
+    let mut heartbeat = tokio::time::interval(state.event_heartbeat_interval);
+    poll.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    heartbeat.tick().await;
+    loop {
+        tokio::select! {
+            message = messages.next() => {
+                match message {
+                    Some(Ok(actix_ws::Message::Ping(bytes))) => {
+                        if session.pong(&bytes).await.is_err() {
+                            break;
+                        }
+                    }
+                    Some(Ok(actix_ws::Message::Pong(_))) => {}
+                    Some(Ok(actix_ws::Message::Close(reason))) => {
+                        let _ = session.close(reason).await;
+                        return;
+                    }
+                    Some(Err(_)) | None => break,
+                    Some(Ok(_)) => break,
+                }
+            }
+            _ = poll.tick() => {
+                if poll_progress(&state).await.is_err() {
+                    break;
+                }
+                match replay_events(
+                    &state,
+                    &principal,
+                    &mut after_id,
+                    &mut sequence,
+                    &mut session,
+                ).await {
+                    Ok(ReplayStatus::Current) => {}
+                    Ok(ReplayStatus::HistoryUnavailable) => {
+                        let now = OffsetDateTime::now_utc();
+                        let Ok(cursor) = event_cursor(&state, &principal).await else {
+                            break;
+                        };
+                        let Ok(frame) = resync_frame(
+                            ResyncReasonDto::HistoryUnavailable,
+                            next_sequence(&mut sequence),
+                            &cursor,
+                            now,
+                        ) else {
+                            break;
+                        };
+                        let _ = send_frame(&mut session, &frame).await;
+                        break;
+                    }
+                    Err(()) => break,
+                }
+            }
+            _ = heartbeat.tick() => {
+                let now = OffsetDateTime::now_utc();
+                let Ok(cursor) = encode_event_cursor(&state, &principal, (after_id, now)) else {
+                    break;
+                };
+                let Ok(frame) = heartbeat_frame(next_sequence(&mut sequence), &cursor, now) else {
+                    break;
+                };
+                if send_frame(&mut session, &frame).await.is_err() {
+                    break;
+                }
+            }
+        }
+    }
+    let _ = session.close(None).await;
+}
+
+#[derive(FromRow)]
+struct ReplayRow {
+    id: i64,
+    topic: String,
+    payload: Value,
+    created_at: OffsetDateTime,
+}
+
+#[derive(FromRow)]
+struct ReplayBatchRow {
+    floor_outbox_id: i64,
+    id: Option<i64>,
+    topic: Option<String>,
+    payload: Option<Value>,
+    created_at: Option<OffsetDateTime>,
+}
+
+enum ReplayStatus {
+    Current,
+    HistoryUnavailable,
+}
+
+async fn replay_events(
+    state: &PublicApiState,
+    principal: &Principal,
+    after_id: &mut i64,
+    sequence: &mut u64,
+    session: &mut actix_ws::Session,
+) -> Result<ReplayStatus, ()> {
+    for _ in 0..10 {
+        let batch = sqlx::query_as::<_, ReplayBatchRow>(
+            "SELECT history.floor_outbox_id, events.id, events.topic,
+                    events.payload, events.created_at
+             FROM live_event_history_state history
+             LEFT JOIN LATERAL (
+                 SELECT id, topic, payload, created_at
+                 FROM outbox_messages
+                 WHERE id > $1 AND topic = ANY($2)
+                 ORDER BY id
+                 LIMIT $3
+             ) events ON TRUE
+             WHERE history.singleton
+             ORDER BY events.id",
+        )
+        .bind(*after_id)
+        .bind(&EVENT_TOPICS[..])
+        .bind(EVENT_BATCH)
+        .fetch_all(&state.pool)
+        .await
+        .map_err(|_| ())?;
+        let Some(first) = batch.first() else {
+            return Err(());
+        };
+        if *after_id < first.floor_outbox_id {
+            return Ok(ReplayStatus::HistoryUnavailable);
+        }
+        let rows = batch
+            .into_iter()
+            .filter_map(|row| {
+                Some(ReplayRow {
+                    id: row.id?,
+                    topic: row.topic?,
+                    payload: row.payload?,
+                    created_at: row.created_at?,
+                })
+            })
+            .collect::<Vec<_>>();
+        let count = rows.len();
+        for row in rows {
+            *after_id = row.id;
+            let now = OffsetDateTime::now_utc();
+            let cursor = encode_event_cursor(state, principal, (row.id, now)).map_err(|_| ())?;
+            let candidate_sequence = sequence.saturating_add(1);
+            let frame = match row.topic.as_str() {
+                "job.updated" => {
+                    project_job_event(state, principal, &row, candidate_sequence, &cursor).await
+                }
+                "deletion.updated" => {
+                    project_deletion_event(state, principal, &row, candidate_sequence, &cursor)
+                        .await
+                }
+                _ => None,
+            };
+            if let Some(frame) = frame {
+                *sequence = candidate_sequence;
+                send_frame(session, &frame).await?;
+            }
+        }
+        if count < EVENT_BATCH as usize {
+            break;
+        }
+    }
+    Ok(ReplayStatus::Current)
+}
+
+async fn project_job_event(
+    state: &PublicApiState,
+    principal: &Principal,
+    event: &ReplayRow,
+    sequence: u64,
+    cursor: &str,
+) -> Option<EventFrameDto> {
+    let job_id = parse_payload_id(&event.payload, "job_id")?;
+    let row = load_job_row(&state.pool, &job_id).await.ok()??;
+    authorize_row(principal, &row, ManagementOperation::Read).ok()?;
+    let progress = current_progress(state, &row)
+        .await
+        .ok()
+        .flatten()
+        .map(progress_fingerprint);
+    job_frame(
+        &row,
+        progress.as_ref(),
+        stable_event_id(event.id),
+        sequence,
+        cursor,
+        event.created_at,
+    )
+    .ok()
+}
+
+async fn project_deletion_event(
+    state: &PublicApiState,
+    principal: &Principal,
+    event: &ReplayRow,
+    sequence: u64,
+    cursor: &str,
+) -> Option<EventFrameDto> {
+    let operation_id = parse_payload_id(&event.payload, "operation_id")?;
+    let (job_id, state_name): (String, String) =
+        sqlx::query_as("SELECT job_id, state FROM deletion_operations WHERE operation_id = $1")
+            .bind(operation_id.as_str())
+            .fetch_optional(&state.pool)
+            .await
+            .ok()??;
+    let job_id = parse_id(job_id).ok()?;
+    let row = load_job_row(&state.pool, &job_id).await.ok()??;
+    authorize_row(principal, &row, ManagementOperation::Read).ok()?;
+    deletion_frame(
+        job_id,
+        operation_id,
+        parse_deletion_state(&state_name)?,
+        stable_event_id(event.id),
+        sequence,
+        cursor,
+        event.created_at,
+    )
+    .ok()
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+struct ProgressFingerprint {
+    attempt_id: OpaqueId,
+    fence: u64,
+    completed_bytes: u64,
+    total_bytes: Option<u64>,
+    completed_members: u64,
+    total_members: Option<u64>,
+}
+
+fn progress_fingerprint(snapshot: remora_attempt::ProgressSnapshot) -> ProgressFingerprint {
+    ProgressFingerprint {
+        attempt_id: snapshot.attempt_id,
+        fence: snapshot.fence,
+        completed_bytes: snapshot.progress.completed_bytes,
+        total_bytes: snapshot.progress.total_bytes,
+        completed_members: snapshot.progress.completed_members,
+        total_members: snapshot.progress.total_members,
+    }
+}
+
+async fn poll_progress(state: &PublicApiState) -> Result<(), ()> {
+    let delay = state.event_poll_interval.as_secs_f64();
+    let claimed = sqlx::query_scalar::<_, bool>(
+        "UPDATE live_event_history_state
+         SET progress_poll_not_before = now() + ($1 * interval '1 second')
+         WHERE singleton AND progress_poll_not_before <= now()
+         RETURNING singleton",
+    )
+    .bind(delay)
+    .fetch_optional(&state.pool)
+    .await
+    .map_err(|_| ())?
+    .is_some();
+    if !claimed {
+        return Ok(());
+    }
+    let mut after_job = None::<String>;
+    loop {
+        let job_ids: Vec<String> = sqlx::query_scalar(
+            "SELECT job_id FROM jobs
+             WHERE phase IN ('running', 'cancelling')
+               AND ($1::text IS NULL OR job_id > $1)
+             ORDER BY job_id
+             LIMIT 200",
+        )
+        .bind(after_job.as_deref())
+        .fetch_all(&state.pool)
+        .await
+        .map_err(|_| ())?;
+        if job_ids.is_empty() {
+            break;
+        }
+        let count = job_ids.len();
+        for job_id in job_ids {
+            after_job = Some(job_id.clone());
+            let Some(row) = load_job_row(&state.pool, &parse_id(job_id.clone()).map_err(|_| ())?)
+                .await
+                .map_err(|_| ())?
+            else {
+                continue;
+            };
+            let progress = current_progress(state, &row)
+                .await
+                .ok()
+                .flatten()
+                .map(progress_fingerprint);
+            record_progress_change(&state.pool, &row, progress.as_ref())
+                .await
+                .map_err(|_| ())?;
+        }
+        if count < 200 {
+            break;
+        }
+    }
+    Ok(())
+}
+
+async fn record_progress_change(
+    pool: &PgPool,
+    row: &JobRow,
+    progress: Option<&ProgressFingerprint>,
+) -> Result<(), PublicApiError> {
+    let Some(attempt_id) = row.authority_attempt_id.as_deref() else {
+        return Ok(());
+    };
+    let mut tx = pool.begin().await?;
+    let now = OffsetDateTime::now_utc();
+    let changed = if let Some(progress) = progress {
+        sqlx::query_scalar::<_, String>(
+            "INSERT INTO live_progress_event_state (
+                job_id, attempt_id, fence, progress_present, completed_bytes,
+                total_bytes, completed_members, total_members, updated_at
+             ) VALUES ($1, $2, $3, TRUE, $4, $5, $6, $7, $8)
+             ON CONFLICT (job_id) DO UPDATE
+             SET attempt_id = EXCLUDED.attempt_id,
+                 fence = EXCLUDED.fence,
+                 progress_present = TRUE,
+                 completed_bytes = EXCLUDED.completed_bytes,
+                 total_bytes = EXCLUDED.total_bytes,
+                 completed_members = EXCLUDED.completed_members,
+                 total_members = EXCLUDED.total_members,
+                 updated_at = EXCLUDED.updated_at
+             WHERE (
+                 live_progress_event_state.attempt_id,
+                 live_progress_event_state.fence,
+                 live_progress_event_state.progress_present,
+                 live_progress_event_state.completed_bytes,
+                 live_progress_event_state.total_bytes,
+                 live_progress_event_state.completed_members,
+                 live_progress_event_state.total_members
+             ) IS DISTINCT FROM (
+                 EXCLUDED.attempt_id,
+                 EXCLUDED.fence,
+                 TRUE,
+                 EXCLUDED.completed_bytes,
+                 EXCLUDED.total_bytes,
+                 EXCLUDED.completed_members,
+                 EXCLUDED.total_members
+             )
+             RETURNING job_id",
+        )
+        .bind(&row.job_id)
+        .bind(progress.attempt_id.as_str())
+        .bind(i64::try_from(progress.fence).map_err(|_| PublicApiError::internal())?)
+        .bind(i64::try_from(progress.completed_bytes).map_err(|_| PublicApiError::internal())?)
+        .bind(
+            progress
+                .total_bytes
+                .map(i64::try_from)
+                .transpose()
+                .map_err(|_| PublicApiError::internal())?,
+        )
+        .bind(i64::try_from(progress.completed_members).map_err(|_| PublicApiError::internal())?)
+        .bind(
+            progress
+                .total_members
+                .map(i64::try_from)
+                .transpose()
+                .map_err(|_| PublicApiError::internal())?,
+        )
+        .bind(now)
+        .fetch_optional(tx.as_mut())
+        .await?
+        .is_some()
+    } else {
+        sqlx::query_scalar::<_, String>(
+            "UPDATE live_progress_event_state
+             SET attempt_id = $2, fence = $3, progress_present = FALSE,
+                 completed_bytes = NULL, total_bytes = NULL,
+                 completed_members = NULL, total_members = NULL, updated_at = $4
+             WHERE job_id = $1
+               AND (
+                   attempt_id IS DISTINCT FROM $2
+                   OR fence IS DISTINCT FROM $3
+                   OR progress_present
+                   OR completed_bytes IS NOT NULL
+                   OR total_bytes IS NOT NULL
+                   OR completed_members IS NOT NULL
+                   OR total_members IS NOT NULL
+               )
+             RETURNING job_id",
+        )
+        .bind(&row.job_id)
+        .bind(attempt_id)
+        .bind(row.current_fence)
+        .bind(now)
+        .fetch_optional(tx.as_mut())
+        .await?
+        .is_some()
+    };
+    if changed {
+        remora_durable::enqueue(
+            &mut tx,
+            "job.updated",
+            serde_json::json!({ "job_id": row.job_id }),
+        )
+        .await?;
+    }
+    tx.commit().await?;
+    Ok(())
+}
+
+fn job_frame(
+    row: &JobRow,
+    progress: Option<&ProgressFingerprint>,
+    event_id: OpaqueId,
+    sequence: u64,
+    cursor: &str,
+    occurred_at: OffsetDateTime,
+) -> Result<EventFrameDto, PublicApiError> {
+    let authoritative_attempt = row
+        .authority_attempt_id
+        .as_deref()
+        .map(|attempt_id| {
+            Ok::<_, PublicApiError>(EventAttemptDto {
+                attempt_id: parse_id(attempt_id)?,
+                fence: PositiveIntegerDto::new(
+                    u64::try_from(row.current_fence).map_err(|_| PublicApiError::internal())?,
+                )
+                .map_err(|_| PublicApiError::internal())?,
+            })
+        })
+        .transpose()?;
+    let progress = progress
+        .filter(|progress| {
+            row.authority_attempt_id.as_deref() == Some(progress.attempt_id.as_str())
+                && u64::try_from(row.current_fence).ok() == Some(progress.fence)
+        })
+        .map(event_progress)
+        .transpose()?;
+    Ok(EventFrameDto::JobChanged(JobChangedDto {
+        schema_version: SchemaVersionDto,
+        event_id,
+        sequence: positive(sequence)?,
+        cursor: event_cursor_dto(cursor)?,
+        occurred_at: event_time(occurred_at)?,
+        job_id: parse_id(row.job_id.clone())?,
+        job_revision: positive(
+            u64::try_from(row.revision).map_err(|_| PublicApiError::internal())?,
+        )?,
+        phase: parse_phase(&row.phase)?,
+        lifecycle: parse_lifecycle(&row.lifecycle)?,
+        authoritative_attempt: NullableDto(authoritative_attempt),
+        progress: NullableDto(progress),
+    }))
+}
+
+fn deletion_frame(
+    job_id: OpaqueId,
+    operation_id: OpaqueId,
+    state: DeletionStateDto,
+    event_id: OpaqueId,
+    sequence: u64,
+    cursor: &str,
+    occurred_at: OffsetDateTime,
+) -> Result<EventFrameDto, PublicApiError> {
+    Ok(EventFrameDto::DeletionChanged(DeletionChangedDto {
+        schema_version: SchemaVersionDto,
+        event_id,
+        sequence: positive(sequence)?,
+        cursor: event_cursor_dto(cursor)?,
+        occurred_at: event_time(occurred_at)?,
+        job_id,
+        operation_id,
+        state,
+    }))
+}
+
+fn resync_frame(
+    reason: ResyncReasonDto,
+    sequence: u64,
+    cursor: &str,
+    occurred_at: OffsetDateTime,
+) -> Result<EventFrameDto, PublicApiError> {
+    Ok(EventFrameDto::ResyncRequired(ResyncRequiredDto {
+        schema_version: SchemaVersionDto,
+        event_id: request_id(),
+        sequence: positive(sequence)?,
+        cursor: event_cursor_dto(cursor)?,
+        occurred_at: event_time(occurred_at)?,
+        reason,
+    }))
+}
+
+fn heartbeat_frame(
+    sequence: u64,
+    cursor: &str,
+    occurred_at: OffsetDateTime,
+) -> Result<EventFrameDto, PublicApiError> {
+    Ok(EventFrameDto::Heartbeat(HeartbeatDto {
+        schema_version: SchemaVersionDto,
+        event_id: request_id(),
+        sequence: positive(sequence)?,
+        cursor: event_cursor_dto(cursor)?,
+        occurred_at: event_time(occurred_at)?,
+    }))
+}
+
+fn event_progress(progress: &ProgressFingerprint) -> Result<EventProgressDto, PublicApiError> {
+    Ok(EventProgressDto {
+        attempt_id: progress.attempt_id.clone(),
+        fence: positive(progress.fence)?,
+        completed_bytes: safe(progress.completed_bytes)?,
+        total_bytes: progress
+            .total_bytes
+            .map(|value| safe(value).map(|value| NullableDto(Some(value))))
+            .transpose()?,
+        completed_members: safe(progress.completed_members)?,
+        total_members: progress
+            .total_members
+            .map(|value| safe(value).map(|value| NullableDto(Some(value))))
+            .transpose()?,
+    })
+}
+
+fn parse_payload_id(payload: &Value, field: &str) -> Option<OpaqueId> {
+    payload
+        .get(field)?
+        .as_str()
+        .and_then(|value| OpaqueId::parse(value).ok())
+}
+
+fn stable_event_id(id: i64) -> OpaqueId {
+    OpaqueId::parse(format!("event_{id:016}")).expect("positive outbox ids form opaque event ids")
+}
+
+fn parse_phase(value: &str) -> Result<JobPhase, PublicApiError> {
+    match value {
+        "queued" => Ok(JobPhase::Queued),
+        "running" => Ok(JobPhase::Running),
+        "retry_wait" => Ok(JobPhase::RetryWait),
+        "cancelling" => Ok(JobPhase::Cancelling),
+        "succeeded" => Ok(JobPhase::Succeeded),
+        "failed" => Ok(JobPhase::Failed),
+        "cancelled" => Ok(JobPhase::Cancelled),
+        _ => Err(PublicApiError::internal()),
+    }
+}
+
+fn parse_lifecycle(value: &str) -> Result<JobLifecycle, PublicApiError> {
+    match value {
+        "visible" => Ok(JobLifecycle::Visible),
+        "soft_deleted" => Ok(JobLifecycle::SoftDeleted),
+        "hard_deleting" => Ok(JobLifecycle::HardDeleting),
+        "hard_deleted" => Ok(JobLifecycle::HardDeleted),
+        _ => Err(PublicApiError::internal()),
+    }
+}
+
+fn parse_deletion_state(value: &str) -> Option<DeletionStateDto> {
+    match value {
+        "queued" => Some(DeletionStateDto::Queued),
+        "running" => Some(DeletionStateDto::Running),
+        "retry_wait" => Some(DeletionStateDto::RetryWait),
+        "blocked" => Some(DeletionStateDto::Blocked),
+        "completed" => Some(DeletionStateDto::Completed),
+        _ => None,
+    }
+}
+
+fn positive(value: u64) -> Result<PositiveIntegerDto, PublicApiError> {
+    PositiveIntegerDto::new(value).map_err(|_| PublicApiError::internal())
+}
+
+fn safe(value: u64) -> Result<SafeIntegerDto, PublicApiError> {
+    SafeIntegerDto::new(value).map_err(|_| PublicApiError::internal())
+}
+
+fn event_cursor_dto(value: &str) -> Result<CursorDto, PublicApiError> {
+    CursorDto::parse(value).map_err(|_| PublicApiError::internal())
+}
+
+fn event_time(value: OffsetDateTime) -> Result<DateTimeDto, PublicApiError> {
+    DateTimeDto::parse(
+        value
+            .format(&Rfc3339)
+            .map_err(|_| PublicApiError::internal())?,
+    )
+    .map_err(|_| PublicApiError::internal())
+}
+
+fn next_sequence(sequence: &mut u64) -> u64 {
+    *sequence = sequence.saturating_add(1);
+    *sequence
+}
+
+async fn send_frame(session: &mut actix_ws::Session, frame: &EventFrameDto) -> Result<(), ()> {
+    session
+        .text(serde_json::to_string(frame).map_err(|_| ())?)
+        .await
+        .map_err(|_| ())
+}
+
+#[cfg(test)]
+mod tests {
+    use remora_domain::{BucketName, StorageScope, TargetPrefix, TargetStoreId};
+    use remora_durable::{CursorCodec, KeysetPosition, first_page_cursor};
+
+    use super::*;
+
+    #[test]
+    fn event_cursor_rejects_other_subject_and_non_event_positions() {
+        let codec = CursorCodec::new(b"event-cursor-test-secret").unwrap();
+        let scope = StorageScope {
+            store: TargetStoreId::parse("target-main").unwrap(),
+            bucket: BucketName::parse("target-bucket").unwrap(),
+            prefix: TargetPrefix::parse("models", std::iter::empty()).unwrap(),
+        };
+        let principal = Principal {
+            id: OpaqueId::parse("principal_one").unwrap(),
+            issuer: "https://issuer.example".into(),
+            subject: "one".into(),
+            administrator: false,
+            scopes: vec![scope],
+        };
+        let fingerprint = CursorCodec::filter_fingerprint(&serde_json::json!({
+            "resource": "events",
+            "authorization": authorization_snapshot(&principal)
+        }))
+        .unwrap();
+        let issued = OffsetDateTime::now_utc();
+        let cursor = first_page_cursor(
+            "principal_two",
+            fingerprint,
+            KeysetPosition {
+                created_at: issued,
+                id: "event-12".into(),
+            },
+            KeysetPosition {
+                created_at: issued,
+                id: "event-12".into(),
+            },
+        );
+        let token = codec.encode(&cursor).unwrap();
+        assert_eq!(
+            decode_event_cursor(
+                &codec,
+                std::time::Duration::from_secs(60),
+                &principal,
+                &token,
+                issued,
+            )
+            .unwrap(),
+            Err(ResyncReasonDto::AuthorizationChanged)
+        );
+
+        let page = first_page_cursor(
+            principal.id.as_str(),
+            fingerprint,
+            KeysetPosition {
+                created_at: issued,
+                id: "job_000001".into(),
+            },
+            KeysetPosition {
+                created_at: issued,
+                id: "job_000001".into(),
+            },
+        );
+        let token = codec.encode(&page).unwrap();
+        assert_eq!(
+            decode_event_cursor(
+                &codec,
+                std::time::Duration::from_secs(60),
+                &principal,
+                &token,
+                issued,
+            )
+            .unwrap(),
+            Err(ResyncReasonDto::CursorInvalid)
+        );
+    }
+}