pipecircuit 0.0.18

//! Owner-managed projections at explicit durability boundaries.
use super::{Authority, Error, Guarantee, PersistentRow, TableProtocol};
use crate::storage::warp_table::observed::*;
use crate::storage::warp_table::{
    Capacity, WarpTable, WarpTableInsertWithIdOutcome, WarpTableRemoveOutcome, WarpTableRow,
    WarpTableUpdateOutcome,
};
use alloc::vec::Vec;
use core::num::NonZeroU64;
#[cfg(test)]
mod evidence;

/// A closed, parameterized catalog of projections exposed by one table protocol.
pub trait ViewProtocol<Row: PersistentRow>: TableProtocol<Row> {
    type View: ViewDefinition<Row> + Clone + Eq;
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct SubscriptionId(u64);

#[derive(Clone, Copy, Debug)]
pub struct SubscriptionLimits {
    pub views: Capacity,
    pub subscribers: Capacity,
    pub retained_changes: Capacity,
}

#[derive(Debug)]
pub enum SubscriptionError {
    Authority(Error),
    Creation(ViewCreationError),
    Refresh(ViewRefreshError),
    Unknown,
    Full,
    Exhausted,
    Corrupt,
}

struct Materialization<Row: PersistentRow, Definition: ViewDefinition<Row>> {
    view: MaintainedView<Row, Definition>,
    delivery: ViewDeliveryBuffer<Row::Id, Definition::Value>,
}
struct Projection<Row: PersistentRow, Definition: ViewDefinition<Row>> {
    definition: Definition,
    guarantee: Guarantee,
    state: Option<Materialization<Row, Definition>>,
    subscribers: usize,
}
#[derive(WarpTableRow)]
#[index(primary = u64)]
struct RegisteredView<Row: PersistentRow, Definition: ViewDefinition<Row>> {
    projection: Projection<Row, Definition>,
}
struct Cursor {
    view: u64,
    delivery: Option<ViewSubscription>,
}
#[derive(WarpTableRow)]
#[index(primary = u64)]
struct Subscriber {
    cursor: Cursor,
}

pub struct Subscriptions<Row: PersistentRow, Definition: ViewDefinition<Row>> {
    views: WarpTable<RegisteredView<Row, Definition>>,
    subscribers: WarpTable<Subscriber>,
    limits: SubscriptionLimits,
    next: u64,
}

impl<Row: PersistentRow, Definition: ViewDefinition<Row> + Clone + Eq>
    Subscriptions<Row, Definition>
{
    pub fn new(limits: SubscriptionLimits) -> Result<Self, SubscriptionError> {
        Ok(Self {
            views: WarpTable::try_new(limits.views).map_err(|_| SubscriptionError::Full)?,
            subscribers: WarpTable::try_new(limits.subscribers)
                .map_err(|_| SubscriptionError::Full)?,
            limits,
            next: 0,
        })
    }

    pub fn subscribe(
        &mut self,
        authority: &Authority<Row>,
        definition: Definition,
        guarantee: Guarantee,
    ) -> Result<SubscriptionId, SubscriptionError> {
        match authority.observation(guarantee) {
            Ok(_) | Err(Error::Busy) => {}
            Err(error) => return Err(SubscriptionError::Authority(error)),
        }
        if self.subscribers.len() == self.limits.subscribers.value().get() as usize {
            return Err(SubscriptionError::Full);
        }
        let id = self
            .next
            .checked_add(1)
            .ok_or(SubscriptionError::Exhausted)?;
        let existing = self
            .views
            .rows()
            .find(|row| {
                row.projection.guarantee == guarantee && row.projection.definition == definition
            })
            .map(|row| *row.id);
        let view = match existing {
            Some(view) => {
                match self.views.update(view, |mut row| {
                    let projection = row.projection_mut();
                    projection.subscribers = projection
                        .subscribers
                        .checked_add(1)
                        .ok_or(SubscriptionError::Exhausted)?;
                    Ok(())
                }) {
                    WarpTableUpdateOutcome::Updated { output, .. } => output?,
                    WarpTableUpdateOutcome::Absent { .. } => {
                        return Err(SubscriptionError::Corrupt);
                    }
                }
                view
            }
            None => {
                let projection = Projection {
                    definition,
                    guarantee,
                    state: None,
                    subscribers: 1,
                };
                match self.views.insert_with_id(id, RegisteredView { projection }) {
                    WarpTableInsertWithIdOutcome::Inserted { .. } => {}
                    WarpTableInsertWithIdOutcome::Full { .. } => {
                        return Err(SubscriptionError::Full);
                    }
                    WarpTableInsertWithIdOutcome::AlreadyPresent { .. }
                    | WarpTableInsertWithIdOutcome::UniqueConflict { .. } => {
                        return Err(SubscriptionError::Corrupt);
                    }
                }
                id
            }
        };
        match self.subscribers.insert_with_id(
            id,
            Subscriber {
                cursor: Cursor {
                    view,
                    delivery: None,
                },
            },
        ) {
            WarpTableInsertWithIdOutcome::Inserted { .. } => {}
            WarpTableInsertWithIdOutcome::Full { .. }
            | WarpTableInsertWithIdOutcome::AlreadyPresent { .. }
            | WarpTableInsertWithIdOutcome::UniqueConflict { .. } => {
                return Err(SubscriptionError::Corrupt);
            }
        }
        self.next = id;
        Ok(SubscriptionId(id))
    }

    /// Called by the owner before delivery. Several successful command and
    /// completion turns may coalesce into one coherent publication.
    pub fn refresh(&mut self, authority: &Authority<Row>) -> Result<(), SubscriptionError> {
        for index in 0..self.views.len() {
            let id = *self
                .views
                .ids()
                .get(index)
                .ok_or(SubscriptionError::Corrupt)?;
            let retention = self.limits.retained_changes;
            match self.views.update(id, |mut row| {
                let projection = row.projection_mut();
                let source = match authority.observation(projection.guarantee) {
                    Ok(source) => source,
                    Err(Error::Busy) => return Ok(()),
                    Err(error) => return Err(SubscriptionError::Authority(error)),
                };
                match &mut projection.state {
                    Some(state) => {
                        state
                            .view
                            .refresh(source)
                            .map_err(SubscriptionError::Refresh)?;
                    }
                    None => {
                        let view = source
                            .maintain_with_retention(
                                ViewId::new(NonZeroU64::new(id).ok_or(SubscriptionError::Corrupt)?),
                                projection.definition.clone(),
                                retention,
                            )
                            .map_err(SubscriptionError::Creation)?;
                        let scratch = view
                            .delivery_buffer()
                            .map_err(SubscriptionError::Creation)?;
                        projection.state = Some(Materialization {
                            view,
                            delivery: scratch,
                        });
                    }
                }
                Ok(())
            }) {
                WarpTableUpdateOutcome::Updated { output, .. } => output?,
                WarpTableUpdateOutcome::Absent { .. } => return Err(SubscriptionError::Corrupt),
            }
        }
        Ok(())
    }

    pub fn ids(&self) -> impl Iterator<Item = SubscriptionId> + '_ {
        self.subscribers.ids().iter().copied().map(SubscriptionId)
    }
    pub fn capacity(&self) -> Capacity {
        self.limits.subscribers
    }

