diff --git a/telegram-bot/.env.example b/telegram-bot/.env.example
new file mode 100644
index 000000000..1615a96c4
--- /dev/null
+++ b/telegram-bot/.env.example
@@ -0,0 +1,21 @@
+# Telegram Bot
+TELEGRAM_BOT_TOKEN=your_telegram_bot_token_here
+TELEGRAM_CHANNEL_ID=-1001234567890
+
+# GitHub
+GITHUB_TOKEN=your_github_token_here
+GITHUB_REPO=SolFoundry/solfoundry
+GITHUB_WEBHOOK_SECRET=your_webhook_secret_here
+
+# SolFoundry API (optional - for leaderboard)
+SOLFOUNDRY_API_URL=https://api.solfoundry.org
+
+# Redis
+REDIS_URL=redis://localhost:6379/0
+
+# Bot mode: polling (dev) or webhook (prod)
+BOT_MODE=polling
+WEBHOOK_HOST=https://your-domain.com
+WEBHOOK_PATH=/telegram/webhook
+
+LOG_LEVEL=INFO
diff --git a/telegram-bot/bot/__init__.py b/telegram-bot/bot/__init__.py
new file mode 100644
index 000000000..9d00861eb
--- /dev/null
+++ b/telegram-bot/bot/__init__.py
@@ -0,0 +1 @@
+# SolFoundry Telegram Bot
diff --git a/telegram-bot/bot/app.py b/telegram-bot/bot/app.py
new file mode 100644
index 000000000..b9646d9ef
--- /dev/null
+++ b/telegram-bot/bot/app.py
@@ -0,0 +1,184 @@
+"""
+SolFoundry Telegram Bot — polling + FastAPI webhook server.
+
+Usage:
+ python -m bot.app # polling mode
+ uvicorn bot.app:app --port 8080 # webhook mode
+"""
+import asyncio
+import logging
+import sys
+from datetime import datetime, timezone
+from typing import Optional
+
+from fastapi import FastAPI, Request, Response
+from telegram import Update
+from telegram.ext import (
+ Application, CommandHandler, CallbackQueryHandler,
+ MessageHandler, filters,
+)
+
+from bot.config import config
+from bot.formatters import format_bounty_notification
+from bot.github_client import GitHubClient
+from bot.handlers import (
+ cmd_start, cmd_help, cmd_leaderboard, cmd_filters,
+ cmd_filter, cmd_subscribe, cmd_unsubscribe, cmd_bounty,
+ handle_callback,
+)
+from bot.subscription_store import SubscriptionStore
+
+logging.basicConfig(
+ level=getattr(logging, config.log_level, logging.INFO),
+ format="%(asctime)s %(levelname)s %(name)s — %(message)s",
+ handlers=[logging.StreamHandler(sys.stdout)],
+)
+logger = logging.getLogger(__name__)
+
+# ─────────────────────────────────────────────────────────────────────────────
+# FastAPI webhook app
+# ─────────────────────────────────────────────────────────────────────────────
+
+app_fastapi = FastAPI(title="SolFoundry Telegram Bot Webhook")
+telegram_app: Optional[Application] = None
+
+
+@app_fastapi.post(config.webhook_path)
+async def telegram_webhook(request: Request) -> Response:
+ global telegram_app
+ if telegram_app is None:
+ return Response(status_code=503, content="Bot not initialized")
+ try:
+ data = await request.json()
+ update = Update.de_json(data, telegram_app.bot)
+ await telegram_app.process_update(update)
+ except Exception:
+ logger.exception("Webhook update failed")
+ return Response(status_code=200)
+
+
+@app_fastapi.get("/health")
+async def health() -> dict:
+ return {"status": "ok", "mode": config.bot_mode}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Bounty polling loop
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def poll_bounties(application: Application) -> None:
+ """Background task: poll GitHub for new bounties and post to channel."""
+ store = SubscriptionStore()
+ github = GitHubClient()
+ channel_id = config.telegram_channel_id
+ logger.info("Bounty polling loop started (interval: 60s)")
+
+ while True:
+ try:
+ bounties = github.fetch_open_bounties()
+ logger.debug("Fetched %d open bounties", len(bounties))
+
+ for bounty in bounties:
+ if not bounty.is_open:
+ continue
+ if store.is_notified(bounty.number):
+ continue
+
+ text, keyboard = format_bounty_notification(bounty)
+ try:
+ await application.bot.send_message(
+ chat_id=channel_id,
+ text=text,
+ parse_mode="HTML",
+ reply_markup=keyboard,
+ disable_web_page_preview=True,
+ )
+ store.mark_notified(bounty.number)
+ logger.info("Notified channel about bounty #%d: %s", bounty.number, bounty.title)
+ except Exception as e:
+ logger.error("Failed to send bounty #%d: %s", bounty.number, e)
+
+ store.set_last_check(datetime.now(timezone.utc).isoformat())
+
+ except Exception:
+ logger.exception("Polling loop error")
+
+ await asyncio.sleep(60)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Bot setup
+# ─────────────────────────────────────────────────────────────────────────────
+
+def build_app() -> Application:
+ application = (
+ Application.builder()
+ .token(config.telegram_bot_token)
+ .read_timeout(30)
+ .write_timeout(30)
+ .build()
+ )
+ application.add_handler(CommandHandler("start", cmd_start))
+ application.add_handler(CommandHandler("help", cmd_help))
+ application.add_handler(CommandHandler("leaderboard", cmd_leaderboard))
+ application.add_handler(CommandHandler("filters", cmd_filters))
+ application.add_handler(CommandHandler("filter", cmd_filter))
+ application.add_handler(CommandHandler("subscribe", cmd_subscribe))
+ application.add_handler(CommandHandler("unsubscribe", cmd_unsubscribe))
+ application.add_handler(CommandHandler("bounty", cmd_bounty))
+ application.add_handler(CallbackQueryHandler(handle_callback))
+ return application
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Entrypoints
+# ─────────────────────────────────────────────────────────────────────────────
+
+async def run_polling() -> None:
+ """Polling mode — development."""
+ global telegram_app
+ telegram_app = build_app()
+ await telegram_app.initialize()
+ await telegram_app.start()
+ asyncio.create_task(poll_bounties(telegram_app))
+ await telegram_app.updater.start_polling(allowed_updates=Update.ALL_TYPES)
+ await telegram_app.updater.stop()
+ await telegram_app.stop()
+ await telegram_app.shutdown()
+
+
+async def run_webhook() -> None:
+ """Webhook mode — production."""
+ global telegram_app
+ telegram_app = build_app()
+ await telegram_app.initialize()
+ await telegram_app.start()
+ webhook_url = config.full_webhook_url
+ await telegram_app.bot.set_webhook(webhook_url)
+ logger.info("Webhook set to: %s", webhook_url)
+ try:
+ while True:
+ await asyncio.sleep(3600)
+ except asyncio.CancelledError:
+ pass
+ finally:
+ await telegram_app.stop()
+ await telegram_app.shutdown()
+
+
+if __name__ == "__main__":
+ errors = config.validate()
+ if errors:
+ for e in errors:
+ print(f"CONFIG ERROR: {e}", file=sys.stderr)
+ sys.exit(1)
+
+ mode = config.bot_mode.lower()
+ if mode == "polling":
+ asyncio.run(run_polling())
+ elif mode == "webhook":
+ import uvicorn
+ uvicorn.run("bot.app:app", host="0.0.0.0", port=8080, reload=False)
+ else:
+ print(f"Unknown BOT_MODE: {mode}", file=sys.stderr)
+ sys.exit(1)
diff --git a/telegram-bot/bot/config.py b/telegram-bot/bot/config.py
new file mode 100644
index 000000000..3209fa1d7
--- /dev/null
+++ b/telegram-bot/bot/config.py
@@ -0,0 +1,39 @@
+"""Bot configuration loaded from environment variables."""
+from dataclasses import dataclass
+from os import getenv
+from typing import Optional
+
+
+@dataclass
+class BotConfig:
+ telegram_bot_token: str = getenv("TELEGRAM_BOT_TOKEN", "")
+ telegram_channel_id: str = getenv("TELEGRAM_CHANNEL_ID", "")
+
+ github_token: str = getenv("GITHUB_TOKEN", "")
+ github_repo: str = getenv("GITHUB_REPO", "SolFoundry/solfoundry")
+ github_webhook_secret: str = getenv("GITHUB_WEBHOOK_SECRET", "")
+
+ solfoundry_api_url: Optional[str] = getenv("SOLFOUNDRY_API_URL")
+
+ redis_url: str = getenv("REDIS_URL", "redis://localhost:6379/0")
+
+ bot_mode: str = getenv("BOT_MODE", "polling")
+ webhook_host: str = getenv("WEBHOOK_HOST", "")
+ webhook_path: str = getenv("WEBHOOK_PATH", "/telegram/webhook")
+
+ log_level: str = getenv("LOG_LEVEL", "INFO")
+
+ @property
+ def full_webhook_url(self) -> str:
+ return f"{self.webhook_host.rstrip('/')}{self.webhook_path}"
+
+ def validate(self) -> list[str]:
+ errors = []
+ if not self.telegram_bot_token:
+ errors.append("TELEGRAM_BOT_TOKEN is required")
+ if not self.telegram_channel_id:
+ errors.append("TELEGRAM_CHANNEL_ID is required")
+ return errors
+
+
+config = BotConfig()
diff --git a/telegram-bot/bot/formatters.py b/telegram-bot/bot/formatters.py
new file mode 100644
index 000000000..d81a84016
--- /dev/null
+++ b/telegram-bot/bot/formatters.py
@@ -0,0 +1,135 @@
+"""Rich embed formatters for Telegram messages."""
+import textwrap
+from typing import Optional, Tuple
+
+from telegram import InlineKeyboardButton, InlineKeyboardMarkup
+
+from bot.models import Bounty, LeaderboardEntry
+
+TIER_EMOJI = {"1": "🔧", "2": "⚡", "3": "🚀", None: "📋"}
+TIER_LABEL = {
+ "1": "Tier 1 — Bug Fixes & Docs",
+ "2": "Tier 2 — Module & Integration",
+ "3": "Tier 3 — Major Feature",
+ None: "Bounty",
+}
+
+
+def format_bounty_notification(bounty: Bounty) -> Tuple[str, InlineKeyboardMarkup]:
+ tier = bounty.tier
+ tier_emoji = TIER_EMOJI.get(tier, "📋")
+ tier_text = TIER_LABEL.get(tier, "Bounty")
+ body_preview = (bounty.body[:300].strip() + "...") if len(bounty.body) > 300 else bounty.body[:300].strip()
+ reward_text = f"💰 {bounty.reward}" if bounty.reward else ""
+
+ meta_parts = []
+ if reward_text:
+ meta_parts.append(reward_text)
+ if bounty.bounty_type:
+ meta_parts.append(f"🏷 {bounty.bounty_type.capitalize()}")
+ meta_parts.append(f"🕐 {bounty.created_at.strftime('%Y-%m-%d')}")
+
+ lines = [
+ f"{tier_emoji} New {tier_text}",
+ f"#{bounty.number} {bounty.title}",
+ "",
+ ]
+ if body_preview:
+ lines.append(textwrap.fill(body_preview, width=80))
+ lines.append("")
+ lines.append(" | ".join(meta_parts))
+
+ keyboard = [
+ [
+ InlineKeyboardButton("🔍 Details", url=bounty.html_url),
+ ],
+ [
+ InlineKeyboardButton("✅ I want this", callback_data=f"claim:{bounty.number}"),
+ ],
+ ]
+ return "\n".join(lines), InlineKeyboardMarkup(keyboard)
+
+
+def format_bounty_detail(bounty: Bounty) -> Tuple[str, InlineKeyboardMarkup]:
+ tier = bounty.tier
+ tier_emoji = TIER_EMOJI.get(tier, "📋")
+ tier_text = TIER_LABEL.get(tier, "Bounty")
+ reward_text = f"💰 {bounty.reward}" if bounty.reward else "💰 Reward TBD"
+
+ lines = [
+ f"{tier_emoji} {tier_text}",
+ f"#{bounty.number} — {bounty.title}",
+ "",
+ reward_text,
+ "",
+ ]
+ if bounty.body:
+ wrapped = textwrap.fill(bounty.body[:1000], width=80)
+ lines.append(wrapped)
+ lines.append("")
+
+ meta = [
+ f"📌 State: {bounty.state.upper()}",
+ f"🕐 Posted: {bounty.created_at.strftime('%Y-%m-%d %H:%M UTC')}",
+ f"🔗 {bounty.html_url}",
+ ]
+ if bounty.assignee:
+ meta.append(f"👤 Assignee: @{bounty.assignee}")
+ lines.extend(meta)
+
+ keyboard = [
+ [InlineKeyboardButton("🐙 Open on GitHub", url=bounty.html_url)],
+ [InlineKeyboardButton("✅ I want this bounty", callback_data=f"claim:{bounty.number}")],
+ ]
+ return "\n".join(lines), InlineKeyboardMarkup(keyboard)
+
+
+def format_leaderboard(entries: list[LeaderboardEntry]) -> str:
+ if not entries:
+ return "📊 No leaderboard data available yet."
+
+ header = "🏆 Top Contributors\n\n```\n{'Rank':<6}{'Contributor':<25}{'Merged':<10}{'Est. Reward':<12}"
+ lines = [header, "─" * 55]
+
+ medal = {1: "🥇", 2: "🥈", 3: "🥉"}
+ for entry in entries:
+ medal_str = medal.get(entry.rank, f"#{entry.rank}")
+ row = f"{medal_str:<6}{'@' + entry.username:<25}{entry.merged_count:<10}{entry.total_reward} FNDRY"
+ lines.append(row)
+ lines.append("```")
+ return "\n".join(lines)
+
+
+def format_filter_status(user_filter: "UserFilter") -> str:
+ tier_str = ", ".join(user_filter.tiers) if user_filter.tiers else "All"
+ type_str = ", ".join(user_filter.types) if user_filter.types else "All types"
+ reward_str = f"≥ {user_filter.min_reward} FNDRY" if user_filter.min_reward else "No minimum"
+
+ lines = [
+ "🔔 Your Notification Filters",
+ "",
+ f"Tiers: {tier_str}",
+ f"Types: {type_str}",
+ f"Min reward: {reward_str}",
+ "",
+ "Use /filter to update your settings.",
+ ]
+ return "\n".join(lines)
+
+
+FILTER_HELP = """\
+🔔 Set Bounty Notification Filters
+
+/filter tier:1,2 type:feature,bug min:500 keyword:api
+
+Options:
+• tier:N,N — Filter by tier (1, 2, 3)
+• type:t1,t2 — Type: feature, bug, docs, integration, security
+• min:N — Minimum FNDRY reward
+• keyword:word — Title must contain keyword
+
+Examples:
+/filter tier:2,3 min:500 → T2/T3 ≥500 FNDRY
+/filter type:bug → Bug bounties only
+/filter clear → Reset to defaults
+"""
diff --git a/telegram-bot/bot/github_client.py b/telegram-bot/bot/github_client.py
new file mode 100644
index 000000000..df0805bb7
--- /dev/null
+++ b/telegram-bot/bot/github_client.py
@@ -0,0 +1,183 @@
+"""GitHub API client for fetching bounty issues and leaderboard data."""
+import logging
+from datetime import datetime
+from typing import Optional
+
+import httpx
+
+from bot.config import config
+from bot.models import Bounty, LeaderboardEntry
+
+logger = logging.getLogger(__name__)
+GITHUB_API = "https://api.github.com"
+
+
+class GitHubClient:
+ def __init__(self, token: Optional[str] = None):
+ self.token = token or config.github_token
+ self.repo = config.github_repo
+ self.headers = {
+ "Accept": "application/vnd.github+json",
+ "User-Agent": "SolFoundry-Telegram-Bot/1.0",
+ "X-GitHub-Api-Version": "2022-11-28",
+ }
+ if self.token:
+ self.headers["Authorization"] = f"Bearer {self.token}"
+
+ def _request(self, method: str, url: str, **kwargs) -> httpx.Response:
+ with httpx.Client(timeout=30.0) as client:
+ response = client.request(method, url, headers=self.headers, **kwargs)
+ response.raise_for_status()
+ return response
+
+ def fetch_open_bounties(
+ self, state: str = "open", per_page: int = 100,
+ ) -> list[Bounty]:
+ """Fetch all open issues tagged as bounties."""
+ all_bounties = []
+ page = 1
+ bounty_labels = ["bounty-tier-1", "bounty-tier-2", "bounty-tier-3"]
+
+ while True:
+ params = {
+ "state": state,
+ "labels": ",".join(bounty_labels),
+ "per_page": per_page,
+ "page": page,
+ "sort": "created",
+ "direction": "desc",
+ }
+ url = f"{GITHUB_API}/repos/{self.repo}/issues"
+ resp = self._request("GET", url, params=params)
+ items = resp.json()
+ if not items:
+ break
+ for item in items:
+ if "pull_request" in item:
+ continue
+ bounty = self._parse_bounty(item)
+ if bounty:
+ all_bounties.append(bounty)
+ if len(items) < per_page:
+ break
+ page += 1
+ return all_bounties
+
+ def fetch_bounty(self, issue_number: int) -> Optional[Bounty]:
+ url = f"{GITHUB_API}/repos/{self.repo}/issues/{issue_number}"
+ try:
+ resp = self._request("GET", url)
+ item = resp.json()
+ if "pull_request" in item:
+ return None
+ return self._parse_bounty(item)
+ except httpx.HTTPStatusError as e:
+ if e.response.status_code == 404:
+ return None
+ raise
+
+ def _parse_bounty(self, item: dict) -> Optional[Bounty]:
+ try:
+ labels = [l["name"] for l in item.get("labels", [])]
+ if not any(l.startswith("bounty-tier-") for l in labels):
+ return None
+ created_at = item["created_at"]
+ if isinstance(created_at, str):
+ created_at = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
+ updated_at = item["updated_at"]
+ if isinstance(updated_at, str):
+ updated_at = datetime.fromisoformat(updated_at.replace("Z", "+00:00"))
+ return Bounty(
+ number=item["number"],
+ title=item["title"],
+ body=item.get("body") or "",
+ state=item["state"],
+ labels=labels,
+ assignee=item.get("assignee", {}).get("login") if item.get("assignee") else None,
+ created_at=created_at,
+ updated_at=updated_at,
+ html_url=item["html_url"],
+ )
+ except Exception as e:
+ logger.warning("Failed to parse bounty #%s: %s", item.get("number"), e)
+ return None
+
+ def fetch_contributor_stats(self) -> list[LeaderboardEntry]:
+ """Fetch leaderboard: tries SolFoundry API first, then GitHub GraphQL."""
+ if config.solfoundry_api_url:
+ try:
+ return self._fetch_leaderboard_from_api()
+ except Exception as e:
+ logger.warning("API leaderboard failed: %s, falling back to GraphQL", e)
+ return self._fetch_leaderboard_from_graphql()
+
+ def _fetch_leaderboard_from_api(self) -> list[LeaderboardEntry]:
+ url = f"{config.solfoundry_api_url}/leaderboard"
+ resp = self._request("GET", url)
+ data = resp.json()
+ entries = []
+ for i, item in enumerate(data[:20], 1):
+ entries.append(LeaderboardEntry(
+ rank=i,
+ username=item.get("username", "unknown"),
+ merged_count=item.get("merged_count", 0),
+ total_reward=item.get("total_reward", 0),
+ avatar_url=item.get("avatar_url"),
+ ))
+ return entries
+
+ LEADERBOARD_QUERY = """
+ query($owner: String!, $name: String!, $cursor: String) {
+ repository(owner: $owner, name: $name) {
+ defaultBranchRef {
+ target {
+ ... on Commit {
+ history(first: 100, after: $cursor) {
+ pageInfo { hasNextPage endCursor }
+ nodes {
+ author { user { login avatarUrl } }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ """
+
+ def _fetch_leaderboard_from_graphql(self) -> list[LeaderboardEntry]:
+ owner, name = self.repo.split("/")
+ url = f"{GITHUB_API}/graphql"
+ contributors: dict[str, int] = {}
+ cursor = None
+
+ for _ in range(5):
+ variables = {"owner": owner, "name": name, "cursor": cursor}
+ payload = {"query": self.LEADERBOARD_QUERY, "variables": variables}
+ resp = self._request("POST", url, json=payload)
+ data = resp.json()
+ history = data.get("data", {}).get("repository", {}).get("defaultBranchRef", {})
+ history = history.get("target", {}).get("history", {})
+ nodes = history.get("nodes", [])
+ page_info = history.get("pageInfo", {})
+ for commit in nodes:
+ author = commit.get("author", {})
+ user = author.get("user")
+ if user:
+ login = user.get("login")
+ if login:
+ contributors[login] = contributors.get(login, 0) + 1
+ if not page_info.get("hasNextPage"):
+ break
+ cursor = page_info.get("endCursor")
+
+ sorted_contributors = sorted(contributors.items(), key=lambda x: -x[1])
+ return [
+ LeaderboardEntry(
+ rank=i + 1,
+ username=login,
+ merged_count=count,
+ total_reward=count * 100,
+ )
+ for i, (login, count) in enumerate(sorted_contributors[:20])
+ ]
diff --git a/telegram-bot/bot/handlers.py b/telegram-bot/bot/handlers.py
new file mode 100644
index 000000000..d77ef7ff8
--- /dev/null
+++ b/telegram-bot/bot/handlers.py
@@ -0,0 +1,198 @@
+"""Telegram bot command and callback handlers."""
+import logging
+import re
+from typing import Optional
+
+from telegram import InlineKeyboardMarkup, Update
+from telegram.ext import ContextTypes
+
+from bot.formatters import FILTER_HELP, format_bounty_detail, format_filter_status, format_leaderboard
+from bot.github_client import GitHubClient
+from bot.models import UserFilter
+from bot.subscription_store import SubscriptionStore
+
+logger = logging.getLogger(__name__)
+
+
+def parse_filter_args(text: str) -> tuple[Optional[UserFilter], Optional[str]]:
+ """Parse /filter args into UserFilter. Returns (filter, error)."""
+ text = text.strip()
+ if not text or text == "clear":
+ return None, None
+
+ tiers: Optional[list[str]] = None
+ types: Optional[list[str]] = None
+ min_reward: Optional[int] = None
+
+ for token in text.split():
+ token = token.strip()
+ if not token or ':' not in token:
+ continue
+ key, _, value = token.partition(':')
+ key, value = key.lower(), value.lower()
+ if key == "tier":
+ tiers = [v.strip() for v in value.split(',') if v.strip()]
+ elif key == "type":
+ types = [v.strip() for v in value.split(',') if v.strip()]
+ elif key == "min":
+ try:
+ min_reward = int(value)
+ except ValueError:
+ return None, f"Invalid min value: {value}"
+
+ return UserFilter(user_id=0, tiers=tiers, types=types, min_reward=min_reward), None
+
+
+async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ await update.message.reply_text(
+ "👋 Welcome to SolFoundry Bounty Bot!\n\n"
+ "I'll notify you about new SolFoundry bounties.\n\n"
+ "📋 Commands:\n"
+ "/bounty N — Bounty details\n"
+ "/leaderboard — 🏆 Top contributors\n"
+ "/filters — View filters\n"
+ "/filter — Set filters\n"
+ "/subscribe — Subscribe\n"
+ "/help — All commands",
+ parse_mode="HTML", disable_web_page_preview=True,
+ )
+
+
+async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ await update.message.reply_text(
+ "📖 Commands\n\n"
+ "/start — Welcome\n"
+ "/bounty N — Bounty #N detail\n"
+ "/leaderboard — 🏆 Top contributors\n"
+ "/filters — View your filters\n"
+ "/filter tier:1,2 min:500 — Set filters\n"
+ "/filter clear — Reset filters\n"
+ "/subscribe — Subscribe\n"
+ "/unsubscribe — Unsubscribe\n"
+ "/help — This message\n\n"
+ "Acceptance Criteria (Bounty #847):\n"
+ "✅ Real-time bounty posting with rich embeds\n"
+ "✅ /leaderboard command\n"
+ "✅ Customizable notification filters per user",
+ parse_mode="HTML", disable_web_page_preview=True,
+ )
+
+
+async def cmd_leaderboard(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ msg = await update.message.reply_text("⏳ Fetching leaderboard...")
+ try:
+ entries = GitHubClient().fetch_contributor_stats()
+ text = format_leaderboard(entries)
+ await msg.edit_text(text, parse_mode="HTML")
+ except Exception as e:
+ logger.exception("Leaderboard failed")
+ await msg.edit_text(f"❌ Failed: {e}")
+
+
+async def cmd_filters(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ store = SubscriptionStore()
+ user_filter = store.get_filter(update.effective_user.id)
+ await update.message.reply_text(format_filter_status(user_filter), parse_mode="HTML")
+
+
+async def cmd_filter(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ store = SubscriptionStore()
+ user_id = update.effective_user.id
+ raw_args = " ".join(context.args) if context.args else ""
+
+ if raw_args == "clear":
+ store.delete_filter(user_id)
+ await update.message.reply_text("✅ Filters reset to defaults.")
+ return
+
+ if not raw_args:
+ user_filter = store.get_filter(user_id)
+ await update.message.reply_text(format_filter_status(user_filter), parse_mode="HTML")
+ return
+
+ user_filter, error = parse_filter_args(raw_args)
+ if error:
+ await update.message.reply_text(f"❌ {error}\n\n{FILTER_HELP}", parse_mode="HTML")
+ return
+
+ valid_tiers = {"1", "2", "3"}
+ valid_types = {"feature", "bug", "docs", "integration", "security"}
+
+ if user_filter.tiers and set(user_filter.tiers) - valid_tiers:
+ await update.message.reply_text("❌ Invalid tier. Valid: 1, 2, 3.", parse_mode="HTML")
+ return
+ if user_filter.types and set(user_filter.types) - valid_types:
+ await update.message.reply_text(
+ "❌ Invalid type. Valid: feature, bug, docs, integration, security.", parse_mode="HTML",
+ )
+ return
+
+ user_filter.user_id = user_id
+ store.save_filter(user_filter)
+ await update.message.reply_text(
+ f"✅ Filters updated!\n\n{format_filter_status(user_filter)}", parse_mode="HTML",
+ )
+
+
+async def cmd_subscribe(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ store = SubscriptionStore()
+ store.save_filter(UserFilter(user_id=update.effective_user.id))
+ await update.message.reply_text(
+ "✅ Subscribed to all bounties!\nUse /filter to narrow down notifications.",
+ )
+
+
+async def cmd_unsubscribe(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ store = SubscriptionStore()
+ store.delete_filter(update.effective_user.id)
+ await update.message.reply_text("❌ Unsubscribed. Use /subscribe to re-enable.")
+
+
+async def cmd_bounty(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ if not context.args:
+ await update.message.reply_text("Usage: /bounty \nExample: /bounty 847")
+ return
+ try:
+ number = int(context.args[0])
+ except ValueError:
+ await update.message.reply_text("❌ Invalid issue number.")
+ return
+
+ msg = await update.message.reply_text(f"🔍 Fetching bounty #{number}...")
+ try:
+ bounty = GitHubClient().fetch_bounty(number)
+ if not bounty:
+ await msg.edit_text(f"❌ Bounty #{number} not found.")
+ return
+ text, keyboard = format_bounty_detail(bounty)
+ await msg.edit_text(text, parse_mode="HTML", reply_markup=keyboard, disable_web_page_preview=True)
+ except Exception as e:
+ logger.exception("Bounty fetch failed")
+ await msg.edit_text(f"❌ Failed to fetch bounty #{number}: {e}")
+
+
+async def handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+ query = update.callback_query
+ if not query or not query.data:
+ return
+ await query.answer()
+ data = query.data
+
+ if data == "subscribe":
+ store = SubscriptionStore()
+ store.save_filter(UserFilter(user_id=query.from_user.id))
+ await query.edit_message_text(
+ "✅ Subscribed! Use /filter to customize what you get notified about.",
+ )
+ return
+
+ if data.startswith("claim:"):
+ number = data.split(":", 1)[1]
+ await query.edit_message_text(
+ f"✅ Head to GitHub to claim this bounty:\n"
+ f"https://github.com/SolFoundry/solfoundry/issues/{number}\n\n"
+ "1. Fork the repo\n"
+ "2. Open a PR with your solution\n"
+ "3. Include Closes #{number} in the PR description",
+ parse_mode="HTML", disable_web_page_preview=True,
+ )
diff --git a/telegram-bot/bot/models.py b/telegram-bot/bot/models.py
new file mode 100644
index 000000000..7d58ce0f0
--- /dev/null
+++ b/telegram-bot/bot/models.py
@@ -0,0 +1,96 @@
+"""Pydantic models for GitHub issues / bounty data."""
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Optional
+
+
+@dataclass
+class Bounty:
+ """Represents a SolFoundry bounty (GitHub issue)."""
+ number: int
+ title: str
+ body: str
+ state: str
+ labels: list[str]
+ assignee: Optional[str]
+ created_at: datetime
+ updated_at: datetime
+ html_url: str
+
+ @property
+ def is_open(self) -> bool:
+ return self.state == "open"
+
+ @property
+ def tier(self) -> Optional[str]:
+ for label in self.labels:
+ if label.startswith("bounty-tier-"):
+ return label.replace("bounty-tier-", "")
+ return None
+
+ @property
+ def reward(self) -> Optional[str]:
+ for label in self.labels:
+ if label.startswith("bounty-reward-"):
+ return label.replace("bounty-reward-", "") + " FNDRY"
+ return None
+
+ @property
+ def bounty_type(self) -> Optional[str]:
+ type_labels = {"bounty-feature", "bounty-bug", "bounty-docs", "bounty-integration", "bounty-security"}
+ for label in self.labels:
+ if label in type_labels:
+ return label.replace("bounty-", "")
+ return None
+
+ def matches_filter(self, user_filter: "UserFilter") -> bool:
+ if user_filter.tiers and self.tier not in user_filter.tiers:
+ return False
+ if user_filter.types and self.bounty_type not in user_filter.types:
+ return False
+ if user_filter.min_reward and self.reward:
+ try:
+ reward_num = int("".join(filter(str.isdigit, self.reward)))
+ if reward_num < user_filter.min_reward:
+ return False
+ except ValueError:
+ pass
+ return True
+
+
+@dataclass
+class UserFilter:
+ """Per-user notification filter settings."""
+ user_id: int
+ tiers: Optional[list[str]] = None
+ types: Optional[list[str]] = None
+ min_reward: Optional[int] = None
+ keywords: Optional[list[str]] = None
+
+ def to_dict(self) -> dict:
+ return {
+ "user_id": self.user_id,
+ "tiers": self.tiers,
+ "types": self.types,
+ "min_reward": self.min_reward,
+ "keywords": self.keywords,
+ }
+
+ @classmethod
+ def from_dict(cls, data: dict) -> "UserFilter":
+ return cls(
+ user_id=data["user_id"],
+ tiers=data.get("tiers"),
+ types=data.get("types"),
+ min_reward=data.get("min_reward"),
+ keywords=data.get("keywords"),
+ )
+
+
+@dataclass
+class LeaderboardEntry:
+ rank: int
+ username: str
+ merged_count: int
+ total_reward: int
+ avatar_url: Optional[str] = None
diff --git a/telegram-bot/bot/subscription_store.py b/telegram-bot/bot/subscription_store.py
new file mode 100644
index 000000000..cd7fa44d8
--- /dev/null
+++ b/telegram-bot/bot/subscription_store.py
@@ -0,0 +1,78 @@
+"""Redis-backed subscription store for per-user filter settings."""
+import json
+import logging
+from typing import Optional
+
+import redis
+
+from bot.config import config
+from bot.models import UserFilter
+
+logger = logging.getLogger(__name__)
+
+
+class SubscriptionStore:
+ """
+ Persists per-user filter settings in Redis.
+
+ Key layout:
+ solfoundry:subscriptions:{user_id} -> JSON UserFilter
+ solfoundry:notified:{issue_number} -> "1" if notified (dedup, 7d TTL)
+ solfoundry:last_check -> ISO timestamp
+ """
+
+ SUBSCRIPTIONS_KEY = "solfoundry:subscriptions"
+ NOTIFIED_KEY_PREFIX = "solfoundry:notified:"
+ LAST_CHECK_KEY = "solfoundry:last_check"
+
+ def __init__(self, redis_url: Optional[str] = None):
+ self.redis_url = redis_url or config.redis_url
+ self._client: Optional[redis.Redis] = None
+
+ @property
+ def client(self) -> redis.Redis:
+ if self._client is None:
+ self._client = redis.from_url(self.redis_url, decode_responses=True)
+ return self._client
+
+ def _sub_key(self, user_id: int) -> str:
+ return f"{self.SUBSCRIPTIONS_KEY}:{user_id}"
+
+ def get_filter(self, user_id: int) -> UserFilter:
+ raw = self.client.get(self._sub_key(user_id))
+ if raw:
+ try:
+ return UserFilter.from_dict(json.loads(raw))
+ except (json.JSONDecodeError, KeyError):
+ pass
+ return UserFilter(user_id=user_id)
+
+ def save_filter(self, user_filter: UserFilter) -> None:
+ self.client.set(self._sub_key(user_filter.user_id), json.dumps(user_filter.to_dict()))
+
+ def delete_filter(self, user_id: int) -> None:
+ self.client.delete(self._sub_key(user_id))
+
+ def list_subscriptions(self) -> list[UserFilter]:
+ keys = self.client.keys(f"{self.SUBSCRIPTIONS_KEY}:*")
+ filters = []
+ for key in keys:
+ raw = self.client.get(key)
+ if raw:
+ try:
+ filters.append(UserFilter.from_dict(json.loads(raw)))
+ except (json.JSONDecodeError, KeyError):
+ pass
+ return filters
+
+ def is_notified(self, issue_number: int) -> bool:
+ return self.client.exists(f"{self.NOTIFIED_KEY_PREFIX}{issue_number}") > 0
+
+ def mark_notified(self, issue_number: int, ttl: int = 86400 * 7) -> None:
+ self.client.setex(f"{self.NOTIFIED_KEY_PREFIX}{issue_number}", ttl, "1")
+
+ def get_last_check(self) -> Optional[str]:
+ return self.client.get(self.LAST_CHECK_KEY)
+
+ def set_last_check(self, iso_timestamp: str) -> None:
+ self.client.set(self.LAST_CHECK_KEY, iso_timestamp)
diff --git a/telegram-bot/main.py b/telegram-bot/main.py
new file mode 100644
index 000000000..3a7191cb8
--- /dev/null
+++ b/telegram-bot/main.py
@@ -0,0 +1,22 @@
+"""Entry point for the SolFoundry Telegram Bot."""
+from bot.app import app_fastapi, run_polling, run_webhook
+from bot.config import config
+
+if __name__ == "__main__":
+ import asyncio, sys
+
+ errors = config.validate()
+ if errors:
+ for e in errors:
+ print(f"CONFIG ERROR: {e}", file=sys.stderr)
+ sys.exit(1)
+
+ mode = config.bot_mode.lower()
+ if mode == "polling":
+ asyncio.run(run_polling())
+ elif mode == "webhook":
+ import uvicorn
+ uvicorn.run("bot.app:app", host="0.0.0.0", port=8080, reload=False)
+ else:
+ print(f"Unknown BOT_MODE: {mode}", file=sys.stderr)
+ sys.exit(1)
diff --git a/telegram-bot/pyproject.toml b/telegram-bot/pyproject.toml
new file mode 100644
index 000000000..cc9587b3c
--- /dev/null
+++ b/telegram-bot/pyproject.toml
@@ -0,0 +1,21 @@
+[project]
+name = "solfoundry-telegram-bot"
+version = "0.1.0"
+description = "Telegram bot for SolFoundry bounty notifications"
+requires-python = ">=3.10"
+dependencies = [
+ "python-telegram-bot==22.0",
+ "httpx>=0.27.0",
+ "redis>=5.0.0",
+ "python-dotenv>=1.0.0",
+ "pydantic>=2.0.0",
+ "fastapi>=0.100.0",
+ "uvicorn>=0.30.0",
+]
+
+[project.optional-dependencies]
+dev = ["pytest", "pytest-asyncio", "pytest-mock"]
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+testpaths = ["tests"]
diff --git a/telegram-bot/tests/__init__.py b/telegram-bot/tests/__init__.py
new file mode 100644
index 000000000..a9fcea6bc
--- /dev/null
+++ b/telegram-bot/tests/__init__.py
@@ -0,0 +1 @@
+# Tests for SolFoundry Telegram Bot
diff --git a/telegram-bot/tests/test_handlers.py b/telegram-bot/tests/test_handlers.py
new file mode 100644
index 000000000..5d1f252a3
--- /dev/null
+++ b/telegram-bot/tests/test_handlers.py
@@ -0,0 +1,47 @@
+"""Tests for bot/handlers.py — filter parsing."""
+from bot.handlers import parse_filter_args
+
+
+def test_parse_filter_empty():
+ f, err = parse_filter_args("")
+ # Empty string = signal caller to show current filter status
+ assert f is None
+ assert err is None
+
+
+def test_parse_filter_clear():
+ f, err = parse_filter_args("clear")
+ assert f is None
+ assert err is None
+
+
+def test_parse_filter_tier():
+ f, err = parse_filter_args("tier:1,2")
+ assert err is None
+ assert f.tiers == ["1", "2"]
+
+
+def test_parse_filter_type():
+ f, err = parse_filter_args("type:feature,bug")
+ assert err is None
+ assert f.types == ["feature", "bug"]
+
+
+def test_parse_filter_min():
+ f, err = parse_filter_args("min:500")
+ assert err is None
+ assert f.min_reward == 500
+
+
+def test_parse_filter_combined():
+ f, err = parse_filter_args("tier:2,3 type:feature min:500")
+ assert err is None
+ assert f.tiers == ["2", "3"]
+ assert f.types == ["feature"]
+ assert f.min_reward == 500
+
+
+def test_parse_filter_invalid_min():
+ f, err = parse_filter_args("min:abc")
+ assert f is None
+ assert "Invalid min value" in err
diff --git a/telegram-bot/tests/test_models.py b/telegram-bot/tests/test_models.py
new file mode 100644
index 000000000..30aa3b895
--- /dev/null
+++ b/telegram-bot/tests/test_models.py
@@ -0,0 +1,66 @@
+"""Tests for bot/models.py."""
+from datetime import datetime, timezone
+
+from bot.models import Bounty, UserFilter
+
+
+def test_bounty_tier_from_label():
+ b = Bounty(
+ number=1, title="Test", body="", state="open",
+ labels=["bounty-tier-2"], assignee=None,
+ created_at=datetime.now(timezone.utc),
+ updated_at=datetime.now(timezone.utc),
+ html_url="https://github.com/test",
+ )
+ assert b.tier == "2"
+ assert b.is_open is True
+
+
+def test_bounty_reward_from_label():
+ b = Bounty(
+ number=1, title="Test", body="", state="open",
+ labels=["bounty-tier-1", "bounty-reward-500"],
+ assignee=None,
+ created_at=datetime.now(timezone.utc),
+ updated_at=datetime.now(timezone.utc),
+ html_url="https://github.com/test",
+ )
+ assert b.reward == "500 FNDRY"
+
+
+def test_bounty_bounty_type():
+ b = Bounty(
+ number=1, title="Test", body="", state="open",
+ labels=["bounty-tier-1", "bounty-feature"],
+ assignee=None,
+ created_at=datetime.now(timezone.utc),
+ updated_at=datetime.now(timezone.utc),
+ html_url="https://github.com/test",
+ )
+ assert b.bounty_type == "feature"
+
+
+def test_bounty_matches_filter():
+ b = Bounty(
+ number=1, title="AI API Feature", body="", state="open",
+ labels=["bounty-tier-2", "bounty-feature"],
+ assignee=None,
+ created_at=datetime.now(timezone.utc),
+ updated_at=datetime.now(timezone.utc),
+ html_url="https://github.com/test",
+ )
+ assert b.matches_filter(UserFilter(user_id=1)) is True
+ assert b.matches_filter(UserFilter(user_id=1, tiers=["2"])) is True
+ assert b.matches_filter(UserFilter(user_id=1, tiers=["1"])) is False
+ assert b.matches_filter(UserFilter(user_id=1, types=["feature"])) is True
+ assert b.matches_filter(UserFilter(user_id=1, types=["bug"])) is False
+ assert b.matches_filter(UserFilter(user_id=1, tiers=["2"], types=["feature"])) is True
+
+
+def test_user_filter_serialization():
+ f = UserFilter(user_id=123, tiers=["2", "3"], types=["bug"], min_reward=500)
+ restored = UserFilter.from_dict(f.to_dict())
+ assert restored.user_id == 123
+ assert restored.tiers == ["2", "3"]
+ assert restored.types == ["bug"]
+ assert restored.min_reward == 500