use std::{net::TcpListener, time::Duration};

use actix_web::{App, HttpServer, web};
use futures_util::{SinkExt as _, StreamExt as _};
use remora_admission::{AdmissionConfig, AdmissionService, migrate as admission_migrate};
use remora_api::{
    PublicApiConfig, PublicApiServices, PublicApiState, configure_public_api,
    migrate as api_migrate, prune_live_event_history,
};
use remora_attempt::{AttemptConfig, AttemptService, migrate as attempt_migrate};
use remora_contracts::{EventFrameDto, ResyncReasonDto};
use remora_coordination::{
    ActivateOptions, CoordinationStore, LeaseOwner, ProgressValue, StoreConfig,
};
use remora_credentials::{
    CredentialRegistry, VaultClient, VaultConfig, migrate as credentials_migrate,
};
use remora_deletion::{
    DeletionConfig, DeletionService, TargetRegistry, migrate as deletion_migrate,
};
use remora_domain::{BucketName, OpaqueId, TargetPrefix, TargetStoreId};
use remora_durable::{PoolConfig, connect, enqueue, migrate as durable_migrate};
use remora_identity::{DefaultTarget, migrate as identity_migrate};
use remora_result::migrate as result_migrate;
use reqwest::{Client, StatusCode};
use serde_json::{Value, json};
use sqlx::PgPool;
use time::OffsetDateTime;
use tokio_tungstenite::{
    MaybeTlsStream, WebSocketStream, connect_async,
    tungstenite::{
        Error as WebSocketError, Message,
        client::IntoClientRequest,
        http::{
            HeaderValue,
            header::{AUTHORIZATION, ORIGIN},
        },
    },
};

mod support;

use support::{
    mint_provider_token, oidc_provider, required_env, restore_remora_grants, revoke_remora_grants,
};

type Socket = WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>;
const BROWSER_ORIGIN: &str = "https://remora.test";

struct Harness {
    pool: PgPool,
    coordination: CoordinationStore,
    client: Client,
    token: String,
    http_base: String,
    ws_base: String,
    server: actix_web::dev::ServerHandle,
}

impl Harness {
    async fn new() -> Option<Self> {
        if std::env::var("REMORA_LIVE_EVENTS_REQUIRE").is_err() {
            return None;
        }
        restore_remora_grants().await.unwrap();
        let pool = connect(
            &required_env("DATABASE_URL"),
            PoolConfig {
                max_connections: 20,
                acquire_timeout: Duration::from_secs(10),
                idle_timeout: Some(Duration::from_secs(30)),
            },
        )
        .await
        .unwrap();
        durable_migrate(&pool).await.unwrap();
        identity_migrate(&pool).await.unwrap();
        credentials_migrate(&pool).await.unwrap();
        admission_migrate(&pool).await.unwrap();
        attempt_migrate(&pool).await.unwrap();
        result_migrate(&pool).await.unwrap();
        deletion_migrate(&pool).await.unwrap();
        api_migrate(&pool).await.unwrap();
        sqlx::query("TRUNCATE TABLE jobs CASCADE")
            .execute(&pool)
            .await
            .unwrap();
        sqlx::query("TRUNCATE TABLE outbox_messages RESTART IDENTITY")
            .execute(&pool)
            .await
            .unwrap();
        sqlx::query(
            "UPDATE live_event_history_state
             SET floor_outbox_id = 0, progress_poll_not_before = '-infinity'
             WHERE singleton",
        )
        .execute(&pool)
        .await
        .unwrap();

        let coordination = CoordinationStore::connect(
            &required_env("REMORA_COORDINATION_URL"),
            format!(
                "{}wp22:{}:",
                required_env("REMORA_COORDINATION_NAMESPACE"),
                nonce()
            ),
            StoreConfig {
                global_limit: 16,
                lease_ttl: Duration::from_secs(10),
                reserve_ttl: Duration::from_secs(10),
                progress_ttl: Duration::from_secs(10),
                cancel_ttl: None,
            },
        )
        .await
        .unwrap();
        let attempts = AttemptService::new(
            pool.clone(),
            coordination.clone(),
            AttemptConfig { max_attempts: 3 },
        )
        .unwrap();
        let admission = AdmissionService::new(
            pool.clone(),
            AdmissionConfig {
                default_target: DefaultTarget {
                    store: TargetStoreId::parse("target-main").unwrap(),
                    bucket: BucketName::parse("target-bucket").unwrap(),
                    prefix: TargetPrefix::parse("models", ["_remora"]).unwrap(),
                },
                reserved_target_segments: vec!["_remora".into()],
                policy_version: "live-events-v1".into(),
            },
        )
        .unwrap();
        let vault = VaultClient::new(VaultConfig {
            address: "http://127.0.0.1:8200".into(),
            token: "synthetic-unused-token".into(),
            mount: "secret".into(),
            path_prefix: "unused".into(),
            timeout: Duration::from_secs(1),
        })
        .unwrap();
        let credentials = CredentialRegistry::new(pool.clone(), vault);
        let targets = TargetRegistry::default();
        let deletion =
            DeletionService::new(pool.clone(), targets.clone(), DeletionConfig::default()).unwrap();
        let state = PublicApiState::new(
            pool.clone(),
            PublicApiServices {
                identity: oidc_provider(),
                admission,
                attempts,
                credentials,
                deletion,
                targets,
            },
            PublicApiConfig {
                cursor_secret: b"live-events-fixture-cursor-secret".to_vec(),
                member_access_ttl: Duration::from_secs(60),
                event_retention: Duration::from_secs(2),
                event_poll_interval: Duration::from_millis(250),
                event_heartbeat_interval: Duration::from_millis(100),
                event_ticket_ttl: Duration::from_secs(30),
                event_allowed_origins: vec!["https://remora.test".into()],
            },
        )
        .unwrap();
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let address = listener.local_addr().unwrap();
        let server = HttpServer::new(move || {
            App::new()
                .app_data(web::Data::new(state.clone()))
                .configure(configure_public_api)
        })
        .listen(listener)
        .unwrap()
        .run();
        let handle = server.handle();
        actix_web::rt::spawn(server);
        let token = mint_provider_token().await.unwrap();
        Some(Self {
            pool,
            coordination,
            client: Client::new(),
            token,
            http_base: format!("http://{address}/v1"),
            ws_base: format!("ws://{address}/v1/events"),
            server: handle,
        })
    }

