"""Catalog schema.

Two rules from the project plan shape this file, and both are deliberate:

* A price is an OBSERVATION, not a fact. Duty-free pricing varies by destination,
  loyalty tier and increasingly by traveller, so we never store "the price" -- we
  store a timestamped observation with its context and surface the date in the UI.
* We store facts, not expression. There is intentionally no column for retailer
  marketing copy and no rehosted product imagery.
"""

import uuid
from datetime import datetime

from sqlalchemy import (
    JSON,
    Boolean,
    DateTime,
    ForeignKey,
    Index,
    func,
    Integer,
    Numeric,
    String,
    Text,
    UniqueConstraint,
    Uuid,
    text,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.models.base import Base, TimestampMixin
# Moved to models/discussion.py (Stream R2, 2026-09-11); re-exported so every importer stands.
from app.models.discussion import DiscussionComment, DiscussionItem, FeaturePriority  # noqa: F401


class Account(Base):
    """Who did something: the one identity table, the FK target for every who-column.

    `username` is the kit's key and the `@mention` handle: lowercase, immutable (members,
    grants, sessions and the audit log key on it), NULL only on a row that cannot sign in.
    `email` is the login alias; `display_name` the only editable name. The credential lives
    in `account_credentials` (`app/models/accounts.py`), never here, so no listing over this
    table can carry a hash. Status: active | invited | disabled.
    """

    __tablename__ = "accounts"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str | None] = mapped_column(String(64), unique=True)
    email: Mapped[str | None] = mapped_column(String(320), unique=True)
    display_name: Mapped[str] = mapped_column(String(80), nullable=False)
    status: Mapped[str] = mapped_column(String(12), nullable=False, default="active", server_default="active")
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )
    last_login_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    disabled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    invited_by_id: Mapped[int | None] = mapped_column(ForeignKey("accounts.id"))


class Source(Base, TimestampMixin):
    """A collector definition, with the per-source kill switch and politeness caps.

    The kill switch exists so a collector can be stopped in seconds without a
    deploy -- if a retailer ever asks us to stop, we stop that day.
    """

    __tablename__ = "sources"

    id: Mapped[int] = mapped_column(primary_key=True)
    slug: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    name: Mapped[str] = mapped_column(String(160), nullable=False)
    enabled: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
    delay_seconds: Mapped[float] = mapped_column(Numeric(5, 2), default=1.0, nullable=False)
    max_concurrency: Mapped[int] = mapped_column(Integer, default=1, nullable=False)
    notes: Mapped[str | None] = mapped_column(Text)
    # How we present ourselves to this host: "declared" is the honest bot UA.
    # "browser_like" is allowed only when permission_record names who agreed,
    # when and how; without the record the collector refuses to switch. A UA
    # change is the ceiling; there is no stealth tier.
    identity_mode: Mapped[str] = mapped_column(
        String(16), default="declared", server_default="declared", nullable=False
    )
    permission_record: Mapped[str | None] = mapped_column(Text)
    # The cooperative control the page sets and the loop reads between requests (Stream AW4):
    # run | pause | stop, written by the API only, never by the loop. `control_set_at` against
    # the run's `started_at` decides whether it applies (`control.effective_control`): a stop
    # pressed yesterday never kills tonight's shell run, and nothing resets the field.
    control: Mapped[str] = mapped_column(String(8), default="run", server_default="run", nullable=False)
    control_set_by: Mapped[str | None] = mapped_column(String(80))
    control_set_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # Who last set the pace (`delay_seconds`) from the page, and when.
    delay_set_by: Mapped[str | None] = mapped_column(String(80))
    delay_set_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # The host's robots.txt Crawl-delay as the loop last read it: a published fact the pace
    # floor is built from, never a control. NULL until a run has read the host.
    robots_crawl_delay: Mapped[float | None] = mapped_column(Numeric(6, 2))
    robots_read_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # The mode the next Start uses: discover (the listing pages) or recheck (the held listings).
    mode: Mapped[str] = mapped_column(String(12), default="discover", server_default="discover", nullable=False)

    runs: Mapped[list["CollectionRun"]] = relationship(back_populates="source")


