skilj-core 0.2.0

//! Tests for the `EventFetch` surface (`specs/skilj.allium`) - entity
//! `ReadCursor`, rules `FetchEvents`/`ConsumeEvents`/`AcknowledgeEvents`,
//! invariant `UniqueReadCursorPerToken`. Generated by `/allium:propagate`
//! as the pilot surface named in [docs/architecture.md §9](../../docs/architecture.md#next-steps); see that
//! section and the commit introducing this file for why EventFetch was
//! picked first.
//!
//! Obligations covered here (from `allium plan specs/skilj.allium`,
//! filtered to this surface's five source constructs): 21 total.
//! `surface-actor.EventFetch`/`surface-exposure.EventFetch`/
//! `surface-provides.EventFetch` (3) were left unbookkept here, pending
//! REST scaffolding that didn't exist yet at the time - see the doc
//! comment at the bottom of this file for why that reasoning is now
//! stale, and where these three are actually exercised today.

use chrono::{TimeZone, Utc};
use proptest::prelude::*;
use skilj_core::access_control::{self, EventReadStartPosition, EventReadToken, TokenStatus};
use skilj_core::error::SkiljRejection;
use skilj_core::event_store::{
    self, AckMode, BoundedContext, BoundedContextStatus, ConsumeEventsResult, CursorUpdate, Event,
    EventOrigin, EventType, ReadCursor,
};
use skilj_core::shared::{Filter, FilterOperator};

// ---------------------------------------------------------------------
// Fixtures
//
// `EventType`/`Event` grew fields beyond what EventFetch itself touches
// when ExternalEventIngestion/DirectEventCreation/CommandTrigger were
// propagated (see event_store's module doc) - these fixtures fill the
// rest with values this surface's rules never read (`event_read_allowed`
// is the only field FetchEvents/ConsumeEvents/AcknowledgeEvents care
// about).
// ---------------------------------------------------------------------

fn bounded_context(name: &str) -> BoundedContext {
    BoundedContext {
        name: name.into(),
        status: BoundedContextStatus::Active,
        created_at: timestamp(0),
        created_by: skilj_core::bootstrap::ContextCreator::SystemCreator,
        template: None,
    }
}

fn event_type(event_read_allowed: bool) -> EventType {
    EventType {
        bounded_context: bounded_context("orders"),
        name: "OrderPlaced".into(),
        schema: "{}".into(),
        schema_version: 1,
        tag_mappings: Vec::new(),
        owner_tag_key: None,
        sensitive_fields: Vec::new(),
        private_fields: Vec::new(),
        external_creation_allowed: false,
        direct_creation_allowed: false,
        system_triggered_allowed: false,
        system_triggered_schedule: None,
        missed_occurrence_policy: None,
        schedule_position: None,
        last_fired_at: None,
        event_read_allowed,
    }
}

fn other_event_type() -> EventType {
    EventType {
        name: "OrderCancelled".into(),
        ..event_type(true)
    }
}

fn token(status: TokenStatus, event_type: EventType) -> EventReadToken {
    EventReadToken {
        id: "reader-1".into(),
        secret: "s3cr3t".into(),
        status,
        created_at: timestamp(0),
        revoked_at: None,
        event_type,
        scope: None,
        start_from: EventReadStartPosition::Beginning,
        start_at_sequence: None,
        start_at_time: None,
    }
}

fn event(event_type: EventType, sequence: i64) -> Event {
    Event {
        bounded_context: bounded_context("orders"),
        event_type,
        payload: "{}".into(),
        metadata: skilj_core::shared::Metadata {
            r#type: "OrderPlaced".into(),
            version: 1,
            client_id: "test-client".into(),
            created_at: timestamp(0),
            correlation_id: None,
            causation_id: None,
        },
        sequence,
        tags: Vec::new(),
        encryption_keys: Vec::new(),
        origin: EventOrigin::DirectlyCreated,
    }
}

fn timestamp(secs: i64) -> chrono::DateTime<Utc> {
    Utc.timestamp_opt(secs, 0).unwrap()
}

