"""Insert-only operational history import for the pinned acceptance tenant.

Authoritative balances and actual transactions are retained independently.
Discrepancies remain explicit; no invented credit or shared-user broadcasts.
"""
from __future__ import annotations

import datetime
import hashlib
import json
import re
import sqlite3
import uuid
from collections import Counter, defaultdict

from weknora_asset_etl import LEDGER as ASSET_LEDGER, _mapping
from weknora_business_etl import archive_text, _canonical
from weknora_identity_etl import ADMIN_TARGET_ID, IDENTITY_LEDGER, TARGET_DATABASE, TARGET_TENANT

SOURCE = "yawan-workbench"
LEDGER = "yawan_legacy_operations_mappings"
NAMESPACE = uuid.UUID("3fce9870-2fb6-51c6-b727-37ab10674612")
KINDS = ("pointaccount", "pointtransaction", "usageevent", "knowledgeusage", "notification",
         "pendingmemory", "auditlog", "systemlog")
TABLES = ("workbench_balances", "workbench_transactions", "workbench_usage", "workbench_consumption",
          "workbench_events", "workbench_pending_memories", "workbench_operation_logs")
NATIVE_TABLE = dict(zip(KINDS, (*TABLES[:6], TABLES[-1], TABLES[-1])))
POLICY = "insert_only_authoritative_balances_owned_history_no_inferred_credit"
SENSITIVE_KEY = re.compile(r"(?i)(api.?key|authorization|password|passwd|secret|access.?token|refresh.?token|cookie)")
SECRET_PATTERNS = (
    re.compile(r"(?i)(bearer\s+)[A-Za-z0-9._~+/-]{8,}"),
    re.compile(r"\bsk-[A-Za-z0-9_-]{12,}"),
    re.compile(r'''(?i)((?:api[_-]?key|authorization|password|passwd|secret|access[_-]?token|refresh[_-]?token)\s*[=:]\s*["']?)[^\s,;"'&}\]]+'''),
    re.compile(r"(?i)(https?://)[^/\s:@]+:[^/\s@]+@"),
)


def operation_id(kind, source_id):
    if kind not in (*KINDS, "history") or type(source_id) is not int or source_id <= 0:
        raise ValueError("Invalid operational source identity")
    return str(uuid.uuid5(NAMESPACE, f"{SOURCE}:tenant:{TARGET_TENANT}:organization:1:{kind}:{source_id}"))


def history_id(kind, source_id):
    return str(uuid.uuid5(NAMESPACE, f"history:{kind}:{source_id}"))


def redact(value):
    if isinstance(value, dict):
        return {k: "[REDACTED]" if SENSITIVE_KEY.search(k) else redact(v) for k, v in value.items()}
    if isinstance(value, list):
        return [redact(v) for v in value]
    if not isinstance(value, str):
        return value
    value = value.replace("\x00", "[NUL]")
    value = "".join(c if not 0xd800 <= ord(c) <= 0xdfff else "[INVALID-UNICODE]" for c in value)
    for pattern in SECRET_PATTERNS:
        value = pattern.sub(lambda m: (m.group(1) if m.lastindex else "") + "[REDACTED]", value)
    return value


def _read(db, table):
    if table not in (*KINDS, "campaign", "segment"):
        raise ValueError("Unreviewed operations source table")
    result = db.execute(f'SELECT * FROM "{table}" ORDER BY id')
    fields = [c[0] for c in result.description]
    rows = [dict(zip(fields, row)) for row in result]
    for row in rows:
        if type(row["id"]) is not int or row["id"] <= 0:
            raise ValueError("Invalid operations source ID")
        if row.get("organization_id", 1) != 1:
            raise ValueError("Operations source organization differs")
        if table in ("pointtransaction", "usageevent"):
            row["metadata"] = json.loads(row["metadata"] or "{}")
    return rows


