bot.py

# A Discord bot with slash commands, ready to deploy on velixir.
#
# Two things share one process and one event loop: the bot, which keeps a WebSocket open to
# Discord's gateway, and a small web server on $PORT for the status page and /healthz. The web
# server starts first and never depends on the bot, so a missing or rejected token gets you a
# page explaining what to fix instead of a crash loop.
#
# Discord never sends a gateway bot an HTTP request, so nothing could wake it from sleep. That
# is why it needs a plan that stays on, and why velixir.json says "alwaysOn": true.

import asyncio
import html
import logging
import math
import os
import random
import signal
import time
from contextlib import suppress
from dataclasses import dataclass

import discord
from aiohttp import web
from discord import app_commands

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(name)s: %(message)s")
log = logging.getLogger("bot")

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

TOKEN = os.environ.get("DISCORD_TOKEN", "").strip()
# Pasting the token with a "Bot " prefix is a common slip. discord.py adds the prefix itself.
if TOKEN.lower().startswith("bot "):
    TOKEN = TOKEN[4:].strip()

# While developing, register the commands to one server: changes there show up instantly,
# where global ones can take a while to reach every server. See the README.
GUILD_ID_TEXT = os.environ.get("DISCORD_GUILD_ID", "").strip()
GUILD_ID = int(GUILD_ID_TEXT) if GUILD_ID_TEXT.isdigit() else None

# This bot does not do voice (and velixir apps cannot send the UDP it needs), so skip
# discord.py's warnings at every start that the voice libraries are not installed.
discord.VoiceClient.warn_nacl = False
discord.VoiceClient.warn_dave = False

STARTED_AT = time.time()


# ─── Commands ───────────────────────────────────────────────────────────────
#
# The part to make your own. Each function is a slash command: the function name is the
# command, the description is what Discord shows under it, and typed parameters become its
# options. Add one to COMMANDS and it is registered with Discord on the next start.

@app_commands.command(description="Check the bot is alive, and how quickly Discord hears from it")
async def ping(interaction: discord.Interaction) -> None:
    await interaction.response.send_message(f"Pong. Gateway latency: {latency_text(interaction.client.latency)}.")


@app_commands.command(description="Roll some dice")
@app_commands.describe(sides="Sides on each die (default 6)", count="How many dice to roll (default 1)")
async def roll(
    interaction: discord.Interaction,
    sides: app_commands.Range[int, 2, 1000] = 6,
    count: app_commands.Range[int, 1, 20] = 1,
) -> None:
    # Range makes Discord enforce the limits in its own interface, so a bad value never gets here.
    rolls = [random.randint(1, sides) for _ in range(count)]
    if count == 1:
        text = f"Rolled a d{sides}: **{rolls[0]}**"
    else:
        text = f"Rolled {count}d{sides}: {' + '.join(map(str, rolls))} = **{sum(rolls)}**"
    await interaction.response.send_message(text)


@app_commands.command(description="What this bot is and where it runs")
async def about(interaction: discord.Interaction) -> None:
    client = interaction.client
    servers = len(client.guilds)
    await interaction.response.send_message(
        f"**{client.user.name}** is a starter bot built on discord.py {discord.__version__}, running on "
        f"velixir. It is in {servers} server{'' if servers == 1 else 's'} and has been up for "
        f"{uptime_text(time.time() - STARTED_AT)}."
    )


COMMANDS = [ping, roll, about]


async def on_command_error(interaction: discord.Interaction, error: app_commands.AppCommandError) -> None:
    name = interaction.command.name if interaction.command else "unknown"
    original = getattr(error, "original", error)
    if isinstance(original, discord.HTTPException) and original.code == 40060:
        # "Already acknowledged": the user has their answer. During a redeploy the old and new
        # copies are both connected for a moment and both hear each command; one answers
        # first and the other lands here. (Outside a redeploy it means the code answered twice.)
        log.warning("/%s was already answered, by another copy of the bot or earlier in this command", name)
        return

    log.error("/%s failed", name, exc_info=original)
    record_error(f"/{name} failed: {type(original).__name__}: {original}")
    if not interaction.response.is_done():
        with suppress(discord.HTTPException):
            await interaction.response.send_message("Sorry, that command failed. The details are in the bot's log.", ephemeral=True)


