서버 코드를 Router, Service, Repository 계층으로 분리하고 오류 처리 및 파일 삭제 로직 개선

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/lib/server/services/auth.ts

@@ -0,0 +1,99 @@
+import argon2 from "argon2";
+import { ClientRepo, IntegrityError, SessionRepo, UserRepo } from "#lib/server/db/index.js";
+import env from "#lib/server/config.js";
+import { issueSessionId } from "#lib/server/crypto.js";
+import { AppError } from "#lib/server/errors.js";
+import { createChallenge, verifyChallengeAnswer } from "./client";
+
+const MIN_PASSWORD_LENGTH = 8;
+
+export const login = async (email: string, password: string, ip: string, userAgent: string) => {
+  const user = await UserRepo.getUserByEmail(email);
+  if (!user || !(await argon2.verify(user.password, password))) {
+    throw new AppError("UNAUTHORIZED", "Invalid email or password");
+  }
+
+  const { sessionId, sessionIdSigned } = await issueSessionId(32, env.session.secret);
+  await SessionRepo.createSession(user.id, sessionId, ip, userAgent);
+  return { sessionIdSigned };
+};
+
+export const logout = async (sessionId: string) => {
+  await SessionRepo.deleteSession(sessionId);
+};
+
+export const changePassword = async (
+  userId: number,
+  sessionId: string,
+  oldPassword: string,
+  newPassword: string,
+) => {
+  if (oldPassword === newPassword) {
+    throw new AppError("BAD_REQUEST", "Same passwords");
+  } else if (newPassword.length < MIN_PASSWORD_LENGTH) {
+    throw new AppError("BAD_REQUEST", "Too short password");
+  }
+
+  const user = await UserRepo.getUser(userId);
+  if (!user) {
+    throw new AppError("INTERNAL_SERVER_ERROR", "Invalid session id");
+  } else if (!(await argon2.verify(user.password, oldPassword))) {
+    throw new AppError("FORBIDDEN", "Invalid password");
+  }
+
+  await UserRepo.setUserPassword(userId, await argon2.hash(newPassword));
+  await SessionRepo.deleteAllOtherSessions(userId, sessionId);
+};
+
+export const requestSessionUpgrade = async (
+  userId: number,
+  sessionId: string,
+  ip: string,
+  encPubKey: string,
+  sigPubKey: string,
+) => {
+  const client = await ClientRepo.getClientByPubKeys(encPubKey, sigPubKey);
+  if (!client) {
+    throw new AppError("UNAUTHORIZED", "Invalid public key(s)");
+  }
+
+  const userClient = await ClientRepo.getUserClient(userId, client.id);
+  if (!userClient || userClient.state === "challenging") {
+    throw new AppError("FORBIDDEN", "Unregistered client");
+  }
+
+  return await createChallenge(encPubKey, (answer) =>
+    SessionRepo.registerSessionUpgradeChallenge(
+      sessionId,
+      client.id,
+      answer,
+      ip,
+      new Date(Date.now() + env.challenge.sessionUpgradeExp),
+    ),
+  );
+};
+
+export const verifySessionUpgrade = async (
+  userId: number,
+  sessionId: string,
+  ip: string,
+  challengeId: number,
+  answerSig: string,
+  force: boolean,
+) => {
+  const challenge = await SessionRepo.consumeSessionUpgradeChallenge(challengeId, sessionId, ip);
+  const client = await verifyChallengeAnswer(challenge, answerSig);
+
+  try {
+    await SessionRepo.upgradeSession(userId, sessionId, client.id, force);
+  } catch (e) {
+    if (e instanceof IntegrityError) {
+      if (e.message === "Session not found") {
+        throw new AppError("INTERNAL_SERVER_ERROR", "Invalid challenge answer");
+      } else if (e.message === "Session already exists") {
+        throw new AppError("CONFLICT", "Already logged in");
+      }
+    }
+    throw e;
+  }
+};

src/lib/server/services/category.ts

