/********************************************************************************
 * Copyright (c) 2026 CrossBreeze, EclipseSource and others.
 *
 * This program and the accompanying materials are made available under the
 * terms of the MIT License which is available in the project root.
 *
 * SPDX-License-Identifier: MIT
 ********************************************************************************/

import { SystemClock, describeError, describeErrorWithStack } from '@hydranium/protocol';
import {
   type Channel,
   CommandService,
   type ConnectionHandler,
   type Disposable,
   ILogger,
   type MessageProvider,
   MessageService
} from '@theia/core';
import { AbstractChannel } from '@theia/core/lib/common/message-rpc/channel';
import { Deferred } from '@theia/core/lib/common/promise-util';
import { inject, injectable, optional } from '@theia/core/shared/inversify';
import * as net from 'node:net';
import { Clock } from '../common/clock';
import { SocketChannelForwarder } from './socket-channel-forwarder';

/** Resolved configuration for a {@link AbstractSocketForwardingConnectionHandler}.
 *  A head-specific subclass derives these from its own (adopter-facing) options
 *  and passes them to `super(...)`. */
export interface SocketForwardingConnectionHandlerOptions {
   /** Theia service path the browser frontend opens a channel to. */
   readonly path: string;
   /** Command id whose return value is the target server's listening port (the
    *  server publishes it on the LSP connection at startup). */
   readonly portCommand: string;
   /** Short bracket prefix for this head's diagnostic logs, rendered as
    *  `[<logComponent>] …`. */
   readonly logComponent: string;
   /** Human-readable name of the server this head connects to — used in log and
    *  error messages. */
   readonly serverName: string;
   readonly findPortTimeout?: number;
   readonly findPortAttempts?: number;
   readonly connectTimeoutMs?: number;
   /**
    * Optional diagnostic hook called immediately after the outbound
    * `net.Socket` to the server is created, BEFORE `socket.connect()` is
    * invoked. Adopters attach `'data'` / `'close'` listeners here for
    * byte-level observability when debugging wire-level issues; attaching
    * before `connect()` guarantees the very first bytes are observed.
    */
   readonly onSocketCreated?: (socket: net.Socket) => void;
}

/**
 * Cross-head base for the Theia backend half of a protocol head's transport:
 * it bridges a Theia browser-frontend {@link Channel} to a server's TCP socket
 * by relaying bytes between the two.
 *
 * Theia delivers each frontend connection to a registered `ConnectionHandler`
 * keyed by its `path`. The handler asks the language server (via a registered
 * command) for the target server's listening port, opens a `net.Socket` to that
 * port, and forwards the Theia channel onto the socket. The frontend terminates
 * the same wire protocol the server speaks, so the backend performs no semantic
 * re-proxy — it relays bytes only and stays oblivious to the server's method set.
 *
 * The byte forwarder that bridges the channel and the socket is
 * {@link forwardToSocketConnection}; both heads use {@link SocketChannelForwarder},
 * and the port discovery, buffer-and-replay race fix and connect orchestration
 * here are shared.
 */
@injectable()
export abstract class AbstractSocketForwardingConnectionHandler implements ConnectionHandler {
   @inject(MessageService) protected messageService!: MessageService;
   @inject(CommandService) protected commandService!: CommandService;
   // Theia backend `ILogger` — this handler runs in the Theia backend process,
   // not the spawned language-server process, so the framework `LspLogger`
   // (which needs the Langium connection/services) is out of reach here.
   @inject(ILogger) protected readonly logger!: ILogger;
   /** The clock each {@link SocketChannelForwarder} bounds its flush on: the container's {@link Clock}, or a `SystemClock`. */
   @inject(Clock) @optional() protected readonly clock: Clock = new SystemClock();

   readonly path: string;