class CollectionRun(Base):
    __tablename__ = "collection_runs"

    id: Mapped[int] = mapped_column(primary_key=True)
    source_id: Mapped[int] = mapped_column(ForeignKey("sources.id"), nullable=False)
    started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    status: Mapped[str] = mapped_column(String(24), default="running", nullable=False)
    products_seen: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
    prices_written: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
    skipped_no_price: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
    # The per-reason split of skipped_no_price ({"no_price": 40, "gtin_size_veto":
    # 2}); the total above is kept. Reassigned, never mutated in place, so the
    # session notices the change (migration #2, Stream Q).
    skip_counts: Mapped[dict] = mapped_column(
        JSON().with_variant(JSONB(), "postgresql"), default=dict, server_default="{}", nullable=False
    )
    error: Mapped[str | None] = mapped_column(Text)
    # "live" or "fallback", and when the rates were fetched: a run on the
    # fallback table must stay visibly different from a live one.
    fx_source: Mapped[str | None] = mapped_column(String(16))
    fx_fetched_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    # The control plane's columns (Stream AW4, migration aw4a1b2c3d4e). `heartbeat_at` and
    # `requests_made` are written by the hook with a plain UPDATE guarded on status='running'
    # (never through this object, whose copy is stale by design); the counters by the loop after
    # resolution; `pid`, `mode`, `limit_n`, `started_by`, `expected_total` by the child at run
    # creation; `stopped_by` at a cooperative stop (`control_set_by`, kill-switch, signal, rules).
    heartbeat_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    requests_made: Mapped[int] = mapped_column(Integer, default=0, server_default=text("0"), nullable=False)
    pid: Mapped[int | None] = mapped_column(Integer)
    mode: Mapped[str] = mapped_column(String(12), default="discover", server_default="discover", nullable=False)
    limit_n: Mapped[int | None] = mapped_column(Integer)
    expected_total: Mapped[int | None] = mapped_column(Integer)
    existing_checked: Mapped[int] = mapped_column(Integer, default=0, server_default=text("0"), nullable=False)
    existing_changed: Mapped[int] = mapped_column(Integer, default=0, server_default=text("0"), nullable=False)
    existing_missing: Mapped[int] = mapped_column(Integer, default=0, server_default=text("0"), nullable=False)
    new_found: Mapped[int] = mapped_column(Integer, default=0, server_default=text("0"), nullable=False)
    new_brands: Mapped[int] = mapped_column(Integer, default=0, server_default=text("0"), nullable=False)
    new_lines: Mapped[int] = mapped_column(Integer, default=0, server_default=text("0"), nullable=False)
    new_variants: Mapped[int] = mapped_column(Integer, default=0, server_default=text("0"), nullable=False)
    started_by: Mapped[str | None] = mapped_column(String(80))
    stopped_by: Mapped[str | None] = mapped_column(String(80))

    source: Mapped[Source] = relationship(back_populates="runs")


class QuoteSelection(Base, TimestampMixin):
    """The client's yes/no on one quote line item; latest choice wins.

    One row per item (upsert), like FeaturePriority: the quote page shows
    where each line currently stands, and he can change his mind freely.
    """

    __tablename__ = "quote_selections"

    id: Mapped[int] = mapped_column(primary_key=True)
    item_key: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    included: Mapped[bool] = mapped_column(Boolean, nullable=False)
    author: Mapped[str] = mapped_column(String(80), nullable=False)
    author_id: Mapped[int | None] = mapped_column(ForeignKey("accounts.id"))


class QuoteRequest(Base):
    """A submitted "please quote this" with the basket frozen at that moment.

    Stored rather than emailed so the exact selection survives, whatever is
    toggled afterwards.
    """

    __tablename__ = "quote_requests"

    id: Mapped[int] = mapped_column(primary_key=True)
    author: Mapped[str] = mapped_column(String(80), nullable=False)
    author_id: Mapped[int | None] = mapped_column(ForeignKey("accounts.id"))
    note: Mapped[str | None] = mapped_column(String(2000))
    items: Mapped[str] = mapped_column(Text, nullable=False)
    total_usd: Mapped[int] = mapped_column(Integer, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )


class Retailer(Base, TimestampMixin):
    __tablename__ = "retailers"

    id: Mapped[int] = mapped_column(primary_key=True)
    slug: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    name: Mapped[str] = mapped_column(String(160), nullable=False)
    operator: Mapped[str | None] = mapped_column(String(160))
    homepage: Mapped[str | None] = mapped_column(String(400))

    shops: Mapped[list["Shop"]] = relationship(back_populates="retailer")


