tangled-cli 0.1.0

//! Repository operations: PDS record CRUD plus knot orchestration.

use anyhow::{anyhow, Result};
use serde::Deserialize;
use tangled_api::lexicon::com::atproto::{identity, repo as atproto_repo};
use tangled_api::lexicon::sh::tangled::repo as tangled_repo;
use tangled_api::xrpc;

use super::auth::PdsAuth;
use super::types::{
    CreateRepoOptions, DefaultBranch, Language, Languages, Pipeline,
    PipelineRecord, RepoDescribe, RepoRecord, Repository, StarRecord,
    REPO_COLLECTION,
};
use super::{http, service_auth, uri_did, uri_rkey};

pub async fn resolve_handle_to_did(
    pds_base: &str,
    user: &str,
    auth: &PdsAuth,
) -> Result<String> {
    if user.starts_with("did:") {
        return Ok(user.to_string());
    }
    let params = identity::resolve_handle::Params {
        handle: user.to_string(),
    };
    let rb = identity::resolve_handle(http(), pds_base, &params);
    let out: identity::resolve_handle::Output = auth.send(rb).await?;
    Ok(out.did)
}

async fn list_collection_records(
    pds_base: &str,
    repo: &str,
    collection: &str,
    auth: &PdsAuth,
) -> Result<Vec<atproto_repo::list_records::Record>> {
    let params = atproto_repo::list_records::Params {
        repo: repo.to_string(),
        collection: collection.to_string(),
        limit: Some(100),
        cursor: None,
        reverse: None,
    };
    let rb = atproto_repo::list_records(http(), pds_base, &params);
    let out: atproto_repo::list_records::Output = auth.send(rb).await?;
    Ok(out.records)
}

pub async fn list_repos(
    pds_base: &str,
    user: Option<&str>,
    knot: Option<&str>,
    starred: bool,
    auth: &PdsAuth,
) -> Result<Vec<Repository>> {
    let user = user.ok_or_else(|| {
        anyhow!("missing user for list_repos; provide handle or DID")
    })?;
    let did = resolve_handle_to_did(pds_base, user, auth).await?;

    let records =
        list_collection_records(pds_base, &did, REPO_COLLECTION, auth).await?;
    let mut repos: Vec<Repository> = Vec::new();
    for record in records {
        let mut val: Repository = serde_json::from_value(record.value)?;
        let rkey = uri_rkey(&record.uri);
        if val.rkey.is_none() {
            val.rkey = rkey.clone();
        }
        if val.name.is_empty() {
            if let Some(k) = rkey {
                val.name = k;
            }
        }
        if val.did.is_none() {
            if let Some(d) = uri_did(&record.uri) {
                val.did = Some(d);
            }
        }
        repos.push(val);
    }
    // Apply optional filters client-side
    if let Some(k) = knot {
        repos.retain(|r| r.knot.as_deref().unwrap_or("") == k);
    }
    if starred {
        // TODO: implement starred filtering when API is available. For now, no-op.
    }
    Ok(repos)
}

