// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

import { EventEmitter } from 'node:events';
import type {
  ClientConfig,
  ClientConfigOrString,
  ClientCredentials, CommandResponse,
  PasswordCredentials, RawClient, SendCommandOptions,
  TokenCredentials
} from '../client/client.type.js';
import { ResponseError, responseError } from '../wire/error.utils.js';
import { debug } from './client.debug.js';
import {
  endpointKey,
  type Endpoint,
  IggyConnection
} from './client.connection.js';
import { LOGIN, LOGIN_WITH_TOKEN, LOGOUT, PING } from '../wire/index.js';
import { GET_CLUSTER_METADATA } from '../wire/cluster/get-cluster-metadata.command.js';
import { COMMAND_CODE } from '../wire/command.code.js';
import {
  decodeVsrResponse,
  prepareVsrCommand,
  readRegisteredSession,
  VsrEvictionError,
  VsrSession,
} from '../wire/vsr/index.js';
import { normalizeClientConfig } from './client.config.js';

const VSR_RESPONSE_TIMEOUT_MS = 30_000;
const VSR_RETRY_INTERVAL_MS = 50;
const LEADERLESS_WAIT_BUDGET_MS = 5_000;
const LEADERLESS_POLL_INTERVAL_MS = 250;
const MAX_LEADER_REDIRECTS = 3;
const TRANSIENT_NOT_COMMITTED = 57;
const TRANSIENT_NOT_ACCEPTED = 58;
/**
 * How long a `TRANSIENT_NOT_ACCEPTED` request replays on the same connection
 * before the roster is re-read. A node that stopped being primary refuses
 * forever, so replaying alone never recovers. Matches the Rust SDK.
 */
const VSR_FAILOVER_CHECK_MS = 2_000;

/**
 * Whether a request's budget still holds enough for another attempt. One
 * exchange needs at least a replay interval to be worth starting; below that
 * the attempt can only end in a timeout, which would hide the refusal that
 * actually came back.
 */
const worthAnotherAttempt = (deadline: number): boolean =>
  deadline - Date.now() > VSR_RETRY_INTERVAL_MS;

/**
 * A request the current node keeps refusing as not-admitted. Carries the
 * refusal so the caller can surface it when the roster turns out to still name
 * this node as the leader. Never escapes `sendCommand`.
 */
class LeaderMovedError extends Error {
  constructor(readonly refusal: ResponseError) {
    super('the node refused the request as not-admitted; re-reading the roster');
  }
}

/**
 * How a leader move resolved: onto the metadata leader, onto the next roster
 * node (for a request the metadata leader itself keeps refusing), or not at
 * all.
 */
type RosterWalkVerdict = {
  endpoint: Endpoint,
  moved: boolean
};
type LeaderMoveVerdict = 'leader' | RosterWalkVerdict | false;

/**
 * Command codes that can be executed without authentication.
 */
const UNLOGGED_COMMAND_CODE = [
  PING.code,
  LOGIN.code,
  LOGIN_WITH_TOKEN.code
];

/**
 * Represents a queued command job waiting to be executed.
 */
type Job = {
  /** Command code */
  command: number,
  /** Command payload */
  payload: Buffer,
  /** Whether to parse the response */
  handleResponse: boolean,
  /** Whether the command is appended rather than prepended to the queue */
  last: boolean,
  /** Whether a not-admitted refusal re-checks the leader and re-issues */
  followsLeaderMoves: boolean,
  /** When the whole request gives up, however often it is re-issued */
  deadline: number,
  /** Promise resolve function */
  resolve: (v: CommandResponse | PromiseLike<CommandResponse>) => void,
  /** Promise reject function */
  reject: (e: unknown) => void
};

type ExchangeState = {
  written: boolean
};

export class VsrResponseTimeoutError extends Error {
  constructor(timeout: number) {
    super(`timed out after ${timeout} ms waiting for VSR response`);
    this.name = 'VsrResponseTimeoutError';
    Object.setPrototypeOf(this, VsrResponseTimeoutError.prototype);
  }
}

/**
 * Manages command execution and response handling for the Iggy server.
 * Implements command queuing, authentication, and heartbeat functionality.
 */
