PersonalBlog/packages/database-postgres/src/lock.ts

/**
 * Migration advisory lock — [E00-S03-T04].
 *
 * The advisory lock is the PostgreSQL-side guarantee that concurrent migration
 * runners cannot run at the same time: the migration runner (built on this
 * boundary — the failure diagnostic E00-S03-T05; the ready gate E00-S03-T06)
 * takes the lock before applying migrations, so a second runner
 * either waits (`acquire()`) or fails fast (`tryAcquire()`) while the first
 * holds it.
 *
 * The lock lives in `database-postgres` — the single workspace package
 * allowed to import the PostgreSQL driver (E00-S03-T02) — and talks to the
 * database exclusively through the package-owned `pg` Pool, so no other
 * package needs the driver to lock migration state.
 *
 * Lock model:
 *   - session-scoped advisory lock (`pg_advisory_lock`/`pg_try_advisory_lock`
 *     on the `bigint` key `hashtextextended($1, 0)`), so it lives exactly as
 *     long as the holding database session and no longer;
 *   - held on a dedicated connection checked out of the caller's pool
 *     (`pool.connect()`), so the lock never taints a pooled connection that
 *     other queries share;
 *   - released explicitly by `release()` (`pg_advisory_unlock` first, then
 *     the client goes back to the pool; if the unlock statement itself fails,
 *     the connection is destroyed — `client.release(error)` — so a pooled
 *     connection is never reused while its session still holds the lock),
 *     and released implicitly by the server when the holding session ends —
 *     the issue's rollback note: "the lock releases when the runner exits".
 *
 * The lock key is always passed as a bound parameter (`$1`) — the SQL never
 * interpolates it, so the only interpolated value is the `$1` placeholder
 * itself. `hashtextextended($1, 0)` maps the key string to a stable `bigint`
 * advisory-lock key: deterministic for a given database, identical across all
 * sessions, so every runner contends on the same lock.
 */

import type { Pool, PoolClient } from 'pg';

/** Namespace string for the migration advisory lock. */
export const MIGRATION_LOCK_KEY = 'personal-blog-migrations';

/**
 * The migration advisory lock: serializes migration runners so two runners
 * can never apply migrations at the same time. Instances are cheap and share
 * the caller's pool; the lock performs no ledger writes (migration ledger is
 * E00-S03-T03) and no diagnostics (E00-S03-T05).
 */
export class MigrationLock {
  private readonly pool: Pool;
  private client: PoolClient | null = null;
  private held = false;
  /** In-flight acquire, memoized so concurrent acquire() calls share one connection (re-entrant-safe). */
  private acquireInFlight: Promise<void> | null = null;

  /** @param pool The package-owned PostgreSQL pool (`pg.Pool`). */
  constructor(pool: Pool) {
    this.pool = pool;
  }

  /** True while this instance holds the advisory lock. */
  get isHeld(): boolean {
    return this.held;
  }

  /**
   * Acquires the migration advisory lock, blocking until it is free — a
   * second runner waits here while the first holds the lock. Idempotent:
   * acquiring an already-held instance is a no-op. Re-entrant-safe:
   * concurrent `acquire()` calls on the same instance share the single
   * in-flight acquire (memoized in `acquireInFlight`), so exactly one
   * connection is checked out and no locked connection leaks. The lock is
   * held on a dedicated connection until `release()` (or until the session
   * ends — e.g. the runner exits and the pool closes its connections).
   */
  async acquire(): Promise<void> {
    if (this.held) return;
    // A second concurrent acquire() on this instance returns the in-flight
    // acquire instead of checking out another connection: exactly one
    // connection is checked out and no locked connection leaks.
    if (this.acquireInFlight !== null) return this.acquireInFlight;
    const inFlight = this.doAcquire();
    this.acquireInFlight = inFlight;
    try {
      await inFlight;
    } finally {
      this.acquireInFlight = null;
    }
  }

  /** The memoized acquire body: checks out one dedicated connection and takes the session-scoped lock on it. */
  private async doAcquire(): Promise<void> {
    const client = await this.pool.connect();
    try {
      await client.query(
        'SELECT pg_advisory_lock(hashtextextended($1, 0))',
        [MIGRATION_LOCK_KEY],
      );
    } catch (error) {
      client.release();
      throw error;
    }
    this.client = client;
    this.held = true;
  }

  /**
   * Non-blocking acquire: resolves `true` when this runner got the lock and
   * `false` when another runner holds it — a second runner fails fast while
   * the first holds the lock. Also resolves `true` when this instance
   * already holds the lock.
   */
  async tryAcquire(): Promise<boolean> {
    if (this.held) return true;
    const client = await this.pool.connect();
    try {
      const result = await client.query<{ acquired: boolean }>(
        'SELECT pg_try_advisory_lock(hashtextextended($1, 0)) AS acquired',
        [MIGRATION_LOCK_KEY],
      );
      const acquired = result.rows[0]?.acquired === true;
      if (acquired) {
        this.client = client;
        this.held = true;
        return true;
      }
      client.release();
      return false;
    } catch (error) {
      client.release();
      throw error;
    }
  }

  /**
   * Releases the advisory lock: unlocks it on the holding session
   * (`pg_advisory_unlock`) and returns the connection to the pool, so the
   * lock is gone before any other query could reuse that pooled connection.
   * If the unlock statement fails (e.g. statement timeout or cancellation)
   * the connection is destroyed instead — `client.release(error)` tells the
   * pool to drop the client, ending the session (and its lock), so a pooled
   * connection is never reused while its session still holds the lock; the
   * failure is re-thrown. Idempotent: releasing an instance that does not
   * hold the lock is a no-op.
   */
  async release(): Promise<void> {
    if (!this.held || this.client === null) return;
    const client = this.client;
    this.client = null;
    this.held = false;
    try {
      await client.query(
        'SELECT pg_advisory_unlock(hashtextextended($1, 0))',
        [MIGRATION_LOCK_KEY],
      );
    } catch (error) {
      // The unlock statement failed while this session may still hold the
      // advisory lock. The connection must not go back into the pool in that
      // state — the next borrower would block every other runner. Destroy it
      // (release(error) removes the client from the pool), so the session —
      // and its lock — ends.
      client.release(error as Error);
      throw error;
    }
    client.release();
  }
}