dbtProject.ts

import { existsSync, writeFileSync } from "fs";

import {
  Catalog,
  CATALOG_FILE,
  ColumnMetaData,
  DataPilotHealtCheckParams,
  DBColumn,
  DBTCommand,
  DBTCommandExecution,
  DBTCommandExecutionInfrastructure,
  DBTCommandFactory,
  DBTDiagnosticData,
  DBTNode,
  DBTProjectIntegration,
  DBTProjectIntegrationAdapter,
  DBTProjectIntegrationAdapterEvents,
  DBTTerminal,
  DBT_PROJECT_FILE,
  DeferConfig,
  extractOutputColumns,
  HealthcheckArgs,
  isResourceHasDbColumns,
  isResourceNode,
  MANIFEST_FILE,
  NoCredentialsError,
  NodeMetaData,
  ParsedManifest,
  ProjectHealthcheck,
  QueryExecution,
  QueryExecutionResult,
  RESOURCE_TYPE_MODEL,
  RESOURCE_TYPE_SOURCE,
  RunModelParams,
  RunResultsEventData,
  SourceNode,
  Table,
  validateSQL,
} from "@altimateai/dbt-integration";
import { inject } from "inversify";
import * as path from "path";
import { PythonException } from "python-bridge";
import {
  commands,
  Diagnostic,
  DiagnosticCollection,
  DiagnosticSeverity,
  Disposable,
  Event,
  EventEmitter,
  languages,
  ProgressLocation,
  Range,
  RelativePattern,
  Uri,
  ViewColumn,
  window,
  workspace,
} from "vscode";
import { AltimateRequest, ModelNode } from "../altimate";
import { AltimateAuthService } from "../services/altimateAuthService";
import { RunHistoryService } from "../services/runHistoryService";
import { SharedStateService } from "../services/sharedStateService";
import { TelemetryService } from "../telemetry";
import { TelemetryEvents } from "../telemetry/events";
import {
  extendErrorWithSupportLinks,
  getColumnNameByCase,
  getProjectRelativePath,
  resolveSettingsVariables,
} from "../utils";
import { ValidationProvider } from "../validation_provider";
import { DBTProjectLog } from "./dbtProjectLog";
import {
  ManifestCacheChangedEvent,
  ManifestCacheProjectAddedEvent,
  RebuildManifestStatusChange,
} from "./event/manifestCacheChangedEvent";
import { ProjectConfigChangedEvent } from "./event/projectConfigChangedEvent";
import { RunResultsEvent } from "./event/runResultsEvent";
import { PythonEnvironment } from "./pythonEnvironment";

interface FileNameTemplateMap {
  [key: string]: string;
}

interface JsonObj {
  [key: string]: string | number | undefined;
}

export class DBTProject implements Disposable {
  private _manifestCacheEvent?: ManifestCacheProjectAddedEvent;
  readonly projectRoot: Uri;
  private dbtProjectIntegration: DBTProjectIntegrationAdapter;

  private _onProjectConfigChanged =
    new EventEmitter<ProjectConfigChangedEvent>();
  public onProjectConfigChanged = this._onProjectConfigChanged.event;
  private _onRunResults = new EventEmitter<RunResultsEvent>();
  public onRunResults = this._onRunResults.event;
  private _onSourceFileChanged = new EventEmitter<void>();
  public onSourceFileChanged = this._onSourceFileChanged.event;
  private dbtProjectLog?: DBTProjectLog;
  public readonly projectHealth = languages.createDiagnosticCollection("dbt");
  public readonly pythonBridgeDiagnostics =
    languages.createDiagnosticCollection("dbt-python-bridge");
  public readonly rebuildManifestDiagnostics =
    languages.createDiagnosticCollection("dbt-rebuild-manifest");
  public readonly projectConfigDiagnostics =
    languages.createDiagnosticCollection("dbt-project-config");
  private disposables: Disposable[] = [
    this._onProjectConfigChanged,
    this._onSourceFileChanged,
    this.projectHealth,
    this.pythonBridgeDiagnostics,
    this.rebuildManifestDiagnostics,
    this.projectConfigDiagnostics,
  ];
  private _onRebuildManifestStatusChange =
    new EventEmitter<RebuildManifestStatusChange>();
  readonly onRebuildManifestStatusChange =
    this._onRebuildManifestStatusChange.event;

  private dbSchemaCache: Record<string, ModelNode> = {};
  private queues: Map<string, DBTCommandExecution[]> = new Map<
    string,
    DBTCommandExecution[]
  >();
  private queueStates: Map<string, boolean> = new Map<string, boolean>();

