feat(hub): patch a policy file in place and open one pull request

apps/hub/src/bitbucket-datacenter/policy-pull.test.ts

@@ -1,5 +1,6 @@
 import { expect, test } from "bun:test";
-import { bitbucketFetch } from "./client.js";
+import { BitbucketFailed, bitbucketFetch } from "./client.js";
+import { bodyWith, DescriptionNotUpdated, PolicyFileMoved } from "../github/policy-pull.js";
 import { openPolicyPull, proposePolicy } from "./policy-pull.js";
 
 const REPO = "ACME/widgets";
@@ -12,12 +13,21 @@ const pulls: Array<{ id: number; version: number; title: string; description: st
 const edits: Array<Record<string, string>> = [];
 /** Set, the next commit is followed by another caller opening the pull request first. */
 let raced: (typeof pulls)[number] | undefined;
+/** Each commit made through the browse API, newest last. */
+const log: Array<{ branch: string; id: string; message: string }> = [];
+/** How many description edits fail next. */
+let pullPutFailures = 0;
+/** Set, the description is edited on Bitbucket just before the next edit, so that edit's version is stale. */
+let editedBeforePut = false;
+/** Set, the policy file appears on the branch just before the next commit. */
+let appearsBeforePut: string | undefined;
 
 async function bitbucket(req: Request): Promise<Response> {
   const url = new URL(req.url);
   if (req.method === "DELETE" && url.pathname === "/rest/branch-utils/1.0/projects/ACME/repos/widgets/branches") {
     const { name } = await req.json() as { name: string };
     branches.delete(name.slice("refs/heads/".length));
+    log.splice(0, log.length, ...log.filter((row) => row.branch !== name.slice("refs/heads/".length)));
     return new Response(null, { status: 204 });
   }
   const path = url.pathname.slice(BASE.length);
@@ -34,9 +44,12 @@ async function bitbucket(req: Request): Promise<Response> {
     expect(req.headers.get("content-type")).toStartWith("multipart/form-data");
     const form = Object.fromEntries(await Promise.all([...(await req.formData()).entries()].map(async ([key, value]) => [key, typeof value === "string" ? value : await value.text()])));
     edits.push(form);
+    if (appearsBeforePut !== undefined) branches.set(form.branch!, { text: appearsBeforePut, commit: "c-appeared" });
+    appearsBeforePut = undefined;
     const from = branches.get(form.sourceBranch ?? form.branch!)!;
     if (form.sourceCommitId !== (from.text === undefined ? undefined : from.commit)) return new Response(null, { status: 409 });
     branches.set(form.branch!, { text: form.content, commit: `c${edits.length}` });
+    log.push({ branch: form.branch!, id: `c${edits.length}`, message: form.message! });
     if (raced) pulls.push(raced);
     raced = undefined;
     return Response.json({ id: `c${edits.length}` });
@@ -49,8 +62,23 @@ async function bitbucket(req: Request): Promise<Response> {
     pulls.push({ id: pulls.length + 1, version: 0, open: true, ...(await req.json() as Omit<(typeof pulls)[number], "id" | "version" | "open">) });
     return Response.json(pullJson(pulls.at(-1)!), { status: 201 });
   }
+  if (req.method === "GET" && /^\/pull-requests\/\d+\/commits$/.test(path)) {
+    const pull = pulls.find((row) => row.id === Number(path.split("/")[2]))!;
+    const all = log.filter((row) => `refs/heads/${row.branch}` === pull.fromRef.id).toReversed();
+    const start = Number(url.searchParams.get("start") ?? "0");
+    const end = start + Number(url.searchParams.get("limit") ?? "25");
+    const values = all.slice(start, end).map((row) => ({ id: row.id, message: row.message, author: { name: "triage" }, committerTimestamp: 0 }));
+    return Response.json({ values, isLastPage: end >= all.length, ...(end < all.length && { nextPageStart: end }) });
+  }
+  if (req.method === "GET" && path.startsWith("/pull-requests/")) return Response.json(pullJson(pulls.find((row) => row.id === Number(path.slice("/pull-requests/".length)))!));
   if (req.method === "PUT" && path.startsWith("/pull-requests/")) {
+    if (pullPutFailures > 0) {
+      pullPutFailures -= 1;
+      return new Response(null, { status: 502 });
+    }
     const pull = pulls.find((row) => row.id === Number(path.slice("/pull-requests/".length)))!;
+    if (editedBeforePut) Object.assign(pull, { description: `${pull.description}\n\nedited on Bitbucket`, version: pull.version + 1 });
+    editedBeforePut = false;
     const body = await req.json() as { version: number; title: string; description: string };
     if (body.version !== pull.version) return new Response(null, { status: 409 });
     Object.assign(pull, { title: body.title, description: body.description, version: pull.version + 1 });
@@ -60,7 +88,7 @@ async function bitbucket(req: Request): Promise<Response> {
 }
 
 function pullJson(pull: (typeof pulls)[number]) {
-  return { id: pull.id, version: pull.version, links: { self: [{ href: `https://bitbucket.test/projects/ACME/repos/widgets/pull-requests/${pull.id}` }] } };
+  return { id: pull.id, version: pull.version, title: pull.title, description: pull.description, links: { self: [{ href: `https://bitbucket.test/projects/ACME/repos/widgets/pull-requests/${pull.id}` }] } };
 }
 
 test("the Bitbucket pull request is made once from a new branch, then brought up to date in place", async () => {
@@ -80,11 +108,63 @@ test("the Bitbucket pull request is made once from a new branch, then brought up
     expect(pulls[0]!.description).toBe("second");
     expect(await openPolicyPull(bb, REPO)).toEqual(opened);
 
+    expect(await proposePolicy(bb, REPO, "main", { ...change, text: "rules: []\nlabels: { pass: Go }\n", title: "chore: another title", body: "appended", append: true })).toEqual(opened);
+    expect(pulls[0]!.title).toBe(change.title);
+    expect(pulls[0]!.description).toStartWith("second\n\n");
+    expect(pulls[0]!.description).toEndWith("\nappended");
+    const appended = pulls[0]!.description;
+    await proposePolicy(bb, REPO, "main", { ...change, text: "rules: []\nlabels: { pass: Go }\n", body: "unchanged", append: true });
+    expect(pulls[0]!.description).toBe(appended);
+    await expect(proposePolicy(bb, REPO, "main", { ...change, text: "rules: [] # stale\n", body: "stale", append: true, readSha: "c1" })).rejects.toBeInstanceOf(PolicyFileMoved);
+    expect(pulls[0]!.description).toBe(appended);
+
+    // A description edit that keeps failing names the commit that landed, and the next batch adds its section.
+    pullPutFailures = 4;
+    const lost = proposePolicy(bb, REPO, "main", { ...change, text: "rules: []\nlabels: { pass: Lost }\n", body: "lost", append: true });
+    await expect(lost).rejects.toBeInstanceOf(DescriptionNotUpdated);
+    await expect(lost).rejects.toMatchObject({ commit: log.at(-1)!.id });
+    expect(pulls[0]!.description).not.toContain("lost");
+    await proposePolicy(bb, REPO, "main", { ...change, text: "rules: []\nlabels: { pass: Found }\n", body: "found", append: true });
+    expect(pulls[0]!.description).toStartWith(appended);
+    expect(pulls[0]!.description).toMatch(/\nlost\n[\s\S]*\nfound$/);
+
+    // A description edited on Bitbucket between the read and the edit is read again and keeps both.
+    editedBeforePut = true;
+    await proposePolicy(bb, REPO, "main", { ...change, text: "rules: []\nlabels: { pass: Raced }\n", body: "raced", append: true });
+    expect(pulls[0]!.description).toContain("\nedited on Bitbucket\n");
+    expect(pulls[0]!.description).toEndWith("\nraced");
+
+    // An export's refused edit is Bitbucket's refusal, sent once and not retried.
+    pullPutFailures = 1;
+    await expect(proposePolicy(bb, REPO, "main", { ...change, text: "rules: []\nlabels: { pass: Export }\n", body: "export" })).rejects.toBeInstanceOf(BitbucketFailed);
+    expect(pullPutFailures).toBe(0);
+
+    // A file that appears on a branch read without one is a moved file.
+    branches.set("corbits-triage/policy", { commit: branches.get("corbits-triage/policy")!.commit });
+    appearsBeforePut = "rules: []\n";
+    await expect(proposePolicy(bb, REPO, "main", { ...change, text: "rules: [] # new\n", body: "new", append: true, readSha: null })).rejects.toBeInstanceOf(PolicyFileMoved);
+
+    // An export over a description holding batch sections keeps them and adds its own, so the next batch adds none back.
+    await proposePolicy(bb, REPO, "main", { ...change, text: "rules: []\nlabels: { pass: Kept }\n", body: "kept", append: true });
+    await proposePolicy(bb, REPO, "main", { ...change, text: "rules: []\nlabels: { pass: Exported }\n", body: "exported" });
+    expect(pulls[0]!.description).toContain("\nkept\n");
+    expect(pulls[0]!.description).toEndWith("\nexported");
+    await proposePolicy(bb, REPO, "main", { ...change, text: "rules: []\nlabels: { pass: After }\n", body: "after", append: true });
+    expect(pulls[0]!.description.split("\nkept\n")).toHaveLength(2);
+    expect(pulls[0]!.description.split("\nlost\n")).toHaveLength(2);
+
+    // A lost section older than the newest hundred commits is found on a later page.
+    const older = Array.from({ length: 101 }, (_, at) => ({ branch: "corbits-triage/policy", id: `x${at + 1}`, message: `${change.title}\n\nolder ${at + 1}` }));
+    log.unshift(...older);
+    pulls[0]!.description = bodyWith(pulls[0]!.description, older.slice(1).map((row) => ({ commit: row.id, text: row.message.slice(change.title.length + 2) })));
+    await proposePolicy(bb, REPO, "main", { ...change, text: "rules: []\nlabels: { pass: Paged }\n", body: "paged", append: true });
+    expect(pulls[0]!.description).toContain("(corbits-triage added x1)\n\nolder 1\n");
+
     // Merged and left behind, the branch starts over from main for the next pull request.
     pulls[0]!.open = false;
     branches.set("main", { text: "rules: []\nlabels: { pass: Ready }\n", commit: "c9" });
     expect(await proposePolicy(bb, REPO, "main", { ...change, text: "rules: []\n", body: "third" })).toMatchObject({ number: 2 });
-    expect(edits[2]).toEqual({ content: "rules: []\n", message: change.title, branch: "corbits-triage/policy", sourceBranch: "main", sourceCommitId: "c9" });
+    expect(edits.at(-1)).toEqual({ content: "rules: []\n", message: change.title, branch: "corbits-triage/policy", sourceBranch: "main", sourceCommitId: "c9" });
 
     // Another caller opening it between the check and the create is answered with theirs.
     pulls[1]!.open = false;

apps/hub/src/bitbucket-datacenter/policy-pull.ts

@@ -1,13 +1,14 @@
 // The Bitbucket Data Center side of policy-pull.ts: one pull request from `corbits-triage/policy`, opened once and
 // brought up to date when asked again.
 import { type } from "arktype";
-import { filePath, readRepoFile, repoParts, repoPath, type BitbucketFetch } from "@corbits/bitbucket-datacenter-reads";
-import { POLICY_BRANCH, type PolicyChange, type PolicyPull } from "../github/policy-pull.js";
+import { filePath, listPrCommits, readRepoFile, repoParts, repoPath, type BitbucketFetch } from "@corbits/bitbucket-datacenter-reads";
+import { bodyWith, commitMessage, missingSections, openingBody, POLICY_BRANCH, PolicyFileMoved, replacedBody, retriedEdit, withCommitted, type PolicyChange, type PolicyPull } from "../github/policy-pull.js";
 import { BitbucketFailed } from "./client.js";
 
-const Pull = type({ id: "number", version: "number", links: { self: type({ href: "string" }).array().atLeastLength(1) } });
+const Pull = type({ id: "number", version: "number", title: "string", "description?": "string", links: { self: type({ href: "string" }).array().atLeastLength(1) } });
 const Pulls = type({ values: Pull.array() });
 const Branches = type({ values: type({ displayId: "string" }).array() });
+const Written = type({ id: "string" });
 
 const BRANCH_REF = `refs/heads/${POLICY_BRANCH}`;
 
@@ -42,37 +43,63 @@ async function deleteBranch(bb: BitbucketFetch, repo: string): Promise<void> {
   await send(bb, `/rest/branch-utils/1.0/projects/${project}/repos/${slug}/branches`, { method: "DELETE", body: JSON.stringify({ name: BRANCH_REF, dryRun: false }) });
 }
 
-/** Commits the file to the branch, making the branch from `base` when it is not there; the commit that last changed the file is how Data Center refuses a stale edit. */
-async function commitFile(bb: BitbucketFetch, repo: string, base: string, onBranch: boolean, { path, text, title }: PolicyChange): Promise<void> {
+/**
+ * Commits the file to the branch, making the branch from `base` when it is not there; the commit that last changed the
+ * file is how Data Center refuses a stale edit. The commit made, undefined when the branch already held the text.
+ */
+async function commitFile(bb: BitbucketFetch, repo: string, base: string, onBranch: boolean, change: PolicyChange): Promise<string | undefined> {
+  const { path, text, readSha } = change;
   const current = await readRepoFile(bb, repo, path, onBranch ? POLICY_BRANCH : base);
-  if (onBranch && current?.kind === "file" && current.text === text) return;
+  if (onBranch && current?.kind === "file" && current.text === text) return undefined;
+  if (readSha !== undefined && (current?.kind === "file" ? current.sha : null) !== readSha) throw new PolicyFileMoved(path);
   const form = new FormData();
   // A file part keeps the text's line endings; a plain field would send them as CRLF.
   form.set("content", new Blob([text], { type: "text/plain" }), path);
-  form.set("message", title);
+  form.set("message", commitMessage(change));
   form.set("branch", POLICY_BRANCH);
   if (!onBranch) form.set("sourceBranch", base);
   if (current?.kind === "file") form.set("sourceCommitId", current.sha);
-  await send(bb, `${repoPath(repo)}/browse/${filePath(path)}`, { method: "PUT", body: form });
+  const written = `${repoPath(repo)}/browse/${filePath(path)}`;
+  // Data Center refuses the edit with 409 when a commit has changed the file since `sourceCommitId`, or when a file
+  // appeared where none was read.
+  const res = await bb(written, { method: "PUT", body: form });
+  if (res.status === 409 && readSha !== undefined) throw new PolicyFileMoved(path);
+  if (!res.ok) throw new BitbucketFailed(res.status, written);
+  return Written.assert(await res.json()).id;
+}
+
+/** Adds a section for each batch commit the description lacks, the new one included; with none lacking, no edit is made. */
+async function appendSections(bb: BitbucketFetch, repo: string, id: number, change: PolicyChange, committed: string | undefined): Promise<void> {
+  const path = `${repoPath(repo)}/pull-requests/${id}`;
+  await retriedEdit(committed, async function appendMissing() {
+    const latest = Pull.assert(await send(bb, path));
+    // Data Center lists the newest commit first.
+    const listed = (await listPrCommits(bb, repo, id)).toReversed();
+    const commits = withCommitted(listed.map((row) => ({ commit: row.sha, message: row.message })), committed, change);
+    const added = missingSections(latest.description, commits, change.title);
+    if (added.length === 0) return;
+    await send(bb, path, { method: "PUT", body: JSON.stringify({ version: latest.version, title: latest.title, description: bodyWith(latest.description, added) }) });
+  });
 }
 
 export async function proposePolicy(bb: BitbucketFetch, repo: string, base: string, change: PolicyChange): Promise<PolicyPull> {
   const open = await openPull(bb, repo);
   const left = !open && (await branchExists(bb, repo));
   if (left) await deleteBranch(bb, repo);
-  await commitFile(bb, repo, base, Boolean(open), change);
-  const pulls = `${repoPath(repo)}/pull-requests`;
-  if (open) {
-    await send(bb, `${pulls}/${open.id}`, { method: "PUT", body: JSON.stringify({ version: open.version, title: change.title, description: change.body }) });
+  const committed = await commitFile(bb, repo, base, Boolean(open), change);
+  if (!open) return createPull(bb, repo, base, change, committed);
+  if (change.append) {
+    await appendSections(bb, repo, open.id, change, committed);
     return pullOf(open);
   }
-  return createPull(bb, repo, base, change);
+  await send(bb, `${repoPath(repo)}/pull-requests/${open.id}`, { method: "PUT", body: JSON.stringify({ version: open.version, title: change.title, description: replacedBody(open.description, change, committed) }) });
+  return pullOf(open);
 }
 
 /** A pull request opened by a concurrent call in between is answered instead of refused. */
-async function createPull(bb: BitbucketFetch, repo: string, base: string, { title, body }: PolicyChange): Promise<PolicyPull> {
+async function createPull(bb: BitbucketFetch, repo: string, base: string, change: PolicyChange, committed: string | undefined): Promise<PolicyPull> {
   const path = `${repoPath(repo)}/pull-requests`;
-  const res = await bb(path, { method: "POST", body: JSON.stringify({ title, description: body, fromRef: { id: BRANCH_REF }, toRef: { id: `refs/heads/${base}` } }) });
+  const res = await bb(path, { method: "POST", body: JSON.stringify({ title: change.title, description: openingBody(change, committed), fromRef: { id: BRANCH_REF }, toRef: { id: `refs/heads/${base}` } }) });
   if (res.ok) return pullOf(Pull.assert(await res.json()));
   const raced = res.status === 409 ? await openPull(bb, repo) : undefined;
   if (raced) return pullOf(raced);

apps/hub/src/github/coalescer.ts

@@ -31,6 +31,7 @@ import type { HeadChecksReader, OpenHeadReader } from "./tenant-open-heads.js";
 import { inFlight, waitedOut, waitsForCi, type ReconcilePolicy } from "./reconcile-plan.js";
 import { deliveryFailure, failHead, queueHead, restoreHead, waitHead, type QueuedHead, type Remembered } from "./triage-queue.js";
 import { TriageStateConflictError, type TriageStateStore } from "./triage-state-store.js";
+import { createKeyedSerial } from "../keyed-serial.js";
 
 export type CoalesceWindow = { quietMs: number; maxWaitMs: number; burstMs: number };
 
@@ -130,18 +131,10 @@ function eventsOf(input: CoalescedEvent, rows: readonly PrTriageRow[]): TriageEv
   return untriaged ? ["ready", "opened"] : [input.event];
 }
 
-async function settled(work: Promise<unknown>): Promise<void> {
-  try {
-    await work;
-  } catch {
-    // The caller of `work` handles its failure; the chain only waits for it.
-  }
-}
-
 export function createCoalescer(deps: CoalescerDeps): Coalescer {
   const windows = new Map<string, Open>();
-  /** The last accept or flush of each pull request, so they never interleave. */
-  const tails = new Map<string, Promise<void>>();
+  /** Each pull request's accepts and flushes, so they never interleave. */
+  const serialized = createKeyedSerial();
   /** The head each pull request's latest new commit or opening named, kept past its window so a late event cannot move it back. */
   const newest = new Map<string, string>();
 
@@ -153,23 +146,6 @@ export function createCoalescer(deps: CoalescerDeps): Coalescer {
     return { tenantId, repo: triageStateKey(record), number, headSha };
   }
 
-  async function serialized<T>(key: string, work: () => Promise<T>): Promise<T> {
-    const previous = tails.get(key);
-    const current = runAfter(previous, work);
-    const tail = settled(current);
-    tails.set(key, tail);
-    try {
-      return await current;
-    } finally {
-      if (tails.get(key) === tail) tails.delete(key);
-    }
-  }
-
-  async function runAfter<T>(previous: Promise<void> | undefined, work: () => Promise<T>): Promise<T> {
-    await previous;
-    return work();
-  }
-
   function timer(key: string, ms: number): () => void {
     return deps.setTimer(function onWindowEnd() {
       return flushLogged(key);