liminal-server 0.25.4

use super::{
    BTreeMap, BTreeSet, Decided, MappedChannel, MembershipAction, MembershipRefusal,
    MembershipSubject, Mode, PageRow, Resource, Subject,
};

#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub(super) struct DecisionKey {
    pub(super) subject: MembershipSubject,
    pub(super) channel: String,
    pub(super) action: MembershipAction,
}

/// A retained decision, or the ticket of the ask settling it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum Indexed {
    Applying { ticket: u64 },
    Allowed { grant: String, revision: u64 },
    Refused { refusal: String },
    Held { mode: Mode },
}

impl Indexed {
    pub(super) fn result(&self) -> Result<Decided, MembershipRefusal> {
        match self {
            Self::Applying { .. } => Err(MembershipRefusal::ChangeApplying),
            Self::Allowed { grant, revision } => Ok(Decided::Allowed {
                grant: grant.clone(),
                revision: *revision,
            }),
            Self::Refused { refusal } => Err(MembershipRefusal::Refused {
                refusal: refusal.clone(),
            }),
            Self::Held { mode } => Err(MembershipRefusal::Held { mode: *mode }),
        }
    }
}

pub(super) type ResourceKey = (String, String);

pub(super) fn resource_key(resource: &Resource) -> ResourceKey {
    (resource.kind.clone(), resource.id.clone())
}

#[derive(Debug)]
pub(super) struct Entry {
    pub(super) verdict: Indexed,
    pub(super) connections: BTreeSet<u64>,
    pub(super) reached_by: [ResourceKey; 2],
}

#[derive(Debug, Default)]
pub(super) struct DecisionIndex {
    pub(super) generation: Option<u64>,
    pub(super) applied: u64,
    pub(super) next_ticket: u64,
    pub(super) entries: BTreeMap<DecisionKey, Entry>,
    pub(super) by_resource: BTreeMap<ResourceKey, BTreeSet<DecisionKey>>,
    pub(super) by_connection: BTreeMap<u64, BTreeSet<DecisionKey>>,
}

impl DecisionIndex {
    pub(super) const fn adopt(&mut self, generation: u64) -> Result<(), MembershipRefusal> {
        match self.generation {
            None => {
                self.generation = Some(generation);
                Ok(())
            }
            Some(current) if current == generation => Ok(()),
            Some(_) => Err(MembershipRefusal::ResumePending),
        }
    }

    pub(super) fn ticket(&mut self) -> Result<u64, MembershipRefusal> {
        let ticket = self.next_ticket;
        self.next_ticket = ticket
            .checked_add(1)
            .ok_or(MembershipRefusal::CapacityExceeded { limit: usize::MAX })?;
        Ok(ticket)
    }

    /// The retained verdict for `key`, binding `pid` to it.
    pub(super) fn known(&mut self, key: &DecisionKey, pid: u64) -> Option<Indexed> {
        let verdict = self.entries.get_mut(key).map(|entry| {
            entry.connections.insert(pid);
            entry.verdict.clone()
        })?;
        self.by_connection
            .entry(pid)
            .or_default()
            .insert(key.clone());
        Some(verdict)
    }

    /// Reserves `key` for one ask, refusing at `limit` before allocating.
    pub(super) fn reserve(
        &mut self,
        key: &DecisionKey,
        mapped: &MappedChannel,
        pid: u64,
        limit: usize,
    ) -> Result<u64, MembershipRefusal> {
        if self.entries.len() >= limit {
            return Err(MembershipRefusal::CapacityExceeded { limit });
        }
        let ticket = self.ticket()?;
        let reached_by = [
            resource_key(&mapped.resource),
            resource_key(&mapped.workspace),
        ];
        for reach in &reached_by {
            self.by_resource
                .entry(reach.clone())
                .or_default()
                .insert(key.clone());
        }
        self.by_connection
            .entry(pid)
            .or_default()
            .insert(key.clone());
        self.entries.insert(
            key.clone(),
            Entry {
                verdict: Indexed::Applying { ticket },
                connections: BTreeSet::from([pid]),
                reached_by,
            },
        );
        Ok(ticket)
    }

    /// Re-reserves an existing binding for a change or resume ask.
    pub(super) fn reapply(&mut self, key: &DecisionKey) -> Result<Option<u64>, MembershipRefusal> {
        if !self.entries.contains_key(key) {
            return Ok(None);
        }
        let ticket = self.ticket()?;
        if let Some(entry) = self.entries.get_mut(key) {
            entry.verdict = Indexed::Applying { ticket };
        }
        Ok(Some(ticket))
    }

