tga 4.0.2

//! Bitbucket Cloud REST v2.0 client — pull-request collection only.
//!
//! Mirrors the structure of [`crate::collect::github::client`] so the two
//! providers can be reviewed side-by-side. The notable differences:
//!
//! - **Pagination** is cursor-style: each response carries an absolute
//!   `next` URL. We follow `next` until it is `None` rather than walking
//!   a page counter.
//! - **Auth** supports either Bearer token (workspace / repo access token)
//!   or Basic auth (`username` + App Password). Tokens win when both are set.
//! - **Per-PR commit fetch** — the PR list payload only carries the (often
//!   abbreviated) merge commit hash, which cannot be joined against
//!   `commits.sha` and discards the PR's actual commit composition (#841).
//!   [`BitbucketClient::fetch_pr_commits`] hits the dedicated per-PR commits
//!   endpoint, paginating with the same `next`-cursor convention, so
//!   `commit_shas` carries the full list — one extra round-trip (or more,
//!   under pagination) per PR, same tradeoff GitHub's equivalent
//!   `fetch_pr_commits` makes.

use std::time::Duration;

use async_trait::async_trait;
use chrono::{DateTime, TimeZone, Utc};
use reqwest::header::{HeaderMap, HeaderValue, ACCEPT, USER_AGENT};
use rusqlite::params;
use tracing::{debug, warn};

use crate::collect::bitbucket::types::{BbCommit, BbPaged, BbPullRequest};
use crate::collect::errors::{CollectError, Result};
use crate::collect::pr_provider::PrProvider;
use crate::core::config::BitbucketConfig;
use crate::core::db::Database;
use crate::core::models::{PrState, PullRequest};

/// HTTP `User-Agent` string sent on every request.
const USER_AGENT_VALUE: &str = "trusty-git-analytics/0.1";
/// Default Bitbucket Cloud REST base URL.
const BITBUCKET_API_BASE: &str = "https://api.bitbucket.org/2.0";
/// Page size for paginated list endpoints (Bitbucket Cloud caps at 50).
const PAGE_SIZE: u32 = 50;
/// Maximum retry attempts for transient failures (5xx, 429).
const MAX_RETRIES: u32 = 3;
/// Base delay (in milliseconds) for exponential backoff: 1s, 2s, 4s.
const RETRY_BASE_MS: u64 = 1000;

/// Authentication credentials for the Bitbucket Cloud REST API.
///
/// Bearer wins when both modes are populated — repo / workspace access
/// tokens supersede legacy App Passwords.
// #5770: no `Debug` derive, deliberately — both variants carry a live
// Bitbucket credential, and nothing formats one, so the derive was pure risk.
#[derive(Clone)]
enum BbAuth {
    /// `Authorization: Bearer <token>` (workspace or repository access token).
    Bearer(String),
    /// `Authorization: Basic <base64(username:app_password)>`.
    Basic { username: String, password: String },
}

