feat(ai): pipeline assincrono — transcricao, analise, prompts (fase 20)

Sub-fase B do modulo de IA: novo servico apps/ai-worker (poll + FOR UPDATE
SKIP LOCKED) processa AIJob de TRANSCRIPTION/ANALYSIS disparados
automaticamente apos uma gravacao ficar disponivel, respeitando a cascata
de privacidade Tenant>Queue>Campaign e o entitlement do Plan. Transcricao
separa o WAV estereo em 2 canais (parser proprio, sem ffmpeg) e transcreve
cada perna independente; analise sempre redige dados sensiveis antes de
sair pro provider e valida o resultado contra o schema antes de persistir.
CRUD de AIPromptTemplate/AIPromptVersion em apps/api.

Testado ponta a ponta contra o worker real em Docker e Postgres real com
RLS (cascata de privacidade em 3 cenarios, WAV sintetico real no object
storage, claim/retry/dead-letter reais) — chamada de rede contra
OpenAI/Anthropic continua nunca exercitada (mesma restricao de rede desde
o Provider Layer). Detalhes em docs/AI_PIPELINE.md.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01X1HxY46WGU4G1zmVDNKcWw

apps/ai-worker/src/claim.ts

@@ -0,0 +1,41 @@
+import type { Prisma } from "@b2bcall/database";
+
+export interface ClaimedJob {
+  id: string;
+  callId: string;
+  type: "TRANSCRIPTION" | "ANALYSIS" | "REANALYSIS";
+  attemptCount: number;
+  maxAttempts: number;
+}
+
+/**
+ * Reserva atômica de AIJob pendentes (mesmo padrão de
+ * apps/predictive-dialer/src/lead-reservation.ts § `FOR UPDATE SKIP
+ * LOCKED`): dois workers concorrentes nunca processam o mesmo job. Precisa
+ * rodar dentro do MESMO `withTenantContext` transaction — o lock de linha
+ * só vale até o commit/rollback.
+ */
+export async function claimPendingJobs(
+  tx: Prisma.TransactionClient,
+  tenantId: string,
+  limit: number,
+): Promise<ClaimedJob[]> {
+  const rows = await tx.$queryRaw<ClaimedJob[]>`
+    SELECT id, call_id AS "callId", type, attempt_count AS "attemptCount", max_attempts AS "maxAttempts"
+    FROM ai_jobs
+    WHERE tenant_id = ${tenantId}::uuid
+      AND status IN ('PENDING', 'RETRYING')
+      AND scheduled_at <= now()
+    ORDER BY scheduled_at ASC
+    LIMIT ${limit}
+    FOR UPDATE SKIP LOCKED
+  `;
+  if (rows.length === 0) return [];
+
+  await tx.aIJob.updateMany({
+    where: { id: { in: rows.map((r) => r.id) } },
+    data: { status: "PROCESSING" },
+  });
+
+  return rows;
+}

apps/ai-worker/src/job-outcome.ts

@@ -0,0 +1,50 @@
+import { getPrismaClient, withTenantContext } from "@b2bcall/database";
+import { computeNextScheduledAt, isDeadLetter } from "@b2bcall/ai";
+import { createLogger } from "@b2bcall/shared";
+
+const logger = createLogger("b2bcall-ai-worker");
+
+export async function markJobCompleted(tenantId: string, jobId: string): Promise<void> {
+  const prisma = getPrismaClient();
+  await withTenantContext(prisma, tenantId, (tx) =>
+    tx.aIJob.update({ where: { id: jobId }, data: { status: "COMPLETED", completedAt: new Date() } }),
+  );
+}
+
+/**
+ * Retry/dead-letter (agente.md secao 108): nunca retry infinito.
+ * `attemptCount` já foi incrementado ANTES de chamar isto — quem processa
+ * o job soma 1 na entrada, chama o handler, e só chega aqui se o handler
+ * lançou.
+ */
+export async function markJobFailed(
+  tenantId: string,
+  jobId: string,
+  attemptCount: number,
+  maxAttempts: number,
+  error: unknown,
+): Promise<void> {
+  const prisma = getPrismaClient();
+  const message = error instanceof Error ? error.message : String(error);
+  const deadLetter = isDeadLetter(attemptCount, maxAttempts);
+
+  await withTenantContext(prisma, tenantId, (tx) =>
+    tx.aIJob.update({
+      where: { id: jobId },
+      data: deadLetter
+        ? { status: "FAILED", lastError: message, attemptCount }
+        : {
+            status: "RETRYING",
+            lastError: message,
+            attemptCount,
+            scheduledAt: computeNextScheduledAt(attemptCount),
+          },
+    }),
+  );
+
+  if (deadLetter) {
+    logger.error("job de IA foi pra dead-letter (tentativas esgotadas)", { jobId, attemptCount, error: message });
+  } else {
+    logger.warn("job de IA falhou, agendado retry com backoff", { jobId, attemptCount, error: message });
+  }
+}

