import { Inject, Logger, OnModuleDestroy, OnModuleInit } from "@nestjs/common";
import {
  ConnectedSocket,
  MessageBody,
  OnGatewayConnection,
  OnGatewayDisconnect,
  SubscribeMessage,
  WebSocketGateway,
  WebSocketServer,
} from "@nestjs/websockets";
import Redis from "ioredis";
import { Server, Socket } from "socket.io";
import * as messaging from "@slackwsh/messaging";
import { MESSAGING_EVENTS_CHANNEL, type RoomEvent } from "@slackwsh/messaging";
import { REDIS_PUBSUB } from "./redis-pubsub.provider";
import { verifyAccessToken } from "./ws-auth";
import { PresenceService } from "./presence.service";

const TYPING_THROTTLE_SECONDS = 3;

interface JoinPayload {
  workspaceId: string;
  channelId: string;
}

interface WorkspaceJoinPayload {
  workspaceId: string;
}

interface SendPayload {
  workspaceId: string;
  channelId: string;
  clientMsgId: string;
  text: string;
  blocks?: unknown;
  parentId?: string | null;
  isBroadcast?: boolean;
}

interface AckReadPayload {
  workspaceId: string;
  channelId: string;
  seq: number;
}

interface AckThreadPayload {
  workspaceId: string;
  channelId: string;
  rootMessageId: string;
  seq: number;
}

interface CallSignalPayload {
  workspaceId: string | number;
  callId: string | number;
  toUserId: string | number;
  signal: "offer" | "answer" | "ice";
  /** Opaque SDP blob or ICE candidate — see CallSignalRequest in contracts. */
  data: unknown;
}

/**
 * Connection lifecycle per §4.3: tenant scope (here, just `userId` — see
 * ws-auth.ts for the ticket-vs-JWT simplification) is fixed at connect and
 * never re-read from client input. Handlers stay trivial — validate, then
 * delegate to libs/messaging, which is the same code apps/api's HTTP routes
 * call. Nothing is broadcast directly from a handler; every mutation
 * publishes to Redis (events-bus.ts) and onModuleInit's subscription below
 * is the only thing that ever calls `server.to(room).emit(...)`, so the
 * send-via-HTTP and send-via-socket paths broadcast identically.
 */
