"""Read queries behind the catalog API.

Routers stay thin; the shaping of a product plus its latest price at every
location lives here. Every query scopes its shops through `publishable(db)`:
visible locations only, minus any shop whose source verify has blocked
(Decision 10), so one rule decides what the public site shows.
"""

from collections import Counter
from datetime import datetime

from sqlalchemy import Select, and_, case, distinct, func, or_, select, tuple_
from sqlalchemy.orm import Session, aliased

from app.models import Award, Brand, Listing, Location, PriceObservation, Product, Retailer, Source
from app.models.hubs import (
    AirportCategoryDetail,
    AirportDetail,
    AirportShop,
    AirportSummary,
    BrandAirport,
    BrandDetail,
    BrandSummary,
    CategoryPageLink,
    DatasetFacts,
)
from app.services import airport_featured, airport_guides, award_picker, editorial, taxonomy, verify
from app.services.collectors.registry import COLLECTORS
from app.services.featured import SavingRecord, pick_featured
from app.services.urls import airport_category_path, airport_path, brand_page_path, category_slug
from app.models.schemas import (
    TopAward,
    AwardOut,
    CategoryCount,
    PriceOut,
    ProductDetail,
    ProductSuggestion,
    ProductSummary,
    ShopPrice,
)


def _latest_observation_subquery(
    product_ids: list[int] | None = None, location_ids: list[int] | None = None
):
    """The most recent observation per listing.

    `product_ids` confines the ranking to those products' listings, `location_ids`
    to those shops' listings (a route). The answer for them is the same; the cost
    is not: ranking every observation in the catalogue to price the eight featured
    cards took about 450 of the home page's 600 ms on staging (11 Sep), a
    two-airport savings page 0.7 to 1.2 s, and both grow with every collection."""
    ranked = select(
        PriceObservation.id.label("obs_id"),
        PriceObservation.listing_id.label("listing_id"),
        func.row_number()
        .over(
            partition_by=PriceObservation.listing_id,
            # id breaks same-timestamp ties, else "latest" is nondeterministic.
            order_by=(PriceObservation.observed_at.desc(), PriceObservation.id.desc()),
        )
        .label("rn"),
    )
    if product_ids is not None or location_ids is not None:
        listings = select(Listing.id)
        if product_ids is not None:
            listings = listings.where(Listing.product_id.in_(product_ids))
        if location_ids is not None:
            listings = listings.where(Listing.location_id.in_(location_ids))
        ranked = ranked.where(PriceObservation.listing_id.in_(listings))
    ranked = ranked.subquery()
    return select(ranked.c.obs_id).where(ranked.c.rn == 1).subquery()


def _available():
    return PriceObservation.in_stock.isnot(False)


def blocked_location_ids(db: Session | None) -> list[int]:
    """The shops whose prices the site must not show as current (Decision 10).

    An uncleared correctness failure (MISMATCH_*, PARSE_FAIL: `verify.blocked_sources`)
    blocks publication of its source until a human clears the check. The block is a
    fact about that source's parser on that host, so it covers every shop the source
    writes to: the failing checks' own listings name the shops exactly, and the
    collector's declared locations cover a platform collector that serves several.
    Nothing else hides a row: a retailer's refusal (BLOCKED) and the freshness findings
    (PRICE_MOVED, GONE) keep their prices visible with their date. Empty when nothing
    is blocked, the common case, at the cost of one small query; memoised on the
    session so one request asks once and every query in it agrees.
    """
    if db is None or not hasattr(db, "scalars"):
        return []
    info = getattr(db, "info", None)
    cache = info if isinstance(info, dict) else None
    if cache is not None and "blocked_location_ids" in cache:
        return cache["blocked_location_ids"]
    blocked = verify.blocked_sources(db)
    ids: set[int] = set()
    if blocked:
        listing_ids = {check.listing_id for checks in blocked.values() for check in checks}
        ids.update(db.scalars(select(Listing.location_id).where(Listing.id.in_(listing_ids))))
        ids.update(_declared_location_ids(db, list(blocked)))
    out = sorted(ids)
    if cache is not None:
        cache["blocked_location_ids"] = out
    return out


def _declared_location_ids(db: Session, source_ids: list[int]) -> list[int]:
    """The locations these sources' collectors declare, by retailer slug and code (the
    pair `ingest.ensure_retailer_and_locations` creates them under)."""
    wanted: list[tuple[str, str]] = []
    for slug in db.scalars(select(Source.slug).where(Source.id.in_(source_ids))):
        collector = COLLECTORS.get(slug)
        if collector is None:
            continue
        wanted.extend((collector.retailer_slug, spec.code) for spec in collector.locations())
    if not wanted:
        return []
    return list(
        db.scalars(
            select(Location.id)
            .join(Retailer, Retailer.id == Location.retailer_id)
            .where(tuple_(Retailer.slug, Location.code).in_(wanted))
        )
    )


def publishable(db: Session | None):
    """The locations the site shows: visible, and not publication-blocked.

    Hidden locations exist nowhere on the site; their collectors still run. A shop
    blocked by verify's tripwire is hidden the same way, everywhere at once, until a
    human clears the check (`blocked_location_ids`); the site never shows its prices
    as current in one place and not another.
    """
    condition = Location.visible.is_(True)
    blocked = blocked_location_ids(db)
    if blocked:
        condition = and_(condition, Location.id.not_in(blocked))
    return condition


def shown_category():
    """The storefront's category rule (taxonomy.HIDDEN_CATEGORIES): a hidden category's
    products are listed, counted and mapped nowhere, like a hidden airport's prices."""
    return or_(Product.category.is_(None), Product.category.not_in(taxonomy.HIDDEN_CATEGORIES))


def _like_pattern(query: str) -> str:
    """Escape LIKE wildcards so a shopper's % or _ is a character, not an operator."""
    return "%" + query.strip().replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + "%"


def _prefix_pattern(query: str) -> str:
    """`_like_pattern` anchored at the start: the words begin the text."""
    return _like_pattern(query)[1:]


#: Under this many characters the header search suggests nothing ("a" is everything).
SUGGEST_MIN_CHARS = 2


