"""REST API for the ServerKit Queue Bus.""" from flask import Blueprint, request, jsonify from flask_jwt_extended import jwt_required from app.queue_bus.service import QueueBusService, QueueBusError from app.middleware.rbac import get_current_user from app.api._request import json_object queue_bus_bp = Blueprint('queue_bus', __name__) def _is_admin(user): return user is not None and getattr(user, 'role', None) == 'admin' def _ensure_group_mutable(group_slug): """Reject HTTP mutations against system-owned queue groups. System groups (jobs, notifications, webhook deliveries, ...) are managed by the panel itself and are read-only over the REST API. Internal producers and consumers call ``QueueBusService`` directly and are unaffected by this guard. """ group = QueueBusService.get_group(group_slug) if group and group.get('owner_type') == 'system': raise QueueBusError('System queue groups are read-only', 403) def _ensure_group_accessible(group_slug): """Reject message reads/acks against queue groups the caller doesn't own. System-owned groups (jobs, notifications, webhook deliveries, ...) carry internal payloads and are admin-only over the REST API — otherwise any user could pop messages (visibility timeout) and stall internal processing, or read/ack/kill them. User-owned groups are limited to the owning user (and admins). Internal producers and consumers call ``QueueBusService`` directly and are unaffected by this guard. """ group = QueueBusService.get_group(group_slug) if not group: return # missing group — the service call below raises the 404 user = get_current_user() if _is_admin(user): return if (group.get('owner_type') == 'user' and user is not None and group.get('owner_id') == str(user.id)): return raise QueueBusError('Access denied', 403) # ---------------------------------------------------------------------- # Groups # ---------------------------------------------------------------------- @queue_bus_bp.route('/groups', methods=['GET']) @jwt_required() def list_groups(): owner_type = request.args.get('owner_type') owner_id = request.args.get('owner_id') limit = request.args.get('limit', 100, type=int) offset = request.args.get('offset', 0, type=int) groups = QueueBusService.list_groups( owner_type=owner_type, owner_id=owner_id, limit=limit, offset=offset, ) return jsonify({'groups': groups}), 200 @queue_bus_bp.route('/groups', methods=['POST']) @jwt_required() def create_group(): data = json_object() slug = data.get('slug') name = data.get('name') description = data.get('description') owner_type = data.get('owner_type', 'user') owner_id = data.get('owner_id') config = data.get('config') if not slug and not name: return jsonify({'error': 'name or slug is required'}), 400 user = get_current_user() if not _is_admin(user): # Non-admins create under their own user ownership. owner_type = 'user' owner_id = str(user.id) if user else None group = QueueBusService.create_group( slug=slug, name=name, description=description, owner_type=owner_type, owner_id=owner_id, config=config, ) return jsonify({'group': group}), 201 @queue_bus_bp.route('/groups/', methods=['GET']) @jwt_required() def get_group(group_slug): group = QueueBusService.get_group(group_slug) if not group: return jsonify({'error': 'Group not found'}), 404 return jsonify({'group': group}), 200 @queue_bus_bp.route('/groups/', methods=['PATCH']) @jwt_required() def update_group(group_slug): _ensure_group_mutable(group_slug) _ensure_group_accessible(group_slug) data = json_object() group = QueueBusService.update_group( slug=group_slug, name=data.get('name'), description=data.get('description'), config=data.get('config'), ) return jsonify({'group': group}), 200 @queue_bus_bp.route('/groups/', methods=['DELETE']) @jwt_required() def delete_group(group_slug): _ensure_group_mutable(group_slug) _ensure_group_accessible(group_slug) result = QueueBusService.delete_group(group_slug) return jsonify(result), 200 # ---------------------------------------------------------------------- # Queues # ---------------------------------------------------------------------- @queue_bus_bp.route('/groups//queues', methods=['GET']) @jwt_required() def list_queues(group_slug): limit = request.args.get('limit', 100, type=int) offset = request.args.get('offset', 0, type=int) queues = QueueBusService.list_queues(group_slug, limit=limit, offset=offset) return jsonify({'queues': queues}), 200 @queue_bus_bp.route('/groups//queues', methods=['POST']) @jwt_required() def create_queue(group_slug): _ensure_group_mutable(group_slug) _ensure_group_accessible(group_slug) data = json_object() slug = data.get('slug') name = data.get('name') description = data.get('description') config = data.get('config') if not slug and not name: return jsonify({'error': 'name or slug is required'}), 400 queue = QueueBusService.create_queue( group_slug=group_slug, slug=slug, name=name, description=description, config=config, ) return jsonify({'queue': queue}), 201 @queue_bus_bp.route('/groups//queues/', methods=['GET']) @jwt_required() def get_queue(group_slug, queue_slug): queue = QueueBusService.get_queue(group_slug, queue_slug) if not queue: return jsonify({'error': 'Queue not found'}), 404 return jsonify({'queue': queue}), 200 @queue_bus_bp.route('/groups//queues/', methods=['PATCH']) @jwt_required() def update_queue(group_slug, queue_slug): _ensure_group_mutable(group_slug) _ensure_group_accessible(group_slug) data = json_object() queue = QueueBusService.update_queue( group_slug=group_slug, slug=queue_slug, name=data.get('name'), description=data.get('description'), config=data.get('config'), ) return jsonify({'queue': queue}), 200 @queue_bus_bp.route('/groups//queues/', methods=['DELETE']) @jwt_required() def delete_queue(group_slug, queue_slug): _ensure_group_mutable(group_slug) _ensure_group_accessible(group_slug) result = QueueBusService.delete_queue(group_slug, queue_slug) return jsonify(result), 200 # ---------------------------------------------------------------------- # Messages # ---------------------------------------------------------------------- @queue_bus_bp.route('/groups//queues//messages', methods=['GET']) @jwt_required() def list_messages(group_slug, queue_slug): _ensure_group_accessible(group_slug) status = request.args.get('status') limit = request.args.get('limit', 100, type=int) offset = request.args.get('offset', 0, type=int) messages = QueueBusService.list_messages( group_slug=group_slug, queue_slug=queue_slug, status=status, limit=limit, offset=offset, ) return jsonify({'messages': messages}), 200 @queue_bus_bp.route('/groups//queues//messages', methods=['POST']) @jwt_required() def send_message(group_slug, queue_slug): _ensure_group_mutable(group_slug) _ensure_group_accessible(group_slug) data = json_object() payload = data.get('payload') priority = data.get('priority', 0) delay_ms = data.get('delay_ms', 0) max_attempts = data.get('max_attempts') if payload is None: return jsonify({'error': 'payload is required'}), 400 message = QueueBusService.send( group_slug=group_slug, queue_slug=queue_slug, payload=payload, priority=priority, delay_ms=delay_ms, max_attempts=max_attempts, ) return jsonify({'message': message}), 201 @queue_bus_bp.route('/groups//queues//messages/receive', methods=['POST']) @jwt_required() def receive_messages(group_slug, queue_slug): _ensure_group_accessible(group_slug) data = json_object() visibility_timeout_ms = data.get('visibility_timeout_ms', 30000) max_messages = data.get('max_messages', 1) messages = QueueBusService.receive( group_slug=group_slug, queue_slug=queue_slug, visibility_timeout_ms=visibility_timeout_ms, max_messages=max_messages, ) return jsonify({'messages': messages}), 200 @queue_bus_bp.route('/groups//queues//messages/', methods=['GET']) @jwt_required() def get_message(group_slug, queue_slug, message_id): _ensure_group_accessible(group_slug) message = QueueBusService.get_message(group_slug, queue_slug, message_id) if not message: return jsonify({'error': 'Message not found'}), 404 return jsonify({'message': message}), 200 @queue_bus_bp.route('/groups//queues//messages//complete', methods=['POST']) @jwt_required() def complete_message(group_slug, queue_slug, message_id): _ensure_group_accessible(group_slug) message = QueueBusService.complete(group_slug, queue_slug, message_id) return jsonify({'message': message}), 200 @queue_bus_bp.route('/groups//queues//messages//fail', methods=['POST']) @jwt_required() def fail_message(group_slug, queue_slug, message_id): _ensure_group_accessible(group_slug) data = json_object() message = QueueBusService.fail( group_slug=group_slug, queue_slug=queue_slug, message_id=message_id, error_message=data.get('error_message'), requeue=data.get('requeue', False), ) return jsonify({'message': message}), 200 @queue_bus_bp.route('/groups//queues//messages//requeue', methods=['POST']) @jwt_required() def requeue_message(group_slug, queue_slug, message_id): _ensure_group_mutable(group_slug) _ensure_group_accessible(group_slug) message = QueueBusService.requeue(group_slug, queue_slug, message_id) return jsonify({'message': message}), 200 @queue_bus_bp.route('/groups//queues//messages/', methods=['DELETE']) @jwt_required() def delete_message(group_slug, queue_slug, message_id): _ensure_group_mutable(group_slug) _ensure_group_accessible(group_slug) result = QueueBusService.delete_message(group_slug, queue_slug, message_id) return jsonify(result), 200 # ---------------------------------------------------------------------- # Shorthand routes # ---------------------------------------------------------------------- @queue_bus_bp.route('///messages', methods=['POST']) @jwt_required() def send_message_shorthand(group_slug, queue_slug): return send_message(group_slug, queue_slug) @queue_bus_bp.route('///messages/receive', methods=['POST']) @jwt_required() def receive_messages_shorthand(group_slug, queue_slug): return receive_messages(group_slug, queue_slug) # ---------------------------------------------------------------------- # Stats # ---------------------------------------------------------------------- @queue_bus_bp.route('/stats', methods=['GET']) @jwt_required() def global_stats(): stats = QueueBusService.get_stats() return jsonify(stats), 200 @queue_bus_bp.route('/groups//stats', methods=['GET']) @jwt_required() def group_stats(group_slug): stats = QueueBusService.get_stats(group_slug=group_slug) return jsonify(stats), 200 @queue_bus_bp.route('/groups//queues//stats', methods=['GET']) @jwt_required() def queue_stats(group_slug, queue_slug): stats = QueueBusService.get_stats(group_slug=group_slug, queue_slug=queue_slug) return jsonify(stats), 200