   protected readonly portCommand: string;
   protected readonly logComponent: string;
   protected readonly serverName: string;
   protected readonly findPortTimeout: number;
   protected readonly findPortAttempts: number;
   protected readonly connectTimeoutMs: number;
   protected readonly onSocketCreated?: (socket: net.Socket) => void;

   constructor(options: SocketForwardingConnectionHandlerOptions) {
      this.path = options.path;
      this.portCommand = options.portCommand;
      this.logComponent = options.logComponent;
      this.serverName = options.serverName;
      this.findPortTimeout = options.findPortTimeout ?? 500;
      this.findPortAttempts = options.findPortAttempts ?? -1;
      this.connectTimeoutMs = options.connectTimeoutMs ?? 10000;
      this.onSocketCreated = options.onSocketCreated;
   }

   onConnection(connection: Channel): void {
      this.initializeServerConnection(connection);
   }

   protected async initializeServerConnection(channel: Channel): Promise<void> {
      // RACE FIX: subscribe to `channel.onMessage` synchronously and buffer
      // every `MessageProvider` until the forwarder is wired. Theia's
      // `ForwardingChannel.onMessage` is a plain `Emitter` — it does NOT replay
      // to listeners that subscribe later. Without this buffer, frontend writes
      // arriving between `onConnection` and the forwarder subscribing (worst
      // case: the whole `findPort` delay plus the socket connect time) are
      // dropped silently. A frontend builds its proxy synchronously on the
      // channel and may send requests immediately, so it reliably hits this
      // window.
      const buffered: MessageProvider[] = [];
      const bufferSub = channel.onMessage(provider => buffered.push(provider));
      const closed = new AbortController();
      const closeSub = channel.onClose(() => closed.abort());
      try {
         const port = await this.findPort(closed.signal);
         closed.signal.throwIfAborted();
         this.logger.info(`[${this.logComponent}] Connecting to ${this.serverName} on port ${port}...`);
         await this.connectToServer(channel, port, { bufferSub, buffered });
         this.logger.info(`[${this.logComponent}] Connected to ${this.serverName} on port ${port}.`);
      } catch (error) {
         bufferSub.dispose();
         if (closed.signal.aborted) {
            this.logger.debug(`[${this.logComponent}] Stopped connecting to ${this.serverName}: the frontend closed the channel.`);
            return;
         }
         this.logger.error(`[${this.logComponent}] Could not connect to ${this.serverName}: ${describeErrorWithStack(error)}`);
         this.reportConnectFailure(describeError(error));
         // Closed so the frontend learns of it: left open, the channel has no
         // server behind it, and every request on it waits for good.
         channel.close();
      } finally {
         closeSub.dispose();
      }
   }

   /** Tell the user this handler gave up; `message` is the reason it logged. */
   protected reportConnectFailure(message: string): void {
      this.messageService.error(`Could not connect to ${this.serverName}: ` + message);
   }

   /** Poll the port command until it answers, its attempts run out, or `signal` aborts. */
   protected async findPort(signal?: AbortSignal): Promise<number> {
      const pendingContent = new Deferred<number>();
      let counter = 0;
      let timer: ReturnType<typeof setTimeout> | undefined;
      const stop = (): void => {
         clearTimeout(timer);
         pendingContent.reject(signal?.reason);
      };
      if (signal?.aborted) {
         stop();
      }
      signal?.addEventListener('abort', stop, { once: true });
      const tryQueryingPort = (): void => {
         if (signal?.aborted) {
            return;
         }
         timer = setTimeout(async () => {
            try {
               const port = await this.commandService.executeCommand<number>(this.portCommand);
               // An empty answer fails the attempt like a throw does: left
               // unanswered, it neither resolves nor re-queues, and the lookup
               // stays pending with nothing logged.
               if (!port) {
                  throw new Error(`Port command '${this.portCommand}' answered without a port.`);
               }
               pendingContent.resolve(port);
            } catch (error) {
               counter++;
               const message = describeError(error);
               this.logger.debug(`[${this.logComponent}] Port command '${this.portCommand}' attempt ${counter} failed: ${message}`);
               if (this.findPortAttempts >= 0 && counter > this.findPortAttempts) {
                  pendingContent.reject(error);
               } else {
                  tryQueryingPort();
               }
            }
         }, this.findPortTimeout);
      };
      tryQueryingPort();
      return pendingContent.promise;
   }

