/* Copyright (c) 2016, 2022, Oracle and/or its affiliates. */

/******************************************************************************
 *
 * You may not use the identified files except in compliance with the Apache
 * License, Version 2.0 (the "License.")
 *
 * You may obtain a copy of the License at
 * http://www.apache.org/licenses/LICENSE-2.0.
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * NAME
 *   13. stream1.js
 *
 * DESCRIPTION
 *   Testing driver query results via stream feature.
 *
 *****************************************************************************/
'use strict';

const oracledb = require('oracledb');
const should   = require('should');
const async    = require('async');
const dbConfig = require('./dbconfig.js');

// Need to skip some tests if Node.js version is < 8
const nodeMajorVersion = Number(process.versions.node.split('.')[0]);

describe('13. stream1.js', function() {

  let connection = null;
  let rowsAmount = 217;

  before(function(done) {
    async.series([
      function getConn(cb) {
        oracledb.getConnection(
          dbConfig,
          function(err, conn) {
            should.not.exist(err);
            connection = conn;
            cb();
          }
        );
      },
      function createTab(cb) {
        let proc = "BEGIN \n" +
                   "    DECLARE \n" +
                   "        e_table_missing EXCEPTION; \n" +
                   "        PRAGMA EXCEPTION_INIT(e_table_missing, -00942);\n " +
                   "    BEGIN \n" +
                   "        EXECUTE IMMEDIATE ('DROP TABLE nodb_stream1 PURGE'); \n" +
                   "    EXCEPTION \n" +
                   "        WHEN e_table_missing \n" +
                   "        THEN NULL; \n" +
                   "    END; \n" +
                   "    EXECUTE IMMEDIATE (' \n" +
                   "        CREATE TABLE nodb_stream1 ( \n" +
                   "            employee_id NUMBER, \n" +
                   "            employee_name VARCHAR2(20), \n" +
                   "            employee_history CLOB \n" +
                   "        ) \n" +
                   "    '); \n" +
                   "END; ";

        connection.execute(
          proc,
          function(err) {
            should.not.exist(err);
            cb();
          }
        );
      },
      function insertRows(cb) {
        const proc = `DECLARE
                       x NUMBER := 0;
                       n VARCHAR2(20);
                      BEGIN
                          FOR i IN 1..217 LOOP
                              x := x + 1;
                              n := 'staff ' || x;
                              INSERT INTO nodb_stream1 VALUES (x, n, x||'12345678901234567890');
                          END LOOP;
                      END;`;

        connection.execute(
          proc,
          function(err) {
            should.not.exist(err);
            cb();
          }
        );
      }
    ], done);
  }); // before

  after(function(done) {
    async.series([
      function(callback) {
        connection.execute(
          "DROP TABLE nodb_stream1 PURGE",
          function(err) {
            should.not.exist(err);
            callback();
          }
        );
      },
      function(callback) {
        connection.release(function(err) {
          should.not.exist(err);
          callback();
        });
      },
    ], done);
  }); // after

  describe('13.1 Testing QueryStream', function() {

    async function readLob(rowIndex, clobloc) {
      should.equal(clobloc.constructor.name, 'Lob');

      const doLobStreamHelper = new Promise((resolve, reject) => {

        clobloc.setEncoding('utf8');

        clobloc.on('error', (err) => {
          reject(err);
        });

        clobloc.on('data', (data) => {
          should.equal(data, (rowIndex + 1) + '12345678901234567890');
        });

        clobloc.on('end', () => {
          clobloc.destroy();
        });

        clobloc.on('close', () => {
          resolve();
        });
      });

      await doLobStreamHelper;
    }

    it('13.1.1 stream results for oracle connection', function(done) {
      let stream = connection.queryStream('SELECT employee_name FROM nodb_stream1 ORDER BY employee_name');

      stream.on('error', function(error) {
        should.fail(error, null, 'Error event should not be triggered');
      });

      let counter = 0;
      stream.on('data', function(data) {
        should.exist(data);
        counter++;
      });

      stream.on('end', function() {
        should.equal(counter, rowsAmount);
        stream.destroy();
      });

      stream.on('close', function() {
        done();
      });
    });

    it('13.1.2 stream results for oracle connection (outFormat: oracledb.OUT_FORMAT_OBJECT)', function(done) {
      let stream = connection.queryStream(
        'SELECT employee_name FROM nodb_stream1 ORDER BY employee_name',
        {},
        { outFormat: oracledb.OUT_FORMAT_OBJECT }
      );

      stream.on('error', function(error) {
        should.fail(error, null, 'Error event should not be triggered');
      });

      let counter = 0;
      stream.on('data', function(data) {
        should.exist(data);
        counter++;
      });

      stream.on('end', function() {
        should.equal(counter, rowsAmount);
        stream.destroy();
      });

      stream.on('close', function() {
        done();
      });
    });

    it('13.1.3 errors in query', function(done) {
      let stream = connection.queryStream('SELECT no_such_column FROM nodb_stream1');

      stream.on('error', function(err) {
        should.exist(err);
      });

      stream.on('data', function(data) {
        should.fail(data, null, 'Data event should not be triggered');
      });

      stream.on('close', function() {
        done();
      });
    });

    it('13.1.4 no result', function(done) {

      let stream = connection.queryStream('SELECT * FROM nodb_stream1 WHERE employee_name = :name', {
        name: 'TEST_NO_RESULT'
      });

      stream.on('error', function(error) {
        should.fail(error, null, 'Error event should not be triggered: ' + error);
      });

      let counter = 0;
      stream.on('data', function(data) {
        should.fail(data, null, 'Data event should not be triggered');
      });

      stream.on('end', function() {
        should.equal(counter, 0);
        stream.destroy();
      });

      stream.on('close', function() {
        done();
      });
    });

    it('13.1.5 single row', function(done) {

      let stream = connection.queryStream('SELECT employee_name FROM nodb_stream1 WHERE employee_name = :name', {
        name: 'staff 10'
      });

      stream.on('error', function(error) {
        should.fail(error, null, 'Error event should not be triggered: ' + error);
      });

      let counter = 0;
      stream.on('data', function(data) {
        should.exist(data);
        should.deepEqual(data, ['staff 10']);

        counter++;
      });

      stream.on('end', function() {
        should.equal(counter, 1);
        stream.destroy();
      });

      stream.on('close', function() {
        done();
      });
    });

    it('13.1.6 multiple row', function(done) {
      let stream = connection.queryStream('SELECT employee_name FROM nodb_stream1 WHERE employee_id <= :maxId ORDER BY employee_id', {
        maxId: 10
      }, {
        outFormat: oracledb.OUT_FORMAT_OBJECT
      });

      stream.on('error', function(error) {
        should.fail(error, null, 'Error event should not be triggered: ' + error);
      });

      let counter = 0;
      stream.on('data', function(data) {
        should.exist(data);
        should.deepEqual(data, {
          EMPLOYEE_NAME: 'staff ' + (counter + 1)
        });
        counter++;
      });

      stream.on('end', function() {
        should.equal(counter, 10);
        stream.destroy();
      });

      stream.on('close', function() {
        done();
      });
    });

    it('13.1.7 invalid SQL', function(done) {
      let stream = connection.queryStream(
        'UPDATE nodb_stream1 SET employee_name = :name WHERE rownum < 1',
        {
          name: 'test_update'
        },
        {
          outFormat: oracledb.OUT_FORMAT_OBJECT
        }
      );

      stream.on('error', function(error) {
        should.exist(error);
      });

      stream.on('data', function(data) {
        should.fail(data, null, 'Data event should not be triggered');
      });

      stream.on('close', function() {
        done();
      });
    });


    it('13.1.8 Read CLOBs', async function() {
      connection.should.be.ok();

      let stream = connection.queryStream(
        'SELECT employee_name, employee_history FROM nodb_stream1 where employee_id <= :maxId ORDER BY employee_id',
        { maxId: 10 },
        { outFormat: oracledb.OUT_FORMAT_OBJECT }
      );

      const consumeQryStream = new Promise((resolve, reject) => {
        let counter = 0;

        stream.on('data', async function(data) {
          let rowIndex = counter;

          const reading = new Promise((resolve1) => {
            should.exist(data);
            should.equal(data.EMPLOYEE_NAME, 'staff ' + (rowIndex + 1));
            should.exist(data.EMPLOYEE_HISTORY);
            should.equal(data.EMPLOYEE_HISTORY.constructor.name, 'Lob');
            counter++;
            resolve1(data.EMPLOYEE_HISTORY);
          });
          const lob = await reading;
          await readLob(rowIndex, lob);
        });

        stream.on('error', function(error) {
          should.fail(error, null, 'Error event should not be triggered: ' + error);
          reject();
        });
        stream.on('end', function() {
          should.equal(counter, 10);
          stream.destroy();
        });

        stream.on('close', function() {
          resolve();
        });
      });

      await consumeQryStream;
    });

    it('13.1.9 Read CLOBs after stream close', async function() {

      let stream = connection.queryStream('SELECT employee_name, employee_history FROM nodb_stream1 where employee_id <= :maxId ORDER BY employee_id', {
        maxId: 10
      }, {
        outFormat: oracledb.OUT_FORMAT_OBJECT
      });

      const consumeQryStream = new Promise((resolve, reject) => {

        let counter = 0;
        let cloblocs = [];  // the CLOB locators

        stream.on('data', function(data) {
          should.exist(data);
          should.equal(data.EMPLOYEE_NAME, 'staff ' + (++counter));
          should.exist(data.EMPLOYEE_HISTORY);
          should.equal(data.EMPLOYEE_HISTORY.constructor.name, 'Lob');
          cloblocs.push(data.EMPLOYEE_HISTORY);
        });

        stream.on('error', function(error) {
          should.fail(error, null, 'Error event should not be triggered: ' + error);
          reject(error);
        });

        stream.on('end', function() {
          stream.destroy();
        });

        stream.on('close', function() {
          resolve(cloblocs);
        });
      });

      const cloblocs = await consumeQryStream;
      should.equal(cloblocs.length, 10);

      for (let i = 0; i < 10; i++) {
        await readLob(i, cloblocs[i]);
      }

    });

    it('13.1.10 meta data', function(done) {

      let stream = connection.queryStream('SELECT employee_name FROM nodb_stream1 WHERE employee_name = :name', {
        name: 'staff 10'
      });

      let metaDataRead = false;
      stream.on('metadata', function(metaData) {
        should.deepEqual(metaData, [
          {
            name: 'EMPLOYEE_NAME'
          }
        ]);
        metaDataRead = true;
      });

      stream.on('error', function(error) {
        should.fail(error, null, 'Error event should not be triggered: ' + error);
      });

      stream.on('data', function() {
        should.equal(metaDataRead, true);
      });

      stream.on('end', function() {
        should.equal(metaDataRead, true);
        stream.destroy();
      });

      stream.on('close', function() {
        done();
      });
    });

    it('13.1.11 should emit events in the correct order', function(done) {
      let stream = connection.queryStream('SELECT employee_name FROM nodb_stream1 WHERE rownum = 1');
      let events = [];

      stream.on('open', function() {
        events.push('open');
      });

      stream.on('metadata', function() {
        events.push('metadata');
      });

      stream.on('data', function() {
        events.push('data');
      });

      stream.on('end', function() {
        events.push('end');
        stream.destroy();
      });

      stream.on('close', function() {
        events[0].should.equal('open');
        events[1].should.equal('metadata');
        events[2].should.equal('data');
        events[3].should.equal('end');
        // This is close so no need to check

        done();
      });

      stream.on('error', function() {
        done(new Error('Test should not have thrown an error'));
      });
    });

    it('13.1.12 query with logical error should throw error', function(done) {
      const sql = 'select 1 from dual union all select 1 from dual union all select 1/0 from dual';
      let stream = connection.queryStream(sql);

      stream.on('error', function(error) {
        should.exist(error);
        should.strictEqual(error.message, 'ORA-01476: divisor is equal to zero');
      });

      stream.on('data', function(data) {
        should.not.exist(data);
      });

      stream.on('end', function() {
        stream.destroy();
      });

      stream.on('close', done);
    });
  });

  describe('13.2 Testing QueryStream.destroy', function() {
    let it = (nodeMajorVersion >= 8) ? global.it : global.it.skip;

    it('13.2.1 should be able to stop the stream early with destroy', function(done) {
      let stream = connection.queryStream('SELECT employee_name FROM nodb_stream1 ORDER BY employee_name');
      let err;

      stream.on('data', function() {
        stream.pause();
        stream.destroy();
      });

      stream.on('close', function() {
        done(err);
      });

      stream.on('end', function() {
        err = new Error('Reached the end of the stream');
      });

      stream.on('error', function(err1) {
        err = err1;
      });
    });

    it('13.2.2 should be able to stop the stream before any data', function(done) {
      let stream = connection.queryStream('SELECT employee_name FROM nodb_stream1 ORDER BY employee_name');
      let err;

      stream.on('close', function() {
        done(err);
      });

      stream.on('open', function() {
        stream.destroy();
      });

      stream.on('data', function() {
        err = new Error('Received data');
      });

      stream.on('end', function() {
        err = new Error('Reached the end of the stream');
      });

      stream.on('error', function(err1) {
        err = err1;
      });
    });

    it('13.2.3 should invoke an optional callback passed to destroy', function(done) {
      let stream = connection.queryStream('SELECT employee_name FROM nodb_stream1 ORDER BY employee_name');
      let err;

      stream.on('open', function(err1) {
        if (err1) err = err1;
        stream.destroy(null, done(err)); // Not documented, but the second param can be a callback
      });

      stream.on('data', function() {
        err = new Error('Received data');
      });

      stream.on('end', function() {
        err = new Error('Reached the end of the stream');
      });

      stream.on('error', function(err1) {
        err = err1;
      });

    });

    it('13.2.4 should work if querystream is destroyed before resultset is opened', function(done) {
      let stream = connection.queryStream('SELECT employee_name FROM nodb_stream1');
      let err;

      stream.destroy();

      stream.on('data', function() {
        err = new Error('Received a row');
      });

      stream.on('end', function() {
        err = new Error('Reached the end of the stream');
      });

      stream.on('error', function(err1) {
        err = err1;
      });

      stream.on('close', function() {
        done(err);
      });
    });

    it('13.2.5 should work if querystream is destroyed after end event', function(done) {
      let stream = connection.queryStream('SELECT employee_name FROM nodb_stream1');
      let err;

      stream.on('data', function() {});

      stream.on('end', function() {
        stream.destroy();
      });

      stream.on('error', function(err1) {
        err = err1;
      });

      stream.on('close', function() {
        done(err);
      });
    });

    it('13.2.6 should emit the error passed in', function(done) {
      let stream = connection.queryStream('SELECT employee_name FROM nodb_stream1 ORDER BY employee_name');
      let customError = new Error('Ouch!');
      let err;

      stream.on('open', function() {
        stream.destroy(customError);
      });

      stream.on('data', function() {
        err = new Error('Received data');
      });

      stream.on('end', function() {
        err = new Error('Reached the end of the stream');
      });

      stream.on('error', function(err1) {
        err1.should.be.equal(customError);
      });

      stream.on('close', function() {
        done(err);
      });
    });
  });

  describe('13.3 Testing QueryStream\'s fetchArraySize option', function() {
    it('13.3.1 should use oracledb.fetchArraySize for fetching', function(done) {
      let defaultFetchArraySize;
      let testFetchArraySize = 9;

      defaultFetchArraySize = oracledb.fetchArraySize;
      oracledb.fetchArraySize = testFetchArraySize;
      let stream = connection.queryStream('SELECT employee_name FROM nodb_stream1 ORDER BY employee_name');
      let err;

      stream.on('data', function() {
        stream.pause();

        // Using the internal/private caches to validate
        should.equal(stream._resultSet._rowCache.length, testFetchArraySize - (1 + stream._readableState.buffer.length));

        stream.destroy();
        oracledb.fetchArraySize = defaultFetchArraySize;
      });

      stream.on('end', function() {
        oracledb.fetchArraySize = defaultFetchArraySize;
        err = new Error('Reached the end of the stream');
      });

      stream.on('error', function(err1) {
        oracledb.fetchArraySize = defaultFetchArraySize;
        err = err1;
      });

      stream.on('close', function() {
        oracledb.fetchArraySize = defaultFetchArraySize;
        done(err);
      });
    });

    it('13.3.2 should use execute options fetchArraySize for fetching', function(done) {
      let testFetchArraySize = 8;
      let stream = connection.queryStream('SELECT employee_name FROM nodb_stream1 ORDER BY employee_name', [], {fetchArraySize: testFetchArraySize});
      let err;

      stream.on('data', function() {
        stream.pause();

        // Using the internal/private caches to validate
        should.equal(stream._resultSet._rowCache.length, testFetchArraySize - (1 + stream._readableState.buffer.length));

        stream.destroy();
      });

      stream.on('end', function() {
        err = new Error('Reached the end of the stream');
      });

      stream.on('error', function(err1) {
        err = err1;
      });

      stream.on('close', function() {
        done(err);
      });
    });
  });

  describe('13.4 Testing QueryStream race conditions', function() {
    it('13.4.1 queryStream from toQueryStream should get open event', function(done) {
      connection.execute(
        'SELECT employee_name FROM nodb_stream1',
        [],
        {
          resultSet: true
        },
        function(err, result) {
          if (err) {
            done(err);
            return;
          }

          const stream = result.resultSet.toQueryStream();
          let receivedEvent = false;
          let err1;

          stream.on('open', function() {
            receivedEvent = true;
          });

          stream.on('data', function() {});

          stream.on('error', function(err2) {
            err1 = err2;
          });

          stream.on('end', function() {
            stream.destroy();
          });

          stream.on('close', function() {
            if (receivedEvent) {
              done(err1);
            } else {
              done(new Error('Did not receive event'));
            }
          });
        }
      );
    });

    it('13.4.2 queryStream from toQueryStream should get metadata event', function(done) {
      connection.execute(
        'SELECT employee_name FROM nodb_stream1',
        [],
        {
          resultSet: true
        },
        function(err, result) {
          if (err) {
            done(err);
            return;
          }

          const stream = result.resultSet.toQueryStream();
          let receivedEvent = false;
          let err1;

          stream.on('metadata', function() {
            receivedEvent = true;
          });

          stream.on('data', function() {});

          stream.on('error', function(err2) {
            err1 = err2;
          });

          stream.on('end', function() {
            stream.destroy();
          });

          stream.on('close', function() {
            if (receivedEvent) {
              done(err1);
            } else {
              done(new Error('Did not receive event'));
            }
          });
        }
      );
    });
  });
});