"""Turning collected listings into catalog rows.

This is where cross-retailer matching happens. A barcode, when we have one, makes
that a join; the normalised brand+name+size key is the fallback for the minority
of records without one.
"""

import html
import json
import logging
from dataclasses import asdict
from datetime import UTC, datetime, timedelta

from sqlalchemy import inspect, select, text
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
from sqlalchemy.orm import Session

from app.models import (
    ProductMerge,
    Brand,
    CollectionRun,
    Listing,
    Location,
    PriceObservation,
    Product,
    RawRecord,
    Retailer,
    Source,
)
from app.models.catalog import IDENTITY_RULES_VERSION
from app.services.collectors.base import (
    Collector,
    ListingRef,
    RawListing,
    bind_skip_sink,
    quantity_of,
    unbind_skip_sink,
)
from app.services.collected import write_listed
from app.services.collectors.fetch import SourceBlocked
from app.services.fx import FxRates, load_rates
from app.services import keying
from app.services.normalize import (
    brand_key,
    flat_key,
    looks_exclusive,
    parse_concentration,
    parse_size,
    size_ml_of,
)
from app.services.taxonomy import classify, vertical_of

DEFAULT_VERTICAL = "liquor"
# A run still "running" this long after it started belongs to a process that
# is gone: a deploy recreates the container and kills whatever was collecting.
STUCK_AFTER = timedelta(hours=24)


class RejectedListing(Exception):
    """Ingest refuses this listing; it must not be published or guessed at."""

    def __init__(self, reason: str, detail: str = "") -> None:
        super().__init__(f"{reason}: {detail}" if detail else reason)
        self.reason = reason
        self.detail = detail


def count_skip(run: CollectionRun, reason: str) -> None:
    """One skip, counted in the total and under its reason (migration #2)."""
    run.skipped_no_price += 1
    counts = dict(run.skip_counts or {})
    counts[reason] = counts.get(reason, 0) + 1
    run.skip_counts = counts


def record_rejection(
    db: Session, run: CollectionRun, *, stage: str, reason: str, raw: RawListing | None,
    detail: str = "",
) -> None:
    """The one place a refused listing is counted and, once the table exists, kept.

    Stream Q's migration #2 adds `rejected_observations(stage, reason,
    source_sku, url, payload, run_id)`; until then this logs and counts, and a
    schema mismatch there can never end a run.
    """
    count_skip(run, reason)
    logger.warning(
        "listing_rejected stage=%s reason=%s run=%s sku=%s detail=%s",
        stage, reason, run.id, raw.source_sku if raw else None, detail,
    )
    if not _has_table(db, "rejected_observations"):
        return
    savepoint = db.begin_nested()
    try:
        db.execute(
            text(
                "INSERT INTO rejected_observations (stage, reason, source_sku, url, payload, run_id)"
                " VALUES (:stage, :reason, :sku, :url, CAST(:payload AS jsonb), :run_id)"
            ),
            {
                "stage": stage, "reason": reason,
                "sku": raw.source_sku if raw else None, "url": raw.url if raw else None,
                "payload": json.dumps(
                    {"listing": asdict(raw) if raw else None, "detail": detail}, default=str
                ),
                "run_id": run.id,
            },
        )
        savepoint.commit()
    except SQLAlchemyError as exc:
        savepoint.rollback()
        logger.error("rejection_not_stored reason=%s error=%s", reason, exc)


_table_cache: dict[int, dict[str, bool]] = {}


def _has_table(db: Session, name: str) -> bool:
    bind = db.get_bind()
    known = _table_cache.setdefault(id(bind), {})
    if name not in known:
        known[name] = inspect(bind).has_table(name)
    return known[name]

logger = logging.getLogger(__name__)

# How many price rows to accumulate before committing. Small enough to keep lock
# windows short under concurrency, large enough not to thrash.
COMMIT_EVERY = 50


def _get_or_create(db: Session, model, defaults: dict, **lookup):
    """Fetch a row, creating it if absent, tolerating a concurrent creator.

    Collectors for the same retailer run concurrently (one process per airport,
    so each host's crawl delay applies independently), and they share retailer
    rows. Without this, whichever process loses the race dies on a unique
    constraint and its whole run is lost.
    """
    instance = db.scalar(select(model).filter_by(**lookup))
    if instance is not None:
        return instance
    savepoint = db.begin_nested()
    try:
        instance = model(**lookup, **defaults)
        db.add(instance)
        savepoint.commit()
        return instance
    except IntegrityError:
        savepoint.rollback()
        return db.scalar(select(model).filter_by(**lookup))


