onebox/ts/opsserver/handlers/logs.handler.ts

import * as plugins from '../../plugins.ts';
import { logger } from '../../logging.ts';
import type { OpsServer } from '../classes.opsserver.ts';
import * as interfaces from '../../../ts_interfaces/index.ts';
import { requireValidIdentity } from '../helpers/guards.ts';

export class LogsHandler {
  public typedrouter = new plugins.typedrequest.TypedRouter();

  constructor(private opsServerRef: OpsServer) {
    this.opsServerRef.typedrouter.addTypedRouter(this.typedrouter);
    this.registerHandlers();
  }

  private registerHandlers(): void {
    // Service log stream
    this.typedrouter.addTypedHandler(
      new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_GetServiceLogStream>(
        'getServiceLogStream',
        async (dataArg) => {
          await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);

          const service = this.opsServerRef.oneboxRef.database.getServiceByName(dataArg.serviceName);
          if (!service) {
            throw new plugins.typedrequest.TypedResponseError('Service not found');
          }

          const virtualStream = new plugins.typedrequest.VirtualStream<Uint8Array>();
          const encoder = new TextEncoder();

          // Get container and start streaming in background
          (async () => {
            try {
              let container = await this.opsServerRef.oneboxRef.docker.getContainerById(service.containerID!);
              if (!container) {
                // Try finding by service label
                const containers = await this.opsServerRef.oneboxRef.docker.listAllContainers();
                const serviceContainer = containers.find((c: any) => {
                  const labels = c.Labels || {};
                  return labels['com.docker.swarm.service.id'] === service.containerID;
                });
                if (serviceContainer) {
                  container = await this.opsServerRef.oneboxRef.docker.getContainerById(serviceContainer.Id);
                }
              }

              if (!container) {
                virtualStream.sendData(encoder.encode(JSON.stringify({ error: 'Container not found' })));
                return;
              }

              const logStream = await container.streamLogs({
                stdout: true,
                stderr: true,
                timestamps: true,
                tail: 100,
              });

              let buffer = new Uint8Array(0);

              logStream.on('data', (chunk: Uint8Array) => {
                // Append to buffer
                const newBuffer = new Uint8Array(buffer.length + chunk.length);
                newBuffer.set(buffer);
                newBuffer.set(chunk, buffer.length);
                buffer = newBuffer;

                // Process Docker multiplexed frames
                while (buffer.length >= 8) {
                  const frameSize = (buffer[4] << 24) | (buffer[5] << 16) | (buffer[6] << 8) | buffer[7];
                  if (buffer.length < 8 + frameSize) break;

                  const frameData = buffer.slice(8, 8 + frameSize);
                  try {
                    virtualStream.sendData(frameData);
                  } catch {
                    logStream.destroy();
                    return;
                  }
                  buffer = buffer.slice(8 + frameSize);
                }
              });

              logStream.on('error', (error: Error) => {
                logger.error(`Log stream error for ${dataArg.serviceName}: ${error.message}`);
              });
            } catch (error) {
              logger.error(`Failed to start log stream: ${error}`);
            }
          })();

          return { logStream: virtualStream as any };
        },
      ),
    );

    // Platform service log stream
    this.typedrouter.addTypedHandler(
      new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_GetPlatformServiceLogStream>(
        'getPlatformServiceLogStream',
        async (dataArg) => {
          await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);

          const platformService = this.opsServerRef.oneboxRef.database.getPlatformServiceByType(
            dataArg.serviceType,
          );
          if (!platformService || !platformService.containerId) {
            throw new plugins.typedrequest.TypedResponseError('Platform service has no container');
          }

          const virtualStream = new plugins.typedrequest.VirtualStream<Uint8Array>();

          (async () => {
            try {
              const container = await this.opsServerRef.oneboxRef.docker.getContainerById(
                platformService.containerId!,
              );
              if (!container) return;

              const logStream = await container.streamLogs({
                stdout: true,
                stderr: true,
                timestamps: true,
                tail: 100,
              });

              let buffer = new Uint8Array(0);

              logStream.on('data', (chunk: Uint8Array) => {
                const newBuffer = new Uint8Array(buffer.length + chunk.length);
                newBuffer.set(buffer);
                newBuffer.set(chunk, buffer.length);
                buffer = newBuffer;

                while (buffer.length >= 8) {
                  const frameSize = (buffer[4] << 24) | (buffer[5] << 16) | (buffer[6] << 8) | buffer[7];
                  if (buffer.length < 8 + frameSize) break;
                  const frameData = buffer.slice(8, 8 + frameSize);
                  try {
                    virtualStream.sendData(frameData);
                  } catch {
                    logStream.destroy();
                    return;
                  }
                  buffer = buffer.slice(8 + frameSize);
                }
              });
            } catch (error) {
              logger.error(`Failed to start platform log stream: ${error}`);
            }
          })();

          return { logStream: virtualStream as any };
        },
      ),
    );

    // Network log stream
    this.typedrouter.addTypedHandler(
      new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_GetNetworkLogStream>(
        'getNetworkLogStream',
        async (dataArg) => {
          await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);

          const virtualStream = new plugins.typedrequest.VirtualStream<Uint8Array>();
          const encoder = new TextEncoder();
          const clientId = crypto.randomUUID();

          // Create a mock WebSocket-like object for the CaddyLogReceiver
          const mockSocket = {
            readyState: 1, // WebSocket.OPEN
            send: (data: string) => {
              try {
                virtualStream.sendData(encoder.encode(data));
              } catch {
                this.opsServerRef.oneboxRef.caddyLogReceiver.removeClient(clientId);
              }
            },
          };

          const filter = dataArg.filter || {};
          this.opsServerRef.oneboxRef.caddyLogReceiver.addClient(
            clientId,
            mockSocket as any,
            filter,
          );

          return { logStream: virtualStream as any };
        },
      ),
    );

    // Event stream (general updates)
    this.typedrouter.addTypedHandler(
      new plugins.typedrequest.TypedHandler<interfaces.requests.IReq_GetEventStream>(
        'getEventStream',
        async (dataArg) => {
          await requireValidIdentity(this.opsServerRef.adminHandler, dataArg);

          const virtualStream = new plugins.typedrequest.VirtualStream<Uint8Array>();
          const encoder = new TextEncoder();

          // Send initial connection message
          virtualStream.sendData(
            encoder.encode(
              JSON.stringify({
                type: 'connected',
                message: 'Connected to Onebox event stream',
                timestamp: Date.now(),
              }),
            ),
          );

          return { eventStream: virtualStream as any };
        },
      ),
    );
  }
}