"""Fixtures and helpers for the transfer-ownership integration suite.

See README.md for the full scenario. In short: product 5244974 ("Travessia")
is slated to be transferred Ala Comunicação (vendor 81790) → Sued Nunes
Produções LTDA (vendor 797716) — and every endpoint must serve each owner a
view of the product's streams correctly sliced in time by
`PRODUCT_OWNERSHIP_ACCESS`.

The transfer is now live in QA data (confirmed 2026-06-05):
`FACTS.QA.PRODUCT_OWNERSHIP_ACCESS` carries both owners — vendor 81790 with
`access_until_date = 2026-04-30` and vendor 797716 current — and
`FACTS.QA.DIM_RELEASE` is updated to the current owner 797716. The boundary
day (2026-04-30, the last day of Ala's tenure) matches
`FACTS.QA.DIM_RELEASE_HISTORY`.

All three profiles carry real QA InsightsProfile ids: `employee` (FF-ON
474215 / FF-OFF 1100) is carried over from the prior scenario; `ala_vendor`
(vendor 81790: FF-ON 9900038 / FF-OFF 9900039) and `sued_vendor` (vendor
797716: FF-ON 9900042 / FF-OFF 9900043) are newly provisioned for this
scenario.
"""

from __future__ import annotations

import json
from typing import NamedTuple

import pytest
from owsrequest.constants import headers as owsrequest_headers

# ---------------------------------------------------------------------------
# Profiles  (all expected to carry insights_transfer_product_ownership in QA)
# ---------------------------------------------------------------------------
#
# The three profile keys map to roles in the scenario as follows:
#   - "employee"     → full-access oncall view
#   - "ala_vendor"   → ORIGINATING vendor 81790 (Ala Comunicação) — former
#                       owner whose tenure ended 2026-04-30
#   - "sued_vendor"  → DESTINATION vendor 797716 (Sued Nunes Produções LTDA)
#                       — current owner from 2026-05-01

# (key, InsightsProfile id) — key is the stable string used by the visibility
# matrix and pytest param ids.
PROFILES = (
    ("employee", "474215"),
    ("ala_vendor", "9900038"),
    ("sued_vendor", "9900042"),
)

# What each profile can reach. "FULL_ACCESS" is the employee sentinel; the
# others are a tuple of (account_id, account_type) pairs.
PROFILE_ACCOUNTS = {
    "employee": "FULL_ACCESS",
    "ala_vendor": (("81790", "vendor"),),
    "sued_vendor": (("797716", "vendor"),),
}


def _insights_headers(profile_id: str) -> dict:
    """Build the request header dict for an InsightsProfile id."""
    return {
        owsrequest_headers.ORCHARD_PROFILE_ID: profile_id,
        owsrequest_headers.ORCHARD_PROFILE_TYPE: "InsightsProfile",
    }


PROFILE_HEADERS = {key: _insights_headers(pid) for key, pid in PROFILES}

# Module-level aliases for direct use in tests.
EMPLOYEE = PROFILE_HEADERS["employee"]
ALA_VENDOR = PROFILE_HEADERS["ala_vendor"]
SUED_VENDOR = PROFILE_HEADERS["sued_vendor"]

# Parametrize axis: every profile, with the key as the pytest id.
TRANSFER_PROFILES = [
    pytest.param(headers, id=key) for key, headers in PROFILE_HEADERS.items()
]

# ---------------------------------------------------------------------------
# FF-OFF twins  (same account access, insights_transfer_product_ownership OFF)
# ---------------------------------------------------------------------------
#
# Each profile above has a flag-disabled twin with *identical* account access,
# provisioned in QA and listed in
# tests/integration/endpoint_diff/retarget_manifest.py. A request made with an
# FF-OFF profile runs the production (legacy) code path: current-ownership
# only, via dim_release — a former owner sees nothing of a transferred-away
# product; the current owner sees its whole history.
#
# Pairing each FF-ON profile with its FF-OFF twin turns the flag into a
# *differential oracle*. The flag only ever changes scoping for transferred
# products, so the FF-ON-minus-FF-OFF delta on any response is, by
# construction, exactly the contribution of the transfer. This is what lets
# the whole-catalogue endpoints — which cannot isolate product 5244974 with a
# query param — be tested precisely rather than merely smoke-tested.

