Server components

The three long-running processes: Telegram bot, APScheduler, FastAPI webhook. How runners sit under them and which CLI commands start each piece.

The application is organized into modular servers, runners, services, and clients, with centralized configuration and daemon utilities. The main CLI coordinates server startup and operational modes.

graph TB
subgraph "CLI and Orchestration"
MAIN["app/main.py"]
CFG["app/core/config.py"]
DAEMON["app/core/daemon.py"]
end
subgraph "Servers"
BOT["app/servers/bot_server.py"]
SCHED["app/servers/scheduler_server.py"]
WEB["app/servers/webhook_server.py"]
end
subgraph "Runners"
UPD_RUNNER["app/runners/update_runner.py"]
NOTIF_RUNNER["app/runners/notification_runner.py"]
end
subgraph "Services"
NOTIF_SVC["app/services/notification_service.py"]
DB_SVC["app/services/database_service.py"]
end
subgraph "Clients"
DB_CLIENT["app/clients/db_client.py"]
end
MAIN --> BOT
MAIN --> SCHED
MAIN --> WEB
SCHED --> UPD_RUNNER
SCHED --> NOTIF_RUNNER
BOT --> DB_SVC
WEB --> NOTIF_SVC
NOTIF_SVC --> DB_SVC
DB_SVC --> DB_CLIENT

• Telegram Bot Server: User commands (/start, /help, /stop, /status, /placement_year, /stats, /noticestats, /web) plus admin /userstats. Polling mode, graceful shutdown.
• Scheduler Server: Uses APScheduler to run periodic update jobs (SuperSet + Emails) and official placement scraping at fixed intervals.
• Webhook Server: FastAPI-based server exposing health, push subscription, notification dispatch, and statistics endpoints for external integrations.
• Runner Services: Encapsulate update fetching and notification sending logic for reuse across servers and CLI commands.
• Services and Clients: Notification routing, database operations, and MongoDB connectivity form the data and delivery backbone.

The system separates concerns across three servers and a runner layer:

• Bot Server: Interactive Telegram commands and user lifecycle
• Scheduler Server: Automated ingestion and notifications
• Webhook Server: External integrations and administrative APIs
• Runners: Shared orchestration for updates and notifications
• Services/Clients: Data persistence and channel delivery

graph TB
CLIENT["Telegram Users"]
ADMIN["Admin Users"]
BOT["Bot Server<br/>Command Handlers"]
SCHED["Scheduler Server<br/>APScheduler Jobs"]
WEB["Webhook Server<br/>FastAPI Endpoints"]
UPD_RUN["UpdateRunner<br/>Fetch + Process"]
NOTIF_RUN["NotificationRunner<br/>Send Unsents"]
NOTIF_SVC["NotificationService"]
DB_SVC["DatabaseService"]
DB_CLIENT["DBClient"]
CLIENT --> BOT
ADMIN --> BOT
SCHED --> UPD_RUN
SCHED --> NOTIF_RUN
WEB --> NOTIF_SVC
BOT --> DB_SVC
WEB --> DB_SVC
NOTIF_RUN --> NOTIF_SVC
NOTIF_SVC --> DB_SVC
DB_SVC --> DB_CLIENT

Telegram bot server#

• Responsibilities:
• Initialize Telegram Application with DI-enabled services
• Register command handlers for user/admin commands
• Manage lifecycle: initialize, start polling, keep alive, graceful shutdown
• Command Handlers:
• /start: user registration and activation
• /help: command reference
• /stop: deactivation
• /status: subscription status
• /placement_year: pick which placement year to follow
• /stats: placement statistics
• /noticestats: notice delivery stats
• /web: suite links
• /userstats: admin-only user stats
• Admin commands wired via injected AdminTelegramService (ADMIN_TELEGRAM_USER_IDS)
• Lifecycle:
• run_async builds Application, sets up handlers, starts polling
• run blocks until interrupted, then shutdown gracefully

sequenceDiagram
participant User as "Telegram User"
participant Bot as "BotServer"
participant App as "Telegram Application"
participant DB as "DatabaseService"
User->>Bot : "/start"
Bot->>App : "register handlers"
App-->>Bot : "ready"
Bot->>DB : "add_user(...)"
DB-->>Bot : "success/exists"
Bot-->>User : "welcome message"