    async fn submit_job(&self) -> Value {
        let response = self
            .client
            .post(format!("{}/jobs", self.http_base))
            .bearer_auth(&self.token)
            .header("Idempotency-Key", idempotency_key())
            .json(&json!({
                "label": "live events",
                "source": {
                    "kind": "http",
                    "url": "https://example.invalid/live",
                    "result_path": "payload.bin"
                }
            }))
            .send()
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::ACCEPTED);
        response.json().await.unwrap()
    }

    async fn get_job(&self, token: &str, job_id: &str) -> reqwest::Response {
        self.client
            .get(format!("{}/jobs/{job_id}", self.http_base))
            .bearer_auth(token)
            .send()
            .await
            .unwrap()
    }

    async fn list_jobs(&self, token: &str) -> Value {
        let response = self
            .client
            .get(format!("{}/jobs", self.http_base))
            .bearer_auth(token)
            .send()
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::OK);
        response.json().await.unwrap()
    }

    async fn issue_ticket(&self, token: &str) -> String {
        let response = self
            .client
            .post(format!("{}/event-tickets", self.http_base))
            .bearer_auth(token)
            .header("Origin", BROWSER_ORIGIN)
            .send()
            .await
            .unwrap();
        assert_eq!(response.status(), StatusCode::CREATED);
        assert_eq!(response.headers().get("Cache-Control").unwrap(), "no-store");
        response.json::<Value>().await.unwrap()["ticket"]
            .as_str()
            .unwrap()
            .to_owned()
    }

    async fn connect(&self, token: &str, cursor: Option<&str>) -> Socket {
        let ticket = self.issue_ticket(token).await;
        self.connect_ticket(&ticket, cursor, BROWSER_ORIGIN)
            .await
            .unwrap()
    }

    async fn connect_ticket(
        &self,
        ticket: &str,
        cursor: Option<&str>,
        origin: &str,
    ) -> Result<Socket, WebSocketError> {
        let url = cursor.map_or_else(
            || format!("{}?ticket={ticket}", self.ws_base),
            |cursor| format!("{}?ticket={ticket}&cursor={cursor}", self.ws_base),
        );
        let mut request = url.into_client_request().unwrap();
        request
            .headers_mut()
            .insert(ORIGIN, HeaderValue::from_str(origin).unwrap());
        let (socket, response) = connect_async(request).await?;
        assert_eq!(response.status(), 101);
        Ok(socket)
    }

    async fn connect_bearer(&self, token: &str, cursor: Option<&str>) -> Socket {
        let url = cursor.map_or_else(
            || self.ws_base.clone(),
            |cursor| format!("{}?cursor={cursor}", self.ws_base),
        );
        let mut request = url.into_client_request().unwrap();
        request.headers_mut().insert(
            AUTHORIZATION,
            HeaderValue::from_str(&format!("Bearer {token}")).unwrap(),
        );
        let (socket, response) = connect_async(request).await.unwrap();
        assert_eq!(response.status(), 101);
        socket
    }
}