apps/ai-worker/src/main.ts

@@ -0,0 +1,81 @@
+import { getPrismaClient, withTenantContext } from "@b2bcall/database";
+import { createLogger } from "@b2bcall/shared";
+import { claimPendingJobs, type ClaimedJob } from "./claim";
+import { processTranscriptionJob } from "./process-transcription";
+import { processAnalysisJob } from "./process-analysis";
+import { markJobCompleted, markJobFailed } from "./job-outcome";
+
+const logger = createLogger("b2bcall-ai-worker");
+
+const TICK_INTERVAL_MS = Number(process.env.AI_WORKER_TICK_INTERVAL_MS ?? 5000);
+const JOBS_PER_TENANT_PER_TICK = Number(process.env.AI_WORKER_JOBS_PER_TICK ?? 5);
+
+async function processJob(tenantId: string, job: ClaimedJob): Promise<void> {
+  const attemptCount = job.attemptCount + 1;
+  try {
+    if (job.type === "TRANSCRIPTION") {
+      await processTranscriptionJob(tenantId, job.callId);
+    } else if (job.type === "ANALYSIS" || job.type === "REANALYSIS") {
+      await processAnalysisJob(tenantId, job.callId);
+    } else {
+      throw new Error(`Tipo de AIJob desconhecido: ${job.type}`);
+    }
+    await markJobCompleted(tenantId, job.id);
+  } catch (err) {
+    await markJobFailed(tenantId, job.id, attemptCount, job.maxAttempts, err);
+  }
+}
+
+/**
+ * Um tick = uma passada por todos os tenants ativos, reservando até
+ * `JOBS_PER_TENANT_PER_TICK` jobs pendentes de cada um (secao 106-108:
+ * "pipeline sempre assíncrono, nunca bloqueia a chamada esperando IA").
+ * Jobs de tenants diferentes processam em paralelo; dentro de um mesmo
+ * tenant, sequencial (evita disparar muitas chamadas simultâneas contra o
+ * mesmo provider/credencial).
+ */
+async function runTick(): Promise<void> {
+  const prisma = getPrismaClient();
+  const tenants = await prisma.tenant.findMany({ where: { status: "ACTIVE" } });
+
+  await Promise.all(
+    tenants.map(async (tenant) => {
+      const claimed = await withTenantContext(prisma, tenant.id, (tx) =>
+        claimPendingJobs(tx, tenant.id, JOBS_PER_TENANT_PER_TICK),
+      );
+      for (const job of claimed) {
+        await processJob(tenant.id, job);
+      }
+    }),
+  );
+}
+
+async function main(): Promise<void> {
+  let stopped = false;
+
+  const tick = async () => {
+    if (stopped) return;
+    try {
+      await runTick();
+    } catch (err) {
+      logger.error("falha no tick do ai-worker", { error: String(err) });
+    }
+    if (!stopped) setTimeout(tick, TICK_INTERVAL_MS);
+  };
+  setTimeout(tick, TICK_INTERVAL_MS);
+
+  logger.info("b2bcall-ai-worker iniciado", { tickIntervalMs: TICK_INTERVAL_MS });
+
+  const shutdown = () => {
+    stopped = true;
+    logger.info("encerrando b2bcall-ai-worker");
+    process.exit(0);
+  };
+  process.on("SIGTERM", shutdown);
+  process.on("SIGINT", shutdown);
+}
+
+main().catch((err) => {
+  logger.error("falha fatal ao iniciar b2bcall-ai-worker", { error: String(err) });
+  process.exit(1);
+});

apps/ai-worker/src/privacy-lookup.ts

