alkgit-substrate 0.1.0

//! The wave-9 real-lfs e2e's shared harness (substrate/w9-e2e-real-lfs):
//! the LFS test door + the LFS-capable git client fixtures, over the
//! same posture as [`crate::common`] (the git doors' shared fixture
//! code; the row/stamp discipline restated there — the core's
//! `tests/common` is not importable across the crate seam). The POC-L
//! reference door's shape re-landed: every machine consumes the
//! composition entry's complete backend (the object pool with both
//! liveness registrations, the LFS rows surface, the cached provider),
//! not doubles.
//!
//! # The LFS test door
//!
//! The alkhttp `lfs` feature stays family infrastructure (doors.md);
//! the test door mirrors the feature's route/auth facts (lfs.md §1):
//! the batch at `{repo}/objects/batch` (the `/info/lfs`-embedded shape
//! mounted beside it), the transfer at `{repo}/lfs/objects/{oid}`,
//! Basic tolerance on the batch (the real client's URL userinfo),
//! hrefs door-minted absolute with per-action `Authorization` tokens
//! (`lfs:<hex>` Bearer) over the adapter's mint inputs, the
//! anonymous-public `anon` synthesis on public repos (no header minted
//! for it — the anonymous action rides unauthenticated), HTTP `Range`
//! parsing with the 206/416 framing, and the door-mapped error
//! statuses (`Forbidden` → 401, `HashAlgo` → 409, the
//! mislabeled/invalid classes → 400, `TooLarge` → 413,
//! `NotSatisfiable` → 416). Every grant is re-checked by the adapter's
//! wired [`LfsGrantProvider`](alkgit_substrate::LfsGrantProvider) —
//! the door's resolution is never an authority; unknown credentials
//! and unknown or repo-mismatched tokens refuse rather than degrade
//! (the wrong-bearer 401 posture — invalid is never anonymous).
//!
//! The transfer routes resolve their caller from a valid **minted
//! token** (per-action: an upload-action token grants PUTs, a
//! download-action token GETs — a mismatch is the door's 403) or from
//! a resolved credential. The route grammar is a full repo-id path
//! parse (the enumerated-pattern constraint is alkhttp's static route
//! tables; this harness door carries the POC's suffix parsing instead
//! — the same protocol surface, no route-fallback machinery).
//!
//! The door is a test client's server only (the serving path spawns
//! nothing — the crate convention); its credential/token tables are
//! the harness's own.

#![allow(dead_code)]

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};

use alkblobs::Store as BlobStore;
use alkcall::core::Identity;
use alkgit::GitRegistry;
use alkgit_substrate::lfs_media_key;
use alkgit_substrate::{
    LfsAdapter, LfsAdapterConfig, LfsAdapterError, LfsBatchObject, LfsBatchOperation,
    LfsBatchRequest, LfsGrantProvider, LfsHrefMint, LfsRange,
};
use parking_lot::{Mutex, RwLock};
use serde_json::{json, Value};
use tokio::io::AsyncWriteExt;
use tokio::net::{TcpListener, TcpStream};
use tokio::task::JoinHandle;

use crate::common::{git, read_http_request, Harness, HttpRequest};

/// The batch request bodies' budget (the batch is small JSON; the real
/// client's batch rides far under it).
pub const BATCH_BODY_BUDGET: usize = 8 * 1024 * 1024;

/// The transfer PUT bodies' budget (an oversized body is the door's
/// 413; the rows' media rides hundreds of KB).
pub const TRANSFER_BODY_BUDGET: usize = 16 * 1024 * 1024;

/// The door's Basic credential seed for the harness's writer identity
/// (the rows' real-client batch credential — the URL userinfo's
/// password).
pub const WRITER_PASSWORD: &str = "lfs-writer-pass";

/// The door's Basic credential seed for the ungranted caller (the
/// wired-provider regression row's identity — a **static** resources
/// claim that rides nothing through the wired provider).
pub const OUTSIDER_PASSWORD: &str = "lfs-outside-pass";

const ANON_ID: &str = "anon";

/// The door's default adapter config — the rows' matrix breadth: the
/// POC-L split datum scaled down (a 128 KiB verified window, so
/// real-client downloads ride several windows) with the shipped 64 MiB
/// object ceiling.
pub fn default_config() -> LfsAdapterConfig {
    LfsAdapterConfig {
        window_bytes: 128 * 1024,
        ..LfsAdapterConfig::default()
    }
}

/// One logged door request (the rows' evidence): the routed method and
/// path, the response status, the resolved auth kind (`token` /
/// `basic` / `anonymous` / `none`), and — for batch responses — the
/// served objects' JSON shape.
#[derive(Clone, Debug)]
pub struct LfsDoorRequest {
    pub method: String,
    pub path: String,
    pub status: u16,
    pub auth: &'static str,
    pub batch_objects: Option<Vec<Value>>,
}

