nexus-sdk 2.1.0

use {
    super::{
        driver::{MIN_REPLAY_RANGE_CHECKPOINTS, RECONNECT_DELAY},
        *,
    },
    crate::{sui, test_utils::sui_mocks},
    futures::StreamExt as _,
    std::sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    },
    sui_rpc::field::FieldMaskUtil as _,
    tokio::{
        sync::Barrier,
        time::{timeout, Duration},
    },
    tokio_util::sync::CancellationToken,
};

fn watermark(cursor: &[u8], checkpoint: Option<u64>) -> sui::grpc::Watermark {
    let mut watermark = sui::grpc::Watermark::default();
    watermark.set_cursor(cursor.to_vec());
    if let Some(checkpoint) = checkpoint {
        watermark.set_checkpoint(checkpoint);
    }
    watermark
}

#[tokio::test]
async fn partial_replay_is_bounded_and_restarts_the_incomplete_range() {
    let mut subscription = sui_mocks::grpc::MockSubscriptionService::new();
    subscription
        .expect_subscribe_events()
        .times(2)
        .returning(|_| {
            let first = subscription_frame(None, watermark(b"live", Some(511)));
            Ok(tonic::Response::new(Box::pin(
                futures::stream::iter([Ok(first)]).chain(futures::stream::pending()),
            )
                as sui_mocks::grpc::BoxEventStream))
        });
    let delivered = Arc::new(tokio::sync::Notify::new());
    let attempts = Arc::new(AtomicUsize::new(0));
    let mut ledger = sui_mocks::grpc::MockLedgerService::new();
    ledger.expect_list_events().times(2).returning({
        let delivered = Arc::clone(&delivered);
        let attempts = Arc::clone(&attempts);
        move |request| {
            let request = request.into_inner();
            // An emitted page at checkpoint 300 does not complete this range.
            assert_eq!(request.start_checkpoint, Some(0));
            assert_eq!(request.end_checkpoint, Some(512));
            assert!(request.options().after.is_none());
            let first_attempt = attempts.fetch_add(1, Ordering::SeqCst) == 0;
            let count = if first_attempt { 64 } else { 193 };
            let events = futures::stream::iter((0..count).map(|index: u32| {
                let mut frame = sui::grpc::ListEventsResponse::default();
                frame.set_event(
                    sui::grpc::Event::default()
                        .with_checkpoint(300)
                        .with_transaction_digest(sui::types::Digest::ZERO)
                        .with_event_index(index)
                        .with_contents(vec![index as u8; 4_096]),
                );
                frame.set_watermark(watermark(&index.to_be_bytes(), None));
                Ok(frame)
            }));
            let delivered = Arc::clone(&delivered);
            let terminal = futures::stream::once(async move {
                if first_attempt {
                    // The consumer must receive a page before this range ends.
                    delivered.notified().await;
                    Err(tonic::Status::unavailable("interrupted dense range"))
                } else {
                    Ok(list_frame(
                        None,
                        watermark(b"end", Some(511)),
                        sui::grpc::QueryEndReason::CheckpointBound,
                    ))
                }
            });
            Ok(tonic::Response::new(
                Box::pin(events.chain(terminal)) as sui_mocks::grpc::BoxListEventsStream
            ))
        }
    });
    let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        ledger_service_mock: Some(ledger),
        subscription_service_mock: Some(subscription),
        ..Default::default()
    });
    let query = RawEventQuery::new(
        sui::grpc::EventFilter::default(),
        sui::grpc::FieldMask::from_paths(["contents"]),
    );
    let mut pages = EventIngestor::new(&rpc_url, query)
        .with_replay_concurrency(std::num::NonZeroUsize::MIN)
        .start(Some(0))
        .unwrap();
    timeout(Duration::from_secs(5), async {
        let first = pages.recv().await.unwrap().unwrap();
        assert_eq!(first.events.len(), 64);
        assert_eq!(first.checkpoint, 300);
        delivered.notify_one();
        assert!(pages.recv().await.unwrap().is_err());
        let mut indices = Vec::new();
        loop {
            let page = pages.recv().await.unwrap().unwrap();
            assert!(page.events.len() <= 64);
            indices.extend(page.events.iter().map(|event| event.event_index()));
            if page.checkpoint == 511 {
                break;
            }
        }
        assert_eq!(indices, (0..193).collect::<Vec<_>>());
    })
    .await
    .expect("dense replay did not progress");
    assert_eq!(attempts.load(Ordering::SeqCst), 2);
}

