pipecircuit 0.0.18

use super::super::inconsistency::invariant_violation;
use super::super::{
    Capacity, IndexKey, WarpTable, WarpTableGetOutcome, WarpTableInsertWithIdOutcome,
    WarpTableRemoveOutcome, WarpTableUpsertOutcome,
};
use super::view::{Projected, Seen, clear, projected_storage, seen_storage};
use super::{SourceId, ViewChange, ViewCreationError, ViewId, ViewRevision, ViewRow};
use crate::{
    ApplySynchronizedViewRevisionSpanOutcome, ApplySynchronizedViewSnapshotOutcome,
    SynchronizedView, SynchronizedViewRead,
};
use alloc::vec::Vec;

type Stamp = (u64, u64);

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ViewApplyOutcome {
    Applied,
    Stale,
    BaselineRequired,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ViewApplyError {
    WrongView,
    InvalidRevision,
    CapacityExceeded,
    DuplicateIdentity,
    InvalidIncarnation,
    UnknownRemoval,
    BaselineNotStarted,
    BaselineIncomplete,
    BaselineContainsRemoval,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ViewReplicaStatus {
    AwaitingBaseline,
    Current {
        through: ViewRevision,
    },
    BaselineRequired {
        through: ViewRevision,
        required: ViewRevision,
    },
}

/// Transport connectivity is independent of the last coherent state.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ViewConnection {
    Connected,
    Disconnected,
}

enum Staging {
    Empty,
    Receiving {
        through: ViewRevision,
        expected_rows: usize,
    },
}

pub struct ViewReplica<Id: IndexKey, Value: Clone + Eq> {
    source: SourceId,
    view: ViewId,
    sync: SynchronizedView<Stamp, ()>,
    rows: WarpTable<Projected<Id, Value>>,
    staging_rows: WarpTable<Projected<Id, Value>>,
    staging: Staging,
    seen: WarpTable<Seen<Id>>,
    pending: Vec<ViewChange<Id, Value>>,
    maximum_rows: usize,
    maximum_changes: usize,
    connection: ViewConnection,
}

impl<Id: IndexKey, Value: Clone + Eq> ViewReplica<Id, Value> {
    pub fn try_new(
        source: SourceId,
        view: ViewId,
        capacity: Capacity,
    ) -> Result<Self, ViewCreationError> {
        let maximum_rows = usize::try_from(capacity.value().get())
            .map_err(|_| ViewCreationError::CapacityOverflow)?;
        let maximum_changes = maximum_rows
            .checked_mul(2)
            .ok_or(ViewCreationError::CapacityOverflow)?;
        // A publication can remove all old identities and introduce all new ones.
        let changes_capacity = capacity
            .value()
            .get()
            .checked_mul(2)
            .and_then(core::num::NonZeroU32::new)
            .ok_or(ViewCreationError::CapacityOverflow)?;
        let rows = projected_storage(capacity)?;
        let staging_rows = projected_storage(capacity)?;
        let seen = seen_storage(Capacity::new(changes_capacity))?;
        let mut pending = Vec::new();
        pending
            .try_reserve_exact(maximum_changes)
            .map_err(ViewCreationError::StorageUnavailable)?;
        Ok(Self {
            source,
            view,
            sync: SynchronizedView::new((0, 0)),
            rows,
            staging_rows,
            staging: Staging::Empty,
            seen,
            pending,
            maximum_rows,
            maximum_changes,
            connection: ViewConnection::Disconnected,
        })
    }

    pub fn connection(&self) -> ViewConnection {
        self.connection
    }
    pub fn set_connection(&mut self, connection: ViewConnection) {
        self.connection = connection;
    }

    pub fn status(&self) -> ViewReplicaStatus {
        match self.sync.read() {
            SynchronizedViewRead::AwaitingSnapshot { .. } => ViewReplicaStatus::AwaitingBaseline,
            SynchronizedViewRead::Synchronized { through, .. } => ViewReplicaStatus::Current {
                through: self.revision(through),
            },
            SynchronizedViewRead::SynchronizationRequired {
                synchronized_through,
                observed_through,
                ..
            } => ViewReplicaStatus::BaselineRequired {
                through: self.revision(synchronized_through),
                required: self.revision(observed_through),
            },
        }
    }

    pub const fn len(&self) -> usize {
        self.rows.len()
    }

    pub const fn is_empty(&self) -> bool {
        self.rows.is_empty()
    }

    /// Reads the last coherent copy. Inspect `status` separately for freshness.
    pub fn get(&self, id: Id) -> Option<ViewRow<'_, Id, Value>> {
        match self.rows.get(id) {
            WarpTableGetOutcome::Present { row } => Some(ViewRow {
                id,
                incarnation: *row.incarnation,
                value: row.value,
            }),
            WarpTableGetOutcome::Absent { .. } => None,
        }
    }

    pub fn rows(&self) -> impl Iterator<Item = ViewRow<'_, Id, Value>> {
        self.rows.rows().map(|row| ViewRow {
            id: *row.id,
            incarnation: *row.incarnation,
            value: row.value,
        })
    }

    pub fn begin_baseline(
        &mut self,
        through: ViewRevision,
        expected_rows: usize,
    ) -> Result<ViewApplyOutcome, ViewApplyError> {
        self.validate_revision(through)?;
        if stamp(through) < self.sync.required_snapshot_through() {
            return Ok(ViewApplyOutcome::Stale);
        }
        if expected_rows > self.maximum_rows {
            return Err(ViewApplyError::CapacityExceeded);
        }
        clear(&mut self.staging_rows);
        self.sync.require_snapshot_through(stamp(through));
        self.staging = Staging::Receiving {
            through,
            expected_rows,
        };
        Ok(ViewApplyOutcome::BaselineRequired)
    }

    pub fn baseline_chunk(
        &mut self,
        through: ViewRevision,
        rows: &[ViewChange<Id, Value>],
    ) -> Result<(), ViewApplyError> {
        self.validate_revision(through)?;
        let Staging::Receiving {
            through: expected,
            expected_rows,
        } = self.staging
        else {
            return Err(ViewApplyError::BaselineNotStarted);
        };
        if through != expected {
            return Err(ViewApplyError::InvalidRevision);
        }
        let result = self.stage_chunk(rows, expected_rows);
        if result.is_err() {
            self.staging = Staging::Empty;
            clear(&mut self.staging_rows);
        }
        result
    }

    fn stage_chunk(
        &mut self,
        rows: &[ViewChange<Id, Value>],
        expected_rows: usize,
    ) -> Result<(), ViewApplyError> {
        let count = self
            .staging_rows
            .len()
            .checked_add(rows.len())
            .ok_or(ViewApplyError::CapacityExceeded)?;
        if count > expected_rows {
            return Err(ViewApplyError::CapacityExceeded);
        }
        for change in rows {
            let ViewChange::Upsert {
                id,
                incarnation,
                value,
            } = change
            else {
                return Err(ViewApplyError::BaselineContainsRemoval);
            };
            if *incarnation == 0 {
                return Err(ViewApplyError::InvalidIncarnation);
            }
            match self.staging_rows.insert_with_id(
                *id,
                Projected {
                    incarnation: *incarnation,
                    value: value.clone(),
                    marker: core::marker::PhantomData,
                },
            ) {
                WarpTableInsertWithIdOutcome::Inserted { .. } => {}
                WarpTableInsertWithIdOutcome::AlreadyPresent { .. } => {
                    return Err(ViewApplyError::DuplicateIdentity);
                }
                WarpTableInsertWithIdOutcome::Full { .. } => {
                    return Err(ViewApplyError::CapacityExceeded);
                }
                WarpTableInsertWithIdOutcome::UniqueConflict { conflict, .. } => match conflict {},
            }
        }
        Ok(())
    }

    pub fn finish_baseline(
        &mut self,
        through: ViewRevision,
    ) -> Result<ViewApplyOutcome, ViewApplyError> {
        Ok(match self.prepare_baseline(through)? {
            ViewBaselinePreparation::Ready(prepared) => {
                prepared.publish();
                ViewApplyOutcome::Applied
            }
            ViewBaselinePreparation::Stale => ViewApplyOutcome::Stale,
        })
    }

    /// Validates a complete staged baseline without exposing it. Prepare all
    /// parts of a grouped publication before publishing any of them.
    pub fn prepare_baseline(
        &mut self,
        through: ViewRevision,
    ) -> Result<ViewBaselinePreparation<'_, Id, Value>, ViewApplyError> {
        self.validate_revision(through)?;
        let Staging::Receiving {
            through: expected,
            expected_rows,
        } = self.staging
        else {
            return Err(ViewApplyError::BaselineNotStarted);
        };
        if through != expected {
            return Err(ViewApplyError::InvalidRevision);
        }
        if self.staging_rows.len() != expected_rows {
            return Err(ViewApplyError::BaselineIncomplete);
        }
        if stamp(through) < self.sync.required_snapshot_through() {
            clear(&mut self.staging_rows);
            self.staging = Staging::Empty;
            return Ok(ViewBaselinePreparation::Stale);
        }
        Ok(ViewBaselinePreparation::Ready(PreparedReplicaUpdate {
            replica: self,
            through,
            kind: PreparedKind::Baseline,
        }))
    }

    pub fn apply_changes(
        &mut self,
        base: ViewRevision,
        through: ViewRevision,
        changes: &[ViewChange<Id, Value>],
    ) -> Result<ViewApplyOutcome, ViewApplyError> {
        Ok(match self.prepare_changes(base, through, changes)? {
            ViewPreparation::Ready(prepared) => {
                prepared.publish();
                ViewApplyOutcome::Applied
            }
            ViewPreparation::Stale => ViewApplyOutcome::Stale,
            ViewPreparation::BaselineRequired => ViewApplyOutcome::BaselineRequired,
        })
    }

    /// Prepares an atomic update. An abandoned guard leaves readable rows intact.
    pub fn prepare_changes(
        &mut self,
        base: ViewRevision,
        through: ViewRevision,
        changes: &[ViewChange<Id, Value>],
    ) -> Result<ViewPreparation<'_, Id, Value>, ViewApplyError> {
        self.validate_revision(base)?;
        self.validate_revision(through)?;
        if base.epoch != through.epoch || base.revision >= through.revision {
            return Err(ViewApplyError::InvalidRevision);
        }
        match self.sync.read() {
            SynchronizedViewRead::Synchronized {
                through: current, ..
            } if stamp(through) <= current => return Ok(ViewPreparation::Stale),
            SynchronizedViewRead::SynchronizationRequired {
                synchronized_through,
                ..
            } if stamp(through) <= synchronized_through => return Ok(ViewPreparation::Stale),
            SynchronizedViewRead::Synchronized {
                through: current, ..
            } if stamp(base) == current => {}
            SynchronizedViewRead::AwaitingSnapshot { .. }
            | SynchronizedViewRead::Synchronized { .. }
            | SynchronizedViewRead::SynchronizationRequired { .. } => {
                self.sync.require_snapshot_through(stamp(through));
                return Ok(ViewPreparation::BaselineRequired);
            }
        }
        if let Err(error) = self.stage_changes(changes) {
            self.sync.require_snapshot_through(stamp(through));
            return Err(error);
        }
        Ok(ViewPreparation::Ready(PreparedReplicaUpdate {
            replica: self,
            through,
            kind: PreparedKind::Changes { base },
        }))
    }

    fn stage_changes(&mut self, changes: &[ViewChange<Id, Value>]) -> Result<(), ViewApplyError> {
        if changes.len() > self.maximum_changes {
            return Err(ViewApplyError::CapacityExceeded);
        }
        clear(&mut self.seen);
        self.pending.clear();
        let mut removed = 0_usize;
        let mut inserted = 0_usize;
        for change in changes {
            let (id, incarnation) = match change {
                ViewChange::Upsert {
                    id, incarnation, ..
                }
                | ViewChange::Remove { id, incarnation } => (*id, *incarnation),
            };
            if incarnation == 0 {
                return Err(ViewApplyError::InvalidIncarnation);
            }
            match self.seen.insert_with_id(
                id,
                Seen {
                    marker: core::marker::PhantomData,
                },
            ) {
                WarpTableInsertWithIdOutcome::Inserted { .. } => {}
                WarpTableInsertWithIdOutcome::AlreadyPresent { .. } => {
                    return Err(ViewApplyError::DuplicateIdentity);
                }
                WarpTableInsertWithIdOutcome::Full { .. } => {
                    return Err(ViewApplyError::CapacityExceeded);
                }
                WarpTableInsertWithIdOutcome::UniqueConflict { conflict, .. } => match conflict {},
            }
            match (change, self.rows.get(id)) {
                (ViewChange::Upsert { .. }, WarpTableGetOutcome::Present { row }) => {
                    if incarnation < *row.incarnation {
                        return Err(ViewApplyError::InvalidIncarnation);
                    }
                }
                (ViewChange::Upsert { .. }, WarpTableGetOutcome::Absent { .. }) => {
                    inserted = inserted
                        .checked_add(1)
                        .ok_or(ViewApplyError::CapacityExceeded)?;
                }
                (ViewChange::Remove { .. }, WarpTableGetOutcome::Present { row }) => {
                    if incarnation != *row.incarnation {
                        return Err(ViewApplyError::InvalidIncarnation);
                    }
                    removed = removed
                        .checked_add(1)
                        .ok_or(ViewApplyError::CapacityExceeded)?;
                }
                (ViewChange::Remove { .. }, WarpTableGetOutcome::Absent { .. }) => {
                    return Err(ViewApplyError::UnknownRemoval);
                }
            }
        }
        let count = self
            .rows
            .len()
            .checked_sub(removed)
            .and_then(|count| count.checked_add(inserted))
            .ok_or(ViewApplyError::CapacityExceeded)?;
        if count > self.maximum_rows {
            return Err(ViewApplyError::CapacityExceeded);
        }
        self.pending.extend(changes.iter().cloned());
        Ok(())
    }

    fn validate_revision(&self, revision: ViewRevision) -> Result<(), ViewApplyError> {
        if revision.source != self.source || revision.view != self.view {
            return Err(ViewApplyError::WrongView);
        }
        if revision.epoch == 0 {
            return Err(ViewApplyError::InvalidRevision);
        }
        Ok(())
    }

    fn revision(&self, (epoch, revision): Stamp) -> ViewRevision {
        ViewRevision {
            source: self.source,
            view: self.view,
            epoch,
            revision,
        }
    }
}

