Merge a814aac069 into 393441faa1

src/db/db.ts

@@ -4,8 +4,21 @@ import { Database } from "bun:sqlite";
 mkdirSync("./data", { recursive: true });
 const db = new Database("./data/mydb.sqlite", { create: true });
 
-if (!db.query("SELECT * FROM sqlite_master WHERE type='table'").get()) {
-  db.exec(`
+function getTableInfo(tableName: string) {
+  try {
+    return db.query(`PRAGMA table_info('${tableName}')`).all() as Array<{ name: string }>;
+  } catch (error) {
+    console.error(`Error getting table info for ${tableName}:`, error);
+    return [];
+  }
+}
+
+function hasColumn(tableName: string, columnName: string) {
+  const info = getTableInfo(tableName);
+  return info.some((c) => c.name === columnName);
+}
+
+db.exec(`
 CREATE TABLE IF NOT EXISTS users (
 	id INTEGER PRIMARY KEY AUTOINCREMENT,
 	email TEXT NOT NULL,
@@ -17,6 +30,7 @@ CREATE TABLE IF NOT EXISTS file_names (
   file_name TEXT NOT NULL,
   output_file_name TEXT NOT NULL,
   status TEXT DEFAULT 'not started',
+  storage_key TEXT, -- v2 column
   FOREIGN KEY (job_id) REFERENCES jobs(id)
 );
 CREATE TABLE IF NOT EXISTS jobs (
@@ -27,17 +41,45 @@ CREATE TABLE IF NOT EXISTS jobs (
   num_files INTEGER DEFAULT 0,
   FOREIGN KEY (user_id) REFERENCES users(id)
 );
-PRAGMA user_version = 1;`);
-}
+-- Ensure storage_metadata exists for legacy DBs
+CREATE TABLE IF NOT EXISTS storage_metadata (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  user_id INTEGER NOT NULL,
+  job_id INTEGER NOT NULL,
+  file_name TEXT NOT NULL,
+  storage_key TEXT NOT NULL,
+  FOREIGN KEY (job_id) REFERENCES jobs(id),
+  FOREIGN KEY (user_id) REFERENCES users(id)
+);
+`);
 
-const dbVersion = (db.query("PRAGMA user_version").get() as { user_version?: number }).user_version;
-if (dbVersion === 0) {
-  db.exec("ALTER TABLE file_names ADD COLUMN status TEXT DEFAULT 'not started';");
-  db.exec("PRAGMA user_version = 1;");
-  console.log("Updated database to version 1.");
+try {
+  if (!hasColumn("file_names", "status")) {
+    db.exec("ALTER TABLE file_names ADD COLUMN status TEXT DEFAULT 'not started';");
+    console.log("Added column file_names.status");
+  }
+
+  if (!hasColumn("file_names", "storage_key")) {
+    db.exec("ALTER TABLE file_names ADD COLUMN storage_key TEXT;");
+    console.log("Added column file_names.storage_key");
+  }
+
+  const currentVersion =
+    (db.query("PRAGMA user_version").get() as { user_version?: number }).user_version ?? 0;
+
+  if (currentVersion < 2) {
+    db.exec("PRAGMA user_version = 2;");
+    console.log(`Updated database to version 2 (was ${currentVersion}).`);
+  }
+} catch (error) {
+  console.error("Error running migrations: ", error);
 }
 
 // enable WAL mode
-db.exec("PRAGMA journal_mode = WAL;");
+try {
+  db.exec("PRAGMA journal_mode = WAL;");
+} catch (error) {
+  console.warn("Could not enable WAL mode: ", error);
+}
 
 export default db;

src/pages/download.tsx

@@ -1,18 +1,18 @@
-import path from "node:path";
 import { Elysia } from "elysia";
 import sanitize from "sanitize-filename";
-import * as tar from "tar";
-import { outputDir } from "..";
 import db from "../db/db";
 import { WEBROOT } from "../helpers/env";
 import { userService } from "./user";
+import { getStorage, getStorageType } from "../storage/index";
+import { outputDir } from "..";
+import path from "path";
+import * as tar from "tar";
 
 export const download = new Elysia()
   .use(userService)
   .get(
     "/download/:userId/:jobId/:fileName",
     async ({ params, redirect, user }) => {
-      const userId = user.id;
       const job = await db
         .query("SELECT * FROM jobs WHERE user_id = ? AND id = ?")
         .get(user.id, params.jobId);
@@ -20,21 +20,40 @@ export const download = new Elysia()
       if (!job) {
         return redirect(`${WEBROOT}/results`, 302);
       }
-      // parse from URL encoded string
-      const jobId = decodeURIComponent(params.jobId);
+
       const fileName = sanitize(decodeURIComponent(params.fileName));
 
-      const filePath = `${outputDir}${userId}/${jobId}/${fileName}`;
-      return Bun.file(filePath);
+      const fileRow = db
+        .query(
+          `
+          SELECT storage_key, file_name 
+          FROM file_names
+          WHERE job_id = ? AND file_name = ?
+        `,
+        )
+        .get(params.jobId, fileName) as { storage_key?: string; file_name?: string } | undefined;
+      if (!fileRow) {
+        return redirect(`${WEBROOT}/results`, 302);
+      }
+
+      const storage = getStorage();
+      const stream = storage.getStream(fileRow.storage_key!);
+
+      return new Response(stream, {
+        headers: {
+          "Content-Type": "application/octet-stream",
+          "Content-Disposition": `attachment; filename="${fileRow.file_name ?? fileName}"`,
+        },
+      });
     },
     {
       auth: true,
     },
   )
+
   .get(
     "/archive/:jobId",
     async ({ params, redirect, user }) => {
-      const userId = user.id;
       const job = await db
         .query("SELECT * FROM jobs WHERE user_id = ? AND id = ?")
         .get(user.id, params.jobId);
@@ -43,21 +62,40 @@ export const download = new Elysia()
         return redirect(`${WEBROOT}/results`, 302);
       }
 
-      const jobId = decodeURIComponent(params.jobId);
-      const outputPath = `${outputDir}${userId}/${jobId}`;
-      const outputTar = path.join(outputPath, `converted_files_${jobId}.tar`);
+      const storageType = getStorageType();
+      if (storageType === "local") {
+        const userId = user.id;
+        const jobId = decodeURIComponent(params.jobId);
+        const outputPath = `${outputDir}${userId}/${jobId}`;
+        const outputTar = path.join(outputPath, `converted_files_${jobId}.tar`);
 
-      await tar.create(
+        await tar.create(
+          {
+            file: outputTar,
+            cwd: outputPath,
+            filter: (path) => {
+              return !path.match(".*\\.tar");
+            },
+          },
+          ["."],
+        );
+
+        return Bun.file(outputTar);
+      }
+
+      return new Response(
+        JSON.stringify({
+          ok: false,
+          message:
+            "Archive download is not supported when object storage is enabled. This is intentional it avoid 404s - please use per-file downloads or request a follow-up for server-side archiving",
+        }),
         {
-          file: outputTar,
-          cwd: outputPath,
-          filter: (path) => {
-            return !path.match(".*\\.tar");
+          status: 501,
+          headers: {
+            "Content-Type": "application/json",
           },
         },
-        ["."],
       );
-      return Bun.file(outputTar);
     },
     {
       auth: true,

src/pages/upload.tsx

@@ -1,9 +1,10 @@
 import { Elysia, t } from "elysia";
 import db from "../db/db";
 import { WEBROOT } from "../helpers/env";
-import { uploadsDir } from "../index";
 import { userService } from "./user";
 import sanitize from "sanitize-filename";
+import { getStorage } from "../storage";
+import crypto from "node:crypto";
 
 export const upload = new Elysia().use(userService).post(
   "/upload",
@@ -12,6 +13,8 @@ export const upload = new Elysia().use(userService).post(
       return redirect(`${WEBROOT}/`, 302);
     }
 
+    const jobIdValue = jobId.value;
+
     const existingJob = await db
       .query("SELECT * FROM jobs WHERE id = ? AND user_id = ?")
       .get(jobId.value, user.id);
@@ -20,17 +23,29 @@ export const upload = new Elysia().use(userService).post(
       return redirect(`${WEBROOT}/`, 302);
     }
 
-    const userUploadsDir = `${uploadsDir}${user.id}/${jobId.value}/`;
+    const storage = getStorage();
+
+    const saveFile = async (file: File) => {
+      const sanitizedFileName = sanitize(file.name);
+      const storageKey = `${user.id}/${jobId.value}/${crypto.randomUUID()}`;
+      const buffer = Buffer.from(await file.arrayBuffer());
+      await storage.save(storageKey, buffer);
+
+      db.query(
+        `
+        INSERT INTO file_names (job_id, file_name, storage_key)
+        VALUES (?, ?, ?)
+        `,
+      ).run(jobIdValue, sanitizedFileName, storageKey);
+    };
 
     if (body?.file) {
       if (Array.isArray(body.file)) {
         for (const file of body.file) {
-          const santizedFileName = sanitize(file.name);
-          await Bun.write(`${userUploadsDir}${santizedFileName}`, file);
+          await saveFile(file);
         }
       } else {
-        const santizedFileName = sanitize(body.file["name"]);
-        await Bun.write(`${userUploadsDir}${santizedFileName}`, body.file);
+        await saveFile(body.file);
       }
     }
 

src/storage/LocalStorageAdapter.ts

@@ -0,0 +1,44 @@
+import { IStorageAdapter } from "./index";
+import { promises as fs } from "fs";
+import path from "path";
+
+export class LocalStorageAdapter implements IStorageAdapter {
+  private baseDir: string;
+
+  constructor(baseDir: string) {
+    this.baseDir = baseDir;
+  }
+
+  async save(key: string, data: Buffer): Promise<string> {
+    const fullPath = path.join(this.baseDir, key);
+    await fs.mkdir(path.dirname(fullPath), { recursive: true });
+    await fs.writeFile(fullPath, data);
+    return key;
+  }
+
+  async get(key: string): Promise<Buffer> {
+    const fullPath = path.join(this.baseDir, key);
+    return fs.readFile(fullPath);
+  }
+
+  async delete(key: string): Promise<void> {
+    const fullPath = path.join(this.baseDir, key);
+    try {
+      await fs.unlink(fullPath);
+    } catch (error) {
+      const err = error as NodeJS.ErrnoException;
+      if (err?.code === "ENOENT" || err?.code === "ENOTDIR") {
+        return;
+      }
+
+      console.error(`Failed to delete file at ${fullPath}: `, error);
+      throw error;
+    }
+  }
+
+  getStream(key: string): ReadableStream<Uint8Array> {
+    const fullPath = path.join(this.baseDir, key);
+    const file = Bun.file(fullPath);
+    return file.stream();
+  }
+}

src/storage/S3StorageAdapter.ts

@@ -0,0 +1,50 @@
+import { s3, S3File } from "bun";
+import { IStorageAdapter } from "./index";
+
+export class S3StorageAdapter implements IStorageAdapter {
+  private bucket: string;
+
+  constructor(bucket: string) {
+    this.bucket = bucket;
+  }
+
+  async save(key: string, data: Buffer): Promise<string> {
+    const opts: Record<string, unknown> = {
+      bucket: this.bucket,
+    };
+
+    if (process.env.S3_USE_ACL === "true") {
+      const aclValue = process.env.S3_ACL_VALUE || "private";
+      opts.acl = aclValue;
+    }
+
+    const file: S3File = s3.file(key, opts);
+
+    await file.write(data);
+    return key;
+  }
+
+  async get(key: string): Promise<Buffer> {
+    const file: S3File = s3.file(key, {
+      bucket: this.bucket,
+    });
+
+    const buf = await file.bytes();
+    return Buffer.from(buf);
+  }
+
+  async delete(key: string): Promise<void> {
+    const file: S3File = s3.file(key, {
+      bucket: this.bucket,
+    });
+
+    await file.delete();
+  }
+
+  getStream(key: string): ReadableStream<Uint8Array> {
+    const file: S3File = s3.file(key, {
+      bucket: this.bucket,
+    });
+    return file.stream();
+  }
+}

src/storage/index.ts

@@ -0,0 +1,27 @@
+import { LocalStorageAdapter } from "./LocalStorageAdapter";
+import { S3StorageAdapter } from "./S3StorageAdapter";
+
+export interface IStorageAdapter {
+  save(key: string, data: Buffer): Promise<string>;
+  get(key: string): Promise<Buffer>;
+  delete(key: string): Promise<void>;
+  getStream(key: string): ReadableStream<Uint8Array>;
+}
+
+export function getStorageType(): "local" | "s3" {
+  return process.env.STORAGE_TYPE === "s3" ? "s3" : "local";
+}
+
+export function getStorage(): IStorageAdapter {
+  if (getStorageType() === "s3") {
+    const bucket = process.env.S3_BUCKET_NAME;
+    if (!bucket) {
+      throw new Error("S3_BUCKET_NAME must be set when STORAGE_TYPE=s3");
+    }
+
+    return new S3StorageAdapter(bucket);
+  }
+
+  const baseDir = process.env.LOCAL_STORAGE_PATH || "./data/storage";
+  return new LocalStorageAdapter(baseDir);
+}