@WebSocketGateway({ cors: { origin: true } })
export class AppGateway implements OnGatewayConnection, OnGatewayDisconnect, OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(AppGateway.name);
  private eventsSub?: Redis;

  @WebSocketServer()
  server!: Server;

  constructor(
    @Inject(REDIS_PUBSUB) private readonly redis: Redis,
    private readonly presence: PresenceService,
  ) {}

  async onModuleInit() {
    this.eventsSub = new Redis(process.env.REDIS_URL ?? "redis://localhost:6379");
    await this.eventsSub.subscribe(MESSAGING_EVENTS_CHANNEL);
    this.eventsSub.on("message", (_channel, raw) => {
      try {
        const { room, event } = JSON.parse(raw) as RoomEvent;
        this.server.to(room).emit(event.type, event.payload);
      } catch (err) {
        this.logger.error(`failed to process messaging event: ${String(err)}`);
      }
    });
  }

  async onModuleDestroy() {
    await this.eventsSub?.quit();
  }

  handleConnection(client: Socket) {
    try {
      const userId = verifyAccessToken(client);
      client.data.userId = userId;
      client.data.workspaces = new Set<string>();
      client.join(`u:${userId}`);
      this.logger.log(`socket connected: ${client.id} (user ${userId})`);
    } catch (err) {
      this.logger.warn(`rejecting socket ${client.id}: ${String(err)}`);
      client.disconnect(true);
    }
  }

  /** Presence is unregistered here for every workspace this socket ever
   * joined (§4.4) — device-count only drops to zero, and `offline` only
   * broadcasts, once every workspace the user was active in agrees. */
  async handleDisconnect(client: Socket) {
    this.logger.log(`socket disconnected: ${client.id}`);
    const userId = client.data.userId as number | undefined;
    const workspaces = client.data.workspaces as Set<string> | undefined;
    if (!userId || !workspaces) return;

    for (const workspaceId of workspaces) {
      await this.presence.disconnect(workspaceId, userId, client.id).catch((err) => this.logger.error(String(err)));
    }
  }

  @SubscribeMessage("workspace:join")
  async handleJoinWorkspace(@ConnectedSocket() client: Socket, @MessageBody() payload: WorkspaceJoinPayload) {
    const userId = client.data.userId as number;
    const isMember = await messaging.checkWorkspaceMembership(payload.workspaceId, userId);
    if (!isMember) return { error: "not a member of this workspace" };

    const workspaces = client.data.workspaces as Set<string>;
    if (!workspaces.has(payload.workspaceId)) {
      workspaces.add(payload.workspaceId);
      client.join(`ws:${payload.workspaceId}`);
      await this.presence.connect(payload.workspaceId, userId, client.id);
    }

    return { ok: true, workspaceId: payload.workspaceId };
  }

  @SubscribeMessage("channel:join")
  async handleJoinChannel(@ConnectedSocket() client: Socket, @MessageBody() payload: JoinPayload) {
    const userId = client.data.userId as number;
    const isMember = await messaging.checkChannelMembership(payload.workspaceId, payload.channelId, userId);
    if (!isMember) return { error: "not a member of this channel" };

    client.join(`ch:${payload.channelId}`);

    const workspaces = client.data.workspaces as Set<string>;
    if (!workspaces.has(payload.workspaceId)) {
      workspaces.add(payload.workspaceId);
      client.join(`ws:${payload.workspaceId}`);
      await this.presence.connect(payload.workspaceId, userId, client.id);
    }

    return { ok: true, joined: payload.channelId };
  }

  @SubscribeMessage("presence:heartbeat")
  async handleHeartbeat(@ConnectedSocket() client: Socket, @MessageBody() payload: { active?: boolean }) {
    const userId = client.data.userId as number;
    const workspaces = client.data.workspaces as Set<string>;
    await this.presence.heartbeat(client.id, userId, workspaces, payload?.active !== false);
    return { ok: true };
  }

  @SubscribeMessage("typing:start")
  async handleTyping(
    @ConnectedSocket() client: Socket,
    @MessageBody() payload: { workspaceId: string | number; channelId: string | number },
  ) {
    const userId = client.data.userId as number;
    if (payload.workspaceId == null || payload.channelId == null) return { error: "workspaceId and channelId are required" };
    const isMember = await messaging.checkChannelMembership(payload.workspaceId, payload.channelId, userId);
    if (!isMember) return { error: "not a member of this channel" };
    const channelId = messaging.asEntityId(payload.channelId);
    // Throttled server-side to one event per user per channel per 3s (§4.3)
    // via SET NX EX — the atomic "only if absent" is what makes this safe
    // under concurrent events from the same user's multiple tabs.
    const key = `typing:${channelId}:${userId}`;
    const acquired = await this.redis.set(key, "1", "EX", TYPING_THROTTLE_SECONDS, "NX");
    if (!acquired) return { ok: true, throttled: true };

    await messaging.publishRoomEvent(this.redis, {
      room: `ch:${channelId}`,
      event: { type: "typing:changed", payload: { channelId, userId, isTyping: true } },
    });
    return { ok: true };
  }

  /**
   * Relays one WebRTC offer/answer/ICE candidate to one other participant.
   *
   * The only handler that forwards a client's payload rather than acting on it,
   * because SDP and ICE belong to the browser's WebRTC implementation, not to
   * this protocol. What the gateway *does* enforce is who may talk to whom:
   * both ends must currently be on the named call, checked against the
   * database rather than trusted from the payload, and `fromUserId` is stamped
   * from the authenticated socket so a sender cannot impersonate a peer.
   *
   * Emitted directly instead of through the Redis events bus: this is a
   * point-to-point message with nothing to persist and no other subscriber, and
   * the negotiation is latency-sensitive enough that the extra hop isn't worth
   * it. `u:{userId}` spans nodes via the Redis socket.io adapter regardless,
   * so a peer on another gateway instance still receives it.
   */
  @SubscribeMessage("call:signal")
  async handleCallSignal(
    @ConnectedSocket() client: Socket,
    @MessageBody() payload: CallSignalPayload,
  ) {
    const userId = client.data.userId as number;
    const toUserId = messaging.asEntityId(payload.toUserId);
    if (toUserId === userId) return { error: "cannot signal yourself" };

    /**
     * Relays are serialised per sending socket.
     *
     * Each signal is its own async invocation, and the authorisation below is
     * two database round-trips, so without this chain the emit order is decided
     * by whichever pair of queries finishes first — not by the order the client
     * sent them. That reordering lets trickled ICE candidates overtake the offer
     * they belong to; the receiver then has candidates for a connection with no
     * remote description yet. The client queues them too (belt and braces), but
     * ordering is this layer's to preserve, not the client's to repair.
     */
    const chain: Promise<unknown> = client.data.signalChain ?? Promise.resolve();
    const relay = chain.then(async () => {
      const [maySend, mayReceive] = await Promise.all([
        messaging.isCallParticipant(payload.workspaceId, payload.callId, userId),
        messaging.isCallParticipant(payload.workspaceId, payload.callId, toUserId),
      ]);
      if (!maySend || !mayReceive) return { relayed: false };

      this.server.to(`u:${toUserId}`).emit("call:signal", {
        callId: messaging.asEntityId(payload.callId),
        fromUserId: userId,
        signal: payload.signal,
        data: payload.data,
      });
      return { relayed: true };
    });
    // A rejected link must not poison every later signal from this socket.
    client.data.signalChain = relay.catch(() => undefined);

    const result = await relay;
    return result.relayed ? { ok: true } : { error: "not on this call" };
  }

  @SubscribeMessage("channel:leave")
  handleLeaveChannel(@ConnectedSocket() client: Socket, @MessageBody() payload: { channelId: string | number }) {
    client.leave(`ch:${payload.channelId}`);
    return { left: payload.channelId };
  }

  @SubscribeMessage("message:send")
  async handleSendMessage(@ConnectedSocket() client: Socket, @MessageBody() payload: SendPayload) {
    const userId = client.data.userId as number;
    const message = await messaging.sendMessage(this.redis, payload.workspaceId, payload.channelId, userId, {
      clientMsgId: payload.clientMsgId,
      text: payload.text,
      blocks: payload.blocks,
      parentId: payload.parentId,
      isBroadcast: payload.isBroadcast,
    });
    return { ok: true, messageId: message.id, seq: message.seq };
  }

  @SubscribeMessage("read:ack")
  async handleReadAck(@ConnectedSocket() client: Socket, @MessageBody() payload: AckReadPayload) {
    const userId = client.data.userId as number;
    await messaging.markChannelRead(this.redis, payload.workspaceId, payload.channelId, userId, payload.seq);
    return { ok: true };
  }

  @SubscribeMessage("thread:ack")
  async handleThreadAck(@ConnectedSocket() client: Socket, @MessageBody() payload: AckThreadPayload) {
    const userId = client.data.userId as number;
    await messaging.markThreadRead(this.redis, payload.workspaceId, payload.channelId, userId, payload.rootMessageId, payload.seq);
    return { ok: true };
  }
}