api/ig_master.py

"""ig_master.py — the PLATFORM-WIDE Instagram master store (wave 22, contract C6 / ruling R2).

WHY A MASTER AT ALL. History is unbuyable: no vendor sells "followers on 1 January", so the
series starts the day capture starts — and a profile two tenants both track was, until this
wave, two separate paid pulls writing two private copies. R2 pools the SNAPSHOT HISTORY on the
operator plane: every pull WRITE-THROUGHS here as well as to the tenant's own `ut_ig_*` tables,
and the time-window metric fields (C7) read THIS series — so a handle any tenant tracks has one
continuous history, and nobody re-bills for what the platform already knows.

RESIDENCY (C6 + amendment): its OWN private dataset repo — `royal-imports/aios-ig-master`, env
`AIOS_IG_MASTER_REPO` — reached through the `core/store.py::handle()` seam so R9's Postgres
flip carries it unchanged. ⛔ NEVER the tenant `user_tables` bucket: that bucket serialises
whole (MEASURED 35.8 MB at the append-table ceiling), so parking a platform-wide series inside
any tenant's bucket would tax every unrelated write in that tenant. Three buckets of its own:

    master_snapshots        append   key `<handle>@<pulled_at>`   profile counts over time
    master_posts            upsert   key `<shortcode>`            post identity (handle, when)
    master_post_snapshots   append   key `<shortcode>@<pulled_at>` engagement over time

TENANT ATTRIBUTION IS FOR AUDIT/ERASURE ONLY (C6, verbatim): every row carries the tenant slug
that paid for the pull, and NOTHING reads it to answer a data question — the pooling benefit
is exactly that reads are attribution-blind. D-24's subject purge walks these buckets too
(`purge_handle`), because a right-to-erasure that missed the pooled copy would not be erasure.

FAILURE POSTURE: three states, never conflated. `off` = no repo configured on this deployment
(local dev, the gates) — the tenant copy is the whole story and the run says so WITHOUT going
partial; `ok` = written; `error` = configured and the write failed — a LOUD partial in the run
summary, never a silent skip (C6, verbatim).
"""
from __future__ import annotations

import os

ENV_REPO = "AIOS_IG_MASTER_REPO"
SNAP_BUCKET = "master_snapshots"
POST_BUCKET = "master_posts"
PSNAP_BUCKET = "master_post_snapshots"
#: One bucket's row ceiling. The append tables grow forever by design; the cap exists so a
#: runaway writer is a LOUD partial long before a bucket becomes unserialisable. At ~1k
#: profiles/day this is ~16 months of profile snapshots — re-derive before raising.
MASTER_MAX_ROWS = 500_000


def repo_id():
    return (os.environ.get(ENV_REPO) or "").strip()


def configured():
    return bool(repo_id())


def _handle():
    """The store handle — THE seam (C6): `core.store.handle` picks the backend, so the pg flip
    (R9) moves this store the day it moves every other one. Monkeypatched by the gate."""
    import core.store as store
    return store.handle(repo_id=repo_id(), slug="ig_master")


def _upsert():
    # Lazy: the engine imports this module, so a top-level engine import would be a cycle.
    import automation_engine as engine
    return engine.upsert_rows


