alkblobs 0.2.2

//! The facade put seam (the write path's paths): the derivation-explicit
//! one seam — the known-length one-pass path, the unknown-length
//! pre-threshold buffering path and its mid-stream restart derivation,
//! the `Batch` batching seam with its algorithm parameter, and the
//! kv-only rejection point — behind the one public `put_as`/
//! `put_in_batch` signature (ADR-020 §Decision 5). The non-fleet paths
//! are the ADR-005 in-process protocol verbatim; the fleet branches
//! route to the fleet file's drivers.

use super::engines::AnyStaging;
#[cfg(all(feature = "postgres", feature = "pg-lo"))]
use super::engines::LargeEngine;
use super::Store;
use super::StoreMode::KvOnly;
use crate::backend::{ReadCursor, StagingCursor};
use crate::dispatch::Tier;
use crate::facade::{Batch, Pin};
use crate::hashing::{Algorithm, BlobHasher};
use crate::keys::Key;
use crate::Error;
use std::sync::Arc;
use tokio::io::{AsyncRead, AsyncReadExt};

/// The write path's streaming chunk granularity — every put loop reads
/// the stream in windows of this size. Sized inside the kv tier's
/// economics regime (bounded memory per iteration; engine bodies
/// aggregate upstream — the pg-lo engine's measured 512 KiB
/// `lowrite` knee, [`crate::pg_lo::LOWRITE_CHUNK`], aggregates these
/// windows into its C2 statement shape; that engine's constant is the
/// authority there, landed with its wave).
pub(crate) const PUT_CHUNK: usize = 64 * 1024;