pub async fn create_repo(
    knot_base: &str,
    opts: CreateRepoOptions<'_>,
) -> Result<()> {
    // 1) Create the sh.tangled.repo record on the user's PDS
    let record = tangled_repo::Record {
        r#type: Some(REPO_COLLECTION.to_string()),
        name: Some(opts.name.to_string()),
        knot: opts.knot.to_string(),
        description: opts.description.map(str::to_string),
        created_at: chrono::Utc::now().to_rfc3339(),
        ..Default::default()
    };
    let input = atproto_repo::create_record::Input {
        repo: opts.did.to_string(),
        collection: REPO_COLLECTION.to_string(),
        rkey: Some(opts.name.to_string()),
        validate: Some(false),
        record: serde_json::to_value(&record)?,
        swap_commit: None,
    };
    let rb = atproto_repo::create_record(http(), opts.pds_base, &input);
    let created: Result<atproto_repo::create_record::Output> =
        opts.auth.send(rb).await;

    // Extract rkey from at-uri: at://did/collection/rkey. Some PDSes
    // occasionally return a 500 after committing createRecord; because we use
    // a deterministic rkey, verify the record before failing hard.
    let rkey = match created {
        Ok(created) => uri_rkey(&created.uri)
            .ok_or_else(|| anyhow!("failed to parse rkey from uri"))?,
        Err(err) => {
            if repo_record_exists(opts.pds_base, opts.did, opts.name, opts.auth)
                .await
            {
                opts.name.to_string()
            } else {
                return Err(err);
            }
        }
    };

    // 2) Obtain a service auth token for the knot (aud = did:web:<host>)
    let token = service_auth::mint(
        opts.pds_base,
        opts.auth,
        knot_base,
        tangled_repo::create::NSID,
    )
    .await?;

    // 3) Call sh.tangled.repo.create with the rkey
    let input = tangled_repo::create::Input {
        rkey: rkey.clone(),
        name: opts.name.to_string(),
        default_branch: opts.default_branch.map(str::to_string),
        source: opts.source.map(str::to_string),
        repo_did: None,
    };
    let rb =
        tangled_repo::create(http(), knot_base, &input).bearer_auth(&token);
    let created_repo: tangled_repo::create::Output =
        super::xrpc_send(rb).await?;
    if let Some(repo_did) = created_repo.repo_did.as_deref() {
        update_repo_record(opts.pds_base, opts.did, &rkey, opts.auth, |obj| {
            obj.insert("repoDid".to_string(), serde_json::json!(repo_did));
        })
        .await?;
    }
    Ok(())
}

async fn repo_record_exists(
    pds_base: &str,
    did: &str,
    rkey: &str,
    auth: &PdsAuth,
) -> bool {
    let params = atproto_repo::get_record::Params {
        repo: did.to_string(),
        collection: REPO_COLLECTION.to_string(),
        rkey: rkey.to_string(),
        cid: None,
    };
    let rb = atproto_repo::get_record(http(), pds_base, &params);
    auth.send::<atproto_repo::get_record::Output>(rb)
        .await
        .is_ok()
}

/// Fetches the sh.tangled.repo record, applies `mutate` to it, and writes it
/// back with putRecord. Unknown fields are preserved because the record is
/// round-tripped as raw JSON.
async fn update_repo_record(
    pds_base: &str,
    did: &str,
    rkey: &str,
    auth: &PdsAuth,
    mutate: impl FnOnce(&mut serde_json::Map<String, serde_json::Value>),
) -> Result<()> {
    let params = atproto_repo::get_record::Params {
        repo: did.to_string(),
        collection: REPO_COLLECTION.to_string(),
        rkey: rkey.to_string(),
        cid: None,
    };
    let rb = atproto_repo::get_record(http(), pds_base, &params);
    let got: atproto_repo::get_record::Output = auth.send(rb).await?;

    let mut record = got.value;
    let Some(obj) = record.as_object_mut() else {
        return Err(anyhow!("repo record is not a JSON object"));
    };
    mutate(obj);
    obj.entry("$type".to_string())
        .or_insert_with(|| serde_json::json!(REPO_COLLECTION));

    let input = atproto_repo::put_record::Input {
        repo: did.to_string(),
        collection: REPO_COLLECTION.to_string(),
        rkey: rkey.to_string(),
        validate: Some(false),
        record,
        swap_commit: None,
        swap_record: None,
    };
    let rb = atproto_repo::put_record(http(), pds_base, &input);
    let _: serde_json::Value = auth.send(rb).await?;
    Ok(())
}

fn repo_record_name(uri: &str, value: &Repository) -> String {
    if value.name.is_empty() {
        uri_rkey(uri).unwrap_or_default()
    } else {
        value.name.clone()
    }
}

pub async fn get_repo_info(
    pds_base: &str,
    owner: &str,
    name: &str,
    auth: &PdsAuth,
) -> Result<RepoRecord> {
    let did = resolve_handle_to_did(pds_base, owner, auth).await?;
    let records =
        list_collection_records(pds_base, &did, REPO_COLLECTION, auth).await?;
    for item in records {
        let value: Repository = serde_json::from_value(item.value)?;
        let record_name = repo_record_name(&item.uri, &value);
        if record_name == name {
            let rkey = uri_rkey(&item.uri)
                .ok_or_else(|| anyhow!("missing rkey in uri"))?;
            return Ok(RepoRecord {
                did: did.clone(),
                name: record_name,
                rkey,
                knot: value.knot.unwrap_or_default(),
                description: value.description,
                source: value.source,
                spindle: value.spindle,
                repo_did: value.repo_did,
            });
        }
    }
    Err(anyhow!("repo not found for owner/name"))
}

