Build Automated Database Backup with Point-in-Time Recovery

// src/config/wal-archiving.ts — PostgreSQL WAL archive configuration generator
import { writeFile } from "node:fs/promises";
import { execSync } from "node:child_process";

interface BackupConfig {
  s3Bucket: string;
  s3Region: string;
  s3Prefix: string;          // e.g., "backups/production"
  pgDataDir: string;         // e.g., "/var/lib/postgresql/16/main"
  retentionDays: number;     // how long to keep backups
  baseBackupSchedule: string; // cron expression for full backups
  encryptionKey: string;     // GPG key ID for backup encryption
}

// Generate postgresql.conf settings for WAL archiving
export function generateWalConfig(config: BackupConfig): string {
  return `
# WAL Archiving Configuration — generated by backup system
# These settings enable continuous archiving for point-in-time recovery

wal_level = replica                    # required for WAL archiving (default in PG16+)
archive_mode = on                      # enable WAL archiving
archive_command = '/usr/local/bin/wal-push %p %f'  # script that uploads each WAL segment
archive_timeout = 300                  # force archive every 5 minutes even during low activity
max_wal_senders = 3                    # allow streaming replication connections
wal_keep_size = 1GB                    # keep 1GB of WAL locally as buffer
`;
}

// WAL push script — called by PostgreSQL for each completed WAL segment
export function generateWalPushScript(config: BackupConfig): string {
  return `#!/bin/bash
# wal-push — Upload WAL segment to S3 with encryption and compression
# Called by PostgreSQL archive_command with %p (path) and %f (filename)

set -euo pipefail

WAL_PATH="$1"
WAL_NAME="$2"
S3_DEST="s3://${config.s3Bucket}/${config.s3Prefix}/wal/$WAL_NAME.zst.gpg"

# Compress with zstd (10x faster than gzip, better ratio) then encrypt
zstd -3 -c "$WAL_PATH" | \\
  gpg --batch --yes --encrypt --recipient "${config.encryptionKey}" | \\
  aws s3 cp - "$S3_DEST" --region ${config.s3Region} --expected-size 17000000

# Verify upload succeeded
if aws s3 ls "$S3_DEST" --region ${config.s3Region} > /dev/null 2>&1; then
  echo "$(date -Iseconds) WAL archived: $WAL_NAME" >> /var/log/wal-archive.log
  exit 0
else
  echo "$(date -Iseconds) FAILED to archive: $WAL_NAME" >> /var/log/wal-archive.log
  exit 1  # PostgreSQL will retry on non-zero exit
fi
`;
}

// src/services/backup-manager.ts — Scheduled base backups with S3 upload
import { exec } from "node:child_process";
import { promisify } from "node:util";
import { stat, unlink } from "node:fs/promises";
import { pool } from "../db";

const execAsync = promisify(exec);

interface BackupResult {
  id: string;
  startTime: Date;
  endTime: Date;
  sizeBytes: number;
  walStartLsn: string;    // WAL position at backup start
  walStopLsn: string;     // WAL position at backup end
  s3Path: string;
  durationSeconds: number;
  success: boolean;
  errorMessage?: string;
}

