typeorm / typeorm / 23098575398

import type { ObjectLiteral } from "../../common/ObjectLiteral"

import { TypeORMError } from "../../error"

import { QueryFailedError } from "../../error/QueryFailedError"

import { QueryRunnerAlreadyReleasedError } from "../../error/QueryRunnerAlreadyReleasedError"

import { TransactionNotStartedError } from "../../error/TransactionNotStartedError"

import type { ReadStream } from "../../platform/PlatformTools"

import { BaseQueryRunner } from "../../query-runner/BaseQueryRunner"

import { QueryLock } from "../../query-runner/QueryLock"

import { QueryResult } from "../../query-runner/QueryResult"

import type { QueryRunner } from "../../query-runner/QueryRunner"

import type { TableIndexOptions } from "../../schema-builder/options/TableIndexOptions"

import { Table } from "../../schema-builder/table/Table"

import { TableCheck } from "../../schema-builder/table/TableCheck"

import { TableColumn } from "../../schema-builder/table/TableColumn"

import type { TableExclusion } from "../../schema-builder/table/TableExclusion"

import { TableForeignKey } from "../../schema-builder/table/TableForeignKey"

import { TableIndex } from "../../schema-builder/table/TableIndex"

import { TableUnique } from "../../schema-builder/table/TableUnique"

import { View } from "../../schema-builder/view/View"

import { Broadcaster } from "../../subscriber/Broadcaster"

import { BroadcasterResult } from "../../subscriber/BroadcasterResult"

import { InstanceChecker } from "../../util/InstanceChecker"

import { OrmUtils } from "../../util/OrmUtils"

import { Query } from "../Query"

import type { ColumnType } from "../types/ColumnTypes"

import type { IsolationLevel } from "../types/IsolationLevel"

import { MetadataTableType } from "../types/MetadataTableType"

import type { ReplicationMode } from "../types/ReplicationMode"

import type { MssqlParameter } from "./MssqlParameter"

import type { SqlServerDriver } from "./SqlServerDriver"

/**

 * Runs queries on a single SQL Server database connection.

 */

export class SqlServerQueryRunner

    extends BaseQueryRunner

    implements QueryRunner

