Notification service

NotificationService fans out across Telegram and Web Push. Routing, batches, preference filters, retries, and when a notice gets marked sent.

The notification system is organized around a service-oriented architecture :

• NotificationService acts as the orchestrator and router for multiple channels.
• Channel services implement a simple interface contract (channel_name property and broadcast/send methods).
• DatabaseService manages persistent state for notices and users.
• Clients encapsulate low-level interactions with external APIs (Telegram Bot API).
• Runners and servers coordinate execution and expose endpoints for triggering notifications.

graph TB
subgraph "CLI and Orchestration"
MAIN["main.py"]
RUNNER["notification_runner.py"]
end
subgraph "Notification Layer"
NS["NotificationService"]
TS["TelegramService"]
WPS["WebPushService"]
end
subgraph "Data Layer"
DBS["DatabaseService"]
DBC["DBClient"]
end
subgraph "External APIs"
TC["TelegramClient"]
end
MAIN --> RUNNER
RUNNER --> NS
NS --> TS
NS --> WPS
TS --> TC
NS --> DBS
DBS --> DBC

• NotificationService: Central coordinator that routes notifications to enabled channels and performs batch processing of pending notices.
• TelegramService: Implements channel-specific logic for Telegram, including message formatting, chunking, and broadcasting to users.
• WebPushService: Implements channel-specific logic for web push notifications, including VAPID authentication and subscription management.
• DatabaseService: Provides access to notices and users collections, including retrieval of unsent notices and marking them as sent.
• TelegramClient: Low-level client for Telegram Bot API with retry and rate-limit handling.
• NotificationRunner: CLI runner that wires dependencies and triggers sending of unsent notices.
• WebhookServer: Exposes REST endpoints for programmatic notification dispatch and subscription management.

The notification workflow begins with pending notices stored in MongoDB. The NotificationService retrieves unsent notices and broadcasts them to enabled channels. TelegramService formats and sends messages to users, while WebPushService delivers push notifications to subscribed browsers. Results are recorded back to the database to mark notices as sent.

sequenceDiagram
participant CLI as "CLI/main.py"
participant Runner as "NotificationRunner"
participant NS as "NotificationService"
participant DB as "DatabaseService"
participant TS as "TelegramService"
participant WPS as "WebPushService"
participant TC as "TelegramClient"
CLI->>Runner : send_updates(telegram, web)
Runner->>NS : send_unsent_notices(telegram, web)
NS->>DB : get_unsent_notices()
DB-->>NS : unsent_posts[]
loop For each post
NS->>TS : broadcast_to_all_users(message) [if telegram]
TS->>TC : send_message(text, parse_mode)
TC-->>TS : success/failure
NS->>WPS : broadcast_to_all_users(message) [if web]
WPS-->>NS : success/failure
NS->>DB : mark_as_sent(post_id)
end
NS-->>Runner : results
Runner-->>CLI : results

NotificationService#

• Responsibilities:
• Aggregates multiple channels and routes notifications accordingly.
• Performs batch processing of unsent notices.
• Broadcasts messages to specified channels and records delivery outcomes.
• Key methods:
• add_channel: Adds a channel implementation.
• send_to_channel: Sends a message to a specific channel by name.
• broadcast: Broadcasts to specified channels or all if none specified.
• send_unsent_notices: Retrieves unsent notices and sends them to target channels, marking as sent upon success.
• send_new_posts_to_all_users: Alternative entry point for scheduled jobs.

classDiagram
class NotificationService {
+channels : List
+db_service : DatabaseService
+add_channel(channel)
+send_to_channel(message, channel_name, **kwargs) bool
+broadcast(message, channels, **kwargs) Dict
+send_unsent_notices(telegram, web) Dict
+send_new_posts_to_all_users(telegram, web) Dict
}
class TelegramService {
+channel_name : str
+broadcast_to_all_users(message, **kwargs) Dict
+send_message(message, parse_mode, **kwargs) bool
+send_to_user(user_id, message, parse_mode, **kwargs) bool
}
class WebPushService {
+channel_name : str
+is_enabled : bool
+broadcast_to_all_users(message, **kwargs) Dict
+send_message(message, **kwargs) bool
+send_to_user(user_id, message, **kwargs) bool
}
NotificationService --> TelegramService : "routes to"
NotificationService --> WebPushService : "routes to"

