import { createConnection } from 'node:net'
import { TURN_LIMITS, type ReviewFinding, type TurnRpcRequest, type TurnRpcResponse } from '@radial/core'
import { artifactTitle } from './commands.js'

function allValues(args: string[], flag: string): string[] {
  const results: string[] = []
  for (let index = 0; index < args.length; index += 1) {
    if (args[index] === flag && args[index + 1]) results.push(args[index + 1] as string)
  }
  return results
}

function lastValue(args: string[], flag: string): string | undefined {
  return allValues(args, flag).at(-1)
}

async function readBody(args: string[], readText: (path: string) => Promise<string>): Promise<string> {
  const inline = lastValue(args, '--body')
  const path = lastValue(args, '--body-file')
  if (Boolean(inline) === Boolean(path)) {
    throw new Error('Provide exactly one of --body or --body-file')
  }
  if (inline !== undefined) return inline
  return readText(path as string)
}

/**
 * Maps CLI args to a turn-mode RPC, enforcing the turn allowlist: inside a turn container there
 * are no credentials, so only the commands the daemon can safely broker on the agent's behalf are
 * permitted — one terminal command per turn kind (`artifact submit`, `review submit`, `answer
 * submit`) plus the can't-complete question (`message post`). Every other command (including auth,
 * and the direct-PDS `artifact post` / `review post` / etc.) would either need credentials this
 * process doesn't have or bypass the daemon's anchoring. Which terminal is actually allowed is the
 * DAEMON's decision, made from turn mode — the sidecar only shapes the envelope.
 */
export async function buildTurnRpc(
  args: string[],
  token: string,
  readText: (path: string) => Promise<string>,
): Promise<TurnRpcRequest> {
  const [group, action] = args

  if (group === 'artifact' && action === 'submit') {
    // Same rule and the same refusals as the direct `artifact post` path: the title is the artifact's
    // own, so a turn supplies it exactly as a human would. The daemon validates it again at the
    // socket edge — this is a legible error message, not the trust boundary.
    const title = artifactTitle(args)
    const body = await readBody(args, readText)
    const criteria = allValues(args, '--criterion')
    // Implementation turns pass the pushed HEAD sha via --commit; the daemon validates its shape and
    // owns everything else (branch naming, push, PR). Plan-style submits omit it.
    const commit = lastValue(args, '--commit')
    const branch = lastValue(args, '--branch')
    const pr = lastValue(args, '--pr')
    return {
      method: 'submitArtifact',
      token,
      title,
      body,
      ...(criteria.length ? { criteria } : {}),
      ...(commit !== undefined ? { commit } : {}),
      ...(branch !== undefined ? { branch } : {}),
      ...(pr !== undefined ? { pr } : {}),
    }
  }

  if (group === 'review' && action === 'submit') {
    // A review turn's terminal command. The daemon owns the subject and the request backref (from
    // turn context), so `--subject` is rejected here — a review turn cannot redirect its verdict at
    // a different artifact than the one its request named.
    if (lastValue(args, '--subject') !== undefined) {
      throw new Error('In turn socket mode `radial review submit` must not pass --subject; the daemon owns the review subject')
    }
    const verdict = lastValue(args, '--verdict')
    if (verdict !== 'approve' && verdict !== 'request_changes') {
      throw new Error('radial review submit requires --verdict approve|request_changes')
    }
    const findingsFile = lastValue(args, '--findings-file')
    if (findingsFile === undefined) {
      throw new Error('radial review submit requires --findings-file <path to a JSON array of findings>')
    }
    const raw = await readText(findingsFile)
    let findings: unknown
    try {
      findings = JSON.parse(raw)
    } catch (error) {
      throw new Error(`--findings-file is not valid JSON: ${error instanceof Error ? error.message : String(error)}`)
    }
    if (!Array.isArray(findings)) {
      throw new Error('--findings-file must contain a JSON array of findings')
    }
    // The daemon performs full per-finding validation at the socket boundary (the untrusted edge);
    // the sidecar only shapes the envelope.
    return { method: 'submitReview', token, verdict, findings: findings as ReviewFinding[] }
  }

  if (group === 'answer' && action === 'submit') {
    // An answer turn's terminal command. The daemon owns every anchoring field — the goal, the
    // `parent`/`root` threading, the `re` backref, the mention of whoever is being answered — from
    // turn context, so the envelope carries prose and nothing else.
    const body = await readBody(args, readText)
    return { method: 'submitAnswer', token, body }
  }

  if (group === 'message' && action === 'post') {
    // In socket mode a message is always the can't-complete question; the daemon anchors it to
    // the turn's goal/artifact, so --goal/--artifact/--re (used in direct-PDS mode) are ignored.
    //
    // `--parent` is REFUSED rather than ignored. In direct-PDS mode it is what makes a message a
    // reply; here the daemon owns anchoring, and silently dropping it would let a turn believe it had
    // threaded a reply — or answered its own request — when it had posted a bare question. An answer
    // turn's reply is `radial answer submit`; nothing else threads.
    if (lastValue(args, '--parent') !== undefined) {
      throw new Error(
        'In turn socket mode `radial message post` must not pass --parent; the daemon owns message anchoring. Use `radial answer submit` to post a reply on an answer turn.',
      )
    }
    const body = await readBody(args, readText)
    return { method: 'askQuestion', token, body }
  }

  const command = args.slice(0, 2).filter((part) => part !== undefined).join(' ') || '(none)'
  throw new Error(
    `In turn socket mode only "artifact submit", "review submit", "answer submit", and "message post" are permitted, got: ${command}`,
  )
}

