Gemini Live API で音声 SaaS を構築する実装 — リアルタイム会話と Stripe 課金の連動

// src/lib/gemini-live-client.ts
import WebSocket from "ws";
 
export interface GeminiLiveConfig {
  apiKey: string;
  model?: string;
  systemInstruction?: string;
}
 
export class GeminiLiveClient {
  private ws: WebSocket | null = null;
  private config: GeminiLiveConfig;
  private onAudioCallback: ((audio: ArrayBuffer) => void) | null = null;
 
  constructor(config: GeminiLiveConfig) {
    this.config = config;
  }
 
  async connect(): Promise<void> {
    const url = `wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent?key=${this.config.apiKey}`;
    this.ws = new WebSocket(url);
 
    return new Promise((resolve, reject) => {
      this.ws!.on("open", () => {
        // 初期セットアップメッセージを送信
        this.ws!.send(JSON.stringify({
          setup: {
            model: `models/${this.config.model ?? "gemini-2.5-flash-live"}`,
            generation_config: {
              response_modalities: ["AUDIO"],
              speech_config: {
                voice_config: { prebuilt_voice_config: { voice_name: "Aoede" } }
              }
            },
            system_instruction: this.config.systemInstruction
              ? { parts: [{ text: this.config.systemInstruction }] }
              : undefined,
          }
        }));
        resolve();
      });
      this.ws!.on("error", reject);
      this.ws!.on("message", (data: Buffer) => {
        const msg = JSON.parse(data.toString());
        if (msg.serverContent?.modelTurn?.parts?.[0]?.inlineData?.data) {
          const audio = Buffer.from(msg.serverContent.modelTurn.parts[0].inlineData.data, "base64");
          this.onAudioCallback?.(audio.buffer);
        }
      });
    });
  }
 
  sendAudio(audioChunk: ArrayBuffer): void {
    if (!this.ws || this.ws.readyState !== WebSocket.OPEN) return;
    const base64Audio = Buffer.from(audioChunk).toString("base64");
    this.ws.send(JSON.stringify({
      realtime_input: {
        media_chunks: [{
          mime_type: "audio/pcm;rate=16000",
          data: base64Audio
        }]
      }
    }));
  }
 
  onAudio(callback: (audio: ArrayBuffer) => void): void {
    this.onAudioCallback = callback;
  }
 
  close(): void {
    this.ws?.close();
  }
}

// src/durable-objects/voice-session.ts
import { DurableObject } from "cloudflare:workers";
import { GeminiLiveClient } from "../lib/gemini-live-client";
 
export class VoiceSession extends DurableObject {
  private clientWs: WebSocket | null = null;
  private geminiClient: GeminiLiveClient | null = null;
  private startTime: number = 0;
  private userId: string = "";
  private stripeCustomerId: string = "";
 
  async fetch(req: Request): Promise<Response> {
    const url = new URL(req.url);
    this.userId = url.searchParams.get("userId") ?? "";
    this.stripeCustomerId = url.searchParams.get("customerId") ?? "";
 
    const upgradeHeader = req.headers.get("Upgrade");
    if (upgradeHeader !== "websocket") {
      return new Response("Expected WebSocket", { status: 400 });
    }
 
    const { 0: client, 1: server } = new WebSocketPair();
    server.accept();
    this.clientWs = server;
 
    // Gemini Live クライアント初期化
    this.geminiClient = new GeminiLiveClient({
      apiKey: this.env.GEMINI_API_KEY,
      systemInstruction: "あなたは親切なカスタマーサポートです。",
    });
    await this.geminiClient.connect();
 
    // Gemini からの応答音声をクライアントに転送
    this.geminiClient.onAudio((audio) => {
      this.clientWs?.send(audio);
    });
 
    // クライアントからの音声を Gemini に転送
    server.addEventListener("message", (event) => {
      if (event.data instanceof ArrayBuffer) {
        this.geminiClient?.sendAudio(event.data);
      }
    });
 
    // セッション終了時に課金処理
    server.addEventListener("close", async () => {
      const durationSeconds = Math.ceil((Date.now() - this.startTime) / 1000);
      await this.recordUsage(durationSeconds);
      this.geminiClient?.close();
    });
 
    this.startTime = Date.now();
    return new Response(null, { status: 101, webSocket: client });
  }
 
  private async recordUsage(durationSeconds: number) {
    // 後述の Stripe Meter Events 送信処理を呼ぶ
    await fetch("https://internal.api/record-voice-usage", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        customerId: this.stripeCustomerId,
        userId: this.userId,
        durationSeconds,
        sessionId: this.ctx.id.toString(),
      }),
    });
  }
}

// src/lib/voice-billing.ts
import Stripe from "stripe";
 