impl Drop for Harness {
    fn drop(&mut self) {
        actix_web::rt::spawn(self.server.stop(true));
    }
}

#[actix_web::test]
async fn live_event_replay_authorization_progress_and_resync_converge() {
    let Some(harness) = Harness::new().await else {
        return;
    };

    let ticket = harness.issue_ticket(&harness.token).await;
    let forbidden_ticket = harness
        .client
        .post(format!("{}/event-tickets", harness.http_base))
        .bearer_auth(&harness.token)
        .header("Origin", "https://attacker.invalid")
        .send()
        .await
        .unwrap();
    assert_eq!(forbidden_ticket.status(), StatusCode::FORBIDDEN);
    assert!(
        harness
            .connect_ticket(&ticket, None, "https://attacker.invalid")
            .await
            .is_err()
    );
    let valid = harness
        .connect_ticket(&ticket, None, BROWSER_ORIGIN)
        .await
        .unwrap();
    drop(valid);
    assert!(
        harness
            .connect_ticket(&ticket, None, BROWSER_ORIGIN)
            .await
            .is_err()
    );

    let concurrent_ticket = harness.issue_ticket(&harness.token).await;
    let (left, right) = tokio::join!(
        harness.connect_ticket(&concurrent_ticket, None, BROWSER_ORIGIN),
        harness.connect_ticket(&concurrent_ticket, None, BROWSER_ORIGIN)
    );
    assert_eq!(usize::from(left.is_ok()) + usize::from(right.is_ok()), 1);

    let expired_ticket = harness.issue_ticket(&harness.token).await;
    sqlx::query(
        "UPDATE live_event_tickets
         SET expires_at = created_at + interval '1 microsecond'
         WHERE expires_at > now()",
    )
    .execute(&harness.pool)
    .await
    .unwrap();
    assert!(
        harness
            .connect_ticket(&expired_ticket, None, BROWSER_ORIGIN)
            .await
            .is_err()
    );

    let mut bearer = harness.connect_bearer(&harness.token, None).await;
    assert_resync(next_frame(&mut bearer).await, ResyncReasonDto::CursorAbsent);

    let mut absent = harness.connect(&harness.token, None).await;
    assert_resync(next_frame(&mut absent).await, ResyncReasonDto::CursorAbsent);

    let submitted = harness.submit_job().await;
    let job_id = submitted["id"].as_str().unwrap().to_owned();
    let initial_cursor = submitted["event_cursor"].as_str().unwrap().to_owned();

    let mut invalid = harness
        .connect(&harness.token, Some(&format!("{initial_cursor}x")))
        .await;
    assert_resync(
        next_frame(&mut invalid).await,
        ResyncReasonDto::CursorInvalid,
    );

    update_job_and_enqueue(&harness.pool, &job_id).await;
    update_job_and_enqueue(&harness.pool, &job_id).await;
    let mut first_replay = harness.connect(&harness.token, Some(&initial_cursor)).await;
    let first = next_job_frame(&mut first_replay, &job_id).await;
    let second = next_job_frame(&mut first_replay, &job_id).await;
    let (first_id, first_sequence) = job_event_identity(&first);
    let (second_id, second_sequence) = job_event_identity(&second);
    assert_ne!(first_id, second_id);
    assert_eq!((first_sequence, second_sequence), (1, 2));
    let replay_cursor = frame_cursor(&second).to_owned();

    let mut duplicate_replay = harness.connect(&harness.token, Some(&initial_cursor)).await;
    let duplicate_first = next_job_frame(&mut duplicate_replay, &job_id).await;
    let duplicate_second = next_job_frame(&mut duplicate_replay, &job_id).await;
    assert_eq!(job_event_identity(&duplicate_first).0, first_id);
    assert_eq!(job_event_identity(&duplicate_second).0, second_id);

    let mut caught_up = harness.connect(&harness.token, Some(&replay_cursor)).await;
    assert!(matches!(
        next_frame(&mut caught_up).await,
        EventFrameDto::Heartbeat(_)
    ));
    first_replay.close(None).await.unwrap();
    duplicate_replay.close(None).await.unwrap();
    caught_up.close(None).await.unwrap();

    let fresh = harness
        .get_job(&harness.token, &job_id)
        .await
        .json::<Value>()
        .await
        .unwrap();
    let expiring_cursor = fresh["event_cursor"].as_str().unwrap().to_owned();
    tokio::time::sleep(Duration::from_millis(2_100)).await;
    let mut expired = harness
        .connect(&harness.token, Some(&expiring_cursor))
        .await;
    assert_resync(
        next_frame(&mut expired).await,
        ResyncReasonDto::CursorExpired,
    );

    let before_gap = harness
        .get_job(&harness.token, &job_id)
        .await
        .json::<Value>()
        .await
        .unwrap()["event_cursor"]
        .as_str()
        .unwrap()
        .to_owned();
    update_job_and_enqueue(&harness.pool, &job_id).await;
    assert!(
        prune_live_event_history(
            &harness.pool,
            OffsetDateTime::now_utc() + time::Duration::seconds(1)
        )
        .await
        .unwrap()
            > 0
    );
    let mut unavailable = harness.connect(&harness.token, Some(&before_gap)).await;
    assert_resync(
        next_frame(&mut unavailable).await,
        ResyncReasonDto::HistoryUnavailable,
    );
    let after_gap = harness
        .get_job(&harness.token, &job_id)
        .await
        .json::<Value>()
        .await
        .unwrap()["event_cursor"]
        .as_str()
        .unwrap()
        .to_owned();
    let mut recovered = harness.connect(&harness.token, Some(&after_gap)).await;
    assert!(matches!(
        next_frame(&mut recovered).await,
        EventFrameDto::Heartbeat(_)
    ));
    recovered.close(None).await.unwrap();

    let active_gap_cursor = harness
        .get_job(&harness.token, &job_id)
        .await
        .json::<Value>()
        .await
        .unwrap()["event_cursor"]
        .as_str()
        .unwrap()
        .to_owned();
    let mut active_gap = harness
        .connect(&harness.token, Some(&active_gap_cursor))
        .await;
    assert!(matches!(
        next_frame(&mut active_gap).await,
        EventFrameDto::Heartbeat(_)
    ));
    update_job_and_enqueue(&harness.pool, &job_id).await;
    prune_live_event_history(
        &harness.pool,
        OffsetDateTime::now_utc() + time::Duration::seconds(1),
    )
    .await
    .unwrap();
    assert_resync_reason(
        next_resync(&mut active_gap).await,
        ResyncReasonDto::HistoryUnavailable,
    );

    let first_attempt = insert_running_attempt(&harness.pool, &job_id, 1).await;
    let first_owner = LeaseOwner::new(first_attempt.as_str(), 1, "worker_live_01").unwrap();
    assert!(
        harness
            .coordination
            .activate(&job_id, &first_owner, ActivateOptions::default())
            .await
            .unwrap()
            .ok()
    );
    let progress_cursor = harness
        .get_job(&harness.token, &job_id)
        .await
        .json::<Value>()
        .await
        .unwrap()["event_cursor"]
        .as_str()
        .unwrap()
        .to_owned();
    let mut progress_socket = harness
        .connect(&harness.token, Some(&progress_cursor))
        .await;
    assert!(
        harness
            .coordination
            .report_progress_value(
                &job_id,
                &first_owner,
                ProgressValue {
                    completed_bytes: 42,
                    total_bytes: Some(100),
                    completed_members: 1,
                    total_members: Some(2),
                },
            )
            .await
            .unwrap()
            .ok()
    );
    let progress = next_job_progress(&mut progress_socket, &job_id, Some(42)).await;
    assert_matching_progress(&progress, &first_attempt, 1, 42);
    let mut progress_replay = harness
        .connect(&harness.token, Some(&progress_cursor))
        .await;
    let replayed_progress = next_job_progress(&mut progress_replay, &job_id, Some(42)).await;
    assert_eq!(
        job_event_identity(&replayed_progress).0,
        job_event_identity(&progress).0
    );

    let second_attempt = insert_running_attempt(&harness.pool, &job_id, 2).await;
    let stale_suppressed = next_job_progress(&mut progress_socket, &job_id, None).await;
    assert_job_authority(&stale_suppressed, &second_attempt, 2);

    assert!(
        harness
            .coordination
            .release(&job_id, &first_owner)
            .await
            .unwrap()
            .ok()
    );
    harness.coordination.clean_terminal(&job_id).await.unwrap();
    let second_owner = LeaseOwner::new(second_attempt.as_str(), 2, "worker_live_02").unwrap();
    assert!(
        harness
            .coordination
            .activate(&job_id, &second_owner, ActivateOptions::default())
            .await
            .unwrap()
            .ok()
    );
    assert!(
        harness
            .coordination
            .report_progress(&job_id, &second_owner, 77)
            .await
            .unwrap()
            .ok()
    );
    let restored_progress = next_job_progress(&mut progress_socket, &job_id, Some(77)).await;
    assert_matching_progress(&restored_progress, &second_attempt, 2, 77);
    assert!(
        harness
            .coordination
            .release(&job_id, &second_owner)
            .await
            .unwrap()
            .ok()
    );
    harness.coordination.clean_terminal(&job_id).await.unwrap();
    let coordination_lost = next_job_progress(&mut progress_socket, &job_id, None).await;
    assert_job_authority(&coordination_lost, &second_attempt, 2);
    let snapshot_after_loss: Value = harness
        .get_job(&harness.token, &job_id)
        .await
        .json()
        .await
        .unwrap();
    assert!(snapshot_after_loss["progress"].is_null());

    transition_cancelled(&harness.pool, &job_id).await;
    let deletion_cursor = harness
        .get_job(&harness.token, &job_id)
        .await
        .json::<Value>()
        .await
        .unwrap()["event_cursor"]
        .as_str()
        .unwrap()
        .to_owned();
    let deletion_response = harness
        .client
        .post(format!("{}/jobs/{job_id}/hard-delete", harness.http_base))
        .bearer_auth(&harness.token)
        .header("Idempotency-Key", idempotency_key())
        .send()
        .await
        .unwrap();
    assert_eq!(deletion_response.status(), StatusCode::ACCEPTED);
    let operation: Value = deletion_response.json().await.unwrap();
    let operation_id = operation["id"].as_str().unwrap();
    let mut deletion_socket = harness
        .connect(&harness.token, Some(&deletion_cursor))
        .await;
    let deletion = next_deletion_frame(&mut deletion_socket, operation_id).await;
    let EventFrameDto::DeletionChanged(deletion) = deletion else {
        unreachable!()
    };
    assert_eq!(deletion.job_id.as_str(), job_id);
    assert_eq!(
        serde_json::to_value(deletion.state).unwrap(),
        json!("completed")
    );

    let authorized_cursor = harness.list_jobs(&harness.token).await["event_cursor"]
        .as_str()
        .unwrap()
        .to_owned();
    revoke_remora_grants().await.unwrap();
    let reduced_token = mint_provider_token().await.unwrap();
    let mut changed = harness
        .connect(&reduced_token, Some(&authorized_cursor))
        .await;
    assert_resync(
        next_frame(&mut changed).await,
        ResyncReasonDto::AuthorizationChanged,
    );
    assert_eq!(
        harness.get_job(&reduced_token, &job_id).await.status(),
        StatusCode::NOT_FOUND
    );
    let reduced_snapshot = harness.list_jobs(&reduced_token).await;
    assert!(reduced_snapshot["items"].as_array().unwrap().is_empty());
    let reduced_cursor = reduced_snapshot["event_cursor"].as_str().unwrap();
    let mut filtered = harness.connect(&reduced_token, Some(reduced_cursor)).await;
    enqueue_job_event(&harness.pool, &job_id).await;
    for _ in 0..2 {
        assert!(matches!(
            next_frame(&mut filtered).await,
            EventFrameDto::Heartbeat(_)
        ));
    }

    progress_socket.close(None).await.unwrap();
    progress_replay.close(None).await.unwrap();
    deletion_socket.close(None).await.unwrap();
    filtered.close(None).await.unwrap();
}

