refactor(backend): Clear out Notification Service code blockage

[majdyz]

Some of the code paths in the notification & scheduler service were synchronous HTTP calls that execute a long-running job that blocks. This makes the service threads busy waiting.

Changes 🏗️

• Remove queue_notification API
• Remove DTO
• Move heavy tasks intothe executor

Checklist 📋

For code changes:

• I have clearly listed my changes in the PR description
• I have made a test plan
• I have tested my changes according to the test plan:
  • Manually executing notification service jobs through the scheduler API

[netlify]

✅ Deploy Preview for auto-gpt-docs-dev canceled.

Name	Link
🔨 Latest commit	e4afd3a
🔍 Latest deploy log	https://app.netlify.com/sites/auto-gpt-docs-dev/deploys/681d9cc523282000081cf33a

[qodo-merge-pro]

PR Reviewer Guide 🔍

(Review updated until commit 4e6b66f)

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 3 🔵🔵🔵⚪⚪

🧪 No relevant tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Error Handling The new queue_notification and queue_notification_async functions have similar error handling but don't follow the same pattern as the service methods they replace. Consider standardizing error handling across these functions. def queue_notification(event: NotificationEventModel) -> NotificationResult: """Queue a notification - exposed method for other services to call""" try: logger.debug(f"Received Request to queue {event=}") exchange = "notifications" routing_key = get_routing_key(event.type) queue = get_notification_queue() queue.publish_message( routing_key=routing_key, message=event.model_dump_json(), exchange=next(ex for ex in EXCHANGES if ex.name == exchange), ) return NotificationResult( success=True, message=f"Notification queued with routing key: {routing_key}", ) except Exception as e: logger.exception(f"Error queueing notification: {e}") return NotificationResult(success=False, message=str(e)) async def queue_notification_async(event: NotificationEventModel) -> NotificationResult: """Queue a notification - exposed method for other services to call""" try: logger.debug(f"Received Request to queue {event=}") exchange = "notifications" routing_key = get_routing_key(event.type) queue = await get_async_notification_queue() await queue.publish_message( routing_key=routing_key, message=event.model_dump_json(), exchange=next(ex for ex in EXCHANGES if ex.name == exchange), ) return NotificationResult( success=True, message=f"Notification queued with routing key: {routing_key}", ) except Exception as e: logger.exception(f"Error queueing notification: {e}") return NotificationResult(success=False, message=str(e)) Thread Safety The background_executor is shared globally but there's no clear lifecycle management. Consider adding proper initialization and shutdown for the ThreadPoolExecutor to prevent resource leaks. background_executor = ThreadPoolExecutor(max_workers=2) Code Duplication The queue_notification and queue_notification_async functions share almost identical code. Consider refactoring to reduce duplication by extracting common logic into a shared helper function. def queue_notification(event: NotificationEventModel) -> NotificationResult: """Queue a notification - exposed method for other services to call""" try: logger.debug(f"Received Request to queue {event=}") exchange = "notifications" routing_key = get_routing_key(event.type) queue = get_notification_queue() queue.publish_message( routing_key=routing_key, message=event.model_dump_json(), exchange=next(ex for ex in EXCHANGES if ex.name == exchange), ) return NotificationResult( success=True, message=f"Notification queued with routing key: {routing_key}", ) except Exception as e: logger.exception(f"Error queueing notification: {e}") return NotificationResult(success=False, message=str(e)) async def queue_notification_async(event: NotificationEventModel) -> NotificationResult: """Queue a notification - exposed method for other services to call""" try: logger.debug(f"Received Request to queue {event=}") exchange = "notifications" routing_key = get_routing_key(event.type) queue = await get_async_notification_queue() await queue.publish_message( routing_key=routing_key, message=event.model_dump_json(), exchange=next(ex for ex in EXCHANGES if ex.name == exchange), ) return NotificationResult( success=True, message=f"Notification queued with routing key: {routing_key}", ) except Exception as e: logger.exception(f"Error queueing notification: {e}") return NotificationResult(success=False, message=str(e))

[deepsource-io]

Here's the code health analysis summary for commits 089e7aa..e4afd3a. View details on DeepSource ↗.

Analysis Summary

Analyzer	Status	Summary	Link
JavaScript	✅ Success		View Check ↗
Python	✅ Success	❗ 104 occurences introduced 🎯 106 occurences resolved	View Check ↗

---

💡 If you’re a repository administrator, you can configure the quality gates from the settings.

[netlify]

✅ Deploy Preview for auto-gpt-docs canceled.

Name	Link
🔨 Latest commit	e4afd3a
🔍 Latest deploy log	https://app.netlify.com/sites/auto-gpt-docs/deploys/681d9cc5b6471400084da577