lean-ctx 3.8.8

//! Edge client to the private commercial control-plane (`lean-ctx-cloud`).
//!
//! This is the *only* place the open community backend learns an account's paid
//! plan. It calls the private billing service's `/api/billing/entitlements`
//! endpoint with the shared internal key. If the billing service is not
//! configured or unreachable, every account resolves to
//! [`Plan::Free`](crate::core::billing::Plan) — so the open backend runs fully
//! standalone and **no local capability is ever gated** (Local-Free Invariant).

use axum::Json;
use axum::extract::{Path, Query, State};
use axum::http::{HeaderMap, StatusCode, header};
use axum::response::{IntoResponse, Response};
use serde::Deserialize;
use serde_json::{Value, json};
use uuid::Uuid;

use crate::core::billing::Plan;

use super::auth::{AppState, auth_user};
use super::config::Config;

/// Resolve a user's effective plan via the private billing service. Any failure
/// (unconfigured, network error, bad response) degrades gracefully to
/// [`Plan::Free`] — the safe default that grants no commercial entitlements.
pub(super) async fn resolve_plan(cfg: &Config, user_id: Uuid) -> Plan {
    resolve_entitlements_raw(cfg, user_id)
        .await
        .and_then(|v| v.get("plan").and_then(Value::as_str).map(Plan::parse))
        .unwrap_or(Plan::Free)
}

/// The raw entitlements payload from the private billing service (plan,
/// entitlements, org membership — GL #468). `None` on any failure, so callers
/// degrade exactly like [`resolve_plan`].
async fn resolve_entitlements_raw(cfg: &Config, user_id: Uuid) -> Option<Value> {
    let (Some(base), Some(key)) = (
        cfg.billing_base_url.clone(),
        cfg.billing_internal_key.clone(),
    ) else {
        return None;
    };

    let url = format!("{base}/api/billing/entitlements/{user_id}");
    let body = tokio::task::spawn_blocking(move || {
        ureq::get(&url)
            .header("X-Internal-Key", &key)
            .call()
            .ok()?
            .into_body()
            .read_to_string()
            .ok()
    })
    .await
    .ok()
    .flatten()?;

    serde_json::from_str::<Value>(&body).ok()
}

/// Billing-side account deletion (GL #535): cancels any live subscription
/// immediately and purges the user's billing rows.
///
/// Tri-state by design:
/// - billing not configured → `Ok(None)` (nothing to delete — standalone deploy)
/// - billing reachable + 2xx → `Ok(Some(response))`
/// - anything else → `Err(502)` — the caller MUST abort the account deletion,
///   otherwise a paid subscription could keep charging a deleted account.
pub(super) async fn billing_delete_account(
    cfg: &Config,
    user_id: Uuid,
) -> Result<Option<Value>, (StatusCode, String)> {
    let (Some(base), Some(key)) = (
        cfg.billing_base_url.clone(),
        cfg.billing_internal_key.clone(),
    ) else {
        return Ok(None);
    };

    let url = format!("{base}/api/billing/account/{user_id}");
    let body = tokio::task::spawn_blocking(move || {
        ureq::delete(&url)
            .header("X-Internal-Key", &key)
            .call()
            .map_err(|e| e.to_string())?
            .into_body()
            .read_to_string()
            .map_err(|e| e.to_string())
    })
    .await
    .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("join: {e}")))?
    .map_err(|e| {
        (
            StatusCode::BAD_GATEWAY,
            format!("billing deletion failed — account NOT deleted, please retry: {e}"),
        )
    })?;

    Ok(Some(serde_json::from_str(&body).unwrap_or(Value::Null)))
}

/// Whether this deployment leaves cloud sync **ungated** for everyone: either no
/// commercial plane is wired (`billing_base_url` unset), or the operator
/// explicitly opted out (`LEANCTX_CLOUD_SYNC_OPEN=1`). leanctx.com has neither,
/// so sync there is gated to the `cloud_sync` entitlement.
fn sync_is_open(cfg: &Config) -> bool {
    cfg.sync_open || cfg.billing_base_url.is_none()
}

/// Pure cloud-sync gate policy, factored out so it is unit-testable without a
/// DB/billing round-trip. Sync is allowed when the deployment does not gate it
/// at all, or when the caller's plan grants the `cloud_sync` entitlement
/// (Pro/Team/Enterprise). Free and Supporter are denied on a gated deployment.
pub(super) fn cloud_sync_allowed(cfg: &Config, plan: Plan) -> bool {
    sync_is_open(cfg) || plan.entitlements().cloud_sync
}

