//! owner-side record emission: `knot-static repo record <repoDid>`.

use std::sync::Arc;

use anyhow::Context as _;
use http::StatusCode;
use knot_runtime::HttpTransport;
use knot_types::RepoDid;

use crate::config::ValidatedConfig;
use crate::registry::format_rfc3339;

pub async fn record(
    config: Arc<ValidatedConfig>,
    repo_did: &str,
    publish: bool,
) -> anyhow::Result<()> {
    let repo_did = RepoDid::new(repo_did.to_string()).context("invalid repo DID")?;
    let plan = build_record_plan(&config, &repo_did)?;
    let RecordPlan {
        owner: owner_str,
        payload,
        ..
    } = &plan;

    println!("{}", serde_json::to_string_pretty(&payload)?);

    if publish {
        let owner_account = knot_types::AccountDid::new(owner_str.clone())
            .context("invalid owner DID in plan")?;
        let mut session = crate::oauth::ensure_auth_for_owner(&config, &owner_account, true)
            .await?
            .context("OAuth session required for publishing")?;
        publish_with_resync_ping(config, &repo_did, &plan, &mut session).await?;
        println!(
            "{}",
            serde_json::to_string_pretty(&serde_json::json!({
                "published": true,
            }))?
        );
    }
    Ok(())
}

/// everything needed to emit (and optionally publish) one repo's record:
/// rkey/name/createdAt/owner from the registry COB, falling back to the index.
pub(crate) struct RecordPlan {
    pub rkey: String,
    pub owner: String,
    pub payload: serde_json::Value,
}

fn build_record_plan(
    config: &Arc<ValidatedConfig>,
    repo_did: &RepoDid,
) -> anyhow::Result<RecordPlan> {
    let layout = config.layout()?;
    let meta_path = layout.meta_path(&config.knot_did)?;
    let meta_repo = layout
        .bootstrap_meta(&config.knot_did)
        .context("open meta.git")?;

    let cob_store = knot_cob::CobStore::new(&meta_repo);
    let cob_objects = cob_store.list::<knot_cobs::RepoRegistryCob>()?;
    let registry_state = cob_objects
        .first()
        .map(|&object| cob_store.get::<knot_cobs::RepoRegistryCob>(object))
        .transpose()?
        .map(|cob| cob.into_state());

    let (rkey_str, name_str, created_at_iso, owner_str) = if let Some(record) = registry_state
        .as_ref()
        .and_then(|reg| reg.record_of(repo_did))
    {
        (
            record.rkey.to_string(),
            record.name.to_string(),
            chrono::DateTime::from_timestamp(record.created_at.get(), 0)
                .map(|dt| dt.to_rfc3339_opts(chrono::SecondsFormat::Secs, true))
                .unwrap_or_else(|| format_rfc3339(record.created_at.get())),
            record.owner.to_string(),
        )
    } else {
        let index = knot_index::Index::new(&meta_path, layout);
        let _ = index.rebuild();
        match index.ownership_of(repo_did) {
            knot_index::Resolved::Ready(Some(ownership)) => {
                let rkey = ownership.rkey.to_string();
                let owner = ownership.owner.to_string();
                let now = std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_secs() as i64;
                // the index has no cosmetic name; fall back to the rkey
                (rkey.clone(), rkey, format_rfc3339(now), owner)
            }
            _ => anyhow::bail!("repo {repo_did} is not registered"),
        }
    };

    let payload = serde_json::json!({
        "collection": "sh.tangled.repo",
        "rkey": rkey_str,
        "record": {
            "$type": "sh.tangled.repo",
            "name": name_str,
            "knot": config.hostname.as_str(),
            // `repo` is the lexicon field (the repo DID); repoDid is kept for
            // compatibility with our own earlier readers.
            "repo": repo_did.as_str(),
            "repoDid": repo_did.as_str(),
            "createdAt": created_at_iso,
        }
    });

    Ok(RecordPlan {
        rkey: rkey_str,
        owner: owner_str,
        payload,
    })
}

