use std::borrow::Cow;
use std::collections::{HashMap, HashSet};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};

use rand::rng;
use russh::server::{Auth, ChannelOpenHandle, Handler, Msg, Server, Session};
use russh::{Channel, ChannelId, MethodKind, MethodSet, Preferred, Pty};
use ssh_key::{Algorithm, EcdsaCurve, PrivateKey, PublicKey};
use thiserror::Error;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpListener;
use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
use tokio::task::JoinHandle;

use crate::auth::SshPublicKey;
use crate::git::packetline::{MAX_REQUEST_BYTES, Packet, decode, encode_data, first_flush_end};
use crate::git::receive_pack::{ReceivePack, ReceivePackError};
use crate::git::transport::{GitRepositories, GitSshService};
use crate::git::upload_pack::{ProtocolVersion, UploadPack, UploadPackError};
use crate::issue::{IssueError, IssueService, MAX_BODY_BYTES, MAX_TITLE_BYTES};
use crate::policy::RepositoryOperation;
use crate::pull_request::{PullRequestError, PullRequestService};
use crate::rate_limit::AttemptLimiter;
use crate::repository::{RepositoryService, RepositoryServiceError};
use crate::store::{Store, StoreError};
use crate::telemetry::Telemetry;

const VERSION_COMMAND: &[u8] = b"tit --version";
const HELP_COMMAND: &[u8] = b"help";
const AUTH_COMMAND_PREFIX: &[u8] = b"auth ";
const GIT_PROTOCOL_VARIABLE: &str = "GIT_PROTOCOL";
const MAX_RECEIVE_PACK_BYTES: u64 = 128 * 1024 * 1024;
const MAX_REPOSITORY_COMMAND_BYTES: usize = 512;
const MAX_ISSUE_COMMAND_BYTES: usize = 512;
const MAX_PULL_REQUEST_COMMAND_BYTES: usize = 512;
const MAX_ISSUE_INPUT_BYTES: usize = MAX_TITLE_BYTES + 1 + MAX_BODY_BYTES;
const SSH_ATTEMPTS_PER_MINUTE: usize = 30;
const MAX_SSH_CLIENTS: usize = 4096;
const MAX_GIT_CHANNELS_PER_CONNECTION: usize = 4;
const MAX_GIT_CHANNELS_GLOBAL: usize = 64;
const GIT_CHANNEL_INACTIVITY_LIMIT: Duration = Duration::from_secs(30);
const GIT_CHANNEL_WALL_CLOCK_LIMIT: Duration = Duration::from_secs(120);
const MAX_GIT_CHANNEL_BYTES: u64 = 257 * 1024 * 1024;
const REPOSITORY_CREATE_USAGE: &str = "repo create NAME [--output human|json]";
const ISSUE_LIST_USAGE: &str = "issue list OWNER/REPOSITORY [--output human|json]";
const ISSUE_CREATE_USAGE: &str = "issue create OWNER/REPOSITORY [--output human|json]";
const ISSUE_COMMENT_USAGE: &str = "issue comment OWNER/REPOSITORY NUMBER [--output human|json]";
const ISSUE_CLOSE_USAGE: &str = "issue close OWNER/REPOSITORY NUMBER [--output human|json]";
const ISSUE_REOPEN_USAGE: &str = "issue reopen OWNER/REPOSITORY NUMBER [--output human|json]";
const PULL_REQUEST_LIST_USAGE: &str =
    "pr list OWNER/REPOSITORY [--state open|closed|merged|all] [--output human|json]";
const PULL_REQUEST_CREATE_USAGE: &str =
    "pr create OWNER/REPOSITORY BASE HEAD [--output human|json]";
const PULL_REQUEST_CLOSE_USAGE: &str = "pr close OWNER/REPOSITORY NUMBER [--output human|json]";
const PULL_REQUEST_REOPEN_USAGE: &str = "pr reopen OWNER/REPOSITORY NUMBER [--output human|json]";
const PULL_REQUEST_CHECKOUT_USAGE: &str =
    "pr checkout OWNER/REPOSITORY NUMBER [--output human|json]";
const HELP_TEXT: &str = "\
Available tit SSH commands:
  help
  tit --version
  auth ONE-TIME-SECRET
  repo create NAME [--output human|json]
  issue list OWNER/REPOSITORY [--output human|json]
  issue create OWNER/REPOSITORY [--output human|json]
  issue comment OWNER/REPOSITORY NUMBER [--output human|json]
  issue close OWNER/REPOSITORY NUMBER [--output human|json]
  issue reopen OWNER/REPOSITORY NUMBER [--output human|json]
  pr list OWNER/REPOSITORY [--state open|closed|merged|all] [--output human|json]
  pr create OWNER/REPOSITORY BASE HEAD [--output human|json]
  pr close OWNER/REPOSITORY NUMBER [--output human|json]
  pr reopen OWNER/REPOSITORY NUMBER [--output human|json]
  pr checkout OWNER/REPOSITORY NUMBER [--output human|json]

Git clients can also use git-upload-pack and git-receive-pack.
";
const HELP_GUIDANCE: &str = "Use 'help' to list the available commands.";

pub(crate) type LoginApprover =
    Arc<dyn Fn(&str, &str, &str) -> Result<(String, String), ()> + Send + Sync>;

struct SshRuntime {
    max_connections: usize,
    telemetry: Telemetry,
    login: Option<LoginApprover>,
}

fn parse_auth_command(command: &[u8]) -> Option<String> {
    let secret = command.strip_prefix(AUTH_COMMAND_PREFIX)?;
    if secret.len() != 64
        || !secret
            .iter()
            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
    {
        return None;
    }
    String::from_utf8(secret.to_vec()).ok()
}

pub(crate) struct RunningSshServer {
    address: SocketAddr,
    handle: russh::server::RunningServerHandle,
    task: JoinHandle<std::io::Result<()>>,
    audit: Arc<RequestAudit>,
}

#[derive(Clone)]
pub(crate) struct AuthorizedSshKeys {
    keys: Arc<RwLock<HashMap<PublicKey, SshIdentity>>>,
}

#[derive(Clone, PartialEq, Eq)]
struct SshIdentity {
    username: String,
    fingerprint: String,
}

impl AuthorizedSshKeys {
    pub(crate) fn new(keys: &[SshPublicKey]) -> Self {
        Self {
            keys: Arc::new(RwLock::new(key_map(keys))),
        }
    }

    pub(crate) fn for_accounts(keys: Vec<(String, SshPublicKey)>) -> Self {
        Self {
            keys: Arc::new(RwLock::new(account_key_map(keys))),
        }
    }

    pub(crate) fn replace_accounts(&self, keys: Vec<(String, SshPublicKey)>) {
        if let Ok(mut current) = self.keys.write() {
            *current = account_key_map(keys);
        }
    }

    fn identity(&self, key: &PublicKey) -> Option<SshIdentity> {
        self.keys.read().ok()?.get(key).cloned()
    }

    fn contains(&self, key: &PublicKey) -> bool {
        self.keys.read().is_ok_and(|keys| keys.contains_key(key))
    }
}

fn key_map(keys: &[SshPublicKey]) -> HashMap<PublicKey, SshIdentity> {
    keys.iter()
        .map(|key| {
            let fingerprint = key.fingerprint().to_owned();
            (
                key.public_key().clone(),
                SshIdentity {
                    username: fingerprint.clone(),
                    fingerprint,
                },
            )
        })
        .collect()
}

fn account_key_map(keys: Vec<(String, SshPublicKey)>) -> HashMap<PublicKey, SshIdentity> {
    keys.into_iter()
        .map(|(username, key)| {
            (
                key.public_key().clone(),
                SshIdentity {
                    username,
                    fingerprint: key.fingerprint().to_owned(),
                },
            )
        })
        .collect()
}

impl RunningSshServer {
    pub(crate) async fn start(
        address: SocketAddr,
        authorized_keys: &[SshPublicKey],
    ) -> Result<Self, SshServerError> {
        let host_key = PrivateKey::random(&mut rng(), Algorithm::Ed25519)?;
        Self::start_inner(address, authorized_keys, &[], None, host_key).await
    }

    pub(crate) async fn start_with_git(
        address: SocketAddr,
        authorized_keys: &[SshPublicKey],
        repositories: GitRepositories,
    ) -> Result<Self, SshServerError> {
        let host_key = PrivateKey::random(&mut rng(), Algorithm::Ed25519)?;
        Self::start_inner(address, authorized_keys, &[], Some(repositories), host_key).await
    }

    pub(crate) async fn start_with_git_and_host_key(
        address: SocketAddr,
        authorized_keys: &[SshPublicKey],
        repositories: GitRepositories,
        host_key: PrivateKey,
    ) -> Result<Self, SshServerError> {
        Self::start_inner(address, authorized_keys, &[], Some(repositories), host_key).await
    }

    pub(crate) async fn start_with_dynamic_keys(
        address: SocketAddr,
        authorized_keys: AuthorizedSshKeys,
        repositories: GitRepositories,
        host_key: PrivateKey,
        max_connections: usize,
        telemetry: Telemetry,
        login: LoginApprover,
    ) -> Result<Self, SshServerError> {
        recover_pushes(&repositories).await?;
        Self::start_inner_with_keys(
            address,
            authorized_keys,
            &[],
            Some(repositories),
            host_key,
            SshRuntime {
                max_connections,
                telemetry,
                login: Some(login),
            },
        )
        .await
    }

    pub(crate) async fn start_with_git_writes(
        address: SocketAddr,
        authorized_keys: &[SshPublicKey],
        writable_keys: &[SshPublicKey],
        repositories: GitRepositories,
    ) -> Result<Self, SshServerError> {
        recover_pushes(&repositories).await?;
        let host_key = PrivateKey::random(&mut rng(), Algorithm::Ed25519)?;
        Self::start_inner(
            address,
            authorized_keys,
            writable_keys,
            Some(repositories),
            host_key,
        )
        .await
    }

    async fn start_inner(
        address: SocketAddr,
        authorized_keys: &[SshPublicKey],
        writable_keys: &[SshPublicKey],
        repositories: Option<GitRepositories>,
        host_key: PrivateKey,
    ) -> Result<Self, SshServerError> {
        Self::start_inner_with_keys(
            address,
            AuthorizedSshKeys::new(authorized_keys),
            writable_keys,
            repositories,
            host_key,
            SshRuntime {
                max_connections: 1024,
                telemetry: Telemetry::default(),
                login: None,
            },
        )
        .await
    }

