import { Encoder, Schema, type StateView } from '@colyseus/schema';
import type { PacketChannel } from './channels';
import type { Envelope } from './protocol';
import { P2P_CLOSE_CODES } from './protocol';
import type { BroadcastOptions, CompatClient, Room } from './room';
import { registerLoopbackRoom } from './room-registry';
import { bytesToBase64 } from './wire-bytes';

export type RoomClass<T extends Room = Room> = new () => T;

const EMPTY_BYTES = new Uint8Array(0);

interface RuntimeClient extends CompatClient {
  readonly channel: PacketChannel;
  /** Per-client monotonic patch sequence — the client resyncs on a gap. */
  clock: number;
}

interface PendingReconnection {
  readonly previousClient: RuntimeClient;
  readonly deferred: ReconnectionDeferred;
  readonly timer?: ReturnType<typeof setTimeout>;
}

export class UniversalRoomRuntime<T extends Room = Room> {
  readonly room: T;
  private readonly clients = new Map<string, RuntimeClient>();
  private readonly reconnections = new Map<string, PendingReconnection>();
  /**
   * The REAL `@colyseus/schema` encoder bound to `room.state`. Created lazily
   * (after `onCreate`, so field-initialized or `setState`-assigned Schema is
   * populated) and rebuilt if `room.state` is replaced. Rooms whose state is
   * not a `Schema` (lifecycle-only fixtures) simply do not replicate — there
   * is nothing the real encoder can carry.
   */
  private encoder?: Encoder<Schema>;
  private hasFilters = false;
  /** Set when `room.state` is REPLACED (setState) so all clients get a fresh full snapshot. */
  private pendingResyncAll = false;

  constructor(
    roomClass: RoomClass<T>,
    private readonly roomName: string,
    options?: unknown,
  ) {
    // The room class is also the client-side decoder root source; register it
    // so a co-located client (loopback, browser host-local, universal Node)
    // resolves the decoder from the same class the host encodes with.
    registerLoopbackRoom(roomName, roomClass);
    this.room = new roomClass();
    this.room._attachRuntime({
      broadcastMessage: (type, payload, options) => this.broadcastMessage(type, payload, options),
      broadcastState: () => this.broadcastState(),
      disconnect: (code) => this.disconnect(code),
      send: (client, type, payload) => this.sendToClient(client, type, payload),
      removeClient: (client, code, reason) =>
        this.removeClient(client as RuntimeClient, true, code, reason),
      allowReconnection: (client, seconds) =>
        this.allowReconnection(client as RuntimeClient, seconds),
    });
    this.room.onCreate?.(options);
  }

  attach(channel: PacketChannel): void {
    channel.onEnvelope((envelope) => this.handleEnvelope(channel, envelope));
    channel.onClose(() => this.removeByChannel(channel, true));
  }

  private handleEnvelope(channel: PacketChannel, envelope: Envelope): void {
    if (envelope.kind === 'ping') {
      channel.send({ kind: 'pong', t: envelope.t });
      return;
    }

    if (envelope.kind === 'join') {
      if (envelope.room !== this.roomName) {
        channel.send({
          kind: 'join-error',
          requestId: envelope.requestId,
          message: `Room not found: ${envelope.room}`,
        });
        closeAfterSend(channel, String(P2P_CLOSE_CODES.roomNotFound));
        return;
      }
      void this.join(channel, envelope.requestId, envelope.options);
      return;
    }

    const client = this.findByChannel(channel);
    if (!client) return;

    if (envelope.kind === 'message') {
      try {
        this.room._dispatchMessage(client, envelope.type, envelope.payload);
      } catch (error) {
        channel.send({
          kind: 'message',
          type: 'error',
          payload: {
            code: P2P_CLOSE_CODES.internalError,
            message: error instanceof Error ? error.message : 'server error',
          },
        });
      }
    } else if (envelope.kind === 'leave') {
      this.removeClient(client, true);
    } else if (envelope.kind === 'state-resync') {
      this.sendSnapshot(client);
    }
  }

