server.mjs

import { createServer } from "node:http"
import next from "next"
import { Server as SocketIOServer } from "socket.io"
import { decode } from "next-auth/jwt"
import Redis from "ioredis"
import { randomUUID } from "node:crypto"

const dev = process.env.NODE_ENV !== "production"
const hostname = process.env.HOSTNAME || "localhost"
const port = parseInt(process.env.PORT || "3000", 10)

const app = next({ dev, hostname, port })
const handle = app.getRequestHandler()

function parseCookies(cookieHeader) {
    const cookies = {}
    if (!cookieHeader) return cookies
    for (const cookie of cookieHeader.split(";")) {
        const eqIdx = cookie.indexOf("=")
        if (eqIdx === -1) continue
        const key = cookie.slice(0, eqIdx).trim()
        const value = cookie.slice(eqIdx + 1).trim()
        try {
            cookies[key] = decodeURIComponent(value)
        } catch {
            cookies[key] = value
        }
    }
    return cookies
}

await app.prepare()

const httpServer = createServer(async (req, res) => {
    const hostHeader = Array.isArray(req.headers.host)
        ? req.headers.host[0]
        : (req.headers.host || `${hostname}:${port}`)
    const parsedUrl = new URL(req.url, `http://${hostHeader}`)
    await handle(req, res, { pathname: parsedUrl.pathname, query: Object.fromEntries(parsedUrl.searchParams) })
})

const allowedOrigin = process.env.ALLOWED_ORIGIN || (dev ? `http://${hostname}:${port}` : false)

const io = new SocketIOServer(httpServer, {
    path: "/api/ws",
    transports: ["websocket", "polling"],
    cors: {
        origin: allowedOrigin,
        credentials: true,
    },
})

// Store Socket.IO instance globally for Next.js modules to access
globalThis.__socketIO = io

// Auth middleware — verify NextAuth JWT from cookie
io.use(async (socket, next) => {
    const cookieHeader = socket.request.headers.cookie || ""
    const cookies = parseCookies(cookieHeader)

    const secureName = "__Secure-authjs.session-token"
    const devName = "authjs.session-token"
    const tokenValue = cookies[secureName] || cookies[devName]
    const salt = cookies[secureName] ? secureName : devName

    if (!tokenValue) {
        return next(new Error("Unauthorized"))
    }

    try {
        const decoded = await decode({
            token: tokenValue,
            secret: process.env.AUTH_SECRET,
            salt,
        })

        if (!decoded?.sub) {
            return next(new Error("Unauthorized"))
        }

        socket.data.userId = decoded.sub
        next()
    } catch {
        return next(new Error("Unauthorized"))
    }
})

// Handle connections
io.on("connection", (socket) => {
    const userId = socket.data.userId
    if (userId) {
        socket.join(`user:${userId}`)
    }
})

// --- Shared Prisma & Redis clients ---
// Used by the FiveM namespace, heartbeat watchdog, cleanup worker, and simulation worker.
// A single instance of each is shared across all subsystems.
let sharedPrisma = null
let sharedRedis = null

async function getSharedPrisma() {
    if (sharedPrisma) return sharedPrisma
    const m = await import("@prisma/client")
    if (!globalThis.__sharedPrismaClient) {
        globalThis.__sharedPrismaClient = new m.PrismaClient()
    }
    sharedPrisma = globalThis.__sharedPrismaClient
    return sharedPrisma
}

async function getSharedRedis() {
    if (sharedRedis) return sharedRedis
    const url = process.env.REDIS_URL || "redis://localhost:6379"
    if (!globalThis.__sharedRedisClient) {
        globalThis.__sharedRedisClient = new Redis(url, {
            maxRetriesPerRequest: 3,
            lazyConnect: true,
            enableReadyCheck: false,
        })
        globalThis.__sharedRedisClient.connect().catch((err) => {
            console.error("[redis] Shared Redis connection failed:", err.message || err)
        })
    }
    sharedRedis = globalThis.__sharedRedisClient
    return sharedRedis
}

/**
 * Collect matching Redis keys using scanStream (non-blocking).
 * Shared utility to avoid repeating the SCAN + collect pattern.
 */
async function scanAndCollectKeys(redisClient, patterns) {
    const allKeys = []
    for (const pattern of patterns) {
        const keys = []
        const stream = redisClient.scanStream({ match: pattern, count: 200 })
        await new Promise((resolve, reject) => {
            stream.on("data", (batch) => keys.push(...batch))
            stream.on("end", resolve)
            stream.on("error", reject)
        })
        allKeys.push(...keys)
    }
    return allKeys
}