def ensure_source(db: Session, collector: Collector) -> Source:
    return _get_or_create(
        db, Source, {"name": collector.retailer_name}, slug=collector.slug
    )


def ensure_retailer_and_locations(db: Session, collector: Collector) -> dict[str, Location]:
    retailer = _get_or_create(
        db,
        Retailer,
        {
            "name": collector.retailer_name,
            "operator": collector.operator,
            "homepage": collector.homepage,
        },
        slug=collector.retailer_slug,
    )

    locations: dict[str, Location] = {}
    for spec in collector.locations():
        location = _get_or_create(
            db,
            Location,
            {
                "iata": spec.iata,
                "name": spec.name,
                "city": spec.city,
                "country": spec.country,
                "currency": spec.currency,
                "is_catalogue_only": spec.is_catalogue_only,
            },
            retailer_id=retailer.id,
            code=spec.code,
        )
        sync_location_currency(location, spec.currency)
        locations[spec.code] = location
    return locations


def held_listing_refs(db: Session, locations: list[Location]) -> list[ListingRef]:
    """Every listing with a URL the database holds at these locations.

    Handed to a collector that declares `wants_held_listings`, so a store whose
    listing pages may only be read one page deep (robots forbids the query
    string that pages them) re-reads the product URLs it already knows instead
    of letting page-2+ products fall silent.
    """
    if not locations:
        return []
    code_of = {location.id: location.code for location in locations}
    rows = db.execute(
        select(Listing.source_sku, Listing.url, Listing.location_id, Product.vertical)
        .join(Product, Product.id == Listing.product_id)
        .where(Listing.location_id.in_(list(code_of)), Listing.url.isnot(None))
        .order_by(Listing.id)
    ).all()
    return [
        ListingRef(source_sku=sku, url=url, location_code=code_of[location_id], vertical=vertical)
        for sku, url, location_id, vertical in rows
    ]


def sync_location_currency(location: Location, declared: str) -> bool:
    """The collector's declared currency wins over a stale row.

    One store was registered as MXN, then found to quote in USD; the
    collector was corrected but the row was not, so every one of its USD
    observations disagreed with its location for two weeks. The location's
    currency is collector data, not a human value, so it follows the code.
    """
    if location.currency == declared:
        return False
    logger.warning(
        "location_currency_synced code=%s was=%s now=%s", location.code, location.currency, declared
    )
    location.currency = declared
    return True


def _sizes_disagree(a: int | None, b: int | None) -> bool:
    """Two known sizes that cannot be the same bottle.

    Feeds are sloppy at 700-vs-750ml, so small differences are tolerated;
    beyond ~12 percent they are different bottles no matter what the barcode
    says -- retailers do mislabel barcodes, and trusting one puts a 70cl
    liqueur's price on a 1L whisky's page.
    """
    if not a or not b:
        return False
    return abs(a - b) / max(a, b) > 0.12


def _quantities_disagree(ours, theirs) -> bool:
    """Two known quantities that cannot be the same item, compared like with like: ml with
    ml (the 12 percent tolerance of `_sizes_disagree`), g with g, pcs with pcs; a different
    dimension is a disagreement (100 ml is never 100 g); an unknown on either side never
    vetoes."""
    if ours is None or theirs is None or ours.state != "stated" or theirs.state != "stated":
        return False
    if ours.unit is None or theirs.unit is None or ours.value is None or theirs.value is None:
        return False
    if ours.unit != theirs.unit:
        return True
    a, b = float(ours.value), float(theirs.value)
    if not a or not b:
        return False
    return abs(a - b) / max(a, b) > 0.12


def write_quantity(product: Product, q, *, name: str | None = None) -> None:
    """The seven quantity columns and `name_key` on a product, from one Quantity."""
    product.quantity_value = q.value
    product.quantity_unit = q.unit
    product.pack_count = q.pack_count
    product.pack_unit_value = q.pack_unit_value
    product.form = q.form
    product.set_contents = q.set_contents
    product.quantity_state = q.state
    product.name_key = flat_key(name if name is not None else product.name)[:400]


def brand_slug(brand: str | None) -> str:
    """The brands.slug for a spelling: the fold key, hyphenated."""
    return brand_key(brand).replace(" ", "-")


