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.