// mssql's admin client — mssql's own code, living in the module (novox/hq ADR 0039). Both this
// module's tools and its provisioner import it, and nothing outside mssql does.
//
// SQL is executed through `sqlcmd`, not a wire-protocol driver: the module may take NO npm
// dependency beyond @novox/mesh-sdk, and hand-rolling the TDS handshake, pre-login and query
// protocol is more surface than this should carry — so it shells out to the client the mssql
// tools ship, the same way postgres drives itself through `psql`, minio through `mc`, and mailu
// through doveadm. One boundary, `run()`, and every method is built on it.
//
// Structured rows come back as JSON: SQL Server itself renders the result with `FOR JSON`, and
// this parses the single JSON document sqlcmd prints — far more robust than parsing sqlcmd's
// column-aligned text, since SQL Server owns the quoting and typing.

import { randomBytes } from "node:crypto";
import { readFileSync } from "node:fs";
import { execFile } from "node:child_process";
import { promisify } from "node:util";

const run = promisify(execFile);

export interface QueryResult {
  /** The leading keyword of the statement, e.g. "SELECT", "CREATE". */
  readonly command: string;
  readonly rows: Record<string, unknown>[];
}

export interface MssqlConn {
  readonly host: string;
  readonly port: number;
  readonly user: string;
  readonly password: string;
  /**
   * The read-only login's password, which the mesh mints for this module (`own-secrets.reader`).
   * Absent when the mesh has not delivered it: then a caller's statement is refused, never run as
   * the administrator (novox/hq issue 193).
   */
  readonly readerPassword?: string;
}

/**
 * The login a caller's statement runs as (novox/hq issue 193). It may connect to every database and
 * read every table, and holds no other permission. A statement cannot climb out of a login the way it
 * could out of a transaction wrapped around it as text, and the administrator — who can run programs
 * on the server — never runs a caller's text.
 */
export const READER = "mesh_mssql_reader";

/** Who a sqlcmd invocation logs in as, and whether the text is a caller's rather than the module's. */
interface Invocation {
  readonly user: string;
  readonly password: string;
  /**
   * A caller's text: sqlcmd substitutes no `$(NAME)` in it, which would read this process's
   * environment — the administrator's password among it. (Its own commands are kept out by the
   * caller's text never beginning a line; see readOnlyQuery.)
   */
  readonly caller: boolean;
}

export class MssqlClient {
  constructor(private readonly conn: MssqlConn) {}

  /** The reader is made once per process: idempotent, and repeating it re-sets a rotated password. */
  private readerReady?: Promise<void>;

  /**
   * Build from the module's resolved environment. Reads MESH_MSSQL_* first (the documented
   * names), falling back to the MESH_PROVISION_* keys the manifest already sets on the provisioner
   * container. Throws if it cannot find a host and an admin password.
   */
  static fromEnv(env: NodeJS.ProcessEnv = process.env): MssqlClient {
    const url = env.MESH_PROVISION_MSSQL ? safeUrl(env.MESH_PROVISION_MSSQL) : undefined;
    const host = env.MESH_MSSQL_HOST ?? url?.hostname;
    const port = Number(env.MESH_MSSQL_PORT ?? url?.port ?? "1433") || 1433;
    const user = env.MESH_MSSQL_USER ?? url?.username ?? "sa";
    const password = env.MESH_MSSQL_PASSWORD ?? readSecretFile(env.MESH_PROVISION_PASSWORD_FILE);
    if (!host || !password) {
      throw new Error("mssql host or admin password is not set — mssql's own code cannot reach the server");
    }
    const readerPassword = env.MESH_MSSQL_READER_PASSWORD ??
      readSecretFile(env.MESH_MSSQL_READER_PASSWORD_FILE);
    return new MssqlClient({ host, port, user, password, readerPassword });
  }

  get host(): string {
    return this.conn.host;
  }

  get port(): number {
    return this.conn.port;
  }

  /**
   * Execute a batch that returns no rows (DDL and the like), through `sqlcmd`. The password is
   * passed by SQLCMDPASSWORD, never on argv, the way postgres passes PGPASSWORD; `-b` makes a
   * failed statement an error here rather than a success with a warning, and `-C` trusts the
   * server's self-signed certificate the mssql image ships with.
   */
  async exec(sql: string, database = "master"): Promise<void> {
    await this.sqlcmd(sql, database);
  }