class Shop(Base, TimestampMixin):
    """An airport storefront. `iata` is null for a retailer-wide catalogue."""

    __tablename__ = "shops"
    __table_args__ = (UniqueConstraint("retailer_id", "code", name="uq_shop_retailer_code"),)

    id: Mapped[int] = mapped_column(primary_key=True)
    retailer_id: Mapped[int] = mapped_column(ForeignKey("retailers.id"), nullable=False)
    code: Mapped[str] = mapped_column(String(32), nullable=False)
    iata: Mapped[str | None] = mapped_column(String(4), index=True)
    name: Mapped[str] = mapped_column(String(160), nullable=False)
    city: Mapped[str | None] = mapped_column(String(120))
    country: Mapped[str | None] = mapped_column(String(80))
    currency: Mapped[str] = mapped_column(String(3), nullable=False)
    is_catalogue_only: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
    # Display switch, distinct from Source.enabled: a hidden shop keeps
    # collecting (its history accrues) but appears nowhere on the site. This is
    # how coverage is scoped per audience -- demo, launch, full. Off by default
    # (Decision 1): a newly collected airport is switched on deliberately.
    visible: Mapped[bool] = mapped_column(
        Boolean, default=False, server_default="false", nullable=False
    )

    retailer: Mapped[Retailer] = relationship(back_populates="shops")
    listings: Mapped[list["Listing"]] = relationship(back_populates="shop")


class Brand(Base, TimestampMixin):
    """One row per brand, whatever the shops call it (migration #3, Decision 6).

    `slug` is the fold key (`normalize.brand_key`, hyphenated): "Moët & Chandon",
    "MOET CHANDON" and "moet-chandon" are one row. `name` is the spelling shown,
    the most common one at fold time until a human sets it. `alias_of_id`
    points an alias ("YSL") at the brand it belongs to ("Yves Saint Laurent");
    NULL means the row is canonical itself. Products keep their `brand` text as
    collected; `brand_id` is the fold.
    """

    __tablename__ = "brands"

    id: Mapped[int] = mapped_column(primary_key=True)
    slug: Mapped[str] = mapped_column(String(160), unique=True, nullable=False)
    #: Born at insert: the identity a decision names across hosts (Stream K2, spec rule 3).
    uid: Mapped[uuid.UUID] = mapped_column(Uuid(as_uuid=True), nullable=False, unique=True, default=uuid.uuid4)
    #: The materialised publish decisions (plan W18): hidden takes a page off the site; indexed
    #: lets it into the sitemap. Written only through the ledger (`services/decisions`).
    hidden: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default=text("false"))
    indexed: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default=text("false"))
    name: Mapped[str] = mapped_column(String(160), nullable=False)
    alias_of_id: Mapped[int | None] = mapped_column(ForeignKey("brands.id"), index=True)
    # Who pointed this row at its brand (or chose its name) and when: the alias is a
    # recorded human decision, never a rule's guess (migration #6, Stream M).
    decided_by: Mapped[int | None] = mapped_column(ForeignKey("accounts.id"))
    decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    # The brand's picture (Stream AW3, migration aw3b1c2d3e4f): a brand mark, with who supplied
    # it and under what licence; `image_level` says what it depicts. Written only through
    # `services/imagery.set_image`, which never lets a fetched picture replace a supplied one.
    image_url: Mapped[str | None] = mapped_column(String(600))
    thumb_url: Mapped[str | None] = mapped_column(String(600))
    image_source: Mapped[str | None] = mapped_column(String(80))
    image_level: Mapped[str | None] = mapped_column(String(16))
    image_licence: Mapped[str | None] = mapped_column(String(200))
    image_attribution: Mapped[str | None] = mapped_column(String(400))
    image_set_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    product_variants: Mapped[list["ProductVariant"]] = relationship(back_populates="brand_row")
    lines: Mapped[list["ProductLine"]] = relationship(back_populates="brand_row")