fn subscription_frame(
    event: Option<sui::grpc::Event>,
    watermark: sui::grpc::Watermark,
) -> sui::grpc::SubscribeEventsResponse {
    let mut response = sui::grpc::SubscribeEventsResponse::default();
    if let Some(event) = event {
        response.set_event(event);
    }
    response.set_watermark(watermark);
    response
}

fn list_frame(
    event: Option<sui::grpc::Event>,
    watermark: sui::grpc::Watermark,
    end: sui::grpc::QueryEndReason,
) -> sui::grpc::ListEventsResponse {
    let mut response = sui::grpc::ListEventsResponse::default();
    if let Some(event) = event {
        response.set_event(event);
    }
    response.set_watermark(watermark);
    let mut query_end = sui::grpc::QueryEnd::default();
    query_end.set_reason(end);
    response.set_end(query_end);
    response
}

#[test]
fn zero_channel_capacity_is_a_configuration_error() {
    let query = RawEventQuery::new(
        sui::grpc::EventFilter::default(),
        sui::grpc::FieldMask::default(),
    );
    let result = EventIngestor::new("http://127.0.0.1:1", query)
        .with_channel_capacity(0)
        .start(None);

    match result {
        Err(EventIngestionError::Configuration(message)) => {
            assert!(message.contains("channel capacity"));
        }
        Err(other) => {
            panic!("expected configuration failure, found {other:?}");
        }
        Ok(_) => panic!("zero channel capacity was accepted"),
    }
}

#[test]
fn invalid_event_field_is_a_configuration_error() {
    let query = RawEventQuery::new(
        sui::grpc::EventFilter::default(),
        sui::grpc::FieldMask::from_paths(["not_an_event_field"]),
    );
    let result = EventIngestor::new("http://127.0.0.1:1", query).start(None);

    match result {
        Err(EventIngestionError::Configuration(message)) => {
            assert!(message.contains("not_an_event_field"));
        }
        Err(other) => {
            panic!("expected configuration failure, found {other:?}");
        }
        Ok(_) => panic!("invalid event field was accepted"),
    }
}

#[test]
fn invalid_replay_endpoint_is_a_configuration_error() {
    let query = RawEventQuery::new(
        sui::grpc::EventFilter::default(),
        sui::grpc::FieldMask::default(),
    );
    let result = EventIngestor::new("http://127.0.0.1:1", query)
        .with_replay_rpc_url("not a url")
        .start(Some(0));

    match result {
        Err(EventIngestionError::Configuration(message)) => {
            assert!(message.contains("invalid replay gRPC URL"));
        }
        Err(other) => {
            panic!("expected configuration failure, found {other:?}");
        }
        Ok(_) => panic!("invalid replay endpoint was accepted"),
    }
}

#[tokio::test]
async fn first_matching_checkpoint_uses_the_exact_filter_and_ascending_order() {
    let filter = sui::grpc::EventFilter::default();
    let expected_filter = filter.clone();
    let mut event = sui::grpc::Event::default();
    event.set_checkpoint(17);

    let mut ledger = sui_mocks::grpc::MockLedgerService::new();
    ledger
        .expect_list_events()
        .once()
        .returning(move |request| {
            let request = request.into_inner();
            assert_eq!(request.filter(), &expected_filter);
            assert_eq!(request.options().limit_opt(), Some(1));
            assert_eq!(
                request.options().ordering,
                Some(sui::grpc::Ordering::Ascending as i32)
            );
            assert_eq!(request.read_mask.as_ref().unwrap().paths, ["checkpoint"]);
            let frame = list_frame(
                Some(event.clone()),
                watermark(b"first", Some(17)),
                sui::grpc::QueryEndReason::ItemLimit,
            );
            Ok(tonic::Response::new(
                Box::pin(futures::stream::iter([Ok(frame)]))
                    as sui_mocks::grpc::BoxListEventsStream,
            ))
        });
    let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        ledger_service_mock: Some(ledger),
        ..Default::default()
    });
    let query = RawEventQuery::new(filter, sui::grpc::FieldMask::default());
    let ingestor = EventIngestor::new(rpc_url, query);

    assert_eq!(
        ingestor.first_matching_checkpoint().await.unwrap(),
        Some(17)
    );
}

#[tokio::test]
async fn first_matching_checkpoint_returns_none_for_an_empty_index() {
    let mut ledger = sui_mocks::grpc::MockLedgerService::new();
    ledger.expect_list_events().once().returning(|_request| {
        let frame = list_frame(
            None,
            watermark(b"end", Some(17)),
            sui::grpc::QueryEndReason::LedgerTip,
        );
        Ok(tonic::Response::new(
            Box::pin(futures::stream::iter([Ok(frame)])) as sui_mocks::grpc::BoxListEventsStream,
        ))
    });
    let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        ledger_service_mock: Some(ledger),
        ..Default::default()
    });
    let query = RawEventQuery::new(
        sui::grpc::EventFilter::default(),
        sui::grpc::FieldMask::default(),
    );
    let ingestor = EventIngestor::new(rpc_url, query);

    assert_eq!(ingestor.first_matching_checkpoint().await.unwrap(), None);
}

