packages/forkhorse-core/src/subscriptions.ts

1import type { Env } from "./types";
2
3export interface RepoEvents {
4  namespace: string;
5  repoName: string;
6  events: "pending" | "active";
7  subscriptionId?: string;
8  attempts: number;
9  retryAt: number;
10}
11const prefix = "repoEvents:";
12const backoff = (attempts: number) => Math.min(30_000 * 2 ** Math.min(attempts, 7), 3_600_000);
13const identifier = (value: unknown): value is string =>
14  typeof value === "string" && /^(?:[a-f\d]{32}|[a-f\d]{8}(?:-[a-f\d]{4}){3}-[a-f\d]{12})$/i.test(value);
15
16/** Keep the registry's membership value unchanged; subscription state lives beside it. */
17export async function registerEvents(
18  storage: DurableObjectTransaction,
19  namespace: string,
20  repoName: string,
21): Promise<void> {
22  const retryAt = Date.now() + backoff(0);
23  await storage.put(`${prefix}${repoName}`, {
24    namespace, repoName, events: "pending", attempts: 0, retryAt,
25  } satisfies RepoEvents);
26  const alarm = await storage.getAlarm();
27  if (alarm === null || alarm > retryAt) await storage.setAlarm(retryAt);
28}
29
30export async function publicEvents(storage: DurableObjectStorage, repoName: string) {
31  const record = await storage.get<RepoEvents>(`${prefix}${repoName}`);
32  return {
33    events: record?.events ?? "pending",
34    ...(record?.subscriptionId && { subscriptionId: record.subscriptionId }),
35  };
36}
37
38interface Subscription {
39  id: string;
40  enabled: boolean;
41  source: { type: string; namespace: string; repo_name: string };
42  destination: { type: string; queue_id: string };
43  events: string[];
44}
45
46/** Matches Wrangler 4.149.0 queues/client.ts and subscription/create.ts.
47 * Queue ID is owner-resolved so this token never needs Queue Read permission.
48 * Only fixed-origin requests; no redirects, provider errors, or bodies escape.
49 */
50export class EventSubscriptions {
51  constructor(private storage: DurableObjectStorage, private env: Env) {}
52
53  private async ensure(record: RepoEvents): Promise<string> {
54    const { CF_ACCOUNT_ID: account, CF_EVENTS_API_TOKEN: token, PUSH_QUEUE_ID: queue } = this.env;
55    if (!identifier(account) || !identifier(queue) || !token)
56      throw new Error("Event subscription configuration is incomplete.");
57    const endpoint = `https://api.cloudflare.com/client/v4/accounts/${account}/event_subscriptions/subscriptions`;
58    const signal = AbortSignal.timeout(10_000);
59    const request = async (url: string, payload?: unknown) => {
60      const response = await fetch(url, {
61        method: payload ? "POST" : "GET",
62        headers: { authorization: `Bearer ${token}`, "content-type": "application/json" },
63        body: payload ? JSON.stringify(payload) : undefined,
64        redirect: "manual",
65        signal,
66      });
67      if (!response.ok) throw new Error("Event subscription request failed.");
68      const envelope = await response.json<{ success: boolean; result: unknown }>();
69      if (envelope.success !== true) throw new Error("Event subscription request failed.");
70      return envelope.result;
71    };
72    // Scan every page before creating. An incomplete listing must never imply absence.
73    for (let page = 1; page <= 100; page++) {
74      const result = await request(`${endpoint}?${new URLSearchParams({ queue_id: queue, page: String(page), per_page: "20" })}`);
75      if (!Array.isArray(result)) throw new Error("Invalid subscription listing.");
76      const existing = result.find((value: Subscription | null) =>
77        value && identifier(value.id) && value.enabled === true &&
78        value.source?.type === "artifacts.repo" &&
79        value.source.namespace === record.namespace && value.source.repo_name === record.repoName &&
80        value.destination?.type === "queues.queue" && value.destination.queue_id === queue &&
81        Array.isArray(value.events) && value.events.includes("pushed"),
82      );
83      if (existing) return existing.id;
84      if (result.length < 20) {
85        const created = await request(endpoint, {
86          name: `${record.namespace}/${record.repoName} pushed`,
87          enabled: true,
88          source: { type: "artifacts.repo", namespace: record.namespace, repo_name: record.repoName },
89          destination: { type: "queues.queue", queue_id: queue },
90          events: ["pushed"],
91        }) as Subscription | null;
92        if (!created || !identifier(created.id)) throw new Error("Invalid subscription result.");
93        return created.id;
94      }
95    }
96    throw new Error("Subscription listing limit reached.");
97  }
98
99  private async schedule(): Promise<void> {
100    const records = await this.storage.list<RepoEvents>({ prefix });
101    const pending = [...records.values()].filter((record) => record.events === "pending");
102    if (pending.length) await this.storage.setAlarm(Math.min(...pending.map((record) => record.retryAt)));
103    else await this.storage.deleteAlarm();
104  }
105
106  async attempt(repoName: string): Promise<void> {
107    const key = `${prefix}${repoName}`;
108    const record = await this.storage.get<RepoEvents>(key);
109    if (!record || record.events === "active") return;
110    // Advance and arm recovery BEFORE external I/O. A crash after POST is recovered by LIST.
111    record.retryAt = Date.now() + backoff(record.attempts);
112    record.attempts++;
113    await this.storage.transaction(async (txn) => {
114      await txn.put(key, record);
115      await this.schedule();
116    });
117    let subscriptionId: string;
118    try {
119      subscriptionId = await this.ensure(record);
120    } catch {
121      // Never log or persist provider bodies/errors (they may echo authorization headers).
122      return;
123    }
124    await this.storage.transaction(async (txn) => {
125      await txn.put(key, { ...record, events: "active", subscriptionId } satisfies RepoEvents);
126      await this.schedule();
127    });
128  }
129
130  async alarm(): Promise<void> {
131    const records = await this.storage.list<RepoEvents>({ prefix });
132    // Bound one alarm's external work; remaining due records retain an immediate alarm.
133    const due = [...records.values()]
134      .filter((record) => record.events === "pending" && record.retryAt <= Date.now())
135      .toSorted((a, b) => a.retryAt - b.retryAt).slice(0, 10);
136    for (const record of due) await this.attempt(record.repoName);
137    await this.schedule();
138  }
139}
140