feat(creation): 新增 patbond-creation 模块骨架——端点、租约队列 Worker、provider 适配层(T4-02)
CI / backend-test (push) Successful in 5m43s

M4 新模块,:8085,前缀 /api/v1/creation/**(ADR-023 §2,E3 拍板方案 A)。
它编排生成任务但不调用 AI 服务商——上游调用在独立的 patbond-ai(Python)
仓库,经 REST 到达。

落地内容:
- 客户端端点 4 个(创建/查询/列表/取消)+ 目录端点。创建强制 Idempotency-Key
  (ADR-019:生成在上游要花钱,重试绝不能变成第二个计费任务)
- 租约队列仓储:FOR UPDATE SKIP LOCKED 领取 + 租约续期 + 回收器。目标模型
  只给了索引没给回收器代码,这里补上
- Worker 三条独立调度通道(提交/轮询/回收),任一卡住不饿死其余两条。
  轮询与续租在同一语句内完成——分开做会让长任务在跑时租约过期被抢,
  那正是重复扣费的经典路径
- provider 适配层接口按「两阶段 submit/poll」定形 + 两个实现位
- 服务身份 JWT 签发器(独立密钥对)
- common 增 3 个错误码(40407/40408/42206);配额与限流的两个 429 码随
  配额工单落地,此处不加死枚举

关键实现纪律(注释中已逐条说明原因):
- provider_request_id 在上游产生任何计费动作之前落库。顺序反过来,每次
  超时都会变成重复扣费
- 租约回收时刻意不清 provider_request_id:保留它,下次尝试才知道「已提交过,
  该轮询而非重发」
- provider 按 provider_code_snapshot 而非当前目录解析,故目录改指向后
  在途任务仍由原适配器轮询
- Worker 各 pass 捕获异常:Spring 会取消抛异常的 fixedDelay 任务,
  不捕获等于一次偶发失败静默停掉整个队列

刻意未实现且会响亮失败(不装作能用):provider 两个实现位(待服务商文档
确认幂等键支持、查询接口与频率限制、单任务最长时长、产物返回形式、
限流与配额响应形态五项)、预签名上传目标获取、产物入 media.assets
(后两者取决于挂起中的对象存储决策,ADR-023 §7)。

共享物按 ADR-023 §2 复制 3 件(JWT 校验栈、UuidV7、RequestHashes);
**存储适配层不复制**——presignPut 已存在,产物走预签名上传由 patbond-user
签发,故无需第 4 份存储层,也使 patbond-ai 零对象存储凭证。

测试:新增 GenerationJobQueueIntegrationTest 11 项,全部针对「写错了也
能跑、只在生产表现为丢任务或重复扣费」的语义:SKIP LOCKED 不把同一行
交给两个 worker、轮询确实续了它依赖的租约、回收器保留 provider_request_id、
first_attempt_at 跨重排队存活、write-once 保护、取消被状态守卫。
全量 393 → 404,零失败零跳过。

过程中被测试抓出两个真问题:UPDATE...FROM 的 RETURNING 列表未限定别名
导致 id 歧义(列表已改为单一来源派生两种形式,避免漂移);以及夹具伪造
了物理上不可能的历史(租约过期但 started_at 是当下),被
ck_generation_jobs_lease 正确拒绝。

Readme.md 全文改中文并补齐 M2 之后的内容:原文停在 M2(pet 骨架阶段)、
完全没有 patbond-community、配置表缺 M2 之后全部条目。现覆盖六模块、
迁移链归属说明、两对密钥的区别与原因、以及各服务端点。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-09-18 17:36:52 +08:00
parent 7896a6d20b
commit 4b2550f88c
38 changed files with 3467 additions and 89 deletions
+139 -89
View File
@@ -1,82 +1,113 @@
# Patbond API # Patbond API
Patbond API is a Spring Boot multi-module backend. Patbond 后端,Spring Boot 多模块。
## Modules ## 模块
- `patbond-common`: stable cross-module contracts (response wrapper, shared DTOs). No web/messaging dependencies. | 模块 | 端口 | 职责 |
- `patbond-user`: user service for user creation, profile lookup, and password verification. | --- | --- | --- |
- `patbond-auth`: authentication service for registration and login. Calls `patbond-user` over HTTP via OpenFeign with a configured static URL (no service discovery in the MVP, see ADR-002). | `patbond-common` | —— | 跨模块稳定契约(响应信封、错误码、共享 DTO)。**刻意不引入 web/消息依赖**,保持纯契约。 |
- `patbond-pet`: pet profile and health record service (M2, ADR-009). Shares the database with `patbond-user` and only reads/writes the `pet_health` schema; the Flyway migration chain stays owned by `patbond-user`. First-wave skeleton: liveness endpoint only. | `patbond-auth` | 8081 | 注册与登录,签发 RS256 access token。经 OpenFeign 静态直连调用 `patbond-user`(MVP 不做服务发现,ADR-002)。**私钥只在本服务**。 |
| `patbond-user` | 8082 | 用户与会话、当前用户资料、媒体上传流程(ADR-016/017)、埋点接收端。**独占 Flyway 迁移链**。 |
| `patbond-pet` | 8083 | 宠物档案与健康记录(M2,ADR-009)。与 `patbond-user` 共库,只读写 `pet_health` schema。 |
| `patbond-community` | 8084 | 社区 Feed、帖子、评论、点赞收藏与关注(M3,ADR-017)。只读写 `community` schema,另按 ADR-017 例外只读 `media.assets`。 |
| `patbond-creation` | 8085 | **AI 生成任务编排(M4,ADR-023)**。见下节。 |
## Technology Stack ### patbond-creation(M4 新增)
- Java 17 (build baseline; use JDK 17 for release builds) 承载 AI 图片与视频生成的任务编排。**它编排任务,但不调用 AI 服务商**——上游调用在独立的 `patbond-ai`(Python)仓库,经 REST 到达。
**本模块拥有**:`creation` schema 的读写、`/api/v1/creation/**` 客户端端点、以及租约队列 Worker。
**本模块刻意不拥有**:
- **AI 服务商调用** —— 在 `patbond-ai`。Java 侧持有队列与任务状态,Python 侧只负责调厂商 API,两者经 REST(不用 gRPC)通信,鉴权用**独立密钥对签发的服务身份 JWT**(不转发用户 token)。
- **对象存储** —— 本服务从不直连。产物上传凭证来自 `patbond-user` 的内部媒体接口(预签名 PUT),由生成方直传,ADR-017 的边界不动。
**队列为什么在数据库里而不是消息中间件**(ADR-023 §3):已评审的目标数据模型本就把整套租约机制建成了列(`lease_owner`/`lease_expires_at`/`next_attempt_at`/`attempt_count`/`max_attempts`/`priority`),外加两条列序与出队语句逐列对齐的偏索引。因此领取是索引扫描而非排序,任务状态只有一个事实来源、无双写;引入中间件反而要让这批列作废。「Worker 重启不丢任务」由租约回收器天然满足。
**Worker 的三条独立调度通道**(任一卡住不会饿死其余两条):领取待办 QUEUED 任务并提交上游 → 轮询本实例持有的 RUNNING 任务并在同一语句内续租 → 回收租约过期的任务重新入队。
两处值得留意的实现纪律:
- **`provider_request_id` 在上游产生任何计费动作之前落库**。顺序反过来,每次超时都会变成重复扣费。
- **租约回收时不清 `provider_request_id`**。保留它,下次尝试才知道「这活已经提交过了,该去轮询而不是重发」。
**骨架阶段刻意未实现**且会响亮失败(不装作能用):provider 适配层的两个实现位、预签名上传目标获取、产物入 `media.assets`。前者待服务商官方文档确认五项(幂等键支持、查询接口与频率限制、单任务最长时长、产物返回形式、限流与配额响应形态),后两者取决于**对象存储方案,该决策明确挂起**(ADR-023 §7)。
## 技术栈
- Java 17(构建基线;发布构建请用 JDK 17)
- Spring Boot 3.5.16 - Spring Boot 3.5.16
- Spring Cloud 2025.0.3 (OpenFeign only) - Spring Cloud 2025.0.3(只用 OpenFeign)
- PostgreSQL 18 + Flyway (patbond-user owns the migration chain: `identity`/`media`/`platform`/`pet_health` schemas) - PostgreSQL 18 + Flyway
- Maven (use the committed Maven Wrapper `./mvnw`) - Maven(用仓库内的 wrapper `./mvnw`)
## Build and Test **Flyway 迁移链由 `patbond-user` 独占**,V1..V6 覆盖 `identity`/`media`/`platform`(V1)、`pet_health`(V3/V4)、`community`(V5)、`creation`(V6)。注意「谁拥有 schema」与「迁移文件放哪」在本项目是分开的:pet、community、creation 的业务代码各在自己的模块,迁移文件全部在 `patbond-user`,`flyway_schema_history` 只有一份、不拆。
## 构建与测试
```bash ```bash
./mvnw clean test ./mvnw clean test
``` ```
Integration tests start a disposable `postgres:18` via Testcontainers, so a 集成测试经 Testcontainers 拉起一次性的 `postgres:18`,因此需要一个在跑的 Docker daemon(不需要本机装 PostgreSQL、也不需要本机库凭证)。
running Docker daemon is required (no local PostgreSQL installation or
credentials are needed).
If your default JDK is not 17, point `JAVA_HOME` at a JDK 17 installation first, e.g.: 默认 JDK 不是 17 时,先指向一个 JDK 17:
```bash ```bash
JAVA_HOME=/usr/lib/jvm/java-17-openjdk ./mvnw clean test JAVA_HOME=/usr/lib/jvm/java-17-openjdk ./mvnw clean test
``` ```
## Run ## 本机运行
Running `patbond-user` requires a reachable PostgreSQL 18 database; Flyway `patbond-user` 启动时 Flyway 会自动应用 `patbond-user/src/main/resources/db/migration` 下的版本化迁移,因此它需要一个可达的 PostgreSQL 18。开发种子数据(`db/dev`)通过 dev profile 可选启用,见 `application.yml.sample`。
applies the versioned migrations in
`patbond-user/src/main/resources/db/migration` automatically on startup.
Development seed data (`db/dev`, regions reference rows) is opt-in via a dev
profile — see `application.yml.sample`.
Each service ships a committed `application.yml.sample`; the real `application.yml` 每个服务都提交了 `application.yml.sample`,真实的 `application.yml` 被 gitignore。先复制样例(默认值本机可用,实际值用环境变量覆盖):
is git-ignored. First copy the samples (defaults work locally, overrides via
environment variables):
```bash ```bash
cp patbond-user/src/main/resources/application.yml.sample \ for m in user auth pet community creation; do
patbond-user/src/main/resources/application.yml cp patbond-$m/src/main/resources/application.yml.sample \
cp patbond-auth/src/main/resources/application.yml.sample \ patbond-$m/src/main/resources/application.yml
patbond-auth/src/main/resources/application.yml done
cp patbond-pet/src/main/resources/application.yml.sample \
patbond-pet/src/main/resources/application.yml
``` ```
Install the shared module once, then run each service in its own terminal: 先装一次共享模块,然后各服务各开一个终端:
```bash ```bash
./mvnw -pl patbond-common install ./mvnw -pl patbond-common install
./mvnw -pl patbond-user spring-boot:run ./mvnw -pl patbond-user spring-boot:run
./mvnw -pl patbond-auth spring-boot:run ./mvnw -pl patbond-auth spring-boot:run
./mvnw -pl patbond-pet spring-boot:run
./mvnw -pl patbond-community spring-boot:run
./mvnw -pl patbond-creation spring-boot:run
``` ```
Running the services also needs an RS256 key pair for access tokens (the ### 密钥对
private key signs in `patbond-auth`, the public key verifies in
`patbond-user`; neither is committed): 运行需要一对 RS256 密钥用于 access token(`patbond-auth` 用私钥签,其余服务用公钥验;两者都不入库):
```bash ```bash
openssl genpkey -algorithm RSA -pkeyopt rsa_keygen_bits:2048 -out jwt-private.pem openssl genpkey -algorithm RSA -pkeyopt rsa_keygen_bits:2048 -out jwt-private.pem
openssl pkey -in jwt-private.pem -pubout -out jwt-public.pem openssl pkey -in jwt-private.pem -pubout -out jwt-public.pem
export PATBOND_JWT_PRIVATE_KEY=$PWD/jwt-private.pem # patbond-auth export PATBOND_JWT_PRIVATE_KEY=$PWD/jwt-private.pem # 只给 patbond-auth
export PATBOND_JWT_PUBLIC_KEY=$PWD/jwt-public.pem # patbond-user export PATBOND_JWT_PUBLIC_KEY=$PWD/jwt-public.pem # 其余服务验签用
``` ```
Tests do not require this copy step — `patbond-auth/src/test/resources/application.yml` **`patbond-creation` 调用 `patbond-ai` 需要另一对独立的密钥**,不要复用上面那对:
keeps `./mvnw clean test` self-contained on a clean checkout.
Smoke check: ```bash
openssl genpkey -algorithm RSA -pkeyopt rsa_keygen_bits:2048 -out service-jwt-private.pem
openssl pkey -in service-jwt-private.pem -pubout -out service-jwt-public.pem
export PATBOND_SERVICE_JWT_PRIVATE_KEY=$PWD/service-jwt-private.pem # patbond-creation
# service-jwt-public.pem 交给 patbond-ai 验签
```
**为什么必须独立**(ADR-023 §5):现有 `JwtVerifier` **只验签名,既不校验 `aud` 也不校验 `iss`**,而 `patbond-auth` 签发的用户 token **不含 `aud` claim**。若服务票复用同一密钥对,user/pet/community/creation 四个模块都会把它当合法用户 token 接受——`sub` 不是 UUID 大概会在后续解析时炸掉,但那是**意外挡住的,不是设计挡住的**。用独立密钥对则无需改动任何已发布的验证器。
测试不需要上述复制与密钥步骤:`patbond-auth/src/test/resources/application.yml` 让 `./mvnw clean test` 在干净 checkout 上自给自足。
冒烟:
```bash ```bash
curl -X POST http://127.0.0.1:8081/api/v1/auth/register \ curl -X POST http://127.0.0.1:8081/api/v1/auth/register \
@@ -84,36 +115,44 @@ curl -X POST http://127.0.0.1:8081/api/v1/auth/register \
-d '{"username":"demo_user","password":"secret123"}' -d '{"username":"demo_user","password":"secret123"}'
``` ```
### Configuration ### 配置
| Environment variable | Default | Used by | | 环境变量 | 默认值 | 使用方 |
| --- | --- | --- | | --- | --- | --- |
| `PATBOND_USER_PORT` | `8082` | patbond-user |
| `PATBOND_AUTH_PORT` | `8081` | patbond-auth | | `PATBOND_AUTH_PORT` | `8081` | patbond-auth |
| `PATBOND_USER_SERVICE_URL` | `http://127.0.0.1:8082` | patbond-auth (Feign target for patbond-user) | | `PATBOND_USER_PORT` | `8082` | patbond-user |
| `PATBOND_DB_URL` | `jdbc:postgresql://127.0.0.1:5432/patbond` | patbond-user | | `PATBOND_PET_PORT` | `8083` | patbond-pet |
| `PATBOND_DB_USER` | `patbond` | patbond-user | | `PATBOND_COMMUNITY_PORT` | `8084` | patbond-community |
| `PATBOND_DB_PASSWORD` | `patbond` | patbond-user | | `PATBOND_CREATION_PORT` | `8085` | patbond-creation |
| `PATBOND_INTERNAL_TOKEN` | `dev-only-internal-token` | both (shared secret for `/internal/**`; inject a strong random value outside local dev) | | `PATBOND_DB_URL` | `jdbc:postgresql://127.0.0.1:5432/patbond` | 全部数据库使用方 |
| `PATBOND_JWT_PRIVATE_KEY` | _(none, required)_ | patbond-auth (RS256 private key: PEM path or inline PEM) | | `PATBOND_DB_USER` | `patbond` | 同上 |
| `PATBOND_JWT_PUBLIC_KEY` | _(none)_ | patbond-user (RS256 public key: PEM path or inline PEM) | | `PATBOND_DB_PASSWORD` | `patbond` | 同上 |
| `PATBOND_ACCESS_TTL` | `15m` | patbond-auth (access token lifetime, ADR-003) | | `PATBOND_USER_SERVICE_URL` | `http://127.0.0.1:8082` | auth / community / creation(Feign 指向 patbond-user) |
| `PATBOND_REFRESH_TTL` | `30d` | patbond-user (refresh token lifetime, ADR-003) | | `PATBOND_INTERNAL_TOKEN` | `dev-only-internal-token` | 全部(`/internal/**` 服务间共享密钥;本机开发之外必须注入强随机值) |
| `PATBOND_LOGIN_LOCK_MAX_FAILURES` | `5` | patbond-user (login failures before lockout) | | `PATBOND_JWT_PRIVATE_KEY` | _(无,必填)_ | patbond-auth(RS256 私钥:PEM 路径或内联 PEM) |
| `PATBOND_LOGIN_LOCK_WINDOW` | `15m` | patbond-user (failure counting window) | | `PATBOND_JWT_PUBLIC_KEY` | _(无)_ | user / pet / community / creation(RS256 公钥) |
| `PATBOND_LOGIN_LOCK_DURATION` | `15m` | patbond-user (lock duration) | | `PATBOND_ACCESS_TTL` | `15m` | patbond-auth(access token 有效期,ADR-003) |
| `PATBOND_REFRESH_TTL` | `30d` | patbond-user(refresh token 有效期,ADR-003) |
| `PATBOND_LOGIN_LOCK_MAX_FAILURES` | `5` | patbond-user(锁定前允许的失败次数) |
| `PATBOND_LOGIN_LOCK_WINDOW` | `15m` | patbond-user(失败计数窗口) |
| `PATBOND_LOGIN_LOCK_DURATION` | `15m` | patbond-user(锁定时长) |
| `PATBOND_MINIO_ENDPOINT` | `http://minio:9000` | patbond-user(服务自身访问对象存储的地址) |
| `PATBOND_MINIO_PUBLIC_ENDPOINT` | `http://127.0.0.1:9000` | user / community(**客户端可达**地址,预签名 URL 的 host 由它决定;容器内外不同) |
| `PATBOND_MINIO_ACCESS_KEY` / `_SECRET_KEY` | _(无)_ | user / community |
| `PATBOND_MINIO_BUCKET` | `patbond-media` | user / community |
| `PATBOND_AI_SERVICE_URL` | `http://127.0.0.1:8000` | patbond-creation(指向 patbond-ai) |
| `PATBOND_SERVICE_JWT_PRIVATE_KEY` | _(无)_ | patbond-creation(**独立于用户 token 的** RS256 私钥) |
| `PATBOND_CREATION_WORKER_ENABLED` | `true` | patbond-creation(关掉则只提供端点、不消费队列) |
| `PATBOND_CREATION_LEASE_TTL` | `60s` | patbond-creation(**必须大于最长轮询间隔**,否则任务在跑租约就过期被抢) |
Machine-specific values live in the git-ignored `application.yml` (copied from the 各服务的 `application.yml.sample` 是环境变量的完整清单与说明,本表只列常用项。机器相关的实际值放在被 gitignore 的 `application.yml` 里;**样例文件里永远不要提交真实密钥**。
committed `.sample`); never commit secrets to the samples.
## Docker Compose ## Docker Compose
MVP 编排(ADR-007):`postgres:18`(数据落 volume)+ 两个无状态应用容器。 编排(ADR-007/016):`postgres:18` + 自托管 `minio` + 无状态应用容器。配置与本机运行同一套约定 —— 容器内挂载 `application.yml.sample` 作为配置,`PATBOND_*` 环境变量注入实际值。
配置与本机运行同一套约定 —— 容器内挂载 `application.yml.sample` 作为配置,
`PATBOND_*` 环境变量注入实际值。
```bash ```bash
# 1. 生成 RS256 密钥对与 .env(DB 口令、内部令牌;产物被 .gitignore 忽略) # 1. 生成 RS256 密钥对与 .env(DB 口令、内部令牌、MinIO 凭证;产物被 gitignore)
./deploy/init-secrets.sh ./deploy/init-secrets.sh
# 2. 构建可执行 jar # 2. 构建可执行 jar
@@ -127,40 +166,51 @@ curl -s -X POST http://127.0.0.1:8081/api/v1/auth/register \
-H 'Content-Type: application/json' \ -H 'Content-Type: application/json' \
-d '{"username":"demo_user","password":"secret123"}' -d '{"username":"demo_user","password":"secret123"}'
# 停止(-v 同时删除数据库数据) # 停止(-v 同时删除数据库与对象存储数据)
docker compose down docker compose down
``` ```
注意:数据库端口不对宿主机发布;`8082`(user)在 MVP 阶段直连暴露以提供 `patbond-creation` 尚未加入 compose——它的 Worker 依赖 `patbond-ai` 与对象存储方案,两者都还没落地。
`/api/v1/me`,`/internal/**` 由服务间令牌保护,规模化阶段应改由网关统一入口。
## Services 注意:数据库端口不对宿主机发布;`8082`(user)在 MVP 阶段直连暴露以提供 `/api/v1/me`,`/internal/**` 由服务间令牌保护,规模化阶段应改由网关统一入口。
### Auth Service ## 服务端点
- Application name: `patbond-auth` ### Auth(8081)
- Default port: `8081`
- Register: `POST /api/v1/auth/register`
- Login: `POST /api/v1/auth/login`
- Refresh (rotates the refresh token): `POST /api/v1/auth/refresh`
- Logout (revokes the current session): `POST /api/v1/auth/logout`
### User Service - 注册:`POST /api/v1/auth/register`
- 登录:`POST /api/v1/auth/login`
- 刷新(轮换 refresh token):`POST /api/v1/auth/refresh`
- 登出(吊销当前会话):`POST /api/v1/auth/logout`
- Application name: `patbond-user` ### User(8082)
- Default port: `8082`
- Current user profile: `GET /api/v1/me` (Bearer access token, verified locally with the RS256 public key)
- Internal (require `X-Internal-Token`):
- Create user: `POST /internal/users`
- Verify password: `POST /internal/users/verify-password`
- Get user by id: `GET /internal/users/{id}`
- Get user by username: `GET /internal/users/by-username/{username}`
- Sessions: `POST /internal/sessions`, `POST /internal/sessions/refresh`, `POST /internal/sessions/revoke`
> Note: user data is persisted in PostgreSQL (`identity.users` / - 当前用户资料:`GET /api/v1/me`、`PATCH /api/v1/me`
> `identity.user_credentials`, bcrypt password hashes, UUIDv7 ids generated in - 社区统计:`GET /api/v1/me/community-stats`
> the application). Refresh sessions live in `identity.auth_sessions` (SHA-256 - 媒体上传:创建上传凭据与确认(ADR-016:私有桶 + 预签名)
> digests only, rotation on every refresh, token-family revocation on reuse, - 埋点接收:`POST /api/v1/events`(批量 1–50、逐条结果、eventId 幂等、白名单剥离)
> ADR-003). Errors follow the `{code, message, data}` envelope with stable - 内部接口(需 `X-Internal-Token`):`POST /internal/users`、`POST /internal/users/verify-password`、`GET /internal/users/{id}`、`GET /internal/users/by-username/{username}`、`GET /internal/users/profiles`、`POST /internal/sessions{,/refresh,/revoke}`
> business codes and matching HTTP statuses; the public contract is documented
> in `patbond-doc/docs/api/openapi.yaml`. ### Pet(8083)
宠物档案、体重与疫苗记录、照护提醒;权限按 ADR-015 三档模型(owner / caregiver / viewer)。
### Community(8084)
Feed 游标分页、帖子草稿与发布、单层评论、点赞与收藏(幂等)、关注与我的收藏。
### Creation(8085)
- 创建生成任务:`POST /api/v1/creation/jobs`(**必带 `Idempotency-Key`**,ADR-019——生成在上游是要花钱的,重试绝不能变成第二个计费任务)
- 查询任务:`GET /api/v1/creation/jobs/{jobId}`(**轮询端点**)
- 我的任务列表:`GET /api/v1/creation/jobs`
- 取消任务:`DELETE /api/v1/creation/jobs/{jobId}`
- 模型与风格目录:`GET /api/v1/creation/catalog`
进度靠轮询上面那个 GET,**没有流式端点**:SSE 会旁路整条「信封解包 + 401 单飞刷新重放」的客户端网络层、形成第二套鉴权路径,而任何推送方案都必须有这个 GET 兜底(ADR-023 §10)。
目录端点会返回「哪些 `media_kind` 真的有可用模型」,客户端据此决定是否渲染类型切换器——所以目录为空的类型不会出现,不会给用户一个承诺了却做不到的控件;V7 种子落地后 UI 会自己长出视频分组,无需发版。
任务响应里的 `timeoutSec` 与 `stalled` 由服务端下发而非客户端硬编码:图片与视频差一个量级,硬编码两个常量意味着换 provider 就要发版。`stalled`(第六个 UI 态)**不是错误态**——任务仍可能完成。
> 用户数据落在 PostgreSQL(`identity.users` / `identity.user_credentials`,bcrypt 口令散列,应用侧生成 UUIDv7)。refresh 会话在 `identity.auth_sessions`(只存 SHA-256 摘要,每次刷新轮换,重用即吊销整个 token family,ADR-003)。错误统一走 `{code, message, data}` 信封与稳定业务码;对外契约见 `patbond-doc/docs/api/openapi.yaml`,架构决策见 `patbond-doc/docs/architecture/decisions.md`。
@@ -28,11 +28,16 @@ public enum ErrorCode {
MEDIA_NOT_FOUND(40405, 404, "媒体资源不存在"), MEDIA_NOT_FOUND(40405, 404, "媒体资源不存在"),
COMMENT_NOT_FOUND(40404, 404, "评论不存在"), COMMENT_NOT_FOUND(40404, 404, "评论不存在"),
TARGET_USER_NOT_FOUND(40406, 404, "用户不存在"), TARGET_USER_NOT_FOUND(40406, 404, "用户不存在"),
// M4 生成域(ADR-023 §9)。40407/40408 与既有 404 族同列;配额与限流的
// 两个 429 码随配额工单落地时补入——此处只加骨架实际使用的码,避免死枚举。
GENERATION_JOB_NOT_FOUND(40407, 404, "生成任务不存在"),
GENERATION_MODEL_NOT_FOUND(40408, 404, "模型或风格不可用"),
VACCINATION_RULE_VIOLATION(42201, 422, "疫苗状态或日期约束不满足"), VACCINATION_RULE_VIOLATION(42201, 422, "疫苗状态或日期约束不满足"),
REMINDER_RULE_VIOLATION(42202, 422, "提醒状态或 completedAt 约束不满足"), REMINDER_RULE_VIOLATION(42202, 422, "提醒状态或 completedAt 约束不满足"),
MEDIA_NOT_READY(42203, 422, "媒体尚未就绪"), MEDIA_NOT_READY(42203, 422, "媒体尚未就绪"),
FOLLOW_RULE_VIOLATION(42204, 422, "不能关注自己"), FOLLOW_RULE_VIOLATION(42204, 422, "不能关注自己"),
MEDIA_UPLOAD_STATE_INVALID(42205, 422, "上传状态不允许确认"), MEDIA_UPLOAD_STATE_INVALID(42205, 422, "上传状态不允许确认"),
GENERATION_STATE_INVALID(42206, 422, "生成任务状态不允许该操作"),
LOGIN_LOCKED(42300, 423, "登录失败次数过多,账号已临时锁定"), LOGIN_LOCKED(42300, 423, "登录失败次数过多,账号已临时锁定"),
INTERNAL_ERROR(50000, 500, "服务器内部错误"), INTERNAL_ERROR(50000, 500, "服务器内部错误"),
DOWNSTREAM_UNAVAILABLE(50300, 503, "依赖服务暂不可用"); DOWNSTREAM_UNAVAILABLE(50300, 503, "依赖服务暂不可用");
+11
View File
@@ -0,0 +1,11 @@
# Runtime image only — build the jar first: ./mvnw -pl patbond-creation -am package
# Stateless by design (ADR-007): no local state, config via env / mounted files.
# Job state lives in postgres; the worker is crash-safe through the DB lease,
# so any replica may be restarted without losing in-flight work (ADR-023 §3).
FROM eclipse-temurin:17-jre
RUN useradd --system --uid 10001 patbond
USER patbond
WORKDIR /app
COPY target/patbond-creation-1.0.0-SNAPSHOT-exec.jar app.jar
EXPOSE 8085
ENTRYPOINT ["java", "-jar", "/app/app.jar"]
+148
View File
@@ -0,0 +1,148 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.patbond.patbond</groupId>
<artifactId>patbond-api</artifactId>
<version>1.0.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<artifactId>patbond-creation</artifactId>
<packaging>jar</packaging>
<name>patbond-creation</name>
<description>AI generation job orchestration service for Patbond (ADR-023)</description>
<!-- M4 skeleton: RS256 bearer auth on /api/v1/** (fourth copy of the
user/pet/community stack), creation schema access via JDBC, and the
lease queue worker. This module owns job state; the upstream provider
call lives in patbond-ai (Python) and is reached over REST.
Flyway remains absent — the migration chain is owned by patbond-user. -->
<dependencies>
<dependency>
<groupId>com.patbond.patbond</groupId>
<artifactId>patbond-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<!-- Outbound calls to patbond-ai (generation) and patbond-user
(/internal/media/** upload credentials), static direct URLs per
ADR-002. feign-hc5 for the same reason as auth/community: the JDK
default client loses error bodies on some replies. -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>
<dependency>
<groupId>io.github.openfeign</groupId>
<artifactId>feign-hc5</artifactId>
</dependency>
<dependency>
<groupId>org.postgresql</groupId>
<artifactId>postgresql</artifactId>
<scope>runtime</scope>
</dependency>
<!-- Access token verification (RS256 public key) AND service-identity
token signing (RS256 private key, a SEPARATE key pair per
ADR-023 §5). jjwt is not in the Boot BOM; version pinned in step
with patbond-user/auth/pet/community. -->
<dependency>
<groupId>io.jsonwebtoken</groupId>
<artifactId>jjwt-api</artifactId>
<version>0.12.6</version>
</dependency>
<dependency>
<groupId>io.jsonwebtoken</groupId>
<artifactId>jjwt-impl</artifactId>
<version>0.12.6</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>io.jsonwebtoken</groupId>
<artifactId>jjwt-jackson</artifactId>
<version>0.12.6</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!-- Tests need the creation schema. The migration chain (V1..V6) is
owned by patbond-user (single flyway_schema_history); pulling its
plain jar plus Flyway into the TEST classpath lets Boot's Flyway
auto-config apply the same chain to the disposable container.
Production wiring is unchanged (same mechanism as pet/community). -->
<dependency>
<groupId>com.patbond.patbond</groupId>
<artifactId>patbond-user</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.flywaydb</groupId>
<artifactId>flyway-core</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.flywaydb</groupId>
<artifactId>flyway-database-postgresql</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-testcontainers</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>postgresql</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<!-- No spring-boot-starter-parent in this build, so the
executable-jar repackaging must be bound explicitly. -->
<executions>
<execution>
<goals>
<goal>repackage</goal>
</goals>
<configuration>
<!-- Keep the plain jar as the main artifact so other
modules can depend on this one; the runnable fat
jar gets the -exec classifier and is what the
Dockerfile ships (same pattern as the others). -->
<classifier>exec</classifier>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
@@ -0,0 +1,30 @@
package com.patbond.patbond.creation;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;
/**
* AI generation orchestration service (M4, ADR-023 §2: the creation domain
* lives in its own Maven module on :8085).
*
* <p>What this service owns: the {@code creation} schema, the client-facing
* generation endpoints under {@code /api/v1/creation/**}, and the lease queue
* worker. What it does NOT own: the upstream vendor call, which lives in
* patbond-ai (Python) and is reached over REST with a service-identity JWT; and
* object storage, which it never touches — upload credentials come from
* patbond-user's internal media API, keeping the ADR-017 boundary intact.
*
* <p>{@code @EnableScheduling} drives the worker. The worker is crash-safe
* through the database lease rather than through in-process state, so replicas
* may be restarted freely; {@code patbond.creation.worker.enabled} turns
* consumption off for a request-only replica or a test.
*/
@SpringBootApplication
@EnableScheduling
public class CreationApplication {
public static void main(String[] args) {
SpringApplication.run(CreationApplication.class, args);
}
}
@@ -0,0 +1,329 @@
package com.patbond.patbond.creation.config;
import com.patbond.patbond.creation.domain.MediaKind;
import org.springframework.boot.context.properties.ConfigurationProperties;
import java.time.Duration;
/**
* Generation-specific knobs: the upstream service address, the service-identity
* token, the worker's lease and polling ladder, and the per-kind timeouts the
* server hands to clients.
*
* <p>Defaults here are the ones ADR-023 §4/§10 argues for, so a service with no
* overrides behaves as designed rather than as whatever happened to be
* convenient.
*/
@ConfigurationProperties(prefix = "patbond.creation")
public class CreationProperties {
private final AiService aiService = new AiService();
private final ServiceToken serviceToken = new ServiceToken();
private final Worker worker = new Worker();
/**
* Hard timeout handed to the client per media kind. Issued by the server
* rather than hardcoded client-side: image and video differ by an order of
* magnitude, and two hardcoded constants would mean shipping a new app
* build every time a provider changes (ADR-023 §10).
*/
private Duration clientTimeoutImage = Duration.ofMinutes(5);
private Duration clientTimeoutVideo = Duration.ofMinutes(20);
/**
* How long a RUNNING job may go before the client shows the sixth UI
* state — stalled/timed-out. That state is deliberately NOT an error state:
* the job may still complete.
*/
private Duration staleAfterImage = Duration.ofMinutes(10);
private Duration staleAfterVideo = Duration.ofMinutes(40);
public AiService getAiService() {
return aiService;
}
public ServiceToken getServiceToken() {
return serviceToken;
}
public Worker getWorker() {
return worker;
}
public Duration clientTimeoutFor(MediaKind kind) {
return kind == MediaKind.VIDEO ? clientTimeoutVideo : clientTimeoutImage;
}
public Duration staleAfterFor(MediaKind kind) {
return kind == MediaKind.VIDEO ? staleAfterVideo : staleAfterImage;
}
public Duration getClientTimeoutImage() {
return clientTimeoutImage;
}
public void setClientTimeoutImage(Duration value) {
this.clientTimeoutImage = value;
}
public Duration getClientTimeoutVideo() {
return clientTimeoutVideo;
}
public void setClientTimeoutVideo(Duration value) {
this.clientTimeoutVideo = value;
}
public Duration getStaleAfterImage() {
return staleAfterImage;
}
public void setStaleAfterImage(Duration value) {
this.staleAfterImage = value;
}
public Duration getStaleAfterVideo() {
return staleAfterVideo;
}
public void setStaleAfterVideo(Duration value) {
this.staleAfterVideo = value;
}
public static class AiService {
private String url = "http://127.0.0.1:8000";
/**
* Timeout of the SUBMIT/POLL call, not of the generation. The
* Java↔Python boundary is asynchronous regardless of whether the vendor
* is (ADR-023 §4), so this must not be inflated to "wait for the
* picture" — doing so reintroduces the held-connection and
* ambiguous-billing problems the async boundary exists to remove.
*/
private Duration connectTimeout = Duration.ofSeconds(2);
private Duration readTimeout = Duration.ofSeconds(5);
public String getUrl() {
return url;
}
public void setUrl(String url) {
this.url = url;
}
public Duration getConnectTimeout() {
return connectTimeout;
}
public void setConnectTimeout(Duration value) {
this.connectTimeout = value;
}
public Duration getReadTimeout() {
return readTimeout;
}
public void setReadTimeout(Duration value) {
this.readTimeout = value;
}
}
public static class ServiceToken {
/**
* RS256 private key, inline PEM or path — a key pair SEPARATE from the
* user-token one (ADR-023 §5). Absent, the service still starts and
* only fails when it actually needs to call patbond-ai, matching how a
* missing JWT public key is handled elsewhere.
*/
private String privateKey;
private String issuer = "patbond-api";
private String audience = "patbond-ai";
private String subject = "service:creation-worker";
/** Signed per call, never cached — see ServiceTokenSigner. */
private Duration ttl = Duration.ofMinutes(5);
public String getPrivateKey() {
return privateKey;
}
public void setPrivateKey(String value) {
this.privateKey = value;
}
public String getIssuer() {
return issuer;
}
public void setIssuer(String value) {
this.issuer = value;
}
public String getAudience() {
return audience;
}
public void setAudience(String value) {
this.audience = value;
}
public String getSubject() {
return subject;
}
public void setSubject(String value) {
this.subject = value;
}
public Duration getTtl() {
return ttl;
}
public void setTtl(Duration value) {
this.ttl = value;
}
}
public static class Worker {
/**
* When false the service still serves read/write endpoints but consumes
* no jobs. Used to keep tests off the queue and to run a
* request-serving replica without a worker.
*/
private boolean enabled = true;
/**
* Scheduler tick. The real cadence comes from {@code next_attempt_at}
* and the lease; the tick only bounds how soon due work is noticed.
*/
private Duration tickInterval = Duration.ofSeconds(3);
/** Cap per tick so one replica cannot drain the whole queue. */
private int batchSize = 5;
/**
* Lease duration. MUST exceed the longest polling interval below,
* otherwise a job still being worked on has its lease expire and gets
* reclaimed mid-flight — the classic route to paying a provider twice.
* Renewed on every poll.
*/
private Duration leaseTtl = Duration.ofSeconds(60);
/**
* Backoff ladder for polling patbond-ai. A flat 3s would mean hundreds
* of pointless polls and lease renewals over a multi-minute video job.
*/
private Duration pollIntervalEarly = Duration.ofSeconds(3);
private Duration pollIntervalMid = Duration.ofSeconds(10);
private Duration pollIntervalLate = Duration.ofSeconds(30);
private Duration pollEarlyUntil = Duration.ofSeconds(60);
private Duration pollMidUntil = Duration.ofMinutes(5);
/** Retry backoff base; actual delay is base * 2^(attempt-1). */
private Duration retryBackoffBase = Duration.ofSeconds(30);
/**
* Chooses the polling interval from how long the job has been running.
*
* @param runningFor elapsed time since the current attempt started
*/
public Duration pollIntervalFor(Duration runningFor) {
if (runningFor.compareTo(pollEarlyUntil) < 0) {
return pollIntervalEarly;
}
return runningFor.compareTo(pollMidUntil) < 0 ? pollIntervalMid : pollIntervalLate;
}
/** Exponential backoff for a retryable failure, capped at 10 minutes. */
public Duration retryBackoffFor(int attemptCount) {
long multiplier = 1L << Math.min(Math.max(attemptCount - 1, 0), 8);
Duration backoff = retryBackoffBase.multipliedBy(multiplier);
Duration cap = Duration.ofMinutes(10);
return backoff.compareTo(cap) > 0 ? cap : backoff;
}
public boolean isEnabled() {
return enabled;
}
public void setEnabled(boolean value) {
this.enabled = value;
}
public Duration getTickInterval() {
return tickInterval;
}
public void setTickInterval(Duration value) {
this.tickInterval = value;
}
public int getBatchSize() {
return batchSize;
}
public void setBatchSize(int value) {
this.batchSize = value;
}
public Duration getLeaseTtl() {
return leaseTtl;
}
public void setLeaseTtl(Duration value) {
this.leaseTtl = value;
}
public Duration getPollIntervalEarly() {
return pollIntervalEarly;
}
public void setPollIntervalEarly(Duration value) {
this.pollIntervalEarly = value;
}
public Duration getPollIntervalMid() {
return pollIntervalMid;
}
public void setPollIntervalMid(Duration value) {
this.pollIntervalMid = value;
}
public Duration getPollIntervalLate() {
return pollIntervalLate;
}
public void setPollIntervalLate(Duration value) {
this.pollIntervalLate = value;
}
public Duration getPollEarlyUntil() {
return pollEarlyUntil;
}
public void setPollEarlyUntil(Duration value) {
this.pollEarlyUntil = value;
}
public Duration getPollMidUntil() {
return pollMidUntil;
}
public void setPollMidUntil(Duration value) {
this.pollMidUntil = value;
}
public Duration getRetryBackoffBase() {
return retryBackoffBase;
}
public void setRetryBackoffBase(Duration value) {
this.retryBackoffBase = value;
}
}
}
@@ -0,0 +1,104 @@
package com.patbond.patbond.creation.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
import java.time.Duration;
/**
* Security knobs of the creation service: the RS256 public key for verifying
* access tokens issued by patbond-auth (same contract as the other services'
* {@code patbond.jwt.public-key}), and the shared service secret presented on
* outbound {@code /internal/**} calls to patbond-user.
*
* <p>Note that the credential for calling patbond-ai is NOT here — that hop
* uses a service-identity JWT with its own key pair, configured under
* {@code patbond.creation.service-token} (ADR-023 §5). Two internal auth
* mechanisms coexist on purpose: the existing Java↔Java channel keeps its
* shared token, the Python hop gets the stronger scheme. Reworking the former
* is deliberately out of M4's scope.
*/
@ConfigurationProperties(prefix = "patbond")
public class CreationSecurityProperties {
/**
* Shared secret sent as X-Internal-Token on calls to patbond-user's
* /internal/** API; must equal the value patbond-user expects.
*/
private String internalToken;
private final Jwt jwt = new Jwt();
private final UserService userService = new UserService();
public String getInternalToken() {
return internalToken;
}
public void setInternalToken(String value) {
this.internalToken = value;
}
public Jwt getJwt() {
return jwt;
}
public UserService getUserService() {
return userService;
}
public static class Jwt {
/**
* RS256 public key for verifying access tokens signed by patbond-auth:
* either inline PEM (starts with -----BEGIN) or a filesystem path. The
* private key never reaches this service.
*/
private String publicKey;
public String getPublicKey() {
return publicKey;
}
public void setPublicKey(String publicKey) {
this.publicKey = publicKey;
}
}
public static class UserService {
/**
* patbond-user base URL, ADR-002 static direct addressing. Source of
* presigned upload credentials via /internal/media/** — this service
* never talks to the object store itself (ADR-017 boundary).
*/
private String url = "http://127.0.0.1:8082";
private Duration connectTimeout = Duration.ofSeconds(1);
private Duration readTimeout = Duration.ofSeconds(2);
public String getUrl() {
return url;
}
public void setUrl(String url) {
this.url = url;
}
public Duration getConnectTimeout() {
return connectTimeout;
}
public void setConnectTimeout(Duration connectTimeout) {
this.connectTimeout = connectTimeout;
}
public Duration getReadTimeout() {
return readTimeout;
}
public void setReadTimeout(Duration readTimeout) {
this.readTimeout = readTimeout;
}
}
}
@@ -0,0 +1,21 @@
package com.patbond.patbond.creation.config;
import com.fasterxml.jackson.databind.DeserializationFeature;
import org.springframework.boot.autoconfigure.jackson.Jackson2ObjectMapperBuilderCustomizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* Integer fields must reject decimal input (development-plan 4.3) — Jackson's
* default is to silently truncate 45.5 to 45 in an integer field, which
* would corrupt values instead of rejecting them. Same setting as the other
* services so all envelopes behave identically.
*/
@Configuration
public class JacksonConfig {
@Bean
public Jackson2ObjectMapperBuilderCustomizer rejectFloatAsInt() {
return builder -> builder.featuresToDisable(DeserializationFeature.ACCEPT_FLOAT_AS_INT);
}
}
@@ -0,0 +1,43 @@
package com.patbond.patbond.creation.config;
import com.patbond.patbond.creation.provider.GenerationProvider;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
import java.util.stream.Collectors;
/**
* {@code provider_code} → adapter lookup.
*
* <p>Resolution is by the job's frozen {@code provider_code_snapshot}, never by
* the current catalog. That is the entire reason the snapshot columns exist: a
* model row may later be disabled or re-pointed at a different vendor, and jobs
* submitted under the old configuration must still be pollable by the adapter
* that submitted them. Routing by the live catalog would silently ask the wrong
* vendor about someone else's request id.
*/
public class ProviderRegistry {
private final Map<String, GenerationProvider> byCode;
public ProviderRegistry(List<GenerationProvider> providers) {
this.byCode = providers.stream().collect(Collectors.toUnmodifiableMap(
GenerationProvider::providerCode, Function.identity()));
}
/**
* @return the adapter that serves this provider code, empty when the code is
* unknown — which for a persisted job means its vendor adapter was
* removed while work was in flight. The worker must treat that as a
* non-retryable failure rather than looping.
*/
public Optional<GenerationProvider> find(String providerCode) {
return Optional.ofNullable(byCode.get(providerCode));
}
public java.util.Set<String> knownCodes() {
return byCode.keySet();
}
}
@@ -0,0 +1,76 @@
package com.patbond.patbond.creation.config;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.patbond.patbond.creation.provider.AiServiceGenerationProvider;
import com.patbond.patbond.creation.provider.FixtureGenerationProvider;
import com.patbond.patbond.creation.provider.GenerationProvider;
import com.patbond.patbond.creation.security.BearerAuthFilter;
import com.patbond.patbond.creation.security.JwtVerifier;
import com.patbond.patbond.creation.security.ServiceTokenSigner;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.web.servlet.FilterRegistrationBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.List;
/**
* Wires bearer authentication for {@code /api/v1/**} without pulling in
* spring-security — the same single-filter pattern user/pet/community use. The
* {@code /health} probe stays outside {@code /api/v1} and therefore
* unauthenticated.
*
* <p>Also wires the provider registry. Providers are looked up by
* {@code provider_code} so a job's frozen {@code provider_code_snapshot} keeps
* routing to the right adapter even after the catalog is re-pointed — that is
* the whole point of snapshotting it.
*/
@Configuration
@EnableConfigurationProperties({CreationSecurityProperties.class, CreationProperties.class})
public class SecurityConfig {
@Bean
public JwtVerifier jwtVerifier(CreationSecurityProperties properties) {
return new JwtVerifier(properties.getJwt().getPublicKey());
}
@Bean
public FilterRegistrationBean<BearerAuthFilter> bearerAuthFilter(
JwtVerifier jwtVerifier, ObjectMapper objectMapper) {
FilterRegistrationBean<BearerAuthFilter> registration = new FilterRegistrationBean<>(
new BearerAuthFilter(jwtVerifier, objectMapper));
registration.addUrlPatterns("/api/v1/*");
registration.setOrder(20);
return registration;
}
/**
* Built even when no private key is configured: an absent key must not stop
* the service from starting and serving reads, exactly as an absent JWT
* public key does not. The failure surfaces at {@code sign()} — when a call
* to patbond-ai is actually attempted — and reports itself as a server
* misconfiguration rather than an upstream fault.
*/
@Bean
public ServiceTokenSigner serviceTokenSigner(CreationProperties properties) {
CreationProperties.ServiceToken config = properties.getServiceToken();
return new ServiceTokenSigner(config.getPrivateKey(), config.getIssuer(),
config.getAudience(), config.getSubject(), config.getTtl());
}
@Bean
public FixtureGenerationProvider fixtureGenerationProvider() {
return new FixtureGenerationProvider();
}
@Bean
public AiServiceGenerationProvider aiServiceGenerationProvider() {
return new AiServiceGenerationProvider();
}
/** provider_code → adapter, so a job's snapshot routes deterministically. */
@Bean
public ProviderRegistry providerRegistry(List<GenerationProvider> providers) {
return new ProviderRegistry(providers);
}
}
@@ -0,0 +1,72 @@
package com.patbond.patbond.creation.controller;
import com.patbond.patbond.common.error.BusinessException;
import com.patbond.patbond.common.error.ErrorCode;
import com.patbond.patbond.common.response.ApiResponse;
import com.patbond.patbond.creation.domain.MediaKind;
import com.patbond.patbond.creation.repository.CatalogRepository;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
import java.util.UUID;
/**
* The model / style catalog the create screen renders from.
*
* <p>{@code GET /catalog} reports which media kinds actually have enabled
* models. The client is specified to render its kind switcher from that list
* rather than from a hardcoded one, so a kind with an empty catalog simply does
* not appear — instead of offering a control that promises a capability the
* backend cannot deliver. It also means the UI grows a video section on its own
* once the V7 seed lands, with no client release.
*/
@RestController
@RequestMapping("/api/v1/creation")
public class CatalogController {
private final CatalogRepository catalog;
public CatalogController(CatalogRepository catalog) {
this.catalog = catalog;
}
/** @param mediaKind optional filter; absent returns every enabled kind */
@GetMapping("/catalog")
public ApiResponse<CatalogResponse> catalog(
@RequestParam(required = false) String mediaKind) {
List<String> kinds = mediaKind == null
? catalog.availableMediaKinds()
: List.of(parse(mediaKind).dbValue());
List<ModelSummary> models = kinds.stream()
.flatMap(kind -> catalog.listEnabledModels(MediaKind.fromDb(kind)).stream())
.map(row -> new ModelSummary(row.id(), row.code(), row.mediaKind().dbValue()))
.toList();
return ApiResponse.success(new CatalogResponse(kinds, models));
}
private static MediaKind parse(String value) {
try {
return MediaKind.fromDb(value);
} catch (IllegalArgumentException e) {
throw new BusinessException(ErrorCode.VALIDATION_ERROR, "mediaKind 只能是 image 或 video");
}
}
/**
* @param availableMediaKinds drives whether the client shows a kind switcher
* at all, and with which segments
*/
public record CatalogResponse(List<String> availableMediaKinds, List<ModelSummary> models) {
}
/**
* Deliberately omits {@code providerCode} and {@code providerModelName}:
* which vendor serves a model is an internal detail, and exposing it would
* both leak the supply chain and invite clients to branch on it.
*/
public record ModelSummary(UUID id, String code, String mediaKind) {
}
}
@@ -0,0 +1,102 @@
package com.patbond.patbond.creation.controller;
import com.patbond.patbond.common.response.ApiResponse;
import com.patbond.patbond.creation.dto.CreateGenerationJobRequest;
import com.patbond.patbond.creation.dto.GenerationJobResponse;
import com.patbond.patbond.creation.security.BearerAuthFilter;
import com.patbond.patbond.creation.service.GenerationService;
import jakarta.validation.Valid;
import jakarta.validation.constraints.Max;
import jakarta.validation.constraints.Min;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.Size;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestAttribute;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
import java.util.UUID;
/**
* Client-facing generation endpoints (M4, ADR-023 §2).
*
* <p>All four are plain request/response: creation enqueues and returns
* immediately, progress is read by polling {@code GET .../jobs/{id}}. There is
* deliberately no streaming endpoint — SSE would bypass the envelope-and-refresh
* chain the whole client network layer is built on, creating a second auth path,
* and every push design needs this GET as a fallback anyway (ADR-023 §10).
*/
@RestController
@RequestMapping("/api/v1/creation")
@Validated
public class GenerationController {
private final GenerationService generationService;
public GenerationController(GenerationService generationService) {
this.generationService = generationService;
}
/**
* Enqueues a job. The Idempotency-Key header is mandatory (ADR-019): a
* generation request costs real money upstream, so a retried request must
* never be able to become a second billable job.
*/
@PostMapping("/jobs")
public ApiResponse<GenerationJobResponse> create(
@RequestAttribute(BearerAuthFilter.USER_ID_ATTRIBUTE) UUID userId,
@RequestHeader("Idempotency-Key")
@NotBlank(message = "Idempotency-Key 不能为空")
@Size(max = 128, message = "Idempotency-Key 最长 128 字符")
String idempotencyKey,
@Valid @RequestBody CreateGenerationJobRequest request) {
return ApiResponse.success(
generationService.create(userId, idempotencyKey.trim(), request));
}
/** The polling endpoint. Also the fallback any future push design needs. */
@GetMapping("/jobs/{jobId}")
public ApiResponse<GenerationJobResponse> get(
@RequestAttribute(BearerAuthFilter.USER_ID_ATTRIBUTE) UUID userId,
@PathVariable UUID jobId) {
return ApiResponse.success(generationService.get(userId, jobId));
}
/**
* The user's own jobs, newest first.
*
* <p>This is what makes "leaving the waiting screen" safe: a job in flight
* can always be found again here, which is why the client is allowed to
* navigate away — and why it must never tell the user it will notify them,
* since the app has no push capability at all (ADR-023 §10).
*/
@GetMapping("/jobs")
public ApiResponse<List<GenerationJobResponse>> list(
@RequestAttribute(BearerAuthFilter.USER_ID_ATTRIBUTE) UUID userId,
@RequestParam(defaultValue = "20")
@Min(value = 1, message = "limit 最小为 1")
@Max(value = 100, message = "limit 最大为 100")
int limit) {
return ApiResponse.success(generationService.list(userId, limit));
}
/**
* Cancels a job that has not finished. Answers 42206 when it already
* reached a terminal state rather than reporting a cancellation that did not
* happen.
*/
@DeleteMapping("/jobs/{jobId}")
public ApiResponse<GenerationJobResponse> cancel(
@RequestAttribute(BearerAuthFilter.USER_ID_ATTRIBUTE) UUID userId,
@PathVariable UUID jobId) {
return ApiResponse.success(generationService.cancel(userId, jobId));
}
}
@@ -0,0 +1,63 @@
package com.patbond.patbond.creation.domain;
import java.time.OffsetDateTime;
import java.util.UUID;
/**
* A row of {@code creation.generation_jobs} (V6), as far as this service needs
* it. Deliberately not the full column set: the snapshot columns and the
* request hash are write-once bookkeeping the worker never reads back, and
* loading them into every poll would only invite accidental mutation.
*
* @param firstAttemptAt written once on the first transition into RUNNING and
* never overwritten — it is what survives re-queueing so that queue wait
* ({@code firstAttemptAt - createdAt}) stays separable from pure execution
* time ({@code completedAt - firstAttemptAt}). The target model lacks this
* column; V6 adds it deliberately (ADR-023 §6).
* @param startedAt the CURRENT attempt's start. Cleared on every re-queue by
* {@code ck_generation_jobs_state}, which is exactly why
* {@code firstAttemptAt} has to exist separately.
*/
public record GenerationJob(
UUID id,
UUID userId,
UUID petId,
MediaKind mediaKind,
UUID modelId,
UUID styleId,
UUID inputAssetId,
UUID outputAssetId,
String prompt,
String negativePrompt,
Integer widthPx,
Integer heightPx,
Long durationMs,
JobStatus status,
short progress,
short attemptCount,
short maxAttempts,
String providerCodeSnapshot,
String providerRequestId,
String errorCode,
String errorMessage,
OffsetDateTime createdAt,
OffsetDateTime firstAttemptAt,
OffsetDateTime startedAt,
OffsetDateTime completedAt,
OffsetDateTime leaseExpiresAt) {
/** @return true when the job has never been attempted (attemptCount == 0) */
public boolean isFirstAttempt() {
return attemptCount == 0;
}
/** @return true when the next failure exhausts the retry budget */
public boolean isLastAttempt() {
return attemptCount >= maxAttempts;
}
/** @return true when the job was already handed to the provider */
public boolean isSubmittedUpstream() {
return providerRequestId != null;
}
}
@@ -0,0 +1,44 @@
package com.patbond.patbond.creation.domain;
/**
* Generation job lifecycle, mirroring {@code ck_generation_jobs_status} and
* the branch structure of {@code ck_generation_jobs_state} (V6).
*
* <p>The database — not this enum — is the authority on which transitions are
* legal: every branch of the state CHECK pins the full column tuple for its
* status, so an illegal transition fails as a constraint violation rather than
* silently persisting a half-state. Keep this enum in step with V6; do not
* encode transition rules here that the CHECK does not enforce.
*/
public enum JobStatus {
QUEUED("queued"),
RUNNING("running"),
SUCCEEDED("succeeded"),
FAILED("failed"),
CANCELLED("cancelled");
private final String dbValue;
JobStatus(String dbValue) {
this.dbValue = dbValue;
}
public String dbValue() {
return dbValue;
}
/** @return true when no further work will ever be done on the job */
public boolean isTerminal() {
return this == SUCCEEDED || this == FAILED || this == CANCELLED;
}
public static JobStatus fromDb(String value) {
for (JobStatus status : values()) {
if (status.dbValue.equals(value)) {
return status;
}
}
throw new IllegalArgumentException("Unknown generation job status: " + value);
}
}
@@ -0,0 +1,36 @@
package com.patbond.patbond.creation.domain;
/**
* The two generation kinds, mirroring {@code ck_generation_jobs_kind} and the
* catalog's {@code media_kind} (V6).
*
* <p>Both are structurally supported from V6 onwards and both ship in M4
* (ADR-023 §1, superseding ADR-018's "视频后置"). The catalog is keyed by
* {@code (id, media_kind)} and jobs carry a composite FK into it, so a video
* job referencing an image model is rejected by the database rather than by
* application code.
*/
public enum MediaKind {
IMAGE("image"),
VIDEO("video");
private final String dbValue;
MediaKind(String dbValue) {
this.dbValue = dbValue;
}
public String dbValue() {
return dbValue;
}
public static MediaKind fromDb(String value) {
for (MediaKind kind : values()) {
if (kind.dbValue.equals(value)) {
return kind;
}
}
throw new IllegalArgumentException("Unknown media kind: " + value);
}
}
@@ -0,0 +1,79 @@
package com.patbond.patbond.creation.dto;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;
import jakarta.validation.constraints.Size;
import java.util.UUID;
/**
* Request to start a generation job.
*
* <p><b>{@code inputAssetId} is optional on purpose.</b> The reference image is
* optional (decision A2): mainstream image and video providers support both
* text-to-* and reference-guided generation, and which one a request means is
* decided by whether an input asset is present. The target data model had this
* column NOT NULL, which would have made text-to-image unrepresentable — V6
* relaxes it deliberately (ADR-023 §6).
*
* <p>Whether a PARTICULAR model requires an input is a catalog question, checked
* by the service. It is intentionally not expressed as a schema constraint: no
* {@code requires_input} flag was invented beyond the reviewed model.
*
* <p>{@code mediaKind} is validated against the catalog rather than trusted:
* the composite FK {@code (model_id, media_kind)} would reject a mismatch at the
* database level anyway, but answering 404 beats surfacing a constraint
* violation.
*/
public record CreateGenerationJobRequest(
@NotBlank(message = "mediaKind 不能为空")
String mediaKind,
@NotNull(message = "modelId 不能为空")
UUID modelId,
UUID styleId,
/** Optional — absent means text-to-*, present means reference-guided. */
UUID inputAssetId,
UUID petId,
@Size(max = 10000, message = "prompt 最长 10000 字符")
String prompt,
@Size(max = 3000, message = "negativePrompt 最长 3000 字符")
String negativePrompt,
Integer widthPx,
Integer heightPx,
Long durationMs) {
/**
* Canonical form hashed into {@code request_hash} for idempotency
* comparison (ADR-019). Hashes the NORMALIZED command rather than raw bytes
* so the comparison is insensitive to JSON formatting while still catching
* every semantic difference — a keyed retry with the same meaning returns
* the first job, a different payload answers 40905.
*/
public String canonical() {
return String.join("|",
nullSafe(mediaKind),
nullSafe(modelId),
nullSafe(styleId),
nullSafe(inputAssetId),
nullSafe(petId),
nullSafe(prompt == null ? null : prompt.trim()),
nullSafe(negativePrompt == null ? null : negativePrompt.trim()),
nullSafe(widthPx),
nullSafe(heightPx),
nullSafe(durationMs));
}
private static String nullSafe(Object value) {
return value == null ? "" : value.toString();
}
}
@@ -0,0 +1,100 @@
package com.patbond.patbond.creation.dto;
import com.patbond.patbond.creation.domain.GenerationJob;
import com.patbond.patbond.creation.domain.JobStatus;
import java.time.Duration;
import java.time.OffsetDateTime;
import java.util.UUID;
/**
* A generation job as the client sees it.
*
* <p>Two fields deserve explanation because they encode client-facing decisions
* from ADR-023 §10 rather than raw columns:
*
* <ul>
* <li>{@code timeoutSec} is issued BY THE SERVER, not hardcoded per kind in
* the app. Image and video differ by an order of magnitude, and baking two
* constants into the client would mean a new app build whenever a provider
* changes.</li>
* <li>{@code stalled} is the sixth UI state — a RUNNING job that has exceeded
* its expected window. It is deliberately NOT an error: the job may still
* finish, so the client shows a different affordance rather than a
* failure. Computing it here keeps the threshold configurable in one
* place.</li>
* </ul>
*
* <p>{@code progress} is reported as-is and the contract promises nothing about
* it advancing smoothly: the state CHECK pins it to 0 while queued and puts no
* floor on it while running. Clients are told to treat an indeterminate
* indicator as the primary form, precisely so a truthful 0 does not read as
* "stuck at 0%".
*/
public record GenerationJobResponse(
UUID id,
String mediaKind,
String status,
int progress,
UUID modelId,
UUID styleId,
UUID inputAssetId,
UUID outputAssetId,
String prompt,
Integer widthPx,
Integer heightPx,
Long durationMs,
int attemptCount,
String errorCode,
String errorMessage,
OffsetDateTime createdAt,
OffsetDateTime startedAt,
OffsetDateTime completedAt,
long timeoutSec,
boolean stalled,
Long elapsedSec) {
public static GenerationJobResponse of(GenerationJob job, Duration clientTimeout,
Duration staleAfter) {
Duration elapsed = elapsedOf(job);
boolean stalled = job.status() == JobStatus.RUNNING
&& elapsed != null
&& elapsed.compareTo(staleAfter) > 0;
return new GenerationJobResponse(
job.id(),
job.mediaKind().dbValue(),
job.status().dbValue(),
job.progress(),
job.modelId(),
job.styleId(),
job.inputAssetId(),
job.outputAssetId(),
job.prompt(),
job.widthPx(),
job.heightPx(),
job.durationMs(),
job.attemptCount(),
job.errorCode(),
job.errorMessage(),
job.createdAt(),
job.startedAt(),
job.completedAt(),
clientTimeout.toSeconds(),
stalled,
elapsed == null ? null : elapsed.toSeconds());
}
/**
* Elapsed wall time of the job so far — measured from creation, not from the
* current attempt, because that is what the user actually waited. Shown
* INSTEAD of an ETA: elapsed time is a fact, an ETA is a promise, and M4
* cannot make a credible one.
*/
private static Duration elapsedOf(GenerationJob job) {
if (job.createdAt() == null) {
return null;
}
OffsetDateTime end = job.completedAt() != null ? job.completedAt() : OffsetDateTime.now();
return Duration.between(job.createdAt().toInstant(), end.toInstant());
}
}
@@ -0,0 +1,65 @@
package com.patbond.patbond.creation.provider;
import com.patbond.patbond.creation.domain.GenerationJob;
/**
* Talks to patbond-ai (Python) over REST for real vendor generation —
* 火山引擎 Seedance for video, gpt-image-2 for images (ADR-023 §1).
*
* <p>Deliberately unimplemented at skeleton stage. Five facts must come out of
* the vendors' official documentation before the request/response shape can be
* fixed, and guessing any of them would produce an adapter that compiles and
* silently misbehaves:
* <ol>
* <li><b>Idempotency key support</b> — decides whether the double-charge
* window on a synchronous vendor can be closed at all, or only bounded
* by {@code max_attempts} (ADR-023 §4). This is the one item no amount of
* design on our side can work around.</li>
* <li><b>Query endpoint shape and any polling rate limit</b> — decides
* whether the 3s/10s/30s backoff ladder is admissible.</li>
* <li><b>Maximum job duration and timeout behaviour</b> — feeds the
* server-issued client timeout, which must not be hardcoded per kind.</li>
* <li><b>Artifact return form (URL vs bytes)</b> — a time-limited vendor URL
* lets patbond-ai stream straight into the presigned PUT without ever
* landing the file.</li>
* <li><b>Rate-limit and quota response shapes</b> — decides how they map onto
* the two distinct 429 codes (quota exhausted waits until tomorrow and
* gets a sheet; too-fast waits seconds and gets a snackbar — one shared
* code would make half the copy and half the exits wrong).</li>
* </ol>
*
* <p>Note what is NOT blocked by the above: the transport, auth and error
* envelope are already settled and independent of vendor specifics — REST/JSON
* (not gRPC), a service-identity JWT signed with a separate key pair, and a
* request body that carries {@code jobId}, generation parameters and the
* presigned upload target but deliberately NOT {@code userId} (ADR-023 §5 data
* minimisation).
*/
public class AiServiceGenerationProvider implements GenerationProvider {
/**
* Placeholder code. The real value comes from
* {@code generation_models.provider_code} in the V7 catalog seed, which is
* itself waiting on the same vendor documentation.
*/
public static final String PROVIDER_CODE = "patbond-ai";
@Override
public String providerCode() {
return PROVIDER_CODE;
}
@Override
public SubmitResult submit(GenerationJob job, String uploadTarget, String idempotencyKey) {
throw new UnsupportedOperationException(
"AiServiceGenerationProvider.submit 未实现:待 Seedance 与 gpt-image-2 官方文档"
+ "确认五项(幂等键支持、查询接口与频率限制、单任务最长时长、产物返回形式、"
+ "限流与配额响应形态)后落地。");
}
@Override
public PollResult poll(GenerationJob job, String providerRequestId) {
throw new UnsupportedOperationException(
"AiServiceGenerationProvider.poll 未实现(同 submit)。");
}
}
@@ -0,0 +1,51 @@
package com.patbond.patbond.creation.provider;
import com.patbond.patbond.creation.domain.GenerationJob;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Stand-in provider used while real vendor credentials and API shapes are not
* yet settled.
*
* <p><b>This is not a placeholder to be deleted — it is a deliberate M4
* asset.</b> All four M4 acceptance criteria (queue mechanics, lease recovery,
* idempotent retry, artifact reaching the media table and then a community
* draft) are independent of whether the picture looks good. A fixture provider
* additionally lets us inject failures a real vendor cannot reproduce on
* demand: timeouts, mid-flight worker death, a 500 on the third attempt. The
* canonical seed itself specifies {@code provider_code = 'fixture'}.
*
* <p>Deliberately NOT implemented yet: producing real bytes and PUTting them to
* the presigned target. That step is the one place where the storage decision
* shows up (local object storage vs a cloud provider changes what patbond-ai
* returns), and that decision is explicitly deferred (ADR-023 §7). The methods
* fail loudly rather than pretending to work, so nothing downstream can mature
* against a fake success.
*/
public class FixtureGenerationProvider implements GenerationProvider {
public static final String PROVIDER_CODE = "fixture";
private static final Logger log = LoggerFactory.getLogger(FixtureGenerationProvider.class);
@Override
public String providerCode() {
return PROVIDER_CODE;
}
@Override
public SubmitResult submit(GenerationJob job, String uploadTarget, String idempotencyKey) {
log.warn("Fixture provider is not implemented yet; job {} will not be submitted", job.id());
throw new UnsupportedOperationException(
"FixtureGenerationProvider.submit 未实现:产物生成与预签名上传的形态取决于"
+ "对象存储方案(ADR-023 §7 明确挂起)。实现前请勿让任务进入队列,"
+ "否则会以「假成功」污染指标基线。");
}
@Override
public PollResult poll(GenerationJob job, String providerRequestId) {
throw new UnsupportedOperationException(
"FixtureGenerationProvider.poll 未实现(同 submit)。");
}
}
@@ -0,0 +1,108 @@
package com.patbond.patbond.creation.provider;
import com.patbond.patbond.creation.domain.GenerationJob;
/**
* The upstream generation boundary. One implementation talks to patbond-ai
* over REST; a fixture implementation stands in while provider credentials are
* not yet available (ADR-023 §1/§4).
*
* <p><b>Two phases, always — even for synchronous providers.</b> Some vendors
* are asynchronous (submit + query, e.g. video), others complete inside one
* long-lived HTTP call (e.g. image). That difference must NOT reach this
* interface, the job table or the client contract: {@code submit} returns as
* soon as the work is accepted and {@code poll} reports progress. A
* synchronous vendor is absorbed on the patbond-ai side by running the long
* call as a background task.
*
* <p>Why it matters beyond tidiness: {@code submit} returning a request id
* immediately is what lets the worker persist {@code provider_request_id}
* before any billable work happens. Without it, a timeout leaves us unable to
* tell whether we were charged, and the retry charges again. The partial
* unique index {@code uq_generation_jobs_provider_request} exists to enforce
* exactly that de-duplication.
*/
public interface GenerationProvider {
/**
* @return the {@code provider_code} this implementation serves, matching
* {@code generation_models.provider_code} and the value frozen into
* {@code provider_code_snapshot}
*/
String providerCode();
/**
* Hands the job upstream. MUST NOT block for the generation itself.
*
* @param job the claimed job, already RUNNING under this worker's lease
* @param uploadTarget where the artifact is to be PUT once produced —
* a presigned URL obtained from patbond-user's internal media API.
* Passing the target in at submit time is what keeps object-storage
* credentials out of this service and out of patbond-ai entirely
* (ADR-023 §4).
* @param idempotencyKey the job's own key, to be forwarded upstream as the
* vendor's idempotency header where the vendor supports one. Where it
* does not, the double-charge window is bounded only by
* {@code max_attempts} — a residual risk ADR-023 §4 records rather
* than pretends to have solved.
*/
SubmitResult submit(GenerationJob job, String uploadTarget, String idempotencyKey);
/**
* Asks after a previously submitted job.
*
* @param providerRequestId the id returned by {@link #submit}
*/
PollResult poll(GenerationJob job, String providerRequestId);
/**
* Outcome of a submission.
*
* @param providerRequestId upstream handle; persisted immediately
*/
record SubmitResult(String providerRequestId) {
}
/**
* Outcome of a poll.
*
* @param state where the upstream job is now
* @param progress 0..100 when known; the client contract deliberately does
* not promise a percentage (ADR-023 §10), so 0 while queued is normal
* rather than a bug
* @param objectKey set only on {@link State#SUCCEEDED} — the key the
* artifact was uploaded under, which this service turns into a
* {@code media.assets} row
* @param errorCode set only on {@link State#FAILED}
* @param retryable whether a FAILED outcome is worth another attempt; a
* rejected prompt is not, a vendor 5xx is
*/
record PollResult(State state, int progress, String objectKey,
String errorCode, String errorMessage, boolean retryable) {
public enum State {
/** Accepted upstream, not started. */
PENDING,
/** Upstream is working on it. */
RUNNING,
SUCCEEDED,
FAILED
}
public static PollResult pending(int progress) {
return new PollResult(State.PENDING, progress, null, null, null, false);
}
public static PollResult running(int progress) {
return new PollResult(State.RUNNING, progress, null, null, null, false);
}
public static PollResult succeeded(String objectKey) {
return new PollResult(State.SUCCEEDED, 100, objectKey, null, null, false);
}
public static PollResult failed(String errorCode, String errorMessage, boolean retryable) {
return new PollResult(State.FAILED, 0, null, errorCode, errorMessage, retryable);
}
}
}
@@ -0,0 +1,118 @@
package com.patbond.patbond.creation.repository;
import com.patbond.patbond.creation.domain.MediaKind;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.stereotype.Repository;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
/**
* {@code creation.generation_models} / {@code generation_styles} reads.
*
* <p>The catalog is a table rather than a config file, and not by preference:
* {@code generation_jobs.model_id} is a NOT NULL uuid participating in a
* composite foreign key, so a config file has no row for it to point at.
*
* <p>Every lookup is scoped by {@code media_kind}. The database enforces the
* same rule through the {@code (id, media_kind)} composite FK — a video job
* cannot reference an image model — but resolving it here lets the API answer a
* clean 404 instead of surfacing a constraint violation.
*/
@Repository
public class CatalogRepository {
private final JdbcClient jdbcClient;
public CatalogRepository(JdbcClient jdbcClient) {
this.jdbcClient = jdbcClient;
}
/**
* Model row needed to freeze a job's provider snapshots.
*
* @param modelVersion currently sourced from {@code provider_model_name} —
* the catalog has no separate version column, while jobs snapshot a
* {@code model_version_snapshot}. Whether the vendors expose a distinct
* version string is one of the open items awaiting their documentation;
* until then the two snapshots carry the same value rather than a
* fabricated one.
*/
public record ModelRow(UUID id, String code, String providerCode, String providerModelName,
String modelVersion, MediaKind mediaKind) {
}
public record StyleRow(UUID id, String code, MediaKind mediaKind) {
}
public Optional<ModelRow> findEnabledModel(UUID modelId, MediaKind mediaKind) {
return jdbcClient.sql("""
SELECT id, code, provider_code, provider_model_name, media_kind
FROM creation.generation_models
WHERE id = :id AND media_kind = :mediaKind AND enabled
""")
.param("id", modelId)
.param("mediaKind", mediaKind.dbValue())
.query((rs, rowNum) -> new ModelRow(
rs.getObject("id", UUID.class),
rs.getString("code"),
rs.getString("provider_code"),
rs.getString("provider_model_name"),
rs.getString("provider_model_name"),
MediaKind.fromDb(rs.getString("media_kind"))))
.optional();
}
public Optional<StyleRow> findEnabledStyle(UUID styleId, MediaKind mediaKind) {
return jdbcClient.sql("""
SELECT id, code, media_kind
FROM creation.generation_styles
WHERE id = :id AND media_kind = :mediaKind AND enabled
""")
.param("id", styleId)
.param("mediaKind", mediaKind.dbValue())
.query((rs, rowNum) -> new StyleRow(
rs.getObject("id", UUID.class),
rs.getString("code"),
MediaKind.fromDb(rs.getString("media_kind"))))
.optional();
}
/**
* Enabled models of one kind, ordered as
* {@code ix_generation_models_kind_order} indexes them.
*/
public List<ModelRow> listEnabledModels(MediaKind mediaKind) {
return jdbcClient.sql("""
SELECT id, code, provider_code, provider_model_name, media_kind
FROM creation.generation_models
WHERE media_kind = :mediaKind AND enabled
ORDER BY sort_order, id
""")
.param("mediaKind", mediaKind.dbValue())
.query((rs, rowNum) -> new ModelRow(
rs.getObject("id", UUID.class),
rs.getString("code"),
rs.getString("provider_code"),
rs.getString("provider_model_name"),
rs.getString("provider_model_name"),
MediaKind.fromDb(rs.getString("media_kind"))))
.list();
}
/**
* Which media kinds actually have enabled models. The client renders its
* kind switcher from this rather than from a hardcoded list — so a kind with
* no catalog entries simply does not appear, instead of offering a control
* that promises a capability the backend cannot deliver.
*/
public List<String> availableMediaKinds() {
return jdbcClient.sql("""
SELECT DISTINCT media_kind FROM creation.generation_models
WHERE enabled ORDER BY media_kind
""")
.query(String.class)
.list();
}
}
@@ -0,0 +1,431 @@
package com.patbond.patbond.creation.repository;
import com.patbond.patbond.creation.domain.GenerationJob;
import com.patbond.patbond.creation.domain.JobStatus;
import com.patbond.patbond.creation.domain.MediaKind;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.stereotype.Repository;
import java.time.Duration;
import java.time.OffsetDateTime;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
/**
* {@code creation.generation_jobs} access, including the lease queue the
* worker runs on.
*
* <p><b>Why the queue lives here and not in a broker</b> (ADR-023 §3): the
* reviewed target model already carries the whole lease apparatus as columns
* ({@code lease_owner}, {@code lease_expires_at}, {@code next_attempt_at},
* {@code attempt_count}, {@code max_attempts}, {@code priority}) plus two
* partial indexes whose column order matches the statements below exactly. The
* claim is therefore an index scan, not a sort, and job state has a single
* source of truth with no dual-write.
*
* <p><b>The invariant every caller must hold</b>: status and its companion
* columns move together in one statement. {@code ck_generation_jobs_state}
* pins the full column tuple per status, so a partial update does not persist
* a half-state — it fails. That is deliberate: the constraint is the last line
* of defence against a worker that crashed mid-transition.
*/
@Repository
public class GenerationJobRepository {
/**
* Columns loaded by every read. Excludes the write-once snapshot columns
* and {@code request_hash} — see {@link GenerationJob}.
*
* <p>Kept as a list rather than two hand-written strings because the
* {@code UPDATE … FROM … RETURNING} statements below need the same columns
* qualified with the target alias: both the target table and the CTE expose
* {@code id}, so an unqualified RETURNING list is ambiguous and postgres
* rejects it. Deriving both forms from one list means the two can never
* drift apart.
*/
private static final List<String> COLUMNS = List.of(
"id", "user_id", "pet_id", "media_kind", "model_id", "style_id",
"input_asset_id", "output_asset_id", "prompt", "negative_prompt",
"width_px", "height_px", "duration_ms", "status", "progress",
"attempt_count", "max_attempts", "provider_code_snapshot",
"provider_request_id", "error_code", "error_message",
"created_at", "first_attempt_at", "started_at", "completed_at",
"lease_expires_at");
private static final String SELECT_COLUMNS = String.join(", ", COLUMNS);
/** The same columns qualified with the UPDATE target's alias {@code j}. */
private static final String RETURNING_COLUMNS =
COLUMNS.stream().map(column -> "j." + column).collect(java.util.stream.Collectors.joining(", "));
private final JdbcClient jdbcClient;
public GenerationJobRepository(JdbcClient jdbcClient) {
this.jdbcClient = jdbcClient;
}
// ---------------------------------------------------------------- reads
public Optional<GenerationJob> findByIdForUser(UUID jobId, UUID userId) {
return jdbcClient.sql("SELECT " + SELECT_COLUMNS
+ " FROM creation.generation_jobs WHERE id = :id AND user_id = :userId")
.param("id", jobId)
.param("userId", userId)
.query(GenerationJobRepository::mapRow)
.optional();
}
/**
* Returns an existing job for the same (user, idempotency key) so a keyed
* retry answers the first write instead of creating a second job
* (ADR-019). The caller compares {@code request_hash} to decide between
* "same request, return it" and 40905.
*/
public Optional<UUID> findIdByIdempotencyKey(UUID userId, String idempotencyKey, byte[] requestHash) {
return jdbcClient.sql("""
SELECT id FROM creation.generation_jobs
WHERE user_id = :userId AND idempotency_key = :key
AND request_hash = :hash
""")
.param("userId", userId)
.param("key", idempotencyKey)
.param("hash", requestHash)
.query(UUID.class)
.optional();
}
/** @return true when the key exists under this user with a DIFFERENT payload */
public boolean idempotencyKeyConflicts(UUID userId, String idempotencyKey, byte[] requestHash) {
return jdbcClient.sql("""
SELECT EXISTS (SELECT 1 FROM creation.generation_jobs
WHERE user_id = :userId AND idempotency_key = :key
AND request_hash <> :hash)
""")
.param("userId", userId)
.param("key", idempotencyKey)
.param("hash", requestHash)
.query(Boolean.class)
.single();
}
/** Newest first, matching {@code ix_generation_jobs_user_created}. */
public List<GenerationJob> listForUser(UUID userId, int limit) {
return jdbcClient.sql("SELECT " + SELECT_COLUMNS + """
FROM creation.generation_jobs
WHERE user_id = :userId
ORDER BY created_at DESC, id DESC
LIMIT :limit
""")
.param("userId", userId)
.param("limit", limit)
.query(GenerationJobRepository::mapRow)
.list();
}
// ------------------------------------------------------------- creation
/**
* Inserts a QUEUED job. All snapshot columns are frozen here on purpose:
* the catalog may later be re-pointed or disabled, and historical jobs
* must stay readable without it.
*
* @return the new job id
*/
public UUID insertQueued(UUID id, UUID userId, UUID petId, MediaKind mediaKind,
UUID modelId, UUID styleId, UUID inputAssetId,
String prompt, String negativePrompt,
Integer widthPx, Integer heightPx, Long durationMs,
String providerCode, String providerModel, String modelVersion,
String styleCode, String idempotencyKey, byte[] requestHash) {
jdbcClient.sql("""
INSERT INTO creation.generation_jobs (
id, user_id, pet_id, media_kind, model_id, style_id,
input_asset_id, prompt, negative_prompt,
width_px, height_px, duration_ms,
provider_code_snapshot, provider_model_snapshot,
model_version_snapshot, style_code_snapshot,
idempotency_key, request_hash
) VALUES (
:id, :userId, :petId, :mediaKind, :modelId, :styleId,
:inputAssetId, :prompt, :negativePrompt,
:widthPx, :heightPx, :durationMs,
:providerCode, :providerModel,
:modelVersion, :styleCode,
:idempotencyKey, :requestHash
)
""")
.param("id", id)
.param("userId", userId)
.param("petId", petId)
.param("mediaKind", mediaKind.dbValue())
.param("modelId", modelId)
.param("styleId", styleId)
.param("inputAssetId", inputAssetId)
.param("prompt", prompt)
.param("negativePrompt", negativePrompt)
.param("widthPx", widthPx)
.param("heightPx", heightPx)
.param("durationMs", durationMs)
.param("providerCode", providerCode)
.param("providerModel", providerModel)
.param("modelVersion", modelVersion)
.param("styleCode", styleCode)
.param("idempotencyKey", idempotencyKey)
.param("requestHash", requestHash)
.update();
return id;
}
// ---------------------------------------------------------- lease queue
/**
* Claims up to {@code batchSize} due QUEUED jobs and moves them to RUNNING
* under a lease, in ONE statement.
*
* <p>{@code FOR UPDATE SKIP LOCKED} is what makes several workers safe
* without a broker: a row another worker is claiming is skipped rather
* than waited on. The inner ORDER BY matches
* {@code ix_generation_jobs_queue} column for column
* ({@code priority DESC, next_attempt_at, created_at, id}).
*
* <p>{@code first_attempt_at} uses COALESCE so it is written on the first
* attempt and preserved on every retry — the entire reason the column
* exists (ADR-023 §6). {@code started_at} is overwritten each attempt
* because the state CHECK requires it to be the current attempt's start.
*/
public List<GenerationJob> claimQueued(String leaseOwner, Duration leaseTtl, int batchSize) {
return jdbcClient.sql("""
WITH claimed AS (
SELECT id FROM creation.generation_jobs
WHERE status = 'queued' AND next_attempt_at <= now()
ORDER BY priority DESC, next_attempt_at, created_at, id
LIMIT :batchSize
FOR UPDATE SKIP LOCKED
)
UPDATE creation.generation_jobs j
SET status = 'running',
attempt_count = j.attempt_count + 1,
first_attempt_at = COALESCE(j.first_attempt_at, now()),
started_at = now(),
lease_owner = :leaseOwner,
lease_expires_at = now() + :leaseTtl::interval,
error_code = NULL,
error_message = NULL
FROM claimed
WHERE j.id = claimed.id
RETURNING\s""" + RETURNING_COLUMNS)
.param("leaseOwner", leaseOwner)
.param("leaseTtl", leaseTtl.toSeconds() + " seconds")
.param("batchSize", batchSize)
.query(GenerationJobRepository::mapRow)
.list();
}
/**
* Claims RUNNING jobs this owner already holds whose lease is close enough
* to expiry to be worth a poll, extending the lease in the same statement.
* Backed by {@code ix_generation_jobs_running}.
*
* <p>Polling and renewing together is not an optimisation but a
* correctness requirement: a poll that does not renew leaves a long
* upstream job open to being reclaimed mid-flight, which is how you end up
* paying a provider twice.
*/
public List<GenerationJob> claimRunningDueForPoll(String leaseOwner, Duration leaseTtl,
Duration pollDue, int batchSize) {
return jdbcClient.sql("""
WITH due AS (
SELECT id FROM creation.generation_jobs
WHERE status = 'running'
AND lease_owner = :leaseOwner
AND lease_expires_at <= now() + :pollDue::interval
ORDER BY lease_expires_at, id
LIMIT :batchSize
FOR UPDATE SKIP LOCKED
)
UPDATE creation.generation_jobs j
SET lease_expires_at = now() + :leaseTtl::interval
FROM due
WHERE j.id = due.id
RETURNING\s""" + RETURNING_COLUMNS)
.param("leaseOwner", leaseOwner)
.param("pollDue", pollDue.toSeconds() + " seconds")
.param("leaseTtl", leaseTtl.toSeconds() + " seconds")
.param("batchSize", batchSize)
.query(GenerationJobRepository::mapRow)
.list();
}
/**
* Returns expired-lease RUNNING jobs to the queue so a crashed worker's
* work is picked up again. The target model supplied the index for this
* but no implementation (ADR-023 §3) — this is that implementation.
*
* <p>{@code provider_request_id} is deliberately NOT cleared: it is how the
* next attempt discovers the work was already submitted upstream and polls
* instead of re-submitting. Clearing it here would turn every worker crash
* into a double charge.
*
* @return number of jobs returned to the queue
*/
public int reapExpiredLeases(int batchSize) {
return jdbcClient.sql("""
WITH expired AS (
SELECT id FROM creation.generation_jobs
WHERE status = 'running' AND lease_expires_at <= now()
ORDER BY lease_expires_at, id
LIMIT :batchSize
FOR UPDATE SKIP LOCKED
)
UPDATE creation.generation_jobs j
SET status = 'queued',
progress = 0,
started_at = NULL,
lease_owner = NULL,
lease_expires_at = NULL,
next_attempt_at = now()
FROM expired
WHERE j.id = expired.id
""")
.param("batchSize", batchSize)
.update();
}
// ------------------------------------------------------- state advances
/**
* Records the upstream submission. Writing this BEFORE the provider does
* any billable work is what makes a timeout recoverable: the next attempt
* sees the id and polls rather than re-submitting.
*/
public int recordProviderRequest(UUID jobId, String providerRequestId) {
return jdbcClient.sql("""
UPDATE creation.generation_jobs
SET provider_request_id = :requestId
WHERE id = :id AND provider_request_id IS NULL
""")
.param("id", jobId)
.param("requestId", providerRequestId)
.update();
}
public int updateProgress(UUID jobId, int progress) {
return jdbcClient.sql("""
UPDATE creation.generation_jobs
SET progress = :progress
WHERE id = :id AND status = 'running'
""")
.param("id", jobId)
.param("progress", (short) progress)
.update();
}
/** RUNNING → SUCCEEDED. progress is forced to 100 by the state CHECK. */
public int markSucceeded(UUID jobId, UUID outputAssetId) {
return jdbcClient.sql("""
UPDATE creation.generation_jobs
SET status = 'succeeded',
progress = 100,
output_asset_id = :assetId,
completed_at = now(),
lease_owner = NULL,
lease_expires_at = NULL
WHERE id = :id AND status = 'running'
""")
.param("id", jobId)
.param("assetId", outputAssetId)
.update();
}
/** RUNNING → FAILED, terminal: the retry budget is spent. */
public int markFailed(UUID jobId, String errorCode, String errorMessage) {
return jdbcClient.sql("""
UPDATE creation.generation_jobs
SET status = 'failed',
error_code = :errorCode,
error_message = :errorMessage,
completed_at = now(),
lease_owner = NULL,
lease_expires_at = NULL
WHERE id = :id AND status = 'running'
""")
.param("id", jobId)
.param("errorCode", errorCode)
.param("errorMessage", errorMessage)
.update();
}
/**
* RUNNING → QUEUED with backoff, for a failure that still has retries
* left. {@code provider_request_id} is preserved for the same reason as in
* {@link #reapExpiredLeases(int)}.
*/
public int requeueWithBackoff(UUID jobId, Duration backoff) {
return jdbcClient.sql("""
UPDATE creation.generation_jobs
SET status = 'queued',
progress = 0,
started_at = NULL,
lease_owner = NULL,
lease_expires_at = NULL,
next_attempt_at = now() + :backoff::interval
WHERE id = :id AND status = 'running'
""")
.param("id", jobId)
.param("backoff", backoff.toSeconds() + " seconds")
.update();
}
/**
* Cancels a job that has not reached a terminal state. Guarded by status
* in the WHERE clause rather than by a read-then-write so a concurrent
* completion cannot be overwritten — 0 rows means "it finished first".
*/
public int markCancelled(UUID jobId, UUID userId) {
return jdbcClient.sql("""
UPDATE creation.generation_jobs
SET status = 'cancelled',
completed_at = now(),
lease_owner = NULL,
lease_expires_at = NULL
WHERE id = :id AND user_id = :userId
AND status IN ('queued', 'running')
""")
.param("id", jobId)
.param("userId", userId)
.update();
}
// ----------------------------------------------------------- row mapper
private static GenerationJob mapRow(java.sql.ResultSet rs, int rowNum) throws java.sql.SQLException {
return new GenerationJob(
rs.getObject("id", UUID.class),
rs.getObject("user_id", UUID.class),
rs.getObject("pet_id", UUID.class),
MediaKind.fromDb(rs.getString("media_kind")),
rs.getObject("model_id", UUID.class),
rs.getObject("style_id", UUID.class),
rs.getObject("input_asset_id", UUID.class),
rs.getObject("output_asset_id", UUID.class),
rs.getString("prompt"),
rs.getString("negative_prompt"),
rs.getObject("width_px", Integer.class),
rs.getObject("height_px", Integer.class),
rs.getObject("duration_ms", Long.class),
JobStatus.fromDb(rs.getString("status")),
rs.getShort("progress"),
rs.getShort("attempt_count"),
rs.getShort("max_attempts"),
rs.getString("provider_code_snapshot"),
rs.getString("provider_request_id"),
rs.getString("error_code"),
rs.getString("error_message"),
rs.getObject("created_at", OffsetDateTime.class),
rs.getObject("first_attempt_at", OffsetDateTime.class),
rs.getObject("started_at", OffsetDateTime.class),
rs.getObject("completed_at", OffsetDateTime.class),
rs.getObject("lease_expires_at", OffsetDateTime.class));
}
}
@@ -0,0 +1,83 @@
package com.patbond.patbond.creation.security;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.patbond.patbond.common.error.BusinessException;
import com.patbond.patbond.common.error.ErrorCode;
import com.patbond.patbond.common.response.ApiResponse;
import io.jsonwebtoken.Claims;
import jakarta.servlet.FilterChain;
import jakarta.servlet.ServletException;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.MediaType;
import org.springframework.web.filter.OncePerRequestFilter;
import java.io.IOException;
import java.util.UUID;
/**
* Bearer authentication for /api/v1/** community endpoints. Verifies the
* RS256 signature locally with the auth service's public key and exposes the
* authenticated user id as a request attribute. Missing, forged or expired
* tokens all answer 401/40101 without detail.
*
* <p>FOURTH copy of the user/pet/community filter. Sinking the shared
* pure-Java parts into patbond-common is registered as a separate M4 ticket
* (ADR-023 §2); the retreat position is exactly this file. Note that the
* service-identity tokens this module ISSUES (ServiceTokenSigner) are signed
* with a DIFFERENT key pair and are never accepted here — precisely because
* this filter verifies the signature only and checks neither aud nor iss
* (ADR-023 §5).
*/
public class BearerAuthFilter extends OncePerRequestFilter {
/** Request attribute holding the authenticated user's UUID. */
public static final String USER_ID_ATTRIBUTE = "patbond.authenticatedUserId";
private static final Logger log = LoggerFactory.getLogger(BearerAuthFilter.class);
private final JwtVerifier jwtVerifier;
private final ObjectMapper objectMapper;
public BearerAuthFilter(JwtVerifier jwtVerifier, ObjectMapper objectMapper) {
this.jwtVerifier = jwtVerifier;
this.objectMapper = objectMapper;
}
@Override
protected boolean shouldNotFilter(HttpServletRequest request) {
return !request.getRequestURI().startsWith("/api/v1/");
}
@Override
protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response,
FilterChain filterChain) throws ServletException, IOException {
String header = request.getHeader("Authorization");
if (header == null || !header.startsWith("Bearer ")) {
reject(response, ErrorCode.TOKEN_INVALID);
return;
}
try {
Claims claims = jwtVerifier.verify(header.substring("Bearer ".length()).trim());
request.setAttribute(USER_ID_ATTRIBUTE, UUID.fromString(claims.getSubject()));
} catch (BusinessException e) {
reject(response, ErrorCode.TOKEN_INVALID);
return;
} catch (IllegalStateException | IllegalArgumentException e) {
log.error("Access token verification unavailable: {}", e.getMessage());
reject(response, ErrorCode.INTERNAL_ERROR);
return;
}
filterChain.doFilter(request, response);
}
private void reject(HttpServletResponse response, ErrorCode errorCode) throws IOException {
response.setStatus(errorCode.getHttpStatus());
response.setContentType(MediaType.APPLICATION_JSON_VALUE);
response.setCharacterEncoding("UTF-8");
objectMapper.writeValue(response.getWriter(),
ApiResponse.failure(errorCode.getCode(), errorCode.getDefaultMessage()));
}
}
@@ -0,0 +1,41 @@
package com.patbond.patbond.creation.security;
import com.patbond.patbond.common.error.BusinessException;
import com.patbond.patbond.common.error.ErrorCode;
import io.jsonwebtoken.Claims;
import io.jsonwebtoken.JwtException;
import io.jsonwebtoken.JwtParser;
import io.jsonwebtoken.Jwts;
/**
* Verifies RS256 access tokens issued by patbond-auth against the configured
* public key. The key is optional at startup so contexts that never serve
* protected routes (most tests) can boot without one; any verification
* attempt without a key fails loudly as a server misconfiguration instead of
* being reported to the client as an authentication problem.
*/
public class JwtVerifier {
private final JwtParser parser;
public JwtVerifier(String publicKeyLocation) {
this.parser = publicKeyLocation == null || publicKeyLocation.isBlank()
? null
: Jwts.parser().verifyWith(RsaPublicKeyLoader.load(publicKeyLocation)).build();
}
/**
* @return the verified claims
* @throws BusinessException 40101 when the token is forged, malformed or expired
*/
public Claims verify(String token) {
if (parser == null) {
throw new IllegalStateException("patbond.jwt.public-key 未配置,无法校验 access token");
}
try {
return parser.parseSignedClaims(token).getPayload();
} catch (JwtException | IllegalArgumentException e) {
throw new BusinessException(ErrorCode.TOKEN_INVALID);
}
}
}
@@ -0,0 +1,59 @@
package com.patbond.patbond.creation.security;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.KeyFactory;
import java.security.NoSuchAlgorithmException;
import java.security.interfaces.RSAPrivateCrtKey;
import java.security.interfaces.RSAPublicKey;
import java.security.spec.InvalidKeySpecException;
import java.security.spec.PKCS8EncodedKeySpec;
import java.security.spec.RSAPublicKeySpec;
import java.util.Base64;
/**
* Loads the RS256 signing key from either inline PEM content (value starts
* with -----BEGIN, e.g. injected through an environment variable) or a
* filesystem path to a PEM file. Only the PKCS#8 form produced by
* `openssl genpkey` is supported. The matching public key is derived from
* the CRT parameters, so this service needs no second configuration value.
*/
public final class RsaPrivateKeyLoader {
private RsaPrivateKeyLoader() {
}
public static RSAPrivateCrtKey load(String pemOrPath) {
String pem = pemOrPath.trim();
if (!pem.startsWith("-----BEGIN")) {
try {
pem = Files.readString(Path.of(pem));
} catch (IOException e) {
throw new UncheckedIOException("无法读取 JWT 私钥文件: " + pemOrPath, e);
}
}
String base64 = pem
.replace("-----BEGIN PRIVATE KEY-----", "")
.replace("-----END PRIVATE KEY-----", "")
.replaceAll("\\s", "");
try {
byte[] der = Base64.getDecoder().decode(base64);
KeyFactory keyFactory = KeyFactory.getInstance("RSA");
return (RSAPrivateCrtKey) keyFactory.generatePrivate(new PKCS8EncodedKeySpec(der));
} catch (IllegalArgumentException | ClassCastException
| NoSuchAlgorithmException | InvalidKeySpecException e) {
throw new IllegalStateException("JWT 私钥不是有效的 PEM(PKCS#8) RSA 私钥", e);
}
}
public static RSAPublicKey derivePublicKey(RSAPrivateCrtKey privateKey) {
try {
return (RSAPublicKey) KeyFactory.getInstance("RSA").generatePublic(
new RSAPublicKeySpec(privateKey.getModulus(), privateKey.getPublicExponent()));
} catch (NoSuchAlgorithmException | InvalidKeySpecException e) {
throw new IllegalStateException("无法从 RSA 私钥推导公钥", e);
}
}
}
@@ -0,0 +1,43 @@
package com.patbond.patbond.creation.security;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.KeyFactory;
import java.security.NoSuchAlgorithmException;
import java.security.PublicKey;
import java.security.spec.InvalidKeySpecException;
import java.security.spec.X509EncodedKeySpec;
import java.util.Base64;
/**
* Loads an RSA public key for JWT signature verification. Accepts either
* inline PEM or a filesystem path; the inline form (environment variable
* PATBOND_JWT_PUBLIC_KEY starting with -----BEGIN) is production's choice
* because mounted secrets beat files-in-the-image.
*/
public final class RsaPublicKeyLoader {
private RsaPublicKeyLoader() {
}
public static PublicKey load(String pemOrPath) {
String pem = pemOrPath.startsWith("-----BEGIN") ? pemOrPath : readFile(pemOrPath);
String stripped = pem.replaceAll("-----BEGIN PUBLIC KEY-----|-----END PUBLIC KEY-----", "")
.replaceAll("\\s", "");
byte[] decoded = Base64.getDecoder().decode(stripped);
try {
return KeyFactory.getInstance("RSA").generatePublic(new X509EncodedKeySpec(decoded));
} catch (NoSuchAlgorithmException | InvalidKeySpecException e) {
throw new IllegalArgumentException("Invalid RSA public key", e);
}
}
private static String readFile(String path) {
try {
return Files.readString(Path.of(path));
} catch (IOException e) {
throw new IllegalArgumentException("Cannot read public key from " + path, e);
}
}
}
@@ -0,0 +1,87 @@
package com.patbond.patbond.creation.security;
import io.jsonwebtoken.Jwts;
import java.security.interfaces.RSAPrivateCrtKey;
import java.time.Duration;
import java.time.Instant;
import java.util.Date;
import java.util.UUID;
/**
* Signs short-lived service-identity tokens for outbound calls to patbond-ai
* (ADR-023 §5).
*
* <p><b>Why a separate key pair, and why not just forward the user's token.</b>
* Forwarding is wrong on two counts: the worker runs asynchronously, so the
* user's token may well have expired by the time the job is picked up, and
* patbond-ai has no business holding user credentials.
*
* <p>Reusing the USER key pair for a service token would be worse still, and
* for a reason specific to this codebase: {@link JwtVerifier} verifies the
* signature and nothing else — it checks neither {@code aud} nor {@code iss} —
* and the user tokens patbond-auth issues carry no {@code aud} claim at all. A
* service token signed with that key pair would therefore be accepted as a
* valid user token by user, pet, community and this module. It would probably
* fall over later when {@code sub} failed to parse as a UUID, but that is an
* accident stopping it, not a design. A separate key pair needs no change to
* any already-released verifier.
*
* <p>The private key lives only here; patbond-ai holds the matching public key
* and MUST enforce {@code aud} and {@code typ}.
*/
public class ServiceTokenSigner {
/** Marks the token as a service identity rather than an end user. */
public static final String TYPE_CLAIM = "typ";
public static final String SERVICE_TYPE = "service";
private final RSAPrivateCrtKey privateKey;
private final String issuer;
private final String audience;
private final String subject;
private final Duration ttl;
public ServiceTokenSigner(String privateKeyLocation, String issuer, String audience,
String subject, Duration ttl) {
// Tolerated at construction, fatal on use — the same contract as
// JwtVerifier's optional public key. A context that never calls
// patbond-ai (most tests, and a request-only replica) must be able to
// boot without a key, but a missing key must never be mistaken for an
// upstream problem: it is a server misconfiguration and says so.
this.privateKey = privateKeyLocation == null || privateKeyLocation.isBlank()
? null
: RsaPrivateKeyLoader.load(privateKeyLocation);
this.issuer = issuer;
this.audience = audience;
this.subject = subject;
this.ttl = ttl;
}
/**
* Mints a token for one outbound call. Tokens are signed per call and never
* cached: the TTL is minutes, signing is cheap, and a cached token outliving
* its purpose is the kind of thing that quietly becomes a long-lived secret.
*
* @throws IllegalStateException when no private key is configured
*/
public String sign() {
if (privateKey == null) {
throw new IllegalStateException(
"patbond.creation.service-token.private-key 未配置:请生成一对"
+ "**独立于用户 token 的** RS256 密钥并通过环境变量注入"
+ "(见 application.yml.sample)");
}
Instant now = Instant.now();
return Jwts.builder()
.issuer(issuer)
.audience().add(audience).and()
.subject(subject)
.claim(TYPE_CLAIM, SERVICE_TYPE)
.id(UUID.randomUUID().toString())
.issuedAt(Date.from(now))
.expiration(Date.from(now.plus(ttl)))
.signWith(privateKey)
.compact();
}
}
@@ -0,0 +1,140 @@
package com.patbond.patbond.creation.service;
import com.patbond.patbond.common.error.BusinessException;
import com.patbond.patbond.common.error.ErrorCode;
import com.patbond.patbond.creation.config.CreationProperties;
import com.patbond.patbond.creation.domain.GenerationJob;
import com.patbond.patbond.creation.domain.MediaKind;
import com.patbond.patbond.creation.dto.CreateGenerationJobRequest;
import com.patbond.patbond.creation.dto.GenerationJobResponse;
import com.patbond.patbond.creation.repository.CatalogRepository;
import com.patbond.patbond.creation.repository.GenerationJobRepository;
import com.patbond.patbond.creation.support.RequestHashes;
import com.patbond.patbond.creation.support.UuidV7;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
/**
* Generation job lifecycle as the client drives it: create, read, list, cancel.
* Execution is the worker's business — creating a job only enqueues it.
*/
@Service
public class GenerationService {
private final GenerationJobRepository jobs;
private final CatalogRepository catalog;
private final CreationProperties properties;
public GenerationService(GenerationJobRepository jobs, CatalogRepository catalog,
CreationProperties properties) {
this.jobs = jobs;
this.catalog = catalog;
this.properties = properties;
}
/**
* Enqueues a job, or returns the existing one for a repeated
* Idempotency-Key.
*
* <p>Idempotency follows ADR-019: same key + same payload returns the first
* job, same key + different payload answers 40905. The check runs before the
* insert as a courtesy, but correctness does not depend on it — the unique
* constraint {@code (user_id, idempotency_key)} is the real guard, so two
* concurrent identical requests cannot both create a job.
*/
@Transactional
public GenerationJobResponse create(UUID userId, String idempotencyKey,
CreateGenerationJobRequest request) {
MediaKind mediaKind = parseMediaKind(request.mediaKind());
byte[] requestHash = RequestHashes.sha256(request.canonical());
Optional<UUID> existing = jobs.findIdByIdempotencyKey(userId, idempotencyKey, requestHash);
if (existing.isPresent()) {
return jobs.findByIdForUser(existing.get(), userId)
.map(this::toResponse)
.orElseThrow(() -> new BusinessException(ErrorCode.GENERATION_JOB_NOT_FOUND));
}
if (jobs.idempotencyKeyConflicts(userId, idempotencyKey, requestHash)) {
throw new BusinessException(ErrorCode.IDEMPOTENCY_PAYLOAD_MISMATCH);
}
// Resolving the catalog here does two things: it turns a bad model id
// into a clean 404 instead of a composite-FK violation, and it yields
// the provider values frozen onto the job. Those snapshots are why a job
// stays pollable after the catalog is re-pointed or disabled.
CatalogRepository.ModelRow model = catalog.findEnabledModel(request.modelId(), mediaKind)
.orElseThrow(() -> new BusinessException(ErrorCode.GENERATION_MODEL_NOT_FOUND));
String styleCode = null;
if (request.styleId() != null) {
styleCode = catalog.findEnabledStyle(request.styleId(), mediaKind)
.orElseThrow(() -> new BusinessException(ErrorCode.GENERATION_MODEL_NOT_FOUND))
.code();
}
UUID jobId = UuidV7.generate();
jobs.insertQueued(jobId, userId, request.petId(), mediaKind,
model.id(), request.styleId(), request.inputAssetId(),
request.prompt(), request.negativePrompt(),
request.widthPx(), request.heightPx(), request.durationMs(),
model.providerCode(), model.providerModelName(), model.modelVersion(),
styleCode, idempotencyKey, requestHash);
return jobs.findByIdForUser(jobId, userId)
.map(this::toResponse)
.orElseThrow(() -> new BusinessException(ErrorCode.INTERNAL_ERROR));
}
public GenerationJobResponse get(UUID userId, UUID jobId) {
return jobs.findByIdForUser(jobId, userId)
.map(this::toResponse)
// Another user's job and a non-existent job answer identically,
// so the endpoint cannot be used to probe for job ids.
.orElseThrow(() -> new BusinessException(ErrorCode.GENERATION_JOB_NOT_FOUND));
}
public List<GenerationJobResponse> list(UUID userId, int limit) {
return jobs.listForUser(userId, limit).stream().map(this::toResponse).toList();
}
/**
* Cancels a job that has not finished.
*
* <p>The status guard lives in the UPDATE's WHERE clause rather than in a
* read-then-write, so a job that completes concurrently is not overwritten:
* 0 rows affected means it reached a terminal state first, which is reported
* as 42206 rather than pretended to be a successful cancellation.
*
* <p>Note this cancels the job record only. Whether the upstream request can
* also be withdrawn depends on the vendor and is not knowable until their
* documentation lands; until then a cancelled job may still be paid for.
*/
@Transactional
public GenerationJobResponse cancel(UUID userId, UUID jobId) {
int cancelled = jobs.markCancelled(jobId, userId);
if (cancelled == 0) {
GenerationJob job = jobs.findByIdForUser(jobId, userId)
.orElseThrow(() -> new BusinessException(ErrorCode.GENERATION_JOB_NOT_FOUND));
throw new BusinessException(ErrorCode.GENERATION_STATE_INVALID,
"任务已处于终态:" + job.status().dbValue());
}
return get(userId, jobId);
}
private GenerationJobResponse toResponse(GenerationJob job) {
return GenerationJobResponse.of(job,
properties.clientTimeoutFor(job.mediaKind()),
properties.staleAfterFor(job.mediaKind()));
}
private static MediaKind parseMediaKind(String value) {
try {
return MediaKind.fromDb(value);
} catch (IllegalArgumentException e) {
throw new BusinessException(ErrorCode.VALIDATION_ERROR, "mediaKind 只能是 image 或 video");
}
}
}
@@ -0,0 +1,30 @@
package com.patbond.patbond.creation.support;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
/**
* SHA-256 of the canonical form of a create request (ADR-019: creation-type
* writes carry a mandatory Idempotency-Key and the request body's hash is
* stored next to it — a keyed retry with the same payload returns the first
* write, a different payload answers 40905). Hashing the NORMALIZED command
* (trimmed fields, defaults applied, media positions resolved) rather than
* the raw bytes makes the comparison insensitive to JSON formatting while
* still catching every semantic difference. 32 bytes, matching
* ck_posts_idempotency's octet_length(request_hash) = 32.
*/
public final class RequestHashes {
private RequestHashes() {
}
public static byte[] sha256(String canonical) {
try {
return MessageDigest.getInstance("SHA-256")
.digest(canonical.getBytes(StandardCharsets.UTF_8));
} catch (NoSuchAlgorithmException e) {
throw new IllegalStateException(e);
}
}
}
@@ -0,0 +1,31 @@
package com.patbond.patbond.creation.support;
import java.security.SecureRandom;
import java.util.UUID;
/**
* Application-side UUIDv7 generator (RFC 9562): 48-bit Unix millisecond
* timestamp, version/variant bits, 74 random bits. Time-ordered values keep
* B-tree page churn low on uuid primary keys; the database DEFAULT
* gen_random_uuid() remains the fallback for rows not inserted through the
* application. Fourth copy after patbond-user/pet/community — the services
* deploy independently and patbond-common stays contract-only; upgrading the
* shared artifacts is an M4 ticket of its own (ADR-023 §2).
*/
public final class UuidV7 {
private static final SecureRandom RANDOM = new SecureRandom();
private UuidV7() {
}
public static UUID generate() {
long timestampMs = System.currentTimeMillis();
long randA = RANDOM.nextLong() & 0x0FFFL;
long randB = RANDOM.nextLong() & 0x3FFFFFFFFFFFFFFFL;
long msb = (timestampMs << 16) | 0x7000L | randA;
long lsb = 0x8000000000000000L | randB;
return new UUID(msb, lsb);
}
}
@@ -0,0 +1,68 @@
package com.patbond.patbond.creation.web;
import com.patbond.patbond.common.error.BusinessException;
import com.patbond.patbond.common.error.ErrorCode;
import com.patbond.patbond.common.response.ApiResponse;
import jakarta.validation.ConstraintViolationException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.ResponseEntity;
import org.springframework.http.converter.HttpMessageNotReadableException;
import org.springframework.validation.FieldError;
import org.springframework.web.bind.MethodArgumentNotValidException;
import org.springframework.web.bind.MissingRequestHeaderException;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.RestControllerAdvice;
import org.springframework.web.method.annotation.HandlerMethodValidationException;
import org.springframework.web.method.annotation.MethodArgumentTypeMismatchException;
import org.springframework.web.servlet.resource.NoResourceFoundException;
/**
* Single place that turns exceptions into the {code, message, data} envelope
* with a matching HTTP status (development-plan 6.1) — same contract as the
* user/auth/pet handlers. Unexpected exceptions are logged in full but never
* leak internals to the client.
*/
@RestControllerAdvice
public class GlobalExceptionHandler {
private static final Logger log = LoggerFactory.getLogger(GlobalExceptionHandler.class);
@ExceptionHandler(BusinessException.class)
public ResponseEntity<ApiResponse<Void>> handleBusiness(BusinessException e) {
return ResponseEntity.status(e.getHttpStatus())
.body(ApiResponse.failure(e.getCode(), e.getMessage()));
}
@ExceptionHandler(MethodArgumentNotValidException.class)
public ResponseEntity<ApiResponse<Void>> handleValidation(MethodArgumentNotValidException e) {
String message = e.getBindingResult().getFieldErrors().stream()
.findFirst()
.map(FieldError::getDefaultMessage)
.orElse(ErrorCode.VALIDATION_ERROR.getDefaultMessage());
return failure(ErrorCode.VALIDATION_ERROR, message);
}
@ExceptionHandler({HttpMessageNotReadableException.class, MethodArgumentTypeMismatchException.class,
ConstraintViolationException.class, HandlerMethodValidationException.class,
MissingRequestHeaderException.class})
public ResponseEntity<ApiResponse<Void>> handleMalformedRequest(Exception e) {
return failure(ErrorCode.VALIDATION_ERROR, ErrorCode.VALIDATION_ERROR.getDefaultMessage());
}
@ExceptionHandler(NoResourceFoundException.class)
public ResponseEntity<ApiResponse<Void>> handleNoResource(NoResourceFoundException e) {
return ResponseEntity.status(404).body(ApiResponse.failure(40400, "资源不存在"));
}
@ExceptionHandler(Exception.class)
public ResponseEntity<ApiResponse<Void>> handleUnexpected(Exception e) {
log.error("Unhandled exception", e);
return failure(ErrorCode.INTERNAL_ERROR, ErrorCode.INTERNAL_ERROR.getDefaultMessage());
}
private static ResponseEntity<ApiResponse<Void>> failure(ErrorCode errorCode, String message) {
return ResponseEntity.status(errorCode.getHttpStatus())
.body(ApiResponse.failure(errorCode.getCode(), message));
}
}
@@ -0,0 +1,228 @@
package com.patbond.patbond.creation.worker;
import com.patbond.patbond.creation.config.CreationProperties;
import com.patbond.patbond.creation.config.ProviderRegistry;
import com.patbond.patbond.creation.domain.GenerationJob;
import com.patbond.patbond.creation.provider.GenerationProvider;
import com.patbond.patbond.creation.repository.GenerationJobRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.time.OffsetDateTime;
import java.util.List;
/**
* The lease queue worker (ADR-023 §3/§4).
*
* <p>Three responsibilities, each a separate scheduled pass so a stall in one
* cannot starve the others:
* <ol>
* <li>{@link #submitPass()} — claim due QUEUED jobs and hand them upstream;</li>
* <li>{@link #pollPass()} — ask after RUNNING jobs this worker holds, renewing
* the lease in the same statement;</li>
* <li>{@link #reapPass()} — return expired-lease jobs to the queue.</li>
* </ol>
*
* <p><b>Crash safety comes from the database, not from this class.</b> No job
* state is held in memory between ticks: a claim writes RUNNING plus a lease, a
* poll renews it, and a worker that dies simply stops renewing until the reaper
* requeues the row. That is why no broker is needed and why replicas may be
* restarted freely.
*
* <p><b>Why the ordering inside {@link #submitPass()} matters.</b>
* {@code provider_request_id} is persisted BEFORE the provider is asked to do
* anything billable. If the call then times out we do not know whether the
* vendor charged us — but we do know the request id, so the next attempt polls
* instead of re-submitting. Reverse the order and every timeout becomes a
* double charge.
*/
@Component
@ConditionalOnProperty(prefix = "patbond.creation.worker", name = "enabled",
havingValue = "true", matchIfMissing = true)
public class GenerationWorker {
private static final Logger log = LoggerFactory.getLogger(GenerationWorker.class);
private final GenerationJobRepository repository;
private final ProviderRegistry providers;
private final CreationProperties properties;
private final UploadTargetGateway uploadTargets;
private final String leaseOwner;
public GenerationWorker(GenerationJobRepository repository, ProviderRegistry providers,
CreationProperties properties, UploadTargetGateway uploadTargets) {
this.repository = repository;
this.providers = providers;
this.properties = properties;
this.uploadTargets = uploadTargets;
this.leaseOwner = WorkerIdentity.current();
}
// ------------------------------------------------------------- submit
@Scheduled(fixedDelayString = "${patbond.creation.worker.tick-interval:3s}")
public void submitPass() {
CreationProperties.Worker config = properties.getWorker();
List<GenerationJob> claimed;
try {
claimed = repository.claimQueued(leaseOwner, config.getLeaseTtl(), config.getBatchSize());
} catch (RuntimeException e) {
// A failing claim must not kill the scheduled task: Spring cancels a
// fixedDelay task whose method throws, which would silently stop the
// whole queue until the next restart.
log.error("Claiming queued generation jobs failed", e);
return;
}
for (GenerationJob job : claimed) {
try {
submitOne(job);
} catch (RuntimeException e) {
log.error("Submitting generation job {} failed", job.id(), e);
failOrRetry(job, "PROVIDER_SUBMIT_FAILED", e.getMessage(), true);
}
}
}
private void submitOne(GenerationJob job) {
GenerationProvider provider = providers.find(job.providerCodeSnapshot()).orElse(null);
if (provider == null) {
// The adapter that submitted this job is gone. Retrying cannot help,
// so fail terminally rather than loop until the budget is spent.
log.error("No provider adapter for code {} (job {}); known codes: {}",
job.providerCodeSnapshot(), job.id(), providers.knownCodes());
repository.markFailed(job.id(), "PROVIDER_UNKNOWN",
"生成服务适配器不可用:" + job.providerCodeSnapshot());
return;
}
if (job.isSubmittedUpstream()) {
// Already handed over on an earlier attempt (timeout, or a worker
// that died after submitting). Do NOT submit again — poll.
log.info("Job {} already has provider_request_id {}, polling instead of re-submitting",
job.id(), job.providerRequestId());
pollOne(job, provider);
return;
}
String uploadTarget = uploadTargets.presignedPutFor(job);
GenerationProvider.SubmitResult result =
provider.submit(job, uploadTarget, job.id().toString());
int recorded = repository.recordProviderRequest(job.id(), result.providerRequestId());
if (recorded == 0) {
// Another attempt recorded an id first; ours is a duplicate
// submission we cannot un-send. Worth a loud log — it means the
// guard above was bypassed by a race.
log.warn("provider_request_id for job {} was already set; possible duplicate submission",
job.id());
}
}
// --------------------------------------------------------------- poll
@Scheduled(fixedDelayString = "${patbond.creation.worker.tick-interval:3s}")
public void pollPass() {
CreationProperties.Worker config = properties.getWorker();
List<GenerationJob> due;
try {
// "Due for a poll" is expressed as "lease is within one poll
// interval of expiring", so renewal and polling stay one decision.
due = repository.claimRunningDueForPoll(leaseOwner, config.getLeaseTtl(),
config.getPollIntervalEarly(), config.getBatchSize());
} catch (RuntimeException e) {
log.error("Claiming running generation jobs failed", e);
return;
}
for (GenerationJob job : due) {
GenerationProvider provider = providers.find(job.providerCodeSnapshot()).orElse(null);
if (provider == null || !job.isSubmittedUpstream()) {
continue;
}
try {
pollOne(job, provider);
} catch (RuntimeException e) {
log.error("Polling generation job {} failed", job.id(), e);
}
}
}
private void pollOne(GenerationJob job, GenerationProvider provider) {
GenerationProvider.PollResult result = provider.poll(job, job.providerRequestId());
switch (result.state()) {
case PENDING, RUNNING -> {
if (result.progress() > job.progress()) {
repository.updateProgress(job.id(), result.progress());
}
}
case SUCCEEDED -> {
// The artifact is already in object storage (the provider PUT it
// to the presigned target). Turning the object key into a
// media.assets row is patbond-user's job — ADR-017 keeps the
// media write flow there — so this step waits on the internal
// media API, which in turn waits on the storage decision.
throw new UnsupportedOperationException(
"产物入库未实现:objectKey=" + result.objectKey()
+ "。需先落地 patbond-user 的 /internal/media/** 确认端点,"
+ "且其返回形态取决于对象存储方案(ADR-023 §7 挂起)。");
}
case FAILED -> failOrRetry(job, result.errorCode(), result.errorMessage(),
result.retryable());
}
}
// ---------------------------------------------------------------- reap
@Scheduled(fixedDelayString = "${patbond.creation.worker.tick-interval:3s}")
public void reapPass() {
try {
int reaped = repository.reapExpiredLeases(properties.getWorker().getBatchSize());
if (reaped > 0) {
// Expected after a worker restart; unexpected in steady state,
// where it means the lease TTL is shorter than a poll cycle.
log.warn("Returned {} generation job(s) with expired leases to the queue", reaped);
}
} catch (RuntimeException e) {
log.error("Reaping expired generation leases failed", e);
}
}
// ------------------------------------------------------------- helpers
/**
* Terminal failure or another attempt, decided by the retry budget and
* whether the failure is retryable at all. A rejected prompt is not worth
* retrying; a vendor 5xx is.
*/
private void failOrRetry(GenerationJob job, String errorCode, String errorMessage,
boolean retryable) {
if (!retryable || job.isLastAttempt()) {
repository.markFailed(job.id(), errorCode, truncate(errorMessage));
return;
}
Duration backoff = properties.getWorker().retryBackoffFor(job.attemptCount());
repository.requeueWithBackoff(job.id(), backoff);
}
/** error_message is varchar(1000); a vendor stack trace must not blow the insert. */
private static String truncate(String message) {
if (message == null) {
return null;
}
return message.length() <= 1000 ? message : message.substring(0, 1000);
}
/** Exposed for tests that need to know which owner string to expect. */
public String leaseOwner() {
return leaseOwner;
}
/** @return how long the current attempt has been running, for the poll ladder */
static Duration runningFor(GenerationJob job) {
OffsetDateTime startedAt = job.startedAt();
return startedAt == null ? Duration.ZERO
: Duration.between(startedAt.toInstant(), OffsetDateTime.now().toInstant());
}
}
@@ -0,0 +1,41 @@
package com.patbond.patbond.creation.worker;
import com.patbond.patbond.creation.domain.GenerationJob;
import org.springframework.stereotype.Component;
/**
* Obtains a presigned PUT target for a job's artifact from patbond-user's
* internal media API.
*
* <p><b>Why this indirection instead of writing to the object store here.</b>
* ADR-017 keeps the media upload flow in patbond-user, and {@code ObjectStorage}
* already offers {@code presignPut}/{@code presignGet} — so nothing needs a new
* "server-side put" primitive. Handing patbond-ai a presigned URL means the
* generation service holds no object-storage credentials and the Python service
* holds none either: the artifact goes straight from the vendor to storage,
* never through this JVM. For video that is the difference between streaming
* bytes through two extra hops and not.
*
* <p>Unimplemented at skeleton stage, and for a specific reason: the endpoint's
* response shape depends on the object-storage decision, which is explicitly
* deferred (ADR-023 §7). With local object storage the reply is a presigned URL
* against the internal endpoint; with a cloud provider it is that vendor's
* upload credential, and patbond-ai additionally grows fetch code for it. Both
* are one method here — but guessing which would bake the pending decision into
* code.
*/
@Component
public class UploadTargetGateway {
/**
* @return a presigned PUT URL the provider may upload the artifact to
* @throws UnsupportedOperationException until the internal media endpoint and
* the storage decision land
*/
public String presignedPutFor(GenerationJob job) {
throw new UnsupportedOperationException(
"预签名上传目标获取未实现:需先在 patbond-user 落地 /internal/media/** "
+ "(签发 presigned PUT + confirm 建 media.assets 行),"
+ "且其返回形态取决于对象存储方案(ADR-023 §7 挂起)。job=" + job.id());
}
}
@@ -0,0 +1,35 @@
package com.patbond.patbond.creation.worker;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.UUID;
/**
* Produces the {@code lease_owner} string a worker instance claims jobs under.
*
* <p>It must be stable for the life of the process (a worker has to be able to
* find the jobs it holds) and distinct between processes (two replicas must not
* be able to poll each other's jobs). Host plus PID satisfies both; the random
* suffix covers the container case where two replicas can share a hostname and
* PID namespace, which would otherwise let them silently steal each other's
* leases.
*
* <p>Capped at 128 characters to match {@code lease_owner varchar(128)}.
*/
final class WorkerIdentity {
private WorkerIdentity() {
}
static String current() {
String host;
try {
host = InetAddress.getLocalHost().getHostName();
} catch (UnknownHostException e) {
host = "unknown-host";
}
String identity = host + "/" + ProcessHandle.current().pid()
+ "/" + UUID.randomUUID().toString().substring(0, 8);
return identity.length() <= 128 ? identity : identity.substring(identity.length() - 128);
}
}
@@ -0,0 +1,86 @@
server:
port: ${PATBOND_CREATION_PORT:8085}
spring:
application:
name: patbond-creation
datasource:
# 与 patbond-user 共库(MVP 单库多 schema);本服务只读写 creation schema,
# 并只读 community.posts 的 generation_job_id 关联(建草稿由 community 端点完成)。
# Flyway 迁移链(V1..V6,含 creation 基线)由 patbond-user 启动时统一执行,
# 本服务不携带 Flyway —— 单一 flyway_schema_history 归属不拆(ADR-023 §2)。
url: ${PATBOND_DB_URL:jdbc:postgresql://127.0.0.1:5432/patbond}
username: ${PATBOND_DB_USER:patbond}
password: ${PATBOND_DB_PASSWORD:patbond}
# /api/v1/** 业务端点自骨架起即接 RS256 校验,与 user/pet/community 同一约定。
patbond:
jwt:
# RS256 公钥,用于本地校验 patbond-auth 签发的 access token。
# 值可以是 PEM 文件路径,也可以是内联 PEM 内容(以 -----BEGIN 开头)。
# 私钥只给 patbond-auth,绝不入库。
public-key: ${PATBOND_JWT_PUBLIC_KEY:}
# /internal/** 服务间共享密钥(Java↔Java 既有通道),需与 patbond-user 同值。
# 注意:调用 patbond-ai 用的**不是**这个 token,而是下面的服务身份 JWT
# ——两套机制并存的原因见 ADR-023 §5。
internal-token: ${PATBOND_INTERNAL_TOKEN:dev-only-internal-token}
user-service:
# 预签名上传凭证来源:patbond-user 的 /internal/media/**(ADR-017 边界:
# 媒体上传流程归 user,本服务不直连对象存储、也不持有其凭证)。
url: ${PATBOND_USER_SERVICE_URL:http://127.0.0.1:8082}
creation:
# ---- 上游生成服务(patbond-ai,Python)----
ai-service:
url: ${PATBOND_AI_SERVICE_URL:http://127.0.0.1:8000}
# 单次 HTTP 调用超时。注意这是**提交/查询**的超时,不是生成耗时——
# Java↔Python 边界一律异步(ADR-023 §4),故此值与出图出片时长无关,
# 不得为了「等出图」而调大。
connect-timeout: ${PATBOND_AI_CONNECT_TIMEOUT:2s}
read-timeout: ${PATBOND_AI_READ_TIMEOUT:5s}
# ---- 服务身份 JWT(ADR-023 §5)----
# **独立密钥对**,与用户 token 的密钥对不同。理由:现有 JwtVerifier 只验
# 签名、不校验 aud/iss,用户 token 亦无 aud claim;若复用同一密钥对,
# 四个模块都会把服务票当合法用户 token 接受。独立密钥对无需改动任何
# 已发布的验证器。私钥只给本服务,公钥给 patbond-ai。
service-token:
private-key: ${PATBOND_SERVICE_JWT_PRIVATE_KEY:}
issuer: ${PATBOND_SERVICE_JWT_ISSUER:patbond-api}
audience: ${PATBOND_SERVICE_JWT_AUDIENCE:patbond-ai}
subject: ${PATBOND_SERVICE_JWT_SUBJECT:service:creation-worker}
# 短有效期:服务票每次调用现签,不缓存、不复用。
ttl: ${PATBOND_SERVICE_JWT_TTL:5m}
# ---- Worker 与租约(ADR-023 §3/§4)----
worker:
# 关掉后本服务只提供读写端点、不消费队列(便于单实例调试与测试隔离)。
enabled: ${PATBOND_CREATION_WORKER_ENABLED:true}
# 调度器 tick 间隔。真正的轮询节律由 next_attempt_at 控制,
# tick 只决定「最快多久发现一件该做的事」。
tick-interval: ${PATBOND_CREATION_TICK_INTERVAL:3s}
# 单 tick 最多领取多少条,防止一个实例吃光整个队列。
batch-size: ${PATBOND_CREATION_BATCH_SIZE:5}
# 租约时长。**必须大于最长的单次轮询间隔**(见下 poll-interval-late),
# 否则会出现「任务仍在跑但租约过期被他人抢占」。每次轮询顺带续租。
lease-ttl: ${PATBOND_CREATION_LEASE_TTL:60s}
# 服务端轮询 patbond-ai 的退避节律(ADR-023 §4)。视频任务绝对次数很多,
# 固定 3s 会产生数百次无谓轮询与续租写入。
poll-interval-early: ${PATBOND_CREATION_POLL_EARLY:3s} # 前 60 秒
poll-interval-mid: ${PATBOND_CREATION_POLL_MID:10s} # 60s ~ 5min
poll-interval-late: ${PATBOND_CREATION_POLL_LATE:30s} # 5min 以后
poll-early-until: ${PATBOND_CREATION_POLL_EARLY_UNTIL:60s}
poll-mid-until: ${PATBOND_CREATION_POLL_MID_UNTIL:5m}
# 失败重试的指数退避基数(实际间隔 = base * 2^(attempt-1))。
retry-backoff-base: ${PATBOND_CREATION_RETRY_BACKOFF:30s}
# 客户端硬超时由服务端下发而非客户端硬编码(ADR-023 §10):图片与视频
# 差一个量级,硬编码两个数会使换 provider 就要发版。
client-timeout-image: ${PATBOND_CREATION_CLIENT_TIMEOUT_IMAGE:5m}
client-timeout-video: ${PATBOND_CREATION_CLIENT_TIMEOUT_VIDEO:20m}
# 任务被判定为僵死的阈值(running 超过此时长)。对应客户端的第六个 UI 态,
# 它不是错误态(ADR-023 §10)。
stale-after-image: ${PATBOND_CREATION_STALE_IMAGE:10m}
stale-after-video: ${PATBOND_CREATION_STALE_VIDEO:40m}
@@ -0,0 +1,27 @@
package com.patbond.patbond.creation;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
import org.springframework.context.annotation.Bean;
import org.testcontainers.containers.PostgreSQLContainer;
import org.testcontainers.utility.DockerImageName;
/**
* Shared Testcontainers setup: a disposable postgres:18 (the production target
* version) wired into the Spring context via {@code @ServiceConnection}.
*
* <p>The production module carries no Flyway — the single migration chain is
* owned by patbond-user — but the TEST classpath adds patbond-user's jar plus
* Flyway, so Boot applies the full V1..V6 chain, including the creation schema
* these tests exercise, to the fresh container exactly as the shared database
* gets it in production (same mechanism as pet/community).
*/
@TestConfiguration(proxyBeanMethods = false)
public class TestcontainersConfiguration {
@Bean
@ServiceConnection
PostgreSQLContainer<?> postgresContainer() {
return new PostgreSQLContainer<>(DockerImageName.parse("postgres:18"));
}
}
@@ -0,0 +1,292 @@
package com.patbond.patbond.creation.repository;
import com.patbond.patbond.creation.TestcontainersConfiguration;
import com.patbond.patbond.creation.domain.GenerationJob;
import com.patbond.patbond.creation.domain.JobStatus;
import com.patbond.patbond.creation.domain.MediaKind;
import com.patbond.patbond.creation.support.UuidV7;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Import;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.test.context.TestPropertySource;
import java.time.Duration;
import java.util.List;
import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat;
/**
* The lease queue's actual semantics against a real postgres — the one part of
* the M4 skeleton that carries real logic rather than wiring.
*
* <p>Every test here pins behaviour that would still "work" if written wrongly
* and only show up as lost jobs or double vendor charges in production:
* SKIP LOCKED not handing the same row to two workers, a poll renewing the lease
* it depends on, the reaper preserving {@code provider_request_id}, and
* {@code first_attempt_at} surviving requeue.
*
* <p>The worker itself is disabled: these tests drive the repository directly so
* a scheduled tick cannot race the assertions.
*/
@SpringBootTest
@Import(TestcontainersConfiguration.class)
@TestPropertySource(properties = "patbond.creation.worker.enabled=false")
class GenerationJobQueueIntegrationTest {
@Autowired
private GenerationJobRepository repository;
@Autowired
private JdbcClient jdbcClient;
private static final Duration LEASE = Duration.ofSeconds(60);
@Test
void claimMovesJobToRunningUnderLeaseAndStampsFirstAttempt() {
UUID userId = insertUser("claim");
UUID modelId = insertModel("claim-model", MediaKind.IMAGE);
UUID jobId = insertQueued(userId, modelId, MediaKind.IMAGE);
List<GenerationJob> claimed = repository.claimQueued("worker-a", LEASE, 10);
assertThat(claimed).extracting(GenerationJob::id).contains(jobId);
GenerationJob job = claimed.stream().filter(j -> j.id().equals(jobId)).findFirst().orElseThrow();
assertThat(job.status()).isEqualTo(JobStatus.RUNNING);
assertThat(job.attemptCount()).isEqualTo((short) 1);
assertThat(job.firstAttemptAt()).isNotNull();
assertThat(job.startedAt()).isNotNull();
assertThat(job.leaseExpiresAt()).isNotNull();
}
@Test
void twoWorkersNeverClaimTheSameJob() {
UUID userId = insertUser("skiplocked");
UUID modelId = insertModel("skiplocked-model", MediaKind.IMAGE);
for (int i = 0; i < 6; i++) {
insertQueued(userId, modelId, MediaKind.IMAGE);
}
List<GenerationJob> first = repository.claimQueued("worker-a", LEASE, 3);
List<GenerationJob> second = repository.claimQueued("worker-b", LEASE, 3);
assertThat(first).hasSize(3);
assertThat(second).hasSize(3);
// The point of SKIP LOCKED: disjoint claims, no waiting, no duplicates.
assertThat(first).extracting(GenerationJob::id)
.doesNotContainAnyElementsOf(second.stream().map(GenerationJob::id).toList());
}
@Test
void claimSkipsJobsWhoseBackoffHasNotElapsed() {
UUID userId = insertUser("backoff");
UUID modelId = insertModel("backoff-model", MediaKind.IMAGE);
UUID jobId = insertQueued(userId, modelId, MediaKind.IMAGE);
jdbcClient.sql("UPDATE creation.generation_jobs "
+ "SET next_attempt_at = now() + interval '1 hour' WHERE id = :id")
.param("id", jobId)
.update();
assertThat(repository.claimQueued("worker-a", LEASE, 10))
.extracting(GenerationJob::id)
.doesNotContain(jobId);
}
@Test
void pollClaimRenewsTheLeaseItDependsOn() {
UUID userId = insertUser("renew");
UUID modelId = insertModel("renew-model", MediaKind.IMAGE);
UUID jobId = insertQueued(userId, modelId, MediaKind.IMAGE);
repository.claimQueued("worker-a", Duration.ofSeconds(5), 10);
// Due when the lease is within one poll interval of expiring.
List<GenerationJob> due = repository.claimRunningDueForPoll(
"worker-a", LEASE, Duration.ofSeconds(30), 10);
assertThat(due).extracting(GenerationJob::id).contains(jobId);
// Renewed to the longer TTL: polling without renewing would let a long
// upstream job be reclaimed mid-flight, which is how a provider gets
// paid twice.
long secondsLeft = jdbcClient.sql(
"SELECT EXTRACT(EPOCH FROM (lease_expires_at - now()))::bigint "
+ "FROM creation.generation_jobs WHERE id = :id")
.param("id", jobId)
.query(Long.class)
.single();
assertThat(secondsLeft).isGreaterThan(30L);
}
@Test
void pollClaimIgnoresJobsHeldByAnotherWorker() {
UUID userId = insertUser("otherowner");
UUID modelId = insertModel("otherowner-model", MediaKind.IMAGE);
UUID jobId = insertQueued(userId, modelId, MediaKind.IMAGE);
repository.claimQueued("worker-a", Duration.ofSeconds(5), 10);
assertThat(repository.claimRunningDueForPoll("worker-b", LEASE, Duration.ofSeconds(30), 10))
.extracting(GenerationJob::id)
.doesNotContain(jobId);
}
@Test
void reaperRequeuesExpiredLeaseButKeepsProviderRequestId() {
UUID userId = insertUser("reaper");
UUID modelId = insertModel("reaper-model", MediaKind.IMAGE);
UUID jobId = insertQueued(userId, modelId, MediaKind.IMAGE);
repository.claimQueued("worker-a", LEASE, 10);
// Unique per test on purpose: uq_generation_jobs_provider_request is a
// TABLE-WIDE partial unique index on (provider, request id), so reusing
// a literal across tests collides — which is the index doing its job.
String providerRequestId = "vendor-req-" + UUID.randomUUID();
repository.recordProviderRequest(jobId, providerRequestId);
// Simulate the worker dying: nobody renews, the lease lapses. The whole
// timeline has to move back, not just the lease — ck_generation_jobs_lease
// requires lease_expires_at > started_at and
// ck_generation_jobs_first_attempt requires first_attempt_at <= started_at,
// so an "expired lease on a job that started a moment ago" is a state the
// database rightly refuses to store. Simulating a plausible history rather
// than an impossible one is the point.
jdbcClient.sql("""
UPDATE creation.generation_jobs
SET first_attempt_at = now() - interval '2 minutes',
started_at = now() - interval '2 minutes',
lease_expires_at = now() - interval '1 second'
WHERE id = :id
""")
.param("id", jobId)
.update();
assertThat(repository.reapExpiredLeases(10)).isEqualTo(1);
GenerationJob job = repository.findByIdForUser(jobId, userId).orElseThrow();
assertThat(job.status()).isEqualTo(JobStatus.QUEUED);
assertThat(job.startedAt()).isNull();
// Preserved deliberately: it is how the next attempt learns the work was
// already submitted upstream and polls instead of re-submitting.
// Clearing it here would turn every worker crash into a double charge.
assertThat(job.providerRequestId()).isEqualTo(providerRequestId);
// And the reason the added column exists: the original attempt time
// survives, so queue wait stays separable from execution time.
assertThat(job.firstAttemptAt()).isNotNull();
assertThat(job.attemptCount()).isEqualTo((short) 1);
}
@Test
void secondClaimAfterRequeuePreservesFirstAttemptAndIncrementsCount() {
UUID userId = insertUser("secondclaim");
UUID modelId = insertModel("secondclaim-model", MediaKind.IMAGE);
UUID jobId = insertQueued(userId, modelId, MediaKind.IMAGE);
repository.claimQueued("worker-a", LEASE, 10);
var firstAttemptAt = repository.findByIdForUser(jobId, userId).orElseThrow().firstAttemptAt();
repository.requeueWithBackoff(jobId, Duration.ZERO);
repository.claimQueued("worker-a", LEASE, 10);
GenerationJob job = repository.findByIdForUser(jobId, userId).orElseThrow();
assertThat(job.attemptCount()).isEqualTo((short) 2);
assertThat(job.firstAttemptAt()).isEqualTo(firstAttemptAt);
assertThat(job.startedAt()).isAfterOrEqualTo(firstAttemptAt);
}
@Test
void recordProviderRequestIsWriteOnce() {
UUID userId = insertUser("writeonce");
UUID modelId = insertModel("writeonce-model", MediaKind.IMAGE);
UUID jobId = insertQueued(userId, modelId, MediaKind.IMAGE);
repository.claimQueued("worker-a", LEASE, 10);
String firstId = "vendor-first-" + UUID.randomUUID();
String secondId = "vendor-second-" + UUID.randomUUID();
assertThat(repository.recordProviderRequest(jobId, firstId)).isEqualTo(1);
// A second attempt must not overwrite it — the stored id is the only
// handle on work the vendor may already have billed.
assertThat(repository.recordProviderRequest(jobId, secondId)).isEqualTo(0);
assertThat(repository.findByIdForUser(jobId, userId).orElseThrow().providerRequestId())
.isEqualTo(firstId);
}
@Test
void terminalTransitionsClearTheLease() {
UUID userId = insertUser("terminal");
UUID modelId = insertModel("terminal-model", MediaKind.IMAGE);
UUID failedId = insertQueued(userId, modelId, MediaKind.IMAGE);
repository.claimQueued("worker-a", LEASE, 10);
assertThat(repository.markFailed(failedId, "PROVIDER_TIMEOUT", "上游超时")).isEqualTo(1);
GenerationJob job = repository.findByIdForUser(failedId, userId).orElseThrow();
assertThat(job.status()).isEqualTo(JobStatus.FAILED);
assertThat(job.leaseExpiresAt()).isNull();
// A terminal job must never be claimable again.
assertThat(repository.claimQueued("worker-a", LEASE, 10))
.extracting(GenerationJob::id)
.doesNotContain(failedId);
}
@Test
void cancelIsGuardedByStatusSoAConcurrentFinishWins() {
UUID userId = insertUser("cancelrace");
UUID modelId = insertModel("cancelrace-model", MediaKind.IMAGE);
UUID jobId = insertQueued(userId, modelId, MediaKind.IMAGE);
repository.claimQueued("worker-a", LEASE, 10);
repository.markFailed(jobId, "PROVIDER_REJECTED", "拒绝");
// 0 rows, not an overwrite: the guard lives in the WHERE clause so a job
// that reached a terminal state first is not silently rewritten.
assertThat(repository.markCancelled(jobId, userId)).isEqualTo(0);
assertThat(repository.findByIdForUser(jobId, userId).orElseThrow().status())
.isEqualTo(JobStatus.FAILED);
}
@Test
void jobsAreScopedToTheirOwner() {
UUID owner = insertUser("owner");
UUID stranger = insertUser("stranger");
UUID modelId = insertModel("scope-model", MediaKind.IMAGE);
UUID jobId = insertQueued(owner, modelId, MediaKind.IMAGE);
assertThat(repository.findByIdForUser(jobId, stranger)).isEmpty();
assertThat(repository.findByIdForUser(jobId, owner)).isPresent();
}
// --- helpers ---
private UUID insertUser(String name) {
UUID id = UuidV7.generate();
jdbcClient.sql("INSERT INTO identity.users (id, username) VALUES (:id, :username)")
.param("id", id)
.param("username", name + "-" + id.toString().substring(0, 8))
.update();
return id;
}
private UUID insertModel(String code, MediaKind kind) {
UUID id = UuidV7.generate();
jdbcClient.sql("""
INSERT INTO creation.generation_models
(id, code, display_name, provider_code, provider_model_name, media_kind)
VALUES (:id, :code, :code, 'fixture', 'fixture-model', :kind)
""")
.param("id", id)
.param("code", code + "-" + id.toString().substring(0, 8))
.param("kind", kind.dbValue())
.update();
return id;
}
private UUID insertQueued(UUID userId, UUID modelId, MediaKind kind) {
UUID id = UuidV7.generate();
return repository.insertQueued(id, userId, null, kind, modelId, null, null,
"a prompt", null, null, null, null,
"fixture", "fixture-model", "v1", null,
"key-" + id, sha256Placeholder());
}
/** request_hash is constrained to exactly 32 bytes by ck_generation_jobs_idempotency. */
private static byte[] sha256Placeholder() {
byte[] hash = new byte[32];
new java.security.SecureRandom().nextBytes(hash);
return hash;
}
}
+1
View File
@@ -18,6 +18,7 @@
<module>patbond-auth</module> <module>patbond-auth</module>
<module>patbond-pet</module> <module>patbond-pet</module>
<module>patbond-community</module> <module>patbond-community</module>
<module>patbond-creation</module>
</modules> </modules>
<properties> <properties>