def _time(value):
    if not value:
        return "1970-01-01T00:00:00+00:00"
    date = datetime.datetime.fromisoformat(value.replace("Z", "+00:00"))
    if date.tzinfo is None:
        date = date.replace(tzinfo=datetime.timezone.utc)
    return date.isoformat()


def reconcile(accounts, transactions):
    totals = defaultdict(int)
    for tx in transactions:
        totals[tx["user_id"]] += tx["delta_points"]
    by_user = {}
    for account in accounts:
        if account["user_id"] in by_user:
            raise ValueError("Duplicate legacy point account")
        by_user[account["user_id"]] = account
    result = []
    for user in sorted(set(totals) | set(by_user)):
        account = by_user.get(user)
        balance = account["balance_points"] if account else None
        matched = account is not None and balance == totals[user]
        result.append({"source_user_id": user, "source_balance": balance, "transaction_sum": totals[user],
                       "difference": balance - totals[user] if account else None,
                       "reconciled": matched, "native_import": account is not None,
                       "reason": "" if matched else "balance_mismatch" if account else "missing_source_account"})
    return result


def _relation_owner(row, campaigns, segments, identities):
    campaign = row.get("campaign_id")
    segment = row.get("segment_id")
    if segment:
        sc = segments.get(segment)
        if not sc or (campaign and sc != campaign):
            return "", "segment_campaign_missing_or_mismatch"
        campaign = sc
    owner = campaigns.get(campaign)
    if not owner or str(owner) not in identities:
        return "", "unresolved_campaign_owner"
    return identities[str(owner)], ""


def notification_owner(row, campaigns, segments, identities):
    ref, kind = row["ref_id"], row["type"]
    if kind == "pending_review":
        return _relation_owner({"segment_id": ref}, campaigns, segments, identities)
    if kind == "info":
        return _relation_owner({"campaign_id": ref}, campaigns, segments, identities)
    if kind == "failure":
        candidates = {campaigns[c] for c in (ref, segments.get(ref)) if c in campaigns}
        if len(candidates) == 1 and str(next(iter(candidates))) in identities:
            return identities[str(next(iter(candidates)))], ""
        return "", "ambiguous_or_missing_notification_reference"
    return "", "notification_has_no_reliable_recipient"


