From 4fef8324dafba36813a0fb870e49cb7c0836ffa3 Mon Sep 17 00:00:00 2001 From: Nathan Disidore Date: Mon, 9 Nov 2026 14:43:11 -0600 Subject: [PATCH] Keep the GitHub, Gmail, Chat and Outlook hooks in the kit's hook registry Each driver kept its own reg:/caps: records, stub disposal and startHook() -> deliver() call around the kit's delivery queue. They now use HookRegistry for all of it, under the same storage keys, and alias their delivery interfaces to HookDelivery, which drops the suppression on the deliver call. Each calls set() inside ctx.storage.transactionSync(), so a registration that can't be stored leaves the hook as it was. Alarms, provider lifecycles, ingest filters, controllers and GitHub's disconnect sweep are unchanged. --- .../gatekeeper-github/src/github-hooks.ts | 69 ++++-------------- packages/gatekeeper-google/src/chat-hooks.ts | 70 +++++-------------- packages/gatekeeper-google/src/gmail-hooks.ts | 69 +++++------------- .../__tests__/outlook.test.ts | 5 +- .../gatekeeper-microsoft/src/outlook-hooks.ts | 47 ++++--------- plans/gatekeeper-kit.md | 9 ++- 6 files changed, 72 insertions(+), 197 deletions(-) diff --git a/packages/gatekeeper-github/src/github-hooks.ts b/packages/gatekeeper-github/src/github-hooks.ts index ac71cab15e..3d9195d2ef 100644 --- a/packages/gatekeeper-github/src/github-hooks.ts +++ b/packages/gatekeeper-github/src/github-hooks.ts @@ -17,14 +17,11 @@ import { DurableObject, RpcTarget, WorkerEntrypoint, type RpcStub } from "cloudflare:workers"; import { validateRpc } from "capnweb-validate"; -import { - DELIVERED_RETENTION_MS, HookDeliveryQueue, disposeStubs, -} from "@gadgets/gatekeeper-kit/hook-delivery-queue"; +import { DELIVERED_RETENTION_MS, disposeStubs } from "@gadgets/gatekeeper-kit/hook-delivery-queue"; +import { HookRegistry, type HookCapabilities, type HookDelivery } from "@gadgets/gatekeeper-kit/hook-registry"; import { ResponseTooLargeError, readBytesCapped } from "@gadgets/gatekeeper-kit/response-body"; import { SerialTaskQueue } from "@gadgets/gatekeeper-kit/serial-queue"; -import type { - ApprovalQueue, HookController, HookInitiator, HookTargetMetadata, -} from "@gadgets/workshop-shared/gatekeeper"; +import type { HookController, HookInitiator, HookTargetMetadata } from "@gadgets/workshop-shared/gatekeeper"; import { GitHubApiError, isConnectionExpired, withAccountApi, type GitHubApi, type GitHubRepoWebhookResponse, type GitHubWebhookDeliveryResponse, type PinnedRepo, @@ -47,14 +44,7 @@ export type GitHubEventHookTarget = RpcTarget & GitHubEventHook; export type GitHubHookParams = { number?: number }; /** What a hook's delivery stub reaches: the connection's facet, narrowed to delivering. */ -export interface GitHubHookDelivery extends RpcTarget { - /** - * Deliver `event` to one firing of the hook, as `HookInitiator.startHook()` returned it, if the - * binding admits it; otherwise return without calling it. - */ - deliver(callback: RpcStub, approvalQueue: RpcStub, - event: GitHubWebhookEvent): Promise; -} +export type GitHubHookDelivery = HookDelivery; /** Everything a hook needs once enabled, captured when the facet binds it. */ export type GitHubHookProps = GitHubHookParams & { @@ -122,17 +112,12 @@ const MAX_REDELIVERIES_PER_CHECK = 20; const FINAL_REFUSALS = new Set([400, 404, 413]); type Registration = Omit; -type Capabilities = { - delivery: RpcStub; - initiator: Fetcher>; -}; +type Capabilities = HookCapabilities; /** A webhook this driver added to a repository. */ type Webhook = { id: number; repo: PinnedRepo }; /** A webhook this driver asked GitHub to add or change at `url`, whose id it may never have learned. */ type Unrecorded = { repo: PinnedRepo; url: string }; -const registrationKey = (key: string) => `reg:${key}`; -const capabilitiesKey = (key: string) => `caps:${key}`; const webhookKey = (repoId: number) => `webhook:${repoId}`; const unrecordedKey = (repoId: number, url: string) => `unrecorded:${repoId}:${url}`; const repairedKey = (repoId: number) => `repaired:${repoId}`; @@ -164,7 +149,7 @@ export class GitHubHookDriver extends DurableObject { * it. */ #viewerReads = new SerialTaskQueue(); - #queue = new HookDeliveryQueue(this.ctx.storage.kv, () => { + #hooks = new HookRegistry(this.ctx.storage.kv, () => { logger.warn("dropped a GitHub event after repeated delivery failures", { event: "hooks.delivery.dropped" }); }); @@ -178,24 +163,17 @@ export class GitHubHookDriver extends DurableObject { // Disconnected while an earlier change ran. if (kv.get("revoked")) throw new Error(DISCONNECTED); await this.#readViewer(userObjectId); - const others = this.#registrations(repo).filter(([regKey]) => regKey !== registrationKey(key)); + const others = this.#registrations(repo).filter(([other]) => other !== key); await this.#ensureWebhook(repo, [registration, ...others.map(([, other]) => other)]); // Recorded in the same change, so a removal queued after it sees this hook. - const replaced = kv.get(capabilitiesKey(key)); - kv.put(registrationKey(key), registration); - kv.put(capabilitiesKey(key), capabilities); - disposeStubs(replaced); + this.ctx.storage.transactionSync(() => this.#hooks.set(key, registration, capabilities)); }); await this.#reschedule(); } async unregister(key: string): Promise { const kv = this.ctx.storage.kv; - const registration = kv.get(registrationKey(key)); - disposeStubs(kv.get(capabilitiesKey(key))); - kv.delete(registrationKey(key)); - kv.delete(capabilitiesKey(key)); - this.#queue.cancel(key); + const registration = this.#hooks.remove(key); if (!registration) return; const { repo } = registration; await this.#webhookChanges.run(async () => { @@ -314,9 +292,9 @@ export class GitHubHookDriver extends DurableObject { // Read before any hook is registered (see register()), and kept until the account disconnects. const viewerId = kv.get("viewer")!; let queued = false; - for (const [regKey, registration] of this.#registrations({ id: repoId })) { + for (const [key, registration] of this.#registrations({ id: repoId })) { if (!watches(registration, event, senderId, viewerId)) continue; - this.#queue.enqueue(regKey.slice("reg:".length), id, event, now); + this.#hooks.enqueue(key, id, event, now); queued = true; } if (queued) await this.#wakeBy(now); @@ -331,26 +309,10 @@ export class GitHubHookDriver extends DurableObject { */ async alarm(): Promise { if ((this.ctx.storage.kv.get("checkAt") ?? Infinity) <= Date.now()) await this.#checkWebhooks(); - await this.#queue.run(Date.now(), (hookKey, event) => this.#deliver(hookKey, event)); + await this.#hooks.run(Date.now()); await this.#reschedule(); } - async #deliver(hookKey: string, event: GitHubWebhookEvent): Promise { - // An unregistered hook's rows are finished, and it gets no new ones. - const capabilities = this.ctx.storage.kv.get(capabilitiesKey(hookKey)); - if (!capabilities) return; - try { - // A refused firing is retried like a failed one, being indistinguishable from a transient - // failure; disabling or deleting the hook unregisters it, which ends the retries. Awaited, - // not pipelined: workerd can't pass a property of a pending result into another call. - using hook = await capabilities.initiator.startHook(); - // @ts-expect-error Worker RPC's mapped types can't relate an ApprovalQueue stub to itself. - await capabilities.delivery.deliver(hook.callback, hook.approvalQueue, event); - } finally { - disposeStubs(capabilities); - } - } - /** * Check each webhook on GitHub (see #checkWebhook), then ask GitHub to redeliver what it failed to * deliver to them, sharing one budget among them so that a busy repository can't starve the rest. @@ -588,10 +550,9 @@ export class GitHubHookDriver extends DurableObject { return [...this.ctx.storage.kv.list({ prefix: "unrecorded:" })].map(([, unrecorded]) => unrecorded); } - /** The enabled hooks on `repo`, by their storage key. */ + /** The enabled hooks on `repo`, by key. */ #registrations(repo: Pick): [string, Registration][] { - return [...this.ctx.storage.kv.list({ prefix: "reg:" })] - .filter(([, registration]) => registration.repo.id === repo.id); + return this.#hooks.entries().filter(([, registration]) => registration.repo.id === repo.id); } async #wakeBy(time: number): Promise { @@ -606,7 +567,7 @@ export class GitHubHookDriver extends DurableObject { async #reschedule(): Promise { const kv = this.ctx.storage.kv; const times: number[] = []; - const due = this.#queue.nextDue(); + const due = this.#hooks.nextDue(); if (due !== undefined) times.push(due); if (this.#webhooks().length > 0 || this.#unrecorded().length > 0) { let checkAt = kv.get("checkAt"); diff --git a/packages/gatekeeper-google/src/chat-hooks.ts b/packages/gatekeeper-google/src/chat-hooks.ts index 6675686f7b..4faaa973e4 100644 --- a/packages/gatekeeper-google/src/chat-hooks.ts +++ b/packages/gatekeeper-google/src/chat-hooks.ts @@ -13,11 +13,9 @@ import { DurableObject, RpcTarget, WorkerEntrypoint, type RpcStub } from "cloudflare:workers"; import { validateRpc } from "capnweb-validate"; -import { HookDeliveryQueue, disposeStubs } from "@gadgets/gatekeeper-kit/hook-delivery-queue"; +import { HookRegistry, type HookCapabilities, type HookDelivery } from "@gadgets/gatekeeper-kit/hook-registry"; import { SingleFlight } from "@gadgets/gatekeeper-kit/single-flight"; -import type { - ApprovalQueue, HookController, HookInitiator, HookTargetMetadata, -} from "@gadgets/workshop-shared/gatekeeper"; +import type { HookController, HookInitiator, HookTargetMetadata } from "@gadgets/workshop-shared/gatekeeper"; import { fetchWithAuthRetry, type AccessTokenProvider } from "./auth-retry"; import { ChatApiError, chatApiFailure, type ChatMessageRaw } from "./chat-api"; import type { ChatMessageHook } from "./chat-types"; @@ -37,11 +35,7 @@ export type ChatMessageHookTarget = RpcTarget & ChatMessageHook; export type ChatHookParams = { spaceName: string; threadName?: string }; /** What a hook's delivery stub reaches: the connection's facet, narrowed to delivering. */ -export interface ChatHookDelivery extends RpcTarget { - /** Deliver `message` to one firing of the hook, as `HookInitiator.startHook()` returned it. */ - deliver(callback: RpcStub, approvalQueue: RpcStub, - message: ChatMessageRaw): Promise; -} +export type ChatHookDelivery = HookDelivery; /** Everything a hook needs once enabled, captured when the facet binds it. */ export type ChatHookProps = ChatHookParams & { @@ -89,14 +83,9 @@ const RENEW_RETRY_MS = 15 * MINUTE_MS; const TRACKING_RECHECK_MS = 24 * HOUR_MS; type Registration = Omit; -type Capabilities = { - delivery: RpcStub; - initiator: Fetcher>; -}; +type Capabilities = HookCapabilities; type Subscription = { name: string; expireTime: number; renewAt: number }; -const registrationKey = (key: string) => `reg:${key}`; -const capabilitiesKey = (key: string) => `caps:${key}`; const subscriptionKey = (authority: string) => `sub:${authority}`; /** When the account last confirmed it knows this driver holds hooks of its (see UserAccount.trackHookDriver()). */ const trackedKey = (userObjectId: string) => `tracked:${userObjectId}`; @@ -108,7 +97,7 @@ const DISCONNECTED = "This Google account has been disconnected."; export class ChatHookDriver extends DurableObject { /** Subscription creations in flight, by account, which concurrent enables and renewals join. */ #subscribing = new SingleFlight(); - #queue = new HookDeliveryQueue(this.ctx.storage.kv, () => { + #hooks = new HookRegistry(this.ctx.storage.kv, () => { logger.warn("dropped a Chat message after repeated delivery failures", { event: "chat.hooks.delivery.dropped" }); }); @@ -122,19 +111,13 @@ export class ChatHookDriver extends DurableObject { } // Nothing from here to the registration awaits, so neither a disconnect nor the alarm, which // drops an expired subscription no hook uses, can come between the subscription and it. - const replaced = this.ctx.storage.kv.get(capabilitiesKey(key)); - this.ctx.storage.kv.put(registrationKey(key), registration); - this.ctx.storage.kv.put(capabilitiesKey(key), capabilities); - disposeStubs(replaced); + this.ctx.storage.transactionSync(() => this.#hooks.set(key, registration, capabilities)); // A reused subscription's renewal may have come due while no hook used it. await this.#reschedule(); } async unregister(key: string): Promise { - disposeStubs(this.ctx.storage.kv.get(capabilitiesKey(key))); - this.ctx.storage.kv.delete(registrationKey(key)); - this.ctx.storage.kv.delete(capabilitiesKey(key)); - this.#queue.cancel(key); + this.#hooks.remove(key); } /** @@ -143,10 +126,8 @@ export class ChatHookDriver extends DurableObject { */ async forgetAccount(userObjectId: string): Promise { this.ctx.storage.kv.put(forgottenKey(userObjectId), Date.now()); - // Listed up front, as unregistering deletes from the listing. - const registrations = [...this.#registrations()]; - for (const [regKey, registration] of registrations) { - if (registration.userObjectId === userObjectId) await this.unregister(regKey.slice("reg:".length)); + for (const [key, registration] of this.#hooks.entries()) { + if (registration.userObjectId === userObjectId) await this.unregister(key); } this.ctx.storage.kv.delete(trackedKey(userObjectId)); } @@ -157,7 +138,7 @@ export class ChatHookDriver extends DurableObject { .find(([, subscription]) => subscription.name === subscriptionName)?.[0].slice("sub:".length); if (authority === undefined) return; const now = Date.now(); - for (const [regKey, registration] of this.#registrations()) { + for (const [key, registration] of this.#hooks.entries()) { if (registration.authority !== authority) continue; for (const message of messages) { // Omit app-private messages as every other Chat read does. @@ -169,7 +150,7 @@ export class ChatHookDriver extends DurableObject { if (message.sender?.name === authority && (!registration.includeOwnMessages || message.clientAssignedMessageId !== undefined)) continue; if (registration.threadName !== undefined && message.thread?.name !== registration.threadName) continue; - this.#queue.enqueue(regKey.slice("reg:".length), message.name, message, now); + this.#hooks.enqueue(key, message.name, message, now); } } await this.#wakeBy(now); @@ -186,7 +167,7 @@ export class ChatHookDriver extends DurableObject { async alarm(): Promise { await this.#trackAccounts(); const now = Date.now(); - await this.#queue.run(now, (hookKey, message) => this.#deliver(hookKey, message)); + await this.#hooks.run(now); // Listed up front: a request handled while a renewal awaits may list too, which would // invalidate a live iterator. Each is read again in its turn, as an enable handled meanwhile // may have subscribed again, or come to use it. @@ -195,7 +176,7 @@ export class ChatHookDriver extends DurableObject { const subscription = this.ctx.storage.kv.get(key); if (!subscription) continue; const authority = key.slice("sub:".length); - const registration = [...this.#registrations()].find(([, candidate]) => candidate.authority === authority)?.[1]; + const registration = this.#hooks.entries().find(([, candidate]) => candidate.authority === authority)?.[1]; if (!registration) { // Unused subscriptions simply lapse. if (subscription.expireTime <= now) this.ctx.storage.kv.delete(key); @@ -230,7 +211,7 @@ export class ChatHookDriver extends DurableObject { */ async #trackAccounts(): Promise { const kv = this.ctx.storage.kv; - const due = new Map([...this.#registrations()] + const due = new Map(this.#hooks.entries() .filter(([, { userObjectId }]) => (kv.get(trackedKey(userObjectId)) ?? 0) <= Date.now() - TRACKING_RECHECK_MS) .map(([, { userObjectId, spaceName }]) => [userObjectId, spaceName])); for (const [userObjectId, spaceName] of due) { @@ -246,21 +227,6 @@ export class ChatHookDriver extends DurableObject { } } - async #deliver(hookKey: string, message: ChatMessageRaw): Promise { - // An unregistered hook's rows are finished, and it gets no new ones. - const capabilities = this.ctx.storage.kv.get(capabilitiesKey(hookKey)); - if (!capabilities) return; - try { - // A refused firing is retried like a failed one, being indistinguishable from a transient - // failure; disabling or deleting the hook unregisters it, which ends the retries. - using hook = await capabilities.initiator.startHook(); - // @ts-expect-error Worker RPC's mapped types can't relate an ApprovalQueue stub to itself. - await capabilities.delivery.deliver(hook.callback, hook.approvalQueue, message); - } finally { - disposeStubs(capabilities); - } - } - /** * Create the account's subscription to this space, joining any creation already in flight. * Refuses, and deletes, a subscription Google attributes to anyone else. @@ -294,10 +260,6 @@ export class ChatHookDriver extends DurableObject { }); } - #registrations() { - return this.ctx.storage.kv.list({ prefix: "reg:" }); - } - #tokens(userObjectId: string): AccessTokenProvider { const account = this.#account(userObjectId); return async opts => (await account.getAccessToken(opts)).token; @@ -314,9 +276,9 @@ export class ChatHookDriver extends DurableObject { async #reschedule(): Promise { const times: number[] = []; - const queueDue = this.#queue.nextDue(); + const queueDue = this.#hooks.nextDue(); if (queueDue !== undefined) times.push(queueDue); - const authorities = new Set([...this.#registrations()].map(([, registration]) => registration.authority)); + const authorities = new Set(this.#hooks.entries().map(([, registration]) => registration.authority)); for (const [key, subscription] of this.ctx.storage.kv.list({ prefix: "sub:" })) { times.push(authorities.has(key.slice("sub:".length)) ? subscription.renewAt : subscription.expireTime); } diff --git a/packages/gatekeeper-google/src/gmail-hooks.ts b/packages/gatekeeper-google/src/gmail-hooks.ts index 50d70ffa94..fb23f534ff 100644 --- a/packages/gatekeeper-google/src/gmail-hooks.ts +++ b/packages/gatekeeper-google/src/gmail-hooks.ts @@ -11,11 +11,9 @@ import { DurableObject, RpcTarget, WorkerEntrypoint, type RpcStub } from "cloudflare:workers"; import { validateRpc } from "capnweb-validate"; -import { HookDeliveryQueue, disposeStubs } from "@gadgets/gatekeeper-kit/hook-delivery-queue"; +import { HookRegistry, type HookCapabilities, type HookDelivery } from "@gadgets/gatekeeper-kit/hook-registry"; import { SingleFlight } from "@gadgets/gatekeeper-kit/single-flight"; -import type { - ApprovalQueue, HookController, HookInitiator, HookTargetMetadata, -} from "@gadgets/workshop-shared/gatekeeper"; +import type { HookController, HookInitiator, HookTargetMetadata } from "@gadgets/workshop-shared/gatekeeper"; import { GmailApi, GmailApiError } from "./google-api"; import { obsContext } from "./observability"; import type { PushHooksEnv } from "./pubsub-push"; @@ -36,15 +34,11 @@ export type GmailMessageHookTarget = RpcTarget & GmailMessageHook; */ export type GmailHookParams = { threadId?: string }; -/** What a hook's delivery stub reaches: the connection's facet, narrowed to delivering. */ -export interface GmailHookDelivery extends RpcTarget { - /** - * Deliver message `messageId` to one firing of the hook if it is new mail the hook watches for; - * otherwise return without calling it. - */ - deliver(callback: RpcStub, approvalQueue: RpcStub, - messageId: string): Promise; -} +/** + * What a hook's delivery stub reaches: the connection's facet, narrowed to delivering a message, by + * id, if it is new mail the hook watches for. + */ +export type GmailHookDelivery = HookDelivery; /** Everything a hook needs once enabled, captured when the facet binds it. */ export type GmailHookProps = GmailHookParams & { @@ -104,16 +98,11 @@ type Registration = Omit & { /** The mailbox's history ID when the hook was enabled: only later records reach it. */ since: string; }; -type Capabilities = { - delivery: RpcStub; - initiator: Fetcher>; -}; +type Capabilities = HookCapabilities; type Watch = { expiration: number; renewAt: number }; type Profile = { emailAddress: string; historyId: string }; type AddedMessage = { id: string; threadId: string; labelIds?: string[] }; -const registrationKey = (key: string) => `reg:${key}`; -const capabilitiesKey = (key: string) => `caps:${key}`; /** When the account last confirmed it knows this driver holds hooks of its (see UserAccount.trackHookDriver()). */ const trackedKey = (userObjectId: string) => `tracked:${userObjectId}`; /** Set when a disconnected account's hooks are dropped, so that an enable it was making is refused. */ @@ -139,7 +128,7 @@ class MailboxChangedError extends Error { */ export class GmailHookDriver extends DurableObject { #watching = new SingleFlight(); - #queue = new HookDeliveryQueue(this.ctx.storage.kv, () => { + #hooks = new HookRegistry(this.ctx.storage.kv, () => { logger.warn("dropped a Gmail message after repeated delivery failures", { event: "gmail.hooks.delivery.dropped" }); }); @@ -162,19 +151,14 @@ export class GmailHookDriver extends DurableObject { } else if (kv.get("syncAt") === undefined) { kv.put("syncAt", Date.now() + SAFETY_SYNC_INTERVAL_MS); } - const replaced = kv.get(capabilitiesKey(key)); - kv.put(registrationKey(key), { ...registration, since: profile.historyId }); - kv.put(capabilitiesKey(key), capabilities); - disposeStubs(replaced); + this.ctx.storage.transactionSync(() => + this.#hooks.set(key, { ...registration, since: profile.historyId }, capabilities)); await this.#reschedule(); } async unregister(key: string): Promise { const kv = this.ctx.storage.kv; - disposeStubs(kv.get(capabilitiesKey(key))); - kv.delete(registrationKey(key)); - kv.delete(capabilitiesKey(key)); - this.#queue.cancel(key); + this.#hooks.remove(key); // The watch is left to lapse rather than stopped: stopping would race a concurrent enable's // fresh watch, and needs credentials a removed connection may no longer have. if (this.#registrations().length === 0) { @@ -186,8 +170,8 @@ export class GmailHookDriver extends DurableObject { /** Drop the hooks of an account being disconnected (see UserAccount.revoke()). */ async forgetAccount(userObjectId: string): Promise { this.ctx.storage.kv.put(forgottenKey(userObjectId), Date.now()); - for (const [regKey, registration] of this.#registrations()) { - if (registration.userObjectId === userObjectId) await this.unregister(regKey.slice("reg:".length)); + for (const [key, registration] of this.#registrations()) { + if (registration.userObjectId === userObjectId) await this.unregister(key); } this.ctx.storage.kv.delete(trackedKey(userObjectId)); } @@ -234,7 +218,7 @@ export class GmailHookDriver extends DurableObject { } // After the sync, so the rows it just queued are due in this same run. - await this.#queue.run(Date.now(), (hookKey, messageId) => this.#deliver(hookKey, messageId)); + await this.#hooks.run(Date.now()); await this.#reschedule(); } @@ -321,26 +305,11 @@ export class GmailHookDriver extends DurableObject { // The facet re-checks authoritatively. This prefilter exists because every delivery attempt // starts a hook firing in its workspace, so unfiltered spam would wake every subscriber. if (labelIds?.some(label => UNDELIVERED_GMAIL_LABELS.includes(label))) return; - for (const [regKey, registration] of registrations) { + for (const [key, registration] of registrations) { if (labelIds && registration.labelId !== undefined && !labelIds.includes(registration.labelId)) continue; if (BigInt(recordId) <= BigInt(registration.since)) continue; if (registration.threadId !== undefined && registration.threadId !== threadId) continue; - this.#queue.enqueue(regKey.slice("reg:".length), id, id, Date.now()); - } - } - - async #deliver(hookKey: string, messageId: string): Promise { - // An unregistered hook's rows are finished, and it gets no new ones. - const capabilities = this.ctx.storage.kv.get(capabilitiesKey(hookKey)); - if (!capabilities) return; - try { - // A refused firing is retried like a failed one, being indistinguishable from a transient - // failure; disabling or deleting the hook unregisters it, which ends the retries. - using hook = await capabilities.initiator.startHook(); - // @ts-expect-error Worker RPC's mapped types can't relate an ApprovalQueue stub to itself. - await capabilities.delivery.deliver(hook.callback, hook.approvalQueue, messageId); - } finally { - disposeStubs(capabilities); + this.#hooks.enqueue(key, id, id, Date.now()); } } @@ -436,7 +405,7 @@ export class GmailHookDriver extends DurableObject { } #registrations(): [string, Registration][] { - return [...this.ctx.storage.kv.list({ prefix: "reg:" })]; + return this.#hooks.entries(); } async #wakeBy(time: number): Promise { @@ -447,7 +416,7 @@ export class GmailHookDriver extends DurableObject { async #reschedule(): Promise { const kv = this.ctx.storage.kv; const times: number[] = []; - const queueDue = this.#queue.nextDue(); + const queueDue = this.#hooks.nextDue(); if (queueDue !== undefined) times.push(queueDue); const watch = kv.get("watch"); if (this.#registrations().length > 0) { diff --git a/packages/gatekeeper-microsoft/__tests__/outlook.test.ts b/packages/gatekeeper-microsoft/__tests__/outlook.test.ts index 7ec9b5ce88..e69b61cc02 100644 --- a/packages/gatekeeper-microsoft/__tests__/outlook.test.ts +++ b/packages/gatekeeper-microsoft/__tests__/outlook.test.ts @@ -753,7 +753,10 @@ describe("reading new mail by cursor", () => { }; const ctx = { id: { name: `${USER}.${MAILBOX}` }, - storage: { kv, getAlarm: async () => alarm, setAlarm: async (at: number) => { alarm = at; }, deleteAlarm: async () => { alarm = null; } }, + storage: { + kv, transactionSync: (callback: () => T) => callback(), + getAlarm: async () => alarm, setAlarm: async (at: number) => { alarm = at; }, deleteAlarm: async () => { alarm = null; }, + }, exports: { UserAccount: { idFromString: (id: string) => id, get: () => account } }, }; const delivered: string[] = []; diff --git a/packages/gatekeeper-microsoft/src/outlook-hooks.ts b/packages/gatekeeper-microsoft/src/outlook-hooks.ts index 746ac277a9..1c6c52aff9 100644 --- a/packages/gatekeeper-microsoft/src/outlook-hooks.ts +++ b/packages/gatekeeper-microsoft/src/outlook-hooks.ts @@ -13,7 +13,7 @@ import { DurableObject, RpcTarget, WorkerEntrypoint, type RpcStub } from "cloudflare:workers"; import { validateRpc } from "capnweb-validate"; import { sanitizeTitle } from "@gadgets/gatekeeper-kit/action-description"; -import { HookDeliveryQueue, disposeStubs } from "@gadgets/gatekeeper-kit/hook-delivery-queue"; +import { HookRegistry, type HookCapabilities, type HookDelivery } from "@gadgets/gatekeeper-kit/hook-registry"; import { escapeObservationValue } from "@gadgets/gatekeeper-kit/observers"; import { ResponseTooLargeError, readTextCapped } from "@gadgets/gatekeeper-kit/response-body"; import type { ApprovalQueue, HookController, HookInitiator, HookTargetMetadata } from "@gadgets/workshop-shared/gatekeeper"; @@ -49,9 +49,7 @@ export type OutlookHookTarget = RpcTarget & NewMessageHook; export type OutlookHookParams = Record; /** What a hook's delivery stub reaches: the binding, narrowed to delivering. */ -export interface OutlookHookDelivery extends RpcTarget { - deliver(callback: RpcStub, approvalQueue: RpcStub, messageId: string): Promise; -} +export type OutlookHookDelivery = HookDelivery; /** Everything a hook needs once enabled, captured when the binding subscribes it. */ export type OutlookHookProps = { @@ -149,7 +147,7 @@ export class OutlookHookController extends WorkerEntrypoint; initiator: Fetcher> }; +type Capabilities = HookCapabilities; /** * A folder being watched: its subscription, if Graph would take one, and how far it has been read. * `continueAt` is when the newest message a scan cut short by `MAX_PAGES_PER_SYNC` had read was @@ -157,8 +155,6 @@ type Capabilities = { delivery: RpcStub; initiator: Fetcher */ type Watch = { subscriptionId?: string; expiresAt?: number; renewAt: number; cursor: string; continueAt?: string }; -const registrationKey = (key: string) => `reg:${key}`; -const capabilitiesKey = (key: string) => `caps:${key}`; const watchKey = (folderId: string) => `watch:${folderId}`; /** The watched folders, read up front: the loops over them write and await as they go. */ const watches = (kv: SyncKvStorage) => Array.from(kv.list({ prefix: "watch:" })); @@ -172,7 +168,7 @@ const SYNC_AT_KEY = "syncAt"; * they depend on. */ export class OutlookHookDriver extends DurableObject { - #queue = new HookDeliveryQueue(this.ctx.storage.kv, () => { + #hooks = new HookRegistry(this.ctx.storage.kv, () => { logger.warn("dropped an Outlook message after repeated delivery failures", { event: "outlook.hooks.delivery.dropped" }); }); @@ -185,27 +181,20 @@ export class OutlookHookDriver extends DurableObject { if (kv.get(watchKey(registration.folderId)) === undefined) { kv.put(watchKey(registration.folderId), { renewAt: Date.now(), cursor: now }); } - const replaced = kv.get(capabilitiesKey(key)); - kv.put(registrationKey(key), { ...registration, since: now }); - kv.put(capabilitiesKey(key), capabilities); - disposeStubs(replaced); + this.ctx.storage.transactionSync(() => this.#hooks.set(key, { ...registration, since: now }, capabilities)); // Subscribe now, so the approval that enabled the hook learns if Graph can't be reached. await this.#ensureSubscriptions(); await this.#reschedule(); } async unregister(key: string): Promise { - const kv = this.ctx.storage.kv; - disposeStubs(kv.get(capabilitiesKey(key))); - kv.delete(registrationKey(key)); - kv.delete(capabilitiesKey(key)); - this.#queue.cancel(key); + this.#hooks.remove(key); await this.#dropUnwatched(); } /** Drops every hook here: the account is being disconnected. */ async forgetAccount(): Promise { - for (const [regKey] of this.#registrations()) await this.unregister(regKey.slice("reg:".length)); + for (const [key] of this.#registrations()) await this.unregister(key); } /** A change notification arrived. Only a read follows, and only for the right secret. */ @@ -228,7 +217,7 @@ export class OutlookHookDriver extends DurableObject { await this.#ensureSubscriptions(); if ((kv.get(SYNC_AT_KEY) ?? 0) <= Date.now()) await this.#sync(); } - await this.#queue.run(Date.now(), (hookKey, messageId) => this.#deliver(hookKey, messageId)); + await this.#hooks.run(Date.now()); await this.#reschedule(); } @@ -265,9 +254,9 @@ export class OutlookHookDriver extends DurableObject { const registrations = this.#registrations().filter(([, registration]) => registration.folderId === folderId); for (const row of page.value ?? []) { if (typeof row.id !== "string" || typeof row.receivedDateTime !== "string" || row.isDraft === true) continue; - for (const [regKey, registration] of registrations) { + for (const [hookKey, registration] of registrations) { if (Date.parse(row.receivedDateTime) < Date.parse(registration.since)) continue; - this.#queue.enqueue(regKey.slice("reg:".length), row.id, row.id, Date.now()); + this.#hooks.enqueue(hookKey, row.id, row.id, Date.now()); } if (Date.parse(row.receivedDateTime) > Date.parse(newest)) newest = row.receivedDateTime; } @@ -348,18 +337,6 @@ export class OutlookHookDriver extends DurableObject { } } - async #deliver(hookKey: string, messageId: string): Promise { - const capabilities = this.ctx.storage.kv.get(capabilitiesKey(hookKey)); - if (!capabilities) return; - try { - using hook = await capabilities.initiator.startHook(); - // @ts-expect-error Worker RPC's mapped types can't relate an ApprovalQueue stub to itself. - await capabilities.delivery.deliver(hook.callback, hook.approvalQueue, messageId); - } finally { - disposeStubs(capabilities); - } - } - #pushing(): boolean { return [...this.ctx.storage.kv.list({ prefix: "watch:" })].every(([, watch]) => watch.subscriptionId !== undefined); } @@ -383,7 +360,7 @@ export class OutlookHookDriver extends DurableObject { } #registrations(): [string, Registration][] { - return [...this.ctx.storage.kv.list({ prefix: "reg:" })]; + return this.#hooks.entries(); } #account(userObjectId: string) { @@ -393,7 +370,7 @@ export class OutlookHookDriver extends DurableObject { async #reschedule(): Promise { const kv = this.ctx.storage.kv; const times: number[] = []; - const queueDue = this.#queue.nextDue(); + const queueDue = this.#hooks.nextDue(); if (queueDue !== undefined) times.push(queueDue); if (this.#registrations().length > 0) { times.push(kv.get(SYNC_AT_KEY) ?? Date.now()); diff --git a/plans/gatekeeper-kit.md b/plans/gatekeeper-kit.md index bfd3c6cc16..b7aea733cd 100644 --- a/plans/gatekeeper-kit.md +++ b/plans/gatekeeper-kit.md @@ -2437,12 +2437,14 @@ export class HookRegistry { } ``` -The drivers that queue provider pushes (GitHub, Gmail, Chat, Outlook) each keep the same three +The drivers that queue provider pushes (GitHub, Gmail, Chat, Outlook) each kept the same three things by hand: a `reg:{key}` record the driver filters events by, a `caps:{key}` record holding the hook's stubs, and one delivery — start a firing through the `HookInitiator`, hand it with the message to the binding's delivery stub, dispose what was read. `HookRegistry` is that copy made once, over the `HookDeliveryQueue` (`./hook-delivery-queue`: deduplication, backoff, bounded runs) -the same drivers already share. The keys are the ones they wrote, so a port migrates nothing. +the same drivers already shared. The keys are the ones they wrote, so the ports migrated nothing. +Each now aliases its delivery interface to `HookDelivery`, calls `set()` inside +`ctx.storage.transactionSync()`, and keeps its own alarm. - **Synchronous but for `run()`.** Gmail and Chat rely on nothing awaiting between their provider work and the registration, GitHub writes it inside a serialized webhook change, and Outlook @@ -2480,7 +2482,8 @@ Rejected: differently around `run()` (sync before, check before, renew after). - **A kit controller.** About fifteen lines each, and each projects its own registration and looks up its own driver. -- **`clear()`**, with one caller (GitHub's disconnect, which can remove its hooks one at a time), and +- **`clear()`**, with one caller (GitHub's disconnect, which sweeps its storage and disposes each + `caps:` record as it goes), and **`offer(id, message, admits)`**, which does not fit Gmail's and Outlook's cursor syncs. - **Account-disconnect tracking.** Google, Microsoft and GitHub have three different contracts.