/// The write path (store-api.md §Write path): `put_as` — the two
/// internal paths behind one signature, the `Batch` batching seam, and
/// the kv-only rejection point (ADR-012 §6.2), on the pin-before-publish
/// ordering (ADR-005). The named design points (stream shape, len
/// mismatch, kv-only carrier, key-at-commit amendment) are recorded in
/// the module doc; their behaviors are pinned by the tests below.
impl Store {
    /// **`put_as(algorithm, len, stream) → (Key, Pin)`** — the one
    /// derivation-explicit put seam (store-api.md §Write path;
    /// ADR-020 §Decision 5): the caller names the derivation. The
    /// same bytes have multiple equally valid addresses, one per
    /// registry member, so the crate never picks a derivation for
    /// its caller — the caller that cannot name an algorithm cannot
    /// form an address (the ADR-020 §Decision 1 invariant). Every
    /// registered algorithm writes through this one signature: no
    /// `matches!` restriction, no canonical default, no family-named
    /// sibling method. Two arms behind one signature:
    ///
    /// **Known-length (`Some(len)`) — the encouraged path** (covers
    /// git objects, `stat`-derived files, `Content-Length`-bearing
    /// uploads, network offers): one pass, no staging needed for
    /// hashing — the preamble (`"blob <len>\0"`) initializes the
    /// derivation before content flows. Preamble-carrying members
    /// (the blob family) hash preamble-then-content; identity-family
    /// members ([`Algorithm::GitObjectSha1`]/[`Algorithm::GitObjectSha256`],
    /// ADR-018) take the preimage verbatim — git's loose-object form,
    /// the caller provides `"<type> <len>\0" + content` — with `len`
    /// as the stored-length check; the derivation is the plain hash
    /// over the preimage, so the resulting key's digest is the
    /// object's git oid, byte-identical, every kind included.
    /// Verification-on-get re-derives with the key's algorithm
    /// byte-identically. ≤ the threshold → the kv tier's whole-value
    /// put; > the threshold → the large tier's staged put (content
    /// streams into the staging cursor while hashing; the derivation
    /// completes at stream end; commit-rename publishes). A stream
    /// that yields fewer or more bytes than the declared `len`
    /// rejects with `Error::Verification` — zero residue either side
    /// (the module doc's named design point).
    ///
    /// **Unknown-length (`None`)** — retained on the facade (ADR-020
    /// §Decision 3: "stream in, get the key back" is a legitimate
    /// in-process posture for edge producers, and the hard part is
    /// the crash-safe staging hygiene every outer-layer workaround
    /// re-implements): content buffers in memory up to the dispatch
    /// threshold, ADR-003's option (d) semantics pinned in the three
    /// numbered rules:
    ///
    /// 1. **Mid-stream threshold overflow** (non-restartable sources
    ///    are normal): the buffered prefix flushes *into a fresh stage
    ///    file* and the stream continues appending into it — there is
    ///    no "restart the stream" requirement anywhere.
    /// 2. **Verification runs over the eventual stored artifact as a
    ///    whole**: the memory buffer for the sub-threshold kv commit;
    ///    the complete staged bytes for the large commit — the
    ///    derivation pass restarts over preamble + staged bytes for
    ///    preamble-carrying members (the preamble is inside the
    ///    hashed input, ADR-002); identity-family members restart over
    ///    the staged bytes — the permitted simple implementation,
    ///    single-pass capable if ever measured worth it. Reading the
    ///    staged artifact back goes through the staging cursor's
    ///    `ReadCursor` seam.
    /// 3. **Stream error before commit**: no entry lands and no stage
    ///    file remains — buffer discard on the failure path, exactly
    ///    like stage discard (the stage-hygiene invariant extended to
    ///    the buffered case).
    ///
    /// Error mapping is uniform across both arms and all algorithms:
    /// length mismatch `Error::Verification` (zero residue), media
    /// failure `Error::Io`.
    ///
    /// **Pin-before-publish (ADR-005):** the pin refcount is minted
    /// under the arbitration lock *before* the entry becomes visible —
    /// an entry cannot appear in a sweep's `list()` without already
    /// carrying its pin. A dedup re-put acquires no new pool entry; the
    /// returned `Pin` refcounts the existing one under the same lock.
    ///
    /// **kv-only mode (ADR-012 §6.2):** a known-length put over the
    /// threshold rejects immediately; an unknown-length put rejects at
    /// mid-stream threshold overflow — lengths are not known at
    /// construction, so put time is the only possible rejection point.
    /// Zero residue on either rejection.
    ///
    /// **Fleet puts (ADR-012 §3/§4; ADR-018 §Decision 7):** under a
    /// `fleet: true` declaration the same signature routes the fleet
    /// forms — the kv tier through the joint entry+pin tx (the
    /// joint-tx driver's named design points, the termination design
    /// included, are recorded on this module's fleet impl block), the
    /// pg-lo tier through the one-tx consolidation form, the local tier
    /// pin-first/publish-second; the named derivation threads through
    /// them (derivation-parameterized sites, not family-branching
    /// copies), and the returned guard carries the pin row's TTL/3
    /// renewal duty. Under `fleet: false` (every pre-existing
    /// constructor) this doc applies verbatim and nothing of the fleet
    /// machinery runs.
    ///
    /// **`stat` semantic (ADR-018 §Decision 3):** for a family entry,
    /// `stat`'s `EntryMeta.len` is the length of the stored value —
    /// the loose-object form, git's header included — not the content
    /// length. Deliberate and per-family; see store-api.md §Read path.
    pub async fn put_as(
        &self,
        algorithm: Algorithm,
        len: Option<u64>,
        stream: impl AsyncRead + Unpin + Send,
    ) -> Result<(Key, Pin), Error> {
        match len {
            Some(len) => {
                if self.mode == KvOnly && len > self.threshold {
                    return Err(Error::Io(format!(
                        "kv-only mode rejects an over-threshold put: declared len {len} > threshold {}",
                        self.threshold
                    )));
                }
                self.put_known_as(algorithm, len, stream).await
            }
            None => self.put_unknown(algorithm, stream).await,
        }
    }

