//! Read-only access to Tangled: the public appview index, repo records on a
//! PDS, and Git history from a knot (via the mirror when it has a copy).

use std::collections::BTreeMap;

use anyhow::{Context, Result, bail};
use serde_json::Value;
use url::Url;

use crate::db::canonical_time;
use crate::http::HttpClient;

#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct RepoRecord {
    /// The repository's DID, or for a record from before repositories had
    /// one, `<owner DID>/<name>`: the identifier its knot still serves it by
    /// (see [`is_legacy_repo`]).
    pub repo_did: String,
    pub owner_did: String,
    pub name: String,
    pub knot: String,
    pub url: String,
    pub website: Option<String>,
    /// A fork record carries the upstream repo's `source`.
    pub fork: bool,
    pub created_at: Option<String>,
    /// The repo record's own description, as its owner wrote it.
    pub description: Option<String>,
}

/// Every repository record search returned, and how many of them were
/// legacy records (no `repoDid`) and distinct owners.
#[derive(Debug, Clone, Default)]
pub struct SearchResults {
    pub repos: Vec<RepoRecord>,
    pub legacy: usize,
    pub owners: std::collections::BTreeSet<String>,
}

/// Whether `repo_id` is a legacy `<owner DID>/<name>` identifier. Knots
/// address such a repository by owner and name; the mirror doesn't have it.
pub fn is_legacy_repo(repo_id: &str) -> bool {
    repo_id.contains('/')
}

#[derive(Debug, Clone)]
pub struct BranchHead {
    pub name: String,
    pub head_sha: String,
    pub is_default: bool,
}

#[derive(Debug, Clone)]
pub struct CommitRecord {
    pub sha: String,
    pub change_id: String,
    pub author_name: String,
    pub author_email: String,
    pub authored_at: String,
    pub committed_at: String,
    pub message: String,
}

#[derive(Debug, Clone)]
pub struct CommitPage {
    pub commits: Vec<CommitRecord>,
    pub total: Option<u64>,
}

#[derive(Debug, Clone)]
pub struct DidDescription {
    pub handle: String,
    pub operator: Option<String>,
    pub collections: Vec<String>,
}

#[derive(Debug, Clone)]
pub struct Tangled {
    http: HttpClient,
    api: String,
    mirror: String,
}

impl Tangled {
    pub fn new(http: HttpClient, api: impl Into<String>, mirror: impl Into<String>) -> Self {
        Self {
            http,
            api: api.into().trim_end_matches('/').to_string(),
            mirror: mirror.into().trim_end_matches('/').to_string(),
        }
    }

    /// The whole public repo index. Search is the only public enumeration that
    /// reaches every knot; `sh.tangled.repo.listRepos` needs an owner. Records
    /// without a `repoDid` (repositories created before they had one) are kept
    /// under their legacy `<owner>/<name>` identifier.
    pub async fn search_repos(&self) -> Result<SearchResults> {
        let mut out = BTreeMap::<String, RepoRecord>::new();
        let mut owners = std::collections::BTreeSet::new();
        let mut cursor: Option<String> = None;
        let mut pages = 0usize;

        loop {
            let mut url = Url::parse(&format!("{}/xrpc/sh.tangled.search.query", self.api))?;
            {
                let mut query = url.query_pairs_mut();
                query.append_pair("q", "*");
                query.append_pair("nsid", "sh.tangled.repo");
                query.append_pair("limit", "100");
                if let Some(cursor) = &cursor {
                    query.append_pair("cursor", cursor);
                }
            }

            let body: Value = self
                .http
                .get_json(url.as_str(), None)
                .await
                .with_context(|| format!("enumerating Tangled repos (page {})", pages + 1))?;
            let hits = body
                .get("hits")
                .and_then(Value::as_array)
                .cloned()
                .unwrap_or_default();
            if hits.is_empty() {
                break;
            }

            for hit in hits {
                if let Some(record) = repo_from_search_hit(&hit) {
                    owners.insert(record.owner_did.clone());
                    out.entry(record.repo_did.clone()).or_insert(record);
                }
            }

            cursor = body
                .get("cursor")
                .and_then(Value::as_str)
                .map(ToString::to_string);
            pages += 1;
            if cursor.is_none() {
                break;
            }
            if pages >= 1_000 {
                bail!("Tangled search did not terminate after 1000 pages");
            }
        }

        let repos: Vec<RepoRecord> = out.into_values().collect();
        Ok(SearchResults {
            legacy: repos
                .iter()
                .filter(|repo| is_legacy_repo(&repo.repo_did))
                .count(),
            repos,
            owners,
        })
    }

