Merge pull request #713 from mikewheeleer/codex/issue-662

feat: route read-only queries to an optional replica

Author: Jagadeeshftw

docs/db/schema.md

@@ -84,6 +84,15 @@ The schema enforces eight invariants (I1–I8), verified by
   be declared in Drizzle's metadata (`schema.ts`). The test suite detects drift
   between these two representations.
 
+## Read replica routing
+
+Set `POSTGRES_READ_REPLICA_CONNECTION_STRING` to route the read-only analytics,
+transactions, and indexed-data handlers through a separate pool. When unset,
+`readDb` is an alias for the primary database and behavior is unchanged. The
+write and read-after-write-sensitive routes continue using `db` so callers do
+not observe replica lag immediately after a mutation. `getReadPoolStats()`
+reports the active read pool without exposing connection details.
+
 ## Tables
 
 | Table | SQL Name | Description |

src/config.ts

@@ -23,6 +23,9 @@ export const EnvSchema = z
     /** Database connection string for indexed data (required for startup and health checks). */
     POSTGRES_CONNECTION_STRING: z.string().url(),
 
+    /** Optional read-only replica; primary is used when unset. */
+    POSTGRES_READ_REPLICA_CONNECTION_STRING: z.string().url().optional(),
+
     // ═════════════════════════════════════════════════════════════════════
     // OPTIONAL — fall back to documented defaults when unset.
     // ═════════════════════════════════════════════════════════════════════

src/db/index.ts

@@ -23,47 +23,42 @@ function maskConnectionString(connectionString: string): string {
   }
 }
 
