api/meta_relational.py

"""meta_relational.py — the Meta Ads mirror becomes LOCKED DATABASES (W31-T47, ruling R2).

The second half of the owner's instruction: *"we just need to pull the data into our template
database for Meta"*, *"much like how we have done so for Odoo"*. `harness/meta_store.py` pulls
Graph into the tenant's DuckDB mirror; this reads that mirror and spawns the `ut_meta_*` grids a
person opens. It is `odoo_relational.py`'s shape with one source swapped.

    plan(cur)                 the rows that WOULD be written — no store write at all
    apply_plan(rt, built)     create-or-merge every table and its rows, in ONE sync write
    refresh(rt, why)          both, with the mirror opened for this tenant

⭐⭐ THE COLUMNS ARE DERIVED FROM THE MIRROR, NOT TYPED OUT, AND THAT IS THE R2 REQUIREMENT MADE
STRUCTURAL. The owner's words were *"understand FULLY the schema, do not drop any column etc."* A
hand-written field list is a second place for the schema to live and the first place for it to
rot — so `_fields_for` reads `PRAGMA table_info` off the table `meta_store` just wrote and emits
one preset field per column. Add a column to the loader's measured list and it appears in the grid
with no edit here. **285 columns across five levels plus the daily Insights grain.**

⛔ WHY IT REUSES `odoo_relational._ensure_table_inplace` RATHER THAN COPYING IT. That function is
where a `ut_*` table becomes a LOCKED DATABASE — `recordMode = AUTOMATION_RECORD_MODE`, the preset
stamp, the MAX_TABLES refusal, the row reconcile. Two copies of that would be two answers to "is
this database locked", and the copies would diverge on the first ruling that touches one of them.
One implementation, two callers, exactly as `_LentTables` became `user_tables.lend` this wave.

⚠ MATERIALISED, NOT READ-THROUGH, AND THE REASON IS THE NUMBERS. Odoo's `ut_odoo_gl_lines` reads
THROUGH the mirror because it is 963,783 rows against a `MAX_ROWS` of 60,000. Meta at this tenant's
scale is 1 account · 55 campaigns · 115 ad sets · 227 ads · creatives · ~90 days of daily insights
— every table two to four orders of magnitude inside the cap, so they are ordinary materialised
`ut_*` tables like `ut_odoo_agents`. ⛔ If a bigger account ever crosses the cap, `plan()` says so
out loud rather than truncating: R6's second sentence, and the check is at the bottom of `plan`.
"""
import os
import sys
from pathlib import Path

_API = Path(__file__).resolve().parent
if str(_API.parent.parent / "platform") not in sys.path:
    sys.path.insert(0, str(_API.parent.parent / "platform"))

import odoo_relational as _odoo                                  # noqa: E402
from harness import datastore, meta_store                        # noqa: E402

Refused = _odoo.Refused

ACCOUNTS_KEY = "ut_meta_ad_accounts"
CAMPAIGNS_KEY = "ut_meta_campaigns"
ADSETS_KEY = "ut_meta_adsets"
ADS_KEY = "ut_meta_ads"
CREATIVES_KEY = "ut_meta_creatives"
INSIGHTS_KEY = "ut_meta_insights"

#: ut_* key -> the mirror table `meta_store` writes.
SOURCE = {
    ACCOUNTS_KEY: "meta_ad_accounts",
    CAMPAIGNS_KEY: "meta_campaigns",
    ADSETS_KEY: "meta_adsets",
    ADS_KEY: "meta_ads",
    CREATIVES_KEY: "meta_creatives",
    INSIGHTS_KEY: meta_store.INSIGHTS_TABLE,
}

LABELS = {
    ACCOUNTS_KEY: "Meta ad accounts", CAMPAIGNS_KEY: "Meta campaigns",
    ADSETS_KEY: "Meta ad sets", ADS_KEY: "Meta ads",
    CREATIVES_KEY: "Meta creatives", INSIGHTS_KEY: "Meta insights (daily)",
}

#: The column each grid shows first. Everything else keeps the API's own order, which is the order
#: the schema was measured in — a reader comparing the grid to Graph sees the same sequence.
PINNED = {ACCOUNTS_KEY: "name", CAMPAIGNS_KEY: "name", ADSETS_KEY: "name",
          ADS_KEY: "name", CREATIVES_KEY: "name", INSIGHTS_KEY: "date_start"}

#: (this table's column) -> (target ut_ key, the column it matches there). Derived links, so a
#: person can walk account -> campaign -> ad set -> ad in the product the way they do in Ads
#: Manager. ⚠ Only emitted when BOTH sides exist in the mirror.
LINKS = {
    CAMPAIGNS_KEY: ("account_id", ACCOUNTS_KEY, "account_id"),
    ADSETS_KEY: ("campaign_id", CAMPAIGNS_KEY, "id"),
    ADS_KEY: ("adset_id", ADSETS_KEY, "id"),
    INSIGHTS_KEY: ("ad_id", ADS_KEY, "id"),
}

