Phase2

apps/worker/src/handlers/server-lifecycle.ts

@@ -0,0 +1,128 @@
+import { randomUUID } from 'node:crypto';
+
+import { Job } from 'bullmq';
+import { z } from 'zod';
+
+import { prisma } from '@hexahost/database';
+
+import { logger } from '../logger';
+import { publishNodeCommand } from '../redis';
+import { transitionServerStatus } from '../server-state';
+
+const lifecycleJobSchema = z.object({
+  serverId: z.string().uuid(),
+});
+
+async function loadServerForLifecycle(serverId: string) {
+  const server = await prisma.gameServer.findUnique({
+    where: { id: serverId },
+  });
+
+  if (!server) {
+    throw new Error(`Server not found: ${serverId}`);
+  }
+
+  if (!server.nodeId) {
+    throw new Error(`Server ${serverId} is not assigned to a node`);
+  }
+
+  return server;
+}
+
+async function publishLifecycleCommand(
+  serverId: string,
+  nodeId: string,
+  type: 'server.start' | 'server.stop',
+  generation: number,
+): Promise<string> {
+  const messageId = randomUUID();
+
+  await publishNodeCommand({
+    nodeId,
+    envelope: {
+      protocolVersion: 1,
+      messageId,
+      type,
+      timestamp: new Date().toISOString(),
+      payload: {
+        serverId,
+        generation,
+        desiredGeneration: generation,
+      },
+    },
+  });
+
+  return messageId;
+}
+
+export async function processStartServerJob(job: Job): Promise<{ status: string }> {
+  if (job.name !== 'start-server') {
+    logger.warn({ jobName: job.name }, 'Unknown lifecycle job');
+    return { status: 'ignored' };
+  }
+
+  const { serverId } = lifecycleJobSchema.parse(job.data);
+  const server = await loadServerForLifecycle(serverId);
+
+  if (server.status !== 'STOPPED') {
+    throw new Error(`Cannot start server ${serverId} from status ${server.status}`);
+  }
+
+  const messageId = await publishLifecycleCommand(
+    serverId,
+    server.nodeId!,
+    'server.start',
+    server.version,
+  );
+
+  await transitionServerStatus(serverId, 'STARTING', {
+    expectedFrom: 'STOPPED',
+    reason: 'Start requested',
+    correlationId: messageId,
+  });
+
+  logger.info({ serverId, messageId }, 'Server start command published');
+  return { status: 'starting' };
+}
+
+export async function processStopServerJob(job: Job): Promise<{ status: string }> {
+  if (job.name !== 'stop-server') {
+    logger.warn({ jobName: job.name }, 'Unknown lifecycle job');
+    return { status: 'ignored' };
+  }
+
+  const { serverId } = lifecycleJobSchema.parse(job.data);
+  const server = await loadServerForLifecycle(serverId);
+
+  if (server.status !== 'RUNNING') {
+    throw new Error(`Cannot stop server ${serverId} from status ${server.status}`);
+  }
+
+  const messageId = await publishLifecycleCommand(
+    serverId,
+    server.nodeId!,
+    'server.stop',
+    server.version,
+  );
+
+  await transitionServerStatus(serverId, 'STOPPING', {
+    expectedFrom: 'RUNNING',
+    reason: 'Stop requested',
+    correlationId: messageId,
+  });
+
+  logger.info({ serverId, messageId }, 'Server stop command published');
+  return { status: 'stopping' };
+}
+
+export async function processLifecycleJob(job: Job): Promise<{ status: string }> {
+  switch (job.name) {
+    case 'start-server':
+      return processStartServerJob(job);
+    case 'stop-server':
+      return processStopServerJob(job);
+    default:
+      logger.warn({ jobName: job.name }, 'Unknown lifecycle job name');
+      return { status: 'ignored' };
+  }
+}

apps/worker/src/handlers/server-provisioning.ts