/// A running LFS door: the bound address (the batch hrefs' minting
/// base), its accept loop (the row aborts it at the end), and the
/// request log.
pub struct LfsDoor {
    pub addr: std::net::SocketAddr,
    pub accept: JoinHandle<()>,
    pub log: Arc<Mutex<Vec<LfsDoorRequest>>>,
}

impl LfsDoor {
    /// The repo's batch URL (the `remote.*.lfsurl` value): plain (the
    /// anonymous-public path) unless a password is given (the Basic
    /// tolerance's real-client shape — the URL userinfo).
    pub fn lfs_url(&self, repo_id: &str, password: Option<&str>) -> String {
        match password {
            Some(password) => format!("http://lfs:{password}@{}/{}", self.addr, repo_id),
            None => format!("http://{}/{}", self.addr, repo_id),
        }
    }

    /// The requests of one routed class, in arrival order.
    pub fn requests_of(&self, method: &str, path_suffix: &str) -> Vec<LfsDoorRequest> {
        self.log
            .lock()
            .iter()
            .filter(|request| request.method == method && request.path.ends_with(path_suffix))
            .cloned()
            .collect()
    }
}

/// The LFS door over the harness's assembly (the adapter assembled
/// from the composition's faces: the object pool, the wrapped LFS rows
/// — the assertions bump the liveness counter for free — and the
/// cached provider behind the wired `LfsGrantProvider`).
pub async fn spawn_lfs_door(harness: &Harness) -> LfsDoor {
    spawn_lfs_door_with(harness, default_config()).await
}

/// [`spawn_lfs_door`] with a row's own adapter config.
pub async fn spawn_lfs_door_with(harness: &Harness, config: LfsAdapterConfig) -> LfsDoor {
    let store = harness.assembly.blobs().clone();
    let ownership: Arc<dyn alkcall::core::ownership::OwnershipProvider> =
        Arc::new(LfsGrantProvider::new(harness.assembly.provider().clone()));
    let adapter = LfsAdapter::new(
        Arc::clone(&store),
        harness.assembly.lfs_rows().clone(),
        Arc::clone(&ownership),
        config,
    )
    .expect("the lfs door's adapter assembles");

    let listener = TcpListener::bind("127.0.0.1:0")
        .await
        .expect("the lfs door binds");
    let addr = listener.local_addr().expect("the lfs door's local addr");
    let shared = Arc::new(Shared {
        adapter,
        store,
        ownership,
        registry: harness.assembly.registry().clone(),
        creds: RwLock::new(creds(harness)),
        tokens: RwLock::new(HashMap::new()),
        token_counter: AtomicU64::new(0),
        log: Arc::new(Mutex::new(Vec::new())),
        addr,
    });
    let log = Arc::clone(&shared.log);
    let accept = tokio::spawn(async move {
        loop {
            let Ok((conn, _peer)) = listener.accept().await else {
                break;
            };
            let shared = Arc::clone(&shared);
            tokio::spawn(async move {
                let inner = lfs_conn(shared, conn);
                match tokio::time::timeout(crate::common::DOOR_CONN_TIMEOUT, inner).await {
                    Ok(Ok(())) | Ok(Err(())) | Err(_) => {}
                }
            });
        }
    });
    LfsDoor { addr, accept, log }
}

struct Shared {
    adapter: LfsAdapter,
    store: Arc<BlobStore>,
    ownership: Arc<dyn alkcall::core::ownership::OwnershipProvider>,
    registry: Arc<dyn GitRegistry>,
    creds: RwLock<HashMap<String, Identity>>,
    tokens: RwLock<HashMap<String, TokenGrant>>,
    token_counter: AtomicU64,
    log: Arc<Mutex<Vec<LfsDoorRequest>>>,
    addr: std::net::SocketAddr,
}

#[derive(Clone)]
struct TokenGrant {
    identity: Identity,
    repo: String,
    action: LfsBatchOperation,
}

/// The door's credential seeds: the harness's writer identity plus the
/// static-claim outsider (the wired-provider regression's shape).
fn creds(harness: &Harness) -> HashMap<String, Identity> {
    HashMap::from([
        (WRITER_PASSWORD.to_owned(), harness.identity.clone()),
        (OUTSIDER_PASSWORD.to_owned(), static_claim_outsider()),
    ])
}

/// The ungranted caller: an identity carrying a **static** resources
/// claim for the ops resource — admitted nowhere through the wired
/// provider (the live grant table decides; the static fallback is
/// unreachable through the adapter).
fn static_claim_outsider() -> Identity {
    Identity {
        id: "u-outside".to_owned(),
        scopes: Vec::new(),
        resources: HashMap::from([(
            "blob-namespace".to_owned(),
            vec!["read".to_owned(), "write".to_owned()],
        )]),
    }
}