  /**
   * Run a SELECT and return its rows as objects. The caller's SQL must be a single SELECT; it is
   * wrapped so SQL Server renders the result with `FOR JSON PATH`, and the JSON document sqlcmd
   * prints (split across output lines for a large result, and reassembled here) is parsed. An
   * empty result yields no output at all — an empty array.
   */
  async query(
    select: string,
    database = "master",
    variables: Record<string, string> = {},
  ): Promise<Record<string, unknown>[]> {
    const wrapped = `SET NOCOUNT ON;\n${stripTrailingSemis(select)}\nFOR JSON PATH, INCLUDE_NULL_VALUES;`;
    const stdout = await this.sqlcmd(wrapped, database, variables);
    return parseJsonRows(stdout);
  }

  /** The one execution boundary: invoke `sqlcmd` and return its concatenated stdout. */
  private async sqlcmd(
    sql: string,
    database: string,
    variables: Record<string, string> = {},
    as: Invocation = { user: this.conn.user, password: this.conn.password, caller: false },
  ): Promise<string> {
    // `-h -1` drops the column-header rule; `-y 0`/`-Y 0` lift the display-width cap so a long
    // JSON document is not truncated; `-W` trims trailing whitespace so the JSON chunks rejoin
    // cleanly. sqlcmd from the mssql-tools ships in the runtime container, the way `psql` ships
    // with postgres's — the module owns its own code (ADR 0039) and shells out to it.
    const { stdout } = await run(
      "sqlcmd",
      [
        "-S", `${this.conn.host},${this.conn.port}`,
        "-U", as.user,
        "-d", database,
        ...(as.caller ? ["-x"] : []),
        "-C",
        "-b",
        "-h", "-1",
        "-y", "0",
        "-Y", "0",
        "-W",
        "-Q", sql,
      ],
      // `variables` reach sqlcmd as environment variables, which it substitutes as `$(NAME)` scripting
      // variables: a value that must not appear on argv, or in the message of a failed command.
      { env: { ...process.env, ...variables, SQLCMDPASSWORD: as.password }, maxBuffer: 16 << 20 },
    );
    return stdout;
  }

  /**
   * Create a login and a database it owns (mapped as a db_owner user), idempotently. The login,
   * the database and the user all carry the consumer's minted name, so the consumer owns exactly
   * its own database — a name it cannot learn is a database it cannot reach (ADR 0048).
   */
  async createDatabaseAndLogin(database: string, login: string, password: string): Promise<void> {
    const logins = await this.query(
      `SELECT 1 AS ok FROM sys.server_principals WHERE name = ${literal(login)}`,
    );
    if (logins.length === 0) {
      await this.exec(
        `CREATE LOGIN ${ident(login)} WITH PASSWORD = ${literal(password)}, CHECK_POLICY = OFF`,
      );
    } else {
      await this.exec(`ALTER LOGIN ${ident(login)} WITH PASSWORD = ${literal(password)}`);
      // A disabled login is refused like a wrong password; the check the provisioner runs reports it
      // lost, so applying again must enable it or the two would disagree for ever.
      await this.exec(`ALTER LOGIN ${ident(login)} ENABLE`);
    }

    const dbs = await this.query(
      `SELECT 1 AS ok FROM sys.databases WHERE name = ${literal(database)}`,
    );
    if (dbs.length === 0) {
      // CREATE DATABASE must stand alone in its batch; it runs as its own sqlcmd invocation.
      await this.exec(`CREATE DATABASE ${ident(database)}`);
    }

    // Map the login to a db_owner user inside the database it owns.
    const users = await this.query(
      `SELECT 1 AS ok FROM sys.database_principals WHERE name = ${literal(login)}`,
      database,
    );
    if (users.length === 0) {
      await this.exec(`CREATE USER ${ident(login)} FOR LOGIN ${ident(login)}`, database);
    } else {
      // Re-point an existing user at the login when its SID is not the login's: a database restored
      // from elsewhere keeps its user under the old login's SID, orphaned. Only then, so a user that
      // is already mapped is left alone.
      const orphaned = await this.query(
        `SELECT 1 AS ok FROM sys.database_principals WHERE name = ${literal(login)} ` +
          `AND (sid IS NULL OR sid <> SUSER_SID(${literal(login)}))`,
        database,
      );
      if (orphaned.length > 0) {
        await this.exec(`ALTER USER ${ident(login)} WITH LOGIN = ${ident(login)}`, database);
      }
    }
    await this.exec(`ALTER ROLE db_owner ADD MEMBER ${ident(login)}`, database);
  }

