"""Purge configurable-tile listings superseded by per-variant rows.

Per-variant emission (0.22.0) gives a configurable tile one listing per priced
size, keyed `<sku>::<size_ml>`. The pre-variant listing for the same tile kept
its plain `<sku>-P` key, so it was never updated again: it lingered as a ghost
carrying the tile's old ambiguous price, and where it had no barcode it stood
as a duplicate product beside the real per-size family (e.g. a "70cl" ghost
next to the true 750ml row).

A plain `-P` listing is deleted only where a `::` sibling exists for the same
tile at the same location — a tile whose page could not be re-read keeps its
old listing, dated honestly. Products left with no listings at all go too.

Revision ID: d6e7f8a9b0c1
Revises: c5d6e7f8a9b0
Create Date: 2026-08-25
"""

from alembic import op
from sqlalchemy import text

revision = "d6e7f8a9b0c1"
down_revision = "c5d6e7f8a9b0"
branch_labels = None
depends_on = None

GHOSTS = (
    "SELECT li.id FROM listings li "
    "WHERE li.source_sku LIKE '%-P' AND li.source_sku NOT LIKE '%::%' "
    "AND EXISTS (SELECT 1 FROM listings li2 "
    "WHERE li2.location_id = li.location_id "
    "AND li2.source_sku LIKE li.source_sku || '::%')"
)


def upgrade() -> None:
    conn = op.get_bind()
    touched = [
        row[0]
        for row in conn.execute(
            text(f"SELECT DISTINCT product_id FROM listings WHERE id IN ({GHOSTS})")
        ).all()
    ]
    if not touched:
        return
    conn.execute(
        text(f"DELETE FROM price_observations WHERE listing_id IN ({GHOSTS})")
    )
    conn.execute(text(f"DELETE FROM listings WHERE id IN ({GHOSTS})"))
    empty = [
        row[0]
        for row in conn.execute(
            text(
                "SELECT p.id FROM products p WHERE p.id = ANY(:ids) "
                "AND NOT EXISTS (SELECT 1 FROM listings l WHERE l.product_id = p.id)"
            ),
            {"ids": touched},
        ).all()
    ]
    if empty:
        conn.execute(text("DELETE FROM awards WHERE product_id = ANY(:ids)"), {"ids": empty})
        conn.execute(text("DELETE FROM products WHERE id = ANY(:ids)"), {"ids": empty})


def downgrade() -> None:
    pass