#[tokio::test]
async fn ingestor_adds_engine_fields_and_returns_raw_events() {
    let mut event = sui::grpc::Event::default();
    event.set_contents(vec![1, 2, 3]);
    event.set_checkpoint(12);
    event.set_transaction_digest(sui::types::Digest::ZERO);
    event.set_event_index(4);

    let expected_event = event.clone();
    let mut subscription = sui_mocks::grpc::MockSubscriptionService::new();
    subscription
        .expect_subscribe_events()
        .once()
        .returning(move |request| {
            let request = request.into_inner();
            let paths = &request.read_mask.as_ref().unwrap().paths;
            for path in [
                "contents",
                "checkpoint",
                "transaction_digest",
                "event_index",
            ] {
                assert!(paths.iter().any(|candidate| candidate == path));
            }

            let frames = [
                Ok(subscription_frame(None, watermark(b"live", None))),
                Ok(subscription_frame(
                    Some(event.clone()),
                    watermark(b"event", Some(11)),
                )),
            ];
            let stream = futures::stream::iter(frames).chain(futures::stream::pending());
            Ok(tonic::Response::new(
                Box::pin(stream) as sui_mocks::grpc::BoxEventStream
            ))
        });

    let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        subscription_service_mock: Some(subscription),
        ..Default::default()
    });
    let query = RawEventQuery::new(
        sui::grpc::EventFilter::default(),
        sui::grpc::FieldMask::from_paths(["contents"]),
    );
    let mut pages = EventIngestor::new(&rpc_url, query)
        .start(None)
        .expect("ingestor should start");

    let page = timeout(Duration::from_secs(2), pages.recv())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(page.checkpoint, 12);
    assert_eq!(page.events, [expected_event]);
    assert_eq!(page.source, EventPageSource::Live);
}

#[tokio::test]
async fn replay_uses_the_live_query_and_stops_at_its_cursor() {
    let mut event = sui::grpc::Event::default();
    event.set_contents(vec![4, 5, 6]);
    event.set_checkpoint(8);
    event.set_transaction_digest(sui::types::Digest::ZERO);
    event.set_event_index(2);
    let expected_event = event.clone();

    let filter = sui::grpc::EventFilter::default();
    let expected_filter = filter.clone();
    let mut subscription = sui_mocks::grpc::MockSubscriptionService::new();
    subscription
        .expect_subscribe_events()
        .once()
        .returning(move |_request| {
            let first = subscription_frame(None, watermark(b"live", None));
            let stream = futures::stream::iter([Ok(first)]).chain(futures::stream::pending());
            Ok(tonic::Response::new(
                Box::pin(stream) as sui_mocks::grpc::BoxEventStream
            ))
        });

    let mut ledger = sui_mocks::grpc::MockLedgerService::new();
    ledger
        .expect_list_events()
        .once()
        .returning(move |request| {
            let request = request.into_inner();
            assert_eq!(request.start_checkpoint, Some(7));
            assert_eq!(
                request.options().before.as_deref(),
                Some(b"live".as_slice())
            );
            assert!(request.end_checkpoint.is_none());
            assert_eq!(request.filter(), &expected_filter);
            let paths = &request.read_mask.as_ref().unwrap().paths;
            for path in [
                "contents",
                "checkpoint",
                "transaction_digest",
                "event_index",
            ] {
                assert!(paths.iter().any(|candidate| candidate == path));
            }

            let frame = list_frame(
                Some(event.clone()),
                watermark(b"live", Some(8)),
                sui::grpc::QueryEndReason::CursorBound,
            );
            Ok(tonic::Response::new(
                Box::pin(futures::stream::iter([Ok(frame)]))
                    as sui_mocks::grpc::BoxListEventsStream,
            ))
        });

    let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        ledger_service_mock: Some(ledger),
        subscription_service_mock: Some(subscription),
        ..Default::default()
    });
    let query = RawEventQuery::new(filter, sui::grpc::FieldMask::from_paths(["contents"]));
    let mut pages = EventIngestor::new(&rpc_url, query)
        .start(Some(7))
        .expect("ingestor should start");

    let page = timeout(Duration::from_secs(2), async {
        loop {
            let page = pages.recv().await.unwrap().unwrap();
            if !page.events.is_empty() {
                return page;
            }
        }
    })
    .await
    .unwrap();

    assert_eq!(page.checkpoint, 8);
    assert_eq!(page.events, [expected_event]);
    assert_eq!(page.source, EventPageSource::Replay);
}

