alkblobs 0.1.0

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::task::{Context, Poll};

use alkcall::channels::operations::{
    Establishment, EstablishmentError, OpenEstablisher, PubOpenHandler,
};
use alkcall::core::types::Connection;
use alkcall::protocol::wire::CallError;
use alkcall::registry::spec::{
    AccessControl, ChannelOpenSpec, OperationSpec, OperationType, Visibility,
};
use serde_json::{json, Value};
use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf};
use tokio::sync::mpsc;

use super::fetch::{FanoutPublisher, SlowSubscriberPolicy};
use super::tokens::OpsContext;
use super::wire::{
    declared_error, digest_ingress, digest_mismatch_error, digest_wire_schema, mint_refused_error,
    namespace_schema, put_message_details_schema, put_refused_error, BLOBS_CHANNEL_ALPN,
    DIGEST_MISMATCH, MINT_REFUSED, PUT_OP_ID, PUT_REFUSED, VERIFICATION_REFUSED,
};
use crate::facade::Pin;
use crate::hashing::{Algorithm, BlobHasher};
use crate::keys::Key;
use crate::Error;

// ------------------------------------------------------------------
// blobs/put (the binary data plane's producer side): the verified Sink
// op — the offer/stream/commit path, the pin hand-over, and the fanout
// seam's producer side.
// ------------------------------------------------------------------

/// The tee's wire read window (one read chunk's size — the tee loop
/// reads the incoming stream once, in bounded windows, and each read's
/// bytes are handed on exactly once per arm: the hash pass, the
/// broadcast's subscriber taps, and the store arm's pipe feed).
const PUT_WIRE_CHUNK: usize = 16 * 1024;
/// The store arm's pipe bound — in-flight chunks between the put's tee
/// reader and the facade put's consumption (the facade put consumes at
/// its own pace; the putter's stream is backpressured by both arms).
const PUT_PIPE_BOUND: usize = 64;
/// The establishment ack's field (`accepted: true`): the offer passed
/// its gate, the channel is adopted, and the put's stream rides it —
/// informative-in-result (ADR-014 §2: the ack and the `{token, digest}`
/// result merge into the single response).
const PUT_ACK_FIELD: &str = "accepted";

