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 { exitCode } from '../../../../common/index.js';
import type { ServiceLogsOptions } from './service-logs-options.js';

const { red, dim } = chalk;

/**
 * The GraphQL shape returned by Hosting, which is deliberately declared here
 * rather than imported: this package is published, and the SDU packages that
 * own the equivalent REST DTO are private.
 */
export interface ServiceReplicaLog {
  replicaName: string;
  lines: string[];
}

/**
 * Log lines go to stdout and nothing else does; status messages and errors go
 * to stderr. This keeps `mosaic hosting service logs -i svc > out.log` a file
 * of pure log output, and keeps `| grep` results free of decoration.
 */
const writeLogLine = (line: string): void => {
  process.stdout.write(line + '\n');
};
const writeStatus = (message: string): void => {
  process.stderr.write(message + '\n');
};

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

/**
 * Replica names are full K8s pod names, e.g.
 * `d-8ip51mklekzyxmtk-5ihjg3ycvjkpnmb2-ax-media-service-bb9c8cfmfq`. Everything
 * up to and including the service ID is identical across replicas of the
 * service that was asked for, so only the trailing segment carries
 * information -- and 60+ characters in front of every line is unreadable.
 * `--json` keeps the full name.
 */
export const toShortReplicaName = (
  replicaName: string,
  serviceId: string,
): string => {
  const marker = `-${serviceId}-`;
  const index = replicaName.lastIndexOf(marker);
  if (index === -1) {
    return replicaName;
  }
  const suffix = replicaName.slice(index + marker.length);
  return suffix.length > 0 ? suffix : replicaName;
};

/**
 * Drops the empty element left by the log's trailing newline, while preserving
 * blank lines inside the log: a service not using
 * `@axinom/mosaic-service-common` may emit plain multi-line text where blank
 * lines are meaningful.
 */
const stripTrailingBlank = (lines: string[]): string[] => {
  const result = [...lines];
  while (result.length > 0 && result[result.length - 1] === '') {
    result.pop();
  }
  return result;
};

const SINCE_UNITS_IN_SECONDS: Record<string, number> = {
  s: 1,
  m: 60,
  h: 60 * 60,
  d: 24 * 60 * 60,
};

const SINCE_PATTERN = /^(\d+)(s|m|h|d)?$/;

/**
 * Reads an age such as `30s`, `5m`, `2h` or `1d` into seconds, matching how
 * `kubectl logs --since` and `docker logs --since` are written. A bare number
 * is read as seconds. Returns undefined when the value cannot be read.
 */
export const parseSinceToSeconds = (
  since: string | undefined,
): number | undefined => {
  if (since === undefined) {
    return undefined;
  }
  const match = SINCE_PATTERN.exec(since.trim());
  if (!match) {
    return undefined;
  }
  const amount = Number(match[1]);
  if (amount <= 0) {
    return undefined;
  }
  return amount * SINCE_UNITS_IN_SECONDS[match[2] ?? 's'];
};

interface OrderedLine {
  /** RFC3339 timestamp Kubernetes recorded for the line, used for ordering. */
  sortKey: string;
  replicaName: string;
  timestamp: string;
  message: string;
}

const TIMESTAMP_PATTERN = /^(\d{4}-\d{2}-\d{2}T\S+Z) (.*)$/s;

const NANO_DIGITS = 9;
const RFC3339_PATTERN = /^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d+))?Z$/;

/**
 * Pads the fractional seconds to a fixed width so timestamps can be ordered by
 * plain string comparison.
 *
 * RFC3339Nano permits a variable-length fraction, and comparing those directly
 * misorders them: `.12Z` is earlier than `.1201Z`, but sorts after it because
 * `Z` compares above `0`. Padding avoids that while keeping the precision a
 * `Date` would round away. A value that does not parse is returned unchanged
 * and simply orders lexicographically.
 */
export const toTimestampSortKey = (timestamp: string): string => {
  const match = RFC3339_PATTERN.exec(timestamp);
  if (!match) {
    return timestamp;
  }
  return `${match[1]}.${(match[2] ?? '').padEnd(NANO_DIGITS, '0')}Z`;
};

/**
 * Splits the timestamp Kubernetes prefixed onto each line from the line itself.
 *
 * A line that does not carry one (which should not happen, but a malformed or
 * truncated log could produce it) inherits the previous line's key so it stays
 * next to the line it belongs with, instead of sorting to the top.
 */
const toOrderedLines = (
  replicaName: string,
  lines: string[],
  previousKey = '',
): OrderedLine[] => {
  let lastKey = previousKey;
  return lines.map((line) => {
    const match = TIMESTAMP_PATTERN.exec(line);
    if (match) {
      lastKey = toTimestampSortKey(match[1]);
      return {
        sortKey: lastKey,
        replicaName,
        timestamp: match[1],
        message: match[2],
      };
    }
    return { sortKey: lastKey, replicaName, timestamp: '', message: line };
  });
};