PROFILES_FF_OFF = (
    ("employee", "1100"),
    ("ala_vendor", "9900039"),
    ("sued_vendor", "9900043"),
)

PROFILE_HEADERS_FF_OFF = {key: _insights_headers(pid) for key, pid in PROFILES_FF_OFF}


class ProfilePair(NamedTuple):
    """An FF-ON / FF-OFF profile twin sharing one account scope."""

    key: str  # stable profile key, e.g. "sued_vendor"
    ff_on: dict  # headers for the flag-enabled (v2) profile
    ff_off: dict  # headers for the flag-disabled (legacy) twin


# profile key -> ProfilePair. Both members share account access; only the
# insights_transfer_product_ownership flag differs.
PROFILE_PAIRS = {
    key: ProfilePair(key, PROFILE_HEADERS[key], PROFILE_HEADERS_FF_OFF[key])
    for key in PROFILE_HEADERS
}

# Parametrize axis for the differential tests: one ProfilePair per profile.
TRANSFER_PROFILE_PAIRS = [
    pytest.param(pair, id=key) for key, pair in PROFILE_PAIRS.items()
]

# ---------------------------------------------------------------------------
# Participation-axis profiles (LabelParticipant) — the two
# permission_label_participant_ids branches of _permissions_filter_v2
# ---------------------------------------------------------------------------
#
# The move that transfers product 5244974's *ownership* (vendor 81790 → 797716)
# also moves the artist's *participation*: lambda-kinesis-to-neo4j
# move_project.py re-points the PARTICIPATED_IN edge onto a freshly-created
# destination-vendor LabelParticipant and hard-deletes the originating LP's
# edge. So the two LP branches of the v2 filter are exercised by two DIFFERENT
# artist profiles — not by an FF-pair of one artist:
#
#   - DESTINATION artist (vendor 797716, LP 1567012701053) — the *current*
#     participant. Its row is live in LABEL_PARTICIPANT_PARTICIPATED_IN_ORCHARD_
#     PRODUCT, so the current LP arm matches and it sees the product's full
#     lifetime — like the sued_vendor current owner, via the participation axis.
#     The flag is a no-op: legacy reads the same current participation row
#     (untime-sliced) and also returns the full lifetime.
#
#   - ORIGINATING artist (vendor 81790, LP 1702386309417) — the *former*
#     participant. The move deleted its PARTICIPATED_IN row, so both the current
#     arm and the entire legacy path match nothing; only the
#     USED_TO_PARTICIPATE_IN shadow re-grants access, frozen to the former-owner
#     window by the historical arm — same shape as the ala_vendor former owner.
#     The flag *grants* the moved-away product back: FF-OFF == 0 (legacy hides
#     it), FF-ON == the W1 historical slice.
#
# Each artist is an FF-ON / FF-OFF ProfilePair, kept on its own parametrize axis
# (ARTIST_PROFILE_PAIRS / ARTIST_PROFILES, read by test_used_to_participate.py)
# rather than folded into the ownership-axis TRANSFER_PROFILES — the gate here
# is participation, not label ownership.
#
# Data-layer state (confirmed in FACTS.QA 2026-06-05):
#   - PRODUCT_OWNERSHIP_ACCESS is rebuilt: 81790 access_until_date 2026-04-30,
#     797716 current.
#   - LABEL_PARTICIPANT_PARTICIPATED_IN_ORCHARD_PRODUCT holds only the
#     destination LP 1567012701053 for product 5244974.
#   - LABEL_PARTICIPANT_USED_TO_PARTICIPATE_IN_ORCHARD_PRODUCT is backfilled
#     with the originating LP 1702386309417 (vendor_id 81790, subaccount_id 0).
# The originating grant depends on the permissions_filter.sql subaccount
# normalization (COALESCE 0/NULL) landing in the queried env — until it
# deploys to QA, the FF-ON originating profile reads 0 there (deploy-lag, same
# class as README §10 cluster A). It passes against a local dev server, which
# runs the patched macro over FACTS.QA.