# ─── The bot ────────────────────────────────────────────────────────────────

class Bot(discord.Client):
    def __init__(self) -> None:
        # Slash commands arrive as interactions, and interactions need no intents at all. The
        # guilds intent is only here so the bot knows which servers it is in, for the count on
        # the status page, and it is not a privileged one. The README covers adding more.
        intents = discord.Intents.none()
        intents.guilds = True
        # Nothing the bot says pings @everyone or a role unless a command asks for it by name.
        # Otherwise one command that repeats user input is all it takes to ping a whole server.
        super().__init__(intents=intents, allowed_mentions=discord.AllowedMentions(everyone=False, roles=False))

        self.tree = app_commands.CommandTree(self)
        self.tree.error(on_command_error)
        for command in COMMANDS:
            self.tree.add_command(command)

    async def setup_hook(self) -> None:
        # Runs once after logging in, before the gateway connects: the moment to tell Discord
        # which commands exist.
        await sync_commands(self)

    async def on_ready(self) -> None:
        status.phase = "online"
        count = len(self.guilds)
        log.info("online as %s, in %d server%s", self.user, count, "" if count == 1 else "s")

    async def on_disconnect(self) -> None:
        # discord.py reconnects by itself. This only keeps the status page honest meanwhile.
        if status.phase == "online":
            status.phase = "reconnecting"

    async def on_resumed(self) -> None:
        status.phase = "online"


async def sync_commands(bot: Bot) -> None:
    """Tells Discord which commands exist. A failure is shown on the status page, never raised."""
    # One request that replaces the whole list, so registering the same commands on every start
    # is cheap and changes nothing. What Discord does limit is creating new commands each day,
    # which only adding or renaming one does.
    try:
        if GUILD_ID:
            guild = discord.Object(id=GUILD_ID)
            bot.tree.copy_global_to(guild=guild)
            synced = await bot.tree.sync(guild=guild)
            where = f"to server {GUILD_ID}, where changes show up instantly"
        else:
            synced = await bot.tree.sync()
            where = "globally"
        # Commands registered to a server stay there after you go back to global, so that
        # server may show each one twice. To clear them, deploy once with DISCORD_GUILD_ID set
        # and these two lines in place of the sync above:
        #     bot.tree.clear_commands(guild=guild)
        #     await bot.tree.sync(guild=guild)
    except discord.Forbidden:
        status.commands = (f"Discord would not register commands in server {GUILD_ID}. Add the bot to that "
                           "server with the invite link below, then redeploy.")
        record_error(status.commands)
        return
    except discord.HTTPException as exc:
        status.commands = "Registering the commands failed. The last error is below."
        record_error(f"Registering commands failed: {exc}")
        return

    names = ", ".join(f"/{command.name}" for command in synced)
    status.commands = f"{names} registered {where}"
    status.commands_ok = True
    log.info("commands registered %s: %s", where, names)


# ─── Keeping it connected ───────────────────────────────────────────────────

@dataclass
class Status:
    phase: str = "starting"             # see PHASES below
    bot: discord.Client | None = None
    application_id: int | None = None
    endpoint_url_set: bool = False
    commands: str = "Not registered yet."
    commands_ok: bool = False
    last_error: str | None = None
    last_error_at: float | None = None


status = Status(phase="starting" if TOKEN else "setup")


def record_error(message: str) -> None:
    # discord.py never puts the token in an error, but this page is public, so make sure.
    if TOKEN:
        message = message.replace(TOKEN, "<token>")
    status.last_error, status.last_error_at = message, time.time()
    log.warning(message)


def stop(message: str) -> None:
    record_error(message)
    status.phase = "stopped"
    log.error("The bot has stopped. The status page stays up until the next deploy.")