function concatBytes(chunks: Uint8Array[]): Uint8Array {
  const total = chunks.reduce((sum, chunk) => sum + chunk.byteLength, 0)
  const result = new Uint8Array(total)
  let offset = 0
  for (const chunk of chunks) {
    result.set(chunk, offset)
    offset += chunk.byteLength
  }
  return result
}

/**
 * One RPC per connection: write a newline-delimited JSON frame, read a newline-delimited JSON
 * frame back, then close. There is no multiplexing and no keep-alive — the CLI process exits
 * right after, so a fresh connection per invocation is simpler than pooling.
 */
export async function sendTurnRpc(
  socketPath: string,
  rpc: TurnRpcRequest,
  options?: { timeoutMs?: number; maxFrameBytes?: number },
): Promise<TurnRpcResponse> {
  const maxFrameBytes = options?.maxFrameBytes ?? TURN_LIMITS.maxFrameBytes
  const timeoutMs = options?.timeoutMs ?? TURN_LIMITS.frameTimeoutMs
  const frame = `${JSON.stringify(rpc)}\n`
  const frameBytes = new TextEncoder().encode(frame)
  if (frameBytes.byteLength > maxFrameBytes) {
    throw new Error(`Turn RPC request frame of ${frameBytes.byteLength} bytes exceeds the ${maxFrameBytes}-byte limit`)
  }

  // `socketPath` is a filesystem path in the production unix-transport case; the dev-only tcp
  // transport (see daemon's turn.ts / cli.ts fail-closed rule) instead hands us a `tcp://host:port`
  // URL, since there is no socket file to point at.
  const connectTarget = socketPath.startsWith('tcp://')
    ? (() => {
        const url = new URL(socketPath)
        return { host: url.hostname, port: Number(url.port) }
      })()
    : { path: socketPath }

  return new Promise<TurnRpcResponse>((resolve, reject) => {
    const socket = createConnection(connectTarget)
    let settled = false
    let received = 0
    const chunks: Uint8Array[] = []

    const finish = (fn: () => void): void => {
      if (settled) return
      settled = true
      fn()
    }
    const fail = (error: Error): void => {
      finish(() => {
        socket.destroy()
        reject(error)
      })
    }

    socket.setTimeout(timeoutMs, () => fail(new Error(`Turn RPC timed out after ${timeoutMs}ms`)))
    socket.on('error', (error) => fail(error))
    socket.on('close', () => fail(new Error('Turn socket closed before a response was received')))
    socket.on('connect', () => socket.write(frame))
    socket.on('data', (chunk) => {
      if (settled) return
      const newlineIndex = chunk.indexOf(0x0a)
      const upTo = newlineIndex === -1 ? chunk : chunk.subarray(0, newlineIndex)
      received += upTo.byteLength
      if (received > maxFrameBytes) {
        fail(new Error(`Turn RPC response exceeds the ${maxFrameBytes}-byte limit`))
        return
      }
      chunks.push(upTo)
      if (newlineIndex === -1) return
      const text = new TextDecoder().decode(concatBytes(chunks))
      finish(() => {
        socket.destroy()
        try {
          resolve(JSON.parse(text) as TurnRpcResponse)
        } catch (error) {
          reject(error instanceof Error ? error : new Error(String(error)))
        }
      })
    })
  })
}