/// `consume_events`'s own `checkout_lease` parameter (issue #25,
/// docs/architecture.md §53) - matches both `config.read_cursor_checkout_lease`'s
/// own spec default and `SkiljBuilder::read_cursor_checkout_lease`'s own
/// Rust default, so a test relying on this fixture is exercising the
/// same real-world value a default deployment would.
fn checkout_lease() -> chrono::Duration {
    chrono::Duration::minutes(5)
}

/// `event()` above with a caller-chosen `metadata.created_at`, for
/// `at_time` fixtures that need to distinguish events by when they
/// happened rather than only by their own sequence.
fn event_at(event_type: EventType, sequence: i64, created_at: chrono::DateTime<Utc>) -> Event {
    Event {
        metadata: skilj_core::shared::Metadata {
            created_at,
            ..event(event_type.clone(), sequence).metadata
        },
        ..event(event_type, sequence)
    }
}

// ---------------------------------------------------------------------
// entity-fields.ReadCursor
// ---------------------------------------------------------------------

#[test]
fn read_cursor_carries_its_declared_fields() {
    let t = token(TokenStatus::Active, event_type(true));
    let cursor = ReadCursor {
        token: t.clone(),
        ack_mode: AckMode::ManualAck,
        sequence: 41,
        updated_at: timestamp(1000),
        checked_out_at: Some(timestamp(1000)),
    };

    assert_eq!(cursor.token, t);
    assert_eq!(cursor.ack_mode, AckMode::ManualAck);
    assert_eq!(cursor.sequence, 41);
    assert_eq!(cursor.updated_at, timestamp(1000));
    assert_eq!(cursor.checked_out_at, Some(timestamp(1000)));
}

// ---------------------------------------------------------------------
// rule-success.FetchEvents / rule-failure.FetchEvents.{1,2,3,4}
// ---------------------------------------------------------------------

#[test]
fn fetch_events_succeeds_and_returns_only_matching_later_events() {
    let et = event_type(true);
    let t = token(TokenStatus::Active, et.clone());
    let events = vec![
        event(et.clone(), 1),
        event(et.clone(), 2),
        event(other_event_type(), 3), // different EventType - excluded
        event(et.clone(), 4),
    ];

    let result = event_store::fetch_events(&t, &events, &[], Some(1), None).unwrap();

    assert_eq!(
        result.iter().map(|e| e.sequence).collect::<Vec<_>>(),
        vec![2, 4]
    );
}

/// docs/architecture.md §179: an event names its type and bounded context
/// by identity. The copies it carries are snapshots - from the event
/// cache, taken before the context was archived or the type's scheduled
/// fire moved `last_fired_at` - and the token holds today's. Compared
/// whole, they used to say every such event belonged to another type.
#[test]
fn fetch_events_matches_an_event_whose_type_snapshot_is_stale() {
    let then = event_type(true);
    let mut now = then.clone();
    now.bounded_context.status = BoundedContextStatus::Archived;
    now.last_fired_at = Some(timestamp(60));
    now.schema_version += 1;
    let t = token(TokenStatus::Active, now.clone());
    let events = vec![event(then, 0), event(now, 1)];

    let result = event_store::fetch_events(&t, &events, &[], None, None).unwrap();

    assert_eq!(
        result.iter().map(|e| e.sequence).collect::<Vec<_>>(),
        vec![0, 1]
    );
}

#[test]
fn fetch_events_with_no_after_sequence_starts_from_the_beginning() {
    let et = event_type(true);
    let t = token(TokenStatus::Active, et.clone());
    let events = vec![event(et.clone(), 0), event(et.clone(), 1)];

    let result = event_store::fetch_events(&t, &events, &[], None, None).unwrap();

    assert_eq!(
        result.iter().map(|e| e.sequence).collect::<Vec<_>>(),
        vec![0, 1]
    );
}