#[tokio::test]
async fn replay_and_subscription_use_their_configured_endpoints() {
    let mut event = sui::grpc::Event::default();
    event.set_checkpoint(8);
    event.set_transaction_digest(sui::types::Digest::ZERO);
    event.set_event_index(2);

    let mut subscription = sui_mocks::grpc::MockSubscriptionService::new();
    subscription
        .expect_subscribe_events()
        .once()
        .returning(move |_request| {
            let first = subscription_frame(None, watermark(b"live", Some(8)));
            let stream = futures::stream::iter([Ok(first)]).chain(futures::stream::pending());
            Ok(tonic::Response::new(
                Box::pin(stream) as sui_mocks::grpc::BoxEventStream
            ))
        });
    let live_rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        subscription_service_mock: Some(subscription),
        ..Default::default()
    });

    let mut ledger = sui_mocks::grpc::MockLedgerService::new();
    ledger
        .expect_list_events()
        .once()
        .returning(move |request| {
            let request = request.into_inner();
            assert_eq!(request.start_checkpoint, Some(7));
            assert!(request.options().before.is_none());
            assert_eq!(request.end_checkpoint, Some(9));
            let frame = list_frame(
                Some(event.clone()),
                watermark(b"live", Some(8)),
                sui::grpc::QueryEndReason::CheckpointBound,
            );
            Ok(tonic::Response::new(
                Box::pin(futures::stream::iter([Ok(frame)]))
                    as sui_mocks::grpc::BoxListEventsStream,
            ))
        });
    let replay_rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        ledger_service_mock: Some(ledger),
        ..Default::default()
    });

    let query = RawEventQuery::new(
        sui::grpc::EventFilter::default(),
        sui::grpc::FieldMask::default(),
    );
    let mut pages = EventIngestor::new("not a url", query)
        .with_subscription_rpc_url(&live_rpc_url)
        .with_replay_rpc_url(replay_rpc_url)
        .start(Some(7))
        .expect("ingestor should start");

    let page = timeout(Duration::from_secs(2), async {
        loop {
            let page = pages.recv().await.unwrap().unwrap();
            if !page.events.is_empty() {
                return page;
            }
        }
    })
    .await
    .expect("archive replay did not emit the event");

    assert_eq!(page.checkpoint, 8);
    assert_eq!(page.source, EventPageSource::Replay);
}

#[tokio::test]
async fn checkpoint_ranges_fetch_concurrently_and_emit_in_order() {
    let replay_end = MIN_REPLAY_RANGE_CHECKPOINTS * 2;
    let mut subscription = sui_mocks::grpc::MockSubscriptionService::new();
    subscription
        .expect_subscribe_events()
        .once()
        .returning(move |_request| {
            let first =
                subscription_frame(None, watermark(b"live", Some(replay_end.saturating_sub(1))));
            let stream = futures::stream::iter([Ok(first)]).chain(futures::stream::pending());
            Ok(tonic::Response::new(
                Box::pin(stream) as sui_mocks::grpc::BoxEventStream
            ))
        });

    let active = Arc::new(AtomicUsize::new(0));
    let maximum_active = Arc::new(AtomicUsize::new(0));
    let concurrent_requests = Arc::new(Barrier::new(2));
    let mut ledger = sui_mocks::grpc::MockLedgerService::new();
    ledger.expect_list_events().times(2).returning({
        let active = Arc::clone(&active);
        let maximum_active = Arc::clone(&maximum_active);
        let concurrent_requests = Arc::clone(&concurrent_requests);
        move |request| {
            let request = request.into_inner();
            let start = request.start_checkpoint.unwrap();
            let end = request.end_checkpoint.unwrap();
            assert_eq!(end, start + MIN_REPLAY_RANGE_CHECKPOINTS);
            assert_eq!(
                request.options().ordering,
                Some(sui::grpc::Ordering::Ascending as i32)
            );

            let active = Arc::clone(&active);
            let maximum_active = Arc::clone(&maximum_active);
            let concurrent_requests = Arc::clone(&concurrent_requests);
            let stream = futures::stream::once(async move {
                let current = active.fetch_add(1, Ordering::SeqCst) + 1;
                maximum_active.fetch_max(current, Ordering::SeqCst);
                concurrent_requests.wait().await;
                let mut event = sui::grpc::Event::default();
                event.set_contents(vec![(start / MIN_REPLAY_RANGE_CHECKPOINTS) as u8]);
                event.set_checkpoint(start);
                event.set_transaction_digest(sui::types::Digest::ZERO);
                event.set_event_index(0);
                let frame = list_frame(
                    Some(event),
                    watermark(&start.to_be_bytes(), Some(end - 1)),
                    sui::grpc::QueryEndReason::CheckpointBound,
                );
                active.fetch_sub(1, Ordering::SeqCst);
                Ok(frame)
            });
            Ok(tonic::Response::new(
                Box::pin(stream) as sui_mocks::grpc::BoxListEventsStream
            ))
        }
    });

    let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        ledger_service_mock: Some(ledger),
        subscription_service_mock: Some(subscription),
        ..Default::default()
    });
    let query = RawEventQuery::new(
        sui::grpc::EventFilter::default(),
        sui::grpc::FieldMask::from_paths(["contents"]),
    );
    let mut pages = EventIngestor::new(&rpc_url, query)
        .with_replay_concurrency(std::num::NonZeroUsize::new(2).unwrap())
        .start(Some(0))
        .expect("ingestor should start");

    let pages = timeout(Duration::from_secs(2), async {
        let mut replay = Vec::new();
        while replay.len() != 2 {
            let page = pages.recv().await.unwrap().unwrap();
            if !page.events.is_empty() {
                replay.push(page);
            }
        }
        replay
    })
    .await
    .expect("parallel replay did not finish");

    assert_eq!(maximum_active.load(Ordering::SeqCst), 2);
    assert_eq!(
        pages.iter().map(|page| page.checkpoint).collect::<Vec<_>>(),
        [MIN_REPLAY_RANGE_CHECKPOINTS - 1, replay_end - 1]
    );
    assert_eq!(pages[0].events[0].contents().value(), &[0]);
    assert_eq!(pages[1].events[0].contents().value(), &[1]);
}

