bot.py

# A Telegram bot on webhooks, ready to deploy on velixir.
#
# Telegram can hand a bot its messages in two ways. With long polling the bot keeps asking
# Telegram for news, so it has to run around the clock. With a webhook, Telegram sends each
# update to the bot as an HTTPS request instead. This template uses a webhook, and that is
# what lets it run on velixir's Free plan: the app sleeps after 15 minutes without a request,
# and the next message wakes it.
#
# The whole flow: on boot the app tells Telegram its own address (setWebhook). Telegram then
# POSTs every update to /telegram with a secret header, and the app checks the header,
# replies, and answers 200. Everything else is the status page at / and /healthz.

import asyncio
import hashlib
import hmac
import html
import logging
import os
import time
from collections import deque
from contextlib import asynccontextmanager, suppress
from dataclasses import dataclass, field
from urllib.parse import urlsplit

import httpx
import uvicorn
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import HTMLResponse, PlainTextResponse, Response
from starlette.routing import Route

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
# httpx logs the URL of every request at INFO, and every Bot API URL contains the token.
logging.getLogger("httpx").setLevel(logging.WARNING)
log = logging.getLogger("bot")

PORT = int(os.environ.get("PORT", "8080"))
TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "").strip()

# Where the Bot API lives. Change it only to point the bot at a mock server in tests, or at a
# self-hosted Bot API server.
API_BASE = os.environ.get("TELEGRAM_API_BASE", "https://api.telegram.org").strip().rstrip("/")

WEBHOOK_PATH = "/telegram"

# The kinds of update Telegram should send. Every kind left out is a request that never
# arrives, which on the Free plan is one less wake-up. Add "callback_query" when you add
# inline keyboards, or "edited_message" to hear about edits.
ALLOWED_UPDATES = ["message"]


# ─── Commands ───────────────────────────────────────────────────────────────
#
# The part to make your own. A command gets the incoming message and whatever text followed
# the command, and returns its reply as Telegram HTML. Anything that came from a user goes
# through html.escape on the way out, as it would on any web page. Add a command to COMMANDS
# and it shows up in /help, and in Telegram's "/" menu after the next deploy.

async def start(message: dict, args: str) -> str:
    name = html.escape(message.get("from", {}).get("first_name", "there"))
    return (
        f"Hello, {name}. I am a small bot running on velixir.\n\n"
        "Send me a message and I will send it straight back, or try /help."
    )


async def help_(message: dict, args: str) -> str:
    lines = [f"/{name}  {description}" for name, (description, _) in COMMANDS.items()]
    return "\n".join(lines) + "\n\nIn a private chat, anything that is not a command comes back to you."


async def chat_id(message: dict, args: str) -> str:
    # Handy once the bot talks to other systems: a chat id is what sendMessage needs to post
    # into this chat later, without waiting to be spoken to first.
    reply = f"This chat's id is <code>{message['chat']['id']}</code>."
    if "from" in message:
        reply += f"\nYour user id is <code>{message['from']['id']}</code>."
    return reply


COMMANDS = {
    "start": ("Say hello", start),
    "help": ("What this bot can do", help_),
    "id": ("Show this chat's id", chat_id),
}


async def echo(message: dict) -> str | None:
    # Private chats only. In a group the bot would repeat everyone, which nobody wants.
    if message["chat"].get("type") != "private":
        return None
    # Telegram caps a message at 4096 characters, and the prefix needs room.
    return "You said: " + html.escape(message["text"][:4000])


async def handle_update(update: dict) -> None:
    message = update.get("message")
    if not isinstance(message, dict) or not isinstance(message.get("text"), str):
        return  # photos, stickers, people joining: nothing to answer

    if message["text"].startswith("/"):
        reply = await run_command(message)
    else:
        reply = await echo(message)
    if reply:
        await send_message(message["chat"]["id"], reply)


async def run_command(message: dict) -> str | None:
    word, *rest = message["text"].split(maxsplit=1)
    # In a group a command can name its bot, as /help@SomeBot. Leave other bots' commands alone.
    name, _, addressee = word[1:].partition("@")
    if addressee and addressee.lower() != (state.username or "").lower():
        return None

    command = COMMANDS.get(name.lower())
    if command:
        return await command[1](message, rest[0] if rest else "")
    if message["chat"].get("type") == "private":
        return "I do not know that one. Try /help."
    return None  # an unknown command in a group is probably meant for another bot