TelegramService#

• Responsibilities:
• Implements channel_name property for routing.
• Formats messages for Telegram (MarkdownV2 and HTML).
• Splits long messages into chunks respecting Telegram limits.
• Broadcasts to all active users with rate limiting.
• Retries without formatting on failures.
• Key methods:
• send_message: Sends a message to default chat with optional parse_mode.
• send_to_user: Sends a message to a specific user.
• broadcast_to_all_users: Iterates active users and sends messages with throttling.
• Message formatting helpers: convert_markdown_to_telegram, convert_markdown_to_html, escape_markdown_v2, split_long_message.

flowchart TD
Start(["send_message(message, parse_mode)"]) --> CheckLong["Message length > 4000?"]
CheckLong --> |Yes| Split["split_long_message(max_length=4000)"]
Split --> LoopChunks["For each chunk"]
LoopChunks --> SendChunk["client.send_message(formatted, parse_mode)"]
SendChunk --> RetryPlain{"Sent?"}
RetryPlain --> |No| PlainText["send_message(original, parse_mode='')"]
RetryPlain --> |Yes| NextChunk["Next chunk"]
NextChunk --> DoneChunks["All chunks sent?"]
DoneChunks --> |Yes| ReturnTrue["Return True"]
DoneChunks --> |No| ReturnFalse["Return False"]
CheckLong --> |No| Format["Format based on parse_mode"]
Format --> SendSingle["client.send_message(formatted)"]
SendSingle --> Sent{"Sent?"}
Sent --> |No| PlainFallback["Retry without formatting"]
Sent --> |Yes| ReturnTrue
PlainFallback --> ReturnTrue

WebPushService#

• Responsibilities:
• Implements channel_name property for routing.
• Checks availability of pywebpush and VAPID keys.
• Broadcasts to all users with push subscriptions.
• Handles subscription removal for expired endpoints.
• Manages subscription persistence (save/remove/get_public_key).
• Key methods:
• send_message: Broadcasts to all subscriptions.
• send_to_user: Sends to a user’s subscriptions.
• broadcast_to_all_users: Iterates users and subscriptions.
• _send_push: Sends a single push with VAPID claims and handles WebPushException.

flowchart TD
Start(["broadcast_to_all_users(message)"]) --> Enabled{"is_enabled?"}
Enabled --> |No| ReturnDisabled["Return disabled result"]
Enabled --> |Yes| GetUsers["get_active_users()"]
GetUsers --> LoopUsers["For each user"]
LoopUsers --> GetSubs["user.push_subscriptions[]"]
GetSubs --> LoopSubs["For each subscription"]
LoopSubs --> SendOne["webpush(subscription, data, vapid)"]
SendOne --> HandleErr{"Exception?"}
HandleErr --> |WebPushException 404/410| RemoveSub["_remove_subscription()"]
HandleErr --> |Other| LogErr["Log error"]
RemoveSub --> NextSub["Next subscription"]
LogErr --> NextSub
NextSub --> NextUser["Next user"]
NextUser --> Done["Return totals"]

DatabaseService and user management#

• Responsibilities:
• Provides get_unsent_notices and mark_as_sent for queued notifications.
• Supplies get_active_users for broadcasting.
• Manages user registration and deactivation.
• Integration with NotificationService:
• NotificationService relies on DatabaseService to fetch unsent notices and mark them as sent after successful delivery.

classDiagram
class DatabaseService {
+get_unsent_notices() List
+mark_as_sent(post_id) bool
+get_active_users() List
+add_user(user_id, chat_id, username, first_name, last_name) Tuple
+deactivate_user(user_id) bool
+get_user_by_id(user_id) Dict
}
class NotificationService {
+send_unsent_notices(telegram, web) Dict
+send_new_posts_to_all_users(telegram, web) Dict
}
NotificationService --> DatabaseService : "uses"

NotificationRunner and CLI integration#

