"""meta_store.py — the REST loader for a mirror whose only other loader is XML-RPC (W31-T48).

    python platform/harness/meta_store.py --sync                 pull into tenant #0's mirror
    python platform/harness/meta_store.py --sync --tenant gtmlab
    python platform/harness/meta_store.py --status               what is in the mirror now

⛔ WHY THIS IS A SIBLING AND NOT A ROW IN `datastore.ENTITIES`. `datastore.sync_entity` is
hardwired to XML-RPC — it does `import core.odoo as O` and speaks `search_read` — so no amount of
spec data makes it fetch over HTTPS. What IS reusable is everything *below* the fetch: the
per-tenant file (`path_for`), the process-singleton connection (`connect`), and the
delete-then-insert upsert. This module borrows those and brings its own reader. **The mirror is one
store with two loaders, not two stores.**

⭐ THE SHAPE OF THE WHOLE THING, because it is the owner's actual request: *"we just need to pull
the data into our template database for Meta"*, working *"just like Odoo"*. Odoo's path is
`XML-RPC -> DuckDB mirror -> odoo_relational -> ut_odoo_* locked grids`. Meta's is
`Graph -> DuckDB mirror -> meta_relational -> ut_meta_* locked grids`. Same middle, same end, one
different first hop.

⚠ EVERY COLUMN NAME BELOW WAS MEASURED, NOT READ OFF A DOC (R8). `proto/meta-entity-fields.json`
carries the run: each level's list is what the API ACCEPTED when asked, on a real ad account.
  Ad Account 45 · Campaign 35 · Ad Set 54 · Ad 36 · Creative 58 · Insights 57
⛔ AND `ACCEPTED` IS NOT `RETURNED`. Graph omits null fields from a response, so a 20-field ask
came back with 14 keys. Building a schema from what came back would drop ~30% of the columns with
nothing going red — which is exactly the "do not drop any column" instruction, broken silently.
The lists below are therefore the ASKED-AND-ACCEPTED set, and a column with no value is NULL.

⚠ `owner` IS ABSENT FROM THE AD ACCOUNT LIST AND THAT IS A REPORTED LIMIT, NOT AN OMISSION: it
answers `403 (#200) Requires business_management permission`, which this token does not carry. One
column of 285. R6's second sentence — a limit that cannot be removed gets stated with its cause.

⛔ IDS ARE TEXT, ALWAYS. A Meta object id is a 17-digit decimal string; `120273975028650555` is
larger than 2^53, so any float or JS-number path silently corrupts it. Same ruling as `ig_id`
(W26/R3), for the same reason, and it is why every `id` column here is VARCHAR.
"""
import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

_HERE = Path(__file__).resolve().parent
if str(_HERE.parent) not in sys.path:
    sys.path.insert(0, str(_HERE.parent))

from harness import datastore                                    # noqa: E402

GRAPH_VERSION = os.environ.get("META_GRAPH_VERSION") or "v21.0"
GRAPH = f"https://graph.facebook.com/{GRAPH_VERSION}"

#: Numeric columns, by name. Everything else is VARCHAR — including ids (see the header) and
#: including anything Graph returns as a nested object, which is stored as compact JSON text.
_INT = {"impressions", "reach", "clicks", "unique_clicks", "inline_link_clicks",
        "inline_post_engagement", "full_view_impressions", "estimated_ad_recallers",
        "account_status", "age", "timezone_id", "timezone_offset_hours_utc", "io_number",
        "min_daily_budget", "min_campaign_group_spend_cap"}
_DBL = {"spend", "social_spend", "ctr", "unique_ctr", "cpc", "cpm", "cpp", "frequency",
        "inline_link_click_ctr", "outbound_clicks_ctr", "amount_spent", "balance", "spend_cap",
        "daily_budget", "lifetime_budget", "budget_remaining", "bid_amount", "canvas_avg_view_time",
        "daily_min_spend_target", "daily_spend_cap", "lifetime_min_spend_target",
        "lifetime_spend_cap", "lifetime_imps"}