export async function createBaseBackup(config: {
  s3Bucket: string;
  s3Prefix: string;
  s3Region: string;
  encryptionKey: string;
}): Promise<BackupResult> {
  const backupId = `base-${Date.now()}`;
  const startTime = new Date();
  const localPath = `/tmp/${backupId}.tar.zst.gpg`;
  const s3Path = `s3://${config.s3Bucket}/${config.s3Prefix}/base/${backupId}.tar.zst.gpg`;

  // Record backup start in database
  await pool.query(
    "INSERT INTO backup_history (id, type, status, started_at) VALUES ($1, 'base', 'running', NOW())",
    [backupId]
  );

  try {
    // pg_basebackup streams the entire database cluster
    // -Ft = tar format, -z = no built-in compression (we use zstd instead)
    // -X stream = include WAL segments generated during backup
    // -l = backup label for identification
    await execAsync(
      `pg_basebackup -h localhost -U replication_user -D - -Ft -X stream -l "${backupId}" | ` +
      `zstd -3 -T4 | ` +  // compress with 4 threads, level 3 (good balance)
      `gpg --batch --yes --encrypt --recipient "${config.encryptionKey}" > "${localPath}"`,
      { maxBuffer: 1024 * 1024 * 10 }  // 10MB stdout buffer for progress output
    );

    // Get file size before upload
    const fileStats = await stat(localPath);

    // Upload to S3 with multipart for large files
    await execAsync(
      `aws s3 cp "${localPath}" "${s3Path}" ` +
      `--region ${config.s3Region} ` +
      `--expected-size ${fileStats.size} ` +
      `--storage-class STANDARD_IA`,  // cheaper storage for backups
      { maxBuffer: 1024 * 1024 }
    );

    // Get WAL positions for this backup
    const { rows } = await pool.query(
      "SELECT pg_current_wal_lsn() as current_lsn"
    );

    const endTime = new Date();
    const result: BackupResult = {
      id: backupId,
      startTime,
      endTime,
      sizeBytes: fileStats.size,
      walStartLsn: "", // extracted from pg_basebackup label
      walStopLsn: rows[0].current_lsn,
      s3Path,
      durationSeconds: Math.round((endTime.getTime() - startTime.getTime()) / 1000),
      success: true,
    };

    // Update backup history
    await pool.query(
      `UPDATE backup_history SET status = 'completed', size_bytes = $2, 
       s3_path = $3, duration_seconds = $4, completed_at = NOW()
       WHERE id = $1`,
      [backupId, fileStats.size, s3Path, result.durationSeconds]
    );

    // Clean up local file
    await unlink(localPath).catch(() => {});

    return result;
  } catch (error) {
    const errorMsg = (error as Error).message;
    await pool.query(
      "UPDATE backup_history SET status = 'failed', error_message = $2 WHERE id = $1",
      [backupId, errorMsg]
    );
    await unlink(localPath).catch(() => {});

    return {
      id: backupId,
      startTime,
      endTime: new Date(),
      sizeBytes: 0,
      walStartLsn: "",
      walStopLsn: "",
      s3Path: "",
      durationSeconds: Math.round((Date.now() - startTime.getTime()) / 1000),
      success: false,
      errorMessage: errorMsg,
    };
  }
}

// Cleanup old backups based on retention policy
export async function pruneOldBackups(config: {
  s3Bucket: string;
  s3Prefix: string;
  s3Region: string;
  retentionDays: number;
}): Promise<{ deletedBackups: number; freedBytes: number }> {
  const cutoff = new Date(Date.now() - config.retentionDays * 86400 * 1000);

  // Find backups older than retention period
  const { rows: oldBackups } = await pool.query(
    "SELECT id, s3_path, size_bytes FROM backup_history WHERE completed_at < $1 AND status = 'completed'",
    [cutoff]
  );

  let freedBytes = 0;
  for (const backup of oldBackups) {
    try {
      await execAsync(
        `aws s3 rm "${backup.s3_path}" --region ${config.s3Region}`
      );
      freedBytes += backup.size_bytes || 0;
    } catch {
      // Log but continue — don't let one failure block cleanup
    }
  }

  // Also prune WAL segments older than the oldest remaining base backup
  const { rows: oldestBase } = await pool.query(
    "SELECT started_at FROM backup_history WHERE status = 'completed' ORDER BY started_at ASC LIMIT 1"
  );

  if (oldestBase.length > 0) {
    // Delete WAL segments from before the oldest base backup
    const walPrefix = `s3://${config.s3Bucket}/${config.s3Prefix}/wal/`;
    // WAL cleanup handled by listing and filtering by date
    await execAsync(
      `aws s3 ls "${walPrefix}" --region ${config.s3Region} | ` +
      `awk '{if ($1 < "${cutoff.toISOString().split("T")[0]}") print $4}' | ` +
      `xargs -I{} aws s3 rm "${walPrefix}{}" --region ${config.s3Region}`
    ).catch(() => {}); // best-effort WAL cleanup
  }

  await pool.query(
    "DELETE FROM backup_history WHERE completed_at < $1",
    [cutoff]
  );

  return { deletedBackups: oldBackups.length, freedBytes };
}

// src/services/recovery.ts — Point-in-time recovery orchestrator
import { exec } from "node:child_process";
import { promisify } from "node:util";
import { mkdir, writeFile, rm } from "node:fs/promises";
import { pool } from "../db";

