videofolxtv/server/videoUpload.ts

import {
  S3Client,
  CreateMultipartUploadCommand,
  UploadPartCommand,
  CompleteMultipartUploadCommand,
  AbortMultipartUploadCommand,
  ListObjectsV2Command,
  HeadObjectCommand,
  DeleteObjectCommand,
  PutObjectCommand,
} from "@aws-sdk/client-s3";
import { getSignedUrl } from "@aws-sdk/s3-request-presigner";

const S3_ENDPOINT = process.env.S3_ENDPOINT || "https://fsn1.your-objectstorage.com";
const S3_BUCKET = process.env.S3_BUCKET || "folxvideos-pub";
const S3_PUBLIC_BASE = process.env.S3_PUBLIC_BASE || "https://folxvideos.b-cdn.net";

// Prefix where finished HLS packages live: folx-tv/{guid}/master.m3u8
export const FOLX_PREFIX = "folx-tv/";
// Staging prefix the transcode worker polls: folx-tv/_incoming/{guid}/original.<ext>
export const INCOMING_PREFIX = "folx-tv/_incoming/";

// 64 MB parts -> 10 000 parts max = 640 GB ceiling, plenty for our masters.
export const PART_SIZE = 64 * 1024 * 1024;

const s3 = new S3Client({
  region: process.env.S3_REGION || "fsn1",
  endpoint: S3_ENDPOINT,
  forcePathStyle: true,
  credentials: {
    accessKeyId: process.env.S3_ACCESS_KEY_ID || "",
    secretAccessKey: process.env.S3_SECRET_ACCESS_KEY || "",
  },
});

export interface InitResult {
  guid: string;
  key: string;
  uploadId: string;
  partSize: number;
  partCount: number;
}

function safeExt(fileName: string): string {
  const m = /\.([a-zA-Z0-9]{1,5})$/.exec(fileName || "");
  const ext = (m ? m[1] : "mp4").toLowerCase();
  const allowed = ["mp4", "mov", "mkv", "m4v", "avi", "mxf", "mpg", "mpeg", "ts", "webm"];
  return allowed.includes(ext) ? ext : "mp4";
}

export function publicUrlFor(guid: string): string {
  return `${S3_PUBLIC_BASE}/${FOLX_PREFIX}${guid}/master.m3u8`;
}

/** Start a multipart upload for the original master file. */
export async function initUpload(
  guid: string,
  fileName: string,
  fileSize: number,
  contentType?: string,
): Promise<InitResult> {
  const ext = safeExt(fileName);
  const key = `${INCOMING_PREFIX}${guid}/original.${ext}`;
  const cmd = new CreateMultipartUploadCommand({
    Bucket: S3_BUCKET,
    Key: key,
    ContentType: contentType || "video/mp4",
    ACL: "public-read",
  });
  const res = await s3.send(cmd);
  if (!res.UploadId) throw new Error("S3 did not return an UploadId");
  return {
    guid,
    key,
    uploadId: res.UploadId,
    partSize: PART_SIZE,
    partCount: Math.max(1, Math.ceil(fileSize / PART_SIZE)),
  };
}

/** Presigned PUT URL for one part. Valid 6 h so slow lines can finish. */
export async function partUrl(key: string, uploadId: string, partNumber: number): Promise<string> {
  const cmd = new UploadPartCommand({
    Bucket: S3_BUCKET,
    Key: key,
    UploadId: uploadId,
    PartNumber: partNumber,
  });
  return getSignedUrl(s3, cmd, { expiresIn: 6 * 3600 });
}

export async function completeUpload(
  key: string,
  uploadId: string,
  parts: Array<{ PartNumber: number; ETag: string }>,
): Promise<void> {
  const ordered = [...parts].sort((a, b) => a.PartNumber - b.PartNumber);
  await s3.send(
    new CompleteMultipartUploadCommand({
      Bucket: S3_BUCKET,
      Key: key,
      UploadId: uploadId,
      MultipartUpload: { Parts: ordered },
    }),
  );
}

export async function abortUpload(key: string, uploadId: string): Promise<void> {
  try {
    await s3.send(new AbortMultipartUploadCommand({ Bucket: S3_BUCKET, Key: key, UploadId: uploadId }));
  } catch {
    /* best effort */
  }
}

/** Drop a small JSON sidecar next to the original so the worker knows the metadata. */
export async function writeJobFile(guid: string, job: Record<string, any>): Promise<void> {
  await s3.send(
    new PutObjectCommand({
      Bucket: S3_BUCKET,
      Key: `${INCOMING_PREFIX}${guid}/job.json`,
      Body: JSON.stringify(job, null, 2),
      ContentType: "application/json",
      ACL: "public-read",
    }),
  );
}

export async function objectExists(key: string): Promise<boolean> {
  try {
    await s3.send(new HeadObjectCommand({ Bucket: S3_BUCKET, Key: key }));
    return true;
  } catch {
    return false;
  }
}

export async function deleteObject(key: string): Promise<void> {
  try {
    await s3.send(new DeleteObjectCommand({ Bucket: S3_BUCKET, Key: key }));
  } catch {
    /* best effort */
  }
}

/**
 * Every GUID that has a finished master.m3u8 in the bucket.
 * Used by the admin inventory page to diff bucket against database.
 */
export async function listPackagedGuids(): Promise<Set<string>> {
  const found = new Set<string>();
  let token: string | undefined = undefined;
  do {
    const res: any = await s3.send(
      new ListObjectsV2Command({
        Bucket: S3_BUCKET,
        Prefix: FOLX_PREFIX,
        MaxKeys: 1000,
        ContinuationToken: token,
      }),
    );
    for (const obj of res.Contents || []) {
      const k: string = obj.Key || "";
      if (k.endsWith("/master.m3u8") && !k.startsWith(INCOMING_PREFIX)) {
        const guid = k.slice(FOLX_PREFIX.length, -"/master.m3u8".length);
        if (guid && !guid.includes("/")) found.add(guid);
      }
    }
    token = res.IsTruncated ? res.NextContinuationToken : undefined;
  } while (token);
  return found;
}

/** Jobs still sitting in the staging prefix (uploaded but not yet packaged). */
export async function listPendingJobs(): Promise<string[]> {
  const guids = new Set<string>();
  let token: string | undefined = undefined;
  do {
    const res: any = await s3.send(
      new ListObjectsV2Command({
        Bucket: S3_BUCKET,
        Prefix: INCOMING_PREFIX,
        MaxKeys: 1000,
        ContinuationToken: token,
      }),
    );
    for (const obj of res.Contents || []) {
      const rest = (obj.Key || "").slice(INCOMING_PREFIX.length);
      const guid = rest.split("/")[0];
      if (guid) guids.add(guid);
    }
    token = res.IsTruncated ? res.NextContinuationToken : undefined;
  } while (token);
  return Array.from(guids);
}