Genarrative/server-rs/crates/api-server/src/wallet_refund_outbox.rs

use std::{
    fmt,
    path::{Path, PathBuf},
    sync::Arc,
    time::{Duration, SystemTime, UNIX_EPOCH},
};

use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use spacetime_client::{SpacetimeClient, SpacetimeClientError};
use tokio::{
    fs::{self, File, OpenOptions},
    io::{AsyncReadExt, AsyncWriteExt},
    sync::{Mutex, Notify},
    time::sleep,
};
use tracing::{debug, warn};

use crate::config::AppConfig;

const PENDING_FILE_PREFIX: &str = "refund-";
const OVERFLOW_FILE_PREFIX: &str = "refund-overflow-";
const CORRUPT_FILE_PREFIX: &str = "corrupt-";
const TEMP_FILE_PREFIX: &str = "tmp-";
const OUTBOX_FILE_EXTENSION: &str = ".json";

#[derive(Clone)]
pub struct WalletRefundOutbox {
    dir: PathBuf,
    batch_size: usize,
    flush_interval: Duration,
    max_bytes: u64,
    spacetime_client: SpacetimeClient,
    enqueue_lock: Arc<Mutex<()>>,
    flush_notify: Arc<Notify>,
}

#[derive(Clone)]
pub struct ProfileWalletRefundOutboxWorker {
    batch_size: u32,
    flush_interval: Duration,
    spacetime_client: SpacetimeClient,
    worker_id: String,
}

impl ProfileWalletRefundOutboxWorker {
    pub fn from_config(config: &AppConfig, spacetime_client: SpacetimeClient) -> Arc<Self> {
        Arc::new(Self {
            batch_size: config
                .wallet_refund_outbox_batch_size
                .max(1)
                .min(u32::MAX as usize) as u32,
            flush_interval: config.wallet_refund_outbox_flush_interval,
            spacetime_client,
            worker_id: format!("api-server-refund-outbox-{}", std::process::id()),
        })
    }

    pub fn spawn_worker(self: Arc<Self>) {
        tokio::spawn(async move {
            self.process_once().await;
            loop {
                sleep(self.flush_interval).await;
                self.process_once().await;
            }
        });
    }

    async fn process_once(&self) {
        match self
            .spacetime_client
            .process_profile_wallet_refund_outbox(self.worker_id.clone(), self.batch_size)
            .await
        {
            Ok(result) if result.failed_count > 0 => {
                warn!(
                    worker_id = %self.worker_id,
                    processed_count = result.processed_count,
                    retry_count = result.retry_count,
                    failed_count = result.failed_count,
                    "profile wallet refund outbox 处理部分失败，将按库内 available_at 重试"
                );
            }
            Ok(result) if result.processed_count > 0 => {
                debug!(
                    worker_id = %self.worker_id,
                    processed_count = result.processed_count,
                    "profile wallet refund outbox 已完成库内退款"
                );
            }
            Ok(_) => {}
            Err(error) => {
                warn!(
                    worker_id = %self.worker_id,
                    error = %error,
                    "profile wallet refund outbox worker 暂时无法连接 SpacetimeDB"
                );
            }
        }
    }
}

#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
pub(crate) struct WalletRefundOutboxRecord {
    pub owner_user_id: String,
    pub amount: u64,
    pub ledger_id: String,
    pub created_at_micros: i64,
    pub asset_kind: String,
    pub asset_id: String,
    #[serde(default = "default_settlement_reason")]
    pub settlement_reason: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub external_generation_job_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub external_generation_claim_attempt: Option<u32>,
}

#[derive(Debug)]
pub enum WalletRefundOutboxEnqueueOutcome {
    Enqueued,
    OverflowEnqueued { reason: &'static str },
}

#[derive(Debug)]
pub enum WalletRefundOutboxError {
    Io(std::io::Error),
    Json(serde_json::Error),
    Spacetime(SpacetimeClientError),
}

impl WalletRefundOutbox {
    pub fn from_config(config: &AppConfig, spacetime_client: SpacetimeClient) -> Option<Arc<Self>> {
        if !config.wallet_refund_outbox_enabled {
            return None;
        }

        Some(Arc::new(Self {
            dir: config.wallet_refund_outbox_dir.clone(),
            batch_size: config.wallet_refund_outbox_batch_size.max(1),
            flush_interval: config.wallet_refund_outbox_flush_interval,
            max_bytes: config.wallet_refund_outbox_max_bytes,
            spacetime_client,
            enqueue_lock: Arc::new(Mutex::new(())),
            flush_notify: Arc::new(Notify::new()),
        }))
    }

