udb 0.4.20

//! XA / 2PC opt-in coordinator (U20).
//!
//! Pre-U20 the capability matrix declared `supports_xa` and
//! `supports_two_phase_commit` per backend, but no live runtime path
//! actually drove a distributed transaction. Operators saw `pg=true` in
//! `GetCapabilities` and couldn't act on it.
//!
//! U20 adds the live path:
//!
//! - [`TransactionStrategy`] — request-level enum chosen by the caller
//!   (`Saga` / `BestEffort` / `TwoPhase`). Mirrors the consistency-mode
//!   typed pattern from U6.
//! - [`XaCoordinator`] — orchestrates PREPARE → vote → COMMIT/ROLLBACK
//!   across a list of participants. The participants are
//!   capability-checked **before** any side-effect (so we fail fast
//!   instead of committing on PG and then discovering Qdrant doesn't
//!   support 2PC).
//! - [`XaParticipant`] trait — what a backend implements to participate.
//!   PostgresExecutor is the first impl (emits `PREPARE TRANSACTION 'xid'`
//!   / `COMMIT PREPARED 'xid'` SQL). Tests plug stubs.
//! - [`XaLedgerEntry`] — durable record of the coordinator's decision,
//!   used by the recovery worker on broker restart to drive in-doubt
//!   transactions to a terminal state.
//!
//! The acceptance gate ("A request asking for 2PC either commits all
//! supported participants or fails before side effects are issued") is
//! satisfied by `XaCoordinator::execute`:
//!
//! 1. **Capability gate** — `validate_participants` checks every
//!    participant's `supports_two_phase_commit` against the manifest's
//!    capability matrix. If ANY participant can't 2PC, the coordinator
//!    refuses **before** PREPARE. The request fails with
//!    `tonic::Code::FailedPrecondition`.
//! 2. **PREPARE phase** — issued to every participant in declaration
//!    order. Any failure → all already-prepared participants
//!    ROLLBACK PREPARED. No participant has committed any side-effect.
//! 3. **COMMIT phase** — only after every participant voted yes.
//!    Failure here means the in-doubt state is durably recorded in
//!    `XaLedgerEntry`; the recovery worker drives it to completion.

use std::time::{SystemTime, UNIX_EPOCH};

use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::backend::BackendCapabilityMatrixEntry;

/// Reason stamped on the write-ahead (pre-PHASE-2) ledger row. MUST NOT
/// contain any of the prepare-failure phrases
/// `xa_recovery::InDoubtLedgerRow::target_intent` matches on, so the
/// recovery worker resolves the row towards COMMIT.
pub const XA_COMMIT_INTENT_REASON: &str = "commit decided; phase 2 in flight";

/// Which transaction strategy the caller wants for a cross-backend
/// mutation. The runtime picks based on this + the participant set;
/// the strategy is the **upper bound** on commitment guarantees.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TransactionStrategy {
    /// Default — write succeeds locally per backend; failures land in
    /// the saga driver for compensation. Eventually consistent.
    Saga,
    /// Try each backend; on failure, **best-effort** compensate the
    /// already-succeeded ones inline. No durable in-doubt state — if the
    /// broker crashes mid-compensation, the saga driver recovers it. The
    /// pre-U20 default for the typed transaction RPC.
    BestEffort,
    /// True 2PC across XA/2PC-capable participants. Failure modes:
    /// - Capability gate fails → request refused, zero side effects.
    /// - PREPARE fails on any participant → all prepared participants
    ///   ROLLBACK PREPARED, request fails, zero durable side effects.
    /// - COMMIT fails on any participant → `XaLedgerEntry` records the
    ///   in-doubt state; the recovery worker drives it to completion
    ///   on broker restart.
    TwoPhase,
}

impl Default for TransactionStrategy {
    fn default() -> Self {
        Self::Saga
    }
}

impl TransactionStrategy {
    /// Pinned wire token. Mirrors `ConsistencyMode::as_str` — SDK enums
    /// and `x-udb-transaction-strategy` header use these strings.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Saga => "saga",
            Self::BestEffort => "best_effort",
            Self::TwoPhase => "two_phase",
        }
    }

    pub fn parse(token: &str) -> Option<Self> {
        match token.trim().to_ascii_lowercase().as_str() {
            "saga" => Some(Self::Saga),
            "best_effort" | "best-effort" => Some(Self::BestEffort),
            "two_phase" | "two-phase" | "2pc" | "xa" => Some(Self::TwoPhase),
            _ => None,
        }
    }

    pub fn parse_or_default(token: &str) -> Self {
        Self::parse(token).unwrap_or_default()
    }

    pub fn requires_two_phase(self) -> bool {
        matches!(self, Self::TwoPhase)
    }
}

/// One participant in a 2PC transaction. The runtime constructs a
/// `XaParticipantHandle` per backend involved and hands it to the
/// coordinator.
#[derive(Debug, Clone)]
pub struct XaParticipantHandle {
    pub backend: String,
    pub instance: String,
    /// Label used in audit logs + the ledger row (`postgres:primary` etc).
    pub label: String,
}

impl XaParticipantHandle {
    pub fn new(backend: impl Into<String>, instance: impl Into<String>) -> Self {
        let backend = backend.into();
        let instance = instance.into();
        let label = format!("{backend}:{instance}");
        Self {
            backend,
            instance,
            label,
        }
    }
}

/// Result of one participant's vote in PHASE 1 (PREPARE). The
/// coordinator collects every vote before deciding to commit.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum PrepareVote {
    /// Participant successfully prepared and is ready to commit.
    Prepared,
    /// Participant refused (typed reason).
    Aborted { reason: String },
}

