Support for Cloudflare Workers

.eslintrc.json

@@ -214,7 +214,7 @@
     ],
     "max-len": [
       2,
-      120
+      150
     ],
     "max-nested-callbacks": [
       2,

.github/workflows/test.yml

@@ -35,6 +35,7 @@ jobs:
           sudo apt-get -y install "postgresql-${{ matrix.postgres }}"
           sudo cp ./tests/pg_hba.conf /etc/postgresql/${{ matrix.postgres }}/main/pg_hba.conf
           sudo sed -i 's/.*wal_level.*/wal_level = logical/' /etc/postgresql/${{ matrix.postgres }}/main/postgresql.conf
+          sudo sed -i 's/.*max_prepared_transactions.*/max_prepared_transactions = 100/' /etc/postgresql/${{ matrix.postgres }}/main/postgresql.conf
           sudo sed -i 's/.*ssl = .*/ssl = on/' /etc/postgresql/${{ matrix.postgres }}/main/postgresql.conf
           openssl req -new -x509 -nodes -days 365 -text -subj "/CN=localhost" -extensions v3_req -config <(cat /etc/ssl/openssl.cnf <(printf "\n[v3_req]\nbasicConstraints=critical,CA:TRUE\nkeyUsage=nonRepudiation,digitalSignature,keyEncipherment\nsubjectAltName=DNS:localhost")) -keyout server.key -out server.crt
           sudo cp server.key /etc/postgresql/${{ matrix.postgres }}/main/server.key

README.md

@@ -455,9 +455,14 @@ const result = await sql.file('query.sql', ['Murray', 68])
 ```
 
 ### Multiple statements in one query
-#### `await sql`select 1;select 2`.simple()
+#### ```await sql``.simple()```
 
-The postgres wire protocol supports "simple" and "extended" queries. "simple" queries supports multiple statements, but does not support any dynamic parameters. "extended" queries support parameters but only one statement. To use "simple" queries you can use sql``.simple(). That will create it as a simple query.
+The postgres wire protocol supports ["simple"](https://www.postgresql.org/docs/current/protocol-flow.html#id-1.10.6.7.4) and ["extended"](https://www.postgresql.org/docs/current/protocol-flow.html#PROTOCOL-FLOW-EXT-QUERY) queries. "simple" queries supports multiple statements, but does not support any dynamic parameters. "extended" queries support parameters but only one statement. To use "simple" queries you can use 
+```sql``.simple()```. That will create it as a simple query.
+
+```js
+await sql`select 1; select 2;`.simple()
+```
 
 ### Copy to/from as Streams
 
@@ -632,6 +637,26 @@ sql.begin('read write', async sql => {
 })
 ```
 
+
+#### PREPARE `await sql.prepare([name]) -> fn()`
+
+Indicates that the transactions should be prepared using the `PREPARED TRANASCTION [NAME]` statement
+instead of being committed.
+
+```js
+sql.begin('read write', async sql => {
+  const [user] = await sql`
+    insert into users (
+      name
+    ) values (
+      'Murray'
+    )
+  `
+
+  await sql.prepare('tx1')
+})
+```
+
 Do note that you can often achieve the same result using [`WITH` queries (Common Table Expressions)](https://www.postgresql.org/docs/current/queries-with.html) instead of using transactions.
 
 ## Data Transformation

cf/polyfills.js

@@ -0,0 +1,218 @@
+import { EventEmitter } from 'node:events'
+import { Buffer } from 'node:buffer'
+
+const Crypto = globalThis.crypto
+
+let ids = 1
+const tasks = new Set()
+
+const v4Seg = '(?:[0-9]|[1-9][0-9]|1[0-9][0-9]|2[0-4][0-9]|25[0-5])'
+const v4Str = `(${v4Seg}[.]){3}${v4Seg}`
+const IPv4Reg = new RegExp(`^${v4Str}$`)
+
+const v6Seg = '(?:[0-9a-fA-F]{1,4})'
+const IPv6Reg = new RegExp(
+  '^(' +
+  `(?:${v6Seg}:){7}(?:${v6Seg}|:)|` +
+  `(?:${v6Seg}:){6}(?:${v4Str}|:${v6Seg}|:)|` +
+  `(?:${v6Seg}:){5}(?::${v4Str}|(:${v6Seg}){1,2}|:)|` +
+  `(?:${v6Seg}:){4}(?:(:${v6Seg}){0,1}:${v4Str}|(:${v6Seg}){1,3}|:)|` +
+  `(?:${v6Seg}:){3}(?:(:${v6Seg}){0,2}:${v4Str}|(:${v6Seg}){1,4}|:)|` +
+  `(?:${v6Seg}:){2}(?:(:${v6Seg}){0,3}:${v4Str}|(:${v6Seg}){1,5}|:)|` +
+  `(?:${v6Seg}:){1}(?:(:${v6Seg}){0,4}:${v4Str}|(:${v6Seg}){1,6}|:)|` +
+  `(?::((?::${v6Seg}){0,5}:${v4Str}|(?::${v6Seg}){1,7}|:))` +
+  ')(%[0-9a-zA-Z-.:]{1,})?$'
+)
+
+const textEncoder = new TextEncoder()
+export const crypto = {
+  randomBytes: l => Crypto.getRandomValues(Buffer.alloc(l)),
+  pbkdf2Sync: async(password, salt, iterations, keylen) =>
+    Crypto.subtle.deriveBits(
+      {
+        name: 'PBKDF2',
+        hash: 'SHA-256',
+        salt,
+        iterations
+      },
+      await Crypto.subtle.importKey(
+        'raw',
+        textEncoder.encode(password),
+        'PBKDF2',
+        false,
+        ['deriveBits']
+      ),
+      keylen * 8,
+      ['deriveBits']
+    ),
+  createHash: type => ({
+    update: x => ({
+      digest: () => {
+        if (type !== 'sha256')
+          throw Error('createHash only supports sha256 on cloudflare.')
+        if (!(x instanceof Uint8Array))
+          x = textEncoder.encode(x)
+        return Crypto.subtle.digest('SHA-256', x)
+      }
+    })
+  }),
+  createHmac: (type, key) => ({
+    update: x => ({
+      digest: async() =>
+        Buffer.from(
+          await Crypto.subtle.sign(
+            'HMAC',
+            await Crypto.subtle.importKey('raw', key, { name: 'HMAC', hash: 'SHA-256' }, false, ['sign']),
+            textEncoder.encode(x)
+          )
+        )
+    })
+  })
+}
+
+export const process = {
+  env: {}
+}
+
+export const os = {
+  userInfo() {
+    return { username: 'postgres' }
+  }
+}
+
+export const fs = {
+  readFile() {
+    throw new Error('Reading files not supported on CloudFlare')
+  }
+}
+
+export const net = {
+  isIP: (x) => RegExp.prototype.test.call(IPv4Reg, x) ? 4 : RegExp.prototype.test.call(IPv6Reg, x) ? 6 : 0,
+  Socket
+}
+
+export { setImmediate, clearImmediate }
+
+export const tls = {
+  connect({ socket: tcp, servername }) {
+    tcp.writer.releaseLock()
+    tcp.reader.releaseLock()
+    tcp.readyState = 'upgrading'
+    tcp.raw = tcp.raw.startTls({ servername })
+    tcp.raw.closed.then(
+      () => tcp.emit('close'),
+      (e) => tcp.emit('error', e)
+    )
+    tcp.writer = tcp.raw.writable.getWriter()
+    tcp.reader = tcp.raw.readable.getReader()
+
+    tcp.writer.ready.then(() => {
+      tcp.read()
+      tcp.readyState = 'upgrade'
+    })
+    return tcp
+  }
+}
+
+function Socket() {
+  const tcp = Object.assign(new EventEmitter(), {
+    readyState: 'open',
+    raw: null,
+    writer: null,
+    reader: null,
+    connect,
+    write,
+    end,
+    destroy,
+    read
+  })
+
+  return tcp
+
+  async function connect(port, host) {
+    try {
+      tcp.readyState = 'opening'
+      const { connect } = await import('cloudflare:sockets')
+      tcp.raw = connect(host + ':' + port, tcp.ssl ? { secureTransport: 'starttls' } : {})
+      tcp.raw.closed.then(
+        () => {
+          tcp.readyState !== 'upgrade'
+            ? close()
+            : ((tcp.readyState = 'open'), tcp.emit('secureConnect'))
+        },
+        (e) => tcp.emit('error', e)
+      )
+      tcp.writer = tcp.raw.writable.getWriter()
+      tcp.reader = tcp.raw.readable.getReader()
+
+      tcp.ssl ? readFirst() : read()
+      tcp.writer.ready.then(() => {
+        tcp.readyState = 'open'
+        tcp.emit('connect')
+      })
+    } catch (err) {
+      error(err)
+    }
+  }
+
+  function close() {
+    if (tcp.readyState === 'closed')
+      return
+
+    tcp.readyState = 'closed'
+    tcp.emit('close')
+  }
+
+  function write(data, cb) {
+    tcp.writer.write(data).then(cb, error)
+    return true
+  }
+
+  function end(data) {
+    return data
+      ? tcp.write(data, () => tcp.raw.close())
+      : tcp.raw.close()
+  }
+
+  function destroy() {
+    tcp.destroyed = true
+    tcp.end()
+  }
+
+  async function read() {
+    try {
+      let done
+        , value
+      while (({ done, value } = await tcp.reader.read(), !done))
+        tcp.emit('data', Buffer.from(value))
+    } catch (err) {
+      error(err)
+    }
+  }
+
+  async function readFirst() {
+    const { value } = await tcp.reader.read()
+    tcp.emit('data', Buffer.from(value))
+  }
+
+  function error(err) {
+    tcp.emit('error', err)
+    tcp.emit('close')
+  }
+}
+
+function setImmediate(fn) {
+  const id = ids++
+  tasks.add(id)
+  queueMicrotask(() => {
+    if (tasks.has(id)) {
+      fn()
+      tasks.delete(id)
+    }
+  })
+  return id
+}
+
+function clearImmediate(id) {
+  tasks.delete(id)
+}

cf/src/bytes.js

@@ -0,0 +1,79 @@
+import { Buffer } from 'node:buffer'
+const size = 256
+let buffer = Buffer.allocUnsafe(size)
+
+const messages = 'BCcDdEFfHPpQSX'.split('').reduce((acc, x) => {
+  const v = x.charCodeAt(0)
+  acc[x] = () => {
+    buffer[0] = v
+    b.i = 5
+    return b
+  }
+  return acc
+}, {})
+
+const b = Object.assign(reset, messages, {
+  N: String.fromCharCode(0),
+  i: 0,
+  inc(x) {
+    b.i += x
+    return b
+  },
+  str(x) {
+    const length = Buffer.byteLength(x)
+    fit(length)
+    b.i += buffer.write(x, b.i, length, 'utf8')
+    return b
+  },
+  i16(x) {
+    fit(2)
+    buffer.writeUInt16BE(x, b.i)
+    b.i += 2
+    return b
+  },
+  i32(x, i) {
+    if (i || i === 0) {
+      buffer.writeUInt32BE(x, i)
+      return b
+    }
+    fit(4)
+    buffer.writeUInt32BE(x, b.i)
+    b.i += 4
+    return b
+  },
+  z(x) {
+    fit(x)
+    buffer.fill(0, b.i, b.i + x)
+    b.i += x
+    return b
+  },
+  raw(x) {
+    buffer = Buffer.concat([buffer.subarray(0, b.i), x])
+    b.i = buffer.length
+    return b
+  },
+  end(at = 1) {
+    buffer.writeUInt32BE(b.i - at, at)
+    const out = buffer.subarray(0, b.i)
+    b.i = 0
+    buffer = Buffer.allocUnsafe(size)
+    return out
+  }
+})
+
+export default b
+
+function fit(x) {
+  if (buffer.length - b.i < x) {
+    const prev = buffer
+        , length = prev.length
+
+    buffer = Buffer.allocUnsafe(length + (length >> 1) + x)
+    prev.copy(buffer)
+  }
+}
+
+function reset() {
+  b.i = 0
+  return b
+}