DESTINATION_ARTIST_LP_ID = 1567012701053
ORIGINATING_ARTIST_LP_ID = 1702386309417

ORIGINATING_ARTIST = ProfilePair(
    key="originating_artist",
    ff_on=_insights_headers("9900036"),
    ff_off=_insights_headers("9900040"),
)
DESTINATION_ARTIST = ProfilePair(
    key="destination_artist",
    ff_on=_insights_headers("9900037"),
    ff_off=_insights_headers("9900041"),
)

# FF-pair axis (both artists) for the differential tests; FF-ON-only axis for
# the precise per-product visibility probes. id == the ProfilePair.key so the
# visibility matrix and pytest ids line up.
ARTIST_PROFILE_PAIRS = [
    pytest.param(ORIGINATING_ARTIST, id="originating_artist"),
    pytest.param(DESTINATION_ARTIST, id="destination_artist"),
]
ARTIST_PROFILES = [
    pytest.param(ORIGINATING_ARTIST.ff_on, id="originating_artist"),
    pytest.param(DESTINATION_ARTIST.ff_on, id="destination_artist"),
]

# ---------------------------------------------------------------------------
# The transferred product
# ---------------------------------------------------------------------------

PRODUCT_ID = "5244974"
PRODUCT_NAME = "Travessia"
UPC = "198000137366"
ARTIST_ID = 3125999
GLOBAL_PARTICIPANT_ID = "611a71d6-b9a9-4598-a83f-6aed1bf73904"

# Canonical single ISRC for the heavy parametrize axes — ~1,377,810 lifetime
# streams, and (unlike a compilation track) it appears only on product
# 5244974, so /sound-recording/BX69Y2100018/* is a clean product-5244974
# probe.
ISRC = "BX69Y2100018"

# All 13 ISRCs under product 5244974 (FACTS.PROD streams-by-track).
PRODUCT_ISRCS = [f"BX69Y21000{n:02d}" for n in range(12, 25)]

# ---------------------------------------------------------------------------
# Ownership timeline  (FACTS.QA.DIM_RELEASE_HISTORY, product 5244974)
# ---------------------------------------------------------------------------
#
#   ... 2026-04-30 | 2026-05-01 ... now
#   W1 Ala / 81790 | W2 Sued / 797716 (current)
#
# NB: PRODUCT_OWNERSHIP_ACCESS in QA does not yet reflect this transfer —
# tests are written for the post-transfer state (cutoff 2026-04-30 inclusive,
# Sued as current owner). They become live once POA is rebuilt.

# W1 — originating: Ala Comunicação vendor.
ORIGINATING_VENDOR_ID = 81790
ORIGINATING_ACCESS_UNTIL = "2026-04-30"  # inclusive last day of tenure

# W2 — destination (current): Sued Nunes Produções LTDA vendor.
DESTINATION_VENDOR_ID = 797716

# Account id reachable by each non-employee profile, and its type.
PROFILE_ACCOUNT = {
    "ala_vendor": (ORIGINATING_VENDOR_ID, "vendor"),
    "sued_vendor": (DESTINATION_VENDOR_ID, "vendor"),
}

# The /account/<id>/* endpoints time-slice a transferred product by the
# *account's* own ownership window, not the viewer's — account_scope_filter
# gates on PRODUCT_OWNERSHIP_ACCESS for the queried account. So the expected
# visibility of an account query is keyed by the account: each account shares
# the window of the transfer profile that owns it. Maps account_id -> profile
# key whose _VISIBILITY row applies.
ACCOUNT_WINDOW_PROFILE = {
    ORIGINATING_VENDOR_ID: "ala_vendor",  # 81790  -> W1 (former)
    DESTINATION_VENDOR_ID: "sued_vendor",  # 797716 -> W2 (current)
}