class ProductLine(Base, TimestampMixin):
    """The real product above the priced one: "1 Million" over its EDT, Parfum and Elixir
    at every size (migration #6, rian's decision of 12 Sep: one page per real product).

    One row per brand and line key (`services/product_lines.py`, a pure function of the name);
    `name` is the preferred spelling, the most common one at backfill time until a person
    sets it; `slug` is the address the line page will answer at. `alias_of_id` points an
    alias line at the one it was folded into (never deleted), and `decided_by` /
    `decided_at` say who and when. A product joins its line through `product_line_id`; readers
    follow the alias to the canonical.
    """

    __tablename__ = "product_lines"
    __table_args__ = (UniqueConstraint("brand_id", "key", name="uq_product_line_brand_key"),)

    id: Mapped[int] = mapped_column(primary_key=True)
    brand_id: Mapped[int] = mapped_column(ForeignKey("brands.id"), nullable=False, index=True)
    #: The rule key, or `decided:<slug>` once a person named the line (spec §2): a namespace no
    #: rule produces, so an arrival never lands on an approved line.
    key: Mapped[str] = mapped_column(String(260), nullable=False)
    #: Born at insert: the identity a decision names across hosts (Stream K2, spec rule 3).
    uid: Mapped[uuid.UUID] = mapped_column(Uuid(as_uuid=True), nullable=False, unique=True, default=uuid.uuid4)
    #: The materialised publish decisions (plan W18): hidden takes a page off the site; indexed
    #: lets it into the sitemap. Written only through the ledger (`services/decisions`).
    hidden: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default=text("false"))
    indexed: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default=text("false"))
    #: An admin's pin (Stream AW2, migration aw2c1d2e3f4a): the line's best variant leads every
    #: featured list it is a comparison in. Written only through the ledger (`services/featured_pins`).
    featured: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default=text("false"))
    name: Mapped[str] = mapped_column(String(200), nullable=False)
    slug: Mapped[str] = mapped_column(String(240), unique=True, nullable=False)
    alias_of_id: Mapped[int | None] = mapped_column(ForeignKey("product_lines.id"), index=True)
    decided_by: Mapped[int | None] = mapped_column(ForeignKey("accounts.id"))
    decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    # The product line's picture (Stream AW3, migration aw3b1c2d3e4f): the best front bottle
    # shot the line holds, with its provenance; the same seven columns as `brands`, written
    # only through `services/imagery.set_image`.
    image_url: Mapped[str | None] = mapped_column(String(600))
    thumb_url: Mapped[str | None] = mapped_column(String(600))
    image_source: Mapped[str | None] = mapped_column(String(80))
    image_level: Mapped[str | None] = mapped_column(String(16))
    image_licence: Mapped[str | None] = mapped_column(String(200))
    image_attribution: Mapped[str | None] = mapped_column(String(400))
    image_set_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

    brand_row: Mapped[Brand] = relationship(back_populates="lines")
    product_variants: Mapped[list["ProductVariant"]] = relationship(back_populates="line")


class AttributeAlias(Base):
    """One wording of a attribute and the canonical one it means, per vertical.

    The attribute vocabulary (migration #6): "elixir parfum intense" and "elixir eau de
    parfum intense" are both the Elixir of a line; `display` is the wording shown. Rows
    are seeded by `backfill attribute_values` from the rules in `services/product_lines.py` and set by
    a person in the merge session (`decided_by`, `decided_at`); a rule never overwrites a
    row a person decided. A vertical with no attribute_values (drinks) has no rows.
    """

    __tablename__ = "attribute_aliases"
    __table_args__ = (UniqueConstraint("vertical", "raw", name="uq_attribute_alias_vertical_raw"),)

    id: Mapped[int] = mapped_column(primary_key=True)
    vertical: Mapped[str] = mapped_column(String(32), nullable=False)
    raw: Mapped[str] = mapped_column(String(80), nullable=False)
    canonical: Mapped[str] = mapped_column(String(80), nullable=False)
    display: Mapped[str | None] = mapped_column(String(80))
    # What kind of thing the attribute is (`lines.ATTRIBUTE_KINDS`: concentration, color,
    # flavor, ...); metadata beside the canonical string, never a key slot (Stream L).
    kind: Mapped[str | None] = mapped_column(String(16))
    decided_by: Mapped[int | None] = mapped_column(ForeignKey("accounts.id"))
    decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )


# The version of the identity rules a product was last resolved under. Bumped
# when match_key semantics or the veto set change; `rederive` recomputes rows
# carrying an older version. "1": brand+name+size key, size veto. "2": the
# per-vertical attribute veto (fragrance concentration) as well. "3": the key is
# brand | line | attribute | size with a form marker (migration #6, Stream M). "4": the
# size slot is a quantity with its unit in every dimension (`100ml`, `50g`, `200pcs`,
# `unknown`; a set on its sorted contents), so 100 ml never equals 100 g and an unknown
# never agrees with anything (Stream L, rian 15 Sep; `services/quantity.py`). "5": a
# attribute a reader finds never stays in the line key (a Makeup shade and a confectionery
# flavour the shop marked leave the line as the concentration does), a attribute is read only
# from what the shop marked, and a pack word in a liquor name is the expression unless the
# name is a pack or a set (the catalogue decisions of 15 Sep, §2.3, §2.5, §2.6).
IDENTITY_RULES_VERSION = "6"