def operations_plan(db, identities, agents, kbs, business, models):
    if db.execute("PRAGMA query_only").fetchone()[0] != 1:
        raise ValueError("Operations import requires a read-only consistent snapshot")
    identities, agents, kbs = (_mapping(m) for m in (identities, agents, kbs))
    if identities.get("1") != ADMIN_TARGET_ID:
        raise ValueError("Pinned administrator identity differs")
    sources = {table: _read(db, table) for table in KINDS}
    campaigns = {r["id"]: r["owner_user_id"] for r in _read(db, "campaign")}
    segments = {r["id"]: r["campaign_id"] for r in _read(db, "segment")}
    balances = reconcile(sources["pointaccount"], sources["pointtransaction"])
    reconciliation = {r["source_user_id"]: r for r in balances}
    charges = defaultdict(list)
    for tx in sources["pointtransaction"]:
        event = tx["metadata"].get("usage_event_id")
        if event:
            charges[event].append(tx)
    plan = {"schema_version": 1, "source": SOURCE, "target_database": TARGET_DATABASE,
            "target_tenant_id": TARGET_TENANT, "policy": POLICY, "identity_mapping": identities,
            "agent_mapping": agents, "kb_mapping": kbs, "business_mapping": business,
            "model_mapping": models, "reconciliation": balances, "items": []}
    for kind, rows in sources.items():
        for source in rows:
            sid, review = source["id"], []
            owner = identities.get(str(source.get("user_id", source.get("owner_user_id", source.get("actor_user_id")))), "")
            timestamp = _time(source.get("created_at", source.get("used_at", source.get("updated_at"))))
            target = {"id": operation_id(kind, sid), "tenant_id": TARGET_TENANT, "created_at": timestamp}
            table = ""
            metadata = {"legacy_import": {"source_kind": kind, "source_id": sid, "read_only": True},
                        "legacy_metadata": redact(source.get("metadata", {}))}
            if kind in ("pointaccount", "pointtransaction"):
                reconciliation_row = reconciliation[source["user_id"]]
                if not reconciliation_row["reconciled"]:
                    review.append("balance_reconciliation_requires_admin_review")
                metadata["balance_reconciliation"] = reconciliation_row
                if not owner:
                    review.append("missing_identity")
                elif kind == "pointaccount":
                    table = "workbench_balances"
                    target = {"tenant_id": TARGET_TENANT, "user_id": owner, "balance": source["balance_points"],
                              "reserved": 0, "updated_at": _time(source["updated_at"])}
                else:
                    table = "workbench_transactions"
                    target.update(user_id=owner, delta=source["delta_points"], kind=source["kind"],
                                  reason=redact(source["reason"]), actor_id=identities.get(str(source["operator_user_id"]), ""),
                                  idempotency_key=f"legacy:pointtransaction:{sid}", metadata=metadata)
            elif kind == "usageevent":
                if not owner:
                    review.append("missing_usage_owner")
                else:
                    table = "workbench_usage"
                    linked = charges[sid]
                    if any(tx["user_id"] != source["user_id"] or tx["delta_points"] >= 0 for tx in linked):
                        raise ValueError("Usage charge has conflicting owner or sign")
                    points = -sum(tx["delta_points"] for tx in linked)
                    metadata.update(legacy_unit=source["unit"], legacy_quantity=source["quantity"],
                                    legacy_currency=source["currency"], legacy_amount_minor=source["amount_minor"],
                                    legacy_transaction_ids=[tx["id"] for tx in linked],
                                    points_basis="exact_linked_debits" if linked else "no_linked_debit", legacy_call_id=source["llm_call_id"],
                                    legacy_campaign_id=source["campaign_id"], legacy_run_id=source["run_id"], legacy_model=redact(source["model"]))
                    token = source["unit"] == "token"
                    target.update(owner_id=owner, job_id="", call_id=f"legacy:usageevent:{sid}",
                                  agent_id=agents.get(str(source["agent_id"]), ""),
                                  model_id=models.get(f"{source['provider_id']}:{source['model']}", ""),
                                  action=redact(source["metadata"].get("operation") or source["service_type"]),
                                  input_tokens=source["input_quantity"] if token else 0,
                                  output_tokens=source["output_quantity"] if token else 0,
                                  points=points, status=source["status"], metadata=metadata)
            elif kind == "knowledgeusage":
                owner, reason = _relation_owner(source, campaigns, segments, identities)
                if reason:
                    review.append(reason)
                if not agents.get(str(source["agent_id"])) or not kbs.get(str(source["kb_id"])):
                    review.append("missing_consumption_asset_mapping")
                if not review:
                    table = "workbench_consumption"
                    # Reindexing changes chunks. Namespacing preserves historical usage
                    # without falsely marking an unrelated native chunk consumed.
                    chunk = "legacy:" + str(source["chunk_id"])
                    if len(chunk) > 200:
                        chunk = "legacy:sha256:" + hashlib.sha256(chunk.encode()).hexdigest()
                    target.update(owner_id=owner, agent_id=agents[str(source["agent_id"])], kb_id=kbs[str(source["kb_id"])],
                                  chunk_id=chunk, knowledge_id="", file_name=redact(source["file_name"]),
                                  category=redact(source["heading"]), excerpt=redact(source["excerpt"]),
                                  resource_id=business.get(f"campaign:{source['campaign_id']}", ""),
                                  segment_id=business.get(f"segment:{source['segment_id']}", ""), job_id="",
                                  status=source["status"], banned=bool(source["banned"]), lease_until=None,
                                  updated_at=_time(source["released_at"] or source["used_at"]))
            elif kind == "notification":
                owner, reason = notification_owner(source, campaigns, segments, identities)
                if reason:
                    review.append(reason)
                else:
                    table = "workbench_events"
                    target.update(owner_id=owner, resource_id="", action=source["type"], message=redact(source["message"]),
                                  level="error" if source["type"] == "failure" else "info", read=source["status"] == "read")
            elif kind == "pendingmemory":
                if not owner or not agents.get(str(source["target_agent_id"])):
                    review.append("missing_memory_owner_or_agent")
                else:
                    table = "workbench_pending_memories"
                    target.update(owner_id=owner, agent_id=agents[str(source["target_agent_id"])], content=redact(source["content"]),
                                  source_ref=redact(source["source_ref"]), status={"approved": "accepted"}.get(source["status"], source["status"]),
                                  reviewed_by="", updated_at=timestamp)
            else:
                table = "workbench_operation_logs"
                system = kind == "systemlog"
                level = str(source["level"]).lower() if system else ("info" if source["result"] in ("ok", "success") else "error")
                status = source.get("status", 0) if system else (200 if level == "info" else 400)
                status = int(status) if str(status).isdigit() else 0
                target.update(owner_id=owner, owner_name=redact(source.get("actor_display_name", "")),
                              kind="system" if system else "audit", level=level,
                              action=redact(source["source"] if system else source["action"]),
                              message=redact(source["message"] if system else source["result"]),
                              resource_id=redact(source.get("resource", "")), trace_id=redact(source.get("trace_id", ""))[:100],
                              status=status, metadata=metadata)
            plan["items"].append({"source_kind": kind, "source_id": sid, "source_snapshot": archive_text(source),
                                  "history_id": history_id(kind, sid), "owner_id": owner, "review": review,
                                  "target_table": table, "target_key": target.get("user_id") if table == "workbench_balances" else target["id"],
                                  "native_snapshot": target if table else {}, "created_at": timestamp,
                                  "history_payload": {"source": archive_text(redact(source)), "review": review,
                                                      "read_only": True, "original_in_private_export": True}})
            if kind in ("pointaccount", "pointtransaction"):
                plan["items"][-1]["history_payload"]["balance_reconciliation"] = reconciliation[source["user_id"]]
    return plan