    pub async fn enqueue(
        &self,
        record: WalletRefundOutboxRecord,
    ) -> Result<WalletRefundOutboxEnqueueOutcome, WalletRefundOutboxError> {
        let _guard = self.enqueue_lock.lock().await;
        fs::create_dir_all(&self.dir).await?;

        let pending_path = self.pending_path_for_ledger(&record.ledger_id);
        let overflow_path = self.overflow_path_for_ledger(&record.ledger_id);
        if self
            .reuse_existing_pending_file(&pending_path, &record)
            .await?
            || self
                .reuse_existing_pending_file(&overflow_path, &record)
                .await?
        {
            self.flush_notify.notify_one();
            return Ok(WalletRefundOutboxEnqueueOutcome::Enqueued);
        }

        let bytes = serde_json::to_vec(&record)?;
        let line_bytes = bytes.len().min(u64::MAX as usize) as u64;
        let current_bytes = directory_size_if_exists(&self.dir).unwrap_or(0);
        let overflow = current_bytes.saturating_add(line_bytes) > self.max_bytes;
        let target_path = if overflow {
            &overflow_path
        } else {
            &pending_path
        };

        let temp_path = self.temp_path();
        let mut file = OpenOptions::new()
            .create_new(true)
            .write(true)
            .open(&temp_path)
            .await?;
        file.write_all(&bytes).await?;
        file.flush().await?;
        file.sync_data().await?;
        drop(file);
        if self
            .reuse_existing_pending_file(&pending_path, &record)
            .await?
            || self
                .reuse_existing_pending_file(&overflow_path, &record)
                .await?
        {
            let _ = fs::remove_file(&temp_path).await;
            self.flush_notify.notify_one();
            return Ok(WalletRefundOutboxEnqueueOutcome::Enqueued);
        }
        for _ in 0..2 {
            match fs::hard_link(&temp_path, target_path).await {
                Ok(()) => {
                    sync_directory_metadata(&self.dir).await?;
                    remove_file_and_sync(&temp_path, &self.dir).await?;
                    self.flush_notify.notify_one();
                    return Ok(if overflow {
                        WalletRefundOutboxEnqueueOutcome::OverflowEnqueued {
                            reason: "max_bytes",
                        }
                    } else {
                        WalletRefundOutboxEnqueueOutcome::Enqueued
                    });
                }
                Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
                    if self
                        .reuse_existing_pending_file(target_path, &record)
                        .await?
                    {
                        remove_file_and_sync(&temp_path, &self.dir).await?;
                        self.flush_notify.notify_one();
                        return Ok(WalletRefundOutboxEnqueueOutcome::Enqueued);
                    }
                }
                Err(error) => return Err(error.into()),
            }
        }
        Err(std::io::Error::new(
            std::io::ErrorKind::AlreadyExists,
            format!(
                "refund pending path could not be installed: {}",
                target_path.display()
            ),
        )
        .into())
    }

    async fn reuse_existing_pending_file(
        &self,
        pending_path: &Path,
        expected: &WalletRefundOutboxRecord,
    ) -> Result<bool, WalletRefundOutboxError> {
        let metadata = match fs::metadata(pending_path).await {
            Ok(metadata) => metadata,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
            Err(error) => return Err(error.into()),
        };
        if !metadata.is_file() {
            return Err(std::io::Error::new(
                std::io::ErrorKind::AlreadyExists,
                format!(
                    "refund pending path is not a regular file: {}",
                    pending_path.display()
                ),
            )
            .into());
        }

        match read_refund_record(pending_path).await {
            Ok(existing) if existing == *expected => Ok(true),
            Ok(existing) if existing.ledger_id == expected.ledger_id => Err(std::io::Error::new(
                std::io::ErrorKind::AlreadyExists,
                format!(
                    "refund ledger {} 已存在但退款事实不一致: {}",
                    expected.ledger_id,
                    pending_path.display()
                ),
            )
            .into()),
            Ok(_) => Err(std::io::Error::new(
                std::io::ErrorKind::AlreadyExists,
                format!("refund ledger hash collision at {}", pending_path.display()),
            )
            .into()),
            Err(error) if error.is_data_corruption() => {
                // Preserve the malformed durable file for inspection, then allow this
                // enqueue to install a valid file for the same ledger id.
                self.quarantine_file(pending_path).await?;
                warn!(
                    source = %pending_path.display(),
                    "wallet refund outbox 已隔离损坏 pending 文件，继续写入新的幂等退款记录"
                );
                Ok(false)
            }
            Err(error) => Err(error),
        }
    }