#[tokio::test]
async fn replay_continues_after_an_item_limit() {
    let mut event = sui::grpc::Event::default();
    event.set_checkpoint(8);
    event.set_transaction_digest(sui::types::Digest::ZERO);
    event.set_event_index(2);

    let mut subscription = sui_mocks::grpc::MockSubscriptionService::new();
    subscription
        .expect_subscribe_events()
        .once()
        .returning(move |_request| {
            let first = subscription_frame(None, watermark(b"live", None));
            let stream = futures::stream::iter([Ok(first)]).chain(futures::stream::pending());
            Ok(tonic::Response::new(
                Box::pin(stream) as sui_mocks::grpc::BoxEventStream
            ))
        });

    let calls = Arc::new(AtomicUsize::new(0));
    let mut ledger = sui_mocks::grpc::MockLedgerService::new();
    ledger.expect_list_events().times(2).returning({
        let calls = Arc::clone(&calls);
        move |request| {
            let call = calls.fetch_add(1, Ordering::SeqCst);
            let request = request.into_inner();
            assert_eq!(request.start_checkpoint, Some(7));
            assert_eq!(
                request.options().before.as_deref(),
                Some(b"live".as_slice())
            );

            let frame = if call == 0 {
                assert!(request.options().after.is_none());
                list_frame(
                    Some(event.clone()),
                    watermark(b"page-one", Some(7)),
                    sui::grpc::QueryEndReason::ItemLimit,
                )
            } else {
                assert_eq!(
                    request.options().after.as_deref(),
                    Some(b"page-one".as_slice())
                );
                list_frame(
                    None,
                    watermark(b"live", Some(8)),
                    sui::grpc::QueryEndReason::CursorBound,
                )
            };
            Ok(tonic::Response::new(
                Box::pin(futures::stream::iter([Ok(frame)]))
                    as sui_mocks::grpc::BoxListEventsStream,
            ))
        }
    });

    let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        ledger_service_mock: Some(ledger),
        subscription_service_mock: Some(subscription),
        ..Default::default()
    });
    let query = RawEventQuery::new(
        sui::grpc::EventFilter::default(),
        sui::grpc::FieldMask::default(),
    );
    let mut pages = EventIngestor::new(&rpc_url, query)
        .start(Some(7))
        .expect("ingestor should start");

    timeout(Duration::from_secs(2), async {
        loop {
            let page = pages.recv().await.unwrap().unwrap();
            if !page.events.is_empty() {
                return;
            }
        }
    })
    .await
    .expect("replay did not emit the event");

    timeout(Duration::from_secs(2), async {
        while calls.load(Ordering::SeqCst) != 2 {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("replay did not continue after the item limit");
}

#[tokio::test]
async fn replay_gap_is_terminal_by_default() {
    let mut subscription = sui_mocks::grpc::MockSubscriptionService::new();
    subscription
        .expect_subscribe_events()
        .once()
        .returning(move |_request| {
            let first = subscription_frame(None, watermark(b"live", None));
            let stream = futures::stream::iter([Ok(first)]).chain(futures::stream::pending());
            Ok(tonic::Response::new(
                Box::pin(stream) as sui_mocks::grpc::BoxEventStream
            ))
        });

    let mut ledger = sui_mocks::grpc::MockLedgerService::new();
    ledger
        .expect_list_events()
        .once()
        .returning(move |_request| {
            Ok(tonic::Response::new(Box::pin(futures::stream::iter([Err(
                tonic::Status::out_of_range("checkpoint was pruned"),
            )]))
                as sui_mocks::grpc::BoxListEventsStream))
        });

    let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        ledger_service_mock: Some(ledger),
        subscription_service_mock: Some(subscription),
        ..Default::default()
    });
    let query = RawEventQuery::new(
        sui::grpc::EventFilter::default(),
        sui::grpc::FieldMask::default(),
    );
    let mut pages = EventIngestor::new(&rpc_url, query)
        .start(Some(7))
        .expect("ingestor should start");

    let error = timeout(Duration::from_secs(2), pages.recv())
        .await
        .expect("ingestor did not report unavailable replay")
        .expect("ingestor closed without reporting unavailable replay")
        .expect_err("unavailable replay unexpectedly succeeded");
    assert!(matches!(
        error,
        EventIngestionError::ReplayGap {
            start_checkpoint: 7,
            ..
        }
    ));
    assert!(pages.recv().await.is_none());
}