async fn update_job_and_enqueue(pool: &PgPool, job_id: &str) {
    let mut tx = pool.begin().await.unwrap();
    sqlx::query("UPDATE jobs SET revision = revision + 1, updated_at = now() WHERE job_id = $1")
        .bind(job_id)
        .execute(tx.as_mut())
        .await
        .unwrap();
    enqueue(&mut tx, "job.updated", json!({ "job_id": job_id }))
        .await
        .unwrap();
    tx.commit().await.unwrap();
}

async fn enqueue_job_event(pool: &PgPool, job_id: &str) {
    let mut tx = pool.begin().await.unwrap();
    enqueue(&mut tx, "job.updated", json!({ "job_id": job_id }))
        .await
        .unwrap();
    tx.commit().await.unwrap();
}

async fn insert_running_attempt(pool: &PgPool, job_id: &str, fence: i64) -> OpaqueId {
    let attempt_id = opaque_id("attempt");
    sqlx::query(
        "INSERT INTO attempts (
            attempt_id, job_id, fence, worker_incarnation, state, initial_progress,
            authorized_at, executing_at, execution_envelope, staging_origin_attempt_id
         ) VALUES ($1, $2, $3, $4, 'executing', 0, now(), now(), '{}'::jsonb, $1)",
    )
    .bind(attempt_id.as_str())
    .bind(job_id)
    .bind(fence)
    .bind(format!("worker_live_{fence:02}"))
    .execute(pool)
    .await
    .unwrap();
    let mut tx = pool.begin().await.unwrap();
    sqlx::query(
        "UPDATE jobs
         SET phase = 'running', current_fence = $2, authority_attempt_id = $3,
             revision = revision + 1, updated_at = now()
         WHERE job_id = $1",
    )
    .bind(job_id)
    .bind(fence)
    .bind(attempt_id.as_str())
    .execute(tx.as_mut())
    .await
    .unwrap();
    enqueue(&mut tx, "job.updated", json!({ "job_id": job_id }))
        .await
        .unwrap();
    tx.commit().await.unwrap();
    attempt_id
}

