Add anonymous multi-user vote command

alembic/versions/e1a2b3c4d5e6_add_anonymous_vote_tables.py

@@ -0,0 +1,76 @@
+"""Add anonymous vote tables
+
+Revision ID: e1a2b3c4d5e6
+Revises: d4f8c2a6e1b7
+Create Date: 2026-07-31 13:53:00.000000
+
+"""
+import sqlalchemy as sa
+from sqlalchemy.dialects import mysql
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "e1a2b3c4d5e6"
+down_revision = "d4f8c2a6e1b7"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "anonymous_vote_session",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("guild_id", mysql.BIGINT(display_width=18), nullable=False),
+        sa.Column("channel_id", mysql.BIGINT(display_width=18), nullable=False),
+        sa.Column("message_id", mysql.BIGINT(display_width=18), nullable=True),
+        sa.Column("topic", mysql.TEXT(), nullable=True),
+        sa.Column("created_by_id", mysql.BIGINT(display_width=18), nullable=False),
+        sa.Column("closes_at", mysql.TIMESTAMP(), nullable=False),
+        sa.Column("closed", sa.Boolean(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_table(
+        "anonymous_vote_candidate",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("session_id", sa.Integer(), nullable=False),
+        sa.Column("user_id", mysql.BIGINT(display_width=18), nullable=False),
+        sa.Column("display_name", mysql.TEXT(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["session_id"],
+            ["anonymous_vote_session.id"],
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_table(
+        "anonymous_vote_ballot",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("session_id", sa.Integer(), nullable=False),
+        sa.Column("candidate_id", sa.Integer(), nullable=False),
+        sa.Column("voter_id", mysql.BIGINT(display_width=18), nullable=False),
+        sa.Column("choice", sa.String(length=16), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["candidate_id"],
+            ["anonymous_vote_candidate.id"],
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["session_id"],
+            ["anonymous_vote_session.id"],
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "session_id",
+            "candidate_id",
+            "voter_id",
+            name="uq_anonymous_vote_ballot_session_candidate_voter",
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("anonymous_vote_ballot")
+    op.drop_table("anonymous_vote_candidate")
+    op.drop_table("anonymous_vote_session")

src/bot.py

@@ -95,13 +95,19 @@ async def on_ready(self) -> None:
 
     async def _register_persistent_views(self) -> None:
         """Re-register persistent UI views so buttons survive bot restarts."""
+        from src.views.anonymous_vote import register_anonymous_vote_views
         from src.views.bandecisionview import register_ban_views
 
         try:
             await register_ban_views(self)
         except Exception:
             logger.exception("Failed to register persistent ban decision views")
 
+        try:
+            await register_anonymous_vote_views(self)
+        except Exception:
+            logger.exception("Failed to register persistent anonymous vote views")
+
     async def on_application_command(self, ctx: ApplicationContext) -> None:
         """A global handler cog."""
         logger.debug(f"Command '{ctx.command}' received.")

src/cmds/core/admin.py

@@ -1,19 +1,48 @@
 """Admin command group for bot administration commands."""
 
 import logging
+import re
+from datetime import datetime
 
 import discord
 from discord import ApplicationContext, Interaction, Option, WebhookMessage
 from discord.ext import commands
 from discord.ext.commands import has_any_role
+from sqlalchemy import select
+from sqlalchemy.orm import selectinload
 
 from src.bot import Bot
 from src.core import settings
+from src.database.models import AnonymousVoteCandidate, AnonymousVoteSession
 from src.database.models.dynamic_role import RoleCategory
+from src.database.session import AsyncSessionLocal
+from src.helpers.duration import validate_duration
+from src.views.anonymous_vote import (
+    AnonymousVoteView,
+    build_poll_embed,
+    schedule_vote_close,
+)
 
 logger = logging.getLogger(__name__)
 
 CATEGORY_CHOICES = [c.value for c in RoleCategory]
+_MEMBER_TOKEN_RE = re.compile(r"<@!?(\d+)>|^(\d+)$")
+
+
+def _parse_member_ids(raw: str) -> list[int]:
+    """Parse space/comma-separated mentions or snowflake IDs into unique IDs."""
+    ids: list[int] = []
+    for part in re.split(r"[\s,]+", raw.strip()):
+        if not part:
+            continue
+        match = _MEMBER_TOKEN_RE.fullmatch(part)
+        if not match:
+            raise ValueError(
+                f"Could not parse `{part}`. Use mentions or numeric user IDs."
+            )
+        ids.append(int(match.group(1) or match.group(2)))
+    # Preserve order, drop duplicates
+    return list(dict.fromkeys(ids))
 
 
 class AdminCog(commands.Cog):
@@ -149,6 +178,109 @@ async def reload(self, ctx: ApplicationContext) -> Interaction | WebhookMessage:
         await self.bot.role_manager.reload()
         return await ctx.respond("Dynamic roles reloaded from database.", ephemeral=True)
 
+    @admin.command(
+        name="vote",
+        description="Start an anonymous timed vote on multiple members.",
+    )
+    @has_any_role(*settings.role_groups.get("VOTE_STARTERS"))
+    async def vote(
+        self,
+        ctx: ApplicationContext,
+        members: Option(
+            str,
+            "Nominees as mentions or user IDs (space/comma separated, max 25)",
+        ),
+        duration: Option(str, "How long the vote stays open (e.g. 12h, 1d, 30m)"),
+        topic: Option(str, "Optional topic shown on the poll", required=False),
+    ) -> Interaction | WebhookMessage:
+        """Start an anonymous vote; tallies reveal automatically when duration ends."""
+        closes_at_ts, error = validate_duration(duration)
+        if error:
+            return await ctx.respond(error, ephemeral=True)
+
+        try:
+            member_ids = _parse_member_ids(members)
+        except ValueError as exc:
+            return await ctx.respond(str(exc), ephemeral=True)
+
+        if not member_ids:
+            return await ctx.respond("Provide at least one nominee.", ephemeral=True)
+        if len(member_ids) > 25:
+            return await ctx.respond(
+                "Discord select menus support at most 25 nominees.",
+                ephemeral=True,
+            )
+
+        resolved: list[tuple[int, str]] = []
+        missing: list[str] = []
+        for user_id in member_ids:
+            member = ctx.guild.get_member(user_id)
+            if member is None:
+                try:
+                    member = await ctx.guild.fetch_member(user_id)
+                except discord.HTTPException:
+                    missing.append(str(user_id))
+                    continue
+            resolved.append((member.id, member.display_name))
+
+        if missing:
+            return await ctx.respond(
+                "Could not find member(s) in this server: " + ", ".join(f"`{m}`" for m in missing),
+                ephemeral=True,
+            )
+
+        closes_at = datetime.fromtimestamp(closes_at_ts)
+        await ctx.defer(ephemeral=True)
+
+        async with AsyncSessionLocal() as session:
+            vote_session = AnonymousVoteSession(
+                guild_id=ctx.guild.id,
+                channel_id=ctx.channel.id,
+                message_id=None,
+                topic=topic,
+                created_by_id=ctx.author.id,
+                closes_at=closes_at,
+                closed=False,
+            )
+            session.add(vote_session)
+            await session.flush()
+
+            for user_id, display_name in resolved:
+                session.add(
+                    AnonymousVoteCandidate(
+                        session_id=vote_session.id,
+                        user_id=user_id,
+                        display_name=display_name,
+                    )
+                )
+            await session.commit()
+
+            loaded = await session.scalar(
+                select(AnonymousVoteSession)
+                .where(AnonymousVoteSession.id == vote_session.id)
+                .options(selectinload(AnonymousVoteSession.candidates))
+            )
+            session_id = loaded.id
+            candidates = list(loaded.candidates)
+            poll_embed = build_poll_embed(loaded, candidates)
+
+        view = AnonymousVoteView(session_id, self.bot, candidates)
+        self.bot.add_view(view)
+        message = await ctx.channel.send(embed=poll_embed, view=view)
+
+        async with AsyncSessionLocal() as session:
+            vote_session = await session.get(AnonymousVoteSession, session_id)
+            if vote_session:
+                vote_session.message_id = message.id
+                await session.commit()
+
+        schedule_vote_close(self.bot, session_id, closes_at)
+        return await ctx.followup.send(
+            f"Anonymous vote #{session_id} started in {ctx.channel.mention}. "
+            f"Closes {discord.utils.format_dt(closes_at, style='R')}.",
+            ephemeral=True,
+        )
+
 
 def setup(bot: Bot) -> None:
     """Load the AdminCog."""

src/core/config.py

@@ -259,6 +259,19 @@ def role_groups(self) -> dict[str, list[int]]:
             ],
             "ALL_HTB_STAFF": [self.roles.HTB_STAFF],
             "ALL_HTB_SUPPORT": [self.roles.HTB_SUPPORT],
+            "VOTE_STARTERS": [
+                self.roles.ADMINISTRATOR,
+                self.roles.COMMUNITY_MANAGER,
+                self.roles.COMMUNITY_TEAM,
+            ],
+            "VOTE_CASTERS": [
+                self.roles.ADMINISTRATOR,
+                self.roles.COMMUNITY_MANAGER,
+                self.roles.COMMUNITY_TEAM,
+                self.roles.SR_MODERATOR,
+                self.roles.MODERATOR,
+                self.roles.JR_MODERATOR,
+            ],
         }
 
 

src/database/models/__init__.py

@@ -1,6 +1,7 @@
 # flake8: noqa
 from src.database.base_class import Base  # noqa
 
+from .anonymous_vote import AnonymousVoteBallot, AnonymousVoteCandidate, AnonymousVoteSession
 from .ban import Ban
 from .ctf import Ctf
 from .dynamic_role import DynamicRole, RoleCategory

src/database/models/anonymous_vote.py

@@ -0,0 +1,73 @@
+# flake8: noqa: D101
+from datetime import datetime
+
+from sqlalchemy import Boolean, ForeignKey, Integer, String, UniqueConstraint
+from sqlalchemy.dialects.mysql import BIGINT, TEXT, TIMESTAMP
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from . import Base
+
+
+class AnonymousVoteSession(Base):
+    """Timed anonymous vote session over one or more nominees."""
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    guild_id: Mapped[int] = mapped_column(BIGINT(18), nullable=False)
+    channel_id: Mapped[int] = mapped_column(BIGINT(18), nullable=False)
+    message_id: Mapped[int | None] = mapped_column(BIGINT(18), nullable=True)
+    topic: Mapped[str | None] = mapped_column(TEXT, nullable=True)
+    created_by_id: Mapped[int] = mapped_column(BIGINT(18), nullable=False)
+    closes_at: Mapped[datetime] = mapped_column(TIMESTAMP, nullable=False)
+    closed: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+
+    candidates: Mapped[list["AnonymousVoteCandidate"]] = relationship(
+        back_populates="session",
+        cascade="all, delete-orphan",
+    )
+    ballots: Mapped[list["AnonymousVoteBallot"]] = relationship(
+        back_populates="session",
+        cascade="all, delete-orphan",
+    )
+
+
+class AnonymousVoteCandidate(Base):
+    """A nominee in an anonymous vote session."""
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    session_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("anonymous_vote_session.id", ondelete="CASCADE"), nullable=False
+    )
+    user_id: Mapped[int] = mapped_column(BIGINT(18), nullable=False)
+    display_name: Mapped[str] = mapped_column(TEXT, nullable=False)
+
+    session: Mapped["AnonymousVoteSession"] = relationship(back_populates="candidates")
+    ballots: Mapped[list["AnonymousVoteBallot"]] = relationship(
+        back_populates="candidate",
+        cascade="all, delete-orphan",
+    )
+
+
+class AnonymousVoteBallot(Base):
+    """A single voter's choice for one nominee. voter_id is never shown in Discord."""
+
+    __table_args__ = (
+        UniqueConstraint(
+            "session_id",
+            "candidate_id",
+            "voter_id",
+            name="uq_anonymous_vote_ballot_session_candidate_voter",
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    session_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("anonymous_vote_session.id", ondelete="CASCADE"), nullable=False
+    )
+    candidate_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("anonymous_vote_candidate.id", ondelete="CASCADE"), nullable=False
+    )
+    voter_id: Mapped[int] = mapped_column(BIGINT(18), nullable=False)
+    choice: Mapped[str] = mapped_column(String(16), nullable=False)
+
+    session: Mapped["AnonymousVoteSession"] = relationship(back_populates="ballots")
+    candidate: Mapped["AnonymousVoteCandidate"] = relationship(back_populates="ballots")