enum Route {
    Batch(String),
    Put(String, String),
    Get(String, String),
}

/// The route parse (the POC's suffix grammar, both batch shapes): the
/// batch route's repo id is the path's prefix (leading slash
/// stripped); the transfer route's is the path before the
/// `/lfs/objects/` marker (the one served transfer shape — hrefs are
/// door-minted; an embedded `/info/lfs` segment is not a transfer
/// shape, so the doubled `info/lfs/lfs/…` form is unrouted — 404).
/// The oid is the transfer route's last segment, lowercased and
/// shape-checked (a non-oid is an unrouted shape — the door's 404,
/// not the adapter's per-object 422).
fn route(method: &str, path: &str) -> Option<Route> {
    if method == "POST" {
        for suffix in ["/info/lfs/objects/batch", "/objects/batch"] {
            if let Some(prefix) = path.strip_suffix(suffix) {
                let repo = prefix.strip_prefix('/').unwrap_or(prefix);
                if !repo.is_empty() {
                    return Some(Route::Batch(repo.to_owned()));
                }
            }
        }
        return None;
    }
    let marker = "/lfs/objects/";
    let position = path.find(marker)?;
    let repo = &path[1..position];
    let oid = path[position + marker.len()..].to_lowercase();
    if repo.is_empty() || repo.ends_with("/info/lfs") || lfs_media_key(&oid).is_err() {
        return None;
    }
    match method {
        "PUT" => Some(Route::Put(repo.to_owned(), oid)),
        "GET" => Some(Route::Get(repo.to_owned(), oid)),
        _ => None,
    }
}

fn anon_identity() -> Identity {
    Identity {
        id: ANON_ID.to_owned(),
        scopes: Vec::new(),
        resources: HashMap::new(),
    }
}

/// The resolved request auth: the identity the route hands to the
/// adapter (the authority), whether transfer-action headers mint for
/// it, and the log's kind. Unknown credentials and unknown or
/// repo-mismatched tokens are `None` (the 401 — refused, not
/// degraded).
struct Resolved {
    identity: Option<Identity>,
    mint: bool,
    kind: &'static str,
}

async fn resolve_auth(shared: &Shared, request: &HttpRequest, repo_id: &str) -> Option<Resolved> {
    let Some(authorization) = request.authorization.as_deref() else {
        return anonymous_read(shared, repo_id).await;
    };
    if let Some(rest) = authorization
        .strip_prefix("Bearer ")
        .or_else(|| authorization.strip_prefix("bearer "))
    {
        let token = rest.strip_prefix("lfs:")?;
        let granted = shared.tokens.read().get(token).cloned()?;
        if granted.repo != repo_id {
            return None;
        }
        return Some(Resolved {
            identity: Some(granted.identity),
            mint: true,
            kind: "token",
        });
    }
    if let Some(encoded) = authorization
        .strip_prefix("Basic ")
        .or_else(|| authorization.strip_prefix("basic "))
    {
        let password = decode_basic(encoded)?;
        let identity = shared.creds.read().get(&password).cloned()?;
        return Some(Resolved {
            identity: Some(identity),
            mint: true,
            kind: "basic",
        });
    }
    None
}

/// The no-header posture: a public repo resolves the synthesized
/// `anon` identity (the anonymous-public read posture — the adapter's
/// provider re-grants it); anything else refuses (the 401). The
/// anonymous ride is read-shaped: an upload through it is refused by
/// the adapter (push is always authenticated).
async fn anonymous_read(shared: &Shared, repo_id: &str) -> Option<Resolved> {
    let record = shared.registry.resolve(repo_id).await.ok()?;
    if record.visibility == alkgit::Visibility::Public {
        return Some(Resolved {
            identity: Some(anon_identity()),
            mint: false,
            kind: "anonymous",
        });
    }
    None
}

/// RFC 4648 base64 (standard alphabet, padding ignored) — the
/// harness's decoder for the real client's Basic userinfo.
fn basic64_decode(input: &str) -> Option<Vec<u8>> {
    const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
    let mut output = Vec::new();
    let mut acc: u32 = 0;
    let mut bits = 0u32;
    for byte in input.bytes() {
        if byte == b'=' {
            break;
        }
        let position = ALPHABET.iter().position(|c| *c == byte)? as u32;
        acc = (acc << 6) | position;
        bits += 6;
        if bits >= 8 {
            bits -= 8;
            output.push((acc >> bits) as u8);
        }
    }
    Some(output)
}