class ProductVariant(Base, TimestampMixin):
    """A canonical product, keyed on GTIN where we have one.

    GTIN is what turns cross-retailer matching from fuzzy string comparison into
    a join, so it is the primary identity and everything else is a fallback.
    A merged product stays in the table with `merged_into_id` set (a tombstone
    that forwards, never a delete), so old links and observations keep meaning.
    """

    __tablename__ = "product_variants"

    id: Mapped[int] = mapped_column(primary_key=True)
    #: Born at insert: the identity a decision names across hosts (Stream K2, spec rule 3).
    uid: Mapped[uuid.UUID] = mapped_column(Uuid(as_uuid=True), nullable=False, unique=True, default=uuid.uuid4)
    gtin: Mapped[str | None] = mapped_column(String(20), unique=True, index=True)
    match_key: Mapped[str] = mapped_column(String(255), index=True, nullable=False)
    brand: Mapped[str | None] = mapped_column(String(160), index=True)
    brand_id: Mapped[int | None] = mapped_column(ForeignKey("brands.id"), index=True)
    name: Mapped[str] = mapped_column(String(400), nullable=False)
    vertical: Mapped[str] = mapped_column(String(32), default="liquor", nullable=False, index=True)
    category: Mapped[str | None] = mapped_column(String(80), index=True)
    # The stated volume pair from before v4 (70 cl, 100 ml; a unit of ml, l or cl and
    # nothing else: grams were never stored here, whatever the earlier comment said; a 75 g
    # stick sat here as 75.00 ml). It is our past parse, kept as the record, not the shop's
    # word; the shop's word is `listings.listed_quantity_text`. `quantity_ml` is the derived
    # millilitre figure the public site and the verification read, NULL for a weight or a
    # count; it stays the comparable figure for volumes.
    quantity_stated_value: Mapped[float | None] = mapped_column(Numeric(9, 2))
    quantity_stated_unit: Mapped[str | None] = mapped_column(String(8))
    quantity_ml: Mapped[int | None] = mapped_column(Integer)
    # The standard quantity (identity rules v4, `services/quantity.py`): the value in its
    # canonical unit (ml | g | pcs), a pack's count and unit value, the form (single | pack |
    # set | refill), a set's sorted contents, and whether a quantity was read at all
    # (stated | none | unparsed). Recomputable: `backfill quantities`, `rederive`, ingest.
    quantity_value: Mapped[float | None] = mapped_column(Numeric(10, 3))
    quantity_unit: Mapped[str | None] = mapped_column(String(8))
    pack_count: Mapped[int | None] = mapped_column(Integer)
    pack_unit_value: Mapped[float | None] = mapped_column(Numeric(10, 3))
    form: Mapped[str | None] = mapped_column(String(12))
    set_contents: Mapped[str | None] = mapped_column(String(120))
    quantity_state: Mapped[str | None] = mapped_column(String(12))
    # The `_flat` fold of the name, so "the shop's name differs from ours" is a SQL compare
    # against `listings.listed_name_key`.
    name_key: Mapped[str | None] = mapped_column(String(400))
    abv: Mapped[float | None] = mapped_column(Numeric(5, 2))
    # Per-vertical facts that VETO a fallback match and never key it (Decision
    # 6): {"concentration": "edp"} for a fragrance. A missing value never splits
    # a product; two declared values that differ never join one.
    # JSONB on Postgres; plain JSON elsewhere, so the identity backfills can be tested on the
    # in-memory SQLite the account suites use (tests/test_line_backfills.py).
    attributes: Mapped[dict] = mapped_column(
        JSON().with_variant(JSONB(), "postgresql"), default=dict, server_default="{}", nullable=False
    )
    identity_rules_version: Mapped[str] = mapped_column(
        String(16), default=IDENTITY_RULES_VERSION, server_default="1", nullable=False
    )
    merged_into_id: Mapped[int | None] = mapped_column(ForeignKey("product_variants.id"), index=True)
    #: An admin's pin (Stream AW2, migration aw2c1d2e3f4a): the variant leads every featured list
    #: it is a comparison in; it never admits one. Written only through the ledger.
    featured: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default=text("false"))
    # The line this product is one size and attribute of (migration #6); NULL until
    # `backfill lines` has run or for a product with no brand.
    product_line_id: Mapped[int | None] = mapped_column(ForeignKey("product_lines.id"), index=True)
    # Where the barcode came from: "collected" (a shop published it) or "merge" (it
    # arrived from a row folded into this one), so markup can tell (M7). NULL on rows
    # keyed before migration #6, which all collected theirs.
    gtin_source: Mapped[str | None] = mapped_column(String(16))
    country_of_origin: Mapped[str | None] = mapped_column(String(80))
    is_exclusive: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
    # The picture: brand-supplied or openly licensed (Open Food Facts by barcode), never
    # retailer photography. `image_source` is the controlled vocabulary in
    # `services/imagery.py` (`admin:<supplier>` or `public:<source>`), `image_level` what the
    # picture depicts, `image_licence` and `image_attribution` the record, `image_set_at` when it
    # was set; written only through `imagery.set_image`, which keeps a supplied picture against
    # any fetched one. image_checked records that Open Food Facts was asked, so a product
    # variant with no picture is not retried on every run.
    image_url: Mapped[str | None] = mapped_column(String(600))
    thumb_url: Mapped[str | None] = mapped_column(String(600))
    image_source: Mapped[str | None] = mapped_column(String(80))
    image_level: Mapped[str | None] = mapped_column(String(16))
    image_licence: Mapped[str | None] = mapped_column(String(200))
    image_attribution: Mapped[str | None] = mapped_column(String(400))
    image_set_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    image_checked: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)

    __table_args__ = (
        Index("ix_products_quantity", "quantity_unit", "quantity_value"),
        Index("ix_products_name_key", "name_key"),
    )

    listings: Mapped[list["Listing"]] = relationship(back_populates="variant", foreign_keys="Listing.variant_id")
    awards: Mapped[list["Award"]] = relationship(back_populates="variant")
    brand_row: Mapped[Brand | None] = relationship(back_populates="product_variants")
    line: Mapped[ProductLine | None] = relationship(back_populates="product_variants")


