Inefficient Octokit instantiation per PR poll

0import { Octokit } from "@octokit/rest";
1import { and, eq, isNull, lt, or, sql } from "drizzle-orm";
2import { db } from "@/db/index";
3import { outgoingPrs, type OutgoingPr } from "@/db/schema";
4import { logError, logInfo, logWarn } from "./log";
5
6// Cross-repo receipts — Slice B of Phase-2 receipt density work.
7//
8// AntFleet's GitHub App is not installed on third-party repos where
9// antfleet-ops opens upstream PRs. The standard sweep (finding_status →
10// reviews) is therefore blind to those PRs' merge state. This module
11// is the parallel data path: rows in outgoing_prs are seeded by an
12// operator script when an upstream PR is opened, then this poll loop
13// runs on the cron tick to transition status open → merged | closed.
14// Merged rows surface on /receipts as cross-repo receipts; declined
15// rows are logged only (honest-report: a closed-without-merge PR is
16// not a receipt).
17//
18// The poll rate limit (one hour per row) is two things at once: a
19// politeness budget toward the upstream owner's GitHub rate limit, and
20// a guard on antfleet-ops's PAT which is shared with manual `gh` work.
21
22const POLL_INTERVAL_MS = 60 * 60 * 1000;
23
24export type OpenOutgoingPr = {
25  id: string;
26  upstreamOwner: string;
27  upstreamRepo: string;
28  upstreamPrNumber: number;
29};
30
31export type UpstreamPrState = {
32  // True only when GitHub returned a non-null merged_at. A PR can be
33  // closed without merging — in which case merged_at is null and
34  // state === "closed".
35  merged: boolean;
36  mergedAt: Date | null;
37  mergeSha: string | null;
38  // The raw GitHub `state` field: "open" or "closed".
39  state: "open" | "closed";
40};
41
42export type PollOutgoingDeps = {
43  loadOpenPrs: (now: Date) => Promise<OpenOutgoingPr[]>;
44  getUpstreamPrState: (args: {
45    owner: string;
46    repo: string;
47    number: number;
48  }) => Promise<UpstreamPrState>;
49  markMerged: (args: {
50    id: string;
51    mergedAt: Date;
52    mergeSha: string;
53    polledAt: Date;
54  }) => Promise<void>;
55  markClosed: (args: { id: string; polledAt: Date }) => Promise<void>;
56  stampPolled: (args: { id: string; polledAt: Date }) => Promise<void>;
57};
58
59export type PollOutgoingResult = {
60  attempted: number;
61  merged: number;
62  closed: number;
63  unchanged: number;
64  errors: number;
65};
66
67export async function pollOutgoingPrs(
68  deps: PollOutgoingDeps,
69  now: Date = new Date(),
70): Promise<PollOutgoingResult> {
71  const open = await deps.loadOpenPrs(now);
72  let merged = 0;
73  let closed = 0;
74  let unchanged = 0;
75  let errors = 0;
76  for (const pr of open) {
77    try {
78      const state = await deps.getUpstreamPrState({
79        owner: pr.upstreamOwner,
80        repo: pr.upstreamRepo,
81        number: pr.upstreamPrNumber,
82      });
83      if (state.merged && state.mergedAt !== null && state.mergeSha !== null) {
84        await deps.markMerged({
85          id: pr.id,
86          mergedAt: state.mergedAt,
87          mergeSha: state.mergeSha,
88          polledAt: now,
89        });
90        merged += 1;
91        logInfo("outgoing_prs.merged", {
92          id: pr.id,
93          upstream: `${pr.upstreamOwner}/${pr.upstreamRepo}#${pr.upstreamPrNumber}`,
94          mergeSha: state.mergeSha,
95        });
96      } else if (state.state === "closed") {
97        await deps.markClosed({ id: pr.id, polledAt: now });
98        closed += 1;
99        logInfo("outgoing_prs.declined", {
100          id: pr.id,
101          upstream: `${pr.upstreamOwner}/${pr.upstreamRepo}#${pr.upstreamPrNumber}`,
102        });
103      } else {
104        await deps.stampPolled({ id: pr.id, polledAt: now });
105        unchanged += 1;
106      }
107    } catch (err) {
108      const message = err instanceof Error ? err.message : String(err);
109      logWarn("outgoing_prs.poll_failed", {
110        id: pr.id,
111        upstream: `${pr.upstreamOwner}/${pr.upstreamRepo}#${pr.upstreamPrNumber}`,
112        message,
113      });
114      errors += 1;
115    }
116  }
117  return { attempted: open.length, merged, closed, unchanged, errors };
118}
119
120// ─── Real-deps wiring (used by cron sweep) ─────────────────────────────────
121
122function getAntfleetOpsToken(): string {
123  const token = process.env["ANTFLEET_OPS_GH_TOKEN"];
124  if (token === undefined || token.length === 0) {
125    throw new Error("ANTFLEET_OPS_GH_TOKEN is required to poll outgoing PRs");
126  }
127  return token;
128}
129
130export function realPollDeps(): PollOutgoingDeps {
131  // Lazy Octokit construction so the cron route only instantiates when
132  // there's actually work to do. Token is read each call so a rotated
133  // PAT picks up without a deploy.
134  const tokenReader = () => new Octokit({ auth: getAntfleetOpsToken() });
135  return {
136    loadOpenPrs: async (now: Date) => {
137      const threshold = new Date(now.getTime() - POLL_INTERVAL_MS);
138      const rows = await db
139        .select({
140          id: outgoingPrs.id,
141          upstreamOwner: outgoingPrs.upstreamOwner,
142          upstreamRepo: outgoingPrs.upstreamRepo,
143          upstreamPrNumber: outgoingPrs.upstreamPrNumber,
144        })
145        .from(outgoingPrs)
146        .where(
147          and(
148            eq(outgoingPrs.status, "open"),
149            or(
150              isNull(outgoingPrs.lastPolledAt),
151              lt(outgoingPrs.lastPolledAt, threshold),
152            ),
153          ),
154        );
155      return rows;
156    },
157    getUpstreamPrState: async ({ owner, repo, number }) => {
158      const octokit = tokenReader();
159      const resp = await octokit.rest.pulls.get({
160        owner,
161        repo,
162        pull_number: number,
163      });
164      const data = resp.data;
165      const mergedAt = data.merged_at !== null ? new Date(data.merged_at) : null;
166      const merged = data.merged === true && mergedAt !== null;
167      // GitHub's `state` for pulls is "open" | "closed"; we never
168      // expect anything else but tighten via a fallback to "open".
169      const state: "open" | "closed" = data.state === "closed" ? "closed" : "open";
170      return {
171        merged,
172        mergedAt,
173        mergeSha: data.merge_commit_sha ?? null,
174        state,
175      };
176    },
177    markMerged: async ({ id, mergedAt, mergeSha, polledAt }) => {
178      await db
179        .update(outgoingPrs)
180        .set({
181          status: "merged",
182          mergedAt,
183          mergeSha,
184          lastPolledAt: polledAt,
185        })
186        .where(eq(outgoingPrs.id, id));
187    },
188    markClosed: async ({ id, polledAt }) => {
189      await db
190        .update(outgoingPrs)
191        .set({ status: "closed", lastPolledAt: polledAt })
192        .where(eq(outgoingPrs.id, id));
193    },
194    stampPolled: async ({ id, polledAt }) => {
195      await db
196        .update(outgoingPrs)
197        .set({ lastPolledAt: polledAt })
198        .where(eq(outgoingPrs.id, id));
199    },
200  };
201}
202
203// ─── Display layer — used by /receipts page ────────────────────────────────
204
205export type CrossRepoReceiptRow = {
206  id: string;
207  sourceFindingId: string;
208  upstreamOwner: string;
209  upstreamRepo: string;
210  upstreamPrNumber: number;
211  mergedAt: Date;
212  mergeSha: string;
213  prUrl: string;
214};
215
216export type CrossRepoReceiptsPage = {
217  total: number;
218  recent: CrossRepoReceiptRow[];
219  lastMergedAt: Date | null;
220};
221
222export async function loadCrossRepoReceipts(
223  limit: number,
224): Promise<CrossRepoReceiptsPage> {
225  const rows = await db
226    .select()
227    .from(outgoingPrs)
228    .where(eq(outgoingPrs.status, "merged"))
229    .orderBy(sql`${outgoingPrs.mergedAt} DESC NULLS LAST`)
230    .limit(limit);
231
232  const recent: CrossRepoReceiptRow[] = [];
233  let lastMergedAt: Date | null = null;
234  for (const r of rows as OutgoingPr[]) {
235    if (r.mergedAt === null || r.mergeSha === null) continue;
236    if (lastMergedAt === null || r.mergedAt > lastMergedAt) {
237      lastMergedAt = r.mergedAt;
238    }
239    recent.push({
240      id: r.id,
241      sourceFindingId: r.sourceFindingId,
242      upstreamOwner: r.upstreamOwner,
243      upstreamRepo: r.upstreamRepo,
244      upstreamPrNumber: r.upstreamPrNumber,
245      mergedAt: r.mergedAt,
246      mergeSha: r.mergeSha,
247      prUrl: upstreamPrUrl(r.upstreamOwner, r.upstreamRepo, r.upstreamPrNumber),
248    });
249  }
250
251  const [{ value }] = await db
252    .select({ value: sql<number>`count(*)::int`.as("value") })
253    .from(outgoingPrs)
254    .where(eq(outgoingPrs.status, "merged"));
255
256  return { total: value ?? 0, recent, lastMergedAt };
257}
258
259export function upstreamPrUrl(owner: string, repo: string, number: number): string {
260  return `https://github.com/${owner}/${repo}/pull/${number}`;
261}
262
263// Wrapping for the cron tick — never throw out of the poll loop. The
264// cron handler treats sweep success as the load-bearing outcome; an
265// outgoing-PR poll failure logs and continues.
266export async function runOutgoingPrsPoll(): Promise<PollOutgoingResult | null> {
267  try {
268    const deps = realPollDeps();
269    return await pollOutgoingPrs(deps);
270  } catch (err) {
271    const message = err instanceof Error ? err.message : String(err);
272    logError("outgoing_prs.poll_orchestrator_failed", { message });
273    return null;
274  }
275}
276

