Build Multi-Region Data Replication

// src/db/multi-region.ts: region-aware routing with read replicas and LSN-based read-after-write
import { Pool, QueryResult } from "pg";
import { Redis } from "ioredis";

interface Region {
  name: string;
  isPrimary: boolean;
  pool: Pool;
  redis: Redis;
}

const regions = new Map<string, Region>();

const configs = [
  { name: "us-east-1", isPrimary: true, db: process.env.DB_URL_US!, redis: process.env.REDIS_URL_US! },
  { name: "eu-west-1", isPrimary: false, db: process.env.DB_URL_EU!, redis: process.env.REDIS_URL_EU! },
  { name: "ap-northeast-1", isPrimary: false, db: process.env.DB_URL_AP!, redis: process.env.REDIS_URL_AP! },
];

for (const c of configs) {
  const pool = new Pool({ connectionString: c.db, max: 20 });
  const redis = new Redis(c.redis);
  // Without these handlers a dropped idle connection (a standby restarting) crashes the process
  pool.on("error", () => {});
  redis.on("error", () => {});
  regions.set(c.name, { name: c.name, isPrimary: c.isPrimary, pool, redis });
}

function closest(userRegion: string): Region {
  return regions.get(userRegion) ?? primary();
}

function primary(): Region {
  for (const r of regions.values()) if (r.isPrimary) return r;
  throw new Error("No primary region configured");
}

// Eventually consistent read from the nearest standby
export function readLocal(userRegion: string, sql: string, params?: unknown[]): Promise<QueryResult> {
  return closest(userRegion).pool.query(sql, params);
}

// Strongly consistent read from the primary
export function readPrimary(sql: string, params?: unknown[]): Promise<QueryResult> {
  return primary().pool.query(sql, params);
}

// Write on the primary, delete cache keys in every region, return the WAL position of the write
export async function writePrimary(
  sql: string,
  params: unknown[],
  cacheKeys: string[] = []
): Promise<{ result: QueryResult; lsn: string }> {
  const p = primary();
  const result = await p.pool.query(sql, params);
  // The WAL position after our commit is at least the commit record's position
  const { rows } = await p.pool.query("SELECT pg_current_wal_lsn()::text AS lsn");

  if (cacheKeys.length > 0) {
    await Promise.allSettled(
      [...regions.values()].flatMap((r) => cacheKeys.map((k) => r.redis.del(k)))
    );
  }
  return { result, lsn: rows[0].lsn };
}

// Read-after-write: use the standby only if it has replayed the client's LSN
export async function readConsistent(
  userRegion: string,
  sql: string,
  params: unknown[],
  minLsn: string
): Promise<QueryResult> {
  const region = closest(userRegion);
  if (region.isPrimary) return region.pool.query(sql, params);

  const { rows } = await region.pool.query(
    "SELECT pg_last_wal_replay_lsn() >= $1::pg_lsn AS caught_up",
    [minLsn]
  );
  return rows[0].caught_up ? region.pool.query(sql, params) : readPrimary(sql, params);
}

// Cache-first read with regional locality
export async function cachedRead(
  userRegion: string,
  cacheKey: string,
  sql: string,
  params: unknown[],
  ttlSeconds = 60
): Promise<{ rows: unknown[]; fromCache: boolean }> {
  const region = closest(userRegion);
  const cached = await region.redis.get(cacheKey);
  if (cached) return { rows: JSON.parse(cached), fromCache: true };

  const result = await region.pool.query(sql, params);
  await region.redis.setex(cacheKey, ttlSeconds, JSON.stringify(result.rows));
  return { rows: result.rows, fromCache: false };
}

// Per-region health: a standby's lag is the byte distance between received and replayed WAL
export async function getRegionHealth() {
  const out = [];
  for (const r of regions.values()) {
    let dbHealthy = false;
    let replayLagBytes = 0;
    let cacheHealthy = false;
    try {
      await r.pool.query("SELECT 1");
      dbHealthy = true;
      if (!r.isPrimary) {
        const { rows } = await r.pool.query(
          "SELECT pg_wal_lsn_diff(pg_last_wal_receive_lsn(), pg_last_wal_replay_lsn())::bigint AS lag"
        );
        replayLagBytes = Number(rows[0].lag ?? 0);
      }
    } catch {
      /* reported as unhealthy */
    }
    try {
      cacheHealthy = (await r.redis.ping()) === "PONG";
    } catch {
      /* reported as unhealthy */
    }
    out.push({ region: r.name, isPrimary: r.isPrimary, dbHealthy, replayLagBytes, cacheHealthy });
  }
  return out;
}