From 0e894f8e6701bb5a5bac71a0813605dec2305791 Mon Sep 17 00:00:00 2001 From: Gabriel Brown Date: Sat, 11 Jul 2026 13:52:51 -0400 Subject: [PATCH] refactor(threads): dedup maintenance threads by stable key via index --- packages/backend/convex/threads.ts | 18 ++- .../tests/unit/maintenance-thread.test.ts | 146 ++++++++++++++++++ 2 files changed, 157 insertions(+), 7 deletions(-) create mode 100644 packages/backend/tests/unit/maintenance-thread.test.ts diff --git a/packages/backend/convex/threads.ts b/packages/backend/convex/threads.ts index a88a6c2..8c28720 100644 --- a/packages/backend/convex/threads.ts +++ b/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, diff --git a/packages/backend/tests/unit/maintenance-thread.test.ts b/packages/backend/tests/unit/maintenance-thread.test.ts new file mode 100644 index 0000000..8b6068c --- /dev/null +++ b/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) => + 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, + 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(); + }); +});