feat(kbdb): bulk 寫入端點(缺口1)+owner_id 寫入必填——D28 keystone
- POST /entries/bulk、/records/bulk:任意N筆約3 subrequest(批量冪等查+db.batch()+hydrate),解 Too many subrequests - 部分失敗不整批炸(逐筆驗證剔除+batch錯退化逐句隔離);page_name 冪等 - embedManyOnWrite 批次嵌(單次AI.run+upsert+UPDATE IN) - owner_id 缺→400(validateEntryInput,擋在HTTP路由層;內部無主路徑recipe-stat/value entries不受傷) - Phase 13 tasks;tsc 0、vitest 18/18
This commit is contained in:
@@ -21,6 +21,20 @@ export interface CreateEntryInput {
|
||||
id?: string;
|
||||
}
|
||||
|
||||
// owner_id 寫入必填(D27/D28:機制強制,設計成 AI 錯不了)。
|
||||
// 缺 owner_id → 資料無主,會被 mira 等 caller 的 owner 過濾濾掉(實測 gloss owner=None 查不到)。
|
||||
// 這是通用驗證器,POST /entries(單筆)與 POST /entries/bulk(逐筆)共用同一把尺。
|
||||
// 註(機制邊界):強制點在「HTTP 寫入路由」,不在 createEntry() 本身——因為 createEntry 被
|
||||
// record-crud 的 slot value entry(owner 由 record 帶)與 recipe-stat(全域統計、天生無主)內部呼叫。
|
||||
// 所有「外部」寫入(graph 插件 / mira / sync 腳本)都經 POST /entries|/entries/bulk,故路由層擋 = 擋死外部漏欄。
|
||||
export function validateEntryInput(input: unknown): string | null {
|
||||
if (!input || typeof input !== 'object') return 'entry must be an object';
|
||||
const e = input as Partial<CreateEntryInput>;
|
||||
if (!e.entry_type) return 'entry_type required';
|
||||
if (!e.owner_id) return 'owner_id required: 每筆寫入必須指定租戶,避免跨租戶資料無主';
|
||||
return null;
|
||||
}
|
||||
|
||||
export async function createEntry(db: D1Database, input: CreateEntryInput): Promise<Entry> {
|
||||
const id = input.id ?? uid('e');
|
||||
await db
|
||||
@@ -123,6 +137,147 @@ export async function deleteEntry(db: D1Database, id: string): Promise<void> {
|
||||
await db.prepare('DELETE FROM entries WHERE id = ?').bind(id).run();
|
||||
}
|
||||
|
||||
// ── Bulk write(缺口1 keystone:一次收整個 envelope,少 subrequest 寫法)────────────────
|
||||
//
|
||||
// 為什麼:base 舊寫入是「每筆一個 await db.prepare().run()」=每筆一個 subrequest,串行。
|
||||
// 大筆記(journals 11 三元組+10 node)撞 CF「Too many subrequests by single Worker invocation」→ 500。
|
||||
// 解:用 D1 `db.batch([...stmts])`(一次 call 跑多語句)→ N 筆 ≈ 常數個 subrequest。
|
||||
// 本函式對任意 N 筆 entries 固定約 3 個 subrequest:① 一次 page_name 冪等查詢 ② 一次 batch 寫入
|
||||
// ③ 一次 hydrate(IN(...) 撈回寫入列)。不隨 N 線性增長 → free/paid tier 都安全。
|
||||
// 冪等:沿用既有 page_name idempotency(同 sync-registry-to-kbdb.py 的 GET→PATCH/POST 模式)——
|
||||
// payload 帶 page_name 且庫內已有 → UPDATE(不重複建);否則 INSERT。單次查詢批量解析。
|
||||
// 部分失敗不整批炸:逐筆驗證(缺 owner_id/entry_type)先剔除記為 failed,其餘照寫;
|
||||
// batch 若整體拋錯(如 DB 約束)→ 退化成逐句執行隔離出真正壞的那筆(只在錯誤路徑付 N 個 subrequest)。
|
||||
|
||||
export interface BulkEntryItemResult {
|
||||
index: number;
|
||||
status: 'created' | 'updated' | 'failed';
|
||||
id?: string;
|
||||
page_name?: string | null;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
export interface BulkEntriesResult {
|
||||
created: number;
|
||||
updated: number;
|
||||
failed: number;
|
||||
results: BulkEntryItemResult[]; // 逐筆結果(與輸入同順序)
|
||||
entries: Entry[]; // 成功寫入(created+updated)的完整列,供 caller / embed 用
|
||||
}
|
||||
|
||||
interface EntryOp {
|
||||
index: number;
|
||||
id: string;
|
||||
kind: 'created' | 'updated';
|
||||
stmt: D1PreparedStatement;
|
||||
}
|
||||
|
||||
export async function bulkCreateEntries(db: D1Database, inputs: CreateEntryInput[]): Promise<BulkEntriesResult> {
|
||||
const results: BulkEntryItemResult[] = new Array(inputs.length);
|
||||
|
||||
// 1. 逐筆驗證(owner_id/entry_type 必填)——壞的先剔除,不阻塞好的。
|
||||
const valid: { index: number; input: CreateEntryInput; id: string }[] = [];
|
||||
for (let i = 0; i < inputs.length; i++) {
|
||||
const err = validateEntryInput(inputs[i]);
|
||||
if (err) {
|
||||
results[i] = { index: i, status: 'failed', error: err, page_name: (inputs[i] as CreateEntryInput | undefined)?.page_name ?? null };
|
||||
continue;
|
||||
}
|
||||
valid.push({ index: i, input: inputs[i], id: inputs[i].id ?? uid('e') });
|
||||
}
|
||||
|
||||
// 2. 冪等:單次查詢庫內已存在的 page_name → id(取代逐筆 GET,省 subrequest)。
|
||||
const pageNames = [...new Set(valid.map((v) => v.input.page_name).filter((p): p is string => !!p))];
|
||||
const existingByPage = new Map<string, string>();
|
||||
if (pageNames.length > 0) {
|
||||
const ph = pageNames.map(() => '?').join(',');
|
||||
const res = await db
|
||||
.prepare(`SELECT id, page_name FROM entries WHERE page_name IN (${ph})`)
|
||||
.bind(...pageNames)
|
||||
.all<{ id: string; page_name: string }>();
|
||||
for (const r of res.results ?? []) existingByPage.set(r.page_name, r.id);
|
||||
}
|
||||
|
||||
// 3. 組 INSERT / UPDATE 語句。
|
||||
const ops: EntryOp[] = [];
|
||||
for (const v of valid) {
|
||||
const pn = v.input.page_name ?? null;
|
||||
const existingId = pn ? existingByPage.get(pn) : undefined;
|
||||
if (existingId) {
|
||||
// 同 page_name 重入 → 就地更新可變欄位(entry_type/owner_id/page_name 不動)。不重複建列。
|
||||
const stmt = db
|
||||
.prepare(
|
||||
`UPDATE entries SET content = ?, parent_id = ?, refs_json = ?, tags_json = ?, task_status = ?, confidence = ?, metadata_json = ?, updated_at = unixepoch() WHERE id = ?`,
|
||||
)
|
||||
.bind(
|
||||
v.input.content ?? null,
|
||||
v.input.parent_id ?? null,
|
||||
v.input.refs_json ?? '[]',
|
||||
v.input.tags_json ?? '[]',
|
||||
v.input.task_status ?? null,
|
||||
v.input.confidence ?? null,
|
||||
v.input.metadata_json ?? null,
|
||||
existingId,
|
||||
);
|
||||
ops.push({ index: v.index, id: existingId, kind: 'updated', stmt });
|
||||
} else {
|
||||
const stmt = db
|
||||
.prepare(
|
||||
`INSERT INTO entries (id, content, entry_type, owner_id, parent_id, page_name, refs_json, tags_json, task_status, confidence, metadata_json)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
)
|
||||
.bind(
|
||||
v.id,
|
||||
v.input.content ?? null,
|
||||
v.input.entry_type,
|
||||
v.input.owner_id ?? null,
|
||||
v.input.parent_id ?? null,
|
||||
pn,
|
||||
v.input.refs_json ?? '[]',
|
||||
v.input.tags_json ?? '[]',
|
||||
v.input.task_status ?? null,
|
||||
v.input.confidence ?? null,
|
||||
v.input.metadata_json ?? null,
|
||||
);
|
||||
ops.push({ index: v.index, id: v.id, kind: 'created', stmt });
|
||||
}
|
||||
}
|
||||
|
||||
// 4. 一次 batch 寫入;整體失敗才退化逐句隔離(避免一筆壞牽連整批)。
|
||||
if (ops.length > 0) {
|
||||
try {
|
||||
await db.batch(ops.map((o) => o.stmt));
|
||||
for (const o of ops) results[o.index] = { index: o.index, status: o.kind, id: o.id };
|
||||
} catch {
|
||||
for (const o of ops) {
|
||||
try {
|
||||
await o.stmt.run();
|
||||
results[o.index] = { index: o.index, status: o.kind, id: o.id };
|
||||
} catch (e) {
|
||||
results[o.index] = { index: o.index, status: 'failed', id: o.id, error: e instanceof Error ? e.message : String(e) };
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 5. hydrate 成功寫入的列(單次 IN(...) 查詢,不逐筆 getEntry)。
|
||||
const okIds = [...new Set(ops.filter((o) => results[o.index]?.status !== 'failed').map((o) => o.id))];
|
||||
let entries: Entry[] = [];
|
||||
if (okIds.length > 0) {
|
||||
const ph = okIds.map(() => '?').join(',');
|
||||
const res = await db.prepare(`SELECT * FROM entries WHERE id IN (${ph})`).bind(...okIds).all<Entry>();
|
||||
entries = res.results ?? [];
|
||||
}
|
||||
|
||||
let created = 0, updated = 0, failed = 0;
|
||||
for (const r of results) {
|
||||
if (r.status === 'created') created++;
|
||||
else if (r.status === 'updated') updated++;
|
||||
else failed++;
|
||||
}
|
||||
return { created, updated, failed, results, entries };
|
||||
}
|
||||
|
||||
// D1 LIKE keyword search (base; semantic search is the optional embed module).
|
||||
// entry_type: optional base filter (generic — caller passes any type, base stays type-agnostic).
|
||||
export async function searchEntries(
|
||||
|
||||
@@ -134,6 +134,144 @@ export async function updateRecord(
|
||||
return getRecord(db, recordId);
|
||||
}
|
||||
|
||||
// ── Bulk records(缺口1:三元組批量寫入,avoid subrequest 撞牆)─────────────────────────
|
||||
//
|
||||
// 為什麼:一筆 record = 1 次 template 查詢 + N 個 slot(各一 INSERT entries + INSERT entry_values)。
|
||||
// 舊 createRecord 逐筆逐句串行 → 大批三元組(journals 11 筆)疊加 node/gloss 就撞 subrequest 500。
|
||||
// 解:整個 envelope 的所有 record 的所有 INSERT 攤平進「單一 db.batch()」→ 常數個 subrequest。
|
||||
// template 也一次查回(WHERE id IN (...) OR name IN (...)),不逐筆 getTemplate。
|
||||
// 租戶必填(D27/D28):每筆 record 缺 owner_id → 記 failed(value entries 的 owner 由 record 帶下去)。
|
||||
// 部分失敗不整批炸:驗證失敗 / template 不存在的先剔除;batch 整體拋錯 → 退化「逐 record 子 batch」隔離。
|
||||
|
||||
export interface BulkRecordItemResult {
|
||||
index: number;
|
||||
status: 'created' | 'failed';
|
||||
record_id?: string;
|
||||
template_id?: string;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
export interface BulkRecordsResult {
|
||||
created: number;
|
||||
failed: number;
|
||||
results: BulkRecordItemResult[];
|
||||
records: RecordResult[];
|
||||
}
|
||||
|
||||
export async function bulkCreateRecords(db: D1Database, inputs: CreateRecordInput[]): Promise<BulkRecordsResult> {
|
||||
const results: BulkRecordItemResult[] = new Array(inputs.length);
|
||||
|
||||
// 1. 驗證(template + values + owner_id 必填)+ 收集要查的 template key。
|
||||
const need: { index: number; input: CreateRecordInput }[] = [];
|
||||
const templKeys = new Set<string>();
|
||||
for (let i = 0; i < inputs.length; i++) {
|
||||
const inp = inputs[i];
|
||||
if (!inp || !inp.template || !inp.values || typeof inp.values !== 'object') {
|
||||
results[i] = { index: i, status: 'failed', error: 'template and values required' };
|
||||
continue;
|
||||
}
|
||||
if (!inp.owner_id) {
|
||||
results[i] = { index: i, status: 'failed', error: 'owner_id required: 每筆寫入必須指定租戶,避免跨租戶資料無主' };
|
||||
continue;
|
||||
}
|
||||
need.push({ index: i, input: inp });
|
||||
templKeys.add(inp.template);
|
||||
}
|
||||
|
||||
// 2. 一次查回所有用到的 template(id 或 name 皆可)。
|
||||
const templMap = new Map<string, Template>();
|
||||
if (templKeys.size > 0) {
|
||||
const keys = [...templKeys];
|
||||
const ph = keys.map(() => '?').join(',');
|
||||
const res = await db
|
||||
.prepare(`SELECT * FROM templates WHERE id IN (${ph}) OR name IN (${ph})`)
|
||||
.bind(...keys, ...keys)
|
||||
.all<Template>();
|
||||
for (const t of res.results ?? []) {
|
||||
templMap.set(t.id, t);
|
||||
templMap.set(t.name, t);
|
||||
}
|
||||
}
|
||||
|
||||
// 3. 每筆 record 組其 INSERT 語句(entries + entry_values),依 index 分組(供隔離退化用)。
|
||||
const opsByIndex = new Map<number, D1PreparedStatement[]>();
|
||||
const built = new Map<number, RecordResult>();
|
||||
for (const { index, input } of need) {
|
||||
const tpl = templMap.get(input.template);
|
||||
if (!tpl) {
|
||||
results[index] = { index, status: 'failed', error: `template not found: ${input.template}` };
|
||||
continue;
|
||||
}
|
||||
let slots: string[];
|
||||
try {
|
||||
slots = JSON.parse(tpl.slots_json);
|
||||
} catch {
|
||||
results[index] = { index, status: 'failed', error: 'template slots_json invalid' };
|
||||
continue;
|
||||
}
|
||||
const recordId = input.record_id ?? uid('rec');
|
||||
const stmts: D1PreparedStatement[] = [];
|
||||
const appliedValues: Record<string, string> = {};
|
||||
for (const slot of slots) {
|
||||
if (!(slot in input.values)) continue;
|
||||
const entryId = uid('e');
|
||||
stmts.push(
|
||||
db
|
||||
.prepare(`INSERT INTO entries (id, content, entry_type, owner_id) VALUES (?, ?, ?, ?)`)
|
||||
.bind(entryId, input.values[slot], 'value', input.owner_id ?? null),
|
||||
);
|
||||
stmts.push(
|
||||
db
|
||||
.prepare(`INSERT INTO entry_values (id, record_id, template_id, slot_name, entry_id) VALUES (?, ?, ?, ?, ?)`)
|
||||
.bind(uid('ev'), recordId, tpl.id, slot, entryId),
|
||||
);
|
||||
appliedValues[slot] = input.values[slot];
|
||||
}
|
||||
opsByIndex.set(index, stmts);
|
||||
built.set(index, { record_id: recordId, template_id: tpl.id, values: appliedValues });
|
||||
}
|
||||
|
||||
// 4. 攤平所有語句進單一 batch;整體失敗 → 逐 record 子 batch 隔離。
|
||||
const flat: D1PreparedStatement[] = [];
|
||||
for (const stmts of opsByIndex.values()) flat.push(...stmts);
|
||||
if (flat.length > 0) {
|
||||
try {
|
||||
await db.batch(flat);
|
||||
for (const index of opsByIndex.keys()) markCreated(results, built, index);
|
||||
} catch {
|
||||
for (const [index, stmts] of opsByIndex.entries()) {
|
||||
try {
|
||||
if (stmts.length > 0) await db.batch(stmts);
|
||||
markCreated(results, built, index);
|
||||
} catch (e) {
|
||||
results[index] = { index, status: 'failed', error: e instanceof Error ? e.message : String(e) };
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// 沒有任何語句(例:所有 record 的 values 都不含 template 的 slot)→ 仍記為 created(空 record)。
|
||||
for (const index of opsByIndex.keys()) markCreated(results, built, index);
|
||||
}
|
||||
|
||||
const records: RecordResult[] = [];
|
||||
let created = 0, failed = 0;
|
||||
for (const r of results) {
|
||||
if (r.status === 'created') {
|
||||
created++;
|
||||
const rec = built.get(r.index);
|
||||
if (rec) records.push(rec);
|
||||
} else {
|
||||
failed++;
|
||||
}
|
||||
}
|
||||
return { created, failed, results, records };
|
||||
}
|
||||
|
||||
function markCreated(results: BulkRecordItemResult[], built: Map<number, RecordResult>, index: number): void {
|
||||
const rec = built.get(index);
|
||||
results[index] = { index, status: 'created', record_id: rec?.record_id, template_id: rec?.template_id };
|
||||
}
|
||||
|
||||
export async function getRecord(db: D1Database, recordId: string): Promise<RecordResult | null> {
|
||||
const res = await db
|
||||
.prepare(
|
||||
|
||||
@@ -57,6 +57,37 @@ export async function embedOnWrite(env: Bindings, entry: Entry): Promise<boolean
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* 批次 embed-on-write(bulk 寫入用)。對一組已寫入的 entry,一次批處理避免逐筆 AI.run/upsert 撞 subrequest。
|
||||
* - 模組未開 → no-op。
|
||||
* - 只 embed isEmbeddable(metadata.embed===true)且 content 非空的 entry(精耕,非地毯式)。
|
||||
* - 效率:單次 AI.run(陣列輸入)+ 單次 VECTORIZE.upsert(陣列)+ 單次 UPDATE...IN(...)——
|
||||
* 與 backfillEmbeddings 同一「一批 ≈ 常數 subrequest」精神,不隨 N 線性增長。
|
||||
* 回傳實際嵌入筆數。由 caller 用 waitUntil 包成 fire-and-forget(失敗不致命)。
|
||||
*/
|
||||
export async function embedManyOnWrite(env: Bindings, entries: Entry[]): Promise<number> {
|
||||
if (!embedEnabled(env)) return 0;
|
||||
const targets = entries.filter((e) => isEmbeddable(e) && (e.content ?? '').trim().length > 0);
|
||||
if (targets.length === 0) return 0;
|
||||
const texts = targets.map((e) => (e.content ?? '').trim());
|
||||
const out = (await env.AI!.run(EMBED_MODEL, { text: texts })) as { data: number[][] };
|
||||
const data = out?.data ?? [];
|
||||
const vectors = targets
|
||||
.map((e, i) => ({ e, vec: data[i] }))
|
||||
.filter((x): x is { e: Entry; vec: number[] } => Array.isArray(x.vec) && x.vec.length > 0)
|
||||
.map((x) => ({
|
||||
id: x.e.id,
|
||||
values: x.vec,
|
||||
metadata: { owner_id: x.e.owner_id ?? '', entry_type: x.e.entry_type, source: readSource(x.e) ?? '' },
|
||||
}));
|
||||
if (vectors.length === 0) return 0;
|
||||
await env.VECTORIZE!.upsert(vectors);
|
||||
const ids = vectors.map((v) => v.id);
|
||||
const placeholders = ids.map(() => '?').join(',');
|
||||
await env.DB.prepare(`UPDATE entries SET is_embedded = 1 WHERE id IN (${placeholders})`).bind(...ids).run();
|
||||
return vectors.length;
|
||||
}
|
||||
|
||||
/** entry 是否該被 embed:caller 在 metadata_json 標 embed:true(精耕,非地毯式)。 */
|
||||
function isEmbeddable(entry: Entry): boolean {
|
||||
const meta = parseMeta(entry.metadata_json);
|
||||
|
||||
@@ -8,21 +8,44 @@ import {
|
||||
updateEntry,
|
||||
deleteEntry,
|
||||
searchEntries,
|
||||
validateEntryInput,
|
||||
bulkCreateEntries,
|
||||
} from '../actions/entry-crud';
|
||||
import { embedEnabled, embedOnWrite, semanticSearch } from '../embed';
|
||||
import { embedEnabled, embedOnWrite, embedManyOnWrite, semanticSearch } from '../embed';
|
||||
|
||||
export const entryRoutes = new Hono<{ Bindings: Bindings }>();
|
||||
|
||||
// POST /entries — create (entry_type=block/value/project/workflow/...)
|
||||
// owner_id 必填(D27/D28 機制強制):缺 → 400,避免跨租戶資料無主(gloss owner=None 查不到之坑)。
|
||||
entryRoutes.post('/', async (c) => {
|
||||
const body = await c.req.json().catch(() => null);
|
||||
if (!body || !body.entry_type) return c.json({ success: false, error: 'entry_type required' }, 400);
|
||||
const err = validateEntryInput(body);
|
||||
if (err) return c.json({ success: false, error: err }, 400);
|
||||
const entry = await createEntry(c.env.DB, body);
|
||||
// embed-on-write (#7 / #5 第4點):模組開 + entry 標 embed:true 才做;fire-and-forget,不阻塞回應、失敗不致命。
|
||||
if (embedEnabled(c.env)) c.executionCtx.waitUntil(embedOnWrite(c.env, entry).catch(() => {}));
|
||||
return c.json({ success: true, entry });
|
||||
});
|
||||
|
||||
// POST /entries/bulk — 批次寫入(缺口1 keystone)。收 { entries: [...] } 或裸陣列 [...]。
|
||||
// - 用 D1 batch,N 筆 ≈ 常數 subrequest(解「Too many subrequests」500,見 journals 11三元組+10node)。
|
||||
// - 冪等:帶 page_name 且已存在 → 就地 update;否則 insert(同 sync 腳本 GET→PATCH/POST 語意)。
|
||||
// - 逐筆結果(created/updated/failed 計數 + 失敗明細),部分失敗不整批炸。
|
||||
// - owner_id 逐筆必填(缺 → 該筆 failed,不阻塞其餘)。
|
||||
// - embed-on-write:對成功寫入且 embed:true 的 entry 批次嵌(單次 AI.run+upsert,不在迴圈逐筆撞 subrequest)。
|
||||
entryRoutes.post('/bulk', async (c) => {
|
||||
const body = await c.req.json().catch(() => null);
|
||||
const items = Array.isArray(body) ? body : body?.entries;
|
||||
if (!Array.isArray(items)) {
|
||||
return c.json({ success: false, error: 'body must be an array of entries or { entries: [...] }' }, 400);
|
||||
}
|
||||
const result = await bulkCreateEntries(c.env.DB, items);
|
||||
if (embedEnabled(c.env) && result.entries.length > 0) {
|
||||
c.executionCtx.waitUntil(embedManyOnWrite(c.env, result.entries).catch(() => {}));
|
||||
}
|
||||
return c.json({ success: true, ...result });
|
||||
});
|
||||
|
||||
// GET /entries — list with filters (entry_type, owner_id, parent_id, page_name, source, q/search)
|
||||
// e.g. list workflows under a project: ?parent_id=PROJECT&entry_type=workflow
|
||||
// e.g. get one by idempotency key: ?page_name=skill-rag_with_arcrun
|
||||
|
||||
@@ -1,16 +1,20 @@
|
||||
// Records route — structured records (entry_values composed by a template).
|
||||
import { Hono } from 'hono';
|
||||
import type { Bindings } from '../types';
|
||||
import { createRecord, getRecord, searchByTemplate, updateRecord } from '../actions/record-crud';
|
||||
import { bulkCreateRecords, createRecord, getRecord, searchByTemplate, updateRecord } from '../actions/record-crud';
|
||||
|
||||
export const recordRoutes = new Hono<{ Bindings: Bindings }>();
|
||||
|
||||
// POST /records — { template, values:{slot:content}, owner_id? }
|
||||
// POST /records — { template, values:{slot:content}, owner_id }
|
||||
// owner_id 必填(D27/D28 機制強制):record 的底層 value entries 靠此隔離租戶,缺 → 無主 → 400。
|
||||
recordRoutes.post('/', async (c) => {
|
||||
const body = await c.req.json().catch(() => null);
|
||||
if (!body || !body.template || !body.values) {
|
||||
return c.json({ success: false, error: 'template and values required' }, 400);
|
||||
}
|
||||
if (!body.owner_id) {
|
||||
return c.json({ success: false, error: 'owner_id required: 每筆寫入必須指定租戶,避免跨租戶資料無主' }, 400);
|
||||
}
|
||||
try {
|
||||
const rec = await createRecord(c.env.DB, body);
|
||||
return c.json({ success: true, record: rec });
|
||||
@@ -19,6 +23,19 @@ recordRoutes.post('/', async (c) => {
|
||||
}
|
||||
});
|
||||
|
||||
// POST /records/bulk — 批次寫入 record(缺口1:三元組批量)。收 { records: [...] } 或裸陣列 [...]。
|
||||
// - 所有 record 的所有 slot INSERT 攤平進單一 D1 batch → 常數 subrequest(解三元組批量 500)。
|
||||
// - owner_id 逐筆必填(缺 → 該筆 failed);逐筆結果 + 部分失敗不整批炸。
|
||||
recordRoutes.post('/bulk', async (c) => {
|
||||
const body = await c.req.json().catch(() => null);
|
||||
const items = Array.isArray(body) ? body : body?.records;
|
||||
if (!Array.isArray(items)) {
|
||||
return c.json({ success: false, error: 'body must be an array of records or { records: [...] }' }, 400);
|
||||
}
|
||||
const result = await bulkCreateRecords(c.env.DB, items);
|
||||
return c.json({ success: true, ...result });
|
||||
});
|
||||
|
||||
// GET /records/by-template/:template — list records of a template
|
||||
recordRoutes.get('/by-template/:template', async (c) => {
|
||||
const records = await searchByTemplate(c.env.DB, c.req.param('template'), c.req.query('owner_id') || undefined);
|
||||
|
||||
@@ -0,0 +1,260 @@
|
||||
import { describe, it, expect } from 'vitest';
|
||||
import { Hono } from 'hono';
|
||||
import { bulkCreateEntries, validateEntryInput } from '../src/actions/entry-crud';
|
||||
import { bulkCreateRecords } from '../src/actions/record-crud';
|
||||
import { entryRoutes } from '../src/routes/entries';
|
||||
import { recordRoutes } from '../src/routes/records';
|
||||
import type { Bindings } from '../src/types';
|
||||
|
||||
// ── In-memory fake D1 ────────────────────────────────────────────────────────
|
||||
// Interprets only the statement shapes the bulk/CRUD paths issue. Backed by real
|
||||
// Maps so idempotency (page_name), batch writes, and IN(...) hydration behave for real.
|
||||
interface Row { [k: string]: unknown }
|
||||
|
||||
function collapse(sql: string): string {
|
||||
return sql.replace(/\s+/g, ' ').trim();
|
||||
}
|
||||
|
||||
class FakeDB {
|
||||
entries = new Map<string, Row>();
|
||||
entryValues: Row[] = [];
|
||||
templates = new Map<string, Row>(); // keyed by id; name lookup scans
|
||||
|
||||
seedTemplate(t: { id: string; name: string; slots: string[] }) {
|
||||
this.templates.set(t.id, { id: t.id, name: t.name, slots_json: JSON.stringify(t.slots), description: null, created_by: 'system' });
|
||||
}
|
||||
|
||||
prepare(sql: string) {
|
||||
const db = this;
|
||||
let bound: unknown[] = [];
|
||||
const stmt = {
|
||||
bind(...args: unknown[]) { bound = args; return stmt; },
|
||||
async run() { db._exec(sql, bound); return { success: true, meta: {} }; },
|
||||
async all<T>() { return { results: db._query(sql, bound) as T[], success: true, meta: {} }; },
|
||||
async first<T>() { const r = db._query(sql, bound); return (r[0] ?? null) as T; },
|
||||
};
|
||||
return stmt;
|
||||
}
|
||||
|
||||
async batch(stmts: { run: () => Promise<unknown> }[]) {
|
||||
const out: unknown[] = [];
|
||||
for (const s of stmts) out.push(await s.run());
|
||||
return out;
|
||||
}
|
||||
|
||||
_exec(rawSql: string, bound: unknown[]) {
|
||||
const sql = collapse(rawSql);
|
||||
let m = sql.match(/^INSERT INTO (\w+) \(([^)]*)\) VALUES/i);
|
||||
if (m) {
|
||||
const table = m[1];
|
||||
const cols = m[2].split(',').map((s) => s.trim());
|
||||
const row: Row = {};
|
||||
cols.forEach((c, i) => { row[c] = bound[i]; });
|
||||
if (table === 'entries') {
|
||||
this.entries.set(row.id as string, {
|
||||
content: null, entry_type: null, owner_id: null, parent_id: null, page_name: null,
|
||||
refs_json: '[]', tags_json: '[]', task_status: null, content_hash: null, is_embedded: 0,
|
||||
confidence: null, metadata_json: null, created_at: 1, updated_at: 1, ...row,
|
||||
});
|
||||
} else if (table === 'entry_values') {
|
||||
this.entryValues.push(row);
|
||||
}
|
||||
return;
|
||||
}
|
||||
m = sql.match(/^UPDATE entries SET (.*) WHERE id = \?$/i);
|
||||
if (m) {
|
||||
const setClause = m[1];
|
||||
const assigns = setClause.split(',').map((s) => s.trim());
|
||||
const id = bound[bound.length - 1] as string;
|
||||
const row = this.entries.get(id);
|
||||
let bi = 0;
|
||||
for (const a of assigns) {
|
||||
const col = a.split('=')[0].trim();
|
||||
if (/=\s*unixepoch\(\)/i.test(a)) {
|
||||
if (row) row[col] = 2;
|
||||
} else {
|
||||
if (row) row[col] = bound[bi];
|
||||
bi++;
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
// ignore anything else in exec
|
||||
}
|
||||
|
||||
_query(rawSql: string, bound: unknown[]): Row[] {
|
||||
const sql = collapse(rawSql);
|
||||
if (/FROM templates/i.test(sql)) {
|
||||
const all = [...this.templates.values()];
|
||||
if (/IN \(/i.test(sql)) {
|
||||
const keys = new Set(bound.map(String));
|
||||
return all.filter((t) => keys.has(String(t.id)) || keys.has(String(t.name)));
|
||||
}
|
||||
// getTemplate: WHERE id = ? OR name = ? LIMIT 1
|
||||
const key = String(bound[0]);
|
||||
return all.filter((t) => String(t.id) === key || String(t.name) === key).slice(0, 1);
|
||||
}
|
||||
if (/FROM entries/i.test(sql)) {
|
||||
const all = [...this.entries.values()];
|
||||
if (/SELECT id, page_name/i.test(sql)) {
|
||||
const set = new Set(bound.map(String));
|
||||
return all.filter((e) => e.page_name != null && set.has(String(e.page_name))).map((e) => ({ id: e.id, page_name: e.page_name }));
|
||||
}
|
||||
if (/WHERE id IN \(/i.test(sql)) {
|
||||
const set = new Set(bound.map(String));
|
||||
return all.filter((e) => set.has(String(e.id)));
|
||||
}
|
||||
if (/WHERE id = \?/i.test(sql)) {
|
||||
return all.filter((e) => String(e.id) === String(bound[0]));
|
||||
}
|
||||
}
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
function makeEnv(db: FakeDB): Bindings {
|
||||
return { DB: asDb(db), ENVIRONMENT: 'test' } as Bindings;
|
||||
}
|
||||
|
||||
const asDb = (db: FakeDB): D1Database => db as unknown as D1Database;
|
||||
|
||||
// ── validateEntryInput ───────────────────────────────────────────────────────
|
||||
describe('validateEntryInput (owner_id 機制強制)', () => {
|
||||
it('rejects missing owner_id', () => {
|
||||
expect(validateEntryInput({ entry_type: 'block' })).toMatch(/owner_id required/);
|
||||
});
|
||||
it('rejects missing entry_type', () => {
|
||||
expect(validateEntryInput({ owner_id: 'leo' })).toBe('entry_type required');
|
||||
});
|
||||
it('rejects non-object', () => {
|
||||
expect(validateEntryInput(null)).toMatch(/must be an object/);
|
||||
});
|
||||
it('accepts a complete entry', () => {
|
||||
expect(validateEntryInput({ entry_type: 'block', owner_id: 'leo' })).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
// ── bulkCreateEntries ────────────────────────────────────────────────────────
|
||||
describe('bulkCreateEntries', () => {
|
||||
it('creates many entries in one shot; partial failure does not abort the batch', async () => {
|
||||
const db = new FakeDB();
|
||||
const r = await bulkCreateEntries(asDb(db), [
|
||||
{ entry_type: 'block', owner_id: 'leo', content: 'a' },
|
||||
{ entry_type: 'block', owner_id: 'leo', content: 'b' },
|
||||
{ entry_type: 'block', content: 'no-owner' } as never, // missing owner_id → failed
|
||||
]);
|
||||
expect(r.created).toBe(2);
|
||||
expect(r.failed).toBe(1);
|
||||
expect(r.results[2].status).toBe('failed');
|
||||
expect(r.results[2].error).toMatch(/owner_id required/);
|
||||
expect(r.entries.length).toBe(2);
|
||||
expect(db.entries.size).toBe(2); // the bad one was never written
|
||||
});
|
||||
|
||||
it('is idempotent by page_name: same key re-entry updates, does not duplicate', async () => {
|
||||
const db = new FakeDB();
|
||||
const first = await bulkCreateEntries(asDb(db), [
|
||||
{ entry_type: 'agent-skill', owner_id: 'registry', page_name: 'skill-x', content: 'v1' },
|
||||
]);
|
||||
expect(first.created).toBe(1);
|
||||
expect(db.entries.size).toBe(1);
|
||||
|
||||
const second = await bulkCreateEntries(asDb(db), [
|
||||
{ entry_type: 'agent-skill', owner_id: 'registry', page_name: 'skill-x', content: 'v2' },
|
||||
]);
|
||||
expect(second.updated).toBe(1);
|
||||
expect(second.created).toBe(0);
|
||||
expect(db.entries.size).toBe(1); // no duplicate row
|
||||
const row = [...db.entries.values()][0];
|
||||
expect(row.content).toBe('v2'); // content flipped to the new value
|
||||
});
|
||||
|
||||
it('accepts an empty array (no-op)', async () => {
|
||||
const db = new FakeDB();
|
||||
const r = await bulkCreateEntries(asDb(db), []);
|
||||
expect(r).toMatchObject({ created: 0, updated: 0, failed: 0 });
|
||||
});
|
||||
});
|
||||
|
||||
// ── bulkCreateRecords ────────────────────────────────────────────────────────
|
||||
describe('bulkCreateRecords', () => {
|
||||
it('creates records via a template; owner_id missing → that record fails', async () => {
|
||||
const db = new FakeDB();
|
||||
db.seedTemplate({ id: 'tpl-1', name: 'triplet', slots: ['subject', 'predicate', 'object'] });
|
||||
const r = await bulkCreateRecords(asDb(db), [
|
||||
{ template: 'triplet', owner_id: 'leo', values: { subject: 'cat', predicate: 'is', object: 'animal' } },
|
||||
{ template: 'triplet', owner_id: 'leo', values: { subject: 'dog', predicate: 'is', object: 'animal' } },
|
||||
{ template: 'triplet', values: { subject: 'x', predicate: 'y', object: 'z' } } as never, // no owner
|
||||
]);
|
||||
expect(r.created).toBe(2);
|
||||
expect(r.failed).toBe(1);
|
||||
expect(r.records.length).toBe(2);
|
||||
// each created record wrote 3 value entries + 3 entry_values
|
||||
expect(db.entryValues.length).toBe(6);
|
||||
expect(db.entries.size).toBe(6);
|
||||
// value entries carry the record owner (tenant isolation)
|
||||
for (const e of db.entries.values()) expect(e.owner_id).toBe('leo');
|
||||
});
|
||||
|
||||
it('unknown template → failed, not thrown', async () => {
|
||||
const db = new FakeDB();
|
||||
const r = await bulkCreateRecords(asDb(db), [{ template: 'nope', owner_id: 'leo', values: { a: 'b' } }]);
|
||||
expect(r.failed).toBe(1);
|
||||
expect(r.results[0].error).toMatch(/template not found/);
|
||||
});
|
||||
});
|
||||
|
||||
// ── Route-level owner_id enforcement (D27/D28) ───────────────────────────────
|
||||
describe('POST /entries owner_id enforcement (route 層擋死)', () => {
|
||||
const app = new Hono<{ Bindings: Bindings }>();
|
||||
app.route('/entries', entryRoutes);
|
||||
app.route('/records', recordRoutes);
|
||||
|
||||
it('POST /entries without owner_id → 400', async () => {
|
||||
const db = new FakeDB();
|
||||
const res = await app.request('/entries', {
|
||||
method: 'POST', headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ entry_type: 'block', content: 'x' }),
|
||||
}, makeEnv(db));
|
||||
expect(res.status).toBe(400);
|
||||
expect((await res.json() as { error: string }).error).toMatch(/owner_id required/);
|
||||
expect(db.entries.size).toBe(0);
|
||||
});
|
||||
|
||||
it('POST /entries with owner_id → 200 created', async () => {
|
||||
const db = new FakeDB();
|
||||
const res = await app.request('/entries', {
|
||||
method: 'POST', headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ entry_type: 'block', owner_id: 'leo', content: 'x' }),
|
||||
}, makeEnv(db));
|
||||
expect(res.status).toBe(200);
|
||||
expect((await res.json() as { success: boolean }).success).toBe(true);
|
||||
expect(db.entries.size).toBe(1);
|
||||
});
|
||||
|
||||
it('POST /entries/bulk mixed → 200 with per-item breakdown', async () => {
|
||||
const db = new FakeDB();
|
||||
const res = await app.request('/entries/bulk', {
|
||||
method: 'POST', headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ entries: [
|
||||
{ entry_type: 'block', owner_id: 'leo', content: 'a' },
|
||||
{ entry_type: 'block', content: 'b' }, // no owner
|
||||
] }),
|
||||
}, makeEnv(db));
|
||||
expect(res.status).toBe(200);
|
||||
const body = await res.json() as { created: number; failed: number };
|
||||
expect(body.created).toBe(1);
|
||||
expect(body.failed).toBe(1);
|
||||
});
|
||||
|
||||
it('POST /records without owner_id → 400', async () => {
|
||||
const db = new FakeDB();
|
||||
db.seedTemplate({ id: 'tpl-1', name: 'triplet', slots: ['s'] });
|
||||
const res = await app.request('/records', {
|
||||
method: 'POST', headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ template: 'triplet', values: { s: 'x' } }),
|
||||
}, makeEnv(db));
|
||||
expect(res.status).toBe(400);
|
||||
expect((await res.json() as { error: string }).error).toMatch(/owner_id required/);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user