/**
 * Fully clear all queue data for a server in Redis and DB.
 * Mirrors the same function in fivem-bridge.mjs — used by the heartbeat
 * watchdog to wipe queue/session state when a server goes offline.
 */
async function fullClearServerData(serverId) {
    const redis = await getSharedRedis()
    if (redis) {
        try {
            const keysToDelete = [`queue:zset:${serverId}`, `queue:lock:${serverId}`]
            const scannedKeys = await scanAndCollectKeys(redis, [
                `queue:entry:${serverId}:*`,
                `queue:steam:${serverId}:*`,
                `queue:simulation:${serverId}`,
            ])
            keysToDelete.push(...scannedKeys)
            if (keysToDelete.length > 0) {
                const pipeline = redis.pipeline()
                for (const key of keysToDelete) pipeline.del(key)
                await pipeline.exec()
            }
            // Invalidate reconnect priority
            await redis.set(
                `server:restart:${serverId}`,
                Date.now().toString(),
                "EX",
                2 * 60 * 60
            )
        } catch (e) {
            console.error(`[heartbeat-watchdog] Redis cleanup failed for ${serverId}:`, e.message || e)
        }
    }

    const prisma = await getSharedPrisma()
    if (prisma) {
        await prisma.queueEntry.deleteMany({ where: { serverId } })
        await prisma.playerSession.updateMany({
            where: { serverId, disconnectedAt: null },
            data: { disconnectedAt: new Date() },
        })
    }
}

// Redis pub/sub channel — same as lib/queue-events.ts and fivem-bridge.mjs
const PUBSUB_CHANNEL = "softora:queue:events"

/**
 * Emit an enriched queue event via Redis pub/sub.
 * The queue-events.ts subscriber picks this up and forwards to browser Socket.IO clients.
 */
async function emitEnrichedEvent(serverId, type, extraData = {}) {
    const redis = await getSharedRedis()
    const prisma = await getSharedPrisma()

    let playerCount = 0
    let queueCount = 0
    try {
        ;[playerCount, queueCount] = await Promise.all([
            prisma.playerSession.count({ where: { serverId, disconnectedAt: null } }),
            redis.zcard(`queue:zset:${serverId}`),
        ])
    } catch { /* use defaults */ }

    const status = extraData.status || undefined

    const event = {
        type,
        serverId,
        ...(Object.keys(extraData).length > 0 ? { data: extraData } : {}),
        payload: { playerCount, queueCount, ...(status ? { status } : {}) },
    }

    try {
        await redis.publish(PUBSUB_CHANNEL, JSON.stringify(event))
    } catch (e) {
        console.error(`[server] Failed to publish event:`, e.message || e)
    }
}

httpServer.listen(port, hostname, () => {
    console.log(`> Ready on http://${hostname}:${port}`)
})

