From 0b69ce699ca5ee476ff2dfb4719c6b64b7068a57 Mon Sep 17 00:00:00 2001 From: JoaquinBN Date: Mon, 6 Jul 2026 13:36:26 +0200 Subject: [PATCH 01/12] Add Telegram bot for notifications, stats commands, and private linking Portal contributors can now link Telegram from their profile header via a one-time deep link to the portal bot. The connection is strictly private: the API serializes it for the owner (and staff) only, it never renders on other users' profiles, and the old manually-typed public Telegram handle was removed from profile editing, public payloads, and user search. High-signal notification events (submission reviews, highlights, validator graduation, missions, node versions, alerts) and admin campaigns with the new Telegram channel checkbox now also deliver to linked accounts through a message outbox. A cron-triggered drain endpoint sends at Telegram's rate budget with retries, dedupe (re-broadcasts and campaign resends never double-send), and automatic disabling of connections that blocked the bot. The bot answers /rank, /points, /missions, /mute, /unmute, /unlink and /help; every inbound and outbound message is logged, laying the groundwork for a future per-user agent experience. - backend/social_connections/models.py: TelegramConnection (extends SocialConnection; notifications_enabled, blocked_at) and TelegramMessage (message log + delivery outbox, partial unique (notification, connection) for idempotent enqueues); PendingOAuthState.consume() user_id now optional for token-only lookup - backend/social_connections/telegram.py: sendMessage client (HTML mode, escape/truncate, 403 marks blocked, 429 retry_after, parse-error plain-text fallback) - backend/social_connections/telegram_bot.py: link-token + disconnect endpoints, secret-validated webhook, handle_update command router (/start token consume, rank/points/missions/mute/unmute/unlink/help) - backend/social_connections/urls.py: /api/webhooks/telegram/, /api/v1/users/telegram/link-token/ and /disconnect/ - backend/social_connections/management/commands/: set_telegram_webhook, run_telegram_polling (local dev getUpdates loop) - backend/social_connections/{serializers,admin}.py + migrations/0005: owner-only TelegramConnectionSerializer (no public variant), admin registrations - backend/notifications/telegram.py: render_notification_text, enqueue_personal/broadcast/campaign, cancel_pending_for_campaign, deliver_pending (skip_locked claims, 25 msg/s pacing, 3 attempts, per-run reclaim guard) - backend/notifications/services.py: users_for_audience() (estimate_broadcast_reach refactored onto it); notify()/broadcast() enqueue when the event has the telegram channel - backend/notifications/campaigns.py: send_campaign enqueues on telegram channel; recall_campaign returns (deleted, cancelled) and cancels pending outbox rows - backend/notifications/views.py: cron-protected POST notifications/telegram/deliver/ action - backend/notifications/registry.py: telegram channel on submission.*, contribution.highlighted, validator.graduated, mission/node_version/alert published, custom.announcement - backend/notifications/admin.py: campaign channels checkboxes (portal forced on), recall messaging includes cancelled Telegram sends - backend/notifications/management/commands/deliver_telegram_messages.py: CLI drain wrapper - backend/users/serializers.py: telegram_connection SerializerMethodField (owner/staff only) + strip telegram_connection/telegram_handle for other viewers; telegram_handle removed from UserProfileUpdateSerializer - backend/users/views.py: search no longer matches telegram_handle - backend/tally/settings.py + backend/.env.example: TELEGRAM_BOT_USERNAME, TELEGRAM_WEBHOOK_SECRET - .github/workflows/telegram-deliver.yml: 5-minute cron drain via X-Cron-Token - frontend/src/components/TelegramLink.svelte: connect pill (deep link + poll) / connected pill with unlink - frontend/src/components/profile/ProfileHeader.svelte: TelegramLink rendered only in the isOwnProfile branch - frontend/src/lib/api.js: telegramAPI (getLinkToken, disconnect) - frontend/src/routes/ProfileEdit.svelte: Telegram text input removed - backend/social_connections/tests/test_telegram_{link,commands}.py, backend/users/tests/test_telegram_privacy.py, backend/notifications/tests.py: link/webhook/command/delivery/campaign coverage + mandatory privacy leak tests; users/tests/test_profile_update.py updated for the removed field - backend/CLAUDE.md, frontend/CLAUDE.md: endpoints, env vars, channel + component docs --- .github/workflows/telegram-deliver.yml | 32 ++ backend/.env.example | 8 +- backend/CLAUDE.md | 24 +- backend/notifications/admin.py | 55 ++- backend/notifications/campaigns.py | 19 +- .../commands/deliver_telegram_messages.py | 18 + backend/notifications/registry.py | 38 +- backend/notifications/services.py | 49 ++- backend/notifications/telegram.py | 294 +++++++++++++++ backend/notifications/tests.py | 293 +++++++++++++++ backend/notifications/views.py | 18 + backend/social_connections/admin.py | 24 ++ .../commands/run_telegram_polling.py | 50 +++ .../commands/set_telegram_webhook.py | 36 ++ ...0006_telegramconnection_telegrammessage.py | 60 ++++ backend/social_connections/models.py | 139 ++++++++ backend/social_connections/serializers.py | 19 +- backend/social_connections/telegram.py | 108 ++++++ backend/social_connections/telegram_bot.py | 335 ++++++++++++++++++ .../tests/test_telegram_commands.py | 250 +++++++++++++ .../tests/test_telegram_link.py | 271 ++++++++++++++ backend/social_connections/urls.py | 10 + backend/tally/settings.py | 10 +- backend/users/serializers.py | 26 +- backend/users/tests/test_profile_update.py | 21 +- backend/users/tests/test_telegram_privacy.py | 109 ++++++ backend/users/views.py | 3 +- frontend/CLAUDE.md | 4 +- frontend/src/components/TelegramLink.svelte | 255 +++++++++++++ .../components/profile/ProfileHeader.svelte | 7 + frontend/src/lib/api.js | 7 + frontend/src/routes/ProfileEdit.svelte | 26 -- 32 files changed, 2515 insertions(+), 103 deletions(-) create mode 100644 .github/workflows/telegram-deliver.yml create mode 100644 backend/notifications/management/commands/deliver_telegram_messages.py create mode 100644 backend/notifications/telegram.py create mode 100644 backend/social_connections/management/commands/run_telegram_polling.py create mode 100644 backend/social_connections/management/commands/set_telegram_webhook.py create mode 100644 backend/social_connections/migrations/0006_telegramconnection_telegrammessage.py create mode 100644 backend/social_connections/telegram.py create mode 100644 backend/social_connections/telegram_bot.py create mode 100644 backend/social_connections/tests/test_telegram_commands.py create mode 100644 backend/social_connections/tests/test_telegram_link.py create mode 100644 backend/users/tests/test_telegram_privacy.py create mode 100644 frontend/src/components/TelegramLink.svelte diff --git a/.github/workflows/telegram-deliver.yml b/.github/workflows/telegram-deliver.yml new file mode 100644 index 00000000..5f91409e --- /dev/null +++ b/.github/workflows/telegram-deliver.yml @@ -0,0 +1,32 @@ +name: Deliver Telegram Notifications + +on: + schedule: + # GitHub Actions schedules are best-effort; avoid crowded :00/:15/:30/:45 slots. + - cron: '4-59/5 * * * *' + workflow_dispatch: + +jobs: + deliver: + runs-on: ubuntu-latest + environment: cron-job + steps: + - name: Drain Telegram outbox + run: | + response=$(curl -s -w "\n%{http_code}" -X POST \ + -H "Content-Type: application/json" \ + -H "X-Cron-Token: ${{ secrets.CRON_SYNC_TOKEN }}" \ + "${{ secrets.API_BASE_URL }}/api/v1/notifications/telegram/deliver/") + + http_code=$(echo "$response" | tail -n1) + body=$(echo "$response" | sed '$d') + + echo "Response: $body" + echo "HTTP Code: $http_code" + + if [ "$http_code" = "200" ]; then + echo "Telegram delivery run completed" + else + echo "Telegram delivery failed with status $http_code" + exit 1 + fi diff --git a/backend/.env.example b/backend/.env.example index f0a1e9b2..5ca8b566 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -135,9 +135,15 @@ DEFILLAMA_FEES_RANK_URL=https://defillama.com/fees/chains # JSON array of curated validators, for example: # [{"name":"Validator","subtitle":"GenLayer validator","aum":"$42.6M","logo_url":"https://..."}] OVERVIEW_TOP_VALIDATORS= +# Telegram portal bot (notifications + commands). Token from BotFather; the +# same bot also feeds the member-count metric below when it is in the group. +TELEGRAM_BOT_TOKEN= +# Bot username without @, used for t.me account-linking deep links. +TELEGRAM_BOT_USERNAME= +# Random secret Telegram echoes back on webhook calls (openssl rand -hex 32). +TELEGRAM_WEBHOOK_SECRET= # Telegram members shown in the overview hero. Set bot token + chat id for the live count, # otherwise the backend falls back to TELEGRAM_MEMBERS or 13300. -TELEGRAM_BOT_TOKEN= TELEGRAM_CHAT_ID= TELEGRAM_MEMBERS=13300 # GenLayer Studio executive-metrics dashboard (decisions/chain-tx time series for the overview chart) diff --git a/backend/CLAUDE.md b/backend/CLAUDE.md index 5231a82c..c4b1c6ea 100644 --- a/backend/CLAUDE.md +++ b/backend/CLAUDE.md @@ -27,7 +27,7 @@ backend/ ├── api/ # Core API app ├── contributions/ # Contribution tracking ├── leaderboard/ # Leaderboard and rankings -├── social_connections/ # OAuth (GitHub, Twitter, Discord) + encrypted token storage +├── social_connections/ # OAuth (GitHub, Twitter, Discord) + Telegram bot link + encrypted token storage ├── social_tasks/ # Repeatable social tasks (follow, join, like) and completions ├── users/ # User management and auth ├── partners/ # Ecosystem partners directory @@ -46,13 +46,13 @@ backend/ - Validator model with node_version field (OneToOne with User) - Custom UserManager for email-based auth - **Views**: `users/views.py` - - `/api/v1/users/me/` - GET/PATCH current user profile (name/description/website/socials editable; node_version is NOT editable — Grafana-sourced, display only) + - `/api/v1/users/me/` - GET/PATCH current user profile (name/description/website/linkedin editable; telegram_handle removed — Telegram links via the bot's verified TelegramConnection; node_version is NOT editable — Grafana-sourced, display only) - `/api/v1/users/by-address/{address}/` - Get user by wallet address - `/api/v1/users/validators/` - Get validator list from blockchain - **Serializers**: `users/serializers.py` - UserSerializer - Full user data including validator info - ValidatorSerializer - Validator node version and target matching - - UserProfileUpdateSerializer - Allows name/description/website/socials updates (node_version removed — Grafana is source of truth) + - UserProfileUpdateSerializer - Allows name/description/website/linkedin updates (telegram_handle removed — replaced by the private TelegramConnection; node_version removed — Grafana is source of truth) - UserCreateSerializer - Registration ### Authentication @@ -306,9 +306,9 @@ backend/ - **Models**: `notifications/models.py` - `Notification` - Personal (has `recipient`) or broadcast (`recipient=None` + `audience`: all/validators/stewards/builders/community). Audiences resolve via the role OneToOnes (Validator/Steward/Builder/Creator) in `services.audiences_for`. Broadcasts are ONE row regardless of user count; users see broadcasts created after their `date_joined`. Frozen copy (`title`/`body`/`link_url`), `payload` JSON for future channel renderers, `dedupe_key` (re-broadcasting a source object refreshes + resurfaces instead of duplicating). - `NotificationReceipt` - Lazy per-user read state for broadcast rows (created on read). - - `CustomNotification` - Admin-composed campaign: title/markdown body/optional link + targeting (`everyone` | `roles` union of builders/validators/stewards/creators | hand-picked `target_users` M2M | pasted `target_wallets`) + delivery record (`status` draft/sent, `sent_count`, `unmatched_wallets`, `channels` reserved for email/Telegram). + - `CustomNotification` - Admin-composed campaign: title/markdown body/optional link + targeting (`everyone` | `roles` union of builders/validators/stewards/creators | hand-picked `target_users` M2M | pasted `target_wallets`) + delivery record (`status` draft/sent, `sent_count`, `unmatched_wallets`, `channels` JSON list: `portal` always + optional `telegram` checkbox in the admin form). - **Campaigns**: `notifications/campaigns.py` - - `resolve_recipients(campaign)` - The channel-agnostic enumeration step (always `is_active=True`; banned/invisible users included by design). Future email/Telegram channels reuse this and add their own delivery. + - `resolve_recipients(campaign)` - The channel-agnostic enumeration step (always `is_active=True`; banned/invisible users included by design). The Telegram channel reuses this (`notifications/telegram.py:enqueue_campaign`); a future email channel would do the same. - `send_campaign(campaign, actor=...)` - Fans out personal `Notification` rows (snapshot semantics, never broadcast rows, so campaigns stay private to recipients). Idempotent via dedupe key `custom.announcement:{pk}`; resend refreshes copy + resurfaces unread, scoped to the currently resolved audience. - `recall_campaign(campaign)` - Deletes delivered portal notification rows for that campaign while keeping the campaign record for audit/resend. - Compose flow: Django admin > Notifications > Custom notifications. Saving is a silent draft with reach preview; off-by-default "Send now" checkbox or `send_selected`/`resend_selected` actions deliver. "Recall delivered portal notifications" or the `recall_selected` action removes delivered portal rows. The send/recall runs in `save_related` (M2M targeting commits after `save_model`). @@ -322,7 +322,9 @@ backend/ - `/api/v1/notifications/unread-count/` - Unread badge count - `/api/v1/notifications/{id}/mark-read/` - Personal sets `read_at`; broadcast creates a receipt - `/api/v1/notifications/mark-all-read/` -- **Future channels**: email/Telegram slot in via registry `channels` + a delivery outbox and `NotificationPreference` model when the first external channel ships (Telegram link would follow the `social_connections` pattern). +- **Telegram channel** (`notifications/telegram.py`): events whose registry entry includes `'telegram'` in `channels` (all `submission.*`, `contribution.highlighted`, `validator.graduated`, `mission.published`, `node_version.published`, `alert.published`, `custom.announcement`) also enqueue rows into the `social_connections.TelegramMessage` outbox for recipients with a linked, unmuted, unblocked `TelegramConnection`. `notify()` enqueues per recipient; `broadcast()` fans out via `services.users_for_audience(audience)` (the queryset twin of `estimate_broadcast_reach`); `send_campaign()` gates on `campaign.channels`. Enqueues are idempotent (partial unique constraint on `(notification, connection)` — re-broadcasts/resends only reach users who linked after the original send) and best-effort (failures never break notification creation). The outbox is drained by `POST /api/v1/notifications/telegram/deliver/` (IsCronToken; GitHub Action `telegram-deliver.yml` every 5 min; ~25 msg/s pacing, 3 attempts, 429-aware) or `python manage.py deliver_telegram_messages`. Campaign recall cancels pending outbox rows; already-sent Telegram messages cannot be recalled. +- **Telegram bot** (`social_connections/telegram_bot.py` + `telegram.py`): deep-link account linking (portal issues a one-time `PendingOAuthState` token; `https://t.me/?start=`; the webhook binds the sender's numeric Telegram id to a `TelegramConnection`), webhook `POST /api/webhooks/telegram/` (validated via `X-Telegram-Bot-Api-Secret-Token` == `TELEGRAM_WEBHOOK_SECRET`, always 200 after auth), and one-shot commands: `/rank`, `/points` (LeaderboardEntry), `/missions` (active Missions), `/mute`/`/unmute` (`notifications_enabled`), `/unlink`, `/help`. All inbound/outbound messages are logged in `TelegramMessage`. **Privacy: TelegramConnection is owner-only** — `UserSerializer.get_telegram_connection` returns it for owner/staff only, `to_representation` strips `telegram_connection`/`telegram_handle` for other viewers, user search does not match `telegram_handle`, and `UserProfileUpdateSerializer` no longer accepts `telegram_handle` (replaced by the verified connection). Dev: `python manage.py run_telegram_polling` (getUpdates loop, no public URL needed); prod one-time setup: `python manage.py set_telegram_webhook --url https:///api/webhooks/telegram/`. +- **Future channels**: email slots in the same way via registry `channels` + its own delivery module; per-user `NotificationPreference` is still future work (Telegram has a single `/mute`). ### Gen TV - **Models**: `gen_tv/models.py` @@ -504,6 +506,12 @@ GET /api/v1/notifications/ (requires auth, ?unread=true ?categor GET /api/v1/notifications/unread-count/ (requires auth) POST /api/v1/notifications/{id}/mark-read/ (requires auth) POST /api/v1/notifications/mark-all-read/ (requires auth) +POST /api/v1/notifications/telegram/deliver/ (cron-protected, X-Cron-Token; drains the Telegram outbox) + +# Telegram bot +POST /api/v1/users/telegram/link-token/ (requires auth; returns one-time t.me deep link) +POST /api/v1/users/telegram/disconnect/ (requires auth; idempotent unlink) +POST /api/webhooks/telegram/ (Telegram only; X-Telegram-Bot-Api-Secret-Token) ``` ### Leaderboard monthly date ranges @@ -558,7 +566,9 @@ The cron `POST /api/v1/metrics/overview/refresh/` (GitHub Action `sync-overview- - `OVERVIEW_TOP_VALIDATORS` - optional JSON array of curated validators; superseded by the per-wallet `ValidatorWallet.show_in_overview` + `assets_under_management_usd` admin fields when any are set. - `DEFILLAMA_FEES_RANK` / `DEFILLAMA_FEES_RANK_URL` - the DeFiLlama fees-rank value/source shown on the overview. - `DISCORD_BOT_TOKEN` + `DISCORD_GUILD_ID` (Discord members), `SORSA_API_KEY` + `X_METRICS_USERNAME` (X followers), `GITHUB_METRICS_REPO` + `GITHUB_METRICS_TOKEN` (boilerplate stars). -- `TELEGRAM_BOT_TOKEN` + `TELEGRAM_CHAT_ID` for the live Telegram member count, else `TELEGRAM_MEMBERS` or the built-in `13300` curated fallback. +- `TELEGRAM_BOT_TOKEN` + `TELEGRAM_CHAT_ID` for the live Telegram member count, else `TELEGRAM_MEMBERS` or the built-in `13300` curated fallback. The same token powers the portal Telegram bot (notifications + commands); the bot must be a member of the chat for the count to work. +- `TELEGRAM_BOT_USERNAME` - Portal bot username without `@`; used to build `t.me/?start=` account-linking deep links (link-token endpoint returns 503 when unset). +- `TELEGRAM_WEBHOOK_SECRET` - Random secret registered via `set_telegram_webhook`; Telegram echoes it in `X-Telegram-Bot-Api-Secret-Token` and the webhook rejects everything when it is unset or mismatched. **AWS Deployment:** For production deployments on AWS App Runner, all environment variables must be stored in AWS Systems Manager (SSM) Parameter Store. See `aws-deployment-guide.md` for setup instructions. diff --git a/backend/notifications/admin.py b/backend/notifications/admin.py index d6069d28..b94b0d69 100644 --- a/backend/notifications/admin.py +++ b/backend/notifications/admin.py @@ -60,6 +60,17 @@ class CustomNotificationAdminForm(forms.ModelForm): widget=forms.CheckboxSelectMultiple, help_text='Roles mode only. Union: a user with any selected role receives it once.', ) + channels = forms.MultipleChoiceField( + choices=[('portal', 'Portal'), ('telegram', 'Telegram')], + initial=['portal'], + required=False, + widget=forms.CheckboxSelectMultiple, + help_text=( + 'Portal is always included (it is the delivery record Telegram ' + 'fans out from). Telegram reaches recipients with a linked, ' + 'unmuted Telegram account.' + ), + ) send_now = forms.BooleanField( required=False, label='Send now', @@ -68,7 +79,11 @@ class CustomNotificationAdminForm(forms.ModelForm): recall_now = forms.BooleanField( required=False, label='Recall delivered portal notifications', - help_text='Deletes delivered portal notification rows for this campaign; the campaign record stays for audit.', + help_text=( + 'Deletes delivered portal notification rows and cancels queued ' + 'Telegram deliveries; already-sent Telegram messages cannot be ' + 'recalled. The campaign record stays for audit.' + ), ) class Meta: @@ -76,6 +91,7 @@ class Meta: fields = [ 'title', 'body', 'link_url', 'link_label', 'priority', 'target_mode', 'target_roles', 'target_users', 'target_wallets', + 'channels', ] widgets = { 'target_mode': forms.RadioSelect, @@ -101,6 +117,13 @@ def clean(self): detail = f" Invalid lines: {', '.join(invalid_lines[:5])}" if invalid_lines else '' self.add_error('target_wallets', f'Paste at least one valid wallet address.{detail}') + # Portal is the fan-out backbone: without its rows there is nothing + # for the Telegram channel to enqueue from. + channels = list(cleaned.get('channels') or []) + if 'portal' not in channels: + channels.insert(0, 'portal') + cleaned['channels'] = channels + # Keep stored targeting unambiguous: clear the fields that don't # belong to the chosen mode (the users M2M clears in save_related). if mode != CustomNotification.TARGET_ROLES: @@ -118,7 +141,7 @@ class CustomNotificationAdmin(admin.ModelAdmin): list_filter = ('status', 'target_mode', 'priority') search_fields = ('title', 'body') readonly_fields = ( - 'audience_preview', 'channels_display', 'status', 'sent_at', 'sent_by', + 'audience_preview', 'status', 'sent_at', 'sent_by', 'sent_count', 'unmatched_report', 'created_at', 'updated_at', ) actions = ('send_selected', 'resend_selected', 'recall_selected') @@ -132,7 +155,7 @@ class CustomNotificationAdmin(admin.ModelAdmin): 'description': 'Fill only the field that matches the chosen mode; the others are ignored.', }), ('Send', { - 'fields': ('send_now', 'recall_now', 'audience_preview', 'channels_display'), + 'fields': ('channels', 'send_now', 'recall_now', 'audience_preview'), }), ('Delivery record', { 'fields': ('status', 'sent_at', 'sent_by', 'sent_count', 'unmatched_report'), @@ -150,11 +173,6 @@ def audience_preview(self, obj): preview += f' · {len(audience.unmatched_wallets)} wallet line(s) unmatched' return preview - @admin.display(description='Channels') - def channels_display(self, obj): - channels = obj.channels if obj and obj.pk else ['portal'] - return ', '.join(channels) - @admin.display(description='Unmatched wallets') def unmatched_report(self, obj): if not obj or not obj.unmatched_wallets: @@ -222,7 +240,7 @@ def _send_campaign(self, request, campaign): def _recall_campaign(self, request, campaign): try: - deleted = campaigns.recall_campaign(campaign) + deleted, cancelled = campaigns.recall_campaign(campaign) except Exception: logger.exception('Failed to recall campaign %r', campaign) self.message_user( @@ -232,12 +250,15 @@ def _recall_campaign(self, request, campaign): ) return - if deleted: - self.message_user( - request, - f'Recalled {deleted} delivered portal notification(s). The campaign record was kept.', - level=messages.SUCCESS, + if deleted or cancelled: + message = f'Recalled {deleted} delivered portal notification(s).' + if cancelled: + message += f' Cancelled {cancelled} queued Telegram message(s).' + message += ( + ' Already-sent Telegram messages cannot be recalled.' + ' The campaign record was kept.' ) + self.message_user(request, message, level=messages.SUCCESS) else: self.message_user( request, @@ -263,15 +284,17 @@ def resend_selected(self, request, queryset): def recall_selected(self, request, queryset): campaigns_recalled = 0 deleted = 0 + cancelled = 0 failed = 0 for campaign in queryset: try: - count = campaigns.recall_campaign(campaign) + count, cancelled_count = campaigns.recall_campaign(campaign) except Exception: logger.exception('Failed to recall campaign %r', campaign) failed += 1 continue + cancelled += cancelled_count if count: campaigns_recalled += 1 deleted += count @@ -280,6 +303,8 @@ def recall_selected(self, request, queryset): f'Recalled {deleted} delivered portal notification(s) ' f'from {campaigns_recalled} custom notification(s).' ) + if cancelled: + message += f' Cancelled {cancelled} queued Telegram message(s).' if failed: message += f' Failed {failed}; check server logs.' self.message_user( diff --git a/backend/notifications/campaigns.py b/backend/notifications/campaigns.py index 18cc0204..a0f009c5 100644 --- a/backend/notifications/campaigns.py +++ b/backend/notifications/campaigns.py @@ -171,6 +171,13 @@ def send_campaign(campaign, *, actor=None): if batch: Notification.objects.bulk_create(batch, ignore_conflicts=True) + if 'telegram' in (campaign.channels or []): + # Deduped per (notification, connection): a resend keeps the same + # notification rows, so it never re-pushes to Telegram. Recall + + # resend (new rows) is the deliberate way to push again. + from . import telegram + telegram.enqueue_campaign(campaign, audience.users) + campaign.status = CustomNotification.STATUS_SENT campaign.sent_at = now campaign.sent_by = actor @@ -191,10 +198,14 @@ def send_campaign(campaign, *, actor=None): def recall_campaign(campaign): """Delete delivered portal notifications for a custom campaign. - The campaign record is kept for audit and can be resent later. Future - email/Telegram channels should add their own outbox recall/cancel logic - next to this portal-row deletion. + The campaign record is kept for audit and can be resent later. Queued + (not yet sent) Telegram deliveries are cancelled first; messages Telegram + already delivered cannot be recalled. """ + from . import telegram + + cancelled = telegram.cancel_pending_for_campaign(campaign) + queryset = Notification.objects.filter( event_type='custom.announcement', dedupe_key=campaign.dedupe_key, @@ -204,4 +215,4 @@ def recall_campaign(campaign): ) count = queryset.count() queryset.delete() - return count + return count, cancelled diff --git a/backend/notifications/management/commands/deliver_telegram_messages.py b/backend/notifications/management/commands/deliver_telegram_messages.py new file mode 100644 index 00000000..c68840a6 --- /dev/null +++ b/backend/notifications/management/commands/deliver_telegram_messages.py @@ -0,0 +1,18 @@ +"""Drain the Telegram delivery outbox from the CLI (dev/ops convenience; +production uses the cron-triggered /api/v1/notifications/telegram/deliver/).""" +from django.core.management.base import BaseCommand + +from notifications.telegram import DEFAULT_RUN_LIMIT, deliver_pending + + +class Command(BaseCommand): + help = "Send pending Telegram notification messages." + + def add_arguments(self, parser): + parser.add_argument('--limit', type=int, default=DEFAULT_RUN_LIMIT) + + def handle(self, *args, **options): + stats = deliver_pending(limit=options['limit']) + self.stdout.write( + f"sent={stats['sent']} failed={stats['failed']} remaining={stats['remaining']}" + ) diff --git a/backend/notifications/registry.py b/backend/notifications/registry.py index a8f22ffb..9b74ca4e 100644 --- a/backend/notifications/registry.py +++ b/backend/notifications/registry.py @@ -27,22 +27,33 @@ class EventType: _EVENT_TYPES = [ # --- Personal, automatic --- - EventType('submission.accepted', category='submission'), - EventType('submission.rejected', category='submission', priority=Notification.PRIORITY_HIGH), - EventType('submission.more_info_needed', category='submission', priority=Notification.PRIORITY_HIGH), - EventType('submission.proposal_questioned', category='submission', priority=Notification.PRIORITY_HIGH), - EventType('submission.appealed', category='submission', priority=Notification.PRIORITY_HIGH), - EventType('submission.more_info_resubmitted', category='submission', priority=Notification.PRIORITY_HIGH), - EventType('contribution.highlighted', category='contribution'), + # High-signal events also push to linked Telegram accounts; content-noise + # events stay portal-only. Flipping one is a one-line channels change. + EventType('submission.accepted', category='submission', + channels=('portal', 'telegram')), + EventType('submission.rejected', category='submission', priority=Notification.PRIORITY_HIGH, + channels=('portal', 'telegram')), + EventType('submission.more_info_needed', category='submission', priority=Notification.PRIORITY_HIGH, + channels=('portal', 'telegram')), + EventType('submission.proposal_questioned', category='submission', priority=Notification.PRIORITY_HIGH, + channels=('portal', 'telegram')), + EventType('submission.appealed', category='submission', priority=Notification.PRIORITY_HIGH, + channels=('portal', 'telegram')), + EventType('submission.more_info_resubmitted', category='submission', priority=Notification.PRIORITY_HIGH, + channels=('portal', 'telegram')), + EventType('contribution.highlighted', category='contribution', + channels=('portal', 'telegram')), EventType('referral.joined', category='community'), - EventType('validator.graduated', category='validator', priority=Notification.PRIORITY_HIGH), + EventType('validator.graduated', category='validator', priority=Notification.PRIORITY_HIGH, + channels=('portal', 'telegram')), EventType('email.verify_reminder', category='system', priority=Notification.PRIORITY_HIGH), # --- Broadcast, admin-explicit --- EventType('featured.published', category='content'), EventType('partner.published', category='content'), EventType('contribution_type.published', category='content'), - EventType('mission.published', category='content'), + EventType('mission.published', category='content', + channels=('portal', 'telegram')), EventType('stream.published', category='content'), EventType('poap.published', category='content'), # Audience resolved per task category (builders/validators/community) @@ -53,11 +64,16 @@ class EventType: category='validator', priority=Notification.PRIORITY_HIGH, audience=Notification.AUDIENCE_VALIDATORS, + channels=('portal', 'telegram'), ), - EventType('alert.published', category='system', priority=Notification.PRIORITY_HIGH), + EventType('alert.published', category='system', priority=Notification.PRIORITY_HIGH, + channels=('portal', 'telegram')), # --- Campaigns (admin-composed, fan-out via notifications.campaigns) --- - EventType('custom.announcement', category='announcement'), + # Campaign Telegram delivery gates on campaign.channels, not this entry; + # listed here so the registry stays the single source of truth. + EventType('custom.announcement', category='announcement', + channels=('portal', 'telegram')), ] EVENT_TYPES = {} diff --git a/backend/notifications/services.py b/backend/notifications/services.py index 4a30c6e3..932b727d 100644 --- a/backend/notifications/services.py +++ b/backend/notifications/services.py @@ -107,9 +107,15 @@ def notify( if not created: # Same event delivered twice: refresh the copy, keep read state. _apply_values(notification, values) - return notification + else: + notification = Notification.objects.create(recipient=recipient, **values) - return Notification.objects.create(recipient=recipient, **values) + if 'telegram' in event.channels: + # Idempotent (deduped per connection) and best-effort: a Telegram + # failure never breaks portal notification creation. + from . import telegram + telegram.enqueue_personal(notification) + return notification def broadcast( @@ -162,9 +168,15 @@ def broadcast( Notification.objects.filter(pk=notification.pk).update(created_at=timezone.now()) notification.receipts.all().delete() notification.refresh_from_db(fields=['created_at']) - return notification + else: + notification = Notification.objects.create(recipient=None, **values) - return Notification.objects.create(recipient=None, **values) + if 'telegram' in event.channels: + # Deduped per (notification, connection): a re-broadcast only reaches + # users who linked Telegram after the original send. + from . import telegram + telegram.enqueue_broadcast(notification) + return notification def recall_broadcast(event_slug, source): @@ -183,22 +195,29 @@ def recall_broadcast(event_slug, source): return count -def estimate_broadcast_reach(audience): - """Approximate audience size, used for admin feedback messages.""" +def users_for_audience(audience): + """Active users belonging to a broadcast audience, as a User queryset. + + The queryset twin of estimate_broadcast_reach (which is defined on top of + this so the two can never drift); external delivery channels use it to + fan a broadcast out into per-user sends. + """ User = get_user_model() + active_users = User.objects.filter(is_active=True) if audience == Notification.AUDIENCE_VALIDATORS: - from validators.models import Validator - return Validator.objects.filter(user__is_active=True).count() + return active_users.filter(validator__isnull=False) if audience == Notification.AUDIENCE_STEWARDS: - from stewards.models import Steward - return Steward.objects.filter(user__is_active=True).count() + return active_users.filter(steward__isnull=False) if audience == Notification.AUDIENCE_BUILDERS: - from builders.models import Builder - return Builder.objects.filter(user__is_active=True).count() + return active_users.filter(builder__isnull=False) if audience == Notification.AUDIENCE_COMMUNITY: - from creators.models import Creator - return Creator.objects.filter(user__is_active=True).count() - return User.objects.filter(is_active=True).count() + return active_users.filter(creator__isnull=False) + return active_users + + +def estimate_broadcast_reach(audience): + """Approximate audience size, used for admin feedback messages.""" + return users_for_audience(audience).count() # --------------------------------------------------------------------------- diff --git a/backend/notifications/telegram.py b/backend/notifications/telegram.py new file mode 100644 index 00000000..2bdffa9d --- /dev/null +++ b/backend/notifications/telegram.py @@ -0,0 +1,294 @@ +"""Telegram delivery channel for portal notifications. + +Enqueue helpers are called from services.notify()/broadcast() and +campaigns.send_campaign() when the event or campaign includes the +'telegram' channel. They write pending rows into the TelegramMessage +outbox; deliver_pending() drains it, triggered by the cron-protected +/api/v1/notifications/telegram/deliver/ endpoint (or the +deliver_telegram_messages management command). + +Enqueueing is best-effort and idempotent: failures are logged and swallowed +so Telegram can never break portal notification creation, and the partial +unique constraint on (notification, connection) makes repeated enqueues +(re-broadcasts, campaign resends) no-ops. +""" +import logging +import time +from datetime import timedelta + +from django.conf import settings +from django.db import connection as db_connection +from django.db import transaction +from django.db.models import F, Q +from django.utils import timezone + +from social_connections.models import TelegramConnection, TelegramMessage +from social_connections.telegram import TELEGRAM_MAX_LEN, escape, send_telegram_message + +logger = logging.getLogger(__name__) + +# Telegram's global budget is ~30 messages/second; one 25-row batch per second +# keeps a safety margin. ponytail: fixed pacing, adaptive throttling only if +# broadcasts ever grow past a few thousand linked users. +CLAIM_BATCH_SIZE = 25 +DEFAULT_RUN_LIMIT = 400 +MAX_ATTEMPTS = 3 +STALE_SENDING_MINUTES = 10 + + +# --------------------------------------------------------------------------- +# Rendering +# --------------------------------------------------------------------------- + +def render_notification_text(notification): + """Render a Notification row as a Telegram HTML message.""" + header = f"{escape(notification.title)}" + + link_html = '' + url = notification.link_url or '' + if url: + if url.startswith('#/'): + url = url[1:] + if not url.startswith(('http://', 'https://')): + url = f"{settings.FRONTEND_URL}{url}" + label = escape(notification.link_label or 'Open portal') + link_html = f'\n\n{label}' + + body = (notification.body or '').strip() + # Truncate the raw body so title and link always survive. Escaping can + # still overshoot the limit in pathological cases; send_telegram_message's + # final truncate + plain-text parse fallback covers those. + budget = TELEGRAM_MAX_LEN - len(header) - len(link_html) - 2 + if body and budget > 20: + if len(body) > budget: + body = body[: budget - 1] + '…' + return f"{header}\n\n{escape(body)}{link_html}" + return f"{header}{link_html}" + + +# --------------------------------------------------------------------------- +# Enqueue +# --------------------------------------------------------------------------- + +def eligible_connections(users): + return TelegramConnection.objects.filter( + user__in=users, + notifications_enabled=True, + blocked_at__isnull=True, + ) + + +def enqueue_personal(notification): + """Queue a personal notification for its recipient, if linked.""" + try: + conn = ( + TelegramConnection.objects + .filter( + user=notification.recipient, + notifications_enabled=True, + blocked_at__isnull=True, + ) + .first() + ) + if conn is None: + return + TelegramMessage.objects.get_or_create( + notification=notification, + connection=conn, + direction=TelegramMessage.DIRECTION_OUT, + defaults={ + 'chat_id': conn.platform_user_id, + 'text': render_notification_text(notification), + 'status': TelegramMessage.STATUS_PENDING, + }, + ) + except Exception: + logger.exception("Telegram enqueue_personal failed (notification %s)", notification.pk) + + +def enqueue_broadcast(notification): + """Queue a broadcast for every linked user in its audience. + + On re-broadcast the dedupe constraint skips users already queued/sent, + so only users who linked Telegram after the original broadcast get it. + """ + try: + from .services import users_for_audience + + text = render_notification_text(notification) + batch = [] + connections = eligible_connections(users_for_audience(notification.audience)) + for conn in connections.iterator(chunk_size=500): + batch.append(TelegramMessage( + direction=TelegramMessage.DIRECTION_OUT, + connection=conn, + chat_id=conn.platform_user_id, + text=text, + status=TelegramMessage.STATUS_PENDING, + notification=notification, + )) + if len(batch) >= 500: + TelegramMessage.objects.bulk_create(batch, ignore_conflicts=True) + batch = [] + if batch: + TelegramMessage.objects.bulk_create(batch, ignore_conflicts=True) + except Exception: + logger.exception("Telegram enqueue_broadcast failed (notification %s)", notification.pk) + + +def enqueue_campaign(campaign, users): + """Queue a campaign's fanned-out personal rows for linked recipients.""" + try: + from .models import Notification + + connections = {c.user_id: c for c in eligible_connections(users)} + if not connections: + return + notifications = Notification.objects.filter( + dedupe_key=campaign.dedupe_key, + recipient_id__in=connections.keys(), + ) + batch = [] + for notification in notifications.iterator(chunk_size=500): + conn = connections[notification.recipient_id] + batch.append(TelegramMessage( + direction=TelegramMessage.DIRECTION_OUT, + connection=conn, + chat_id=conn.platform_user_id, + text=render_notification_text(notification), + status=TelegramMessage.STATUS_PENDING, + notification=notification, + )) + if len(batch) >= 500: + TelegramMessage.objects.bulk_create(batch, ignore_conflicts=True) + batch = [] + if batch: + TelegramMessage.objects.bulk_create(batch, ignore_conflicts=True) + except Exception: + logger.exception("Telegram enqueue_campaign failed (campaign %s)", campaign.pk) + + +def cancel_pending_for_campaign(campaign): + """Delete not-yet-sent outbox rows on campaign recall. Sent messages + cannot be recalled from Telegram.""" + from .models import Notification + + return TelegramMessage.objects.filter( + notification__in=Notification.objects.filter(dedupe_key=campaign.dedupe_key), + status__in=[TelegramMessage.STATUS_PENDING, TelegramMessage.STATUS_SENDING], + ).delete()[0] + + +# --------------------------------------------------------------------------- +# Drain +# --------------------------------------------------------------------------- + +def _claim_batch(size, exclude_pks): + """Atomically claim up to `size` deliverable rows (multi-worker safe). + + exclude_pks keeps rows that already failed retryably in THIS run from + being re-claimed immediately; they retry on the next cron tick instead of + burning all their attempts against the same transient failure. + """ + now = timezone.now() + stale_cutoff = now - timedelta(minutes=STALE_SENDING_MINUTES) + with transaction.atomic(): + queryset = ( + TelegramMessage.objects + .filter(direction=TelegramMessage.DIRECTION_OUT, attempts__lt=MAX_ATTEMPTS) + .filter( + Q(status=TelegramMessage.STATUS_PENDING) + | Q(status=TelegramMessage.STATUS_SENDING, updated_at__lt=stale_cutoff) + ) + .exclude(pk__in=exclude_pks) + .order_by('created_at') + ) + if db_connection.features.has_select_for_update_skip_locked: + queryset = queryset.select_for_update(skip_locked=True) + elif db_connection.features.has_select_for_update: + queryset = queryset.select_for_update() + pks = list(queryset.values_list('pk', flat=True)[:size]) + if pks: + TelegramMessage.objects.filter(pk__in=pks).update( + status=TelegramMessage.STATUS_SENDING, + attempts=F('attempts') + 1, + updated_at=now, + ) + if not pks: + return [] + return list( + TelegramMessage.objects + .filter(pk__in=pks) + .select_related('connection') + .order_by('created_at') + ) + + +def _finish(message, status, error=''): + message.status = status + message.error = error[:200] + message.sent_at = timezone.now() if status == TelegramMessage.STATUS_SENT else message.sent_at + message.save(update_fields=['status', 'error', 'sent_at', 'updated_at']) + + +def deliver_pending(limit=DEFAULT_RUN_LIMIT): + """Send queued messages, pacing to Telegram's rate budget. + + Returns {'sent': n, 'failed': n, 'remaining': n}. On a 429 the whole + run stops and unclaimed rows return to pending; the next cron tick + resumes. Delivery is at-least-once, bounded by MAX_ATTEMPTS. + """ + sent = failed = processed = 0 + rate_limited = False + seen_pks = set() + + while processed < limit and not rate_limited: + batch = _claim_batch(min(CLAIM_BATCH_SIZE, limit - processed), seen_pks) + if not batch: + break + processed += len(batch) + seen_pks.update(message.pk for message in batch) + batch_started = time.monotonic() + + for index, message in enumerate(batch): + conn = message.connection + if conn is None or not conn.notifications_enabled or conn.blocked_at: + _finish(message, TelegramMessage.STATUS_FAILED, 'connection_gone') + failed += 1 + continue + + # Send to the connection's LIVE chat id: if the user re-linked a + # different Telegram account, message.chat_id is a stale record. + ok, retry_after, description = send_telegram_message( + conn.platform_user_id, message.text, connection=conn + ) + if ok: + _finish(message, TelegramMessage.STATUS_SENT) + sent += 1 + elif retry_after is not None: + # Rate limited: unclaim this row and the rest of the batch, + # stop the run. The next cron tick resumes. + remaining_pks = [m.pk for m in batch[index:]] + TelegramMessage.objects.filter(pk__in=remaining_pks).update( + status=TelegramMessage.STATUS_PENDING, + attempts=F('attempts') - 1, + ) + rate_limited = True + logger.warning("Telegram rate limited (retry_after=%s); run stopped", retry_after) + break + else: + if message.attempts >= MAX_ATTEMPTS: + _finish(message, TelegramMessage.STATUS_FAILED, description) + failed += 1 + else: + _finish(message, TelegramMessage.STATUS_PENDING, description) + + elapsed = time.monotonic() - batch_started + if elapsed < 1 and processed < limit and not rate_limited: + time.sleep(1 - elapsed) + + remaining = TelegramMessage.objects.filter( + direction=TelegramMessage.DIRECTION_OUT, + status=TelegramMessage.STATUS_PENDING, + ).count() + return {'sent': sent, 'failed': failed, 'remaining': remaining} diff --git a/backend/notifications/tests.py b/backend/notifications/tests.py index c9f1018b..f4cd1125 100644 --- a/backend/notifications/tests.py +++ b/backend/notifications/tests.py @@ -1308,3 +1308,296 @@ def test_clear_removes_reminder_after_verification(self): services.clear_email_verification_reminder(self.unverified) self.assertEqual(Notification.objects.filter(recipient=self.unverified).count(), 0) + + +class TelegramEnqueueTests(TestCase): + """notify()/broadcast() enqueue Telegram outbox rows for telegram-channel events.""" + + def setUp(self): + from social_connections.models import TelegramConnection + self.linked = make_user('linked@test.com', '0x1111111111111111111111111111111111111111') + self.unlinked = make_user('unlinked@test.com', '0x2222222222222222222222222222222222222222') + self.connection = TelegramConnection.objects.create( + user=self.linked, + platform_user_id='111', + platform_username='linked', + linked_at=timezone.now(), + ) + + def outbox(self): + from social_connections.models import TelegramMessage + return TelegramMessage.objects.filter( + direction=TelegramMessage.DIRECTION_OUT, + status=TelegramMessage.STATUS_PENDING, + ) + + def test_personal_telegram_event_enqueues_for_linked_recipient(self): + services.notify( + 'submission.accepted', + recipient=self.linked, + title='Submission accepted + +{#if connection} + + {@html icon} + + {connection.platform_username ? `@${connection.platform_username}` : "Telegram"} + + + +{:else} + +{/if} + + diff --git a/frontend/src/components/profile/ProfileHeader.svelte b/frontend/src/components/profile/ProfileHeader.svelte index 7287ede5..3667146a 100644 --- a/frontend/src/components/profile/ProfileHeader.svelte +++ b/frontend/src/components/profile/ProfileHeader.svelte @@ -5,6 +5,7 @@ import EmailVerificationModal from "../EmailVerificationModal.svelte"; import CategoryIcon from "../portal/CategoryIcon.svelte"; import SocialLink from "../SocialLink.svelte"; + import TelegramLink from "../TelegramLink.svelte"; import { hasStartedJourney } from "../../lib/roleState.js"; let { @@ -303,6 +304,12 @@ onLinked={onParticipantUpdated} compact={true} /> + + {:else} {#if participant?.github_connection?.platform_username} diff --git a/frontend/src/lib/api.js b/frontend/src/lib/api.js index 3fd670b8..3bb01932 100644 --- a/frontend/src/lib/api.js +++ b/frontend/src/lib/api.js @@ -312,6 +312,13 @@ export const socialAPI = { checkDiscordGuild: () => api.get('/users/discord/check-guild/'), }; +// Telegram bot linking (deep-link flow, no OAuth). The connection is +// private: the API only ever returns it for the user's own profile. +export const telegramAPI = { + getLinkToken: () => api.post('/users/telegram/link-token/'), + disconnect: () => api.post('/users/telegram/disconnect/'), +}; + // Social tasks API export const socialTasksAPI = { list: (params = {}) => api.get('/social-tasks/', { params }), diff --git a/frontend/src/routes/ProfileEdit.svelte b/frontend/src/routes/ProfileEdit.svelte index fd847f4e..68498dc2 100644 --- a/frontend/src/routes/ProfileEdit.svelte +++ b/frontend/src/routes/ProfileEdit.svelte @@ -23,7 +23,6 @@ let email = $state(""); let description = $state(""); let website = $state(""); - let telegramHandle = $state(""); let linkedinHandle = $state(""); let profileImageUrl = $state(""); let bannerImageUrl = $state(""); @@ -81,7 +80,6 @@ (name !== (user.name || "") || description !== (user.description || "") || website !== (user.website || "") || - telegramHandle !== (user.telegram_handle || "") || linkedinHandle !== (user.linkedin_handle || "")), ); @@ -113,7 +111,6 @@ email = rawEmail.endsWith("@ethereum.address") ? "" : rawEmail; description = userData.description || ""; website = userData.website || ""; - telegramHandle = userData.telegram_handle || ""; linkedinHandle = userData.linkedin_handle || ""; profileImageUrl = userData.profile_image_url || ""; bannerImageUrl = userData.banner_image_url || ""; @@ -191,7 +188,6 @@ name: name.trim(), description: description.trim(), website: website.trim(), - telegram_handle: telegramHandle.trim(), linkedin_handle: linkedinHandle.trim(), }; @@ -771,28 +767,6 @@ /> -
- -
- @ - -
-
-