# ---------------------------------------------------------------------------
# Date-range probes
# ---------------------------------------------------------------------------
#
# Each probe is the `start_date`/`end_date` query fragment. Probes deliberately
# sit *inside* a window, away from the transfer-boundary day — boundary
# behaviour is tested separately in test_boundary.py.
#
# `early` and `mid` both land in W1 (Ala's era); `late` is the only probe in
# W2 (Sued's era). Two W1 probes are kept to preserve the suite's 3-probe
# parametrize axis from the prior scenario without sacrificing the time-slice
# coverage.

PROBES = {
    "early": "start_date=2025-01-01&end_date=2025-12-31",  # within W1, early
    "mid": "start_date=2026-01-01&end_date=2026-04-25",  # within W1, late
    "late": "start_date=2026-05-10&end_date=2026-05-25",  # within W2
    "full": "start_date=2024-10-24&end_date=2026-05-25",  # spans both windows
}

# Single-window probes — the visibility matrix is unambiguous for these.
WINDOW_PROBES = [pytest.param(p, id=p) for p in ("early", "mid", "late")]
# Every probe, including the all-spanning `full`.
ALL_PROBES = [pytest.param(p, id=p) for p in PROBES]

# Transfer-boundary day (the `access_until_date` value itself) — used by
# test_boundary.py to verify the former owner sees their cutoff day, since
# access_until_date is inclusive (README §7).
ORIGINATING_BOUNDARY = ORIGINATING_ACCESS_UNTIL  # "2026-04-30"

# ---------------------------------------------------------------------------
# Visibility matrix
# ---------------------------------------------------------------------------
#
# "data"  -> endpoint returns rows whose metric sums > 0
# "empty" -> endpoint returns no rows, or rows whose metric sums to 0
#
# Each owner sees [beginning-of-time, their cutoff]; windows are nested.
# Probes sit inside a window, away from the transfer-boundary day — the
# boundary day itself is covered by test_boundary.py.

_VISIBILITY: dict[tuple[str, str], str] = {
    # employee — full access, sees every window
    ("employee", "early"): "data",
    ("employee", "mid"): "data",
    ("employee", "late"): "data",
    ("employee", "full"): "data",
    # sued_vendor — destination/current owner, access_until_date IS NULL,
    # sees everything
    ("sued_vendor", "early"): "data",
    ("sued_vendor", "mid"): "data",
    ("sued_vendor", "late"): "data",
    ("sued_vendor", "full"): "data",
    # ala_vendor — originating former owner, cutoff 2026-04-30:
    # sees W1 (early/mid), not W2 (late)
    ("ala_vendor", "early"): "data",
    ("ala_vendor", "mid"): "data",
    ("ala_vendor", "late"): "empty",
    ("ala_vendor", "full"): "data",
    # originating_artist — former participant frozen at the same move boundary
    # via USED_TO_PARTICIPATE_IN; identical shape to ala_vendor (the former
    # owner), reached through the participation axis. Read by
    # test_used_to_participate.py — see ORIGINATING_ARTIST above.
    ("originating_artist", "early"): "data",
    ("originating_artist", "mid"): "data",
    ("originating_artist", "late"): "empty",
    ("originating_artist", "full"): "data",
    # destination_artist — current participant (live PARTICIPATED_IN row), sees
    # the product's full lifetime in every window; identical shape to
    # sued_vendor (the current owner), via the participation axis.
    ("destination_artist", "early"): "data",
    ("destination_artist", "mid"): "data",
    ("destination_artist", "late"): "data",
    ("destination_artist", "full"): "data",
}

# NOTE: /account/<id>/* endpoints have no separate visibility matrix — they
# reuse _VISIBILITY via ACCOUNT_WINDOW_PROFILE, which maps each account to the
# transfer profile whose ownership window it shares. test_account.py isolates
# product 5244974 on the timeseries/summary endpoints with
# `type=*_BY_PRODUCT&ids=5244974`; the products/top-content/video endpoints
# have no per-product selector and are tested with the FF-pair differential
# below.

