feat: redesign UPLOADS_UPDATE to send only changed items

backend/src/core/handler.ts

@@ -24,14 +24,6 @@ import type { Redis } from 'ioredis'
 
 const BATCH_UPLOADS_PAGE_SIZE = 100
 
-const UPLOAD_DONE_STATUSES = new Set([
-  'completed',
-  'failed',
-  'duplicate',
-  'duplicated_sdc_updated',
-  'duplicated_sdc_not_updated',
-])
-
 const BATCH_RETRIEVAL_CHUNK_SIZE = 100
 
 type SessionUserWithAuth = SessionUser & {
@@ -644,23 +636,40 @@ export class Handler {
   }
 
   private startUploadStream(batchid: number): ReturnType<typeof setTimeout> {
-    let lastSerialized: string | null = null
+    let cursorTime: Date = new Date(0)
+    // Track last-sent state per id: catches same-second re-updates that id-ordering misses
+    const sentState = new Map<number, string>()
     const poll = async () => {
       try {
-        const items = await this.getAllUploadsForBatch(batchid)
-        const updateItems = items.map(toUploadUpdateItem)
-        const serialized = JSON.stringify(updateItems)
-        if (serialized !== lastSerialized) {
+        // minId=0 makes this equivalent to updated_at >= cursorTime; sentState deduplicates
+        const candidates = await this.services.uploads.getUploadsByBatchChangedSinceWithIdCursor(
+          batchid,
+          cursorTime,
+          0,
+        )
+        const toSend = candidates.filter((r) => {
+          const key = `${r.status}|${JSON.stringify(r.error)}|${r.success ?? ''}`
+          if (sentState.get(r.id) === key) return false
+          sentState.set(r.id, key)
+          return true
+        })
+        if (toSend.length > 0) {
           this.sender.send({
             type: 'UPLOADS_UPDATE',
-            data: updateItems,
+            data: toSend.map(toUploadUpdateItem),
+            partial: true,
             nonce: nonce(),
           })
-          lastSerialized = serialized
         }
-        const total = await this.services.batches.countUploadsInBatch(batchid)
-        const completed = items.filter((i) => UPLOAD_DONE_STATUSES.has(i.status)).length
-        if (total > 0 && completed >= total) {
+        if (candidates.length > 0) {
+          const maxTime = candidates[candidates.length - 1]!.updated_at
+          if (maxTime > cursorTime) cursorTime = maxTime
+        }
+        const [total, active] = await Promise.all([
+          this.services.batches.countUploadsInBatch(batchid),
+          this.services.uploads.countActiveUploadsInBatch(batchid),
+        ])
+        if (total > 0 && active === 0) {
           this.sender.send({
             type: 'UPLOADS_COMPLETE',
             data: batchid,

backend/src/db/dal/uploads.ts

@@ -529,4 +529,44 @@ export class UploadService {
 
     return { items, total }
   }
+
+  async getUploadsByBatchChangedSinceWithIdCursor(
+    batchId: number,
+    since: Date,
+    minId: number,
+  ): Promise<SafeUploadRow[]> {
+    const {
+      access_token: _,
+      collection: __,
+      copyright_override: ___,
+      celery_task_id: ____,
+      ...cols
+    } = getTableColumns(uploadRequests)
+    return this.db
+      .select(cols)
+      .from(uploadRequests)
+      .where(
+        and(
+          eq(uploadRequests.batchid, batchId),
+          or(
+            gt(uploadRequests.updated_at, since),
+            and(eq(uploadRequests.updated_at, since), gt(uploadRequests.id, minId)),
+          ),
+        ),
+      )
+      .orderBy(asc(uploadRequests.updated_at), asc(uploadRequests.id))
+  }
+
+  async countActiveUploadsInBatch(batchId: number): Promise<number> {
+    const [row] = await this.db
+      .select({ n: count(uploadRequests.id) })
+      .from(uploadRequests)
+      .where(
+        and(
+          eq(uploadRequests.batchid, batchId),
+          inArray(uploadRequests.status, ['queued', 'in_progress']),
+        ),
+      )
+    return row?.n ?? 0
+  }
 }

backend/src/types/ws.ts

@@ -386,6 +386,7 @@ export const SubscribedSchema = t.Object({
 export const UploadsUpdateSchema = t.Object({
   type: t.Literal('UPLOADS_UPDATE'),
   data: t.Array(UploadUpdateItemSchema),
+  partial: t.Boolean(),
   nonce: t.String(),
 })
 

frontend/src/App.vue

@@ -19,7 +19,6 @@ initCollectionsListeners()
     <Header />
     <DevAuthBanner v-if="isDev" />
     <MaintenanceBanner />
-    <BetaBanner />
 
     <template v-if="auth.isAuthenticated">
       <div

frontend/src/composables/useCollections.ts

@@ -116,7 +116,7 @@ export const initCollectionsListeners = () => {
       }
 
       // Update batch uploads list if present
-      const index = newBatchUploads.findIndex((u) => u.key === update.key)
+      const index = newBatchUploads.findIndex((u) => u.id === update.id)
       if (index !== -1) {
         batchUploadsChanged = true
         const upload = { ...newBatchUploads[index] } as BatchUploadItem
@@ -141,11 +141,13 @@ export const initCollectionsListeners = () => {
           stats: {
             ...store.batch.stats,
             queued: newBatchUploads.filter((u) => u.status === UPLOAD_STATUS.Queued).length,
-            in_progress: newBatchUploads.filter((u) => u.status === UPLOAD_STATUS.InProgress).length,
+            in_progress: newBatchUploads.filter((u) => u.status === UPLOAD_STATUS.InProgress)
+              .length,
             completed: newBatchUploads.filter((u) => u.status === UPLOAD_STATUS.Completed).length,
             failed: newBatchUploads.filter((u) => u.status === UPLOAD_STATUS.Failed).length,
             cancelled: newBatchUploads.filter((u) => u.status === UPLOAD_STATUS.Cancelled).length,
-            duplicate: newBatchUploads.filter((u) => isDuplicateStatus(u.status as UploadStatus)).length,
+            duplicate: newBatchUploads.filter((u) => isDuplicateStatus(u.status as UploadStatus))
+              .length,
           },
         }
       }