export class CommandResponseStream extends EventEmitter {
  /** Client configuration */
  private options: ClientConfig;
  /** Underlying connection to the server */
  private connection: IggyConnection;
  /** Queue of pending command jobs */
  private _execQueue: Job[];
  /** Consensus session used by VSR framing */
  private vsrSession: VsrSession;
  /** Shared authentication attempt for concurrent callers */
  private authenticationPromise?: Promise<boolean>;
  /** Whether a login is already being moved to the leader */
  private settlingLeader: boolean;
  /**
   * The leader re-check a refused request started, shared with every other
   * request refused by the same node so one demotion moves the client once.
   */
  private leaderMoveInFlight?: Promise<LeaderMoveVerdict>;
  /**
   * Set when a roster walk just redirected this client, so the login that
   * re-authenticates it stays on the dialed node instead of settling back on
   * the metadata leader whose partition replica refused the request.
   */
  private walkSettleSuppressed = false;
  /**
   * Refusals handed out to callers that have not decided what to do with them
   * yet. The queue holds while any are outstanding: the caller of a refused
   * command re-checks the leader, and a command written in the meantime goes
   * out on the socket that check is about to replace.
   */
  private leaderMovesUndecided: number;
  /** How long a leaderless roster is polled before settling in place */
  private leaderlessWaitBudget: number;
  /** Delay between roster reads while the cluster elects */
  private leaderlessPollInterval: number;
  /** Calls that have acquired this stream but have not fully settled */
  private pendingSubmissions: number;
  /** Whether the stream is currently processing a command */
  public busy: boolean;
  /** Whether the client has been authenticated */
  isAuthenticated: boolean;
  /** Authenticated user ID */
  userId?: number;
  /** Heartbeat interval timer handle */
  heartbeatIntervalHandler?: NodeJS.Timeout;
  /** Whether a heartbeat ping is still awaiting its response */
  private heartbeatInFlight: boolean;

  /**
   * Creates a new CommandResponseStream.
   *
   * @param options - Client configuration
   */
  constructor(options: ClientConfigOrString) {
    super();
    const normalizedConfig = normalizeClientConfig(options);
    this.options = normalizedConfig;
    this.connection = new IggyConnection(normalizedConfig);
    this.busy = false;
    this.isAuthenticated = false;
    this._execQueue = [];
    this.vsrSession = new VsrSession();
    this.authenticationPromise = undefined;
    this.settlingLeader = false;
    this.leaderMovesUndecided = 0;
    this.leaderlessWaitBudget = LEADERLESS_WAIT_BUDGET_MS;
    this.leaderlessPollInterval = LEADERLESS_POLL_INTERVAL_MS;
    this.pendingSubmissions = 0;
    this.heartbeatInFlight = false;
    this._init();
  };

  /**
   * Initializes the stream by setting up heartbeat and connection event handlers.
   */
  _init() {
    this.heartbeat(this.options.heartbeatInterval);
    this.connection.on('error', (error: Error) => {
      this._failQueue(error);
    });
    this.connection.on('eviction', (error: VsrEvictionError) => {
      this._resetSession();
      this.emit('eviction', error);
    });
    this.connection.on('disconnected', () => {
      this._resetSession();
      if (this.connection.redirecting) {
        // The client is moving to the leader, which is its own doing: a queued
        // command has not been written, so it belongs on the node being moved
        // to rather than in an error.
        this._reissueQueue();
        return;
      }
      this._failQueue(
        new Error('connection closed before queued commands were sent')
      );
    });
  }

  /**
   * Re-submits queued commands through the full send path, so each one
   * reconnects, re-authenticates and re-checks the leader as if it had just
   * been called.
   *
   * Only for a drop the client caused. Nothing here was written, so there is no
   * outcome in doubt: a command still in the queue when the socket is replaced
   * would otherwise fail with a lost-connection error the caller can do nothing
   * about.
   */
  private _reissueQueue(): void {
    const queued = this._execQueue;
    this._execQueue = [];
    for (const job of queued) {
      debug('re-issuing a queued command after a leader move', job.command);
      // The whole job, not just the payload: a fresh budget would let a command
      // caught in a move take twice the response timeout, and a roster read
      // re-issued as leader-following would answer a leader check with another
      // leader check.
      this.sendCommand(job.command, job.payload, {
        handleResponse: job.handleResponse,
        last: job.last,
        followsLeaderMoves: job.followsLeaderMoves,
        deadline: job.deadline
      }).then(job.resolve, job.reject);
    }
  }