// --- Discord Gateway Bot ---
// Listens for guildMemberUpdate events to instantly update queue priorities
// when a user's Discord roles change, without waiting for their next API call.
if (!globalThis.__discordGatewayStarted) {
    globalThis.__discordGatewayStarted = true

    const discordBotToken = process.env.DISCORD_BOT_TOKEN
    const discordGuildId = process.env.DISCORD_GUILD_ID

    if (discordBotToken && discordGuildId) {
        // Import all required modules upfront so they're available throughout the handler
        Promise.all([
            import("discord.js"),
            import("./lib/discord.ts"),
            import("./lib/queue.ts"),
        ]).then(([
            { Client, GatewayIntentBits },
            { DISCORD_PLACEHOLDER_GUILD_ID },
            { determinePriority, generateQueueScore },
        ]) => {
            if (discordGuildId === DISCORD_PLACEHOLDER_GUILD_ID) {
                console.log("[discord-gateway] Skipped — DISCORD_GUILD_ID is the placeholder value")
                return
            }

            const discordClient = new Client({
                intents: [GatewayIntentBits.Guilds, GatewayIntentBits.GuildMembers],
            })

            // Expose client globally so fetchDiscordRoles() in lib/discord.ts can use the cache
            globalThis.__discordClient = discordClient

            discordClient.on("guildMemberUpdate", async (oldMember, newMember) => {
                try {
                    // Ignore updates from guilds other than the configured one
                    if (newMember.guild.id !== discordGuildId) return

                    // Skip if roles haven't changed
                    const oldRoleIds = new Set(oldMember.roles.cache.keys())
                    const newRoleIds = new Set(newMember.roles.cache.keys())
                    const hasChanged =
                        oldRoleIds.size !== newRoleIds.size ||
                        [...newRoleIds].some((id) => !oldRoleIds.has(id))
                    if (!hasChanged) return

                    const discordId = newMember.id
                    const currentRoles = [...newMember.roles.cache.keys()]

                    const prisma = await getSharedPrisma()
                    const redis = await getSharedRedis()

                    // Find user in DB by their Discord ID
                    const user = await prisma.user.findFirst({ where: { discordId } })
                    if (!user) return

                    const userId = user.id

                    // Scan Redis for all servers where this user has an active queue entry,
                    // instead of loading all servers from the DB.
                    const entryKeys = await scanAndCollectKeys(redis, [`queue:entry:*:${userId}`])

                    for (const entryKey of entryKeys) {
                        // Key pattern: queue:entry:{serverId}:{userId} — destructure to extract serverId
                        const [, , serverId] = entryKey.split(":")
                        if (!serverId) continue

                        const raw = await redis.get(entryKey)
                        if (!raw) continue

                        let entry
                        try {
                            entry = JSON.parse(raw)
                        } catch {
                            continue
                        }

                        if (entry.status !== "waiting") continue

                        // Recalculate priority via determinePriority() to include rejoin-priority rules
                        const priorityInfo = await determinePriority(userId, serverId, currentRoles)
                        const newPriority = priorityInfo.priority
                        const newQueueName = priorityInfo.queueName
                        // Use the joinedAt from priorityInfo if provided (e.g. rejoin priority uses a
                        // backdated timestamp to place the user at the front of the queue)
                        const scoreJoinedAt = priorityInfo.joinedAt || new Date(entry.joinedAt)
                        const newScore = generateQueueScore(newPriority, scoreJoinedAt)

                        if (newPriority === entry.priority) continue

                        // Update the Redis entry with the new priority, preserving TTL
                        const updatedEntry = { ...entry, priority: newPriority, queueName: newQueueName }

                        const pipeline = redis.pipeline()
                        pipeline.set(entryKey, JSON.stringify(updatedEntry), "KEEPTTL")
                        pipeline.zadd(`queue:zset:${serverId}`, newScore, userId)
                        await pipeline.exec()

                        // Notify all clients of the queue update
                        await emitEnrichedEvent(serverId, "queue:update")

                        // Notify the specific user so their dashboard updates instantly
                        if (globalThis.__socketIO) {
                            globalThis.__socketIO.to(`user:${userId}`).emit("queue:priority-update", {
                                serverId,
                                priority: newPriority,
                                queueName: newQueueName,
                            })
                        }

                        console.log(
                            `[discord-gateway] Updated queue priority for user ${userId} in server ${serverId}: ${entry.priority} → ${newPriority} (${newQueueName})`
                        )
                    }

                    // Invalidate roleSyncCache so the next auth() call re-fetches admin/mod roles
                    if (globalThis.__roleSyncCache) {
                        globalThis.__roleSyncCache.delete(discordId)
                    }
                } catch (err) {
                    console.error("[discord-gateway] guildMemberUpdate error:", err.message || err)
                }
            })

            discordClient.once("ready", () => {
                console.log(`[discord-gateway] Bot ready — logged in as ${discordClient.user?.tag}`)
            })

            discordClient.login(discordBotToken).catch((err) => {
                console.error("[discord-gateway] Login failed:", err.message || err)
            })
        }).catch((err) => {
            console.error("[discord-gateway] Failed to import modules:", err.message || err)
        })
    } else {
        console.log("[discord-gateway] Skipped — DISCORD_BOT_TOKEN or DISCORD_GUILD_ID not configured")
    }
}

