import { getServiceAccountToken } from '@axinom/mosaic-id-link-be';
import { isNullOrWhitespace } from '@axinom/mosaic-service-common';
import axios, { AxiosError, type AxiosInstance } from 'axios';
import chalk from 'chalk';
import net from 'net';
import ora from 'ora';
import WebSocket, { createWebSocketStream } from 'ws';
import { exitCode } from '../../../../common/index.js';
import type { DatabaseProxyOptions } from './database-proxy-options.js';

const { red, yellow, green, dim, bold } = chalk;

export const DEFAULT_LOCAL_PROXY_PORT = 5433;

/** Only loopback is ever bound: the credential is not meant to leave the machine. */
const LOOPBACK = '127.0.0.1';

/** Declared here rather than imported: the Hosting schema types are private. */
export interface DatabaseAccessSession {
  databaseAccessSessionId: string;
  username: string;
  password: string;
  databaseName: string;
  expiresAt: string;
  databaseProxyUrl: string;
}

const OPEN_SESSION = `mutation OpenDatabaseAccessSession($input: OpenDatabaseAccessSessionInput!) {
  openDatabaseAccessSession(input: $input) {
    databaseAccessSessionId
    username
    password
    databaseName
    expiresAt
    databaseProxyUrl
  }
}`;

const CLOSE_SESSION = `mutation CloseDatabaseAccessSession($input: CloseDatabaseAccessSessionInput!) {
  closeDatabaseAccessSession(input: $input) {
    status
    terminatedBackends
  }
}`;

/** Connection details go to stdout and nothing else does, so `--json` pipes. */
const writeOut = (text: string): void => {
  process.stdout.write(text + '\n');
};
const writeStatus = (text: string): void => {
  process.stderr.write(text + '\n');
};

export const minutesUntil = (expiresAt: string, now = Date.now()): number =>
  Math.max(0, Math.floor((new Date(expiresAt).getTime() - now) / 60_000));

/**
 * Hosting may grant less than was asked for. The minute of tolerance absorbs
 * the round trip.
 */
export const formatClampedTtlWarning = (
  requestedMinutes: number | undefined,
  expiresAt: string,
  now = Date.now(),
): string | undefined => {
  if (requestedMinutes === undefined) {
    return undefined;
  }
  const granted = minutesUntil(expiresAt, now);
  if (granted >= requestedMinutes - 1) {
    return undefined;
  }
  return `Requested ${requestedMinutes} minutes, granted ${granted} - database access for this service expires then.`;
};

export const WARN_BEFORE_EXPIRY_MS = 5 * 60_000;

/** The byte a terminal delivers for Ctrl+C once it stops raising SIGINT. */
const ETX = 0x03;

/** Undefined when the session is too short for the warning to precede it. */
export const toExpiryWarningDelay = (
  expiresAt: string,
  now = Date.now(),
): number | undefined => {
  const remaining = new Date(expiresAt).getTime() - now;
  return remaining > WARN_BEFORE_EXPIRY_MS
    ? remaining - WARN_BEFORE_EXPIRY_MS
    : undefined;
};

export const toConnectionJson = (
  session: DatabaseAccessSession,
  port: number,
): Record<string, string | number> => ({
  databaseAccessSessionId: session.databaseAccessSessionId,
  host: LOOPBACK,
  port,
  database: session.databaseName,
  user: session.username,
  password: session.password,
  expiresAt: session.expiresAt,
});

export const formatConnectionDetails = (
  session: DatabaseAccessSession,
  port: number,
  now = Date.now(),
): string => {
  const rows: [string, string][] = [
    ['Host', LOOPBACK],
    ['Port', String(port)],
    ['Database', session.databaseName],
    ['User', session.username],
    ['Password', session.password],
    [
      'Expires',
      `${session.expiresAt} (${minutesUntil(session.expiresAt, now)} minutes)`,
    ],
  ];
  return rows
    .map(([label, value]) => `  ${label.padEnd(9)} ${value}`)
    .join('\n');
};

interface GraphQlError {
  message: string;
  code?: string;
}

const getAxiosInstance = (
  hostingServiceBaseUrl: string,
  serviceAccountToken: string,
): AxiosInstance => {
  return axios.create({
    baseURL: new URL(hostingServiceBaseUrl).toString(),
    headers: {
      Authorization: `Bearer ${serviceAccountToken}`,
      'content-type': 'application/json',
    },
  });
};

