Soloseng/add-combiner-db

packages/phone-number-privacy/combiner/package.json

@@ -23,9 +23,15 @@
  "test": "jest --runInBand --testPathIgnorePatterns test/end-to-end",
  "test:coverage": "yarn test --coverage",
  "test:integration": "jest --runInBand test/integration",
+ "test:integration:debugdb": "VERBOSE_DB_LOGGING=true jest --runInBand test/integration",
  "test:e2e": "jest test/end-to-end --verbose",
  "test:e2e:staging": "CONTEXT_NAME=staging yarn test:e2e",
- "test:e2e:alfajores": "CONTEXT_NAME=alfajores yarn test:e2e"
+ "test:e2e:alfajores": "CONTEXT_NAME=alfajores yarn test:e2e",
+ "db:migrate": "NODE_ENV=dev FIREBASE_CONFIG=./firebase.json ts-node ./scripts/run-migrations.ts",
+ "db:migrate:staging": "GCLOUD_PROJECT=celo-phone-number-privacy-stg yarn db:migrate",
+ "db:migrate:alfajores": "GCLOUD_PROJECT=celo-phone-number-privacy yarn db:migrate",
+ "db:migrate:mainnet": "GCLOUD_PROJECT=celo-pgpnp-mainnet yarn db:migrate",
+ "db:migrate:make": "knex --migrations-directory ./migrations migrate:make -x ts"
  },
  "dependencies": {
  "@celo/contractkit": "^4.1.1-beta.1",

packages/phone-number-privacy/combiner/scripts/run-migrations.ts

@@ -0,0 +1,19 @@
+// tslint:disable: no-console
+// TODO de-dupe with signer script
+import { initDatabase } from '../src/database/database'
+import config from '../src/config'
+
+async function start() {
+ console.info('Running migrations')
+ await initDatabase(config, undefined, false)
+}
+
+start()
+ .then(() => {
+ console.info('Migrations complete')
+ process.exit(0)
+ })
+ .catch((e) => {
+ console.error('Migration failed', e)
+ process.exit(1)
+ })

packages/phone-number-privacy/combiner/src/config.ts

@@ -1,5 +1,7 @@
 import {
  BlockchainConfig,
+ DB_POOL_MAX_SIZE,
+ DB_TIMEOUT,
  FULL_NODE_TIMEOUT_IN_MS,
  RETRY_COUNT,
  RETRY_DELAY_IN_MS,
@@ -13,6 +15,7 @@ export function getCombinerVersion(): string {
 }
 export const DEV_MODE =
  process.env.NODE_ENV !== 'production' || process.env.FUNCTIONS_EMULATOR === 'true'
+export const VERBOSE_DB_LOGGING = toBool(process.env.VERBOSE_DB_LOGGING, false)
 
 export const FORNO_ALFAJORES = 'https://alfajores-forno.celo-testnet.org'
 
@@ -25,6 +28,13 @@ export const MAX_BLOCK_DISCREPANCY_THRESHOLD = 3
 export const MAX_TOTAL_QUOTA_DISCREPANCY_THRESHOLD = 5
 export const MAX_QUERY_COUNT_DISCREPANCY_THRESHOLD = 5
 
+export enum SupportedDatabase {
+ Postgres = 'postgres', // PostgresSQL
+ MySql = 'mysql', // MySQL
+ MsSql = 'mssql', // Microsoft SQL Server
+ Sqlite = 'sqlite3', // SQLite (for testing)
+}
+
 export interface OdisConfig {
  serviceName: string
  enabled: boolean
@@ -46,6 +56,17 @@ export interface CombinerConfig {
  blockchain: BlockchainConfig
  phoneNumberPrivacy: OdisConfig
  domains: OdisConfig
+ db: {
+ type: SupportedDatabase
+ user: string
+ password: string
+ database: string
+ host: string
+ port?: number
+ ssl: boolean
+ poolMaxSize: number
+ timeout: number
+ }
 }
 
 let config: CombinerConfig
@@ -141,6 +162,17 @@ if (DEV_MODE) {
  fullNodeRetryCount: RETRY_COUNT,
  fullNodeRetryDelayMs: RETRY_DELAY_IN_MS,
  },
+ db: {
+ type: SupportedDatabase.Sqlite,
+ user: '',
+ password: '',
+ database: 'phoneNumber+privacy',
+ host: 'http://localhost',
+ port: undefined,
+ ssl: true,
+ poolMaxSize: DB_POOL_MAX_SIZE,
+ timeout: DB_TIMEOUT,
+ },
  }
 } else {
  const functionConfig = functions.config()
@@ -188,6 +220,21 @@ if (DEV_MODE) {
  functionConfig.pnp.full_node_retry_delay_ms ?? RETRY_DELAY_IN_MS
  ),
  },
+ db: {
+ type: functionConfig.db.type
+ ? functionConfig.db.type.toLowerCase()
+ : SupportedDatabase.Postgres,
+ user: functionConfig.db.username,
+ password: functionConfig.db.pass,
+ database: functionConfig.db.name,
+ host: `/cloudsql/${functionConfig.db.host}`,
+ port: functionConfig.db.port ? Number(functionConfig.db.port) : undefined,
+ ssl: toBool(functionConfig.db.ssl, true),
+ poolMaxSize: functionConfig.db.pool_max_size
+ ? Number(functionConfig.db.pool_max_size)
+ : DB_POOL_MAX_SIZE,
+ timeout: functionConfig.db.timeout ? Number(functionConfig.db.timeout) : DB_TIMEOUT,
+ },
  }
 }
 export default config