  /**
   * Sends a command to the server.
   * Automatically handles connection, authentication and leader settlement.
   *
   * @param command - Command code to send
   * @param payload - Command payload buffer
   * @param options - Response and queue options
   * @returns Promise resolving to the command response
   */
  async sendCommand(
    command: number,
    payload: Buffer,
    options: SendCommandOptions = {}
  ): Promise<CommandResponse> {
    this.pendingSubmissions += 1;
    try {
      const {
        handleResponse = true,
        last = true,
        followsLeaderMoves = true
      } = options;

      if (!this.connection.connected)
        await this.connection.connect()

      if (!this.isAuthenticated && !this.isUnloggedCommand(command))
        await this.authenticate(this.options.credentials);

      // The roster read is itself a queued command and the queue is
      // single-flighted, so the leader re-check cannot happen inside
      // `_processVsr`. The refusal comes back out here instead, where the
      // queue is free, and the command is re-issued on the node that now
      // leads.
      //
      // A not-admitted refusal means the request was never applied, so it is
      // re-issued for the whole request budget rather than given up on after
      // one window: the roster can still name this node -- an election in
      // flight, a leader that has not moved yet -- and that is a wait, not a
      // verdict.
      //
      // One budget for the whole request: the transient replays on a
      // connection, the leader re-checks, and the re-issues after a move all
      // spend it, so a request cannot outlive it by moving. A command re-issued
      // after a move keeps the budget it was first submitted with, rather than
      // opening a second one.
      const deadline = options.deadline ?? Date.now() + VSR_RESPONSE_TIMEOUT_MS;
      let response: CommandResponse;
      let walkingRoster = false;
      const visitedRosterEndpoints = new Set<string>();
      for (;;) {
        try {
          response = await this._queueCommand(command, payload, handleResponse,
            last, followsLeaderMoves, deadline);
          break;
        } catch (error) {
          if (!(error instanceof LeaderMovedError))
            throw error;
          // The roster read that a re-check runs is itself a command that can
          // be refused this way, and answering a leader check with another
          // leader check would recurse. Its caller reads a failure as "stay
          // where you are".
          //
          // A budget too small to carry another attempt ends it here, with the
          // refusal the server actually gave: re-issued into what is left, the
          // request would time out instead and the caller would see a timeout
          // where the answer was "not admitted".
          let moved: LeaderMoveVerdict = false;
          try {
            if (!followsLeaderMoves || !worthAnotherAttempt(deadline))
              throw responseError(command, error.refusal.errorCode);
            // Once this request starts walking the roster it keeps walking: a
            // leader recheck between hops would put it straight back on the
            // metadata leader whose partition replica refused it, and the
            // walk would bounce between two nodes without reaching the rest.
            moved = await this._followLeaderMove(
              walkingRoster,
              visitedRosterEndpoints
            );
            if (typeof moved === 'object') {
              walkingRoster = true;
              visitedRosterEndpoints.add(endpointKey(moved.endpoint));
            }
          } finally {
            // Released as soon as the move is decided, before the pace below
            // and before any re-authentication: those go through the queue
            // themselves, and a queue still held for this refusal would never
            // reach them.
            if (followsLeaderMoves)
              this._releaseUndecidedMove();
          }
          const connectionMoved = moved === 'leader' ||
            (typeof moved === 'object' && moved.moved);
          if (!connectionMoved) {
            // Nowhere else to go yet: the roster still names this node, or it
            // could not be read. Paced, because the in-connection replay
            // window belongs to the request's budget and has already been
            // spent -- re-issuing straight away would spin.
            await delay(Math.min(
              VSR_FAILOVER_CHECK_MS,
              Math.max(0, deadline - Date.now())
            ));
          }
          if (!worthAnotherAttempt(deadline))
            throw responseError(command, error.refusal.errorCode);
          // A move drops the session with the socket it was bound to, so the
          // re-issue would otherwise go out under no session: a replicated
          // command fails client-side, a non-replicated one goes out with
          // session 0.
          if (!this.isAuthenticated && !this.isUnloggedCommand(command))
            await this.authenticate(this.options.credentials);
        }
      }
      if (!isLoginCommand(command) || this.settlingLeader)
        return response;
      // A login that re-authenticates a roster walk stays on the dialed node:
      // settling would put the client back on the metadata leader whose
      // partition replica refused the walked request. One login only.
      if (this.walkSettleSuppressed) {
        this.walkSettleSuppressed = false;
        return response;
      }
      this.settlingLeader = true;
      try {
        const settled = await this._settleOnLeader(command, payload);
        return settled ?? response;
      } finally {
        this.settlingLeader = false;
      }
    } finally {
      this.pendingSubmissions -= 1;
      this._emitFinishQueue();
    }
  }

