C データエンジニアリング — ポイント付与基盤 Pub/Sub at-least-once重複配信（BigQuery Storage Write APIにオフセット未指定で重複書き込み × dbt incrementalがappendで生ログ重複がマートへ伝播 × ポイント計算ロジックにdbt Unit Testsが無く倍率参照ミスを検知できず × パーティションが取り込み日ベースでバックフィル時にズレる × クラスタリング未設定 × バックフィルDAGにチェックポイントが無く失敗のたび全再実行）（MOps 販促システムチーム ポイント10倍誤付与事故 Bad→Good）

📋 コピーversion: 2

models:
  - name: mart_point_ledger
    description: >
      ユーザーごとのポイント付与台帳。生ログ(raw_events.point_grant_events)から
      incremental mergeで構築し、event_id単位で重複を最終的に収束させる。
    config:
      materialized: incremental
      incremental_strategy: merge   # 修正③: append→mergeへ変更
      unique_key: event_id          # 修正③: 生ログ側の重複をここで吸収する最終防波堤
    columns:
      - name: event_id
        data_type: string
        constraints: [{ type: not_null }, { type: primary_key }]
      - name: user_id
        data_type: string
        constraints: [{ type: not_null }]
      - name: point_amount
        data_type: numeric

📋 コピーunit_tests:
  - name: test_point_amount_normal_campaign
    description: >
      通常キャンペーン(倍率2倍)時、購入金額1000円に対して
      1%基本レート×2倍 = 20ポイントが計算されることを検証する。
      今回の事故のような「倍率マスタの結合ミスで倍率が異常値になる」
      回帰をCIで検知するための固定入力テスト。
    model: mart_point_ledger
    given:
      - input: ref('stg_point_grant_events')
        rows:
          - { event_id: "evt_1", purchase_amount: 1000, campaign_multiplier: 2 }
      - input: ref('stg_campaign_master')
        rows:
          - { campaign_id: "camp_default", multiplier: 2 }
    expect:
      rows:
        - { event_id: "evt_1", point_amount: 20 }   # 1000 * 0.01 * 2

  - name: test_point_amount_multiplier_join_miss_is_capped
    description: >
      倍率マスタの結合キーが一致せずNULLになった場合、
      異常な倍率でポイントが計算されない(基本レートのみが
      適用される)ことを検証する。
    model: mart_point_ledger
    given:
      - input: ref('stg_point_grant_events')
        rows:
          - { event_id: "evt_2", purchase_amount: 1000, campaign_multiplier: null }
    expect:
      rows:
        - { event_id: "evt_2", point_amount: 10 }   # 1000 * 0.01 * 1(デフォルト倍率)

📋 コピーapiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: backfill-point-events-
spec:
  entrypoint: backfill
  arguments:
    parameters:
      - name: target-date   # 例: 2026-08-20

  templates:
    - name: backfill
      steps:
        # 修正⑦-1: 対象日をマイクロバッチ(1時間単位)に分割し、
        #          チェックポイントテーブルで成功済みバッチを事前に除外する
        - - name: list-pending-batches
            template: list-pending-batches
            arguments:
              parameters: [{ name: target-date, value: "{{workflow.parameters.target-date}}" }]
        - - name: process-batch
            template: process-hourly-batch
            arguments:
              parameters: [{ name: hour, value: "{{item}}" }]
            withParam: "{{steps.list-pending-batches.outputs.result}}"

    - name: list-pending-batches
      inputs:
        parameters: [{ name: target-date }]
      script:
        image: gcr.io/mops-project/bq-tools:1.8
        command: [python]
        source: |
          # 修正⑦-2: ops_backfill_checkpoints に status='succeeded' が
          #          記録されていない時間帯(hour)のみを再処理対象として返す
          from google.cloud import bigquery
          client = bigquery.Client()
          query = """
            SELECT hour
            FROM UNNEST(GENERATE_ARRAY(0, 23)) AS hour
            WHERE hour NOT IN (
              SELECT hour FROM `mops-project.ops.backfill_checkpoints`
              WHERE target_date = @target_date AND status = 'succeeded'
            )
          """
          job = client.query(query, job_config=bigquery.QueryJobConfig(
              query_parameters=[
                  bigquery.ScalarQueryParameter("target_date", "DATE", "{{inputs.parameters.target-date}}")
              ]
          ))
          print([row.hour for row in job.result()])

    - name: process-hourly-batch
      inputs:
        parameters: [{ name: hour }]
      steps:
        - - name: reload-batch
            template: reload-hourly-batch
        - - name: mark-succeeded
            # 修正⑦-3: マイクロバッチ単位で成功をチェックポイントに記録
            #          (dbt側のmergeと合わせた二重防御)
            template: mark-checkpoint-succeeded