feat(git): shared idempotent ingest core

src/lib/git/ingest.ts

@@ -0,0 +1,180 @@
+import "server-only";
+
+import { getAdminPb } from "@/lib/pb/admin";
+import { parseCommitMessage } from "@/lib/git/conventional";
+import { fetchCommitPage, type ForgejoCommit } from "@/lib/git/forgejo";
+import type { ProjectRecord } from "@/lib/pb/types";
+
+export interface IngestResult {
+  created: number;
+  skipped: number;
+  pagesRead: number;
+}
+
+/**
+ * Turn one Forgejo commit into a row.
+ *
+ * `commit.author.date` — the authoring time — not `created` and not
+ * `committer.date`. The latter two move on rebase, and a rebased branch would
+ * then look like one enormous coding session in Plan 2c's effort estimate.
+ *
+ * `parents.length > 1` is the authoritative merge test. Matching on a
+ * "Merge branch" message prefix would also catch a commit that merely talks
+ * about merging, and would miss a merge with a rewritten message.
+ */
+function toRecord(projectId: string, commit: ForgejoCommit) {
+  const parsed = parseCommitMessage(commit.commit.message);
+  const isMerge = (commit.parents?.length ?? 0) > 1;
+
+  return {
+    project: projectId,
+    sha: commit.sha,
+    short_sha: commit.sha.slice(0, 7),
+    message: commit.commit.message.slice(0, 50000),
+    subject: parsed.subject,
+    // A merge is real history but is not a unit of work. Typing it here keeps
+    // every consumer from having to remember to exclude it.
+    type: isMerge ? "merge" : parsed.type,
+    scope: parsed.scope,
+    author_name: commit.commit.author.name,
+    author_email: commit.commit.author.email,
+    authored_at: commit.commit.author.date,
+    is_merge: isMerge,
+    additions: commit.stats?.additions ?? 0,
+    deletions: commit.stats?.deletions ?? 0,
+    link_source: "none" as const,
+    url: commit.html_url,
+  };
+}
+
+/**
+ * Does this PocketBase error mean "that sha is already stored"?
+ *
+ * PocketBase answers 400 for EVERY validation failure, so status alone cannot
+ * distinguish a duplicate from a genuinely malformed record. Treating all 400s
+ * as duplicates would silently drop a commit that failed for any other reason —
+ * an over-long field, say — count it as "skipped", and under-report the
+ * dashboard with no error anywhere. That is exactly the kind of quiet wrongness
+ * this phase must not have.
+ */
+function isDuplicateSha(error: unknown): boolean {
+  const e = error as { status?: number; data?: { data?: Record<string, { code?: string }> } };
+  if (e?.status !== 400) return false;
+  return e?.data?.data?.sha?.code === "validation_not_unique";
+}
+
+/**
+ * Store a batch, skipping anything already present.
+ *
+ * Existing shas are read in one query rather than discovered through failed
+ * creates — a backfill would otherwise generate 787 deliberate errors. The
+ * unique-violation catch remains as a race backstop for the case where the
+ * webhook and the hourly poll process the same commit concurrently, which the
+ * pre-read cannot prevent.
+ */
+export async function storeCommits(
+  projectId: string,
+  commits: ForgejoCommit[]
+): Promise<{ created: number; skipped: number }> {
+  if (commits.length === 0) return { created: 0, skipped: 0 };
+
+  const pb = await getAdminPb();
+
+  const filter = commits.map((c) => `sha = "${c.sha}"`).join(" || ");
+  const existing = await pb.collection("commits").getFullList({
+    filter: `project = "${projectId}" && (${filter})`,
+    fields: "sha",
+  });
+  const known = new Set(existing.map((r) => r.sha as string));
+
+  let created = 0;
+  let skipped = 0;
+
+  for (const commit of commits) {
+    if (known.has(commit.sha)) {
+      skipped += 1;
+      continue;
+    }
+
+    try {
+      await pb.collection("commits").create(toRecord(projectId, commit));
+      created += 1;
+    } catch (error: unknown) {
+      if (isDuplicateSha(error)) {
+        skipped += 1;
+        continue;
+      }
+      // Anything else is a real problem. Let it surface: the webhook returns
+      // 500 so Forgejo shows the delivery as failed, and the backfill stops
+      // rather than reporting a total that quietly omits rows.
+      throw error;
+    }
+  }
+
+  return { created, skipped };
+}
+
+/**
+ * Walk a repository from the newest commit backwards.
+ *
+ * `stopAfterKnownPages` decides the mode:
+ *   - backfill passes Infinity and reads to the end
+ *   - reconciliation passes 1 and stops once a whole page is already known
+ *
+ * The Forgejo commits endpoint has no `since` parameter, so "what is new" is
+ * necessarily expressed as "walk newest-first until everything is familiar".
+ * Stopping on a whole page rather than on the first known sha tolerates a
+ * commit arriving out of order between the webhook and the poll.
+ */
+export async function ingestProject(
+  project: Pick<ProjectRecord, "id" | "repo_owner" | "repo_name">,
+  options: { stopAfterKnownPages: number } = { stopAfterKnownPages: 1 }
+): Promise<IngestResult> {
+  if (!project.repo_owner || !project.repo_name) {
+    throw new Error("This project has no repository linked");
+  }
+
+  let page = 1;
+  let created = 0;
+  let skipped = 0;
+  let consecutiveKnownPages = 0;
+
+  for (;;) {
+    const result = await fetchCommitPage(
+      project.repo_owner,
+      project.repo_name,
+      page
+    );
+
+    if (result.commits.length === 0) break;
+
+    const stored = await storeCommits(project.id, result.commits);
+    created += stored.created;
+    skipped += stored.skipped;
+
+    if (stored.created === 0) {
+      consecutiveKnownPages += 1;
+      if (consecutiveKnownPages >= options.stopAfterKnownPages) break;
+    } else {
+      consecutiveKnownPages = 0;
+    }
+
+    if (!result.hasMore || page >= result.pageCount) break;
+    page += 1;
+  }
+
+  const pb = await getAdminPb();
+  await pb
+    .collection("projects")
+    .update(project.id, { last_ingest_at: new Date().toISOString() });
+
+  return { created, skipped, pagesRead: page };
+}
+
+/** Every project with a repository linked. Used by the hourly reconciliation. */
+export async function projectsWithRepos(): Promise<ProjectRecord[]> {
+  const pb = await getAdminPb();
+  return pb.collection("projects").getFullList<ProjectRecord>({
+    filter: 'repo_owner != "" && repo_name != "" && status != "archived"',
+  });
+}