"""drop scheduled message duplicates

Revision ID: e1c7a4b62d90
Revises: d4a7e2b91f38
Create Date: 2026-08-13 10:00:00.000000

"""

from collections.abc import Sequence

from alembic import op

revision: str = "e1c7a4b62d90"
down_revision: str | None = "d4a7e2b91f38"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

_STAGE = """
CREATE TEMP TABLE scheduled_keys ON COMMIT DROP AS
SELECT DISTINCT s.account_id, s.chat_id, s.message_id
FROM messages s
WHERE s.raw->>'scheduled' = 'true'
  AND NOT EXISTS (
    SELECT 1 FROM messages r
    WHERE r.account_id = s.account_id AND r.chat_id = s.chat_id
      AND r.message_id = s.message_id
      AND r.raw->>'scheduled' IS DISTINCT FROM 'true'
  )
"""

_DELETE_CHILD = """
DELETE FROM {table} t
USING scheduled_keys k
WHERE t.account_id = k.account_id AND t.chat_id = k.chat_id
  AND t.message_id = k.message_id
"""

_DELETE_MESSAGES = """
DELETE FROM messages m
USING scheduled_keys k
WHERE m.account_id = k.account_id AND m.chat_id = k.chat_id
  AND m.message_id = k.message_id AND m.raw->>'scheduled' = 'true'
"""

_RECOUNT = """
UPDATE chat_stats cs
SET message_count = fresh.message_count
FROM (
    SELECT k.account_id, k.chat_id,
           (SELECT count(*) FROM messages m
            WHERE m.account_id = k.account_id AND m.chat_id = k.chat_id)
           AS message_count
    FROM (SELECT DISTINCT account_id, chat_id FROM scheduled_keys) k
) fresh
WHERE cs.account_id = fresh.account_id AND cs.chat_id = fresh.chat_id
"""

_CHILD_TABLES = ("media", "media_versions", "message_versions", "links", "callbacks")

_ALLOW_DECOMPRESSION = (
    "SET LOCAL timescaledb.max_tuples_decompressed_per_dml_transaction = 0"
)


def upgrade() -> None:
    op.execute(_ALLOW_DECOMPRESSION)
    op.execute(_STAGE)
    for table in _CHILD_TABLES:
        op.execute(_DELETE_CHILD.format(table=table))
    op.execute(_DELETE_MESSAGES)
    op.execute(_RECOUNT)


def downgrade() -> None:
    pass