#: ⭐ THE MEASURED CATALOG. `edge` is the connection on the ad account; `None` means the account
#: itself. `parent` names the column that ties a row to its parent, used by nothing here and by
#: the grid links in `meta_relational`.
ACCOUNT_FIELDS = (
    "account_id account_status age amount_spent balance business_city business_country_code "
    "business_name business_state business_street business_street2 business_zip capabilities "
    "created_time currency disable_reason end_advertiser end_advertiser_name funding_source "
    "has_migrated_permissions id io_number is_attribution_spec_system_default "
    "is_direct_deals_enabled is_notifications_enabled is_personal is_prepay_account "
    "is_tax_id_required line_numbers media_agency min_campaign_group_spend_cap min_daily_budget "
    "name offsite_pixels_tos_accepted partner spend_cap tax_id tax_id_status tax_id_type "
    "timezone_id timezone_name timezone_offset_hours_utc tos_accepted user_tasks user_tos_accepted"
).split()

CAMPAIGN_FIELDS = (
    "account_id bid_strategy boosted_object_id budget_rebalance_flag budget_remaining buying_type "
    "campaign_group_active_time can_create_brand_lift_study can_use_spend_cap configured_status "
    "created_time daily_budget effective_status id is_skadnetwork_attribution issues_info "
    "last_budget_toggling_time lifetime_budget name objective pacing_type primary_attribution "
    "promoted_object smart_promotion_type source_campaign source_campaign_id special_ad_categories "
    "special_ad_category special_ad_category_country spend_cap start_time status stop_time "
    "topline_id updated_time"
).split()

ADSET_FIELDS = (
    "account_id adlabels adset_schedule asset_feed_id attribution_spec bid_adjustments bid_amount "
    "bid_constraints bid_info bid_strategy billing_event budget_remaining campaign "
    "campaign_active_time campaign_attribution campaign_id configured_status created_time "
    "creative_sequence daily_budget daily_min_spend_target daily_spend_cap destination_type "
    "effective_status end_time frequency_control_specs id instagram_actor_id is_dynamic_creative "
    "issues_info learning_stage_info lifetime_budget lifetime_imps lifetime_min_spend_target "
    "lifetime_spend_cap multi_optimization_goal_weight name optimization_goal "
    "optimization_sub_event pacing_type promoted_object recurring_budget_semantics review_feedback "
    "rf_prediction_id source_adset source_adset_id start_time status targeting "
    "targeting_optimization_types time_based_ad_rotation_id_blocks "
    "time_based_ad_rotation_intervals updated_time use_new_app_click"
).split()

AD_FIELDS = (
    "account_id ad_active_time ad_review_feedback ad_schedule_end_time ad_schedule_start_time "
    "adlabels adset adset_id bid_amount bid_info bid_type campaign campaign_id configured_status "
    "conversion_domain created_time creative demolink_hash display_sequence effective_status "
    "engagement_audience failed_delivery_checks id issues_info last_updated_by_app_id name "
    "preview_shareable_link priority recommendations source_ad source_ad_id status targeting "
    "tracking_and_conversion_with_defaults tracking_specs updated_time"
).split()

CREATIVE_FIELDS = (
    "account_id actor_id adlabels applink_treatment asset_feed_spec authorization_category body "
    "branded_content_sponsor_page_id bundle_folder_id call_to_action_type categorization_criteria "
    "category_media_source collaborative_ads_lsb_image_bank_id degrees_of_freedom_spec "
    "destination_set_id dynamic_ad_voice effective_authorization_category "
    "effective_instagram_media_id effective_object_story_id enable_direct_install "
    "enable_launch_instant_app id image_crops image_hash image_url instagram_actor_id "
    "instagram_permalink_url instagram_story_id instagram_user_id interactive_components_spec "
    "link_deep_link_url link_destination_display_url link_og_id link_url "
    "messenger_sponsored_message name object_id object_store_url object_story_id object_story_spec "
    "object_type object_url place_page_set_id platform_customizations playable_asset_id "
    "portrait_customizations product_set_id recommender_settings source_instagram_media_id status "
    "template_url template_url_spec thumbnail_id thumbnail_url title url_tags "
    "use_page_actor_override video_id"
).split()