async def run_bot() -> None:
    """Logs in and keeps the bot connected. Never raises: every failure ends up on the status page."""
    delay = 5
    while True:
        try:
            # A fresh client for every attempt: one that failed half-way is closed and cannot
            # be reused.
            bot = Bot()
        except Exception as exc:
            # Most likely a mistake in the commands (two with the same name, say). Retrying
            # the same code would not help.
            log.exception("could not set up the bot")
            stop(f"Setting up the bot failed: {type(exc).__name__}: {exc}")
            return

        status.bot = bot
        status.phase = "starting"
        try:
            await bot.login(TOKEN)
            status.application_id = bot.application_id
            status.endpoint_url_set = bool(bot.application and bot.application.interactions_endpoint_url)
            status.phase = "connecting"
            await bot.connect(reconnect=True)  # handles dropped connections itself
            return
        except discord.LoginFailure:
            # Never retried. Discord temporarily blocks IP addresses that keep making
            # unauthorised requests, and on a shared platform that address is not only yours.
            stop("Discord rejected DISCORD_TOKEN. Reset the token on the Bot page of the Developer Portal, "
                 "set the new one, and redeploy.")
            return
        except discord.PrivilegedIntentsRequired:
            stop("bot.py asks for a privileged intent that is not switched on for this application. "
                 "Enable it on the Bot page of the Developer Portal, or remove it from bot.py, then redeploy.")
            return
        except discord.ConnectionClosed as exc:
            # The close codes discord.py will not reconnect after (failed authentication, invalid
            # intents, an unsupported API version) do not fix themselves either.
            stop(f"Discord closed the gateway connection with code {exc.code} "
                 f"({exc.reason or 'no reason given'}). Check the token and the intents in bot.py, then redeploy.")
            return
        except Exception as exc:
            # Network trouble or a Discord outage: worth retrying, with a growing pause.
            status.phase = "retrying"
            record_error(f"Could not connect to Discord: {type(exc).__name__}: {exc}")
        finally:
            await bot.close()
        await asyncio.sleep(delay)
        delay = min(delay * 2, 300)


# ─── Web server ─────────────────────────────────────────────────────────────

async def status_page(request: web.Request) -> web.Response:
    if not TOKEN:
        return web.Response(text=page("Connect your Discord bot", setup_body()), content_type="text/html")
    return web.Response(text=page(bot_name(), status_body()), content_type="text/html")


async def healthz(request: web.Request) -> web.Response:
    # Liveness only: 200 while the process runs, whatever the gateway is doing. A restart does
    # not fix a bad token, and a check that failed on one would turn a single error into a
    # crash loop. The gateway state is in the body, for people and monitors.
    return web.Response(text=f"ok ({health_detail()})")


def health_detail() -> str:
    if not TOKEN:
        return "setup mode: DISCORD_TOKEN is not set"
    if status.phase == "online" and status.bot:
        return f"gateway: online, {latency_text(status.bot.latency)}"
    if status.phase == "stopped":
        return "gateway: stopped, see /"
    return f"gateway: {status.phase}"


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

PHASES = {
    "starting": ("warn", "Logging in"),
    "connecting": ("warn", "Connecting to the gateway"),
    "online": ("ok", "Online"),
    "reconnecting": ("warn", "Reconnecting"),
    "retrying": ("bad", "Offline, retrying on its own"),
    "stopped": ("bad", "Offline until the next deploy"),
}


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 latency_text(seconds: float) -> str:
    # NaN before the first heartbeat, infinity while heartbeats go unanswered.
    return f"{seconds * 1000:.0f} ms" if math.isfinite(seconds) else "not measured yet"


