bot.py

__version__ = "4.2.1"


import asyncio
import copy
import hashlib
import os
import re
import string
import struct
import sys
import platform
import typing
from datetime import datetime, timezone
from subprocess import PIPE
from types import SimpleNamespace

import discord
import isodate
from aiohttp import ClientSession, ClientResponseError
from discord.ext import commands, tasks
from discord.ext.commands.view import StringView
from emoji import is_emoji
from packaging.version import Version


try:
    # noinspection PyUnresolvedReferences
    from colorama import init

    init()
except ImportError:
    pass

from core import checks
from core.changelog import Changelog
from core.clients import ApiClient, MongoDBClient, PluginDatabaseClient
from core.config import ConfigManager
from core.models import (
    DMDisabled,
    HostingMethod,
    InvalidConfigError,
    PermissionLevel,
    SafeFormatter,
    configure_logging,
    getLogger,
)
from core.thread import ThreadManager
from core.time import human_timedelta
from core.utils import (
    extract_block_timestamp,
    normalize_alias,
    parse_alias,
    truncate,
    tryint,
    human_join,
    extract_forwarded_content,
)

logger = getLogger(__name__)

temp_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "temp")
if not os.path.exists(temp_dir):
    os.mkdir(temp_dir)

if sys.platform == "win32":
    try:
        asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
    except AttributeError:
        logger.error("Failed to use WindowsProactorEventLoopPolicy.", exc_info=True)