def append_run(tenant, snaps, posts, psnaps):
    """Write-through one run's accumulated rows. Returns `(status, note)`,
    status ∈ ok | off | error — see the module header for what each obliges the caller to say.

    One coalesced update PER BUCKET per run (the 256-commits/hr law travels — this repo has its
    own budget, and a per-row writer would spend it the same way). Empty inputs cost nothing.
    """
    if not configured():
        return "off", "no master store on this deployment — tenant copy only"
    slug = str(tenant or "").strip() or "unknown"
    stamped = {
        SNAP_BUCKET: ([{**r, "tenant": slug} for r in snaps or []], "snapshot_key"),
        POST_BUCKET: ([{**r, "tenant": slug} for r in posts or []], "shortcode"),
        PSNAP_BUCKET: ([{**r, "tenant": slug} for r in psnaps or []], "post_snapshot_key"),
    }
    upsert = _upsert()
    capped = []
    try:
        h = _handle()
        for bucket, (rows_in, key_field) in stamped.items():
            if not rows_in:
                continue
            existing = dict(h.get(bucket) or {})
            merged, counts = upsert(existing, rows_in, key_field, cap=MASTER_MAX_ROWS)
            if counts.get("capped"):
                capped.append(f"{bucket} is FULL ({MASTER_MAX_ROWS:,}) — "
                              f"{counts['capped']} rows lost")
            h.update(bucket, lambda cur, m=merged: m, flush="sync")
    except Exception as e:                                        # noqa: BLE001
        return "error", f"the master store did not take the write ({type(e).__name__})"
    if capped:
        return "error", "; ".join(capped)
    return "ok", ""


def series_for(handles):
    """The C7 read seam: `{handle: {snapshots: [...], posts: [...], postSnapshots:
    {shortcode: [...]}}}` for the handles asked about, each series sorted ascending by its
    time key. Attribution-blind by design (the pooling benefit). `{}` when unconfigured —
    a metric over no master reads BLANK, never zero."""
    if not configured():
        return {}
    want = {str(x or "").strip().lower() for x in handles or [] if str(x or "").strip()}
    if not want:
        return {}
    try:
        h = _handle()
        snaps = h.get(SNAP_BUCKET) or {}
        posts = h.get(POST_BUCKET) or {}
        psnaps = h.get(PSNAP_BUCKET) or {}
    except Exception:                                             # noqa: BLE001
        return {}
    out = {w: {"snapshots": [], "posts": [], "postSnapshots": {}} for w in want}
    code_owner = {}
    for r in posts.values():
        hkey = str((r or {}).get("influencer_key") or "").strip().lower()
        if hkey in out:
            out[hkey]["posts"].append(dict(r))
            code_owner[str(r.get("shortcode") or "")] = hkey
    for r in snaps.values():
        hkey = str((r or {}).get("influencer_key") or "").strip().lower()
        if hkey in out:
            out[hkey]["snapshots"].append(dict(r))
    for r in psnaps.values():
        code = str((r or {}).get("shortcode") or "")
        hkey = code_owner.get(code)
        if hkey:
            out[hkey]["postSnapshots"].setdefault(code, []).append(dict(r))
    for w in out.values():
        w["snapshots"].sort(key=lambda r: str(r.get("pulled_at") or ""))
        w["posts"].sort(key=lambda r: str(r.get("posted_at") or ""))
        for series in w["postSnapshots"].values():
            series.sort(key=lambda r: str(r.get("pulled_at") or ""))
    return out


def purge_handle(handle):
    """D-24's master half: remove EVERY row about one subject from all three buckets.
    Returns `{bucket: removed}`; `{}` when unconfigured (nothing exists to purge). The tenant
    half lives in the engine (`purge_subject`) — the two run together, and the gate proves a
    purged handle leaves no row on either side."""
    if not configured():
        return {}
    subject = str(handle or "").strip().lower()
    if not subject:
        return {}
    h = _handle()
    removed = {}
    posts = h.get(POST_BUCKET) or {}
    codes = {str(r.get("shortcode") or "") for r in posts.values()
             if str((r or {}).get("influencer_key") or "").strip().lower() == subject}

    def _sweep(bucket, keep):
        cur = h.get(bucket) or {}
        nxt = {k: v for k, v in cur.items() if keep(v)}
        removed[bucket] = len(cur) - len(nxt)
        if removed[bucket]:
            h.update(bucket, lambda c, m=nxt: m, flush="sync")

    _sweep(SNAP_BUCKET,
           lambda r: str((r or {}).get("influencer_key") or "").strip().lower() != subject)
    _sweep(POST_BUCKET,
           lambda r: str((r or {}).get("influencer_key") or "").strip().lower() != subject)
    _sweep(PSNAP_BUCKET, lambda r: str((r or {}).get("shortcode") or "") not in codes)
    return removed