/// Resolve Bitbucket auth credentials from config plus an injected env lookup.
///
/// Why: the original logic read `std::env::var` directly, which made the auth
/// unit tests mutate process-global environment variables. Under the
/// multi-threaded test harness those mutations raced across tests, flaking
/// `app_password_falls_back_to_env_when_config_absent` in CI (issue #1653).
/// Threading the environment in as a closure removes the ambient-state
/// coupling: production passes `std::env::var`; tests pass an in-memory map,
/// so no test ever touches the real environment and the race cannot occur.
/// What: applies the documented precedence — (1) Bearer `token` (config,
/// then `BITBUCKET_TOKEN`), else (2) Basic auth `username` + `app_password`
/// (config with `${VAR}` expansion via the same `env` lookup, then
/// `BITBUCKET_APP_PASSWORD`). Config values written as `${MY_SECRET}` are
/// expanded against `env` before use (closes #842). Returns
/// [`CollectError::Config`] when no usable credential is available.
/// Test: `client::tests` — `resolve_auth_*` cases inject env maps directly
/// (no `std::env` mutation), covering bearer, basic, env-expansion,
/// precedence, env-fallback, and the missing-credential error branches.
fn resolve_auth(config: &BitbucketConfig, env: impl Fn(&str) -> Option<String>) -> Result<BbAuth> {
    // Expand `${VAR}` placeholders using the injected env lookup so config
    // and env resolution share one source of truth (no `std::env` here).
    let expand = |raw: &str| -> String {
        if let Some(var) = raw
            .strip_prefix("${")
            .and_then(|s| s.strip_suffix('}'))
            .filter(|v| !v.is_empty())
        {
            env(var).unwrap_or_default()
        } else {
            raw.to_string()
        }
    };
    let clean = |s: String| -> Option<String> {
        let t = s.trim().to_string();
        if t.is_empty() {
            None
        } else {
            Some(t)
        }
    };

    let token = config
        .token
        .as_deref()
        .map(&expand)
        .and_then(clean)
        .or_else(|| env("BITBUCKET_TOKEN").and_then(clean));

    if let Some(t) = token {
        return Ok(BbAuth::Bearer(t));
    }

    let username = config
        .username
        .as_deref()
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .ok_or_else(|| {
            CollectError::Config(
                "bitbucket auth missing: provide `token` or `username` + `app_password`".into(),
            )
        })?
        .to_string();

    // `app_password` may be written as `${MY_SECRET}` in YAML; without
    // expansion the literal `${MY_SECRET}` is sent as the Basic-auth password,
    // yielding a silent 401 (closes #842). Falls back to the
    // `BITBUCKET_APP_PASSWORD` env var, mirroring `token` above.
    let password = config
        .app_password
        .as_deref()
        .map(&expand)
        .and_then(clean)
        .or_else(|| env("BITBUCKET_APP_PASSWORD").and_then(clean))
        .ok_or_else(|| {
            CollectError::Config(
                "bitbucket auth missing: app_password (or BITBUCKET_APP_PASSWORD) \
                 required when token is unset"
                    .into(),
            )
        })?;

    Ok(BbAuth::Basic { username, password })
}

/// Async Bitbucket Cloud REST client.
pub struct BitbucketClient {
    client: reqwest::Client,
    auth: BbAuth,
    workspace: String,
    repo_slug: String,
    api_base: String,
}

impl BitbucketClient {
    /// Build a client from a [`BitbucketConfig`].
    ///
    /// Credential precedence:
    /// 1. Bearer `token` (config, then `BITBUCKET_TOKEN` env).
    /// 2. Basic auth: `username` + `app_password` (config, then
    ///    `BITBUCKET_APP_PASSWORD` env).
    ///
    /// # Errors
    ///
    /// - [`CollectError::Config`] if `workspace` / `repo_slug` are missing
    ///   or no usable auth mode is available.
    /// - [`CollectError::Http`] if the underlying `reqwest::Client` cannot
    ///   be built.
    pub fn new(config: &BitbucketConfig) -> Result<Self> {
        let workspace = config
            .workspace
            .as_deref()
            .map(str::trim)
            .filter(|s| !s.is_empty())
            .ok_or_else(|| CollectError::Config("bitbucket.workspace is required".into()))?
            .to_string();
        let repo_slug = config
            .repo_slug
            .as_deref()
            .map(str::trim)
            .filter(|s| !s.is_empty())
            .ok_or_else(|| CollectError::Config("bitbucket.repo_slug is required".into()))?
            .to_string();

        // Resolve credentials through the live process environment. The
        // env-coupled logic lives in `resolve_auth` so tests can exercise it
        // with an injected lookup instead of mutating `std::env` (issue #1653).
        let auth = resolve_auth(config, |name| std::env::var(name).ok())?;

        let mut headers = HeaderMap::new();
        headers.insert(USER_AGENT, HeaderValue::from_static(USER_AGENT_VALUE));
        headers.insert(ACCEPT, HeaderValue::from_static("application/json"));

        let client = reqwest::Client::builder()
            .default_headers(headers)
            .timeout(Duration::from_secs(30))
            .build()?;

        let api_base = config
            .api_base_url
            .as_deref()
            .map(str::trim)
            .filter(|s| !s.is_empty())
            .map(|s| s.trim_end_matches('/').to_string())
            .unwrap_or_else(|| BITBUCKET_API_BASE.to_string());

        Ok(Self {
            client,
            auth,
            workspace,
            repo_slug,
            api_base,
        })
    }