  constructor(
    @inject(PythonEnvironment)
    private PythonEnvironment: PythonEnvironment,
    @inject("Factory<DBTProjectLog>")
    private dbtProjectLogFactory: (
      onProjectConfigChanged: Event<ProjectConfigChangedEvent>,
    ) => DBTProjectLog,
    private dbtCommandFactory: DBTCommandFactory,
    private terminal: DBTTerminal,
    private eventEmitterService: SharedStateService,
    private telemetry: TelemetryService,
    private executionInfrastructure: DBTCommandExecutionInfrastructure,
    private dbtIntegrationAdapterFactory: (
      projectRoot: string,
      deferConfig: DeferConfig | undefined,
    ) => DBTProjectIntegrationAdapter,
    private altimate: AltimateRequest,
    private validationProvider: ValidationProvider,
    private altimateAuthService: AltimateAuthService,
    private runHistoryService: RunHistoryService,
    path: Uri,
    _projectConfig: any,
    private _onManifestChanged: EventEmitter<ManifestCacheChangedEvent>,
  ) {
    this.projectRoot = path;
    try {
      this.validationProvider.validateCredentialsSilently();
    } catch (error) {
      this.terminal.error(
        "validateCredentialsSilently",
        "Credential validation failed",
        error,
        false,
      );
    }

    this.dbtProjectLog = this.dbtProjectLogFactory(this.onProjectConfigChanged);

    // Check if dbt loom is installed for telemetry (only for core integration)
    const dbtIntegrationMode = workspace
      .getConfiguration("dbt")
      .get<string>("dbtIntegration", "core");

    if (dbtIntegrationMode === "core") {
      this.isDbtLoomInstalled().then((isInstalled) => {
        this.telemetry.setTelemetryCustomAttribute(
          "dbtLoomInstalled",
          `${isInstalled}`,
        );
      });
    }

    // Create the integration adapter which will handle the integration selection internally
    this.dbtProjectIntegration = this.dbtIntegrationAdapterFactory(
      this.projectRoot.fsPath,
      this.retrieveDeferConfigFromSettings(),
    );

    // Set up Node.js watcher events to emit VSCode events directly
    this.dbtProjectIntegration.on(
      DBTProjectIntegrationAdapterEvents.SOURCE_FILE_CHANGED,
      () => {
        this.terminal.debug(
          "DBTProject",
          "Received sourceFileChanged event from Node.js file watchers",
        );
        this._onSourceFileChanged.fire();
      },
    );

    this.dbtProjectIntegration.on(
      DBTProjectIntegrationAdapterEvents.PROJECT_CONFIG_CHANGED,
      () => {
        this.terminal.debug(
          "DBTProject",
          "Received projectConfigChanged event from Node.js project config watcher",
        );
        const event = new ProjectConfigChangedEvent(this);
        this._onProjectConfigChanged.fire(event);
      },
    );

    this.dbtProjectIntegration.on(
      DBTProjectIntegrationAdapterEvents.REBUILD_MANIFEST_STATUS_CHANGE,
      (status: { inProgress: boolean }) => {
        this.terminal.debug(
          "DBTProject",
          `Received rebuildManifestStatusChange event: inProgress=${status.inProgress}`,
        );
        const event: RebuildManifestStatusChange = {
          project: this,
          inProgress: status.inProgress,
        };
        this._onRebuildManifestStatusChange.fire(event);
      },
    );

    // Handle manifestCreated events from dbtIntegrationAdapter
    this.dbtProjectIntegration.on(
      DBTProjectIntegrationAdapterEvents.MANIFEST_PARSED,
      (parsedManifest: ParsedManifest) => {
        this.terminal.debug(
          "DBTProject",
          "Received manifestParsed event from dbtIntegrationAdapter",
        );
        const manifestCacheEvent: ManifestCacheProjectAddedEvent = {
          project: this,
          nodeMetaMap: parsedManifest.nodeMetaMap,
          macroMetaMap: parsedManifest.macroMetaMap,
          metricMetaMap: parsedManifest.metricMetaMap,
          sourceMetaMap: parsedManifest.sourceMetaMap,
          graphMetaMap: parsedManifest.graphMetaMap,
          testMetaMap: parsedManifest.testMetaMap,
          docMetaMap: parsedManifest.docMetaMap,
          exposureMetaMap: parsedManifest.exposureMetaMap,
          functionMetaMap: parsedManifest.functionMetaMap,
          modelDepthMap: parsedManifest.modelDepthMap,
        };
        this._manifestCacheEvent = manifestCacheEvent;
        this._onManifestChanged.fire({ added: [manifestCacheEvent] });
      },
    );

    // Handle runResultsCreated events from dbtIntegrationAdapter
    this.dbtProjectIntegration.on(
      DBTProjectIntegrationAdapterEvents.RUN_RESULTS_PARSED,
      (eventData: RunResultsEventData) => {
        this.terminal.debug(
          "DBTProject",
          "Received runResultsParsed event from dbtIntegrationAdapter",
        );

        this.runHistoryService.addEntry(eventData);

        const uniqueIds = eventData.results.map((r) => r.uniqueId);
        const runResultsEvent = new RunResultsEvent(this, uniqueIds);
        this._onRunResults.fire(runResultsEvent);
      },
    );

    // Handle diagnosticsChanged events from dbtIntegrationAdapter
    this.dbtProjectIntegration.on(
      DBTProjectIntegrationAdapterEvents.DIAGNOSTICS_CHANGED,
      () => {
        this.terminal.debug(
          "DBTProject",
          "Received diagnosticsChanged event from dbtIntegrationAdapter",
        );
        this.updateDiagnosticsInProblemsPanel();
      },
    );

    this.disposables.push(
      this.dbtProjectIntegration,
      this._onManifestChanged.event((event) => {
        const addedEvent = event.added?.find(
          (e) => e.project.projectRoot === this.projectRoot,
        );
        if (addedEvent) {
          this._manifestCacheEvent = addedEvent;
        }
      }),
      this.onRunResults((event) => {
        this.invalidateCacheUsingUniqueIds(event.uniqueIds || []);
      }),
    );

    // Initialize Python environment and set up change listener
    this.initializePythonEnvironmentListener();

    this.terminal.debug(
      "DbtProject",
      `Created ${dbtIntegrationMode} dbt project ${this.getProjectName()} at ${
        this.projectRoot
      }`,
    );
  }

  private initializePythonEnvironmentListener(): void {
    this.PythonEnvironment.initialize()
      .then(() => {
        this.disposables.push(
          this.PythonEnvironment.onPythonEnvironmentChanged(() =>
            this.onPythonEnvironmentChanged(),
          ),
        );
      })
      .catch((err) => {
        this.terminal.error(
          "dbtProject:initializePythonEnvironmentListener",
          "Failed to initialize Python environment listener",
          err,
        );
      });
  }

  private async isDbtLoomInstalled(): Promise<boolean> {
    const dbtLoomThread = this.executionInfrastructure.createPythonBridge(
      this.projectRoot.fsPath,
    );
    try {
      await dbtLoomThread.ex`from dbt_loom import *`;
      return true;
    } catch (error) {
      return false;
    } finally {
      await this.executionInfrastructure.closePythonBridge(dbtLoomThread);
    }
  }

  private invalidateCacheUsingUniqueIds(uniqueIds: string[]) {
    for (const uniqueId of uniqueIds) {
      if (uniqueId in this.dbSchemaCache) {
        delete this.dbSchemaCache[uniqueId];
      }
    }
  }

  getProjectName() {
    return this.dbtProjectIntegration.getProjectName();
  }

  getProjectRoot() {
    return this.projectRoot.fsPath;
  }

  getSelectedTarget() {
    return this.dbtProjectIntegration.getSelectedTarget();
  }

  getTargetNames() {
    return this.dbtProjectIntegration.getTargetNames();
  }

  async setSelectedTarget(targetName: string) {
    await window.withProgress(
      {
        location: ProgressLocation.Notification,
        title: "Changing target...",
        cancellable: false,
      },
      () => this.dbtProjectIntegration.setSelectedTarget(targetName),
    );
  }

  getDBTProjectFilePath() {
    return path.join(this.projectRoot.fsPath, DBT_PROJECT_FILE);
  }

  getTargetPath() {
    return this.dbtProjectIntegration.getTargetPath();
  }

  getPackageInstallPath() {
    return this.dbtProjectIntegration.getPackageInstallPath();
  }

  getModelPaths() {
    return this.dbtProjectIntegration.getModelPaths();
  }

  getSeedPaths() {
    return this.dbtProjectIntegration.getSeedPaths();
  }

  getMacroPaths() {
    return this.dbtProjectIntegration.getMacroPaths();
  }

  getManifestPath() {
    const targetPath = this.getTargetPath();
    if (!targetPath) {
      return;
    }
    return path.join(targetPath, MANIFEST_FILE);
  }

  getCatalogPath() {
    const targetPath = this.getTargetPath();
    if (!targetPath) {
      return;
    }
    return path.join(targetPath, CATALOG_FILE);
  }

  getPythonBridgeStatus() {
    return this.dbtProjectIntegration.getPythonBridgeStatus();
  }