  private _queueCommand(
    command: number,
    payload: Buffer,
    handleResponse: boolean,
    last: boolean,
    followsLeaderMoves: boolean,
    deadline: number
  ): Promise<CommandResponse> {
    return new Promise<CommandResponse>((resolve, reject) => {
      const job: Job = {
        command,
        payload,
        handleResponse,
        last,
        followsLeaderMoves,
        deadline,
        resolve,
        reject
      };
      if (last)
        this._execQueue.push(job);
      else
        this._execQueue.unshift(job);
      this._processQueue();
    });
  }

  /**
   * Re-reads the roster and moves to the leader it names.
   *
   * Best effort: an unreadable roster, or one that still names this node,
   * leaves the client where it is and the refused request is re-issued anyway.
   *
   * Single-flighted, and concurrent callers share the outcome instead of
   * failing: several commands are refused by the same demoted node, and each
   * starting its own redirect would move the client once per command. The
   * first redirect's `'disconnected'` also fails the others' roster reads, so
   * a caller that raced one would report a refusal it never had to.
   *
   * @returns Whether the client moved
   */
  private _followLeaderMove(
    walkPastLeader = false,
    visitedRosterEndpoints = new Set<string>()
  ): Promise<LeaderMoveVerdict> {
    const inFlight = this.leaderMoveInFlight;
    if (inFlight)
      return inFlight;

    const move = (async (): Promise<LeaderMoveVerdict> => {
      try {
        if (!walkPastLeader) {
          const leader = await this._readLeaderEndpoint();
          if (leader && !this.connection.isConnectedTo(leader.host, leader.port)) {
            debug(`the leader moved to ${leader.host}:${leader.port}, following it`);
            await this.connection.redirect(leader.host, leader.port);
            return 'leader';
          }
        }
        // The roster names this node as the metadata leader (or said nothing
        // usable), yet it keeps refusing to admit the request: its replica of
        // the target partition group is not that group's primary, because
        // metadata and partition consensus groups elect independently. Walk
        // the roster instead of re-issuing into the same refusal.
        const next = this.connection.nextRosterEndpoint(visitedRosterEndpoints);
        if (!next)
          return false;
        debug(
          'the request keeps being refused here, walking the roster to ' +
          `${next.host}:${next.port}`
        );
        // The re-authentication after this redirect runs a login, and a login
        // normally settles on the metadata leader, which would put the walk
        // right back on the node that refused. One login only.
        this.walkSettleSuppressed = true;
        try {
          await this.connection.redirect(next.host, next.port);
        } catch (error) {
          // The suppression belongs to the redirect above. If that redirect
          // never lands, a later unrelated login must settle normally.
          this.walkSettleSuppressed = false;
          debug('the roster endpoint could not be reached', error);
          return { endpoint: next, moved: false };
        }
        return { endpoint: next, moved: true };
      } catch (error) {
        debug('the leader could not be re-checked, staying on this node', error);
        return false;
      }
    })();
    this.leaderMoveInFlight = move;
    void move.finally(() => {
      if (this.leaderMoveInFlight !== move)
        return;
      this.leaderMoveInFlight = undefined;
      // The drain stopped while the move was being decided. A move that
      // happened re-issues what was held back on the new socket; one that did
      // not leaves it here, with nothing else due to pick it up.
      if (!this.connection.redirecting)
        void this._processQueue();
    });
    return move;
  }

  /** Whether a leader move is being decided or carried out. */
  private _movePending(): boolean {
    return this.leaderMovesUndecided > 0 || this.leaderMoveInFlight !== undefined;
  }

  /**
   * Releases the queue hold one refusal took, and drains what was held back
   * once the last of them is decided.
   */
  private _releaseUndecidedMove(): void {
    if (this.leaderMovesUndecided > 0)
      this.leaderMovesUndecided -= 1;
    if (this._movePending() || this.connection.redirecting)
      return;
    void this._processQueue();
  }