const execAsync = promisify(exec);

interface RecoveryParams {
  targetTime: Date;           // recover to this exact moment
  s3Bucket: string;
  s3Prefix: string;
  s3Region: string;
  encryptionKey: string;
  recoveryDir: string;        // where to restore (e.g., /var/lib/postgresql/recovery)
  pgPort: number;             // port for recovered instance (different from production)
}

interface RecoveryResult {
  success: boolean;
  baseBackupUsed: string;
  targetTime: Date;
  walSegmentsReplayed: number;
  recoveryDurationSeconds: number;
  pgPort: number;
  errorMessage?: string;
}

export async function performRecovery(params: RecoveryParams): Promise<RecoveryResult> {
  const startTime = Date.now();
  const walDir = `${params.recoveryDir}/wal_restore`;

  try {
    // Step 1: Find the most recent base backup before the target time
    const { rows: backups } = await pool.query(
      `SELECT id, s3_path, started_at FROM backup_history 
       WHERE status = 'completed' AND started_at <= $1
       ORDER BY started_at DESC LIMIT 1`,
      [params.targetTime]
    );

    if (backups.length === 0) {
      throw new Error(`No base backup found before ${params.targetTime.toISOString()}`);
    }

    const baseBackup = backups[0];
    console.log(`Using base backup: ${baseBackup.id} from ${baseBackup.started_at}`);

    // Step 2: Prepare recovery directory
    await rm(params.recoveryDir, { recursive: true, force: true });
    await mkdir(params.recoveryDir, { recursive: true });
    await mkdir(walDir, { recursive: true });

    // Step 3: Download and decrypt base backup
    console.log("Downloading base backup...");
    await execAsync(
      `aws s3 cp "${baseBackup.s3_path}" - --region ${params.s3Region} | ` +
      `gpg --batch --yes --decrypt | ` +
      `zstd -d | ` +
      `tar xf - -C "${params.recoveryDir}"`,
      { maxBuffer: 1024 * 1024 * 100 }
    );

    // Step 4: Download WAL segments between base backup and target time
    console.log("Downloading WAL segments...");
    const walPrefix = `${params.s3Prefix}/wal/`;
    const { stdout: walList } = await execAsync(
      `aws s3 ls "s3://${params.s3Bucket}/${walPrefix}" --region ${params.s3Region}`
    );

    let walCount = 0;
    const walFiles = walList.trim().split("\n").filter(Boolean);
    for (const line of walFiles) {
      const filename = line.trim().split(/\s+/).pop();
      if (!filename) continue;

      await execAsync(
        `aws s3 cp "s3://${params.s3Bucket}/${walPrefix}${filename}" - --region ${params.s3Region} | ` +
        `gpg --batch --yes --decrypt | ` +
        `zstd -d > "${walDir}/${filename.replace('.zst.gpg', '')}"`,
        { maxBuffer: 1024 * 1024 * 50 }
      );
      walCount++;
    }

    // Step 5: Configure recovery parameters
    const recoveryConf = `
# Recovery configuration — restore to specific point in time
restore_command = 'cp ${walDir}/%f %p'
recovery_target_time = '${params.targetTime.toISOString()}'
recovery_target_action = 'promote'    # promote to read-write after recovery
`;

    await writeFile(`${params.recoveryDir}/recovery.signal`, "");
    await writeFile(`${params.recoveryDir}/postgresql.auto.conf`, `
port = ${params.pgPort}
${recoveryConf}
`);

    // Step 6: Start PostgreSQL with recovery configuration
    console.log(`Starting recovered instance on port ${params.pgPort}...`);
    await execAsync(
      `pg_ctl -D "${params.recoveryDir}" -l "${params.recoveryDir}/recovery.log" start`
    );

    // Wait for recovery to complete (PostgreSQL replays WAL segments)
    let recovered = false;
    for (let i = 0; i < 120; i++) { // wait up to 10 minutes
      await new Promise((r) => setTimeout(r, 5000));
      try {
        await execAsync(
          `psql -p ${params.pgPort} -c "SELECT 1" postgres`
        );
        recovered = true;
        break;
      } catch {
        // Still recovering — continue waiting
      }
    }

    if (!recovered) {
      throw new Error("Recovery timed out after 10 minutes");
    }

    const durationSeconds = Math.round((Date.now() - startTime) / 1000);

    return {
      success: true,
      baseBackupUsed: baseBackup.id,
      targetTime: params.targetTime,
      walSegmentsReplayed: walCount,
      recoveryDurationSeconds: durationSeconds,
      pgPort: params.pgPort,
    };
  } catch (error) {
    return {
      success: false,
      baseBackupUsed: "",
      targetTime: params.targetTime,
      walSegmentsReplayed: 0,
      recoveryDurationSeconds: Math.round((Date.now() - startTime) / 1000),
      pgPort: params.pgPort,
      errorMessage: (error as Error).message,
    };
  }
}

