alkblobs 0.2.0

use std::sync::Arc;

use alkcall::protocol::wire::{CallError, ResponseEnvelope};
use alkcall::registry::registration::{make_handler, Handler};
use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
use serde_json::json;

use super::tokens::{OpsContext, PinToken};
use super::wire::{
    declared_error, digest_details_schema, digest_ingress, digest_wire_schema, digests_wire_schema,
    envelope_error, json_string_array, json_string_field, namespace_schema, present_wire_schema,
    schema_validation_error, store_internal_error, token_wire_schema, BLOB_MISSING, GC_REFUSED,
    INVALID_DIGEST, INVALID_RENEWAL_FORM,
};
use crate::Error;

/// **`blobs/stat`** `{namespace, digest}` → `{len}` — the probe before
/// offering (ops-surface.md §Op family): a **Query**, the **read**
/// action on the namespace resource; the handler rides `Store::stat`
/// (kv first, then the large tier — the fall-through general rule,
/// ADR-012 §6.1). A miss is the typed missing-error schema
/// ([`BLOB_MISSING`] with the digest as details) — never a
/// `{len: null}`.
pub fn stat_pair(ctx: Arc<OpsContext>) -> (OperationSpec, Handler) {
    let spec = OperationSpec::new(
        "blobs/stat",
        OperationType::Query,
        Visibility::External,
        json!({
            "type": "object",
            "properties": {
                "namespace": namespace_schema(),
                "digest": digest_wire_schema()
            },
            "required": ["namespace", "digest"],
            "additionalProperties": false
        }),
        json!({
            "type": "object",
            "properties": { "len": { "type": "integer", "minimum": 0 } },
            "required": ["len"],
            "additionalProperties": false
        }),
        vec![
            declared_error(
                BLOB_MISSING,
                "the digest is absent from the pool",
                digest_details_schema(),
            ),
            declared_error(
                INVALID_DIGEST,
                "the digest is not tag-prefixed key hex (lowercase, 42 or 66 chars)",
                digest_details_schema(),
            ),
        ],
        AccessControl {
            resource_type: Some("blob-namespace".to_owned()),
            resource_action: Some("read".to_owned()),
            ..AccessControl::default()
        },
        Some("$.namespace".to_owned()),
    )
    .with_description(
        "the entry's byte length for a digest, no content load (the probe before offering)",
    );
    let handler = make_handler(move |input, context| {
        let ctx = Arc::clone(&ctx);
        async move {
            let request_id = context.request_id.clone();
            let Some(digest_hex) = json_string_field(&input, "digest") else {
                return schema_validation_error(&request_id);
            };
            let key = match digest_ingress(digest_hex) {
                Ok(key) => key,
                Err(error) => return envelope_error(&request_id, error),
            };
            match ctx.store().stat(&key.key_bytes()).await {
                Ok(Some(meta)) => ResponseEnvelope::ok(request_id, json!({ "len": meta.len })),
                Ok(None) => envelope_error(
                    &request_id,
                    CallError::new(BLOB_MISSING, "the digest is absent from the pool", false)
                        .with_details(json!({ "digest": digest_hex })),
                ),
                Err(error) => store_internal_error(&request_id, "stat", error),
            }
        }
    });
    (spec, handler)
}