    pub fn unsubscribe(&mut self, id: SubscriptionId) -> Result<(), SubscriptionError> {
        let cursor = match self.subscribers.remove(id.0) {
            WarpTableRemoveOutcome::Removed { removed, .. } => removed.cursor,
            WarpTableRemoveOutcome::Absent { .. } => return Err(SubscriptionError::Unknown),
        };
        let remaining = match self.views.update(cursor.view, |mut row| {
            let projection = row.projection_mut();
            projection.subscribers = projection
                .subscribers
                .checked_sub(1)
                .ok_or(SubscriptionError::Corrupt)?;
            Ok(projection.subscribers)
        }) {
            WarpTableUpdateOutcome::Updated { output, .. } => output?,
            WarpTableUpdateOutcome::Absent { .. } => return Err(SubscriptionError::Corrupt),
        };
        if remaining == 0 {
            match self.views.remove(cursor.view) {
                WarpTableRemoveOutcome::Removed { .. } => {}
                WarpTableRemoveOutcome::Absent { .. } => return Err(SubscriptionError::Corrupt),
            }
        }
        Ok(())
    }

    /// The carrier must own the entire delivery before admitting it. A pending
    /// replicated baseline has no delivery yet; its cursor remains unchanged.
    pub fn transmit<E>(
        &mut self,
        id: SubscriptionId,
        send: impl FnOnce(ViewDelivery<'_, Row::Id, Definition::Value>) -> Result<ViewAdmission, E>,
    ) -> Result<Option<Result<ViewTransmitOutcome, ViewTransmitError<E>>>, SubscriptionError> {
        let views = &mut self.views;
        match self.subscribers.update(id.0, |mut subscriber| {
            let cursor = subscriber.cursor_mut();
            match views.update(cursor.view, |mut row| {
                let projection = row.projection_mut();
                let Some(Materialization {
                    view,
                    delivery: scratch,
                }) = &mut projection.state
                else {
                    return None;
                };
                let delivery = cursor.delivery.get_or_insert_with(|| view.subscribe());
                Some(view.transmit(delivery, scratch, send))
            }) {
                WarpTableUpdateOutcome::Updated { output, .. } => Ok(output),
                WarpTableUpdateOutcome::Absent { .. } => Err(SubscriptionError::Corrupt),
            }
        }) {
            WarpTableUpdateOutcome::Updated { output, .. } => output,
            WarpTableUpdateOutcome::Absent { .. } => Err(SubscriptionError::Unknown),
        }
    }

    pub fn require_baseline(&mut self, id: SubscriptionId) -> Result<(), SubscriptionError> {
        match self.subscribers.update(id.0, |mut row| {
            row.cursor_mut().delivery = None;
        }) {
            WarpTableUpdateOutcome::Updated { .. } => Ok(()),
            WarpTableUpdateOutcome::Absent { .. } => Err(SubscriptionError::Unknown),
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub enum ViewUpdate<Id, Value> {
    Baseline {
        through: ViewRevision,
        rows: Vec<ViewChange<Id, Value>>,
    },
    Changes {
        base: ViewRevision,
        through: ViewRevision,
        changes: Vec<ViewChange<Id, Value>>,
    },
}

impl<Id: crate::storage::warp_table::IndexKey, Value: Clone> ViewUpdate<Id, Value> {
    pub fn capture(delivery: ViewDelivery<'_, Id, Value>) -> Option<Self> {
        match delivery {
            ViewDelivery::Baseline { through, rows } => Some(Self::Baseline {
                through,
                rows: rows
                    .map(|row| ViewChange::Upsert {
                        id: row.id,
                        incarnation: row.incarnation,
                        value: row.value.clone(),
                    })
                    .collect(),
            }),
            ViewDelivery::Changes {
                base,
                through,
                changes,
            } => Some(Self::Changes {
                base,
                through,
                changes: changes.to_vec(),
            }),
            ViewDelivery::Current { .. } => None,
        }
    }
}