    async fn start_inner_with_keys(
        address: SocketAddr,
        authorized_keys: AuthorizedSshKeys,
        writable_keys: &[SshPublicKey],
        repositories: Option<GitRepositories>,
        host_key: PrivateKey,
        runtime: SshRuntime,
    ) -> Result<Self, SshServerError> {
        let listener = TcpListener::bind(address).await?;
        let address = listener.local_addr()?;
        let mut methods = MethodSet::empty();
        methods.push(MethodKind::PublicKey);
        let config = Arc::new(russh::server::Config {
            methods,
            auth_rejection_time: Duration::from_millis(250),
            auth_rejection_time_initial: Some(Duration::ZERO),
            keys: vec![host_key],
            preferred: Preferred {
                key: Cow::Owned(vec![
                    Algorithm::Ed25519,
                    Algorithm::Ecdsa {
                        curve: EcdsaCurve::NistP256,
                    },
                ]),
                ..Preferred::default()
            },
            max_auth_attempts: 3,
            inactivity_timeout: Some(Duration::from_secs(30)),
            nodelay: true,
            ..Default::default()
        });
        let writable_keys = Arc::new(
            writable_keys
                .iter()
                .map(|key| key.public_key().clone())
                .collect(),
        );
        let audit = Arc::new(RequestAudit::default());
        let server = SshServer {
            authorized_keys,
            writable_keys,
            audit: Arc::clone(&audit),
            repositories: repositories.map(Arc::new),
            login: runtime.login,
            connections: Arc::new(Semaphore::new(runtime.max_connections)),
            git_channels: Arc::new(Semaphore::new(MAX_GIT_CHANNELS_GLOBAL)),
            attempts: AttemptLimiter::new(
                SSH_ATTEMPTS_PER_MINUTE,
                Duration::from_secs(60),
                MAX_SSH_CLIENTS,
            ),
            telemetry: runtime.telemetry,
        };
        let (handle_sender, handle_receiver) = oneshot::channel();
        let task = tokio::spawn(async move {
            let mut server = server;
            let running = server.run_on_socket(config, &listener);
            let _ = handle_sender.send(running.handle());
            running.await
        });
        let handle = handle_receiver.await.map_err(|_| SshServerError::Startup)?;
        Ok(Self {
            address,
            handle,
            task,
            audit,
        })
    }

    pub(crate) fn address(&self) -> SocketAddr {
        self.address
    }

    pub(crate) fn audit(&self) -> RequestAuditSnapshot {
        self.audit.snapshot()
    }

    pub(crate) async fn shutdown(self) -> Result<(), SshServerError> {
        self.handle.shutdown("tit test shutdown".to_owned());
        self.task.await.map_err(|_| SshServerError::Join)??;
        Ok(())
    }

    pub(crate) async fn shutdown_bounded(
        mut self,
        limit: Duration,
    ) -> Result<bool, SshServerError> {
        self.handle.shutdown("tit shutdown".to_owned());
        match tokio::time::timeout(limit, &mut self.task).await {
            Ok(result) => {
                result.map_err(|_| SshServerError::Join)??;
                Ok(true)
            }
            Err(_) => {
                self.task.abort();
                let _ = self.task.await;
                Ok(false)
            }
        }
    }
}

async fn recover_pushes(repositories: &GitRepositories) -> Result<(), SshServerError> {
    let database = repositories
        .push_database()
        .ok_or_else(|| SshServerError::Recovery("push storage is not configured".to_owned()))?
        .to_owned();
    tokio::task::spawn_blocking(move || {
        crate::git::receive_pack::recover_incomplete_pushes(&database)
    })
    .await
    .map_err(|_| SshServerError::Join)?
    .map_err(|error| SshServerError::Recovery(error.to_string()))
}