@@ -0,0 +1,169 @@
+import { randomUUID } from 'node:crypto';
+
+import { prisma } from '@hexahost/database';
+import { Job } from 'bullmq';
+import { z } from 'zod';
+
+import { logger } from '../logger';
+import { publishNodeCommand, waitForNodeResponse } from '../redis';
+import { transitionServerStatus } from '../server-state';
+
+const BASE_HOST_PORT = 25565;
+const PROVISION_TIMEOUT_MS = 5 * 60 * 1000;
+
+export const provisionServerJobSchema = z.object({
+  serverId: z.string().uuid(),
+});
+
+function buildDataPath(serverId: string): string {
+  return `/var/lib/hgc/servers/${serverId}`;
+}
+
+async function findAvailableHostPort(nodeId: string): Promise<number> {
+  const [allocations, servers] = await Promise.all([
+    prisma.gameServerAllocation.findMany({
+      where: { nodeId, status: 'ACTIVE' },
+      select: { hostPort: true },
+    }),
+    prisma.gameServer.findMany({
+      where: { nodeId, hostPort: { not: null } },
+      select: { hostPort: true },
+    }),
+  ]);
+
+  const usedPorts = new Set<number>();
+  for (const allocation of allocations) {
+    usedPorts.add(allocation.hostPort);
+  }
+  for (const server of servers) {
+    if (server.hostPort !== null) {
+      usedPorts.add(server.hostPort);
+    }
+  }
+
+  let port = BASE_HOST_PORT;
+  while (usedPorts.has(port)) {
+    port += 1;
+  }
+
+  return port;
+}
+
+export async function processProvisionServerJob(job: Job): Promise<{ status: string }> {
+  if (job.name !== 'provision-server') {
+    logger.warn({ jobName: job.name }, 'Unknown provisioning job');
+    return { status: 'ignored' };
+  }
+
+  const { serverId } = provisionServerJobSchema.parse(job.data);
+  const server = await prisma.gameServer.findUnique({
+    where: { id: serverId },
+    include: { plan: true },
+  });
+
+  if (!server) {
+    throw new Error(`Server not found: ${serverId}`);
+  }
+
+  const node = await prisma.gameNode.findFirst({
+    where: { status: 'ONLINE' },
+    orderBy: { createdAt: 'asc' },
+  });
+
+  if (!node) {
+    await transitionServerStatus(serverId, 'ERROR', {
+      reason: 'No online game node available',
+      errorCode: 'NO_NODE_AVAILABLE',
+      errorMessage: 'No online game node is available for provisioning',
+    });
+    throw new Error('No online game node available');
+  }
+
+  const hostPort = await findAvailableHostPort(node.id);
+  const ramMb = server.plan?.maxRamMb ?? server.ramMb;
+  const dataPath = buildDataPath(serverId);
+  const messageId = randomUUID();
+
+  await prisma.$transaction(async (tx) => {
+    await tx.gameServerAllocation.create({
+      data: {
+        serverId,
+        nodeId: node.id,
+        hostPort,
+        ramMbReserved: ramMb,
+        status: 'ACTIVE',
+      },
+    });
+
+    await tx.gameServer.update({
+      where: { id: serverId },
+      data: {
+        nodeId: node.id,
+        hostPort,
+        ramMb,
+        dataPath,
+        status: 'PROVISIONING',
+      },
+    });
+
+    await tx.gameServerStateTransition.create({
+      data: {
+        gameServerId: serverId,
+        fromStatus: server.status,
+        toStatus: 'PROVISIONING',
+        reason: 'Worker allocated node resources',
+        correlationId: messageId,
+      },
+    });
+  });
+
+  await publishNodeCommand({
+    nodeId: node.id,
+    envelope: {
+      protocolVersion: 1,
+      messageId,
+      type: 'server.provision',
+      timestamp: new Date().toISOString(),
+      payload: {
+        serverId,
+        generation: server.version,
+        edition: server.edition,
+        softwareFamily: server.softwareFamily,
+        minecraftVersion: server.minecraftVersion,
+        hostPort,
+        ramMb,
+        dataPath,
+        eulaAccepted: server.eulaAccepted,
+      },
+    },
+  });
+
+  const response = await waitForNodeResponse(messageId, PROVISION_TIMEOUT_MS);
+
+  if (response === null) {
+    await transitionServerStatus(serverId, 'INSTALLING', {
+      reason: 'Provision command sent; awaiting agent completion',
+      correlationId: messageId,
+    });
+    logger.info({ serverId, messageId }, 'Provision timed out; marked INSTALLING');
+    return { status: 'installing' };
+  }
+
+  if (!response.success) {
+    await transitionServerStatus(serverId, 'ERROR', {
+      reason: 'Provision failed',
+      correlationId: messageId,
+      errorCode: response.errorCode ?? response.resultCode ?? 'PROVISION_FAILED',
+      errorMessage: response.errorMessage ?? 'Server provisioning failed',
+    });
+    throw new Error(response.errorMessage ?? 'Server provisioning failed');
+  }
+
+  await transitionServerStatus(serverId, 'STOPPED', {
+    reason: 'Provision completed',
+    correlationId: messageId,
+  });
+
+  logger.info({ serverId, nodeId: node.id, hostPort }, 'Server provisioned');
+  return { status: 'stopped' };
+}

