From 6e9948ac9e9521dd3b871693610694202ee244b4 Mon Sep 17 00:00:00 2001 From: kdletters Date: Thu, 27 Aug 2026 19:11:23 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E8=AE=A4=E8=AF=81=E7=9F=AD?= =?UTF-8?q?=E7=8A=B6=E6=80=81=E4=B8=8E=E9=80=80=E6=AC=BE=20outbox=20?= =?UTF-8?q?=E8=B7=A8=E8=8A=82=E7=82=B9=E6=81=A2=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将短信验证码与微信 OAuth state 纳入 SpacetimeDB typed projection 并支持 CAS 只读刷新 为退款 emergency spool 增加 overflow 文件与崩溃恢复，避免容量上限静默丢失 为 API 与 external worker 配置独立 tracking/refund 持久卷 使用 profile refund outbox 的 status/available_at 索引扫描并同步架构运维文档 --- deploy/container/docker-compose.loadtest.yml | 7 + .../shared-memory/decision-log.md | 16 ++ ...】server-rs与SpacetimeDB数据契约-2026-05-15.md | 7 +- ...发运维】本地开发验证与生产运维-2026-05-15.md | 4 +- .../crates/api-server/src/asset_billing.rs | 4 +- .../api-server/src/password_management.rs | 15 +- server-rs/crates/api-server/src/phone_auth.rs | 58 ++++- server-rs/crates/api-server/src/state.rs | 45 +++- .../api-server/src/wallet_refund_outbox.rs | 205 +++++++++------ .../crates/api-server/src/wechat/auth.rs | 85 ++++++- server-rs/crates/module-auth/src/domain.rs | 33 ++- server-rs/crates/module-auth/src/lib.rs | 238 ++++++++++++++++-- .../src/active/mapper/auth.rs | 62 +++++ .../spacetime-client/src/mapper/auth.rs | 58 +++++ .../spacetime-client/src/module_bindings.rs | 4 + .../auth_store_projection_meta_type.rs | 6 + .../auth_store_projection_phone_code_type.rs | 21 ++ .../auth_store_projection_view_type.rs | 4 + ...auth_store_projection_wechat_state_type.rs | 24 ++ .../spacetime-module/src/auth/procedures.rs | 72 +++++- .../spacetime-module/src/auth/tables.rs | 4 + .../src/runtime/active/profile.rs | 7 +- 22 files changed, 850 insertions(+), 129 deletions(-) create mode 100644 server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_phone_code_type.rs create mode 100644 server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_wechat_state_type.rs diff --git a/deploy/container/docker-compose.loadtest.yml b/deploy/container/docker-compose.loadtest.yml index c2b78c73b..cdb7aed22 100644 --- a/deploy/container/docker-compose.loadtest.yml +++ b/deploy/container/docker-compose.loadtest.yml @@ -53,6 +53,7 @@ services: - "host.docker.internal:host-gateway" volumes: - api-tracking-outbox:/var/lib/genarrative/tracking-outbox + - api-wallet-refund-outbox:/var/lib/genarrative/wallet-refund-outbox ulimits: nofile: soft: 4096 @@ -85,6 +86,9 @@ services: OTEL_SERVICE_NAME: genarrative-external-generation-worker extra_hosts: - "host.docker.internal:host-gateway" + volumes: + - external-generation-tracking-outbox:/var/lib/genarrative/tracking-outbox-worker + - external-generation-wallet-refund-outbox:/var/lib/genarrative/wallet-refund-outbox ulimits: nofile: soft: 4096 @@ -142,4 +146,7 @@ services: volumes: spacetime-data: api-tracking-outbox: + api-wallet-refund-outbox: + external-generation-tracking-outbox: + external-generation-wallet-refund-outbox: nginx-logs: diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index a155d0931..a2e9ad552 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -16,6 +16,22 @@ --- +## 2026-08-27 退款 emergency spool 容量溢出保持可恢复 + +- 背景：本机 emergency spool 仅作为 SpacetimeDB 完全不可达时的最后恢复路径，原有 `MAX_BYTES` 分支会直接返回 `Dropped`，导致扣费已经完成但没有可重放记录。 +- 决策：达到普通 outbox `MAX_BYTES` 时，将退款记录写入同一持久目录的 `refund-overflow-*` 文件；该文件与普通 pending 文件一样由启动恢复和后台 worker 重放到 SpacetimeDB，且按 refund ledger id 保持幂等。溢出文件不计入普通阈值，但必须触发容量告警；底层磁盘写入失败仍进入关键退款人工补偿流程。 +- 影响范围：api-server wallet refund emergency spool、资产失败退款日志、loadtest / 预览 Compose 持久卷、后端架构与开发运维文档。 +- 验证方式：运行 api-server `wallet_refund_outbox` 定向测试，确认超限写入并保留 overflow 文件；运行 SpacetimeDB profile 测试、Compose 配置校验、编码和 diff 门禁。 + +## 2026-08-27 短期认证状态进入共享 typed projection + +- 背景：短信验证码和微信 OAuth state 仍只存在 API 进程内 HashMap，多节点请求或 API 重启会直接丢失，无法满足无粘性会话的鉴权恢复要求。 +- 决策：`AuthStoreProjectionView` 增加 `phone_codes` 与 `wechat_states` typed 字段，由 `auth_store_projection_meta` 以 JSON 投影持久化；启动恢复、CAS 同步和失败后的权威刷新都覆盖这两类短期状态。验证码哈希使用部署级稳定盐（当前复用 `GENARRATIVE_JWT_SECRET`），各 API 节点必须一致；认证 handler 在发码、消费验证码、创建/消费微信 state 后都要完成 projection sync，失败即返回服务错误；短期状态首次未命中时允许从正式投影做一次受 CAS 保护的只读刷新，再重试读取，不能依赖粘性会话。短期状态仍由 `module-auth` 内存工作集执行领域校验，但不再把本机 HashMap 当作持久化或跨节点真相。 +- 影响范围：`module-auth` projection、`spacetime-module` auth schema/procedure、`spacetime-client` bindings/facade、api-server 手机号 / 微信 handler、认证架构与运维文档。 +- 验证方式：运行 module-auth projection roundtrip（验证码可跨恢复校验、微信 state 可跨恢复消费）、SpacetimeDB schema/runtime/DDD 门禁、api-server 定向测试、编码和 diff 检查。 + +--- + ## 2026-08-27 SpacetimeDB 工具链统一升级到 2.8.3 - 背景：SpacetimeDB 2.8.0 引入 TypeScript submodule 与调度延迟观测，2.8.1 修复 v1 WebSocket 订阅移除死锁、TypeScript SDK `array` 读缓存别名和 Rust string 默认值支持，2.8.2 修复 table accessor 改名自动迁移，2.8.3 修复 scheduled function 从实际执行时间重排导致的长期漂移。仓库若继续锁定 2.7.0，会保留这些已知运行时与 SDK 问题。 diff --git a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md index 707e834c8..b3086ed82 100644 --- a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md +++ b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md @@ -412,10 +412,11 @@ Responses 的终态载荷既是工具调用的恢复源，也是正文的恢复 - Rust 结构体：`AuthStoreProjectionMeta` - 源码：`server-rs/crates/spacetime-module/src/auth/tables.rs` +- 职责：保存 typed 认证投影的单调版本，以及短期手机号验证码和微信 OAuth state 的序列化投影；`phone_codes_json` / `wechat_states_json` 只承载短期认证状态，不替代 `user_account`、`auth_identity` 或 `refresh_session` 的正式表语义。 -认证恢复策略：`api-server` 启动时只从 SpacetimeDB 正式认证表（`user_account` / `auth_identity` / `refresh_session`）导出 typed `AuthStoreProjectionView`，再恢复 `module-auth` 的进程内认证工作集；生产 Bearer 中间件不再从 `InMemoryAuthStore` 读取用户或会话，而是每次通过 typed `validate_auth_session` procedure 在 SpacetimeDB 事务内校验 `token_version`、会话归属、撤销时间和过期时间，SpacetimeDB 不可用时 fail closed 返回服务错误。`validate_auth_session`、投影导出和投影同步均从 `ctx.sender()` 派生调用方，并复用现役 runtime service identity 白名单；启动恢复先完成该服务身份初始化，普通 SpacetimeDB identity 不能读取或改写私有认证表。测试构建仍可使用显式的内存测试夹具。运行中 refresh cookie 在本进程工作集内未命中时直接按失效处理，不再从 SpacetimeDB 导出整包认证状态刷新内存，避免旧投影把重复手机号或旧会话重新灌回进程。`module-auth` 只保留内存工作集和 projection 导入 / 导出能力，不再保留 JSON 快照导入 / 导出能力，也不写本地持久化文件；`auth-store.json` / `GENARRATIVE_AUTH_STORE_PATH` 不再是兼容恢复源。认证创建、登录会话、刷新、退出、改密、重置密码、绑定和资料变更等写操作仍必须在返回客户端前通过 `sync_auth_store_projection` 成功同步 SpacetimeDB 正式认证表；同步失败时接口返回错误，不允许把只存在于当前进程内存的账号或会话当成成功结果。每个 API 工作集绑定启动恢复或上次成功同步得到的 `auth_store_projection_meta.updated_at` 版本作为 `base_updated_at_micros`，SpacetimeDB 在同一事务内执行基线 CAS，并要求新的 `updated_at_micros` 严格递增；基线不一致或版本不晚于当前值时整包写入失败，冲突节点在确认本次同步前没有遗留未确认本地变更且同步期间没有其它本地认证变更后，才可丢弃本地工作集并从正式表恢复，不能用陈旧工作集删除、恢复或覆盖另一节点的新状态；若此前已有待重试 revision 或同期还有本地变更则保留旧基线并继续失败关闭，不用自动恢复覆盖未提交变更；同步成功但期间又出现新本地变更时最多连续补同步三轮，仍未稳定则失败关闭。这只是迁移期并发保护，不改变正式认证表的权威地位。新用户注册奖励、邀请码绑定和登录埋点必须排在认证同步成功之后，避免认证没落库时先写出钱包或邀请关系。若启动恢复阶段 SpacetimeDB 不可连接或超时，`api-server` 会按固定间隔持续重试认证工作集恢复，恢复成功后才开始监听 HTTP，避免一次短超时让进程永久停留在依赖不可用状态。 +认证恢复策略：`api-server` 启动时从 SpacetimeDB 正式认证表（`user_account` / `auth_identity` / `refresh_session`）以及 `auth_store_projection_meta` 中的短期状态投影导出 typed `AuthStoreProjectionView`，再恢复 `module-auth` 的进程内认证工作集；生产 Bearer 中间件不再从 `InMemoryAuthStore` 读取用户或会话，而是每次通过 typed `validate_auth_session` procedure 在 SpacetimeDB 事务内校验 `token_version`、会话归属、撤销时间和过期时间，SpacetimeDB 不可用时 fail closed 返回服务错误。`validate_auth_session`、投影导出和投影同步均从 `ctx.sender()` 派生调用方，并复用现役 runtime service identity 白名单；启动恢复先完成该服务身份初始化，普通 SpacetimeDB identity 不能读取或改写私有认证表。测试构建仍可使用显式的内存测试夹具。短期认证请求在本进程工作集内未命中且工作集没有待同步变更时，先对正式投影做一次受 CAS 保护的只读刷新，再决定验证码 / OAuth state 是否失效，不依赖粘性会话；refresh cookie 仍只按正式 `refresh_session` 校验，其他认证数据也不得绕过正式同步。`module-auth` 只保留内存工作集和 projection 导入 / 导出能力，不再保留 JSON 快照导入 / 导出能力，也不写本地持久化文件；`auth-store.json` / `GENARRATIVE_AUTH_STORE_PATH` 不再是兼容恢复源。认证创建、登录会话、刷新、退出、改密、重置密码、绑定和资料变更等写操作仍必须在返回客户端前通过 `sync_auth_store_projection` 成功同步 SpacetimeDB 正式认证表；同步失败时接口返回错误，不允许把只存在于当前进程内存的账号、会话、短信验证码或微信 state 当成成功结果。每个 API 工作集绑定启动恢复或上次成功同步得到的 `auth_store_projection_meta.updated_at` 版本作为 `base_updated_at_micros`，SpacetimeDB 在同一事务内执行基线 CAS，并要求新的 `updated_at_micros` 严格递增；基线不一致或版本不晚于当前值时整包写入失败，冲突节点在确认本次同步前没有遗留未确认本地变更且同步期间没有其它本地认证变更后，才可丢弃本地工作集并从正式表恢复，不能用陈旧工作集删除、恢复或覆盖另一节点的新状态；若此前已有待重试 revision 或同期还有本地变更则保留旧基线并继续失败关闭，不用自动恢复覆盖未提交变更；同步成功但期间又出现新本地变更时最多连续补同步三轮，仍未稳定则失败关闭。这只是迁移期并发保护，不改变正式认证表的权威地位。新用户注册奖励、邀请码绑定和登录埋点必须排在认证同步成功之后，避免认证没落库时先写出钱包或邀请关系。若启动恢复阶段 SpacetimeDB 不可连接或超时，`api-server` 会按固定间隔持续重试认证工作集恢复，恢复成功后才开始监听 HTTP，避免一次短超时让进程永久停留在依赖不可用状态。 -`auth_store_snapshot` 表和旧 `import_auth_store_snapshot_json` / `export_auth_store_snapshot_from_tables` procedure 已删除。认证投影同步只读写 `user_account`、`auth_identity`、`refresh_session` 和 `auth_store_projection_meta`；`auth_identity` 不再写 `phone_e164`、`display_name`、`avatar_url`，这些账号资料只以 `user_account` 为准。 +`auth_store_snapshot` 表和旧 `import_auth_store_snapshot_json` / `export_auth_store_snapshot_from_tables` procedure 已删除。认证投影同步只读写 `user_account`、`auth_identity`、`refresh_session` 和 `auth_store_projection_meta`；`auth_identity` 不再写 `phone_e164`、`display_name`、`avatar_url`，这些账号资料只以 `user_account` 为准。`api-server` 多节点必须使用相同的部署级验证码哈希盐（当前复用 `GENARRATIVE_JWT_SECRET`）；轮换该 secret 会使尚未消费的短信验证码失效，但不会改变已持久化账号或 session。 ### `bark_battle_draft_config` @@ -1061,7 +1062,7 @@ Responses 的终态载荷既是工具调用的恢复源，也是正文的恢复 - 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。 +- 说明：跨节点资产退款的正式 pending 队列。主键为 refund ledger ID，保存 consume/refund 配对、用户、金额、资源、生成任务 attempt、失败原因和重试时间；失败事务先写入该表，worker 在 SpacetimeDB 事务内幂等执行钱包退款并删除成功行。只有数据库不可达时，api-server 才使用本机 `wallet-refund-outbox` emergency spool；本机 `MAX_BYTES` 达到阈值时改写入同目录 `refund-overflow-*` 溢出文件，保持可恢复而不静默丢弃。 - 索引：`(status, available_at)`。 ### `profile_wallet_config` diff --git a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md index 9691ad80d..cb42ae393 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 或因硬上限 / 保护阈值被丢弃的审计不在该保证内。 -钱包退款正式 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` 并纳入节点恢复 / 备份演练。 +钱包退款正式 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-*` 隔离目录。达到 `MAX_BYTES` 时不再静默丢弃退款，而是写入同一持久目录下的 `refund-overflow-*` 溢出文件并继续重放；溢出文件不计入普通容量阈值，但必须接入容量告警和人工补偿预案，底层磁盘写入失败仍按关键退款告警处理。worker 连接失败、库内 retry、emergency spool 写入 / 容量失败和 `corrupt-*` 出现都必须接入告警；人工补偿先按 refund ledger id 对账 `profile_wallet_ledger`、`asset_operation_wallet_settlement` 与两类 outbox，再通过受控退款 procedure 幂等重放，禁止直接手写钱包表。该目录不能替代库内 outbox；发布和主机替换必须保留 `/var/lib/genarrative/wallet-refund-outbox` 并纳入节点恢复 / 备份演练。容器 loadtest / 预览环境必须分别为 `api-server` 与 `external-generation-worker` 挂载各自的 tracking 与 wallet refund 命名卷，不能让节点重建清空本机恢复队列。 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`。修复顺序： @@ -866,7 +866,7 @@ systemctl restart genarrative-api.service journalctl -u genarrative-api.service --since '30 seconds ago' --no-pager | grep -E 'tracking outbox|Permission denied|os error 13' ``` -`Genarrative-Server-Provision` 和 `Genarrative-Api-Deploy` 会在保留旧 `/etc/genarrative/api-server.env` 的前提下补齐缺失的 tracking outbox 运行态路径，并确保 `/var/lib/genarrative/tracking-outbox` 归属 `genarrative:genarrative`。用户认证真相源只允许在 SpacetimeDB 正式认证表（`user_account` / `auth_identity` / `refresh_session`）恢复；不要再配置或依赖 `GENARRATIVE_AUTH_STORE_PATH` / `auth-store.json`，`module-auth` 也不再维护本地文件持久化；`auth_store_snapshot` 不再作为备查或运行期恢复源，只在正式认证表为空时一次性转移最新旧快照并清空，且旧 `get_auth_store_snapshot` / `upsert_auth_store_snapshot` / `import_auth_store_snapshot` 入口已经删除。如果 `api-server` 启动时连不上 SpacetimeDB，会持续重试启动恢复，直到认证工作集从 SpacetimeDB 正式表恢复成功后才开始监听 HTTP，以避免用空本地状态或旧快照覆盖认证表。 +`Genarrative-Server-Provision` 和 `Genarrative-Api-Deploy` 会在保留旧 `/etc/genarrative/api-server.env` 的前提下补齐缺失的 tracking outbox 运行态路径，并确保 `/var/lib/genarrative/tracking-outbox` 归属 `genarrative:genarrative`。用户认证真相源只允许从 SpacetimeDB 正式认证表（`user_account` / `auth_identity` / `refresh_session`）和 `auth_store_projection_meta` 中的短期验证码 / 微信 state 投影恢复；请求首次未命中短期状态时会从正式投影做一次只读刷新，不依赖粘性会话。不要再配置或依赖 `GENARRATIVE_AUTH_STORE_PATH` / `auth-store.json`，`module-auth` 也不再维护本地文件持久化；`auth_store_snapshot` 不再作为备查或运行期恢复源，只在正式认证表为空时一次性转移最新旧快照并清空，且旧 `get_auth_store_snapshot` / `upsert_auth_store_snapshot` / `import_auth_store_snapshot` 入口已经删除。所有 API 节点必须使用相同的 `GENARRATIVE_JWT_SECRET`，它也作为验证码哈希盐；轮换后尚未消费的验证码会失效。如果 `api-server` 启动时连不上 SpacetimeDB，会持续重试启动恢复，直到认证工作集从 SpacetimeDB 正式表和短期投影恢复成功后才开始监听 HTTP，以避免用空本地状态或旧快照覆盖认证表。 前端登录态恢复只把 `/api/auth/refresh` 的 `401` / `403` 当成权威失效信号；服务器重启窗口里的 `502` / `503` / `504`、浏览器 `Failed to fetch` 或 refresh 响应契约异常都必须保留已有本地 access token，不触发全局 auth 变化。refresh 成功响应以共享契约 `RefreshSessionResponse { token }` 为准，前端不要额外要求业务 `ok` 字段。排查“重启后用户都掉线”时，先区分前端是否被暂时不可用清掉本地 token，再检查 SpacetimeDB 正式认证表是否缺 `user_account` / `refresh_session` 数据。 diff --git a/server-rs/crates/api-server/src/asset_billing.rs b/server-rs/crates/api-server/src/asset_billing.rs index 83545d528..e0ce18d6d 100644 --- a/server-rs/crates/api-server/src/asset_billing.rs +++ b/server-rs/crates/api-server/src/asset_billing.rs @@ -579,7 +579,7 @@ async fn refund_asset_operation_points_with_job_id( "SpacetimeDB refund outbox 不可达，已写入本机 emergency spool" ); } - Ok(WalletRefundOutboxEnqueueOutcome::Dropped { reason }) => { + Ok(WalletRefundOutboxEnqueueOutcome::OverflowEnqueued { reason }) => { tracing::error!( owner_user_id, asset_kind, @@ -588,7 +588,7 @@ async fn refund_asset_operation_points_with_job_id( ledger_id, reason, error = %refund_error, - "SpacetimeDB refund outbox 不可达，且本机 emergency spool 因容量限制丢弃" + "SpacetimeDB refund outbox 不可达，退款已写入本机 emergency spool overflow 文件；需监控并尽快恢复库内队列" ); } Err(outbox_error) => { diff --git a/server-rs/crates/api-server/src/password_management.rs b/server-rs/crates/api-server/src/password_management.rs index 70f620791..38a0e40ab 100644 --- a/server-rs/crates/api-server/src/password_management.rs +++ b/server-rs/crates/api-server/src/password_management.rs @@ -9,6 +9,7 @@ use shared_contracts::auth::{ PasswordChangeRequest, PasswordChangeResponse, PasswordResetRequest, PasswordResetResponse, }; use time::OffsetDateTime; +use tracing::warn; use crate::{ api_response::json_success_body, @@ -81,7 +82,7 @@ pub async fn reset_password( ); } - let result = state + let result = match state .phone_auth_service() .reset_password( ResetPasswordInput { @@ -93,7 +94,17 @@ pub async fn reset_password( OffsetDateTime::now_utc(), ) .await - .map_err(map_phone_auth_error)?; + { + Ok(result) => result, + Err(error) => { + if let Err(sync_error) = state.sync_auth_store_tables_to_spacetime().await { + warn!(error = %sync_error, "重置密码失败后的短信验证码状态同步失败"); + return Err(AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message("同步短信验证码状态失败")); + } + return Err(map_phone_auth_error(error)); + } + }; let session_client = resolve_session_client_context(&headers); let signed_session = create_auth_session( &state, diff --git a/server-rs/crates/api-server/src/phone_auth.rs b/server-rs/crates/api-server/src/phone_auth.rs index a0e9cd753..02237d372 100644 --- a/server-rs/crates/api-server/src/phone_auth.rs +++ b/server-rs/crates/api-server/src/phone_auth.rs @@ -91,6 +91,14 @@ pub async fn send_phone_code( } }; + state + .sync_auth_store_tables_to_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("同步短信验证码状态失败：{error}")) + })?; + Ok(json_success_body( Some(&request_context), PhoneSendCodeResponse { @@ -115,18 +123,36 @@ pub async fn phone_login( ); } let invite_code = payload.invite_code.clone(); - let result = match state + let login_input = PhoneLoginInput { + country_code: payload.country_code, + pure_phone_number: payload.pure_phone_number, + verify_code: payload.code, + }; + let login_now = OffsetDateTime::now_utc(); + let login_result = state .phone_auth_service() - .login( - PhoneLoginInput { - country_code: payload.country_code, - pure_phone_number: payload.pure_phone_number, - verify_code: payload.code, - }, - OffsetDateTime::now_utc(), - ) - .await - { + .login(login_input.clone(), login_now) + .await; + let login_result = match login_result { + Err(PhoneAuthError::VerifyCodeNotFound) => { + if let Err(sync_error) = state.refresh_auth_store_from_spacetime().await { + warn!( + request_id = request_context.request_id(), + operation = request_context.operation(), + error = %sync_error, + "手机号验证码未命中后的认证投影刷新失败" + ); + return Err(AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message("刷新短信验证码状态失败")); + } + state + .phone_auth_service() + .login(login_input, login_now) + .await + } + result => result, + }; + let result = match login_result { Ok(result) => { info!( request_id = request_context.request_id(), @@ -142,6 +168,16 @@ pub async fn phone_login( result } Err(error) => { + if let Err(sync_error) = state.sync_auth_store_tables_to_spacetime().await { + warn!( + request_id = request_context.request_id(), + operation = request_context.operation(), + error = %sync_error, + "手机号验证码登录失败后的状态同步失败" + ); + return Err(AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message("同步短信验证码状态失败")); + } warn!( request_id = request_context.request_id(), operation = request_context.operation(), diff --git a/server-rs/crates/api-server/src/state.rs b/server-rs/crates/api-server/src/state.rs index 600a1d0e1..cf9792f80 100644 --- a/server-rs/crates/api-server/src/state.rs +++ b/server-rs/crates/api-server/src/state.rs @@ -571,7 +571,11 @@ impl AppState { ORPHAN_WORK_AUTHOR_PUBLIC_USER_CODE, ) .map_err(|error| AppStateInitError::AuthStore(error.to_string()))?; - let phone_auth_service = PhoneAuthService::new(auth_store.clone(), sms_provider); + let phone_auth_service = PhoneAuthService::new_with_verify_code_salt( + auth_store.clone(), + sms_provider, + config.jwt_secret.clone(), + ); let wechat_auth_state_service = WechatAuthStateService::new(auth_store.clone(), config.wechat_state_ttl_minutes); let wechat_auth_service = WechatAuthService::new(auth_store.clone()); @@ -1402,6 +1406,43 @@ impl AppState { )) } + /// 在只读短期认证状态未命中时，从正式投影刷新一次本地工作集，避免 OAuth / 短信 + /// 请求落到另一节点后必须依赖粘性会话才能成功。 + pub async fn refresh_auth_store_from_spacetime(&self) -> Result<(), SpacetimeClientError> { + #[cfg(test)] + return Ok(()); + + #[cfg(not(test))] + { + let _sync_guard = self.auth_projection_sync_lock.lock().await; + let expected_revision = self.auth_store.revision(); + if self.auth_projection_synced_revision.load(Ordering::Acquire) != expected_revision { + return Err(SpacetimeClientError::Runtime( + "认证工作集存在待同步变更，跳过只读刷新".to_string(), + )); + } + let projection = self + .spacetime_client + .export_auth_store_projection_from_tables() + .await?; + let updated_at_micros = projection.updated_at_micros; + let refreshed = self + .auth_store + .refresh_from_projection_view_if_revision(projection, expected_revision) + .map_err(SpacetimeClientError::Runtime)?; + if !refreshed { + return Err(SpacetimeClientError::Runtime( + "认证工作集刷新期间发生并发变更".to_string(), + )); + } + self.auth_projection_version + .store(updated_at_micros, Ordering::Release); + self.auth_projection_synced_revision + .store(self.auth_store.revision(), Ordering::Release); + Ok(()) + } + } + pub async fn try_restore_auth_store_from_spacetime( config: AppConfig, ) -> Result { @@ -1946,6 +1987,8 @@ fn auth_store_candidate_from_projection_view( if projection.users.is_empty() && projection.identities.is_empty() && projection.refresh_sessions.is_empty() + && projection.phone_codes.is_empty() + && projection.wechat_states.is_empty() && projection.updated_at_micros == 0 { return Ok(None); 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 bb2232a52..6d79051ff 100644 --- a/server-rs/crates/api-server/src/wallet_refund_outbox.rs +++ b/server-rs/crates/api-server/src/wallet_refund_outbox.rs @@ -19,6 +19,7 @@ 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"; @@ -118,7 +119,7 @@ pub(crate) struct WalletRefundOutboxRecord { #[derive(Debug)] pub enum WalletRefundOutboxEnqueueOutcome { Enqueued, - Dropped { reason: &'static str }, + OverflowEnqueued { reason: &'static str }, } #[derive(Debug)] @@ -153,9 +154,13 @@ impl WalletRefundOutbox { 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); @@ -164,11 +169,12 @@ impl WalletRefundOutbox { 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); - if current_bytes.saturating_add(line_bytes) > self.max_bytes { - return Ok(WalletRefundOutboxEnqueueOutcome::Dropped { - reason: "max_bytes", - }); - } + 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() @@ -183,22 +189,31 @@ impl WalletRefundOutbox { 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, &pending_path).await { + 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(WalletRefundOutboxEnqueueOutcome::Enqueued); + 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(&pending_path, &record) + .reuse_existing_pending_file(target_path, &record) .await? { remove_file_and_sync(&temp_path, &self.dir).await?; @@ -213,7 +228,7 @@ impl WalletRefundOutbox { std::io::ErrorKind::AlreadyExists, format!( "refund pending path could not be installed: {}", - pending_path.display() + target_path.display() ), ) .into()) @@ -369,7 +384,7 @@ impl WalletRefundOutbox { async fn recover_temporary_files(&self) -> Result<(), WalletRefundOutboxError> { let _guard = self.enqueue_lock.lock().await; let temporary_files = self.list_temporary_files().await?; - for path in temporary_files { + '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() => { @@ -385,82 +400,72 @@ impl WalletRefundOutbox { }; let pending_path = self.pending_path_for_ledger(&record.ledger_id); - match fs::metadata(&pending_path).await { - Ok(metadata) if metadata.is_file() => { - match read_refund_record(&pending_path).await { - Ok(existing) if existing == record => { - remove_file_and_sync(&path, &self.dir).await?; - debug!( - ledger_id = %record.ledger_id, - source = %path.display(), - target = %pending_path.display(), - "wallet refund outbox 临时文件与已有幂等文件重复，已删除临时副本" - ); - continue; - } - Ok(existing) if existing.ledger_id == record.ledger_id => { - self.quarantine_file(&path).await?; - warn!( - ledger_id = %record.ledger_id, - source = %path.display(), - target = %pending_path.display(), - "wallet refund outbox 临时文件与已有幂等文件事实不一致，已隔离临时文件" - ); - continue; - } - Err(error) if error.is_data_corruption() => { - self.quarantine_file(&pending_path).await?; - } - Ok(_) => { - self.quarantine_file(&path).await?; - warn!( - ledger_id = %record.ledger_id, - source = %path.display(), - target = %pending_path.display(), - "wallet refund outbox 临时文件命名冲突，已隔离临时文件" - ); - continue; - } - Err(error) => return Err(error), + 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), } - Ok(_) => { - return Err(std::io::Error::new( - std::io::ErrorKind::AlreadyExists, - format!( - "refund pending path is not a regular file: {}", - pending_path.display() - ), - ) - .into()); - } - Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} - Err(error) => return Err(error.into()), } - match fs::hard_link(&path, &pending_path).await { + 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 = %pending_path.display(), + target = %target_path.display(), "wallet refund outbox 崩溃遗留临时文件已恢复为幂等退款文件" ); } Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { - match self - .reuse_existing_pending_file(&pending_path, &record) - .await - { + 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, &pending_path).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?; @@ -475,7 +480,7 @@ impl WalletRefundOutbox { warn!( ledger_id = %record.ledger_id, source = %path.display(), - target = %pending_path.display(), + target = %target_path.display(), "wallet refund outbox 临时文件无法与现有幂等文件合并，已隔离临时文件" ); } @@ -488,7 +493,7 @@ impl WalletRefundOutbox { warn!( ledger_id = %record.ledger_id, source = %path.display(), - target = %pending_path.display(), + target = %target_path.display(), error = %error, "wallet refund outbox 临时文件与现有幂等文件事实冲突，已隔离临时文件" ); @@ -537,7 +542,9 @@ impl WalletRefundOutbox { let Some(name) = path.file_name().and_then(|value| value.to_str()) else { continue; }; - if name.starts_with(PENDING_FILE_PREFIX) && name.ends_with(OUTBOX_FILE_EXTENSION) { + if (name.starts_with(PENDING_FILE_PREFIX) || name.starts_with(OVERFLOW_FILE_PREFIX)) + && name.ends_with(OUTBOX_FILE_EXTENSION) + { files.push(path); } } @@ -552,6 +559,13 @@ impl WalletRefundOutbox { )) } + 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}", @@ -642,7 +656,7 @@ fn directory_size_if_exists(path: &Path) -> Result { let mut total = 0u64; for entry in std::fs::read_dir(path)? { let entry = entry?; - if !is_pending_outbox_file_name(&entry.file_name()) { + if !is_capped_outbox_file_name(&entry.file_name()) { continue; } let metadata = entry.metadata()?; @@ -666,7 +680,17 @@ fn ledger_id_hash(ledger_id: &str) -> String { 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(TEMP_FILE_PREFIX)) + (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) }) } @@ -786,7 +810,7 @@ mod tests { } #[tokio::test] - async fn enqueue_drops_when_outbox_exceeds_max_bytes() { + 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); @@ -794,11 +818,18 @@ mod tests { assert!(matches!( outcome, - WalletRefundOutboxEnqueueOutcome::Dropped { + WalletRefundOutboxEnqueueOutcome::OverflowEnqueued { reason: "max_bytes" } )); - assert!(!dir.exists() || std::fs::read_dir(&dir).unwrap().next().is_none()); + 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); } @@ -899,6 +930,30 @@ mod tests { 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"); diff --git a/server-rs/crates/api-server/src/wechat/auth.rs b/server-rs/crates/api-server/src/wechat/auth.rs index 420438146..d4f26dcd2 100644 --- a/server-rs/crates/api-server/src/wechat/auth.rs +++ b/server-rs/crates/api-server/src/wechat/auth.rs @@ -16,6 +16,7 @@ use shared_contracts::auth::{ }; use shared_kernel::normalize_optional_string; use time::OffsetDateTime; +use tracing::warn; use url::Url; use crate::{ @@ -62,6 +63,13 @@ pub async fn start_wechat_login( OffsetDateTime::now_utc(), ) .map_err(map_wechat_auth_error)?; + state + .sync_auth_store_tables_to_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("同步微信登录状态失败：{error}")) + })?; let authorization_url = state .wechat_provider() .build_authorization_url( @@ -107,6 +115,13 @@ pub async fn start_wechat_bind( OffsetDateTime::now_utc(), ) .map_err(map_wechat_auth_error)?; + state + .sync_auth_store_tables_to_spacetime() + .await + .map_err(|error| { + AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message(format!("同步微信绑定状态失败：{error}")) + })?; let authorization_url = state .wechat_provider() .build_authorization_url( @@ -149,11 +164,45 @@ pub async fn handle_wechat_callback( .into_response()); } - let consumed = match state + let consume_result = state .wechat_auth_state_service() - .consume_state(&state_token, OffsetDateTime::now_utc()) - { + .consume_state(&state_token, OffsetDateTime::now_utc()); + let consumed = match consume_result { Ok(value) => value, + Err(WechatAuthError::StateNotFound) => { + if let Err(error) = state.refresh_auth_store_from_spacetime().await { + warn!( + request_id = request_context.request_id(), + operation = request_context.operation(), + error = %error, + "微信 state 未命中后的认证投影刷新失败" + ); + return Ok(Redirect::to(&build_auth_result_redirect_url( + &fallback_redirect, + &[ + ("auth_provider", "wechat"), + ("auth_error", "微信登录状态已失效，请重新发起登录。"), + ], + )) + .into_response()); + } + match state + .wechat_auth_state_service() + .consume_state(&state_token, OffsetDateTime::now_utc()) + { + Ok(value) => value, + Err(_) => { + return Ok(Redirect::to(&build_auth_result_redirect_url( + &fallback_redirect, + &[ + ("auth_provider", "wechat"), + ("auth_error", "微信登录状态已失效，请重新发起登录。"), + ], + )) + .into_response()); + } + } + } Err(_) => { return Ok(Redirect::to(&build_auth_result_redirect_url( &fallback_redirect, @@ -165,6 +214,22 @@ pub async fn handle_wechat_callback( .into_response()); } }; + if let Err(error) = state.sync_auth_store_tables_to_spacetime().await { + warn!( + request_id = request_context.request_id(), + operation = request_context.operation(), + error = %error, + "微信回调消费 state 后同步失败" + ); + return Ok(Redirect::to(&build_auth_result_redirect_url( + &fallback_redirect, + &[ + ("auth_provider", "wechat"), + ("auth_error", "微信登录服务暂时不可用，请稍后重试。"), + ], + )) + .into_response()); + } let redirect_path = consumed.state.redirect_path.clone(); let session_client = resolve_session_client_context(&headers); @@ -293,7 +358,7 @@ pub async fn bind_wechat_phone( .ok_or_else(|| { AppError::from_status(StatusCode::BAD_REQUEST).with_message("缺少短信验证码") })?; - state + match state .phone_auth_service() .bind_wechat_phone( BindWechatPhoneInput { @@ -306,7 +371,17 @@ pub async fn bind_wechat_phone( OffsetDateTime::now_utc(), ) .await - .map_err(map_wechat_bind_phone_error)? + { + Ok(result) => result, + Err(error) => { + if let Err(sync_error) = state.sync_auth_store_tables_to_spacetime().await { + warn!(error = %sync_error, "微信绑定手机号失败后的短信验证码状态同步失败"); + return Err(AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) + .with_message("同步短信验证码状态失败")); + } + return Err(map_wechat_bind_phone_error(error)); + } + } }; let session_client = resolve_session_client_context(&headers); let signed_session = create_auth_session( diff --git a/server-rs/crates/module-auth/src/domain.rs b/server-rs/crates/module-auth/src/domain.rs index 90c12cd90..4a29af646 100644 --- a/server-rs/crates/module-auth/src/domain.rs +++ b/server-rs/crates/module-auth/src/domain.rs @@ -81,7 +81,7 @@ pub struct PhoneNumberSnapshot { } /// 手机验证码使用场景。 -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum PhoneAuthScene { Login, BindPhone, @@ -101,7 +101,7 @@ impl PhoneAuthScene { } /// 微信授权入口场景。 -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum WechatAuthScene { Desktop, WechatInApp, @@ -187,6 +187,10 @@ pub struct AuthStoreProjectionView { pub users: Vec, pub identities: Vec, pub refresh_sessions: Vec, + #[serde(default)] + pub phone_codes: Vec, + #[serde(default)] + pub wechat_states: Vec, /// 当前进程工作集所基于的正式投影版本，用于事务内 CAS。 #[serde(default)] pub base_updated_at_micros: i64, @@ -233,6 +237,31 @@ pub struct AuthStoreProjectionRefreshSession { pub last_seen_at: String, } +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct AuthStoreProjectionPhoneCode { + pub phone_number: String, + pub scene: String, + pub verify_code_hash: String, + pub expires_at: String, + pub last_sent_at: String, + pub failed_attempts: u32, + pub provider_out_id: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct AuthStoreProjectionWechatState { + pub wechat_state_id: String, + pub state_token: String, + pub redirect_path: String, + pub scene: String, + pub request_user_agent: Option, + pub bind_user_id: Option, + pub expires_at: String, + pub consumed_at: Option, + pub created_at: String, + pub updated_at: String, +} + pub fn validate_password(password: &str) -> Result<(), PasswordEntryError> { let length = password.chars().count(); if !(PASSWORD_MIN_LENGTH..=PASSWORD_MAX_LENGTH).contains(&length) { diff --git a/server-rs/crates/module-auth/src/lib.rs b/server-rs/crates/module-auth/src/lib.rs index a35682b04..586e3eec3 100644 --- a/server-rs/crates/module-auth/src/lib.rs +++ b/server-rs/crates/module-auth/src/lib.rs @@ -31,6 +31,8 @@ use shared_kernel::{ use time::{Duration, OffsetDateTime}; use tracing::{info, warn}; +const DEFAULT_PHONE_VERIFY_CODE_SALT: &str = "genarrative-phone-verify-code-v1"; + #[derive(Clone, Debug)] pub struct InMemoryAuthStore { inner: Arc>, @@ -135,6 +137,24 @@ fn parse_auth_binding_status(value: &str) -> AuthBindingStatus { } } +fn parse_phone_auth_scene(value: &str) -> Option { + match value.trim() { + "login" => Some(PhoneAuthScene::Login), + "bind_phone" => Some(PhoneAuthScene::BindPhone), + "change_phone" => Some(PhoneAuthScene::ChangePhone), + "reset_password" => Some(PhoneAuthScene::ResetPassword), + _ => None, + } +} + +fn parse_wechat_auth_scene(value: &str) -> Option { + match value.trim() { + "desktop" => Some(WechatAuthScene::Desktop), + "wechat_in_app" => Some(WechatAuthScene::WechatInApp), + _ => None, + } +} + fn next_sequence_from_public_user_code(public_user_code: &str) -> u64 { public_user_code .trim() @@ -503,10 +523,19 @@ impl RefreshSessionService { impl PhoneAuthService { pub fn new(store: InMemoryAuthStore, sms_provider: SmsAuthProvider) -> Self { + Self::new_with_verify_code_salt(store, sms_provider, DEFAULT_PHONE_VERIFY_CODE_SALT) + } + + /// 使用部署级稳定盐值构造服务，确保验证码投影恢复到另一节点后仍可校验。 + pub fn new_with_verify_code_salt( + store: InMemoryAuthStore, + sms_provider: SmsAuthProvider, + verify_code_salt: impl Into, + ) -> Self { Self { store, sms_provider, - verify_code_salt: new_uuid_simple_string(), + verify_code_salt: verify_code_salt.into(), } } @@ -1026,6 +1055,8 @@ impl InMemoryAuthStoreState { let mut wechat_identity_by_provider_uid = HashMap::new(); let mut user_id_by_provider_union_id = HashMap::new(); let mut phone_number_by_user_id = HashMap::new(); + let mut phone_codes_by_key = HashMap::new(); + let mut wechat_states_by_token = HashMap::new(); for user in &view.users { if let Some(phone_number) = normalize_optional_string(user.phone_number_e164.clone()) { @@ -1098,6 +1129,46 @@ impl InMemoryAuthStoreState { ); } + for phone_code in view.phone_codes { + let scene = parse_phone_auth_scene(&phone_code.scene) + .ok_or_else(|| format!("未知短信验证码场景：{}", phone_code.scene))?; + let key = build_phone_code_key(&phone_code.phone_number, &scene); + phone_codes_by_key.insert( + key, + StoredPhoneCode { + phone_number: phone_code.phone_number, + scene, + verify_code_hash: phone_code.verify_code_hash, + expires_at: phone_code.expires_at, + last_sent_at: phone_code.last_sent_at, + failed_attempts: phone_code.failed_attempts, + provider_out_id: phone_code.provider_out_id, + }, + ); + } + + for wechat_state in view.wechat_states { + let scene = parse_wechat_auth_scene(&wechat_state.scene) + .ok_or_else(|| format!("未知微信授权 state 场景：{}", wechat_state.scene))?; + wechat_states_by_token.insert( + wechat_state.state_token.clone(), + StoredWechatAuthState { + state: WechatAuthStateRecord { + wechat_state_id: wechat_state.wechat_state_id, + state_token: wechat_state.state_token, + redirect_path: wechat_state.redirect_path, + scene, + request_user_agent: wechat_state.request_user_agent, + bind_user_id: wechat_state.bind_user_id, + expires_at: wechat_state.expires_at, + consumed_at: wechat_state.consumed_at, + created_at: wechat_state.created_at, + updated_at: wechat_state.updated_at, + }, + }, + ); + } + for user in view.users { let wechat_identity = wechat_identity_by_provider_uid .values() @@ -1145,8 +1216,8 @@ impl InMemoryAuthStoreState { phone_to_user_id, sessions_by_id, session_id_by_refresh_token_hash, - phone_codes_by_key: HashMap::new(), - wechat_states_by_token: HashMap::new(), + phone_codes_by_key, + wechat_states_by_token, wechat_identity_by_provider_uid, user_id_by_provider_union_id, }) @@ -1158,11 +1229,33 @@ impl InMemoryAuthStoreState { self.phone_to_user_id = next_state.phone_to_user_id; self.sessions_by_id = next_state.sessions_by_id; self.session_id_by_refresh_token_hash = next_state.session_id_by_refresh_token_hash; + self.phone_codes_by_key = next_state.phone_codes_by_key; + self.wechat_states_by_token = next_state.wechat_states_by_token; self.wechat_identity_by_provider_uid = next_state.wechat_identity_by_provider_uid; self.user_id_by_provider_union_id = next_state.user_id_by_provider_union_id; } } +fn prune_expired_short_lived_state( + state: &mut InMemoryAuthStoreState, + now: OffsetDateTime, +) -> bool { + let phone_code_count = state.phone_codes_by_key.len(); + state.phone_codes_by_key.retain(|_, code| { + parse_rfc3339(&code.expires_at) + .map(|expires_at| expires_at > now) + .unwrap_or(true) + }); + let wechat_state_count = state.wechat_states_by_token.len(); + state.wechat_states_by_token.retain(|_, stored| { + parse_rfc3339(&stored.state.expires_at) + .map(|expires_at| expires_at > now) + .unwrap_or(true) + }); + phone_code_count != state.phone_codes_by_key.len() + || wechat_state_count != state.wechat_states_by_token.len() +} + impl InMemoryAuthStore { pub fn from_projection_view(view: AuthStoreProjectionView) -> Result { Ok(Self { @@ -1215,10 +1308,14 @@ impl InMemoryAuthStore { &self, updated_at_micros: i64, ) -> Result { - let state = self + let mut state = self .inner .lock() .map_err(|_| "认证仓储锁已中毒".to_string())?; + let pruned = prune_expired_short_lived_state(&mut state, OffsetDateTime::now_utc()); + if pruned { + self.revision.fetch_add(1, Ordering::Release); + } let users = state .users_by_username .values() @@ -1283,6 +1380,35 @@ impl InMemoryAuthStore { }) }) .collect::, String>>()?; + let phone_codes = state + .phone_codes_by_key + .values() + .map(|stored| AuthStoreProjectionPhoneCode { + phone_number: stored.phone_number.clone(), + scene: stored.scene.as_str().to_string(), + verify_code_hash: stored.verify_code_hash.clone(), + expires_at: stored.expires_at.clone(), + last_sent_at: stored.last_sent_at.clone(), + failed_attempts: stored.failed_attempts, + provider_out_id: stored.provider_out_id.clone(), + }) + .collect(); + let wechat_states = state + .wechat_states_by_token + .values() + .map(|stored| AuthStoreProjectionWechatState { + wechat_state_id: stored.state.wechat_state_id.clone(), + state_token: stored.state.state_token.clone(), + redirect_path: stored.state.redirect_path.clone(), + scene: stored.state.scene.as_str().to_string(), + request_user_agent: stored.state.request_user_agent.clone(), + bind_user_id: stored.state.bind_user_id.clone(), + expires_at: stored.state.expires_at.clone(), + consumed_at: stored.state.consumed_at.clone(), + created_at: stored.state.created_at.clone(), + updated_at: stored.state.updated_at.clone(), + }) + .collect(); Ok(AuthStoreProjectionView { base_updated_at_micros: 0, @@ -1290,6 +1416,8 @@ impl InMemoryAuthStore { users, identities, refresh_sessions, + phone_codes, + wechat_states, }) } @@ -1966,6 +2094,7 @@ impl InMemoryAuthStore { // 手机号和业务场景共同决定同一份验证码快照，重复发送时直接覆盖旧值。 let key = build_phone_code_key(&code.phone_number, &code.scene); state.phone_codes_by_key.insert(key, code); + self.persist_phone_state(&state)?; Ok(()) } @@ -2047,6 +2176,7 @@ impl InMemoryAuthStore { .map_err(|_| PhoneAuthError::Store("短信验证码仓储锁已中毒".to_string()))?; let key = build_phone_code_key(phone_number, scene); state.phone_codes_by_key.remove(&key); + self.persist_phone_state(&state)?; Ok(()) } @@ -2066,11 +2196,13 @@ impl InMemoryAuthStore { let next_failed_attempts = stored.failed_attempts.saturating_add(1); if next_failed_attempts >= SMS_CODE_MAX_FAILED_ATTEMPTS { state.phone_codes_by_key.remove(&key); + self.persist_phone_state(&state)?; return Err(PhoneAuthError::VerifyAttemptsExceeded); } if let Some(current) = state.phone_codes_by_key.get_mut(&key) { current.failed_attempts = next_failed_attempts; } + self.persist_phone_state(&state)?; Err(PhoneAuthError::InvalidVerifyCode) } @@ -2094,6 +2226,7 @@ impl InMemoryAuthStore { state: state_record, }, ); + self.persist_wechat_state(&state)?; Ok(()) } @@ -2131,7 +2264,10 @@ impl InMemoryAuthStore { .ok_or(WechatAuthError::StateNotFound)?; current.state.consumed_at = Some(now_iso.clone()); current.state.updated_at = now_iso; - Ok(current.clone()) + let consumed = current.clone(); + state.wechat_states_by_token.remove(state_token.trim()); + self.persist_wechat_state(&state)?; + Ok(consumed) } fn bind_wechat_phone_to_user( @@ -2835,6 +2971,8 @@ mod tests { users: vec![], identities: vec![], refresh_sessions: vec![], + phone_codes: vec![], + wechat_states: vec![], }) .expect("projection should restore") } @@ -3265,6 +3403,8 @@ mod tests { )], identities: vec![], refresh_sessions: vec![], + phone_codes: vec![], + wechat_states: vec![], }) .expect("projection should restore"), ); @@ -3347,6 +3487,64 @@ mod tests { assert_eq!(rotated.user.id, user.id); } + #[tokio::test] + async fn projection_roundtrip_preserves_phone_code_and_wechat_state() { + let store = InMemoryAuthStore::default(); + let phone_service = build_phone_service(store.clone()); + let wechat_state_service = WechatAuthStateService::new(store.clone(), 5); + let now = OffsetDateTime::now_utc(); + + phone_service + .send_code( + SendPhoneCodeInput { + country_code: None, + pure_phone_number: "13800138040".to_string(), + scene: PhoneAuthScene::Login, + }, + now, + ) + .await + .expect("phone code should send before projection export"); + let created_state = wechat_state_service + .create_state( + CreateWechatAuthStateInput { + redirect_path: "/studio".to_string(), + scene: WechatAuthScene::Desktop, + request_user_agent: Some("test-agent".to_string()), + bind_user_id: None, + }, + now, + ) + .expect("wechat state should be created before projection export"); + + let projection = store + .export_projection_view(1) + .expect("projection export should include short-lived auth state"); + assert_eq!(projection.phone_codes.len(), 1); + assert_eq!(projection.wechat_states.len(), 1); + + let restored_store = InMemoryAuthStore::from_projection_view(projection) + .expect("projection should restore short-lived auth state"); + let restored_phone_service = build_phone_service(restored_store.clone()); + let login = restored_phone_service + .login( + PhoneLoginInput { + country_code: None, + pure_phone_number: "13800138040".to_string(), + verify_code: DEFAULT_SMS_MOCK_VERIFY_CODE.to_string(), + }, + now + Duration::seconds(1), + ) + .await + .expect("restored phone code should verify"); + assert!(login.created); + + let consumed_state = WechatAuthStateService::new(restored_store, 5) + .consume_state(&created_state.state.state_token, now + Duration::seconds(1)) + .expect("restored wechat state should be consumable"); + assert_eq!(consumed_state.state.redirect_path, "/studio"); + } + #[tokio::test] async fn refresh_from_projection_view_merges_session_created_by_another_process() { let source_store = InMemoryAuthStore::default(); @@ -3393,19 +3591,19 @@ mod tests { ) .expect("refreshed session active check should succeed") ); - assert!(matches!( - local_phone_service - .send_code( - SendPhoneCodeInput { - country_code: None, - pure_phone_number: "13800138034".to_string(), - scene: PhoneAuthScene::Login, - }, - local_now + Duration::seconds(5), - ) - .await, - Err(PhoneAuthError::SendCoolingDown { .. }) - )); + // 刷新到数据库正式投影后，短期认证状态也以数据库快照为准；本地未同步的验证码 + // 不得继续留在工作集里，避免消费已被其他节点清理的验证码。 + local_phone_service + .send_code( + SendPhoneCodeInput { + country_code: None, + pure_phone_number: "13800138034".to_string(), + scene: PhoneAuthScene::Login, + }, + local_now + Duration::seconds(5), + ) + .await + .expect("phone code should be resendable after authoritative refresh"); } #[test] @@ -3417,6 +3615,8 @@ mod tests { users: vec![], identities: vec![], refresh_sessions: vec![], + phone_codes: vec![], + wechat_states: vec![], }; assert_eq!(store.revision(), 0); @@ -4330,6 +4530,8 @@ mod tests { )], identities: vec![], refresh_sessions: vec![], + phone_codes: vec![], + wechat_states: vec![], }) .expect("projection should restore"); let phone_service = build_phone_service(store.clone()); diff --git a/server-rs/crates/spacetime-client/src/active/mapper/auth.rs b/server-rs/crates/spacetime-client/src/active/mapper/auth.rs index fd32f8fdc..3195f4514 100644 --- a/server-rs/crates/spacetime-client/src/active/mapper/auth.rs +++ b/server-rs/crates/spacetime-client/src/active/mapper/auth.rs @@ -105,6 +105,39 @@ pub(crate) fn map_auth_store_projection_view_input( }, ) .collect(), + phone_codes: view + .phone_codes + .into_iter() + .map( + |code| crate::module_bindings::AuthStoreProjectionPhoneCode { + phone_number: code.phone_number, + scene: code.scene, + verify_code_hash: code.verify_code_hash, + expires_at: code.expires_at, + last_sent_at: code.last_sent_at, + failed_attempts: code.failed_attempts, + provider_out_id: code.provider_out_id, + }, + ) + .collect(), + wechat_states: view + .wechat_states + .into_iter() + .map( + |state| crate::module_bindings::AuthStoreProjectionWechatState { + wechat_state_id: state.wechat_state_id, + state_token: state.state_token, + redirect_path: state.redirect_path, + scene: state.scene, + request_user_agent: state.request_user_agent, + bind_user_id: state.bind_user_id, + expires_at: state.expires_at, + consumed_at: state.consumed_at, + created_at: state.created_at, + updated_at: state.updated_at, + }, + ) + .collect(), } } @@ -161,5 +194,34 @@ fn map_auth_store_projection_view( last_seen_at: session.last_seen_at, }) .collect(), + phone_codes: view + .phone_codes + .into_iter() + .map(|code| module_auth::AuthStoreProjectionPhoneCode { + phone_number: code.phone_number, + scene: code.scene, + verify_code_hash: code.verify_code_hash, + expires_at: code.expires_at, + last_sent_at: code.last_sent_at, + failed_attempts: code.failed_attempts, + provider_out_id: code.provider_out_id, + }) + .collect(), + wechat_states: view + .wechat_states + .into_iter() + .map(|state| module_auth::AuthStoreProjectionWechatState { + wechat_state_id: state.wechat_state_id, + state_token: state.state_token, + redirect_path: state.redirect_path, + scene: state.scene, + request_user_agent: state.request_user_agent, + bind_user_id: state.bind_user_id, + expires_at: state.expires_at, + consumed_at: state.consumed_at, + created_at: state.created_at, + updated_at: state.updated_at, + }) + .collect(), } } diff --git a/server-rs/crates/spacetime-client/src/mapper/auth.rs b/server-rs/crates/spacetime-client/src/mapper/auth.rs index b85eeb959..0e0c7b5ac 100644 --- a/server-rs/crates/spacetime-client/src/mapper/auth.rs +++ b/server-rs/crates/spacetime-client/src/mapper/auth.rs @@ -88,6 +88,35 @@ pub(crate) fn map_auth_store_projection_view_input( }, ) .collect(), + phone_codes: view + .phone_codes + .into_iter() + .map(|code| crate::module_bindings::AuthStoreProjectionPhoneCode { + phone_number: code.phone_number, + scene: code.scene, + verify_code_hash: code.verify_code_hash, + expires_at: code.expires_at, + last_sent_at: code.last_sent_at, + failed_attempts: code.failed_attempts, + provider_out_id: code.provider_out_id, + }) + .collect(), + wechat_states: view + .wechat_states + .into_iter() + .map(|state| crate::module_bindings::AuthStoreProjectionWechatState { + wechat_state_id: state.wechat_state_id, + state_token: state.state_token, + redirect_path: state.redirect_path, + scene: state.scene, + request_user_agent: state.request_user_agent, + bind_user_id: state.bind_user_id, + expires_at: state.expires_at, + consumed_at: state.consumed_at, + created_at: state.created_at, + updated_at: state.updated_at, + }) + .collect(), } } @@ -144,5 +173,34 @@ fn map_auth_store_projection_view( last_seen_at: session.last_seen_at, }) .collect(), + phone_codes: view + .phone_codes + .into_iter() + .map(|code| module_auth::AuthStoreProjectionPhoneCode { + phone_number: code.phone_number, + scene: code.scene, + verify_code_hash: code.verify_code_hash, + expires_at: code.expires_at, + last_sent_at: code.last_sent_at, + failed_attempts: code.failed_attempts, + provider_out_id: code.provider_out_id, + }) + .collect(), + wechat_states: view + .wechat_states + .into_iter() + .map(|state| module_auth::AuthStoreProjectionWechatState { + wechat_state_id: state.wechat_state_id, + state_token: state.state_token, + redirect_path: state.redirect_path, + scene: state.scene, + request_user_agent: state.request_user_agent, + bind_user_id: state.bind_user_id, + expires_at: state.expires_at, + consumed_at: state.consumed_at, + created_at: state.created_at, + updated_at: state.updated_at, + }) + .collect(), } } diff --git a/server-rs/crates/spacetime-client/src/module_bindings.rs b/server-rs/crates/spacetime-client/src/module_bindings.rs index 64674f251..b1fc29d7b 100644 --- a/server-rs/crates/spacetime-client/src/module_bindings.rs +++ b/server-rs/crates/spacetime-client/src/module_bindings.rs @@ -120,12 +120,14 @@ pub mod auth_session_validation_procedure_result_type; pub mod auth_store_projection_identity_type; pub mod auth_store_projection_meta_table; pub mod auth_store_projection_meta_type; +pub mod auth_store_projection_phone_code_type; pub mod auth_store_projection_procedure_result_type; pub mod auth_store_projection_refresh_session_type; pub mod auth_store_projection_sync_procedure_result_type; pub mod auth_store_projection_sync_record_type; pub mod auth_store_projection_user_type; pub mod auth_store_projection_view_type; +pub mod auth_store_projection_wechat_state_type; pub mod authenticate_external_api_key_and_return_procedure; pub mod authorize_database_migration_operator_procedure; pub mod backfill_editor_canvas_layout_and_return_procedure; @@ -989,12 +991,14 @@ pub use auth_session_validation_procedure_result_type::AuthSessionValidationProc pub use auth_store_projection_identity_type::AuthStoreProjectionIdentity; pub use auth_store_projection_meta_table::*; pub use auth_store_projection_meta_type::AuthStoreProjectionMeta; +pub use auth_store_projection_phone_code_type::AuthStoreProjectionPhoneCode; pub use auth_store_projection_procedure_result_type::AuthStoreProjectionProcedureResult; pub use auth_store_projection_refresh_session_type::AuthStoreProjectionRefreshSession; pub use auth_store_projection_sync_procedure_result_type::AuthStoreProjectionSyncProcedureResult; pub use auth_store_projection_sync_record_type::AuthStoreProjectionSyncRecord; pub use auth_store_projection_user_type::AuthStoreProjectionUser; pub use auth_store_projection_view_type::AuthStoreProjectionView; +pub use auth_store_projection_wechat_state_type::AuthStoreProjectionWechatState; pub use authenticate_external_api_key_and_return_procedure::authenticate_external_api_key_and_return; pub use authorize_database_migration_operator_procedure::authorize_database_migration_operator; pub use backfill_editor_canvas_layout_and_return_procedure::backfill_editor_canvas_layout_and_return; diff --git a/server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_meta_type.rs b/server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_meta_type.rs index 309dceff7..2e7b2d303 100644 --- a/server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_meta_type.rs +++ b/server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_meta_type.rs @@ -9,6 +9,8 @@ use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws}; pub struct AuthStoreProjectionMeta { pub meta_id: String, pub updated_at: __sdk::Timestamp, + pub phone_codes_json: Option, + pub wechat_states_json: Option, } impl __sdk::InModule for AuthStoreProjectionMeta { @@ -21,6 +23,8 @@ impl __sdk::InModule for AuthStoreProjectionMeta { pub struct AuthStoreProjectionMetaCols { pub meta_id: __sdk::__query_builder::Col, pub updated_at: __sdk::__query_builder::Col, + pub phone_codes_json: __sdk::__query_builder::Col>, + pub wechat_states_json: __sdk::__query_builder::Col>, } impl __sdk::__query_builder::HasCols for AuthStoreProjectionMeta { @@ -29,6 +33,8 @@ impl __sdk::__query_builder::HasCols for AuthStoreProjectionMeta { AuthStoreProjectionMetaCols { meta_id: __sdk::__query_builder::Col::new(table_name, "meta_id"), updated_at: __sdk::__query_builder::Col::new(table_name, "updated_at"), + phone_codes_json: __sdk::__query_builder::Col::new(table_name, "phone_codes_json"), + wechat_states_json: __sdk::__query_builder::Col::new(table_name, "wechat_states_json"), } } } diff --git a/server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_phone_code_type.rs b/server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_phone_code_type.rs new file mode 100644 index 000000000..10c7bffc2 --- /dev/null +++ b/server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_phone_code_type.rs @@ -0,0 +1,21 @@ +// 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 AuthStoreProjectionPhoneCode { + pub phone_number: String, + pub scene: String, + pub verify_code_hash: String, + pub expires_at: String, + pub last_sent_at: String, + pub failed_attempts: u32, + pub provider_out_id: Option, +} + +impl __sdk::InModule for AuthStoreProjectionPhoneCode { + type Module = super::RemoteModule; +} diff --git a/server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_view_type.rs b/server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_view_type.rs index b771d4b3a..fa743157a 100644 --- a/server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_view_type.rs +++ b/server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_view_type.rs @@ -5,8 +5,10 @@ use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws}; use super::auth_store_projection_identity_type::AuthStoreProjectionIdentity; +use super::auth_store_projection_phone_code_type::AuthStoreProjectionPhoneCode; use super::auth_store_projection_refresh_session_type::AuthStoreProjectionRefreshSession; use super::auth_store_projection_user_type::AuthStoreProjectionUser; +use super::auth_store_projection_wechat_state_type::AuthStoreProjectionWechatState; #[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)] #[sats(crate = __lib)] @@ -15,6 +17,8 @@ pub struct AuthStoreProjectionView { pub users: Vec, pub identities: Vec, pub refresh_sessions: Vec, + pub phone_codes: Vec, + pub wechat_states: Vec, pub base_updated_at_micros: i64, } diff --git a/server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_wechat_state_type.rs b/server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_wechat_state_type.rs new file mode 100644 index 000000000..7de462fa8 --- /dev/null +++ b/server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_wechat_state_type.rs @@ -0,0 +1,24 @@ +// 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 AuthStoreProjectionWechatState { + pub wechat_state_id: String, + pub state_token: String, + pub redirect_path: String, + pub scene: String, + pub request_user_agent: Option, + pub bind_user_id: Option, + pub expires_at: String, + pub consumed_at: Option, + pub created_at: String, + pub updated_at: String, +} + +impl __sdk::InModule for AuthStoreProjectionWechatState { + type Module = super::RemoteModule; +} diff --git a/server-rs/crates/spacetime-module/src/auth/procedures.rs b/server-rs/crates/spacetime-module/src/auth/procedures.rs index 7bf0ded36..9f181c018 100644 --- a/server-rs/crates/spacetime-module/src/auth/procedures.rs +++ b/server-rs/crates/spacetime-module/src/auth/procedures.rs @@ -1,4 +1,5 @@ use crate::{ProcedureContext, ReducerContext, SpacetimeType, Table, Timestamp}; +use serde::{Deserialize, Serialize}; use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; @@ -15,6 +16,8 @@ pub struct AuthStoreProjectionView { pub users: Vec, pub identities: Vec, pub refresh_sessions: Vec, + pub phone_codes: Vec, + pub wechat_states: Vec, pub base_updated_at_micros: i64, } @@ -59,6 +62,31 @@ pub struct AuthStoreProjectionRefreshSession { pub last_seen_at: String, } +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, SpacetimeType)] +pub struct AuthStoreProjectionPhoneCode { + pub phone_number: String, + pub scene: String, + pub verify_code_hash: String, + pub expires_at: String, + pub last_sent_at: String, + pub failed_attempts: u32, + pub provider_out_id: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, SpacetimeType)] +pub struct AuthStoreProjectionWechatState { + pub wechat_state_id: String, + pub state_token: String, + pub redirect_path: String, + pub scene: String, + pub request_user_agent: Option, + pub bind_user_id: Option, + pub expires_at: String, + pub consumed_at: Option, + pub created_at: String, + pub updated_at: String, +} + #[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)] pub struct AuthStoreProjectionSyncRecord { pub imported_user_count: u32, @@ -214,6 +242,10 @@ fn sync_auth_store_projection_tx( current_updated_at_micros.unwrap_or_default(), )?; ensure_newer_auth_projection_version(current_updated_at_micros, input.updated_at_micros)?; + let phone_codes_json = serde_json::to_string(&input.phone_codes) + .map_err(|error| format!("序列化短信验证码投影失败：{error}"))?; + let wechat_states_json = serde_json::to_string(&input.wechat_states) + .map_err(|error| format!("序列化微信授权 state 投影失败：{error}"))?; let user_ids = input .users @@ -343,7 +375,12 @@ fn sync_auth_store_projection_tx( imported_refresh_session_count += 1; } - upsert_auth_projection_meta(ctx, input.updated_at_micros); + upsert_auth_projection_meta( + ctx, + input.updated_at_micros, + phone_codes_json, + wechat_states_json, + ); Ok(AuthStoreProjectionSyncRecord { imported_user_count, @@ -387,13 +424,31 @@ fn ensure_newer_auth_projection_version( fn export_auth_store_projection_from_tables_tx( ctx: &ReducerContext, ) -> Result { - let updated_at_micros = ctx + let meta = ctx .db .auth_store_projection_meta() .meta_id() - .find(&AUTH_STORE_PROJECTION_META_ID.to_string()) + .find(&AUTH_STORE_PROJECTION_META_ID.to_string()); + let updated_at_micros = meta + .as_ref() .map(|row| row.updated_at.to_micros_since_unix_epoch()) .unwrap_or(0); + let phone_codes = meta + .as_ref() + .and_then(|row| row.phone_codes_json.as_deref()) + .filter(|value| !value.trim().is_empty()) + .map(serde_json::from_str) + .transpose() + .map_err(|error| format!("解析短信验证码投影失败：{error}"))? + .unwrap_or_default(); + let wechat_states = meta + .as_ref() + .and_then(|row| row.wechat_states_json.as_deref()) + .filter(|value| !value.trim().is_empty()) + .map(serde_json::from_str) + .transpose() + .map_err(|error| format!("解析微信授权 state 投影失败：{error}"))? + .unwrap_or_default(); let users = ctx .db .user_account() @@ -451,6 +506,8 @@ fn export_auth_store_projection_from_tables_tx( users, identities, refresh_sessions, + phone_codes, + wechat_states, }) } @@ -493,7 +550,12 @@ fn delete_missing_refresh_sessions( } } -fn upsert_auth_projection_meta(ctx: &ReducerContext, updated_at_micros: i64) { +fn upsert_auth_projection_meta( + ctx: &ReducerContext, + updated_at_micros: i64, + phone_codes_json: String, + wechat_states_json: String, +) { let meta_id = AUTH_STORE_PROJECTION_META_ID.to_string(); if ctx .db @@ -512,6 +574,8 @@ fn upsert_auth_projection_meta(ctx: &ReducerContext, updated_at_micros: i64) { .insert(AuthStoreProjectionMeta { meta_id, updated_at: Timestamp::from_micros_since_unix_epoch(updated_at_micros), + phone_codes_json: Some(phone_codes_json), + wechat_states_json: Some(wechat_states_json), }); } diff --git a/server-rs/crates/spacetime-module/src/auth/tables.rs b/server-rs/crates/spacetime-module/src/auth/tables.rs index e9f6e6c11..49251e549 100644 --- a/server-rs/crates/spacetime-module/src/auth/tables.rs +++ b/server-rs/crates/spacetime-module/src/auth/tables.rs @@ -5,6 +5,10 @@ pub struct AuthStoreProjectionMeta { #[primary_key] pub(crate) meta_id: String, pub(crate) updated_at: Timestamp, + #[default(None::)] + pub(crate) phone_codes_json: Option, + #[default(None::)] + pub(crate) wechat_states_json: Option, } #[spacetimedb::table( 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 e553a58e4..4fd3e8818 100644 --- a/server-rs/crates/spacetime-module/src/runtime/active/profile.rs +++ b/server-rs/crates/spacetime-module/src/runtime/active/profile.rs @@ -9451,10 +9451,9 @@ fn process_profile_wallet_refund_outbox_tx( let mut rows = ctx .db .profile_wallet_refund_outbox() - .iter() - .filter(|row| { - row.status == PROFILE_WALLET_REFUND_OUTBOX_STATUS_PENDING && row.available_at <= now - }) + .by_profile_wallet_refund_outbox_status_available() + .filter(&PROFILE_WALLET_REFUND_OUTBOX_STATUS_PENDING.to_string()) + .filter(|row| row.available_at <= now) .collect::>(); rows.sort_by(|left, right| { left.available_at