SQLiteでクロススタックオブザーバビリティを実現する：Next.js + FastAPIのトレースID連携ログ基盤

CREATE TABLE IF NOT EXISTS logs (
  id            INTEGER PRIMARY KEY AUTOINCREMENT,
  ts_unix_ms    INTEGER NOT NULL,
  level         TEXT NOT NULL,       -- DEBUG/INFO/WARN/ERROR
  source        TEXT NOT NULL,       -- "bff" or "backend"
  logger        TEXT,
  message       TEXT NOT NULL,
  trace_id      TEXT,
  request_id    TEXT,
  span_id       TEXT,
  pid           INTEGER,
  host          TEXT,
  route         TEXT,
  method        TEXT,
  status_code   INTEGER,
  duration_ms   INTEGER,
  attrs_json    TEXT                  -- 任意のJSON属性
);

-- クエリ高速化用インデックス
CREATE INDEX IF NOT EXISTS idx_logs_ts ON logs(ts_unix_ms);
CREATE INDEX IF NOT EXISTS idx_logs_level ON logs(level);
CREATE INDEX IF NOT EXISTS idx_logs_source ON logs(source);
CREATE INDEX IF NOT EXISTS idx_logs_trace ON logs(trace_id);

-- 全文検索用FTS5テーブル
CREATE VIRTUAL TABLE IF NOT EXISTS logs_fts USING fts5(
  message, trace_id, logger, route, attrs,
  content='logs', content_rowid='id'
);

-- FTS5自動同期トリガー
CREATE TRIGGER IF NOT EXISTS logs_ai AFTER INSERT ON logs BEGIN
  INSERT INTO logs_fts(rowid, message, trace_id, logger, route, attrs)
  VALUES (new.id, new.message, new.trace_id, new.logger, new.route, new.attrs_json);  -- attrs_jsonをFTS5のattrsカラムにマッピング
END;

PRAGMA journal_mode=WAL;
PRAGMA synchronous=NORMAL;
PRAGMA busy_timeout=5000;
PRAGMA wal_autocheckpoint=1000;

import time
import uuid
from contextvars import ContextVar
from starlette.requests import Request
from starlette.responses import Response

from .db_logger import insert_log  # SQLiteへの書き込み関数

# リクエストスコープでtrace_idを保持するcontextvar
_trace_id_var: ContextVar[str | None] = ContextVar("trace_id", default=None)

def get_trace_id() -> str | None:
    return _trace_id_var.get()

def set_trace_id(value: str | None) -> None:
    _trace_id_var.set(value)

def log_request(request: Request, response, *, started_ms: int) -> None:
    """リクエスト完了時のログを書き込む"""
    insert_log(
        ts_unix_ms=int(time.time() * 1000),
        level="INFO",
        source="backend",
        logger="http",
        message="request",
        trace_id=get_trace_id(),
        route=request.url.path,
        method=request.method,
        status_code=response.status_code,
        duration_ms=max(0, int(time.time() * 1000) - started_ms),
    )

async def trace_id_middleware(request: Request, call_next):
    # BFFから転送されたtrace_idがあればそれを使う
    trace_id = request.headers.get("x-trace-id") or uuid.uuid4().hex
    started_ms = int(time.time() * 1000)
    set_trace_id(trace_id)  # contextvarsに保存

    try:
        response = await call_next(request)
        response.headers["x-trace-id"] = trace_id

        # リクエスト完了ログ
        log_request(request, response, started_ms=started_ms)
        return response
    except Exception as e:
        # 例外発生時もログを残す
        insert_log(
            ts_unix_ms=int(time.time() * 1000),
            level="ERROR",
            source="backend",
            logger="http",
            message="request_exception",
            trace_id=trace_id,
            duration_ms=max(0, int(time.time() * 1000) - started_ms),
            attrs={"error": str(e)},
        )
        raise
    finally:
        set_trace_id(None)

import logging
import time

_LEVEL_MAP = {
    logging.DEBUG: "DEBUG",
    logging.INFO: "INFO",
    logging.WARNING: "WARN",
    logging.ERROR: "ERROR",
    logging.CRITICAL: "ERROR",
}