pub async fn get_repo_by_rkey(
    pds_base: &str,
    did: &str,
    rkey: &str,
    auth: &PdsAuth,
) -> Result<Repository> {
    let params = atproto_repo::get_record::Params {
        repo: did.to_string(),
        collection: REPO_COLLECTION.to_string(),
        rkey: rkey.to_string(),
        cid: None,
    };
    let rb = atproto_repo::get_record(http(), pds_base, &params);
    let out: atproto_repo::get_record::Output = auth.send(rb).await?;
    Ok(serde_json::from_value(out.value)?)
}

pub async fn resolve_did_to_handle(
    pds_base: &str,
    did: &str,
    auth: &PdsAuth,
) -> Result<String> {
    let params = atproto_repo::describe_repo::Params {
        repo: did.to_string(),
    };
    let rb = atproto_repo::describe_repo(http(), pds_base, &params);
    let out: atproto_repo::describe_repo::Output = auth.send(rb).await?;
    Ok(out.handle)
}

/// Resolve either a repository DID or a legacy repository AT-URI to an
/// owner/name string suitable for human-readable list output.
pub async fn repo_display_name(
    pds_base: &str,
    repo_ref: &str,
    auth: &PdsAuth,
) -> Result<String> {
    let (owner_did, rkey) = if repo_ref.starts_with("at://") {
        let owner_did = uri_did(repo_ref)
            .ok_or_else(|| anyhow!("repository URI missing owner DID"))?;
        let rkey = uri_rkey(repo_ref)
            .ok_or_else(|| anyhow!("repository URI missing rkey"))?;
        (owner_did, rkey)
    } else {
        let resolver_base = std::env::var("TANGLED_API_BASE")
            .unwrap_or_else(|_| super::DEFAULT_API_BASE.to_string());
        let described = describe_repo(&resolver_base, repo_ref, None).await?;
        (described.owner_did, described.rkey)
    };

    let repo = get_repo_by_rkey(pds_base, &owner_did, &rkey, auth).await?;
    let owner = resolve_did_to_handle(pds_base, &owner_did, auth)
        .await
        .unwrap_or(owner_did);
    let name = if repo.name.is_empty() {
        rkey
    } else {
        repo.name
    };
    Ok(format!("{}/{}", owner, name))
}

pub async fn describe_repo(
    api_base: &str,
    repo_did: &str,
    bearer: Option<&str>,
) -> Result<RepoDescribe> {
    // Bobbin's native lookup for repo DID -> owner/rkey is
    // `sh.tangled.repo.getRepoByRepoDid`. `describeRepo` is a knot endpoint;
    // Bobbin only proxies it for already-known repo AT-URIs, so calling it with
    // just `repoDid` returns "missing repo".
    if bearer.is_none() {
        match get_repo_by_repo_did(api_base, repo_did).await {
            Ok(out) => return Ok(out),
            Err(err) if is_default_bobbin_base(api_base) => return Err(err),
            Err(_) => {}
        }
    }

    describe_repo_via_knot(api_base, repo_did, bearer).await
}

async fn describe_repo_via_knot(
    knot_base: &str,
    repo_did: &str,
    bearer: Option<&str>,
) -> Result<RepoDescribe> {
    let params = tangled_repo::describe_repo::Params {
        repo_did: repo_did.to_string(),
    };
    let mut rb = tangled_repo::describe_repo(http(), knot_base, &params);
    if let Some(token) = bearer {
        rb = rb.bearer_auth(token);
    }
    Ok(super::xrpc_send(rb).await?)
}

fn is_default_bobbin_base(base: &str) -> bool {
    let with_scheme;
    let normalized =
        if base.starts_with("http://") || base.starts_with("https://") {
            base.trim_end_matches('/')
        } else {
            with_scheme = format!("https://{}", base.trim_end_matches('/'));
            &with_scheme
        };
    reqwest::Url::parse(normalized)
        .ok()
        .and_then(|url| url.host_str().map(str::to_ascii_lowercase))
        .is_some_and(|host| host == "api.tangled.org")
}

#[derive(Debug, Deserialize)]
struct BobbinRepoRecord {
    uri: String,
}