apps/worker/src/index.ts

@@ -4,9 +4,18 @@ import { validateConfig } from '@hexahost/config';
 import { prisma } from '@hexahost/database';
 
 import { processNotificationJob } from './handlers/notifications';
+import { processLifecycleJob } from './handlers/server-lifecycle';
+import { processProvisionServerJob } from './handlers/server-provisioning';
 import { startHealthServer } from './health';
 import { logger } from './logger';
-import { WORKER_QUEUES, type WorkerQueueName } from './queues';
+import { closeRedisConnections } from './redis';
+import {
+  QUEUE_NOTIFICATIONS,
+  QUEUE_SERVER_LIFECYCLE,
+  QUEUE_SERVER_PROVISIONING,
+  WORKER_QUEUES,
+  type WorkerQueueName,
+} from './queues';
 
 function createQueueWorker(
   queueName: WorkerQueueName,
@@ -20,11 +29,19 @@ function createQueueWorker(
         'Processing job',
       );
 
-      if (queueName === 'notifications') {
+      if (queueName === QUEUE_NOTIFICATIONS) {
         return processNotificationJob(job);
       }
 
-      return { acknowledged: true, queue: queueName };
+      if (queueName === QUEUE_SERVER_PROVISIONING) {
+        return processProvisionServerJob(job);
+      }
+
+      if (queueName === QUEUE_SERVER_LIFECYCLE) {
+        return processLifecycleJob(job);
+      }
+
+      logger.warn({ queue: queueName, jobName: job.name }, 'Unhandled job');
     },
     { connection },
   );