def validate_plan(plan):
    if (plan.get("schema_version") != 1 or plan.get("source") != SOURCE or plan.get("target_database") != TARGET_DATABASE
            or plan.get("target_tenant_id") != TARGET_TENANT or plan.get("policy") != POLICY):
        raise ValueError("Operations migration scope differs")
    if _mapping(plan["identity_mapping"]).get("1") != ADMIN_TARGET_ID:
        raise ValueError("Operations administrator mapping differs")
    seen = set()
    for item in plan["items"]:
        key = item["source_kind"], item["source_id"]
        if key in seen or key[0] not in KINDS or item["source_snapshot"]["id"] != key[1] or item["history_id"] != history_id(*key):
            raise ValueError("Operations source identity conflict")
        seen.add(key)
        native, table = item["native_snapshot"], item["target_table"]
        if table and table != NATIVE_TABLE[key[0]]:
            raise ValueError("Operations native table differs")
        if table and native.get("tenant_id") != TARGET_TENANT:
            raise ValueError("Operations native tenant differs")
        if table and table != "workbench_balances" and native["id"] != operation_id(*key):
            raise ValueError("Operations native identity differs")
        if table and table != "workbench_operation_logs" and not item["owner_id"]:
            raise ValueError("Owned operations cannot be broadcast")
        if table:
            owner_column = "user_id" if table in ("workbench_balances", "workbench_transactions") else "owner_id"
            native_key = native["user_id"] if table == "workbench_balances" else native["id"]
            if native[owner_column] != item["owner_id"] or item["target_key"] != native_key:
                raise ValueError("Operations native owner or key differs")
        if item["owner_id"] and item["owner_id"] not in plan["identity_mapping"].values():
            raise ValueError("Operations owner is not mapped")
    accounts = [i["source_snapshot"] for i in plan["items"] if i["source_kind"] == "pointaccount"]
    transactions = [i["source_snapshot"] for i in plan["items"] if i["source_kind"] == "pointtransaction"]
    if plan["reconciliation"] != reconcile(accounts, transactions):
        raise ValueError("Balance reconciliation was changed")
    for item in plan["items"]:
        source, native = item["source_snapshot"], item["native_snapshot"]
        if item["target_table"] == "workbench_balances" and native["balance"] != source["balance_points"]:
            raise ValueError("Authoritative current balance must remain exact")
        if item["target_table"] == "workbench_transactions" and native["delta"] != source["delta_points"]:
            raise ValueError("Historical transactions must remain exact")