#[must_use]
pub enum ViewBaselinePreparation<'replica, Id: IndexKey, Value: Clone + Eq> {
    Ready(PreparedReplicaUpdate<'replica, Id, Value>),
    Stale,
}

/// Validation may require recovery without yielding an update to publish.
#[must_use]
pub enum ViewPreparation<'replica, Id: IndexKey, Value: Clone + Eq> {
    Ready(PreparedReplicaUpdate<'replica, Id, Value>),
    Stale,
    BaselineRequired,
}

enum PreparedKind {
    Baseline,
    Changes { base: ViewRevision },
}

#[must_use = "publish the prepared update or drop it to abandon it"]
pub struct PreparedReplicaUpdate<'replica, Id: IndexKey, Value: Clone + Eq> {
    replica: &'replica mut ViewReplica<Id, Value>,
    through: ViewRevision,
    kind: PreparedKind,
}

impl<Id: IndexKey, Value: Clone + Eq> PreparedReplicaUpdate<'_, Id, Value> {
    /// Cannot refuse after preparation. The exclusive guard keeps the validated
    /// revision and storage unchanged until publication.
    pub fn publish(self) -> ViewRevision {
        let replica = &mut *self.replica;
        match self.kind {
            PreparedKind::Baseline => match replica.sync.apply_snapshot(stamp(self.through), ()) {
                ApplySynchronizedViewSnapshotOutcome::Synchronized { .. } => {
                    core::mem::swap(&mut replica.rows, &mut replica.staging_rows);
                }
                ApplySynchronizedViewSnapshotOutcome::Stale { .. } => {
                    invariant_violation("prepared baseline became stale under exclusive guard")
                }
            },
            PreparedKind::Changes { base } => {
                let rows = &mut replica.rows;
                let pending = &mut replica.pending;
                match replica.sync.apply_revision_span(
                    stamp(base),
                    stamp(self.through),
                    |()| Ok::<(), core::convert::Infallible>(()),
                    |(), ()| Ok(()),
                    |(), ()| commit(rows, pending),
                ) {
                    ApplySynchronizedViewRevisionSpanOutcome::Applied { .. } => {}
                    ApplySynchronizedViewRevisionSpanOutcome::Stale { .. }
                    | ApplySynchronizedViewRevisionSpanOutcome::SnapshotRequired { .. } => {
                        invariant_violation(
                            "prepared changes lost continuity under exclusive guard",
                        )
                    }
                }
            }
        }
        self.through
    }
}