def suggest_products(db: Session, query: str, limit: int = 6) -> list[ProductSuggestion]:
    """The header search's product suggestions while a shopper types.

    The same match as `list_products` (the words anywhere in the name or the brand,
    accents folded, over the visible listings), so "see all results" holds what
    was suggested (unless the shopper's airports narrow that page). Ranked for typing: names that begin with
    the words, then brands that do, then the most widely stocked, then the shortest
    name. No prices: `list_products` spends about half a second per search on the
    price aggregation over the whole catalogue (measured on staging, 11 Sep), too
    slow per keystroke; this answers in about a tenth of that."""
    q = query.strip()
    if len(q) < SUGGEST_MIN_CHARS:
        return []
    name = func.unaccent(Product.name)
    brand = func.unaccent(func.coalesce(Product.brand, ""))
    anywhere = func.unaccent(_like_pattern(q))
    starting = func.unaccent(_prefix_pattern(q))
    shops = func.count(distinct(Listing.location_id))
    rank = case((name.ilike(starting, escape="\\"), 0), (brand.ilike(starting, escape="\\"), 1), else_=2)
    rows = db.execute(
        select(Product.id, Product.name, Product.brand, Product.category, Product.thumb_url, shops)
        .join(Listing, Listing.product_id == Product.id)
        .join(Location, Location.id == Listing.location_id)
        .where(
            publishable(db),
            shown_category(),
            or_(name.ilike(anywhere, escape="\\"), brand.ilike(anywhere, escape="\\")),
        )
        .group_by(Product.id)
        .order_by(rank, shops.desc(), func.length(Product.name), Product.id)
        .limit(limit)
    ).all()
    return [
        ProductSuggestion(id=pid, name=n, brand=b, category=c, thumb_url=t, location_count=count)
        for pid, n, b, c, t, count in rows
    ]


def _summary_base(
    db: Session, at_codes: list[str] | None = None, product_ids: list[int] | None = None
) -> Select:
    latest = _latest_observation_subquery(product_ids)
    stmt = (
        select(
            Product.id,
            Product.name,
            Product.brand,
            Product.brand_id,
            Product.gtin,
            Product.category,
            Product.size_ml,
            Product.abv,
            Product.is_exclusive,
            Product.thumb_url,
            Product.image_source,
            func.count(distinct(Listing.location_id)).label("location_count"),
            # Airport shops only: a product in one airport plus the online
            # catalogue cannot be compared airport-to-airport, and counting it
            # as comparable overstated that number by nearly a third.
            func.count(distinct(Listing.location_id))
            .filter(Location.is_catalogue_only.is_(False))
            .label("airport_count"),
            # A sold-out shelf is not the cheapest place to buy: out-of-stock
            # observations only count when nothing in stock exists at all.
            func.coalesce(
                func.min(PriceObservation.price_usd).filter(_available()),
                func.min(PriceObservation.price_usd),
            ).label("cheapest_usd"),
            func.coalesce(
                func.max(PriceObservation.price_usd).filter(_available()),
                func.max(PriceObservation.price_usd),
            ).label("dearest_usd"),
        )
        .join(Listing, Listing.product_id == Product.id)
        .join(PriceObservation, PriceObservation.listing_id == Listing.id)
        .join(Location, Location.id == Listing.location_id)
        .where(PriceObservation.id.in_(select(latest.c.obs_id)), publishable(db), shown_category())
        .group_by(Product.id)
    )
    if at_codes:
        # Everything downstream -- price, saving, shop count -- is then relative
        # to where this shopper is actually travelling.
        stmt = stmt.where(Location.code.in_(at_codes))
    return stmt


def _priced_locations(
    db: Session, product_ids: list[int], at_codes: list[str] | None
) -> dict[int, list[tuple[str, str | None, float, bool | None]]]:
    """Where each product is sold with its price, cheapest first, one row per
    shop. A card that SHOWS the comparison answers more of the shopper's
    question than a bare price with a promise behind a click."""
    if not product_ids:
        return {}
    latest = _latest_observation_subquery(product_ids)
    stmt = (
        select(
            Listing.product_id,
            Location.name,
            Location.iata,
            Location.code,
            Location.is_catalogue_only,
            PriceObservation.price_usd,
            PriceObservation.in_stock,
        )
        .join(PriceObservation, PriceObservation.listing_id == Listing.id)
        .join(Location, Location.id == Listing.location_id)
        .where(
            Listing.product_id.in_(product_ids),
            PriceObservation.id.in_(select(latest.c.obs_id)),
            PriceObservation.price_usd.isnot(None),
            publishable(db),
        )
        .order_by(
            Listing.product_id,
            case((PriceObservation.in_stock.is_(False), 1), else_=0),
            PriceObservation.price_usd.asc(),
            # Two shops at the same price keep one order, whatever plan the
            # database picks (a tie at BCN and MAD swapped with the query's shape).
            Location.code,
        )
    )
    if at_codes:
        stmt = stmt.where(Location.code.in_(at_codes))
    out: dict[int, list[tuple[str, str | None, float, bool | None]]] = {}
    seen: set[tuple[int, str]] = set()
    for product_id, name, iata, code, catalogue_only, price, in_stock in db.execute(stmt):
        # Bars have room for a short code, not a shop's full trading name.
        label = iata or ("Online" if catalogue_only else (code if len(code) <= 4 else name))
        # Two listings can share a shop; the first (cheapest, in-stock-first)
        # row is that shop's price on the card.
        if (product_id, label) in seen:
            continue
        seen.add((product_id, label))
        out.setdefault(product_id, []).append((label, iata, float(price), in_stock))
    return out


def _top_award(award: Award) -> TopAward:
    return TopAward(
        competition_slug=award.competition_slug,
        competition=award.competition,
        medal=award.medal,
        year=award.year,
    )


def _top_awards(
    db: Session, product_ids: list[int], at_codes: list[str] | None = None
) -> dict[int, TopAward]:
    """Each product's featured medal for its card corner, by the strategy in
    `award_picker` (the latest result per competition, then level, recency and
    the visitor's airports). The first version kept the best medal a bottle ever
    won, so a 2024 Double Gold outranked every later result for good."""
    if not product_ids:
        return {}
    rows = db.scalars(select(Award).where(Award.product_id.in_(product_ids))).all()
    held: dict[int, list[Award]] = {}
    for award in rows:
        held.setdefault(award.product_id, []).append(award)
    out: dict[int, TopAward] = {}
    for pid, awards in held.items():
        picked = award_picker.pick(awards, visitor_airports=at_codes or (), listing=True)
        if picked.featured is not None:
            out[pid] = _top_award(picked.featured)
    return out