/// rule-failure.FetchEvents.1 - `requires: token.status = active`.
#[test]
fn fetch_events_rejects_a_revoked_token() {
    let t = token(TokenStatus::Revoked, event_type(true));

    let err = event_store::fetch_events(&t, &[], &[], None, None).unwrap_err();

    assert_eq!(err.code(), access_control::Error::TokenNotActive.code());
}

/// rule-failure.FetchEvents.2 - `requires: token.event_type = read_type`.
/// Structurally unreachable rather than runtime-tested: `fetch_events`
/// doesn't accept `read_type` as a separate parameter, it derives it from
/// `token.event_type` (see the doc comment on `fetch_events`), so there is
/// no call shape that can violate this - it's true by construction, the
/// same way a type system rules out an illegal state rather than
/// rejecting it at runtime.
#[test]
fn fetch_events_read_type_is_always_the_tokens_event_type_by_construction() {
    let et = event_type(true);
    let t = token(TokenStatus::Active, et.clone());
    // No `read_type` argument exists to disagree with `t.event_type` -
    // this test exists to document the obligation, not exercise a branch.
    let _ = event_store::fetch_events(&t, &[], &[], None, None);
    assert_eq!(t.event_type, et);
}

/// rule-failure.FetchEvents.3 - `requires: read_type.event_read_allowed = true`.
#[test]
fn fetch_events_rejects_an_event_type_not_opted_into_reads() {
    let t = token(TokenStatus::Active, event_type(false));

    let err = event_store::fetch_events(&t, &[], &[], None, None).unwrap_err();

    assert_eq!(err.code(), event_store::Error::EventReadNotAllowed.code());
}

/// rule-failure.FetchEvents.4 - `requires: valid_filters(read_type, filters)`.
#[test]
fn fetch_events_rejects_an_invalid_filter() {
    let t = token(TokenStatus::Active, event_type(true));
    // event_type()'s schema is "{}" - "amount" names no field it declares.
    let bogus_filter = vec![Filter {
        field: "amount".into(),
        operator: FilterOperator::GreaterThan,
        value: "10".into(),
    }];

    let err = event_store::fetch_events(&t, &[], &bogus_filter, None, None).unwrap_err();

    assert_eq!(err.code(), event_store::Error::InvalidFilter.code());
}

// ---------------------------------------------------------------------
// rule-success.ConsumeEvents / rule-failure.ConsumeEvents.{1..5}
// ---------------------------------------------------------------------

#[test]
fn consume_events_provisions_a_cursor_on_first_use() {
    let et = event_type(true);
    let t = token(TokenStatus::Active, et.clone());
    let events = vec![event(et.clone(), 0), event(et.clone(), 1)];

    let ConsumeEventsResult {
        served,
        cursor_update,
    } = event_store::consume_events(
        &t,
        None,
        Some(AckMode::AutoAdvance),
        &events,
        &[],
        timestamp(100),
        checkout_lease(),
    )
    .unwrap();

    assert_eq!(
        served.iter().map(|e| e.sequence).collect::<Vec<_>>(),
        vec![0, 1]
    );
    match cursor_update {
        CursorUpdate::Created(cursor) => {
            assert_eq!(cursor.token, t);
            assert_eq!(cursor.ack_mode, AckMode::AutoAdvance);
            assert_eq!(cursor.sequence, 1); // highest served
            assert_eq!(cursor.updated_at, timestamp(100));
            // issue #25 (docs/architecture.md §53) -
            // auto_advance never claims: it has no equivalent gap to
            // close (see `ReadCursor.checked_out_at`'s own spec doc
            // comment).
            assert_eq!(cursor.checked_out_at, None);
        }
        other => panic!("expected ReadCursor.created, got {other:?}"),
    }
}

