import { Pool , PoolClient , PoolConfig , QueryResult , types } from "pg" ; import { parse } from "pg-connection-string" ; import { Async , CoatyObject , DbConnectionInfo , filterOp , ObjectFilterOperator , ObjectMatcher , Runtime , Uuid } from "../.." ; import { DbAdapterBase } from "../db-adapter" ; import { DbContext } from "../db-context" ; import { AggregateProperties , DbJoinCondition , DbObjectFilter , IQueryIterator , QueryIterator , SqlQueryBuilder , SqlQueryOptions , SqlQueryResultSet , } from "../db-operations" ; import { AggregateOp } from "../db-operators" ; types . setTypeParser ( 20 , value => parseInt ( value , 10 ) ) ; /**
 * Provides access to a PostgreSQL 9.5 (or later) database server. Supports
 * connection pooling and query iteration. Supports SQL operations
 * as well as NoSQL operations using the JSONB data type.
 *
 * The Postgres connection string has the following general form:
 * `postgres://<user>:<password>@<hostname>:<port>/<database>?ssl=true&application_name=PGAPPNAME&fallback_application_name=false`
 *
 * PostgreSQL supports the following connection options (with the given default values):
 * ```
 * {
 *    // database host
 *    host: process.env.PGHOST || 'localhost',
 *
 *    // database user's name
 *    user: process.env.PGUSER || process.env.USER
 *
 *    // name of database to connect
 *    database: process.env.PGDATABASE || process.env.USER
 *
 *    // database user's password
 *    password: process.env.PGPASSWORD || null,
 *
 *    // database port
 *    port: process.env.PGPORT || 5432,
 *
 *    // number of rows to return at a time from a prepared statement's
 *    // portal. 0 will return all rows at once
 *    rows: 0,
 *
 *    // binary result mode
 *    binary: false,
 *
 *    client_encoding: "",
 *
 *    ssl: false,
 *
 *    // TCP keep alive: this should help with backends incorrectly considering
 *    // idle clients to be dead and prematurely disconnecting them.
 *    keepAlive: false,
 *
 *    application_name: undefined,
 *
 *    fallback_application_name: undefined,
 *
 *    parseInputDatesAsUTC: false
 *
 *    // Pool Options (see also https://github.com/brianc/node-pg-pool and https://github.com/coopernurse/node-pool)
 *
 *    // maximum number of resources to create at any given time optional (default=1)
 *    max: 10
 *
 *    // minimum number of resources to keep in pool at any given time
 *    // if this is set >= max, the pool will silently set the min
 *    // to factory.max - 1 (Note: min==max case is expected to change in v3 release)
 *    // optional (default=0)
 *    min: 0
 *
 *    // max milliseconds a resource can go unused before it should be destroyed
 *    // (default 30000)
 *    idleTimeoutMillis: 30000
 *
 *    // return an error after given milliseconds if connection could not be established
 *    connectionTimeoutMillis: 1000
 *
 *    // If true, logs pool activities to console; can also be a function.
 *    // Note: true doesn't work properly in a Node.js process, use a function instead.
 *    log: true/false or function(msg: string, level: string) => {}
 *
 * }
 * ```
 *
 * This adapter supports the following adapter configuration options:
 * ```
 * {
 *    // The default chunk size of chunking fetch operations such as findObjects
 *    // or iquery (default 100). This value can be overridden in individual
 *    // fetch operations by specifying the DbObjectFilter.fetchSize property.
 *    defaultFetchSize: 100
 * }
 * ```
 *
 * This adapter supports the following extensions for creating/deleting users and databases dynamically:
 * ```
 * createUser
 * deleteUser
 * initDatabase
 * createDatabase
 * deleteDatabase
 * ```
 * Note that these operations must be performed on a database context with administrative privileges, i.e.
 * the connection info must specify a user/pwd with administration rights.
 */ export class PostgresAdapter extends DbAdapterBase { private static readonly PG_POOLS = new Map < DbConnectionInfo , PoolItem > ( ) ; private static readonly COLLECTION_SCHEMA = "collections" ; private static readonly DEFAULT_FETCH_SIZE = 100 ; private static readonly MINIMUM_FETCH_SIZE = 10 ; private _transactionClient : PooledClient ; constructor ( connectionInfo ) { super ( connectionInfo ) ; this . registerExtension ( "initDatabase" ) ; this . registerExtension ( "createDatabase" ) ; this . registerExtension ( "deleteDatabase" ) ; this . registerExtension ( "createUser" ) ; this . registerExtension ( "deleteUser" ) ; } addCollection ( name ) { const tableId = this . _getTableId ( name ) ; const indexId = asIdentifier ( name + "_object_idx" ) ; const sql = `do $$
                    BEGIN
                    CREATE SCHEMA IF NOT EXISTS ${ PostgresAdapter . COLLECTION_SCHEMA };
                    EXCEPTION when unique_violation then
                    END;
                    $$ language 'plpgsql';
                    CREATE TABLE IF NOT EXISTS ${ tableId } (
                       objectid uuid PRIMARY KEY,
                       object jsonb NOT NULL
                     );
                    CREATE INDEX IF NOT EXISTS ${ indexId }
                       ON ${ tableId } USING GIN(object jsonb_path_ops);` ; return this . _queryPooled ( sql ) . then ( result => { } ) ; } removeCollection ( name ) { const sql = `DROP TABLE IF EXISTS ${ this . _getTableId ( name ) };` ; return this . _queryPooled ( sql ) . then ( result => { } ) ; } clearCollection ( name ) { const sql = `TRUNCATE TABLE ${ this . _getTableId ( name ) };` ; return this . _queryPooled ( sql ) . then ( result => { } ) ; } insertObjects ( collectionName , objects , shouldReplaceExisting ) { const tableId = this . _getTableId ( collectionName ) ; const text = `INSERT INTO ${ tableId } (objectid, object)
                          SELECT (value->>'objectId')::uuid, value FROM jsonb_array_elements($1)
                          ON CONFLICT (objectid) ${ shouldReplaceExisting ? "DO UPDATE SET object = EXCLUDED.object" : "DO NOTHING" }
                          RETURNING objectid;` ; return this . _queryOnObjects ( objects , text ) ; } updateObjects ( collectionName , objects ) { const tableId = this . _getTableId ( collectionName ) ; const text = `UPDATE ${ tableId } 
                          SET object = objs.object
                          FROM (SELECT (value->>'objectId')::uuid as objectid, value as object FROM jsonb_array_elements($1)) as objs
                          WHERE ${ tableId }.objectid = objs.objectid
                          RETURNING ${ tableId }.objectid;` ; return this . _queryOnObjects ( objects , text ) ; } updateObjectProperty ( collectionName , objectIds , property , value , shouldCreateMissing = true ) { const tableId = this . _getTableId ( collectionName ) ; const values = this . _getQueryValuesOnObjects ( objectIds ) ; if ( values . length === 0 ) { return Promise . resolve ( ) ; } values . push ( [ property ] ) ; let text ; if ( value !== undefined ) { const createMissing = shouldCreateMissing ? "true" : "false" ; text = `UPDATE ${ tableId } 
                        SET object = jsonb_set(
                            object,
                            $2,
                            $3::jsonb,
                            ${ createMissing })
                        WHERE ${ tableId }.objectid in (SELECT value::uuid FROM jsonb_array_elements_text($1));` ; values . push ( JSON . stringify ( value ) ) ; } else { text = `UPDATE ${ tableId } 
                        SET object = object - $2
                        WHERE ${ tableId }.objectid in (SELECT value::uuid FROM jsonb_array_elements_text($1));` ; } return this . _queryPooled ( text , values ) ; } deleteObjectsById ( collectionName , objectIds ) { const tableId = this . _getTableId ( collectionName ) ; const text = `DELETE FROM ${ tableId }
                          WHERE objectid in (SELECT value::uuid FROM jsonb_array_elements_text($1))
                          RETURNING objectid;` ; return this . _queryOnObjects ( objectIds , text ) ; } deleteObjects ( collectionName , filter ? ) { const createQuery = info => { const tableId = this . _getTableId ( collectionName ) ; const [ whereClause , whereParams ] = this . _getWhereClauseAndParams ( filter , this . _supportsJsonpath ( info ) ) ; const text = `DELETE FROM ${ tableId }
                              ${ whereClause }
                              RETURNING objectid;` ; return [ text , whereParams ] ; } ; return this . _queryPooledLazy ( createQuery ) . then ( result => result . rows . length ) ; } findObjectById < T extends CoatyObject > ( collectionName , id , isExternalId = false ) { const createQuery = info => { const tableId = this . _getTableId ( collectionName ) ; let text ; let values ; if ( isExternalId ) { const filter = { conditions : [ "externalId" , filterOp . equals ( id ) ] } ; const [ whereClause , whereParams ] = this . _getWhereClauseAndParams ( filter , this . _supportsJsonpath ( info ) ) ; text = `SELECT object FROM ${ tableId } ${ whereClause };` ; values = whereParams ; } else { text = `SELECT object FROM ${ tableId } WHERE objectid = $1;` ; values = [ id ] ; } return [ text , values ] ; } ; return this . _queryPooledLazy ( createQuery ) . then ( result => result . rows . length === 0 ? undefined : result . rows [ 0 ] . object ) ; } findObjects < T extends CoatyObject > ( collectionName , filter ? , joinConditions ? ) { const fetchInChunks = filter && filter . shouldFetchInChunks ; const rowField = "object" ; const createQuery = info => { const tableId = this . _getTableId ( collectionName ) ; const [ whereClause , whereParams ] = this . _getWhereClauseAndParams ( filter , this . _supportsJsonpath ( info ) , fetchInChunks ) ; const orderBy = this . _getOrderByClause ( filter ) ; const limit = ! fetchInChunks && filter ? ( filter . take !== undefined ? Math . max ( 0 , filter . take ) : "ALL" ) : "ALL" ; const offset = ! fetchInChunks && filter ? ( filter . skip !== undefined ? Math . max ( 0 , filter . skip ) : 0 ) : 0 ; const text = `SELECT ${ rowField } FROM ${ tableId }
                              ${ whereClause }
                              ${ orderBy } LIMIT ${ limit } OFFSET ${ offset }` ; const query = this . _getJoinQuery ( tableId , text , rowField , joinConditions , filter ) ; return fetchInChunks ? [ query ] : [ query , whereParams ] ; } ; if ( fetchInChunks ) { const fetchSize = this . _getFetchSize ( filter && filter . fetchSize ) ; return this . _fetchInChunks < T > ( createQuery , rowField , fetchSize , filter && filter . take , filter && filter . skip ) ; } else { return this . _fetchOnce < T > ( createQuery , rowField ) ; } } countObjects ( collectionName , filter ? ) { const createQuery = info => { const tableId = this . _getTableId ( collectionName ) ; const [ whereClause , whereParams ] = this . _getWhereClauseAndParams ( filter , this . _supportsJsonpath ( info ) ) ; const text = `SELECT COUNT(objectid) FROM ${ tableId } ${ whereClause };` ; return [ text , whereParams ] ; } ; return this . _queryPooledLazy ( createQuery ) . then ( result => result . rows [ 0 ] . count ) ; } aggregateObjects ( collectionName , aggregateProps , aggregateOp , filter ? ) { const createQuery = info => { const propsArray = ObjectMatcher . getFilterProperties ( aggregateProps ) ; const tableId = this . _getTableId ( collectionName ) ; const [ aggFunc , aggType ] = this . _getAggregateParams ( aggregateOp ) ; const [ whereClause , whereParams ] = this . _getWhereClauseAndParams ( filter , this . _supportsJsonpath ( info ) ) ; const text = `SELECT ${ aggFunc }((${ this . _getObjectIndexer ( propsArray , true ) })::${ aggType }) AS result
                              FROM ${ tableId } ${ whereClause };` ; return [ text , whereParams ] ; } ; return this . _queryPooledLazy ( createQuery ) . then ( result => result . rows [ 0 ] . result === null ? undefined : result . rows [ 0 ] . result ) ; } query ( sql ) { const [ text , params ] = sql ( "$" , true , 0 , id => asIdentifier ( id ) , lit => asLiteral ( lit ) ) ; return this . _queryPooled ( text , params ) . then ( result => result ) ; } iquery ( sql , queryOptions ? ) { const [ text ] = sql ( "$" , true , 0 , id => asIdentifier ( id ) , lit => asLiteral ( lit ) , true ) ; const fetchSize = this . _getFetchSize ( queryOptions && queryOptions . fetchSize ) ; return this . _fetchInChunks < any > ( ( ) => [ text ] , undefined , fetchSize , queryOptions && queryOptions . take , queryOptions && queryOptions . skip ) ; } transaction < T > ( action ) { return new Promise < T > ( ( resolve , reject ) => { const transact = ( client ) => { Promise . resolve ( ) . then ( ( ) => { const ctx = new DbContext ( this . connectionInfo ) ; ( ctx . adapter as PostgresAdapter ) . _transactionClient = client ; return action ( ctx ) . then ( result => { ( ctx . adapter as PostgresAdapter ) . _transactionClient = undefined ; return result ; } ) . catch ( err => { ( ctx . adapter as PostgresAdapter ) . _transactionClient = undefined ; return Promise . reject ( err ) ; } ) ; } ) . then ( result => { this . _commitTransaction ( client , ( ) => resolve ( result ) , error => reject ( error ) ) ; } ) . catch ( err => { this . _rollbackTransaction ( client , err , error => reject ( error ) ) ; } ) ; } ; this . _getPooledClient ( true ) . then ( client => { this . _beginTransaction ( client , clnt => transact ( clnt ) , error => reject ( error ) ) ; } ) . catch ( error => reject ( error ) ) ; } ) ; } param dbInfo param collections param clearCollections initDatabase ( dbInfo , collections ? , clearCollections = false ) { const dbc = new DbContext ( dbInfo , PostgresAdapter ) ; const dbConfig = this . _getPoolOptions ( dbInfo ) ; return this . createUser ( dbConfig . user , dbConfig . password as string ) . then ( ( ) => this . createDatabase ( dbConfig . database , dbConfig . user ) ) . then ( ( ) => Async . inSeries ( collections , coll => dbc . addCollection ( coll ) ) ) . then ( ( ) => clearCollections && Async . inSeries ( collections , coll => dbc . clearCollection ( coll ) ) ) ; } param name param owner param comment createDatabase ( name , owner , comment ? ) { const dbId = asIdentifier ( name ) ; let sql = `SELECT true FROM pg_database WHERE datname = ${ asLiteral ( name ) };` ; return this . _queryPooled ( sql ) . then ( result => { if ( result . rows . length === 1 ) { return false ; } sql = `CREATE DATABASE ${ dbId }
                       WITH OWNER = ${ asIdentifier ( owner ) }
                       ENCODING = 'UTF8';` ; return this . _queryPooled ( sql ) . then ( ( ) => { if ( comment ) { sql = `COMMENT ON DATABASE ${ dbId } IS ${ asLiteral ( comment ) };` ; return this . _queryPooled ( sql ) . then ( res => true ) ; } return true ; } ) ; } ) ; } param name deleteDatabase ( name ) { const sql = `DROP DATABASE IF EXISTS ${ asIdentifier ( name ) };` ; return this . _queryPooled ( sql ) . then ( result => { } ) ; } param name param password param comment createUser ( name , password , comment ? ) { const userId = asIdentifier ( name ) ; let sql = `SELECT true FROM pg_roles WHERE rolname = ${ asLiteral ( name ) };` ; return this . _queryPooled ( sql ) . then ( result => { if ( result . rows . length === 1 ) { return false ; } sql = `CREATE ROLE ${ userId } LOGIN
                       PASSWORD ${ asLiteral ( password ) }
                       NOSUPERUSER INHERIT CREATEDB NOCREATEROLE NOREPLICATION;` ; return this . _queryPooled ( sql ) . then ( ( ) => { if ( comment ) { const cSql = `COMMENT ON ROLE ${ userId } IS ${ asLiteral ( comment ) };` ; return this . _queryPooled ( cSql ) . then ( res => true ) ; } return true ; } ) ; } ) ; } param name deleteUser ( name ) { const sql = `DROP ROLE IF EXISTS ${ asIdentifier ( name ) };` ; return this . _queryPooled ( sql ) . then ( result => { } ) ; } private _queryOnObjects ( objects , text ) { const values = this . _getQueryValuesOnObjects ( objects ) ; if ( values . length === 0 ) { return Promise . resolve ( [ ] ) ; } return this . _queryPooled ( text , values ) . then ( result => result . rows . map ( row => row . objectid ) ) ; } private _getQueryValuesOnObjects ( objects ) { if ( Array . isArray ( objects ) ) { return [ JSON . stringify ( objects ) ] ; } return [ JSON . stringify ( [ objects ] ) ] ; } private _getWhereClauseAndParams ( filter , supportsJsonpath , embedParams = false ) { if ( ! filter || ! filter . conditions ) { return [ "" , [ ] ] ; } if ( Array . isArray ( filter . conditions ) ) { filter . conditions = { and : [ filter . conditions ] } ; } const combineAnd = ! filter . conditions . or ; const clauses = [ ] ; const params = [ ] ; const contains = [ ] ; const notContains = [ ] ; let pi = 1 ; let p1 ; let p2 ; const para = ( param , index , isJsonpathFilterExpression = false ) => { return isJsonpathFilterExpression ? asJsonpathLiteral ( param , true ) : ( embedParams ? asLiteral ( param ) : `$${ index }` ) ; } ; const checkExistence = ( propsArray , isNegated = false ) => { if ( supportsJsonpath ) { clauses . push ( `(object @@ '${ isNegated ? "!" : "" }exists(${ this . _getJsonpathKeys ( propsArray ) })')` ) ; } else { if ( propsArray . length === 0 ) { if ( isNegated ) { clauses . push ( `(false)` ) ; } return ; } let clause = `object ? ${ asLiteral ( propsArray [ 0 ] ) }` ; for ( let ip = 1 ; ip < propsArray . length ; ip ++ ) { clause += ` AND ${ this . _getObjectIndexer ( propsArray . slice ( 0 , ip ) ) } ? ${ asLiteral ( propsArray [ ip ] ) }` ; } clauses . push ( `(${ isNegated ? "NOT (" : "" }${ clause }${ isNegated ? ")" : "" })` ) ; } } ; const checkContainments = ( containment , isNegated = false ) => { for ( const [ propsArray , opnd ] of containment ) { const obj = Object . create ( null ) ; let subobj = obj ; propsArray . forEach ( ( prop , index ) => { if ( index === propsArray . length - 1 ) { subobj [ prop ] = opnd ; return ; } subobj [ prop ] = Object . create ( null ) ; subobj = subobj [ prop ] ; } ) ; p1 = JSON . stringify ( obj ) ; clauses . push ( `(${ isNegated ? "NOT " : "" }object @> ${ para ( p1 , pi ++ ) }::jsonb)` ) ; params . push ( p1 ) ; } } ; for ( const cond of ( combineAnd ? filter . conditions . and : filter . conditions . or ) ) { const [ props , expr ] = cond ; const propsArray = ObjectMatcher . getFilterProperties ( props ) ; const op = expr [ 0 ] ; const opnd1 = expr [ 1 ] ; const opnd2 = expr [ 2 ] ; switch ( op ) { case ObjectFilterOperator . LessThan : p1 = JSON . stringify ( opnd1 ) ; if ( supportsJsonpath ) { clauses . push ( `(object @@ '${ this . _getJsonpathKeys ( propsArray ) } < ${ para ( p1 , pi , true ) }')` ) ; } else { clauses . push ( `(${ this . _getObjectIndexer ( propsArray ) } < ${ para ( p1 , pi ++ ) }::jsonb)` ) ; params . push ( p1 ) ; } break ; case ObjectFilterOperator . LessThanOrEqual : p1 = JSON . stringify ( opnd1 ) ; if ( supportsJsonpath ) { clauses . push ( `(object @@ '${ this . _getJsonpathKeys ( propsArray ) } <= ${ para ( p1 , pi , true ) }')` ) ; } else { clauses . push ( `(${ this . _getObjectIndexer ( propsArray ) } <= ${ para ( p1 , pi ++ ) }::jsonb)` ) ; params . push ( p1 ) ; } break ; case ObjectFilterOperator . GreaterThan : p1 = JSON . stringify ( opnd1 ) ; if ( supportsJsonpath ) { clauses . push ( `(object @@ '${ this . _getJsonpathKeys ( propsArray ) } > ${ para ( p1 , pi , true ) }')` ) ; } else { clauses . push ( `(${ this . _getObjectIndexer ( propsArray ) } > ${ para ( p1 , pi ++ ) }::jsonb)` ) ; params . push ( p1 ) ; } break ; case ObjectFilterOperator . GreaterThanOrEqual : p1 = JSON . stringify ( opnd1 ) ; if ( supportsJsonpath ) { clauses . push ( `(object @@ '${ this . _getJsonpathKeys ( propsArray ) } >= ${ para ( p1 , pi , true ) }')` ) ; } else { clauses . push ( `(${ this . _getObjectIndexer ( propsArray ) } >= ${ para ( p1 , pi ++ ) }::jsonb)` ) ; params . push ( p1 ) ; } break ; case ObjectFilterOperator . Between : p1 = JSON . stringify ( opnd1 ) ; p2 = JSON . stringify ( opnd2 ) ; if ( supportsJsonpath ) { const swapOpnds = ObjectMatcher . matchesFilter ( { value : opnd1 } as any as CoatyObject , { conditions : [ "value" , filterOp . greaterThan ( opnd2 ) ] } ) ; if ( swapOpnds ) { const ptmp = p1 ; p1 = p2 ; p2 = ptmp ; } clauses . push ( `(object @? '${ this . _getJsonpathKeys ( propsArray ) } ? (@ >= ${ para ( p1 , pi , true ) } && @ <= ${ para ( p2 , pi + 1 , true ) })')` ) ; } else { clauses . push ( `(${ this . _getObjectIndexer ( propsArray ) } BETWEEN SYMMETRIC ${ para ( p1 , pi ) }::jsonb AND ${ para ( p2 , pi + 1 ) }::jsonb)` ) ; pi += 2 ; params . push ( p1 ) ; params . push ( p2 ) ; } break ; case ObjectFilterOperator . NotBetween : p1 = JSON . stringify ( opnd1 ) ; p2 = JSON . stringify ( opnd2 ) ; if ( supportsJsonpath ) { const swapOpnds = ObjectMatcher . matchesFilter ( { value : opnd1 } as any as CoatyObject , { conditions : [ "value" , filterOp . greaterThan ( opnd2 ) ] } ) ; if ( swapOpnds ) { const ptmp = p1 ; p1 = p2 ; p2 = ptmp ; } clauses . push ( `(object @? '${ this . _getJsonpathKeys ( propsArray ) } ? (@ < ${ para ( p1 , pi , true ) } || @ > ${ para ( p2 , pi + 1 , true ) })')` ) ; } else { clauses . push ( `(${ this . _getObjectIndexer ( propsArray ) } NOT BETWEEN SYMMETRIC ${ para ( p1 , pi ) }::jsonb AND ${ para ( p2 , pi + 1 ) }::jsonb)` ) ; pi += 2 ; params . push ( p1 ) ; params . push ( p2 ) ; } break ; case ObjectFilterOperator . Like : p1 = opnd1 ; if ( supportsJsonpath ) { const likeRegex = like2PosixRegex ( p1 ) ; clauses . push ( `(object @@ '${ this . _getJsonpathKeys ( propsArray ) } like_regex ${ para ( JSON . stringify ( likeRegex ) , pi , true ) }')` ) ; } else { clauses . push ( `(${ this . _getObjectIndexer ( propsArray , true ) } LIKE ${ para ( p1 , pi ++ ) })` ) ; params . push ( p1 ) ; } break ; case ObjectFilterOperator . Equals : if ( opnd1 === undefined ) { checkExistence ( propsArray , true ) ; } else { if ( typeof opnd1 === "number" || typeof opnd1 === "string" || typeof opnd1 === "boolean" || ! opnd1 ) { if ( supportsJsonpath ) { p1 = JSON . stringify ( opnd1 ) ; clauses . push ( `(object @@ '${ this . _getJsonpathKeys ( propsArray ) } == ${ para ( p1 , pi , true ) }')` ) ; } else { contains . push ( [ propsArray , opnd1 ] ) ; } } else { p1 = JSON . stringify ( opnd1 ) ; clauses . push ( `(${ this . _getObjectIndexer ( propsArray ) } = ${ para ( p1 , pi ++ ) }::jsonb)` ) ; params . push ( p1 ) ; } } break ; case ObjectFilterOperator . NotEquals : if ( opnd1 === undefined ) { checkExistence ( propsArray , false ) ; } else { if ( typeof opnd1 === "number" || typeof opnd1 === "string" || typeof opnd1 === "boolean" || ! opnd1 ) { if ( supportsJsonpath ) { p1 = JSON . stringify ( opnd1 ) ; clauses . push ( `(object @@ '${ this . _getJsonpathKeys ( propsArray ) } != ${ para ( p1 , pi , true ) }')` ) ; } else { notContains . push ( [ propsArray , opnd1 ] ) ; } } else { p1 = JSON . stringify ( opnd1 ) ; clauses . push ( `(${ this . _getObjectIndexer ( propsArray ) } <> ${ para ( p1 , pi ++ ) }::jsonb)` ) ; params . push ( p1 ) ; } } break ; case ObjectFilterOperator . Exists : checkExistence ( propsArray , false ) ; break ; case ObjectFilterOperator . NotExists : checkExistence ( propsArray , true ) ; break ; case ObjectFilterOperator . Contains : contains . push ( [ propsArray , opnd1 ] ) ; break ; case ObjectFilterOperator . NotContains : notContains . push ( [ propsArray , opnd1 ] ) ; break ; case ObjectFilterOperator . In : case ObjectFilterOperator . NotIn : if ( ! Array . isArray ( opnd1 ) ) { throw new TypeError ( `operand for filter operator ${ op } must be an array` ) ; } const sqlOp = op === ObjectFilterOperator . In ? "in" : "not in" ; p1 = JSON . stringify ( opnd1 ) ; clauses . push ( `(${ this . _getObjectIndexer ( propsArray ) } ${ sqlOp } (SELECT value FROM jsonb_array_elements(${ para ( p1 , pi ++ ) }::jsonb)))` ) ; params . push ( p1 ) ; break ; default : throw new TypeError ( `unknown filter operator: ${ op }` ) ; } } checkContainments ( notContains , true ) ; checkContainments ( contains , false ) ; if ( clauses . length === 0 ) { return [ "" , [ ] ] ; } return [ "WHERE " + clauses . join ( combineAnd ? " AND " : " OR " ) , params ] ; } private _getOrderByClause ( filter , tableId ? ) { if ( ! filter || ! filter . orderByProperties ) { return "" ; } tableId = tableId ? tableId + "." : "" ; return "ORDER BY " + filter . orderByProperties . map ( ( order , i ) => `${ tableId }${ this . _getObjectIndexer ( ObjectMatcher . getFilterProperties ( order [ 0 ] ) ) } ${ this . _getSortOrder ( order [ 1 ] ) }` ) . join ( ", " ) ; } private _getObjectIndexer ( propsChain , asText = false ) { return propsChain . length === 1 ? `object ->${ asText ? ">" : "" }${ asLiteral ( propsChain [ 0 ] ) }` : `object #>${ asText ? ">" : "" }array[${ propsChain . map ( p => asLiteral ( p ) ) . join ( "," ) }]` ; } private _getJsonpathKeys ( propsChain ) { const keys = propsChain . map ( p => asJsonpathLiteral ( p , false ) ) . join ( "." ) ; return keys === "" ? `$` : `$.${ keys }` ; } private _getSortOrder ( order ) { return order === "Asc" ? "ASC" : "DESC" ; } private _getAggregateParams ( aggregateOp ) { switch ( aggregateOp ) { case AggregateOp . Avg : return [ "avg" , "float" ] ; case AggregateOp . Sum : return [ "sum" , "float" ] ; case AggregateOp . Max : return [ "max" , "float" ] ; case AggregateOp . Min : return [ "min" , "float" ] ; case AggregateOp . Every : return [ "bool_and" , "boolean" ] ; case AggregateOp . Some : return [ "bool_or" , "boolean" ] ; default : throw new TypeError ( `Unhandled aggregate operator ${ aggregateOp }` ) ; } } private _getJoinQuery ( tableId , query , resultColumn , joinConditions , filter ) { if ( ! joinConditions ) { return query + ";" ; } if ( ! Array . isArray ( joinConditions ) ) { joinConditions = [ joinConditions ] ; } if ( joinConditions . length === 0 ) { return query + ";" ; } const local = "local" ; const joinedSelectList = joinConditions . map ( ( c , i ) => { const isSingleResult = c . isOneToOneRelation && ! c . isLocalPropertyArray ; let aggExpr = `COALESCE(jsonb_agg(DISTINCT from${ i }.object ORDER BY from${ i }.object ASC) FILTER (WHERE from${ i }.object IS NOT NULL), '[]'::jsonb)` ; if ( isSingleResult ) { aggExpr += "->0" ; } const buildExpr = `jsonb_build_object(${ asLiteral ( c . asProperty ) }, ${ aggExpr })` ; return isSingleResult ? `jsonb_strip_nulls(${ buildExpr })` : buildExpr ; } ) ; const joinClauses = joinConditions . map ( ( c , i ) => { const localExpr = c . isLocalPropertyArray ? `IN (SELECT value FROM jsonb_array_elements(${ local }.object->${ asLiteral ( c . localProperty ) }))` : `= ${ local }.object->${ asLiteral ( c . localProperty ) }` ; return `LEFT OUTER JOIN ${ this . _getTableId ( c . fromCollection ) } AS from${ i }
                ON (from${ i }.object->${ asLiteral ( c . fromProperty ) } ${ localExpr })` ; } ) ; const orderBy = this . _getOrderByClause ( filter , local ) ; return `SELECT (${ local }.object || ${ joinedSelectList . join ( " || " ) }) as ${ resultColumn } 
                            FROM (${ query }) as ${ local }
                            ${ joinClauses . join ( "\n" ) }
                            GROUP BY ${ local }.object ${ orderBy };` ; } private _fetchInChunks < T > ( createQuery , rowField , fetchSize , take , skip ) { return new Promise < IQueryIterator < T > > ( ( resolve , reject ) => { let resultIndex = 0 ; let currentBatch = [ ] ; take = take !== undefined ? Math . max ( 0 , take ) : - 1 ; skip = skip !== undefined ? Math . max ( 0 , skip ) : 0 ; const fetchNextChunk = ( client , cursor , fetchCount , skipCount , iter , resolveDone , rejectDone ) => { client . readCursor ( cursor , fetchCount , skipCount ) . then ( results => { let isBreakInvoked = false ; let allTaken = false ; let result ; if ( Array . isArray ( results ) ) { result = results . find ( res => res . command === "FETCH" ) ; } else { result = results ; } const rowLen = result . rows . length ; let i = 0 ; while ( i < rowLen ) { const obj = ( rowField ? result . rows [ i ] [ rowField ] : result . rows [ i ] ) as T ; let dontBreak = true ; if ( iter . forEachAction ) { dontBreak = iter . forEachAction ( obj , resultIndex ) ; } if ( iter . forBatchAction ) { currentBatch . push ( obj ) ; if ( iter . batchSize > 0 && currentBatch . length === iter . batchSize ) { dontBreak = iter . forBatchAction ( currentBatch ) ; currentBatch = [ ] ; } } resultIndex ++ ; if ( dontBreak === false ) { isBreakInvoked = true ; break ; } if ( take > 0 && resultIndex >= take ) { allTaken = true ; break ; } i ++ ; } const readCompletely = rowLen < fetchCount ; if ( isBreakInvoked || allTaken || readCompletely ) { if ( iter . forBatchAction && ( currentBatch . length > 0 || resultIndex === 0 ) ) { if ( iter . forBatchAction ( currentBatch ) === false ) { isBreakInvoked = true ; } } client . closeCursor ( cursor , undefined , ( errc ) => { client . releaseIfNotTx ( errc ) ; resolveDone ( [ resultIndex , isBreakInvoked ] ) ; } ) ; return ; } fetchNextChunk ( client , cursor , fetchSize , 0 , iter , resolveDone , rejectDone ) ; } ) . catch ( error => { client . closeCursor ( cursor , error , ( errc ) => { client . releaseIfNotTx ( errc ) ; rejectDone ( error ) ; } ) ; } ) ; } ; const iterator = new QueryIterator < T > ( ) ; const donePromise = new Promise < [ number , boolean ] > ( ( resolveDone , rejectDone ) => { iterator . setStarter ( ( ) => { const cpromise = this . _transactionClient ? Promise . resolve ( this . _transactionClient ) : this . _getPooledClient ( ) ; let pc ; cpromise . then ( poolClient => { pc = poolClient ; return poolClient . queryCursorLazy ( createQuery ) ; } ) . then ( cursor => { if ( take === 0 ) { resolveDone ( [ 0 , false ] ) ; return ; } fetchNextChunk ( pc , cursor , fetchSize , skip , iterator , resolveDone , rejectDone ) ; } ) . catch ( error => rejectDone ( error ) ) ; } ) ; } ) ; iterator . setDonePromise ( donePromise ) ; resolve ( iterator ) ; } ) ; } private _fetchOnce < T > ( createQuery , rowField ) { return new Promise < IQueryIterator < T > > ( ( resolve , reject ) => { const iterator = new QueryIterator < T > ( ) ; const donePromise = new Promise < [ number , boolean ] > ( ( resolveDone , rejectDone ) => { iterator . setStarter ( ( ) => { this . _queryPooledLazy ( createQuery ) . then ( result => { const rows = result . rows ; let resultIndex = 0 ; let currentBatch = [ ] ; let isBreakInvoked = false ; const rowLen = rows . length ; let i = 0 ; while ( i < rowLen ) { const obj = ( rowField ? rows [ i ] [ rowField ] : rows [ i ] ) as T ; let dontBreak = true ; if ( iterator . forEachAction ) { dontBreak = iterator . forEachAction ( obj , resultIndex ) ; } if ( iterator . forBatchAction ) { currentBatch . push ( obj ) ; if ( iterator . batchSize > 0 && currentBatch . length === iterator . batchSize ) { dontBreak = iterator . forBatchAction ( currentBatch ) ; currentBatch = [ ] ; } } resultIndex ++ ; if ( dontBreak === false ) { isBreakInvoked = true ; break ; } i ++ ; } if ( iterator . forBatchAction && ( currentBatch . length > 0 || resultIndex === 0 ) ) { if ( iterator . forBatchAction ( currentBatch ) === false ) { isBreakInvoked = true ; } } resolveDone ( [ resultIndex , isBreakInvoked ] ) ; } ) . catch ( error => rejectDone ( error ) ) ; } ) ; } ) ; iterator . setDonePromise ( donePromise ) ; resolve ( iterator ) ; } ) ; } private _beginTransaction ( client , action , errorAction ) { if ( this . _transactionClient !== undefined ) { client . release ( ) ; throw new Error ( "Nested transactions (subtransactions or savepoints) are not supported." ) ; } client . query ( "BEGIN" ) . then ( ( ) => action ( client ) , err => { client . release ( err ) ; errorAction ( err ) ; } ) ; } private _commitTransaction ( client , action , errorAction ) { client . query ( "COMMIT" ) . then ( ( ) => { client . release ( ) ; action ( ) ; } , err => { client . release ( err ) ; errorAction ( err ) ; } ) ; } private _rollbackTransaction ( client , error , errorAction ) { client . query ( "ROLLBACK" ) . then ( ( ) => { client . release ( ) ; errorAction ( error ) ; } , err => { client . release ( err ) ; errorAction ( error ) ; } ) ; } private _getTableId ( name ) { return PostgresAdapter . COLLECTION_SCHEMA + "." + asIdentifier ( name ) ; } private _getFetchSize ( filterFetchSize ) { return Math . max ( PostgresAdapter . MINIMUM_FETCH_SIZE , filterFetchSize || ( this . connectionInfo . adapterOptions && this . connectionInfo . adapterOptions . defaultFetchSize ) || PostgresAdapter . DEFAULT_FETCH_SIZE ) ; } private _supportsJsonpath ( info ) { return info . majorVersion >= 12 ; } private _queryPooled ( text , values ? ) { if ( this . _transactionClient ) { return this . _transactionClient . query ( text , values ) ; } return this . _getPoolItem ( ) . then ( item => item . pool . query ( text , values ) ) ; } private _queryPooledLazy ( createQuery ) { if ( this . _transactionClient ) { return this . _transactionClient . query ( ... createQuery ( this . _transactionClient . poolItem . serverInfo ) ) ; } return this . _getPoolItem ( ) . then ( item => item . pool . query ( ... createQuery ( item . serverInfo ) ) ) ; } private _getPooledClient ( isTransactionClient = false ) { return this . _getPoolItem ( ) . then ( item => item . pool . connect ( ) . then ( client => new PooledClient ( client , item , isTransactionClient ) ) ) ; } private _getPoolItem ( ) { let poolItem = PostgresAdapter . PG_POOLS . get ( this . connectionInfo ) ; if ( poolItem !== undefined ) { return Promise . resolve ( poolItem ) ; } poolItem = { pool : new Pool ( this . _getPoolOptions ( this . connectionInfo ) ) , serverInfo : { majorVersion : 0 , minorVersion : 0 } , } ; poolItem . pool . on ( "error" , ( error , client ) => { console . log ( `PostgresAdapter: pool idle client error: ${ error . message }` ) ; } ) ; return poolItem . pool . query ( "show server_version" ) . then ( result => { const [ major , minor ] = ( result . rows [ 0 ] . server_version as string ) . split ( "." ) ; poolItem . serverInfo . majorVersion = parseInt ( major , 10 ) ; poolItem . serverInfo . minorVersion = parseInt ( minor , 10 ) ; PostgresAdapter . PG_POOLS . set ( this . connectionInfo , poolItem ) ; return poolItem ; } ) ; } private _getPoolOptions ( connectionInfo ) { const connString = connectionInfo . connectionString ; let config = { } ; if ( connString && connString . length > 0 ) { config = parse ( connString ) ; } if ( connectionInfo . connectionOptions ) { Object . assign ( config , connectionInfo . connectionOptions ) ; } return config as PoolConfig ; } } type CreateQuery = ( info : ServerInfo ) => [ text : string , values ? : any [ ] ] ; interface ServerInfo { majorVersion : number ; minorVersion : number ; } interface PoolItem { pool : Pool ; serverInfo : ServerInfo ; } class PooledClient { constructor ( private _client , public poolItem , public isTransactionClient = false ) { } query ( text , values ? ) { return new Promise < QueryResult > ( ( resolve , reject ) => { const callback = ( err , result ) => { err ? reject ( err ) : resolve ( result ) ; } ; this . _client . query ( text , values || [ ] , callback ) ; } ) ; } queryCursorLazy ( createQuery ) { const [ text ] = createQuery ( this . poolItem . serverInfo ) ; const cursor = asIdentifier ( Runtime . newUuid ( ) ) ; const txQuery = this . isTransactionClient ? Promise . resolve ( { } as QueryResult ) : this . _client . query ( "BEGIN" ) ; return txQuery . then ( ( ) => this . _client . query ( `DECLARE ${ cursor } NO SCROLL CURSOR WITHOUT HOLD FOR ${ text }` ) ) . then ( ( ) => cursor ) ; } readCursor ( cursor , take , skip ? ) { let text = `FETCH ${ take } FROM ${ cursor };` ; if ( skip ) { text = `MOVE ${ skip } FROM ${ cursor }; ` + text ; } return this . _client . query ( text ) ; } closeCursor ( cursor , error , closeAction ) { const p = this . isTransactionClient ? this . _client . query ( `CLOSE ${ cursor };` ) : this . _client . query ( error ? "ROLLBACK;" : "COMMIT;" ) ; return p . then ( ( ) => closeAction ( ) ) . catch ( err => closeAction ( err ) ) ; } /**
     * Release the client to the pool. This is equivalent to calling the `done`
     * DoneCallback if Pool.connect is used with a connect callback instead of a
     * promise.
     *
     * If the error parameter is specified, the client instance is destroyed
     * instead of being returned back to the pool. This is e.g. useful if a
     * transaction error occurs on rollback. In this case the client instance
     * should no longer be used.
     */ release ( err ? ) { this . _client . release ( err ) ; } /**
     * Release the client to the pool if not inside a transaction block. Note
     * that open cursors (see _fetchWithCursor) are implicitly closed at
     * transaction end by the database server. Hopefully, this also affects
     * cursors that yielded an error on reading or closing. Internally, the
     * connection is kept from hanging when a cursor operation fails.
     *
     */ releaseIfNotTx ( err ? ) { if ( ! this . isTransactionClient ) { this . release ( err ) ; } } } remarks param like returns function like2PosixRegex ( likePattern ) { let regexStr = "^" ; let isEscaped = false ; for ( const c of likePattern ) { if ( c === "\\" ) { if ( isEscaped ) { isEscaped = false ; regexStr += "\\\\" ; } else { isEscaped = true ; } continue ; } if ( ".*+?^${}()|[]" . indexOf ( c ) !== - 1 ) { regexStr += "\\" + c ; isEscaped = false ; continue ; } if ( c === "_" && ! isEscaped ) { regexStr += "." ; continue ; } if ( c === "%" && ! isEscaped ) { regexStr += ".*" ; continue ; } regexStr += c ; isEscaped = false ; } regexStr += "$" ; return regexStr ; } param text function asIdentifier ( text ) { if ( typeof text !== "string" ) { text = JSON . stringify ( text ) ; } const escape = "\"" ; const len = text . length ; let escaped = escape ; for ( let i = 0 ; i < len ; i ++ ) { const c = text [ i ] ; if ( c === escape ) { escaped += c + c ; } else { escaped += c ; } } escaped += escape ; return escaped ; } param text function asLiteral ( text ) { if ( typeof text !== "string" ) { text = JSON . stringify ( text ) ; } const escape = "'" ; const len = text . length ; let hasBackslash = false ; let escaped = escape ; for ( let i = 0 ; i < len ; i ++ ) { const c = text [ i ] ; if ( c === escape ) { escaped += c + c ; } else if ( c === "\\" ) { escaped += c + c ; hasBackslash = true ; } else { escaped += c ; } } escaped += escape ; if ( hasBackslash === true ) { escaped = " E" + escaped ; } return escaped ; } param text param isFilterExpression function asJsonpathLiteral ( text , isFilterExpression ) { const escape = "\"" ; const backslash = "\\" ; const len = text . length ; let escaped = isFilterExpression ? "" : escape ; for ( let i = 0 ; i < len ; i ++ ) { const c = text [ i ] ; if ( c === "'" ) { escaped += c + c ; } else if ( ! isFilterExpression && ( c === backslash || c === escape ) ) { escaped += backslash + c ; } else { escaped += c ; } } if ( ! isFilterExpression ) { escaped += escape ; } return escaped ; }