/// **`blobs/have`** `{namespace, digests[], token?}` → `{present[]}` —
/// the have half of have/need set diffing (ops-surface.md §Op family;
/// ADR-012 §2): a **Query**, **read-gated** exactly like fetch (an
/// existence probe over arbitrary digests is a discovery surface) —
/// hashes only, content never traverses this op; `present[]`
/// corresponds to `digests[]` by index.
///
/// **Without `token`** — the pure existence probe (ADR-008's has
/// discipline) over the facade's `has`; an engine failure folds to
/// `present: false` (the probes stay probes; the caller's next act —
/// fetch against the miss — is always safe, and the putter's backstop
/// re-put is CAS-idempotent, never harmful).
///
/// **With `token`** — the **renewal form** (ADR-012 §2; ADR-010 §4):
/// exactly one digest accompanies the token (a different count is the
/// typed [`INVALID_RENEWAL_FORM`] error); the handler rides
/// [`OpsContext::renew`] — the foundation's verify-the-mapping-then-
/// renew probe (the validity domain, ADR-017 §3's namespace parity,
/// the one-token-one-row digest check, the store core's
/// classification then the renewal) — and answers `present[]` as
/// usual: a live row renews (`present: true`), and an unknown
/// (foreign, or namespace-mismatched — the validity-domain refusals),
/// mismatched, or expired answer is `present: false, no renewal
/// occurred` (never an error — the putter's backstop is re-put).
/// A **malformed** token (over-length or non-hex — review-001 S-5's
/// `token` field bound) is the **schema gate's `INVALID_INPUT`**, not
/// this arm; the handler's malformed-token mapping (`from_hex` →
/// `present: false`, [`crate::ops::RenewalOutcome::UnknownToken`]'s
/// store-free form) remains for direct and composed calls that ride
/// past the wire gate.
pub fn have_pair(ctx: Arc<OpsContext>) -> (OperationSpec, Handler) {
    let spec = OperationSpec::new(
        "blobs/have",
        OperationType::Query,
        Visibility::External,
        json!({
            "type": "object",
            "properties": {
                "namespace": namespace_schema(),
                "digests": digests_wire_schema(),
                "token": token_wire_schema()
            },
            "required": ["namespace", "digests"],
            "additionalProperties": false,
            "if": { "required": ["token"] },
            "then": { "properties": { "digests": { "minItems": 1, "maxItems": 1 } } }
        }),
        present_wire_schema(),
        vec![
            declared_error(
                INVALID_RENEWAL_FORM,
                "a renewal probe carries exactly one digest alongside its token",
                json!({
                    "type": "object",
                    "properties": { "digests_len": { "type": "integer", "minimum": 0 } },
                    "required": ["digests_len"],
                    "additionalProperties": false
                }),
            ),
            declared_error(
                INVALID_DIGEST,
                "the digest is not tag-prefixed key hex (lowercase, 42 or 66 chars)",
                digest_details_schema(),
            ),
        ],
        AccessControl {
            resource_type: Some("blob-namespace".to_owned()),
            resource_action: Some("read".to_owned()),
            ..AccessControl::default()
        },
        Some("$.namespace".to_owned()),
    )
    .with_description("the existence probe over digests (index-corresponded present[]); the pin-token renewal form is its single-digest shape");
    let handler = make_handler(move |input, context| {
        let ctx = Arc::clone(&ctx);
        async move {
            let request_id = context.request_id.clone();
            let Some(namespace) = json_string_field(&input, "namespace") else {
                return schema_validation_error(&request_id);
            };
            let Some(digests_wire) = json_string_array(&input, "digests") else {
                return schema_validation_error(&request_id);
            };
            match json_string_field(&input, "token") {
                Some(token_hex) => {
                    have_renewal(ctx, &request_id, namespace, &digests_wire, token_hex).await
                }
                None => have_probe(ctx, &request_id, &digests_wire).await,
            }
        }
    });
    (spec, handler)
}

/// The renewal form's handler arm: the one-digest arity check, the
/// digest's ingress, the token's ingress (malformed tokens answer the
/// unknown-token form, `present: false`, not an error — for direct
/// and composed calls; the wire gate has already bounded the field,
/// review-001 S-5, so a wire-ridden malformed form never reaches
/// here), then the foundation's renewal probe ([`OpsContext::renew`])
/// with the
/// payload's namespace (ADR-017 §3's parity — the mint record decides)
/// — its outcome is the wire answer.
async fn have_renewal(
    ctx: Arc<OpsContext>,
    request_id: &str,
    namespace: &str,
    digests_wire: &[&str],
    token_hex: &str,
) -> ResponseEnvelope {
    if digests_wire.len() != 1 {
        return envelope_error(
            request_id,
            CallError::new(
                INVALID_RENEWAL_FORM,
                "a renewal probe carries exactly one digest alongside its token",
                false,
            )
            .with_details(json!({ "digests_len": digests_wire.len() })),
        );
    }
    let key = match digest_ingress(digests_wire[0]) {
        Ok(key) => key,
        Err(error) => return envelope_error(request_id, error),
    };
    let present = match PinToken::from_hex(token_hex) {
        Some(token) => ctx.renew(namespace, &token, &key).await.present(),
        None => false,
    };
    ResponseEnvelope::ok(request_id, json!({ "present": [present] }))
}

/// The pure existence probe's handler arm: the digests' ingress, then
/// `Store::has` per digest, index-corresponded; engine failures fold to
/// `present: false` (the probes stay probes — ADR-008's discipline).
async fn have_probe(
    ctx: Arc<OpsContext>,
    request_id: &str,
    digests_wire: &[&str],
) -> ResponseEnvelope {
    let mut keys = Vec::with_capacity(digests_wire.len());
    for digest_hex in digests_wire {
        match digest_ingress(digest_hex) {
            Ok(key) => keys.push(key),
            Err(error) => return envelope_error(request_id, error),
        }
    }
    let mut present = Vec::with_capacity(keys.len());
    for key in &keys {
        present.push(ctx.store().has(&key.key_bytes()).await.unwrap_or(false));
    }
    ResponseEnvelope::ok(request_id, json!({ "present": present }))
}