    /// Every account a relay has seen write a `sh.tangled.repo` record. Some
    /// of them search doesn't index; most of the rest have since deleted the
    /// record or the account, which a listing of their records shows.
    pub async fn relay_repo_owners(&self, relay: &str) -> Result<Vec<String>> {
        let mut out = Vec::new();
        let mut cursor: Option<String> = None;
        for page in 1..=1_000 {
            let mut url = Url::parse(&format!(
                "{}/xrpc/com.atproto.sync.listReposByCollection",
                relay.trim_end_matches('/')
            ))?;
            {
                let mut query = url.query_pairs_mut();
                query.append_pair("collection", "sh.tangled.repo");
                query.append_pair("limit", "2000");
                if let Some(cursor) = &cursor {
                    query.append_pair("cursor", cursor);
                }
            }
            let body: Value = self
                .http
                .get_json(url.as_str(), None)
                .await
                .with_context(|| {
                    format!("listing sh.tangled.repo owners on {relay} (page {page})")
                })?;
            let repos = body
                .get("repos")
                .and_then(Value::as_array)
                .cloned()
                .unwrap_or_default();
            out.extend(
                repos
                    .iter()
                    .filter_map(|repo| repo.get("did").and_then(Value::as_str))
                    .map(ToString::to_string),
            );
            cursor = body
                .get("cursor")
                .and_then(Value::as_str)
                .map(ToString::to_string);
            if cursor.is_none() || repos.is_empty() {
                return Ok(out);
            }
        }
        bail!("{relay} did not finish listing sh.tangled.repo owners after 1000 pages")
    }

    /// Repo records published by one owner. Used to make sure the person's own
    /// repositories are present even if search is behind.
    pub async fn repos_for_owner(&self, pds: &str, did: &str) -> Result<Vec<RepoRecord>> {
        let mut out = Vec::new();
        let mut cursor: Option<String> = None;
        loop {
            let mut url = Url::parse(&format!(
                "{}/xrpc/com.atproto.repo.listRecords",
                pds.trim_end_matches('/')
            ))?;
            {
                let mut query = url.query_pairs_mut();
                query.append_pair("repo", did);
                query.append_pair("collection", "sh.tangled.repo");
                query.append_pair("limit", "100");
                if let Some(cursor) = &cursor {
                    query.append_pair("cursor", cursor);
                }
            }
            let body: Value = self
                .http
                .get_json(url.as_str(), None)
                .await
                .with_context(|| format!("listing Tangled repos for {did}"))?;
            let records = body
                .get("records")
                .and_then(Value::as_array)
                .cloned()
                .unwrap_or_default();
            for record in records {
                if let Some(value) = record.get("value") {
                    let rkey = record
                        .get("uri")
                        .and_then(Value::as_str)
                        .and_then(|uri| uri.rsplit('/').next());
                    if let Some(repo) = repo_from_record(did, value, rkey) {
                        out.push(repo);
                    }
                }
            }
            cursor = body
                .get("cursor")
                .and_then(Value::as_str)
                .map(ToString::to_string);
            if cursor.is_none() {
                break;
            }
        }
        Ok(out)
    }