/// What the coordinator decided at PHASE 2. Stored in `XaLedgerEntry`
/// so the recovery worker knows which way to drive an in-doubt xid.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum XaDecision {
    /// PHASE 1 succeeded, PHASE 2 committed on every participant.
    Committed,
    /// Some participant aborted in PHASE 1 or the coordinator
    /// pre-emptively rolled back; every prepared participant was
    /// rolled back.
    RolledBack,
    /// PHASE 2 mid-flight — coordinator crashed or some participant is
    /// unreachable. The recovery worker scans this state on restart and
    /// drives it to `Committed` (re-issuing COMMIT PREPARED) or
    /// `RolledBack` based on the per-participant outcome ledger.
    InDoubt,
    /// The recovery worker failed `RecoveryConfig::max_attempts` times
    /// to drive this xid terminal. Parked: no automatic retries, an
    /// operator must resolve it manually (the backend's prepared xact —
    /// if any — is intentionally left untouched so no decision is lost).
    ManualReview,
}

impl XaDecision {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Committed => "committed",
            Self::RolledBack => "rolled_back",
            Self::InDoubt => "in_doubt",
            Self::ManualReview => "manual_review",
        }
    }

    pub fn is_terminal(self) -> bool {
        matches!(self, Self::Committed | Self::RolledBack)
    }
}

/// One row in the `udb_xa_ledger` table — durable record of a 2PC
/// transaction's lifecycle. The recovery worker scans rows where
/// `decision = InDoubt` on broker startup.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct XaLedgerEntry {
    /// Cross-participant transaction id. Same value goes into
    /// `PREPARE TRANSACTION '<xid>'` on every PG participant.
    pub xid: String,
    pub tenant_id: String,
    pub project_id: String,
    /// Where the request originated (RPC + correlation id).
    pub origin_rpc: String,
    pub correlation_id: String,
    /// Per-participant labels (`postgres:primary` …) in declaration
    /// order. The recovery worker reproduces the participant set from
    /// this so it can re-issue COMMIT/ROLLBACK PREPARED on the right
    /// pool.
    pub participants: Vec<String>,
    pub decision: XaDecision,
    /// Unix milliseconds when the coordinator made its decision.
    pub decided_at_unix_ms: i64,
    /// Optional reason — populated when decision is `RolledBack` (the
    /// failing participant's PREPARE vote) or `InDoubt` (the
    /// transport / coordinator error).
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub reason: String,
}

impl XaLedgerEntry {
    pub fn new(
        xid: impl Into<String>,
        tenant_id: impl Into<String>,
        project_id: impl Into<String>,
        origin_rpc: impl Into<String>,
        correlation_id: impl Into<String>,
        participants: Vec<String>,
        decision: XaDecision,
    ) -> Self {
        Self {
            xid: xid.into(),
            tenant_id: tenant_id.into(),
            project_id: project_id.into(),
            origin_rpc: origin_rpc.into(),
            correlation_id: correlation_id.into(),
            participants,
            decision,
            decided_at_unix_ms: SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .map(|d| d.as_millis() as i64)
                .unwrap_or(0),
            reason: String::new(),
        }
    }

    pub fn with_reason(mut self, reason: impl Into<String>) -> Self {
        self.reason = reason.into();
        self
    }
}

/// What the runtime returns to the gRPC handler after running the
/// coordinator. Carries the ledger entry so the handler can audit-log
/// it, plus the per-participant outcome for response shaping.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct XaOutcome {
    pub ledger: XaLedgerEntry,
    pub participants: Vec<ParticipantOutcome>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ParticipantOutcome {
    pub label: String,
    pub vote: PrepareVote,
    pub committed: bool,
}

/// Error from the coordinator. Mapped to `tonic::Status` by the gRPC
/// handler — `CapabilityRefused` → `FailedPrecondition`, `PrepareFailed`
/// → `Aborted`, `InDoubt` → `Internal` (operator must consult the
/// recovery worker logs).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum XaError {
    /// At least one participant doesn't support 2PC. Refused before any
    /// side effect.
    CapabilityRefused { unsupported: Vec<String> },
    /// PHASE 1 failed; every prepared participant was rolled back.
    PrepareFailed {
        ledger: XaLedgerEntry,
        failures: Vec<ParticipantOutcome>,
    },
    /// PHASE 2 was reached but couldn't be driven to completion. The
    /// ledger is durably recorded; recovery is the operator's
    /// responsibility (the recovery worker will retry).
    InDoubt { ledger: XaLedgerEntry },
}

impl std::fmt::Display for XaError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::CapabilityRefused { unsupported } => write!(
                f,
                "2PC refused: {} participant(s) lack supports_two_phase_commit: {}",
                unsupported.len(),
                unsupported.join(", ")
            ),
            Self::PrepareFailed { ledger, .. } => write!(
                f,
                "2PC PREPARE phase failed for xid {}; rolled back: {}",
                ledger.xid, ledger.reason
            ),
            Self::InDoubt { ledger } => write!(
                f,
                "2PC xid {} is in-doubt; recovery worker will drive to terminal state",
                ledger.xid
            ),
        }
    }
}

impl std::error::Error for XaError {}

/// What a backend implements to participate in 2PC. The runtime
/// constructs one of these per participating instance and hands them
/// to the coordinator.
///
/// All three methods take `&self` so a coordinator can call them
/// concurrently against different participants. Implementations bind
/// the per-request connection / pool internally.
#[async_trait::async_trait]
pub trait XaParticipant: Send + Sync {
    fn handle(&self) -> &XaParticipantHandle;

    /// PHASE 1: persist the transaction in a prepared state so a
    /// subsequent `commit_prepared` or `rollback_prepared` is guaranteed
    /// to succeed. For Postgres this is `PREPARE TRANSACTION '<xid>'`.
    async fn prepare(&self, xid: &str) -> PrepareVote;

    /// PHASE 2 (success path): commit the prepared transaction.
    /// For Postgres this is `COMMIT PREPARED '<xid>'`.
    async fn commit_prepared(&self, xid: &str) -> Result<(), String>;

    /// PHASE 2 (failure path): roll back the prepared transaction.
    /// For Postgres this is `ROLLBACK PREPARED '<xid>'`.
    async fn rollback_prepared(&self, xid: &str) -> Result<(), String>;
}

/// Per-request inputs the coordinator needs.
#[derive(Debug, Clone)]
pub struct XaRequest {
    pub tenant_id: String,
    pub project_id: String,
    pub origin_rpc: String,
    pub correlation_id: String,
}

