/**
 * Container RPC handlers.
 */
import { PassThrough } from 'stream'
import { docker, extractIpAddress } from '../services/docker.js'
import * as validation from '../utils/validation.js'
import { Pushes } from '../../shared/protocol.js'
import { peers } from '../core/peer-registry.js'
import { getHistory } from '../services/stats-history.js'
import { validateCreateOptions } from '../utils/engine-capabilities.js'
import logger from '../utils/logger.js'

export function registerContainerHandlers(session) {
  session.respond('validateCreateOptions', async (args) => {
    const result = await validateCreateOptions(args?.options || args || {}, {
      strict: args?.strict !== false,
    })
    return { success: result.ok, type: 'validateCreateOptions', ...result }
  })

  session.respond('createContainer', async (args) => {
    // Full Docker Engine create API passthrough with Engine-version validation
    const name = args.name || args.Name
    if (name) {
      const sanitized = validation.sanitizeString(name, 63)
      if (!validation.isValidContainerName(sanitized)) {
        throw new Error('Invalid container name')
      }
      args.name = sanitized
    }
    const createOpts = { ...args }
    delete createOpts.start
    delete createOpts.skipValidation

    if (args.skipValidation !== true) {
      const check = await validateCreateOptions(createOpts, { strict: true })
      if (!check.ok) {
        const err = new Error(
          `Create options incompatible with Engine API ${check.apiVersion}: ${check.errors.join('; ')}`
        )
        err.code = 'ENGINE_CAPABILITY'
        throw err
      }
    }

    // dockerode uses `name` as separate option
    const container = await docker.createContainer(createOpts)
    if (args.start) {
      try {
        await container.start()
      } catch (startErr) {
        // Rollback: remove container if start failed after create
        try {
          await container.remove({ force: true })
        } catch {
          // ignore
        }
        throw new Error(`Container created but start failed (rolled back): ${startErr.message}`)
      }
    }
    await broadcastContainers()
    return {
      success: true,
      message: `Container created${args.start ? ' and started' : ''}`,
      id: container.id,
      data: { Id: container.id },
    }
  })

  session.respond('listContainers', async (args = {}) => {
    const listOpts = { all: args.all !== false }
    if (args.filters) listOpts.filters = args.filters
    // Convenience state filter → Docker filters
    if (args.state && !listOpts.filters) {
      listOpts.filters = { status: [String(args.state)] }
    }

    let containers = await docker.listContainers(listOpts)

    if (args.name) {
      const q = String(args.name).toLowerCase()
      containers = containers.filter((c) =>
        (c.Names || []).some((n) => n.toLowerCase().includes(q))
      )
    }

    const total = containers.length
    const offset = Math.max(0, Number(args.offset) || 0)
    const limit = args.limit != null ? Math.min(Number(args.limit) || 50, 1000) : null
    const page = limit != null ? containers.slice(offset, offset + limit) : containers.slice(offset)

    const detailed = await Promise.all(
      page.map(async (container) => {
        try {
          const details = await docker.getContainer(container.Id).inspect()
          return { ...container, ipAddress: extractIpAddress(details) }
        } catch (error) {
          logger.error('Failed to inspect container', { id: container.Id, error: error.message })
          return { ...container, ipAddress: 'Error Retrieving IP' }
        }
      })
    )
    return {
      type: 'containers',
      data: detailed,
      total,
      offset,
      limit: limit ?? detailed.length,
      hasMore: limit != null ? offset + limit < total : false,
    }
  })

  session.respond('inspectContainer', async (args) => {
    const config = await docker.getContainer(args.id).inspect()
    return { type: 'containerConfig', data: config }
  })

  session.respond('startContainer', async (args) => {
    await docker.getContainer(args.id).start()
    return { success: true, message: `Container ${args.id} started` }
  })

  session.respond('stopContainer', async (args) => {
    const opts = {}
    if (args.t != null || args.timeout != null) {
      opts.t = Number(args.t ?? args.timeout)
    }
    await docker.getContainer(args.id).stop(opts)
    return { success: true, message: `Container ${args.id} stopped` }
  })

  session.respond('restartContainer', async (args) => {
    const opts = {}
    if (args.t != null || args.timeout != null) {
      opts.t = Number(args.t ?? args.timeout)
    }
    await docker.getContainer(args.id).restart(opts)
    return { success: true, message: `Container ${args.id} restarted` }
  })

  session.respond('killContainer', async (args) => {
    const opts = {}
    if (args.signal) opts.signal = String(args.signal)
    await docker.getContainer(args.id).kill(opts)
    return {
      success: true,
      message: `Container ${args.id} killed${opts.signal ? ` (${opts.signal})` : ''}`,
    }
  })

  session.respond('pauseContainer', async (args) => {
    await docker.getContainer(args.id).pause()
    return { success: true, message: `Container ${args.id} paused` }
  })

  session.respond('unpauseContainer', async (args) => {
    await docker.getContainer(args.id).unpause()
    return { success: true, message: `Container ${args.id} unpaused` }
  })

  session.respond('removeContainer', async (args) => {
    const id = args.id
    session._cleanupLogsForContainer?.(id)
    await docker.getContainer(id).remove({
      force: args.force !== false,
      v: Boolean(args.v || args.removeVolumes),
      link: Boolean(args.link),
    })
    return { success: true, message: `Container ${id} removed` }
  })

  session.respond('renameContainer', async (args) => {
    const newName = validation.sanitizeString(args.name, 63)
    if (!newName || !validation.isValidContainerName(newName)) {
      throw new Error('Invalid container name. Must be alphanumeric with dashes/underscores, 1-63 characters.')
    }
    await docker.getContainer(args.id).rename({ name: newName })
    return { success: true, message: `Container renamed to "${newName}"` }
  })

  session.respond('commitContainer', async (args) => {
    const commitOptions = {
      repo: validation.sanitizeString(args.repo, 255),
      tag: validation.sanitizeString(args.tag || 'latest', 128),
    }
    if (args.message) commitOptions.comment = validation.sanitizeString(args.message, 500)
    if (args.author) commitOptions.author = validation.sanitizeString(args.author, 255)
    const image = await docker.getContainer(args.id).commit(commitOptions)
    return {
      success: true,
      message: `Container committed as ${commitOptions.repo}:${commitOptions.tag}`,
      data: image.id,
    }
  })

  session.respond('exportContainer', async (args) => {
    // Full container filesystem export as tar (size-capped, chunked push)
    const maxBytes = Math.min(Number(args.maxBytes) || 50 * 1024 * 1024, 100 * 1024 * 1024)
    const chunkSize = Math.min(Number(args.chunkSize) || 256 * 1024, 512 * 1024)
    const stream = await docker.getContainer(args.id).export()
    const transferId = `export-${String(args.id).slice(0, 12)}-${Date.now()}`
    const chunks = []
    let total = 0
    await new Promise((resolve, reject) => {
      stream.on('data', (chunk) => {
        total += chunk.length
        if (total > maxBytes) {
          stream.destroy()
          reject(new Error(`Export exceeds maxBytes (${maxBytes})`))
          return
        }
        chunks.push(chunk)
      })
      stream.on('end', resolve)
      stream.on('error', reject)
    })
    const buf = Buffer.concat(chunks)
    let index = 0
    for (let offset = 0; offset < buf.length; offset += chunkSize) {
      const slice = buf.subarray(offset, Math.min(offset + chunkSize, buf.length))
      session.push(Pushes.binaryChunk, {
        type: 'binaryChunk',
        transferId,
        kind: 'containerExport',
        containerId: args.id,
        index,
        totalBytes: buf.length,
        encoding: 'base64',
        data: slice.toString('base64'),
        done: offset + chunkSize >= buf.length,
      })
      index += 1
    }
    return {
      success: true,
      type: 'containerExport',
      transferId,
      id: args.id,
      size: buf.length,
      chunks: index,
      encoding: 'base64',
      data: buf.length <= 2 * 1024 * 1024 ? buf.toString('base64') : null,
    }
  })

  session.respond('attachContainer', async (args) => {
    const container = docker.getContainer(args.id)
    const stream = await container.attach({
      stream: true,
      stdin: args.stdin !== false,
      stdout: true,
      stderr: true,
      logs: Boolean(args.logs),
    })
    const attachKey = `attach:${args.id}`
    // Close previous attach for this container
    const prev = session.state.get(attachKey)
    if (prev?.stream) {
      try {
        prev.stream.end()
      } catch {
        // ignore
      }
    }
    session.state.set(attachKey, { stream, containerId: args.id })
    stream.on('data', (chunk) => {
      session.push(Pushes.attachOutput, {
        type: 'attachOutput',
        containerId: args.id,
        data: chunk.toString('base64'),
        encoding: 'base64',
      })
    })
    stream.on('end', () => {
      session.state.delete(attachKey)
    })
    stream.on('error', (err) => {
      logger.error('Attach stream error', { id: args.id, error: err.message })
      session.state.delete(attachKey)
    })
    return { success: true, message: `Attached to container ${args.id}`, containerId: args.id }
  })

  session.respond('attachInput', async (args) => {
    const key = `attach:${args.containerId || args.id}`
    const entry = session.state.get(key)
    if (!entry) throw new Error('No active attach session for container')
    const inputData =
      args.encoding === 'base64'
        ? Buffer.from(args.data, 'base64')
        : Buffer.from(args.data || '', 'utf8')
    entry.stream.write(inputData)
    return { success: true }
  })

  session.respond('getStatsHistory', async (args) => {
    if (!args.id) throw new Error('container id required')
    const data = getHistory(args.id, { limit: args.limit })
    return { success: true, type: 'statsHistory', id: args.id, data }
  })

  session.respond('getContainerLogs', async (args) => {
    const id = args.id || args.containerId
    if (!id) throw new Error('container id required')
    const opts = {
      stdout: args.stdout !== false,
      stderr: args.stderr !== false,
      timestamps: args.timestamps !== false,
      tail: Math.min(Number(args.tail) || 500, 10000),
      follow: false,
    }
    if (args.since != null) opts.since = args.since
    if (args.until != null) opts.until = args.until
    const buf = await docker.getContainer(id).logs(opts)
    let text = Buffer.isBuffer(buf) ? demuxLogs(buf) : String(buf)
    // Client-side style filters applied server-side for download
    if (args.search) {
      const q = String(args.search).toLowerCase()
      text = text
        .split('\n')
        .filter((line) => line.toLowerCase().includes(q))
        .join('\n')
    }
    if (args.level) {
      const level = String(args.level).toLowerCase()
      const patterns = {
        error: /\b(error|err|fatal|panic)\b/i,
        warn: /\b(warn|warning)\b/i,
        info: /\b(info)\b/i,
        debug: /\b(debug|trace)\b/i,
      }
      const re = patterns[level]
      if (re) {
        text = text
          .split('\n')
          .filter((line) => re.test(line))
          .join('\n')
      }
    }
    return {
      success: true,
      type: 'containerLogs',
      id,
      encoding: 'utf8',
      size: text.length,
      data: text,
    }
  })

  session.respond('updateContainer', async (args) => {
    const id = args.id
    if (!id) throw new Error('Container id required')

    // Docker Engine update API: HostConfig subset (CPU/memory/restart/…)
    const update = {}
    if (args.Memory != null || args.memory != null) {
      const mem = args.Memory ?? args.memory
      update.Memory = typeof mem === 'number' && mem < 1e6 ? mem * 1024 * 1024 : Number(mem)
    }
    if (args.MemoryReservation != null) update.MemoryReservation = Number(args.MemoryReservation)
    if (args.MemorySwap != null) update.MemorySwap = Number(args.MemorySwap)
    if (args.CpuShares != null) update.CpuShares = Number(args.CpuShares)
    if (args.NanoCpus != null) update.NanoCpus = Number(args.NanoCpus)
    if (args.CpuQuota != null) update.CpuQuota = Number(args.CpuQuota)
    if (args.CpuPeriod != null) update.CpuPeriod = Number(args.CpuPeriod)
    if (args.CpusetCpus != null) update.CpusetCpus = String(args.CpusetCpus)
    if (args.CpusetMems != null) update.CpusetMems = String(args.CpusetMems)
    if (args.BlkioWeight != null) update.BlkioWeight = Number(args.BlkioWeight)
    if (args.RestartPolicy) {
      update.RestartPolicy =
        typeof args.RestartPolicy === 'string'
          ? { Name: args.RestartPolicy }
          : args.RestartPolicy
    }
    if (args.PidsLimit != null) update.PidsLimit = Number(args.PidsLimit)

    // Allow passthrough of raw HostConfig-style keys under args.update
    if (args.update && typeof args.update === 'object') {
      Object.assign(update, args.update)
    }

    if (Object.keys(update).length === 0) {
      throw new Error(
        'No update fields provided. Supported: Memory, CpuShares, NanoCpus, RestartPolicy, CpusetCpus, …'
      )
    }

    const data = await docker.getContainer(id).update(update)
    return {
      success: true,
      message: `Container ${id} updated`,
      data,
      applied: update,
    }
  })

  session.respond('containerTop', async (args) => {
    const opts = {}
    if (args.ps_args || args.psArgs) opts.ps_args = args.ps_args || args.psArgs
    const data = await docker.getContainer(args.id).top(opts)
    return { success: true, type: 'containerTop', id: args.id, data }
  })

  session.respond('containerStats', async (args) => {
    const stream = args.stream === true
    const stats = await docker.getContainer(args.id).stats({ stream: false })
    // dockerode returns a stream when stream:true; one-shot when false
    if (stream) {
      return {
        success: true,
        type: 'containerStats',
        id: args.id,
        note: 'Live stats are pushed via push:allStats; this is a one-shot snapshot',
        data: stats,
      }
    }
    return { success: true, type: 'containerStats', id: args.id, data: stats }
  })

  session.respond('waitContainer', async (args) => {
    const opts = {}
    if (args.condition) opts.condition = String(args.condition)
    const result = await docker.getContainer(args.id).wait(opts)
    return { success: true, type: 'containerWait', id: args.id, data: result }
  })

  session.respond('archiveContainerGet', async (args) => {
    const path = validation.sanitizeString(args.path || '/', 4096)
    if (!path || path.includes('..')) throw new Error('Invalid path')
    const maxBytes = Math.min(Number(args.maxBytes) || 5 * 1024 * 1024, 20 * 1024 * 1024)

    const stream = await docker.getContainer(args.id).getArchive({ path })
    const chunks = []
    let total = 0
    await new Promise((resolve, reject) => {
      stream.on('data', (chunk) => {
        total += chunk.length
        if (total > maxBytes) {
          stream.destroy()
          reject(new Error(`Archive exceeds maxBytes (${maxBytes})`))
          return
        }
        chunks.push(chunk)
      })
      stream.on('end', resolve)
      stream.on('error', reject)
    })

    const buf = Buffer.concat(chunks)
    return {
      success: true,
      type: 'containerArchive',
      id: args.id,
      path,
      encoding: 'base64',
      size: buf.length,
      data: buf.toString('base64'),
    }
  })

  session.respond('archiveContainerPut', async (args) => {
    const path = validation.sanitizeString(args.path || '/', 4096)
    if (!path || path.includes('..')) throw new Error('Invalid path')
    if (!args.data) throw new Error('Archive data (base64) required')

    const buf = Buffer.from(args.data, args.encoding === 'utf8' ? 'utf8' : 'base64')
    const maxBytes = 20 * 1024 * 1024
    if (buf.length > maxBytes) throw new Error(`Archive exceeds ${maxBytes} bytes`)

    await docker.getContainer(args.id).putArchive(buf, { path })
    return {
      success: true,
      message: `Archive extracted to ${path} in container ${args.id}`,
      size: buf.length,
    }
  })

  session.respond('pruneContainers', async (args) => {
    const opts = {}
    if (args.filters) opts.filters = args.filters
    const result = await docker.pruneContainers(opts)
    await broadcastContainers()
    return {
      success: true,
      type: 'pruneContainers',
      message: 'Unused containers pruned',
      data: result,
    }
  })

  session.respond('bulkContainerOperation', async (args) => {
    const { containerIds, operation } = args
    if (!Array.isArray(containerIds) || containerIds.length === 0) {
      throw new Error('No containers specified')
    }
    if (!['start', 'stop', 'restart', 'pause', 'unpause', 'remove', 'kill'].includes(operation)) {
      throw new Error('Invalid operation')
    }

    const results = []
    for (const containerId of containerIds) {
      try {
        const container = docker.getContainer(containerId)
        switch (operation) {
          case 'start':
            await container.start()
            break
          case 'stop':
            await container.stop()
            break
          case 'restart':
            await container.restart()
            break
          case 'pause':
            await container.pause()
            break
          case 'unpause':
            await container.unpause()
            break
          case 'kill':
            await container.kill(args.signal ? { signal: args.signal } : {})
            break
          case 'remove':
            await container.remove({ force: true })
            break
        }
        results.push({ id: containerId, success: true })
      } catch (err) {
        results.push({ id: containerId, success: false, error: err.message })
      }
    }
    return { success: true, message: 'Bulk operation completed', results }
  })

  session.respond('duplicateContainer', async (args) => {
    return duplicateContainer(args, session)
  })

  session.respond('execContainer', async (args) => {
    const container = docker.getContainer(args.id)
    const exec = await container.exec({
      Cmd: args.cmd || ['/bin/sh'],
      AttachStdin: true,
      AttachStdout: true,
      AttachStderr: true,
      Tty: args.tty !== false,
    })
    const stream = await exec.start({ hijack: true, stdin: true })
    const stdout = new PassThrough()
    const stderr = new PassThrough()
    container.modem.demuxStream(stream, stdout, stderr)

    const execKey = `exec:${exec.id}`
    session.state.set(execKey, { stream, exec, containerId: args.id })

    stdout.on('data', (chunk) => {
      session.push(Pushes.execOutput, {
        type: 'execOutput',
        containerId: args.id,
        execId: exec.id,
        data: chunk.toString('base64'),
        encoding: 'base64',
      })
    })
    stderr.on('data', (chunk) => {
      session.push(Pushes.execErrorOutput, {
        type: 'execErrorOutput',
        containerId: args.id,
        execId: exec.id,
        data: chunk.toString('base64'),
        encoding: 'base64',
      })
    })

    return { success: true, message: 'Exec session started', execId: exec.id }
  })

  session.respond('execInput', async (args) => {
    const key = `exec:${args.execId}`
    const entry = session.state.get(key)
    if (!entry) throw new Error('Exec session not found')
    const inputData =
      args.encoding === 'base64'
        ? Buffer.from(args.data, 'base64')
        : Buffer.from(args.data || '', 'utf8')
    entry.stream.write(inputData)
    return { success: true }
  })
}

