refactor(hub): delete the worker coordinator, rotation and code review feed

apps/hub/src/bitbucket-datacenter/hook.test.ts

@@ -67,7 +67,6 @@ function harness(overrides: Partial<ProviderHookDeps> = {}) {
     } as never,
     cipher: { async decrypt() { return SECRET; } } as never,
     cache: new DeliveryCache(),
-    codeReview: { async offer() {} },
     coalescer: {
       async accept(input) {
         accepted.push(input);

apps/hub/src/bitbucket-datacenter/integration.ts

@@ -59,7 +59,7 @@ type BitbucketDeps = PortalCredentialDeps & {
   egressAllowlist?: readonly string[];
   /** Called once a repository is enabled, so its open pull requests are triaged without waiting for the next pass. */
   onEnabled: (tenantId: string) => void;
-  /** Called after connect and disconnect: the deployed pr-triage entry follows the connected hosts. */
+  /** Called after connect and disconnect, so a disconnected host's listener and managers are released. */
   onConnectionChanged: (tenantId: string) => Promise<void>;
   now: () => Date;
   log: (entry: Record<string, unknown>) => void;

apps/hub/src/bitbucket-datacenter/tenant-bitbucket.test.ts

@@ -1,41 +1,27 @@
 import { expect, test } from "bun:test";
 import type { DB } from "@intx/db";
 import type { CredentialCipher } from "@intx/types";
-import type { TriageJob } from "../github/coordinator.js";
-import { jobWorkflowsFor, triageWorkflowFor } from "../github/source-workflow.js";
-import { checkedBitbucketSpawner, createTenantBitbucket } from "./tenant-bitbucket.js";
+import { checkedBitbucketDelivery, createTenantBitbucket } from "./tenant-bitbucket.js";
 
 const TENANT = { config: { corbitsTriage: { bitbucketDatacenter: { baseUrl: "https://git.example" } } } };
 
-const KNOWN_BITBUCKET_WORKFLOWS = ["pr-triage-bitbucket-datacenter", "pr-code-review-bitbucket-datacenter", "pr-workflow-anthropic-bitbucket-datacenter"];
-
-function job(workflow: string): TriageJob {
-  return { tenantId: "tnt_spawn", workflow, repo: "ACME/widgets", number: 1, headSha: "abc", source: "event", payload: {} };
-}
-
-test("no Bitbucket worker is spawned once the stored host resolves to an address the token may not reach", async () => {
+test("no Bitbucket mail reaches the listener once the stored host resolves to an address the token may not reach", async () => {
   const tenantBitbucket = createTenantBitbucket({
     db: {
       query: {
         tenant: { async findFirst() { return TENANT; } },
-        credential: { async findFirst() { return { id: "crd_bb", tenantId: "tnt_spawn", status: "active", principalId: null }; } },
+        credential: { async findFirst() { return { id: "crd_bb", tenantId: "tnt_mail", status: "active", principalId: null }; } },
       },
     } as unknown as DB["db"],
     cipher: {} as CredentialCipher,
     fetchImpl: fetch,
     async lookup() { return [{ address: "169.254.169.254", family: 4 }]; },
   });
-  const spawned: string[] = [];
-  const spawner = checkedBitbucketSpawner({
-    async spawn(next) { spawned.push(next.workflow); return "run_worker"; },
-    async ended() {},
-    async release() {},
+  const delivered: unknown[] = [];
+  const deliver = checkedBitbucketDelivery(async function record(_tenantId, _address, payload) {
+    delivered.push(payload);
   }, tenantBitbucket);
-  const workflows = jobWorkflowsFor("bitbucket-datacenter");
-  expect(workflows).toEqual(expect.arrayContaining(KNOWN_BITBUCKET_WORKFLOWS));
-  for (const workflow of workflows) {
-    await expect(spawner.spawn(job(workflow))).rejects.toThrow("host is refused");
-  }
-  expect(await spawner.spawn(job(triageWorkflowFor("github").id))).toBe("run_worker");
-  expect(spawned).toEqual([triageWorkflowFor("github").id]);
+  await expect(deliver("tnt_mail", "run_listener@tenant.example", { repoSource: "bitbucket-datacenter", repo: "ACME/widgets" })).rejects.toThrow("host is refused");
+  await deliver("tnt_mail", "run_listener@tenant.example", { repoSource: "github", repo: "acme/widgets" });
+  expect(delivered).toEqual([{ repoSource: "github", repo: "acme/widgets" }]);
 });

apps/hub/src/bitbucket-datacenter/tenant-bitbucket.ts

@@ -3,11 +3,11 @@
 import type { DB } from "@intx/db";
 import type { CredentialCipher } from "@intx/types";
 import type { BitbucketFetch } from "@corbits/bitbucket-datacenter-reads";
+import { repoSourceOf } from "@corbits/triage-contracts";
+import { asRecord } from "@corbits/common";
 import { tenantConfigOf } from "../github/bridge.js";
 import { activeCredential, credentialSecret } from "../github/portal-credential.js";
 import { triageNs } from "../github/tenant-config.js";
-import type { TriageJob, WorkerSpawner } from "../github/coordinator.js";
-import { jobWorkflowsFor } from "../github/source-workflow.js";
 import { bitbucketFetch, lookupAddresses, resolveInstance, type AddressLookup } from "./client.js";
 import { bitbucketConnection, TOKEN_CREDENTIAL } from "./records.js";
 
@@ -36,15 +36,14 @@ export function createTenantBitbucket(deps: TenantBitbucketDeps): TenantBitbucke
   };
 }
 
-const BITBUCKET_WORKFLOWS = new Set(jobWorkflowsFor("bitbucket-datacenter"));
+export type Deliver = (tenantId: string, address: string, payload: unknown) => Promise<void>;
 
-/** A Bitbucket run sends the token to the stored instance, so its worker is spawned only while that host passes the check. */
-export function checkedBitbucketSpawner(spawner: WorkerSpawner, tenantBitbucket: TenantBitbucket): WorkerSpawner {
-  async function spawn(job: TriageJob): Promise<string> {
-    if (BITBUCKET_WORKFLOWS.has(job.workflow) && (await tenantBitbucket(job.tenantId)) === undefined) {
+/** A Bitbucket run sends the token to the stored instance, so its mail goes out only while that host passes the check. */
+export function checkedBitbucketDelivery(deliver: Deliver, tenantBitbucket: TenantBitbucket): Deliver {
+  return async function deliverChecked(tenantId, address, payload) {
+    if (repoSourceOf(asRecord(payload)?.["repoSource"]) === "bitbucket-datacenter" && (await tenantBitbucket(tenantId)) === undefined) {
       throw new Error("Bitbucket is not connected, has no active token, or its host is refused");
     }
-    return spawner.spawn(job);
-  }
-  return { ...spawner, spawn };
+    await deliver(tenantId, address, payload);
+  };
 }

apps/hub/src/code-review-runs.postgres.test.ts

@@ -3,7 +3,7 @@ import { expect, test } from "bun:test";
 import { randomBytes } from "node:crypto";
 import { sql } from "drizzle-orm";
 import { createDB, runMigrations, schema } from "@intx/db";
-import { diffHash, type PullRequestToReview } from "./code-review-feed.js";
+import { diffHash, type PullRequestToReview } from "./pull-request-reader.js";
 import { createCodeReviewRuns } from "./code-review-runs.js";
 import { recordEvent, type PullRequestRef } from "./manager-state.js";
 import { migrateTriageTables } from "./triage-db.js";
@@ -100,26 +100,27 @@ test("PostgreSQL: a pull request's latest code review is what its manager record
     // A finished review names its diff's hash, so a later head with the same diff takes it over.
     await recordEvent(db, TENANT, GITHUB, "run_live", { type: "reviewer-result", head: "sha4", payload: { kind: "code-review", result: OUTPUT, diffHash: diffHash(DIFF) } });
     const hashed = { ...finished, headSha: "sha4", diffHash: diffHash(DIFF) };
-    expect(await runs.carries(TENANT, RECORD, 7, head)).toBe(true);
     expect(await reviewsOf(GITHUB, "sha5")).toEqual({ review: hashed, completed: hashed, carriedTo: "sha5" });
-    expect(hostReads).toBe(0);
+    expect(await reviewsOf(GITHUB, "sha5")).toMatchObject({ carriedTo: "sha5" });
+    expect(hostReads).toBe(1);
     expect(await reviewsOf(GITHUB, "sha4")).toMatchObject({ carriedTo: null });
 
-    expect(await runs.carries(TENANT, RECORD, 7, { ...head, headSha: "sha6", diff: BASE_EDITED })).toBe(false);
-    expect(await runs.carries(TENANT, RECORD, 7, { ...head, headSha: "sha7", diff: REBUILT })).toBe(false);
-    expect(await runs.carries(TENANT, RECORD, 7, { ...head, headSha: "sha8", diff: `${DIFF}+c\n` })).toBe(false);
+    for (const [headSha, diff] of [["sha6", BASE_EDITED], ["sha7", REBUILT], ["sha8", `${DIFF}+c\n`]] as const) {
+      head = { ...head, headSha, diff };
+      expect(await reviewsOf(GITHUB, headSha)).toMatchObject({ carriedTo: null });
+    }
 
-    head = { ...head, headSha: "sha10" };
+    head = { ...head, headSha: "sha10", diff: DIFF };
     expect(await reviewsOf(GITHUB, "sha10")).toMatchObject({ carriedTo: "sha10" });
     expect(await reviewsOf(GITHUB, "sha11")).toMatchObject({ carriedTo: null });
-    expect(hostReads).toBe(2);
+    expect(hostReads).toBe(6);
 
     // A read that fails, or finds another head, is not tried again on every poll.
     hostDown = true;
     expect(await reviewsOf(GITHUB, "sha12")).toMatchObject({ carriedTo: null });
     expect(await reviewsOf(GITHUB, "sha12")).toMatchObject({ carriedTo: null });
     expect(await reviewsOf(GITHUB, "sha11")).toMatchObject({ carriedTo: null });
-    expect(hostReads).toBe(3);
+    expect(hostReads).toBe(7);
   } finally {
     await handle?.close();
     await admin.db.execute(sql.raw(`drop database if exists ${database} with (force)`));

apps/hub/src/code-review-runs.ts

@@ -5,7 +5,7 @@ import { eq } from "drizzle-orm";
 import { type } from "arktype";
 import { schema, type DB } from "@intx/db";
 import { CodeReviewComment, DiffCut, REPO_SOURCES, repoSourceOf, StageResult, type RepoSource } from "@corbits/triage-contracts";
-import { diffHash, type PullRequestToReview } from "./code-review-feed.js";
+import { diffHash, type PullRequestToReview } from "./pull-request-reader.js";
 import { repoRecords, sourceOf, triageNs, type RepoRecord } from "./github/tenant-config.js";
 import type { Member } from "./integrations/app.js";
 import { codeReviews, type CodeReviewRecord } from "./manager-state.js";
@@ -132,22 +132,11 @@ export function createCodeReviewRuns(deps: CodeReviewRunsDeps) {
     return { ...reviews, carriedTo: await carriedTo(tenantId, tenant.config, query, reviews) };
   }
 
-  /** Whether `pull`'s head takes the newest review over rather than being reviewed; false when that can't be told. */
-  async function carries(tenantId: string, record: RepoRecord, number: number, pull: PullRequestToReview): Promise<boolean> {
-    try {
-      const completed = carriable(await reviewsOf(tenantId, sourceOf(record), record.name, number), pull.headSha);
-      return completed !== undefined && sameDiff(tenantId, record, number, completed, pull);
-    } catch (err) {
-      deps.log({ level: "warn", msg: "code_review_carry_unread", tenantId, repo: record.name, pr: number, error: String(err) });
-      return false;
-    }
-  }
-
   async function latestCodeReview(query: typeof CodeReviewQuery.infer, { tenantId }: Member): Promise<Response> {
     return Response.json(await latest(tenantId, query));
   }
 
-  return { latest, carries, latestCodeReview };
+  return { latest, latestCodeReview };
 }
 
 export type CodeReviewRuns = ReturnType<typeof createCodeReviewRuns>;

apps/hub/src/code-review.postgres.test.ts

@@ -9,7 +9,6 @@ import { DOCKER_PLACE, DOCKER_REMOTE_PLACE } from "@corbits/provisioner-docker";
 import { emptyPack, type CodeReviewPlace } from "@corbits/triage-contracts";
 import { createCodeReviewPlace, createCodeReviewPlacement, createCodeReviewRun, createCodeReviewSwitch } from "./code-review.js";
 import { NoLiveDeploymentError } from "./github/deployment.js";
-import { codeReviewSources } from "./github/source-workflow.js";
 import { repoRecords, triageNs } from "./github/tenant-config.js";
 
 const TEST_DB = { host: "localhost", port: 5432, user: "postgres", password: "postgres", database: "interchange" };
@@ -94,10 +93,8 @@ test("PostgreSQL: code review turns on only with an isolated place to run, the h
     expect((await answer({ repo: "acme/missing", enabled: true })).status).toBe(404);
     expect((await answer({ repo: "acme/gone", enabled: true })).body.error?.code).toBe("repo_not_connected");
     expect(await answer({ repo: "acme/widgets", enabled: true })).toEqual({ status: 200, body: { repo: "acme/widgets", source: "github", codeReview: true } });
-    expect(await codeReviewSources(db, TENANT)).toEqual(["github"]);
 
     expect((await answer({ repo: "acme/widgets", enabled: false })).body.codeReview).toBe(false);
-    expect(await codeReviewSources(db, TENANT)).toEqual([]);
     const tenant = await db.query.tenant.findFirst({ where: eq(schema.tenant.id, TENANT) });
     expect(repoRecords(triageNs(tenant?.config)).map((record) => [record.name, record.codeReview])).toEqual([["acme/widgets", false], ["TRI/widgets", undefined], ["acme/gone", undefined]]);
     expect(changed).toEqual([TENANT, TENANT]);

apps/hub/src/code-review.ts

@@ -10,7 +10,7 @@ import { failure } from "./github/portal-credential.js";
 import { connectedSources, hasCodeReviewKey } from "./github/source-workflow.js";
 import { mailPayload, tenantConfigOf } from "./github/bridge.js";
 import { patchCorbitsTriage, repoRecords, sourceOf, triageNs, type RepoRecord } from "./github/tenant-config.js";
-import type { PullRequestHead } from "./code-review-feed.js";
+import type { PullRequestHead } from "./pull-request-reader.js";
 import type { PolicyReader } from "./github/check-pack-store.js";
 import { NoLiveDeploymentError } from "./github/deployment.js";
 import { triageInputs } from "./github/pr-triage.js";

apps/hub/src/github/bridge.test.ts

@@ -161,7 +161,6 @@ function bridge({ refreshed = [], ...overrides }: Partial<BridgeDeps> & { sent?:
     cipher: stubCipher,
     cache: new DeliveryCache(),
     coalescer: { accept },
-    codeReview: { async offer() {} },
     log: ignoreLog,
     readCheckPack: packsFor(["octocat/hello"]),
     openHeadFor: openHeadOf([]),
@@ -469,21 +468,14 @@ describe("bridge handler", () => {
     expect([sent, rechecks, refreshed]).toEqual([[], [], []]);
   });
 
-  test("a disabled repo is 202 ignored, and its pull request event still reaches code review", async () => {
+  test("a disabled repo is 202 ignored", async () => {
     const sent: Sent[] = [];
-    const offered: string[] = [];
     const db = stubDb([hookRow()], {
       corbitsTriage: { repos: [{ name: "octocat/hello", connected: true, enabled: false }] },
     });
-    const codeReview = {
-      async offer(_tenantId: string, record: { name: string }, mail: { action: string | null }) {
-        offered.push(`${record.name}:${mail.action}`);
-      },
-    };
-    const res = await bridge({ db, sent, codeReview })(githubRequest(prPayload), TARGET);
+    const res = await bridge({ db, sent })(githubRequest(prPayload), TARGET);
     expect(await res.json()).toEqual({ status: "ignored" });
     expect(sent).toHaveLength(0);
-    expect(offered).toEqual([`octocat/hello:${JSON.parse(prPayload).action}`]);
   });
 
   test("a draft is 202 ignored as its event says, and triaged once marked ready; a check run is left to its window to read", async () => {

apps/hub/src/github/bridge.ts

@@ -45,7 +45,6 @@ import type { PolicyReader, StoredPack } from "./check-pack-store.js";
 import type { Accepted, Coalescer } from "./coalescer.js";
 import type { OpenHeadReader } from "./tenant-open-heads.js";
 import type { RepoReads } from "./repo-reads.js";
-import type { CodeReviewFeed } from "../code-review-feed.js";
 
 const HOOK_VERIFY = ["bearer", "standard-webhooks", "slack", "x-hub-signature"] as const;
 
@@ -68,8 +67,6 @@ export interface BridgeDeps {
   cipher: CredentialCipher;
   cache: DeliveryCache;
   coalescer: Pick<Coalescer, "accept">;
-  /** Code review is fed whether or not triage is on for the repository. */
-  codeReview: Pick<CodeReviewFeed, "offer">;
   log?: (entry: Record<string, unknown>) => void;
   readCheckPack: PolicyReader;
   openHeadFor: (tenantId: string) => Promise<OpenHeadReader | undefined>;
@@ -552,7 +549,6 @@ export function createBridgeHandler(d: BridgeDeps) {
     }
     const record = await connectedRecord(d, log, loaded, delivery, event, mail.repo);
     if (record instanceof Response) return record;
-    void d.codeReview.offer(loaded.tenantId, record, mail);
     const policy = repoPolicy(record);
     if (!policy.enabled) {
       log({ level: "info", msg: "ignored", delivery, event, repo: mail.repo, hook: loaded.credentialId, reason: "repo_not_enabled" });

apps/hub/src/github/coalescer.ts

@@ -23,7 +23,7 @@
 import { dependsOnCi, marksWaiting, type ActionKind, type CheckPack, type PolicySource, type PolicyWorkflow, type PrTriageRow, type RepoPolicy, type TriageEvent } from "@corbits/triage-contracts";
 import { summarizeChecks, type CheckRun } from "@corbits/github-reads";
 import { mailPayload, unreadablePayload, waitingPayload, type PolicyUnreadable } from "./bridge.js";
-import type { TriageJob } from "./coordinator.js";
+import type { TriageJob } from "./triage-queue.js";
 import { GithubRateLimited } from "./github-usage.js";
 import type { PrMail } from "./normalize.js";
 import { sourceOf, triageStateKey, type RepoRecord } from "./tenant-config.js";
@@ -207,7 +207,7 @@ export function createCoalescer(deps: CoalescerDeps): Coalescer {
   async function markWaiting({ latest, head }: Open, noCi: boolean): Promise<void> {
     if (!("pack" in latest) || !marksWaiting(sourceOf(latest.record), latest.automatic)) return;
     try {
-      await deps.sendMail({ ...head, workflow: latest.workflow, source: "event", payload: waitingPayload(latest.record, latest.policy, latest, head, noCi) });
+      await deps.sendMail({ ...head, payload: waitingPayload(latest.record, latest.policy, latest, head, noCi) });
     } catch (err) {
       deps.log({ level: "warn", msg: "waiting_mail_failed", tenantId: head.tenantId, repo: latest.record.name, pr: head.number, headSha: head.headSha, error: String(err) });
     }
@@ -262,7 +262,7 @@ export function createCoalescer(deps: CoalescerDeps): Coalescer {
       const noWait = waitedOut(firstSeenAt, deps.now(), deps.ciWait);
       const extra = { ...latest.mail, headSha: head.headSha, events, ...(noWait && { noWait }) };
       const payload = "pack" in latest ? mailPayload(latest.record, latest.policy, latest, extra) : unreadablePayload(latest.record, latest.policy, latest, extra);
-      await deps.sendMail({ ...head, workflow: latest.workflow, source: "event", record: latest.record, payload });
+      await deps.sendMail({ ...head, record: latest.record, payload });
     } catch (err) {
       if (err instanceof GithubRateLimited) return drop({ ...open, remembered }, entry, "github_rate_limited");
       deps.log({ level: "error", msg: "forward_failed", ...entry, error: String(err) });