  private async join(channel: PacketChannel, requestId: string, options?: unknown): Promise<void> {
    if (this.room.locked) {
      channel.send({
        kind: 'join-error',
        requestId,
        message: String(P2P_CLOSE_CODES.roomFull),
      });
      closeAfterSend(channel, String(P2P_CLOSE_CODES.roomFull));
      return;
    }

    const reconnectionToken = getReconnectionToken(options);
    const pendingReconnection = reconnectionToken
      ? this.reconnections.get(reconnectionToken)
      : undefined;

    const sessionId = pendingReconnection?.previousClient.sessionId ?? channel.peerId;
    const events = new MiniEmitter();
    const client: RuntimeClient = {
      id: sessionId,
      sessionId,
      state: 1,
      reconnectionToken: pendingReconnection?.previousClient.reconnectionToken ?? sessionId,
      ref: events,
      channel,
      clock: 0,
      raw: (data, _options, cb) => {
        channel.send({ kind: 'message', type: 'raw', payload: data });
        cb?.();
      },
      enqueueRaw: (data) => channel.send({ kind: 'message', type: 'raw', payload: data }),
      send: (type, payload) => channel.send({ kind: 'message', type: String(type), payload }),
      sendBytes: (type, bytes) =>
        channel.send({ kind: 'message', type: String(type), payload: bytes }),
      leave: (code, data) => this.removeClient(client, true, code, data),
      close: (code, data) => this.removeClient(client, true, code, data),
      error: (code, message) =>
        channel.send({ kind: 'message', type: 'error', payload: { code, message } }),
    };
    try {
      const authResult = await this.room.onAuth?.(client, options);
      if (authResult === false) {
        channel.send({
          kind: 'join-error',
          requestId,
          message: String(P2P_CLOSE_CODES.unauthorized),
        });
        closeAfterSend(channel, String(P2P_CLOSE_CODES.unauthorized));
        return;
      }
      client.auth = authResult;
    } catch (error) {
      channel.send({
        kind: 'join-error',
        requestId,
        message: error instanceof Error ? error.message : String(P2P_CLOSE_CODES.unauthorized),
      });
      closeAfterSend(channel, String(P2P_CLOSE_CODES.unauthorized));
      return;
    }
    if (pendingReconnection) {
      if (pendingReconnection.timer) clearTimeout(pendingReconnection.timer);
      this.reconnections.delete(reconnectionToken!);
    }
    if (!pendingReconnection && this.room.locked) {
      channel.send({
        kind: 'join-error',
        requestId,
        message: String(P2P_CLOSE_CODES.roomFull),
      });
      closeAfterSend(channel, String(P2P_CLOSE_CODES.roomFull));
      return;
    }
    this.clients.set(sessionId, client);
    this.room.clients.push(client);
    if (pendingReconnection) {
      this.room.onReconnect?.(client);
      pendingReconnection.deferred.resolve(client);
    } else {
      this.room.onJoin?.(client, options);
    }
    // Full state for THIS client (respecting its `@view()`), then the delta
    // that onJoin produced goes to the OTHER clients only — the joiner already
    // has it in its full snapshot, so it must not also receive the patch.
    const fullBytes = this.encodeFullState(client.view);
    client.clock += 1;
    channel.send({
      kind: 'join-ok',
      requestId,
      sessionId,
      state: bytesToBase64(fullBytes),
      clock: client.clock,
    });
    this.flushPatches(sessionId);
  }

  private removeByChannel(channel: PacketChannel, consented: boolean): void {
    const client = this.findByChannel(channel);
    if (client) this.removeClient(client, consented);
  }

  private removeClient(
    client: RuntimeClient,
    consented: boolean,
    code = 1000,
    reason?: string,
  ): void {
    if (!this.clients.delete(client.sessionId)) return;
    client.state = 5;
    this.room.clients.delete(client);
    if (!consented) this.room.onDrop?.(client, code);
    this.room.onLeave?.(client, code);
    client.channel.send(
      reason === undefined ? { kind: 'leave', code } : { kind: 'leave', code, reason },
    );
    closeAfterSend(client.channel, reason ?? 'left');
    this.broadcastState();
    this.disposeIfIdle();
  }

  private findByChannel(channel: PacketChannel): RuntimeClient | undefined {
    for (const client of this.clients.values()) {
      if (client.channel === channel) return client;
    }
    return undefined;
  }

  private sendToClient(client: CompatClient, type: string | number, payload?: unknown): void {
    const runtimeClient = this.clients.get(client.sessionId);
    runtimeClient?.channel.send({ kind: 'message', type: String(type), payload });
  }

  private broadcastMessage(
    type: string | number,
    payload?: unknown,
    options?: BroadcastOptions,
  ): void {
    const except = new Set(
      Array.isArray(options?.except)
        ? options.except.map((client) => client.sessionId)
        : options?.except
          ? [options.except.sessionId]
          : [],
    );
    for (const client of this.clients.values()) {
      if (!except.has(client.sessionId))
        client.channel.send({ kind: 'message', type: String(type), payload });
    }
  }