def _award_counts(db: Session, product_ids: list[int]) -> dict[int, int]:
    if not product_ids:
        return {}
    rows = db.execute(
        select(Award.product_id, func.count(Award.id))
        .where(Award.product_id.in_(product_ids))
        .group_by(Award.product_id)
    ).all()
    return dict(rows)


def list_products(
    db: Session,
    *,
    query: str | None = None,
    category: str | None = None,
    family: str | None = None,
    multi_only: bool = False,
    awarded_only: bool = False,
    exclusives_only: bool = False,
    with_image_only: bool = False,
    at_codes: list[str] | None = None,
    only_ids: list[int] | None = None,
    stocked_at: list[int] | None = None,
    brand_ids: list[int] | None = None,
    sort: str = "featured",
    limit: int = 24,
    offset: int = 0,
) -> tuple[int, list[ProductSummary]]:
    # A known handful of products (the featured cards, a picker's list) ranks only
    # their own observations; a browse ranks the catalogue's.
    stmt = _summary_base(db, at_codes, product_ids=only_ids or None)

    if stocked_at:
        # An airport page: only products carried at these locations, but every
        # figure on the card still spans all our airports. Unlike `at_codes`
        # this narrows WHICH products, not where their prices come from, so a
        # bottle at Heathrow still shows that it is cheaper in Dubai.
        stmt = stmt.where(
            Product.id.in_(select(Listing.product_id).where(Listing.location_id.in_(stocked_at)))
        )
    if brand_ids:
        # A brand page: the fold row and its aliases (Decision 6), never a name match.
        stmt = stmt.where(Product.brand_id.in_(brand_ids))

    if query:
        # unaccent so "patron" finds Patrón; escaped so % and _ are literals.
        pattern = func.unaccent(_like_pattern(query))
        stmt = stmt.where(
            or_(
                func.unaccent(Product.name).ilike(pattern, escape="\\"),
                func.unaccent(func.coalesce(Product.brand, "")).ilike(pattern, escape="\\"),
            )
        )
    if category:
        stmt = stmt.where(Product.category == category)
    if family:
        # A department ("All drinks"): every shown category of the family. An
        # unknown family names no category and so matches nothing.
        stmt = stmt.where(Product.category.in_(taxonomy.categories_of(family)))
    if awarded_only:
        stmt = stmt.where(Product.id.in_(select(Award.product_id)))
    if exclusives_only:
        stmt = stmt.where(Product.is_exclusive.is_(True))
    if with_image_only:
        stmt = stmt.where(Product.thumb_url.isnot(None))
    if only_ids:
        stmt = stmt.where(Product.id.in_(only_ids))
    if multi_only:
        stmt = stmt.having(
            func.count(distinct(Listing.location_id))
            .filter(Location.is_catalogue_only.is_(False))
            > 1
        )

    total = db.scalar(select(func.count()).select_from(stmt.subquery())) or 0

    if sort == "price":
        stmt = stmt.order_by(
            func.min(PriceObservation.price_usd).asc().nulls_last(), Product.id
        )
    elif sort == "name":
        stmt = stmt.order_by(Product.name.asc(), Product.id)
    elif sort == "compared":
        stmt = stmt.order_by(
            func.count(distinct(Listing.location_id)).desc(),
            func.min(PriceObservation.price_usd).asc().nulls_last(),
            Product.id,
        )
    else:
        # "featured" (the default): lead with products that look and read as
        # complete -- a photo, a medal, and a real cross-shop comparison. A grid
        # of grey placeholders is a bad first impression and is also, genuinely,
        # a worse set of results for a shopper.
        awarded = select(Award.product_id).distinct().subquery()
        stmt = stmt.order_by(
            case((Product.thumb_url.isnot(None), 0), else_=1),
            case((Product.id.in_(select(awarded.c.product_id)), 0), else_=1),
            func.count(distinct(Listing.location_id)).desc(),
            func.min(PriceObservation.price_usd).asc().nulls_last(),
            Product.id,
        )

    rows = db.execute(stmt.limit(limit).offset(offset)).all()
    awards = _award_counts(db, [r.id for r in rows])
    top_awards = _top_awards(db, [r.id for r in rows], at_codes)
    priced_at = _priced_locations(db, [r.id for r in rows], at_codes)
    brand_slugs = brand_page_slugs(db, [r.brand_id for r in rows])
    items = [
        ProductSummary(
            id=r.id,
            name=r.name,
            brand=r.brand,
            brand_slug=brand_slugs.get(r.brand_id),
            gtin=r.gtin,
            category=r.category,
            size_ml=r.size_ml,
            abv=float(r.abv) if r.abv is not None else None,
            is_exclusive=r.is_exclusive,
            thumb_url=r.thumb_url,
            image_source=r.image_source,
            location_count=r.location_count,
            cheapest_usd=float(r.cheapest_usd) if r.cheapest_usd is not None else None,
            dearest_usd=float(r.dearest_usd) if r.dearest_usd is not None else None,
            award_count=awards.get(r.id, 0),
            best_location=(priced_at.get(r.id) or [(None, None, 0.0, None)])[0][0],
            best_location_iata=(priced_at.get(r.id) or [(None, None, 0.0, None)])[0][1],
            location_labels=[
                label for label, _, _, _ in (priced_at.get(r.id) or [])[:4]
            ],
            top_prices=[
                ShopPrice(label=label, usd=round(price, 2), in_stock=in_stock)
                for label, _, price, in_stock in (priced_at.get(r.id) or [])[:4]
            ],
            top_award=top_awards.get(r.id),
        )
        for r in rows
    ]
    return total, items


# How many forwards to follow before giving up: a merge chain is one or two
# hops in practice, and a cycle (which nothing should write) must not hang a
# request.
MERGE_CHAIN_LIMIT = 8


def resolve_product_id(db: Session, product_id: int) -> int:
    """The id a request for `product_id` should be answered under.

    Every merged product must 301 to its survivor (build plan §6, Decision 6):
    the page route and the product API both pass their id through here first.
    Follows `products.merged_into_id` (migration #3) hop by hop, bounded, and
    returns the last id that forwards nowhere; an unknown id comes back as
    itself so the caller's 404 still fires.
    """
    scalar = getattr(db, "scalar", None)
    if scalar is None:
        # No session behind the request (the route tests fake the dependency):
        # the identity, exactly as before migration #3.
        return product_id
    current = product_id
    for _ in range(MERGE_CHAIN_LIMIT):
        target = scalar(select(Product.merged_into_id).where(Product.id == current))
        if not target or target == current:
            return current
        current = target
    return current