@@ -0,0 +1,62 @@
+import { CategoryRepo, FileRepo } from "#lib/server/db/index.js";
+import type { Ciphertext } from "#lib/server/db/schema/index.js";
+import { AppError } from "#lib/server/errors.js";
+import { ensureFreshDekVersion } from "./common";
+
+export const getCategory = async (userId: number, categoryId: CategoryId, recurse: boolean) => {
+  const category =
+    categoryId !== "root" ? await CategoryRepo.getCategory(userId, categoryId) : undefined;
+  if (category === null) {
+    throw new AppError("NOT_FOUND", "Category not found");
+  }
+
+  const [subCategories, files] = await Promise.all([
+    CategoryRepo.getAllCategoriesByParent(userId, categoryId),
+    categoryId !== "root" ? FileRepo.getAllFilesByCategory(userId, categoryId, recurse) : undefined,
+  ]);
+  return { category, subCategories, files };
+};
+
+export const createCategory = async (params: CategoryRepo.NewCategory) => {
+  ensureFreshDekVersion(params.dekVersion);
+  await CategoryRepo.registerCategory(params);
+};
+
+export const renameCategory = async (
+  userId: number,
+  categoryId: number,
+  dekVersion: Date,
+  encName: Ciphertext,
+) => {
+  await CategoryRepo.setCategoryEncName(userId, categoryId, dekVersion, encName);
+};
+
+export const deleteCategory = async (userId: number, categoryId: number) => {
+  await CategoryRepo.unregisterCategory(userId, categoryId);
+};
+
+const ensureCategoryAndFileExist = async (userId: number, categoryId: number, fileId: number) => {
+  const [category, file] = await Promise.all([
+    CategoryRepo.getCategory(userId, categoryId),
+    FileRepo.getFile(userId, fileId),
+  ]);
+  if (!category) {
+    throw new AppError("NOT_FOUND", "Category not found");
+  } else if (!file) {
+    throw new AppError("NOT_FOUND", "File not found");
+  }
+};
+
+export const addFileToCategory = async (userId: number, categoryId: number, fileId: number) => {
+  await ensureCategoryAndFileExist(userId, categoryId, fileId);
+  await FileRepo.addFileToCategory(fileId, categoryId);
+};
+
+export const removeFileFromCategory = async (
+  userId: number,
+  categoryId: number,
+  fileId: number,
+) => {
+  await ensureCategoryAndFileExist(userId, categoryId, fileId);
+  await FileRepo.removeFileFromCategory(fileId, categoryId);
+};

src/lib/server/services/client.ts

@@ -0,0 +1,76 @@
+import { ClientRepo } from "#lib/server/db/index.js";
+import env from "#lib/server/config.js";
+import { generateChallenge, verifyPubKey, verifySignature } from "#lib/server/crypto.js";
+import { AppError } from "#lib/server/errors.js";
+
+const CHALLENGE_LENGTH = 32;
+
+export const createChallenge = async (
+  encPubKey: string,
+  register: (answer: string) => Promise<{ id: number }>,
+) => {
+  const { answer, challenge } = await generateChallenge(CHALLENGE_LENGTH, encPubKey);
+  const { id } = await register(answer.toString("base64"));
+  return { id, challenge: challenge.toString("base64") };
+};
+
+export const verifyChallengeAnswer = async (
+  challenge: { clientId: number; answer: string } | null,
+  answerSig: string,
+) => {
+  if (!challenge) {
+    throw new AppError("FORBIDDEN", "Invalid challenge answer");
+  }
+
+  const client = await ClientRepo.getClient(challenge.clientId);
+  if (!client) {
+    throw new AppError("INTERNAL_SERVER_ERROR", "Invalid challenge answer");
+  } else if (
+    !verifySignature(Buffer.from(challenge.answer, "base64"), answerSig, client.sigPubKey)
+  ) {
+    throw new AppError("FORBIDDEN", "Invalid challenge answer signature");
+  }
+  return client;
+};
+
+export const registerClient = async (
+  userId: number,
+  ip: string,
+  encPubKey: string,
+  sigPubKey: string,
+) => {
+  const getClientId = async () => {
+    const client = await ClientRepo.getClientByPubKeys(encPubKey, sigPubKey);
+    if (client) {
+      await ClientRepo.createUserClient(userId, client.id);
+      return client.id;
+    } else if (encPubKey === sigPubKey) {
+      throw new AppError("BAD_REQUEST", "Same public keys");
+    } else if (!verifyPubKey(encPubKey) || !verifyPubKey(sigPubKey)) {
+      throw new AppError("BAD_REQUEST", "Invalid public key(s)");
+    }
+    return (await ClientRepo.createClient(encPubKey, sigPubKey, userId)).id;
+  };
+
+  const clientId = await getClientId();
+  return await createChallenge(encPubKey, (answer) =>
+    ClientRepo.registerUserClientChallenge(
+      userId,
+      clientId,
+      answer,
+      ip,
+      new Date(Date.now() + env.challenge.userClientExp),
+    ),
+  );
+};
+
+export const verifyClient = async (
+  userId: number,
+  ip: string,
+  challengeId: number,
+  answerSig: string,
+) => {
+  const challenge = await ClientRepo.consumeUserClientChallenge(challengeId, userId, ip);
+  const client = await verifyChallengeAnswer(challenge, answerSig);
+  await ClientRepo.setUserClientStateToPending(userId, client.id);
+};

