From 426c20b476cafa191fbd1ba6f6b20bb3b69d8e2b Mon Sep 17 00:00:00 2001 From: kdletters Date: Thu, 27 Aug 2026 16:15:46 +0800 Subject: [PATCH] =?UTF-8?q?=E5=B0=86=E9=80=80=E6=AC=BE=20outbox=20?= =?UTF-8?q?=E4=B8=BB=E8=B7=AF=E5=BE=84=E8=BF=81=E5=85=A5=20SpacetimeDB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 profile_wallet_refund_outbox 表与事务处理 procedure 失败任务事务写库内 pending 记录，API 节点跨实例处理并重试 SpacetimeDB 不可达时保留本机 emergency spool 与 attempt 追踪 同步 runtime bindings、架构文档、运维恢复和项目决策记录 --- .../shared-memory/decision-log.md | 8 +- ...】server-rs与SpacetimeDB数据契约-2026-05-15.md | 9 +- ...发运维】本地开发验证与生产运维-2026-05-15.md | 2 +- .../crates/api-server/src/asset_billing.rs | 212 ++++++---- server-rs/crates/api-server/src/main.rs | 12 +- server-rs/crates/api-server/src/state.rs | 10 +- .../api-server/src/wallet_refund_outbox.rs | 106 ++++- .../crates/module-runtime/src/commands.rs | 53 +++ server-rs/crates/module-runtime/src/domain.rs | 32 ++ server-rs/crates/module-runtime/src/errors.rs | 4 + .../src/active/mapper/runtime_profile.rs | 47 +++ .../spacetime-client/src/active/runtime.rs | 60 +++ .../spacetime-client/src/module_bindings.rs | 38 ++ ...llet_refund_outbox_and_return_procedure.rs | 62 +++ ...llet_refund_outbox_and_return_procedure.rs | 62 +++ .../profile_wallet_refund_outbox_table.rs | 235 +++++++++++ .../profile_wallet_refund_outbox_type.rs | 103 +++++ ...wallet_refund_outbox_enqueue_input_type.rs | 23 ++ ...let_refund_outbox_procedure_result_type.rs | 20 + ...wallet_refund_outbox_process_input_type.rs | 16 + .../src/external_generation.rs | 4 + .../crates/spacetime-module/src/migration.rs | 1 + .../src/runtime/active/profile.rs | 381 +++++++++++++++++- 23 files changed, 1391 insertions(+), 109 deletions(-) create mode 100644 server-rs/crates/spacetime-client/src/module_bindings/enqueue_profile_wallet_refund_outbox_and_return_procedure.rs create mode 100644 server-rs/crates/spacetime-client/src/module_bindings/process_profile_wallet_refund_outbox_and_return_procedure.rs create mode 100644 server-rs/crates/spacetime-client/src/module_bindings/profile_wallet_refund_outbox_table.rs create mode 100644 server-rs/crates/spacetime-client/src/module_bindings/profile_wallet_refund_outbox_type.rs create mode 100644 server-rs/crates/spacetime-client/src/module_bindings/runtime_profile_wallet_refund_outbox_enqueue_input_type.rs create mode 100644 server-rs/crates/spacetime-client/src/module_bindings/runtime_profile_wallet_refund_outbox_procedure_result_type.rs create mode 100644 server-rs/crates/spacetime-client/src/module_bindings/runtime_profile_wallet_refund_outbox_process_input_type.rs diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index 731298ee0..bc69a4236 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -7752,4 +7752,10 @@ CI 上 `background_agent_runtime_recovers_stale_running_before_pending_task` 在 - 生产 Bearer 中间件的会话有效性改由 SpacetimeDB typed `validate_auth_session` procedure 在事务内校验 `user_account.token_version`、会话归属、撤销时间和过期时间；`InMemoryAuthStore` 仅保留启动恢复工作集及测试夹具，不作为生产请求鉴权读取源。该切片不等于登录、刷新、验证码和微信 state 的全量迁移，后续写路径仍需继续收口。 - `validate_auth_session`、认证投影导出和同步 procedure 都从 `ctx.sender()` 取调用方并要求现役 runtime service identity。`sync_auth_store_projection` 使用 API 工作集启动恢复或上次成功同步绑定的 `base_updated_at_micros` 做事务内 CAS，并要求 `updated_at_micros` 严格递增；基线不一致或版本不晚于当前值的整包快照失败关闭，只有确认本次同步前没有遗留未确认本地变更且同步期间无其它本地认证变更时才从正式表恢复，否则保留旧基线继续失败关闭，避免自动恢复覆盖并发未提交变更。同步成功但期间出现新本地变更时最多连续补同步三轮，仍未稳定则失败关闭，并保留待重试 revision。正式认证表仍是权威源，这只是迁移期跨 API 实例的延迟快照保护。 -- tracking outbox 与 wallet refund outbox worker 启动即执行恢复；退款 outbox 会恢复崩溃遗留 `tmp-*` 文件并隔离损坏 / 冲突文件。两者仍是节点本地 durable fallback，不能用粘性会话替代共享持久化；生产节点必须保留对应目录并纳入恢复演练。 +- tracking outbox worker 与 wallet refund emergency spool worker 启动即执行恢复；退款 emergency spool 会恢复崩溃遗留 `tmp-*` 文件并隔离损坏 / 冲突文件。tracking 仍是节点本地 durable fallback，退款的正式队列已迁入 SpacetimeDB；两者都不能用粘性会话替代共享持久化，生产节点必须保留对应目录并纳入恢复演练。 + +## 2026-08-27 退款 outbox 主路径迁入 SpacetimeDB + +- 决策：`profile_wallet_refund_outbox` 是跨 API 节点退款的正式持久化队列。扣费失败、外部生成 attempt 失败或最终 lease 过期时，在同一个 SpacetimeDB 事务内按 `refund_ledger_id` 幂等写入 pending 行；worker 从库内 pending 行批量处理，退款账本写入与 outbox 成功删除保持在同一事务内，失败由 `available_at` / `attempts` 驱动重试。`asset_operation_wallet_settlement` 继续负责退款先于 consume 可见时的取消 intent，阻止迟到扣费。只有 SpacetimeDB 完全不可达时，api-server 才写本机 `wallet-refund-outbox` emergency spool；本机文件不能替代库内队列，必须持久挂载、告警、恢复演练并支持人工补偿。 +- 影响范围：`profile_wallet_refund_outbox` 表及 bindings、runtime enqueue/process procedure、外部生成失败事务、inline 资产退款、api-server 跨节点 worker 和 emergency spool、后端架构与开发运维文档。 +- 验证方式：运行 `npm run spacetime:generate`、`npm run check:spacetime-schema`、`npm run check:server-rs-ddd`、`cargo check -p spacetime-module -p spacetime-client -p api-server --manifest-path server-rs/Cargo.toml`、退款 outbox / asset billing / external generation 定向测试、`npm run check:encoding` 和 `git diff --check`。 diff --git a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md index a7dc7d6dd..038ff0f32 100644 --- a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md +++ b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md @@ -232,7 +232,7 @@ npm run check:server-rs-ddd 3. 结果页单图重生成、发布、道具使用和其它独立资产操作仍按各自业务操作成本执行；不要把初始草稿成本误套到这些单次操作上。 4. 资产操作的预扣费必须 fail-closed：钱包或 SpacetimeDB 预扣费不可达、超时或返回业务错误时，`api-server` 直接返回错误，不允许继续调用图片、音频、GLB 等外部生成 provider。 5. 需要支持 HTTP retry 的计费 ledger id 必须包含当前请求的 `request_id`；前端 `fetchWithApiAuth` 同一次业务请求的静默刷新重试复用同一个 `x-request-id`，后端不得再使用 prompt 指纹或随机 asset id 作为扣费幂等键。 -6. 外部生成已预扣费但后续失败时必须先同步调用钱包退款；若 SpacetimeDB 暂不可用，退款请求写入 `wallet-refund-outbox` 本地文件并由后台 worker 重放。默认启用，配置项为 `GENARRATIVE_WALLET_REFUND_OUTBOX_ENABLED`、`GENARRATIVE_WALLET_REFUND_OUTBOX_DIR`、`GENARRATIVE_WALLET_REFUND_OUTBOX_BATCH_SIZE`、`GENARRATIVE_WALLET_REFUND_OUTBOX_FLUSH_INTERVAL_MS` 和 `GENARRATIVE_WALLET_REFUND_OUTBOX_MAX_BYTES`。outbox 文件按 refund ledger id 幂等落盘；成功重放后删除，坏文件隔离为 `corrupt-*`。外部生成任务触发的扣费和退款必须在 `profile_wallet_ledger.metadata_json` 中写入 `externalGenerationJobId`，outbox 重放也必须保留同一任务 ID，便于从退款记录追溯到正式生成任务。 +6. 外部生成已预扣费但后续失败时，失败/任务状态变更事务必须在 SpacetimeDB 内按 refund ledger id 幂等写入 `profile_wallet_refund_outbox` pending 行；跨节点 worker 从库内 pending 行批量处理并在库内事务执行退款，成功后删除 outbox 行，失败按 `available_at` 和 `attempts` 重试。当前 attempt 若 consume 尚不可见，事务仍必须先写 `asset_operation_wallet_settlement` 取消 intent，阻止迟到扣费。普通 inline 资产失败也先调用同一 DB outbox procedure；只有 SpacetimeDB 完全不可达时才写 `wallet-refund-outbox` 本机 emergency spool。默认启用，配置项为 `GENARRATIVE_WALLET_REFUND_OUTBOX_ENABLED`、`GENARRATIVE_WALLET_REFUND_OUTBOX_DIR`、`GENARRATIVE_WALLET_REFUND_OUTBOX_BATCH_SIZE`、`GENARRATIVE_WALLET_REFUND_OUTBOX_FLUSH_INTERVAL_MS` 和 `GENARRATIVE_WALLET_REFUND_OUTBOX_MAX_BYTES`。本机文件按 refund ledger id 幂等落盘；成功重放后删除，坏文件隔离为 `corrupt-*`，不能替代库内 outbox。外部生成任务触发的扣费和退款必须在 `profile_wallet_ledger.metadata_json` 与两类 outbox 中保留 `externalGenerationJobId` 和 `externalGenerationClaimAttempt`，便于从退款记录追溯到具体 attempt。 7. 拼图首图后台生成的跨实例互斥锁必须落在 SpacetimeDB `puzzle_background_compile_task` 表，claim id 由 `task_id + request_id` 构成，释放时必须校验 claim id，避免旧后台任务释放新请求抢到的租约。 ## 用户钱包与编辑器生成扣费契约 @@ -1057,6 +1057,13 @@ Responses 的终态载荷既是工具调用的恢复源，也是正文的恢复 - 说明：资产操作 consume/refund 配对结算事实表，主键为 consume ledger ID，并保存配对 refund ledger、用户、金额和结算时间。退款先到且 consume 尚不可见时，该表作为持久化取消 intent；迟到 consume 必须检测该行并拒绝扣费，避免 worker 崩溃重领期间双扣。 - 索引：主键 `consume_ledger_id`。 +### `profile_wallet_refund_outbox` + +- Rust 结构体：`ProfileWalletRefundOutbox` +- 源码：`server-rs/crates/spacetime-module/src/runtime/active/profile.rs` +- 说明：跨节点资产退款的正式 pending 队列。主键为 refund ledger ID，保存 consume/refund 配对、用户、金额、资源、生成任务 attempt、失败原因和重试时间；失败事务先写入该表，worker 在 SpacetimeDB 事务内幂等执行钱包退款并删除成功行。只有数据库不可达时，api-server 才使用本机 `wallet-refund-outbox` emergency spool。 +- 索引：`(status, available_at)`。 + ### `profile_wallet_config` - Rust 结构体：`ProfileWalletConfig` diff --git a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md index bfc5e04f8..ff5e6f725 100644 --- a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md +++ b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md @@ -855,7 +855,7 @@ GENARRATIVE_API_SHUTDOWN_OUTBOX_FLUSH_TIMEOUT_MS=5000 outbox 采用 NDJSON 文件保存原始事件。达到 `BATCH_SIZE` 时会立刻把当前 active 文件原子封存为 sealed 文件，并马上切到新的 active 继续写入；后台 worker 异步 flush sealed 文件，HTTP 请求线程不等待 SpacetimeDB。worker 启动时会先封存并 flush 已存在的 active / sealed 文件，恢复窗口内 SpacetimeDB 暂不可用则保留文件并按后续周期重试；`FLUSH_INTERVAL_MS` 只负责兜底封存长时间未满批的 active 文件。SpacetimeDB 批量 procedure 返回成功后删除 sealed 文件，失败则保留文件并重试。`MAX_BYTES` 是每个 outbox 实例的磁盘保护阈值，不是 flush 阈值；超过后低价值 route tracking 和 BgFilter provider 失败审计可以被丢弃并记录日志 / 指标，关键同步事件不进入该丢弃路径。api-server 使用配置目录本身，BgFilter worker 固定使用其 `bgfilter-worker/` 子目录，两个进程不得操作同一个 active 文件。sealed 文件若出现无法解析的坏行，会重命名为 `corrupt-*` 隔离并记录 `genarrative.tracking_outbox.files.corrupt` 指标，避免一个坏文件阻塞后续批量入库。进程收到退出信号后会在 `GENARRATIVE_API_SHUTDOWN_OUTBOX_FLUSH_TIMEOUT_MS` 窗口内封存各自 active 文件并尽力 flush sealed 文件，超时或 SpacetimeDB 暂不可用时保留本地文件给下次同角色启动继续投递。该机制对已 enqueue 记录提供至少一次投递语义，依赖 `tracking_event.event_id` 幂等跳过重复事件；BgFilter 尚未 enqueue 或因硬上限 / 保护阈值被丢弃的审计不在该保证内。 -钱包退款 outbox 同样在 worker 启动时立即扫描 pending 文件；如果进程在“临时文件写完但尚未改名”阶段崩溃，启动恢复会校验 `tmp-*` 内容并原子提升为按 ledger id 命名的 pending 文件，损坏或冲突文件移入 `corrupt-*` 隔离目录，不会静默丢弃关键退款。该目录仍是节点本地 durable fallback，不能替代 SpacetimeDB 钱包账本；发布和主机替换必须保留 `/var/lib/genarrative/wallet-refund-outbox` 并纳入节点恢复 / 备份演练。 +钱包退款正式 pending 队列在 SpacetimeDB 的 `profile_wallet_refund_outbox` 表中，由每个 API 节点的 worker 共同处理；worker 启动即扫描库内 pending 行，成功在同一事务内写钱包账本并删除 outbox 行，失败按库内 `available_at` / `attempts` 重试。只有 SpacetimeDB 完全不可达时才写本机 `wallet-refund-outbox` emergency spool；如果进程在“临时文件写完但尚未改名”阶段崩溃，启动恢复会校验 `tmp-*` 内容并原子提升为按 ledger id 命名的 pending 文件，损坏或冲突文件移入 `corrupt-*` 隔离目录，不会静默丢失关键退款。worker 连接失败、库内 retry、emergency spool 写入 / 容量失败和 `corrupt-*` 出现都必须接入告警；人工补偿先按 refund ledger id 对账 `profile_wallet_ledger`、`asset_operation_wallet_settlement` 与两类 outbox，再通过受控退款 procedure 幂等重放，禁止直接手写钱包表。该目录不能替代库内 outbox；发布和主机替换必须保留 `/var/lib/genarrative/wallet-refund-outbox` 并纳入节点恢复 / 备份演练。 release 机器如果日志每秒刷 `tracking outbox ... Permission denied (os error 13)`，先检查 `/etc/genarrative/api-server.env` 是否缺少 `GENARRATIVE_TRACKING_OUTBOX_DIR`。缺少时 `api-server` 会回退到本地开发默认相对路径 `server-rs/.data/tracking-outbox`，而 systemd 的工作目录是只读发布目录 `/opt/genarrative/releases/`，`genarrative` 用户无法在其中创建 `server-rs`。修复顺序： diff --git a/server-rs/crates/api-server/src/asset_billing.rs b/server-rs/crates/api-server/src/asset_billing.rs index 11e7369eb..83545d528 100644 --- a/server-rs/crates/api-server/src/asset_billing.rs +++ b/server-rs/crates/api-server/src/asset_billing.rs @@ -499,87 +499,127 @@ async fn refund_asset_operation_points_with_job_id( external_generation_claim_attempt: Option, ) -> Result<(), AppError> { let created_at_micros = current_utc_micros(); - let metadata_json = wallet_metadata_json( - external_generation_job_id.as_deref(), + let current_attempt_is_owned_by_failure_transaction = + external_generation_job_id.as_deref().is_some_and(|job_id| { + current_external_generation_billing_context().is_some_and(|context| { + context.job_id == job_id + && Some(context.claim_attempt) == external_generation_claim_attempt + }) + }); + if current_attempt_is_owned_by_failure_transaction { + // 队列当前 attempt 的 refund 由 fail_external_generation_job transaction 原子写入 + // SpacetimeDB outbox；这里不能先写另一笔独立退款，避免任务成功写回后被误退。 + return Ok(()); + } + let settlement_reason = if external_generation_job_id.is_some() { + "stale_attempt_recovery" + } else { + "asset_operation_failed" + }; + let enqueue_input = module_runtime::build_runtime_profile_wallet_refund_outbox_enqueue_input( + owner_user_id.clone(), + points_cost, + ledger_id.clone(), + created_at_micros, + asset_kind.clone(), + asset_id.clone(), + settlement_reason.to_string(), + external_generation_job_id.clone(), external_generation_claim_attempt, - ); - let result = state + ) + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR).with_details(json!({ + "provider": "profile-wallet-refund-outbox", + "message": error.to_string(), + })) + })?; + let enqueue_result = state .spacetime_client() - .refund_profile_wallet_points_with_metadata( - owner_user_id.clone(), - points_cost, - ledger_id.clone(), - created_at_micros, - metadata_json, - ) + .enqueue_profile_wallet_refund_outbox(enqueue_input) .await; - if let Err(error) = result { - let refund_error = error.to_string(); - let app_error = map_asset_operation_wallet_error(error); - if let Some(outbox) = state.wallet_refund_outbox() { - match outbox - .enqueue(WalletRefundOutboxRecord { - owner_user_id: owner_user_id.clone(), - amount: points_cost, - ledger_id: ledger_id.clone(), - created_at_micros, - asset_kind: asset_kind.clone(), - asset_id: asset_id.clone(), - external_generation_job_id: external_generation_job_id.clone(), - }) - .await - { - Ok(WalletRefundOutboxEnqueueOutcome::Enqueued) => { - tracing::warn!( - owner_user_id, - asset_kind, - asset_id, - external_generation_job_id, - ledger_id, - error = %refund_error, - "资产操作失败后的泥点退款立即执行失败，已写入 wallet refund outbox" - ); - } - Ok(WalletRefundOutboxEnqueueOutcome::Dropped { reason }) => { - tracing::error!( - owner_user_id, - asset_kind, - asset_id, - external_generation_job_id, - ledger_id, - reason, - error = %refund_error, - "资产操作失败后的泥点退款立即执行失败，且 wallet refund outbox 因容量限制丢弃" - ); - } - Err(outbox_error) => { - tracing::error!( - owner_user_id, - asset_kind, - asset_id, - external_generation_job_id, - ledger_id, - refund_error = %refund_error, - outbox_error = %outbox_error, - "资产操作失败后的泥点退款立即执行失败，且写入 wallet refund outbox 失败" - ); - } - } - } else { - tracing::error!( + match enqueue_result { + Ok(_) => { + tracing::info!( owner_user_id, asset_kind, asset_id, external_generation_job_id, external_generation_claim_attempt, ledger_id, - error = %refund_error, - "资产操作失败后的泥点退款失败，且 wallet refund outbox 未启用" + "资产操作失败后的泥点退款已写入 SpacetimeDB refund outbox" ); + Ok(()) } - return Err(app_error); + Err(error) if should_use_wallet_refund_emergency_spool(&error) => { + let refund_error = error.to_string(); + let app_error = map_asset_operation_wallet_error(error); + if let Some(outbox) = state.wallet_refund_outbox() { + match outbox + .enqueue(WalletRefundOutboxRecord { + owner_user_id: owner_user_id.clone(), + amount: points_cost, + ledger_id: ledger_id.clone(), + created_at_micros, + asset_kind: asset_kind.clone(), + asset_id: asset_id.clone(), + settlement_reason: settlement_reason.to_string(), + external_generation_job_id: external_generation_job_id.clone(), + external_generation_claim_attempt, + }) + .await + { + Ok(WalletRefundOutboxEnqueueOutcome::Enqueued) => { + tracing::warn!( + owner_user_id, + asset_kind, + asset_id, + external_generation_job_id, + ledger_id, + error = %refund_error, + "SpacetimeDB refund outbox 不可达，已写入本机 emergency spool" + ); + } + Ok(WalletRefundOutboxEnqueueOutcome::Dropped { reason }) => { + tracing::error!( + owner_user_id, + asset_kind, + asset_id, + external_generation_job_id, + ledger_id, + reason, + error = %refund_error, + "SpacetimeDB refund outbox 不可达，且本机 emergency spool 因容量限制丢弃" + ); + } + Err(outbox_error) => { + tracing::error!( + owner_user_id, + asset_kind, + asset_id, + external_generation_job_id, + ledger_id, + refund_error = %refund_error, + outbox_error = %outbox_error, + "SpacetimeDB refund outbox 不可达，且写入本机 emergency spool 失败" + ); + } + } + } else { + tracing::error!( + owner_user_id, + asset_kind, + asset_id, + external_generation_job_id, + external_generation_claim_attempt, + ledger_id, + error = %refund_error, + "SpacetimeDB refund outbox 不可达，且本机 emergency spool 未启用" + ); + } + Err(app_error) + } + Err(error) => Err(map_asset_operation_wallet_error(error)), } - Ok(()) } fn current_external_generation_billing_context() -> Option { @@ -683,6 +723,22 @@ pub(crate) fn should_skip_asset_operation_billing_for_connectivity( } } +fn should_use_wallet_refund_emergency_spool(error: &SpacetimeClientError) -> bool { + match error { + SpacetimeClientError::ConnectDropped | SpacetimeClientError::Timeout(_) => true, + SpacetimeClientError::Build(message) + | SpacetimeClientError::Procedure(message) + | SpacetimeClientError::Runtime(message) => { + message.contains("503") + || message.contains("Service Unavailable") + || message.contains("Failed to connect") + || message.contains("WebSocket") + || message.contains("连接已断开") + || message.contains("连接在返回结果前已断开") + } + } +} + fn current_utc_micros() -> i64 { time::OffsetDateTime::now_utc().unix_timestamp_nanos() as i64 / 1_000 } @@ -838,6 +894,24 @@ mod tests { )); } + #[test] + fn wallet_refund_emergency_spool_requires_database_unavailability() { + assert!(should_use_wallet_refund_emergency_spool( + &SpacetimeClientError::ConnectDropped + )); + assert!(should_use_wallet_refund_emergency_spool( + &SpacetimeClientError::Runtime("503 Service Unavailable".to_string()) + )); + assert!(!should_use_wallet_refund_emergency_spool( + &SpacetimeClientError::Procedure( + "No such procedure: enqueue_profile_wallet_refund_outbox_and_return".to_string(), + ) + )); + assert!(!should_use_wallet_refund_emergency_spool( + &SpacetimeClientError::Procedure("泥点余额不足".to_string()) + )); + } + #[test] fn asset_operation_wallet_insufficient_balance_is_public_message() { for domain_message in [ diff --git a/server-rs/crates/api-server/src/main.rs b/server-rs/crates/api-server/src/main.rs index 1d73da766..441936e1e 100644 --- a/server-rs/crates/api-server/src/main.rs +++ b/server-rs/crates/api-server/src/main.rs @@ -529,21 +529,24 @@ async fn finalize_shutdown(context: ShutdownContext) { } if let Some(outbox) = context.wallet_refund_outbox { - info!(timeout_ms, "api-server 退出前 flush wallet refund outbox"); + info!( + timeout_ms, + "api-server 退出前 flush wallet refund emergency spool" + ); match timeout(context.outbox_flush_timeout, outbox.flush_for_shutdown()).await { Ok(Ok(())) => { - info!("api-server 退出前 wallet refund outbox flush 完成"); + info!("api-server 退出前 wallet refund emergency spool flush 完成"); } Ok(Err(error)) => { warn!( error = %error, - "api-server 退出前 wallet refund outbox flush 未完成，已保留本地文件等待下次启动重试" + "api-server 退出前 wallet refund emergency spool flush 未完成，已保留本地文件等待下次启动重试" ); } Err(_) => { warn!( timeout_ms, - "api-server 退出前 wallet refund outbox flush 超时，已保留本地文件等待下次启动重试" + "api-server 退出前 wallet refund emergency spool flush 超时，已保留本地文件等待下次启动重试" ); } } @@ -557,6 +560,7 @@ fn spawn_common_app_state_background_workers(state: &AppState) { if let Some(outbox) = state.wallet_refund_outbox() { outbox.spawn_worker(); } + state.profile_wallet_refund_outbox_worker().spawn_worker(); } fn spawn_http_app_state_background_workers(state: &AppState, process_role: ProcessRole) { diff --git a/server-rs/crates/api-server/src/state.rs b/server-rs/crates/api-server/src/state.rs index fc82fab00..600a1d0e1 100644 --- a/server-rs/crates/api-server/src/state.rs +++ b/server-rs/crates/api-server/src/state.rs @@ -45,7 +45,7 @@ use crate::editor_generation_config::{ EditorGenerationPricingUnit, }; use crate::tracking_outbox::TrackingOutbox; -use crate::wallet_refund_outbox::WalletRefundOutbox; +use crate::wallet_refund_outbox::{ProfileWalletRefundOutboxWorker, WalletRefundOutbox}; use crate::wechat::pay::{build_wechat_pay_config, map_wechat_pay_init_error}; use crate::wechat::provider::build_wechat_provider; use crate::work_author::{ @@ -297,6 +297,7 @@ pub struct AppStateInner { puzzle_gallery_cache: PuzzleGalleryCache, tracking_outbox: Option>, wallet_refund_outbox: Option>, + profile_wallet_refund_outbox_worker: Arc, editor_generation_pricing_store: EditorGenerationPricingStore, llm_client: Option, vector_engine_llm_client: Option, @@ -586,6 +587,8 @@ impl AppState { let tracking_outbox = TrackingOutbox::from_config(&config, spacetime_client.clone()); let wallet_refund_outbox = WalletRefundOutbox::from_config(&config, spacetime_client.clone()); + let profile_wallet_refund_outbox_worker = + ProfileWalletRefundOutboxWorker::from_config(&config, spacetime_client.clone()); let editor_generation_pricing_store = EditorGenerationPricingStore::load( config.editor_generation_pricing_override_path.clone(), ) @@ -656,6 +659,7 @@ impl AppState { puzzle_gallery_cache: PuzzleGalleryCache::new(), tracking_outbox, wallet_refund_outbox, + profile_wallet_refund_outbox_worker, editor_generation_pricing_store, llm_client, vector_engine_llm_client, @@ -1508,6 +1512,10 @@ impl AppState { self.wallet_refund_outbox.clone() } + pub fn profile_wallet_refund_outbox_worker(&self) -> Arc { + self.profile_wallet_refund_outbox_worker.clone() + } + pub fn llm_client(&self) -> Option<&LlmClient> { self.llm_client.as_ref() } diff --git a/server-rs/crates/api-server/src/wallet_refund_outbox.rs b/server-rs/crates/api-server/src/wallet_refund_outbox.rs index 64ae5d884..b260ad1c3 100644 --- a/server-rs/crates/api-server/src/wallet_refund_outbox.rs +++ b/server-rs/crates/api-server/src/wallet_refund_outbox.rs @@ -34,6 +34,71 @@ pub struct WalletRefundOutbox { flush_notify: Arc, } +#[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 { + 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) { + 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, @@ -42,8 +107,12 @@ pub(crate) struct WalletRefundOutboxRecord { 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, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub external_generation_claim_attempt: Option, } #[derive(Debug)] @@ -250,15 +319,26 @@ impl WalletRefundOutbox { Err(error) => return Err(error), }; - match self - .spacetime_client - .refund_profile_wallet_points_with_metadata( + 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, - refund_metadata_json(record.external_generation_job_id.as_deref()), + record.asset_kind.clone(), + record.asset_id.clone(), + record.settlement_reason.clone(), + record.external_generation_job_id.clone(), + record.external_generation_claim_attempt, ) + .map_err(|error| { + WalletRefundOutboxError::Spacetime(SpacetimeClientError::Runtime( + error.to_string(), + )) + })?; + match self + .spacetime_client + .enqueue_profile_wallet_refund_outbox(enqueue_input) .await { Ok(_) => { @@ -275,7 +355,7 @@ impl WalletRefundOutbox { asset_id = %record.asset_id, external_generation_job_id = ?record.external_generation_job_id, path = %path.display(), - "wallet refund outbox 退款已重放并删除文件" + "wallet refund emergency spool 已恢复到 SpacetimeDB outbox 并删除文件" ); } Err(error) => return Err(WalletRefundOutboxError::Spacetime(error)), @@ -491,18 +571,8 @@ impl WalletRefundOutbox { } } -fn refund_metadata_json(external_generation_job_id: Option<&str>) -> String { - let Some(external_generation_job_id) = external_generation_job_id - .map(str::trim) - .filter(|value| !value.is_empty()) - else { - return module_runtime::PROFILE_INVITE_CODE_METADATA_DEFAULT_JSON.to_string(); - }; - - serde_json::json!({ - "externalGenerationJobId": external_generation_job_id, - }) - .to_string() +fn default_settlement_reason() -> String { + "emergency_spool_replay".to_string() } impl fmt::Debug for WalletRefundOutbox { @@ -621,7 +691,9 @@ mod tests { 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), } } diff --git a/server-rs/crates/module-runtime/src/commands.rs b/server-rs/crates/module-runtime/src/commands.rs index d282cb10e..0fc648db8 100644 --- a/server-rs/crates/module-runtime/src/commands.rs +++ b/server-rs/crates/module-runtime/src/commands.rs @@ -285,6 +285,59 @@ pub fn build_runtime_profile_wallet_adjustment_input_with_metadata( }) } +pub fn build_runtime_profile_wallet_refund_outbox_enqueue_input( + owner_user_id: String, + amount: u64, + refund_ledger_id: String, + created_at_micros: i64, + asset_kind: String, + asset_id: String, + settlement_reason: String, + external_generation_job_id: Option, + external_generation_claim_attempt: Option, +) -> Result { + let adjustment = build_runtime_profile_wallet_adjustment_input( + owner_user_id, + amount, + refund_ledger_id, + created_at_micros, + )?; + let asset_kind = + normalize_required_string(asset_kind).ok_or(RuntimeProfileFieldError::MissingLedgerId)?; + let asset_id = + normalize_required_string(asset_id).ok_or(RuntimeProfileFieldError::MissingLedgerId)?; + let settlement_reason = normalize_required_string(settlement_reason) + .ok_or(RuntimeProfileFieldError::MissingLedgerId)?; + let external_generation_job_id = + external_generation_job_id.and_then(|value| normalize_required_string(value)); + if external_generation_job_id.is_some() != external_generation_claim_attempt.is_some() { + return Err(RuntimeProfileFieldError::InvalidExternalGenerationAttempt); + } + Ok(RuntimeProfileWalletRefundOutboxEnqueueInput { + owner_user_id: adjustment.user_id, + amount: adjustment.amount, + refund_ledger_id: adjustment.ledger_id, + created_at_micros: adjustment.created_at_micros, + asset_kind, + asset_id, + settlement_reason, + external_generation_job_id, + external_generation_claim_attempt, + }) +} + +pub fn build_runtime_profile_wallet_refund_outbox_process_input( + worker_id: String, + limit: u32, +) -> Result { + let worker_id = + normalize_required_string(worker_id).ok_or(RuntimeProfileFieldError::MissingLedgerId)?; + if limit == 0 { + return Err(RuntimeProfileFieldError::InvalidWalletAmount); + } + Ok(RuntimeProfileWalletRefundOutboxProcessInput { worker_id, limit }) +} + pub fn build_runtime_profile_recharge_center_get_input( user_id: String, ) -> Result { diff --git a/server-rs/crates/module-runtime/src/domain.rs b/server-rs/crates/module-runtime/src/domain.rs index d4334fc1f..7a4a8c928 100644 --- a/server-rs/crates/module-runtime/src/domain.rs +++ b/server-rs/crates/module-runtime/src/domain.rs @@ -2077,6 +2077,38 @@ pub struct RuntimeProfileWalletAdjustmentProcedureResult { pub error_message: Option, } +#[cfg_attr(feature = "spacetime-types", derive(SpacetimeType))] +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct RuntimeProfileWalletRefundOutboxEnqueueInput { + pub owner_user_id: String, + pub amount: u64, + pub refund_ledger_id: String, + pub created_at_micros: i64, + pub asset_kind: String, + pub asset_id: String, + pub settlement_reason: String, + pub external_generation_job_id: Option, + pub external_generation_claim_attempt: Option, +} + +#[cfg_attr(feature = "spacetime-types", derive(SpacetimeType))] +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct RuntimeProfileWalletRefundOutboxProcessInput { + pub worker_id: String, + pub limit: u32, +} + +#[cfg_attr(feature = "spacetime-types", derive(SpacetimeType))] +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct RuntimeProfileWalletRefundOutboxProcedureResult { + pub ok: bool, + pub enqueued_count: u32, + pub processed_count: u32, + pub retry_count: u32, + pub failed_count: u32, + pub error_message: Option, +} + #[cfg_attr(feature = "spacetime-types", derive(SpacetimeType))] #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct RuntimeProfileWalletLedgerListInput { diff --git a/server-rs/crates/module-runtime/src/errors.rs b/server-rs/crates/module-runtime/src/errors.rs index 74e1924ed..e8f1f0457 100644 --- a/server-rs/crates/module-runtime/src/errors.rs +++ b/server-rs/crates/module-runtime/src/errors.rs @@ -82,6 +82,7 @@ pub enum RuntimeProfileFieldError { TaskNotClaimable, TaskAlreadyClaimed, MissingWorkerId, + InvalidExternalGenerationAttempt, MissingOrderId, MissingProductId, MissingProductTitle, @@ -172,6 +173,9 @@ impl std::fmt::Display for RuntimeProfileFieldError { Self::TaskNotClaimable => f.write_str("任务尚未达成"), Self::TaskAlreadyClaimed => f.write_str("任务奖励已领取"), Self::MissingWorkerId => f.write_str("worker_id 不能为空"), + Self::InvalidExternalGenerationAttempt => { + f.write_str("external_generation_job_id 与 claim_attempt 必须成对提供") + } Self::MissingOrderId => f.write_str("recharge.order_id 不能为空"), Self::MissingProductId => f.write_str("recharge.product_id 不能为空"), Self::MissingProductTitle => f.write_str("recharge.product_title 不能为空"), diff --git a/server-rs/crates/spacetime-client/src/active/mapper/runtime_profile.rs b/server-rs/crates/spacetime-client/src/active/mapper/runtime_profile.rs index 1077b5b0c..4573466b9 100644 --- a/server-rs/crates/spacetime-client/src/active/mapper/runtime_profile.rs +++ b/server-rs/crates/spacetime-client/src/active/mapper/runtime_profile.rs @@ -117,6 +117,35 @@ impl From } } +impl From + for RuntimeProfileWalletRefundOutboxEnqueueInput +{ + fn from(input: module_runtime::RuntimeProfileWalletRefundOutboxEnqueueInput) -> Self { + Self { + owner_user_id: input.owner_user_id, + amount: input.amount, + refund_ledger_id: input.refund_ledger_id, + created_at_micros: input.created_at_micros, + asset_kind: input.asset_kind, + asset_id: input.asset_id, + settlement_reason: input.settlement_reason, + external_generation_job_id: input.external_generation_job_id, + external_generation_claim_attempt: input.external_generation_claim_attempt, + } + } +} + +impl From + for RuntimeProfileWalletRefundOutboxProcessInput +{ + fn from(input: module_runtime::RuntimeProfileWalletRefundOutboxProcessInput) -> Self { + Self { + worker_id: input.worker_id, + limit: input.limit, + } + } +} + impl From for RuntimeProfileRechargeOrderGetInput { @@ -677,6 +706,24 @@ pub(crate) fn map_runtime_profile_wallet_adjustment_procedure_result( )) } +pub(crate) fn map_runtime_profile_wallet_refund_outbox_procedure_result( + result: RuntimeProfileWalletRefundOutboxProcedureResult, +) -> Result { + if !result.ok { + return Err(SpacetimeClientError::procedure_failed(result.error_message)); + } + Ok( + module_runtime::RuntimeProfileWalletRefundOutboxProcedureResult { + ok: true, + enqueued_count: result.enqueued_count, + processed_count: result.processed_count, + retry_count: result.retry_count, + failed_count: result.failed_count, + error_message: result.error_message, + }, + ) +} + pub(crate) fn map_runtime_profile_recharge_center_procedure_result( result: RuntimeProfileRechargeCenterProcedureResult, ) -> Result { diff --git a/server-rs/crates/spacetime-client/src/active/runtime.rs b/server-rs/crates/spacetime-client/src/active/runtime.rs index f11c2cb89..119393fb1 100644 --- a/server-rs/crates/spacetime-client/src/active/runtime.rs +++ b/server-rs/crates/spacetime-client/src/active/runtime.rs @@ -312,6 +312,66 @@ impl SpacetimeClient { .await } + pub async fn enqueue_profile_wallet_refund_outbox( + &self, + input: module_runtime::RuntimeProfileWalletRefundOutboxEnqueueInput, + ) -> Result + { + let procedure_input: RuntimeProfileWalletRefundOutboxEnqueueInput = input.into(); + self.call_after_connect( + "enqueue_profile_wallet_refund_outbox_and_return", + move |connection, sender| { + connection + .procedures() + .enqueue_profile_wallet_refund_outbox_and_return_then( + procedure_input, + move |_, result| { + let mapped = result + .map_err(SpacetimeClientError::from_sdk_error) + .and_then( + map_runtime_profile_wallet_refund_outbox_procedure_result, + ); + send_once(&sender, mapped); + }, + ); + }, + ) + .await + } + + pub async fn process_profile_wallet_refund_outbox( + &self, + worker_id: String, + limit: u32, + ) -> Result + { + let procedure_input = + module_runtime::build_runtime_profile_wallet_refund_outbox_process_input( + worker_id, limit, + ) + .map_err(SpacetimeClientError::validation_failed)? + .into(); + self.call_after_connect( + "process_profile_wallet_refund_outbox_and_return", + move |connection, sender| { + connection + .procedures() + .process_profile_wallet_refund_outbox_and_return_then( + procedure_input, + move |_, result| { + let mapped = result + .map_err(SpacetimeClientError::from_sdk_error) + .and_then( + map_runtime_profile_wallet_refund_outbox_procedure_result, + ); + send_once(&sender, mapped); + }, + ); + }, + ) + .await + } + pub async fn get_profile_recharge_center( &self, user_id: String, diff --git a/server-rs/crates/spacetime-client/src/module_bindings.rs b/server-rs/crates/spacetime-client/src/module_bindings.rs index f808362ee..a8a42efb0 100644 --- a/server-rs/crates/spacetime-client/src/module_bindings.rs +++ b/server-rs/crates/spacetime-client/src/module_bindings.rs @@ -381,6 +381,7 @@ pub mod editor_spritesheet_slice_batch_persist_result_type; pub mod editor_spritesheet_slice_persist_item_input_type; pub mod editor_spritesheet_slice_persisted_item_type; pub mod enqueue_external_generation_job_and_return_procedure; +pub mod enqueue_profile_wallet_refund_outbox_and_return_procedure; pub mod ensure_analytics_date_dimension_for_date_reducer; pub mod expire_profile_recharge_order_timer_reducer; pub mod export_auth_store_projection_from_tables_procedure; @@ -518,6 +519,7 @@ pub mod preflight_editor_generation_target_and_return_procedure; pub mod preflight_editor_pixel_art_result_and_return_procedure; pub mod prepare_profile_recharge_refund_hold_and_return_procedure; pub mod preview_profile_recharge_refund_hold_and_return_procedure; +pub mod process_profile_wallet_refund_outbox_and_return_procedure; pub mod profile_code_operation_table; pub mod profile_code_operation_type; pub mod profile_daily_free_points_table; @@ -572,6 +574,8 @@ pub mod profile_wallet_ledger_table; pub mod profile_wallet_ledger_type; pub mod profile_wallet_manual_restriction_table; pub mod profile_wallet_manual_restriction_type; +pub mod profile_wallet_refund_outbox_table; +pub mod profile_wallet_refund_outbox_type; pub mod public_work_like_table; pub mod public_work_like_type; pub mod public_work_play_daily_stat_table; @@ -780,6 +784,9 @@ pub mod runtime_profile_wallet_ledger_procedure_result_type; pub mod runtime_profile_wallet_ledger_source_type_type; pub mod runtime_profile_wallet_manual_restriction_snapshot_type; pub mod runtime_profile_wallet_manual_restriction_upsert_input_type; +pub mod runtime_profile_wallet_refund_outbox_enqueue_input_type; +pub mod runtime_profile_wallet_refund_outbox_procedure_result_type; +pub mod runtime_profile_wallet_refund_outbox_process_input_type; pub mod runtime_referral_invite_center_get_input_type; pub mod runtime_referral_invite_center_procedure_result_type; pub mod runtime_referral_invite_center_snapshot_type; @@ -1243,6 +1250,7 @@ pub use editor_spritesheet_slice_batch_persist_result_type::EditorSpritesheetSli pub use editor_spritesheet_slice_persist_item_input_type::EditorSpritesheetSlicePersistItemInput; pub use editor_spritesheet_slice_persisted_item_type::EditorSpritesheetSlicePersistedItem; pub use enqueue_external_generation_job_and_return_procedure::enqueue_external_generation_job_and_return; +pub use enqueue_profile_wallet_refund_outbox_and_return_procedure::enqueue_profile_wallet_refund_outbox_and_return; pub use ensure_analytics_date_dimension_for_date_reducer::ensure_analytics_date_dimension_for_date; pub use expire_profile_recharge_order_timer_reducer::expire_profile_recharge_order_timer; pub use export_auth_store_projection_from_tables_procedure::export_auth_store_projection_from_tables; @@ -1380,6 +1388,7 @@ pub use preflight_editor_generation_target_and_return_procedure::preflight_edito pub use preflight_editor_pixel_art_result_and_return_procedure::preflight_editor_pixel_art_result_and_return; pub use prepare_profile_recharge_refund_hold_and_return_procedure::prepare_profile_recharge_refund_hold_and_return; pub use preview_profile_recharge_refund_hold_and_return_procedure::preview_profile_recharge_refund_hold_and_return; +pub use process_profile_wallet_refund_outbox_and_return_procedure::process_profile_wallet_refund_outbox_and_return; pub use profile_code_operation_table::*; pub use profile_code_operation_type::ProfileCodeOperation; pub use profile_daily_free_points_table::*; @@ -1434,6 +1443,8 @@ pub use profile_wallet_ledger_table::*; pub use profile_wallet_ledger_type::ProfileWalletLedger; pub use profile_wallet_manual_restriction_table::*; pub use profile_wallet_manual_restriction_type::ProfileWalletManualRestriction; +pub use profile_wallet_refund_outbox_table::*; +pub use profile_wallet_refund_outbox_type::ProfileWalletRefundOutbox; pub use public_work_like_table::*; pub use public_work_like_type::PublicWorkLike; pub use public_work_play_daily_stat_table::*; @@ -1642,6 +1653,9 @@ pub use runtime_profile_wallet_ledger_procedure_result_type::RuntimeProfileWalle pub use runtime_profile_wallet_ledger_source_type_type::RuntimeProfileWalletLedgerSourceType; pub use runtime_profile_wallet_manual_restriction_snapshot_type::RuntimeProfileWalletManualRestrictionSnapshot; pub use runtime_profile_wallet_manual_restriction_upsert_input_type::RuntimeProfileWalletManualRestrictionUpsertInput; +pub use runtime_profile_wallet_refund_outbox_enqueue_input_type::RuntimeProfileWalletRefundOutboxEnqueueInput; +pub use runtime_profile_wallet_refund_outbox_procedure_result_type::RuntimeProfileWalletRefundOutboxProcedureResult; +pub use runtime_profile_wallet_refund_outbox_process_input_type::RuntimeProfileWalletRefundOutboxProcessInput; pub use runtime_referral_invite_center_get_input_type::RuntimeReferralInviteCenterGetInput; pub use runtime_referral_invite_center_procedure_result_type::RuntimeReferralInviteCenterProcedureResult; pub use runtime_referral_invite_center_snapshot_type::RuntimeReferralInviteCenterSnapshot; @@ -1943,6 +1957,7 @@ pub struct DbUpdate { profile_wallet_consumption_total: __sdk::TableUpdate, profile_wallet_ledger: __sdk::TableUpdate, profile_wallet_manual_restriction: __sdk::TableUpdate, + profile_wallet_refund_outbox: __sdk::TableUpdate, public_work_like: __sdk::TableUpdate, public_work_play_daily_stat: __sdk::TableUpdate, puzzle_agent_message: __sdk::TableUpdate, @@ -2346,6 +2361,9 @@ impl TryFrom<__ws::v2::TransactionUpdate> for DbUpdate { profile_wallet_manual_restriction_table::parse_table_update(table_update)?, ) } + "profile_wallet_refund_outbox" => db_update.profile_wallet_refund_outbox.append( + profile_wallet_refund_outbox_table::parse_table_update(table_update)?, + ), "public_work_like" => db_update .public_work_like .append(public_work_like_table::parse_table_update(table_update)?), @@ -3036,6 +3054,12 @@ impl __sdk::DbUpdate for DbUpdate { &self.profile_wallet_manual_restriction, ) .with_updates_by_pk(|row| &row.user_id); + diff.profile_wallet_refund_outbox = cache + .apply_diff_to_table::( + "profile_wallet_refund_outbox", + &self.profile_wallet_refund_outbox, + ) + .with_updates_by_pk(|row| &row.refund_ledger_id); diff.public_work_like = cache .apply_diff_to_table::("public_work_like", &self.public_work_like) .with_updates_by_pk(|row| &row.like_id); @@ -3531,6 +3555,9 @@ impl __sdk::DbUpdate for DbUpdate { "profile_wallet_manual_restriction" => db_update .profile_wallet_manual_restriction .append(__sdk::parse_row_list_as_inserts(table_rows.rows)?), + "profile_wallet_refund_outbox" => db_update + .profile_wallet_refund_outbox + .append(__sdk::parse_row_list_as_inserts(table_rows.rows)?), "public_work_like" => db_update .public_work_like .append(__sdk::parse_row_list_as_inserts(table_rows.rows)?), @@ -3961,6 +3988,9 @@ impl __sdk::DbUpdate for DbUpdate { "profile_wallet_manual_restriction" => db_update .profile_wallet_manual_restriction .append(__sdk::parse_row_list_as_deletes(table_rows.rows)?), + "profile_wallet_refund_outbox" => db_update + .profile_wallet_refund_outbox + .append(__sdk::parse_row_list_as_deletes(table_rows.rows)?), "public_work_like" => db_update .public_work_like .append(__sdk::parse_row_list_as_deletes(table_rows.rows)?), @@ -4202,6 +4232,7 @@ pub struct AppliedDiff<'r> { profile_wallet_consumption_total: __sdk::TableAppliedDiff<'r, ProfileWalletConsumptionTotal>, profile_wallet_ledger: __sdk::TableAppliedDiff<'r, ProfileWalletLedger>, profile_wallet_manual_restriction: __sdk::TableAppliedDiff<'r, ProfileWalletManualRestriction>, + profile_wallet_refund_outbox: __sdk::TableAppliedDiff<'r, ProfileWalletRefundOutbox>, public_work_like: __sdk::TableAppliedDiff<'r, PublicWorkLike>, public_work_play_daily_stat: __sdk::TableAppliedDiff<'r, PublicWorkPlayDailyStat>, puzzle_agent_message: __sdk::TableAppliedDiff<'r, PuzzleAgentMessageRow>, @@ -4743,6 +4774,11 @@ impl<'r> __sdk::AppliedDiff<'r> for AppliedDiff<'r> { &self.profile_wallet_manual_restriction, event, ); + callbacks.invoke_table_row_callbacks::( + "profile_wallet_refund_outbox", + &self.profile_wallet_refund_outbox, + event, + ); callbacks.invoke_table_row_callbacks::( "public_work_like", &self.public_work_like, @@ -5690,6 +5726,7 @@ impl __sdk::SpacetimeModule for RemoteModule { profile_wallet_consumption_total_table::register_table(client_cache); profile_wallet_ledger_table::register_table(client_cache); profile_wallet_manual_restriction_table::register_table(client_cache); + profile_wallet_refund_outbox_table::register_table(client_cache); public_work_like_table::register_table(client_cache); public_work_play_daily_stat_table::register_table(client_cache); puzzle_agent_message_table::register_table(client_cache); @@ -5831,6 +5868,7 @@ impl __sdk::SpacetimeModule for RemoteModule { "profile_wallet_consumption_total", "profile_wallet_ledger", "profile_wallet_manual_restriction", + "profile_wallet_refund_outbox", "public_work_like", "public_work_play_daily_stat", "puzzle_agent_message", diff --git a/server-rs/crates/spacetime-client/src/module_bindings/enqueue_profile_wallet_refund_outbox_and_return_procedure.rs b/server-rs/crates/spacetime-client/src/module_bindings/enqueue_profile_wallet_refund_outbox_and_return_procedure.rs new file mode 100644 index 000000000..4f5f28c68 --- /dev/null +++ b/server-rs/crates/spacetime-client/src/module_bindings/enqueue_profile_wallet_refund_outbox_and_return_procedure.rs @@ -0,0 +1,62 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#![allow(unused, clippy::all)] +use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws}; + +use super::runtime_profile_wallet_refund_outbox_enqueue_input_type::RuntimeProfileWalletRefundOutboxEnqueueInput; +use super::runtime_profile_wallet_refund_outbox_procedure_result_type::RuntimeProfileWalletRefundOutboxProcedureResult; + +#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)] +#[sats(crate = __lib)] +struct EnqueueProfileWalletRefundOutboxAndReturnArgs { + pub input: RuntimeProfileWalletRefundOutboxEnqueueInput, +} + +impl __sdk::InModule for EnqueueProfileWalletRefundOutboxAndReturnArgs { + type Module = super::RemoteModule; +} + +#[allow(non_camel_case_types)] +/// Extension trait for access to the procedure `enqueue_profile_wallet_refund_outbox_and_return`. +/// +/// Implemented for [`super::RemoteProcedures`]. +pub trait enqueue_profile_wallet_refund_outbox_and_return { + fn enqueue_profile_wallet_refund_outbox_and_return( + &self, + input: RuntimeProfileWalletRefundOutboxEnqueueInput, + ) { + self.enqueue_profile_wallet_refund_outbox_and_return_then(input, |_, _| {}); + } + + fn enqueue_profile_wallet_refund_outbox_and_return_then( + &self, + input: RuntimeProfileWalletRefundOutboxEnqueueInput, + + __callback: impl FnOnce( + &super::ProcedureEventContext, + Result, + ) + Send + + 'static, + ); +} + +impl enqueue_profile_wallet_refund_outbox_and_return for super::RemoteProcedures { + fn enqueue_profile_wallet_refund_outbox_and_return_then( + &self, + input: RuntimeProfileWalletRefundOutboxEnqueueInput, + + __callback: impl FnOnce( + &super::ProcedureEventContext, + Result, + ) + Send + + 'static, + ) { + self.imp + .invoke_procedure_with_callback::<_, RuntimeProfileWalletRefundOutboxProcedureResult>( + "enqueue_profile_wallet_refund_outbox_and_return", + EnqueueProfileWalletRefundOutboxAndReturnArgs { input }, + __callback, + ); + } +} diff --git a/server-rs/crates/spacetime-client/src/module_bindings/process_profile_wallet_refund_outbox_and_return_procedure.rs b/server-rs/crates/spacetime-client/src/module_bindings/process_profile_wallet_refund_outbox_and_return_procedure.rs new file mode 100644 index 000000000..8b6f64ccd --- /dev/null +++ b/server-rs/crates/spacetime-client/src/module_bindings/process_profile_wallet_refund_outbox_and_return_procedure.rs @@ -0,0 +1,62 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#![allow(unused, clippy::all)] +use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws}; + +use super::runtime_profile_wallet_refund_outbox_procedure_result_type::RuntimeProfileWalletRefundOutboxProcedureResult; +use super::runtime_profile_wallet_refund_outbox_process_input_type::RuntimeProfileWalletRefundOutboxProcessInput; + +#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)] +#[sats(crate = __lib)] +struct ProcessProfileWalletRefundOutboxAndReturnArgs { + pub input: RuntimeProfileWalletRefundOutboxProcessInput, +} + +impl __sdk::InModule for ProcessProfileWalletRefundOutboxAndReturnArgs { + type Module = super::RemoteModule; +} + +#[allow(non_camel_case_types)] +/// Extension trait for access to the procedure `process_profile_wallet_refund_outbox_and_return`. +/// +/// Implemented for [`super::RemoteProcedures`]. +pub trait process_profile_wallet_refund_outbox_and_return { + fn process_profile_wallet_refund_outbox_and_return( + &self, + input: RuntimeProfileWalletRefundOutboxProcessInput, + ) { + self.process_profile_wallet_refund_outbox_and_return_then(input, |_, _| {}); + } + + fn process_profile_wallet_refund_outbox_and_return_then( + &self, + input: RuntimeProfileWalletRefundOutboxProcessInput, + + __callback: impl FnOnce( + &super::ProcedureEventContext, + Result, + ) + Send + + 'static, + ); +} + +impl process_profile_wallet_refund_outbox_and_return for super::RemoteProcedures { + fn process_profile_wallet_refund_outbox_and_return_then( + &self, + input: RuntimeProfileWalletRefundOutboxProcessInput, + + __callback: impl FnOnce( + &super::ProcedureEventContext, + Result, + ) + Send + + 'static, + ) { + self.imp + .invoke_procedure_with_callback::<_, RuntimeProfileWalletRefundOutboxProcedureResult>( + "process_profile_wallet_refund_outbox_and_return", + ProcessProfileWalletRefundOutboxAndReturnArgs { input }, + __callback, + ); + } +} diff --git a/server-rs/crates/spacetime-client/src/module_bindings/profile_wallet_refund_outbox_table.rs b/server-rs/crates/spacetime-client/src/module_bindings/profile_wallet_refund_outbox_table.rs new file mode 100644 index 000000000..6f6233437 --- /dev/null +++ b/server-rs/crates/spacetime-client/src/module_bindings/profile_wallet_refund_outbox_table.rs @@ -0,0 +1,235 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#![allow(unused, clippy::all)] +use super::profile_wallet_refund_outbox_type::ProfileWalletRefundOutbox; +use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws}; + +/// Table handle for the table `profile_wallet_refund_outbox`. +/// +/// Obtain a handle from the [`ProfileWalletRefundOutboxTableAccess::profile_wallet_refund_outbox`] method on [`super::RemoteTables`], +/// like `ctx.db.profile_wallet_refund_outbox()`. +/// +/// Users are encouraged not to explicitly reference this type, +/// but to directly chain method calls, +/// like `ctx.db.profile_wallet_refund_outbox().on_insert(...)`. +pub struct ProfileWalletRefundOutboxTableHandle<'ctx> { + imp: __sdk::TableHandle, + ctx: std::marker::PhantomData<&'ctx super::RemoteTables>, +} + +/// Lifetime-aware accessor marker for the table `profile_wallet_refund_outbox`. +pub struct ProfileWalletRefundOutboxTableAccessor; + +impl __sdk::TableAccessor for ProfileWalletRefundOutboxTableAccessor { + type Row = ProfileWalletRefundOutbox; + type Handle<'db> = ProfileWalletRefundOutboxTableHandle<'db>; + + fn get<'db>(db: &'db super::RemoteTables) -> Self::Handle<'db> { + db.profile_wallet_refund_outbox() + } +} + +#[allow(non_camel_case_types)] +/// Extension trait for access to the table `profile_wallet_refund_outbox`. +/// +/// Implemented for [`super::RemoteTables`]. +pub trait ProfileWalletRefundOutboxTableAccess { + #[allow(non_snake_case)] + /// Obtain a [`ProfileWalletRefundOutboxTableHandle`], which mediates access to the table `profile_wallet_refund_outbox`. + fn profile_wallet_refund_outbox(&self) -> ProfileWalletRefundOutboxTableHandle<'_>; +} + +impl ProfileWalletRefundOutboxTableAccess for super::RemoteTables { + fn profile_wallet_refund_outbox(&self) -> ProfileWalletRefundOutboxTableHandle<'_> { + ProfileWalletRefundOutboxTableHandle { + imp: self + .imp + .get_table::("profile_wallet_refund_outbox"), + ctx: std::marker::PhantomData, + } + } +} + +pub struct ProfileWalletRefundOutboxInsertCallbackId(__sdk::CallbackId); +pub struct ProfileWalletRefundOutboxDeleteCallbackId(__sdk::CallbackId); + +impl<'ctx> __sdk::TableLike for ProfileWalletRefundOutboxTableHandle<'ctx> { + type Row = ProfileWalletRefundOutbox; + type EventContext = super::EventContext; + + fn count(&self) -> u64 { + self.imp.count() + } + fn iter(&self) -> impl Iterator + '_ { + self.imp.iter() + } +} + +impl<'ctx> __sdk::Table for ProfileWalletRefundOutboxTableHandle<'ctx> { + type Row = ProfileWalletRefundOutbox; + type EventContext = super::EventContext; + + fn count(&self) -> u64 { + self.imp.count() + } + fn iter(&self) -> impl Iterator + '_ { + self.imp.iter() + } + + type InsertCallbackId = ProfileWalletRefundOutboxInsertCallbackId; + + fn on_insert( + &self, + callback: impl FnMut(&Self::EventContext, &Self::Row) + Send + 'static, + ) -> ProfileWalletRefundOutboxInsertCallbackId { + ProfileWalletRefundOutboxInsertCallbackId(self.imp.on_insert(Box::new(callback))) + } + + fn remove_on_insert(&self, callback: ProfileWalletRefundOutboxInsertCallbackId) { + self.imp.remove_on_insert(callback.0) + } + + type DeleteCallbackId = ProfileWalletRefundOutboxDeleteCallbackId; + + fn on_delete( + &self, + callback: impl FnMut(&Self::EventContext, &Self::Row) + Send + 'static, + ) -> ProfileWalletRefundOutboxDeleteCallbackId { + ProfileWalletRefundOutboxDeleteCallbackId(self.imp.on_delete(Box::new(callback))) + } + + fn remove_on_delete(&self, callback: ProfileWalletRefundOutboxDeleteCallbackId) { + self.imp.remove_on_delete(callback.0) + } +} + +impl<'ctx> __sdk::WithInsert for ProfileWalletRefundOutboxTableHandle<'ctx> { + type InsertCallbackId = ProfileWalletRefundOutboxInsertCallbackId; + + fn on_insert( + &self, + callback: impl FnMut(&Self::EventContext, &Self::Row) + Send + 'static, + ) -> ProfileWalletRefundOutboxInsertCallbackId { + ProfileWalletRefundOutboxInsertCallbackId(self.imp.on_insert(Box::new(callback))) + } + + fn remove_on_insert(&self, callback: ProfileWalletRefundOutboxInsertCallbackId) { + self.imp.remove_on_insert(callback.0) + } +} + +impl<'ctx> __sdk::WithDelete for ProfileWalletRefundOutboxTableHandle<'ctx> { + type DeleteCallbackId = ProfileWalletRefundOutboxDeleteCallbackId; + + fn on_delete( + &self, + callback: impl FnMut(&Self::EventContext, &Self::Row) + Send + 'static, + ) -> ProfileWalletRefundOutboxDeleteCallbackId { + ProfileWalletRefundOutboxDeleteCallbackId(self.imp.on_delete(Box::new(callback))) + } + + fn remove_on_delete(&self, callback: ProfileWalletRefundOutboxDeleteCallbackId) { + self.imp.remove_on_delete(callback.0) + } +} + +pub struct ProfileWalletRefundOutboxUpdateCallbackId(__sdk::CallbackId); + +impl<'ctx> __sdk::TableWithPrimaryKey for ProfileWalletRefundOutboxTableHandle<'ctx> { + type UpdateCallbackId = ProfileWalletRefundOutboxUpdateCallbackId; + + fn on_update( + &self, + callback: impl FnMut(&Self::EventContext, &Self::Row, &Self::Row) + Send + 'static, + ) -> ProfileWalletRefundOutboxUpdateCallbackId { + ProfileWalletRefundOutboxUpdateCallbackId(self.imp.on_update(Box::new(callback))) + } + + fn remove_on_update(&self, callback: ProfileWalletRefundOutboxUpdateCallbackId) { + self.imp.remove_on_update(callback.0) + } +} + +impl<'ctx> __sdk::WithUpdate for ProfileWalletRefundOutboxTableHandle<'ctx> { + type UpdateCallbackId = ProfileWalletRefundOutboxUpdateCallbackId; + + fn on_update( + &self, + callback: impl FnMut(&Self::EventContext, &Self::Row, &Self::Row) + Send + 'static, + ) -> ProfileWalletRefundOutboxUpdateCallbackId { + ProfileWalletRefundOutboxUpdateCallbackId(self.imp.on_update(Box::new(callback))) + } + + fn remove_on_update(&self, callback: ProfileWalletRefundOutboxUpdateCallbackId) { + self.imp.remove_on_update(callback.0) + } +} + +/// Access to the `refund_ledger_id` unique index on the table `profile_wallet_refund_outbox`, +/// which allows point queries on the field of the same name +/// via the [`ProfileWalletRefundOutboxRefundLedgerIdUnique::find`] method. +/// +/// Users are encouraged not to explicitly reference this type, +/// but to directly chain method calls, +/// like `ctx.db.profile_wallet_refund_outbox().refund_ledger_id().find(...)`. +pub struct ProfileWalletRefundOutboxRefundLedgerIdUnique<'ctx> { + imp: __sdk::UniqueConstraintHandle, + phantom: std::marker::PhantomData<&'ctx super::RemoteTables>, +} + +impl<'ctx> ProfileWalletRefundOutboxTableHandle<'ctx> { + /// Get a handle on the `refund_ledger_id` unique index on the table `profile_wallet_refund_outbox`. + pub fn refund_ledger_id(&self) -> ProfileWalletRefundOutboxRefundLedgerIdUnique<'ctx> { + ProfileWalletRefundOutboxRefundLedgerIdUnique { + imp: self.imp.get_unique_constraint::("refund_ledger_id"), + phantom: std::marker::PhantomData, + } + } +} + +impl<'ctx> ProfileWalletRefundOutboxRefundLedgerIdUnique<'ctx> { + /// Find the subscribed row whose `refund_ledger_id` column value is equal to `col_val`, + /// if such a row is present in the client cache. + pub fn find(&self, col_val: &String) -> Option { + self.imp.find(col_val) + } +} + +#[doc(hidden)] +pub(super) fn register_table(client_cache: &mut __sdk::ClientCache) { + let _table = + client_cache.get_or_make_table::("profile_wallet_refund_outbox"); + _table.add_unique_constraint::("refund_ledger_id", |row| &row.refund_ledger_id); +} + +#[doc(hidden)] +pub(super) fn parse_table_update( + raw_updates: __ws::v2::TableUpdate, +) -> __sdk::Result<__sdk::TableUpdate> { + __sdk::TableUpdate::parse_table_update(raw_updates).map_err(|e| { + __sdk::InternalError::failed_parse("TableUpdate", "TableUpdate") + .with_cause(e) + .into() + }) +} + +#[allow(non_camel_case_types)] +/// Extension trait for query builder access to the table `ProfileWalletRefundOutbox`. +/// +/// Implemented for [`__sdk::QueryTableAccessor`]. +pub trait profile_wallet_refund_outboxQueryTableAccess { + #[allow(non_snake_case)] + /// Get a query builder for the table `ProfileWalletRefundOutbox`. + fn profile_wallet_refund_outbox( + &self, + ) -> __sdk::__query_builder::Table; +} + +impl profile_wallet_refund_outboxQueryTableAccess for __sdk::QueryTableAccessor { + fn profile_wallet_refund_outbox( + &self, + ) -> __sdk::__query_builder::Table { + __sdk::__query_builder::Table::new("profile_wallet_refund_outbox") + } +} diff --git a/server-rs/crates/spacetime-client/src/module_bindings/profile_wallet_refund_outbox_type.rs b/server-rs/crates/spacetime-client/src/module_bindings/profile_wallet_refund_outbox_type.rs new file mode 100644 index 000000000..f2c38e984 --- /dev/null +++ b/server-rs/crates/spacetime-client/src/module_bindings/profile_wallet_refund_outbox_type.rs @@ -0,0 +1,103 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#![allow(unused, clippy::all)] +use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws}; + +#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)] +#[sats(crate = __lib)] +pub struct ProfileWalletRefundOutbox { + pub refund_ledger_id: String, + pub consume_ledger_id: String, + pub owner_user_id: String, + pub amount: u64, + pub created_at: __sdk::Timestamp, + pub asset_kind: String, + pub asset_id: String, + pub settlement_reason: String, + pub external_generation_job_id: Option, + pub external_generation_claim_attempt: Option, + pub status: String, + pub available_at: __sdk::Timestamp, + pub attempts: u32, + pub last_error: Option, + pub last_attempted_at: Option<__sdk::Timestamp>, + pub last_worker_id: Option, +} + +impl __sdk::InModule for ProfileWalletRefundOutbox { + type Module = super::RemoteModule; +} + +/// Column accessor struct for the table `ProfileWalletRefundOutbox`. +/// +/// Provides typed access to columns for query building. +pub struct ProfileWalletRefundOutboxCols { + pub refund_ledger_id: __sdk::__query_builder::Col, + pub consume_ledger_id: __sdk::__query_builder::Col, + pub owner_user_id: __sdk::__query_builder::Col, + pub amount: __sdk::__query_builder::Col, + pub created_at: __sdk::__query_builder::Col, + pub asset_kind: __sdk::__query_builder::Col, + pub asset_id: __sdk::__query_builder::Col, + pub settlement_reason: __sdk::__query_builder::Col, + pub external_generation_job_id: + __sdk::__query_builder::Col>, + pub external_generation_claim_attempt: + __sdk::__query_builder::Col>, + pub status: __sdk::__query_builder::Col, + pub available_at: __sdk::__query_builder::Col, + pub attempts: __sdk::__query_builder::Col, + pub last_error: __sdk::__query_builder::Col>, + pub last_attempted_at: + __sdk::__query_builder::Col>, + pub last_worker_id: __sdk::__query_builder::Col>, +} + +impl __sdk::__query_builder::HasCols for ProfileWalletRefundOutbox { + type Cols = ProfileWalletRefundOutboxCols; + fn cols(table_name: &'static str) -> Self::Cols { + ProfileWalletRefundOutboxCols { + refund_ledger_id: __sdk::__query_builder::Col::new(table_name, "refund_ledger_id"), + consume_ledger_id: __sdk::__query_builder::Col::new(table_name, "consume_ledger_id"), + owner_user_id: __sdk::__query_builder::Col::new(table_name, "owner_user_id"), + amount: __sdk::__query_builder::Col::new(table_name, "amount"), + created_at: __sdk::__query_builder::Col::new(table_name, "created_at"), + asset_kind: __sdk::__query_builder::Col::new(table_name, "asset_kind"), + asset_id: __sdk::__query_builder::Col::new(table_name, "asset_id"), + settlement_reason: __sdk::__query_builder::Col::new(table_name, "settlement_reason"), + external_generation_job_id: __sdk::__query_builder::Col::new( + table_name, + "external_generation_job_id", + ), + external_generation_claim_attempt: __sdk::__query_builder::Col::new( + table_name, + "external_generation_claim_attempt", + ), + status: __sdk::__query_builder::Col::new(table_name, "status"), + available_at: __sdk::__query_builder::Col::new(table_name, "available_at"), + attempts: __sdk::__query_builder::Col::new(table_name, "attempts"), + last_error: __sdk::__query_builder::Col::new(table_name, "last_error"), + last_attempted_at: __sdk::__query_builder::Col::new(table_name, "last_attempted_at"), + last_worker_id: __sdk::__query_builder::Col::new(table_name, "last_worker_id"), + } + } +} + +/// Indexed column accessor struct for the table `ProfileWalletRefundOutbox`. +/// +/// Provides typed access to indexed columns for query building. +pub struct ProfileWalletRefundOutboxIxCols { + pub refund_ledger_id: __sdk::__query_builder::IxCol, +} + +impl __sdk::__query_builder::HasIxCols for ProfileWalletRefundOutbox { + type IxCols = ProfileWalletRefundOutboxIxCols; + fn ix_cols(table_name: &'static str) -> Self::IxCols { + ProfileWalletRefundOutboxIxCols { + refund_ledger_id: __sdk::__query_builder::IxCol::new(table_name, "refund_ledger_id"), + } + } +} + +impl __sdk::__query_builder::CanBeLookupTable for ProfileWalletRefundOutbox {} diff --git a/server-rs/crates/spacetime-client/src/module_bindings/runtime_profile_wallet_refund_outbox_enqueue_input_type.rs b/server-rs/crates/spacetime-client/src/module_bindings/runtime_profile_wallet_refund_outbox_enqueue_input_type.rs new file mode 100644 index 000000000..14e090000 --- /dev/null +++ b/server-rs/crates/spacetime-client/src/module_bindings/runtime_profile_wallet_refund_outbox_enqueue_input_type.rs @@ -0,0 +1,23 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#![allow(unused, clippy::all)] +use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws}; + +#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)] +#[sats(crate = __lib)] +pub struct RuntimeProfileWalletRefundOutboxEnqueueInput { + pub owner_user_id: String, + pub amount: u64, + pub refund_ledger_id: String, + pub created_at_micros: i64, + pub asset_kind: String, + pub asset_id: String, + pub settlement_reason: String, + pub external_generation_job_id: Option, + pub external_generation_claim_attempt: Option, +} + +impl __sdk::InModule for RuntimeProfileWalletRefundOutboxEnqueueInput { + type Module = super::RemoteModule; +} diff --git a/server-rs/crates/spacetime-client/src/module_bindings/runtime_profile_wallet_refund_outbox_procedure_result_type.rs b/server-rs/crates/spacetime-client/src/module_bindings/runtime_profile_wallet_refund_outbox_procedure_result_type.rs new file mode 100644 index 000000000..dae26bc0d --- /dev/null +++ b/server-rs/crates/spacetime-client/src/module_bindings/runtime_profile_wallet_refund_outbox_procedure_result_type.rs @@ -0,0 +1,20 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#![allow(unused, clippy::all)] +use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws}; + +#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)] +#[sats(crate = __lib)] +pub struct RuntimeProfileWalletRefundOutboxProcedureResult { + pub ok: bool, + pub enqueued_count: u32, + pub processed_count: u32, + pub retry_count: u32, + pub failed_count: u32, + pub error_message: Option, +} + +impl __sdk::InModule for RuntimeProfileWalletRefundOutboxProcedureResult { + type Module = super::RemoteModule; +} diff --git a/server-rs/crates/spacetime-client/src/module_bindings/runtime_profile_wallet_refund_outbox_process_input_type.rs b/server-rs/crates/spacetime-client/src/module_bindings/runtime_profile_wallet_refund_outbox_process_input_type.rs new file mode 100644 index 000000000..1c3f25e7c --- /dev/null +++ b/server-rs/crates/spacetime-client/src/module_bindings/runtime_profile_wallet_refund_outbox_process_input_type.rs @@ -0,0 +1,16 @@ +// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE +// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD. + +#![allow(unused, clippy::all)] +use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws}; + +#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)] +#[sats(crate = __lib)] +pub struct RuntimeProfileWalletRefundOutboxProcessInput { + pub worker_id: String, + pub limit: u32, +} + +impl __sdk::InModule for RuntimeProfileWalletRefundOutboxProcessInput { + type Module = super::RemoteModule; +} diff --git a/server-rs/crates/spacetime-module/src/external_generation.rs b/server-rs/crates/spacetime-module/src/external_generation.rs index 9cfa4fa13..547c4282c 100644 --- a/server-rs/crates/spacetime-module/src/external_generation.rs +++ b/server-rs/crates/spacetime-module/src/external_generation.rs @@ -888,6 +888,8 @@ fn finalize_external_generation_job_after_lease_exhaustion( row.price_mud_points, failed_at, "final_attempt_lease_expired", + &row.job_kind, + &row.source_entity_id, )?; let row = mark_external_generation_job_lease_exhausted(row, failed_at, refund_ledger_id); persist_external_generation_job_row(ctx, row.clone()); @@ -1437,6 +1439,8 @@ fn fail_external_generation_job_tx( row.price_mud_points, failed_at, "worker_attempt_failed", + &row.job_kind, + &row.source_entity_id, )?; let requested_refund_ledger_id = input .refund_ledger_id diff --git a/server-rs/crates/spacetime-module/src/migration.rs b/server-rs/crates/spacetime-module/src/migration.rs index 4b6993bf7..32e02063c 100644 --- a/server-rs/crates/spacetime-module/src/migration.rs +++ b/server-rs/crates/spacetime-module/src/migration.rs @@ -199,6 +199,7 @@ macro_rules! migration_tables { profile_wallet_ledger, profile_wallet_consumption_total, asset_operation_wallet_settlement, + profile_wallet_refund_outbox, profile_wallet_config, analytics_date_dimension, tracking_event, diff --git a/server-rs/crates/spacetime-module/src/runtime/active/profile.rs b/server-rs/crates/spacetime-module/src/runtime/active/profile.rs index e53d8e6b9..e3610969e 100644 --- a/server-rs/crates/spacetime-module/src/runtime/active/profile.rs +++ b/server-rs/crates/spacetime-module/src/runtime/active/profile.rs @@ -19,6 +19,8 @@ const PROFILE_RECHARGE_ORDER_EXPIRATION_CHECK_LIMIT_DEFAULT: u32 = 50; const PROFILE_RECHARGE_ORDER_EXPIRATION_CHECK_LIMIT_MAX: u32 = 200; const ASSET_OPERATION_CONSUME_LEDGER_PREFIX: &str = "asset_operation_consume:"; const ASSET_OPERATION_REFUND_LEDGER_PREFIX: &str = "asset_operation_refund:"; +const PROFILE_WALLET_REFUND_OUTBOX_STATUS_PENDING: &str = "pending"; +const PROFILE_WALLET_REFUND_OUTBOX_MAX_BATCH_SIZE: u32 = 100; #[spacetimedb::table(accessor = profile_dashboard_state)] pub struct ProfileDashboardState { @@ -82,6 +84,34 @@ pub struct AssetOperationWalletSettlement { pub(crate) settled_at: Timestamp, } +#[spacetimedb::table( + accessor = profile_wallet_refund_outbox, + index( + accessor = by_profile_wallet_refund_outbox_status_available, + btree(columns = [status, available_at]) + ) +)] +#[derive(Clone)] +pub struct ProfileWalletRefundOutbox { + #[primary_key] + pub(crate) refund_ledger_id: String, + pub(crate) consume_ledger_id: String, + pub(crate) owner_user_id: String, + pub(crate) amount: u64, + pub(crate) created_at: Timestamp, + pub(crate) asset_kind: String, + pub(crate) asset_id: String, + pub(crate) settlement_reason: String, + pub(crate) external_generation_job_id: Option, + pub(crate) external_generation_claim_attempt: Option, + pub(crate) status: String, + pub(crate) available_at: Timestamp, + pub(crate) attempts: u32, + pub(crate) last_error: Option, + pub(crate) last_attempted_at: Option, + pub(crate) last_worker_id: Option, +} + #[spacetimedb::table(accessor = profile_wallet_config)] #[derive(Clone)] pub struct ProfileWalletConfig { @@ -1159,6 +1189,82 @@ pub fn refund_profile_wallet_points_and_return( } } +#[spacetimedb::procedure] +pub fn enqueue_profile_wallet_refund_outbox_and_return( + ctx: &mut ProcedureContext, + input: RuntimeProfileWalletRefundOutboxEnqueueInput, +) -> RuntimeProfileWalletRefundOutboxProcedureResult { + let caller = ctx.sender(); + match ctx.try_with_tx(|tx| { + crate::editor_project_storage::require_editor_generation_runtime_service_identity( + tx, caller, + )?; + enqueue_profile_wallet_refund_outbox_tx( + tx, + input.owner_user_id.clone(), + input.amount, + input.refund_ledger_id.clone(), + input.created_at_micros, + input.asset_kind.clone(), + input.asset_id.clone(), + input.settlement_reason.clone(), + input.external_generation_job_id.clone(), + input.external_generation_claim_attempt, + ) + .map(|enqueued| u32::from(enqueued)) + }) { + Ok(enqueued_count) => RuntimeProfileWalletRefundOutboxProcedureResult { + ok: true, + enqueued_count, + processed_count: 0, + retry_count: 0, + failed_count: 0, + error_message: None, + }, + Err(message) => RuntimeProfileWalletRefundOutboxProcedureResult { + ok: false, + enqueued_count: 0, + processed_count: 0, + retry_count: 0, + failed_count: 0, + error_message: Some(message), + }, + } +} + +#[spacetimedb::procedure] +pub fn process_profile_wallet_refund_outbox_and_return( + ctx: &mut ProcedureContext, + input: RuntimeProfileWalletRefundOutboxProcessInput, +) -> RuntimeProfileWalletRefundOutboxProcedureResult { + let caller = ctx.sender(); + match ctx.try_with_tx(|tx| { + crate::editor_project_storage::require_editor_generation_runtime_service_identity( + tx, caller, + )?; + process_profile_wallet_refund_outbox_tx(tx, input.clone()) + }) { + Ok((processed_count, retry_count, failed_count)) => { + RuntimeProfileWalletRefundOutboxProcedureResult { + ok: true, + enqueued_count: 0, + processed_count, + retry_count, + failed_count, + error_message: None, + } + } + Err(message) => RuntimeProfileWalletRefundOutboxProcedureResult { + ok: false, + enqueued_count: 0, + processed_count: 0, + retry_count: 0, + failed_count: 0, + error_message: Some(message), + }, + } +} + // play stats 与 dashboard 共用 dashboard projection 的 total_play_time / updated_at，避免 Axum 侧拼装。 #[cfg(any())] #[spacetimedb::procedure] @@ -9125,6 +9231,254 @@ pub(crate) fn grant_profile_wallet_points_with_metadata( ) } +pub(crate) fn enqueue_profile_wallet_refund_outbox_tx( + ctx: &ReducerContext, + owner_user_id: String, + amount: u64, + refund_ledger_id: String, + created_at_micros: i64, + asset_kind: String, + asset_id: String, + settlement_reason: String, + external_generation_job_id: Option, + external_generation_claim_attempt: Option, +) -> Result { + let owner_user_id = owner_user_id.trim().to_string(); + if owner_user_id.is_empty() { + return Err("资产操作退款用户不能为空".to_string()); + } + if amount == 0 { + return Err("资产操作退款金额必须大于 0".to_string()); + } + let refund_ledger_id = refund_ledger_id.trim().to_string(); + let consume_ledger_id = asset_operation_consume_ledger_id(&refund_ledger_id)?; + let asset_kind = asset_kind.trim().to_string(); + if asset_kind.is_empty() { + return Err("资产操作退款 asset_kind 不能为空".to_string()); + } + let asset_id = asset_id.trim().to_string(); + if asset_id.is_empty() { + return Err("资产操作退款 asset_id 不能为空".to_string()); + } + let settlement_reason = settlement_reason.trim().to_string(); + if settlement_reason.is_empty() { + return Err("资产操作退款 settlement_reason 不能为空".to_string()); + } + let external_generation_job_id = normalize_optional_text(external_generation_job_id); + if external_generation_job_id.is_some() != external_generation_claim_attempt.is_some() { + return Err("资产操作退款 outbox 任务与 attempt 必须成对提供".to_string()); + } + let created_at = Timestamp::from_micros_since_unix_epoch(created_at_micros); + let disposition = resolve_asset_operation_refund_disposition_from_ledger( + ctx, + owner_user_id.as_str(), + i64::try_from(amount).map_err(|_| "资产操作退款金额超出范围".to_string())?, + refund_ledger_id.as_str(), + )?; + + if matches!(disposition, AssetOperationRefundDisposition::Noop) { + if let Some(existing) = ctx + .db + .profile_wallet_refund_outbox() + .refund_ledger_id() + .find(&refund_ledger_id) + { + validate_profile_wallet_refund_outbox_fact( + &existing, + &refund_ledger_id, + &consume_ledger_id, + &owner_user_id, + amount, + &asset_kind, + &asset_id, + &settlement_reason, + external_generation_job_id.as_deref(), + external_generation_claim_attempt, + )?; + ctx.db + .profile_wallet_refund_outbox() + .refund_ledger_id() + .delete(&refund_ledger_id); + } + return Ok(false); + } + if matches!(disposition, AssetOperationRefundDisposition::RecordIntent) { + record_asset_operation_wallet_settlement( + ctx, + owner_user_id.as_str(), + i64::try_from(amount).map_err(|_| "资产操作退款金额超出范围".to_string())?, + refund_ledger_id.as_str(), + created_at, + )?; + // consume 尚不可见时，settlement 是取消 intent；不应再创建会被 worker + // 误认为可退款的 outbox 行。迟到 consume 会在同一 resolver 中被拒绝。 + return Ok(false); + } + + let existing = ctx + .db + .profile_wallet_refund_outbox() + .refund_ledger_id() + .find(&refund_ledger_id); + if let Some(existing) = existing { + validate_profile_wallet_refund_outbox_fact( + &existing, + &refund_ledger_id, + &consume_ledger_id, + &owner_user_id, + amount, + &asset_kind, + &asset_id, + &settlement_reason, + external_generation_job_id.as_deref(), + external_generation_claim_attempt, + )?; + let mut existing = existing; + existing.status = PROFILE_WALLET_REFUND_OUTBOX_STATUS_PENDING.to_string(); + existing.available_at = ctx.timestamp; + existing.last_error = None; + ctx.db + .profile_wallet_refund_outbox() + .refund_ledger_id() + .update(existing); + return Ok(false); + } + + ctx.db + .profile_wallet_refund_outbox() + .insert(ProfileWalletRefundOutbox { + refund_ledger_id, + consume_ledger_id, + owner_user_id, + amount, + created_at, + asset_kind, + asset_id, + settlement_reason, + external_generation_job_id, + external_generation_claim_attempt, + status: PROFILE_WALLET_REFUND_OUTBOX_STATUS_PENDING.to_string(), + available_at: ctx.timestamp, + attempts: 0, + last_error: None, + last_attempted_at: None, + last_worker_id: None, + }); + Ok(true) +} + +fn validate_profile_wallet_refund_outbox_fact( + row: &ProfileWalletRefundOutbox, + expected_refund_ledger_id: &str, + expected_consume_ledger_id: &str, + expected_owner_user_id: &str, + expected_amount: u64, + expected_asset_kind: &str, + expected_asset_id: &str, + expected_settlement_reason: &str, + expected_external_generation_job_id: Option<&str>, + expected_external_generation_claim_attempt: Option, +) -> Result<(), String> { + if row.refund_ledger_id != expected_refund_ledger_id + || row.consume_ledger_id != expected_consume_ledger_id + || row.owner_user_id != expected_owner_user_id + || row.amount != expected_amount + { + return Err("资产操作退款 outbox 金融事实不匹配".to_string()); + } + if row.asset_kind != expected_asset_kind || row.asset_id != expected_asset_id { + return Err("资产操作退款 outbox 资源事实不匹配".to_string()); + } + if row.settlement_reason != expected_settlement_reason { + return Err("资产操作退款 outbox settlement_reason 不匹配".to_string()); + } + if row.external_generation_job_id.as_deref() != expected_external_generation_job_id + || row.external_generation_claim_attempt != expected_external_generation_claim_attempt + { + return Err("资产操作退款 outbox 任务 attempt 不匹配".to_string()); + } + Ok(()) +} + +fn process_profile_wallet_refund_outbox_tx( + ctx: &ReducerContext, + input: RuntimeProfileWalletRefundOutboxProcessInput, +) -> Result<(u32, u32, u32), String> { + let worker_id = input.worker_id.trim(); + if worker_id.is_empty() { + return Err("退款 outbox worker_id 不能为空".to_string()); + } + let limit = input + .limit + .clamp(1, PROFILE_WALLET_REFUND_OUTBOX_MAX_BATCH_SIZE); + let now = ctx.timestamp; + let mut rows = ctx + .db + .profile_wallet_refund_outbox() + .iter() + .filter(|row| { + row.status == PROFILE_WALLET_REFUND_OUTBOX_STATUS_PENDING && row.available_at <= now + }) + .collect::>(); + rows.sort_by(|left, right| { + left.available_at + .cmp(&right.available_at) + .then_with(|| left.created_at.cmp(&right.created_at)) + .then_with(|| left.refund_ledger_id.cmp(&right.refund_ledger_id)) + }); + + let mut processed_count: u32 = 0; + let mut retry_count: u32 = 0; + let mut failed_count: u32 = 0; + for mut row in rows.into_iter().take(limit as usize) { + row.attempts = row.attempts.saturating_add(1); + row.last_attempted_at = Some(now); + row.last_worker_id = Some(worker_id.to_string()); + let metadata_json = serde_json::json!({ + "externalGenerationJobId": &row.external_generation_job_id, + "externalGenerationClaimAttempt": row.external_generation_claim_attempt, + "assetKind": &row.asset_kind, + "assetId": &row.asset_id, + "settlementReason": &row.settlement_reason, + "refundOutboxWorkerId": worker_id, + }) + .to_string(); + let result = apply_profile_wallet_adjustment( + ctx, + RuntimeProfileWalletAdjustmentInput { + user_id: row.owner_user_id.clone(), + amount: row.amount, + ledger_id: row.refund_ledger_id.clone(), + created_at_micros: row.created_at.to_micros_since_unix_epoch(), + metadata_json, + }, + RuntimeProfileWalletLedgerSourceType::AssetOperationRefund, + false, + ); + match result { + Ok(_) => { + ctx.db + .profile_wallet_refund_outbox() + .refund_ledger_id() + .delete(&row.refund_ledger_id); + processed_count = processed_count.saturating_add(1); + } + Err(error) => { + row.available_at = + now + std::time::Duration::from_secs(2u64.saturating_pow(row.attempts.min(10))); + row.last_error = Some(error); + ctx.db + .profile_wallet_refund_outbox() + .refund_ledger_id() + .update(row); + retry_count = retry_count.saturating_add(1); + failed_count = failed_count.saturating_add(1); + } + } + } + Ok((processed_count, retry_count, failed_count)) +} + fn apply_profile_wallet_adjustment( ctx: &ReducerContext, input: RuntimeProfileWalletAdjustmentInput, @@ -9242,6 +9596,8 @@ pub(crate) fn settle_external_generation_attempt_refund( amount: u64, settled_at: Timestamp, settlement_reason: &str, + asset_kind: &str, + asset_id: &str, ) -> Result, String> { if amount == 0 { return Ok(None); @@ -9250,22 +9606,17 @@ pub(crate) fn settle_external_generation_attempt_refund( "{ASSET_OPERATION_REFUND_LEDGER_PREFIX}external_generation_job:{}:attempt:{attempt}", job_id.trim() ); - apply_profile_wallet_adjustment( + enqueue_profile_wallet_refund_outbox_tx( ctx, - RuntimeProfileWalletAdjustmentInput { - user_id: user_id.to_string(), - amount, - ledger_id: refund_ledger_id.clone(), - created_at_micros: settled_at.to_micros_since_unix_epoch(), - metadata_json: serde_json::json!({ - "externalGenerationJobId": job_id.trim(), - "claimAttempt": attempt, - "settlementReason": settlement_reason, - }) - .to_string(), - }, - RuntimeProfileWalletLedgerSourceType::AssetOperationRefund, - false, + user_id.to_string(), + amount, + refund_ledger_id.clone(), + settled_at.to_micros_since_unix_epoch(), + asset_kind.to_string(), + asset_id.to_string(), + settlement_reason.to_string(), + Some(job_id.trim().to_string()), + Some(attempt), )?; Ok(Some(refund_ledger_id)) }