diff --git a/src/lib/server/db/error.ts b/src/lib/server/db/error.ts deleted file mode 100644 index 3fac93f..0000000 --- a/src/lib/server/db/error.ts +++ /dev/null @@ -1,41 +0,0 @@ -import { AppError, type ErrorCode } from "#lib/server/errors.ts"; - -const errorCodes = { - // Category - "Category not found": "NOT_FOUND", - // Challenge - "Challenge already registered": "CONFLICT", - // Client - "Public key(s) already registered": "CONFLICT", - "User client already exists": "CONFLICT", - // File - "Directory not found": "NOT_FOUND", - "Directory already favorited": "BAD_REQUEST", - "Directory not favorited": "BAD_REQUEST", - "File not found": "NOT_FOUND", - "File not found in category": "BAD_REQUEST", - "File already added to category": "BAD_REQUEST", - "File already favorited": "BAD_REQUEST", - "File not favorited": "BAD_REQUEST", - "Invalid DEK version": "CONFLICT", - // HSK - "HSK already registered": "CONFLICT", - "Inactive HSK version": "BAD_REQUEST", - // MEK - "MEK already registered": "CONFLICT", - "Inactive MEK version": "BAD_REQUEST", - // Session - "Session not found": "UNAUTHORIZED", - "Already logged in": "CONFLICT", - // Upload - "Upload session not found": "NOT_FOUND", -} as const satisfies Record; - -type IntegrityErrorMessage = keyof typeof errorCodes; - -export class IntegrityError extends AppError { - constructor(public override readonly message: IntegrityErrorMessage) { - super(errorCodes[message], message); - this.name = "IntegrityError"; - } -} diff --git a/src/lib/server/db/index.ts b/src/lib/server/db/index.ts index 74e37a0..533f2e2 100644 --- a/src/lib/server/db/index.ts +++ b/src/lib/server/db/index.ts @@ -11,6 +11,5 @@ export * as SessionRepo from "./repositories/session.ts"; export * as UploadRepo from "./repositories/upload.ts"; export * as UserRepo from "./repositories/user.ts"; -export * from "./error.ts"; export * from "./executor.ts"; export { default as db } from "./kysely.ts"; diff --git a/src/lib/server/db/repositories/category.ts b/src/lib/server/db/repositories/category.ts index 215b9df..ef0467e 100644 --- a/src/lib/server/db/repositories/category.ts +++ b/src/lib/server/db/repositories/category.ts @@ -1,8 +1,7 @@ import type { Selectable } from "kysely"; -import { IntegrityError } from "../error.ts"; -import db from "../kysely.ts"; +import type { Executor, Tx } from "../executor.ts"; +import { decodeParentId, encodeParentId, parentIdOperator } from "../queries.ts"; import type { CategoryTable, Ciphertext } from "../schema/index.ts"; -import { lockActiveMekVersion } from "./mek.ts"; export interface Category { id: number; @@ -23,7 +22,7 @@ type CategoryRow = Omit, "data_encryption_key_version" export const toCategory = (row: CategoryRow): Category => ({ id: row.id, - parentId: row.parent_id ?? "root", + parentId: decodeParentId(row.parent_id), userId: row.user_id, mekVersion: row.master_encryption_key_version, encDek: row.encrypted_data_encryption_key, @@ -31,70 +30,24 @@ export const toCategory = (row: CategoryRow): Category => ({ encName: row.encrypted_name, }); -export const ensureCategoryExists = async ( - trx: typeof db, - userId: number, - categoryId: CategoryId, -) => { - if (categoryId === "root") return; - - const category = await trx - .selectFrom("category") - .select("id") - .where("id", "=", categoryId) - .where("user_id", "=", userId) - .limit(1) - .forShare() - .executeTakeFirst(); - if (!category) { - throw new IntegrityError("Category not found"); - } +export const createCategory = async (executor: Executor, params: NewCategory) => { + const { id } = await executor + .insertInto("category") + .values({ + parent_id: encodeParentId(params.parentId), + user_id: params.userId, + master_encryption_key_version: params.mekVersion, + encrypted_data_encryption_key: params.encDek, + data_encryption_key_version: params.dekVersion, + encrypted_name: params.encName, + }) + .returning("id") + .executeTakeFirstOrThrow(); + return { id }; }; -export const registerCategory = async (params: NewCategory) => { - await db.transaction().execute(async (trx) => { - if ((await lockActiveMekVersion(trx, params.userId)) !== params.mekVersion) { - throw new IntegrityError("Inactive MEK version"); - } - - await ensureCategoryExists(trx, params.userId, params.parentId); - - const { categoryId } = await trx - .insertInto("category") - .values({ - parent_id: params.parentId !== "root" ? params.parentId : null, - user_id: params.userId, - master_encryption_key_version: params.mekVersion, - encrypted_data_encryption_key: params.encDek, - data_encryption_key_version: params.dekVersion, - encrypted_name: params.encName, - }) - .returning("id as categoryId") - .executeTakeFirstOrThrow(); - await trx - .insertInto("category_log") - .values({ - category_id: categoryId, - timestamp: new Date(), - action: "create", - new_name: params.encName, - }) - .execute(); - }); -}; - -export const getAllCategoriesByParent = async (userId: number, parentId: CategoryId) => { - let query = db.selectFrom("category").selectAll().where("user_id", "=", userId); - query = - parentId === "root" - ? query.where("parent_id", "is", null) - : query.where("parent_id", "=", parentId); - const categories = await query.execute(); - return categories.map(toCategory); -}; - -export const getCategory = async (userId: number, categoryId: number) => { - const category = await db +export const getCategory = async (executor: Executor, userId: number, categoryId: number) => { + const category = await executor .selectFrom("category") .selectAll() .where("id", "=", categoryId) @@ -104,52 +57,59 @@ export const getCategory = async (userId: number, categoryId: number) => { return category ? toCategory(category) : null; }; +// Locks the category in share mode so that it isn't deleted until the transaction ends. Returns +// whether it exists. +export const lockCategory = async (trx: Tx, userId: number, categoryId: number) => { + const category = await trx + .selectFrom("category") + .select("id") + .where("id", "=", categoryId) + .where("user_id", "=", userId) + .limit(1) + .forShare() + .executeTakeFirst(); + return !!category; +}; + +export const getAllCategoriesByParent = async ( + executor: Executor, + userId: number, + parentId: CategoryId, +) => { + const categories = await executor + .selectFrom("category") + .selectAll() + .where("user_id", "=", userId) + .where("parent_id", parentIdOperator(parentId), encodeParentId(parentId)) + .execute(); + return categories.map(toCategory); +}; + +// Returns whether the name has been changed, which fails if the category doesn't exist or its DEK +// version differs export const setCategoryEncName = async ( + executor: Executor, userId: number, categoryId: number, dekVersion: Date, encName: Ciphertext, ) => { - await db.transaction().execute(async (trx) => { - const category = await trx - .selectFrom("category") - .select("data_encryption_key_version") - .where("id", "=", categoryId) - .where("user_id", "=", userId) - .limit(1) - .forUpdate() - .executeTakeFirst(); - if (!category) { - throw new IntegrityError("Category not found"); - } else if (category.data_encryption_key_version.getTime() !== dekVersion.getTime()) { - throw new IntegrityError("Invalid DEK version"); - } - - await trx - .updateTable("category") - .set({ encrypted_name: encName }) - .where("id", "=", categoryId) - .where("user_id", "=", userId) - .execute(); - await trx - .insertInto("category_log") - .values({ - category_id: categoryId, - timestamp: new Date(), - action: "rename", - new_name: encName, - }) - .execute(); - }); + const { numUpdatedRows } = await executor + .updateTable("category") + .set({ encrypted_name: encName }) + .where("id", "=", categoryId) + .where("user_id", "=", userId) + .where("data_encryption_key_version", "=", dekVersion) + .executeTakeFirst(); + return numUpdatedRows > 0n; }; -export const unregisterCategory = async (userId: number, categoryId: number) => { - const res = await db +// Subcategories are deleted together by the database. Returns whether the category existed. +export const deleteCategory = async (executor: Executor, userId: number, categoryId: number) => { + const { numDeletedRows } = await executor .deleteFrom("category") .where("id", "=", categoryId) .where("user_id", "=", userId) .executeTakeFirst(); - if (res.numDeletedRows === 0n) { - throw new IntegrityError("Category not found"); - } + return numDeletedRows > 0n; }; diff --git a/src/lib/server/db/repositories/directory.ts b/src/lib/server/db/repositories/directory.ts index 3b51635..e416907 100644 --- a/src/lib/server/db/repositories/directory.ts +++ b/src/lib/server/db/repositories/directory.ts @@ -1,9 +1,12 @@ import type { Selectable } from "kysely"; -import { IntegrityError } from "../error.ts"; -import db from "../kysely.ts"; +import type { Executor, Tx } from "../executor.ts"; +import { + decodeParentId, + encodeParentId, + parentIdOperator, + withDirectoryTrees, +} from "../queries.ts"; import type { Ciphertext, DirectoryTable } from "../schema/index.ts"; -import { enqueueFileDeletions } from "./file-deletion.ts"; -import { lockActiveMekVersion } from "./mek.ts"; export interface Directory { id: number; @@ -16,9 +19,11 @@ export interface Directory { isFavorite: boolean; } +export type NewDirectory = Omit; + const toDirectory = (row: Selectable): Directory => ({ id: row.id, - parentId: row.parent_id ?? "root", + parentId: decodeParentId(row.parent_id), userId: row.user_id, mekVersion: row.master_encryption_key_version, encDek: row.encrypted_data_encryption_key, @@ -27,81 +32,24 @@ const toDirectory = (row: Selectable): Directory => ({ isFavorite: row.is_favorite, }); -export const ensureDirectoryExists = async ( - trx: typeof db, - userId: number, - directoryId: DirectoryId, -) => { - if (directoryId === "root") return; - - const directory = await trx - .selectFrom("directory") - .select("id") - .where("id", "=", directoryId) - .where("user_id", "=", userId) - .limit(1) - .forShare() - .executeTakeFirst(); - if (!directory) { - throw new IntegrityError("Directory not found"); - } +export const createDirectory = async (executor: Executor, params: NewDirectory) => { + const { id } = await executor + .insertInto("directory") + .values({ + parent_id: encodeParentId(params.parentId), + user_id: params.userId, + master_encryption_key_version: params.mekVersion, + encrypted_data_encryption_key: params.encDek, + data_encryption_key_version: params.dekVersion, + encrypted_name: params.encName, + }) + .returning("id") + .executeTakeFirstOrThrow(); + return { id }; }; -export const registerDirectory = async (params: Omit) => { - await db.transaction().execute(async (trx) => { - if ((await lockActiveMekVersion(trx, params.userId)) !== params.mekVersion) { - throw new IntegrityError("Inactive MEK version"); - } - - await ensureDirectoryExists(trx, params.userId, params.parentId); - - const { directoryId } = await trx - .insertInto("directory") - .values({ - parent_id: params.parentId !== "root" ? params.parentId : null, - user_id: params.userId, - master_encryption_key_version: params.mekVersion, - encrypted_data_encryption_key: params.encDek, - data_encryption_key_version: params.dekVersion, - encrypted_name: params.encName, - }) - .returning("id as directoryId") - .executeTakeFirstOrThrow(); - await trx - .insertInto("directory_log") - .values({ - directory_id: directoryId, - timestamp: new Date(), - action: "create", - new_name: params.encName, - }) - .execute(); - }); -}; - -export const getAllDirectoriesByParent = async (userId: number, parentId: DirectoryId) => { - const directories = await db - .selectFrom("directory") - .selectAll() - .where("user_id", "=", userId) - .$if(parentId === "root", (qb) => qb.where("parent_id", "is", null)) - .$if(parentId !== "root", (qb) => qb.where("parent_id", "=", parentId as number)) - .execute(); - return directories.map(toDirectory); -}; - -export const getAllFavoriteDirectories = async (userId: number) => { - const directories = await db - .selectFrom("directory") - .selectAll() - .where("user_id", "=", userId) - .where("is_favorite", "=", true) - .execute(); - return directories.map(toDirectory); -}; - -export const getDirectory = async (userId: number, directoryId: number) => { - const directory = await db +export const getDirectory = async (executor: Executor, userId: number, directoryId: number) => { + const directory = await executor .selectFrom("directory") .selectAll() .where("id", "=", directoryId) @@ -111,46 +59,57 @@ export const getDirectory = async (userId: number, directoryId: number) => { return directory ? toDirectory(directory) : null; }; -export const searchDirectories = async ( +// Locks the directory in share mode so that it isn't deleted until the transaction ends. Returns +// whether it exists. +export const lockDirectory = async (trx: Tx, userId: number, directoryId: number) => { + const directory = await trx + .selectFrom("directory") + .select("id") + .where("id", "=", directoryId) + .where("user_id", "=", userId) + .limit(1) + .forShare() + .executeTakeFirst(); + return !!directory; +}; + +export const getAllDirectoriesByParent = async ( + executor: Executor, userId: number, - filters: { - parentId: DirectoryId; - inFavorites: boolean; - }, + parentId: DirectoryId, ) => { - const directories = await db - .withRecursive("directory_tree", (db) => - db - .selectFrom("directory") - .select("id") - .where("user_id", "=", userId) - .$if(filters.parentId === "root", (qb) => qb.where((eb) => eb.lit(false))) // directory_tree will be empty if parentId is "root" - .$if(filters.parentId !== "root", (qb) => qb.where("id", "=", filters.parentId as number)) - .unionAll( - db - .selectFrom("directory as d") - .innerJoin("directory_tree as dt", "d.parent_id", "dt.id") - .select("d.id"), - ), - ) - .withRecursive("favorite_directory_tree", (db) => - db - .selectFrom("directory") - .select("id") - .where("user_id", "=", userId) - .$if(!filters.inFavorites, (qb) => qb.where((eb) => eb.lit(false))) // favorite_directory_tree will be empty if inFavorites is false - .$if(filters.inFavorites, (qb) => qb.where("is_favorite", "=", true)) - .unionAll((db) => - db - .selectFrom("directory as d") - .innerJoin("favorite_directory_tree as dt", "d.parent_id", "dt.id") - .select("d.id"), - ), - ) + const directories = await executor .selectFrom("directory") .selectAll() .where("user_id", "=", userId) - .$if(filters.parentId !== "root", (qb) => + .where("parent_id", parentIdOperator(parentId), encodeParentId(parentId)) + .execute(); + return directories.map(toDirectory); +}; + +export const getAllFavoriteDirectories = async (executor: Executor, userId: number) => { + const directories = await executor + .selectFrom("directory") + .selectAll() + .where("user_id", "=", userId) + .where("is_favorite", "=", true) + .execute(); + return directories.map(toDirectory); +}; + +export const searchDirectories = async ( + executor: Executor, + userId: number, + filters: { ancestorId: DirectoryId; inFavorites: boolean }, +) => { + const directories = await withDirectoryTrees(executor, userId, { + rootId: filters.ancestorId, + inFavorites: filters.inFavorites, + }) + .selectFrom("directory") + .selectAll() + .where("user_id", "=", userId) + .$if(filters.ancestorId !== "root", (qb) => qb.where((eb) => eb.exists(eb.selectFrom("directory_tree as dt").whereRef("dt.id", "=", "parent_id")), ), @@ -166,138 +125,59 @@ export const searchDirectories = async ( return directories.map(toDirectory); }; +// Returns whether the name has been changed, which fails if the directory doesn't exist or its DEK +// version differs export const setDirectoryEncName = async ( + executor: Executor, userId: number, directoryId: number, dekVersion: Date, encName: Ciphertext, ) => { - await db.transaction().execute(async (trx) => { - const directory = await trx - .selectFrom("directory") - .select("data_encryption_key_version") - .where("id", "=", directoryId) - .where("user_id", "=", userId) - .limit(1) - .forUpdate() - .executeTakeFirst(); - if (!directory) { - throw new IntegrityError("Directory not found"); - } else if (directory.data_encryption_key_version.getTime() !== dekVersion.getTime()) { - throw new IntegrityError("Invalid DEK version"); - } - - await trx - .updateTable("directory") - .set({ encrypted_name: encName }) - .where("id", "=", directoryId) - .where("user_id", "=", userId) - .execute(); - await trx - .insertInto("directory_log") - .values({ - directory_id: directoryId, - timestamp: new Date(), - action: "rename", - new_name: encName, - }) - .execute(); - }); + const { numUpdatedRows } = await executor + .updateTable("directory") + .set({ encrypted_name: encName }) + .where("id", "=", directoryId) + .where("user_id", "=", userId) + .where("data_encryption_key_version", "=", dekVersion) + .executeTakeFirst(); + return numUpdatedRows > 0n; }; +// Returns whether the state has been changed, which fails if the directory doesn't exist or is +// already in the state export const setDirectoryFavorite = async ( + executor: Executor, userId: number, directoryId: number, isFavorite: boolean, ) => { - await db.transaction().execute(async (trx) => { - const directory = await trx - .selectFrom("directory") - .select("is_favorite") - .where("id", "=", directoryId) - .where("user_id", "=", userId) - .limit(1) - .forUpdate() - .executeTakeFirst(); - if (!directory) { - throw new IntegrityError("Directory not found"); - } else if (directory.is_favorite === isFavorite) { - throw new IntegrityError( - isFavorite ? "Directory already favorited" : "Directory not favorited", - ); - } - - await trx - .updateTable("directory") - .set({ is_favorite: isFavorite }) - .where("id", "=", directoryId) - .where("user_id", "=", userId) - .execute(); - await trx - .insertInto("directory_log") - .values({ - directory_id: directoryId, - timestamp: new Date(), - action: isFavorite ? "add-to-favorites" : "remove-from-favorites", - }) - .execute(); - }); + const { numUpdatedRows } = await executor + .updateTable("directory") + .set({ is_favorite: isFavorite }) + .where("id", "=", directoryId) + .where("user_id", "=", userId) + .where("is_favorite", "!=", isFavorite) + .executeTakeFirst(); + return numUpdatedRows > 0n; }; -export const unregisterDirectory = async (userId: number, directoryId: number) => { - return await db.transaction().execute(async (trx) => { - const directories = await trx - .withRecursive("directory_tree", (db) => - db - .selectFrom("directory") - .select("id") - .where("id", "=", directoryId) - .where("user_id", "=", userId) - .unionAll( - db - .selectFrom("directory as d") - .innerJoin("directory_tree as dt", "d.parent_id", "dt.id") - .select("d.id"), - ), - ) - .selectFrom("directory") - .select("id") - .where("id", "in", (eb) => eb.selectFrom("directory_tree").select("id")) - .forUpdate() - .execute(); - if (directories.length === 0) { - throw new IntegrityError("Directory not found"); - } +// Locks the directory and its descendants, and returns their ids (empty if it doesn't exist) +export const lockDirectoryTree = async (trx: Tx, userId: number, directoryId: number) => { + const directories = await withDirectoryTrees(trx, userId, { + rootId: directoryId, + inFavorites: false, + }) + .selectFrom("directory") + .select("id") + .where("id", "in", (eb) => eb.selectFrom("directory_tree").select("id")) + .forUpdate() + .execute(); + return directories.map(({ id }) => id); +}; - const directoryIds = directories.map(({ id }) => id); - const files = await trx - .selectFrom("file") - .leftJoin("thumbnail", "file.id", "thumbnail.file_id") - .select(["file.id", "file.path", "thumbnail.path as thumbnailPath"]) - .where("file.parent_id", "in", directoryIds) - .forUpdate("file") - .execute(); - const fileIds = files.map(({ id }) => id); - - const uploadSessions = await trx - .deleteFrom("upload_session") - .where((eb) => - eb.or([ - eb("parent_id", "in", directoryIds), - ...(fileIds.length > 0 ? [eb("file_id", "in", fileIds)] : []), - ]), - ) - .returning("path") - .execute(); - if (fileIds.length > 0) { - await trx.deleteFrom("file").where("id", "in", fileIds).execute(); - } - await trx.deleteFrom("directory").where("id", "in", directoryIds).execute(); - - await enqueueFileDeletions(trx, [ - ...files.flatMap(({ path, thumbnailPath }) => [path, thumbnailPath]), - ...uploadSessions.map(({ path }) => path), - ]); - return { deletedFileIds: fileIds }; - }); +export const deleteDirectories = async (executor: Executor, directoryIds: number[]) => { + if (directoryIds.length > 0) { + await executor.deleteFrom("directory").where("id", "in", directoryIds).execute(); + } }; diff --git a/src/lib/server/db/repositories/file-deletion.ts b/src/lib/server/db/repositories/file-deletion.ts index 995dad2..0a79ecf 100644 --- a/src/lib/server/db/repositories/file-deletion.ts +++ b/src/lib/server/db/repositories/file-deletion.ts @@ -1,19 +1,21 @@ -import db from "../kysely.ts"; +import type { Executor } from "../executor.ts"; +// Paths are enqueued in the transaction which deletes the files from the database, and the files +// are deleted from the storage after it is committed export const enqueueFileDeletions = async ( - trx: typeof db, + executor: Executor, paths: (string | null | undefined)[], ) => { const values = paths .filter((path) => !!path) .map((path) => ({ path: path!, enqueued_at: new Date() })); if (values.length > 0) { - await trx.insertInto("file_deletion_queue").values(values).execute(); + await executor.insertInto("file_deletion_queue").values(values).execute(); } }; -export const getPendingFileDeletions = async (limit: number) => { - return await db +export const getPendingFileDeletions = async (executor: Executor, limit: number) => { + return await executor .selectFrom("file_deletion_queue") .select(["id", "path"]) .orderBy("id") @@ -21,8 +23,8 @@ export const getPendingFileDeletions = async (limit: number) => { .execute(); }; -export const completeFileDeletions = async (ids: number[]) => { +export const completeFileDeletions = async (executor: Executor, ids: number[]) => { if (ids.length > 0) { - await db.deleteFrom("file_deletion_queue").where("id", "in", ids).execute(); + await executor.deleteFrom("file_deletion_queue").where("id", "in", ids).execute(); } }; diff --git a/src/lib/server/db/repositories/file.ts b/src/lib/server/db/repositories/file.ts index 1d7c7dd..52b6c2d 100644 --- a/src/lib/server/db/repositories/file.ts +++ b/src/lib/server/db/repositories/file.ts @@ -1,11 +1,14 @@ import { sql, type Selectable } from "kysely"; import { jsonArrayFrom } from "kysely/helpers/postgres"; -import pg from "pg"; -import { IntegrityError } from "../error.ts"; -import db from "../kysely.ts"; +import type { Executor, Tx } from "../executor.ts"; +import { + decodeParentId, + encodeParentId, + parentIdOperator, + withDirectoryTrees, +} from "../queries.ts"; import type { Ciphertext, FileTable } from "../schema/index.ts"; import { toCategory } from "./category.ts"; -import { enqueueFileDeletions } from "./file-deletion.ts"; export interface File { id: number; @@ -26,9 +29,11 @@ export interface File { isFavorite: boolean; } +export type NewFile = Omit; + const toFile = (row: Selectable): File => ({ id: row.id, - parentId: row.parent_id ?? "root", + parentId: decodeParentId(row.parent_id), userId: row.user_id, path: row.path, mekVersion: row.master_encryption_key_version, @@ -45,15 +50,11 @@ const toFile = (row: Selectable): File => ({ isFavorite: row.is_favorite, }); -export const registerFile = async (trx: typeof db, params: Omit) => { - if ((params.hskVersion && !params.contentHmac) || (!params.hskVersion && params.contentHmac)) { - throw new Error("Invalid arguments"); - } - - const { fileId } = await trx +export const createFile = async (executor: Executor, params: NewFile) => { + const { id } = await executor .insertInto("file") .values({ - parent_id: params.parentId !== "root" ? params.parentId : null, + parent_id: encodeParentId(params.parentId), user_id: params.userId, path: params.path, master_encryption_key_version: params.mekVersion, @@ -68,46 +69,101 @@ export const registerFile = async (trx: typeof db, params: Omit { - const files = await db +export const getFile = async (executor: Executor, userId: number, fileId: number) => { + const file = await executor + .selectFrom("file") + .selectAll() + .where("id", "=", fileId) + .where("user_id", "=", userId) + .limit(1) + .executeTakeFirst(); + return file ? toFile(file) : null; +}; + +// Locks the file in share mode so that it isn't deleted until the transaction ends. Returns +// whether it exists. +export const lockFile = async (trx: Tx, userId: number, fileId: number) => { + const file = await trx + .selectFrom("file") + .select("id") + .where("id", "=", fileId) + .where("user_id", "=", userId) + .limit(1) + .forShare() + .executeTakeFirst(); + return !!file; +}; + +// Locks the file in share mode so that its DEK isn't changed until the transaction ends. Returns +// its DEK version if it exists. +export const lockFileDekVersion = async (trx: Tx, userId: number, fileId: number) => { + const file = await trx + .selectFrom("file") + .select("data_encryption_key_version") + .where("id", "=", fileId) + .where("user_id", "=", userId) + .limit(1) + .forShare() + .executeTakeFirst(); + return file?.data_encryption_key_version ?? null; +}; + +// Locks the files to delete them, and selects the paths of their contents and thumbnails +const selectFilesForDeletion = (trx: Tx) => + trx + .selectFrom("file") + .leftJoin("thumbnail", "file.id", "thumbnail.file_id") + .select(["file.id", "file.path", "thumbnail.path as thumbnailPath"]) + .forUpdate("file"); + +export const lockFileForDeletion = async (trx: Tx, userId: number, fileId: number) => { + return await selectFilesForDeletion(trx) + .where("file.id", "=", fileId) + .where("file.user_id", "=", userId) + .executeTakeFirst(); +}; + +export const lockFilesInDirectoriesForDeletion = async (trx: Tx, directoryIds: number[]) => { + if (directoryIds.length === 0) return []; + return await selectFilesForDeletion(trx).where("file.parent_id", "in", directoryIds).execute(); +}; + +export const getAllFilesByParent = async ( + executor: Executor, + userId: number, + parentId: DirectoryId, +) => { + const files = await executor .selectFrom("file") .selectAll() .where("user_id", "=", userId) - .$if(parentId === "root", (qb) => qb.where("parent_id", "is", null)) - .$if(parentId !== "root", (qb) => qb.where("parent_id", "=", parentId as number)) + .where("parent_id", parentIdOperator(parentId), encodeParentId(parentId)) .execute(); return files.map(toFile); }; +// Files of subcategories are included if `recurse`, which are marked as recursive export const getAllFilesByCategory = async ( + executor: Executor, userId: number, categoryId: number, recurse: boolean, ) => { - const files = await db - .withRecursive("category_tree", (db) => - db + const files = await executor + .withRecursive("category_tree", (qc) => + qc .selectFrom("category") .select(["id", sql`0`.as("depth")]) .where("id", "=", categoryId) .where("user_id", "=", userId) .$if(recurse, (qb) => - qb.unionAll((db) => - db + qb.unionAll((qc) => + qc .selectFrom("category") .innerJoin("category_tree", "category.parent_id", "category_tree.id") .select(["category.id", sql`depth + 1`.as("depth")]), @@ -129,13 +185,17 @@ export const getAllFilesByCategory = async ( })); }; -export const getAllFileIds = async (userId: number) => { - const files = await db.selectFrom("file").select("id").where("user_id", "=", userId).execute(); +export const getAllFileIds = async (executor: Executor, userId: number) => { + const files = await executor + .selectFrom("file") + .select("id") + .where("user_id", "=", userId) + .execute(); return files.map(({ id }) => id); }; -export const getFilesWithoutThumbnail = async (userId: number, limit: number = 100) => { - const files = await db +export const getFilesWithoutThumbnail = async (executor: Executor, userId: number, limit = 100) => { + const files = await executor .selectFrom("file") .selectAll() .where("user_id", "=", userId) @@ -159,11 +219,12 @@ export const getFilesWithoutThumbnail = async (userId: number, limit: number = 1 }; export const getAllFileIdsByContentHmac = async ( + executor: Executor, userId: number, hskVersion: number, contentHmac: string, ) => { - const files = await db + const files = await executor .selectFrom("file") .select("id") .where("user_id", "=", userId) @@ -173,19 +234,12 @@ export const getAllFileIdsByContentHmac = async ( return files.map(({ id }) => id); }; -export const getFile = async (userId: number, fileId: number) => { - const file = await db - .selectFrom("file") - .selectAll() - .where("id", "=", fileId) - .where("user_id", "=", userId) - .limit(1) - .executeTakeFirst(); - return file ? toFile(file) : null; -}; - -export const getFilesWithCategories = async (userId: number, fileIds: number[]) => { - const files = await db +export const getFilesWithCategories = async ( + executor: Executor, + userId: number, + fileIds: number[], +) => { + const files = await executor .selectFrom("file") .selectAll() .select((eb) => @@ -206,8 +260,8 @@ export const getFilesWithCategories = async (userId: number, fileIds: number[]) })); }; -export const getAllFavoriteFiles = async (userId: number) => { - const files = await db +export const getAllFavoriteFiles = async (executor: Executor, userId: number) => { + const files = await executor .selectFrom("file") .selectAll() .where("user_id", "=", userId) @@ -217,64 +271,40 @@ export const getAllFavoriteFiles = async (userId: number) => { }; export const searchFiles = async ( + executor: Executor, userId: number, filters: { - parentId: DirectoryId; + ancestorId: DirectoryId; inFavorites: boolean; includeCategoryIds: number[]; excludeCategoryIds: number[]; }, ) => { - const baseQuery = db - .withRecursive("directory_tree", (db) => - db - .selectFrom("directory") - .select("id") - .where("user_id", "=", userId) - .$if(filters.parentId === "root", (qb) => qb.where((eb) => eb.lit(false))) // directory_tree will be empty if parentId is "root" - .$if(filters.parentId !== "root", (qb) => qb.where("id", "=", filters.parentId as number)) - .unionAll( - db - .selectFrom("directory as d") - .innerJoin("directory_tree as dt", "d.parent_id", "dt.id") - .select("d.id"), - ), - ) - .withRecursive("favorite_directory_tree", (db) => - db - .selectFrom("directory") - .select("id") - .where("user_id", "=", userId) - .$if(!filters.inFavorites, (qb) => qb.where((eb) => eb.lit(false))) // favorite_directory_tree will be empty if inFavorites is false - .$if(filters.inFavorites, (qb) => qb.where("is_favorite", "=", true)) - .unionAll((db) => - db - .selectFrom("directory as d") - .innerJoin("favorite_directory_tree as dt", "d.parent_id", "dt.id") - .select("d.id"), - ), - ) - .withRecursive("include_category_tree", (db) => - db + const baseQuery = withDirectoryTrees(executor, userId, { + rootId: filters.ancestorId, + inFavorites: filters.inFavorites, + }) + .withRecursive("include_category_tree", (qc) => + qc .selectFrom("category") .select(["id", "id as root_id"]) .where("id", "=", (eb) => eb.fn.any(eb.val(filters.includeCategoryIds))) .where("user_id", "=", userId) - .unionAll( - db + .unionAll((qc) => + qc .selectFrom("category as c") .innerJoin("include_category_tree as ct", "c.parent_id", "ct.id") .select(["c.id", "ct.root_id"]), ), ) - .withRecursive("exclude_category_tree", (db) => - db + .withRecursive("exclude_category_tree", (qc) => + qc .selectFrom("category") .select("id") .where("id", "=", (eb) => eb.fn.any(eb.val(filters.excludeCategoryIds))) .where("user_id", "=", userId) - .unionAll((db) => - db + .unionAll((qc) => + qc .selectFrom("category as c") .innerJoin("exclude_category_tree as ct", "c.parent_id", "ct.id") .select("c.id"), @@ -283,7 +313,7 @@ export const searchFiles = async ( .selectFrom("file") .selectAll("file") .where("user_id", "=", userId) - .$if(filters.parentId !== "root", (qb) => + .$if(filters.ancestorId !== "root", (qb) => qb.where((eb) => eb.exists(eb.selectFrom("directory_tree as dt").whereRef("dt.id", "=", "file.parent_id")), ), @@ -310,6 +340,7 @@ export const searchFiles = async ( ), ), ); + // Files must belong to all the included categories (or their subcategories) const files = filters.includeCategoryIds.length > 0 ? await baseQuery @@ -330,100 +361,73 @@ export const searchFiles = async ( return files.map(toFile); }; +// Returns whether the name has been changed, which fails if the file doesn't exist or its DEK +// version differs export const setFileEncName = async ( + executor: Executor, userId: number, fileId: number, dekVersion: Date, encName: Ciphertext, ) => { - await db.transaction().execute(async (trx) => { - const file = await trx - .selectFrom("file") - .select("data_encryption_key_version") - .where("id", "=", fileId) - .where("user_id", "=", userId) - .limit(1) - .forUpdate() - .executeTakeFirst(); - if (!file) { - throw new IntegrityError("File not found"); - } else if (file.data_encryption_key_version.getTime() !== dekVersion.getTime()) { - throw new IntegrityError("Invalid DEK version"); - } - - await trx - .updateTable("file") - .set({ encrypted_name: encName }) - .where("id", "=", fileId) - .where("user_id", "=", userId) - .execute(); - await trx - .insertInto("file_log") - .values({ - file_id: fileId, - timestamp: new Date(), - action: "rename", - new_name: encName, - }) - .execute(); - }); + const { numUpdatedRows } = await executor + .updateTable("file") + .set({ encrypted_name: encName }) + .where("id", "=", fileId) + .where("user_id", "=", userId) + .where("data_encryption_key_version", "=", dekVersion) + .executeTakeFirst(); + return numUpdatedRows > 0n; }; -export const unregisterFile = async (userId: number, fileId: number) => { - await db.transaction().execute(async (trx) => { - const file = await trx - .selectFrom("file") - .leftJoin("thumbnail", "file.id", "thumbnail.file_id") - .select(["file.path", "thumbnail.path as thumbnailPath"]) - .where("file.id", "=", fileId) - .where("file.user_id", "=", userId) - .forUpdate("file") - .executeTakeFirst(); - if (!file) { - throw new IntegrityError("File not found"); - } - - const uploadSessions = await trx - .deleteFrom("upload_session") - .where("file_id", "=", fileId) - .returning("path") - .execute(); - await trx.deleteFrom("file").where("id", "=", fileId).execute(); - await enqueueFileDeletions(trx, [ - file.path, - file.thumbnailPath, - ...uploadSessions.map(({ path }) => path), - ]); - }); +// Returns whether the state has been changed, which fails if the file doesn't exist or is already +// in the state +export const setFileFavorite = async ( + executor: Executor, + userId: number, + fileId: number, + isFavorite: boolean, +) => { + const { numUpdatedRows } = await executor + .updateTable("file") + .set({ is_favorite: isFavorite }) + .where("id", "=", fileId) + .where("user_id", "=", userId) + .where("is_favorite", "!=", isFavorite) + .executeTakeFirst(); + return numUpdatedRows > 0n; }; -export const addFileToCategory = async (fileId: number, categoryId: number) => { - await db.transaction().execute(async (trx) => { - try { - await trx - .insertInto("file_category") - .values({ file_id: fileId, category_id: categoryId }) - .execute(); - await trx - .insertInto("file_log") - .values({ - file_id: fileId, - timestamp: new Date(), - action: "add-to-category", - category_id: categoryId, - }) - .execute(); - } catch (e) { - if (e instanceof pg.DatabaseError && e.code === "23505") { - throw new IntegrityError("File already added to category"); - } - throw e; - } - }); +export const deleteFiles = async (executor: Executor, fileIds: number[]) => { + if (fileIds.length > 0) { + await executor.deleteFrom("file").where("id", "in", fileIds).execute(); + } }; -export const getAllFileCategories = async (fileId: number) => { - const categories = await db +// Violates `file_category_pk` if the file is already in the category +export const addFileToCategory = async (executor: Executor, fileId: number, categoryId: number) => { + await executor + .insertInto("file_category") + .values({ file_id: fileId, category_id: categoryId }) + .execute(); +}; + +// Returns whether the file has been removed, which fails if it isn't in the category +export const removeFileFromCategory = async ( + executor: Executor, + fileId: number, + categoryId: number, +) => { + const { numDeletedRows } = await executor + .deleteFrom("file_category") + .where("file_id", "=", fileId) + .where("category_id", "=", categoryId) + .executeTakeFirst(); + return numDeletedRows > 0n; +}; + +export const getAllFileCategories = async (executor: Executor, fileId: number) => { + const categories = await executor .selectFrom("file_category") .innerJoin("category", "file_category.category_id", "category.id") .selectAll("category") @@ -431,59 +435,3 @@ export const getAllFileCategories = async (fileId: number) => { .execute(); return categories.map(toCategory); }; - -export const removeFileFromCategory = async (fileId: number, categoryId: number) => { - await db.transaction().execute(async (trx) => { - const res = await trx - .deleteFrom("file_category") - .where("file_id", "=", fileId) - .where("category_id", "=", categoryId) - .executeTakeFirst(); - if (res.numDeletedRows === 0n) { - throw new IntegrityError("File not found in category"); - } - - await trx - .insertInto("file_log") - .values({ - file_id: fileId, - timestamp: new Date(), - action: "remove-from-category", - category_id: categoryId, - }) - .execute(); - }); -}; - -export const setFileFavorite = async (userId: number, fileId: number, isFavorite: boolean) => { - await db.transaction().execute(async (trx) => { - const file = await trx - .selectFrom("file") - .select("is_favorite") - .where("id", "=", fileId) - .where("user_id", "=", userId) - .limit(1) - .forUpdate() - .executeTakeFirst(); - if (!file) { - throw new IntegrityError("File not found"); - } else if (file.is_favorite === isFavorite) { - throw new IntegrityError(isFavorite ? "File already favorited" : "File not favorited"); - } - - await trx - .updateTable("file") - .set({ is_favorite: isFavorite }) - .where("id", "=", fileId) - .where("user_id", "=", userId) - .execute(); - await trx - .insertInto("file_log") - .values({ - file_id: fileId, - timestamp: new Date(), - action: isFavorite ? "add-to-favorites" : "remove-from-favorites", - }) - .execute(); - }); -}; diff --git a/src/lib/server/db/repositories/media.ts b/src/lib/server/db/repositories/media.ts index c2e1d4c..efc87b3 100644 --- a/src/lib/server/db/repositories/media.ts +++ b/src/lib/server/db/repositories/media.ts @@ -1,86 +1,60 @@ -import type { NotNull } from "kysely"; -import { IntegrityError } from "../error.ts"; -import db from "../kysely.ts"; -import { enqueueFileDeletions } from "./file-deletion.ts"; +import type { Executor, Tx } from "../executor.ts"; -interface Thumbnail { +interface FileThumbnail { id: number; + fileId: number; path: string; updatedAt: Date; encContentIv: string | null; } -interface FileThumbnail extends Thumbnail { - fileId: number; -} - -export const updateFileThumbnail = async ( - trx: typeof db, - userId: number, - fileId: number, - dekVersion: Date, - path: string, - encContentIv: string | null, -) => { - const file = await trx - .selectFrom("file") - .select("data_encryption_key_version") - .where("id", "=", fileId) - .where("user_id", "=", userId) - .limit(1) - .forUpdate() - .executeTakeFirst(); - if (!file) { - throw new IntegrityError("File not found"); - } else if (file.data_encryption_key_version.getTime() !== dekVersion.getTime()) { - throw new IntegrityError("Invalid DEK version"); - } - - const thumbnail = await trx - .selectFrom("thumbnail") - .select("path as oldPath") - .where("file_id", "=", fileId) - .limit(1) - .forUpdate() - .executeTakeFirst(); - const now = new Date(); - - await trx - .insertInto("thumbnail") - .values({ - file_id: fileId, - path, - updated_at: now, - encrypted_content_iv: encContentIv, - }) - .onConflict((oc) => - oc.column("file_id").doUpdateSet({ - path, - updated_at: now, - encrypted_content_iv: encContentIv, - }), - ) - .execute(); - await enqueueFileDeletions(trx, [thumbnail?.oldPath]); -}; - -export const getFileThumbnail = async (userId: number, fileId: number) => { - const thumbnail = await db +export const getFileThumbnail = async (executor: Executor, userId: number, fileId: number) => { + const thumbnail = await executor .selectFrom("thumbnail") .innerJoin("file", "thumbnail.file_id", "file.id") - .selectAll("thumbnail") + .select([ + "thumbnail.id", + "thumbnail.path", + "thumbnail.updated_at", + "thumbnail.encrypted_content_iv", + ]) .where("file.id", "=", fileId) .where("file.user_id", "=", userId) - .$narrowType<{ file_id: NotNull }>() .limit(1) .executeTakeFirst(); return thumbnail ? ({ id: thumbnail.id, - fileId: thumbnail.file_id, + fileId, path: thumbnail.path, - encContentIv: thumbnail.encrypted_content_iv, updatedAt: thumbnail.updated_at, + encContentIv: thumbnail.encrypted_content_iv, } satisfies FileThumbnail) : null; }; + +// Locks the thumbnail of the file to replace it, and returns its path if any +export const lockFileThumbnailPath = async (trx: Tx, fileId: number) => { + const thumbnail = await trx + .selectFrom("thumbnail") + .select("path") + .where("file_id", "=", fileId) + .limit(1) + .forUpdate() + .executeTakeFirst(); + return thumbnail?.path ?? null; +}; + +export const upsertFileThumbnail = async ( + executor: Executor, + fileId: number, + path: string, + encContentIv: string | null, +) => { + const values = { path, updated_at: new Date(), encrypted_content_iv: encContentIv }; + await executor + .insertInto("thumbnail") + .values({ file_id: fileId, ...values }) + .onConflict((oc) => oc.column("file_id").doUpdateSet(values)) + .execute(); +}; diff --git a/src/lib/server/db/repositories/upload.ts b/src/lib/server/db/repositories/upload.ts index 002bb28..8d8bf69 100644 --- a/src/lib/server/db/repositories/upload.ts +++ b/src/lib/server/db/repositories/upload.ts @@ -1,11 +1,7 @@ import { sql, type Selectable } from "kysely"; -import { IntegrityError } from "../error.ts"; -import db from "../kysely.ts"; +import type { Executor, Tx } from "../executor.ts"; +import { decodeParentId, encodeParentId } from "../queries.ts"; import type { Ciphertext, UploadSessionTable } from "../schema/index.ts"; -import { ensureDirectoryExists } from "./directory.ts"; -import { enqueueFileDeletions } from "./file-deletion.ts"; -import { lockActiveHskVersion } from "./hsk.ts"; -import { lockActiveMekVersion } from "./mek.ts"; interface BaseUploadSession { id: string; @@ -36,93 +32,72 @@ interface ThumbnailUploadSession extends BaseUploadSession { dekVersion: Date; } +const createBitmap = (totalChunks: number) => Buffer.alloc(Math.ceil(totalChunks / 8)); + export const createFileUploadSession = async ( + executor: Executor, params: Omit, ) => { - await db.transaction().execute(async (trx) => { - if ((await lockActiveMekVersion(trx, params.userId)) !== params.mekVersion) { - throw new IntegrityError("Inactive MEK version"); - } - - if (params.hskVersion) { - if ((await lockActiveHskVersion(trx, params.userId)) !== params.hskVersion) { - throw new IntegrityError("Inactive HSK version"); - } - } - - await ensureDirectoryExists(trx, params.userId, params.parentId); - - await trx - .insertInto("upload_session") - .values({ - id: params.id, - type: "file", - user_id: params.userId, - path: params.path, - bitmap: Buffer.alloc(Math.ceil(params.totalChunks / 8)), - total_chunks: params.totalChunks, - expires_at: params.expiresAt, - parent_id: params.parentId !== "root" ? params.parentId : null, - master_encryption_key_version: params.mekVersion, - encrypted_data_encryption_key: params.encDek, - data_encryption_key_version: params.dekVersion, - hmac_secret_key_version: params.hskVersion, - content_type: params.contentType, - encrypted_name: params.encName, - encrypted_created_at: params.encCreatedAt, - encrypted_last_modified_at: params.encLastModifiedAt, - }) - .execute(); - }); + await executor + .insertInto("upload_session") + .values({ + id: params.id, + type: "file", + user_id: params.userId, + path: params.path, + bitmap: createBitmap(params.totalChunks), + total_chunks: params.totalChunks, + expires_at: params.expiresAt, + parent_id: encodeParentId(params.parentId), + master_encryption_key_version: params.mekVersion, + encrypted_data_encryption_key: params.encDek, + data_encryption_key_version: params.dekVersion, + hmac_secret_key_version: params.hskVersion, + content_type: params.contentType, + encrypted_name: params.encName, + encrypted_created_at: params.encCreatedAt, + encrypted_last_modified_at: params.encLastModifiedAt, + }) + .execute(); }; export const createThumbnailUploadSession = async ( - params: Omit, + executor: Executor, + params: Omit, ) => { - await db.transaction().execute(async (trx) => { - const file = await trx - .selectFrom("file") - .select("data_encryption_key_version") - .where("id", "=", params.fileId) - .where("user_id", "=", params.userId) - .limit(1) - .forUpdate() - .executeTakeFirst(); - if (!file) { - throw new IntegrityError("File not found"); - } else if (file.data_encryption_key_version.getTime() !== params.dekVersion.getTime()) { - throw new IntegrityError("Invalid DEK version"); - } - - await trx - .insertInto("upload_session") - .values({ - id: params.id, - type: params.type, - user_id: params.userId, - path: params.path, - bitmap: Buffer.alloc(Math.ceil(params.totalChunks / 8)), - total_chunks: params.totalChunks, - expires_at: params.expiresAt, - file_id: params.fileId, - data_encryption_key_version: params.dekVersion, - }) - .execute(); - }); + await executor + .insertInto("upload_session") + .values({ + id: params.id, + type: "thumbnail", + user_id: params.userId, + path: params.path, + bitmap: createBitmap(params.totalChunks), + total_chunks: params.totalChunks, + expires_at: params.expiresAt, + file_id: params.fileId, + data_encryption_key_version: params.dekVersion, + }) + .execute(); }; -const toUploadSession = (session: Selectable) => { +const toUploadSession = ( + session: Selectable, +): FileUploadSession | ThumbnailUploadSession => { + const base = { + id: session.id, + userId: session.user_id, + path: session.path, + bitmap: session.bitmap, + totalChunks: session.total_chunks, + uploadedChunks: session.uploaded_chunks, + expiresAt: session.expires_at, + }; if (session.type === "file") { return { + ...base, type: "file", - id: session.id, - userId: session.user_id, - path: session.path, - bitmap: session.bitmap, - totalChunks: session.total_chunks, - uploadedChunks: session.uploaded_chunks, - expiresAt: session.expires_at, - parentId: session.parent_id ?? "root", + parentId: decodeParentId(session.parent_id), mekVersion: session.master_encryption_key_version!, encDek: session.encrypted_data_encryption_key!, dekVersion: session.data_encryption_key_version!, @@ -131,54 +106,45 @@ const toUploadSession = (session: Selectable) => { encName: session.encrypted_name!, encCreatedAt: session.encrypted_created_at, encLastModifiedAt: session.encrypted_last_modified_at!, - } satisfies FileUploadSession; + }; } else { return { - type: session.type, - id: session.id, - userId: session.user_id, - path: session.path, - bitmap: session.bitmap, - totalChunks: session.total_chunks, - uploadedChunks: session.uploaded_chunks, - expiresAt: session.expires_at, + ...base, + type: "thumbnail", fileId: session.file_id!, dekVersion: session.data_encryption_key_version!, - } satisfies ThumbnailUploadSession; + }; } }; -export const getUploadSession = async (sessionId: string, userId: number) => { - const session = await db +const selectActiveUploadSession = (executor: Executor, sessionId: string, userId: number) => + executor .selectFrom("upload_session") .selectAll() .where("id", "=", sessionId) .where("user_id", "=", userId) .where("expires_at", ">", new Date()) - .limit(1) - .executeTakeFirst(); + .limit(1); + +export const getUploadSession = async (executor: Executor, sessionId: string, userId: number) => { + const session = await selectActiveUploadSession(executor, sessionId, userId).executeTakeFirst(); return session ? toUploadSession(session) : null; }; // Locks the session until the transaction ends so that it can be completed only once -export const lockUploadSession = async (trx: typeof db, sessionId: string, userId: number) => { - const session = await trx - .selectFrom("upload_session") - .selectAll() - .where("id", "=", sessionId) - .where("user_id", "=", userId) - .where("expires_at", ">", new Date()) - .limit(1) +export const lockUploadSession = async (trx: Tx, sessionId: string, userId: number) => { + const session = await selectActiveUploadSession(trx, sessionId, userId) .forUpdate() .executeTakeFirst(); - if (!session) { - throw new IntegrityError("Upload session not found"); - } - return toUploadSession(session); + return session ? toUploadSession(session) : null; }; -export const markChunkAsUploaded = async (sessionId: string, chunkIndex: number) => { - await db +export const markChunkAsUploaded = async ( + executor: Executor, + sessionId: string, + chunkIndex: number, +) => { + await executor .updateTable("upload_session") .set({ bitmap: sql`set_bit(${sql.ref("bitmap")}, ${chunkIndex - 1}, 1)`, @@ -187,34 +153,42 @@ export const markChunkAsUploaded = async (sessionId: string, chunkIndex: number) .execute(); }; -export const deleteUploadSession = async (trx: typeof db, sessionId: string) => { - await trx.deleteFrom("upload_session").where("id", "=", sessionId).execute(); +// The following functions return the paths of the files uploaded by the deleted sessions + +export const deleteUploadSession = async (executor: Executor, sessionId: string) => { + const sessions = await executor + .deleteFrom("upload_session") + .where("id", "=", sessionId) + .returning("path") + .execute(); + return sessions.map(({ path }) => path); }; -export const discardUploadSession = async (sessionId: string) => { - await db.transaction().execute(async (trx) => { - const sessions = await trx - .deleteFrom("upload_session") - .where("id", "=", sessionId) - .returning("path") - .execute(); - await enqueueFileDeletions( - trx, - sessions.map(({ path }) => path), - ); - }); +// Deletes the sessions uploading files into the directories or thumbnails of the files +export const deleteUploadSessionsOf = async ( + executor: Executor, + { directoryIds = [], fileIds = [] }: { directoryIds?: number[]; fileIds?: number[] }, +) => { + if (directoryIds.length === 0 && fileIds.length === 0) return []; + + const sessions = await executor + .deleteFrom("upload_session") + .where((eb) => + eb.or([ + ...(directoryIds.length > 0 ? [eb("parent_id", "in", directoryIds)] : []), + ...(fileIds.length > 0 ? [eb("file_id", "in", fileIds)] : []), + ]), + ) + .returning("path") + .execute(); + return sessions.map(({ path }) => path); }; -export const cleanupExpiredUploadSessions = async () => { - await db.transaction().execute(async (trx) => { - const sessions = await trx - .deleteFrom("upload_session") - .where("expires_at", "<=", new Date()) - .returning("path") - .execute(); - await enqueueFileDeletions( - trx, - sessions.map(({ path }) => path), - ); - }); +export const deleteExpiredUploadSessions = async (executor: Executor) => { + const sessions = await executor + .deleteFrom("upload_session") + .where("expires_at", "<=", new Date()) + .returning("path") + .execute(); + return sessions.map(({ path }) => path); }; diff --git a/src/lib/server/services/category.ts b/src/lib/server/services/category.ts index 41a23d1..99f4e8e 100644 --- a/src/lib/server/services/category.ts +++ b/src/lib/server/services/category.ts @@ -1,25 +1,57 @@ -import { CategoryRepo, FileRepo } from "#lib/server/db/index.ts"; +import { + CategoryRepo, + FileRepo, + LogRepo, + db, + isUniqueViolation, + transaction, + type Tx, +} from "#lib/server/db/index.ts"; import type { Ciphertext } from "#lib/server/db/schema/index.ts"; import { AppError } from "#lib/server/errors.ts"; -import { ensureFreshDekVersion } from "./common.ts"; +import { + dekVersionMismatchOrNotFound, + ensureActiveMekVersion, + ensureFreshDekVersion, +} from "./common.ts"; + +const categoryNotFoundError = () => new AppError("NOT_FOUND", "Category not found"); export const getCategory = async (userId: number, categoryId: CategoryId, recurse: boolean) => { const category = - categoryId !== "root" ? await CategoryRepo.getCategory(userId, categoryId) : undefined; + categoryId !== "root" ? await CategoryRepo.getCategory(db, userId, categoryId) : undefined; if (category === null) { - throw new AppError("NOT_FOUND", "Category not found"); + throw categoryNotFoundError(); } const [subCategories, files] = await Promise.all([ - CategoryRepo.getAllCategoriesByParent(userId, categoryId), - categoryId !== "root" ? FileRepo.getAllFilesByCategory(userId, categoryId, recurse) : undefined, + CategoryRepo.getAllCategoriesByParent(db, userId, categoryId), + categoryId !== "root" + ? FileRepo.getAllFilesByCategory(db, userId, categoryId, recurse) + : undefined, ]); return { category, subCategories, files }; }; export const createCategory = async (params: CategoryRepo.NewCategory) => { ensureFreshDekVersion(params.dekVersion); - await CategoryRepo.registerCategory(params); + + await transaction(async (trx) => { + await ensureActiveMekVersion(trx, params.userId, params.mekVersion); + if ( + params.parentId !== "root" && + !(await CategoryRepo.lockCategory(trx, params.userId, params.parentId)) + ) { + throw categoryNotFoundError(); + } + + const { id } = await CategoryRepo.createCategory(trx, params); + await LogRepo.appendCategoryLog(trx, { + category_id: id, + action: "create", + new_name: params.encName, + }); + }); }; export const renameCategory = async ( @@ -28,28 +60,54 @@ export const renameCategory = async ( dekVersion: Date, encName: Ciphertext, ) => { - await CategoryRepo.setCategoryEncName(userId, categoryId, dekVersion, encName); + await transaction(async (trx) => { + if (!(await CategoryRepo.setCategoryEncName(trx, userId, categoryId, dekVersion, encName))) { + throw await dekVersionMismatchOrNotFound( + CategoryRepo.getCategory(trx, userId, categoryId), + "Category not found", + ); + } + + await LogRepo.appendCategoryLog(trx, { + category_id: categoryId, + action: "rename", + new_name: encName, + }); + }); }; export const deleteCategory = async (userId: number, categoryId: number) => { - await CategoryRepo.unregisterCategory(userId, categoryId); + if (!(await CategoryRepo.deleteCategory(db, userId, categoryId))) { + throw categoryNotFoundError(); + } }; -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) { +// Locks the category and the file so that they aren't deleted until the transaction ends +const lockCategoryAndFile = async (trx: Tx, userId: number, categoryId: number, fileId: number) => { + if (!(await CategoryRepo.lockCategory(trx, userId, categoryId))) { + throw categoryNotFoundError(); + } else if (!(await FileRepo.lockFile(trx, userId, fileId))) { 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); + try { + await transaction(async (trx) => { + await lockCategoryAndFile(trx, userId, categoryId, fileId); + await FileRepo.addFileToCategory(trx, fileId, categoryId); + await LogRepo.appendFileLog(trx, { + file_id: fileId, + action: "add-to-category", + category_id: categoryId, + }); + }); + } catch (e) { + if (isUniqueViolation(e)) { + throw new AppError("BAD_REQUEST", "File already added to category"); + } + throw e; + } }; export const removeFileFromCategory = async ( @@ -57,6 +115,16 @@ export const removeFileFromCategory = async ( categoryId: number, fileId: number, ) => { - await ensureCategoryAndFileExist(userId, categoryId, fileId); - await FileRepo.removeFileFromCategory(fileId, categoryId); + await transaction(async (trx) => { + await lockCategoryAndFile(trx, userId, categoryId, fileId); + if (!(await FileRepo.removeFileFromCategory(trx, fileId, categoryId))) { + throw new AppError("BAD_REQUEST", "File not found in category"); + } + + await LogRepo.appendFileLog(trx, { + file_id: fileId, + action: "remove-from-category", + category_id: categoryId, + }); + }); }; diff --git a/src/lib/server/services/common.ts b/src/lib/server/services/common.ts index a9985dc..254f270 100644 --- a/src/lib/server/services/common.ts +++ b/src/lib/server/services/common.ts @@ -1,4 +1,4 @@ -import { HskRepo, MekRepo, type Tx } from "#lib/server/db/index.ts"; +import { FileRepo, HskRepo, MekRepo, type Tx } from "#lib/server/db/index.ts"; import { AppError } from "#lib/server/errors.ts"; const DEK_VERSION_TOLERANCE = 60 * 1000; // 1 minute @@ -10,6 +10,31 @@ export const ensureFreshDekVersion = (dekVersion: Date) => { } }; +const invalidDekVersionError = () => new AppError("CONFLICT", "Invalid DEK version"); + +// Updates conditioned on the DEK version fail if the entry doesn't exist or its DEK has changed +export const dekVersionMismatchOrNotFound = async ( + entry: Promise, + notFoundMessage: string, +) => { + return (await entry) ? invalidDekVersionError() : new AppError("NOT_FOUND", notFoundMessage); +}; + +// Locks the file so that its DEK isn't changed until the transaction ends +export const ensureFileDekVersion = async ( + trx: Tx, + userId: number, + fileId: number, + dekVersion: Date, +) => { + const currentDekVersion = await FileRepo.lockFileDekVersion(trx, userId, fileId); + if (!currentDekVersion) { + throw new AppError("NOT_FOUND", "File not found"); + } else if (currentDekVersion.getTime() !== dekVersion.getTime()) { + throw invalidDekVersionError(); + } +}; + // New data must be encrypted with the active keys, which are locked until the transaction ends export const ensureActiveMekVersion = async (trx: Tx, userId: number, mekVersion: number) => { diff --git a/src/lib/server/services/directory.ts b/src/lib/server/services/directory.ts index ac79fff..ca6bd53 100644 --- a/src/lib/server/services/directory.ts +++ b/src/lib/server/services/directory.ts @@ -1,33 +1,56 @@ -import { DirectoryRepo, FileRepo } from "#lib/server/db/index.ts"; +import { + DirectoryRepo, + FileDeletionRepo, + FileRepo, + LogRepo, + UploadRepo, + db, + transaction, +} from "#lib/server/db/index.ts"; import type { Ciphertext } from "#lib/server/db/schema/index.ts"; import { AppError } from "#lib/server/errors.ts"; -import { ensureFreshDekVersion } from "./common.ts"; +import { + dekVersionMismatchOrNotFound, + ensureActiveMekVersion, + ensureFreshDekVersion, +} from "./common.ts"; import { schedulePendingFileDeletions } from "./file-deletion.ts"; +const directoryNotFoundError = () => new AppError("NOT_FOUND", "Directory not found"); + export const getDirectory = async (userId: number, directoryId: DirectoryId) => { const directory = - directoryId !== "root" ? await DirectoryRepo.getDirectory(userId, directoryId) : undefined; + directoryId !== "root" ? await DirectoryRepo.getDirectory(db, userId, directoryId) : undefined; if (directory === null) { - throw new AppError("NOT_FOUND", "Directory not found"); + throw directoryNotFoundError(); } const [subDirectories, files] = await Promise.all([ - DirectoryRepo.getAllDirectoriesByParent(userId, directoryId), - FileRepo.getAllFilesByParent(userId, directoryId), + DirectoryRepo.getAllDirectoriesByParent(db, userId, directoryId), + FileRepo.getAllFilesByParent(db, userId, directoryId), ]); return { directory, subDirectories, files }; }; -export const createDirectory = async (params: { - userId: number; - parentId: DirectoryId; - mekVersion: number; - encDek: string; - dekVersion: Date; - encName: Ciphertext; -}) => { +export const createDirectory = async (params: DirectoryRepo.NewDirectory) => { ensureFreshDekVersion(params.dekVersion); - await DirectoryRepo.registerDirectory(params); + + await transaction(async (trx) => { + await ensureActiveMekVersion(trx, params.userId, params.mekVersion); + if ( + params.parentId !== "root" && + !(await DirectoryRepo.lockDirectory(trx, params.userId, params.parentId)) + ) { + throw directoryNotFoundError(); + } + + const { id } = await DirectoryRepo.createDirectory(trx, params); + await LogRepo.appendDirectoryLog(trx, { + directory_id: id, + action: "create", + new_name: params.encName, + }); + }); }; export const renameDirectory = async ( @@ -36,23 +59,68 @@ export const renameDirectory = async ( dekVersion: Date, encName: Ciphertext, ) => { - await DirectoryRepo.setDirectoryEncName(userId, directoryId, dekVersion, encName); + await transaction(async (trx) => { + if (!(await DirectoryRepo.setDirectoryEncName(trx, userId, directoryId, dekVersion, encName))) { + throw await dekVersionMismatchOrNotFound( + DirectoryRepo.getDirectory(trx, userId, directoryId), + "Directory not found", + ); + } + + await LogRepo.appendDirectoryLog(trx, { + directory_id: directoryId, + action: "rename", + new_name: encName, + }); + }); }; +// Deletes the directory with its descendants and files export const deleteDirectory = async (userId: number, directoryId: number) => { - const { deletedFileIds } = await DirectoryRepo.unregisterDirectory(userId, directoryId); + const deletedFileIds = await transaction(async (trx) => { + const directoryIds = await DirectoryRepo.lockDirectoryTree(trx, userId, directoryId); + if (directoryIds.length === 0) { + throw directoryNotFoundError(); + } + + const files = await FileRepo.lockFilesInDirectoriesForDeletion(trx, directoryIds); + const fileIds = files.map(({ id }) => id); + const uploadPaths = await UploadRepo.deleteUploadSessionsOf(trx, { directoryIds, fileIds }); + await FileRepo.deleteFiles(trx, fileIds); + await DirectoryRepo.deleteDirectories(trx, directoryIds); + + await FileDeletionRepo.enqueueFileDeletions(trx, [ + ...files.flatMap(({ path, thumbnailPath }) => [path, thumbnailPath]), + ...uploadPaths, + ]); + return fileIds; + }); + schedulePendingFileDeletions(); return { deletedFileIds }; }; export const getFavoriteDirectories = async (userId: number) => { - return await DirectoryRepo.getAllFavoriteDirectories(userId); + return await DirectoryRepo.getAllFavoriteDirectories(db, userId); }; +// Does nothing if the directory is already in the state export const setDirectoryFavorite = async ( userId: number, directoryId: number, isFavorite: boolean, ) => { - await DirectoryRepo.setDirectoryFavorite(userId, directoryId, isFavorite); + await transaction(async (trx) => { + if (!(await DirectoryRepo.setDirectoryFavorite(trx, userId, directoryId, isFavorite))) { + if (!(await DirectoryRepo.getDirectory(trx, userId, directoryId))) { + throw directoryNotFoundError(); + } + return; + } + + await LogRepo.appendDirectoryLog(trx, { + directory_id: directoryId, + action: isFavorite ? "add-to-favorites" : "remove-from-favorites", + }); + }); }; diff --git a/src/lib/server/services/file-deletion.ts b/src/lib/server/services/file-deletion.ts index d117cd6..b2ed334 100644 --- a/src/lib/server/services/file-deletion.ts +++ b/src/lib/server/services/file-deletion.ts @@ -1,4 +1,4 @@ -import { FileDeletionRepo } from "#lib/server/db/index.ts"; +import { FileDeletionRepo, db } from "#lib/server/db/index.ts"; import { deleteFile } from "#lib/server/storage/index.ts"; const BATCH_SIZE = 100; @@ -6,7 +6,7 @@ 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); + const deletions = await FileDeletionRepo.getPendingFileDeletions(db, BATCH_SIZE); if (deletions.length === 0) return; const results = await Promise.allSettled(deletions.map(({ path }) => deleteFile(path))); @@ -17,7 +17,7 @@ export const processPendingFileDeletions = async () => { if (result.status === "rejected") console.error(result.reason); }); - await FileDeletionRepo.completeFileDeletions(deletedIds); + await FileDeletionRepo.completeFileDeletions(db, deletedIds); if (deletedIds.length < deletions.length) return; // Retry later } }; diff --git a/src/lib/server/services/file.ts b/src/lib/server/services/file.ts index cfea308..1372c08 100644 --- a/src/lib/server/services/file.ts +++ b/src/lib/server/services/file.ts @@ -1,26 +1,37 @@ -import { FileRepo, MediaRepo } from "#lib/server/db/index.ts"; +import { + FileDeletionRepo, + FileRepo, + LogRepo, + MediaRepo, + UploadRepo, + db, + transaction, +} from "#lib/server/db/index.ts"; import type { Ciphertext } from "#lib/server/db/schema/index.ts"; import { AppError } from "#lib/server/errors.ts"; import { createEncContentStream } from "#lib/server/storage/index.ts"; import type { RangeRequest } from "#lib/shared/http.ts"; +import { dekVersionMismatchOrNotFound } from "./common.ts"; import { schedulePendingFileDeletions } from "./file-deletion.ts"; +const fileNotFoundError = () => new AppError("NOT_FOUND", "File not found"); + export const getFile = async (userId: number, fileId: number) => { - const file = await FileRepo.getFile(userId, fileId); + const file = await FileRepo.getFile(db, userId, fileId); if (!file) { - throw new AppError("NOT_FOUND", "File not found"); + throw fileNotFoundError(); } - const categories = await FileRepo.getAllFileCategories(fileId); + const categories = await FileRepo.getAllFileCategories(db, fileId); return { ...file, categories }; }; export const getFiles = async (userId: number, fileIds: number[]) => { - return await FileRepo.getFilesWithCategories(userId, fileIds); + return await FileRepo.getFilesWithCategories(db, userId, fileIds); }; export const getAllFileIds = async (userId: number) => { - return await FileRepo.getAllFileIds(userId); + return await FileRepo.getAllFileIds(db, userId); }; export const getFileIdsByContentHmac = async ( @@ -28,11 +39,11 @@ export const getFileIdsByContentHmac = async ( hskVersion: number, contentHmac: string, ) => { - return await FileRepo.getAllFileIdsByContentHmac(userId, hskVersion, contentHmac); + return await FileRepo.getAllFileIdsByContentHmac(db, userId, hskVersion, contentHmac); }; export const getFilesWithoutThumbnail = async (userId: number) => { - return await FileRepo.getFilesWithoutThumbnail(userId); + return await FileRepo.getFilesWithoutThumbnail(db, userId); }; export const renameFile = async ( @@ -41,24 +52,60 @@ export const renameFile = async ( dekVersion: Date, encName: Ciphertext, ) => { - await FileRepo.setFileEncName(userId, fileId, dekVersion, encName); + await transaction(async (trx) => { + if (!(await FileRepo.setFileEncName(trx, userId, fileId, dekVersion, encName))) { + throw await dekVersionMismatchOrNotFound( + FileRepo.getFile(trx, userId, fileId), + "File not found", + ); + } + + await LogRepo.appendFileLog(trx, { file_id: fileId, action: "rename", new_name: encName }); + }); }; export const deleteFile = async (userId: number, fileId: number) => { - await FileRepo.unregisterFile(userId, fileId); + await transaction(async (trx) => { + const file = await FileRepo.lockFileForDeletion(trx, userId, fileId); + if (!file) { + throw fileNotFoundError(); + } + + const uploadPaths = await UploadRepo.deleteUploadSessionsOf(trx, { fileIds: [fileId] }); + await FileRepo.deleteFiles(trx, [fileId]); + await FileDeletionRepo.enqueueFileDeletions(trx, [ + file.path, + file.thumbnailPath, + ...uploadPaths, + ]); + }); + schedulePendingFileDeletions(); }; export const getFavoriteFiles = async (userId: number) => { - return await FileRepo.getAllFavoriteFiles(userId); + return await FileRepo.getAllFavoriteFiles(db, userId); }; +// Does nothing if the file is already in the state export const setFileFavorite = async (userId: number, fileId: number, isFavorite: boolean) => { - await FileRepo.setFileFavorite(userId, fileId, isFavorite); + await transaction(async (trx) => { + if (!(await FileRepo.setFileFavorite(trx, userId, fileId, isFavorite))) { + if (!(await FileRepo.getFile(trx, userId, fileId))) { + throw fileNotFoundError(); + } + return; + } + + await LogRepo.appendFileLog(trx, { + file_id: fileId, + action: isFavorite ? "add-to-favorites" : "remove-from-favorites", + }); + }); }; export const getFileThumbnail = async (userId: number, fileId: number) => { - const thumbnail = await MediaRepo.getFileThumbnail(userId, fileId); + const thumbnail = await MediaRepo.getFileThumbnail(db, userId, fileId); if (!thumbnail) { throw new AppError("NOT_FOUND", "File or its thumbnail not found"); } @@ -68,9 +115,9 @@ export const getFileThumbnail = async (userId: number, fileId: number) => { const decodeIv = (iv: string | null) => (iv ? Buffer.from(iv, "base64") : undefined); export const getFileStream = async (userId: number, fileId: number, range?: RangeRequest) => { - const file = await FileRepo.getFile(userId, fileId); + const file = await FileRepo.getFile(db, userId, fileId); if (!file) { - throw new AppError("NOT_FOUND", "File not found"); + throw fileNotFoundError(); } return await createEncContentStream(file.path, decodeIv(file.encContentIv), range); }; diff --git a/src/lib/server/services/search.ts b/src/lib/server/services/search.ts index 9bbca1b..984948e 100644 --- a/src/lib/server/services/search.ts +++ b/src/lib/server/services/search.ts @@ -1,4 +1,4 @@ -import { DirectoryRepo, FileRepo } from "#lib/server/db/index.ts"; +import { DirectoryRepo, FileRepo, db } from "#lib/server/db/index.ts"; export const search = async ( userId: number, @@ -15,17 +15,12 @@ export const search = async ( const [directories, files] = await Promise.all([ !hasCategoryFilter - ? DirectoryRepo.searchDirectories(userId, { - parentId: filters.ancestorId, + ? DirectoryRepo.searchDirectories(db, userId, { + ancestorId: filters.ancestorId, inFavorites: filters.inFavorites, }) : [], - FileRepo.searchFiles(userId, { - parentId: filters.ancestorId, - inFavorites: filters.inFavorites, - includeCategoryIds: filters.includeCategoryIds, - excludeCategoryIds: filters.excludeCategoryIds, - }), + FileRepo.searchFiles(db, userId, filters), ]); return { directories, files }; }; diff --git a/src/lib/server/services/upload.ts b/src/lib/server/services/upload.ts index 8c08507..7b0360f 100644 --- a/src/lib/server/services/upload.ts +++ b/src/lib/server/services/upload.ts @@ -1,5 +1,14 @@ import type { Readable } from "stream"; -import { db, FileRepo, MediaRepo, UploadRepo } from "#lib/server/db/index.ts"; +import { + DirectoryRepo, + FileDeletionRepo, + FileRepo, + LogRepo, + MediaRepo, + UploadRepo, + db, + transaction, +} from "#lib/server/db/index.ts"; import type { Ciphertext } from "#lib/server/db/schema/index.ts"; import { AppError } from "#lib/server/errors.ts"; import { @@ -8,7 +17,12 @@ import { reserveFile, writeChunkAtOffset, } from "#lib/server/storage/index.ts"; -import { ensureFreshDekVersion } from "./common.ts"; +import { + ensureActiveHskVersion, + ensureActiveMekVersion, + ensureFileDekVersion, + ensureFreshDekVersion, +} from "./common.ts"; import { schedulePendingFileDeletions } from "./file-deletion.ts"; const UPLOAD_EXPIRES = 24 * 3600 * 1000; // 24 hours @@ -51,11 +65,24 @@ export const startFileUpload = async (params: { 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), + await transaction(async (trx) => { + await ensureActiveMekVersion(trx, params.userId, params.mekVersion); + if (params.hskVersion) { + await ensureActiveHskVersion(trx, params.userId, params.hskVersion); + } + if ( + params.parentId !== "root" && + !(await DirectoryRepo.lockDirectory(trx, params.userId, params.parentId)) + ) { + throw new AppError("NOT_FOUND", "Directory not found"); + } + + await UploadRepo.createFileUploadSession(trx, { + ...params, + id, + path, + expiresAt: new Date(Date.now() + UPLOAD_EXPIRES), + }); }); return { uploadId: id }; }); @@ -68,15 +95,17 @@ export const startFileThumbnailUpload = async ( ) => { 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, + await transaction(async (trx) => { + await ensureFileDekVersion(trx, userId, fileId, dekVersion); + await UploadRepo.createThumbnailUploadSession(trx, { + id, + userId, + path, + totalChunks: THUMBNAIL_CHUNKS, + expiresAt: new Date(Date.now() + UPLOAD_EXPIRES), + fileId, + dekVersion, + }); }); return { uploadId: id }; }); @@ -88,7 +117,7 @@ export const uploadChunk = async ( chunkIndex: number, encChunkStream: Readable, ) => { - const session = await UploadRepo.getUploadSession(sessionId, userId); + const session = await UploadRepo.getUploadSession(db, sessionId, userId); if (!session) { throw new AppError("NOT_FOUND", "Invalid upload id"); } else if (chunkIndex > session.totalChunks) { @@ -101,7 +130,7 @@ export const uploadChunk = async ( // hash verification on completion const isLastChunk = chunkIndex === session.totalChunks; await writeChunkAtOffset(session.path, encChunkStream, chunkIndex, isLastChunk); - await UploadRepo.markChunkAsUploaded(sessionId, chunkIndex); + await UploadRepo.markChunkAsUploaded(db, sessionId, chunkIndex); }; const getCompletedUploadSession = async ( @@ -109,7 +138,7 @@ const getCompletedUploadSession = async ( uploadId: string, type: T, ) => { - const session = await UploadRepo.getUploadSession(uploadId, userId); + const session = await UploadRepo.getUploadSession(db, uploadId, userId); if (session?.type !== type) { throw new AppError("NOT_FOUND", "Invalid upload id"); } else if (session.uploadedChunks < session.totalChunks) { @@ -131,19 +160,24 @@ export const completeFileUpload = async ( const hash = await hashFile(session.path); if (hash !== encContentHash) { - await UploadRepo.discardUploadSession(uploadId); + await transaction(async (trx) => { + await FileDeletionRepo.enqueueFileDeletions( + trx, + await UploadRepo.deleteUploadSession(trx, uploadId), + ); + }); schedulePendingFileDeletions(); throw new AppError("CONFLICT", "Uploaded file corrupted"); } - return await db.transaction().execute(async (trx) => { + return await transaction(async (trx) => { // Locking the session ensures that it is completed only once const lockedSession = await UploadRepo.lockUploadSession(trx, uploadId, userId); - if (lockedSession.type !== "file") { + if (lockedSession?.type !== "file") { throw new AppError("NOT_FOUND", "Invalid upload id"); } - const { id: fileId } = await FileRepo.registerFile(trx, { + const { id: fileId } = await FileRepo.createFile(trx, { parentId: lockedSession.parentId, userId, path: lockedSession.path, @@ -159,6 +193,11 @@ export const completeFileUpload = async ( encCreatedAt: lockedSession.encCreatedAt, encLastModifiedAt: lockedSession.encLastModifiedAt, }); + await LogRepo.appendFileLog(trx, { + file_id: fileId, + action: "create", + new_name: lockedSession.encName, + }); await UploadRepo.deleteUploadSession(trx, uploadId); return { fileId }; }); @@ -167,26 +206,27 @@ export const completeFileUpload = async ( export const completeFileThumbnailUpload = async (userId: number, uploadId: string) => { await getCompletedUploadSession(userId, uploadId, "thumbnail"); - await db.transaction().execute(async (trx) => { + await transaction(async (trx) => { const session = await UploadRepo.lockUploadSession(trx, uploadId, userId); - if (session.type !== "thumbnail") { + if (session?.type !== "thumbnail") { throw new AppError("NOT_FOUND", "Invalid upload id"); } - await MediaRepo.updateFileThumbnail( - trx, - userId, - session.fileId, - session.dekVersion, - session.path, - null, - ); + await ensureFileDekVersion(trx, userId, session.fileId, session.dekVersion); + const oldPath = await MediaRepo.lockFileThumbnailPath(trx, session.fileId); + await MediaRepo.upsertFileThumbnail(trx, session.fileId, session.path, null); await UploadRepo.deleteUploadSession(trx, uploadId); + await FileDeletionRepo.enqueueFileDeletions(trx, [oldPath]); }); schedulePendingFileDeletions(); }; export const cleanupExpiredUploadSessions = async () => { - await UploadRepo.cleanupExpiredUploadSessions(); + await transaction(async (trx) => { + await FileDeletionRepo.enqueueFileDeletions( + trx, + await UploadRepo.deleteExpiredUploadSessions(trx), + ); + }); schedulePendingFileDeletions(); }; diff --git a/tests/server/deletion.test.ts b/tests/server/deletion.test.ts index a19e226..faa7950 100644 --- a/tests/server/deletion.test.ts +++ b/tests/server/deletion.test.ts @@ -1,6 +1,6 @@ import { access } from "fs/promises"; import { beforeEach, describe, expect, it } from "vitest"; -import { UploadRepo } from "#lib/server/db/index.ts"; +import { UploadRepo, db } from "#lib/server/db/index.ts"; import { processPendingFileDeletions } from "#lib/server/services/file-deletion.ts"; import { DirectoryService, FileService } from "#lib/server/services/index.ts"; import { @@ -58,7 +58,7 @@ describe("file deletion", () => { await createFile(userId, grandchildId, paths[1]!), ]; const otherFileId = await createFile(userId, otherId, paths[2]!); - await UploadRepo.createFileUploadSession({ + await UploadRepo.createFileUploadSession(db, { ...newEntryParams(userId, childId), id: crypto.randomUUID(), path: paths[3]!, diff --git a/tests/server/ownership.test.ts b/tests/server/ownership.test.ts index a62a4b3..14f6fb5 100644 --- a/tests/server/ownership.test.ts +++ b/tests/server/ownership.test.ts @@ -1,5 +1,5 @@ import { beforeEach, describe, expect, it } from "vitest"; -import { CategoryRepo, DirectoryRepo, UploadRepo, IntegrityError } from "#lib/server/db/index.ts"; +import { CategoryService, DirectoryService, UploadService } from "#lib/server/services/index.ts"; import { createCategory, createDirectory, @@ -16,10 +16,10 @@ describe("parent ownership", () => { const { userId } = await createUser(); const parentId = await createDirectory(userId); - await DirectoryRepo.registerDirectory(newEntryParams(userId, parentId)); + await DirectoryService.createDirectory(newEntryParams(userId, parentId)); - const directories = await DirectoryRepo.getAllDirectoriesByParent(userId, parentId); - expect(directories).toHaveLength(1); + const { subDirectories } = await DirectoryService.getDirectory(userId, parentId); + expect(subDirectories).toHaveLength(1); }); it("rejects creating a directory under another user's directory", async () => { @@ -28,8 +28,8 @@ describe("parent ownership", () => { const victimDirectoryId = await createDirectory(victimId); await expect( - DirectoryRepo.registerDirectory(newEntryParams(attackerId, victimDirectoryId)), - ).rejects.toThrow(new IntegrityError("Directory not found")); + DirectoryService.createDirectory(newEntryParams(attackerId, victimDirectoryId)), + ).rejects.toMatchObject({ code: "NOT_FOUND", message: "Directory not found" }); }); it("rejects creating a category under another user's category", async () => { @@ -38,8 +38,8 @@ describe("parent ownership", () => { const victimCategoryId = await createCategory(victimId); await expect( - CategoryRepo.registerCategory(newEntryParams(attackerId, victimCategoryId)), - ).rejects.toThrow(new IntegrityError("Category not found")); + CategoryService.createCategory(newEntryParams(attackerId, victimCategoryId)), + ).rejects.toMatchObject({ code: "NOT_FOUND", message: "Category not found" }); }); it("rejects uploading a file into another user's directory", async () => { @@ -48,17 +48,14 @@ describe("parent ownership", () => { const victimDirectoryId = await createDirectory(victimId); await expect( - UploadRepo.createFileUploadSession({ + UploadService.startFileUpload({ ...newEntryParams(attackerId, victimDirectoryId), - id: crypto.randomUUID(), - path: "unused", totalChunks: 1, - expiresAt: new Date(Date.now() + 60_000), hskVersion: null, contentType: "image/png", encCreatedAt: null, encLastModifiedAt: ciphertext(), }), - ).rejects.toThrow(new IntegrityError("Directory not found")); + ).rejects.toMatchObject({ code: "NOT_FOUND", message: "Directory not found" }); }); }); diff --git a/tests/server/upload.test.ts b/tests/server/upload.test.ts index eda34a8..4a6e057 100644 --- a/tests/server/upload.test.ts +++ b/tests/server/upload.test.ts @@ -1,7 +1,7 @@ import { createHash } from "crypto"; import { Readable } from "stream"; import { beforeEach, describe, expect, it } from "vitest"; -import { UploadRepo } from "#lib/server/db/index.ts"; +import { UploadRepo, db } from "#lib/server/db/index.ts"; import { FileService, UploadService } from "#lib/server/services/index.ts"; import { isChunkUploaded } from "#lib/server/services/upload.ts"; import { ciphertext, createUser, newEntryParams, resetDatabase } from "../helpers/database.ts"; @@ -12,7 +12,7 @@ describe("upload session bitmap", () => { it("matches the bit order of PostgreSQL's set_bit", async () => { const { userId } = await createUser(); const id = crypto.randomUUID(); - await UploadRepo.createFileUploadSession({ + await UploadRepo.createFileUploadSession(db, { ...newEntryParams(userId, "root"), id, path: "unused", @@ -26,10 +26,10 @@ describe("upload session bitmap", () => { const uploaded = [1, 8, 9, 20]; for (const chunkIndex of uploaded) { - await UploadRepo.markChunkAsUploaded(id, chunkIndex); + await UploadRepo.markChunkAsUploaded(db, id, chunkIndex); } - const session = await UploadRepo.getUploadSession(id, userId); + const session = await UploadRepo.getUploadSession(db, id, userId); expect(session?.uploadedChunks).toBe(uploaded.length); for (let chunkIndex = 1; chunkIndex <= 20; chunkIndex++) { expect(isChunkUploaded(session!.bitmap, chunkIndex)).toBe(uploaded.includes(chunkIndex)); @@ -80,7 +80,7 @@ describe("upload completion", () => { await expect( UploadService.completeFileUpload(userId, uploadId, null, "invalid"), ).rejects.toMatchObject({ code: "CONFLICT" }); - expect(await UploadRepo.getUploadSession(uploadId, userId)).toBeNull(); + expect(await UploadRepo.getUploadSession(db, uploadId, userId)).toBeNull(); }); it("rejects chunks of invalid sizes", async () => {