#[derive(Debug, Error)]
pub(crate) enum SshServerError {
    #[error("SSH listener error: {0}")]
    Io(#[from] std::io::Error),
    #[error("SSH key error: {0}")]
    Key(#[from] ssh_key::Error),
    #[error("SSH server did not start")]
    Startup,
    #[error("SSH server task failed")]
    Join,
    #[error("cannot recover Git writes: {0}")]
    Recovery(String),
}

#[derive(Clone)]
struct SshServer {
    authorized_keys: AuthorizedSshKeys,
    writable_keys: Arc<HashSet<PublicKey>>,
    audit: Arc<RequestAudit>,
    repositories: Option<Arc<GitRepositories>>,
    login: Option<LoginApprover>,
    connections: Arc<Semaphore>,
    git_channels: Arc<Semaphore>,
    attempts: AttemptLimiter<IpAddr>,
    telemetry: Telemetry,
}

impl Server for SshServer {
    type Handler = SshSession;

    fn new_client(&mut self, peer_address: Option<SocketAddr>) -> Self::Handler {
        let connection_id = format!("{:032x}", rand::random::<u128>());
        self.telemetry.ssh_connection(&connection_id);
        SshSession {
            authorized_keys: self.authorized_keys.clone(),
            writable_keys: Arc::clone(&self.writable_keys),
            audit: Arc::clone(&self.audit),
            repositories: self.repositories.clone(),
            login: self.login.clone(),
            protocol: ProtocolVersion::V0,
            exec_channels: HashMap::new(),
            authenticated_identity: None,
            authenticated_key: None,
            authenticated_writer: false,
            peer_address: peer_address
                .map(|address| address.ip())
                .unwrap_or(IpAddr::V4(Ipv4Addr::LOCALHOST)),
            attempts: self.attempts.clone(),
            _connection_permit: self.connections.clone().try_acquire_owned().ok(),
            git_channels: Arc::clone(&self.git_channels),
            telemetry: self.telemetry.clone(),
            connection_id,
        }
    }
}

struct SshSession {
    authorized_keys: AuthorizedSshKeys,
    writable_keys: Arc<HashSet<PublicKey>>,
    audit: Arc<RequestAudit>,
    repositories: Option<Arc<GitRepositories>>,
    login: Option<LoginApprover>,
    protocol: ProtocolVersion,
    exec_channels: HashMap<ChannelId, ExecChannel>,
    authenticated_identity: Option<SshIdentity>,
    authenticated_key: Option<PublicKey>,
    authenticated_writer: bool,
    peer_address: IpAddr,
    attempts: AttemptLimiter<IpAddr>,
    _connection_permit: Option<OwnedSemaphorePermit>,
    git_channels: Arc<Semaphore>,
    telemetry: Telemetry,
    connection_id: String,
}

enum ExecChannel {
    Upload(Box<UploadChannel>),
    Receive(Box<ReceiveChannel>),
    IssueInput(IssueInputChannel),
    PullRequestCreate(PullRequestCreateChannel),
}

enum IssueInputKind {
    Create,
    Comment(i64),
}

struct IssueInputChannel {
    actor: String,
    owner: String,
    repository: String,
    output: CommandOutput,
    kind: IssueInputKind,
    input: Vec<u8>,
}

struct PullRequestCreateChannel {
    actor: String,
    owner: String,
    repository: String,
    base: String,
    head: String,
    output: CommandOutput,
    input: Vec<u8>,
}

struct UploadChannel {
    service: UploadPack,
    request: Vec<u8>,
    started_at: Instant,
    last_activity: Instant,
    transferred: u64,
    cancelled: Arc<AtomicBool>,
    _global_permit: OwnedSemaphorePermit,
}

struct ReceiveChannel {
    service: ReceivePack,
    owner: String,
    repository: String,
    identity: SshIdentity,
    public_key: PublicKey,
    authorized_keys: AuthorizedSshKeys,
    commands: Vec<u8>,
    commands_complete: bool,
    pack: tokio::fs::File,
    pack_bytes: u64,
    started_at: Instant,
    last_activity: Instant,
    transferred: u64,
    _global_permit: OwnedSemaphorePermit,
}

impl UploadChannel {
    fn expired(&self) -> bool {
        git_channel_expired(self.started_at, self.last_activity)
    }
}

impl Drop for UploadChannel {
    fn drop(&mut self) {
        self.cancelled.store(true, Ordering::Relaxed);
    }
}

impl ReceiveChannel {
    fn expired(&self) -> bool {
        git_channel_expired(self.started_at, self.last_activity)
    }
}

fn git_channel_expired(started_at: Instant, last_activity: Instant) -> bool {
    started_at.elapsed() > GIT_CHANNEL_WALL_CLOCK_LIMIT
        || last_activity.elapsed() > GIT_CHANNEL_INACTIVITY_LIMIT
}

fn reserve_git_channel(
    active_channels: usize,
    global_channels: &Arc<Semaphore>,
) -> Option<OwnedSemaphorePermit> {
    if active_channels >= MAX_GIT_CHANNELS_PER_CONNECTION {
        return None;
    }
    global_channels.clone().try_acquire_owned().ok()
}

impl Handler for SshSession {
    type Error = russh::Error;

    async fn auth_publickey_offered(
        &mut self,
        _user: &str,
        public_key: &PublicKey,
    ) -> Result<Auth, Self::Error> {
        Ok(self.authorize(public_key))
    }

    async fn auth_publickey(
        &mut self,
        _user: &str,
        public_key: &PublicKey,
    ) -> Result<Auth, Self::Error> {
        if self._connection_permit.is_none() || !self.attempts.allow(self.peer_address) {
            self.telemetry.ssh_auth(&self.connection_id, false);
            return Ok(Auth::reject());
        }
        if let Some(identity) = self.authorized_keys.identity(public_key) {
            self.authenticated_identity = Some(identity);
            self.authenticated_key = Some(public_key.clone());
            self.authenticated_writer = self.writable_keys.contains(public_key);
            self.telemetry.ssh_auth(&self.connection_id, true);
            Ok(Auth::Accept)
        } else {
            self.telemetry.ssh_auth(&self.connection_id, false);
            Ok(Auth::reject())
        }
    }

    async fn channel_open_session(
        &mut self,
        _channel: Channel<Msg>,
        reply: ChannelOpenHandle,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        reply.accept().await;
        Ok(())
    }

    async fn channel_open_x11(
        &mut self,
        _channel: Channel<Msg>,
        _originator_address: &str,
        _originator_port: u32,
        _reply: ChannelOpenHandle,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.audit.rejected_forward.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    async fn channel_open_direct_tcpip(
        &mut self,
        _channel: Channel<Msg>,
        _host_to_connect: &str,
        _port_to_connect: u32,
        _originator_address: &str,
        _originator_port: u32,
        _reply: ChannelOpenHandle,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.audit.rejected_forward.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    async fn channel_open_direct_streamlocal(
        &mut self,
        _channel: Channel<Msg>,
        _socket_path: &str,
        _reply: ChannelOpenHandle,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.audit.rejected_forward.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    async fn pty_request(
        &mut self,
        channel: ChannelId,
        _term: &str,
        _col_width: u32,
        _row_height: u32,
        _pix_width: u32,
        _pix_height: u32,
        _modes: &[(Pty, u32)],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.audit.rejected_pty.fetch_add(1, Ordering::Relaxed);
        session.channel_failure(channel)?;
        Ok(())
    }

    async fn x11_request(
        &mut self,
        channel: ChannelId,
        _single_connection: bool,
        _x11_auth_protocol: &str,
        _x11_auth_cookie: &str,
        _x11_screen_number: u32,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.audit.rejected_forward.fetch_add(1, Ordering::Relaxed);
        session.channel_failure(channel)?;
        Ok(())
    }

    async fn env_request(
        &mut self,
        channel: ChannelId,
        variable_name: &str,
        variable_value: &str,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        if variable_name == GIT_PROTOCOL_VARIABLE && valid_git_protocol(variable_value) {
            self.protocol = match variable_value {
                "version=0" => ProtocolVersion::V0,
                "version=1" => ProtocolVersion::V1,
                "version=2" => ProtocolVersion::V2,
                _ => unreachable!("the Git protocol value was validated"),
            };
            self.audit.accepted_env.fetch_add(1, Ordering::Relaxed);
            session.channel_success(channel)?;
        } else {
            self.audit.rejected_env.fetch_add(1, Ordering::Relaxed);
            session.channel_failure(channel)?;
        }
        Ok(())
    }

    async fn shell_request(
        &mut self,
        channel: ChannelId,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.audit.rejected_shell.fetch_add(1, Ordering::Relaxed);
        session.channel_failure(channel)?;
        session.close(channel)?;
        Ok(())
    }

    async fn exec_request(
        &mut self,
        channel: ChannelId,
        command: &[u8],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        let operation_id = format!("{:032x}", rand::random::<u128>());
        self.telemetry.ssh_operation(&operation_id);
        if command == VERSION_COMMAND {
            self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
            session.channel_success(channel)?;
            session.data(
                channel,
                format!("tit {}\n", env!("CARGO_PKG_VERSION")).into_bytes(),
            )?;
            session.exit_status_request(channel, 0)?;
            session.eof(channel)?;
            session.close(channel)?;
        } else if command == HELP_COMMAND {
            self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
            session.channel_success(channel)?;
            session.data(channel, HELP_TEXT.as_bytes())?;
            finish_git_channel(channel, 0, session)?;
        } else if command.starts_with(AUTH_COMMAND_PREFIX) {
            self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
            session.channel_success(channel)?;
            let result = match (
                parse_auth_command(command),
                self.active_identity(),
                self.login.clone(),
            ) {
                (Some(secret), Some(identity), Some(login)) => {
                    let username = identity.username;
                    let fingerprint = identity.fingerprint;
                    tokio::task::spawn_blocking(move || login(&secret, &username, &fingerprint))
                        .await
                        .ok()
                        .and_then(Result::ok)
                }
                _ => None,
            };
            match result {
                Some((origin, username)) => {
                    session.data(
                        channel,
                        format!(
                            "Authentication approved.\nOrigin: {}\nAccount: {}\nReturn to your browser and select Continue.\n",
                            origin, username
                        )
                        .into_bytes(),
                    )?;
                    finish_git_channel(channel, 0, session)?;
                }
                None => {
                    session.extended_data(
                        channel,
                        1,
                        format!(
                            "tit: The authentication approval is invalid or expired.\n{HELP_GUIDANCE}\n"
                        )
                        .into_bytes(),
                    )?;
                    finish_git_channel(channel, 1, session)?;
                }
            }
        } else if is_repository_command(command) {
            self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
            session.channel_success(channel)?;
            let machine_requested = requests_json(command);
            let result = match parse_repository_command(command) {
                Ok(command) => match self.active_identity() {
                    Some(identity) => {
                        run_repository_command(
                            self.repositories.clone(),
                            identity.username.clone(),
                            command,
                        )
                        .await
                    }
                    None => Err(RepositoryCommandError::Unavailable),
                },
                Err(()) => Err(RepositoryCommandError::Usage),
            };
            send_repository_command_result(channel, result, machine_requested, session)?;
        } else if is_issue_command(command) {
            self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
            session.channel_success(channel)?;
            let machine_requested = requests_json(command);
            match (parse_issue_command(command), self.active_identity()) {
                (Ok(IssueCommand::List(command)), Some(identity)) => {
                    let result =
                        run_issue_list(self.repositories.clone(), identity.username, command).await;
                    send_issue_list_result(channel, result, machine_requested, session)?;
                }
                (Ok(IssueCommand::Create(command)), Some(identity)) => {
                    self.exec_channels.insert(
                        channel,
                        ExecChannel::IssueInput(IssueInputChannel {
                            actor: identity.username,
                            owner: command.owner,
                            repository: command.repository,
                            output: command.output,
                            kind: IssueInputKind::Create,
                            input: Vec::new(),
                        }),
                    );
                }
                (Ok(IssueCommand::Comment(command)), Some(identity)) => {
                    self.exec_channels.insert(
                        channel,
                        ExecChannel::IssueInput(IssueInputChannel {
                            actor: identity.username,
                            owner: command.owner,
                            repository: command.repository,
                            output: command.output,
                            kind: IssueInputKind::Comment(command.number),
                            input: Vec::new(),
                        }),
                    );
                }
                (Ok(IssueCommand::SetState(command)), Some(identity)) => {
                    let result =
                        run_issue_set_state(self.repositories.clone(), identity.username, command)
                            .await;
                    send_issue_mutation_result(channel, result, machine_requested, session)?;
                }
                (Ok(_), None) => send_issue_error(
                    channel,
                    IssueCommandError::Unavailable,
                    machine_requested,
                    session,
                )?,
                (Err(()), _) => send_issue_error(
                    channel,
                    IssueCommandError::Usage,
                    machine_requested,
                    session,
                )?,
            }
        } else if is_pull_request_command(command) {
            self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
            session.channel_success(channel)?;
            let machine_requested = requests_json(command);
            match (parse_pull_request_command(command), self.active_identity()) {
                (Ok(PullRequestCommand::Checkout(command)), Some(identity)) => {
                    let result = run_pull_request_checkout(
                        self.repositories.clone(),
                        identity.username,
                        command,
                    )
                    .await;
                    send_pull_request_checkout_result(channel, result, machine_requested, session)?;
                }
                (Ok(PullRequestCommand::List(command)), Some(identity)) => {
                    let result = run_pull_request_list(
                        self.repositories.clone(),
                        identity.username,
                        command,
                    )
                    .await;
                    send_pull_request_list_result(channel, result, machine_requested, session)?;
                }
                (Ok(PullRequestCommand::Create(command)), Some(identity)) => {
                    self.exec_channels.insert(
                        channel,
                        ExecChannel::PullRequestCreate(PullRequestCreateChannel {
                            actor: identity.username,
                            owner: command.owner,
                            repository: command.repository,
                            base: command.base,
                            head: command.head,
                            output: command.output,
                            input: Vec::new(),
                        }),
                    );
                }
                (Ok(PullRequestCommand::SetState(command)), Some(identity)) => {
                    let result = run_pull_request_set_state(
                        self.repositories.clone(),
                        identity.username,
                        command,
                    )
                    .await;
                    send_pull_request_mutation_result(channel, result, machine_requested, session)?;
                }
                (Ok(_), None) => send_pull_request_error(
                    channel,
                    PullRequestCommandError::Unavailable,
                    machine_requested,
                    session,
                )?,
                (Err(()), _) => send_pull_request_error(
                    channel,
                    PullRequestCommandError::Usage,
                    machine_requested,
                    session,
                )?,
            }
        } else {
            let service = self.open_git_service(command).await;
            if let Some(service) = service {
                self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
                session.channel_success(channel)?;
                match service {
                    InitialGitService::Upload {
                        service,
                        advertisement,
                        global_permit,
                    } => {
                        let started_at = Instant::now();
                        let advertisement_bytes =
                            u64::try_from(advertisement.len()).unwrap_or(u64::MAX);
                        session.data(channel, advertisement)?;
                        self.exec_channels.insert(
                            channel,
                            ExecChannel::Upload(Box::new(UploadChannel {
                                service: *service,
                                request: Vec::new(),
                                started_at,
                                last_activity: started_at,
                                transferred: advertisement_bytes,
                                cancelled: Arc::new(AtomicBool::new(false)),
                                _global_permit: global_permit,
                            })),
                        );
                    }
                    InitialGitService::Receive(receive) => {
                        let InitialReceiveService {
                            service,
                            advertisement,
                            owner,
                            repository,
                            identity,
                            public_key,
                            global_permit,
                        } = *receive;
                        let started_at = Instant::now();
                        let advertisement_bytes =
                            u64::try_from(advertisement.len()).unwrap_or(u64::MAX);
                        session.data(channel, advertisement)?;
                        let pack = tokio::fs::File::create(service.incoming_pack()).await;
                        match pack {
                            Ok(pack) => {
                                self.exec_channels.insert(
                                    channel,
                                    ExecChannel::Receive(Box::new(ReceiveChannel {
                                        service: *service,
                                        owner,
                                        repository,
                                        identity,
                                        public_key,
                                        authorized_keys: self.authorized_keys.clone(),
                                        commands: Vec::new(),
                                        commands_complete: false,
                                        pack,
                                        pack_bytes: 0,
                                        started_at,
                                        last_activity: started_at,
                                        transferred: advertisement_bytes,
                                        _global_permit: global_permit,
                                    })),
                                );
                            }
                            Err(_) => fail_git_channel(channel, session)?,
                        }
                    }
                }
            } else {
                self.audit.accepted_exec.fetch_add(1, Ordering::Relaxed);
                session.channel_success(channel)?;
                if requests_json(command) {
                    session.data(
                        channel,
                        json_line(serde_json::json!({
                            "version": 1,
                            "status": "error",
                            "error": { "code": "invalid-command" },
                        })),
                    )?;
                } else {
                    session.extended_data(
                        channel,
                        1,
                        format!("tit: The command is not valid.\n{HELP_GUIDANCE}\n").into_bytes(),
                    )?;
                }
                finish_git_channel(channel, 1, session)?;
            }
        }
        Ok(())
    }

    async fn data(
        &mut self,
        channel: ChannelId,
        data: &[u8],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        let Some(exec) = self.exec_channels.remove(&channel) else {
            return Ok(());
        };
        let mut git = match exec {
            ExecChannel::IssueInput(mut issue) => {
                if append_issue_input(&mut issue.input, data).is_err() {
                    send_issue_error(
                        channel,
                        IssueCommandError::Input,
                        issue.output == CommandOutput::Json,
                        session,
                    )?;
                } else {
                    self.exec_channels
                        .insert(channel, ExecChannel::IssueInput(issue));
                }
                return Ok(());
            }
            ExecChannel::PullRequestCreate(mut pull_request) => {
                if append_issue_input(&mut pull_request.input, data).is_err() {
                    send_pull_request_error(
                        channel,
                        PullRequestCommandError::Input,
                        pull_request.output == CommandOutput::Json,
                        session,
                    )?;
                } else {
                    self.exec_channels
                        .insert(channel, ExecChannel::PullRequestCreate(pull_request));
                }
                return Ok(());
            }
            ExecChannel::Upload(git) => git,
            ExecChannel::Receive(mut git) => {
                if git.expired() {
                    fail_git_channel(channel, session)?;
                    return Ok(());
                }
                git.last_activity = Instant::now();
                git.transferred = git
                    .transferred
                    .saturating_add(u64::try_from(data.len()).unwrap_or(u64::MAX));
                if !git_channel_bytes_within_limit(git.transferred, 0) {
                    fail_git_channel(channel, session)?;
                    return Ok(());
                }
                if receive_data(&mut git, data).await.is_err() {
                    fail_git_channel(channel, session)?;
                } else if git.commands_complete
                    && matches!(git.service.expects_pack(&git.commands), Ok(false))
                {
                    send_receive_result(
                        channel,
                        finish_receive(self.repositories.clone(), git).await,
                        session,
                    )?;
                } else {
                    self.exec_channels
                        .insert(channel, ExecChannel::Receive(git));
                }
                return Ok(());
            }
        };
        if git.expired() {
            fail_git_channel(channel, session)?;
            return Ok(());
        }
        git.last_activity = Instant::now();
        let bytes = u64::try_from(data.len()).unwrap_or(u64::MAX);
        git.transferred = git.transferred.saturating_add(bytes);
        if !git_channel_bytes_within_limit(git.transferred, 0) {
            fail_git_channel(channel, session)?;
            return Ok(());
        }
        if git.request.len().saturating_add(data.len()) > MAX_REQUEST_BYTES {
            fail_git_channel(channel, session)?;
            return Ok(());
        }
        git.request.extend_from_slice(data);

        let packets = match decode(&git.request) {
            Ok(packets) => packets,
            Err(super::git::packetline::PacketLineError::TruncatedHeader)
            | Err(super::git::packetline::PacketLineError::TruncatedPacket) => {
                self.exec_channels.insert(channel, ExecChannel::Upload(git));
                return Ok(());
            }
            Err(_) => {
                fail_git_channel(channel, session)?;
                return Ok(());
            }
        };
        if packets == [Packet::Flush] {
            finish_git_channel(channel, 0, session)?;
            return Ok(());
        }

        match self.protocol {
            ProtocolVersion::V0 | ProtocolVersion::V1 => {
                let done = packets.last().is_some_and(
                    |packet| matches!(packet, Packet::Data(line) if trim_line(line) == b"done"),
                );
                if done {
                    match respond_git(self.repositories.clone(), self.protocol, git).await {
                        Some((git, Ok(response))) => {
                            if !git_response_within_limit(&git, response.len()) {
                                fail_git_channel(channel, session)?;
                                return Ok(());
                            }
                            session.data(channel, response)?;
                            finish_git_channel(channel, 0, session)?;
                        }
                        Some((_, Err(_))) | None => fail_git_channel(channel, session)?,
                    }
                } else {
                    self.exec_channels.insert(channel, ExecChannel::Upload(git));
                }
            }
            ProtocolVersion::V2 => {
                if packets.last() != Some(&Packet::Flush) {
                    self.exec_channels.insert(channel, ExecChannel::Upload(git));
                    return Ok(());
                }
                let fetch = packets.iter().any(
                    |packet| matches!(packet, Packet::Data(line) if trim_line(line) == b"command=fetch"),
                );
                let done = packets.iter().any(
                    |packet| matches!(packet, Packet::Data(line) if trim_line(line) == b"done"),
                );
                match respond_git(self.repositories.clone(), self.protocol, git).await {
                    Some((mut git, Ok(response))) => {
                        if !git_response_within_limit(&git, response.len()) {
                            fail_git_channel(channel, session)?;
                            return Ok(());
                        }
                        git.transferred = git
                            .transferred
                            .saturating_add(u64::try_from(response.len()).unwrap_or(u64::MAX));
                        session.data(channel, response)?;
                        if fetch && done {
                            finish_git_channel(channel, 0, session)?;
                        } else {
                            git.request.clear();
                            self.exec_channels.insert(channel, ExecChannel::Upload(git));
                        }
                    }
                    Some((_, Err(_))) | None => fail_git_channel(channel, session)?,
                }
            }
        }
        Ok(())
    }

    async fn channel_close(
        &mut self,
        channel: ChannelId,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.exec_channels.remove(&channel);
        Ok(())
    }

    async fn channel_eof(
        &mut self,
        channel: ChannelId,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        match self.exec_channels.remove(&channel) {
            Some(ExecChannel::Receive(git)) => {
                if !git.commands_complete || git.expired() {
                    fail_git_channel(channel, session)?;
                    return Ok(());
                }
                let result = finish_receive(self.repositories.clone(), git).await;
                send_receive_result(channel, result, session)?;
            }
            Some(ExecChannel::IssueInput(issue)) => {
                let active = self
                    .active_identity()
                    .is_some_and(|identity| identity.username == issue.actor);
                let machine_requested = issue.output == CommandOutput::Json;
                let result = if active {
                    run_issue_input(self.repositories.clone(), issue).await
                } else {
                    Err(IssueCommandError::Unavailable)
                };
                send_issue_mutation_result(channel, result, machine_requested, session)?;
            }
            Some(ExecChannel::PullRequestCreate(pull_request)) => {
                let active = self
                    .active_identity()
                    .is_some_and(|identity| identity.username == pull_request.actor);
                let machine_requested = pull_request.output == CommandOutput::Json;
                let result = if active {
                    run_pull_request_create(self.repositories.clone(), pull_request).await
                } else {
                    Err(PullRequestCommandError::Unavailable)
                };
                send_pull_request_mutation_result(channel, result, machine_requested, session)?;
            }
            Some(ExecChannel::Upload(_)) | None => {}
        }
        Ok(())
    }

    async fn subsystem_request(
        &mut self,
        channel: ChannelId,
        _name: &str,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.audit.rejected_exec.fetch_add(1, Ordering::Relaxed);
        session.channel_failure(channel)?;
        session.close(channel)?;
        Ok(())
    }

    async fn agent_request(
        &mut self,
        channel: ChannelId,
        session: &mut Session,
    ) -> Result<bool, Self::Error> {
        self.audit.rejected_agent.fetch_add(1, Ordering::Relaxed);
        session.channel_failure(channel)?;
        Ok(false)
    }

    async fn tcpip_forward(
        &mut self,
        _address: &str,
        _port: &mut u32,
        _session: &mut Session,
    ) -> Result<bool, Self::Error> {
        self.audit.rejected_forward.fetch_add(1, Ordering::Relaxed);
        Ok(false)
    }

    async fn streamlocal_forward(
        &mut self,
        _socket_path: &str,
        _session: &mut Session,
    ) -> Result<bool, Self::Error> {
        self.audit.rejected_forward.fetch_add(1, Ordering::Relaxed);
        Ok(false)
    }
}

#[derive(Clone, Copy, Eq, PartialEq)]
enum CommandOutput {
    Human,
    Json,
}

struct RepositoryCreateCommand {
    slug: String,
    output: CommandOutput,
}

fn is_repository_command(command: &[u8]) -> bool {
    command == b"repo" || command.starts_with(b"repo ")
}

fn requests_json(command: &[u8]) -> bool {
    std::str::from_utf8(command).is_ok_and(|command| {
        let tokens = command.split_ascii_whitespace().collect::<Vec<_>>();
        tokens
            .windows(2)
            .any(|tokens| tokens == ["--output", "json"])
    })
}

fn parse_repository_command(command: &[u8]) -> Result<RepositoryCreateCommand, ()> {
    if command.len() > MAX_REPOSITORY_COMMAND_BYTES || !command.is_ascii() {
        return Err(());
    }
    let command = std::str::from_utf8(command).map_err(|_| ())?;
    if command
        .bytes()
        .any(|byte| byte.is_ascii_control() && byte != b' ')
    {
        return Err(());
    }
    let mut tokens = command.split_ascii_whitespace();
    if tokens.next() != Some("repo") || tokens.next() != Some("create") {
        return Err(());
    }
    let slug = tokens.next().ok_or(())?.to_owned();
    let mut output = None;
    while let Some(option) = tokens.next() {
        let value = tokens.next().ok_or(())?;
        match option {
            "--output" if output.is_none() => {
                output = Some(match value {
                    "human" => CommandOutput::Human,
                    "json" => CommandOutput::Json,
                    _ => return Err(()),
                });
            }
            _ => return Err(()),
        }
    }
    Ok(RepositoryCreateCommand {
        slug,
        output: output.unwrap_or(CommandOutput::Human),
    })
}

enum RepositoryCommandError {
    Usage,
    Unavailable,
    Create(RepositoryServiceError),
}

async fn run_repository_command(
    repositories: Option<Arc<GitRepositories>>,
    actor: String,
    command: RepositoryCreateCommand,
) -> Result<(crate::store::RepositoryRecord, CommandOutput), RepositoryCommandError> {
    let repositories = repositories.ok_or(RepositoryCommandError::Unavailable)?;
    let database = repositories
        .push_database()
        .ok_or(RepositoryCommandError::Unavailable)?
        .to_owned();
    let root = repositories.repository_root().to_owned();
    let permit = repositories
        .blocking_permit()
        .await
        .map_err(|_| RepositoryCommandError::Unavailable)?;
    let output = command.output;
    let correlation_id = format!("{:032x}", rand::random::<u128>());
    tokio::task::spawn_blocking(move || {
        let _permit = permit;
        RepositoryService::new(&database, &root)
            .create_for_account(
                &actor,
                &command.slug,
                gix::hash::Kind::Sha1,
                &correlation_id,
            )
            .map(|repository| (repository, output))
            .map_err(RepositoryCommandError::Create)
    })
    .await
    .map_err(|_| RepositoryCommandError::Unavailable)?
}

fn send_repository_command_result(
    channel: ChannelId,
    result: Result<(crate::store::RepositoryRecord, CommandOutput), RepositoryCommandError>,
    machine_requested: bool,
    session: &mut Session,
) -> Result<(), russh::Error> {
    match result {
        Ok((repository, CommandOutput::Human)) => {
            session.data(
                channel,
                format!(
                    "Created repository {}/{}.\n",
                    repository.owner, repository.slug
                )
                .into_bytes(),
            )?;
            finish_git_channel(channel, 0, session)
        }
        Ok((repository, CommandOutput::Json)) => {
            session.data(
                channel,
                format!(
                    "{{\"version\":1,\"status\":\"success\",\"repository\":{{\"owner\":\"{}\",\"name\":\"{}\"}}}}\n",
                    repository.owner, repository.slug
                )
                .into_bytes(),
            )?;
            finish_git_channel(channel, 0, session)
        }
        Err(error) => {
            if machine_requested {
                session.data(
                    channel,
                    format!(
                        "{{\"version\":1,\"status\":\"error\",\"error\":{{\"code\":\"{}\"}}}}\n",
                        repository_command_error_code(&error)
                    )
                    .into_bytes(),
                )?;
            } else {
                session.extended_data(
                    channel,
                    1,
                    format!("tit: {}\n", repository_command_error_message(&error)).into_bytes(),
                )?;
            }
            finish_git_channel(channel, 1, session)
        }
    }
}

fn repository_command_error_code(error: &RepositoryCommandError) -> &'static str {
    match error {
        RepositoryCommandError::Usage => "invalid-command",
        RepositoryCommandError::Unavailable => "service-unavailable",
        RepositoryCommandError::Create(RepositoryServiceError::Store(
            crate::store::StoreError::RepositoryExists(_, _),
        )) => "repository-exists",
        RepositoryCommandError::Create(RepositoryServiceError::Auth(_))
        | RepositoryCommandError::Create(RepositoryServiceError::RepositoryName(_)) => {
            "invalid-name"
        }
        RepositoryCommandError::Create(RepositoryServiceError::Store(
            crate::store::StoreError::AccountNotFound(_),
        )) => "account-unavailable",
        RepositoryCommandError::Create(_) => "repository-create-failed",
    }
}

fn repository_command_error_message(error: &RepositoryCommandError) -> String {
    match repository_command_error_code(error) {
        "invalid-command" => {
            format!("usage: {REPOSITORY_CREATE_USAGE}\n{HELP_GUIDANCE}")
        }
        "repository-exists" => "A repository with this name already exists.".to_owned(),
        "invalid-name" => "The repository name is not valid.".to_owned(),
        "account-unavailable" => "The account is not active.".to_owned(),
        "service-unavailable" => "The repository service is not available.".to_owned(),
        _ => "The repository could not be created.".to_owned(),
    }
}

enum IssueCommand {
    List(IssueListCommand),
    Create(IssueCreateCommand),
    Comment(IssueNumberCommand),
    SetState(IssueStateCommand),
}

struct IssueListCommand {
    owner: String,
    repository: String,
    output: CommandOutput,
}

struct IssueCreateCommand {
    owner: String,
    repository: String,
    output: CommandOutput,
}

struct IssueNumberCommand {
    owner: String,
    repository: String,
    number: i64,
    output: CommandOutput,
}

struct IssueStateCommand {
    owner: String,
    repository: String,
    number: i64,
    state: &'static str,
    output: CommandOutput,
}

struct IssueListResult {
    repository: crate::store::RepositoryRecord,
    issues: Vec<crate::store::IssueRecord>,
    output: CommandOutput,
}

struct IssueMutationResult {
    owner: String,
    repository: String,
    number: i64,
    operation: &'static str,
    issue: Option<crate::store::IssueRecord>,
    comment_id: Option<String>,
    output: CommandOutput,
}

enum IssueCommandError {
    Usage,
    Input,
    Unavailable,
    Service(IssueError),
}

fn is_issue_command(command: &[u8]) -> bool {
    command == b"issue" || command.starts_with(b"issue ")
}

fn parse_issue_command(command: &[u8]) -> Result<IssueCommand, ()> {
    if command.len() > MAX_ISSUE_COMMAND_BYTES || !command.is_ascii() {
        return Err(());
    }
    let command = std::str::from_utf8(command).map_err(|_| ())?;
    if command
        .bytes()
        .any(|byte| byte.is_ascii_control() && byte != b' ')
    {
        return Err(());
    }
    let mut tokens = command.split_ascii_whitespace();
    if tokens.next() != Some("issue") {
        return Err(());
    }
    let operation = tokens.next().ok_or(())?;
    let target = tokens.next().ok_or(())?;
    let (owner, repository) = target.split_once('/').ok_or(())?;
    if owner.is_empty() || repository.is_empty() || repository.contains('/') {
        return Err(());
    }
    match operation {
        "list" | "create" => {
            let output = parse_output_options(tokens)?;
            if operation == "list" {
                Ok(IssueCommand::List(IssueListCommand {
                    owner: owner.to_owned(),
                    repository: repository.to_owned(),
                    output,
                }))
            } else {
                Ok(IssueCommand::Create(IssueCreateCommand {
                    owner: owner.to_owned(),
                    repository: repository.to_owned(),
                    output,
                }))
            }
        }
        "comment" | "close" | "reopen" => {
            let number = tokens.next().ok_or(())?.parse::<i64>().map_err(|_| ())?;
            if number < 1 {
                return Err(());
            }
            let output = parse_output_options(tokens)?;
            if operation == "comment" {
                Ok(IssueCommand::Comment(IssueNumberCommand {
                    owner: owner.to_owned(),
                    repository: repository.to_owned(),
                    number,
                    output,
                }))
            } else {
                Ok(IssueCommand::SetState(IssueStateCommand {
                    owner: owner.to_owned(),
                    repository: repository.to_owned(),
                    number,
                    state: if operation == "close" {
                        "closed"
                    } else {
                        "open"
                    },
                    output,
                }))
            }
        }
        _ => Err(()),
    }
}

fn parse_output_options<'a>(
    mut tokens: impl Iterator<Item = &'a str>,
) -> Result<CommandOutput, ()> {
    let mut output = None;
    while let Some(option) = tokens.next() {
        let value = tokens.next().ok_or(())?;
        if option != "--output" || output.is_some() {
            return Err(());
        }
        output = Some(match value {
            "human" => CommandOutput::Human,
            "json" => CommandOutput::Json,
            _ => return Err(()),
        });
    }
    Ok(output.unwrap_or(CommandOutput::Human))
}

async fn run_issue_list(
    repositories: Option<Arc<GitRepositories>>,
    actor: String,
    command: IssueListCommand,
) -> Result<IssueListResult, IssueCommandError> {
    let repositories = repositories.ok_or(IssueCommandError::Unavailable)?;
    let database = repositories
        .push_database()
        .ok_or(IssueCommandError::Unavailable)?
        .to_owned();
    let permit = repositories
        .blocking_permit()
        .await
        .map_err(|_| IssueCommandError::Unavailable)?;
    tokio::task::spawn_blocking(move || {
        let _permit = permit;
        IssueService::new(&database)
            .list(&command.owner, &command.repository, Some(&actor))
            .map(|(repository, issues)| IssueListResult {
                repository,
                issues,
                output: command.output,
            })
            .map_err(IssueCommandError::Service)
    })
    .await
    .map_err(|_| IssueCommandError::Unavailable)?
}

async fn run_issue_input(
    repositories: Option<Arc<GitRepositories>>,
    command: IssueInputChannel,
) -> Result<IssueMutationResult, IssueCommandError> {
    let repositories = repositories.ok_or(IssueCommandError::Unavailable)?;
    let database = repositories
        .push_database()
        .ok_or(IssueCommandError::Unavailable)?
        .to_owned();
    let permit = repositories
        .blocking_permit()
        .await
        .map_err(|_| IssueCommandError::Unavailable)?;
    tokio::task::spawn_blocking(move || {
        let _permit = permit;
        let service = IssueService::new(&database);
        match command.kind {
            IssueInputKind::Create => {
                let (title, body) = parse_issue_input(&command.input)?;
                service
                    .create(
                        &command.owner,
                        &command.repository,
                        &command.actor,
                        &title,
                        &body,
                    )
                    .map(|issue| IssueMutationResult {
                        owner: command.owner,
                        repository: command.repository,
                        number: issue.number,
                        operation: "created",
                        issue: Some(issue),
                        comment_id: None,
                        output: command.output,
                    })
                    .map_err(IssueCommandError::Service)
            }
            IssueInputKind::Comment(number) => {
                let body =
                    std::str::from_utf8(&command.input).map_err(|_| IssueCommandError::Input)?;
                service
                    .comment(
                        &command.owner,
                        &command.repository,
                        number,
                        &command.actor,
                        body,
                    )
                    .map(|comment_id| IssueMutationResult {
                        owner: command.owner,
                        repository: command.repository,
                        number,
                        operation: "commented",
                        issue: None,
                        comment_id: Some(comment_id),
                        output: command.output,
                    })
                    .map_err(IssueCommandError::Service)
            }
        }
    })
    .await
    .map_err(|_| IssueCommandError::Unavailable)?
}

async fn run_issue_set_state(
    repositories: Option<Arc<GitRepositories>>,
    actor: String,
    command: IssueStateCommand,
) -> Result<IssueMutationResult, IssueCommandError> {
    let repositories = repositories.ok_or(IssueCommandError::Unavailable)?;
    let database = repositories
        .push_database()
        .ok_or(IssueCommandError::Unavailable)?
        .to_owned();
    let permit = repositories
        .blocking_permit()
        .await
        .map_err(|_| IssueCommandError::Unavailable)?;
    tokio::task::spawn_blocking(move || {
        let _permit = permit;
        IssueService::new(&database)
            .set_state(
                &command.owner,
                &command.repository,
                command.number,
                &actor,
                command.state,
            )
            .map(|()| IssueMutationResult {
                owner: command.owner,
                repository: command.repository,
                number: command.number,
                operation: if command.state == "closed" {
                    "closed"
                } else {
                    "reopened"
                },
                issue: None,
                comment_id: None,
                output: command.output,
            })
            .map_err(IssueCommandError::Service)
    })
    .await
    .map_err(|_| IssueCommandError::Unavailable)?
}

fn parse_issue_input(input: &[u8]) -> Result<(String, String), IssueCommandError> {
    let (title, body) = input
        .iter()
        .position(|byte| *byte == b'\n')
        .map_or((input, &[][..]), |end| (&input[..end], &input[end + 1..]));
    let title = title.strip_suffix(b"\r").unwrap_or(title);
    Ok((
        std::str::from_utf8(title)
            .map_err(|_| IssueCommandError::Input)?
            .to_owned(),
        std::str::from_utf8(body)
            .map_err(|_| IssueCommandError::Input)?
            .to_owned(),
    ))
}

fn append_issue_input(input: &mut Vec<u8>, data: &[u8]) -> Result<(), ()> {
    if input.len().saturating_add(data.len()) > MAX_ISSUE_INPUT_BYTES {
        return Err(());
    }
    input.extend_from_slice(data);
    Ok(())
}

fn send_issue_list_result(
    channel: ChannelId,
    result: Result<IssueListResult, IssueCommandError>,
    machine_requested: bool,
    session: &mut Session,
) -> Result<(), russh::Error> {
    match result {
        Ok(result) => {
            let data = match result.output {
                CommandOutput::Human => issue_list_human(&result),
                CommandOutput::Json => issue_list_json(&result),
            };
            session.data(channel, data)?;
            finish_git_channel(channel, 0, session)
        }
        Err(error) => send_issue_error(channel, error, machine_requested, session),
    }
}

fn issue_list_human(result: &IssueListResult) -> Vec<u8> {
    if result.issues.is_empty() {
        return b"No issues.\n".to_vec();
    }
    let mut output = String::new();
    for issue in &result.issues {
        output.push_str(&format!(
            "#{} {} {}\n",
            issue.number, issue.state, issue.title
        ));
    }
    output.into_bytes()
}

fn issue_list_json(result: &IssueListResult) -> Vec<u8> {
    let issues = result
        .issues
        .iter()
        .map(|issue| {
            serde_json::json!({
                "number": issue.number,
                "title": issue.title,
                "state": issue.state,
                "author": issue.author,
                "created_at": issue.created_at,
                "updated_at": issue.updated_at,
            })
        })
        .collect::<Vec<_>>();
    json_line(serde_json::json!({
        "version": 1,
        "status": "success",
        "repository": {
            "owner": result.repository.owner,
            "name": result.repository.slug,
        },
        "issues": issues,
    }))
}

fn send_issue_mutation_result(
    channel: ChannelId,
    result: Result<IssueMutationResult, IssueCommandError>,
    machine_requested: bool,
    session: &mut Session,
) -> Result<(), russh::Error> {
    match result {
        Ok(result) => {
            let data = match result.output {
                CommandOutput::Human => format!(
                    "{} issue {}/{}#{}.\n",
                    sentence_case(result.operation),
                    result.owner,
                    result.repository,
                    result.number
                )
                .into_bytes(),
                CommandOutput::Json => json_line(serde_json::json!({
                    "version": 1,
                    "status": "success",
                    "repository": {
                        "owner": result.owner,
                        "name": result.repository,
                    },
                    "issue": {
                        "number": result.number,
                        "title": result.issue.as_ref().map(|issue| &issue.title),
                        "state": result.issue.as_ref().map(|issue| &issue.state),
                        "author": result.issue.as_ref().map(|issue| &issue.author),
                        "created_at": result.issue.as_ref().map(|issue| issue.created_at),
                        "updated_at": result.issue.as_ref().map(|issue| issue.updated_at),
                    },
                    "operation": result.operation,
                    "comment_id": result.comment_id,
                })),
            };
            session.data(channel, data)?;
            finish_git_channel(channel, 0, session)
        }
        Err(error) => send_issue_error(channel, error, machine_requested, session),
    }
}

fn sentence_case(value: &str) -> String {
    let mut characters = value.chars();
    match characters.next() {
        Some(first) => first.to_uppercase().chain(characters).collect(),
        None => String::new(),
    }
}

fn send_issue_error(
    channel: ChannelId,
    error: IssueCommandError,
    machine_requested: bool,
    session: &mut Session,
) -> Result<(), russh::Error> {
    if machine_requested {
        session.data(
            channel,
            json_line(serde_json::json!({
                "version": 1,
                "status": "error",
                "error": { "code": issue_command_error_code(&error) },
            })),
        )?;
    } else {
        session.extended_data(
            channel,
            1,
            format!("tit: {}\n", issue_command_error_message(&error)).into_bytes(),
        )?;
    }
    finish_git_channel(channel, 1, session)
}

fn json_line(value: serde_json::Value) -> Vec<u8> {
    let mut output = serde_json::to_vec(&value).expect("a JSON value can be serialized");
    output.push(b'\n');
    output
}

fn issue_command_error_code(error: &IssueCommandError) -> &'static str {
    match error {
        IssueCommandError::Usage => "invalid-command",
        IssueCommandError::Input
        | IssueCommandError::Service(IssueError::Title | IssueError::Body) => "invalid-input",
        IssueCommandError::Service(IssueError::Auth(_) | IssueError::RepositoryName(_)) => {
            "invalid-target"
        }
        IssueCommandError::Service(IssueError::Store(
            crate::store::StoreError::RepositoryNotFound(_, _)
            | crate::store::StoreError::IssueHidden,
        )) => "repository-unavailable",
        IssueCommandError::Service(IssueError::Store(crate::store::StoreError::IssueDenied)) => {
            "permission-denied"
        }
        IssueCommandError::Service(IssueError::Store(
            crate::store::StoreError::AccountNotFound(_),
        )) => "account-unavailable",
        IssueCommandError::Unavailable => "service-unavailable",
        IssueCommandError::Service(_) => "issue-command-failed",
    }
}

fn issue_command_error_message(error: &IssueCommandError) -> String {
    match issue_command_error_code(error) {
        "invalid-command" => {
            format!(
                "usage: {ISSUE_LIST_USAGE}, {ISSUE_CREATE_USAGE}, {ISSUE_COMMENT_USAGE}, \
                 {ISSUE_CLOSE_USAGE}, or {ISSUE_REOPEN_USAGE}\n{HELP_GUIDANCE}"
            )
        }
        "invalid-input" => {
            "The first input line must be a valid title. The remaining input is the body."
                .to_owned()
        }
        "invalid-target" => "The repository name is not valid.".to_owned(),
        "repository-unavailable" => "The repository is not available.".to_owned(),
        "permission-denied" => "The account cannot create an issue in this repository.".to_owned(),
        "account-unavailable" => "The account is not active.".to_owned(),
        "service-unavailable" => "The issue service is not available.".to_owned(),
        _ => "The issue command could not be completed.".to_owned(),
    }
}

struct PullRequestCheckoutCommand {
    owner: String,
    repository: String,
    number: i64,
    output: CommandOutput,
}

enum PullRequestCommand {
    Checkout(PullRequestCheckoutCommand),
    List(PullRequestListCommand),
    Create(PullRequestCreateCommand),
    SetState(PullRequestStateCommand),
}

struct PullRequestListCommand {
    owner: String,
    repository: String,
    state: String,
    output: CommandOutput,
}

struct PullRequestCreateCommand {
    owner: String,
    repository: String,
    base: String,
    head: String,
    output: CommandOutput,
}

struct PullRequestStateCommand {
    owner: String,
    repository: String,
    number: i64,
    state: &'static str,
    output: CommandOutput,
}

struct PullRequestCheckoutResult {
    owner: String,
    repository: String,
    number: i64,
    output: CommandOutput,
}

struct PullRequestListResult {
    repository: crate::store::RepositoryRecord,
    pull_requests: Vec<crate::store::PullRequestRecord>,
    state: String,
    output: CommandOutput,
}

struct PullRequestMutationResult {
    owner: String,
    repository: String,
    number: i64,
    operation: &'static str,
    pull_request: Option<crate::store::PullRequestRecord>,
    output: CommandOutput,
}

enum PullRequestCommandError {
    Usage,
    Input,
    Unavailable,
    Store(StoreError),
    Service(PullRequestError),
}

fn is_pull_request_command(command: &[u8]) -> bool {
    command == b"pr" || command.starts_with(b"pr ")
}

fn parse_pull_request_command(command: &[u8]) -> Result<PullRequestCommand, ()> {
    if command.len() > MAX_PULL_REQUEST_COMMAND_BYTES || !command.is_ascii() {
        return Err(());
    }
    let command = std::str::from_utf8(command).map_err(|_| ())?;
    if command
        .bytes()
        .any(|byte| byte.is_ascii_control() && byte != b' ')
    {
        return Err(());
    }
    let mut tokens = command.split_ascii_whitespace();
    if tokens.next() != Some("pr") {
        return Err(());
    }
    let operation = tokens.next().ok_or(())?;
    let target = tokens.next().ok_or(())?;
    let (owner, repository) = target.split_once('/').ok_or(())?;
    if owner.is_empty() || repository.is_empty() || repository.contains('/') {
        return Err(());
    }
    match operation {
        "list" => {
            let (state, output) = parse_pull_request_list_options(tokens)?;
            Ok(PullRequestCommand::List(PullRequestListCommand {
                owner: owner.to_owned(),
                repository: repository.to_owned(),
                state,
                output,
            }))
        }
        "create" => {
            let base = tokens.next().ok_or(())?.to_owned();
            let head = tokens.next().ok_or(())?.to_owned();
            let output = parse_output_options(tokens)?;
            Ok(PullRequestCommand::Create(PullRequestCreateCommand {
                owner: owner.to_owned(),
                repository: repository.to_owned(),
                base,
                head,
                output,
            }))
        }
        "checkout" | "close" | "reopen" => {
            let number = tokens.next().ok_or(())?.parse::<i64>().map_err(|_| ())?;
            if number < 1 {
                return Err(());
            }
            let output = parse_output_options(tokens)?;
            if operation == "checkout" {
                Ok(PullRequestCommand::Checkout(PullRequestCheckoutCommand {
                    owner: owner.to_owned(),
                    repository: repository.to_owned(),
                    number,
                    output,
                }))
            } else {
                Ok(PullRequestCommand::SetState(PullRequestStateCommand {
                    owner: owner.to_owned(),
                    repository: repository.to_owned(),
                    number,
                    state: if operation == "close" {
                        "closed"
                    } else {
                        "open"
                    },
                    output,
                }))
            }
        }
        _ => Err(()),
    }
}

fn parse_pull_request_list_options<'a>(
    mut tokens: impl Iterator<Item = &'a str>,
) -> Result<(String, CommandOutput), ()> {
    let mut state = None;
    let mut output = None;
    while let Some(option) = tokens.next() {
        let value = tokens.next().ok_or(())?;
        match option {
            "--state" if state.is_none() => {
                if !matches!(value, "open" | "closed" | "merged" | "all") {
                    return Err(());
                }
                state = Some(value.to_owned());
            }
            "--output" if output.is_none() => {
                output = Some(match value {
                    "human" => CommandOutput::Human,
                    "json" => CommandOutput::Json,
                    _ => return Err(()),
                });
            }
            _ => return Err(()),
        }
    }
    Ok((
        state.unwrap_or_else(|| "open".to_owned()),
        output.unwrap_or(CommandOutput::Human),
    ))
}

async fn run_pull_request_checkout(
    repositories: Option<Arc<GitRepositories>>,
    actor: String,
    command: PullRequestCheckoutCommand,
) -> Result<PullRequestCheckoutResult, PullRequestCommandError> {
    let repositories = repositories.ok_or(PullRequestCommandError::Unavailable)?;
    let database = repositories
        .push_database()
        .ok_or(PullRequestCommandError::Unavailable)?
        .to_owned();
    let permit = repositories
        .blocking_permit()
        .await
        .map_err(|_| PullRequestCommandError::Unavailable)?;
    tokio::task::spawn_blocking(move || {
        let _permit = permit;
        Store::open(&database)
            .and_then(|store| {
                store.pull_request(
                    &command.owner,
                    &command.repository,
                    command.number,
                    Some(&actor),
                )
            })
            .map(|_| PullRequestCheckoutResult {
                owner: command.owner,
                repository: command.repository,
                number: command.number,
                output: command.output,
            })
            .map_err(PullRequestCommandError::Store)
    })
    .await
    .map_err(|_| PullRequestCommandError::Unavailable)?
}

async fn run_pull_request_list(
    repositories: Option<Arc<GitRepositories>>,
    actor: String,
    command: PullRequestListCommand,
) -> Result<PullRequestListResult, PullRequestCommandError> {
    let repositories = repositories.ok_or(PullRequestCommandError::Unavailable)?;
    let database = repositories
        .push_database()
        .ok_or(PullRequestCommandError::Unavailable)?
        .to_owned();
    let root = repositories.repository_root().to_owned();
    let permit = repositories
        .blocking_permit()
        .await
        .map_err(|_| PullRequestCommandError::Unavailable)?;
    tokio::task::spawn_blocking(move || {
        let _permit = permit;
        PullRequestService::new(&database, &root)
            .list_page(
                &command.owner,
                &command.repository,
                Some(&actor),
                &command.state,
                1,
            )
            .map(|(repository, page, _)| PullRequestListResult {
                repository,
                pull_requests: page.items,
                state: command.state,
                output: command.output,
            })
            .map_err(PullRequestCommandError::Service)
    })
    .await
    .map_err(|_| PullRequestCommandError::Unavailable)?
}

async fn run_pull_request_create(
    repositories: Option<Arc<GitRepositories>>,
    command: PullRequestCreateChannel,
) -> Result<PullRequestMutationResult, PullRequestCommandError> {
    let (title, body) =
        parse_issue_input(&command.input).map_err(|_| PullRequestCommandError::Input)?;
    let repositories = repositories.ok_or(PullRequestCommandError::Unavailable)?;
    let database = repositories
        .push_database()
        .ok_or(PullRequestCommandError::Unavailable)?
        .to_owned();
    let root = repositories.repository_root().to_owned();
    let permit = repositories
        .blocking_permit()
        .await
        .map_err(|_| PullRequestCommandError::Unavailable)?;
    tokio::task::spawn_blocking(move || {
        let _permit = permit;
        PullRequestService::new(&database, &root)
            .open(
                &command.owner,
                &command.repository,
                &command.actor,
                &title,
                &body,
                &command.base,
                &command.head,
            )
            .map(|pull_request| PullRequestMutationResult {
                owner: command.owner,
                repository: command.repository,
                number: pull_request.number,
                operation: "created",
                pull_request: Some(pull_request),
                output: command.output,
            })
            .map_err(PullRequestCommandError::Service)
    })
    .await
    .map_err(|_| PullRequestCommandError::Unavailable)?
}

async fn run_pull_request_set_state(
    repositories: Option<Arc<GitRepositories>>,
    actor: String,
    command: PullRequestStateCommand,
) -> Result<PullRequestMutationResult, PullRequestCommandError> {
    let repositories = repositories.ok_or(PullRequestCommandError::Unavailable)?;
    let database = repositories
        .push_database()
        .ok_or(PullRequestCommandError::Unavailable)?
        .to_owned();
    let root = repositories.repository_root().to_owned();
    let permit = repositories
        .blocking_permit()
        .await
        .map_err(|_| PullRequestCommandError::Unavailable)?;
    tokio::task::spawn_blocking(move || {
        let _permit = permit;
        PullRequestService::new(&database, &root)
            .set_state(
                &command.owner,
                &command.repository,
                command.number,
                &actor,
                command.state,
            )
            .map(|()| PullRequestMutationResult {
                owner: command.owner,
                repository: command.repository,
                number: command.number,
                operation: if command.state == "closed" {
                    "closed"
                } else {
                    "reopened"
                },
                pull_request: None,
                output: command.output,
            })
            .map_err(PullRequestCommandError::Service)
    })
    .await
    .map_err(|_| PullRequestCommandError::Unavailable)?
}

fn send_pull_request_checkout_result(
    channel: ChannelId,
    result: Result<PullRequestCheckoutResult, PullRequestCommandError>,
    machine_requested: bool,
    session: &mut Session,
) -> Result<(), russh::Error> {
    match result {
        Ok(result) => {
            let fetch = format!(
                "git fetch origin refs/pull/{}/head:refs/heads/pr-{}",
                result.number, result.number
            );
            let checkout = format!("git checkout pr-{}", result.number);
            let data = match result.output {
                CommandOutput::Human => format!("{fetch}\n{checkout}\n"),
                CommandOutput::Json => format!(
                    "{{\"version\":1,\"status\":\"success\",\"repository\":{{\"owner\":\"{}\",\"name\":\"{}\"}},\"pull_request\":{{\"number\":{},\"ref\":\"refs/pull/{}/head\",\"local_branch\":\"pr-{}\"}},\"commands\":{{\"fetch\":\"{}\",\"checkout\":\"{}\"}}}}\n",
                    result.owner,
                    result.repository,
                    result.number,
                    result.number,
                    result.number,
                    fetch,
                    checkout,
                ),
            };
            session.data(channel, data.into_bytes())?;
            finish_git_channel(channel, 0, session)
        }
        Err(error) => send_pull_request_error(channel, error, machine_requested, session),
    }
}

fn send_pull_request_list_result(
    channel: ChannelId,
    result: Result<PullRequestListResult, PullRequestCommandError>,
    machine_requested: bool,
    session: &mut Session,
) -> Result<(), russh::Error> {
    match result {
        Ok(result) => {
            let data = match result.output {
                CommandOutput::Human => {
                    if result.pull_requests.is_empty() {
                        b"No pull requests.\n".to_vec()
                    } else {
                        result
                            .pull_requests
                            .iter()
                            .map(|pull_request| {
                                format!(
                                    "#{} {} {} ({} <- {})\n",
                                    pull_request.number,
                                    pull_request.state,
                                    pull_request.title,
                                    pull_request.base_ref,
                                    pull_request.head_ref
                                )
                            })
                            .collect::<String>()
                            .into_bytes()
                    }
                }
                CommandOutput::Json => {
                    let pull_requests = result
                        .pull_requests
                        .iter()
                        .map(pull_request_json)
                        .collect::<Vec<_>>();
                    json_line(serde_json::json!({
                        "version": 1,
                        "status": "success",
                        "repository": {
                            "owner": result.repository.owner,
                            "name": result.repository.slug,
                        },
                        "state": result.state,
                        "pull_requests": pull_requests,
                    }))
                }
            };
            session.data(channel, data)?;
            finish_git_channel(channel, 0, session)
        }
        Err(error) => send_pull_request_error(channel, error, machine_requested, session),
    }
}

fn send_pull_request_mutation_result(
    channel: ChannelId,
    result: Result<PullRequestMutationResult, PullRequestCommandError>,
    machine_requested: bool,
    session: &mut Session,
) -> Result<(), russh::Error> {
    match result {
        Ok(result) => {
            let data = match result.output {
                CommandOutput::Human => format!(
                    "{} pull request {}/{}#{}.\n",
                    sentence_case(result.operation),
                    result.owner,
                    result.repository,
                    result.number
                )
                .into_bytes(),
                CommandOutput::Json => json_line(serde_json::json!({
                    "version": 1,
                    "status": "success",
                    "repository": {
                        "owner": result.owner,
                        "name": result.repository,
                    },
                    "pull_request": result.pull_request.as_ref().map(pull_request_json)
                        .unwrap_or_else(|| serde_json::json!({ "number": result.number })),
                    "operation": result.operation,
                })),
            };
            session.data(channel, data)?;
            finish_git_channel(channel, 0, session)
        }
        Err(error) => send_pull_request_error(channel, error, machine_requested, session),
    }
}

fn pull_request_json(pull_request: &crate::store::PullRequestRecord) -> serde_json::Value {
    serde_json::json!({
        "number": pull_request.number,
        "title": pull_request.title,
        "state": pull_request.state,
        "author": pull_request.author,
        "base": pull_request.base_ref,
        "head": pull_request.head_ref,
        "created_at": pull_request.created_at,
        "updated_at": pull_request.updated_at,
    })
}

fn send_pull_request_error(
    channel: ChannelId,
    error: PullRequestCommandError,
    machine_requested: bool,
    session: &mut Session,
) -> Result<(), russh::Error> {
    if machine_requested {
        session.data(
            channel,
            json_line(serde_json::json!({
                "version": 1,
                "status": "error",
                "error": { "code": pull_request_command_error_code(&error) },
            })),
        )?;
    } else {
        session.extended_data(
            channel,
            1,
            format!("tit: {}\n", pull_request_command_error_message(&error)).into_bytes(),
        )?;
    }
    finish_git_channel(channel, 1, session)
}

fn pull_request_command_error_code(error: &PullRequestCommandError) -> &'static str {
    match error {
        PullRequestCommandError::Usage => "invalid-command",
        PullRequestCommandError::Input
        | PullRequestCommandError::Service(
            PullRequestError::Title | PullRequestError::Body | PullRequestError::Branch,
        ) => "invalid-input",
        PullRequestCommandError::Unavailable => "service-unavailable",
        PullRequestCommandError::Store(
            StoreError::PullRequestHidden
            | StoreError::PullRequestNotFound(_, _, _)
            | StoreError::RepositoryNotFound(_, _),
        ) => "pull-request-unavailable",
        PullRequestCommandError::Service(PullRequestError::Auth(_))
        | PullRequestCommandError::Service(PullRequestError::RepositoryName(_)) => "invalid-target",
        PullRequestCommandError::Service(PullRequestError::Store(
            StoreError::PullRequestHidden
            | StoreError::PullRequestNotFound(_, _, _)
            | StoreError::RepositoryNotFound(_, _),
        )) => "pull-request-unavailable",
        PullRequestCommandError::Service(PullRequestError::Store(
            StoreError::PullRequestDenied,
        )) => "permission-denied",
        PullRequestCommandError::Store(_) | PullRequestCommandError::Service(_) => {
            "pull-request-command-failed"
        }
    }
}