#: DuckDB type -> `core.user_tables.UT_FIELD_TYPES` member. ⚠ `_clean_field` returns None for an
#: unknown type, which DELETES the column on the next read rather than erroring — so this map only
#: ever emits members of that set, and anything unrecognised falls back to `text`.
_TYPE = {"BIGINT": "int", "INTEGER": "int", "HUGEINT": "int",
         "DOUBLE": "currency", "FLOAT": "currency", "DECIMAL": "currency"}

#: Columns whose NAME makes them a date regardless of storage. Graph returns ISO-ish strings.
_DATEISH = ("_time", "date_start", "date_stop", "created_time", "updated_time",
            "start_time", "stop_time", "end_time")

#: ⛔ A MONEY COLUMN IS `currency`, NOT `int`, and the distinction is not cosmetic: `spend` is
#: fractional in every currency Meta reports and an int column would round every row. The account
#: reports HKD; the grid does not convert, it shows what the API said.
_MONEY = {"spend", "social_spend", "cpc", "cpm", "cpp", "amount_spent", "balance", "spend_cap",
          "daily_budget", "lifetime_budget", "budget_remaining", "bid_amount",
          "cost_per_inline_link_click", "cost_per_unique_click", "cost_per_thruplay"}


def _label(col):
    return col.replace("_", " ").strip().capitalize()


def _ftype(col, duck_type):
    if any(col.endswith(s) or col == s for s in _DATEISH):
        return "date"
    if col in _MONEY:
        return "currency"
    base = str(duck_type or "").upper().split("(")[0]
    if base in ("BIGINT", "INTEGER", "HUGEINT") or base.startswith("DECIMAL"):
        return _TYPE.get(base, "int")
    if base in ("DOUBLE", "FLOAT"):
        return "currency"
    return "text"


def mirror_columns(cur, table):
    """[(name, type)] for a mirror table, or [] when it has never been synced."""
    try:
        return [(r[1], r[2]) for r in cur.execute(f"PRAGMA table_info('{table}')").fetchall()]
    except Exception:
        return []


def _fields_for(cur, key):
    """One preset field per mirror column — the whole schema, derived (see the header)."""
    cols = mirror_columns(cur, SOURCE[key])
    if not cols:
        return []
    pinned = PINNED.get(key)
    link = LINKS.get(key)
    out = []
    for name, dtype in cols:
        f = {"key": name, "label": _label(name), "type": _ftype(name, dtype),
             "source": "overlay", "default": name in (pinned, "id", "name", "status", "spend",
                                                      "impressions", "clicks", "date_start"),
             "pinned": name == pinned}
        out.append(_odoo._preset(f, flow="meta_relational"))
    if link:
        col, target, on = link
        if any(c[0] == col for c in cols):
            out.append(_odoo._preset(
                {"key": f"{target}_link", "label": LABELS[target], "type": "link",
                 "source": "overlay", "default": True,
                 "link": {"table": target, "on": on, "from": col}}, flow="meta_relational"))
    out.append(_odoo._preset(_odoo._refreshed_field(), flow="meta_relational"))
    return out


def _read(cur, key):
    """[(row dict)] straight off the mirror. `_id` is the object's own Meta id — so a re-sync
    updates in place and 'every id Graph returned is in the table' is checkable, not hopeful."""
    cols = [c[0] for c in mirror_columns(cur, SOURCE[key])]
    if not cols:
        return []
    rows = cur.execute(f"SELECT {', '.join(cols)} FROM {SOURCE[key]}").fetchall()
    out = []
    for r in rows:
        row = {"_id": str(r[cols.index("id")]) if "id" in cols else str(len(out))}
        for name, val in zip(cols, r):
            row[name] = "" if val is None else (val if isinstance(val, (int, float)) else str(val))
        out.append(row)
    return out


def available(tenant_key="royal-imports"):
    """Has this tenant's mirror ever been fed by `meta_store`? Cheap; never fetches."""
    try:
        counts = meta_store.status(tenant_key)
    except Exception:
        return False
    return any(bool(v) for v in counts.values())