@@ -0,0 +1,37 @@
+import type { Prisma, Tenant } from "@b2bcall/database";
+import { resolvePrivacyLevel, type AIPrivacyLevelName } from "@b2bcall/ai";
+
+/**
+ * Mesma cascata Tenant > Queue > Campaign usada em
+ * apps/freeswitch-events/src/ai-trigger.ts pra decidir o job de
+ * TRANSCRIPTION — reconsultada aqui pra decidir se, depois de transcrever,
+ * a chamada também pode virar um job de ANALYSIS. Recalculada (não lida de
+ * um valor salvo) porque a config pode ter mudado entre os dois jobs.
+ */
+export async function resolveEffectivePrivacy(
+  tx: Prisma.TransactionClient,
+  tenant: Tenant,
+  queueId: string | null,
+  campaignId: string | null,
+): Promise<AIPrivacyLevelName> {
+  let queueLevel: AIPrivacyLevelName | null = null;
+  let campaignTranscriptionEnabled = false;
+  let campaignAnalysisEnabled = false;
+
+  if (queueId) {
+    const queue = await tx.queue.findFirst({ where: { id: queueId } });
+    queueLevel = (queue?.aiPrivacyLevel as AIPrivacyLevelName | null) ?? null;
+  }
+  if (campaignId) {
+    const campaign = await tx.campaign.findFirst({ where: { id: campaignId } });
+    campaignTranscriptionEnabled = campaign?.aiTranscriptionEnabled ?? false;
+    campaignAnalysisEnabled = campaign?.aiAnalysisEnabled ?? false;
+  }
+
+  return resolvePrivacyLevel({
+    tenantLevel: tenant.aiPrivacyLevel as AIPrivacyLevelName,
+    queueLevel,
+    campaignTranscriptionEnabled,
+    campaignAnalysisEnabled,
+  });
+}

apps/ai-worker/src/process-analysis.ts