  /**
   * Whether `login` exists, is enabled, has exactly `password`, and is a db_owner user of
   * `database`. Read-only: the password is compared with PWDCOMPARE against the stored hash, so
   * nothing logs in and no failed-login is recorded (novox/hq issue 120).
   */
  async holdsLogin(database: string, login: string, password: string): Promise<boolean> {
    // The password reaches sqlcmd as a scripting variable from the environment, never inside the
    // query text, so it is neither on argv nor in the message of a failed command. It is the mesh's
    // minted value, which carries no quote.
    const server = await this.query(
      `SELECT CAST(CASE WHEN EXISTS (SELECT 1 FROM sys.sql_logins WHERE name = ${literal(login)} ` +
        `AND is_disabled = 0 AND PWDCOMPARE(N'$(MESHHOLDSPW)', password_hash) = 1) ` +
        `AND DB_ID(${literal(database)}) IS NOT NULL THEN 1 ELSE 0 END AS int) AS ok`,
      "master",
      { MESHHOLDSPW: password },
    );
    if (Number(server[0]?.ok) !== 1) return false;
    // The user must be this login's, by SID, and a db_owner. A user orphaned by a restore has the
    // right name and the wrong SID, and cannot be reached through the login.
    const owner = await this.query(
      `SELECT CAST(CASE WHEN EXISTS (SELECT 1 FROM sys.database_principals dp ` +
        `JOIN sys.server_principals sp ON dp.sid = sp.sid ` +
        `WHERE dp.name = ${literal(login)} AND sp.name = ${literal(login)}) ` +
        `AND IS_ROLEMEMBER('db_owner', ${literal(login)}) = 1 THEN 1 ELSE 0 END AS int) AS ok`,
      database,
    );
    return Number(owner[0]?.ok) === 1;
  }

  /** Drop a database and its login, idempotently, after evicting live connections. */
  async dropDatabaseAndLogin(database: string, login: string): Promise<void> {
    const dbs = await this.query(
      `SELECT 1 AS ok FROM sys.databases WHERE name = ${literal(database)}`,
    );
    if (dbs.length > 0) {
      // SINGLE_USER WITH ROLLBACK IMMEDIATE evicts every other session before the drop.
      await this.exec(`ALTER DATABASE ${ident(database)} SET SINGLE_USER WITH ROLLBACK IMMEDIATE`);
      await this.exec(`DROP DATABASE ${ident(database)}`);
    }
    const logins = await this.query(
      `SELECT 1 AS ok FROM sys.server_principals WHERE name = ${literal(login)}`,
    );
    if (logins.length > 0) {
      await this.exec(`DROP LOGIN ${ident(login)}`);
    }
  }

  /** List the user databases (database_id > 4 excludes the system four), with size, for the tool. */
  async listDatabases(): Promise<{ name: string; sizeBytes: number; state: string }[]> {
    const rows = await this.query(
      "SELECT d.name AS name, d.state_desc AS state, " +
        "SUM(CAST(f.size AS bigint)) * 8 * 1024 AS size_bytes " +
        "FROM sys.databases d JOIN sys.master_files f ON d.database_id = f.database_id " +
        "WHERE d.database_id > 4 GROUP BY d.name, d.state_desc ORDER BY d.name",
    );
    return rows.map((r) => ({
      name: String(r.name),
      sizeBytes: Number(r.size_bytes ?? 0),
      state: String(r.state ?? ""),
    }));
  }

  /**
   * Make the read-only login, idempotently, with the password the mesh minted for it: it may connect
   * to every database and read every table, and is taken out of the administrators' role should
   * anyone have put it there. Run as the administrator, because only it can make a login.
   */
  async ensureReader(): Promise<void> {
    const password = this.conn.readerPassword;
    if (!password) throw readerMissing();
    const logins = await this.query(
      `SELECT 1 AS ok FROM sys.server_principals WHERE name = ${literal(READER)}`,
    );
    if (logins.length === 0) {
      await this.exec(
        `CREATE LOGIN ${ident(READER)} WITH PASSWORD = ${literal(password)}, CHECK_POLICY = OFF`,
      );
    } else {
      await this.exec(`ALTER LOGIN ${ident(READER)} WITH PASSWORD = ${literal(password)}`);
      await this.exec(`ALTER LOGIN ${ident(READER)} ENABLE`);
    }
    await this.exec(
      `IF IS_SRVROLEMEMBER('sysadmin', ${literal(READER)}) = 1 ` +
        `ALTER SERVER ROLE sysadmin DROP MEMBER ${ident(READER)}`,
    );
    await this.exec(`GRANT CONNECT ANY DATABASE TO ${ident(READER)}`);
    await this.exec(`GRANT SELECT ALL USER SECURABLES TO ${ident(READER)}`);
  }