def safe_operations_summary(plan):
    validate_plan(plan)
    return {"database": TARGET_DATABASE, "tenant_id": TARGET_TENANT,
            "source_counts": dict(Counter(i["source_kind"] for i in plan["items"])),
            "native_counts": dict(Counter(i["target_table"] for i in plan["items"] if i["target_table"])),
            "archive_only_counts": dict(Counter(i["source_kind"] for i in plan["items"] if not i["target_table"])),
            "review_counts": dict(Counter(r for i in plan["items"] for r in i["review"])),
            "reconciliation": plan["reconciliation"], "source_snapshot_sha256": plan.get("source_snapshot_sha256"),
            "history_owner_policy": "admin_only_archive", "full_original_export_required": True,
            "consumption_chunk_policy": "legacy_namespace_no_inferred_native_chunk", "jobs_created": 0}


def operations_sql(plan):
    validate_plan(plan)
    fingerprint = plan.get("source_snapshot_sha256", "")
    if not re.fullmatch("[a-f0-9]{64}", fingerprint):
        raise ValueError("Operations source snapshot fingerprint required")
    literal = "'" + _canonical(plan).replace("'", "''") + "'::jsonb"
    sql = f"""SET LOCAL standard_conforming_strings=on;
SET LOCAL log_statement='none'; SET LOCAL log_min_error_statement='panic';
CREATE TEMP TABLE _ops_plan(payload jsonb) ON COMMIT DROP;
INSERT INTO _ops_plan VALUES ({literal});
DO $scope$ BEGIN
 IF current_database()<>'{TARGET_DATABASE}' THEN RAISE EXCEPTION 'Operations database mismatch'; END IF;
 PERFORM pg_advisory_xact_lock({TARGET_TENANT},11907);
 PERFORM 1 FROM tenants WHERE id={TARGET_TENANT} AND status='active' AND deleted_at IS NULL FOR SHARE;
 IF NOT FOUND THEN RAISE EXCEPTION 'Operations target tenant inactive'; END IF;
 IF EXISTS (SELECT 1 FROM _ops_plan p,jsonb_each_text(p.payload->'identity_mapping') i
  LEFT JOIN {IDENTITY_LEDGER} m ON m.source_system='{SOURCE}' AND m.source_organization_id=1
   AND m.source_user_id=i.key::bigint AND m.target_tenant_id={TARGET_TENANT} AND m.target_user_id=i.value
  LEFT JOIN users u ON u.id=i.value AND u.deleted_at IS NULL
  LEFT JOIN tenant_members tm ON tm.user_id=i.value AND tm.tenant_id={TARGET_TENANT} AND tm.deleted_at IS NULL
  WHERE m.target_user_id IS NULL OR u.id IS NULL OR tm.id IS NULL)
 THEN RAISE EXCEPTION 'Operations durable owner mapping conflict'; END IF;
 IF EXISTS (SELECT 1 FROM _ops_plan p,jsonb_each_text(p.payload->'agent_mapping') i
  LEFT JOIN {ASSET_LEDGER} m ON m.source_system='{SOURCE}' AND m.source_organization_id=1
   AND m.source_kind='agent' AND m.source_id=i.key::bigint AND m.target_tenant_id={TARGET_TENANT} AND m.target_id=i.value
  LEFT JOIN custom_agents a ON a.id=i.value AND a.tenant_id={TARGET_TENANT}
  WHERE m.target_id IS NULL OR a.id IS NULL)
 THEN RAISE EXCEPTION 'Operations durable Agent mapping conflict'; END IF;
 IF EXISTS (SELECT 1 FROM _ops_plan p,jsonb_each_text(p.payload->'kb_mapping') i
  LEFT JOIN knowledge_bases k ON k.id=i.value AND k.tenant_id={TARGET_TENANT} AND k.deleted_at IS NULL
  WHERE k.id IS NULL)
 THEN RAISE EXCEPTION 'Operations KB scope conflict'; END IF;
 IF EXISTS (SELECT 1 FROM _ops_plan p,jsonb_each_text(p.payload->'business_mapping') i
  LEFT JOIN yawan_legacy_business_mappings b ON b.source_system='{SOURCE}' AND b.source_organization_id=1
   AND b.source_kind||':'||b.source_id=i.key AND b.target_tenant_id={TARGET_TENANT} AND b.target_id=i.value
  WHERE b.target_id IS NULL)
 THEN RAISE EXCEPTION 'Operations durable business mapping conflict'; END IF;
 IF EXISTS (SELECT 1 FROM _ops_plan p,jsonb_each_text(p.payload->'model_mapping') i
  LEFT JOIN yawan_legacy_model_mappings m ON m.source_system='{SOURCE}' AND m.source_organization_id=1
   AND m.target_tenant_id={TARGET_TENANT} AND m.target_model_id=i.value
  WHERE m.target_model_id IS NULL)
 THEN RAISE EXCEPTION 'Operations durable model mapping conflict'; END IF;
END $scope$;
LOCK TABLE {','.join(TABLES)},workbench_history IN SHARE ROW EXCLUSIVE MODE;
CREATE TABLE IF NOT EXISTS {LEDGER} (
 source_system text NOT NULL, source_organization_id bigint NOT NULL, source_kind text NOT NULL, source_id bigint NOT NULL,
 target_tenant_id bigint NOT NULL, history_id varchar(36) NOT NULL, target_table text NOT NULL, target_key text NOT NULL,
 source_snapshot jsonb NOT NULL, native_snapshot jsonb NOT NULL, review jsonb NOT NULL,
 source_snapshot_sha256 varchar(64) NOT NULL CHECK(source_snapshot_sha256 ~ '^[a-f0-9]{{64}}$'),
 created_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY(source_system,source_organization_id,source_kind,source_id),
 UNIQUE(target_tenant_id,history_id)
);
LOCK TABLE {LEDGER} IN SHARE ROW EXCLUSIVE MODE;
CREATE TEMP TABLE _ops_items ON COMMIT DROP AS SELECT item.* FROM _ops_plan p,
 jsonb_to_recordset(p.payload->'items') AS item(source_kind text,source_id bigint,source_snapshot jsonb,
 history_id text,owner_id text,review jsonb,target_table text,target_key text,native_snapshot jsonb,created_at timestamptz,history_payload jsonb);
CREATE INDEX ON _ops_items(source_kind,source_id);
DO $provenance$ BEGIN
 IF EXISTS (SELECT 1 FROM _ops_items i JOIN {LEDGER} l USING(source_kind,source_id)
  WHERE l.source_system='{SOURCE}' AND l.source_organization_id=1 AND
   (l.target_tenant_id IS DISTINCT FROM {TARGET_TENANT} OR l.history_id IS DISTINCT FROM i.history_id
    OR l.target_table IS DISTINCT FROM i.target_table OR l.target_key IS DISTINCT FROM i.target_key
    OR l.source_snapshot IS DISTINCT FROM i.source_snapshot OR l.native_snapshot IS DISTINCT FROM i.native_snapshot
    OR l.review IS DISTINCT FROM i.review))
 THEN RAISE EXCEPTION 'Operations provenance conflict'; END IF;
 IF EXISTS (SELECT 1 FROM _ops_items i JOIN workbench_history h ON h.id=i.history_id
  WHERE NOT EXISTS (SELECT 1 FROM {LEDGER} l WHERE l.source_system='{SOURCE}' AND l.source_organization_id=1
   AND l.source_kind=i.source_kind AND l.source_id=i.source_id))
 THEN RAISE EXCEPTION 'Operations history collision'; END IF;
 IF EXISTS (SELECT 1 FROM _ops_items i JOIN {LEDGER} l USING(source_kind,source_id)
  LEFT JOIN workbench_history h ON h.id=i.history_id AND h.tenant_id={TARGET_TENANT} AND h.owner_id=''
  WHERE l.source_system='{SOURCE}' AND l.source_organization_id=1 AND h.id IS NULL)
 THEN RAISE EXCEPTION 'Operations history removed or scope changed'; END IF;
END $provenance$;
CREATE TEMP TABLE _ops_new ON COMMIT DROP AS SELECT i.* FROM _ops_items i WHERE NOT EXISTS
 (SELECT 1 FROM {LEDGER} l WHERE l.source_system='{SOURCE}' AND l.source_organization_id=1 AND l.source_kind=i.source_kind AND l.source_id=i.source_id);
"""
    for table in TABLES:
        key = "user_id" if table == "workbench_balances" else "id"
        owner = "user_id" if table in ("workbench_balances", "workbench_transactions") else "owner_id"
        sql += f"""DO $native$ BEGIN
 IF EXISTS (SELECT 1 FROM _ops_new i JOIN {table} n ON n.{key}=i.target_key
  WHERE i.target_table='{table}' {'AND n.tenant_id=' + str(TARGET_TENANT) if key == 'user_id' else ''})
 THEN RAISE EXCEPTION 'Operations native collision in {table}'; END IF;
 IF EXISTS (SELECT 1 FROM _ops_items i JOIN {LEDGER} l USING(source_kind,source_id)
  LEFT JOIN {table} n ON n.{key}=i.target_key AND n.tenant_id={TARGET_TENANT} AND n.{owner}=i.owner_id
  WHERE l.source_system='{SOURCE}' AND l.source_organization_id=1 AND i.target_table='{table}' AND n.{key} IS NULL)
 THEN RAISE EXCEPTION 'Operations native row removed or ownership changed in {table}'; END IF;
END $native$;
INSERT INTO {table} SELECT (jsonb_populate_record(NULL::{table},i.native_snapshot)).*
 FROM _ops_new i WHERE i.target_table='{table}';
"""
    sql += f"""INSERT INTO workbench_history(id,tenant_id,owner_id,source_kind,source_id,payload,created_at)
 SELECT history_id,{TARGET_TENANT},'',source_kind,source_id::text,history_payload,created_at FROM _ops_new;
INSERT INTO {LEDGER}(source_system,source_organization_id,source_kind,source_id,target_tenant_id,history_id,target_table,target_key,
 source_snapshot,native_snapshot,review,source_snapshot_sha256)
 SELECT '{SOURCE}',1,source_kind,source_id,{TARGET_TENANT},history_id,target_table,target_key,source_snapshot,native_snapshot,review,'{fingerprint}' FROM _ops_new;
SELECT json_build_object('inserted_history',(SELECT count(*) FROM _ops_new),
 'inserted_native',(SELECT count(*) FROM _ops_new WHERE target_table<>''),
 'ledger_total',(SELECT count(*) FROM {LEDGER} WHERE target_tenant_id={TARGET_TENANT}));
"""
    return sql