GenarrativeAI/Genarrative
10
0
Fork 0
Code Issues 20 Pull Requests 8 Actions Packages Projects 1 Releases Wiki Activity

修复认证短状态与退款 outbox 跨节点恢复

Browse Source 
将短信验证码与微信 OAuth state 纳入 SpacetimeDB typed projection 并支持 CAS 只读刷新

为退款 emergency spool 增加 overflow 文件与崩溃恢复，避免容量上限静默丢失

为 API 与 external worker 配置独立 tracking/refund 持久卷

使用 profile refund outbox 的 status/available_at 索引扫描并同步架构运维文档
This commit is contained in:
段舒康
2026-08-27 19:11:23 +08:00
parent 6da1eecbb6
commit 6e9948ac9e
22 changed files with 850 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/container/docker-compose.loadtest.yml
+7
View File
@@ -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:
docs/project-memory/shared-memory/decision-log.md
+16
View File
@@ -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<u8>` 读缓存别名和 Rust string 默认值支持，2.8.2 修复 table accessor 改名自动迁移，2.8.3 修复 scheduled function 从实际执行时间重排导致的长期漂移。仓库若继续锁定 2.7.0，会保留这些已知运行时与 SDK 问题。
docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md
+4 -3
View File
File diff suppressed because one or more lines are too long
docs/【开发运维】本地开发验证与生产运维-2026-05-15.md
+2 -2
View File
@@ -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/<version>`，`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` 数据。
server-rs/crates/api-server/src/asset_billing.rs
+2 -2
View File
@@ -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) => {
server-rs/crates/api-server/src/password_management.rs
+13 -2
View File
@@ -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,
server-rs/crates/api-server/src/phone_auth.rs
+47 -11
View File
@@ -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(),
server-rs/crates/api-server/src/state.rs
+44 -1
View File
@@ -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<Self, AppStateInitError> {
@@ -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);
server-rs/crates/api-server/src/wallet_refund_outbox.rs
+130 -75
View File
@@ -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<u64, std::io::Error> {
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");
server-rs/crates/api-server/src/wechat/auth.rs
+80 -5
View File
@@ -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(
server-rs/crates/module-auth/src/domain.rs
+31 -2
View File
@@ -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<AuthStoreProjectionUser>,
pub identities: Vec<AuthStoreProjectionIdentity>,
pub refresh_sessions: Vec<AuthStoreProjectionRefreshSession>,
#[serde(default)]
pub phone_codes: Vec<AuthStoreProjectionPhoneCode>,
#[serde(default)]
pub wechat_states: Vec<AuthStoreProjectionWechatState>,
/// 当前进程工作集所基于的正式投影版本，用于事务内 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<String>,
}
#[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<String>,
pub bind_user_id: Option<String>,
pub expires_at: String,
pub consumed_at: Option<String>,
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) {
server-rs/crates/module-auth/src/lib.rs
+220 -18
View File
@@ -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<Mutex<InMemoryAuthStoreState>>,
@@ -135,6 +137,24 @@ fn parse_auth_binding_status(value: &str) -> AuthBindingStatus {
}
}
fn parse_phone_auth_scene(value: &str) -> Option<PhoneAuthScene> {
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<WechatAuthScene> {
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<String>,
) -> 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<Self, String> {
Ok(Self {
@@ -1215,10 +1308,14 @@ impl InMemoryAuthStore {
&self,
updated_at_micros: i64,
) -> Result<AuthStoreProjectionView, String> {
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::<Result<Vec<_>, 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());
server-rs/crates/spacetime-client/src/active/mapper/auth.rs
+62
View File
@@ -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(),
}
}
server-rs/crates/spacetime-client/src/mapper/auth.rs
+58
View File
@@ -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(),
}
}
server-rs/crates/spacetime-client/src/module_bindings.rs
+4
View File
@@ -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;
server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_meta_type.rs
+6
View File
@@ -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<String>,
pub wechat_states_json: Option<String>,
}
impl __sdk::InModule for AuthStoreProjectionMeta {
@@ -21,6 +23,8 @@ impl __sdk::InModule for AuthStoreProjectionMeta {
pub struct AuthStoreProjectionMetaCols {
pub meta_id: __sdk::__query_builder::Col<AuthStoreProjectionMeta, String>,
pub updated_at: __sdk::__query_builder::Col<AuthStoreProjectionMeta, __sdk::Timestamp>,
pub phone_codes_json: __sdk::__query_builder::Col<AuthStoreProjectionMeta, Option<String>>,
pub wechat_states_json: __sdk::__query_builder::Col<AuthStoreProjectionMeta, Option<String>>,
}
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"),
}
}
}
server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_phone_code_type.rs
+21
View File
@@ -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<String>,
}
impl __sdk::InModule for AuthStoreProjectionPhoneCode {
type Module = super::RemoteModule;
}
server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_view_type.rs
+4
View File
@@ -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<AuthStoreProjectionUser>,
pub identities: Vec<AuthStoreProjectionIdentity>,
pub refresh_sessions: Vec<AuthStoreProjectionRefreshSession>,
pub phone_codes: Vec<AuthStoreProjectionPhoneCode>,
pub wechat_states: Vec<AuthStoreProjectionWechatState>,
pub base_updated_at_micros: i64,
}
server-rs/crates/spacetime-client/src/module_bindings/auth_store_projection_wechat_state_type.rs
+24
View File
@@ -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<String>,
pub bind_user_id: Option<String>,
pub expires_at: String,
pub consumed_at: Option<String>,
pub created_at: String,
pub updated_at: String,
}
impl __sdk::InModule for AuthStoreProjectionWechatState {
type Module = super::RemoteModule;
}
server-rs/crates/spacetime-module/src/auth/procedures.rs
+68 -4
View File
@@ -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<AuthStoreProjectionUser>,
pub identities: Vec<AuthStoreProjectionIdentity>,
pub refresh_sessions: Vec<AuthStoreProjectionRefreshSession>,
pub phone_codes: Vec<AuthStoreProjectionPhoneCode>,
pub wechat_states: Vec<AuthStoreProjectionWechatState>,
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<String>,
}
#[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<String>,
pub bind_user_id: Option<String>,
pub expires_at: String,
pub consumed_at: Option<String>,
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<AuthStoreProjectionView, String> {
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),
});
}
server-rs/crates/spacetime-module/src/auth/tables.rs
+4
View File
@@ -5,6 +5,10 @@ pub struct AuthStoreProjectionMeta {
#[primary_key]
pub(crate) meta_id: String,
pub(crate) updated_at: Timestamp,
#[default(None::<String>)]
pub(crate) phone_codes_json: Option<String>,
#[default(None::<String>)]
pub(crate) wechat_states_json: Option<String>,
}
#[spacetimedb::table(
server-rs/crates/spacetime-module/src/runtime/active/profile.rs
+3 -4
View File
@@ -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::<Vec<_>>();
rows.sort_by(|left, right| {
left.available_at