/// `at_sequence`'s whole point over `latest`: the token's first call can
/// replay *some* history from a caller-chosen cutoff, not just none at
/// all - here sequence 0 stays unserved (at or before the cutoff) while
/// 1 and 2, already historical relative to this call, are served
/// alongside nothing-new-yet.
#[test]
fn consume_events_at_sequence_starts_strictly_after_the_given_sequence() {
    let et = event_type(true);
    let t = EventReadToken {
        start_from: EventReadStartPosition::AtSequence,
        start_at_sequence: Some(0),
        ..token(TokenStatus::Active, et.clone())
    };
    let events = vec![event(et.clone(), 0), event(et.clone(), 1), event(et, 2)];

    let ConsumeEventsResult {
        served,
        cursor_update,
    } = event_store::consume_events(
        &t,
        None,
        Some(AckMode::AutoAdvance),
        &events,
        &[],
        timestamp(100),
        checkout_lease(),
    )
    .unwrap();

    assert_eq!(
        served.iter().map(|e| e.sequence).collect::<Vec<_>>(),
        vec![1, 2]
    );
    match cursor_update {
        CursorUpdate::Created(cursor) => assert_eq!(cursor.sequence, 2),
        other => panic!("expected ReadCursor.created, got {other:?}"),
    }
}

/// `at_time`'s own version of the same proof: the cutoff sits strictly
/// between two events' own `created_at`, so the earlier one is excluded
/// and the later one - already historical relative to this call - is
/// served, exactly the "replay from a chosen point in time" behaviour
/// `at_sequence` gives by number instead.
#[test]
fn consume_events_at_time_starts_strictly_after_events_at_or_before_that_cutoff() {
    let et = event_type(true);
    let cutoff = timestamp(50);
    let t = EventReadToken {
        start_from: EventReadStartPosition::AtTime,
        start_at_time: Some(cutoff),
        ..token(TokenStatus::Active, et.clone())
    };
    let events = vec![
        event_at(et.clone(), 0, timestamp(10)), // before the cutoff - excluded
        event_at(et.clone(), 1, timestamp(50)), // exactly at the cutoff - excluded
        event_at(et.clone(), 2, timestamp(90)), // after the cutoff - served
    ];

    let ConsumeEventsResult { served, .. } = event_store::consume_events(
        &t,
        None,
        Some(AckMode::AutoAdvance),
        &events,
        &[],
        timestamp(100),
        checkout_lease(),
    )
    .unwrap();

    assert_eq!(
        served.iter().map(|e| e.sequence).collect::<Vec<_>>(),
        vec![2]
    );
}

/// A `start_at_time` with nothing committed at or before it behaves
/// exactly as `beginning` would - the spec's own "empty means the token
/// gets the whole stream" note on `at_time_position`.
#[test]
fn consume_events_at_time_with_no_events_before_the_cutoff_serves_everything() {
    let et = event_type(true);
    let t = EventReadToken {
        start_from: EventReadStartPosition::AtTime,
        start_at_time: Some(timestamp(5)),
        ..token(TokenStatus::Active, et.clone())
    };
    let events = vec![
        event_at(et.clone(), 0, timestamp(50)),
        event_at(et, 1, timestamp(60)),
    ];

    let ConsumeEventsResult { served, .. } = event_store::consume_events(
        &t,
        None,
        Some(AckMode::AutoAdvance),
        &events,
        &[],
        timestamp(100),
        checkout_lease(),
    )
    .unwrap();

    assert_eq!(
        served.iter().map(|e| e.sequence).collect::<Vec<_>>(),
        vec![0, 1]
    );
}

#[test]
fn consume_events_auto_advance_moves_the_cursor_to_the_highest_served_sequence() {
    let et = event_type(true);
    let t = token(TokenStatus::Active, et.clone());
    let existing = ReadCursor {
        token: t.clone(),
        ack_mode: AckMode::AutoAdvance,
        sequence: 5,
        updated_at: timestamp(0),
        checked_out_at: None,
    };
    let events = vec![event(et.clone(), 6), event(et.clone(), 7)];

    let ConsumeEventsResult {
        served,
        cursor_update,
    } = event_store::consume_events(
        &t,
        Some(&existing),
        None,
        &events,
        &[],
        timestamp(200),
        checkout_lease(),
    )
    .unwrap();

    assert_eq!(
        served.iter().map(|e| e.sequence).collect::<Vec<_>>(),
        vec![6, 7]
    );
    assert_eq!(
        cursor_update,
        CursorUpdate::Advanced {
            sequence: 7,
            updated_at: timestamp(200)
        }
    );
}