async fn transition_cancelled(pool: &PgPool, job_id: &str) {
    let mut tx = pool.begin().await.unwrap();
    sqlx::query(
        "UPDATE jobs
         SET phase = 'cancelled', cancel_requested = TRUE, authority_attempt_id = NULL,
             dispatch_needed = FALSE, terminal_at = now(), revision = revision + 1,
             updated_at = now()
         WHERE job_id = $1",
    )
    .bind(job_id)
    .execute(tx.as_mut())
    .await
    .unwrap();
    enqueue(&mut tx, "job.updated", json!({ "job_id": job_id }))
        .await
        .unwrap();
    tx.commit().await.unwrap();
}

async fn next_frame(socket: &mut Socket) -> EventFrameDto {
    tokio::time::timeout(Duration::from_secs(3), async {
        loop {
            match socket.next().await {
                Some(Ok(Message::Text(text))) => {
                    return serde_json::from_str(text.as_ref()).unwrap();
                }
                Some(Ok(Message::Ping(bytes))) => {
                    socket.send(Message::Pong(bytes)).await.unwrap();
                }
                Some(Ok(Message::Close(reason))) => {
                    panic!("websocket closed before a frame: {reason:?}");
                }
                Some(Ok(_)) => {}
                Some(Err(error)) => panic!("websocket error: {error}"),
                None => panic!("websocket ended before a frame"),
            }
        }
    })
    .await
    .expect("event frame timeout")
}

