solid-pod-rs-server 0.5.0-alpha.18

//! Provenance `_prov` API + the composition wiring for the LDP write hook
//! (master-plan §2.4, ADR-059 D1/D5/D7 — Phase 5).
//!
//! Three concerns live here:
//!
//! 1. **`GET /{pod}/_prov/{commit_sha}`** — resolve a git-mark commit SHA back
//!    to the resource it wrote + its [`ProvenanceMark`] (and any block-trail
//!    anchor recorded in the sidecar). Public (the same audience as the
//!    `.prov.ttl` sidecar the route complements).
//! 2. **`POST /{pod}/_prov/anchor`** — NIP-98-authenticated (pod owner),
//!    payment-gated *explicit upgrade* of an existing git-mark to a Bitcoin
//!    block-trail anchor (the high-value opt-in). Debits the caller's Web
//!    Ledger by the pod's configured anchor price, anchors the commit SHA on
//!    the pod's trail (via [`GitmarkAnchorer`], or [`MempoolBlockAnchorer`]
//!    for an MRC20 trail already issued), and rewrites the resource's
//!    `.prov.ttl` sidecar to carry the anchor.
//! 3. **Composition helpers for the LDP write hook** — [`resolve_anchor_policy`]
//!    reads a resource's effective ACL and maps a `ProvenanceAnchor` condition
//!    onto an [`AnchorPolicy`]; [`build_anchorer`] constructs the optional
//!    [`ProvenanceLog::anchorer`]; [`epoch_push_and_maybe_anchor`] backs the
//!    `AnchorPolicy::Epoch` batch (one Bitcoin tx notarises N commits, ADR-059
//!    D5) with a small per-pod persisted [`EpochAccumulator`].
//!
//! All of this is native-only (it shells to `git` and talks to the mempool);
//! the pure composition types ([`ProvenanceLog`], [`AnchorPolicy`],
//! [`EpochAccumulator`]) live in `solid-pod-rs::provenance` and compile to wasm.

use std::sync::Arc;

use actix_web::{web, Error as ActixError, HttpRequest, HttpResponse};
use bytes::Bytes;

use solid_pod_rs::provenance::{
    prov_ttl, AnchorPolicy, BlockAnchorer, BlockTrailAnchor, ClosedEpoch, EpochAccumulator,
    ProvenanceError, ProvenanceMark,
};
use solid_pod_rs::storage::Storage;
use solid_pod_rs::wac::{anchor_mode_of, AnchorMode};
use solid_pod_rs::PodError;

use crate::mempool::{GitmarkAnchorer, MempoolBlockAnchorer, MempoolHttpClient};
use crate::trail_store::{load_gitmark_trail, load_trail, DEFAULT_GITMARK_TRAIL};
use crate::{extract_pubkey_with_body, pod_repo_path, require_pod_owner_with_body, AppState};

/// Default epoch close threshold (commit count) when the operator has not
/// configured one. Bounds on-chain cost: one anchor per this many commits
/// (master-plan §5 "ACL writes epoch-only to bound cost"). Read from
/// `JSS_PROV_EPOCH_SIZE` when set.
const DEFAULT_EPOCH_SIZE: usize = 16;

/// Environment override for the epoch close threshold.
pub const EPOCH_SIZE_ENV: &str = "JSS_PROV_EPOCH_SIZE";

/// Pod-storage path of the persisted epoch accumulator (the pending batch of
/// git commit SHAs awaiting a single anchoring tx).
const EPOCH_PATH: &str = "/.well-known/prov/epoch.json";

/// Pod-storage path of the batch last anchored, written only when the epoch
/// reset after that anchor failed: the commits to drop from the pending batch
/// so they are not anchored (and paid for) a second time.
const EPOCH_ANCHORED_PATH: &str = "/.well-known/prov/epoch-anchored.json";

/// The configured epoch close threshold (clamped to ≥ 1).
fn epoch_size() -> usize {
    std::env::var(EPOCH_SIZE_ENV)
        .ok()
        .and_then(|v| v.trim().parse::<usize>().ok())
        .filter(|n| *n >= 1)
        .unwrap_or(DEFAULT_EPOCH_SIZE)
}

// ---------------------------------------------------------------------------
// Composition wiring for the LDP write hook
// ---------------------------------------------------------------------------

/// Read `resource_path`'s **effective** ACL and resolve the
/// [`AnchorPolicy`] + optional ticker override for a write to it.
///
/// - No `ProvenanceAnchor` condition on the resource (the common case) ⇒
///   [`AnchorPolicy::Never`] — git-mark only, no on-chain cost.
/// - A `ProvenanceAnchor` with `anchorMode "always"`/`"inline"` ⇒
///   [`AnchorPolicy::HighValue`] (anchored inline; the hook passes `high_value
///   = true`).
/// - A `ProvenanceAnchor` with the default `"epoch"` mode ⇒
///   [`AnchorPolicy::Epoch`] (batched into the per-pod epoch).
///
/// Fail-open for provenance: any ACL read error ⇒ `Never` (a missing/broken
/// ACL must never *force* expensive anchoring, and the write itself was
/// already authorised upstream).
pub(crate) async fn resolve_anchor_policy(
    state: &AppState,
    resource_path: &str,
) -> (AnchorPolicy, Option<String>) {
    let acl = match crate::find_effective_acl_dyn(&*state.storage, resource_path).await {
        Ok(Some(doc)) => doc,
        _ => return (AnchorPolicy::Never, None),
    };
    let Some(graph) = acl.graph.as_ref() else {
        return (AnchorPolicy::Never, None);
    };
    // Scan every authorisation's conditions for the first ProvenanceAnchor.
    for auth in graph {
        if let Some(conds) = auth.condition.as_ref() {
            if let Some(mode) = anchor_mode_of(conds) {
                let ticker = conds.iter().find_map(|c| match c {
                    solid_pod_rs::wac::Condition::ProvenanceAnchor(b) => b.ticker.clone(),
                    _ => None,
                });
                let policy = match mode {
                    AnchorMode::Inline => AnchorPolicy::HighValue,
                    AnchorMode::Epoch => AnchorPolicy::Epoch,
                };
                return (policy, ticker);
            }
        }
    }
    (AnchorPolicy::Never, None)
}