# ---------------------------------------------------------------------------
# Recent-window visibility
# ---------------------------------------------------------------------------
#
# The last-28-day and 7-day-rollup endpoints (`/aggregated-streams`,
# `/metrics-by-track`, `/top-markets`) probe only the *current* era (W2). The
# profiles below still see that window; a former owner — or, on the
# participation axis, a former participant — is frozen at 2026-04-30, so as
# soon as the transfer cutoff is comfortably outside the 28-day window they
# must see zero (no forward leak on a ROLLUP endpoint). destination_artist is
# the *current* participant and sees the recent window; originating_artist
# (the former participant) does not.

SEES_RECENT_WINDOW = {"employee", "sued_vendor", "destination_artist"}

# ---------------------------------------------------------------------------
# Ownership roles  (drive the FF-pair differential)
# ---------------------------------------------------------------------------
#
# A *former owner* cannot see product 5244974 at all without the flag — legacy
# current-ownership hides the transferred-away product — so the flag *grants*
# it a time-sliced historical slice: FF-OFF figure == 0, FF-ON figure > 0.
#
# A *current-view* profile sees product 5244974's full lifetime under both
# flags — the current owner via dim_release, the employee via the is_current
# rollup rows — so the flag is a no-op for the product: FF-ON == FF-OFF.
#
# For viewer-scoped endpoints the role is the viewer's own key. For the
# account-scoped /account/<id>/* endpoints it is the *queried account's* key:
# account_scope_filter time-slices by the account, not the viewer, so use
# ACCOUNT_WINDOW_PROFILE to map the account id to its role key.
#
# The participation axis mirrors the ownership axis: originating_artist is a
# former participant (its PARTICIPATED_IN edge was deleted by the move, so the
# flag *grants* it the USED_TO_PARTICIPATE_IN historical slice — FF-OFF == 0,
# FF-ON > 0, exactly like a former owner), and destination_artist is the
# current participant (live PARTICIPATED_IN row, so the flag is a no-op, like a
# current owner). This lets the artist FF-pairs reuse assert_isolated_transfer.

FORMER_OWNERS = frozenset({"ala_vendor", "originating_artist"})
CURRENT_VIEW = frozenset({"employee", "sued_vendor", "destination_artist"})

# ---------------------------------------------------------------------------
# Lookups
# ---------------------------------------------------------------------------

# Participation-axis profile ids -> stable artist key. Both the FF-ON profile
# and its FF-OFF twin resolve to the same key, mirroring PROFILE_HEADERS /
# PROFILE_HEADERS_FF_OFF for the ownership axis.
_ARTIST_KEY_BY_ID = {
    headers[owsrequest_headers.ORCHARD_PROFILE_ID]: pair.key
    for pair in (ORIGINATING_ARTIST, DESTINATION_ARTIST)
    for headers in (pair.ff_on, pair.ff_off)
}


def profile_key_for(headers: dict) -> str:
    """Reverse-lookup the short profile key from a header dict.

    Resolves both the FF-ON profiles and their FF-OFF twins to the same key,
    for the ownership axis (employee/ala_vendor/sued_vendor) and the
    participation axis (originating_artist/destination_artist) alike.
    """
    target = headers[owsrequest_headers.ORCHARD_PROFILE_ID]
    for table in (PROFILE_HEADERS, PROFILE_HEADERS_FF_OFF):
        for key, hdrs in table.items():
            if hdrs[owsrequest_headers.ORCHARD_PROFILE_ID] == target:
                return key
    if target in _ARTIST_KEY_BY_ID:
        return _ARTIST_KEY_BY_ID[target]
    raise KeyError(f"Unknown profile id in headers: {target}")


def expected_visibility(profile_key: str, probe: str) -> str:
    """Return 'data' or 'empty' for a (profile, probe) pair."""
    return _VISIBILITY[(profile_key, probe)]


def profile_can_reach_account(profile_key: str, account_id: int) -> bool:
    """True if the profile is allowed to query /account/<account_id>/*.

    The employee profile reaches every account; each label/subaccount profile
    reaches only its own account.
    """
    if profile_key == "employee":
        return True
    own = PROFILE_ACCOUNT.get(profile_key)
    return bool(own) and own[0] == int(account_id)


