"""source-collapse: one Source per (artist, platform) — consolidate junk per-post Sources

Revision ID: 0022
Revises: 0021
Create Date: 2026-05-26

Closes the operator-flagged 2026-05-26 issue where the filesystem importer
called _find_or_create_source(url=sd.post_url), creating one Source row per
imported post URL. Operator's Atole artist had 406 Source rows where there
should have been 1 (the /cw/Atole subscription Source).

Source represents a subscription feed (one per artist+platform — the
gallery-dl URL polled by the FC-3 downloader). Posts hang off it. The
filesystem importer was misusing Source as a per-post key.

Migration steps per (artist_id, platform) group with >1 Source:
  1. Pick canonical — prefer a URL NOT matching '/posts/<id>$' (real
     campaign URL like /cw/Atole); else min(id).
  2. PRE-merge any Posts under non-canonical sources whose
     external_post_id ALREADY exists under the canonical source. (Same
     gallery-dl post imported via two different sidecar paths can plant
     two Post rows with identical external_post_id under different
     Sources for the same artist.) Repoint ImageProvenance +
     ImageRecord.primary_post_id to the canonical-side Post, dedupe
     ImageProvenance against alembic 0021's uq, then delete the
     non-canonical-side Post. This MUST happen before step 3 — Postgres
     fires uq_post_source_external_id row-by-row during the bulk UPDATE
     and the merge-after-reparent ordering 500s on first collision
     (operator-hit during v26.05.26.1 deploy, 2026-05-26).
  3. Reparent remaining Posts onto canonical (no collisions possible now).
  4. Reparent ImageProvenance.source_id off the non-canonical sources.
  5. Delete the orphan Source rows.
  6. If the canonical Source's URL still looks like a per-post URL (no
     campaign URL existed among candidates), rewrite it to
     'sidecar:<platform>:<artist_slug>' so the artist detail page shows
     something readable.
"""
from typing import Sequence, Union

from alembic import op
from sqlalchemy import text

revision: str = "0022"
down_revision: Union[str, None] = "0021"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None

