import { createPrivateKey } from "crypto";
import path from "path";
import { components } from "../api";

type ApiConnection = components["schemas"]["Connection"];
type AttachedDatabase = components["schemas"]["AttachedDatabase"];

export type CoreConnectionEntry = {
   is: string;
   [key: string]: unknown;
};

export type CoreConnectionsPojo = {
   connections: Record<string, CoreConnectionEntry>;
};

export type EnvironmentConnectionMetadata = {
   apiConnection: ApiConnection;
   attachedDatabases: AttachedDatabase[];
   hasAzureAttachment: boolean;
   hasSnowflakePrivateKey: boolean;
   isDuckLake: boolean;
   databasePath?: string;
   workingDirectory: string;
   proxy?: ApiConnection["proxy"];
};

export type AssembledEnvironmentConnections = {
   pojo: CoreConnectionsPojo;
   metadata: Map<string, EnvironmentConnectionMetadata>;
   apiConnections: ApiConnection[];
};

const PUBLISHER_DUCKDB_API_FIELDS = new Set<string>(["attachedDatabases"]);

export function normalizeSnowflakePrivateKey(privateKey: string): string {
   let privateKeyContent = privateKey.trim();

   if (!privateKeyContent.includes("\n")) {
      const keyPatterns = [
         {
            beginRegex: /-----BEGIN\s+ENCRYPTED\s+PRIVATE\s+KEY-----/i,
            endRegex: /-----END\s+ENCRYPTED\s+PRIVATE\s+KEY-----/i,
            beginMarker: "-----BEGIN ENCRYPTED PRIVATE KEY-----",
            endMarker: "-----END ENCRYPTED PRIVATE KEY-----",
         },
         {
            beginRegex: /-----BEGIN\s+PRIVATE\s+KEY-----/i,
            endRegex: /-----END\s+PRIVATE\s+KEY-----/i,
            beginMarker: "-----BEGIN PRIVATE KEY-----",
            endMarker: "-----END PRIVATE KEY-----",
         },
         {
            beginRegex: /-----BEGIN\s+RSA\s+PRIVATE\s+KEY-----/i,
            endRegex: /-----END\s+RSA\s+PRIVATE\s+KEY-----/i,
            beginMarker: "-----BEGIN RSA PRIVATE KEY-----",
            endMarker: "-----END RSA PRIVATE KEY-----",
         },
      ];

      for (const pattern of keyPatterns) {
         const beginMatch = privateKeyContent.match(pattern.beginRegex);
         const endMatch = privateKeyContent.match(pattern.endRegex);

         if (beginMatch && endMatch) {
            const beginPos = beginMatch.index! + beginMatch[0].length;
            const endPos = endMatch.index!;
            const keyData = privateKeyContent
               .substring(beginPos, endPos)
               .replace(/\s+/g, "");

            const lines: string[] = [];
            for (let i = 0; i < keyData.length; i += 64) {
               lines.push(keyData.slice(i, i + 64));
            }
            privateKeyContent = `${pattern.beginMarker}\n${lines.join("\n")}\n${pattern.endMarker}\n`;
            break;
         }
      }
   } else if (!privateKeyContent.endsWith("\n")) {
      privateKeyContent += "\n";
   }

   // Snowflake's Node SDK requires PKCS#8 ("BEGIN PRIVATE KEY"). Convert
   // PKCS#1 ("BEGIN RSA PRIVATE KEY") so users can paste either format.
   if (/-----BEGIN\s+RSA\s+PRIVATE\s+KEY-----/i.test(privateKeyContent)) {
      try {
         privateKeyContent = createPrivateKey({
            key: privateKeyContent,
            format: "pem",
         })
            .export({ type: "pkcs8", format: "pem" })
            .toString();
      } catch (err) {
         throw new Error(
            `Failed to convert Snowflake RSA private key (PKCS#1) to PKCS#8: ${
               err instanceof Error ? err.message : String(err)
            }`,
         );
      }
      if (!privateKeyContent.endsWith("\n")) {
         privateKeyContent += "\n";
      }
   }

   return privateKeyContent;
}