src/lib/server/services/common.ts

@@ -0,0 +1,10 @@
+import { AppError } from "#lib/server/errors.js";
+
+const DEK_VERSION_TOLERANCE = 60 * 1000; // 1 minute
+
+// DEK versions are creation timestamps generated by clients
+export const ensureFreshDekVersion = (dekVersion: Date) => {
+  if (Math.abs(dekVersion.getTime() - Date.now()) >= DEK_VERSION_TOLERANCE) {
+    throw new AppError("BAD_REQUEST", "Invalid DEK version");
+  }
+};

src/lib/server/services/directory.ts

@@ -0,0 +1,58 @@
+import { DirectoryRepo, FileRepo } from "#lib/server/db/index.js";
+import type { Ciphertext } from "#lib/server/db/schema/index.js";
+import { AppError } from "#lib/server/errors.js";
+import { ensureFreshDekVersion } from "./common";
+import { schedulePendingFileDeletions } from "./fileDeletion";
+
+export const getDirectory = async (userId: number, directoryId: DirectoryId) => {
+  const directory =
+    directoryId !== "root" ? await DirectoryRepo.getDirectory(userId, directoryId) : undefined;
+  if (directory === null) {
+    throw new AppError("NOT_FOUND", "Directory not found");
+  }
+
+  const [subDirectories, files] = await Promise.all([
+    DirectoryRepo.getAllDirectoriesByParent(userId, directoryId),
+    FileRepo.getAllFilesByParent(userId, directoryId),
+  ]);
+  return { directory, subDirectories, files };
+};
+
+export const createDirectory = async (params: {
+  userId: number;
+  parentId: DirectoryId;
+  mekVersion: number;
+  encDek: string;
+  dekVersion: Date;
+  encName: Ciphertext;
+}) => {
+  ensureFreshDekVersion(params.dekVersion);
+  await DirectoryRepo.registerDirectory(params);
+};
+
+export const renameDirectory = async (
+  userId: number,
+  directoryId: number,
+  dekVersion: Date,
+  encName: Ciphertext,
+) => {
+  await DirectoryRepo.setDirectoryEncName(userId, directoryId, dekVersion, encName);
+};
+
+export const deleteDirectory = async (userId: number, directoryId: number) => {
+  const { deletedFileIds } = await DirectoryRepo.unregisterDirectory(userId, directoryId);
+  schedulePendingFileDeletions();
+  return { deletedFileIds };
+};
+
+export const getFavoriteDirectories = async (userId: number) => {
+  return await DirectoryRepo.getAllFavoriteDirectories(userId);
+};
+
+export const setDirectoryFavorite = async (
+  userId: number,
+  directoryId: number,
+  isFavorite: boolean,
+) => {
+  await DirectoryRepo.setDirectoryFavorite(userId, directoryId, isFavorite);
+};

src/lib/server/services/file.ts

