Antigravity と Inngest で組み立てるイベント駆動 AI ワークフロー — 本番運用を見据えた実装ガイド

# Antigravity のターミナルから実行
npx create-next-app@latest ai-workflows --typescript --app --tailwind
cd ai-workflows
npm install inngest @google/genai zod
npm install -D @types/node tsx

// src/inngest/client.ts
import { Inngest } from "inngest";
import { z } from "zod";
 
// イベント型を Zod で定義しておくと、関数側で自動的に型が効きます
const eventSchemas = {
  "support/inquiry.received": {
    data: z.object({
      inquiryId: z.string().uuid(),
      userId: z.string(),
      body: z.string().min(1).max(8000),
      locale: z.enum(["ja", "en"]).default("ja"),
    }),
  },
  "support/reply.approved": {
    data: z.object({
      inquiryId: z.string().uuid(),
      approvedBy: z.string(),
    }),
  },
} as const;
 
export const inngest = new Inngest({
  id: "antigravity-support-bot",
  // 本番では環境変数経由で signing key を渡す
  schemas: eventSchemas,
});

// src/inngest/functions/handle-inquiry.ts
import { inngest } from "../client";
import { GoogleGenAI } from "@google/genai";
import { NonRetriableError } from "inngest";
 
const ai = new GoogleGenAI({ apiKey: process.env.GEMINI_API_KEY! });
 
export const handleInquiry = inngest.createFunction(
  {
    id: "handle-inquiry",
    // 同一ユーザーから連投された場合、5 秒以内のものは 1 件として扱う
    debounce: { period: "5s", key: "event.data.userId" },
    retries: 3,
  },
  { event: "support/inquiry.received" },
  async ({ event, step, logger }) => {
    const { inquiryId, body, locale } = event.data;
 
    // 1. AI 要約（高コストなので step.run でチェックポイント化）
    const summary = await step.run("summarize-with-gemini", async () => {
      const resp = await ai.models.generateContent({
        model: "gemini-2.5-flash",
        contents: `次の問い合わせを ${locale === "ja" ? "日本語" : "英語"}で 80 字以内に要約してください:\n\n${body}`,
      });
      const text = resp.text?.trim();
      if (!text) {
        // 空応答は再試行しても無駄なので非リトライ扱い
        throw new NonRetriableError("Gemini returned empty response");
      }
      return text;
    });
 
    // 2. DB に下書き保存(冪等キーは inquiryId)
    await step.run("save-draft-reply", async () => {
      await fetch(`${process.env.APP_URL}/api/internal/save-draft`, {
        method: "POST",
        headers: { "content-type": "application/json", "x-idempotency-key": inquiryId },
        body: JSON.stringify({ inquiryId, summary }),
      });
    });
 
    // 3. Slack に通知
    await step.run("notify-slack", async () => {
      await fetch(process.env.SLACK_WEBHOOK_URL!, {
        method: "POST",
        headers: { "content-type": "application/json" },
        body: JSON.stringify({
          text: `📩 新しい問い合わせ要約: ${summary}\n承認 → /approve ${inquiryId}`,
        }),
      });
    });
 
    logger.info({ inquiryId }, "inquiry summarized and notified");
    return { inquiryId, summary };
  }
);

// src/inngest/functions/generate-monthly-report.ts
import { inngest } from "../client";
 
export const generateMonthlyReport = inngest.createFunction(
  {
    id: "generate-monthly-report",
    // 同時実行は最大 8 つ。ユーザー単位ではなく関数全体で制御
    concurrency: {
      limit: 8,
      key: "event.data.tenantId", // テナントごとに独立した concurrency
    },
    // 1 分あたり最大 60 リクエストに絞る(Gemini Free の上限想定)
    throttle: {
      limit: 60,
      period: "1m",
      key: "event.data.tenantId",
    },
    retries: 5,
  },
  { event: "report/generate.requested" },
  async ({ event, step }) => {
    const { userId, month } = event.data;
 
    const sections = await step.run("collect-data", async () => {
      // ユーザーのアクティビティを集計して返す
      return await collectUserActivity(userId, month);
    });
 
    const draft = await step.run("ai-write-report", async () => {
      // 集計結果を Gemini に渡してレポート本文を生成
      return await composeReportWithGemini(sections);
    });
 
    await step.run("publish-report", async () => {
      await publishToUser(userId, draft);
    });
 
    return { userId, month, draftId: draft.id };
  }
);
 
// ヘルパは実装簡略化のため省略しています
async function collectUserActivity(userId: string, month: string) { /* ... */ }
async function composeReportWithGemini(sections: unknown) { /* ... */ }
async function publishToUser(userId: string, draft: unknown) { /* ... */ }

// src/inngest/functions/auto-reply-with-approval.ts
import { inngest } from "../client";
 
export const autoReplyWithApproval = inngest.createFunction(
  { id: "auto-reply-with-approval", retries: 2 },
  { event: "support/inquiry.received" },
  async ({ event, step }) => {
    const { inquiryId } = event.data;
 
    // 下書き生成(別関数に丸投げしてもよい)
    const draft = await step.run("generate-draft", async () => {
      return await generateDraftWithGemini(event.data.body);
    });
 
    // 24 時間以内に承認イベントが来たら受け取り、来なければタイムアウト
    const approval = await step.waitForEvent("wait-for-approval", {
      event: "support/reply.approved",
      timeout: "24h",
      // 同じ inquiryId の承認だけを待つ
      if: `event.data.inquiryId == "${inquiryId}"`,
    });
 
    if (!approval) {
      // タイムアウト時は人間にエスカレーション
      await step.run("escalate", async () => {
        await sendEscalationEmail(inquiryId, draft);
      });
      return { status: "escalated", inquiryId };
    }
 
    // 承認されたら送信
    await step.run("send-reply", async () => {
      await sendEmail({
        inquiryId,
        body: draft.body,
        approvedBy: approval.data.approvedBy,
      });
    });
 
    return { status: "sent", inquiryId, approvedBy: approval.data.approvedBy };
  }
);
 
async function generateDraftWithGemini(body: string) { /* ... */ return { body: "" }; }
async function sendEscalationEmail(inquiryId: string, draft: { body: string }) { /* ... */ }
async function sendEmail(args: { inquiryId: string; body: string; approvedBy: string }) { /* ... */ }

// src/inngest/functions/safe-charge.ts
import { inngest } from "../client";
import { NonRetriableError, RetryAfterError } from "inngest";
 
export const safeCharge = inngest.createFunction(
  { id: "safe-charge", retries: 4 },
  { event: "billing/charge.requested" },
  async ({ event, step }) => {
    await step.run("charge-stripe", async () => {
      const res = await fetch("https://api.stripe.com/v1/charges", {
        method: "POST",
        headers: {
          authorization: `Bearer ${process.env.STRIPE_SECRET_KEY}`,
          // 冪等キーを必ず付ける
          "idempotency-key": event.data.checkoutSessionId,
        },
        body: new URLSearchParams({ amount: String(event.data.amount), currency: "jpy" }),
      });
 
      if (res.status === 429) {
        const retryAfter = Number(res.headers.get("retry-after") ?? "10");
        throw new RetryAfterError("rate limited by Stripe", `${retryAfter}s`);
      }
      if (res.status === 400) {
        // 400 は再試行しても回復しないので非リトライ
        throw new NonRetriableError(`Stripe rejected: ${await res.text()}`);
      }
      if (!res.ok) {
        // 5xx は通常リトライに任せる
        throw new Error(`Stripe error: ${res.status}`);
      }
      return await res.json();
    });
  }
);