Skip to content

Keep the GitHub, Gmail, Chat and Outlook hooks in the kit's hook registry #1076

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ndisidore merged 1 commit into from
Nov 9, 2026
+72 −197
Merged

Keep the GitHub, Gmail, Chat and Outlook hooks in the kit's hook registry #1076

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 15 additions & 54 deletions packages/gatekeeper-github/src/github-hooks.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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<GitHubEventHookTarget>, approvalQueue: RpcStub<ApprovalQueue>,
event: GitHubWebhookEvent): Promise<void>;
}
export type GitHubHookDelivery = HookDelivery<GitHubEventHookTarget, GitHubWebhookEvent>;

/** Everything a hook needs once enabled, captured when the facet binds it. */
export type GitHubHookProps = GitHubHookParams & {
Expand Down Expand Up @@ -122,17 +112,12 @@ const MAX_REDELIVERIES_PER_CHECK = 20;
const FINAL_REFUSALS = new Set([400, 404, 413]);

type Registration = Omit<GitHubHookProps, "key" | "userObjectId" | "delivery">;
type Capabilities = {
delivery: RpcStub<GitHubHookDelivery>;
initiator: Fetcher<HookInitiator<GitHubEventHookTarget>>;
};
type Capabilities = HookCapabilities<GitHubEventHookTarget, GitHubWebhookEvent>;
/** 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}`;
Expand Down Expand Up @@ -164,7 +149,7 @@ export class GitHubHookDriver extends DurableObject<Env> {
* it.
*/
#viewerReads = new SerialTaskQueue();
#queue = new HookDeliveryQueue<GitHubWebhookEvent>(this.ctx.storage.kv, () => {
#hooks = new HookRegistry<GitHubEventHookTarget, GitHubWebhookEvent, Registration>(this.ctx.storage.kv, () => {
logger.warn("dropped a GitHub event after repeated delivery failures", { event: "hooks.delivery.dropped" });
});