@@ -1,75 +1,82 @@
-import { error } from "@sveltejs/kit";
-import { createReadStream } from "fs";
-import { stat } from "fs/promises";
-import { Readable } from "stream";
 import { FileRepo, MediaRepo } from "#lib/server/db/index.js";
+import type { Ciphertext } from "#lib/server/db/schema/index.js";
+import { AppError } from "#lib/server/errors.js";
+import { createEncContentStream } from "#lib/server/storage/index.js";
+import { schedulePendingFileDeletions } from "./fileDeletion";
 
-export const createEncContentStream = async (
-  path: string,
-  iv?: Buffer,
-  range?: { start?: number; end?: number },
-) => {
-  const { size: fileSize } = await stat(path);
-  const ivSize = iv?.byteLength ?? 0;
-  const totalSize = fileSize + ivSize;
+type Range = { start?: number; end?: number };
 
-  const start = range?.start ?? 0;
-  const end = range?.end ?? totalSize - 1;
-  if (start > end || start < 0 || end >= totalSize) {
-    error(416, "Invalid range");
-  }
-
-  return {
-    encContentStream: Readable.toWeb(
-      Readable.from(
-        (async function* () {
-          if (start < ivSize) {
-            yield iv!.subarray(start, Math.min(end + 1, ivSize));
-          }
-          if (end >= ivSize) {
-            yield* createReadStream(path, {
-              start: Math.max(0, start - ivSize),
-              end: end - ivSize,
-            });
-          }
-        })(),
-      ),
-    ),
-    range: { start, end, total: totalSize },
-    isRangeRequest: !!range,
-  };
-};
-
-export const getFileStream = async (
-  userId: number,
-  fileId: number,
-  range?: { start?: number; end?: number },
-) => {
+export const getFile = async (userId: number, fileId: number) => {
   const file = await FileRepo.getFile(userId, fileId);
   if (!file) {
-    error(404, "Invalid file id");
+    throw new AppError("NOT_FOUND", "File not found");
   }
 
-  return createEncContentStream(
-    file.path,
-    file.encContentIv ? Buffer.from(file.encContentIv, "base64") : undefined,
-    range,
-  );
+  const categories = await FileRepo.getAllFileCategories(fileId);
+  return { ...file, categories };
 };
 
-export const getFileThumbnailStream = async (
+export const getFiles = async (userId: number, fileIds: number[]) => {
+  return await FileRepo.getFilesWithCategories(userId, fileIds);
+};
+
+export const getAllFileIds = async (userId: number) => {
+  return await FileRepo.getAllFileIds(userId);
+};
+
+export const getFileIdsByContentHmac = async (
+  userId: number,
+  hskVersion: number,
+  contentHmac: string,
+) => {
+  return await FileRepo.getAllFileIdsByContentHmac(userId, hskVersion, contentHmac);
+};
+
+export const getFilesWithoutThumbnail = async (userId: number) => {
+  return await FileRepo.getFilesWithoutThumbnail(userId);
+};
+
+export const renameFile = async (
   userId: number,
   fileId: number,
-  range?: { start?: number; end?: number },
+  dekVersion: Date,
+  encName: Ciphertext,
 ) => {
+  await FileRepo.setFileEncName(userId, fileId, dekVersion, encName);
+};
+
+export const deleteFile = async (userId: number, fileId: number) => {
+  await FileRepo.unregisterFile(userId, fileId);
+  schedulePendingFileDeletions();
+};
+
+export const getFavoriteFiles = async (userId: number) => {
+  return await FileRepo.getAllFavoriteFiles(userId);
+};
+
+export const setFileFavorite = async (userId: number, fileId: number, isFavorite: boolean) => {
+  await FileRepo.setFileFavorite(userId, fileId, isFavorite);
+};
+
+export const getFileThumbnail = async (userId: number, fileId: number) => {
   const thumbnail = await MediaRepo.getFileThumbnail(userId, fileId);
   if (!thumbnail) {
-    error(404, "File or its thumbnail not found");
+    throw new AppError("NOT_FOUND", "File or its thumbnail not found");
   }
+  return thumbnail;
+};
 
-  return createEncContentStream(
-    thumbnail.path,
-    thumbnail.encContentIv ? Buffer.from(thumbnail.encContentIv, "base64") : undefined,
-    range,
-  );
+const decodeIv = (iv: string | null) => (iv ? Buffer.from(iv, "base64") : undefined);
+
+export const getFileStream = async (userId: number, fileId: number, range?: Range) => {
+  const file = await FileRepo.getFile(userId, fileId);
+  if (!file) {
+    throw new AppError("NOT_FOUND", "File not found");
+  }
+  return await createEncContentStream(file.path, decodeIv(file.encContentIv), range);
+};
+
+export const getFileThumbnailStream = async (userId: number, fileId: number, range?: Range) => {
+  const thumbnail = await getFileThumbnail(userId, fileId);
+  return await createEncContentStream(thumbnail.path, decodeIv(thumbnail.encContentIv), range);
 };

src/lib/server/services/fileDeletion.ts

@@ -0,0 +1,44 @@
+import { FileDeletionRepo } from "#lib/server/db/index.js";
+import { deleteFile } from "#lib/server/storage/index.js";
+
+const BATCH_SIZE = 100;
+
+// Deletes files enqueued by committed transactions. Failed deletions are retried later.
+export const processPendingFileDeletions = async () => {
+  while (true) {
+    const deletions = await FileDeletionRepo.getPendingFileDeletions(BATCH_SIZE);
+    if (deletions.length === 0) return;
+
+    const results = await Promise.allSettled(deletions.map(({ path }) => deleteFile(path)));
+    const deletedIds = deletions
+      .filter((_, i) => results[i]!.status === "fulfilled")
+      .map(({ id }) => id);
+    results.forEach((result) => {
+      if (result.status === "rejected") console.error(result.reason);
+    });
+
+    await FileDeletionRepo.completeFileDeletions(deletedIds);
+    if (deletedIds.length < deletions.length) return; // Retry later
+  }
+};
+
+let processing: Promise<void> | undefined;
+let isRequestedAgain = false;
+
+export const schedulePendingFileDeletions = () => {
+  if (processing) {
+    isRequestedAgain = true;
+    return;
+  }
+
+  processing = (async () => {
+    do {
+      isRequestedAgain = false;
+      await processPendingFileDeletions();
+    } while (isRequestedAgain);
+  })()
+    .catch(console.error)
+    .finally(() => {
+      processing = undefined;
+    });
+};

src/lib/server/services/index.ts

@@ -0,0 +1,10 @@
+export * as AuthService from "./auth";
+export * as CategoryService from "./category";
+export * as ClientService from "./client";
+export * as DirectoryService from "./directory";
+export * as FileService from "./file";
+export * as FileDeletionService from "./fileDeletion";
+export * as KeyService from "./key";
+export * as SearchService from "./search";
+export * as UploadService from "./upload";
+export * as UserService from "./user";

src/lib/server/services/key.ts

@@ -0,0 +1,50 @@
+import { ClientRepo, HskRepo, MekRepo } from "#lib/server/db/index.js";
+import { verifySignature } from "#lib/server/crypto.js";
+import { AppError } from "#lib/server/errors.js";
+
+export const getClientMeks = async (userId: number, clientId: number) => {
+  return await MekRepo.getAllValidClientMeks(userId, clientId);
+};
+
+const verifyClientEncMekSig = async (
+  userId: number,
+  clientId: number,
+  version: number,
+  encMek: string,
+  encMekSig: string,
+) => {
+  const userClient = await ClientRepo.getUserClientWithDetails(userId, clientId);
+  if (!userClient) {
+    throw new AppError("INTERNAL_SERVER_ERROR", "Invalid session id");
+  }
+
+  const data = JSON.stringify({ version, key: encMek });
+  return verifySignature(Buffer.from(data), encMekSig, userClient.sigPubKey);
+};
+
+export const registerInitialMek = async (
+  userId: number,
+  clientId: number,
+  encMek: string,
+  encMekSig: string,
+) => {
+  if (!(await verifyClientEncMekSig(userId, clientId, 1, encMek, encMekSig))) {
+    throw new AppError("BAD_REQUEST", "Invalid signature");
+  }
+
+  await MekRepo.registerInitialMek(userId, clientId, encMek, encMekSig);
+  await ClientRepo.setUserClientStateToActive(userId, clientId);
+};
+
+export const getHsks = async (userId: number) => {
+  return await HskRepo.getAllValidHsks(userId);
+};
+
+export const registerInitialHsk = async (
+  userId: number,
+  clientId: number,
+  mekVersion: number,
+  encHsk: string,
+) => {
+  await HskRepo.registerInitialHsk(userId, clientId, mekVersion, encHsk);
+};

src/lib/server/services/search.ts

@@ -0,0 +1,31 @@
+import { DirectoryRepo, FileRepo } from "#lib/server/db/index.js";
+
+export const search = async (
+  userId: number,
+  filters: {
+    ancestorId: DirectoryId;
+    inFavorites: boolean;
+    includeCategoryIds: number[];
+    excludeCategoryIds: number[];
+  },
+) => {
+  // Directories can't belong to categories
+  const hasCategoryFilter =
+    filters.includeCategoryIds.length > 0 || filters.excludeCategoryIds.length > 0;
+
+  const [directories, files] = await Promise.all([
+    !hasCategoryFilter
+      ? DirectoryRepo.searchDirectories(userId, {
+          parentId: filters.ancestorId,
+          inFavorites: filters.inFavorites,
+        })
+      : [],
+    FileRepo.searchFiles(userId, {
+      parentId: filters.ancestorId,
+      inFavorites: filters.inFavorites,
+      includeCategoryIds: filters.includeCategoryIds,
+      excludeCategoryIds: filters.excludeCategoryIds,
+    }),
+  ]);
+  return { directories, files };
+};

src/lib/server/services/upload.ts

@@ -1,11 +1,18 @@
-import { error } from "@sveltejs/kit";
-import { open } from "fs/promises";
-import { Readable } from "stream";
-import { ENCRYPTION_OVERHEAD, ENCRYPTED_CHUNK_SIZE } from "#lib/constants/index.js";
-import { UploadRepo } from "#lib/server/db/index.js";
-import { safeUnlink } from "#lib/server/modules/filesystem.js";
+import type { Readable } from "stream";
+import { db, FileRepo, MediaRepo, UploadRepo } from "#lib/server/db/index.js";
+import type { Ciphertext } from "#lib/server/db/schema/index.js";
+import { AppError } from "#lib/server/errors.js";
+import {
+  deleteFile,
+  hashFile,
+  reserveFile,
+  writeChunkAtOffset,
+} from "#lib/server/storage/index.js";
+import { ensureFreshDekVersion } from "./common";
+import { schedulePendingFileDeletions } from "./fileDeletion";
 
-const chunkLocks = new Set<string>();
+const UPLOAD_EXPIRES = 24 * 3600 * 1000; // 24 hours
+const THUMBNAIL_CHUNKS = 1; // Up to 4 MiB
 
 export const isChunkUploaded = (bitmap: Buffer, chunkIndex: number) => {
   chunkIndex -= 1;
@@ -13,93 +20,173 @@ export const isChunkUploaded = (bitmap: Buffer, chunkIndex: number) => {
   return !!byte && (byte & (1 << (chunkIndex % 8))) !== 0; // Postgres sucks
 };
 
-const writeChunkAtOffset = async (
-  path: string,
-  encChunkStream: Readable,
-  chunkIndex: number,
-  isLastChunk: boolean,
+const withReservedFile = async <T>(
+  kind: "library" | "thumbnails",
+  userId: number,
+  callback: (path: string) => Promise<T>,
 ) => {
-  const offset = (chunkIndex - 1) * ENCRYPTED_CHUNK_SIZE;
-  const file = await open(path, "r+");
-  let written = 0;
-
+  const path = await reserveFile(kind, userId);
   try {
-    for await (const chunk of encChunkStream) {
-      const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
-      written += buffer.length;
-      if (written > ENCRYPTED_CHUNK_SIZE) {
-        throw new Error("Invalid chunk size");
-      }
-
-      let chunkOffset = 0;
-      while (chunkOffset < buffer.length) {
-        const { bytesWritten } = await file.write(
-          buffer,
-          chunkOffset,
-          buffer.length - chunkOffset,
-          offset + written - buffer.length + chunkOffset,
-        );
-        if (bytesWritten <= 0) {
-          throw new Error("Failed to write chunk");
-        }
-        chunkOffset += bytesWritten;
-      }
-    }
-
-    if (
-      (!isLastChunk && written !== ENCRYPTED_CHUNK_SIZE) ||
-      (isLastChunk && (written <= ENCRYPTION_OVERHEAD || written > ENCRYPTED_CHUNK_SIZE))
-    ) {
-      throw new Error("Invalid chunk size");
-    }
-
-    if (isLastChunk) {
-      await file.truncate(offset + written);
-    }
-
-    return written;
-  } finally {
-    await file.close();
+    return await callback(path);
+  } catch (e) {
+    await deleteFile(path).catch(console.error);
+    throw e;
   }
 };
 
+export const startFileUpload = async (params: {
+  userId: number;
+  totalChunks: number;
+  parentId: DirectoryId;
+  mekVersion: number;
+  encDek: string;
+  dekVersion: Date;
+  hskVersion: number | null;
+  contentType: string;
+  encName: Ciphertext;
+  encCreatedAt: Ciphertext | null;
+  encLastModifiedAt: Ciphertext;
+}) => {
+  ensureFreshDekVersion(params.dekVersion);
+
+  return await withReservedFile("library", params.userId, async (path) => {
+    const id = crypto.randomUUID();
+    await UploadRepo.createFileUploadSession({
+      ...params,
+      id,
+      path,
+      expiresAt: new Date(Date.now() + UPLOAD_EXPIRES),
+    });
+    return { uploadId: id };
+  });
+};
+
+export const startFileThumbnailUpload = async (
+  userId: number,
+  fileId: number,
+  dekVersion: Date,
+) => {
+  return await withReservedFile("thumbnails", userId, async (path) => {
+    const id = crypto.randomUUID();
+    await UploadRepo.createThumbnailUploadSession({
+      id,
+      type: "thumbnail",
+      userId,
+      path,
+      totalChunks: THUMBNAIL_CHUNKS,
+      expiresAt: new Date(Date.now() + UPLOAD_EXPIRES),
+      fileId,
+      dekVersion,
+    });
+    return { uploadId: id };
+  });
+};
+
 export const uploadChunk = async (
   userId: number,
   sessionId: string,
   chunkIndex: number,
   encChunkStream: Readable,
 ) => {
-  const lockKey = `${sessionId}/${chunkIndex}`;
-  if (chunkLocks.has(lockKey)) {
-    error(409, "Chunk upload already in progress");
-  } else {
-    chunkLocks.add(lockKey);
+  const session = await UploadRepo.getUploadSession(sessionId, userId);
+  if (!session) {
+    throw new AppError("NOT_FOUND", "Invalid upload id");
+  } else if (chunkIndex > session.totalChunks) {
+    throw new AppError("BAD_REQUEST", "Invalid chunk index");
+  } else if (isChunkUploaded(session.bitmap, chunkIndex)) {
+    throw new AppError("CONFLICT", "Chunk already uploaded");
   }
 
-  try {
-    const session = await UploadRepo.getUploadSession(sessionId, userId);
-    if (!session) {
-      error(404, "Invalid upload id");
-    } else if (chunkIndex > session.totalChunks) {
-      error(400, "Invalid chunk index");
-    } else if (isChunkUploaded(session.bitmap, chunkIndex)) {
-      error(409, "Chunk already uploaded");
+  // Concurrent uploads of the same chunk may corrupt the file, which will be detected by the
+  // hash verification on completion
+  const isLastChunk = chunkIndex === session.totalChunks;
+  await writeChunkAtOffset(session.path, encChunkStream, chunkIndex, isLastChunk);
+  await UploadRepo.markChunkAsUploaded(sessionId, chunkIndex);
+};
+
+const getCompletedUploadSession = async <T extends "file" | "thumbnail">(
+  userId: number,
+  uploadId: string,
+  type: T,
+) => {
+  const session = await UploadRepo.getUploadSession(uploadId, userId);
+  if (session?.type !== type) {
+    throw new AppError("NOT_FOUND", "Invalid upload id");
+  } else if (session.uploadedChunks < session.totalChunks) {
+    throw new AppError("BAD_REQUEST", "Upload not completed");
+  }
+  return session as Extract<NonNullable<typeof session>, { type: T }>;
+};
+
+export const completeFileUpload = async (
+  userId: number,
+  uploadId: string,
+  contentHmac: string | null,
+  encContentHash: string,
+) => {
+  const session = await getCompletedUploadSession(userId, uploadId, "file");
+  if (!!session.hskVersion !== !!contentHmac) {
+    throw new AppError("BAD_REQUEST", "Invalid content HMAC");
+  }
+
+  const hash = await hashFile(session.path);
+  if (hash !== encContentHash) {
+    await UploadRepo.discardUploadSession(uploadId);
+    schedulePendingFileDeletions();
+    throw new AppError("CONFLICT", "Uploaded file corrupted");
+  }
+
+  return await db.transaction().execute(async (trx) => {
+    // Locking the session ensures that it is completed only once
+    const lockedSession = await UploadRepo.lockUploadSession(trx, uploadId, userId);
+    if (lockedSession.type !== "file") {
+      throw new AppError("NOT_FOUND", "Invalid upload id");
     }
 
-    const isLastChunk = chunkIndex === session.totalChunks;
-    await writeChunkAtOffset(session.path, encChunkStream, chunkIndex, isLastChunk);
-    await UploadRepo.markChunkAsUploaded(sessionId, chunkIndex);
-  } catch (e) {
-    if (e instanceof Error && e.message === "Invalid chunk size") {
-      error(400, "Invalid request body");
+    const { id: fileId } = await FileRepo.registerFile(trx, {
+      parentId: lockedSession.parentId,
+      userId,
+      path: lockedSession.path,
+      mekVersion: lockedSession.mekVersion,
+      encDek: lockedSession.encDek,
+      dekVersion: lockedSession.dekVersion,
+      hskVersion: lockedSession.hskVersion,
+      contentHmac,
+      contentType: lockedSession.contentType,
+      encContentIv: null,
+      encContentHash: hash,
+      encName: lockedSession.encName,
+      encCreatedAt: lockedSession.encCreatedAt,
+      encLastModifiedAt: lockedSession.encLastModifiedAt,
+    });
+    await UploadRepo.deleteUploadSession(trx, uploadId);
+    return { fileId };
+  });
+};
+
+export const completeFileThumbnailUpload = async (userId: number, uploadId: string) => {
+  await getCompletedUploadSession(userId, uploadId, "thumbnail");
+
+  await db.transaction().execute(async (trx) => {
+    const session = await UploadRepo.lockUploadSession(trx, uploadId, userId);
+    if (session.type !== "thumbnail") {
+      throw new AppError("NOT_FOUND", "Invalid upload id");
     }
-    throw e;
-  } finally {
-    chunkLocks.delete(lockKey);
-  }
+
+    await MediaRepo.updateFileThumbnail(
+      trx,
+      userId,
+      session.fileId,
+      session.dekVersion,
+      session.path,
+      null,
+    );
+    await UploadRepo.deleteUploadSession(trx, uploadId);
+  });
+  schedulePendingFileDeletions();
 };
 
 export const cleanupExpiredUploadSessions = async () => {
-  const paths = await UploadRepo.cleanupExpiredUploadSessions();
-  await Promise.all(paths.map(safeUnlink));
+  await UploadRepo.cleanupExpiredUploadSessions();
+  schedulePendingFileDeletions();
 };

src/lib/server/services/user.ts

@@ -0,0 +1,10 @@
+import { UserRepo } from "#lib/server/db/index.js";
+import { AppError } from "#lib/server/errors.js";
+
+export const getUser = async (userId: number) => {
+  const user = await UserRepo.getUser(userId);
+  if (!user) {
+    throw new AppError("INTERNAL_SERVER_ERROR", "Invalid session id");
+  }
+  return user;
+};