// --- Heartbeat Watchdog ---
// Uses per-server timers for instant offline detection.
// Each heartbeat resets a timer; when it fires (stale threshold exceeded)
// the server is immediately marked offline and clients are notified via
// Socket.IO — no polling delay at all.
// A slow background sweep runs every 60s as a safety net to catch any
// servers whose timer was lost (e.g. after a hot-reload).
if (!globalThis.__heartbeatWatchdogStarted) {
    globalThis.__heartbeatWatchdogStarted = true

    const STALE_THRESHOLD_MS = 12_000 // 12s — ~2× the 5s heartbeat interval

    // Initialise timers for all servers that are currently online/restarting
    getSharedPrisma().then(async (prisma) => {
        try {
            const activeServers = await prisma.server.findMany({
                where: { status: { in: ["online", "restarting"] } },
                select: { id: true, name: true, lastHeartbeatAt: true },
            })
            for (const s of activeServers) {
                const age = s.lastHeartbeatAt ? Date.now() - s.lastHeartbeatAt.getTime() : Infinity
                const remaining = Math.max(0, STALE_THRESHOLD_MS - age)
                scheduleStaleTimer(s.id, s.name, remaining)
            }
            if (activeServers.length > 0) {
                console.log(`[heartbeat-watchdog] Initialised timers for ${activeServers.length} active server(s)`)
            }
        } catch (e) {
            console.error("[heartbeat-watchdog] Failed to initialise server timers:", e.message || e)
        }
    }).catch((err) => {
        console.error("[heartbeat-watchdog] Failed to init Prisma:", err)
    })

    /**
     * Fully clear all queue data for a server in Redis and DB.
     * Delegates to the shared fullClearServerData() function.
     */
    async function watchdogClearServer(serverId) {
        await fullClearServerData(serverId)
    }

    /**
     * Mark a server offline, clear its queue, and notify all web clients.
     */
    async function markServerOffline(serverId, serverName) {
        const watchdogPrisma = await getSharedPrisma()
        if (!watchdogPrisma) return

        // Double-check current status to avoid race conditions
        const current = await watchdogPrisma.server.findUnique({
            where: { id: serverId },
            select: { status: true },
        })
        if (!current || (current.status !== "online" && current.status !== "restarting")) return

        console.log(`[heartbeat-watchdog] Server "${serverName}" (${serverId}) heartbeat stale — marking offline, clearing queue`)

        await watchdogPrisma.server.update({
            where: { id: serverId },
            data: { status: "offline" },
        })

        // Full queue + session + reconnect priority wipe
        await watchdogClearServer(serverId)

        // Notify web clients via Socket.IO instantly with enriched payload
        if (globalThis.__socketIO) {
            globalThis.__socketIO.emit("queue:event", {
                type: "server:status",
                serverId,
                data: { status: "offline" },
                payload: { playerCount: 0, queueCount: 0, status: "offline" },
            })
            globalThis.__socketIO.emit("queue:event", {
                type: "queue:update",
                serverId,
                payload: { playerCount: 0, queueCount: 0, status: "offline" },
            })
        }
    }

    // --- Per-server timer management ---
    if (!globalThis.__heartbeatTimers) {
        globalThis.__heartbeatTimers = {}
    }

    function scheduleStaleTimer(serverId, serverName, delayMs) {
        // Clear any existing timer for this server
        if (globalThis.__heartbeatTimers[serverId]) {
            clearTimeout(globalThis.__heartbeatTimers[serverId])
        }
        globalThis.__heartbeatTimers[serverId] = setTimeout(() => {
            delete globalThis.__heartbeatTimers[serverId]
            markServerOffline(serverId, serverName).catch((e) => {
                console.error(`[heartbeat-watchdog] Timer error for ${serverId}:`, e.message || e)
            })
        }, delayMs)
    }

    // Expose timer functions globally so the heartbeat & status API routes can call them
    globalThis.__resetHeartbeatTimer = function (serverId, serverName) {
        scheduleStaleTimer(serverId, serverName, STALE_THRESHOLD_MS)
    }

    globalThis.__clearHeartbeatTimer = function (serverId) {
        if (globalThis.__heartbeatTimers[serverId]) {
            clearTimeout(globalThis.__heartbeatTimers[serverId])
            delete globalThis.__heartbeatTimers[serverId]
        }
    }

    console.log("[heartbeat-watchdog] Watchdog started (per-server timers, 60s safety sweep)")

    // Safety-net sweep: catches servers whose timer was somehow lost
    setInterval(async () => {
        const sweepPrisma = await getSharedPrisma().catch(() => null)
        if (!sweepPrisma) return

        try {
            const cutoff = new Date(Date.now() - STALE_THRESHOLD_MS)

            const staleServers = await sweepPrisma.server.findMany({
                where: {
                    status: { in: ["online", "restarting"] },
                    OR: [
                        { lastHeartbeatAt: null },
                        { lastHeartbeatAt: { lt: cutoff } },
                    ],
                },
                select: { id: true, name: true },
            })

            for (const server of staleServers) {
                // Only act if no active timer exists (timer already handles normal case)
                if (!globalThis.__heartbeatTimers[server.id]) {
                    await markServerOffline(server.id, server.name)
                }
            }
        } catch (e) {
            console.error("[heartbeat-watchdog] Sweep error:", e.message || e)
        }
    }, 60_000) // Safety sweep every 60s
}

