liminal-rs 0.16.1

use crate::durability::settle::append_settled;
use std::fmt;
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};

use super::head_checkpoint::READ_BATCH_SIZE;
use super::{DurabilityError, DurableStore, ProcessingReceipt, StoredEntry};
use codec::DedupRecord;
pub use head::{DedupHeadCounts, dedup_head_counts};
use in_flight::InFlightIndex;
pub use sweep::{DedupSweepReport, DedupSweeper};

#[cfg(test)]
mod ceiling_in_flight;
mod codec;
mod head;
#[cfg(test)]
mod head_tests;
mod in_flight;
#[cfg(test)]
mod in_flight_tests;
#[cfg(test)]
mod lookup_tests;
#[cfg(test)]
mod start_count_tests;
mod sweep;
#[cfg(test)]
mod tests;
mod unresolved;

/// Hashes a producer idempotency key into a deterministic stream-key suffix.
#[must_use]
pub fn key_hash(idempotency_key: &str) -> String {
    let mut hash = 0xcbf2_9ce4_8422_2325_u64;
    for byte in idempotency_key.as_bytes() {
        hash ^= u64::from(*byte);
        hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
    }
    format!("{hash:016x}")
}

/// Where an in-flight claim's row would be, recorded with the claim.
///
/// The claim is written before its publish appends, so the row, if it exists,
/// sits at or after the position each partition of the channel stood at when
/// the claim was made: a lookup starts there and never reads the history
/// before it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ClaimEvidence {
    channel: String,
    positions: Vec<u64>,
}

impl ClaimEvidence {
    /// Evidence naming the durable `channel` and, per partition, the sequence
    /// the channel stood at when the claim was made.
    #[must_use]
    pub fn new(channel: impl Into<String>, positions: Vec<u64>) -> Self {
        Self {
            channel: channel.into(),
            positions,
        }
    }

    /// Durable channel whose log can establish the claim's outcome.
    #[must_use]
    pub fn channel(&self) -> &str {
        &self.channel
    }

    /// Per partition, the first sequence the claim's row can sit at.
    #[must_use]
    pub fn positions(&self) -> &[u64] {
        &self.positions
    }
}

/// One claim in flight at a start: its key, when it was claimed, and where
/// its row would be when the claim recorded that.
///
/// `claimed_at_millis` is the epoch millisecond the claim's entry carries,
/// written at the claim; a start reads it from the same bounded page as the
/// evidence. A claim with no evidence was written by a binary before the
/// evidence lane (`liminal-server` 0.16.0 and earlier) and nothing can settle
/// it by looking its row up (DUR-022).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct InFlightClaim {
    /// The producer idempotency key the claim holds.
    pub idempotency_key: String,
    /// Epoch milliseconds the claim was written at.
    pub claimed_at_millis: u64,
    /// Where the claim's row would be, when the claim recorded it.
    pub evidence: Option<ClaimEvidence>,
}

impl InFlightClaim {
    fn of(idempotency_key: String, entry: &DedupEntry) -> Self {
        Self {
            idempotency_key,
            claimed_at_millis: entry.timestamp_millis(),
            evidence: entry.evidence().cloned(),
        }
    }
}

/// Persisted dedup cache entry for one producer idempotency key.
#[derive(Clone, PartialEq, Eq)]
pub struct DedupEntry {
    idempotency_key: String,
    receipt: Option<Vec<u8>>,
    timestamp_millis: u64,
    evidence: Option<ClaimEvidence>,
}

impl DedupEntry {
    /// Builds an entry containing the original key, optional receipt bytes, and epoch millis.
    #[must_use]
    pub fn new(
        idempotency_key: impl Into<String>,
        receipt: Option<Vec<u8>>,
        timestamp_millis: u64,
    ) -> Self {
        Self {
            idempotency_key: idempotency_key.into(),
            receipt,
            timestamp_millis,
            evidence: None,
        }
    }

    /// Returns the original idempotency key stored with this entry.
    #[must_use]
    pub fn idempotency_key(&self) -> &str {
        &self.idempotency_key
    }

    /// Durable channel whose log can establish this in-flight claim's outcome.
    #[must_use]
    pub fn evidence_channel(&self) -> Option<&str> {
        self.evidence.as_ref().map(ClaimEvidence::channel)
    }

    /// Where this in-flight claim's row would be, when the claim recorded it.
    #[must_use]
    pub const fn evidence(&self) -> Option<&ClaimEvidence> {
        self.evidence.as_ref()
    }