  broadcastState(): boolean {
    return this.flushPatches();
  }

  /**
   * Bind (or rebind) the real encoder to `room.state`. Returns undefined when
   * the room has no `Schema` state — such a room simply does not replicate.
   */
  private ensureEncoder(): Encoder<Schema> | undefined {
    const state = this.room.state as unknown;
    // Stateless / lifecycle-only rooms are legitimate — nothing to replicate,
    // stay silent.
    if (state === undefined || state === null) return undefined;
    // A non-Schema state (e.g. `setState({ score: 0 })` with a plain object)
    // CANNOT be replicated by the real encoder. Real Colyseus throws on this;
    // surface it LOUDLY at the call rather than silently dropping replication —
    // a silent divergence is exactly the defect this seam exists to kill.
    if (!(state instanceof Schema)) {
      const got =
        typeof state === 'object'
          ? ((state as { constructor?: { name?: string } }).constructor?.name ?? 'plain object')
          : typeof state;
      throw new Error(
        `@vgai/p2p-colyseus: room "${this.roomName}" state must extend @colyseus/schema Schema; ` +
          `got ${got}. setState(plainObject) does not replicate — declare a Schema state class ` +
          `(e.g. \`class State extends Schema { @type('number') score = 0 }\`) and ` +
          `setState(new State()).`,
      );
    }
    if (this.encoder && this.encoder.state === state) return this.encoder;
    if (this.encoder) this.pendingResyncAll = true; // state instance was replaced (setState)
    this.encoder = new Encoder(state);
    this.hasFilters = this.encoder.context.hasFilters;
    return this.encoder;
  }

  /**
   * Encode the FULL current state for a client — filtered to its `@view()`
   * when the schema has view-gated fields and the client has a view. Mirrors
   * `@colyseus/core` `SchemaSerializer.getFullState`.
   */
  private encodeFullState(view?: StateView): Uint8Array {
    const encoder = this.ensureEncoder();
    if (!encoder) return EMPTY_BYTES;
    const it = { offset: 0 };
    const full = encoder.encodeAll(it);
    const sharedOffset = it.offset;
    if (this.hasFilters && view) {
      return Uint8Array.from(encoder.encodeAllView(view, sharedOffset, it));
    }
    return Uint8Array.from(full);
  }

  /**
   * Encode the pending delta once and fan it out per client — mirrors
   * `@colyseus/core` `SchemaSerializer.applyPatches`: viewless clients share
   * one encoded patch, while each `@view()` client gets `encodeView` bytes that
   * NEVER contain another view's filtered fields (the confidentiality property).
   * `exceptSessionId` skips a just-joined client that already holds the full
   * state.
   */
  private flushPatches(exceptSessionId?: string): boolean {
    const encoder = this.ensureEncoder();
    const recipients = [...this.clients.values()].filter(
      (client) => client.sessionId !== exceptSessionId,
    );
    if (!encoder || recipients.length === 0) {
      encoder?.discardChanges();
      return false;
    }

    if (this.pendingResyncAll) {
      this.pendingResyncAll = false;
      for (const client of recipients) this.sendSnapshotTo(client);
      encoder.discardChanges();
      return true;
    }

    if (!encoder.hasChanges) {
      // No state mutation, but a client may have manual view add/remove ops.
      if (this.hasFilters) {
        const it = { offset: 0 };
        const sharedOffset = it.offset;
        for (const client of recipients) {
          if (client.view && client.view.changes.size > 0) {
            this.sendPatchTo(
              client,
              Uint8Array.from(encoder.encodeView(client.view, sharedOffset, it)),
            );
          }
        }
      }
      return false;
    }

    const it = { offset: 0 };
    const encodedChanges = encoder.encode(it);
    const sharedOffset = it.offset;
    // Copy the shared (non-filtered) changes before any encodeView reuses the buffer.
    const sharedCopy = Uint8Array.from(encodedChanges);
    if (!this.hasFilters) {
      for (const client of recipients) this.sendPatchTo(client, sharedCopy);
    } else {
      const perView = new Map<StateView, Uint8Array>();
      for (const client of recipients) {
        if (!client.view) {
          this.sendPatchTo(client, sharedCopy);
          continue;
        }
        let bytes = perView.get(client.view);
        if (!bytes) {
          bytes = Uint8Array.from(encoder.encodeView(client.view, sharedOffset, it));
          perView.set(client.view, bytes);
        }
        this.sendPatchTo(client, bytes);
      }
    }
    encoder.discardChanges();
    return true;
  }

