postgres-copy.js

const { Client } = require("pg");
const AWS = require("aws-sdk");
const { from } = require("pg-copy-streams");
const stream = require("stream");

// Configure AWS S3
const s3 = new AWS.S3({
  accessKeyId: "", // Your AWS access key
  secretAccessKey: "", // Your AWS secret key
  region: "ap-southeast-2", // The region where your bucket is located
});

// PostgreSQL connection configuration
const client = new Client({
  user: "postgres",
  host: "host.docker.internal",
  database: "postgres",
  password: "postgres",
  port: 5432,
});

// Function to get CSV file from S3
async function getS3CSVFile(bucketName, key) {
  const params = {
    Bucket: "guangtoutou-s3-event-notification",
    Key: "users.csv",
  };

  // Return the S3 object as a stream
  return s3.getObject(params).createReadStream();
}

// Function to copy CSV data from S3 into PostgreSQL
async function copyCSVFromS3ToPostgres(bucketName, key) {
  try {
    await client.connect();

    // Get the CSV file from S3 as a stream
    const s3Stream = await getS3CSVFile(bucketName, key);

    // Set up the COPY command for PostgreSQL
    const copyStream = client.query(
      from("COPY users(name, age) FROM STDIN WITH CSV HEADER")
    );

    // Pipe the S3 stream to PostgreSQL
    s3Stream.pipe(copyStream);

    // Handle stream completion
    copyStream.on("end", async () => {
      console.log("CSV file successfully imported into the database");
      await client.end();
    });

    // Handle errors in the copy stream
    copyStream.on("error", async (err) => {
      console.error("Error while copying CSV to PostgreSQL:", err);
      await client.end();
    });
  } catch (err) {
    console.error("Error:", err);
    await client.end();
  }
}

// Invoke the function to load data from S3 into PostgreSQL
copyCSVFromS3ToPostgres("your_s3_bucket_name", "path/to/your/file.csv");