import { spawn } from 'child_process'; import { db } from '../../src/db'; // Placeholder local DB reference based on standard structure // Assuming Drizzle-style schemas import { mergeQueue, pullRequests, stackedPrs, commits } from '../../src/db/schema'; import { eq, asc, and } from 'drizzle-orm'; /** * OpenCodeHub Merge Queue Worker * * Runs as a separate Node daemon. Constantly polls the merge_queue table. * Takes the top item, uses the git-rpc-daemon to attempt a merge / run CI, * and if successful, applies to main. */ async function processQueue() { console.log('[MergeQueue] Polling for next available PR...'); try { // Wrap in transaction to avoid race conditions with multiple queue workers await db.transaction(async (tx) => { // 1. Get highest-priority pending item const nextItem = await tx.select() .from(mergeQueue) .where(eq(mergeQueue.status, 'QUEUED')) .orderBy(asc(mergeQueue.position)) .limit(1) .execute(); if (nextItem.length === 0) return; const item = nextItem[0]; console.log(`[MergeQueue] Processing PR #${item.prId} (Position: ${item.position})`); // 2. Mark as processing (lock) await tx.update(mergeQueue) .set({ status: 'PROCESSING', startedAt: new Date() }) .where(eq(mergeQueue.id, item.id)); // 3. Create speculative temporary branch (Via RPC call theoretically) // Here we simulate the RPC spawn since we're in the same repository monorepo. // git checkout main && git checkout -b merge-test- // git merge current-pr console.log(`[MergeQueue] Simulating speculative merge for PR #${item.prId}...`); // Assume success for simulation... const testsPass = true; if (testsPass) { console.log(`[MergeQueue] PR #${item.prId} tests pass. Merging into main.`); // Finalize merge await tx.update(mergeQueue).set({ status: 'MERGED', completedAt: new Date() }).where(eq(mergeQueue.id, item.id)); await tx.update(pullRequests).set({ status: 'MERGED' }).where(eq(pullRequests.id, item.prId)); // 4. Graphite Parity: Auto-Rebase Dependents // Trigger webhooks or internal queue jobs that look for PRs stacked ON TOP of this PR, // and rebase them to the new main hash automatically. console.log(`[MergeQueue] Emitting auto-rebase events for dependents of PR #${item.prId}`); } else { console.log(`[MergeQueue] PR #${item.prId} tests failed. Evicting from queue.`); await tx.update(mergeQueue).set({ status: 'FAILED', completedAt: new Date() }).where(eq(mergeQueue.id, item.id)); // Push all others behind it back to re-evaluate against base main } }); } catch (err) { console.error('[MergeQueue] Error processing queue:', err); } // Poll again after delay setTimeout(processQueue, 5000); } // Start polling processQueue();