def resolve_brand(db: Session, brand: str | None) -> Brand | None:
    """The brands row a spelling folds to, created on first sight (migration #3).

    The row's `name` is the first spelling seen; `backfill brands` picks the
    most common spelling across the catalogue, and a human value stays.
    """
    slug = brand_slug(brand)
    if not slug or not _has_table(db, "brands"):
        return None
    return _get_or_create(db, Brand, {"name": brand.strip()}, slug=slug)


def raw_attributes(raw: RawListing) -> dict:
    """The per-vertical facts a listing declares (Decision 6: veto, never key)."""
    vertical = raw.vertical or vertical_of(classify(raw.name, raw.brand, raw.feed_categories))
    if vertical == "beauty":
        concentration = parse_concentration(raw.name)
        if concentration:
            return {"concentration": concentration}
    return {}


def attributes_disagree(ours: dict | None, theirs: dict | None) -> bool:
    """Two declared values for the same attribute that differ: not the same product.

    A missing value on either side is not a disagreement; most feeds omit the
    concentration, and a silent feed must never split a product.
    """
    ours, theirs = ours or {}, theirs or {}
    # The canonical variation subsumes the parsed concentration (identity rules v3): two
    # Elixir rows parsed as "parfum" and "edp" are one variation, not two bottles.
    if ours.get("variation") and theirs.get("variation"):
        return ours["variation"] != theirs["variation"]
    return any(key in theirs and theirs[key] != value for key, value in ours.items() if value)


def stated_size(raw: RawListing) -> tuple[float | None, str | None]:
    """The size as the name states it, kept only when it agrees with size_ml."""
    stated = parse_size(raw.name)
    if stated and (raw.size_ml is None or size_ml_of(*stated) == raw.size_ml):
        return stated
    if raw.size_ml:
        return float(raw.size_ml), "ml"
    return None, None


def survivor_of(db: Session, product: Product | None, limit: int = 8) -> Product | None:
    """The product a merged row forwards to; the row itself when it forwards nowhere."""
    hops = 0
    while product is not None and product.merged_into_id and hops < limit:
        product = db.get(Product, product.merged_into_id)
        hops += 1
    return product


