"use strict";
var __importDefault = (this && this.__importDefault) || function (mod) {
    return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.RDSClient = void 0;
const node_async_hooks_1 = require("node:async_hooks");
const node_util_1 = require("node:util");
const mysql_1 = __importDefault(require("mysql"));
const operator_1 = require("./operator");
const connection_1 = require("./connection");
const transaction_1 = require("./transaction");
const PoolConfig_1 = require("./PoolConfig");
const literals_1 = __importDefault(require("./literals"));
const channels_1 = __importDefault(require("./channels"));
class RDSClient extends operator_1.Operator {
    static get literals() { return literals_1.default; }
    static get escape() { return mysql_1.default.escape; }
    static get escapeId() { return mysql_1.default.escapeId; }
    static get format() { return mysql_1.default.format; }
    static get raw() { return mysql_1.default.raw; }
    static #DEFAULT_STORAGE_KEY = Symbol('RDSClient#storage#default');
    static #TRANSACTION_NEST_COUNT = Symbol('RDSClient#transaction#nestCount');
    #pool;
    #connectionStorage;
    #connectionStorageKey;
    constructor(options) {
        super();
        const { connectionStorage, connectionStorageKey, ...mysqlOptions } = options;
        // get connection options from getConnectionConfig method every time
        if (mysqlOptions.getConnectionConfig) {
            // eslint-disable-next-line @typescript-eslint/no-var-requires
            const MySQLPool = require('mysql/lib/Pool');
            this.#pool = new MySQLPool({
                config: new PoolConfig_1.RDSPoolConfig(mysqlOptions, mysqlOptions.getConnectionConfig),
            });
            // override _needsChangeUser to return false
            this.#pool._needsChangeUser = () => false;
        }
        else {
            this.#pool = mysql_1.default.createPool(mysqlOptions);
        }
        [
            'query',
            'getConnection',
            'end',
        ].forEach(method => {
            this.#pool[method] = (0, node_util_1.promisify)(this.#pool[method]);
        });
        this.#connectionStorage = connectionStorage || new node_async_hooks_1.AsyncLocalStorage();
        this.#connectionStorageKey = connectionStorageKey || RDSClient.#DEFAULT_STORAGE_KEY;
        // https://github.com/mysqljs/mysql#pool-events
        this.#pool.on('connection', (connection) => {
            channels_1.default.connectionNew.publish({
                client: this,
                connection,
            });
        });
        this.#pool.on('enqueue', () => {
            channels_1.default.connectionEnqueue.publish({
                client: this,
            });
        });
        this.#pool.on('acquire', (connection) => {
            channels_1.default.connectionAcquire.publish({
                client: this,
                connection,
            });
        });
        this.#pool.on('release', (connection) => {
            channels_1.default.connectionRelease.publish({
                client: this,
                connection,
            });
        });
    }
    async query(sql, values) {
        const conn = await this.getConnection();
        try {
            return await conn.query(sql, values);
        }
        finally {
            conn.release();
        }
    }
    get pool() {
        return this.#pool;
    }
    get stats() {
        const acquiringConnections = this.#pool._acquiringConnections.length;
        const allConnections = this.#pool._allConnections.length;
        const freeConnections = this.#pool._freeConnections.length;
        const connectionQueue = this.#pool._connectionQueue.length;
        const busyConnections = allConnections - freeConnections - acquiringConnections;
        return {
            acquiringConnections,
            allConnections,
            freeConnections,
            connectionQueue,
            busyConnections,
        };
    }
    async getConnection() {
        try {
            const _conn = await this.#pool.getConnection();
            const conn = new connection_1.RDSConnection(_conn);
            if (this.beforeQueryHandlers.length > 0) {
                for (const handler of this.beforeQueryHandlers) {
                    conn.beforeQuery(handler);
                }
            }
            if (this.afterQueryHandlers.length > 0) {
                for (const handler of this.afterQueryHandlers) {
                    conn.afterQuery(handler);
                }
            }
            return conn;
        }
        catch (err) {
            if (err.name === 'Error') {
                err.name = 'RDSClientGetConnectionError';
            }
            throw err;
        }
    }
    /**
     * Begin a transaction
     *
     * @return {RDSTransaction} transaction instance
     */
    async beginTransaction() {
        const conn = await this.getConnection();
        try {
            await conn.beginTransaction();
        }
        catch (err) {
            conn.release();
            throw err;
        }
        const tran = new transaction_1.RDSTransaction(conn);
        tran[RDSClient.#TRANSACTION_NEST_COUNT] = 1;
        if (this.beforeQueryHandlers.length > 0) {
            for (const handler of this.beforeQueryHandlers) {
                tran.beforeQuery(handler);
            }
        }
        if (this.afterQueryHandlers.length > 0) {
            for (const handler of this.afterQueryHandlers) {
                tran.afterQuery(handler);
            }
        }
        return tran;
    }
    /**
     * Auto commit or rollback on a transaction scope
     *
     * @param {Function} scope - scope with code
     * @param {Object} [ctx] - transaction context
     * @return {Object} - scope return result
     */
    async #beginTransactionScope(scope, ctx) {
        let tran;
        let shouldRelease = false;
        if (!ctx[this.#connectionStorageKey]) {
            // there is no transaction in ctx, create a new one
            tran = await this.beginTransaction();
            ctx[this.#connectionStorageKey] = tran;
            shouldRelease = true;
        }
        else {
            // use transaction in ctx
            tran = ctx[this.#connectionStorageKey];
            tran[RDSClient.#TRANSACTION_NEST_COUNT]++;
        }
        let result;
        let scopeError;
        let internalError;
        try {
            result = await scope(tran);
        }
        catch (err) {
            scopeError = err;
        }
        tran[RDSClient.#TRANSACTION_NEST_COUNT]--;
        // null connection means the nested scope has been rollback, we can do nothing here
        if (tran.conn) {
            try {
                // execution error, should rollback
                if (scopeError) {
                    await tran.rollback();
                }
                else if (tran[RDSClient.#TRANSACTION_NEST_COUNT] < 1) {
                    // nestedCount smaller than 1 means all the nested scopes have executed successfully
                    await tran.commit();
                }
            }
            catch (err) {
                internalError = err;
            }
        }
        // remove transaction in ctx
        if (shouldRelease && tran[RDSClient.#TRANSACTION_NEST_COUNT] < 1) {
            ctx[this.#connectionStorageKey] = null;
        }
        if (internalError) {
            if (scopeError) {
                internalError.cause = scopeError;
            }
            throw internalError;
        }
        if (scopeError) {
            throw scopeError;
        }
        return result;
    }
    /**
     * Auto commit or rollback on a transaction scope
     *
     * @param scope - scope with code
     * @return {Object} - scope return result
     */
    async beginTransactionScope(scope) {
        let ctx = this.#connectionStorage.getStore();
        if (ctx) {
            return await this.#beginTransactionScope(scope, ctx);
        }
        ctx = {};
        return await this.#connectionStorage.run(ctx, async () => {
            return await this.#beginTransactionScope(scope, ctx);
        });
    }
    /**
     * doomed to be rollbacked after transaction scope
     * useful on writing tests which are related with database
     *
     * @param scope - scope with code
     * @param ctx - transaction context
     * @return {Object} - scope return result
     */
    async #beginDoomedTransactionScope(scope, ctx) {
        let tran;
        if (!ctx[this.#connectionStorageKey]) {
            // there is no transaction in ctx, create a new one
            tran = await this.beginTransaction();
            ctx[this.#connectionStorageKey] = tran;
        }
        else {
            // use transaction in ctx
            tran = ctx[this.#connectionStorageKey];
            tran[RDSClient.#TRANSACTION_NEST_COUNT]++;
        }
        try {
            const result = await scope(tran);
            tran[RDSClient.#TRANSACTION_NEST_COUNT]--;
            if (tran[RDSClient.#TRANSACTION_NEST_COUNT] === 0) {
                ctx[this.#connectionStorageKey] = null;
                await tran.rollback();
            }
            return result;
        }
        catch (err) {
            if (ctx[this.#connectionStorageKey]) {
                ctx[this.#connectionStorageKey] = null;
                await tran.rollback();
            }
            throw err;
        }
    }
    /**
     * doomed to be rollbacked after transaction scope
     * useful on writing tests which are related with database
     *
     * @param scope - scope with code
     * @return {Object} - scope return result
     */
    async beginDoomedTransactionScope(scope) {
        let ctx = this.#connectionStorage.getStore();
        if (ctx) {
            return await this.#beginDoomedTransactionScope(scope, ctx);
        }
        ctx = {};
        return await this.#connectionStorage.run(ctx, async () => {
            return await this.#beginDoomedTransactionScope(scope, ctx);
        });
    }
    async end() {
        await this.#pool.end();
    }
}
exports.RDSClient = RDSClient;
//# sourceMappingURL=data:application/json;base64, 