peardock/server/handlers/logs.js

/**
 * Container log streaming over protomux-rpc pushes.
 * Demuxes Docker multiplexed frames server-side so clients receive plain UTF-8.
 */
import { docker } from '../services/docker.js'
import { Pushes } from '../../shared/protocol.js'
import logger from '../utils/logger.js'

/**
 * Incremental demux of docker log multiplex protocol.
 * Frame: 1 byte stream type (0–2) + 3 zero bytes + 4 byte BE size + payload.
 */
function createServerDemuxer() {
  let pending = Buffer.alloc(0)
  /** Once we see a valid mux frame, stay in mux mode for the stream. */
  let muxMode = null // null | true | false

  return {
    /**
     * @param {Buffer} chunk
     * @returns {Buffer[]} payloads
     */
    push(chunk) {
      pending = Buffer.concat([pending, chunk])
      const out = []

      if (muxMode === false) {
        if (pending.length) {
          out.push(pending)
          pending = Buffer.alloc(0)
        }
        return out
      }

      // Detect mode from first full frame or non-mux text
      if (muxMode === null && pending.length >= 1) {
        const b0 = pending[0]
        if (b0 > 2) {
          muxMode = false
          out.push(pending)
          pending = Buffer.alloc(0)
          return out
        }
        // Need header to confirm
        if (pending.length >= 8) {
          const size = pending.readUInt32BE(4)
          // Plausible frame: type 0-2, zeros in 1-3, reasonable size
          const zerosOk = pending[1] === 0 && pending[2] === 0 && pending[3] === 0
          if (b0 <= 2 && zerosOk && size > 0 && size < 16 * 1024 * 1024) {
            muxMode = true
          } else if (b0 <= 2 && !zerosOk) {
            // Not mux — treat as raw (e.g. TTY)
            muxMode = false
            out.push(pending)
            pending = Buffer.alloc(0)
            return out
          }
        }
      }

      if (muxMode !== true) {
        // Wait for more bytes to decide, or emit if clearly not mux
        return out
      }

      while (pending.length >= 8) {
        const size = pending.readUInt32BE(4)
        if (size < 0 || size > 16 * 1024 * 1024) {
          // Corrupt header — fall back to raw remainder
          out.push(pending)
          pending = Buffer.alloc(0)
          muxMode = false
          break
        }
        if (pending.length < 8 + size) break
        out.push(pending.subarray(8, 8 + size))
        pending = pending.subarray(8 + size)
      }
      return out
    },
    flush() {
      if (!pending.length) return []
      const rest = pending
      pending = Buffer.alloc(0)
      return [rest]
    },
  }
}

async function startLogStream(session, streams, args) {
  const containerId = args.id || args.containerId
  if (!containerId) throw new Error('container id required')

  if (streams.has(containerId)) {
    try {
      streams.get(containerId).stream?.destroy?.()
    } catch {
      // ignore
    }
    streams.delete(containerId)
  }

  const logOpts = {
    stdout: args.stdout !== false,
    stderr: args.stderr !== false,
    tail: args.tail ?? 200,
    follow: args.follow !== false,
    timestamps: args.timestamps !== false,
  }
  if (args.since != null) logOpts.since = args.since
  if (args.until != null) logOpts.until = args.until

  const logsStream = await docker.getContainer(containerId).logs(logOpts)
  const demux = createServerDemuxer()

  streams.set(containerId, { stream: logsStream, demux })

  logsStream.on('data', (chunk) => {
    const buf = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)
    const payloads = demux.push(buf)
    for (const payload of payloads) {
      if (!payload.length) continue
      try {
        session.push(Pushes.logs, {
          type: 'logs',
          containerId,
          data: payload.toString('base64'),
          encoding: 'base64',
          format: 'utf8-demuxed',
        })
      } catch (e) {
        logger.debug('log push failed', { error: e.message })
      }
    }
  })
  logsStream.on('end', () => {
    for (const payload of demux.flush()) {
      if (!payload.length) continue
      try {
        session.push(Pushes.logs, {
          type: 'logs',
          containerId,
          data: payload.toString('base64'),
          encoding: 'base64',
          format: 'utf8-demuxed',
        })
      } catch {
        // ignore
      }
    }
    streams.delete(containerId)
  })
  logsStream.on('error', (err) => {
    logger.error('Log stream error', { containerId, error: err.message })
    session.push(Pushes.error, {
      error: `Log stream error: ${err.message}`,
      containerId,
    })
    streams.delete(containerId)
  })

  return { success: true, message: `Log stream started for ${containerId}` }
}

export function registerLogsHandlers(session) {
  /** @type {Map<string, { stream: import('stream').Readable, demux: ReturnType<typeof createServerDemuxer> }>} */
  const streams = new Map()
  session.state.set('logsStreams', streams)

  session._cleanupLogsForContainer = (containerId) => {
    for (const [key, entry] of streams.entries()) {
      if (key === containerId) {
        try {
          entry.stream?.destroy?.()
        } catch {
          // ignore
        }
        streams.delete(key)
      }
    }
  }

  session.respond('startLogs', (args) => startLogStream(session, streams, args))
  session.respond('logs', (args) => startLogStream(session, streams, args))

  session.respond('stopLogs', async (args) => {
    const containerId = args.id || args.containerId
    if (containerId && streams.has(containerId)) {
      try {
        streams.get(containerId).stream?.destroy?.()
      } catch {
        // ignore
      }
      streams.delete(containerId)
    }
    return { success: true }
  })
}

export function cleanupLogsOnClose(session) {
  const streams = session.state.get('logsStreams')
  if (!streams) return
  for (const entry of streams.values()) {
    try {
      entry.stream?.destroy?.()
    } catch {
      // ignore
    }
  }
  streams.clear()
}