/// **`blobs/delete`** `{namespace, digests[]}` — the operator's
/// correction flow (ops-surface.md §Op family; ADR-017's privilege
/// model): a **Mutation** with **`Visibility::Internal`** — not callable
/// from the wire (an external discovery answers the protocol's
/// `NOT_FOUND`, the visibility discipline's non-leakage form; only
/// internal composition reaches it) and **not namespace-gated** — the
/// `namespace` field exists for audit/triage scoping (which slice of the
/// pool the deletion targets), not for gating; the gate is the
/// **internal authority context**: the ACL check runs against the
/// composition authority the embedder declares at registration (alkcall
/// ADR-017 — internal calls switch authority context, never skip ACL).
///
/// The handler rides `Store::delete` (the kv-large fall-through consult
/// then the full-protection-set arbitration; ADR-012 §6.1's general
/// rule): a deleted digest answers `deleted[]` in input order; a
/// protected key is the typed [`GC_REFUSED`] error (nothing deleted —
/// also the preceding digests in this call are the only ones ever
/// deleted: the path is fail-fast); an absent key is
/// [`BLOB_MISSING`]; the embedder's correction flow for a refusal is
/// deregisters/re-registers around the delete (ADR-012 §6.4).
pub fn delete_pair(ctx: Arc<OpsContext>) -> (OperationSpec, Handler) {
    let spec = OperationSpec::new(
        "blobs/delete",
        OperationType::Mutation,
        Visibility::Internal,
        json!({
            "type": "object",
            "properties": {
                "namespace": namespace_schema(),
                "digests": digests_wire_schema()
            },
            "required": ["namespace", "digests"],
            "additionalProperties": false
        }),
        json!({
            "type": "object",
            "properties": { "deleted": digests_wire_schema() },
            "required": ["deleted"],
            "additionalProperties": false
        }),
        vec![
            declared_error(
                BLOB_MISSING,
                "the digest is absent: nothing deleted",
                digest_details_schema(),
            ),
            declared_error(
                GC_REFUSED,
                "the digest is protected (pinned, liveness-live, or protect-callback-named): \
                 nothing deleted",
                digest_details_schema(),
            ),
            declared_error(
                INVALID_DIGEST,
                "the digest is not tag-prefixed key hex (lowercase, 42 or 66 chars)",
                digest_details_schema(),
            ),
        ],
        AccessControl::default(),
        None,
    )
    .with_description("the operator's correction-flow deletion (internal authority context; failures are fail-fast)");
    let handler = make_handler(move |input, context| {
        let ctx = Arc::clone(&ctx);
        async move {
            let request_id = context.request_id.clone();
            let Some(digests_wire) = json_string_array(&input, "digests") else {
                return schema_validation_error(&request_id);
            };
            let mut keys = Vec::with_capacity(digests_wire.len());
            for digest_hex in &digests_wire {
                match digest_ingress(digest_hex) {
                    Ok(key) => keys.push(key),
                    Err(error) => return envelope_error(&request_id, error),
                }
            }
            let mut deleted = Vec::with_capacity(digests_wire.len());
            for (key, digest_hex) in keys.iter().zip(&digests_wire) {
                match ctx.store().delete(&key.key_bytes()).await {
                    Ok(()) => deleted.push(digest_hex.to_owned()),
                    Err(Error::Missing) => {
                        return envelope_error(
                            &request_id,
                            CallError::new(
                                BLOB_MISSING,
                                "the digest is absent: nothing deleted",
                                false,
                            )
                            .with_details(json!({ "digest": digest_hex })),
                        );
                    }
                    Err(Error::GcRefuse) => {
                        return envelope_error(
                            &request_id,
                            CallError::new(
                                GC_REFUSED,
                                "the digest is protected (pinned, liveness-live, or \
                                 protect-callback-named): nothing deleted",
                                false,
                            )
                            .with_details(json!({ "digest": digest_hex })),
                        );
                    }
                    Err(error) => return store_internal_error(&request_id, "delete", error),
                }
            }
            ResponseEnvelope::ok(request_id, json!({ "deleted": deleted }))
        }
    });
    (spec, handler)
}