{

    // -------------------------------------------------------------------------

    // Public Implemented Properties

    // -------------------------------------------------------------------------

    /**

     * Database driver used by connection.

     */

    driver: SqlServerDriver

    // -------------------------------------------------------------------------

    // Private Properties

    // -------------------------------------------------------------------------

    private lock: QueryLock = new QueryLock()

    // -------------------------------------------------------------------------

    // Constructor

    // -------------------------------------------------------------------------

    constructor(driver: SqlServerDriver, mode: ReplicationMode) {

        super()

        this.driver = driver

        this.connection = driver.connection

        this.broadcaster = new Broadcaster(this)

        this.mode = mode

    }

    // -------------------------------------------------------------------------

    // Public Methods

    // -------------------------------------------------------------------------

    /**

     * Creates/uses database connection from the connection pool to perform further operations.

     * Returns obtained database connection.

     */

    connect(): Promise<void> {

        return Promise.resolve()

    }

    /**

     * Releases used database connection.

     * You cannot use query runner methods once its released.

     */

    release(): Promise<void> {

        this.isReleased = true

        return Promise.resolve()

    }

    /**

     * Starts transaction.

     * @param isolationLevel

     */

    async startTransaction(isolationLevel?: IsolationLevel): Promise<void> {

        this.isTransactionActive = true

        try {

            await this.broadcaster.broadcast("BeforeTransactionStart")

        await new Promise<void>(async (ok, fail) => {

            const transactionCallback = (err: any) => {

                ok()

            }

            if (this.transactionDepth === 0) {

                const pool = await (this.mode === "slave"

                    : this.driver.obtainMasterConnection())

                this.databaseConnection = pool.transaction()

                this.connection.logger.logQuery("BEGIN TRANSACTION")

                if (isolationLevel) {

                    this.databaseConnection.begin(

                        this.convertIsolationLevel(isolationLevel),

                        transactionCallback,

                    )

                    this.connection.logger.logQuery(

                        "SET TRANSACTION ISOLATION LEVEL " + isolationLevel,

                    )

                } else {

                    this.databaseConnection.begin(transactionCallback)

                }

            } else {

                await this.query(

                    `SAVE TRANSACTION typeorm_${this.transactionDepth}`,

                )

                ok()

            }

            this.transactionDepth += 1

        })

        await this.broadcaster.broadcast("AfterTransactionStart")

    }

    /**

     * Commits transaction.

     * Error will be thrown if transaction was not started.

     */

    async commitTransaction(): Promise<void> {

        await this.broadcaster.broadcast("BeforeTransactionCommit")

        if (this.transactionDepth === 1) {

            return new Promise<void>((ok, fail) => {

                this.databaseConnection.commit(async (err: any) => {

                    this.isTransactionActive = false

                    this.databaseConnection = null

                    await this.broadcaster.broadcast("AfterTransactionCommit")

                    ok()

                    this.connection.logger.logQuery("COMMIT")

                    this.transactionDepth -= 1

                })

            })

        }

        this.transactionDepth -= 1

    }

    /**

     * Rollbacks transaction.

     * Error will be thrown if transaction was not started.

     */

    async rollbackTransaction(): Promise<void> {

        await this.broadcaster.broadcast("BeforeTransactionRollback")

        if (this.transactionDepth > 1) {

            await this.query(

                `ROLLBACK TRANSACTION typeorm_${this.transactionDepth - 1}`,

            )

            this.transactionDepth -= 1

        } else {

            return new Promise<void>((ok, fail) => {

                this.databaseConnection.rollback(async (err: any) => {

                    if (err) return fail(err)

                    this.isTransactionActive = false

                    this.databaseConnection = null

                    await this.broadcaster.broadcast("AfterTransactionRollback")

                    ok()

                    this.connection.logger.logQuery("ROLLBACK")

                    this.transactionDepth -= 1

                })

            })

        }

    }

    /**

     * Executes a given SQL query.

     * @param query

     * @param parameters

     * @param useStructuredResult

     */

    async query(

        query: string,

        parameters?: any[],

        useStructuredResult = false,

    ): Promise<any> {

        const release = await this.lock.acquire()

        this.driver.connection.logger.logQuery(query, parameters, this)

        await this.broadcaster.broadcast("BeforeQuery", query, parameters)

        const broadcasterResult = new BroadcasterResult()

        try {

            const pool = await (this.mode === "slave"

                ? this.driver.obtainSlaveConnection()

                : this.driver.obtainMasterConnection())

            const request = new this.driver.mssql.Request(

                this.isTransactionActive ? this.databaseConnection : pool,

            )

            if (parameters && parameters.length) {

                parameters.forEach((parameter, index) => {

                    const parameterName = index.toString()

                    if (InstanceChecker.isMssqlParameter(parameter)) {

                        const mssqlParameter =

                            this.mssqlParameterToNativeParameter(parameter)

                        if (mssqlParameter) {

                            request.input(

                                parameterName,

                                mssqlParameter,

                                parameter.value,

                            )

                        } else {

                            request.input(parameterName, parameter.value)

                        }

                    } else {

                        request.input(parameterName, parameter)

                    }

                })

            }

            const queryStartTime = Date.now()

            const raw = await new Promise<any>((ok, fail) => {

                request.query(query, (err: any, raw: any) => {

                    // log slow queries if maxQueryExecution time is set

                    const maxQueryExecutionTime =

                        this.driver.options.maxQueryExecutionTime

                    const queryEndTime = Date.now()

                    const queryExecutionTime = queryEndTime - queryStartTime

                    this.broadcaster.broadcastAfterQueryEvent(

                        broadcasterResult,

                        query,

                        parameters,

                        true,

                        queryExecutionTime,

                        raw,

                        undefined,

                    )

                    if (

                    if (err) {

                        fail(new QueryFailedError(query, parameters, err))

                    }

                    ok(raw)

                })

            })

            const result = new QueryResult()

            if (raw?.hasOwnProperty("recordset")) {

                result.records = raw.recordset

            }

            if (raw?.hasOwnProperty("rowsAffected")) {

                result.affected = raw.rowsAffected[0]

            }

            const queryType = query.slice(0, query.indexOf(" "))

            switch (queryType) {

                case "DELETE":

                    // for DELETE query additionally return number of affected rows

                    result.raw = [raw.recordset, raw.rowsAffected[0]]

                    break

                default:

                    result.raw = raw.recordset

            }

            if (useStructuredResult) {

                return result

            } else {

                return result.raw

            }

        } catch (err) {

            this.driver.connection.logger.logQueryError(

                err,

                query,

                parameters,

                this,

            )

            this.broadcaster.broadcastAfterQueryEvent(

                broadcasterResult,

                query,

                parameters,

                false,

                undefined,

                undefined,

                err,

            )

            throw err

            await broadcasterResult.wait()

            release()

        }

    }

    /**

     * Returns raw data stream.

     * @param query

     * @param parameters

     * @param onEnd

     * @param onError

     */

    async stream(

        query: string,

        parameters?: any[],

        onEnd?: Function,

        onError?: Function,

    ): Promise<ReadStream> {

        const release = await this.lock.acquire()

        this.driver.connection.logger.logQuery(query, parameters, this)

        const pool = await (this.mode === "slave"

            : this.driver.obtainMasterConnection())

        const request = new this.driver.mssql.Request(

        )

        request.query(query)

        const streamRequest = request.toReadableStream()

        streamRequest.on("error", (err: any) => {

        })

        streamRequest.on("end", () => {

            release()

        })

        return streamRequest

    }

    /**

     * Returns all available database names including system databases.

     */

    async getDatabases(): Promise<string[]> {

    /**

     * Returns all available schema names including system schemas.

     * If database parameter specified, returns schemas of that database.

     * @param database

     */

    async getSchemas(database?: string): Promise<string[]> {

    /**

     * Checks if database with the given name exist.

     * @param database

     */

    async hasDatabase(database: string): Promise<boolean> {

        const result = await this.query(`SELECT DB_ID(@0) as "db_id"`, [

            database,

        ])

        const dbId = result[0]["db_id"]

        return !!dbId

    }

    /**

     * Loads currently using database

     */

    async getCurrentDatabase(): Promise<string> {

        const currentDBQuery = await this.query(`SELECT DB_NAME() AS "db_name"`)

        return currentDBQuery[0]["db_name"]

    }

    /**

     * Checks if schema with the given name exist.

     * @param schema

     */

    async hasSchema(schema: string): Promise<boolean> {

        const result = await this.query(`SELECT SCHEMA_ID(@0) as "schema_id"`, [

            schema,

        ])

        const schemaId = result[0]["schema_id"]

        return !!schemaId

    }

    /**

     * Loads currently using database schema

     */

    async getCurrentSchema(): Promise<string> {

        const currentSchemaQuery = await this.query(

            `SELECT SCHEMA_NAME() AS "schema_name"`,

        )

        return currentSchemaQuery[0]["schema_name"]

    }

    /**

     * Checks if table with the given name exist in the database.

     * @param tableOrName

     */

    async hasTable(tableOrName: Table | string): Promise<boolean> {

        const parsedTableName = this.driver.parseTableName(tableOrName)

        const escapedDatabase = parsedTableName.database.replace(/"/g, '""')

        const sql = `SELECT * FROM "${escapedDatabase}"."INFORMATION_SCHEMA"."TABLES" WHERE "TABLE_NAME" = @0 AND "TABLE_SCHEMA" = @1`

        const result = await this.query(sql, [

            parsedTableName.tableName,

            parsedTableName.schema,

        ])

        return result.length ? true : false

    }

    /**

     * Checks if column exist in the table.

     * @param tableOrName

     * @param columnName

     */

    async hasColumn(

        tableOrName: Table | string,

        columnName: string,

    ): Promise<boolean> {

        const parsedTableName = this.driver.parseTableName(tableOrName)

        const escapedDatabase = parsedTableName.database.replace(/"/g, '""')

        const sql = `SELECT * FROM "${escapedDatabase}"."INFORMATION_SCHEMA"."COLUMNS" WHERE "TABLE_NAME" = @0 AND "TABLE_SCHEMA" = @1 AND "COLUMN_NAME" = @2`

        const result = await this.query(sql, [

            parsedTableName.tableName,

            parsedTableName.schema,

            columnName,

        ])

        return result.length ? true : false

    }

    /**

     * Creates a new database.

     * @param database

     * @param ifNotExists

     */

    async createDatabase(

        database: string,

        ifNotExists?: boolean,

    ): Promise<void> {

        const up = ifNotExists

            ? `IF DB_ID('${database}') IS NULL CREATE DATABASE "${database}"`

        const down = `DROP DATABASE "${database}"`

        await this.executeQueries(new Query(up), new Query(down))

    }

    /**

     * Drops database.

     * @param database

     * @param ifExists

     */

    async dropDatabase(database: string, ifExists?: boolean): Promise<void> {

        const up = ifExists

            ? `IF DB_ID('${database}') IS NOT NULL DROP DATABASE "${database}"`

            : `DROP DATABASE "${database}"`

        const down = `CREATE DATABASE "${database}"`

        await this.executeQueries(new Query(up), new Query(down))

    }

    /**

     * Creates table schema.

     * If database name also specified (e.g. 'dbName.schemaName') schema will be created in specified database.

     * @param schemaPath

     * @param ifNotExists

     */

    async createSchema(

        schemaPath: string,

        ifNotExists?: boolean,

    ): Promise<void> {

        const upQueries: Query[] = []

        const downQueries: Query[] = []

        if (schemaPath.indexOf(".") === -1) {

            const upQuery = ifNotExists

                ? `IF SCHEMA_ID('${schemaPath}') IS NULL BEGIN EXEC ('CREATE SCHEMA "${schemaPath}"') END`

            upQueries.push(new Query(upQuery))

            downQueries.push(new Query(`DROP SCHEMA "${schemaPath}"`))

        } else {

            const dbName = schemaPath.split(".")[0]

            const schema = schemaPath.split(".")[1]

            const currentDB = await this.getCurrentDatabase()

            upQueries.push(new Query(`USE "${dbName}"`))

            downQueries.push(new Query(`USE "${currentDB}"`))

            const upQuery = ifNotExists

                ? `IF SCHEMA_ID('${schema}') IS NULL BEGIN EXEC ('CREATE SCHEMA "${schema}"') END`

            upQueries.push(new Query(upQuery))

            downQueries.push(new Query(`DROP SCHEMA "${schema}"`))

            upQueries.push(new Query(`USE "${currentDB}"`))

            downQueries.push(new Query(`USE "${dbName}"`))

        }

        await this.executeQueries(upQueries, downQueries)

    }

    /**

     * Drops table schema.

     * If database name also specified (e.g. 'dbName.schemaName') schema will be dropped in specified database.

     * @param schemaPath

     * @param ifExists

     */

    async dropSchema(schemaPath: string, ifExists?: boolean): Promise<void> {

        const upQueries: Query[] = []

        const downQueries: Query[] = []

        if (schemaPath.indexOf(".") === -1) {

            const upQuery = ifExists

                ? `IF SCHEMA_ID('${schemaPath}') IS NOT NULL BEGIN EXEC ('DROP SCHEMA "${schemaPath}"') END`

                : `DROP SCHEMA "${schemaPath}"`

            upQueries.push(new Query(upQuery))

            downQueries.push(new Query(`CREATE SCHEMA "${schemaPath}"`))

        await this.executeQueries(upQueries, downQueries)

    }

    /**

     * Creates a new table.

     * @param table

     * @param ifNotExists

     * @param createForeignKeys

     * @param createIndices

     */

    async createTable(

        table: Table,

        ifNotExists: boolean = false,

        createForeignKeys: boolean = true,

        createIndices: boolean = true,

    ): Promise<void> {

        if (ifNotExists) {

            const isTableExist = await this.hasTable(table)

            if (isTableExist) return Promise.resolve()

        }

        const upQueries: Query[] = []

        const downQueries: Query[] = []

        upQueries.push(this.createTableSql(table, createForeignKeys))

        downQueries.push(this.dropTableSql(table))

        // if createForeignKeys is true, we must drop created foreign keys in down query.

        // createTable does not need separate method to create foreign keys, because it create fk's in the same query with table creation.

        if (createForeignKeys)

            table.foreignKeys.forEach((foreignKey) =>

                downQueries.push(this.dropForeignKeySql(table, foreignKey)),

            )

        if (createIndices) {

            table.indices.forEach((index) => {

                // new index may be passed without name. In this case we generate index name manually.

                if (!index.name)

                    index.name = this.connection.namingStrategy.indexName(

                        table,

                        index.columnNames,

                        index.where,

                    )

                upQueries.push(this.createIndexSql(table, index))

                downQueries.push(this.dropIndexSql(table, index))

            })

        }

        // if table have column with generated type, we must add the expression to the metadata table

        const generatedColumns = table.columns.filter(

            (column) => column.generatedType && column.asExpression,

        )

        for (const column of generatedColumns) {

            const parsedTableName = this.driver.parseTableName(table)

            const insertQuery = this.insertTypeormMetadataSql({

                database: parsedTableName.database,

                schema: parsedTableName.schema,

                table: parsedTableName.tableName,

                type: MetadataTableType.GENERATED_COLUMN,

                name: column.name,

                value: column.asExpression,

            })

            const deleteQuery = this.deleteTypeormMetadataSql({

                database: parsedTableName.database,

                schema: parsedTableName.schema,

                table: parsedTableName.tableName,

                type: MetadataTableType.GENERATED_COLUMN,

                name: column.name,

            })

            upQueries.push(insertQuery)

            downQueries.push(deleteQuery)

        }

        await this.executeQueries(upQueries, downQueries)

    }

    /**

     * Drops the table.

     * @param tableOrName

     * @param ifExists

     * @param dropForeignKeys

     * @param dropIndices

     */

    async dropTable(

        tableOrName: Table | string,

        ifExists?: boolean,

        dropForeignKeys: boolean = true,

        dropIndices: boolean = true,

    ): Promise<void> {

        if (ifExists) {

            const isTableExist = await this.hasTable(tableOrName)

            if (!isTableExist) return Promise.resolve()

        }

        // if dropTable called with dropForeignKeys = true, we must create foreign keys in down query.

        const createForeignKeys: boolean = dropForeignKeys

        const table = InstanceChecker.isTable(tableOrName)

            ? tableOrName

            : await this.getCachedTable(tableOrName)

        const upQueries: Query[] = []

        const downQueries: Query[] = []

        // It needs because if table does not exist and dropForeignKeys or dropIndices is true, we don't need

        // to perform drop queries for foreign keys and indices.

        if (dropIndices) {

            table.indices.forEach((index) => {

                upQueries.push(this.dropIndexSql(table, index))

                downQueries.push(this.createIndexSql(table, index))

            })

        }

        // if dropForeignKeys is true, we just drop the table, otherwise we also drop table foreign keys.

        // createTable does not need separate method to create foreign keys, because it create fk's in the same query with table creation.

        if (dropForeignKeys)

            table.foreignKeys.forEach((foreignKey) =>

                upQueries.push(this.dropForeignKeySql(table, foreignKey)),

            )

        upQueries.push(this.dropTableSql(table))

        downQueries.push(this.createTableSql(table, createForeignKeys))

        // if table had columns with generated type, we must remove the expression from the metadata table

        const generatedColumns = table.columns.filter(

            (column) => column.generatedType && column.asExpression,

        )

        for (const column of generatedColumns) {

            const parsedTableName = this.driver.parseTableName(table)

            const deleteQuery = this.deleteTypeormMetadataSql({

                database: parsedTableName.database,

                schema: parsedTableName.schema,

                table: parsedTableName.tableName,

                type: MetadataTableType.GENERATED_COLUMN,

                name: column.name,

            })

            const insertQuery = this.insertTypeormMetadataSql({

                database: parsedTableName.database,

                schema: parsedTableName.schema,

                table: parsedTableName.tableName,

                type: MetadataTableType.GENERATED_COLUMN,

                name: column.name,

                value: column.asExpression,

            })

            upQueries.push(deleteQuery)

            downQueries.push(insertQuery)

        }

        await this.executeQueries(upQueries, downQueries)

    }

    /**

     * Creates a new view.

     * @param view

     * @param syncWithMetadata

     */

    async createView(

        view: View,

        syncWithMetadata: boolean = false,

    ): Promise<void> {

        const upQueries: Query[] = []

        const downQueries: Query[] = []

        upQueries.push(this.createViewSql(view))

        if (syncWithMetadata)

            upQueries.push(await this.insertViewDefinitionSql(view))

        downQueries.push(this.dropViewSql(view))

        if (syncWithMetadata)

            downQueries.push(await this.deleteViewDefinitionSql(view))

        await this.executeQueries(upQueries, downQueries)

    }

    /**

     * Drops the view.

     * @param target

     * @param ifExists

     */

    async dropView(target: View | string, ifExists?: boolean): Promise<void> {

    /**

     * Renames a table.

     * @param oldTableOrName

     * @param newTableName

     */

    async renameTable(

        oldTableOrName: Table | string,

        newTableName: string,

    ): Promise<void> {

        const upQueries: Query[] = []

        const downQueries: Query[] = []

        const oldTable = InstanceChecker.isTable(oldTableOrName)

            ? oldTableOrName

            : await this.getCachedTable(oldTableOrName)

        const newTable = oldTable.clone()

        // we need database name and schema name to rename FK constraints

        let dbName: string | undefined = undefined

        let schemaName: string | undefined = undefined

        let oldTableName: string = oldTable.name

        const splittedName = oldTable.name.split(".")

        if (splittedName.length === 3) {

            dbName = splittedName[0]

            oldTableName = splittedName[2]

            if (splittedName[1] !== "") schemaName = splittedName[1]

        newTable.name = this.driver.buildTableName(

            newTableName,

            schemaName,

            dbName,

        )

        // if we have tables with database which differs from database specified in config, we must change currently used database.

        // This need because we can not rename objects from another database.

        const currentDB = await this.getCurrentDatabase()

        if (dbName && dbName !== currentDB) {

            upQueries.push(new Query(`USE "${dbName}"`))

            downQueries.push(new Query(`USE "${currentDB}"`))

        }