    pub fn spawn_worker(self: Arc<Self>) {
        tokio::spawn(async move {
            if let Err(error) = self.flush_pending_files_once().await {
                warn!(error = %error, "wallet refund outbox 启动恢复退款失败，将保留文件等待重试");
            }

            loop {
                tokio::select! {
                    _ = sleep(self.flush_interval) => {
                        if let Err(error) = self.flush_pending_files_once().await {
                            warn!(error = %error, "wallet refund outbox 重放退款失败，将保留文件等待重试");
                        }
                    }
                    _ = self.flush_notify.notified() => {
                        if let Err(error) = self.flush_pending_files_once().await {
                            warn!(error = %error, "wallet refund outbox 主动重放退款失败，将保留文件等待重试");
                        }
                    }
                }
            }
        });
    }

    pub async fn flush_for_shutdown(&self) -> Result<(), WalletRefundOutboxError> {
        self.flush_pending_files_once().await
    }

    async fn flush_pending_files_once(&self) -> Result<(), WalletRefundOutboxError> {
        fs::create_dir_all(&self.dir).await?;
        self.recover_temporary_files().await?;
        let pending_files = self.list_pending_files().await?;
        for path in pending_files.into_iter().take(self.batch_size) {
            let record = match read_refund_record(&path).await {
                Ok(record) => record,
                Err(error) if error.is_data_corruption() => {
                    let corrupt_path = self.corrupt_path_for(&path);
                    fs::rename(&path, &corrupt_path).await?;
                    sync_directory_metadata(&self.dir).await?;
                    warn!(
                        error = %error,
                        source = %path.display(),
                        target = %corrupt_path.display(),
                        "wallet refund outbox 文件无法解析，已隔离"
                    );
                    continue;
                }
                Err(error) => return Err(error),
            };

            let enqueue_input =
                module_runtime::build_runtime_profile_wallet_refund_outbox_enqueue_input(
                    record.owner_user_id.clone(),
                    record.amount,
                    record.ledger_id.clone(),
                    record.created_at_micros,
                    record.asset_kind.clone(),
                    record.asset_id.clone(),
                    record.settlement_reason.clone(),
                    record.external_generation_job_id.clone(),
                    record
                        .external_generation_claim_attempt
                        .or_else(|| infer_external_generation_claim_attempt(&record)),
                )
                .map_err(|error| {
                    WalletRefundOutboxError::Spacetime(SpacetimeClientError::Runtime(
                        error.to_string(),
                    ))
                })?;
            match self
                .spacetime_client
                .enqueue_profile_wallet_refund_outbox(enqueue_input)
                .await
            {
                Ok(_) => {
                    match fs::remove_file(&path).await {
                        Ok(()) => {}
                        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                        Err(error) => return Err(error.into()),
                    }
                    sync_directory_metadata(&self.dir).await?;
                    debug!(
                        ledger_id = %record.ledger_id,
                        owner_user_id = %record.owner_user_id,
                        asset_kind = %record.asset_kind,
                        asset_id = %record.asset_id,
                        external_generation_job_id = ?record.external_generation_job_id,
                        path = %path.display(),
                        "wallet refund emergency spool 已恢复到 SpacetimeDB outbox 并删除文件"
                    );
                }
                Err(error) => return Err(WalletRefundOutboxError::Spacetime(error)),
            }
        }
        Ok(())
    }