Scheduler server (APScheduler)#

• Responsibilities:
• Schedule SuperSet + email updates at midnight and every hour from 8 AM to 11 PM IST (hour="0,8-23")
• Daily official placement scrape at noon IST
• Orchestrate runners for update and notification sending
• Job Execution:
• run_scheduled_update mirrors legacy behavior: fetch SuperSet updates, fetch email updates, send via Telegram
• run_official_placement_scrape scrapes official data and persists to DB
• Scheduler Setup:
• AsyncIOScheduler with Asia/Kolkata timezone
• Two cron jobs in setup_scheduler(), not a 24-hour hourly loop

flowchart TD
Start(["Scheduler Start"]) --> Init["Initialize AsyncIOScheduler"]
Init --> Jobs["Add Cron Jobs (0,8-23 + Noon)"]
Jobs --> Loop{"Every Tick"}
Loop --> Hourly["run_scheduled_update()"]
Hourly --> SS["fetch_and_process_updates()"]
Hourly --> EMAIL["run_email_updates()"]
Hourly --> SEND["send_updates(telegram=True, web=False)"]
Loop --> Noon["run_official_placement_scrape()"]
Loop --> Loop

Webhook server (FastAPI)#

• Responsibilities:
• Health endpoints
• Web push subscription management
• Notification dispatch to Telegram and/or Web Push
• Statistics endpoints for placements, notices, users
• External webhook to trigger unsent notice delivery
• Dependencies:
• DI via lifespan to construct DatabaseService, WebPushService, and NotificationService
• CORS enabled for external integrations
• Endpoints:
• GET /, GET /health
• POST /api/push/subscribe, POST /api/push/unsubscribe, GET /api/push/vapid-key
• POST /api/notify, POST /api/notify/telegram, POST /api/notify/web-push
• GET /api/stats, /api/stats/placements, /api/stats/notices, /api/stats/users
• POST /webhook/update

sequenceDiagram
participant Ext as "External Client"
participant API as "WebhookServer"
participant Notif as "NotificationService"
participant DB as "DatabaseService"
Ext->>API : "POST /api/notify"
API->>Notif : "broadcast(message, channels, title)"
Notif->>DB : "get_unsent_notices()"
DB-->>Notif : "unsent posts"
Notif-->>API : "results per channel"
API-->>Ext : "NotifyResponse"

Runner orchestration services#

• UpdateRunner:
• Logs in to SuperSet, pre-fetches existing IDs, filters new notices/jobs, enriches only new jobs, processes notices with job enrichment callback, saves to DB
• Optimizes API calls by enriching only new jobs and using existing job structures for linking
• NotificationRunner:
• Enables sending unsent notices via Telegram and/or Web Push
• Creates channels on demand and delegates to NotificationService
• Supports context manager for resource cleanup

flowchart TD
UStart(["UpdateRunner.fetch_and_process_updates"]) --> Login["Login to SuperSet"]
Login --> IDs["Load existing notice/job IDs from DB"]
IDs --> Notices["Fetch notices and filter new"]
IDs --> JobsBasic["Fetch basic jobs and filter new"]
JobsBasic --> Enrich["Enrich new jobs (expensive)"]
Notices --> Link["Link notices to jobs (basic/enriched)"]
Enrich --> SaveN["Save notices to DB"]
Link --> SaveJ["Upsert jobs to DB"]
SaveN --> UEnd(["Return counts"])
SaveJ --> UEnd

Notification service and database service#

• NotificationService:
• Aggregates channels (Telegram, Web Push), broadcasts to all users, and tracks sent status
• Sends unsent notices and marks them sent upon successful delivery
• DatabaseService:
• Manages notices, jobs, placement offers, users, policies, and official placement data
• Provides statistics and helpers for unsent notices and user management
• Implements deduplication and merging for placement offers