packages/phone-number-privacy/combiner/src/database/database.ts

@@ -0,0 +1,96 @@
+import { rootLogger } from '@celo/phone-number-privacy-common'
+import Logger from 'bunyan'
+import knex, { Knex } from 'knex'
+import { CombinerConfig, DEV_MODE, SupportedDatabase, VERBOSE_DB_LOGGING } from '../config'
+import { ACCOUNTS_COLUMNS, ACCOUNTS_TABLE } from './models/account'
+
+export async function initDatabase(
+ config: CombinerConfig,
+ migrationsPath?: string,
+ doTestQuery = true
+): Promise<Knex> {
+ const logger = rootLogger(config.serviceName)
+ logger.info({ config: config.db }, 'Initializing database connection')
+ const { type, host, port, user, password, database, ssl, poolMaxSize } = config.db
+
+ let connection: any
+ let client: string
+ if (type === SupportedDatabase.Postgres) {
+ logger.info('Using Postgres')
+ client = 'pg'
+ connection = {
+ user,
+ password,
+ database,
+ host,
+ port: port ?? 5432,
+ ssl,
+ pool: { max: poolMaxSize },
+ }
+ } else if (type === SupportedDatabase.MySql) {
+ logger.info('Using MySql')
+ client = 'mysql2'
+ connection = {
+ user,
+ password,
+ database,
+ host,
+ port: port ?? 3306,
+ ssl,
+ pool: { max: poolMaxSize },
+ }
+ } else if (type === SupportedDatabase.MsSql) {
+ logger.info('Using MS SQL')
+ client = 'mssql'
+ connection = {
+ user,
+ password,
+ database,
+ server: host,
+ port: port ?? 1433,
+ pool: { max: poolMaxSize },
+ }
+ } else if (type === SupportedDatabase.Sqlite) {
+ logger.info('Using SQLite - combiner')
+ client = 'sqlite3'
+ connection = ':memory:'
+ } else {
+ throw new Error(`Unsupported database type: ${type}`)
+ }
+ const db = knex({
+ client,
+ useNullAsDefault: type === SupportedDatabase.Sqlite,
+ connection,
+ debug: DEV_MODE && VERBOSE_DB_LOGGING,
+ })
+
+ logger.info('Running Migrations')
+
+ await db.migrate.latest({
+ directory: migrationsPath ?? './dist/database/migrations',
+ loadExtensions: ['.js'],
+ })
+
+ if (doTestQuery) {
+ await executeTestQuery(db, logger)
+ }
+
+ logger.info('Database initialized successfully')
+ return db
+}
+
+async function executeTestQuery(db: Knex, logger: Logger) {
+ logger.info('Counting accounts')
+ const result = await db(ACCOUNTS_TABLE).count(ACCOUNTS_COLUMNS.address).first()
+
+ if (!result) {
+ throw new Error('No result from count, have migrations been run?')
+ }
+
+ const count = Object.values(result)[0]
+ if (count === undefined || count === null || count === '') {
+ throw new Error('No result from count, have migrations been run?')
+ }
+
+ logger.info(`Found ${count} accounts`)
+}