/// `sequence`/`position` never moves on serve, in either mode - but
/// unlike before issue #25's investigation (docs/architecture.md
/// §53), a `manual_ack` cursor with no live claim now *does* get one:
/// `Claimed`, not `Unchanged`. See
/// `consume_events_manual_ack_with_a_live_claim_serves_nothing` below for
/// the case this test used to conflate with "nothing happens" - a second
/// call while a claim is still live really is `Unchanged`.
#[test]
fn consume_events_manual_ack_never_moves_the_cursor_on_serve() {
    let et = event_type(true);
    let t = token(TokenStatus::Active, et.clone());
    let existing = ReadCursor {
        token: t.clone(),
        ack_mode: AckMode::ManualAck,
        sequence: 5,
        updated_at: timestamp(0),
        checked_out_at: None,
    };
    let events = vec![event(et.clone(), 6)];

    let ConsumeEventsResult {
        served,
        cursor_update,
    } = event_store::consume_events(
        &t,
        Some(&existing),
        None,
        &events,
        &[],
        timestamp(200),
        checkout_lease(),
    )
    .unwrap();

    assert_eq!(
        served.iter().map(|e| e.sequence).collect::<Vec<_>>(),
        vec![6]
    );
    assert_eq!(
        cursor_update,
        CursorUpdate::Claimed {
            checked_out_at: timestamp(200)
        }
    );
}

/// issue #25's investigation (docs/architecture.md §53) - the
/// gap this closes: a second caller racing a still-live claim on the
/// same token must be served nothing, not the identical unacknowledged
/// batch a first caller may still be acting on externally.
#[test]
fn consume_events_manual_ack_with_a_live_claim_serves_nothing() {
    let et = event_type(true);
    let t = token(TokenStatus::Active, et.clone());
    let existing = ReadCursor {
        token: t.clone(),
        ack_mode: AckMode::ManualAck,
        sequence: 5,
        updated_at: timestamp(0),
        // Claimed one minute ago - well inside the five-minute
        // checkout_lease() fixture, so still live.
        checked_out_at: Some(timestamp(150)),
    };
    let events = vec![event(et, 6)];

    let ConsumeEventsResult {
        served,
        cursor_update,
    } = event_store::consume_events(
        &t,
        Some(&existing),
        None,
        &events,
        &[],
        timestamp(200),
        checkout_lease(),
    )
    .unwrap();

    assert!(
        served.is_empty(),
        "a second caller must not be served the same unacknowledged batch"
    );
    assert_eq!(cursor_update, CursorUpdate::Unchanged);
}

/// issue #25's investigation - the other half of the same
/// guarantee: a claim older than `checkout_lease` is abandoned, not
/// live, so this call succeeds exactly as if nothing had claimed it -
/// serving the batch and reclaiming the cursor with a fresh
/// `checked_out_at`, the self-healing path for a caller that was served
/// but crashed before acknowledging.
#[test]
fn consume_events_manual_ack_reclaims_a_stale_claim() {
    let et = event_type(true);
    let t = token(TokenStatus::Active, et.clone());
    let existing = ReadCursor {
        token: t.clone(),
        ack_mode: AckMode::ManualAck,
        sequence: 5,
        updated_at: timestamp(0),
        // Six minutes before `now` below - past the five-minute
        // checkout_lease() fixture, so stale.
        checked_out_at: Some(timestamp(200 - 6 * 60)),
    };
    let events = vec![event(et, 6)];

    let ConsumeEventsResult {
        served,
        cursor_update,
    } = event_store::consume_events(
        &t,
        Some(&existing),
        None,
        &events,
        &[],
        timestamp(200),
        checkout_lease(),
    )
    .unwrap();

    assert_eq!(
        served.iter().map(|e| e.sequence).collect::<Vec<_>>(),
        vec![6],
        "a stale claim must be reclaimed, not treated as still live"
    );
    assert_eq!(
        cursor_update,
        CursorUpdate::Claimed {
            checked_out_at: timestamp(200)
        }
    );
}