class ModmailBot(commands.Bot):
    def __init__(self):
        self.config = ConfigManager(self)
        self.config.populate_cache()

        intents = discord.Intents.all()
        if not self.config["enable_presence_intent"]:
            intents.presences = False

        super().__init__(command_prefix=None, intents=intents)  # implemented in `get_prefix`
        self.session = None
        self._api = None
        self.formatter = SafeFormatter()
        self.loaded_cogs = [
            "cogs.modmail",
            "cogs.plugins",
            "cogs.utility",
            "cogs.threadmenu",
        ]
        self._connected = None
        self.start_time = discord.utils.utcnow()
        self._started = False

        self.threads = ThreadManager(self)
        self._message_queues = {}  # User ID -> asyncio.Queue for message ordering

        log_dir = os.path.join(temp_dir, "logs")
        if not os.path.exists(log_dir):
            os.mkdir(log_dir)
        self.log_file_path = os.path.join(log_dir, "modmail.log")
        configure_logging(self)

        self.plugin_db = PluginDatabaseClient(self)  # Deprecated
        self.startup()

    def get_guild_icon(
        self,
        guild: typing.Optional[discord.Guild],
        *,
        size: typing.Optional[int] = None,
    ) -> str:
        if guild is None:
            guild = self.guild
        if guild.icon is None:
            return "https://cdn.discordapp.com/embed/avatars/0.png"
        if size is None:
            return guild.icon.url
        return guild.icon.with_size(size).url

    def _resolve_snippet(self, name: str) -> typing.Optional[str]:
        """
        Get actual snippet names from direct aliases to snippets.

        If the provided name is a snippet, it's returned unchanged.
        If there is an alias by this name, it is parsed to see if it
        refers only to a snippet, in which case that snippet name is
        returned.

        If no snippets were found, None is returned.
        """
        if name in self.snippets:
            return name

        try:
            (command,) = parse_alias(self.aliases[name])
        except (KeyError, ValueError):
            # There is either no alias by this name present or the
            # alias has multiple steps.
            pass
        else:
            if command in self.snippets:
                return command

    @property
    def uptime(self) -> str:
        now = discord.utils.utcnow()
        delta = now - self.start_time
        hours, remainder = divmod(int(delta.total_seconds()), 3600)
        minutes, seconds = divmod(remainder, 60)
        days, hours = divmod(hours, 24)

        fmt = "{h}h {m}m {s}s"
        if days:
            fmt = "{d}d " + fmt

        return self.formatter.format(fmt, d=days, h=hours, m=minutes, s=seconds)

    @property
    def hosting_method(self) -> HostingMethod:
        # use enums
        if ".heroku" in os.environ.get("PYTHONHOME", ""):
            return HostingMethod.HEROKU

        if os.environ.get("pm_id"):
            return HostingMethod.PM2

        if os.environ.get("INVOCATION_ID"):
            return HostingMethod.SYSTEMD

        if os.environ.get("USING_DOCKER"):
            return HostingMethod.DOCKER

        if os.environ.get("TERM"):
            return HostingMethod.SCREEN

        return HostingMethod.OTHER

    def startup(self):
        logger.line()
        logger.info("┌┬┐┌─┐┌┬┐┌┬┐┌─┐┬┬")
        logger.info("││││ │ │││││├─┤││")
        logger.info("┴ ┴└─┘─┴┘┴ ┴┴ ┴┴┴─┘")
        logger.info("v%s", __version__)
        logger.info("Authors: kyb3r, fourjr, Taaku18")
        logger.line()
        logger.info("discord.py: v%s", discord.__version__)
        logger.line()

    async def load_extensions(self):
        for cog in self.loaded_cogs:
            if cog in self.extensions:
                continue
            logger.debug("Loading %s.", cog)
            try:
                await self.load_extension(cog)
                logger.debug("Successfully loaded %s.", cog)
            except Exception:
                logger.exception("Failed to load %s.", cog)
        logger.line("debug")

    @property
    def version(self):
        return Version(__version__)

    @property
    def api(self) -> ApiClient:
        if self._api is None:
            if self.config["database_type"].lower() == "mongodb":
                self._api = MongoDBClient(self)
            else:
                logger.critical("Invalid database type.")
                raise RuntimeError
        return self._api

    @property
    def db(self):
        # deprecated
        return self.api.db

    async def get_prefix(self, message=None):
        return [self.prefix, f"<@{self.user.id}> ", f"<@!{self.user.id}> "]

    def run(self):
        async def runner():
            async with self:
                self._connected = asyncio.Event()
                self.session = ClientSession(loop=self.loop)

                if self.config["enable_presence_intent"]:
                    logger.info("Starting bot with presence intent.")
                else:
                    logger.info("Starting bot without presence intent.")

                try:
                    await self.start(self.token)
                except discord.PrivilegedIntentsRequired:
                    logger.critical(
                        "Privileged intents are not explicitly granted in the discord developers dashboard."
                    )
                except discord.LoginFailure:
                    logger.critical("Invalid token")
                except Exception:
                    logger.critical("Fatal exception", exc_info=True)
                finally:
                    if self.session:
                        await self.session.close()
                    if not self.is_closed():
                        await self.close()

        async def _cancel_tasks():
            async with self:
                task_retriever = asyncio.all_tasks
                loop = self.loop
                tasks = {t for t in task_retriever() if not t.done() and t.get_coro() != cancel_tasks_coro}

                if not tasks:
                    return

                logger.info("Cleaning up after %d tasks.", len(tasks))
                for task in tasks:
                    task.cancel()

                await asyncio.gather(*tasks, return_exceptions=True)
                logger.info("All tasks finished cancelling.")

                for task in tasks:
                    try:
                        if task.exception() is not None:
                            loop.call_exception_handler(
                                {
                                    "message": "Unhandled exception during Client.run shutdown.",
                                    "exception": task.exception(),
                                    "task": task,
                                }
                            )
                    except (asyncio.InvalidStateError, asyncio.CancelledError):
                        pass

        try:
            asyncio.run(runner(), debug=bool(os.getenv("DEBUG_ASYNCIO")))
        except (KeyboardInterrupt, SystemExit):
            logger.info("Received signal to terminate bot and event loop.")
        finally:
            logger.info("Cleaning up tasks.")

            try:
                cancel_tasks_coro = _cancel_tasks()
                asyncio.run(cancel_tasks_coro)
            finally:
                logger.info("Closing the event loop.")

    @property
    def bot_owner_ids(self):
        owner_ids = self.config["owners"]
        if owner_ids is not None:
            owner_ids = set(map(int, str(owner_ids).split(",")))
        if self.owner_id is not None:
            owner_ids.add(self.owner_id)
        permissions = self.config["level_permissions"].get(PermissionLevel.OWNER.name, [])
        for perm in permissions:
            owner_ids.add(int(perm))
        return owner_ids

    async def is_owner(self, user: discord.User) -> bool:
        if user.id in self.bot_owner_ids:
            return True
        return await super().is_owner(user)

    @property
    def log_channel(self) -> typing.Optional[discord.TextChannel]:
        channel_id = self.config["log_channel_id"]
        if channel_id is not None:
            try:
                channel = self.get_channel(int(channel_id))
                if channel is not None:
                    return channel
            except ValueError:
                pass
            logger.debug("LOG_CHANNEL_ID was invalid, removed.")
            self.config.remove("log_channel_id")
        if self.main_category is not None:
            try:
                channel = self.main_category.channels[0]
                self.config["log_channel_id"] = channel.id
                logger.warning(
                    "No log channel set, setting #%s to be the log channel.",
                    channel.name,
                )
                return channel
            except IndexError:
                pass
        logger.warning(
            "No log channel set, set one with `%ssetup` or `%sconfig set log_channel_id <id>`.",
            self.prefix,
            self.prefix,
        )
        return None

    @property
    def mention_channel(self):
        channel_id = self.config["mention_channel_id"]
        if channel_id is not None:
            try:
                channel = self.get_channel(int(channel_id))
                if channel is not None:
                    return channel
            except ValueError:
                pass
            logger.debug("MENTION_CHANNEL_ID was invalid, removed.")
            self.config.remove("mention_channel_id")

        return self.log_channel

    @property
    def update_channel(self):
        channel_id = self.config["update_channel_id"]
        if channel_id is not None:
            try:
                channel = self.get_channel(int(channel_id))
                if channel is not None:
                    return channel
            except ValueError:
                pass
            logger.debug("UPDATE_CHANNEL_ID was invalid, removed.")
            self.config.remove("update_channel_id")

        return self.log_channel

    async def wait_for_connected(self) -> None:
        await self.wait_until_ready()
        await self._connected.wait()
        await self.config.wait_until_ready()

    @property
    def snippets(self) -> typing.Dict[str, str]:
        return self.config["snippets"]

    @property
    def aliases(self) -> typing.Dict[str, str]:
        return self.config["aliases"]

    @property
    def auto_triggers(self) -> typing.Dict[str, str]:
        return self.config["auto_triggers"]

    @property
    def token(self) -> str:
        token = self.config["token"]
        if token is None:
            logger.critical("TOKEN must be set, set this as bot token found on the Discord Developer Portal.")
            sys.exit(0)
        return token

    @property
    def guild_id(self) -> typing.Optional[int]:
        guild_id = self.config["guild_id"]
        if guild_id is not None:
            try:
                return int(str(guild_id))
            except ValueError:
                self.config.remove("guild_id")
                logger.critical("Invalid GUILD_ID set.")
        else:
            logger.debug("No GUILD_ID set.")
        return None

    @property
    def guild(self) -> typing.Optional[discord.Guild]:
        """
        The guild that the bot is serving
        (the server where users message it from)
        """
        return discord.utils.get(self.guilds, id=self.guild_id)

    @property
    def modmail_guild(self) -> typing.Optional[discord.Guild]:
        """
        The guild that the bot is operating in
        (where the bot is creating threads)
        """
        modmail_guild_id = self.config["modmail_guild_id"]
        if modmail_guild_id is None:
            return self.guild
        try:
            guild = discord.utils.get(self.guilds, id=int(modmail_guild_id))
            if guild is not None:
                return guild
        except ValueError:
            pass
        self.config.remove("modmail_guild_id")
        logger.critical("Invalid MODMAIL_GUILD_ID set.")
        return self.guild

    @property
    def using_multiple_server_setup(self) -> bool:
        return self.modmail_guild != self.guild

    @property
    def main_category(self) -> typing.Optional[discord.CategoryChannel]:
        if self.modmail_guild is not None:
            category_id = self.config["main_category_id"]
            if category_id is not None:
                try:
                    cat = discord.utils.get(self.modmail_guild.categories, id=int(category_id))
                    if cat is not None:
                        return cat
                except ValueError:
                    pass
                self.config.remove("main_category_id")
                logger.debug("MAIN_CATEGORY_ID was invalid, removed.")
            cat = discord.utils.get(self.modmail_guild.categories, name="Modmail")
            if cat is not None:
                self.config["main_category_id"] = cat.id
                logger.debug(
                    'No main category set explicitly, setting category "Modmail" as the main category.'
                )
                return cat
        return None

    @property
    def blocked_users(self) -> typing.Dict[str, str]:
        return self.config["blocked"]

    @property
    def blocked_roles(self) -> typing.Dict[str, str]:
        return self.config["blocked_roles"]

    @property
    def blocked_whitelisted_users(self) -> typing.List[str]:
        return self.config["blocked_whitelist"]

    @property
    def prefix(self) -> str:
        return str(self.config["prefix"])

    @property
    def mod_color(self) -> int:
        return self.config.get("mod_color")

    @property
    def recipient_color(self) -> int:
        return self.config.get("recipient_color")

    @property
    def main_color(self) -> int:
        return self.config.get("main_color")

    @property
    def error_color(self) -> int:
        return self.config.get("error_color")

    def command_perm(self, command_name: str) -> PermissionLevel:
        level = self.config["override_command_level"].get(command_name)
        if level is not None:
            try:
                return PermissionLevel[level.upper()]
            except KeyError:
                logger.warning("Invalid override_command_level for command %s.", command_name)
                self.config["override_command_level"].pop(command_name)

        command = self.get_command(command_name)
        if command is None:
            logger.debug("Command %s not found.", command_name)
            return PermissionLevel.INVALID
        level = next(
            (check.permission_level for check in command.checks if hasattr(check, "permission_level")),
            None,
        )
        if level is None:
            logger.debug("Command %s does not have a permission level.", command_name)
            return PermissionLevel.INVALID
        return level

    async def on_connect(self):
        try:
            await self.api.validate_database_connection()
        except Exception:
            logger.debug("Logging out due to failed database connection.")
            return await self.close()

        logger.debug("Connected to gateway.")
        await self.config.refresh()
        await self.api.setup_indexes()
        await self.load_extensions()
        self._connected.set()

    async def on_ready(self):
        """Bot startup, sets uptime."""

        # Wait until config cache is populated with stuff from db and on_connect ran
        await self.wait_for_connected()

        if self.guild is None:
            logger.error("Logging out due to invalid GUILD_ID.")
            return await self.close()

        if self._started:
            # Bot has started before
            logger.line()
            logger.warning("Bot restarted due to internal discord reloading.")
            logger.line()
            return

        logger.line()
        logger.debug("Client ready.")
        logger.info("Logged in as: %s", self.user)
        logger.info("Bot ID: %s", self.user.id)
        owners = ", ".join(
            getattr(self.get_user(owner_id), "name", str(owner_id)) for owner_id in self.bot_owner_ids
        )
        logger.info("Owners: %s", owners)
        logger.info("Prefix: %s", self.prefix)
        logger.info("Guild Name: %s", self.guild.name)
        logger.info("Guild ID: %s", self.guild.id)
        if self.using_multiple_server_setup:
            logger.info("Receiving guild ID: %s", self.modmail_guild.id)
        logger.line()

        if "dev" in __version__:
            logger.warning(
                "You are running a developmental version. This should not be used in production. (v%s)",
                __version__,
            )
            logger.line()

        await self.threads.populate_cache()

        # closures
        closures = self.config["closures"]
        logger.info("There are %d thread(s) pending to be closed.", len(closures))
        logger.line()

        for recipient_id, items in tuple(closures.items()):
            after = (
                datetime.fromisoformat(items["time"]).astimezone(timezone.utc) - discord.utils.utcnow()
            ).total_seconds()
            if after <= 0:
                logger.debug("Closing thread for recipient %s.", recipient_id)
                after = 0
            else:
                logger.debug(
                    "Thread for recipient %s will be closed after %s seconds.",
                    recipient_id,
                    after,
                )

            thread = await self.threads.find(recipient_id=int(recipient_id))

            if not thread:
                # If the channel is deleted
                logger.debug("Failed to close thread for recipient %s.", recipient_id)
                self.config["closures"].pop(recipient_id)
                await self.config.update()
                continue

            await thread.close(
                closer=await self.get_or_fetch_user(items["closer_id"]),
                after=after,
                silent=items["silent"],
                delete_channel=items["delete_channel"],
                message=items["message"],
                auto_close=items.get("auto_close", False),
            )

        for log in await self.api.get_open_logs():
            if log.get("channel_id") is None or self.get_channel(int(log["channel_id"])) is None:
                logger.debug("Unable to resolve thread with channel %s.", log["channel_id"])
                log_data = await self.api.post_log(
                    log["channel_id"],
                    {
                        "open": False,
                        "title": None,
                        "closed_at": str(discord.utils.utcnow()),
                        "close_message": "Channel has been deleted, no closer found.",
                        "closer": {
                            "id": str(self.user.id),
                            "name": self.user.name,
                            "discriminator": self.user.discriminator,
                            "avatar_url": self.user.display_avatar.url,
                            "mod": True,
                        },
                    },
                )
                if log_data:
                    logger.debug("Successfully closed thread with channel %s.", log["channel_id"])
                else:
                    logger.debug(
                        "Failed to close thread with channel %s, skipping.",
                        log["channel_id"],
                    )

        other_guilds = [guild for guild in self.guilds if guild not in {self.guild, self.modmail_guild}]
        if any(other_guilds):
            logger.warning(
                "The bot is in more servers other than the main and staff server. "
                "This may cause data compromise (%s).",
                ", ".join(str(guild.name) for guild in other_guilds),
            )
            logger.warning("If the external servers are valid, you may ignore this message.")

        self.post_metadata.start()
        self.autoupdate.start()
        self.log_expiry.start()
        self._started = True

    async def convert_emoji(self, name: str) -> str:
        ctx = SimpleNamespace(bot=self, guild=self.modmail_guild)
        converter = commands.EmojiConverter()

        if not is_emoji(name):
            try:
                name = await converter.convert(ctx, name.strip(":"))
            except commands.BadArgument as e:
                logger.warning("%s is not a valid emoji: %s", name, e)
                raise
        return name

    async def get_or_fetch_user(self, id: int) -> discord.User:
        """
        Retrieve a User based on their ID.

        This tries getting the user from the cache and falls back to making
        an API call if they're not found in the cache.
        """
        return self.get_user(id) or await self.fetch_user(id)

    @staticmethod
    async def get_or_fetch_member(guild: discord.Guild, member_id: int) -> typing.Optional[discord.Member]:
        """
        Attempt to get a member from cache; on failure fetch from the API.

        Returns:
            The :obj:`discord.Member` or :obj:`None` to indicate the member could not be found.
        """
        return guild.get_member(member_id) or await guild.fetch_member(member_id)

    async def retrieve_emoji(self) -> typing.Tuple[str, str]:
        sent_emoji = self.config["sent_emoji"]
        blocked_emoji = self.config["blocked_emoji"]

        if sent_emoji != "disable":
            try:
                sent_emoji = await self.convert_emoji(sent_emoji)
            except commands.BadArgument:
                logger.warning("Removed sent emoji (%s).", sent_emoji)
                sent_emoji = self.config.remove("sent_emoji")
                await self.config.update()

        if blocked_emoji != "disable":
            try:
                blocked_emoji = await self.convert_emoji(blocked_emoji)
            except commands.BadArgument:
                logger.warning("Removed blocked emoji (%s).", blocked_emoji)
                blocked_emoji = self.config.remove("blocked_emoji")
                await self.config.update()

        return sent_emoji, blocked_emoji

    def check_account_age(self, author: discord.Member) -> bool:
        account_age = self.config.get("account_age")
        now = discord.utils.utcnow()

        try:
            min_account_age = author.created_at + account_age
        except ValueError:
            logger.warning("Error with 'account_age'.", exc_info=True)
            min_account_age = author.created_at + self.config.remove("account_age")

        if min_account_age > now:
            # User account has not reached the required time
            delta = human_timedelta(min_account_age)
            logger.debug("Blocked due to account age, user %s.", author.name)

            if str(author.id) not in self.blocked_users:
                new_reason = f"System Message: New Account. User can try again {delta}."
                self.blocked_users[str(author.id)] = new_reason

            return False
        return True

    def check_guild_age(self, author: discord.Member) -> bool:
        guild_age = self.config.get("guild_age")
        now = discord.utils.utcnow()

        if not hasattr(author, "joined_at"):
            logger.warning("Not in guild, cannot verify guild_age, %s.", author.name)
            return True

        try:
            min_guild_age = author.joined_at + guild_age
        except ValueError:
            logger.warning("Error with 'guild_age'.", exc_info=True)
            min_guild_age = author.joined_at + self.config.remove("guild_age")

        if min_guild_age > now:
            # User has not stayed in the guild for long enough
            delta = human_timedelta(min_guild_age)
            logger.debug("Blocked due to guild age, user %s.", author.name)

            if str(author.id) not in self.blocked_users:
                new_reason = f"System Message: Recently Joined. User can try again {delta}."
                self.blocked_users[str(author.id)] = new_reason

            return False
        return True

    def check_manual_blocked_roles(self, author: discord.Member) -> bool:
        if isinstance(author, discord.Member):
            for r in author.roles:
                if str(r.id) in self.blocked_roles:
                    blocked_reason = self.blocked_roles.get(str(r.id)) or ""

                    try:
                        end_time, after = extract_block_timestamp(blocked_reason, author.id)
                    except ValueError:
                        return False

                    if end_time is not None:
                        if after <= 0:
                            # No longer blocked
                            self.blocked_roles.pop(str(r.id))
                            logger.debug("No longer blocked, role %s.", r.name)
                            return True
                    logger.debug("User blocked, role %s.", r.name)
                    return False

        return True

    def check_manual_blocked(self, author: discord.Member) -> bool:
        if str(author.id) not in self.blocked_users:
            return True

        blocked_reason = self.blocked_users.get(str(author.id)) or ""

        if blocked_reason.startswith("System Message:"):
            # Met the limits already, otherwise it would've been caught by the previous checks
            logger.debug("No longer internally blocked, user %s.", author.name)
            self.blocked_users.pop(str(author.id))
            return True

        try:
            end_time, after = extract_block_timestamp(blocked_reason, author.id)
        except ValueError:
            return False

        if end_time is not None:
            if after <= 0:
                # No longer blocked
                self.blocked_users.pop(str(author.id))
                logger.debug("No longer blocked, user %s.", author.name)
                return True
        logger.debug("User blocked, user %s.", author.name)
        return False

    async def _process_blocked(self, message):
        _, blocked_emoji = await self.retrieve_emoji()
        if await self.is_blocked(message.author, channel=message.channel, send_message=True):
            await self.add_reaction(message, blocked_emoji)
            return True
        return False

    async def is_blocked(
        self,
        author: discord.User,
        *,
        channel: discord.TextChannel = None,
        send_message: bool = False,
    ) -> bool:
        member = self.guild.get_member(author.id)
        if member is None:
            # try to find in other guilds
            for g in self.guilds:
                member = g.get_member(author.id)
                if member:
                    break

            if member is None:
                logger.debug("User not in guild, %s.", author.id)

        if member is not None:
            author = member

        if str(author.id) in self.blocked_whitelisted_users:
            if str(author.id) in self.blocked_users:
                self.blocked_users.pop(str(author.id))
                await self.config.update()
            return False

        blocked_reason = self.blocked_users.get(str(author.id)) or ""

        if not self.check_account_age(author) or not self.check_guild_age(author):
            new_reason = self.blocked_users.get(str(author.id))
            if new_reason != blocked_reason:
                if send_message:
                    await channel.send(
                        embed=discord.Embed(
                            title="Message not sent!",
                            description=new_reason,
                            color=self.error_color,
                        )
                    )
            return True

        if not self.check_manual_blocked(author):
            return True

        if not self.check_manual_blocked_roles(author):
            return True

        await self.config.update()
        return False

    async def get_thread_cooldown(self, author: discord.Member):
        thread_cooldown = self.config.get("thread_cooldown")
        now = discord.utils.utcnow()

        if thread_cooldown == isodate.Duration():
            return

        last_log = await self.api.get_latest_user_logs(author.id)

        if last_log is None:
            logger.debug("Last thread wasn't found, %s.", author.name)
            return

        last_log_closed_at = last_log.get("closed_at")

        if not last_log_closed_at:
            logger.debug("Last thread was not closed, %s.", author.name)
            return

        try:
            cooldown = datetime.fromisoformat(last_log_closed_at).astimezone(timezone.utc) + thread_cooldown
        except ValueError:
            logger.warning("Error with 'thread_cooldown'.", exc_info=True)
            cooldown = datetime.fromisoformat(last_log_closed_at).astimezone(
                timezone.utc
            ) + self.config.remove("thread_cooldown")

        if cooldown > now:
            # User messaged before thread cooldown ended
            delta = human_timedelta(cooldown)
            logger.debug("Blocked due to thread cooldown, user %s.", author.name)
            return delta
        return

    @staticmethod
    async def add_reaction(
        msg,
        reaction: typing.Union[discord.Emoji, discord.Reaction, discord.PartialEmoji, str],
    ) -> bool:
        if reaction != "disable":
            try:
                await msg.add_reaction(reaction)
            except (discord.HTTPException, TypeError) as e:
                logger.warning("Failed to add reaction %s: %s.", reaction, e)
                return False
        return True

    async def _queue_dm_message(self, message: discord.Message) -> None:
        """Queue DM messages to ensure they're processed in order per user."""
        user_id = message.author.id

        if user_id not in self._message_queues:
            self._message_queues[user_id] = asyncio.Queue()
            # Start processing task for this user
            self.loop.create_task(self._process_user_messages(user_id))

        await self._message_queues[user_id].put(message)

    async def _process_user_messages(self, user_id: int) -> None:
        """Process messages for a specific user in order."""
        queue = self._message_queues[user_id]

        while True:
            try:
                # Wait for a message with timeout to clean up inactive queues
                message = await asyncio.wait_for(queue.get(), timeout=300)  # 5 minutes
                await self.process_dm_modmail(message)
                queue.task_done()
            except asyncio.TimeoutError:
                # Clean up inactive queue
                if queue.empty():
                    self._message_queues.pop(user_id, None)
                    break
            except Exception as e:
                logger.error(f"Error processing message for user {user_id}: {e}", exc_info=True)
                queue.task_done()

    async def process_dm_modmail(self, message: discord.Message) -> None:
        """Processes messages sent to the bot."""
        blocked = await self._process_blocked(message)
        if blocked:
            return
        sent_emoji, blocked_emoji = await self.retrieve_emoji()

        # Handle forwarded messages (Discord forwards)
        # See: https://discord.com/developers/docs/resources/message#message-reference-content-attribution-forwards
        # 1. Multi-forward (message_snapshots)
        if hasattr(message, "flags") and getattr(message.flags, "has_snapshot", False):
            if hasattr(message, "message_snapshots") and message.message_snapshots:
                thread = await self.threads.find(recipient=message.author)
                if thread is None:
                    delta = await self.get_thread_cooldown(message.author)
                    if delta:
                        await message.channel.send(
                            embed=discord.Embed(
                                title=self.config["cooldown_thread_title"],
                                description=self.config["cooldown_thread_response"].format(delta=delta),
                                color=self.error_color,
                            )
                        )
                        return
                    if self.config["dm_disabled"] in (DMDisabled.NEW_THREADS, DMDisabled.ALL_THREADS):
                        embed = discord.Embed(
                            title=self.config["disabled_new_thread_title"],
                            color=self.error_color,
                            description=self.config["disabled_new_thread_response"],
                        )
                        embed.set_footer(
                            text=self.config["disabled_new_thread_footer"],
                            icon_url=self.get_guild_icon(guild=message.guild, size=128),
                        )
                        logger.info(
                            "A new thread was blocked from %s due to disabled Modmail.", message.author
                        )
                        await self.add_reaction(message, blocked_emoji)
                        return await message.channel.send(embed=embed)
                    thread = await self.threads.create(message.author, message=message)
                else:
                    if self.config["dm_disabled"] == DMDisabled.ALL_THREADS:
                        embed = discord.Embed(
                            title=self.config["disabled_current_thread_title"],
                            color=self.error_color,
                            description=self.config["disabled_current_thread_response"],
                        )
                        embed.set_footer(
                            text=self.config["disabled_current_thread_footer"],
                            icon_url=self.get_guild_icon(guild=message.guild, size=128),
                        )
                        logger.info("A message was blocked from %s due to disabled Modmail.", message.author)
                        await self.add_reaction(message, blocked_emoji)
                        return await message.channel.send(embed=embed)
                # Extract forwarded content using utility function
                combined_content = extract_forwarded_content(message) or "[Forwarded message with no content]"

                class ForwardedMessage:
                    def __init__(self, original_message, forwarded_content):
                        self.author = original_message.author
                        self.content = forwarded_content
                        self.attachments = []
                        self.stickers = []
                        self.created_at = original_message.created_at
                        self.embeds = []
                        self.id = original_message.id
                        self.flags = original_message.flags
                        self.message_snapshots = original_message.message_snapshots
                        self.type = getattr(original_message, "type", None)

                forwarded_msg = ForwardedMessage(message, combined_content)
                await thread.send(forwarded_msg)
                await self.add_reaction(message, sent_emoji)
                self.dispatch("thread_reply", thread, False, message, False, False)
                return
            else:
                message.content = "[Forwarded message with no content]"
        # 2. Single-message forward (MessageType.forward)
        elif getattr(message, "type", None) == getattr(discord.MessageType, "forward", None):
            # Check for message.reference and its type
            ref = getattr(message, "reference", None)
            if ref and getattr(ref, "type", None) == getattr(discord, "MessageReferenceType", None).forward:
                # Try to fetch the referenced message
                ref_msg = None
                try:
                    if ref.resolved:
                        ref_msg = ref.resolved
                    elif ref.message_id and ref.channel_id:
                        channel = self.get_channel(ref.channel_id) or (
                            await self.fetch_channel(ref.channel_id)
                        )
                        ref_msg = await channel.fetch_message(ref.message_id)
                except Exception:
                    ref_msg = None
                if ref_msg:
                    # Forward the referenced message as if it was sent
                    thread = await self.threads.find(recipient=message.author)
                    if thread is None:
                        delta = await self.get_thread_cooldown(message.author)
                        if delta:
                            await message.channel.send(
                                embed=discord.Embed(
                                    title=self.config["cooldown_thread_title"],
                                    description=self.config["cooldown_thread_response"].format(delta=delta),
                                    color=self.error_color,
                                )
                            )
                            return
                        if self.config["dm_disabled"] in (DMDisabled.NEW_THREADS, DMDisabled.ALL_THREADS):
                            embed = discord.Embed(
                                title=self.config["disabled_new_thread_title"],
                                color=self.error_color,
                                description=self.config["disabled_new_thread_response"],
                            )
                            embed.set_footer(
                                text=self.config["disabled_new_thread_footer"],
                                icon_url=self.get_guild_icon(guild=message.guild, size=128),
                            )
                            logger.info(
                                "A new thread was blocked from %s due to disabled Modmail.", message.author
                            )
                            await self.add_reaction(message, blocked_emoji)
                            return await message.channel.send(embed=embed)
                        thread = await self.threads.create(message.author, message=message)
                    else:
                        if self.config["dm_disabled"] == DMDisabled.ALL_THREADS:
                            embed = discord.Embed(
                                title=self.config["disabled_current_thread_title"],
                                color=self.error_color,
                                description=self.config["disabled_current_thread_response"],
                            )
                            embed.set_footer(
                                text=self.config["disabled_current_thread_footer"],
                                icon_url=self.get_guild_icon(guild=message.guild, size=128),
                            )
                            logger.info(
                                "A message was blocked from %s due to disabled Modmail.", message.author
                            )
                            await self.add_reaction(message, blocked_emoji)
                            return await message.channel.send(embed=embed)

                    # Create a forwarded message wrapper to preserve forward info
                    class ForwardedMessage:
                        def __init__(self, original_message, ref_message):
                            self.author = original_message.author
                            # Use the utility function to extract content or fallback to ref message content
                            extracted_content = extract_forwarded_content(original_message)
                            self.content = (
                                extracted_content
                                or ref_message.content
                                or "[Forwarded message with no text content]"
                            )
                            self.attachments = getattr(ref_message, "attachments", [])
                            self.stickers = getattr(ref_message, "stickers", [])
                            self.created_at = original_message.created_at
                            self.embeds = getattr(ref_message, "embeds", [])
                            self.id = original_message.id
                            self.type = getattr(original_message, "type", None)
                            self.reference = original_message.reference

                    forwarded_msg = ForwardedMessage(message, ref_msg)
                    await thread.send(forwarded_msg)
                    await self.add_reaction(message, sent_emoji)
                    self.dispatch("thread_reply", thread, False, message, False, False)
                    return
                else:
                    message.content = "[Forwarded message with no content]"

        if message.type not in [discord.MessageType.default, discord.MessageType.reply]:
            return

        thread = await self.threads.find(recipient=message.author)
        if thread and thread.snoozed:
            await thread.restore_from_snooze()
            self.threads.cache[thread.id] = thread
            # No need to re-fetch the thread - it's already restored and cached properly

        # If the previous thread was closed with delete_channel=True the channel object
        # stored on the thread will now be invalid (deleted). In some rare race cases
        # the thread can still be returned from the cache (or reconstructed) while the
        # channel lookup returns None, causing downstream relay attempts to raise
        # discord.NotFound ("Channel not found when trying to send message."). Treat
        # this situation as "no active thread" so the user's new DM starts a fresh
        # thread instead of silently failing.
        try:
            if (
                thread
                and thread.channel
                and isinstance(thread.channel, discord.TextChannel)
                and self.get_channel(getattr(thread.channel, "id", None)) is None
            ):
                logger.info(
                    "Stale thread detected for %s (channel deleted). Purging cache entry and creating new thread.",
                    message.author,
                )
                # Best-effort removal; ignore if already gone.
                self.threads.cache.pop(thread.id, None)
                thread = None
        except Exception:
            # If any attribute access fails, fall back to treating it as closed.
            self.threads.cache.pop(getattr(thread, "id", None), None)
            thread = None

        if thread is None:
            delta = await self.get_thread_cooldown(message.author)
            if delta:
                await message.channel.send(
                    embed=discord.Embed(
                        title=self.config["cooldown_thread_title"],
                        description=self.config["cooldown_thread_response"].format(delta=delta),
                        color=self.error_color,
                    )
                )
                return

            if self.config["dm_disabled"] in (
                DMDisabled.NEW_THREADS,
                DMDisabled.ALL_THREADS,
            ):
                embed = discord.Embed(
                    title=self.config["disabled_new_thread_title"],
                    color=self.error_color,
                    description=self.config["disabled_new_thread_response"],
                )
                embed.set_footer(
                    text=self.config["disabled_new_thread_footer"],
                    icon_url=self.get_guild_icon(guild=message.guild, size=128),
                )
                logger.info(
                    "A new thread was blocked from %s due to disabled Modmail.",
                    message.author,
                )
                await self.add_reaction(message, blocked_emoji)
                return await message.channel.send(embed=embed)

            thread = await self.threads.create(message.author, message=message)
            # If thread menu is enabled, thread creation is deferred until user selects an option.
            if getattr(thread, "_pending_menu", False):
                return
        else:
            if self.config["dm_disabled"] == DMDisabled.ALL_THREADS:
                embed = discord.Embed(
                    title=self.config["disabled_current_thread_title"],
                    color=self.error_color,
                    description=self.config["disabled_current_thread_response"],
                )
                embed.set_footer(
                    text=self.config["disabled_current_thread_footer"],
                    icon_url=self.get_guild_icon(guild=message.guild, size=128),
                )
                logger.info(
                    "A message was blocked from %s due to disabled Modmail.",
                    message.author,
                )
                await self.add_reaction(message, blocked_emoji)
                return await message.channel.send(embed=embed)

        if not thread.cancelled:
            try:
                await thread.send(message)
            except Exception:
                logger.error("Failed to send message:", exc_info=True)
                await self.add_reaction(message, blocked_emoji)

                try:
                    # Re-check channel existence
                    if thread and thread.channel and isinstance(thread.channel, discord.TextChannel):
                        if self.get_channel(thread.channel.id) is None:
                            logger.info(
                                "Relay failed due to deleted channel for %s; creating new thread.",
                                message.author,
                            )
                            self.threads.cache.pop(thread.id, None)
                            new_thread = await self.threads.create(message.author, message=message)
                            if not getattr(new_thread, "_pending_menu", False) and not new_thread.cancelled:
                                try:
                                    await new_thread.send(message)
                                except Exception:
                                    logger.error(
                                        "Failed to relay message after creating new thread:",
                                        exc_info=True,
                                    )
                                else:
                                    for user in new_thread.recipients:
                                        if user != message.author:
                                            try:
                                                await new_thread.send(message, user)
                                            except Exception:
                                                logger.error(
                                                    "Failed to send message to additional recipient:",
                                                    exc_info=True,
                                                )
                                    await self.add_reaction(message, sent_emoji)
                                    self.dispatch(
                                        "thread_reply",
                                        new_thread,
                                        False,
                                        message,
                                        False,
                                        False,
                                    )
                except Exception:
                    logger.warning(
                        "Unexpected failure in DM relay/new-thread follow-up block.",
                        exc_info=True,
                    )
            else:
                for user in thread.recipients:
                    # send to all other recipients
                    if user != message.author:
                        try:
                            await thread.send(message, user)
                        except Exception:
                            # silently ignore
                            logger.error("Failed to send message:", exc_info=True)

                await self.add_reaction(message, sent_emoji)
                self.dispatch("thread_reply", thread, False, message, False, False)

    def _get_snippet_command(self) -> commands.Command:
        """Get the correct reply command based on the snippet config"""
        modifiers = "f"
        if self.config["plain_snippets"]:
            modifiers += "p"
        if self.config["anonymous_snippets"]:
            modifiers += "a"

        return self.get_command(f"{modifiers}reply")

    async def get_contexts(self, message, *, cls=commands.Context):
        """
        Returns all invocation contexts from the message.
        Supports getting the prefix from database as well as command aliases.
        """

        view = StringView(message.content)
        ctx = cls(prefix=self.prefix, view=view, bot=self, message=message)
        thread = await self.threads.find(channel=ctx.channel)

        if message.author.id == self.user.id:  # type: ignore
            return [ctx]

        prefixes = await self.get_prefix()

        invoked_prefix = discord.utils.find(view.skip_string, prefixes)
        if invoked_prefix is None:
            return [ctx]

        invoker = view.get_word().lower()

        # Check if a snippet is being called.
        # This needs to be done before checking for aliases since
        # snippets can have multiple words.
        try:
            # Use removeprefix once PY3.9+
            snippet_text = self.snippets[message.content[len(invoked_prefix) :]]
        except KeyError:
            snippet_text = None

        # Check if there is any aliases being called.
        alias = self.aliases.get(invoker)
        if alias is not None and snippet_text is None:
            ctxs = []
            aliases = normalize_alias(alias, message.content[len(f"{invoked_prefix}{invoker}") :])
            if not aliases:
                logger.warning("Alias %s is invalid, removing.", invoker)
                self.aliases.pop(invoker)

            for alias in aliases:
                command = None
                try:
                    snippet_text = self.snippets[alias]
                except KeyError:
                    command_invocation_text = alias
                else:
                    command = self._get_snippet_command()
                    command_invocation_text = f"{invoked_prefix}{command} {snippet_text}"
                view = StringView(invoked_prefix + command_invocation_text)
                ctx_ = cls(prefix=self.prefix, view=view, bot=self, message=message)
                ctx_.thread = thread
                discord.utils.find(view.skip_string, prefixes)
                ctx_.invoked_with = view.get_word().lower()
                ctx_.command = command or self.all_commands.get(ctx_.invoked_with)
                ctxs += [ctx_]
            return ctxs

        ctx.thread = thread

        if snippet_text is not None:
            # Process snippets
            ctx.command = self._get_snippet_command()
            reply_view = StringView(f"{invoked_prefix}{ctx.command} {snippet_text}")
            discord.utils.find(reply_view.skip_string, prefixes)
            ctx.invoked_with = reply_view.get_word().lower()
            ctx.view = reply_view
        else:
            ctx.command = self.all_commands.get(invoker)
            ctx.invoked_with = invoker

        return [ctx]

    async def trigger_auto_triggers(self, message, channel, *, cls=commands.Context):
        message.author = self.modmail_guild.me
        message.channel = channel
        message.guild = channel.guild

        view = StringView(message.content)
        ctx = cls(prefix=self.prefix, view=view, bot=self, message=message)
        thread = await self.threads.find(channel=ctx.channel)

        invoked_prefix = self.prefix
        invoker = None

        if self.config.get("use_regex_autotrigger"):
            trigger = next(filter(lambda x: re.search(x, message.content), self.auto_triggers.keys()))
            if trigger:
                invoker = re.search(trigger, message.content).group(0)
        else:
            trigger = next(
                filter(
                    lambda x: x.lower() in message.content.lower(),
                    self.auto_triggers.keys(),
                )
            )
            if trigger:
                invoker = trigger.lower()

        alias = self.auto_triggers[trigger]

        ctxs = []

        if alias is not None:
            ctxs = []
            aliases = normalize_alias(alias)
            if not aliases:
                logger.warning("Alias %s is invalid as called in autotrigger.", invoker)

        message.author = thread.recipient  # Allow for get_contexts to work

        for alias in aliases:
            message.content = invoked_prefix + alias
            ctxs += await self.get_contexts(message)

        message.author = self.modmail_guild.me  # Fix message so commands execute properly

        for ctx in ctxs:
            if ctx.command:
                old_checks = copy.copy(ctx.command.checks)
                ctx.command.checks = [checks.has_permissions(PermissionLevel.INVALID)]

                await self.invoke(ctx)

                ctx.command.checks = old_checks
                continue

    async def get_context(self, message, *, cls=commands.Context):
        """
        Returns the invocation context from the message.
        Supports getting the prefix from database.
        """

        view = StringView(message.content)
        ctx = cls(prefix=self.prefix, view=view, bot=self, message=message)

        if message.author.id == self.user.id:
            return ctx

        ctx.thread = await self.threads.find(channel=ctx.channel)

        prefixes = await self.get_prefix()

        invoked_prefix = discord.utils.find(view.skip_string, prefixes)
        if invoked_prefix is None:
            return ctx

        invoker = view.get_word().lower()

        ctx.invoked_with = invoker
        ctx.command = self.all_commands.get(invoker)

        return ctx

    async def update_perms(
        self, name: typing.Union[PermissionLevel, str], value: int, add: bool = True
    ) -> None:
        if value != -1:
            value = str(value)
        if isinstance(name, PermissionLevel):
            level = True
            permissions = self.config["level_permissions"]
            name = name.name
        else:
            level = False
            permissions = self.config["command_permissions"]
        if name not in permissions:
            if add:
                permissions[name] = [value]
        else:
            if add:
                if value not in permissions[name]:
                    permissions[name].append(value)
            else:
                if value in permissions[name]:
                    permissions[name].remove(value)

        if level:
            self.config["level_permissions"] = permissions
        else:
            self.config["command_permissions"] = permissions
        logger.info("Updating permissions for %s, %s (add=%s).", name, value, add)
        await self.config.update()

    async def on_message(self, message):
        await self.wait_for_connected()
        if message.type == discord.MessageType.pins_add and message.author == self.user:
            await message.delete()

        if (
            (f"<@{self.user.id}" in message.content or f"<@!{self.user.id}" in message.content)
            and self.config["alert_on_mention"]
            and not message.author.bot
        ):
            em = discord.Embed(
                title="Bot mention",
                description=f"[Jump URL]({message.jump_url})\n{truncate(message.content, 50)}",
                color=self.main_color,
            )
            if self.config["show_timestamp"]:
                em.timestamp = discord.utils.utcnow()

            if not self.config["silent_alert_on_mention"]:
                content = self.config["mention"]
            else:
                content = ""
            await self.mention_channel.send(content=content, embed=em)

        # --- MODERATOR-ONLY MESSAGE LOGGING ---
        # If a moderator sends a message directly in a thread channel (not via modmail command), log it
        if not message.author.bot and not isinstance(message.channel, discord.DMChannel):
            thread = await self.threads.find(channel=message.channel)
            if thread is not None:
                ctxs = await self.get_contexts(message)
                is_command = any(ctx.command for ctx in ctxs)
                if not is_command:
                    # Only log if not a command
                    perms = message.channel.permissions_for(message.author)
                    if perms.manage_messages or perms.administrator:
                        await self.api.append_log(message, type_="internal")

        await self.process_commands(message)

    async def process_commands(self, message):
        if message.author.bot:
            return

        if isinstance(message.channel, discord.DMChannel):
            return await self._queue_dm_message(message)

        ctxs = await self.get_contexts(message)
        for ctx in ctxs:
            if ctx.command:
                if not any(1 for check in ctx.command.checks if hasattr(check, "permission_level")):
                    logger.debug(
                        "Command %s has no permissions check, adding invalid level.",
                        ctx.command.qualified_name,
                    )
                    checks.has_permissions(PermissionLevel.INVALID)(ctx.command)

                # Check if thread is unsnoozing and queue command if so
                thread = await self.threads.find(channel=ctx.channel)
                if thread and thread._unsnoozing:
                    queued = await thread.queue_command(ctx, ctx.command)
                    if queued:
                        # Send a brief acknowledgment that command is queued
                        try:
                            await ctx.message.add_reaction("⏳")
                        except Exception as e:
                            logger.warning("Failed to add queued-reaction: %s", e)
                        continue

                await self.invoke(ctx)
                continue

            thread = await self.threads.find(channel=ctx.channel)
            if thread is not None:
                # If thread is snoozed (moved), auto-unsnooze when a mod sends a message directly in channel
                behavior = (self.config.get("snooze_behavior") or "delete").lower()
                if thread.snoozed and behavior == "move":
                    if not thread.snooze_data:
                        try:
                            log_entry = await self.api.logs.find_one(
                                {"recipient.id": str(thread.id), "snoozed": True}
                            )
                            if log_entry:
                                thread.snooze_data = log_entry.get("snooze_data")
                        except Exception:
                            logger.debug(
                                "Failed to add queued command reaction (⏳).",
                                exc_info=True,
                            )
                    try:
                        await thread.restore_from_snooze()
                        # refresh local cache
                        self.threads.cache[thread.id] = thread
                    except Exception as e:
                        logger.warning("Auto-unsnooze on direct message failed: %s", e)
                anonymous = False
                plain = False
                if self.config.get("anon_reply_without_command"):
                    anonymous = True
                if self.config.get("plain_reply_without_command"):
                    plain = True

                if (
                    self.config.get("reply_without_command")
                    or self.config.get("anon_reply_without_command")
                    or self.config.get("plain_reply_without_command")
                ):
                    # When replying without a command in a thread channel, use the raw content
                    # from the sent message as reply text while still preserving attachments.
                    await thread.reply(message, message.content, anonymous=anonymous, plain=plain)
            elif ctx.invoked_with:
                exc = commands.CommandNotFound('Command "{}" is not found'.format(ctx.invoked_with))
                self.dispatch("command_error", ctx, exc)

    async def on_typing(self, channel, user, _):
        await self.wait_for_connected()

        if user.bot:
            return

        if isinstance(channel, discord.DMChannel):
            if not self.config.get("user_typing"):
                return

            thread = await self.threads.find(recipient=user)

            if thread:
                try:
                    await thread.channel.typing()
                except Exception:
                    logger.debug(
                        "Failed to trigger typing indicator in recipient DM.",
                        exc_info=True,
                    )
        else:
            if not self.config.get("mod_typing"):
                return

            thread = await self.threads.find(channel=channel)
            if thread is not None and thread.recipient:
                for user in thread.recipients:
                    if await self.is_blocked(user):
                        continue
                    try:
                        await user.typing()
                    except Exception:
                        logger.debug(
                            "Failed to trigger typing for recipient %s.",
                            getattr(user, "id", "?"),
                            exc_info=True,
                        )

    async def handle_reaction_events(self, payload):
        user = self.get_user(payload.user_id)
        if user is None or user.bot:
            return

        channel = self.get_channel(payload.channel_id)
        thread = None
        # dm channel not in internal cache
        if not channel:
            thread = await self.threads.find(recipient=user)
            if not thread:
                return
            channel = await thread.recipient.create_dm()
            if channel.id != payload.channel_id:
                return

        from_dm = isinstance(channel, discord.DMChannel)
        from_txt = isinstance(channel, discord.TextChannel)
        if not from_dm and not from_txt:
            return

        if not thread:
            params = {"recipient": user} if from_dm else {"channel": channel}
            thread = await self.threads.find(**params)
            if not thread:
                return

        # thread must exist before doing this API call
        try:
            message = await channel.fetch_message(payload.message_id)
        except (discord.NotFound, discord.Forbidden):
            return

        reaction = payload.emoji
        close_emoji = await self.convert_emoji(self.config["close_emoji"])
        if from_dm:
            if (
                payload.event_type == "REACTION_ADD"
                and message.embeds
                and str(reaction) == str(close_emoji)
                and self.config.get("recipient_thread_close")
            ):
                ts = message.embeds[0].timestamp
                if ts == thread.channel.created_at:
                    # the reacted message is the corresponding thread creation embed
                    # closing thread
                    return await thread.close(closer=user)
            if (
                message.author == self.user
                and message.embeds
                and self.config.get("confirm_thread_creation")
                and message.embeds[0].title == self.config["confirm_thread_creation_title"]
                and message.embeds[0].description == self.config["confirm_thread_response"]
            ):
                return
            if not thread.recipient.dm_channel:
                await thread.recipient.create_dm()
            try:
                linked_messages = await thread.find_linked_message_from_dm(message, either_direction=True)
            except ValueError as e:
                logger.warning("Failed to find linked message for reactions: %s", e)
                return
        else:
            try:
                _, *linked_messages = await thread.find_linked_messages(
                    message1=message, either_direction=True
                )
            except ValueError as e:
                logger.warning("Failed to find linked message for reactions: %s", e)
                return

        if self.config["transfer_reactions"] and linked_messages != [None]:
            if payload.event_type == "REACTION_ADD":
                for msg in linked_messages:
                    await self.add_reaction(msg, reaction)
                await self.add_reaction(message, reaction)
            else:
                try:
                    for msg in linked_messages:
                        await msg.remove_reaction(reaction, self.user)
                    await message.remove_reaction(reaction, self.user)
                except (discord.HTTPException, TypeError) as e:
                    logger.warning("Failed to remove reaction: %s", e)

    async def handle_react_to_contact(self, payload):
        react_message_id = tryint(self.config.get("react_to_contact_message"))
        react_message_emoji = self.config.get("react_to_contact_emoji")
        if not all((react_message_id, react_message_emoji)) or payload.message_id != react_message_id:
            return
        if payload.emoji.is_unicode_emoji():
            emoji_fmt = payload.emoji.name
        else:
            emoji_fmt = f"<:{payload.emoji.name}:{payload.emoji.id}>"

        if emoji_fmt != react_message_emoji:
            return
        channel = self.get_channel(payload.channel_id)
        member = channel.guild.get_member(payload.user_id)
        if member.bot:
            return
        message = await channel.fetch_message(payload.message_id)
        await message.remove_reaction(payload.emoji, member)
        await message.add_reaction(emoji_fmt)  # bot adds as well

        if self.config["dm_disabled"] in (
            DMDisabled.NEW_THREADS,
            DMDisabled.ALL_THREADS,
        ):
            embed = discord.Embed(
                title=self.config["disabled_new_thread_title"],
                color=self.error_color,
                description=self.config["disabled_new_thread_response"],
            )
            embed.set_footer(
                text=self.config["disabled_new_thread_footer"],
                icon_url=self.get_guild_icon(guild=channel.guild, size=128),
            )
            logger.info(
                "A new thread using react to contact was blocked from %s due to disabled Modmail.",
                member,
            )
            return await member.send(embed=embed)

        # Check if user has a snoozed thread
        existing_thread = await self.threads.find(recipient=member)
        if existing_thread and existing_thread.snoozed:
            # Unsnooze the thread
            await existing_thread.restore_from_snooze()
            self.threads.cache[existing_thread.id] = existing_thread
            # Send notification to the thread channel
            if existing_thread.channel:
                await existing_thread.channel.send(
                    f"ℹ️ {member.mention} reacted to contact and their snoozed thread has been unsnoozed."
                )
            return

        ctx = await self.get_context(message)
        await ctx.invoke(self.get_command("contact"), users=[member], manual_trigger=False)

    async def on_raw_reaction_add(self, payload):
        await asyncio.gather(
            self.handle_reaction_events(payload),
            self.handle_react_to_contact(payload),
        )

    async def on_raw_reaction_remove(self, payload):
        if self.config["transfer_reactions"]:
            await self.handle_reaction_events(payload)

    async def on_guild_channel_delete(self, channel):
        if channel.guild != self.modmail_guild:
            return

        if isinstance(channel, discord.CategoryChannel):
            if self.main_category == channel:
                logger.debug("Main category was deleted.")
                self.config.remove("main_category_id")
                await self.config.update()
            return

        if not isinstance(channel, discord.TextChannel):
            return

        if self.log_channel is None or self.log_channel == channel:
            logger.info("Log channel deleted.")
            self.config.remove("log_channel_id")
            await self.config.update()
            return

        # Attempt to attribute channel deletion to a moderator via audit logs.
        # This requires the "View Audit Log" permission; if missing, skip silently.
        if not self.modmail_guild.me.guild_permissions.view_audit_log:
            logger.debug(
                "Skipping audit log lookup for deleted channel %d: missing view_audit_log permission.",
                channel.id,
            )
            return

        try:
            audit_logs = self.modmail_guild.audit_logs(limit=10, action=discord.AuditLogAction.channel_delete)
            found_entry = False
            async for entry in audit_logs:
                if int(entry.target.id) == channel.id:
                    found_entry = True
                    break
        except discord.Forbidden:
            logger.debug(
                "Forbidden when fetching audit logs for deleted channel %d (missing permission).", channel.id
            )
            return
        except discord.HTTPException as e:
            logger.debug("HTTPException when fetching audit logs for deleted channel %d: %s", channel.id, e)
            return

        if not found_entry:
            logger.debug("Cannot find the audit log entry for channel delete of %d.", channel.id)
            return

        mod = entry.user
        if mod == self.user:
            return

        thread = await self.threads.find(channel=channel)
        if thread and thread.channel == channel:
            logger.debug("Manually closed channel %s.", channel.name)
            await thread.close(closer=mod, silent=True, delete_channel=False)

    async def on_member_remove(self, member):
        thread = await self.threads.find(recipient=member)
        if thread:
            if member.guild == self.guild and self.config["close_on_leave"]:
                await thread.close(
                    closer=member.guild.me,
                    message=self.config["close_on_leave_reason"],
                    silent=True,
                )
            else:
                if len(self.guilds) > 1:
                    guild_left = member.guild
                    remaining_guilds = member.mutual_guilds

                    if remaining_guilds:
                        remaining_guild_names = [guild.name for guild in remaining_guilds]
                        leave_message = (
                            f"The recipient has left {guild_left}. "
                            f"They are still in {human_join(remaining_guild_names, final='and')}."
                        )
                    else:
                        leave_message = (
                            f"The recipient has left {guild_left}. We no longer share any mutual servers."
                        )
                else:
                    leave_message = "The recipient has left the server."

                embed = discord.Embed(description=leave_message, color=self.error_color)
                await thread.channel.send(embed=embed)

    async def on_member_join(self, member):
        thread = await self.threads.find(recipient=member)
        if thread:
            if len(self.guilds) > 1:
                guild_joined = member.guild
                join_message = f"The recipient has joined {guild_joined}."
            else:
                join_message = "The recipient has joined the server."
            embed = discord.Embed(description=join_message, color=self.mod_color)
            await thread.channel.send(embed=embed)

    async def on_message_delete(self, message):
        """Support for deleting linked messages"""

        if message.is_system():
            return

        if isinstance(message.channel, discord.DMChannel):
            if message.author == self.user:
                return
            thread = await self.threads.find(recipient=message.author)
            if not thread:
                return
            try:
                message = await thread.find_linked_message_from_dm(message, get_thread_channel=True)
            except ValueError as e:
                if str(e) != "Thread channel message not found.":
                    logger.debug("Failed to find linked message to delete: %s", e)
                return
            message = message[0]
            embed = message.embeds[0]

            if embed.footer.icon:
                icon_url = embed.footer.icon.url
            else:
                icon_url = None

            embed.set_footer(text=f"{embed.footer.text} (deleted)", icon_url=icon_url)
            await message.edit(embed=embed)
            return

        if message.author != self.user:
            return

        thread = await self.threads.find(channel=message.channel)
        if not thread:
            return

        try:
            await thread.delete_message(message, note=False)
            embed = discord.Embed(description="Successfully deleted message.", color=self.main_color)
        except ValueError as e:
            # Treat common non-fatal cases as benign: relay counterpart not present, note embeds, etc.
            if str(e) not in {
                "DM message not found.",
                "Malformed thread message.",
                "Thread message not found.",
            }:
                logger.debug("Failed to find linked message to delete: %s", e)
                embed = discord.Embed(description="Failed to delete message.", color=self.error_color)
            else:
                return
        except discord.NotFound:
            return
        embed.set_footer(text=f"Message ID: {message.id} from {message.author}.")
        return await message.channel.send(embed=embed)

    async def on_bulk_message_delete(self, messages):
        await discord.utils.async_all(self.on_message_delete(msg) for msg in messages)

    async def on_message_edit(self, before, after):
        if after.author.bot:
            return
        if before.content == after.content:
            return

        if isinstance(after.channel, discord.DMChannel):
            thread = await self.threads.find(recipient=before.author)
            if not thread:
                return

            try:
                await thread.edit_dm_message(after, after.content)
            except ValueError:
                _, blocked_emoji = await self.retrieve_emoji()
                await self.add_reaction(after, blocked_emoji)
            else:
                embed = discord.Embed(description="Successfully Edited Message", color=self.main_color)
                embed.set_footer(text=f"Message ID: {after.id}")
                await after.channel.send(embed=embed)

    async def on_error(self, event_method, *args, **kwargs):
        logger.error("Ignoring exception in %s.", event_method)
        logger.error("Unexpected exception:", exc_info=sys.exc_info())

    async def on_command_error(
        self,
        context: commands.Context,
        exception: Exception,
        *,
        unhandled_by_cog: bool = False,
    ) -> None:
        if not unhandled_by_cog:
            command = context.command
            if command and command.has_error_handler():
                return
            cog = context.cog
            if cog and cog.has_error_handler():
                return

        if isinstance(exception, (commands.BadArgument, commands.BadUnionArgument)):
            try:
                await context.typing()
            except Exception:
                logger.debug(
                    "Failed to start typing context for command error feedback.",
                    exc_info=True,
                )
            await context.send(embed=discord.Embed(color=self.error_color, description=str(exception)))
        elif isinstance(exception, commands.CommandNotFound):
            logger.warning("CommandNotFound: %s", exception)
        elif isinstance(exception, commands.MissingRequiredArgument):
            await context.send_help(context.command)
        elif isinstance(exception, commands.CommandOnCooldown):
            await context.send(
                embed=discord.Embed(
                    title="Command on cooldown",
                    description=f"Try again in {exception.retry_after:.2f} seconds",
                    color=self.error_color,
                )
            )
        elif isinstance(exception, commands.CheckFailure):
            for check in context.command.checks:
                if not await check(context):
                    if hasattr(check, "fail_msg"):
                        await context.send(
                            embed=discord.Embed(color=self.error_color, description=check.fail_msg)
                        )
                    if hasattr(check, "permission_level"):
                        corrected_permission_level = self.command_perm(context.command.qualified_name)
                        logger.warning(
                            "User %s does not have permission to use this command: `%s` (%s).",
                            context.author.name,
                            context.command.qualified_name,
                            corrected_permission_level.name,
                        )
            logger.warning("CheckFailure: %s", exception)
        elif isinstance(exception, commands.DisabledCommand):
            logger.info(
                "DisabledCommand: %s is trying to run eval but it's disabled",
                context.author.name,
            )
        else:
            logger.error("Unexpected exception:", exc_info=exception)

    @tasks.loop(hours=1)
    async def post_metadata(self):
        info = await self.application_info()

        delta = discord.utils.utcnow() - self.start_time
        data = {
            "bot_id": self.user.id,
            "bot_name": str(self.user),
            "avatar_url": self.user.display_avatar.url,
            "guild_id": self.guild_id,
            "guild_name": self.guild.name,
            "member_count": len(self.guild.members),
            "uptime": delta.total_seconds(),
            "latency": f"{self.ws.latency * 1000:.4f}",
            "version": str(self.version),
            "selfhosted": True,
            "last_updated": str(discord.utils.utcnow()),
        }

        if info.team is not None:
            data.update(
                {
                    "owner_name": info.team.owner.name if info.team.owner is not None else "No Owner",
                    "owner_id": info.team.owner_id,
                    "team": True,
                }
            )
        else:
            data.update(
                {
                    "owner_name": info.owner.name,
                    "owner_id": info.owner.id,
                    "team": False,
                }
            )

        async with self.session.post("https://api.modmail.dev/metadata", json=data):
            logger.debug("Uploading metadata to Modmail server.")

    @post_metadata.before_loop
    async def before_post_metadata(self):
        await self.wait_for_connected()
        if not self.config.get("data_collection") or not self.guild:
            self.post_metadata.cancel()
            return

        logger.debug("Starting metadata loop.")
        logger.line("debug")

    @tasks.loop(hours=1)
    async def autoupdate(self):
        changelog = await Changelog.from_url(self)
        latest = changelog.latest_version

        if self.version < Version(latest.version):
            error = None
            data = {}
            try:
                # update fork if gh_token exists
                data = await self.api.update_repository()
            except InvalidConfigError:
                pass
            except ClientResponseError as exc:
                error = exc
            if self.hosting_method == HostingMethod.HEROKU:
                if error is not None:
                    logger.error(f"Autoupdate failed! Status: {error.status}.")
                    logger.error(f"Error message: {error.message}")
                    self.autoupdate.cancel()
                    return

                commit_data = data.get("data")
                if not commit_data:
                    return

                logger.info("Bot has been updated.")

                if not self.config["update_notifications"]:
                    return

                embed = discord.Embed(color=self.main_color)
                message = commit_data["commit"]["message"]
                html_url = commit_data["html_url"]
                short_sha = commit_data["sha"][:6]
                user = data["user"]
                embed.add_field(
                    name="Merge Commit",
                    value=f"[`{short_sha}`]({html_url}) {message} - {user['username']}",
                )
                embed.set_author(
                    name=user["username"] + " - Updating Bot",
                    icon_url=user["avatar_url"],
                    url=user["url"],
                )

                embed.set_footer(text=f"Updating Modmail v{self.version} -> v{latest.version}")

                embed.description = latest.description
                for name, value in latest.fields.items():
                    embed.add_field(name=name, value=value)

                channel = self.update_channel
                await channel.send(embed=embed)
            else:
                command = "git pull"
                proc = await asyncio.create_subprocess_shell(
                    command,
                    stderr=PIPE,
                    stdout=PIPE,
                )
                err = await proc.stderr.read()
                err = err.decode("utf-8").rstrip()
                res = await proc.stdout.read()
                res = res.decode("utf-8").rstrip()

                if err and not res:
                    logger.warning(f"Autoupdate failed: {err}")
                    self.autoupdate.cancel()
                    return

                elif res != "Already up to date.":
                    if os.getenv("PIPENV_ACTIVE"):
                        # Update pipenv if possible
                        await asyncio.create_subprocess_shell(
                            "pipenv sync",
                            stderr=PIPE,
                            stdout=PIPE,
                        )
                        message = ""
                    else:
                        message = "\n\nDo manually update dependencies if your bot has crashed."

                    logger.info("Bot has been updated.")
                    channel = self.update_channel
                    if self.hosting_method in (
                        HostingMethod.PM2,
                        HostingMethod.SYSTEMD,
                    ):
                        embed = discord.Embed(title="Bot has been updated", color=self.main_color)
                        embed.set_footer(
                            text=f"Updating Modmail v{self.version} " f"-> v{latest.version} {message}"
                        )
                        if self.config["update_notifications"]:
                            await channel.send(embed=embed)
                    else:
                        embed = discord.Embed(
                            title="Bot has been updated and is logging out.",
                            description=f"If you do not have an auto-restart setup, please manually start the bot. {message}",
                            color=self.main_color,
                        )
                        embed.set_footer(text=f"Updating Modmail v{self.version} -> v{latest.version}")
                        if self.config["update_notifications"]:
                            await channel.send(embed=embed)
                    return await self.close()

    @autoupdate.before_loop
    async def before_autoupdate(self):
        await self.wait_for_connected()
        logger.debug("Starting autoupdate loop")

        if self.config.get("disable_autoupdates"):
            logger.warning("Autoupdates disabled.")
            self.autoupdate.cancel()
            return

        if self.hosting_method == HostingMethod.DOCKER:
            logger.warning("Autoupdates disabled as using Docker.")
            self.autoupdate.cancel()
            return

        if not self.config.get("github_token") and self.hosting_method == HostingMethod.HEROKU:
            logger.warning("GitHub access token not found.")
            logger.warning("Autoupdates disabled.")
            self.autoupdate.cancel()
            return

    @tasks.loop(hours=1, reconnect=False)
    async def log_expiry(self):
        log_expire_after = self.config.get("log_expiration")
        if log_expire_after == isodate.Duration():
            return self.log_expiry.stop()

        now = discord.utils.utcnow()
        expiration_datetime = now - log_expire_after
        # WARNING: comparison is done lexicographically, not by date.
        # This is fine as long as the date is in zero-padded ISO format, which it should be.
        expired_logs = await self.db.logs.delete_many({"closed_at": {"$lte": str(expiration_datetime)}})

        logger.info(f"Deleted {expired_logs.deleted_count} expired logs.")

    def format_channel_name(self, author, exclude_channel=None, force_null=False):
        """Sanitises a username for use with text channel names

        Placed in main bot class to be extendable to plugins"""
        guild = self.modmail_guild

        if force_null:
            name = new_name = "null"
        else:
            if self.config["use_random_channel_name"]:
                to_hash = self.token.split(".")[-1] + str(author.id)
                digest = hashlib.md5(to_hash.encode("utf8"), usedforsecurity=False)
                name = new_name = digest.hexdigest()[-8:]
            elif self.config["use_user_id_channel_name"]:
                name = new_name = str(author.id)
            elif self.config["use_timestamp_channel_name"]:
                name = new_name = author.created_at.isoformat(sep="-", timespec="minutes")
            else:
                if self.config["use_nickname_channel_name"]:
                    author_member = self.guild.get_member(author.id)
                    name = author_member.display_name.lower()
                else:
                    name = author.name.lower()

                if force_null:
                    name = "null"

                name = "".join(l for l in name if l not in string.punctuation and l.isprintable()) or "null"
                if author.discriminator != "0":
                    name += f"-{author.discriminator}"
                new_name = name

        counter = 1
        existed = set(c.name for c in guild.text_channels if c != exclude_channel)
        while new_name in existed:
            new_name = f"{name}_{counter}"  # multiple channels with same name
            counter += 1

        return new_name


