build

cjs/src/connection.js

@@ -429,10 +429,8 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
  lifeTimer.cancel()
  connectTimer.cancel()
 
- if (socket.encrypted) {
- socket.removeAllListeners()
- socket = null
- }
+ socket.removeAllListeners()
+ socket = null
 
  if (initial)
  return reconnect()
@@ -790,7 +788,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
  const error = Errors.postgres(parseError(x))
  query && query.retried
  ? errored(query.retried)
- : query && retryRoutines.has(error.routine)
+ : query && query.prepare && retryRoutines.has(error.routine)
  ? retry(query, error)
  : errored(error)
  }

cjs/src/subscribe.js

@@ -47,7 +47,7 @@ module.exports = Subscribe;function Subscribe(postgres, options) {
 
  return subscribe
 
- async function subscribe(event, fn, onsubscribe = noop) {
+ async function subscribe(event, fn, onsubscribe = noop, onerror = noop) {
  event = parseEvent(event)
 
  if (!connection)
@@ -66,6 +66,7 @@ module.exports = Subscribe;function Subscribe(postgres, options) {
  return connection.then(x => {
  connected(x)
  onsubscribe()
+ stream && stream.on('error', onerror)
  return { unsubscribe, state, sql }
  })
  }
@@ -109,8 +110,10 @@ module.exports = Subscribe;function Subscribe(postgres, options) {
  function data(x) {
  if (x[0] === 0x77)
  parse(x.subarray(25), state, sql.options.parsers, handle, options.transform)
- else if (x[0] === 0x6b && x[17])
+ else if (x[0] === 0x6b && x[17]) {
+ state.lsn = x.subarray(1, 9)
  pong()
+ }
  }
 
  function handle(a, b) {

cjs/tests/index.js

@@ -1789,6 +1789,21 @@ t('Recreate prepared statements on RevalidateCachedQuery error', async() => {
  ]
 })
 
+t('Properly throws routing error on not prepared statements', async() => {
+ await sql`create table x (x text[])`
+ const { routine } = await sql.unsafe(`insert into x(x) values (('a', 'b'))`).catch(e => e)
+
+ return ['transformAssignedExpr', routine, await sql`drop table x`]
+})
+
+t('Properly throws routing error on not prepared statements in transaction', async() => {
+ const { routine } = await sql.begin(sql => [
+ sql`create table x (x text[])`,
+ sql`insert into x(x) values (('a', 'b'))`,
+ ]).catch(e => e)
+
+ return ['transformAssignedExpr', routine]
+})
 
 t('Catches connection config errors', async() => {
  const sql = postgres({ ...options, user: { toString: () => { throw new Error('wat') } }, database: 'prut' })

deno/README.md

@@ -266,7 +266,7 @@ const users = [
 ]
 
 await sql`
- update users set name = update_data.name, (age = update_data.age)::int
+ update users set name = update_data.name, age = (update_data.age)::int
  from (values ${sql(users)}) as update_data (id, name, age)
  where users.id = (update_data.id)::int
  returning users.id, users.name, users.age
@@ -286,7 +286,7 @@ const users = await sql`
 
 or
 ```js
-const [{ a, b, c }] => await sql`
+const [{ a, b, c }] = await sql`
  select
  *
  from (values ${ sql(['a', 'b', 'c']) }) as x(a, b, c)
@@ -913,7 +913,7 @@ The `Result` Array returned from queries is a custom array allowing for easy des
 
 ### .count
 
-The `count` property is the number of affected rows returned by the database. This is usefull for insert, update and delete operations to know the number of rows since .length will be 0 in these cases if not using `RETURNING ...`.
+The `count` property is the number of affected rows returned by the database. This is useful for insert, update and delete operations to know the number of rows since .length will be 0 in these cases if not using `RETURNING ...`.
 
 ### .command
 
@@ -1099,10 +1099,10 @@ export default async fetch(req: Request, env: Env, ctx: ExecutionContext) {
 }
 ```
 
-In `wrangler.toml` you will need to enable `node_compat` to allow Postgres.js to operate in the Workers environment:
+In `wrangler.toml` you will need to enable the `nodejs_compat` compatibility flag to allow Postgres.js to operate in the Workers environment:
 
 ```toml
-node_compat = true # required for database drivers to function
+compatibility_flags = ["nodejs_compat"]
 ```
 
 ### Auto fetching of array types

deno/src/connection.js

@@ -432,10 +432,8 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
  lifeTimer.cancel()
  connectTimer.cancel()
 
- if (socket.encrypted) {
- socket.removeAllListeners()
- socket = null
- }
+ socket.removeAllListeners()
+ socket = null
 
  if (initial)
  return reconnect()
@@ -793,7 +791,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
  const error = Errors.postgres(parseError(x))
  query && query.retried
  ? errored(query.retried)
- : query && retryRoutines.has(error.routine)
+ : query && query.prepare && retryRoutines.has(error.routine)
  ? retry(query, error)
  : errored(error)
  }

deno/src/subscribe.js

@@ -48,7 +48,7 @@ export default function Subscribe(postgres, options) {
 
  return subscribe
 
- async function subscribe(event, fn, onsubscribe = noop) {
+ async function subscribe(event, fn, onsubscribe = noop, onerror = noop) {
  event = parseEvent(event)
 
  if (!connection)
@@ -67,6 +67,7 @@ export default function Subscribe(postgres, options) {
  return connection.then(x => {
  connected(x)
  onsubscribe()
+ stream && stream.on('error', onerror)
  return { unsubscribe, state, sql }
  })
  }
@@ -110,8 +111,10 @@ export default function Subscribe(postgres, options) {
  function data(x) {
  if (x[0] === 0x77)
  parse(x.subarray(25), state, sql.options.parsers, handle, options.transform)
- else if (x[0] === 0x6b && x[17])
+ else if (x[0] === 0x6b && x[17]) {
+ state.lsn = x.subarray(1, 9)
  pong()
+ }
  }
 
  function handle(a, b) {

deno/tests/index.js

@@ -1791,6 +1791,21 @@ t('Recreate prepared statements on RevalidateCachedQuery error', async() => {
  ]
 })
 
+t('Properly throws routing error on not prepared statements', async() => {
+ await sql`create table x (x text[])`
+ const { routine } = await sql.unsafe(`insert into x(x) values (('a', 'b'))`).catch(e => e)
+
+ return ['transformAssignedExpr', routine, await sql`drop table x`]
+})
+
+t('Properly throws routing error on not prepared statements in transaction', async() => {
+ const { routine } = await sql.begin(sql => [
+ sql`create table x (x text[])`,
+ sql`insert into x(x) values (('a', 'b'))`,
+ ]).catch(e => e)
+
+ return ['transformAssignedExpr', routine]
+})
 
 t('Catches connection config errors', async() => {
  const sql = postgres({ ...options, user: { toString: () => { throw new Error('wat') } }, database: 'prut' })

deno/types/index.d.ts

@@ -601,6 +601,7 @@ declare namespace postgres {
  type RowList<T extends readonly any[]> = T & Iterable<NonNullable<T[number]>> & ResultQueryMeta<T['length'], keyof T[number]>;
 
  interface PendingQueryModifiers<TRow extends readonly any[]> {
+ simple(): this;
  readable(): Promise<Readable>;
  writable(): Promise<Writable>;
 
@@ -692,7 +693,7 @@ declare namespace postgres {
  listen(channel: string, onnotify: (value: string) => void, onlisten?: (() => void) | undefined): ListenRequest;
  notify(channel: string, payload: string): PendingRequest;
 
- subscribe(event: string, cb: (row: Row | null, info: ReplicationEvent) => void, onsubscribe?: (() => void) | undefined): Promise<SubscriptionHandle>;
+ subscribe(event: string, cb: (row: Row | null, info: ReplicationEvent) => void, onsubscribe?: (() => void), onerror?: (() => any)): Promise<SubscriptionHandle>;
 
  largeObject(oid?: number | undefined, /** @default 0x00020000 | 0x00040000 */ mode?: number | undefined): Promise<LargeObject>;