def get_product(
    db: Session, product_id: int, at_codes: list[str] | None = None
) -> ProductDetail | None:
    """The product page's object. `at_codes` (the visitor's airports) only order
    the medals; the server-rendered page passes none, so it is deterministic."""
    product = db.get(Product, product_id)
    if product is None:
        return None

    latest = _latest_observation_subquery()
    rows = db.execute(
        select(PriceObservation, Location, Retailer, Listing)
        .join(Listing, PriceObservation.listing_id == Listing.id)
        .join(Location, Listing.location_id == Location.id)
        .join(Retailer, Location.retailer_id == Retailer.id)
        .where(
            Listing.product_id == product_id,
            PriceObservation.id.in_(select(latest.c.obs_id)),
            publishable(db),
        )
        .order_by(
            case((PriceObservation.in_stock.is_(False), 1), else_=0),
            PriceObservation.price_usd.asc().nulls_last(),
        )
    ).all()

    # One row per shop: two source SKUs occasionally resolve to one product at
    # one location, and showing the same airport twice reads as a data error.
    # Rows are ordered best-buyable-first, so the first per location wins.
    seen_locations: set[int] = set()
    deduped = []
    for row in rows:
        if row[1].id in seen_locations:
            continue
        seen_locations.add(row[1].id)
        deduped.append(row)
    rows = deduped

    prices = [
        PriceOut(
            location_code=location.code,
            location_name=location.name,
            location_iata=location.iata,
            location_city=location.city,
            location_country=location.country,
            location_path=(
                airport_path(location.iata, location.city, location.name)
                if location.iata and not location.is_catalogue_only else None
            ),
            retailer_name=retailer.name,
            is_catalogue_only=location.is_catalogue_only,
            currency=obs.currency,
            price=float(obs.price),
            price_usd=float(obs.price_usd) if obs.price_usd is not None else None,
            was_price=float(obs.was_price) if obs.was_price is not None else None,
            price_type=obs.price_type,
            in_stock=obs.in_stock,
            observed_at=obs.observed_at,
            url=listing.url,
        )
        for obs, location, retailer, listing in rows
    ]

    # The page shows the retained medals (the latest result per competition),
    # featured first; seo.py's `award` property reads the same list. The count
    # stays the number of medals held. A pin will arrive through
    # award_picker.pin_from_override once the overrides table exists (R's
    # migration #4); the read path only ever falls back, a GET never writes.
    held = db.scalars(select(Award).where(Award.product_id == product_id)).all()
    picked = award_picker.pick(held, visitor_airports=at_codes or ())
    awards = [AwardOut.model_validate(a) for a in picked.retained]

    usd = [p.price_usd for p in prices if p.price_usd is not None]
    return ProductDetail(
        id=product.id,
        name=product.name,
        brand=product.brand,
        brand_slug=brand_page_slugs(db, [product.brand_id]).get(product.brand_id),
        gtin=product.gtin,
        category=product.category,
        size_ml=product.size_ml,
        abv=float(product.abv) if product.abv is not None else None,
        is_exclusive=product.is_exclusive,
        thumb_url=product.thumb_url,
        image_url=product.image_url,
        image_source=product.image_source,
        country_of_origin=product.country_of_origin,
        location_count=len({p.location_name for p in prices}),
        cheapest_usd=min(usd) if usd else None,
        dearest_usd=max(usd) if usd else None,
        award_count=len(held),
        top_award=_top_award(picked.featured) if picked.featured is not None else None,
        prices=prices,
        awards=awards,
    )


def featured_savings(
    db: Session, at_codes: list[str] | None = None, total: int = 8
) -> list[ProductSummary]:
    """The home page's lead cards, selected by the agreed rules (see featured.py)."""
    latest = _latest_observation_subquery()
    awarded = select(Award.product_id).distinct().subquery()
    rows = db.execute(
        select(
            Product.id,
            func.coalesce(
                func.min(PriceObservation.price_usd).filter(_available()),
                func.min(PriceObservation.price_usd),
            ).label("cheapest"),
            func.max(PriceObservation.price_usd).label("dearest"),
            Product.id.in_(select(awarded.c.product_id)).label("has_award"),
            Product.thumb_url.isnot(None).label("has_image"),
            func.count(distinct(Listing.location_id)).label("shop_count"),
        )
        .join(Listing, Listing.product_id == Product.id)
        .join(PriceObservation, PriceObservation.listing_id == Listing.id)
        .join(Location, Location.id == Listing.location_id)
        .where(
            PriceObservation.id.in_(select(latest.c.obs_id)),
            PriceObservation.price_usd.isnot(None),
            publishable(db),
            shown_category(),
            *( [Location.code.in_(at_codes)] if at_codes else [] ),
        )
        .group_by(Product.id)
        .having(
            func.count(distinct(Listing.location_id))
            .filter(Location.is_catalogue_only.is_(False))
            > 1
        )
    ).all()
    records = [
        SavingRecord(
            product_id=r.id,
            saving_usd=float(r.dearest) - float(r.cheapest),
            saving_pct=(float(r.dearest) - float(r.cheapest)) / float(r.dearest)
            if float(r.dearest) > 0
            else 0.0,
            awarded=bool(r.has_award),
            has_image=bool(r.has_image),
            shop_count=int(r.shop_count),
        )
        for r in rows
        if r.cheapest is not None and r.dearest is not None
    ]
    order = pick_featured(records, total=total)
    if not order:
        return []
    _, items = list_products(db, at_codes=at_codes, only_ids=order, limit=len(order))
    by_id = {item.id: item for item in items}
    return [by_id[pid] for pid in order if pid in by_id]


def similar_products(
    db: Session,
    product_id: int,
    limit: int = 8,
    at_codes: list[str] | None = None,
) -> list[ProductSummary]:
    """Other bottles a shopper looking at this one would plausibly consider:
    same category, comparable across shops, complete-looking first.

    Scoped to the shopper's airports when they have set any -- suggesting a
    bottle they cannot walk up to is worse than suggesting nothing.
    """
    product = db.get(Product, product_id)
    if product is None:
        return []
    _, items = list_products(
        db,
        category=product.category,
        at_codes=at_codes,
        sort="featured",
        limit=limit + 1,
    )
    return [item for item in items if item.id != product_id][:limit]