/// The coordinator. Stateless — call `execute` per request with the
/// participant list. The runtime composes participants from the
/// dispatch resolver (one `XaParticipant` per backend the request
/// touches).
pub struct XaCoordinator;

impl XaCoordinator {
    /// Generate an xid. PostgreSQL identifier rules apply
    /// (max 200 chars, ascii); we use `udb-<uuid>` which is 40 chars.
    pub fn new_xid() -> String {
        format!("udb-{}", Uuid::new_v4())
    }

    /// Check every participant's capability against the manifest's
    /// capability matrix. Refuses the request when any participant
    /// can't 2PC — **before** any PREPARE side-effect. The acceptance
    /// gate ("commits all or fails before side effects").
    pub fn validate_participants(
        participants: &[XaParticipantHandle],
        capability_matrix: &[BackendCapabilityMatrixEntry],
    ) -> Result<(), XaError> {
        let mut unsupported: Vec<String> = Vec::new();
        for p in participants {
            let entry = capability_matrix
                .iter()
                .find(|e| e.backend.eq_ignore_ascii_case(&p.backend));
            match entry {
                Some(e) if e.supports_two_phase_commit => {}
                Some(_) => unsupported.push(p.label.clone()),
                None => unsupported.push(format!("{}:unknown", p.label)),
            }
        }
        if !unsupported.is_empty() {
            return Err(XaError::CapabilityRefused { unsupported });
        }
        Ok(())
    }

    /// Run the full 2PC lifecycle. Returns `XaOutcome` on success, or
    /// `XaError` on capability/PREPARE failure / in-doubt state.
    pub async fn execute(
        request: XaRequest,
        participants: Vec<Box<dyn XaParticipant>>,
        capability_matrix: &[BackendCapabilityMatrixEntry],
    ) -> Result<XaOutcome, XaError> {
        Self::execute_with_write_ahead(request, participants, capability_matrix, |_| async {
            Ok(())
        })
        .await
    }

    /// Like [`Self::execute`] but invokes `write_ahead` with a
    /// commit-intent `XaLedgerEntry` (decision `InDoubt`, reason
    /// signalling commit) after every participant voted Prepared and
    /// BEFORE the first `commit_prepared` is issued. This is the
    /// write-ahead decision record: a crash mid-PHASE-2 leaves a durable
    /// commit-intent row, so the presumed-abort sweep can never roll
    /// back a commit-decided transaction. If `write_ahead` fails, the
    /// coordinator fails CLOSED — every prepared participant is rolled
    /// back and the request aborts without committing anything.
    pub async fn execute_with_write_ahead<F, Fut>(
        request: XaRequest,
        participants: Vec<Box<dyn XaParticipant>>,
        capability_matrix: &[BackendCapabilityMatrixEntry],
        write_ahead: F,
    ) -> Result<XaOutcome, XaError>
    where
        F: FnOnce(XaLedgerEntry) -> Fut + Send,
        Fut: std::future::Future<Output = Result<(), String>> + Send,
    {
        let handles: Vec<XaParticipantHandle> =
            participants.iter().map(|p| p.handle().clone()).collect();
        let labels: Vec<String> = handles.iter().map(|h| h.label.clone()).collect();
        Self::validate_participants(&handles, capability_matrix)?;

        let xid = Self::new_xid();

        // PHASE 1: PREPARE every participant. Collect votes.
        let mut outcomes: Vec<ParticipantOutcome> = Vec::with_capacity(participants.len());
        let mut had_failure = false;
        let mut first_failure_reason = String::new();
        for p in &participants {
            let vote = p.prepare(&xid).await;
            if let PrepareVote::Aborted { reason } = &vote {
                had_failure = true;
                if first_failure_reason.is_empty() {
                    first_failure_reason = reason.clone();
                }
            }
            outcomes.push(ParticipantOutcome {
                label: p.handle().label.clone(),
                vote,
                committed: false,
            });
        }

        if had_failure {
            // Roll back every prepared participant. Best-effort —
            // rollback failures are logged but don't escalate (the
            // transaction is "rolled back" from the coordinator's
            // perspective; recovery worker handles stragglers).
            for (i, p) in participants.iter().enumerate() {
                if matches!(outcomes[i].vote, PrepareVote::Prepared)
                    && let Err(err) = p.rollback_prepared(&xid).await
                {
                    tracing::warn!(
                        xid = %xid,
                        participant = %outcomes[i].label,
                        error = %err,
                        "ROLLBACK PREPARED failed during prepare-phase abort",
                    );
                }
            }
            let ledger = XaLedgerEntry::new(
                xid,
                request.tenant_id,
                request.project_id,
                request.origin_rpc,
                request.correlation_id,
                labels,
                XaDecision::RolledBack,
            )
            .with_reason(first_failure_reason);
            return Err(XaError::PrepareFailed {
                ledger,
                failures: outcomes,
            });
        }

        // Write-ahead decision record (item 5): persist the COMMIT
        // intent BEFORE the first commit_prepared so a mid-PHASE-2
        // crash leaves a commit-intent ledger row instead of nothing.
        // The reason string deliberately avoids the prepare-failure
        // phrases `InDoubtLedgerRow::target_intent` recognises, so
        // recovery drives this xid towards COMMIT.
        let write_ahead_entry = XaLedgerEntry::new(
            xid.clone(),
            request.tenant_id.clone(),
            request.project_id.clone(),
            request.origin_rpc.clone(),
            request.correlation_id.clone(),
            labels.clone(),
            XaDecision::InDoubt,
        )
        .with_reason(XA_COMMIT_INTENT_REASON);
        if let Err(err) = write_ahead(write_ahead_entry).await {
            // Fail closed: without a durable decision record we must not
            // commit (a crash mid-PHASE-2 would be unrecoverable). Every
            // prepared participant rolls back.
            for (i, p) in participants.iter().enumerate() {
                if let Err(rb_err) = p.rollback_prepared(&xid).await {
                    tracing::warn!(
                        xid = %xid,
                        participant = %outcomes[i].label,
                        error = %rb_err,
                        "ROLLBACK PREPARED failed during write-ahead-ledger abort",
                    );
                }
            }
            let ledger = XaLedgerEntry::new(
                xid,
                request.tenant_id,
                request.project_id,
                request.origin_rpc,
                request.correlation_id,
                labels,
                XaDecision::RolledBack,
            )
            .with_reason(format!("write-ahead XA ledger insert failed: {err}"));
            return Err(XaError::PrepareFailed {
                ledger,
                failures: outcomes,
            });
        }

        // PHASE 2: COMMIT PREPARED. If any participant fails commit,
        // the transaction enters in-doubt state — durably record and
        // surface to the operator. Recovery worker drives stragglers.
        let mut commit_error: Option<String> = None;
        for (i, p) in participants.iter().enumerate() {
            match p.commit_prepared(&xid).await {
                Ok(()) => outcomes[i].committed = true,
                Err(err) => {
                    if commit_error.is_none() {
                        commit_error = Some(format!("{}: {err}", outcomes[i].label));
                    }
                }
            }
        }

        if let Some(reason) = commit_error {
            let ledger = XaLedgerEntry::new(
                xid,
                request.tenant_id,
                request.project_id,
                request.origin_rpc,
                request.correlation_id,
                labels,
                XaDecision::InDoubt,
            )
            .with_reason(reason);
            return Err(XaError::InDoubt { ledger });
        }

        Ok(XaOutcome {
            ledger: XaLedgerEntry::new(
                xid,
                request.tenant_id,
                request.project_id,
                request.origin_rpc,
                request.correlation_id,
                labels,
                XaDecision::Committed,
            ),
            participants: outcomes,
        })
    }
}