@@ -0,0 +1,134 @@
+import { getPrismaClient, withTenantContext, type Prisma } from "@b2bcall/database";
+import {
+  SensitiveDataRedactor,
+  CALL_ANALYSIS_JSON_SCHEMA,
+  validateCallAnalysisResult,
+} from "@b2bcall/ai";
+import { createLogger } from "@b2bcall/shared";
+import { resolveProviderForCapability } from "./provider-resolution";
+
+const logger = createLogger("b2bcall-ai-worker");
+const redactor = new SensitiveDataRedactor();
+
+/**
+ * Resolução de template (agente.md secao 114-116): `Campaign.
+ * analysisPromptTemplateId` sobrescreve quando setado; senão o primeiro
+ * `AIPromptTemplate` de purpose=ANALYSIS com versão ativa, preferindo um
+ * template do próprio tenant sobre o padrão GLOBAL da plataforma.
+ */
+async function resolvePromptContent(
+  tx: Prisma.TransactionClient,
+  tenantId: string,
+  campaignId: string | null,
+): Promise<string | null> {
+  if (campaignId) {
+    const campaign = await tx.campaign.findFirst({ where: { id: campaignId } });
+    if (campaign?.analysisPromptTemplateId) {
+      const template = await tx.aIPromptTemplate.findFirst({
+        where: { id: campaign.analysisPromptTemplateId },
+        include: { activeVersion: true },
+      });
+      if (template?.activeVersion) return template.activeVersion.content;
+    }
+  }
+
+  const templates = await tx.aIPromptTemplate.findMany({
+    where: { purpose: "ANALYSIS" },
+    include: { activeVersion: true },
+  });
+  const tenantSpecific = templates.find((t) => t.tenantId === tenantId && t.activeVersion);
+  const global = templates.find((t) => t.tenantId === null && t.activeVersion);
+  return (tenantSpecific ?? global)?.activeVersion?.content ?? null;
+}
+
+export async function processAnalysisJob(tenantId: string, callId: string): Promise<void> {
+  const prisma = getPrismaClient();
+
+  const { call, transcription } = await withTenantContext(prisma, tenantId, async (tx) => {
+    const call = await tx.call.findUniqueOrThrow({ where: { id: callId } });
+    const transcription = await tx.callTranscription.findFirst({
+      where: { callId, status: "COMPLETED" },
+      orderBy: { createdAt: "desc" },
+    });
+    return { call, transcription };
+  });
+
+  if (!transcription?.text) {
+    throw new Error(`Nenhuma transcricao concluida encontrada pra chamada ${callId}`);
+  }
+
+  const promptContent = await withTenantContext(prisma, tenantId, (tx) =>
+    resolvePromptContent(tx, tenantId, call.campaignId),
+  );
+  if (!promptContent) {
+    throw new Error("Nenhum AIPromptTemplate (purpose=ANALYSIS) com versao ativa configurado pra este tenant");
+  }
+
+  const resolved = await withTenantContext(prisma, tenantId, (tx) =>
+    resolveProviderForCapability(tx, tenantId, "STRUCTURED_OUTPUT"),
+  );
+  if (!resolved) {
+    throw new Error("Nenhum provider/modelo com capability STRUCTURED_OUTPUT habilitado pra este tenant");
+  }
+  if (!resolved.instance.analyze) {
+    throw new Error(`Provider ${resolved.providerId} nao implementa analyze()`);
+  }
+
+  // Secao 123: nunca manda texto sem passar pelo redactor antes de sair
+  // pra um provider terceiro, independente do nivel de privacidade (que só
+  // controla SE roda, não O QUE é enviado quando roda).
+  const redactedText = redactor.redact(transcription.text);
+
+  const result = await resolved.instance.analyze({
+    transcriptText: redactedText,
+    promptContent,
+    jsonSchema: CALL_ANALYSIS_JSON_SCHEMA,
+  });
+
+  const validated = validateCallAnalysisResult(result.data);
+
+  await withTenantContext(prisma, tenantId, async (tx) => {
+    await tx.callAIAnalysis.create({
+      data: {
+        tenantId,
+        callId,
+        providerId: resolved.providerId,
+        model: resolved.externalModelId,
+        summary: validated.summary,
+        customerIntent: validated.customerIntent,
+        outcome: validated.outcome,
+        sentiment: validated.sentiment,
+        topics: validated.topics,
+        keywords: validated.keywords,
+        objections: validated.objections,
+        questions: validated.questions,
+        actionItems: validated.actionItems,
+        complianceFlags: validated.complianceFlags,
+        riskFlags: validated.riskFlags,
+        qualityScore: validated.qualityScore,
+        agentScore: validated.agentScore,
+        customerSentimentScore: validated.customerSentimentScore,
+        salesOpportunity: validated.salesOpportunity,
+        nextBestAction: validated.nextBestAction,
+      },
+    });
+
+    const usageRows = [
+      { type: "AI_ANALYSIS_REQUEST" as const, quantity: 1 },
+      ...(result.inputTokens ? [{ type: "AI_INPUT_TOKENS" as const, quantity: result.inputTokens }] : []),
+      ...(result.outputTokens ? [{ type: "AI_OUTPUT_TOKENS" as const, quantity: result.outputTokens }] : []),
+    ];
+    await tx.aIUsageRecord.createMany({
+      data: usageRows.map((row) => ({
+        tenantId,
+        callId,
+        type: row.type,
+        quantity: row.quantity,
+        providerId: resolved.providerId,
+        model: resolved.externalModelId,
+      })),
+    });
+  });
+
+  logger.info("analise de IA concluida", { callId, qualityScore: validated.qualityScore });
+}

apps/ai-worker/src/process-transcription.ts

