alkblobs 0.2.2

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, DIGEST_WIRE_PATTERN, MINT_REFUSED, PUT_OP_ID, PUT_REFUSED,
    VERIFICATION_REFUSED,
};
use crate::facade::Pin;
use crate::hashing::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, key}`
/// 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): the `{namespace, key, len}` offer — both
/// claim fields mandatory (ADR-020 §Decision 3: the wire never
/// discovers addresses; an unmeasured remote stream buffers at its own
/// layer before offering) — + the byte stream in; the response is
/// `{token, key}` (ADR-020 §Decision 4, amending 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 key's claim cross-checks the derivation under the key's
/// own algorithm pre-EOF (the one claim-check moment for every family
/// — ADR-020 §Decision 4), 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(),
                "key": digest_wire_schema(),
                "len": { "type": "integer", "minimum": 0 }
            },
            "required": ["namespace", "key", "len"],
            "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}$" },
                "key": { "type": "string", "pattern": DIGEST_WIRE_PATTERN }
            },
            "required": ["channel_id", "token", "key"],
            "additionalProperties": false
        }),
        vec![
            declared_error(
                DIGEST_MISMATCH,
                "the offered key is not the stream's derivation under the key's own \
                 algorithm (a claim contradicted by the authority): nothing handed over",
                put_message_details_schema(),
            ),
            declared_error(
                VERIFICATION_REFUSED,
                "the stream failed verification against the offered key's own 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, key, len} offer + the byte stream over the \
         alk/blobs channel; server verifies against the offered key's derivation before \
         commit; the single response merges the offer ack and {token, key}",
    );
    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 mandatory `key` claim's form
/// check; the schema gate precedes) and the offer ack's reply fields.
async fn put_establish(input: Value) -> Result<Establishment, EstablishmentError> {
    let Some(key_hex) = input.get("key").and_then(Value::as_str) else {
        return Err(EstablishmentError::HandlerError {
            message: "the key field is missing after the schema gate (unreachable \
                      post-validation)"
                .to_owned(),
        });
    };
    let _ = digest_ingress(key_hex).map_err(|_| EstablishmentError::HandlerError {
        message: format!("the key failed its ingress: {key_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
    /// claim 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 mandatory key + len claims),
/// the fanout broadcast's begin (the offered key 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 Some(key_hex) = input.get("key").and_then(Value::as_str) else {
        return Err(CallError::invalid_input(
            "the key field is missing after the schema gate (unreachable post-validation)",
        ));
    };
    let offered_key = digest_ingress(key_hex)?;
    let Some(offered_len) = input.get("len").and_then(Value::as_u64) else {
        return Err(CallError::invalid_input(
            "the len field is missing after the schema gate (unreachable post-validation)",
        ));
    };
    let offered_hex = hex::encode(offered_key.as_bytes());

    // The fanout seam's producer side (POC #3 finding A3): the
    // broadcast begins under the offered key 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 keys
    // off the tap path). `None` (a broadcast already live for the
    // key) continues store-arm-only — the incumbent's bytes are the
    // key's bytes under CAS.
    let publisher = ctx
        .fanout()
        .begin(&offered_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);
        let value = offered_key.clone();
        tokio::spawn(async move {
            let reader = PutPipeReader {
                rx,
                fail: reader_fail,
                buffer: None,
                offset: 0,
            };
            store
                .put_as(value.algorithm(), Some(offered_len), reader)
                .await
        })
    };
    let mut writer = PutPipeWriter {
        tx: Some(tx),
        fail: Arc::clone(&fail),
    };

    // The claim's cross-check pass: the tee's hasher opens at stream
    // start under the offered key's algorithm — every registered
    // family (the preamble-carrying members hash the preamble from
    // the mandatory `len`; the identity family hashes the preimage
    // verbatim). The tee's derivation rides the same one stream read
    // as the store arm — no second read of the wire.
    let mut hasher = BlobHasher::new(offered_key.algorithm(), offered_len);

    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());
                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-claim cross-check (the one claim-check
            // moment, pre-EOF, for every family — ADR-020 §Decision
            // 4): the tee's derivation pass completed with the stream
            // — a claim the derivation under the key's own algorithm
            // 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
            // releases the facade's EOF: the commit is the facade's
            // own verification-then-commit (the backstop behind the
            // cross-check).
            let derived = Key::from(hasher.finish());
            if derived != offered_key {
                writer.kill();
                drop(put_task.await);
                return Err(digest_mismatch_error(
                    &offered_hex,
                    &hex::encode(derived.as_bytes()),
                ));
            }
            writer.eof();
            finish_put(ctx, &namespace, put_task, 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, Some(&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<&str>,
) -> Result<Value, CallError> {
    match put_task.await {
        Ok(Err(error)) => Err(put_refused_error("the store arm", offered_hex, 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 backstop behind the pre-EOF claim cross-check,
/// which is the only claim-check moment), the hand-over — the custody
/// split (ADR-010 §4) — the broadcast's end, and the `{token, key}`
/// result.
async fn finish_put(
    ctx: Arc<OpsContext>,
    namespace: &str,
    put_task: tokio::task::JoinHandle<Result<(Key, Pin), Error>>,
    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();
    }
    let key_hex = hex::encode(key.as_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
    // key, 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(&key_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(&key_hex)),
            Err(_) => return Err(CallError::internal("the custodial token mint refused")),
        },
    };
    Ok(json!({
        "token": token.to_hex(),
        "key": key_hex,
    }))
}