//! Serving and talking to remotes (spec §10.2, §16).

use super::tags::TagCheck;

use super::*;

/// What a push did, for hosts (hub spec §6.12).

#[derive(Clone, Debug, Default)]

pub struct PushReport {

    pub results: Vec<RefResult>,

    /// Accepted reference updates, in request order.

    pub updated: Vec<RefUpdated>,

    /// Changes this push stored that were not already present.

    pub imported: Vec<ChangeHash>,

}

#[derive(Clone, Debug)]

pub struct RefUpdated {

    pub name: String,

    pub old_state: ObjectRef,

    pub new_state: ObjectRef,

    pub heads: Vec<ChangeHash>,

}

/// What a fetch brought in.

#[derive(Debug, Clone, PartialEq, Eq, Default)]

pub struct FetchReport {

    /// Bookmark name, its heads, and how many changes were new.

    pub refs: Vec<(String, Vec<ChangeHash>, usize)>,

    /// Tags adopted from the remote.

    pub new_tags: usize,

    /// Tags the remote withdrew, and so withdrawn here (oxygenhub#301).

    pub withdrawn_tags: Vec<String>,

    /// The remote has never presented a policy, so nothing was verified

    /// against one (spec §14.5: warn loudly).

    pub unverified: bool,

    /// What checking the host's transparency log found short of a fork

    /// (spec §14.8): attestations not logged yet, a log that could not be

    /// read. A fork fails the fetch instead.

    pub log_notes: Vec<String>,

    /// Bookmarks left as they were because their attestation could not be

    /// verified (its signer unknown here, or its signature bad), each with

    /// why (oxygenhub#266). The current bookmark is never skipped: its

    /// fetch fails instead.

    pub skipped: Vec<(String, String)>,

}

#[derive(Debug, Clone, PartialEq, Eq)]

pub struct PullReport {

    /// The current bookmark's heads after the pull.

    pub heads: Vec<ChangeHash>,

    /// Whether the current bookmark changed.

    pub changed: bool,

    pub checked_out: bool,

    /// Other bookmarks the pull moved: each local bookmark tracks the

    /// remote's of the same name.

    pub others: Vec<String>,

    /// What the fetch left as it was ([`FetchReport::skipped`]).

    pub skipped: Vec<(String, String)>,

}

/// The result a push of objects alone reports (a batch of a long history):

/// not a reference.

pub const OBJECTS_RESULT: &str = "objects";

/// Graph blobs asked for in one `objects` request when a shallow clone

/// fetches its checkpoint in parts (#137): their chunk lists and chunks,

/// a few MiB a request rather than one answer of gigabytes.

const CHECKPOINT_PART: usize = 2048;

impl Repo {

    // ------------------------------------------------------------ serving (spec §16)

    fn ref_attest_path(&self, bookmark: &str) -> PathBuf {

        self.oxy.join("attest").join(name_file(bookmark))

    }

    /// The latest reference attestation stored for a bookmark.

