pipecircuit 0.0.22

//! An atomic local copy fed by an owner's publications, including UI copies.
use super::ViewUpdate;
use crate::storage::warp_table::observed::*;
use crate::storage::warp_table::{Capacity, IndexKey};

pub struct ContinuousView<Id: IndexKey, Value: Clone + Eq> {
    capacity: Capacity,
    replica: Option<ViewReplica<Id, Value>>,
}
#[derive(Debug)]
pub enum ContinuousViewError {
    Creation(ViewCreationError),
    Apply(ViewApplyError),
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ContinuousViewStatus {
    AwaitingBaseline,
    Current,
    BaselineRequired,
}

impl<Id: IndexKey, Value: Clone + Eq> ContinuousView<Id, Value> {
    pub const fn new(capacity: Capacity) -> Self {
        Self {
            capacity,
            replica: None,
        }
    }
    /// Advanced stream diagnostics; ordinary consumers use `get` and `status`.
    pub fn copy(&self) -> Option<&ViewReplica<Id, Value>> {
        self.replica.as_ref()
    }
    pub fn get(&self, id: Id) -> Option<&Value> {
        self.replica.as_ref()?.get(id).map(|row| row.value)
    }
    pub fn rows(&self) -> impl Iterator<Item = (Id, &Value)> {
        self.replica
            .iter()
            .flat_map(|replica| replica.rows())
            .map(|row| (row.id, row.value))
    }
    pub fn status(&self) -> ContinuousViewStatus {
        match self.replica.as_ref().map(ViewReplica::status) {
            None | Some(ViewReplicaStatus::AwaitingBaseline) => {
                ContinuousViewStatus::AwaitingBaseline
            }
            Some(ViewReplicaStatus::Current { .. }) => ContinuousViewStatus::Current,
            Some(ViewReplicaStatus::BaselineRequired { .. }) => {
                ContinuousViewStatus::BaselineRequired
            }
        }
    }
    pub fn connection(&self) -> ViewConnection {
        self.replica
            .as_ref()
            .map_or(ViewConnection::Disconnected, ViewReplica::connection)
    }
    pub fn set_connection(&mut self, connection: ViewConnection) {
        if let Some(replica) = &mut self.replica {
            replica.set_connection(connection);
        }
    }
    /// A carrier must bind this copy to one subscription. A new authority
    /// incarnation requires an explicit reconnect baseline; old traffic cannot
    /// silently change which authority this copy represents.
    pub fn receive(
        &mut self,
        update: &ViewUpdate<Id, Value>,
    ) -> Result<ViewApplyOutcome, ContinuousViewError> {
        self.receive_reporting(update, None)
    }
    pub(crate) fn receive_reporting(
        &mut self,
        update: &ViewUpdate<Id, Value>,
        mut report: Option<&mut AppliedViewChanges<Id>>,
    ) -> Result<ViewApplyOutcome, ContinuousViewError> {
        match update {
            ViewUpdate::Baseline { through, rows } => {
                if self.replica.is_none() {
                    let mut replica =
                        ViewReplica::try_new(through.source, through.view, self.capacity)
                            .map_err(ContinuousViewError::Creation)?;
                    replica
                        .begin_baseline(*through, rows.len())
                        .map_err(ContinuousViewError::Apply)?;
                    replica
                        .baseline_chunk(*through, rows)
                        .map_err(ContinuousViewError::Apply)?;
                    let outcome = replica
                        .finish_baseline_reporting(*through, report.take())
                        .map_err(ContinuousViewError::Apply)?;
                    replica.set_connection(ViewConnection::Connected);
                    self.replica = Some(replica);
                    return Ok(outcome);
                }
                let replica = self.replica.as_mut().ok_or(ContinuousViewError::Apply(
                    ViewApplyError::BaselineNotStarted,
                ))?;
                if replica
                    .begin_baseline(*through, rows.len())
                    .map_err(ContinuousViewError::Apply)?
                    == ViewApplyOutcome::Stale
                {
                    return Ok(ViewApplyOutcome::Stale);
                }
                replica
                    .baseline_chunk(*through, rows)
                    .map_err(ContinuousViewError::Apply)?;
                replica
                    .finish_baseline_reporting(*through, report)
                    .map_err(ContinuousViewError::Apply)
            }
            ViewUpdate::Changes {
                base,
                through,
                changes,
            } => match &mut self.replica {
                Some(replica) => match replica
                    .prepare_changes(*base, *through, changes)
                    .map_err(ContinuousViewError::Apply)?
                {
                    ViewPreparation::Ready(prepared) => {
                        prepared
                            .publish_reporting(report)
                            .map_err(ContinuousViewError::Apply)?;
                        Ok(ViewApplyOutcome::Applied)
                    }
                    ViewPreparation::Stale => Ok(ViewApplyOutcome::Stale),
                    ViewPreparation::BaselineRequired => Ok(ViewApplyOutcome::BaselineRequired),
                },
                None => Ok(ViewApplyOutcome::BaselineRequired),
            },
        }
    }
    /// Called only after the carrier establishes a new subscription with its
    /// intended authority. Validate the entire baseline before replacing the
    /// old readable copy. A failed resynchronization leaves that copy intact.
    pub fn reconnect(
        &mut self,
        update: &ViewUpdate<Id, Value>,
    ) -> Result<ViewApplyOutcome, ContinuousViewError> {
        self.reconnect_reporting(update, None)
    }
    pub(crate) fn reconnect_reporting(
        &mut self,
        update: &ViewUpdate<Id, Value>,
        report: Option<&mut AppliedViewChanges<Id>>,
    ) -> Result<ViewApplyOutcome, ContinuousViewError> {
        let ViewUpdate::Baseline { through, .. } = update else {
            return Ok(ViewApplyOutcome::BaselineRequired);
        };
        let mut replacement = Self::new(self.capacity);
        let result = replacement.receive(update)?;
        if let Some(report) = report {
            let replica = replacement
                .replica
                .as_ref()
                .ok_or(ContinuousViewError::Apply(
                    ViewApplyError::BaselineNotStarted,
                ))?;
            report
                .replacement(self.replica.as_ref(), replica, *through)
                .map_err(ContinuousViewError::Apply)?;
        }
        self.replica = replacement.replica;
        Ok(result)
    }
}