Function addToQueue

apps/webapp/app/trigger/utils/queue.ts:11–107

(
  body: z.infer<typeof IngestBodyRequest>,
  userId: string,
  workspaceId: string,
  activityId?: string,
  ingestionQueueId?: string,
)

// Used in the trigger
export const addToQueue = async (
  body: z.infer<typeof IngestBodyRequest>,
  userId: string,
  workspaceId: string,
  activityId?: string,
  ingestionQueueId?: string,
) => {
  if (!workspaceId) {
    throw new Error(
      "Workspace ID is required to create an ingestion queue entry.",
    );
  }

  // Check if workspace has sufficient credits before processing
  const hasSufficientCredits = await hasCredits(
    workspaceId,
    userId,
    "addEpisode",
  );

  if (!hasSufficientCredits) {
    throw new Error("no credits");
  }

  let labels: string[] = body.labelIds ?? [];

  if (body.sessionId) {
    const lastEpisode = await prisma.ingestionQueue.findFirst({
      where: {
        sessionId: body.sessionId,
      },
      orderBy: {
        createdAt: "desc",
      },
    });

    if (lastEpisode?.labels && lastEpisode?.labels.length > 0) {
      labels = lastEpisode?.labels;
    }
  }

  // Validate label access if labelIds are provided
  if (body.labelIds && body.labelIds.length > 0) {
    const labelService = new LabelService();
    const hasAccess = await labelService.validateLabelAccess(
      body.labelIds,
      workspaceId,
    );

    if (!hasAccess) {
      throw new Error(
        "One or more labels are invalid or not accessible in this workspace",
      );
    }
  }

  // Upsert: update existing or create new ingestion queue entry
  const queuePersist = await prisma.ingestionQueue.upsert({