/// Authenticate the caller **and** require the `cloud_sync` entitlement before a
/// `/api/sync/*` handler proceeds. Returns the same `(user_id, email)` tuple as
/// [`auth_user`], so a call site is a drop-in swap.
///
/// Gating only applies where a commercial plane is actually wired:
/// - **No billing configured** (`billing_base_url` unset) ⇒ open. The community
///   backend runs standalone and sync stays fully usable — nothing is gated
///   without an explicit paid plane (Local-Free Invariant).
/// - **`LEANCTX_CLOUD_SYNC_OPEN=1`** ⇒ open. Operator opt-out for self-hosters
///   who run billing for other reasons but want sync free for everyone.
/// - **Otherwise** ⇒ the account must resolve to a plan whose entitlements grant
///   `cloud_sync` (Pro/Team/Enterprise). Free/Supporter get `402 Payment Required`.
pub(super) async fn require_cloud_sync(
    state: &AppState,
    headers: &HeaderMap,
) -> Result<(Uuid, String), (StatusCode, String)> {
    let (user_id, email) = auth_user(state, headers).await?;

    // Resolve the paid plan only where sync is actually gated; open deployments
    // short-circuit without a billing round-trip. `Plan::Free` is a safe stand-in
    // there — `cloud_sync_allowed` returns `true` via its open checks regardless.
    let plan = if sync_is_open(&state.cfg) {
        Plan::Free
    } else {
        resolve_plan(&state.cfg, user_id).await
    };

    if cloud_sync_allowed(&state.cfg, plan) {
        return Ok((user_id, email));
    }

    Err((
        StatusCode::PAYMENT_REQUIRED,
        format!(
            "cloud sync requires lean-ctx Pro (current plan: {}). \
             Run `lean-ctx upgrade` to enable hosted cross-device sync.",
            plan.as_str()
        ),
    ))
}

/// The account's hosted-index quota in MB (GL #392). Paid plans use their
/// `hosted_index_mb` entitlement (Pro: 1000). Open deployments — no billing
/// plane wired, or sync explicitly opened — get a 1000 MB default so the
/// feature works standalone without ever paying (Local-Free Invariant: the
/// hosted bucket is additive, the local index is never gated).
pub(super) async fn hosted_index_quota_mb(state: &AppState, user_id: Uuid) -> u32 {
    if !sync_is_open(&state.cfg) {
        let quota = resolve_plan(&state.cfg, user_id)
            .await
            .entitlements()
            .hosted_index_mb;
        if quota > 0 {
            return quota;
        }
    }
    1_000
}

/// `GET /api/account/entitlements` — the logged-in user's plan, the additive
/// Team/Cloud entitlements it grants, and the org membership (GL #468) the
/// plan may be inherited through (`org: null` for solo accounts).
pub(super) async fn get_account_entitlements(
    State(state): State<AppState>,
    headers: HeaderMap,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let raw = resolve_entitlements_raw(&state.cfg, user_id).await;
    let plan = raw
        .as_ref()
        .and_then(|v| v.get("plan").and_then(Value::as_str).map(Plan::parse))
        .unwrap_or(Plan::Free);
    let org = raw
        .as_ref()
        .and_then(|v| v.get("org").cloned())
        .unwrap_or(Value::Null);
    // Subscription lifecycle (GL #535): lets the dashboard show a scheduled
    // cancellation ("ends on July 10 — resume anytime") instead of silence.
    let subscription = raw
        .as_ref()
        .and_then(|v| v.get("subscription").cloned())
        .unwrap_or(Value::Null);
    Ok(Json(json!({
        "plan": plan.as_str(),
        "entitlements": plan.entitlements(),
        "org": org,
        "subscription": subscription,
    })))
}

/// Authenticated server-to-server POST to the private billing service. The shared
/// internal key never leaves the backend. Returns the parsed JSON body, or a
/// `503` when billing is not enabled / `502` when the upstream is unreachable.
async fn billing_post(
    cfg: &Config,
    path: &str,
    payload: Value,
) -> Result<Value, (StatusCode, String)> {
    let (Some(base), Some(key)) = (
        cfg.billing_base_url.clone(),
        cfg.billing_internal_key.clone(),
    ) else {
        return Err((
            StatusCode::SERVICE_UNAVAILABLE,
            "billing is not enabled on this deployment".to_string(),
        ));
    };

    let url = format!("{base}{path}");
    let bytes = serde_json::to_vec(&payload)
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("encode: {e}")))?;

    let text = tokio::task::spawn_blocking(move || {
        ureq::post(&url)
            .header("X-Internal-Key", &key)
            .header("Content-Type", "application/json")
            .send(&bytes)
            .map_err(|e| e.to_string())?
            .into_body()
            .read_to_string()
            .map_err(|e| e.to_string())
    })
    .await
    .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("join: {e}")))?
    .map_err(|e| (StatusCode::BAD_GATEWAY, format!("billing upstream: {e}")))?;

    serde_json::from_str::<Value>(&text).map_err(|e| {
        (
            StatusCode::BAD_GATEWAY,
            format!("billing returned non-JSON: {e}"),
        )
    })
}

/// Request body for `POST /api/account/checkout`. `interval` defaults to monthly
/// on the billing side when omitted.
#[derive(Deserialize)]
pub(super) struct CheckoutBody {
    plan: String,
    #[serde(default)]
    interval: Option<String>,
}

