refactor(threads): dedup maintenance threads by stable key via index

packages/backend/convex/threads.ts

@@ -387,19 +387,20 @@ export const findOpenMaintenanceThread = internalQuery({
   args: {
     spoonId: v.id('spoons'),
     ownerId: v.id('users'),
-    upstreamTo: v.string(),
+    dedupKey: v.string(),
   },
-  handler: async (ctx, { spoonId, ownerId, upstreamTo }) => {
+  handler: async (ctx, { spoonId, ownerId, dedupKey }) => {
     const threads = await ctx.db
       .query('threads')
-      .withIndex('by_spoon', (q) => q.eq('spoonId', spoonId))
+      .withIndex('by_spoon_dedup', (q) =>
+        q.eq('spoonId', spoonId).eq('dedupKey', dedupKey),
+      )
       .order('desc')
       .collect();
     return (
       threads.find(
         (thread) =>
           thread.ownerId === ownerId &&
-          thread.upstreamTo === upstreamTo &&
           !['resolved', 'ignored', 'failed', 'cancelled'].includes(
             thread.status,
           ),
@@ -416,7 +417,8 @@ export const createMaintenanceThread = internalMutation({
     title: v.string(),
     summary: v.string(),
     upstreamFrom: v.optional(v.string()),
-    upstreamTo: v.string(),
+    upstreamTo: v.optional(v.string()),
+    dedupKey: v.string(),
     forkHeadAtCreation: v.optional(v.string()),
     mergeBaseAtCreation: v.optional(v.string()),
     relatedSyncRunId: v.optional(v.id('syncRuns')),
@@ -429,14 +431,15 @@ export const createMaintenanceThread = internalMutation({
     const now = Date.now();
     const existing = await ctx.db
       .query('threads')
-      .withIndex('by_spoon', (q) => q.eq('spoonId', args.spoonId))
+      .withIndex('by_spoon_dedup', (q) =>
+        q.eq('spoonId', args.spoonId).eq('dedupKey', args.dedupKey),
+      )
       .order('desc')
       .collect()
       .then((threads) =>
         threads.find(
           (thread) =>
             thread.ownerId === args.ownerId &&
-            thread.upstreamTo === args.upstreamTo &&
             !['resolved', 'ignored', 'failed', 'cancelled'].includes(
               thread.status,
             ),
@@ -470,6 +473,7 @@ export const createMaintenanceThread = internalMutation({
       priority: args.source === 'merge_conflict' ? 'high' : 'normal',
       upstreamFrom: args.upstreamFrom,
       upstreamTo: args.upstreamTo,
+      dedupKey: args.dedupKey,
       forkHeadAtCreation: args.forkHeadAtCreation,
       mergeBaseAtCreation: args.mergeBaseAtCreation,
       relatedSyncRunId: args.relatedSyncRunId,

packages/backend/tests/unit/maintenance-thread.test.ts

@@ -0,0 +1,146 @@
+import { convexTest } from 'convex-test';
+import { describe, expect, test } from 'vitest';
+
+import type { Id } from '../../convex/_generated/dataModel.js';
+import { internal } from '../../convex/_generated/api.js';
+import schema from '../../convex/schema';
+
+const modules = import.meta.glob('../../convex/**/*.*s');
+
+const seedSpoon = async (t: ReturnType<typeof convexTest>) =>
+  await t.mutation(async (ctx) => {
+    const now = Date.now();
+    const ownerId = await ctx.db.insert('users', {
+      email: 'maintenance-thread@example.com',
+      name: 'Maintenance Thread Owner',
+    });
+    const spoonId = await ctx.db.insert('spoons', {
+      ownerId,
+      name: 'Maintenance Thread Spoon',
+      provider: 'github',
+      upstreamOwner: 'upstream',
+      upstreamRepo: 'editor',
+      upstreamDefaultBranch: 'main',
+      upstreamUrl: 'https://github.com/upstream/editor',
+      forkOwner: 'team',
+      forkRepo: 'editor-spoon',
+      forkUrl: 'https://github.com/team/editor-spoon',
+      visibility: 'private',
+      maintenanceMode: 'watch',
+      syncCadence: 'daily',
+      productionRefStrategy: 'default_branch',
+      status: 'active',
+      createdAt: now,
+      updatedAt: now,
+    });
+    return { ownerId, spoonId };
+  });
+
+const baseArgs = (
+  ownerId: Id<'users'>,
+  spoonId: Id<'spoons'>,
+  dedupKey: string,
+) => ({
+  spoonId,
+  ownerId,
+  source: 'upstream_update' as const,
+  title: 'Upstream maintenance review',
+  summary: 'Upstream advanced; review recommended.',
+  upstreamTo: 'headsha',
+  dedupKey,
+  jobType: 'maintenance_review' as const,
+});
+
+const countSpoonThreads = async (
+  t: ReturnType<typeof convexTest>,
+  spoonId: Id<'spoons'>,
+) =>
+  await t.run(async (ctx) => {
+    const threads = await ctx.db.query('threads').collect();
+    return threads.filter((thread) => thread.spoonId === spoonId).length;
+  });
+
+describe('createMaintenanceThread dedup by key', () => {
+  test('reuses an open thread when called twice with the same dedupKey', async () => {
+    const t = convexTest(schema, modules);
+    const { ownerId, spoonId } = await seedSpoon(t);
+
+    const first = await t.mutation(
+      internal.threads.createMaintenanceThread,
+      baseArgs(ownerId, spoonId, 'base:head'),
+    );
+    const second = await t.mutation(
+      internal.threads.createMaintenanceThread,
+      baseArgs(ownerId, spoonId, 'base:head'),
+    );
+
+    expect(second).toBe(first);
+    expect(await countSpoonThreads(t, spoonId)).toBe(1);
+  });
+
+  test('creates a new thread for a different dedupKey', async () => {
+    const t = convexTest(schema, modules);
+    const { ownerId, spoonId } = await seedSpoon(t);
+
+    const first = await t.mutation(
+      internal.threads.createMaintenanceThread,
+      baseArgs(ownerId, spoonId, 'base:head1'),
+    );
+    const second = await t.mutation(
+      internal.threads.createMaintenanceThread,
+      baseArgs(ownerId, spoonId, 'base:head2'),
+    );
+
+    expect(second).not.toBe(first);
+    expect(await countSpoonThreads(t, spoonId)).toBe(2);
+  });
+
+  test('creates a new thread when the prior thread for the key is terminal', async () => {
+    const t = convexTest(schema, modules);
+    const { ownerId, spoonId } = await seedSpoon(t);
+
+    const first = await t.mutation(
+      internal.threads.createMaintenanceThread,
+      baseArgs(ownerId, spoonId, 'base:head'),
+    );
+
+    await t.run(async (ctx) => {
+      await ctx.db.patch(first, { status: 'resolved' });
+    });
+
+    const second = await t.mutation(
+      internal.threads.createMaintenanceThread,
+      baseArgs(ownerId, spoonId, 'base:head'),
+    );
+
+    expect(second).not.toBe(first);
+    expect(await countSpoonThreads(t, spoonId)).toBe(2);
+  });
+
+  test('findOpenMaintenanceThread returns the open thread by key and null when terminal', async () => {
+    const t = convexTest(schema, modules);
+    const { ownerId, spoonId } = await seedSpoon(t);
+
+    const threadId = await t.mutation(
+      internal.threads.createMaintenanceThread,
+      baseArgs(ownerId, spoonId, 'base:head'),
+    );
+
+    const found = await t.query(internal.threads.findOpenMaintenanceThread, {
+      spoonId,
+      ownerId,
+      dedupKey: 'base:head',
+    });
+    expect(found?._id).toBe(threadId);
+
+    await t.run(async (ctx) => {
+      await ctx.db.patch(threadId, { status: 'failed' });
+    });
+
+    const afterTerminal = await t.query(
+      internal.threads.findOpenMaintenanceThread,
+      { spoonId, ownerId, dedupKey: 'base:head' },
+    );
+    expect(afterTerminal).toBeNull();
+  });
+});