impl<Id: IndexKey, Value: Clone + Eq> Drop for PreparedReplicaUpdate<'_, Id, Value> {
    fn drop(&mut self) {
        match self.kind {
            PreparedKind::Baseline => {
                clear(&mut self.replica.staging_rows);
                self.replica.staging = Staging::Empty;
            }
            PreparedKind::Changes { .. } => self.replica.pending.clear(),
        }
    }
}

fn stamp(revision: ViewRevision) -> Stamp {
    (revision.epoch, revision.revision)
}

fn commit<Id: IndexKey, Value>(
    rows: &mut WarpTable<Projected<Id, Value>>,
    pending: &mut Vec<ViewChange<Id, Value>>,
) {
    for change in pending.iter() {
        if let ViewChange::Remove { id, .. } = change {
            match rows.remove(*id) {
                WarpTableRemoveOutcome::Removed { .. } => {}
                WarpTableRemoveOutcome::Absent { .. } => {
                    invariant_violation("validated view removal absent")
                }
            }
        }
    }
    for change in pending.drain(..) {
        if let ViewChange::Upsert {
            id,
            incarnation,
            value,
        } = change
        {
            match rows.upsert(
                id,
                Projected {
                    incarnation,
                    value,
                    marker: core::marker::PhantomData,
                },
            ) {
                WarpTableUpsertOutcome::Inserted { .. }
                | WarpTableUpsertOutcome::Replaced { .. } => {}
                WarpTableUpsertOutcome::Full { .. } => {
                    invariant_violation("validated view capacity exceeded")
                }
                WarpTableUpsertOutcome::UniqueConflict { conflict, .. } => match conflict {},
            }
        }
    }
}