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 }));
}