import * as pg from 'pg' ; import { isDatabaseError } from './pgErrors' ; import { InsertSignatures , UpdateSignatures , DeleteSignatures , SelectSignatures , SelectOneSignatures , CountSignatures , Insertable , Updatable , Whereable , Table , Column , UpsertSignatures , } from './schema' ; const config = { dbURL : 'postgresql://localhost/mostly_ormless' , dbTransactionAttempts : 5 , dbTransactionRetryDelayRange : [ 25 , 125 ] , verbose : true , } ; export const pool = new pg . Pool ( { connectionString : config . dbURL } ) ; export const Default = Symbol ( 'DEFAULT' ) ; export type DefaultType = typeof Default ; export const self = Symbol ( 'self' ) ; export type SelfType = typeof self ; export type JSONValue = null | boolean | number | string | JSONObject | JSONArray ; export interface JSONObject { [ k ] : JSONValue ; } export interface JSONArray extends Array < JSONValue > { } export class Parameter { constructor ( public value ) { } } export function param ( x ) { return new Parameter ( x ) ; } export class DangerousRawString { constructor ( public value ) { } } export function raw ( x ) { return new DangerousRawString ( x ) ; } export class ColumnNames < T > { constructor ( public value ) { } } export function cols < T > ( x ) { return new ColumnNames < T > ( x ) ; } export class ColumnValues < T > { constructor ( public value ) { } } export function vals < T > ( x ) { return new ColumnValues < T > ( x ) ; } export type GenericSQLExpression = SQLFragment | Parameter | DefaultType | DangerousRawString | SelfType ; export type SQLExpression = Table | ColumnNames < Updatable | ( keyof Updatable ) [ ] > | ColumnValues < Updatable > | Whereable | Column | GenericSQLExpression ; export type SQL = SQLExpression | SQLExpression [ ] ; export type Queryable = pg . Pool | PoolClient < any > ; export const insert = async function ( client , table , values ) : Promise < any > { const completedValues = Array . isArray ( values ) ? completeKeysWithDefault ( values ) : values , colsSQL = cols ( Array . isArray ( completedValues ) ? completedValues [ 0 ] : completedValues ) , valuesSQL = Array . isArray ( completedValues ) ? mapWithSeparator ( completedValues as Insertable [ ] , sql < SQL > `, ` , v => sql < SQL > `(${ vals ( v ) })` ) : sql < SQL > `(${ vals ( completedValues ) })` , query = sql < SQL > `INSERT INTO ${ table } (${ colsSQL }) VALUES ${ valuesSQL } RETURNING *` , rows = await query . run ( client ) ; return Array . isArray ( completedValues ) ? rows : rows [ 0 ] ; } ; export interface UpsertAction { $action : 'INSERT' | 'UPDATE' ; } ; export const upsert = async function ( client , table , values , uniqueCols , noNullUpdateCols = [ ] ) : Promise < any > { if ( ! Array . isArray ( uniqueCols ) ) uniqueCols = [ uniqueCols ] ; if ( ! Array . isArray ( noNullUpdateCols ) ) noNullUpdateCols = [ noNullUpdateCols ] ; const completedValues = Array . isArray ( values ) ? completeKeysWithDefault ( values ) : values , colsSQL = cols ( Array . isArray ( completedValues ) ? completedValues [ 0 ] : completedValues ) , valuesSQL = Array . isArray ( completedValues ) ? mapWithSeparator ( completedValues as Insertable [ ] , sql `, ` , v => sql `(${ vals ( v ) })` ) : sql `(${ vals ( completedValues ) })` , nonUniqueCols = ( Object . keys ( Array . isArray ( completedValues ) ? completedValues [ 0 ] : completedValues ) as Column [ ] ) . filter ( v => ! uniqueCols . includes ( v ) ) , uniqueColsSQL = mapWithSeparator ( uniqueCols . slice ( ) . sort ( ) , sql `, ` , c => c ) , updateColsSQL = mapWithSeparator ( nonUniqueCols . slice ( ) . sort ( ) , sql `, ` , c => c ) , updateValuesSQL = mapWithSeparator ( nonUniqueCols . slice ( ) . sort ( ) , sql `, ` , c => noNullUpdateCols . includes ( c ) ? sql `CASE WHEN EXCLUDED.${ c } IS NULL THEN ${ table }.${ c } ELSE EXCLUDED.${ c } END` : sql `EXCLUDED.${ c }` ) ; const query = sql < SQL > `INSERT INTO ${ table } (${ colsSQL }) VALUES ${ valuesSQL } ON CONFLICT (${ uniqueColsSQL }) DO UPDATE SET (${ updateColsSQL }) = ROW(${ updateValuesSQL }) RETURNING *, CASE xmax WHEN 0 THEN 'INSERT' ELSE 'UPDATE' END AS "$action"` ; const rows = await query . run ( client ) ; return Array . isArray ( completedValues ) ? rows : rows [ 0 ] ; } ; export const update = async function ( client , table , values , where ) : Promise < any [ ] > { const query = sql < SQL > `UPDATE ${ table } SET (${ cols ( values ) }) = ROW(${ vals ( values ) }) WHERE ${ where } RETURNING *` , rows = await query . run ( client ) ; return rows ; } ; export const deletes = async function ( client , table , where ) : Promise < any [ ] > { const query = sql < SQL > `DELETE FROM ${ table } WHERE ${ where } RETURNING *` , rows = await query . run ( client ) ; return rows ; } ; type TruncateIdentityOpts = 'CONTINUE IDENTITY' | 'RESTART IDENTITY' ; type TruncateForeignKeyOpts = 'RESTRICT' | 'CASCADE' ; interface TruncateSignatures { ( client , table , optId ) : Promise < void > ; ( client , table , optFK ) : Promise < void > ; ( client , table , optId , optFK ) : Promise < void > ; } export const truncate = async function ( client , table , ... opts ) : Promise < void > { if ( ! Array . isArray ( table ) ) table = [ table ] ; const tables = mapWithSeparator ( table , sql `, ` , t => t ) , query = sql < SQL > `TRUNCATE ${ tables }${ raw ( ( opts . length ? ' ' : '' ) + opts . join ( ' ' ) ) }` ; await query . run ( client ) ; } ; interface OrderSpec { by : SQL ; direction : 'ASC' | 'DESC' ; nulls ? : 'FIRST' | 'LAST' ; } interface SelectOptions { order ? : OrderSpec [ ] ; limit ? : number ; offset ? : number ; columns ? : Column [ ] ; } export const select = async function ( client , table , where = { } , options = { } , count = false ) : Promise < any [ ] > { const colsSQL = ! options . columns ? raw ( '*' ) : cols ( options . columns ) , whereSQL = Object . keys ( where ) . length > 0 ? [ sql ` WHERE ` , where ] : [ ] , orderSQL = ! options . order ? [ ] : [ sql ` ORDER BY ` , ... mapWithSeparator ( options . order , sql `, ` , o => sql `${ o . by } ${ raw ( o . direction ) }${ o . nulls ? sql ` NULLS ${ raw ( o . nulls ) }` : [ ] }` ) ] , limitSQL = options . limit === undefined ? [ ] : sql ` LIMIT ${ raw ( String ( options . limit ) ) }` , offsetSQL = options . offset === undefined ? [ ] : sql ` OFFSET ${ raw ( String ( options . offset ) ) }` ; const query = sql < SQL > `SELECT ${ count ? sql `count(${ colsSQL })` : colsSQL } FROM ${ table }${ whereSQL }${ orderSQL }${ limitSQL }${ offsetSQL }` , rows = query . run ( client ) ; return rows ; } ; export const selectOne = async function ( client , table , where = { } , options = { } ) : Promise < any > { const limitedOptions = Object . assign ( { } , options , { limit : 1 } ) , rows = await select ( client , table as any , where as any , limitedOptions ) ; return rows [ 0 ] ; } ; type CountResult = [ { count : string ; } ] ; export const count = async function ( client , table , where = { } , options = { } ) : Promise < number > { const rows = await select ( client , table as any , where as any , options as any , true ) as unknown as CountResult ; return Number ( rows [ 0 ] . count ) ; } ; export enum Isolation { Serializable = "SERIALIZABLE" , RepeatableRead = "REPEATABLE READ" , ReadCommitted = "READ COMMITTED" , SerializableRO = "SERIALIZABLE, READ ONLY" , RepeatableReadRO = "REPEATABLE READ, READ ONLY" , ReadCommittedRO = "READ COMMITTED, READ ONLY" , SerializableRODeferrable = "SERIALIZABLE, READ ONLY, DEFERRABLE" } export namespace TxnSatisfying { export type Serializable = Isolation . Serializable ; export type RepeatableRead = Serializable | Isolation . RepeatableRead ; export type ReadCommitted = RepeatableRead | Isolation . ReadCommitted ; export type SerializableRO = Serializable | Isolation . SerializableRO ; export type RepeatableReadRO = SerializableRO | RepeatableRead | Isolation . RepeatableReadRO ; export type ReadCommittedRO = RepeatableReadRO | ReadCommitted | Isolation . ReadCommittedRO ; export type SerializableRODeferrable = SerializableRO | Isolation . SerializableRODeferrable ; } export interface PoolClient < T extends Isolation | undefined > extends pg . PoolClient { transactionMode : T ; } let txnSeq = 0 ; export async function transaction < T , M extends Isolation > ( isolationMode , callback ) { const txnId = txnSeq ++ , txnClient = await pool . connect ( ) as PoolClient < typeof isolationMode > , maxAttempts = config . dbTransactionAttempts , [ delayMin , delayMax ] = config . dbTransactionRetryDelayRange ; txnClient . transactionMode = isolationMode ; try { for ( let attempt = 1 ; ; attempt ++ ) { try { if ( attempt > 1 ) console . log ( `Retrying transaction #${ txnId }, attempt ${ attempt } of ${ maxAttempts }` ) ; await sql `START TRANSACTION ISOLATION LEVEL ${ raw ( isolationMode ) }` . run ( txnClient ) ; const result = await callback ( txnClient ) ; await sql `COMMIT` . run ( txnClient ) ; return result ; } catch ( err ) { await sql `ROLLBACK` . run ( txnClient ) ; if ( isDatabaseError ( err , "TransactionRollback_SerializationFailure" , "TransactionRollback_DeadlockDetected" ) ) { if ( attempt < maxAttempts ) { const delayBeforeRetry = Math . round ( delayMin + ( delayMax - delayMin ) * Math . random ( ) ) ; console . log ( `Transaction #${ txnId } rollback (code ${ err . code }) on attempt ${ attempt } of ${ maxAttempts }, retrying in ${ delayBeforeRetry }ms` ) ; await wait ( delayBeforeRetry ) ; } else { console . log ( `Transaction #${ txnId } rollback (code ${ err . code }) on attempt ${ attempt } of ${ maxAttempts }, giving up` ) ; throw err ; } } else { throw err ; } } } } finally { ( txnClient as any ) . transactionMode = undefined ; txnClient . release ( ) ; } } interface SQLResultType { text : string ; values : any [ ] ; } ; export function sql < T = SQL > ( literals , ... expressions ) { return new SQLFragment ( Array . prototype . slice . apply ( literals ) , expressions ) ; } export class SQLFragment { constructor ( private literals , private expressions ) { } async run ( queryable ) { const query = this . compile ( ) ; if ( config . verbose ) console . log ( query ) ; const { rows } = await queryable . query ( query ) ; return rows ; } compile ( result = { text : '' , values : [ ] } , currentAlias ? ) { result . text += this . literals [ 0 ] ; for ( let i = 1 , len = this . literals . length ; i < len ; i ++ ) { this . compileExpression ( this . expressions [ i - 1 ] , result , currentAlias ) ; result . text += this . literals [ i ] ; } return result ; } compileExpression ( expression , result = { text : '' , values : [ ] } , currentAlias ? ) { if ( expression instanceof SQLFragment ) { expression . compile ( result , currentAlias ) ; } else if ( typeof expression === 'string' ) { result . text += `"${ expression }"` ; } else if ( expression instanceof DangerousRawString ) { result . text += expression . value ; } else if ( Array . isArray ( expression ) ) { for ( let i = 0 , len = expression . length ; i < len ; i ++ ) this . compileExpression ( expression [ i ] , result , currentAlias ) ; } else if ( expression instanceof Parameter ) { result . values . push ( expression . value ) ; result . text += '$' + String ( result . values . length ) ; } else if ( expression === Default ) { result . text += 'DEFAULT' ; } else if ( expression === self ) { if ( ! currentAlias ) throw new Error ( `The 'self' column alias has no meaning here` ) ; result . text += `"${ currentAlias }"` ; } else if ( expression instanceof ColumnNames ) { const columnNames = Array . isArray ( expression . value ) ? expression . value : Object . keys ( expression . value ) . sort ( ) ; result . text += columnNames . map ( k => `"${ k }"` ) . join ( ', ' ) ; } else if ( expression instanceof ColumnValues ) { const columnNames = Object . keys ( expression . value ) . sort ( ) , columnValues = columnNames . map ( k => ( < any > expression . value ) [ k ] ) ; for ( let i = 0 , len = columnValues . length ; i < len ; i ++ ) { const columnName = columnNames [ i ] , columnValue = columnValues [ i ] ; if ( i > 0 ) result . text += ', ' ; if ( columnValue instanceof SQLFragment || columnValue === Default ) this . compileExpression ( columnValue , result , columnName ) ; else this . compileExpression ( new Parameter ( columnValue ) , result , columnName ) ; } } else if ( typeof expression === 'object' ) { const columnNames = Object . keys ( expression ) . sort ( ) ; if ( columnNames . length ) { result . text += '(' ; for ( let i = 0 , len = columnNames . length ; i < len ; i ++ ) { const columnName = columnNames [ i ] , columnValue = ( < any > expression ) [ columnName ] ; if ( i > 0 ) result . text += ' AND ' ; if ( columnValue instanceof SQLFragment ) { result . text += '(' ; this . compileExpression ( columnValue , result , columnName ) ; result . text += ')' ; } else { result . text += `"${ columnName }" = ` ; this . compileExpression ( new Parameter ( columnValue ) , result , columnName ) ; } } result . text += ')' ; } else { result . text += 'TRUE' ; } } else { throw new Error ( `Alien object while interpolating SQL: ${ expression }` ) ; } } } const wait = ( delayMs ) => new Promise ( resolve => setTimeout ( resolve , delayMs ) ) ; const mapWithSeparator = < TIn , TSep , TOut > ( arr , separator , cb ) : ( TOut | TSep ) [ ] => { const result = [ ] ; for ( let i = 0 , len = arr . length ; i < len ; i ++ ) { if ( i > 0 ) result . push ( separator ) ; result . push ( cb ( arr [ i ] , i , arr ) ) ; } return result ; } ; const completeKeysWithDefault = < T extends object > ( objs ) : T [ ] => { const unionKeys = Object . assign ( { } , ... objs ) ; for ( let k in unionKeys ) unionKeys [ k ] = Default ; return objs . map ( o => Object . assign ( { } , unionKeys , o ) ) ; } ;