  getAllDiagnostic(): Diagnostic[] {
    const projectURI = Uri.file(
      path.join(this.projectRoot.fsPath, DBT_PROJECT_FILE),
    );
    const integrationDiagnostics =
      this.getCurrentProjectIntegration().getDiagnostics();

    // Convert diagnostic data to VSCode Diagnostics
    const convertedDiagnostics = [
      ...integrationDiagnostics.pythonBridgeDiagnostics.map(
        (data) =>
          new Diagnostic(
            new Range(
              data.range?.startLine || 0,
              data.range?.startColumn || 0,
              data.range?.endLine || 999,
              data.range?.endColumn || 999,
            ),
            data.message,
            this.mapSeverityToVSCode(data.severity),
          ),
      ),
      ...integrationDiagnostics.rebuildManifestDiagnostics.map(
        (data) =>
          new Diagnostic(
            new Range(
              data.range?.startLine || 0,
              data.range?.startColumn || 0,
              data.range?.endLine || 999,
              data.range?.endColumn || 999,
            ),
            data.message,
            this.mapSeverityToVSCode(data.severity),
          ),
      ),
      ...(integrationDiagnostics.projectConfigDiagnostics || []).map(
        (data) =>
          new Diagnostic(
            new Range(
              data.range?.startLine || 0,
              data.range?.startColumn || 0,
              data.range?.endLine || 999,
              data.range?.endColumn || 999,
            ),
            data.message,
            this.mapSeverityToVSCode(data.severity),
          ),
      ),
    ];

    return [
      ...convertedDiagnostics,
      ...(this.projectHealth.get(projectURI) || []),
    ];
  }

  private mapSeverityToVSCode(severity: string): DiagnosticSeverity {
    switch (severity) {
      case "error":
        return DiagnosticSeverity.Error;
      case "warning":
        return DiagnosticSeverity.Warning;
      case "info":
        return DiagnosticSeverity.Information;
      case "hint":
        return DiagnosticSeverity.Hint;
      default:
        return DiagnosticSeverity.Error;
    }
  }

  private convertDiagnosticDataToVSCode(data: DBTDiagnosticData): Diagnostic {
    return new Diagnostic(
      new Range(
        data.range?.startLine || 0,
        data.range?.startColumn || 0,
        data.range?.endLine || 999,
        data.range?.endColumn || 999,
      ),
      data.message,
      this.mapSeverityToVSCode(data.severity),
    );
  }

  updateDiagnosticsInProblemsPanel(): void {
    const projectURI = Uri.file(
      path.join(this.projectRoot.fsPath, DBT_PROJECT_FILE),
    );
    const integrationDiagnostics =
      this.getCurrentProjectIntegration().getDiagnostics();

    // Update each diagnostic collection separately
    this.pythonBridgeDiagnostics.set(
      projectURI,
      integrationDiagnostics.pythonBridgeDiagnostics.map((data) =>
        this.convertDiagnosticDataToVSCode(data),
      ),
    );

    this.rebuildManifestDiagnostics.set(
      projectURI,
      integrationDiagnostics.rebuildManifestDiagnostics.map((data) =>
        this.convertDiagnosticDataToVSCode(data),
      ),
    );

    this.projectConfigDiagnostics.set(
      projectURI,
      integrationDiagnostics.projectConfigDiagnostics.map((data) =>
        this.convertDiagnosticDataToVSCode(data),
      ),
    );
  }

  async performDatapilotHealthcheck(args: DataPilotHealtCheckParams) {
    const manifestPath = this.getManifestPath();
    if (!manifestPath) {
      throw new Error(
        `Unable to find manifest path for project ${this.getProjectName()}`,
      );
    }
    const healthcheckArgs: HealthcheckArgs = { manifestPath };
    if (args.configType === "Manual") {
      healthcheckArgs.configPath = args.configPath;
    } else {
      if (args.configType === "Saas") {
        healthcheckArgs.config = args.config;
      }
      if (
        args.configType === "All" ||
        args.config_schema.some((i) => i.files_required.includes("Catalog"))
      ) {
        const docsGenerateCommand =
          this.dbtCommandFactory.createDocsGenerateCommand();
        docsGenerateCommand.focus = false;
        docsGenerateCommand.logToTerminal = false;
        docsGenerateCommand.showProgress = false;
        await this.unsafeGenerateDocsImmediately();
        healthcheckArgs.catalogPath = this.getCatalogPath();
        if (!healthcheckArgs.catalogPath) {
          throw new Error(
            `Unable to find catalog path for project ${this.getProjectName()}`,
          );
        }
      }
    }
    this.terminal.debug(
      "performDatapilotHealthcheck",
      "Performing healthcheck",
      healthcheckArgs,
    );
    // Create isolated Python bridge for healthcheck
    const healthCheckThread = this.executionInfrastructure.createPythonBridge(
      this.projectRoot.fsPath,
    );

    let projectHealthcheck: ProjectHealthcheck;
    try {
      await healthCheckThread.ex`from dbt_utils import *`;
      projectHealthcheck = await healthCheckThread.lock<ProjectHealthcheck>(
        (python) =>
          python!`to_dict(project_healthcheck(${healthcheckArgs.manifestPath}, ${healthcheckArgs.catalogPath}, ${healthcheckArgs.configPath}, ${healthcheckArgs.config}, ${this.altimate.getAIKey()}, ${this.altimate.getInstanceName()}, ${this.altimate.getAltimateUrl()}))`,
      );
    } finally {
      await this.executionInfrastructure.closePythonBridge(healthCheckThread);
    }
    // temp fix: ideally datapilot should return absolute path
    for (const key in projectHealthcheck.model_insights) {
      for (const item of projectHealthcheck.model_insights[key]) {
        item.path = path.join(this.projectRoot.fsPath, item.original_file_path);
      }
    }
    return projectHealthcheck;
  }

  async initialize(): Promise<void> {
    // Create command queue for this project
    this.createQueue("all");

    try {
      await this.dbtProjectIntegration.initialize();
    } catch (error) {
      window.showErrorMessage(
        extendErrorWithSupportLinks(
          "An unexpected error occured while initializing the dbt project at " +
            this.projectRoot +
            ": " +
            error +
            ".",
        ),
      );
    }

    // ensure all watchers are cleaned up
    if (this.dbtProjectLog) {
      this.disposables.push(this.dbtProjectLog);
    }

    this.terminal.debug(
      "DbtProject",
      `Initialized dbt project ${this.getProjectName()} at ${this.projectRoot}`,
    );
  }

  async rebuildManifest(): Promise<void> {
    this.dbtProjectIntegration.rebuildManifest();
  }

  private async onPythonEnvironmentChanged() {
    this.terminal.debug(
      "DbtProject",
      `Python environment for dbt project ${this.getProjectName()} at ${
        this.projectRoot
      } has changed`,
    );
    await this.initialize();
  }

  async refreshProjectConfig(): Promise<void> {
    this.dbtProjectIntegration.refreshProjectConfig();
  }

  async parseManifest(): Promise<ParsedManifest | undefined> {
    return await this.dbtProjectIntegration.parseManifest();
  }

  getAdapterType() {
    return this.dbtProjectIntegration.getAdapterType() || "unknown";
  }

  findPackageName(uri: Uri): string | undefined {
    const documentPath = uri.path;
    const pathSegments = documentPath
      .replace(new RegExp(this.projectRoot + "/", "g"), "")
      .split("/");
    const packagesInstallPath = this.getPackageInstallPath();
    if (packagesInstallPath && uri.fsPath.startsWith(packagesInstallPath)) {
      return pathSegments[1];
    }
    return undefined;
  }

  contains(uri: Uri) {
    return (
      uri.fsPath === this.projectRoot.fsPath ||
      uri.fsPath.startsWith(this.projectRoot.fsPath + path.sep)
    );
  }

