WRNexusJS/packages/uploader/src/operations.ts

import type { PutMeta, StorageDriver } from "./driver.ts";

export interface QuotaUsage {
  owner: string;
  bytes: number;
  objects: number;
  updatedAt: number;
}
export interface QuotaStore {
  get(owner: string): Promise<QuotaUsage>;
  reserve(
    owner: string,
    bytes: number,
    limits: { bytes: number; objects?: number },
  ): Promise<boolean>;
  release(owner: string, bytes: number): Promise<void>;
}

export function memoryQuotaStore(): QuotaStore {
  const values = new Map<string, QuotaUsage>();
  return {
    async get(owner) {
      return structuredClone(
        values.get(owner) ?? { owner, bytes: 0, objects: 0, updatedAt: Date.now() },
      );
    },
    async reserve(owner, bytes, limits) {
      if (!Number.isInteger(bytes) || bytes < 0)
        throw new RangeError("Quota bytes must be non-negative");
      const current = values.get(owner) ?? { owner, bytes: 0, objects: 0, updatedAt: Date.now() };
      if (
        current.bytes + bytes > limits.bytes ||
        current.objects + 1 > (limits.objects ?? Number.MAX_SAFE_INTEGER)
      )
        return false;
      values.set(owner, {
        owner,
        bytes: current.bytes + bytes,
        objects: current.objects + 1,
        updatedAt: Date.now(),
      });
      return true;
    },
    async release(owner, bytes) {
      const current = values.get(owner);
      if (!current) return;
      values.set(owner, {
        ...current,
        bytes: Math.max(0, current.bytes - Math.max(0, bytes)),
        objects: Math.max(0, current.objects - 1),
        updatedAt: Date.now(),
      });
    },
  };
}

export interface QuotaSqlClient {
  query<T = any>(sql: string, parameters?: unknown[]): Promise<{ rows: T[] }>;
}

/** PostgreSQL quota accounting using a single atomic conditional upsert. */
export function postgresQuotaStore(
  db: QuotaSqlClient,
  table = "wrnexus_storage_quota",
): QuotaStore {
  if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(table)) throw new Error("Invalid quota table name");
  return {
    async get(owner) {
      const result = await db.query<QuotaUsage>(
        `SELECT owner,bytes,objects,updated_at AS "updatedAt" FROM ${table} WHERE owner=$1`,
        [owner],
      );
      return result.rows[0] ?? { owner, bytes: 0, objects: 0, updatedAt: Date.now() };
    },
    async reserve(owner, bytes, limits) {
      const result = await db.query(
        `INSERT INTO ${table} (owner,bytes,objects,updated_at) VALUES ($1,$2,1,$5) ON CONFLICT (owner) DO UPDATE SET bytes=${table}.bytes+$2,objects=${table}.objects+1,updated_at=$5 WHERE ${table}.bytes+$2 <= $3 AND ${table}.objects+1 <= $4 RETURNING owner`,
        [owner, bytes, limits.bytes, limits.objects ?? 2_147_483_647, Date.now()],
      );
      return result.rows.length === 1;
    },
    async release(owner, bytes) {
      await db.query(
        `UPDATE ${table} SET bytes=GREATEST(0,bytes-$2),objects=GREATEST(0,objects-1),updated_at=$3 WHERE owner=$1`,
        [owner, bytes, Date.now()],
      );
    },
  };
}

export const POSTGRES_QUOTA_SCHEMA = `CREATE TABLE IF NOT EXISTS wrnexus_storage_quota (owner text PRIMARY KEY, bytes bigint NOT NULL DEFAULT 0, objects integer NOT NULL DEFAULT 0, updated_at bigint NOT NULL);`;

export interface MultipartObjectClient {
  create(key: string, meta: PutMeta): Promise<string>;
  uploadPart(uploadId: string, key: string, part: number, bytes: Uint8Array): Promise<string>;
  complete(
    uploadId: string,
    key: string,
    parts: Array<{ part: number; etag: string }>,
  ): Promise<void>;
  abort(uploadId: string, key: string): Promise<void>;
}