classDiagram
class NotificationService {
+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 DatabaseService {
+notice_exists(id) bool
+get_all_notice_ids() set
+save_notice(notice) (bool, str)
+get_unsent_notices() List
+mark_as_sent(post_id) bool
+upsert_structured_job(job) (bool, str)
+save_placement_offers(offers) Dict
+get_placement_stats() Dict
+add_user(user_id, chat_id, ...) (bool, str)
+get_users_stats() Dict
}
class DBClient {
+connect()
+close_connection()
+notices_collection
+jobs_collection
+placement_offers_collection
+users_collection
+policies_collection
+official_placement_data_collection
}
NotificationService --> DatabaseService : "uses"
DatabaseService --> DBClient : "wraps"

• Configuration:
• Centralized Settings with environment variable loading and logging setup
• Daemon mode toggles stdout vs file logging
• Daemon Utilities:
• Double-fork daemonization, PID file management, stop/status helpers
• Server-to-Service Coupling:
• Servers depend on DI-created services; runners encapsulate orchestration
• External Dependencies:
• Python packages pinned in requirements.txt and pyproject.toml

graph LR
CFG["Settings (config.py)"] --> BOT["BotServer"]
CFG --> SCHED["SchedulerServer"]
CFG --> WEB["WebhookServer"]
CFG --> DAEMON["daemon.py"]
BOT --> DB_SVC["DatabaseService"]
WEB --> NOTIF_SVC["NotificationService"]
SCHED --> UPD_RUN["UpdateRunner"]
SCHED --> NOTIF_RUN["NotificationRunner"]
NOTIF_RUN --> NOTIF_SVC
NOTIF_SVC --> DB_SVC
DB_SVC --> DB_CLIENT["DBClient"]

• UpdateRunner optimization:
• Pre-fetch existing IDs to avoid redundant API calls
• Enrich only new jobs with detailed info to minimize cost
• Use job lookups to avoid repeated enrich calls
• Notification batching:
• Broadcast to all users per channel and mark sent atomically
• Logging:
• Reduce noise from third-party libraries and route to file in daemon mode
• Scheduler cadence:
• Updates at midnight and 8 AM through 11 PM IST. Overnight 1 AM through 7 AM is idle. Change the cron in scheduler_server.py.

[No sources needed since this section provides general guidance]

• Environment variables:
• Store tokens and secrets in .env; Settings validates and loads them
• Webhook server:
• CORS enabled broadly for development; restrict origins in production
• No built-in authentication on endpoints; consider API keys or signed requests for external integrations
• Rate limiting:
• No explicit rate limiting in code; consider adding throttling at the FastAPI layer or upstream proxy
• Transport security:
• Use HTTPS in production; configure reverse proxies accordingly
• Daemon logging:
• Ensure logs do not expose sensitive data; sanitize outputs

• Local development:
• docker-compose.dev.yaml provisions MongoDB locally with exposed port and volume storage
• Containerization:
• Pin Python version and dependencies via pyproject.toml and requirements.txt
• Process management:
• Use daemon mode (-d) for long-running servers; manage via PID files and stop/status commands
• Scaling strategies:
• Horizontal scaling: run multiple instances behind a load balancer; ensure shared MongoDB
• Vertical scaling: increase CPU/RAM for CPU-bound tasks (enrichment, notifications)
• Queue-based offloading: introduce message queues for high-volume notifications
• Cloud platforms:
• Kubernetes: deploy stateless webhook and scheduler pods; persistent MongoDB via StatefulSet or managed service
• Platform-as-a-Service: run as separate processes (bot, scheduler, webhook) with environment variables

• Logging:
• Centralized setup via setup_logging; file handler plus optional stream handler
• Daemon mode suppresses stdout; logs written to files
• Metrics:
• No built-in metrics endpoints; consider adding Prometheus metrics for job durations and error rates
• Health checks:
• Webhook server exposes / and /health endpoints
• Alerting:
• Integrate with external monitoring systems to track server uptime and error rates

• Telegram bot not starting:
• Verify TELEGRAM_BOT_TOKEN is set; ensure network connectivity to Telegram
• Check logs for initialization errors
• Scheduler not running jobs:
• Confirm daemon mode and logging; verify cron schedule and timezone
• Inspect scheduler logs for exceptions during job execution
• Webhook server errors:
• Validate MONGO_CONNECTION_STR and database availability
• Check CORS configuration and endpoint permissions
• Database connectivity:
• Confirm MongoDB is reachable and credentials are correct
• Review DBClient connection logs
• Daemon control:
• Use stop/status commands to inspect PID and process state

Three servers, Telegram, scheduler, webhook, plus runners underneath. DI keeps them from growing into one blob. Harden auth, add rate limits, and watch the logs before you call it production.