migrate token pooling to postgres

.github/actions/integration-tests/action.yml

@@ -7,11 +7,19 @@ inputs:
 runs:
   using: 'composite'
   steps:
+    - name: Migrate DB
+      if: always()
+      run: npm run migrate up
+      env:
+        POSTGRES_URL: postgresql://postgres:postgres@localhost:5432/ci_test
+      shell: bash
+
     - name: Integration Tests
       if: always()
       run: npm run test:integration -- --reporter json --reporter-option 'output=reports/integration-tests.json'
       env:
         GH_TOKEN: '${{ inputs.github-token }}'
+        POSTGRES_URL: postgresql://postgres:postgres@localhost:5432/ci_test
       shell: bash
 
     - name: Write Markdown Summary

.github/workflows/test-integration-17.yml

@@ -23,6 +23,19 @@ jobs:
           --health-retries 5
         ports:
           - 6379:6379
+      postgres:
+        image: postgres
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: ci_test
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 5432:5432
 
     steps:
       - name: Checkout

.github/workflows/test-integration.yml

@@ -23,6 +23,19 @@ jobs:
           --health-retries 5
         ports:
           - 6379:6379
+      postgres:
+        image: postgres
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: ci_test
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 5432:5432
 
     steps:
       - name: Checkout

.gitignore

@@ -117,3 +117,6 @@ service-definitions.yml
 
 # Flamebearer
 flamegraph.html
+
+# config file for node-pg-migrate
+migrations-config.json

config/custom-environment-variables.yml

@@ -94,6 +94,7 @@ private:
   obs_user: 'OBS_USER'
   obs_pass: 'OBS_PASS'
   redis_url: 'REDIS_URL'
+  postgres_url: 'POSTGRES_URL'
   sentry_dsn: 'SENTRY_DSN'
   sl_insight_userUuid: 'SL_INSIGHT_USER_UUID'
   sl_insight_apiToken: 'SL_INSIGHT_API_TOKEN'

core/server/server.js

