refactor(sync,scripts): harden exhausted-job cli and share filter

Use a true exhausted count with truncation signal on list, reject clear
on non-failed jobs, and check remove success. Share the exhausted-job
Mongo filter between count/list helpers.

Co-authored-by: Tyler Dane <tyler-dane@users.noreply.github.com>

Author: cursoragent

packages/scripts/src/commands/manage-failed-jobs.ts

@@ -25,10 +25,6 @@ function flagValue(args: string[], name: string): string | undefined {
   return args[index + 1];
 }
 
-function hasFlag(args: string[], name: string): boolean {
-  return args.includes(name);
-}
-
 /**
  * Operator tooling for Sync jobs that exhausted the self-heal requeue budget
  * and still occupy a coalescing key. Default actions are dry-run; `--apply`
@@ -47,7 +43,7 @@ export async function runManageFailedJobs(): Promise<void> {
     );
   }
 
-  const apply = hasFlag(args, "--apply");
+  const apply = args.includes("--apply");
   const syncMongo = new SyncMongoService();
   try {
     await syncMongo.connect({
@@ -58,11 +54,16 @@ export async function runManageFailedJobs(): Promise<void> {
     const jobs = new JobRepository(syncMongo.db);
 
     if (action === "list") {
-      const exhausted = await jobs.listExhaustedFailed(FAILED_JOB_MAX_REQUEUES);
+      const [count, sample] = await Promise.all([
+        jobs.countExhaustedFailed(FAILED_JOB_MAX_REQUEUES),
+        jobs.listExhaustedFailed(FAILED_JOB_MAX_REQUEUES),
+      ]);
       const report = {
         maxRequeues: FAILED_JOB_MAX_REQUEUES,
-        count: exhausted.length,
-        jobs: exhausted.map((job) => ({
+        count,
+        sampleSize: sample.length,
+        truncated: count > sample.length,
+        jobs: sample.map((job) => ({
           id: job.id,
           coalescingKey: job.coalescingKey,
           connectionId: job.connectionId,
@@ -73,7 +74,7 @@ export async function runManageFailedJobs(): Promise<void> {
       };
       process.stdout.write(`${JSON.stringify(report, null, 2)}\n`);
       logger.info(
-        `manage-failed-jobs list count=${report.count} maxRequeues=${FAILED_JOB_MAX_REQUEUES}`,
+        `manage-failed-jobs list count=${report.count} sample=${report.sampleSize} truncated=${report.truncated}`,
       );
       await syncMongo.disconnect();
       process.exit(0);
@@ -88,6 +89,11 @@ export async function runManageFailedJobs(): Promise<void> {
     if (!existing) {
       throw new Error(`No job found for id=${id}`);
     }
+    if (existing.state !== "failed") {
+      throw new Error(
+        `Job id=${id} is state=${existing.state}; ${action} only accepts failed jobs`,
+      );
+    }
 
     if (action === "clear") {
       const coalescingKey =
@@ -99,13 +105,16 @@ export async function runManageFailedJobs(): Promise<void> {
       }
       const report = {
         dryRun: !apply,
-        action: "clear",
+        action: "clear" as const,
         id,
         coalescingKey,
         state: existing.state,
       };
       if (apply) {
-        await jobs.remove(id, coalescingKey);
+        const ok = await jobs.remove(id, coalescingKey);
+        if (!ok) {
+          throw new Error(`Failed to clear id=${id} key=${coalescingKey}`);
+        }
       }
       process.stdout.write(`${JSON.stringify(report, null, 2)}\n`);
       logger.info(
@@ -115,15 +124,9 @@ export async function runManageFailedJobs(): Promise<void> {
       process.exit(0);
     }
 
-    // requeue
-    if (existing.state !== "failed") {
-      throw new Error(
-        `Job id=${id} is state=${existing.state}; requeue only accepts failed jobs`,
-      );
-    }
     const report = {
       dryRun: !apply,
-      action: "requeue",
+      action: "requeue" as const,
       id,
       coalescingKey: existing.coalescingKey,
       requeuedCount: existing.requeuedCount,

packages/sync/src/storage/repositories/job.repository.ts

@@ -22,6 +22,14 @@ export type ExhaustedFailedJob = {
   updatedAt: Date;
 };
 
+function exhaustedFailedFilter(maxRequeues: number) {
+  return {
+    state: "failed" as const,
+    failureClass: { $ne: "permanent" as const },
+    requeuedCount: { $gte: maxRequeues },
+  };
+}
+
 // Repository for `jobs`. Enqueue coalesces on a unique key so repeated
 // notifications for the same resource collapse into one pending job instead of
 // an unbounded queue. Terminal jobs are removed so a later notification can
@@ -283,11 +291,7 @@ export class JobRepository {
   // the sweep will not touch them again; an operator must. Used to drive a
   // loud, recurring alert rather than a silent terminal state.
   async countExhaustedFailed(maxRequeues: number): Promise<number> {
-    return this.collection.countDocuments({
-      state: "failed",
-      failureClass: { $ne: "permanent" },
-      requeuedCount: { $gte: maxRequeues },
-    });
+    return this.collection.countDocuments(exhaustedFailedFilter(maxRequeues));
   }
 
   // Same filter as countExhaustedFailed, returning the rows an operator needs
@@ -297,11 +301,7 @@ export class JobRepository {
     limit = 50,
   ): Promise<ExhaustedFailedJob[]> {
     const rows = await this.collection
-      .find({
-        state: "failed",
-        failureClass: { $ne: "permanent" },
-        requeuedCount: { $gte: maxRequeues },
-      })
+      .find(exhaustedFailedFilter(maxRequeues))
       .sort({ updatedAt: 1 })
       .limit(limit)
       .project({
@@ -315,7 +315,7 @@ export class JobRepository {
       .toArray();
 
     return rows.map((row) => ({
-      id: row._id as SyncJobId,
+      id: row["_id"] as SyncJobId,
       coalescingKey: String(row["coalescingKey"]),
       connectionId: row["connectionId"] as ConnectionId,
       failureClass: (row["failureClass"] ?? "retryableTransient") as Exclude<
@@ -351,9 +351,7 @@ export class JobRepository {
       tenantId,
       principalId,
       connectionId,
-      state: "failed",
-      failureClass: { $ne: "permanent" },
-      requeuedCount: { $gte: maxRequeues },
+      ...exhaustedFailedFilter(maxRequeues),
     });
   }
 

packages/web/src/events/grid-event-draft.adapter.test.ts

@@ -1,3 +1,4 @@
+import { Origin } from "@core/constants/core.constants";
 import {
   type Calendar,
   getCalendarCapabilities,
@@ -150,17 +151,17 @@ test("projects a grid draft into a grid event without a CompassEvent bridge", ()
   expect(allDayGrid.isAllDay).toBe(true);
   expect(allDayGrid.title).toBe("All day");
   expect(allDayGrid.calendarId).toBe(timedEvent.calendarId);
-  expect(allDayGrid.origin).toBe("compass");
+  expect(allDayGrid.origin).toBe(Origin.COMPASS);
 
   const timed = editGridEventDraft(timedEvent);
   if (!timed) throw new Error("Expected timed edit draft");
-  timed.values.color = "tomato";
+  timed.values.color = "coral";
   const timedGrid = gridEventDraftToGridEvent(timed);
 
   expect(timedGrid.isAllDay).toBe(false);
   expect(timedGrid.startDate).toBe(dayjs(timed.values.schedule.start).format());
   expect(timedGrid.endDate).toBe(dayjs(timed.values.schedule.end).format());
-  expect(timedGrid.color).toBe("tomato");
+  expect(timedGrid.color).toBe("coral");
   expect(timedGrid.isBusy).toBe(false);
 });