/** Built per call rather than once: a long session outlives its startup token. */
const callHosting = async <T>(
  hostingServiceBaseUrl: string,
  serviceAccountToken: string,
  query: string,
  variables: unknown,
): Promise<T> => {
  const axiosInstance = getAxiosInstance(
    hostingServiceBaseUrl,
    serviceAccountToken,
  );
  const payload = JSON.stringify({ query, variables });
  const response = await axiosInstance.post('graphql', payload);

  const errors: GraphQlError[] = response.data.errors ?? [];
  if (errors.length > 0) {
    const { message, code } = errors[0];
    throw new Error(code ? `${code}: ${message}` : message);
  }
  return response.data.data as T;
};

/** Caches the token and renews it a minute before it lapses. */
const createTokenProvider = (
  authEndpoint: string,
  clientId: string,
  clientSecret: string,
): (() => Promise<string>) => {
  let token = '';
  let renewAt = 0;
  return async () => {
    if (token === '' || Date.now() >= renewAt) {
      const result = await getServiceAccountToken(
        authEndpoint,
        clientId,
        clientSecret,
      );
      token = result.accessToken;
      renewAt = Date.now() + (result.expiresInSeconds - 60) * 1_000;
    }
    return token;
  };
};

const describeError = (error: unknown): string => {
  const axiosError = error as AxiosError;
  if (axiosError.response) {
    return `${axiosError.message} (HTTP ${
      axiosError.response.status
    }: ${JSON.stringify(axiosError.response.data)})`;
  }
  return error instanceof Error ? error.message : String(error);
};

/**
 * Pipes one accepted TCP connection through one WebSocket, never multiplexed.
 * The bytes are the TLS session the client negotiated with Postgres directly.
 */
const databaseProxyConnection = (
  client: net.Socket,
  session: DatabaseAccessSession,
  getToken: () => Promise<string>,
): void => {
  // Paused until the carrier is open, so nothing the client sends is dropped.
  client.pause();

  void (async () => {
    // Everything up to the carrier existing is guarded: an escape here is an
    // unhandled rejection, which kills the process before the session is
    // closed and strands the database role until it expires.
    let ws: WebSocket;
    try {
      const token = await getToken();
      ws = new WebSocket(session.databaseProxyUrl, {
        headers: {
          authorization: `Bearer ${token}`,
          'x-mosaic-database-access-session': session.databaseAccessSessionId,
        },
      });
    } catch (error) {
      writeStatus(
        red(`Could not reach the database proxy: ${describeError(error)}`),
      );
      client.destroy();
      return;
    }

    // end(), not destroy(): destroy drops bytes still buffered for the client.
    const stop = (): void => {
      client.end();
      if (
        ws.readyState === WebSocket.CONNECTING ||
        ws.readyState === WebSocket.OPEN
      ) {
        ws.close();
      }
    };

    // Piped rather than forwarded by hand: a slow reader on either end pauses
    // the other instead of queueing without limit.
    ws.on('open', () => {
      const carrier = createWebSocketStream(ws);
      carrier.on('error', stop);
      client.pipe(carrier);
      carrier.pipe(client);
      client.resume();
    });
    ws.on('unexpected-response', (_request, response) => {
      writeStatus(
        red(
          `The database proxy refused the connection (HTTP ${response.statusCode}).`,
        ),
      );
      client.destroy();
    });
    ws.on('error', (error) => {
      writeStatus(red(`Database proxy error: ${error.message}`));
      stop();
    });
    ws.on('close', stop);
    client.on('error', stop);
    client.on('close', () => {
      if (
        ws.readyState === WebSocket.CONNECTING ||
        ws.readyState === WebSocket.OPEN
      ) {
        ws.close();
      }
    });
  })();
};

const listen = (server: net.Server, port: number): Promise<void> =>
  new Promise((resolve, reject) => {
    server.once('error', reject);
    server.listen(port, LOOPBACK, () => {
      server.removeListener('error', reject);
      resolve();
    });
  });

/** Hands the terminal back; the mode belongs to it, not to this process. */
const restoreStdinMode = (): void => {
  if (process.stdin.isTTY && process.stdin.isRaw) {
    process.stdin.setRawMode(false);
    process.stdin.pause();
  }
};