    /// **The batch-scoped pin group** (store-api.md §Write path; ADR-005
    /// §Marks 2). Multi-put writes — manifest writes are exactly this —
    /// group under one batch-scoped pin: [`Store::put_in_batch`] runs
    /// each member's lone-put flow and contributes its pin to the
    /// group; the group's pins drop together on [`Batch::commit`] or
    /// drop (no early single-pin release; member liveness never
    /// outlives the batch — batch drop is the failure path).
    pub fn batch(&self) -> Batch {
        self.gc().new_batch()
    }

    /// A member put inside a [`Batch`] — the batching put seam
    /// (store-api.md §Write path). Member puts publish pin-before-
    /// publish exactly as lone puts do: the two paths behind
    /// [`Store::put_as`]'s one signature run, and the minted guard is
    /// contributed to the group instead of returned; the algorithm is
    /// named like every other write (ADR-020 §Decision 5). The batch's
    /// pins drop together on [`Batch::commit`] or batch drop.
    ///
    /// **Foreign-Batch refusal (review-001 N-5):** the batch must have
    /// been minted by this store — the seam opens with a
    /// `debug_assert!(Arc::ptr_eq)` on the batch's registry Arc against
    /// the store's, before any pin is minted. The debug-assert point is
    /// the recorded choice over a named error: the mis-use is
    /// developer-error territory, and the error family is ADR-pinned
    /// (ADR-012 §6.5 — new variants are new ADRs). Release builds run
    /// on: a foreign batch's members would carry their pins in the
    /// other store's in-process registry, where the group release is a
    /// silently no-op — the leak the assert's window exists to catch.
    pub async fn put_in_batch(
        &self,
        batch: &mut Batch,
        algorithm: Algorithm,
        len: Option<u64>,
        stream: impl AsyncRead + Unpin + Send,
    ) -> Result<Key, Error> {
        debug_assert!(
            Arc::ptr_eq(batch.registry(), self.gc().core()),
            "put_in_batch: the Batch was minted by a foreign Store — its members' pins would release against the wrong registry (a silently no-op release there), leaking the group's protection"
        );
        let (key, pin) = self.put_as(algorithm, len, stream).await?;
        batch.add(pin);
        Ok(key)
    }