    /// Branch heads. The mirror is tried first because it returns 1000 refs a
    /// page; the knot's own API is the fallback and is capped at 100.
    pub async fn branches(&self, knot: &str, repo_did: &str) -> Result<Vec<BranchHead>> {
        if is_legacy_repo(repo_did) {
            return self
                .branches_knot(knot, repo_did)
                .await
                .with_context(|| format!("listing branches for {repo_did}"));
        }
        match self.branches_mirror(repo_did).await {
            Ok(branches) => Ok(branches),
            Err(mirror_error) => {
                tracing::debug!(repo_did, error = %mirror_error, "mirror branch listing failed; asking the knot");
                self.branches_knot(knot, repo_did)
                    .await
                    .with_context(|| format!("listing branches for {repo_did}"))
            }
        }
    }

    async fn branches_mirror(&self, repo_did: &str) -> Result<Vec<BranchHead>> {
        let mut out = Vec::new();
        let mut offset = 0u64;
        let limit = 1000u64;
        loop {
            let mut url = Url::parse(&format!(
                "{}/xrpc/sh.tangled.git.temp.listBranches",
                self.mirror
            ))?;
            {
                let mut query = url.query_pairs_mut();
                query.append_pair("repo", repo_did);
                query.append_pair("limit", &limit.to_string());
                query.append_pair("cursor", &offset.to_string());
            }
            let body: Value = self.http.get_json(url.as_str(), None).await?;
            let branches = body
                .get("branches")
                .and_then(Value::as_array)
                .cloned()
                .unwrap_or_default();
            let count = branches.len();
            for branch in branches {
                if let Some(head) = branch_head(&branch) {
                    out.push(head);
                }
            }
            if count < limit as usize {
                break;
            }
            offset += limit;
        }
        Ok(out)
    }

    async fn branches_knot(&self, knot: &str, repo_did: &str) -> Result<Vec<BranchHead>> {
        let mut out = Vec::new();
        let mut offset = 0u64;
        let limit = 100u64;
        loop {
            let mut url = Url::parse(&format!(
                "https://{}/xrpc/sh.tangled.repo.branches",
                knot.trim_end_matches('/')
            ))?;
            {
                let mut query = url.query_pairs_mut();
                query.append_pair("repo", repo_did);
                query.append_pair("limit", &limit.to_string());
                query.append_pair("cursor", &offset.to_string());
            }
            let body: Value = self.http.get_json(url.as_str(), None).await?;
            let branches = body
                .get("branches")
                .and_then(Value::as_array)
                .cloned()
                .unwrap_or_default();
            let count = branches.len();
            for branch in branches {
                if let Some(head) = branch_head(&branch) {
                    out.push(head);
                }
            }
            if count < limit as usize {
                break;
            }
            offset += limit;
        }
        Ok(out)
    }

    /// A page of commits reachable from `reference`.
    pub async fn commits(
        &self,
        knot: &str,
        repo_did: &str,
        reference: &str,
        offset: u64,
        limit: u64,
    ) -> Result<CommitPage> {
        if is_legacy_repo(repo_did) {
            return self
                .commits_knot(knot, repo_did, reference, offset, limit.min(100))
                .await
                .with_context(|| format!("reading commits for {repo_did} at {reference}"));
        }
        match self
            .commits_mirror(repo_did, reference, offset, limit.min(1000))
            .await
        {
            Ok(page) => Ok(page),
            Err(mirror_error) => {
                tracing::debug!(repo_did, reference, error = %mirror_error, "mirror commit log failed; asking the knot");
                self.commits_knot(knot, repo_did, reference, offset, limit.min(100))
                    .await
                    .with_context(|| format!("reading commits for {repo_did} at {reference}"))
            }
        }
    }

    async fn commits_mirror(
        &self,
        repo_did: &str,
        reference: &str,
        offset: u64,
        limit: u64,
    ) -> Result<CommitPage> {
        let mut url = Url::parse(&format!(
            "{}/xrpc/sh.tangled.git.temp.listCommits",
            self.mirror
        ))?;
        {
            let mut query = url.query_pairs_mut();
            query.append_pair("repo", repo_did);
            query.append_pair("ref", reference);
            query.append_pair("limit", &limit.to_string());
            query.append_pair("cursor", &offset.to_string());
        }
        let body: Value = self.http.get_json(url.as_str(), None).await?;
        parse_commit_page(&body)
    }