@@ -0,0 +1,139 @@
+import { join } from "node:path";
+import { getPrismaClient, withTenantContext } from "@b2bcall/database";
+import { getObjectStorageProvider } from "@b2bcall/storage";
+import { splitStereoWav, allowsAnalysis } from "@b2bcall/ai";
+import { createLogger } from "@b2bcall/shared";
+import { createJobWorkDir, cleanupJobWorkDir, streamToFile } from "./temp-files";
+import { resolveProviderForCapability } from "./provider-resolution";
+import { resolveEffectivePrivacy } from "./privacy-lookup";
+import { writeFile } from "node:fs/promises";
+
+const logger = createLogger("b2bcall-ai-worker");
+
+/**
+ * Convenção de canal (agente.md secao 111, ver packages/ai/src/
+ * wav-stereo-split.ts): canal 0 = perna onde `record_session` foi chamado
+ * (member/"cliente" no discador), canal 1 = perna bridgeada (agente). Não
+ * confirmada contra áudio real distinguível nesta sessão.
+ */
+export async function processTranscriptionJob(tenantId: string, callId: string): Promise<void> {
+  const prisma = getPrismaClient();
+  const workDir = await createJobWorkDir();
+
+  try {
+    const { call, tenant, recording } = await withTenantContext(prisma, tenantId, async (tx) => {
+      const call = await tx.call.findUniqueOrThrow({ where: { id: callId } });
+      const tenant = await tx.tenant.findUniqueOrThrow({ where: { id: tenantId } });
+      const recording = await tx.recording.findUnique({ where: { callId } });
+      return { call, tenant, recording };
+    });
+
+    if (!recording || recording.status !== "AVAILABLE") {
+      throw new Error(`Gravacao indisponivel pra chamada ${callId} (status: ${recording?.status ?? "inexistente"})`);
+    }
+
+    const storage = getObjectStorageProvider();
+    const stereoPath = join(workDir, "stereo.wav");
+    const stream = await storage.getObjectStream(recording.objectKey);
+    const stereoBuffer = await streamToFile(stream, stereoPath);
+
+    const { channel0, channel1, sampleRate } = splitStereoWav(stereoBuffer);
+    const ch0Path = join(workDir, "channel0-customer.wav");
+    const ch1Path = join(workDir, "channel1-agent.wav");
+    await writeFile(ch0Path, channel0);
+    await writeFile(ch1Path, channel1);
+
+    const resolved = await withTenantContext(prisma, tenantId, (tx) =>
+      resolveProviderForCapability(tx, tenantId, "TRANSCRIPTION"),
+    );
+    if (!resolved) {
+      throw new Error("Nenhum provider/modelo com capability TRANSCRIPTION habilitado pra este tenant");
+    }
+    if (!resolved.instance.transcribe) {
+      throw new Error(`Provider ${resolved.providerId} nao implementa transcribe()`);
+    }
+
+    const [customerResult, agentResult] = await Promise.all([
+      resolved.instance.transcribe({ audioFilePath: ch0Path }),
+      resolved.instance.transcribe({ audioFilePath: ch1Path }),
+    ]);
+
+    const combinedText = [
+      customerResult.text ? `[CUSTOMER]\n${customerResult.text}` : null,
+      agentResult.text ? `[AGENT]\n${agentResult.text}` : null,
+    ]
+      .filter(Boolean)
+      .join("\n\n");
+
+    const durationSeconds =
+      (customerResult.durationSeconds ?? 0) || (agentResult.durationSeconds ?? 0) || recording.durationSeconds || 0;
+
+    await withTenantContext(prisma, tenantId, async (tx) => {
+      const transcription = await tx.callTranscription.create({
+        data: {
+          tenantId,
+          callId,
+          providerId: resolved.providerId,
+          model: resolved.externalModelId,
+          language: customerResult.language ?? agentResult.language,
+          text: combinedText,
+          status: "COMPLETED",
+          durationSeconds,
+          providerRequestId: [customerResult.providerRequestId, agentResult.providerRequestId]
+            .filter(Boolean)
+            .join(","),
+          inputUsage: (customerResult.inputUsage ?? 0) + (agentResult.inputUsage ?? 0),
+          outputUsage: (customerResult.outputUsage ?? 0) + (agentResult.outputUsage ?? 0),
+        },
+      });
+
+      const segmentsFor = (result: typeof customerResult, speaker: "CUSTOMER" | "AGENT") =>
+        (result.segments && result.segments.length > 0
+          ? result.segments
+          : result.text
+            ? [{ startMs: 0, endMs: (result.durationSeconds ?? 0) * 1000, text: result.text }]
+            : []
+        ).map((seg) => ({
+          tenantId,
+          transcriptionId: transcription.id,
+          speaker,
+          startMs: seg.startMs,
+          endMs: seg.endMs,
+          text: seg.text,
+          confidence: seg.confidence,
+        }));
+
+      const segments = [...segmentsFor(customerResult, "CUSTOMER"), ...segmentsFor(agentResult, "AGENT")];
+      if (segments.length > 0) {
+        await tx.callTranscriptSegment.createMany({ data: segments });
+      }
+
+      if (durationSeconds > 0) {
+        await tx.aIUsageRecord.create({
+          data: {
+            tenantId,
+            callId,
+            type: "AI_TRANSCRIPTION_SECONDS",
+            quantity: durationSeconds,
+            providerId: resolved.providerId,
+            model: resolved.externalModelId,
+          },
+        });
+      }
+    });
+
+    logger.info("transcricao concluida", { callId, durationSeconds, sampleRate });
+
+    const effectiveLevel = await withTenantContext(prisma, tenantId, (tx) =>
+      resolveEffectivePrivacy(tx, tenant, call.queueId, call.campaignId),
+    );
+    if (allowsAnalysis(effectiveLevel)) {
+      await withTenantContext(prisma, tenantId, (tx) =>
+        tx.aIJob.create({ data: { tenantId, callId, type: "ANALYSIS" } }),
+      );
+      logger.info("job de analise de IA encadeado", { callId });
+    }
+  } finally {
+    await cleanupJobWorkDir(workDir);
+  }
+}

