yaas

Author: Dylan-J

app/api/urls.py

@@ -34,6 +34,7 @@
     OpsQueuePeekView,
     OpsQueuePurgeView,
     OpsQueueEvictPurgeView,
+    OpsConnectionCloseView,
     OpsTaskForceCloseView,
 )
 from .views.analytics import (
@@ -176,6 +177,7 @@
     re_path(r'^ops/queues/(?P<name>[^/]+)/peek/?$', OpsQueuePeekView.as_view(), name='ops-queue-peek'),
     re_path(r'^ops/queues/(?P<name>[^/]+)/purge/?$', OpsQueuePurgeView.as_view(), name='ops-queue-purge'),
     re_path(r'^ops/queues/?$', OpsQueuesView.as_view(), name='ops-queues'),
+    re_path(r'^ops/connections/close/?$', OpsConnectionCloseView.as_view(), name='ops-connection-close'),
     re_path(r'^ops/tasks/(?P<task_id>[0-9]+)/force-close/?$', OpsTaskForceCloseView.as_view(), name='ops-task-force-close'),
 
     # OpenAPI schema (support both with and without trailing slash)

app/api/views/ops.py

@@ -34,6 +34,34 @@
 logger = logging.getLogger('trawlr.api.ops')
 
 
+def _slim_queue_depth(q):
+    """
+    Return just the fields the frontend renders for a queue depth row, or
+    None if the queue doesn't exist on the broker. Avoids leaking the full
+    RabbitMQ management payload over the API.
+    """
+    if not q:
+        return None
+    return {
+        'messages_ready': q.get('messages_ready', 0),
+        'messages_unacknowledged': q.get('messages_unacknowledged', 0),
+    }
+
+
+def _slim_queue_rows(rows):
+    """Reshape `get_queues_panel_context`'s queue rows into the API contract."""
+    out = []
+    for row in rows:
+        out.append({
+            'name': row['name'],
+            'main': _slim_queue_depth(row.get('main')),
+            'dq': _slim_queue_depth(row.get('dq')),
+            'xq': _slim_queue_depth(row.get('xq')),
+            'consumers': row.get('consumers', []),
+        })
+    return out
+
+
 class OpsQueuesView(APIView):
     """
     GET /api/v1/ops/queues
@@ -47,7 +75,7 @@ class OpsQueuesView(APIView):
     def get(self, request):
         ctx = get_queues_panel_context()
         return Response({
-            'queues': ctx['queue_rows'],
+            'queues': _slim_queue_rows(ctx['queue_rows']),
             'taskSummary': ctx['task_summary'],
             'orphans': ctx['orphans'],
             'error': ctx.get('queues_error'),
@@ -130,6 +158,35 @@ def post(self, request, name):
         })
 
 
+class OpsConnectionCloseView(APIView):
+    """
+    POST /api/v1/ops/connections/close
+    Body: {"name": "<connection name>"}
+
+    Force-close a single RabbitMQ consumer connection (worker will
+    auto-reconnect). Mirrors the v1 web `ops:connection_close` action.
+    """
+
+    permission_classes = [IsAdminUser]
+
+    def post(self, request):
+        conn_name = (request.data.get('name') or '').strip()
+        if not conn_name:
+            return Response(
+                {'error': 'name is required'},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        ok = rabbitmq.close_connection(conn_name, reason='closed via API')
+        return Response(
+            {'closed': bool(ok), 'name': conn_name},
+            status=(
+                status.HTTP_200_OK
+                if ok
+                else status.HTTP_502_BAD_GATEWAY
+            ),
+        )
+
+
 class OpsTaskForceCloseView(APIView):
     """
     POST /api/v1/ops/tasks/{id}/force-close