From 5bf036bb81e8a47eef295683ea820903eb865452 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=AE=B5=E8=88=92=E5=BA=B7?= Date: Sat, 19 Sep 2026 12:29:19 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=B6=E6=95=9B=E5=90=8E=E7=AB=AF=E4=BE=9D?= =?UTF-8?q?=E8=B5=96=E8=A3=85=E9=85=8D=E4=B8=8E=E9=89=B4=E6=9D=83=E5=B9=B6?= =?UTF-8?q?=E5=AE=8C=E5=96=84=E5=BC=82=E6=AD=A5=E8=BF=BD=E8=B8=AA=20(#425)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit HTTP 请求取消后，在途计数原先无法释放；项目元数据和 External API 鉴权依赖完整 AppState，相同鉴权和追踪配置也散落在多个入口。本次集中装配这些依赖和横切能力，保持现有公开 API、权限、计费、幂等和事务规则。 ## 修改 - 91 个受保护路由集中应用鉴权；保留方法级 404/405/HEAD/Allow、公开入口、MCP、精选缓存头及 2/4 MiB 请求限制。 - 七个项目元数据入口改用缓存的 EditorProjectState，External/MCP 鉴权改用 ExternalApiAuthState；生产实现复用 SpacetimeClient，媒体修复维持原有上传和登记顺序。 - RAII 覆盖请求 Future 取消和 panic unwind 的计数清理；正常与降级服务复用 TraceLayer，指标采用 MatchedPath 模板及固定兜底。 - LLM 普通与流式调用增加跳过参数的异步 span，保持父上下文、流式回调、错误与重试行为；补充替代依赖测试并同步锁文件和文档。 ## 验证 | 验证面 | 结果 | | --- | --- | | platform-llm 完整本地回归 | 161 个单元测试、3 个集成测试通过；1 个真实 Provider 用例按原配置忽略 | | api-server 完整回归 | 执行时 1075 通过、12 失败、6 忽略；其中 1 个新增公开读取 fixture 断言已修正，14 个路由契约回归随后全部通过；剩余 11 个是下述既有 Windows 失败 | | 窄依赖、取消与追踪 | 元数据 owner/幂等/revision、鉴权及 MCP 错误传播、取消/panic/流式响应、追踪父子关系与敏感参数省略均通过 | | 实际本地服务 | 独立 SpacetimeDB 上 102/102 检查通过，两个动态项目 ID 的路由模板及请求 ID 日志核验 3/3 通过 | | 编译与边界 | api-server cargo check、AGC 锁文件下 platform-llm cargo check、rustfmt、编码、文档索引、DDD 与 diff 检查通过 | | 合入最新 master 后 | 后端源码及锁文件保持已测内容；再次通过 14 个路由契约测试、3 个 Provider 追踪测试及编码/文档/DDD/diff 检查 | 实际服务检查覆盖 health/ready、两账号登录、项目 CRUD、幂等重复、跨 owner 拒绝、revision 冲突、External/MCP 读取、Key 撤销及 404/405。使用既有 test 环境的本地 Router 拒绝 fixture，未调用真实付费 Provider；自建服务已关闭，原开发实例保留。 ## 已知测试限制 API 全量测试尚未全绿：11 个 wallet_refund_outbox 用例在 Windows 的目录同步处失败。其生产文件与变更前内容一致；标准库隔离复现确认 File::open(目录) 返回 OS 5，而普通文件写入、同步及 hard_link 正常。这个已有的目录持久化问题未混入本次重构，也未通过跳过或弱化相关断言掩盖。 --------- Co-authored-by: kdletters <61648117+kdletters@users.noreply.github.com> Reviewed-on: http://192.168.35.82/git/GenarrativeAI/Genarrative/pulls/425 --- .../src-tauri/Cargo.lock | 1 + .../shared-memory/team-conventions.md | 2 + ...】server-rs与SpacetimeDB数据契约-2026-05-15.md | 9 + ...发运维】本地开发验证与生产运维-2026-05-15.md | 6 +- server-rs/Cargo.lock | 2 + server-rs/crates/api-server/src/app.rs | 384 ++++++--- .../crates/api-server/src/editor_project.rs | 90 +- .../src/editor_project/metadata_tests.rs | 364 ++++++++ .../api-server/src/external_api_auth.rs | 186 ++++- .../crates/api-server/src/modules/admin.rs | 556 +++++++------ .../api-server/src/modules/editor_project.rs | 779 +++++++++++++----- .../api-server/src/modules/external_api.rs | 595 +++++++++---- server-rs/crates/api-server/src/state.rs | 26 + .../api-server/src/state/external_api_auth.rs | 47 ++ .../api-server/src/state/project_metadata.rs | 195 +++++ server-rs/crates/api-server/src/telemetry.rs | 274 +++++- server-rs/crates/platform-llm/Cargo.toml | 4 +- server-rs/crates/platform-llm/src/lib.rs | 23 + .../platform-llm/src/observability_tests.rs | 332 ++++++++ 19 files changed, 3059 insertions(+), 816 deletions(-) create mode 100644 server-rs/crates/api-server/src/editor_project/metadata_tests.rs create mode 100644 server-rs/crates/api-server/src/state/external_api_auth.rs create mode 100644 server-rs/crates/api-server/src/state/project_metadata.rs create mode 100644 server-rs/crates/platform-llm/src/observability_tests.rs diff --git a/apps/ai-game-creator-shell/src-tauri/Cargo.lock b/apps/ai-game-creator-shell/src-tauri/Cargo.lock index 21f1d6233..7a8d82d5f 100644 --- a/apps/ai-game-creator-shell/src-tauri/Cargo.lock +++ b/apps/ai-game-creator-shell/src-tauri/Cargo.lock @@ -3931,6 +3931,7 @@ dependencies = [ "serde", "serde_json", "tokio", + "tracing", ] [[package]] diff --git a/docs/project-memory/shared-memory/team-conventions.md b/docs/project-memory/shared-memory/team-conventions.md index 84df79771..3e3c71e64 100644 --- a/docs/project-memory/shared-memory/team-conventions.md +++ b/docs/project-memory/shared-memory/team-conventions.md @@ -32,6 +32,8 @@ - 修改 `/api/external/v1` 时，同批更新 `docs/openapi/genarrative-external-v1.openapi.json` 与契约测试。 - 修改 SpacetimeDB schema 时遵守字段追加/default 约束，同步 migration、表目录、生成绑定，并运行 schema 检查；删除、改名、重排或改类型前先确认迁移计划。 - 日志不递归输出完整配置、应用状态或 provider client；新增字段默认不进入安全摘要。 +- HTTP 横切能力集中在 Axum/Tower 中间件：正常与降级路由复用追踪层；指标与 trace 使用 `MatchedPath` 模板及固定兜底，不把请求 ID、实际资源 ID 或 query 放入指标标签。在途请求通过 RAII guard 覆盖 Future 取消与 panic unwind；请求执行和响应体存活分别计量，不能把 handler 耗时当作 SSE 全生命周期。 +- 业务依赖在组合根显式装配，Axum `FromRef` 只抽取可浅拷贝的窄能力。项目元数据与 External API 鉴权不持有完整 `AppState`，测试经相同接口注入替代依赖。集中鉴权仍保留方法级 fallback、公开入口、MCP 和 body limit 顺序；Provider span 跳过完整参数，不隐藏计费、重试、幂等或事务规则。 - 中文文案、注释和文档保持 UTF-8，优先局部补丁，不擅自翻译成英文。 ## 文档生命周期 diff --git a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md index dfab9a03e..a65dd79ef 100644 --- a/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md +++ b/docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md @@ -47,6 +47,15 @@ SpacetimeDB 版本口径：当前 Rust crate `spacetimedb`、`spacetimedb-sdk` npm run check:server-rs-ddd ``` +## 依赖装配与横切能力 + +- 启动入口负责构造共享依赖，业务入口通过 Axum `State` / `FromRef` 获取所需能力。项目元数据链路与 External API 鉴权使用窄状态；窄状态不得持有完整 `AppState`、通过 `Deref` 暴露完整配置，或提供 `root_state()` 逃逸。仅在需要替换或隔离测试的外部能力处抽取接口，生产实现仍复用现有 `spacetime-client` facade，不新建数据库访问通道。 +- 后台、External API 和编辑器路由按同一种身份策略聚合鉴权，公开登录、公开文档、公开精选读取及 MCP 保持独立边界。路由、HTTP 方法、404/405、请求大小限制、授权和错误响应合同保持不变；后台 member 的实时权限校验和资源 owner 校验继续由现有权威逻辑执行。 +- HTTP 观测使用统一 Tower 追踪层与 `MatchedPath` 模板；请求计数通过 RAII 覆盖取消与 unwind，响应体存活单独统计。具体指标合同见开发运维文档。 +- Provider 函数级追踪采用现有 tracing 能力，显式列出 provider、operation、model 等非秘密字段，跳过完整参数、配置、凭据和消息正文；异步 span 覆盖实际执行与等待，不能只记录 Future 创建。装饰与追踪保持原始成功值、错误、重试次数、流式回调和取消语义。 +- 扣费、退款、幂等、资源登记、重试判定与事务边界保持显式业务流程；不引入自动扫描 IOC 容器、通用 AOP 框架或跨外部副作用的隐式事务。 +- 验收覆盖：所有受保护路由的匿名拒绝及公开入口可访问；原有 404/405 和 body limit；窄依赖可独立构造和替换；Provider 成功/失败、异步追踪归属与脱敏；已有 API、计费和幂等回归。真实本地服务 smoke 必须记录所用数据库及依赖可用性，不能用单元测试代替运行时证据。 + ## `spacetime-client` mapper 组织 `spacetime-client` 的 Cargo `lib.path` 指向 `src/active.rs`，现役 mapper 聚合入口是 `src/active/mapper.rs`；原旧 facade 和 mapper 已删除。 diff --git a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md index 92e4d4fc6..2b884aad3 100644 --- a/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md +++ b/docs/【开发运维】本地开发验证与生产运维-2026-05-15.md @@ -808,13 +808,15 @@ OpenTelemetry 现阶段默认开启 OTLP traces / metrics / logs，但本地日 - 应用日志按进程查看：父 API 使用 `journalctl -u genarrative-api.service`，独立 BgFilter worker 使用 `journalctl -u genarrative-bgfilter-worker.service`；Nginx 日志仍写文件。日志等级继续用 `GENARRATIVE_API_LOG` / `RUST_LOG` 控制，例如 `info,tower_http=info,spacetime_client=info`。 - debug exporter / Rider 转发都会同时接收 traces、metrics 和 logs。 - api-server 会随 metrics 发送进程级指标：`process.memory.usage`、`process.memory.virtual`、`process.cpu.time`、`genarrative.process.cpu.usage_percent`、`process.thread.count`、`genarrative.process.memory.private`；Windows 额外发送 `process.windows.handle.count`，Linux 额外发送 `process.unix.file_descriptor.count`。这些指标只描述当前进程，不携带请求、用户或作品 label。 -- HTTP 运行态补充发送 `genarrative.http.server.response_bodies.in_flight` 与 `genarrative.http.server.request_permits.available`，后者带低基数 `pool=default|gallery|detail|admin` label，用于区分业务 handler / 背压 permit 是否仍被占用；拼图广场热点缓存补充发送 `genarrative.puzzle_gallery.cache.*` 指标，记录 fresh hit、stale hit、未命中、后台刷新开始 / 失败、重建耗时和预序列化 data JSON 字节数。 +- HTTP 请求在途指标 `http.server.active_requests` 从进入观测中间件计数，到生成 `Response` 时结束；使用 RAII guard 保证正常返回、Future 被取消释放和 panic unwind 时都按原始 method / route 标签且仅递减一次。`http.server.request.duration` 采用同一执行区间，仅记录已生成响应的请求，不表示响应已发送到客户端，也不表示 SSE 已结束。响应体存活另由 `genarrative.http.server.response_bodies.in_flight` 统计；背压 permit 由 `genarrative.http.server.request_permits.available{pool=default|admin}` 统计。 +- 正常服务与 SpacetimeDB 不可用时的降级路由复用同一 HTTP `TraceLayer` 构造函数。请求上下文在追踪层外侧，错误归一化和响应头回写在追踪层内侧，保证提前拒绝和降级响应仍记录同一 request ID 与最终状态。 +- `platform-llm` 的普通与流式调用统一生成 `llm.request` 子 span，字段白名单为 `provider`、`operation`、`api_kind`、`model`，继承调用方当前 span；范围覆盖响应读取、流式回调和异步等待，取消释放 Future 时结束。完整 client/config、API Key、请求/响应正文不进入该 span；既有 Provider 错误和重试策略保持原样，不能用遥测包装吞掉失败或自动重放。 - 外部 API 失败统一发送 OTLP 并落库。当前 VectorEngine 图片生成 / 编辑失败由 `platform-image` provider 输出结构化日志字段，字段包括 provider、endpoint、failure_stage、status、source、source_chain、source_chain_depth、timeout、retryable、latency_ms、prompt_chars、reference_image_count、实际 provider `image_model`、request_params 和 raw_excerpt；发生模型回退时另带 `fallback_from_model` / `fallback_to_model`。图片编辑请求参数日志还会带 reference_image_bytes_total，并在 request_params.referenceImages 中记录每个 multipart `image` part 的 fileName、mimeType 和 bytes，不记录 API key 或原始图片 bytes；`api-server` 再记录指标 `genarrative.external_api.failures{provider,failure_stage,status_class,retryable}`，并写入 `tracking_event`，`event_key = external_api_call_failure`、`module_key = external-api`、`scope_kind = module`、`scope_id = provider`。调用方能拿到身份上下文时，失败事件还会在行级 `user_id` / `owner_user_id` / `profile_id` 和 `metadata_json.userId` / `metadata_json.profileId` / `metadata_json.requestId` / `metadata_json.errorSource` 中记录触发者、草稿 / 作品作用域、请求标识和传输错误链。排障时先按 provider / failureStage / imageModel 聚合，再下钻 userId / profileId，最后结合 request 日志、errorSource 和上游响应 excerpt 判断是模型不可用、限流、超时、解析失败还是未返回图片。 - OSS 平台适配器也输出结构化日志，覆盖 `sign_post_object`、`sign_get_object_url`、`head_object` 和 `put_object`。排查资产签名、上传或确认失败时，先按 `provider=aliyun-oss` 与 `operation` 过滤，再看 `object_key` / `key_prefix`、`status`、`status_class`、`error_kind`、`content_length`、`content_type` 和 `elapsed_ms`；角色动画逐帧额外按 `frame_index`、`operation=source_put|final_put|final_head`、`attempt/max_attempts`、`will_retry`、`oss_code` 和 `oss_request_id` 对齐同一对象的请求尝试。`请求 OSS 失败` 时，`timeout/connect/transport=true` 表示传输类失败，OSS PutObject 的 `status=400, oss_code=RequestTimeout, timeout=true`、`status=429` 或 `500–599` 表示暂时性失败，PUT 的 `status=400`、`oss_code` 为空且 `timeout=true` 或 `transport=true`（message 含「错误响应体读取失败」，即 400 错误体读取超时/断流）也会重试；除这两类例外外，其他 400、401/403/404、配置、URL 和签名错误是确定性失败，不会重试。最终帧 HEAD 失败只会重试 HEAD，不会重复 PUT。日志不得包含 AccessKey、policy、signature、Authorization header、完整 signed URL 或 OSS 错误响应体；`oss_request_id` 只用于关联 OSS 服务端排障。排查 generated 图片重复下载时，先确认前端输入是否为 `/generated-*` legacy path 或可归一化的 `https://*.oss-*.aliyuncs.com/generated-*`；正确链路应先调 `/api/assets/read-url`，再由浏览器请求 signed URL，且同一路径、同一 `refreshKey` 版本和未临近过期的 signed URL 应复用。新上传 generated 私有对象应带 `Cache-Control: public, max-age=31536000, immutable`；旧对象若只有 `ETag` / `Last-Modified`，浏览器会走 304 协商缓存而不是长期强缓存，可通过刷新 OSS 元数据或 CDN 配置补齐。 - SpacetimeDB 观测分为两类：procedure / reducer 调用继续用 `genarrative.spacetime.procedure.*`，订阅本地 cache 读使用 `genarrative.spacetime.read.*`。`read=list_puzzle_gallery` 表示拼图广场当前从 `puzzle_gallery_card_view` 本地 cache 读取，不再每个 HTTP 请求调用 `list_puzzle_gallery` procedure。 - 本地 Windows 直连压测的内存高水位要结合 K6 VU / 连接数解释。250 RPS 下过高 `PREALLOCATED_VUS` 可能让 300 个本地 Established 连接把 `api-server` private memory 瞬时推到 GB 级，且 `/healthz` 小响应也能复现；若压测结束后回落、`response_bodies.in_flight` 和背压 permit 未显示业务积压，应优先按连接 / 发送链路高水位处理，而不是判断为 SpacetimeDB 或 JSON 缓存泄漏。 - Rider 的 Logs 面板只展示 log event 自身字段，不会自动展开父 span 的全部 attributes；请求完成日志会直接带 `request_id`、`http.request.method`、`http.route`、`url.scheme`、`url.path`、`http.response.status_code`、`status_class`、`latency_ms` 和 `slow_request`，完整链路继续到 Traces 面板按 trace/span 查看。 -- 指标 label 只允许低基数字段：HTTP 使用 `method`、`route`、`status_class`，SpacetimeDB 调用使用 `procedure`、`status_class`；`request_id` 只进入 trace/log attribute，不进入 metric label。 +- 指标 label 只允许低基数字段：HTTP 使用 `http.request.method`、`http.route`、`status_class`，SpacetimeDB 调用使用 `procedure`、`status_class`；`request_id` 只进入 trace/log attribute，不进入 metric label。HTTP trace、完成日志与指标共用 Axum `MatchedPath` 路由模板，例如 `/api/editor/projects/{project_id}`；未匹配路由或无路由模板的降级请求只使用 `/api/*`、`/admin/api/*`、`other` 三种固定兜底，不把实际 ID 或 query 写入 route 标签。原来按 `/api/*` 聚合的监控查询应改为按路由模板汇总。 常见外部服务变量： diff --git a/server-rs/Cargo.lock b/server-rs/Cargo.lock index 5de6bdfbe..614e6da6b 100644 --- a/server-rs/Cargo.lock +++ b/server-rs/Cargo.lock @@ -4101,6 +4101,8 @@ dependencies = [ "serde", "serde_json", "tokio", + "tracing", + "tracing-subscriber", ] [[package]] diff --git a/server-rs/crates/api-server/src/app.rs b/server-rs/crates/api-server/src/app.rs index 4c34bda86..ded1dafe0 100644 --- a/server-rs/crates/api-server/src/app.rs +++ b/server-rs/crates/api-server/src/app.rs @@ -1,3 +1,5 @@ +use std::time::Duration; + use axum::{ Router, body::Body, @@ -9,8 +11,8 @@ use axum::{ }; use serde_json::json; use tower_http::{ - classify::ServerErrorsFailureClass, - trace::{DefaultOnRequest, TraceLayer}, + classify::{ServerErrorsAsFailures, ServerErrorsFailureClass, SharedClassifier}, + trace::{DefaultOnBodyChunk, DefaultOnEos, DefaultOnRequest, TraceLayer}, }; use tracing::{Level, Span, error, info_span}; @@ -85,66 +87,7 @@ pub fn build_router(state: AppState) -> Router { state.clone(), record_http_observability, )) - // 当前阶段先统一挂接 HTTP tracing，后续 request_id、响应头与错误中间件继续在这里扩展。 - .layer( - TraceLayer::new_for_http() - .make_span_with(|request: &Request| { - let request_id = - resolve_request_id(request).unwrap_or_else(|| "unknown".to_string()); - let route = crate::telemetry::observability_route(request.uri().path()); - let scheme = crate::telemetry::resolve_request_scheme(request.headers()); - let span_name = format!("{} {}", request.method(), route); - - info_span!( - "http.request", - otel.kind = "server", - otel.name = %span_name, - otel.status_code = tracing::field::Empty, - http.response.status_code = tracing::field::Empty, - method = %request.method(), - http.request.method = %request.method(), - http.route = %route, - url.scheme = %scheme, - url.path = %request.uri().path(), - request_id = %request_id, - status = tracing::field::Empty, - latency_ms = tracing::field::Empty, - ) - }) - .on_request(DefaultOnRequest::new().level(Level::INFO)) - .on_response( - |response: &axum::response::Response, - latency: std::time::Duration, - span: &Span| { - let latency_ms = latency.as_millis().min(u64::MAX as u128) as u64; - let status = response.status().as_u16(); - span.record("status", status); - span.record("http.response.status_code", status); - span.record( - "otel.status_code", - if response.status().is_server_error() { - "ERROR" - } else { - "OK" - }, - ); - span.record("latency_ms", latency_ms); - }, - ) - .on_failure( - |failure: ServerErrorsFailureClass, - latency: std::time::Duration, - span: &Span| { - let latency_ms = latency.as_millis().min(u64::MAX as u128) as u64; - error!( - parent: span, - latency_ms, - failure = %failure, - "http request failed" - ); - }, - ), - ) + .layer(http_trace_layer()) // request_id 中间件先进入请求链，确保后续 tracing、错误处理和响应头层都能复用同一份请求标识。 .layer(middleware::from_fn(attach_request_context)) .with_state(state) @@ -159,68 +102,72 @@ pub fn build_spacetime_unavailable_router(message: String) -> Router { // 依赖不可用模式不挂业务 state，统一返回 503，并继续保留 request_id / API 版本 / 耗时响应头。 .layer(middleware::from_fn(normalize_error_response)) .layer(middleware::from_fn(propagate_request_id_header)) - .layer( - TraceLayer::new_for_http() - .make_span_with(|request: &Request| { - let request_id = - resolve_request_id(request).unwrap_or_else(|| "unknown".to_string()); - let route = crate::telemetry::observability_route(request.uri().path()); - let scheme = crate::telemetry::resolve_request_scheme(request.headers()); - let span_name = format!("{} {}", request.method(), route); - - info_span!( - "http.request", - otel.kind = "server", - otel.name = %span_name, - otel.status_code = tracing::field::Empty, - http.response.status_code = tracing::field::Empty, - method = %request.method(), - http.request.method = %request.method(), - http.route = %route, - url.scheme = %scheme, - url.path = %request.uri().path(), - request_id = %request_id, - status = tracing::field::Empty, - latency_ms = tracing::field::Empty, - ) - }) - .on_request(DefaultOnRequest::new().level(Level::INFO)) - .on_response( - |response: &axum::response::Response, - latency: std::time::Duration, - span: &Span| { - let latency_ms = latency.as_millis().min(u64::MAX as u128) as u64; - let status = response.status().as_u16(); - span.record("status", status); - span.record("http.response.status_code", status); - span.record( - "otel.status_code", - if response.status().is_server_error() { - "ERROR" - } else { - "OK" - }, - ); - span.record("latency_ms", latency_ms); - }, - ) - .on_failure( - |failure: ServerErrorsFailureClass, - latency: std::time::Duration, - span: &Span| { - let latency_ms = latency.as_millis().min(u64::MAX as u128) as u64; - error!( - parent: span, - latency_ms, - failure = %failure, - "http request failed" - ); - }, - ), - ) + .layer(http_trace_layer()) .layer(middleware::from_fn(attach_request_context)) } +type HttpTraceLayer = TraceLayer< + SharedClassifier, + fn(&Request) -> Span, + DefaultOnRequest, + fn(&Response, Duration, &Span), + DefaultOnBodyChunk, + DefaultOnEos, + fn(ServerErrorsFailureClass, Duration, &Span), +>; + +fn http_trace_layer() -> HttpTraceLayer { + TraceLayer::new_for_http() + .make_span_with(make_http_request_span as fn(&Request) -> Span) + .on_request(DefaultOnRequest::new().level(Level::INFO)) + .on_response(record_http_response_span as fn(&Response, Duration, &Span)) + .on_failure(record_http_failure as fn(ServerErrorsFailureClass, Duration, &Span)) +} + +fn make_http_request_span(request: &Request) -> Span { + let request_id = resolve_request_id(request).unwrap_or_else(|| "unknown".to_string()); + let route = crate::telemetry::observability_route(request); + let scheme = crate::telemetry::resolve_request_scheme(request.headers()); + let span_name = format!("{} {}", request.method(), route); + + info_span!( + "http.request", + otel.kind = "server", + otel.name = %span_name, + otel.status_code = tracing::field::Empty, + http.response.status_code = tracing::field::Empty, + method = %request.method(), + http.request.method = %request.method(), + http.route = %route, + url.scheme = %scheme, + url.path = %request.uri().path(), + request_id = %request_id, + status = tracing::field::Empty, + latency_ms = tracing::field::Empty, + ) +} + +fn record_http_response_span(response: &Response, latency: Duration, span: &Span) { + let latency_ms = latency.as_millis().min(u64::MAX as u128) as u64; + let status = response.status().as_u16(); + span.record("status", status); + span.record("http.response.status_code", status); + span.record( + "otel.status_code", + if response.status().is_server_error() { + "ERROR" + } else { + "OK" + }, + ); + span.record("latency_ms", latency_ms); +} + +fn record_http_failure(failure: ServerErrorsFailureClass, latency: Duration, span: &Span) { + let latency_ms = latency.as_millis().min(u64::MAX as u128) as u64; + error!(parent: span, latency_ms, failure = %failure, "http request failed"); +} + #[derive(Clone, Debug)] struct SpacetimeUnavailableState { message: std::sync::Arc, @@ -308,6 +255,201 @@ mod tests { const TEST_PASSWORD: &str = "secret123"; const INTERNAL_TEST_SECRET: &str = "test-internal-secret"; + mod http_tracing { + use std::{ + collections::BTreeMap, + sync::{Arc, Mutex}, + }; + + use axum::extract::FromRef; + use tracing::{ + Metadata, Subscriber, + field::{Field, Visit}, + instrument::WithSubscriber, + span::{Attributes, Id, Record}, + }; + + use super::*; + use crate::state::{BackpressureState, HttpRequestPermitPoolKind}; + + #[derive(Clone, Default)] + struct HttpSpanCapture(Arc>>>); + + struct SpanFields<'a>(&'a mut BTreeMap); + + impl Visit for SpanFields<'_> { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.0 + .insert(field.name().to_string(), format!("{value:?}")); + } + + fn record_str(&mut self, field: &Field, value: &str) { + self.0.insert(field.name().to_string(), value.to_string()); + } + } + + impl Subscriber for HttpSpanCapture { + fn enabled(&self, metadata: &Metadata<'_>) -> bool { + metadata.is_span() && metadata.name() == "http.request" + } + + fn new_span(&self, attributes: &Attributes<'_>) -> Id { + let mut spans = self.0.lock().expect("span capture should lock"); + let mut fields = BTreeMap::new(); + attributes.record(&mut SpanFields(&mut fields)); + spans.push(fields); + Id::from_u64(spans.len() as u64) + } + + fn record(&self, span: &Id, values: &Record<'_>) { + let mut spans = self.0.lock().expect("span capture should lock"); + values.record(&mut SpanFields(&mut spans[span.into_u64() as usize - 1])); + } + + fn record_follows_from(&self, _: &Id, _: &Id) {} + fn event(&self, _: &tracing::Event<'_>) {} + fn enter(&self, _: &Id) {} + fn exit(&self, _: &Id) {} + } + + async fn assert_rejection_observed( + app: Router, + request: Request, + expected_status: StatusCode, + expected_route: &str, + expected_code: &str, + ) -> (Value, String) { + let expected_request_id = request.headers().get("x-request-id").cloned(); + let path = request.uri().path().to_string(); + let capture = HttpSpanCapture::default(); + let response = app + .oneshot(request) + .with_subscriber(capture.clone()) + .await + .expect("rejected request should complete"); + + assert_eq!(response.status(), expected_status); + let request_id = response.headers()["x-request-id"] + .to_str() + .expect("request id should be a header string") + .to_string(); + assert!(!request_id.is_empty()); + assert_ne!(request_id, "unknown"); + if let Some(expected_request_id) = expected_request_id { + assert_eq!(response.headers()["x-request-id"], expected_request_id); + } + for name in ["x-api-version", "x-route-version"] { + assert_eq!(response.headers()[name], shared_contracts::api::API_VERSION); + } + assert!( + response.headers()["x-response-time-ms"] + .to_str() + .unwrap() + .parse::() + .is_ok() + ); + if expected_status == StatusCode::TOO_MANY_REQUESTS { + assert_eq!(response.headers()["retry-after"], "1"); + } + let payload = read_json_response(response).await; + assert_eq!(payload["error"]["code"], expected_code); + + let spans = capture.0.lock().expect("span capture should lock"); + assert_eq!( + spans.len(), + 1, + "each rejected request should have one HTTP span" + ); + let fields = &spans[0]; + assert_eq!(fields["request_id"], request_id); + assert_eq!(fields["http.route"], expected_route); + assert_eq!(fields["otel.name"], format!("GET {expected_route}")); + assert_eq!(fields["url.path"], path); + assert_eq!(fields["http.request.method"], "GET"); + assert_eq!(fields["status"], expected_status.as_u16().to_string()); + assert_eq!( + fields["http.response.status_code"], + expected_status.as_u16().to_string() + ); + assert_eq!( + fields["otel.status_code"], + if expected_status.is_server_error() { + "ERROR" + } else { + "OK" + } + ); + assert!(fields["latency_ms"].parse::().is_ok()); + (payload, request_id) + } + + #[tokio::test] + async fn auth_rejections_share_the_matched_route_template() { + let app = + build_router(AppState::new(AppConfig::default()).expect("state should build")); + for project_id in ["project-one", "project-two"] { + assert_rejection_observed( + app.clone(), + Request::builder() + .uri(format!( + "/api/editor/projects/{project_id}/agent-conversations" + )) + .header("x-request-id", format!("req-trace-{project_id}")) + .body(Body::empty()) + .expect("request should build"), + StatusCode::UNAUTHORIZED, + "/api/editor/projects/{project_id}/agent-conversations", + "UNAUTHORIZED", + ) + .await; + } + } + + #[tokio::test] + async fn backpressure_rejection_keeps_generated_context_and_headers() { + let config = AppConfig { + max_concurrent_requests: Some(1), + ..AppConfig::default() + }; + let state = AppState::new(config).expect("state should build"); + let (_, pool) = BackpressureState::from_ref(&state) + .request_permit_pool(HttpRequestPermitPoolKind::Default) + .expect("default request pool should exist"); + let _held_permit = pool + .try_acquire_owned() + .expect("pool should have one permit"); + + let (payload, request_id) = assert_rejection_observed( + build_router(state), + Request::builder() + .uri("/api/editor/projects/project-one/agent-conversations") + .body(Body::empty()) + .expect("request should build"), + StatusCode::TOO_MANY_REQUESTS, + "/api/editor/projects/{project_id}/agent-conversations", + "TOO_MANY_REQUESTS", + ) + .await; + assert_eq!(payload["meta"]["requestId"], request_id); + } + + #[tokio::test] + async fn unavailable_router_rejection_keeps_generated_context_and_headers() { + let (payload, request_id) = assert_rejection_observed( + build_spacetime_unavailable_router("test unavailable".to_string()), + Request::builder() + .uri("/api/auth/login-options") + .body(Body::empty()) + .expect("request should build"), + StatusCode::SERVICE_UNAVAILABLE, + "/api/*", + "SERVICE_UNAVAILABLE", + ) + .await; + assert_eq!(payload["meta"]["requestId"], request_id); + } + } + async fn seed_phone_user_with_password( state: &AppState, phone_number: &str, diff --git a/server-rs/crates/api-server/src/editor_project.rs b/server-rs/crates/api-server/src/editor_project.rs index b880b06e6..100d89b41 100644 --- a/server-rs/crates/api-server/src/editor_project.rs +++ b/server-rs/crates/api-server/src/editor_project.rs @@ -12,7 +12,7 @@ use std::{ use axum::{ Json, - extract::{Extension, Path, Query, State, rejection::JsonRejection}, + extract::{Extension, FromRef, Path, Query, State, rejection::JsonRejection}, http::{HeaderMap, HeaderValue, StatusCode}, }; use module_assets::{ @@ -122,7 +122,7 @@ use crate::{ platform_errors::map_oss_error, prompt::editor_scene::build_editor_scene_prompt, request_context::RequestContext, - state::AppState, + state::{AppState, EditorMediaStorageState, EditorProjectState}, work_author::{ORPHAN_WORK_AUTHOR_PUBLIC_USER_CODE, ORPHAN_WORK_OWNER_USER_ID}, }; @@ -1893,19 +1893,19 @@ pub struct EditorAssetPayload { } pub async fn load_recent_editor_project( - State(state): State, + State(state): State, Extension(request_context): Extension, Extension(authenticated): Extension, ) -> Result, AppError> { let owner_user_id = authenticated.claims().user_id().to_string(); let project = state - .spacetime_client() + .projects() .get_recent_editor_project(owner_user_id) .await .map_err(map_editor_project_error)?; let project = match project { Some(project) => Some(editor_project_payload_from_record( - repair_editor_project_record_inline_media(&state, project).await, + state.repair_project_media(project).await, )), None => None, }; @@ -1931,20 +1931,20 @@ pub async fn get_editor_generation_pricing( } pub async fn list_editor_projects( - State(state): State, + State(state): State, Extension(request_context): Extension, Extension(authenticated): Extension, ) -> Result, AppError> { let owner_user_id = authenticated.claims().user_id().to_string(); let project_records = state - .spacetime_client() + .projects() .list_editor_projects(owner_user_id) .await .map_err(map_editor_project_error)?; let mut projects = Vec::with_capacity(project_records.len()); for project in project_records { projects.push(editor_project_payload_from_record( - repair_editor_project_record_inline_media(&state, project).await, + state.repair_project_media(project).await, )); } @@ -2049,7 +2049,7 @@ fn parse_editor_timestamp_micros(value: &str) -> Option { } pub async fn create_editor_project( - State(state): State, + State(state): State, headers: HeaderMap, Extension(request_context): Extension, Extension(authenticated): Extension, @@ -2073,7 +2073,7 @@ pub async fn create_editor_project( }) .unwrap_or_else(|| build_prefixed_uuid_id(EDITOR_PROJECT_ID_PREFIX)); let project = state - .spacetime_client() + .projects() .create_editor_project(EditorProjectCreateRecordInput { project_id, owner_user_id, @@ -2092,20 +2092,20 @@ pub async fn create_editor_project( } pub async fn get_editor_project( - State(state): State, + State(state): State, Path(project_id): Path, Extension(request_context): Extension, Extension(authenticated): Extension, ) -> Result, AppError> { let project = state - .spacetime_client() + .projects() .get_editor_project(EditorProjectGetRecordInput { project_id, owner_user_id: authenticated.claims().user_id().to_string(), }) .await .map_err(map_editor_project_error)?; - let project = repair_editor_project_record_inline_media(&state, project).await; + let project = state.repair_project_media(project).await; Ok(json_success_body( Some(&request_context), @@ -2116,7 +2116,7 @@ pub async fn get_editor_project( } pub async fn save_editor_project_layout( - State(state): State, + State(state): State, Path(project_id): Path, Extension(request_context): Extension, Extension(authenticated): Extension, @@ -2128,7 +2128,7 @@ pub async fn save_editor_project_layout( let owner_user_id = authenticated.claims().user_id().to_string(); let updated_at_micros = current_utc_micros(); let ack = state - .spacetime_client() + .projects() .save_editor_project_layout_v2_ack(EditorProjectLayoutSaveV2RecordInput { project_id, owner_user_id, @@ -2152,14 +2152,14 @@ pub async fn save_editor_project_layout( } pub async fn rename_editor_project( - State(state): State, + State(state): State, Path(project_id): Path, Extension(request_context): Extension, Extension(authenticated): Extension, Json(payload): Json, ) -> Result, AppError> { let project = state - .spacetime_client() + .projects() .rename_editor_project(EditorProjectRenameRecordInput { project_id, owner_user_id: authenticated.claims().user_id().to_string(), @@ -2178,13 +2178,13 @@ pub async fn rename_editor_project( } pub async fn delete_editor_project( - State(state): State, + State(state): State, Path(project_id): Path, Extension(request_context): Extension, Extension(authenticated): Extension, ) -> Result, AppError> { let deleted_project_id = state - .spacetime_client() + .projects() .delete_editor_project(EditorProjectDeleteRecordInput { project_id, owner_user_id: authenticated.claims().user_id().to_string(), @@ -7754,7 +7754,7 @@ pub async fn snap_editor_image_to_pixel_art( } result = async move { let mut uploaded = upload_editor_generated_image_object_prepared( - &state, + &EditorMediaStorageState::from_ref(&state), owner_user_id.as_str(), persistence_identity.task_id.as_str(), prepared_upload, @@ -8946,8 +8946,8 @@ pub(crate) async fn extract_editor_ui_design_assets_for_owner( )) } -async fn repair_editor_project_record_inline_media( - state: &AppState, +pub(crate) async fn repair_editor_project_record_inline_media( + state: &EditorMediaStorageState, mut record: EditorProjectRecord, ) -> EditorProjectRecord { let owner_user_id = record.owner_user_id.clone(); @@ -8976,7 +8976,7 @@ async fn repair_editor_asset_library_inline_media( } async fn repair_editor_project_resource_inline_media( - state: &AppState, + state: &EditorMediaStorageState, owner_user_id: &str, resource: EditorProjectResourceRecord, ) -> EditorProjectResourceRecord { @@ -9058,7 +9058,7 @@ async fn repair_editor_asset_inline_media( } let repair = persist_editor_legacy_inline_image( - state, + &EditorMediaStorageState::from_ref(state), owner_user_id, asset.image_src.as_str(), asset.task_id.as_deref().unwrap_or(asset.asset_id.as_str()), @@ -9112,7 +9112,7 @@ async fn repair_editor_asset_inline_media( } async fn persist_editor_legacy_inline_image( - state: &AppState, + state: &EditorMediaStorageState, owner_user_id: &str, image_src: &str, task_id: &str, @@ -9133,11 +9133,14 @@ async fn persist_editor_legacy_inline_image( mime_type: decoded.format.mime_type, extension: decoded.format.extension, }; - persist_editor_generated_image( + persist_editor_generated_image_data( state, owner_user_id, task_id, - &image, + GeneratedImageAssetDataUrl { + format: normalize_generated_image_asset_mime(image.mime_type.as_str()), + bytes: image.bytes, + }, prompt, actual_prompt, asset_kind.unwrap_or(EDITOR_LEGACY_INLINE_IMAGE_ASSET_KIND), @@ -10407,7 +10410,11 @@ pub(crate) async fn complete_editor_canvas_generation( }) .await .map_err(map_editor_project_error)?; - let project = repair_editor_project_record_inline_media(state, project).await; + let project = repair_editor_project_record_inline_media( + &EditorMediaStorageState::from_ref(state), + project, + ) + .await; let expected_revision = project.canvas.revision; let viewport = project.viewport.clone(); let project_payload = editor_project_payload_from_record(project); @@ -10683,7 +10690,11 @@ pub(crate) async fn prepare_editor_canvas_generation_layout( }) .await .map_err(map_editor_project_error)?; - let project = repair_editor_project_record_inline_media(state, project).await; + let project = repair_editor_project_record_inline_media( + &EditorMediaStorageState::from_ref(state), + project, + ) + .await; let expected_revision = project.canvas.revision; let viewport = project.viewport.clone(); let project_payload = editor_project_payload_from_record(project); @@ -10776,7 +10787,11 @@ async fn prepare_editor_canvas_background_removal_layout( }) .await .map_err(map_editor_project_error)?; - let project = repair_editor_project_record_inline_media(state, project).await; + let project = repair_editor_project_record_inline_media( + &EditorMediaStorageState::from_ref(state), + project, + ) + .await; let expected_revision = project.canvas.revision; let viewport = project.viewport.clone(); let project_payload = editor_project_payload_from_record(project); @@ -12123,7 +12138,7 @@ pub(crate) async fn persist_editor_generated_image( provider: &str, ) -> Result { persist_editor_generated_image_data( - state, + &EditorMediaStorageState::from_ref(state), owner_user_id, task_id, GeneratedImageAssetDataUrl { @@ -12162,7 +12177,7 @@ pub(crate) async fn prepare_editor_generated_image( })) })?; let uploaded = upload_editor_generated_image_object_data( - state, + &EditorMediaStorageState::from_ref(state), caller.owner_user_id.as_str(), task_id, GeneratedImageAssetDataUrl { @@ -12207,7 +12222,7 @@ async fn persist_editor_generated_image_owned( bytes: image.bytes, }; persist_editor_generated_image_data( - state, + &EditorMediaStorageState::from_ref(state), owner_user_id, task_id, image_data, @@ -12223,7 +12238,7 @@ async fn persist_editor_generated_image_owned( } async fn persist_editor_generated_image_data( - state: &AppState, + state: &EditorMediaStorageState, owner_user_id: &str, task_id: &str, image: GeneratedImageAssetDataUrl, @@ -12270,7 +12285,7 @@ async fn persist_editor_generated_image_data( #[allow(clippy::too_many_arguments)] async fn upload_editor_generated_image_object_data( - state: &AppState, + state: &EditorMediaStorageState, owner_user_id: &str, task_id: &str, image: GeneratedImageAssetDataUrl, @@ -12306,7 +12321,7 @@ async fn upload_editor_generated_image_object_data( } async fn upload_editor_generated_image_object_prepared( - state: &AppState, + state: &EditorMediaStorageState, owner_user_id: &str, task_id: &str, prepared: GeneratedImageAssetPreparedPut, @@ -13679,6 +13694,9 @@ pub(crate) fn current_utc_micros() -> i64 { i64::try_from(duration.as_micros()).expect("current unix micros should fit in i64") } +#[cfg(test)] +mod metadata_tests; + #[cfg(test)] mod tests { use super::*; diff --git a/server-rs/crates/api-server/src/editor_project/metadata_tests.rs b/server-rs/crates/api-server/src/editor_project/metadata_tests.rs new file mode 100644 index 000000000..cf3b2b04f --- /dev/null +++ b/server-rs/crates/api-server/src/editor_project/metadata_tests.rs @@ -0,0 +1,364 @@ +use super::*; +use crate::{ + request_context::attach_request_context, + state::project_metadata::{EditorProjectMediaRepair, EditorProjectRepository}, +}; +use axum::{ + Router, + body::{Body, to_bytes}, + http::Request, + middleware, + routing::{get, patch}, +}; +use futures_util::future::BoxFuture; +use platform_auth::{AccessTokenClaims, AuthProvider, BindingStatus}; +use spacetime_client::EditorProjectLayoutSaveV2AckRecord; +use std::sync::Mutex; +use tower::ServiceExt; + +#[derive(Default)] +struct RecordingProjects { + calls: Mutex>, + error: Option<&'static str>, +} + +impl RecordingProjects { + fn finish( + &self, + operation: &'static str, + input: Value, + result: T, + ) -> BoxFuture<'_, Result> { + self.calls.lock().unwrap().push((operation, input)); + Box::pin(async move { + match self.error { + Some(message) => Err(SpacetimeClientError::Procedure(message.to_string())), + None => Ok(result), + } + }) + } +} + +fn project_record(project_id: &str, owner: &str, title: &str) -> EditorProjectRecord { + let viewport = EditorCanvasViewportRecord { + x: 1.0, + y: 2.0, + scale: 1.0, + }; + EditorProjectRecord { + project_id: project_id.to_string(), + owner_user_id: owner.to_string(), + title: title.to_string(), + canvas: EditorCanvasRecord { + canvas_id: "canvas-fixture".to_string(), + project_id: project_id.to_string(), + title: title.to_string(), + viewport: viewport.clone(), + layers: json!([]), + revision: 7, + layout_storage_version: 2, + background_color: None, + created_at: "0.000000Z".to_string(), + updated_at: "0.000000Z".to_string(), + }, + viewport, + layers: json!([]), + resources: vec![], + created_at: "0.000000Z".to_string(), + updated_at: "0.000000Z".to_string(), + } +} + +impl EditorProjectRepository for RecordingProjects { + fn get_recent_editor_project( + &self, + owner_user_id: String, + ) -> BoxFuture<'_, Result, SpacetimeClientError>> { + let record = project_record("project-fixture", &owner_user_id, "原名称"); + self.finish("recent", json!({"owner": owner_user_id}), Some(record)) + } + + fn list_editor_projects( + &self, + owner_user_id: String, + ) -> BoxFuture<'_, Result, SpacetimeClientError>> { + let record = project_record("project-fixture", &owner_user_id, "原名称"); + self.finish("list", json!({"owner": owner_user_id}), vec![record]) + } + + fn get_editor_project( + &self, + input: EditorProjectGetRecordInput, + ) -> BoxFuture<'_, Result> { + let record = project_record(&input.project_id, &input.owner_user_id, "原名称"); + self.finish( + "get", + json!({"owner": input.owner_user_id, "project": input.project_id}), + record, + ) + } + + fn create_editor_project( + &self, + input: EditorProjectCreateRecordInput, + ) -> BoxFuture<'_, Result> { + let record = project_record(&input.project_id, &input.owner_user_id, &input.title); + self.finish("create", json!({"owner": input.owner_user_id, "project": input.project_id, "title": input.title, "at": input.now_micros}), record) + } + + fn rename_editor_project( + &self, + input: EditorProjectRenameRecordInput, + ) -> BoxFuture<'_, Result> { + let record = project_record(&input.project_id, &input.owner_user_id, &input.title); + self.finish("rename", json!({"owner": input.owner_user_id, "project": input.project_id, "title": input.title, "at": input.updated_at_micros}), record) + } + + fn delete_editor_project( + &self, + input: EditorProjectDeleteRecordInput, + ) -> BoxFuture<'_, Result> { + self.finish( + "delete", + json!({"owner": input.owner_user_id, "project": input.project_id}), + input.project_id, + ) + } + + fn save_editor_project_layout_v2_ack( + &self, + input: EditorProjectLayoutSaveV2RecordInput, + ) -> BoxFuture<'_, Result> { + let ack = EditorProjectLayoutSaveV2AckRecord { + project_id: input.project_id.clone(), + canvas_id: "canvas-fixture".to_string(), + revision: input.expected_revision + 1, + updated_at: "1.000000Z".to_string(), + updated_at_micros: input.updated_at_micros, + }; + self.finish("layout", json!({"owner": input.owner_user_id, "project": input.project_id, "revision": input.expected_revision, "layers": input.layers_json, "viewport": input.viewport}), ack) + } +} + +#[derive(Default)] +struct RecordingMediaRepair { + calls: Mutex>, +} + +impl EditorProjectMediaRepair for RecordingMediaRepair { + fn repair_project( + &self, + mut project: EditorProjectRecord, + ) -> BoxFuture<'_, EditorProjectRecord> { + self.calls + .lock() + .unwrap() + .push((project.project_id.clone(), project.owner_user_id.clone())); + Box::pin(async move { + project.updated_at = "media-repaired".to_string(); + project + }) + } +} + +fn authenticated_owner() -> AuthenticatedAccessToken { + AuthenticatedAccessToken::new(AccessTokenClaims { + iss: "fixture".to_string(), + sub: "owner-from-token".to_string(), + sid: "fixture-session".to_string(), + provider: AuthProvider::Password, + roles: vec!["user".to_string()], + ver: 1, + phone_verified: true, + binding_status: BindingStatus::Active, + display_name: None, + device: None, + iat: 1, + exp: u64::MAX, + }) +} + +fn metadata_router(projects: Arc, media: Arc) -> Router { + // 直接以窄状态组装实际 handler，不构造 AppConfig、连接数据库或初始化其它平台。 + Router::new() + .route("/projects/recent", get(load_recent_editor_project)) + .route( + "/projects", + get(list_editor_projects).post(create_editor_project), + ) + .route( + "/projects/{project_id}", + get(get_editor_project) + .patch(save_editor_project_layout) + .delete(delete_editor_project), + ) + .route( + "/projects/{project_id}/metadata", + patch(rename_editor_project), + ) + .layer(Extension(authenticated_owner())) + .layer(middleware::from_fn(attach_request_context)) + .with_state(EditorProjectState::new(projects, media)) +} + +async fn send(router: &Router, method: &str, path: &str, body: Value) -> (StatusCode, Value) { + let response = router + .clone() + .oneshot( + Request::builder() + .method(method) + .uri(path) + .header("content-type", "application/json") + .header("idempotency-key", "metadata-fixture") + .body(Body::from(body.to_string())) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let body = to_bytes(response.into_body(), 65_536).await.unwrap(); + (status, serde_json::from_slice(&body).unwrap()) +} + +#[tokio::test] +async fn narrow_project_reads_preserve_owner_and_use_media_repair_results() { + let projects = Arc::new(RecordingProjects::default()); + let media = Arc::new(RecordingMediaRepair::default()); + let router = metadata_router(projects.clone(), media.clone()); + for (path, result_path) in [ + ("/projects/recent", "/project"), + ("/projects", "/projects/0"), + ("/projects/project-fixture", "/project"), + ] { + let (status, body) = send(&router, "GET", path, Value::Null).await; + assert_eq!(status, StatusCode::OK); + let result = body.pointer(result_path).unwrap(); + assert_eq!(result["projectId"], "project-fixture"); + assert_eq!(result["updatedAt"], "media-repaired"); + } + let calls = projects.calls.lock().unwrap(); + assert_eq!( + calls.iter().map(|call| call.0).collect::>(), + ["recent", "list", "get"] + ); + assert!( + calls + .iter() + .all(|call| call.1["owner"] == "owner-from-token") + ); + assert_eq!( + *media.calls.lock().unwrap(), + vec![ + ( + "project-fixture".to_string(), + "owner-from-token".to_string() + ); + 3 + ] + ); +} + +#[tokio::test] +async fn narrow_project_writes_preserve_owner_idempotency_and_revision() { + let projects = Arc::new(RecordingProjects::default()); + let media = Arc::new(RecordingMediaRepair::default()); + let router = metadata_router(projects.clone(), media.clone()); + let (status, created) = send( + &router, + "POST", + "/projects", + json!({"title": " 测试项目 ", "ownerUserId": "untrusted-owner"}), + ) + .await; + assert_eq!(status, StatusCode::OK); + let project_id = editor_idempotent_create_id( + EDITOR_PROJECT_ID_PREFIX, + "owner-from-token", + "internal-editor-project", + "metadata-fixture", + ); + assert_eq!(created["project"]["projectId"], project_id); + assert_eq!(created["project"]["title"], "测试项目"); + + let (status, renamed) = send( + &router, + "PATCH", + "/projects/project-fixture/metadata", + json!({"title": "新名称"}), + ) + .await; + assert_eq!(status, StatusCode::OK); + assert_eq!(renamed["project"]["title"], "新名称"); + + let (status, saved) = send( + &router, + "PATCH", + "/projects/project-fixture", + json!({"viewport": {"x": 10, "y": 20, "scale": 2}, "layers": [], "expectedRevision": 7}), + ) + .await; + assert_eq!(status, StatusCode::OK); + assert_eq!(saved["revision"], 8); + assert_eq!(saved["projectId"], "project-fixture"); + + let (status, deleted) = send(&router, "DELETE", "/projects/project-fixture", Value::Null).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(deleted["deletedProjectId"], "project-fixture"); + let calls = projects.calls.lock().unwrap(); + assert_eq!( + calls.iter().map(|call| call.0).collect::>(), + ["create", "rename", "layout", "delete"] + ); + assert!( + calls + .iter() + .all(|call| call.1["owner"] == "owner-from-token") + ); + assert_eq!(calls[0].1["project"], project_id); + assert_eq!(calls[2].1["revision"], 7); + assert_eq!(calls[2].1["layers"], "[]"); + assert_eq!( + calls[2].1["viewport"], + json!({"x": 10.0, "y": 20.0, "scale": 2.0}) + ); + assert!(media.calls.lock().unwrap().is_empty()); +} + +#[tokio::test] +async fn narrow_project_handlers_propagate_facade_errors_without_media_side_effects() { + for (message, expected_status) in [ + ("项目不存在", StatusCode::NOT_FOUND), + ("无权访问该项目", StatusCode::FORBIDDEN), + ("项目版本冲突", StatusCode::CONFLICT), + ] { + let projects = Arc::new(RecordingProjects { + error: Some(message), + ..Default::default() + }); + let media = Arc::new(RecordingMediaRepair::default()); + let router = metadata_router(projects.clone(), media.clone()); + for (method, path, body) in [ + ("GET", "/projects/recent", Value::Null), + ("GET", "/projects", Value::Null), + ("GET", "/projects/project-fixture", Value::Null), + ("POST", "/projects", json!({"title": "测试"})), + ( + "PATCH", + "/projects/project-fixture/metadata", + json!({"title": "测试"}), + ), + ( + "PATCH", + "/projects/project-fixture", + json!({"viewport": {"x": 0, "y": 0, "scale": 1}, "layers": [], "expectedRevision": 7}), + ), + ("DELETE", "/projects/project-fixture", Value::Null), + ] { + let (status, body) = send(&router, method, path, body).await; + assert_eq!(status, expected_status, "{method} {path}"); + assert!(body.to_string().contains(message)); + } + assert_eq!(projects.calls.lock().unwrap().len(), 7); + assert!(media.calls.lock().unwrap().is_empty()); + } +} diff --git a/server-rs/crates/api-server/src/external_api_auth.rs b/server-rs/crates/api-server/src/external_api_auth.rs index 4ce511d76..307203bb1 100644 --- a/server-rs/crates/api-server/src/external_api_auth.rs +++ b/server-rs/crates/api-server/src/external_api_auth.rs @@ -16,7 +16,7 @@ use crate::{ external_api_keys::{hash_external_api_key, map_external_api_key_error}, http_error::AppError, request_context::RequestContext, - state::AppState, + state::ExternalApiAuthState, }; #[derive(Clone, Debug)] @@ -50,7 +50,7 @@ impl ExternalApiPrincipal { } pub async fn require_external_api_key( - State(state): State, + State(state): State, mut request: Request, next: Next, ) -> Result { @@ -61,7 +61,7 @@ pub async fn require_external_api_key( .unwrap_or_else(|| "unknown".to_string()); let raw_key = extract_external_api_bearer(request.headers())?; let key = state - .spacetime_client() + .authenticator() .authenticate_external_api_key(ExternalApiKeyAuthenticateRecordInput { key_hash: hash_external_api_key(raw_key.as_str()), used_at_micros: current_utc_micros(), @@ -88,7 +88,7 @@ pub async fn require_external_api_key( } pub async fn require_external_mcp_api_key( - State(state): State, + State(state): State, request: Request, next: Next, ) -> Result { @@ -171,6 +171,184 @@ fn extract_external_api_bearer(headers: &HeaderMap) -> Result #[cfg(test)] mod tests { use super::*; + use crate::state::external_api_auth::ExternalApiKeyAuthenticator; + use axum::{ + Router, + body::{Body, to_bytes}, + extract::Extension, + middleware, + routing::get, + }; + use futures_util::future::BoxFuture; + use spacetime_client::{ExternalApiKeyRecord, SpacetimeClientError}; + use std::sync::{ + Arc, Mutex, + atomic::{AtomicUsize, Ordering}, + }; + use tower::ServiceExt; + + struct RecordingAuthenticator { + requests: Mutex>, + error: Option<&'static str>, + } + + impl ExternalApiKeyAuthenticator for RecordingAuthenticator { + fn authenticate_external_api_key( + &self, + input: ExternalApiKeyAuthenticateRecordInput, + ) -> BoxFuture<'_, Result> { + self.requests.lock().unwrap().push(input); + Box::pin(async move { + if let Some(error) = self.error { + return Err(SpacetimeClientError::Procedure(error.to_string())); + } + Ok(ExternalApiKeyRecord { + key_id: "key-from-store".to_string(), + owner_user_id: "owner-from-store".to_string(), + name: "测试密钥".to_string(), + key_prefix: "tnr_sk_fixture".to_string(), + scopes: vec!["editor:project".to_string()], + created_at: "0.000000Z".to_string(), + last_used_at: None, + revoked_at: None, + updated_at: "0.000000Z".to_string(), + }) + }) + } + } + + fn auth_test_router( + dependency: Arc, + entered: Arc, + mcp: bool, + ) -> Router { + let state = ExternalApiAuthState::new(dependency); + let router = Router::new().route("/protected", get(move |Extension(principal): Extension| { + let entered = entered.clone(); + async move { + entered.fetch_add(1, Ordering::Relaxed); + axum::Json(json!({"owner": principal.owner_user_id(), "projectScope": principal.has_scope("editor:project")})) + } + })); + if mcp { + router.layer(middleware::from_fn_with_state( + state, + require_external_mcp_api_key, + )) + } else { + router.layer(middleware::from_fn_with_state( + state, + require_external_api_key, + )) + } + } + + #[tokio::test] + async fn narrow_external_auth_forwards_store_identity_and_only_hashes_credentials() { + let dependency = Arc::new(RecordingAuthenticator { + requests: Mutex::default(), + error: None, + }); + let entered = Arc::new(AtomicUsize::new(0)); + let response = auth_test_router(dependency.clone(), entered.clone(), false) + .oneshot( + Request::builder() + .uri("/protected") + .header(AUTHORIZATION, "Bearer tnr_sk_fixture_secret") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let principal = response.extensions().get::().unwrap(); + assert_eq!(principal.owner_user_id(), "owner-from-store"); + assert_eq!(principal.key_id(), "key-from-store"); + let body: serde_json::Value = + serde_json::from_slice(&to_bytes(response.into_body(), 1024).await.unwrap()).unwrap(); + assert_eq!( + body, + json!({"owner": "owner-from-store", "projectScope": true}) + ); + assert_eq!(entered.load(Ordering::Relaxed), 1); + let requests = dependency.requests.lock().unwrap(); + assert_eq!(requests.len(), 1); + assert_eq!( + requests[0].key_hash, + hash_external_api_key("tnr_sk_fixture_secret") + ); + assert!(requests[0].used_at_micros > 0); + } + + #[tokio::test] + async fn narrow_external_auth_preserves_failure_mapping_and_mcp_guide() { + for (message, expected_status) in [ + ("API Key 不存在", StatusCode::UNAUTHORIZED), + ("无权使用此密钥", StatusCode::FORBIDDEN), + ("校验请求失败", StatusCode::BAD_REQUEST), + ] { + for mcp in [false, true] { + let dependency = Arc::new(RecordingAuthenticator { + requests: Mutex::default(), + error: Some(message), + }); + let entered = Arc::new(AtomicUsize::new(0)); + let response = auth_test_router(dependency.clone(), entered.clone(), mcp) + .oneshot( + Request::builder() + .uri("/protected") + .header(AUTHORIZATION, "Bearer invalid-fixture-key") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), expected_status); + assert!( + response + .extensions() + .get::() + .is_none() + ); + assert_eq!(entered.load(Ordering::Relaxed), 0); + assert_eq!(dependency.requests.lock().unwrap().len(), 1); + let body = String::from_utf8( + to_bytes(response.into_body(), 16_384) + .await + .unwrap() + .to_vec(), + ) + .unwrap(); + if mcp && expected_status == StatusCode::UNAUTHORIZED { + assert!(body.contains("MCP_AUTHENTICATION_REQUIRED")); + assert!(!body.contains(message)); + } else { + assert!(body.contains(message)); + } + } + } + } + + #[tokio::test] + async fn narrow_external_auth_rejects_missing_credentials_before_store_access() { + let dependency = Arc::new(RecordingAuthenticator { + requests: Mutex::default(), + error: None, + }); + let entered = Arc::new(AtomicUsize::new(0)); + let response = auth_test_router(dependency.clone(), entered.clone(), false) + .oneshot( + Request::builder() + .uri("/protected") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + assert!(dependency.requests.lock().unwrap().is_empty()); + assert_eq!(entered.load(Ordering::Relaxed), 0); + } #[test] fn mcp_authentication_guide_replaces_sensitive_key_diagnostics() { diff --git a/server-rs/crates/api-server/src/modules/admin.rs b/server-rs/crates/api-server/src/modules/admin.rs index c8fdc40cc..77925d2e6 100644 --- a/server-rs/crates/api-server/src/modules/admin.rs +++ b/server-rs/crates/api-server/src/modules/admin.rs @@ -37,283 +37,349 @@ use crate::{ }; pub fn router(state: AppState) -> Router { - Router::new() - .route( + let auth = middleware::from_fn_with_state(state, require_admin_auth); + let protected_routes = [ + ( "/admin/api/agc-models", get(crate::agc_models::admin_get_agc_models) - .put(crate::agc_models::admin_save_agc_models) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route("/admin/api/login", axum::routing::post(admin_login)) - .route( + .put(crate::agc_models::admin_save_agc_models), + ), + ( "/admin/api/accounts", - get(admin_list_accounts) - .post(admin_create_account) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_list_accounts).post(admin_create_account), + ), + ( "/admin/api/accounts/{account_id}", - axum::routing::put(admin_update_account).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( - "/admin/api/me", - get(admin_me).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( - "/admin/api/overview", - get(admin_overview).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( - "/admin/api/dashboard", - get(admin_dashboard).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + axum::routing::put(admin_update_account), + ), + ("/admin/api/me", get(admin_me)), + ("/admin/api/overview", get(admin_overview)), + ("/admin/api/dashboard", get(admin_dashboard)), + ( "/admin/api/debug/http", - axum::routing::post(admin_debug_http).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + axum::routing::post(admin_debug_http), + ), + ( "/admin/api/tracking/events", - get(admin_list_tracking_events).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_list_tracking_events), + ), + ( "/admin/api/tracking/event-keys", - get(admin_list_tracking_event_keys).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_list_tracking_event_keys), + ), + ( "/admin/api/database/tables", - get(admin_list_database_tables).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_list_database_tables), + ), + ( "/admin/api/database/tables/{table_name}/rows", - get(admin_list_database_table_rows).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_list_database_table_rows), + ), + ( "/admin/api/external-api-keys", - get(admin_list_external_api_keys).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_list_external_api_keys), + ), + ( "/admin/api/feature-gates", - get(admin_get_feature_gate_config) - .put(admin_upsert_feature_gate_config) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_get_feature_gate_config).put(admin_upsert_feature_gate_config), + ), + ( "/admin/api/editor-generation-pricing", - get(admin_get_editor_generation_pricing) - .post(admin_upsert_editor_generation_pricing) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( - "/admin/api/editor-assets", - get(admin_list_editor_assets).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( - "/admin/api/assets/read-url", - get(admin_get_asset_read_url).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_get_editor_generation_pricing).post(admin_upsert_editor_generation_pricing), + ), + ("/admin/api/editor-assets", get(admin_list_editor_assets)), + ("/admin/api/assets/read-url", get(admin_get_asset_read_url)), + ( "/admin/api/editor-showcase/assets", - get(admin_list_editor_showcase_assets).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_list_editor_showcase_assets), + ), + ( "/admin/api/editor-showcase/assets/review", - post(admin_review_editor_showcase_asset).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + post(admin_review_editor_showcase_asset), + ), + ( "/admin/api/editor-showcase/assets/display", - post(admin_update_editor_showcase_asset_display).route_layer( - middleware::from_fn_with_state(state.clone(), require_admin_auth), - ), - ) - .route( + post(admin_update_editor_showcase_asset_display), + ), + ( "/admin/api/editor-showcase/campaign", - get(admin_get_editor_showcase_campaign) - .post(admin_upsert_editor_showcase_campaign) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_get_editor_showcase_campaign).post(admin_upsert_editor_showcase_campaign), + ), + ( "/admin/api/editor-showcase/campaign/image-upload-ticket", - post(admin_create_editor_showcase_campaign_image_upload_ticket).route_layer( - middleware::from_fn_with_state(state.clone(), require_admin_auth), - ), - ) - .route( + post(admin_create_editor_showcase_campaign_image_upload_ticket), + ), + ( "/admin/api/editor-showcase/campaign/image-upload-confirm", - post(admin_confirm_editor_showcase_campaign_image_upload).route_layer( - middleware::from_fn_with_state(state.clone(), require_admin_auth), - ), - ) - .route( + post(admin_confirm_editor_showcase_campaign_image_upload), + ), + ( "/admin/api/profile/redeem-codes", - get(admin_list_profile_redeem_codes) - .post(admin_upsert_profile_redeem_code) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_list_profile_redeem_codes).post(admin_upsert_profile_redeem_code), + ), + ( "/admin/api/profile/redeem-codes/disable", - axum::routing::post(admin_disable_profile_redeem_code).route_layer( - middleware::from_fn_with_state(state.clone(), require_admin_auth), - ), - ) - .route( + axum::routing::post(admin_disable_profile_redeem_code), + ), + ( "/admin/api/profile/invite-codes", - get(admin_list_profile_invite_codes) - .post(admin_upsert_profile_invite_code) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_list_profile_invite_codes).post(admin_upsert_profile_invite_code), + ), + ( "/admin/api/profile/tasks", - get(admin_list_profile_task_configs) - .post(admin_upsert_profile_task_config) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_list_profile_task_configs).post(admin_upsert_profile_task_config), + ), + ( "/admin/api/profile/tasks/disable", - axum::routing::post(admin_disable_profile_task_config).route_layer( - middleware::from_fn_with_state(state.clone(), require_admin_auth), - ), - ) - .route( + axum::routing::post(admin_disable_profile_task_config), + ), + ( "/admin/api/profile/wallet-config", - get(admin_get_profile_wallet_config) - .post(admin_upsert_profile_wallet_config) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_get_profile_wallet_config).post(admin_upsert_profile_wallet_config), + ), + ( "/admin/api/profile/recharge-products", - get(admin_list_profile_recharge_products) - .post(admin_upsert_profile_recharge_product) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_list_profile_recharge_products).post(admin_upsert_profile_recharge_product), + ), + ( "/admin/api/profile/recharge-orders", - get(admin_list_recharge_orders).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_list_recharge_orders), + ), + ( "/admin/api/profile/recharge-refunds/preview", - post(admin_preview_recharge_refund).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + post(admin_preview_recharge_refund), + ), + ( "/admin/api/profile/recharge-refunds/execute", - post(admin_execute_recharge_refund).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + post(admin_execute_recharge_refund), + ), + ( "/admin/api/profile/recharge-refunds/register", - post(admin_register_recharge_refund).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + post(admin_register_recharge_refund), + ), + ( "/admin/api/profile/recharge-refunds/manual-review/resolve", - post(admin_resolve_recharge_refund_manual_review).route_layer( - middleware::from_fn_with_state(state.clone(), require_admin_auth), - ), - ) - .route( + post(admin_resolve_recharge_refund_manual_review), + ), + ( "/admin/api/profile/users/detail", - get(admin_get_user_detail).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + get(admin_get_user_detail), + ), + ( "/admin/api/profile/users/reconcile-consumption", - post(admin_reconcile_user_consumption).route_layer(middleware::from_fn_with_state( - state.clone(), - require_admin_auth, - )), - ) - .route( + post(admin_reconcile_user_consumption), + ), + ( "/admin/api/profile/users/initialize-consumption-projections", - post(admin_initialize_user_consumption_projections).route_layer( - middleware::from_fn_with_state(state.clone(), require_admin_auth), - ), - ) - .route( + post(admin_initialize_user_consumption_projections), + ), + ( "/admin/api/profile/wallet-restriction", - post(admin_update_wallet_restriction) - .route_layer(middleware::from_fn_with_state(state, require_admin_auth)), - ) + post(admin_update_wallet_restriction), + ), + ]; + // 在方法层统一鉴权，未支持的方法仍由 Axum 返回 405。 + let protected_router = protected_routes + .into_iter() + .fold(Router::new(), |router, (path, methods)| { + router.route(path, methods.route_layer(auth.clone())) + }); + + Router::new() + .route("/admin/api/login", axum::routing::post(admin_login)) + .merge(protected_router) +} + +#[cfg(test)] +mod route_contract_tests { + use axum::{ + body::Body, + http::{Request, StatusCode, header}, + middleware, + }; + use http_body_util::BodyExt; + use tower::ServiceExt; + + use super::router; + use crate::{config::AppConfig, request_context::attach_request_context, state::AppState}; + + const PROTECTED_ROUTES: &[(&str, &[&str])] = &[ + ("/admin/api/agc-models", &["GET", "PUT"]), + ("/admin/api/accounts", &["GET", "POST"]), + ("/admin/api/accounts/{account_id}", &["PUT"]), + ("/admin/api/me", &["GET"]), + ("/admin/api/overview", &["GET"]), + ("/admin/api/dashboard", &["GET"]), + ("/admin/api/debug/http", &["POST"]), + ("/admin/api/tracking/events", &["GET"]), + ("/admin/api/tracking/event-keys", &["GET"]), + ("/admin/api/database/tables", &["GET"]), + ("/admin/api/database/tables/{table_name}/rows", &["GET"]), + ("/admin/api/external-api-keys", &["GET"]), + ("/admin/api/feature-gates", &["GET", "PUT"]), + ("/admin/api/editor-generation-pricing", &["GET", "POST"]), + ("/admin/api/editor-assets", &["GET"]), + ("/admin/api/assets/read-url", &["GET"]), + ("/admin/api/editor-showcase/assets", &["GET"]), + ("/admin/api/editor-showcase/assets/review", &["POST"]), + ("/admin/api/editor-showcase/assets/display", &["POST"]), + ("/admin/api/editor-showcase/campaign", &["GET", "POST"]), + ( + "/admin/api/editor-showcase/campaign/image-upload-ticket", + &["POST"], + ), + ( + "/admin/api/editor-showcase/campaign/image-upload-confirm", + &["POST"], + ), + ("/admin/api/profile/redeem-codes", &["GET", "POST"]), + ("/admin/api/profile/redeem-codes/disable", &["POST"]), + ("/admin/api/profile/invite-codes", &["GET", "POST"]), + ("/admin/api/profile/tasks", &["GET", "POST"]), + ("/admin/api/profile/tasks/disable", &["POST"]), + ("/admin/api/profile/wallet-config", &["GET", "POST"]), + ("/admin/api/profile/recharge-products", &["GET", "POST"]), + ("/admin/api/profile/recharge-orders", &["GET"]), + ("/admin/api/profile/recharge-refunds/preview", &["POST"]), + ("/admin/api/profile/recharge-refunds/execute", &["POST"]), + ("/admin/api/profile/recharge-refunds/register", &["POST"]), + ( + "/admin/api/profile/recharge-refunds/manual-review/resolve", + &["POST"], + ), + ("/admin/api/profile/users/detail", &["GET"]), + ("/admin/api/profile/users/reconcile-consumption", &["POST"]), + ( + "/admin/api/profile/users/initialize-consumption-projections", + &["POST"], + ), + ("/admin/api/profile/wallet-restriction", &["POST"]), + ]; + + fn app() -> axum::Router { + let state = AppState::new(AppConfig { + admin_username: Some("root".to_string()), + admin_password: Some("secret123".to_string()), + ..AppConfig::default() + }) + .expect("state should build"); + router(state.clone()) + .layer(middleware::from_fn(attach_request_context)) + .with_state(state) + } + + #[tokio::test] + async fn protected_route_matrix_keeps_auth_methods_and_head_behavior() { + let app = app(); + for (template, supported) in PROTECTED_ROUTES { + let path = template + .split('/') + .map(|segment| { + if segment.starts_with('{') { + "fixture" + } else { + segment + } + }) + .collect::>() + .join("/"); + let mut allowed = supported.to_vec(); + if allowed.contains(&"GET") { + allowed.push("HEAD"); + } + allowed.sort_unstable(); + for credential in [None, Some("Bearer invalid-token")] { + for method in [ + "GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "TRACE", + ] { + let mut request = Request::builder().method(method).uri(&path); + if let Some(credential) = credential { + request = request.header(header::AUTHORIZATION, credential); + } + let response = app + .clone() + .oneshot(request.body(Body::empty()).expect("request should build")) + .await + .expect("request should complete"); + let expected = if allowed.contains(&method) { + StatusCode::UNAUTHORIZED + } else { + StatusCode::METHOD_NOT_ALLOWED + }; + assert_eq!( + response.status(), + expected, + "{method} {path}, credential={credential:?}" + ); + if expected == StatusCode::METHOD_NOT_ALLOWED { + let mut actual = response.headers()[header::ALLOW] + .to_str() + .expect("Allow should be ASCII") + .split(',') + .map(str::trim) + .collect::>(); + actual.sort_unstable(); + assert_eq!(actual, allowed, "Allow for {method} {path}"); + } + if method == "HEAD" { + assert!( + response + .into_body() + .collect() + .await + .expect("HEAD body should collect") + .to_bytes() + .is_empty(), + "HEAD {path} must not return a body" + ); + } + } + } + } + } + + #[tokio::test] + async fn unknown_paths_remain_not_found_before_authentication() { + for credential in [None, Some("Bearer invalid-token")] { + let mut request = Request::builder().uri("/admin/api/missing/unknown/route"); + if let Some(credential) = credential { + request = request.header(header::AUTHORIZATION, credential); + } + let response = app() + .oneshot(request.body(Body::empty()).expect("request should build")) + .await + .expect("request should complete"); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + } + } + + #[tokio::test] + async fn login_remains_public_with_invalid_authorization_header() { + let response = app() + .oneshot( + Request::builder() + .method("POST") + .uri("/admin/api/login") + .header(header::AUTHORIZATION, "Bearer invalid-token") + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(r#"{"username":"root","password":"secret123"}"#)) + .expect("login request should build"), + ) + .await + .expect("login should complete"); + assert_eq!(response.status(), StatusCode::OK); + let body = response + .into_body() + .collect() + .await + .expect("login body should collect") + .to_bytes(); + let payload: serde_json::Value = + serde_json::from_slice(&body).expect("login should return JSON"); + assert!( + payload["token"] + .as_str() + .is_some_and(|token| !token.is_empty()) + ); + } } diff --git a/server-rs/crates/api-server/src/modules/editor_project.rs b/server-rs/crates/api-server/src/modules/editor_project.rs index 1abcb5095..8ff87a7ee 100644 --- a/server-rs/crates/api-server/src/modules/editor_project.rs +++ b/server-rs/crates/api-server/src/modules/editor_project.rs @@ -51,6 +51,132 @@ async fn add_showcase_private_response_headers(request: Request, next: Next) -> } pub fn router(state: AppState) -> Router { + let auth = middleware::from_fn_with_state(state, require_bearer_auth); + let protected_routes = [ + ( + "/api/editor/projects/recent", + get(load_recent_editor_project), + ), + ( + "/api/editor/projects", + get(list_editor_projects).post(create_editor_project), + ), + ( + "/api/editor/projects/{project_id}", + get(get_editor_project) + .patch( + save_editor_project_layout + .layer(DefaultBodyLimit::max(EDITOR_LAYOUT_REQUEST_BODY_MAX_BYTES)), + ) + .delete(delete_editor_project), + ), + ( + "/api/editor/projects/{project_id}/metadata", + patch(rename_editor_project), + ), + ( + "/api/editor/projects/{project_id}/resources", + post(create_editor_project_resource), + ), + ( + "/api/editor/projects/{project_id}/agent-conversations", + get(list_editor_agent_conversations).post(create_editor_agent_conversation), + ), + ( + "/api/editor/agent-conversations/{conversation_id}", + get(get_editor_agent_conversation).delete(delete_editor_agent_conversation), + ), + ( + "/api/editor/agent-conversations/{conversation_id}/messages", + post(editor_agent_message) + .layer(DefaultBodyLimit::max(EDITOR_AGENT_MESSAGE_BODY_LIMIT_BYTES)), + ), + ( + "/api/editor/agent-conversations/{conversation_id}/messages/{message_id}/confirm", + post(confirm_editor_agent_tool_call), + ), + ( + "/api/editor/agent-conversations/{conversation_id}/messages/{message_id}/cancel", + post(cancel_editor_agent_tool_call), + ), + ( + "/api/editor/project-resources/{resource_id}/showcase", + patch(update_editor_project_resource_showcase), + ), + ("/api/editor/assets/library", get(get_editor_asset_library)), + ( + "/api/editor/assets/folders", + post(create_editor_asset_folder), + ), + ( + "/api/editor/assets/folders/{folder_id}", + patch(update_editor_asset_folder).delete(delete_editor_asset_folder), + ), + ("/api/editor/assets", post(create_editor_asset)), + ( + "/api/editor/assets/{asset_id}", + patch(update_editor_asset).delete(delete_editor_asset), + ), + ( + "/api/editor/assets/{asset_id}/showcase-submissions", + post(submit_editor_asset_showcase), + ), + ( + "/api/editor/images/generations", + post(generate_editor_image), + ), + ( + "/api/editor/scenes/generations", + post(generate_editor_scene), + ), + ( + "/api/editor/llm/icon-specs/refine-game-play", + post(refine_game_play), + ), + ( + "/api/editor/llm/icon-specs/refine-art-style", + post(refine_art_style), + ), + ( + "/api/editor/icon-specs/generations", + post(generate_icon_spec), + ), + ("/api/editor/images/edits", post(edit_editor_image)), + ( + "/api/editor/images/background-removals", + post(remove_editor_image_background), + ), + ( + "/api/editor/images/pixel-art-snaps", + post(snap_editor_image_to_pixel_art), + ), + ( + "/api/editor/icon-spritesheets/generations", + post(generate_editor_icon_spritesheet), + ), + ( + "/api/editor/icon-spritesheets/slices", + post(split_editor_icon_spritesheet), + ), + ( + "/api/editor/ui-designs/assets/extractions", + post(extract_editor_ui_design_assets), + ), + ]; + // 在方法层统一鉴权，未支持的方法仍由 Axum 返回 405。 + let protected_router = protected_routes + .into_iter() + .fold(Router::new(), |router, (path, methods)| { + router.route(path, methods.route_layer(auth.clone())) + }); + // 点赞拒绝响应也必须带私有缓存头，保持该层位于鉴权外侧。 + let showcase_likes_router = Router::new().route( + "/api/editor/showcase/assets/{showcase_id}/likes", + post(toggle_editor_showcase_asset_like) + .route_layer(auth) + .route_layer(middleware::from_fn(add_showcase_private_response_headers)), + ); + Router::new() .route( "/api/editor/generation-pricing", @@ -61,228 +187,8 @@ pub fn router(state: AppState) -> Router { get(list_public_editor_project_resources) .route_layer(middleware::from_fn(add_showcase_private_response_headers)), ) - .route( - "/api/editor/projects/recent", - get(load_recent_editor_project).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/projects", - get(list_editor_projects) - .post(create_editor_project) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/projects/{project_id}", - get(get_editor_project) - .patch( - save_editor_project_layout - .layer(DefaultBodyLimit::max(EDITOR_LAYOUT_REQUEST_BODY_MAX_BYTES)), - ) - .delete(delete_editor_project) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/projects/{project_id}/metadata", - patch(rename_editor_project).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/projects/{project_id}/resources", - post(create_editor_project_resource).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/projects/{project_id}/agent-conversations", - get(list_editor_agent_conversations) - .post(create_editor_agent_conversation) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/agent-conversations/{conversation_id}", - get(get_editor_agent_conversation) - .delete(delete_editor_agent_conversation) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/agent-conversations/{conversation_id}/messages", - post(editor_agent_message) - .layer(DefaultBodyLimit::max(EDITOR_AGENT_MESSAGE_BODY_LIMIT_BYTES)) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/agent-conversations/{conversation_id}/messages/{message_id}/confirm", - post(confirm_editor_agent_tool_call).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/agent-conversations/{conversation_id}/messages/{message_id}/cancel", - post(cancel_editor_agent_tool_call).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/project-resources/{resource_id}/showcase", - patch(update_editor_project_resource_showcase).route_layer( - middleware::from_fn_with_state(state.clone(), require_bearer_auth), - ), - ) - .route( - "/api/editor/assets/library", - get(get_editor_asset_library).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/assets/folders", - post(create_editor_asset_folder).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/assets/folders/{folder_id}", - patch(update_editor_asset_folder) - .delete(delete_editor_asset_folder) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/assets", - post(create_editor_asset).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/assets/{asset_id}", - patch(update_editor_asset) - .delete(delete_editor_asset) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/assets/{asset_id}/showcase-submissions", - post(submit_editor_asset_showcase).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/showcase/assets/{showcase_id}/likes", - post(toggle_editor_showcase_asset_like) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )) - .route_layer(middleware::from_fn(add_showcase_private_response_headers)), - ) - .route( - "/api/editor/images/generations", - post(generate_editor_image).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/scenes/generations", - post(generate_editor_scene).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/llm/icon-specs/refine-game-play", - post(refine_game_play).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/llm/icon-specs/refine-art-style", - post(refine_art_style).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/icon-specs/generations", - post(generate_icon_spec).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/images/edits", - post(edit_editor_image).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/images/background-removals", - post(remove_editor_image_background).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/images/pixel-art-snaps", - post(snap_editor_image_to_pixel_art).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/icon-spritesheets/generations", - post(generate_editor_icon_spritesheet).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/icon-spritesheets/slices", - post(split_editor_icon_spritesheet).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) - .route( - "/api/editor/ui-designs/assets/extractions", - post(extract_editor_ui_design_assets).route_layer(middleware::from_fn_with_state( - state.clone(), - require_bearer_auth, - )), - ) + .merge(protected_router) + .merge(showcase_likes_router) } #[cfg(test)] @@ -334,3 +240,432 @@ mod tests { assert_eq!(vary_values, vec!["Origin", "Authorization"]); } } + +#[cfg(test)] +mod route_contract_tests { + use axum::{ + body::Body, + http::{Request, StatusCode, header}, + middleware, + }; + use http_body_util::BodyExt; + use tower::ServiceExt; + + use super::router; + use crate::{config::AppConfig, request_context::attach_request_context, state::AppState}; + + const PROTECTED_ROUTES: &[(&str, &[&str])] = &[ + ("/api/editor/projects/recent", &["GET"]), + ("/api/editor/projects", &["GET", "POST"]), + ( + "/api/editor/projects/{project_id}", + &["GET", "PATCH", "DELETE"], + ), + ("/api/editor/projects/{project_id}/metadata", &["PATCH"]), + ("/api/editor/projects/{project_id}/resources", &["POST"]), + ( + "/api/editor/projects/{project_id}/agent-conversations", + &["GET", "POST"], + ), + ( + "/api/editor/agent-conversations/{conversation_id}", + &["GET", "DELETE"], + ), + ( + "/api/editor/agent-conversations/{conversation_id}/messages", + &["POST"], + ), + ( + "/api/editor/agent-conversations/{conversation_id}/messages/{message_id}/confirm", + &["POST"], + ), + ( + "/api/editor/agent-conversations/{conversation_id}/messages/{message_id}/cancel", + &["POST"], + ), + ( + "/api/editor/project-resources/{resource_id}/showcase", + &["PATCH"], + ), + ("/api/editor/assets/library", &["GET"]), + ("/api/editor/assets/folders", &["POST"]), + ( + "/api/editor/assets/folders/{folder_id}", + &["PATCH", "DELETE"], + ), + ("/api/editor/assets", &["POST"]), + ("/api/editor/assets/{asset_id}", &["PATCH", "DELETE"]), + ( + "/api/editor/assets/{asset_id}/showcase-submissions", + &["POST"], + ), + ("/api/editor/showcase/assets/{showcase_id}/likes", &["POST"]), + ("/api/editor/images/generations", &["POST"]), + ("/api/editor/scenes/generations", &["POST"]), + ("/api/editor/llm/icon-specs/refine-game-play", &["POST"]), + ("/api/editor/llm/icon-specs/refine-art-style", &["POST"]), + ("/api/editor/icon-specs/generations", &["POST"]), + ("/api/editor/images/edits", &["POST"]), + ("/api/editor/images/background-removals", &["POST"]), + ("/api/editor/images/pixel-art-snaps", &["POST"]), + ("/api/editor/icon-spritesheets/generations", &["POST"]), + ("/api/editor/icon-spritesheets/slices", &["POST"]), + ("/api/editor/ui-designs/assets/extractions", &["POST"]), + ]; + + fn app() -> axum::Router { + let state = AppState::new(AppConfig::default()).expect("state should build"); + router(state.clone()) + .layer(middleware::from_fn(attach_request_context)) + .with_state(state) + } + + #[tokio::test] + async fn protected_route_matrix_keeps_auth_methods_and_head_behavior() { + let app = app(); + for (template, supported) in PROTECTED_ROUTES { + let path = template + .split('/') + .map(|segment| { + if segment.starts_with('{') { + "fixture" + } else { + segment + } + }) + .collect::>() + .join("/"); + let mut allowed = supported.to_vec(); + if allowed.contains(&"GET") { + allowed.push("HEAD"); + } + allowed.sort_unstable(); + for credential in [None, Some("Bearer invalid-token")] { + for method in [ + "GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "TRACE", + ] { + let mut request = Request::builder().method(method).uri(&path); + if let Some(credential) = credential { + request = request.header(header::AUTHORIZATION, credential); + } + let response = app + .clone() + .oneshot(request.body(Body::empty()).expect("request should build")) + .await + .expect("request should complete"); + let expected = if allowed.contains(&method) { + StatusCode::UNAUTHORIZED + } else { + StatusCode::METHOD_NOT_ALLOWED + }; + assert_eq!( + response.status(), + expected, + "{method} {path}, credential={credential:?}" + ); + if expected == StatusCode::METHOD_NOT_ALLOWED { + let mut actual = response.headers()[header::ALLOW] + .to_str() + .expect("Allow should be ASCII") + .split(',') + .map(str::trim) + .collect::>(); + actual.sort_unstable(); + assert_eq!(actual, allowed, "Allow for {method} {path}"); + } + if method == "HEAD" { + assert!( + response + .into_body() + .collect() + .await + .expect("HEAD body should collect") + .to_bytes() + .is_empty(), + "HEAD {path} must not return a body" + ); + } + } + } + } + } + + #[tokio::test] + async fn unknown_paths_remain_not_found_before_authentication() { + for credential in [None, Some("Bearer invalid-token")] { + let mut request = Request::builder().uri("/api/editor/missing/unknown/route"); + if let Some(credential) = credential { + request = request.header(header::AUTHORIZATION, credential); + } + let response = app() + .oneshot(request.body(Body::empty()).expect("request should build")) + .await + .expect("request should complete"); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + } + } + + #[tokio::test] + async fn public_reads_and_showcase_rejections_keep_cache_boundaries() { + use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }; + use std::time::Duration; + + // 精选读取会访问正式 facade；隔离上游返回 503，明确验证匿名请求已进入 handler。 + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("fixture should bind"); + let address = listener + .local_addr() + .expect("fixture should have an address"); + let calls = Arc::new(AtomicUsize::new(0)); + let fixture_calls = calls.clone(); + let fixture = axum::Router::new().fallback(move || { + fixture_calls.fetch_add(1, Ordering::SeqCst); + async { StatusCode::SERVICE_UNAVAILABLE } + }); + let server = tokio::spawn(async move { + axum::serve(listener, fixture) + .await + .expect("fixture should serve"); + }); + let state = AppState::new(AppConfig { + spacetime_server_url: format!("http://{address}"), + spacetime_database: "route-showcase-unavailable-fixture".to_string(), + spacetime_procedure_timeout: Duration::from_secs(2), + ..AppConfig::default() + }) + .expect("state should build"); + let app = router(state.clone()) + .layer(middleware::from_fn(attach_request_context)) + .with_state(state); + + let pricing = app + .clone() + .oneshot( + Request::builder() + .uri("/api/editor/generation-pricing") + .body(Body::empty()) + .expect("pricing request should build"), + ) + .await + .expect("pricing request should complete"); + assert_eq!(pricing.status(), StatusCode::OK); + assert_eq!(calls.load(Ordering::SeqCst), 0); + + let response = tokio::time::timeout( + Duration::from_secs(5), + app.clone().oneshot( + Request::builder() + .uri("/api/editor/showcase/resources") + .body(Body::empty()) + .expect("showcase request should build"), + ), + ) + .await + .expect("showcase request should remain bounded") + .expect("showcase request should complete"); + server.abort(); + let _ = server.await; + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + assert!( + calls.load(Ordering::SeqCst) > 0, + "anonymous request must reach the database facade" + ); + assert_eq!( + response.headers()[header::CACHE_CONTROL], + "private, no-store" + ); + assert!( + response + .headers() + .get_all(header::VARY) + .iter() + .any(|value| value == "Authorization") + ); + let body = response + .into_body() + .collect() + .await + .expect("error body should collect") + .to_bytes(); + let payload: serde_json::Value = + serde_json::from_slice(&body).expect("error should be JSON"); + assert_eq!(payload["error"]["code"], "UPSTREAM_ERROR"); + assert_eq!(payload["error"]["details"]["provider"], "spacetimedb"); + assert!( + payload["error"]["details"]["message"] + .as_str() + .is_some_and(|message| message.contains("503")) + ); + + for (method, path) in [ + ("GET", "/api/editor/showcase/resources"), + ("POST", "/api/editor/showcase/assets/fixture/likes"), + ] { + let response = app + .clone() + .oneshot( + Request::builder() + .method(method) + .uri(path) + .header(header::AUTHORIZATION, "Bearer invalid-token") + .body(Body::empty()) + .expect("showcase request should build"), + ) + .await + .expect("showcase request should complete"); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + assert_eq!( + response.headers()[header::CACHE_CONTROL], + "private, no-store" + ); + assert!( + response + .headers() + .get_all(header::VARY) + .iter() + .any(|value| value == "Authorization") + ); + } + } + + #[tokio::test] + async fn authentication_precedes_layout_and_agent_body_limits() { + let app = app(); + for (method, path, limit) in [ + ( + "PATCH", + "/api/editor/projects/fixture", + super::EDITOR_LAYOUT_REQUEST_BODY_MAX_BYTES, + ), + ( + "POST", + "/api/editor/agent-conversations/fixture/messages", + super::EDITOR_AGENT_MESSAGE_BODY_LIMIT_BYTES, + ), + ] { + let response = app + .clone() + .oneshot( + Request::builder() + .method(method) + .uri(path) + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from("x".repeat(limit + 1))) + .expect("oversized request should build"), + ) + .await + .expect("oversized request should complete"); + assert_eq!( + response.status(), + StatusCode::UNAUTHORIZED, + "{method} {path}" + ); + } + } + + #[tokio::test] + async fn authenticated_layout_and_agent_requests_keep_distinct_body_limits() { + use platform_auth::{ + AccessTokenClaims, AccessTokenClaimsInput, AuthProvider, BindingStatus, + sign_access_token, + }; + + let state = AppState::new(AppConfig::default()).expect("state should build"); + let user = state + .seed_test_phone_user_with_password("13800138991", "secret123") + .await; + let session_id = + state.seed_test_refresh_session_for_user(&user, "session-route-body-limits"); + let claims = AccessTokenClaims::from_input( + AccessTokenClaimsInput { + user_id: user.id.clone(), + session_id, + provider: AuthProvider::Password, + roles: vec!["user".to_string()], + token_version: user.token_version, + phone_verified: false, + binding_status: BindingStatus::Active, + display_name: Some(user.display_name.clone()), + }, + state.auth_jwt_config(), + time::OffsetDateTime::now_utc(), + ) + .expect("claims should build"); + let token = sign_access_token(&claims, state.auth_jwt_config()).expect("token should sign"); + let app = router(state.clone()) + .layer(middleware::from_fn(attach_request_context)) + .with_state(state); + + for (method, path, size, expected) in [ + ( + "PATCH", + "/api/editor/projects/fixture", + 2 * 1024 * 1024 + 1, + StatusCode::UNPROCESSABLE_ENTITY, + ), + ( + "PATCH", + "/api/editor/projects/fixture", + super::EDITOR_LAYOUT_REQUEST_BODY_MAX_BYTES + 1, + StatusCode::PAYLOAD_TOO_LARGE, + ), + ( + "POST", + "/api/editor/agent-conversations/fixture/messages", + super::EDITOR_AGENT_MESSAGE_BODY_LIMIT_BYTES + 1, + StatusCode::PAYLOAD_TOO_LARGE, + ), + ] { + // 用有效 JSON 字符串触发类型校验，避免进入业务 handler 或产生持久化副作用。 + let body = format!("\"{}\"", "x".repeat(size)); + let response = app + .clone() + .oneshot( + Request::builder() + .method(method) + .uri(path) + .header(header::AUTHORIZATION, format!("Bearer {token}")) + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(body)) + .expect("body limit request should build"), + ) + .await + .expect("body limit request should complete"); + assert_eq!(response.status(), expected, "{method} {path}, size={size}"); + } + } + + #[tokio::test] + async fn method_auth_layer_leaves_405_outside_authentication() { + async fn reject(_: Request, _: axum::middleware::Next) -> StatusCode { + StatusCode::UNAUTHORIZED + } + let method_layer = axum::Router::new().route( + "/only-get", + axum::routing::get(|| async { "ok" }).route_layer(middleware::from_fn(reject)), + ); + let router_layer = axum::Router::new() + .route("/only-get", axum::routing::get(|| async { "ok" })) + .route_layer(middleware::from_fn(reject)); + for (app, expected) in [ + (method_layer, StatusCode::METHOD_NOT_ALLOWED), + (router_layer, StatusCode::UNAUTHORIZED), + ] { + let response = app + .oneshot( + Request::builder() + .method("OPTIONS") + .uri("/only-get") + .body(Body::empty()) + .expect("request should build"), + ) + .await + .expect("request should complete"); + assert_eq!(response.status(), expected); + } + } +} diff --git a/server-rs/crates/api-server/src/modules/external_api.rs b/server-rs/crates/api-server/src/modules/external_api.rs index 9da0014fb..214baf8e3 100644 --- a/server-rs/crates/api-server/src/modules/external_api.rs +++ b/server-rs/crates/api-server/src/modules/external_api.rs @@ -44,8 +44,114 @@ pub fn router(state: AppState) -> Router { require_external_mcp_api_key, )); + let auth = middleware::from_fn_with_state(state, require_external_api_key); + let protected_routes = [ + ( + "/api/external/v1/assets/direct-upload-tickets", + post(create_external_direct_upload_ticket), + ), + ( + "/api/external/v1/assets/objects/confirm", + post(confirm_external_asset_object), + ), + ( + "/api/external/v1/assets/read-url", + get(get_external_asset_read_url), + ), + ( + "/api/external/v1/editor/projects", + get(list_external_editor_projects).post(create_external_editor_project), + ), + ( + "/api/external/v1/editor/projects/recent", + get(load_recent_external_editor_project), + ), + ( + "/api/external/v1/editor/projects/{project_id}", + get(get_external_editor_project).delete(delete_external_editor_project), + ), + ( + "/api/external/v1/editor/projects/{project_id}/metadata", + patch(rename_external_editor_project), + ), + ( + "/api/external/v1/editor/projects/{project_id}/canvas", + patch(save_external_editor_canvas) + .layer(DefaultBodyLimit::max(EDITOR_LAYOUT_REQUEST_BODY_MAX_BYTES)), + ), + ( + "/api/external/v1/editor/projects/{project_id}/resources", + post(create_external_editor_project_resource), + ), + ( + "/api/external/v1/editor/assets/library", + get(get_external_editor_asset_library), + ), + ( + "/api/external/v1/editor/assets/folders", + post(create_external_editor_asset_folder), + ), + ( + "/api/external/v1/editor/assets/folders/{folder_id}", + patch(update_external_editor_asset_folder).delete(delete_external_editor_asset_folder), + ), + ( + "/api/external/v1/editor/assets", + post(create_external_editor_asset), + ), + ( + "/api/external/v1/editor/assets/{asset_id}", + patch(update_external_editor_asset).delete(delete_external_editor_asset), + ), + ( + "/api/external/v1/generations/{operation_id}", + get(get_external_editor_generation_job), + ), + ( + "/api/external/v1/editor/images/generations", + post(generate_external_editor_image), + ), + ( + "/api/external/v1/editor/images/edits", + post(edit_external_editor_image), + ), + ( + "/api/external/v1/editor/images/background-removals", + post(remove_external_editor_image_background), + ), + ( + "/api/external/v1/editor/icon-spritesheets/generations", + post(generate_external_editor_icon_spritesheet), + ), + ( + "/api/external/v1/editor/ui-designs/assets/extractions", + post(extract_external_editor_ui_design_assets), + ), + ( + "/api/external/v1/editor/character-animations/generations", + post(generate_external_editor_character_animation), + ), + ( + "/api/external/v1/editor/videos/generations", + post(generate_external_editor_video), + ), + ( + "/api/external/v1/editor/audios/sound-effects/generations", + post(generate_external_editor_sound_effect), + ), + ( + "/api/external/v1/editor/audios/background-music/generations", + post(generate_external_editor_background_music), + ), + ]; + // 在方法层统一鉴权，未支持的方法仍由 Axum 返回 405。 + let protected_router = protected_routes + .into_iter() + .fold(Router::new(), |router, (path, methods)| { + router.route(path, methods.route_layer(auth.clone())) + }); + Router::new() - .merge(mcp_router) .route("/api/external/v1/openapi.json", get(openapi_json)) .route( "/api/external/v1/agent-integration.json", @@ -59,175 +165,320 @@ pub fn router(state: AppState) -> Router { "/api/external/v1/skill.zip", get(download_external_skill_archive), ) - .route( - "/api/external/v1/assets/direct-upload-tickets", - post(create_external_direct_upload_ticket).route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/assets/objects/confirm", - post(confirm_external_asset_object).route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/assets/read-url", - get(get_external_asset_read_url).route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/editor/projects", - get(list_external_editor_projects) - .post(create_external_editor_project) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/editor/projects/recent", - get(load_recent_external_editor_project).route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/editor/projects/{project_id}", - get(get_external_editor_project) - .delete(delete_external_editor_project) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/editor/projects/{project_id}/metadata", - patch(rename_external_editor_project).route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/editor/projects/{project_id}/canvas", - patch(save_external_editor_canvas) - .layer(DefaultBodyLimit::max(EDITOR_LAYOUT_REQUEST_BODY_MAX_BYTES)) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/editor/projects/{project_id}/resources", - post(create_external_editor_project_resource).route_layer( - middleware::from_fn_with_state(state.clone(), require_external_api_key), - ), - ) - .route( - "/api/external/v1/editor/assets/library", - get(get_external_editor_asset_library).route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/editor/assets/folders", - post(create_external_editor_asset_folder).route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/editor/assets/folders/{folder_id}", - patch(update_external_editor_asset_folder) - .delete(delete_external_editor_asset_folder) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/editor/assets", - post(create_external_editor_asset).route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/editor/assets/{asset_id}", - patch(update_external_editor_asset) - .delete(delete_external_editor_asset) - .route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/generations/{operation_id}", - get(get_external_editor_generation_job).route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/editor/images/generations", - post(generate_external_editor_image).route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/editor/images/edits", - post(edit_external_editor_image).route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/editor/images/background-removals", - post(remove_external_editor_image_background).route_layer( - middleware::from_fn_with_state(state.clone(), require_external_api_key), - ), - ) - .route( - "/api/external/v1/editor/icon-spritesheets/generations", - post(generate_external_editor_icon_spritesheet).route_layer( - middleware::from_fn_with_state(state.clone(), require_external_api_key), - ), - ) - .route( - "/api/external/v1/editor/ui-designs/assets/extractions", - post(extract_external_editor_ui_design_assets).route_layer( - middleware::from_fn_with_state(state.clone(), require_external_api_key), - ), - ) - .route( - "/api/external/v1/editor/character-animations/generations", - post(generate_external_editor_character_animation).route_layer( - middleware::from_fn_with_state(state.clone(), require_external_api_key), - ), - ) - .route( - "/api/external/v1/editor/videos/generations", - post(generate_external_editor_video).route_layer(middleware::from_fn_with_state( - state.clone(), - require_external_api_key, - )), - ) - .route( - "/api/external/v1/editor/audios/sound-effects/generations", - post(generate_external_editor_sound_effect).route_layer( - middleware::from_fn_with_state(state.clone(), require_external_api_key), - ), - ) - .route( - "/api/external/v1/editor/audios/background-music/generations", - post(generate_external_editor_background_music).route_layer( - middleware::from_fn_with_state(state, require_external_api_key), - ), - ) + .merge(protected_router) + .merge(mcp_router) +} + +#[cfg(test)] +mod route_contract_tests { + use std::sync::Arc; + + use axum::{ + body::Body, + http::{Request, StatusCode, header}, + middleware, + }; + use futures_util::future::BoxFuture; + use http_body_util::BodyExt; + use spacetime_client::{ + ExternalApiKeyAuthenticateRecordInput, ExternalApiKeyRecord, SpacetimeClientError, + }; + use tower::ServiceExt; + + use super::router; + use crate::{ + config::AppConfig, + request_context::attach_request_context, + state::{AppState, ExternalApiAuthState, external_api_auth::ExternalApiKeyAuthenticator}, + }; + + const VALID_KEY: &str = "tnr_sk_route-body-limit-fixture"; + + struct FixtureKeyAuthenticator; + + impl ExternalApiKeyAuthenticator for FixtureKeyAuthenticator { + fn authenticate_external_api_key( + &self, + input: ExternalApiKeyAuthenticateRecordInput, + ) -> BoxFuture<'_, Result> { + Box::pin(async move { + if input.key_hash != crate::external_api_keys::hash_external_api_key(VALID_KEY) { + return Err(SpacetimeClientError::Procedure( + "API Key 不存在".to_string(), + )); + } + Ok(ExternalApiKeyRecord { + key_id: "key-route-fixture".to_string(), + owner_user_id: "owner-route-fixture".to_string(), + name: "route fixture".to_string(), + key_prefix: "tnr_sk_route".to_string(), + scopes: Vec::new(), + created_at: "2026-09-18T00:00:00Z".to_string(), + last_used_at: None, + revoked_at: None, + updated_at: "2026-09-18T00:00:00Z".to_string(), + }) + }) + } + } + + const PROTECTED_ROUTES: &[(&str, &[&str])] = &[ + ("/api/external/v1/assets/direct-upload-tickets", &["POST"]), + ("/api/external/v1/assets/objects/confirm", &["POST"]), + ("/api/external/v1/assets/read-url", &["GET"]), + ("/api/external/v1/editor/projects", &["GET", "POST"]), + ("/api/external/v1/editor/projects/recent", &["GET"]), + ( + "/api/external/v1/editor/projects/{project_id}", + &["GET", "DELETE"], + ), + ( + "/api/external/v1/editor/projects/{project_id}/metadata", + &["PATCH"], + ), + ( + "/api/external/v1/editor/projects/{project_id}/canvas", + &["PATCH"], + ), + ( + "/api/external/v1/editor/projects/{project_id}/resources", + &["POST"], + ), + ("/api/external/v1/editor/assets/library", &["GET"]), + ("/api/external/v1/editor/assets/folders", &["POST"]), + ( + "/api/external/v1/editor/assets/folders/{folder_id}", + &["PATCH", "DELETE"], + ), + ("/api/external/v1/editor/assets", &["POST"]), + ( + "/api/external/v1/editor/assets/{asset_id}", + &["PATCH", "DELETE"], + ), + ("/api/external/v1/generations/{operation_id}", &["GET"]), + ("/api/external/v1/editor/images/generations", &["POST"]), + ("/api/external/v1/editor/images/edits", &["POST"]), + ( + "/api/external/v1/editor/images/background-removals", + &["POST"], + ), + ( + "/api/external/v1/editor/icon-spritesheets/generations", + &["POST"], + ), + ( + "/api/external/v1/editor/ui-designs/assets/extractions", + &["POST"], + ), + ( + "/api/external/v1/editor/character-animations/generations", + &["POST"], + ), + ("/api/external/v1/editor/videos/generations", &["POST"]), + ( + "/api/external/v1/editor/audios/sound-effects/generations", + &["POST"], + ), + ( + "/api/external/v1/editor/audios/background-music/generations", + &["POST"], + ), + ]; + + fn app() -> axum::Router { + let state = AppState::new(AppConfig::default()) + .expect("state should build") + .with_external_api_auth_state(ExternalApiAuthState::new(Arc::new( + FixtureKeyAuthenticator, + ))); + router(state.clone()) + .layer(middleware::from_fn(attach_request_context)) + .with_state(state) + } + + #[tokio::test] + async fn protected_route_matrix_keeps_auth_methods_and_head_behavior() { + let app = app(); + for (template, supported) in PROTECTED_ROUTES { + let path = template + .split('/') + .map(|segment| { + if segment.starts_with('{') { + "fixture" + } else { + segment + } + }) + .collect::>() + .join("/"); + let mut allowed = supported.to_vec(); + if allowed.contains(&"GET") { + allowed.push("HEAD"); + } + allowed.sort_unstable(); + for credential in [None, Some("Bearer invalid-token")] { + for method in [ + "GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "TRACE", + ] { + let mut request = Request::builder().method(method).uri(&path); + if let Some(credential) = credential { + request = request.header(header::AUTHORIZATION, credential); + } + let response = app + .clone() + .oneshot(request.body(Body::empty()).expect("request should build")) + .await + .expect("request should complete"); + let expected = if allowed.contains(&method) { + StatusCode::UNAUTHORIZED + } else { + StatusCode::METHOD_NOT_ALLOWED + }; + assert_eq!( + response.status(), + expected, + "{method} {path}, credential={credential:?}" + ); + if expected == StatusCode::METHOD_NOT_ALLOWED { + let mut actual = response.headers()[header::ALLOW] + .to_str() + .expect("Allow should be ASCII") + .split(',') + .map(str::trim) + .collect::>(); + actual.sort_unstable(); + assert_eq!(actual, allowed, "Allow for {method} {path}"); + } + if method == "HEAD" { + assert!( + response + .into_body() + .collect() + .await + .expect("HEAD body should collect") + .to_bytes() + .is_empty(), + "HEAD {path} must not return a body" + ); + } + } + } + } + } + + #[tokio::test] + async fn unknown_paths_remain_not_found_before_authentication() { + for credential in [None, Some("Bearer invalid-token")] { + let mut request = Request::builder().uri("/api/external/v1/missing/unknown/route"); + if let Some(credential) = credential { + request = request.header(header::AUTHORIZATION, credential); + } + let response = app() + .oneshot(request.body(Body::empty()).expect("request should build")) + .await + .expect("request should complete"); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + } + } + + #[tokio::test] + async fn integration_documents_remain_public_and_mcp_keeps_its_auth_challenge() { + let app = app(); + for path in [ + "/api/external/v1/openapi.json", + "/api/external/v1/agent-integration.json", + "/api/external/v1/skill/SKILL.md", + "/api/external/v1/skill.zip", + ] { + for method in ["GET", "HEAD"] { + let response = app + .clone() + .oneshot( + Request::builder() + .method(method) + .uri(path) + .header(header::AUTHORIZATION, "Bearer invalid-token") + .body(Body::empty()) + .expect("document request should build"), + ) + .await + .expect("document request should complete"); + assert_eq!(response.status(), StatusCode::OK, "{method} {path}"); + if method == "HEAD" { + assert!( + response + .into_body() + .collect() + .await + .expect("HEAD body should collect") + .to_bytes() + .is_empty() + ); + } + } + } + for method in ["GET", "POST", "DELETE", "OPTIONS"] { + let response = app + .clone() + .oneshot( + Request::builder() + .method(method) + .uri("/api/external/v1/mcp") + .body(Body::empty()) + .expect("MCP request should build"), + ) + .await + .expect("MCP request should complete"); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED, "MCP {method}"); + assert_eq!( + response.headers()[header::WWW_AUTHENTICATE], + "Bearer realm=\"genarrative-external-editor\"" + ); + } + } + + #[tokio::test] + async fn canvas_authentication_precedes_body_limit_rejection() { + let response = app() + .oneshot( + Request::builder() + .method("PATCH") + .uri("/api/external/v1/editor/projects/fixture/canvas") + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from("x".repeat( + crate::editor_project::EDITOR_LAYOUT_REQUEST_BODY_MAX_BYTES + 1, + ))) + .expect("oversized request should build"), + ) + .await + .expect("oversized request should complete"); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + } + + #[tokio::test] + async fn authenticated_canvas_keeps_its_four_mebibyte_body_limit() { + let app = app(); + for (size, expected) in [ + (2 * 1024 * 1024 + 1, StatusCode::UNPROCESSABLE_ENTITY), + ( + crate::editor_project::EDITOR_LAYOUT_REQUEST_BODY_MAX_BYTES + 1, + StatusCode::PAYLOAD_TOO_LARGE, + ), + ] { + let response = app + .clone() + .oneshot( + Request::builder() + .method("PATCH") + .uri("/api/external/v1/editor/projects/fixture/canvas") + .header(header::AUTHORIZATION, format!("Bearer {VALID_KEY}")) + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(format!("\"{}\"", "x".repeat(size)))) + .expect("canvas request should build"), + ) + .await + .expect("canvas request should complete"); + assert_eq!(response.status(), expected, "size={size}"); + } + } } diff --git a/server-rs/crates/api-server/src/state.rs b/server-rs/crates/api-server/src/state.rs index 91c58792f..98033a8f5 100644 --- a/server-rs/crates/api-server/src/state.rs +++ b/server-rs/crates/api-server/src/state.rs @@ -52,6 +52,12 @@ use crate::work_author::{ ORPHAN_WORK_AUTHOR_DISPLAY_NAME, ORPHAN_WORK_AUTHOR_PUBLIC_USER_CODE, ORPHAN_WORK_OWNER_USER_ID, }; +pub(crate) mod external_api_auth; +pub(crate) mod project_metadata; +pub use external_api_auth::ExternalApiAuthState; +pub(crate) use project_metadata::EditorMediaStorageState; +pub use project_metadata::EditorProjectState; + const ADMIN_ROLE: &str = "admin"; const EDITOR_AGENT_LLM_MAX_RETRIES: u32 = 1; const EDITOR_AGENT_LLM_MAX_RETRY_BACKOFF_MS: u64 = 60_000; @@ -297,6 +303,8 @@ pub struct AppStateInner { #[cfg_attr(not(test), allow(dead_code))] ai_task_service: AiTaskService, spacetime_client: SpacetimeClient, + external_api_auth_state: ExternalApiAuthState, + editor_project_state: EditorProjectState, #[cfg(any())] puzzle_gallery_cache: PuzzleGalleryCache, tracking_outbox: Option>, @@ -638,6 +646,14 @@ impl AppState { let character_animation_oss_io_limiter = Arc::new(Semaphore::new(CHARACTER_ANIMATION_OSS_MAX_CONCURRENCY)); let editor_oss_http_client = build_editor_oss_http_client()?; + let editor_project_state = EditorProjectState::new( + Arc::new(spacetime_client.clone()), + Arc::new(EditorMediaStorageState::new( + spacetime_client.clone(), + oss_client.clone(), + editor_oss_http_client.clone(), + )), + ); let http_request_permit_pools = HttpRequestPermitPools::from_config(&config); let (profile_recharge_order_updates, _) = broadcast::channel(128); // `ensure_orphan_work_owner_user` 只为公开作品作者回退提供进程内占位账号， @@ -685,6 +701,8 @@ impl AppState { wechat_client, wechat_pay_client, ai_task_service, + external_api_auth_state: ExternalApiAuthState::new(Arc::new(spacetime_client.clone())), + editor_project_state, spacetime_client, #[cfg(any())] puzzle_gallery_cache: PuzzleGalleryCache::new(), @@ -1577,6 +1595,14 @@ impl AppState { &self.spacetime_client } + #[cfg(test)] + pub(crate) fn with_external_api_auth_state(mut self, state: ExternalApiAuthState) -> Self { + Arc::get_mut(&mut self.0) + .expect("configure external authentication before sharing AppState") + .external_api_auth_state = state; + self + } + #[cfg(any())] pub fn puzzle_gallery_cache(&self) -> &PuzzleGalleryCache { &self.puzzle_gallery_cache diff --git a/server-rs/crates/api-server/src/state/external_api_auth.rs b/server-rs/crates/api-server/src/state/external_api_auth.rs new file mode 100644 index 000000000..2e7f7cb85 --- /dev/null +++ b/server-rs/crates/api-server/src/state/external_api_auth.rs @@ -0,0 +1,47 @@ +use std::sync::Arc; + +use axum::extract::FromRef; +use futures_util::future::BoxFuture; +use spacetime_client::{ + ExternalApiKeyAuthenticateRecordInput, ExternalApiKeyRecord, SpacetimeClient, + SpacetimeClientError, +}; + +use super::AppState; + +pub(crate) trait ExternalApiKeyAuthenticator: Send + Sync { + fn authenticate_external_api_key( + &self, + input: ExternalApiKeyAuthenticateRecordInput, + ) -> BoxFuture<'_, Result>; +} + +impl ExternalApiKeyAuthenticator for SpacetimeClient { + fn authenticate_external_api_key( + &self, + input: ExternalApiKeyAuthenticateRecordInput, + ) -> BoxFuture<'_, Result> { + Box::pin(SpacetimeClient::authenticate_external_api_key(self, input)) + } +} + +#[derive(Clone)] +pub struct ExternalApiAuthState { + authenticator: Arc, +} + +impl ExternalApiAuthState { + pub(crate) fn new(authenticator: Arc) -> Self { + Self { authenticator } + } + + pub(crate) fn authenticator(&self) -> &dyn ExternalApiKeyAuthenticator { + self.authenticator.as_ref() + } +} + +impl FromRef for ExternalApiAuthState { + fn from_ref(state: &AppState) -> Self { + state.external_api_auth_state.clone() + } +} diff --git a/server-rs/crates/api-server/src/state/project_metadata.rs b/server-rs/crates/api-server/src/state/project_metadata.rs new file mode 100644 index 000000000..280fd2cea --- /dev/null +++ b/server-rs/crates/api-server/src/state/project_metadata.rs @@ -0,0 +1,195 @@ +use std::sync::Arc; + +use axum::extract::FromRef; +use futures_util::future::BoxFuture; +use platform_oss::OssClient; +use spacetime_client::{ + EditorProjectCreateRecordInput, EditorProjectDeleteRecordInput, EditorProjectGetRecordInput, + EditorProjectLayoutSaveV2AckRecord, EditorProjectLayoutSaveV2RecordInput, EditorProjectRecord, + EditorProjectRenameRecordInput, SpacetimeClient, SpacetimeClientError, +}; + +use super::AppState; + +// 元数据入口只依赖现役 typed facade 的项目能力，不暴露其它领域的数据库操作。 +pub(crate) trait EditorProjectRepository: Send + Sync { + fn get_recent_editor_project( + &self, + owner_user_id: String, + ) -> BoxFuture<'_, Result, SpacetimeClientError>>; + + fn list_editor_projects( + &self, + owner_user_id: String, + ) -> BoxFuture<'_, Result, SpacetimeClientError>>; + + fn get_editor_project( + &self, + input: EditorProjectGetRecordInput, + ) -> BoxFuture<'_, Result>; + + fn create_editor_project( + &self, + input: EditorProjectCreateRecordInput, + ) -> BoxFuture<'_, Result>; + + fn rename_editor_project( + &self, + input: EditorProjectRenameRecordInput, + ) -> BoxFuture<'_, Result>; + + fn delete_editor_project( + &self, + input: EditorProjectDeleteRecordInput, + ) -> BoxFuture<'_, Result>; + + fn save_editor_project_layout_v2_ack( + &self, + input: EditorProjectLayoutSaveV2RecordInput, + ) -> BoxFuture<'_, Result>; +} + +impl EditorProjectRepository for SpacetimeClient { + fn get_recent_editor_project( + &self, + owner_user_id: String, + ) -> BoxFuture<'_, Result, SpacetimeClientError>> { + Box::pin(SpacetimeClient::get_recent_editor_project( + self, + owner_user_id, + )) + } + + fn list_editor_projects( + &self, + owner_user_id: String, + ) -> BoxFuture<'_, Result, SpacetimeClientError>> { + Box::pin(SpacetimeClient::list_editor_projects(self, owner_user_id)) + } + + fn get_editor_project( + &self, + input: EditorProjectGetRecordInput, + ) -> BoxFuture<'_, Result> { + Box::pin(SpacetimeClient::get_editor_project(self, input)) + } + + fn create_editor_project( + &self, + input: EditorProjectCreateRecordInput, + ) -> BoxFuture<'_, Result> { + Box::pin(SpacetimeClient::create_editor_project(self, input)) + } + + fn rename_editor_project( + &self, + input: EditorProjectRenameRecordInput, + ) -> BoxFuture<'_, Result> { + Box::pin(SpacetimeClient::rename_editor_project(self, input)) + } + + fn delete_editor_project( + &self, + input: EditorProjectDeleteRecordInput, + ) -> BoxFuture<'_, Result> { + Box::pin(SpacetimeClient::delete_editor_project(self, input)) + } + + fn save_editor_project_layout_v2_ack( + &self, + input: EditorProjectLayoutSaveV2RecordInput, + ) -> BoxFuture<'_, Result> { + Box::pin(SpacetimeClient::save_editor_project_layout_v2_ack( + self, input, + )) + } +} + +pub(crate) trait EditorProjectMediaRepair: Send + Sync { + fn repair_project(&self, project: EditorProjectRecord) -> BoxFuture<'_, EditorProjectRecord>; +} + +#[derive(Clone)] +pub struct EditorProjectState { + projects: Arc, + media_repair: Arc, +} + +impl EditorProjectState { + pub(crate) fn new( + projects: Arc, + media_repair: Arc, + ) -> Self { + Self { + projects, + media_repair, + } + } + + pub(crate) fn projects(&self) -> &dyn EditorProjectRepository { + self.projects.as_ref() + } + + pub(crate) async fn repair_project_media( + &self, + project: EditorProjectRecord, + ) -> EditorProjectRecord { + self.media_repair.repair_project(project).await + } +} + +impl FromRef for EditorProjectState { + fn from_ref(state: &AppState) -> Self { + state.editor_project_state.clone() + } +} + +// 读项目时的内联媒体修复与生成写入共用相同存储依赖，保持原有上传和登记顺序。 +#[derive(Clone)] +pub(crate) struct EditorMediaStorageState { + spacetime_client: SpacetimeClient, + oss_client: Option, + http_client: reqwest::Client, +} + +impl EditorMediaStorageState { + pub(super) fn new( + spacetime_client: SpacetimeClient, + oss_client: Option, + http_client: reqwest::Client, + ) -> Self { + Self { + spacetime_client, + oss_client, + http_client, + } + } + + pub(crate) fn spacetime_client(&self) -> &SpacetimeClient { + &self.spacetime_client + } + + pub(crate) fn oss_client(&self) -> Option<&OssClient> { + self.oss_client.as_ref() + } + + pub(crate) fn editor_oss_http_client(&self) -> &reqwest::Client { + &self.http_client + } +} + +impl FromRef for EditorMediaStorageState { + fn from_ref(state: &AppState) -> Self { + Self::new( + state.spacetime_client().clone(), + state.oss_client().cloned(), + state.editor_oss_http_client().clone(), + ) + } +} + +impl EditorProjectMediaRepair for EditorMediaStorageState { + fn repair_project(&self, project: EditorProjectRecord) -> BoxFuture<'_, EditorProjectRecord> { + Box::pin(crate::editor_project::repair_editor_project_record_inline_media(self, project)) + } +} diff --git a/server-rs/crates/api-server/src/telemetry.rs b/server-rs/crates/api-server/src/telemetry.rs index 99ca82e41..ba0a78f84 100644 --- a/server-rs/crates/api-server/src/telemetry.rs +++ b/server-rs/crates/api-server/src/telemetry.rs @@ -1,6 +1,6 @@ use axum::{ body::Body, - extract::State, + extract::{MatchedPath, State}, http::{HeaderMap, Request, Response}, middleware::Next, }; @@ -28,28 +28,42 @@ pub async fn record_http_observability( State(state): State, request: Request, next: Next, +) -> Response { + observe_http_request( + http_metrics(), + state.config.slow_request_threshold_ms, + request, + next, + ) + .await +} + +async fn observe_http_request( + metrics: &HttpMetrics, + slow_request_threshold_ms: u64, + request: Request, + next: Next, ) -> Response { let method = request.method().as_str().to_string(); - let route = observability_route(request.uri().path()); + let route = observability_route(&request); let scheme = resolve_request_scheme(request.headers()); let path = request.uri().path().to_string(); let request_id = resolve_request_id(&request).unwrap_or_else(|| "unknown".to_string()); let base_labels = http_base_labels(method.clone(), route.clone()); - let metrics = http_metrics(); - metrics.in_flight.add(1, &base_labels); + let in_flight = RequestInFlightGuard::new(&metrics.in_flight, base_labels.clone()); let started_at = std::time::Instant::now(); let response = next.run(request).await; let status = response.status().as_u16(); let status_class = status_class(status); let latency_ms = started_at.elapsed().as_millis().min(u64::MAX as u128) as u64; - let slow_request = latency_ms >= state.config.slow_request_threshold_ms; + let slow_request = latency_ms >= slow_request_threshold_ms; let labels = http_response_labels(base_labels, status); metrics.requests.add(1, &labels); metrics .duration .record(started_at.elapsed().as_secs_f64(), &labels); - metrics.in_flight.add(-1, &labels[..2]); + drop(in_flight); if slow_request { warn!( @@ -201,12 +215,32 @@ fn track_response_body_in_flight(response: Response) -> Response { }) } +#[derive(Clone)] struct HttpMetrics { requests: Counter, in_flight: opentelemetry::metrics::UpDownCounter, duration: opentelemetry::metrics::Histogram, } +// 请求 Future 被取消或 panic unwind 时也必须释放计数；响应体存活由另一 guard 统计。 +struct RequestInFlightGuard<'a> { + counter: &'a opentelemetry::metrics::UpDownCounter, + labels: Vec, +} + +impl<'a> RequestInFlightGuard<'a> { + fn new(counter: &'a opentelemetry::metrics::UpDownCounter, labels: Vec) -> Self { + counter.add(1, &labels); + Self { counter, labels } + } +} + +impl Drop for RequestInFlightGuard<'_> { + fn drop(&mut self) { + self.counter.add(-1, &self.labels); + } +} + #[cfg(any())] struct PuzzleGalleryCacheMetrics { hits: Counter, @@ -467,7 +501,11 @@ fn status_class(status: u16) -> &'static str { } } -pub(crate) fn observability_route(path: &str) -> String { +pub(crate) fn observability_route(request: &Request) -> String { + if let Some(path) = request.extensions().get::() { + return path.as_str().to_string(); + } + let path = request.uri().path(); if path.starts_with("/admin/api/") { "/admin/api/*".to_string() } else if path.starts_with("/api/") { @@ -490,22 +528,232 @@ pub(crate) fn resolve_request_scheme(headers: &HeaderMap) -> String { #[cfg(test)] mod tests { - use axum::http::{HeaderMap, HeaderValue}; + use axum::{ + Router, + body::{Body, Bytes}, + http::{HeaderMap, HeaderValue, Request, StatusCode}, + middleware, + routing::get, + }; + use http_body_util::BodyExt; + use opentelemetry::{ + KeyValue, + metrics::{Counter, Histogram, SyncInstrument, UpDownCounter}, + }; + use std::{ + convert::Infallible, + sync::{Arc, Mutex}, + time::Duration, + }; + use tokio::sync::Notify; + use tower::ServiceExt; - use super::{observability_route, resolve_request_scheme}; + use super::{HttpMetrics, observability_route, observe_http_request, resolve_request_scheme}; + + #[derive(Default)] + struct Measurements { + values: Mutex)>>, + } + + impl SyncInstrument for Measurements { + fn measure(&self, value: T, attributes: &[KeyValue]) { + self.values + .lock() + .expect("measurement lock") + .push((value, attributes.to_vec())); + } + } + + fn observed_router(router: Router) -> (Router, Arc>) { + let in_flight = Arc::new(Measurements::default()); + let metrics = HttpMetrics { + requests: Counter::new(Arc::new(Measurements::::default())), + in_flight: UpDownCounter::new(in_flight.clone()), + duration: Histogram::new(Arc::new(Measurements::::default())), + }; + let router = router.layer(middleware::from_fn(move |request, next| { + let metrics = metrics.clone(); + async move { observe_http_request(&metrics, u64::MAX, request, next).await } + })); + (router, in_flight) + } + + fn request(uri: &str) -> Request { + Request::builder() + .uri(uri) + .body(Body::empty()) + .expect("request") + } + + fn assert_request_finished(measurements: &Measurements) { + let values = measurements.values.lock().expect("measurement lock"); + assert_eq!( + values.iter().map(|(value, _)| *value).collect::>(), + vec![1, -1] + ); + assert_eq!( + values[0].1, values[1].1, + "decrement must use the original labels" + ); + } + + #[tokio::test] + async fn cancelled_request_releases_in_flight_measurement() { + let entered = Arc::new(Notify::new()); + let handler_entered = entered.clone(); + let (router, measurements) = observed_router(Router::new().route( + "/api/pending/{id}", + get(move || { + let entered = handler_entered.clone(); + async move { + entered.notify_one(); + std::future::pending::().await + } + }), + )); + let task = tokio::spawn(router.oneshot(request("/api/pending/123"))); + tokio::time::timeout(Duration::from_secs(5), entered.notified()) + .await + .expect("handler entered"); + assert_eq!( + measurements.values.lock().expect("measurement lock")[0].0, + 1 + ); + + task.abort(); + assert!( + task.await + .expect_err("request should be cancelled") + .is_cancelled() + ); + assert_request_finished(&measurements); + } + + #[tokio::test] + async fn panicking_request_releases_in_flight_measurement() { + async fn panic_handler() -> StatusCode { + panic!("handler panic for cancellation cleanup test"); + } + let (router, measurements) = + observed_router(Router::new().route("/panic", get(panic_handler))); + let task = tokio::spawn(router.oneshot(request("/panic"))); + assert!(task.await.expect_err("handler should panic").is_panic()); + assert_request_finished(&measurements); + } + + #[tokio::test] + async fn responses_release_in_flight_once_for_success_and_errors() { + for status in [ + StatusCode::OK, + StatusCode::UNAUTHORIZED, + StatusCode::INTERNAL_SERVER_ERROR, + ] { + let (router, measurements) = observed_router( + Router::new().route("/response", get(move || async move { status })), + ); + let response = router + .oneshot(request("/response")) + .await + .expect("response"); + assert_eq!(response.status(), status); + assert_request_finished(&measurements); + drop(response); + assert_request_finished(&measurements); + } + } + + #[tokio::test] + async fn streaming_response_releases_request_before_body_completion() { + use futures_util::StreamExt; + + let (router, measurements) = observed_router(Router::new().route( + "/events", + get(|| async { + let stream = futures_util::stream::iter([Ok::<_, Infallible>(Bytes::from_static( + b"data: ready\n\n", + ))]) + .chain(futures_util::stream::pending()); + ( + [("content-type", "text/event-stream")], + Body::from_stream(stream), + ) + }), + )); + let mut response = router + .oneshot(request("/events")) + .await + .expect("streaming response"); + assert_request_finished(&measurements); + let frame = response + .body_mut() + .frame() + .await + .expect("first frame") + .expect("body frame"); + assert_eq!(frame.into_data().expect("data frame"), "data: ready\n\n"); + drop(response); + assert_request_finished(&measurements); + } + + #[tokio::test] + async fn matched_routes_share_templates_and_preserve_distinct_endpoints() { + let routes = Router::new().nest( + "/api", + Router::new() + .route("/projects/{project_id}", get(|| async {})) + .route("/assets/{asset_id}", get(|| async {})), + ); + let (router, measurements) = observed_router(routes); + for (uri, template) in [ + ( + "/api/projects/project-1?cursor=private", + "/api/projects/{project_id}", + ), + ("/api/projects/project-2", "/api/projects/{project_id}"), + ("/api/assets/asset-1", "/api/assets/{asset_id}"), + ("/api/missing/private-id?token=private", "/api/*"), + ] { + let response = router + .clone() + .oneshot(request(uri)) + .await + .expect("response"); + drop(response); + assert_request_finished(&measurements); + let mut values = measurements.values.lock().expect("measurement lock"); + assert_eq!( + values[0].1, + vec![ + KeyValue::new("http.request.method", "GET"), + KeyValue::new("http.route", template) + ] + ); + values.clear(); + } + } #[test] fn observability_route_keeps_metrics_labels_low_cardinality() { assert_eq!( - observability_route("/api/editor/showcase/resources?cursor=abc"), + observability_route(&request("/api/editor/showcase/resources?cursor=abc")), "/api/*" ); assert_eq!( - observability_route("/api/editor/projects/project-1"), + observability_route(&request("/api/editor/projects/project-1")), "/api/*" ); - assert_eq!(observability_route("/api/runtime/settings"), "/api/*"); - assert_eq!(observability_route("/admin/api/debug/http"), "/admin/api/*"); + assert_eq!( + observability_route(&request("/api/runtime/settings")), + "/api/*" + ); + assert_eq!( + observability_route(&request("/admin/api/debug/http")), + "/admin/api/*" + ); + assert_eq!( + observability_route(&request("/missing/private-id")), + "other" + ); } #[test] diff --git a/server-rs/crates/platform-llm/Cargo.toml b/server-rs/crates/platform-llm/Cargo.toml index 312631f45..b0fa9e34a 100644 --- a/server-rs/crates/platform-llm/Cargo.toml +++ b/server-rs/crates/platform-llm/Cargo.toml @@ -11,6 +11,8 @@ reqwest = { workspace = true, features = ["json", "rustls-tls-native-roots", "st serde = { workspace = true } serde_json = { workspace = true } tokio = { workspace = true, features = ["time"] } +tracing = { workspace = true } [dev-dependencies] -tokio = { workspace = true, features = ["macros", "rt"] } +tokio = { workspace = true, features = ["macros", "rt", "sync"] } +tracing-subscriber = { workspace = true } diff --git a/server-rs/crates/platform-llm/src/lib.rs b/server-rs/crates/platform-llm/src/lib.rs index 8b87ff6bb..5f7180d33 100644 --- a/server-rs/crates/platform-llm/src/lib.rs +++ b/server-rs/crates/platform-llm/src/lib.rs @@ -18,6 +18,9 @@ use tokio::time::sleep; mod provider_adapter; pub mod router_billing; +#[cfg(test)] +mod observability_tests; + pub use provider_adapter::{ ANTHROPIC_PROVIDER_INSTANCE_ID, ANTHROPIC_PROVIDER_PROTOCOL_ID, AnthropicProviderAdapter, OPENAI_CHAT_PROVIDER_INSTANCE_ID, OPENAI_CHAT_PROVIDER_PROTOCOL_ID, @@ -1640,6 +1643,16 @@ impl LlmClient { self } + #[tracing::instrument( + name = "llm.request", + skip_all, + fields( + provider = self.config.provider().as_str(), + operation = "run", + api_kind = request.api_kind.as_str(), + model = request.resolved_model(self.config.model()), + ) + )] pub async fn run(&self, request: LlmRunRequest) -> Result { request.validate()?; let resolved_model = request.resolved_model(self.config.model()).to_string(); @@ -1687,6 +1700,16 @@ impl LlmClient { .await } + #[tracing::instrument( + name = "llm.request", + skip_all, + fields( + provider = self.config.provider().as_str(), + operation = "stream_run", + api_kind = request.api_kind.as_str(), + model = request.resolved_model(self.config.model()), + ) + )] pub async fn stream_run( &self, request: LlmRunRequest, diff --git a/server-rs/crates/platform-llm/src/observability_tests.rs b/server-rs/crates/platform-llm/src/observability_tests.rs new file mode 100644 index 000000000..16cb87dfc --- /dev/null +++ b/server-rs/crates/platform-llm/src/observability_tests.rs @@ -0,0 +1,332 @@ +use std::{ + collections::BTreeMap, + io::{Read, Write}, + net::TcpListener, + sync::{Arc, Mutex, mpsc}, + thread, + time::Duration, +}; + +use tokio::sync::oneshot; +use tracing::{ + Instrument, Subscriber, + field::{Field, Visit}, + instrument::WithSubscriber, + span::{Attributes, Id, Record}, +}; +use tracing_subscriber::{Layer, layer::Context, prelude::*, registry::LookupSpan}; + +use super::{LlmClient, LlmConfig, LlmError, LlmProvider, LlmRunRequest}; + +const PRIVATE_INPUT: &str = "PRIVATE_MESSAGE_MUST_NOT_ENTER_SPAN"; +const PRIVATE_KEY: &str = "PRIVATE_API_KEY_MUST_NOT_ENTER_SPAN"; + +#[derive(Clone, Debug, Default)] +struct CapturedSpan { + name: String, + parent: Option, + fields: BTreeMap, + closed: bool, +} + +#[derive(Clone, Default)] +struct Capture { + spans: Arc>>, + delta_parents: Arc>>, +} + +struct Fields<'a>(&'a mut BTreeMap); + +impl Visit for Fields<'_> { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.0.insert(field.name().into(), format!("{value:?}")); + } + + fn record_str(&mut self, field: &Field, value: &str) { + self.0.insert(field.name().into(), value.into()); + } +} + +impl LookupSpan<'a>> Layer for Capture { + fn on_new_span(&self, attributes: &Attributes<'_>, id: &Id, context: Context<'_, S>) { + let span = context.span(id).expect("registered span"); + let mut captured = CapturedSpan { + name: attributes.metadata().name().into(), + parent: span.parent().map(|parent| parent.id().into_u64()), + ..CapturedSpan::default() + }; + attributes.record(&mut Fields(&mut captured.fields)); + self.spans.lock().unwrap().insert(id.into_u64(), captured); + } + + fn on_record(&self, id: &Id, values: &Record<'_>, _: Context<'_, S>) { + let mut spans = self.spans.lock().unwrap(); + values.record(&mut Fields( + &mut spans.get_mut(&id.into_u64()).unwrap().fields, + )); + } + + fn on_close(&self, id: Id, _: Context<'_, S>) { + self.spans + .lock() + .unwrap() + .get_mut(&id.into_u64()) + .unwrap() + .closed = true; + } + + fn on_event(&self, event: &tracing::Event<'_>, context: Context<'_, S>) { + if event.metadata().target() == "llm_observability_test_delta" { + let parent = context + .event_span(event) + .expect("delta should have a parent"); + self.delta_parents + .lock() + .unwrap() + .push(parent.id().into_u64()); + } + } +} + +impl Capture { + fn provider_span(&self) -> (u64, CapturedSpan) { + let spans = self.spans.lock().unwrap(); + let provider_spans = spans + .iter() + .filter(|(_, span)| span.name == "llm.request") + .collect::>(); + assert_eq!(provider_spans.len(), 1); + let (id, span) = provider_spans[0]; + (*id, span.clone()) + } + + fn assert_completed(&self, operation: &str) { + let (_, span) = self.provider_span(); + assert!( + span.closed, + "provider span must close after completion or cancellation" + ); + assert_eq!( + span.fields.get("operation").map(String::as_str), + Some(operation) + ); + assert_eq!( + span.fields.get("provider").map(String::as_str), + Some("openai_compatible") + ); + assert_eq!( + span.fields.get("api_kind").map(String::as_str), + Some("openai_chat") + ); + assert_eq!( + span.fields.get("model").map(String::as_str), + Some("requested-model") + ); + let spans = self.spans.lock().unwrap(); + assert_eq!( + spans[&span.parent.expect("HTTP/application parent")].name, + "test.request" + ); + let captured = format!("{spans:?}"); + assert!(!captured.contains(PRIVATE_INPUT)); + assert!(!captured.contains(PRIVATE_KEY)); + assert!(!span.fields.contains_key("self")); + assert!(!span.fields.contains_key("request")); + } +} + +struct ProviderFixture { + client: LlmClient, + entered: oneshot::Receiver<()>, + release: mpsc::Sender<()>, + server: thread::JoinHandle<()>, +} + +// 使用本地替代上游和通道控制响应时点，不依赖实际 Provider 或计时猜测。 +fn provider_fixture(status: &str, content_type: &str, body: &str) -> ProviderFixture { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let response = format!( + "HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + let (entered, entered_rx) = oneshot::channel(); + let (release, release_rx) = mpsc::channel(); + let server = thread::spawn(move || { + listener.set_nonblocking(true).unwrap(); + let deadline = std::time::Instant::now() + Duration::from_secs(10); + let mut stream = loop { + match listener.accept() { + Ok((stream, _)) => break stream, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + assert!( + std::time::Instant::now() < deadline, + "request did not arrive" + ); + thread::sleep(Duration::from_millis(5)); + } + Err(error) => panic!("fixture accept: {error}"), + } + }; + stream.set_nonblocking(false).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut bytes = Vec::new(); + let mut chunk = [0; 4096]; + loop { + let count = stream.read(&mut chunk).unwrap(); + assert_ne!(count, 0); + bytes.extend_from_slice(&chunk[..count]); + if let Some(headers_end) = bytes.windows(4).position(|part| part == b"\r\n\r\n") { + let headers = String::from_utf8_lossy(&bytes[..headers_end]); + let length = headers + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().unwrap()) + }) + .unwrap_or(0); + if bytes.len() >= headers_end + 4 + length { + break; + } + } + } + entered.send(()).unwrap(); + release_rx.recv_timeout(Duration::from_secs(10)).unwrap(); + // 取消用例已关闭客户端连接，允许写回失败。 + let _ = stream.write_all(response.as_bytes()); + }); + let config = LlmConfig::new( + LlmProvider::OpenAiCompatible, + format!("http://{address}"), + PRIVATE_KEY.into(), + "default-model".into(), + 5_000, + 0, + 1, + ) + .unwrap(); + ProviderFixture { + client: LlmClient::new(config).unwrap(), + entered: entered_rx, + release, + server, + } +} + +fn request() -> LlmRunRequest { + LlmRunRequest::single_turn("system", PRIVATE_INPUT) + .with_openai_chat() + .with_model("requested-model") +} + +#[tokio::test] +async fn provider_span_covers_awaited_execution_and_keeps_parent_without_arguments() { + let fixture = provider_fixture( + "200 OK", + "application/json", + r#"{"choices":[{"message":{"content":"completed"},"finish_reason":"stop"}]}"#, + ); + let capture = Capture::default(); + let response = + async { + async { + let mut future = Box::pin(fixture.client.run(request())); + assert!(capture.spans.lock().unwrap().values().all(|span| span.name != "llm.request")); + tokio::select! { + response = &mut future => panic!("request returned before released: {response:?}"), + entered = fixture.entered => entered.unwrap(), + } + assert!(!capture.provider_span().1.closed); + // 挂起时不得把 Provider span 留在当前异步执行上下文。 + assert_eq!(tracing::Span::current().metadata().unwrap().name(), "test.request"); + fixture.release.send(()).unwrap(); + future.await.unwrap() + }.instrument(tracing::info_span!("test.request")).await + } + .with_subscriber(tracing_subscriber::registry().with(capture.clone())) + .await; + fixture.server.join().unwrap(); + assert_eq!(response.text, "completed"); + capture.assert_completed("run"); +} + +#[tokio::test] +async fn stream_callbacks_inherit_provider_span_and_keep_result() { + let fixture = provider_fixture( + "200 OK", + "text/event-stream", + "data: {\"choices\":[{\"delta\":{\"content\":\"hello\"},\"finish_reason\":null}]}\n\ndata: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n", + ); + let capture = Capture::default(); + let mut deltas = Vec::new(); + let response = + async { + async { + let mut future = Box::pin(fixture.client.stream_run(request(), |delta| { + tracing::info!(target: "llm_observability_test_delta", "delta received"); + deltas.push(delta.delta_text.clone()); + })); + tokio::select! { + response = &mut future => panic!("stream returned before released: {response:?}"), + entered = fixture.entered => entered.unwrap(), + } + fixture.release.send(()).unwrap(); + future.await.unwrap() + }.instrument(tracing::info_span!("test.request")).await + } + .with_subscriber(tracing_subscriber::registry().with(capture.clone())) + .await; + fixture.server.join().unwrap(); + assert_eq!(response.text, "hello"); + assert_eq!(deltas.concat(), "hello"); + capture.assert_completed("stream_run"); + let id = capture.provider_span().0; + let parents = capture.delta_parents.lock().unwrap(); + assert!(!parents.is_empty()); + assert!(parents.iter().all(|parent| *parent == id)); +} + +#[tokio::test] +async fn provider_span_preserves_upstream_errors_and_closes_on_cancellation() { + let fixture = provider_fixture( + "401 Unauthorized", + "application/json", + r#"{"error":{"message":"upstream-rejected"}}"#, + ); + let capture = Capture::default(); + let error = async { + async { + fixture.release.send(()).unwrap(); + fixture.client.run(request()).await.unwrap_err() + } + .instrument(tracing::info_span!("test.request")) + .await + } + .with_subscriber(tracing_subscriber::registry().with(capture.clone())) + .await; + fixture.server.join().unwrap(); + assert!( + matches!(error, LlmError::Upstream { status_code: 401, message } if message.contains("upstream-rejected")) + ); + capture.assert_completed("run"); + + let fixture = provider_fixture("200 OK", "application/json", "{}"); + let capture = Capture::default(); + async { + async { + let mut future = Box::pin(fixture.client.run(request())); + tokio::select! { + response = &mut future => panic!("request returned before cancellation: {response:?}"), + entered = fixture.entered => entered.unwrap(), + } + assert!(!capture.provider_span().1.closed); + drop(future); + fixture.release.send(()).unwrap(); + }.instrument(tracing::info_span!("test.request")).await + }.with_subscriber(tracing_subscriber::registry().with(capture.clone())).await; + fixture.server.join().unwrap(); + capture.assert_completed("run"); +}