pipecircuit 0.0.11

use super::*;

#[derive(Clone, Copy, Eq, Ord, PartialEq, PartialOrd)]
struct Revision(u8);

impl SynchronizedViewRevision for Revision {
    type ChangeId = u8;

    fn changed_through(change_id: Self::ChangeId) -> Self {
        Self(change_id)
    }

    fn next_change_id(self) -> Option<Self::ChangeId> {
        self.0.checked_add(1)
    }
}

#[test]
fn a_gap_retains_the_last_coherent_projection_until_a_covering_snapshot() {
    let mut view = SynchronizedView::new(Revision(0));
    assert!(matches!(
        view.apply_snapshot(Revision(2), 20),
        ApplySynchronizedViewSnapshotOutcome::Synchronized {
            through: Revision(2)
        }
    ));

    assert!(matches!(
        view.apply_change(
            4,
            |_| Ok::<_, ()>(40),
            |_, _| Ok(()),
            |projection, value| {
                *projection = value;
            }
        ),
        ApplySynchronizedViewChangeOutcome::SnapshotRequired {
            observed_through: Revision(4),
            reason: SynchronizedViewSnapshotRequiredReason::ChangeGap {
                synchronized_through: Revision(2)
            }
        }
    ));
    assert!(matches!(
        view.read(),
        SynchronizedViewRead::SynchronizationRequired {
            synchronized_through: Revision(2),
            observed_through: Revision(4),
            projection: 20,
        }
    ));
    assert!(matches!(
        view.apply_snapshot(Revision(3), 30),
        ApplySynchronizedViewSnapshotOutcome::Stale {
            snapshot_through: Revision(3),
            required_through: Revision(4),
        }
    ));
    assert!(matches!(
        view.apply_snapshot(Revision(4), 40),
        ApplySynchronizedViewSnapshotOutcome::Synchronized {
            through: Revision(4)
        }
    ));
    assert!(matches!(
        view.read(),
        SynchronizedViewRead::Synchronized {
            through: Revision(4),
            projection: 40,
        }
    ));
}

#[test]
fn only_the_exact_next_change_reaches_the_projection_reducer() {
    let mut view = SynchronizedView::new(Revision(0));
    assert!(matches!(
        view.apply_snapshot(Revision(1), 10),
        ApplySynchronizedViewSnapshotOutcome::Synchronized { .. }
    ));

    let mut stale_called = false;
    assert!(matches!(
        view.apply_change(
            1,
            |_| {
                stale_called = true;
                Ok::<_, ()>(())
            },
            |_, _| Ok(()),
            |_, ()| {},
        ),
        ApplySynchronizedViewChangeOutcome::Stale { .. }
    ));
    assert!(!stale_called);

    assert!(matches!(
        view.apply_change(
            2,
            |_| Ok::<_, ()>(20),
            |_, _| Ok(()),
            |projection, value| {
                *projection = value;
            }
        ),
        ApplySynchronizedViewChangeOutcome::Applied {
            through: Revision(2)
        }
    ));
    assert!(matches!(
        view.read(),
        SynchronizedViewRead::Synchronized {
            through: Revision(2),
            projection: 20,
        }
    ));
}

#[test]
fn a_rejected_next_change_requires_recovery_without_advancing_the_revision() {
    let mut view = SynchronizedView::new(Revision(0));
    assert!(matches!(
        view.apply_snapshot(Revision(1), 10),
        ApplySynchronizedViewSnapshotOutcome::Synchronized { .. }
    ));

    assert!(matches!(
        view.apply_change(2, |_| Err("invalid"), |_, &()| Ok(()), |_, ()| {}),
        ApplySynchronizedViewChangeOutcome::SnapshotRequired {
            observed_through: Revision(2),
            reason: SynchronizedViewSnapshotRequiredReason::ChangeRejected("invalid"),
        }
    ));
    assert!(matches!(
        view.read(),
        SynchronizedViewRead::SynchronizationRequired {
            synchronized_through: Revision(1),
            observed_through: Revision(2),
            projection: 10,
        }
    ));
}

#[test]
fn changes_seen_before_hydration_raise_the_minimum_acceptable_snapshot() {
    let mut view = SynchronizedView::new(Revision(0));
    assert!(matches!(
        view.apply_change(2, |_| Ok::<_, ()>(()), |_, _| Ok(()), |_, ()| {}),
        ApplySynchronizedViewChangeOutcome::SnapshotRequired {
            observed_through: Revision(2),
            reason: SynchronizedViewSnapshotRequiredReason::NoSnapshotApplied,
        }
    ));
    assert!(matches!(
        view.apply_change(4, |_| Ok::<_, ()>(()), |_, _| Ok(()), |_, ()| {}),
        ApplySynchronizedViewChangeOutcome::SnapshotRequired {
            observed_through: Revision(4),
            reason: SynchronizedViewSnapshotRequiredReason::NoSnapshotApplied,
        }
    ));
    assert!(matches!(
        view.apply_snapshot(Revision(3), 30),
        ApplySynchronizedViewSnapshotOutcome::Stale {
            required_through: Revision(4),
            ..
        }
    ));
}

#[test]
fn an_explicit_requirement_accepts_an_equal_revision_snapshot() {
    let mut view = SynchronizedView::new(Revision(0));
    assert!(matches!(
        view.apply_snapshot(Revision(2), 20),
        ApplySynchronizedViewSnapshotOutcome::Synchronized { .. }
    ));
    view.require_snapshot_through(Revision(2));
    assert!(matches!(
        view.read(),
        SynchronizedViewRead::SynchronizationRequired {
            synchronized_through: Revision(2),
            observed_through: Revision(2),
            projection: 20,
        }
    ));
    assert!(matches!(
        view.apply_snapshot(Revision(2), 21),
        ApplySynchronizedViewSnapshotOutcome::Synchronized {
            through: Revision(2)
        }
    ));
}

#[test]
fn preparation_and_reservation_failures_do_not_commit() {
    let mut view = SynchronizedView::new(Revision(0));
    assert!(matches!(
        view.apply_snapshot(Revision(1), 10),
        ApplySynchronizedViewSnapshotOutcome::Synchronized { .. }
    ));
    let mut committed = false;

    assert!(matches!(
        view.apply_change(
            2,
            |_| Ok::<_, &str>(20),
            |_, _| Err("unavailable"),
            |projection, value| {
                committed = true;
                *projection = value;
            },
        ),
        ApplySynchronizedViewChangeOutcome::SnapshotRequired {
            reason: SynchronizedViewSnapshotRequiredReason::ChangeRejected("unavailable"),
            ..
        }
    ));
    assert!(!committed);
    assert!(matches!(
        view.read(),
        SynchronizedViewRead::SynchronizationRequired { projection: 10, .. }
    ));
}