diff --git a/README.md b/README.md index 8e59b81..f13ca86 100644 --- a/README.md +++ b/README.md @@ -91,7 +91,8 @@ PR #42 "Fix login timeout" was opened by @alice in repo acme/web-app > "I've reviewed all the push notifications, mark them as done." **Expected output:** -The AI calls `list_pending_events` to find push events, then `mark_processed` for each one: +The AI calls `list_pending_events` to find push events, then clears them with a single +`mark_processed({ event_ids: [...] })` call: ``` Marked 2 push events as processed: @@ -123,7 +124,7 @@ All checks passed. | `list_pending_events` | Summaries of pending events (no full payloads) | | `get_event` | Full payload for a single event by ID | | `get_webhook_events` | Full payloads for all pending events | -| `mark_processed` | Mark an event as processed | +| `mark_processed` | Mark events as processed (`event_id` for one, `event_ids` for a batch) | ## Event Retention diff --git a/docs/0-requirements.ja.md b/docs/0-requirements.ja.md index c0c3ba4..5ce38a5 100644 --- a/docs/0-requirements.ja.md +++ b/docs/0-requirements.ja.md @@ -102,12 +102,21 @@ WebhookMcpAgent DO が以下のツールセットを提供する。ローカル | F3.2 | `list_pending_events` | limit (1-100, default 20) | サマリー配列 | 未処理イベントのメタデータ一覧を返す（ペイロード含まず） | | F3.3 | `get_event` | event_id | 完全イベント or error | UUID 指定で完全なペイロードを返す | | F3.4 | `get_webhook_events` | limit (1-100, default 20) | 未処理イベント配列 | 未処理イベントをフルペイロード付きで返す | -| F3.5 | `mark_processed` | event_id | success, event_id, purged | イベントを処理済みにマークし、保持期間超過の処理済みイベントを自動削除する。`purged` は今回削除された件数 | +| F3.5 | `mark_processed` | event_id **または** event_ids (1-100) | 単数形: success, event_id, purged／バッチ形: success, marked, failed, results (id 単位の判定), purged | イベントを処理済みにマークし、保持期間超過の処理済みイベントを自動削除する。`purged` は今回削除された件数 | **F3.1 ローカルブリッジ整形:** ローカルブリッジは `get_pending_status` の戻り値を Claude Code UserPromptSubmit hook の decision JSON shape (`hookSpecificOutput.hookEventName="UserPromptSubmit"` + `additionalContext` に pending_count / types / latest_received_at の自然文要約) にラップして返す。これは `type: "mcp_tool"` UserPromptSubmit hook 経由の呼び出しで戻り値が AI 文脈に注入されるための要件であり、手動 tool 呼び出し時も同 shape で返る。リモート (Worker + DO) 側の戻り値構造は変更しない。 **F3.1 空状態の silent (empty silent, #221):** `pending_count == 0` の場合はラップせずリモート戻り値をそのまま返す。Claude Code 側で decision schema に一致しない JSON は silent discard されるため、hook 経由の呼び出しで `additionalContext` に何も注入されず、毎ターン空 reminder のノイズが消える。手動 tool 呼び出しではリモートの raw payload (`{pending_count: 0, types: {}, latest_received_at: null}`) が返り、AI は内容を直接判定できる。 +**F3.5 バッチ形 (#245):** `mark_processed` が 1 呼び出しにつき 1 イベントしか受けなかったため、自己操作 1 回で 6〜10 件生じる到達確認イベントの消費が呼び出し回数に比例していた。`event_ids: string[]` を追加し、id を明示列挙したまま往復だけ畳む（filter 一括消費は外部イベントを誤って消しうるため採らない）。 + +- **後方互換:** 既存の単数 `event_id` 呼び出しは戻り値の形も含めて不変。存在しない id でも `success: true` を返す旧挙動を維持する。 +- **id 単位の判定:** バッチ形は id ごとに `{event_id, success, error?}` を返す。tool 表層で `error` が取る値は `not found`（どの store にも無い）のみ。空の id は store に届く前に tool schema が弾くため、誤解を招く `not found` に化けることはない。 +- **部分失敗:** 1 件の失敗で全体を落とさない。成功した id のマークは確定済みで、呼び出し側は失敗した id だけ再送すればよい。部分失敗は tool error ではなく本文で報告する。 +- **マルチアカウント:** イベントは 1 つの store にのみ存在するため、アクセス可能な全 store に同じバッチを投げ、**いずれかの store が一致した id を成功**とする。全 store が取り逃した id のみ失敗。 +- **purge 回数:** 保持期間 purge はバッチ 1 回につき 1 回のみ走る（id ごとではない）。 +- **MCP proxy:** 静的スキーマ (`mcp-server/server/index.js` / `local-mcp/src/index.ts`) の更新が要る。新 param は再接続では反映されず npm 再公開が必要。 + **イベントサマリー構造:** ```json @@ -154,7 +163,7 @@ WebhookMcpAgent DO が以下のツールセットを提供する。ローカル | 1 | `get_pending_status()` を 60 秒間隔でポーリング | | 2 | `pending_count > 0` なら `list_pending_events()` でサマリー取得 | | 3 | フルペイロードが必要なイベントのみ `get_event(event_id)` で取得 | -| 4 | 処理完了後 `mark_processed(event_id)` でマーク | +| 4 | 処理完了後 `mark_processed` でマーク。複数件をまとめて処理した場合は `event_ids` で 1 呼び出しに畳む | ### F7. OAuth 認証（Worker-hosted web OAuth） diff --git a/docs/0-requirements.md b/docs/0-requirements.md index b3c0ca3..a90ed74 100644 --- a/docs/0-requirements.md +++ b/docs/0-requirements.md @@ -102,12 +102,21 @@ WebhookMcpAgent DO が以下のツールセットを提供する。ローカル | F3.2 | `list_pending_events` | limit (1-100, default 20) | サマリー配列 | 未処理イベントのメタデータ一覧を返す（ペイロード含まず） | | F3.3 | `get_event` | event_id | 完全イベント or error | UUID 指定で完全なペイロードを返す | | F3.4 | `get_webhook_events` | limit (1-100, default 20) | 未処理イベント配列 | 未処理イベントをフルペイロード付きで返す | -| F3.5 | `mark_processed` | event_id | success, event_id, purged | イベントを処理済みにマークし、保持期間超過の処理済みイベントを自動削除する。`purged` は今回削除された件数 | +| F3.5 | `mark_processed` | event_id **または** event_ids (1-100) | 単数形: success, event_id, purged／バッチ形: success, marked, failed, results (id 単位の判定), purged | イベントを処理済みにマークし、保持期間超過の処理済みイベントを自動削除する。`purged` は今回削除された件数 | **F3.1 Local bridge shaping:** The local bridge wraps `get_pending_status` results into the Claude Code UserPromptSubmit hook decision JSON shape (`hookSpecificOutput.hookEventName="UserPromptSubmit"` plus a natural-language summary of pending_count / types / latest_received_at in `additionalContext`). This is required so values returned via `type: "mcp_tool"` UserPromptSubmit hooks reach the AI prompt context; manual tool calls receive the same shape. The remote (Worker + DO) return contract is unchanged. **F3.1 Empty silent (#221):** When `pending_count == 0`, the local bridge returns the remote payload untouched (no wrap). Claude Code silently discards JSON that does not match a decision schema, so hook callers receive nothing in `additionalContext` — eliminating the per-turn empty-reminder noise. Manual tool callers see the raw payload (`{pending_count: 0, types: {}, latest_received_at: null}`) and can interpret it directly. +**F3.5 Batch form (#245):** `mark_processed` accepted exactly one event per call, so consuming the 6-10 self-operation acknowledgement events a single PR generates cost one round trip each. `event_ids: string[]` is added so the round trips collapse while the caller still enumerates ids explicitly (filter-based bulk consumption is deliberately not offered: it can silently consume external events). + +- **Backward compatibility:** the existing singular `event_id` call is unchanged, response shape included. It still answers `success: true` for an id that was never ingested. +- **Per-id verdict:** the batch form returns `{event_id, success, error?}` per id. `not found` (no store held that id) is the only value `error` takes at the tool surface — an empty id is rejected by the tool schema before it reaches a store, so it never turns into a misleading `not found`. +- **Partial failure:** one failing id does not fail the call. Marks for the successful ids are already committed, so the caller retries only the failed ids. Partial failure is reported in the body, not as a tool error. +- **Multi-account:** an event lives in exactly one store, so the same batch goes to every accessible store and an id marked by **any** store counts as marked. Only an id missed by every store failed. +- **Purge cadence:** the retention purge runs once per batch call, not once per id. +- **MCP proxy:** the static schemas (`mcp-server/server/index.js` / `local-mcp/src/index.ts`) must be updated. A new param is not picked up by reconnecting; the npm package has to be republished. + **イベントサマリー構造:** ```json @@ -154,7 +163,7 @@ WebhookMcpAgent DO が以下のツールセットを提供する。ローカル | 1 | `get_pending_status()` を 60 秒間隔でポーリング | | 2 | `pending_count > 0` なら `list_pending_events()` でサマリー取得 | | 3 | フルペイロードが必要なイベントのみ `get_event(event_id)` で取得 | -| 4 | 処理完了後 `mark_processed(event_id)` でマーク | +| 4 | 処理完了後 `mark_processed` でマーク。複数件をまとめて処理した場合は `event_ids` で 1 呼び出しに畳む | ### F7. OAuth 認証（Worker-hosted web OAuth） diff --git a/docs/Home.md b/docs/Home.md index 51f451d..0e80a4c 100644 --- a/docs/Home.md +++ b/docs/Home.md @@ -55,7 +55,7 @@ GitHub --POST--> Cloudflare Worker --> Durable Object (SQLite) | `list_pending_events` | 未処理イベントのサマリー（フルペイロードなし） | | `get_event` | ID 指定で単一イベントのフルペイロード取得 | | `get_webhook_events` | 全未処理イベントのフルペイロード取得 | -| `mark_processed` | イベントを処理済みにマーク | +| `mark_processed` | イベントを処理済みにマーク（`event_ids` で複数件を 1 呼び出しにまとめられる） | ## モノレポ構成 diff --git a/local-mcp/src/index.ts b/local-mcp/src/index.ts index 7d36683..8593217 100644 --- a/local-mcp/src/index.ts +++ b/local-mcp/src/index.ts @@ -798,13 +798,21 @@ const TOOLS = [ }, { name: "mark_processed", - description: "Mark a webhook event as processed", + description: + "Mark webhook events as processed. Pass event_ids to clear a whole batch in one call (preferred when several events were handled together); event_id marks a single event.", inputSchema: { type: "object" as const, properties: { - event_id: { type: "string", description: "The event ID to mark" }, + event_id: { type: "string", description: "A single event ID to mark" }, + event_ids: { + type: "array", + items: { type: "string", minLength: 1 }, + minItems: 1, + maxItems: 100, + description: + "Event IDs to mark in one call (1-100). Returns a per-id success/failure verdict; ids that succeed stay marked even if others fail.", + }, }, - required: ["event_id"], }, }, ]; diff --git a/mcp-server/README.md b/mcp-server/README.md index d1d9c86..e6c2fb9 100644 --- a/mcp-server/README.md +++ b/mcp-server/README.md @@ -138,14 +138,14 @@ All tools are read-only except `mark_processed`. | `list_pending_events` | Summary list of pending events (`limit`: 1-100, default 20). Returns metadata only — `id`, `type`, `action`, `repo`, `sender`, `number`, `title`, `url`, `received_at` — without the full payload. | | `get_event` | Full payload for a single webhook event by `event_id`. | | `get_webhook_events` | Pending events with full payloads. Prefer `get_pending_status` or `list_pending_events` for polling and only fall back to this when you really need everything. | -| `mark_processed` | Mark an event as processed by `event_id` so it will no longer appear in pending queries. Required to keep the pending queue from growing unbounded. | +| `mark_processed` | Mark events as processed so they no longer appear in pending queries. Pass `event_id` for one event, or `event_ids` (1-100) to clear a whole batch in a single call. Required to keep the pending queue from growing unbounded. | ### Recommended polling flow 1. Poll `get_pending_status()` periodically (e.g. every 60 seconds). 2. If `pending_count > 0`, call `list_pending_events()` for summaries. 3. Call `get_event(event_id)` only for events that need the full payload. -4. Call `mark_processed(event_id)` after handling each event. +4. Call `mark_processed` after handling the events — `event_ids: [...]` clears the whole set in one call, which is the normal case when a batch of events was handled together. If real-time channel notifications are enabled (Claude Code), step 1 can be skipped — the proxy will push event summaries as soon as the Worker receives them. You still need to call `mark_processed` to clear the queue. diff --git a/mcp-server/manifest.json b/mcp-server/manifest.json index 705523e..4fe12d2 100644 --- a/mcp-server/manifest.json +++ b/mcp-server/manifest.json @@ -66,7 +66,7 @@ }, { "name": "mark_processed", - "description": "Mark a webhook event as processed so it won't appear again." + "description": "Mark webhook events as processed so they won't appear again. Accepts a single event_id or a batch of event_ids." } ], "compatibility": { diff --git a/mcp-server/server/index.js b/mcp-server/server/index.js index 188705f..ee7f1c3 100644 --- a/mcp-server/server/index.js +++ b/mcp-server/server/index.js @@ -758,21 +758,28 @@ const TOOLS = [ }, { name: "mark_processed", - title: "Mark Event Processed", + title: "Mark Events Processed", description: - "Mark a webhook event as processed so it won't appear again.", + "Mark webhook events as processed so they won't appear again. Pass event_ids to clear a whole batch in one call (preferred when several events were handled together); event_id marks a single event.", inputSchema: { type: "object", properties: { event_id: { type: "string", - description: "The event ID to mark as processed", + description: "A single event ID to mark as processed", + }, + event_ids: { + type: "array", + items: { type: "string", minLength: 1 }, + minItems: 1, + maxItems: 100, + description: + "Event IDs to mark as processed in one call (1-100). Returns a per-id success/failure verdict; ids that succeed stay marked even if others fail.", }, }, - required: ["event_id"], }, annotations: { - title: "Mark Event Processed", + title: "Mark Events Processed", destructiveHint: true, }, }, diff --git a/worker/src/agent.ts b/worker/src/agent.ts index dfa0f26..b1de27f 100644 --- a/worker/src/agent.ts +++ b/worker/src/agent.ts @@ -13,6 +13,7 @@ import { McpAgent } from "agents/mcp"; import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { z } from "zod"; import type { PendingStatus, EventSummary, WebhookEvent } from "../../shared/src/types.js"; +import { mergeMarkResults, type StoreBatchResponse } from "./mark-results.js"; interface Env { MCP_OBJECT: DurableObjectNamespace; @@ -20,6 +21,13 @@ interface Env { TENANT_REGISTRY: DurableObjectNamespace; } +/** + * Upper bound on ids per batched mark_processed call. Matches the 100 ceiling + * the listing tools use for `limit`, so a batch can always clear one full page + * of pending events. + */ +const MARK_BATCH_MAX = 100; + /** Tenant context passed via props when creating per-tenant instances */ export type TenantProps = { account_id?: number; @@ -57,6 +65,17 @@ export class WebhookMcpAgent extends McpAgent { }); } + /** POST a mark-processed body (singular or batch) to one store. */ + private markRequest(store: DurableObjectStub, body: unknown): Promise { + return store.fetch( + new Request("https://store/mark-processed", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(body), + }), + ); + } + async init() { this.server.tool( "get_pending_status", @@ -152,21 +171,42 @@ export class WebhookMcpAgent extends McpAgent { this.server.tool( "mark_processed", - "Mark a webhook event as processed", - { event_id: z.string() }, - async ({ event_id }) => { - // Try all stores — the event lives in exactly one, others are no-ops + "Mark webhook events as processed. Pass event_ids to clear a whole batch in one call (preferred when several events were handled together); event_id marks a single event.", + { + event_id: z.string().optional(), + // min(1) on the ITEM, not just the array: an empty-string id would pass + // schema validation, miss in every store, and reach the caller as + // "not found" — a misleading verdict for what is really a malformed + // request. Rejecting it here keeps "not found" the only per-id error. + event_ids: z.array(z.string().min(1)).min(1).max(MARK_BATCH_MAX).optional(), + }, + async ({ event_id, event_ids }) => { const stores = this.getStores(); + + // ── Batch form (#245): one round trip for N ids ── + if (event_ids) { + const perStore = await Promise.all( + stores.map((s) => + this.markRequest(s, { event_ids }).then((r) => r.json() as Promise), + ), + ); + + // Per-id verdict resolution across the fan-out lives in + // mark-results.ts (unit-tested there). + const summary = mergeMarkResults(event_ids, perStore); + return { content: [{ type: "text", text: JSON.stringify(summary, null, 2) }] }; + } + + // ── Singular form: response shape unchanged ── + if (!event_id) { + return { + content: [{ type: "text", text: "mark_processed requires event_id or event_ids" }], + isError: true, + }; + } + // Try all stores — the event lives in exactly one, others are no-ops const results = await Promise.all( - stores.map((s) => - s.fetch( - new Request("https://store/mark-processed", { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ event_id }), - }), - ).then((r) => r.json()), - ), + stores.map((s) => this.markRequest(s, { event_id }).then((r) => r.json())), ); // Return the first successful result return { content: [{ type: "text", text: JSON.stringify(results[0]) }] }; diff --git a/worker/src/mark-results.ts b/worker/src/mark-results.ts new file mode 100644 index 0000000..6ca1898 --- /dev/null +++ b/worker/src/mark-results.ts @@ -0,0 +1,79 @@ +/** + * Cross-store merge for the batched mark_processed form (#245). + * + * Kept out of agent.ts so it can be unit-tested without the McpAgent / + * Durable Object runtime: the merge is where a batch's per-id verdict is + * actually decided, and that decision is easy to get subtly wrong under + * multi-account fan-out. + * + * Fan-out shape: a session may read several tenant stores (user + orgs). The + * same batch is POSTed to every accessible store, but each event lives in + * exactly ONE of them — so "not found" from the other stores is the normal + * case, not a failure. An id is marked when ANY store matched it; only an id + * that every store missed has actually failed. + */ + +/** Per-id verdict surfaced to the caller. */ +export type MarkResult = { event_id: string; success: boolean; error?: string }; + +/** Batch response shape from one WebhookStore DO's /mark-processed route. */ +export type StoreBatchResponse = { + success: boolean; + results?: { event_id: unknown; success: boolean; error?: string }[]; + purged?: number; +}; + +/** Aggregate returned by the batched mark_processed tool. */ +export type MarkBatchSummary = { + success: boolean; + marked: number; + failed: number; + results: MarkResult[]; + purged: number; +}; + +/** + * Merge per-store batch responses into one per-id verdict list. + * + * `results` follows the order of `event_ids` (the order the caller asked in), + * so a caller can line the verdicts up against its own list. `success` is the + * all-ids-marked verdict; partial failure is reported here rather than as a + * tool error, because the ids that succeeded are already committed and the + * caller needs to see exactly which ids to retry. + */ +export function mergeMarkResults( + event_ids: string[], + perStore: StoreBatchResponse[], +): MarkBatchSummary { + const verdicts = new Map( + event_ids.map((id) => [id, { event_id: id, success: false, error: "not found" }]), + ); + + let purged = 0; + for (const store of perStore) { + purged += store.purged ?? 0; + for (const r of store.results ?? []) { + // Only successes are promoted: a miss from one store says nothing about + // the others, and the map already holds "not found" as the default. + // Store-side `error` strings are deliberately NOT propagated — a + // per-store reason describes one store, not the merged verdict. This is + // why "not found" is the only error the tool surface emits, and why the + // tool schema rejects empty ids rather than letting them arrive here as + // a store-level "invalid event_id" that this merge would flatten. + if (r.success && typeof r.event_id === "string" && verdicts.has(r.event_id)) { + verdicts.set(r.event_id, { event_id: r.event_id, success: true }); + } + } + } + + const results = [...verdicts.values()]; + const failed = results.filter((r) => !r.success).length; + + return { + success: failed === 0, + marked: results.length - failed, + failed, + results, + purged, + }; +} diff --git a/worker/src/store.ts b/worker/src/store.ts index ea68987..83fb07c 100644 --- a/worker/src/store.ts +++ b/worker/src/store.ts @@ -125,6 +125,48 @@ export class WebhookStore extends DurableObject { return { processed, unprocessed }; } + /** + * Purge processed events whose received_at is older than PURGE_AFTER_DAYS. + * Unprocessed events are never touched here regardless of age (the time-based + * sweep owns that class). Returns the number of rows deleted. + * + * Shared by the singular and the batch mark-processed paths so a batch runs + * exactly one purge instead of one per id — that per-call collapse is the + * point of the batch form (#245). + */ + private purgeProcessed(): number { + const cutoff = new Date(Date.now() - purgeDays(this.env) * 86_400_000).toISOString(); + const cursor = this.ctx.storage.sql.exec( + `DELETE FROM events WHERE processed = 1 AND received_at < ?`, cutoff, + ); + return cursor.rowsWritten; + } + + /** + * Mark one event processed, reporting whether a row in THIS store matched. + * `success: false` here means "not in this store" — with multi-account + * fan-out the caller asks every accessible store and merges, so a miss is + * expected on all but one. The agent decides the final per-id verdict. + * + * Each id is isolated: a throw on one id is caught and reported, leaving the + * writes already applied for the other ids intact (partial-failure contract). + */ + private markOne(rawId: unknown): { event_id: unknown; success: boolean; error?: string } { + if (typeof rawId !== "string" || rawId.length === 0) { + return { event_id: rawId, success: false, error: "invalid event_id" }; + } + try { + const cursor = this.ctx.storage.sql.exec( + `UPDATE events SET processed = 1 WHERE id = ?`, rawId, + ); + return cursor.rowsWritten > 0 + ? { event_id: rawId, success: true } + : { event_id: rawId, success: false, error: "not found" }; + } catch (err) { + return { event_id: rawId, success: false, error: String(err) }; + } + } + /** * DO Alarm handler — the consumption-independent retention guarantee. Runs the * full sweep (processed + unprocessed) and reschedules the next sweep so the @@ -322,23 +364,28 @@ export class WebhookStore extends DurableObject { } // ── mark_processed ── + // Two request forms on one route: + // { event_id } → singular, response shape frozen for compatibility + // { event_ids: [...] } → batch, per-id verdicts in `results` + // Auto-purge runs once per call either way: it deletes processed events whose + // received_at is older than the retention window, bounding DO storage growth + // from dead processed rows (re-port of #29). Unprocessed events are never + // deleted here regardless of age. if (url.pathname === "/mark-processed" && request.method === "POST") { - const { event_id } = await request.json() as { event_id: string }; + const body = await request.json() as { event_id?: string; event_ids?: unknown[] }; + + if (Array.isArray(body.event_ids)) { + // Marks are applied id-by-id before the purge, so ids that failed do not + // undo the ones that succeeded — the successes are already committed. + const results = body.event_ids.map((id) => this.markOne(id)); + return Response.json({ success: true, results, purged: this.purgeProcessed() }); + } + + const { event_id } = body as { event_id: string }; this.ctx.storage.sql.exec( `UPDATE events SET processed = 1 WHERE id = ?`, event_id, ); - - // Auto-purge: delete processed events whose received_at is older than the - // retention window. Unprocessed events are never deleted regardless of age. - // This bounds DO storage growth from dead processed rows (re-port of #29). - const days = purgeDays(this.env); - const cutoff = new Date(Date.now() - days * 86_400_000).toISOString(); - const cursor = this.ctx.storage.sql.exec( - `DELETE FROM events WHERE processed = 1 AND received_at < ?`, cutoff, - ); - const purged = cursor.rowsWritten; - - return Response.json({ success: true, event_id, purged }); + return Response.json({ success: true, event_id, purged: this.purgeProcessed() }); } // ── sweep (time-based retention purge) ── diff --git a/worker/test/mark-results.test.ts b/worker/test/mark-results.test.ts new file mode 100644 index 0000000..4f4b712 --- /dev/null +++ b/worker/test/mark-results.test.ts @@ -0,0 +1,125 @@ +/** + * Unit tests for worker/src/mark-results.ts :: mergeMarkResults (#245). + * + * This is the cross-store merge behind the batched mark_processed form. The + * batch is POSTed to every store the session can read, but each event lives in + * exactly one of them — so the merge has to read "not found" from the other + * stores as normal, not as failure. Getting that backwards would report every + * id as failed on any multi-account session, so it is pinned directly here + * rather than only through the DO integration tests. + */ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { mergeMarkResults, type StoreBatchResponse } from "../src/mark-results.js"; + +/** One store's batch response, with the shape the DO route returns. */ +function store( + results: { event_id: unknown; success: boolean; error?: string }[], + purged = 0, +): StoreBatchResponse { + return { success: true, results, purged }; +} + +test("single store: all ids matched", () => { + const merged = mergeMarkResults( + ["a", "b"], + [store([{ event_id: "a", success: true }, { event_id: "b", success: true }], 2)], + ); + assert.deepEqual(merged, { + success: true, + marked: 2, + failed: 0, + results: [ + { event_id: "a", success: true }, + { event_id: "b", success: true }, + ], + purged: 2, + }); +}); + +test("multi-store fan-out: an id matched by ANY store counts as marked", () => { + // "a" lives in store 1, "b" in store 2. Each store misses the other's id. + const merged = mergeMarkResults( + ["a", "b"], + [ + store([ + { event_id: "a", success: true }, + { event_id: "b", success: false, error: "not found" }, + ]), + store([ + { event_id: "a", success: false, error: "not found" }, + { event_id: "b", success: true }, + ]), + ], + ); + assert.equal(merged.success, true); + assert.equal(merged.marked, 2); + assert.equal(merged.failed, 0); +}); + +test("an id missed by EVERY store is the only real failure", () => { + const merged = mergeMarkResults( + ["a", "ghost"], + [ + store([ + { event_id: "a", success: true }, + { event_id: "ghost", success: false, error: "not found" }, + ]), + store([ + { event_id: "a", success: false, error: "not found" }, + { event_id: "ghost", success: false, error: "not found" }, + ]), + ], + ); + assert.equal(merged.success, false); + assert.equal(merged.marked, 1); + assert.equal(merged.failed, 1); + assert.deepEqual(merged.results, [ + { event_id: "a", success: true }, + { event_id: "ghost", success: false, error: "not found" }, + ]); +}); + +test("results follow the caller's id order regardless of store response order", () => { + const merged = mergeMarkResults( + ["z", "y", "x"], + [store([{ event_id: "x", success: true }, { event_id: "z", success: true }])], + ); + assert.deepEqual( + merged.results.map((r) => r.event_id), + ["z", "y", "x"], + ); +}); + +test("purged counts sum across stores", () => { + const merged = mergeMarkResults( + ["a"], + [store([{ event_id: "a", success: true }], 3), store([], 4)], + ); + assert.equal(merged.purged, 7); +}); + +test("a store omitting results / purged does not break the merge", () => { + // Defensive against an older DO revision answering the singular shape. + const merged = mergeMarkResults(["a"], [{ success: true } as StoreBatchResponse]); + assert.deepEqual(merged, { + success: false, + marked: 0, + failed: 1, + results: [{ event_id: "a", success: false, error: "not found" }], + purged: 0, + }); +}); + +test("an id a store reports that the caller never asked for is ignored", () => { + const merged = mergeMarkResults( + ["a"], + [store([{ event_id: "a", success: true }, { event_id: "stray", success: true }])], + ); + assert.deepEqual(merged.results, [{ event_id: "a", success: true }]); +}); + +test("empty id list yields an empty, successful merge", () => { + const merged = mergeMarkResults([], [store([])]); + assert.deepEqual(merged, { success: true, marked: 0, failed: 0, results: [], purged: 0 }); +}); diff --git a/worker/test/workers/store.test.ts b/worker/test/workers/store.test.ts index e7bc424..b68f545 100644 --- a/worker/test/workers/store.test.ts +++ b/worker/test/workers/store.test.ts @@ -218,6 +218,136 @@ describe("WebhookStore: mark-processed", () => { const ev = (await evRes.json()) as WebhookEvent; expect(ev.processed).toBe(true); }); + + // The batch form (#245) reports a per-id "not found", but the singular form + // never did and must not start: callers depend on the flat, always-success + // response shape. Pinned so adding batch strictness cannot leak into it. + it("still answers success for an unknown id in the singular form", async () => { + const stub = storeFor("mark-processed-unknown-id"); + const mp = await stub.fetch( + new Request(`${BASE}/mark-processed`, { + method: "POST", + body: JSON.stringify({ event_id: "never-ingested" }), + }), + ); + expect(mp.status).toBe(200); + expect(await mp.json()).toEqual({ success: true, event_id: "never-ingested", purged: 0 }); + }); +}); + +// Batch form of /mark-processed (#245): { event_ids: [...] } in place of +// { event_id }. Returns a per-id verdict so the caller can see which ids missed +// without re-sending the batch. +type BatchResult = { event_id: unknown; success: boolean; error?: string }; +type BatchResponse = { success: boolean; results: BatchResult[]; purged: number }; + +async function markBatch(stub: DurableObjectStub, event_ids: unknown[]) { + const res = await stub.fetch( + new Request(`${BASE}/mark-processed`, { + method: "POST", + body: JSON.stringify({ event_ids }), + }), + ); + expect(res.status).toBe(200); + return res.json() as Promise; +} + +describe("WebhookStore: mark-processed batch (event_ids)", () => { + it("marks every id in one call and reports a per-id verdict", async () => { + const stub = storeFor("batch-marks-all"); + for (const id of ["b1", "b2", "b3"]) { + await ingest(stub, makeEvent({ id, received_at: isoFromNow(-1 * DAY_MS) })); + } + + const body = await markBatch(stub, ["b1", "b2", "b3"]); + expect(body.success).toBe(true); + expect(body.results).toEqual([ + { event_id: "b1", success: true }, + { event_id: "b2", success: true }, + { event_id: "b3", success: true }, + ]); + + // all three dropped out of pending, all three still fetchable as processed + const status = (await (await stub.fetch(new Request(`${BASE}/pending-status`))).json()) as PendingStatus; + expect(status.pending_count).toBe(0); + for (const id of ["b1", "b2", "b3"]) { + const ev = (await (await stub.fetch(new Request(`${BASE}/event?id=${id}`))).json()) as WebhookEvent; + expect(ev.processed).toBe(true); + } + }); + + it("commits the successful ids when one id in the batch fails", async () => { + const stub = storeFor("batch-partial-failure"); + await ingest(stub, makeEvent({ id: "ok1", received_at: isoFromNow(-1 * DAY_MS) })); + await ingest(stub, makeEvent({ id: "ok2", received_at: isoFromNow(-1 * DAY_MS) })); + + // "ghost" is not in the store, and "" is not a usable id — neither may + // prevent ok1 / ok2 from being marked. + const body = await markBatch(stub, ["ok1", "ghost", "ok2", ""]); + expect(body.success).toBe(true); // the call itself did not fail + expect(body.results).toEqual([ + { event_id: "ok1", success: true }, + { event_id: "ghost", success: false, error: "not found" }, + { event_id: "ok2", success: true }, + { event_id: "", success: false, error: "invalid event_id" }, + ]); + + // the successes are durable: both are processed and out of pending + const status = (await (await stub.fetch(new Request(`${BASE}/pending-status`))).json()) as PendingStatus; + expect(status.pending_count).toBe(0); + for (const id of ["ok1", "ok2"]) { + const ev = (await (await stub.fetch(new Request(`${BASE}/event?id=${id}`))).json()) as WebhookEvent; + expect(ev.processed).toBe(true); + } + }); + + it("runs the retention purge once for the whole batch", async () => { + const stub = storeFor("batch-purges-once"); + // one stale processed row: a per-id purge would report it repeatedly, + // a single per-call purge reports it exactly once + await ingest(stub, makeEvent({ id: "stale", received_at: isoFromNow(-30 * DAY_MS), processed: true })); + await ingest(stub, makeEvent({ id: "n1", received_at: isoFromNow(-1 * DAY_MS) })); + await ingest(stub, makeEvent({ id: "n2", received_at: isoFromNow(-1 * DAY_MS) })); + + const body = await markBatch(stub, ["n1", "n2"]); + expect(body.purged).toBe(1); + expect((await stub.fetch(new Request(`${BASE}/event?id=stale`))).status).toBe(404); + }); + + it("treats an empty id list as a no-op that still succeeds", async () => { + const stub = storeFor("batch-empty"); + await ingest(stub, makeEvent({ id: "untouched", received_at: isoFromNow(-1 * DAY_MS) })); + + const body = await markBatch(stub, []); + expect(body).toEqual({ success: true, results: [], purged: 0 }); + + const status = (await (await stub.fetch(new Request(`${BASE}/pending-status`))).json()) as PendingStatus; + expect(status.pending_count).toBe(1); + }); + + it("is idempotent: re-marking an already-processed id still reports success", async () => { + const stub = storeFor("batch-idempotent"); + await ingest(stub, makeEvent({ id: "twice", received_at: isoFromNow(-1 * DAY_MS) })); + + expect((await markBatch(stub, ["twice"])).results).toEqual([{ event_id: "twice", success: true }]); + expect((await markBatch(stub, ["twice"])).results).toEqual([{ event_id: "twice", success: true }]); + }); + + it("leaves the singular event_id form untouched when both keys could apply", async () => { + // event_ids selects the batch form; the singular response shape is reserved + // for requests that carry event_id alone. + const stub = storeFor("batch-form-selection"); + await ingest(stub, makeEvent({ id: "s1", received_at: isoFromNow(-1 * DAY_MS) })); + + const res = await stub.fetch( + new Request(`${BASE}/mark-processed`, { + method: "POST", + body: JSON.stringify({ event_id: "s1", event_ids: ["s1"] }), + }), + ); + const body = (await res.json()) as BatchResponse; + expect(body.results).toEqual([{ event_id: "s1", success: true }]); + }); }); describe("WebhookStore: mark-processed auto-purge", () => {