INSIGHT_FIELDS = (
    "account_currency account_id account_name action_values actions ad_id ad_name adset_id "
    "adset_name attribution_setting buying_type campaign_id campaign_name canvas_avg_view_time "
    "clicks conversion_rate_ranking conversion_values conversions cost_per_action_type "
    "cost_per_inline_link_click cost_per_thruplay cost_per_unique_click cpc cpm cpp ctr date_start "
    "date_stop engagement_rate_ranking estimated_ad_recallers frequency full_view_impressions "
    "impressions inline_link_click_ctr inline_link_clicks inline_post_engagement objective "
    "optimization_goal outbound_clicks outbound_clicks_ctr purchase_roas quality_ranking reach "
    "social_spend spend unique_clicks unique_ctr unique_outbound_clicks "
    "video_avg_time_watched_actions video_p100_watched_actions video_p25_watched_actions "
    "video_p50_watched_actions video_p75_watched_actions video_p95_watched_actions "
    "video_play_actions video_thruplay_watched_actions website_purchase_roas"
).split()

#: table -> (graph edge on the account | None, measured fields, parent column)
SPECS = {
    "meta_ad_accounts": (None, ACCOUNT_FIELDS, None),
    "meta_campaigns": ("campaigns", CAMPAIGN_FIELDS, "account_id"),
    "meta_adsets": ("adsets", ADSET_FIELDS, "campaign_id"),
    "meta_ads": ("ads", AD_FIELDS, "adset_id"),
    "meta_creatives": ("adcreatives", CREATIVE_FIELDS, "account_id"),
}

#: The daily Insights grain (R2). One row per (ad, day) — the id is synthesised because Insights
#: has no id of its own, and it must be STABLE so a re-run updates instead of appending.
INSIGHTS_TABLE = "meta_insights_daily"
INSIGHTS_LEVEL = os.environ.get("META_INSIGHTS_LEVEL") or "ad"

#: ⛔ INSIGHTS IS FETCHED IN TIME SLICES, AND THE REASON IS MEASURED, NOT DEFENSIVE. All 57 fields
#: at ad level over `last_90d` with a 100-row page answers **HTTP 500 "An unknown error occurred"**
#: — Graph's way of saying the synchronous query is too heavy (the async report-run API is the
#: other answer, and it costs a poll loop this does not need). The SAME 57 fields over 7 days at
#: page 25 answer 200. So the window is walked in slices with every column intact:
#:   57 fields · ad level · 7d · limit 25   -> 200, 25 rows
#:   57 fields · account level · 7d          -> 200, 7 rows
#: ⚠ Narrowing the FIELD list would also have "fixed" it, and that is the wrong fix twice over —
#: it drops columns R2 requires, and it does so invisibly.
INSIGHTS_DAYS = int(os.environ.get("META_INSIGHTS_DAYS") or 90)
INSIGHTS_SLICE = int(os.environ.get("META_INSIGHTS_SLICE_DAYS") or 7)
INSIGHTS_PAGE = int(os.environ.get("META_INSIGHTS_PAGE") or 25)


def _slices(days, size):
    """[(since, until)] covering the last `days`, oldest first, in `size`-day windows."""
    from datetime import date, timedelta
    end = date.today()
    start = end - timedelta(days=max(1, days) - 1)
    out, cur = [], start
    while cur <= end:
        stop = min(cur + timedelta(days=max(1, size) - 1), end)
        out.append((cur.isoformat(), stop.isoformat()))
        cur = stop + timedelta(days=1)
    return out

#: ⚠ A REAL CEILING, AND R6's SECOND SENTENCE APPLIES TO IT. Graph pages at 25-100 rows; this is
#: the number of PAGES a single edge may walk before the loader stops and SAYS it stopped. It is
#: not a row cap on a connected source (R6 forbids that) — it is a runaway guard, and reaching it
#: is reported as a problem, never absorbed.
MAX_PAGES = int(os.environ.get("META_MAX_PAGES") or 200)
PAGE = int(os.environ.get("META_PAGE_SIZE") or 100)