fn pull_request_command_error_message(error: &PullRequestCommandError) -> String {
    match pull_request_command_error_code(error) {
        "invalid-command" => {
            format!(
                "usage: {PULL_REQUEST_LIST_USAGE}, {PULL_REQUEST_CREATE_USAGE}, \
                 {PULL_REQUEST_CLOSE_USAGE}, {PULL_REQUEST_REOPEN_USAGE}, or \
                 {PULL_REQUEST_CHECKOUT_USAGE}\n{HELP_GUIDANCE}"
            )
        }
        "invalid-input" => {
            "The refs or input are not valid. The first input line is the title and the remaining input is the body."
                .to_owned()
        }
        "pull-request-unavailable" => "The pull request is not available.".to_owned(),
        "invalid-target" => "The pull-request target is not valid.".to_owned(),
        "permission-denied" => {
            "The account cannot change a pull request in this repository.".to_owned()
        }
        "service-unavailable" => "The pull-request service is not available.".to_owned(),
        _ => "The pull-request command could not be completed.".to_owned(),
    }
}

impl SshSession {
    fn authorize(&self, public_key: &PublicKey) -> Auth {
        if self.authorized_keys.contains(public_key) {
            Auth::Accept
        } else {
            Auth::reject()
        }
    }

    async fn open_git_service(&mut self, command: &[u8]) -> Option<InitialGitService> {
        let active_channels = self
            .exec_channels
            .values()
            .filter(|channel| matches!(channel, ExecChannel::Upload(_) | ExecChannel::Receive(_)))
            .count();
        let global_permit = reserve_git_channel(active_channels, &self.git_channels)?;
        let repositories = self.repositories.as_ref()?;
        let identity = self.active_identity()?;
        let service = repositories
            .resolve_ssh_service_for(Some(&identity.username), command)
            .ok()?;
        match service {
            GitSshService::Upload { path, .. } => {
                let permit = repositories.blocking_permit().await.ok()?;
                let protocol = self.protocol;
                tokio::task::spawn_blocking(move || {
                    let _permit = permit;
                    let service = UploadPack::open(&path)?;
                    let advertisement = service.advertisement(protocol, false)?;
                    Ok::<_, UploadPackError>(InitialGitService::Upload {
                        service: Box::new(service),
                        advertisement,
                        global_permit,
                    })
                })
                .await
                .ok()?
                .ok()
            }
            GitSshService::Receive {
                path,
                owner,
                repository,
            } => {
                if !repositories.uses_policy() && !self.authenticated_writer {
                    return None;
                }
                let database = repositories.push_database()?.to_owned();
                let actor = identity.username.clone();
                let public_key = self.authenticated_key.clone()?;
                let uses_policy = repositories.uses_policy();
                let permit = repositories.blocking_permit().await.ok()?;
                tokio::task::spawn_blocking(move || {
                    let _permit = permit;
                    let service = if uses_policy {
                        ReceivePack::open_authorized(
                            &path,
                            &database,
                            actor,
                            owner.clone(),
                            repository.clone(),
                        )?
                    } else {
                        ReceivePack::open(&path, &database, actor)?
                    };
                    let advertisement = service.advertisement()?;
                    Ok::<_, ReceivePackError>(InitialGitService::Receive(Box::new(
                        InitialReceiveService {
                            service: Box::new(service),
                            advertisement,
                            owner,
                            repository,
                            identity,
                            public_key,
                            global_permit,
                        },
                    )))
                })
                .await
                .ok()?
                .ok()
            }
        }
    }

