podcastdistributiona/app/(app)/series/actions.ts

"use server";

import { revalidatePath } from "next/cache";
import { z } from "zod";
import type { Prisma } from "@prisma/client";
import { getServerSession } from "@/lib/auth/guards";
import { prisma } from "@/lib/db";
import { getEffectivePlan, subjectHasFeature } from "@/lib/billing/subscription";
import { reserveLimit, LimitExceededError } from "@/lib/usage/limits";
import { refundUsage } from "@/lib/usage/meter";
import { enqueueEpisodeGeneration } from "@/lib/queue/pgboss";
import type { UsageMetric } from "@/lib/billing/plans";
import { FORMAT_SPEAKERS } from "@/lib/episodes/options";
import { DEFAULT_VOICE_IDS, VOICE_CATALOG } from "@/lib/ai/voices";
import { isFlagEnabled } from "@/lib/flags";

const createSchema = z.object({
  theme: z.string().min(5).max(500),
  count: z.number().int().min(2).max(12),
  tone: z.string().min(1),
  audience: z.string().max(200).optional(),
  language: z.string().min(2).max(5),
});

export async function createSeriesAction(
  input: z.infer<typeof createSchema>
): Promise<{ ok: boolean; error?: string; seriesId?: string }> {
  const session = await getServerSession();
  if (!session) return { ok: false, error: "You must be signed in." };
  if (!(await subjectHasFeature(session.user.id, "series_generator", session.session.activeOrganizationId))) {
    return { ok: false, error: "The series generator requires the Pro plan." };
  }
  if (!(await isFlagEnabled("episode_generation_enabled"))) {
    return { ok: false, error: "Generation is temporarily paused. Please try again shortly." };
  }
  const parsed = createSchema.safeParse(input);
  if (!parsed.success) return { ok: false, error: parsed.error.issues[0]?.message ?? "Invalid input." };

  const { planSeason } = await import("@/lib/ai/series");
  const { plan } = await planSeason(parsed.data);

  const series = await prisma.series.create({
    data: {
      userId: session.user.id,
      organizationId: session.session.activeOrganizationId ?? undefined,
      title: plan.title,
      description: plan.description,
      plannedCount: plan.episodes.length,
      plan: plan.episodes as unknown as Prisma.InputJsonValue,
    },
  });
  revalidatePath("/series");
  return { ok: true, seriesId: series.id };
}

export async function generateFromSeriesAction(
  seriesId: string,
  index: number
): Promise<{ ok: boolean; error?: string; episodeId?: string }> {
  const session = await getServerSession();
  if (!session) return { ok: false, error: "You must be signed in." };

  if (!(await isFlagEnabled("episode_generation_enabled"))) {
    return { ok: false, error: "Episode generation is temporarily paused. Please try again shortly." };
  }

  // `index` is client-supplied and only TS-typed — validate it at runtime.
  if (!Number.isInteger(index) || index < 0) {
    return { ok: false, error: "Invalid episode index." };
  }

  const series = await prisma.series.findUnique({ where: { id: seriesId } });
  if (!series || (series.userId !== session.user.id && session.user.role !== "admin")) {
    return { ok: false, error: "Not allowed." };
  }

  const episodes = (series.plan as unknown as { title: string; topic: string; summary: string }[]) ?? [];
  const item = episodes[index];
  if (!item) return { ok: false, error: "Episode not found in plan." };

  // Bill the generation against the org that OWNS the series (the resource being
  // acted on), and stamp the new episode with that same org, so the billing
  // subject and the episode's organizationId are always consistent — regardless
  // of the caller's currently-active org. getEffectivePlan verifies membership of
  // series.organizationId internally and falls back to the user subject otherwise.
  const orgId = series.organizationId ?? undefined;
  const { subjectId, subjectType } = await getEffectivePlan(session.user.id, orgId);

  // Reserve quota atomically up front (a series generation consumes script +
  // audio). The worker won't re-meter; refund below if create/enqueue fails.
  const reserved: UsageMetric[] = [];
  const refundReserved = async () => {
    for (const m of reserved) await refundUsage(subjectId, subjectType, m);
  };
  try {
    await reserveLimit(session.user.id, "script", orgId);
    reserved.push("script");
    await reserveLimit(session.user.id, "audio", orgId);
    reserved.push("audio");
  } catch (err) {
    await refundReserved();
    if (err instanceof LimitExceededError) {
      return { ok: false, error: `Monthly ${err.check.metric} limit reached.` };
    }
    throw err;
  }

  const speakers = FORMAT_SPEAKERS.SOLO.map((s, i) => ({
    speakerKey: s.speakerKey,
    displayName: s.defaultName,
    elevenVoiceId: DEFAULT_VOICE_IDS[s.speakerKey] ?? VOICE_CATALOG[i % VOICE_CATALOG.length].id,
  }));

  try {
    const episode = await prisma.episode.create({
      data: {
        userId: session.user.id,
        organizationId: orgId,
        seriesId: series.id,
        title: item.title,
        topic: item.topic,
        tone: "Conversational",
        format: "SOLO",
        language: "en",
        targetLengthMin: 10,
        status: "QUEUED",
        stage: "Queued for generation",
        speakers: { create: speakers },
        jobs: { create: { type: "full", status: "queued" } },
      },
    });
    await enqueueEpisodeGeneration({ episodeId: episode.id, type: "full" });

    revalidatePath(`/series/${seriesId}`);
    return { ok: true, episodeId: episode.id };
  } catch (err) {
    await refundReserved();
    throw err;
  }
}