async fn get_repo_by_repo_did(
    api_base: &str,
    repo_did: &str,
) -> Result<RepoDescribe> {
    let rb = http()
        .get(xrpc::xrpc_url(api_base, "sh.tangled.repo.getRepoByRepoDid"))
        .query(&[("repoDid", repo_did)]);
    let out: BobbinRepoRecord = super::xrpc_send(rb).await?;
    let owner_did = uri_did(&out.uri)
        .ok_or_else(|| anyhow!("Bobbin repo response uri missing owner DID"))?;
    let rkey = uri_rkey(&out.uri)
        .ok_or_else(|| anyhow!("Bobbin repo response uri missing rkey"))?;
    Ok(RepoDescribe {
        owner_did,
        repo_did: repo_did.to_string(),
        rkey,
    })
}

pub async fn delete_repo(
    knot_base: &str,
    did: &str,
    name: &str,
    pds_base: &str,
    auth: &PdsAuth,
) -> Result<()> {
    let info = get_repo_info(pds_base, did, name, auth).await?;

    let input = atproto_repo::delete_record::Input {
        repo: did.to_string(),
        collection: REPO_COLLECTION.to_string(),
        rkey: info.rkey.clone(),
        swap_commit: None,
        swap_record: None,
    };
    let rb = atproto_repo::delete_record(http(), pds_base, &input);
    auth.send_unit(rb).await?;

    let token = service_auth::mint(
        pds_base,
        auth,
        knot_base,
        tangled_repo::delete::NSID,
    )
    .await?;
    let input = tangled_repo::delete::Input {
        did: did.to_string(),
        name: name.to_string(),
        rkey: info.rkey,
    };
    let rb =
        tangled_repo::delete(http(), knot_base, &input).bearer_auth(&token);
    super::xrpc_send_unit(rb).await?;
    Ok(())
}

pub async fn update_repo_knot(
    pds_base: &str,
    did: &str,
    rkey: &str,
    new_knot: &str,
    auth: &PdsAuth,
) -> Result<()> {
    update_repo_record(pds_base, did, rkey, auth, |obj| {
        obj.insert("knot".to_string(), serde_json::json!(new_knot));
    })
    .await
}

pub async fn update_repo_spindle(
    pds_base: &str,
    did: &str,
    rkey: &str,
    new_spindle: Option<&str>,
    auth: &PdsAuth,
) -> Result<()> {
    update_repo_record(pds_base, did, rkey, auth, |obj| {
        if let Some(spindle) = new_spindle {
            obj.insert("spindle".to_string(), serde_json::json!(spindle));
        } else {
            obj.remove("spindle");
        }
    })
    .await
}

pub async fn get_default_branch(
    knot_host: &str,
    did: &str,
    name: &str,
) -> Result<DefaultBranch> {
    let params = tangled_repo::get_default_branch::Params {
        repo: format!("{}/{}", did, name),
    };
    let rb = tangled_repo::get_default_branch(http(), knot_host, &params);
    let out: tangled_repo::get_default_branch::Output =
        super::xrpc_send(rb).await?;
    Ok(DefaultBranch {
        name: out.name,
        hash: out.hash,
        short_hash: out.short_hash,
        when: out.when,
        message: out.message,
    })
}

pub async fn get_languages(
    knot_host: &str,
    did: &str,
    name: &str,
) -> Result<Languages> {
    let params = tangled_repo::languages::Params {
        repo: format!("{}/{}", did, name),
        r#ref: None,
    };
    let rb = tangled_repo::languages(http(), knot_host, &params);
    // Live knots don't always return the full lexicon shape; stay lenient.
    let res: serde_json::Value = super::xrpc_send(rb).await?;
    let langs = res
        .get("languages")
        .cloned()
        .unwrap_or(serde_json::json!([]));
    let languages: Vec<Language> = serde_json::from_value(langs)?;
    Ok(Languages {
        languages,
        total_size: res.get("totalSize").and_then(|v| v.as_u64()),
        total_files: res.get("totalFiles").and_then(|v| v.as_u64()),
    })
}