    async fn recover_temporary_files(&self) -> Result<(), WalletRefundOutboxError> {
        let _guard = self.enqueue_lock.lock().await;
        let temporary_files = self.list_temporary_files().await?;
        'temporary_files: for path in temporary_files {
            let record = match read_refund_record(&path).await {
                Ok(record) => record,
                Err(error) if error.is_data_corruption() => {
                    self.quarantine_file(&path).await?;
                    warn!(
                        error = %error,
                        source = %path.display(),
                        "wallet refund outbox 崩溃遗留临时文件无法解析，已隔离"
                    );
                    continue;
                }
                Err(error) => return Err(error),
            };

            let pending_path = self.pending_path_for_ledger(&record.ledger_id);
            let overflow_path = self.overflow_path_for_ledger(&record.ledger_id);
            for existing_path in [&pending_path, &overflow_path] {
                match self
                    .reuse_existing_pending_file(existing_path, &record)
                    .await
                {
                    Ok(true) => {
                        remove_file_and_sync(&path, &self.dir).await?;
                        debug!(
                            ledger_id = %record.ledger_id,
                            source = %path.display(),
                            target = %existing_path.display(),
                            "wallet refund outbox 临时文件与已有幂等文件重复，已删除临时副本"
                        );
                        continue 'temporary_files;
                    }
                    Ok(false) => {}
                    Err(WalletRefundOutboxError::Io(error))
                        if error.kind() == std::io::ErrorKind::AlreadyExists =>
                    {
                        self.quarantine_file(&path).await?;
                        warn!(
                            ledger_id = %record.ledger_id,
                            source = %path.display(),
                            target = %existing_path.display(),
                            error = %error,
                            "wallet refund outbox 临时文件与现有幂等文件事实冲突，已隔离临时文件"
                        );
                        continue 'temporary_files;
                    }
                    Err(error) => return Err(error),
                }
            }

            let temp_bytes = fs::metadata(&path).await?.len();
            let record_bytes = serde_json::to_vec(&record)?;
            let current_bytes = directory_size_if_exists(&self.dir)
                .unwrap_or(0)
                .saturating_sub(temp_bytes);
            let target_path =
                if current_bytes.saturating_add(record_bytes.len() as u64) > self.max_bytes {
                    &overflow_path
                } else {
                    &pending_path
                };

            match fs::hard_link(&path, target_path).await {
                Ok(()) => {
                    sync_directory_metadata(&self.dir).await?;
                    remove_file_and_sync(&path, &self.dir).await?;
                    debug!(
                        ledger_id = %record.ledger_id,
                        source = %path.display(),
                        target = %target_path.display(),
                        "wallet refund outbox 崩溃遗留临时文件已恢复为幂等退款文件"
                    );
                }
                Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
                    match self.reuse_existing_pending_file(target_path, &record).await {
                        Ok(true) => {
                            // Another writer won the same ledger id with identical facts. Keep
                            // the first durable file and remove only this duplicate temporary
                            // link.
                            remove_file_and_sync(&path, &self.dir).await?;
                        }
                        Ok(false) => match fs::hard_link(&path, target_path).await {
                            Ok(()) => {
                                sync_directory_metadata(&self.dir).await?;
                                remove_file_and_sync(&path, &self.dir).await?;
                            }
                            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                                // The file may have been completed by another process after the
                                // scan.
                                continue;
                            }
                            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
                                self.quarantine_file(&path).await?;
                                warn!(
                                    ledger_id = %record.ledger_id,
                                    source = %path.display(),
                                    target = %target_path.display(),
                                    "wallet refund outbox 临时文件无法与现有幂等文件合并，已隔离临时文件"
                                );
                            }
                            Err(error) => return Err(error.into()),
                        },
                        Err(WalletRefundOutboxError::Io(error))
                            if error.kind() == std::io::ErrorKind::AlreadyExists =>
                        {
                            self.quarantine_file(&path).await?;
                            warn!(
                                ledger_id = %record.ledger_id,
                                source = %path.display(),
                                target = %target_path.display(),
                                error = %error,
                                "wallet refund outbox 临时文件与现有幂等文件事实冲突，已隔离临时文件"
                            );
                        }
                        Err(error) => return Err(error),
                    }
                }
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                    // The file may have been completed by another process after the scan.
                    continue;
                }
                Err(error) => return Err(error.into()),
            }
        }
        Ok(())
    }

    async fn list_temporary_files(&self) -> Result<Vec<PathBuf>, WalletRefundOutboxError> {
        let mut entries = fs::read_dir(&self.dir).await?;
        let mut files = Vec::new();
        while let Some(entry) = entries.next_entry().await? {
            let path = entry.path();
            let Some(name) = path.file_name().and_then(|value| value.to_str()) else {
                continue;
            };
            if name.starts_with(TEMP_FILE_PREFIX) && name.ends_with(OUTBOX_FILE_EXTENSION) {
                files.push(path);
            }
        }
        files.sort();
        Ok(files)
    }

    async fn quarantine_file(&self, path: &Path) -> Result<(), WalletRefundOutboxError> {
        let corrupt_path = self.corrupt_path_for(path);
        fs::rename(path, &corrupt_path).await?;
        sync_directory_metadata(&self.dir).await?;
        Ok(())
    }

    async fn list_pending_files(&self) -> Result<Vec<PathBuf>, WalletRefundOutboxError> {
        let mut entries = fs::read_dir(&self.dir).await?;
        let mut files = Vec::new();
        while let Some(entry) = entries.next_entry().await? {
            let path = entry.path();
            let Some(name) = path.file_name().and_then(|value| value.to_str()) else {
                continue;
            };
            if (name.starts_with(PENDING_FILE_PREFIX) || name.starts_with(OVERFLOW_FILE_PREFIX))
                && name.ends_with(OUTBOX_FILE_EXTENSION)
            {
                files.push(path);
            }
        }
        files.sort();
        Ok(files)
    }

    fn pending_path_for_ledger(&self, ledger_id: &str) -> PathBuf {
        self.dir.join(format!(
            "{PENDING_FILE_PREFIX}{}{OUTBOX_FILE_EXTENSION}",
            ledger_id_hash(ledger_id)
        ))
    }

    fn overflow_path_for_ledger(&self, ledger_id: &str) -> PathBuf {
        self.dir.join(format!(
            "{OVERFLOW_FILE_PREFIX}{}{OUTBOX_FILE_EXTENSION}",
            ledger_id_hash(ledger_id)
        ))
    }

    fn temp_path(&self) -> PathBuf {
        self.dir.join(format!(
            "{TEMP_FILE_PREFIX}{}-{uuid}{OUTBOX_FILE_EXTENSION}",
            current_unix_micros(),
            uuid = uuid::Uuid::new_v4()
        ))
    }

    fn corrupt_path_for(&self, path: &Path) -> PathBuf {
        let name = path
            .file_name()
            .and_then(|value| value.to_str())
            .unwrap_or("unknown.json");
        self.dir.join(format!(
            "{CORRUPT_FILE_PREFIX}{}-{uuid}-{name}",
            current_unix_micros(),
            uuid = uuid::Uuid::new_v4()
        ))
    }
}