  private _rememberRoster(response: CommandResponse): void {
    try {
      const metadata = GET_CLUSTER_METADATA.deserialize(response);
      this.connection.rememberRoster(
        metadata.nodes
          .filter((node) => node.endpoints.tcp !== 0)
          .map((node) => ({ host: node.ip, port: node.endpoints.tcp }))
      );
    } catch (error) {
      debug('an unreadable roster leaves the redial candidates as they are',
        error);
    }
  }

  /**
   * Processes queued commands sequentially.
   * Emits 'finishQueue' when all commands are processed.
   *
   */
  async _processQueue(): Promise<void> {
    if (this.busy)
      return;
    this.busy = true;
    while (this._execQueue.length > 0 && this.connection.socket.writable) {
      // While a leader move is being decided, only the roster read the move
      // itself runs goes out -- it is what decides where the client lands, and
      // it is the one command that does not follow moves. Draining the rest
      // would write them to the socket `redirect()` is about to replace, and a
      // command in flight when that happens dies with a lost-connection error
      // instead of being re-issued on the node the move lands on.
      const index = this._movePending()
        ? this._execQueue.findIndex((job) => !job.followsLeaderMoves)
        : 0;
      if (index < 0) break;
      const [next] = this._execQueue.splice(index, 1);
      if (!next) break;
      const { command, payload, handleResponse, deadline, resolve, reject } = next;
      try {
        resolve(await this._processNext(command, payload, handleResponse, deadline));
      } catch (err) {
        if (err instanceof LeaderMovedError && next.followsLeaderMoves)
          // Counted before the rejection is handed out, not after: the caller
          // resumes as a microtask, so this loop would otherwise write the next
          // command before the re-check it is about to start has begun.
          this.leaderMovesUndecided += 1;
        reject(err);
      }
    }
    if (this._execQueue.length > 0) {
      // The same distinction as on 'disconnected': the socket a leader move
      // replaced stops being writable, and what is still queued belongs on the
      // node being moved to.
      if (this.connection.redirecting)
        this._reissueQueue();
      else if (!this._movePending())
        this._failQueue(new Error('connection is not writable'));
      // Otherwise the move is still being decided: these commands were never
      // written, and they are drained again once it settles -- here if the
      // client stays, on the new socket if it moves.
    }
    this.busy = false;
    this._emitFinishQueue();
  }

  private _emitFinishQueue(): void {
    if (this.pendingSubmissions === 0 &&
        !this.busy &&
        this._execQueue.length === 0)
      this.emit('finishQueue');
  }

  /**
   * Processes a single command by writing it to the connection and waiting for response.
   *
   * @param command - Command code
   * @param payload - Command payload
   * @param handleResp - Whether to parse the response
   * @param deadline - When the whole request gives up, shared with the leader
   * re-checks and the re-issues after a move
   * @returns Promise resolving to the command response
   */
  _processNext(
    command: number,
    payload: Buffer,
    handleResp = true,
    deadline = Date.now() + VSR_RESPONSE_TIMEOUT_MS
  ): Promise<CommandResponse> {
    if (isLoginCommand(command) && this.isAuthenticated)
      return this._processVsrLogin(command, payload, handleResp, deadline);
    return this._processVsr(command, payload, handleResp, deadline);
  }

  private async _processVsrLogin(
    command: number,
    payload: Buffer,
    handleResp: boolean,
    deadline: number
  ): Promise<CommandResponse> {
    await this._processVsr(LOGOUT.code, LOGOUT.serialize(), true, deadline);
    return this._processVsr(command, payload, handleResp, deadline);
  }