-// Create connection pool with proper error handling.
-let pool: Pool;
-try {
-  const connectionString = env.POSTGRES_CONNECTION_STRING;
-
-  if (!connectionString || typeof connectionString !== "string") {
-    console.warn("[db] POSTGRES_CONNECTION_STRING not set, database features will be unavailable");
-    pool = new Pool({
-      connectionString: "postgresql://localhost:5432/stellopay_indexer",
-      ...poolTuning,
+function createPool(connectionString: string | undefined, label: string, tuning = poolTuning): Pool {
+  try {
+    if (!connectionString) console.warn(`[db] ${label} connection string not set, using local fallback`);
+    const url = new URL(connectionString ?? "postgresql://localhost:5432/stellopay_indexer");
+    if (url.password === null || url.password === undefined) url.password = "";
+    const createdPool = new Pool({ connectionString: url.toString(), ...tuning });
+    createdPool.on("error", (error: Error & { code?: string }) => {
+      console.error(`[db] Unexpected ${label} pool error`, {
+        message: error.message,
+        code: error.code,
+        stack: error.stack,
+      });
     });
-  } else {
-    const url = new URL(connectionString);
-    if (url.password === null || url.password === undefined) {
-      url.password = "";
-    }
-
-    pool = new Pool({
-      connectionString: url.toString(),
-      ...poolTuning,
+    return createdPool;
+  } catch (error) {
+    console.error(`[db] Failed to initialize ${label} connection pool`, {
+      message: error instanceof Error ? error.message : String(error),
+    });
+    return new Pool({
+      connectionString: "postgresql://localhost:5432/stellopay_indexer",
+      ...tuning,
     });
   }
-} catch (error) {
-  console.error("[db] Failed to initialize connection pool", {
-    message: error instanceof Error ? error.message : String(error),
-  });
-  pool = new Pool({
-    connectionString: "postgresql://localhost:5432/stellopay_indexer",
-    ...poolTuning,
-  });
 }
 
-pool.on("error", (error: Error & { code?: string }) => {
-  console.error("[db] Unexpected pool error", {
-    message: error.message,
-    code: error.code,
-    stack: error.stack,
-  });
-});
+const pool = createPool(env.POSTGRES_CONNECTION_STRING, "primary");
+const readPool = env.POSTGRES_READ_REPLICA_CONNECTION_STRING
+  ? createPool(env.POSTGRES_READ_REPLICA_CONNECTION_STRING, "read replica", {
+      ...poolTuning,
+      max: Math.max(1, Math.floor(env.DB_POOL_MAX / 2)),
+    })
+  : null;
 
 export const db = drizzle(pool, { schema });
+/** Read-only queries use the replica when configured, otherwise the primary. */
+export const readDb = drizzle(readPool ?? pool, { schema });
 export { schema };
 
 /** Current utilization counters for the shared Postgres connection pool. */
@@ -81,14 +76,23 @@ export interface PoolStats {
  * details are included, and reading the snapshot does not acquire a client.
  */
 export function getPoolStats(): PoolStats {
-  const total = pool.totalCount;
-  const idle = pool.idleCount;
+  return getStatsForPool(pool);
+}
+
+/** Returns the replica pool snapshot, or the primary snapshot when disabled. */
+export function getReadPoolStats(): PoolStats {
+  return getStatsForPool(readPool ?? pool);
+}
+
+function getStatsForPool(activePool: Pool): PoolStats {
+  const total = activePool.totalCount;
+  const idle = activePool.idleCount;
 
   return {
     total,
     idle,
     active: total - idle,
-    waiting: pool.waitingCount,
+    waiting: activePool.waitingCount,
   };
 }
 
@@ -133,6 +137,7 @@ export async function waitForDbReadiness(): Promise<void> {
 export async function closePool(): Promise<void> {
   console.log("[db] Closing Postgres connection pool...");
   await pool.end();
+  if (readPool && readPool !== pool) await readPool.end();
   console.log("[db] Postgres connection pool closed.");
 }
 

src/routes/analytics.ts

@@ -1,7 +1,7 @@
 import { createHash } from "node:crypto";
 import { Router } from "express";
 import { z } from "zod";
-import { db, schema } from "../db/index.js";
+import { readDb, schema } from "../db/index.js";
 import { asc, eq, and, gt, gte, lte, or, sql } from "drizzle-orm";
 import { StarknetAddress } from "../utils/validation.js";
 import { DEFAULT_TOKEN_DECIMALS } from "../utils/codec.js";
@@ -287,7 +287,7 @@ analyticsRouter.get("/analytics/:user_address", async (req, res, next) => {
             : undefined;
 
           const whereCondition = cursorFilter ? and(baseFilter, cursorFilter) : baseFilter;
-          const query = db
+          const query = readDb
             .select({
               id: schema.payments.id,
               createdAt: schema.payments.createdAt,
@@ -327,7 +327,7 @@ analyticsRouter.get("/analytics/:user_address", async (req, res, next) => {
             : undefined;
 
           const whereCondition = cursorFilter ? and(baseFilter, cursorFilter) : baseFilter;
-          const query = db
+          const query = readDb
             .select({
               id: schema.escrowEvents.id,
               createdAt: schema.escrowEvents.createdAt,
@@ -369,7 +369,7 @@ analyticsRouter.get("/analytics/:user_address", async (req, res, next) => {
             : undefined;
 
           const whereCondition = cursorFilter ? and(baseFilter, cursorFilter) : baseFilter;
-          const query = db
+          const query = readDb
             .select({
               id: schema.agreementEvents.id,
               createdAt: schema.agreementEvents.createdAt,

src/routes/indexed.ts

@@ -1,6 +1,6 @@
 import { Router } from "express";
 import { z } from "zod";
-import { db, schema } from "../db/index.js";
+import { readDb, schema } from "../db/index.js";
 import { eq, and, or, desc } from "drizzle-orm";
 import { StarknetAddress, AgreementId, parsePagination } from "../utils/validation.js";
 import { defaults, env } from "../config.js";
@@ -238,7 +238,7 @@ async function resolveCheckpoint(): Promise<{
   checkpointBlock: number;
   records: Array<{ blockNumber: unknown }>;
 }> {
-  const records = await db
+  const records = await readDb
     .select({ blockNumber: schema.agreementEvents.blockNumber })
     .from(schema.agreementEvents)
     .orderBy(desc(schema.agreementEvents.blockNumber))
@@ -393,7 +393,7 @@ indexedRouter.get(
       const { limit, offset } = parsePagination(req.query);
 
       const [agreements, employeeAgreements] = await Promise.all([
-        db
+        readDb
           .select()
           .from(schema.agreements)
           .where(
@@ -409,7 +409,7 @@ indexedRouter.get(
           .limit(limit)
           .offset(offset),
 
-        db
+        readDb
           .select({
             agreement: schema.agreements,
           })
@@ -487,7 +487,7 @@ indexedRouter.get("/indexed/agreement/:contract_address/:agreement_id", async (r
     }
     const agreementId = AgreementId.parse(req.params.agreement_id);
 
-    const agreement = await db
+    const agreement = await readDb
       .select()
       .from(schema.agreements)
       .where(
@@ -504,23 +504,23 @@ indexedRouter.get("/indexed/agreement/:contract_address/:agreement_id", async (r
     }
 
     const [events, payments, milestones, employees, escrowEvents] = await Promise.all([
-      db.select().from(schema.agreementEvents)
+      readDb.select().from(schema.agreementEvents)
         .where(eq(schema.agreementEvents.agreementId, agreementId))
         .orderBy(desc(schema.agreementEvents.blockNumber)).limit(MAX_INTERNAL_LIMIT),
 
-      db.select().from(schema.payments)
+      readDb.select().from(schema.payments)
         .where(eq(schema.payments.agreementId, agreementId))
         .orderBy(desc(schema.payments.blockNumber)).limit(MAX_INTERNAL_LIMIT),
 
-      db.select().from(schema.milestones)
+      readDb.select().from(schema.milestones)
         .where(eq(schema.milestones.agreementId, agreementId))
         .orderBy(schema.milestones.milestoneId).limit(MAX_INTERNAL_LIMIT),
 
-      db.select().from(schema.employees)
+      readDb.select().from(schema.employees)
         .where(eq(schema.employees.agreementId, agreementId))
         .orderBy(schema.employees.employeeIndex).limit(MAX_INTERNAL_LIMIT),
 
-      db.select().from(schema.escrowEvents)
+      readDb.select().from(schema.escrowEvents)
         .where(eq(schema.escrowEvents.agreementId, agreementId))
         .orderBy(desc(schema.escrowEvents.blockNumber)).limit(MAX_INTERNAL_LIMIT),
     ]);
@@ -586,7 +586,7 @@ indexedRouter.get("/indexed/payments/user/:user_address", async (req, res, next)
     const userAddress = StarknetAddress.parse(req.params.user_address);
     const { limit, offset } = parsePagination(req.query);
 
-    const payments = await db
+    const payments = await readDb
       .select()
       .from(schema.payments)
       .where(or(eq(schema.payments.from, userAddress), eq(schema.payments.to, userAddress)))
@@ -642,7 +642,7 @@ indexedRouter.get(
       }
       const agreementId = AgreementId.parse(req.params.agreement_id);
 
-      const escrowEvents = await db
+      const escrowEvents = await readDb
         .select()
         .from(schema.escrowEvents)
         .where(
@@ -704,4 +704,3 @@ indexedRouter.get(
 );
 
 export default indexedRouter;
-

src/routes/transactions.ts

@@ -27,7 +27,7 @@
 import { Router } from "express";
 import type { Request, Response } from "express";
 import { z } from "zod";
-import { db, schema } from "../db/index.js";
+import { readDb, schema } from "../db/index.js";
 import { eq, and, or, desc, gte, lte, inArray, sql, count } from "drizzle-orm";
 import { agreementContract } from "../starknet/client.js";
 import { toHexString } from "../utils/codec.js";
@@ -859,31 +859,31 @@ async function fetchAndBuildTransactions(
     milestoneEventsData,
   ] = await Promise.all([
     // ── counts ────────────────────────────────────────────────────────
-    db
+    readDb
       .select({ count: count() })
       .from(schema.payments)
       .where(conds.payments),
-    db
+    readDb
       .select({ count: count() })
       .from(schema.escrowEvents)
       .where(conds.escrowEvents),
-    db
+    readDb
       .select({ count: count() })
       .from(schema.agreementEvents)
       .innerJoin(
         schema.agreements,
         eq(schema.agreementEvents.agreementId, schema.agreements.id),
       )
       .where(conds.agreementEvents),
-    db
+    readDb
       .select({ count: count() })
       .from(schema.employees)
       .leftJoin(
         schema.agreements,
         eq(schema.employees.agreementId, schema.agreements.id),
       )
       .where(conds.employees),
-    db
+    readDb
       .select({ count: count() })
       .from(schema.milestones)
       .leftJoin(
@@ -892,19 +892,19 @@ async function fetchAndBuildTransactions(
       )
       .where(conds.milestones),
     // ── data ──────────────────────────────────────────────────────────
-    db
+    readDb
       .select()
       .from(schema.payments)
       .where(conds.payments)
       .orderBy(desc(schema.payments.createdAt), desc(schema.payments.id))
       .limit(queryLimit),
-    db
+    readDb
       .select()
       .from(schema.escrowEvents)
       .where(conds.escrowEvents)
       .orderBy(desc(schema.escrowEvents.createdAt), desc(schema.escrowEvents.id))
       .limit(queryLimit),
-    db
+    readDb
       .select({
         id: schema.agreementEvents.id,
         agreementId: schema.agreementEvents.agreementId,
@@ -925,7 +925,7 @@ async function fetchAndBuildTransactions(
       .where(conds.agreementEvents)
       .orderBy(desc(schema.agreementEvents.createdAt), desc(schema.agreementEvents.id))
       .limit(queryLimit),
-    db
+    readDb
       .select({
         id: schema.employees.id,
         agreementId: schema.employees.agreementId,
@@ -947,7 +947,7 @@ async function fetchAndBuildTransactions(
       .where(conds.employees)
       .orderBy(desc(schema.employees.createdAt), desc(schema.employees.id))
       .limit(queryLimit),
-    db
+    readDb
       .select({
         id: schema.milestones.id,
         agreementId: schema.milestones.agreementId,
@@ -997,7 +997,7 @@ async function fetchAndBuildTransactions(
 
   const escrowAgreements =
     escrowAgreementIds.length > 0
-      ? await db
+      ? await readDb
           .select({
             id: schema.agreements.id,
             token: schema.agreements.token,