pub async fn star_repo(
    pds_base: &str,
    auth: &PdsAuth,
    subject_at_uri: &str,
    user_did: &str,
) -> Result<String> {
    let input = atproto_repo::create_record::Input {
        repo: user_did.to_string(),
        collection: "sh.tangled.feed.star".to_string(),
        rkey: None,
        validate: Some(false),
        record: serde_json::json!({
            "subject": subject_at_uri,
            "createdAt": super::now_rfc3339(),
        }),
        swap_commit: None,
    };
    let rb = atproto_repo::create_record(http(), pds_base, &input);
    let out: atproto_repo::create_record::Output = auth.send(rb).await?;
    uri_rkey(&out.uri).ok_or_else(|| anyhow!("missing rkey in star uri"))
}

pub async fn unstar_repo(
    pds_base: &str,
    auth: &PdsAuth,
    subject_at_uri: &str,
    user_did: &str,
) -> Result<()> {
    let records = list_collection_records(
        pds_base,
        user_did,
        "sh.tangled.feed.star",
        auth,
    )
    .await?;
    let mut rkey = None;
    for item in records {
        let value: StarRecord = match serde_json::from_value(item.value) {
            Ok(v) => v,
            Err(_) => continue,
        };
        if value.subject == subject_at_uri {
            rkey = uri_rkey(&item.uri);
            if rkey.is_some() {
                break;
            }
        }
    }
    let rkey = rkey.ok_or_else(|| anyhow!("star record not found"))?;
    let input = atproto_repo::delete_record::Input {
        repo: user_did.to_string(),
        collection: "sh.tangled.feed.star".to_string(),
        rkey,
        swap_commit: None,
        swap_record: None,
    };
    let rb = atproto_repo::delete_record(http(), pds_base, &input);
    auth.send_unit(rb).await?;
    Ok(())
}

pub async fn list_pipelines(
    pds_base: &str,
    repo_did: &str,
    auth: &PdsAuth,
) -> Result<Vec<PipelineRecord>> {
    let records = list_collection_records(
        pds_base,
        repo_did,
        "sh.tangled.pipeline",
        auth,
    )
    .await?;
    let mut out = vec![];
    for item in records {
        let pipeline: Pipeline = serde_json::from_value(item.value)?;
        out.push(PipelineRecord {
            rkey: uri_rkey(&item.uri).unwrap_or_default(),
            pipeline,
        });
    }
    Ok(out)
}

#[cfg(test)]
mod tests {
    use mockito::{Matcher, Server};
    use serde_json::json;

    use super::*;

    #[tokio::test]
    async fn get_repo_info_uses_rkey_when_live_record_has_no_name() {
        let mut pds = Server::new_async().await;
        let _list = pds
            .mock("GET", "/xrpc/com.atproto.repo.listRecords")
            .match_query(Matcher::AllOf(vec![
                Matcher::UrlEncoded("repo".into(), "did:plc:owner".into()),
                Matcher::UrlEncoded(
                    "collection".into(),
                    REPO_COLLECTION.into(),
                ),
            ]))
            .with_status(200)
            .with_body(
                json!({
                    "records": [{
                        "uri": "at://did:plc:owner/sh.tangled.repo/core",
                        "cid": "bafycid",
                        "value": {
                            "$type": "sh.tangled.repo",
                            "knot": "knot1.tangled.sh",
                            "repoDid": "did:plc:repo",
                            "createdAt": "2025-02-23T18:43:18Z"
                        }
                    }]
                })
                .to_string(),
            )
            .create_async()
            .await;

        let auth = PdsAuth::Bearer("access".to_string());
        let info = get_repo_info(&pds.url(), "did:plc:owner", "core", &auth)
            .await
            .expect("repo info should deserialize live records that omit name");

        assert_eq!(info.name, "core");
        assert_eq!(info.rkey, "core");
        assert_eq!(info.repo_did.as_deref(), Some("did:plc:repo"));
    }

    #[tokio::test]
    async fn describe_repo_uses_bobbin_get_repo_by_repo_did() {
        let mut api = Server::new_async().await;
        let repo_did = "did:plc:repo";

        let _bobbin_get = api
            .mock("GET", "/xrpc/sh.tangled.repo.getRepoByRepoDid")
            .match_query(Matcher::UrlEncoded("repoDid".into(), repo_did.into()))
            .with_status(200)
            .with_body(
                json!({
                    "uri": "at://did:plc:owner/sh.tangled.repo/demo",
                    "cid": "bafycid",
                    "value": {"$type":"sh.tangled.repo","repoDid": repo_did}
                })
                .to_string(),
            )
            .create_async()
            .await;

        let described = describe_repo(&api.url(), repo_did, None)
            .await
            .expect("Bobbin getRepoByRepoDid should resolve repo metadata");

        assert_eq!(described.owner_did, "did:plc:owner");
        assert_eq!(described.repo_did, repo_did);
        assert_eq!(described.rkey, "demo");
    }