# --- airports ---------------------------------------------------------------
#
# An airport page is keyed by IATA code and gathers every visible storefront
# there (one today, but two operators at one airport is the normal case in
# the world), so each query below takes the airport's location ids.

SORTS = ("featured", "compared", "price", "name")


def airport_locations(db: Session, iata: str | None = None) -> list[Location]:
    """Visible airport storefronts that hold stock, in IATA order; one
    airport's when `iata` is given. The same rule as the airport picker: a
    configured store we hold nothing for has no page."""
    stmt = (
        select(Location)
        .where(
            Location.is_catalogue_only.is_(False),
            Location.iata.isnot(None),
            publishable(db),
            Location.id.in_(select(Listing.location_id).distinct()),
        )
        .order_by(Location.iata, Location.id)
    )
    if iata:
        stmt = stmt.where(Location.iata == iata.upper())
    return list(db.scalars(stmt))


def _airport_shops(db: Session, locations: list[Location]) -> list[AirportShop]:
    ids = [loc.id for loc in locations]
    rows = db.execute(
        select(
            Listing.location_id,
            func.count(distinct(Listing.product_id)),
            func.max(PriceObservation.observed_at),
        )
        .join(PriceObservation, PriceObservation.listing_id == Listing.id)
        .where(Listing.location_id.in_(ids))
        .group_by(Listing.location_id)
    ).all()
    by_id = {location_id: (products, last) for location_id, products, last in rows}
    return [
        AirportShop(
            code=loc.code,
            retailer_name=loc.retailer.name,
            products=by_id.get(loc.id, (0, None))[0],
            last_collected_at=by_id.get(loc.id, (0, None))[1],
        )
        for loc in locations
    ]


def _airport_summary(db: Session, locations: list[Location], shops: list[AirportShop]) -> AirportSummary:
    first = locations[0]
    city = next((loc.city for loc in locations if loc.city), None)
    country = next((loc.country for loc in locations if loc.country), None)
    # Several shops can carry the same bottle; the airport's count is of
    # distinct products, not of shelves.
    products = db.scalar(
        select(func.count(distinct(Listing.product_id))).where(
            Listing.location_id.in_([loc.id for loc in locations])
        )
    ) or 0
    return AirportSummary(
        iata=first.iata,
        path=airport_path(first.iata, city, first.name),
        name=first.name,
        city=city,
        country=country,
        currency=first.currency,
        products=products,
        last_collected_at=max((s.last_collected_at for s in shops if s.last_collected_at), default=None),
    )


def list_airports(db: Session) -> list[AirportSummary]:
    """Every airport with a page, for the sitemap and the airports index."""
    grouped: dict[str, list[Location]] = {}
    for loc in airport_locations(db):
        grouped.setdefault(loc.iata, []).append(loc)
    return [
        _airport_summary(db, locations, _airport_shops(db, locations))
        for locations in grouped.values()
    ]


def _airport_rows(db: Session, location_ids: list[int]):
    """One row per product carried at the airport: the card figures across
    every airport plus `here_usd`, what it costs at this one."""
    stocked = select(Listing.product_id).where(Listing.location_id.in_(location_ids))
    here = Listing.location_id.in_(location_ids)
    stmt = (
        _summary_base(db)
        .add_columns(
            func.coalesce(
                func.min(PriceObservation.price_usd).filter(here, _available()),
                func.min(PriceObservation.price_usd).filter(here),
            ).label("here_usd")
        )
        .where(Product.id.in_(stocked))
    )
    return db.execute(stmt).all()


def airport_detail(
    db: Session,
    iata: str,
    *,
    category: str | None = None,
    multi_only: bool = False,
    awarded_only: bool = False,
    exclusives_only: bool = False,
    sort: str = "featured",
    limit: int = 24,
    offset: int = 0,
    sample: int = 8,
) -> AirportDetail | None:
    """Everything the airport page shows, or None when we hold no visible shop
    at that code (a hidden airport is a 404, not a teaser). The three feature flags
    are the storefront's own chips, applied to the full list only."""
    locations = airport_locations(db, iata)
    if not locations:
        return None
    ids = [loc.id for loc in locations]
    shops = _airport_shops(db, locations)
    summary = _airport_summary(db, locations, shops)
    sort = sort if sort in SORTS else "featured"

    rows = _airport_rows(db, ids)
    # "Cheapest here" is the airport's card: this shop's latest price is the
    # lowest we know for a bottle sold at two or more of our airports. The saving
    # is against the dearest of them, the same figure the product page shows.
    winners = []
    for r in rows:
        if r.airport_count < 2 or r.here_usd is None or r.cheapest_usd is None:
            continue
        if float(r.here_usd) <= float(r.cheapest_usd) + 0.005:
            winners.append((float(r.dearest_usd or r.here_usd) - float(r.here_usd), r.id))
    winners.sort(key=lambda w: (-w[0], w[1]))
    top_ids = [pid for saving, pid in winners if saving > 0.5][:sample]
    savings: list[ProductSummary] = []
    if top_ids:
        _, found = list_products(db, only_ids=top_ids, limit=len(top_ids))
        by_id = {item.id: item for item in found}
        savings = [by_id[pid] for pid in top_ids if pid in by_id]

    _, exclusive_items = list_products(
        db, stocked_at=ids, exclusives_only=True, sort="featured", limit=sample
    )
    # Featured by family (services/airport_featured.py): the same rows, the home page's picker
    # per family, one summaries fetch for every card.
    featured = airport_featured.featured_at(
        db, rows, lambda wanted: list_products(db, only_ids=wanted, limit=len(wanted))[1]
    )
    total, items = list_products(
        db, stocked_at=ids, category=category, multi_only=multi_only, awarded_only=awarded_only,
        exclusives_only=exclusives_only, sort=sort, limit=limit, offset=offset,
    )
    counted = Counter(r.category for r in rows if r.category)
    writeup = editorial.airport_writeup(db, iata)
    category_pages = category_page_links(db, iata)

    return AirportDetail(
        **summary.model_dump(),
        shops=shops,
        comparable=sum(1 for r in rows if r.airport_count > 1),
        cheapest_here=len(winners),
        exclusives=sum(1 for r in rows if r.is_exclusive),
        categories=[CategoryCount(category=c, count=n) for c, n in counted.most_common()],
        savings=savings,
        exclusive_items=exclusive_items,
        featured=featured,
        category_pages=category_pages,
        category=category,
        multi_only=multi_only,
        awarded_only=awarded_only,
        exclusives_only=exclusives_only,
        sort=sort,
        total=total,
        limit=limit,
        offset=offset,
        items=items,
        writeup=editorial.article_out(writeup) if writeup else None,
        guide=airport_guides.guide_for(iata, db),
    )


