alkblobs 0.1.0

//! The facade put seam (the write path's paths): the known-length
//! one-pass path, the unknown-length pre-threshold buffering path and
//! its mid-stream restart derivation, the `Batch` batching seam, and
//! the kv-only rejection point — behind the one public `put`/
//! `put_in_batch` signature. 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` — 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(len, stream) → (Key, Pin)`** — the streaming put seam
    /// (store-api.md §Write path; POC #3 finding A1): two internal
    /// paths behind one signature, routing a pure function of content
    /// length (ADR-003) — no content ever migrates tiers.
    ///
    /// **Known-length (`Some(len)`) — the encouraged path** (covers
    /// git objects, `stat`-derived files, `Content-Length`-bearing
    /// uploads, network offers): the preamble (`"blob <len>\0"`)
    /// initializes the derivation before content flows, so one pass
    /// suffices — no staging needed for hashing. ≤ 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`):** 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 (the
    ///    preamble is inside the hashed input, ADR-002), reading the
    ///    staged artifact back 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).
    ///
    /// **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):** 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 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.
    pub async fn put(
        &self,
        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(len, stream).await
            }
            None => self.put_unknown(Algorithm::GitBlobSha256, stream).await,
        }
    }

    /// **`put_git_object(algorithm, len, stream) → (Key, Pin)`** — the
    /// git-object family's put seam (store-api.md §Write path;
    /// ADR-018): the known-length put path under the family's
    /// preimage-identity derivation. The caller provides git's
    /// loose-object form (`"<type> <len>\0" + content` — the header
    /// lives *inside* the stored value), the digest is the plain hash
    /// of that value, 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 over the stored value
    /// byte-identically, so every read verifies against git's own oid
    /// function; dedup within the family is structural (CAS by oid).
    ///
    /// `algorithm` must be a git-object family member
    /// ([`Algorithm::GitObjectSha1`] / [`Algorithm::GitObjectSha256`]);
    /// anything else is [`Error::KeyInvalid`] — blob-family content
    /// addresses through the ordinary [`Store::put`]. Known-length
    /// only, no `Option<u64>`: a git object's length is always known
    /// before its content (the pack-entry and loose-file forms both
    /// carry it). A stream that yields fewer or more bytes than `len`
    /// rejects with `Error::Verification` — zero residue either side
    /// (the known-length path's mapping inherited verbatim);
    /// kv-only mode rejects an over-threshold put with the inherited
    /// [`Error::Io`] rejection; media failures ride `Error::Io`.
    /// Pin-before-publish ordering and CAS re-put are inherited as-is.
    ///
    /// **Fleet puts (ADR-018 §Decision 7; ADR-012 §3/§4):** under a
    /// `fleet: true` declaration the same fleet forms run — the kv
    /// tier through the joint entry+pin tx, the pg-lo tier through the
    /// one-tx consolidation form, the local tier pin-first/
    /// publish-second; the family derivation threads through them
    /// (derivation-parameterized sites, not family-branching copies).
    ///
    /// **`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_git_object(
        &self,
        algorithm: Algorithm,
        len: u64,
        stream: impl AsyncRead + Unpin + Send,
    ) -> Result<(Key, Pin), Error> {
        if !matches!(
            algorithm,
            Algorithm::GitObjectSha1 | Algorithm::GitObjectSha256
        ) {
            return Err(Error::KeyInvalid);
        }
        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
    }

    /// **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`]'s one signature run, and the minted guard is
    /// contributed to the group instead of returned. 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,
        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(len, stream).await?;
        batch.add(pin);
        Ok(key)
    }

    /// The known-length internal path: one streaming pass, the preamble
    /// initialized into the derivation up front. Routing by the declared
    /// length (the pure function); a stream that yields fewer or more
    /// bytes than `len` rejects `Verification` with zero residue on
    /// either tier.
    async fn put_known<R: AsyncRead + Unpin + Send>(
        &self,
        len: u64,
        stream: R,
    ) -> Result<(Key, Pin), Error> {
        self.put_known_as(Algorithm::GitBlobSha256, len, stream)
            .await
    }

    /// The known-length internal path under a caller-chosen
    /// derivation (ADR-018 §Decision 6/§7): the canonical blob put
    /// routes the preamble hasher; the git-object family put routes
    /// the plain hasher (no preamble — the loose-object header is
    /// inside the preimage). 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 only caller is
    /// [`Store::put`]'s unknown-length arm — blob-family by
    /// construction (ADR-018 §Decision 6 rejected the family's
    /// unknown-length arm).
    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::GitBlobSha256, 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))
                }
            },
        }
    }
}