    /// Apply the configured auth to a request builder.
    fn authed(&self, rb: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
        match &self.auth {
            BbAuth::Bearer(t) => rb.bearer_auth(t),
            BbAuth::Basic { username, password } => rb.basic_auth(username, Some(password)),
        }
    }

    /// Fetch every pull request, following `next` cursors until exhausted.
    ///
    /// State filter is `OPEN,MERGED,DECLINED,SUPERSEDED` — i.e. everything.
    /// Bitbucket's default is open-only, which would drop merged history.
    ///
    /// # Errors
    ///
    /// Returns [`CollectError::Http`] on transport or non-success HTTP
    /// responses, and [`CollectError::Json`] on payload parse failures.
    pub async fn fetch_pull_requests(&self) -> Result<Vec<PullRequest>> {
        let initial = format!(
            "{}/repositories/{}/{}/pullrequests\
             ?state=OPEN&state=MERGED&state=DECLINED&state=SUPERSEDED\
             &pagelen={PAGE_SIZE}",
            self.api_base, self.workspace, self.repo_slug
        );

        let mut out: Vec<PullRequest> = Vec::new();
        let mut next_url: Option<String> = Some(initial);
        while let Some(url) = next_url.take() {
            debug!(url = %url, "GET (bitbucket)");
            let resp = self.retry_request(&url).await?;

            if let Some(rem) = resp
                .headers()
                .get("x-ratelimit-remaining")
                .and_then(|v| v.to_str().ok())
                .and_then(|s| s.parse::<u32>().ok())
            {
                if rem < 5 {
                    warn!(remaining = rem, "Bitbucket rate limit nearly exhausted");
                }
            }

            let resp = resp.error_for_status()?;
            let page: BbPaged<BbPullRequest> = resp.json().await?;
            let repository = format!("{}/{}", self.workspace, self.repo_slug);
            for pr in page.values {
                let pr_number = pr.id;
                let mut mapped = map_pr(pr, &repository);
                match self.fetch_pr_commits(pr_number).await {
                    Ok(shas) => {
                        mapped.commit_shas = encode_commit_shas(&shas)?;
                    }
                    Err(e) => {
                        // Graceful degradation (mirrors the per-repo partial-success
                        // precedent in the GitHub client, issue #87): one PR's
                        // commits endpoint hiccuping must not drop the whole PR,
                        // nor abort the rest of the batch. `mapped.commit_shas`
                        // already holds the merge-commit fallback from `map_pr`.
                        warn!(
                            pr_number,
                            error = %e,
                            "Bitbucket PR commit fetch failed; keeping merge-commit fallback"
                        );
                    }
                }
                out.push(mapped);
            }
            next_url = page.next;
        }
        Ok(out)
    }

    /// Fetch every commit SHA attached to a pull request, following `next`
    /// cursors until exhausted.
    ///
    /// Why: the PR list endpoint only carries the merge commit hash (often
    /// abbreviated), which cannot be joined against `commits.sha` by equality
    /// and discards the PR's real commit composition (issue #841). This is
    /// the per-PR enrichment call `fetch_pull_requests` makes for every PR it
    /// returns.
    /// What: `GET /2.0/repositories/{workspace}/{repo_slug}/pullrequests/{pr_number}/commits`,
    /// following `next` cursors exactly like [`Self::fetch_pull_requests`].
    /// Test: `fetch_pr_commits_follows_next_cursor`,
    /// `fetch_pull_requests_persists_full_commit_list`.
    ///
    /// # Errors
    ///
    /// Returns [`CollectError::Http`] on transport or non-success HTTP
    /// responses, and [`CollectError::Json`] on payload parse failures.
    pub async fn fetch_pr_commits(&self, pr_number: u64) -> Result<Vec<String>> {
        let initial = format!(
            "{}/repositories/{}/{}/pullrequests/{pr_number}/commits?pagelen={PAGE_SIZE}",
            self.api_base, self.workspace, self.repo_slug
        );

        let mut out: Vec<String> = Vec::new();
        let mut next_url: Option<String> = Some(initial);
        while let Some(url) = next_url.take() {
            debug!(url = %url, "GET (bitbucket pr commits)");
            let resp = self.retry_request(&url).await?.error_for_status()?;
            let page: BbPaged<BbCommit> = resp.json().await?;
            out.extend(page.values.into_iter().map(|c| c.hash));
            next_url = page.next;
        }
        Ok(out)
    }