    async fn commits_knot(
        &self,
        knot: &str,
        repo_did: &str,
        reference: &str,
        offset: u64,
        limit: u64,
    ) -> Result<CommitPage> {
        let mut url = Url::parse(&format!(
            "https://{}/xrpc/sh.tangled.repo.log",
            knot.trim_end_matches('/')
        ))?;
        {
            let mut query = url.query_pairs_mut();
            query.append_pair("repo", repo_did);
            query.append_pair("ref", reference);
            query.append_pair("limit", &limit.to_string());
            query.append_pair("cursor", &offset.to_string());
        }
        let body: Value = self.http.get_json(url.as_str(), None).await?;
        parse_commit_page(&body)
    }

    /// Resolve a DID's PDS and operator. `pds` is the endpoint to ask when it
    /// is already known (a didbot directory listing gives it to us); otherwise
    /// the DID document is fetched.
    pub async fn describe_did(&self, did: &str, pds: Option<&str>) -> Result<DidDescription> {
        let pds = match pds {
            Some(pds) => pds.trim_end_matches('/').to_string(),
            None => self.resolve_pds(did).await?,
        };
        let mut url = Url::parse(&format!("{pds}/xrpc/com.atproto.repo.describeRepo"))?;
        url.query_pairs_mut().append_pair("repo", did);
        let body: Value = self.http.get_json(url.as_str(), None).await?;
        let handle = body
            .get("handle")
            .and_then(Value::as_str)
            .unwrap_or(did)
            .to_string();
        let operator = body
            .get("didDoc")
            .and_then(|doc| doc.get("operator"))
            .and_then(Value::as_str)
            .map(ToString::to_string);
        let collections = body
            .get("collections")
            .and_then(Value::as_array)
            .map(|values| {
                values
                    .iter()
                    .filter_map(Value::as_str)
                    .map(ToString::to_string)
                    .collect()
            })
            .unwrap_or_default();
        Ok(DidDescription {
            handle,
            operator,
            collections,
        })
    }

    pub async fn resolve_pds(&self, did: &str) -> Result<String> {
        let doc_url = if let Some(rest) = did.strip_prefix("did:plc:") {
            format!("https://plc.directory/did:plc:{rest}")
        } else if let Some(rest) = did.strip_prefix("did:web:") {
            let decoded = rest.replace("%3A", ":").replace("%3a", ":");
            let mut parts = decoded.split(':');
            let host = parts.next().unwrap_or("");
            let path = parts.collect::<Vec<_>>().join("/");
            if path.is_empty() {
                format!("https://{host}/.well-known/did.json")
            } else {
                format!("https://{host}/{path}/did.json")
            }
        } else {
            bail!("unsupported DID method in {did}");
        };

        let doc: Value = self
            .http
            .get_json(&doc_url, None)
            .await
            .with_context(|| format!("resolving DID document for {did}"))?;
        let endpoints = doc
            .get("service")
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_default();
        for service in endpoints {
            let id = service.get("id").and_then(Value::as_str).unwrap_or("");
            let kind = service.get("type").and_then(Value::as_str).unwrap_or("");
            if (id.ends_with("atproto_pds") || kind == "AtprotoPersonalDataServer")
                && let Some(endpoint) = service.get("serviceEndpoint").and_then(Value::as_str)
            {
                return Ok(endpoint.trim_end_matches('/').to_string());
            }
        }
        bail!("DID document for {did} has no atproto PDS service")
    }
}

