const PSQL = require('cartodb-psql');
const copyTo = require('pg-copy-streams').to;
const copyFrom = require('pg-copy-streams').from;
const { Client } = require('pg');

const ACTION_TO = 'to';
const ACTION_FROM = 'from';

module.exports = class StreamCopy {
    constructor(sql, userDbParams) {
        this.pg = new PSQL(userDbParams);
        this.sql = sql;
        this.connectionClosedByClient = false;

        this.stream = null;
    }

    static get ACTION_TO() {
        return ACTION_TO;
    }

    static get ACTION_FROM() {
        return ACTION_FROM;
    }

    to(cb) {
        this.pg.connect((err, client, done) => {
            if (err) {
                return cb(err);
            }

            this.stream = copyTo(this.sql);
            const pgstream = client.query(this.stream);

            pgstream
                .on('end', () => done())
                .on('error', err => done(err))
                .on('cancelQuery', err => {
                    // See https://www.postgresql.org/docs/9.5/static/protocol-flow.html#PROTOCOL-COPY
                    const cancelingClient = new Client(client.connectionParameters);
                    cancelingClient.cancel(client, pgstream);

                    // see https://node-postgres.com/api/pool#releasecallback
                    done(err);
                });

            cb(null, pgstream);
        });
    }

    from(cb) {
        this.pg.connect((err, client, done) => {
            if (err) {
                return cb(err);
            }

            this.stream = copyFrom(this.sql);
            const pgstream = client.query(this.stream);

            pgstream
                .on('end', () => done())
                .on('error', err => done(err))
                .on('cancelQuery', () => {
                    client.connection.sendCopyFail('CARTO SQL API: Connection closed by client');
                });

            cb(null, pgstream);
        });
    }

    getRowCount() {
        return this.stream.rowCount;
    }
};

@@ -401,17 +401,16 @@
  false;%0A
-%0A
         
@@ -572,11 +572,26 @@
     
-to(
+getStream(action, 
 cb) 
@@ -727,28 +727,105 @@
     
-this.
+let 
 stream
- = copyTo
+Maker = action === ACTION_TO ? copyTo : copyFrom;%0A            this.stream = streamMaker
 (thi
@@ -1038,24 +1038,75 @@
 ', err =%3E %7B%0A
+                    if(action === ACTION_TO) %7B%0A    
             
@@ -1212,24 +1212,28 @@
             
+    
 const cancel
@@ -1305,16 +1305,20 @@
         
+  
+  
 cancelin
@@ -1344,32 +1344,36 @@
 nt, pgstream);%0A%0A
+    
                 
@@ -1454,16 +1454,20 @@
         
+  
+  
 done(err
@@ -1489,453 +1489,53 @@
     
-%7D);%0A%0A            cb(null, pgstream);%0A        %7D);%0A    %7D%0A%0A    from(cb) %7B%0A        this.pg.connect((err, client, done) =%3E %7B%0A            if (err) %7B%0A                return cb(err);%0A            %7D%0A%0A            this.stream = copyFrom(this.sql);%0A            const pgstream = client.query(this.stream);%0A%0A            pgstream%0A                .on('end', () =%3E done())%0A                .on('error', err =%3E done(err))%0A                .on('cancelQuery', () =%3E %7B%0A
+    %7D else if (action === ACTION_FROM) %7B%0A    
     
@@ -1628,16 +1628,38 @@
 ient');%0A
+                    %7D%0A
         

unify to and from method in only one getPGStream