#[tokio::test]
async fn replay_gap_recovery_reports_the_gap_and_resumes_live() {
    let mut subscription = sui_mocks::grpc::MockSubscriptionService::new();
    subscription
        .expect_subscribe_events()
        .once()
        .returning(move |_request| {
            let frames = [
                Ok(subscription_frame(None, watermark(b"live", None))),
                Ok(subscription_frame(None, watermark(b"progress", Some(20)))),
            ];
            let stream = futures::stream::iter(frames).chain(futures::stream::pending());
            Ok(tonic::Response::new(
                Box::pin(stream) as sui_mocks::grpc::BoxEventStream
            ))
        });

    let mut ledger = sui_mocks::grpc::MockLedgerService::new();
    ledger
        .expect_list_events()
        .once()
        .returning(move |_request| {
            Ok(tonic::Response::new(Box::pin(futures::stream::iter([Err(
                tonic::Status::out_of_range("checkpoint was pruned"),
            )]))
                as sui_mocks::grpc::BoxListEventsStream))
        });

    let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        ledger_service_mock: Some(ledger),
        subscription_service_mock: Some(subscription),
        ..Default::default()
    });
    let query = RawEventQuery::new(
        sui::grpc::EventFilter::default(),
        sui::grpc::FieldMask::default(),
    );
    let mut pages = EventIngestor::new(&rpc_url, query)
        .with_replay_gap_recovery()
        .start(Some(7))
        .expect("ingestor should start");

    let error = timeout(Duration::from_secs(2), pages.recv())
        .await
        .expect("ingestor did not report unavailable replay")
        .expect("ingestor closed without reporting unavailable replay")
        .expect_err("unavailable replay unexpectedly succeeded");
    assert!(matches!(
        error,
        EventIngestionError::ReplayGap {
            start_checkpoint: 7,
            ..
        }
    ));

    let page = timeout(Duration::from_secs(2), pages.recv())
        .await
        .expect("ingestor did not resume at the live cursor")
        .expect("ingestor closed after unavailable replay")
        .expect("live ingestion failed after unavailable replay");
    assert_eq!(page.checkpoint, 20);
    assert!(page.events.is_empty());
    assert_eq!(page.source, EventPageSource::Live);
}

