pipecircuit 0.0.23

pub trait SynchronizedViewRevision: Copy + Ord {
    type ChangeId: Copy + Ord;

    fn changed_through(change_id: Self::ChangeId) -> Self;

    fn next_change_id(self) -> Option<Self::ChangeId>;
}

pub enum SynchronizedViewSnapshotRequiredReason<Revision, Rejection> {
    NoSnapshotApplied,
    ChangeGap { synchronized_through: Revision },
    ChangeRejected(Rejection),
    AlreadyRequired,
}

#[must_use]
pub enum ApplySynchronizedViewSnapshotOutcome<Revision> {
    Synchronized {
        through: Revision,
    },
    Stale {
        snapshot_through: Revision,
        required_through: Revision,
    },
}

#[must_use]
pub enum ApplySynchronizedViewChangeOutcome<Revision, Rejection> {
    Applied {
        through: Revision,
    },
    Stale {
        change_through: Revision,
        synchronized_through: Revision,
    },
    SnapshotRequired {
        observed_through: Revision,
        reason: SynchronizedViewSnapshotRequiredReason<Revision, Rejection>,
    },
}

pub enum ApplySynchronizedViewRevisionSpanOutcome<Revision, Rejection> {
    Applied {
        through: Revision,
    },
    Stale {
        span_through: Revision,
        synchronized_through: Revision,
    },
    SnapshotRequired {
        observed_through: Revision,
        reason: SynchronizedViewSnapshotRequiredReason<Revision, Rejection>,
    },
}

pub enum SynchronizedViewRead<'a, Revision, Projection> {
    AwaitingSnapshot {
        observed_through: Revision,
    },
    Synchronized {
        through: Revision,
        projection: &'a Projection,
    },
    SynchronizationRequired {
        synchronized_through: Revision,
        observed_through: Revision,
        projection: &'a Projection,
    },
}

enum SynchronizedViewState<Revision, Projection> {
    AwaitingSnapshot {
        observed_through: Revision,
    },
    Observing {
        synchronized_through: Revision,
        observed_through: Revision,
        synchronization: SynchronizedViewSynchronization,
        projection: Projection,
    },
}

enum SynchronizedViewSynchronization {
    Synchronized,
    Required,
}

pub struct SynchronizedView<Revision, Projection> {
    state: SynchronizedViewState<Revision, Projection>,
}

impl<Revision, Projection> SynchronizedView<Revision, Projection>
where
    Revision: Copy + Ord,
{
    pub const fn new(initial_revision: Revision) -> Self {
        Self {
            state: SynchronizedViewState::AwaitingSnapshot {
                observed_through: initial_revision,
            },
        }
    }

    pub fn apply_snapshot(
        &mut self,
        snapshot_through: Revision,
        projection: Projection,
    ) -> ApplySynchronizedViewSnapshotOutcome<Revision> {
        let required_through = self.required_snapshot_through();
        if snapshot_through < required_through {
            return ApplySynchronizedViewSnapshotOutcome::Stale {
                snapshot_through,
                required_through,
            };
        }
        self.state = SynchronizedViewState::Observing {
            synchronized_through: snapshot_through,
            observed_through: snapshot_through,
            synchronization: SynchronizedViewSynchronization::Synchronized,
            projection,
        };
        ApplySynchronizedViewSnapshotOutcome::Synchronized {
            through: snapshot_through,
        }
    }

    pub fn required_snapshot_through(&self) -> Revision {
        match &self.state {
            SynchronizedViewState::AwaitingSnapshot { observed_through }
            | SynchronizedViewState::Observing {
                observed_through, ..
            } => *observed_through,
        }
    }

    pub fn read(&self) -> SynchronizedViewRead<'_, Revision, Projection> {
        match &self.state {
            SynchronizedViewState::AwaitingSnapshot { observed_through } => {
                SynchronizedViewRead::AwaitingSnapshot {
                    observed_through: *observed_through,
                }
            }
            SynchronizedViewState::Observing {
                synchronized_through,
                observed_through: _,
                synchronization: SynchronizedViewSynchronization::Synchronized,
                projection,
            } => SynchronizedViewRead::Synchronized {
                through: *synchronized_through,
                projection,
            },
            SynchronizedViewState::Observing {
                synchronized_through,
                observed_through,
                synchronization: SynchronizedViewSynchronization::Required,
                projection,
            } => SynchronizedViewRead::SynchronizationRequired {
                synchronized_through: *synchronized_through,
                observed_through: *observed_through,
                projection,
            },
        }
    }

    pub fn require_snapshot_through(&mut self, required_through: Revision) {
        match &mut self.state {
            SynchronizedViewState::AwaitingSnapshot { observed_through } => {
                if required_through > *observed_through {
                    *observed_through = required_through;
                }
            }
            SynchronizedViewState::Observing {
                observed_through,
                synchronization,
                ..
            } => {
                if required_through > *observed_through {
                    *observed_through = required_through;
                }
                *synchronization = SynchronizedViewSynchronization::Required;
            }
        }
    }

    pub fn apply_revision_span<Prepared, Rejection, Prepare, Reserve, Commit>(
        &mut self,
        expected: Revision,
        through: Revision,
        prepare: Prepare,
        reserve: Reserve,
        commit: Commit,
    ) -> ApplySynchronizedViewRevisionSpanOutcome<Revision, Rejection>
    where
        Prepare: FnOnce(&Projection) -> Result<Prepared, Rejection>,
        Reserve: FnOnce(&mut Projection, &Prepared) -> Result<(), Rejection>,
        Commit: FnOnce(&mut Projection, Prepared),
    {
        match &mut self.state {
            SynchronizedViewState::AwaitingSnapshot { observed_through } => {
                if through > *observed_through {
                    *observed_through = through;
                }
                ApplySynchronizedViewRevisionSpanOutcome::SnapshotRequired {
                    observed_through: *observed_through,
                    reason: SynchronizedViewSnapshotRequiredReason::NoSnapshotApplied,
                }
            }
            SynchronizedViewState::Observing {
                synchronized_through,
                ..
            } if through <= *synchronized_through => {
                ApplySynchronizedViewRevisionSpanOutcome::Stale {
                    span_through: through,
                    synchronized_through: *synchronized_through,
                }
            }
            SynchronizedViewState::Observing {
                observed_through,
                synchronization: SynchronizedViewSynchronization::Required,
                ..
            } => {
                if through > *observed_through {
                    *observed_through = through;
                }
                ApplySynchronizedViewRevisionSpanOutcome::SnapshotRequired {
                    observed_through: *observed_through,
                    reason: SynchronizedViewSnapshotRequiredReason::AlreadyRequired,
                }
            }
            SynchronizedViewState::Observing {
                synchronized_through,
                observed_through,
                synchronization,
                projection,
            } => {
                if expected != *synchronized_through || through <= expected {
                    if through > *observed_through {
                        *observed_through = through;
                    }
                    *synchronization = SynchronizedViewSynchronization::Required;
                    return ApplySynchronizedViewRevisionSpanOutcome::SnapshotRequired {
                        observed_through: *observed_through,
                        reason: SynchronizedViewSnapshotRequiredReason::ChangeGap {
                            synchronized_through: *synchronized_through,
                        },
                    };
                }
                match prepare(projection).and_then(|prepared| {
                    reserve(projection, &prepared)?;
                    Ok(prepared)
                }) {
                    Ok(prepared) => {
                        commit(projection, prepared);
                        *synchronized_through = through;
                        *observed_through = through;
                        ApplySynchronizedViewRevisionSpanOutcome::Applied { through }
                    }
                    Err(rejection) => {
                        *observed_through = through;
                        *synchronization = SynchronizedViewSynchronization::Required;
                        ApplySynchronizedViewRevisionSpanOutcome::SnapshotRequired {
                            observed_through: through,
                            reason: SynchronizedViewSnapshotRequiredReason::ChangeRejected(
                                rejection,
                            ),
                        }
                    }
                }
            }
        }
    }
}