async fn next_job_frame(socket: &mut Socket, job_id: &str) -> EventFrameDto {
    tokio::time::timeout(Duration::from_secs(10), async {
        loop {
            let frame = next_frame(socket).await;
            match &frame {
                EventFrameDto::JobChanged(event) if event.job_id.as_str() == job_id => {
                    return frame;
                }
                EventFrameDto::ResyncRequired(event) => {
                    panic!(
                        "unexpected resync while waiting for job event: {:?}",
                        event.reason
                    )
                }
                _ => {}
            }
        }
    })
    .await
    .expect("job event timeout")
}

async fn next_job_progress(
    socket: &mut Socket,
    job_id: &str,
    completed_bytes: Option<u64>,
) -> EventFrameDto {
    tokio::time::timeout(Duration::from_secs(10), async {
        loop {
            let frame = next_job_frame(socket, job_id).await;
            let EventFrameDto::JobChanged(event) = &frame else {
                unreachable!()
            };
            let observed = event
                .progress
                .0
                .as_ref()
                .map(|progress| progress.completed_bytes.get());
            if observed == completed_bytes {
                return frame;
            }
        }
    })
    .await
    .expect("job progress timeout")
}

async fn next_deletion_frame(socket: &mut Socket, operation_id: &str) -> EventFrameDto {
    tokio::time::timeout(Duration::from_secs(10), async {
        loop {
            let frame = next_frame(socket).await;
            if matches!(
                &frame,
                EventFrameDto::DeletionChanged(event)
                    if event.operation_id.as_str() == operation_id
            ) {
                return frame;
            }
        }
    })
    .await
    .expect("deletion event timeout")
}

