feat: speed up file uploads (#1121)

Author: finaIcutpro

README.md

@@ -100,6 +100,8 @@ volumes:
 - `./public` - The folder where all the public assets are stored (must mount to `/zipline/public`)
 - `./themes` - The folder where all the custom themes are stored (must mount to `/zipline/themes`)
 
+Temporary files default to `./uploads/.tmp`. Setting `CORE_TEMP_DIRECTORY` to another filesystem, such as tmpfs, can reduce local upload performance.
+
 ### Generating Secrets
 
 ```bash

prisma/migrations/20260724223042_add_file_user_id_index/migration.sql

@@ -0,0 +1,2 @@
+-- CreateIndex
+CREATE INDEX "File_userId_size_idx" ON "public"."File"("userId", "size");

prisma/schema.prisma

@@ -17,7 +17,7 @@ model Zipline {
 
   coreReturnHttpsUrls Boolean @default(false)
   coreDefaultDomain   String?
-  coreTempDirectory   String // default join(tmpdir(), 'zipline')
+  coreTempDirectory   String // default resolve('./uploads/.tmp')
   coreTrustProxy      Boolean @default(false)
 
   chunksEnabled Boolean @default(true)
@@ -298,6 +298,7 @@ model File {
   thumbnail Thumbnail?
 
   @@index([name])
+  @@index([userId, size])
   @@index([folderId, createdAt])
 }
 

src/lib/api/upload.ts

@@ -33,30 +33,26 @@ export function getExtension(filename: string, override?: string) {
 }
 
 export async function checkQuota(
-  user: User | null,
+  user: Pick<User, 'id' | 'quota'> | null,
   newSize: number,
   fileCount: number,
 ): Promise<true | string> {
   if (!user?.quota) return true;
 
-  const stats = await prisma.file.aggregate({
-    where: {
-      userId: user.id,
-    },
-    _sum: {
-      size: true,
-    },
-    _count: {
-      _all: true,
-    },
-  });
+  if (user.quota.filesQuota === 'BY_BYTES') {
+    const stats = await prisma.file.aggregate({
+      where: { userId: user.id },
+      _sum: { size: true },
+    });
 
-  const aggSize = stats?._sum?.size ? stats._sum.size : 0n;
+    if (Number(stats._sum.size ?? 0n) + newSize > bytes(user.quota.maxBytes!))
+      return `uploading will exceed your storage quota of ${user.quota.maxBytes}`;
 
-  if (user.quota.filesQuota === 'BY_BYTES' && Number(aggSize) + newSize > bytes(user.quota.maxBytes!))
-    return `uploading will exceed your storage quota of ${user.quota.maxFiles} files`;
+    return true;
+  }
 
-  if (user.quota.filesQuota === 'BY_FILES' && stats?._count?._all + fileCount > user.quota.maxFiles!)
+  const count = await prisma.file.count({ where: { userId: user.id } });
+  if (count + fileCount > user.quota.maxFiles!)
     return `uploading will exceed your file count quota of ${user.quota.maxFiles} files`;
 
   return true;
@@ -81,14 +77,16 @@ export async function getFilename(
   originalName: string,
   extension: string,
   override?: string,
+  reservedNames?: Set<string>,
 ): Promise<{ error: string } | { fileName: string }> {
   try {
     let fileName = override ? sanitizeFilename(override) : formatFileName(format, originalName);
 
     if (!fileName) return { error: 'invalid file name' };
 
     let fullFileName = `${fileName}${extension}`;
-    let existing = await prisma.file.findFirst({ where: { name: fullFileName } });
+    let existing =
+      reservedNames?.has(fullFileName) || (await prisma.file.findFirst({ where: { name: fullFileName } }));
 
     if (existing && (override || format === 'name')) {
       return { error: 'file with the same name already exists' };
@@ -101,9 +99,11 @@ export async function getFilename(
       if (!fileName) return { error: 'invalid file name' };
 
       fullFileName = `${fileName}${extension}`;
-      existing = await prisma.file.findFirst({ where: { name: fullFileName } });
+      existing =
+        reservedNames?.has(fullFileName) || (await prisma.file.findFirst({ where: { name: fullFileName } }));
     }
 
+    reservedNames?.add(fullFileName);
     return { fileName };
   } catch (e) {
     logger.warn(`error generating file name: ${e}`);

src/lib/config/read/db.ts

@@ -1,6 +1,5 @@
 import { prisma } from '@/lib/db';
-import { tmpdir } from 'os';
-import { join } from 'path';
+import { resolve } from 'path';
 
 export const DATABASE_TO_PROP = {
   coreReturnHttpsUrls: 'core.returnHttpsUrls',
@@ -154,7 +153,7 @@ export async function readDatabaseSettings() {
   if (!ziplineTable) {
     ziplineTable = await prisma.zipline.create({
       data: {
-        coreTempDirectory: join(tmpdir(), 'zipline'),
+        coreTempDirectory: resolve('./uploads/.tmp'),
       },
       omit: {
         createdAt: true,

src/lib/config/validate.ts

@@ -1,5 +1,4 @@
-import { tmpdir } from 'os';
-import { join, resolve } from 'path';
+import { resolve } from 'path';
 import { z } from 'zod';
 import { log } from '../logger';
 import { ParsedConfig } from './read';
@@ -99,7 +98,7 @@ export const schema = z.object({
     tempDirectory: z
       .string()
       .transform((s) => resolve(s))
-      .default(join(tmpdir(), 'zipline')),
+      .default(resolve('./uploads/.tmp')),
     trustProxy: z.boolean().default(false),
 
     databaseUrl: z.url(),

src/lib/datasource/Local.ts

@@ -101,15 +101,15 @@ export class LocalDatasource extends Datasource {
   }
 
   public async totalSize(): Promise<number> {
-    const files = await readdir(this.dir);
-    const sizes = await Promise.all(files.map((file) => this.size(file)));
+    const files = (await readdir(this.dir, { withFileTypes: true })).filter((file) => file.isFile());
+    const sizes = await Promise.all(files.map((file) => this.size(file.name)));
 
     return sizes.reduce((a, b) => a + b, 0);
   }
 
   public async clear(): Promise<void> {
-    for (const file of await readdir(this.dir)) {
-      await rm(join(this.dir, file));
+    for (const file of await readdir(this.dir, { withFileTypes: true })) {
+      if (file.isFile()) await rm(join(this.dir, file.name));
     }
   }
 

src/lib/datasource/S3.ts

@@ -13,7 +13,9 @@ import {
   UploadPartCopyCommand,
 } from '@aws-sdk/client-s3';
 import { NodeHttpHandler } from '@smithy/node-http-handler';
+import { Upload } from '@aws-sdk/lib-storage';
 import { createReadStream } from 'fs';
+import { stat } from 'fs/promises';
 import { Agent as HttpAgent } from 'http';
 import { Agent as HttpsAgent } from 'https';
 import { Readable } from 'stream';
@@ -168,36 +170,49 @@ export class S3Datasource extends Datasource {
   }
 
   public async put(file: string, data: Buffer | string, options: PutOptions = {}): Promise<void> {
-    let command = new PutObjectCommand({
-      Bucket: this.options.bucket,
-      Key: this.key(file),
-      Body: data,
-      ...(options.mimetype ? { ContentType: options.mimetype } : {}),
-    });
+    try {
+      if (typeof data === 'string') {
+        const size = await stat(data).then((file) => file.size);
+        if (size > 25 * 1024 * 1024) {
+          // 25mb
+          this.logger.debug('putting object with multipart upload', { file, key: this.key(file) });
+
+          try {
+            const upload = new Upload({
+              client: this.client,
+              params: {
+                Bucket: this.options.bucket,
+                Key: this.key(file),
+                Body: createReadStream(data),
+                ...(options.mimetype ? { ContentType: options.mimetype } : {}),
+              },
+              leavePartsOnError: false,
+            });
+
+            await upload.done();
+            return;
+          } catch (error) {
+            this.logger.warn('multipart upload failed, retrying with a single request', {
+              error: error instanceof Error ? error.message : error,
+            });
+          }
+        }
+      }
 
-    if (typeof data === 'string') {
-      const readStream = createReadStream(data);
-      command = new PutObjectCommand({
+      const command = new PutObjectCommand({
         Bucket: this.options.bucket,
         Key: this.key(file),
-        Body: readStream,
+        Body: typeof data === 'string' ? createReadStream(data) : data,
         ...(options.mimetype ? { ContentType: options.mimetype } : {}),
       });
-
-      this.logger.debug('putting object from stream', { file, key: this.key(file) });
-    }
-
-    try {
       const res = await this.client.send(command);
 
       if (!isOk(res.$metadata.httpStatusCode || 0)) {
-        this.logger.error(
-          'there was an error while putting object',
-          res.$metadata as Record<string, unknown>,
-        );
+        throw new Error(`S3 put failed with status ${res.$metadata.httpStatusCode ?? 'unknown'}`);
       }
     } catch (e) {
       this.logger.error('there was an error while putting object', e as Record<string, unknown>);
+      throw e;
     }
   }
 

src/lib/db/models/zipline.ts

@@ -1,14 +1,12 @@
-import { tmpdir } from 'os';
 import { prisma } from '..';
-import { join } from 'path';
+import { resolve } from 'path';
 
 export async function getZipline() {
   const zipline = await prisma.zipline.findFirst();
   if (!zipline) {
-    const tmp = join(tmpdir(), 'zipline');
     return prisma.zipline.create({
       data: {
-        coreTempDirectory: tmp,
+        coreTempDirectory: resolve('./uploads/.tmp'),
       },
     });
   }

src/lib/mapConcurrent.ts

@@ -0,0 +1,27 @@
+export async function mapConcurrent<T, R>(
+  values: T[],
+  concurrency: number,
+  map: (value: T, index: number) => Promise<R>,
+): Promise<R[]> {
+  const results = Array<R>(values.length);
+  let next = 0;
+  let failed = false;
+  let failure: unknown;
+
+  async function worker() {
+    while (!failed && next < values.length) {
+      const index = next++;
+      try {
+        results[index] = await map(values[index], index);
+      } catch (error) {
+        failed = true;
+        failure = error;
+      }
+    }
+  }
+
+  await Promise.all(Array.from({ length: Math.min(concurrency, values.length) }, worker));
+  if (failed) throw failure;
+
+  return results;
+}