// --- Queue Cleanup Worker ---
// Periodically cleans up expired queue entries (offers/accepts that timed out).
// Runs every 10s. Previously this work was done inside /api/queue/status on
// every poll, which caused massive latency for read-only status checks.
if (!globalThis.__queueCleanupWorkerStarted) {
    globalThis.__queueCleanupWorkerStarted = true

    console.log("[queue-cleanup] Cleanup worker started (10s interval)")

    setInterval(async () => {
        const cleanupPrisma = await getSharedPrisma().catch(() => null)
        if (!cleanupPrisma) return

        try {
            const { cleanupExpiredEntries } = await import("./lib/queue.ts")

            // Find all online servers and clean up their expired queue entries
            const servers = await cleanupPrisma.server.findMany({
                where: { status: "online" },
                select: { id: true },
            })

            // Use allSettled so a failure for one server doesn't skip the rest
            const results = await Promise.allSettled(
                servers.map(s => cleanupExpiredEntries(s.id)),
            )

            results.forEach((result, index) => {
                if (result.status === "rejected") {
                    const serverId = servers[index]?.id
                    console.error(
                        `[queue-cleanup] Error cleaning server ${serverId}:`,
                        (result.reason && result.reason.message) || result.reason,
                    )
                }
            })
        } catch (e) {
            console.error("[queue-cleanup] Error:", e.message || e)
        }
    }, 10_000)
}

// --- Queue Simulation Worker ---
if (!globalThis.__queueSimWorkerStarted) {
    globalThis.__queueSimWorkerStarted = true

    console.log("[queue-sim] Simulation worker started in background")

    setInterval(async () => {
        const simRedis = await getSharedRedis().catch(() => null)
        const prismaClient = await getSharedPrisma().catch(() => null)
        if (!simRedis || !prismaClient) return
        try {
            // Find all active simulation flags
            const keys = await scanAndCollectKeys(simRedis, ["queue:simulation:*"])

            if (keys.length === 0) return

            for (const key of keys) {
                const val = await simRedis.get(key)
                if (val !== "true") continue

                const serverId = key.split(":")[2]
                if (!serverId) continue

                // Auto-inject 1 user
                console.log(`[queue-sim] Injecting 1 test user for server ${serverId}`)

                // We recreate the inject logic here to avoid importing Next.js Server Actions outside of Next.js context
                const fakeId = `usr_${randomUUID()}`
                const fakeName = `AutoTest ${Math.floor(Math.random() * 9000 + 1000)}`
                const fakeDiscord = `999${Math.floor(Math.random() * 1000000)}`
                const fakeSteam = `1100001auto${Math.floor(Math.random() * 8999 + 1000)}`

                await prismaClient.user.upsert({
                    where: { steamHex: fakeSteam },
                    update: {},
                    create: {
                        id: fakeId,
                        name: fakeName,
                        discordId: fakeDiscord,
                        steamHex: fakeSteam
                    }
                })

                const now = Date.now()
                const priority = Math.floor(Math.random() * 3)
                const score = (Math.max(0, Math.min(999, priority))) * 1e12 + now
                const joinedAtStr = new Date(now).toISOString()
                const entryKey = `queue:entry:${serverId}:${fakeId}`

                const newEntry = {
                    id: randomUUID(),
                    userId: fakeId,
                    serverId,
                    status: "waiting",
                    priority,
                    queueName: `Auto Tier ${priority}`,
                    joinedAt: joinedAtStr,
                    steamHex: fakeSteam
                }

                const pipeline = simRedis.pipeline()
                pipeline.set(entryKey, JSON.stringify(newEntry), "EX", 30 * 60)
                pipeline.zadd(`queue:zset:${serverId}`, score, fakeId)
                pipeline.set(`queue:steam:${serverId}:${fakeSteam}`, fakeId)
                await pipeline.exec()

                // Notify websockets with enriched payload
                if (globalThis.__socketIO) {
                    const freshQueueCount = await simRedis.zcard(`queue:zset:${serverId}`)
                    const event = {
                        type: "queue:update",
                        serverId,
                        payload: { queueCount: freshQueueCount },
                    }
                    globalThis.__socketIO.emit("queue:event", event)
                }
            }
        } catch (e) {
            console.error("[queue-sim] Worker error:", e)
        }
    }, 15000) // Run every 15 seconds
}