Make upload-slice sending resilient to WebSocket reconnects

backend/src/__tests__/rateLimiter.test.ts

@@ -1,6 +1,74 @@
 import { makeRedisMock } from '@backend/__tests__/helpers'
-import { getNextUploadDelay } from '@backend/core/rateLimiter'
-import { describe, expect, it } from 'bun:test'
+import {
+  getNextUploadDelay,
+  getRateLimitForBatch,
+  type RateLimitInfo,
+} from '@backend/core/rateLimiter'
+import type { MediaWikiClient } from '@backend/mediawiki/client'
+import { describe, expect, it, mock } from 'bun:test'
+
+const makeMwClient = (
+  overrides: Partial<Awaited<ReturnType<MediaWikiClient['getUserRateLimits']>>> = {},
+) =>
+  ({
+    getUserRateLimits: mock(async () => ({
+      ratelimits: {},
+      rights: [],
+      ...overrides,
+    })),
+  }) as unknown as MediaWikiClient
+
+describe('getRateLimitForBatch', () => {
+  it('returns cached value without calling MediaWiki on cache hit', async () => {
+    const cached: RateLimitInfo = { uploadsPerPeriod: 20, periodSeconds: 30 }
+    const { redis, getMock } = makeRedisMock(JSON.stringify(cached))
+    const client = makeMwClient()
+
+    const result = await getRateLimitForBatch('u1', client, redis)
+
+    expect(result).toEqual(cached)
+    expect(client.getUserRateLimits as ReturnType<typeof mock>).not.toHaveBeenCalled()
+    expect(getMock).toHaveBeenCalledWith(expect.stringContaining('u1'))
+  })
+
+  it('calls MediaWiki and caches result with 1-hour TTL on cache miss', async () => {
+    const { redis, setMock } = makeRedisMock(null)
+    const client = makeMwClient({
+      ratelimits: { upload: { user: { hits: 8, seconds: 60 } } },
+      rights: [],
+    })
+
+    const result = await getRateLimitForBatch('u1', client, redis)
+
+    expect(client.getUserRateLimits as ReturnType<typeof mock>).toHaveBeenCalledTimes(1)
+    expect(setMock).toHaveBeenCalledTimes(1)
+    const [key, value, ex, ttl] = (
+      setMock.mock.calls as unknown as [string, string, string, number][]
+    )[0]
+    expect(key).toContain('u1')
+    expect(JSON.parse(value)).toEqual(result)
+    expect(ex).toBe('EX')
+    expect(ttl).toBe(3600)
+  })
+
+  it('does not call MediaWiki again within the cache window', async () => {
+    let stored: string | null = null
+    const redis = {
+      get: mock(async () => stored),
+      set: mock(async (_k: string, v: string) => {
+        stored = v
+        return 'OK' as const
+      }),
+      del: mock(async () => 1),
+    } as unknown as import('ioredis').Redis
+    const client = makeMwClient()
+
+    await getRateLimitForBatch('u1', client, redis)
+    await getRateLimitForBatch('u1', client, redis)
+
+    expect(client.getUserRateLimits as ReturnType<typeof mock>).toHaveBeenCalledTimes(1)
+  })
+})
 
 describe('getNextUploadDelay', () => {
   it('returns the existing delay without updating Redis when uploadsPerPeriod is 0', async () => {

backend/src/__tests__/upload.worker.duplicate.test.ts

@@ -6,11 +6,13 @@
 //
 // mock.module() is intentionally avoided for @backend/core/crypto and
 // @backend/mediawiki/client (both imported directly by other test files).
-// Instead, WorkerDeps injection supplies mock implementations inline.
+// @backend/db/dal/uploads is a real value import in uploads.dal.test.ts, so it is
+// injected via WorkerDeps.uploads instead of mock.module() here.
 // @backend/handlers/mapillary is module-mocked here because no other test
 // file imports it directly.
 
 import { DuplicateUploadError } from '@backend/core/errors'
+import type { UploadService } from '@backend/db/dal/uploads'
 import type { MediaWikiClient } from '@backend/mediawiki/client'
 import { buildStatementsFromMapillaryImage } from '@backend/mediawiki/sdc'
 import type { MediaImage } from '@backend/types/ws'
@@ -34,15 +36,15 @@ mock.module('@backend/db/client', () => ({ lazyDb: { client: {} } }))
 
 const mockUpdateStatus = mock(async () => {})
 const mockClearToken = mock(async () => {})
-const mockGetById = mock(async (_id: number) => null as unknown)
-
-mock.module('@backend/db/dal/uploads', () => ({
-  UploadService: class {
-    updateUploadStatus = mockUpdateStatus
-    clearUploadAccessToken = mockClearToken
-    getUploadById = mockGetById
-  },
-}))
+const mockGetById = mock(
+  async (_id: number): Promise<Awaited<ReturnType<UploadService['getUploadById']>>> => null,
+)
+
+const mockUploads = {
+  updateUploadStatus: mockUpdateStatus,
+  clearUploadAccessToken: mockClearToken,
+  getUploadById: mockGetById,
+}
 
 // MapillaryHandler is NOT imported by any other test file — safe to mock.
 const FAKE_IMAGE: MediaImage = {
@@ -148,9 +150,13 @@ function setupWorker(clientStub: MediaWikiClient, uploadOverrides: Record<string
   mockGetById.mockClear()
   capturedProcessor = undefined
 
-  mockGetById.mockImplementation(async () => makeUpload(uploadOverrides))
+  mockGetById.mockImplementation(
+    async () =>
+      makeUpload(uploadOverrides) as unknown as Awaited<ReturnType<UploadService['getUploadById']>>,
+  )
 
   createUploadWorker({} as Redis, {
+    uploads: mockUploads,
     decryptToken: () => ['tok-key', 'tok-secret'],
     makeClient: () => clientStub,
   })

backend/src/__tests__/upload.worker.test.ts

@@ -1,4 +1,5 @@
 import { HashLockError, SourceCdnError, StorageError } from '@backend/core/errors'
+import type { UploadService } from '@backend/db/dal/uploads'
 import { beforeEach, describe, expect, it, mock } from 'bun:test'
 import type { Redis } from 'ioredis'
 
@@ -7,7 +8,8 @@ import type { Redis } from 'ioredis'
 // Only modules that no other test file imports directly are safe to mock here.
 // - bullmq: not imported by any other test ✓
 // - @backend/db/client: not imported by any other test ✓
-// - @backend/db/dal/uploads: only type-imported (erased at runtime) by ws.handler.test.ts ✓
+// @backend/db/dal/uploads is a real value import in uploads.dal.test.ts, so it is
+// injected via WorkerDeps.uploads instead of mock.module() here.
 
 // Capture the processor and event handlers registered by createUploadWorker.
 let capturedProcessor: ((job: unknown) => Promise<void>) | undefined
@@ -28,15 +30,15 @@ mock.module('@backend/db/client', () => ({ lazyDb: { client: {} } }))
 
 const mockUpdateStatus = mock(async () => {})
 const mockClearToken = mock(async () => {})
-const mockGetById = mock(async (_id: number) => null as unknown)
-
-mock.module('@backend/db/dal/uploads', () => ({
-  UploadService: class {
-    updateUploadStatus = mockUpdateStatus
-    clearUploadAccessToken = mockClearToken
-    getUploadById = mockGetById
-  },
-}))
+const mockGetById = mock(
+  async (_id: number): Promise<Awaited<ReturnType<UploadService['getUploadById']>>> => null,
+)
+
+const mockUploads = {
+  updateUploadStatus: mockUpdateStatus,
+  clearUploadAccessToken: mockClearToken,
+  getUploadById: mockGetById,
+}
 
 import { createUploadWorker } from '@backend/workers/upload.worker'
 
@@ -66,7 +68,7 @@ beforeEach(() => {
   mockUpdateStatus.mockClear()
   mockClearToken.mockClear()
   mockGetById.mockClear()
-  createUploadWorker(mockRedis)
+  createUploadWorker(mockRedis, { uploads: mockUploads })
 })
 
 // === Processor rethrows retryable errors without touching the DB ===
@@ -157,6 +159,7 @@ describe('upload worker — permanent BullMQ failure marks upload as failed in D
     const mockGetDelay = mock(async () => 1500)
     const mockRemoveJob = mock(async () => {})
     createUploadWorker(mockRedis, {
+      uploads: mockUploads,
       enqueueUpload: mockEnqueue,
       getNextUploadDelay: mockGetDelay,
       removeUploadJob: mockRemoveJob,
@@ -178,7 +181,11 @@ describe('upload worker — permanent BullMQ failure marks upload as failed in D
   it('StorageError: permanently fails when requeueCount reaches the limit', async () => {
     const mockEnqueue = mock(async () => 'new-job-id')
     const mockGetDelay = mock(async () => 1500)
-    createUploadWorker(mockRedis, { enqueueUpload: mockEnqueue, getNextUploadDelay: mockGetDelay })
+    createUploadWorker(mockRedis, {
+      uploads: mockUploads,
+      enqueueUpload: mockEnqueue,
+      getNextUploadDelay: mockGetDelay,
+    })
 
     const failedHandler = capturedHandlers.get('failed')
     expect(failedHandler).toBeDefined()

backend/src/__tests__/uploads.dal.test.ts

@@ -0,0 +1,129 @@
+import type { DB } from '@backend/db/client'
+import { UploadService } from '@backend/db/dal/uploads'
+import type { Handler, UploadItem } from '@backend/types/ws'
+import { describe, expect, it, mock } from 'bun:test'
+
+const makeItem = (id: string): UploadItem => ({
+  id,
+  input: 'col-1',
+  title: `${id}.jpg`,
+  wikitext: '== wikitext ==',
+  labels: null,
+  copyright_override: false,
+})
+
+const BASE_PARAMS = {
+  userid: 'u1',
+  username: 'alice',
+  batchid: 42,
+  handler: 'mapillary' as Handler,
+  encryptedAccessToken: 'tok',
+}
+
+const makeDb = ({
+  preExisting,
+  existing,
+}: {
+  preExisting: { key: string }[]
+  existing: { id: number; key: string; status: string }[]
+}) => {
+  const insertChain = {
+    values: mock(() => insertChain),
+    onDuplicateKeyUpdate: mock(() => Promise.resolve()),
+  }
+  const preExistingChain = {
+    from: mock(() => preExistingChain),
+    where: mock(() => Promise.resolve(preExisting)),
+  }
+  const existingChain = {
+    from: mock(() => existingChain),
+    where: mock(() => existingChain),
+    orderBy: mock(() => Promise.resolve(existing)),
+  }
+  let selectCallCount = 0
+  const db = {
+    insert: mock(() => insertChain),
+    select: mock(() => {
+      selectCallCount += 1
+      return selectCallCount % 2 === 1 ? preExistingChain : existingChain
+    }),
+  } as unknown as DB
+  return { db, insertChain, preExistingChain, existingChain }
+}
+
+describe('createUploadRequestsForBatch', () => {
+  it('returns empty array and skips DB when items list is empty', async () => {
+    const { db } = makeDb({ preExisting: [], existing: [] })
+    const service = new UploadService(db)
+    const result = await service.createUploadRequestsForBatch({ ...BASE_PARAMS, items: [] })
+    expect(result).toEqual([])
+    expect(db.insert).not.toHaveBeenCalled()
+    expect(db.select).not.toHaveBeenCalled()
+  })
+
+  it('marks freshly inserted rows as new with queued status', async () => {
+    const { db } = makeDb({
+      preExisting: [],
+      existing: [
+        { id: 1, key: 'img-1', status: 'queued' },
+        { id: 2, key: 'img-2', status: 'queued' },
+      ],
+    })
+    const service = new UploadService(db)
+    const result = await service.createUploadRequestsForBatch({
+      ...BASE_PARAMS,
+      items: [makeItem('img-1'), makeItem('img-2')],
+    })
+    expect(result).toEqual([
+      { id: 1, key: 'img-1', status: 'queued', isNew: true },
+      { id: 2, key: 'img-2', status: 'queued', isNew: true },
+    ])
+  })
+
+  it('uses onDuplicateKeyUpdate so the same slice can be resent on reconnect', async () => {
+    const { db, insertChain } = makeDb({
+      preExisting: [],
+      existing: [{ id: 1, key: 'img-1', status: 'queued' }],
+    })
+    const service = new UploadService(db)
+    const params = { ...BASE_PARAMS, items: [makeItem('img-1')] }
+
+    await service.createUploadRequestsForBatch(params)
+    await service.createUploadRequestsForBatch(params)
+
+    expect(db.insert).toHaveBeenCalledTimes(2)
+    expect(insertChain.onDuplicateKeyUpdate).toHaveBeenCalledTimes(2)
+  })
+
+  it('marks a resent row as not new and reports its real current status', async () => {
+    const { db } = makeDb({
+      preExisting: [{ key: 'img-1' }],
+      existing: [{ id: 1, key: 'img-1', status: 'in_progress' }],
+    })
+    const service = new UploadService(db)
+    const params = { ...BASE_PARAMS, items: [makeItem('img-1')] }
+
+    const result = await service.createUploadRequestsForBatch(params)
+
+    expect(result).toEqual([{ id: 1, key: 'img-1', status: 'in_progress', isNew: false }])
+  })
+
+  it('reports a mix of new and resent rows independently within the same slice', async () => {
+    const { db } = makeDb({
+      preExisting: [{ key: 'img-1' }],
+      existing: [
+        { id: 1, key: 'img-1', status: 'completed' },
+        { id: 2, key: 'img-2', status: 'queued' },
+      ],
+    })
+    const service = new UploadService(db)
+    const params = { ...BASE_PARAMS, items: [makeItem('img-1'), makeItem('img-2')] }
+
+    const result = await service.createUploadRequestsForBatch(params)
+
+    expect(result).toEqual([
+      { id: 1, key: 'img-1', status: 'completed', isNew: false },
+      { id: 2, key: 'img-2', status: 'queued', isNew: true },
+    ])
+  })
+})