treewide: make less webhook-centric

Signed-off-by: NotAShelf <raf@notashelf.dev>
Change-Id: Ifab58fcb523549ca9cb83dc8467be51e6a6a6964

src/polling.ts

@@ -0,0 +1,223 @@
+import type { Config, WebhookEvent } from './types.js';
+import {
+  listRecentComments,
+  fetchPR,
+  fetchIssue,
+  hasExistingComment,
+  postComment,
+  updateComment,
+  formatComment,
+  type RecentComment,
+} from './github.js';
+import { createEngine } from './engine/index.js';
+import { getLogger } from './logger.js';
+import { recordEvent } from './events.js';
+
+interface ProcessedComment {
+  id: number;
+  timestamp: number;
+}
+
+const processedComments: Map<string, ProcessedComment> = new Map();
+const MAX_PROCESSED_CACHE = 1000;
+
+function getCacheKey(owner: string, repo: string, commentId: number): string {
+  return `${owner}/${repo}#${commentId}`;
+}
+
+function isProcessed(owner: string, repo: string, commentId: number): boolean {
+  return processedComments.has(getCacheKey(owner, repo, commentId));
+}
+
+function markProcessed(owner: string, repo: string, commentId: number): void {
+  const key = getCacheKey(owner, repo, commentId);
+  processedComments.set(key, { id: commentId, timestamp: Date.now() });
+
+  // Clean up old entries if cache is too large
+  if (processedComments.size > MAX_PROCESSED_CACHE) {
+    const entries = Array.from(processedComments.entries());
+    entries.sort((a, b) => a[1].timestamp - b[1].timestamp);
+    const toRemove = entries.slice(0, entries.length - MAX_PROCESSED_CACHE);
+    for (const [k] of toRemove) {
+      processedComments.delete(k);
+    }
+  }
+}
+
+function containsMention(body: string): boolean {
+  return body.includes('@troutbot');
+}
+
+async function analyzeAndComment(
+  event: WebhookEvent,
+  config: Config
+): Promise<Record<string, unknown>> {
+  const logger = getLogger();
+  const engine = createEngine(config.engine);
+
+  // Run analysis
+  const analysis = await engine.analyze(event);
+  logger.info(
+    `Analyzed ${event.owner}/${event.repo}#${event.number}: impact=${analysis.impact}, confidence=${analysis.confidence.toFixed(2)}`
+  );
+
+  // Check for existing comment
+  const { commentMarker, allowUpdates } = config.response;
+  const existing = await hasExistingComment(event.owner, event.repo, event.number, commentMarker);
+
+  if (existing.exists && !allowUpdates) {
+    logger.info(`Already commented on ${event.owner}/${event.repo}#${event.number}, skipping`);
+    const result = { skipped: true, reason: 'Already commented' };
+    recordEvent(event, result, analysis);
+    return result;
+  }
+
+  const body = formatComment(
+    config.response,
+    event.type,
+    analysis.impact,
+    analysis.confidence,
+    analysis.reasoning
+  );
+
+  if (existing.exists && allowUpdates && existing.commentId) {
+    logger.info(`Updating existing comment on ${event.owner}/${event.repo}#${event.number}`);
+    await updateComment(event.owner, event.repo, existing.commentId, body);
+  } else {
+    await postComment(event.owner, event.repo, event.number, body);
+  }
+
+  const result = { processed: true, impact: analysis.impact, confidence: analysis.confidence };
+  recordEvent(event, result, analysis);
+  return result;
+}
+
+async function processComment(
+  comment: RecentComment,
+  owner: string,
+  repo: string,
+  config: Config
+): Promise<void> {
+  const logger = getLogger();
+
+  if (!containsMention(comment.body)) {
+    return;
+  }
+
+  if (isProcessed(owner, repo, comment.id)) {
+    logger.debug(`Comment ${owner}/${repo}#${comment.id} already processed, skipping`);
+    return;
+  }
+
+  logger.info(`Found @troutbot mention in ${owner}/${repo}#${comment.issueNumber}`);
+
+  try {
+    // First, try to fetch as a PR to check if it's a pull request
+    const prData = await fetchPR(owner, repo, comment.issueNumber);
+
+    let event: WebhookEvent;
+
+    if (prData) {
+      // It's a pull request
+      event = {
+        action: 'on_demand',
+        type: 'pull_request',
+        number: comment.issueNumber,
+        title: prData.title,
+        body: prData.body,
+        owner,
+        repo,
+        author: prData.author,
+        labels: prData.labels,
+        branch: prData.branch,
+        sha: prData.sha,
+      };
+    } else {
+      // It's an issue
+      const issueData = await fetchIssue(owner, repo, comment.issueNumber);
+      if (!issueData) {
+        logger.warn(`Could not fetch issue ${owner}/${repo}#${comment.issueNumber}`);
+        return;
+      }
+
+      event = {
+        action: 'on_demand',
+        type: 'issue',
+        number: comment.issueNumber,
+        title: issueData.title,
+        body: issueData.body,
+        owner,
+        repo,
+        author: issueData.author,
+        labels: issueData.labels,
+      };
+    }
+
+    await analyzeAndComment(event, config);
+    markProcessed(owner, repo, comment.id);
+
+    logger.info(
+      `Successfully processed on-demand analysis for ${owner}/${repo}#${comment.issueNumber}`
+    );
+  } catch (err) {
+    logger.error(`Failed to process mention in ${owner}/${repo}#${comment.issueNumber}`, err);
+  }
+}
+
+async function pollRepository(
+  owner: string,
+  repo: string,
+  config: Config,
+  since: Date
+): Promise<void> {
+  const logger = getLogger();
+
+  try {
+    const comments = await listRecentComments(owner, repo, since);
+    logger.debug(`Fetched ${comments.length} recent comments from ${owner}/${repo}`);
+
+    for (const comment of comments) {
+      await processComment(comment, owner, repo, config);
+    }
+  } catch (err) {
+    logger.error(`Failed to poll ${owner}/${repo}`, err);
+  }
+}
+
+export async function startPolling(config: Config): Promise<void> {
+  const logger = getLogger();
+  const pollingConfig = config.polling;
+
+  if (!pollingConfig || !pollingConfig.enabled) {
+    logger.info('Polling is disabled');
+    return;
+  }
+
+  if (config.repositories.length === 0) {
+    logger.warn('Polling enabled but no repositories configured');
+    return;
+  }
+
+  const intervalMs = pollingConfig.intervalMinutes * 60 * 1000;
+  const lookbackMs = pollingConfig.lookbackMinutes * 60 * 1000;
+
+  logger.info(`Starting polling for ${config.repositories.length} repositories`);
+  logger.info(
+    `Poll interval: ${pollingConfig.intervalMinutes} minutes, lookback: ${pollingConfig.lookbackMinutes} minutes`
+  );
+
+  // Do an initial poll
+  const initialSince = new Date(Date.now() - lookbackMs);
+  for (const repo of config.repositories) {
+    await pollRepository(repo.owner, repo.repo, config, initialSince);
+  }
+
+  // Set up recurring polling
+  setInterval(async () => {
+    const since = new Date(Date.now() - lookbackMs);
+
+    for (const repo of config.repositories) {
+      await pollRepository(repo.owner, repo.repo, config, since);
+    }
+  }, intervalMs);
+}