0import { Octokit } from "@octokit/rest";
1import { and, eq, isNull, lt, or } from "drizzle-orm";
2import { db } from "@/db/index";
3import { outgoingPrs } from "@/db/schema";
4import { logError, logInfo, logWarn } from "./log";
5
6// Cross-repo receipts — Slice B of Phase-2 receipt density work.
7//
8// AntFleet's GitHub App is not installed on third-party repos where
9// antfleet-ops opens upstream PRs. The standard sweep (finding_status →
10// reviews) is therefore blind to those PRs' merge state. This module
11// is the parallel data path: rows in outgoing_prs are seeded by an
12// operator script when an upstream PR is opened, then this poll loop
13// runs on the cron tick to transition status open → merged | closed.
14// Merged rows surface on /receipts as cross-repo receipts; declined
15// rows are logged only (honest-report: a closed-without-merge PR is
16// not a receipt).
17//
18// The poll rate limit (one hour per row) is two things at once: a
19// politeness budget toward the upstream owner's GitHub rate limit, and
20// a guard on antfleet-ops's PAT which is shared with manual `gh` work.
21
22const POLL_INTERVAL_MS = 60 * 60 * 1000;
23
24export type OpenOutgoingPr = {
25  id: string;
26  upstreamOwner: string;
27  upstreamRepo: string;
28  upstreamPrNumber: number;
29};
30
31export type UpstreamPrState = {
32  // True only when GitHub returned a non-null merged_at. A PR can be
33  // closed without merging — in which case merged_at is null and
34  // state === "closed".
35  merged: boolean;
36  mergedAt: Date | null;
37  mergeSha: string | null;
38  // The raw GitHub `state` field: "open" or "closed".
39  state: "open" | "closed";
40};
41
42export type PollOutgoingDeps = {
43  loadOpenPrs: (now: Date) => Promise<OpenOutgoingPr[]>;
44  getUpstreamPrState: (args: {
45    owner: string;
46    repo: string;
47    number: number;
48  }) => Promise<UpstreamPrState>;
49  markMerged: (args: {
50    id: string;
51    mergedAt: Date;
52    mergeSha: string;
53    polledAt: Date;
54  }) => Promise<void>;
55  markClosed: (args: { id: string; polledAt: Date }) => Promise<void>;
56  stampPolled: (args: { id: string; polledAt: Date }) => Promise<void>;
57};
58
59export type PollOutgoingResult = {
60  attempted: number;
61  merged: number;
62  closed: number;
63  unchanged: number;
64  errors: number;
65};
66
67export async function pollOutgoingPrs(
68  deps: PollOutgoingDeps,
69  now: Date = new Date(),
70): Promise<PollOutgoingResult> {
71  const open = await deps.loadOpenPrs(now);
72  let merged = 0;
73  let closed = 0;
74  let unchanged = 0;
75  let errors = 0;
76  for (const pr of open) {
77    try {
78      const state = await deps.getUpstreamPrState({
79        owner: pr.upstreamOwner,
80        repo: pr.upstreamRepo,
81        number: pr.upstreamPrNumber,
82      });
83      if (state.merged && state.mergedAt !== null && state.mergeSha !== null) {
84        await deps.markMerged({
85          id: pr.id,
86          mergedAt: state.mergedAt,
87          mergeSha: state.mergeSha,
88          polledAt: now,
89        });
90        merged += 1;
91        logInfo("outgoing_prs.merged", {
92          id: pr.id,
93          upstream: `${pr.upstreamOwner}/${pr.upstreamRepo}#${pr.upstreamPrNumber}`,
94          mergeSha: state.mergeSha,
95        });
96      } else if (state.merged) {
97        // GitHub returned merged=true but merge_commit_sha is still null
98        // (async commit propagation can lag the API's `merged` flag by
99        // seconds-to-minutes). Treat as still-open and re-poll next tick
100        // rather than mis-marking as declined — a merged PR is the
101        // receipt-eligible class and we don't want to lose it.
102        await deps.stampPolled({ id: pr.id, polledAt: now });
103        unchanged += 1;
104        logWarn("outgoing_prs.merged_pending_sha", {
105          id: pr.id,
106          upstream: `${pr.upstreamOwner}/${pr.upstreamRepo}#${pr.upstreamPrNumber}`,
107        });
108      } else if (state.state === "closed") {
109        await deps.markClosed({ id: pr.id, polledAt: now });
110        closed += 1;
111        logInfo("outgoing_prs.declined", {
112          id: pr.id,
113          upstream: `${pr.upstreamOwner}/${pr.upstreamRepo}#${pr.upstreamPrNumber}`,
114        });
115      } else {
116        await deps.stampPolled({ id: pr.id, polledAt: now });
117        unchanged += 1;
118      }
119    } catch (err) {
120      const message = err instanceof Error ? err.message : String(err);
121      logWarn("outgoing_prs.poll_failed", {
122        id: pr.id,
123        upstream: `${pr.upstreamOwner}/${pr.upstreamRepo}#${pr.upstreamPrNumber}`,
124        message,
125      });
126      errors += 1;
127    }
128  }
129  return { attempted: open.length, merged, closed, unchanged, errors };
130}
131
132// ─── Real-deps wiring (used by cron sweep) ─────────────────────────────────
133
134function getAntfleetOpsToken(): string {
135  const token = process.env["ANTFLEET_OPS_GH_TOKEN"];
136  if (token === undefined || token.length === 0) {
137    throw new Error("ANTFLEET_OPS_GH_TOKEN is required to poll outgoing PRs");
138  }
139  return token;
140}
141
142export function realPollDeps(): PollOutgoingDeps {
143  // Lazy Octokit construction so the cron route only instantiates when
144  // there's actually work to do. Token is read each call so a rotated
145  // PAT picks up without a deploy.
146  const tokenReader = () => new Octokit({ auth: getAntfleetOpsToken() });
147  return {
148    loadOpenPrs: async (now: Date) => {
149      const threshold = new Date(now.getTime() - POLL_INTERVAL_MS);
150      const rows = await db
151        .select({
152          id: outgoingPrs.id,
153          upstreamOwner: outgoingPrs.upstreamOwner,
154          upstreamRepo: outgoingPrs.upstreamRepo,
155          upstreamPrNumber: outgoingPrs.upstreamPrNumber,
156        })
157        .from(outgoingPrs)
158        .where(
159          and(
160            eq(outgoingPrs.status, "open"),
161            or(
162              isNull(outgoingPrs.lastPolledAt),
163              lt(outgoingPrs.lastPolledAt, threshold),
164            ),
165          ),
166        );
167      return rows;
168    },
169    getUpstreamPrState: async ({ owner, repo, number }) => {
170      const octokit = tokenReader();
171      const resp = await octokit.rest.pulls.get({
172        owner,
173        repo,
174        pull_number: number,
175      });
176      const data = resp.data;
177      const mergedAt = data.merged_at !== null ? new Date(data.merged_at) : null;
178      const merged = data.merged === true && mergedAt !== null;
179      // GitHub's `state` for pulls is "open" | "closed"; we never
180      // expect anything else but tighten via a fallback to "open".
181      const state: "open" | "closed" = data.state === "closed" ? "closed" : "open";
182      return {
183        merged,
184        mergedAt,
185        mergeSha: data.merge_commit_sha ?? null,
186        state,
187      };
188    },
189    markMerged: async ({ id, mergedAt, mergeSha, polledAt }) => {
190      await db
191        .update(outgoingPrs)
192        .set({
193          status: "merged",
194          mergedAt,
195          mergeSha,
196          lastPolledAt: polledAt,
197        })
198        .where(eq(outgoingPrs.id, id));
199    },
200    markClosed: async ({ id, polledAt }) => {
201      await db
202        .update(outgoingPrs)
203        .set({ status: "closed", lastPolledAt: polledAt })
204        .where(eq(outgoingPrs.id, id));
205    },
206    stampPolled: async ({ id, polledAt }) => {
207      await db
208        .update(outgoingPrs)
209        .set({ lastPolledAt: polledAt })
210        .where(eq(outgoingPrs.id, id));
211    },
212  };
213}
214
215// Display loader + URL helper for cross-repo receipts live in lib/receipts.ts
216// alongside the same-repo loaders. This module owns the write/poll path
217// only; readers import from lib/receipts.ts.
218
219// Wrapping for the cron tick — never throw out of the poll loop. The
220// cron handler treats sweep success as the load-bearing outcome; an
221// outgoing-PR poll failure logs and continues.
222export async function runOutgoingPrsPoll(): Promise<PollOutgoingResult | null> {
223  try {
224    const deps = realPollDeps();
225    return await pollOutgoingPrs(deps);
226  } catch (err) {
227    const message = err instanceof Error ? err.message : String(err);
228    logError("outgoing_prs.poll_orchestrator_failed", { message });
229    return null;
230  }
231}
232