DB 접근 시 Executor를 명시적으로 전달하고 서비스에서 트랜잭션 경계를 관리하도록 인증 및 키 관련 코드를 개선하고 직렬화 실패 시 트랜잭션을 재시도하도록 변경

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

src/lib/server/auth.ts

@@ -1,4 +1,4 @@
-import { ClientRepo, SessionRepo } from "#lib/server/db/index.ts";
+import { ClientRepo, SessionRepo, db } from "#lib/server/db/index.ts";
 import env from "#lib/server/config.ts";
 import { verifySessionId } from "#lib/server/crypto.ts";
 import { AppError, invalidSessionError } from "#lib/server/errors.ts";
@@ -39,7 +39,7 @@ export const authenticate = async (sessionIdSigned: string, ip: string, userAgen
     throw new AuthenticationError(400, "Invalid session id");
   }
 
-  const session = await SessionRepo.refreshSession(sessionId, ip, userAgent);
+  const session = await SessionRepo.refreshSession(db, sessionId, ip, userAgent);
   if (!session) {
     throw new AuthenticationError(401, "Invalid session id");
   }
@@ -59,7 +59,7 @@ const ensureUserClientState = async (
     throw new AppError("FORBIDDEN", "Forbidden");
   }
 
-  const userClient = await ClientRepo.getUserClient(userId, clientId);
+  const userClient = await ClientRepo.getUserClient(db, userId, clientId);
   if (!userClient) {
     throw invalidSessionError();
   } else if (userClient.state !== state) {

src/lib/server/db/executor.ts

@@ -0,0 +1,39 @@
+import type { IsolationLevel, Kysely, Transaction } from "kysely";
+import pg from "pg";
+import db from "./kysely.ts";
+import type { Database } from "./schema/index.ts";
+
+// Runs queries either directly or in a transaction
+export type Executor = Kysely<Database>;
+// Required by queries that only make sense in a transaction (e.g. locking rows)
+export type Tx = Transaction<Database>;
+
+const isDatabaseError = (e: unknown, code: string): e is pg.DatabaseError => {
+  return e instanceof pg.DatabaseError && e.code === code;
+};
+
+// `constraint` narrows down the violated constraint if a query may violate several of them
+export const isUniqueViolation = (e: unknown, constraint?: string) => {
+  return isDatabaseError(e, "23505") && (!constraint || e.constraint === constraint);
+};
+
+export const isForeignKeyViolation = (e: unknown) => isDatabaseError(e, "23503");
+
+// Serialization failures and deadlocks succeed if the transaction is retried
+const isRetryable = (e: unknown) => isDatabaseError(e, "40001") || isDatabaseError(e, "40P01");
+
+export const transaction = async <T>(
+  callback: (trx: Tx) => Promise<T>,
+  { isolationLevel, maxRetries = 3 }: { isolationLevel?: IsolationLevel; maxRetries?: number } = {},
+) => {
+  for (let attempt = 0; ; attempt++) {
+    try {
+      const builder = db.transaction();
+      return await (isolationLevel ? builder.setIsolationLevel(isolationLevel) : builder).execute(
+        callback,
+      );
+    } catch (e) {
+      if (!isRetryable(e) || attempt >= maxRetries) throw e;
+    }
+  }
+};

src/lib/server/db/index.ts

@@ -4,6 +4,7 @@ export * as DirectoryRepo from "./repositories/directory.ts";
 export * as FileRepo from "./repositories/file.ts";
 export * as FileDeletionRepo from "./repositories/file-deletion.ts";
 export * as HskRepo from "./repositories/hsk.ts";
+export * as LogRepo from "./repositories/log.ts";
 export * as MediaRepo from "./repositories/media.ts";
 export * as MekRepo from "./repositories/mek.ts";
 export * as SessionRepo from "./repositories/session.ts";
@@ -11,4 +12,5 @@ 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";

src/lib/server/db/queries.ts

@@ -0,0 +1,50 @@
+import type { Executor } from "./executor.ts";
+
+// Entries in the root have NULL as their parent, so use `is` instead of `=` to compare it
+export const encodeParentId = (parentId: DirectoryId | CategoryId) => {
+  return parentId === "root" ? null : parentId;
+};
+
+export const decodeParentId = (parentId: number | null) => parentId ?? ("root" as const);
+
+export const parentIdOperator = (parentId: DirectoryId | CategoryId) => {
+  return parentId === "root" ? ("is" as const) : ("=" as const);
+};
+
+// Adds recursive CTEs of the directories in the subtree of `rootId` including itself (empty if the
+// root), and of the directories in the subtrees of favorite directories (empty unless needed)
+export const withDirectoryTrees = (
+  executor: Executor,
+  userId: number,
+  { rootId, inFavorites }: { rootId: DirectoryId; inFavorites: boolean },
+) => {
+  return executor
+    .withRecursive("directory_tree", (qc) =>
+      qc
+        .selectFrom("directory")
+        .select("id")
+        .where("user_id", "=", userId)
+        .$if(rootId === "root", (qb) => qb.where((eb) => eb.lit(false)))
+        .$if(rootId !== "root", (qb) => qb.where("id", "=", rootId as number))
+        .unionAll((qc) =>
+          qc
+            .selectFrom("directory as d")
+            .innerJoin("directory_tree as dt", "d.parent_id", "dt.id")
+            .select("d.id"),
+        ),
+    )
+    .withRecursive("favorite_directory_tree", (qc) =>
+      qc
+        .selectFrom("directory")
+        .select("id")
+        .where("user_id", "=", userId)
+        .$if(!inFavorites, (qb) => qb.where((eb) => eb.lit(false)))
+        .$if(inFavorites, (qb) => qb.where("is_favorite", "=", true))
+        .unionAll((qc) =>
+          qc
+            .selectFrom("directory as d")
+            .innerJoin("favorite_directory_tree as dt", "d.parent_id", "dt.id")
+            .select("d.id"),
+        ),
+    );
+};

src/lib/server/db/repositories/category.ts

@@ -53,7 +53,9 @@ export const ensureCategoryExists = async (
 
 export const registerCategory = async (params: NewCategory) => {
   await db.transaction().execute(async (trx) => {
-    await lockActiveMekVersion(trx, params.userId, params.mekVersion);
+    if ((await lockActiveMekVersion(trx, params.userId)) !== params.mekVersion) {
+      throw new IntegrityError("Inactive MEK version");
+    }
 
     await ensureCategoryExists(trx, params.userId, params.parentId);
 

src/lib/server/db/repositories/client.ts

@@ -1,7 +1,6 @@
-import pg from "pg";
+import type { Selectable } from "kysely";
-import { IntegrityError } from "../error.ts";
+import type { Executor } from "../executor.ts";
-import db from "../kysely.ts";
+import type { ClientTable, UserClientState, UserClientTable } from "../schema/index.ts";
-import type { UserClientState } from "../schema/index.ts";
 
 interface Client {
   id: number;
@@ -15,154 +14,117 @@ interface UserClient {
   state: UserClientState;
 }
 
-interface UserClientWithDetails extends UserClient {
+const toClient = (row: Selectable<ClientTable>): Client => ({
-  encPubKey: string;
+  id: row.id,
-  sigPubKey: string;
+  encPubKey: row.encryption_public_key,
-}
+  sigPubKey: row.signature_public_key,
+});
 
-export const createClient = async (encPubKey: string, sigPubKey: string, userId: number) => {
+const toUserClient = (row: Selectable<UserClientTable>): UserClient => ({
-  return await db
+  userId: row.user_id,
-    .transaction()
+  clientId: row.client_id,
-    .setIsolationLevel("serializable")
+  state: row.state,
-    .execute(async (trx) => {
+});
-      const client = await trx
-        .selectFrom("client")
-        .where((eb) =>
-          eb.or([
-            eb("encryption_public_key", "=", encPubKey),
-            eb("encryption_public_key", "=", sigPubKey),
-            eb("signature_public_key", "=", encPubKey),
-            eb("signature_public_key", "=", sigPubKey),
-          ]),
-        )
-        .limit(1)
-        .executeTakeFirst();
-      if (client) {
-        throw new IntegrityError("Public key(s) already registered");
-      }
 
-      const { clientId } = await trx
+// Whether any of the keys is used by a client as either of its keys
-        .insertInto("client")
+export const isAnyPublicKeyRegistered = async (
-        .values({ encryption_public_key: encPubKey, signature_public_key: sigPubKey })
+  executor: Executor,
-        .returning("id as clientId")
+  encPubKey: string,
-        .executeTakeFirstOrThrow();
+  sigPubKey: string,
-      await trx
+) => {
-        .insertInto("user_client")
+  const client = await executor
-        .values({ user_id: userId, client_id: clientId })
+    .selectFrom("client")
-        .execute();
+    .select("id")
-      return { id: clientId };
+    .where((eb) =>
-    });
+      eb.or([
+        eb("encryption_public_key", "in", [encPubKey, sigPubKey]),
+        eb("signature_public_key", "in", [encPubKey, sigPubKey]),
+      ]),
+    )
+    .limit(1)
+    .executeTakeFirst();
+  return !!client;
 };
 
-export const getClient = async (clientId: number) => {
+export const createClient = async (executor: Executor, encPubKey: string, sigPubKey: string) => {
-  const client = await db
+  const { id } = await executor
+    .insertInto("client")
+    .values({ encryption_public_key: encPubKey, signature_public_key: sigPubKey })
+    .returning("id")
+    .executeTakeFirstOrThrow();
+  return { id };
+};
+
+export const getClient = async (executor: Executor, clientId: number) => {
+  const client = await executor
     .selectFrom("client")
     .selectAll()
     .where("id", "=", clientId)
     .limit(1)
     .executeTakeFirst();
-  return client
+  return client ? toClient(client) : null;
-    ? ({
-        id: client.id,
-        encPubKey: client.encryption_public_key,
-        sigPubKey: client.signature_public_key,
-      } satisfies Client)
-    : null;
 };
 
-export const getClientByPubKeys = async (encPubKey: string, sigPubKey: string) => {
+export const getClientByPubKeys = async (
-  const client = await db
+  executor: Executor,
+  encPubKey: string,
+  sigPubKey: string,
+) => {
+  const client = await executor
     .selectFrom("client")
     .selectAll()
     .where("encryption_public_key", "=", encPubKey)
     .where("signature_public_key", "=", sigPubKey)
     .limit(1)
     .executeTakeFirst();
-  return client
+  return client ? toClient(client) : null;
-    ? ({
-        id: client.id,
-        encPubKey: client.encryption_public_key,
-        sigPubKey: client.signature_public_key,
-      } satisfies Client)
-    : null;
 };
 
-export const createUserClient = async (userId: number, clientId: number) => {
+// Violates a unique constraint if the user already has the client
-  try {
+export const createUserClient = async (executor: Executor, userId: number, clientId: number) => {
-    await db.insertInto("user_client").values({ user_id: userId, client_id: clientId }).execute();
+  await executor
-  } catch (e) {
+    .insertInto("user_client")
-    if (e instanceof pg.DatabaseError && e.code === "23505") {
+    .values({ user_id: userId, client_id: clientId })
-      throw new IntegrityError("User client already exists");
+    .execute();
-    }
-    throw e;
-  }
 };
 
-export const getUserClient = async (userId: number, clientId: number) => {
+export const getUserClient = async (executor: Executor, userId: number, clientId: number) => {
-  const userClient = await db
+  const userClient = await executor
     .selectFrom("user_client")
     .selectAll()
     .where("user_id", "=", userId)
     .where("client_id", "=", clientId)
     .limit(1)
     .executeTakeFirst();
-  return userClient
+  return userClient ? toUserClient(userClient) : null;
-    ? ({
-        userId: userClient.user_id,
-        clientId: userClient.client_id,
-        state: userClient.state,
-      } satisfies UserClient)
-    : null;
 };
 
-export const getUserClientWithDetails = async (userId: number, clientId: number) => {
+// Returns whether the state has changed, which fails if the client isn't in the `from` state
-  const userClient = await db
+export const transitionUserClientState = async (
-    .selectFrom("user_client")
+  executor: Executor,
-    .innerJoin("client", "user_client.client_id", "client.id")
+  userId: number,
-    .selectAll()
+  clientId: number,
+  { from, to }: { from: UserClientState; to: UserClientState },
+) => {
+  const { numUpdatedRows } = await executor
+    .updateTable("user_client")
+    .set({ state: to })
     .where("user_id", "=", userId)
     .where("client_id", "=", clientId)
-    .limit(1)
+    .where("state", "=", from)
     .executeTakeFirst();
-  return userClient
+  return numUpdatedRows > 0n;
-    ? ({
-        userId: userClient.user_id,
-        clientId: userClient.client_id,
-        state: userClient.state,
-        encPubKey: userClient.encryption_public_key,
-        sigPubKey: userClient.signature_public_key,
-      } satisfies UserClientWithDetails)
-    : null;
 };
 
-export const setUserClientStateToPending = async (userId: number, clientId: number) => {
+export const createUserClientChallenge = async (
-  await db
+  executor: Executor,
-    .updateTable("user_client")
-    .set({ state: "pending" })
-    .where("user_id", "=", userId)
-    .where("client_id", "=", clientId)
-    .where("state", "=", "challenging")
-    .execute();
-};
-
-export const setUserClientStateToActive = async (userId: number, clientId: number) => {
-  await db
-    .updateTable("user_client")
-    .set({ state: "active" })
-    .where("user_id", "=", userId)
-    .where("client_id", "=", clientId)
-    .where("state", "=", "pending")
-    .execute();
-};
-
-export const registerUserClientChallenge = async (
   userId: number,
   clientId: number,
   answer: string,
   allowedIp: string,
   expiresAt: Date,
 ) => {
-  const { id } = await db
+  const { id } = await executor
     .insertInto("user_client_challenge")
     .values({
       user_id: userId,
@@ -177,11 +139,12 @@ export const registerUserClientChallenge = async (
 };
 
 export const consumeUserClientChallenge = async (
+  executor: Executor,
   challengeId: number,
   userId: number,
   ip: string,
 ) => {
-  const challenge = await db
+  const challenge = await executor
     .deleteFrom("user_client_challenge")
     .where("id", "=", challengeId)
     .where("user_id", "=", userId)
@@ -192,6 +155,9 @@ export const consumeUserClientChallenge = async (
   return challenge ? { clientId: challenge.client_id, answer: challenge.answer } : null;
 };
 
-export const cleanupExpiredUserClientChallenges = async () => {
+export const deleteExpiredUserClientChallenges = async (executor: Executor) => {
-  await db.deleteFrom("user_client_challenge").where("expires_at", "<=", new Date()).execute();
+  await executor
+    .deleteFrom("user_client_challenge")
+    .where("expires_at", "<=", new Date())
+    .execute();
 };

src/lib/server/db/repositories/directory.ts

@@ -49,7 +49,9 @@ export const ensureDirectoryExists = async (
 
 export const registerDirectory = async (params: Omit<Directory, "id" | "isFavorite">) => {
   await db.transaction().execute(async (trx) => {
-    await lockActiveMekVersion(trx, params.userId, params.mekVersion);
+    if ((await lockActiveMekVersion(trx, params.userId)) !== params.mekVersion) {
+      throw new IntegrityError("Inactive MEK version");
+    }
 
     await ensureDirectoryExists(trx, params.userId, params.parentId);
 

src/lib/server/db/repositories/hsk.ts

@@ -1,17 +1,7 @@
-import pg from "pg";
+import type { Executor, Tx } from "../executor.ts";
-import { IntegrityError } from "../error.ts";
-import db from "../kysely.ts";
-import type { HskState } from "../schema/index.ts";
 
-interface Hsk {
+// Locks the active version until the transaction ends, so that it isn't retired in the meantime
-  userId: number;
+export const lockActiveHskVersion = async (trx: Tx, userId: number) => {
-  version: number;
-  state: HskState;
-  mekVersion: number;
-  encHsk: string;
-}
-
-export const lockActiveHskVersion = async (trx: typeof db, userId: number, version: number) => {
   const hsk = await trx
     .selectFrom("hmac_secret_key")
     .select("version")
@@ -20,63 +10,40 @@ export const lockActiveHskVersion = async (trx: typeof db, userId: number, versi
     .limit(1)
     .forUpdate()
     .executeTakeFirst();
-  if (hsk?.version !== version) {
+  return hsk?.version ?? null;
-    throw new IntegrityError("Inactive HSK version");
-  }
 };
 
-export const registerInitialHsk = async (
+// Violates a unique constraint if the user already has an HMAC secret key
+export const createInitialHsk = async (
+  executor: Executor,
   userId: number,
-  createdBy: number,
   mekVersion: number,
   encHsk: string,
 ) => {
-  await db.transaction().execute(async (trx) => {
+  await executor
-    try {
+    .insertInto("hmac_secret_key")
-      await trx
+    .values({
-        .insertInto("hmac_secret_key")
+      user_id: userId,
-        .values({
+      version: 1,
-          user_id: userId,
+      state: "active",
-          version: 1,
+      master_encryption_key_version: mekVersion,
-          state: "active",
+      encrypted_key: encHsk,
-          master_encryption_key_version: mekVersion,
+    })
-          encrypted_key: encHsk,
+    .execute();
-        })
+  return { version: 1 };
-        .execute();
-      await trx
-        .insertInto("hmac_secret_key_log")
-        .values({
-          user_id: userId,
-          hmac_secret_key_version: 1,
-          timestamp: new Date(),
-          action: "create",
-          action_by: createdBy,
-        })
-        .execute();
-    } catch (e) {
-      if (e instanceof pg.DatabaseError && e.code === "23505") {
-        throw new IntegrityError("HSK already registered");
-      }
-      throw e;
-    }
-  });
 };
 
-export const getAllValidHsks = async (userId: number) => {
+export const getAllValidHsks = async (executor: Executor, userId: number) => {
-  const hsks = await db
+  const hsks = await executor
     .selectFrom("hmac_secret_key")
-    .selectAll()
+    .select(["version", "state", "master_encryption_key_version", "encrypted_key"])
     .where("user_id", "=", userId)
     .where("state", "=", "active")
     .execute();
-  return hsks.map(
+  return hsks.map(({ version, state, master_encryption_key_version, encrypted_key }) => ({
-    ({ user_id, version, state, master_encryption_key_version, encrypted_key }) =>
+    version,
-      ({
+    state,
-        userId: user_id,
+    mekVersion: master_encryption_key_version,
-        version,
+    encHsk: encrypted_key,
-        state: state as "active",
+  }));
-        mekVersion: master_encryption_key_version,
-        encHsk: encrypted_key,
-      }) satisfies Hsk,
-  );
 };

src/lib/server/db/repositories/log.ts

@@ -0,0 +1,48 @@
+import type { Insertable } from "kysely";
+import type { Executor } from "../executor.ts";
+import type {
+  CategoryLogTable,
+  DirectoryLogTable,
+  FileLogTable,
+  HskLogTable,
+  MekLogTable,
+} from "../schema/index.ts";
+
+// Audit logs of changes, timestamped when appended
+
+type NewLog<T> = Omit<Insertable<T>, "id" | "timestamp">;
+
+export const appendDirectoryLog = async (executor: Executor, log: NewLog<DirectoryLogTable>) => {
+  await executor
+    .insertInto("directory_log")
+    .values({ ...log, timestamp: new Date() })
+    .execute();
+};
+
+export const appendFileLog = async (executor: Executor, log: NewLog<FileLogTable>) => {
+  await executor
+    .insertInto("file_log")
+    .values({ ...log, timestamp: new Date() })
+    .execute();
+};
+
+export const appendCategoryLog = async (executor: Executor, log: NewLog<CategoryLogTable>) => {
+  await executor
+    .insertInto("category_log")
+    .values({ ...log, timestamp: new Date() })
+    .execute();
+};
+
+export const appendMekLog = async (executor: Executor, log: NewLog<MekLogTable>) => {
+  await executor
+    .insertInto("master_encryption_key_log")
+    .values({ ...log, timestamp: new Date() })
+    .execute();
+};
+
+export const appendHskLog = async (executor: Executor, log: NewLog<HskLogTable>) => {
+  await executor
+    .insertInto("hmac_secret_key_log")
+    .values({ ...log, timestamp: new Date() })
+    .execute();
+};

src/lib/server/db/repositories/mek.ts

@@ -1,21 +1,7 @@
-import pg from "pg";
+import type { Executor, Tx } from "../executor.ts";
-import { IntegrityError } from "../error.ts";
-import db from "../kysely.ts";
-import type { MekState } from "../schema/index.ts";
 
-interface Mek {
+// Locks the active version until the transaction ends, so that it isn't retired in the meantime
-  userId: number;
+export const lockActiveMekVersion = async (trx: Tx, userId: number) => {
-  version: number;
-  state: MekState;
-}
-
-interface ClientMekWithDetails extends Mek {
-  clientId: number;
-  encMek: string;
-  encMekSig: string;
-}
-
-export const lockActiveMekVersion = async (trx: typeof db, userId: number, version: number) => {
   const mek = await trx
     .selectFrom("master_encryption_key")
     .select("version")
@@ -24,78 +10,60 @@ export const lockActiveMekVersion = async (trx: typeof db, userId: number, versi
     .limit(1)
     .forUpdate()
     .executeTakeFirst();
-  if (mek?.version !== version) {
+  return mek?.version ?? null;
-    throw new IntegrityError("Inactive MEK version");
-  }
 };
 
-export const registerInitialMek = async (
+// Violates a unique constraint if the user already has a master key
+export const createInitialMek = async (
+  trx: Tx,
   userId: number,
-  createdBy: number,
+  clientId: number,
   encMek: string,
   encMekSig: string,
 ) => {
-  await db.transaction().execute(async (trx) => {
+  await trx
-    try {
+    .insertInto("master_encryption_key")
-      await trx
+    .values({ user_id: userId, version: 1, state: "active" })
-        .insertInto("master_encryption_key")
+    .execute();
-        .values({
+  await trx
-          user_id: userId,
+    .insertInto("client_master_encryption_key")
-          version: 1,
+    .values({
-          state: "active",
+      user_id: userId,
-        })
+      client_id: clientId,
-        .execute();
+      version: 1,
-      await trx
+      encrypted_key: encMek,
-        .insertInto("client_master_encryption_key")
+      encrypted_key_signature: encMekSig,
-        .values({
+    })
-          user_id: userId,
+    .execute();
-          client_id: createdBy,
+  return { version: 1 };
-          version: 1,
-          encrypted_key: encMek,
-          encrypted_key_signature: encMekSig,
-        })
-        .execute();
-      await trx
-        .insertInto("master_encryption_key_log")
-        .values({
-          user_id: userId,
-          master_encryption_key_version: 1,
-          timestamp: new Date(),
-          action: "create",
-          action_by: createdBy,
-        })
-        .execute();
-    } catch (e) {
-      if (e instanceof pg.DatabaseError && e.code === "23505") {
-        throw new IntegrityError("MEK already registered");
-      }
-      throw e;
-    }
-  });
 };
 
-export const getAllValidClientMeks = async (userId: number, clientId: number) => {
+export const getAllValidClientMeks = async (
-  const clientMeks = await db
+  executor: Executor,
+  userId: number,
+  clientId: number,
+) => {
+  const clientMeks = await executor
     .selectFrom("client_master_encryption_key")
     .innerJoin("master_encryption_key", (join) =>
       join
         .onRef("client_master_encryption_key.user_id", "=", "master_encryption_key.user_id")
         .onRef("client_master_encryption_key.version", "=", "master_encryption_key.version"),
     )
-    .selectAll()
+    .select([
+      "client_master_encryption_key.version",
+      "state",
+      "encrypted_key",
+      "encrypted_key_signature",
+    ])
     .where("client_master_encryption_key.user_id", "=", userId)
     .where("client_master_encryption_key.client_id", "=", clientId)
     .where("state", "in", ["active", "retired"])
     .execute();
-  return clientMeks.map(
+  return clientMeks.map(({ version, state, encrypted_key, encrypted_key_signature }) => ({
-    ({ user_id, client_id, version, state, encrypted_key, encrypted_key_signature }) =>
+    version,
-      ({
+    state: state as "active" | "retired",
-        userId: user_id,
+    encMek: encrypted_key,
-        version,
+    encMekSig: encrypted_key_signature,
-        state: state as "active" | "retired",
+  }));
-        clientId: client_id,
-        encMek: encrypted_key,
-        encMekSig: encrypted_key_signature,
-      }) satisfies ClientMekWithDetails,
-  );
 };

src/lib/server/db/repositories/session.ts

@@ -1,16 +1,15 @@
-import pg from "pg";
 import env from "#lib/server/config.ts";
-import { IntegrityError } from "../error.ts";
+import type { Executor } from "../executor.ts";
-import db from "../kysely.ts";
 
 export const createSession = async (
+  executor: Executor,
   userId: number,
   sessionId: string,
   ip: string | null,
   agent: string | null,
 ) => {
   const now = new Date();
-  await db
+  await executor
     .insertInto("session")
     .values({
       id: sessionId,
@@ -23,13 +22,15 @@ export const createSession = async (
     .execute();
 };
 
+// Returns `null` if the session doesn't exist or has expired
 export const refreshSession = async (
+  executor: Executor,
   sessionId: string,
   ip: string | null,
   agent: string | null,
 ) => {
   const now = new Date();
-  const session = await db
+  const session = await executor
     .updateTable("session")
     .set({
       last_used_at: now,
@@ -43,93 +44,74 @@ export const refreshSession = async (
   return session ? { userId: session.user_id, clientId: session.client_id } : null;
 };
 
-export const upgradeSession = async (
+// Returns whether the session has been upgraded, which fails if it was already upgraded. Violates
+// `session_ak01` if the client has another session.
+export const setSessionClient = async (executor: Executor, sessionId: string, clientId: number) => {
+  const { numUpdatedRows } = await executor
+    .updateTable("session")
+    .set({ client_id: clientId })
+    .where("id", "=", sessionId)
+    .where("client_id", "is", null)
+    .executeTakeFirst();
+  return numUpdatedRows > 0n;
+};
+
+export const deleteSession = async (executor: Executor, sessionId: string) => {
+  await executor.deleteFrom("session").where("id", "=", sessionId).execute();
+};
+
+// Deletes the other sessions of the user, or only those of the client if given
+export const deleteOtherSessions = async (
+  executor: Executor,
   userId: number,
   sessionId: string,
-  clientId: number,
+  clientId?: number,
-  force: boolean,
 ) => {
-  try {
+  await executor
-    await db.transaction().execute(async (trx) => {
-      if (force) {
-        await trx
-          .deleteFrom("session")
-          .where("id", "!=", sessionId)
-          .where("user_id", "=", userId)
-          .where("client_id", "=", clientId)
-          .execute();
-      }
-      const res = await trx
-        .updateTable("session")
-        .set({ client_id: clientId })
-        .where("id", "=", sessionId)
-        .where("client_id", "is", null)
-        .executeTakeFirst();
-      if (res.numUpdatedRows === 0n) {
-        throw new IntegrityError("Session not found");
-      }
-    });
-  } catch (e) {
-    if (e instanceof pg.DatabaseError && e.code === "23505") {
-      throw new IntegrityError("Already logged in");
-    }
-    throw e;
-  }
-};
-
-export const deleteSession = async (sessionId: string) => {
-  await db.deleteFrom("session").where("id", "=", sessionId).execute();
-};
-
-export const deleteAllOtherSessions = async (userId: number, sessionId: string) => {
-  await db
     .deleteFrom("session")
     .where("id", "!=", sessionId)
     .where("user_id", "=", userId)
+    .$if(clientId !== undefined, (qb) => qb.where("client_id", "=", clientId!))
     .execute();
 };
 
-export const cleanupExpiredSessions = async () => {
+export const deleteExpiredSessions = async (executor: Executor) => {
-  await db
+  await executor
     .deleteFrom("session")
     .where("last_used_at", "<=", new Date(Date.now() - env.session.exp))
     .execute();
 };
 
-export const registerSessionUpgradeChallenge = async (
+// Violates a unique constraint if the session already has a challenge
+export const createSessionUpgradeChallenge = async (
+  executor: Executor,
   sessionId: string,
   clientId: number,
   answer: string,
   allowedIp: string,
   expiresAt: Date,
 ) => {
-  try {
+  const { id } = await executor
-    const { id } = await db
+    .insertInto("session_upgrade_challenge")
-      .insertInto("session_upgrade_challenge")
+    .values({
-      .values({
+      session_id: sessionId,
-        session_id: sessionId,
+      client_id: clientId,
-        client_id: clientId,
+      answer,
-        answer,
+      allowed_ip: allowedIp,
-        allowed_ip: allowedIp,
+      expires_at: expiresAt,
-        expires_at: expiresAt,
+    })
-      })
+    .returning("id")
-      .returning("id")
+    .executeTakeFirstOrThrow();
-      .executeTakeFirstOrThrow();
+  return { id };
-    return { id };
-  } catch (e) {
-    if (e instanceof pg.DatabaseError && e.code === "23505") {
-      throw new IntegrityError("Challenge already registered");
-    }
-    throw e;
-  }
 };
 
 export const consumeSessionUpgradeChallenge = async (
+  executor: Executor,
   challengeId: number,
   sessionId: string,
   ip: string,
 ) => {
-  const challenge = await db
+  const challenge = await executor
     .deleteFrom("session_upgrade_challenge")
     .where("id", "=", challengeId)
     .where("session_id", "=", sessionId)
@@ -140,6 +122,9 @@ export const consumeSessionUpgradeChallenge = async (
   return challenge ? { clientId: challenge.client_id, answer: challenge.answer } : null;
 };
 
-export const cleanupExpiredSessionUpgradeChallenges = async () => {
+export const deleteExpiredSessionUpgradeChallenges = async (executor: Executor) => {
-  await db.deleteFrom("session_upgrade_challenge").where("expires_at", "<=", new Date()).execute();
+  await executor
+    .deleteFrom("session_upgrade_challenge")
+    .where("expires_at", "<=", new Date())
+    .execute();
 };

src/lib/server/db/repositories/upload.ts

@@ -40,10 +40,14 @@ export const createFileUploadSession = async (
   params: Omit<FileUploadSession, "type" | "bitmap" | "uploadedChunks">,
 ) => {
   await db.transaction().execute(async (trx) => {
-    await lockActiveMekVersion(trx, params.userId, params.mekVersion);
+    if ((await lockActiveMekVersion(trx, params.userId)) !== params.mekVersion) {
+      throw new IntegrityError("Inactive MEK version");
+    }
 
     if (params.hskVersion) {
-      await lockActiveHskVersion(trx, params.userId, params.hskVersion);
+      if ((await lockActiveHskVersion(trx, params.userId)) !== params.hskVersion) {
+        throw new IntegrityError("Inactive HSK version");
+      }
     }
 
     await ensureDirectoryExists(trx, params.userId, params.parentId);

src/lib/server/db/repositories/user.ts

@@ -1,4 +1,6 @@
-import db from "../kysely.ts";
+import type { Selectable } from "kysely";
+import type { Executor } from "../executor.ts";
+import type { UserTable } from "../schema/index.ts";
 
 interface User {
   id: number;
@@ -7,26 +9,33 @@ interface User {
   password: string;
 }
 
-export const getUser = async (userId: number) => {
+const toUser = (row: Selectable<UserTable>): User => ({
-  const user = await db
+  id: row.id,
+  email: row.email,
+  nickname: row.nickname,
+  password: row.password,
+});
+
+export const getUser = async (executor: Executor, userId: number) => {
+  const user = await executor
     .selectFrom("user")
     .selectAll()
     .where("id", "=", userId)
     .limit(1)
     .executeTakeFirst();
-  return user ? (user satisfies User) : null;
+  return user ? toUser(user) : null;
 };
 
-export const getUserByEmail = async (email: string) => {
+export const getUserByEmail = async (executor: Executor, email: string) => {
-  const user = await db
+  const user = await executor
     .selectFrom("user")
     .selectAll()
     .where("email", "=", email)
     .limit(1)
     .executeTakeFirst();
-  return user ? (user satisfies User) : null;
+  return user ? toUser(user) : null;
 };
 
-export const setUserPassword = async (userId: number, password: string) => {
+export const setUserPassword = async (executor: Executor, userId: number, password: string) => {
-  await db.updateTable("user").set({ password }).where("id", "=", userId).execute();
+  await executor.updateTable("user").set({ password }).where("id", "=", userId).execute();
 };

src/lib/server/services/auth.ts

@@ -1,5 +1,12 @@
 import argon2 from "argon2";
-import { ClientRepo, SessionRepo, UserRepo } from "#lib/server/db/index.ts";
+import {
+  ClientRepo,
+  SessionRepo,
+  UserRepo,
+  db,
+  isUniqueViolation,
+  transaction,
+} from "#lib/server/db/index.ts";
 import env from "#lib/server/config.ts";
 import { issueSessionId } from "#lib/server/crypto.ts";
 import { AppError, invalidSessionError } from "#lib/server/errors.ts";
@@ -8,18 +15,18 @@ import { createChallenge, verifyChallengeAnswer } from "./client.ts";
 const MIN_PASSWORD_LENGTH = 8;
 
 export const login = async (email: string, password: string, ip: string, userAgent: string) => {
-  const user = await UserRepo.getUserByEmail(email);
+  const user = await UserRepo.getUserByEmail(db, 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);
+  await SessionRepo.createSession(db, user.id, sessionId, ip, userAgent);
   return { sessionIdSigned };
 };
 
 export const logout = async (sessionId: string) => {
-  await SessionRepo.deleteSession(sessionId);
+  await SessionRepo.deleteSession(db, sessionId);
 };
 
 export const changePassword = async (
@@ -34,15 +41,18 @@ export const changePassword = async (
     throw new AppError("BAD_REQUEST", "Too short password");
   }
 
-  const user = await UserRepo.getUser(userId);
+  const user = await UserRepo.getUser(db, userId);
   if (!user) {
     throw invalidSessionError();
   } else if (!(await argon2.verify(user.password, oldPassword))) {
     throw new AppError("FORBIDDEN", "Invalid password");
   }
 
-  await UserRepo.setUserPassword(userId, await argon2.hash(newPassword));
+  const passwordHash = await argon2.hash(newPassword);
-  await SessionRepo.deleteAllOtherSessions(userId, sessionId);
+  await transaction(async (trx) => {
+    await UserRepo.setUserPassword(trx, userId, passwordHash);
+    await SessionRepo.deleteOtherSessions(trx, userId, sessionId);
+  });
 };
 
 export const requestSessionUpgrade = async (
@@ -52,25 +62,33 @@ export const requestSessionUpgrade = async (
   encPubKey: string,
   sigPubKey: string,
 ) => {
-  const client = await ClientRepo.getClientByPubKeys(encPubKey, sigPubKey);
+  const client = await ClientRepo.getClientByPubKeys(db, encPubKey, sigPubKey);
   if (!client) {
     throw new AppError("UNAUTHORIZED", "Invalid public key(s)");
   }
 
-  const userClient = await ClientRepo.getUserClient(userId, client.id);
+  const userClient = await ClientRepo.getUserClient(db, userId, client.id);
   if (!userClient || userClient.state === "challenging") {
     throw new AppError("FORBIDDEN", "Unregistered client");
   }
 
-  return await createChallenge(encPubKey, (answer) =>
+  try {
-    SessionRepo.registerSessionUpgradeChallenge(
+    return await createChallenge(encPubKey, (answer) =>
-      sessionId,
+      SessionRepo.createSessionUpgradeChallenge(
-      client.id,
+        db,
-      answer,
+        sessionId,
-      ip,
+        client.id,
-      new Date(Date.now() + env.challenge.sessionUpgradeExp),
+        answer,
-    ),
+        ip,
-  );
+        new Date(Date.now() + env.challenge.sessionUpgradeExp),
+      ),
+    );
+  } catch (e) {
+    if (isUniqueViolation(e)) {
+      throw new AppError("CONFLICT", "Challenge already registered");
+    }
+    throw e;
+  }
 };
 
 export const verifySessionUpgrade = async (
@@ -81,14 +99,34 @@ export const verifySessionUpgrade = async (
   answerSig: string,
   force: boolean,
 ) => {
-  const challenge = await SessionRepo.consumeSessionUpgradeChallenge(challengeId, sessionId, ip);
+  const challenge = await SessionRepo.consumeSessionUpgradeChallenge(
+    db,
+    challengeId,
+    sessionId,
+    ip,
+  );
   const client = await verifyChallengeAnswer(challenge, answerSig);
-  await SessionRepo.upgradeSession(userId, sessionId, client.id, force);
+
+  try {
+    await transaction(async (trx) => {
+      if (force) {
+        await SessionRepo.deleteOtherSessions(trx, userId, sessionId, client.id);
+      }
+      if (!(await SessionRepo.setSessionClient(trx, sessionId, client.id))) {
+        throw new AppError("UNAUTHORIZED", "Session not found");
+      }
+    });
+  } catch (e) {
+    if (isUniqueViolation(e, "session_ak01")) {
+      throw new AppError("CONFLICT", "Already logged in");
+    }
+    throw e;
+  }
 };
 
 export const cleanupExpiredSessions = async () => {
   await Promise.all([
-    SessionRepo.cleanupExpiredSessions(),
+    SessionRepo.deleteExpiredSessions(db),
-    SessionRepo.cleanupExpiredSessionUpgradeChallenges(),
+    SessionRepo.deleteExpiredSessionUpgradeChallenges(db),
   ]);
 };

src/lib/server/services/client.ts

@@ -1,4 +1,4 @@
-import { ClientRepo } from "#lib/server/db/index.ts";
+import { ClientRepo, db, isUniqueViolation, transaction } from "#lib/server/db/index.ts";
 import env from "#lib/server/config.ts";
 import { generateChallenge, verifyPubKey, verifySignature } from "#lib/server/crypto.ts";
 import { AppError } from "#lib/server/errors.ts";
@@ -22,7 +22,7 @@ export const verifyChallengeAnswer = async (
     throw new AppError("FORBIDDEN", "Invalid challenge answer");
   }
 
-  const client = await ClientRepo.getClient(challenge.clientId);
+  const client = await ClientRepo.getClient(db, challenge.clientId);
   if (!client) {
     throw new AppError("INTERNAL_SERVER_ERROR", "Invalid challenge answer");
   } else if (
@@ -33,28 +33,66 @@ export const verifyChallengeAnswer = async (
   return client;
 };
 
+const addExistingClient = async (userId: number, clientId: number) => {
+  try {
+    await ClientRepo.createUserClient(db, userId, clientId);
+  } catch (e) {
+    if (isUniqueViolation(e)) {
+      throw new AppError("CONFLICT", "User client already exists");
+    }
+    throw e;
+  }
+};
+
+const addNewClient = async (userId: number, encPubKey: string, sigPubKey: string) => {
+  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)");
+  }
+
+  const publicKeysRegistered = () => new AppError("CONFLICT", "Public key(s) already registered");
+  try {
+    // Serializable, as a key must not be used as either key of another client, which can't be
+    // ensured by unique constraints
+    return await transaction(
+      async (trx) => {
+        if (await ClientRepo.isAnyPublicKeyRegistered(trx, encPubKey, sigPubKey)) {
+          throw publicKeysRegistered();
+        }
+
+        const { id } = await ClientRepo.createClient(trx, encPubKey, sigPubKey);
+        await ClientRepo.createUserClient(trx, userId, id);
+        return id;
+      },
+      { isolationLevel: "serializable" },
+    );
+  } catch (e) {
+    if (isUniqueViolation(e)) {
+      throw publicKeysRegistered();
+    }
+    throw e;
+  }
+};
+
 export const registerClient = async (
   userId: number,
   ip: string,
   encPubKey: string,
   sigPubKey: string,
 ) => {
-  const getClientId = async () => {
+  const client = await ClientRepo.getClientByPubKeys(db, encPubKey, sigPubKey);
-    const client = await ClientRepo.getClientByPubKeys(encPubKey, sigPubKey);
+  let clientId;
-    if (client) {
+  if (client) {
-      await ClientRepo.createUserClient(userId, client.id);
+    await addExistingClient(userId, client.id);
-      return client.id;
+    clientId = client.id;
-    } else if (encPubKey === sigPubKey) {
+  } else {
-      throw new AppError("BAD_REQUEST", "Same public keys");
+    clientId = await addNewClient(userId, encPubKey, sigPubKey);
-    } 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(
+    ClientRepo.createUserClientChallenge(
+      db,
       userId,
       clientId,
       answer,
@@ -70,11 +108,14 @@ export const verifyClient = async (
   challengeId: number,
   answerSig: string,
 ) => {
-  const challenge = await ClientRepo.consumeUserClientChallenge(challengeId, userId, ip);
+  const challenge = await ClientRepo.consumeUserClientChallenge(db, challengeId, userId, ip);
   const client = await verifyChallengeAnswer(challenge, answerSig);
-  await ClientRepo.setUserClientStateToPending(userId, client.id);
+  await ClientRepo.transitionUserClientState(db, userId, client.id, {
+    from: "challenging",
+    to: "pending",
+  });
 };
 
 export const cleanupExpiredChallenges = async () => {
-  await ClientRepo.cleanupExpiredUserClientChallenges();
+  await ClientRepo.deleteExpiredUserClientChallenges(db);
 };

src/lib/server/services/common.ts

@@ -1,3 +1,4 @@
+import { 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
@@ -8,3 +9,17 @@ export const ensureFreshDekVersion = (dekVersion: Date) => {
     throw new AppError("BAD_REQUEST", "Invalid DEK version");
   }
 };
+
+// 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) => {
+  if ((await MekRepo.lockActiveMekVersion(trx, userId)) !== mekVersion) {
+    throw new AppError("BAD_REQUEST", "Inactive MEK version");
+  }
+};
+
+export const ensureActiveHskVersion = async (trx: Tx, userId: number, hskVersion: number) => {
+  if ((await HskRepo.lockActiveHskVersion(trx, userId)) !== hskVersion) {
+    throw new AppError("BAD_REQUEST", "Inactive HSK version");
+  }
+};

src/lib/server/services/key.ts

@@ -1,26 +1,34 @@
-import { ClientRepo, HskRepo, MekRepo } from "#lib/server/db/index.ts";
+import {
+  ClientRepo,
+  HskRepo,
+  LogRepo,
+  MekRepo,
+  db,
+  isUniqueViolation,
+  transaction,
+} from "#lib/server/db/index.ts";
 import { verifySignature } from "#lib/server/crypto.ts";
 import { AppError, invalidSessionError } from "#lib/server/errors.ts";
 import { serializeMasterKeyWrapped } from "#lib/shared/crypto/index.ts";
+import { ensureActiveMekVersion } from "./common.ts";
 
 export const getClientMeks = async (userId: number, clientId: number) => {
-  return await MekRepo.getAllValidClientMeks(userId, clientId);
+  return await MekRepo.getAllValidClientMeks(db, userId, clientId);
 };
 
 const verifyClientEncMekSig = async (
-  userId: number,
   clientId: number,
   version: number,
   encMek: string,
   encMekSig: string,
 ) => {
-  const userClient = await ClientRepo.getUserClientWithDetails(userId, clientId);
+  const client = await ClientRepo.getClient(db, clientId);
-  if (!userClient) {
+  if (!client) {
     throw invalidSessionError();
   }
 
   const data = serializeMasterKeyWrapped(encMek, version);
-  return verifySignature(Buffer.from(data), encMekSig, userClient.sigPubKey);
+  return verifySignature(Buffer.from(data), encMekSig, client.sigPubKey);
 };
 
 export const registerInitialMek = async (
@@ -29,16 +37,35 @@ export const registerInitialMek = async (
   encMek: string,
   encMekSig: string,
 ) => {
-  if (!(await verifyClientEncMekSig(userId, clientId, 1, encMek, encMekSig))) {
+  if (!(await verifyClientEncMekSig(clientId, 1, encMek, encMekSig))) {
     throw new AppError("BAD_REQUEST", "Invalid signature");
   }
 
-  await MekRepo.registerInitialMek(userId, clientId, encMek, encMekSig);
+  try {
-  await ClientRepo.setUserClientStateToActive(userId, clientId);
+    // The client that registered the master key is approved to access it at the same time
+    await transaction(async (trx) => {
+      const { version } = await MekRepo.createInitialMek(trx, userId, clientId, encMek, encMekSig);
+      await LogRepo.appendMekLog(trx, {
+        user_id: userId,
+        master_encryption_key_version: version,
+        action: "create",
+        action_by: clientId,
+      });
+      await ClientRepo.transitionUserClientState(trx, userId, clientId, {
+        from: "pending",
+        to: "active",
+      });
+    });
+  } catch (e) {
+    if (isUniqueViolation(e)) {
+      throw new AppError("CONFLICT", "MEK already registered");
+    }
+    throw e;
+  }
 };
 
 export const getHsks = async (userId: number) => {
-  return await HskRepo.getAllValidHsks(userId);
+  return await HskRepo.getAllValidHsks(db, userId);
 };
 
 export const registerInitialHsk = async (
@@ -47,5 +74,21 @@ export const registerInitialHsk = async (
   mekVersion: number,
   encHsk: string,
 ) => {
-  await HskRepo.registerInitialHsk(userId, clientId, mekVersion, encHsk);
+  try {
+    await transaction(async (trx) => {
+      await ensureActiveMekVersion(trx, userId, mekVersion);
+      const { version } = await HskRepo.createInitialHsk(trx, userId, mekVersion, encHsk);
+      await LogRepo.appendHskLog(trx, {
+        user_id: userId,
+        hmac_secret_key_version: version,
+        action: "create",
+        action_by: clientId,
+      });
+    });
+  } catch (e) {
+    if (isUniqueViolation(e)) {
+      throw new AppError("CONFLICT", "HSK already registered");
+    }
+    throw e;
+  }
 };

src/lib/server/services/user.ts

@@ -1,8 +1,8 @@
-import { UserRepo } from "#lib/server/db/index.ts";
+import { UserRepo, db } from "#lib/server/db/index.ts";
 import { invalidSessionError } from "#lib/server/errors.ts";
 
 export const getUser = async (userId: number) => {
-  const user = await UserRepo.getUser(userId);
+  const user = await UserRepo.getUser(db, userId);
   if (!user) {
     throw invalidSessionError();
   }