    /// The known-length internal path under a caller-chosen
    /// derivation (ADR-018 §Decision 6/§7; ADR-020 §Decision 5): the
    /// blob family routes the preamble hasher; the git-object family
    /// routes the plain hasher (no preamble — the loose-object header
    /// is inside the preimage), with `len` as the stored-length check.
    /// Everything after the hasher construction is pin-before-publish
    /// identical: the same tier routing, the same fleet forms (the kv
    /// joint entry+pin tx, the pg-lo consolidation tx, the pin-first/
    /// publish-second staged commit), the same declared-len
    /// `Verification` mapping, the same CAS dedup.
    async fn put_known_as<R: AsyncRead + Unpin + Send>(
        &self,
        algorithm: Algorithm,
        len: u64,
        mut stream: R,
    ) -> Result<(Key, Pin), Error> {
        let mut hasher = BlobHasher::new(algorithm, len);
        let mut chunk = vec![0u8; PUT_CHUNK];
        let mut received = 0u64;

        match self.tier_for(len) {
            Tier::Kv => {
                let mut value: Vec<u8> = Vec::new();
                loop {
                    let n = stream
                        .read(&mut chunk)
                        .await
                        .map_err(|e| Error::Io(e.to_string()))?;
                    if n == 0 {
                        break;
                    }
                    received = received.saturating_add(n as u64);
                    if received > len {
                        return Err(Error::Verification);
                    }
                    hasher.update(&chunk[..n]);
                    value.extend_from_slice(&chunk[..n]);
                }
                if received != len {
                    return Err(Error::Verification);
                }
                let key = Key::from(hasher.finish());
                if let Some(fleet) = self.fleet() {
                    return self.fleet_put_kv(fleet, key, &value).await;
                }
                let pin = self.gc().acquire(&key).await;
                self.kv().put(&key.as_bytes(), &value).await?;
                Ok((key, pin))
            }
            Tier::Large => {
                let engine = match self.large() {
                    Some(engine) => engine,
                    None => return Err(Error::Io("no large tier in this mode".to_owned())),
                };
                #[cfg(all(feature = "postgres", feature = "pg-lo"))]
                if let Some(fleet) = self.fleet() {
                    if let LargeEngine::PgLo(lo) = engine {
                        // The consolidation posture: the staged form *is*
                        // the joint form (one tx; ADR-012 §4).
                        return self
                            .fleet_put_pglo_known(fleet, lo, algorithm, len, stream)
                            .await;
                    }
                }
                let mut stage = engine.put_staged().await?;
                let fed: Result<(), Error> = 'stream: loop {
                    let n = match stream.read(&mut chunk).await {
                        Ok(n) => n,
                        Err(e) => break 'stream Err(Error::Io(e.to_string())),
                    };
                    if n == 0 {
                        break 'stream Ok(());
                    }
                    received = received.saturating_add(n as u64);
                    if received > len {
                        break 'stream Err(Error::Verification);
                    }
                    hasher.update(&chunk[..n]);
                    if let Err(e) = stage.append(&chunk[..n]).await {
                        break 'stream Err(e);
                    }
                };
                match fed {
                    Err(err) => {
                        stage.discard().await;
                        Err(err)
                    }
                    Ok(()) if received != len => {
                        stage.discard().await;
                        Err(Error::Verification)
                    }
                    Ok(()) => {
                        let key = Key::from(hasher.finish());
                        if let Some(fleet) = self.fleet() {
                            // The shared-media ordering (ADR-012 §4):
                            // pin row first, publish-second.
                            return self.fleet_staged_commit(fleet, key, stage).await;
                        }
                        let pin = self.gc().acquire(&key).await;
                        stage.commit(&key.as_bytes()).await?;
                        Ok((key, pin))
                    }
                }
            }
        }
    }

    /// The unknown-length internal path: buffer to the dispatch
    /// threshold; on mid-stream overflow the buffered prefix flushes
    /// into a fresh stage file and the stream continues appending
    /// (rule 1) — kv-only mode rejects at that same overflow
    /// (ADR-012 §6.2). At stream end, the derivation pass runs over the
    /// eventual stored artifact as a whole: the buffer (kv form) or the
    /// staged bytes read back through the cursor's `ReadCursor` seam
    /// (staged form, rule 2). Every failure path discards (rule 3).
    /// The derivation threads from the caller; the one caller is
    /// [`Store::put_as`]'s unknown-length arm — every registered
    /// algorithm routes here (ADR-020 §Decision 5's retained arm), so
    /// the parameter is honored at every branch, fleet arms included.
    async fn put_unknown<R: AsyncRead + Unpin + Send>(
        &self,
        algorithm: Algorithm,
        mut stream: R,
    ) -> Result<(Key, Pin), Error> {
        #[cfg(all(feature = "postgres", feature = "pg-lo"))]
        if let (Some(fleet), Some(LargeEngine::PgLo(lo))) = (self.fleet(), self.large()) {
            // The consolidation posture: the unknown-length staged form
            // *is* the joint form (one tx; ADR-012 §4).
            return self
                .fleet_put_pglo_unknown(fleet, lo, algorithm, stream)
                .await;
        }
        let mut buffered = Vec::with_capacity(PUT_CHUNK);
        let mut stage: Option<AnyStaging> = None;
        let mut streamed_total = 0u64;
        let mut chunk = vec![0u8; PUT_CHUNK];

        let fed: Result<(), Error> = 'feed: loop {
            let n = match stream.read(&mut chunk).await {
                Ok(n) => n,
                Err(e) => break 'feed Err(Error::Io(e.to_string())),
            };
            if n == 0 {
                break 'feed Ok(());
            }
            streamed_total = streamed_total.saturating_add(n as u64);

            if let Some(staged) = stage.as_mut() {
                let append = staged.append(&chunk[..n]).await;
                if let Err(e) = append {
                    if let Some(cursor) = stage.take() {
                        cursor.discard().await;
                    }
                    break 'feed Err(e);
                }
                continue;
            }

            let overflow = match self.threshold() {
                Some(t) => (buffered.len() as u64).saturating_add(n as u64) > t,
                None => false,
            };
            if !overflow {
                buffered.extend_from_slice(&chunk[..n]);
                continue;
            }
            match self.mode {
                KvOnly => {
                    let t = self.threshold;
                    break 'feed Err(Error::Io(format!(
                        "kv-only mode rejects an over-threshold put: streamed {streamed_total} bytes past the threshold {t}"
                    )));
                }
                _ => match self.large() {
                    Some(engine) => {
                        let mut staged = match engine.put_staged().await {
                            Ok(staged) => staged,
                            Err(e) => break 'feed Err(e),
                        };
                        if let Err(e) = staged.append(&buffered).await {
                            staged.discard().await;
                            break 'feed Err(e);
                        }
                        if let Err(e) = staged.append(&chunk[..n]).await {
                            staged.discard().await;
                            break 'feed Err(e);
                        }
                        stage = Some(staged);
                    }
                    None => {
                        break 'feed Err(Error::Io("no large tier in this mode".to_owned()));
                    }
                },
            }
        };

        match fed {
            Err(err) => {
                if let Some(cursor) = stage.take() {
                    cursor.discard().await;
                }
                Err(err)
            }
            Ok(()) => match stage.take() {
                Some(staged) => {
                    // Rule 2, staged form: the derivation pass restarts
                    // over preamble + staged bytes, the artifact read
                    // back through the cursor's ReadCursor seam.
                    let mut hasher = BlobHasher::new(algorithm, streamed_total);
                    let mut offset = 0u64;
                    let derived: Result<(), Error> = 'pass: loop {
                        if offset >= streamed_total {
                            break 'pass Ok(());
                        }
                        let want = (streamed_total - offset).min(PUT_CHUNK as u64);
                        let bytes = match staged.read_at(offset, want).await {
                            Ok(bytes) => bytes,
                            Err(e) => break 'pass Err(e),
                        };
                        if bytes.is_empty() {
                            break 'pass Err(Error::Io(
                                "staged artifact under-read during the derivation pass".to_owned(),
                            ));
                        }
                        hasher.update(&bytes);
                        offset += bytes.len() as u64;
                    };
                    match derived {
                        Err(err) => {
                            staged.discard().await;
                            Err(err)
                        }
                        Ok(()) => {
                            let key = Key::from(hasher.finish());
                            if let Some(fleet) = self.fleet() {
                                // The shared-media ordering (ADR-012
                                // §4): pin row first, publish-second.
                                return self.fleet_staged_commit(fleet, key, staged).await;
                            }
                            let pin = self.gc().acquire(&key).await;
                            staged.commit(&key.as_bytes()).await?;
                            Ok((key, pin))
                        }
                    }
                }
                None => {
                    // Rule 2, buffer form: one derivation pass over the
                    // complete buffered artifact — the kv commit's whole
                    // stored value.
                    if buffered.len() as u64 != streamed_total {
                        return Err(Error::Io(
                            "buffered artifact length disagrees with the streamed total".to_owned(),
                        ));
                    }
                    let mut hasher = BlobHasher::new(algorithm, streamed_total);
                    hasher.update(&buffered);
                    let key = Key::from(hasher.finish());
                    if let Some(fleet) = self.fleet() {
                        return self.fleet_put_kv(fleet, key, &buffered).await;
                    }
                    let pin = self.gc().acquire(&key).await;
                    self.kv().put(&key.as_bytes(), &buffered).await?;
                    Ok((key, pin))
                }
            },
        }
    }
}