[Feat] Notify task threads when pull requests conflict

apps/api/src/handlers/github/index.ts

@@ -40,6 +40,10 @@ import { queuePrCiFailureNotification } from './notifyPrCiFailure';
 // Conflict Resolution:
 import { handlePushConflictCheck } from './handlePushConflictCheck';
 import { handleWorkflowRunCompleted } from './handleWorkflowRunCompleted';
+import {
+  queueBaseBranchMergeabilityCheck,
+  queueTrackedPullRequestMergeabilityCheck,
+} from './queuePullRequestMergeabilityCheck';
 
 // Repository metadata sync:
 import { handleRepositoryEdited } from './handleRepositoryEdited';
@@ -56,8 +60,8 @@ function syncPrStatus(
   repo: string,
   prNumber: number,
   status: PullRequestStatus,
-): void {
-  updateTaskPrStatus('github', repo, prNumber, status).catch((error) =>
+): Promise<void> {
+  return updateTaskPrStatus('github', repo, prNumber, status).catch((error) =>
     console.warn(
       `[syncPrStatus] Failed to update PR status for ${repo}#${prNumber}: ${
         error instanceof Error ? error.message : String(error)
@@ -300,7 +304,7 @@ github.post('/', async (c) => {
 
     webhooks.on('pull_request.opened', ({ id, name, payload }) =>
       recordWebhook(id, `${name}.${payload.action}`, payload, async () => {
-        syncPrStatus(
+        await syncPrStatus(
           payload.repository.full_name,
           payload.pull_request.number,
           payload.pull_request.draft ? 'draft' : 'open',
@@ -324,6 +328,7 @@ github.post('/', async (c) => {
             url: payload.pull_request.html_url,
           },
         });
+        await queueTrackedPullRequestMergeabilityCheck(payload);
 
         if (isRepoSkipped(payload.repository.full_name)) {
           return {
@@ -338,7 +343,7 @@ github.post('/', async (c) => {
 
     webhooks.on('pull_request.reopened', ({ id, name, payload }) =>
       recordWebhook(id, `${name}.${payload.action}`, payload, async () => {
-        syncPrStatus(
+        await syncPrStatus(
           payload.repository.full_name,
           payload.pull_request.number,
           'open',
@@ -362,6 +367,7 @@ github.post('/', async (c) => {
             url: payload.pull_request.html_url,
           },
         });
+        await queueTrackedPullRequestMergeabilityCheck(payload);
 
         if (isRepoSkipped(payload.repository.full_name)) {
           return {
@@ -395,6 +401,7 @@ github.post('/', async (c) => {
             url: payload.pull_request.html_url,
           },
         });
+        await queueTrackedPullRequestMergeabilityCheck(payload);
 
         if (isRepoSkipped(payload.repository.full_name)) {
           return {
@@ -407,13 +414,30 @@ github.post('/', async (c) => {
       }),
     );
 
+    webhooks.on('pull_request.edited', ({ id, name, payload }) =>
+      recordWebhook(id, `${name}.${payload.action}`, payload, async () => {
+        if (!payload.changes.base) {
+          return { status: 'ok' as const };
+        }
+
+        await queueTrackedPullRequestMergeabilityCheck(payload, {
+          updateBaseRef: true,
+        });
+        return { status: 'ok' as const };
+      }),
+    );
+
     webhooks.on('pull_request.ready_for_review', ({ id, name, payload }) =>
       recordWebhook(id, `${name}.${payload.action}`, payload, async () => {
-        syncPrStatus(
+        // Awaited so the mergeability check below sees the row as 'open';
+        // conflicts accrued while the PR was a draft surface at this
+        // transition.
+        await syncPrStatus(
           payload.repository.full_name,
           payload.pull_request.number,
           'open',
         );
+        await queueTrackedPullRequestMergeabilityCheck(payload);
         syncPullRequestFact({
           githubRepoId: payload.repository.id,
           repositoryFullName: payload.repository.full_name,
@@ -523,7 +547,13 @@ github.post('/', async (c) => {
     );
 
     webhooks.on('push', ({ id, name, payload }) =>
-      recordWebhook(id, name, payload, () => handlePushConflictCheck(payload)),
+      recordWebhook(id, name, payload, async () => {
+        const [result] = await Promise.all([
+          handlePushConflictCheck(payload),
+          queueBaseBranchMergeabilityCheck(payload),
+        ]);
+        return result;
+      }),
     );
 
     webhooks.on('repository.edited', ({ id, name, payload }) =>

apps/api/src/handlers/github/queuePullRequestMergeabilityCheck.ts

@@ -0,0 +1,105 @@
+import {
+  listTrackedPullRequestsForMergeability,
+  updateTrackedPullRequestBaseRef,
+} from '@roomote/db/server';
+import { isRepoSkipped } from '@roomote/github';
+import { enqueuePullRequestMergeabilityCheck } from '@roomote/sdk/server';
+
+type PushPayload = {
+  ref: string;
+  installation?: { id: number } | null;
+  repository: { full_name: string };
+};
+
+type PullRequestPayload = {
+  installation?: { id: number } | null;
+  repository: { full_name: string };
+  pull_request: {
+    number: number;
+    base: { ref: string };
+  };
+};
+
+function logQueueFailure(context: string, error: unknown): void {
+  console.error(
+    `[queuePullRequestMergeabilityCheck] ${context}: ${
+      error instanceof Error ? error.message : String(error)
+    }`,
+  );
+}
+
+export async function queueBaseBranchMergeabilityCheck(
+  payload: PushPayload,
+): Promise<void> {
+  // Best effort: mergeability queueing must never fail the webhook delivery,
+  // which would also drop the co-located pre-existing handlers.
+  try {
+    const installationId = payload.installation?.id;
+    const refPrefix = 'refs/heads/';
+    if (!installationId || !payload.ref.startsWith(refPrefix)) return;
+    const repository = payload.repository.full_name;
+    if (isRepoSkipped(repository)) return;
+
+    const baseRef = payload.ref.slice(refPrefix.length);
+    const candidates = await listTrackedPullRequestsForMergeability({
+      repository,
+      baseRef,
+    });
+    if (candidates.length === 0) return;
+
+    await enqueuePullRequestMergeabilityCheck({
+      installationId,
+      repository,
+      baseRef,
+      deduplicationKey: `base:${repository}:${baseRef}`,
+      retryAttempt: 0,
+      allowNotifiedConflictCheck: true,
+    });
+  } catch (error) {
+    logQueueFailure(`push ${payload.repository.full_name}`, error);
+  }
+}
+
+export async function queueTrackedPullRequestMergeabilityCheck(
+  payload: PullRequestPayload,
+  options: { updateBaseRef?: boolean } = {},
+): Promise<void> {
+  // Best effort: see queueBaseBranchMergeabilityCheck.
+  try {
+    const installationId = payload.installation?.id;
+    if (!installationId) return;
+
+    const repository = payload.repository.full_name;
+    if (isRepoSkipped(repository)) return;
+    const prNumber = payload.pull_request.number;
+
+    // Only base-change edits need the eager write; the job re-syncs
+    // baseRefName from GitHub for every checked PR anyway.
+    if (options.updateBaseRef) {
+      await updateTrackedPullRequestBaseRef({
+        repository,
+        prNumber,
+        baseRef: payload.pull_request.base.ref,
+      });
+    }
+
+    // Enqueued unconditionally: the job resolves the tracked rows at run
+    // time, so an opened webhook that races the task's own PR persistence
+    // still gets its baseline check 45 seconds later.
+    await enqueuePullRequestMergeabilityCheck({
+      installationId,
+      repository,
+      prNumber,
+      deduplicationKey: `pr:${repository}:${prNumber}`,
+      retryAttempt: 0,
+      // A synchronize event is how a previously conflicting PR reports a
+      // conflict-resolution push, so it must be allowed to re-arm the state.
+      allowNotifiedConflictCheck: true,
+    });
+  } catch (error) {
+    logQueueFailure(
+      `pull_request ${payload.repository.full_name}#${payload.pull_request.number}`,
+      error,
+    );
+  }
+}