Messenger/alembic/versions/0026_deduplicate_saved_chats.py

"""deduplicate saved chats per user

Revision ID: 0026_deduplicate_saved_chats
Revises: 0025_user_twofa_recovery_codes
Create Date: 2026-03-10 00:25:00.000000
"""

from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


revision: str = "0026_deduplicate_saved_chats"
down_revision: Union[str, Sequence[str], None] = "0025_user_twofa_recovery_codes"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    bind = op.get_bind()

    duplicate_user_ids = [
        int(row[0])
        for row in bind.execute(
            sa.text(
                """
                SELECT cm.user_id
                FROM chat_members cm
                JOIN chats c ON c.id = cm.chat_id
                WHERE c.is_saved IS TRUE
                GROUP BY cm.user_id
                HAVING COUNT(*) > 1
                """
            )
        ).fetchall()
    ]

    for user_id in duplicate_user_ids:
        saved_chat_ids = [
            int(row[0])
            for row in bind.execute(
                sa.text(
                    """
                    SELECT c.id
                    FROM chats c
                    JOIN chat_members cm ON cm.chat_id = c.id
                    WHERE c.is_saved IS TRUE
                      AND cm.user_id = :user_id
                    ORDER BY c.id ASC
                    """
                ),
                {"user_id": user_id},
            ).fetchall()
        ]
        if len(saved_chat_ids) <= 1:
            continue

        keep_chat_id = saved_chat_ids[0]
        duplicate_chat_ids = saved_chat_ids[1:]

        for duplicate_chat_id in duplicate_chat_ids:
            bind.execute(
                sa.text(
                    """
                    UPDATE chats keep
                    SET pinned_message_id = COALESCE(
                        keep.pinned_message_id,
                        (SELECT pinned_message_id FROM chats WHERE id = :dup_chat_id)
                    )
                    WHERE keep.id = :keep_chat_id
                    """
                ),
                {"keep_chat_id": keep_chat_id, "dup_chat_id": duplicate_chat_id},
            )

            bind.execute(
                sa.text(
                    """
                    UPDATE messages
                    SET chat_id = :keep_chat_id
                    WHERE chat_id = :dup_chat_id
                    """
                ),
                {"keep_chat_id": keep_chat_id, "dup_chat_id": duplicate_chat_id},
            )

            bind.execute(
                sa.text(
                    """
                    INSERT INTO message_receipts (chat_id, user_id, last_delivered_message_id, last_read_message_id, updated_at)
                    SELECT :keep_chat_id, mr.user_id, mr.last_delivered_message_id, mr.last_read_message_id, mr.updated_at
                    FROM message_receipts mr
                    WHERE mr.chat_id = :dup_chat_id
                    ON CONFLICT (chat_id, user_id) DO UPDATE
                    SET last_delivered_message_id = GREATEST(
                        COALESCE(message_receipts.last_delivered_message_id, 0),
                        COALESCE(EXCLUDED.last_delivered_message_id, 0)
                    ),
                        last_read_message_id = GREATEST(
                            COALESCE(message_receipts.last_read_message_id, 0),
                            COALESCE(EXCLUDED.last_read_message_id, 0)
                        ),
                        updated_at = GREATEST(message_receipts.updated_at, EXCLUDED.updated_at)
                    """
                ),
                {"keep_chat_id": keep_chat_id, "dup_chat_id": duplicate_chat_id},
            )
            bind.execute(
                sa.text("DELETE FROM message_receipts WHERE chat_id = :dup_chat_id"),
                {"dup_chat_id": duplicate_chat_id},
            )

            bind.execute(
                sa.text(
                    """
                    INSERT INTO chat_notification_settings (chat_id, user_id, muted, updated_at)
                    SELECT :keep_chat_id, cns.user_id, cns.muted, cns.updated_at
                    FROM chat_notification_settings cns
                    WHERE cns.chat_id = :dup_chat_id
                    ON CONFLICT (chat_id, user_id) DO UPDATE
                    SET muted = chat_notification_settings.muted OR EXCLUDED.muted,
                        updated_at = GREATEST(chat_notification_settings.updated_at, EXCLUDED.updated_at)
                    """
                ),
                {"keep_chat_id": keep_chat_id, "dup_chat_id": duplicate_chat_id},
            )
            bind.execute(
                sa.text("DELETE FROM chat_notification_settings WHERE chat_id = :dup_chat_id"),
                {"dup_chat_id": duplicate_chat_id},
            )

            bind.execute(
                sa.text(
                    """
                    INSERT INTO chat_user_settings (chat_id, user_id, archived, pinned, pinned_at, updated_at)
                    SELECT :keep_chat_id, cus.user_id, cus.archived, cus.pinned, cus.pinned_at, cus.updated_at
                    FROM chat_user_settings cus
                    WHERE cus.chat_id = :dup_chat_id
                    ON CONFLICT (chat_id, user_id) DO UPDATE
                    SET archived = chat_user_settings.archived OR EXCLUDED.archived,
                        pinned = chat_user_settings.pinned OR EXCLUDED.pinned,
                        pinned_at = CASE
                            WHEN chat_user_settings.pinned_at IS NULL THEN EXCLUDED.pinned_at
                            WHEN EXCLUDED.pinned_at IS NULL THEN chat_user_settings.pinned_at
                            ELSE GREATEST(chat_user_settings.pinned_at, EXCLUDED.pinned_at)
                        END,
                        updated_at = GREATEST(chat_user_settings.updated_at, EXCLUDED.updated_at)
                    """
                ),
                {"keep_chat_id": keep_chat_id, "dup_chat_id": duplicate_chat_id},
            )
            bind.execute(
                sa.text("DELETE FROM chat_user_settings WHERE chat_id = :dup_chat_id"),
                {"dup_chat_id": duplicate_chat_id},
            )

            bind.execute(
                sa.text(
                    """
                    INSERT INTO message_idempotency_keys (chat_id, sender_id, client_message_id, message_id, created_at)
                    SELECT :keep_chat_id, mik.sender_id, mik.client_message_id, mik.message_id, mik.created_at
                    FROM message_idempotency_keys mik
                    WHERE mik.chat_id = :dup_chat_id
                    ON CONFLICT (chat_id, sender_id, client_message_id) DO NOTHING
                    """
                ),
                {"keep_chat_id": keep_chat_id, "dup_chat_id": duplicate_chat_id},
            )
            bind.execute(
                sa.text("DELETE FROM message_idempotency_keys WHERE chat_id = :dup_chat_id"),
                {"dup_chat_id": duplicate_chat_id},
            )

            bind.execute(
                sa.text(
                    """
                    INSERT INTO chat_members (chat_id, user_id, role, joined_at)
                    SELECT :keep_chat_id, cm.user_id, cm.role, cm.joined_at
                    FROM chat_members cm
                    WHERE cm.chat_id = :dup_chat_id
                    ON CONFLICT (chat_id, user_id) DO UPDATE
                    SET role = CASE
                        WHEN chat_members.role = 'OWNER' OR EXCLUDED.role = 'OWNER' THEN 'OWNER'::chatmemberrole
                        WHEN chat_members.role = 'ADMIN' OR EXCLUDED.role = 'ADMIN' THEN 'ADMIN'::chatmemberrole
                        ELSE 'MEMBER'::chatmemberrole
                    END,
                        joined_at = LEAST(chat_members.joined_at, EXCLUDED.joined_at)
                    """
                ),
                {"keep_chat_id": keep_chat_id, "dup_chat_id": duplicate_chat_id},
            )
            bind.execute(
                sa.text("DELETE FROM chat_members WHERE chat_id = :dup_chat_id"),
                {"dup_chat_id": duplicate_chat_id},
            )

            bind.execute(
                sa.text("DELETE FROM chat_bans WHERE chat_id = :dup_chat_id"),
                {"dup_chat_id": duplicate_chat_id},
            )
            bind.execute(
                sa.text("DELETE FROM chat_invite_links WHERE chat_id = :dup_chat_id"),
                {"dup_chat_id": duplicate_chat_id},
            )

            bind.execute(
                sa.text("DELETE FROM chats WHERE id = :dup_chat_id"),
                {"dup_chat_id": duplicate_chat_id},
            )


def downgrade() -> None:
    # data-cleanup migration; no reversible schema changes
    pass