// ── Concrete XA participants ─────────────────────────────────────────────────

/// Translate one Postgres-dialect `SqlOperationPlan` statement (the
/// machine-generated shape from `build_upsert_plan` / `build_delete_plan`:
/// `$N` placeholders in ascending order, double-quoted identifiers,
/// `ON CONFLICT (…) DO …`) into MySQL dialect for replay inside an XA
/// transaction (`XaMysqlParticipant::prepared_statements`). Plan SQL never
/// embeds literal values — everything is parameterized — so the
/// identifier-quote and placeholder rewrites are purely structural. Fails
/// CLOSED on any shape it cannot prove a faithful translation for.
pub fn translate_pg_plan_sql_to_mysql(sql: &str) -> Result<String, String> {
    if sql.to_ascii_uppercase().contains("RETURNING") {
        return Err("RETURNING is not supported by the MySQL XA participant".to_string());
    }
    if sql.contains("\"\"") {
        return Err(
            "identifiers containing embedded double quotes cannot be translated to MySQL"
                .to_string(),
        );
    }
    if sql.contains('`') {
        return Err("plan SQL already contains backticks; refusing ambiguous translation".into());
    }
    // Plan identifiers use PG double-quote quoting; MySQL (without
    // ANSI_QUOTES) treats `"x"` as a string literal, so rewrite to
    // backticks. `ILIKE` maps to `LIKE`, which is case-insensitive under
    // MySQL's default *_ci collations.
    let quoted = sql.replace('"', "`").replace(" ILIKE ", " LIKE ");
    let mut out = String::with_capacity(quoted.len());
    let mut expected = 1usize;
    let mut chars = quoted.chars().peekable();
    while let Some(c) = chars.next() {
        if c != '$' {
            out.push(c);
            continue;
        }
        let mut digits = String::new();
        while let Some(d) = chars.peek().copied() {
            if d.is_ascii_digit() {
                digits.push(d);
                chars.next();
            } else {
                break;
            }
        }
        if digits.is_empty() {
            return Err("unexpected '$' without parameter number in plan SQL".to_string());
        }
        let n: usize = digits
            .parse()
            .map_err(|e| format!("invalid placeholder ${digits}: {e}"))?;
        if n != expected {
            return Err(format!(
                "placeholder ${n} out of order (expected ${expected}); MySQL binds positionally"
            ));
        }
        expected += 1;
        out.push('?');
    }
    rewrite_on_conflict_for_mysql(&out)
}

/// Rewrite the PG `ON CONFLICT` tail into MySQL `ON DUPLICATE KEY UPDATE`.
/// Caveat (documented, accepted): MySQL's clause fires on ANY unique-key
/// conflict, not only the columns PG's plan named — same rows-or-fewer
/// semantics for the no-op form, and the standard MySQL upsert idiom for
/// the update form.
fn rewrite_on_conflict_for_mysql(sql: &str) -> Result<String, String> {
    const MARKER: &str = " ON CONFLICT (";
    let Some(pos) = sql.find(MARKER) else {
        return Ok(sql.to_string());
    };
    let head = &sql[..pos];
    let rest = &sql[pos + MARKER.len()..];
    let close = rest
        .find(')')
        .ok_or_else(|| "malformed ON CONFLICT clause".to_string())?;
    let conflict_columns = &rest[..close];
    let tail = rest[close + 1..].trim();
    if tail == "DO NOTHING" {
        // MySQL no-op idiom: assign the first conflict column to itself.
        // Unlike INSERT IGNORE this only suppresses duplicate-key rows.
        let first = conflict_columns
            .split(',')
            .next()
            .map(str::trim)
            .filter(|c| !c.is_empty())
            .ok_or_else(|| "ON CONFLICT clause without columns".to_string())?;
        return Ok(format!("{head} ON DUPLICATE KEY UPDATE {first} = {first}"));
    }
    let Some(assignments) = tail.strip_prefix("DO UPDATE SET ") else {
        return Err(format!("unsupported ON CONFLICT action '{tail}'"));
    };
    let mut rewritten = Vec::new();
    for assignment in assignments.split(", ") {
        let (lhs, rhs) = assignment
            .split_once(" = ")
            .ok_or_else(|| format!("unsupported ON CONFLICT assignment '{assignment}'"))?;
        let column = rhs
            .strip_prefix("EXCLUDED.")
            .ok_or_else(|| format!("unsupported ON CONFLICT assignment source '{rhs}'"))?;
        if column != lhs {
            return Err(format!(
                "unsupported cross-column ON CONFLICT assignment '{assignment}'"
            ));
        }
        rewritten.push(format!("{lhs} = VALUES({column})"));
    }
    Ok(format!(
        "{head} ON DUPLICATE KEY UPDATE {}",
        rewritten.join(", ")
    ))
}