/**
 * Merges every replica's lines into one chronological stream.
 *
 * The keys are RFC3339 UTC strings of identical shape, so comparing them as
 * strings orders them correctly and preserves the sub-millisecond precision
 * that parsing into a `Date` would discard. Sorting is stable, so lines sharing
 * a timestamp keep their original per-replica order.
 */
export const mergeReplicaLines = (
  replicas: ServiceReplicaLog[],
): OrderedLine[] =>
  replicas
    .flatMap((replica) =>
      toOrderedLines(replica.replicaName, stripTrailingBlank(replica.lines)),
    )
    .sort((a, b) =>
      a.sortKey < b.sortKey ? -1 : a.sortKey > b.sortKey ? 1 : 0,
    );

export const getServiceLogs = async (
  args: Required<Pick<ServiceLogsOptions, 'serviceId'>> & ServiceLogsOptions,
): Promise<void> => {
  const serviceAccountToken = await getServiceAccountToken(
    args.idServiceAuthBaseURL as string,
    args.mosaicHostingClientId as string,
    args.mosaicHostingClientSecret as string,
  );

  const axiosInstance: AxiosInstance = getAxiosInstance(
    args.hostingServiceBaseURL as string,
    serviceAccountToken.accessToken,
  );

  const query = `query GetServiceRuntimeLogs($serviceId: String!, $tailLines: Int, $sinceSeconds: Int) {
    serviceRuntimeLogs(input: { serviceId: $serviceId, tailLines: $tailLines, sinceSeconds: $sinceSeconds }) {
      replicas {
        replicaName
        lines
      }
    }
  }`;

  const payload = JSON.stringify({
    query,
    variables: {
      serviceId: args.serviceId,
      tailLines: args.tail,
      sinceSeconds: parseSinceToSeconds(args.since),
    },
  });

  try {
    const response = await axiosInstance.post('graphql', payload);

    if (response.data.errors !== undefined && response.data.errors.length > 0) {
      writeStatus(
        red(`Failed to retrieve logs for service [${args.serviceId}].`),
      );
      writeStatus(red(response.data.errors[0].message));
      process.exit(exitCode);
    }

    const replicas: ServiceReplicaLog[] =
      response.data.data?.serviceRuntimeLogs?.replicas ?? [];

    const merged = mergeReplicaLines(replicas);

    // Newline-delimited JSON: the same ordering as the human output, one object
    // per line so it streams into `jq` and friends. The replica and timestamp
    // are always present as fields, which is why --prefix and --timestamps do
    // not apply here.
    if (args.json) {
      for (const line of merged) {
        writeLogLine(
          JSON.stringify({
            replicaName: line.replicaName,
            timestamp: line.timestamp,
            message: line.message,
          }),
        );
      }
      return;
    }

    if (replicas.length === 0) {
      writeStatus(
        dim(`No running replicas found for service [${args.serviceId}].`),
      );
      return;
    }

    // With one replica the prefix is noise on every line; with several it is
    // the only way to tell which replica said what.
    const usePrefix = args.prefix ?? replicas.length > 1;

    writeStatus(
      dim(
        `# ${args.serviceId}: ${replicas.length} replica(s)${
          args.tail ? `, last ${args.tail} line(s) each` : ''
        }`,
      ),
    );

    for (const line of merged) {
      // The service's own text is emitted verbatim: it may be plain text
      // rather than JSON, and any reformatting would corrupt it.
      const prefix = usePrefix
        ? `[${toShortReplicaName(line.replicaName, args.serviceId)}] `
        : '';
      const stamp =
        args.timestamps && line.timestamp ? `${line.timestamp} ` : '';
      writeLogLine(`${prefix}${stamp}${line.message}`);
    }
  } catch (error) {
    const axiosError = error as AxiosError;
    writeStatus(
      red(`Error while retrieving logs for service [${args.serviceId}].`),
    );
    writeStatus(red(`Error: ${axiosError.message}`));

    if (axiosError.response) {
      writeStatus(red(`Status: ${axiosError.response.status}`));
      writeStatus(
        red(`Response: ${JSON.stringify(axiosError.response.data, null, 2)}`),
      );
    }
    process.exit(exitCode);
  }
};

/**
 * Validate arguments for reading service logs.
 */
export const validateServiceLogsArgs = (
  args: ServiceLogsOptions,
): [ServiceLogsOptions, 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 ?? '';

  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(args.serviceId)) {
    errorMessages.push('[serviceId] is required.');
  }
  if (args.tail !== undefined && (args.tail < 1 || args.tail > 1000)) {
    errorMessages.push('[tail] must be between 1 and 1000.');
  }
  if (
    args.since !== undefined &&
    parseSinceToSeconds(args.since) === undefined
  ) {
    errorMessages.push(
      '[since] must be a positive age such as 30s, 5m, 2h or 1d.',
    );
  }

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