Gemini API × Cloudflare D1：エッジSQLiteでAIバックエンドを月$10以下に収めた実装メモ

name = "gemini-ai-backend"
main = "src/index.ts"
compatibility_date = "2026-01-01"
compatibility_flags = ["nodejs_compat"]
 
[vars]
MODEL_ID = "gemini-2.5-flash"
 
[[d1_databases]]
binding = "DB"
database_name = "ai-conversations"
database_id = "your-d1-database-id"
 
# wrangler secret put GEMINI_API_KEY で設定（toml には書かない）

-- セッション管理テーブル
CREATE TABLE IF NOT EXISTS conversations (
  id TEXT PRIMARY KEY DEFAULT (lower(hex(randomblob(16)))),
  user_id TEXT NOT NULL,
  created_at INTEGER NOT NULL DEFAULT (unixepoch()),
  updated_at INTEGER NOT NULL DEFAULT (unixepoch()),
  title TEXT,
  model_id TEXT NOT NULL DEFAULT 'gemini-2.5-flash'
);
 
-- メッセージ（会話履歴）テーブル
CREATE TABLE IF NOT EXISTS messages (
  id TEXT PRIMARY KEY DEFAULT (lower(hex(randomblob(16)))),
  conversation_id TEXT NOT NULL REFERENCES conversations(id),
  role TEXT NOT NULL CHECK (role IN ('user', 'model')),
  content TEXT NOT NULL,
  token_count INTEGER DEFAULT 0,
  created_at INTEGER NOT NULL DEFAULT (unixepoch())
);
 
-- 使用量・コスト追跡テーブル
CREATE TABLE IF NOT EXISTS usage_records (
  id TEXT PRIMARY KEY DEFAULT (lower(hex(randomblob(16)))),
  user_id TEXT NOT NULL,
  date TEXT NOT NULL,
  request_count INTEGER NOT NULL DEFAULT 0,
  input_tokens INTEGER NOT NULL DEFAULT 0,
  output_tokens INTEGER NOT NULL DEFAULT 0,
  estimated_cost_usd REAL NOT NULL DEFAULT 0.0,
  UNIQUE(user_id, date)
);
 
-- 検索高速化インデックス
CREATE INDEX IF NOT EXISTS idx_messages_conversation
  ON messages(conversation_id, created_at);
CREATE INDEX IF NOT EXISTS idx_usage_user_date
  ON usage_records(user_id, date);

// src/gemini.ts
export interface Message {
  role: 'user' | 'model';
  parts: [{ text: string }];
}
 
// ストリーミング版：最速レスポンスが必要な場合
export async function streamGemini(
  apiKey: string,
  modelId: string,
  messages: Message[],
  systemInstruction?: string
): Promise<Response> {
  const endpoint =
    `https://generativelanguage.googleapis.com/v1beta/models/${modelId}` +
    `:streamGenerateContent?alt=sse&key=${apiKey}`;
 
  const requestBody: Record<string, unknown> = {
    contents: messages,
    generationConfig: {
      temperature: 0.7,
      maxOutputTokens: 2048,
    },
  };
 
  if (systemInstruction) {
    requestBody.systemInstruction = {
      parts: [{ text: systemInstruction }],
    };
  }
 
  const upstream = await fetch(endpoint, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(requestBody),
  });
 
  if (!upstream.ok) {
    const errorText = await upstream.text();
    throw new Error(`Gemini API error ${upstream.status}: ${errorText}`);
  }
 
  // SSE ストリームをクライアントへそのままパイプ（バッファリングなし）
  return new Response(upstream.body, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Access-Control-Allow-Origin': '*',
    },
  });
}
 
// 非ストリーミング版：使用量を記録したい場合
export async function generateWithUsage(
  apiKey: string,
  modelId: string,
  messages: Message[]
): Promise<{ text: string; inputTokens: number; outputTokens: number }> {
  const endpoint =
    `https://generativelanguage.googleapis.com/v1beta/models/${modelId}` +
    `:generateContent?key=${apiKey}`;
 
  const response = await fetch(endpoint, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ contents: messages }),
  });
 
  if (!response.ok) {
    const errText = await response.text();
    throw new Error(`Gemini API error ${response.status}: ${errText}`);
  }
 
  const data = await response.json() as {
    candidates: [{ content: { parts: [{ text: string }] } }];
    usageMetadata: { promptTokenCount: number; candidatesTokenCount: number };
  };
 
  return {
    text: data.candidates[0].content.parts[0].text,
    inputTokens: data.usageMetadata.promptTokenCount,
    outputTokens: data.usageMetadata.candidatesTokenCount,
  };
}