// NOTE: This narrows the environment-author API surface (it rejects securityPolicy,
// allowedDirectories, setupSQL, etc.). It is NOT a filesystem isolation
// boundary: attachedDatabases[].path is not normalized or constrained to stay
// under the environment root, and DuckDB's local-file access is unchanged.
// Adversarial filesystem isolation is an explicit non-goal here: DuckDB
// hardening knobs are not exposed and there is no adversarial DuckDB
// filesystem isolation. Future work owns any path-traversal/allowlist
// enforcement.
export function validateDuckdbApiSurface(connection: ApiConnection): void {
   if (connection.type !== "duckdb" || !connection.duckdbConnection) return;

   const unsupportedFields = Object.keys(connection.duckdbConnection).filter(
      (field) =>
         !PUBLISHER_DUCKDB_API_FIELDS.has(field) &&
         (connection.duckdbConnection as Record<string, unknown>)[field] !==
            undefined,
   );

   if (unsupportedFields.length > 0) {
      throw new Error(
         `Unsupported DuckDB connection field(s): ${unsupportedFields.join(
            ", ",
         )}. Publisher only supports attachedDatabases for environment-authored DuckDB connections.`,
      );
   }
}

function cloneApiConnection(connection: ApiConnection): ApiConnection {
   return { ...connection };
}

function getStaticConnectionAttributes(
   type: ApiConnection["type"],
): components["schemas"]["ConnectionAttributes"] | undefined {
   switch (type) {
      case "postgres":
         return {
            dialectName: "postgres",
            isPool: false,
            canPersist: true,
            canStream: true,
         };
      case "bigquery":
         return {
            dialectName: "standardsql",
            isPool: false,
            canPersist: true,
            canStream: true,
         };
      case "snowflake":
         return {
            dialectName: "snowflake",
            isPool: true,
            canPersist: true,
            canStream: true,
         };
      case "trino":
         return {
            dialectName: "trino",
            isPool: false,
            canPersist: true,
            canStream: false,
         };
      case "databricks":
         return {
            dialectName: "databricks",
            isPool: false,
            canPersist: true,
            canStream: false,
         };
      case "mysql":
         return {
            dialectName: "mysql",
            isPool: false,
            canPersist: true,
            canStream: false,
         };
      case "duckdb":
      case "motherduck":
      case "ducklake":
         return {
            dialectName: "duckdb",
            isPool: false,
            canPersist: true,
            canStream: true,
         };
      default:
         return undefined;
   }
}

type ServiceAccountKey = {
   type?: string;
   environment_id?: string;
   private_key?: string;
   client_email?: string;
   [key: string]: unknown;
};

function parseServiceAccountKey(json?: string): ServiceAccountKey | undefined {
   if (!json) return undefined;
   const keyData = JSON.parse(json) as ServiceAccountKey;
   const requiredFields = ["type", "project_id", "private_key", "client_email"];
   for (const field of requiredFields) {
      if (!keyData[field]) {
         throw new Error(
            `Invalid service account key: missing "${field}" field`,
         );
      }
   }
   if (keyData.type !== "service_account") {
      throw new Error('Invalid service account key: incorrect "type" field');
   }
   return keyData;
}

function buildPostgresConnectionString(
   config: components["schemas"]["PostgresConnection"],
): string | undefined {
   if (config.connectionString || !process.env.PGSSLMODE) {
      return config.connectionString;
   }

   const params = new URLSearchParams();
   params.set("sslmode", process.env.PGSSLMODE);
   const auth =
      config.userName && config.password
         ? `${encodeURIComponent(config.userName)}:${encodeURIComponent(
              config.password,
           )}@`
         : config.userName
           ? `${encodeURIComponent(config.userName)}@`
           : "";
   const host = config.host ?? "localhost";
   const port = config.port ? `:${config.port}` : "";
   const database = config.databaseName
      ? `/${encodeURIComponent(config.databaseName)}`
      : "";
   return `postgresql://${auth}${host}${port}${database}?${params.toString()}`;
}