impl<Revision, Projection> SynchronizedView<Revision, Projection>
where
    Revision: SynchronizedViewRevision,
{
    pub fn apply_change<Prepared, Rejection, Prepare, Reserve, Commit>(
        &mut self,
        change_id: Revision::ChangeId,
        prepare: Prepare,
        reserve: Reserve,
        commit: Commit,
    ) -> ApplySynchronizedViewChangeOutcome<Revision, Rejection>
    where
        Prepare: FnOnce(&Projection) -> Result<Prepared, Rejection>,
        Reserve: FnOnce(&mut Projection, &Prepared) -> Result<(), Rejection>,
        Commit: FnOnce(&mut Projection, Prepared),
    {
        let change_through = Revision::changed_through(change_id);
        match &mut self.state {
            SynchronizedViewState::AwaitingSnapshot { observed_through } => {
                if change_through > *observed_through {
                    *observed_through = change_through;
                }
                ApplySynchronizedViewChangeOutcome::SnapshotRequired {
                    observed_through: *observed_through,
                    reason: SynchronizedViewSnapshotRequiredReason::NoSnapshotApplied,
                }
            }
            SynchronizedViewState::Observing {
                synchronized_through,
                ..
            } if change_through <= *synchronized_through => {
                ApplySynchronizedViewChangeOutcome::Stale {
                    change_through,
                    synchronized_through: *synchronized_through,
                }
            }
            SynchronizedViewState::Observing {
                synchronized_through: _,
                observed_through,
                synchronization: SynchronizedViewSynchronization::Required,
                ..
            } => {
                if change_through > *observed_through {
                    *observed_through = change_through;
                }
                ApplySynchronizedViewChangeOutcome::SnapshotRequired {
                    observed_through: *observed_through,
                    reason: SynchronizedViewSnapshotRequiredReason::AlreadyRequired,
                }
            }
            SynchronizedViewState::Observing {
                synchronized_through,
                observed_through,
                synchronization,
                projection,
            } => {
                if synchronized_through.next_change_id() != Some(change_id) {
                    *observed_through = change_through;
                    *synchronization = SynchronizedViewSynchronization::Required;
                    return ApplySynchronizedViewChangeOutcome::SnapshotRequired {
                        observed_through: change_through,
                        reason: SynchronizedViewSnapshotRequiredReason::ChangeGap {
                            synchronized_through: *synchronized_through,
                        },
                    };
                }
                match prepare(projection).and_then(|prepared| {
                    reserve(projection, &prepared)?;
                    Ok(prepared)
                }) {
                    Ok(prepared) => {
                        commit(projection, prepared);
                        *synchronized_through = change_through;
                        *observed_through = change_through;
                        ApplySynchronizedViewChangeOutcome::Applied {
                            through: change_through,
                        }
                    }
                    Err(rejection) => {
                        *observed_through = change_through;
                        *synchronization = SynchronizedViewSynchronization::Required;
                        ApplySynchronizedViewChangeOutcome::SnapshotRequired {
                            observed_through: change_through,
                            reason: SynchronizedViewSnapshotRequiredReason::ChangeRejected(
                                rejection,
                            ),
                        }
                    }
                }
            }
        }
    }
}

#[cfg(test)]
mod tests;

#[cfg(kani)]
mod proofs;