/// `POST /api/account/checkout` — start a Stripe Checkout session for the
/// logged-in user and return the hosted `url` to redirect to.
pub(super) async fn post_account_checkout(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(body): Json<CheckoutBody>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, email) = auth_user(&state, &headers).await?;
    let payload = json!({
        "user_id": user_id,
        "email": email,
        "plan": body.plan,
        "interval": body.interval,
    });
    Ok(Json(
        billing_post(&state.cfg, "/api/billing/checkout", payload).await?,
    ))
}

/// `POST /api/account/portal` — open the Stripe billing portal for the logged-in
/// user (manage payment method, invoices, cancel) and return the redirect `url`.
pub(super) async fn post_account_portal(
    State(state): State<AppState>,
    headers: HeaderMap,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let payload = json!({ "user_id": user_id });
    Ok(Json(
        billing_post(&state.cfg, "/api/billing/portal", payload).await?,
    ))
}

// ── Hosted Team server dashboard ──────────────────────────────────────────────
//
// Thin, status-preserving proxies to the private plane's team control endpoints.
// The shared internal key never reaches the browser; the caller is identified by
// their session, so the dashboard can only ever act on its own team instance.

/// Forward a team control call to the private plane, preserving the upstream
/// status so the dashboard can surface 404 (no instance yet) / 400 (seat limit)
/// distinctly. Errors only for unset billing (503) or an unreachable plane (502).
async fn billing_forward(
    cfg: &Config,
    method: &'static str,
    path: String,
    payload: Option<Value>,
) -> Result<(StatusCode, Value), (StatusCode, String)> {
    let (Some(base), Some(key)) = (
        cfg.billing_base_url.clone(),
        cfg.billing_internal_key.clone(),
    ) else {
        return Err((
            StatusCode::SERVICE_UNAVAILABLE,
            "billing is not enabled on this deployment".to_string(),
        ));
    };

    let url = format!("{base}{path}");
    let (code, text) = tokio::task::spawn_blocking(move || -> Result<(u16, String), String> {
        // Read non-2xx as a normal response so the upstream status is preserved.
        let agent: ureq::Agent = ureq::Agent::config_builder()
            .http_status_as_error(false)
            .build()
            .into();
        let resp = match method {
            "GET" => agent.get(&url).header("X-Internal-Key", &key).call(),
            "DELETE" => agent.delete(&url).header("X-Internal-Key", &key).call(),
            // Body methods (POST default, PATCH for partial updates, PUT for
            // full settings replacement). All carry the caller's JSON unchanged.
            _ => {
                let bytes = serde_json::to_vec(&payload.unwrap_or_else(|| json!({})))
                    .map_err(|e| e.to_string())?;
                let builder = match method {
                    "PATCH" => agent.patch(&url),
                    "PUT" => agent.put(&url),
                    _ => agent.post(&url),
                };
                builder
                    .header("X-Internal-Key", &key)
                    .header("Content-Type", "application/json")
                    .send(&bytes)
            }
        }
        .map_err(|e| e.to_string())?;
        let code = resp.status().as_u16();
        let body = resp
            .into_body()
            .read_to_string()
            .map_err(|e| e.to_string())?;
        Ok((code, body))
    })
    .await
    .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("join: {e}")))?
    .map_err(|e| (StatusCode::BAD_GATEWAY, format!("billing upstream: {e}")))?;

    let json = serde_json::from_str::<Value>(&text).unwrap_or(Value::Null);
    let status = StatusCode::from_u16(code).unwrap_or(StatusCode::BAD_GATEWAY);
    Ok((status, json))
}

/// Turn a forwarded `(status, body)` into a handler result, propagating the
/// upstream error message on non-2xx so the dashboard can display it.
fn finish(status: StatusCode, json: Value) -> Result<Json<Value>, (StatusCode, String)> {
    if status.is_success() {
        return Ok(Json(json));
    }
    let msg = json
        .get("error")
        .and_then(Value::as_str)
        .or_else(|| json.get("message").and_then(Value::as_str))
        .unwrap_or("team request failed")
        .to_string();
    Err((status, msg))
}

/// Request body for issuing a team member token.
#[derive(Deserialize)]
pub(super) struct MemberBody {
    #[serde(default)]
    role: Option<String>,
    #[serde(default)]
    label: Option<String>,
}

/// `GET /api/account/team` — the logged-in owner's hosted team server status and
/// member roster (no secrets). `provisioned:false` until a Team plan deploys one.
pub(super) async fn get_account_team(
    State(state): State<AppState>,
    headers: HeaderMap,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "GET",
        format!("/api/billing/team/{user_id}"),
        None,
    )
    .await?;
    finish(status, json)
}