#[tokio::test]
async fn reconnect_replays_from_the_last_inclusive_checkpoint() {
    let mut first_event = sui::grpc::Event::default();
    first_event.set_checkpoint(10);
    first_event.set_transaction_digest(sui::types::Digest::ZERO);
    first_event.set_event_index(1);

    let mut replayed_event = sui::grpc::Event::default();
    replayed_event.set_checkpoint(11);
    replayed_event.set_transaction_digest(sui::types::Digest::ZERO);
    replayed_event.set_event_index(2);

    let subscription_calls = Arc::new(AtomicUsize::new(0));
    let mut subscription = sui_mocks::grpc::MockSubscriptionService::new();
    subscription.expect_subscribe_events().times(2).returning({
        let subscription_calls = Arc::clone(&subscription_calls);
        move |_request| {
            let call = subscription_calls.fetch_add(1, Ordering::SeqCst);
            let stream: sui_mocks::grpc::BoxEventStream = match call {
                0 => Box::pin(futures::stream::iter([
                    Ok(subscription_frame(None, watermark(b"live-one", None))),
                    Ok(subscription_frame(
                        Some(first_event.clone()),
                        watermark(b"event-ten", Some(9)),
                    )),
                    Err(tonic::Status::unavailable("stream interrupted")),
                ])),
                1 => Box::pin(
                    futures::stream::iter([Ok(subscription_frame(
                        None,
                        watermark(b"live-two", None),
                    ))])
                    .chain(futures::stream::pending()),
                ),
                _ => panic!("unexpected subscription call"),
            };
            Ok(tonic::Response::new(stream))
        }
    });

    let mut ledger = sui_mocks::grpc::MockLedgerService::new();
    ledger
        .expect_list_events()
        .once()
        .returning(move |request| {
            let request = request.into_inner();
            assert_eq!(request.start_checkpoint, Some(10));
            assert_eq!(
                request.options().before.as_deref(),
                Some(b"live-two".as_slice())
            );

            let frame = list_frame(
                Some(replayed_event.clone()),
                watermark(b"live-two", Some(11)),
                sui::grpc::QueryEndReason::CursorBound,
            );
            Ok(tonic::Response::new(
                Box::pin(futures::stream::iter([Ok(frame)]))
                    as sui_mocks::grpc::BoxListEventsStream,
            ))
        });

    let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        ledger_service_mock: Some(ledger),
        subscription_service_mock: Some(subscription),
        ..Default::default()
    });
    let query = RawEventQuery::new(
        sui::grpc::EventFilter::default(),
        sui::grpc::FieldMask::default(),
    );
    let mut pages = EventIngestor::new(&rpc_url, query)
        .start(None)
        .expect("ingestor should start");
    let mut checkpoints = Vec::new();

    timeout(Duration::from_secs(2), async {
        while checkpoints.len() != 2 {
            match pages.recv().await.unwrap() {
                Ok(page) if !page.events.is_empty() => {
                    checkpoints.push(page.checkpoint);
                }
                Ok(_) | Err(_) => {}
            }
        }
    })
    .await
    .expect("ingestor did not replay after reconnecting");

    assert_eq!(checkpoints, [10, 11]);
    assert_eq!(subscription_calls.load(Ordering::SeqCst), 2);
}

struct FailingQuery {
    decode_calls: Arc<AtomicUsize>,
}

impl EventQuery for FailingQuery {
    type Error = std::io::Error;
    type Output = ();

    fn filter(&self) -> sui::grpc::EventFilter {
        sui::grpc::EventFilter::default()
    }

    fn read_mask(&self) -> sui::grpc::FieldMask {
        sui::grpc::FieldMask::default()
    }

    async fn decode(&self, _event: sui::grpc::Event) -> Result<Option<Self::Output>, Self::Error> {
        self.decode_calls.fetch_add(1, Ordering::SeqCst);
        Err(std::io::Error::other("unsupported event shape"))
    }
}

#[tokio::test]
async fn decode_failure_is_terminal_and_preserves_event_identity() {
    let mut event = sui::grpc::Event::default();
    event.set_checkpoint(10);
    event.set_transaction_digest(sui::types::Digest::ZERO);
    event.set_event_index(3);

    let subscription_calls = Arc::new(AtomicUsize::new(0));
    let mut subscription = sui_mocks::grpc::MockSubscriptionService::new();
    subscription.expect_subscribe_events().once().returning({
        let subscription_calls = Arc::clone(&subscription_calls);
        move |_request| {
            subscription_calls.fetch_add(1, Ordering::SeqCst);
            let frames = [
                Ok(subscription_frame(None, watermark(b"live", None))),
                Ok(subscription_frame(
                    Some(event.clone()),
                    watermark(b"event", Some(10)),
                )),
            ];
            Ok(tonic::Response::new(
                Box::pin(futures::stream::iter(frames)) as sui_mocks::grpc::BoxEventStream,
            ))
        }
    });

    let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        subscription_service_mock: Some(subscription),
        ..Default::default()
    });
    let decode_calls = Arc::new(AtomicUsize::new(0));
    let query = FailingQuery {
        decode_calls: Arc::clone(&decode_calls),
    };
    let mut pages = EventIngestor::new(&rpc_url, query)
        .start(None)
        .expect("ingestor should start");

    let error = pages.recv().await.unwrap().unwrap_err();
    match error {
        EventIngestionError::Decode {
            checkpoint,
            transaction_digest,
            event_index,
            source,
        } => {
            assert_eq!(checkpoint, 10);
            assert_eq!(transaction_digest, sui::types::Digest::ZERO.to_string());
            assert_eq!(event_index, 3);
            assert_eq!(source.to_string(), "unsupported event shape");
        }
        other => panic!("expected decode failure, found {other:?}"),
    }

    assert!(pages.recv().await.is_none());
    assert_eq!(decode_calls.load(Ordering::SeqCst), 1);
    assert_eq!(subscription_calls.load(Ordering::SeqCst), 1);
}