fn repo_from_search_hit(hit: &Value) -> Option<RepoRecord> {
    let uri = hit.get("uri")?.as_str()?;
    let value = hit.get("value")?;
    let owner = uri.strip_prefix("at://")?.split('/').next()?.to_string();
    let rkey = uri.rsplit('/').next()?.to_string();
    let name = value
        .get("name")
        .and_then(Value::as_str)
        .filter(|name| !name.is_empty())
        .unwrap_or(&rkey)
        .to_string();
    let repo_did = repo_id(&owner, &name, value)?;
    let knot = value
        .get("knot")
        .and_then(Value::as_str)
        .unwrap_or("knot1.tangled.sh")
        .to_string();
    let website = value
        .get("website")
        .and_then(Value::as_str)
        .map(ToString::to_string);
    let fork = value.get("source").is_some();
    let created_at = value
        .get("createdAt")
        .and_then(Value::as_str)
        .map(ToString::to_string);
    let description = value
        .get("description")
        .and_then(Value::as_str)
        .map(ToString::to_string);
    Some(RepoRecord {
        url: format!("https://tangled.org/{owner}/{name}"),
        repo_did,
        owner_did: owner,
        name,
        knot,
        website,
        fork,
        created_at,
        description,
    })
}

/// A record's repository identifier: its `repoDid`, or for a legacy record
/// `<owner>/<name>`, which is how its knot serves it. A legacy record needs a
/// knot and a name that can be a path segment to be addressable.
fn repo_id(owner_did: &str, name: &str, value: &Value) -> Option<String> {
    if let Some(repo_did) = value.get("repoDid").and_then(Value::as_str) {
        return Some(repo_did.to_string());
    }
    let usable = !name.is_empty() && !name.contains('/') && value.get("knot").is_some();
    usable.then(|| format!("{owner_did}/{name}"))
}

fn repo_from_record(owner_did: &str, value: &Value, rkey: Option<&str>) -> Option<RepoRecord> {
    // Older records omit `name`; the record key is the repository name.
    let name = value
        .get("name")
        .and_then(Value::as_str)
        .filter(|name| !name.is_empty())
        .or(rkey)
        .unwrap_or("")
        .to_string();
    let repo_did = repo_id(owner_did, &name, value)?;
    let knot = value
        .get("knot")
        .and_then(Value::as_str)
        .unwrap_or("knot1.tangled.sh")
        .to_string();
    let website = value
        .get("website")
        .and_then(Value::as_str)
        .map(ToString::to_string);
    let fork = value.get("source").is_some();
    let created_at = value
        .get("createdAt")
        .and_then(Value::as_str)
        .map(ToString::to_string);
    let description = value
        .get("description")
        .and_then(Value::as_str)
        .map(ToString::to_string);
    Some(RepoRecord {
        url: format!("https://tangled.org/{owner_did}/{name}"),
        repo_did,
        owner_did: owner_did.to_string(),
        name,
        knot,
        website,
        fork,
        created_at,
        description,
    })
}

fn branch_head(branch: &Value) -> Option<BranchHead> {
    let reference = branch.get("reference")?;
    Some(BranchHead {
        name: reference.get("name")?.as_str()?.to_string(),
        head_sha: reference.get("hash")?.as_str()?.to_string(),
        is_default: branch
            .get("is_default")
            .and_then(Value::as_bool)
            .unwrap_or(false),
    })
}

fn parse_commit_page(body: &Value) -> Result<CommitPage> {
    let total = body.get("total").and_then(Value::as_u64);
    let raw = body
        .get("commits")
        .and_then(Value::as_array)
        .cloned()
        .unwrap_or_default();
    let mut commits = Vec::with_capacity(raw.len());
    for commit in raw {
        let Some(hash) = commit.get("hash").and_then(hash_to_hex) else {
            continue;
        };
        let author = commit.get("author");
        let author_name = author
            .and_then(|value| value.get("Name"))
            .and_then(Value::as_str)
            .unwrap_or("")
            .to_string();
        let author_email = author
            .and_then(|value| value.get("Email"))
            .and_then(Value::as_str)
            .unwrap_or("")
            .to_string();
        let authored_raw = author
            .and_then(|value| value.get("When"))
            .and_then(Value::as_str)
            .unwrap_or("");
        let committed_raw = commit
            .get("committer")
            .and_then(|value| value.get("When"))
            .and_then(Value::as_str)
            .unwrap_or("");
        let authored_at = canonical_time(authored_raw).unwrap_or_else(|| authored_raw.to_string());
        let committed_at =
            canonical_time(committed_raw).unwrap_or_else(|| committed_raw.to_string());
        let full_message = commit.get("message").and_then(Value::as_str).unwrap_or("");
        let change_id = crate::model::change_id_from_message(full_message);
        let message = full_message
            .lines()
            .next()
            .unwrap_or("")
            .chars()
            .take(500)
            .collect::<String>();
        commits.push(CommitRecord {
            sha: hash,
            change_id,
            author_name,
            author_email,
            authored_at,
            committed_at,
            message,
        });
    }
    Ok(CommitPage { commits, total })
}

