// **The forge's announcements carry a message id** (novox/hq ADR 0406 §6 and §7, issue 976). A merge, a new head and a // close are announced by more than one path — the poll, the forge's own event, the merge tool's ask of a check — and // were kept from being announced twice by files of what had been announced. They are now announced with a message id // that is the same for the same announcement, whichever path makes it: the runtime makes an event's `x-event-id` header // the bus's message id (mesh-tools node-tools/internal/bus/bus.go, `nats.MsgId`), and the bus stores a message it has // seen the id of, within its de-duplication window, once. // // The SDK's `emit` takes no id, so this publishes through the broker with the headers `emit` stamps and the id given. // It returns to `emit` once the SDK takes an id. // // The ids: a merge by its merge commit, a head by the head and the trunk commit it was announced against (the check asked // again against a new trunk commit is another announcement), a close by its head. import { broker } from "@novox/mesh-sdk/messaging"; import { state as keptState } from "@novox/mesh-sdk/state"; type Said = { owner?: string; repo?: string; number?: number; head_sha?: string; base_sha?: string; merge_commit_sha?: string }; /** The message id of an announcement, or undefined for an event that is not one of these three. */ export function announcementId(type: string, body: unknown): string | undefined { const b = (body ?? {}) as Said; const pull = `${b.owner}/${b.repo}#${b.number}`; switch (type) { case "pull.merged": return `merged:${pull}@${b.merge_commit_sha ?? ""}`; case "pull.updated": return `updated:${pull}@${b.head_sha ?? ""}:${b.base_sha ?? ""}`; case "pull.closed": return `closed:${pull}@${b.head_sha ?? ""}`; default: return undefined; } } /** Announce a merge, a head or a close: published with its message id. */ export async function announce(type: string, body: unknown): Promise { const id = announcementId(type, body); if (!id) throw new Error(`${type} is not an announcement with a message id`); const node = process.env.MESH_NODE ?? "unknown"; await broker().publish({ key: type, node, body, headers: { "x-event-id": id, "x-source": process.env.MESH_MODULE ?? "unknown", "x-node": node, "x-time": new Date().toISOString(), "content-type": "application/json", }, }); } // --- the watermark: the one value of the old files that is not the bus's to remember -------------------------------- /** What the module keeps of how far its poll of merges has looked: a time, in its own state. */ interface Kept { get(key: string): Promise<{ value: unknown } | null>; put(key: string, value: unknown): Promise; } const WATERMARK = "merges-since"; /** The time the poll of merges last looked from, or "" when none is kept or the state cannot be reached: a merge made * before it is history, one after it is announced. */ export async function readWatermark(state: Kept = keptState("announced")): Promise { try { const entry = await state.get(WATERMARK); const since = (entry?.value as { since?: string } | undefined)?.since; return typeof since === "string" ? since : ""; } catch (err) { console.error(`[gitea] where the poll of merges last looked from could not be read, so none is kept — ${err instanceof Error ? err.message : err}`); return ""; } } /** Keep the time the poll of merges looked from; a state that cannot be reached is said, and the next pass keeps it. */ export async function keepWatermark(since: string, state: Kept = keptState("announced")): Promise { try { await state.put(WATERMARK, { since }); } catch (err) { console.error(`[gitea] where the poll of merges looked from could not be kept — ${err instanceof Error ? err.message : err}`); } }