From 50285d24108fc4b7a2dbcdcb5edecdabc7197bae Mon Sep 17 00:00:00 2001 From: Shiroha7z Date: Thu, 23 Jul 2026 01:20:14 +0800 Subject: [PATCH] fix(pr31): interactive db.tx readback + hot enable runtime sync Complete PR #31 leftovers: SDK db.tx uses begin/step/commit so query/get/call return real mid-tx results; enable/disable hot-updates enabledSet, tokens, manifests, and builtin service handlers without restarting db-server. Co-authored-by: Cursor --- db-server/src/index.ts | 52 +++++- db-server/src/module-auth.ts | 28 ++++ db-server/src/routes/db-routes.ts | 70 ++++++-- db-server/src/runtime.test.ts | 98 +++++++++++ db-server/src/services/builtin-handlers.ts | 44 ++++- db-server/src/tx-runner.ts | 157 +++++++++++++++--- docs/api/db.md | 17 +- docs/api/sdk/db.md | 8 +- .../sdk/@sfmc-sdk/src/node/config/index.ts | 4 + modules/sdk/@sfmc-sdk/src/sapi/db/client.ts | 106 +++++++----- modules/sdk/@sfmc-sdk/src/sapi/db/index.ts | 3 +- modules/sdk/@sfmc-sdk/src/sapi/db/types.ts | 1 + .../sdk/@sfmc-sdk/src/sapi/service/client.ts | 2 +- 13 files changed, 499 insertions(+), 91 deletions(-) diff --git a/db-server/src/index.ts b/db-server/src/index.ts index d050cfb1..3085a505 100644 --- a/db-server/src/index.ts +++ b/db-server/src/index.ts @@ -23,7 +23,13 @@ import http from "node:http"; import { createIdempotencyStore } from "./lib/idempotency-store.js"; import { loadEnv } from "./env.js"; -import { buildModuleAuth, verifyModuleAuth } from "./module-auth.js"; +import { + buildModuleAuth, + ensureModuleToken, + persistModuleAuth, + revokeModuleToken, + verifyModuleAuth, +} from "./module-auth.js"; import { loadManifestV2 } from "./manifest-loader.js"; import { log } from "./lib/log.js"; import { assertNodeVersion } from "./lib/runtime.js"; @@ -34,7 +40,11 @@ import { initSchema } from "./domain/schema.js"; import { SchemaRegistry } from "./schema-registry.js"; import { ServiceRegistry } from "./service-registry.js"; import { TxRunner } from "./tx-runner.js"; -import { registerEnabledBuiltinServices } from "./services/builtin-handlers.js"; +import { + registerBuiltinPluginForModule, + registerEnabledBuiltinServices, + unregisterBuiltinPluginForModule, +} from "./services/builtin-handlers.js"; import { readJson } from "@sfmc-bds/sdk/node/config"; @@ -199,6 +209,44 @@ function setModuleEnabled(mod: { id: string; canDisable: boolean }, enabled: boo updateModuleState(lockFile, mod.id, { enabled: !!enabled }); // DRY:与 loadModuleLock 对称走 saveModuleLock,勿散落 writeJson saveModuleLock(env.MODULE_LOCK_PATH, lockFile); + + // 热更新运行时图(PR #31 未完成项):enabledSet / tokens / manifests / builtin handlers + // 与启动期同源,避免「lock 已开但鉴权仍 401 / handler 未注册」。 + syncRuntimeEnabled(mod.id, !!enabled); +} + +/** + * 启停后同步 db-server 进程内运行时状态(不重启)。 + * TxRunner / serviceRoutes 持有 enabledManifests 引用,Map 就地改即可。 + */ +function syncRuntimeEnabled(moduleId: string, enabled: boolean): void { + if (enabled) { + const manifest = loadedManifest.modules[moduleId]; + if (!manifest) { + log.warn(`[modules] 热启用 ${moduleId}: manifest 缺失(未安装?),仅写 lock`); + return; + } + enabledSet.add(moduleId); + enabledManifests.set(moduleId, manifest); + if (ensureModuleToken(moduleAuth, moduleId)) { + log.info(`[modules] 热启用 ${moduleId}: 派生 module token`); + } + persistModuleAuth(env.PROJECT_ROOT, moduleAuth, env.AUTH_TOKEN); + if (registerBuiltinPluginForModule(serviceRegistry, { query, db }, moduleId)) { + log.success(`[modules] 热启用 ${moduleId}: 已注册内置 service handlers`); + } + return; + } + + enabledSet.delete(moduleId); + enabledManifests.delete(moduleId); + if (revokeModuleToken(moduleAuth, moduleId)) { + persistModuleAuth(env.PROJECT_ROOT, moduleAuth, env.AUTH_TOKEN); + } + const n = unregisterBuiltinPluginForModule(serviceRegistry, moduleId); + if (n > 0) { + log.info(`[modules] 热禁用 ${moduleId}: 卸下 ${n} 个内置 handlers`); + } } // ── 平台路由(非模块业务) ─────────────────────────────────── diff --git a/db-server/src/module-auth.ts b/db-server/src/module-auth.ts index 7b786ae4..689644dc 100644 --- a/db-server/src/module-auth.ts +++ b/db-server/src/module-auth.ts @@ -102,6 +102,34 @@ export function loadModuleAuth(projectRoot: string): ModuleAuthMap | null { return { tokens: parsed.tokens, secret: parsed.secret }; } +/** 把当前 auth map 写回 data/module-tokens.json(热更新 token 时复用) */ +export function persistModuleAuth(projectRoot: string, auth: ModuleAuthMap, envAuthToken: string): void { + const outFile = join(projectRoot, "data", "module-tokens.json"); + writeJson(outFile, { + tokens: auth.tokens, + secret: auth.secret, + generatedAt: new Date().toISOString(), + secretGenerated: !envAuthToken, + } satisfies TokenStore); +} + +/** + * 确保 moduleId 在 auth.tokens 中有派生值(复用现有 secret,勿重建随机 secret)。 + * @returns 是否新写入了 token + */ +export function ensureModuleToken(auth: ModuleAuthMap, moduleId: string): boolean { + if (auth.tokens[moduleId]) return false; + auth.tokens[moduleId] = deriveToken(moduleId, auth.secret); + return true; +} + +/** 禁用时撤销内存中的 token(磁盘由 persistModuleAuth 同步) */ +export function revokeModuleToken(auth: ModuleAuthMap, moduleId: string): boolean { + if (!(moduleId in auth.tokens)) return false; + delete auth.tokens[moduleId]; + return true; +} + /** * 校验请求方身份: * - Bearer 头 / X-Module-Token 二选一 diff --git a/db-server/src/routes/db-routes.ts b/db-server/src/routes/db-routes.ts index 65a453ec..dd0b0d62 100644 --- a/db-server/src/routes/db-routes.ts +++ b/db-server/src/routes/db-routes.ts @@ -3,17 +3,13 @@ * * 端点: * POST /api/sfmc/db/define-table body {name, columns, softDelete?} → {table, created} - * POST /api/sfmc/db/tx body {steps[]} → TxResponse | TxError + * POST /api/sfmc/db/tx body {steps[]} → TxResponse | TxError (批量) + * POST /api/sfmc/db/tx/begin body {} → {txId} + * POST /api/sfmc/db/tx/step body {txId, step} → {result} + * POST /api/sfmc/db/tx/commit body {txId} → TxResponse + * POST /api/sfmc/db/tx/rollback body {txId} → {ok} * POST /api/sfmc/db/query body {table, opts?} → {rows} - * POST /api/sfmc/db/get body {table, id} → {row} - * POST /api/sfmc/db/insert body {table, row} → {row} - * POST /api/sfmc/db/update body {table, id, patch} → {row} - * POST /api/sfmc/db/delete body {table, id, hard?} → {changes} - * POST /api/sfmc/db/audit body {table, rowId, action, data?} → {ok} - * POST /api/sfmc/db/idempotent/probe body {action, key} → {replayed, cached?} - * POST /api/sfmc/db/idempotent/commit body {action, key, value?} → {ok} - * - * 鉴权:handle() 校验后把 {id, permissions} 写到 ctx.moduleAuth(不挂 req)。 + * ... */ import type { IncomingMessage, ServerResponse } from "node:http"; @@ -95,6 +91,60 @@ export function createDbRoutes(depsIn: Partial) { return true; } + // 交互式事务会话 — 供 SDK 在回调内读回 query/get/call 结果 + if (path === "/api/sfmc/db/tx/begin") { + try { + const result = deps.txRunner!.beginSession(moduleId); + json(res, result as unknown as Record, result.ok ? 200 : 400); + } catch (e) { + jsonV2Fail(res, (e as Error).message, 500); + } + return true; + } + if (path === "/api/sfmc/db/tx/step") { + try { + const txId = String(body.txId || ""); + const step = body.step as TxStep | undefined; + if (!txId || !step || typeof step !== "object" || !("op" in step)) { + jsonV2Fail(res, "tx/step 需要 { txId, step }", 400); + return true; + } + const result = await deps.txRunner!.stepSession(txId, moduleId, step); + json(res, result as unknown as Record, result.ok ? 200 : 400); + } catch (e) { + jsonV2Fail(res, (e as Error).message, 500); + } + return true; + } + if (path === "/api/sfmc/db/tx/commit") { + try { + const txId = String(body.txId || ""); + if (!txId) { + jsonV2Fail(res, "tx/commit 需要 { txId }", 400); + return true; + } + const result = deps.txRunner!.commitSession(txId, moduleId); + json(res, result as unknown as Record, result.ok ? 200 : 400); + } catch (e) { + jsonV2Fail(res, (e as Error).message, 500); + } + return true; + } + if (path === "/api/sfmc/db/tx/rollback") { + try { + const txId = String(body.txId || ""); + if (!txId) { + jsonV2Fail(res, "tx/rollback 需要 { txId }", 400); + return true; + } + const result = deps.txRunner!.rollbackSession(txId, moduleId); + json(res, result as unknown as Record, result.ok ? 200 : 400); + } catch (e) { + jsonV2Fail(res, (e as Error).message, 500); + } + return true; + } + type SingleOp = "query" | "get" | "insert" | "update" | "delete"; const singles: { match: RegExp; op: SingleOp }[] = [ { match: /^\/api\/sfmc\/db\/query$/, op: "query" }, diff --git a/db-server/src/runtime.test.ts b/db-server/src/runtime.test.ts index 54315c78..e51a85e2 100644 --- a/db-server/src/runtime.test.ts +++ b/db-server/src/runtime.test.ts @@ -130,3 +130,101 @@ test("normalizeOrderBy: SDK field 与遗留 col / 数组互通(LSP)", async () = ]); throws(() => normalizeOrderBy({ dir: "asc" }), /field\/col/); }); + +test("module-auth: ensureModuleToken / revoke 复用 secret(DRY)", async () => { + const { deriveToken, ensureModuleToken, revokeModuleToken } = await import("./module-auth.js"); + const secret = "test-secret"; + const auth = { secret, tokens: {} as Record }; + equal(ensureModuleToken(auth, "feature-afk"), true); + equal(auth.tokens["feature-afk"], deriveToken("feature-afk", secret)); + equal(ensureModuleToken(auth, "feature-afk"), false); + equal(revokeModuleToken(auth, "feature-afk"), true); + equal(auth.tokens["feature-afk"], undefined); + equal(revokeModuleToken(auth, "feature-afk"), false); +}); + +test("builtin-handlers: 热禁用按 serviceNames 卸载(OCP)", async () => { + const { BUILTIN_SERVICE_PLUGINS, unregisterBuiltinPluginForModule } = await import( + "./services/builtin-handlers.js" + ); + const economy = BUILTIN_SERVICE_PLUGINS.find((p) => p.moduleId === "feature-economy"); + equal(!!economy?.serviceNames?.includes("economy.account.get"), true); + + const reg = new ServiceRegistry(); + reg.registerHandler("feature-economy", "economy.account.get", async () => ({})); + reg.registerHandler("feature-economy", "economy.account.debit", async () => ({})); + equal(unregisterBuiltinPluginForModule(reg, "feature-economy"), 2); + equal(reg.list().length, 0); + equal(unregisterBuiltinPluginForModule(reg, "feature-unknown"), 0); +}); + +test("TxRunner 交互会话: step 中途读回 insert/get(PR #31 leftover)", async () => { + const { DatabaseSync } = await import("node:sqlite"); + const { createQuery } = await import("./lib/sqlite.js"); + const { SchemaRegistry } = await import("./schema-registry.js"); + const { TxRunner } = await import("./tx-runner.js"); + + const db = new DatabaseSync(":memory:"); + const query = createQuery(db); + const schema = new SchemaRegistry(db); + schema.define("feature-demo", { + name: "demo_items", + columns: { + id: { type: "text", primary: true }, + name: { type: "text", notNull: true }, + }, + softDelete: false, + }); + + const enabled = new Map([ + [ + "feature-demo", + { + id: "feature-demo", + version: "1.0.0", + permissions: ["db:read:demo_items", "db:write:demo_items"], + services: { provides: [], requires: [] }, + db: { tables: [] }, + config: { key: "demo" }, + } as unknown as import("./manifest-loader.js").ModuleManifestV2, + ], + ]); + + const runner = new TxRunner({ + db, + query, + schema, + serviceRegistry: new ServiceRegistry(), + enabled, + }); + + const begin = runner.beginSession("feature-demo"); + equal(begin.ok, true); + if (!begin.ok) return; + const { txId } = begin; + + const ins = await runner.stepSession(txId, "feature-demo", { + op: "insert", + table: "demo_items", + row: { id: "a1", name: "alpha" }, + }); + equal(ins.ok, true); + if (!ins.ok) return; + equal((ins.result as { op: string; row: { id: string } }).row.id, "a1"); + + const got = await runner.stepSession(txId, "feature-demo", { + op: "get", + table: "demo_items", + id: "a1", + }); + equal(got.ok, true); + if (!got.ok) return; + equal((got.result as { row: { name: string } | null }).row?.name, "alpha"); + + const committed = runner.commitSession(txId, "feature-demo"); + equal(committed.ok, true); + + const rows = db.prepare("SELECT name FROM demo_items WHERE id = ?").all("a1") as Array<{ name: string }>; + equal(rows[0]?.name, "alpha"); + db.close(); +}); diff --git a/db-server/src/services/builtin-handlers.ts b/db-server/src/services/builtin-handlers.ts index 14450ea8..260de84d 100644 --- a/db-server/src/services/builtin-handlers.ts +++ b/db-server/src/services/builtin-handlers.ts @@ -15,11 +15,25 @@ export type BuiltinServiceDeps = { query: QueryFn; db: DatabaseSync }; export type BuiltinServicePlugin = { moduleId: string; register: (registry: ServiceRegistry, deps: BuiltinServiceDeps) => void; + /** 与 register 写入的 handler 名对齐,供热禁用 unregister */ + serviceNames?: string[]; }; /** 内置 service 插件清单 — 唯一扩展点 */ export const BUILTIN_SERVICE_PLUGINS: BuiltinServicePlugin[] = [ - { moduleId: "feature-economy", register: registerEconomyHandlers }, + { + moduleId: "feature-economy", + register: registerEconomyHandlers, + serviceNames: [ + "economy.account.get", + "economy.account.debit", + "economy.account.credit", + "economy.account.transfer", + "economy.dailyTasks.list", + "economy.dailyTasks.submit", + "economy.stats.monthly", + ], + }, ]; /** 按 enabledSet 注册内置插件,返回已注册插件数 */ @@ -36,3 +50,31 @@ export function registerEnabledBuiltinServices( } return n; } + +/** 热启用:只注册单个内置插件(若尚未注册) */ +export function registerBuiltinPluginForModule( + registry: ServiceRegistry, + deps: BuiltinServiceDeps, + moduleId: string +): boolean { + const plugin = BUILTIN_SERVICE_PLUGINS.find((p) => p.moduleId === moduleId); + if (!plugin) return false; + const already = (plugin.serviceNames ?? []).some((n) => registry.list().some((h) => h.name === n)); + if (already) return false; + plugin.register(registry, deps); + return true; +} + +/** 热禁用:卸掉该模块的内置 handler */ +export function unregisterBuiltinPluginForModule(registry: ServiceRegistry, moduleId: string): number { + const plugin = BUILTIN_SERVICE_PLUGINS.find((p) => p.moduleId === moduleId); + if (!plugin?.serviceNames) return 0; + let n = 0; + for (const name of plugin.serviceNames) { + if (registry.list().some((h) => h.name === name)) { + registry.unregisterHandler(name); + n += 1; + } + } + return n; +} diff --git a/db-server/src/tx-runner.ts b/db-server/src/tx-runner.ts index 20d4312e..3981d099 100644 --- a/db-server/src/tx-runner.ts +++ b/db-server/src/tx-runner.ts @@ -1,16 +1,15 @@ /** * tx-runner.ts — 事务 RPC 处理器 * - * 协议: - * POST /api/sfmc/db/tx - * body = { moduleId, steps: TxStep[] } - * reply: { ok: true } | { ok: false, step: , error, code } + * 协议(两种): + * A. 批量: POST /api/sfmc/db/tx body={ steps } → { ok, results? } | { ok:false, step, error, code } + * B. 交互: begin → step* → commit|rollback(SDK db.tx 用;回调内可读回 query/get/call) * - * 流程: - * 1. 校验 moduleId in enabled + permission "db:write:*" + * 流程(批量): + * 1. 校验 moduleId in enabled * 2. BEGIN IMMEDIATE * 3. 顺序跑 steps — 任一抛错 → ROLLBACK + reply { ok:false, step:, ... } - * 4. 全过 → COMMIT + reply { ok:true } + * 4. 全过 → COMMIT + reply { ok:true, results } * * step.op: * - query { table, opts? } @@ -22,8 +21,7 @@ * - service { name, input } → 派发;失败抛错回退整事务 * * 事务内 service call:不持有 db handle 给 handler(避免并发写) — - * 实现:handler 是"同步纯函数",返回 result 写回 step 的输出。 - * 若 handler 自己也想写 db,应通过其它接口在事务外做(避免嵌套)。 + * 实现:handler 返回 result 写回 step 输出;若 handler 还要写 db,应走其它接口在事务外做。 */ import { randomUUID } from "node:crypto"; @@ -143,6 +141,12 @@ export interface TxRequest { steps: TxStep[]; } +interface TxSession { + moduleId: string; + results: TxStepResult[]; + openedAt: number; +} + export interface TxRunnerDeps { db: DatabaseSync; query: import("./lib/sqlite.js").QueryFn; @@ -152,6 +156,8 @@ export interface TxRunnerDeps { } export class TxRunner { + private readonly sessions = new Map(); + constructor(private readonly deps: TxRunnerDeps) {} async run(req: TxRequest): Promise { @@ -179,22 +185,7 @@ export class TxRunner { results.push(r); } catch (err) { this.rollback(); - const code: TxError["code"] = - err instanceof PermissionDeniedError - ? "permission_denied" - : err instanceof DispatchError && - (err.code === "no_such_service" || - err.code === "not_in_requires" || - err.code === "forbidden" || - err.code === "domain_error") - ? err.code - : (err as { code?: string }).code === "no_such_service" - ? "no_such_service" - : (err as { code?: string }).code === "not_in_requires" - ? "not_in_requires" - : (err as { code?: string }).code === "no_such_table" - ? "no_such_table" - : "internal"; + const code = this.mapErrorCode(err); log.warn(`[tx ${traceId}] step=${i} failed: ${(err as Error).message}`); return { ok: false, step: i, error: (err as Error).message, code }; } @@ -211,6 +202,122 @@ export class TxRunner { return { ok: true, results }; } + /** + * 交互式事务:begin → step* → commit|rollback。 + * 供 SDK db.tx 在回调内 await query/get/call 真实结果(PR #31 未完成项)。 + * 单连接模型:新会话会先清掉残留会话(避免 BEGIN 锁死)。 + */ + beginSession(moduleId: string): { ok: true; txId: string } | TxError { + const manifest = this.deps.enabled.get(moduleId); + if (!manifest) return { ok: false, step: -1, error: "模块未 enabled", code: "forbidden" }; + // 清残留会话(崩溃/未 commit),再开新事务 + if (this.sessions.size > 0) { + for (const id of [...this.sessions.keys()]) { + this.abortSession(id); + } + } + try { + this.deps.db.exec("BEGIN IMMEDIATE"); + } catch (e) { + return { ok: false, step: -1, error: `BEGIN 失败: ${(e as Error).message}`, code: "internal" }; + } + const txId = randomUUID(); + this.sessions.set(txId, { + moduleId, + results: [], + openedAt: Date.now(), + }); + log.info(`[tx-session ${txId.slice(0, 8)}] begin module=${moduleId}`); + return { ok: true, txId }; + } + + async stepSession( + txId: string, + moduleId: string, + step: TxStep + ): Promise<{ ok: true; result: TxStepResult } | TxError> { + const session = this.sessions.get(txId); + if (!session) { + return { ok: false, step: -1, error: "事务会话不存在或已结束", code: "internal" }; + } + if (session.moduleId !== moduleId) { + return { ok: false, step: session.results.length, error: "moduleId 与会话不符", code: "forbidden" }; + } + const manifest = this.deps.enabled.get(moduleId); + if (!manifest) { + this.abortSession(txId); + return { ok: false, step: session.results.length, error: "模块未 enabled", code: "forbidden" }; + } + const stepIndex = session.results.length; + try { + const r = await this.runOne(moduleId, manifest, step); + session.results.push(r); + return { ok: true, result: r }; + } catch (err) { + this.abortSession(txId); + const code = this.mapErrorCode(err); + log.warn(`[tx-session ${txId.slice(0, 8)}] step=${stepIndex} failed: ${(err as Error).message}`); + return { ok: false, step: stepIndex, error: (err as Error).message, code }; + } + } + + commitSession(txId: string, moduleId: string): TxResponse | TxError { + const session = this.sessions.get(txId); + if (!session) { + return { ok: false, step: -1, error: "事务会话不存在或已结束", code: "internal" }; + } + if (session.moduleId !== moduleId) { + return { ok: false, step: -1, error: "moduleId 与会话不符", code: "forbidden" }; + } + try { + this.deps.db.exec("COMMIT"); + } catch (e) { + this.abortSession(txId); + return { ok: false, step: -1, error: `COMMIT 失败: ${(e as Error).message}`, code: "internal" }; + } + const results = session.results; + this.sessions.delete(txId); + log.info(`[tx-session ${txId.slice(0, 8)}] commit ${results.length} steps OK`); + return { ok: true, results }; + } + + rollbackSession(txId: string, moduleId: string): { ok: true } | TxError { + const session = this.sessions.get(txId); + if (!session) { + // 幂等:已结束视为成功回滚 + return { ok: true }; + } + if (session.moduleId !== moduleId) { + return { ok: false, step: -1, error: "moduleId 与会话不符", code: "forbidden" }; + } + this.abortSession(txId); + return { ok: true }; + } + + private abortSession(txId: string): void { + this.sessions.delete(txId); + this.rollback(); + log.info(`[tx-session ${txId.slice(0, 8)}] aborted`); + } + + private mapErrorCode(err: unknown): TxError["code"] { + return err instanceof PermissionDeniedError + ? "permission_denied" + : err instanceof DispatchError && + (err.code === "no_such_service" || + err.code === "not_in_requires" || + err.code === "forbidden" || + err.code === "domain_error") + ? err.code + : (err as { code?: string }).code === "no_such_service" + ? "no_such_service" + : (err as { code?: string }).code === "not_in_requires" + ? "not_in_requires" + : (err as { code?: string }).code === "no_such_table" + ? "no_such_table" + : "internal"; + } + private rollback(): void { try { this.deps.db.exec("ROLLBACK"); diff --git a/docs/api/db.md b/docs/api/db.md index 1e472776..6d01505e 100644 --- a/docs/api/db.md +++ b/docs/api/db.md @@ -34,9 +34,22 @@ `opts` 使用表达式树 `WhereExpr`，**不要**传 SQL 字符串。 -## POST /api/sfmc/db/tx +## 事务 -事务，body `{ steps: TxStep[] }`。步骤类型包括 query、get、insert、update、delete、audit、**call**（调 service）。 +### 交互会话（SDK `db.tx` 默认） + +| 路径 | Body | 响应要点 | +|------|------|----------| +| `/tx/begin` | `{}` | `{ ok, txId }` | +| `/tx/step` | `{ txId, step }` | `{ ok, result }`（`query`/`get`/`call` 可当场读回） | +| `/tx/commit` | `{ txId }` | `{ ok, results }` | +| `/tx/rollback` | `{ txId }` | `{ ok }` | + +### 批量（工具/测试） + +`POST /api/sfmc/db/tx`，body `{ steps: TxStep[] }` → `{ ok, results }`。 + +步骤类型：query、get、insert、update、delete、audit、**service**（SDK 侧 `tx.call`）。 在 tx 内用 `tx.*`，不要混用外面的单步 CRUD。 diff --git a/docs/api/sdk/db.md b/docs/api/sdk/db.md index 5a8aeae1..3c7da4e0 100644 --- a/docs/api/sdk/db.md +++ b/docs/api/sdk/db.md @@ -30,12 +30,16 @@ await db.audit("lands", id, "transfer", { to: "..." }); ## 事务 +`db.tx` 走交互会话（begin → step* → commit）：回调里 `await tx.query/get/call` 返回**真实**服务端结果，可据此分支。 + ```ts await db.tx(async (tx: TxContext) => { + const row = await tx.get("lands", landId); + if (!row) throw new Error("missing"); await tx.update("lands", landId, { owner_player_id: newOwner }); await tx.audit("lands", landId, "transfer", { from, to }); - await tx.call("economy.debit", { playerId, amount: 100 }); - return { ok: true }; + const debit = await tx.call<{ ok: boolean }>("economy.debit", { playerId, amount: 100 }); + return { ok: true, debit }; }); ``` diff --git a/modules/sdk/@sfmc-sdk/src/node/config/index.ts b/modules/sdk/@sfmc-sdk/src/node/config/index.ts index e9cd2a84..e2efe4b0 100644 --- a/modules/sdk/@sfmc-sdk/src/node/config/index.ts +++ b/modules/sdk/@sfmc-sdk/src/node/config/index.ts @@ -119,6 +119,10 @@ export interface Catalog { export interface TokenStore { tokens?: Record; secret?: string; + /** ISO 时间戳：最近一次写入 module-tokens.json */ + generatedAt?: string; + /** secret 是否为本次启动随机生成（未配置 AUTH_TOKEN） */ + secretGenerated?: boolean; } /** diff --git a/modules/sdk/@sfmc-sdk/src/sapi/db/client.ts b/modules/sdk/@sfmc-sdk/src/sapi/db/client.ts index f654d412..fc24d2e1 100644 --- a/modules/sdk/@sfmc-sdk/src/sapi/db/client.ts +++ b/modules/sdk/@sfmc-sdk/src/sapi/db/client.ts @@ -17,8 +17,8 @@ import type { DeleteResult, InsertResult, QueryOptions, - TxResponse, TxStep, + TxStepResult, UpdateResult, } from "./types.js"; @@ -39,7 +39,7 @@ export function clearDbModuleContext(): void { _currentTxId = null; } -/** 供 service 客户端判断是否处于 db.tx 录制期(LSP:与 service.get 互斥)。 */ +/** 供 service 客户端判断是否处于 db.tx 交互会话(LSP:与 service.get 互斥)。 */ export function isDbTxRecording(): boolean { return _currentTxId != null; } @@ -96,14 +96,19 @@ async function post(path: string, body: unknown): Promise { return res.data as T; } -/** 事务录制期不支持 query/get 读回:返回 []/null stub 会破坏 LSP(误导业务分支)。 */ -function txReadNotSupported(op: string): never { - throw new DbError( - `db.tx 内暂不支持 ${op} 读回结果(录制后一次性提交,无交互协议);` + - `请在事务外先 db.${op},再在 tx 内做写操作`, - "tx_interactive_required", - 0 - ); +/** 规范化 step(去掉 undefined 可选字段,避免 JSON 脏键) */ +function normalizeStep(s: TxStep): TxStep { + if (s.op === "query" && s.opts === undefined) delete (s as { opts?: unknown }).opts; + if (s.op === "audit" && s.data === undefined) delete (s as { data?: unknown }).data; + return s; +} + +type StepOk = { ok: true; result: TxStepResult }; +type SessionBegin = { ok: true; txId: string }; + +async function txStep(txId: string, step: TxStep): Promise { + const res = await post("/api/sfmc/db/tx/step", { txId, step: normalizeStep(step) }); + return res.result; } /* ── 公开 API ──────────────────────────────────────────────────── */ @@ -170,62 +175,72 @@ export const db = { /** * 事务:边界在 db-server 进程。失败自动回滚,成功提交。 - * 录制期:写操作(insert/update/delete/audit/call)可入队;query/get 显式抛错, - * 避免返回 []/null 假数据破坏 LSP。call 为 fire-and-forget(Promise), - * 完整两阶段读回协议另议。 + * 交互会话协议:begin → step* → commit|rollback。 + * 回调内 await query/get/call 返回真实服务端结果(PR #31 未完成项补齐)。 + * 批量 POST /db/tx 仍保留给工具/测试;模块侧统一走交互路径。 */ async tx(fn: (tx: TxContext) => Promise): Promise { requireModuleContext("tx"); if (_currentTxId) throw new DbError("嵌套事务暂不支持", "nested_tx", 0); - const steps: TxStep[] = []; - _currentTxId = "pending"; - const push = (s: TxStep) => { - if (s.op === "query" && s.opts === undefined) delete (s as { opts?: unknown }).opts; - if (s.op === "audit" && s.data === undefined) delete (s as { data?: unknown }).data; - steps.push(s); - }; + const begin = await post("/api/sfmc/db/tx/begin", {}); + const txId = begin.txId; + _currentTxId = txId; - const recorder: TxContext = { - query: async () => txReadNotSupported("query"), - get: async () => txReadNotSupported("get"), + const interactive: TxContext = { + query: async = Record>( + table: string, + opts?: QueryOptions + ): Promise => { + const r = await txStep(txId, { op: "query", table, ...(opts ? { opts } : {}) }); + return (r.rows ?? []) as U[]; + }, + get: async = Record>( + table: string, + id: string | number + ): Promise => { + const r = await txStep(txId, { op: "get", table, id: String(id) }); + return (r.row ?? null) as U | null; + }, insert: async >(table: string, row: U): Promise => { - push({ op: "insert", table, row }); - return row; + const r = await txStep(txId, { op: "insert", table, row }); + return (r.row ?? row) as U; }, update: async >( table: string, id: string | number, patch: Partial - ): Promise => { - // LSP:录制期无完整行可读回;勿把 patch 假扮成 U - push({ op: "update", table, id: String(id), patch }); + ): Promise => { + const r = await txStep(txId, { op: "update", table, id: String(id), patch }); + return (r.row ?? ({ ...patch, id } as unknown as U)) as U; }, delete: async (table: string, id: string | number, opts?: { hard?: boolean }) => { - push({ op: "delete", table, id: String(id), hard: opts?.hard ?? false }); + await txStep(txId, { op: "delete", table, id: String(id), hard: opts?.hard ?? false }); }, audit: async (table: string, rowId: string | number, action: string, data?: Record) => { - if (data) push({ op: "audit", table, rowId: String(rowId), action, data }); - else push({ op: "audit", table, rowId: String(rowId), action }); + if (data) await txStep(txId, { op: "audit", table, rowId: String(rowId), action, data }); + else await txStep(txId, { op: "audit", table, rowId: String(rowId), action }); }, - call: async (name: string, input: Record): Promise => { - push({ op: "service", name, input }); - // 录制期无服务端 result 可回放;契约为 void(与 service.get 区分 — LSP) + call: async (name: string, input: Record): Promise => { + const r = await txStep(txId, { op: "service", name, input }); + return r.result as U; }, }; - let userResult: T; try { - userResult = await fn(recorder); + const userResult = await fn(interactive); + await post("/api/sfmc/db/tx/commit", { txId }); + return userResult; } catch (e) { - _currentTxId = null; + try { + await post("/api/sfmc/db/tx/rollback", { txId }); + } catch { + /* best-effort */ + } throw e; + } finally { + _currentTxId = null; } - - // post 在 !ok 时已抛带 step/code 的 DbError;此处只需成功路径 - await post("/api/sfmc/db/tx", { steps }); - _currentTxId = null; - return userResult; }, /** 平台预置:审计日志(自动写 _audit 表) */ @@ -259,14 +274,13 @@ export interface TxContext { id: string | number ): Promise; insert>(table: string, row: T): Promise; - /** 录制期无行读回,返回 void(与事务外 update→T 区分)。 */ update>( table: string, id: string | number, patch: Partial - ): Promise; + ): Promise; delete(table: string, id: string | number, opts?: { hard?: boolean }): Promise; audit(table: string, rowId: string | number, action: string, data?: Record): Promise; - /** fire-and-forget:入队 service step,不返回 result。 */ - call(name: string, input: Record): Promise; + /** 交互会话内返回服务端真实 result */ + call(name: string, input: Record): Promise; } diff --git a/modules/sdk/@sfmc-sdk/src/sapi/db/index.ts b/modules/sdk/@sfmc-sdk/src/sapi/db/index.ts index d6178259..395539f8 100644 --- a/modules/sdk/@sfmc-sdk/src/sapi/db/index.ts +++ b/modules/sdk/@sfmc-sdk/src/sapi/db/index.ts @@ -6,8 +6,7 @@ * * 设计: * - db.query / get / insert / update / delete / audit / idempotent:单 RPC - * - db.tx(fn):把 fn 内写 step 录下来,一次性发 /api/sfmc/db/tx,server 端事务跑 - * (query/get 在录制期不可用 — 避免 stub 假数据;交互读回需两阶段协议) + * - db.tx(fn):交互会话 begin→step*→commit,回调内 await query/get/call 可读回真实结果 * - 不允许原始 SQL;只能传 WhereExpr 表达式树,平台翻译 * - 模块不能 require("fs");只能走这里 */ diff --git a/modules/sdk/@sfmc-sdk/src/sapi/db/types.ts b/modules/sdk/@sfmc-sdk/src/sapi/db/types.ts index ef041844..f10aa43a 100644 --- a/modules/sdk/@sfmc-sdk/src/sapi/db/types.ts +++ b/modules/sdk/@sfmc-sdk/src/sapi/db/types.ts @@ -76,6 +76,7 @@ export interface TxStepResult { rows?: Record[]; row?: Record; id?: Primitive; + changes?: number; result?: unknown; } diff --git a/modules/sdk/@sfmc-sdk/src/sapi/service/client.ts b/modules/sdk/@sfmc-sdk/src/sapi/service/client.ts index f852d2d4..70fa0d80 100644 --- a/modules/sdk/@sfmc-sdk/src/sapi/service/client.ts +++ b/modules/sdk/@sfmc-sdk/src/sapi/service/client.ts @@ -6,7 +6,7 @@ * * 设计: * - service.get(name, input):发 GET /api/sfmc/services/:name?input=... - * - 事务内调用走 tx.call(name, input),step 由 db.tx 一并提交 + * - 事务内调用走 tx.call(name, input),交互会话内返回真实 result * - service.list:列所有 enabled 模块 provides 的 service */