fn decode_basic(encoded: &str) -> Option<String> {
    let decoded = basic64_decode(encoded)?;
    let text = String::from_utf8(decoded).ok()?;
    let (_, password) = text.split_once(':')?;
    Some(password.to_owned())
}

async fn lfs_conn(shared: Arc<Shared>, mut conn: TcpStream) -> Result<(), ()> {
    let mut buffer: Vec<u8> = Vec::new();
    loop {
        let Some(request) = read_http_request(&mut conn, &mut buffer).await else {
            return Ok(());
        };
        let Some(path) = request.path.clone() else {
            return Ok(());
        };
        match route(&request.method, &path) {
            Some(Route::Batch(repo_id)) => batch(&shared, &mut conn, &repo_id, &request).await,
            Some(Route::Put(repo_id, oid)) => {
                transfer_put(&shared, &mut conn, &repo_id, &oid, &request).await
            }
            Some(Route::Get(repo_id, oid)) => {
                transfer_get(&shared, &mut conn, &repo_id, &oid, &request).await
            }
            None => respond_error(&mut conn, 404, "unknown route").await,
        }
    }
}

/// The batch route: the JSON request parsed → the adapter's `check` →
/// the response's JSON with the door-minted hrefs. Logged as served.
async fn batch(shared: &Shared, conn: &mut TcpStream, repo_id: &str, request: &HttpRequest) {
    let outcome = batch_inner(shared, repo_id, request).await;
    let (kind, status, objects) = match &outcome {
        Ok((kind, body)) => (
            *kind,
            200,
            body.get("objects").and_then(Value::as_array).cloned(),
        ),
        Err((kind, status, _message)) => (*kind, *status, None),
    };
    shared.log.lock().push(LfsDoorRequest {
        method: request.method.clone(),
        path: request.path.clone().unwrap_or_default(),
        status,
        auth: kind,
        batch_objects: objects,
    });
    match outcome {
        Ok((_kind, body)) => respond_json(conn, 200, &body).await,
        Err((_kind, status, message)) => {
            respond_json(conn, status, &json!({ "message": message })).await
        }
    }
}

/// The batch's mapping: the door's own gates (body budget, the JSON
/// parse, the credential resolution) precede the adapter; the
/// adapter's refusals map status-only (the door's resolution is never
/// an authority — every grant is re-checked through the wired
/// provider).
async fn batch_inner(
    shared: &Shared,
    repo_id: &str,
    request: &HttpRequest,
) -> Result<(&'static str, Value), (&'static str, u16, String)> {
    if request.body.len() > BATCH_BODY_BUDGET {
        return Err(("none", 413, "the batch body exceeds the budget".to_owned()));
    }
    let resolved = match resolve_auth(shared, request, repo_id).await {
        Some(resolved) => resolved,
        None => return Err(("none", 401, "authentication required".to_owned())),
    };
    let kind = resolved.kind;
    let batch = match parse_batch(&request.body) {
        Ok(batch) => batch,
        Err((status, message)) => return Err((kind, status, message)),
    };
    match shared
        .adapter
        .check(resolved.identity.as_ref(), repo_id, batch)
        .await
    {
        Ok(response) => {
            let objects = response
                .objects
                .iter()
                .map(|answer| {
                    let mut object = json!({
                        "oid": answer.oid,
                        "size": answer.size,
                        "authenticated": answer.authenticated,
                    });
                    if let Some(error) = &answer.error {
                        object["error"] = json!({ "code": error.code, "message": error.message });
                    }
                    if let Some(actions) = &answer.actions {
                        let mut shaped = json!({});
                        if let Some(mint) = &actions.upload {
                            shaped["upload"] = action_value(
                                shared,
                                mint,
                                LfsBatchOperation::Upload,
                                resolved.identity.as_ref(),
                                resolved.mint,
                            );
                        }
                        if let Some(mint) = &actions.download {
                            shaped["download"] = action_value(
                                shared,
                                mint,
                                LfsBatchOperation::Download,
                                resolved.identity.as_ref(),
                                resolved.mint,
                            );
                        }
                        object["actions"] = shaped;
                    }
                    object
                })
                .collect::<Vec<_>>();
            Ok((
                kind,
                json!({
                    "transfer": response.transfer,
                    "hash_algo": response.hash_algo,
                    "objects": objects,
                }),
            ))
        }
        Err(error) => {
            let (status, _) = error_status(&error);
            Err((kind, status, error.to_string()))
        }
    }
}