# ─── State ──────────────────────────────────────────────────────────────────

# What the status page shows. It lives in memory, so it starts from nothing on every boot,
# which is also every wake-up on the Free plan.
@dataclass
class State:
    username: str | None = None                # the bot's @username, from getMe
    webhook_ok: bool = False
    webhook: str = "Registering with Telegram."  # one line for the status page
    telegram_view: dict | None = None          # getWebhookInfo: Telegram's side of the story
    telegram_view_at: float = 0.0
    updates: int = 0
    replies: int = 0
    last_error: str | None = None
    last_error_at: float | None = None
    started_at: float = field(default_factory=time.time)


state = State()


def redact(text: str) -> str:
    # The token is part of every Bot API URL, and a network error can quote the URL. Keep the
    # token out of the logs and off the status page.
    return text.replace(TOKEN, "<token>") if TOKEN else text


def record_error(message: str) -> None:
    state.last_error, state.last_error_at = redact(message), time.time()
    log.warning(state.last_error)


# ─── Talking to Telegram ────────────────────────────────────────────────────

client: httpx.AsyncClient  # created on startup, in lifespan() below


class TelegramError(Exception):
    """A Bot API call that failed, with Telegram's error code when there was one."""

    def __init__(self, method: str, description: str, code: int | None = None):
        super().__init__(redact(f"{method} failed: {description}"))
        self.code = code


async def telegram(method: str, **params):
    """Calls one Bot API method and returns its result, or raises TelegramError."""
    try:
        response = await client.post(f"{API_BASE}/bot{TOKEN}/{method}", json=params)
        body = response.json()
    except Exception as exc:  # unreachable, timed out, or an answer that is not JSON
        reason = f"{type(exc).__name__}: {exc}" if str(exc) else type(exc).__name__
        # "from None": the original exception can carry the request URL, token and all.
        raise TelegramError(method, f"could not reach Telegram ({reason})") from None

    if not isinstance(body, dict) or not body.get("ok"):
        body = body if isinstance(body, dict) else {}
        raise TelegramError(
            method,
            body.get("description", f"HTTP {response.status_code}"),
            body.get("error_code", response.status_code),
        )
    return body.get("result")


async def send_message(chat: int, text: str) -> None:
    try:
        await telegram("sendMessage", chat_id=chat, text=text, parse_mode="HTML")
        state.replies += 1
    except TelegramError as exc:
        # Logged and shown on the status page, never raised: the update still gets its 200,
        # so Telegram does not deliver it again and again.
        record_error(str(exc))


# ─── Webhook registration ───────────────────────────────────────────────────

def public_url() -> str | None:
    """The address Telegram should deliver to, with no trailing slash, or None."""
    # PUBLIC_URL wins, for a custom domain. Otherwise every velixir app answers on
    # https://<slug>.velixir.run, and the platform tells the app its slug.
    url = os.environ.get("PUBLIC_URL", "").strip().rstrip("/")
    if url:
        return url if "://" in url else f"https://{url}"
    slug = os.environ.get("VELIXIR_APP_SLUG", "").strip()
    return f"https://{slug}.velixir.run" if slug else None


def webhook_secret(token: str) -> str:
    # Telegram repeats this secret in a header on every update, which is how the app tells a
    # real update from anyone else who finds the URL. It is derived from the token rather
    # than made up at random, so it survives restarts, agrees across replicas and needs no
    # configuration. HMAC-SHA256 in hex fits Telegram's rules for it (A-Z, a-z, 0-9, _ and
    # -, up to 256 characters). A new token from BotFather means a new secret.
    return hmac.new(token.encode(), b"velixir-telegram-webhook", hashlib.sha256).hexdigest()


SECRET = webhook_secret(TOKEN) if TOKEN else ""