    /// Returns the stored opaque receipt bytes, when processing has completed.
    #[must_use]
    pub fn receipt(&self) -> Option<&[u8]> {
        self.receipt.as_deref()
    }

    /// Returns the entry timestamp in epoch milliseconds.
    #[must_use]
    pub const fn timestamp_millis(&self) -> u64 {
        self.timestamp_millis
    }

    /// Serializes this active cache entry into deterministic storage bytes.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError::EnvelopeError`] when a field length cannot be encoded.
    pub fn serialize(&self) -> Result<Vec<u8>, DurabilityError> {
        DedupRecord::Active(self.clone()).serialize()
    }

    /// Deserializes an active cache entry previously produced by [`Self::serialize`].
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError::EnvelopeError`] when bytes are malformed or contain a tombstone.
    pub fn deserialize(bytes: &[u8]) -> Result<Self, DurabilityError> {
        match DedupRecord::deserialize(bytes)? {
            DedupRecord::Active(entry) => Ok(entry),
            DedupRecord::Tombstone { .. } => Err(DurabilityError::EnvelopeError(
                "dedup tombstone is not an active entry".to_owned(),
            )),
        }
    }
}

impl fmt::Debug for DedupEntry {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("DedupEntry")
            .field("idempotency_key", &self.idempotency_key)
            .field("receipt_bytes", &self.receipt.as_ref().map(Vec::len))
            .field("timestamp_millis", &self.timestamp_millis)
            .field("evidence", &self.evidence)
            .finish()
    }
}

/// Result of checking or claiming an idempotency key.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum DedupDecision {
    /// The caller claimed the key and may deliver the message for processing.
    Claimed,
    /// The key was already completed; return this receipt without re-delivery.
    Completed(ProcessingReceipt),
    /// The key is already being processed and delivery must be deferred.
    InFlight,
}

/// Haematite-backed idempotency-key cache for the lightweight dedup contract.
///
/// Clones share one in-flight index (`dedup/in_flight.rs`). The index assumes
/// it is the only one kept for its namespace in the process, as the store
/// assumes one writing process: a second cache built over the same namespace
/// is a second process's view, as a restart would be.
#[derive(Clone)]
pub struct DedupCache {
    store: Arc<dyn DurableStore>,
    namespace: String,
    in_flight: Arc<InFlightIndex>,
}

impl DedupCache {
    /// Creates a dedup cache over the given durable store and namespace prefix.
    #[must_use]
    pub fn new(store: Arc<dyn DurableStore>, namespace: impl Into<String>) -> Self {
        Self {
            store,
            namespace: namespace.into(),
            in_flight: Arc::new(InFlightIndex::default()),
        }
    }

    /// Creates a dedup cache over a namespace nothing has written to: one in
    /// a store this process created, or one no earlier process used.
    ///
    /// Its in-flight index starts empty without reading the store, so
    /// [`Self::in_flight_claims`] reads nothing and no claim walks the
    /// namespace. [`Self::record_empty_start`], or failing that the first
    /// claim, records the index start before anything is written, so a later
    /// process opening the namespace with [`Self::new`] reads the index and
    /// never walks it. A namespace that may hold claims is opened with
    /// [`Self::new`].
    #[must_use]
    pub fn new_empty(store: Arc<dyn DurableStore>, namespace: impl Into<String>) -> Self {
        Self {
            store,
            namespace: namespace.into(),
            in_flight: Arc::new(InFlightIndex::empty()),
        }
    }

    /// Returns the configured dedup namespace.
    #[must_use]
    pub fn namespace(&self) -> &str {
        &self.namespace
    }

    /// Formats the haematite stream key for an idempotency key.
    #[must_use]
    pub fn stream_key_for(&self, idempotency_key: &str) -> String {
        format!("{}:{}", self.namespace, key_hash(idempotency_key))
    }

    /// Looks up an existing key without appending a record. A stream of
    /// fewer than one page of records is read in one page, as before DUR-008
    /// A12; a longer one is read from its verified recorded head, and one
    /// whose head was never recorded is walked once and its head recorded.
    ///
    /// # Errors
    ///
    /// Propagates store read errors and returns [`DurabilityError::DedupCollision`] when
    /// the hashed stream contains a different original idempotency key.
    pub async fn lookup(
        &self,
        idempotency_key: &str,
    ) -> Result<Option<DedupDecision>, DurabilityError> {
        let stream_key = self.stream_key_for(idempotency_key);
        let snapshot = self.load_snapshot(&stream_key, idempotency_key).await?;
        Ok(snapshot.current.as_ref().map(decision_for_entry))
    }

    /// Claims a new key or returns the existing completed/in-flight decision.
    ///
    /// Callers must only deliver the message when this returns [`DedupDecision::Claimed`].
    ///
    /// # Errors
    ///
    /// Propagates store errors and serialization errors. A concurrent first claim that wins
    /// the append race is converted into the duplicate completed/in-flight decision.
    pub async fn claim_or_get(
        &self,
        idempotency_key: &str,
        timestamp_millis: u64,
    ) -> Result<DedupDecision, DurabilityError> {
        let stream_key = self.stream_key_for(idempotency_key);
        let snapshot = self.load_snapshot(&stream_key, idempotency_key).await?;
        if let Some(entry) = snapshot.current.as_ref() {
            return Ok(decision_for_entry(entry));
        }

        let entry = DedupEntry::new(idempotency_key, None, timestamp_millis);
        self.record_empty_start().await?;
        match append_settled(
            self.store.as_ref(),
            &stream_key,
            entry.serialize()?,
            snapshot.next_seq,
        )
        .await
        {
            Ok(_) => Ok(DedupDecision::Claimed),
            Err(DurabilityError::SequenceConflict { expected, actual }) => {
                self.decision_after_conflict(&stream_key, idempotency_key, expected, actual)
                    .await
            }
            Err(error) => Err(error),
        }
    }

    /// Stores a processing receipt by appending a completed entry for an existing key.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError::ConfigError`] when the current system time cannot be encoded.
    /// Returns [`DurabilityError::DedupCollision`] when the key is missing or collides with
    /// another original key, and propagates serialization/store append errors.
    pub async fn complete_receipt(
        &self,
        idempotency_key: &str,
        receipt: ProcessingReceipt,
    ) -> Result<(), DurabilityError> {
        self.complete_receipt_at(idempotency_key, receipt, current_epoch_millis()?)
            .await
    }

    /// Stores a processing receipt with an explicit receipt timestamp.
    ///
    /// This is useful for deterministic tests and for callers that already have a trusted
    /// processing-completion timestamp. The timestamp is the TTL anchor for the stored receipt.
    ///
    /// # Errors
    ///
    /// Returns [`DurabilityError::DedupCollision`] when the key is missing, collides with
    /// another original key, or has already completed with different receipt bytes. Propagates
    /// serialization/store append errors.
    async fn complete_receipt_at(
        &self,
        idempotency_key: &str,
        receipt: ProcessingReceipt,
        timestamp_millis: u64,
    ) -> Result<(), DurabilityError> {
        let stream_key = self.stream_key_for(idempotency_key);
        let snapshot = self.load_snapshot(&stream_key, idempotency_key).await?;
        let Some(entry) = snapshot.current.as_ref() else {
            return Err(DurabilityError::DedupCollision {
                key: idempotency_key.to_owned(),
            });
        };

        let receipt_bytes = receipt.into_bytes();
        if let Some(existing_receipt) = entry.receipt() {
            if existing_receipt == receipt_bytes.as_slice() {
                self.leave_in_flight(idempotency_key, snapshot.claim_sequence()?)
                    .await;
                return Ok(());
            }
            return Err(DurabilityError::DedupCollision {
                key: idempotency_key.to_owned(),
            });
        }

        let completed = DedupEntry::new(
            entry.idempotency_key().to_owned(),
            Some(receipt_bytes.clone()),
            timestamp_millis,
        );
        match append_settled(
            self.store.as_ref(),
            &stream_key,
            completed.serialize()?,
            snapshot.next_seq,
        )
        .await
        {
            Ok(_) => {}
            Err(DurabilityError::SequenceConflict { expected, actual }) => {
                self.confirm_matching_receipt(
                    &stream_key,
                    idempotency_key,
                    &receipt_bytes,
                    expected,
                    actual,
                )
                .await?;
            }
            Err(error) => return Err(error),
        }
        self.leave_in_flight(idempotency_key, snapshot.claim_sequence()?)
            .await;
        Ok(())
    }

    /// Releases an in-flight claim so the key becomes re-claimable.
    ///
    /// Callers use this on the publish failure path: a key was claimed
    /// ([`DedupDecision::Claimed`]) but the downstream delivery failed before a
    /// receipt could be recorded, leaving the key dangling [`DedupDecision::InFlight`]
    /// forever and permanently suppressing every re-publish. Releasing appends a
    /// tombstone so the next claim succeeds.
    ///
    /// This NEVER clobbers a stored receipt: an already-completed key (receipt
    /// present) and an absent key are both no-ops, preserving the at-most-once
    /// guarantee. Only a current in-flight entry (active, no receipt) is tombstoned.
    ///
    /// # Errors
    ///
    /// Propagates store read/append and serialization errors. A [`DurabilityError::SequenceConflict`]
    /// on the tombstone append is re-checked: if the latest state is now completed
    /// or already a tombstone the release is treated as successful, otherwise the
    /// conflict is propagated.
    pub async fn release_claim(&self, idempotency_key: &str) -> Result<(), DurabilityError> {
        self.release_claim_at(idempotency_key, current_epoch_millis()?)
            .await
    }

    async fn release_claim_at(
        &self,
        idempotency_key: &str,
        timestamp_millis: u64,
    ) -> Result<(), DurabilityError> {
        let stream_key = self.stream_key_for(idempotency_key);
        let snapshot = self.load_snapshot(&stream_key, idempotency_key).await?;
        // No-op when there is nothing in flight: absent key, or a completed key
        // whose receipt must never be clobbered (guards at-most-once).
        let Some(entry) = snapshot.current.as_ref() else {
            return Ok(());
        };
        if entry.receipt().is_some() {
            return Ok(());
        }

        let tombstone = DedupRecord::tombstone(idempotency_key.to_owned(), timestamp_millis);
        match append_settled(
            self.store.as_ref(),
            &stream_key,
            tombstone.serialize()?,
            snapshot.next_seq,
        )
        .await
        {
            Ok(_) => {}
            Err(DurabilityError::SequenceConflict { expected, actual }) => {
                self.confirm_release_after_conflict(&stream_key, idempotency_key, expected, actual)
                    .await?;
            }
            Err(error) => return Err(error),
        }
        self.leave_in_flight(idempotency_key, snapshot.claim_sequence()?)
            .await;
        Ok(())
    }

    async fn confirm_release_after_conflict(
        &self,
        stream_key: &str,
        idempotency_key: &str,
        expected: u64,
        actual: u64,
    ) -> Result<(), DurabilityError> {
        // A concurrent writer advanced the stream after our snapshot. Re-load and
        // re-check the latest record directly (not via `into_active`, so a fresh
        // tombstone is distinguishable): if it is now completed (receipt present)
        // or already a tombstone, the in-flight claim is gone and the release goal
        // is met. A still-active no-receipt entry means a legitimate re-claim won
        // the race, so we must not clobber it -- propagate the conflict.
        let latest = self.latest_record(stream_key, idempotency_key).await?;
        match latest {
            Some(DedupRecord::Tombstone { .. }) => Ok(()),
            Some(DedupRecord::Active(entry)) if entry.receipt().is_some() => Ok(()),
            _ => Err(DurabilityError::SequenceConflict { expected, actual }),
        }
    }

    async fn latest_record(
        &self,
        stream_key: &str,
        idempotency_key: &str,
    ) -> Result<Option<DedupRecord>, DurabilityError> {
        Ok(self.latest(stream_key, idempotency_key).await?.0)
    }

    /// The key stream's latest record and its next append sequence.
    ///
    /// A lookup reads the stream's first page. A page short of
    /// [`READ_BATCH_SIZE`] rows is the whole stream, so a short stream, the
    /// common case, costs one `read_from` and nothing else, as it did before
    /// DUR-008 A12, and every record's key is checked. Only a stream that
    /// fills its first page is read past it from its verified recorded head
    /// (`dedup/head.rs`), at a cost bounded by that page plus the rows past
    /// the head, never the key's whole history. No append records a head.
    async fn latest(
        &self,
        stream_key: &str,
        idempotency_key: &str,
    ) -> Result<(Option<DedupRecord>, u64), DurabilityError> {
        // ast-grep-allow: no-read-from-start: one bounded page is the whole of a short key stream (the J5 path, no head read); a stream that fills it is read past it from its verified head
        let page = self.store.read_from(stream_key, 0, READ_BATCH_SIZE).await?;
        if page.len() < READ_BATCH_SIZE {
            let latest = fold_page(stream_key, idempotency_key, &page, 0)?;
            return Ok((latest, len_to_u64(page.len())?));
        }
        self.latest_past_first_page(stream_key, idempotency_key, &page)
            .await
    }

    fn scan_prefix(&self) -> String {
        format!("{}:", self.namespace)
    }

    async fn decision_after_conflict(
        &self,
        stream_key: &str,
        idempotency_key: &str,
        expected: u64,
        actual: u64,
    ) -> Result<DedupDecision, DurabilityError> {
        let snapshot = self.load_snapshot(stream_key, idempotency_key).await?;
        snapshot.current.as_ref().map_or(
            Err(DurabilityError::SequenceConflict { expected, actual }),
            |entry| Ok(decision_for_entry(entry)),
        )
    }

    async fn confirm_matching_receipt(
        &self,
        stream_key: &str,
        idempotency_key: &str,
        receipt_bytes: &[u8],
        expected: u64,
        actual: u64,
    ) -> Result<(), DurabilityError> {
        let snapshot = self.load_snapshot(stream_key, idempotency_key).await?;
        if snapshot
            .current
            .as_ref()
            .and_then(DedupEntry::receipt)
            .is_some_and(|bytes| bytes == receipt_bytes)
        {
            Ok(())
        } else {
            Err(DurabilityError::SequenceConflict { expected, actual })
        }
    }

    async fn load_snapshot(
        &self,
        stream_key: &str,
        idempotency_key: &str,
    ) -> Result<StreamSnapshot, DurabilityError> {
        let (latest, next_seq) = self.latest(stream_key, idempotency_key).await?;
        Ok(StreamSnapshot {
            current: latest.and_then(DedupRecord::into_active),
            next_seq,
        })
    }
}