// src/db.ts
export interface Env {
  DB: D1Database;
  GEMINI_API_KEY: string;
}
 
// 会話履歴を取得（直近 N 件のメッセージ）
export async function getConversationHistory(
  db: D1Database,
  conversationId: string,
  limit = 20
): Promise<Message[]> {
  const { results } = await db
    .prepare(`
      SELECT role, content
      FROM messages
      WHERE conversation_id = ?
      ORDER BY created_at DESC
      LIMIT ?
    `)
    .bind(conversationId, limit)
    .all<{ role: string; content: string }>();
 
  // 時系列順に並び替えて Gemini API 形式に変換
  return results.reverse().map(row => ({
    role: row.role as 'user' | 'model',
    parts: [{ text: row.content }],
  }));
}
 
// メッセージをバッチ保存（会話の updated_at も同時更新）
export async function saveMessage(
  db: D1Database,
  conversationId: string,
  role: 'user' | 'model',
  content: string,
  tokenCount = 0
): Promise<void> {
  await db.batch([
    db.prepare(`
      INSERT INTO messages (conversation_id, role, content, token_count)
      VALUES (?, ?, ?, ?)
    `).bind(conversationId, role, content, tokenCount),
 
    db.prepare(`
      UPDATE conversations SET updated_at = unixepoch() WHERE id = ?
    `).bind(conversationId),
  ]);
}
 
// 使用量をアップサート（SELECT 不要の1クエリ更新）
export async function recordUsage(
  db: D1Database,
  userId: string,
  inputTokens: number,
  outputTokens: number
): Promise<void> {
  const today = new Date().toISOString().split('T')[0];
 
  // Gemini 2.5 Flash 料金例: 入力 $0.15/1M tokens, 出力 $0.60/1M tokens
  const costUsd =
    (inputTokens / 1_000_000) * 0.15 +
    (outputTokens / 1_000_000) * 0.60;
 
  await db
    .prepare(`
      INSERT INTO usage_records
        (user_id, date, request_count, input_tokens, output_tokens, estimated_cost_usd)
      VALUES (?, ?, 1, ?, ?, ?)
      ON CONFLICT(user_id, date) DO UPDATE SET
        request_count      = request_count + 1,
        input_tokens       = input_tokens + excluded.input_tokens,
        output_tokens      = output_tokens + excluded.output_tokens,
        estimated_cost_usd = estimated_cost_usd + excluded.estimated_cost_usd
    `)
    .bind(userId, today, inputTokens, outputTokens, costUsd)
    .run();
}

// src/index.ts
import { streamGemini } from './gemini';
import { getConversationHistory, saveMessage, recordUsage } from './db';
 
export interface Env {
  DB: D1Database;
  GEMINI_API_KEY: string;
}
 
async function checkRateLimit(
  db: D1Database,
  userId: string,
  dailyLimit = 100,
  dailyCostLimitUsd = 1.0
): Promise<{ allowed: boolean; reason?: string }> {
  const today = new Date().toISOString().split('T')[0];
 
  const record = await db
    .prepare(`
      SELECT request_count, estimated_cost_usd
      FROM usage_records
      WHERE user_id = ? AND date = ?
    `)
    .bind(userId, today)
    .first<{ request_count: number; estimated_cost_usd: number }>();
 
  if (!record) return { allowed: true };
 
  if (record.request_count >= dailyLimit) {
    return {
      allowed: false,
      reason: `1日のリクエスト数上限（${dailyLimit}回）に達しました`,
    };
  }
 
  if (record.estimated_cost_usd >= dailyCostLimitUsd) {
    return {
      allowed: false,
      reason: `1日の利用コスト上限（$${dailyCostLimitUsd}）に達しました`,
    };
  }
 
  return { allowed: true };
}
 