async def register_webhook() -> None:
    """Points Telegram at this app. Runs in the background on every boot, and never raises."""
    base = public_url()
    if base is None:
        state.webhook = "No public address to give Telegram. On velixir it is automatic; elsewhere set PUBLIC_URL."
        log.warning(state.webhook)
        return
    if not base.startswith("https://"):
        state.webhook = f"Telegram only delivers to https:// addresses, and PUBLIC_URL is {base}."
        log.warning(state.webhook)
        return

    # On every boot, not only the first. setWebhook is idempotent, it is the only way to be
    # sure Telegram holds the current secret, and it takes the webhook back from any other
    # copy of the bot that registered its own address in the meantime.
    delay = 5
    while True:
        try:
            state.username = (await telegram("getMe")).get("username")
            await telegram(
                "setWebhook",
                url=base + WEBHOOK_PATH,
                secret_token=SECRET,
                allowed_updates=ALLOWED_UPDATES,
            )
            break
        except Exception as exc:
            record_error(str(exc))
            if getattr(exc, "code", None) in (401, 404):
                # 401 is a token Telegram does not know (mistyped, or revoked in BotFather),
                # 404 one that is not even shaped like a token. Retrying cannot fix either.
                state.webhook = "Telegram rejected TELEGRAM_BOT_TOKEN. Copy it again from @BotFather, set it, and redeploy."
                return
            state.webhook = "Not registered yet. Retrying on its own; the last error is below."
        await asyncio.sleep(delay)
        delay = min(delay * 2, 300)

    state.webhook_ok = True
    state.webhook = f"Registered at {base}{WEBHOOK_PATH}"
    log.info("webhook registered at %s%s for @%s", base, WEBHOOK_PATH, state.username)

    # The menu Telegram shows when someone types "/". Nice to have, so a failure is only logged.
    try:
        await telegram("setMyCommands", commands=[
            {"command": name, "description": description}
            for name, (description, _) in COMMANDS.items()
        ])
    except TelegramError as exc:
        record_error(str(exc))

    await refresh_telegram_view()


async def refresh_telegram_view() -> None:
    try:
        state.telegram_view = await telegram("getWebhookInfo")
        state.telegram_view_at = time.time()
    except TelegramError as exc:
        record_error(str(exc))


# ─── Routes ─────────────────────────────────────────────────────────────────

# Telegram delivers a lost update again, for example when a reply took too long to reach it.
# Remembering the last few ids keeps a redelivery from being answered twice.
recent_updates: deque = deque(maxlen=200)


async def webhook(request: Request) -> Response:
    if not TOKEN:
        return PlainTextResponse("No TELEGRAM_BOT_TOKEN is set, so this bot cannot take updates yet.", status_code=503)

    # compare_digest takes as long for a nearly right guess as for a wildly wrong one, so the
    # secret cannot be worked out a character at a time from how fast the app says no.
    received = request.headers.get("x-telegram-bot-api-secret-token", "")
    if not hmac.compare_digest(received.encode(), SECRET.encode()):
        return PlainTextResponse("Unauthorized", status_code=401)

    try:
        update = await request.json()
    except ValueError:
        return PlainTextResponse("Bad Request", status_code=400)
    if not isinstance(update, dict):
        return PlainTextResponse("Bad Request", status_code=400)

    update_id = update.get("update_id")
    if update_id is not None:
        if update_id in recent_updates:
            return Response(status_code=200)
        recent_updates.append(update_id)

    state.updates += 1
    log.info("update %s received", update_id)
    try:
        await handle_update(update)
    except Exception as exc:
        # Answer 200 anyway. Telegram retries an update until it gets a 2xx, so a message this
        # code chokes on would otherwise come back again and again.
        log.exception("handling update %s failed", update_id)
        record_error(f"Handling an update failed: {type(exc).__name__}: {exc}")
    return Response(status_code=200)


async def healthz(request: Request) -> Response:
    # Liveness only. Whether the bot works depends on its token and on Telegram, and
    # restarting the container fixes neither, so this stays 200 and the status page explains.
    if not TOKEN:
        detail = "setup mode: TELEGRAM_BOT_TOKEN is not set"
    elif state.webhook_ok:
        detail = "webhook registered"
    else:
        detail = "webhook not registered, see /"
    return PlainTextResponse(f"ok ({detail})")