/// re-publish changed repos' records for owners with an active OAuth session.
/// a static knot emits no firehose, so a record update is the only signal that
/// re-enqueues a knotmirror resync.
///
/// a re-publish with identical bytes yields an identical cid, which mirrors treat as
/// no change. so when the built payload matches what's already published, createdAt is
/// nudged +-1s across two writes: the record ends up semantically identical but both
/// writes carry fresh cids, forcing the change through cid-deduping pipelines.
/// per-repo failures are logged; they never fail an activation.
pub(crate) async fn publish_all(config: Arc<ValidatedConfig>, changed_repos: &[String]) {
    let layout = match config.layout() {
        Ok(layout) => layout,
        Err(error) => {
            eprintln!("warn: record publish: {error:#}");
            return;
        }
    };
    let meta_path = match layout.meta_path(&config.knot_did) {
        Ok(meta_path) => meta_path,
        Err(error) => {
            eprintln!("warn: record publish: {error:#}");
            return;
        }
    };
    let index = knot_index::Index::new(&meta_path, layout.clone());
    let _ = index.rebuild();

    let mut targets = Vec::new();
    for did in index.hosted_repos() {
        if !changed_repos
            .iter()
            .any(|repo_did| repo_did == did.as_str())
        {
            continue;
        }
        let knot_index::Resolved::Ready(Some(ownership)) = index.ownership_of(&did) else {
            continue;
        };
        let owner_did_str = ownership.owner.to_string();
        let Ok(owner_account) = knot_types::AccountDid::new(owner_did_str.clone()) else {
            continue;
        };
        let plan = match build_record_plan(&config, &did) {
            Ok(plan) => plan,
            Err(error) => {
                eprintln!("warn: record publish {}: {error:#}", did.as_str());
                continue;
            }
        };
        targets.push((did, owner_account, plan));
    }

    if targets.is_empty() {
        return;
    }

    let mut sessions: std::collections::HashMap<String, Option<crate::oauth::StoredSession>> =
        std::collections::HashMap::new();
    for (_, owner_account, _) in &targets {
        let owner_key = owner_account.to_string();
        if sessions.contains_key(&owner_key) {
            continue;
        }
        let session = match crate::oauth::get_or_refresh_session(&config, owner_account).await {
            Ok(Some(s)) => Some(s),
            Ok(None) => {
                eprintln!("record publish: skipped (no OAuth session for owner {owner_key})");
                None
            }
            Err(error) => {
                eprintln!("warn: oauth session lookup for {owner_key}: {error:#}");
                None
            }
        };
        sessions.insert(owner_key, session);
    }

    let publish_futures = targets.into_iter().filter_map(|(did, owner_account, plan)| {
        let session = sessions.get(&owner_account.to_string())?.clone()?;
        let config = config.clone();
        Some(async move {
            let mut session = session;
            match publish_with_resync_ping(config, &did, &plan, &mut session).await {
                Ok(()) => eprintln!(
                    "published sh.tangled.repo record for {} via OAuth (mirror resync requested)",
                    did.as_str()
                ),
                Err(error) => eprintln!("warn: record publish {}: {error:#}", did.as_str()),
            }
        })
    });

    futures::future::join_all(publish_futures).await;
}

/// publish the record; when the built payload equals what's already published,
/// the +-1s createdAt nudge makes the write observable downstream.
async fn publish_with_resync_ping(
    config: Arc<ValidatedConfig>,
    _did: &RepoDid,
    plan: &RecordPlan,
    session: &mut crate::oauth::StoredSession,
) -> anyhow::Result<()> {
    let payload = plan.payload.clone();
    crate::oauth::put_record_oauth(
        session,
        "sh.tangled.repo",
        &plan.rkey,
        &payload["record"],
    )
    .await?;

    let published = fetch_published_record(&config, plan).await?;
    if published.as_ref() != Some(&payload["record"]) {
        // first write was a real change: nothing was published before, or the content drifted.
        return Ok(());
    }

    // Identical bytes published: nudge createdAt +1s then restore to force fresh CIDs downstream.
    let original = payload["record"]["createdAt"]
        .as_str()
        .context("record payload missing createdAt")?
        .to_owned();
    let shifted = shift_rfc3339_seconds(&original, 1).context("shift createdAt")?;

    let mut nudged = payload["record"].clone();
    nudged["createdAt"] = serde_json::Value::String(shifted);
    crate::oauth::put_record_oauth(session, "sh.tangled.repo", &plan.rkey, &nudged).await?;

    let mut restored = payload["record"].clone();
    restored["createdAt"] = serde_json::Value::String(original);
    crate::oauth::put_record_oauth(session, "sh.tangled.repo", &plan.rkey, &restored).await?;
    Ok(())
}

/// read the currently published record from the repo owner's PDS for the no-change check.
async fn fetch_published_record(
    config: &Arc<ValidatedConfig>,
    plan: &RecordPlan,
) -> anyhow::Result<Option<serde_json::Value>> {
    let transport = knot_runtime::ReqwestHttp::new(knot_runtime::HttpLimits::default())
        .context("build http transport")?;
    let atproto = knot_atproto::Atproto::new(
        knot_runtime::ReqwestHttp::new(knot_runtime::HttpLimits::default())
            .context("build identity http transport")?,
        knot_runtime::SystemClock,
        config.knot_did.clone(),
        knot_atproto::PlcDirectory::new(config.plc_directory.clone())?,
    );
    let owner_account = knot_types::AccountDid::new(plan.owner.clone())
        .context("repo owner DID is not a valid account DID")?;
    let identity = atproto
        .resolve_identity(&owner_account)
        .await
        .context("resolve repo owner DID to find its PDS")?;
    let mut get_url = identity
        .pds
        .url()
        .join("xrpc/com.atproto.repo.getRecord")
        .context("build getRecord url")?;
    get_url
        .query_pairs_mut()
        .append_pair("repo", plan.owner.as_str())
        .append_pair("collection", "sh.tangled.repo")
        .append_pair("rkey", &plan.rkey);
    let mut request = knot_runtime::HttpRequest::get(get_url);
    request.headers.insert(
        http::header::ACCEPT,
        http::HeaderValue::from_static("application/json"),
    );
    let response = transport
        .execute(request)
        .await
        .context("getRecord request")?;
    match response.status {
        StatusCode::NOT_FOUND => Ok(None),
        StatusCode::OK => {
            let body: serde_json::Value =
                serde_json::from_slice(&response.body).context("parse getRecord response")?;
            Ok(body.get("value").cloned())
        }
        status => anyhow::bail!("getRecord failed with HTTP {status}"),
    }
}

/// shift an RFC3339 timestamp by whole seconds, keeping second precision.
fn shift_rfc3339_seconds(stamp: &str, delta: i64) -> Option<String> {
    chrono::DateTime::parse_from_rfc3339(stamp).ok().map(|dt| {
        (dt + chrono::Duration::seconds(delta)).to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
    })
}