Send websocket updates during rehash.

core/tasks.py

@@ -326,6 +326,20 @@ def rehash_streams(keys):
     total_records = queryset.count()
     logger.info(f"Starting rehash of {total_records} streams with keys: {keys}")
 
+    # Send initial WebSocket update
+    send_websocket_update(
+        'updates',
+        'update',
+        {
+            "success": True,
+            "type": "stream_rehash",
+            "action": "starting",
+            "progress": 0,
+            "total_records": total_records,
+            "message": f"Starting rehash of {total_records} streams"
+        }
+    )
+
     for start in range(0, total_records, batch_size):
         batch_processed = 0
         batch_duplicates = 0
@@ -398,12 +412,51 @@ def rehash_streams(keys):
         total_processed += batch_processed
         duplicates_merged += batch_duplicates
 
-        logger.info(f"Rehashed batch {start//batch_size + 1}/{(total_records//batch_size) + 1}: "
+        # Calculate progress percentage
+        progress_percent = int((total_processed / total_records) * 100)
+        current_batch = start // batch_size + 1
+        total_batches = (total_records // batch_size) + 1
+
+        # Send progress update via WebSocket
+        send_websocket_update(
+            'updates',
+            'update',
+            {
+                "success": True,
+                "type": "stream_rehash",
+                "action": "processing",
+                "progress": progress_percent,
+                "batch": current_batch,
+                "total_batches": total_batches,
+                "processed": total_processed,
+                "duplicates_merged": duplicates_merged,
+                "message": f"Processed batch {current_batch}/{total_batches}: {batch_processed} streams, {batch_duplicates} duplicates merged"
+            }
+        )
+
+        logger.info(f"Rehashed batch {current_batch}/{total_batches}: "
                    f"{batch_processed} processed, {batch_duplicates} duplicates merged")
 
     logger.info(f"Rehashing complete: {total_processed} streams processed, "
                f"{duplicates_merged} duplicates merged")
 
+    # Send completion update via WebSocket
+    send_websocket_update(
+        'updates',
+        'update',
+        {
+            "success": True,
+            "type": "stream_rehash",
+            "action": "completed",
+            "progress": 100,
+            "total_processed": total_processed,
+            "duplicates_merged": duplicates_merged,
+            "final_count": total_processed - duplicates_merged,
+            "message": f"Rehashing complete: {total_processed} streams processed, {duplicates_merged} duplicates merged"
+        },
+        collect_garbage=True  # Force garbage collection after completion
+    )
+
     return {
         'total_processed': total_processed,
         'duplicates_merged': duplicates_merged,

frontend/src/WebSocket.jsx

@@ -372,6 +372,43 @@ export const WebsocketProvider = ({ children }) => {
               }
               break;
 
+            case 'stream_rehash':
+              // Handle stream rehash progress updates
+              if (parsedEvent.data.action === 'starting') {
+                notifications.show({
+                  id: 'stream-rehash-progress', // Persistent ID
+                  title: 'Stream Rehash Started',
+                  message: parsedEvent.data.message,
+                  color: 'blue.5',
+                  autoClose: false, // Don't auto-close
+                  withCloseButton: false, // No close button during processing
+                  loading: true, // Show loading indicator
+                });
+              } else if (parsedEvent.data.action === 'processing') {
+                // Update the existing notification
+                notifications.update({
+                  id: 'stream-rehash-progress',
+                  title: 'Stream Rehash in Progress',
+                  message: `${parsedEvent.data.progress}% complete - ${parsedEvent.data.processed} streams processed, ${parsedEvent.data.duplicates_merged} duplicates merged`,
+                  color: 'blue.5',
+                  autoClose: false,
+                  withCloseButton: false,
+                  loading: true,
+                });
+              } else if (parsedEvent.data.action === 'completed') {
+                // Update to completion state
+                notifications.update({
+                  id: 'stream-rehash-progress',
+                  title: 'Stream Rehash Complete',
+                  message: `Processed ${parsedEvent.data.total_processed} streams, merged ${parsedEvent.data.duplicates_merged} duplicates. Final count: ${parsedEvent.data.final_count}`,
+                  color: 'green.5',
+                  autoClose: 8000, // Auto-close after completion
+                  withCloseButton: true, // Allow manual close
+                  loading: false, // Remove loading indicator
+                });
+              }
+              break;
+
             default:
               console.error(
                 `Unknown websocket event type: ${parsedEvent.data?.type}`