pipecircuit 0.0.23

use super::*;

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum AppliedViewChange<Id> {
    Entered { id: Id },
    Updated { id: Id },
    Replaced { id: Id },
    Left { id: Id },
}

pub struct AppliedViewChanges<Id> {
    changes: Vec<AppliedViewChange<Id>>,
    maximum: usize,
}

impl<Id> AppliedViewChanges<Id> {
    pub fn try_new(view_capacity: Capacity) -> Result<Self, ViewCreationError> {
        let maximum = usize::try_from(view_capacity.value().get())
            .ok()
            .and_then(|rows| rows.checked_mul(2))
            .ok_or(ViewCreationError::CapacityOverflow)?;
        let mut changes = Vec::new();
        changes
            .try_reserve_exact(maximum)
            .map_err(ViewCreationError::StorageUnavailable)?;
        Ok(Self { changes, maximum })
    }

    pub(crate) fn clear(&mut self) {
        self.changes.clear();
    }

    pub(crate) fn as_slice(&self) -> &[AppliedViewChange<Id>] {
        &self.changes
    }

    fn push(&mut self, change: AppliedViewChange<Id>) -> Result<(), ViewApplyError> {
        if self.changes.len() == self.maximum {
            return Err(ViewApplyError::ReportCapacityExceeded);
        }
        self.changes.push(change);
        Ok(())
    }
}

impl<Id: IndexKey> AppliedViewChanges<Id> {
    fn compare_row<Value: Eq>(
        &mut self,
        id: Id,
        before: Option<(u64, &Value)>,
        after: (u64, &Value),
    ) -> Result<(), ViewApplyError> {
        let change = match (before, after) {
            (None, _) => AppliedViewChange::Entered { id },
            (Some((old_identity, _)), (new_identity, _)) if old_identity != new_identity => {
                AppliedViewChange::Replaced { id }
            }
            (Some((_, old)), (_, new)) if old != new => AppliedViewChange::Updated { id },
            (Some(_), _) => return Ok(()),
        };
        self.push(change)
    }

    pub(crate) fn replacement<Value: Clone + Eq>(
        &mut self,
        before: Option<&ViewReplica<Id, Value>>,
        after: &ViewReplica<Id, Value>,
        through: ViewRevision,
    ) -> Result<(), ViewApplyError> {
        self.clear();
        if let Some(before) = before {
            let same_authority = before.source == after.source
                && before.view == after.view
                && before.installed_epoch() == Some(through.epoch);
            for row in before.rows() {
                match after.get(row.id) {
                    Some(_) if !same_authority => {
                        self.push(AppliedViewChange::Replaced { id: row.id })?
                    }
                    Some(next) => self.compare_row(
                        row.id,
                        Some((row.incarnation, row.value)),
                        (next.incarnation, next.value),
                    )?,
                    None => self.push(AppliedViewChange::Left { id: row.id })?,
                }
            }
        }
        for row in after.rows() {
            if before.is_none_or(|before| before.get(row.id).is_none()) {
                self.push(AppliedViewChange::Entered { id: row.id })?;
            }
        }
        Ok(())
    }
}

impl<Id: IndexKey, Value: Clone + Eq> PreparedReplicaUpdate<'_, Id, Value> {
    pub(crate) fn publish_reporting(
        self,
        report: Option<&mut AppliedViewChanges<Id>>,
    ) -> Result<ViewRevision, ViewApplyError> {
        if let Some(report) = report {
            report.clear();
            let replica = &self.replica;
            match self.kind {
                PreparedKind::Changes { .. } => {
                    for change in &replica.pending {
                        match change {
                            ViewChange::Upsert {
                                id,
                                incarnation,
                                value,
                            } => report.compare_row(
                                *id,
                                replica.get(*id).map(|row| (row.incarnation, row.value)),
                                (*incarnation, value),
                            )?,
                            ViewChange::Remove { id, .. } => {
                                report.push(AppliedViewChange::Left { id: *id })?
                            }
                        }
                    }
                }
                PreparedKind::Baseline => {
                    let same_epoch = replica.installed_epoch() == Some(self.through.epoch);
                    for row in replica.rows() {
                        let next = match replica.staging_rows.get(row.id) {
                            WarpTableGetOutcome::Present { row } => {
                                Some((*row.incarnation, row.value))
                            }
                            WarpTableGetOutcome::Absent { .. } => None,
                        };
                        if !same_epoch && next.is_some() {
                            report.push(AppliedViewChange::Replaced { id: row.id })?;
                        } else if let Some(next) = next {
                            report.compare_row(row.id, Some((row.incarnation, row.value)), next)?;
                        } else {
                            report.push(AppliedViewChange::Left { id: row.id })?;
                        }
                    }
                    for row in replica.staging_rows.rows() {
                        if replica.get(*row.id).is_none() {
                            report.push(AppliedViewChange::Entered { id: *row.id })?;
                        }
                    }
                }
            }
        }
        Ok(self.publish())
    }
}

impl<Id: IndexKey, Value: Clone + Eq> ViewReplica<Id, Value> {
    fn installed_epoch(&self) -> Option<u64> {
        match self.status() {
            ViewReplicaStatus::Current { through }
            | ViewReplicaStatus::BaselineRequired { through, .. } => Some(through.epoch),
            ViewReplicaStatus::AwaitingBaseline => None,
        }
    }
}

#[cfg(kani)]
mod proofs;