/// **`blobs/put`** (`channels/blobs/pub` over [`BLOBS_CHANNEL_ALPN`])
/// — the verified `Sink` op (ops-surface.md §Op family; alkcall
/// ADR-046/047 + ADR-014): `{namespace, digest?, len?}` offer + the
/// byte stream in; the response is `{token, digest}` (ADR-012 §2).
/// Exported as the open triple the embedder assembles — the
/// `OperationSpec` (the `channel_open` marker, `Pub`-typed, the
/// **write** action on the namespace resource), the establisher (the
/// awaited establishment phase whose reply fields carry the offer
/// ack), and the `PubOpenHandler` (the data plane) — registered
/// through `ChannelCore::register_pub_openable` (the harness's test
/// tree realizes the assembly pattern).
///
/// See the module doc's `blobs/put` section for the full contract: the
/// offer ack is informative-in-result (a putter streams without
/// waiting for it), verification runs before commit with zero residue,
/// the offered-digest claim cross-checks the canonical derivation, the
/// pin hand-over splits custody fleet/non-fleet per ADR-010 §4, and
/// the put's one stream read feeds the store arm and the fanout's
/// subscriber arms (serve-while-persisting).
pub fn put_openables(ctx: Arc<OpsContext>) -> (OperationSpec, OpenEstablisher, PubOpenHandler) {
    let spec = OperationSpec::new(
        PUT_OP_ID,
        OperationType::Pub,
        Visibility::External,
        json!({
            "type": "object",
            "properties": {
                "namespace": namespace_schema(),
                "digest": digest_wire_schema(),
                "len": { "type": "integer", "minimum": 0 }
            },
            "required": ["namespace"],
            "additionalProperties": false
        }),
        json!({
            "type": "object",
            "properties": {
                "channel_id": { "type": "integer", "minimum": 1 },
                PUT_ACK_FIELD: { "type": "boolean" },
                "token": { "type": "string", "pattern": "^[0-9a-f]{64}$" },
                "digest": { "type": "string", "pattern": "^[0-9a-f]{40}$|^[0-9a-f]{64}$" }
            },
            "required": ["channel_id", "token", "digest"],
            "additionalProperties": false
        }),
        vec![
            declared_error(
                DIGEST_MISMATCH,
                "the offered digest is not the stream's canonical derivation (a claim \
                 contradicted by the authority): nothing handed over",
                put_message_details_schema(),
            ),
            declared_error(
                VERIFICATION_REFUSED,
                "the stream failed verification against the canonical derivation \
                 (length or byte mismatch): nothing committed, zero residue on both tiers",
                put_message_details_schema(),
            ),
            declared_error(
                PUT_REFUSED,
                "the put refused before or during the stream (the kv-only over-threshold \
                 rejection, ADR-012 §6.2, and other put-time store failures): nothing committed",
                put_message_details_schema(),
            ),
            declared_error(
                MINT_REFUSED,
                "the hand-over's mint was refused at the token registry's admission cap \
                 (ADR-019): the put's stream already committed and verified on this node, \
                 but no token was issued and no pin custody was taken — non-fleet the \
                 entry stands as unreferenced content a sweep may reclaim, fleet the pin \
                 row stands and expires by its TTL; the re-put is CAS-idempotent",
                put_message_details_schema(),
            ),
        ],
        AccessControl {
            resource_type: Some("blob-namespace".to_owned()),
            resource_action: Some("write".to_owned()),
            ..AccessControl::default()
        },
        Some("$.namespace".to_owned()),
    )
    .with_channel_open(ChannelOpenSpec::new(BLOBS_CHANNEL_ALPN))
    .with_description(
        "the verified put: {namespace, digest?, len?} offer + the byte stream over the \
         alk/blobs channel; server verifies against the canonical derivation before commit; \
         the single response merges the offer ack and {token, digest}",
    );
    let establisher: OpenEstablisher =
        { Arc::new(move |input, _auth| Box::pin(async move { put_establish(input).await })) };
    let handler: PubOpenHandler = Arc::new(move |input, _plan, channel_conn, _auth| {
        let ctx = Arc::clone(&ctx);
        Box::pin(async move { put_handle(ctx, input, channel_conn).await })
    });
    (spec, establisher, handler)
}

/// The establisher's body — the awaited establishment phase: the offer's
/// post-schema ingress (the digest claim's form check; the schema gate
/// precedes) and the offer ack's reply fields.
async fn put_establish(input: Value) -> Result<Establishment, EstablishmentError> {
    if let Some(digest_hex) = input.get("digest").and_then(Value::as_str) {
        let _ = digest_ingress(digest_hex).map_err(|_| EstablishmentError::HandlerError {
            message: format!("the digest failed its ingress: {digest_hex}"),
        })?;
    }
    Ok(Establishment::default().with_reply_field(PUT_ACK_FIELD, json!(true)))
}

/// The tee→facade pipe: the put's one-read chunks flow to the facade
/// put's consumption through it. Bounded ([`PUT_PIPE_BOUND`]) — the
/// facade's consumption pace backpressures the tee (and with it the
/// putter's stream).
struct PutPipeWriter {
    tx: Option<mpsc::Sender<Arc<Vec<u8>>>>,
    fail: Arc<AtomicBool>,
}

impl PutPipeWriter {
    /// Feed the next chunk onward. `false` = the pipe died (the facade
    /// put ended ahead of the stream — errored or was killed); the tee
    /// stops feeding (the facade's own error is the response).
    async fn feed(&mut self, chunk: Arc<Vec<u8>>) -> bool {
        let Some(tx) = &self.tx else {
            return false;
        };
        tx.send(chunk).await.is_ok()
    }

    /// The normal stream end: release the facade's EOF (its
    /// derivation-and-commit follows the pipe's end).
    fn eof(&mut self) {
        self.tx = None;
    }

    /// The pre-commit kill — the failure path: the failure flag arms
    /// the reader's next read into an error (the facade's stream-error
    /// path: no commit, zero residue), then the sender's drop wakes the
    /// blocked read. Order is load-bearing: the flag must be visible
    /// ahead of the wake.
    fn kill(&mut self) {
        self.fail.store(true, Ordering::Release);
        self.tx = None;
    }
}