/// issue #25's investigation - the `manual_ack` twin of
/// `consume_events_provisions_a_cursor_on_first_use`: a brand new
/// cursor's very first serve claims it too, unlike `auto_advance`, which
/// never sets `checked_out_at` at all (that test's own assertion).
#[test]
fn consume_events_manual_ack_provisions_a_claimed_cursor_on_first_use() {
    let et = event_type(true);
    let t = token(TokenStatus::Active, et.clone());
    let events = vec![event(et, 0)];

    let ConsumeEventsResult { cursor_update, .. } = event_store::consume_events(
        &t,
        None,
        Some(AckMode::ManualAck),
        &events,
        &[],
        timestamp(100),
        checkout_lease(),
    )
    .unwrap();

    match cursor_update {
        CursorUpdate::Created(cursor) => {
            assert_eq!(cursor.ack_mode, AckMode::ManualAck);
            assert_eq!(cursor.checked_out_at, Some(timestamp(100)));
        }
        other => panic!("expected ReadCursor.created, got {other:?}"),
    }
}

/// Per the spec's `ensures` block, the `auto_advance` branch writes
/// `cursor.sequence = next_position` unconditionally, even when nothing
/// new matched and `next_position` falls back to the unchanged `position`
/// (`highest_sequence(served) ?? position`) - so this is still
/// `Advanced`, at the same sequence, not `Unchanged`. `Unchanged` is only
/// reachable for `manual_ack` with a still-live claim (issue
/// #25's investigation narrowed this - see
/// `consume_events_manual_ack_with_a_live_claim_serves_nothing` below).
#[test]
fn consume_events_auto_advance_with_nothing_new_still_reports_advanced_at_the_same_sequence() {
    let et = event_type(true);
    let t = token(TokenStatus::Active, et.clone());
    let existing = ReadCursor {
        token: t.clone(),
        ack_mode: AckMode::AutoAdvance,
        sequence: 5,
        updated_at: timestamp(0),
        checked_out_at: None,
    };

    let ConsumeEventsResult {
        served,
        cursor_update,
    } = event_store::consume_events(
        &t,
        Some(&existing),
        None,
        &[],
        &[],
        timestamp(200),
        checkout_lease(),
    )
    .unwrap();

    assert!(served.is_empty());
    assert_eq!(
        cursor_update,
        CursorUpdate::Advanced {
            sequence: 5,
            updated_at: timestamp(200)
        }
    );
}

/// rule-failure.ConsumeEvents.1 - `requires: token.status = active`.
#[test]
fn consume_events_rejects_a_revoked_token() {
    let t = token(TokenStatus::Revoked, event_type(true));

    let err = event_store::consume_events(
        &t,
        None,
        Some(AckMode::AutoAdvance),
        &[],
        &[],
        timestamp(0),
        checkout_lease(),
    )
    .unwrap_err();

    assert_eq!(err.code(), access_control::Error::TokenNotActive.code());
}

/// rule-failure.ConsumeEvents.2 - `requires: token.event_type.event_read_allowed = true`.
#[test]
fn consume_events_rejects_an_event_type_not_opted_into_reads() {
    let t = token(TokenStatus::Active, event_type(false));

    let err = event_store::consume_events(
        &t,
        None,
        Some(AckMode::AutoAdvance),
        &[],
        &[],
        timestamp(0),
        checkout_lease(),
    )
    .unwrap_err();

    assert_eq!(err.code(), event_store::Error::EventReadNotAllowed.code());
}

/// rule-failure.ConsumeEvents.3 - `requires: valid_filters(token.event_type, filters)`.
#[test]
fn consume_events_rejects_an_invalid_filter() {
    let t = token(TokenStatus::Active, event_type(true));
    // event_type()'s schema is "{}" - "amount" names no field it declares.
    let bogus_filter = vec![Filter {
        field: "amount".into(),
        operator: FilterOperator::GreaterThan,
        value: "10".into(),
    }];

    let err = event_store::consume_events(
        &t,
        None,
        Some(AckMode::AutoAdvance),
        &[],
        &bogus_filter,
        timestamp(0),
        checkout_lease(),
    )
    .unwrap_err();

    assert_eq!(err.code(), event_store::Error::InvalidFilter.code());
}