# --- category at airport ----------------------------------------------------
#
# The pairing page (`/airports/<airport>/<category>`) exists only where the pair holds at
# least coverage.CATEGORY_AT_AIRPORT_MIN_PRODUCTS published products, read at request time:
# one count, `category_pair_counts`, feeds the page's 404, the airport page's rail of links
# and the sitemap, so they can never disagree.


def category_pair_counts(db: Session, iata: str | None = None) -> list[tuple[str, str, int]]:
    """`(iata, category, distinct products)` over the shops the site shows and the categories
    it shows, one airport's when `iata` is given; biggest first within an airport. No session
    (a sitemap built from stubs in a test) means no pairs, as `blocked_location_ids` answers."""
    if db is None or not hasattr(db, "execute"):
        return []
    stmt = (
        select(Location.iata, Product.category, func.count(distinct(Product.id)))
        .join(Listing, Listing.location_id == Location.id)
        .join(Product, Product.id == Listing.product_id)
        .where(
            Location.is_catalogue_only.is_(False),
            Location.iata.isnot(None),
            Product.category.isnot(None),
            publishable(db),
            shown_category(),
        )
        .group_by(Location.iata, Product.category)
        .order_by(Location.iata, func.count(distinct(Product.id)).desc(), Product.category)
    )
    if iata:
        stmt = stmt.where(Location.iata == iata.upper())
    return [(code, category, int(n)) for code, category, n in db.execute(stmt).all()]


def category_page_links(db: Session, iata: str, airport_path_: str | None = None) -> list[CategoryPageLink]:
    """The category pages one airport has, biggest first (the airport page's rail)."""
    from app.services import coverage  # lazy: coverage imports this module

    if airport_path_ is None:
        locations = airport_locations(db, iata)
        if not locations:
            return []
        airport_path_ = _airport_summary(db, locations, _airport_shops(db, locations)).path
    return [
        CategoryPageLink(category=category, slug=category_slug(category), path=airport_category_path(airport_path_, category), count=n)
        for _, category, n in coverage.qualifying_pairs(db, iata)
        if category_slug(category)
    ]


def category_page_rows(db: Session) -> list[tuple[str, datetime | None]]:
    """(path, lastmod) for every pairing page that exists, for the sitemap."""
    from app.services import coverage  # lazy: coverage imports this module

    airports = {a.iata: a for a in list_airports(db)}
    return [
        (airport_category_path(airports[code].path, category), airports[code].last_collected_at)
        for code, category, _ in coverage.qualifying_pairs(db)
        if code in airports and category_slug(category)
    ]


def airport_category_detail(
    db: Session,
    iata: str,
    category: str,
    *,
    multi_only: bool = False,
    awarded_only: bool = False,
    exclusives_only: bool = False,
    sort: str = "featured",
    limit: int = 24,
    offset: int = 0,
    sample: int = 4,
) -> AirportCategoryDetail | None:
    """Everything the pairing page shows, or None where the airport is hidden, the category
    has no address, or the pair is under the coverage bar (a 404, never a thin page)."""
    from app.services import coverage  # lazy: coverage imports this module

    locations = airport_locations(db, iata)
    if not locations or not category_slug(category):
        return None
    counts = {c: n for _, c, n in category_pair_counts(db, iata)}
    if counts.get(category, 0) < coverage.CATEGORY_AT_AIRPORT_MIN_PRODUCTS:
        return None
    ids = [loc.id for loc in locations]
    shops = _airport_shops(db, locations)
    summary = _airport_summary(db, locations, shops)
    sort = sort if sort in SORTS else "featured"

    rows = [r for r in _airport_rows(db, ids) if r.category == category]
    winners = 0
    for r in rows:
        if r.airport_count < 2 or r.here_usd is None or r.cheapest_usd is None:
            continue
        if float(r.here_usd) <= float(r.cheapest_usd) + 0.005 and float(r.dearest_usd or r.here_usd) - float(r.here_usd) > 0.5:
            winners += 1
    _, exclusive_items = list_products(
        db, stocked_at=ids, category=category, exclusives_only=True, sort="featured", limit=sample
    )
    total, items = list_products(
        db, stocked_at=ids, category=category, multi_only=multi_only, awarded_only=awarded_only,
        exclusives_only=exclusives_only, sort=sort, limit=limit, offset=offset,
    )
    siblings = [link for link in category_page_links(db, iata, summary.path) if link.category != category]
    return AirportCategoryDetail(
        # The airport's identity, then the pair's own address and the category's count: the
        # first rehearsal on a copy kept the airport's path here and the canonical pointed at
        # the airport page (tests/test_airport_category.py pins the path).
        **{**summary.model_dump(), "path": airport_category_path(summary.path, category), "products": counts[category]},
        shops=shops,
        category=category,
        category_slug=category_slug(category),
        airport_path=summary.path,
        comparable=sum(1 for r in rows if r.airport_count > 1),
        cheapest_here=winners,
        exclusives=sum(1 for r in rows if r.is_exclusive),
        exclusive_items=exclusive_items,
        siblings=siblings,
        multi_only=multi_only,
        awarded_only=awarded_only,
        exclusives_only=exclusives_only,
        sort=sort,
        total=total,
        limit=limit,
        offset=offset,
        items=items,
    )


# --- brands -----------------------------------------------------------------
#
# A brand page is keyed by the brands table's slug (migration #3): the fold row
# plus any alias rows pointing at it through canonical_id. The review's floor
# (Adam, 10 Sep: "three or more where stocked at two or more airports"): a
# house has a page when three or more of its published products are priced,
# between them, at two or more distinct airports. Read that way because the
# page is a comparison: three bottles at three airports beat five at one, and
# a single-bottle perfume house stays out. A thin brand has no page, no
# sitemap entry and no link from its products until it grows. One definition,
# `_house_counts`, so the index, the page, the sitemap and the product page's
# brand link can never disagree.

BRAND_PAGE_MIN_PRODUCTS = 3
BRAND_PAGE_MIN_AIRPORTS = 2