function buildDuckdbEntry(
   name: string,
   environmentPath: string,
   databaseFilename = `${name}.duckdb`,
): CoreConnectionEntry {
   return {
      is: "duckdb",
      databasePath: path.join(environmentPath, databaseFilename),
   };
}

function validateConnectionShape(connection: ApiConnection): void {
   if (connection.proxy) {
      // A connection proxy makes THIS server open an outbound SSH tunnel to a
      // tenant-configured bastion. It's a normal connection capability,
      // authorized by whoever configures the connection — deliberately NOT gated
      // by an env flag, and kept separate from the `publisher` HTTP multi-hop
      // type's PUBLISHER_ALLOW_PROXY_CONNECTIONS gate below (that flag is about
      // publisher-to-publisher proxying, a distinct operator decision). Host-key
      // pinning is fail-closed at connect time (see openProxy).
      if (connection.proxy.type !== "ssh") {
         throw new Error(
            `Connection '${connection.name}' has an unsupported proxy type '${connection.proxy.type}'. Only 'ssh' is supported.`,
         );
      }
      if (connection.type !== "postgres") {
         throw new Error(
            `Connection proxy is not supported for type '${connection.type}' (only 'postgres' today).`,
         );
      }
      if (!connection.proxy.ssh) {
         throw new Error(
            `Connection proxy on '${connection.name}' has type 'ssh' but no 'ssh' config object.`,
         );
      }
      // The tunnel forwards to an explicit host:port; the connectionString form
      // can't be rewritten to the local endpoint. Reject it outright when a
      // proxy is set — normal postgres gives connectionString precedence over
      // host/port, so a config carrying BOTH would silently tunnel to
      // host/port and ignore the connectionString, connecting to a different
      // database than the operator configured. Require discrete host/port.
      if (connection.postgresConnection?.connectionString) {
         throw new Error(
            `Connection proxy on '${connection.name}' does not support the connectionString form; ` +
               `provide discrete host and port instead (the tunnel forwards to an explicit endpoint).`,
         );
      }
      if (
         !connection.postgresConnection?.host ||
         !connection.postgresConnection?.port
      ) {
         throw new Error(
            `Connection proxy on '${connection.name}' requires explicit host and port on the ` +
               `postgres connection; the connectionString form is not supported with a proxy.`,
         );
      }
   }

   switch (connection.type) {
      case "postgres":
      case "mysql":
      case "bigquery":
         break;
      case "duckdb":
         if (!connection.duckdbConnection) {
            throw new Error("DuckDB connection configuration is missing.");
         }
         {
            const attached =
               connection.duckdbConnection.attachedDatabases ?? [];
            if (attached.length === 0) {
               throw new Error(
                  `DuckDB connection "${connection.name}" has no attached databases. Add at least one foreign database (BigQuery, Snowflake, Postgres, GCS, S3, Azure) to attachedDatabases, or remove this connection entirely — each package already gets a per-package DuckDB sandbox named "duckdb" automatically.`,
               );
            }
         }
         break;
      case "motherduck":
         if (!connection.motherduckConnection) {
            throw new Error("MotherDuck connection configuration is missing.");
         }
         if (!connection.motherduckConnection.accessToken) {
            throw new Error("MotherDuck access token is required.");
         }
         break;
      case "trino":
         if (!connection.trinoConnection) {
            throw new Error("Trino connection configuration is missing.");
         }
         break;
      case "databricks": {
         const databricks = connection.databricksConnection;
         if (!databricks) {
            throw new Error("Databricks connection configuration is missing.");
         }
         if (!databricks.host) {
            throw new Error("Databricks host is required.");
         }
         if (!databricks.path) {
            throw new Error("Databricks SQL warehouse HTTP path is required.");
         }
         const hasToken = !!databricks.token;
         const hasOAuth =
            !!databricks.oauthClientId && !!databricks.oauthClientSecret;
         if (!hasToken && !hasOAuth) {
            throw new Error(
               "Databricks requires either a personal access token or OAuth M2M client ID and secret.",
            );
         }
         const hasDefaultCatalog = !!databricks.defaultCatalog;
         if (!hasDefaultCatalog) {
            throw new Error("Databricks default catalog is required.");
         }
         break;
      }
      case "snowflake": {
         const snowflakeConnection = connection.snowflakeConnection;
         if (!snowflakeConnection) {
            throw new Error("Snowflake connection configuration is missing.");
         }
         if (!snowflakeConnection.account) {
            throw new Error("Snowflake account is required.");
         }
         if (!snowflakeConnection.username) {
            throw new Error("Snowflake username is required.");
         }
         if (!snowflakeConnection.password && !snowflakeConnection.privateKey) {
            throw new Error(
               "Snowflake password or private key or private key path is required.",
            );
         }
         if (!snowflakeConnection.warehouse) {
            throw new Error("Snowflake warehouse is required.");
         }
         break;
      }
      case "publisher": {
         // SSRF gate (default-deny / fail-closed). A `publisher` connection
         // makes THIS server issue outbound HTTP to a tenant-controlled
         // `connectionUri` (both the query path in db-publisher's
         // PublisherConnection and the introspection path in db_utils). That is
         // the intended behavior for local `--watch-env` authoring, but in a
         // hosted multi-tenant deployment (e.g. Credible running this server) it
         // is an SSRF surface. Require an explicit opt-in so the type is refused
         // unless the operator deliberately enabled it. This is the single
         // choke point — every connection passes validateConnectionShape before
         // it can be assembled, queried, or introspected — so denying here shuts
         // off all three at once.
         if (process.env.PUBLISHER_ALLOW_PROXY_CONNECTIONS !== "true") {
            throw new Error(
               `Publisher proxy connection '${connection.name}' is disabled in this deployment. ` +
                  `'publisher' connections make the server issue outbound requests to a configured connectionUri, ` +
                  `which is only appropriate for local --watch-env authoring. ` +
                  `Fix: set the environment variable PUBLISHER_ALLOW_PROXY_CONNECTIONS=true to enable them.`,
            );
         }
         const publisher = connection.publisherConnection;
         if (!publisher?.connectionUri) {
            throw new Error(
               `Invalid publisher connection '${connection.name}': missing connectionUri. ` +
                  `Fix: { "name": "${connection.name}", "type": "publisher", ` +
                  `"publisherConnection": { "connectionUri": "https://…/connections/${connection.name}", "accessToken": "<jwt>" } }`,
            );
         }
         // Reject a malformed connectionUri here, at config-load, rather than
         // letting it fail deep in the request path — where the thrown error can
         // echo the raw value back, leaking any credentials embedded in it
         // (`redactUrlCredentials` returns the URI unchanged when it can't parse
         // it). Never include the raw connectionUri in these messages; the
         // scheme is safe to name.
         let parsedUri: URL;
         try {
            parsedUri = new URL(publisher.connectionUri);
         } catch {
            throw new Error(
               `Invalid publisher connection '${connection.name}': connectionUri is not a valid URL. ` +
                  `Fix: set connectionUri to an absolute https URL like "https://…/connections/${connection.name}".`,
            );
         }
         if (
            parsedUri.protocol !== "http:" &&
            parsedUri.protocol !== "https:"
         ) {
            throw new Error(
               `Invalid publisher connection '${connection.name}': connectionUri must use http or https (got '${parsedUri.protocol}'). ` +
                  `Fix: set connectionUri to an absolute https URL like "https://…/connections/${connection.name}".`,
            );
         }
         break;
      }
   }
}