    fn active_identity(&self) -> Option<SshIdentity> {
        let public_key = self.authenticated_key.as_ref()?;
        let authenticated = self.authenticated_identity.as_ref()?;
        let current = self.authorized_keys.identity(public_key)?;
        (current == *authenticated).then_some(current)
    }
}

enum InitialGitService {
    Upload {
        service: Box<UploadPack>,
        advertisement: Vec<u8>,
        global_permit: OwnedSemaphorePermit,
    },
    Receive(Box<InitialReceiveService>),
}

struct InitialReceiveService {
    service: Box<ReceivePack>,
    advertisement: Vec<u8>,
    owner: String,
    repository: String,
    identity: SshIdentity,
    public_key: PublicKey,
    global_permit: OwnedSemaphorePermit,
}

async fn receive_data(git: &mut ReceiveChannel, data: &[u8]) -> Result<(), ()> {
    if git.commands_complete {
        write_receive_pack(git, data).await?;
        return Ok(());
    }
    git.commands.extend_from_slice(data);
    let boundary = first_flush_end(&git.commands).map_err(|_| ())?;
    let Some(boundary) = boundary else {
        return Ok(());
    };
    let pack = git.commands.split_off(boundary);
    git.commands_complete = true;
    write_receive_pack(git, &pack).await
}