#[tokio::test]
async fn permanent_rpc_failure_is_terminal() {
    let subscription_calls = Arc::new(AtomicUsize::new(0));
    let mut subscription = sui_mocks::grpc::MockSubscriptionService::new();
    subscription.expect_subscribe_events().returning({
        let subscription_calls = Arc::clone(&subscription_calls);
        move |_request| {
            subscription_calls.fetch_add(1, Ordering::SeqCst);
            Err(tonic::Status::invalid_argument("query was rejected"))
        }
    });

    let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        subscription_service_mock: Some(subscription),
        ..Default::default()
    });
    let query = RawEventQuery::new(
        sui::grpc::EventFilter::default(),
        sui::grpc::FieldMask::default(),
    );
    let mut pages = EventIngestor::new(&rpc_url, query)
        .start(None)
        .expect("ingestor should start");

    let error = pages.recv().await.unwrap().unwrap_err();
    assert!(error.to_string().contains("query was rejected"));
    assert!(pages.recv().await.is_none());
    assert_eq!(subscription_calls.load(Ordering::SeqCst), 1);
}

#[tokio::test]
async fn cancellation_stops_blocked_error_delivery() {
    let subscription_calls = Arc::new(AtomicUsize::new(0));
    let mut subscription = sui_mocks::grpc::MockSubscriptionService::new();
    subscription.expect_subscribe_events().returning({
        let subscription_calls = Arc::clone(&subscription_calls);
        move |_request| {
            subscription_calls.fetch_add(1, Ordering::SeqCst);
            Err(tonic::Status::unavailable("stream interrupted"))
        }
    });

    let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        subscription_service_mock: Some(subscription),
        ..Default::default()
    });
    let cancellation_token = CancellationToken::new();
    let query = RawEventQuery::new(
        sui::grpc::EventFilter::default(),
        sui::grpc::FieldMask::default(),
    );
    let mut pages = EventIngestor::new(&rpc_url, query)
        .with_channel_capacity(1)
        .with_cancellation_token(cancellation_token.clone())
        .start(None)
        .expect("ingestor should start");

    timeout(Duration::from_secs(2), async {
        while subscription_calls.load(Ordering::SeqCst) < 2 {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("ingestor did not attempt to reconnect");
    tokio::time::sleep(RECONNECT_DELAY * 2).await;
    cancellation_token.cancel();
    timeout(Duration::from_secs(2), async {
        while !pages.is_closed() {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("cancellation did not stop error delivery");

    assert!(pages.recv().await.unwrap().is_err());
    assert!(pages.recv().await.is_none());
}

#[tokio::test]
async fn dropping_the_receiver_stops_reconnection() {
    let subscription_calls = Arc::new(AtomicUsize::new(0));
    let mut subscription = sui_mocks::grpc::MockSubscriptionService::new();
    subscription.expect_subscribe_events().returning({
        let subscription_calls = Arc::clone(&subscription_calls);
        move |_request| {
            subscription_calls.fetch_add(1, Ordering::SeqCst);
            Ok(tonic::Response::new(Box::pin(futures::stream::iter([Err(
                tonic::Status::unavailable("stream interrupted"),
            )]))
                as sui_mocks::grpc::BoxEventStream))
        }
    });

    let rpc_url = sui_mocks::grpc::mock_server(sui_mocks::grpc::ServerMocks {
        subscription_service_mock: Some(subscription),
        ..Default::default()
    });
    let query = RawEventQuery::new(
        sui::grpc::EventFilter::default(),
        sui::grpc::FieldMask::default(),
    );
    let mut pages = EventIngestor::new(&rpc_url, query)
        .start(None)
        .expect("ingestor should start");

    assert!(pages.recv().await.unwrap().is_err());
    drop(pages);
    tokio::time::sleep(RECONNECT_DELAY * 4).await;

    assert_eq!(subscription_calls.load(Ordering::SeqCst), 1);
}