import { emitter } from '@amatiasq/emitter';

import { Message } from '../Message.js';
import { SocketTransport } from '../transport.js';
import { ClientId } from './ClientId.js';
import { ICMType, InternalClientMessage } from './messaging.js';
import { parseJson } from './parseJson.js';
import { SessionClients } from './SessionClients.js';
import { SessionConnection } from './SessionConnection.js';

/**
 * The server side of the session layer: hand it each incoming socket and it
 * resolves which client it belongs to, new or returning.
 *
 * **It does not listen on anything.** This was `WebSocketServer` — a name that
 * collided with `ws`'s own class, which it built internally from an
 * `http.Server`. The caller now owns the listening socket, so this package needs
 * no WebSocket implementation at all; see `SocketTransport`.
 *
 *     const server = new SessionServer<ServerMessage, ClientMessage>();
 *     server.onConnection(client => client.onMessageType('ping', …));
 *
 *     // Node, with `ws`:
 *     wss.on('connection', ws => server.accept(fromEventEmitter(ws)));
 *
 * `accept` does not emit a connection straight away: it waits for the client's
 * handshake, because that is what says whether this is a new client or one
 * coming back with a `ClientId`. A returning client is bound to its existing
 * `SessionConnection` and does **not** re-emit `onConnection`.
 */
export class SessionServer<
  ServerMessage extends Message<any, any>,
  ClientMessage extends Message<any, any>
> {
  private readonly clients = new SessionClients<ServerMessage, ClientMessage>();

  private readonly emitConnection = emitter<
    SessionConnection<ServerMessage, ClientMessage>
  >();
  readonly onConnection = this.emitConnection.subscribe;

  accept(transport: SocketTransport) {
    const unbind: (() => void)[] = [];

    const stopListening = () => {
      for (const off of unbind) off();
      unbind.length = 0;
    };

    unbind.push(
      transport.onClose(stopListening),
      transport.onMessage(payload => {
        const msg = parseJson(payload) as InternalClientMessage;

        if (msg.type === ICMType.AMQ_CONNECT) {
          this.onClientConnected(transport);
        } else if (msg.type === ICMType.AMQ_RECONNECT) {
          this.onClientReconnect(transport, msg.data);
        } else {
          console.error('Unexpected message', msg.type);
        }

        // The handshake is over either way: from here on the socket belongs to
        // a `SessionConnection`, which binds its own listeners.
        stopListening();
      }),
    );
  }

  private onClientConnected(transport: SocketTransport) {
    const client = this.clients.add();
    client.bindTo(transport);
    this.emitConnection(client);
  }

  /**
   * An id we do not know is not adopted — the client gets a fresh session, and
   * `onConnection` fires as it would for any newcomer.
   *
   * This used to be `get(id) || add(id)`, which took the id from the wire at face
   * value. Combined with sequential ids it meant naming another client's id was
   * enough to be handed their session; on its own it still let a client choose an
   * id nobody issued.
   */
  private onClientReconnect(transport: SocketTransport, id: ClientId) {
    const client = this.clients.get(id);

    if (!client) {
      this.onClientConnected(transport);
      return;
    }

    client.bindTo(transport);
  }
}