CubeAdmin/lib/minecraft/process.ts

import { EventEmitter } from "node:events"
import { db } from "@/lib/db/index"
import { serverSettings } from "@/lib/db/schema"
import { rconClient } from "@/lib/minecraft/rcon"

// Maximum number of output lines kept in the ring buffer
const RING_BUFFER_SIZE = 500

export interface ServerStatus {
  running: boolean
  pid?: number
  uptime?: number
  startedAt?: Date
}

type OutputCallback = (line: string) => void

export class McProcessManager extends EventEmitter {
  private process: ReturnType<typeof Bun.spawn> | null = null
  private startedAt: Date | null = null
  private outputBuffer: string[] = []
  private outputCallbacks: Set<OutputCallback> = new Set()
  private restartOnCrash = false
  private isIntentionalStop = false
  private stdoutReader: Promise<void> | null = null

  // -------------------------------------------------------------------------
  // Public API
  // -------------------------------------------------------------------------

  /**
   * Start the Minecraft server process.
   * Reads java command configuration from the `server_settings` table.
   */
  async start(): Promise<void> {
    if (this.process !== null) {
      throw new Error("Server is already running")
    }

    const settings = await this.loadSettings()
    const cmd = this.buildCommand(settings)

    console.log(`[MC] Starting server: ${cmd.join(" ")}`)

    this.isIntentionalStop = false
    this.restartOnCrash = settings.restartOnCrash ?? false

    this.process = Bun.spawn(cmd, {
      cwd: settings.minecraftPath ?? process.env.MC_SERVER_PATH ?? process.cwd(),
      stdout: "pipe",
      stderr: "pipe",
      stdin: "pipe",
    })

    this.startedAt = new Date()

    // Pipe stdout
    this.stdoutReader = this.readStream(
      this.process.stdout as ReadableStream<Uint8Array> | null ?? null,
      "stdout",
    )
    // Pipe stderr into the same output stream
    void this.readStream(
      this.process.stderr as ReadableStream<Uint8Array> | null ?? null,
      "stderr",
    )

    this.emit("started", { pid: this.process.pid })
    console.log(`[MC] Server started with PID ${this.process.pid}`)

    // Watch for exit
    void this.watchExit()
  }

  /**
   * Stop the Minecraft server.
   * @param force - if true, kills the process immediately; if false, sends the
   *                RCON `stop` command and waits for graceful shutdown.
   */
  async stop(force = false): Promise<void> {
    if (this.process === null) {
      throw new Error("Server is not running")
    }

    this.isIntentionalStop = true

    if (force) {
      console.log("[MC] Force-killing server process")
      this.process.kill()
    } else {
      console.log("[MC] Sending RCON stop command")
      try {
        await rconClient.sendCommand("stop")
      } catch (err) {
        console.warn("[MC] RCON stop failed, killing process:", err)
        this.process.kill()
      }
    }

    // Wait up to 30 s for the process to exit
    await Promise.race([
      this.process.exited,
      new Promise<void>((_, reject) =>
        setTimeout(() => reject(new Error("Server did not stop in 30 s")), 30_000),
      ),
    ])
  }

  /**
   * Restart the Minecraft server.
   * @param force - passed through to `stop()`
   */
  async restart(force = false): Promise<void> {
    if (this.process !== null) {
      await this.stop(force)
    }
    await this.start()
  }

  /** Returns current process status */
  getStatus(): ServerStatus {
    const running = this.process !== null
    if (!running) return { running: false }

    return {
      running: true,
      pid: this.process!.pid,
      startedAt: this.startedAt ?? undefined,
      uptime: this.startedAt
        ? Math.floor((Date.now() - this.startedAt.getTime()) / 1000)
        : undefined,
    }
  }

  /** Returns the last RING_BUFFER_SIZE lines of console output */
  getOutput(): string[] {
    return [...this.outputBuffer]
  }

  /**
   * Register a callback that receives each new output line.
   * Returns an unsubscribe function.
   */
  onOutput(cb: OutputCallback): () => void {
    this.outputCallbacks.add(cb)
    return () => this.outputCallbacks.delete(cb)
  }

  /**
   * Write a raw string to the server's stdin (for when RCON is unavailable).
   */
  writeStdin(line: string): void {
    const stdin = this.process?.stdin
    if (!stdin) throw new Error("Server is not running")
    // Bun.spawn stdin is a FileSink (not a WritableStream)
    const fileSink = stdin as import("bun").FileSink
    fileSink.write(new TextEncoder().encode(line + "\n"))
    void fileSink.flush()
  }

  // -------------------------------------------------------------------------
  // Private helpers
  // -------------------------------------------------------------------------

  private async loadSettings() {
    const rows = await db.select().from(serverSettings).limit(1)
    const s = rows[0]
    if (!s) throw new Error("No server settings found in database")
    return s
  }

  private buildCommand(settings: Awaited<ReturnType<typeof this.loadSettings>>): string[] {
    const jarPath = settings.serverJar ?? "server.jar"
    const minRam = settings.minRam ?? 1024
    const maxRam = settings.maxRam ?? 4096
    const extraArgs: string[] = settings.javaArgs
      ? settings.javaArgs.split(/\s+/).filter(Boolean)
      : []

    return [
      "java",
      `-Xms${minRam}M`,
      `-Xmx${maxRam}M`,
      ...extraArgs,
      "-jar",
      jarPath,
      "--nogui",
    ]
  }

  private async readStream(
    stream: ReadableStream<Uint8Array> | null,
    _tag: string,
  ): Promise<void> {
    if (!stream) return
    const reader = stream.getReader()
    const decoder = new TextDecoder()
    let partial = ""

    try {
      while (true) {
        const { done, value } = await reader.read()
        if (done) break

        const chunk = decoder.decode(value, { stream: true })
        partial += chunk

        const lines = partial.split("\n")
        partial = lines.pop() ?? ""

        for (const line of lines) {
          this.pushLine(line)
        }
      }
      // Flush remaining partial content
      if (partial) this.pushLine(partial)
    } catch {
      // Stream closed - normal during shutdown
    } finally {
      reader.releaseLock()
    }
  }

  private pushLine(line: string): void {
    // Ring buffer
    this.outputBuffer.push(line)
    if (this.outputBuffer.length > RING_BUFFER_SIZE) {
      this.outputBuffer.shift()
    }

    this.emit("output", line)
    for (const cb of this.outputCallbacks) {
      try {
        cb(line)
      } catch {
        // Ignore callback errors
      }
    }
  }

  private async watchExit(): Promise<void> {
    if (!this.process) return
    const exitCode = await this.process.exited

    const wasRunning = this.process !== null
    this.process = null
    this.startedAt = null

    await rconClient.disconnect().catch(() => {})

    if (wasRunning) {
      this.emit("stopped", { exitCode })
      console.log(`[MC] Server stopped with exit code ${exitCode}`)

      if (!this.isIntentionalStop && this.restartOnCrash) {
        this.emit("crash", { exitCode })
        console.warn(`[MC] Server crashed (exit ${exitCode}), restarting in 5 s…`)
        await new Promise((resolve) => setTimeout(resolve, 5_000))
        try {
          await this.start()
        } catch (err) {
          console.error("[MC] Auto-restart failed:", err)
        }
      }
    }
  }
}

export const mcProcessManager = new McProcessManager()