From d835d97331990590cb9dd8a779f7fe357f4464d4 Mon Sep 17 00:00:00 2001
From: Daniil Okhlopkov <5613295+ohld@users.noreply.github.com>
Date: Wed, 20 May 2026 13:54:30 +0300
Subject: [PATCH 1/4] Add early moderator source vote rejection

---
 scripts/serve_flows.py                        |   2 +-
 specs/moderator-community-loop.md             |  24 +-
 src/storage/source_voting.py                  | 187 ++++++++++++++--
 .../handlers/moderator/source_candidates.py   |  13 +-
 tests/storage/test_source_voting.py           | 208 +++++++++++++++++-
 .../test_source_candidate_vote_handler.py     | 112 ++++++++++
 6 files changed, 513 insertions(+), 33 deletions(-)
 create mode 100644 tests/tgbot/test_source_candidate_vote_handler.py

diff --git a/scripts/serve_flows.py b/scripts/serve_flows.py
index 1cd7ec77..03e8d2ac 100644
--- a/scripts/serve_flows.py
+++ b/scripts/serve_flows.py
@@ -120,7 +120,7 @@
         # ── Moderator community loops ──
         daily_moderator_source_voting.to_deployment(
             name="Daily Moderator Source Voting",
-            schedules=[CronSchedule(cron="0 10 * * *", timezone=MSK)],
+            schedules=[CronSchedule(cron="*/15 * * * *", timezone=MSK)],
         ),
         # ── Broadcasts ──
         broadcast_next_meme_to_active_15m_ago.to_deployment(
diff --git a/specs/moderator-community-loop.md b/specs/moderator-community-loop.md
index 67fd4356..af68e078 100644
--- a/specs/moderator-community-loop.md
+++ b/specs/moderator-community-loop.md
@@ -183,15 +183,20 @@ If the poll is closed, answer `Голосование уже закрыто` and
 
 ## Daily Source Cycle
 
-Run once every 24 hours from a small scheduled flow or admin command.
+Run from a small scheduled flow or admin command. The production schedule may
+check frequently (for example every 15 minutes) so early negative polls can be
+closed soon after they become eligible.
 
 Order matters:
 
 1. Post the **Next-Day Source Report** for the last source that passed and was enabled in a previous cycle, if it has not been reported yet.
-2. Close the currently `open` poll, if one exists:
+2. Close the currently `open` poll if its 24-hour window has elapsed or if it
+   matches the early negative rule:
    - recompute likes/dislikes from `meme_source_candidate_vote`;
    - mark the poll `passed`, `rejected`, or `expired_no_quorum`;
-   - edit the moderator-chat message so voting is visibly closed.
+   - edit the moderator-chat message so it keeps the source URL and shows
+     final vote results;
+   - unpin the closed voting message.
 3. If the closed poll passed, enable the prepared source:
    - load `poll.prepared_meme_source_id` or the candidate's `promoted_meme_source_id`;
    - set `language_code='ru'`;
@@ -208,7 +213,7 @@ Order matters:
    - create or reuse a `meme_source` row with `status='in_moderation'`;
    - fetch the latest public Telegram posts once;
    - inspect the fetched posts for Cyrillic evidence;
-   - if Cyrillic is absent, dismiss the candidate and stop the cycle for the day without posting a poll;
+   - if Cyrillic is absent, dismiss the candidate and stop the current run without posting a poll;
    - if Cyrillic is present, save the fetched posts into `meme_raw_telegram`;
    - mark the candidate `prepared` and store `promoted_meme_source_id`.
 6. Insert `meme_source_candidate_poll` as `draft`, with `prepared_meme_source_id` and `closes_at = now() + interval '24 hours'`.
@@ -216,7 +221,9 @@ Order matters:
 8. Update poll with `chat_id`, `message_id`, `opened_at`, `status='open'`.
 9. If send fails, mark `status='cancelled'` and keep the prepared source in `in_moderation` for an admin retry.
 
-Hard limit: 1 new source poll per day. The point is a steady community ritual, not a high-volume ops feed.
+Hard limit: exactly one active source poll at a time. The point is a steady
+community ritual, not a high-volume ops feed; early-rejected non-meme sources
+may be replaced before the full 24-hour window.
 
 Message content should include:
 
@@ -285,7 +292,12 @@ Outcomes:
 
 Rejected sources must not return to the automatic daily cycle. If the owner wants to revisit one, they can add it manually later.
 
-Do not early-close. The daily cycle closes the poll after the full 24-hour window.
+Early negative close: after the poll has been open for at least 90 minutes,
+if it has 0 likes and at least 6 dislikes, close it immediately as
+`rejected`, set the prepared source to `parsing_disabled`, dismiss the
+candidate with `dismissed_reason='source_vote:{poll_id}:early_negative_not_meme_source'`,
+write `meme_source.data.source_vote_rejection.reason='early_negative_not_meme_source'`,
+and try to post the next candidate.
 
 ## Passing Source Flow
 
diff --git a/src/storage/source_voting.py b/src/storage/source_voting.py
index 269d4a0c..d7dc76f4 100644
--- a/src/storage/source_voting.py
+++ b/src/storage/source_voting.py
@@ -43,6 +43,9 @@
 SOURCE_VOTE_MIN_LIKES = 2
 SOURCE_VOTE_LIKE_SHARE_THRESHOLD = 0.30
 SOURCE_VOTE_WINDOW = timedelta(hours=24)
+SOURCE_VOTE_EARLY_REJECT_MIN_OPEN = timedelta(minutes=90)
+SOURCE_VOTE_EARLY_REJECT_MIN_DISLIKES = 6
+SOURCE_VOTE_EARLY_REJECT_REASON = "early_negative_not_meme_source"
 
 _CYRILLIC_RE = re.compile(r"[\u0400-\u04FF]")
 
@@ -363,6 +366,25 @@ async def record_source_candidate_vote(
     await execute(stmt)
 
     counts = await get_source_candidate_vote_counts(poll_id)
+    if source_vote_should_reject_early(poll, counts, now):
+        close_result = await close_source_candidate_poll(
+            poll_id,
+            now=now,
+            close_reason=SOURCE_VOTE_EARLY_REJECT_REASON,
+        )
+        if close_result["status"] == POLL_STATUS_REJECTED:
+            return {
+                "status": "early_rejected",
+                "poll": close_result["poll"],
+                "counts": close_result["counts"],
+                "close_result": close_result,
+            }
+        return {
+            "status": "closed",
+            "poll": close_result.get("poll", poll),
+            "counts": counts,
+        }
+
     return {
         "status": "changed" if existing and existing["vote"] != vote else "recorded",
         "poll": poll,
@@ -379,6 +401,20 @@ def source_vote_passed(counts: dict[str, int]) -> bool:
     )
 
 
+def source_vote_should_reject_early(
+    poll: dict[str, Any],
+    counts: dict[str, int],
+    now: datetime,
+) -> bool:
+    opened_at = poll.get("opened_at")
+    return (
+        opened_at is not None
+        and opened_at + SOURCE_VOTE_EARLY_REJECT_MIN_OPEN <= now
+        and counts["yes"] == 0
+        and counts["no"] >= SOURCE_VOTE_EARLY_REJECT_MIN_DISLIKES
+    )
+
+
 async def _set_poll_status(
     poll_id: int,
     *,
@@ -432,6 +468,37 @@ async def _append_source_vote_metadata(
     )
 
 
+async def _append_source_vote_rejection_metadata(
+    source_id: int,
+    *,
+    poll: dict[str, Any],
+    counts: dict[str, int],
+    rejected_at: datetime,
+    reason: str,
+) -> dict[str, Any] | None:
+    source = await fetch_one(select(meme_source).where(meme_source.c.id == source_id))
+    if source is None:
+        return None
+    data = dict(source.get("data") or {})
+    data["source_vote_rejection"] = {
+        **dict(data.get("source_vote_rejection") or {}),
+        "poll_id": poll["id"],
+        "candidate_id": poll["candidate_id"],
+        "chat_id": poll["chat_id"],
+        "message_id": poll["message_id"],
+        "yes": counts["yes"],
+        "no": counts["no"],
+        "closed_at": rejected_at.isoformat(),
+        "reason": reason,
+    }
+    return await fetch_one(
+        meme_source.update()
+        .where(meme_source.c.id == source_id)
+        .values(data=data, updated_at=_utcnow())
+        .returning(meme_source)
+    )
+
+
 async def enable_passed_source_poll(
     poll: dict[str, Any],
     counts: dict[str, int],
@@ -470,6 +537,7 @@ async def close_source_candidate_poll(
     poll_id: int,
     *,
     now: datetime | None = None,
+    close_reason: str | None = None,
 ) -> dict[str, Any]:
     now = now or _utcnow()
     poll = await get_source_candidate_poll(poll_id)
@@ -477,10 +545,15 @@ async def close_source_candidate_poll(
         return {"status": "not_found"}
     if poll["status"] != POLL_STATUS_OPEN:
         return {"status": "already_closed", "poll": poll}
+    candidate = await fetch_one(
+        select(meme_source_candidate).where(meme_source_candidate.c.id == poll["candidate_id"])
+    )
 
     counts = await get_source_candidate_vote_counts(poll_id)
     poll_data = dict(poll.get("data") or {})
     poll_data["final_counts"] = counts
+    if close_reason is not None:
+        poll_data["close_reason"] = close_reason
 
     result_source = None
     if counts["total"] < SOURCE_VOTE_QUORUM:
@@ -497,10 +570,21 @@ async def close_source_candidate_poll(
                 status=MemeSourceStatus.PARSING_DISABLED.value,
                 trigger_parse=False,
             )
+            await _append_source_vote_rejection_metadata(
+                poll["prepared_meme_source_id"],
+                poll=poll,
+                counts=counts,
+                rejected_at=now,
+                reason=close_reason or "vote_failed",
+            )
         await _mark_candidate(
             poll["candidate_id"],
             status=CANDIDATE_STATUS_DISMISSED,
-            dismissed_reason=f"source_vote:{poll_id}",
+            dismissed_reason=(
+                f"source_vote:{poll_id}"
+                if close_reason is None
+                else f"source_vote:{poll_id}:{close_reason}"
+            ),
         )
 
     updated_poll = await _set_poll_status(
@@ -513,11 +597,26 @@ async def close_source_candidate_poll(
     return {
         "status": status,
         "poll": updated_poll,
+        "candidate": candidate,
         "counts": counts,
         "source": result_source,
     }
 
 
+async def get_early_rejectable_source_candidate_poll(
+    now: datetime | None = None,
+) -> dict[str, Any] | None:
+    now = now or _utcnow()
+    poll = await get_active_source_candidate_poll()
+    if poll is None or poll["status"] != POLL_STATUS_OPEN:
+        return None
+
+    counts = await get_source_candidate_vote_counts(poll["id"])
+    if source_vote_should_reject_early(poll, counts, now):
+        return poll
+    return None
+
+
 async def select_daily_source_candidate() -> dict[str, Any] | None:
     return await fetch_one(
         text(
@@ -578,19 +677,66 @@ def format_closed_poll_message(result: dict[str, Any]) -> str:
     counts = result["counts"]
     status = result["status"]
     if status == POLL_STATUS_PASSED:
-        outcome = "Источник добавлен."
+        outcome = "Голосование завершено: источник добавлен."
     elif status == POLL_STATUS_REJECTED:
-        outcome = "Источник отклонён."
+        outcome = "Голосование завершено: мем-источник отклонён."
     elif status == POLL_STATUS_EXPIRED_NO_QUORUM:
-        outcome = "Недостаточно голосов. Автоматически не возвращаем."
+        outcome = "Голосование завершено: недостаточно голосов. Автоматически не возвращаем."
     else:
         outcome = f"Голосование закрыто: {status}."
-    return (
-        f"{outcome}\n\n"
-        f"За: {counts['yes']}\n"
-        f"Против: {counts['no']}\n"
-        f"Всего голосов: {counts['total']}"
+    candidate = result.get("candidate")
+    lines: list[str] = []
+    if candidate is not None:
+        lines.extend(
+            [
+                "Добавляли новый источник мемов?",
+                "",
+                f"🔗 {candidate['url']}",
+                "",
+                f"Наши паблики пересылали мемы оттуда {candidate['times_forwarded']} раз.",
+                "",
+            ]
+        )
+    lines.extend(
+        [
+            outcome,
+            "",
+            "Результаты голосования:",
+            f"За: {counts['yes']}",
+            f"Против: {counts['no']}",
+            f"Всего голосов: {counts['total']}",
+        ]
     )
+    return "\n".join(lines)
+
+
+async def update_closed_source_candidate_poll_message(
+    bot: Bot,
+    result: dict[str, Any],
+) -> None:
+    poll = result.get("poll")
+    if not poll or not poll.get("message_id"):
+        return
+
+    try:
+        await bot.edit_message_text(
+            chat_id=poll["chat_id"],
+            message_id=poll["message_id"],
+            text=format_closed_poll_message(result),
+            disable_web_page_preview=True,
+        )
+    except TelegramError:
+        pass
+
+    if not hasattr(bot, "unpin_chat_message"):
+        return
+    try:
+        await bot.unpin_chat_message(
+            chat_id=poll["chat_id"],
+            message_id=poll["message_id"],
+        )
+    except TelegramError:
+        pass
 
 
 async def create_source_candidate_poll(
@@ -869,16 +1015,21 @@ async def advance_daily_source_cycle(
     result["report"] = await post_next_day_source_report(bot, now=now)
 
     due_poll = await get_due_source_candidate_poll(now)
-    if due_poll is not None:
-        close_result = await close_source_candidate_poll(due_poll["id"], now=now)
+    poll_to_close = due_poll
+    close_reason = None
+    if poll_to_close is None:
+        poll_to_close = await get_early_rejectable_source_candidate_poll(now)
+        if poll_to_close is not None:
+            close_reason = SOURCE_VOTE_EARLY_REJECT_REASON
+
+    if poll_to_close is not None:
+        close_result = await close_source_candidate_poll(
+            poll_to_close["id"],
+            now=now,
+            close_reason=close_reason,
+        )
         result["closed_poll"] = close_result
-        if due_poll["message_id"]:
-            await bot.edit_message_text(
-                chat_id=due_poll["chat_id"],
-                message_id=due_poll["message_id"],
-                text=format_closed_poll_message(close_result),
-                disable_web_page_preview=True,
-            )
+        await update_closed_source_candidate_poll_message(bot, close_result)
 
     active_poll = await get_active_source_candidate_poll()
     if active_poll is not None:
diff --git a/src/tgbot/handlers/moderator/source_candidates.py b/src/tgbot/handlers/moderator/source_candidates.py
index 9f51da6e..751d628a 100644
--- a/src/tgbot/handlers/moderator/source_candidates.py
+++ b/src/tgbot/handlers/moderator/source_candidates.py
@@ -2,7 +2,11 @@
 from telegram.error import BadRequest
 from telegram.ext import ContextTypes
 
-from src.storage.source_voting import record_source_candidate_vote
+from src.storage.source_voting import (
+    post_new_source_candidate_poll,
+    record_source_candidate_vote,
+    update_closed_source_candidate_poll_message,
+)
 from src.tgbot.constants import UserType
 from src.tgbot.handlers.moderator.meme_source import meme_source_admin_pipeline
 from src.tgbot.logs import log
@@ -139,6 +143,13 @@ async def handle_source_candidate_vote(
                 raise
         return
 
+    if status == "early_rejected":
+        close_result = result["close_result"]
+        await query.answer("Источник отклонён, открываю следующий")
+        await update_closed_source_candidate_poll_message(context.bot, close_result)
+        await post_new_source_candidate_poll(context.bot)
+        return
+
     if status == "closed":
         await query.answer("Голосование уже закрыто")
         return
diff --git a/tests/storage/test_source_voting.py b/tests/storage/test_source_voting.py
index a6de1a26..0b776eb4 100644
--- a/tests/storage/test_source_voting.py
+++ b/tests/storage/test_source_voting.py
@@ -23,6 +23,8 @@
     CANDIDATE_STATUS_DISCOVERED,
     POLL_STATUS_OPEN,
     POLL_STATUS_PASSED,
+    POLL_STATUS_REJECTED,
+    SOURCE_VOTE_EARLY_REJECT_REASON,
     VOTE_ADD_SOURCE,
     VOTE_SKIP_SOURCE,
     advance_daily_source_cycle,
@@ -40,6 +42,8 @@
 
 CANDIDATE_ID = TEST_ID_START + 2200
 CANDIDATE_URL = "https://t.me/ffm_source_vote_candidate"
+SECOND_CANDIDATE_ID = TEST_ID_START + 2201
+SECOND_CANDIDATE_URL = "https://t.me/ffm_source_vote_candidate_next"
 
 
 def _post(post_id: int, content: str = "смешной мем") -> TgChannelPostParsingResult:
@@ -53,15 +57,21 @@ def _post(post_id: int, content: str = "смешной мем") -> TgChannelPost
     )
 
 
-async def _create_candidate(conn: AsyncConnection) -> None:
+async def _create_candidate(
+    conn: AsyncConnection,
+    *,
+    candidate_id: int = CANDIDATE_ID,
+    url: str = CANDIDATE_URL,
+    times_forwarded: int = 5,
+) -> None:
     await conn.execute(
         insert(meme_source_candidate)
         .values(
-            id=CANDIDATE_ID,
+            id=candidate_id,
             type="telegram",
-            url=CANDIDATE_URL,
+            url=url,
             status="discovered",
-            times_forwarded=5,
+            times_forwarded=times_forwarded,
             first_seen_at=datetime(2026, 5, 1, 10, 0, 0),
             last_seen_at=datetime(2026, 5, 2, 10, 0, 0),
         )
@@ -80,13 +90,19 @@ async def conn():
         )
         await conn.execute(
             delete(meme_source_candidate_poll).where(
-                meme_source_candidate_poll.c.candidate_id == CANDIDATE_ID
+                meme_source_candidate_poll.c.candidate_id.in_(
+                    [CANDIDATE_ID, SECOND_CANDIDATE_ID]
+                )
+            )
+        )
+        await conn.execute(
+            delete(meme_source_candidate).where(
+                meme_source_candidate.c.id.in_([CANDIDATE_ID, SECOND_CANDIDATE_ID])
             )
         )
         await conn.execute(
-            delete(meme_source_candidate).where(meme_source_candidate.c.id == CANDIDATE_ID)
+            delete(meme_source).where(meme_source.c.url.in_([CANDIDATE_URL, SECOND_CANDIDATE_URL]))
         )
-        await conn.execute(delete(meme_source).where(meme_source.c.url == CANDIDATE_URL))
         await conn.commit()
         await cleanup_test_data(conn)
 
@@ -234,6 +250,114 @@ async def test_source_candidate_vote_rejects_poll_outside_moderator_chat(
     assert result["status"] == "wrong_chat"
 
 
+@pytest.mark.asyncio
+async def test_source_candidate_vote_early_rejects_clear_negative_poll(
+    conn: AsyncConnection,
+):
+    now = datetime.utcnow()
+    await _create_candidate(conn)
+    await conn.commit()
+    prepared = await prepare_source_candidate(CANDIDATE_ID, posts=[_post(4013)])
+    poll = await create_source_candidate_poll(
+        CANDIDATE_ID,
+        prepared_meme_source_id=prepared["source"]["id"],
+        chat_id=TELEGRAM_MODERATOR_CHAT_ID,
+        now=now,
+    )
+    await conn.execute(
+        meme_source_candidate_poll.update()
+        .where(meme_source_candidate_poll.c.id == poll["id"])
+        .values(
+            status=POLL_STATUS_OPEN,
+            message_id=131,
+            opened_at=now - timedelta(hours=2),
+            closes_at=now + timedelta(hours=22),
+        )
+    )
+    await conn.commit()
+
+    result = None
+    for offset in range(1, 7):
+        result = await record_source_candidate_vote(
+            poll_id=poll["id"],
+            user_id=TEST_ID_START + 20 + offset,
+            vote=VOTE_SKIP_SOURCE,
+            chat_id=TELEGRAM_MODERATOR_CHAT_ID,
+            now=now,
+        )
+
+    assert result is not None
+    assert result["status"] == "early_rejected"
+    assert result["counts"] == {"yes": 0, "no": 6, "total": 6}
+
+    closed_poll = await fetch_one(
+        select(meme_source_candidate_poll).where(meme_source_candidate_poll.c.id == poll["id"])
+    )
+    assert closed_poll["status"] == POLL_STATUS_REJECTED
+    assert closed_poll["data"]["close_reason"] == SOURCE_VOTE_EARLY_REJECT_REASON
+
+    source = await fetch_one(
+        select(meme_source).where(meme_source.c.id == prepared["source"]["id"])
+    )
+    assert source["status"] == MemeSourceStatus.PARSING_DISABLED.value
+    assert source["data"]["source_vote_rejection"]["reason"] == SOURCE_VOTE_EARLY_REJECT_REASON
+    assert source["data"]["source_vote_rejection"]["no"] == 6
+
+    candidate = await fetch_one(
+        select(meme_source_candidate).where(meme_source_candidate.c.id == CANDIDATE_ID)
+    )
+    assert candidate["status"] == "dismissed"
+    assert candidate["dismissed_reason"] == (
+        f"source_vote:{poll['id']}:{SOURCE_VOTE_EARLY_REJECT_REASON}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_source_candidate_vote_does_not_early_reject_before_min_open_time(
+    conn: AsyncConnection,
+):
+    now = datetime.utcnow()
+    await _create_candidate(conn)
+    await conn.commit()
+    prepared = await prepare_source_candidate(CANDIDATE_ID, posts=[_post(4014)])
+    poll = await create_source_candidate_poll(
+        CANDIDATE_ID,
+        prepared_meme_source_id=prepared["source"]["id"],
+        chat_id=TELEGRAM_MODERATOR_CHAT_ID,
+        now=now,
+    )
+    await conn.execute(
+        meme_source_candidate_poll.update()
+        .where(meme_source_candidate_poll.c.id == poll["id"])
+        .values(
+            status=POLL_STATUS_OPEN,
+            message_id=132,
+            opened_at=now - timedelta(minutes=89),
+            closes_at=now + timedelta(hours=22),
+        )
+    )
+    await conn.commit()
+
+    result = None
+    for offset in range(1, 7):
+        result = await record_source_candidate_vote(
+            poll_id=poll["id"],
+            user_id=TEST_ID_START + 40 + offset,
+            vote=VOTE_SKIP_SOURCE,
+            chat_id=TELEGRAM_MODERATOR_CHAT_ID,
+            now=now,
+        )
+
+    assert result is not None
+    assert result["status"] == "recorded"
+    assert result["counts"] == {"yes": 0, "no": 6, "total": 6}
+
+    open_poll = await fetch_one(
+        select(meme_source_candidate_poll).where(meme_source_candidate_poll.c.id == poll["id"])
+    )
+    assert open_poll["status"] == POLL_STATUS_OPEN
+
+
 @pytest.mark.asyncio
 async def test_close_passed_poll_enables_prepared_source_without_reparsing(
     conn: AsyncConnection,
@@ -343,6 +467,76 @@ async def test_daily_source_cycle_resumes_existing_draft_poll(conn: AsyncConnect
     bot.send_message.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+async def test_daily_source_cycle_early_rejects_negative_poll_and_posts_next_candidate(
+    conn: AsyncConnection,
+):
+    now = datetime.utcnow()
+    await _create_candidate(conn)
+    await _create_candidate(
+        conn,
+        candidate_id=SECOND_CANDIDATE_ID,
+        url=SECOND_CANDIDATE_URL,
+        times_forwarded=4,
+    )
+    await conn.commit()
+    prepared = await prepare_source_candidate(CANDIDATE_ID, posts=[_post(4015)])
+    poll = await create_source_candidate_poll(
+        CANDIDATE_ID,
+        prepared_meme_source_id=prepared["source"]["id"],
+        chat_id=TELEGRAM_MODERATOR_CHAT_ID,
+        now=now,
+    )
+    await conn.execute(
+        meme_source_candidate_poll.update()
+        .where(meme_source_candidate_poll.c.id == poll["id"])
+        .values(
+            status=POLL_STATUS_OPEN,
+            message_id=558,
+            opened_at=now - timedelta(hours=2),
+            closes_at=now + timedelta(hours=22),
+        )
+    )
+    await conn.commit()
+
+    for offset in range(1, 7):
+        await record_source_candidate_vote(
+            poll_id=poll["id"],
+            user_id=TEST_ID_START + 60 + offset,
+            vote=VOTE_SKIP_SOURCE,
+            chat_id=TELEGRAM_MODERATOR_CHAT_ID,
+            now=now - timedelta(minutes=31),
+        )
+
+    bot = SimpleNamespace(
+        edit_message_text=AsyncMock(),
+        unpin_chat_message=AsyncMock(),
+        send_message=AsyncMock(return_value=SimpleNamespace(message_id=559)),
+    )
+    with patch(
+        "src.storage.source_voting.fetch_telegram_candidate_posts",
+        new=AsyncMock(return_value=[_post(4016)]),
+    ):
+        result = await advance_daily_source_cycle(bot, now=now)
+
+    assert result["closed_poll"]["status"] == POLL_STATUS_REJECTED
+    assert (
+        result["closed_poll"]["poll"]["data"]["close_reason"]
+        == SOURCE_VOTE_EARLY_REJECT_REASON
+    )
+    assert result["new_poll"]["status"] == "posted"
+    assert result["new_poll"]["candidate"]["id"] == SECOND_CANDIDATE_ID
+    bot.edit_message_text.assert_awaited_once()
+    edited_text = bot.edit_message_text.await_args.kwargs["text"]
+    assert CANDIDATE_URL in edited_text
+    assert "Голосование завершено: мем-источник отклонён." in edited_text
+    bot.unpin_chat_message.assert_awaited_once_with(
+        chat_id=TELEGRAM_MODERATOR_CHAT_ID,
+        message_id=558,
+    )
+    bot.send_message.assert_awaited_once()
+
+
 @pytest.mark.asyncio
 async def test_opening_draft_poll_refreshes_vote_window(conn: AsyncConnection):
     now = datetime.utcnow()
diff --git a/tests/tgbot/test_source_candidate_vote_handler.py b/tests/tgbot/test_source_candidate_vote_handler.py
new file mode 100644
index 00000000..3ece0798
--- /dev/null
+++ b/tests/tgbot/test_source_candidate_vote_handler.py
@@ -0,0 +1,112 @@
+from datetime import datetime, timedelta
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from src.storage.source_voting import (
+    format_closed_poll_message,
+    source_vote_should_reject_early,
+)
+from src.tgbot.constants import TELEGRAM_MODERATOR_CHAT_ID
+from src.tgbot.handlers.moderator import source_candidates
+
+
+def test_source_vote_should_reject_early_requires_age_zero_likes_and_six_dislikes():
+    opened_at = datetime(2026, 5, 20, 10, 0, 0)
+    poll = {"opened_at": opened_at}
+
+    assert source_vote_should_reject_early(
+        poll,
+        {"yes": 0, "no": 6, "total": 6},
+        opened_at + timedelta(minutes=90),
+    )
+    assert not source_vote_should_reject_early(
+        poll,
+        {"yes": 0, "no": 6, "total": 6},
+        opened_at + timedelta(minutes=89),
+    )
+    assert not source_vote_should_reject_early(
+        poll,
+        {"yes": 1, "no": 6, "total": 7},
+        opened_at + timedelta(hours=2),
+    )
+    assert not source_vote_should_reject_early(
+        poll,
+        {"yes": 0, "no": 5, "total": 5},
+        opened_at + timedelta(hours=2),
+    )
+
+
+@pytest.mark.asyncio
+async def test_handle_source_candidate_vote_early_rejects_and_posts_next_poll():
+    query = SimpleNamespace(
+        data="mscv:123:2",
+        answer=AsyncMock(),
+    )
+    update = SimpleNamespace(
+        callback_query=query,
+        effective_user=SimpleNamespace(id=42),
+        effective_chat=SimpleNamespace(id=TELEGRAM_MODERATOR_CHAT_ID),
+    )
+    context = SimpleNamespace(
+        bot=SimpleNamespace(
+            edit_message_text=AsyncMock(),
+            unpin_chat_message=AsyncMock(),
+        )
+    )
+    close_result = {
+        "status": "rejected",
+        "poll": {"chat_id": TELEGRAM_MODERATOR_CHAT_ID, "message_id": 123},
+        "candidate": {"url": "https://t.me/ravememe", "times_forwarded": 12},
+        "counts": {"yes": 0, "no": 6, "total": 6},
+    }
+
+    with (
+        patch.object(
+            source_candidates,
+            "record_source_candidate_vote",
+            new=AsyncMock(
+                return_value={
+                    "status": "early_rejected",
+                    "close_result": close_result,
+                }
+            ),
+        ),
+        patch.object(
+            source_candidates,
+            "post_new_source_candidate_poll",
+            new=AsyncMock(return_value={"status": "posted"}),
+        ) as post_new_poll,
+    ):
+        await source_candidates.handle_source_candidate_vote(update, context)
+
+    query.answer.assert_awaited_once_with("Источник отклонён, открываю следующий")
+    context.bot.edit_message_text.assert_awaited_once_with(
+        chat_id=TELEGRAM_MODERATOR_CHAT_ID,
+        message_id=123,
+        text=format_closed_poll_message(close_result),
+        disable_web_page_preview=True,
+    )
+    context.bot.unpin_chat_message.assert_awaited_once_with(
+        chat_id=TELEGRAM_MODERATOR_CHAT_ID,
+        message_id=123,
+    )
+    post_new_poll.assert_awaited_once_with(context.bot)
+
+
+def test_format_closed_poll_message_keeps_source_context_and_results():
+    text = format_closed_poll_message(
+        {
+            "status": "rejected",
+            "candidate": {"url": "https://t.me/ravememe", "times_forwarded": 12},
+            "counts": {"yes": 0, "no": 6, "total": 6},
+        }
+    )
+
+    assert "https://t.me/ravememe" in text
+    assert "Наши паблики пересылали мемы оттуда 12 раз." in text
+    assert "Голосование завершено: мем-источник отклонён." in text
+    assert "Результаты голосования:" in text
+    assert "За: 0" in text
+    assert "Против: 6" in text

From 2b9306a3fa04521b198f20adcdd40c5700dad4f0 Mon Sep 17 00:00:00 2001
From: Daniil Okhlopkov <5613295+ohld@users.noreply.github.com>
Date: Wed, 20 May 2026 13:54:43 +0300
Subject: [PATCH 2/4] Resolve duplicate memes after OCR and file-id sweeps

---
 src/flows/storage/describe_memes.py        |  14 ++
 src/flows/storage/memes.py                 |   5 +
 src/storage/service.py                     | 106 ++++++++-
 tests/storage/test_duplicate_resolution.py | 265 +++++++++++++++++++++
 4 files changed, 378 insertions(+), 12 deletions(-)
 create mode 100644 tests/storage/test_duplicate_resolution.py

diff --git a/src/flows/storage/describe_memes.py b/src/flows/storage/describe_memes.py
index 02c91e58..af3fab5b 100644
--- a/src/flows/storage/describe_memes.py
+++ b/src/flows/storage/describe_memes.py
@@ -39,6 +39,7 @@
 from src.flows.events import safe_emit
 from src.flows.hooks import notify_telegram_on_failure
 from src.redis import redis_client
+from src.storage.service import find_meme_duplicate, resolve_meme_duplicate
 from src.storage.upload import download_meme_content_from_tg
 
 OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
@@ -242,6 +243,7 @@ async def get_memes_to_describe(limit: int = 30) -> list[dict]:
             M.id,
             M.telegram_file_id,
             M.ocr_result,
+            M.status,
             M.language_code
         FROM meme M
         LEFT JOIN meme_stats MS ON MS.meme_id = M.id
@@ -646,6 +648,18 @@ async def describe_single_meme(meme_row: dict, log, *, deadline: float | None =
 
     update_query = meme.update().where(meme.c.id == meme_id).values(**update_kwargs).returning(meme)
     await fetch_one(update_query)
+
+    if meme_row.get("status") == "ok":
+        duplicate_meme_id = await find_meme_duplicate(meme_id, merged.get("text", ""))
+        if duplicate_meme_id:
+            result = await resolve_meme_duplicate(meme_id, duplicate_meme_id)
+            log.info(
+                "Meme %s resolved as OCR duplicate of %s after describe: %s",
+                meme_id,
+                duplicate_meme_id,
+                result,
+            )
+
     return "ok"
 
 
diff --git a/src/flows/storage/memes.py b/src/flows/storage/memes.py
index 66b9c680..4b4ca049 100644
--- a/src/flows/storage/memes.py
+++ b/src/flows/storage/memes.py
@@ -16,6 +16,7 @@
     get_pending_memes,
     get_unloaded_tg_memes,
     get_unloaded_vk_memes,
+    resolve_all_file_id_duplicates,
     resolve_meme_duplicate,
     update_meme,
     update_meme_status_of_ready_memes,
@@ -220,6 +221,10 @@ async def final_meme_pipeline() -> None:
     # next step of a pipeline
     await update_meme_status_of_ready_memes()
 
+    file_id_duplicates = await resolve_all_file_id_duplicates()
+    if file_id_duplicates["resolved"]:
+        logger.info("Resolved file_id duplicates: %s", file_id_duplicates)
+
     safe_emit(
         "ff.pipeline.final.completed",
         "ff.pipeline.final",
diff --git a/src/storage/service.py b/src/storage/service.py
index d899e44a..990db459 100644
--- a/src/storage/service.py
+++ b/src/storage/service.py
@@ -434,15 +434,16 @@ async def find_meme_duplicate(meme_id: int, imagetext: str) -> int | None:
 async def resolve_meme_duplicate(dupe_id: int, original_id: int) -> dict[str, int]:
     """Mark a meme as duplicate with full cleanup.
 
-    1. Move reactions from dupe → original (skip conflicts)
-    2. Delete remaining reactions on dupe
-    3. Delete meme_stats for dupe
-    4. Set meme status='duplicate', duplicate_of=original_id
-
-    Stats for original will auto-recalculate on next 5-15 min cycle.
-    Returns counts: {moved, conflicts, deleted_stats}.
+    1. Move user reactions from dupe -> original (skip conflicts)
+    2. Move group-chat reactions from dupe -> original (skip conflicts)
+    3. Delete remaining reactions on dupe
+    4. Delete meme_stats for dupe
+    5. Set meme status='duplicate', duplicate_of=original_id
+    6. Refresh basic meme_stats counters for original
+
+    Returns counts for moved/deleted reaction rows.
     """
-    # 1. Move non-conflicting reactions to original
+    # 1. Move non-conflicting user reactions to original
     move_query = text(
         """
         WITH moved AS (
@@ -465,7 +466,31 @@ async def resolve_meme_duplicate(dupe_id: int, original_id: int) -> dict[str, in
     res = await fetch_one(move_query, {"dupe_id": dupe_id, "original_id": original_id})
     moved = res["moved"] if res else 0
 
-    # 2. Delete all reactions remaining on dupe (conflicts + already moved)
+    # 2. Move non-conflicting group-chat reactions to original
+    move_chat_reactions = text(
+        """
+        WITH moved AS (
+            INSERT INTO chat_meme_reaction
+                (chat_id, meme_id, user_id, reaction, reacted_at)
+            SELECT chat_id, :original_id, user_id, reaction, reacted_at
+            FROM chat_meme_reaction
+            WHERE meme_id = :dupe_id
+              AND NOT EXISTS (
+                  SELECT 1 FROM chat_meme_reaction existing
+                  WHERE existing.chat_id = chat_meme_reaction.chat_id
+                    AND existing.user_id = chat_meme_reaction.user_id
+                    AND existing.meme_id = :original_id
+              )
+            ON CONFLICT (chat_id, meme_id, user_id) DO NOTHING
+            RETURNING 1
+        )
+        SELECT count(*) AS moved FROM moved
+    """
+    )
+    res = await fetch_one(move_chat_reactions, {"dupe_id": dupe_id, "original_id": original_id})
+    chat_moved = res["moved"] if res else 0
+
+    # 3. Delete all reactions remaining on dupe (conflicts + already moved)
     delete_reactions = text(
         """
         WITH deleted AS (
@@ -477,13 +502,24 @@ async def resolve_meme_duplicate(dupe_id: int, original_id: int) -> dict[str, in
     res = await fetch_one(delete_reactions, {"dupe_id": dupe_id})
     conflicts = res["conflicts"] if res else 0
 
-    # 3. Delete meme_stats for dupe (stale, will not regenerate since no reactions)
+    delete_chat_reactions = text(
+        """
+        WITH deleted AS (
+            DELETE FROM chat_meme_reaction WHERE meme_id = :dupe_id RETURNING 1
+        )
+        SELECT count(*) AS deleted FROM deleted
+    """
+    )
+    res = await fetch_one(delete_chat_reactions, {"dupe_id": dupe_id})
+    chat_deleted = res["deleted"] if res else 0
+
+    # 4. Delete meme_stats for dupe (stale, will not regenerate since no reactions)
     await execute(
         text("DELETE FROM meme_stats WHERE meme_id = :dupe_id"),
         {"dupe_id": dupe_id},
     )
 
-    # 4. Mark meme as duplicate
+    # 5. Mark meme as duplicate
     await execute(
         text(
             """
@@ -495,7 +531,53 @@ async def resolve_meme_duplicate(dupe_id: int, original_id: int) -> dict[str, in
         {"dupe_id": dupe_id, "original_id": original_id},
     )
 
-    return {"moved": moved, "conflicts": conflicts}
+    # 6. Refresh basic original counters now. Incremental stats only revisits recent
+    # reactions, while duplicate cleanup often moves old rows.
+    await execute(
+        text(
+            """
+            INSERT INTO meme_stats (
+                meme_id, nlikes, ndislikes, nmemes_sent, age_days, sec_to_react, updated_at
+            )
+            SELECT
+                M.id AS meme_id,
+                COUNT(R.*) FILTER (WHERE R.reaction_id = 1) AS nlikes,
+                COUNT(R.*) FILTER (WHERE R.reaction_id = 2) AS ndislikes,
+                COUNT(R.*) AS nmemes_sent,
+                EXTRACT('DAYS' FROM NOW() - M.published_at)::int AS age_days,
+                COALESCE(EXTRACT(
+                    EPOCH FROM percentile_cont(0.5)
+                        WITHIN GROUP (ORDER BY R.reacted_at - R.sent_at)
+                        FILTER (
+                            WHERE R.reacted_at - R.sent_at
+                            BETWEEN '0.5 second'
+                            AND '1 minute'
+                        )
+                ), 99999) AS sec_to_react,
+                NOW() AS updated_at
+            FROM meme M
+            LEFT JOIN user_meme_reaction R
+                ON R.meme_id = M.id
+            WHERE M.id = :original_id
+            GROUP BY M.id
+            ON CONFLICT (meme_id) DO UPDATE SET
+                nlikes = EXCLUDED.nlikes,
+                ndislikes = EXCLUDED.ndislikes,
+                nmemes_sent = EXCLUDED.nmemes_sent,
+                age_days = EXCLUDED.age_days,
+                sec_to_react = EXCLUDED.sec_to_react,
+                updated_at = EXCLUDED.updated_at
+        """
+        ),
+        {"original_id": original_id},
+    )
+
+    return {
+        "moved": moved,
+        "conflicts": conflicts,
+        "chat_moved": chat_moved,
+        "chat_deleted": chat_deleted,
+    }
 
 
 async def resolve_all_file_id_duplicates() -> dict[str, int]:
diff --git a/tests/storage/test_duplicate_resolution.py b/tests/storage/test_duplicate_resolution.py
new file mode 100644
index 00000000..e13f8055
--- /dev/null
+++ b/tests/storage/test_duplicate_resolution.py
@@ -0,0 +1,265 @@
+import logging
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import insert, select
+
+from src.database import chat_meme_reaction, engine, meme, meme_stats, user_meme_reaction
+from src.flows.storage import describe_memes
+from src.flows.storage import memes as storage_memes
+from src.storage.constants import MemeStatus
+from src.storage.service import resolve_all_file_id_duplicates, resolve_meme_duplicate
+from tests.factories import (
+    cleanup_test_data,
+    create_meme,
+    create_meme_source,
+    create_meme_stats,
+    create_reaction,
+    create_user,
+)
+
+
+@pytest_asyncio.fixture()
+async def duplicate_setup():
+    async with engine.connect() as conn:
+        await create_meme_source(conn, id=10001)
+        for user_id in range(10001, 10006):
+            await create_user(conn, id=user_id)
+        await conn.commit()
+
+    yield
+
+    async with engine.connect() as conn:
+        await cleanup_test_data(conn)
+
+
+async def _row(table, **where):
+    async with engine.connect() as conn:
+        query = select(table)
+        for column, value in where.items():
+            query = query.where(getattr(table.c, column) == value)
+        result = await conn.execute(query)
+        row = result.first()
+        return row._asdict() if row else None
+
+
+@pytest.mark.asyncio
+async def test_resolve_meme_duplicate_moves_reactions_and_refreshes_stats(duplicate_setup):
+    async with engine.connect() as conn:
+        await create_meme(conn, id=10001, meme_source_id=10001)
+        await create_meme(conn, id=10002, meme_source_id=10001)
+        await create_meme_stats(conn, meme_id=10001, nlikes=1, ndislikes=1, nmemes_sent=2)
+        await create_meme_stats(conn, meme_id=10002, nlikes=2, ndislikes=1, nmemes_sent=3)
+        await create_reaction(conn, user_id=10001, meme_id=10001, reaction_id=1)
+        await create_reaction(conn, user_id=10002, meme_id=10001, reaction_id=2)
+        await create_reaction(conn, user_id=10002, meme_id=10002, reaction_id=1)
+        await create_reaction(conn, user_id=10003, meme_id=10002, reaction_id=1)
+        await create_reaction(conn, user_id=10004, meme_id=10002, reaction_id=2)
+        await conn.commit()
+
+    result = await resolve_meme_duplicate(10002, 10001)
+
+    assert result["moved"] == 2
+    assert result["conflicts"] == 3
+
+    original_stats = await _row(meme_stats, meme_id=10001)
+    assert original_stats["nlikes"] == 2
+    assert original_stats["ndislikes"] == 2
+    assert original_stats["nmemes_sent"] == 4
+
+    dupe = await _row(meme, id=10002)
+    assert dupe["status"] == MemeStatus.DUPLICATE.value
+    assert dupe["duplicate_of"] == 10001
+    assert await _row(meme_stats, meme_id=10002) is None
+
+    async with engine.connect() as conn:
+        reaction_rows = await conn.execute(
+            select(user_meme_reaction).where(user_meme_reaction.c.meme_id == 10002)
+        )
+        assert reaction_rows.all() == []
+
+
+@pytest.mark.asyncio
+async def test_resolve_all_file_id_duplicates_sweeps_ok_exact_duplicates(duplicate_setup):
+    async with engine.connect() as conn:
+        await create_meme(
+            conn,
+            id=10001,
+            meme_source_id=10001,
+            telegram_file_id="same-file-id",
+        )
+        await create_meme(
+            conn,
+            id=10002,
+            meme_source_id=10001,
+            telegram_file_id="same-file-id",
+        )
+        await create_reaction(conn, user_id=10001, meme_id=10001, reaction_id=1)
+        await create_reaction(conn, user_id=10002, meme_id=10002, reaction_id=2)
+        await conn.commit()
+
+    result = await resolve_all_file_id_duplicates()
+
+    assert result["resolved"] == 1
+    assert result["reactions_moved"] == 1
+
+    original_stats = await _row(meme_stats, meme_id=10001)
+    assert original_stats["nlikes"] == 1
+    assert original_stats["ndislikes"] == 1
+    assert original_stats["nmemes_sent"] == 2
+
+    dupe = await _row(meme, id=10002)
+    assert dupe["status"] == MemeStatus.DUPLICATE.value
+    assert dupe["duplicate_of"] == 10001
+
+
+@pytest.mark.asyncio
+async def test_resolve_meme_duplicate_moves_chat_reactions(duplicate_setup):
+    async with engine.connect() as conn:
+        await create_meme(conn, id=10001, meme_source_id=10001)
+        await create_meme(conn, id=10002, meme_source_id=10001)
+        await conn.execute(
+            insert(chat_meme_reaction),
+            [
+                {"chat_id": 1, "meme_id": 10001, "user_id": 10001, "reaction": 1},
+                {"chat_id": 1, "meme_id": 10002, "user_id": 10001, "reaction": 2},
+                {"chat_id": 1, "meme_id": 10002, "user_id": 10002, "reaction": 1},
+            ],
+        )
+        await conn.commit()
+
+    result = await resolve_meme_duplicate(10002, 10001)
+
+    assert result["chat_moved"] == 1
+    assert result["chat_deleted"] == 2
+
+    async with engine.connect() as conn:
+        original_rows = await conn.execute(
+            select(chat_meme_reaction)
+            .where(chat_meme_reaction.c.meme_id == 10001)
+            .order_by(chat_meme_reaction.c.user_id)
+        )
+        dupe_rows = await conn.execute(
+            select(chat_meme_reaction).where(chat_meme_reaction.c.meme_id == 10002)
+        )
+
+    assert [row._asdict()["user_id"] for row in original_rows.all()] == [10001, 10002]
+    assert dupe_rows.all() == []
+
+
+@pytest.mark.asyncio
+async def test_final_meme_pipeline_runs_file_id_duplicate_sweep(monkeypatch):
+    calls = []
+
+    class FakeLogger:
+        def info(self, *args):
+            calls.append(("log_info", args))
+
+    async def fake_get_pending_memes():
+        calls.append(("get_pending",))
+        return []
+
+    async def fake_update_ready():
+        calls.append(("update_ready",))
+        return []
+
+    async def fake_resolve_all():
+        calls.append(("resolve_all",))
+        return {"resolved": 1, "reactions_moved": 2, "reactions_dropped": 0}
+
+    monkeypatch.setattr(storage_memes, "get_run_logger", lambda: FakeLogger())
+    monkeypatch.setattr(storage_memes, "get_pending_memes", fake_get_pending_memes)
+    monkeypatch.setattr(storage_memes, "update_meme_status_of_ready_memes", fake_update_ready)
+    monkeypatch.setattr(storage_memes, "resolve_all_file_id_duplicates", fake_resolve_all)
+    monkeypatch.setattr(storage_memes, "safe_emit", lambda *args, **kwargs: None)
+
+    await storage_memes.final_meme_pipeline.fn()
+
+    assert ("resolve_all",) in calls
+    assert calls.index(("update_ready",)) < calls.index(("resolve_all",))
+
+
+@pytest.mark.asyncio
+async def test_describe_single_meme_resolves_ok_ocr_duplicate(monkeypatch, duplicate_setup):
+    async with engine.connect() as conn:
+        await create_meme(
+            conn,
+            id=10001,
+            meme_source_id=10001,
+            ocr_result={"text": "same visible meme text", "calculated_at": "2026-05-20T00:00:00Z"},
+        )
+        await create_meme(conn, id=10002, meme_source_id=10001, status=MemeStatus.OK.value)
+        await conn.commit()
+
+    async def fake_download(_file_id):
+        return b"image-bytes"
+
+    async def fake_call(_image_b64, _log, *, deadline=None):
+        return {
+            "ocr_text": "same visible meme text",
+            "description": "A duplicate text meme.",
+            "language": "en",
+            "__model": "test/free",
+        }
+
+    monkeypatch.setattr(describe_memes, "download_meme_content_from_tg", fake_download)
+    monkeypatch.setattr(describe_memes, "call_openrouter_vision", fake_call)
+
+    status = await describe_memes.describe_single_meme(
+        {"id": 10002, "telegram_file_id": "file", "ocr_result": None, "status": "ok"},
+        logging.getLogger(__name__),
+    )
+
+    dupe = await _row(meme, id=10002)
+    assert status == "ok"
+    assert dupe["status"] == MemeStatus.DUPLICATE.value
+    assert dupe["duplicate_of"] == 10001
+
+
+@pytest.mark.asyncio
+async def test_describe_single_meme_does_not_resolve_upload_review_duplicate(
+    monkeypatch, duplicate_setup
+):
+    async with engine.connect() as conn:
+        await create_meme(
+            conn,
+            id=10001,
+            meme_source_id=10001,
+            ocr_result={"text": "same visible meme text", "calculated_at": "2026-05-20T00:00:00Z"},
+        )
+        await create_meme(
+            conn,
+            id=10002,
+            meme_source_id=10001,
+            status=MemeStatus.WAITING_REVIEW.value,
+        )
+        await conn.commit()
+
+    async def fake_download(_file_id):
+        return b"image-bytes"
+
+    async def fake_call(_image_b64, _log, *, deadline=None):
+        return {
+            "ocr_text": "same visible meme text",
+            "description": "A duplicate text meme.",
+            "language": "en",
+            "__model": "test/free",
+        }
+
+    monkeypatch.setattr(describe_memes, "download_meme_content_from_tg", fake_download)
+    monkeypatch.setattr(describe_memes, "call_openrouter_vision", fake_call)
+
+    status = await describe_memes.describe_single_meme(
+        {
+            "id": 10002,
+            "telegram_file_id": "file",
+            "ocr_result": None,
+            "status": MemeStatus.WAITING_REVIEW.value,
+        },
+        logging.getLogger(__name__),
+    )
+
+    meme_row = await _row(meme, id=10002)
+    assert status == "ok"
+    assert meme_row["status"] == MemeStatus.WAITING_REVIEW.value
+    assert meme_row["duplicate_of"] is None

From bc14f8137df078925b27332b86773e961bd933e1 Mon Sep 17 00:00:00 2001
From: Daniil Okhlopkov <5613295+ohld@users.noreply.github.com>
Date: Wed, 20 May 2026 11:08:30 +0000
Subject: [PATCH 3/4] fix: format source voting tests

---
 tests/storage/test_source_voting.py | 9 ++-------
 1 file changed, 2 insertions(+), 7 deletions(-)

diff --git a/tests/storage/test_source_voting.py b/tests/storage/test_source_voting.py
index 0b776eb4..d4bc72ed 100644
--- a/tests/storage/test_source_voting.py
+++ b/tests/storage/test_source_voting.py
@@ -90,9 +90,7 @@ async def conn():
         )
         await conn.execute(
             delete(meme_source_candidate_poll).where(
-                meme_source_candidate_poll.c.candidate_id.in_(
-                    [CANDIDATE_ID, SECOND_CANDIDATE_ID]
-                )
+                meme_source_candidate_poll.c.candidate_id.in_([CANDIDATE_ID, SECOND_CANDIDATE_ID])
             )
         )
         await conn.execute(
@@ -520,10 +518,7 @@ async def test_daily_source_cycle_early_rejects_negative_poll_and_posts_next_can
         result = await advance_daily_source_cycle(bot, now=now)
 
     assert result["closed_poll"]["status"] == POLL_STATUS_REJECTED
-    assert (
-        result["closed_poll"]["poll"]["data"]["close_reason"]
-        == SOURCE_VOTE_EARLY_REJECT_REASON
-    )
+    assert result["closed_poll"]["poll"]["data"]["close_reason"] == SOURCE_VOTE_EARLY_REJECT_REASON
     assert result["new_poll"]["status"] == "posted"
     assert result["new_poll"]["candidate"]["id"] == SECOND_CANDIDATE_ID
     bot.edit_message_text.assert_awaited_once()

From 66beeae589497d7035a6159c73bbae12a320cb8c Mon Sep 17 00:00:00 2001
From: Daniil Okhlopkov <5613295+ohld@users.noreply.github.com>
Date: Wed, 20 May 2026 11:13:51 +0000
Subject: [PATCH 4/4] fix: make source vote early rejection idempotent

---
 src/storage/source_voting.py        |  67 ++++++++++++------
 tests/storage/test_source_voting.py | 101 ++++++++++++++++++++++++++++
 2 files changed, 148 insertions(+), 20 deletions(-)

diff --git a/src/storage/source_voting.py b/src/storage/source_voting.py
index d7dc76f4..7b33cc51 100644
--- a/src/storage/source_voting.py
+++ b/src/storage/source_voting.py
@@ -422,6 +422,7 @@ async def _set_poll_status(
     closed_at: datetime | None = None,
     result_meme_source_id: int | None = None,
     data: dict[str, Any] | None = None,
+    expected_status: str | None = None,
 ) -> dict[str, Any] | None:
     values: dict[str, Any] = {"status": status, "updated_at": _utcnow()}
     if closed_at is not None:
@@ -430,12 +431,12 @@ async def _set_poll_status(
         values["result_meme_source_id"] = result_meme_source_id
     if data is not None:
         values["data"] = data
-    return await fetch_one(
-        meme_source_candidate_poll.update()
-        .where(meme_source_candidate_poll.c.id == poll_id)
-        .values(**values)
-        .returning(meme_source_candidate_poll)
+    update_stmt = meme_source_candidate_poll.update().where(
+        meme_source_candidate_poll.c.id == poll_id
     )
+    if expected_status is not None:
+        update_stmt = update_stmt.where(meme_source_candidate_poll.c.status == expected_status)
+    return await fetch_one(update_stmt.values(**values).returning(meme_source_candidate_poll))
 
 
 async def _append_source_vote_metadata(
@@ -563,6 +564,21 @@ async def close_source_candidate_poll(
         result_source = await enable_passed_source_poll(poll, counts, now)
     else:
         status = POLL_STATUS_REJECTED
+        early_reject_close = close_reason == SOURCE_VOTE_EARLY_REJECT_REASON
+        if early_reject_close:
+            claimed_poll = await _set_poll_status(
+                poll_id,
+                status=status,
+                closed_at=now,
+                data=poll_data,
+                expected_status=POLL_STATUS_OPEN,
+            )
+            if claimed_poll is None:
+                return {
+                    "status": "already_closed",
+                    "poll": await get_source_candidate_poll(poll_id) or poll,
+                }
+            poll = claimed_poll
         if poll["prepared_meme_source_id"]:
             await advance_meme_source(
                 poll["prepared_meme_source_id"],
@@ -587,13 +603,21 @@ async def close_source_candidate_poll(
             ),
         )
 
-    updated_poll = await _set_poll_status(
-        poll_id,
-        status=status,
-        closed_at=now,
-        result_meme_source_id=None if result_source is None else result_source["id"],
-        data=poll_data,
-    )
+    updated_poll = poll
+    if status != POLL_STATUS_REJECTED or close_reason != SOURCE_VOTE_EARLY_REJECT_REASON:
+        updated_poll = await _set_poll_status(
+            poll_id,
+            status=status,
+            closed_at=now,
+            result_meme_source_id=None if result_source is None else result_source["id"],
+            data=poll_data,
+            expected_status=POLL_STATUS_OPEN,
+        )
+        if updated_poll is None:
+            return {
+                "status": "already_closed",
+                "poll": await get_source_candidate_poll(poll_id) or poll,
+            }
     return {
         "status": status,
         "poll": updated_poll,
@@ -758,6 +782,7 @@ async def create_source_candidate_poll(
                 "closes_at": now + SOURCE_VOTE_WINDOW,
             }
         )
+        .on_conflict_do_nothing()
         .returning(meme_source_candidate_poll)
     )
     return await fetch_one(stmt)
@@ -869,14 +894,11 @@ async def post_new_source_candidate_poll(
     now: datetime | None = None,
 ) -> dict[str, Any]:
     now = now or _utcnow()
-    draft_poll = await fetch_one(
-        select(meme_source_candidate_poll)
-        .where(meme_source_candidate_poll.c.status == POLL_STATUS_DRAFT)
-        .order_by(meme_source_candidate_poll.c.created_at.asc())
-        .limit(1)
-    )
-    if draft_poll is not None:
-        return await post_source_candidate_poll_message(bot, draft_poll, now=now)
+    active_poll = await get_active_source_candidate_poll()
+    if active_poll is not None:
+        if active_poll["status"] == POLL_STATUS_DRAFT:
+            return await post_source_candidate_poll_message(bot, active_poll, now=now)
+        return {"status": "active_poll_exists", "poll": active_poll}
 
     candidate = await select_daily_source_candidate()
     if candidate is None:
@@ -895,6 +917,11 @@ async def post_new_source_candidate_poll(
         now=now,
     )
     if poll is None:
+        active_poll = await get_active_source_candidate_poll()
+        if active_poll is not None:
+            if active_poll["status"] == POLL_STATUS_DRAFT:
+                return await post_source_candidate_poll_message(bot, active_poll, now=now)
+            return {"status": "active_poll_exists", "poll": active_poll}
         return {"status": "poll_create_failed", "candidate": candidate}
 
     return await post_source_candidate_poll_message(bot, poll, now=now, prepared=prepared)
diff --git a/tests/storage/test_source_voting.py b/tests/storage/test_source_voting.py
index d4bc72ed..03f49c15 100644
--- a/tests/storage/test_source_voting.py
+++ b/tests/storage/test_source_voting.py
@@ -1,3 +1,4 @@
+import asyncio
 from datetime import datetime, timedelta
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, patch
@@ -310,6 +311,78 @@ async def test_source_candidate_vote_early_rejects_clear_negative_poll(
     )
 
 
+@pytest.mark.asyncio
+async def test_early_reject_close_has_single_winner_under_concurrent_calls(
+    conn: AsyncConnection,
+):
+    now = datetime.utcnow()
+    await _create_candidate(conn)
+    await conn.commit()
+    prepared = await prepare_source_candidate(CANDIDATE_ID, posts=[_post(4017)])
+    poll = await create_source_candidate_poll(
+        CANDIDATE_ID,
+        prepared_meme_source_id=prepared["source"]["id"],
+        chat_id=TELEGRAM_MODERATOR_CHAT_ID,
+        now=now,
+    )
+    await conn.execute(
+        meme_source_candidate_poll.update()
+        .where(meme_source_candidate_poll.c.id == poll["id"])
+        .values(
+            status=POLL_STATUS_OPEN,
+            message_id=133,
+            opened_at=now - timedelta(hours=2),
+            closes_at=now + timedelta(hours=22),
+        )
+    )
+    await conn.execute(
+        insert(meme_source_candidate_vote),
+        [
+            {
+                "poll_id": poll["id"],
+                "user_id": TEST_ID_START + 30 + offset,
+                "vote": VOTE_SKIP_SOURCE,
+            }
+            for offset in range(1, 7)
+        ],
+    )
+    await conn.commit()
+
+    advance_calls = 0
+    entered_advance = asyncio.Event()
+    release_advance = asyncio.Event()
+
+    async def blocking_advance_meme_source(*args, **kwargs):
+        nonlocal advance_calls
+        advance_calls += 1
+        entered_advance.set()
+        await release_advance.wait()
+
+    with patch("src.storage.source_voting.advance_meme_source", new=blocking_advance_meme_source):
+        first = asyncio.create_task(
+            close_source_candidate_poll(
+                poll["id"],
+                now=now,
+                close_reason=SOURCE_VOTE_EARLY_REJECT_REASON,
+            )
+        )
+        await asyncio.wait_for(entered_advance.wait(), timeout=1)
+
+        second = asyncio.create_task(
+            close_source_candidate_poll(
+                poll["id"],
+                now=now,
+                close_reason=SOURCE_VOTE_EARLY_REJECT_REASON,
+            )
+        )
+        await asyncio.sleep(0.05)
+        release_advance.set()
+        results = await asyncio.gather(first, second)
+
+    assert advance_calls == 1
+    assert {result["status"] for result in results} == {POLL_STATUS_REJECTED, "already_closed"}
+
+
 @pytest.mark.asyncio
 async def test_source_candidate_vote_does_not_early_reject_before_min_open_time(
     conn: AsyncConnection,
@@ -532,6 +605,34 @@ async def test_daily_source_cycle_early_rejects_negative_poll_and_posts_next_can
     bot.send_message.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+async def test_post_new_source_candidate_poll_is_idempotent_when_active_poll_exists(
+    conn: AsyncConnection,
+):
+    now = datetime.utcnow()
+    await _create_candidate(conn)
+    await _create_candidate(
+        conn,
+        candidate_id=SECOND_CANDIDATE_ID,
+        url=SECOND_CANDIDATE_URL,
+        times_forwarded=4,
+    )
+    await conn.commit()
+
+    bot = SimpleNamespace(send_message=AsyncMock(return_value=SimpleNamespace(message_id=560)))
+    with patch(
+        "src.storage.source_voting.fetch_telegram_candidate_posts",
+        new=AsyncMock(return_value=[_post(4018)]),
+    ):
+        first = await post_new_source_candidate_poll(bot, now=now)
+        second = await post_new_source_candidate_poll(bot, now=now)
+
+    assert first["status"] == "posted"
+    assert second["status"] == "active_poll_exists"
+    assert second["poll"]["id"] == first["poll"]["id"]
+    bot.send_message.assert_awaited_once()
+
+
 @pytest.mark.asyncio
 async def test_opening_draft_poll_refreshes_vote_window(conn: AsyncConnection):
     now = datetime.utcnow()