def _resolve_product(db: Session, raw: RawListing) -> Product:
    """Find or create the canonical product a listing refers to."""
    # The keying call (identity rules v3, `services/keying.py`): the house with its alias
    # followed and the canonical variation, then the key.
    category = classify(raw.name, raw.brand, raw.feed_categories)
    quantity = quantity_of(raw, category=category)
    key, attributes, line_id = keying.listing_key(db, raw, raw_attributes(raw), quantity)

    # A barcode is identity -- unless the quantities flatly disagree, in which case
    # the retailer mislabelled the barcode and we refuse the join. The listing
    # then resolves by brand+name+quantity and must NOT carry the disputed code.
    gtin = raw.gtin
    product: Product | None = None
    if gtin:
        product = survivor_of(db, db.scalar(select(Product).where(Product.gtin == gtin)))
        if product is not None and (
            _quantities_disagree(keying.product_quantity(product, keying.maps_for(db)), quantity)
            or (product.quantity_state != "stated" and _sizes_disagree(product.size_ml, raw.size_ml))
        ):
            # Refuse, do not re-route: falling back to brand+name+size here
            # once created two 7,000 ml Glenfiddichs from a "0.70cl" misread,
            # each a published product with a price. Contradictory evidence
            # is a rejection, and the human review decides which side lied.
            raise RejectedListing(
                "gtin_size_veto",
                f"gtin={gtin} product={product.id}({product.size_ml}ml) "
                f"listing={raw.source_sku}({raw.size_ml}ml)",
            )
    if product is None:
        # The fallback key can join a barcode-bearing product (the key IS our
        # identity for barcode-less rows), but never one holding a DIFFERENT
        # barcode -- that would be a different bottle with the same name.
        # Taking only the first key-holder here once created a fresh duplicate
        # product per airport per run; consider every candidate.
        for candidate in db.scalars(
            select(Product)
            .where(Product.match_key == key, Product.merged_into_id.is_(None))
            .order_by(Product.gtin.isnot(None).desc(), Product.id)
        ):
            if gtin and candidate.gtin and candidate.gtin != gtin:
                continue
            # The per-vertical attribute vetoes: an EDP and an EDT of one line
            # and size share a key and are different bottles.
            if attributes_disagree(candidate.attributes, attributes):
                continue
            product = candidate
            break

    if product is None:
        # A human product merge holds across the next collection: a tombstone at this key
        # forwards to its survivor (LT3 d). Without this, `_resolve_listing` re-pinned the
        # listing to a fresh product at the old key on every sighting (576 pairs on staging).
        product = _follow_tombstone(db, key, gtin)

    if product is None:
        # Two collectors running concurrently can stock the same bottle and both
        # reach here. Create inside a savepoint so the loser re-reads the winner's
        # row instead of poisoning the whole session on the barcode constraint.
        brand_row = resolve_brand(db, raw.brand)
        savepoint = db.begin_nested()
        try:
            size_value, size_unit = stated_size(raw)
            product = Product(
                gtin=gtin,
                match_key=key,
                brand=raw.brand,
                brand_id=brand_row.id if brand_row else None,
                line_id=line_id,
                name=raw.name,
                category=category,
                vertical=resolve_vertical(raw.vertical, category, None),
                size_value=size_value,
                size_unit=size_unit,
                size_ml=raw.size_ml,
                abv=raw.abv,
                attributes=attributes,
                identity_rules_version=IDENTITY_RULES_VERSION,
                country_of_origin=raw.country_of_origin,
                is_exclusive=raw.is_exclusive or looks_exclusive(raw.name),
            )
            write_quantity(product, quantity)
            db.add(product)
            savepoint.commit()
            return product
        except IntegrityError:
            savepoint.rollback()
            product = None
            if gtin:
                product = db.scalar(select(Product).where(Product.gtin == gtin))
            if product is None:
                product = db.scalar(select(Product).where(Product.match_key == key))
            if product is None:
                raise

    # Enrich: a later source often knows a fact the first one did not. The
    # barcode enrichment runs in a savepoint because a concurrent collector may
    # have created a product with this code since we looked.
    if product.gtin is None and gtin:
        savepoint = db.begin_nested()
        try:
            product.gtin = gtin
            db.flush()
            savepoint.commit()
        except IntegrityError:
            savepoint.rollback()
    for attribute in ("brand", "size_ml", "abv", "country_of_origin"):
        if getattr(product, attribute) is None and getattr(raw, attribute) is not None:
            setattr(product, attribute, getattr(raw, attribute))
    if product.size_value is None and product.size_ml is not None:
        product.size_value, product.size_unit = stated_size(raw)
    # Enrich the quantity where nothing was read before; a stated one, or a decided one, stays.
    if product.quantity_state != "stated" and quantity.state == "stated" \
            and "quantity" not in keying.maps_for(db).decided.get(product.id, {}):
        write_quantity(product, quantity)
    if product.name_key is None:
        product.name_key = flat_key(product.name)[:400]
    if product.brand_id is None and product.brand:
        brand_row = resolve_brand(db, product.brand)
        product.brand_id = brand_row.id if brand_row else None
    missing = {k: v for k, v in attributes.items() if k not in (product.attributes or {})}
    if missing:
        # Reassigned, never mutated in place, so the session sees the change.
        product.attributes = {**(product.attributes or {}), **missing}
    if product.identity_rules_version != IDENTITY_RULES_VERSION:
        product.identity_rules_version = IDENTITY_RULES_VERSION
    # Re-run classification on every sighting: a second source often supplies the
    # category hint the first one lacked.
    if product.category is None:
        product.category = classify(raw.name, raw.brand, raw.feed_categories)
    product.vertical = resolve_vertical(raw.vertical, product.category, product.vertical)
    if raw.is_exclusive or looks_exclusive(raw.name):
        product.is_exclusive = True
    return product


def _follow_tombstone(db: Session, key: str, gtin: str | None) -> Product | None:
    """The survivor of a HUMAN merge whose tombstone carries this key under the current
    rules: `product_merges.reason == "confirmed"` (never a rule's fold), the tombstone's
    `identity_rules_version` current (`rederive` rekeys tombstones too), its quantity slot
    not `unknown` (unknown never agrees with known), and the survivor holding no barcode
    different from the listing's. Several tombstones on one key follow the newest merge."""
    parts = key.split("|")
    if len(parts) < 4 or parts[3] == "unknown":
        return None
    rows = db.execute(
        select(Product, ProductMerge)
        .join(ProductMerge, ProductMerge.from_id == Product.id)
        .where(Product.match_key == key, Product.merged_into_id.isnot(None),
               Product.identity_rules_version == IDENTITY_RULES_VERSION, ProductMerge.reason == "confirmed")
        .order_by(ProductMerge.id.desc())
    ).all()
    for tombstone, _merge in rows:
        survivor = survivor_of(db, tombstone)
        if survivor is None or survivor.merged_into_id is not None:
            continue
        if gtin and survivor.gtin and survivor.gtin != gtin:
            continue
        return survivor
    return None


