Troubleshooting

def _is_thread_missing(err: TelegramBadRequest) -> bool:
    msg = (getattr(err, "message", None) or "").lower()
    return (
        "message thread not found" in msg
        or "message thread is not found" in msg
        or "thread not found" in msg
        or ("topic" in msg and "closed" in msg)
    )

except TelegramBadRequest as err:
    if not _is_thread_missing(err):
        try:
            await bot.send_message(
                chat_id=self._operator_group_id,
                message_thread_id=topic.topic_id,
                text=(
                    "Failed to copy the user's message.\n"
                    f"type={message.content_type}, message_id={message.message_id}\n"
                    f"error={getattr(err, 'message', str(err))}"
                ),
            )
        except Exception:
            pass
        return topic
    await self._db.deactivate_conversation(message.from_user.id)
    topic = await self.ensure_topic(bot, message.from_user)
    reply_params = None
    copy_result = await bot.copy_message(
        chat_id=self._operator_group_id,
        from_chat_id=message.chat.id,
        message_id=message.message_id,
        message_thread_id=topic.topic_id,
        reply_parameters=reply_params,
    )

except TelegramForbiddenError:
    await message.reply(
        "The user has blocked the bot or has not opened the chat with the bot."
    )
    return

async def connect(self) -> None:
    self._conn = await aiosqlite.connect(self._path)
    await self._conn.execute("PRAGMA journal_mode=WAL;")
    await self._conn.execute("PRAGMA foreign_keys=ON;")
    await self._conn.commit()

self._tx_lock = asyncio.Lock()

@asynccontextmanager
async def transaction(self) -> Any:
    async with self._tx_lock:
        await self.conn.execute("BEGIN;")
        try:
            yield
        except BaseException:
            await self.conn.rollback()
            raise
        else:
            await self.conn.commit()

self._locks: dict[int, asyncio.Lock] = {}
self._locks_guard = asyncio.Lock()

async def _lock_for(self, user_id: int) -> asyncio.Lock:
    async with self._locks_guard:
        lock = self._locks.get(user_id)
        if lock is None:
            lock = asyncio.Lock()
            self._locks[user_id] = lock
        return lock

async def ensure_topic(self, bot: Bot, user: User) -> TopicRef:
    lock = await self._lock_for(user.id)
    async with lock:
        existing = await self._db.get_active_conversation(user.id)
        if existing:
            return TopicRef(user_id=user.id, topic_id=existing.topic_id)

me = await bot.get_me()
log.info("Started as @%s (id=%s)", me.username, me.id)

user_id = await db.find_user_id_by_topic(int(message.message_thread_id))
if user_id is None:
    return

async def find_user_id_by_topic(self, topic_id: int) -> int | None:
    cur = await self.conn.execute(
        "SELECT user_id FROM conversations WHERE topic_id = ? AND active = 1",
        (topic_id,),
    )
    row = await cur.fetchone()
    await cur.close()
    return int(row[0]) if row else None

async def _log_message_link(
    self,
    *,
    user_id: int,
    source_chat_id: int,
    source_message_id: int,
    target_chat_id: int,
    target_message_id: int,
) -> None:
    async with self._db.transaction():
        await self._db.log_message_link(
            user_id=user_id,
            source_chat_id=source_chat_id,
            source_message_id=source_message_id,
            target_chat_id=target_chat_id,
            target_message_id=target_message_id,
            commit=False,
        )
        await self._db.log_message_link(
            user_id=user_id,
            source_chat_id=target_chat_id,
            source_message_id=target_message_id,
            target_chat_id=source_chat_id,
            target_message_id=source_message_id,
            commit=False,
        )