use std::collections::{HashMap, HashSet};
use std::fs::{self, File};
use std::io::{BufReader, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use gix::bstr::ByteSlice;
use gix::hash::{Kind, ObjectId};
use gix::objs::{CommitRef, Kind as ObjectKind, TagRef, TreeRefIter};
use gix::refs::transaction::{Change, LogChange, PreviousValue, RefEdit, RefLog};
use gix::refs::{FullName, Target};
use rand::TryRng;
use thiserror::Error;

use super::packetline::{Packet, PacketLineError, decode, encode_data, encode_flush};
use super::upload_pack::hash_name;
use crate::policy::{PolicyError, RefChange, RepositoryPolicy};
use crate::store::{GitIntentRecord, GitOperationIntent, NewAuditEvent, Store};

const MAX_COMMANDS: usize = 256;
const MAX_OBJECTS: usize = 100_000;
const MAX_OBJECT_BYTES: usize = 64 * 1024 * 1024;
const MAX_PACK_BYTES: u64 = 256 * 1024 * 1024;
const MAX_WALK_OBJECTS: usize = 500_000;
const MAX_DELTA_DEPTH: usize = 64;
const MAX_PROCESSING_TIME: Duration = Duration::from_secs(30);

pub(crate) struct ReceivePack {
    repository_path: PathBuf,
    database_path: PathBuf,
    actor: String,
    object_format: Kind,
    intent_id: String,
    quarantine: PathBuf,
    incoming_pack: PathBuf,
    cleanup_on_drop: bool,
    policy_context: Option<(String, String)>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct RefCommand {
    old: ObjectId,
    new: ObjectId,
    name: FullName,
}

impl ReceivePack {
    pub(crate) fn open(
        repository_path: &Path,
        database_path: &Path,
        actor: String,
    ) -> Result<Self, ReceivePackError> {
        Self::open_inner(repository_path, database_path, actor, None)
    }

    pub(crate) fn open_authorized(
        repository_path: &Path,
        database_path: &Path,
        actor: String,
        owner: String,
        repository: String,
    ) -> Result<Self, ReceivePackError> {
        Self::open_inner(
            repository_path,
            database_path,
            actor,
            Some((owner, repository)),
        )
    }

    fn open_inner(
        repository_path: &Path,
        database_path: &Path,
        actor: String,
        policy_context: Option<(String, String)>,
    ) -> Result<Self, ReceivePackError> {
        let repository = open_bare(repository_path)?;
        let object_format = repository.object_hash();
        let intent_id = random_id()?;
        let quarantine = repository_path
            .join("objects")
            .join("tit-quarantine")
            .join(&intent_id);
        fs::create_dir_all(quarantine.join("pack"))?;
        let incoming_pack = quarantine.join("incoming.pack");
        Ok(Self {
            repository_path: repository_path.to_owned(),
            database_path: database_path.to_owned(),
            actor,
            object_format,
            intent_id,
            quarantine,
            incoming_pack,
            cleanup_on_drop: true,
            policy_context,
        })
    }

    pub(crate) fn incoming_pack(&self) -> &Path {
        &self.incoming_pack
    }

    pub(crate) fn advertisement(&self) -> Result<Vec<u8>, ReceivePackError> {
        let repository = open_bare(&self.repository_path)?;
        let mut references = repository
            .references()
            .map_err(|error| ReceivePackError::Repository(error.to_string()))?
            .all()
            .map_err(|error| ReceivePackError::Repository(error.to_string()))?
            .filter_map(|reference| reference.ok())
            .filter_map(|reference| {
                let id = reference.try_id()?.detach();
                Some((reference.name().as_bstr().to_vec(), id))
            })
            .collect::<Vec<_>>();
        references.sort_by(|left, right| left.0.cmp(&right.0));
        let capabilities = format!(
            "report-status report-status-v2 delete-refs atomic ofs-delta object-format={} agent=tit/{}",
            hash_name(self.object_format),
            env!("CARGO_PKG_VERSION")
        );
        let mut output = Vec::new();
        if references.is_empty() {
            encode_data(
                format!(
                    "{} capabilities^{{}}\0{capabilities}\n",
                    self.object_format.null()
                )
                .as_bytes(),
                &mut output,
            )?;
        } else {
            for (index, (name, id)) in references.iter().enumerate() {
                let suffix = if index == 0 {
                    format!("\0{capabilities}")
                } else {
                    String::new()
                };
                encode_data(
                    format!("{id} {}{suffix}\n", String::from_utf8_lossy(name)).as_bytes(),
                    &mut output,
                )?;
            }
        }
        encode_flush(&mut output);
        Ok(output)
    }

    pub(crate) fn expects_pack(&self, command_bytes: &[u8]) -> Result<bool, ReceivePackError> {
        Ok(parse_commands(command_bytes, self.object_format)?
            .iter()
            .any(|command| !command.new.is_null()))
    }

    pub(crate) fn finish(&mut self, command_bytes: &[u8]) -> Result<Vec<u8>, ReceivePackError> {
        let commands = parse_commands(command_bytes, self.object_format)?;
        let repository = open_bare(&self.repository_path)?;
        validate_initial_refs(&repository, &commands)?;

        let initial = serialize_refs(&commands, false);
        let proposed = serialize_refs(&commands, true);
        let repository_text = path_text(&self.repository_path)?;
        let quarantine_text = path_text(&self.quarantine)?;
        let created_at = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_err(|_| ReceivePackError::Clock)?
            .as_secs()
            .try_into()
            .map_err(|_| ReceivePackError::Clock)?;
        let mut store = Store::open(&self.database_path)?;
        store.begin_git_intent(&GitOperationIntent {
            id: &self.intent_id,
            repository_path: repository_text,
            actor: &self.actor,
            initial_refs: &initial,
            proposed_refs: &proposed,
            event_payload: &proposed,
            quarantine_path: quarantine_text,
            created_at,
        })?;
        self.cleanup_on_drop = false;
        crash_point("intent");

        let result = (|| {
            let has_pack =
                self.incoming_pack.exists() && fs::metadata(&self.incoming_pack)?.len() != 0;
            let pack_name = if has_pack {
                self.index_and_validate_pack(&repository, &commands)?
            } else {
                let changes = validate_proposed_objects(&repository, None, &commands)?;
                self.validate_policy(&commands, &changes)?;
                None
            };

            let pack_name = match pack_name.as_ref() {
                Some(path) => Some(
                    path.file_name()
                        .and_then(|name| name.to_str())
                        .ok_or(ReceivePackError::Path)?,
                ),
                None => None,
            };
            store.mark_git_objects_promoted(&self.intent_id, pack_name)?;
            crash_point("objects");
            apply_ref_transaction(&repository, &commands)?;
            crash_point("refs");
            store.complete_git_intent(&self.intent_id)?;
            let _ = fs::remove_dir_all(&self.quarantine);
            crash_point("completed");
            Ok(status_response(&commands, None)?)
        })();
        if result.is_err() {
            drop(store);
            recover_incomplete_pushes(&self.database_path)?;
            self.cleanup_on_drop = true;
        }
        result
    }

    pub(crate) fn record_rejection(&self) -> Result<(), ReceivePackError> {
        let store = Store::open(&self.database_path)?;
        if store.git_intent_completed(&self.intent_id)? {
            return Ok(());
        }
        let target = self
            .repository_path
            .file_name()
            .and_then(|name| name.to_str())
            .and_then(|name| name.strip_suffix(".git"))
            .unwrap_or("repository");
        let created_at = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_err(|_| ReceivePackError::Clock)?
            .as_secs()
            .try_into()
            .map_err(|_| ReceivePackError::Clock)?;
        store.record_audit_event(&NewAuditEvent {
            action: "ref.update",
            actor: &self.actor,
            target,
            outcome: "failure",
            correlation_id: &self.intent_id,
            created_at,
        })?;
        Ok(())
    }

    pub(crate) fn rejection_response(
        &self,
        command_bytes: &[u8],
        error: &ReceivePackError,
    ) -> Vec<u8> {
        let Ok(commands) = parse_commands(command_bytes, self.object_format) else {
            return Vec::new();
        };
        let mut output = Vec::new();
        let unpack = if error.is_unpack_error() {
            format!("unpack {}\n", error.client_reason())
        } else {
            "unpack ok\n".to_owned()
        };
        if encode_data(unpack.as_bytes(), &mut output).is_err() {
            return Vec::new();
        }
        for command in commands {
            let name = String::from_utf8_lossy(command.name.as_bstr());
            let line = format!("ng {name} {}\n", error.client_reason());
            if encode_data(line.as_bytes(), &mut output).is_err() {
                return Vec::new();
            }
        }
        encode_flush(&mut output);
        output
    }

    fn index_and_validate_pack(
        &self,
        repository: &gix::Repository,
        commands: &[RefCommand],
    ) -> Result<Option<PathBuf>, ReceivePackError> {
        let deadline = Instant::now() + MAX_PROCESSING_TIME;
        let metadata = fs::metadata(&self.incoming_pack)?;
        if metadata.len() == 0 || metadata.len() > MAX_PACK_BYTES {
            return Err(ReceivePackError::PackLimit);
        }
        validate_pack_allocation_limits(&self.incoming_pack, self.object_format)?;
        let mut reader = BufReader::new(File::open(&self.incoming_pack)?);
        let mut progress = gix::progress::Discard;
        let interrupt = Arc::new(AtomicBool::new(false));
        let timer_interrupt = Arc::clone(&interrupt);
        let (done_sender, done_receiver) = std::sync::mpsc::channel();
        let remaining = deadline.saturating_duration_since(Instant::now());
        let timer = std::thread::spawn(move || {
            if done_receiver.recv_timeout(remaining).is_err() {
                timer_interrupt.store(true, Ordering::Relaxed);
            }
        });
        let outcome = gix_pack::Bundle::write_to_directory(
            &mut reader,
            Some(self.quarantine.join("pack").as_path()),
            &mut progress,
            &interrupt,
            Some(Box::new(repository.objects.clone())),
            gix_pack::bundle::write::Options {
                thread_limit: Some(2),
                iteration_mode: gix_pack::data::input::Mode::Verify,
                index_version: gix_pack::index::Version::V2,
                object_hash: self.object_format,
            },
        );
        let _ = done_sender.send(());
        let _ = timer.join();
        let outcome = outcome.map_err(|error| {
            if interrupt.load(Ordering::Relaxed) {
                ReceivePackError::WallClockLimit
            } else {
                ReceivePackError::Pack(error.to_string())
            }
        })?;
        if outcome.index.num_objects > MAX_OBJECTS as u32 {
            return Err(ReceivePackError::ObjectLimit);
        }
        check_deadline(deadline)?;
        if outcome.index.num_objects == 0 {
            let changes = validate_proposed_objects(repository, None, commands)?;
            check_deadline(deadline)?;
            self.validate_policy(commands, &changes)?;
            check_deadline(deadline)?;
            if let Some(path) = outcome.data_path {
                let _ = fs::remove_file(path);
            }
            if let Some(path) = outcome.index_path {
                let _ = fs::remove_file(path);
            }
            if let Some(path) = outcome.keep_path {
                let _ = fs::remove_file(path);
            }
            return Ok(None);
        }
        let mut bundle = outcome
            .to_bundle()
            .ok_or(ReceivePackError::MissingPack)?
            .map_err(|error| ReceivePackError::Pack(error.to_string()))?;
        bundle.pack = bundle.pack.with_alloc_limit_bytes(Some(MAX_OBJECT_BYTES));
        validate_delta_depth(&bundle)?;
        check_deadline(deadline)?;
        let changes = validate_proposed_objects(repository, Some(&bundle), commands)?;
        check_deadline(deadline)?;
        self.validate_policy(commands, &changes)?;
        check_deadline(deadline)?;

        let source_pack = outcome.data_path.ok_or(ReceivePackError::MissingPack)?;
        let source_index = outcome.index_path.ok_or(ReceivePackError::MissingPack)?;
        if fs::metadata(&source_pack)?.len() > MAX_PACK_BYTES {
            return Err(ReceivePackError::PackLimit);
        }
        let destination = self.repository_path.join("objects/pack");
        fs::create_dir_all(&destination)?;
        check_deadline(deadline)?;
        let destination_pack =
            destination.join(source_pack.file_name().ok_or(ReceivePackError::Path)?);
        let destination_index =
            destination.join(source_index.file_name().ok_or(ReceivePackError::Path)?);
        let pack_exists = destination_pack.exists();
        let index_exists = destination_index.exists();
        if pack_exists != index_exists {
            return Err(ReceivePackError::Repository(
                "a pack file and its index do not match".to_owned(),
            ));
        }
        let promoted = if pack_exists {
            fs::remove_file(&source_pack)?;
            fs::remove_file(&source_index)?;
            false
        } else {
            fs::rename(&source_index, &destination_index)?;
            if let Err(error) = fs::rename(&source_pack, &destination_pack) {
                let _ = fs::remove_file(&destination_index);
                return Err(error.into());
            }
            true
        };
        if let Some(keep) = outcome.keep_path {
            let _ = fs::remove_file(keep);
        }
        sync_directory(&destination)?;
        check_deadline(deadline)?;
        Ok(promoted.then_some(destination_pack))
    }

    fn validate_policy(
        &self,
        commands: &[RefCommand],
        changes: &[RefChange],
    ) -> Result<(), ReceivePackError> {
        let Some((owner, repository)) = &self.policy_context else {
            if changes
                .iter()
                .any(|change| matches!(change, RefChange::Force))
            {
                return Err(ReceivePackError::NonFastForward);
            }
            return Ok(());
        };
        let policy = RepositoryPolicy::new(&self.database_path);
        for (command, change) in commands.iter().zip(changes) {
            let authorization = policy.authorize_ref_change(
                &self.actor,
                owner,
                repository,
                command.name.as_bstr(),
                *change,
            );
            match authorization {
                Ok(_) => {}
                Err(PolicyError::Denied) if matches!(change, RefChange::Force) => {
                    return Err(ReceivePackError::NonFastForward);
                }
                Err(PolicyError::Denied) => return Err(ReceivePackError::RefPolicy),
                Err(error) => return Err(ReceivePackError::Policy(error)),
            }
        }
        Ok(())
    }
}

fn validate_pack_allocation_limits(
    path: &Path,
    object_format: Kind,
) -> Result<(), ReceivePackError> {
    let pack = gix_pack::data::File::at(path, object_format)
        .map_err(|error| ReceivePackError::Pack(error.to_string()))?
        .with_alloc_limit_bytes(Some(MAX_OBJECT_BYTES));
    if pack.num_objects() > MAX_OBJECTS as u32 {
        return Err(ReceivePackError::ObjectLimit);
    }
    let entries = pack
        .streaming_iter()
        .map_err(|error| ReceivePackError::Pack(error.to_string()))?;
    let mut inflate = gix::features::zlib::Inflate::default();
    for entry in entries {
        let entry = entry.map_err(|error| ReceivePackError::Pack(error.to_string()))?;
        if entry.decompressed_size > MAX_OBJECT_BYTES as u64 {
            return Err(ReceivePackError::ObjectLimit);
        }
        if !matches!(
            entry.header,
            gix_pack::data::entry::Header::OfsDelta { .. }
                | gix_pack::data::entry::Header::RefDelta { .. }
        ) {
            continue;
        }
        let pack_entry = pack
            .entry(entry.pack_offset)
            .map_err(|error| ReceivePackError::Pack(error.to_string()))?;
        let compressed_end = pack_entry
            .data_offset
            .checked_add(entry.compressed_size)
            .ok_or(ReceivePackError::PackLimit)?;
        let compressed = pack
            .entry_slice(pack_entry.data_offset..compressed_end)
            .ok_or(ReceivePackError::PackLimit)?;
        inflate.reset();
        let mut header = [0_u8; 20];
        let (_, _, written) = inflate
            .once(compressed, &mut header)
            .map_err(|error| ReceivePackError::Pack(error.to_string()))?;
        let (_, base_end) = decode_delta_size(&header[..written])?;
        let (result_size, _) = decode_delta_size(&header[base_end..written])?;
        if result_size > MAX_OBJECT_BYTES as u64 {
            return Err(ReceivePackError::ObjectLimit);
        }
    }
    Ok(())
}

fn decode_delta_size(input: &[u8]) -> Result<(u64, usize), ReceivePackError> {
    let mut size = 0_u64;
    let mut shift = 0_u32;
    for (index, byte) in input.iter().copied().enumerate() {
        if shift >= u64::BITS {
            return Err(ReceivePackError::Pack(
                "delta object size is invalid".to_owned(),
            ));
        }
        size |= u64::from(byte & 0x7f) << shift;
        if byte & 0x80 == 0 {
            return Ok((size, index + 1));
        }
        shift += 7;
    }
    Err(ReceivePackError::Pack(
        "delta object size is incomplete".to_owned(),
    ))
}

fn validate_delta_depth(bundle: &gix_pack::Bundle) -> Result<(), ReceivePackError> {
    let entries = bundle.index.iter().collect::<Vec<_>>();
    let offsets_by_id = entries
        .iter()
        .map(|entry| (entry.oid, entry.pack_offset))
        .collect::<HashMap<_, _>>();
    let headers = entries
        .iter()
        .map(|entry| {
            bundle
                .pack
                .entry(entry.pack_offset)
                .map(|pack_entry| (entry.pack_offset, pack_entry.header))
                .map_err(|error| ReceivePackError::Pack(error.to_string()))
        })
        .collect::<Result<HashMap<_, _>, _>>()?;
    let mut depths = HashMap::new();
    for offset in headers.keys().copied() {
        let depth = delta_depth(offset, &headers, &offsets_by_id, &mut depths)?;
        if depth > MAX_DELTA_DEPTH {
            return Err(ReceivePackError::DeltaDepthLimit);
        }
    }
    Ok(())
}

fn delta_depth(
    offset: u64,
    headers: &HashMap<u64, gix_pack::data::entry::Header>,
    offsets_by_id: &HashMap<ObjectId, u64>,
    depths: &mut HashMap<u64, usize>,
) -> Result<usize, ReceivePackError> {
    if let Some(depth) = depths.get(&offset) {
        return Ok(*depth);
    }
    let mut path = Vec::new();
    let mut seen = HashSet::new();
    let mut current = offset;
    let base_depth = loop {
        if let Some(depth) = depths.get(&current) {
            break *depth;
        }
        if !seen.insert(current) || path.len() > MAX_DELTA_DEPTH {
            return Err(ReceivePackError::DeltaDepthLimit);
        }
        path.push(current);
        let header = headers
            .get(&current)
            .ok_or(ReceivePackError::RecoveryData)?;
        match header {
            gix_pack::data::entry::Header::OfsDelta { base_distance } => {
                current = current
                    .checked_sub(*base_distance)
                    .ok_or(ReceivePackError::DeltaDepthLimit)?;
            }
            gix_pack::data::entry::Header::RefDelta { base_id } => {
                let Some(base) = offsets_by_id.get(base_id) else {
                    break 0;
                };
                current = *base;
            }
            _ => break 0,
        }
    };
    let mut depth = base_depth;
    for item in path.into_iter().rev() {
        let header = headers.get(&item).ok_or(ReceivePackError::RecoveryData)?;
        if matches!(
            header,
            gix_pack::data::entry::Header::OfsDelta { .. }
                | gix_pack::data::entry::Header::RefDelta { .. }
        ) {
            depth += 1;
        }
        if depth > MAX_DELTA_DEPTH {
            return Err(ReceivePackError::DeltaDepthLimit);
        }
        depths.insert(item, depth);
    }
    depths
        .get(&offset)
        .copied()
        .ok_or(ReceivePackError::RecoveryData)
}

fn check_deadline(deadline: Instant) -> Result<(), ReceivePackError> {
    if Instant::now() >= deadline {
        Err(ReceivePackError::WallClockLimit)
    } else {
        Ok(())
    }
}

impl Drop for ReceivePack {
    fn drop(&mut self) {
        if self.cleanup_on_drop {
            let _ = fs::remove_dir_all(&self.quarantine);
        }
    }
}

pub(crate) fn recover_incomplete_pushes(database_path: &Path) -> Result<(), ReceivePackError> {
    let mut store = Store::open(database_path)?;
    for intent in store.incomplete_git_intents()? {
        recover_intent(&mut store, &intent)?;
    }
    Ok(())
}

fn recover_intent(store: &mut Store, intent: &GitIntentRecord) -> Result<(), ReceivePackError> {
    let repository_path = Path::new(&intent.repository_path);
    let repository = open_bare(repository_path)?;
    let initial = parse_ref_snapshot(&intent.initial_refs, repository.object_hash())?;
    let proposed = parse_ref_snapshot(&intent.proposed_refs, repository.object_hash())?;
    let at_initial = refs_match(&repository, &initial)?;
    let at_proposed = refs_match(&repository, &proposed)?;

    match (intent.state.as_str(), at_initial, at_proposed) {
        ("pending", true, false) | ("pending", true, true) => {
            store.abandon_git_intent(&intent.id)?;
        }
        ("pending", false, false) if initial.len() == 1 => {
            store.abandon_git_intent(&intent.id)?;
        }
        ("promoted", false, true) => {
            store.complete_git_intent(&intent.id)?;
        }
        ("promoted", true, false) | ("promoted", true, true) => {
            remove_promoted_pack(repository_path, intent.pack_name.as_deref())?;
            store.abandon_git_intent(&intent.id)?;
        }
        ("promoted", false, false) if initial.len() == 1 => {
            remove_promoted_pack(repository_path, intent.pack_name.as_deref())?;
            store.abandon_git_intent(&intent.id)?;
        }
        _ => return Err(ReceivePackError::MixedRecovery(intent.id.clone())),
    }
    let quarantine = Path::new(&intent.quarantine_path);
    let expected_parent = repository_path.join("objects/tit-quarantine");
    if quarantine.parent() != Some(expected_parent.as_path())
        || quarantine.file_name().and_then(|name| name.to_str()) != Some(intent.id.as_str())
    {
        return Err(ReceivePackError::RecoveryData);
    }
    let _ = fs::remove_dir_all(quarantine);
    Ok(())
}

fn parse_ref_snapshot(
    bytes: &[u8],
    object_format: Kind,
) -> Result<Vec<(FullName, ObjectId)>, ReceivePackError> {
    bytes
        .split(|byte| *byte == b'\n')
        .filter(|line| !line.is_empty())
        .map(|line| {
            let mut fields = line.splitn(2, |byte| *byte == b' ');
            let id = parse_id(fields.next(), object_format)?;
            let name = fields.next().ok_or(ReceivePackError::RecoveryData)?;
            let name =
                FullName::try_from(name.as_bstr()).map_err(|_| ReceivePackError::RecoveryData)?;
            Ok((name, id))
        })
        .collect()
}

fn refs_match(
    repository: &gix::Repository,
    expected: &[(FullName, ObjectId)],
) -> Result<bool, ReceivePackError> {
    for (name, expected_id) in expected {
        let current = repository
            .try_find_reference(name)
            .map_err(|error| ReceivePackError::Repository(error.to_string()))?
            .and_then(|reference| reference.try_id().map(gix::Id::detach));
        if (expected_id.is_null() && current.is_some())
            || (!expected_id.is_null() && current != Some(*expected_id))
        {
            return Ok(false);
        }
    }
    Ok(true)
}

fn remove_promoted_pack(
    repository_path: &Path,
    pack_name: Option<&str>,
) -> Result<(), ReceivePackError> {
    let Some(pack_name) = pack_name else {
        return Ok(());
    };
    if Path::new(pack_name)
        .file_name()
        .and_then(|name| name.to_str())
        != Some(pack_name)
        || !pack_name.starts_with("pack-")
        || !pack_name.ends_with(".pack")
    {
        return Err(ReceivePackError::RecoveryData);
    }
    let pack = repository_path.join("objects/pack").join(pack_name);
    let index = pack.with_extension("idx");
    if pack.exists() {
        fs::remove_file(pack)?;
    }
    if index.exists() {
        fs::remove_file(index)?;
    }
    sync_directory(&repository_path.join("objects/pack"))?;
    Ok(())
}

fn parse_commands(input: &[u8], object_format: Kind) -> Result<Vec<RefCommand>, ReceivePackError> {
    let packets = decode(input)?;
    if packets.last() != Some(&Packet::Flush) {
        return Err(ReceivePackError::Commands);
    }
    let mut commands = Vec::new();
    let mut names = HashSet::new();
    for (index, packet) in packets[..packets.len() - 1].iter().enumerate() {
        let Packet::Data(line) = packet else {
            return Err(ReceivePackError::Commands);
        };
        let line = line.strip_suffix(b"\n").unwrap_or(line);
        let command = if index == 0 {
            line.split(|byte| *byte == 0).next().unwrap_or_default()
        } else if line.contains(&0) {
            return Err(ReceivePackError::Commands);
        } else {
            line
        };
        let mut fields = command.split(|byte| *byte == b' ');
        let old = parse_id(fields.next(), object_format)?;
        let new = parse_id(fields.next(), object_format)?;
        let name = fields.next().ok_or(ReceivePackError::Commands)?;
        if fields.next().is_some() || old == new {
            return Err(ReceivePackError::Commands);
        }
        let name = FullName::try_from(name.as_bstr()).map_err(|_| ReceivePackError::RefName)?;
        if !(name.as_bstr().starts_with(b"refs/heads/")
            || name.as_bstr().starts_with(b"refs/tags/"))
            || !names.insert(name.clone())
        {
            return Err(ReceivePackError::RefName);
        }
        commands.push(RefCommand { old, new, name });
    }
    if commands.is_empty() || commands.len() > MAX_COMMANDS {
        return Err(ReceivePackError::CommandLimit);
    }
    Ok(commands)
}

fn parse_id(input: Option<&[u8]>, object_format: Kind) -> Result<ObjectId, ReceivePackError> {
    let id = ObjectId::from_hex(input.ok_or(ReceivePackError::Commands)?)
        .map_err(|_| ReceivePackError::Commands)?;
    if id.kind() != object_format {
        return Err(ReceivePackError::ObjectFormat);
    }
    Ok(id)
}

fn validate_initial_refs(
    repository: &gix::Repository,
    commands: &[RefCommand],
) -> Result<(), ReceivePackError> {
    for command in commands {
        let current = repository
            .try_find_reference(&command.name)
            .map_err(|error| ReceivePackError::Repository(error.to_string()))?
            .and_then(|reference| reference.try_id().map(gix::Id::detach));
        match (command.old.is_null(), current) {
            (true, None) => {}
            (false, Some(current)) if current == command.old => {}
            _ => return Err(ReceivePackError::StaleRef),
        }
    }
    Ok(())
}

fn validate_proposed_objects(
    repository: &gix::Repository,
    bundle: Option<&gix_pack::Bundle>,
    commands: &[RefCommand],
) -> Result<Vec<RefChange>, ReceivePackError> {
    let mut finder = CombinedObjects::new(repository, bundle);
    let mut changes = Vec::with_capacity(commands.len());
    for command in commands {
        if command.new.is_null() {
            changes.push(RefChange::Delete);
            continue;
        }
        let kind = finder.kind_and_links(command.new)?.0;
        if command.name.as_bstr().starts_with(b"refs/heads/") && kind != ObjectKind::Commit {
            return Err(ReceivePackError::BranchNotCommit);
        }
        finder.validate_reachable(command.new)?;
        let change = if command.old.is_null() {
            RefChange::Create
        } else if command.name.as_bstr().starts_with(b"refs/tags/") {
            RefChange::TagUpdate
        } else if finder.is_ancestor(command.old, command.new)? {
            RefChange::FastForward
        } else {
            RefChange::Force
        };
        changes.push(change);
    }
    Ok(changes)
}

struct CombinedObjects<'a> {
    repository: &'a gix::Repository,
    bundle: Option<&'a gix_pack::Bundle>,
}

impl<'a> CombinedObjects<'a> {
    fn new(repository: &'a gix::Repository, bundle: Option<&'a gix_pack::Bundle>) -> Self {
        Self { repository, bundle }
    }

    fn object(&self, id: ObjectId) -> Result<(ObjectKind, Vec<u8>), ReceivePackError> {
        let mut buffer = Vec::new();
        let mut inflate = gix::features::zlib::Inflate::default();
        let mut cache = gix_pack::cache::Never;
        if let Some(bundle) = self.bundle
            && let Some((data, _)) = bundle
                .find(&id, &mut buffer, &mut inflate, &mut cache)
                .map_err(|error| ReceivePackError::Object(error.to_string()))?
        {
            if data.data.len() > MAX_OBJECT_BYTES {
                return Err(ReceivePackError::ObjectLimit);
            }
            return Ok((data.kind, data.data.to_vec()));
        }
        let object = self
            .repository
            .try_find_object(id)
            .map_err(|error| ReceivePackError::Object(error.to_string()))?
            .ok_or(ReceivePackError::MissingObject(id))?;
        if object.data.len() > MAX_OBJECT_BYTES {
            return Err(ReceivePackError::ObjectLimit);
        }
        Ok((object.kind, object.data.clone()))
    }

    fn kind_and_links(
        &self,
        id: ObjectId,
    ) -> Result<(ObjectKind, Vec<ObjectId>), ReceivePackError> {
        let (kind, data) = self.object(id)?;
        let links = match kind {
            ObjectKind::Blob => Vec::new(),
            ObjectKind::Commit => {
                let commit = CommitRef::from_bytes(&data, id.kind())
                    .map_err(|_| ReceivePackError::MalformedObject(kind))?;
                std::iter::once(commit.tree())
                    .chain(commit.parents())
                    .collect()
            }
            ObjectKind::Tree => TreeRefIter::from_bytes(&data, id.kind())
                .map(|entry| {
                    let entry = entry.map_err(|_| ReceivePackError::MalformedObject(kind))?;
                    Ok((entry.mode.kind() != gix::objs::tree::EntryKind::Commit)
                        .then(|| entry.oid.to_owned()))
                })
                .collect::<Result<Vec<_>, ReceivePackError>>()?
                .into_iter()
                .flatten()
                .collect(),
            ObjectKind::Tag => vec![
                TagRef::from_bytes(&data, id.kind())
                    .map_err(|_| ReceivePackError::MalformedObject(kind))?
                    .target(),
            ],
        };
        Ok((kind, links))
    }

    fn validate_reachable(&mut self, root: ObjectId) -> Result<(), ReceivePackError> {
        let mut pending = vec![root];
        let mut seen = HashSet::new();
        while let Some(id) = pending.pop() {
            if !seen.insert(id) {
                continue;
            }
            if seen.len() > MAX_WALK_OBJECTS {
                return Err(ReceivePackError::ObjectLimit);
            }
            pending.extend(self.kind_and_links(id)?.1);
        }
        Ok(())
    }

    fn is_ancestor(
        &self,
        ancestor: ObjectId,
        descendant: ObjectId,
    ) -> Result<bool, ReceivePackError> {
        let mut pending = vec![descendant];
        let mut seen = HashSet::new();
        while let Some(id) = pending.pop() {
            if id == ancestor {
                return Ok(true);
            }
            if !seen.insert(id) {
                continue;
            }
            if seen.len() > MAX_WALK_OBJECTS {
                return Err(ReceivePackError::ObjectLimit);
            }
            let (kind, links) = self.kind_and_links(id)?;
            if kind != ObjectKind::Commit {
                return Err(ReceivePackError::BranchNotCommit);
            }
            pending.extend(links.into_iter().skip(1));
        }
        Ok(false)
    }
}

fn apply_ref_transaction(
    repository: &gix::Repository,
    commands: &[RefCommand],
) -> Result<(), ReceivePackError> {
    let edits = commands.iter().map(|command| RefEdit {
        name: command.name.clone(),
        deref: false,
        change: if command.new.is_null() {
            Change::Delete {
                expected: PreviousValue::MustExistAndMatch(Target::Object(command.old)),
                log: RefLog::AndReference,
            }
        } else {
            Change::Update {
                expected: if command.old.is_null() {
                    PreviousValue::MustNotExist
                } else {
                    PreviousValue::MustExistAndMatch(Target::Object(command.old))
                },
                new: Target::Object(command.new),
                log: LogChange {
                    mode: RefLog::AndReference,
                    force_create_reflog: false,
                    message: "push".into(),
                },
            }
        },
    });
    repository
        .edit_references_as(edits, None)
        .map_err(|error| ReceivePackError::RefTransaction(error.to_string()))?;
    Ok(())
}

fn status_response(
    commands: &[RefCommand],
    error: Option<&str>,
) -> Result<Vec<u8>, PacketLineError> {
    let mut output = Vec::new();
    encode_data(b"unpack ok\n", &mut output)?;
    for command in commands {
        let name = String::from_utf8_lossy(command.name.as_bstr());
        let line = match error {
            Some(error) => format!("ng {name} {error}\n"),
            None => format!("ok {name}\n"),
        };
        encode_data(line.as_bytes(), &mut output)?;
    }
    encode_flush(&mut output);
    Ok(output)
}

fn serialize_refs(commands: &[RefCommand], proposed: bool) -> Vec<u8> {
    let mut output = Vec::new();
    for command in commands {
        let id = if proposed { command.new } else { command.old };
        writeln!(
            output,
            "{id} {}",
            String::from_utf8_lossy(command.name.as_bstr())
        )
        .expect("a vector write cannot fail");
    }
    output
}

fn open_bare(path: &Path) -> Result<gix::Repository, ReceivePackError> {
    let repository =
        gix::open(path).map_err(|error| ReceivePackError::Repository(error.to_string()))?;
    if !repository.is_bare() {
        return Err(ReceivePackError::NotBare);
    }
    Ok(repository)
}

fn random_id() -> Result<String, ReceivePackError> {
    let mut bytes = [0_u8; 16];
    rand::rngs::SysRng
        .try_fill_bytes(&mut bytes)
        .map_err(|_| ReceivePackError::Random)?;
    Ok(bytes.iter().map(|byte| format!("{byte:02x}")).collect())
}

fn path_text(path: &Path) -> Result<&str, ReceivePackError> {
    path.to_str().ok_or(ReceivePackError::Path)
}

fn sync_directory(path: &Path) -> Result<(), ReceivePackError> {
    File::open(path)?.sync_all()?;
    Ok(())
}

#[cfg(test)]
fn crash_point(point: &str) {
    if std::env::var("TIT_M1D_CRASH_AFTER").as_deref() != Ok(point) {
        return;
    }
    let ready = std::env::var_os("TIT_M1D_READY").expect("read the M1D ready path");
    fs::write(ready, point.as_bytes()).expect("write the M1D ready file");
    loop {
        std::thread::park();
    }
}

#[cfg(not(test))]
fn crash_point(_point: &str) {}

#[derive(Debug, Error)]
pub(crate) enum ReceivePackError {
    #[error(transparent)]
    PacketLine(#[from] PacketLineError),
    #[error(transparent)]
    Io(#[from] std::io::Error),
    #[error(transparent)]
    Store(#[from] crate::store::StoreError),
    #[error("cannot read Git repository: {0}")]
    Repository(String),
    #[error("Git repository is not bare")]
    NotBare,
    #[error("receive-pack commands are not valid")]
    Commands,
    #[error("receive-pack command count exceeds the limit")]
    CommandLimit,
    #[error("receive-pack object ID uses the wrong hash format")]
    ObjectFormat,
    #[error("receive-pack reference name is not allowed")]
    RefName,
    #[error("receive-pack initial reference does not match")]
    StaleRef,
    #[error("receive-pack input is missing a pack")]
    MissingPack,
    #[error("receive-pack input has an unexpected pack")]
    UnexpectedPack,
    #[error("receive-pack exceeds the pack limit")]
    PackLimit,
    #[error("receive-pack exceeds the object limit")]
    ObjectLimit,
    #[error("receive-pack exceeds the delta depth limit")]
    DeltaDepthLimit,
    #[error("receive-pack exceeds the processing time limit")]
    WallClockLimit,
    #[error("cannot index receive-pack input: {0}")]
    Pack(String),
    #[error("cannot read received object: {0}")]
    Object(String),
    #[error("received object {0} does not exist")]
    MissingObject(ObjectId),
    #[error("received {0:?} object is malformed")]
    MalformedObject(ObjectKind),
    #[error("a branch target is not a commit")]
    BranchNotCommit,
    #[error("a branch update is not a fast-forward")]
    NonFastForward,
    #[error("reference policy rejects the update")]
    RefPolicy,
    #[error("cannot check reference policy: {0}")]
    Policy(PolicyError),
    #[error("cannot update Git references: {0}")]
    RefTransaction(String),
    #[error("incomplete Git operation {0} has mixed reference state")]
    MixedRecovery(String),
    #[error("an incomplete Git operation has invalid recovery data")]
    RecoveryData,
    #[error("cannot create a random operation ID")]
    Random,
    #[error("system clock is before the Unix epoch")]
    Clock,
    #[error("filesystem path is not valid UTF-8")]
    Path,
}

impl ReceivePackError {
    fn is_unpack_error(&self) -> bool {
        matches!(
            self,
            Self::MissingPack
                | Self::UnexpectedPack
                | Self::PackLimit
                | Self::ObjectLimit
                | Self::Pack(_)
                | Self::Object(_)
                | Self::MissingObject(_)
                | Self::MalformedObject(_)
        )
    }

    fn client_reason(&self) -> &'static str {
        match self {
            Self::StaleRef => "stale reference",
            Self::NonFastForward => "non-fast-forward",
            Self::RefPolicy => "reference policy rejected the update",
            Self::Policy(_) => "reference policy is unavailable",
            Self::BranchNotCommit => "branch target is not a commit",
            Self::RefName => "reference name is not allowed",
            Self::ObjectFormat => "object format does not match",
            Self::CommandLimit => "too many reference commands",
            Self::PackLimit => "pack exceeds the byte limit",
            Self::ObjectLimit => "object limit exceeded",
            Self::DeltaDepthLimit => "delta depth limit exceeded",
            Self::WallClockLimit => "processing time limit exceeded",
            Self::MissingObject(_) => "object connectivity check failed",
            Self::MixedRecovery(_) => "repository recovery is required",
            _ if self.is_unpack_error() => "pack validation failed",
            _ => "push validation failed",
        }
    }
}

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

    #[test]
    fn rejects_a_compressed_object_before_large_allocation() {
        let directory = tempfile::TempDir::new().expect("create a pack directory");
        let path = directory.path().join("oversized.pack");
        let object_size = MAX_OBJECT_BYTES + 1;
        let mut pack = Vec::new();
        pack.extend_from_slice(b"PACK");
        pack.extend_from_slice(&2_u32.to_be_bytes());
        pack.extend_from_slice(&1_u32.to_be_bytes());
        pack.extend_from_slice(&pack_object_header(3, object_size as u64));

        let mut compressed = gix::features::zlib::stream::deflate::Write::new(Vec::new());
        let block = [0_u8; 64 * 1024];
        for _ in 0..(object_size / block.len()) {
            compressed.write_all(&block).expect("compress a zero block");
        }
        compressed
            .write_all(&block[..object_size % block.len()])
            .expect("compress the final zero block");
        compressed.flush().expect("finish the compressed object");
        pack.extend_from_slice(&compressed.into_inner());

        let mut hasher = gix::hash::hasher(Kind::Sha1);
        hasher.update(&pack);
        let checksum = hasher.try_finalize().expect("hash the pack");
        pack.extend_from_slice(checksum.as_bytes());
        assert!(pack.len() < 1024 * 1024);
        fs::write(&path, pack).expect("write the compressed pack");

        assert!(matches!(
            validate_pack_allocation_limits(&path, Kind::Sha1),
            Err(ReceivePackError::ObjectLimit)
        ));
    }

    fn pack_object_header(kind: u8, mut size: u64) -> Vec<u8> {
        let mut header = vec![((kind & 0x07) << 4) | (size as u8 & 0x0f)];
        size >>= 4;
        while size != 0 {
            *header.last_mut().expect("the header has one byte") |= 0x80;
            header.push((size as u8) & 0x7f);
            size >>= 7;
        }
        header
    }
}
