copy-both.js

'use strict'

module.exports = function (txt, options) {
  return new CopyStreamQuery(txt, options)
}

const { Duplex } = require('stream')
const assert = require('assert')
const BufferList = require('obuf')
const code = require('./message-formats')

// Readable decoder states
const PG_CODE = 0
const PG_LENGTH = 1
const PG_MESSAGE = 2

class CopyStreamQuery extends Duplex {
  constructor(text, options) {
    super(options)
    this.text = text

    // Readable side
    this._state = PG_CODE
    this._buffer = new BufferList()
    this._unreadMessageContentLength = 0
    this._copyDataChunks = new BufferList()
    this._pgDataHandler = null
    this._drained = false
    this._forwarding = false
    this._onReadableEvent = this._onReadable.bind(this)
    this.alignOnCopyDataFrame = options ? options.alignOnCopyDataFrame === true : false

    // Writable side
    this._gotCopyInResponse = false
    this.chunks = []
    this.cb = null
    this.cork()
  }

  submit(connection) {
    this.connection = connection
    this._attach()
    connection.query(this.text)
  }

  /* Readable implementation */
  _attach() {
    const connectionStream = this.connection.stream
    const pgDataListeners = connectionStream.listeners('data')
    assert(pgDataListeners.length == 1)
    this._pgDataHandler = pgDataListeners.pop()
    connectionStream.removeListener('data', this._pgDataHandler)
    connectionStream.pause()
    this._forward()
    connectionStream.on('readable', this._onReadableEvent)
  }

  _detach() {
    const connectionStream = this.connection.stream
    const unreadBuffer = this._buffer.take(this._buffer.size)
    connectionStream.removeListener('readable', this._onReadableEvent)
    connectionStream.addListener('data', this._pgDataHandler)
    this._pgDataHandler(unreadBuffer)

    // unpipe can pause the stream but also underlying onData event can potentially pause the stream because of hitting
    // the highWaterMark and pausing the stream, so we resume the stream in the next tick after the underlying onData
    // event has finished
    process.nextTick(function () {
      connectionStream.resume()
    })
  }

  _cleanup() {
    this._buffer = null
    this._copyDataChunks = null
    this._pgDataHandler = null
    this._onReadableEvent = null
  }

  _onReadable() {
    this._forward()
  }

  _read() {
    this._drained = true
    this._forward()
  }

  _forward() {
    if (this._forwarding || !this._drained || !this.connection) return
    this._forwarding = true
    const connectionStream = this.connection.stream
    let chunk
    while (this._drained && (chunk = connectionStream.read()) !== null) {
      this._drained = this._parse(chunk)
    }
    this._forwarding = false
  }

  _parse(chunk) {
    let done = false
    let drained = true
    this._buffer.push(chunk)

    while (!done && this._buffer.size > 0) {
      if (PG_CODE === this._state) {
        if (!this._buffer.has(1)) break
        this._code = this._buffer.peekUInt8()
        if (this._code === code.ErrorResponse) {
          // ErrorResponse Interception
          // We must let pg parse future messages and handle their consequences on
          // the ActiveQuery
          this._detach()
          return
        }
        this._buffer.readUInt8()
        this._state = PG_LENGTH
      }

      if (PG_LENGTH === this._state) {
        if (!this._buffer.has(4)) break
        this._unreadMessageContentLength = this._buffer.readUInt32BE() - 4
        this._state = PG_MESSAGE
      }

      if (PG_MESSAGE === this._state) {
        if (this._unreadMessageContentLength > 0 && this._buffer.size > 0) {
          const n = Math.min(this._buffer.size, this._unreadMessageContentLength)
          const messageContentChunk = this._buffer.take(n)
          this._unreadMessageContentLength -= n
          if (this._code === code.CopyData) {
            this._copyDataChunks.push(messageContentChunk)
          }
        }

        if (this._unreadMessageContentLength === 0) {
          // a full message has been captured
          switch (this._code) {
            case code.CopyBothResponse:
              this._startCopyIn()
              break
            case code.CopyData:
              if (this.alignOnCopyDataFrame) {
                drained = this._flushCopyData()
              }
              break
            // standard interspersed messages.
            // see https://www.postgresql.org/docs/9.6/protocol-flow.html#PROTOCOL-COPY
            case code.ParameterStatus:
            case code.NoticeResponse:
            case code.NotificationResponse:
              break
            case code.CopyDone:
            default:
              done = true
              break
          }
          this._state = PG_CODE
        }
      }
    }

    // When we are not in alignOnCopyDataFrame, copyData payload is not buffered
    // Forward payload bytes as they arrive
    if (!this.alignOnCopyDataFrame) {
      drained = this._flushCopyData()
    }

    if (done) {
      this._detach()
      this.push(null)
      this._cleanup()
    }

    return drained
  }

  _flushCopyData() {
    let drained = true
    const len = this._copyDataChunks.size
    if (len > 0) {
      drained = this.push(this._copyDataChunks.take(len))
    }
    return drained
  }

  /* Writable implementation */
  _write(chunk, enc, cb) {
    this.chunks.push({ chunk: chunk, encoding: enc })
    if (this._gotCopyInResponse) {
      return this.flush(cb)
    }
    this.cb = cb
  }

  _writev(chunks, cb) {
    this.chunks.push(...chunks)
    if (this._gotCopyInResponse) {
      return this.flush(cb)
    }
    this.cb = cb
  }

  _final(cb) {
    this.flush()
    const Int32Len = 4
    const finBuffer = Buffer.from([code.CopyDone, 0, 0, 0, Int32Len])
    this.connection.stream.write(finBuffer)
    this.cb_flush = cb
  }

  flush(callback) {
    let chunk
    let ok = true
    while (ok && (chunk = this.chunks.shift())) {
      ok = this.flushChunk(chunk.chunk)
    }
    if (callback) {
      if (ok) {
        callback()
      } else {
        if (this.chunks.length) {
          this.connection.stream.once('drain', this.flush.bind(this, callback))
        } else {
          this.connection.stream.once('drain', callback)
        }
      }
    }
  }

  flushChunk(chunk) {
    const Int32Len = 4
    const lenBuffer = Buffer.from([code.CopyData, 0, 0, 0, 0])
    lenBuffer.writeUInt32BE(chunk.length + Int32Len, 1)
    this.connection.stream.write(lenBuffer)
    return this.connection.stream.write(chunk)
  }

  _startCopyIn() {
    this._gotCopyInResponse = true
    this.uncork()
    this.flush()
    if (this.cb) {
      const { cb } = this
      this.cb = null
      cb()
    }
  }

  handleError(err) {
    this.emit('error', err)
    this._cleanup()
  }

  handleCopyData(chunk) {
    // an out of band copyData message
    // is received after copyDone
    // this is currently discarded
  }

  handleCommandComplete() {}

  handleReadyForQuery() {
    this.connection = null
    this.cb_flush()
  }
}