  async runModel(runModelParams: RunModelParams) {
    if (!this.validateIntegrationPrerequisites()) {
      return undefined;
    }

    const runModelCommand =
      this.dbtCommandFactory.createRunModelCommand(runModelParams);

    try {
      const command =
        await this.getCurrentProjectIntegration().runModel(runModelCommand);
      this.telemetry.sendTelemetryEvent("runModel");
      if (command) {
        this.addCommandToQueue("all", command);
      }
    } catch (error) {
      this.handleNoCredentialsError(error);
    }
  }

  async unsafeRunModelImmediately(runModelParams: RunModelParams) {
    this.telemetry.sendTelemetryEvent("runModel");
    return this.dbtProjectIntegration.unsafeRunModelImmediately(runModelParams);
  }

  async buildModel(runModelParams: RunModelParams) {
    if (!this.validateIntegrationPrerequisites()) {
      return undefined;
    }

    const buildModelCommand =
      this.dbtCommandFactory.createBuildModelCommand(runModelParams);

    try {
      const command =
        await this.getCurrentProjectIntegration().buildModel(buildModelCommand);
      this.telemetry.sendTelemetryEvent("buildModel");
      if (command) {
        this.addCommandToQueue("all", command);
      }
    } catch (error) {
      this.handleNoCredentialsError(error);
    }
  }

  async unsafeBuildModelImmediately(runModelParams: RunModelParams) {
    this.telemetry.sendTelemetryEvent("buildModel");
    return this.dbtProjectIntegration.unsafeBuildModelImmediately(
      runModelParams,
    );
  }

  async buildProject() {
    if (!this.validateIntegrationPrerequisites()) {
      return;
    }

    const buildProjectCommand =
      this.dbtCommandFactory.createBuildProjectCommand();

    try {
      const command =
        await this.getCurrentProjectIntegration().buildProject(
          buildProjectCommand,
        );
      this.telemetry.sendTelemetryEvent("buildProject");
      if (command) {
        this.addCommandToQueue("all", command);
      }
    } catch (error) {
      this.handleNoCredentialsError(error);
    }
  }

  async unsafeBuildProjectImmediately() {
    this.telemetry.sendTelemetryEvent("buildProject");
    return this.dbtProjectIntegration.unsafeBuildProjectImmediately();
  }

  async runTest(testName: string) {
    if (!this.validateIntegrationPrerequisites()) {
      return undefined;
    }

    const testModelCommand =
      this.dbtCommandFactory.createTestModelCommand(testName);

    try {
      const command =
        await this.getCurrentProjectIntegration().runTest(testModelCommand);
      this.telemetry.sendTelemetryEvent("runTest");
      if (command) {
        this.addCommandToQueue("all", command);
      }
    } catch (error) {
      this.handleNoCredentialsError(error);
    }
  }

  async unsafeRunTestImmediately(testName: string) {
    this.telemetry.sendTelemetryEvent("runTest");
    return this.dbtProjectIntegration.unsafeRunTestImmediately(testName);
  }

  async runModelTest(modelName: string) {
    if (!this.validateIntegrationPrerequisites()) {
      return undefined;
    }

    const testModelCommand =
      this.dbtCommandFactory.createTestModelCommand(modelName);

    try {
      const command =
        await this.getCurrentProjectIntegration().runModelTest(
          testModelCommand,
        );
      this.telemetry.sendTelemetryEvent("runModelTest");
      if (command) {
        this.addCommandToQueue("all", command);
      }
    } catch (error) {
      this.handleNoCredentialsError(error);
    }
  }

  async unsafeRunModelTestImmediately(modelName: string) {
    this.telemetry.sendTelemetryEvent("runModelTest");
    return this.dbtProjectIntegration.unsafeRunModelTestImmediately(modelName);
  }

  private handleNoCredentialsError(error: unknown) {
    if (error instanceof NoCredentialsError) {
      this.altimateAuthService.handlePreviewFeatures();
      return;
    }
    window.showErrorMessage((error as Error).message);
  }

  private validateIntegrationPrerequisites(): boolean {
    // Validate different prerequisites based on integration type
    const dbtIntegrationMode = workspace
      .getConfiguration("dbt")
      .get<string>("dbtIntegration", "core");

    switch (dbtIntegrationMode) {
      case "cloud":
      case "fusion":
        // For cloud/fusion integrations, validate authentication
        try {
          this.validationProvider.validateCredentialsSilently();
          return true;
        } catch (e) {
          window.showErrorMessage((e as Error).message);
          return false;
        }
      case "core":
      case "corecommand":
      default:
        // For core integrations, check if we have a proper dbt installation
        // We'll validate through the integration's diagnostic system
        const diagnostics =
          this.getCurrentProjectIntegration().getDiagnostics();
        const hasErrors = [
          ...diagnostics.pythonBridgeDiagnostics,
          ...diagnostics.rebuildManifestDiagnostics,
        ].some((diagnostic) => diagnostic.severity === "error");

        if (hasErrors) {
          window.showErrorMessage(
            "dbt installation or Python environment is not properly configured",
          );
          return false;
        }
        return true;
    }
  }

  private requiresAuthentication(): boolean {
    const dbtIntegrationMode = workspace
      .getConfiguration("dbt")
      .get<string>("dbtIntegration", "core");
    return dbtIntegrationMode === "cloud";
  }

  throwIfNotAuthenticated() {
    if (this.requiresAuthentication()) {
      this.validationProvider.throwIfNotAuthenticated();
    }
  }

  async compileModel(runModelParams: RunModelParams) {
    if (!this.validateIntegrationPrerequisites()) {
      return;
    }

    const compileModelCommand =
      this.dbtCommandFactory.createCompileModelCommand(runModelParams);
    const command =
      await this.getCurrentProjectIntegration().compileModel(
        compileModelCommand,
      );
    this.telemetry.sendTelemetryEvent("compileModel");
    if (command) {
      this.addCommandToQueue("all", command);
    }
  }

  async unsafeCompileModelImmediately(runModelParams: RunModelParams) {
    this.telemetry.sendTelemetryEvent("compileModel");
    return this.dbtProjectIntegration.unsafeCompileModelImmediately(
      runModelParams,
    );
  }

  async unsafeGenerateDocsImmediately(args?: string[]) {
    return this.dbtProjectIntegration.unsafeGenerateDocsImmediately(args);
  }

  async generateDocs() {
    if (!this.validateIntegrationPrerequisites()) {
      return;
    }

    const docsGenerateCommand =
      this.dbtCommandFactory.createDocsGenerateCommand();
    const command =
      await this.getCurrentProjectIntegration().generateDocs(
        docsGenerateCommand,
      );
    this.telemetry.sendTelemetryEvent("generateDocs");
    if (command) {
      this.addCommandToQueue("all", command);
    }
  }

  clean() {
    this.throwIfNotAuthenticated();
    this.telemetry.sendTelemetryEvent("clean");
    return this.dbtProjectIntegration.clean();
  }

  debug(focus: boolean = true) {
    this.telemetry.sendTelemetryEvent("debug");
    return this.dbtProjectIntegration.debug(focus);
  }

  async installDbtPackages(packages: string[]) {
    this.telemetry.sendTelemetryEvent("installDbtPackages");
    return this.dbtProjectIntegration.installDbtPackages(packages);
  }

  async installDeps(silent = false) {
    this.telemetry.sendTelemetryEvent("installDeps");
    return this.dbtProjectIntegration.installDeps(silent);
  }

