perf(backend): message writes look up the stored copy by its date and touch one chunk

backend/src/userbot/modules/capture/message.py

@@ -63,7 +63,7 @@ async def capture_message(
     )
     if ctx.pending_deletes.take(chat_id, message.id):
         await repository.mark_deleted_late(
-            ctx.pool, ctx.account_id, chat_id, message.id
+            ctx.pool, ctx.account_id, chat_id, message.id, message.date
         )
     from userbot.modules.capture.identity import capture_identity  # noqa: PLC0415
 

backend/src/userbot/modules/capture/repository.py

@@ -49,7 +49,8 @@ WHERE account_id = $1 AND chat_id = $2 AND message_id = $3 AND date = $4
 
 _MARK_DELETED_LATE = """
 UPDATE messages SET deleted_at = now()
-WHERE account_id = $1 AND chat_id = $2 AND message_id = $3 AND deleted_at IS NULL
+WHERE account_id = $1 AND chat_id = $2 AND message_id = $3 AND date = $4
+  AND deleted_at IS NULL
 """
 
 _INSERT_VERSION = """
@@ -64,7 +65,7 @@ INSERT INTO message_versions
     (account_id, chat_id, message_id, observed_at, edit_date, text, raw)
 SELECT account_id, chat_id, message_id, clock_timestamp(), NULL, text, raw
 FROM messages m
-WHERE m.account_id = $1 AND m.chat_id = $2 AND m.message_id = $3
+WHERE m.account_id = $1 AND m.chat_id = $2 AND m.message_id = $3 AND m.date = $4
   AND NOT EXISTS (
     SELECT 1 FROM message_versions v
     WHERE v.account_id = m.account_id AND v.chat_id = m.chat_id
@@ -103,8 +104,7 @@ SELECT
      WHERE d.account_id = m.account_id AND d.chat_id = m.chat_id
        AND d.message_id = m.message_id) AS media_unique_id
 FROM messages m
-WHERE m.account_id = $1 AND m.chat_id = $2 AND m.message_id = $3
+WHERE m.account_id = $1 AND m.chat_id = $2 AND m.message_id = $3 AND m.date = $4
-ORDER BY m.date DESC LIMIT 1
 """
 
 _CURRENT_MEDIA = """
@@ -188,7 +188,7 @@ async def _lock_current(
 ) -> asyncpg.Record | None:
     await conn.execute(_LOCK_MESSAGE, account_id, row.chat_id, row.message_id)
     return await conn.fetchrow(
-        _CURRENT_CONTENT, account_id, row.chat_id, row.message_id
+        _CURRENT_CONTENT, account_id, row.chat_id, row.message_id, row.date
     )
 
 
@@ -222,7 +222,9 @@ async def _record_past(
 async def _record_edit(
     conn: asyncpg.Connection, account_id: int, row: MessageRow, *, mark_edited: bool
 ) -> None:
-    await conn.execute(_SNAPSHOT_ORIGINAL, account_id, row.chat_id, row.message_id)
+    await conn.execute(
+        _SNAPSHOT_ORIGINAL, account_id, row.chat_id, row.message_id, row.date
+    )
     await conn.execute(_TOUCH_EDITED, account_id, *row.columns, mark_edited)
     await conn.execute(
         _INSERT_VERSION,
@@ -316,9 +318,9 @@ async def add_version(pool: asyncpg.Pool, account_id: int, row: MessageRow) -> b
 
 
 async def mark_deleted_late(
-    pool: asyncpg.Pool, account_id: int, chat_id: int, message_id: int
+    pool: asyncpg.Pool, account_id: int, chat_id: int, message_id: int, date: datetime
 ) -> None:
-    await pool.execute(_MARK_DELETED_LATE, account_id, chat_id, message_id)
+    await pool.execute(_MARK_DELETED_LATE, account_id, chat_id, message_id, date)
 
 
 async def current_media(