import { createHmac, timingSafeEqual } from 'node:crypto';
import { WebSocketServer, WebSocket } from 'ws';
function encode(value) {
    return Buffer.from(value).toString('base64url');
}
export function createRealtimeToken(payload, secret) {
    const body = encode(JSON.stringify(payload));
    const signature = encode(createHmac('sha256', secret).update(body).digest());
    return `${body}.${signature}`;
}
export function verifyRealtimeToken(token, secret) {
    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'));
        return payload.expiresAt > Date.now() ? payload : null;
    }
    catch {
        return null;
    }
}
export class RealtimeHub {
    db;
    env;
    wss = new WebSocketServer({ noServer: true });
    clientChannels = new Map();
    redisPublisher = null;
    redisSubscriber = null;
    constructor(db, env) {
        this.db = db;
        this.env = env;
    }
    async useRedis(client) {
        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);
                this.publishLocal(parsed.channel, parsed.payload);
            }
            catch { /* پیام مخدوش Redis نادیده گرفته می‌شود. */ }
        });
    }
    attach(server) {
        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('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(`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, payload) {
        this.publishLocal(channel, payload);
        if (this.redisPublisher?.isOpen) {
            await this.redisPublisher.publish('mandegar:realtime', JSON.stringify({ channel, payload }));
        }
    }
    publishLocal(channel, payload) {
        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() {
        for (const client of this.clientChannels.keys())
            client.close(1001, 'server shutdown');
        this.wss.close();
        if (this.redisSubscriber?.isOpen)
            await this.redisSubscriber.quit();
    }
}
//# sourceMappingURL=hub.js.map