  private async _processVsr(
    command: number,
    payload: Buffer,
    handleResp: boolean,
    deadline: number
  ): Promise<CommandResponse> {
    let requestWritten = false;
    try {
      const prepared = prepareVsrCommand(command, payload);
      // A transient retry must preserve all request identity fields.
      const frame = this.vsrSession.encode(prepared.command, prepared.payload);
      // Derived from the request's own budget rather than read off the clock,
      // so one request spends one budget however many times it is re-issued.
      const notAcceptedDeadline =
        deadline - VSR_RESPONSE_TIMEOUT_MS + VSR_FAILOVER_CHECK_MS;
      let lastTransientError: ResponseError | undefined;
      let parsed: CommandResponse;
      while (true) {
        const remaining = deadline - Date.now();
        if (remaining <= 0) {
          if (lastTransientError)
            throw lastTransientError;
          throw new VsrResponseTimeoutError(VSR_RESPONSE_TIMEOUT_MS);
        }
        const exchangeState = { written: false };
        let response: Buffer;
        try {
          response = await this._exchange(
            () => this.connection.writeFrame(frame),
            remaining,
            exchangeState
          );
        } finally {
          requestWritten ||= exchangeState.written;
        }
        if (!handleResp)
          return response as unknown as CommandResponse;
        try {
          parsed = decodeVsrResponse(response, command);
          break;
        } catch (error) {
          if (!(error instanceof ResponseError) ||
              !isTransientVsrError(error.errorCode))
            throw error;
          lastTransientError = error;
          // A not-admitted refusal is a statement about who leads, not about
          // load: a node that stopped being primary refuses forever, so
          // replaying on this connection never recovers. Hand it back for a
          // roster re-read once the window is spent. Not-committed (57) stays
          // here: the request is in flight on this very node, and its outcome
          // is unknown anywhere else.
          if (error.errorCode === TRANSIENT_NOT_ACCEPTED &&
              !isLoginCommand(command) &&
              Date.now() >= notAcceptedDeadline)
            throw new LeaderMovedError(error);
          const retryDelay = Math.min(
            VSR_RETRY_INTERVAL_MS,
            Math.max(0, deadline - Date.now())
          );
          if (retryDelay === 0)
            throw error;
          await delay(retryDelay);
        }
      }

      if (prepared.command === COMMAND_CODE.LoginRegister ||
          prepared.command === COMMAND_CODE.LoginRegisterWithAccessToken) {
        this.vsrSession.bind(readRegisteredSession(parsed));
        this.isAuthenticated = true;
        this.userId = parsed.data.readUInt32LE(0);
      }
      if (prepared.command === COMMAND_CODE.LogoutUser) {
        this._resetSession();
      }
      // Every roster read feeds the redial candidates, whoever asked for it
      // and whatever it says: a node dies together with its address, the
      // roster is unreachable exactly when it is needed, and reading it only
      // during a login would leave the candidates stale between logins.
      if (handleResp && command === GET_CLUSTER_METADATA.code)
        this._rememberRoster(parsed);
      return parsed;
    } catch (error) {
      // A not-admitted refusal is an answer, so the session is not in doubt
      // and the request was never applied.
      if (error instanceof LeaderMovedError)
        throw error;
      // Once bytes were handed to the socket, a local transport or decode
      // failure leaves the request outcome ambiguous. Register a fresh session
      // rather than replaying that request under a different client identity.
      if (!(error instanceof ResponseError) && requestWritten)
        this._resetSession();
      if (error instanceof VsrEvictionError)
        throw error;
      if (error instanceof ResponseError)
        throw responseError(command, error.errorCode);
      throw error;
    }
  }

  private _exchange(
    write: () => void,
    timeout?: number,
    state?: ExchangeState
  ): Promise<Buffer> {
    return new Promise((resolve, reject) => {
      let timeoutHandler: NodeJS.Timeout | undefined;
      const cleanup = () => {
        if (timeoutHandler)
          clearTimeout(timeoutHandler);
        this.connection.removeListener('error', errorCallback);
        this.connection.removeListener('disconnected', disconnectedCallback);
        this.connection.removeListener('eviction', evictionCallback);
        this.connection.removeListener('response', responseCallback);
      };
      const errorCallback = (error: unknown) => {
        cleanup();
        reject(error);
      };
      const disconnectedCallback = () => {
        cleanup();
        reject(new Error('connection closed while waiting for response'));
      };
      const evictionCallback = (error: VsrEvictionError) => {
        cleanup();
        reject(error);
      };
      const responseCallback = (response: Buffer) => {
        cleanup();
        resolve(response);
      };
      if (timeout !== undefined) {
        timeoutHandler = setTimeout(() => {
          cleanup();
          this.connection.abort();
          reject(new VsrResponseTimeoutError(timeout));
        }, timeout);
      }
      this.connection.once('error', errorCallback);
      this.connection.once('disconnected', disconnectedCallback);
      this.connection.once('eviction', evictionCallback);
      this.connection.once('response', responseCallback);
      try {
        write();
        if (state)
          state.written = true;
      } catch (error) {
        cleanup();
        reject(error);
      }
    });
  }