/// The batch request's parse (the door's JSON side): the operation
/// mapped through the adapter's wire form, the optional hash
/// algorithm, and the objects' mandatory `{oid, size}` shapes — a
/// shape-less request is the door's 400 (the adapter's contract takes
/// a well-formed wire request; the door refuses malformed before it).
fn parse_batch(body: &[u8]) -> Result<LfsBatchRequest, (u16, String)> {
    let value: Value = serde_json::from_slice(body)
        .map_err(|error| (400u16, format!("bad batch request: {error}")))?;
    let operation = value
        .get("operation")
        .and_then(Value::as_str)
        .ok_or_else(|| (400u16, "the batch carries no operation".to_owned()))?;
    let operation = LfsBatchOperation::from_wire(operation)
        .ok_or_else(|| (400u16, format!("unknown batch operation: {operation}")))?;
    let hash_algo = value
        .get("hash_algo")
        .and_then(Value::as_str)
        .map(str::to_owned);
    let mut objects = Vec::new();
    let request_objects = value
        .get("objects")
        .and_then(Value::as_array)
        .ok_or_else(|| (400u16, "the batch carries no objects".to_owned()))?;
    for object in request_objects {
        let oid = object
            .get("oid")
            .and_then(Value::as_str)
            .ok_or_else(|| (400u16, "a batch object carries no oid".to_owned()))?;
        let size = object
            .get("size")
            .and_then(Value::as_u64)
            .ok_or_else(|| (400u16, "a batch object carries no size".to_owned()))?;
        objects.push(LfsBatchObject {
            oid: oid.to_owned(),
            size,
        });
    }
    Ok(LfsBatchRequest {
        operation,
        hash_algo,
        objects,
    })
}

/// The action's JSON shape: the door-minted absolute href plus the
/// per-action `Authorization` header — minted for named identities
/// (`lfs:<hex>` Bearer, a fresh token per action), omitted for the
/// anonymous ride (the client's unauthenticated action, the
/// anonymous-public read's posture).
fn action_value(
    shared: &Shared,
    mint: &LfsHrefMint,
    action: LfsBatchOperation,
    identity: Option<&Identity>,
    mintable: bool,
) -> Value {
    let href = format!(
        "http://{}/{}/lfs/objects/{}",
        shared.addr, mint.repo_id, mint.oid
    );
    if !mintable {
        return json!({ "href": href });
    }
    let Some(identity) = identity else {
        return json!({ "href": href });
    };
    let token = mint_token(shared, identity, mint.repo_id.clone(), action);
    json!({ "href": href, "header": { "Authorization": format!("Bearer lfs:{token}") } })
}

fn mint_token(
    shared: &Shared,
    identity: &Identity,
    repo: String,
    action: LfsBatchOperation,
) -> String {
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos() as u64)
        .unwrap_or_default();
    let token = format!(
        "{:032x}",
        nanos ^ shared.token_counter.fetch_add(1, Ordering::Relaxed)
    );
    shared.tokens.write().insert(
        token.clone(),
        TokenGrant {
            identity: identity.clone(),
            repo,
            action,
        },
    );
    token
}

/// The transfer upload route: the auth resolved (the minted per-action
/// token or the credential), the pre-keyed offer's mandatory size from
/// the `Content-Length`, the body over the adapter's pre-keyed put.
async fn transfer_put(
    shared: &Shared,
    conn: &mut TcpStream,
    repo_id: &str,
    oid: &str,
    request: &HttpRequest,
) {
    let outcome = put_inner(shared, repo_id, oid, request).await;
    shared.log.lock().push(LfsDoorRequest {
        method: "PUT".to_owned(),
        path: format!("/{repo_id}/lfs/objects/{oid}"),
        status: match &outcome {
            Ok(kind) => {
                let _ = kind;
                200
            }
            Err((_, status)) => *status,
        },
        auth: match &outcome {
            Ok(kind) | Err((kind, _)) => kind,
        },
        batch_objects: None,
    });
    match outcome {
        Ok(_kind) => respond_empty(conn, 200).await,
        Err((_, status)) => respond_error(conn, status, status_message(status)).await,
    }
}

async fn put_inner(
    shared: &Shared,
    repo_id: &str,
    oid: &str,
    request: &HttpRequest,
) -> Result<&'static str, (&'static str, u16)> {
    let resolved = match resolve_auth(shared, request, repo_id).await {
        Some(resolved) => resolved,
        None => return Err(("none", 401)),
    };
    let kind = resolved.kind;
    if kind == "token" {
        let token = minted_token_of(request).expect("the token-auth request carries one");
        let granted = shared.tokens.read().get(&token).cloned();
        let grant = granted.expect("the resolved token is present");
        if grant.action != LfsBatchOperation::Upload {
            return Err((kind, 403));
        }
    }
    let Some(content_length) = request.content_length else {
        return Err((kind, 400));
    };
    if content_length as usize > TRANSFER_BODY_BUDGET || request.body.len() > TRANSFER_BODY_BUDGET {
        return Err((kind, 413));
    }
    let size = request.body.len() as u64;
    if size != content_length {
        return Err((kind, 400));
    }
    match shared
        .adapter
        .upload(
            resolved.identity.as_ref(),
            repo_id,
            oid,
            Some(size),
            &request.body[..],
        )
        .await
    {
        Ok(()) => Ok(kind),
        Err(error) => Err((kind, error_status(&error).0)),
    }
}