export function assembleEnvironmentConnections(
   connections: ApiConnection[] = [],
   environmentPath = "",
): AssembledEnvironmentConnections {
   const pojo: CoreConnectionsPojo = { connections: {} };
   const metadata = new Map<string, EnvironmentConnectionMetadata>();
   const apiConnections: ApiConnection[] = [];
   const processedConnections = new Set<string>();

   for (const connection of connections) {
      if (!connection.name) {
         throw new Error("Invalid connection configuration. No name.");
      }

      if (processedConnections.has(connection.name)) {
         continue;
      }

      if (connection.name === "duckdb") {
         throw new Error(
            "Connection name 'duckdb' is reserved for per-package sandboxes. Choose a different name for environment-level DuckDB connections (e.g. 'shared_duckdb').",
         );
      }

      processedConnections.add(connection.name);
      validateDuckdbApiSurface(connection);
      validateConnectionShape(connection);

      const apiConnection = cloneApiConnection(connection);
      apiConnection.attributes = getStaticConnectionAttributes(connection.type);
      const attachedDatabases =
         connection.duckdbConnection?.attachedDatabases ?? [];
      const isDuckLake = connection.type === "ducklake";
      const isDuckdb = connection.type === "duckdb";
      const databasePath = isDuckLake
         ? path.join(environmentPath, `${connection.name}_ducklake.duckdb`)
         : isDuckdb
           ? path.join(environmentPath, `${connection.name}.duckdb`)
           : undefined;

      metadata.set(connection.name, {
         apiConnection,
         attachedDatabases,
         hasAzureAttachment: attachedDatabases.some(
            (database) => database.type === "azure",
         ),
         hasSnowflakePrivateKey:
            connection.type === "snowflake" &&
            !!connection.snowflakeConnection?.privateKey,
         isDuckLake,
         databasePath,
         workingDirectory: environmentPath,
         proxy: connection.proxy,
      });

      switch (connection.type) {
         case "postgres": {
            const postgresConnection = connection.postgresConnection;
            pojo.connections[connection.name] = {
               is: "postgres",
               host: postgresConnection?.host,
               port: postgresConnection?.port,
               username: postgresConnection?.userName,
               password: postgresConnection?.password,
               databaseName: postgresConnection?.databaseName,
               connectionString: postgresConnection
                  ? buildPostgresConnectionString(postgresConnection)
                  : undefined,
            };
            break;
         }

         case "mysql": {
            pojo.connections[connection.name] = {
               is: "mysql",
               host: connection.mysqlConnection?.host,
               port: connection.mysqlConnection?.port,
               user: connection.mysqlConnection?.user,
               password: connection.mysqlConnection?.password,
               database: connection.mysqlConnection?.database,
            };
            break;
         }

         case "bigquery": {
            const serviceAccountKey = parseServiceAccountKey(
               connection.bigqueryConnection?.serviceAccountKeyJson as
                  | string
                  | undefined,
            );
            pojo.connections[connection.name] = {
               is: "bigquery",
               projectId:
                  connection.bigqueryConnection?.defaultProjectId ??
                  serviceAccountKey?.environment_id,
               serviceAccountKey,
               location: connection.bigqueryConnection?.location,
               maximumBytesBilled:
                  connection.bigqueryConnection?.maximumBytesBilled,
               timeoutMs:
                  connection.bigqueryConnection?.queryTimeoutMilliseconds,
               billingProjectId:
                  connection.bigqueryConnection?.billingProjectId,
            };
            break;
         }

         case "snowflake": {
            pojo.connections[connection.name] = {
               is: "snowflake",
               account: connection.snowflakeConnection?.account,
               username: connection.snowflakeConnection?.username,
               password: connection.snowflakeConnection?.password,
               privateKey: connection.snowflakeConnection?.privateKey
                  ? normalizeSnowflakePrivateKey(
                       connection.snowflakeConnection.privateKey,
                    )
                  : undefined,
               privateKeyPass: connection.snowflakeConnection?.privateKeyPass,
               warehouse: connection.snowflakeConnection?.warehouse,
               database: connection.snowflakeConnection?.database,
               schema: connection.snowflakeConnection?.schema,
               role: connection.snowflakeConnection?.role,
               timeoutMs:
                  connection.snowflakeConnection?.responseTimeoutMilliseconds,
               // Pool sizing is server-owned policy (matches the values
               // main's deleted switch passed pre-MalloyConfig adoption).
               // Not exposed through the public API.
               poolMin: 1,
               poolMax: 20,
            };
            break;
         }

         case "trino": {
            pojo.connections[connection.name] = {
               is: "trino",
               ...validateAndBuildTrinoCoreConfig(connection.trinoConnection),
            };
            break;
         }

         case "databricks": {
            const databricks = connection.databricksConnection;
            pojo.connections[connection.name] = {
               is: "databricks",
               host: databricks?.host,
               path: databricks?.path,
               token: databricks?.token,
               oauthClientId: databricks?.oauthClientId,
               oauthClientSecret: databricks?.oauthClientSecret,
               defaultCatalog: databricks?.defaultCatalog,
               defaultSchema: databricks?.defaultSchema,
               setupSQL: databricks?.setupSQL,
            };
            break;
         }

         case "duckdb": {
            if (
               attachedDatabases.some(
                  (database) => database.name === connection.name,
               )
            ) {
               throw new Error(
                  `DuckDB attached database names cannot conflict with connection name ${connection.name}`,
               );
            }
            pojo.connections[connection.name] = buildDuckdbEntry(
               connection.name,
               environmentPath,
               `${connection.name}.duckdb`,
            );
            break;
         }

         case "motherduck": {
            if (!connection.motherduckConnection?.accessToken) {
               throw new Error("MotherDuck access token is required.");
            }

            pojo.connections[connection.name] = {
               is: "duckdb",
               databasePath: connection.motherduckConnection.database
                  ? `md:${connection.motherduckConnection.database}?attach_mode=single`
                  : "md:",
               motherDuckToken: connection.motherduckConnection.accessToken,
            };
            break;
         }

         case "ducklake": {
            if (!connection.ducklakeConnection) {
               throw new Error("DuckLake connection configuration is missing.");
            }
            if (!connection.ducklakeConnection.catalog?.postgresConnection) {
               throw new Error(
                  `PostgreSQL connection configuration is required for DuckLake catalog: ${connection.name}`,
               );
            }
            pojo.connections[connection.name] = buildDuckdbEntry(
               connection.name,
               environmentPath,
               `${connection.name}_ducklake.duckdb`,
            );
            break;
         }

         case "publisher": {
            // connectionUri presence is validated by validateConnectionShape
            // above. The proxied dataplane owns auth and read-only enforcement;
            // PublisherConnection itself does not reject writes. The real
            // dialect is the remote connection's and is resolved at runtime by
            // the live connection, so getStaticConnectionAttributes returns
            // undefined for publisher (falls through to its default).
            const publisher = connection.publisherConnection!;
            pojo.connections[connection.name] = {
               is: "publisher",
               connectionUri: publisher.connectionUri,
               accessToken: publisher.accessToken,
            };
            break;
         }

         default: {
            throw new Error(`Unsupported connection type: ${connection.type}`);
         }
      }

      apiConnections.push(apiConnection);
   }

   return { pojo, metadata, apiConnections };
}

function validateAndBuildTrinoCoreConfig(
   trinoConfig: components["schemas"]["TrinoConnection"] | undefined,
): Record<string, unknown> {
   if (!trinoConfig) {
      return {};
   }

   const server =
      trinoConfig.server && trinoConfig.port
         ? trinoConfig.server.includes(trinoConfig.port.toString())
            ? trinoConfig.server
            : `${trinoConfig.server}:${trinoConfig.port}`
         : trinoConfig.server;

   const baseConfig: Record<string, unknown> = {
      server,
      port: trinoConfig.port,
      catalog: trinoConfig.catalog,
      schema: trinoConfig.schema,
      user: trinoConfig.user,
   };

   if (trinoConfig.peakaKey) {
      baseConfig.extraCredential = {
         peakaKey: trinoConfig.peakaKey,
      };
      return baseConfig;
   }

   if (server?.startsWith("https://") && trinoConfig.password) {
      baseConfig.password = trinoConfig.password;
   }

   if (server?.startsWith("http://") || server?.startsWith("https://")) {
      return baseConfig;
   }

   throw new Error(
      `Invalid Trino connection: expected "http://server:port" or "https://server:port".`,
   );
}