  // `GetClusterMetadata` is deliberately absent: the server auth-gates it,
  // so the client authenticates before reading the topology. A login dialed
  // at a backup still succeeds because the server forwards the register to
  // the primary.
  private isUnloggedCommand(command: number): boolean {
    return UNLOGGED_COMMAND_CODE.includes(command);
  }

  /**
   * Moves a freshly authenticated session to the cluster leader.
   *
   * Only the leader accepts replicated commands, and the roster read is
   * auth-gated, so the topology cannot be inspected before a login binds a
   * session. The redirect drops that session along with the socket, so the
   * login is replayed on the leader and its answer supersedes the one from the
   * node the client dialed. Leadership can move between the roster read and
   * the replay, so each freshly bound hop rechecks the roster under a bounded
   * redirect budget.
   *
   * @returns The leader's login response, or undefined when the client stays
   */
  private async _settleOnLeader(
    loginCommand: number,
    loginPayload: Buffer
  ): Promise<CommandResponse | undefined> {
    let settledResponse: CommandResponse | undefined;
    for (let redirects = 0; redirects < MAX_LEADER_REDIRECTS; redirects += 1) {
      const leader = await this._readLeaderEndpoint();
      if (!leader || this.connection.isConnectedTo(leader.host, leader.port))
        return settledResponse;
      await this.connection.redirect(leader.host, leader.port);
      settledResponse = await this.sendCommand(
        loginCommand,
        loginPayload,
        { last: false }
      );
    }
    debug(
      `leader settlement reached its ${MAX_LEADER_REDIRECTS}-hop budget, ` +
      'staying on the current node'
    );
    return settledResponse;
  }

  /**
   * Reads the cluster roster and picks the endpoint to settle on.
   *
   * Best effort: an unreadable roster, `Unauthenticated` included (the session
   * died between the login and this read), keeps the client on its current
   * node instead of failing a login that already succeeded.
   */
  private async _readLeaderEndpoint(): Promise<Endpoint | undefined> {
    // A cluster can be transiently leaderless: a restarted node cedes the
    // primaryship its stale view assigns it, and the roster reports no leader
    // until the peers' election completes. That window is roughly one heartbeat
    // timeout, so poll through it rather than settling on a replica that denies
    // every replicated command for its whole retry budget.
    const deadline = Date.now() + this.leaderlessWaitBudget;
    while (true) {
      // Reading without a session would re-enter authentication, which awaits
      // the very login this settlement runs inside of. The session can also die
      // between polls, so this holds for every pass, not just the first.
      if (!this.isAuthenticated)
        return undefined;
      try {
        // Queue the metadata fetch instead of writing directly: a bare write
        // would race an in-flight exchange and both would wake on the same
        // response event.
        const response = await this.sendCommand(
          GET_CLUSTER_METADATA.code,
          GET_CLUSTER_METADATA.serialize(),
          { last: false, followsLeaderMoves: false }
        );
        // The redial candidates are fed by `_processVsr` for every roster
        // read, leaderless ones included: a roster with no leader still names
        // where the nodes are.
        const metadata = GET_CLUSTER_METADATA.deserialize(response);
        if (metadata.nodes.length <= 1)
          return undefined;
        const leader = metadata.nodes.find(
          (node) => node.role === 'Leader' && node.status === 'Healthy'
        );
        if (leader)
          return { host: leader.ip, port: leader.endpoints.tcp };
      } catch (error) {
        debug('cluster metadata is unreadable, staying on this node', error);
        return undefined;
      }
      if (Date.now() >= deadline) {
        debug(
          'cluster metadata named no healthy leader within ' +
          `${this.leaderlessWaitBudget} ms, staying on this node`
        );
        return undefined;
      }
      await delay(this.leaderlessPollInterval);
    }
  }

  /**
   * Fails all queued commands with the given error.
   *
   * @param err - Error to reject all queued commands with
   */
  _failQueue(err: Error) {
    this._execQueue.forEach(({ reject }) => reject(err));
    this._execQueue = [];
  }