    pub fn latest_ref_attest(&self, bookmark: &str) -> Result<Option<(ObjectRef, SignedAttest)>> {

        match fs::read_to_string(self.ref_attest_path(bookmark)) {

            Ok(h) => {

                let r = ObjectRef::from_hex(h.trim())

                    .ok_or_else(|| RepoError::Msg("corrupt attestation pointer".into()))?;

                Ok(Some((

                    r,

                    SignedAttest {

                        attest: self.store.get_attest(&r)?,

                        sigs: self.store.signatures(&r)?,

                    },

                )))

            }

            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),

            Err(e) => Err(e.into()),

        }

    }

    pub(crate) fn set_ref_attest(&self, bookmark: &str, r: &ObjectRef) -> Result<()> {

        fs::create_dir_all(self.oxy.join("attest"))?;

        write_atomic(&self.ref_attest_path(bookmark), r.to_hex().as_bytes())

    }

    pub fn remote_refs(&self) -> Result<RefsResponse> {

        let view = self.view()?;

        let mut out = Vec::new();

        for (name, state) in &view.bookmarks {

            let latest = self.latest_ref_attest(name)?.map(|(_, a)| a);

            let last_serial = latest.as_ref().and_then(|a| match a.attest.body {

                AttestBody::Ref { serial, .. } => Some(serial),

                _ => None,

            });

            let attestation = latest.filter(|a| matches!(&a.attest.body, AttestBody::Ref { new_state, .. } if new_state == state));

            let accepted = match &attestation {

                Some(a) => self.acceptance_of(&a.attest.hash())?,

                None => None,

            };

            out.push(RemoteRef {

                name: name.clone(),

                state: *state,

                heads: self.store.get_state(state)?.heads,

                attestation,

                last_serial,

                accepted,

                logged: None,

                integration: vec![],

            });

        }

        let policy = self.current_policy()?.map(|(_, sp)| sp);

        let history = self

            .policy_history()?

            .into_iter()

            .filter(|sp| {

                policy

                    .as_ref()

                    .is_some_and(|p| sp.policy.serial < p.policy.serial)

            })

            .collect();

        Ok(RefsResponse {

            refs: out,

            policy,

            tags: self.tag_refs()?,

            history,

            default_bookmark: Some(self.config.bookmark.clone()),

            log: None,

            witnesses: vec![],

            withdrawn: self.withdrawn_refs()?,

        })

    }

    /// Every change reachable from `heads` that is in the store.

    fn closure_in_store(&self, heads: &[ChangeHash]) -> Result<BTreeSet<ChangeHash>> {

        let mut seen = BTreeSet::new();

        let mut stack: Vec<ChangeHash> = heads.to_vec();

        while let Some(h) = stack.pop() {

            if seen.contains(&h) || !self.store.has(&h) {

                continue;

            }

            seen.insert(h);

            stack.extend(self.store.get_change_meta(&h)?.deps);

        }

        Ok(seen)

    }

    /// The changes within `depth` dependency steps of `heads` (spec §17),

    /// nearest first; depth 1 is the heads themselves.

    pub fn shallow_changes(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {

        let mut out = Vec::new();

        let mut seen: BTreeSet<ChangeHash> = BTreeSet::new();

        let mut layer: Vec<ChangeHash> = heads.to_vec();

        for _ in 0..depth.max(1) {

            let mut next = Vec::new();

            for h in layer {

                if !seen.insert(h) || !self.store.has(&h) {

                    continue;

                }

                let c = self.store.get_change(&h)?;

                next.extend(c.deps.iter().copied());

                out.push(h);

            }

            if next.is_empty() {

                break;

            }

            layer = next;

        }

        out.reverse();

        Ok(out)

    }

    fn shallow_path(&self) -> PathBuf {

        self.oxy.join("shallow")

    }

    /// The shallow boundary: dependencies this repository does not hold (spec §17).

    pub fn boundary(&self) -> BTreeSet<ChangeHash> {

        fs::read_to_string(self.shallow_path())

            .map(|t| t.lines().filter_map(ObjectRef::from_hex).collect())

            .unwrap_or_default()

    }

    pub fn is_shallow(&self) -> bool {

        !self.boundary().is_empty()

    }

    fn write_boundary(&self, boundary: &BTreeSet<ChangeHash>) -> Result<()> {

        if boundary.is_empty() {

            let _ = fs::remove_file(self.shallow_path());

            return Ok(());

        }

        let text: String = boundary

            .iter()

            .map(|h| format!("{}\n", h.to_hex()))

            .collect();

        write_atomic(&self.shallow_path(), text.as_bytes())

    }

    /// Recompute the boundary from what is stored: every dependency of a

    /// stored change that is not itself stored.

    fn refresh_boundary(&self, imported: &[ChangeHash]) -> Result<()> {

        let mut boundary = self.boundary();

        boundary.retain(|h| !self.store.has(h));

        for h in imported {

            for d in self.store.get_change_meta(h)?.deps {

                if !self.store.has(&d) {

                    boundary.insert(d);

                }

            }

        }

        self.write_boundary(&boundary)

    }

    /// Fetch `depth` more levels of history below the shallow boundary.

    pub fn deepen(&self, remote_name: &str, depth: u32) -> Result<usize> {

        let boundary = self.boundary();

        if boundary.is_empty() {

            return Ok(0);

        }

        let remote = self.open_named_remote(remote_name)?;

        let wanted = remote.shallow(&boundary.iter().copied().collect::<Vec<_>>(), depth)?;

        let bundle = remote.fetch(&wanted, self.config.max_blob)?;

        let imported = self.import_bundle_shallow(&bundle, false)?;

        if self.config.max_blob.is_some() {

            self.register_missing_blobs(&imported, remote_name)?;

        }

        self.refresh_boundary(&imported)?;

        Ok(imported.len())

    }

    /// `closure(want) \ closure(have)`, dependencies first (spec §16.2).

    pub fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {

        let have = self.closure_in_store(have)?;

        // In a shallow repository the walk stops at the boundary, whose

        // changes the remote is assumed to hold.

        let boundary = self.boundary();

        let mut out = Vec::new();

        let mut done: BTreeSet<ChangeHash> = have.clone();

        done.extend(boundary.iter().copied());

        // Iterative post-order DFS.

        let shallow = self.is_shallow();

        let mut stack: Vec<(ChangeHash, bool)> = want.iter().rev().map(|h| (*h, false)).collect();

        while let Some((h, expanded)) = stack.pop() {

            if done.contains(&h) {

                continue;

            }

            if !self.store.has(&h) {

                // A shallow repository holds only part of its history; a

                // change it lacks is below what it cloned (in a depth-0

                // clone, any change of the checkpoint), which the remote

                // is assumed to hold, as it is for the boundary (#137).

                if shallow {

                    done.insert(h);

                    continue;

                }

                return fail(format!("change {} is not in this repository", h.short(12)));

            }

            if expanded {

                done.insert(h);

                out.push(h);

                continue;

            }

            stack.push((h, true));

            for d in self.store.get_change_meta(&h)?.deps {

                if !done.contains(&d) {

                    stack.push((d, false));

                }

            }

        }

        Ok(out)

    }

    /// Bundle changes (in the given dependency order) with their signatures,

    /// seals, identities, blobs, chunks.

    pub fn bundle_for(&self, changes: &[ChangeHash]) -> Result<Bundle> {

        self.bundle_of(changes, None, false, false)

    }

    /// Objects by reference for the `objects` endpoint; a blob brings its chunks.

    pub fn objects_for(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {

        // In two reads in pack order (#137): a checkpoint's graph blobs come

        // this way in parts, tens of thousands for a big state.

        let mut found = Vec::with_capacity(refs.len());

        for got in self.store.get_many(refs) {

            match got {

                Ok(x) => found.push(x),

                Err(StoreError::NotFound(_)) => {}

                Err(e) => return Err(e.into()),

            }

        }

        let mut chunks = Vec::new();

        for (ty, body) in &found {

            if *ty == ObjectType::Blob {

                chunks.extend(Blob::decode(body)?.chunks);

            }

        }

        let mut out = Vec::with_capacity(found.len() + chunks.len());

        for got in self.store.get_many(&chunks) {

            match got {

                Ok(x) => out.push(x),

                Err(StoreError::NotFound(_)) => {}

                Err(e) => return Err(e.into()),

            }

        }

        out.extend(found);

        Ok(out)

    }

    /// Like `bundle_for`, leaving out blobs larger than `max_blob` bytes

    /// (and their chunks) so the receiver fetches them lazily (spec §13.5).

    ///

    /// This is what a server answers a fetch with, so it also carries what

    /// verifies the attestations it serves beside it (see

    /// [`Repo::served_attest_signers`]), also when `changes` is empty.

    pub fn bundle_for_filtered(

        &self,

        changes: &[ChangeHash],

        max_blob: Option<u64>,

    ) -> Result<Bundle> {

        self.bundle_of(changes, max_blob, false, true)

    }

    /// Like [`Repo::bundle_for_filtered`] for a fetch between repositories

    /// on one machine: changes as they are stored ([`Bundle::raw_changes`]),

    /// with their text blobs and chunks among the objects, so neither side

    /// decodes or encodes their text.

    pub fn bundle_raw_for(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {

        self.bundle_of(changes, max_blob, true, true)

    }

    /// The keys that signed the attestations this repository serves with

    /// its references: each bookmark's latest and each tag's. A fetch

    /// bundle carries their identities and certificates (spec §16.2), so a

    /// client can check an attestation whose signer signed none of the

    /// changes it receives, a delegate or a short-lived key that only

    /// moved a bookmark (oxygenhub#266). One pointer and one signature

    /// list per reference; nothing per change.

    pub fn served_attest_signers(&self) -> Result<BTreeSet<KeyId>> {

        let view = self.view()?;

        let pointers = view

            .bookmarks

            .keys()

            .map(|b| self.ref_attest_path(b))

            .chain(view.tags.keys().map(|t| self.tag_attest_path(t)));

        let mut keys = BTreeSet::new();

        for p in pointers {

            let Ok(text) = fs::read_to_string(&p) else {

                continue;

            };

            let Some(r) = ObjectRef::from_hex(text.trim()) else {

                continue;

            };

            keys.extend(self.store.signatures(&r)?.into_iter().map(|s| s.key));

        }

        Ok(keys)

    }

    /// `serving`: a fetch's answer, which carries the identities and

    /// certificates of [`Repo::served_attest_signers`] too.

    fn bundle_of(

        &self,

        changes: &[ChangeHash],

        max_blob: Option<u64>,

        raw: bool,

        serving: bool,

    ) -> Result<Bundle> {

        let mut bundle = Bundle::default();

        let mut objects: BTreeSet<ObjectRef> = BTreeSet::new();

        let mut seals: BTreeSet<ObjectRef> = BTreeSet::new();

        let add_object = |r: &ObjectRef, objects: &mut BTreeSet<ObjectRef>| {

            if !r.is_zero() && self.store.has(r) {

                objects.insert(*r);

            }

        };

        for h in changes {

            // Raw: the stored bytes, and a skeleton for the rest.

            let (c, encoded) = if raw {

                let bytes = self.store.get_typed(h, ObjectType::Change)?;

                let (skel, texts) = Change::decode_skeleton(&bytes, *h)?;

                for t in texts {

                    if let crate::objects::TextRef::Blob(r, _, _) = t {

                        let blob = Blob::decode(&self.store.get_typed(&r, ObjectType::Blob)?)?;

                        objects.insert(r);

                        objects.extend(blob.chunks);

                    }

                }

                (skel, Some(bytes))

            } else {

                (self.store.get_change(h)?, None)

            };

            let sigs = self.store.signatures(h)?;

            add_object(&c.author, &mut objects);

            for s in &sigs {

                add_object(&s.key, &mut objects);

            }

            // A replacement's new content only: its old content is the new

            // content of a change it depends on, which the receiver holds or

            // which this bundle carries (oxygen#136: removing ~1,100 binary

            // files sent them all again).

            for e in &c.edits {

                if let Edit::Replace { new: Some(b), .. } = e {

                    if self.store.has(b) {

                        if let Ok(blob) = Blob::decode(&self.store.get_typed(b, ObjectType::Blob)?)

                        {

                            if max_blob.is_some_and(|m| blob.length > m) {

                                continue;

                            }

                            objects.insert(*b);

                            objects.extend(blob.chunks.iter().copied());

                        }

                    }

                }

            }

            for s in self.store.seals_for(h)? {

                seals.insert(s);

            }

            match encoded {

                Some(bytes) => bundle.raw_changes.push((bytes, sigs)),

                None => {

                    bundle.changes.push((c, sigs));

                    bundle.hashes.push(*h);

                }

            }

        }

        for r in seals {

            let attest = self.store.get_attest(&r)?;

            let sigs = self.store.signatures(&r)?;

            for s in &sigs {

                add_object(&s.key, &mut objects);

            }

            bundle.attests.push(SignedAttest { attest, sigs });

        }