/// The transfer download route: the auth resolved, the Range parsed
/// and framed (the door-side stat gives the framing total), the
/// adapter's verified-window serve collected into the response body,
/// and the 200/206/416 shape written.
async fn transfer_get(
    shared: &Shared,
    conn: &mut TcpStream,
    repo_id: &str,
    oid: &str,
    request: &HttpRequest,
) {
    let outcome = get_inner(shared, repo_id, oid, request).await;
    shared.log.lock().push(LfsDoorRequest {
        method: "GET".to_owned(),
        path: format!("/{repo_id}/lfs/objects/{oid}"),
        status: match &outcome {
            Ok((_, status, _)) => *status,
            Err((_, status, _)) => *status,
        },
        auth: match &outcome {
            Ok((kind, _, _)) | Err((kind, _, _)) => kind,
        },
        batch_objects: None,
    });
    match outcome {
        Ok((_, _, served)) => {
            let mut extra = Vec::new();
            if served.status == 206 {
                extra.push(("Content-Range".to_owned(), served.content_range.clone()));
            }
            extra.push(("Accept-Ranges".to_owned(), "bytes".to_owned()));
            respond_bytes(
                conn,
                served.status,
                "application/octet-stream",
                &extra,
                &served.bytes,
            )
            .await;
        }
        Err((_, status, Some(content_range))) => {
            respond_bytes(
                conn,
                status,
                "application/octet-stream",
                &[("Content-Range".to_owned(), content_range)],
                b"",
            )
            .await;
        }
        Err((_, status, None)) => respond_error(conn, status, status_message(status)).await,
    }
}

/// The served shape (the download route's body + framing).
struct Served {
    status: u16,
    bytes: Vec<u8>,
    content_range: String,
}

async fn get_inner(
    shared: &Shared,
    repo_id: &str,
    oid: &str,
    request: &HttpRequest,
) -> Result<(&'static str, u16, Served), (&'static str, u16, Option<String>)> {
    let resolved = match resolve_auth(shared, request, repo_id).await {
        Some(resolved) => resolved,
        None => return Err(("none", 401, None)),
    };
    let kind = resolved.kind;
    if kind == "token" {
        let token = minted_token_of(request).expect("the token-auth request carries one");
        let granted = shared.tokens.read().get(&token).cloned();
        let grant = granted.expect("the resolved token is present");
        if grant.action != LfsBatchOperation::Download {
            return Err((kind, 403, None));
        }
    }

    let Ok(key) = lfs_media_key(oid) else {
        return Err((kind, 400, None));
    };
    // The grant consult rides **before** the existence read (the
    // adapter's own order): a resolved-but-ungranted credential must
    // not probe the entry's existence through the 404 (the wired
    // provider consulted live here — the adapter re-checks at the
    // download call either way, so this is the pre-stat ordering fix,
    // not a second authority).
    if let Some(identity) = resolved.identity.as_ref() {
        if !shared
            .ownership
            .owns(identity, "blob-namespace", repo_id, "read")
        {
            return Err((kind, 403, None));
        }
    }
    let Ok(Some(meta)) = shared.store.stat(key.key_bytes().as_slice()).await else {
        return Err((kind, 404, None));
    };
    let total = meta.len;

    let range = match resolve_range(request.range.as_deref(), total, oid) {
        Ok(range) => range,
        Err(()) => return Err((kind, 416, Some(format!("bytes */{total}")))),
    };
    let (status, expected, content_range) = match &range {
        None => (200u16, total, None),
        Some(range) => {
            let end = range
                .offset
                .saturating_add(range.len.unwrap_or(total))
                .min(total);
            let expected = end - range.offset;
            (
                206u16,
                expected,
                Some(format!("bytes {}-{}/{total}", range.offset, end - 1)),
            )
        }
    };
    match shared
        .adapter
        .download(resolved.identity.as_ref(), repo_id, oid, range)
        .await
    {
        Ok(mut download) => {
            use tokio::io::AsyncReadExt;
            let mut bytes = Vec::with_capacity(expected as usize);
            if let Err(_error) = download.read_to_end(&mut bytes).await {
                return Err((kind, 500, None));
            }
            assert_eq!(
                bytes.len() as u64,
                expected,
                "the served stream's length contradicts the door's framing"
            );
            Ok((
                kind,
                status,
                Served {
                    status,
                    bytes,
                    content_range: content_range.unwrap_or_default(),
                },
            ))
        }
        Err(error) => {
            let (status, _) = error_status(&error);
            let content_range = if status == 416 {
                Some(format!("bytes */{total}"))
            } else {
                None
            };
            Err((kind, status, content_range))
        }
    }
}