def resolve_vertical(claimed: str | None, category: str | None, current: str | None) -> str:
    """The family a product belongs to.

    Our own category is the strongest evidence (it was read from the name and
    the feed's labels); the collector's claim is which tree it walked; the
    current value is kept when neither says otherwise, and the default is the
    only vertical collected before 2026-09.
    """
    return vertical_of(category) or claimed or current or DEFAULT_VERTICAL


def _resolve_listing(db: Session, product: Product, location: Location, raw: RawListing) -> Listing:
    listing = db.scalar(
        select(Listing).where(
            Listing.location_id == location.id, Listing.source_sku == raw.source_sku
        )
    )
    if listing is None:
        savepoint = db.begin_nested()
        try:
            listing = Listing(
                product_id=product.id,
                location_id=location.id,
                source_sku=raw.source_sku,
                url=raw.url,
            )
            db.add(listing)
            savepoint.commit()
            return listing
        except IntegrityError:
            savepoint.rollback()
            listing = db.scalar(
                select(Listing).where(
                    Listing.location_id == location.id,
                    Listing.source_sku == raw.source_sku,
                )
            )
            if listing is None:
                raise
    if listing is not None:
        # A pinned listing (a person's decision, LT3 e) stays on its pinned product's
        # survivor every sighting and is never re-pinned over by the rules.
        if listing.pinned_product_id:
            pinned = survivor_of(db, db.get(Product, listing.pinned_product_id))
            if pinned is not None:
                listing.product_id = pinned.id
        else:
            listing.product_id = product.id
        if raw.url:
            listing.url = raw.url
    return listing