  async compileNode(modelName: string): Promise<string | undefined> {
    this.telemetry.sendTelemetryEvent("compileNode");
    this.throwDiagnosticsErrorIfAvailable();
    try {
      return await this.dbtProjectIntegration.unsafeCompileNode(modelName);
    } catch (exc: any) {
      if (exc instanceof PythonException) {
        window.showErrorMessage(
          extendErrorWithSupportLinks(
            `An error occured while trying to compile your node: ${modelName}` +
              exc.exception.message +
              ".",
          ),
        );
        this.telemetry.sendTelemetryError("compileNodePythonError", exc);
        return (
          "Exception: " +
          exc.exception.message +
          "\n\n" +
          "Detailed error information:\n" +
          exc
        );
      }
      this.telemetry.sendTelemetryError("compileNodeUnknownError", exc);
      // Unknown error
      window.showErrorMessage(
        extendErrorWithSupportLinks(
          "Could not compile model " +
            modelName +
            ": " +
            (exc as Error).message +
            ".",
        ),
      );
      return "Detailed error information:\n" + exc;
    }
  }

  async unsafeCompileNode(modelName: string): Promise<string | undefined> {
    this.telemetry.sendTelemetryEvent("unsafeCompileNode");
    this.throwDiagnosticsErrorIfAvailable();
    this.throwIfNotAuthenticated();
    return this.dbtProjectIntegration.unsafeCompileNode(modelName);
  }

  async validateSql(request: {
    sql: string;
    dialect: string;
    models: any[];
  }): Promise<Awaited<ReturnType<typeof validateSQL>>> {
    const { sql, dialect, models } = request;
    this.throwDiagnosticsErrorIfAvailable();
    this.throwIfNotAuthenticated();
    const sqlValidationThread = this.executionInfrastructure.createPythonBridge(
      this.projectRoot.fsPath,
    );
    try {
      return await validateSQL(sql, dialect, models, sqlValidationThread);
    } finally {
      await this.executionInfrastructure.closePythonBridge(sqlValidationThread);
    }
  }

  async validateSQLDryRun(query: string) {
    this.throwIfNotAuthenticated();
    try {
      return this.dbtProjectIntegration.validateSQLDryRun(query);
    } catch (exc) {
      const exception = exc as { exception: { message: string } };
      window.showErrorMessage(
        exception.exception.message || "Could not validate sql with dry run.",
      );
      this.telemetry.sendTelemetryError("validateSQLDryRunError", {
        error: exc,
      });
    }
  }

  getDBTVersion(): number[] | undefined {
    return this.getCurrentProjectIntegration().getVersion();
  }

  async compileQuery(
    query: string,
    originalModelName: string | undefined = undefined,
  ): Promise<string | undefined> {
    this.telemetry.sendTelemetryEvent("compileQuery");
    try {
      return await this.dbtProjectIntegration.unsafeCompileQuery(
        query,
        originalModelName,
      );
    } catch (exc: any) {
      if (exc instanceof PythonException) {
        window.showErrorMessage(
          extendErrorWithSupportLinks(
            "An error occured while trying to compile your query: " +
              exc.exception.message +
              ".",
          ),
        );
        this.telemetry.sendTelemetryError("compileQueryPythonError", exc);
        return undefined;
      }
      this.telemetry.sendTelemetryError("compileQueryUnknownError", exc);
      // Unknown error
      window.showErrorMessage(
        extendErrorWithSupportLinks(
          "Could not compile query: " + (exc as Error).message,
        ),
      );
      return undefined;
    }
  }

  showCompiledSql(modelPath: Uri) {
    this.findModelInTargetfolder(modelPath, "compiled");
  }

  showRunSQL(modelPath: Uri) {
    this.findModelInTargetfolder(modelPath, "run");
  }

  createYMLContent(
    columnsInRelation: { [key: string]: string }[],
    modelName: string,
  ): string {
    let yamlString = "version: 2\n\nmodels:\n";
    yamlString += `  - name: ${modelName}\n    description: ""\n    columns:\n`;
    for (const item of columnsInRelation) {
      yamlString += `    - name: ${item.column}\n      description: ""\n`;
    }
    return yamlString;
  }

  async unsafeCompileQuery(
    query: string,
    originalModelName: string | undefined = undefined,
  ) {
    this.throwIfNotAuthenticated();
    return this.dbtProjectIntegration.unsafeCompileQuery(
      query,
      originalModelName,
    );
  }

  async getColumnsOfModel(modelName: string) {
    this.throwIfNotAuthenticated();
    const result =
      await this.dbtProjectIntegration.getColumnsOfModel(modelName);
    await this.getCurrentProjectIntegration().cleanupConnections();
    return result;
  }

  async getColumnsOfSource(sourceName: string, tableName: string) {
    this.throwIfNotAuthenticated();
    const result = await this.dbtProjectIntegration.getColumnsOfSource(
      sourceName,
      tableName,
    );
    await this.getCurrentProjectIntegration().cleanupConnections();
    return result;
  }

  async getColumnValues(model: string, column: string) {
    this.telemetry.startTelemetryEvent(
      TelemetryEvents["DocumentationEditor/GetDistinctColumnValues"],
      { column, model },
    );

    try {
      this.throwIfNotAuthenticated();
      this.terminal.debug(
        "getColumnValues",
        "finding distinct values for column",
        true,
        { model, column },
      );
      const result = this.dbtProjectIntegration.getColumnValues(model, column);
      this.telemetry.endTelemetryEvent(
        TelemetryEvents["DocumentationEditor/GetDistinctColumnValues"],
        undefined,
        { column, model },
      );
      return (result as any).flat();
    } catch (error) {
      this.telemetry.endTelemetryEvent(
        TelemetryEvents["DocumentationEditor/GetDistinctColumnValues"],
        error,
        { column, model },
      );
      throw error;
    } finally {
      await this.getCurrentProjectIntegration().cleanupConnections();
    }
  }

  async getBulkSchemaFromDB(req: DBTNode[], signal: AbortSignal) {
    this.throwIfNotAuthenticated();
    try {
      const result =
        await this.getCurrentProjectIntegration().getBulkSchemaFromDB(
          req,
          signal,
        );
      await this.getCurrentProjectIntegration().cleanupConnections();
      return result;
    } finally {
      await this.getCurrentProjectIntegration().cleanupConnections();
    }
  }

  async getCatalog(): Promise<Catalog> {
    this.throwIfNotAuthenticated();
    try {
      const result = await this.getCurrentProjectIntegration().getCatalog();
      return result;
    } catch (exc: any) {
      if (exc instanceof PythonException) {
        this.telemetry.sendTelemetryError("catalogPythonError", exc, {
          adapter: this.getAdapterType(),
        });
        window.showErrorMessage(
          "Some of the scans could not run as connectivity to database for the project " +
            this.getProjectName() +
            " is not available. ",
        );
        return [];
      }
      // Unknown error
      this.telemetry.sendTelemetryError("catalogUnknownError", exc, {
        adapter: this.getAdapterType(),
      });
      window.showErrorMessage(
        "Some of the scans could not run as connectivity to database for the project " +
          this.getProjectName() +
          " is not available. ",
      );
      return [];
    } finally {
      await this.getCurrentProjectIntegration().cleanupConnections();
    }
  }