/// MySQL XA participant (NW-deep).
///
/// Implements PHASE 1 (PREPARE) and PHASE 2 (COMMIT / ROLLBACK) by
/// emitting MySQL's `XA START / END / PREPARE / COMMIT / ROLLBACK`
/// SQL via a fresh sqlx connection. MySQL XA semantics:
///
/// - `XA START 'xid'` opens a global transaction.
/// - `XA END 'xid'` marks the transaction's work complete.
/// - `XA PREPARE 'xid'` durably persists the transaction in a prepared
///   state. From this point, `XA COMMIT 'xid'` or `XA ROLLBACK 'xid'`
///   are guaranteed to succeed.
///
/// Caveats handled by this impl:
/// - The actual user statements must be issued between `XA START` and
///   `XA END` on the **same connection**. This participant is built
///   with the statements already buffered (`prepared_statements`) so
///   it can replay them in the PREPARE phase. The buffer is supplied by
///   the runtime when it constructs participants from a saga step list.
/// - sqlx's connection pool may hand out a different connection on
///   each call, breaking XA's single-connection requirement. We hold a
///   dedicated `PoolConnection` for the lifetime of the participant
///   (acquired in PREPARE, released after the PHASE 2 outcome).
#[cfg(feature = "mysql")]
pub struct XaMysqlParticipant {
    handle: XaParticipantHandle,
    pool: sqlx::MySqlPool,
    /// Statements issued between `XA START` and `XA END`. Each is
    /// `(sql, bound_params_json)` where the executor's normal
    /// bind-params path is replayed.
    prepared_statements: Vec<(String, Vec<serde_json::Value>)>,
    /// Connection held across PREPARE → COMMIT/ROLLBACK so XA's
    /// single-connection requirement is honoured. `Mutex` because
    /// `XaParticipant` methods take `&self`.
    connection: tokio::sync::Mutex<Option<sqlx::pool::PoolConnection<sqlx::MySql>>>,
}

#[cfg(feature = "mysql")]
impl XaMysqlParticipant {
    pub fn new(
        instance: impl Into<String>,
        pool: sqlx::MySqlPool,
        prepared_statements: Vec<(String, Vec<serde_json::Value>)>,
    ) -> Self {
        Self {
            handle: XaParticipantHandle::new("mysql", instance),
            pool,
            prepared_statements,
            connection: tokio::sync::Mutex::new(None),
        }
    }

    /// Bind one JSON value into a sqlx MySQL query. Same logic as the
    /// compensator's bind helper but co-located here for the XA path.
    fn bind_xa_value<'a>(
        q: sqlx::query::Query<'a, sqlx::MySql, sqlx::mysql::MySqlArguments>,
        v: &'a serde_json::Value,
    ) -> sqlx::query::Query<'a, sqlx::MySql, sqlx::mysql::MySqlArguments> {
        match v {
            serde_json::Value::Null => q.bind(Option::<&str>::None),
            serde_json::Value::Bool(b) => q.bind(*b),
            serde_json::Value::Number(n) => {
                if let Some(i) = n.as_i64() {
                    q.bind(i)
                } else if let Some(f) = n.as_f64() {
                    q.bind(f)
                } else {
                    q.bind(n.to_string())
                }
            }
            serde_json::Value::String(s) => q.bind(s.as_str()),
            other => q.bind(other.to_string()),
        }
    }
}

#[cfg(feature = "mysql")]
#[async_trait::async_trait]
impl XaParticipant for XaMysqlParticipant {
    fn handle(&self) -> &XaParticipantHandle {
        &self.handle
    }

    async fn prepare(&self, xid: &str) -> PrepareVote {
        // Acquire a dedicated connection from the pool — this connection
        // is held across PHASE 2 to honour MySQL's single-connection XA
        // requirement.
        let mut conn = match self.pool.acquire().await {
            Ok(c) => c,
            Err(err) => {
                return PrepareVote::Aborted {
                    reason: format!("XA mysql acquire connection: {err}"),
                };
            }
        };
        use sqlx::Executor;
        // XA START requires a valid xid format. MySQL accepts
        // single-quoted xid; we pass the raw value from the coordinator
        // (`udb-<uuid>`, no quotes inside).
        if let Err(err) = conn.execute(format!("XA START '{xid}'").as_str()).await {
            return PrepareVote::Aborted {
                reason: format!("XA START failed: {err}"),
            };
        }
        // Replay every buffered statement on this connection.
        for (sql, params) in &self.prepared_statements {
            let mut q = sqlx::query(sql);
            for v in params {
                q = Self::bind_xa_value(q, v);
            }
            if let Err(err) = q.execute(&mut *conn).await {
                // Best-effort cleanup: END + ROLLBACK to clear state on
                // this connection before dropping.
                let _ = conn.execute(format!("XA END '{xid}'").as_str()).await;
                let _ = conn.execute(format!("XA ROLLBACK '{xid}'").as_str()).await;
                return PrepareVote::Aborted {
                    reason: format!("XA statement failed: {err}"),
                };
            }
        }
        if let Err(err) = conn.execute(format!("XA END '{xid}'").as_str()).await {
            let _ = conn.execute(format!("XA ROLLBACK '{xid}'").as_str()).await;
            return PrepareVote::Aborted {
                reason: format!("XA END failed: {err}"),
            };
        }
        if let Err(err) = conn.execute(format!("XA PREPARE '{xid}'").as_str()).await {
            return PrepareVote::Aborted {
                reason: format!("XA PREPARE failed: {err}"),
            };
        }
        // Stash the connection for PHASE 2.
        *self.connection.lock().await = Some(conn);
        PrepareVote::Prepared
    }

    async fn commit_prepared(&self, xid: &str) -> Result<(), String> {
        let mut guard = self.connection.lock().await;
        let mut conn = guard
            .take()
            .ok_or_else(|| "XA commit called without prepared connection".to_string())?;
        use sqlx::Executor;
        conn.execute(format!("XA COMMIT '{xid}'").as_str())
            .await
            .map(|_| ())
            .map_err(|err| format!("XA COMMIT failed: {err}"))
    }

