lib/src/orm/connection/PostgresConnection.ts

import {Connection, ConnectionNotReadyError} from './Connection'
import {Client} from 'pg'
import {Inject} from '../../di'
import {QueryResult} from '../types'
import {Awaitable, collect} from '../../util'
import {SQLDialect} from '../dialect/SQLDialect'
import {PostgreSQLDialect} from '../dialect/PostgreSQLDialect'
import {Logging} from '../../service/Logging'
import {Schema} from '../schema/Schema'
import {PostgresSchema} from '../schema/PostgresSchema'

/**
 * Type interface representing the config for a PostgreSQL connection.
 */
export interface PostgresConnectionConfig {
    user: string,
    host: string,
    database: string,
    password?: string,
    port?: string,
}

/**
 * An implementation of a database Connection for dealing with PostgreSQL servers.
 */
export class PostgresConnection extends Connection {
    @Inject()
    protected readonly logging!: Logging

    /** The `pg` database client. */
    protected client?: Client

    public dialect(): SQLDialect {
        return <PostgreSQLDialect> this.app().make(PostgreSQLDialect)
    }

    public async init(): Promise<void> {
        this.logging.debug(`Initializing PostgreSQL connection ${this.name}...`)
        this.client = new Client(this.config)
        await this.client.connect()
    }

    public async close(): Promise<void> {
        this.logging.debug(`Closing PostgreSQL connection ${this.name}...`)
        if ( this.client ) {
            await this.client.end()
        }
    }

    public async query(query: string): Promise<QueryResult> {
        if ( !this.client ) {
            throw new ConnectionNotReadyError(this.name, { config: JSON.stringify(this.config) })
        }
        this.logging.verbose(`Executing query in connection ${this.name}: \n${query.split('\n').map(x => '    ' + x)
            .join('\n')}`)

        try {
            const result = await this.client.query(query)
            await this.queryExecuted(query)

            return {
                rows: collect(result.rows),
                rowCount: result.rowCount,
            }
        } catch (e) {
            if ( e instanceof Error ) {
                throw this.app().errorWrapContext(e, {
                    query,
                    connection: this.name,
                })
            }

            throw e
        }
    }

    public async asTransaction<T>(closure: () => Awaitable<T>): Promise<T> {
        if ( !this.client ) {
            throw new ConnectionNotReadyError(this.name, { config: JSON.stringify(this.config) })
        }

        await this.client.query('BEGIN')
        try {
            const result = await closure()
            await this.client.query('COMMIT')
            return result
        } catch (e) {
            await this.client.query('ROLLBACK')

            if ( e instanceof Error ) {
                throw this.app().errorWrapContext(e, {
                    connection: this.name,
                })
            }

            throw e
        }
    }

    public schema(name?: string): Schema {
        return new PostgresSchema(this, name)
    }
}