async fn write_receive_pack(git: &mut ReceiveChannel, data: &[u8]) -> Result<(), ()> {
    let bytes = u64::try_from(data.len()).map_err(|_| ())?;
    git.pack_bytes = git.pack_bytes.checked_add(bytes).ok_or(())?;
    if git.pack_bytes > MAX_RECEIVE_PACK_BYTES {
        return Err(());
    }
    git.pack.write_all(data).await.map_err(|_| ())
}

type ReceiveResult = Option<Result<Vec<u8>, (ReceivePackError, Vec<u8>)>>;

async fn finish_receive(
    repositories: Option<Arc<GitRepositories>>,
    git: Box<ReceiveChannel>,
) -> ReceiveResult {
    let ReceiveChannel {
        mut service,
        owner,
        repository,
        identity,
        public_key,
        authorized_keys,
        commands,
        mut pack,
        started_at,
        ..
    } = *git;
    if started_at.elapsed() > GIT_CHANNEL_WALL_CLOCK_LIMIT - Duration::from_secs(30) {
        return None;
    }
    pack.flush().await.ok()?;
    pack.sync_all().await.ok()?;
    drop(pack);
    let repositories = repositories?;
    let push_permit = repositories.push_permit().await.ok()?;
    let permit = repositories.blocking_permit().await.ok()?;
    let maintenance = repositories.mutation_permit().await;
    tokio::task::spawn_blocking(move || {
        let _permit = permit;
        let _push_permit = push_permit;
        let _maintenance = maintenance;
        if authorized_keys.identity(&public_key).as_ref() != Some(&identity)
            || !repositories.authorize(
                &identity.username,
                &owner,
                &repository,
                RepositoryOperation::Write,
            )
        {
            return None;
        }
        Some(match service.finish(&commands) {
            Ok(response) => Ok(response),
            Err(error) => {
                let response = service.rejection_response(&commands, &error);
                match service.record_rejection() {
                    Ok(()) => Err((error, response)),
                    Err(audit_error) => {
                        let response = service.rejection_response(&commands, &audit_error);
                        Err((audit_error, response))
                    }
                }
            }
        })
    })
    .await
    .ok()
    .flatten()
}