def _implausibly_cheap(
    db: Session, product: Product, location: Location, price_usd: float | None
) -> bool:
    """A price far below what every other shop charges is a bug, not a bargain.

    Both real incidents so far -- a currency mix-up and a small-variant price
    recorded against the full-size bottle -- would have been caught here: the
    bad rows sat at under half the product's cross-shop median while genuine
    spreads (including border-shop discounts and 20 percent promotions) stay
    well above it. Quarantine means skip and log, never publish.
    """
    if price_usd is None or product.id is None:
        return False
    peers = [
        float(row)
        for row in db.scalars(
            select(PriceObservation.price_usd)
            .join(Listing, Listing.id == PriceObservation.listing_id)
            .where(
                Listing.product_id == product.id,
                Listing.location_id != location.id,
                PriceObservation.price_usd.isnot(None),
            )
            .order_by(Listing.location_id, PriceObservation.observed_at.desc())
            .distinct(Listing.location_id)
        )
    ]
    if len(peers) < 3:
        return False
    peers.sort()
    median = peers[len(peers) // 2]
    return price_usd < 0.45 * median


def facts_only_detail(detail: dict) -> dict:
    from app.services.collectors.base import facts_only
    return facts_only(detail)


def mark_stuck(run: CollectionRun, why: str) -> None:
    """A run whose process died is an error, not a collection in progress."""
    run.status = "error"
    run.error = f"stuck: {why}"
    run.finished_at = run.finished_at or datetime.now(UTC)


def is_stuck(run: CollectionRun, now: datetime) -> bool:
    return run.status == "running" and run.started_at is not None and now - run.started_at > STUCK_AFTER


def run_collector(
    db: Session,
    collector: Collector,
    *,
    limit: int | None = None,
    delay: float | None = None,
    rates: FxRates | None = None,
) -> CollectionRun:
    """Collect one source into the database, recording the run either way."""
    source = ensure_source(db, collector)
    # A source runs one process at a time, so any earlier run of it still
    # "running" belongs to a process that died (a deploy killed it). Eight
    # such runs once sat open for ten days and looked like live collections.
    for stale in db.scalars(
        select(CollectionRun).where(
            CollectionRun.source_id == source.id, CollectionRun.status == "running"
        )
    ):
        mark_stuck(stale, "superseded by a new run of the same source")
    run = CollectionRun(source_id=source.id, started_at=datetime.now(UTC))
    db.add(run)
    db.flush()

    if not source.enabled:
        run.status = "skipped"
        run.error = "source disabled"
        run.finished_at = datetime.now(UTC)
        db.commit()
        logger.info("collector_skipped source=%s reason=disabled", collector.slug)
        return run

    rates = rates or load_rates()
    run.fx_source = "fallback" if rates.is_fallback else "live"
    run.fx_fetched_at = rates.fetched_at
    locations = ensure_retailer_and_locations(db, collector)
    if getattr(collector, "wants_held_listings", False):
        collector.held_listings = held_listing_refs(db, list(locations.values()))
    effective_delay = float(source.delay_seconds) if delay is None else delay
    # The table arrives with migration #1; a collection may run on a container
    # that started before the upgrade, and must not die for want of it.
    keep_raw = _has_table(db, "raw_records")
    parser_version = getattr(collector, "parser_version", collector.slug)

    def collector_skipped(reason: str, detail: dict) -> None:
        record_rejection(
            db, run, stage="collector", reason=reason,
            raw=None, detail=json.dumps(facts_only_detail(detail), default=str)[:2000],
        )

    token = bind_skip_sink(collector_skipped)
    try:
        for raw in collector.collect(limit=limit, delay=effective_delay):
            location = locations.get(raw.location_code)
            if location is None:
                continue
            # A price of zero is not a price. Retailers list gift-with-purchase
            # items and unavailable lines at 0, and showing those as the cheapest
            # way to buy a bottle makes the whole comparison look broken.
            if raw.price is None or raw.price <= 0:
                count_skip(run, "no_price")
                continue
            # Feeds ship HTML entities in names ("Bacard&iacute;"); stored
            # verbatim they render literally everywhere. Decode at the one
            # boundary every collector passes through.
            raw.name = html.unescape(raw.name)
            if raw.brand:
                raw.brand = html.unescape(raw.brand)
            try:
                product = _resolve_product(db, raw)
            except RejectedListing as rejected:
                record_rejection(
                    db, run, stage="ingest", reason=rejected.reason, raw=raw,
                    detail=rejected.detail,
                )
                continue
            listing = _resolve_listing(db, product, location, raw)
            # Every sighting stamps `last_seen_at`; a fragment refreshes the listed columns as
            # one unit from that fragment (`collected.write_listed`). A sighting the price
            # quarantine refuses below still refreshes them: the shop showed the words.
            listing.last_seen_at = datetime.now(UTC)
            if keep_raw and raw.raw is not None:
                record = RawRecord(
                    listing_id=listing.id,
                    run_id=run.id,
                    payload=raw.raw,
                    parser_version=parser_version,
                )
                db.add(record)
                db.flush()
                write_listed(listing, record.id, parser_version, raw.raw)
            price_usd = rates.to_usd(raw.price, raw.currency)
            if _implausibly_cheap(db, product, location, price_usd):
                record_rejection(
                    db, run, stage="ingest", reason="price_quarantined", raw=raw,
                    detail=f"product={product.id} price_usd={price_usd}",
                )
                continue
            # Some shops render a "was" price equal to (or below) the current
            # one. Showing that as a saving would be wrong, so drop it.
            was_price = raw.was_price if raw.was_price and raw.was_price > raw.price else None
            db.add(
                PriceObservation(
                    listing_id=listing.id,
                    price=raw.price,
                    currency=raw.currency,
                    price_usd=price_usd,
                    was_price=was_price,
                    price_type=raw.price_type,
                    in_stock=raw.in_stock,
                    # Stamped as each listing arrives, which is just after its
                    # page was fetched: one timestamp per run made a
                    # six-hour crawl look like a single instant.
                    observed_at=datetime.now(UTC),
                    run_id=run.id,
                    fx_rate=rates.rates.get((raw.currency or "").upper()),
                )
            )
            run.products_seen += 1
            run.prices_written += 1
            # Commit in batches rather than holding one transaction for the whole
            # run. Collectors run concurrently and touch the same product rows, so
            # a long-held transaction deadlocks against its siblings; short ones
            # rarely do. It also means a run that fails part-way keeps what it
            # already collected instead of losing everything.
            if run.prices_written % COMMIT_EVERY == 0:
                db.commit()
        run.status = "ok"
    except SourceBlocked as exc:
        # The source declined. Record it and stop -- we do not work around a block.
        run.status = "blocked"
        run.error = str(exc)
        logger.warning("collector_blocked source=%s detail=%s", collector.slug, exc)
    except Exception as exc:
        run.status = "error"
        run.error = f"{type(exc).__name__}: {exc}"
        logger.exception("collector_failed source=%s", collector.slug)

    unbind_skip_sink(token)
    run.finished_at = datetime.now(UTC)
    db.commit()
    logger.info(
        "collector_finished source=%s status=%s prices=%d",
        collector.slug,
        run.status,
        run.prices_written,
    )
    return run