impl fmt::Debug for DedupCache {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("DedupCache")
            .field("namespace", &self.namespace)
            .field("store", &self.store)
            .field("in_flight", &self.in_flight)
            .finish()
    }
}

struct StreamSnapshot {
    current: Option<DedupEntry>,
    next_seq: u64,
}

impl StreamSnapshot {
    /// The sequence of the stream's latest record, the claim a completion or
    /// release written after this snapshot settles.
    fn claim_sequence(&self) -> Result<u64, DurabilityError> {
        self.next_seq.checked_sub(1).ok_or_else(|| {
            DurabilityError::ConfigError("dedup stream has no record to settle".to_owned())
        })
    }
}

fn decision_for_entry(entry: &DedupEntry) -> DedupDecision {
    entry.receipt().map_or(DedupDecision::InFlight, |bytes| {
        DedupDecision::Completed(ProcessingReceipt::new(bytes.to_vec()))
    })
}

/// Decodes a key stream's record, refusing one written for another original
/// key.
fn checked_record(payload: &[u8], idempotency_key: &str) -> Result<DedupRecord, DurabilityError> {
    let record = DedupRecord::deserialize(payload)?;
    if record.idempotency_key() != idempotency_key {
        return Err(DurabilityError::DedupCollision {
            key: idempotency_key.to_owned(),
        });
    }
    Ok(record)
}