def brand_by_slug(db: Session, slug: str) -> Brand | None:
    """The canonical brand row for a slug: an alias resolves to its house."""
    brand = db.scalar(select(Brand).where(Brand.slug == slug))
    if brand is not None and brand.canonical_id and brand.canonical_id != brand.id:
        brand = db.get(Brand, brand.canonical_id) or brand
    return brand


def _brand_ids(db: Session, brand: Brand) -> list[int]:
    aliases = db.scalars(select(Brand.id).where(Brand.canonical_id == brand.id)).all()
    return [brand.id, *[a for a in aliases if a != brand.id]]


def _brand_rows(db: Session, brand_ids: list[int]):
    """The brand's products at visible airports: each product's latest priced
    observation per airport location, with the location's identity."""
    latest = _latest_observation_subquery()
    return db.execute(
        select(
            Product.id,
            Product.category,
            Product.is_exclusive,
            Location.id,
            Location.iata,
            Location.city,
            Location.name,
            func.coalesce(
                func.min(PriceObservation.price_usd).filter(_available()),
                func.min(PriceObservation.price_usd),
            ),
            func.max(PriceObservation.observed_at),
        )
        .join(Listing, Listing.product_id == Product.id)
        .join(PriceObservation, PriceObservation.listing_id == Listing.id)
        .join(Location, Location.id == Listing.location_id)
        .where(
            Product.brand_id.in_(brand_ids),
            shown_category(),
            PriceObservation.id.in_(select(latest.c.obs_id)),
            Location.is_catalogue_only.is_(False),
            Location.iata.isnot(None),
            publishable(db),
        )
        .group_by(Product.id, Location.id)
    ).all()


def _brand_summary(brand: Brand, rows) -> BrandSummary:
    return BrandSummary(
        slug=brand.slug,
        path=brand_page_path(brand.slug),
        name=brand.name,
        products=len({r[0] for r in rows}),
        last_collected_at=max((r[8] for r in rows if r[8]), default=None),
    )


def _house_counts(
    db: Session,
    *,
    min_products: int = BRAND_PAGE_MIN_PRODUCTS,
    min_airports: int = BRAND_PAGE_MIN_AIRPORTS,
    house_ids: list[int] | None = None,
):
    """The houses (canonical brand rows) that meet the floor for a page:
    (house id, products, last observed). The one definition of "this brand has
    a page": `min_products` published products priced, between them, at
    `min_airports` distinct airports (by IATA, so two shops at one airport are
    one airport). `house_ids` narrows it to the brands a list of cards names."""
    house = func.coalesce(Brand.canonical_id, Brand.id)
    stmt = (
        select(house.label("house"), func.count(distinct(Product.id)), func.max(PriceObservation.observed_at))
        .join(Product, Product.brand_id == Brand.id)
        .join(Listing, Listing.product_id == Product.id)
        .join(PriceObservation, PriceObservation.listing_id == Listing.id)
        .join(Location, Location.id == Listing.location_id)
        .where(publishable(db), shown_category(), Location.is_catalogue_only.is_(False))
        .group_by("house")
        .having(
            func.count(distinct(Product.id)) >= min_products,
            func.count(distinct(Location.iata)) >= min_airports,
        )
    )
    if house_ids is not None:
        stmt = stmt.where(house.in_(house_ids))
    return db.execute(stmt).all()


def brand_page_slugs(db: Session, brand_ids: list[int | None]) -> dict[int, str]:
    """For each product `brand_id` (a fold row or an alias), the slug of its
    house's page, when the house has one.

    A product page links its brand name to `/brands/<slug>` only through this:
    a brand too thin for a page (under the `_house_counts` floor) gets no slug
    and the name links to a catalogue search instead.
    Measured on a copy of the launch data, a blind slug would have put a 404
    link on about a quarter of the visible product pages.
    """
    ids = sorted({i for i in brand_ids if i})
    if not ids:
        return {}
    house = aliased(Brand)
    rows = db.execute(
        select(Brand.id, house.id, house.slug)
        .join(house, house.id == func.coalesce(Brand.canonical_id, Brand.id))
        .where(Brand.id.in_(ids))
    ).all()
    houses = {brand_id: (house_id, slug) for brand_id, house_id, slug in rows}
    if not houses:
        return {}
    with_pages = {h for h, _, _ in _house_counts(db, house_ids=sorted({h for h, _ in houses.values()}))}
    return {brand_id: slug for brand_id, (house_id, slug) in houses.items() if house_id in with_pages}


def list_brands(
    db: Session, min_products: int = BRAND_PAGE_MIN_PRODUCTS, min_airports: int = BRAND_PAGE_MIN_AIRPORTS
) -> list[BrandSummary]:
    """Every brand with a page: canonical rows that meet the floor, for the
    sitemap and the brands index."""
    counted = _house_counts(db, min_products=min_products, min_airports=min_airports)
    if not counted:
        return []
    house_ids = [h for h, _, _ in counted]
    houses = {b.id: b for b in db.scalars(select(Brand).where(Brand.id.in_(house_ids)))}
    families = _house_families(db, house_ids)
    out = [
        BrandSummary(slug=houses[h].slug, path=brand_page_path(houses[h].slug), name=houses[h].name,
                     products=n, last_collected_at=last, family=families.get(h))
        for h, n, last in counted if h in houses
    ]
    return sorted(out, key=lambda b: b.slug)


def _house_families(db: Session, house_ids: list[int]) -> dict[int, str | None]:
    """Each house's family, from where its visible products sit (the same
    listings `_house_counts` counts), so a menu lists a whisky house under
    Drinks and a fragrance house under Beauty without a typed list of either."""
    if not house_ids:
        return {}
    house = func.coalesce(Brand.canonical_id, Brand.id)
    rows = db.execute(
        select(house.label("house"), Product.category, func.count(distinct(Product.id)))
        .join(Product, Product.brand_id == Brand.id)
        .join(Listing, Listing.product_id == Product.id)
        .join(Location, Location.id == Listing.location_id)
        .where(publishable(db), shown_category(), Location.is_catalogue_only.is_(False), house.in_(house_ids))
        .group_by("house", Product.category)
    ).all()
    per_house: dict[int, dict[str | None, int]] = {}
    for h, category, n in rows:
        per_house.setdefault(h, {})[category] = n
    return {h: taxonomy.majority_family(counts) for h, counts in per_house.items()}