export async function recordVoiceUsage(args: {
  customerId: string;
  userId: string;
  durationSeconds: number;
  sessionId: string; // 冪等性キーとして使用
}) {
  const stripe = new Stripe(process.env.STRIPE_SECRET_KEY!, {
    apiVersion: "2024-10-28.acacia",
  });
 
  // 通話時間が極端に短い場合は課金しない（接続失敗とみなす）
  if (args.durationSeconds < 3) {
    console.log(`Session ${args.sessionId} too short, skipping billing`);
    return null;
  }
 
  // Meter Events に送信
  try {
    const event = await stripe.billing.meterEvents.create({
      event_name: "voice_call_seconds",
      payload: {
        stripe_customer_id: args.customerId,
        seconds: args.durationSeconds.toString(),
      },
      identifier: args.sessionId,
    });
    console.log(`Billed ${args.durationSeconds}s for ${args.userId}`);
    return event;
  } catch (err) {
    console.error("Meter event failed", err);
    // DLQ に保存して後でリトライ
    await saveToDLQ(args);
    return null;
  }
}

// public/voice-client.ts
class VoiceClient {
  private ws: WebSocket | null = null;
  private mediaRecorder: MediaRecorder | null = null;
  private audioContext: AudioContext | null = null;
 
  async startCall(userId: string, customerId: string) {
    // マイク権限を取得
    const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
 
    // WebSocket 接続
    this.ws = new WebSocket(`wss://api.example.com/voice?userId=${userId}&customerId=${customerId}`);
    this.ws.binaryType = "arraybuffer";
 
    this.audioContext = new AudioContext({ sampleRate: 16000 });
 
    // サーバーからの音声を再生
    this.ws.addEventListener("message", async (event) => {
      if (event.data instanceof ArrayBuffer) {
        const audioBuffer = await this.audioContext!.decodeAudioData(event.data);
        const source = this.audioContext!.createBufferSource();
        source.buffer = audioBuffer;
        source.connect(this.audioContext!.destination);
        source.start();
      }
    });
 
    // マイク音声を WebSocket で送信
    this.mediaRecorder = new MediaRecorder(stream, { mimeType: "audio/webm" });
    this.mediaRecorder.ondataavailable = async (e) => {
      if (e.data.size > 0 && this.ws?.readyState === WebSocket.OPEN) {
        const arrayBuffer = await e.data.arrayBuffer();
        this.ws.send(arrayBuffer);
      }
    };
    // 100ms ごとに送信（低レイテンシ）
    this.mediaRecorder.start(100);
  }
 
  endCall() {
    this.mediaRecorder?.stop();
    this.ws?.close();
    this.audioContext?.close();
  }
}

// src/durable-objects/voice-session.ts（前述クラスへ追記）
private lastActivity: number = 0;
private readonly IDLE_TIMEOUT_MS = 30_000; // 30秒無音で強制終了
 
private touch() {
  this.lastActivity = Date.now();
  // 既存の alarm を上書きして延長する
  this.ctx.storage.setAlarm(Date.now() + this.IDLE_TIMEOUT_MS);
}
 
// メッセージ受信のたびに touch() を呼ぶ
//   server.addEventListener("message", (event) => { this.touch(); /* ...既存処理... */ });
 
async alarm() {
  const idle = Date.now() - this.lastActivity;
  if (idle < this.IDLE_TIMEOUT_MS) {
    // まだ生きているので alarm を再設定する
    this.ctx.storage.setAlarm(Date.now() + (this.IDLE_TIMEOUT_MS - idle));
    return;
  }
  // 無音が続いた → 切断とみなして確実に課金する
  const durationSeconds = Math.ceil((this.lastActivity - this.startTime) / 1000);
  await this.recordUsage(durationSeconds);
  this.geminiClient?.close();
  this.clientWs?.close();
}

// src/lib/billing-dlq.ts
interface FailedUsage {
  customerId: string;
  userId: string;
  durationSeconds: number;
  sessionId: string;
  failedAt: number;
}
 
// 失敗イベントを KV に退避する
export async function saveToDLQ(args: Omit<FailedUsage, "failedAt">, kv: KVNamespace) {
  const record: FailedUsage = { ...args, failedAt: Date.now() };
  // sessionId をキーにするので、同じ通話の重複退避は自然に防げる
  await kv.put(`dlq:voice:${args.sessionId}`, JSON.stringify(record), {
    expirationTtl: 60 * 60 * 24 * 7, // 7日
  });
}
 
// Cron Trigger で定期的に再送する（例: 10分ごと）
export async function retryDLQ(kv: KVNamespace, stripe: Stripe) {
  const list = await kv.list({ prefix: "dlq:voice:" });
  let recovered = 0;
  for (const key of list.keys) {
    const raw = await kv.get(key.name);
    if (!raw) continue;
    const r: FailedUsage = JSON.parse(raw);
    try {
      await stripe.billing.meterEvents.create({
        event_name: "voice_call_seconds",
        payload: { stripe_customer_id: r.customerId, seconds: r.durationSeconds.toString() },
        identifier: r.sessionId, // 冪等なので二重課金にならない
      });
      await kv.delete(key.name);
      recovered++;
    } catch {
      // まだ復旧していなければ次回へ持ち越す
    }
  }
  return recovered;
}