C データエンジニアリング — 退会会員の個人データ消去を「分析基盤の下流まで」伝播させる（incremental mergeは上流の削除を下流に伝えない × dbt snapshotのhard_deletesは履歴行のPIIを残す × time travel 7日＋fail-safe 7日を無視した即日「完了」通知 × 有効期限なしtable snapshotにPIIが永久残存 × 取引行を行ごと削除し報告済み売上が38万円変動 × 例外握り潰しとストリーミングバッファ起因の部分消去 × 対象テーブルの手書きリストと検証なし・1件ずつDMLで1.2TBを100回スキャン）（MOps 販促システムチーム member-erasure-job Bad→Good）

📋 コピー"""退会会員の個人データ消去を、分析基盤の下流まで伝播させるジョブ。

Argo Workflows の CronWorkflow から1日1回呼ばれる。消去対象テーブルは
コードに書かず、dbt manifest.json の meta.erasure から列挙する。
"""

from __future__ import annotations

import datetime as dt
import json
import uuid
from dataclasses import dataclass
from enum import StrEnum
from pathlib import Path

from google.cloud import bigquery

LEDGER = "mops-project.governance.erasure_ledger"
AUDIT_LOG = "mops-project.governance.erasure_audit_log"

# ⑥ insertAll の行はバッファにある間 DML できない。受付から待つ時間
STREAMING_BUFFER_GRACE = dt.timedelta(hours=2)
# PII データセットの max_time_travel_hours=48 と一致させる
TIME_TRAVEL_DAYS = 2
# time travel の後に BigQuery が保持する期間（利用者は短縮できない）
FAIL_SAFE_DAYS = 7
# ④ PII テーブルの table snapshot に付ける有効期限
TABLE_SNAPSHOT_TTL_DAYS = 7
# ③ 「復元不能」と言えるまでの日数 = 一番遅く消える複製で決まる
PHYSICAL_ERASURE_DAYS = max(TIME_TRAVEL_DAYS + FAIL_SAFE_DAYS, TABLE_SNAPSHOT_TTL_DAYS)
# 集計（件数・金額）は残し、人には戻れない値
ERASED_MEMBER_ID = "ERASED"


class ErasureAction(StrEnum):
    """消去のやり方。"""

    DELETE = "delete"              # マーケ用途だけのテーブル: 行ごと消す
    PSEUDONYMIZE = "pseudonymize"  # ⑤ 取引: 識別子と PII 列だけ外す


@dataclass(frozen=True, slots=True)
class ErasureTarget:
    """消去対象のテーブル1つ。

    Attributes:
        table_id: `project.dataset.table` 形式の完全修飾名。
        action: 行ごと消すか、識別子だけ外すか。
        key_column: 会員を特定する列名。
        pii_columns: pseudonymize のとき NULL にする列。
    """

    table_id: str
    action: ErasureAction
    key_column: str
    pii_columns: tuple[str, ...] = ()


def load_targets(manifest_path: Path) -> list[ErasureTarget]:
    """dbt の manifest.json から meta.erasure を持つノードを列挙する。

    Args:
        manifest_path: `dbt parse` が出力した manifest.json のパス。

    Returns:
        消去対象テーブルのリスト（model と snapshot の両方）。
    """
    manifest = json.loads(manifest_path.read_text())
    targets: list[ErasureTarget] = []
    for node in manifest["nodes"].values():
        # ①② incremental も snapshot も同じ宣言で拾う
        if node["resource_type"] not in ("model", "snapshot"):
            continue
        spec = node["config"].get("meta", {}).get("erasure")
        if spec is None:
            continue
        targets.append(ErasureTarget(
            table_id=f"{node['database']}.{node['schema']}.{node['alias']}",
            action=ErasureAction(spec["action"]),
            key_column=spec["key"],
            pii_columns=tuple(spec.get("pii_columns", [])),
        ))
    return targets


def build_dml(target: ErasureTarget) -> str:
    """1テーブル分の DML。会員IDは SQL に埋め込まず台帳を副問い合わせで引く。"""
    # ⑦ 1バッチ1文。⑥ 値を文字列連結しない
    in_batch = f"(SELECT member_id FROM `{LEDGER}` WHERE batch_id = @batch_id)"
    if target.action is ErasureAction.DELETE:
        return f"DELETE FROM `{target.table_id}` WHERE {target.key_column} IN {in_batch}"
    # ⑤ 金額・日付は残す → 報告済みの売上が変わらない
    sets = [f"{target.key_column} = '{ERASED_MEMBER_ID}'"]
    sets += [f"{col} = NULL" for col in target.pii_columns]
    return (f"UPDATE `{target.table_id}` SET {', '.join(sets)} "
            f"WHERE {target.key_column} IN {in_batch}")


def claim_batch(client: bigquery.Client, now: dt.datetime) -> str | None:
    """受付から十分時間が経った請求に batch_id を振る。対象が無ければ None。"""
    batch_id = uuid.uuid4().hex
    job = client.query(
        f"""UPDATE `{LEDGER}` SET batch_id = @batch_id
            WHERE status = 'requested' AND batch_id IS NULL
              AND requested_at <= @cutoff""",
        job_config=bigquery.QueryJobConfig(query_parameters=[
            bigquery.ScalarQueryParameter("batch_id", "STRING", batch_id),
            bigquery.ScalarQueryParameter(
                "cutoff", "TIMESTAMP", now - STREAMING_BUFFER_GRACE),
        ]),
    )
    job.result()
    return batch_id if job.num_dml_affected_rows else None


def propagate(client: bigquery.Client, targets: list[ErasureTarget],
              now: dt.datetime) -> None:
    """1バッチ分の消去を全対象へ伝播させ、台帳を propagated に進める。

    同じ batch_id で再実行しても影響行数が 0 になるだけなので冪等。
    Argo の retryStrategy でそのまま再試行してよい。

    Raises:
        google.api_core.exceptions.GoogleAPICallError: DML が失敗したとき。
    """
    batch_id = claim_batch(client, now)
    if batch_id is None:
        return
    params = [bigquery.ScalarQueryParameter("batch_id", "STRING", batch_id)]
    audit_rows = []
    for target in targets:
        job = client.query(build_dml(target),
                           job_config=bigquery.QueryJobConfig(query_parameters=params))
        job.result()  # ⑥ 失敗はここで例外。握り潰さない
        audit_rows.append({
            "batch_id": batch_id, "table_id": target.table_id,
            "action": target.action.value,
            "affected_rows": job.num_dml_affected_rows or 0,
            "job_id": job.job_id, "executed_at": now.isoformat(),
        })
    if errors := client.insert_rows_json(AUDIT_LOG, audit_rows):
        raise RuntimeError(f"監査ログの書き込みに失敗: {errors}")

    # ③ 全テーブル成功後に台帳を進める。通知はまだしない
    client.query(
        f"""UPDATE `{LEDGER}`
            SET status = 'propagated', propagated_at = @now,
                erasable_after = TIMESTAMP_ADD(@now, INTERVAL {PHYSICAL_ERASURE_DAYS} DAY)
            WHERE batch_id = @batch_id""",
        job_config=bigquery.QueryJobConfig(query_parameters=params + [
            bigquery.ScalarQueryParameter("now", "TIMESTAMP", now)]),
    ).result()