From e0e3348f798f925b1579bbdd6186bca9e2396133 Mon Sep 17 00:00:00 2001 From: kdletters Date: Tue, 30 Jun 2026 20:58:33 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E8=AE=A4=E8=AF=81=E5=BF=AB?= =?UTF-8?q?=E7=85=A7=E5=9B=9E=E7=81=8C=E5=AF=BC=E8=87=B4=E9=87=8D=E5=A4=8D?= =?UTF-8?q?=E8=B4=A6=E5=8F=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 切断运行中从 SpacetimeDB 整包快照回灌内存的路径 清空旧 auth_store_snapshot 并仅保留一次性迁移 修复手机号字段只在 AuthUser 中时重复创建账号 同步认证排障与后端数据契约文档 --- .../shared-memory/decision-log.md | 7 + docs/project-memory/shared-memory/pitfalls.md | 12 +- ...】server-rs与SpacetimeDB数据契约-2026-05-15.md | 4 +- ...发运维】本地开发验证与生产运维-2026-05-15.md | 2 +- server-rs/crates/api-server/src/auth.rs | 83 +--- .../crates/api-server/src/auth_sessions.rs | 4 +- server-rs/crates/api-server/src/logout.rs | 4 +- server-rs/crates/api-server/src/logout_all.rs | 4 +- .../crates/api-server/src/password_entry.rs | 4 +- .../api-server/src/password_management.rs | 8 +- server-rs/crates/api-server/src/phone_auth.rs | 4 +- .../crates/api-server/src/profile_identity.rs | 2 +- .../crates/api-server/src/refresh_session.rs | 39 +- server-rs/crates/api-server/src/state.rs | 52 +- .../crates/api-server/src/wechat/auth.rs | 12 +- server-rs/crates/module-auth/src/lib.rs | 170 ++++++- .../spacetime-module/src/auth/procedures.rs | 446 +++++------------- 17 files changed, 355 insertions(+), 502 deletions(-) diff --git a/docs/project-memory/shared-memory/decision-log.md b/docs/project-memory/shared-memory/decision-log.md index ba65153e9..814d5f253 100644 --- a/docs/project-memory/shared-memory/decision-log.md +++ b/docs/project-memory/shared-memory/decision-log.md @@ -1891,6 +1891,13 @@ - 影响范围：`spacetime-module` auth procedures、`spacetime-client` auth facade、`api-server` 启动恢复、后端架构文档、开发运维文档、认证排障记忆。 - 验证方式：`cargo check -p spacetime-module --manifest-path server-rs/Cargo.toml`、`cargo check -p api-server --manifest-path server-rs/Cargo.toml`、`cargo test -p api-server spacetime_unavailable_router_returns_service_unavailable_for_requests --manifest-path server-rs/Cargo.toml -- --nocapture`、`npm run check:encoding`。 +## 2026-06-30 auth_store_snapshot 只做一次性迁移并切断运行中回灌 + +- 背景：同手机号重复账号暴露出认证工作集、正式认证表和旧 `auth_store_snapshot` 之间仍有互刷路径；运行中 Bearer / refresh session 未命中后再导出整包状态刷新内存，会把旧手机号索引或旧会话重新带回进程。 +- 决策：`auth_store_snapshot` 不再保留行级备查；正式认证表为空时才从最新旧快照转移一次到 `user_account` / `auth_identity` / `refresh_session`，随后清空旧表。`api-server` 运行中不再因 Bearer 用户、token version、session 或 refresh token 未命中而从 SpacetimeDB 导出整包状态刷新 `InMemoryAuthStore`；启动恢复暂保留从正式表构建工作集，直到认证仓储改为直接读写正式表。 +- 影响范围：`server-rs/crates/spacetime-module/src/auth/procedures.rs`、`server-rs/crates/api-server/src/state.rs`、`server-rs/crates/api-server/src/auth.rs`、`server-rs/crates/api-server/src/refresh_session.rs`、认证排障记忆与后端架构文档。 +- 验证方式：`cargo test -p spacetime-module auth_export -- --nocapture`、`cargo test -p module-auth phone_only_exists -- --nocapture`、`cargo test -p module-auth bind_wechat_phone_merges -- --nocapture`、`npm run check:encoding`、`git diff --check`。 + ## 2026-05-13 微信小程序支付以后端通知为唯一入账事实 - 背景：“我的”账户充值需要接入微信小程序支付，同时保留本地 / H5 mock 支付联调能力。 diff --git a/docs/project-memory/shared-memory/pitfalls.md b/docs/project-memory/shared-memory/pitfalls.md index 870e7bfeb..7d4aa7e20 100644 --- a/docs/project-memory/shared-memory/pitfalls.md +++ b/docs/project-memory/shared-memory/pitfalls.md @@ -86,6 +86,14 @@ - 验证：`http://127.0.0.1:3101/v1/ping` 可访问、`http://127.0.0.1:8082/healthz` 返回 200、`http://127.0.0.1:3000/` 和 `http://127.0.0.1:3102/admin/` 可打开。 - 关联：`scripts/dev.mjs`、`.app/dev-stack.json`、`docs/project-memory/shared-memory/development-workflow.md`。 +## 私有兑换码不适用先查同手机号重复账号 + +- 现象：后台把私有兑换码配给某个陶泥号或手机号后，用户用同一手机号登录兑换仍提示 `该兑换码不适用于当前账号`。 +- 原因：认证表里可能存在同一手机号的多条 `user_account`。如果认证工作集重建 `phone_to_user_id` 时让 `user_account.phone_number_e164` 后写覆盖前写，当前登录态会漂到没有 `auth_identity` 的重复账号，而兑换码白名单仍指向另一个内部 `user_id`。 +- 处理：重建认证工作集时以 `auth_identity(provider="phone")` 指向的账号作为手机号索引权威，`user_account.phone_number_e164` 只补没有 identity 的手机号；旧 `auth_store_snapshot` 只允许在正式认证表为空时一次性转移到正式表，随后清空，不再作为运行期回灌来源；Bearer / refresh session 本进程未命中时不要再从 SpacetimeDB 导出整包快照刷新内存。线上止血先核对失败请求附近的 current session `user_id` 与兑换码 `allowed_user_ids`，不要只看手机号展示值。 +- 验证：`cargo test -p spacetime-module auth_export -- --nocapture` 应覆盖同手机号重复账号时手机号索引优先指向有 phone identity 的账号；`api-server` 中不应再存在运行期 `refresh_auth_store_from_spacetime` 调用。 +- 关联：`server-rs/crates/spacetime-module/src/auth/procedures.rs`、`server-rs/crates/spacetime-module/src/auth/tables.rs`、`server-rs/crates/module-auth/src/lib.rs`。 + ## API Build / Deploy 归档清单不能漏掉随包 Pingora 脚本 - 现象：`Genarrative-Api-Deploy` 在发布阶段报 `发布产物缺少 Pingora TLS 证书同步脚本: build//scripts/deploy/pingora-tls-cert-sync.mjs`。 @@ -1191,8 +1199,8 @@ ## 忘记密码后仍提示手机号或密码错误先查认证快照同步 - 现象：用户通过“忘记密码”重设密码后，接口返回成功或页面进入登录态，但再次使用新密码登录仍提示“手机号或密码错误”；重启后还可能出现 `Bearer JWT 版本已失效`，日志里的 token version 与本地快照不一致。 -- 原因：重置/修改密码会更新 `password_hash`、`password_login_enabled` 和 `token_version`，如果 API 层只更新本地 `InMemoryAuthStore`，没有调用 `sync_auth_store_snapshot_to_spacetime()`，`api-server` 重启时可能从旧的 SpacetimeDB 表或旧快照恢复账号状态。 -- 处理：`POST /api/auth/password/change` 与 `POST /api/auth/password/reset` 成功后必须同步认证快照。2026-05-27 起，启动恢复只允许从 SpacetimeDB 正式认证表恢复；`auth_store_snapshot` 只保留行级记录，不再写 `default` 聚合单行，也不再把本地文件 `auth-store.json` / `GENARRATIVE_AUTH_STORE_PATH` 当作恢复源。认证创建、登录会话、刷新、退出、改密、重置密码、绑定和资料变更等写操作必须在返回客户端前成功同步 SpacetimeDB；同步失败时接口返回错误，不允许把只存在于当前进程内存的账号或会话当成成功结果。新用户注册奖励、邀请码绑定和登录埋点必须排在认证同步成功之后，避免认证没落库时先写出钱包或邀请关系。若启动时连不上 SpacetimeDB，`api-server` 等待启动恢复超时后进入依赖不可用模式，所有请求返回 `503 SERVICE_UNAVAILABLE`，`details.reason = "spacetime_startup_unavailable"`。 +- 原因：重置/修改密码会更新 `password_hash`、`password_login_enabled` 和 `token_version`，如果 API 层只更新本地 `InMemoryAuthStore`，没有调用 `sync_auth_store_tables_to_spacetime()`，`api-server` 重启时可能从旧的 SpacetimeDB 正式认证表恢复账号状态。 +- 处理：`POST /api/auth/password/change` 与 `POST /api/auth/password/reset` 成功后必须同步正式认证表。2026-06-30 起，`auth_store_snapshot` 不再保留行级备查，也不作为运行期回灌来源；只在正式认证表为空时把最新旧快照转移一次到 `user_account` / `auth_identity` / `refresh_session` 并立即清空旧表。认证创建、登录会话、刷新、退出、改密、重置密码、绑定和资料变更等写操作必须在返回客户端前成功同步 SpacetimeDB；同步失败时接口返回错误，不允许把只存在于当前进程内存的账号或会话当成成功结果。新用户注册奖励、邀请码绑定和登录埋点必须排在认证同步成功之后，避免认证没落库时先写出钱包或邀请关系。 - 验证：执行 `cargo test -p module-auth password --manifest-path server-rs/Cargo.toml` 与 `cargo test -p api-server password --manifest-path server-rs/Cargo.toml`；手测时重设密码后旧密码应失败，新密码应成功，重启后仍应保持。 - 关联：`server-rs/crates/api-server/src/password_management.rs`、`server-rs/crates/api-server/src/state.rs`、`docs/technical/PASSWORD_LOGIN_CHANGE_RESET_DESIGN_2026-04-24.md`。 diff --git a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md index 8e4c255be..8df6c4fdb 100644 --- a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md +++ b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md @@ -294,9 +294,9 @@ npm run check:server-rs-ddd - Rust 结构体：`AuthStoreSnapshot` - 源码：`server-rs/crates/spacetime-module/src/auth/tables.rs` -认证恢复策略：`api-server` 启动时只从 SpacetimeDB 正式认证表（`user_account` / `auth_identity` / `refresh_session`）投影恢复进程内认证工作集；运行中若 Bearer `sid` 或 refresh cookie 在本进程工作集内未命中，会先从 SpacetimeDB 正式认证表按需刷新一次认证工作集再复查，避免多实例或滚动重启时新登录设备只被签发它的进程认识。`auth_store_snapshot` 只保留行级快照备查，不再作为启动兜底来源。`module-auth` 只保留内存工作集和 JSON 导入 / 导出能力，不再写本地持久化文件；`auth-store.json` / `GENARRATIVE_AUTH_STORE_PATH` 不再是兼容恢复源，也不得在启动时回写覆盖 `auth_identity` / `user_account`。认证创建、登录会话、刷新、退出、改密、重置密码、绑定和资料变更等写操作必须在返回客户端前成功同步 SpacetimeDB 正式认证表；同步失败时接口返回错误，不允许把只存在于当前进程内存的账号或会话当成成功结果。新用户注册奖励、邀请码绑定和登录埋点必须排在认证同步成功之后，避免认证没落库时先写出钱包或邀请关系。若启动恢复阶段 SpacetimeDB 不可连接或超时，`api-server` 会按固定间隔持续重试认证工作集恢复，恢复成功后才开始监听 HTTP，避免一次短超时让进程永久停留在依赖不可用状态。 +认证恢复策略：`api-server` 启动时只从 SpacetimeDB 正式认证表（`user_account` / `auth_identity` / `refresh_session`）投影恢复进程内认证工作集；运行中 Bearer `sid` 或 refresh cookie 在本进程工作集内未命中时直接按失效处理，不再从 SpacetimeDB 导出整包认证状态刷新内存，避免旧投影把重复手机号或旧会话重新灌回进程。`module-auth` 只保留内存工作集和 JSON 导入 / 导出能力，不再写本地持久化文件；`auth-store.json` / `GENARRATIVE_AUTH_STORE_PATH` 不再是兼容恢复源，也不得在启动时回写覆盖 `auth_identity` / `user_account`。认证创建、登录会话、刷新、退出、改密、重置密码、绑定和资料变更等写操作必须在返回客户端前成功同步 SpacetimeDB 正式认证表；同步失败时接口返回错误，不允许把只存在于当前进程内存的账号或会话当成成功结果。新用户注册奖励、邀请码绑定和登录埋点必须排在认证同步成功之后，避免认证没落库时先写出钱包或邀请关系。若启动恢复阶段 SpacetimeDB 不可连接或超时，`api-server` 会按固定间隔持续重试认证工作集恢复，恢复成功后才开始监听 HTTP，避免一次短超时让进程永久停留在依赖不可用状态。 -`auth_store_snapshot` 禁止再写单行 `snapshot_id = "default"` 聚合 JSON。认证同步入口收到 `module-auth` 整份快照后必须拆成行级记录写入同一张表，当前行键前缀包括：`meta/next_user_id`、`user/`、`phone/`、`session/`、`session_hash/`、`wechat/`、`union/`。SpacetimeDB 模块只保留 `import_auth_store_snapshot_json` 与 `export_auth_store_snapshot_from_tables` 两个认证快照过程；旧 `get_auth_store_snapshot`、`upsert_auth_store_snapshot`、`import_auth_store_snapshot` 兼容入口已删除。导入正式表时只按主键 upsert 本次快照包含的用户、身份和会话，避免过期快照把其他用户整表删除。 +`auth_store_snapshot` 禁止再写单行 `snapshot_id = "default"` 聚合 JSON，也不再保留行级备查。SpacetimeDB 模块只保留 `import_auth_store_snapshot_json` 与 `export_auth_store_snapshot_from_tables` 两个兼容过程：前者把当前 `module-auth` 工作集导入正式认证表并清空旧快照表；后者只在正式认证表为空时把最新旧快照转移一次到正式认证表，然后清空 `auth_store_snapshot`。旧 `get_auth_store_snapshot`、`upsert_auth_store_snapshot`、`import_auth_store_snapshot` 兼容入口已删除。导入正式表时只按主键 upsert 本次快照包含的用户、身份和会话，避免过期快照把其他用户整表删除。 导出认证快照时，`auth_identity` 与 `refresh_session` 只能引用仍存在于 `user_account` 的用户；孤儿手机号 identity、微信 identity、union 索引或 refresh session 必须被过滤，不能恢复成 `module-auth` 内存态里的 `phone_to_user_id` 死索引。`module-auth` 从 JSON 快照恢复时也要二次清理这些孤儿索引，避免历史坏快照导致密码登录提示错误、短信登录又提示手机号已存在。 diff --git a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md index b27e3abbb..aaa645b70 100644 --- a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md +++ b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md @@ -550,7 +550,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` 只保留行级记录，不再保存为单行 `default` 聚合快照，且旧 `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`）恢复；不要再配置或依赖 `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，以避免用空本地状态或旧快照覆盖认证表。 前端登录态恢复只把 `/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/auth.rs b/server-rs/crates/api-server/src/auth.rs index 93380d4d0..be0345c59 100644 --- a/server-rs/crates/api-server/src/auth.rs +++ b/server-rs/crates/api-server/src/auth.rs @@ -231,7 +231,7 @@ async fn authenticate_request( ); AppError::from_status(StatusCode::UNAUTHORIZED) })?; - let mut current_user = state + let current_user = state .auth_user_service() .get_user_by_id(claims.user_id()) .map_err(|error| { @@ -242,27 +242,7 @@ async fn authenticate_request( ); AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) })?; - if current_user.is_none() { - warn!( - %request_id, - user_id = %claims.user_id(), - "Bearer JWT 对应用户不存在，准备刷新认证工作集后复查" - ); - if refresh_auth_store_for_stale_bearer(state, &request_id, claims.user_id()).await { - current_user = state - .auth_user_service() - .get_user_by_id(claims.user_id()) - .map_err(|error| { - warn!( - %request_id, - error = %error, - "Bearer JWT 用户快照刷新后读取失败" - ); - AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) - })?; - } - } - let Some(mut current_user) = current_user else { + let Some(current_user) = current_user else { warn!( %request_id, user_id = %claims.user_id(), @@ -270,23 +250,6 @@ async fn authenticate_request( ); return Err(AppError::from_status(StatusCode::UNAUTHORIZED)); }; - if current_user.token_version != claims.token_version() { - if refresh_auth_store_for_stale_bearer(state, &request_id, claims.user_id()).await - && let Some(refreshed_user) = state - .auth_user_service() - .get_user_by_id(claims.user_id()) - .map_err(|error| { - warn!( - %request_id, - error = %error, - "Bearer JWT 用户版本刷新后读取失败" - ); - AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) - })? - { - current_user = refreshed_user; - } - } if current_user.token_version != claims.token_version() { warn!( %request_id, @@ -299,7 +262,7 @@ async fn authenticate_request( .with_message("当前登录态已失效，请重新登录")); } - let mut session_is_active = state + let session_is_active = state .refresh_session_service() .is_session_active_for_user( claims.user_id(), @@ -316,27 +279,6 @@ async fn authenticate_request( ); AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) })?; - if !session_is_active - && refresh_auth_store_for_stale_bearer(state, &request_id, claims.user_id()).await - { - session_is_active = state - .refresh_session_service() - .is_session_active_for_user( - claims.user_id(), - claims.session_id(), - OffsetDateTime::now_utc(), - ) - .map_err(|error| { - warn!( - %request_id, - user_id = %claims.user_id(), - session_id = %claims.session_id(), - error = %error, - "Bearer JWT refresh session 刷新后状态读取失败" - ); - AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) - })?; - } if !session_is_active { warn!( %request_id, @@ -359,25 +301,6 @@ fn request_id_from_request(request: &Request) -> String { .unwrap_or_else(|| "unknown".to_string()) } -async fn refresh_auth_store_for_stale_bearer( - state: &AppState, - request_id: &str, - user_id: &str, -) -> bool { - match state.refresh_auth_store_from_spacetime().await { - Ok(refreshed) => refreshed, - Err(error) => { - warn!( - %request_id, - user_id = %user_id, - error = %error, - "刷新认证工作集失败，继续按本进程现有状态处理" - ); - false - } - } -} - pub async fn inspect_auth_claims( Extension(request_context): Extension, Extension(authenticated): Extension, diff --git a/server-rs/crates/api-server/src/auth_sessions.rs b/server-rs/crates/api-server/src/auth_sessions.rs index 1ddb32ee0..c9ba471d3 100644 --- a/server-rs/crates/api-server/src/auth_sessions.rs +++ b/server-rs/crates/api-server/src/auth_sessions.rs @@ -94,11 +94,11 @@ pub async fn revoke_auth_session( ); } state - .sync_auth_store_snapshot_to_spacetime() + .sync_auth_store_tables_to_spacetime() .await .map_err(|error| { AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) - .with_message(format!("同步认证快照失败：{error}")) + .with_message(format!("同步认证状态失败：{error}")) })?; Ok(json_success_body( diff --git a/server-rs/crates/api-server/src/logout.rs b/server-rs/crates/api-server/src/logout.rs index 2e4a4f055..4de4f4efb 100644 --- a/server-rs/crates/api-server/src/logout.rs +++ b/server-rs/crates/api-server/src/logout.rs @@ -46,11 +46,11 @@ pub async fn logout( ) .map_err(map_logout_error)?; state - .sync_auth_store_snapshot_to_spacetime() + .sync_auth_store_tables_to_spacetime() .await .map_err(|error| { AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) - .with_message(format!("同步认证快照失败：{error}")) + .with_message(format!("同步认证状态失败：{error}")) })?; let mut headers = HeaderMap::new(); diff --git a/server-rs/crates/api-server/src/logout_all.rs b/server-rs/crates/api-server/src/logout_all.rs index e3f1711e9..e56e1b775 100644 --- a/server-rs/crates/api-server/src/logout_all.rs +++ b/server-rs/crates/api-server/src/logout_all.rs @@ -33,11 +33,11 @@ pub async fn logout_all( ) .map_err(map_logout_error)?; state - .sync_auth_store_snapshot_to_spacetime() + .sync_auth_store_tables_to_spacetime() .await .map_err(|error| { AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) - .with_message(format!("同步认证快照失败：{error}")) + .with_message(format!("同步认证状态失败：{error}")) })?; let mut headers = HeaderMap::new(); diff --git a/server-rs/crates/api-server/src/password_entry.rs b/server-rs/crates/api-server/src/password_entry.rs index 94c3781aa..eaa05fa86 100644 --- a/server-rs/crates/api-server/src/password_entry.rs +++ b/server-rs/crates/api-server/src/password_entry.rs @@ -43,11 +43,11 @@ pub async fn password_entry( let session_client = resolve_session_client_context(&headers); let signed_session = create_password_auth_session(&state, &result.user, &session_client)?; state - .sync_auth_store_snapshot_to_spacetime() + .sync_auth_store_tables_to_spacetime() .await .map_err(|error| { AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) - .with_message(format!("同步认证快照失败：{error}")) + .with_message(format!("同步认证状态失败：{error}")) })?; if result.created { crate::registration_reward::grant_new_user_registration_wallet_reward( diff --git a/server-rs/crates/api-server/src/password_management.rs b/server-rs/crates/api-server/src/password_management.rs index 94e5177e6..596140a2a 100644 --- a/server-rs/crates/api-server/src/password_management.rs +++ b/server-rs/crates/api-server/src/password_management.rs @@ -45,11 +45,11 @@ pub async fn change_password( .await .map_err(map_password_management_error)?; state - .sync_auth_store_snapshot_to_spacetime() + .sync_auth_store_tables_to_spacetime() .await .map_err(|error| { AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) - .with_message(format!("同步认证快照失败：{error}")) + .with_message(format!("同步认证状态失败：{error}")) })?; let mut headers = HeaderMap::new(); @@ -101,11 +101,11 @@ pub async fn reset_password( module_auth::AuthLoginMethod::Password, )?; state - .sync_auth_store_snapshot_to_spacetime() + .sync_auth_store_tables_to_spacetime() .await .map_err(|error| { AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) - .with_message(format!("同步认证快照失败：{error}")) + .with_message(format!("同步认证状态失败：{error}")) })?; record_daily_login_tracking_event_after_auth_success( &state, diff --git a/server-rs/crates/api-server/src/phone_auth.rs b/server-rs/crates/api-server/src/phone_auth.rs index 1b8788caf..55b2cce3e 100644 --- a/server-rs/crates/api-server/src/phone_auth.rs +++ b/server-rs/crates/api-server/src/phone_auth.rs @@ -159,11 +159,11 @@ pub async fn phone_login( AuthLoginMethod::Phone, )?; state - .sync_auth_store_snapshot_to_spacetime() + .sync_auth_store_tables_to_spacetime() .await .map_err(|error| { AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) - .with_message(format!("同步认证快照失败：{error}")) + .with_message(format!("同步认证状态失败：{error}")) })?; if created { crate::registration_reward::grant_new_user_registration_wallet_reward( diff --git a/server-rs/crates/api-server/src/profile_identity.rs b/server-rs/crates/api-server/src/profile_identity.rs index fc5365f21..4a00cd1a7 100644 --- a/server-rs/crates/api-server/src/profile_identity.rs +++ b/server-rs/crates/api-server/src/profile_identity.rs @@ -37,7 +37,7 @@ pub async fn update_profile_identity( .map_err(map_profile_update_error)?; state - .sync_auth_store_snapshot_to_spacetime() + .sync_auth_store_tables_to_spacetime() .await .map_err(|error| { AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR).with_message(error.to_string()) diff --git a/server-rs/crates/api-server/src/refresh_session.rs b/server-rs/crates/api-server/src/refresh_session.rs index 331f2ea70..4cdd87bb8 100644 --- a/server-rs/crates/api-server/src/refresh_session.rs +++ b/server-rs/crates/api-server/src/refresh_session.rs @@ -7,7 +7,6 @@ use module_auth::{RefreshSessionError, RotateRefreshSessionInput}; use platform_auth::hash_refresh_session_token; use shared_contracts::auth::RefreshSessionResponse; use time::OffsetDateTime; -use tracing::warn; use crate::{ api_response::json_success_body, @@ -49,36 +48,10 @@ pub async fn refresh_session( ) { Ok(rotated) => rotated, Err(RefreshSessionError::SessionNotFound) => { - match state.refresh_auth_store_from_spacetime().await { - Ok(true) => {} - Ok(false) => { - return Err(map_refresh_error_with_clear_cookie( - &state, - RefreshSessionError::SessionNotFound, - )); - } - Err(error) => { - warn!( - request_id = request_context.request_id(), - error = %error, - "refresh session 本地未命中后刷新认证工作集失败" - ); - return Err(map_refresh_error_with_clear_cookie( - &state, - RefreshSessionError::SessionNotFound, - )); - } - } - state - .refresh_session_service() - .rotate_session( - RotateRefreshSessionInput { - refresh_token_hash, - next_refresh_token_hash, - }, - OffsetDateTime::now_utc(), - ) - .map_err(|error| map_refresh_error_with_clear_cookie(&state, error))? + return Err(map_refresh_error_with_clear_cookie( + &state, + RefreshSessionError::SessionNotFound, + )); } Err(error) => return Err(map_refresh_error_with_clear_cookie(&state, error)), }; @@ -90,11 +63,11 @@ pub async fn refresh_session( Some(&rotated.session.client_info), )?; state - .sync_auth_store_snapshot_to_spacetime() + .sync_auth_store_tables_to_spacetime() .await .map_err(|error| { AppError::from_status(axum::http::StatusCode::INTERNAL_SERVER_ERROR) - .with_message(format!("同步认证快照失败：{error}")) + .with_message(format!("同步认证状态失败：{error}")) })?; record_daily_login_tracking_event_after_auth_success( &state, diff --git a/server-rs/crates/api-server/src/state.rs b/server-rs/crates/api-server/src/state.rs index 1752ceceb..4dff008f7 100644 --- a/server-rs/crates/api-server/src/state.rs +++ b/server-rs/crates/api-server/src/state.rs @@ -11,7 +11,6 @@ use std::{ use axum::extract::FromRef; use module_ai::{AiTaskService, InMemoryAiTaskStore}; #[cfg(not(test))] -use module_auth::RefreshAuthStoreSnapshotResult; use module_auth::{ AuthUserService, InMemoryAuthStore, PasswordEntryService, PhoneAuthService, RefreshSessionService, WechatAuthService, WechatAuthStateService, @@ -779,7 +778,7 @@ impl AppState { &self.password_entry_service } - pub async fn sync_auth_store_snapshot_to_spacetime(&self) -> Result<(), SpacetimeClientError> { + pub async fn sync_auth_store_tables_to_spacetime(&self) -> Result<(), SpacetimeClientError> { #[cfg(test)] return Ok(()); @@ -792,9 +791,9 @@ impl AppState { let updated_at_micros = i64::try_from( OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000, ) - .map_err(|_| SpacetimeClientError::Runtime("认证快照更新时间超出 i64 范围".to_string()))?; - // 当前进程内 auth_store 是认证请求的即时工作集；SpacetimeDB 正式认证表用于跨进程恢复。 - // 认证变更必须在返回客户端前写入 SpacetimeDB，避免只在本进程内成功、重启后丢失账号或会话。 + .map_err(|_| SpacetimeClientError::Runtime("认证状态更新时间超出 i64 范围".to_string()))?; + // 当前仍由 module-auth 的进程内工作集执行业务规则；这里只同步到 SpacetimeDB 正式认证表， + // 不再读写 auth_store_snapshot 行镜像。 #[cfg(not(test))] if let Err(error) = self .spacetime_client @@ -803,7 +802,7 @@ impl AppState { { warn!( error = %error, - "认证快照导入 SpacetimeDB 正式表失败，当前认证流程中止" + "认证状态导入 SpacetimeDB 正式表失败，当前认证流程中止" ); return Err(error); } @@ -811,47 +810,6 @@ impl AppState { Ok(()) } - #[cfg(not(test))] - pub fn refresh_auth_store_from_snapshot_json( - &self, - snapshot_json: &str, - ) -> Result { - self.auth_store - .refresh_from_snapshot_json(snapshot_json) - .map_err(SpacetimeClientError::Runtime) - } - - pub async fn refresh_auth_store_from_spacetime(&self) -> Result { - #[cfg(test)] - { - return Ok(false); - } - - #[cfg(not(test))] - { - let snapshot = self - .spacetime_client - .export_auth_store_snapshot_from_tables() - .await?; - let Some(snapshot_json) = snapshot - .snapshot_json - .as_deref() - .map(str::trim) - .filter(|value| !value.is_empty()) - else { - return Ok(false); - }; - let result = self.refresh_auth_store_from_snapshot_json(snapshot_json)?; - info!( - user_count = result.user_count, - session_count = result.session_count, - updated_at_micros = snapshot.updated_at_micros, - "已按需刷新本进程认证工作集" - ); - Ok(true) - } - } - pub async fn try_restore_auth_store_from_spacetime( config: AppConfig, ) -> Result { diff --git a/server-rs/crates/api-server/src/wechat/auth.rs b/server-rs/crates/api-server/src/wechat/auth.rs index 28446f334..c5d8437d3 100644 --- a/server-rs/crates/api-server/src/wechat/auth.rs +++ b/server-rs/crates/api-server/src/wechat/auth.rs @@ -147,11 +147,11 @@ pub async fn handle_wechat_callback( AuthLoginMethod::Wechat, )?; state - .sync_auth_store_snapshot_to_spacetime() + .sync_auth_store_tables_to_spacetime() .await .map_err(|error| { AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) - .with_message(format!("同步认证快照失败：{error}")) + .with_message(format!("同步认证状态失败：{error}")) })?; record_daily_login_tracking_event_after_auth_success( &state, @@ -252,11 +252,11 @@ pub async fn bind_wechat_phone( AuthLoginMethod::Wechat, )?; state - .sync_auth_store_snapshot_to_spacetime() + .sync_auth_store_tables_to_spacetime() .await .map_err(|error| { AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) - .with_message(format!("同步认证快照失败：{error}")) + .with_message(format!("同步认证状态失败：{error}")) })?; if result.activated_new_user { crate::registration_reward::grant_new_user_registration_wallet_reward( @@ -328,11 +328,11 @@ pub async fn login_wechat_mini_program( AuthLoginMethod::Wechat, )?; state - .sync_auth_store_snapshot_to_spacetime() + .sync_auth_store_tables_to_spacetime() .await .map_err(|error| { AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR) - .with_message(format!("同步认证快照失败：{error}")) + .with_message(format!("同步认证状态失败：{error}")) })?; let mut response_headers = HeaderMap::new(); diff --git a/server-rs/crates/module-auth/src/lib.rs b/server-rs/crates/module-auth/src/lib.rs index b3f4db238..b34f4d188 100644 --- a/server-rs/crates/module-auth/src/lib.rs +++ b/server-rs/crates/module-auth/src/lib.rs @@ -120,6 +120,13 @@ fn hydrate_private_auth_fields( hydrated } +fn stored_user_phone_number(stored_user: &StoredPasswordUser) -> Option<&str> { + stored_user + .phone_number + .as_deref() + .or(stored_user.user.phone_number.as_deref()) +} + #[derive(Clone, Debug)] pub struct PasswordEntryService { store: InMemoryAuthStore, @@ -1100,7 +1107,7 @@ impl InMemoryAuthStore { let Some(stored_user) = state .users_by_username .values() - .find(|stored_user| stored_user.phone_number.as_deref() == Some(phone_number)) + .find(|stored_user| stored_user_phone_number(stored_user) == Some(phone_number)) .cloned() else { return None; @@ -2837,6 +2844,70 @@ mod tests { ); } + #[tokio::test] + async fn phone_login_reuses_user_when_phone_only_exists_on_auth_user() { + let stored_user = StoredPasswordUser { + user: AuthUser { + id: "user_existing_phone".to_string(), + public_user_code: "SY-00000009".to_string(), + username: "phone_existing".to_string(), + display_name: "138****8009".to_string(), + avatar_url: None, + phone_number: Some("+8613800138009".to_string()), + phone_number_masked: Some("138****8009".to_string()), + login_method: AuthLoginMethod::Phone, + binding_status: AuthBindingStatus::Active, + wechat_bound: false, + wechat_display_name: None, + wechat_account: None, + token_version: 1, + created_at: "2026-06-30T00:00:00Z".to_string(), + }, + password_hash: "hash-existing".to_string(), + password_login_enabled: false, + phone_number: None, + }; + let snapshot = PersistentAuthStoreSnapshot { + next_user_id: 10, + users_by_username: HashMap::from([("phone_existing".to_string(), stored_user)]), + phone_to_user_id: HashMap::new(), + sessions_by_id: HashMap::new(), + session_id_by_refresh_token_hash: HashMap::new(), + wechat_identity_by_provider_uid: HashMap::new(), + user_id_by_provider_union_id: HashMap::new(), + }; + let snapshot_json = + serde_json::to_string(&snapshot).expect("snapshot json should serialize"); + let phone_service = build_phone_service( + InMemoryAuthStore::from_snapshot_json(&snapshot_json).expect("snapshot should restore"), + ); + let now = OffsetDateTime::now_utc(); + phone_service + .send_code( + SendPhoneCodeInput { + phone_number: "13800138009".to_string(), + scene: PhoneAuthScene::Login, + }, + now, + ) + .await + .expect("phone code should send"); + + let reused = phone_service + .login( + PhoneLoginInput { + phone_number: "13800138009".to_string(), + verify_code: "123456".to_string(), + }, + now + Duration::seconds(1), + ) + .await + .expect("phone login should reuse existing user"); + + assert!(!reused.created); + assert_eq!(reused.user.id, "user_existing_phone"); + } + #[tokio::test] async fn snapshot_json_restores_user_and_refresh_session_after_roundtrip() { let store = InMemoryAuthStore::default(); @@ -3773,6 +3844,103 @@ mod tests { ); } + #[tokio::test] + async fn bind_wechat_phone_merges_when_existing_phone_only_exists_on_auth_user() { + let stored_user = StoredPasswordUser { + user: AuthUser { + id: "user_existing_phone_bind".to_string(), + public_user_code: "SY-00000032".to_string(), + username: "phone_existing_bind".to_string(), + display_name: "138****8032".to_string(), + avatar_url: None, + phone_number: Some("+8613800138032".to_string()), + phone_number_masked: Some("138****8032".to_string()), + login_method: AuthLoginMethod::Phone, + binding_status: AuthBindingStatus::Active, + wechat_bound: false, + wechat_display_name: None, + wechat_account: None, + token_version: 1, + created_at: "2026-06-30T00:00:00Z".to_string(), + }, + password_hash: "hash-existing".to_string(), + password_login_enabled: false, + phone_number: None, + }; + let snapshot = PersistentAuthStoreSnapshot { + next_user_id: 33, + users_by_username: HashMap::from([("phone_existing_bind".to_string(), stored_user)]), + phone_to_user_id: HashMap::new(), + sessions_by_id: HashMap::new(), + session_id_by_refresh_token_hash: HashMap::new(), + wechat_identity_by_provider_uid: HashMap::new(), + user_id_by_provider_union_id: HashMap::new(), + }; + let snapshot_json = + serde_json::to_string(&snapshot).expect("snapshot json should serialize"); + let store = + InMemoryAuthStore::from_snapshot_json(&snapshot_json).expect("snapshot should restore"); + let phone_service = build_phone_service(store.clone()); + let wechat_service = WechatAuthService::new(store); + let now = OffsetDateTime::now_utc(); + let wechat_user = wechat_service + .resolve_login(ResolveWechatLoginInput { + profile: WechatIdentityProfile { + provider_uid: "wx-openid-existing-phone-bind".to_string(), + provider_union_id: Some("wx-union-existing-phone-bind".to_string()), + display_name: Some("待绑定微信用户".to_string()), + avatar_url: None, + session_key: None, + }, + }) + .await + .expect("wechat login should succeed") + .user; + + phone_service + .send_code( + SendPhoneCodeInput { + phone_number: "13800138032".to_string(), + scene: PhoneAuthScene::BindPhone, + }, + now, + ) + .await + .expect("bind phone code should send"); + let merged = phone_service + .bind_wechat_phone( + BindWechatPhoneInput { + user_id: wechat_user.id, + phone_number: "13800138032".to_string(), + verify_code: "123456".to_string(), + wechat_display_name: None, + }, + now + Duration::seconds(1), + ) + .await + .expect("bind phone should merge existing phone user"); + + assert_eq!(merged.user.id, "user_existing_phone_bind"); + assert!(merged.user.wechat_bound); + assert_eq!( + wechat_service + .resolve_login(ResolveWechatLoginInput { + profile: WechatIdentityProfile { + provider_uid: "wx-openid-existing-phone-bind".to_string(), + provider_union_id: Some("wx-union-existing-phone-bind".to_string()), + display_name: Some("已归并微信用户".to_string()), + avatar_url: None, + session_key: None, + }, + }) + .await + .expect("wechat login should reuse merged user") + .user + .id, + "user_existing_phone_bind" + ); + } + #[tokio::test] async fn bind_wechat_phone_keeps_account_marker_when_identity_has_no_display_name() { let store = build_store(); diff --git a/server-rs/crates/spacetime-module/src/auth/procedures.rs b/server-rs/crates/spacetime-module/src/auth/procedures.rs index 5beef1efb..a04181d83 100644 --- a/server-rs/crates/spacetime-module/src/auth/procedures.rs +++ b/server-rs/crates/spacetime-module/src/auth/procedures.rs @@ -1,28 +1,18 @@ -use serde::Serialize; - use crate::{ProcedureContext, ReducerContext, SpacetimeType, Table, Timestamp}; use super::{ mapper::{ - sanitize_identity_component, AuthUserSnapshot, PersistentAuthStoreSnapshot, - RefreshSessionSnapshot, StoredPasswordUserSnapshot, StoredRefreshSessionSnapshot, - StoredWechatIdentitySnapshot, + AuthUserSnapshot, PersistentAuthStoreSnapshot, RefreshSessionSnapshot, + StoredPasswordUserSnapshot, StoredRefreshSessionSnapshot, StoredWechatIdentitySnapshot, + sanitize_identity_component, }, tables::{ - auth_identity, auth_store_projection_meta, auth_store_snapshot, refresh_session, - user_account, AuthIdentity, AuthStoreProjectionMeta, AuthStoreSnapshot, RefreshSession, - UserAccount, + AuthIdentity, AuthStoreProjectionMeta, RefreshSession, UserAccount, auth_identity, + auth_store_projection_meta, auth_store_snapshot, refresh_session, user_account, }, }; const AUTH_STORE_PROJECTION_META_ID: &str = "default"; -const AUTH_STORE_SNAPSHOT_META_NEXT_USER_ID: &str = "meta/next_user_id"; -const AUTH_STORE_SNAPSHOT_USER_PREFIX: &str = "user/"; -const AUTH_STORE_SNAPSHOT_PHONE_PREFIX: &str = "phone/"; -const AUTH_STORE_SNAPSHOT_SESSION_PREFIX: &str = "session/"; -const AUTH_STORE_SNAPSHOT_SESSION_HASH_PREFIX: &str = "session_hash/"; -const AUTH_STORE_SNAPSHOT_WECHAT_PREFIX: &str = "wechat/"; -const AUTH_STORE_SNAPSHOT_UNION_PREFIX: &str = "union/"; #[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)] pub struct AuthStoreSnapshotRecord { @@ -49,80 +39,10 @@ fn normalize_user_account_tags( module_runtime::normalize_profile_user_tags(tags.unwrap_or_default()) } -fn prefixed_snapshot_id(prefix: &str, value: &str) -> String { - format!("{prefix}{}", sanitize_identity_component(value)) -} - -fn upsert_auth_snapshot_row( - ctx: &ReducerContext, - snapshot_id: String, - snapshot_json: String, - updated_at: Timestamp, -) { - if ctx - .db - .auth_store_snapshot() - .snapshot_id() - .find(&snapshot_id) - .is_some() - { - ctx.db - .auth_store_snapshot() - .snapshot_id() - .delete(&snapshot_id); - } - - ctx.db.auth_store_snapshot().insert(AuthStoreSnapshot { - snapshot_id, - snapshot_json, - updated_at, - }); -} - -fn auth_store_snapshot_user_row_id(user_id: &str) -> String { - prefixed_snapshot_id(AUTH_STORE_SNAPSHOT_USER_PREFIX, user_id) -} - -fn auth_store_snapshot_phone_row_id(phone_number: &str, user_id: &str) -> String { - prefixed_snapshot_id( - AUTH_STORE_SNAPSHOT_PHONE_PREFIX, - &format!("{phone_number}|{user_id}"), - ) -} - -fn auth_store_snapshot_session_row_id(session_id: &str) -> String { - prefixed_snapshot_id(AUTH_STORE_SNAPSHOT_SESSION_PREFIX, session_id) -} - -fn auth_store_snapshot_session_hash_row_id(refresh_token_hash: &str, session_id: &str) -> String { - prefixed_snapshot_id( - AUTH_STORE_SNAPSHOT_SESSION_HASH_PREFIX, - &format!("{refresh_token_hash}|{session_id}"), - ) -} - -fn auth_store_snapshot_wechat_row_id(provider_uid: &str, user_id: &str) -> String { - prefixed_snapshot_id( - AUTH_STORE_SNAPSHOT_WECHAT_PREFIX, - &format!("{provider_uid}|{user_id}"), - ) -} - -fn auth_store_snapshot_union_row_id(union_id: &str, user_id: &str) -> String { - prefixed_snapshot_id( - AUTH_STORE_SNAPSHOT_UNION_PREFIX, - &format!("{union_id}|{user_id}"), - ) -} - fn snapshot_has_user_rows(snapshot: &PersistentAuthStoreSnapshot) -> bool { !snapshot.users_by_username.is_empty() } -fn to_snapshot_row_json(label: &str, value: &T) -> Result { - serde_json::to_string(value).map_err(|error| format!("{label} 序列化失败：{error}")) -} - #[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)] pub struct AuthStoreSnapshotImportRecord { pub imported_user_count: u32, @@ -137,7 +57,7 @@ pub struct AuthStoreSnapshotImportProcedureResult { pub error_message: Option, } -// Axum 运行期认证变更直接导入正式认证表，并把快照拆成行级记录；禁止再写 `auth_store_snapshot/default`。 +// Axum 运行期认证变更直接导入正式认证表；旧 `auth_store_snapshot` 只做一次性清理。 #[spacetimedb::procedure] pub fn import_auth_store_snapshot_json( ctx: &mut ProcedureContext, @@ -194,11 +114,19 @@ fn import_auth_store_snapshot_json_value_tx( } let parsed = serde_json::from_str::(snapshot_json) .map_err(|error| format!("认证快照 JSON 解析失败：{error}"))?; + import_auth_store_snapshot_value_tx(ctx, parsed, updated_at_micros) +} + +fn import_auth_store_snapshot_value_tx( + ctx: &ReducerContext, + parsed: PersistentAuthStoreSnapshot, + updated_at_micros: i64, +) -> Result { if !snapshot_has_user_rows(&parsed) { return Err("认证快照缺少用户记录，拒绝导入正式表".to_string()); } - upsert_auth_store_snapshot_rows(ctx, &parsed, updated_at_micros)?; + clear_legacy_auth_store_snapshot_rows(ctx); upsert_auth_projection_meta(ctx, updated_at_micros); let mut imported_user_count = 0_u32; @@ -322,126 +250,28 @@ fn import_auth_store_snapshot_json_value_tx( }) } -fn upsert_auth_store_snapshot_rows( - ctx: &ReducerContext, - snapshot: &PersistentAuthStoreSnapshot, - updated_at_micros: i64, -) -> Result<(), String> { - let updated_at = Timestamp::from_micros_since_unix_epoch(updated_at_micros); - let desired_ids = auth_store_snapshot_row_ids(snapshot); +fn clear_legacy_auth_store_snapshot_rows(ctx: &ReducerContext) { for row in ctx.db.auth_store_snapshot().iter().collect::>() { - if !desired_ids.contains(&row.snapshot_id) { - ctx.db - .auth_store_snapshot() - .snapshot_id() - .delete(&row.snapshot_id); - } + ctx.db + .auth_store_snapshot() + .snapshot_id() + .delete(&row.snapshot_id); } - - upsert_auth_snapshot_row( - ctx, - AUTH_STORE_SNAPSHOT_META_NEXT_USER_ID.to_string(), - to_snapshot_row_json("认证快照 next_user_id", &snapshot.next_user_id)?, - updated_at, - ); - - for user in snapshot.users_by_username.values() { - upsert_auth_snapshot_row( - ctx, - auth_store_snapshot_user_row_id(&user.user.id), - to_snapshot_row_json("认证快照用户", user)?, - updated_at, - ); - } - - for (phone_number, user_id) in &snapshot.phone_to_user_id { - upsert_auth_snapshot_row( - ctx, - auth_store_snapshot_phone_row_id(phone_number, user_id), - to_snapshot_row_json("认证快照手机号索引", user_id)?, - updated_at, - ); - } - - for session in snapshot.sessions_by_id.values() { - upsert_auth_snapshot_row( - ctx, - auth_store_snapshot_session_row_id(&session.session.session_id), - to_snapshot_row_json("认证快照会话", session)?, - updated_at, - ); - } - - for (refresh_token_hash, session_id) in &snapshot.session_id_by_refresh_token_hash { - upsert_auth_snapshot_row( - ctx, - auth_store_snapshot_session_hash_row_id(refresh_token_hash, session_id), - to_snapshot_row_json("认证快照 refresh token 索引", session_id)?, - updated_at, - ); - } - - for identity in snapshot.wechat_identity_by_provider_uid.values() { - upsert_auth_snapshot_row( - ctx, - auth_store_snapshot_wechat_row_id(&identity.provider_uid, &identity.user_id), - to_snapshot_row_json("认证快照微信身份", identity)?, - updated_at, - ); - } - - for (union_id, user_id) in &snapshot.user_id_by_provider_union_id { - upsert_auth_snapshot_row( - ctx, - auth_store_snapshot_union_row_id(union_id, user_id), - to_snapshot_row_json("认证快照微信 union 索引", user_id)?, - updated_at, - ); - } - - Ok(()) -} - -fn auth_store_snapshot_row_ids( - snapshot: &PersistentAuthStoreSnapshot, -) -> std::collections::HashSet { - let mut ids = std::collections::HashSet::new(); - ids.insert(AUTH_STORE_SNAPSHOT_META_NEXT_USER_ID.to_string()); - for user in snapshot.users_by_username.values() { - ids.insert(auth_store_snapshot_user_row_id(&user.user.id)); - } - for (phone_number, user_id) in &snapshot.phone_to_user_id { - ids.insert(auth_store_snapshot_phone_row_id(phone_number, user_id)); - } - for session in snapshot.sessions_by_id.values() { - ids.insert(auth_store_snapshot_session_row_id( - &session.session.session_id, - )); - } - for (refresh_token_hash, session_id) in &snapshot.session_id_by_refresh_token_hash { - ids.insert(auth_store_snapshot_session_hash_row_id( - refresh_token_hash, - session_id, - )); - } - for identity in snapshot.wechat_identity_by_provider_uid.values() { - ids.insert(auth_store_snapshot_wechat_row_id( - &identity.provider_uid, - &identity.user_id, - )); - } - for (union_id, user_id) in &snapshot.user_id_by_provider_union_id { - ids.insert(auth_store_snapshot_union_row_id(union_id, user_id)); - } - ids } fn export_auth_store_snapshot_from_tables_tx( ctx: &ReducerContext, ) -> Result { - let users = ctx.db.user_account().iter().collect::>(); - let identities = ctx.db.auth_identity().iter().collect::>(); - let sessions = ctx.db.refresh_session().iter().collect::>(); + let mut users = ctx.db.user_account().iter().collect::>(); + let mut identities = ctx.db.auth_identity().iter().collect::>(); + let mut sessions = ctx.db.refresh_session().iter().collect::>(); + if users.is_empty() && identities.is_empty() && sessions.is_empty() { + transfer_latest_legacy_auth_store_snapshot_to_tables(ctx)?; + users = ctx.db.user_account().iter().collect::>(); + identities = ctx.db.auth_identity().iter().collect::>(); + sessions = ctx.db.refresh_session().iter().collect::>(); + } + clear_legacy_auth_store_snapshot_rows(ctx); if users.is_empty() && identities.is_empty() && sessions.is_empty() { return Ok(AuthStoreSnapshotRecord { snapshot_json: None, @@ -455,9 +285,6 @@ fn export_auth_store_snapshot_from_tables_tx( .find(&AUTH_STORE_PROJECTION_META_ID.to_string()) .map(|row| row.updated_at.to_micros_since_unix_epoch()); let snapshot = build_auth_store_snapshot_from_rows(users, identities, sessions)?; - if let Some(updated_at_micros) = updated_at_micros { - upsert_auth_store_snapshot_rows(ctx, &snapshot, updated_at_micros)?; - } let snapshot_json = serde_json::to_string_pretty(&snapshot) .map_err(|error| format!("序列化认证快照失败：{error}"))?; @@ -467,6 +294,29 @@ fn export_auth_store_snapshot_from_tables_tx( }) } +fn transfer_latest_legacy_auth_store_snapshot_to_tables( + ctx: &ReducerContext, +) -> Result<(), String> { + let Some(row) = ctx + .db + .auth_store_snapshot() + .iter() + .max_by_key(|row| row.updated_at.to_micros_since_unix_epoch()) + else { + return Ok(()); + }; + let snapshot_json = row.snapshot_json; + let updated_at_micros = row.updated_at.to_micros_since_unix_epoch(); + let parsed = serde_json::from_str::(snapshot_json.trim()) + .map_err(|error| format!("旧认证快照 JSON 解析失败：{error}"))?; + if !snapshot_has_user_rows(&parsed) { + clear_legacy_auth_store_snapshot_rows(ctx); + return Ok(()); + } + import_auth_store_snapshot_value_tx(ctx, parsed, updated_at_micros)?; + Ok(()) +} + fn build_auth_store_snapshot_from_rows( users: Vec, identities: Vec, @@ -477,6 +327,7 @@ fn build_auth_store_snapshot_from_rows( .map(|user| user.user_id.clone()) .collect::>(); let mut phone_identity_by_user_id = std::collections::HashMap::new(); + let mut phone_user_id_by_phone = std::collections::HashMap::new(); let mut wechat_identity_by_provider_uid = std::collections::HashMap::new(); let mut user_id_by_provider_union_id = std::collections::HashMap::new(); @@ -487,11 +338,13 @@ fn build_auth_store_snapshot_from_rows( match identity.provider.as_str() { "phone" => { + let user_id = identity.user_id.clone(); let phone_number = identity .phone_e164 .clone() .unwrap_or_else(|| identity.provider_uid.clone()); - phone_identity_by_user_id.insert(identity.user_id, phone_number); + phone_identity_by_user_id.insert(user_id.clone(), phone_number.clone()); + phone_user_id_by_phone.insert(phone_number, user_id); } "wechat" => { if let Some(union_id) = identity.provider_union_id.clone() { @@ -513,7 +366,7 @@ fn build_auth_store_snapshot_from_rows( } let mut next_user_id = 1_u64; - let mut phone_to_user_id = std::collections::HashMap::new(); + let mut phone_to_user_id = phone_user_id_by_phone; let mut users_by_username = std::collections::HashMap::new(); for user in users { next_user_id = @@ -523,7 +376,9 @@ fn build_auth_store_snapshot_from_rows( .clone() .or_else(|| phone_identity_by_user_id.remove(&user.user_id)); if let Some(phone_number) = phone_number.clone() { - phone_to_user_id.insert(phone_number, user.user_id.clone()); + phone_to_user_id + .entry(phone_number) + .or_insert_with(|| user.user_id.clone()); } let auth_user = AuthUserSnapshot { id: user.user_id.clone(), @@ -628,115 +483,6 @@ fn upsert_auth_projection_meta(ctx: &ReducerContext, updated_at_micros: i64) { mod tests { use super::*; - fn sample_snapshot() -> PersistentAuthStoreSnapshot { - let user = StoredPasswordUserSnapshot { - user: AuthUserSnapshot { - id: "user_00000042".to_string(), - public_user_code: "GN-000042".to_string(), - username: "phone_42".to_string(), - display_name: "测试玩家".to_string(), - avatar_url: None, - phone_number_masked: Some("138****8000".to_string()), - login_method: "phone".to_string(), - binding_status: "active".to_string(), - wechat_bound: true, - token_version: 3, - user_tags: vec!["early".to_string()], - }, - password_hash: "hash-42".to_string(), - password_login_enabled: true, - phone_number: Some("+8613800008000".to_string()), - }; - let session = StoredRefreshSessionSnapshot { - session: RefreshSessionSnapshot { - session_id: "usess_42".to_string(), - user_id: "user_00000042".to_string(), - refresh_token_hash: "refresh-hash-42".to_string(), - issued_by_provider: "phone".to_string(), - client_info: serde_json::json!({"clientType":"web"}), - expires_at: "2026-06-01T00:00:00Z".to_string(), - revoked_at: None, - created_at: "2026-05-27T00:00:00Z".to_string(), - updated_at: "2026-05-27T00:00:00Z".to_string(), - last_seen_at: "2026-05-27T00:00:00Z".to_string(), - }, - }; - let identity = StoredWechatIdentitySnapshot { - user_id: "user_00000042".to_string(), - provider_uid: "wx-openid-42".to_string(), - provider_union_id: Some("wx-union-42".to_string()), - display_name: Some("微信玩家".to_string()), - avatar_url: None, - }; - - PersistentAuthStoreSnapshot { - next_user_id: 43, - users_by_username: std::collections::HashMap::from([("phone_42".to_string(), user)]), - phone_to_user_id: std::collections::HashMap::from([( - "+8613800008000".to_string(), - "user_00000042".to_string(), - )]), - sessions_by_id: std::collections::HashMap::from([("usess_42".to_string(), session)]), - session_id_by_refresh_token_hash: std::collections::HashMap::from([( - "refresh-hash-42".to_string(), - "usess_42".to_string(), - )]), - wechat_identity_by_provider_uid: std::collections::HashMap::from([( - "wx-openid-42".to_string(), - identity, - )]), - user_id_by_provider_union_id: std::collections::HashMap::from([( - "wx-union-42".to_string(), - "user_00000042".to_string(), - )]), - } - } - - #[test] - fn auth_store_snapshot_row_ids_are_row_level_without_default_aggregate() { - let ids = auth_store_snapshot_row_ids(&sample_snapshot()); - - assert!(!ids.contains("default")); - assert!(ids.contains(AUTH_STORE_SNAPSHOT_META_NEXT_USER_ID)); - assert!(ids.contains(&auth_store_snapshot_user_row_id("user_00000042"))); - assert!(ids.contains(&auth_store_snapshot_phone_row_id( - "+8613800008000", - "user_00000042" - ))); - assert!(ids.contains(&auth_store_snapshot_session_row_id("usess_42"))); - assert!(ids.contains(&auth_store_snapshot_session_hash_row_id( - "refresh-hash-42", - "usess_42" - ))); - assert!(ids.contains(&auth_store_snapshot_wechat_row_id( - "wx-openid-42", - "user_00000042" - ))); - assert!(ids.contains(&auth_store_snapshot_union_row_id( - "wx-union-42", - "user_00000042" - ))); - } - - #[test] - fn auth_store_snapshot_user_row_key_is_stable_after_username_change() { - let before = sample_snapshot(); - let mut after = sample_snapshot(); - let mut renamed_user = after - .users_by_username - .remove("phone_42") - .expect("sample user exists"); - renamed_user.user.username = "renamed_42".to_string(); - after - .users_by_username - .insert("renamed_42".to_string(), renamed_user); - - assert_eq!( - auth_store_snapshot_row_ids(&before), - auth_store_snapshot_row_ids(&after) - ); - } - #[test] fn auth_export_ignores_phone_identity_without_user_account() { let live_user = UserAccount { @@ -780,6 +526,57 @@ mod tests { assert!(!snapshot.phone_to_user_id.contains_key("+8613900009999")); } + #[test] + fn auth_export_prefers_phone_identity_when_phone_user_accounts_are_duplicated() { + fn phone_user(user_id: &str, public_user_code: &str, username: &str) -> UserAccount { + UserAccount { + user_id: user_id.to_string(), + public_user_code: public_user_code.to_string(), + username: username.to_string(), + display_name: "138****8000".to_string(), + avatar_url: None, + phone_number_masked: Some("138****8000".to_string()), + phone_number_e164: Some("+8613800008000".to_string()), + login_method: "Phone".to_string(), + binding_status: "Active".to_string(), + wechat_bound: false, + password_hash: "hash-live".to_string(), + password_login_enabled: true, + token_version: 1, + user_tags: Some(vec![]), + } + } + + let identity = AuthIdentity { + identity_id: "authi_phone_8613800008000".to_string(), + user_id: "user_identity".to_string(), + provider: "phone".to_string(), + provider_uid: "+8613800008000".to_string(), + provider_union_id: None, + phone_e164: Some("+8613800008000".to_string()), + display_name: None, + avatar_url: None, + }; + + let snapshot = build_auth_store_snapshot_from_rows( + vec![ + phone_user("user_duplicate", "SY-00000042", "phone_duplicate"), + phone_user("user_identity", "SY-00000023", "phone_identity"), + ], + vec![identity], + vec![], + ) + .expect("auth rows should export"); + + assert_eq!( + snapshot + .phone_to_user_id + .get("+8613800008000") + .map(String::as_str), + Some("user_identity") + ); + } + #[test] fn auth_export_next_user_id_follows_public_user_code_for_uuid_user_ids() { let users = vec![ @@ -822,4 +619,23 @@ mod tests { assert_eq!(snapshot.next_user_id, 108); } + + #[test] + fn auth_legacy_empty_snapshot_is_cleanup_only() { + let snapshot = serde_json::to_string(&PersistentAuthStoreSnapshot { + next_user_id: 1, + users_by_username: std::collections::HashMap::new(), + phone_to_user_id: std::collections::HashMap::new(), + sessions_by_id: std::collections::HashMap::new(), + session_id_by_refresh_token_hash: std::collections::HashMap::new(), + wechat_identity_by_provider_uid: std::collections::HashMap::new(), + user_id_by_provider_union_id: std::collections::HashMap::new(), + }) + .expect("legacy snapshot should serialize"); + + let parsed = serde_json::from_str::(&snapshot) + .expect("legacy snapshot should parse"); + + assert!(!snapshot_has_user_rows(&parsed)); + } }