import { EventEmitter } from "node:events";
import { agentFields, agentOf, type Actor, type AgentRef } from "./actor";
import { startEventWorkflows } from "./ci";
import { db, filter, type RecordModel } from "./db";
import type { RepoRecord } from "./repos";
import { getUserById } from "./users";
import { deliverEvent } from "./webhooks";
/*
* Everything that happens in a repository, as a feed: pushes, issues, pull requests, reviews,
* workflow runs, releases. Each event is stored (so agents can catch up from a cursor with
* list_events, and people can see what their agents did), handed to the repository's webhooks, and
* starts the workflows that run on it (on: issues, issue_comment, pull_request, …).
*
* Emitting never fails the action that caused it: problems are logged. Like lib/ci.ts this is used
* by the SSH server, so it must not import anything from next/*.
*/
export const EVENT_KINDS = [
"push",
"create",
"delete",
"issues",
"issue_comment",
"pull_request",
"pull_request_review",
"pull_request_review_comment",
"workflow_run",
"release",
"fork",
] as const;
export type EventKind = (typeof EVENT_KINDS)[number];
export const EVENT_DESCRIPTIONS: Record<EventKind, string> = {
push: "Commits pushed to a branch or tag",
create: "A branch or tag created",
delete: "A branch or tag deleted",
issues: "Issues opened, edited, assigned, labeled, closed or reopened",
issue_comment: "Comments on issues and pull requests",
pull_request: "Pull requests opened, updated, merged, closed or reopened",
pull_request_review: "Reviews submitted",
pull_request_review_comment: "Comments on lines of a pull request's changes",
workflow_run: "Workflow runs finished",
release: "Releases published or deleted",
fork: "The repository forked",
};
export type RepoEvent = {
id: string;
seq: number;
repoId: string;
kind: EventKind;
action: string | null;
actor: string | null;
agent: AgentRef | null;
subject: string;
payload: Record<string, unknown>;
createdAt: number;
};
/** Events are kept this long. */
const RETENTION_MS = 90 * 86400e3;
const g = globalThis as unknown as { __eventSeq?: number; __eventBus?: EventEmitter };
const bus = (g.__eventBus ??= new EventEmitter().setMaxListeners(0));
/** Increasing numbers that also order events by time: microseconds, bumped on collisions. */
function nextSeq() {
const seq = Math.max(Date.now() * 1000, (g.__eventSeq ?? 0) + 1);
g.__eventSeq = seq;
return seq;
}
const toEvent = (r: RecordModel): RepoEvent => ({
id: r.id,
seq: r.seq,
repoId: r.repo,
kind: r.kind,
action: r.action || null,
actor: r.expand?.actor?.username ?? null,
agent: agentOf(r),
subject: r.subject,
payload: r.payload ?? {},
createdAt: r.created_at,
});
export type EmitInput = {
kind: EventKind;
action?: string;
actor: Actor | null;
/** What it's about, for lists: "#12", "main", "v1.0". */
subject: string;
payload: Record<string, unknown>;
};
/** The parts of an event's payload every event has. */
async function envelope(repo: RepoRecord, input: EmitInput) {
const user = input.actor ? await getUserById(input.actor.userId) : null;
return {
action: input.action ?? null,
repository: { full_name: `${repo.owner}/${repo.name}`, owner: repo.owner, name: repo.name, visibility: repo.visibility },
sender: user ? { username: user.username, agent: input.actor?.agent?.name ?? null } : null,
...input.payload,
};
}
async function store(repo: RepoRecord, input: EmitInput): Promise<RepoEvent> {
const payload = await envelope(repo, input);
const row = await (await db()).collection("events").create(
{
seq: nextSeq(),
repo: repo.id,
kind: input.kind,
action: input.action ?? "",
actor: input.actor?.userId ?? "",
...agentFields(input.actor?.agent),
subject: input.subject.slice(0, 300),
payload,
created_at: Date.now(),
},
{ expand: "actor" },
);
if (Math.random() < 0.002) prune().catch((e) => console.error("[events] could not prune:", e));
return toEvent(row);
}
async function prune() {
const pb = await db();
const { items } = await pb.collection("events").getList(1, 500, {
filter: filter("created_at < {:cutoff}", { cutoff: Date.now() - RETENTION_MS }),
fields: "id",
skipTotal: true,
});
for (const { id } of items) await pb.collection("events").delete(id);
}
/** Records an event and sets off what follows from it. Resolves once it's stored; never throws. */
export async function emitEvent(repo: RepoRecord, input: EmitInput): Promise<RepoEvent | null> {
let event: RepoEvent;
try {
event = await store(repo, input);
} catch (e) {
console.error(`[events] could not record ${input.kind}:`, e);
return null;
}
bus.emit("event", event);
deliverEvent(repo, event).catch((e) => console.error("[webhooks] could not deliver:", e));
startEventWorkflows(repo, event, input.actor).catch((e) => console.error("[ci] could not start workflows:", e));
return event;
}
/** For callers that shouldn't wait for the event to be stored. */
export function emitInBackground(repo: RepoRecord, input: EmitInput) {
void emitEvent(repo, input);
}
// ---------------------------------------------------------------------------
// Reading
const anyOf = (field: string, values: string[]) => `(${values.map((v, i) => filter(`${field} = {:v${i}}`, { [`v${i}`]: v })).join(" || ")})`;
/** Events in these repositories after `since` (a seq), oldest first. */
export async function listEvents(opts: { repoIds: string[]; since?: number; kinds?: string[]; limit: number }) {
if (opts.repoIds.length === 0) return [];
const where = [anyOf("repo", opts.repoIds), filter("seq > {:since}", { since: opts.since ?? 0 })];
if (opts.kinds?.length) where.push(anyOf("kind", opts.kinds));
const { items } = await (await db()).collection("events").getList(1, opts.limit, {
filter: where.join(" && "),
// Without a cursor, the latest events (still returned oldest first).
sort: opts.since === undefined ? "-seq" : "seq",
expand: "actor",
skipTotal: true,
});
const events = items.map(toEvent);
return opts.since === undefined ? events.reverse() : events;
}
/** The newest event's seq in these repositories, to start following them from now. */
export async function latestSeq(repoIds: string[]) {
if (repoIds.length === 0) return 0;
const { items } = await (await db()).collection("events").getList(1, 1, { filter: anyOf("repo", repoIds), sort: "-seq", fields: "seq", skipTotal: true });
return (items[0]?.seq as number | undefined) ?? 0;
}
/** Waits until an event in one of these repositories happens, or the time is up. Resolves whether one did. */
export function waitForEvent(repoIds: string[], timeoutMs: number, signal?: AbortSignal) {
const ids = new Set(repoIds);
return new Promise<boolean>((resolve) => {
const done = (happened: boolean) => {
clearTimeout(timer);
bus.off("event", listener);
signal?.removeEventListener("abort", aborted);
resolve(happened);
};
const listener = (e: RepoEvent) => ids.has(e.repoId) && done(true);
const aborted = () => done(false);
const timer = setTimeout(() => done(false), timeoutMs);
bus.on("event", listener);
signal?.addEventListener("abort", aborted);
});
}
/** What a person's agents did, newest first: everything an agent caused, across repositories. */
export async function listAgentActivity(userId: string, opts: { agent?: string; repoId?: string; page: number; perPage: number }) {
const where = [filter("actor = {:userId} && agent != ''", { userId })];
if (opts.agent) where.push(filter("agent = {:agent}", { agent: opts.agent }));
if (opts.repoId) where.push(filter("repo = {:repo}", { repo: opts.repoId }));
const result = await (await db()).collection("events").getList(opts.page, opts.perPage, {
filter: where.join(" && "),
sort: "-seq",
expand: "actor",
});
return { events: result.items.map(toEvent), total: result.totalItems };
}
export async function getEvent(id: string) {
const row = await (await db()).collection("events").getOne(id, { expand: "actor" }).catch(() => null);
return row ? toEvent(row) : null;
}
/** The agents that have done something for this person, for the activity filter. */
export async function listActiveAgents(userId: string) {
const { items: rows } = await (await db()).collection("events").getList(1, 2000, {
filter: filter("actor = {:userId} && agent != ''", { userId }),
fields: "agent,agent_name,seq",
sort: "-seq",
skipTotal: true,
});
const seen = new Map<string, string>();
for (const r of rows) if (!seen.has(r.agent)) seen.set(r.agent, r.agent_name || "Agent");
return [...seen].map(([key, name]) => ({ key, name }));
}