pg-copy-streams

const { to, from, both } = require('pg-copy-streams');

// Alternative destructuring with aliases
const { to: copyTo, from: copyFrom, both: copyBoth } = require('pg-copy-streams');

// Import entire module
const pgCopyStreams = require('pg-copy-streams');
const copyTo = pgCopyStreams.to;

import { to, from, both } from 'pg-copy-streams';

// Alternative with aliases
import { to as copyTo, from as copyFrom, both as copyBoth } from 'pg-copy-streams';

// Default import (not available - module uses named exports only)
// import pgCopyStreams from 'pg-copy-streams'; // ❌ This won't work

const { Pool } = require('pg');
const { to: copyTo, from: copyFrom } = require('pg-copy-streams');

const pool = new Pool();

// Export data from PostgreSQL (COPY TO)
pool.connect((err, client, done) => {
  const stream = client.query(copyTo('COPY my_table TO STDOUT'));
  stream.pipe(process.stdout);
  stream.on('end', done);
  stream.on('error', done);
});

// Import data into PostgreSQL (COPY FROM)
const fs = require('fs');
pool.connect((err, client, done) => {
  const stream = client.query(copyFrom('COPY my_table FROM STDIN'));
  const fileStream = fs.createReadStream('data.csv');
  fileStream.pipe(stream);
  stream.on('finish', done);
  stream.on('error', done);
});

/**
 * Creates a readable stream for COPY TO operations
 * @param {string} text - SQL COPY TO statement
 * @param {object} [options] - Optional stream configuration 
 * @returns {CopyToQueryStream} Readable stream instance
 */
function to(text, options);

/**
 * Creates a writable stream for COPY FROM operations
 * @param {string} text - SQL COPY FROM statement
 * @param {object} [options] - Optional stream configuration
 * @returns {CopyFromQueryStream} Writable stream instance
 */
function from(text, options);

/**
 * Creates a duplex stream for COPY BOTH operations
 * @param {string} text - SQL COPY statement
 * @param {object} [options] - Optional stream configuration
 * @returns {CopyBothQueryStream} Duplex stream instance
 */
function both(text, options);

/**
 * @typedef {object} StreamOptions
 * @property {number} [highWaterMark] - Stream high water mark
 * @property {boolean} [objectMode] - Object mode flag
 * @property {string} [encoding] - Stream encoding
 */

/**
 * @typedef {StreamOptions} CopyBothOptions
 * @property {boolean} [alignOnCopyDataFrame] - Whether to align data on COPY data frame boundaries
 */

/**
 * CopyToQueryStream - Readable stream for COPY TO operations
 * @class
 * @extends {Readable}
 */
class CopyToQueryStream {
  /**
   * @param {string} text - The SQL COPY TO statement
   * @param {StreamOptions} [options] - Stream options
   */
  constructor(text, options) {}
  
  /** @type {string} The SQL COPY statement */
  text;
  
  /** @type {number} Number of rows processed */
  rowCount;
  
  /** @type {object} PostgreSQL connection reference */
  connection;
  
  /**
   * Submits the query to a PostgreSQL connection
   * @param {object} connection - pg connection object
   */
  submit(connection) {}
  
  /**
   * Handles errors during stream processing
   * @param {Error} err - Error object
   */
  handleError(err) {}
  
  /**
   * Processes CommandComplete message and extracts row count
   * @param {object} msg - Message object with text property
   */
  handleCommandComplete(msg) {}
  
  /**
   * Called when connection is ready for next query
   */
  handleReadyForQuery() {}
}

/**
 * CopyFromQueryStream - Writable stream for COPY FROM operations
 * @class
 * @extends {Writable}
 */
class CopyFromQueryStream {
  /**
   * @param {string} text - The SQL COPY FROM statement
   * @param {StreamOptions} [options] - Stream options
   */
  constructor(text, options) {}
  
  /** @type {string} The SQL COPY statement */
  text;
  
  /** @type {number} Number of rows processed */
  rowCount;
  
  /** @type {object} PostgreSQL connection reference */
  connection;
  
  /**
   * Submits the query to a PostgreSQL connection
   * @param {object} connection - pg connection object
   */
  submit(connection) {}
  
  /**
   * Callback placeholder for pg timeout mechanism
   */
  callback() {}
  
  /**
   * Handles errors during stream processing
   * @param {Error} e - Error object
   */
  handleError(e) {}
  
  /**
   * Handles CopyInResponse from PostgreSQL backend
   * @param {object} connection - pg connection object
   */
  handleCopyInResponse(connection) {}
  
  /**
   * Processes CommandComplete message and extracts row count
   * @param {object} msg - Message object with text property
   */
  handleCommandComplete(msg) {}
  
  /**
   * Called when connection is ready for next query
   */
  handleReadyForQuery() {}
}

/**
 * CopyBothQueryStream - Duplex stream for COPY BOTH operations
 * @class
 * @extends {Duplex}
 */
class CopyBothQueryStream {
  /**
   * @param {string} text - The SQL COPY statement
   * @param {CopyBothOptions} [options] - Stream options
   */
  constructor(text, options) {}
  
  /** @type {string} The SQL COPY statement */
  text;
  
  /** @type {object} PostgreSQL connection reference */
  connection;
  
  /** @type {boolean} Whether to align data on COPY data frame boundaries */
  alignOnCopyDataFrame;
  
  /**
   * Submits the query to a PostgreSQL connection
   * @param {object} connection - pg connection object
   */
  submit(connection) {}
  
  /**
   * Handles errors during stream processing
   * @param {Error} err - Error object
   */
  handleError(err) {}
  
  /**
   * Handles out-of-band CopyData messages
   * @param {Buffer} chunk - Data chunk
   */
  handleCopyData(chunk) {}
  
  /**
   * Called when command completes
   */
  handleCommandComplete() {}
  
  /**
   * Called when connection is ready for next query
   */
  handleReadyForQuery() {}
}