/// The Range header's resolution against the entry total (the door's
/// own — the adapter re-validates the explicit arithmetic): the whole
/// form (no header, `None`), `bytes=a-b`, `bytes=a-` (to the end — the
/// client's resume form), and the suffix form `bytes=-n` (the last `n`
/// bytes). Unparseable, inverted, zero-suffix, and past-EOF offsets
/// are unsatisfiable (the door's 416 with `bytes */total`); the
/// single-range grammar is the whole surface — a multi-range header
/// (`bytes=0-99,200-299`) or junk after the dash refuses (416), never
/// degenerating into an open-ended serve.
fn resolve_range(header: Option<&str>, total: u64, _oid: &str) -> Result<Option<LfsRange>, ()> {
    let Some(value) = header else {
        return Ok(None);
    };
    let Some(rest) = value.strip_prefix("bytes=") else {
        return Err(());
    };
    if let Some(suffix_text) = rest.strip_prefix('-') {
        let suffix: u64 = suffix_text.trim().parse().map_err(|_| ())?;
        if suffix == 0 {
            return Err(());
        }
        let from = total - suffix.min(total);
        return Ok(Some(LfsRange {
            offset: from,
            len: Some(total - from),
        }));
    }
    let Some((from_text, to_text)) = rest.split_once('-') else {
        return Err(());
    };
    let from: u64 = from_text.trim().parse().map_err(|_| ())?;
    if from >= total {
        return Err(());
    }
    match to_text.trim().parse::<u64>() {
        Ok(to) if to >= from => Ok(Some(LfsRange {
            offset: from,
            len: Some(to - from + 1),
        })),
        Ok(_) => Err(()),
        Err(_) if to_text.trim().is_empty() => Ok(Some(LfsRange {
            offset: from,
            len: None,
        })),
        Err(_) => Err(()),
    }
}

/// The request's raw minted token (the `lfs:` Bearer's hex form — the
/// door token table's key).
fn minted_token_of(request: &HttpRequest) -> Option<String> {
    let authorization = request.authorization.as_deref()?;
    let rest = authorization
        .strip_prefix("Bearer ")
        .or_else(|| authorization.strip_prefix("bearer "))?;
    Some(rest.strip_prefix("lfs:")?.to_owned())
}

fn error_status(error: &LfsAdapterError) -> (u16, &'static str) {
    match error {
        LfsAdapterError::Forbidden(_) => (401, "authentication or authorization refused"),
        LfsAdapterError::HashAlgo(_) => (409, "unsupported hash_algo"),
        LfsAdapterError::MislabeledBody(_)
        | LfsAdapterError::Invalid(_)
        | LfsAdapterError::SizelessOffer => (400, "invalid transfer offer"),
        LfsAdapterError::TooLarge(_) => (413, "the object exceeds the door's ceiling"),
        LfsAdapterError::NotSatisfiable => (416, "the range is not satisfiable"),
        LfsAdapterError::Storage(_)
        | LfsAdapterError::Rows(_)
        | LfsAdapterError::Custody(_)
        | LfsAdapterError::MintRefused(_) => (500, "storage failure"),
    }
}

// ——— the response writers (hand-rolled http/1.1, the git doors' shape) ———

fn status_message(status: u16) -> &'static str {
    match status {
        200 => "OK",
        206 => "Partial Content",
        400 => "Bad Request",
        401 => "Unauthorized",
        403 => "Forbidden",
        404 => "Not Found",
        409 => "Conflict",
        413 => "Payload Too Large",
        416 => "Range Not Satisfiable",
        500 => "Internal Server Error",
        _ => "Unknown",
    }
}

async fn write_head(
    conn: &mut TcpStream,
    status: u16,
    content_type: &str,
    extra: &[(String, String)],
    length: usize,
) {
    let mut head = format!(
        "HTTP/1.1 {} {}\r\nContent-Type: {content_type}\r\n",
        status,
        status_message(status)
    );
    for (name, value) in extra {
        head.push_str(&format!("{name}: {value}\r\n"));
    }
    head.push_str(&format!("Content-Length: {length}\r\n\r\n"));
    conn.write_all(head.as_bytes())
        .await
        .expect("the door writes the head");
}