fn hash_to_hex(value: &Value) -> Option<String> {
    if let Some(s) = value.as_str() {
        return Some(s.to_string());
    }
    let bytes: Vec<u8> = value
        .as_array()?
        .iter()
        .map(|byte| byte.as_u64().map(|n| n as u8))
        .collect::<Option<Vec<_>>>()?;
    Some(hex::encode(bytes))
}

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

    #[test]
    fn parses_byte_array_hashes() {
        let value = json!([1, 2, 255]);
        assert_eq!(hash_to_hex(&value).unwrap(), "0102ff");
    }

    #[test]
    fn parses_branch_response() {
        let value = json!({
            "reference": {"name": "main", "hash": "abc"},
            "commit": {}
        });
        let head = branch_head(&value).unwrap();
        assert_eq!(head.name, "main");
        assert_eq!(head.head_sha, "abc");
    }

    #[test]
    fn a_record_without_a_repo_did_is_kept_by_owner_and_name() {
        let hit = json!({
            "uri": "at://did:plc:owner/sh.tangled.repo/3abc",
            "value": {"name": "old", "knot": "knot1.tangled.sh"}
        });
        let repo = repo_from_search_hit(&hit).unwrap();
        assert_eq!(repo.repo_did, "did:plc:owner/old");
        assert!(is_legacy_repo(&repo.repo_did));
        assert_eq!(repo.url, "https://tangled.org/did:plc:owner/old");
        let current = json!({
            "uri": "at://did:plc:owner/sh.tangled.repo/3abd",
            "value": {"name": "new", "knot": "knot1.tangled.sh", "repoDid": "did:plc:repo"}
        });
        assert_eq!(
            repo_from_search_hit(&current).unwrap().repo_did,
            "did:plc:repo"
        );
        assert!(!is_legacy_repo("did:plc:repo"));
        // Without a knot, or with a name that can't be a path segment, a
        // legacy record can't be addressed.
        let no_knot = json!({"name": "old"});
        assert!(repo_from_record("did:plc:owner", &no_knot, Some("3abc")).is_none());
        let bad_name = json!({"name": "a/b", "knot": "knot1.tangled.sh"});
        assert!(repo_from_record("did:plc:owner", &bad_name, Some("3abc")).is_none());
        // The record key stands in for a missing name.
        let unnamed = json!({"knot": "knot1.tangled.sh"});
        assert_eq!(
            repo_from_record("did:plc:owner", &unnamed, Some("old"))
                .unwrap()
                .repo_did,
            "did:plc:owner/old"
        );
    }

    #[test]
    fn parses_a_commit_page() {
        let body = json!({
            "total": 1,
            "commits": [{
                "hash": [10, 20],
                "author": {"Name": "@me", "Email": "did:plc:me", "When": "2026-10-20T12:00:00-04:00"},
                "committer": {"When": "2026-10-20T16:00:01Z"},
                "message": "subject\n\nbody"
            }]
        });
        let page = parse_commit_page(&body).unwrap();
        assert_eq!(page.commits[0].sha, "0a14");
        assert_eq!(page.commits[0].authored_at, "2026-10-20T16:00:00Z");
        assert_eq!(page.commits[0].message, "subject");
    }
}