class Suggestion(Base):
    """Two things the rules suspect are one, queued for a person: two product variants, two
    brands or two lines (`level`, migration #6).

    `left_id` / `right_id` are the ids at that level. Every row carries a `reason` a person
    can read without the code and a `score`; `decided_by` is FK accounts, NULL for the
    algorithm; `decision` is same | separate | superseded | withdrawn | NULL (Confirm same and
    Keep separate, rian's words). A pair kept separate never resurfaces: `suggest` skips any
    pair that has a decision.
    """

    __tablename__ = "suggestions"
    __table_args__ = (
        UniqueConstraint("level", "left_id", "right_id", name="uq_suggestion_level_pair"),
    )

    id: Mapped[int] = mapped_column(primary_key=True)
    level: Mapped[str] = mapped_column(String(16), nullable=False, default="product", server_default="product", index=True)
    #: Born at insert: the identity a decision names across hosts (Stream K2, spec rule 3).
    uid: Mapped[uuid.UUID] = mapped_column(Uuid(as_uuid=True), nullable=False, unique=True, default=uuid.uuid4)
    #: Why the algorithm closed the pair (`superseded`, `withdrawn`); `decision` is a person's ruling only
    #: (`same`, `separate`), with the ledger row that carries it (spec §5).
    closed_reason: Mapped[str | None] = mapped_column(String(16))
    decision_id: Mapped[int | None] = mapped_column(ForeignKey("decisions.id", name="fk_suggestions_decision", use_alter=True))
    left_id: Mapped[int | None] = mapped_column(Integer)
    right_id: Mapped[int | None] = mapped_column(Integer)
    reason: Mapped[str] = mapped_column(String(40), nullable=False)
    score: Mapped[float | None] = mapped_column(Numeric(5, 3))
    detail: Mapped[dict | None] = mapped_column(JSON().with_variant(JSONB(), "postgresql"))
    detected_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )
    decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    decided_by: Mapped[int | None] = mapped_column(ForeignKey("accounts.id"))
    decision: Mapped[str | None] = mapped_column(String(16))

    @classmethod
    def is_open(cls):
        """A pair waiting for a person: no ruling and not closed by the algorithm (spec §5)."""
        return (cls.decision.is_(None)) & (cls.closed_reason.is_(None))


class Merge(Base):
    """The record of one merge: `from_id` now forwards to `to_id`.

    Written by `backfill merges` and by any later human merge; never deleted,
    so a merge can be read back and undone by hand. `merged_by` is FK accounts
    (NULL when the algorithm did it); `detail` keeps both sides as they were.
    """

    __tablename__ = "merges"

    id: Mapped[int] = mapped_column(primary_key=True)
    from_id: Mapped[int] = mapped_column(ForeignKey("product_variants.id"), nullable=False, index=True)
    to_id: Mapped[int] = mapped_column(ForeignKey("product_variants.id"), nullable=False, index=True)
    merged_by: Mapped[int | None] = mapped_column(ForeignKey("accounts.id"))
    #: The merge record beside its decision (plan W15): the `merged_into` row, the batch whose tail
    #: fold made it, and the undo row that reversed it (the record itself carries the reversal).
    decision_id: Mapped[int | None] = mapped_column(ForeignKey("decisions.id", name="fk_merges_decision", use_alter=True))
    batch_id: Mapped[int | None] = mapped_column(ForeignKey("decision_batches.id", name="fk_merges_batch", use_alter=True))
    reversed_by_id: Mapped[int | None] = mapped_column(ForeignKey("decisions.id", name="fk_merges_reversed_by", use_alter=True))
    reversed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    merged_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )
    reason: Mapped[str] = mapped_column(String(40), nullable=False, default="duplicate", server_default="duplicate")
    detail: Mapped[dict | None] = mapped_column(JSON().with_variant(JSONB(), "postgresql"))