export default {
  async fetch(
    request: Request,
    env: Env,
    ctx: ExecutionContext
  ): Promise<Response> {
    // CORS プリフライト対応
    if (request.method === 'OPTIONS') {
      return new Response(null, {
        headers: {
          'Access-Control-Allow-Origin': '*',
          'Access-Control-Allow-Methods': 'POST, OPTIONS',
          'Access-Control-Allow-Headers': 'Content-Type, Authorization',
        },
      });
    }
 
    if (request.method !== 'POST') {
      return new Response('Method Not Allowed', { status: 405 });
    }
 
    let body: { userId: string; conversationId?: string; message: string };
    try {
      body = await request.json();
    } catch {
      return new Response(
        JSON.stringify({ error: 'Invalid JSON' }),
        { status: 400, headers: { 'Content-Type': 'application/json' } }
      );
    }
 
    const { userId, conversationId, message } = body;
 
    if (!userId || !message) {
      return new Response(
        JSON.stringify({ error: 'userId と message は必須です' }),
        { status: 400, headers: { 'Content-Type': 'application/json' } }
      );
    }
 
    // ① レート制限チェック（Gemini API 呼び出し前に実行してコスト節約）
    const rateCheck = await checkRateLimit(env.DB, userId);
    if (!rateCheck.allowed) {
      return new Response(
        JSON.stringify({ error: rateCheck.reason }),
        { status: 429, headers: { 'Content-Type': 'application/json' } }
      );
    }
 
    // ② D1 から会話履歴取得（直近20件に絞ってコスト管理）
    const history = conversationId
      ? await getConversationHistory(env.DB, conversationId)
      : [];
 
    history.push({ role: 'user', parts: [{ text: message }] });
 
    try {
      // ③ Gemini API ストリーミング呼び出し
      const streamResponse = await streamGemini(
        env.GEMINI_API_KEY,
        'gemini-2.5-flash',
        history
      );
 
      // ④ レスポンス返却後にバックグラウンドで D1 保存（ユーザーを待たせない）
      if (conversationId) {
        ctx.waitUntil(
          saveMessage(env.DB, conversationId, 'user', message)
        );
      }
 
      return streamResponse;
    } catch (error) {
      const errorMessage =
        error instanceof Error ? error.message : 'Unknown error';
      console.error('Gemini API error:', errorMessage);
 
      return new Response(
        JSON.stringify({ error: 'レスポンスの生成に失敗しました' }),
        { status: 500, headers: { 'Content-Type': 'application/json' } }
      );
    }
  },
};

// ストリームを返し終えてから書き込む → 書き込み時間ぶん完了が遅れる
const fullText = await streamToClient(geminiStream, writer);
await saveMessage(env.DB, conversationId, 'model', fullText, outputTokens);
await recordUsage(env.DB, userId, inputTokens, outputTokens);
return new Response(readable, { headers: streamHeaders });

// 本文の送出が終わった後の永続化を waitUntil に委ねる
ctx.waitUntil(
  (async () => {
    const fullText = await collectStreamedText(); // ストリーム中に蓄積したテキスト
    // D1 書き込みは一時的な競合で失敗しうるので軽くリトライする
    for (let attempt = 0; attempt < 3; attempt++) {
      try {
        await saveMessage(env.DB, conversationId, 'model', fullText, outputTokens);
        await recordUsage(env.DB, userId, inputTokens, outputTokens);
        break;
      } catch (err) {
        if (attempt === 2) console.error('persist failed after retries', err);
        await new Promise(r => setTimeout(r, 50 * (attempt + 1)));
      }
    }
  })()
);
return new Response(readable, { headers: streamHeaders });

# ローカル開発サーバー起動
wrangler dev
 
# D1 スキーマを本番に適用
wrangler d1 execute ai-conversations --file=schema.sql
 
# API キーをシークレットとして設定（wrangler.toml には書かない）
wrangler secret put GEMINI_API_KEY
 
# 本番デプロイ
wrangler deploy
 
# リアルタイムログ監視（エラー調査に必須）
wrangler tail

# .github/workflows/deploy.yml
name: Deploy to Cloudflare Workers
on:
  push:
    branches: [main]
 
jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-node@v4
        with:
          node-version: '20'
      - run: npm ci
      - run: npx wrangler deploy
        env:
          CLOUDFLARE_API_TOKEN: ${{ secrets.CF_API_TOKEN }}