Add resumable uploads, proxy discovery, PWA install and 2 GiB personal vaults

apps/server/src/storage.ts

@@ -3,7 +3,7 @@ import { mkdirSync, statfsSync, existsSync } from 'node:fs';
 import { createWriteStream, createReadStream } from 'node:fs';
 import { rename, unlink, readdir, stat, open } from 'node:fs/promises';
 import { resolve } from 'node:path';
-import { Transform } from 'node:stream';
+import { Transform, Writable } from 'node:stream';
 import { pipeline } from 'node:stream/promises';
 import { and, eq, gt, isNull, inArray, sql, or } from 'drizzle-orm';
 import sharp from 'sharp';
@@ -146,6 +146,7 @@ export class Storage {
         .run();
   }
   release(requestId: string) {
+    this.service.resumable?.abort(requestId);
     const db = this.service.db;
     for (const file of this.fileRows(requestId)) {
       for (const [uploadId, upload] of this.active)
@@ -161,12 +162,51 @@ export class Storage {
           .run();
     }
   }
+  uploadManifest(g: StorageSubject, requestId: string, position: number) {
+    if (g.kind === 'user') {
+      const session = g.sessionId
+        ? this.service.db
+            .select()
+            .from(s.accountSessions)
+            .where(eq(s.accountSessions.id, g.sessionId))
+            .get()
+        : null;
+      if (
+        !session ||
+        session.userId !== g.id ||
+        session.revokedAt ||
+        session.expiresAt <= Date.now()
+      )
+        fail(401, 'SESSION_REVOKED', '계정 세션이 종료되었습니다.');
+    } else {
+      const guest = this.service.db.select().from(s.guests).where(eq(s.guests.id, g.id)).get();
+      if (!guest || guest.revokedAt || guest.expiresAt <= Date.now())
+        fail(401, 'SESSION_EXPIRED', '기기 세션이 종료되었습니다.');
+    }
+    const row = this.request(g, requestId);
+    if (row.kind !== 'file' || !this.isSender(g, row))
+      fail(403, 'UPLOAD_FORBIDDEN', '발신자만 업로드할 수 있습니다.');
+    if (this.service.vault.source(requestId))
+      fail(409, 'VAULT_TRANSFER_MANAGED', '이미 저장된 파일을 전달합니다.');
+    if (
+      row.status !== 'accepted' ||
+      !row.startBy ||
+      row.startBy <= Date.now() ||
+      this.blocked(g, row)
+    )
+      fail(409, 'ACCEPTANCE_REQUIRED', '수락된 유효한 요청이 필요합니다.');
+    const file = this.fileRows(requestId).find((f) => f.position === position);
+    if (!file) fail(404, 'FILE_NOT_REQUESTED', '요청에 없는 파일입니다.');
+    if (file!.status === 'uploading') fail(409, 'UPLOAD_ACTIVE', '파일이 처리 중입니다.');
+    return file!;
+  }
   async upload(
     g: StorageSubject,
     requestId: string,
     position: number,
     request: FastifyRequest,
     key: string,
+    stagedPath?: string,
   ) {
     const { db, cfg } = this.service;
     let file: typeof s.files.$inferSelect;
@@ -230,7 +270,7 @@ export class Storage {
         .run();
     });
     const controller = new AbortController(),
-      tempPath = resolve(this.temp, uploadId!);
+      tempPath = stagedPath ?? resolve(this.temp, uploadId!);
     this.active.set(uploadId!, {
       guestId: `${g.kind ?? 'guest'}:${g.id}`,
       controller,
@@ -265,7 +305,13 @@ export class Storage {
         await pipeline(
           part.file,
           counter,
-          createWriteStream(tempPath, { flags: 'wx', mode: 0o600 }),
+          stagedPath
+            ? new Writable({
+                write(_chunk, _encoding, callback) {
+                  callback();
+                },
+              })
+            : createWriteStream(tempPath, { flags: 'wx', mode: 0o600 }),
           { signal: controller.signal },
         );
         if (part.file.truncated || bytes !== file!.size)
@@ -486,6 +532,11 @@ export class Storage {
           )
           .get();
         if (row) {
+          if (g.kind === 'user' && row.item.fileId) {
+            const file = db.select().from(s.files).where(eq(s.files.id, row.item.fileId)).get();
+            if (file && !this.service.accounts.canStoreFile(g.id, file))
+              fail(413, 'ACCOUNT_QUOTA', '보관함 용량이 부족해 복원하지 못했습니다.');
+          }
           db.update(s.access).set({ deletedAt: null }).where(eq(s.access.id, row.access.id)).run();
           count++;
         }
@@ -505,6 +556,7 @@ export class Storage {
     if (this.cleaning) return;
     this.cleaning = true;
     try {
+      this.service.resumable?.cleanup();
       const { db, cfg } = this.service,
         now = Date.now();
       const expired = db
@@ -580,10 +632,12 @@ export class Storage {
     }
   }
   abortSession(id: string) {
+    this.service.resumable?.abortSession(id);
     for (const upload of this.active.values())
       if (upload.sessionId === id) upload.controller.abort();
   }
   abortUser(id: string) {
+    this.service.resumable?.abortUser(id);
     for (const upload of this.active.values())
       if (upload.guestId === `user:${id}`) upload.controller.abort();
   }