/// Checks one page of a key stream read from `offset`: each row sits at its
/// own position and holds a record of `idempotency_key`. Answers the page's
/// last record.
fn fold_page(
    stream_key: &str,
    idempotency_key: &str,
    page: &[StoredEntry],
    offset: u64,
) -> Result<Option<DedupRecord>, DurabilityError> {
    let mut latest = None;
    let mut expected = offset;
    for stored in page {
        if stored.sequence != expected {
            return Err(DurabilityError::ConfigError(format!(
                "dedup stream {stream_key} answered sequence {} where {expected} was read",
                stored.sequence
            )));
        }
        latest = Some(checked_record(&stored.payload, idempotency_key)?);
        expected = expected.checked_add(1).ok_or_else(|| {
            DurabilityError::ConfigError(format!("dedup stream {stream_key} sequence overflow"))
        })?;
    }
    Ok(latest)
}

fn len_to_u64(len: usize) -> Result<u64, DurabilityError> {
    u64::try_from(len).map_err(|error| {
        DurabilityError::ConfigError(format!("dedup entry count cannot fit u64: {error}"))
    })
}

fn current_epoch_millis() -> Result<u64, DurabilityError> {
    let duration = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_err(|error| {
            DurabilityError::ConfigError(format!("system clock is before Unix epoch: {error}"))
        })?;
    u64::try_from(duration.as_millis()).map_err(|error| {
        DurabilityError::ConfigError(format!(
            "current epoch millis cannot fit u64 for dedup receipt: {error}"
        ))
    })
}