def uptime_text(seconds: float) -> str:
    days, rest = divmod(int(seconds) // 60, 1440)
    hours, minutes = divmod(rest, 60)
    if days:
        return f"{days}d {hours}h"
    if hours:
        return f"{hours}h {minutes}m"
    return f"{minutes}m" if minutes else "under a minute"


def bot_name() -> str:
    bot = status.bot
    return bot.user.name if bot and bot.user else "Your Discord bot"


def status_body() -> str:
    bot = status.bot
    phase_dot, phase_label = PHASES.get(status.phase, ("warn", status.phase))
    gateway = f'<span class="dot {phase_dot}"></span>{phase_label}'
    if status.phase == "online" and bot:
        gateway += f", {esc(latency_text(bot.latency))}"

    rows = [("Gateway", gateway)]
    if bot and status.phase in ("online", "reconnecting"):
        rows.append(("Servers", str(len(bot.guilds))))
    commands_dot = "ok" if status.commands_ok else "warn"
    rows.append(("Commands", f'<span class="dot {commands_dot}"></span>{esc(status.commands)}'))
    if GUILD_ID_TEXT and not GUILD_ID:
        rows.append(("Heads up", '<span class="dot warn"></span>DISCORD_GUILD_ID is not a server id (it should be '
                                 "all digits), so it is ignored and the commands are registered globally."))
    if status.endpoint_url_set:
        rows.append(("Heads up", '<span class="dot warn"></span>This application has an Interactions Endpoint URL '
                                 "in the Developer Portal, so Discord sends slash commands there instead of to "
                                 "this bot. Clear it on the General Information page."))
    rows.append(("Up since", esc(when(STARTED_AT))))
    rows.append(("Last error", f'<code class="error">{esc(status.last_error)}</code> at {esc(when(status.last_error_at))}'
                 if status.last_error else "None"))

    facts = "".join(f"<div><dt>{label}</dt><dd>{value}</dd></div>" for label, value in rows)

    invite = ""
    if status.application_id:
        url = discord.utils.oauth_url(status.application_id, permissions=discord.Permissions.none(),
                                      scopes=("bot", "applications.commands"))
        invite = f"""
  <p><a class="button" href="{esc(url)}">Add it to a server</a></p>
  <p class="note">The link asks for no permissions, because slash-command replies need none. If the
    bot needs some later (posting on its own, managing roles), add them in the Developer Portal's
    OAuth2 URL Generator.</p>"""

    return f"""
  <h1>{esc(bot_name())}</h1>
  <p class="sub">A Discord bot with slash commands, on discord.py {esc(discord.__version__)}.</p>
  <dl class="facts">{facts}</dl>
  {invite}"""


def setup_body() -> str:
    return """
  <h1>Connect your Discord bot</h1>
  <p class="sub">This app is running. It needs a bot token from Discord before the bot can come online.</p>
  <ol class="steps">
    <li><strong>Create an application</strong> in the
      <a href="https://discord.com/developers/applications">Discord Developer Portal</a> with New Application.</li>
    <li><strong>Copy its bot token.</strong> On the application's Bot page, press Reset Token and copy
      the token. Leave the Privileged Gateway Intents switched off: this bot does not need them.</li>
    <li><strong>Give it to this app.</strong> On the app's Environment tab, add <code>DISCORD_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). This page
      then shows whether the bot is online, and a link for adding it to a server.</li>
  </ol>
  <p class="note">Developing? Also set <code>DISCORD_GUILD_ID</code> to your test server's id, and command
    changes show up there instantly instead of rolling out to every server.</p>
  <p class="note">The token is the key to your bot: whoever has it can act as the bot. If it leaks,
    reset it in the Developer Portal and redeploy with the new one.</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; }
  a.button { display:inline-block; padding:.7rem 1.1rem; border-radius:8px; background:#6366f1;
             color:#fff; font-weight:600; text-decoration:none; }
  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 ──────────────────────────────────────────────────────────────────

async def main() -> None:
    app = web.Application()
    app.add_routes([web.get("/", status_page), web.get("/healthz", healthz)])
    # No access log: health checks would bury everything else.
    runner = web.AppRunner(app, access_log=None)
    await runner.setup()
    # 0.0.0.0, not localhost: the platform's traffic arrives from outside the container.
    await web.TCPSite(runner, "0.0.0.0", PORT).start()
    log.info("status page listening on %s", PORT)

    # SIGTERM is how the platform asks the app to stop, on every redeploy. Closing the gateway
    # connection properly takes the bot offline at once and lets the old copy exit straight
    # away, instead of lingering until it is killed.
    stopping = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        with suppress(NotImplementedError):  # Windows has no loop signal handlers; Ctrl+C still works
            loop.add_signal_handler(sig, stopping.set)

    bot_task = None
    if TOKEN:
        bot_task = asyncio.create_task(run_bot())
    else:
        log.info(
            "DISCORD_TOKEN is not set, so this app is serving its setup page.\n"
            "On velixir: add DISCORD_TOKEN on the Environment tab, then redeploy.\n"
            "Locally: DISCORD_TOKEN=... python bot.py"
        )

    try:
        await stopping.wait()
    finally:
        log.info("shutting down")
        if bot_task:
            bot_task.cancel()
            await asyncio.gather(bot_task, return_exceptions=True)
        await runner.cleanup()


if __name__ == "__main__":
    with suppress(KeyboardInterrupt):
        asyncio.run(main())