/**
 * Calls `onShutdown` once, on Ctrl+C or a termination signal.
 *
 * Ctrl+C is read from stdin rather than taken as a signal: in raw mode the
 * terminal stops raising one, which keeps it from also killing the `yarn` and
 * `env-cmd` wrappers a solution runs this through. They would exit at once and
 * hand the shell prompt back while the session was still closing.
 */
const onShutdownRequested = (onShutdown: () => void): void => {
  let requested = false;
  const request = (): void => {
    if (requested) {
      return;
    }
    requested = true;
    onShutdown();
  };

  process.on('exit', restoreStdinMode);

  // The baseline everywhere: `kill`, a shutting-down machine, Ctrl+Break, and
  // any stdin that is not a terminal. SIGKILL cannot be caught; that falls to
  // the session's expiry.
  for (const signal of ['SIGINT', 'SIGTERM', 'SIGHUP', 'SIGBREAK'] as const) {
    process.on(signal, request);
  }

  // On a terminal, take the key before it becomes a SIGINT that would kill the
  // wrappers too. Raw mode suppresses that signal on every platform.
  if (process.stdin.isTTY) {
    process.stdin.setRawMode(true);
    process.stdin.resume();
    process.stdin.on('data', (chunk: Buffer) => {
      if (chunk.includes(ETX)) {
        request();
      }
    });
  }
};

const openSession = async (
  args: DatabaseProxyOptions,
  hostingServiceBaseUrl: string,
  getToken: () => Promise<string>,
): Promise<DatabaseAccessSession> => {
  // Creating the role is a remote call that regularly takes half a minute.
  const spinner = ora({
    text: `Opening a database access session for [${args.serviceId}]...`,
    // Off: it would put stdin in raw mode, which is this command's to manage.
    discardStdin: false,
  }).start();

  try {
    const data = await callHosting<{
      openDatabaseAccessSession: DatabaseAccessSession;
    }>(hostingServiceBaseUrl, await getToken(), OPEN_SESSION, {
      input: {
        serviceId: args.serviceId,
        // Left undefined when not asked for, so the service applies its own
        // default rather than this one.
        ttlMinutes: args.ttlMinutes,
      },
    });
    spinner.stop();
    return data.openDatabaseAccessSession;
  } catch (error) {
    spinner.fail(`Could not open a session for service [${args.serviceId}].`);
    writeStatus(red(describeError(error)));
    process.exit(exitCode);
  }
};

/** Closes the session and reports the outcome. Resolves to the exit code. */
const closeSession = async (
  hostingServiceBaseUrl: string,
  getToken: () => Promise<string>,
  session: DatabaseAccessSession,
): Promise<number> => {
  try {
    // Fresh token: the startup one may have lapsed, and a 401 strands the role.
    const data = await callHosting<{
      closeDatabaseAccessSession: { status: string };
    }>(hostingServiceBaseUrl, await getToken(), CLOSE_SESSION, {
      input: { databaseAccessSessionId: session.databaseAccessSessionId },
    });

    if (data.closeDatabaseAccessSession.status === 'CLOSED') {
      writeStatus(green('Session closed and the database role dropped.'));
    } else {
      // CLOSING: the role outlives the command by at most one reconciliation
      // pass, so there is nothing useful to wait here for.
      writeStatus(
        yellow(
          'Local proxy closed. Dropping the database role is still pending and will be completed by the service.',
        ),
      );
    }
    return 0;
  } catch (error) {
    writeStatus(
      yellow(
        `The session could not be closed and will expire at ${
          session.expiresAt
        }: ${describeError(error)}`,
      ),
    );
    return exitCode;
  }
};

const printSession = (
  args: DatabaseProxyOptions,
  session: DatabaseAccessSession,
  port: number,
): void => {
  const clampWarning = formatClampedTtlWarning(
    args.ttlMinutes,
    session.expiresAt,
  );
  if (clampWarning) {
    writeStatus(yellow(clampWarning));
  }

  if (args.json === true) {
    writeOut(JSON.stringify(toConnectionJson(session, port)));
    return;
  }

  writeStatus(bold(`\nDatabase access to [${args.serviceId}] is open.\n`));
  writeOut(formatConnectionDetails(session, port));
  writeStatus(
    dim('\nThe password is shown once and cannot be retrieved again.'),
  );
  writeStatus(dim('Press Ctrl+C to close the session.'));
};