fn send_receive_result(
    channel: ChannelId,
    result: ReceiveResult,
    session: &mut Session,
) -> Result<(), russh::Error> {
    match result {
        Some(Ok(response)) => {
            session.data(channel, response)?;
            finish_git_channel(channel, 0, session)
        }
        Some(Err((error, response))) => {
            eprintln!("tit: receive-pack failed: {error}");
            session.data(
                channel,
                if response.is_empty() {
                    receive_error_response()
                } else {
                    response
                },
            )?;
            finish_git_channel(channel, 1, session)
        }
        None => {
            session.data(channel, receive_error_response())?;
            finish_git_channel(channel, 1, session)
        }
    }
}

fn receive_error_response() -> Vec<u8> {
    let mut response = Vec::new();
    let _ = encode_data(b"unpack tit rejected the push\n", &mut response);
    super::git::packetline::encode_flush(&mut response);
    response
}

async fn respond_git(
    repositories: Option<Arc<GitRepositories>>,
    protocol: ProtocolVersion,
    git: Box<UploadChannel>,
) -> Option<(Box<UploadChannel>, Result<Vec<u8>, UploadPackError>)> {
    let remaining = GIT_CHANNEL_WALL_CLOCK_LIMIT.checked_sub(git.started_at.elapsed())?;
    let cancelled = Arc::clone(&git.cancelled);
    let permit = repositories?.blocking_permit().await.ok()?;
    let cancellation_for_job = Arc::clone(&cancelled);
    let mut task = tokio::task::spawn_blocking(move || {
        let _permit = permit;
        let response =
            git.service
                .respond_with_cancellation(protocol, &git.request, &cancellation_for_job);
        (git, response)
    });
    match tokio::time::timeout(remaining, &mut task).await {
        Ok(result) => result.ok(),
        Err(_) => {
            cancelled.store(true, Ordering::Relaxed);
            task.await.ok()
        }
    }
}