class MetaError(RuntimeError):
    """A Graph refusal carrying Meta's own words. Safe to print: no token ever reaches it."""


def token():
    """The Meta token from the environment or gitignored `platform/.env`; "" when absent.

    ⚠ Same resolver `aios-web/api/connectors_meta.py` uses. Duplicated deliberately and minimally:
    `platform/` must not import from `aios-web/api/`, which is the layering rule this repo keeps
    (`core` never imports up). Fifteen lines is the price of that boundary.
    """
    tok = os.environ.get("META_ADS_ACCESS_TOKEN") or ""
    if tok:
        return tok.strip()
    env = _HERE.parent / ".env"
    if env.exists():
        for line in env.read_text(encoding="utf-8", errors="replace").splitlines():
            if line.strip().startswith("META_ADS_ACCESS_TOKEN"):
                _, _, v = line.partition("=")
                return v.strip().strip('"').strip("'")
    return ""


def _get(path, tok, **params):
    params["access_token"] = tok
    url = f"{GRAPH}/{path.lstrip('/')}?" + urllib.parse.urlencode(params)
    req = urllib.request.Request(url, headers={"User-Agent": "aios-meta-store/1"})
    try:
        with urllib.request.urlopen(req, timeout=120) as r:
            return json.loads(r.read().decode("utf-8", "replace"))
    except urllib.error.HTTPError as e:
        try:
            msg = (json.loads(e.read().decode("utf-8", "replace")).get("error") or {}
                   ).get("message") or ""
        except Exception:
            msg = ""
        raise MetaError(f"HTTP {e.code} on /{path.lstrip('/')}: {msg[:200]}") from None
    except Exception as e:
        raise MetaError(f"{type(e).__name__} on /{path.lstrip('/')}") from None


#: Graph's own words when a page is too heavy. It arrives as an HTTP **500**, not a 4xx, which is
#: why it cannot be treated as "the server is broken, give up".
_TOO_MUCH = "reduce the amount of data"

#: ⛔ THE ADS-MANAGEMENT RATE LIMIT, WHICH IS PER AD ACCOUNT AND NOT PER TOKEN. Measured: after a
#: heavy backfill Graph answers **HTTP 400 "There have been too many calls to this ad-account.
#: Wait a bit and try again."** It is a 4xx, so nothing about the status code says "retry" — the
#: MESSAGE is the only signal, which is why it is matched here rather than inferred from a code.
#: ⚠ It persists for minutes, so the backoff is measured in minutes and BOUNDED: after
#: `_RATE_TRIES` waits the loader STOPS and reports what it got, rather than sitting in a retry
#: loop nobody can see. A partial mirror that says it is partial beats a hung sync.
_RATE_LIMITED = "too many calls"
_RATE_TRIES = int(os.environ.get("META_RATE_TRIES") or 3)
_RATE_WAIT = int(os.environ.get("META_RATE_WAIT_S") or 90)