export async function multipartUpload(
  client: MultipartObjectClient,
  key: string,
  bytes: Uint8Array,
  meta: PutMeta,
  options: { partBytes?: number; concurrency?: number } = {},
): Promise<void> {
  const partBytes = options.partBytes ?? 8 * 1024 * 1024;
  const concurrency = options.concurrency ?? 4;
  if (partBytes < 5 * 1024 * 1024) throw new RangeError("Multipart parts must be at least 5 MiB");
  if (!Number.isInteger(concurrency) || concurrency < 1 || concurrency > 32)
    throw new RangeError("Multipart concurrency must be between 1 and 32");
  const uploadId = await client.create(key, meta);
  const chunks = Array.from({ length: Math.ceil(bytes.length / partBytes) }, (_, index) => ({
    part: index + 1,
    bytes: bytes.slice(index * partBytes, (index + 1) * partBytes),
  }));
  const completed: Array<{ part: number; etag: string }> = [];
  try {
    for (let offset = 0; offset < chunks.length; offset += concurrency) {
      completed.push(
        ...(await Promise.all(
          chunks.slice(offset, offset + concurrency).map(async (chunk) => ({
            part: chunk.part,
            etag: await client.uploadPart(uploadId, key, chunk.part, chunk.bytes),
          })),
        )),
      );
    }
    await client.complete(uploadId, key, completed);
  } catch (error) {
    await client.abort(uploadId, key);
    throw error;
  }
}

export interface TemporaryObject {
  key: string;
  expiresAt: number;
}
export function createTemporaryObjectCleaner(
  driver: StorageDriver,
  options: { now?: () => number; limit?: number } = {},
) {
  const objects = new Map<string, number>();
  const now = options.now ?? Date.now;
  const limit = options.limit ?? 10_000;
  return {
    track(key: string, ttlMs: number) {
      if (objects.size >= limit) throw new Error("WRN-UPLOAD-TEMP-CAPACITY");
      if (ttlMs < 1) throw new RangeError("Temporary TTL must be positive");
      objects.set(key, now() + ttlMs);
    },
    async cleanup(at = now()) {
      const due = [...objects].filter(([, expiry]) => expiry <= at);
      for (const [key] of due) {
        await driver.delete(key);
        objects.delete(key);
      }
      return due.length;
    },
    snapshot: () => ({
      tracked: objects.size,
      nextExpiry: objects.size ? Math.min(...objects.values()) : undefined,
    }),
  };
}

export interface VideoTranscodeOptions {
  format: "mp4" | "webm";
  width?: number;
  height?: number;
  videoBitrateKbps?: number;
}
export function ffmpegVideoTranscoder(
  options: { executable?: string; spawn?: (args: string[]) => { exited: Promise<number> } } = {},
) {
  return async (input: string, output: string, config: VideoTranscodeOptions): Promise<void> => {
    const validPath = (value: string) =>
      /^[\w .:\\/-]+$/.test(value) && !value.split(/[\\/]/).includes("..");
    if (!validPath(input) || !validPath(output)) throw new Error("Invalid video path");
    if (config.format !== "mp4" && config.format !== "webm")
      throw new Error("Invalid video format");
    // These reach an ffmpeg filter string, so reject anything that is not a
    // plain positive integer rather than trusting the declared type.
    const dimension = (value: number | undefined, name: string): number | undefined => {
      if (value === undefined) return undefined;
      if (!Number.isInteger(value) || value <= 0 || value > 16384)
        throw new Error(`Invalid video ${name}`);
      return value;
    };
    const width = dimension(config.width, "width");
    const height = dimension(config.height, "height");
    const bitrate = dimension(config.videoBitrateKbps, "bitrate");
    const args = [
      options.executable ?? "ffmpeg",
      "-y",
      "-i",
      input,
      ...(width || height ? ["-vf", `scale=${width ?? -2}:${height ?? -2}`] : []),
      ...(bitrate ? ["-b:v", `${bitrate}k`] : []),
      "-f",
      config.format,
      output,
    ];
    const child = options.spawn?.(args) ?? Bun.spawn(args, { stdout: "ignore", stderr: "ignore" });
    if ((await child.exited) !== 0) throw new Error("WRN-VIDEO-TRANSCODE-FAILED");
  };
}