/// rule-failure.ConsumeEvents.4 - `requires: not is_new or ack_mode != null`.
#[test]
fn consume_events_rejects_a_first_call_with_no_ack_mode() {
    let t = token(TokenStatus::Active, event_type(true));

    let err = event_store::consume_events(&t, None, None, &[], &[], timestamp(0), checkout_lease())
        .unwrap_err();

    assert_eq!(err.code(), event_store::Error::CursorAckModeMismatch.code());
}

/// rule-failure.ConsumeEvents.5 - `requires: is_new or ack_mode = null or ack_mode = cursor.ack_mode`.
#[test]
fn consume_events_rejects_an_ack_mode_that_disagrees_with_the_stored_one() {
    let t = token(TokenStatus::Active, event_type(true));
    let existing = ReadCursor {
        token: t.clone(),
        ack_mode: AckMode::ManualAck,
        sequence: 0,
        updated_at: timestamp(0),
        checked_out_at: None,
    };

    let err = event_store::consume_events(
        &t,
        Some(&existing),
        Some(AckMode::AutoAdvance),
        &[],
        &[],
        timestamp(0),
        checkout_lease(),
    )
    .unwrap_err();

    assert_eq!(err.code(), event_store::Error::CursorAckModeMismatch.code());
}

// ---------------------------------------------------------------------
// rule-success.AcknowledgeEvents / rule-failure.AcknowledgeEvents.{1..4}
// ---------------------------------------------------------------------

#[test]
fn acknowledge_events_moves_the_cursor_forward() {
    let t = token(TokenStatus::Active, event_type(true));
    let cursor = ReadCursor {
        token: t.clone(),
        ack_mode: AckMode::ManualAck,
        sequence: 5,
        updated_at: timestamp(0),
        checked_out_at: None,
    };

    let (sequence, updated_at) =
        event_store::acknowledge_events(&t, Some(&cursor), 9, timestamp(300)).unwrap();

    assert_eq!(sequence, 9);
    assert_eq!(updated_at, timestamp(300));
}

#[test]
fn acknowledge_events_allows_reconfirming_ground_already_covered() {
    let t = token(TokenStatus::Active, event_type(true));
    let cursor = ReadCursor {
        token: t.clone(),
        ack_mode: AckMode::ManualAck,
        sequence: 5,
        updated_at: timestamp(0),
        checked_out_at: None,
    };

    let (sequence, _) =
        event_store::acknowledge_events(&t, Some(&cursor), 5, timestamp(300)).unwrap();

    assert_eq!(sequence, 5);
}

/// rule-failure.AcknowledgeEvents.1 - `requires: token.status = active`.
#[test]
fn acknowledge_events_rejects_a_revoked_token() {
    let t = token(TokenStatus::Revoked, event_type(true));
    let cursor = ReadCursor {
        token: t.clone(),
        ack_mode: AckMode::ManualAck,
        sequence: 0,
        updated_at: timestamp(0),
        checked_out_at: None,
    };

    let err = event_store::acknowledge_events(&t, Some(&cursor), 0, timestamp(0)).unwrap_err();

    assert_eq!(err.code(), access_control::Error::TokenNotActive.code());
}

/// rule-failure.AcknowledgeEvents.2 - `requires: exists cursor`.
#[test]
fn acknowledge_events_rejects_a_token_with_no_cursor_yet() {
    let t = token(TokenStatus::Active, event_type(true));

    let err = event_store::acknowledge_events(&t, None, 0, timestamp(0)).unwrap_err();

    assert_eq!(err.code(), event_store::Error::NoReadCursor.code());
}