/// The store arm's reader half: the facade put's `AsyncRead`. Serves
/// the channel's chunks; on the kill flag (set ahead of the sender's
/// drop, so any read — already-woken or actively polling — sees it)
/// errors like a failed stream: the facade put's error path discards
/// and never commits.
struct PutPipeReader {
    rx: mpsc::Receiver<Arc<Vec<u8>>>,
    fail: Arc<AtomicBool>,
    buffer: Option<Arc<Vec<u8>>>,
    offset: usize,
}

impl AsyncRead for PutPipeReader {
    fn poll_read(
        self: std::pin::Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<tokio::io::Result<()>> {
        let this = self.get_mut();
        loop {
            if this.fail.load(Ordering::Acquire) {
                return Poll::Ready(Err(tokio::io::Error::new(
                    tokio::io::ErrorKind::BrokenPipe,
                    "the put's feed failed pre-commit (the stream errored)",
                )));
            }
            if let Some(chunk) = &this.buffer {
                if this.offset < chunk.len() {
                    let src = &chunk[this.offset..];
                    let n = src.len().min(buf.remaining());
                    buf.put_slice(&src[..n]);
                    this.offset += n;
                    return Poll::Ready(Ok(()));
                }
                this.buffer = None;
                this.offset = 0;
            }
            match this.rx.poll_recv(cx) {
                Poll::Ready(Some(chunk)) => {
                    this.buffer = Some(chunk);
                    this.offset = 0;
                }
                Poll::Ready(None) => return Poll::Ready(Ok(())),
                Poll::Pending => return Poll::Pending,
            }
        }
    }
}

/// How the tee's one read of the wire stream ended — the three endings
/// the put handle branches on.
enum PutStreamEnd {
    /// The wire ended cleanly (the putter's stream completed): the
    /// offered-digest cross-check runs here, then the facade's EOF.
    Eof,
    /// The wire died mid-stream (the putter's channel tore down, a
    /// transport failure): the put is killed pre-commit.
    WireFailed,
    /// The facade put already ended (it rejected ahead of the stream —
    /// the kv-only known-length rejection, an over-fed stream).
    PutEnded,
}

/// The put handler's body — the flow after the wrapper's establishment:
/// the offer's post-schema ingress, the fanout broadcast's begin (the
/// offered digest keys it), the tee's one read of the wire (hash pass +
/// subscriber arms + the store arm's pipe), then the verification /
/// commit / mint ordering and the response.
async fn put_handle(
    ctx: Arc<OpsContext>,
    input: Value,
    channel_conn: Connection,
) -> Result<Value, CallError> {
    let namespace = match input.get("namespace").and_then(Value::as_str) {
        Some(namespace) => namespace.to_owned(),
        None => return Err(CallError::invalid_input(
            "the namespace field is missing after the schema gate (unreachable post-validation)",
        )),
    };
    let offered_digest = match input.get("digest").and_then(Value::as_str) {
        Some(digest_hex) => Some(digest_ingress(digest_hex).map_err(|error| {
            CallError::invalid_input(format!(
                "the digest failed its ingress after the schema gate (unreachable \
                     post-validation): {error:?}"
            ))
        })?),
        None => None,
    };
    let offered_hex = input
        .get("digest")
        .and_then(Value::as_str)
        .map(str::to_owned);
    let offered_len = input.get("len").and_then(Value::as_u64);

    // The fanout seam's producer side (POC #3 finding A3): the
    // broadcast begins under the offered digest before the first wire
    // read — a fetch that taps it receives the put's stream from the
    // tap moment (store-first consult keeps already-committed digests
    // off the tap path). `None` (a broadcast already live for the
    // digest, or no offered digest) continues store-arm-only — the
    // incumbent's bytes are the digest's bytes under CAS.
    let publisher = offered_digest
        .as_ref()
        .and_then(|key| ctx.fanout().begin(key, SlowSubscriberPolicy::Stall));

    let mut bidi = channel_conn
        .accept_bi()
        .await
        .map_err(|_| CallError::internal("the put's data plane never opened"))?;

    // The store arm: the facade put consumes the tee's pipe feed
    // concurrently with the tee's reads (bounded pipe; backpressure
    // flows to the putter).
    let (tx, rx) = mpsc::channel(PUT_PIPE_BOUND);
    let fail = Arc::new(AtomicBool::new(false));
    let put_task = {
        let store = Arc::clone(ctx.store());
        let reader_fail = Arc::clone(&fail);
        tokio::spawn(async move {
            let reader = PutPipeReader {
                rx,
                fail: reader_fail,
                buffer: None,
                offset: 0,
            };
            store.put(offered_len, reader).await
        })
    };
    let mut writer = PutPipeWriter {
        tx: Some(tx),
        fail: Arc::clone(&fail),
    };

    // The offered-digest cross-check's hash pass: only on the known-
    // length form (the preamble is known up front there). The tee's
    // derivation rides the same one stream read as the store arm — no
    // second read of the wire.
    let mut hasher = match (offered_len.is_some(), &offered_hex) {
        (true, Some(_)) => Some(BlobHasher::new(
            Algorithm::GitBlobSha256,
            offered_len.expect("the known-length offer's len"),
        )),
        _ => None,
    };

    let mut read_buf = vec![0u8; PUT_WIRE_CHUNK];
    let stream_end = loop {
        if put_task.is_finished() {
            break PutStreamEnd::PutEnded;
        }
        match bidi.read(&mut read_buf).await {
            Ok(0) => {
                break PutStreamEnd::Eof;
            }
            Ok(n) => {
                let chunk: Arc<Vec<u8>> = Arc::new(read_buf[..n].to_vec());
                if let Some(hasher) = hasher.as_mut() {
                    hasher.update(&chunk[..]);
                }
                if let Some(publisher) = &publisher {
                    publisher.broadcast(Arc::clone(&chunk)).await;
                }
                if !writer.feed(Arc::clone(&chunk)).await {
                    break PutStreamEnd::PutEnded;
                }
            }
            Err(_) => {
                break PutStreamEnd::WireFailed;
            }
        }
    };

    match stream_end {
        PutStreamEnd::Eof => {
            // The offered-digest cross-check (known-length form): the
            // tee's derivation pass completed with the stream — a claim
            // the derivation contradicts rejects here, **before the
            // facade's EOF** and so before commit: the facade's put is
            // killed (its stream-error path, zero residue) and the
            // response carries the mismatch. A claim the derivation
            // confirms — or no claim — releases the facade's EOF: the
            // commit is the facade's own verification-then-commit.
            if let Some(hasher) = hasher.take() {
                let derived = Key::from(hasher.finish());
                if let Some(offered) = &offered_digest {
                    if &derived != offered {
                        writer.kill();
                        drop(put_task.await);
                        return Err(digest_mismatch_error(
                            offered_hex.as_deref().expect("the offered hex"),
                            &hex::encode(derived.digest().bytes()),
                        ));
                    }
                }
            }
            writer.eof();
            finish_put(
                ctx,
                &namespace,
                put_task,
                offered_digest,
                offered_hex,
                publisher,
            )
            .await
        }
        PutStreamEnd::WireFailed | PutStreamEnd::PutEnded => {
            // The wire died mid-stream, or the facade already ended:
            // the put must not commit on a torn/failed stream — the
            // same kill discipline on every failure path (zero residue
            // on both tiers), then the facade's own rejection (if it
            // produced one) is the response.
            writer.kill();
            put_failed_response(put_task, offered_hex).await
        }
    }
}

/// The put's failure settle: the facade put's own rejection (it ended
/// ahead of the tee — the kv-only put-time rejection, an over-fed
/// stream) is the response; a killed put's task errored through the
/// injected feed failure. The publisher (when any) drops at scope end —
/// the same broadcast end the abandoned put has.
async fn put_failed_response(
    put_task: tokio::task::JoinHandle<Result<(Key, Pin), Error>>,
    offered_hex: Option<String>,
) -> Result<Value, CallError> {
    match put_task.await {
        Ok(Err(error)) => Err(put_refused_error(
            "the store arm",
            offered_hex.as_deref(),
            error,
        )),
        Ok(Ok(..)) => Err(CallError::internal(
            "the put's store arm completed against a torn stream; ADR-005's CAS re-put is \
             the backstop",
        )),
        Err(join_error) => Err(CallError::internal(format!(
            "the put's store task failed ({})",
            if join_error.is_cancelled() {
                "cancelled"
            } else {
                "panicked"
            }
        ))),
    }
}

/// The post-EOF half of the put: the facade's commit (its verification
/// runs first — a failing stream is zero-residue and surfaces as the
/// facade's error), the offered-digest post-check (the unknown-length
/// form's cross-check — the derivation is the facade's own), then the
/// hand-over — the custody split (ADR-010 §4) — the broadcast's end,
/// and the `{token, digest}` result.
async fn finish_put(
    ctx: Arc<OpsContext>,
    namespace: &str,
    put_task: tokio::task::JoinHandle<Result<(Key, Pin), Error>>,
    offered_digest: Option<Key>,
    offered_hex: Option<String>,
    publisher: Option<FanoutPublisher>,
) -> Result<Value, CallError> {
    let (key, pin) = match put_task.await {
        Ok(Ok(landing)) => landing,
        Ok(Err(error)) => {
            return Err(put_refused_error("the commit verification", None, error));
        }
        Err(join_error) => {
            return Err(CallError::internal(format!(
                "the put's store task failed ({})",
                if join_error.is_cancelled() {
                    "cancelled"
                } else {
                    "panicked"
                }
            )));
        }
    };
    if let Some(publisher) = publisher {
        // The put landed: the broadcast's subscribers have the whole
        // feed and the entry is committed — the tap window closes (late
        // joiners take the store path, same bytes under CAS). The
        // abandoned form (plain drop) is equivalent; this is the
        // completed-put's end.
        publisher.finish();
    }
    if let Some(offered) = &offered_digest {
        // The offered-digest cross-check's unknown-length arm: the
        // derivation is the facade's own (the
        // restart-over-staged-bytes pass ran and the commit followed
        // it), so a claim the derived key contradicts rejects here —
        // no token mints over a put whose claim failed, and the guard
        // drops at scope end (the entry stands as valid CAS content
        // its own derivation justifies; GC reaps an unreferenced one).
        // The known-length arm's pre-commit check above already
        // rejected its mismatches before the facade's EOF; this is the
        // shared post-commit backstop.
        if &key != offered {
            drop(pin);
            return Err(digest_mismatch_error(
                offered_hex.as_deref().expect("the offered hex"),
                &hex::encode(key.digest().bytes()),
            ));
        }
    }
    let digest_hex = hex::encode(key.digest().bytes());
    // The hand-over (ADR-010 §4): the custody split. Fleet — mint over
    // the committed row, then the guard's drop: the row's expiry is
    // the protection from here on (the TTL machinery stays live on the
    // ops path). Non-fleet — the registry takes the guard: the token
    // is the in-process handle, nothing expires.
    //
    // The admission cap's refusal (ADR-019) answers the already-committed
    // put honestly: the declared `MINT_REFUSED` error names the committed
    // digest, no token was issued, and — non-fleet — the guard dropped at
    // the refused mint (no custody taken; the entry is sweep-eligible;
    // fleet, the row expires by its TTL). Operational mint failures
    // (entropy, engine) stay the internal mapping.
    let token = match ctx.fleet_owner() {
        Some(owner) => match ctx.mint_fleet(namespace, key, &owner) {
            Ok(token) => {
                drop(pin);
                token
            }
            Err(Error::RegistryFull) => return Err(mint_refused_error(&digest_hex)),
            Err(_) => {
                drop(pin);
                return Err(CallError::internal("the fleet token mint refused"));
            }
        },
        None => match ctx.mint_custodial(namespace, key, pin) {
            Ok(token) => token,
            Err(Error::RegistryFull) => return Err(mint_refused_error(&digest_hex)),
            Err(_) => return Err(CallError::internal("the custodial token mint refused")),
        },
    };
    Ok(json!({
        "token": token.to_hex(),
        "digest": digest_hex,
    }))
}