diff --git a/.github/workflows/telegram-deliver-dev.yml b/.github/workflows/telegram-deliver-dev.yml new file mode 100644 index 00000000..b0a2ff02 --- /dev/null +++ b/.github/workflows/telegram-deliver-dev.yml @@ -0,0 +1,40 @@ +name: Deliver Telegram Notifications (dev) + +permissions: + contents: read + +concurrency: + group: telegram-deliver-dev + cancel-in-progress: false + +on: + schedule: + # GitHub Actions schedules are best-effort; avoid crowded :00/:15/:30/:45 slots. + # Start at :01 so the last slot (:56) rolls to :01 with no hourly gap. + - cron: '1-59/5 * * * *' + workflow_dispatch: + +jobs: + deliver: + runs-on: ubuntu-latest + environment: cron-job + steps: + - name: Drain Telegram outbox (dev) + run: | + response=$(curl -s -w "\n%{http_code}" -X POST \ + -H "Content-Type: application/json" \ + -H "X-Cron-Token: ${{ secrets.CRON_SYNC_TOKEN }}" \ + "${{ secrets.DEV_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/.github/workflows/telegram-deliver.yml b/.github/workflows/telegram-deliver.yml new file mode 100644 index 00000000..cb181773 --- /dev/null +++ b/.github/workflows/telegram-deliver.yml @@ -0,0 +1,39 @@ +name: Deliver Telegram Notifications + +permissions: + contents: read + +concurrency: + group: telegram-deliver-prod + cancel-in-progress: false + +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/CHANGELOG.md b/CHANGELOG.md index 77164813..65f1c3d0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,8 @@ All notable user-facing changes to this project will be documented in this file. ## Unreleased +- Contributors can link a Telegram account from their profile header (private, visible only to them) to receive portal notifications and announcements on Telegram and check their rank, points, and active missions with bot commands; the old public Telegram handle field was removed from profiles. Post-deploy: set TELEGRAM_BOT_USERNAME and TELEGRAM_WEBHOOK_SECRET, run `python manage.py set_telegram_webhook`, and add the telegram-deliver cron workflow (de6a8726) + - Finishing the Creator or Builder journey now actually grants the role: since late June the final "Become a Creator" / "Claim Builder Role" step failed for every new member with a generic error, and completion errors now show their real reason instead of a dead-end "try again" (9d546e70) - Validators can link Telegram support groups to their validator: generate a one-time code on the new Telegram Support page, paste it in a Telegram group with the Deckard support bot, and the group is bound to the validator (multiple groups supported, codes expire in 48 hours and can be revoked) (0cd7e5f) diff --git a/backend/.env.example b/backend/.env.example index 72d83c46..8e96d3a6 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -138,9 +138,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 bc5b2baa..6922945b 100644 --- a/backend/CLAUDE.md +++ b/backend/CLAUDE.md @@ -27,7 +27,11 @@ backend/ ├── api/ # Core API app ├── contributions/ # Contribution tracking ├── leaderboard/ # Leaderboard and rankings +<<<<<<< HEAD +├── social_connections/ # OAuth (GitHub, Twitter, Discord) + Telegram bot link + encrypted token storage +======= ├── social_connections/ # OAuth (GitHub, Twitter, Discord) + encrypted token storage + Telegram (bot-confirmed, no OAuth) +>>>>>>> origin/dev ├── social_tasks/ # Repeatable social tasks (follow, join, like) and completions ├── users/ # User management and auth ├── partners/ # Ecosystem partners directory @@ -47,13 +51,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 @@ -331,9 +335,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`). @@ -347,7 +351,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; refuses to run while a webhook is registered unless `--delete-webhook` is passed, so a prod token can't be hijacked by accident); prod one-time setup: `python manage.py set_telegram_webhook --url https:///api/webhooks/telegram/`. The three `TELEGRAM_*` env vars are wired into both App Runner deploy scripts as SSM secrets (`telegram_bot_token` / `telegram_bot_username` / `telegram_webhook_secret`) — create the SSM parameters BEFORE deploying or App Runner will fail to start. +- **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` @@ -542,10 +548,19 @@ 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) +<<<<<<< HEAD +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) +======= # Campaign vanity links (public; the portal CDN passes /join/* through to the backend) GET /join/{role}/{alias} (anonymous GET/HEAD, 302 with UTMs, throttled 120/min) GET /campaigns/redirect/{role}/{alias} (same view; original internal path) +>>>>>>> origin/dev ``` ### Leaderboard monthly date ranges @@ -605,7 +620,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/deploy-apprunner-dev.sh b/backend/deploy-apprunner-dev.sh index ea7ce13c..105f3fd3 100755 --- a/backend/deploy-apprunner-dev.sh +++ b/backend/deploy-apprunner-dev.sh @@ -151,6 +151,9 @@ if aws apprunner describe-service --service-arn arn:aws:apprunner:$REGION:$ACCOU "DEFILLAMA_FEES_RANK": "$SSM_PREFIX/$SSM_ENV/defillama_fees_rank", "DEFILLAMA_FEES_RANK_URL": "$SSM_PREFIX/$SSM_ENV/defillama_fees_rank_url", "TELEGRAM_MEMBERS": "$SSM_PREFIX/$SSM_ENV/telegram_members", + "TELEGRAM_BOT_TOKEN": "$SSM_PREFIX/$SSM_ENV/telegram_bot_token", + "TELEGRAM_BOT_USERNAME": "$SSM_PREFIX/$SSM_ENV/telegram_bot_username", + "TELEGRAM_WEBHOOK_SECRET": "$SSM_PREFIX/$SSM_ENV/telegram_webhook_secret", "SOCIAL_ENCRYPTION_KEY": "$SSM_PREFIX/$SSM_ENV/social_encryption_key", "TWITTER_CLIENT_ID": "$SSM_PREFIX/$SSM_ENV/twitter_client_id", "TWITTER_CLIENT_SECRET": "$SSM_PREFIX/$SSM_ENV/twitter_client_secret", @@ -291,6 +294,9 @@ EOF "DEFILLAMA_FEES_RANK": "$SSM_PREFIX/$SSM_ENV/defillama_fees_rank", "DEFILLAMA_FEES_RANK_URL": "$SSM_PREFIX/$SSM_ENV/defillama_fees_rank_url", "TELEGRAM_MEMBERS": "$SSM_PREFIX/$SSM_ENV/telegram_members", + "TELEGRAM_BOT_TOKEN": "$SSM_PREFIX/$SSM_ENV/telegram_bot_token", + "TELEGRAM_BOT_USERNAME": "$SSM_PREFIX/$SSM_ENV/telegram_bot_username", + "TELEGRAM_WEBHOOK_SECRET": "$SSM_PREFIX/$SSM_ENV/telegram_webhook_secret", "SOCIAL_ENCRYPTION_KEY": "$SSM_PREFIX/$SSM_ENV/social_encryption_key", "TWITTER_CLIENT_ID": "$SSM_PREFIX/$SSM_ENV/twitter_client_id", "TWITTER_CLIENT_SECRET": "$SSM_PREFIX/$SSM_ENV/twitter_client_secret", diff --git a/backend/deploy-apprunner.sh b/backend/deploy-apprunner.sh index 4a6ba787..9e5b9d03 100755 --- a/backend/deploy-apprunner.sh +++ b/backend/deploy-apprunner.sh @@ -233,6 +233,9 @@ if aws apprunner describe-service --service-arn arn:aws:apprunner:$REGION:$ACCOU "DEFILLAMA_FEES_RANK": "$SSM_PREFIX/prod/defillama_fees_rank", "DEFILLAMA_FEES_RANK_URL": "$SSM_PREFIX/prod/defillama_fees_rank_url", "TELEGRAM_MEMBERS": "$SSM_PREFIX/prod/telegram_members", + "TELEGRAM_BOT_TOKEN": "$SSM_PREFIX/prod/telegram_bot_token", + "TELEGRAM_BOT_USERNAME": "$SSM_PREFIX/prod/telegram_bot_username", + "TELEGRAM_WEBHOOK_SECRET": "$SSM_PREFIX/prod/telegram_webhook_secret", "SOCIAL_ENCRYPTION_KEY": "$SSM_PREFIX/prod/social_encryption_key", "TWITTER_CLIENT_ID": "$SSM_PREFIX/prod/twitter_client_id", "TWITTER_CLIENT_SECRET": "$SSM_PREFIX/prod/twitter_client_secret", @@ -351,6 +354,9 @@ else "DEFILLAMA_FEES_RANK": "$SSM_PREFIX/prod/defillama_fees_rank", "DEFILLAMA_FEES_RANK_URL": "$SSM_PREFIX/prod/defillama_fees_rank_url", "TELEGRAM_MEMBERS": "$SSM_PREFIX/prod/telegram_members", + "TELEGRAM_BOT_TOKEN": "$SSM_PREFIX/prod/telegram_bot_token", + "TELEGRAM_BOT_USERNAME": "$SSM_PREFIX/prod/telegram_bot_username", + "TELEGRAM_WEBHOOK_SECRET": "$SSM_PREFIX/prod/telegram_webhook_secret", "SOCIAL_ENCRYPTION_KEY": "$SSM_PREFIX/prod/social_encryption_key", "TWITTER_CLIENT_ID": "$SSM_PREFIX/prod/twitter_client_id", "TWITTER_CLIENT_SECRET": "$SSM_PREFIX/prod/twitter_client_secret", diff --git a/backend/ethereum_auth/email_verification.py b/backend/ethereum_auth/email_verification.py index 1d97ccec..76c5b909 100644 --- a/backend/ethereum_auth/email_verification.py +++ b/backend/ethereum_auth/email_verification.py @@ -488,7 +488,6 @@ def _create_user_from_pending_signup(self, pending_signup, email): name=profile.get('name', ''), description=profile.get('description', ''), website=profile.get('website', ''), - telegram_handle=profile.get('telegram_handle', ''), linkedin_handle=profile.get('linkedin_handle', ''), is_email_verified=True, email_verified_at=timezone.now(), @@ -515,7 +514,6 @@ def _clean_profile_data(data): 'name', 'description', 'website', - 'telegram_handle', 'linkedin_handle', 'selected_role', } diff --git a/backend/ethereum_auth/views.py b/backend/ethereum_auth/views.py index 32d03a63..d8472317 100644 --- a/backend/ethereum_auth/views.py +++ b/backend/ethereum_auth/views.py @@ -38,11 +38,12 @@ LOGIN_STATEMENT = 'Sign in with Ethereum to GenLayer Testnet Contributions' email_verification_service = EmailVerificationService() turnstile_verifier = TurnstileVerifier() +# telegram_handle removed: Telegram links only through the bot's verified +# private TelegramConnection, never as a user-typed handle. PENDING_SIGNUP_PROFILE_FIELDS = { 'name', 'description', 'website', - 'telegram_handle', 'linkedin_handle', 'selected_role', } 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/admin_mixins.py b/backend/notifications/admin_mixins.py index 68427c13..1573fa60 100644 --- a/backend/notifications/admin_mixins.py +++ b/backend/notifications/admin_mixins.py @@ -227,11 +227,14 @@ def _recall_broadcast(self, request, obj): return if deleted: - self.message_user( - request, - 'Broadcast notification recalled from user feeds.', - level=messages.SUCCESS, - ) + message = 'Broadcast notification recalled from user feeds.' + from .registry import get_event_type + if 'telegram' in get_event_type(self.broadcast_event_slug).channels: + message += ( + ' Queued Telegram deliveries were cancelled; messages ' + 'Telegram already delivered cannot be recalled.' + ) + self.message_user(request, message, level=messages.SUCCESS) else: self.message_user( request, 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..563be731 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,13 +168,24 @@ 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): - """Delete a source object's broadcast notification, if one exists.""" + """Delete a source object's broadcast notification, if one exists. + + Queued (not yet sent) Telegram deliveries are cancelled first; the FK is + SET_NULL, so deleting the notification first would orphan the pending + rows and the drain would still send the recalled content. + """ if source is None or not source.pk: return 0 @@ -178,27 +195,36 @@ def recall_broadcast(event_slug, source): event_type=event.slug, dedupe_key=broadcast_dedupe_key(event.slug, source), ) + from . import telegram + telegram.cancel_pending_for_notifications(queryset) count = queryset.count() queryset.delete() 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..9427f82b --- /dev/null +++ b/backend/notifications/telegram.py @@ -0,0 +1,483 @@ +"""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 re +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 +# The drain runs inside an HTTP request and App Runner kills requests at +# 120s. The happy path (~16s for 400 rows) never gets near this; the budget +# exists for runs where slow Telegram responses (10s timeout, degraded +# retries) pile up. Unfinished rows return to pending for the next tick. +MAX_RUN_SECONDS = 90 + + +# --------------------------------------------------------------------------- +# Rendering +# --------------------------------------------------------------------------- + +# Per-event accents; the category map is the fallback for future events. +EVENT_EMOJI = { + 'submission.accepted': '✅', + 'submission.rejected': '❌', + 'submission.more_info_needed': '📝', + 'submission.proposal_questioned': '❓', + 'submission.appealed': '⚖️', + 'submission.more_info_resubmitted': '📨', + 'contribution.highlighted': '🌟', + 'validator.graduated': '🎓', + 'mission.published': '🎯', + 'node_version.published': '🚀', + 'alert.published': '🚨', + 'custom.announcement': '📣', +} +CATEGORY_EMOJI = { + 'submission': '📄', + 'contribution': '🏆', + 'community': '👥', + 'content': '📰', + 'validator': '🖥️', + 'system': '⚙️', + 'announcement': '📣', +} +# Long bodies collapse behind Telegram's expandable quote so the card stays +# compact in the chat list. +EXPANDABLE_BODY_THRESHOLD = 400 + + +def _emoji_for(notification): + return ( + EVENT_EMOJI.get(notification.event_type) + or CATEGORY_EMOJI.get(notification.category) + or '🔔' + ) + + +def _absolute_link_url(notification): + url = (notification.link_url or '').strip() + if not url: + return '' + if url.startswith('#/'): + url = url[1:] + if not url.startswith(('http://', 'https://')): + url = f"{settings.FRONTEND_URL}{url}" + return url + + +def notification_link_button(notification): + """Inline-keyboard reply_markup for the notification's link, or None. + + The link travels as a button (not an in the text): that is what makes + the message read as a card in Telegram. + """ + url = _absolute_link_url(notification) + if not url: + return None + label = notification.link_label or 'Open in portal' + return {'inline_keyboard': [[{'text': label[:64], 'url': url}]]} + + +def render_notification_text(notification): + """Render a Notification row as a Telegram HTML "card": + + {emoji} {title} + {category} · GenLayer Portal + +
{body}
+ + The blockquote draws Telegram's accent bar next to the body; the link is + delivered separately as an inline button (notification_link_button). + """ + category_label = notification.get_category_display() if notification.category else '' + byline_parts = [part for part in (category_label, 'GenLayer Portal') if part] + header = ( + f"{_emoji_for(notification)} {escape(notification.title)}\n" + f"{escape(' · '.join(byline_parts))}" + ) + + body = (notification.body or '').strip() + if not body: + return header + + # Escape FIRST, then budget on the escaped length: escaping expands + # & < > up to 5x, so budgeting raw text would overshoot the limit and + # force send-time truncation straight through the HTML tags. + escaped_body = escape(body) + budget = TELEGRAM_MAX_LEN - len(header) - len('\n\n
') + if budget <= 20: + return header + if len(escaped_body) > budget: + cut = escaped_body[: budget - 1] + # Never end inside an entity (& / < / >). + cut = re.sub(r'&[a-zA-Z]{0,4}$', '', cut) + escaped_body = cut + '…' + open_tag = ( + '
' + if len(escaped_body) > EXPANDABLE_BODY_THRESHOLD + else '
' + ) + return f"{header}\n\n{open_tag}{escaped_body}
" + + +# --------------------------------------------------------------------------- +# Enqueue +# --------------------------------------------------------------------------- + +def eligible_connections(users): + return TelegramConnection.objects.filter( + user__in=users, + user__is_active=True, + notifications_enabled=True, + blocked_at__isnull=True, + ) + + +def _refresh_pending_text(notifications, text): + """Sync still-queued rows with refreshed notification copy. + + Dedupe-refreshes and campaign resends update the Notification rows in + place; without this, a pre-cron edit would deliver the original text. + Already-sent rows are history and stay untouched. + """ + TelegramMessage.objects.filter( + notification__in=notifications, + direction=TelegramMessage.DIRECTION_OUT, + status=TelegramMessage.STATUS_PENDING, + ).exclude(text=text).update(text=text, updated_at=timezone.now()) + + +def enqueue_personal(notification): + """Queue a personal notification for its recipient, if linked.""" + try: + conn = ( + TelegramConnection.objects + .filter( + user=notification.recipient, + user__is_active=True, + notifications_enabled=True, + blocked_at__isnull=True, + ) + .first() + ) + if conn is None: + return + text = render_notification_text(notification) + _, created = TelegramMessage.objects.get_or_create( + notification=notification, + connection=conn, + direction=TelegramMessage.DIRECTION_OUT, + defaults={ + 'chat_id': conn.platform_user_id, + 'text': text, + 'status': TelegramMessage.STATUS_PENDING, + }, + ) + if not created: + _refresh_pending_text([notification.pk], text) + 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) + # A re-broadcast refreshed the notification copy; sync queued rows. + _refresh_pending_text([notification.pk], text) + 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(), + ) + # All fanned-out rows carry identical copy; render once. + text = None + batch = [] + for notification in notifications.iterator(chunk_size=500): + if text is None: + text = render_notification_text(notification) + conn = connections[notification.recipient_id] + 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) + # A resend refreshed the campaign copy; sync still-queued rows. + if text is not None: + _refresh_pending_text(notifications, text) + except Exception: + logger.exception("Telegram enqueue_campaign failed (campaign %s)", campaign.pk) + + +def cancel_pending_for_notifications(notifications): + """Delete not-yet-sent outbox rows for the given notifications. + + Must run BEFORE the notifications themselves are deleted: the FK is + SET_NULL, so deleting first would orphan the pending rows and the drain + would still send the recalled content. Sent messages cannot be recalled + from Telegram. + """ + return TelegramMessage.objects.filter( + notification__in=notifications, + status__in=[TelegramMessage.STATUS_PENDING, TelegramMessage.STATUS_SENDING], + ).delete()[0] + + +def cancel_pending_for_campaign(campaign): + from .models import Notification + + return cancel_pending_for_notifications( + Notification.objects.filter(dedupe_key=campaign.dedupe_key) + ) + + +# --------------------------------------------------------------------------- +# 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 [] + # No connection preload: the send loop refetches each connection right + # before delivery so relinks/deactivations during the batch are seen. + return list( + TelegramMessage.objects + .filter(pk__in=pks) + .select_related('notification') + .order_by('created_at') + ) + + +def _finish(message, status, error=''): + # Queryset update, not instance save: a recall can delete the row while + # its send is in flight, and saving a deleted instance would raise and + # abort the rest of the run. An update on a deleted row is a no-op. + updates = {'status': status, 'error': error[:200], 'updated_at': timezone.now()} + if status == TelegramMessage.STATUS_SENT: + updates['sent_at'] = timezone.now() + TelegramMessage.objects.filter(pk=message.pk).update(**updates) + + +def _unclaim(messages): + """Return claimed-but-unprocessed rows to pending, undoing the attempt.""" + TelegramMessage.objects.filter(pk__in=[m.pk for m in messages]).update( + status=TelegramMessage.STATUS_PENDING, + attempts=F('attempts') - 1, + ) + + +def deliver_pending(limit=DEFAULT_RUN_LIMIT, max_run_seconds=MAX_RUN_SECONDS): + """Send queued messages, pacing to Telegram's rate budget. + + Returns {'sent': n, 'failed': n, 'remaining': n}. On a 429 or when the + wall-clock budget runs out, the run stops and unprocessed rows return to + pending; the next cron tick resumes. Delivery is at-least-once, bounded + by MAX_ATTEMPTS. + """ + # A worker that crashed mid-send on a row's FINAL attempt leaves it + # 'sending' with attempts == MAX_ATTEMPTS; claims require attempts < MAX, + # so without this sweep the row would be stranded forever. + stale_cutoff = timezone.now() - timedelta(minutes=STALE_SENDING_MINUTES) + TelegramMessage.objects.filter( + direction=TelegramMessage.DIRECTION_OUT, + status=TelegramMessage.STATUS_SENDING, + attempts__gte=MAX_ATTEMPTS, + updated_at__lt=stale_cutoff, + ).update( + status=TelegramMessage.STATUS_FAILED, + error='max_attempts_exceeded', + updated_at=timezone.now(), + ) + + sent = failed = processed = 0 + stopped = False + seen_pks = set() + deadline = time.monotonic() + max_run_seconds + + while processed < limit and not stopped and time.monotonic() < deadline: + 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): + if time.monotonic() >= deadline: + _unclaim(batch[index:]) + stopped = True + logger.warning("Telegram delivery run hit its %ss budget; run stopped", max_run_seconds) + break + + # Re-check the LIVE row, not the batch-time snapshot: a recall + # during the batch deletes claimed rows (row gone -> skip) or + # nulls the notification FK (enqueued rows always carry one), and + # neither must reach Telegram. + live = ( + TelegramMessage.objects + .filter(pk=message.pk) + .values_list('status', 'notification_id') + .first() + ) + if live is None or live[0] != TelegramMessage.STATUS_SENDING: + continue + if live[1] is None: + _finish(message, TelegramMessage.STATUS_FAILED, 'notification_recalled') + failed += 1 + continue + + # Refetch immediately before delivery (not preloaded with the + # batch): a relink mid-batch must send to the LIVE chat id, a 403 + # must mark the CURRENT row state, and a user deactivated after + # enqueue must not receive anything. + conn = ( + TelegramConnection.objects + .select_related('user') + .filter(pk=message.connection_id) + .first() + ) if message.connection_id else None + if ( + conn is None + or not conn.notifications_enabled + or conn.blocked_at + or not conn.user.is_active + ): + _finish(message, TelegramMessage.STATUS_FAILED, 'connection_gone') + failed += 1 + continue + + # message.chat_id is only the historical record; the button is + # built from the notification at send time, so the outbox stores + # only the rendered text. + ok, retry_after, description = send_telegram_message( + conn.platform_user_id, + message.text, + connection=conn, + reply_markup=notification_link_button(message.notification), + ) + 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. + _unclaim(batch[index:]) + stopped = 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 stopped: + 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..118a1ffd 100644 --- a/backend/notifications/tests.py +++ b/backend/notifications/tests.py @@ -1308,3 +1308,614 @@ 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): + notification = 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 0e57aa0e..85fc12c3 100644 --- a/frontend/src/lib/api.js +++ b/frontend/src/lib/api.js @@ -329,6 +329,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 @@ /> -
- -
- @ - -
-
-