/// rule-failure.AcknowledgeEvents.3 - `requires: cursor.ack_mode = manual_ack`.
#[test]
fn acknowledge_events_rejects_an_auto_advance_cursor() {
    let t = token(TokenStatus::Active, event_type(true));
    let cursor = ReadCursor {
        token: t.clone(),
        ack_mode: AckMode::AutoAdvance,
        sequence: 0,
        updated_at: timestamp(0),
        checked_out_at: None,
    };

    let err = event_store::acknowledge_events(&t, Some(&cursor), 0, timestamp(0)).unwrap_err();

    assert_eq!(err.code(), event_store::Error::NotManualAckCursor.code());
}

/// rule-failure.AcknowledgeEvents.4 - `requires: sequence >= cursor.sequence`.
#[test]
fn acknowledge_events_rejects_moving_the_cursor_backwards() {
    let t = token(TokenStatus::Active, event_type(true));
    let cursor = ReadCursor {
        token: t.clone(),
        ack_mode: AckMode::ManualAck,
        sequence: 5,
        updated_at: timestamp(0),
        checked_out_at: None,
    };

    let err = event_store::acknowledge_events(&t, Some(&cursor), 4, timestamp(0)).unwrap_err();

    assert_eq!(
        err.code(),
        event_store::Error::AcknowledgementRegresses.code()
    );
}

// ---------------------------------------------------------------------
// invariant.UniqueReadCursorPerToken
// ---------------------------------------------------------------------
//
// "At most one cursor per EventReadToken" is enforced by keying the
// (eventual sqlx-backed) cursor store by the token alone - the same
// UNIQUE-constraint pattern docs/architecture.md uses throughout. This
// property test simulates that store with a `HashMap<TokenId,
// ReadCursor>` and drives it through `consume_events` for randomly
// interleaved tokens, checking the map never needs a second entry for a
// token that already has one - i.e. the get-or-create call in
// `consume_events` is idempotent per token, which is what makes the
// invariant hold once a real store enforces the key.

fn token_id(t: &EventReadToken) -> (String, String) {
    (
        t.event_type.bounded_context.name.clone(),
        t.event_type.name.clone(),
    )
}

proptest! {
    #[test]
    fn unique_read_cursor_per_token_holds_across_interleaved_consumes(
        token_indices in prop::collection::vec(0..3usize, 1..30),
    ) {
        let tokens = [
            token(TokenStatus::Active, event_type(true)),
            token(TokenStatus::Active, other_event_type()),
            token(
                TokenStatus::Active,
                EventType {
                    bounded_context: bounded_context("billing"),
                    name: "InvoiceIssued".into(),
                    ..event_type(true)
                },
            ),
        ];
        let mut cursors: std::collections::HashMap<(String, String), ReadCursor> = std::collections::HashMap::new();

        for idx in token_indices {
            let t = &tokens[idx];
            let key = token_id(t);
            let existing = cursors.get(&key);
            let ack_mode = if existing.is_none() { Some(AckMode::AutoAdvance) } else { None };

            let result = event_store::consume_events(
                t,
                existing,
                ack_mode,
                &[],
                &[],
                Utc::now(),
                checkout_lease(),
            )
            .unwrap();
            if let CursorUpdate::Created(cursor) = result.cursor_update {
                let previous = cursors.insert(key.clone(), *cursor);
                prop_assert!(previous.is_none(), "a second ReadCursor.created for a token that already has one");
            }
        }

        // At most one entry per distinct token, by construction of the map key.
        prop_assert!(cursors.len() <= tokens.len());
    }
}

// ---------------------------------------------------------------------
// surface-actor.EventFetch / surface-exposure.EventFetch /
// surface-provides.EventFetch
// ---------------------------------------------------------------------
//
// These three obligation *ids* aren't bookkept as covered by name in
// any pass's own "obligations covered here" count - a documentation gap
// to close with a fresh pass over `allium plan`'s own output, not a
// missing-implementation one. Real coverage exists in
// `skilj/tests/event_fetch_rest.rs` (a wrong-secret 401, a correct
// EventFetch response, and `FetchEvents`/`ConsumeEvents`/
// `AcknowledgeEvents` all at their real routes).