    /// Settles the ask holding `ticket`; `false` when a later change or
    /// resume overtook it.
    pub(super) fn settle(&mut self, key: &DecisionKey, ticket: u64, verdict: Indexed) -> bool {
        match self.entries.get_mut(key) {
            Some(entry) if entry.verdict == (Indexed::Applying { ticket }) => {
                entry.verdict = verdict;
                true
            }
            Some(_) | None => false,
        }
    }

    /// Forgets the binding still held by `ticket`'s ask.
    pub(super) fn forget(&mut self, key: &DecisionKey, ticket: u64) {
        let held = self
            .entries
            .get(key)
            .is_some_and(|entry| entry.verdict == (Indexed::Applying { ticket }));
        if held {
            self.remove(key);
        }
    }

    pub(super) fn remove(&mut self, key: &DecisionKey) {
        let Some(entry) = self.entries.remove(key) else {
            return;
        };
        for reach in &entry.reached_by {
            let emptied = self.by_resource.get_mut(reach).is_some_and(|keys| {
                keys.remove(key);
                keys.is_empty()
            });
            if emptied {
                self.by_resource.remove(reach);
            }
        }
        for pid in &entry.connections {
            let emptied = self.by_connection.get_mut(pid).is_some_and(|keys| {
                keys.remove(key);
                keys.is_empty()
            });
            if emptied {
                self.by_connection.remove(pid);
            }
        }
    }

    /// Releases every binding `pid` held; a binding no connection holds is
    /// removed. Visits only `pid`'s own bindings.
    pub(super) fn release(&mut self, pid: u64) -> u64 {
        let Some(keys) = self.by_connection.remove(&pid) else {
            return 0;
        };
        let mut released = 0_u64;
        for key in keys {
            let unheld = self.entries.get_mut(&key).is_some_and(|entry| {
                entry.connections.remove(&pid);
                entry.connections.is_empty()
            });
            if unheld {
                self.remove(&key);
                released = released.saturating_add(1);
            }
        }
        released
    }
}

/// The question a page continuation is bound to.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum PageQuestion {
    Resources {
        workspace: String,
        subject: MembershipSubject,
        action: MembershipAction,
    },
    Recipients {
        channel: String,
        action: MembershipAction,
    },
}

/// An opaque page continuation, bound to its exact question, grant log and
/// generation; it grants nothing (ACCESS-006 R4).
#[derive(Clone, PartialEq, Eq)]
pub struct PageCursor {
    pub(super) token: String,
    pub(super) question: PageQuestion,
    pub(super) generation: u64,
}

impl std::fmt::Debug for PageCursor {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("PageCursor")
            .field("token", &crate::auth_pass::ByteLength(self.token.len()))
            .field("generation", &self.generation)
            .finish_non_exhaustive()
    }
}

/// One page of the protected channels a holder may act on in a workspace.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ChannelPage {
    /// The liminal channels, in the provider's stable order.
    pub channels: Vec<String>,
    /// Rows the provider returned.
    pub returned: u64,
    /// Candidates the provider decided and left out.
    pub skipped: u64,
    /// Rows held for approval: a membership exists, the act still waits.
    pub held: u64,
    /// Rows naming a resource the product mapping does not list.
    pub unmapped: u64,
    /// The rows' encoded bytes.
    pub bytes: u64,
    /// The revision every row was decided at.
    pub revision: u64,
    /// Whether no row remains.
    pub complete: bool,
    /// The continuation, absent when complete.
    pub next: Option<PageCursor>,
}

/// One page of the subjects who may act on a protected channel.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RecipientPage {
    /// The subjects, in the provider's stable order.
    pub recipients: Vec<Subject>,
    /// Rows the provider returned.
    pub returned: u64,
    /// Candidates the provider decided and left out.
    pub skipped: u64,
    /// Rows held for approval.
    pub held: u64,
    /// The rows' encoded bytes.
    pub bytes: u64,
    /// The revision every row was decided at.
    pub revision: u64,
    /// Whether no row remains.
    pub complete: bool,
    /// The continuation, absent when complete.
    pub next: Option<PageCursor>,
}

/// A validated page's parts.
pub(super) struct PageParts<'p> {
    pub(super) rows: &'p [PageRow],
    pub(super) skipped: u64,
    pub(super) complete: bool,
    pub(super) next: Option<String>,
    pub(super) revision: u64,
    pub(super) bytes: u64,
}