/// `GET /api/account/team/savings` — the logged-in owner's aggregated team
/// savings roll-up (net tokens + USD saved, per member and per model). Returns
/// `savings_available:false` until the hosted server has received at least one
/// signed batch, or `provisioned:false` when no team server exists yet.
pub(super) async fn get_account_team_savings(
    State(state): State<AppState>,
    headers: HeaderMap,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "GET",
        format!("/api/billing/team/{user_id}/savings"),
        None,
    )
    .await?;
    finish(status, json)
}

/// `GET /api/account/team/savings/member/{signer}` — per-member drilldown
/// (GL #389): the signer's own 90-day cumulative series plus model/tool
/// breakdowns. 404 when the signer never reported a batch. The signer id is
/// the truncated public key from `summary.by_member[].signer` (URL-safe by
/// construction; anything else is rejected upstream).
pub(super) async fn get_account_team_savings_member(
    State(state): State<AppState>,
    headers: HeaderMap,
    axum::extract::Path(signer): axum::extract::Path<String>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    // Tight allowlist before the id is embedded in an upstream URL.
    if signer.is_empty()
        || signer.len() > 64
        || !signer
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-')
    {
        return Err((StatusCode::BAD_REQUEST, "invalid signer id".into()));
    }
    let (status, json) = billing_forward(
        &state.cfg,
        "GET",
        format!("/api/billing/team/{user_id}/savings/member/{signer}"),
        None,
    )
    .await?;
    finish(status, json)
}

/// Internal GET against the billing plane for the digest job (GL #386) —
/// no user session involved, the job acts on the server's own behalf.
/// `None` when billing is unconfigured or unreachable.
pub(super) async fn forward_for_digest(cfg: &Config, path: String) -> Option<(u16, Value)> {
    match billing_forward(cfg, "GET", path, None).await {
        Ok((status, json)) => Some((status.as_u16(), json)),
        Err(_) => None,
    }
}

/// Request body for team settings (GL #388).
#[derive(Deserialize)]
pub(super) struct TeamSettingsBody {
    #[serde(default, rename = "roiWebhookUrl")]
    roi_webhook_url: Option<String>,
}

/// `PUT /api/account/team/settings` — owner-tunable team-server settings
/// (currently the weekly ROI webhook URL, GL #388). Validation and the
/// config re-render happen in the control plane; this edge only forwards.
pub(super) async fn put_account_team_settings(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(body): Json<TeamSettingsBody>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "PUT",
        format!("/api/billing/team/{user_id}/settings"),
        Some(json!({ "roiWebhookUrl": body.roi_webhook_url })),
    )
    .await?;
    finish(status, json)
}

/// `POST /api/account/team/owner-token` — (re)issue the owner token, returned
/// exactly once. Rotates any prior owner credential and redeploys the server.
pub(super) async fn post_account_team_owner_token(
    State(state): State<AppState>,
    headers: HeaderMap,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "POST",
        format!("/api/billing/team/{user_id}/owner-token"),
        Some(json!({})),
    )
    .await?;
    finish(status, json)
}

/// `POST /api/account/team/members` — issue a seat-limited member token (returned
/// once). 400 from the plane when the plan's seat limit is reached.
pub(super) async fn post_account_team_member(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(body): Json<MemberBody>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "POST",
        format!("/api/billing/team/{user_id}/tokens"),
        Some(json!({ "role": body.role, "label": body.label })),
    )
    .await?;
    finish(status, json)
}

// ── Invite links (GL #385) ────────────────────────────────────────────────────

/// Request body for `POST /api/account/team/invites`.
#[derive(Deserialize)]
pub(super) struct InviteBody {
    #[serde(default)]
    label: Option<String>,
    #[serde(default)]
    role: Option<String>,
}

/// `POST /api/account/team/invites` — mint a one-time invite link for the
/// logged-in owner's team. The code is returned exactly once; the dashboard
/// turns it into `https://leanctx.com/join/?code=…`.
pub(super) async fn post_account_team_invite(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(body): Json<InviteBody>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "POST",
        format!("/api/billing/team/{user_id}/invites"),
        Some(json!({ "label": body.label, "role": body.role })),
    )
    .await?;
    finish(status, json)
}

/// `GET /api/account/team/invites` — the owner's invite audit list
/// (pending / used / revoked / expired; never the codes themselves).
pub(super) async fn get_account_team_invites(
    State(state): State<AppState>,
    headers: HeaderMap,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "GET",
        format!("/api/billing/team/{user_id}/invites"),
        None,
    )
    .await?;
    finish(status, json)
}

/// `DELETE /api/account/team/invites/{invite_id}` — revoke a pending invite.
pub(super) async fn delete_account_team_invite(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(invite_id): Path<Uuid>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "DELETE",
        format!("/api/billing/team/{user_id}/invites/{invite_id}"),
        None,
    )
    .await?;
    if status == StatusCode::NO_CONTENT {
        return Ok(Json(json!({ "revoked": true })));
    }
    finish(status, json)
}

