Build cjs + deno

Author: porsager

cjs/src/connection.js

@@ -3,7 +3,7 @@ const tls = require('tls')
 const crypto = require('crypto')
 const Stream = require('stream')
 
-const { Identifier, Builder, handleValue, arrayParser, arraySerializer } = require('./types.js')
+const { stringify, handleValue, arrayParser, arraySerializer } = require('./types.js')
 const { Errors } = require('./errors.js')
 const Result = require('./result.js')
 const Queue = require('./queue.js')
@@ -218,9 +218,9 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
     const parameters = []
         , types = []
 
-    const string = stringify(q, q.strings[0], q.args[0], parameters, types)
+    const string = stringify(q, q.strings[0], q.args[0], parameters, types, options)
 
-    !q.tagged && q.args.forEach(x => handleValue(x, parameters, types))
+    !q.tagged && q.args.forEach(x => handleValue(x, parameters, types, options))
 
     q.prepare = options.prepare && ('prepare' in q.options ? q.options.prepare : true)
     q.string = string
@@ -236,25 +236,6 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
     typeof options.debug === 'function' && options.debug(id, string, parameters, types)
   }
 
-  function stringify(q, string, value, parameters, types) {
-    for (let i = 1; i < q.strings.length; i++) {
-      string += (
-        value instanceof Query ? fragment(string, value, parameters, types) :
-        value instanceof Identifier ? value.value :
-        value instanceof Builder ? value.build(string, parameters, types, options.transform) :
-        handleValue(value, parameters, types)
-      ) + q.strings[i]
-      value = q.args[i]
-    }
-
-    return string
-  }
-
-  function fragment(string, q, parameters, types) {
-    q.fragment = true
-    return stringify(q, q.strings[0], q.args[0], parameters, types)
-  }
-
   function write(x, fn) {
     chunk = chunk ? Buffer.concat([chunk, x]) : Buffer.from(x)
     if (fn || chunk.length >= 1024)
@@ -497,7 +478,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
 
       value = length === -1
         ? null
-        : query.isRaw
+        : query.isRaw === true
           ? x.slice(index, index += length)
           : column.parser === undefined
             ? x.toString('utf8', index, index += length)
@@ -506,7 +487,9 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
               : column.parser(x.toString('utf8', index, index += length))
 
       query.isRaw
-        ? (row[i] = value)
+        ? (row[i] = query.isRaw === true
+          ? value
+          : transform.value.from ? transform.value.from(value) : value)
         : (row[column.name] = transform.value.from ? transform.value.from(value) : value)
     }
 

cjs/src/index.js

@@ -42,7 +42,7 @@ module.exports = Postgres
 
 function Postgres(a, b) {
   const options = parseOptions(a, b)
-      , subscribe = Subscribe(Postgres, { ...options })
+      , subscribe = options.no_subscribe || Subscribe(Postgres, { ...options })
 
   let ending = false
 
@@ -71,6 +71,7 @@ function Postgres(a, b) {
     listen,
     notify,
     begin,
+    close,
     end
   })
 
@@ -325,6 +326,10 @@ function Postgres(a, b) {
     ]).then(() => clearTimeout(timer))
   }
 
