peardock/server/services/events.js

/**
 * Docker event stream → peer broadcasts.
 * Auto-reconnects when the daemon restarts or the stream ends.
 */
import { docker, extractVolumesList } from './docker.js'
import { peers } from '../core/peer-registry.js'
import { Pushes } from '../../shared/protocol.js'
import { broadcastContainers } from '../handlers/containers.js'
import logger from '../utils/logger.js'

let dockerEventStream = null
let reconnectTimer = null
let stopped = false
let reconnectAttempt = 0
/** Coalesce noisy container events into one list broadcast */
let containerListBroadcastTimer = null

const BASE_DELAY_MS = 2000
const MAX_DELAY_MS = 30_000
/** Only these container actions should refresh the fleet list (not exec/health spam) */
const CONTAINER_LIST_ACTIONS = new Set([
  'create',
  'start',
  'stop',
  'die',
  'kill',
  'destroy',
  'remove',
  'pause',
  'unpause',
  'restart',
  'rename',
  'update',
  'oom',
])

function scheduleContainerListBroadcast() {
  if (containerListBroadcastTimer) return
  containerListBroadcastTimer = setTimeout(() => {
    containerListBroadcastTimer = null
    broadcastContainers().catch((err) => {
      logger.debug('container list broadcast failed', { error: err.message })
    })
  }, 400)
}

export async function startDockerEventStream() {
  stopped = false
  await openEventStream()
}

async function openEventStream() {
  if (stopped) return
  if (reconnectTimer) {
    clearTimeout(reconnectTimer)
    reconnectTimer = null
  }

  try {
    const stream = await new Promise((resolve, reject) => {
      docker.getEvents({}, (err, s) => (err ? reject(err) : resolve(s)))
    })
    dockerEventStream = stream
    reconnectAttempt = 0
    logger.info('Docker event stream connected')

    stream.on('data', async (chunk) => {
      try {
        const lines = chunk
          .toString()
          .split('\n')
          .map((l) => l.trim())
          .filter(Boolean)

        for (const line of lines) {
          let event
          try {
            event = JSON.parse(line)
          } catch {
            continue
          }
          if (event.status === 'undefined') continue

          logger.info('Docker event', {
            status: event.status,
            id: event.id,
            type: event.Type,
            action: event.Action,
          })

          peers.broadcast(Pushes.dockerEvent, {
            type: 'dockerEvent',
            data: event,
          })

          // Do not rebroadcast the full list on every exec_*/health_status event —
          // that thrashed clients (rows flicker / disappear on each refresh).
          if (event.Type === 'container') {
            const action = String(event.Action || event.status || '').split(':')[0]
            if (CONTAINER_LIST_ACTIONS.has(action)) {
              scheduleContainerListBroadcast()
            }
          }

          if (event.Type === 'image') {
            try {
              const images = await docker.listImages({ all: true })
              peers.broadcast(Pushes.images, { type: 'images', data: images })
            } catch (e) {
              logger.debug('image list on event failed', { error: e.message })
            }
          }

          if (event.Type === 'volume' && (event.Action === 'create' || event.Action === 'destroy')) {
            const volumesResult = await docker.listVolumes()
            const volumesList = extractVolumesList(volumesResult)
            peers.broadcast(Pushes.volumes, {
              type: 'volumes',
              data: volumesList,
              success: true,
              volumes: volumesList,
            })
          }
        }
      } catch (err) {
        logger.error('Failed to process Docker event', { error: err.message })
      }
    })

    stream.on('error', (err) => {
      logger.error('Docker event stream error', { error: err.message })
      scheduleReconnect()
    })
    stream.on('end', () => {
      dockerEventStream = null
      logger.warn('Docker event stream ended')
      scheduleReconnect()
    })
  } catch (err) {
    logger.error('Failed to start Docker event stream', { error: err.message })
    scheduleReconnect()
  }
}

function scheduleReconnect() {
  if (stopped) return
  if (reconnectTimer) return

  if (dockerEventStream) {
    try {
      dockerEventStream.destroy()
    } catch {
      // ignore
    }
    dockerEventStream = null
  }

  const delay = Math.min(MAX_DELAY_MS, BASE_DELAY_MS * 2 ** reconnectAttempt)
  reconnectAttempt += 1
  logger.info('Reconnecting Docker event stream', { delayMs: delay, attempt: reconnectAttempt })
  reconnectTimer = setTimeout(() => {
    reconnectTimer = null
    openEventStream().catch((err) => {
      logger.error('Event stream reconnect failed', { error: err.message })
      scheduleReconnect()
    })
  }, delay)
}

export function stopDockerEventStream() {
  stopped = true
  if (reconnectTimer) {
    clearTimeout(reconnectTimer)
    reconnectTimer = null
  }
  if (dockerEventStream) {
    try {
      dockerEventStream.destroy()
    } catch {
      // ignore
    }
    dockerEventStream = null
  }
}