   protected async connectToServer(
      channel: Channel,
      port: number,
      preForwardBuffer?: { bufferSub: Disposable; buffered: MessageProvider[] }
   ): Promise<void> {
      const connected = new Deferred<void>();
      const socket = new net.Socket();
      this.onSocketCreated?.(socket);
      let ready = false;
      socket.on('ready', () => {
         ready = true;
         connected.resolve();
      });
      socket.on('close', () => connected.reject(`Socket to ${this.serverName} was closed.`));
      // Before `ready` the error fails the connect, whose caller logs it; the `close` that follows rejects too late to count.
      socket.on('error', error => {
         if (ready) {
            this.logger.error(`Error occurred with the ${this.serverName} socket: ${describeErrorWithStack(error)}`);
         } else {
            connected.reject(error);
         }
      });
      // Synchronous hand-off: dispose the pre-forward buffer FIRST, then wire the
      // forwarder, then replay buffered messages. The event loop cannot
      // interleave between these synchronous statements, so no message is both
      // buffered AND forwarded (no double delivery), and none arriving this tick
      // slips through without a subscriber.
      preForwardBuffer?.bufferSub.dispose();
      this.forwardToSocketConnection(channel, socket);
      if (preForwardBuffer && preForwardBuffer.buffered.length > 0) {
         this.replayBufferedMessages(channel, preForwardBuffer.buffered);
      }
      if (channel instanceof AbstractChannel) {
         socket.on('error', error => channel.onErrorEmitter.fire(error));
      }
      // `127.0.0.1`, not the `localhost` Node dials when no host is given, and it
      // has to match the head: the socket launchers bind `127.0.0.1` by default,
      // so on a dual-stack machine where `localhost` resolves to `::1` first the
      // dial fails as ECONNREFUSED with nothing in the message naming the address
      // family as the cause.
      socket.connect({ port, host: '127.0.0.1' });
      setTimeout(() => connected.reject('Timeout reached.'), this.connectTimeoutMs);
      return connected.promise;
   }

   /**
    * Bridge the Theia frontend `clientChannel` and the server `socket` so bytes
    * relay both ways. Called once, after the socket is created and the
    * pre-forward buffer is about to be replayed.
    */
   protected forwardToSocketConnection(clientChannel: Channel, socket: net.Socket): Disposable {
      return new SocketChannelForwarder(clientChannel, socket, this.clock);
   }

   /**
    * Re-fire pre-forward buffered `MessageProvider`s on the channel's
    * `onMessageEmitter` so the now-subscribed forwarder picks them up in arrival
    * order. `MessageProvider` is a thunk (`() => ReadBuffer`) — buffering does
    * not consume the read position, so replay produces the same bytes the
    * forwarder would have seen if wired earlier.
    *
    * The emitter is public on `AbstractChannel`; the `Channel` interface
    * carries just its `Event`, so any other channel has its messages logged as
    * dropped. A Theia rename of the emitter fails to compile here.
    */
   protected replayBufferedMessages(channel: Channel, buffered: MessageProvider[]): void {
      if (!(channel instanceof AbstractChannel)) {
         this.logger.error(`[${this.logComponent}] dropping ${buffered.length} pre-forward message(s) — channel is not an AbstractChannel`);
         return;
      }
      this.logger.debug(`[${this.logComponent}] replaying ${buffered.length} pre-forward message(s) onto the wired socket forwarder`);
      for (const provider of buffered) {
         channel.onMessageEmitter.fire(provider);
      }
   }
}