erika/lib/video/video-upload-storage.ts

import "server-only";

import { mkdir, readdir, stat, writeFile } from "node:fs/promises";
import { randomUUID } from "node:crypto";
import { sse } from "@/lib/realtime/sse";
import { db } from "@/db";
import {
	videoUploadJobs,
	videoUploads,
	type TiktokSettings,
	type YoutubeSettings,
} from "@/db/schema";
import {
	validateThumbnailFile,
	validateVideoFile,
	publishVideoJob,
} from "@/lib/video/video-platforms";
import { eq } from "drizzle-orm";

const VIDEO_DIR = process.env.VIDEO_UPLOAD_DIR || "/nfs/erika";

export async function saveVideoUpload(formData: FormData, discordId: string) {
	const video = formData.get("video");
	const stagedVideoId = String(formData.get("stagedVideoId") ?? "");
	const thumbnail = formData.get("thumbnail");

	let videoPath: string;
	let originalFilename: string;
	let mimeType: string;
	let videoSize: number;

	if (stagedVideoId) {
		if (!/^[0-9a-f-]{36}$/i.test(stagedVideoId)) {
			throw new Error("Invalid staged video.");
		}
		const stagingDirectory = `${VIDEO_DIR}/staging/${discordId}`;
		const stagedFile = (await readdir(stagingDirectory)).find((name) =>
			name.startsWith(`${stagedVideoId}-`),
		);
		if (!stagedFile) throw new Error("Uploaded video was not found.");
		videoPath = `${stagingDirectory}/${stagedFile}`;
		originalFilename = String(formData.get("videoName") ?? stagedFile);
		mimeType = String(formData.get("videoMimeType") ?? "");
		videoSize = (await stat(videoPath)).size;
		if (!["video/mp4", "video/quicktime", "video/webm"].includes(mimeType)) {
			throw new Error("Use an MP4, MOV, or WebM video.");
		}
	} else if (video instanceof File) {
		validateVideoFile(video);
		await mkdir(VIDEO_DIR, { recursive: true });
		const temporaryId = randomUUID();
		const safeVideoName = video.name.replace(/[^a-zA-Z0-9._-]/g, "_");
		videoPath = `${VIDEO_DIR}/${temporaryId}-${safeVideoName}`;
		await writeFile(videoPath, Buffer.from(await video.arrayBuffer()));
		originalFilename = video.name;
		mimeType = video.type;
		videoSize = video.size;
	} else {
		throw new Error("Choose and finish uploading a video first.");
	}

	if (thumbnail instanceof File && thumbnail.size > 0)
		validateThumbnailFile(thumbnail);
	const title = String(formData.get("title") ?? "").trim();
	if (!title) throw new Error("Title is required.");
	const scheduledValue = String(formData.get("scheduledAt") ?? "").trim();
	const scheduledAt = scheduledValue ? new Date(scheduledValue) : null;
	if (scheduledAt && Number.isNaN(scheduledAt.getTime()))
		throw new Error("Invalid schedule date.");
	if (scheduledAt && scheduledAt <= new Date())
		throw new Error("Schedule must be in the future.");

	const id = randomUUID();
	let thumbnailPath: string | null = null;
	if (thumbnail instanceof File && thumbnail.size > 0) {
		thumbnailPath = `${VIDEO_DIR}/${id}-thumbnail-${thumbnail.name.replace(/[^a-zA-Z0-9._-]/g, "_")}`;
		await writeFile(thumbnailPath, Buffer.from(await thumbnail.arrayBuffer()));
	}

	const youtubeSettings: YoutubeSettings = JSON.parse(
		String(formData.get("youtubeSettings") || "{}"),
	);
	const tiktokSettings: TiktokSettings = JSON.parse(
		String(formData.get("tiktokSettings") || "{}"),
	);
	const [upload] = await db
		.insert(videoUploads)
		.values({
			id,
			title,
			description: String(formData.get("description") ?? ""),
			videoPath,
			thumbnailPath,
			originalFilename,
			mimeType,
			size: videoSize,
			scheduledAt,
			status: scheduledAt ? "scheduled" : "processing",
			youtubeSettings,
			tiktokSettings,
			createdByDiscordId: discordId,
		})
		.returning({ id: videoUploads.id });
	const jobs = await db
		.insert(videoUploadJobs)
		.values([
			{ uploadId: upload.id, platform: "youtube" },
			{ uploadId: upload.id, platform: "tiktok" },
		])
		.returning({ id: videoUploadJobs.id });
	await sse.uploads.pub("update", {
		uploadId: upload.id,
		status: scheduledAt ? "scheduled" : "processing",
	});
	if (!scheduledAt) {
		await Promise.allSettled(jobs.map((job) => publishVideoJob(job.id)));
		const remaining = await db.query.videoUploadJobs.findMany({
			where: (table, { eq, and, ne }) =>
				and(eq(table.uploadId, upload.id), ne(table.status, "completed")),
		});
		await db
			.update(videoUploads)
			.set({
				status: remaining.length === 0 ? "completed" : "partial",
				updatedAt: new Date(),
			})
			.where(eq(videoUploads.id, upload.id));
		await sse.uploads.pub("update", {
			uploadId: upload.id,
			status: remaining.length === 0 ? "completed" : "partial",
		});
	}
	return upload.id;
}