import { EventEmitter } from "events";
import * as websocket from "websocket";
import { subscriber as redisClient } from "../../db/redis.js";
import { Users } from "../../models/index.js";
import MainStreamConnection from "./stream/index.js";
import authenticate from "./authenticate.js";
import { apiLogger } from "./logger.js";
import { MastodonStreamingConnection } from "./mastodon/streaming/index.js";
import { getTokenFromOAuth } from "./mastodon/middleware/auth.js";
export const streamingLogger = apiLogger.createSubLogger("streaming");
export const initializeStreamingServer = (server)=>{
    // Init websocket server
    const ws = new websocket.server({
        httpServer: server
    });
    ws.on("request", async (request)=>{
        const q = request.resourceURL.query;
        const headers = request.httpRequest.headers["sec-websocket-protocol"] || "";
        const cred = q.i || q.access_token || headers;
        const rawAccessToken = cred.toString();
        const accessToken = rawAccessToken.length > 0 ? rawAccessToken : null;
        const isMastodon = request.resourceURL.pathname?.startsWith('/api/v1/streaming');
        let main;
        let user;
        let app;
        let token;
        if (!isMastodon) {
            [user, app] = await authenticate(request.httpRequest.headers.authorization, accessToken).catch((err)=>{
                request.reject(403, err.message);
                return [];
            });
        } else {
            token = await getTokenFromOAuth(accessToken);
            if (!token || !token.user) {
                request.reject(400);
                return;
            }
            user = token.user;
        }
        if (user?.isSuspended) {
            request.reject(400);
            return;
        }
        const connection = request.accept(request.requestedProtocols[0] ?? undefined);
        const ev = new EventEmitter();
        async function onRedisMessage(_, data) {
            const parsed = JSON.parse(data);
            ev.emit(parsed.channel, parsed.message);
        }
        redisClient.on("message", onRedisMessage);
        const host = `https://${request.host}`;
        const prepareStream = q.stream?.toString();
        main = isMastodon ? new MastodonStreamingConnection(connection, ev, user, token, q) : new MainStreamConnection(connection, ev, user, app, host, accessToken ?? "", prepareStream);
        const intervalId = user ? setInterval(()=>{
            Users.update(user.id, {
                lastActiveDate: new Date()
            });
        }, 1000 * 60 * 5) : null;
        if (user) {
            Users.update(user.id, {
                lastActiveDate: new Date()
            });
        }
        connection.once("close", ()=>{
            ev.removeAllListeners();
            main.dispose();
            redisClient.off("message", onRedisMessage);
            if (intervalId) clearInterval(intervalId);
        });
        connection.on("message", async (data)=>{
            if (data.type === "utf8" && data.utf8Data === "ping") {
                connection.send("pong");
            }
        });
    });
};