async def status_page(request: Request) -> Response:
    if not TOKEN:
        return HTMLResponse(page("Connect your Telegram bot", setup_body()))

    # Telegram's own view of the webhook (updates queued, its last delivery error) is the
    # quickest way to see why messages are not arriving. Cached for half a minute, since
    # otherwise every view of this page is another Bot API call.
    if state.webhook_ok and time.time() - state.telegram_view_at > 30:
        await refresh_telegram_view()
    return HTMLResponse(page(f"@{state.username}" if state.username else "Telegram bot", status_body()))


# ─── Pages ──────────────────────────────────────────────────────────────────

def esc(value) -> str:
    return html.escape(str(value))


def when(timestamp: float | None) -> str:
    if not timestamp:
        return "an unknown time"
    return time.strftime("%Y-%m-%d %H:%M UTC", time.gmtime(timestamp))


def plural(count: int, one: str, many: str) -> str:
    return f"{count} {one if count == 1 else many}"


def status_body() -> str:
    name = f"@{state.username}" if state.username else "Your Telegram bot"
    expected = f"{public_url()}{WEBHOOK_PATH}"
    view = state.telegram_view or {}

    rows = [("Webhook", f'<span class="dot {"ok" if state.webhook_ok else "bad"}"></span>{esc(state.webhook)}')]

    if state.telegram_view is not None:
        reported = f"{plural(view.get('pending_update_count', 0), 'update', 'updates')} waiting"
        if view.get("last_error_message"):
            reported += (f", last delivery failed at {esc(when(view.get('last_error_date')))}: "
                         f"<code>{esc(view['last_error_message'])}</code>")
        else:
            reported += ", no delivery errors"
        rows.append(("Telegram reports", reported))

        if view.get("url") != expected:
            # Only the host: a webhook URL elsewhere may carry a secret in its path, and this
            # page is public.
            other = urlsplit(view.get("url") or "").netloc
            where = f"to <code>{esc(other)}</code>" if other else "nowhere (no webhook is set)"
            rows.append(("Heads up",
                         f'<span class="dot warn"></span>Telegram is delivering this bot\'s updates {where}, '
                         "not to this app. Another copy of the bot probably registered its own address. "
                         "Redeploy this app to take the webhook back."))

    rows.append(("Since this start",
                 f"{plural(state.updates, 'update', 'updates')} in, {plural(state.replies, 'reply', 'replies')} sent "
                 f"(up since {esc(when(state.started_at))})"))
    rows.append(("Last error", f'<code class="error">{esc(state.last_error)}</code> at {esc(when(state.last_error_at))}'
                 if state.last_error else "None"))

    facts = "".join(f"<div><dt>{label}</dt><dd>{value}</dd></div>" for label, value in rows)
    link = (f'<p class="note">Message <a href="https://t.me/{esc(state.username)}">{esc(name)}</a> on Telegram and it answers. '
            "On the Free plan, the first message after a quiet spell takes a few seconds while the app wakes up.</p>"
            if state.username else "")

    return f"""
  <h1>{esc(name)}</h1>
  <p class="sub">A Telegram bot on webhooks. Telegram posts each message to this app.</p>
  <dl class="facts">{facts}</dl>
  {link}"""


def setup_body() -> str:
    return """
  <h1>Connect your Telegram bot</h1>
  <p class="sub">This app is running. It needs a bot token from Telegram before it can answer anyone.</p>
  <ol class="steps">
    <li><strong>Create a bot.</strong> In Telegram, open <a href="https://t.me/BotFather">@BotFather</a>,
      send <code>/newbot</code>, and choose a name and a username ending in <code>bot</code>.
      BotFather replies with a token that looks like <code>123456789:AAH...</code>.</li>
    <li><strong>Give it to this app.</strong> On the app's Environment tab, add
      <code>TELEGRAM_BOT_TOKEN</code> with the token as its value, and tick Treat as secret.</li>
    <li><strong>Redeploy</strong> with the Redeploy button on the live release (Deploys tab). On
      boot the app registers its webhook with Telegram. Then send your bot <code>/start</code>.</li>
  </ol>
  <p class="note">The token is the key to your bot: whoever has it can act as the bot. If it
    leaks, send <code>/revoke</code> to BotFather, then set the new token and redeploy.</p>"""