@@ -62,6 +79,7 @@ async function bootstrap(): Promise<void> {
     logger.info({ signal }, 'Shutting down worker');
 
     await Promise.all(workers.map((worker) => worker.close()));
+    await closeRedisConnections();
     await prisma.$disconnect();
 
     healthServer.close(() => {

apps/worker/src/redis.ts

@@ -0,0 +1,109 @@
+import Redis from 'ioredis';
+
+import { validateConfig } from '@hexahost/config';
+
+import { logger } from './logger';
+
+export const NODE_COMMANDS_CHANNEL = 'hgc:node:commands';
+
+export interface NodeCommandEnvelope {
+  protocolVersion: number;
+  messageId: string;
+  type: string;
+  timestamp: string;
+  payload: Record<string, unknown>;
+}
+
+export interface NodeCommandMessage {
+  nodeId: string;
+  envelope: NodeCommandEnvelope;
+}
+
+export interface NodeResponseMessage {
+  success: boolean;
+  resultCode?: string;
+  errorCode?: string;
+  errorMessage?: string;
+}
+
+let publisher: Redis | null = null;
+
+export function getRedisPublisher(): Redis {
+  if (!publisher) {
+    const config = validateConfig();
+    publisher = new Redis(config.REDIS_URL, {
+      maxRetriesPerRequest: null,
+    });
+  }
+
+  return publisher;
+}
+
+export async function publishNodeCommand(message: NodeCommandMessage): Promise<void> {
+  const redis = getRedisPublisher();
+  await redis.publish(NODE_COMMANDS_CHANNEL, JSON.stringify(message));
+  logger.info(
+    { nodeId: message.nodeId, type: message.envelope.type, messageId: message.envelope.messageId },
+    'Published node command',
+  );
+}
+
+export function nodeResponseChannel(messageId: string): string {
+  return `hgc:node:responses:${messageId}`;
+}
+
+export async function waitForNodeResponse(
+  messageId: string,
+  timeoutMs: number,
+): Promise<NodeResponseMessage | null> {
+  const config = validateConfig();
+  const subscriber = new Redis(config.REDIS_URL, {
+    maxRetriesPerRequest: null,
+  });
+  const channel = nodeResponseChannel(messageId);
+
+  return new Promise((resolve) => {
+    let settled = false;
+
+    const finish = (value: NodeResponseMessage | null): void => {
+      if (settled) {
+        return;
+      }
+      settled = true;
+      clearTimeout(timer);
+      void subscriber.unsubscribe(channel).finally(() => {
+        subscriber.disconnect();
+      });
+      resolve(value);
+    };
+
+    const timer = setTimeout(() => finish(null), timeoutMs);
+
+    subscriber.on('message', (receivedChannel, message) => {
+      if (receivedChannel !== channel) {
+        return;
+      }
+
+      try {
+        finish(JSON.parse(message) as NodeResponseMessage);
+      } catch {
+        finish({
+          success: false,
+          errorMessage: 'Invalid node response payload',
+        });
+      }
+    });
+
+    void subscriber.subscribe(channel).catch((error: Error) => {
+      logger.error({ err: error, channel }, 'Failed to subscribe for node response');
+      finish(null);
+    });
+  });
+}
+
+export async function closeRedisConnections(): Promise<void> {
+  if (publisher) {
+    await publisher.quit();
+    publisher = null;
+  }
+}

apps/worker/src/server-state.ts

@@ -0,0 +1,58 @@
+import { type Prisma, prisma } from '@hexahost/database';
+
+type GameServerStatus = Prisma.GameServerGetPayload<{ select: { status: true } }>['status'];
+
+interface TransitionOptions {
+  expectedFrom?: GameServerStatus | GameServerStatus[];
+  reason?: string;
+  correlationId?: string;
+  actorId?: string;
+  errorCode?: string;
+  errorMessage?: string;
+  extraData?: Prisma.GameServerUpdateInput;
+}
+
+export async function transitionServerStatus(
+  serverId: string,
+  toStatus: GameServerStatus,
+  options: TransitionOptions = {},
+): Promise<void> {
+  await prisma.$transaction(async (tx) => {
+    const server = await tx.gameServer.findUniqueOrThrow({
+      where: { id: serverId },
+    });
+
+    if (options.expectedFrom) {
+      const allowed = Array.isArray(options.expectedFrom)
+        ? options.expectedFrom
+        : [options.expectedFrom];
+
+      if (!allowed.includes(server.status)) {
+        throw new Error(
+          `Invalid state transition for server ${serverId}: ${server.status} -> ${toStatus}`,
+        );
+      }
+    }
+
+    await tx.gameServer.update({
+      where: { id: serverId },
+      data: {
+        status: toStatus,
+        ...options.extraData,
+      },
+    });
+
+    await tx.gameServerStateTransition.create({
+      data: {
+        gameServerId: serverId,
+        fromStatus: server.status,
+        toStatus,
+        reason: options.reason,
+        correlationId: options.correlationId,
+        actorId: options.actorId,
+        errorCode: options.errorCode,
+        errorMessage: options.errorMessage,
+      },
+    });
+  });
+}