/// Forward an invite redemption to the control plane on behalf of the (login-
/// less) teammate. Used by the public join endpoint (`team_join.rs`).
pub(super) async fn forward_invite_redeem(
    cfg: &Config,
    code: &str,
) -> Result<(StatusCode, Value), (StatusCode, String)> {
    billing_forward(
        cfg,
        "POST",
        "/api/billing/invites/redeem".to_string(),
        Some(json!({ "code": code })),
    )
    .await
}

// ── Org SSO settings (GL #482) ────────────────────────────────────────────────

/// Request body for `PUT /api/account/org/sso`.
#[derive(Deserialize)]
pub(super) struct OrgSsoBody {
    email_domain: String,
    issuer: String,
    client_id: String,
    #[serde(default)]
    client_secret: Option<String>,
}

/// `GET /api/account/org/sso` — the caller's org SSO config (never the secret).
pub(super) async fn get_account_org_sso(
    State(state): State<AppState>,
    headers: HeaderMap,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "GET",
        format!("/api/billing/org/{user_id}/sso"),
        None,
    )
    .await?;
    finish(status, json)
}

/// `PUT /api/account/org/sso` — create/update the org's IdP configuration.
pub(super) async fn put_account_org_sso(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(body): Json<OrgSsoBody>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "PUT",
        format!("/api/billing/org/{user_id}/sso"),
        Some(json!({
            "email_domain": body.email_domain,
            "issuer": body.issuer,
            "client_id": body.client_id,
            "client_secret": body.client_secret,
        })),
    )
    .await?;
    finish(status, json)
}

/// `POST /api/account/org/sso/verify` — run the DNS-TXT domain check.
pub(super) async fn post_account_org_sso_verify(
    State(state): State<AppState>,
    headers: HeaderMap,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "POST",
        format!("/api/billing/org/{user_id}/sso/verify"),
        Some(json!({})),
    )
    .await?;
    finish(status, json)
}

/// Request body for `PUT /api/account/org/sso/required`.
#[derive(Deserialize)]
pub(super) struct OrgSsoRequiredBody {
    sso_required: bool,
}

/// `PUT /api/account/org/sso/required` — toggle SSO enforcement.
pub(super) async fn put_account_org_sso_required(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(body): Json<OrgSsoRequiredBody>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "PUT",
        format!("/api/billing/org/{user_id}/sso/required"),
        Some(json!({ "sso_required": body.sso_required })),
    )
    .await?;
    finish(status, json)
}

/// `DELETE /api/account/org/sso` — remove the IdP configuration.
pub(super) async fn delete_account_org_sso(
    State(state): State<AppState>,
    headers: HeaderMap,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "DELETE",
        format!("/api/billing/org/{user_id}/sso"),
        None,
    )
    .await?;
    if status == StatusCode::NO_CONTENT {
        return Ok(Json(json!({ "removed": true })));
    }
    finish(status, json)
}

/// Read-side query for the org audit log (GL #484). Mirrors the control-plane
/// contract; all three are optional.
#[derive(Deserialize)]
pub(super) struct AuditQuery {
    #[serde(default)]
    before: Option<i64>,
    #[serde(default)]
    limit: Option<i64>,
    #[serde(default)]
    event: Option<String>,
}

/// Build the sanitized upstream query string. `before`/`limit` are numeric and
/// safe; `event` is allowlisted to our snake_case event ids so nothing
/// untrusted is ever spliced into the upstream URL.
fn build_audit_query(q: &AuditQuery) -> String {
    let mut parts: Vec<String> = Vec::new();
    if let Some(b) = q.before
        && b > 0
    {
        parts.push(format!("before={b}"));
    }
    if let Some(l) = q.limit {
        parts.push(format!("limit={}", l.clamp(1, 200)));
    }
    if let Some(ev) = q.event.as_deref() {
        let ev = ev.trim();
        if !ev.is_empty()
            && ev.len() <= 48
            && ev.bytes().all(|b| b.is_ascii_lowercase() || b == b'_')
        {
            parts.push(format!("event={ev}"));
        }
    }
    if parts.is_empty() {
        String::new()
    } else {
        format!("?{}", parts.join("&"))
    }
}

/// `GET /api/account/org/audit` — the owner's paginated governance audit log.
pub(super) async fn get_account_org_audit(
    State(state): State<AppState>,
    headers: HeaderMap,
    Query(q): Query<AuditQuery>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let qs = build_audit_query(&q);
    let (status, json) = billing_forward(
        &state.cfg,
        "GET",
        format!("/api/billing/org/{user_id}/audit{qs}"),
        None,
    )
    .await?;
    finish(status, json)
}