    /// Persist a batch of [`PullRequest`] rows into the database.
    ///
    /// Why: `ON CONFLICT … DO UPDATE` keeps existing `id` so FK-linked `pr_reviewers`
    /// survive re-collection; `INSERT OR REPLACE` cascade-deleted them (#752).
    /// A bare `DO UPDATE` still overwrote `state`/`merged_at`/`commit_shas`
    /// unconditionally on every conflict, so a background job re-ingesting an
    /// OLDER snapshot could downgrade an already-`merged` PR back to `open`
    /// (issue #821). The `WHERE excluded.fetched_at > pull_requests.fetched_at`
    /// guard rejects such stale writes: the conflicting row is left untouched
    /// and the statement still succeeds with no error.
    /// What: `provider='bitbucket'`; `repository="<workspace>/<repo_slug>"`;
    /// deduplicates on `(provider,repository,pr_number)` (0012/#88); existing rows
    /// update `title`/`author`/`state`/`merged_at`/`commit_shas`/`fetched_at` only
    /// when the incoming `fetched_at` is strictly newer. Test: see
    /// reviewer_store integration tests for FK-preservation coverage;
    /// `store_pull_requests_stale_write_guard_rejects_older_fetched_at` and
    /// `store_pull_requests_applies_genuinely_newer_fetched_at` (in `tests.rs`)
    /// cover the guard itself.
    /// # Errors
    ///
    /// Propagates [`crate::core::TgaError::DbError`] on SQL failures.
    pub fn store_pull_requests(
        &self,
        db: &Database,
        prs: &[PullRequest],
    ) -> crate::core::Result<usize> {
        let conn = db.connection();
        let mut count = 0usize;
        for pr in prs {
            conn.execute(
                "INSERT INTO pull_requests \
                 (provider,repository,pr_number,title,author,state,created_at,merged_at,commit_shas,fetched_at) \
                 VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10) \
                 ON CONFLICT(provider,repository,pr_number) DO UPDATE SET \
                   title=excluded.title,author=excluded.author,state=excluded.state,\
                   merged_at=excluded.merged_at,commit_shas=excluded.commit_shas,\
                   fetched_at=excluded.fetched_at \
                 WHERE excluded.fetched_at > pull_requests.fetched_at",
                params![
                    "bitbucket",
                    pr.repository,
                    pr.pr_number as i64,
                    pr.title,
                    pr.author,
                    pr.state.as_str(),
                    pr.created_at.to_rfc3339(),
                    pr.merged_at.map(|t| t.to_rfc3339()),
                    pr.commit_shas,
                    pr.fetched_at,
                ],
            )?;
            count += 1;
        }
        Ok(count)
    }

    /// GET with exponential backoff on transient failures (429, 5xx).
    async fn retry_request(&self, url: &str) -> Result<reqwest::Response> {
        let mut last_err: Option<reqwest::Error> = None;
        for attempt in 0..=MAX_RETRIES {
            debug!(url = %url, attempt, "GET bitbucket (with retry)");
            match self.authed(self.client.get(url)).send().await {
                Ok(resp) => {
                    let status = resp.status();
                    let transient =
                        status.as_u16() == 429 || (500..=599).contains(&status.as_u16());
                    if !transient || attempt == MAX_RETRIES {
                        return Ok(resp);
                    }
                    let delay = RETRY_BASE_MS * (1u64 << attempt);
                    warn!(
                        status = %status,
                        attempt,
                        delay_ms = delay,
                        "Bitbucket returned transient status; retrying"
                    );
                    tokio::time::sleep(Duration::from_millis(delay)).await;
                }
                Err(e) => {
                    if attempt == MAX_RETRIES {
                        return Err(CollectError::Http(e));
                    }
                    let delay = RETRY_BASE_MS * (1u64 << attempt);
                    warn!(error = %e, attempt, delay_ms = delay,
                          "Bitbucket transport error; retrying");
                    last_err = Some(e);
                    tokio::time::sleep(Duration::from_millis(delay)).await;
                }
            }
        }
        Err(CollectError::Http(
            last_err.expect("retry loop preserved error"),
        ))
    }
}

