build

Author: porsager

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>;