+  async function close() {
+    await Promise.all(connections.map(c => c.end()))
+  }
+
   async function destroy(resolve) {
     await Promise.all(connections.map(c => c.terminate()))
     while (queries.length)
@@ -397,7 +402,7 @@ function parseOptions(a, b) {
     onnotify        : o.onnotify,
     onclose         : o.onclose,
     onparameter     : o.onparameter,
-    transform       : parseTransform(o.transform || {}),
+    transform       : parseTransform(o.transform || { undefined: undefined }),
     connection      : Object.assign({ application_name: 'postgres.js' }, o.connection),
     target_session_attrs: tsa(o, url, env),
     debug           : o.debug,
@@ -429,6 +434,7 @@ function max_lifetime() {
 
 function parseTransform(x) {
   return {
+    undefined: x.undefined,
     column: {
       from: typeof x.column === 'function' ? x.column : x.column && x.column.from,
       to: x.column && x.column.to

cjs/src/query.js

@@ -127,6 +127,11 @@ const Query = module.exports.Query = class Query extends Promise {
     return this
   }
 
+  values() {
+    this.isRaw = 'values'
+    return this
+  }
+
   async handle() {
     !this.executed && (this.executed = true) && await 1 && this.handler(this)
   }

cjs/src/subscribe.js

@@ -1,59 +1,90 @@
+const noop = () => { /* noop */ }
+
 module.exports = Subscribe;function Subscribe(postgres, options) {
-  const listeners = new Map()
+  const subscribers = new Map()
+      , slot = 'postgresjs_' + Math.random().toString(36).slice(2)
+      , state = {}
 
   let connection
-
-  return async function subscribe(event, fn) {
-    event = parseEvent(event)
-
-    options.max = 1
-    options.onclose = onclose
-    options.fetch_types = false
-    options.connection = {
+    , stream
+    , ended = false
+
+  const sql = subscribe.sql = postgres({
+    ...options,
+    max: 1,
+    fetch_types: false,
+    idle_timeout: null,
+    max_lifetime: null,
+    connection: {
       ...options.connection,
       replication: 'database'
-    }
+    },
+    onclose: async function() {
+      if (ended)
+        return
+      stream = null
+      state.pid = state.secret = undefined
+      !ended && connected(await init(sql, slot, options.publications))
+      subscribers.forEach(event => event.forEach(({ onsubscribe }) => onsubscribe()))
+    },
+    no_subscribe: true
+  })
 
-    let stream
-      , ended = false
+  const end = sql.end
+      , close = sql.close
 
-    const sql = postgres(options)
-        , slot = 'postgresjs_' + Math.random().toString(36).slice(2)
-        , end = sql.end
+  sql.end = async() => {
+    ended = true
+    stream && (await new Promise(r => (stream.once('end', r), stream.end())))
+    return end()
+  }
 
-    sql.end = async() => {
-      ended = true
-      stream && (await new Promise(r => (stream.once('end', r), stream.end())))
-      return end()
-    }
+  sql.close = async() => {
+    stream && (await new Promise(r => (stream.once('end', r), stream.end())))
+    return close()
+  }
+
+  return subscribe
 
-    !connection && (subscribe.sql = sql, connection = init(sql, slot, options.publications))
+  async function subscribe(event, fn, onsubscribe = noop) {
+    event = parseEvent(event)
 
-    const fns = listeners.has(event)
-      ? listeners.get(event).add(fn)
-      : listeners.set(event, new Set([fn])).get(event)
+    if (!connection)
+      connection = init(sql, slot, options.publications)
+
+    const subscriber = { fn, onsubscribe }
+    const fns = subscribers.has(event)
+      ? subscribers.get(event).add(subscriber)
+      : subscribers.set(event, new Set([subscriber])).get(event)
 
     const unsubscribe = () => {
-      fns.delete(fn)
-      fns.size === 0 && listeners.delete(event)
+      fns.delete(subscriber)
+      fns.size === 0 && subscribers.delete(event)
     }
 
-    return connection.then(x => (stream = x, { unsubscribe }))
+    return connection.then(x => {
+      connected(x)
+      onsubscribe()
+      return { unsubscribe, state, sql }
+    })
+  }
 
-    async function onclose() {
-      stream = null
-      !ended && (stream = await init(sql, slot, options.publications))
-    }
+  function connected(x) {
+    stream = x.stream
+    state.pid = x.state.pid
+    state.secret = x.state.secret
   }
 
   async function init(sql, slot, publications) {
     if (!publications)
       throw new Error('Missing publication names')
 
-    const [x] = await sql.unsafe(
+    const xs = await sql.unsafe(
       `CREATE_REPLICATION_SLOT ${ slot } TEMPORARY LOGICAL pgoutput NOEXPORT_SNAPSHOT`
     )
 
+    const [x] = xs
+
     const stream = await sql.unsafe(
       `START_REPLICATION SLOT ${ slot } LOGICAL ${
         x.consistent_point
@@ -65,12 +96,10 @@ module.exports = Subscribe;function Subscribe(postgres, options) {
     }
 
     stream.on('data', data)
-    stream.on('error', (error) => {
-      console.error('Logical Replication Error - Reconnecting', error) // eslint-disable-line
-      sql.end()
-    })
+    stream.on('error', sql.close)
+    stream.on('close', sql.close)
 
-    return stream
+    return { stream, state: xs.state }
 
     function data(x) {
       if (x[0] === 0x77)
@@ -99,7 +128,7 @@ module.exports = Subscribe;function Subscribe(postgres, options) {
   }
 
   function call(x, a, b) {
-    listeners.has(x) && listeners.get(x).forEach(fn => fn(a, b, x))
+    subscribers.has(x) && subscribers.get(x).forEach(({ fn }) => fn(a, b, x))
   }
 }
 

cjs/src/types.js

@@ -64,19 +64,25 @@ const Builder = module.exports.Builder = class Builder extends NotTagged {
     this.rest = rest
   }
 
-  build(before, parameters, types, transform) {
+  build(before, parameters, types, options) {
     const keyword = builders.map(([x, fn]) => ({ fn, i: before.search(x) })).sort((a, b) => a.i - b.i).pop()
     if (keyword.i === -1)
       throw new Error('Could not infer helper mode')
 
-    return keyword.fn(this.first, this.rest, parameters, types, transform)
+    return keyword.fn(this.first, this.rest, parameters, types, options)
   }
 }
 
-module.exports.handleValue = handleValue;function handleValue(x, parameters, types) {
-  const value = x instanceof Parameter ? x.value : x
-  if (value === undefined)
-    throw Errors.generic('UNDEFINED_VALUE', 'Undefined values are not allowed')
+module.exports.handleValue = handleValue;function handleValue(x, parameters, types, options) {
+  let value = x instanceof Parameter ? x.value : x
+  if (value === undefined) {
+    x instanceof Parameter
+      ? x.value = options.transform.undefined
+      : value = x = options.transform.undefined
+
+    if (value === undefined)
+      throw Errors.generic('UNDEFINED_VALUE', 'Undefined values are not allowed')
+  }
 
   return '$' + (types.push(
     x instanceof Parameter
@@ -90,40 +96,59 @@ module.exports.handleValue = handleValue;function handleValue(x, parameters, typ
 
 const defaultHandlers = typeHandlers(types)
 
-function valuesBuilder(first, parameters, types, transform, columns) {
+module.exports.stringify = stringify;function stringify(q, string, value, parameters, types, options) { // eslint-disable-line
+  for (let i = 1; i < q.strings.length; i++) {
+    string += (
+      value instanceof Query ? fragment(value, parameters, types) :
+      value instanceof Identifier ? value.value :
+      value instanceof Builder ? value.build(string, parameters, types, options) :
+      handleValue(value, parameters, types, options)
+    ) + q.strings[i]
+    value = q.args[i]
+  }
+
+  return string
+}
+
+function fragment(q, parameters, types) {
+  q.fragment = true
+  return stringify(q, q.strings[0], q.args[0], parameters, types)
+}
+
+function valuesBuilder(first, parameters, types, columns, options) {
   let value
   return first.map(row =>
     '(' + columns.map(column => {
       value = row[column]
       return (
-        value instanceof Query ? value.strings[0] :
+        value instanceof Query ? fragment(value, parameters, types) :
         value instanceof Identifier ? value.value :
-        handleValue(value, parameters, types)
+        handleValue(value, parameters, types, options)
       )
     }).join(',') + ')'
   ).join(',')
 }
 
-function values(first, rest, parameters, types, transform) {
+function values(first, rest, parameters, types, options) {
   const multi = Array.isArray(first[0])
   const columns = rest.length ? rest.flat() : Object.keys(multi ? first[0] : first)
-  return valuesBuilder(multi ? first : [first], parameters, types, transform, columns)
+  return valuesBuilder(multi ? first : [first], parameters, types, columns, options)
 }
 
-function select(first, rest, parameters, types, transform) {
+function select(first, rest, parameters, types, options) {
   typeof first === 'string' && (first = [first].concat(rest))
   if (Array.isArray(first))
-    return first.map(x => escapeIdentifier(transform.column.to ? transform.column.to(x) : x)).join(',')
+    return first.map(x => escapeIdentifier(options.transform.column.to ? options.transform.column.to(x) : x)).join(',')
 
   let value
   const columns = rest.length ? rest.flat() : Object.keys(first)
   return columns.map(x => {
     value = first[x]
     return (
-      value instanceof Query ? value.strings[0] :
+      value instanceof Query ? fragment(value, parameters, types) :
       value instanceof Identifier ? value.value :
-      handleValue(value, parameters, types)
-    ) + ' as ' + escapeIdentifier(transform.column.to ? transform.column.to(x) : x)
+      handleValue(value, parameters, types, options)
+    ) + ' as ' + escapeIdentifier(options.transform.column.to ? options.transform.column.to(x) : x)
   }).join(',')
 }
 
@@ -133,19 +158,19 @@ const builders = Object.entries({
   select,
   returning: select,
 
-  update(first, rest, parameters, types, transform) {
+  update(first, rest, parameters, types, options) {
     return (rest.length ? rest.flat() : Object.keys(first)).map(x =>
-      escapeIdentifier(transform.column.to ? transform.column.to(x) : x) +
-      '=' + handleValue(first[x], parameters, types)
+      escapeIdentifier(options.transform.column.to ? options.transform.column.to(x) : x) +
+      '=' + handleValue(first[x], parameters, types, options)
     )
   },
 
-  insert(first, rest, parameters, types, transform) {
+  insert(first, rest, parameters, types, options) {
     const columns = rest.length ? rest.flat() : Object.keys(Array.isArray(first) ? first[0] : first)
     return '(' + columns.map(x =>
-      escapeIdentifier(transform.column.to ? transform.column.to(x) : x)
+      escapeIdentifier(options.transform.column.to ? options.transform.column.to(x) : x)
     ).join(',') + ')values' +
-    valuesBuilder(Array.isArray(first) ? first : [first], parameters, types, transform, columns)
+    valuesBuilder(Array.isArray(first) ? first : [first], parameters, types, columns, options)
   }
 }).map(([x, fn]) => ([new RegExp('(^|[\\s(])' + x + '($|[\\s(])', 'i'), fn]))