STYLE = """
  :root { color-scheme: dark; }
  * { box-sizing: border-box; }
  body { margin:0; min-height:100vh; background:#0b0b10; color:#e8e8ef;
         font:15px/1.6 ui-sans-serif,system-ui,-apple-system,Segoe UI,sans-serif; }
  .wrap { max-width:38rem; margin:0 auto; padding:3rem 1.25rem 4rem; }
  h1 { font-size:1.5rem; margin:0 0 .25rem; letter-spacing:-.02em; overflow-wrap:anywhere; }
  .sub { color:#7a7a8c; font-size:.8125rem; margin:0 0 2rem; }
  a { color:#a5b4fc; }
  code { background:#16161f; border:1px solid #26263a; border-radius:5px;
         padding:.1rem .35rem; color:#c7d2fe; font-size:.8125rem; overflow-wrap:anywhere; }
  code.error { color:#fca5a5; }
  dl.facts { margin:0 0 1.5rem; border:1px solid #1f1f2e; border-radius:10px; }
  dl.facts div { display:grid; grid-template-columns:9rem 1fr; gap:1rem; padding:.8rem 1rem;
                 border-top:1px solid #1f1f2e; }
  dl.facts div:first-child { border-top:0; }
  dt { color:#7a7a8c; font-size:.8125rem; padding-top:.1rem; }
  dd { margin:0; overflow-wrap:anywhere; }
  .dot { display:inline-block; width:.5rem; height:.5rem; border-radius:50%; background:#4b4b5c;
         margin-right:.5rem; vertical-align:middle; }
  .dot.ok { background:#34d399; } .dot.warn { background:#fbbf24; } .dot.bad { background:#f87171; }
  ol.steps { padding-left:1.25rem; margin:0 0 1.5rem; }
  ol.steps li { margin-bottom:.9rem; }
  .note { color:#7a7a8c; font-size:.8125rem; }
  footer { color:#4b4b5c; font-size:.75rem; margin-top:2.5rem; text-align:center; }
  footer a { color:#7a7a8c; }
  @media (max-width: 30rem) { dl.facts div { grid-template-columns:1fr; gap:.15rem; } }
"""


def page(title: str, body: str) -> str:
    return f"""<!doctype html>
<html lang="en">
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>{esc(title)}</title>
<style>{STYLE}</style>
<div class="wrap">
  {body}
  <footer>
    <a href="/healthz">Health</a> &middot; running on <a href="https://velixir.net">velixir</a>
  </footer>
</div>
</html>"""


# ─── Start ──────────────────────────────────────────────────────────────────

@asynccontextmanager
async def lifespan(app: Starlette):
    global client
    client = httpx.AsyncClient(timeout=10)

    # Registration runs in the background, so the server is answering (the setup page,
    # /healthz, and the update that just woke the app) while it talks to Telegram.
    task = None
    if TOKEN:
        task = asyncio.create_task(register_webhook())
    else:
        log.info(
            "TELEGRAM_BOT_TOKEN is not set, so this app is serving its setup page.\n"
            "On velixir: add TELEGRAM_BOT_TOKEN on the Environment tab, then redeploy.\n"
            "Locally: TELEGRAM_BOT_TOKEN=... PUBLIC_URL=https://<your tunnel> python bot.py"
        )

    yield

    if task:
        task.cancel()
        with suppress(asyncio.CancelledError):
            await task
    await client.aclose()


app = Starlette(
    routes=[
        Route("/", status_page),
        Route("/healthz", healthz),
        Route(WEBHOOK_PATH, webhook, methods=["POST"]),
    ],
    lifespan=lifespan,
)

if __name__ == "__main__":
    # 0.0.0.0, not localhost: the platform's traffic arrives from outside the container.
    # No access log: health checks would bury everything else, and updates are logged above.
    uvicorn.run(app, host="0.0.0.0", port=PORT, access_log=False)