# ---------------------------------------------------------------------------
# Assertion helpers
# ---------------------------------------------------------------------------


def _sum_metric(items: list, metric_key: str) -> float:
    """Sum `metric_key` across items, treating missing/None as 0."""
    return sum((item.get(metric_key) or 0) for item in items)


def assert_visibility(
    payload: dict,
    *,
    profile_key: str,
    probe: str,
    items_key: str | None = "items",
    metric_key: str = "streams",
) -> None:
    """Assert the payload matches the visibility matrix for (profile, probe).

    - "data"  -> items non-empty AND the metric sums > 0
    - "empty" -> items empty OR the metric sums to 0

    For payloads that are a single object rather than an items list, pass
    `items_key=None`; the metric is then read from the top-level payload.
    """
    expectation = expected_visibility(profile_key, probe)
    _assert_against_expectation(
        payload,
        expectation=expectation,
        label=f"profile={profile_key} probe={probe}",
        items_key=items_key,
        metric_key=metric_key,
    )


def _assert_against_expectation(
    payload: dict,
    *,
    expectation: str,
    label: str,
    items_key: str | None,
    metric_key: str,
) -> None:
    """Shared 'data' / 'empty' assertion body."""
    if items_key:
        items = payload.get(items_key) or []
        total = _sum_metric(items, metric_key)
        if expectation == "empty":
            assert total == 0, (
                f"{label}: expected empty but {len(items)} item(s) "
                f"totalled {total} on '{metric_key}'"
            )
        else:
            assert items, f"{label}: expected data but items were empty"
            assert total > 0, (
                f"{label}: expected data but items totalled 0 on '{metric_key}'"
            )
    else:
        value = payload.get(metric_key) or 0
        if expectation == "empty":
            assert value == 0, f"{label}: expected empty but {metric_key}={value}"
        else:
            assert value > 0, f"{label}: expected data but {metric_key}={value}"


# ---------------------------------------------------------------------------
# FF-pair differential helpers
# ---------------------------------------------------------------------------
#
# These compare a metric measured with the FF-ON profile against the same
# metric measured with its FF-OFF twin. Three laws hold (see README §6):
#
#   monotonicity  — FF-ON >= FF-OFF always; v2 only ever *grants* visibility.
#   no-op         — FF-ON == FF-OFF where v2 resolves to current-ownership for
#                   the viewer (the flag must not move that figure).
#   grant         — for a former owner's product-5244974 figure: FF-OFF is 0
#                   (legacy hides the transferred-away product) and FF-ON is
#                   the v2-granted historical slice.

# Fractional slack on the FF-pair comparisons — absorbs DIM_RELEASE CDC
# rebuilds between the two requests (see the project_full_set_aggregation
# memory: catalogue-wide aggregates drift run-to-run).
FF_DRIFT = 0.02


def find_row(items: list[dict] | None, key: str, value) -> dict | None:
    """Return the first row whose `key` stringifies to `value`, else None."""
    target = str(value)
    for item in items or []:
        if str(item.get(key)) == target:
            return item
    return None


def rows_for(items: list[dict] | None, key: str, values) -> list[dict]:
    """Return every row whose `key` is one of `values`."""
    wanted = {str(v) for v in values}
    return [item for item in (items or []) if str(item.get(key)) in wanted]


def sum_metric(rows: list[dict] | None, metric_key: str) -> float:
    """Sum `metric_key` across rows, treating missing/None as 0."""
    return _sum_metric(rows or [], metric_key)


def assert_ff_monotonic(ff_on: float, ff_off: float, *, label: str) -> None:
    """FF-ON must see at least as much as FF-OFF.

    v2 only ever grants a former owner back its historical slice — it never
    removes data. A violation means the flag dropped a viewer's data, or
    PRODUCT_OWNERSHIP_ACCESS disagrees with dim_release.
    """
    assert ff_on >= ff_off * (1 - FF_DRIFT), (
        f"{label}: the flag REMOVED data — FF-ON={ff_on} < FF-OFF={ff_off}. "
        f"v2 must never show a viewer less than legacy does."
    )


