Webhook server

FastAPI webhook server for REST and external triggers. Routes, auth/CORS, validation, errors, and how it shares NotificationService and DatabaseService with the bot process.

The webhook server is implemented as a FastAPI application with dependency injection and integrates with database and notification services. The CLI entry point supports running the webhook server alongside other servers.

graph TB
subgraph "CLI and Entrypoints"
MAIN["app/main.py<br/>CLI entrypoint"]
end
subgraph "Webhook Server"
WS["app/servers/webhook_server.py<br/>FastAPI app factory"]
CFG["app/core/config.py<br/>Settings and logging"]
end
subgraph "Services"
DB["app/services/database_service.py<br/>MongoDB wrapper"]
NOTIF["app/services/notification_service.py<br/>Channel router"]
WP["app/services/web_push_service.py<br/>Web Push channel"]
end
subgraph "External Integrations"
TG["Telegram Bot Server<br/>(app/servers/bot_server.py)"]
EXT["External Systems"]
end
MAIN --> WS
WS --> CFG
WS --> DB
WS --> NOTIF
NOTIF --> WP
WS --> TG
WS --> EXT

• FastAPI application factory with dependency injection
• Health and statistics endpoints
• Web push subscription endpoints
• Notification dispatch endpoints
• Webhook trigger endpoint for external integrations

It owns:

• Expose REST endpoints for health, stats, web push, and notifications
• Provide a webhook endpoint to trigger internal update and notification workflows
• Manage CORS and logging
• Validate request schemas using Pydantic models

The webhook server orchestrates data retrieval and notification delivery through injected services. It exposes:

• Health endpoints for readiness/liveness checks
• Statistics endpoints backed by the database service
• Web push subscription management
• Notification dispatch to multiple channels
• A webhook endpoint to trigger unsent notice broadcasts

graph TB
Client["External Clients<br/>and Integrations"] --> API["FastAPI Webhook Server"]
API --> Health["Health Endpoints"]
API --> Stats["Stats Endpoints"]
API --> Push["Web Push Endpoints"]
API --> Notify["Notification Endpoints"]
API --> Trigger["Webhook Trigger"]
Notify --> Router["NotificationService"]
Router --> DB["DatabaseService"]
Router --> WP["WebPushService"]
Router --> TG["TelegramService"]
Push --> DB
Stats --> DB

REST API endpoints#

Health endpoints#

• GET /
• Returns a basic health status
• Response model: HealthResponse
• GET /health
• Returns detailed health status including version

Validation and responses:

• Response models define status and version fields
• No authentication required

Web push endpoints#

• POST /api/push/subscribe
• Request body: PushSubscription
• Response: success boolean
• Requires web push service enabled
• POST /api/push/unsubscribe
• Request body: PushSubscription
• Response: success boolean
• GET /api/push/vapid-key
• Response: public key for VAPID configuration

Validation and responses:

• Request bodies validated by Pydantic models
• HTTP 501 returned when web push is not configured
• HTTP 500 returned for unexpected errors

Notification endpoints#

• POST /api/notify
• Request body: NotifyRequest
• Response model: NotifyResponse
• Broadcasts to configured channels
• POST /api/notify/telegram
• Sends to Telegram only
• POST /api/notify/web-push
• Sends to Web Push only

Validation and responses:

• Request bodies validated by Pydantic models
• HTTP 501 returned when notification service is not configured
• HTTP 500 returned for unexpected errors

Statistics endpoints#

• GET /api/stats
• Response model: StatsResponse
• GET /api/stats/placements
• GET /api/stats/notices
• GET /api/stats/users

Validation and responses:

• HTTP 501 returned when database is not configured
• HTTP 500 returned for unexpected errors

Webhook trigger endpoint#

• POST /webhook/update
• Triggers unsent notice broadcast to Telegram and Web Push
• Returns success boolean and results

Validation and responses:

• HTTP 501 returned when services are not configured
• HTTP 500 returned for unexpected errors

Request/Response schemas#

classDiagram
class HealthResponse {
+string status
+string version
}
class PushSubscription {
+string endpoint
+dict keys
+int user_id
}
class NotifyRequest {
+string message
+string title
+list channels
}
class NotifyResponse {
+bool success
+dict results
}
class StatsResponse {
+dict placement_stats
+dict notice_stats
+dict user_stats
}

Dependency injection and service wiring#

The application uses a factory pattern to construct the FastAPI app with injected services:

• DatabaseService
• NotificationService (comprising Telegram and Web Push channels)
• WebPushService

sequenceDiagram
participant Client as "Client"
participant App as "FastAPI App"
participant DI as "DI Container"
participant DB as "DatabaseService"
participant Notif as "NotificationService"
participant WP as "WebPushService"
Client->>App : HTTP Request
App->>DI : Resolve dependencies
DI-->>App : Services (DB, Notif, WP)
App->>DB : Query/Update
App->>Notif : Broadcast/Dispatch
Notif->>WP : Send to Web Push
App-->>Client : Response

Webhook endpoint implementation#

The webhook endpoint triggers the internal workflow to send unsent notices to all enabled channels.

sequenceDiagram
participant Ext as "External System"
participant API as "Webhook Server"
participant Notif as "NotificationService"
participant DB as "DatabaseService"
Ext->>API : POST /webhook/update
API->>Notif : send_unsent_notices(telegram=True, web=True)
Notif->>DB : get_unsent_notices()
DB-->>Notif : List of notices
Notif->>Notif : Broadcast to channels
Notif->>DB : mark_as_sent(post_id)
API-->>Ext : {success, result}

Data validation and sanitization#

• Pydantic models define strict request schemas for all endpoints
• Validation occurs automatically by FastAPI
• Error responses use HTTP status codes and standardized exceptions

Common validations:

• Required fields enforced by model definitions
• Optional fields defaulted where applicable
• Exceptions raised on invalid payloads

Error handling strategies#

• HTTP 501: Service not configured (e.g., web push, notification, database)
• HTTP 500: Unexpected runtime errors
• Exceptions are caught and mapped to appropriate HTTP responses
• Logging captures errors for diagnostics

External dependencies relevant to the webhook server:

• FastAPI and Uvicorn for the ASGI server
• Pydantic and Pydantic Settings for configuration and validation
• PyWebPush for Web Push notifications
• python-telegram-bot for Telegram integration (used by the bot server)

graph LR
WS["webhook_server.py"] --> FA["fastapi"]
WS --> UV["uvicorn"]
WS --> PD["pydantic"]
WS --> PDS["pydantic-settings"]
WS --> PYW["pywebpush"]
BOT["bot_server.py"] --> PTB["python-telegram-bot"]

• Dependency injection avoids repeated initialization overhead
• Database operations are performed synchronously within request scope
• Web Push broadcasting iterates through subscriptions; consider batching or async processing for scale
• Logging is configured centrally to minimize overhead

[No sources needed since this section provides general guidance]

Common issues and resolutions:

• Web push not configured
• Symptom: HTTP 501 on push endpoints
• Resolution: Set VAPID keys and ensure pywebpush is installed
• Notification service not configured
• Symptom: HTTP 501 on notify endpoints
• Resolution: Verify Telegram bot token and database connectivity
• Database connectivity failures
• Symptom: HTTP 501 on stats endpoints
• Resolution: Check MongoDB connection string and network access
• Webhook trigger failures
• Symptom: HTTP 500 on /webhook/update
• Resolution: Review logs for underlying exceptions

Webhook server is a thin FastAPI front for health, push, stats, and send. FastAPI validates requests; DI wires services. VAPID keys and a working MongoDB URI are required or push and stats routes will no-op.