Index: node_modules/pg-protocol/src/b.ts =================================================================== --- node_modules/pg-protocol/src/b.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) +++ node_modules/pg-protocol/src/b.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) @@ -0,0 +1,25 @@ +// file for microbenchmarking + +import { BufferReader } from './buffer-reader' + +const LOOPS = 1000 +let count = 0 +const start = performance.now() + +const reader = new BufferReader() +const buffer = Buffer.from([33, 33, 33, 33, 33, 33, 33, 0]) + +const run = () => { + if (count > LOOPS) { + console.log(performance.now() - start) + return + } + count++ + for (let i = 0; i < LOOPS; i++) { + reader.setBuffer(0, buffer) + reader.cstring() + } + setImmediate(run) +} + +run() Index: node_modules/pg-protocol/src/buffer-reader.ts =================================================================== --- node_modules/pg-protocol/src/buffer-reader.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) +++ node_modules/pg-protocol/src/buffer-reader.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) @@ -0,0 +1,58 @@ +export class BufferReader { + private buffer: Buffer = Buffer.allocUnsafe(0) + + // TODO(bmc): support non-utf8 encoding? + private encoding: BufferEncoding = 'utf-8' + + constructor(private offset: number = 0) {} + + public setBuffer(offset: number, buffer: Buffer): void { + this.offset = offset + this.buffer = buffer + } + + public int16(): number { + const result = this.buffer.readInt16BE(this.offset) + this.offset += 2 + return result + } + + public byte(): number { + const result = this.buffer[this.offset] + this.offset++ + return result + } + + public int32(): number { + const result = this.buffer.readInt32BE(this.offset) + this.offset += 4 + return result + } + + public uint32(): number { + const result = this.buffer.readUInt32BE(this.offset) + this.offset += 4 + return result + } + + public string(length: number): string { + const result = this.buffer.toString(this.encoding, this.offset, this.offset + length) + this.offset += length + return result + } + + public cstring(): string { + const start = this.offset + let end = start + // eslint-disable-next-line no-empty + while (this.buffer[end++]) {} + this.offset = end + return this.buffer.toString(this.encoding, start, end - 1) + } + + public bytes(length: number): Buffer { + const result = this.buffer.slice(this.offset, this.offset + length) + this.offset += length + return result + } +} Index: node_modules/pg-protocol/src/buffer-writer.ts =================================================================== --- node_modules/pg-protocol/src/buffer-writer.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) +++ node_modules/pg-protocol/src/buffer-writer.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) @@ -0,0 +1,110 @@ +//binary data writer tuned for encoding binary specific to the postgres binary protocol + +export class Writer { + private buffer: Buffer + private offset: number = 5 + private headerPosition: number = 0 + constructor(private size = 256) { + this.buffer = Buffer.allocUnsafe(size) + } + + private ensure(size: number): void { + const remaining = this.buffer.length - this.offset + if (remaining < size) { + const oldBuffer = this.buffer + // exponential growth factor of around ~ 1.5 + // https://stackoverflow.com/questions/2269063/buffer-growth-strategy + const newSize = oldBuffer.length + (oldBuffer.length >> 1) + size + this.buffer = Buffer.allocUnsafe(newSize) + oldBuffer.copy(this.buffer) + } + } + + public addInt32(num: number): Writer { + this.ensure(4) + this.buffer[this.offset++] = (num >>> 24) & 0xff + this.buffer[this.offset++] = (num >>> 16) & 0xff + this.buffer[this.offset++] = (num >>> 8) & 0xff + this.buffer[this.offset++] = (num >>> 0) & 0xff + return this + } + + public addInt16(num: number): Writer { + this.ensure(2) + this.buffer[this.offset++] = (num >>> 8) & 0xff + this.buffer[this.offset++] = (num >>> 0) & 0xff + return this + } + + public addCString(string: string): Writer { + if (!string) { + this.ensure(1) + } else { + const len = Buffer.byteLength(string) + this.ensure(len + 1) // +1 for null terminator + this.buffer.write(string, this.offset, 'utf-8') + this.offset += len + } + + this.buffer[this.offset++] = 0 // null terminator + return this + } + + public addString(string: string = ''): Writer { + const len = Buffer.byteLength(string) + this.ensure(len) + this.buffer.write(string, this.offset) + this.offset += len + return this + } + + // Write an Int32 byte-length prefix immediately followed by the string's UTF-8 + // bytes. Postgres' Bind wire format prefixes every parameter with its length, + // and doing it in one method computes Buffer.byteLength ONCE — the previous + // `addInt32(Buffer.byteLength(s)).addString(s)` pairing scanned the string + // three times (byteLength for the prefix, byteLength again inside addString, + // then the encode), which is costly for large text parameters. + public addInt32PrefixedString(string: string): Writer { + const len = Buffer.byteLength(string) + this.ensure(4 + len) + const buffer = this.buffer + let offset = this.offset + buffer[offset++] = (len >>> 24) & 0xff + buffer[offset++] = (len >>> 16) & 0xff + buffer[offset++] = (len >>> 8) & 0xff + buffer[offset++] = (len >>> 0) & 0xff + buffer.write(string, offset, 'utf-8') + this.offset = offset + len + return this + } + + public add(otherBuffer: Buffer): Writer { + this.ensure(otherBuffer.length) + otherBuffer.copy(this.buffer, this.offset) + this.offset += otherBuffer.length + return this + } + + private join(code?: number): Buffer { + if (code) { + this.buffer[this.headerPosition] = code + //length is everything in this packet minus the code + const length = this.offset - (this.headerPosition + 1) + this.buffer.writeInt32BE(length, this.headerPosition + 1) + } + return this.buffer.slice(code ? 0 : 5, this.offset) + } + + public flush(code?: number): Buffer { + const result = this.join(code) + this.offset = 5 + this.headerPosition = 0 + this.buffer = Buffer.allocUnsafe(this.size) + return result + } + + public clear(): void { + this.offset = 5 + this.headerPosition = 0 + } +} Index: node_modules/pg-protocol/src/inbound-parser.test.ts =================================================================== --- node_modules/pg-protocol/src/inbound-parser.test.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) +++ node_modules/pg-protocol/src/inbound-parser.test.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) @@ -0,0 +1,585 @@ +import buffers from './testing/test-buffers' +import BufferList from './testing/buffer-list' +import { parse } from '.' +import assert from 'assert' +import { PassThrough } from 'stream' +import { BackendMessage } from './messages' +import { Parser } from './parser' + +const authOkBuffer = buffers.authenticationOk() +const paramStatusBuffer = buffers.parameterStatus('client_encoding', 'UTF8') +const readyForQueryBuffer = buffers.readyForQuery() +const backendKeyDataBuffer = buffers.backendKeyData(1, 2) +const commandCompleteBuffer = buffers.commandComplete('SELECT 3') +const parseCompleteBuffer = buffers.parseComplete() +const bindCompleteBuffer = buffers.bindComplete() +const portalSuspendedBuffer = buffers.portalSuspended() + +const row1 = { + name: 'id', + tableID: 1, + attributeNumber: 2, + dataTypeID: 3, + dataTypeSize: 4, + typeModifier: 5, + formatCode: 0, +} +const oneRowDescBuff = buffers.rowDescription([row1]) +row1.name = 'bang' + +const twoRowBuf = buffers.rowDescription([ + row1, + { + name: 'whoah', + tableID: 10, + attributeNumber: 11, + dataTypeID: 12, + dataTypeSize: 13, + typeModifier: 14, + formatCode: 0, + }, +]) + +const rowWithBigOids = { + name: 'bigoid', + tableID: 3000000001, + attributeNumber: 2, + dataTypeID: 3000000003, + dataTypeSize: 4, + typeModifier: 5, + formatCode: 0, +} +const bigOidDescBuff = buffers.rowDescription([rowWithBigOids]) + +const emptyRowFieldBuf = buffers.dataRow([]) + +const oneFieldBuf = buffers.dataRow(['test']) + +const expectedAuthenticationOkayMessage = { + name: 'authenticationOk', + length: 8, +} + +const expectedParameterStatusMessage = { + name: 'parameterStatus', + parameterName: 'client_encoding', + parameterValue: 'UTF8', + length: 25, +} + +const expectedBackendKeyDataMessage = { + name: 'backendKeyData', + processID: 1, + secretKey: 2, +} + +const expectedReadyForQueryMessage = { + name: 'readyForQuery', + length: 5, + status: 'I', +} + +const expectedCommandCompleteMessage = { + name: 'commandComplete', + length: 13, + text: 'SELECT 3', +} +const emptyRowDescriptionBuffer = new BufferList() + .addInt16(0) // number of fields + .join(true, 'T') + +const expectedEmptyRowDescriptionMessage = { + name: 'rowDescription', + length: 6, + fieldCount: 0, + fields: [], +} +const expectedOneRowMessage = { + name: 'rowDescription', + length: 27, + fieldCount: 1, + fields: [ + { + name: 'id', + tableID: 1, + columnID: 2, + dataTypeID: 3, + dataTypeSize: 4, + dataTypeModifier: 5, + format: 'text', + }, + ], +} + +const expectedTwoRowMessage = { + name: 'rowDescription', + length: 53, + fieldCount: 2, + fields: [ + { + name: 'bang', + tableID: 1, + columnID: 2, + dataTypeID: 3, + dataTypeSize: 4, + dataTypeModifier: 5, + format: 'text', + }, + { + name: 'whoah', + tableID: 10, + columnID: 11, + dataTypeID: 12, + dataTypeSize: 13, + dataTypeModifier: 14, + format: 'text', + }, + ], +} +const expectedBigOidMessage = { + name: 'rowDescription', + length: 31, + fieldCount: 1, + fields: [ + { + name: 'bigoid', + tableID: 3000000001, + columnID: 2, + dataTypeID: 3000000003, + dataTypeSize: 4, + dataTypeModifier: 5, + format: 'text', + }, + ], +} + +const emptyParameterDescriptionBuffer = new BufferList() + .addInt16(0) // number of parameters + .join(true, 't') + +const oneParameterDescBuf = buffers.parameterDescription([1111]) + +const twoParameterDescBuf = buffers.parameterDescription([2222, 3333]) + +const bigOidParameterDescBuf = buffers.parameterDescription([3000000003]) + +const expectedEmptyParameterDescriptionMessage = { + name: 'parameterDescription', + length: 6, + parameterCount: 0, + dataTypeIDs: [], +} + +const expectedOneParameterMessage = { + name: 'parameterDescription', + length: 10, + parameterCount: 1, + dataTypeIDs: [1111], +} + +const expectedTwoParameterMessage = { + name: 'parameterDescription', + length: 14, + parameterCount: 2, + dataTypeIDs: [2222, 3333], +} + +const expectedBigOidParameterMessage = { + name: 'parameterDescription', + length: 10, + parameterCount: 1, + dataTypeIDs: [3000000003], +} + +const testForMessage = function (buffer: Buffer, expectedMessage: any) { + it('receives and parses ' + expectedMessage.name, async () => { + const messages = await parseBuffers([buffer]) + const [lastMessage] = messages + + for (const key in expectedMessage) { + assert.deepEqual((lastMessage as any)[key], expectedMessage[key]) + } + }) +} + +const plainPasswordBuffer = buffers.authenticationCleartextPassword() +const md5PasswordBuffer = buffers.authenticationMD5Password() +const SASLBuffer = buffers.authenticationSASL() +const SASLContinueBuffer = buffers.authenticationSASLContinue() +const SASLFinalBuffer = buffers.authenticationSASLFinal() + +const expectedPlainPasswordMessage = { + name: 'authenticationCleartextPassword', +} + +const expectedMD5PasswordMessage = { + name: 'authenticationMD5Password', + salt: Buffer.from([1, 2, 3, 4]), +} + +const expectedSASLMessage = { + name: 'authenticationSASL', + mechanisms: ['SCRAM-SHA-256'], +} + +const expectedSASLContinueMessage = { + name: 'authenticationSASLContinue', + data: 'data', +} + +const expectedSASLFinalMessage = { + name: 'authenticationSASLFinal', + data: 'data', +} + +const notificationResponseBuffer = buffers.notification(4, 'hi', 'boom') +const expectedNotificationResponseMessage = { + name: 'notification', + processId: 4, + channel: 'hi', + payload: 'boom', +} + +const parseBuffers = async (buffers: Buffer[]): Promise => { + const stream = new PassThrough() + for (const buffer of buffers) { + stream.write(buffer) + } + stream.end() + const msgs: BackendMessage[] = [] + await parse(stream, (msg) => msgs.push(msg)) + return msgs +} + +describe('PgPacketStream', function () { + testForMessage(authOkBuffer, expectedAuthenticationOkayMessage) + testForMessage(plainPasswordBuffer, expectedPlainPasswordMessage) + testForMessage(md5PasswordBuffer, expectedMD5PasswordMessage) + testForMessage(SASLBuffer, expectedSASLMessage) + testForMessage(SASLContinueBuffer, expectedSASLContinueMessage) + + // this exercises a found bug in the parser: + // https://github.com/brianc/node-postgres/pull/2210#issuecomment-627626084 + // and adds a test which is deterministic, rather than relying on network packet chunking + const extendedSASLContinueBuffer = Buffer.concat([SASLContinueBuffer, Buffer.from([1, 2, 3, 4])]) + testForMessage(extendedSASLContinueBuffer, expectedSASLContinueMessage) + + testForMessage(SASLFinalBuffer, expectedSASLFinalMessage) + + // this exercises a found bug in the parser: + // https://github.com/brianc/node-postgres/pull/2210#issuecomment-627626084 + // and adds a test which is deterministic, rather than relying on network packet chunking + const extendedSASLFinalBuffer = Buffer.concat([SASLFinalBuffer, Buffer.from([1, 2, 4, 5])]) + testForMessage(extendedSASLFinalBuffer, expectedSASLFinalMessage) + + testForMessage(paramStatusBuffer, expectedParameterStatusMessage) + testForMessage(backendKeyDataBuffer, expectedBackendKeyDataMessage) + testForMessage(readyForQueryBuffer, expectedReadyForQueryMessage) + testForMessage(commandCompleteBuffer, expectedCommandCompleteMessage) + testForMessage(notificationResponseBuffer, expectedNotificationResponseMessage) + testForMessage(buffers.emptyQuery(), { + name: 'emptyQuery', + length: 4, + }) + + testForMessage(Buffer.from([0x6e, 0, 0, 0, 4]), { + name: 'noData', + }) + + describe('rowDescription messages', function () { + testForMessage(emptyRowDescriptionBuffer, expectedEmptyRowDescriptionMessage) + testForMessage(oneRowDescBuff, expectedOneRowMessage) + testForMessage(twoRowBuf, expectedTwoRowMessage) + testForMessage(bigOidDescBuff, expectedBigOidMessage) + }) + + describe('parameterDescription messages', function () { + testForMessage(emptyParameterDescriptionBuffer, expectedEmptyParameterDescriptionMessage) + testForMessage(oneParameterDescBuf, expectedOneParameterMessage) + testForMessage(twoParameterDescBuf, expectedTwoParameterMessage) + testForMessage(bigOidParameterDescBuf, expectedBigOidParameterMessage) + }) + + describe('parsing rows', function () { + describe('parsing empty row', function () { + testForMessage(emptyRowFieldBuf, { + name: 'dataRow', + fieldCount: 0, + }) + }) + + describe('parsing data row with fields', function () { + testForMessage(oneFieldBuf, { + name: 'dataRow', + fieldCount: 1, + fields: ['test'], + }) + }) + }) + + describe('notice message', function () { + // this uses the same logic as error message + const buff = buffers.notice([{ type: 'C', value: 'code' }]) + testForMessage(buff, { + name: 'notice', + code: 'code', + }) + }) + + testForMessage(buffers.error([]), { + name: 'error', + }) + + describe('with all the fields', function () { + const buffer = buffers.error([ + { + type: 'S', + value: 'ERROR', + }, + { + type: 'C', + value: 'code', + }, + { + type: 'M', + value: 'message', + }, + { + type: 'D', + value: 'details', + }, + { + type: 'H', + value: 'hint', + }, + { + type: 'P', + value: '100', + }, + { + type: 'p', + value: '101', + }, + { + type: 'q', + value: 'query', + }, + { + type: 'W', + value: 'where', + }, + { + type: 'F', + value: 'file', + }, + { + type: 'L', + value: 'line', + }, + { + type: 'R', + value: 'routine', + }, + { + type: 'Z', // ignored + value: 'alsdkf', + }, + ]) + + testForMessage(buffer, { + name: 'error', + severity: 'ERROR', + code: 'code', + message: 'message', + detail: 'details', + hint: 'hint', + position: '100', + internalPosition: '101', + internalQuery: 'query', + where: 'where', + file: 'file', + line: 'line', + routine: 'routine', + }) + }) + + testForMessage(parseCompleteBuffer, { + name: 'parseComplete', + }) + + testForMessage(bindCompleteBuffer, { + name: 'bindComplete', + }) + + testForMessage(bindCompleteBuffer, { + name: 'bindComplete', + }) + + testForMessage(buffers.closeComplete(), { + name: 'closeComplete', + }) + + describe('parses portal suspended message', function () { + testForMessage(portalSuspendedBuffer, { + name: 'portalSuspended', + }) + }) + + describe('parses replication start message', function () { + testForMessage(Buffer.from([0x57, 0x00, 0x00, 0x00, 0x04]), { + name: 'replicationStart', + length: 4, + }) + }) + + describe('copy', () => { + testForMessage(buffers.copyIn(0), { + name: 'copyInResponse', + length: 7, + binary: false, + columnTypes: [], + }) + + testForMessage(buffers.copyIn(2), { + name: 'copyInResponse', + length: 11, + binary: false, + columnTypes: [0, 1], + }) + + testForMessage(buffers.copyOut(0), { + name: 'copyOutResponse', + length: 7, + binary: false, + columnTypes: [], + }) + + testForMessage(buffers.copyOut(3), { + name: 'copyOutResponse', + length: 13, + binary: false, + columnTypes: [0, 1, 2], + }) + + testForMessage(buffers.copyDone(), { + name: 'copyDone', + length: 4, + }) + + testForMessage(buffers.copyData(Buffer.from([5, 6, 7])), { + name: 'copyData', + length: 7, + chunk: Buffer.from([5, 6, 7]), + }) + }) + + // since the data message on a stream can randomly divide the incomming + // tcp packets anywhere, we need to make sure we can parse every single + // split on a tcp message + describe('split buffer, single message parsing', function () { + const fullBuffer = buffers.dataRow([null, 'bang', 'zug zug', null, '!']) + + it('parses when full buffer comes in', async function () { + const messages = await parseBuffers([fullBuffer]) + const message = messages[0] as any + assert.equal(message.fields.length, 5) + assert.equal(message.fields[0], null) + assert.equal(message.fields[1], 'bang') + assert.equal(message.fields[2], 'zug zug') + assert.equal(message.fields[3], null) + assert.equal(message.fields[4], '!') + }) + + const testMessageReceivedAfterSplitAt = async function (split: number) { + const firstBuffer = Buffer.alloc(fullBuffer.length - split) + const secondBuffer = Buffer.alloc(fullBuffer.length - firstBuffer.length) + fullBuffer.copy(firstBuffer, 0, 0) + fullBuffer.copy(secondBuffer, 0, firstBuffer.length) + const messages = await parseBuffers([firstBuffer, secondBuffer]) + const message = messages[0] as any + assert.equal(message.fields.length, 5) + assert.equal(message.fields[0], null) + assert.equal(message.fields[1], 'bang') + assert.equal(message.fields[2], 'zug zug') + assert.equal(message.fields[3], null) + assert.equal(message.fields[4], '!') + } + + it('parses when split in the middle', function () { + return testMessageReceivedAfterSplitAt(6) + }) + + it('parses when split at end', function () { + return testMessageReceivedAfterSplitAt(2) + }) + + it('parses when split at beginning', function () { + return Promise.all([ + testMessageReceivedAfterSplitAt(fullBuffer.length - 2), + testMessageReceivedAfterSplitAt(fullBuffer.length - 1), + testMessageReceivedAfterSplitAt(fullBuffer.length - 5), + ]) + }) + }) + + describe('split buffer, multiple message parsing', function () { + const dataRowBuffer = buffers.dataRow(['!']) + const readyForQueryBuffer = buffers.readyForQuery() + const fullBuffer = Buffer.alloc(dataRowBuffer.length + readyForQueryBuffer.length) + dataRowBuffer.copy(fullBuffer, 0, 0) + readyForQueryBuffer.copy(fullBuffer, dataRowBuffer.length, 0) + + const verifyMessages = function (messages: any[]) { + assert.strictEqual(messages.length, 2) + assert.deepEqual(messages[0], { + name: 'dataRow', + fieldCount: 1, + length: 11, + fields: ['!'], + }) + assert.equal(messages[0].fields[0], '!') + assert.deepEqual(messages[1], { + name: 'readyForQuery', + length: 5, + status: 'I', + }) + } + // sanity check + it('receives both messages when packet is not split', async function () { + const messages = await parseBuffers([fullBuffer]) + verifyMessages(messages) + }) + + const splitAndVerifyTwoMessages = async function (split: number) { + const firstBuffer = Buffer.alloc(fullBuffer.length - split) + const secondBuffer = Buffer.alloc(fullBuffer.length - firstBuffer.length) + fullBuffer.copy(firstBuffer, 0, 0) + fullBuffer.copy(secondBuffer, 0, firstBuffer.length) + const messages = await parseBuffers([firstBuffer, secondBuffer]) + verifyMessages(messages) + } + + describe('receives both messages when packet is split', function () { + it('in the middle', function () { + return splitAndVerifyTwoMessages(11) + }) + it('at the front', function () { + return Promise.all([ + splitAndVerifyTwoMessages(fullBuffer.length - 1), + splitAndVerifyTwoMessages(fullBuffer.length - 4), + splitAndVerifyTwoMessages(fullBuffer.length - 6), + ]) + }) + + it('at the end', function () { + return Promise.all([splitAndVerifyTwoMessages(8), splitAndVerifyTwoMessages(1)]) + }) + }) + }) + + it('cleans up the reader after handling a packet', function () { + const parser = new Parser() + parser.parse(oneFieldBuf, () => {}) + assert.strictEqual((parser as any).reader.buffer.byteLength, 0) + }) +}) Index: node_modules/pg-protocol/src/index.ts =================================================================== --- node_modules/pg-protocol/src/index.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) +++ node_modules/pg-protocol/src/index.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) @@ -0,0 +1,11 @@ +import { DatabaseError } from './messages' +import { serialize } from './serializer' +import { Parser, MessageCallback } from './parser' + +export function parse(stream: NodeJS.ReadableStream, callback: MessageCallback): Promise { + const parser = new Parser() + stream.on('data', (buffer: Buffer) => parser.parse(buffer, callback)) + return new Promise((resolve) => stream.on('end', () => resolve())) +} + +export { serialize, DatabaseError } Index: node_modules/pg-protocol/src/messages.ts =================================================================== --- node_modules/pg-protocol/src/messages.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) +++ node_modules/pg-protocol/src/messages.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) @@ -0,0 +1,262 @@ +export type Mode = 'text' | 'binary' + +export type MessageName = + | 'parseComplete' + | 'bindComplete' + | 'closeComplete' + | 'noData' + | 'portalSuspended' + | 'replicationStart' + | 'emptyQuery' + | 'copyDone' + | 'copyData' + | 'rowDescription' + | 'parameterDescription' + | 'parameterStatus' + | 'backendKeyData' + | 'notification' + | 'readyForQuery' + | 'commandComplete' + | 'dataRow' + | 'copyInResponse' + | 'copyOutResponse' + | 'authenticationOk' + | 'authenticationMD5Password' + | 'authenticationCleartextPassword' + | 'authenticationSASL' + | 'authenticationSASLContinue' + | 'authenticationSASLFinal' + | 'error' + | 'notice' + +export interface BackendMessage { + name: MessageName + length: number +} + +export const parseComplete: BackendMessage = { + name: 'parseComplete', + length: 5, +} + +export const bindComplete: BackendMessage = { + name: 'bindComplete', + length: 5, +} + +export const closeComplete: BackendMessage = { + name: 'closeComplete', + length: 5, +} + +export const noData: BackendMessage = { + name: 'noData', + length: 5, +} + +export const portalSuspended: BackendMessage = { + name: 'portalSuspended', + length: 5, +} + +export const replicationStart: BackendMessage = { + name: 'replicationStart', + length: 4, +} + +export const emptyQuery: BackendMessage = { + name: 'emptyQuery', + length: 4, +} + +export const copyDone: BackendMessage = { + name: 'copyDone', + length: 4, +} + +interface NoticeOrError { + message: string | undefined + severity: string | undefined + code: string | undefined + detail: string | undefined + hint: string | undefined + position: string | undefined + internalPosition: string | undefined + internalQuery: string | undefined + where: string | undefined + schema: string | undefined + table: string | undefined + column: string | undefined + dataType: string | undefined + constraint: string | undefined + file: string | undefined + line: string | undefined + routine: string | undefined +} + +export class DatabaseError extends Error implements NoticeOrError { + public severity: string | undefined + public code: string | undefined + public detail: string | undefined + public hint: string | undefined + public position: string | undefined + public internalPosition: string | undefined + public internalQuery: string | undefined + public where: string | undefined + public schema: string | undefined + public table: string | undefined + public column: string | undefined + public dataType: string | undefined + public constraint: string | undefined + public file: string | undefined + public line: string | undefined + public routine: string | undefined + constructor( + message: string, + public readonly length: number, + public readonly name: MessageName + ) { + super(message) + } +} + +export class CopyDataMessage { + public readonly name = 'copyData' + constructor( + public readonly length: number, + public readonly chunk: Buffer + ) {} +} + +export class CopyResponse { + public readonly columnTypes: number[] + constructor( + public readonly length: number, + public readonly name: MessageName, + public readonly binary: boolean, + columnCount: number + ) { + this.columnTypes = new Array(columnCount) + } +} + +export class Field { + constructor( + public readonly name: string, + public readonly tableID: number, + public readonly columnID: number, + public readonly dataTypeID: number, + public readonly dataTypeSize: number, + public readonly dataTypeModifier: number, + public readonly format: Mode + ) {} +} + +export class RowDescriptionMessage { + public readonly name: MessageName = 'rowDescription' + public readonly fields: Field[] + constructor( + public readonly length: number, + public readonly fieldCount: number + ) { + this.fields = new Array(this.fieldCount) + } +} + +export class ParameterDescriptionMessage { + public readonly name: MessageName = 'parameterDescription' + public readonly dataTypeIDs: number[] + constructor( + public readonly length: number, + public readonly parameterCount: number + ) { + this.dataTypeIDs = new Array(this.parameterCount) + } +} + +export class ParameterStatusMessage { + public readonly name: MessageName = 'parameterStatus' + constructor( + public readonly length: number, + public readonly parameterName: string, + public readonly parameterValue: string + ) {} +} + +export class AuthenticationMD5Password implements BackendMessage { + public readonly name: MessageName = 'authenticationMD5Password' + constructor( + public readonly length: number, + public readonly salt: Buffer + ) {} +} + +export class BackendKeyDataMessage { + public readonly name: MessageName = 'backendKeyData' + constructor( + public readonly length: number, + public readonly processID: number, + public readonly secretKey: number + ) {} +} + +export class NotificationResponseMessage { + public readonly name: MessageName = 'notification' + constructor( + public readonly length: number, + public readonly processId: number, + public readonly channel: string, + public readonly payload: string + ) {} +} + +export class ReadyForQueryMessage { + public readonly name: MessageName = 'readyForQuery' + constructor( + public readonly length: number, + public readonly status: string + ) {} +} + +export class CommandCompleteMessage { + public readonly name: MessageName = 'commandComplete' + constructor( + public readonly length: number, + public readonly text: string + ) {} +} + +export class DataRowMessage { + public readonly fieldCount: number + public readonly name: MessageName = 'dataRow' + constructor( + public length: number, + public fields: any[] + ) { + this.fieldCount = fields.length + } +} + +export class NoticeMessage implements BackendMessage, NoticeOrError { + constructor( + public readonly length: number, + public readonly message: string | undefined + ) {} + public readonly name = 'notice' + public severity: string | undefined + public code: string | undefined + public detail: string | undefined + public hint: string | undefined + public position: string | undefined + public internalPosition: string | undefined + public internalQuery: string | undefined + public where: string | undefined + public schema: string | undefined + public table: string | undefined + public column: string | undefined + public dataType: string | undefined + public constraint: string | undefined + public file: string | undefined + public line: string | undefined + public routine: string | undefined +} Index: node_modules/pg-protocol/src/outbound-serializer.test.ts =================================================================== --- node_modules/pg-protocol/src/outbound-serializer.test.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) +++ node_modules/pg-protocol/src/outbound-serializer.test.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) @@ -0,0 +1,356 @@ +import assert from 'assert' +import { serialize } from './serializer' +import BufferList from './testing/buffer-list' + +describe('serializer', () => { + it('builds startup message', function () { + const actual = serialize.startup({ + user: 'brian', + database: 'bang', + }) + assert.deepEqual( + actual, + new BufferList() + .addInt16(3) + .addInt16(0) + .addCString('user') + .addCString('brian') + .addCString('database') + .addCString('bang') + .addCString('client_encoding') + .addCString('UTF8') + .addCString('') + .join(true) + ) + }) + + it('builds password message', function () { + const actual = serialize.password('!') + assert.deepEqual(actual, new BufferList().addCString('!').join(true, 'p')) + }) + + it('builds request ssl message', function () { + const actual = serialize.requestSsl() + const expected = new BufferList().addInt32(80877103).join(true) + assert.deepEqual(actual, expected) + }) + + it('builds SASLInitialResponseMessage message', function () { + const actual = serialize.sendSASLInitialResponseMessage('mech', 'data') + assert.deepEqual(actual, new BufferList().addCString('mech').addInt32(4).addString('data').join(true, 'p')) + }) + + it('builds SCRAMClientFinalMessage message', function () { + const actual = serialize.sendSCRAMClientFinalMessage('data') + assert.deepEqual(actual, new BufferList().addString('data').join(true, 'p')) + }) + + it('builds query message', function () { + const txt = 'select * from boom' + const actual = serialize.query(txt) + assert.deepEqual(actual, new BufferList().addCString(txt).join(true, 'Q')) + }) + + describe('parse message', () => { + it('builds parse message', function () { + const actual = serialize.parse({ text: '!' }) + const expected = new BufferList().addCString('').addCString('!').addInt16(0).join(true, 'P') + assert.deepEqual(actual, expected) + }) + + it('builds parse message with named query', function () { + const actual = serialize.parse({ + name: 'boom', + text: 'select * from boom', + types: [], + }) + const expected = new BufferList().addCString('boom').addCString('select * from boom').addInt16(0).join(true, 'P') + assert.deepEqual(actual, expected) + }) + + it('with multiple parameters', function () { + const actual = serialize.parse({ + name: 'force', + text: 'select * from bang where name = $1', + types: [1, 2, 3, 4], + }) + const expected = new BufferList() + .addCString('force') + .addCString('select * from bang where name = $1') + .addInt16(4) + .addInt32(1) + .addInt32(2) + .addInt32(3) + .addInt32(4) + .join(true, 'P') + assert.deepEqual(actual, expected) + }) + }) + + describe('bind messages', function () { + it('with no values', function () { + const actual = serialize.bind() + + const expectedBuffer = new BufferList() + .addCString('') + .addCString('') + .addInt16(0) + .addInt16(0) + .addInt16(1) + .addInt16(0) + .join(true, 'B') + assert.deepEqual(actual, expectedBuffer) + }) + + it('with named statement, portal, and values', function () { + const actual = serialize.bind({ + portal: 'bang', + statement: 'woo', + values: ['1', 'hi', null, 'zing'], + }) + const expectedBuffer = new BufferList() + .addCString('bang') // portal name + .addCString('woo') // statement name + .addInt16(4) + .addInt16(0) + .addInt16(0) + .addInt16(0) + .addInt16(0) + .addInt16(4) + .addInt32(1) + .add(Buffer.from('1')) + .addInt32(2) + .add(Buffer.from('hi')) + .addInt32(-1) + .addInt32(4) + .add(Buffer.from('zing')) + .addInt16(1) + .addInt16(0) + .join(true, 'B') + assert.deepEqual(actual, expectedBuffer) + }) + + it('encodes a multi-byte string param with its UTF-8 byte length, not char length', function () { + // Guards the single-pass addInt32PrefixedString write path: the Int32 + // length prefix must be the UTF-8 byte count, not String.length. 'héllo中🎉' + // is 7 code points / 8 UTF-16 code units but 13 UTF-8 bytes. + const value = 'héllo中🎉' + const bytes = Buffer.from(value, 'utf8') + assert.notEqual(bytes.length, value.length) // sanity: the divergence we're testing + const actual = serialize.bind({ values: [value] }) + const expectedBuffer = new BufferList() + .addCString('') // portal + .addCString('') // statement + .addInt16(1) // param format code count + .addInt16(0) // format code for the one value (text) + .addInt16(1) // value count + .addInt32(bytes.length) // 13 — the UTF-8 byte length, NOT value.length (8) + .add(bytes) + .addInt16(1) // result format code count + .addInt16(0) // result format (text) + .join(true, 'B') + assert.deepEqual(actual, expectedBuffer) + }) + }) + + it('with custom valueMapper', function () { + const actual = serialize.bind({ + portal: 'bang', + statement: 'woo', + values: ['1', 'hi', null, 'zing'], + valueMapper: () => null, + }) + const expectedBuffer = new BufferList() + .addCString('bang') // portal name + .addCString('woo') // statement name + .addInt16(4) + .addInt16(0) + .addInt16(0) + .addInt16(0) + .addInt16(0) + .addInt16(4) + .addInt32(-1) + .addInt32(-1) + .addInt32(-1) + .addInt32(-1) + .addInt16(1) + .addInt16(0) + .join(true, 'B') + assert.deepEqual(actual, expectedBuffer) + }) + + it('with named statement, portal, and buffer value', function () { + const actual = serialize.bind({ + portal: 'bang', + statement: 'woo', + values: ['1', 'hi', null, Buffer.from('zing', 'utf8')], + }) + const expectedBuffer = new BufferList() + .addCString('bang') // portal name + .addCString('woo') // statement name + .addInt16(4) // value count + .addInt16(0) // string + .addInt16(0) // string + .addInt16(0) // string + .addInt16(1) // binary + .addInt16(4) + .addInt32(1) + .add(Buffer.from('1')) + .addInt32(2) + .add(Buffer.from('hi')) + .addInt32(-1) + .addInt32(4) + .add(Buffer.from('zing', 'utf-8')) + .addInt16(1) + .addInt16(0) + .join(true, 'B') + assert.deepEqual(actual, expectedBuffer) + }) + + describe('builds execute message', function () { + it('for unamed portal with no row limit', function () { + const actual = serialize.execute() + const expectedBuffer = new BufferList().addCString('').addInt32(0).join(true, 'E') + assert.deepEqual(actual, expectedBuffer) + }) + + it('for named portal with row limit', function () { + const actual = serialize.execute({ + portal: 'my favorite portal', + rows: 100, + }) + const expectedBuffer = new BufferList().addCString('my favorite portal').addInt32(100).join(true, 'E') + assert.deepEqual(actual, expectedBuffer) + }) + }) + + it('builds flush command', function () { + const actual = serialize.flush() + const expected = new BufferList().join(true, 'H') + assert.deepEqual(actual, expected) + }) + + it('builds sync command', function () { + const actual = serialize.sync() + const expected = new BufferList().join(true, 'S') + assert.deepEqual(actual, expected) + }) + + it('builds end command', function () { + const actual = serialize.end() + const expected = Buffer.from([0x58, 0, 0, 0, 4]) + assert.deepEqual(actual, expected) + }) + + describe('builds describe command', function () { + it('describe statement', function () { + const actual = serialize.describe({ type: 'S', name: 'bang' }) + const expected = new BufferList().addChar('S').addCString('bang').join(true, 'D') + assert.deepEqual(actual, expected) + }) + + it('describe unnamed portal', function () { + const actual = serialize.describe({ type: 'P' }) + const expected = new BufferList().addChar('P').addCString('').join(true, 'D') + assert.deepEqual(actual, expected) + }) + }) + + describe('builds close command', function () { + it('describe statement', function () { + const actual = serialize.close({ type: 'S', name: 'bang' }) + const expected = new BufferList().addChar('S').addCString('bang').join(true, 'C') + assert.deepEqual(actual, expected) + }) + + it('describe unnamed portal', function () { + const actual = serialize.close({ type: 'P' }) + const expected = new BufferList().addChar('P').addCString('').join(true, 'C') + assert.deepEqual(actual, expected) + }) + }) + + describe('copy messages', function () { + it('builds copyFromChunk', () => { + const actual = serialize.copyData(Buffer.from([1, 2, 3])) + const expected = new BufferList().add(Buffer.from([1, 2, 3])).join(true, 'd') + assert.deepEqual(actual, expected) + }) + + it('builds copy fail', () => { + const actual = serialize.copyFail('err!') + const expected = new BufferList().addCString('err!').join(true, 'f') + assert.deepEqual(actual, expected) + }) + + it('builds copy done', () => { + const actual = serialize.copyDone() + const expected = new BufferList().join(true, 'c') + assert.deepEqual(actual, expected) + }) + }) + + it('builds cancel message', () => { + const actual = serialize.cancel(3, 4) + const expected = new BufferList().addInt16(1234).addInt16(5678).addInt32(3).addInt32(4).join(true) + assert.deepEqual(actual, expected) + }) + + describe('bind error recovery', () => { + const throwingMapper = () => { + throw new Error('valueMapper error') + } + + it('produces correct bind output after a valueMapper exception', () => { + assert.throws(() => { + serialize.bind({ + values: ['fail'], + valueMapper: throwingMapper, + }) + }, /valueMapper error/) + + const actual = serialize.bind({ + portal: 'bang', + statement: 'woo', + values: ['1', 'hi', null, 'zing'], + }) + const expectedBuffer = new BufferList() + .addCString('bang') + .addCString('woo') + .addInt16(4) + .addInt16(0) + .addInt16(0) + .addInt16(0) + .addInt16(0) + .addInt16(4) + .addInt32(1) + .add(Buffer.from('1')) + .addInt32(2) + .add(Buffer.from('hi')) + .addInt32(-1) + .addInt32(4) + .add(Buffer.from('zing')) + .addInt16(1) + .addInt16(0) + .join(true, 'B') + assert.deepEqual(actual, expectedBuffer) + }) + + it('produces correct output from other serializer methods after a failed bind', () => { + assert.throws(() => { + serialize.bind({ + values: ['fail'], + valueMapper: throwingMapper, + }) + }, /valueMapper error/) + + const parseActual = serialize.parse({ text: '!' }) + const parseExpected = new BufferList().addCString('').addCString('!').addInt16(0).join(true, 'P') + assert.deepEqual(parseActual, parseExpected) + + const queryActual = serialize.query('select 1') + const queryExpected = new BufferList().addCString('select 1').join(true, 'Q') + assert.deepEqual(queryActual, queryExpected) + }) + }) +}) Index: node_modules/pg-protocol/src/parser.ts =================================================================== --- node_modules/pg-protocol/src/parser.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) +++ node_modules/pg-protocol/src/parser.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) @@ -0,0 +1,414 @@ +import { TransformOptions } from 'stream' +import { + Mode, + bindComplete, + parseComplete, + closeComplete, + noData, + portalSuspended, + copyDone, + replicationStart, + emptyQuery, + ReadyForQueryMessage, + CommandCompleteMessage, + CopyDataMessage, + CopyResponse, + NotificationResponseMessage, + RowDescriptionMessage, + ParameterDescriptionMessage, + Field, + DataRowMessage, + ParameterStatusMessage, + BackendKeyDataMessage, + DatabaseError, + BackendMessage, + MessageName, + AuthenticationMD5Password, + NoticeMessage, +} from './messages' +import { BufferReader } from './buffer-reader' + +// every message is prefixed with a single byte +const CODE_LENGTH = 1 +// every message has an int32 length which includes itself but does +// NOT include the code in the length +const LEN_LENGTH = 4 + +const HEADER_LENGTH = CODE_LENGTH + LEN_LENGTH + +// A placeholder for a `BackendMessage`’s length value that will be set after construction. +const LATEINIT_LENGTH = -1 + +export type Packet = { + code: number + packet: Buffer +} + +const emptyBuffer = Buffer.allocUnsafe(0) + +type StreamOptions = TransformOptions & { + mode: Mode +} + +const enum MessageCodes { + DataRow = 0x44, // D + ParseComplete = 0x31, // 1 + BindComplete = 0x32, // 2 + CloseComplete = 0x33, // 3 + CommandComplete = 0x43, // C + ReadyForQuery = 0x5a, // Z + NoData = 0x6e, // n + NotificationResponse = 0x41, // A + AuthenticationResponse = 0x52, // R + ParameterStatus = 0x53, // S + BackendKeyData = 0x4b, // K + ErrorMessage = 0x45, // E + NoticeMessage = 0x4e, // N + RowDescriptionMessage = 0x54, // T + ParameterDescriptionMessage = 0x74, // t + PortalSuspended = 0x73, // s + ReplicationStart = 0x57, // W + EmptyQuery = 0x49, // I + CopyIn = 0x47, // G + CopyOut = 0x48, // H + CopyDone = 0x63, // c + CopyData = 0x64, // d +} + +export type MessageCallback = (msg: BackendMessage) => void + +export class Parser { + private buffer: Buffer = emptyBuffer + private bufferLength: number = 0 + private bufferOffset: number = 0 + private reader = new BufferReader() + private mode: Mode + + constructor(opts?: StreamOptions) { + if (opts?.mode === 'binary') { + throw new Error('Binary mode not supported yet') + } + this.mode = opts?.mode || 'text' + } + + public parse(buffer: Buffer, callback: MessageCallback) { + this.mergeBuffer(buffer) + const bufferFullLength = this.bufferOffset + this.bufferLength + let offset = this.bufferOffset + while (offset + HEADER_LENGTH <= bufferFullLength) { + // code is 1 byte long - it identifies the message type + const code = this.buffer[offset] + // length is 1 Uint32BE - it is the length of the message EXCLUDING the code + const length = this.buffer.readUInt32BE(offset + CODE_LENGTH) + const fullMessageLength = CODE_LENGTH + length + if (fullMessageLength + offset <= bufferFullLength) { + const message = this.handlePacket(offset + HEADER_LENGTH, code, length, this.buffer) + callback(message) + offset += fullMessageLength + } else { + break + } + } + if (offset === bufferFullLength) { + // No more use for the buffer + this.buffer = emptyBuffer + this.bufferLength = 0 + this.bufferOffset = 0 + } else { + // Adjust the cursors of remainingBuffer + this.bufferLength = bufferFullLength - offset + this.bufferOffset = offset + } + } + + private mergeBuffer(buffer: Buffer): void { + if (this.bufferLength > 0) { + const newLength = this.bufferLength + buffer.byteLength + const newFullLength = newLength + this.bufferOffset + if (newFullLength > this.buffer.byteLength) { + // We can't concat the new buffer with the remaining one + let newBuffer: Buffer + if (newLength <= this.buffer.byteLength && this.bufferOffset >= this.bufferLength) { + // We can move the relevant part to the beginning of the buffer instead of allocating a new buffer + newBuffer = this.buffer + } else { + // Allocate a new larger buffer + let newBufferLength = this.buffer.byteLength * 2 + while (newLength >= newBufferLength) { + newBufferLength *= 2 + } + newBuffer = Buffer.allocUnsafe(newBufferLength) + } + // Move the remaining buffer to the new one + this.buffer.copy(newBuffer, 0, this.bufferOffset, this.bufferOffset + this.bufferLength) + this.buffer = newBuffer + this.bufferOffset = 0 + } + // Concat the new buffer with the remaining one + buffer.copy(this.buffer, this.bufferOffset + this.bufferLength) + this.bufferLength = newLength + } else { + this.buffer = buffer + this.bufferOffset = 0 + this.bufferLength = buffer.byteLength + } + } + + private handlePacket(offset: number, code: number, length: number, bytes: Buffer): BackendMessage { + const { reader } = this + + // NOTE: This undesirably retains the buffer in `this.reader` if the `parse*Message` calls below throw. However, those should only throw in the case of a protocol error, which normally results in the reader being discarded. + reader.setBuffer(offset, bytes) + + let message: BackendMessage + + switch (code) { + case MessageCodes.BindComplete: + message = bindComplete + break + case MessageCodes.ParseComplete: + message = parseComplete + break + case MessageCodes.CloseComplete: + message = closeComplete + break + case MessageCodes.NoData: + message = noData + break + case MessageCodes.PortalSuspended: + message = portalSuspended + break + case MessageCodes.CopyDone: + message = copyDone + break + case MessageCodes.ReplicationStart: + message = replicationStart + break + case MessageCodes.EmptyQuery: + message = emptyQuery + break + case MessageCodes.DataRow: + message = parseDataRowMessage(reader) + break + case MessageCodes.CommandComplete: + message = parseCommandCompleteMessage(reader) + break + case MessageCodes.ReadyForQuery: + message = parseReadyForQueryMessage(reader) + break + case MessageCodes.NotificationResponse: + message = parseNotificationMessage(reader) + break + case MessageCodes.AuthenticationResponse: + message = parseAuthenticationResponse(reader, length) + break + case MessageCodes.ParameterStatus: + message = parseParameterStatusMessage(reader) + break + case MessageCodes.BackendKeyData: + message = parseBackendKeyData(reader) + break + case MessageCodes.ErrorMessage: + message = parseErrorMessage(reader, 'error') + break + case MessageCodes.NoticeMessage: + message = parseErrorMessage(reader, 'notice') + break + case MessageCodes.RowDescriptionMessage: + message = parseRowDescriptionMessage(reader) + break + case MessageCodes.ParameterDescriptionMessage: + message = parseParameterDescriptionMessage(reader) + break + case MessageCodes.CopyIn: + message = parseCopyInMessage(reader) + break + case MessageCodes.CopyOut: + message = parseCopyOutMessage(reader) + break + case MessageCodes.CopyData: + message = parseCopyData(reader, length) + break + default: + return new DatabaseError('received invalid response: ' + code.toString(16), length, 'error') + } + + reader.setBuffer(0, emptyBuffer) + + message.length = length + return message + } +} + +const parseReadyForQueryMessage = (reader: BufferReader) => { + const status = reader.string(1) + return new ReadyForQueryMessage(LATEINIT_LENGTH, status) +} + +const parseCommandCompleteMessage = (reader: BufferReader) => { + const text = reader.cstring() + return new CommandCompleteMessage(LATEINIT_LENGTH, text) +} + +const parseCopyData = (reader: BufferReader, length: number) => { + const chunk = reader.bytes(length - 4) + return new CopyDataMessage(LATEINIT_LENGTH, chunk) +} + +const parseCopyInMessage = (reader: BufferReader) => parseCopyMessage(reader, 'copyInResponse') + +const parseCopyOutMessage = (reader: BufferReader) => parseCopyMessage(reader, 'copyOutResponse') + +const parseCopyMessage = (reader: BufferReader, messageName: MessageName) => { + const isBinary = reader.byte() !== 0 + const columnCount = reader.int16() + const message = new CopyResponse(LATEINIT_LENGTH, messageName, isBinary, columnCount) + for (let i = 0; i < columnCount; i++) { + message.columnTypes[i] = reader.int16() + } + return message +} + +const parseNotificationMessage = (reader: BufferReader) => { + const processId = reader.int32() + const channel = reader.cstring() + const payload = reader.cstring() + return new NotificationResponseMessage(LATEINIT_LENGTH, processId, channel, payload) +} + +const parseRowDescriptionMessage = (reader: BufferReader) => { + const fieldCount = reader.int16() + const message = new RowDescriptionMessage(LATEINIT_LENGTH, fieldCount) + for (let i = 0; i < fieldCount; i++) { + message.fields[i] = parseField(reader) + } + return message +} + +const parseField = (reader: BufferReader) => { + const name = reader.cstring() + const tableID = reader.uint32() + const columnID = reader.int16() + const dataTypeID = reader.uint32() + const dataTypeSize = reader.int16() + const dataTypeModifier = reader.int32() + const mode = reader.int16() === 0 ? 'text' : 'binary' + return new Field(name, tableID, columnID, dataTypeID, dataTypeSize, dataTypeModifier, mode) +} + +const parseParameterDescriptionMessage = (reader: BufferReader) => { + const parameterCount = reader.int16() + const message = new ParameterDescriptionMessage(LATEINIT_LENGTH, parameterCount) + for (let i = 0; i < parameterCount; i++) { + // OIDs are unsigned, same as dataTypeID in parseField above + message.dataTypeIDs[i] = reader.uint32() + } + return message +} + +const parseDataRowMessage = (reader: BufferReader) => { + const fieldCount = reader.int16() + const fields: any[] = new Array(fieldCount) + for (let i = 0; i < fieldCount; i++) { + const len = reader.int32() + // a -1 for length means the value of the field is null + fields[i] = len === -1 ? null : reader.string(len) + } + return new DataRowMessage(LATEINIT_LENGTH, fields) +} + +const parseParameterStatusMessage = (reader: BufferReader) => { + const name = reader.cstring() + const value = reader.cstring() + return new ParameterStatusMessage(LATEINIT_LENGTH, name, value) +} + +const parseBackendKeyData = (reader: BufferReader) => { + const processID = reader.int32() + const secretKey = reader.int32() + return new BackendKeyDataMessage(LATEINIT_LENGTH, processID, secretKey) +} + +const parseAuthenticationResponse = (reader: BufferReader, length: number) => { + const code = reader.int32() + // TODO(bmc): maybe better types here + const message: BackendMessage & any = { + name: 'authenticationOk', + length, + } + + switch (code) { + case 0: // AuthenticationOk + break + case 3: // AuthenticationCleartextPassword + if (message.length === 8) { + message.name = 'authenticationCleartextPassword' + } + break + case 5: // AuthenticationMD5Password + if (message.length === 12) { + message.name = 'authenticationMD5Password' + const salt = reader.bytes(4) + return new AuthenticationMD5Password(LATEINIT_LENGTH, salt) + } + break + case 10: // AuthenticationSASL + { + message.name = 'authenticationSASL' + message.mechanisms = [] + let mechanism: string + do { + mechanism = reader.cstring() + if (mechanism) { + message.mechanisms.push(mechanism) + } + } while (mechanism) + } + break + case 11: // AuthenticationSASLContinue + message.name = 'authenticationSASLContinue' + message.data = reader.string(length - 8) + break + case 12: // AuthenticationSASLFinal + message.name = 'authenticationSASLFinal' + message.data = reader.string(length - 8) + break + default: + throw new Error('Unknown authenticationOk message type ' + code) + } + return message +} + +const parseErrorMessage = (reader: BufferReader, name: MessageName) => { + const fields: Record = {} + let fieldType = reader.string(1) + while (fieldType !== '\0') { + fields[fieldType] = reader.cstring() + fieldType = reader.string(1) + } + + const messageValue = fields.M + + const message = + name === 'notice' + ? new NoticeMessage(LATEINIT_LENGTH, messageValue) + : new DatabaseError(messageValue, LATEINIT_LENGTH, name) + + message.severity = fields.S + message.code = fields.C + message.detail = fields.D + message.hint = fields.H + message.position = fields.P + message.internalPosition = fields.p + message.internalQuery = fields.q + message.where = fields.W + message.schema = fields.s + message.table = fields.t + message.column = fields.c + message.dataType = fields.d + message.constraint = fields.n + message.file = fields.F + message.line = fields.L + message.routine = fields.R + return message +} Index: node_modules/pg-protocol/src/serializer.ts =================================================================== --- node_modules/pg-protocol/src/serializer.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) +++ node_modules/pg-protocol/src/serializer.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) @@ -0,0 +1,280 @@ +import { Writer } from './buffer-writer' + +const enum code { + startup = 0x70, + query = 0x51, + parse = 0x50, + bind = 0x42, + execute = 0x45, + flush = 0x48, + sync = 0x53, + end = 0x58, + close = 0x43, + describe = 0x44, + copyFromChunk = 0x64, + copyDone = 0x63, + copyFail = 0x66, +} + +const writer = new Writer() + +const startup = (opts: Record): Buffer => { + // protocol version + writer.addInt16(3).addInt16(0) + for (const key of Object.keys(opts)) { + writer.addCString(key).addCString(opts[key]) + } + + writer.addCString('client_encoding').addCString('UTF8') + + const bodyBuffer = writer.addCString('').flush() + // this message is sent without a code + + const length = bodyBuffer.length + 4 + + return new Writer().addInt32(length).add(bodyBuffer).flush() +} + +const requestSsl = (): Buffer => { + const response = Buffer.allocUnsafe(8) + response.writeInt32BE(8, 0) + response.writeInt32BE(80877103, 4) + return response +} + +const password = (password: string): Buffer => { + return writer.addCString(password).flush(code.startup) +} + +const sendSASLInitialResponseMessage = function (mechanism: string, initialResponse: string): Buffer { + // 0x70 = 'p' + writer.addCString(mechanism).addInt32PrefixedString(initialResponse) + + return writer.flush(code.startup) +} + +const sendSCRAMClientFinalMessage = function (additionalData: string): Buffer { + return writer.addString(additionalData).flush(code.startup) +} + +const query = (text: string): Buffer => { + return writer.addCString(text).flush(code.query) +} + +type ParseOpts = { + name?: string + types?: number[] + text: string +} + +const emptyArray: any[] = [] + +const parse = (query: ParseOpts): Buffer => { + // expect something like this: + // { name: 'queryName', + // text: 'select * from blah', + // types: ['int8', 'bool'] } + + // normalize missing query names to allow for null + const name = query.name || '' + if (name.length > 63) { + console.error('Warning! Postgres only supports 63 characters for query names.') + console.error('You supplied %s (%s)', name, name.length) + console.error('This can cause conflicts and silent errors executing queries') + } + + const types = query.types || emptyArray + + const len = types.length + + const buffer = writer + .addCString(name) // name of query + .addCString(query.text) // actual query text + .addInt16(len) + + for (let i = 0; i < len; i++) { + buffer.addInt32(types[i]) + } + + return writer.flush(code.parse) +} + +type ValueMapper = (param: any, index: number) => any + +type BindOpts = { + portal?: string + binary?: boolean + statement?: string + values?: any[] + // optional map from JS value to postgres value per parameter + valueMapper?: ValueMapper +} + +const paramWriter = new Writer() + +// make this a const enum so typescript will inline the value +const enum ParamType { + STRING = 0, + BINARY = 1, +} + +const writeValues = function (values: any[], valueMapper?: ValueMapper): void { + for (let i = 0; i < values.length; i++) { + const mappedVal = valueMapper ? valueMapper(values[i], i) : values[i] + if (mappedVal == null) { + // add the param type (string) to the writer + writer.addInt16(ParamType.STRING) + // write -1 to the param writer to indicate null + paramWriter.addInt32(-1) + } else if (mappedVal instanceof Buffer) { + // add the param type (binary) to the writer + writer.addInt16(ParamType.BINARY) + // add the buffer to the param writer + paramWriter.addInt32(mappedVal.length) + paramWriter.add(mappedVal) + } else { + // add the param type (string) to the writer + writer.addInt16(ParamType.STRING) + // length prefix + UTF-8 bytes in one pass (Buffer.byteLength computed once) + paramWriter.addInt32PrefixedString(mappedVal) + } + } +} + +const bind = (config: BindOpts = {}): Buffer => { + // normalize config + const portal = config.portal || '' + const statement = config.statement || '' + const binary = config.binary || false + const values = config.values || emptyArray + const len = values.length + + writer.addCString(portal).addCString(statement) + writer.addInt16(len) + + try { + writeValues(values, config.valueMapper) + } catch (err) { + writer.clear() + paramWriter.clear() + throw err + } + + writer.addInt16(len) + writer.add(paramWriter.flush()) + + // all results use the same format code + writer.addInt16(1) + // format code + writer.addInt16(binary ? ParamType.BINARY : ParamType.STRING) + return writer.flush(code.bind) +} + +type ExecOpts = { + portal?: string + rows?: number +} + +const emptyExecute = Buffer.from([code.execute, 0x00, 0x00, 0x00, 0x09, 0x00, 0x00, 0x00, 0x00, 0x00]) + +const execute = (config?: ExecOpts): Buffer => { + // this is the happy path for most queries + if (!config || (!config.portal && !config.rows)) { + return emptyExecute + } + + const portal = config.portal || '' + const rows = config.rows || 0 + + const portalLength = Buffer.byteLength(portal) + const len = 4 + portalLength + 1 + 4 + // one extra bit for code + const buff = Buffer.allocUnsafe(1 + len) + buff[0] = code.execute + buff.writeInt32BE(len, 1) + buff.write(portal, 5, 'utf-8') + buff[portalLength + 5] = 0 // null terminate portal cString + buff.writeUInt32BE(rows, buff.length - 4) + return buff +} + +const cancel = (processID: number, secretKey: number): Buffer => { + const buffer = Buffer.allocUnsafe(16) + buffer.writeInt32BE(16, 0) + buffer.writeInt16BE(1234, 4) + buffer.writeInt16BE(5678, 6) + buffer.writeInt32BE(processID, 8) + buffer.writeInt32BE(secretKey, 12) + return buffer +} + +type PortalOpts = { + type: 'S' | 'P' + name?: string +} + +const cstringMessage = (code: code, string: string): Buffer => { + const stringLen = Buffer.byteLength(string) + const len = 4 + stringLen + 1 + // one extra bit for code + const buffer = Buffer.allocUnsafe(1 + len) + buffer[0] = code + buffer.writeInt32BE(len, 1) + buffer.write(string, 5, 'utf-8') + buffer[len] = 0 // null terminate cString + return buffer +} + +const emptyDescribePortal = writer.addCString('P').flush(code.describe) +const emptyDescribeStatement = writer.addCString('S').flush(code.describe) + +const describe = (msg: PortalOpts): Buffer => { + return msg.name + ? cstringMessage(code.describe, `${msg.type}${msg.name || ''}`) + : msg.type === 'P' + ? emptyDescribePortal + : emptyDescribeStatement +} + +const close = (msg: PortalOpts): Buffer => { + const text = `${msg.type}${msg.name || ''}` + return cstringMessage(code.close, text) +} + +const copyData = (chunk: Buffer): Buffer => { + return writer.add(chunk).flush(code.copyFromChunk) +} + +const copyFail = (message: string): Buffer => { + return cstringMessage(code.copyFail, message) +} + +const codeOnlyBuffer = (code: code): Buffer => Buffer.from([code, 0x00, 0x00, 0x00, 0x04]) + +const flushBuffer = codeOnlyBuffer(code.flush) +const syncBuffer = codeOnlyBuffer(code.sync) +const endBuffer = codeOnlyBuffer(code.end) +const copyDoneBuffer = codeOnlyBuffer(code.copyDone) + +const serialize = { + startup, + password, + requestSsl, + sendSASLInitialResponseMessage, + sendSCRAMClientFinalMessage, + query, + parse, + bind, + execute, + describe, + close, + flush: () => flushBuffer, + sync: () => syncBuffer, + end: () => endBuffer, + copyData, + copyDone: () => copyDoneBuffer, + copyFail, + cancel, +} + +export { serialize } Index: node_modules/pg-protocol/src/testing/buffer-list.ts =================================================================== --- node_modules/pg-protocol/src/testing/buffer-list.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) +++ node_modules/pg-protocol/src/testing/buffer-list.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) @@ -0,0 +1,67 @@ +export default class BufferList { + constructor(public buffers: Buffer[] = []) {} + + public add(buffer: Buffer, front?: boolean) { + this.buffers[front ? 'unshift' : 'push'](buffer) + return this + } + + public addInt16(val: number, front?: boolean) { + return this.add(Buffer.from([val >>> 8, val >>> 0]), front) + } + + public getByteLength() { + return this.buffers.reduce(function (previous, current) { + return previous + current.length + }, 0) + } + + public addInt32(val: number, first?: boolean) { + return this.add( + Buffer.from([(val >>> 24) & 0xff, (val >>> 16) & 0xff, (val >>> 8) & 0xff, (val >>> 0) & 0xff]), + first + ) + } + + public addCString(val: string, front?: boolean) { + const len = Buffer.byteLength(val) + const buffer = Buffer.alloc(len + 1) + buffer.write(val) + buffer[len] = 0 + return this.add(buffer, front) + } + + public addString(val: string, front?: boolean) { + const len = Buffer.byteLength(val) + const buffer = Buffer.alloc(len) + buffer.write(val) + return this.add(buffer, front) + } + + public addChar(char: string, first?: boolean) { + return this.add(Buffer.from(char, 'utf8'), first) + } + + public addByte(byte: number) { + return this.add(Buffer.from([byte])) + } + + public join(appendLength?: boolean, char?: string): Buffer { + let length = this.getByteLength() + if (appendLength) { + this.addInt32(length + 4, true) + return this.join(false, char) + } + if (char) { + this.addChar(char, true) + length++ + } + const result = Buffer.alloc(length) + let index = 0 + this.buffers.forEach(function (buffer) { + buffer.copy(result, index, 0) + index += buffer.length + }) + return result + } +} Index: node_modules/pg-protocol/src/testing/test-buffers.ts =================================================================== --- node_modules/pg-protocol/src/testing/test-buffers.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) +++ node_modules/pg-protocol/src/testing/test-buffers.ts (revision 62b296460f935ff5cc47a75f6567f97df5a244a7) @@ -0,0 +1,166 @@ +// https://www.postgresql.org/docs/current/protocol-message-formats.html +import BufferList from './buffer-list' + +const buffers = { + readyForQuery: function () { + return new BufferList().add(Buffer.from('I')).join(true, 'Z') + }, + + authenticationOk: function () { + return new BufferList().addInt32(0).join(true, 'R') + }, + + authenticationCleartextPassword: function () { + return new BufferList().addInt32(3).join(true, 'R') + }, + + authenticationMD5Password: function () { + return new BufferList() + .addInt32(5) + .add(Buffer.from([1, 2, 3, 4])) + .join(true, 'R') + }, + + authenticationSASL: function () { + return new BufferList().addInt32(10).addCString('SCRAM-SHA-256').addCString('').join(true, 'R') + }, + + authenticationSASLContinue: function () { + return new BufferList().addInt32(11).addString('data').join(true, 'R') + }, + + authenticationSASLFinal: function () { + return new BufferList().addInt32(12).addString('data').join(true, 'R') + }, + + parameterStatus: function (name: string, value: string) { + return new BufferList().addCString(name).addCString(value).join(true, 'S') + }, + + backendKeyData: function (processID: number, secretKey: number) { + return new BufferList().addInt32(processID).addInt32(secretKey).join(true, 'K') + }, + + commandComplete: function (string: string) { + return new BufferList().addCString(string).join(true, 'C') + }, + + rowDescription: function (fields: any[]) { + fields = fields || [] + const buf = new BufferList() + buf.addInt16(fields.length) + fields.forEach(function (field) { + buf + .addCString(field.name) + .addInt32(field.tableID || 0) + .addInt16(field.attributeNumber || 0) + .addInt32(field.dataTypeID || 0) + .addInt16(field.dataTypeSize || 0) + .addInt32(field.typeModifier || 0) + .addInt16(field.formatCode || 0) + }) + return buf.join(true, 'T') + }, + + parameterDescription: function (dataTypeIDs: number[]) { + dataTypeIDs = dataTypeIDs || [] + const buf = new BufferList() + buf.addInt16(dataTypeIDs.length) + dataTypeIDs.forEach(function (dataTypeID) { + buf.addInt32(dataTypeID) + }) + return buf.join(true, 't') + }, + + dataRow: function (columns: any[]) { + columns = columns || [] + const buf = new BufferList() + buf.addInt16(columns.length) + columns.forEach(function (col) { + if (col == null) { + buf.addInt32(-1) + } else { + const strBuf = Buffer.from(col, 'utf8') + buf.addInt32(strBuf.length) + buf.add(strBuf) + } + }) + return buf.join(true, 'D') + }, + + error: function (fields: any) { + return buffers.errorOrNotice(fields).join(true, 'E') + }, + + notice: function (fields: any) { + return buffers.errorOrNotice(fields).join(true, 'N') + }, + + errorOrNotice: function (fields: any) { + fields = fields || [] + const buf = new BufferList() + fields.forEach(function (field: any) { + buf.addChar(field.type) + buf.addCString(field.value) + }) + return buf.add(Buffer.from([0])) // terminator + }, + + parseComplete: function () { + return new BufferList().join(true, '1') + }, + + bindComplete: function () { + return new BufferList().join(true, '2') + }, + + notification: function (id: number, channel: string, payload: string) { + return new BufferList().addInt32(id).addCString(channel).addCString(payload).join(true, 'A') + }, + + emptyQuery: function () { + return new BufferList().join(true, 'I') + }, + + portalSuspended: function () { + return new BufferList().join(true, 's') + }, + + closeComplete: function () { + return new BufferList().join(true, '3') + }, + + copyIn: function (cols: number) { + const list = new BufferList() + // text mode + .addByte(0) + // column count + .addInt16(cols) + for (let i = 0; i < cols; i++) { + list.addInt16(i) + } + return list.join(true, 'G') + }, + + copyOut: function (cols: number) { + const list = new BufferList() + // text mode + .addByte(0) + // column count + .addInt16(cols) + for (let i = 0; i < cols; i++) { + list.addInt16(i) + } + return list.join(true, 'H') + }, + + copyData: function (bytes: Buffer) { + return new BufferList().add(bytes).join(true, 'd') + }, + + copyDone: function () { + return new BufferList().join(true, 'c') + }, +} + +export default buffers