backup pg database to aws

render-aws-backup.ts

import { spawn, execSync } from 'child_process';
import through2 from 'through2';
import fs from 'fs';
import * as AWS from 'aws-sdk';

function spawnPgDump(pgDumpPath, args, env) {
  if (!fs.existsSync(pgDumpPath)) {
    throw new Error('pg_dump not found at ' + pgDumpPath);
  }

  return spawn(pgDumpPath, args, {
    env,
  });
}

export class Backups {
  public async run(): Promise<string> {
    const path = execSync('which pg_dump').toString().trim();
    const now = new Date();
    return this.streamToS3(
      await this.getPgDumpStream(path),
      `backups/${now.getFullYear()}${now
        .getMonth()
        .toString()
        .padStart(2, '0')}/${now.getTime()}.dump`,
    );
  }

  private getPgDumpStream(
    pgDumpPath: string,
    pgDumpSpawnFn = spawnPgDump,
  ): Promise<Buffer> {
    return new Promise((resolve, reject) => {
      let headerChecked = false;
      let stderr = '';

      // spawn pg_dump process
      const parts = pgDumpPath.split('/');
      parts.pop();
      const env = { LD_LIBRARY_PATH: parts.join('/') };
      const pgdumpStream = pgDumpSpawnFn(
        pgDumpPath,
        ['-Fc', '-Z1', process.env.PG_CONNECTION_STRING],
        env,
      );

      // hook into the process
      pgdumpStream.stderr.on('data', (data) => {
        stderr += data.toString('utf8');
      });

      pgdumpStream.on('close', (code) => {
        // reject our promise if pg_dump had a non-zero exit
        if (code !== 0) {
          return reject(new Error('pg_dump process failed: ' + stderr));
        }
        // check that pgdump actually gave us some data
        if (!headerChecked) {
          return reject(new Error('pg_dump gave us an unexpected response'));
        }
        return null;
      });

      // use through2 to proxy the pg_dump stdout stream
      // so we can check it's valid
      const buffer = through2(function (chunk, _enc, callback) {
        this.push(chunk);
        // if stdout begins with 'PGDMP' then the backup has begun
        // otherwise, we abort
        if (!headerChecked) {
          headerChecked = true;
          if (chunk.toString('utf8').startsWith('PGDMP')) {
            resolve(buffer);
          } else {
            reject(new Error('pg_dump gave us an unexpected response'));
          }
        }
        callback();
      });

      // pipe pg_dump to buffer
      pgdumpStream.stdout.pipe(buffer);
    });
  }

  private async streamToS3(stream: Buffer, key: string): Promise<string> {
    const s3 = new AWS.S3({
      apiVersion: '2006-03-01',
      region: process.env.S3_REGION || 'us-west-2',
      endpoint: process.env.S3_ENDPOINT || undefined,
      accessKeyId: process.env.AWS_ACCESS_KEY_ID,
      secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY,
      s3ForcePathStyle: true,
    });
    const result = await s3
      .upload({
        Key: key,
        Bucket: process.env.S3_BUCKET,
        Body: stream,
      })
      .promise();
    console.log(`uploaded backup to ${result.Location}`);
    return result.Location;
  }
}