_POST_URL_RE = r"/posts/[^/]+$"


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

    # Find (artist_id, platform) groups with > 1 Source row.
    groups = conn.execute(text("""
        SELECT artist_id, platform
        FROM source
        GROUP BY artist_id, platform
        HAVING COUNT(*) > 1
    """)).fetchall()

    for artist_id, platform in groups:
        rows = conn.execute(
            text("""
                SELECT id, url FROM source
                WHERE artist_id = :a AND platform = :p
                ORDER BY id ASC
            """),
            {"a": artist_id, "p": platform},
        ).fetchall()

        # Canonical: first row whose URL doesn't look like a per-post URL;
        # else min(id).
        canonical_id = None
        for sid, url in rows:
            if not _matches_post_url(url):
                canonical_id = sid
                break
        if canonical_id is None:
            canonical_id = rows[0][0]

        other_ids = [sid for sid, _ in rows if sid != canonical_id]
        if not other_ids:
            continue

        # STEP 2: PRE-merge ALL Posts with duplicate external_post_id
        # across the entire (canonical + others) group, BEFORE the bulk
        # reparent. Two cases must both be handled:
        #   (A) canonical has Post X with epid=N; an "other" source has
        #       Post Y with epid=N → after bulk UPDATE, (canonical, N)
        #       collides with itself.
        #   (B) two different "other" sources each have a Post with
        #       epid=N; canonical has none → after bulk UPDATE, both
        #       are repointed to (canonical, N) and the second collides.
        # The earlier version of this migration only handled (A); the
        # operator's deploy 2026-05-26 tripped (B) at line 139.
        # Fix: group ALL Posts in the (artist, platform) by epid; for
        # any group with count>1, pick the keep (prefer one already
        # under canonical; else lowest id) and merge the rest into it.
        all_posts = conn.execute(
            text("""
                SELECT external_post_id, id, source_id
                FROM post
                WHERE source_id = :canonical OR source_id = ANY(:others)
                ORDER BY external_post_id, id
            """),
            {"canonical": canonical_id, "others": other_ids},
        ).fetchall()
        by_epid: dict = {}
        for epid, post_id, src_id in all_posts:
            by_epid.setdefault(epid, []).append((post_id, src_id))
        for _epid, posts in by_epid.items():
            if len(posts) <= 1:
                continue
            # Prefer a Post already under canonical as the keep.
            canonical_posts = [p for p in posts if p[1] == canonical_id]
            if canonical_posts:
                keep_id = canonical_posts[0][0]
            else:
                keep_id = posts[0][0]  # already sorted by id ASC
            drop_ids = [p[0] for p in posts if p[0] != keep_id]
            for drop_id in drop_ids:
                # Pre-delete image_provenance rows under drop_ whose
                # image_record_id ALREADY has a provenance under keep —
                # the UPDATE below would otherwise repoint them and
                # trip uq_image_provenance_image_post (alembic 0021)
                # row-by-row before any after-the-fact dedupe could
                # run. Operator's v26.05.26.3 deploy 2026-05-26 tripped
                # this at line 123.
                conn.execute(
                    text("""
                        DELETE FROM image_provenance
                        WHERE post_id = :drop_
                          AND image_record_id IN (
                              SELECT image_record_id FROM image_provenance
                              WHERE post_id = :keep
                          )
                    """),
                    {"keep": keep_id, "drop_": drop_id},
                )
                # Now safe to repoint the survivors.
                conn.execute(
                    text("""
                        UPDATE image_provenance SET post_id = :keep
                        WHERE post_id = :drop_
                    """),
                    {"keep": keep_id, "drop_": drop_id},
                )
                conn.execute(
                    text("""
                        UPDATE image_record SET primary_post_id = :keep
                        WHERE primary_post_id = :drop_
                    """),
                    {"keep": keep_id, "drop_": drop_id},
                )
                conn.execute(
                    text("DELETE FROM post WHERE id = :drop_"),
                    {"drop_": drop_id},
                )

        # STEP 3: Bulk reparent the remaining Posts off the other
        # Sources. After step 2, no collisions on
        # (canonical, external_post_id) are possible.
        conn.execute(
            text("""
                UPDATE post SET source_id = :canonical
                WHERE source_id = ANY(:others)
            """),
            {"canonical": canonical_id, "others": other_ids},
        )

        # STEP 4: Reparent ImageProvenance.source_id (denormalized FK).
        # No UNIQUE on source_id; safe bulk update.
        conn.execute(
            text("""
                UPDATE image_provenance SET source_id = :canonical
                WHERE source_id = ANY(:others)
            """),
            {"canonical": canonical_id, "others": other_ids},
        )

        # STEP 5: Drop the orphan Sources.
        conn.execute(
            text("DELETE FROM source WHERE id = ANY(:others)"),
            {"others": other_ids},
        )

        # If the canonical's URL still looks per-post (no campaign URL
        # existed among the candidates), rewrite to a synthetic anchor so
        # the artist detail page renders something readable.
        canonical_url = conn.execute(
            text("SELECT url FROM source WHERE id = :id"),
            {"id": canonical_id},
        ).scalar_one()
        if _matches_post_url(canonical_url):
            slug = conn.execute(
                text("SELECT slug FROM artist WHERE id = :id"),
                {"id": artist_id},
            ).scalar_one()
            conn.execute(
                text("""
                    UPDATE source
                    SET url = :new_url, enabled = false
                    WHERE id = :id
                """),
                {
                    "id": canonical_id,
                    "new_url": f"sidecar:{platform}:{slug}",
                },
            )


def downgrade() -> None:
    # Lossy migration — orphan Sources deleted, Posts reparented, Posts
    # merged. No safe downgrade. If you need to roll back the schema
    # invariant, fork from 0021 and re-run filesystem imports.
    pass


def _matches_post_url(url: str) -> bool:
    """True if url ends with /posts/<token> (gallery-dl-style per-post URL)."""
    import re
    return bool(re.search(_POST_URL_RE, url or ""))