  async generateSchemaYML(modelPath: Uri, modelName: string) {
    try {
      // Create filePath based on model location
      const currentDir = path.dirname(modelPath.fsPath);
      const location = path.join(currentDir, modelName + "_schema.yml");
      if (!existsSync(location)) {
        this.telemetry.sendTelemetryEvent("generateSchemaYML", {
          adapter: this.getAdapterType(),
        });
        const columnsInRelation = await this.getColumnsOfModel(modelName);
        // Generate yml file content
        const fileContents = this.createYMLContent(
          columnsInRelation,
          modelName,
        );
        writeFileSync(location, fileContents);
        const doc = await workspace.openTextDocument(Uri.file(location));
        window.showTextDocument(doc);
      } else {
        window.showErrorMessage(
          `A file called ${modelName}_schema.yml already exists in ${currentDir}. If you want to generate the schema yml, please rename the other file or delete it if you want to generate the yml again.`,
        );
      }
    } catch (exc: any) {
      if (exc instanceof PythonException) {
        this.telemetry.sendTelemetryError("generateSchemaYMLPythonError", exc, {
          adapter: this.getAdapterType(),
        });
        window.showErrorMessage(
          extendErrorWithSupportLinks(
            "An error occured while trying to generate the schema yml " +
              exc.exception.message +
              ".",
          ),
        );
      }
      // Unknown error
      this.telemetry.sendTelemetryError("generateSchemaYMLUnknownError", exc, {
        adapter: this.getAdapterType(),
      });
      window.showErrorMessage(
        extendErrorWithSupportLinks(
          "Could not generate schema yaml: " + (exc as Error).message,
        ),
      );
    }
  }

  async generateModel(
    sourceName: string,
    tableName: string,
    sourcePath: string,
  ) {
    await window.withProgress(
      {
        location: ProgressLocation.Notification,
        title: "Generating model...",
        cancellable: false,
      },
      async () => {
        try {
          const prefix = workspace
            .getConfiguration("dbt")
            .get<string>("prefixGenerateModel", "base");

          // Map setting to fileName
          const fileNameTemplateMap: FileNameTemplateMap = {
            "{prefix}_{sourceName}_{tableName}": `${prefix}_${sourceName}_${tableName}`,
            "{prefix}_{sourceName}__{tableName}": `${prefix}_${sourceName}__${tableName}`,
            "{prefix}_{tableName}": `${prefix}_${tableName}`,
            "{tableName}": `${tableName}`,
          };

          // Default filename template
          let fileName = `${prefix}_${sourceName}_${tableName}`;

          const fileNameTemplate = workspace
            .getConfiguration("dbt")
            .get<string>(
              "fileNameTemplateGenerateModel",
              "{prefix}_{sourceName}_{tableName}",
            );

          this.telemetry.sendTelemetryEvent("generateModel", {
            prefix: prefix,
            filenametemplate: fileNameTemplate,
            adapter: this.getAdapterType(),
          });

          // Parse setting to fileName
          if (fileNameTemplate in fileNameTemplateMap) {
            fileName = fileNameTemplateMap[fileNameTemplate];
          }
          // Create filePath based on source.yml location
          const location = path.join(sourcePath, fileName + ".sql");
          if (!existsSync(location)) {
            const columnsInRelation = await this.getColumnsOfSource(
              sourceName,
              tableName,
            );
            this.terminal.debug(
              "dbtProject:generateModel",
              `Generating columns for source ${sourceName} and table ${tableName}`,
              columnsInRelation,
            );

            const fileContents = `with source as (
        select * from {{ source('${sourceName}', '${tableName}') }}
  ),
  renamed as (
      select
          ${columnsInRelation
            .map((column) => `{{ adapter.quote("${column.column}") }}`)
            .join(",\n        ")}

      from source
  )
  select * from renamed
    `;
            writeFileSync(location, fileContents);
            const doc = await workspace.openTextDocument(Uri.file(location));
            window.showTextDocument(doc);
          } else {
            window.showErrorMessage(
              `A model called ${fileName} already exists in ${sourcePath}. If you want to generate the model, please rename the other model or delete it if you want to generate the model again.`,
            );
          }
        } catch (exc: any) {
          if (exc instanceof PythonException) {
            this.telemetry.sendTelemetryError("generateModelPythonError", exc, {
              adapter: this.getAdapterType(),
            });
            window.showErrorMessage(
              "An error occured while trying to generate the model " +
                exc.exception.message,
            );
          }
          // Unknown error
          this.telemetry.sendTelemetryError("generateModelUnknownError", exc, {
            adapter: this.getAdapterType(),
          });
          window.showErrorMessage(
            extendErrorWithSupportLinks(
              "An error occured while trying to generate the model:" +
                exc +
                ".",
            ),
          );
        }
      },
    );
  }

  async executeSQLOnQueryPanel(
    query: string,
    modelName: string,
    fullRefresh: boolean = false,
  ) {
    const limit = workspace
      .getConfiguration("dbt")
      .get<number>("queryLimit", 500);
    return this.executeSQLWithLimitOnQueryPanel(
      query,
      modelName,
      limit,
      fullRefresh,
    );
  }

  async executeSQLWithLimitOnQueryPanel(
    query: string,
    modelName: string,
    limit: number,
    fullRefresh: boolean = false,
  ) {
    if (limit <= 0) {
      window.showErrorMessage("Please enter a positive number for query limit");
      return;
    }
    this.terminal.info("executeSQL", "Executed query: " + query, true, {
      adapter: this.getAdapterType(),
      limit: limit.toString(),
      fullRefresh: fullRefresh.toString(),
    });
    // TODO(#1670): once `@altimateai/dbt-integration` is bumped to the version
    // containing the `fullRefresh` plumbing
    // (https://github.com/AltimateAI/altimate-dbt-integration/pull/44), drop
    // the `as unknown as` cast — the published 0.2.12 types still declare
    // `executeSQLWithLimit` with 3 params, so we widen locally here to let CI
    // compile against the npm-published typings while the patched runtime
    // accepts the 4th argument.
    const executeSQLWithLimit = this.dbtProjectIntegration
      .executeSQLWithLimit as unknown as (
      query: string,
      modelName: string,
      limit: number,
      fullRefresh?: boolean,
    ) => Promise<QueryExecution>;
    this.eventEmitterService.fire({
      command: "executeQuery",
      payload: {
        query,
        fn: executeSQLWithLimit(query, modelName, limit, fullRefresh),
        projectName: this.getProjectName(),
      },
    });
  }

  async immediatelyExecuteSQLWithLimit(
    query: string,
    modelName: string,
    limit: number,
  ): Promise<QueryExecutionResult> {
    this.throwDiagnosticsErrorIfAvailable();
    this.throwIfNotAuthenticated();
    this.terminal.info("executeSQL", "Executed query: " + query, true, {
      adapter: this.getAdapterType(),
      limit: limit.toString(),
    });
    return this.dbtProjectIntegration.immediatelyExecuteSQLWithLimit(
      query,
      modelName,
      limit,
    );
  }