apps/ai-worker/src/provider-resolution.ts

@@ -0,0 +1,56 @@
+import type { Prisma, AICapability } from "@b2bcall/database";
+import { decryptSecret } from "@b2bcall/shared";
+import { createAIProvider, type AIProvider } from "@b2bcall/ai";
+
+export interface ResolvedProvider {
+  providerId: string;
+  modelId: string;
+  externalModelId: string;
+  instance: AIProvider;
+}
+
+/**
+ * Resolução de "qual provider/modelo usar pra este job" — agente.md não
+ * especifica um algoritmo de seleção (só que BYOK deve poder coexistir com
+ * GLOBAL, secao 100-101). Política adotada, documentada em
+ * docs/AI_PIPELINE.md: prefere BYOK do próprio tenant (custo e
+ * credenciais do próprio tenant) sobre o provider GLOBAL da plataforma;
+ * dentro de cada grupo, o primeiro provider habilitado com um AIModel
+ * habilitado que tenha a capability pedida, por `createdAt` (determinístico,
+ * sem heurística de custo/qualidade ainda). Revisar se um dia precisar de
+ * uma política mais rica (ex.: menor custo, fallback em cadeia).
+ */
+export async function resolveProviderForCapability(
+  tx: Prisma.TransactionClient,
+  tenantId: string,
+  capability: AICapability,
+): Promise<ResolvedProvider | null> {
+  const candidates = await tx.aIModel.findMany({
+    where: {
+      enabled: true,
+      capabilities: { has: capability },
+      provider: { enabled: true },
+    },
+    include: { provider: true },
+    orderBy: { createdAt: "asc" },
+  });
+  if (candidates.length === 0) return null;
+
+  const tenantOwned = candidates.filter((c) => c.provider.tenantId === tenantId);
+  const chosen = tenantOwned[0] ?? candidates.find((c) => c.provider.tenantId === null);
+  if (!chosen) return null;
+
+  const credentials = {
+    apiKey: decryptSecret(chosen.provider.encryptedApiKey),
+    baseUrl: chosen.provider.baseUrl ?? undefined,
+    organization: chosen.provider.organization ?? undefined,
+    project: chosen.provider.project ?? undefined,
+  };
+
+  return {
+    providerId: chosen.provider.id,
+    modelId: chosen.id,
+    externalModelId: chosen.externalModelId,
+    instance: createAIProvider(chosen.provider.providerType, credentials),
+  };
+}

apps/ai-worker/src/temp-files.ts

@@ -0,0 +1,30 @@
+import { randomUUID } from "node:crypto";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { mkdir, rm, writeFile } from "node:fs/promises";
+import type { Readable } from "node:stream";
+
+const WORK_ROOT = join(tmpdir(), "b2bcall-ai-worker");
+
+/** Um diretório de escopo por job — tudo dentro dele some junto no
+ * `cleanup`, nunca deixa arquivo órfão em `/tmp` se o processamento
+ * lançar no meio. */
+export async function createJobWorkDir(): Promise<string> {
+  const dir = join(WORK_ROOT, randomUUID());
+  await mkdir(dir, { recursive: true });
+  return dir;
+}
+
+export async function cleanupJobWorkDir(dir: string): Promise<void> {
+  await rm(dir, { recursive: true, force: true });
+}
+
+export async function streamToFile(stream: Readable, destPath: string): Promise<Buffer> {
+  const chunks: Buffer[] = [];
+  for await (const chunk of stream) {
+    chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
+  }
+  const buffer = Buffer.concat(chunks);
+  await writeFile(destPath, buffer);
+  return buffer;
+}