pr-monitor/server/poller.ts

import { createHash } from 'node:crypto';
import type { PrSnapshot } from '../shared/types.js';
import type { AppConfig } from './config.js';
import type { Db } from './db.js';
import { categorize } from './categorize.js';
import { deriveReviewState, deriveVerdicts } from './reviewState.js';
import { graphql } from './github/client.js';
import {
  DETAIL_QUERY,
  LIST_QUERY,
  type DetailPr,
  type DetailResult,
  type ListPr,
  type ListResult,
} from './github/queries.js';

const DETAIL_BATCH = 10;
const ROLLING_REFRESH_MS = 15 * 60 * 1000;

export interface PollResult {
  changedPrIds: string[];
  newEventCount: number;
  prCount: number;
}

function signature(pr: ListPr): string {
  const resolved = pr.reviewThreads.nodes.filter((t) => t.isResolved).length;
  return createHash('sha1')
    .update(
      JSON.stringify([
        pr.updatedAt,
        pr.headRefOid,
        pr.commits.nodes[0]?.commit.statusCheckRollup?.state ?? null,
        pr.reviewThreads.totalCount,
        resolved,
        pr.comments.totalCount,
        pr.reviews.totalCount,
        pr.latestReviews.nodes.map((r) => `${r.author?.login}:${r.state}`),
        pr.labels.nodes.map((l) => l.name).sort(),
        pr.mergeable,
      ]),
    )
    .digest('hex');
}

async function fetchList(config: AppConfig): Promise<ListPr[]> {
  const prs: ListPr[] = [];
  let cursor: string | null = null;
  do {
    const data: ListResult = await graphql<ListResult>(LIST_QUERY, {
      q: config.searchQuery,
      cursor,
    });
    for (const node of data.search.nodes) {
      if ('id' in node) prs.push(node as ListPr);
    }
    cursor = data.search.pageInfo.hasNextPage ? data.search.pageInfo.endCursor : null;
  } while (cursor);
  return prs;
}

function buildSnapshot(config: AppConfig, pr: ListPr, detail: DetailPr | undefined, db: Db): PrSnapshot {
  const repo = pr.repository.nameWithOwner;
  const rules = config.rulesByRepo.get(repo);
  let categories;
  if (detail) {
    const contexts = detail.commits.nodes[0]?.commit.statusCheckRollup?.contexts.nodes ?? [];
    categories = categorize(rules, contexts);
  } else {
    // Detail node inaccessible this round — keep the previous check breakdown.
    categories = db.getSnapshot(pr.id)?.categories ?? categorize(rules, []);
  }
  const labels = pr.labels.nodes.map((l) => l.name);
  const previewUrl =
    rules?.preview && labels.includes(rules.preview.label)
      ? rules.preview.urlTemplate.replaceAll('{number}', String(pr.number))
      : null;
  return {
    id: pr.id,
    repo,
    number: pr.number,
    title: pr.title,
    url: pr.url,
    author: pr.author?.login ?? 'ghost',
    isDraft: pr.isDraft,
    mergeable: pr.mergeable,
    branch: pr.headRefName,
    updatedAt: pr.updatedAt,
    labels,
    previewUrl,
    ...deriveReviewState(pr),
    verdicts: deriveVerdicts(pr),
    categories,
  };
}

export async function runPoll(config: AppConfig, db: Db): Promise<PollResult> {
  const now = new Date().toISOString();
  const listPrs = await fetchList(config);
  const known = db.getPrSignatures();

  const changed: ListPr[] = [];
  for (const pr of listPrs) {
    const prev = known.get(pr.id);
    const stale =
      !prev?.detailSyncedAt || Date.now() - Date.parse(prev.detailSyncedAt) > ROLLING_REFRESH_MS;
    if (!prev || prev.signature !== signature(pr) || stale) changed.push(pr);
  }

  const present = new Set(listPrs.map((p) => p.id));
  db.markGone([...known.keys()].filter((id) => !present.has(id)));

  const detailById = new Map<string, DetailPr>();
  for (let i = 0; i < changed.length; i += DETAIL_BATCH) {
    const ids = changed.slice(i, i + DETAIL_BATCH).map((p) => p.id);
    const data = await graphql<DetailResult>(DETAIL_QUERY, { ids });
    for (const node of data.nodes) {
      if (node?.id) detailById.set(node.id, node);
    }
  }

  const backfillAsRead =
    db.getMeta('first_sync_done') !== '1' && process.env.BACKFILL_UNREAD !== '1';
  let newEventCount = 0;
  for (const pr of changed) {
    const detail = detailById.get(pr.id);
    const snapshot = buildSnapshot(config, pr, detail, db);
    db.upsertPr(snapshot, signature(pr), detail ? now : null);
    if (detail) {
      newEventCount += ingest(db, detail, config.user, backfillAsRead);
      await syncFailureReports(
        db,
        config.rulesByRepo.get(snapshot.repo),
        snapshot.repo,
        snapshot,
      );
    }
  }

  db.setMeta('first_sync_done', '1');
  db.setMeta('last_sync_at', now);
  db.setMeta('last_sync_error', null);
  return { changedPrIds: changed.map((p) => p.id), newEventCount, prCount: listPrs.length };
}

// Thin wrapper so poller stays the single import for index.ts / sync-once.
import { ingestTimeline } from './timeline.js';
import { syncFailureReports } from './failures.js';
function ingest(db: Db, detail: DetailPr, user: string, backfillAsRead: boolean): number {
  return ingestTimeline(db, detail, { user, backfillAsRead });
}