• Responsibilities:
• Creates and wires dependencies (TelegramService, WebPushService, DatabaseService).
• Initializes NotificationService with selected channels.
• Executes send_unsent_notices and returns results.
• CLI integration:
• main.py subcommand “send” invokes send_updates with telegram/web flags.

sequenceDiagram
participant CLI as "main.py"
participant Runner as "NotificationRunner"
participant DB as "DatabaseService"
participant TS as "TelegramService"
participant WPS as "WebPushService"
participant NS as "NotificationService"
CLI->>Runner : send_updates(telegram, web)
alt telegram enabled
Runner->>TS : create TelegramService(db_service)
Runner->>NS : create NotificationService([TS], db_service)
end
alt web enabled
Runner->>WPS : create WebPushService(db_service)
Runner->>NS : append WPS to channels
end
Runner->>NS : send_unsent_notices(telegram, web)
NS-->>Runner : results
Runner-->>CLI : results

WebhookServer integration#

• Responsibilities:
• Exposes endpoints to trigger notifications programmatically.
• Provides subscription management for web push.
• Integrates NotificationService and WebPushService via dependency injection.
• Endpoints:
• POST /api/notify: Broadcast to specified channels.
• POST /api/notify/telegram: Telegram-only.
• POST /api/notify/web-push: Web push-only.
• POST /api/push/subscribe and /api/push/unsubscribe: Manage subscriptions.
• GET /api/push/vapid-key: Retrieve VAPID public key.

sequenceDiagram
participant Client as "Client"
participant API as "WebhookServer"
participant NS as "NotificationService"
participant TS as "TelegramService"
participant WPS as "WebPushService"
Client->>API : POST /api/notify {message, channels, title}
API->>NS : broadcast(message, channels, title)
NS->>TS : broadcast_to_all_users(message)
NS->>WPS : broadcast_to_all_users(message)
NS-->>API : results
API-->>Client : NotifyResponse(success, results)

• Coupling:
• NotificationService depends on channel implementations via channel_name and broadcast/send contracts.
• Channel services depend on DatabaseService for user and subscription data.
• TelegramService depends on TelegramClient for API calls.
• Cohesion:
• Each service has a single responsibility: routing (NotificationService), channel-specific delivery (TelegramService/WebPushService), and persistence (DatabaseService).
• External dependencies:
• Telegram Bot API (requests).
• Optional web push library (pywebpush) with graceful degradation.
• MongoDB via PyMongo.

graph LR
NS["NotificationService"] --> TS["TelegramService"]
NS --> WPS["WebPushService"]
TS --> TC["TelegramClient"]
NS --> DBS["DatabaseService"]
DBS --> DBC["DBClient"]
WPS -.->|"optional"| PYWP["pywebpush"]

• Rate limiting:
• TelegramService applies throttling between user sends to avoid rate limits.
• TelegramClient handles 429 responses with Retry-After header.
• Message chunking:
• TelegramService splits long messages into chunks respecting Telegram’s character limits.
• Graceful degradation:
• WebPushService disables itself if pywebpush is unavailable or VAPID keys are missing.
• Batch processing:
• NotificationService processes unsent notices in batches and marks as sent upon success.
• Caching and reuse:
• Settings are cached to reduce repeated environment parsing.
• Indexing:
• MongoDB collections are indexed for frequent queries (e.g., notices by sent flags, users by user_id).

• Telegram configuration issues:
• Missing bot token or chat ID leads to early return in TelegramService and TelegramClient.
• Rate limiting: TelegramClient retries with exponential backoff and respects Retry-After.
• Web push configuration issues:
• Missing pywebpush or VAPID keys disables web push functionality.
• Expired subscriptions (404/410) are handled by removing them.
• Database connectivity:
• DBClient requires MONGO_CONNECTION_STR; connection failures are logged and raised.
• DatabaseService wraps operations with try/catch and logs errors.
• Notification delivery failures:
• NotificationService logs errors per channel and continues processing remaining posts.
• WebhookServer returns HTTP 500 on exceptions with error details.

NotificationService routes to channel services, degrades when one channel is down, and marks sent only after a real success. Keep channel logic inside TelegramService / WebPushService.