add safeRead

server/auth/sessions/app.ts

@@ -3,7 +3,14 @@ import {
     encodeHexLowerCase
 } from "@oslojs/encoding";
 import { sha256 } from "@oslojs/crypto/sha2";
-import { resourceSessions, Session, sessions, User, users } from "@server/db";
+import {
+    resourceSessions,
+    safeRead,
+    Session,
+    sessions,
+    User,
+    users
+} from "@server/db";
 import { db } from "@server/db";
 import { eq, inArray } from "drizzle-orm";
 import config from "@server/lib/config";
@@ -54,11 +61,15 @@ export async function validateSessionToken(
     const sessionId = encodeHexLowerCase(
         sha256(new TextEncoder().encode(token))
     );
-    const result = await db
+
-        .select({ user: users, session: sessions })
+    const result = await safeRead((db) =>
-        .from(sessions)
+        db
-        .innerJoin(users, eq(sessions.userId, users.userId))
+            .select({ user: users, session: sessions })
-        .where(eq(sessions.sessionId, sessionId));
+            .from(sessions)
+            .innerJoin(users, eq(sessions.userId, users.userId))
+            .where(eq(sessions.sessionId, sessionId))
+    );
+
     if (result.length < 1) {
         return { session: null, user: null };
     }

server/auth/sessions/resource.ts

@@ -1,7 +1,7 @@
 import { encodeHexLowerCase } from "@oslojs/encoding";
 import { sha256 } from "@oslojs/crypto/sha2";
 import { resourceSessions, ResourceSession } from "@server/db";
-import { db } from "@server/db";
+import { db, safeRead } from "@server/db";
 import { eq, and } from "drizzle-orm";
 import config from "@server/lib/config";
 
@@ -66,15 +66,17 @@ export async function validateResourceSessionToken(
     const sessionId = encodeHexLowerCase(
         sha256(new TextEncoder().encode(token))
     );
-    const result = await db
+    const result = await safeRead((db) =>
-        .select()
+        db
-        .from(resourceSessions)
+            .select()
-        .where(
+            .from(resourceSessions)
-            and(
+            .where(
-                eq(resourceSessions.sessionId, sessionId),
+                and(
-                eq(resourceSessions.resourceId, resourceId)
+                    eq(resourceSessions.sessionId, sessionId),
+                    eq(resourceSessions.resourceId, resourceId)
+                )
             )
-        );
+    );
 
     if (result.length < 1) {
         return { resourceSession: null };

server/db/pg/index.ts

@@ -1,4 +1,5 @@
 export * from "./driver";
+export * from "./safeRead";
 export * from "./schema/schema";
 export * from "./schema/privateSchema";
 export * from "./migrate";

server/db/pg/safeRead.ts

@@ -0,0 +1,24 @@
+import { db, primaryDb } from "./driver";
+
+/**
+ * Runs a read query with replica fallback for Postgres.
+ * Executes the query against the replica first (when replicas exist).
+ * If the query throws or returns no data (null, undefined, or empty array),
+ * runs the same query against the primary.
+ */
+export async function safeRead<T>(
+    query: (d: typeof db | typeof primaryDb) => Promise<T>
+): Promise<T> {
+    try {
+        const result = await query(db);
+        if (result === undefined || result === null) {
+            return query(primaryDb);
+        }
+        if (Array.isArray(result) && result.length === 0) {
+            return query(primaryDb);
+        }
+        return result;
+    } catch {
+        return query(primaryDb);
+    }
+}

server/db/sqlite/index.ts

@@ -1,4 +1,5 @@
 export * from "./driver";
+export * from "./safeRead";
 export * from "./schema/schema";
 export * from "./schema/privateSchema";
 export * from "./migrate";

server/db/sqlite/safeRead.ts

@@ -0,0 +1,11 @@
+import { db } from "./driver";
+
+/**
+ * Runs a read query. For SQLite there is no replica/primary distinction,
+ * so the query is executed once against the database.
+ */
+export async function safeRead<T>(
+    query: (d: typeof db) => Promise<T>
+): Promise<T> {
+    return query(db);
+}