def main():
    try:
        # noinspection PyUnresolvedReferences
        import uvloop  # type: ignore

        logger.debug("Setting up with uvloop.")
        uvloop.install()
    except ImportError:
        pass

    try:
        import cairosvg  # noqa: F401
    except OSError:
        if os.name == "nt":
            if struct.calcsize("P") * 8 != 64:
                logger.error(
                    "Unable to import cairosvg, ensure your Python is a 64-bit version: https://www.python.org/downloads/"
                )
            else:
                logger.error(
                    "Unable to import cairosvg, install GTK Installer for Windows and restart your system (https://github.com/tschoonj/GTK-for-Windows-Runtime-Environment-Installer/releases/latest)"
                )
        else:
            if "ubuntu" in platform.version().lower() or "debian" in platform.version().lower():
                logger.error(
                    "Unable to import cairosvg, try running `sudo apt-get install libpangocairo-1.0-0` or report on our support server with your OS details: https://discord.gg/etJNHCQ"
                )
            else:
                logger.error(
                    "Unable to import cairosvg, report on our support server with your OS details: https://discord.gg/etJNHCQ"
                )
        sys.exit(0)

    # check discord version
    discord_version = "2.6.3"
    if discord.__version__ != discord_version:
        logger.error(
            "Dependencies are not updated, run pipenv install. discord.py version expected %s, received %s",
            discord_version,
            discord.__version__,
        )
        sys.exit(0)

    bot = ModmailBot()
    bot.run()


if __name__ == "__main__":
    main()