// src/services/backup-monitor.ts — Backup health monitoring with Slack alerts
import { pool } from "../db";

interface BackupHealth {
  status: "healthy" | "warning" | "critical";
  lastBaseBackup: { age: string; size: string } | null;
  walArchiveLag: number;       // seconds behind
  totalBackupSize: string;
  issues: string[];
}

export async function checkBackupHealth(): Promise<BackupHealth> {
  const issues: string[] = [];

  // Check last successful base backup
  const { rows: lastBackup } = await pool.query(
    `SELECT id, completed_at, size_bytes, duration_seconds
     FROM backup_history WHERE status = 'completed'
     ORDER BY completed_at DESC LIMIT 1`
  );

  let lastBaseBackup = null;
  if (lastBackup.length === 0) {
    issues.push("CRITICAL: No successful base backup found");
  } else {
    const ageHours = (Date.now() - new Date(lastBackup[0].completed_at).getTime()) / 3600000;
    lastBaseBackup = {
      age: `${Math.round(ageHours)}h ago`,
      size: formatBytes(lastBackup[0].size_bytes),
    };
    if (ageHours > 25) {
      issues.push(`WARNING: Last base backup is ${Math.round(ageHours)} hours old (target: <24h)`);
    }
  }

  // Check WAL archiving lag
  const { rows: walLag } = await pool.query(`
    SELECT EXTRACT(EPOCH FROM (NOW() - pg_last_xact_replay_timestamp())) as lag_seconds,
           pg_walfile_name(pg_current_wal_lsn()) as current_wal
  `);
  const walArchiveLag = walLag[0]?.lag_seconds || 0;

  if (walArchiveLag > 600) { // more than 10 minutes behind
    issues.push(`WARNING: WAL archive lag is ${Math.round(walArchiveLag / 60)} minutes`);
  }

  // Check total backup storage
  const { rows: totalSize } = await pool.query(
    "SELECT COALESCE(SUM(size_bytes), 0) as total FROM backup_history WHERE status = 'completed'"
  );

  const status = issues.some((i) => i.startsWith("CRITICAL"))
    ? "critical"
    : issues.length > 0
      ? "warning"
      : "healthy";

  return {
    status,
    lastBaseBackup,
    walArchiveLag,
    totalBackupSize: formatBytes(totalSize[0].total),
    issues,
  };
}

// Send Slack alert when backup health degrades
export async function alertIfUnhealthy(webhookUrl: string): Promise<void> {
  const health = await checkBackupHealth();
  if (health.status === "healthy") return;

  const color = health.status === "critical" ? "#e01e5a" : "#ecb22e";

  await fetch(webhookUrl, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      attachments: [{
        color,
        title: `🗄️ Backup ${health.status.toUpperCase()}`,
        fields: [
          { title: "Last Base Backup", value: health.lastBaseBackup?.age || "None", short: true },
          { title: "WAL Lag", value: `${Math.round(health.walArchiveLag)}s`, short: true },
          { title: "Total Size", value: health.totalBackupSize, short: true },
          { title: "Issues", value: health.issues.join("\n") },
        ],
        ts: Math.floor(Date.now() / 1000),
      }],
    }),
  });
}

function formatBytes(bytes: number): string {
  if (bytes < 1024) return `${bytes} B`;
  if (bytes < 1048576) return `${(bytes / 1024).toFixed(1)} KB`;
  if (bytes < 1073741824) return `${(bytes / 1048576).toFixed(1)} MB`;
  return `${(bytes / 1073741824).toFixed(1)} GB`;
}