def _walk(path, tok, log, **params):
    """Every page of an edge, paged by CURSOR under our own parameters.

    ⛔ IT DOES NOT FOLLOW GRAPH'S `paging.next` URL, AND THAT IS THE WHOLE POINT. Measured: the
    first call to `/adcreatives` at limit=100 answers **HTTP 500 "Please reduce the amount of data
    you're asking for"**, the retry at limit=25 succeeds — and then `next` carries the ORIGINAL
    limit=100 and fails again on page 2. A backoff that cannot reach every page is not a backoff.
    Re-issuing each page ourselves with `after=<cursor>` keeps the reduced limit for the whole walk.

    ⭐ It shrinks the PAGE, never the FIELD LIST. Dropping columns to make a request fit is the
    silent omission R2 forbids, and nothing downstream could see it. Fewer rows per call, always
    every column per row.

    -> (rows, hit_page_cap)
    """
    out, pages = [], 0
    limit = int(params.pop("limit", None) or PAGE)
    after, waited = None, 0
    while True:
        call = dict(params, limit=limit)
        if after:
            call["after"] = after
        try:
            body = _get(path, tok, **call)
        except MetaError as e:
            if _TOO_MUCH in str(e) and limit > 5:
                limit = max(5, limit // 4)
                log(f"    page too heavy for /{path} - retrying at limit={limit} "
                    f"(a payload ceiling, not a row cap; every column is still asked for)")
                continue
            if _RATE_LIMITED in str(e).lower() and waited < _RATE_TRIES:
                waited += 1
                log(f"    rate-limited on /{path} (per AD ACCOUNT, not per token) - waiting "
                    f"{_RATE_WAIT}s, attempt {waited}/{_RATE_TRIES}")
                time.sleep(_RATE_WAIT)
                continue
            if _RATE_LIMITED in str(e).lower():
                log(f"    !! GIVING UP on /{path} after {waited} waits: still rate-limited. "
                    f"{len(out)} row(s) collected so far are kept. Cause: the ads-management "
                    f"limit is per ad account and persists for minutes. Fix: re-run "
                    f"`--sync --only <table>` later; the upsert is idempotent.")
                return out, True
            raise
        out.extend(body.get("data") or [])
        pages += 1
        after = ((body.get("paging") or {}).get("cursors") or {}).get("after")
        has_next = bool((body.get("paging") or {}).get("next")) and bool(after)
        if not has_next:
            return out, False
        if pages >= MAX_PAGES:
            log(f"    !! STOPPED at MAX_PAGES={MAX_PAGES} on /{path} with more pages left. "
                f"Cause: a runaway guard, not a row cap. Fix: raise META_MAX_PAGES, or narrow the "
                f"window with META_INSIGHTS_DAYS.")
            return out, True


def _cell(value):
    """One Graph value -> one DuckDB cell. Nested structures become compact JSON TEXT rather than
    being dropped: R2 says every field the API returns, and `targeting` is a field."""
    if value is None or isinstance(value, (str, int, float, bool)):
        return value
    return json.dumps(value, separators=(",", ":"), ensure_ascii=False)


def _coltype(name, rows=None):
    """The DuckDB type for one column — decided by the DATA when there is data, by the name only
    as a fallback.

    ⛔ THE NAME LIST WAS WRONG AND ONLY THE API COULD SAY SO. `cost_per_unique_click`,
    `cost_per_action_type`, `purchase_roas` and friends READ like money and are **arrays of
    action-type objects** on the Insights edge:
        [{"action_type":"outbound_click","value":"1.459854"}]
    Typed DOUBLE from `_DBL`, the insert died with `Conversion Error: Could not convert string
    '[{...}]' to DOUBLE` — after five entity tables had already been written, so the sync looked
    like it worked and then blew up on the last table.
    ⭐ Same principle the field CATALOG is built on, applied one layer down: **ask the response,
    do not assert from a list.** A value that ever arrives as a list or dict is JSON TEXT, because
    that is what `_cell` stores; anything else falls back to the measured name hints.
    """
    if rows:
        seen, kinds = 0, set()
        for r in rows:
            v = r.get(name)
            if v is None or v == "":
                continue
            kinds.add("json" if isinstance(v, (list, dict)) else
                      "bool" if isinstance(v, bool) else
                      "int" if isinstance(v, int) else
                      "float" if isinstance(v, float) else "str")
            seen += 1
            if seen >= 200:
                break
        if kinds:
            if "json" in kinds or "str" in kinds:
                return "VARCHAR"                              # a JSON blob, or a numeric STRING
            if kinds <= {"int", "bool"}:
                return "BIGINT" if name in _INT else ("VARCHAR" if "bool" in kinds else "BIGINT")
            return "DOUBLE"
    if name in _INT:
        return "BIGINT"
    if name in _DBL:
        return "DOUBLE"
    return "VARCHAR"                                          # ids included — see the header


def _ensure(con, table, fields, rows=None):
    """Create or widen the table, with every column typed from `rows` when they are available.

    ⚠ AN EXISTING COLUMN WHOSE TYPE IS NOW WRONG IS REBUILT, NOT PATCHED. DuckDB cannot retype a
    column in place, and this table is DERIVED data that can be re-pulled in minutes — so a type
    disagreement drops and recreates rather than limping on with a column that refuses every
    insert. The alternative is a mirror that is permanently unwritable for one bad guess.
    """
    want = {f: _coltype(f, rows) for f in fields}
    have = {r[1]: str(r[2]).upper() for r in con.execute(f"PRAGMA table_info('{table}')").fetchall()}
    if have:
        clash = [f for f, ty in want.items() if f in have and have[f] != ty
                 and not (have[f].startswith("VARCHAR") and ty == "VARCHAR")]
        if clash:
            con.execute(f"DROP TABLE {table}")
            have = {}
    if not have:
        cols = ", ".join(f"{f} {want[f]}" for f in fields)
        con.execute(f"CREATE TABLE IF NOT EXISTS {table} (id VARCHAR PRIMARY KEY, {cols})"
                    if "id" not in fields else
                    f"CREATE TABLE IF NOT EXISTS {table} ({cols})")
        return
    for f in fields:
        if f not in have:
            con.execute(f"ALTER TABLE {table} ADD COLUMN {f} {want[f]}")


def _upsert(con, table, fields, rows):
    """Delete-then-insert by id — the same idempotence `datastore._upsert` gives the Odoo half, so
    a re-sync updates in place and can never append a second copy of the same object."""
    if not rows:
        return 0
    cols = list(fields)
    ids = [str(r.get("id") or "") for r in rows]
    q = ",".join("?" for _ in ids)
    con.execute(f"DELETE FROM {table} WHERE id IN ({q})", ids)
    con.executemany(
        f"INSERT INTO {table} ({', '.join(cols)}) VALUES ({', '.join('?' for _ in cols)})",
        [[_cell(r.get(c)) for c in cols] for r in rows])
    return len(rows)


def sync(tenant_key="royal-imports", tok=None, log=print, insights=True, insights_days=None):
    """Pull every level into the tenant's mirror. -> a report dict; raises only on a bad token.

    Idempotent: re-running updates rows in place. Safe to call at boot and on the resync loop,
    exactly as `odoo_relational.refresh` is.
    """
    tok = tok or token()
    # ⛔ THE WINDOW IS A PARAMETER, NOT AN ENVIRONMENT READ AT CALL TIME — and the difference cost a
    # live deploy. `INSIGHTS_DAYS` binds at IMPORT (module scope), so `main._pull_meta`'s
    # `os.environ.setdefault("META_INSIGHTS_DAYS", "7")` executed AFTER this module was already
    # imported and changed nothing: every boot pulled **90 days**, not 7, which is the slow path
    # that trips the per-ad-account rate limit and never finishes. The comment beside that call
    # claimed a short window the code could not deliver.
    # ⭐ The shape: **a knob read at import cannot be turned by a caller at runtime.** Passing it
    # makes the caller's intent effective instead of aspirational; the env var stays the DEFAULT.
    days = int(insights_days or INSIGHTS_DAYS)
    report = {"tenant": tenant_key, "tables": {}, "problems": [], "accounts": []}
    if not tok:
        report["problems"].append(
            "META_ADS_ACCESS_TOKEN is not set in the environment or in platform/.env, so nothing "
            "was pulled. This is a missing CREDENTIAL, not a missing capability.")
        return report

    path = datastore.path_for(tenant_key)
    if Path(datastore.DB_PATH) != Path(path):
        datastore.use_path(path)
    con = datastore.connect()
    log(f"  mirror: {Path(path).name}")

    accts, _ = _walk("me/adaccounts", tok, log, fields="id,name", limit=PAGE)
    report["accounts"] = [a.get("id") for a in accts]
    if not accts:
        report["problems"].append("the token reaches no ad accounts")
        return report

    # ── the five entity levels ────────────────────────────────────────────────────────────────
    for table, (edge, fields, _parent) in SPECS.items():
        rows, capped = [], False
        for acct in accts:
            aid = acct["id"]
            if edge is None:
                rows.append(_get(aid, tok, fields=",".join(fields)))
            else:
                got, hit = _walk(f"{aid}/{edge}", tok, log, fields=",".join(fields), limit=PAGE)
                rows.extend(got)
                capped = capped or hit
        _ensure(con, table, fields, rows)
        n = _upsert(con, table, fields, rows)
        total = con.execute(f"SELECT count(*) FROM {table}").fetchone()[0]
        report["tables"][table] = {"pulled": n, "in_mirror": total, "capped": capped}
        log(f"  {table:<20} pulled {n:>6}   mirror total {total:>6}"
            + ("   !! PAGE CAP HIT" if capped else ""))
        if capped:
            report["problems"].append(f"{table}: stopped at MAX_PAGES={MAX_PAGES}")

    # ── the daily Insights grain ──────────────────────────────────────────────────────────────
    if insights:
        rows, capped = [], False
        for acct in accts:
            for since, until in _slices(days, INSIGHTS_SLICE):
                got, hit = _walk(f"{acct['id']}/insights", tok, log,
                                 fields=",".join(INSIGHT_FIELDS), level=INSIGHTS_LEVEL,
                                 time_increment="1", limit=INSIGHTS_PAGE,
                                 time_range=json.dumps({"since": since, "until": until}))
                rows.extend(got)
                capped = capped or hit
        # ⛔ Insights rows carry no id. The key must be STABLE across runs or every re-sync
        # appends a second copy of the same day — so it is composed from the grain itself.
        for r in rows:
            r["id"] = ":".join(str(r.get(k) or "") for k in
                               (f"{INSIGHTS_LEVEL}_id", "date_start", "date_stop"))
        fields = ["id"] + INSIGHT_FIELDS
        _ensure(con, INSIGHTS_TABLE, fields, rows)
        n = _upsert(con, INSIGHTS_TABLE, fields, rows)
        total = con.execute(f"SELECT count(*) FROM {INSIGHTS_TABLE}").fetchone()[0]
        report["tables"][INSIGHTS_TABLE] = {"pulled": n, "in_mirror": total, "capped": capped}
        log(f"  {INSIGHTS_TABLE:<20} pulled {n:>6}   mirror total {total:>6}"
            + ("   !! PAGE CAP HIT" if capped else ""))

    con.execute("INSERT OR REPLACE INTO _sync_state VALUES (?,?,?,?,?,?)",
                ["meta", "done", 0, f"{days}d/{INSIGHTS_SLICE}d@{INSIGHTS_LEVEL}",
                 sum(t["in_mirror"] for t in report["tables"].values()),
                 time.strftime("%Y-%m-%d %H:%M:%S")])
    return report


def status(tenant_key="royal-imports"):
    """What is in the mirror right now, per table. Never fetches."""
    path = datastore.path_for(tenant_key)
    if Path(datastore.DB_PATH) != Path(path):
        datastore.use_path(path)
    out = {}
    con = datastore.connect()
    for table in list(SPECS) + [INSIGHTS_TABLE]:
        try:
            out[table] = con.execute(f"SELECT count(*) FROM {table}").fetchone()[0]
        except Exception:
            out[table] = None                                 # table absent = never synced
    return out


def main(argv=None):
    ap = argparse.ArgumentParser(description="Pull Meta Ads into the tenant's DuckDB mirror.")
    ap.add_argument("--sync", action="store_true")
    ap.add_argument("--status", action="store_true")
    ap.add_argument("--tenant", default="royal-imports")
    ap.add_argument("--no-insights", action="store_true")
    ap.add_argument("--insights-days", type=int, default=None,
                    help="override the Insights window for THIS run (default INSIGHTS_DAYS); the env var is the default, this is the caller's say")
    a = ap.parse_args(argv)
    if a.status:
        for k, v in status(a.tenant).items():
            print(f"  {k:<20} {'(never synced)' if v is None else v}")
        return 0
    if not a.sync:
        ap.print_help()
        return 2
    rep = sync(a.tenant, insights=not a.no_insights, insights_days=a.insights_days)
    for p in rep["problems"]:
        print("  PROBLEM:", p)
    print(f"  accounts: {len(rep['accounts'])}   tables: {len(rep['tables'])}")
    return 1 if rep["problems"] else 0


if __name__ == "__main__":
    sys.exit(main())