class SqliteLogHandler(logging.Handler):
    def __init__(self, *, source: str = "backend"):
        super().__init__()
        self.source = source

    def emit(self, record: logging.LogRecord) -> None:
        try:
            insert_log(
                ts_unix_ms=int(time.time() * 1000),
                level=_LEVEL_MAP.get(record.levelno, record.levelname),
                source=self.source,
                logger=record.name,
                message=record.getMessage(),
                trace_id=get_trace_id(),
                attrs=record.args if isinstance(record.args, dict) else None,
            )
        except Exception:
            self.handleError(record)  # ログ書き込み失敗でアプリを壊さない

import { insertLog } from "./logDb";

export function getOrCreateTraceId(req: NextRequest | { headers: Headers }) {
  return req.headers.get("x-trace-id") ?? crypto.randomUUID();
}

export async function logBff(
  level: "DEBUG" | "INFO" | "WARN" | "ERROR",
  message: string,
  opts: {
    traceId?: string;
    logger?: string;
    route?: string;
    method?: string;
    statusCode?: number;
    durationMs?: number;
    attrs?: Record<string, unknown>;
  } = {},
) {
  await insertLog({
    level,
    source: "bff",
    logger: opts.logger,
    message,
    trace_id: opts.traceId,
    request_id: opts.traceId,
    route: opts.route,
    method: opts.method,
    status_code: opts.statusCode,
    duration_ms: opts.durationMs,
    attrs: opts.attrs,
  });
}

type BffHandler = (req: NextRequest, ctx: unknown) => Promise<Response>;

function levelFromStatus(status: number) {
  if (status < 400) return "INFO" as const;
  if (status < 500) return "WARN" as const;
  return "ERROR" as const;
}

function isStreamingResponse(res: Response) {
  return res.headers.get("content-type")?.includes("text/event-stream") ?? false;
}

export function withBffLog(routeLabel: string, handler: BffHandler) {
  return async (req: NextRequest, ctx: unknown) => {
    const started = Date.now();
    const traceId = getOrCreateTraceId(req);

    try {
      const res = await handler(req, ctx);
      const durationMs = Date.now() - started;

      await logBff(levelFromStatus(res.status), "request", {
        traceId,
        logger: `bff:${routeLabel}`,
        route: req.nextUrl.pathname,
        method: req.method,
        statusCode: res.status,
        durationMs,
        attrs: { ok: res.ok, streaming: isStreamingResponse(res) },
      });

      // trace_idをレスポンスヘッダーに付与
      const headers = new Headers(res.headers);
      headers.set("x-trace-id", traceId);
      return new Response(res.body, { status: res.status, headers });
    } catch (e) {
      const msg = e instanceof Error ? e.message : String(e);
      await logBff("ERROR", "request_failed", {
        traceId,
        logger: `bff:${routeLabel}`,
        statusCode: 500,
        durationMs: Date.now() - started,
        attrs: { error: msg },
      });
      return NextResponse.json({ error: msg, traceId }, { status: 500 });
    }
  };
}

export const POST = withBffLog("chat-stream", async (req) => {
  // 通常のルートハンドラーロジック
  const res = await fetch("http://127.0.0.1:8765/v1/chat/stream", { ... });
  return new Response(res.body, { headers: { ... } });
});

// GET /api/logs?from=&to=&levels=&source=&q=&limit=&cursor=
export async function GET(req: NextRequest) {
  const params = req.nextUrl.searchParams;
  const result = await queryLogs({
    from: params.get("from") ? Number(params.get("from")) : undefined,
    to: params.get("to") ? Number(params.get("to")) : undefined,
    levels: params.get("levels")?.split(","),
    source: params.get("source"),
    q: params.get("q"),           // FTS5全文検索
    limit: Number(params.get("limit")) || 50,
    cursor: params.get("cursor"),  // カーソルページネーション
  });
  return NextResponse.json(result, {
    headers: { "cache-control": "no-store" },
  });
}

export async function pruneOldLogs() {
  const cutoffMs = Date.now() - 7 * 24 * 60 * 60 * 1000;
  db.prepare("DELETE FROM logs WHERE ts_unix_ms < @cutoffMs").run({ cutoffMs });
  db.exec("PRAGMA wal_checkpoint(TRUNCATE)");
  db.exec("VACUUM");
}

# Python: handleErrorで握りつぶす
except Exception:
    self.handleError(record)

// TypeScript: try-catchで握りつぶす
try {
  await insertLog({ ... });
} catch {
  // ログ書き込み失敗は無視
}

// ログ関連エンドポイントはwithBffLogで囲まない
// export const GET = withBffLog("logs", handler);  // NG
export async function GET(req: NextRequest) { ... }  // OK