@@ -183,6 +183,7 @@ const privateConfigSchema = Joi.object({
   obs_user: Joi.string(),
   obs_pass: Joi.string(),
   redis_url: Joi.string().uri({ scheme: ['redis', 'rediss'] }),
+  postgres_url: Joi.string().uri({ scheme: 'postgresql' }),
   sentry_dsn: Joi.string(),
   sl_insight_userUuid: Joi.string(),
   sl_insight_apiToken: Joi.string(),

core/token-pooling/redis-token-persistence.integration.js

@@ -84,7 +84,6 @@ describe('Redis token persistence', function () {
         const toRemove = expected.pop()
 
         await persistence.initialize()
-
         await persistence.noteTokenRemoved(toRemove)
 
         const savedTokens = await redis.smembers(key)

core/token-pooling/sql-token-persistence.integration.js

@@ -0,0 +1,103 @@
+import pg from 'pg'
+import { expect } from 'chai'
+import configModule from 'config'
+import SqlTokenPersistence from './sql-token-persistence.js'
+
+const config = configModule.util.toObject()
+const postgresUrl = config?.private?.postgres_url
+const tableName = 'token_persistence_integration_test'
+
+describe('SQL token persistence', function () {
+  let pool
+  let persistence
+
+  before('Mock db connection and load app', async function () {
+    // Create a new pool with a connection limit of 1
+    pool = new pg.Pool({
+      connectionString: postgresUrl,
+
+      // Reuse the connection to make sure we always hit the same pg_temp schema
+      max: 1,
+
+      // Disable auto-disconnection of idle clients to make sure we always hit the same pg_temp schema
+      idleTimeoutMillis: 0,
+    })
+    persistence = new SqlTokenPersistence({
+      url: postgresUrl,
+      table: tableName,
+    })
+  })
+  after(async function () {
+    if (persistence) {
+      await persistence.stop()
+      persistence = undefined
+    }
+  })
+
+  beforeEach('Create temporary table', async function () {
+    await pool.query(
+      `CREATE TEMPORARY TABLE ${tableName} (LIKE github_user_tokens INCLUDING ALL);`
+    )
+  })
+  afterEach('Drop temporary table', async function () {
+    await pool.query(`DROP TABLE IF EXISTS pg_temp.${tableName};`)
+  })
+
+  context('when the key does not exist', function () {
+    it('does nothing', async function () {
+      const tokens = await persistence.initialize(pool)
+      expect(tokens).to.deep.equal([])
+    })
+  })
+
+  context('when the key exists', function () {
+    const initialTokens = ['a', 'b', 'c'].map(char => char.repeat(40))
+
+    beforeEach(async function () {
+      initialTokens.forEach(async token => {
+        await pool.query(
+          `INSERT INTO pg_temp.${tableName} (token) VALUES ($1::text);`,
+          [token]
+        )
+      })
+    })
+
+    it('loads the contents', async function () {
+      const tokens = await persistence.initialize(pool)
+      expect(tokens.sort()).to.deep.equal(initialTokens)
+    })
+
+    context('when tokens are added', function () {
+      it('saves the change', async function () {
+        const newToken = 'e'.repeat(40)
+        const expected = initialTokens.slice()
+        expected.push(newToken)
+
+        await persistence.initialize(pool)
+        await persistence.noteTokenAdded(newToken)
+
+        const result = await pool.query(
+          `SELECT token FROM pg_temp.${tableName};`
+        )
+        const savedTokens = result.rows.map(row => row.token)
+        expect(savedTokens.sort()).to.deep.equal(expected)
+      })
+    })
+
+    context('when tokens are removed', function () {
+      it('saves the change', async function () {
+        const expected = Array.from(initialTokens)
+        const toRemove = expected.pop()
+
+        await persistence.initialize(pool)
+        await persistence.noteTokenRemoved(toRemove)
+
+        const result = await pool.query(
+          `SELECT token FROM pg_temp.${tableName};`
+        )
+        const savedTokens = result.rows.map(row => row.token)
+        expect(savedTokens.sort()).to.deep.equal(expected)
+      })
+    })
+  })
+})

core/token-pooling/sql-token-persistence.js

@@ -0,0 +1,55 @@
+import pg from 'pg'
+import log from '../server/log.js'
+
+export default class SqlTokenPersistence {
+  constructor({ url, table }) {
+    this.url = url
+    this.table = table
+    this.noteTokenAdded = this.noteTokenAdded.bind(this)
+    this.noteTokenRemoved = this.noteTokenRemoved.bind(this)
+  }
+
+  async initialize(pool) {
+    if (pool) {
+      this.pool = pool
+    } else {
+      this.pool = new pg.Pool({ connectionString: this.url })
+    }
+    const result = await this.pool.query(`SELECT token FROM ${this.table};`)
+    return result.rows.map(row => row.token)
+  }
+
+  async stop() {
+    await this.pool.end()
+  }
+
+  async onTokenAdded(token) {
+    return await this.pool.query(
+      `INSERT INTO ${this.table} (token) VALUES ($1::text) ON CONFLICT (token) DO NOTHING;`,
+      [token]
+    )
+  }
+
+  async onTokenRemoved(token) {
+    return await this.pool.query(
+      `DELETE FROM ${this.table} WHERE token=$1::text;`,
+      [token]
+    )
+  }
+
+  async noteTokenAdded(token) {
+    try {
+      await this.onTokenAdded(token)
+    } catch (e) {
+      log.error(e)
+    }
+  }
+
+  async noteTokenRemoved(token) {
+    try {
+      await this.onTokenRemoved(token)
+    } catch (e) {
+      log.error(e)
+    }
+  }
+}

migrations/1676731511125_initialize.cjs

@@ -0,0 +1,14 @@
+/* eslint-disable camelcase */
+
+exports.shorthands = undefined
+
+exports.up = pgm => {
+  pgm.createTable('github_user_tokens', {
+    id: 'id',
+    token: { type: 'varchar(1000)', notNull: true, unique: true },
+  })
+}
+
+exports.down = pgm => {
+  pgm.dropTable('github_user_tokens')
+}