  async executeSQLWithLimit(query: string, modelName: string, limit: number) {
    this.throwDiagnosticsErrorIfAvailable();
    this.throwIfNotAuthenticated();
    this.terminal.info("executeSQL", "Executed query: " + query, true, {
      adapter: this.getAdapterType(),
      limit: limit.toString(),
    });
    return this.dbtProjectIntegration.executeSQLWithLimit(
      query,
      modelName,
      limit,
    );
  }

  async immediatelyExecuteSQL(
    query: string,
    modelName: string,
  ): Promise<QueryExecutionResult> {
    this.throwDiagnosticsErrorIfAvailable();
    this.throwIfNotAuthenticated();
    const limit = workspace
      .getConfiguration("dbt")
      .get<number>("queryLimit", 500);
    this.terminal.info("executeSQL", "Executed query: " + query, true, {
      adapter: this.getAdapterType(),
      limit: limit.toString(),
    });
    return this.dbtProjectIntegration.immediatelyExecuteSQL(query, modelName);
  }

  executeSQL(query: string, modelName: string): Promise<QueryExecution> {
    const limit = workspace
      .getConfiguration("dbt")
      .get<number>("queryLimit", 500);
    this.terminal.info("executeSQL", "Executed query: " + query, true, {
      adapter: this.getAdapterType(),
      limit: limit.toString(),
    });
    return this.dbtProjectIntegration.executeSQL(query, modelName);
  }

  async dispose() {
    while (this.disposables.length) {
      const x = this.disposables.pop();
      if (x) {
        x.dispose();
      }
    }
  }

  private async findModelInTargetfolder(modelPath: Uri, type: string) {
    const targetPath = this.getTargetPath();
    if (!targetPath) {
      return;
    }
    const relativePath = path.relative(
      this.projectRoot.fsPath,
      modelPath.fsPath,
    );

    const targetModels = await workspace.findFiles(
      new RelativePattern(targetPath, path.join(type, "**", relativePath)),
    );
    if (targetModels.length > 0) {
      commands.executeCommand("vscode.open", targetModels[0], {
        preview: false,
        preserveFocus: true,
        viewColumn: ViewColumn.Beside,
      });
    }
  }

  static isResourceNode(resourceType: string): boolean {
    return isResourceNode(resourceType);
  }

  static isResourceHasDbColumns(resourceType: string): boolean {
    return isResourceHasDbColumns(resourceType);
  }

  getNonEphemeralParents(keys: string[]): string[] {
    return this.dbtProjectIntegration.getNonEphemeralParents(keys);
  }

  getChildrenModels({ table }: { table: string }): Table[] {
    return this.dbtProjectIntegration.getChildrenModels({ table });
  }

  getParentModels({ table }: { table: string }): Table[] {
    return this.dbtProjectIntegration.getParentModels({ table });
  }

  mergeColumnsFromDB(
    node: Pick<ModelNode, "columns">,
    columnsFromDB: DBColumn[],
  ) {
    if (!columnsFromDB || columnsFromDB.length === 0) {
      return false;
    }
    if (columnsFromDB.length > 100) {
      // Flagging events where more than 100 columns are fetched from db to get a sense of how many of these happen
      this.telemetry.sendTelemetryEvent("excessiveColumnsFetchedFromDB");
    }
    const columnsFromManifest: Record<string, ColumnMetaData> = {};
    Object.entries(node.columns).forEach(([k, v]) => {
      columnsFromManifest[getColumnNameByCase(k, this.getAdapterType())] = v;
    });

    for (const c of columnsFromDB) {
      const columnNameFromDB = getColumnNameByCase(
        c.column,
        this.getAdapterType(),
      );
      const existing_column = columnsFromManifest[columnNameFromDB];
      if (existing_column) {
        existing_column.data_type = (
          existing_column.data_type || c.dtype
        )?.toLowerCase();
        continue;
      }
      node.columns[columnNameFromDB] = {
        name: columnNameFromDB,
        data_type: c.dtype?.toLowerCase(),
        description: "",
        meta: {},
      };
    }
    if (Object.keys(node.columns).length > columnsFromDB.length) {
      // Flagging events where columns fetched from db are less than the number of columns in the manifest
      this.telemetry.sendTelemetryEvent("possibleStaleSchema");
    }
    return true;
  }

  public findPackageVersion(packageName: string) {
    const version =
      this.getCurrentProjectIntegration().findPackageVersion(packageName);
    this.terminal.debug(
      "dbtProject:findPackageVersion",
      `found ${packageName} version: ${version}`,
    );
    return version;
  }

  async getBulkCompiledSql(models: string[]) {
    if (models.length === 0) {
      return {};
    }
    if (!this._manifestCacheEvent) {
      throw new Error("The dbt manifest is not available");
    }
    const { nodeMetaMap } = this._manifestCacheEvent;
    return this.getCurrentProjectIntegration().getBulkCompiledSQL(
      models
        .map((m) => nodeMetaMap.lookupByUniqueId(m))
        .filter(Boolean) as NodeMetaData[],
    );
  }