/// `GET /api/account/org/audit/export.csv` — the owner's audit log as a CSV
/// download. The control plane renders the CSV; this edge streams it through
/// with the right headers (the body is not JSON, so it bypasses `finish`).
pub(super) async fn get_account_org_audit_export(
    State(state): State<AppState>,
    headers: HeaderMap,
) -> Result<Response, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, body) = billing_forward_text(
        &state.cfg,
        format!("/api/billing/org/{user_id}/audit/export.csv"),
    )
    .await?;
    if !status.is_success() {
        return Err((status, "audit export failed".to_string()));
    }
    Ok((
        [
            (header::CONTENT_TYPE, "text/csv; charset=utf-8"),
            (
                header::CONTENT_DISPOSITION,
                "attachment; filename=\"leanctx-audit-log.csv\"",
            ),
        ],
        body,
    )
        .into_response())
}

// ── ctxpkg registry publisher self-service (GL #406) ─────────────────────────
//
// Namespace + publish-token management for the logged-in account. Thin
// status-preserving proxies to the private plane; publish/download themselves
// never touch this edge — they go straight to the registry via ctxpkg.com.

/// Request body for `PUT /api/account/registry/namespace`.
#[derive(Deserialize)]
pub(super) struct RegistryNamespaceBody {
    namespace: String,
    /// Claim on behalf of an org (GL #524) — requires owner/admin there.
    #[serde(default)]
    org_id: Option<String>,
}

/// `PUT /api/account/registry/namespace` — claim the account's publisher
/// namespace on the ctxpkg registry (permanent in v1).
pub(super) async fn put_account_registry_namespace(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(body): Json<RegistryNamespaceBody>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "PUT",
        format!("/api/billing/registry/{user_id}/namespace"),
        Some(json!({ "namespace": body.namespace, "org_id": body.org_id })),
    )
    .await?;
    finish(status, json)
}

/// `GET /api/account/registry` — publisher profile: namespace + token list
/// (metadata only; plaintext tokens are shown exactly once at mint time).
pub(super) async fn get_account_registry(
    State(state): State<AppState>,
    headers: HeaderMap,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "GET",
        format!("/api/billing/registry/{user_id}"),
        None,
    )
    .await?;
    finish(status, json)
}

/// Request body for `POST /api/account/registry/tokens`.
#[derive(Deserialize, Default)]
pub(super) struct RegistryTokenBody {
    #[serde(default)]
    label: Option<String>,
    /// `publish` (default) or `read` — read tokens are install-only (GL #524).
    #[serde(default)]
    scope: Option<String>,
}

/// `POST /api/account/registry/tokens` — mint a `ctxp_…` publish token or a
/// `ctxr_…` read-only install token.
pub(super) async fn post_account_registry_token(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(body): Json<RegistryTokenBody>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "POST",
        format!("/api/billing/registry/{user_id}/tokens"),
        Some(json!({ "label": body.label, "scope": body.scope })),
    )
    .await?;
    finish(status, json)
}

/// `DELETE /api/account/registry/tokens/{token_id}` — revoke a publish token.
pub(super) async fn delete_account_registry_token(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(token_id): Path<i64>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "DELETE",
        format!("/api/billing/registry/{user_id}/tokens/{token_id}"),
        None,
    )
    .await?;
    finish(status, json)
}

/// Request body for `PUT /api/account/registry/price`.
#[derive(Deserialize)]
pub(super) struct RegistryPriceBody {
    name: String,
    /// `0` / absent clears the price (the pack becomes free again).
    #[serde(default)]
    price_cents: Option<i32>,
}

/// `PUT /api/account/registry/price` — set or clear a package price on the
/// account's namespace (Paid Packs v0, GL #529).
pub(super) async fn put_account_registry_price(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(body): Json<RegistryPriceBody>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "PUT",
        format!("/api/billing/registry/{user_id}/price"),
        Some(json!({ "name": body.name, "price_cents": body.price_cents })),
    )
    .await?;
    finish(status, json)
}

/// Request body for `POST /api/account/registry/buy`.
#[derive(Deserialize)]
pub(super) struct RegistryBuyBody {
    namespace: String,
    name: String,
}

/// `POST /api/account/registry/buy` — start a Stripe Checkout for a paid
/// pack; returns the hosted checkout URL (GL #529).
pub(super) async fn post_account_registry_buy(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(body): Json<RegistryBuyBody>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "POST",
        format!("/api/billing/registry/{user_id}/buy"),
        Some(json!({
            "namespace": body.namespace,
            "name": body.name,
            "email": email,
        })),
    )
    .await?;
    finish(status, json)
}

/// Request body for `POST /api/account/registry/domains`.
#[derive(Deserialize)]
pub(super) struct RegistryDomainBody {
    domain: String,
}

/// `POST /api/account/registry/domains` — register a domain for Verified
/// Publisher and receive the DNS-TXT challenge (GL #516).
pub(super) async fn post_account_registry_domain(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(body): Json<RegistryDomainBody>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "POST",
        format!("/api/billing/registry/{user_id}/domains"),
        Some(json!({ "domain": body.domain })),
    )
    .await?;
    finish(status, json)
}

/// `POST /api/account/registry/domains/{domain_id}/verify` — trigger the
/// DNS-TXT check; flips the publisher to verified on success.
pub(super) async fn post_account_registry_domain_verify(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(domain_id): Path<i64>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "POST",
        format!("/api/billing/registry/{user_id}/domains/{domain_id}/verify"),
        None,
    )
    .await?;
    finish(status, json)
}