/// Stands in for the anchorer when the pod's trail could not be read: every
/// anchor reports the storage error, so the write's receipt carries an
/// `anchor_error` instead of silently degrading to git-mark-only.
struct UnavailableAnchorer(String);

#[async_trait::async_trait(?Send)]
impl BlockAnchorer for UnavailableAnchorer {
    async fn anchor(
        &self,
        _ticker: &str,
        _state_hash: &str,
        _network: &str,
    ) -> Result<BlockTrailAnchor, ProvenanceError> {
        Err(ProvenanceError::Anchor(self.0.clone()))
    }

    async fn verify(&self, _anchor: &BlockTrailAnchor) -> Result<bool, ProvenanceError> {
        Err(ProvenanceError::Anchor(self.0.clone()))
    }
}

/// Build the optional expensive-tier anchorer for a pod, or `None` when the
/// pod has no trail to anchor on — in which case
/// [`ProvenanceLog`](solid_pod_rs::provenance::ProvenanceLog) degrades to
/// git-mark-only.
///
/// A trail that exists but cannot be *read* is not "no trail": the result is
/// an anchorer that fails every anchor with the storage error, which the
/// write hook records as `anchor_error` (see [`try_build_anchorer`]).
pub(crate) async fn build_anchorer(
    state: &AppState,
    ticker_override: Option<&str>,
) -> Option<(Arc<dyn BlockAnchorer>, String, String)> {
    match try_build_anchorer(state, ticker_override).await {
        Ok(bundle) => bundle,
        Err(e) => {
            tracing::warn!(
                target: "solid_pod_rs_server::prov",
                "anchorer unavailable (trail unreadable): {e}"
            );
            Some((
                Arc::new(UnavailableAnchorer(e)),
                ticker_override.unwrap_or_default().to_string(),
                String::new(),
            ))
        }
    }
}

/// [`build_anchorer`] with the storage error kept: `Ok(None)` only when the
/// pod has no trail (a definite not-found), `Err` when reading one failed.
///
/// The trail is named by `ticker_override`, else the pay-token ticker, else
/// [`DEFAULT_GITMARK_TRAIL`]. An MRC20 trail already issued under that name
/// keeps [`MempoolBlockAnchorer`] (its marks and `urn:mono:op:anchor` states
/// are never moved); otherwise a git-mark trail of that name, the profile new
/// trails use, gets a [`GitmarkAnchorer`]. Returns the anchorer plus the
/// resolved `(ticker, network)`; the trail's own network is authoritative.
pub(crate) async fn try_build_anchorer(
    state: &AppState,
    ticker_override: Option<&str>,
) -> Result<Option<(Arc<dyn BlockAnchorer>, String, String)>, String> {
    let token = state.pay_config.token.as_ref();
    let ticker = ticker_override
        .filter(|t| !t.is_empty())
        .map(str::to_string)
        .or_else(|| token.map(|t| t.ticker.clone()).filter(|t| !t.is_empty()))
        .unwrap_or_else(|| DEFAULT_GITMARK_TRAIL.to_string());

    let mempool = || match &state.mempool_url {
        Some(url) => MempoolHttpClient::new(url.clone()),
        None => MempoolHttpClient::from_env(),
    };

    // An issued MRC20 trail (it needs the pod's pay-token) keeps its profile.
    if token.is_some() {
        if let Some(trail) = load_trail(&state.storage, &ticker)
            .await
            .map_err(|e| e.to_string())?
        {
            let anchorer = MempoolBlockAnchorer::with_storage(mempool(), state.storage.clone());
            return Ok(Some((Arc::new(anchorer), ticker, trail.network)));
        }
    }

    // Otherwise the git-mark profile, when the trail has been started.
    let Some(trail) = load_gitmark_trail(&state.storage, &ticker)
        .await
        .map_err(|e| e.to_string())?
    else {
        return Ok(None);
    };
    let anchorer = GitmarkAnchorer::with_storage(mempool(), state.storage.clone());
    Ok(Some((Arc::new(anchorer), ticker, trail.network())))
}

/// Load the persisted per-pod pending epoch batch (the commit SHAs awaiting a
/// single anchoring tx). Only a missing file is an empty batch; a read or
/// parse failure is an error, so the batch is never overwritten unseen.
async fn load_epoch_commits(storage: &Arc<dyn Storage>) -> Result<Vec<String>, String> {
    match storage.get(EPOCH_PATH).await {
        Ok((bytes, _)) => {
            serde_json::from_slice(&bytes).map_err(|e| format!("malformed epoch batch: {e}"))
        }
        Err(PodError::NotFound(_)) => Ok(Vec::new()),
        Err(e) => Err(format!("read epoch batch: {e}")),
    }
}

/// Persist the per-pod pending epoch batch.
async fn save_epoch_commits(storage: &Arc<dyn Storage>, commits: &[String]) -> Result<(), String> {
    let body = serde_json::to_vec(commits).map_err(|e| e.to_string())?;
    storage
        .put(EPOCH_PATH, Bytes::from(body), "application/json")
        .await
        .map(|_| ())
        .map_err(|e| e.to_string())
}

/// How many epochs' worth of commits may wait on a failing anchor before
/// further commits are refused rather than appended to `epoch.json`.
const EPOCH_BACKLOG_FACTOR: usize = 4;