  async getNodesWithDBColumns(modelsToFetch: string[], signal: AbortSignal) {
    const mappedNode: Record<string, ModelNode> = {};
    const relationsWithoutColumns: string[] = [];
    if (modelsToFetch.length === 0) {
      return { mappedNode, relationsWithoutColumns, mappedCompiledSql: {} };
    }
    if (!this._manifestCacheEvent) {
      throw new Error("The dbt manifest is not available");
    }
    const { nodeMetaMap, sourceMetaMap } = this._manifestCacheEvent;
    const bulkSchemaRequest: DBTNode[] = [];

    for (const key of modelsToFetch) {
      if (this.dbSchemaCache[key]) {
        this.telemetry.sendTelemetryEvent("dbSchemaCacheHit", { model: key });
        mappedNode[key] = this.dbSchemaCache[key];
        continue;
      }
      const splits = key.split(".");
      const resource_type = splits[0];
      if (resource_type === RESOURCE_TYPE_SOURCE) {
        const source = sourceMetaMap.get(splits[2]);
        const tableName = splits[3];
        if (!source) {
          continue;
        }
        const table = source?.tables.find((t) => t.name === tableName);
        if (!table) {
          continue;
        }
        bulkSchemaRequest.push({
          unique_id: key,
          name: source.name,
          resource_type,
          table: table.name,
        } as SourceNode);
        const node = {
          database: source.database,
          schema: source.schema,
          name: table.name,
          alias: table.identifier,
          uniqueId: key,
          columns: table.columns,
          path: table.path,
        };
        mappedNode[key] = node;
      } else if (DBTProject.isResourceNode(resource_type)) {
        const node = nodeMetaMap.lookupByUniqueId(key);
        if (!node) {
          continue;
        }
        if (DBTProject.isResourceHasDbColumns(resource_type)) {
          bulkSchemaRequest.push({
            unique_id: key,
            name: node.name,
            resource_type,
          });
        }
        mappedNode[key] = {
          uniqueId: key,
          ...node,
        };
      }
    }

    const dbSchemaRequest = bulkSchemaRequest.filter(
      (r) => r.resource_type !== RESOURCE_TYPE_MODEL,
    );

    const sqlglotSchemaRequest = bulkSchemaRequest.filter(
      (r) => r.resource_type === RESOURCE_TYPE_MODEL,
    );
    let startTime = Date.now();
    const sqlglotSchemaResponse = await this.getBulkCompiledSql(
      sqlglotSchemaRequest.map((r) => r.unique_id),
    );
    const compiledSqlTime = Date.now() - startTime;

    if (signal.aborted) {
      return {
        mappedNode,
        relationsWithoutColumns,
        mappedCompiledSql: sqlglotSchemaResponse,
      };
    }

    const sqlglotSchemas: Record<string, DBColumn[]> = {};
    const dialect = this.getAdapterType();

    startTime = Date.now();
    for (const r of sqlglotSchemaRequest) {
      if (!sqlglotSchemaResponse[r.unique_id]) {
        dbSchemaRequest.push(r);
        continue;
      }

      try {
        const columns = await extractOutputColumns(
          sqlglotSchemaResponse[r.unique_id],
          dialect,
        );
        sqlglotSchemas[r.unique_id] = columns.map((c) => ({
          column: c,
          dtype: "string",
        }));
      } catch (e) {
        this.terminal.warn(
          "sqlglotSchemaFetchingFailed",
          `Error while schema fetching for ${r.unique_id}`,
          true,
          e,
        );
        dbSchemaRequest.push(r);
      }
    }
    const sqlglotSchemaTime = Date.now() - startTime;

    if (signal.aborted) {
      return {
        mappedNode,
        relationsWithoutColumns,
        mappedCompiledSql: sqlglotSchemaResponse,
      };
    }

    startTime = Date.now();
    const dbSchemaResponse =
      await this.getCurrentProjectIntegration().getBulkSchemaFromDB(
        dbSchemaRequest,
        signal,
      );
    const dbFetchTime = Date.now() - startTime;

    const bulkSchemaResponse = { ...dbSchemaResponse, ...sqlglotSchemas };

    for (const key of modelsToFetch) {
      if (!bulkSchemaRequest.find((r) => r.unique_id === key)) {
        continue;
      }
      const node = mappedNode[key];
      if (!node) {
        continue;
      }
      const dbColumnAdded = this.mergeColumnsFromDB(
        node,
        bulkSchemaResponse[key],
      );
      if (!dbColumnAdded) {
        relationsWithoutColumns.push(key);
      } else {
        // only adding to cache when successfully fetched columns from db
        this.dbSchemaCache[key] = mappedNode[key];
      }
    }

    console.log("getNodesWithDBColumnsTimings", {
      compiledSqlTime,
      sqlglotSchemaTime,
      dbFetchTime,
      modelInfosLength: modelsToFetch.length,
    });
    this.telemetry.sendTelemetryEvent("getNodesWithDBColumnsTimings", {
      compiledSqlTime: compiledSqlTime.toString(),
      sqlglotSchemaTime: sqlglotSchemaTime.toString(),
      dbFetchTime: dbFetchTime.toString(),
      modelInfosLength: modelsToFetch.length.toString(),
    });

    return {
      mappedNode,
      relationsWithoutColumns,
      mappedCompiledSql: sqlglotSchemaResponse,
    };
  }

  async applyDeferConfig(): Promise<void> {
    const deferConfig = this.retrieveDeferConfigFromSettings();
    await this.dbtProjectIntegration.applyDeferConfig(deferConfig);
  }

  throwDiagnosticsErrorIfAvailable() {
    // Check integration diagnostics
    const integrationDiagnostics =
      this.getCurrentProjectIntegration().getDiagnostics();
    const allIntegrationDiagnostics = [
      ...integrationDiagnostics.pythonBridgeDiagnostics,
      ...integrationDiagnostics.rebuildManifestDiagnostics,
    ];

    for (const diagnostic of allIntegrationDiagnostics) {
      if (diagnostic.severity === "error") {
        throw new Error(diagnostic.message);
      }
    }

    // Check VSCode diagnostic collections
    const vscodeCollections: DiagnosticCollection[] = [
      this.projectHealth,
      this.pythonBridgeDiagnostics,
      this.rebuildManifestDiagnostics,
      this.projectConfigDiagnostics,
    ];

    for (const diagnosticCollection of vscodeCollections) {
      for (const [_, diagnostics] of diagnosticCollection) {
        const error = diagnostics.find(
          (diagnostic) => diagnostic.severity === DiagnosticSeverity.Error,
        );
        if (error) {
          throw new Error(error.message);
        }
      }
    }
  }

  private retrieveDeferConfigFromSettings(): DeferConfig | undefined {
    const relativePath = getProjectRelativePath(this.projectRoot);
    const currentConfig: Record<string, DeferConfig> = workspace
      .getConfiguration("dbt")
      .get("deferConfigPerProject", {});
    if (currentConfig[relativePath]) {
      const config = currentConfig[relativePath];
      const resolvedManifestPath = config.manifestPathForDeferral
        ? resolveSettingsVariables(
            config.manifestPathForDeferral,
            this.projectRoot,
          )
        : config.manifestPathForDeferral;
      return new DeferConfig(
        config.deferToProduction,
        config.favorState,
        resolvedManifestPath,
        config.manifestPathType,
        config.dbtCoreIntegrationId,
      );
    }
  }

  getDeferConfig(): DeferConfig {
    if (!this.dbtProjectIntegration) {
      throw new Error("DBT Project Integration is not initialized.");
    }
    return this.dbtProjectIntegration.getDeferConfig();
  }

  private createQueue(queueName: string) {
    this.queues.set(queueName, []);
  }

  private addCommandToQueue(queueName: string, command: DBTCommand): void {
    this.queues.get(queueName)!.push({
      command: async (signal) => {
        await command.execute(signal);
      },
      statusMessage: command.statusMessage,
      focus: command.focus,
      signal: command.signal,
      showProgress: command.showProgress,
    });
    this.pickCommandToRun(queueName);
  }

  private async pickCommandToRun(queueName: string): Promise<void> {
    const queue = this.queues.get(queueName)!;
    const running = this.queueStates.get(queueName);
    if (!running && queue.length > 0) {
      this.queueStates.set(queueName, true);
      const { command, statusMessage, focus, showProgress } = queue.shift()!;
      const commandExecution = async (signal?: AbortSignal) => {
        try {
          await command(signal);
        } catch (error) {
          if (error instanceof NoCredentialsError) {
            this.altimateAuthService.handlePreviewFeatures();
            return;
          }
          window.showErrorMessage(
            extendErrorWithSupportLinks(
              `Could not run command '${statusMessage}': ` + error + ".",
            ),
          );
          this.telemetry.sendTelemetryError("queueRunCommandError", error, {
            command: statusMessage,
          });
        }
      };

      if (showProgress) {
        await window.withProgress(
          {
            location: focus
              ? ProgressLocation.Notification
              : ProgressLocation.Window,
            cancellable: true,
            title: statusMessage,
          },
          async (_, token) => {
            const abortController = new AbortController();
            token.onCancellationRequested(() => abortController.abort());
            await commandExecution(abortController.signal);
          },
        );
      } else {
        await commandExecution();
      }
      this.queueStates.set(queueName, false);
      this.pickCommandToRun(queueName);
    }
  }

  private getCurrentProjectIntegration(): DBTProjectIntegration {
    return this.dbtProjectIntegration.getCurrentProjectIntegration();
  }
}