class Reverification(Base):
    """A product or listing queued for a fresh read because something disagreed.

    Decision 7: disagreements are scored and queued, never silently resolved.
    Columns only for now (§10 #1 keeps the columns, defers the scorer); the
    queue is worked by `app.cli verify` once it reads from here.
    """

    __tablename__ = "reverifications"

    id: Mapped[int] = mapped_column(primary_key=True)
    variant_id: Mapped[int | None] = mapped_column(ForeignKey("product_variants.id"), index=True)
    listing_id: Mapped[int | None] = mapped_column(ForeignKey("listings.id"), index=True)
    reason: Mapped[str] = mapped_column(String(40), nullable=False)
    score: Mapped[float | None] = mapped_column(Numeric(6, 3))
    queued_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )
    done_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    result: Mapped[str | None] = mapped_column(String(24))
    detail: Mapped[dict | None] = mapped_column(JSONB)


class Listing(Base, TimestampMixin):
    """One product as carried by one shop, in three layers (rian, 14 Sep; Stream L).

    LISTED is the shop's own words, exactly as the fragment showed them: `listed_brand`,
    `listed_name`, `listed_variant`, `listed_quantity_text`, `listed_category`, `listed_gtin`,
    with the folds `listed_brand_key` (slug-shaped, compares with `brands.slug`) and
    `listed_name_key` (compares with `product_variants.name_key`) and the parsed
    `listed_quantity_value`/`_unit`. Written as one unit by one reader
    (`services/collected.listed_fields`) from the fragment named in `listed_record_id`, at
    ingest and by `backfill listed`; NULL where no fragment exists; never from ProductVariant
    fields; never trimmed, cased or unescaped differently from ingest. STANDARD identity
    lives on the product. DECIDED at listing level is a pin (`pinned_variant_id`: ingest
    honours it every sighting) or an ignore (`ignored_at`: prices keep collecting, the site's
    readers skip it through `catalog_queries.live_listings()`), each with who and when.
    `first_seen` is `created_at`; `last_seen_at` is the newest sighting.
    """

    __tablename__ = "listings"
    __table_args__ = (
        UniqueConstraint("shop_id", "source_sku", name="uq_listing_shop_sku"),
        Index("ix_listings_listed_brand_key", "listed_brand_key"),
        Index("ix_listings_listed_name_key", "listed_name_key"),
        Index("ix_listings_listed_gtin", "listed_gtin"),
        Index("ix_listings_listed_record_id", "listed_record_id"),
        Index("ix_listings_last_seen_at", "last_seen_at"),
        Index("ix_listings_pinned_product_id", "pinned_variant_id"),
    )

    id: Mapped[int] = mapped_column(primary_key=True)
    variant_id: Mapped[int] = mapped_column(ForeignKey("product_variants.id"), nullable=False, index=True)
    shop_id: Mapped[int] = mapped_column(ForeignKey("shops.id"), nullable=False, index=True)
    source_sku: Mapped[str] = mapped_column(String(96), nullable=False)
    url: Mapped[str | None] = mapped_column(String(600))
    listed_brand: Mapped[str | None] = mapped_column(Text)
    listed_name: Mapped[str | None] = mapped_column(Text)
    listed_variant: Mapped[str | None] = mapped_column(Text)
    listed_quantity_text: Mapped[str | None] = mapped_column(Text)
    listed_category: Mapped[str | None] = mapped_column(Text)
    listed_gtin: Mapped[str | None] = mapped_column(String(20))
    listed_brand_key: Mapped[str | None] = mapped_column(String(160))
    listed_name_key: Mapped[str | None] = mapped_column(String(400))
    listed_quantity_value: Mapped[float | None] = mapped_column(Numeric(10, 3))
    listed_quantity_unit: Mapped[str | None] = mapped_column(String(8))
    # `use_alter`: listings and raw_records reference each other; the flag names the cycle so
    # the test kit can order its drops (Postgres needs nothing).
    listed_record_id: Mapped[int | None] = mapped_column(
        ForeignKey("raw_records.id", name="fk_listings_listed_record", use_alter=True))
    last_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    pinned_variant_id: Mapped[int | None] = mapped_column(ForeignKey("product_variants.id", name="fk_listings_pinned_product"))
    pinned_by: Mapped[int | None] = mapped_column(ForeignKey("accounts.id", name="fk_listings_pinned_by"))
    pinned_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    ignored_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
    ignored_by: Mapped[int | None] = mapped_column(ForeignKey("accounts.id", name="fk_listings_ignored_by"))
    ignore_reason: Mapped[str | None] = mapped_column(Text)

    variant: Mapped[ProductVariant] = relationship(back_populates="listings", foreign_keys=[variant_id])
    shop: Mapped[Shop] = relationship(back_populates="listings")
    observations: Mapped[list["PriceObservation"]] = relationship(back_populates="listing")
    raw_records: Mapped[list["RawRecord"]] = relationship(back_populates="listing", foreign_keys="RawRecord.listing_id")