fn git_response_within_limit(git: &UploadChannel, response_bytes: usize) -> bool {
    let response_bytes = u64::try_from(response_bytes).unwrap_or(u64::MAX);
    git_channel_bytes_within_limit(git.transferred, response_bytes)
}

fn git_channel_bytes_within_limit(transferred: u64, additional: u64) -> bool {
    transferred.saturating_add(additional) <= MAX_GIT_CHANNEL_BYTES
}

fn valid_git_protocol(value: &str) -> bool {
    matches!(value, "version=0" | "version=1" | "version=2")
}

fn trim_line(line: &[u8]) -> &[u8] {
    line.strip_suffix(b"\n").unwrap_or(line)
}

fn finish_git_channel(
    channel: ChannelId,
    status: u32,
    session: &mut Session,
) -> Result<(), russh::Error> {
    session.exit_status_request(channel, status)?;
    session.eof(channel)?;
    session.close(channel)?;
    Ok(())
}

fn fail_git_channel(channel: ChannelId, session: &mut Session) -> Result<(), russh::Error> {
    let mut error = Vec::new();
    encode_data(b"ERR invalid Git request\n", &mut error)
        .expect("a Git error packet is within the limit");
    session.data(channel, error)?;
    finish_git_channel(channel, 1, session)
}

#[derive(Default)]
struct RequestAudit {
    accepted_env: AtomicUsize,
    rejected_env: AtomicUsize,
    accepted_exec: AtomicUsize,
    rejected_exec: AtomicUsize,
    rejected_shell: AtomicUsize,
    rejected_pty: AtomicUsize,
    rejected_agent: AtomicUsize,
    rejected_forward: AtomicUsize,
}

impl RequestAudit {
    fn snapshot(&self) -> RequestAuditSnapshot {
        RequestAuditSnapshot {
            accepted_env: self.accepted_env.load(Ordering::Relaxed),
            rejected_env: self.rejected_env.load(Ordering::Relaxed),
            accepted_exec: self.accepted_exec.load(Ordering::Relaxed),
            rejected_exec: self.rejected_exec.load(Ordering::Relaxed),
            rejected_shell: self.rejected_shell.load(Ordering::Relaxed),
            rejected_pty: self.rejected_pty.load(Ordering::Relaxed),
            rejected_agent: self.rejected_agent.load(Ordering::Relaxed),
            rejected_forward: self.rejected_forward.load(Ordering::Relaxed),
        }
    }
}

#[derive(Clone, Copy, Debug, Default)]
pub(crate) struct RequestAuditSnapshot {
    pub(crate) accepted_env: usize,
    pub(crate) rejected_env: usize,
    pub(crate) accepted_exec: usize,
    pub(crate) rejected_exec: usize,
    pub(crate) rejected_shell: usize,
    pub(crate) rejected_pty: usize,
    pub(crate) rejected_agent: usize,
    pub(crate) rejected_forward: usize,
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parses_only_bounded_issue_commands_and_input() {
        assert!(matches!(
            parse_issue_command(b"issue list alice/project --output json"),
            Ok(IssueCommand::List(_))
        ));
        assert!(matches!(
            parse_issue_command(b"issue create alice/project"),
            Ok(IssueCommand::Create(_))
        ));
        for command in [
            b"issue list alice/project --output json extra".as_slice(),
            b"issue list alice/project/extra".as_slice(),
            b"issue create alice/project --output json --output json".as_slice(),
            &[b'x'; MAX_ISSUE_COMMAND_BYTES + 1],
        ] {
            assert!(parse_issue_command(command).is_err());
        }

        let mut input = Vec::new();
        assert!(append_issue_input(&mut input, &[b'x'; MAX_ISSUE_INPUT_BYTES]).is_ok());
        assert!(append_issue_input(&mut input, b"x").is_err());
        assert!(matches!(
            parse_issue_input(b"invalid\xff"),
            Err(IssueCommandError::Input)
        ));
    }

    #[test]
    fn parses_only_bounded_pull_request_commands() {
        let command = parse_pull_request_command(b"pr checkout alice/project 42 --output json")
            .expect("parse a pull-request checkout command");
        let PullRequestCommand::Checkout(command) = command else {
            panic!("expected checkout");
        };
        assert_eq!(command.owner, "alice");
        assert_eq!(command.repository, "project");
        assert_eq!(command.number, 42);
        assert!(command.output == CommandOutput::Json);
        assert!(matches!(
            parse_pull_request_command(
                b"pr list alice/project --output json --state merged"
            ),
            Ok(PullRequestCommand::List(PullRequestListCommand {
                state,
                output: CommandOutput::Json,
                ..
            })) if state == "merged"
        ));
        assert!(matches!(
            parse_pull_request_command(b"pr create alice/project main topic"),
            Ok(PullRequestCommand::Create(_))
        ));
        assert!(matches!(
            parse_pull_request_command(b"pr close alice/project 2"),
            Ok(PullRequestCommand::SetState(_))
        ));
        for command in [
            b"pr checkout alice/project 0".as_slice(),
            b"pr checkout alice/project not-a-number".as_slice(),
            b"pr checkout alice/project/extra 1".as_slice(),
            b"pr checkout alice/project 1 --output json extra".as_slice(),
            b"pr checkout alice/project 1 --output json --output json".as_slice(),
            b"pr list alice/project --state invalid".as_slice(),
            b"pr create alice/project main".as_slice(),
            &[b'x'; MAX_PULL_REQUEST_COMMAND_BYTES + 1],
        ] {
            assert!(parse_pull_request_command(command).is_err());
        }
    }

    #[test]
    fn bounds_git_channel_time_bytes_and_concurrency() {
        let now = Instant::now();
        assert!(!git_channel_expired(now, now));
        assert!(git_channel_expired(
            now - GIT_CHANNEL_WALL_CLOCK_LIMIT - Duration::from_secs(1),
            now
        ));
        assert!(git_channel_expired(
            now,
            now - GIT_CHANNEL_INACTIVITY_LIMIT - Duration::from_secs(1)
        ));

        let global = Arc::new(Semaphore::new(MAX_GIT_CHANNELS_GLOBAL));
        let permits = (0..MAX_GIT_CHANNELS_GLOBAL)
            .map(|_| reserve_git_channel(0, &global).expect("reserve a global Git channel"))
            .collect::<Vec<_>>();
        assert!(reserve_git_channel(0, &global).is_none());
        drop(permits);
        assert!(reserve_git_channel(MAX_GIT_CHANNELS_PER_CONNECTION, &global).is_none());

        let limit = std::hint::black_box(MAX_GIT_CHANNEL_BYTES);
        assert!(git_channel_bytes_within_limit(limit, 0));
        assert!(!git_channel_bytes_within_limit(limit, 1));
    }
}