  private sendPatchTo(client: RuntimeClient, bytes: Uint8Array): void {
    client.channel.send({
      kind: 'state-patch',
      patch: bytesToBase64(bytes),
      clock: ++client.clock,
    });
  }

  private sendSnapshotTo(client: RuntimeClient): void {
    const bytes = this.encodeFullState(client.view);
    client.channel.send({
      kind: 'state-snapshot',
      state: bytesToBase64(bytes),
      clock: ++client.clock,
    });
  }

  private async disconnect(code = 4000): Promise<void> {
    for (const client of [...this.clients.values()])
      this.removeClient(client, true, code, 'disconnect');
  }

  private allowReconnection(
    client: RuntimeClient,
    seconds: number | 'manual',
  ): ReconnectionDeferred {
    const token = client.reconnectionToken;
    const existing = this.reconnections.get(token);
    if (existing?.timer) clearTimeout(existing.timer);
    const deferred = new ReconnectionDeferred();
    const timer =
      seconds === 'manual'
        ? undefined
        : setTimeout(
            () => {
              this.reconnections.delete(token);
              deferred.reject(new Error('reconnection timeout'));
              this.disposeIfIdle();
            },
            Math.max(0, seconds * 1000),
          );
    if (timer) (timer as unknown as { unref?: () => void }).unref?.();
    this.reconnections.set(token, {
      previousClient: client,
      deferred,
      ...(timer ? { timer } : {}),
    });
    return deferred;
  }

  private disposeIfIdle(): void {
    if (this.clients.size === 0 && this.reconnections.size === 0 && this.room.autoDispose) {
      this.room.onDispose?.();
      this.room._disposeRuntime();
    }
  }

  private sendSnapshot(client: RuntimeClient): void {
    this.sendSnapshotTo(client);
  }
}

class MiniEmitter {
  private readonly handlers = new Map<string, Set<(...args: unknown[]) => void>>();

  on(event: string, handler: (...args: unknown[]) => void): void {
    let eventHandlers = this.handlers.get(event);
    if (!eventHandlers) {
      eventHandlers = new Set();
      this.handlers.set(event, eventHandlers);
    }
    eventHandlers.add(handler);
  }

  off(event: string, handler: (...args: unknown[]) => void): void {
    this.handlers.get(event)?.delete(handler);
  }

  emit(event: string, ...args: unknown[]): void {
    for (const handler of this.handlers.get(event) ?? []) handler(...args);
  }
}

class ReconnectionDeferred implements Promise<CompatClient> {
  readonly [Symbol.toStringTag] = 'Promise';
  private resolvePromise!: (client: CompatClient) => void;
  private rejectPromise!: (reason?: unknown) => void;
  private readonly promise = new Promise<CompatClient>((resolve, reject) => {
    this.resolvePromise = resolve;
    this.rejectPromise = reject;
  });

  resolve(client: CompatClient): void {
    this.resolvePromise(client);
  }

  reject(reason?: unknown): void {
    this.rejectPromise(reason);
  }

  // biome-ignore lint/suspicious/noThenProperty: allowReconnection returns a Promise-compatible thenable in Colyseus.
  then<TResult1 = CompatClient, TResult2 = never>(
    onfulfilled?: ((value: CompatClient) => TResult1 | PromiseLike<TResult1>) | null,
    onrejected?: ((reason: unknown) => TResult2 | PromiseLike<TResult2>) | null,
  ): Promise<TResult1 | TResult2> {
    return this.promise.then(onfulfilled, onrejected);
  }

  catch<TResult = never>(
    onrejected?: ((reason: unknown) => TResult | PromiseLike<TResult>) | null,
  ): Promise<CompatClient | TResult> {
    return this.promise.catch(onrejected);
  }

  finally(onfinally?: (() => void) | null): Promise<CompatClient> {
    return this.promise.finally(onfinally);
  }
}

function getReconnectionToken(options: unknown): string | undefined {
  if (typeof options !== 'object' || options === null) return undefined;
  const token = (options as { reconnectionToken?: unknown }).reconnectionToken;
  return typeof token === 'string' ? token : undefined;
}

function closeAfterSend(channel: PacketChannel, reason: string): void {
  queueMicrotask(() => channel.close(reason));
}