/// Feed a git commit SHA into the pod's epoch (the `AnchorPolicy::Epoch` path),
/// and when the epoch fills, anchor the **batch Merkle root** with a single
/// Bitcoin tx (ADR-059 D5) and reset the epoch.
///
/// The pending batch is persisted as a plain SHA list; the Merkle root is
/// computed via [`EpochAccumulator`] (the pure, wasm-safe tree) at close time.
///
/// Returns `Ok(Some(closed))` when this push closed an epoch (the batch was
/// anchored — `closed.root` is the value committed on-chain, `closed.commits`
/// the SHAs it notarises), `Ok(None)` when the commit was merely accumulated.
/// Best-effort: the caller (the LDP hook) swallows the error — a failed epoch
/// anchor must never fail the write.
pub(crate) async fn epoch_push_and_maybe_anchor(
    state: &AppState,
    anchorer: &Arc<dyn BlockAnchorer>,
    ticker: &str,
    network: &str,
    commit_sha: &str,
) -> Result<Option<ClosedEpoch>, String> {
    let size = epoch_size();
    // One read-modify-write of the batch at a time (and one close at a time).
    let _guard = crate::trail_store::lock_path(EPOCH_PATH).await;
    let mut commits = load_epoch_commits(&state.storage).await?;
    // A batch anchored on an earlier push whose reset was lost: drop it now,
    // before anything is counted, so it is not anchored a second time.
    match state.storage.get(EPOCH_ANCHORED_PATH).await {
        Ok((bytes, _)) => {
            let done: Vec<String> = serde_json::from_slice(&bytes)
                .map_err(|e| format!("malformed anchored-epoch marker: {e}"))?;
            commits.retain(|c| !done.contains(c));
            save_epoch_commits(&state.storage, &commits).await?;
            // Dropping the same commits again is a no-op, so a marker that
            // cannot be cleared now is cleared by a later push.
            if let Err(e) = state.storage.delete(EPOCH_ANCHORED_PATH).await {
                tracing::warn!("anchored-epoch marker not cleared: {e}");
            }
        }
        Err(PodError::NotFound(_)) => {}
        Err(e) => return Err(format!("read anchored-epoch marker: {e}")),
    }
    // Idempotency: never double-count the same commit in a batch.
    if !commits.iter().any(|c| c == commit_sha) {
        // A close that keeps failing leaves the batch full, and every push
        // would grow it. Past the backlog cap the commit is not accumulated
        // (its git-mark stands) but the close is still attempted, so the
        // batch drains once an anchor succeeds.
        if commits.len() < size.saturating_mul(EPOCH_BACKLOG_FACTOR) {
            commits.push(commit_sha.to_string());
        } else {
            tracing::warn!(
                backlog = commits.len(),
                "epoch backlog full: commit not accumulated, anchoring keeps failing"
            );
        }
    }

    // Persist the grown batch first: a failed close below must not lose the
    // commit that filled it (the next push retries the close).
    save_epoch_commits(&state.storage, &commits).await?;
    if commits.len() < size {
        return Ok(None);
    }

    // Epoch full → compute the single Merkle root over the batch and anchor it.
    let mut acc = EpochAccumulator::new(size);
    for c in &commits {
        acc.push(c.clone());
    }
    let closed = acc.close().ok_or("epoch close produced nothing")?;
    anchorer
        .anchor(ticker, &closed.root, network)
        .await
        .map_err(|e| e.to_string())?;
    // Reset the on-disk epoch (the batch is now anchored on-chain). If that
    // fails the batch is still on disk, so record what was anchored: the next
    // push drops it rather than paying to anchor it again.
    if let Err(reset) = save_epoch_commits(&state.storage, &[]).await {
        let marker = serde_json::to_vec(&closed.commits).map_err(|e| e.to_string())?;
        if let Err(e) = state
            .storage
            .put(EPOCH_ANCHORED_PATH, Bytes::from(marker), "application/json")
            .await
        {
            tracing::error!(
                root = %closed.root,
                "epoch anchored but neither the reset ({reset}) nor the marker ({e}) was saved; \
                 the next push will anchor this batch again"
            );
            return Err(format!(
                "epoch anchored but not reset: {reset}; marker: {e}"
            ));
        }
        tracing::warn!(root = %closed.root, "epoch anchored; reset failed ({reset}), marker kept");
    }
    Ok(Some(closed))
}

// ---------------------------------------------------------------------------
// GET /{pod}/_prov/{commit_sha}
// ---------------------------------------------------------------------------

/// Resolve a git-mark commit SHA to its resource + [`ProvenanceMark`].
///
/// `git show` the commit for metadata + changed files, pick the content file
/// it wrote (excluding sidecars), then load that resource's `.prov.ttl`
/// sidecar — if present and it matches this commit, return the recorded mark
/// (carrying any block-trail anchor); otherwise reconstruct a git-only mark
/// from the commit metadata.
async fn handle_resolve(
    path: web::Path<(String, String)>,
    state: web::Data<AppState>,
) -> Result<HttpResponse, ActixError> {
    let (pod, commit_sha) = path.into_inner();

    let Some(repo) = pod_repo_path(&state, &pod) else {
        return Ok(prov_err(
            "git provenance not available (no FS backend / invalid pod)",
            501,
        ));
    };
    if !repo.join(".git").is_dir() {
        return Ok(prov_err("pod is not git-backed", 404));
    }

    let resolved = match solid_pod_rs_git::resolve_commit(&repo, &commit_sha).await {
        Ok(r) => r,
        Err(e) => {
            let code = e.status_code();
            return Ok(prov_err(
                &format!("commit not found: {e}"),
                if code == 404 { 404 } else { 400 },
            ));
        }
    };

    // Pick the resource the write targeted: the first non-sidecar file the
    // commit touched. (A git-mark commits exactly one content file + nothing
    // else; ACL/meta/prov writes are never marked, see `git_mark_write`.)
    let resource_rel = resolved.files.iter().find(|f| !is_sidecar(f)).cloned();
    let Some(resource_rel) = resource_rel else {
        return Ok(prov_err("commit touched no content resource", 404));
    };
    let resource = format!("/{pod}/{resource_rel}");

    // Try the persisted sidecar first (it carries any anchor + the exact mark).
    let sidecar = format!("{resource}.prov.ttl");
    let sidecar_ttl = state
        .storage
        .get(&sidecar)
        .await
        .ok()
        .map(|(b, _)| String::from_utf8_lossy(&b).into_owned());

    // Build the JSON response. The mark is authoritative from the sidecar when
    // present; otherwise reconstruct from git metadata (git-only, no anchor).
    let body = serde_json::json!({
        "pod": pod,
        "resource": resource,
        "commit": {
            "sha": resolved.hash,
            "parent": resolved.parent,
            "agent_did": resolved.author_email,
            "committer": resolved.author_name,
            "subject": resolved.subject,
            "committed_at": resolved.committed_at,
        },
        // The PROV-O sidecar (Turtle), inlined so a caller gets the full mark
        // (incl. any block-trail anchor) without a second round-trip. `null`
        // when no sidecar was persisted (e.g. a commit predating git-marks).
        "prov_ttl": sidecar_ttl,
        "anchored": sidecar_ttl.as_deref().map(|t| t.contains("bt:txid")).unwrap_or(false),
    });

    Ok(HttpResponse::Ok()
        .content_type("application/json")
        .json(body))
}