    #[tokio::test]
    async fn create_repo_sends_current_knot_shape_and_preserves_record_fields()
    {
        let mut pds = Server::new_async().await;
        let mut knot = Server::new_async().await;
        let knot_host = knot.url().trim_start_matches("http://").to_string();

        let _create_record = pds
            .mock("POST", "/xrpc/com.atproto.repo.createRecord")
            .match_header("authorization", "Bearer access")
            .match_body(Matcher::PartialJson(json!({
                "repo": "did:plc:owner",
                "collection": "sh.tangled.repo",
                "rkey": "demo",
                "record": {
                    "$type": "sh.tangled.repo",
                    "name": "demo",
                    "knot": knot_host
                }
            })))
            .with_status(200)
            .with_body(
                json!({
                    "uri": "at://did:plc:owner/sh.tangled.repo/demo",
                    "cid": "bafycid"
                })
                .to_string(),
            )
            .create_async()
            .await;

        let _service_auth = pds
            .mock("GET", "/xrpc/com.atproto.server.getServiceAuth")
            .match_header("authorization", "Bearer access")
            .match_query(Matcher::AllOf(vec![
                Matcher::UrlEncoded(
                    "aud".into(),
                    format!("did:web:{knot_host}"),
                ),
                Matcher::UrlEncoded(
                    "lxm".into(),
                    "sh.tangled.repo.create".into(),
                ),
            ]))
            .with_status(200)
            .with_body(json!({"token": "service"}).to_string())
            .create_async()
            .await;

        let _knot_create = knot
            .mock("POST", "/xrpc/sh.tangled.repo.create")
            .match_header("authorization", "Bearer service")
            .match_body(Matcher::PartialJson(json!({
                "rkey": "demo",
                "name": "demo"
            })))
            .with_status(200)
            .with_body(json!({"repoDid": "did:plc:repo"}).to_string())
            .create_async()
            .await;

        let _get_record = pds
            .mock("GET", "/xrpc/com.atproto.repo.getRecord")
            .match_query(Matcher::AllOf(vec![
                Matcher::UrlEncoded("repo".into(), "did:plc:owner".into()),
                Matcher::UrlEncoded("collection".into(), REPO_COLLECTION.into()),
                Matcher::UrlEncoded("rkey".into(), "demo".into()),
            ]))
            .with_status(200)
            .with_body(
                json!({
                    "uri": "at://did:plc:owner/sh.tangled.repo/demo",
                    "value": {
                        "$type": "sh.tangled.repo",
                        "name": "demo",
                        "knot": knot_host,
                        "labels": ["at://did:plc:owner/sh.tangled.label.definition/good-first-issue"],
                        "createdAt": "2026-07-04T12:00:00Z"
                    }
                })
                .to_string(),
            )
            .create_async()
            .await;

        let _put_record = pds
            .mock("POST", "/xrpc/com.atproto.repo.putRecord")
            .match_body(Matcher::PartialJson(json!({
                "repo": "did:plc:owner",
                "collection": "sh.tangled.repo",
                "rkey": "demo",
                "record": {
                    "repoDid": "did:plc:repo",
                    "labels": ["at://did:plc:owner/sh.tangled.label.definition/good-first-issue"]
                }
            })))
            .with_status(200)
            .with_body(
                json!({
                    "uri": "at://did:plc:owner/sh.tangled.repo/demo",
                    "cid": "bafycid"
                })
                .to_string(),
            )
            .create_async()
            .await;

        let auth = PdsAuth::Bearer("access".to_string());
        create_repo(
            &knot.url(),
            CreateRepoOptions {
                did: "did:plc:owner",
                name: "demo",
                knot: &knot_host,
                description: None,
                default_branch: None,
                source: None,
                pds_base: &pds.url(),
                auth: &auth,
            },
        )
        .await
        .expect("repo creation should follow the current knot contract");
    }
}