def plan(cur, rt=None):
    """{bucket_key: rows} plus `problems`. No store write. Every cap checked BEFORE anything is
    committed — the same order `odoo_relational.plan` keeps and for the same reason."""
    built = {"problems": []}
    ut = _odoo._ut()
    for key in SOURCE:
        cap = getattr(ut, "MAX_ROWS", 60_000)
        # ⛔ COUNT BEFORE READ, not after. `_read` materialises every row as a dict — the whole
        # table twice over, once from DuckDB and once as python — and checking the cap afterwards
        # means the one table too big to keep is the one we build in memory first. A `count(*)` is
        # free and answers the same question.
        try:
            n = cur.execute(f"SELECT count(*) FROM {SOURCE[key]}").fetchone()[0]
        except Exception:
            continue                                          # table absent = never synced
        if n > cap:
            built["problems"].append(
                f"{key}: the mirror holds {n:,} rows against MAX_ROWS={cap:,}. Nothing was "
                f"truncated and nothing was written for this table. Cause: this account is larger "
                f"than a materialised ut_* table can hold. Fix: bind it read-through off the "
                f"mirror as `ut_odoo_gl_lines` is (odoo_relational.READ_THROUGH_KEYS), which is "
                f"what `routes_connected_tables` already serves.")
            continue
        rows = _read(cur, key)
        if not rows:
            continue
        if len(rows) > cap:
            # ⛔ REPORTED, NEVER TRUNCATED (R6's second sentence). The fix is named so the reader
            # does not have to derive it: this table becomes read-through like the Odoo line
            # grains, which is a build, not a bigger constant.
            built["problems"].append(
                f"{key}: the mirror holds {len(rows):,} rows against MAX_ROWS={cap:,}. Nothing was "
                f"truncated and nothing was written for this table. Cause: this account is larger "
                f"than a materialised ut_* table can hold. Fix: bind it read-through off the "
                f"mirror as `ut_odoo_gl_lines` is (odoo_relational.READ_THROUGH_KEYS).")
            continue
        built[key] = rows
    return built


def apply_plan(rt, built, username="meta", today=None):
    """Create-or-merge every table in `built`, in ONE `flush="sync"` write.

    ⚠ ONE write for all six tables, not six — `Store.update` is a full download plus a full upload
    of the tenant document, so six would be six round trips and six commits every resync. It is
    also ATOMIC: a `Refused` at MAX_TABLES aborts before anything persists, instead of leaving the
    half-spawn `plan()` exists to prevent.
    """
    if built.get("problems"):
        raise Refused("; ".join(built["problems"]))
    stamp = today or _odoo._iso_today()
    written = {}
    cur_mirror = datastore.ro_con()
    plans = [(key, LABELS[key], _fields_for(cur_mirror, key), built[key])
             for key in SOURCE if key in built]

    def _apply_all(doc):
        doc = doc if isinstance(doc, dict) else {}
        for key, label, fields, rows in plans:
            written[key] = _odoo._ensure_table_inplace(doc, key, label, fields, rows,
                                                       username, stamp)
        return doc

    rt.update(_odoo._ut().STORE_KEY, _apply_all, flush="sync")
    return written


def refresh(rt, why="boot", tenant_key=None, log=print):
    """The one call a caller wants: open this tenant's mirror, plan, apply. -> written counts.

    Returns `{}` and says why when the mirror has no Meta tables — a tenant that never connected
    Meta is a normal state, not a failure.
    """
    tenant = tenant_key or getattr(getattr(rt, "tenant", None), "key", None) or "royal-imports"
    path = datastore.path_for(tenant)
    if Path(datastore.DB_PATH) != Path(path):
        datastore.use_path(path)
    cur = datastore.ro_con()
    if not any(mirror_columns(cur, t) for t in SOURCE.values()):
        log(f"[meta_relational] {why}: no Meta tables in {tenant}'s mirror — nothing to spawn "
            f"(run `python platform/harness/meta_store.py --sync --tenant {tenant}` first)")
        return {}
    built = plan(cur, rt)
    for p in built.get("problems") or []:
        log(f"[meta_relational] PROBLEM {p}")
    if built.get("problems"):
        built = {k: v for k, v in built.items() if k != "problems"}
    written = apply_plan(rt, built)
    for key, counts in written.items():
        log(f"[meta_relational] {why}: {key:<24} {counts}")
    return written


def main(argv=None):
    """`python aios-web/api/meta_relational.py --refresh [--tenant X]` — the manual door."""
    import argparse
    ap = argparse.ArgumentParser()
    ap.add_argument("--refresh", action="store_true")
    ap.add_argument("--plan", action="store_true")
    ap.add_argument("--tenant", default="royal-imports")
    a = ap.parse_args(argv)
    os.environ.setdefault("AIOS_PREWARM", "0")
    from harness import runtime as _rt
    rt = _rt.get_runtime(a.tenant)
    if a.plan:
        path = datastore.path_for(a.tenant)
        if Path(datastore.DB_PATH) != Path(path):
            datastore.use_path(path)
        built = plan(datastore.ro_con(), rt)
        for k, v in built.items():
            print(f"  {k:<26} {len(v) if isinstance(v, list) else v}")
        return 0
    if a.refresh:
        refresh(rt, why="cli", tenant_key=a.tenant)
        return 0
    ap.print_help()
    return 2


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