class RawRecord(Base):
    """The parsed fragment a listing was derived from, kept per run.

    The tile, the JSON-LD offer, the API item: what the collector actually
    saw, not the page. Every derived value can then be re-run over it when a
    rule changes (`app.cli rederive`), instead of re-fetching. parser_version
    names the collector rules that read it.
    """

    __tablename__ = "raw_records"

    id: Mapped[int] = mapped_column(primary_key=True)
    listing_id: Mapped[int] = mapped_column(ForeignKey("listings.id"), nullable=False, index=True)
    run_id: Mapped[int | None] = mapped_column(ForeignKey("collection_runs.id"), index=True)
    payload: Mapped[dict] = mapped_column(JSON().with_variant(JSONB(), "postgresql"), nullable=False)
    parser_version: Mapped[str] = mapped_column(String(40), nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )

    listing: Mapped[Listing] = relationship(back_populates="raw_records", foreign_keys=[listing_id])


class PriceObservation(Base):
    """A price seen at a point in time, with the context it was seen under."""

    __tablename__ = "price_observations"
    __table_args__ = (Index("ix_obs_listing_time", "listing_id", "observed_at"),)

    id: Mapped[int] = mapped_column(primary_key=True)
    listing_id: Mapped[int] = mapped_column(ForeignKey("listings.id"), nullable=False)
    price: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False)
    currency: Mapped[str] = mapped_column(String(3), nullable=False)
    price_usd: Mapped[float | None] = mapped_column(Numeric(12, 2))
    was_price: Mapped[float | None] = mapped_column(Numeric(12, 2))
    price_type: Mapped[str] = mapped_column(String(24), default="list", nullable=False)
    in_stock: Mapped[bool | None] = mapped_column(Boolean)
    observed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    run_id: Mapped[int | None] = mapped_column(ForeignKey("collection_runs.id"))
    # collector | browser | human. A hand-entered or browser-rendered price is
    # never mistaken for a collected one, and the back office can mark it stale.
    source_kind: Mapped[str] = mapped_column(
        String(16), default="collector", server_default="collector", nullable=False
    )
    # Units of `currency` per USD used to derive price_usd, so the derivation
    # is re-doable and a fallback-rate figure can be found and redone.
    fx_rate: Mapped[float | None] = mapped_column(Numeric(14, 6))

    listing: Mapped[Listing] = relationship(back_populates="observations")


class Award(Base, TimestampMixin):
    """A competition medal attached to a product.

    Generic by design: populated today from the existing competition network and
    from awards the retailers themselves publish; a new competition slots in
    with no schema change.
    """

    __tablename__ = "awards"
    __table_args__ = (
        UniqueConstraint("variant_id", "competition", "year", name="uq_award_product_comp_year"),
    )

    id: Mapped[int] = mapped_column(primary_key=True)
    variant_id: Mapped[int] = mapped_column(ForeignKey("product_variants.id"), nullable=False, index=True)
    competition: Mapped[str] = mapped_column(String(160), nullable=False)
    competition_slug: Mapped[str | None] = mapped_column(String(64), index=True)
    year: Mapped[int | None] = mapped_column(Integer)
    medal: Mapped[str | None] = mapped_column(String(64))
    score: Mapped[int | None] = mapped_column(Integer)
    is_own_competition: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
    source: Mapped[str | None] = mapped_column(String(64))

    variant: Mapped[ProductVariant] = relationship(back_populates="awards")