/// `DELETE /api/account/registry/domains/{domain_id}` — remove a domain.
pub(super) async fn delete_account_registry_domain(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(domain_id): Path<i64>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "DELETE",
        format!("/api/billing/registry/{user_id}/domains/{domain_id}"),
        None,
    )
    .await?;
    finish(status, json)
}

/// Like [`billing_forward`] but returns the raw upstream body unparsed — used
/// for the CSV export, whose body is `text/csv`, not JSON.
async fn billing_forward_text(
    cfg: &Config,
    path: String,
) -> Result<(StatusCode, String), (StatusCode, String)> {
    let (Some(base), Some(key)) = (
        cfg.billing_base_url.clone(),
        cfg.billing_internal_key.clone(),
    ) else {
        return Err((
            StatusCode::SERVICE_UNAVAILABLE,
            "billing is not enabled on this deployment".to_string(),
        ));
    };
    let url = format!("{base}{path}");
    let (code, text) = tokio::task::spawn_blocking(move || -> Result<(u16, String), String> {
        let agent: ureq::Agent = ureq::Agent::config_builder()
            .http_status_as_error(false)
            .build()
            .into();
        let resp = agent
            .get(&url)
            .header("X-Internal-Key", &key)
            .call()
            .map_err(|e| e.to_string())?;
        let code = resp.status().as_u16();
        let body = resp
            .into_body()
            .read_to_string()
            .map_err(|e| e.to_string())?;
        Ok((code, body))
    })
    .await
    .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("join: {e}")))?
    .map_err(|e| (StatusCode::BAD_GATEWAY, format!("billing upstream: {e}")))?;
    let status = StatusCode::from_u16(code).unwrap_or(StatusCode::BAD_GATEWAY);
    Ok((status, text))
}

/// `GET /api/supporters` — the public supporters wall (no auth). Proxies the
/// private plane's read model with the shared internal key (which never reaches
/// the browser). Degrades to an empty wall when billing is unconfigured or
/// unreachable, so the website always renders.
pub(super) async fn get_supporters(State(state): State<AppState>) -> Json<Value> {
    let empty = || json!({ "supporters": [], "count": 0 });

    let (Some(base), Some(key)) = (
        state.cfg.billing_base_url.clone(),
        state.cfg.billing_internal_key.clone(),
    ) else {
        return Json(empty());
    };

    let url = format!("{base}/api/billing/supporters");
    let body = tokio::task::spawn_blocking(move || {
        ureq::get(&url)
            .header("X-Internal-Key", &key)
            .call()
            .ok()?
            .into_body()
            .read_to_string()
            .ok()
    })
    .await
    .ok()
    .flatten();

    match body.and_then(|b| serde_json::from_str::<Value>(&b).ok()) {
        Some(v) => Json(v),
        None => Json(empty()),
    }
}

/// Request body for `POST /api/supporters/checkout`: the chosen monthly
/// contribution in USD minor units (cents).
#[derive(Deserialize)]
pub(super) struct SupporterCheckoutBody {
    #[serde(default)]
    amount_cents: i64,
}

/// `POST /api/supporters/checkout` — start a no-account, custom-amount Supporter
/// subscription and return the hosted Stripe `url`. Public: supporting needs no
/// login. The amount is clamped here (defense in depth) and again on the private
/// plane; a 503/502 lets the website fall back to a fixed preset Payment Link.
pub(super) async fn post_supporter_checkout(
    State(state): State<AppState>,
    Json(body): Json<SupporterCheckoutBody>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let amount = body.amount_cents.clamp(100, 100_000);
    Ok(Json(
        billing_post(
            &state.cfg,
            "/api/billing/supporters/checkout",
            json!({ "amount_cents": amount }),
        )
        .await?,
    ))
}

/// `DELETE /api/account/team/members/{token_id}` — revoke a member token and
/// redeploy. The owner token cannot be revoked (the plane rejects it).
pub(super) async fn delete_account_team_member(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(token_id): Path<String>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "DELETE",
        format!("/api/billing/team/{user_id}/tokens/{token_id}"),
        None,
    )
    .await?;
    if status.is_success() {
        return Ok(Json(json!({ "revoked": true })));
    }
    finish(status, json)
}

// ── Team seats, hosted-index storage & managed connectors ─────────────────────
//
// Same thin-proxy pattern as the team roster above: authenticate the owner by
// their session, forward to the private plane with the internal key, and preserve
// the upstream status. Request bodies are passed through unchanged (the plane owns
// validation), so the edge never duplicates the seat/connector schema.

/// `POST /api/account/team/seats` — change the team's seat count (written straight
/// to the Stripe subscription, prorated). Body `{ "seats": N }`; returns the
/// refreshed team payload so the dashboard re-renders in one round-trip.
pub(super) async fn post_account_team_seats(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(body): Json<Value>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "POST",
        format!("/api/billing/team/{user_id}/seats"),
        Some(body),
    )
    .await?;
    finish(status, json)
}