packages/phone-number-privacy/combiner/src/database/migrations/20230815120000_create_accounts_table.ts

@@ -0,0 +1,18 @@
+import { Knex } from 'knex'
+import { ACCOUNTS_COLUMNS, ACCOUNTS_TABLE } from '../models/account'
+
+export async function up(knex: Knex): Promise<any> {
+ if (!(await knex.schema.hasTable(ACCOUNTS_TABLE))) {
+ return knex.schema.createTable(ACCOUNTS_TABLE, (t) => {
+ t.string(ACCOUNTS_COLUMNS.address).notNullable().primary()
+ t.dateTime(ACCOUNTS_COLUMNS.createdAt).notNullable()
+ t.string(ACCOUNTS_COLUMNS.dek)
+ t.dateTime(ACCOUNTS_COLUMNS.onChainDataLastUpdated)
+ })
+ }
+ return null
+}
+
+export async function down(knex: Knex): Promise<any> {
+ return knex.schema.dropTable(ACCOUNTS_TABLE)
+}

packages/phone-number-privacy/combiner/src/database/models/account.ts

@@ -0,0 +1,22 @@
+export const ACCOUNTS_TABLE = 'accounts'
+
+export enum ACCOUNTS_COLUMNS {
+ address = 'address',
+ createdAt = 'created_at',
+ dek = 'dek',
+ onChainDataLastUpdated = 'onChainDataLastUpdated',
+}
+
+export class Account {
+ [ACCOUNTS_COLUMNS.address]: string | undefined;
+ [ACCOUNTS_COLUMNS.createdAt]: Date = new Date();
+ [ACCOUNTS_COLUMNS.dek]: string | undefined;
+ [ACCOUNTS_COLUMNS.onChainDataLastUpdated]: Date | null = null
+
+ constructor(address: string, dek?: string) {
+ this.address = address
+ if (dek) {
+ this.dek = dek
+ }
+ }
+}

packages/phone-number-privacy/combiner/src/database/wrappers/account.ts

