Antigravity エージェントの従量課金が「内部の利用量」と「Stripeの請求」でずれる — 計測の冪等化・遅延吸収・突合の実装メモ

import crypto from 'node:crypto';
 
function meterIdentifier(executionId, stepNumber) {
  // executionId は UUIDv4 以上のエントロピーを持たせる。
  // タイムスタンプ＋ユーザーIDのような推測可能な値は並列実行で衝突しうる。
  return crypto
    .createHash('sha256')
    .update(`${executionId}:${stepNumber}`)
    .digest('hex')
    .slice(0, 40);
}

CREATE TABLE usage_ledger (
  identifier   TEXT PRIMARY KEY,      -- meterIdentifier と同一
  customer_id  TEXT NOT NULL,
  event_name   TEXT NOT NULL,
  value        INTEGER NOT NULL,
  occurred_at  INTEGER NOT NULL,      -- イベント発生時刻（UNIX秒）
  sent_at      INTEGER,               -- Stripe受理時刻。NULL = 未送信
  billing_month TEXT NOT NULL         -- occurred_at から導出した 'YYYY-MM'
);

// 実行側：台帳への書き込みが成功した時点で「計測は確定」とみなす
async function recordUsage(db, { executionId, stepNumber, customerId, value, occurredAt }) {
  const identifier = meterIdentifier(executionId, stepNumber);
  const month = new Date(occurredAt * 1000).toISOString().slice(0, 7);
  await db.prepare(
    `INSERT INTO usage_ledger (identifier, customer_id, event_name, value, occurred_at, billing_month)
     VALUES (?, ?, 'agent_steps', ?, ?, ?)
     ON CONFLICT(identifier) DO NOTHING`
  ).bind(identifier, customerId, value, occurredAt, month).run();
}
 
// ワーカー側：未送信行を拾って Stripe に送り、受理できたら sent_at を打つ
async function flushLedger(db, stripe) {
  const { results } = await db.prepare(
    `SELECT * FROM usage_ledger WHERE sent_at IS NULL ORDER BY occurred_at LIMIT 200`
  ).all();
  for (const row of results) {
    try {
      await stripe.billing.meterEvents.create({
        event_name: row.event_name,
        payload: { stripe_customer_id: row.customer_id, value: String(row.value) },
        identifier: row.identifier,
        timestamp: row.occurred_at,
      });
      await db.prepare(`UPDATE usage_ledger SET sent_at = ? WHERE identifier = ?`)
        .bind(Math.floor(Date.now() / 1000), row.identifier).run();
    } catch (e) {
      // 送信失敗は sent_at を打たずに残す。冪等キーがあるので次回再送して安全。
      console.error('meter flush failed', row.identifier, e.message);
    }
  }
}

async function reconcile(db, stripe, { meterId, month, customerId }) {
  // 1) 内部台帳：当月・送信済みの合計
  const ledger = await db.prepare(
    `SELECT COALESCE(SUM(value), 0) AS total
       FROM usage_ledger
      WHERE customer_id = ? AND billing_month = ? AND sent_at IS NOT NULL`
  ).bind(customerId, month).first();
 
  // 2) Stripe メーター集計サマリ（当月の合計）
  const start = Math.floor(new Date(`${month}-01T00:00:00Z`).getTime() / 1000);
  const end = Math.floor(Date.now() / 1000);
  const summaries = await stripe.billing.meters.eventSummaries.list(meterId, {
    customer: customerId,
    start_time: start,
    end_time: end,
    value_grouping_window: 'day',
  });
  const stripeTotal = summaries.data.reduce((s, x) => s + x.aggregated_value, 0);
 
  // 3) ドリフトを評価。絶対差と相対差の両方でしきい値を切る。
  const drift = ledger.total - stripeTotal;
  const ratio = stripeTotal === 0 ? (ledger.total === 0 ? 0 : 1) : drift / stripeTotal;
  const significant = Math.abs(drift) > 50 || Math.abs(ratio) > 0.02;
 
  return { customerId, month, ledgerTotal: ledger.total, stripeTotal, drift, ratio, significant };
}