From b7e3510553392c8221feef74b7118a8e2f9fe81f Mon Sep 17 00:00:00 2001 From: jwinterm Date: Fri, 10 Jul 2026 15:34:14 -0400 Subject: [PATCH] =?UTF-8?q?feat(tips):=20stage=204=20=E2=80=94=20claim=20/?= =?UTF-8?q?=20clawback=20/=20confirm-sweep=20+=20sweep=20reconciler=20(mon?= =?UTF-8?q?ey-out)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Escrow settlement, audited guards ported verbatim: - mark_tip_claiming: status IN('pending','claiming') AND confs>=required AND funding_amount_verified AND sweep_confirmed_at IS NULL (double-claim guard). - confirm_tip_sweep: RECORD-not-settle, first-recorder-wins (COALESCE + sweep_txid IS NULL OR =$3) — RBF-steal safe; reconciler owns claiming->claimed. - confirm_sweep_onchain: SOLE writer of 'claimed', one-way status='claiming' AND sweep_confirmed_at IS NULL lock; settles only after on-chain sweep confirms to per-asset depth (btc/ltc 1, wow 4, xmr 10). - clawback: 5-status set incl 'cancelled', blocked once sweep_confirmed_at set. - revert_sweep_on_reorg: only from 'claimed', re-opens to 'claiming' + clears txid. - sweep_reconciler: address-history-driven (not recorded txid), pick_winner (earliest height, lexi-smallest tie), pg advisory locks on a pinned conn for single-instance safety; reorg pass reverts only on affirmative vanish, never on probe error. DMs stubbed, exactly-once stamps kept. - POST claim/confirm-sweep/clawback endpoints + openapi. - 16 integration + 12 reconciler unit tests; build clean; SocialTipRow dead_code allow removed (all fields now read). --- src/api/openapi.rs | 7 + src/api/tips.rs | 157 +++++++++ src/infra/db/social_tips.rs | 358 ++++++++++++++++++- src/main.rs | 45 +++ src/tips/mod.rs | 2 + src/tips/sweep_reconciler.rs | 643 +++++++++++++++++++++++++++++++++++ tests/tips.rs | 301 ++++++++++++++++ 7 files changed, 1510 insertions(+), 3 deletions(-) create mode 100644 src/tips/sweep_reconciler.rs diff --git a/src/api/openapi.rs b/src/api/openapi.rs index c24679b..5c08dfe 100644 --- a/src/api/openapi.rs +++ b/src/api/openapi.rs @@ -73,6 +73,9 @@ use utoipa::OpenApi; crate::api::tips::get_public_social_tip, crate::api::tips::cancel_social_tip, crate::api::tips::attach_funding, + crate::api::tips::claim_social_tip, + crate::api::tips::confirm_sweep, + crate::api::tips::clawback_social_tip, ), components(schemas( crate::api::health::HealthResponse, @@ -170,6 +173,10 @@ use utoipa::OpenApi; crate::api::tips::PublicTipInfo, crate::api::tips::CancelTipResponse, crate::api::tips::AttachFundingRequest, + crate::api::tips::ClaimTipResponse, + crate::api::tips::ConfirmSweepRequest, + crate::api::tips::ConfirmSweepResponse, + crate::api::tips::ClawbackTipResponse, )), tags( (name = "system", description = "Service health and metadata."), diff --git a/src/api/tips.rs b/src/api/tips.rs index a8430be..07e2c5b 100644 --- a/src/api/tips.rs +++ b/src/api/tips.rs @@ -125,6 +125,37 @@ pub struct AttachFundingRequest { pub funding_txid: String, } +/// Claim a tip: lock it into `claiming` and receive the encrypted claim key + +/// tip address to sweep. Load-bearing INCONSISTENT key: `success` (not `ok`). +#[derive(Debug, Serialize, utoipa::ToSchema)] +pub struct ClaimTipResponse { + pub success: bool, + /// Hex-encoded AES-GCM ciphertext of the claim secret; `null` if the tip + /// stored none. Useless without the URL-fragment key. + pub encrypted_key: Option, + pub tip_address: Option, +} + +/// Record the claimer's broadcast sweep transaction (first-recorder-wins). +#[derive(Debug, Deserialize, utoipa::ToSchema)] +pub struct ConfirmSweepRequest { + /// The broadcast sweep transaction id. + pub sweep_txid: String, +} + +#[derive(Debug, Serialize, utoipa::ToSchema)] +pub struct ConfirmSweepResponse { + /// The WINNING (first-recorded) sweep txid, or `null` if none is recorded. + pub sweep_txid: Option, + pub status: String, +} + +/// Sender clawback of a tip. Load-bearing INCONSISTENT key: `success` (not `ok`). +#[derive(Debug, Serialize, utoipa::ToSchema)] +pub struct ClawbackTipResponse { + pub success: bool, +} + // ── helpers ────────────────────────────────────────────────────────────────── /// Reject when tips are not enabled on this instance. @@ -476,6 +507,129 @@ pub async fn attach_funding( })) } +/// Claim a public tip: lock it into `claiming` and return the encrypted claim +/// key + tip address so the caller can sweep the funds to their own wallet. +/// +/// AUTHED, no body. The `claiming` state is a UX signal, not a cryptographic +/// lock — any URL holder may claim while the sweep hasn't confirmed (whoever +/// wins the on-chain sweep race wins the tip; the reconciler resolves the +/// winner). A non-claimable tip (missing, underfunded, clawed back, already +/// swept) is a 400. +#[utoipa::path( + security(("bearer_auth" = [])), + post, + path = "/tips/social/{tip_id}/claim", + params(("tip_id" = String, Path, description = "Tip id")), + responses( + (status = 200, description = "Tip locked into claiming", body = ClaimTipResponse), + (status = 400, description = "Tips off or tip not claimable"), + (status = 401, description = "Missing or invalid token") + ), + tag = "tips" +)] +#[instrument(skip(state, headers))] +pub async fn claim_social_tip( + State(state): State>, + headers: HeaderMap, + Path(tip_id): Path, +) -> Result, AppError> { + let user_id = extract_user_id_from_token(&state, &headers).await?; + ensure_tips_enabled(&state)?; + + let row = state + .db + .mark_tip_claiming(tip_id, user_id) + .await? + .ok_or_else(|| AppError::ValidationError("tip not claimable".into()))?; + + Ok(Json(ClaimTipResponse { + success: true, + encrypted_key: row.encrypted_key.map(hex::encode), + tip_address: row.tip_address, + })) +} + +/// Record the claimer's broadcast sweep txid (first-recorder-wins). RECORD not +/// settle: the tip stays `claiming` and the reconciler owns the settle to +/// `claimed` once the sweep confirms on-chain. Idempotent; a second, different +/// txid does NOT overwrite the recorded winner. A tip that was never claimed is +/// a 400. +#[utoipa::path( + security(("bearer_auth" = [])), + post, + path = "/tips/social/{tip_id}/confirm-sweep", + params(("tip_id" = String, Path, description = "Tip id")), + request_body = ConfirmSweepRequest, + responses( + (status = 200, description = "Sweep txid recorded", body = ConfirmSweepResponse), + (status = 400, description = "Tips off, empty/oversized txid, or tip not claiming"), + (status = 401, description = "Missing or invalid token") + ), + tag = "tips" +)] +#[instrument(skip(state, headers, req))] +pub async fn confirm_sweep( + State(state): State>, + headers: HeaderMap, + Path(tip_id): Path, + Json(req): Json, +) -> Result, AppError> { + let user_id = extract_user_id_from_token(&state, &headers).await?; + ensure_tips_enabled(&state)?; + + let sweep_txid = req.sweep_txid.trim(); + if sweep_txid.is_empty() { + return Err(AppError::ValidationError("sweep_txid is required.".into())); + } + if sweep_txid.len() > 128 { + return Err(AppError::ValidationError("sweep_txid is too long.".into())); + } + + let row = state + .db + .confirm_tip_sweep(tip_id, user_id, sweep_txid) + .await? + .ok_or_else(|| AppError::ValidationError("tip not claiming".into()))?; + + Ok(Json(ConfirmSweepResponse { + sweep_txid: row.sweep_txid, + status: row.status, + })) +} + +/// Sender reclaims a tip's funds. AUTHED, no body. Succeeds for the sender on a +/// pending / pending_confirmation / claiming / funding_mismatch / cancelled tip +/// while its sweep hasn't confirmed; a settled or non-clawable tip is a 400. +#[utoipa::path( + security(("bearer_auth" = [])), + post, + path = "/tips/social/{tip_id}/clawback", + params(("tip_id" = String, Path, description = "Tip id")), + responses( + (status = 200, description = "Tip clawed back", body = ClawbackTipResponse), + (status = 400, description = "Tips off or tip not clawable"), + (status = 401, description = "Missing or invalid token") + ), + tag = "tips" +)] +#[instrument(skip(state, headers))] +pub async fn clawback_social_tip( + State(state): State>, + headers: HeaderMap, + Path(tip_id): Path, +) -> Result, AppError> { + let user_id = extract_user_id_from_token(&state, &headers).await?; + ensure_tips_enabled(&state)?; + + state + .db + .clawback_social_tip(tip_id, user_id) + .await? + .ok_or_else(|| AppError::ValidationError("tip not clawable".into()))?; + + Ok(Json(ClawbackTipResponse { success: true })) +} + pub fn routes() -> Router> { Router::new() .route("/tips/social", post(create_social_tip)) @@ -486,4 +640,7 @@ pub fn routes() -> Router> { "/tips/social/:tip_id/attach-funding", post(attach_funding), ) + .route("/tips/social/:tip_id/claim", post(claim_social_tip)) + .route("/tips/social/:tip_id/confirm-sweep", post(confirm_sweep)) + .route("/tips/social/:tip_id/clawback", post(clawback_social_tip)) } diff --git a/src/infra/db/social_tips.rs b/src/infra/db/social_tips.rs index 307caf6..bede64d 100644 --- a/src/infra/db/social_tips.rs +++ b/src/infra/db/social_tips.rs @@ -29,10 +29,8 @@ pub(crate) const TIP_COLS: &str = "\ lws_deactivated_at, created_at, updated_at"; /// A persisted public social tip. Field names/order match [`TIP_COLS`] (sqlx -/// `FromRow` maps by name). Fields not read until later port stages are held -/// behind `allow(dead_code)`. +/// `FromRow` maps by name). #[derive(Debug, Clone, FromRow)] -#[allow(dead_code)] pub struct SocialTipRow { pub id: Uuid, pub sender_user_id: Uuid, @@ -426,4 +424,358 @@ impl Database { .await?; Ok(row) } + + // ── Stage 4: money-out (claim / confirm-sweep / clawback + reconciler) ────── + // + // The HIGHEST money-safety surface. Every method below is a single guarded + // `UPDATE ... WHERE ... RETURNING` and the WHERE guards are non-negotiable — + // a wrong conjunct loses funds. Guards ported VERBATIM from the audited + // legacy `smirk-backend/src/infra/db/social_tips.rs` (public subset: the + // targeted/recipient branches are dropped, `is_public = TRUE` is kept). + + /// Lock a claimable tip into `claiming` and hand the claimer the encrypted + /// key + tip address. THE DOUBLE-CLAIM GUARD — every conjunct is + /// non-negotiable: + /// - `status IN ('pending', 'claiming')` (audit T1): without it a + /// `funding_mismatch` / `clawed_back` row would re-pass the UPDATE + /// because it still satisfies the confirmation/amount/sweep predicates — + /// a URL holder could curl `/claim` on an underfunded tip and sweep the + /// short amount before the sender's clawback wins. + /// - `funding_confirmations >= confirmations_required AND + /// funding_amount_verified = TRUE`: only a fully-funded, amount-verified + /// tip is claimable. + /// - `sweep_confirmed_at IS NULL`: never re-hand the key on a settled tip. + /// - `is_public = TRUE`: public-only port; a public tip's claim state is a + /// UX signal, not a cryptographic lock (whoever wins the on-chain sweep + /// race wins the tip). `claimed_by_user_id` is recorded for DM routing + /// only, carrying no exclusive rights. + /// + /// Returns `None` (not claimable) for a missing / non-claimable / already- + /// swept row. + #[instrument(skip(self))] + pub async fn mark_tip_claiming( + &self, + tip_id: Uuid, + claimed_by_user_id: Uuid, + ) -> Result, AppError> { + let tip = sqlx::query_as::<_, SocialTipRow>(&format!( + "UPDATE social_tips \ + SET status = 'claiming', \ + claimed_at = NOW(), \ + claimed_by_user_id = $2, \ + updated_at = NOW() \ + WHERE id = $1 \ + AND status IN ('pending', 'claiming') \ + AND funding_confirmations >= confirmations_required \ + AND funding_amount_verified = TRUE \ + AND sweep_confirmed_at IS NULL \ + AND is_public = TRUE \ + RETURNING {TIP_COLS}" + )) + .bind(tip_id) + .bind(claimed_by_user_id) + .fetch_optional(self.pool()) + .await?; + Ok(tip) + } + + /// RECORD (not settle) the claimer's broadcast sweep txid, leaving the row in + /// `claiming`. The reconciler — never this call — owns `claiming -> claimed`, + /// and only once the sweep CONFIRMS on-chain. This closes the RBF-steal + /// false-"claimed" vector: a URL holder who RBFs the recorded sweep can't + /// trigger a premature settlement. + /// + /// First-recorder-wins + idempotent: the guard + /// `(sweep_txid IS NULL OR sweep_txid = $3)` records the first txid and makes + /// a second call with the SAME txid a no-op; a second call with a DIFFERENT + /// txid does NOT overwrite (the UPDATE matches nothing) — junk-txid poisoning + /// stays blocked. In that case the fallthrough SELECT returns the current + /// row (with the WINNING first txid) so the caller renders live state; both + /// the first and the different-second call therefore return `Some`. + /// + /// Returns `None` only when the tip doesn't exist or isn't in + /// `claiming`/`claimed` (e.g. `/confirm-sweep` on a never-claimed tip). + #[instrument(skip(self))] + pub async fn confirm_tip_sweep( + &self, + tip_id: Uuid, + claimer_user_id: Uuid, + sweep_txid: &str, + ) -> Result, AppError> { + let recorded = sqlx::query_as::<_, SocialTipRow>(&format!( + "UPDATE social_tips \ + SET sweep_txid = COALESCE(sweep_txid, $3), \ + claimed_by_user_id = COALESCE(claimed_by_user_id, $2), \ + claimed_at = COALESCE(claimed_at, NOW()), \ + updated_at = NOW() \ + WHERE id = $1 \ + AND status = 'claiming' \ + AND sweep_confirmed_at IS NULL \ + AND is_public = TRUE \ + AND (sweep_txid IS NULL OR sweep_txid = $3) \ + RETURNING {TIP_COLS}" + )) + .bind(tip_id) + .bind(claimer_user_id) + .bind(sweep_txid) + .fetch_optional(self.pool()) + .await?; + + if let Some(tip) = recorded { + return Ok(Some(tip)); + } + + // Did not record (already settled, a DIFFERENT txid already recorded, or + // not a claiming/claimed row): return the current row so the caller can + // render live state — first-recorder's txid wins. + let existing = sqlx::query_as::<_, SocialTipRow>(&format!( + "SELECT {TIP_COLS} FROM social_tips \ + WHERE id = $1 AND status IN ('claiming', 'claimed')" + )) + .bind(tip_id) + .fetch_optional(self.pool()) + .await?; + Ok(existing) + } + + /// Sender reclaims the funds. Guard ported verbatim (evolved across two + /// audits): + /// - `sender_user_id = $2`: only the sender can claw back. + /// - `status IN ('pending', 'pending_confirmation', 'claiming', + /// 'funding_mismatch', 'cancelled')`: the 5-status set lets a sender + /// recover a tip racing a claimer (`claiming`), an underfunded tip + /// (`funding_mismatch`), and a GC-cancelled-but-still-funded draft + /// (`cancelled`). + /// - `sweep_confirmed_at IS NULL`: never roll back a settled `claimed` row. + /// + /// Returns `None` when no such clawable row (wrong owner, wrong status, or + /// already settled). + #[instrument(skip(self))] + pub async fn clawback_social_tip( + &self, + tip_id: Uuid, + sender_user_id: Uuid, + ) -> Result, AppError> { + let tip = sqlx::query_as::<_, SocialTipRow>(&format!( + "UPDATE social_tips \ + SET status = 'clawed_back', \ + clawed_back_at = NOW(), \ + updated_at = NOW() \ + WHERE id = $1 \ + AND sender_user_id = $2 \ + AND status IN ( \ + 'pending', \ + 'pending_confirmation', \ + 'claiming', \ + 'funding_mismatch', \ + 'cancelled' \ + ) \ + AND sweep_confirmed_at IS NULL \ + RETURNING {TIP_COLS}" + )) + .bind(tip_id) + .bind(sender_user_id) + .fetch_optional(self.pool()) + .await?; + Ok(tip) + } + + /// The SOLE new writer of `status = 'claimed'`. Settle a tip on-chain after + /// its sweep confirmed to the per-asset depth. + /// + /// One-way lock: the `status = 'claiming' AND sweep_confirmed_at IS NULL` + /// guard means the FIRST reconciler tick to observe the on-chain winner wins + /// the transition; any later tick (or an instance that lost the advisory-lock + /// race) gets `None` and fires no side effects. + /// + /// `claimed_by`: `Some` overwrites `claimed_by_user_id`; `None` PRESERVES it + /// (the `CASE WHEN $5 IS NOT NULL` guard). This method can only set-or-keep, + /// never clear — to NULL the attribution use `clear_claimed_by_on_settle`. + #[instrument(skip(self))] + pub async fn confirm_sweep_onchain( + &self, + tip_id: Uuid, + winner_txid: &str, + block_height: i32, + block_hash: Option<&str>, + claimed_by: Option, + ) -> Result, AppError> { + let tip = sqlx::query_as::<_, SocialTipRow>(&format!( + "UPDATE social_tips \ + SET status = 'claimed', \ + sweep_confirmed_at = NOW(), \ + sweep_txid = $2, \ + sweep_block_height = $3, \ + sweep_block_hash = $4, \ + claimed_by_user_id = CASE \ + WHEN $5::uuid IS NOT NULL THEN $5 \ + ELSE claimed_by_user_id END, \ + updated_at = NOW() \ + WHERE id = $1 \ + AND status = 'claiming' \ + AND sweep_confirmed_at IS NULL \ + RETURNING {TIP_COLS}" + )) + .bind(tip_id) + .bind(winner_txid) + .bind(block_height) + .bind(block_hash) + .bind(claimed_by) + .fetch_optional(self.pool()) + .await?; + Ok(tip) + } + + /// Settle to `claimed` AND NULL `claimed_by_user_id`. Used when the on-chain + /// winner's txid doesn't match the recorded `sweep_txid` (RBF / unknown + /// sweeper) or none was recorded — a public tip swept by a URL holder who + /// isn't the recorded claimer. Same one-way `claiming -> claimed` lock as + /// `confirm_sweep_onchain`; `None` if already settled. + #[instrument(skip(self))] + pub async fn clear_claimed_by_on_settle( + &self, + tip_id: Uuid, + winner_txid: &str, + block_height: i32, + block_hash: Option<&str>, + ) -> Result, AppError> { + let tip = sqlx::query_as::<_, SocialTipRow>(&format!( + "UPDATE social_tips \ + SET status = 'claimed', \ + sweep_confirmed_at = NOW(), \ + sweep_txid = $2, \ + sweep_block_height = $3, \ + sweep_block_hash = $4, \ + claimed_by_user_id = NULL, \ + updated_at = NOW() \ + WHERE id = $1 \ + AND status = 'claiming' \ + AND sweep_confirmed_at IS NULL \ + RETURNING {TIP_COLS}" + )) + .bind(tip_id) + .bind(winner_txid) + .bind(block_height) + .bind(block_hash) + .fetch_optional(self.pool()) + .await?; + Ok(tip) + } + + /// Revert a settled (`claimed`) tip back to `claiming` after a reorg orphaned + /// its recorded sweep. Clearing `sweep_txid` is LOAD-BEARING: the next + /// confirm cycle re-scans the address and records the NEW on-chain winner, + /// not the orphaned one. Also clears `sweep_confirmed_dm_sent_at` so a fresh + /// "claimed" notify can fire once re-confirmation lands. The + /// `status = 'claimed' AND sweep_confirmed_at IS NOT NULL` guard is the only + /// path off `claimed`; `None` if the row isn't settled. + #[instrument(skip(self))] + pub async fn revert_sweep_on_reorg( + &self, + tip_id: Uuid, + ) -> Result, AppError> { + let tip = sqlx::query_as::<_, SocialTipRow>(&format!( + "UPDATE social_tips \ + SET status = 'claiming', \ + sweep_confirmed_at = NULL, \ + sweep_txid = NULL, \ + sweep_block_height = NULL, \ + sweep_block_hash = NULL, \ + sweep_confirmed_dm_sent_at = NULL, \ + updated_at = NOW() \ + WHERE id = $1 \ + AND status = 'claimed' \ + AND sweep_confirmed_at IS NOT NULL \ + RETURNING {TIP_COLS}" + )) + .bind(tip_id) + .fetch_optional(self.pool()) + .await?; + Ok(tip) + } + + /// Reconciler poll: `claiming` tips whose sweep hasn't confirmed yet. Includes + /// rows with `sweep_txid IS NULL` — the reconciler scans the ADDRESS, not any + /// recorded txid, since a public tip's recorded sweep may not be the on-chain + /// winner (RBF / race). `funding_amount_verified = TRUE` keeps a + /// `funding_mismatch`-adjacent row out. Rate-limited on `last_confirmation_check` + /// (30s) so an outage doesn't burn one probe per row per cycle. + #[instrument(skip(self))] + pub async fn get_tips_awaiting_sweep_confirmation( + &self, + limit: i64, + ) -> Result, AppError> { + let tips = sqlx::query_as::<_, SocialTipRow>(&format!( + "SELECT {TIP_COLS} FROM social_tips \ + WHERE status = 'claiming' \ + AND sweep_confirmed_at IS NULL \ + AND funding_amount_verified = TRUE \ + AND (last_confirmation_check IS NULL \ + OR last_confirmation_check < NOW() - INTERVAL '30 seconds') \ + ORDER BY claimed_at ASC \ + LIMIT $1" + )) + .bind(limit) + .fetch_all(self.pool()) + .await?; + Ok(tips) + } + + /// Reorg-check poll: settled tips the reconciler settled ITSELF (they carry a + /// `sweep_block_height` witness) and hasn't re-probed in the last 30s. Rows + /// with `sweep_block_height IS NULL` are excluded on purpose (no recorded + /// height to reorg-check against). + #[instrument(skip(self))] + pub async fn get_settled_tips_for_reorg_check( + &self, + limit: i64, + ) -> Result, AppError> { + let tips = sqlx::query_as::<_, SocialTipRow>(&format!( + "SELECT {TIP_COLS} FROM social_tips \ + WHERE status = 'claimed' \ + AND sweep_confirmed_at IS NOT NULL \ + AND sweep_block_height IS NOT NULL \ + AND (last_confirmation_check IS NULL \ + OR last_confirmation_check < NOW() - INTERVAL '30 seconds') \ + ORDER BY sweep_confirmed_at ASC \ + LIMIT $1" + )) + .bind(limit) + .fetch_all(self.pool()) + .await?; + Ok(tips) + } + + /// Exactly-once stamp for the reconciler-owned "tip claimed" notify. The + /// `WHERE ... IS NULL` guard makes it idempotent so a retried cycle never + /// re-stamps. (The notify itself is stubbed to a no-op in this port, but the + /// stamp is preserved so re-enabling notifications stays exactly-once.) + #[instrument(skip(self))] + pub async fn stamp_sweep_confirmed_dm_sent(&self, tip_id: Uuid) -> Result<(), AppError> { + sqlx::query( + "UPDATE social_tips \ + SET sweep_confirmed_dm_sent_at = NOW(), updated_at = NOW() \ + WHERE id = $1 AND sweep_confirmed_dm_sent_at IS NULL", + ) + .bind(tip_id) + .execute(self.pool()) + .await?; + Ok(()) + } + + /// Exactly-once stamp for the reconciler-owned "claim reversed by reorg" + /// notify. Same `WHERE ... IS NULL` idempotency as + /// `stamp_sweep_confirmed_dm_sent`. + #[instrument(skip(self))] + pub async fn stamp_reorg_notified(&self, tip_id: Uuid) -> Result<(), AppError> { + sqlx::query( + "UPDATE social_tips \ + SET reorg_notified_at = NOW(), updated_at = NOW() \ + WHERE id = $1 AND reorg_notified_at IS NULL", + ) + .bind(tip_id) + .execute(self.pool()) + .await?; + Ok(()) + } } diff --git a/src/main.rs b/src/main.rs index 07ecdb2..e780f11 100644 --- a/src/main.rs +++ b/src/main.rs @@ -285,6 +285,51 @@ async fn main() -> Result<(), Box> { }); } + // Public-tips "money-out" reconciler: the settle-on-sweep-confirmation pass + // (probe each claiming tip's ADDRESS, settle claiming -> claimed once the + // sweep confirms to the per-asset depth). Sole new writer of 'claimed'. Runs + // an immediate startup pass (so a restart doesn't wait a full interval before + // settling anything that confirmed while we were down) then every 60s. Same + // catch_unwind guard as the money-in worker. + if state.config.features.tips { + use futures::FutureExt; + let state = state.clone(); + tokio::spawn(async move { + let mut tick = tokio::time::interval(std::time::Duration::from_secs(60)); + loop { + tick.tick().await; + let cycle = std::panic::AssertUnwindSafe( + smirk_backend_core::tips::run_sweep_reconcile_cycle(state.clone()), + ); + if cycle.catch_unwind().await.is_err() { + tracing::error!("sweep reconcile cycle panicked; continuing after backoff"); + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + } + } + }); + } + + // Public-tips reorg-check pass: re-probe settled tips and revert any whose + // recorded sweep was orphaned by a chain reorg. Lowest-frequency, novel, and + // recoverable, so it runs on a slower cadence (300s) than the confirm pass. + if state.config.features.tips { + use futures::FutureExt; + let state = state.clone(); + tokio::spawn(async move { + let mut tick = tokio::time::interval(std::time::Duration::from_secs(300)); + loop { + tick.tick().await; + let cycle = std::panic::AssertUnwindSafe( + smirk_backend_core::tips::run_sweep_reorg_cycle(state.clone()), + ); + if cycle.catch_unwind().await.is_err() { + tracing::error!("sweep reorg cycle panicked; continuing after backoff"); + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + } + } + }); + } + let app = build_router(state); let listener = tokio::net::TcpListener::bind(&addr).await?; diff --git a/src/tips/mod.rs b/src/tips/mod.rs index b20accf..9f11311 100644 --- a/src/tips/mod.rs +++ b/src/tips/mod.rs @@ -16,5 +16,7 @@ //! outage can therefore never mass-flip pending tips to `funding_mismatch`. pub mod confirmation; +pub mod sweep_reconciler; pub use confirmation::run_tip_confirmation_cycle; +pub use sweep_reconciler::{run_sweep_reconcile_cycle, run_sweep_reorg_cycle}; diff --git a/src/tips/sweep_reconciler.rs b/src/tips/sweep_reconciler.rs new file mode 100644 index 0000000..a157b8b --- /dev/null +++ b/src/tips/sweep_reconciler.rs @@ -0,0 +1,643 @@ +//! Settle-on-sweep-confirmation reconciler for public social tips (the +//! "money-out" path). Ported (public subset) from the legacy +//! `smirk-backend/src/core/sweep_reconciler.rs`. +//! +//! Watches `claiming` tips and settles them to `claimed` ONLY after the sweep +//! transaction confirms on-chain to the per-asset depth. It is the sole new +//! writer of `status = 'claimed'` (via [`Database::confirm_sweep_onchain`] / +//! [`Database::clear_claimed_by_on_settle`]) and the sole writer of the reorg +//! revert ([`Database::revert_sweep_on_reorg`]). +//! +//! ## Source of truth = the ADDRESS, not the recorded `sweep_txid` +//! +//! Public tips carry the spend key in the URL fragment, so multiple claimers can +//! race; only one sweep wins on-chain (and a claimer may RBF-replace its own +//! sweep). So the source of truth is the tip ADDRESS's confirmed spend history: +//! we probe the address, collect confirmed spends, and pick the deterministic +//! winner ([`pick_winner`]). +//! +//! ## Confirmation thresholds (mirror the funding thresholds) +//! BTC/LTC 1, WOW 4, XMR 10. Grin is OUT of this port. +//! +//! ## Money-safety invariants +//! - A probe ERROR (Electrum/LWS unreachable, hostile amount) NEVER settles or +//! reverts — it only stamps `last_confirmation_check` and retries. +//! - Settle requires an affirmative confirmed spend at the required depth. +//! - Reorg-revert requires an affirmative "the recorded sweep is GONE from +//! confirmed history"; a probe error can never trigger it. +//! +//! ## Notifications +//! The sender-DM side-effects are STUBBED to no-ops in this port (public tips +//! deliver by share URL), but the exactly-once stamp writes +//! (`stamp_sweep_confirmed_dm_sent` / `stamp_reorg_notified`) are preserved so +//! re-enabling notifications later stays exactly-once. + +use std::sync::Arc; + +use tracing::{debug, info, instrument, warn}; + +use crate::error::AppError; +use crate::infra::db::SocialTipRow; +use crate::infra::electrum::VerboseTransaction; +use crate::AppState; + +/// Max rows processed per cycle (per pass). +const BATCH_SIZE: i64 = 50; + +/// Advisory-lock key for the confirm/settle cycle. A fixed, arbitrary 64-bit +/// constant — single-instance guard so two replicas don't both probe + settle +/// the same batch. Distinct from the reorg key. +const SWEEP_CONFIRM_ADVISORY_LOCK_KEY: i64 = 0x53EE_D0C1; +/// Advisory-lock key for the reorg-check cycle (distinct from the confirm cycle +/// so the two passes run independently). +const SWEEP_REORG_ADVISORY_LOCK_KEY: i64 = 0x53EE_D0C2; + +/// A confirmed spend of the tip address, distilled to what winner-determination +/// needs. Pure data, no I/O. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ConfirmedSpend { + /// On-chain identifier of the spend (BTC/LTC + XMR/WOW tx hash). + pub txid: String, + /// Block height the spend confirmed at. + pub block_height: i64, + /// Block hash at that height when trivially available (BTC/LTC don't fetch + /// the merkle proof, so usually `None`). + pub block_hash: Option, +} + +/// Per-asset sweep confirmation threshold (confirmations the winning spend needs +/// before we settle to `claimed`). Mirrors the funding-side thresholds. Unknown +/// assets default to the most conservative (10) so a misconfigured asset never +/// settles early. Grin is not supported by this port. +pub(crate) fn sweep_confirm_threshold(asset: &str) -> u64 { + match asset.to_lowercase().as_str() { + "btc" | "ltc" => 1, + "wow" => 4, + "xmr" => 10, + _ => 10, + } +} + +/// Pick the canonical winning spend: EARLIEST `block_height`, ties broken by the +/// lexicographically smallest `txid`. Deterministic + idempotent, so a re-probed +/// address always settles to the same winner. `None` for an empty slice. +pub(crate) fn pick_winner(spends: &[ConfirmedSpend]) -> Option<&ConfirmedSpend> { + spends.iter().min_by(|a, b| { + a.block_height + .cmp(&b.block_height) + .then_with(|| a.txid.cmp(&b.txid)) + }) +} + +/// Is this verbose BTC/LTC tx a SPEND of `address`? A single-use tip address is +/// a pure RECIPIENT in its funding tx and a pure SPENDER only in its sweep, so a +/// tx is a spend of the address iff the address received nothing in it +/// (`total_received_at(address) == 0`). Same `recv == 0` isolation the funding +/// verifier uses to avoid misreading a funding tx's change output as a send. +/// +/// Returns `Err` when core's Result-returning accessor rejects a hostile / +/// non-finite amount — the caller bails the probe (never settles) on that. +pub(crate) fn is_verbose_tx_spend_of( + tx: &VerboseTransaction, + address: &str, +) -> Result { + Ok(tx.total_received_at(address)? == 0) +} + +// ============================================================================ +// Advisory-lock single-instance guard +// ============================================================================ + +/// Acquire a session-scoped Postgres advisory lock on a PINNED connection. pg +/// advisory locks are connection(session)-scoped, so the lock MUST be acquired +/// and released on the SAME connection — releasing on a different pooled +/// connection leaks the lock. Returns the held connection on success (the caller +/// keeps it alive for the cycle, then passes it to `release_advisory_lock`), or +/// `None` if another holder has it (skip the cycle). +async fn acquire_advisory_lock( + state: &AppState, + key: i64, +) -> Result>, AppError> { + let mut conn = state.db.pool().acquire().await?; + let acquired: bool = sqlx::query_scalar("SELECT pg_try_advisory_lock($1)") + .bind(key) + .fetch_one(&mut *conn) + .await?; + if acquired { + Ok(Some(conn)) + } else { + Ok(None) + } +} + +/// Release the advisory lock on the SAME connection it was acquired on, then +/// drop the connection back to the pool. Best-effort — the lock auto-releases +/// when the connection closes. +async fn release_advisory_lock( + mut conn: sqlx::pool::PoolConnection, + key: i64, +) { + if let Err(e) = sqlx::query("SELECT pg_advisory_unlock($1)") + .bind(key) + .execute(&mut *conn) + .await + { + debug!(error = %e, key, "pg_advisory_unlock failed (lock auto-releases on conn close)"); + } +} + +// ============================================================================ +// Confirm cycle +// ============================================================================ + +/// One confirm/settle cycle: probe each awaiting tip's address, settle the +/// winner once it reaches the per-asset threshold. Guarded by a Postgres +/// advisory lock so only one instance processes a batch. Never returns an +/// error to the caller — a single tip's probe/settle failure is logged and +/// swallowed so it can't tank the batch (or the spawn loop). +pub async fn run_sweep_reconcile_cycle(state: Arc) { + if let Err(e) = run_confirm_cycle(&state).await { + warn!(error = %e, "sweep confirm cycle failed"); + } +} + +#[instrument(skip(state))] +async fn run_confirm_cycle(state: &AppState) -> Result<(), AppError> { + let Some(lock_conn) = acquire_advisory_lock(state, SWEEP_CONFIRM_ADVISORY_LOCK_KEY).await? else { + debug!("sweep confirm cycle: advisory lock held by another instance — skipping"); + return Ok(()); + }; + + let result = confirm_cycle_inner(state).await; + + release_advisory_lock(lock_conn, SWEEP_CONFIRM_ADVISORY_LOCK_KEY).await; + result +} + +async fn confirm_cycle_inner(state: &AppState) -> Result<(), AppError> { + let tips = state.db.get_tips_awaiting_sweep_confirmation(BATCH_SIZE).await?; + if tips.is_empty() { + debug!("no tips awaiting sweep confirmation"); + return Ok(()); + } + info!(count = tips.len(), "reconciling tip sweeps"); + + for tip in &tips { + if let Err(e) = process_awaiting_tip(state, tip).await { + // A single tip's probe/settle error must not tank the batch. + warn!(tip_id = %tip.id, error = %e, "process_awaiting_tip failed"); + } + } + Ok(()) +} + +/// Process a single `claiming` tip: probe its address, pick the winner, require +/// the per-asset depth, settle, stamp the (stubbed) notify. +async fn process_awaiting_tip(state: &AppState, tip: &SocialTipRow) -> Result<(), AppError> { + let asset = tip.asset.to_lowercase(); + + // Probe the address for confirmed spends + the current chain tip. + let (spends, tip_height) = match probe_confirmed_spends(state, tip).await { + Ok(v) => v, + Err(e) => { + // Probe error (chain unreachable / disabled / hostile response): + // NEVER settle. Stamp last_check so the rate-limit backs off, skip. + warn!(tip_id = %tip.id, %asset, error = %e, "sweep probe failed — skipping (retry next cycle)"); + if let Err(e2) = state.db.touch_tip_last_check(tip.id).await { + debug!(tip_id = %tip.id, error = %e2, "touch_tip_last_check failed after probe error"); + } + return Ok(()); + } + }; + + let winner = match pick_winner(&spends) { + Some(w) => w.clone(), + None => { + // No confirmed spend of the address yet — sweep not mined, or still + // in mempool. Back off via last_check and retry. + debug!(tip_id = %tip.id, %asset, "no confirmed sweep yet"); + if let Err(e) = state.db.touch_tip_last_check(tip.id).await { + debug!(tip_id = %tip.id, error = %e, "touch_tip_last_check failed (no winner)"); + } + return Ok(()); + } + }; + + // Threshold gate: confirmations = tip_height - winner_height + 1. + let confirmations = tip_height + .saturating_sub(winner.block_height) + .saturating_add(1); + let needed = sweep_confirm_threshold(&asset); + if (confirmations as u64) < needed { + debug!(tip_id = %tip.id, %asset, confirmations, needed, "sweep seen on-chain, waiting for depth"); + if let Err(e) = state.db.touch_tip_last_check(tip.id).await { + debug!(tip_id = %tip.id, error = %e, "touch_tip_last_check failed (below threshold)"); + } + return Ok(()); + } + + // Attribution (public-only): if the on-chain winner matches the recorded + // sweep_txid, keep claimed_by (via confirm_sweep_onchain with claimed_by = + // None). Otherwise the tip was swept by a URL holder who isn't the recorded + // claimer (RBF / unknown sweeper / no recorded txid) — settle the funds-moved + // fact but drop the now-unreliable attribution. + let recorded = tip.sweep_txid.as_deref(); + let txid_matches = recorded == Some(winner.txid.as_str()); + let block_hash = winner.block_hash.as_deref(); + + let settled = if txid_matches { + state + .db + .confirm_sweep_onchain(tip.id, &winner.txid, winner.block_height as i32, block_hash, None) + .await? + } else { + debug!( + tip_id = %tip.id, %asset, + recorded_sweep_txid = ?recorded, onchain_winner_txid = %winner.txid, + "public tip on-chain winner differs from recorded sweep — clearing claimed_by" + ); + state + .db + .clear_claimed_by_on_settle(tip.id, &winner.txid, winner.block_height as i32, block_hash) + .await? + }; + + let settled = match settled { + Some(row) => row, + None => { + // Another tick (or instance) already settled this row — side effects + // belong to whoever won the UPDATE. + debug!(tip_id = %tip.id, "sweep already settled by another tick — skipping side effects"); + return Ok(()); + } + }; + + info!( + tip_id = %settled.id, %asset, + winner_txid = %winner.txid, block_height = winner.block_height, confirmations, + "tip sweep confirmed on-chain — settled to claimed" + ); + + // Notify side-effect is stubbed to a no-op in this port; keep the + // exactly-once stamp so re-enabling it later stays exactly-once. + send_sweep_confirmed_dm_if_needed(state, &settled).await; + + Ok(()) +} + +// ============================================================================ +// Per-asset on-chain spend probe (source of truth = the ADDRESS) +// ============================================================================ + +/// Probe the tip address for confirmed spends and the current chain tip height. +/// Returns `(spends, tip_height)`. Any upstream error / disabled chain / missing +/// field is `Err` — the caller treats that as "skip this row, retry next cycle" +/// (never settle / revert on an error). +async fn probe_confirmed_spends( + state: &AppState, + tip: &SocialTipRow, +) -> Result<(Vec, i64), AppError> { + let asset = tip.asset.to_lowercase(); + match asset.as_str() { + "btc" | "ltc" => probe_electrum_spends(state, tip, &asset).await, + "xmr" | "wow" => probe_lws_spends(state, tip, &asset).await, + other => { + // Grin is out of this port; only btc/ltc/xmr/wow ever create a tip. + warn!(tip_id = %tip.id, asset = %other, "sweep probe: unsupported asset"); + Err(AppError::ValidationError(format!( + "sweep probe: unsupported asset {other}" + ))) + } + } +} + +/// BTC/LTC probe via Electrum. For each confirmed history entry (`height > 0`) +/// fetch the verbose tx and treat it as a SPEND iff the address received nothing +/// in it. `block_hash` stays `None` (no merkle fetch); reorg detection keys on +/// the txid vanishing from confirmed history, not the hash. +async fn probe_electrum_spends( + state: &AppState, + tip: &SocialTipRow, + asset: &str, +) -> Result<(Vec, i64), AppError> { + let electrum = if asset == "btc" { + state.chains.btc.as_ref() + } else { + state.chains.ltc.as_ref() + } + .ok_or_else(|| AppError::NodeError(format!("{asset} Electrum client disabled")))?; + + let address = tip + .tip_address + .as_deref() + .ok_or_else(|| AppError::ValidationError("tip_address missing for Electrum probe".into()))?; + + let history = electrum.get_history(address).await?; + // Chain height comes from ELECTRUM, not a local node (the backend runs none). + let tip_height = electrum.get_tip_height().await?; + + let mut spends = Vec::new(); + for entry in history.iter().filter(|e| e.height > 0) { + let tx = electrum.get_transaction_verbose(&entry.tx_hash).await?; + if is_verbose_tx_spend_of(&tx, address)? { + spends.push(ConfirmedSpend { + txid: entry.tx_hash.clone(), + block_height: entry.height, + block_hash: None, + }); + } + } + Ok((spends, tip_height)) +} + +/// XMR/WOW probe via LWS `get_address_txs`. A SPEND is a confirmed (`height > 0`, +/// not mempool) tx with `total_sent > 0 && total_received == 0` (the single-use +/// tip address only ever spends in its sweep). `tip_height` comes from the LWS +/// response's `blockchain_height`, falling back to the node height if it's 0. +async fn probe_lws_spends( + state: &AppState, + tip: &SocialTipRow, + asset: &str, +) -> Result<(Vec, i64), AppError> { + let lws = if asset == "xmr" { + state.chains.xmr.as_ref() + } else { + state.chains.wow.as_ref() + } + .ok_or_else(|| AppError::NodeError(format!("{asset} LWS client disabled")))?; + + let address = tip + .tip_address + .as_deref() + .ok_or_else(|| AppError::ValidationError("tip_address missing for LWS probe".into()))?; + let view_key = tip + .tip_view_key + .as_deref() + .ok_or_else(|| AppError::ValidationError("tip_view_key missing for LWS probe".into()))?; + + let txs = lws.get_address_txs(address, view_key).await?; + + let tip_height = if txs.blockchain_height > 0 { + txs.blockchain_height as i64 + } else { + lws.get_blockchain_height().await? as i64 + }; + + let spends: Vec = txs + .transactions + .iter() + .filter(|t| !t.mempool && t.height > 0 && t.total_sent > 0 && t.total_received == 0) + .map(|t| ConfirmedSpend { + txid: t.hash.clone(), + block_height: t.height as i64, + block_hash: None, + }) + .collect(); + + Ok((spends, tip_height)) +} + +// ============================================================================ +// Reorg cycle +// ============================================================================ + +/// One reorg-check cycle: re-probe settled tips; if a recorded sweep is no longer +/// a confirmed spend of the address, revert to `claiming`. Guarded by its own +/// advisory lock. Runs on a slower cadence than the confirm cycle (the reorg +/// path is the lowest-frequency one and reverting is recoverable). +pub async fn run_sweep_reorg_cycle(state: Arc) { + if let Err(e) = run_reorg_cycle(&state).await { + warn!(error = %e, "sweep reorg cycle failed"); + } +} + +#[instrument(skip(state))] +async fn run_reorg_cycle(state: &AppState) -> Result<(), AppError> { + let Some(lock_conn) = acquire_advisory_lock(state, SWEEP_REORG_ADVISORY_LOCK_KEY).await? else { + debug!("sweep reorg cycle: advisory lock held by another instance — skipping"); + return Ok(()); + }; + + let result = reorg_cycle_inner(state).await; + + release_advisory_lock(lock_conn, SWEEP_REORG_ADVISORY_LOCK_KEY).await; + result +} + +async fn reorg_cycle_inner(state: &AppState) -> Result<(), AppError> { + let tips = state.db.get_settled_tips_for_reorg_check(BATCH_SIZE).await?; + if tips.is_empty() { + debug!("no settled tips to reorg-check"); + return Ok(()); + } + debug!(count = tips.len(), "reorg-checking settled tip sweeps"); + + for tip in &tips { + if let Err(e) = process_settled_tip(state, tip).await { + warn!(tip_id = %tip.id, error = %e, "process_settled_tip failed"); + } + } + Ok(()) +} + +/// Re-probe a settled tip. If its recorded `sweep_txid` is still a confirmed +/// spend of the address, do nothing (stamp). If it VANISHED (orphaned by a +/// reorg), revert to `claiming`. A probe ERROR never reverts. +async fn process_settled_tip(state: &AppState, tip: &SocialTipRow) -> Result<(), AppError> { + let recorded_txid = match tip.sweep_txid.as_deref() { + Some(t) => t, + None => { + // Settled with no recorded txid — nothing to reorg-check against. + if let Err(e) = state.db.touch_tip_last_check(tip.id).await { + debug!(tip_id = %tip.id, error = %e, "touch_tip_last_check failed (no recorded txid)"); + } + return Ok(()); + } + }; + + let (spends, _tip_height) = match probe_confirmed_spends(state, tip).await { + Ok(v) => v, + Err(e) => { + // NEVER revert on a probe error. Back off. + warn!(tip_id = %tip.id, error = %e, "reorg re-probe failed — skipping (will retry)"); + if let Err(e2) = state.db.touch_tip_last_check(tip.id).await { + debug!(tip_id = %tip.id, error = %e2, "touch_tip_last_check failed after reorg probe error"); + } + return Ok(()); + } + }; + + // Still confirmed iff the recorded txid appears among the confirmed spends. + // Absence = orphaned by a reorg. + let still_confirmed = spends.iter().any(|s| s.txid == recorded_txid); + if still_confirmed { + if let Err(e) = state.db.touch_tip_last_check(tip.id).await { + debug!(tip_id = %tip.id, error = %e, "touch_tip_last_check failed (still confirmed)"); + } + return Ok(()); + } + + warn!( + tip_id = %tip.id, asset = %tip.asset, recorded_sweep_txid = %recorded_txid, + "recorded sweep no longer confirmed — reverting claim (chain reorg)" + ); + + match state.db.revert_sweep_on_reorg(tip.id).await? { + Some(reverted) => send_reorg_dm_if_needed(state, &reverted).await, + None => debug!(tip_id = %tip.id, "reorg revert no-op (already reverted) — skipping notify"), + } + Ok(()) +} + +// ============================================================================ +// Notify side-effects (STUBBED to no-ops; exactly-once stamps preserved) +// ============================================================================ + +/// Sender "your tip was claimed" notify. STUBBED to a no-op in this port (public +/// tips deliver by share URL), but we still stamp `sweep_confirmed_dm_sent_at` +/// (guarded `IS NULL`) so re-enabling notifications later stays exactly-once. +async fn send_sweep_confirmed_dm_if_needed(state: &AppState, tip: &SocialTipRow) { + if tip.sweep_confirmed_dm_sent_at.is_some() { + return; + } + // TODO(notify): send the sender DM here when notifications are re-enabled. + if let Err(e) = state.db.stamp_sweep_confirmed_dm_sent(tip.id).await { + debug!(tip_id = %tip.id, error = %e, "stamp_sweep_confirmed_dm_sent failed"); + } +} + +/// Sender "claim reversed by reorg" notify. STUBBED to a no-op; stamps +/// `reorg_notified_at` (guarded `IS NULL`) to keep exactly-once semantics. +async fn send_reorg_dm_if_needed(state: &AppState, tip: &SocialTipRow) { + if tip.reorg_notified_at.is_some() { + return; + } + // TODO(notify): send the sender reorg DM here when notifications are re-enabled. + if let Err(e) = state.db.stamp_reorg_notified(tip.id).await { + debug!(tip_id = %tip.id, error = %e, "stamp_reorg_notified failed"); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::infra::electrum::{ScriptPubKeyInfo, VerboseTransaction, VerboseVin, VerboseVout}; + + fn spend(txid: &str, height: i64) -> ConfirmedSpend { + ConfirmedSpend { + txid: txid.to_string(), + block_height: height, + block_hash: None, + } + } + + // ---- pick_winner ------------------------------------------------- + + #[test] + fn pick_winner_empty_is_none() { + let spends: Vec = vec![]; + assert!(pick_winner(&spends).is_none()); + } + + #[test] + fn pick_winner_earliest_height_wins() { + let spends = vec![spend("zzz", 110), spend("aaa", 100), spend("mmm", 105)]; + let w = pick_winner(&spends).unwrap(); + assert_eq!(w.txid, "aaa"); + assert_eq!(w.block_height, 100); + } + + #[test] + fn pick_winner_same_height_smallest_txid_tiebreak() { + let spends = vec![spend("ffff", 100), spend("aaaa", 100), spend("cccc", 100)]; + assert_eq!(pick_winner(&spends).unwrap().txid, "aaaa"); + } + + #[test] + fn pick_winner_is_deterministic_across_input_order() { + let a = vec![spend("bbb", 100), spend("aaa", 100)]; + let b = vec![spend("aaa", 100), spend("bbb", 100)]; + assert_eq!(pick_winner(&a).unwrap().txid, pick_winner(&b).unwrap().txid); + assert_eq!(pick_winner(&a).unwrap().txid, "aaa"); + } + + #[test] + fn pick_winner_height_beats_txid() { + let spends = vec![spend("aaa", 200), spend("zzz", 100)]; + assert_eq!(pick_winner(&spends).unwrap().txid, "zzz"); + } + + // ---- sweep_confirm_threshold ------------------------------------- + + #[test] + fn threshold_btc_ltc_is_one() { + assert_eq!(sweep_confirm_threshold("btc"), 1); + assert_eq!(sweep_confirm_threshold("ltc"), 1); + assert_eq!(sweep_confirm_threshold("BTC"), 1); + } + + #[test] + fn threshold_wow_is_four() { + assert_eq!(sweep_confirm_threshold("wow"), 4); + assert_eq!(sweep_confirm_threshold("WOW"), 4); + } + + #[test] + fn threshold_xmr_is_ten() { + assert_eq!(sweep_confirm_threshold("xmr"), 10); + } + + #[test] + fn threshold_unknown_defaults_conservative() { + assert_eq!(sweep_confirm_threshold("doge"), 10); + } + + // ---- is_verbose_tx_spend_of -------------------------------------- + + fn spk(addr: &str) -> ScriptPubKeyInfo { + ScriptPubKeyInfo { + address: Some(addr.to_string()), + addresses: None, + } + } + + #[test] + fn funding_tx_is_not_a_spend_of_tip_address() { + // Funding tx: tip address RECEIVES; recv != 0 → not a spend. + let tx = VerboseTransaction { + txid: "fund".into(), + vin: vec![VerboseVin { txid: Some("alice_prev".into()), vout: Some(0), prevout: None }], + vout: vec![ + VerboseVout { value: 0.001, n: 0, script_pub_key: spk("ltc1qtip") }, + VerboseVout { value: 0.296, n: 1, script_pub_key: spk("ltc1qalice") }, + ], + }; + assert!(!is_verbose_tx_spend_of(&tx, "ltc1qtip").unwrap()); + } + + #[test] + fn sweep_tx_is_a_spend_of_tip_address() { + // Sweep: tip address absent from outputs (recv == 0) → spend. + let tx = VerboseTransaction { + txid: "sweep".into(), + vin: vec![VerboseVin { txid: Some("fund".into()), vout: Some(0), prevout: None }], + vout: vec![VerboseVout { value: 0.00099, n: 0, script_pub_key: spk("ltc1qbob") }], + }; + assert!(is_verbose_tx_spend_of(&tx, "ltc1qtip").unwrap()); + } + + #[test] + fn incoming_payment_with_change_is_not_a_spend() { + // Tip address receives with the payer's change to a third party — + // recv != 0 → not a spend (guards the funding-with-change case). + let tx = VerboseTransaction { + txid: "pay".into(), + vin: vec![VerboseVin { txid: Some("payer_prev".into()), vout: Some(0), prevout: None }], + vout: vec![ + VerboseVout { value: 0.5, n: 0, script_pub_key: spk("ltc1qtip") }, + VerboseVout { value: 0.3, n: 1, script_pub_key: spk("ltc1qpayer") }, + ], + }; + assert!(!is_verbose_tx_spend_of(&tx, "ltc1qtip").unwrap()); + } +} diff --git a/tests/tips.rs b/tests/tips.rs index 9fee471..0699399 100644 --- a/tests/tips.rs +++ b/tests/tips.rs @@ -399,3 +399,304 @@ async fn get_tips_pending_confirmation_filters() { "btc tip (threshold 0) must be excluded from the confirmation query" ); } + +// ── Stage 4: money-out (claim / confirm-sweep / clawback + reconciler) DB tests ── +// +// Exercises the audited money-safety guards directly on the DB layer. BTC tips +// (confirmations_required = 0) manufacture a claimable row with no chain access. + +/// Manufacture a claimable (pending, amount-verified) BTC tip owned by `uid`. +async fn make_pending(app: &common::TestApp, uid: uuid::Uuid, txid: &str) -> uuid::Uuid { + let id = make_draft(app, uid, "btc", 100_000, 0, None).await; + app.state + .db + .attach_funding_to_tip(id, uid, txid) + .await + .expect("attach funding"); + let row = app + .state + .db + .mark_tip_funding_verified(id, 100_000) + .await + .expect("verify") + .expect("verify returns row"); + assert_eq!(row.status, "pending", "make_pending must land in 'pending'"); + id +} + +/// Manufacture a `claiming` tip (pending -> claiming) locked by `claimant`. +async fn make_claiming( + app: &common::TestApp, + uid: uuid::Uuid, + claimant: uuid::Uuid, + txid: &str, +) -> uuid::Uuid { + let id = make_pending(app, uid, txid).await; + let row = app + .state + .db + .mark_tip_claiming(id, claimant) + .await + .expect("claim") + .expect("claiming row"); + assert_eq!(row.status, "claiming"); + id +} + +/// Manufacture a settled `claimed` tip (with a sweep_block_height witness). +async fn make_claimed( + app: &common::TestApp, + uid: uuid::Uuid, + claimant: uuid::Uuid, + txid: &str, +) -> uuid::Uuid { + let id = make_claiming(app, uid, claimant, txid).await; + let row = app + .state + .db + .confirm_sweep_onchain(id, "onchain-swp", 100, None, None) + .await + .expect("settle") + .expect("claimed row"); + assert_eq!(row.status, "claimed"); + id +} + +/// (a) THE DOUBLE-CLAIM GUARD: a fresh claimable tip claims once; a +/// funding_mismatch / unverified / clawed_back / already-swept row is never +/// claimable. Re-claiming a `claiming` row IS intentionally allowed (verbatim +/// `status IN ('pending','claiming')` — public-tip retry semantics). +#[tokio::test] +async fn mark_tip_claiming_double_claim_guard() { + let Some(app) = tips_app().await else { return }; + let sender = app.create_user().await; + let claimant = app.create_user().await; + + // Fresh claimable tip: first claim locks it into 'claiming'. + let id = make_pending(&app, sender, "fund-claim-1").await; + let first = app.state.db.mark_tip_claiming(id, claimant).await.expect("claim"); + assert!(first.is_some(), "first claim on a pending tip must succeed"); + assert_eq!(first.unwrap().status, "claiming"); + + // Re-claim on a 'claiming' public tip is INTENTIONALLY allowed: the claim + // state is a UX signal, not a lock — the reconciler resolves the on-chain + // sweep-race winner. (Verbatim audit-T1 retry semantics.) + let reclaim = app.state.db.mark_tip_claiming(id, claimant).await.expect("reclaim"); + assert!(reclaim.is_some(), "re-claim on a claiming public tip is allowed"); + + // None on funding_mismatch (audit-T1 fund-loss guard). + let mm = make_draft(&app, sender, "btc", 1_000_000, 0, None).await; + app.state.db.attach_funding_to_tip(mm, sender, "fund-mm").await.expect("attach"); + app.state.db.mark_tip_funding_mismatch(mm, 10).await.expect("mismatch").expect("row"); + assert!( + app.state.db.mark_tip_claiming(mm, claimant).await.expect("claim mm").is_none(), + "a funding_mismatch tip must never be claimable" + ); + + // None on an unverified (pending_confirmation) tip. + let pc = make_draft(&app, sender, "btc", 100_000, 0, None).await; + app.state.db.attach_funding_to_tip(pc, sender, "fund-pc").await.expect("attach"); + assert!( + app.state.db.mark_tip_claiming(pc, claimant).await.expect("claim pc").is_none(), + "an unverified tip must not be claimable" + ); + + // None on a clawed_back tip. + let cb = make_pending(&app, sender, "fund-cb").await; + app.state.db.clawback_social_tip(cb, sender).await.expect("clawback").expect("row"); + assert!( + app.state.db.mark_tip_claiming(cb, claimant).await.expect("claim cb").is_none(), + "a clawed_back tip must not be claimable" + ); + + // None on an already-swept (settled) tip. + let swept = make_claimed(&app, sender, claimant, "fund-swept").await; + assert!( + app.state.db.mark_tip_claiming(swept, claimant).await.expect("claim swept").is_none(), + "a settled tip must not be re-claimable" + ); +} + +/// (b) confirm_tip_sweep is first-recorder-wins: a second, DIFFERENT txid must +/// NOT overwrite the recorded winner (both calls return the live row with the +/// FIRST txid). A never-claimed tip returns None. +#[tokio::test] +async fn confirm_tip_sweep_first_recorder_wins() { + let Some(app) = tips_app().await else { return }; + let sender = app.create_user().await; + let claimant = app.create_user().await; + let id = make_claiming(&app, sender, claimant, "fund-sweep").await; + + // First recorder writes txid-A; the row STAYS 'claiming' (record, not settle). + let r1 = app.state.db.confirm_tip_sweep(id, claimant, "txid-A").await.expect("record").expect("row"); + assert_eq!(r1.sweep_txid.as_deref(), Some("txid-A")); + assert_eq!(r1.status, "claiming", "confirm_tip_sweep records, never settles"); + + // A DIFFERENT txid does not overwrite; fallthrough returns the live row (A). + let r2 = app.state.db.confirm_tip_sweep(id, claimant, "txid-B").await.expect("second").expect("live row"); + assert_eq!( + r2.sweep_txid.as_deref(), + Some("txid-A"), + "first-recorder wins: a different txid must not overwrite" + ); + assert_eq!(r2.status, "claiming"); + + // Same txid again is an idempotent no-op returning A. + let r3 = app.state.db.confirm_tip_sweep(id, claimant, "txid-A").await.expect("idem").expect("row"); + assert_eq!(r3.sweep_txid.as_deref(), Some("txid-A")); + + // A never-claimed (pending) tip → None. + let pending = make_pending(&app, sender, "fund-cs-pending").await; + assert!( + app.state.db.confirm_tip_sweep(pending, claimant, "txid-x").await.expect("cs pending").is_none(), + "confirm-sweep on a non-claiming tip returns None" + ); +} + +/// (c) confirm_sweep_onchain settles 'claiming' -> 'claimed' ONLY, and it is the +/// only path to 'claimed'. It records the height witness and preserves +/// claimed_by when passed None; a second call is a no-op. +#[tokio::test] +async fn confirm_sweep_onchain_settles_claiming_only() { + let Some(app) = tips_app().await else { return }; + let sender = app.create_user().await; + let claimant = app.create_user().await; + + let id = make_claiming(&app, sender, claimant, "fund-settle").await; + let settled = app + .state + .db + .confirm_sweep_onchain(id, "swp-1", 500, Some("blockhash"), None) + .await + .expect("settle") + .expect("claimed row"); + assert_eq!(settled.status, "claimed"); + assert!(settled.sweep_confirmed_at.is_some()); + assert_eq!(settled.sweep_block_height, Some(500)); + assert_eq!(settled.sweep_txid.as_deref(), Some("swp-1")); + assert_eq!( + settled.claimed_by_user_id, + Some(claimant), + "claimed_by is preserved when claimed_by=None" + ); + + // Second settle is a no-op (one-way lock). + assert!( + app.state.db.confirm_sweep_onchain(id, "swp-2", 501, None, None).await.expect("resettle").is_none(), + "an already-claimed row cannot be re-settled" + ); + + // A pending (non-claiming) tip cannot be settled — 'claiming' is the only + // entry into 'claimed'. + let pending = make_pending(&app, sender, "fund-settle-pending").await; + assert!( + app.state.db.confirm_sweep_onchain(pending, "swp", 1, None, None).await.expect("settle pending").is_none(), + "confirm_sweep_onchain requires status='claiming'" + ); + let still = app.state.db.get_social_tip(pending).await.expect("get").expect("row"); + assert_eq!(still.status, "pending", "a pending tip must not be flipped to claimed"); +} + +/// (d) revert_sweep_on_reorg reverts 'claimed' -> 'claiming' ONLY, clearing the +/// sweep witness (incl. sweep_txid so the next cycle records the new winner). +#[tokio::test] +async fn revert_sweep_on_reorg_only_from_claimed() { + let Some(app) = tips_app().await else { return }; + let sender = app.create_user().await; + let claimant = app.create_user().await; + + let id = make_claimed(&app, sender, claimant, "fund-reorg").await; + let reverted = app.state.db.revert_sweep_on_reorg(id).await.expect("revert").expect("reverted row"); + assert_eq!(reverted.status, "claiming"); + assert!(reverted.sweep_confirmed_at.is_none()); + assert!( + reverted.sweep_txid.is_none(), + "reorg clears sweep_txid so the next confirm cycle records the new on-chain winner" + ); + assert!(reverted.sweep_block_height.is_none()); + + // Second revert is a no-op (now 'claiming', not 'claimed'). + assert!( + app.state.db.revert_sweep_on_reorg(id).await.expect("re-revert").is_none(), + "revert requires status='claimed'" + ); + + // Revert on a plain pending tip → None. + let pending = make_pending(&app, sender, "fund-reorg-pending").await; + assert!( + app.state.db.revert_sweep_on_reorg(pending).await.expect("revert pending").is_none(), + "a pending tip cannot be reorg-reverted" + ); +} + +/// (e) clawback succeeds for the SENDER on pending / pending_confirmation / +/// claiming / funding_mismatch / cancelled while sweep_confirmed_at IS NULL, +/// is owner-scoped, and is BLOCKED once the sweep confirmed. +#[tokio::test] +async fn clawback_status_matrix_and_settled_block() { + let Some(app) = tips_app().await else { return }; + let sender = app.create_user().await; + let claimant = app.create_user().await; + + // pending → clawable. + let pending = make_pending(&app, sender, "cb-pending").await; + assert!(app.state.db.clawback_social_tip(pending, sender).await.expect("cb").is_some()); + + // pending_confirmation → clawable. + let pc = make_draft(&app, sender, "btc", 100_000, 0, None).await; + app.state.db.attach_funding_to_tip(pc, sender, "cb-pc").await.expect("attach"); + assert!(app.state.db.clawback_social_tip(pc, sender).await.expect("cb").is_some()); + + // claiming → clawable (sender racing a claimer). + let claiming = make_claiming(&app, sender, claimant, "cb-claiming").await; + assert!(app.state.db.clawback_social_tip(claiming, sender).await.expect("cb").is_some()); + + // funding_mismatch → clawable. + let mm = make_draft(&app, sender, "btc", 1_000_000, 0, None).await; + app.state.db.attach_funding_to_tip(mm, sender, "cb-mm").await.expect("attach"); + app.state.db.mark_tip_funding_mismatch(mm, 10).await.expect("mismatch").expect("row"); + assert!(app.state.db.clawback_social_tip(mm, sender).await.expect("cb").is_some()); + + // cancelled (GC'd but still-funded draft) → clawable. + let cancelled = make_draft(&app, sender, "btc", 100_000, 0, None).await; + app.state.db.cancel_draft_social_tip(cancelled, sender).await.expect("cancel").expect("id"); + assert!(app.state.db.clawback_social_tip(cancelled, sender).await.expect("cb").is_some()); + + // Owner-scoped: a stranger cannot claw back; the real sender can. + let owned = make_pending(&app, sender, "cb-owner").await; + let stranger = app.create_user().await; + assert!( + app.state.db.clawback_social_tip(owned, stranger).await.expect("cb stranger").is_none(), + "only the sender can claw back" + ); + assert!(app.state.db.clawback_social_tip(owned, sender).await.expect("cb owner").is_some()); + + // BLOCKED once the sweep confirmed (sweep_confirmed_at IS NOT NULL). + let swept = make_claimed(&app, sender, claimant, "cb-swept").await; + assert!( + app.state.db.clawback_social_tip(swept, sender).await.expect("cb swept").is_none(), + "a settled tip must not be clawed back" + ); +} + +/// Reconciler poll queries select the right rows. A huge limit avoids the +/// ordering / rate-limit windows hiding our rows amid other tests' rows. +#[tokio::test] +async fn reconciler_poll_queries_select_the_right_rows() { + let Some(app) = tips_app().await else { return }; + let sender = app.create_user().await; + let claimant = app.create_user().await; + + let claiming = make_claiming(&app, sender, claimant, "poll-claiming").await; + let pending = make_pending(&app, sender, "poll-pending").await; + let claimed = make_claimed(&app, sender, claimant, "poll-claimed").await; + + let awaiting = app.state.db.get_tips_awaiting_sweep_confirmation(1_000_000).await.expect("awaiting"); + assert!(awaiting.iter().any(|t| t.id == claiming), "a claiming tip awaits sweep confirmation"); + assert!(!awaiting.iter().any(|t| t.id == pending), "a pending tip does not await sweep confirmation"); + assert!(!awaiting.iter().any(|t| t.id == claimed), "a settled tip does not await sweep confirmation"); + + let reorg = app.state.db.get_settled_tips_for_reorg_check(1_000_000).await.expect("reorg"); + assert!(reorg.iter().any(|t| t.id == claimed), "a settled tip with a height witness is reorg-checked"); + assert!(!reorg.iter().any(|t| t.id == claiming), "a claiming tip is not reorg-checked"); +}