Expand All @@ -178,24 +163,17 @@ export class GitHubHookDriver extends DurableObject<Env> {
// 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<Capabilities>(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<void> {
const kv = this.ctx.storage.kv;
const registration = kv.get<Registration>(registrationKey(key));
disposeStubs(kv.get<Capabilities>(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 () => {
Expand Down Expand Up @@ -314,9 +292,9 @@ export class GitHubHookDriver extends DurableObject<Env> {
// Read before any hook is registered (see register()), and kept until the account disconnects.
const viewerId = kv.get<number>("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);
Expand All @@ -331,26 +309,10 @@ export class GitHubHookDriver extends DurableObject<Env> {
*/
async alarm(): Promise<void> {
if ((this.ctx.storage.kv.get<number>("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<void> {
// An unregistered hook's rows are finished, and it gets no new ones.
const capabilities = this.ctx.storage.kv.get<Capabilities>(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.
Expand Down Expand Up @@ -588,10 +550,9 @@ export class GitHubHookDriver extends DurableObject<Env> {
return [...this.ctx.storage.kv.list<Unrecorded>({ prefix: "unrecorded:" })].map(([, unrecorded]) => unrecorded);
}

/** The enabled hooks on `repo`, by their storage key. */
/** The enabled hooks on `repo`, by key. */
#registrations(repo: Pick<PinnedRepo, "id">): [string, Registration][] {
return [...this.ctx.storage.kv.list<Registration>({ 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<void> {
Expand All @@ -606,7 +567,7 @@ export class GitHubHookDriver extends DurableObject<Env> {
async #reschedule(): Promise<void> {
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<number>("checkAt");
Expand Down
70 changes: 16 additions & 54 deletions packages/gatekeeper-google/src/chat-hooks.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand All @@ -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<ChatMessageHookTarget>, approvalQueue: RpcStub<ApprovalQueue>,
message: ChatMessageRaw): Promise<void>;
}
export type ChatHookDelivery = HookDelivery<ChatMessageHookTarget, ChatMessageRaw>;

/** Everything a hook needs once enabled, captured when the facet binds it. */
export type ChatHookProps = ChatHookParams & {
Expand Down Expand Up @@ -89,14 +83,9 @@ const RENEW_RETRY_MS = 15 * MINUTE_MS;
const TRACKING_RECHECK_MS = 24 * HOUR_MS;

type Registration = Omit<ChatHookProps, "key" | "delivery">;
type Capabilities = {
delivery: RpcStub<ChatHookDelivery>;
initiator: Fetcher<HookInitiator<ChatMessageHookTarget>>;
};
type Capabilities = HookCapabilities<ChatMessageHookTarget, ChatMessageRaw>;
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}`;
Expand All @@ -108,7 +97,7 @@ const DISCONNECTED = "This Google account has been disconnected.";
export class ChatHookDriver extends DurableObject<Env> {
/** Subscription creations in flight, by account, which concurrent enables and renewals join. */
#subscribing = new SingleFlight();
#queue = new HookDeliveryQueue<ChatMessageRaw>(this.ctx.storage.kv, () => {
#hooks = new HookRegistry<ChatMessageHookTarget, ChatMessageRaw, Registration>(this.ctx.storage.kv, () => {
logger.warn("dropped a Chat message after repeated delivery failures", { event: "chat.hooks.delivery.dropped" });
});

Expand All @@ -122,19 +111,13 @@ export class ChatHookDriver extends DurableObject<Env> {
}
// 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<Capabilities>(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<void> {
disposeStubs(this.ctx.storage.kv.get<Capabilities>(capabilitiesKey(key)));
this.ctx.storage.kv.delete(registrationKey(key));
this.ctx.storage.kv.delete(capabilitiesKey(key));
this.#queue.cancel(key);
this.#hooks.remove(key);
}

/**
Expand All @@ -143,10 +126,8 @@ export class ChatHookDriver extends DurableObject<Env> {
*/
async forgetAccount(userObjectId: string): Promise<void> {
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));
}
Expand All @@ -157,7 +138,7 @@ export class ChatHookDriver extends DurableObject<Env> {
.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.
Expand All @@ -169,7 +150,7 @@ export class ChatHookDriver extends DurableObject<Env> {
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);
Expand All @@ -186,7 +167,7 @@ export class ChatHookDriver extends DurableObject<Env> {
async alarm(): Promise<void> {
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.
Expand All @@ -195,7 +176,7 @@ export class ChatHookDriver extends DurableObject<Env> {
const subscription = this.ctx.storage.kv.get<Subscription>(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);
Expand Down Expand Up @@ -230,7 +211,7 @@ export class ChatHookDriver extends DurableObject<Env> {
*/
async #trackAccounts(): Promise<void> {
const kv = this.ctx.storage.kv;
const due = new Map([...this.#registrations()]
const due = new Map(this.#hooks.entries()
.filter(([, { userObjectId }]) => (kv.get<number>(trackedKey(userObjectId)) ?? 0) <= Date.now() - TRACKING_RECHECK_MS)
.map(([, { userObjectId, spaceName }]) => [userObjectId, spaceName]));
for (const [userObjectId, spaceName] of due) {
Expand All @@ -246,21 +227,6 @@ export class ChatHookDriver extends DurableObject<Env> {
}
}

async #deliver(hookKey: string, message: ChatMessageRaw): Promise<void> {
// An unregistered hook's rows are finished, and it gets no new ones.
const capabilities = this.ctx.storage.kv.get<Capabilities>(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.
Expand Down Expand Up @@ -294,10 +260,6 @@ export class ChatHookDriver extends DurableObject<Env> {
});
}

#registrations() {
return this.ctx.storage.kv.list<Registration>({ prefix: "reg:" });
}

#tokens(userObjectId: string): AccessTokenProvider {
const account = this.#account(userObjectId);
return async opts => (await account.getAccessToken(opts)).token;
Expand All @@ -314,9 +276,9 @@ export class ChatHookDriver extends DurableObject<Env> {

async #reschedule(): Promise<void> {
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<Subscription>({ prefix: "sub:" })) {
times.push(authorities.has(key.slice("sub:".length)) ? subscription.renewAt : subscription.expireTime);
}
Expand Down
Loading
Loading