/// Map a Bitbucket PR shape into the project's internal [`PullRequest`].
///
/// `repository` is provided by the caller (formatted `"workspace/repo_slug"`)
/// so that the row carries enough identity to participate in the
/// `(provider, repository, pr_number)` unique constraint added in migration
/// v12 (#88).
///
/// State mapping: `MERGED` → [`PrState::Merged`]; `OPEN` → [`PrState::Open`];
/// `DECLINED` and `SUPERSEDED` collapse to [`PrState::Closed`] because the
/// shared schema has no richer variants. The collapse is lossy but documented
/// in the configuration reference.
///
/// `commit_shas` here is seeded from the list payload's (often abbreviated)
/// `merge_commit` hash only as a fallback; [`BitbucketClient::fetch_pull_requests`]
/// overwrites it with the full per-PR commit list from
/// [`BitbucketClient::fetch_pr_commits`] whenever that call succeeds (#841).
fn map_pr(pr: BbPullRequest, repository: &str) -> PullRequest {
    let state = match pr.state.as_str() {
        "MERGED" => PrState::Merged,
        "OPEN" => PrState::Open,
        _ => PrState::Closed,
    };

    let created_at = parse_ts(&pr.created_on).unwrap_or_else(Utc::now);
    let merged_at = if matches!(state, PrState::Merged) {
        pr.updated_on.as_deref().and_then(parse_ts)
    } else {
        None
    };

    let commit_shas = match pr.merge_commit.as_ref().map(|c| c.hash.clone()) {
        Some(h) if !h.is_empty() => serde_json::to_string(&vec![h]).unwrap_or_else(|_| "[]".into()),
        _ => "[]".into(),
    };

    let author = pr
        .author
        .as_ref()
        .map(|a| a.best_name())
        .unwrap_or_default();

    PullRequest {
        id: 0,
        pr_number: pr.id,
        repository: repository.to_string(),
        title: pr.title,
        author,
        state,
        created_at,
        merged_at,
        commit_shas,
        fetched_at: Utc::now().to_rfc3339(),
        // #5734: Bitbucket's list payload does carry `source.branch.name` and
        // `description`, but neither is deserialized yet. `None` states that
        // this provider makes no claim, rather than asserting an empty branch.
        head_ref: None,
        body_ticket_id: None,
    }
}

/// Serialize a full commit SHA list into the JSON-array string stored in
/// `pull_requests.commit_shas`.
///
/// Why: shares the on-disk convention with the GitHub provider's
/// `commit_shas_for_pull` (a JSON array of SHA strings) so downstream
/// consumers — e.g. the `commits.sha` join — don't need per-provider
/// parsing.
/// What: `serde_json::to_string(shas)`. `Vec<String>` serialization cannot
/// fail in practice, but the `Result` return keeps the call site symmetric
/// with the rest of this module's fallible JSON handling.
/// Test: `fetch_pull_requests_persists_full_commit_list`.
fn encode_commit_shas(shas: &[String]) -> Result<String> {
    Ok(serde_json::to_string(shas)?)
}

/// Parse a Bitbucket ISO8601 timestamp into UTC.
///
/// Bitbucket sometimes returns offsets like `+00:00` and sometimes `Z`;
/// chrono's RFC3339 parser handles both. On parse failure the caller falls
/// back to `Utc::now()` so a single malformed row doesn't poison the batch.
fn parse_ts(s: &str) -> Option<DateTime<Utc>> {
    DateTime::parse_from_rfc3339(s)
        .ok()
        .map(|dt| Utc.from_utc_datetime(&dt.naive_utc()))
}

#[async_trait]
impl PrProvider for BitbucketClient {
    fn name(&self) -> &str {
        "bitbucket"
    }

    async fn fetch_pull_requests(&self) -> Result<Vec<PullRequest>> {
        BitbucketClient::fetch_pull_requests(self).await
    }

    fn store_pull_requests(
        &self,
        db: &Database,
        prs: &[PullRequest],
    ) -> crate::core::Result<usize> {
        BitbucketClient::store_pull_requests(self, db, prs)
    }
}

#[cfg(test)]
#[path = "tests.rs"]
mod tests;