def assert_ff_noop(
    ff_on: float, ff_off: float, *, label: str, tolerance: float = FF_DRIFT
) -> None:
    """FF-ON and FF-OFF must agree — the flag is a no-op for this figure.

    Holds where v2 resolves to current-ownership for the viewer: the employee
    (is_current rollup rows) on viewer-scoped endpoints, and every profile on
    non-ownership-scoped endpoints. Any divergence is a v2 regression on the
    common, non-transferred-product code path.
    """
    margin = max(abs(ff_off), 1.0) * tolerance
    assert abs(ff_on - ff_off) <= margin, (
        f"{label}: the flag changed a figure it must not — "
        f"FF-ON={ff_on}, FF-OFF={ff_off} (allowed ±{margin})."
    )


def assert_ff_grants_transfer(
    ff_on: float, ff_off: float, *, label: str, require_data: bool = True
) -> None:
    """Assert product 5244974's own figure for a *former owner*.

    Legacy (FF-OFF) hides the transferred-away product entirely, so its figure
    must be 0 — a non-zero FF-OFF is a legacy permission leak. v2 (FF-ON)
    grants the time-sliced historical slice; with `require_data` it must be
    positive (set False only where QA coverage is genuinely unverified, e.g.
    TikTok — there the 0-leak check still holds).
    """
    assert ff_off == 0, (
        f"{label}: legacy LEAKED a transferred-away product — FF-OFF={ff_off}, "
        f"must be 0 (a former owner cannot see product 5244974 without the flag)."
    )
    if require_data:
        assert ff_on > 0, (
            f"{label}: v2 did NOT grant the former owner its historical "
            f"slice — FF-ON={ff_on}, expected > 0."
        )


def assert_isolated_transfer(
    ff_on: float,
    ff_off: float,
    *,
    role_key: str,
    label: str,
    require_data: bool = True,
) -> None:
    """Assert an FF-pair figure that *is product 5244974's own metric*.

    `role_key` is the profile/account key whose ownership window applies — the
    viewer for viewer-scoped endpoints, the queried account for the
    account-scoped /account/<id>/* endpoints. For a current-view role the flag
    is a no-op for the product (FF-ON == FF-OFF); for a former owner the flag
    grants the historical slice (FF-OFF == 0, FF-ON > 0). Monotonicity is
    asserted for every role, including the unclassified ones.
    """
    assert_ff_monotonic(ff_on, ff_off, label=label)
    if role_key in CURRENT_VIEW:
        assert_ff_noop(ff_on, ff_off, label=label)
    elif role_key in FORMER_OWNERS:
        assert_ff_grants_transfer(ff_on, ff_off, label=label, require_data=require_data)


def _normalize(obj):
    """Recursively make a JSON payload order-insensitive for comparison."""
    if isinstance(obj, dict):
        return {key: _normalize(value) for key, value in obj.items()}
    if isinstance(obj, list):
        normalized = [_normalize(item) for item in obj]
        return sorted(
            normalized, key=lambda x: json.dumps(x, sort_keys=True, default=str)
        )
    return obj


def assert_ff_pair_equal(ff_on: dict, ff_off: dict, *, label: str) -> None:
    """Assert the FF-ON and FF-OFF payloads are identical (order-insensitive).

    For endpoints where the flag is, by design, not an ownership-scoping
    change — it only swaps a fact table, or the endpoint is not
    ownership-scoped at all — every viewer must get a byte-equal response with
    the flag on and off. Use only for small/stable payloads; a catalogue-wide
    aggregate drifts run-to-run and needs the tolerance-based assert_ff_noop.
    """
    assert _normalize(ff_on) == _normalize(ff_off), (
        f"{label}: the flag changed the response — FF-ON and FF-OFF differ, "
        f"but this endpoint's v2 path must be metric-neutral."
    )