    async fn rollback_prepared(&self, xid: &str) -> Result<(), String> {
        let mut guard = self.connection.lock().await;
        // If no connection was stashed (PREPARE failed earlier), we
        // try a fresh one — MySQL's XA RECOVER + XA ROLLBACK works
        // from any connection on the same server.
        let mut conn = match guard.take() {
            Some(c) => c,
            None => self
                .pool
                .acquire()
                .await
                .map_err(|err| format!("XA mysql rollback acquire: {err}"))?,
        };
        use sqlx::Executor;
        conn.execute(format!("XA ROLLBACK '{xid}'").as_str())
            .await
            .map(|_| ())
            .map_err(|err| format!("XA ROLLBACK failed: {err}"))
    }
}

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

    /// Stub participant for testing. Records every method call so the
    /// test can assert PHASE 1 / PHASE 2 ordering, rollback discipline,
    /// and which participants were touched.
    struct StubParticipant {
        handle: XaParticipantHandle,
        prepare_outcome: PrepareVote,
        commit_outcome: Result<(), String>,
        log: Mutex<Vec<String>>,
    }

    impl StubParticipant {
        fn new(label: &str, prepare: PrepareVote, commit: Result<(), String>) -> Self {
            Self {
                handle: XaParticipantHandle::new("postgres", label),
                prepare_outcome: prepare,
                commit_outcome: commit,
                log: Mutex::new(Vec::new()),
            }
        }
    }

    #[async_trait::async_trait]
    impl XaParticipant for StubParticipant {
        fn handle(&self) -> &XaParticipantHandle {
            &self.handle
        }
        async fn prepare(&self, xid: &str) -> PrepareVote {
            self.log.lock().unwrap().push(format!("prepare({xid})"));
            self.prepare_outcome.clone()
        }
        async fn commit_prepared(&self, xid: &str) -> Result<(), String> {
            self.log
                .lock()
                .unwrap()
                .push(format!("commit_prepared({xid})"));
            self.commit_outcome.clone()
        }
        async fn rollback_prepared(&self, xid: &str) -> Result<(), String> {
            self.log
                .lock()
                .unwrap()
                .push(format!("rollback_prepared({xid})"));
            Ok(())
        }
    }

    fn matrix_with_two_phase(backend: &str, supports: bool) -> BackendCapabilityMatrixEntry {
        BackendCapabilityMatrixEntry {
            backend: backend.to_string(),
            tier: "sql".to_string(),
            operations: vec!["query".to_string(), "mutate".to_string()],
            unsupported_error_code: "failed_precondition".to_string(),
            consistency_model: "strong".to_string(),
            max_payload_bytes: 0,
            supports_xa: supports,
            supports_two_phase_commit: supports,
            transport_label: "test".to_string(),
            live_probe: false,
            configured: false,
            role: crate::backend::BackendRole::Canonical,
            // Struct's documented defaults (this test helper predates the B.12
            // canonical-promotion fields added to the matrix entry).
            canonical_candidate: crate::backend::CanonicalCandidateProfile::ExplicitlyNotSupported,
            canonical_goal: String::new(),
            canonical_feasibility: None,
            control_plane_ha_level: crate::backend::ControlPlaneHaLevel::ProjectionOnly,
        }
    }

    fn request() -> XaRequest {
        XaRequest {
            tenant_id: "tenant-a".to_string(),
            project_id: "billing".to_string(),
            origin_rpc: "BeginTransaction".to_string(),
            correlation_id: "corr-1".to_string(),
        }
    }

    /// Tokens are SDK contract — pinned.
    #[test]
    fn strategy_tokens_are_pinned() {
        assert_eq!(TransactionStrategy::Saga.as_str(), "saga");
        assert_eq!(TransactionStrategy::BestEffort.as_str(), "best_effort");
        assert_eq!(TransactionStrategy::TwoPhase.as_str(), "two_phase");
    }

    #[test]
    fn strategy_parses_aliases_back_to_canonical() {
        assert_eq!(
            TransactionStrategy::parse("2pc"),
            Some(TransactionStrategy::TwoPhase)
        );
        assert_eq!(
            TransactionStrategy::parse("xa"),
            Some(TransactionStrategy::TwoPhase)
        );
        assert_eq!(
            TransactionStrategy::parse("best-effort"),
            Some(TransactionStrategy::BestEffort)
        );
        assert_eq!(TransactionStrategy::parse("unknown"), None);
        assert_eq!(
            TransactionStrategy::parse_or_default(""),
            TransactionStrategy::Saga
        );
    }

    /// **Acceptance gate**: 2PC refused before any side effect when a
    /// participant lacks `supports_two_phase_commit`. The matrix lookup
    /// runs FIRST; no PREPARE happens.
    #[tokio::test]
    async fn capability_refused_emits_no_side_effects() {
        let pg = StubParticipant::new("primary", PrepareVote::Prepared, Ok(()));
        // Mongo participant — capability matrix says no 2PC.
        let mongo_handle = XaParticipantHandle::new("mongodb", "default");
        let mongo = StubParticipant {
            handle: mongo_handle,
            prepare_outcome: PrepareVote::Prepared,
            commit_outcome: Ok(()),
            log: Mutex::new(Vec::new()),
        };
        let matrix = vec![
            matrix_with_two_phase("postgres", true),
            matrix_with_two_phase("mongodb", false),
        ];
        let err = XaCoordinator::execute(request(), vec![Box::new(pg), Box::new(mongo)], &matrix)
            .await
            .unwrap_err();
        assert!(matches!(err, XaError::CapabilityRefused { .. }));
        // Critical: capability check ran BEFORE any participant was
        // touched. The mongo log is empty.
        // (We can't easily inspect the boxed values' interior state
        // here — but the proof is that CapabilityRefused is returned
        // synchronously from validate_participants without ever calling
        // prepare().)
    }

    /// PHASE 1 abort path: any participant voting Aborted → all
    /// previously-prepared participants get rolled back, the ledger
    /// records RolledBack with the abort reason, no participant
    /// committed.
    #[tokio::test]
    async fn prepare_abort_triggers_rollback_on_prepared_participants() {
        let p1 = StubParticipant::new("p1", PrepareVote::Prepared, Ok(()));
        let p2 = StubParticipant::new(
            "p2",
            PrepareVote::Aborted {
                reason: "constraint violation".to_string(),
            },
            Ok(()),
        );
        let p3 = StubParticipant::new("p3", PrepareVote::Prepared, Ok(()));
        let matrix = vec![matrix_with_two_phase("postgres", true)];
        let err = XaCoordinator::execute(
            request(),
            vec![Box::new(p1), Box::new(p2), Box::new(p3)],
            &matrix,
        )
        .await
        .unwrap_err();
        match err {
            XaError::PrepareFailed { ledger, failures } => {
                assert_eq!(ledger.decision, XaDecision::RolledBack);
                assert!(ledger.reason.contains("constraint violation"));
                // p2's vote was Aborted; p1 and p3 were Prepared.
                assert_eq!(failures.len(), 3);
                assert!(matches!(failures[0].vote, PrepareVote::Prepared));
                assert!(matches!(failures[1].vote, PrepareVote::Aborted { .. }));
                assert!(matches!(failures[2].vote, PrepareVote::Prepared));
                // No participant committed.
                assert!(failures.iter().all(|f| !f.committed));
            }
            other => panic!("expected PrepareFailed, got {other:?}"),
        }
    }

    /// Happy path: all participants vote Prepared, all commit, the
    /// ledger records Committed with `is_terminal() = true`.
    #[tokio::test]
    async fn happy_path_commits_every_participant() {
        let p1 = StubParticipant::new("p1", PrepareVote::Prepared, Ok(()));
        let p2 = StubParticipant::new("p2", PrepareVote::Prepared, Ok(()));
        let matrix = vec![matrix_with_two_phase("postgres", true)];
        let outcome = XaCoordinator::execute(request(), vec![Box::new(p1), Box::new(p2)], &matrix)
            .await
            .expect("happy path");
        assert_eq!(outcome.ledger.decision, XaDecision::Committed);
        assert!(outcome.ledger.decision.is_terminal());
        assert_eq!(outcome.participants.len(), 2);
        assert!(outcome.participants.iter().all(|p| p.committed));
        assert_eq!(
            outcome.ledger.participants,
            vec!["postgres:p1", "postgres:p2"]
        );
    }

    /// PHASE 2 commit failure → in-doubt ledger. Recovery worker takes
    /// over; the request returns InDoubt error with the durable xid.
    #[tokio::test]
    async fn commit_failure_lands_in_doubt() {
        let p1 = StubParticipant::new("p1", PrepareVote::Prepared, Ok(()));
        let p2 = StubParticipant::new(
            "p2",
            PrepareVote::Prepared,
            Err("network partition".to_string()),
        );
        let matrix = vec![matrix_with_two_phase("postgres", true)];
        let err = XaCoordinator::execute(request(), vec![Box::new(p1), Box::new(p2)], &matrix)
            .await
            .unwrap_err();
        match err {
            XaError::InDoubt { ledger } => {
                assert_eq!(ledger.decision, XaDecision::InDoubt);
                assert!(!ledger.decision.is_terminal());
                assert!(ledger.reason.contains("network partition"));
                assert!(!ledger.xid.is_empty());
            }
            other => panic!("expected InDoubt, got {other:?}"),
        }
    }

    #[test]
    fn decision_tokens_and_terminality_are_pinned() {
        assert_eq!(XaDecision::Committed.as_str(), "committed");
        assert_eq!(XaDecision::RolledBack.as_str(), "rolled_back");
        assert_eq!(XaDecision::InDoubt.as_str(), "in_doubt");
        assert_eq!(XaDecision::ManualReview.as_str(), "manual_review");
        assert!(XaDecision::Committed.is_terminal());
        assert!(XaDecision::RolledBack.is_terminal());
        assert!(!XaDecision::InDoubt.is_terminal());
        assert!(!XaDecision::ManualReview.is_terminal());
    }

    /// Stub that records events into a SHARED log so cross-participant /
    /// hook ordering can be asserted.
    struct SharedLogParticipant {
        handle: XaParticipantHandle,
        log: std::sync::Arc<Mutex<Vec<String>>>,
    }

    #[async_trait::async_trait]
    impl XaParticipant for SharedLogParticipant {
        fn handle(&self) -> &XaParticipantHandle {
            &self.handle
        }
        async fn prepare(&self, _xid: &str) -> PrepareVote {
            self.log
                .lock()
                .unwrap()
                .push(format!("prepare:{}", self.handle.label));
            PrepareVote::Prepared
        }
        async fn commit_prepared(&self, _xid: &str) -> Result<(), String> {
            self.log
                .lock()
                .unwrap()
                .push(format!("commit:{}", self.handle.label));
            Ok(())
        }
        async fn rollback_prepared(&self, _xid: &str) -> Result<(), String> {
            self.log
                .lock()
                .unwrap()
                .push(format!("rollback:{}", self.handle.label));
            Ok(())
        }
    }

    /// Item 5 acceptance: the write-ahead ledger record is durably issued
    /// BEFORE any participant's PHASE 2 commit.
    #[tokio::test]
    async fn write_ahead_record_precedes_every_phase_2_commit() {
        let log = std::sync::Arc::new(Mutex::new(Vec::<String>::new()));
        let p1 = SharedLogParticipant {
            handle: XaParticipantHandle::new("postgres", "p1"),
            log: log.clone(),
        };
        let p2 = SharedLogParticipant {
            handle: XaParticipantHandle::new("postgres", "p2"),
            log: log.clone(),
        };
        let matrix = vec![matrix_with_two_phase("postgres", true)];
        let hook_log = log.clone();
        let outcome = XaCoordinator::execute_with_write_ahead(
            request(),
            vec![Box::new(p1), Box::new(p2)],
            &matrix,
            |entry| async move {
                assert_eq!(entry.decision, XaDecision::InDoubt);
                assert_eq!(entry.reason, XA_COMMIT_INTENT_REASON);
                hook_log.lock().unwrap().push("write_ahead".to_string());
                Ok(())
            },
        )
        .await
        .expect("happy path");
        assert_eq!(outcome.ledger.decision, XaDecision::Committed);
        let events = log.lock().unwrap().clone();
        let write_ahead_idx = events
            .iter()
            .position(|e| e == "write_ahead")
            .expect("write-ahead hook ran");
        for (idx, event) in events.iter().enumerate() {
            if event.starts_with("commit:") {
                assert!(
                    write_ahead_idx < idx,
                    "write-ahead must precede phase-2 commit {event}; got {events:?}"
                );
            }
            if event.starts_with("prepare:") {
                assert!(
                    idx < write_ahead_idx,
                    "write-ahead must follow all prepares; got {events:?}"
                );
            }
        }
    }

    /// Item 5 fail-closed: when the write-ahead record can't be written,
    /// nothing commits — every prepared participant rolls back.
    #[tokio::test]
    async fn write_ahead_failure_rolls_back_and_never_commits() {
        let log = std::sync::Arc::new(Mutex::new(Vec::<String>::new()));
        let p1 = SharedLogParticipant {
            handle: XaParticipantHandle::new("postgres", "p1"),
            log: log.clone(),
        };
        let matrix = vec![matrix_with_two_phase("postgres", true)];
        let err = XaCoordinator::execute_with_write_ahead(
            request(),
            vec![Box::new(p1)],
            &matrix,
            |_entry| async move { Err("ledger down".to_string()) },
        )
        .await
        .unwrap_err();
        match err {
            XaError::PrepareFailed { ledger, .. } => {
                assert_eq!(ledger.decision, XaDecision::RolledBack);
                assert!(ledger.reason.contains("ledger down"));
            }
            other => panic!("expected PrepareFailed, got {other:?}"),
        }
        let events = log.lock().unwrap().clone();
        assert!(events.iter().any(|e| e.starts_with("rollback:")));
        assert!(events.iter().all(|e| !e.starts_with("commit:")));
    }

    #[test]
    fn commit_intent_reason_is_not_misread_as_prepare_failure() {
        let lc = XA_COMMIT_INTENT_REASON.to_ascii_lowercase();
        for phrase in ["prepare failed", "xa prepare", "aborted vote", "aborted: "] {
            assert!(
                !lc.contains(phrase),
                "commit-intent reason must not contain '{phrase}'"
            );
        }
    }

    #[test]
    fn translate_upsert_plan_to_mysql_dialect() {
        let pg = "INSERT INTO \"public\".\"users\" (\"id\", \"name\") VALUES ($1, $2) \
                  ON CONFLICT (\"id\") DO UPDATE SET \"name\" = EXCLUDED.\"name\"";
        let mysql = translate_pg_plan_sql_to_mysql(pg).expect("translates");
        assert_eq!(
            mysql,
            "INSERT INTO `public`.`users` (`id`, `name`) VALUES (?, ?) \
             ON DUPLICATE KEY UPDATE `name` = VALUES(`name`)"
        );
    }

    #[test]
    fn translate_do_nothing_upsert_to_mysql_noop_update() {
        let pg = "INSERT INTO \"s\".\"t\" (\"id\") VALUES ($1) ON CONFLICT (\"id\") DO NOTHING";
        let mysql = translate_pg_plan_sql_to_mysql(pg).expect("translates");
        assert_eq!(
            mysql,
            "INSERT INTO `s`.`t` (`id`) VALUES (?) ON DUPLICATE KEY UPDATE `id` = `id`"
        );
    }

    #[test]
    fn translate_delete_plan_to_mysql_dialect() {
        let pg = "DELETE FROM \"s\".\"t\" WHERE \"tenant_id\" = $1 AND \"id\" IN ($2, $3)";
        let mysql = translate_pg_plan_sql_to_mysql(pg).expect("translates");
        assert_eq!(
            mysql,
            "DELETE FROM `s`.`t` WHERE `tenant_id` = ? AND `id` IN (?, ?)"
        );
    }

    #[test]
    fn translate_fails_closed_on_untranslatable_shapes() {
        // RETURNING has no MySQL equivalent on this path.
        assert!(
            translate_pg_plan_sql_to_mysql(
                "INSERT INTO \"s\".\"t\" (\"id\") VALUES ($1) ON CONFLICT (\"id\") DO NOTHING RETURNING *"
            )
            .is_err()
        );
        // Out-of-order placeholders would bind the wrong values positionally.
        assert!(
            translate_pg_plan_sql_to_mysql("DELETE FROM \"s\".\"t\" WHERE \"a\" = $2").is_err()
        );
        // Embedded identifier quotes can't round-trip the quote rewrite.
        assert!(
            translate_pg_plan_sql_to_mysql("DELETE FROM \"s\".\"t\"\"x\" WHERE \"a\" = $1")
                .is_err()
        );
        // Pre-existing backticks make the quote rewrite ambiguous.
        assert!(translate_pg_plan_sql_to_mysql("DELETE FROM `t` WHERE \"a\" = $1").is_err());
    }

    #[test]
    fn xid_is_valid_postgres_identifier() {
        let xid = XaCoordinator::new_xid();
        assert!(xid.starts_with("udb-"));
        assert!(
            xid.len() < 200,
            "must fit Postgres prepared-xact name limit"
        );
        assert!(
            xid.chars().all(|c| c.is_ascii_alphanumeric() || c == '-'),
            "xid must be ASCII: {xid}"
        );
    }

    #[test]
    fn ledger_round_trips_through_serde() {
        let entry = XaLedgerEntry::new(
            "udb-test",
            "t1",
            "p1",
            "BeginTransaction",
            "corr",
            vec!["postgres:primary".into(), "postgres:replica".into()],
            XaDecision::InDoubt,
        )
        .with_reason("kafka timeout");
        let json = serde_json::to_string(&entry).unwrap();
        let back: XaLedgerEntry = serde_json::from_str(&json).unwrap();
        assert_eq!(back, entry);
    }
}