  private _resetSession(): void {
    if (!this.isAuthenticated &&
        this.userId === undefined &&
        !this.vsrSession.hasActivity)
      return;
    this.isAuthenticated = false;
    this.userId = undefined;
    this.vsrSession.reset();
    this.emit('sessionReset');
  }

  hold(): () => void {
    this.pendingSubmissions += 1;
    let released = false;
    return () => {
      if (released)
        return;
      released = true;
      this.pendingSubmissions -= 1;
      this._emitFinishQueue();
    };
  }

  /**
   * Authenticates the client with the server.
   *
   * @param creds - Authentication credentials (token or password)
   * @returns True if authentication succeeded
   */
  async authenticate(creds: ClientCredentials): Promise<boolean> {
    if (this.isAuthenticated)
      return true;
    if (this.authenticationPromise)
      return this.authenticationPromise;

    this.authenticationPromise = this._authenticate(creds);
    try {
      return await this.authenticationPromise;
    } finally {
      this.authenticationPromise = undefined;
    }
  }

  private async _authenticate(creds: ClientCredentials): Promise<boolean> {
    const r = ('token' in creds) ?
      await this._authWithToken(creds) :
      await this._authWithPassword(creds);
    this.isAuthenticated = true;
    this.userId = r.userId;
    return this.isAuthenticated;
  }

  /**
   * Authenticates using username and password.
   *
   * @param creds - Password credentials
   * @returns Login response with user ID
   */
  async _authWithPassword(creds: PasswordCredentials) {
    const pl = LOGIN.serialize(creds);
    const logr = await this.sendCommand(LOGIN.code, pl, { last: false });
    return LOGIN.deserialize(logr);
  }

  /**
   * Authenticates using a token.
   *
   * @param creds - Token credentials
   * @returns Login response with user ID
   */
  async _authWithToken(creds: TokenCredentials) {
    const pl = LOGIN_WITH_TOKEN.serialize(creds);
    const logr = await this.sendCommand(
      LOGIN_WITH_TOKEN.code,
      pl,
      { last: false }
    );
    return LOGIN_WITH_TOKEN.deserialize(logr);
  }

  /**
   * Sends a ping command to the server.
   *
   * @returns Ping response
   */
  async ping() {
    const pl = PING.serialize();
    const pingR = await this.sendCommand(PING.code, pl);
    return PING.deserialize(pingR);
  }

  /**
   * Starts sending periodic heartbeat pings to keep the connection alive.
   *
   * @param interval - Heartbeat interval in milliseconds
   */
  heartbeat(interval?: number) {
    if (!interval)
      return

    this.heartbeatIntervalHandler = setInterval(async () => {
      if (this.connection.connected && !this.heartbeatInFlight) {
        this.heartbeatInFlight = true;
        debug(`sending heartbeat ping (interval: ${interval} ms)`);
        try {
          await this.ping()
          this.emit('heartbeat');
        } catch (error) {
          debug('heartbeat ping failed', error);
        } finally {
          this.heartbeatInFlight = false;
        }
      }
    }, interval);
    // A pending heartbeat must not be the reason the process stays up: a script
    // that never calls destroy() would otherwise hang on exit.
    this.heartbeatIntervalHandler.unref();
  }

  /**
   * Returns the underlying socket as a readable stream.
   *
   * @returns The connection socket
   */
  getReadStream() {
    return this.connection.socket;
  }

  /**
   * Destroys the stream and cleans up resources.
   * Stops heartbeat and destroys the connection.
   */
  destroy() {
    if (this.heartbeatIntervalHandler)
      clearInterval(this.heartbeatIntervalHandler);
    return this.connection._destroy();
  }
};


/**
 * Creates a new RawClient instance.
 *
 * @param options - Client configuration
 * @returns RawClient instance
 */
export function getRawClient(options: ClientConfigOrString): RawClient {
  return new CommandResponseStream(options);
}

const isLoginCommand = (command: number): boolean =>
  command === COMMAND_CODE.LoginUser ||
  command === COMMAND_CODE.LoginWithAccessToken;

const delay = (milliseconds: number): Promise<void> =>
  new Promise((resolve) => setTimeout(resolve, milliseconds));

const isTransientVsrError = (errorCode: number): boolean =>
  errorCode === TRANSIENT_NOT_COMMITTED ||
  errorCode === TRANSIENT_NOT_ACCEPTED;