const warnBeforeExpiry = (session: DatabaseAccessSession): void => {
  const delay = toExpiryWarningDelay(session.expiresAt);
  if (delay === undefined) {
    return;
  }
  setTimeout(() => {
    writeStatus(
      yellow(
        `\nThis session ends at ${session.expiresAt}. Open connections will be dropped.`,
      ),
    );
  }, delay).unref();
};

export const proxyDatabase = async (
  args: Required<Pick<DatabaseProxyOptions, 'serviceId'>> &
    DatabaseProxyOptions,
): Promise<void> => {
  const port = args.localProxyPort ?? DEFAULT_LOCAL_PROXY_PORT;
  const hostingServiceBaseUrl = args.hostingServiceBaseURL as string;

  const getToken = createTokenProvider(
    args.idServiceAuthBaseURL as string,
    args.mosaicHostingClientId as string,
    args.mosaicHostingClientSecret as string,
  );

  const session = await openSession(args, hostingServiceBaseUrl, getToken);

  const server = net.createServer((client) =>
    databaseProxyConnection(client, session, getToken),
  );

  // Bound before printing: a bind error gives the session back.
  try {
    await listen(server, port);
  } catch (error) {
    writeStatus(red(`Could not listen on ${LOOPBACK}:${port}.`));
    writeStatus(red(describeError(error)));
    await closeSession(hostingServiceBaseUrl, getToken, session);
    process.exit(exitCode);
  }

  onShutdownRequested(() => {
    writeStatus(dim('\nClosing the session...'));
    server.close();
    void closeSession(hostingServiceBaseUrl, getToken, session).then((code) =>
      process.exit(code),
    );
  });

  printSession(args, session, port);
  warnBeforeExpiry(session);
};

/**
 * Validate arguments for proxying a direct database connection.
 */
export const validateDatabaseProxyArgs = (
  args: DatabaseProxyOptions,
): [DatabaseProxyOptions, string[]] => {
  const errorMessages: string[] = [];

  const idServiceAuthBaseURL =
    args.idServiceAuthBaseURL ?? process.env.ID_SERVICE_AUTH_BASE_URL ?? '';
  const mosaicHostingClientId =
    args.mosaicHostingClientId ?? process.env.MOSAIC_HOSTING_CLIENT_ID ?? '';
  const mosaicHostingClientSecret =
    args.mosaicHostingClientSecret ??
    process.env.MOSAIC_HOSTING_CLIENT_SECRET ??
    '';
  const hostingServiceBaseURL =
    args.hostingServiceBaseURL ?? process.env.HOSTING_SERVICE_BASE_URL ?? '';
  // Run from a solution, the service's own .env names it.
  const serviceId = args.serviceId ?? process.env.SERVICE_ID ?? '';

  if (isNullOrWhitespace(idServiceAuthBaseURL)) {
    errorMessages.push('[idServiceAuthBaseURL] is required.');
  }
  if (isNullOrWhitespace(mosaicHostingClientId)) {
    errorMessages.push('[mosaicHostingClientId] is required.');
  }
  if (isNullOrWhitespace(mosaicHostingClientSecret)) {
    errorMessages.push('[mosaicHostingClientSecret] is required.');
  }
  if (isNullOrWhitespace(hostingServiceBaseURL)) {
    errorMessages.push('[hostingServiceBaseURL] is required.');
  }
  if (isNullOrWhitespace(serviceId)) {
    errorMessages.push('[serviceId] is required.');
  }
  if (
    args.ttlMinutes !== undefined &&
    (!Number.isInteger(args.ttlMinutes) || args.ttlMinutes < 1)
  ) {
    errorMessages.push('[ttlMinutes] must be a whole number of 1 or more.');
  }
  if (
    args.localProxyPort !== undefined &&
    (!Number.isInteger(args.localProxyPort) ||
      args.localProxyPort < 1 ||
      args.localProxyPort > 65535)
  ) {
    errorMessages.push(
      '[localProxyPort] must be a whole number between 1 and 65535.',
    );
  }

  return [
    {
      ...args,
      serviceId,
      idServiceAuthBaseURL,
      mosaicHostingClientId,
      mosaicHostingClientSecret,
      hostingServiceBaseURL,
    },
    errorMessages,
  ];
};