fn default_settlement_reason() -> String {
    "emergency_spool_replay".to_string()
}

fn infer_external_generation_claim_attempt(record: &WalletRefundOutboxRecord) -> Option<u32> {
    let job_id = record.external_generation_job_id.as_deref()?.trim();
    let prefix = format!("asset_operation_refund:external_generation_job:{job_id}:attempt:");
    record
        .ledger_id
        .strip_prefix(&prefix)
        .and_then(|value| value.parse::<u32>().ok())
}

impl fmt::Debug for WalletRefundOutbox {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("WalletRefundOutbox")
            .field("dir", &self.dir)
            .field("batch_size", &self.batch_size)
            .field("flush_interval", &self.flush_interval)
            .field("max_bytes", &self.max_bytes)
            .finish()
    }
}

impl fmt::Display for WalletRefundOutboxError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Io(error) => write!(f, "{error}"),
            Self::Json(error) => write!(f, "{error}"),
            Self::Spacetime(error) => write!(f, "{error}"),
        }
    }
}

impl From<std::io::Error> for WalletRefundOutboxError {
    fn from(value: std::io::Error) -> Self {
        Self::Io(value)
    }
}

impl From<serde_json::Error> for WalletRefundOutboxError {
    fn from(value: serde_json::Error) -> Self {
        Self::Json(value)
    }
}

impl WalletRefundOutboxError {
    fn is_data_corruption(&self) -> bool {
        matches!(self, Self::Json(_))
    }
}

async fn read_refund_record(
    path: &Path,
) -> Result<WalletRefundOutboxRecord, WalletRefundOutboxError> {
    let mut file = File::open(path).await?;
    let mut bytes = Vec::new();
    file.read_to_end(&mut bytes).await?;
    Ok(serde_json::from_slice::<WalletRefundOutboxRecord>(&bytes)?)
}

fn directory_size_if_exists(path: &Path) -> Result<u64, std::io::Error> {
    if !path.is_dir() {
        return Ok(0);
    }

    let mut total = 0u64;
    for entry in std::fs::read_dir(path)? {
        let entry = entry?;
        if !is_capped_outbox_file_name(&entry.file_name()) {
            continue;
        }
        let metadata = entry.metadata()?;
        if metadata.is_file() {
            total = total.saturating_add(metadata.len());
        }
    }
    Ok(total)
}