async function duplicateContainer(args, session) {
  const { name, image, hostname, netmode, cpu, memory, config } = args
  const memoryInMB = memory * 1024 * 1024

  const sanitizedConfig = { ...(config || {}) }
  for (const key of [
    'Id', 'State', 'Created', 'NetworkSettings', 'Mounts', 'Path', 'Args',
    'Image', 'Hostname', 'CpuCount', 'Memory', 'CpuShares', 'CpusetCpus',
  ]) {
    delete sanitizedConfig[key]
  }

  const existing = await docker.listContainers({ all: true })
  if (existing.some((c) => c.Names.includes(`/${name}`))) {
    throw new Error(`Container name '${name}' already exists.`)
  }

  const cpusetCpus = Array.from({ length: cpu }, (_, i) => i).join(',')
  const nanoCpus = cpu * 1e9

  const newContainer = await docker.createContainer({
    ...sanitizedConfig.Config,
    name,
    Hostname: hostname,
    Image: image,
    HostConfig: {
      CpusetCpus: cpusetCpus.toString(),
      NanoCpus: nanoCpus,
      Memory: Number(memoryInMB),
      MemoryReservation: Number(memoryInMB),
      NetworkMode: String(netmode),
    },
  })
  await newContainer.start()

  await broadcastContainers()
  return { success: true, message: `Container '${name}' duplicated and started successfully.` }
}

export async function broadcastContainers() {
  try {
    const containers = await docker.listContainers({ all: true })
    peers.broadcast(Pushes.containers, { type: 'containers', data: containers })
  } catch (err) {
    logger.error('Failed to broadcast containers', { error: err.message })
  }
}

/** Strip docker multiplex headers from log buffers when possible. */
function demuxLogs(buffer) {
  try {
    const chunks = []
    let offset = 0
    while (offset + 8 <= buffer.length) {
      const size = buffer.readUInt32BE(offset + 4)
      if (size < 0 || offset + 8 + size > buffer.length) {
        return buffer.toString('utf8')
      }
      chunks.push(buffer.subarray(offset + 8, offset + 8 + size).toString('utf8'))
      offset += 8 + size
    }
    if (chunks.length) return chunks.join('')
  } catch {
    // fall through
  }
  return buffer.toString('utf8')
}