feat: add worker main poll loop

worker/__tests__/index.test.ts

@@ -0,0 +1,51 @@
+import { prisma } from '@/lib/prisma'
+import * as processor from '../processor'
+
+jest.mock('@/lib/prisma', () => ({
+  prisma: {
+    download: {
+      findMany: jest.fn(),
+      updateMany: jest.fn(),
+    },
+  },
+}))
+jest.mock('../processor')
+
+const mockFindMany = prisma.download.findMany as jest.Mock
+const mockUpdateMany = prisma.download.updateMany as jest.Mock
+const mockProcess = processor.processDownload as jest.Mock
+
+describe('pollOnce', () => {
+  beforeEach(() => {
+    jest.clearAllMocks()
+  })
+
+  it('processes all PENDING downloads', async () => {
+    mockFindMany.mockResolvedValue([{ id: 'dl-1' }, { id: 'dl-2' }])
+    mockUpdateMany.mockResolvedValue({ count: 2 })
+    mockProcess.mockResolvedValue(undefined)
+
+    const { pollOnce } = await import('../index')
+    await pollOnce()
+
+    expect(mockUpdateMany).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: expect.objectContaining({ status: 'PENDING' }),
+        data: expect.objectContaining({ status: 'PROCESSING' }),
+      })
+    )
+    expect(mockProcess).toHaveBeenCalledTimes(2)
+    expect(mockProcess).toHaveBeenCalledWith('dl-1')
+    expect(mockProcess).toHaveBeenCalledWith('dl-2')
+  })
+
+  it('does nothing when queue is empty', async () => {
+    mockFindMany.mockResolvedValue([])
+
+    const { pollOnce } = await import('../index')
+    await pollOnce()
+
+    expect(mockUpdateMany).not.toHaveBeenCalled()
+    expect(mockProcess).not.toHaveBeenCalled()
+  })
+})

worker/index.ts

@@ -0,0 +1,41 @@
+import { prisma } from '@/lib/prisma'
+import { config } from '../config/app.config'
+import { processDownload } from './processor'
+
+export async function pollOnce(): Promise<void> {
+  const pending = await prisma.download.findMany({
+    where: { status: 'PENDING' },
+    orderBy: { submittedAt: 'asc' },
+    take: config.WORKER_CONCURRENCY,
+    select: { id: true },
+  })
+
+  if (pending.length === 0) return
+
+  const ids = pending.map((d) => d.id)
+
+  await prisma.download.updateMany({
+    where: { id: { in: ids }, status: 'PENDING' },
+    data: { status: 'PROCESSING', startedAt: new Date() },
+  })
+
+  await Promise.all(ids.map((id) => processDownload(id)))
+}
+
+export async function runWorker(): Promise<never> {
+  console.log('[worker] started')
+  while (true) {
+    try {
+      await pollOnce()
+    } catch (err) {
+      console.error('[worker] poll error:', err)
+    }
+    await new Promise((resolve) =>
+      setTimeout(resolve, config.WORKER_POLL_INTERVAL_MS)
+    )
+  }
+}
+
+if (require.main === module) {
+  runWorker().catch(console.error)
+}