import { createHmac, timingSafeEqual } from 'node:crypto';
import type { Server } from 'node:http';
import { WebSocketServer, WebSocket } from 'ws';
import type { Database } from '../db/types.js';
import type { AppEnv } from '../config/env.js';

interface TokenPayload { userId: string; sessionVersion: number; expiresAt: number }

function encode(value: Buffer | string): string {
  return Buffer.from(value).toString('base64url');
}

export function createRealtimeToken(payload: TokenPayload, secret: string): string {
  const body = encode(JSON.stringify(payload));
  const signature = encode(createHmac('sha256', secret).update(body).digest());
  return `${body}.${signature}`;
}

export function verifyRealtimeToken(token: string, secret: string): TokenPayload | null {
  const [body, supplied] = token.split('.');
  if (!body || !supplied) return null;
  const expected = encode(createHmac('sha256', secret).update(body).digest());
  const left = Buffer.from(supplied);
  const right = Buffer.from(expected);
  if (left.length !== right.length || !timingSafeEqual(left, right)) return null;
  try {
    const payload = JSON.parse(Buffer.from(body, 'base64url').toString('utf8')) as TokenPayload;
    return payload.expiresAt > Date.now() ? payload : null;
  } catch {
    return null;
  }
}

export interface RedisBridge {
  publish(channel: string, message: string): Promise<unknown>;
  subscribe(channel: string, listener: (message: string) => void): Promise<unknown>;
  duplicate(): RedisBridge;
  connect(): Promise<unknown>;
  quit(): Promise<unknown>;
  isOpen?: boolean;
}

export class RealtimeHub {
  private readonly wss = new WebSocketServer({ noServer: true });
  private readonly clientChannels = new Map<WebSocket, Set<string>>();
  private redisPublisher: RedisBridge | null = null;
  private redisSubscriber: RedisBridge | null = null;

  constructor(private readonly db: Database, private readonly env: AppEnv) {}

  async useRedis(client: RedisBridge): Promise<void> {
    this.redisPublisher = client;
    this.redisSubscriber = client.duplicate();
    if (!this.redisSubscriber.isOpen) await this.redisSubscriber.connect();
    await this.redisSubscriber.subscribe('mandegar:realtime', (message) => {
      try {
        const parsed = JSON.parse(message) as { channel: string; payload: Record<string, unknown>; instance?: string };
        this.publishLocal(parsed.channel, parsed.payload);
      } catch { /* پیام مخدوش Redis نادیده گرفته می‌شود. */ }
    });
  }

  attach(server: Server): void {
    server.on('upgrade', async (request, socket, head) => {
      try {
        const url = new URL(request.url ?? '/', this.env.APP_URL);
        if (url.pathname !== '/ws') return;
        const payload = verifyRealtimeToken(url.searchParams.get('token') ?? '', this.env.APP_KEY);
        if (!payload) { socket.destroy(); return; }
        const user = await this.db.query<{ id: string; session_version: number; warehouse_id: string | null; active: boolean }>(
          'SELECT id, session_version, warehouse_id, active FROM users WHERE id = $1 AND archived_at IS NULL', [payload.userId]
        );
        if (!user.rowCount || !user.rows[0]!.active || user.rows[0]!.session_version !== payload.sessionVersion) { socket.destroy(); return; }
        const roles = await this.db.query<{ code: string }>(
          `SELECT r.code FROM roles r JOIN user_roles ur ON ur.role_id = r.id WHERE ur.user_id = $1`, [payload.userId]
        );
        const channels = new Set([`user:${payload.userId}`]);
        if (roles.rows.some((role) => role.code === 'warehouse_operator') && user.rows[0]!.warehouse_id) {
          channels.add(`warehouse:${user.rows[0]!.warehouse_id}`);
        }
        this.wss.handleUpgrade(request, socket, head, (ws) => {
          this.clientChannels.set(ws, channels);
          this.wss.emit('connection', ws, request);
          ws.send(JSON.stringify({ type: 'connected', at: new Date().toISOString() }));
        });
      } catch { socket.destroy(); }
    });
    this.wss.on('connection', (socket) => {
      socket.on('close', () => this.clientChannels.delete(socket));
      socket.on('error', () => this.clientChannels.delete(socket));
      socket.on('message', (raw) => {
        if (raw.toString() === 'ping' && socket.readyState === WebSocket.OPEN) socket.send('pong');
      });
    });
  }

  async publish(channel: string, payload: Record<string, unknown>): Promise<void> {
    this.publishLocal(channel, payload);
    if (this.redisPublisher?.isOpen) {
      await this.redisPublisher.publish('mandegar:realtime', JSON.stringify({ channel, payload }));
    }
  }

  private publishLocal(channel: string, payload: Record<string, unknown>): void {
    const message = JSON.stringify({ ...payload, channel, at: new Date().toISOString() });
    for (const [client, channels] of this.clientChannels) {
      if (client.readyState === WebSocket.OPEN && channels.has(channel)) client.send(message);
    }
  }

  async close(): Promise<void> {
    for (const client of this.clientChannels.keys()) client.close(1001, 'server shutdown');
    this.wss.close();
    if (this.redisSubscriber?.isOpen) await this.redisSubscriber.quit();
  }
}