  /**
   * Run a caller's SELECT against a named database as the read-only login, for the mssql_query tool
   * (novox/hq issue 193). Read-only by the login, not by a transaction wrapped around the text; the
   * rows are rendered by FOR JSON. Never as the administrator: without the reader's password the call
   * is refused.
   */
  async readOnlyQuery(database: string, sql: string): Promise<QueryResult> {
    const password = this.conn.readerPassword;
    if (!password) throw readerMissing();
    // **One line, refused otherwise.** sqlcmd reads a line that BEGINS with `:` or `!!` as its own
    // command rather than SQL, and `:!!` starts a program in this container, which holds the
    // administrator's password. Its switch for refusing those (-X) makes it ignore -Q in the
    // version shipped here, so instead no line of a caller's text can begin one: the text follows
    // this module's own on the first line, and a line break in it is refused. Proven on a throwaway
    // server: the same text at the start of a line ran a program; mid-line it is a syntax error.
    if (/[\r\n]/.test(sql)) {
      throw new Error(
        "mssql_query: the statement must be one line — sqlcmd takes a line beginning with ':' or " +
          "'!!' as a command of its own, which can start a program (novox/hq issue 193)",
      );
    }
    this.readerReady ??= this.ensureReader().catch((err) => {
      this.readerReady = undefined; // asked again next call, not failed for the process's life
      throw err;
    });
    await this.readerReady;
    const stdout = await this.sqlcmd(
      `SET NOCOUNT ON; ${stripTrailingSemis(sql)}\nFOR JSON PATH, INCLUDE_NULL_VALUES;`,
      database,
      {},
      { user: READER, password, caller: true },
    );
    const command = /^\s*([A-Za-z]+)/.exec(sql)?.[1]?.toUpperCase() ?? "";
    return { command, rows: parseJsonRows(stdout) };
  }
}

function readerMissing(): Error {
  return new Error(
    "the read-only login's password was not delivered (own-secrets.reader, " +
      "MESH_MSSQL_READER_PASSWORD_FILE), so the statement is refused rather than run as the " +
      "administrator (novox/hq issue 193)",
  );
}

/** Generate a URL-safe password. */
export function generatePassword(): string {
  return randomBytes(24).toString("base64url");
}

/** Quote a T-SQL identifier (square brackets, doubled internal `]`). */
export function ident(id: string): string {
  return "[" + id.replace(/]/g, "]]") + "]";
}

/** Quote a T-SQL string literal (single quotes, doubled internal quotes). */
export function literal(val: string): string {
  return "'" + val.replace(/'/g, "''") + "'";
}

/** Strip trailing semicolons and whitespace so FOR JSON can be appended to a caller's SELECT. */
function stripTrailingSemis(sql: string): string {
  return sql.replace(/[\s;]+$/, "");
}

function readSecretFile(path: string | undefined): string | undefined {
  if (!path) return undefined;
  try {
    return readFileSync(path, "utf8").trim();
  } catch {
    return undefined;
  }
}

function safeUrl(raw: string): URL | undefined {
  try {
    return new URL(raw);
  } catch {
    return undefined;
  }
}

/**
 * Parse the JSON a FOR JSON query prints through sqlcmd. SQL Server splits a large FOR JSON result
 * into ~2033-character chunks, one per output row; with `-h -1 -W` each lands on its own line, so
 * the document is reassembled by concatenating the non-empty lines. No output (an empty result, or
 * a pure DDL batch) means no rows.
 */
function parseJsonRows(stdout: string): Record<string, unknown>[] {
  const joined = stdout
    .split(/\r?\n/)
    .map((l) => l.trimEnd())
    .filter((l) => l.length > 0)
    .join("");
  if (joined.length === 0) return [];
  const parsed = JSON.parse(joined);
  return Array.isArray(parsed) ? (parsed as Record<string, unknown>[]) : [parsed as Record<string, unknown>];
}