// ---------------------------------------------------------------------------
// POST /{pod}/_prov/anchor
// ---------------------------------------------------------------------------

/// Body of `POST /{pod}/_prov/anchor`: the commit SHA (git-mark) to upgrade to
/// a Bitcoin block-trail anchor, and an optional trail ticker override.
#[derive(Debug, serde::Deserialize)]
struct AnchorBody {
    /// The git commit SHA (a previously-recorded git-mark) to anchor.
    commit_sha: String,
    /// Optional trail ticker; defaults to the pod's configured pay-token.
    #[serde(default)]
    ticker: Option<String>,
}

/// Explicitly upgrade an existing git-mark to a Bitcoin block-trail anchor
/// (the high-value opt-in). NIP-98-authenticated as the pod owner and
/// payment-gated: the caller's Web Ledger is debited by the pod's configured
/// anchor price before the on-chain anchor is produced.
///
/// On success the commit SHA is anchored on the pod's trail (one MRC20 state),
/// and the resource's `.prov.ttl` sidecar is rewritten to carry the anchor so
/// `GET /{pod}/_prov/{commit_sha}` and the sidecar both reflect the upgrade.
async fn handle_anchor(
    path: web::Path<String>,
    req: HttpRequest,
    state: web::Data<AppState>,
    body: Bytes,
) -> Result<HttpResponse, ActixError> {
    let pod = path.into_inner();

    // (1) AuthN: NIP-98, and the caller MUST be the pod owner (only the owner
    // upgrades their own provenance — keys + on-chain spend live pod-side).
    let owner = match require_pod_owner_with_body(&req, &pod, Some(&body)).await {
        Some(pk) => pk,
        None => {
            // Distinguish "no/invalid NIP-98" from "authenticated but not owner".
            return Ok(match extract_pubkey_with_body(&req, Some(&body)).await {
                Some(_) => prov_err("not the pod owner", 403),
                None => prov_err("NIP-98 authentication required", 401),
            });
        }
    };
    let did = format!("did:nostr:{owner}");

    let req_body: AnchorBody = match serde_json::from_slice(&body) {
        Ok(b) => b,
        Err(e) => return Ok(prov_err(&format!("malformed anchor request: {e}"), 400)),
    };
    // Validate the commit id shape early (the git resolver also checks).
    if req_body.commit_sha.is_empty()
        || req_body.commit_sha.len() > 64
        || !req_body.commit_sha.bytes().all(|b| b.is_ascii_hexdigit())
    {
        return Ok(prov_err("invalid commit_sha", 400));
    }

    let Some(repo) = pod_repo_path(&state, &pod) else {
        return Ok(prov_err("git provenance not available", 501));
    };
    if !repo.join(".git").is_dir() {
        return Ok(prov_err("pod is not git-backed", 404));
    }

    // (2) Resolve the commit → resource (must be a real git-mark).
    let resolved = match solid_pod_rs_git::resolve_commit(&repo, &req_body.commit_sha).await {
        Ok(r) => r,
        Err(e) => return Ok(prov_err(&format!("commit not found: {e}"), 404)),
    };
    let Some(resource_rel) = resolved.files.iter().find(|f| !is_sidecar(f)).cloned() else {
        return Ok(prov_err("commit touched no content resource", 404));
    };
    let resource = format!("/{pod}/{resource_rel}");

    // (3) Build the anchorer (pod must have a minted trail).
    let (anchorer, ticker, network) =
        match try_build_anchorer(&state, req_body.ticker.as_deref()).await {
            Ok(Some(bundle)) => bundle,
            Ok(None) => {
                return Ok(prov_err(
                    "pod not configured for Bitcoin anchoring (no pay-token / trail not minted)",
                    400,
                ))
            }
            // Before any charge: an unreadable trail is a server fault, not a 400.
            Err(e) => return Ok(prov_err(&format!("anchor trail unavailable: {e}"), 500)),
        };

    // (4) Payment gate: charge the pod's configured anchor price. Debit BEFORE
    // the on-chain action; if anchoring then fails, refund (credit back).
    let price = anchor_price_sats(&state);
    let charge = if price > 0 {
        match debit(&state, &did, price).await {
            Ok(charge) => Some(charge),
            Err(rsp) => return Ok(rsp),
        }
    } else {
        None
    };

    // (5) Anchor the commit SHA on the trail (the expensive tier). The anchored
    // state_hash IS the git commit SHA — binding git ↔ Bitcoin (§2.3).
    let anchor = match anchorer.anchor(&ticker, &resolved.hash, &network).await {
        Ok(a) => a,
        // The transaction may be on-chain and the next advance settles it:
        // keep the charge and say the anchor is pending.
        Err(e) if crate::mempool::is_pending_anchor(&e) => {
            tracing::warn!(
                target: "solid_pod_rs_server::prov",
                "anchor outcome pending, charge kept: {e}"
            );
            return Ok(HttpResponse::Accepted()
                .content_type("application/json")
                .json(serde_json::json!({
                    "status": "pending",
                    "error": format!("anchor pending: {e}"),
                    "charged_sats": price,
                })));
        }
        Err(e) => {
            // Refund the charge — no anchor was produced. A refund that fails
            // is logged and retained, never dropped.
            let refund_pending = match charge {
                Some(charge) => !refund_or_retain(&state, charge).await,
                None => false,
            };
            return Ok(HttpResponse::BadGateway()
                .content_type("application/json")
                .json(serde_json::json!({
                    "error": format!("anchor failed: {e}"),
                    "refund_pending": refund_pending,
                })));
        }
    };

    // (6) Rewrite the resource's `.prov.ttl` sidecar to carry the anchor.
    let mark = ProvenanceMark {
        resource: resource.clone(),
        git: solid_pod_rs::provenance::GitMark {
            commit_sha: resolved.hash.clone(),
            repo: pod.clone(),
            branch: "main".to_string(),
            parent: resolved.parent.clone(),
        },
        anchor: Some(anchor.clone()),
        agent_did: resolved.author_email.clone(),
        created: resolved.committed_at,
    };
    let ttl = prov_ttl(&mark);
    let sidecar = format!("{resource}.prov.ttl");
    if let Err(e) = state
        .storage
        .put(&sidecar, Bytes::from(ttl.into_bytes()), "text/turtle")
        .await
    {
        // The on-chain anchor succeeded; the sidecar rewrite is best-effort.
        tracing::warn!(
            target: "solid_pod_rs_server::prov",
            sidecar = %sidecar,
            "anchor sidecar rewrite failed (anchor still on-chain): {e}"
        );
    }

    Ok(HttpResponse::Ok()
        .content_type("application/json")
        .json(serde_json::json!({
            "pod": pod,
            "resource": resource,
            "commit_sha": resolved.hash,
            "anchor": {
                "ticker": anchor.ticker,
                "txid": anchor.txid,
                "vout": anchor.vout,
                "address": anchor.address,
                "network": anchor.network,
                "state_hash": anchor.state_hash,
            },
            "charged_sats": price,
        })))
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Whether a repo-relative path is a control-plane / provenance sidecar (never
/// the content resource a git-mark targets).
fn is_sidecar(path: &str) -> bool {
    path.ends_with(".acl") || path.ends_with(".meta") || path.ends_with(".prov.ttl")
}

/// The pod's configured per-anchor price in satoshis (the explicit-upgrade
/// gate). Reuses the pay-token rate as the anchor price; `0` ⇒ ungated.
fn anchor_price_sats(state: &AppState) -> u64 {
    std::env::var("JSS_PROV_ANCHOR_PRICE_SATS")
        .ok()
        .and_then(|v| v.trim().parse::<u64>().ok())
        .unwrap_or_else(|| {
            state
                .pay_config
                .token
                .as_ref()
                .map(|t| t.rate.max(1))
                .unwrap_or(0)
        })
}

/// Charge the caller's Web Ledger `sats`, mapping insufficient-balance to a
/// 402 and other failures to a 500. Uses the same `StoragePaymentStore` ledger
/// path as `/pay/*` so balances are consistent. The returned
/// [`Charge`](solid_pod_rs::payments::Charge) is what [`refund`] gives back.
async fn debit(
    state: &AppState,
    did: &str,
    sats: u64,
) -> Result<solid_pod_rs::payments::Charge, HttpResponse> {
    use crate::handlers::pay::StoragePaymentStore;
    use solid_pod_rs::payments::{PaymentError, PaymentStore};

    // Read, charge and write back under the one payment lock, so a concurrent
    // charge, refund or `/pay/*` transition is never overwritten with a stale
    // ledger. Held only for these storage calls, never across the anchor.
    let _transaction = crate::PAYMENT_STATE_LOCK.lock().await;
    let store = StoragePaymentStore::new(&*state.storage);
    let mut ledger = store
        .read_ledger()
        .await
        .map_err(|e| prov_err(&format!("ledger read failed: {e}"), 500))?;
    let charge = ledger.charge(did, sats).map_err(|e| match e {
        PaymentError::InsufficientBalance { balance, cost } => HttpResponse::PaymentRequired()
            .json(serde_json::json!({
                "error": "Insufficient balance to anchor",
                "balance": balance,
                "cost": cost,
            })),
        other => prov_err(&format!("debit failed: {other}"), 500),
    })?;
    store
        .write_ledger(&ledger)
        .await
        .map_err(|e| prov_err(&format!("ledger write failed: {e}"), 500))?;
    Ok(charge)
}

/// Give back a [`debit`]'s charge (the anchor-failure refund).
async fn refund(state: &AppState, charge: solid_pod_rs::payments::Charge) -> Result<(), String> {
    use crate::handlers::pay::StoragePaymentStore;
    use solid_pod_rs::payments::PaymentStore;

    let _transaction = crate::PAYMENT_STATE_LOCK.lock().await;
    let store = StoragePaymentStore::new(&*state.storage);
    let mut ledger = store.read_ledger().await.map_err(|e| e.to_string())?;
    ledger.refund(charge);
    store.write_ledger(&ledger).await.map_err(|e| e.to_string())
}

/// Pod-storage path of refunds owed but not yet paid back (a failed [`refund`]).
const PENDING_REFUNDS_PATH: &str = "/.well-known/prov/refunds-pending.json";

/// [`refund`] a charge; if that fails, log it and keep a record in
/// [`PENDING_REFUNDS_PATH`] so the debt to the caller survives. Returns whether
/// the refund went through.
async fn refund_or_retain(state: &AppState, charge: solid_pod_rs::payments::Charge) -> bool {
    let (did, amount) = (charge.account().to_string(), charge.amount());
    let Err(reason) = refund(state, charge).await else {
        return true;
    };
    tracing::error!(
        target: "solid_pod_rs_server::prov",
        %did, amount, "anchor refund failed, retaining it as pending: {reason}"
    );
    if let Err(e) = retain_refund(&state.storage, &did, amount, &reason).await {
        tracing::error!(
            target: "solid_pod_rs_server::prov",
            %did, amount, "pending refund could not be recorded either: {e}"
        );
    }
    false
}

/// Append a refund owed to [`PENDING_REFUNDS_PATH`].
async fn retain_refund(
    storage: &Arc<dyn Storage>,
    did: &str,
    amount: u64,
    reason: &str,
) -> Result<(), String> {
    let _guard = crate::trail_store::lock_path(PENDING_REFUNDS_PATH).await;
    let mut owed: Vec<serde_json::Value> = match storage.get(PENDING_REFUNDS_PATH).await {
        Ok((bytes, _)) => {
            serde_json::from_slice(&bytes).map_err(|e| format!("malformed pending refunds: {e}"))?
        }
        Err(PodError::NotFound(_)) => Vec::new(),
        Err(e) => return Err(format!("read pending refunds: {e}")),
    };
    owed.push(serde_json::json!({
        "did": did,
        "amount_sats": amount,
        "error": reason,
        "at": chrono::Utc::now().to_rfc3339(),
    }));
    let body = serde_json::to_vec(&owed).map_err(|e| e.to_string())?;
    storage
        .put(PENDING_REFUNDS_PATH, Bytes::from(body), "application/json")
        .await
        .map(|_| ())
        .map_err(|e| e.to_string())
}

/// JSS-shaped JSON error response.
fn prov_err(msg: &str, status: u16) -> HttpResponse {
    HttpResponse::build(
        actix_web::http::StatusCode::from_u16(status)
            .unwrap_or(actix_web::http::StatusCode::INTERNAL_SERVER_ERROR),
    )
    .content_type("application/json")
    .json(serde_json::json!({ "error": msg }))
}

// ---------------------------------------------------------------------------
// Route registration
// ---------------------------------------------------------------------------

/// Register the `_prov` routes. Must be registered BEFORE the LDP catch-all so
/// `_prov` path segments are never treated as pod resources (mirrors the
/// `/pods/{pk}/_git/*` ordering).
pub fn register(app: &mut web::ServiceConfig) {
    app.route("/{pod}/_prov/anchor", web::post().to(handle_anchor))
        .route("/{pod}/_prov/{commit_sha}", web::get().to(handle_resolve));
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::trail_store::fault_storage::FaultStorage;
    use crate::trail_store::{gitmark_trail_path, trail_path};
    use solid_pod_rs::provenance::{BlockTrailAnchor, ProvenanceError};
    use solid_pod_rs::storage::memory::MemoryBackend;

    /// An anchorer that must never be reached (the epoch never fills).
    struct Unreachable;
    #[async_trait::async_trait(?Send)]
    impl BlockAnchorer for Unreachable {
        async fn anchor(
            &self,
            _: &str,
            _: &str,
            _: &str,
        ) -> Result<BlockTrailAnchor, ProvenanceError> {
            panic!("epoch must not close in this test")
        }
        async fn verify(&self, _: &BlockTrailAnchor) -> Result<bool, ProvenanceError> {
            Ok(false)
        }
    }

    fn state_over(storage: Arc<dyn Storage>) -> AppState {
        AppState::new(storage)
    }

    async fn epoch_on_disk(inner: &Arc<dyn Storage>) -> Vec<String> {
        let (bytes, _) = inner.get(EPOCH_PATH).await.unwrap();
        serde_json::from_slice(&bytes).unwrap()
    }

    #[tokio::test]
    async fn epoch_read_error_keeps_the_pending_batch() {
        let inner: Arc<dyn Storage> = Arc::new(MemoryBackend::new());
        save_epoch_commits(&inner, &["a".into(), "b".into()])
            .await
            .unwrap();
        let faulty = FaultStorage::new(inner.clone(), false);
        faulty.fail_get(EPOCH_PATH);
        let state = state_over(faulty.clone());
        let anchorer: Arc<dyn BlockAnchorer> = Arc::new(Unreachable);

        let out = epoch_push_and_maybe_anchor(&state, &anchorer, "prov", "testnet4", "c").await;
        assert!(
            out.is_err(),
            "a failed read must not look like an empty batch"
        );
        assert_eq!(epoch_on_disk(&inner).await, vec!["a", "b"], "batch lost");
    }

    #[tokio::test]
    async fn epoch_corrupt_batch_is_not_overwritten() {
        let inner: Arc<dyn Storage> = Arc::new(MemoryBackend::new());
        inner
            .put(
                EPOCH_PATH,
                Bytes::from_static(b"{not json"),
                "application/json",
            )
            .await
            .unwrap();
        let state = state_over(inner.clone());
        let anchorer: Arc<dyn BlockAnchorer> = Arc::new(Unreachable);
        let out = epoch_push_and_maybe_anchor(&state, &anchorer, "prov", "testnet4", "c").await;
        assert!(out.is_err());
        let (bytes, _) = inner.get(EPOCH_PATH).await.unwrap();
        assert_eq!(&bytes[..], b"{not json");
    }

    #[tokio::test]
    async fn concurrent_epoch_pushes_both_survive() {
        let inner: Arc<dyn Storage> = Arc::new(MemoryBackend::new());
        let faulty = FaultStorage::new(inner.clone(), true);
        let state = state_over(faulty.clone());
        let anchorer: Arc<dyn BlockAnchorer> = Arc::new(Unreachable);
        let (a, b) = tokio::join!(
            epoch_push_and_maybe_anchor(&state, &anchorer, "prov", "testnet4", "a"),
            epoch_push_and_maybe_anchor(&state, &anchorer, "prov", "testnet4", "b"),
        );
        a.unwrap();
        b.unwrap();
        let mut got = epoch_on_disk(&inner).await;
        got.sort();
        assert_eq!(got, vec!["a", "b"], "lost update in the epoch RMW");
    }

    /// A failing anchor must be reported by the hook (as `anchor_error`), so
    /// `build_anchorer` still hands back an anchorer when the trail read fails.
    #[tokio::test]
    async fn build_anchorer_storage_error_is_not_silent_git_only() {
        let inner: Arc<dyn Storage> = Arc::new(MemoryBackend::new());
        let faulty = FaultStorage::new(inner, false);
        faulty.fail_get(&gitmark_trail_path("prov"));
        let state = state_over(faulty.clone());
        let (anchorer, ticker, network) = build_anchorer(&state, None)
            .await
            .expect("a storage fault must surface as a failing anchorer, not git-only");
        let err = anchorer.anchor(&ticker, "ab", &network).await.unwrap_err();
        assert!(err.to_string().contains("injected fault"), "{err}");
    }

    #[tokio::test]
    async fn build_anchorer_mrc20_trail_read_error_does_not_fall_through() {
        let inner: Arc<dyn Storage> = Arc::new(MemoryBackend::new());
        let faulty = FaultStorage::new(inner, false);
        faulty.fail_get(&trail_path("prov"));
        let mut state = state_over(faulty.clone());
        state.pay_config.token = Some(solid_pod_rs::payments::TokenConfig {
            ticker: "prov".into(),
            rate: 1,
            supply: 1000,
            issuer: "02".to_string() + &"ab".repeat(32),
            accepted_issuers: vec![],
        });
        let (anchorer, ticker, network) = build_anchorer(&state, None)
            .await
            .expect("an MRC20 read fault must not fall through to git-mark/None");
        let err = anchorer.anchor(&ticker, "ab", &network).await.unwrap_err();
        assert!(err.to_string().contains("injected fault"), "{err}");
    }

    #[tokio::test]
    async fn failed_refund_is_logged_and_retained() {
        use solid_pod_rs::payments::{PaymentStore, ReceiptOutcome};
        let inner: Arc<dyn Storage> = Arc::new(MemoryBackend::new());
        let faulty = FaultStorage::new(inner.clone(), false);
        let state = state_over(faulty.clone());
        let did = format!("did:nostr:{}", "ab".repeat(32));
        {
            let store = crate::handlers::pay::StoragePaymentStore::new(&*state.storage);
            let mut ledger = store.read_ledger().await.unwrap();
            let r = ledger
                .credit_by_outpoint(&did, "satoshi", &"cd".repeat(32), 0, 100)
                .unwrap();
            assert_eq!(r, ReceiptOutcome::Applied);
            store.write_ledger(&ledger).await.unwrap();
        }
        let Ok(charge) = debit(&state, &did, 40).await else {
            panic!("debit failed");
        };
        // The ledger write that would give the money back fails.
        faulty.fail_put("/.well-known/webledgers/state.json");
        assert!(!refund_or_retain(&state, charge).await);
        let (bytes, _) = inner.get(PENDING_REFUNDS_PATH).await.unwrap();
        let owed: Vec<serde_json::Value> = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(owed.len(), 1);
        assert_eq!(owed[0]["did"], did.as_str());
        assert_eq!(owed[0]["amount_sats"], 40);
    }

    /// An anchorer whose every anchor fails (the chain is unreachable).
    struct Failing;
    #[async_trait::async_trait(?Send)]
    impl BlockAnchorer for Failing {
        async fn anchor(
            &self,
            _: &str,
            _: &str,
            _: &str,
        ) -> Result<BlockTrailAnchor, ProvenanceError> {
            Err(ProvenanceError::Anchor("chain unreachable".into()))
        }
        async fn verify(&self, _: &BlockTrailAnchor) -> Result<bool, ProvenanceError> {
            Ok(false)
        }
    }

    /// An anchorer that counts its calls and, once reached, makes the next
    /// write of the epoch file fail (the anchor lands, the reset is lost).
    struct CountingThenFault {
        calls: std::sync::atomic::AtomicUsize,
        storage: Arc<FaultStorage>,
    }
    #[async_trait::async_trait(?Send)]
    impl BlockAnchorer for CountingThenFault {
        async fn anchor(
            &self,
            ticker: &str,
            state_hash: &str,
            network: &str,
        ) -> Result<BlockTrailAnchor, ProvenanceError> {
            self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            self.storage.fail_put(EPOCH_PATH);
            Ok(BlockTrailAnchor {
                ticker: ticker.into(),
                state_hash: state_hash.into(),
                txid: "ab".repeat(32),
                vout: 0,
                address: String::new(),
                network: network.into(),
                blockheight: None,
                state_strings: vec![],
                pubkey: None,
            })
        }
        async fn verify(&self, _: &BlockTrailAnchor) -> Result<bool, ProvenanceError> {
            Ok(true)
        }
    }

    #[tokio::test]
    async fn lost_epoch_reset_does_not_anchor_the_same_batch_twice() {
        let inner: Arc<dyn Storage> = Arc::new(MemoryBackend::new());
        let full: Vec<String> = (0..DEFAULT_EPOCH_SIZE - 1)
            .map(|i| format!("c{i}"))
            .collect();
        save_epoch_commits(&inner, &full).await.unwrap();
        let faulty = FaultStorage::new(inner.clone(), false);
        let state = state_over(faulty.clone());
        let counting = Arc::new(CountingThenFault {
            calls: Default::default(),
            storage: faulty.clone(),
        });
        let anchorer: Arc<dyn BlockAnchorer> = counting.clone();
        let first =
            epoch_push_and_maybe_anchor(&state, &anchorer, "prov", "testnet4", "last").await;
        // the storage recovers; another commit arrives
        faulty.heal();
        let second =
            epoch_push_and_maybe_anchor(&state, &anchorer, "prov", "testnet4", "next").await;
        assert_eq!(
            counting.calls.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "the anchored batch was anchored again"
        );
        assert!(first.is_ok_and(|c| c.is_some()), "the batch was anchored");
        assert!(second.is_ok_and(|c| c.is_none()));
        assert_eq!(epoch_on_disk(&inner).await, vec!["next"]);
        assert!(!inner.exists(EPOCH_ANCHORED_PATH).await.unwrap());
    }

    #[tokio::test]
    async fn failing_anchor_cannot_grow_the_epoch_without_bound() {
        let inner: Arc<dyn Storage> = Arc::new(MemoryBackend::new());
        let backlog: Vec<String> = (0..DEFAULT_EPOCH_SIZE * EPOCH_BACKLOG_FACTOR)
            .map(|i| format!("c{i}"))
            .collect();
        save_epoch_commits(&inner, &backlog).await.unwrap();
        let state = state_over(inner.clone());
        let anchorer: Arc<dyn BlockAnchorer> = Arc::new(Failing);
        let out = epoch_push_and_maybe_anchor(&state, &anchorer, "prov", "testnet4", "more").await;
        assert!(out.is_err());
        assert_eq!(epoch_on_disk(&inner).await, backlog, "epoch.json grew");
    }

    #[tokio::test]
    async fn a_full_backlog_still_drains_once_anchoring_recovers() {
        let inner: Arc<dyn Storage> = Arc::new(MemoryBackend::new());
        let backlog: Vec<String> = (0..DEFAULT_EPOCH_SIZE * EPOCH_BACKLOG_FACTOR)
            .map(|i| format!("c{i}"))
            .collect();
        save_epoch_commits(&inner, &backlog).await.unwrap();
        let state = state_over(inner.clone());
        struct Works;
        #[async_trait::async_trait(?Send)]
        impl BlockAnchorer for Works {
            async fn anchor(
                &self,
                ticker: &str,
                state_hash: &str,
                network: &str,
            ) -> Result<BlockTrailAnchor, ProvenanceError> {
                Ok(BlockTrailAnchor {
                    ticker: ticker.into(),
                    state_hash: state_hash.into(),
                    txid: "ab".repeat(32),
                    vout: 0,
                    address: String::new(),
                    network: network.into(),
                    blockheight: None,
                    state_strings: vec![],
                    pubkey: None,
                })
            }
            async fn verify(&self, _: &BlockTrailAnchor) -> Result<bool, ProvenanceError> {
                Ok(true)
            }
        }
        let anchorer: Arc<dyn BlockAnchorer> = Arc::new(Works);
        let closed = epoch_push_and_maybe_anchor(&state, &anchorer, "prov", "testnet4", "more")
            .await
            .unwrap()
            .expect("the epoch must close once anchoring works again");
        assert_eq!(closed.commits, backlog, "every backlogged commit is rooted");
        assert!(epoch_on_disk(&inner).await.is_empty());
    }

    #[tokio::test]
    async fn failed_epoch_close_keeps_the_commit_that_filled_it() {
        let inner: Arc<dyn Storage> = Arc::new(MemoryBackend::new());
        let full: Vec<String> = (0..DEFAULT_EPOCH_SIZE - 1)
            .map(|i| format!("c{i}"))
            .collect();
        save_epoch_commits(&inner, &full).await.unwrap();
        let state = state_over(inner.clone());
        let anchorer: Arc<dyn BlockAnchorer> = Arc::new(Failing);
        let out = epoch_push_and_maybe_anchor(&state, &anchorer, "prov", "testnet4", "last").await;
        assert!(out.is_err());
        let kept = epoch_on_disk(&inner).await;
        assert_eq!(kept.len(), DEFAULT_EPOCH_SIZE);
        assert_eq!(kept.last().map(String::as_str), Some("last"));
    }

    /// A ledger holding `sats` for `did`, over a storage that yields on every
    /// access (so unlocked read-modify-writes interleave).
    async fn funded(sats: u64) -> (AppState, String) {
        use solid_pod_rs::payments::{PaymentStore, ReceiptOutcome};
        let inner: Arc<dyn Storage> = Arc::new(MemoryBackend::new());
        let state = state_over(FaultStorage::new(inner, true));
        let did = format!("did:nostr:{}", "ab".repeat(32));
        let store = crate::handlers::pay::StoragePaymentStore::new(&*state.storage);
        let mut ledger = store.read_ledger().await.unwrap();
        let r = ledger
            .credit_by_outpoint(&did, "satoshi", &"cd".repeat(32), 0, sats)
            .unwrap();
        assert_eq!(r, ReceiptOutcome::Applied);
        store.write_ledger(&ledger).await.unwrap();
        (state, did)
    }

    async fn balance(state: &AppState, did: &str) -> u64 {
        use solid_pod_rs::payments::PaymentStore;
        let store = crate::handlers::pay::StoragePaymentStore::new(&*state.storage);
        store.read_ledger().await.unwrap().get_balance(did)
    }

    #[tokio::test]
    async fn concurrent_debits_do_not_lose_an_update() {
        let (state, did) = funded(100).await;
        let (a, b) = tokio::join!(debit(&state, &did, 40), debit(&state, &did, 40));
        assert!(a.is_ok() && b.is_ok());
        assert_eq!(balance(&state, &did).await, 20);
    }

    #[tokio::test]
    async fn concurrent_debit_and_refund_keep_both_effects() {
        let (state, did) = funded(100).await;
        let held = debit(&state, &did, 40).await.unwrap();
        let (refunded, debited) = tokio::join!(refund(&state, held), debit(&state, &did, 30));
        refunded.unwrap();
        assert!(debited.is_ok());
        assert_eq!(balance(&state, &did).await, 70);
    }

    #[tokio::test]
    async fn concurrent_debits_cannot_overspend() {
        let (state, did) = funded(100).await;
        let (a, b) = tokio::join!(debit(&state, &did, 60), debit(&state, &did, 60));
        assert_eq!(
            a.is_ok() as u8 + b.is_ok() as u8,
            1,
            "exactly one debit fits"
        );
        assert_eq!(balance(&state, &did).await, 40);
    }
}