fn current_unix_micros() -> u128 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_micros()
}

fn ledger_id_hash(ledger_id: &str) -> String {
    hex::encode(Sha256::digest(ledger_id.as_bytes()))
}

fn is_pending_outbox_file_name(name: &std::ffi::OsStr) -> bool {
    name.to_str().is_some_and(|value| {
        (value.starts_with(PENDING_FILE_PREFIX)
            || value.starts_with(OVERFLOW_FILE_PREFIX)
            || value.starts_with(TEMP_FILE_PREFIX))
            && value.ends_with(OUTBOX_FILE_EXTENSION)
    })
}

fn is_capped_outbox_file_name(name: &std::ffi::OsStr) -> bool {
    name.to_str().is_some_and(|value| {
        ((value.starts_with(PENDING_FILE_PREFIX) && !value.starts_with(OVERFLOW_FILE_PREFIX))
            || value.starts_with(TEMP_FILE_PREFIX))
            && value.ends_with(OUTBOX_FILE_EXTENSION)
    })
}

async fn remove_file_and_sync(path: &Path, dir: &Path) -> Result<(), WalletRefundOutboxError> {
    match fs::remove_file(path).await {
        Ok(()) => sync_directory_metadata(dir).await,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(error.into()),
    }
}

async fn sync_directory_metadata(path: &Path) -> Result<(), WalletRefundOutboxError> {
    let path = path.to_path_buf();
    tokio::task::spawn_blocking(move || {
        let dir = std::fs::File::open(path)?;
        dir.sync_all()
    })
    .await
    .map_err(|error| std::io::Error::new(std::io::ErrorKind::Other, error.to_string()))??;
    Ok(())
}

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

    fn sample_record(ledger_id: &str) -> WalletRefundOutboxRecord {
        WalletRefundOutboxRecord {
            owner_user_id: "user-1".to_string(),
            amount: 2,
            ledger_id: ledger_id.to_string(),
            created_at_micros: 1_713_680_000_000_000,
            asset_kind: "puzzle_initial_image".to_string(),
            asset_id: "asset-1".to_string(),
            settlement_reason: "worker_attempt_failed".to_string(),
            external_generation_job_id: Some("extgen-test".to_string()),
            external_generation_claim_attempt: Some(1),
        }
    }

    fn test_dir(name: &str) -> PathBuf {
        let dir = std::env::temp_dir().join(format!(
            "genarrative-wallet-refund-outbox-{name}-{}",
            current_unix_micros()
        ));
        let _ = std::fs::remove_dir_all(&dir);
        dir
    }

    fn test_outbox(dir: PathBuf, max_bytes: u64) -> Arc<WalletRefundOutbox> {
        let config = AppConfig {
            wallet_refund_outbox_dir: dir,
            wallet_refund_outbox_batch_size: 500,
            wallet_refund_outbox_flush_interval: Duration::from_secs(60),
            wallet_refund_outbox_max_bytes: max_bytes,
            ..AppConfig::default()
        };
        WalletRefundOutbox::from_config(
            &config,
            SpacetimeClient::new(spacetime_client::SpacetimeClientConfig {
                server_url: "http://127.0.0.1:1".to_string(),
                database: "missing".to_string(),
                token: None,
                pool_size: 1,
                procedure_timeout: Duration::from_millis(10),
                subscribe_cached_read_models: false,
            }),
        )
        .expect("outbox should be enabled")
    }

    #[tokio::test]
    async fn enqueue_is_idempotent_per_ledger_id() {
        let dir = test_dir("idempotent");
        let outbox = test_outbox(dir.clone(), 1024 * 1024);

        outbox.enqueue(sample_record("ledger-1")).await.unwrap();
        outbox.enqueue(sample_record("ledger-1")).await.unwrap();

        let pending_count = std::fs::read_dir(&dir)
            .unwrap()
            .filter_map(Result::ok)
            .filter(|entry| is_pending_outbox_file_name(&entry.file_name()))
            .count();
        assert_eq!(pending_count, 1);

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn enqueue_rejects_conflicting_existing_ledger_file() {
        let dir = test_dir("conflicting-ledger");
        let outbox = test_outbox(dir.clone(), 1024 * 1024);
        outbox.enqueue(sample_record("ledger-1")).await.unwrap();

        let mut conflicting = sample_record("ledger-1");
        conflicting.amount += 1;
        let error = outbox
            .enqueue(conflicting)
            .await
            .expect_err("conflicting refund must fail");
        assert!(
            matches!(error, WalletRefundOutboxError::Io(error) if error.kind() == std::io::ErrorKind::AlreadyExists)
        );

        let _ = std::fs::remove_dir_all(dir);
    }

    #[test]
    fn legacy_external_refund_record_infers_missing_claim_attempt() {
        let mut record =
            sample_record("asset_operation_refund:external_generation_job:extgen-test:attempt:7");
        record.external_generation_claim_attempt = None;

        assert_eq!(infer_external_generation_claim_attempt(&record), Some(7));
    }

    #[tokio::test]
    async fn enqueue_uses_durable_overflow_file_when_outbox_exceeds_max_bytes() {
        let dir = test_dir("max-bytes");
        let outbox = test_outbox(dir.clone(), 1);

        let outcome = outbox.enqueue(sample_record("ledger-1")).await.unwrap();

        assert!(matches!(
            outcome,
            WalletRefundOutboxEnqueueOutcome::OverflowEnqueued {
                reason: "max_bytes"
            }
        ));
        assert!(outbox.overflow_path_for_ledger("ledger-1").is_file());
        assert_eq!(directory_size_if_exists(&dir).unwrap(), 0);
        assert_eq!(
            read_refund_record(&outbox.overflow_path_for_ledger("ledger-1"))
                .await
                .unwrap(),
            sample_record("ledger-1")
        );

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn flush_quarantines_corrupt_file() {
        let dir = test_dir("corrupt");
        std::fs::create_dir_all(&dir).unwrap();
        let pending_path = dir.join(format!("{PENDING_FILE_PREFIX}bad{OUTBOX_FILE_EXTENSION}"));
        std::fs::write(&pending_path, b"{not-json}").unwrap();
        let outbox = test_outbox(dir.clone(), 1024 * 1024);

        outbox.flush_pending_files_once().await.unwrap();

        assert!(!pending_path.exists());
        let corrupt_count = std::fs::read_dir(&dir)
            .unwrap()
            .filter_map(Result::ok)
            .filter(|entry| {
                entry
                    .file_name()
                    .to_str()
                    .is_some_and(|name| name.starts_with(CORRUPT_FILE_PREFIX))
            })
            .count();
        assert_eq!(corrupt_count, 1);

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn enqueue_does_not_silently_accept_corrupt_pending_file() {
        let dir = test_dir("corrupt-pending-enqueue");
        std::fs::create_dir_all(&dir).unwrap();
        let outbox = test_outbox(dir.clone(), 1024 * 1024);
        let record = sample_record("ledger-corrupt-pending");
        let pending_path = outbox.pending_path_for_ledger(&record.ledger_id);
        std::fs::write(&pending_path, b"{not-json}").unwrap();

        outbox.enqueue(record.clone()).await.unwrap();

        assert_eq!(
            read_refund_record(&pending_path).await.unwrap().ledger_id,
            record.ledger_id
        );
        let corrupt_count = std::fs::read_dir(&dir)
            .unwrap()
            .filter_map(Result::ok)
            .filter(|entry| {
                entry
                    .file_name()
                    .to_str()
                    .is_some_and(|name| name.starts_with(CORRUPT_FILE_PREFIX))
            })
            .count();
        assert_eq!(corrupt_count, 1);

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn shutdown_flush_keeps_file_when_spacetime_is_unavailable() {
        let dir = test_dir("shutdown");
        let outbox = test_outbox(dir.clone(), 1024 * 1024);

        outbox.enqueue(sample_record("ledger-1")).await.unwrap();
        let result = outbox.flush_for_shutdown().await;

        assert!(
            matches!(result, Err(WalletRefundOutboxError::Spacetime(_))),
            "missing test SpacetimeDB should keep refund file for retry"
        );
        let pending_count = std::fs::read_dir(&dir)
            .unwrap()
            .filter_map(Result::ok)
            .filter(|entry| is_pending_outbox_file_name(&entry.file_name()))
            .count();
        assert_eq!(pending_count, 1);

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn flush_recovers_valid_crash_left_temp_file() {
        let dir = test_dir("recover-temp");
        std::fs::create_dir_all(&dir).unwrap();
        let outbox = test_outbox(dir.clone(), 1024 * 1024);
        let record = sample_record("ledger-temp");
        let temp_path = outbox.temp_path();
        std::fs::write(&temp_path, serde_json::to_vec(&record).unwrap()).unwrap();

        let result = outbox.flush_pending_files_once().await;

        assert!(matches!(result, Err(WalletRefundOutboxError::Spacetime(_))));
        assert!(!temp_path.exists());
        assert!(outbox.pending_path_for_ledger(&record.ledger_id).exists());

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn flush_recovers_crash_left_temp_file_into_overflow_when_capped() {
        let dir = test_dir("recover-temp-overflow");
        std::fs::create_dir_all(&dir).unwrap();
        let outbox = test_outbox(dir.clone(), 1);
        let existing = sample_record("ledger-existing");
        std::fs::write(
            outbox.pending_path_for_ledger(&existing.ledger_id),
            serde_json::to_vec(&existing).unwrap(),
        )
        .unwrap();
        let record = sample_record("ledger-temp-overflow");
        let temp_path = outbox.temp_path();
        std::fs::write(&temp_path, serde_json::to_vec(&record).unwrap()).unwrap();

        let result = outbox.flush_pending_files_once().await;

        assert!(matches!(result, Err(WalletRefundOutboxError::Spacetime(_))));
        assert!(!temp_path.exists());
        assert!(outbox.overflow_path_for_ledger(&record.ledger_id).exists());

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn flush_quarantines_conflicting_crash_left_temp_file() {
        let dir = test_dir("recover-conflicting-temp");
        std::fs::create_dir_all(&dir).unwrap();
        let outbox = test_outbox(dir.clone(), 1024 * 1024);
        let record = sample_record("ledger-temp-conflict");
        let mut conflicting = record.clone();
        conflicting.amount += 1;
        let pending_path = outbox.pending_path_for_ledger(&record.ledger_id);
        let temp_path = outbox.temp_path();
        std::fs::write(&pending_path, serde_json::to_vec(&conflicting).unwrap()).unwrap();
        std::fs::write(&temp_path, serde_json::to_vec(&record).unwrap()).unwrap();

        let result = outbox.flush_pending_files_once().await;

        assert!(matches!(result, Err(WalletRefundOutboxError::Spacetime(_))));
        assert!(!temp_path.exists());
        assert_eq!(
            read_refund_record(&pending_path).await.unwrap(),
            conflicting
        );
        let corrupt_count = std::fs::read_dir(&dir)
            .unwrap()
            .filter_map(Result::ok)
            .filter(|entry| {
                entry
                    .file_name()
                    .to_str()
                    .is_some_and(|name| name.starts_with(CORRUPT_FILE_PREFIX))
            })
            .count();
        assert_eq!(corrupt_count, 1);

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn flush_quarantines_corrupt_crash_left_temp_file() {
        let dir = test_dir("recover-corrupt-temp");
        std::fs::create_dir_all(&dir).unwrap();
        let outbox = test_outbox(dir.clone(), 1024 * 1024);
        let temp_path = outbox.temp_path();
        std::fs::write(&temp_path, b"{not-json}").unwrap();

        outbox.flush_pending_files_once().await.unwrap();

        assert!(!temp_path.exists());
        let corrupt_count = std::fs::read_dir(&dir)
            .unwrap()
            .filter_map(Result::ok)
            .filter(|entry| {
                entry
                    .file_name()
                    .to_str()
                    .is_some_and(|name| name.starts_with(CORRUPT_FILE_PREFIX))
            })
            .count();
        assert_eq!(corrupt_count, 1);

        let _ = std::fs::remove_dir_all(dir);
    }

    #[tokio::test]
    async fn worker_recovers_temp_file_immediately_on_startup() {
        let dir = test_dir("worker-startup");
        std::fs::create_dir_all(&dir).unwrap();
        let outbox = test_outbox(dir.clone(), 1024 * 1024);
        let record = sample_record("ledger-worker-startup");
        let temp_path = outbox.temp_path();
        std::fs::write(&temp_path, serde_json::to_vec(&record).unwrap()).unwrap();

        outbox.clone().spawn_worker();

        for _ in 0..100 {
            if !temp_path.exists() {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }

        assert!(!temp_path.exists());
        assert!(outbox.pending_path_for_ledger(&record.ledger_id).exists());

        let _ = std::fs::remove_dir_all(dir);
    }
}