alkgit 0.1.0

//! The `GitAdapter` — the direct-ALPN producer half on `alk/git`
//! (ADR-010's producer shape, the alktty `TtyAdapter` template).
//!
//! `handle` accepts the connection and loops `accept_bi`; each
//! [`BiStream`](alkcall::core::BiStream) becomes one
//! [`drive_native_session`] — the ADR-016 §2 request-line preamble is
//! read and parsed from the stream start (service selection + repo id +
//! the pinned refusal set), then the resolve→authorize gate runs with
//! the service-selected action, and only then does the duplex serving
//! entry emit a protocol byte (ADR-007's order; unknown ≡ unauthorized,
//! ADR-008 — the gate's collapse is the resolved front's denial shape).
//! A connection hosts many sessions — one session per bidi stream, all
//! independent.
//!
//! The adapter is backend-agnostic: it holds the backend handles
//! (`Arc<dyn GitRegistry>` + the fetch-surface and push-surface
//! traits), [`Limits`], the `push-options` negotiated-gate flag (the
//! same assembler config value the stateless advertisement carries,
//! ADR-013 §11) and the server-wide pipeline-concurrency semaphore it
//! hands each session — all assembler config origin (ADR-009). It never
//! touches pkt-line APIs directly: the preamble grammar lives in
//! [`crate::preamble`], the session in [`crate::duplex`].

use std::sync::Arc;

use alkcall::core::auth::AuthContext;
use alkcall::core::Identity;
use alkcall::core::{Connection, HandlerError, ProtocolHandler, StreamError};
use async_trait::async_trait;
use tokio::io::{AsyncRead as TokioAsyncRead, AsyncWrite as TokioAsyncWrite};
use tokio::sync::Semaphore;

use crate::backend::{GitPackGen, GitPackIngest, GitRefs, GitRegistry};
use crate::duplex::{deny_decline, deny_generic, serve_duplex_resolved};
use crate::limits::Limits;
use crate::preamble::{read_request_line, RequestLineError};

/// The `ProtocolHandler` for `alk/git`. Construction takes the backend
/// handles (`Arc<dyn GitRegistry>` + the fetch-surface and push-surface
/// traits), [`Limits`], the `push-options` negotiated-gate flag
/// (ADR-013 §11 — default off in v1), and the server-wide
/// pipeline-concurrency semaphore (shared across every session this
/// adapter drives — ADR-009's server-wide wording; the assembler's
/// config origin even on the in-crate adapter path).
pub struct GitAdapter {
    registry: Arc<dyn GitRegistry>,
    refs: Arc<dyn GitRefs>,
    packgen: Arc<dyn GitPackGen>,
    ingest: Arc<dyn GitPackIngest>,
    limits: Limits,
    push_options_negotiated: bool,
    pipeline: Arc<Semaphore>,
}

impl GitAdapter {
    /// Constructs the adapter over the given backend handles and
    /// budgets.
    pub fn new(
        registry: Arc<dyn GitRegistry>,
        refs: Arc<dyn GitRefs>,
        packgen: Arc<dyn GitPackGen>,
        ingest: Arc<dyn GitPackIngest>,
        push_options_negotiated: bool,
        limits: Limits,
        pipeline: Arc<Semaphore>,
    ) -> Self {
        GitAdapter {
            registry,
            refs,
            packgen,
            ingest,
            limits,
            push_options_negotiated,
            pipeline,
        }
    }
}

#[async_trait]
impl ProtocolHandler for GitAdapter {
    fn alpn(&self) -> &'static [u8] {
        b"alk/git"
    }

    async fn handle(&self, connection: Connection, auth: &AuthContext) -> Result<(), HandlerError> {
        loop {
            let stream = match connection.accept_bi().await {
                Ok(stream) => stream,
                Err(StreamError::ConnectionClosed | StreamError::StreamClosed) => break,
                Err(e) => return Err(HandlerError::from(e)),
            };
            let session = SessionHandles {
                identity: auth.identity.clone(),
                registry: Arc::clone(&self.registry),
                refs: Arc::clone(&self.refs),
                packgen: Arc::clone(&self.packgen),
                ingest: Arc::clone(&self.ingest),
                push_options_negotiated: self.push_options_negotiated,
                limits: self.limits,
                pipeline: Arc::clone(&self.pipeline),
            };
            tokio::spawn(drive_native_session(stream, session));
        }
        Ok(())
    }
}

/// The per-session inputs cloned off the adapter for one bidi stream
/// (one clone set per accepted stream; the adapter itself is shared —
/// one `alk/git` connection hosts many independent sessions).
pub(crate) struct SessionHandles {
    pub(crate) identity: Option<Identity>,
    pub(crate) registry: Arc<dyn GitRegistry>,
    pub(crate) refs: Arc<dyn GitRefs>,
    pub(crate) packgen: Arc<dyn GitPackGen>,
    pub(crate) ingest: Arc<dyn GitPackIngest>,
    pub(crate) push_options_negotiated: bool,
    pub(crate) limits: Limits,
    pub(crate) pipeline: Arc<Semaphore>,
}

/// Drives one native session end-to-end over a duplex stream: the ADR-016
/// §2 request-line preamble (read + parse + the pinned refusal set), then
/// the resolve→authorize gate ([`serve_duplex_resolved`]) with the
/// service-selected action — no protocol byte precedes the gate.
///
/// Failure outcomes:
///
/// - a clean peer close before any preamble byte ends the session
///   quietly (no response is written);
/// - [`RequestLineError::RepoInvalid`] collapses into the same generic
///   denial as unknown-repo (-backend.md §repo-id grammar: on the
///   serving paths a grammar-invalid id is `unauthorized`, never a
///   distinct invalid wire error) — no existence oracle;
/// - every other preamble failure takes the clear repo-blind decline
///   band ([`RequestLineError::decline_message`]) before any
///   advertisement.
pub(crate) async fn drive_native_session<T>(stream: T, handles: SessionHandles)
where
    T: TokioAsyncRead + TokioAsyncWrite + Unpin + Send + 'static,
{
    let SessionHandles {
        identity,
        registry,
        refs,
        packgen,
        ingest,
        push_options_negotiated,
        limits,
        pipeline,
    } = handles;
    let mut stream = stream;
    let line = match read_request_line(&mut stream).await {
        Ok(Some(line)) => line,
        Ok(None) => return,
        Err(RequestLineError::RepoInvalid { repo, .. }) => {
            let _ = deny_generic(&repo, stream).await;
            return;
        }
        Err(error) => {
            if let Some(message) = error.decline_message() {
                deny_decline(stream, &message).await;
            }
            return;
        }
    };
    let _ = serve_duplex_resolved(
        line.repo(),
        identity.as_ref(),
        line.service(),
        stream,
        limits,
        registry,
        refs,
        packgen,
        ingest,
        push_options_negotiated,
        pipeline,
    )
    .await;
}