@@ -0,0 +1,79 @@
+import { DB_TIMEOUT, ErrorMessage } from '@celo/phone-number-privacy-common'
+import Logger from 'bunyan'
+import { Knex } from 'knex'
+import { Account, ACCOUNTS_COLUMNS, ACCOUNTS_TABLE } from '../models/account'
+
+function accounts(db: Knex) {
+ return db<Account>(ACCOUNTS_TABLE)
+}
+
+/*
+ * Get DEK signer record from DB.
+ */
+export async function getDekSignerRecord(
+ db: Knex,
+ account: string,
+ logger: Logger
+): Promise<string | undefined> {
+ try {
+ logger.info('getting Dek Signer Record')
+ const dekSignerRecord = await accounts(db)
+ .where(ACCOUNTS_COLUMNS.address, account)
+ .select(ACCOUNTS_COLUMNS.dek)
+ .first()
+ .timeout(DB_TIMEOUT)
+
+ return dekSignerRecord ? dekSignerRecord[ACCOUNTS_COLUMNS.dek] : undefined
+ } catch (err) {
+ logger.error(ErrorMessage.DATABASE_GET_FAILURE)
+ logger.error(err)
+ return undefined
+ }
+}
+
+export async function updateDekSignerRecord(
+ db: Knex,
+ account: string,
+ newDek: string,
+ logger: Logger,
+ trx: Knex.Transaction
+) {
+ logger.info(`updating Dek Signer Record`)
+ if (await getAccountExist(db, account, trx)) {
+ await accounts(db)
+ .transacting(trx)
+ .timeout(DB_TIMEOUT)
+ .where(ACCOUNTS_COLUMNS.address, account)
+ .update({ [ACCOUNTS_COLUMNS.dek]: newDek })
+ await accounts(db)
+ .transacting(trx)
+ .timeout(DB_TIMEOUT)
+ .where(ACCOUNTS_COLUMNS.address, account)
+ .update({ [ACCOUNTS_COLUMNS.onChainDataLastUpdated]: new Date() })
+ } else {
+ // account does not exists
+ const newAccount = new Account(account, newDek)
+ await accounts(db).transacting(trx).timeout(DB_TIMEOUT).insert(newAccount)
+ }
+}
+
+export function tableWithLockForTrx(baseQuery: Knex.QueryBuilder, trx?: Knex.Transaction) {
+ if (trx) {
+ // Lock relevant database rows for the duration of the transaction
+ return baseQuery.transacting(trx).forUpdate()
+ }
+ return baseQuery
+}
+
+async function getAccountExist(
+ db: Knex,
+ account: string,
+ trx?: Knex.Transaction
+): Promise<boolean> {
+ const accountRecord = await tableWithLockForTrx(accounts(db), trx)
+ .where(ACCOUNTS_COLUMNS.address, account)
+ .first()
+ .timeout(DB_TIMEOUT)
+
+ return !!accountRecord
+}

packages/phone-number-privacy/combiner/src/index.ts

@@ -1,6 +1,8 @@
 import { getContractKitWithAgent } from '@celo/phone-number-privacy-common'
 import * as functions from 'firebase-functions'
+import { Knex } from 'knex'
 import config from './config'
+import { initDatabase } from './database/database'
 import { startCombiner } from './server'
 
 require('dotenv').config()
@@ -12,5 +14,14 @@ export const combiner = functions
  // Defined check required for running tests vs. deployment
  minInstances: functions.config().service ? Number(functions.config().service.min_instances) : 0,
  })
- .https.onRequest(startCombiner(config, getContractKitWithAgent(config.blockchain)))
+ .https.onRequest(async (req, res) => {
+ try {
+ const db: Knex = await initDatabase(config)
+ const app = startCombiner(db, config, getContractKitWithAgent(config.blockchain))
+
+ app(req, res)
+ } catch (e) {
+ res.status(500).send('Internal Server Error')
+ }
+ })
 export * from './config'

packages/phone-number-privacy/combiner/src/pnp/endpoints/quota/action.ts

@@ -1,5 +1,4 @@
 import {
- authenticateUser,
  CombinerEndpoint,
  DataEncryptionKeyFetcher,
  ErrorMessage,
@@ -19,8 +18,11 @@ import { getKeyVersionInfo, sendFailure } from '../../../common/io'
 import { getCombinerVersion, OdisConfig } from '../../../config'
 import { logPnpSignerResponseDiscrepancies } from '../../services/log-responses'
 import { findCombinerQuotaState } from '../../services/threshold-state'
+import { authenticateUser } from '../../../utils/authentication'
+import { Knex } from 'knex'
 
 export function createPnpQuotaHandler(
+ db: Knex,
  signers: Signer[],
  config: OdisConfig,
  dekFetcher: DataEncryptionKeyFetcher
@@ -33,7 +35,7 @@ export function createPnpQuotaHandler(
  return
  }
 
- if (!(await authenticateUser(request, logger, dekFetcher))) {
+ if (!(await authenticateUser(db, request, logger, dekFetcher))) {
  sendFailure(WarningMessage.UNAUTHENTICATED_USER, 401, response)
  return
  }