socket io setup

app/__init__.py

@@ -13,11 +13,11 @@ from flask import current_app, g, has_request_context, jsonify, make_response, r
 from flask.ctx import has_app_context
 from flask_marshmallow import Marshmallow
 from flask_migrate import Migrate
+from flask_socketio import SocketIO
 from flask_sqlalchemy import SQLAlchemy as _SQLAlchemy
 from sqlalchemy import event
 from werkzeug.exceptions import HTTPException as WerkzeugHTTPException
 from werkzeug.local import LocalProxy
-from flask_socketio import SocketIO
 
 from app import config
 from app.clients import NotificationProviderClients
@@ -32,6 +32,13 @@ from notifications_utils.clients.encryption.encryption_client import Encryption
 from notifications_utils.clients.redis.redis_client import RedisClient
 from notifications_utils.clients.zendesk.zendesk_client import ZendeskClient
 
+socketio = SocketIO(
+    cors_allowed_origins="*",
+    message_queue="redis://localhost:6379",
+    logger=True,
+    engineio_logger=True
+)
+
 
 class NotifyCelery(Celery):
     def init_app(self, app):
@@ -101,7 +108,7 @@ notification_provider_clients = NotificationProviderClients()
 api_user = LocalProxy(lambda: g.api_user)
 authenticated_service = LocalProxy(lambda: g.authenticated_service)
 
-socketio = SocketIO(cors_allowed_origins="*")  # ← Global instance
+
 def create_app(application):
     from app.config import configs
 
@@ -114,6 +121,8 @@ def create_app(application):
 
     socketio.init_app(application)
 
+    from app.socket_handlers import register_socket_handlers
+    register_socket_handlers(socketio)
     request_helper.init_app(application)
     db.init_app(application)
     migrate.init_app(application, db=db)
@@ -171,7 +180,6 @@ def register_blueprint(application):
     from app.inbound_number.rest import inbound_number_blueprint
     from app.inbound_sms.rest import inbound_sms as inbound_sms_blueprint
     from app.job.rest import job_blueprint
-    from app.socketio_server.test_socketio import test_bp
     from app.notifications.notifications_ses_callback import ses_callback_blueprint
     from app.notifications.receive_notifications import receive_notifications_blueprint
     from app.notifications.rest import notifications as notifications_blueprint
@@ -193,9 +201,6 @@ def register_blueprint(application):
     from app.user.rest import user_blueprint
     from app.webauthn.rest import webauthn_blueprint
 
-    # this will need to be requires_admin_auth
-    application.register_blueprint(test_bp)
-
     service_blueprint.before_request(requires_admin_auth)
     application.register_blueprint(service_blueprint, url_prefix="/service")
 

app/dao/notifications_dao.py

@@ -26,9 +26,11 @@ from werkzeug.datastructures import MultiDict
 from app import create_uuid, db
 from app.dao.dao_utils import autocommit
 from app.dao.inbound_sms_dao import Pagination
+from app.dao.jobs_dao import dao_get_job_by_id
 from app.enums import KeyType, NotificationStatus, NotificationType
 from app.models import FactNotificationStatus, Notification, NotificationHistory
 from app.utils import (
+    emit_job_update_summary,
     escape_special_characters,
     get_midnight_in_utc,
     midnight_n_days_ago,
@@ -895,6 +897,15 @@ def dao_update_delivery_receipts(receipts, delivered):
         f"#loadtestperformance batch update query time: \
         updated {len(receipts)} notification in {elapsed_time} ms"
     )
+    current_app.logger.info("✅✅✅✅ Reached delivery receipt processing")
+    job_ids = db.session.execute(
+        select(Notification.job_id)
+        .where(Notification.message_id.in_(id_to_carrier.keys()))
+    ).scalars().all()
+
+    for job_id in set(job_ids):
+        job = dao_get_job_by_id(job_id)
+        emit_job_update_summary(job)
 
 
 def dao_close_out_delivery_receipts():

app/socket_handlers.py

@@ -0,0 +1,16 @@
+from flask import current_app, request
+from flask_socketio import join_room, leave_room
+
+
+def register_socket_handlers(socketio):
+    @socketio.on("join")
+    def on_join(data):  # noqa: F401
+        room = data.get("room")
+        join_room(room)
+        current_app.logger.info(f"Socket {request.sid} joined room {room}")
+
+    @socketio.on("leave")
+    def on_leave(data):  # noqa: F401
+        room = data.get("room")
+        leave_room(room)
+        current_app.logger.info(f"Socket {request.sid} left room {room}")

app/socketio_server/test_socketio.py

@@ -1,12 +0,0 @@
-from flask import Blueprint
-from app import socketio
-
-test_bp = Blueprint('test', __name__)
-
-@test_bp.route('/test-emit', methods=["GET"])
-def test_emit():
-    socketio.emit('job_update', {
-        'job_id': 'abc123',
-        'status': 'Test message from API'
-    })
-    return "Event emitted!"

app/utils.py

@@ -131,3 +131,18 @@ def utc_now():
 def debug_not_production(msg):
     if os.getenv("NOTIFY_ENVIRONMENT") not in ["production"]:
         current_app.logger.info(msg)
+
+
+def emit_job_update_summary(job):
+    from app import socketio
+
+    current_app.logger.info(f"Emitting summary for job {job.id}")
+    socketio.emit(
+        "job_updated",
+        {
+            "job_id": str(job.id),
+            "job_status": job.job_status,
+            "notification_count": job.notification_count,
+        },
+        room=f"job-{job.id}"
+    )