async fn respond_json(conn: &mut TcpStream, status: u16, body: &Value) {
    let raw = serde_json::to_vec(body).expect("the batch response serializes");
    write_head(
        conn,
        status,
        "application/vnd.git-lfs+json",
        &[(
            "Cache-Control".to_owned(),
            "no-cache, max-age=0, must-revalidate".to_owned(),
        )],
        raw.len(),
    )
    .await;
    conn.write_all(&raw)
        .await
        .expect("the door writes the body");
    conn.flush().await.expect("the door flushes the body");
}

async fn respond_bytes(
    conn: &mut TcpStream,
    status: u16,
    content_type: &str,
    extra: &[(String, String)],
    body: &[u8],
) {
    write_head(conn, status, content_type, extra, body.len()).await;
    conn.write_all(body)
        .await
        .expect("the door writes the body");
    conn.flush().await.expect("the door flushes the body");
}

async fn respond_empty(conn: &mut TcpStream, status: u16) {
    write_head(conn, status, "application/octet-stream", &[], 0).await;
    conn.flush().await.expect("the door flushes");
}

async fn respond_error(conn: &mut TcpStream, status: u16, message: &str) {
    let bytes = format!("{message}\n");
    respond_bytes(conn, status, "text/plain", &[], bytes.as_bytes()).await;
}

// ——— the LFS-capable git client fixtures (test clients only) ———

/// The `git-lfs` binary's presence (the rows' skip signal, mirroring
/// the `git` lookup's PATH scan for an executable).
pub fn lfs_binary() -> Option<PathBuf> {
    executable_on_path("git-lfs")
}

fn executable_on_path(name: &str) -> Option<PathBuf> {
    let path = std::env::var_os("PATH")?;
    for dir in std::env::split_paths(&path) {
        let candidate = dir.join(name);
        if !candidate.is_file() {
            continue;
        }
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let executable = std::fs::metadata(&candidate)
                .map(|meta| meta.permissions().mode() & 0o111 != 0)
                .unwrap_or(false);
            if !executable {
                continue;
            }
        }
        return Some(candidate);
    }
    None
}

/// Deterministic pseudo-random bytes (xorshift) — the rows' media.
pub fn pseudo_bytes(seed: u64, len: usize) -> Vec<u8> {
    let mut x = seed | 1;
    (0..len)
        .map(|_| {
            x ^= x << 13;
            x ^= x >> 7;
            x ^= x << 17;
            (x & 0xff) as u8
        })
        .collect()
}

/// The content's LFS oid (the raw sha256 — byte-exact the media
/// family's digest half).
pub fn oid_of(content: &[u8]) -> String {
    use sha2::Digest as _;
    let digest = sha2::Sha256::digest(content);
    hex::encode(digest)
}

/// An LFS client repo (test client only): initialized at `base/name`
/// (main, repo-local identity), `git lfs install --local`ed (the
/// fresh-consumer quirk: the filter config is repo-local), `*.bin`
/// tracked, the media committed (the pointer-file blob plus the
/// `.gitattributes` blob — ordinary git blobs for the git path),
/// `lfs.locksverify false` (the lock API is not served v1 — POC-L's
/// pinned quirk). `None` propagates as the row's clean skip.
pub struct LfsClient {
    pub dir: PathBuf,
    pub media: Vec<u8>,
    pub oid: String,
}

impl LfsClient {
    /// The client repo's directory (the git invocations' cwd).
    pub fn path(&self) -> &Path {
        &self.dir
    }
}

pub async fn init_lfs_client(base: &Path, name: &str, media: Vec<u8>) -> Option<LfsClient> {
    let dir = base.join(name);
    std::fs::create_dir_all(&dir).expect("the lfs client dir creates");
    let oid = oid_of(&media);
    git(
        Some(base),
        &["init", "-q", "-b", "main", dir.to_str().expect("utf8 path")],
    )
    .await?;
    git(Some(&dir), &["config", "user.name", "alkgit e2e"]).await?;
    git(Some(&dir), &["config", "user.email", "e2e@alk.dev"]).await?;
    git(Some(&dir), &["lfs", "install", "--local"]).await?;
    git(Some(&dir), &["lfs", "track", "*.bin"]).await?;
    std::fs::write(dir.join("media.bin"), &media).expect("the media file writes");
    git(Some(&dir), &["add", "."]).await?;
    git(
        Some(&dir),
        &["commit", "-q", "-m", "the lfs client's commit"],
    )
    .await?;
    git(Some(&dir), &["config", "lfs.locksverify", "false"]).await?;
    Some(LfsClient { dir, media, oid })
}

/// The local LFS object store's wipe (the pull rows' precondition —
/// the pull re-materializes from the server, not the local store).
pub fn wipe_local_store(client: &LfsClient) {
    let store = client.dir.join(".git/lfs/objects");
    std::fs::remove_dir_all(&store).ok();
}