[Fix] PR review feedback disappears during pull request creation

apps/bullmq/src/jobs/pr-review-notification.ts

@@ -17,14 +17,16 @@ import {
   PR_REVIEW_NOTIFICATION_MAX_DEFERRALS,
   attachPendingPrReviewActionMessage,
   getCommunicationProviderAdapter,
-  type PrReviewNotificationRequest,
+  type PrReviewNotificationQueueRequest,
   type PrReviewNotificationRoute,
   consumePendingPrReviewActivity,
   dispatchPrReviewFollowUp,
   preparePrReviewNotificationDelivery,
   prReviewNotificationRequestSchema,
+  prReviewAssociationReplayRequestSchema,
   recordPrReviewNotificationDeliveryBestEffort,
   requeuePendingPrReviewActivity,
+  replayPrReviewNotificationAssociation,
   schedulePrReviewNotificationJob,
   setPendingPrReviewAction,
 } from '@roomote/sdk/server';
@@ -39,7 +41,11 @@ import {
   WORKER_HEARTBEAT_STALE_MS,
 } from '@roomote/types';
 
-type PrReviewNotificationJob = Job<PrReviewNotificationRequest, void, string>;
+type PrReviewNotificationJob = Job<
+  PrReviewNotificationQueueRequest,
+  void,
+  string
+>;
 
 function buildPrReviewNotificationPostInput(
   route: PrReviewNotificationRoute,
@@ -212,6 +218,13 @@ async function postPrReviewNotification({
 export const prReviewNotificationJob = async (
   job: PrReviewNotificationJob,
 ): Promise<void> => {
+  const replay = prReviewAssociationReplayRequestSchema.safeParse(job.data);
+
+  if (replay.success) {
+    await replayPrReviewNotificationAssociation(replay.data);
+    return;
+  }
+
   const parsed = prReviewNotificationRequestSchema.safeParse(job.data);
 
   if (!parsed.success) {
@@ -225,6 +238,7 @@ export const prReviewNotificationJob = async (
     taskId: data.taskId,
     repository: data.repository,
     prNumber: data.prNumber,
+    sourceControlProvider: data.sourceControlProvider ?? ('github' as const),
     ...(data.batchKind ? { batchKind: data.batchKind } : {}),
     ...(data.batchId ? { batchId: data.batchId } : {}),
     ...(data.immediate ? { immediate: true } : {}),

apps/bullmq/src/pr-review-notification-queue.ts

@@ -2,16 +2,20 @@ import { Queue, QueueEvents, Worker } from 'bullmq';
 
 import {
   PR_REVIEW_NOTIFICATION_QUEUE_NAME,
-  type PrReviewNotificationRequest,
+  type PrReviewNotificationQueueRequest,
 } from '@roomote/sdk/server';
 
 import { prReviewNotificationJob } from './jobs/pr-review-notification';
 import { getRedis } from './redis';
 
+function formatJobTarget(data: PrReviewNotificationQueueRequest): string {
+  return `${data.repository}#${data.prNumber}`;
+}
+
 export function startPrReviewNotificationQueue() {
   const connection = getRedis();
 
-  const queue = new Queue<PrReviewNotificationRequest, void, string>(
+  const queue = new Queue<PrReviewNotificationQueueRequest, void, string>(
     PR_REVIEW_NOTIFICATION_QUEUE_NAME,
     {
       connection,
@@ -24,15 +28,17 @@ export function startPrReviewNotificationQueue() {
     },
   );
 
-  const worker = new Worker<PrReviewNotificationRequest, void, string>(
+  const worker = new Worker<PrReviewNotificationQueueRequest, void, string>(
     PR_REVIEW_NOTIFICATION_QUEUE_NAME,
     prReviewNotificationJob,
     { connection, concurrency: 5, autorun: true },
   );
 
   worker.on('failed', (job, err) =>
     console.error(
-      `[PrReviewNotificationQueue] job ${job?.id} failed for ${job?.data.repository}#${job?.data.prNumber}:`,
+      `[PrReviewNotificationQueue] job ${job?.id} failed for ${
+        job?.data ? formatJobTarget(job.data) : 'unknown pull request'
+      }:`,
       err.message,
     ),
   );

apps/bullmq/src/scheduled-jobs/index.ts

@@ -6,3 +6,4 @@ export { instancePingJob } from './instance-ping';
 export { licenseUsageSyncJob } from './license-usage-sync';
 export { webhookCleanupJob } from './webhook-cleanup';
 export { standbyRetentionJob } from './standby-retention';
+export { prReviewAssociationRepairJob } from './pr-review-association-repair';

apps/bullmq/src/scheduled-jobs/pr-review-association-repair.ts

@@ -0,0 +1,5 @@
+import { repairOrphanPrReviewAssociationReplays } from '@roomote/sdk/server';
+
+export async function prReviewAssociationRepairJob(): Promise<void> {
+  await repairOrphanPrReviewAssociationReplays();
+}

apps/bullmq/src/scheduler.ts

@@ -30,6 +30,7 @@ import {
   licenseUsageSyncJob,
   webhookCleanupJob,
   standbyRetentionJob,
+  prReviewAssociationRepairJob,
 } from './scheduled-jobs';
 
 const QUEUE_NAME = 'scheduled-jobs';
@@ -88,6 +89,10 @@ async function createJobs(queue: Queue): Promise<void> {
     { every: 60 * 1000 }, // Every 60 seconds.
   );
 
+  await queue.upsertJobScheduler(ScheduledJobName.PrReviewAssociationRepair, {
+    every: 60 * 1000,
+  });
+
   await queue.upsertJobScheduler(
     ScheduledJobName.StandbyRetention,
     { every: 5 * 60 * 1000 }, // Every 5 minutes.
@@ -216,6 +221,8 @@ const runJobs = async (job: ScheduledJob): Promise<void> => {
       return webhookCleanupJob();
     case ScheduledJobName.StandbyRetention:
       return standbyRetentionJob();
+    case ScheduledJobName.PrReviewAssociationRepair:
+      return prReviewAssociationRepairJob();
     case ScheduledJobName.CustomAutomations:
       await customAutomationsJob();
       return;

apps/bullmq/src/types.ts

@@ -14,6 +14,7 @@ export enum ScheduledJobName {
   WebhookCleanup = 'WebhookCleanup',
   StandbyRetention = 'StandbyRetention',
   CustomAutomations = 'custom_automations',
+  PrReviewAssociationRepair = 'PrReviewAssociationRepair',
 }
 
 /**

packages/sdk/src/server/index.ts

@@ -197,21 +197,30 @@ export {
   PR_REVIEW_NOTIFICATION_MAX_DEFERRALS,
   PR_REVIEW_NOTIFICATION_QUEUE_NAME,
   PR_REVIEW_NOTIFICATION_ROOMOTE_FALLBACK_MS,
+  PR_REVIEW_ASSOCIATION_REPLAY_DELAYS_MS,
+  ORPHAN_REPLAY_REPAIR_DELAY_MS,
+  wakePrReviewNotificationAssociation,
   consumePendingPrReviewActivity,
   enqueuePrReviewNotification,
   enqueuePrReviewNotificationInputSchema,
   formatPrReviewActivityMessage,
   getPrReviewCompletedCycleKey,
   hasPrReviewNotificationThreadContext,
   prReviewActivityEventSchema,
+  prReviewAssociationReplayRequestSchema,
+  prReviewNotificationQueueRequestSchema,
   prReviewNotificationRequestSchema,
   requeuePendingPrReviewActivity,
+  replayPrReviewNotificationAssociation,
+  repairOrphanPrReviewAssociationReplays,
   resolvePrReviewNotificationRoute,
   schedulePrReviewNotificationJob,
   startPrReviewNotificationCycle,
   startPrReviewNotificationCycleInputSchema,
   type EnqueuePrReviewNotificationInput,
   type PrReviewActivityEvent,
+  type PrReviewAssociationReplayRequest,
+  type PrReviewNotificationQueueRequest,
   type PrReviewNotificationRequest,
   type PrReviewNotificationRoute,
   type StartPrReviewNotificationCycleInput,