def brand_detail(
    db: Session,
    slug: str,
    *,
    category: str | None = None,
    sort: str = "featured",
    limit: int = 24,
    offset: int = 0,
    min_products: int = BRAND_PAGE_MIN_PRODUCTS,
    min_airports: int = BRAND_PAGE_MIN_AIRPORTS,
) -> BrandDetail | None:
    """Everything the brand page shows, or None when the slug is unknown or the
    brand is under the floor (`_house_counts`, the same test the index, the
    sitemap and the product page's brand link apply)."""
    brand = brand_by_slug(db, slug)
    if brand is None:
        return None
    if not _house_counts(db, min_products=min_products, min_airports=min_airports, house_ids=[brand.id]):
        return None
    ids = _brand_ids(db, brand)
    rows = _brand_rows(db, ids)
    summary = _brand_summary(brand, rows)
    sort = sort if sort in SORTS else "featured"

    # Per product: its price at each airport; per airport: how many of the
    # brand's bottles it stocks, and for how many comparable ones it is the
    # cheapest place we know.
    by_product: dict[int, dict[int, float]] = {}
    airports: dict[int, dict] = {}
    for pid, category_name, is_exclusive, loc_id, iata, city, name, usd, last in rows:
        entry = airports.setdefault(loc_id, {"iata": iata, "city": city, "name": name, "products": set(),
                                             "cheapest_for": 0, "last": None})
        entry["products"].add(pid)
        if last and (entry["last"] is None or last > entry["last"]):
            entry["last"] = last
        if usd is not None:
            by_product.setdefault(pid, {})[loc_id] = float(usd)
    comparable = 0
    for pid, prices in by_product.items():
        if len(prices) < 2:
            continue
        comparable += 1
        low = min(prices.values())
        for loc_id, usd in prices.items():
            if usd <= low + 0.005:
                airports[loc_id]["cheapest_for"] += 1
    # One row per IATA even if two shops share an airport.
    per_iata: dict[str, BrandAirport] = {}
    for loc_id, a in sorted(airports.items()):
        row = per_iata.get(a["iata"])
        if row is None:
            per_iata[a["iata"]] = BrandAirport(
                iata=a["iata"], path=airport_path(a["iata"], a["city"], a["name"]), name=a["name"],
                products=len(a["products"]), cheapest_for=a["cheapest_for"], last_collected_at=a["last"],
            )
        else:
            row.products = max(row.products, len(a["products"]))
            row.cheapest_for += a["cheapest_for"]
    airport_list = sorted(per_iata.values(), key=lambda a: (-a.products, -a.cheapest_for, a.iata))

    categories = Counter(category_name for pid, category_name, *_ in {(r[0], r[1]) for r in rows} if category_name)
    exclusives = len({r[0] for r in rows if r[2]})
    total, items = list_products(
        db, brand_ids=ids, category=category, sort=sort, limit=limit, offset=offset
    )
    return BrandDetail(
        **summary.model_dump(),
        airports=airport_list,
        comparable=comparable,
        exclusives=exclusives,
        categories=[CategoryCount(category=c, count=n) for c, n in categories.most_common()],
        category=category,
        sort=sort,
        total=total,
        limit=limit,
        offset=offset,
        items=items,
    )


def newest_products(db: Session, limit: int = 50) -> list[tuple[ProductSummary, object]]:
    """The products most recently added to the catalogue at a visible airport,
    newest first, each with the moment it was first recorded (the feed's items)."""
    rows = db.execute(
        select(Product.id, Product.created_at)
        .where(
            Product.id.in_(
                select(Listing.product_id)
                .join(Location, Location.id == Listing.location_id)
                .where(publishable(db))
            )
        )
        .order_by(Product.created_at.desc(), Product.id.desc())
        .limit(limit)
    ).all()
    if not rows:
        return []
    _, items = list_products(db, only_ids=[pid for pid, _ in rows], limit=len(rows))
    by_id = {item.id: item for item in items}
    return [(by_id[pid], created) for pid, created in rows if pid in by_id]


def dataset_facts(db: Session) -> DatasetFacts:
    """The public data page's figures: visible airports only, like /api/stats."""
    visible_listing = (
        select(Listing.id, Listing.product_id, Listing.location_id)
        .join(Location, Location.id == Listing.location_id)
        .where(publishable(db))
        .subquery()
    )
    obs = select(PriceObservation).where(
        PriceObservation.listing_id.in_(select(visible_listing.c.id))
    ).subquery()
    first, last, observations, currencies = db.execute(
        select(
            func.min(obs.c.observed_at), func.max(obs.c.observed_at),
            func.count(obs.c.id), func.count(distinct(obs.c.currency)),
        )
    ).one()
    products = select(func.count(distinct(visible_listing.c.product_id)))
    return DatasetFacts(
        observations=observations or 0,
        products=db.scalar(products) or 0,
        airports=db.scalar(
            select(func.count(distinct(Location.id))).where(
                Location.id.in_(select(visible_listing.c.location_id)),
                Location.is_catalogue_only.is_(False),
            )
        ) or 0,
        retailers=db.scalar(
            select(func.count(distinct(Location.retailer_id))).where(
                Location.id.in_(select(visible_listing.c.location_id))
            )
        ) or 0,
        currencies=currencies or 0,
        with_barcode=db.scalar(
            select(func.count(Product.id)).where(
                Product.id.in_(select(visible_listing.c.product_id)), Product.gtin.isnot(None)
            )
        ) or 0,
        awards=db.scalar(
            select(func.count(Award.id)).where(
                Award.product_id.in_(select(visible_listing.c.product_id))
            )
        ) or 0,
        first_observed_at=first,
        last_observed_at=last,
        airport_list=list_airports(db),
        categories=category_counts(db),
    )


def category_counts(db: Session) -> list[CategoryCount]:
    # Counted through visible listings so the numbers match what browsing that
    # category actually shows. Each count carries its family (taxonomy's vertical
    # and the shopper's word for it) so the mega menu groups shelves without a
    # copy of the taxonomy in the SPA; a category outside the vocabulary has none.
    rows = db.execute(
        select(Product.category, func.count(func.distinct(Product.id)))
        .join(Listing, Listing.product_id == Product.id)
        .join(Location, Location.id == Listing.location_id)
        .where(Product.category.isnot(None), publishable(db), shown_category())
        .group_by(Product.category)
        .order_by(func.count(func.distinct(Product.id)).desc())
    ).all()
    return [
        CategoryCount(
            category=c, count=n, family=taxonomy.vertical_of(c), family_label=taxonomy.family_label(c),
        )
        for c, n in rows
    ]