/// `GET /api/account/team/storage` — hosted retrieval-index footprint + metering.
/// `available:false` until a team server is provisioned and reports storage.
pub(super) async fn get_account_team_storage(
    State(state): State<AppState>,
    headers: HeaderMap,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "GET",
        format!("/api/billing/team/{user_id}/storage"),
        None,
    )
    .await?;
    finish(status, json)
}

/// `GET /api/account/team/connectors` — the secret-free managed-connector roster,
/// each merged with its latest live sync status.
pub(super) async fn get_account_team_connectors(
    State(state): State<AppState>,
    headers: HeaderMap,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "GET",
        format!("/api/billing/team/{user_id}/connectors"),
        None,
    )
    .await?;
    finish(status, json)
}

/// `POST /api/account/team/connectors` — create a managed connector. The plaintext
/// provider secret is forwarded once to the plane (encrypted at rest there) and is
/// never stored or echoed by the edge. 400 from the plane on validation / limit.
pub(super) async fn post_account_team_connector(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(body): Json<Value>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "POST",
        format!("/api/billing/team/{user_id}/connectors"),
        Some(body),
    )
    .await?;
    finish(status, json)
}

/// `PATCH /api/account/team/connectors/{connector_id}` — pause/resume a connector.
/// Body `{ "enabled": bool }`. The plane returns 204 No Content, so surface a
/// small JSON ack the dashboard can treat as success.
pub(super) async fn patch_account_team_connector(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(connector_id): Path<String>,
    Json(body): Json<Value>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "PATCH",
        format!("/api/billing/team/{user_id}/connectors/{connector_id}"),
        Some(body),
    )
    .await?;
    if status.is_success() {
        return Ok(Json(json!({ "updated": true })));
    }
    finish(status, json)
}

/// `DELETE /api/account/team/connectors/{connector_id}` — remove a connector and
/// redeploy. The plane returns 204 No Content; surface a JSON ack.
pub(super) async fn delete_account_team_connector(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(connector_id): Path<String>,
) -> Result<Json<Value>, (StatusCode, String)> {
    let (user_id, _email) = auth_user(&state, &headers).await?;
    let (status, json) = billing_forward(
        &state.cfg,
        "DELETE",
        format!("/api/billing/team/{user_id}/connectors/{connector_id}"),
        None,
    )
    .await?;
    if status.is_success() {
        return Ok(Json(json!({ "deleted": true })));
    }
    finish(status, json)
}

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

    /// A `Config` carrying only the two knobs the sync gate reads. `billing`
    /// toggles whether a commercial plane is wired; `sync_open` is the operator
    /// opt-out (`LEANCTX_CLOUD_SYNC_OPEN`).
    fn cfg(billing: bool, sync_open: bool) -> Config {
        Config {
            bind_host: "127.0.0.1".into(),
            bind_port: 8088,
            public_base_url: String::new(),
            api_base_url: String::new(),
            database_url: String::new(),
            ip_hash_salt: String::new(),
            smtp_host: None,
            smtp_port: None,
            smtp_username: None,
            smtp_password: None,
            smtp_from: None,
            billing_base_url: billing.then(|| "https://billing.example".to_string()),
            billing_internal_key: billing.then(|| "internal-key".to_string()),
            sync_open,
        }
    }

    #[test]
    fn gated_deployment_blocks_free_and_supporter_only() {
        // leanctx.com: billing wired, no operator opt-out → the gate is live.
        let gated = cfg(true, false);
        // Free and Supporter lack `cloud_sync` ⇒ denied (handler returns 402).
        assert!(!cloud_sync_allowed(&gated, Plan::Free));
        assert!(!cloud_sync_allowed(&gated, Plan::Supporter));
        // Pro and every superset grant `cloud_sync` ⇒ allowed.
        assert!(cloud_sync_allowed(&gated, Plan::Pro));
        assert!(cloud_sync_allowed(&gated, Plan::Team));
        assert!(cloud_sync_allowed(&gated, Plan::Enterprise));
    }

    #[test]
    fn no_billing_plane_never_gates_sync() {
        // A self-hosted community backend without billing keeps sync fully usable
        // for every logged-in user (Local-Free Invariant) — even Free.
        let open = cfg(false, false);
        assert!(sync_is_open(&open));
        assert!(cloud_sync_allowed(&open, Plan::Free));
    }

    #[test]
    fn operator_opt_out_opens_sync_even_with_billing() {
        // Self-host with billing wired but LEANCTX_CLOUD_SYNC_OPEN=1 → sync free
        // for everyone, regardless of plan.
        let opt_out = cfg(true, true);
        assert!(sync_is_open(&opt_out));
        assert!(cloud_sync_allowed(&opt_out, Plan::Free));
    }
}