Replace @databases/pg with pg, make log optional

README.md

@@ -15,16 +15,15 @@ npm install @platformatic/leader
 ```js
 'use strict'
 
-const createConnectionPool = require('@databases/pg')
+const pg = require('pg')
 const createLeaderElector = require('@platformatic/leader')
 
-const db = createConnectionPool({
-  connectionString: 'postgres://localhost/mydb',
-  bigIntMode: 'bigint'
+const pool = new pg.Pool({
+  connectionString: 'postgres://localhost/mydb'
 })
 
 const leader = createLeaderElector({
-  db,
+  pool,
   lock: 4242,
   poll: 10000,
   channels: [
@@ -35,7 +34,6 @@ const leader = createLeaderElector({
       }
     }
   ],
-  log: console,
   onLeadershipChange: (isLeader) => {
     console.log('Leadership changed:', isLeader)
   }
@@ -51,7 +49,7 @@ console.log(leader.isLeader())
 
 // Stop the leader elector
 await leader.stop()
-await db.dispose()
+await pool.end()
 ```
 
 ## API
@@ -64,11 +62,11 @@ Returns an object with `{ start, stop, notify, isLeader }`.
 
 | Option | Type | Required | Default | Description |
 |---|---|---|---|---|
-| `db` | `ConnectionPool` | Yes | - | A `@databases/pg` connection pool |
+| `pool` | `pg.Pool` | Yes | - | A [node-postgres](https://node-postgres.com/) pool instance |
 | `lock` | `number` | Yes | - | PostgreSQL advisory lock ID |
 | `poll` | `number` | No | `10000` | Polling interval in milliseconds |
 | `channels` | `Array` | Yes | - | Notification channel configurations |
-| `log` | `object` | Yes | - | Logger with `info`, `debug`, `warn`, `error` methods |
+| `log` | `object` | No | `pino()` | Logger with `info`, `debug`, `warn`, `error` methods |
 | `onLeadershipChange` | `function` | No | `null` | Callback invoked with `(isLeader: boolean)` when leadership status changes |
 
 Each channel in the `channels` array must have:
@@ -90,6 +88,20 @@ Each channel in the `channels` array must have:
 4. If the leader stops or loses the lock, another instance acquires it and takes over.
 5. The `onLeadershipChange` callback fires whenever the leadership status transitions.
 
+## Development
+
+Start PostgreSQL:
+
+```
+docker compose up -d
+```
+
+Run tests:
+
+```
+node --test test/*.test.js
+```
+
 ## License
 
 Apache-2.0

docker-compose.yml

@@ -0,0 +1,9 @@
+services:
+  postgres:
+    ports:
+      - "127.0.0.1:5433:5432"
+    image: "postgres"
+    environment:
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_DB: leader
+      PGDATA: /data/postgres

index.js

@@ -2,30 +2,27 @@
 
 const { scheduler } = require('timers/promises')
 const { on } = require('events')
+const pino = require('pino')
 
 function createLeaderElector (options) {
   const {
-    db,
+    pool,
     lock,
     poll = 10000,
     channels,
-    log,
+    log = pino(),
     onLeadershipChange = null
   } = options
 
-  log.info('Acquiring advisory lock %d', lock)
-
-  if (!db) {
-    throw new Error('db is required')
+  if (!pool) {
+    throw new Error('pool is required')
   }
 
   if (!lock) {
     throw new Error('lock is required')
   }
 
-  if (!log) {
-    throw new Error('log is required')
-  }
+  log.info('Acquiring advisory lock %d', lock)
 
   if (!channels || !Array.isArray(channels) || channels.length === 0) {
     throw new Error('channels array is required')
@@ -56,22 +53,23 @@ function createLeaderElector (options) {
   }
 
   async function amITheLeader () {
-    const sql = db.sql
-    await db.task(async (t) => {
+    const client = await pool.connect()
+    try {
       while (!abortController.signal.aborted) {
-        const [{ leader }] = await t.query(sql`
-          SELECT pg_try_advisory_lock(${lock}) as leader;
-        `)
+        const { rows: [{ leader }] } = await client.query(
+          'SELECT pg_try_advisory_lock($1) as leader',
+          [lock]
+        )
         if (leader && !elected) {
           log.info('This instance is the leader')
           updateLeadershipStatus(true)
           ;(async () => {
             for (const ch of notificationChannels) {
-              await t.query(sql.__dangerous__rawValue(`LISTEN "${ch.channel}";`))
+              await client.query(`LISTEN "${ch.channel}"`)
               log.info({ channel: ch.channel }, 'Listening to notification channel')
             }
 
-            for await (const notification of on(t._driver.client, 'notification', { signal: abortController.signal })) {
+            for await (const notification of on(client, 'notification', { signal: abortController.signal })) {
               log.debug({ notification }, 'Received notification')
               try {
                 const msg = notification[0]
@@ -110,7 +108,9 @@ function createLeaderElector (options) {
           break
         }
       }
-    })
+    } finally {
+      client.release()
+    }
     log.debug('leader loop stopped')
   }
 
@@ -140,8 +140,7 @@ function createLeaderElector (options) {
 
     payload = JSON.stringify(payload)
 
-    const sql = db.sql
-    await db.query(sql.__dangerous__rawValue(`NOTIFY "${channelName}", '${payload}';`))
+    await pool.query(`NOTIFY "${channelName}", '${payload}'`)
   }
 
   async function stop () {

package.json

@@ -8,10 +8,10 @@
     "lint": "standard ."
   },
   "dependencies": {
-    "@databases/pg": "^5.5.0"
+    "pg": "^8.20.0",
+    "pino": "^10.3.1"
   },
   "devDependencies": {
-    "pg": "^8.13.3",
     "standard": "^17.1.2"
   },
   "engines": {