async fn next_resync(socket: &mut Socket) -> EventFrameDto {
    tokio::time::timeout(Duration::from_secs(10), async {
        loop {
            let frame = next_frame(socket).await;
            if matches!(frame, EventFrameDto::ResyncRequired(_)) {
                return frame;
            }
        }
    })
    .await
    .expect("resync event timeout")
}

fn assert_resync(frame: EventFrameDto, expected: ResyncReasonDto) {
    let sequence = assert_resync_reason(frame, expected);
    assert_eq!(sequence, 1);
}

fn assert_resync_reason(frame: EventFrameDto, expected: ResyncReasonDto) -> u64 {
    let EventFrameDto::ResyncRequired(frame) = frame else {
        panic!("expected resync_required, got {frame:?}");
    };
    assert_eq!(frame.reason, expected);
    frame.sequence.get()
}

fn job_event_identity(frame: &EventFrameDto) -> (&str, u64) {
    let EventFrameDto::JobChanged(frame) = frame else {
        panic!("expected job_changed");
    };
    (frame.event_id.as_str(), frame.sequence.get())
}

fn frame_cursor(frame: &EventFrameDto) -> &str {
    match frame {
        EventFrameDto::JobChanged(frame) => frame.cursor.as_str(),
        EventFrameDto::DeletionChanged(frame) => frame.cursor.as_str(),
        EventFrameDto::ResyncRequired(frame) => frame.cursor.as_str(),
        EventFrameDto::Heartbeat(frame) => frame.cursor.as_str(),
    }
}

fn assert_matching_progress(
    frame: &EventFrameDto,
    attempt_id: &OpaqueId,
    fence: u64,
    completed_bytes: u64,
) {
    let EventFrameDto::JobChanged(frame) = frame else {
        panic!("expected job_changed");
    };
    let progress = frame.progress.0.as_ref().expect("matching progress");
    assert_eq!(progress.attempt_id, *attempt_id);
    assert_eq!(progress.fence.get(), fence);
    assert_eq!(progress.completed_bytes.get(), completed_bytes);
}

fn assert_job_authority(frame: &EventFrameDto, attempt_id: &OpaqueId, fence: u64) {
    let EventFrameDto::JobChanged(frame) = frame else {
        panic!("expected job_changed");
    };
    assert!(frame.progress.0.is_none());
    let authority = frame
        .authoritative_attempt
        .0
        .as_ref()
        .expect("authoritative attempt");
    assert_eq!(authority.attempt_id, *attempt_id);
    assert_eq!(authority.fence.get(), fence);
}

fn opaque_id(prefix: &str) -> OpaqueId {
    OpaqueId::parse(format!("{prefix}_{}", nonce())).unwrap()
}

fn idempotency_key() -> String {
    format!("key-{}-replay-safe", nonce())
}

fn nonce() -> String {
    let mut bytes = [0_u8; 8];
    getrandom::fill(&mut bytes).unwrap();
    bytes.iter().map(|byte| format!("{byte:02x}")).collect()
}