orca/src/relay/protocol.ts

// Self-contained relay protocol — mirrors src/main/ssh/relay-protocol.ts
// but has no Electron dependencies. Deployed standalone to remote hosts.

import {
  FrameDecoder,
  FrameDecoderContinuationError,
  HEADER_LENGTH,
  MAX_MESSAGE_SIZE,
  FRAME_DECODER_MAX_FRAMES_PER_TURN,
  FRAME_DECODER_MAX_BYTES_PER_TURN,
  FRAME_DECODER_MAX_TURN_MS,
  FRAME_DECODER_MAX_RETAINED_BYTES
} from './relay-frame-decoder'

export {
  FrameDecoder,
  FrameDecoderContinuationError,
  HEADER_LENGTH,
  MAX_MESSAGE_SIZE,
  FRAME_DECODER_MAX_FRAMES_PER_TURN,
  FRAME_DECODER_MAX_BYTES_PER_TURN,
  FRAME_DECODER_MAX_TURN_MS,
  FRAME_DECODER_MAX_RETAINED_BYTES
}
export type { DecodedFrame, FrameDecoderOptions } from './relay-frame-decoder'

export const RELAY_VERSION = '0.1.0'
export const RELAY_SENTINEL = `ORCA-RELAY v${RELAY_VERSION} READY\n`

export const MessageType = {
  Regular: 1,
  Handshake: 2,
  KeepAlive: 9
} as const

// Why: a pre-dispatcher envelope on a freshly-accepted Unix socket. The daemon
// reads exactly one Handshake frame before attaching the JSON-RPC dispatcher,
// to refuse mismatched-version --connect bridges that would otherwise drive a
// stale daemon.
export type HandshakeMessage =
  | { type: 'orca-relay-handshake'; version: string; endpointCredential?: string }
  // Why optional and informational: bridge and daemon already match exactly on `version`;
  // this only names the runtime executing the daemon (design D8.1).
  | { type: 'orca-relay-handshake-ok'; version: string; runtime?: RelayHandshakeRuntime }
  | { type: 'orca-relay-handshake-mismatch'; expected: string; got: string }
  // Why a distinct reply: the bridge exits with its own code so the client can tell a refused
  // credential from a crashed relay. Old bridges reject the unknown type and exit 1 pre-sentinel.
  | { type: 'orca-relay-handshake-credential-mismatch' }

export type RelayHandshakeRuntime = { kind: string; version: string }

export function encodeHandshakeFrame(msg: HandshakeMessage): Buffer {
  const payload = Buffer.from(JSON.stringify(msg), 'utf-8')
  return encodeFrame(MessageType.Handshake, 0, 0, payload)
}

// Why the fields are checked and not just the type: this frame arrives before any credential, and
// both sides interpolate its version fields into log lines. `JSON.parse` can produce values a
// template literal throws on, so anything that reaches a reader must already be a string.
const HANDSHAKE_STRING_FIELDS: Readonly<Record<HandshakeMessage['type'], readonly string[]>> = {
  'orca-relay-handshake': ['version'],
  'orca-relay-handshake-ok': ['version'],
  'orca-relay-handshake-mismatch': ['expected', 'got'],
  'orca-relay-handshake-credential-mismatch': []
}

// Optional fields are peer-supplied too, so the parser only proves the type of what it returns if
// it refuses a present-but-wrong one. `endpointCredential` survives today only because its single
// reader compares it and never interpolates it; the next reader to log it would restore the bug
// this function exists to stop. Absent stays absent — refusing that would break a bridge that
// legitimately presents no credential.
const HANDSHAKE_OPTIONAL_STRING_FIELDS: Readonly<
  Record<HandshakeMessage['type'], readonly string[]>
> = {
  'orca-relay-handshake': ['endpointCredential'],
  'orca-relay-handshake-ok': [],
  'orca-relay-handshake-mismatch': [],
  'orca-relay-handshake-credential-mismatch': []
}

export function parseHandshakeMessage(payload: Buffer): HandshakeMessage {
  const parsed: unknown = JSON.parse(payload.toString('utf-8'))
  if (typeof parsed !== 'object' || parsed === null) {
    throw new Error('Handshake payload is not an object')
  }
  // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the typeof/null guard directly above is exactly what makes this an index-able object; every read below still proves its own field.
  const msg = parsed as Record<string, unknown>
  const t = msg.type
  const required =
    typeof t === 'string' && Object.hasOwn(HANDSHAKE_STRING_FIELDS, t)
      ? // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: reached only when Object.hasOwn proved t is a key of this record, on the same line.
        HANDSHAKE_STRING_FIELDS[t as HandshakeMessage['type']]
      : null
  if (required === null) {
    // Why typeof and not String(t): a peer-supplied `{ "type": { "toString": 1 } }` makes String()
    // itself throw "Cannot convert object to primitive value", replacing the one diagnostic this
    // line exists to produce.
    throw new Error(`Unknown handshake type: ${typeof t === 'string' ? t : typeof t}`)
  }
  for (const field of required) {
    if (typeof msg[field] !== 'string') {
      throw new Error(`Handshake field ${field} is not a string`)
    }
  }
  // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the required === null bail above already refused every t that is not one of the four keys.
  for (const field of HANDSHAKE_OPTIONAL_STRING_FIELDS[t as HandshakeMessage['type']]) {
    if (msg[field] !== undefined && typeof msg[field] !== 'string') {
      throw new Error(`Handshake field ${field} is not a string`)
    }
  }
  if (t === 'orca-relay-handshake-ok' && msg.runtime !== undefined) {
    const runtime = msg.runtime
    if (
      typeof runtime !== 'object' ||
      runtime === null ||
      !('kind' in runtime) ||
      typeof runtime.kind !== 'string' ||
      !('version' in runtime) ||
      typeof runtime.version !== 'string'
    ) {
      throw new Error('Handshake field runtime is not a {kind, version} object')
    }
  }
  // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: this is the one place the shape is proved: the type is one of the four literals and every field the union declares has been checked to be a string.
  return msg as unknown as HandshakeMessage
}

export const KEEPALIVE_SEND_MS = 5_000
export const TIMEOUT_MS = 20_000

// ── Streaming constants (see docs/relay-file-stream-design.md) ─────

export const STREAM_CHUNK_SIZE = 256 * 1024
export const MAX_CONCURRENT_STREAMS = 16

/** Max unacked fs.streamChunk frames in flight per stream when the client
 * requested `flowControl: 'ack'`. Bounds how many bulk bytes an interactive
 * pty.data frame can queue behind on the shared SSH channel (~1MB raw) while
 * keeping the pipe full across one ack round-trip on fast links. */
export const STREAM_ACK_WINDOW_CHUNKS = 4

/** Safety-valve poll interval for a pump stalled on acks: re-checks stream
 * abort/staleness so a client that vanished mid-stream cannot park the pump
 * (and its open file handle) forever. */
export const STREAM_ACK_STALL_RECHECK_MS = 1_000

// ── Git response streaming (see docs/relay-git-response-stream-design.md) ──

/** Serialized-JSON size above which a streamable git response (diff family +
 * exec) is chunked onto the bulk lane instead of one JSON-RPC frame, so a large
 * diff cannot head-of-line-block interactive pty.data echo on the shared SSH
 * channel. Below this, single-frame is cheaper and avoids stream overhead. */
export const GIT_RESPONSE_STREAM_THRESHOLD = 256 * 1024

/** Per-chunk size (UTF-8 bytes of the serialized result) for git response
 * streaming. Independent from STREAM_CHUNK_SIZE — this offset math is not
 * shared with fs streams, so tuning it here is cross-version safe as long as
 * the client reassembles by concatenation (it does not depend on chunk size). */
export const GIT_RESPONSE_CHUNK_SIZE = 128 * 1024

/** Sentinel result returned in place of a large git response: the real payload
 * follows as git.responseChunk frames on the bulk lane. Old relays never emit
 * this, so a new client falls back to the plain result they return. */
export type GitResponseStreamMarker = {
  __orcaGitResponseStream: { streamId: number; totalBytes: number; chunkCount: number }
}

export const RelayErrorCode = {
  TooManyStreams: -33006,
  StreamProtocolError: -33007,
  /** Substituted for a response too large for the sink's frame capacity; the request fails
   *  instead of the whole link, so a caller can retry with a narrower scope. */
  ResponseOverCapacity: -33008
} as const

export type JsonRpcRequest = {
  jsonrpc: '2.0'
  id: number
  method: string
  params?: Record<string, unknown>
}

export type JsonRpcResponse = {
  jsonrpc: '2.0'
  id: number
  result?: unknown
  error?: { code: number; message: string; data?: unknown }
}

export type JsonRpcNotification = {
  jsonrpc: '2.0'
  method: string
  params?: Record<string, unknown>
}

export type JsonRpcMessage = JsonRpcRequest | JsonRpcResponse | JsonRpcNotification

const JSON_RPC_PAYLOAD_BYTES = Symbol('jsonRpcPayloadBytes')

export type PreparedJsonRpcPayload = Readonly<{
  byteLength: number
  [JSON_RPC_PAYLOAD_BYTES]: Buffer
}>

export function encodeFrame(
  type: number,
  id: number,
  ack: number,
  payload: Buffer | Uint8Array
): Buffer {
  const header = Buffer.alloc(HEADER_LENGTH)
  header[0] = type
  header.writeUInt32BE(id, 1)
  header.writeUInt32BE(ack, 5)
  header.writeUInt32BE(payload.length, 9)
  return Buffer.concat([header, payload])
}

export function encodeJsonRpcFrame(msg: JsonRpcMessage, id: number, ack: number): Buffer {
  return encodePreparedJsonRpcFrame(prepareJsonRpcPayload(msg), id, ack)
}

export function prepareJsonRpcPayload(msg: JsonRpcMessage): PreparedJsonRpcPayload {
  const payload = Buffer.from(JSON.stringify(msg), 'utf-8')
  if (payload.length > MAX_MESSAGE_SIZE) {
    throw new Error(`Message too large: ${payload.length} bytes`)
  }
  return Object.freeze({ byteLength: payload.length, [JSON_RPC_PAYLOAD_BYTES]: payload })
}

export function encodePreparedJsonRpcFrame(
  payload: PreparedJsonRpcPayload,
  id: number,
  ack: number
): Buffer {
  return encodeFrame(MessageType.Regular, id, ack, payload[JSON_RPC_PAYLOAD_BYTES])
}

export function encodeKeepAliveFrame(id: number, ack: number): Buffer {
  return encodeFrame(MessageType.KeepAlive, id, ack, Buffer.alloc(0))
}

export function parseJsonRpcMessage(payload: Buffer): JsonRpcMessage {
  const text = payload.toString('utf-8')
  const msg = JSON.parse(text) as JsonRpcMessage
  if (msg.jsonrpc !== '2.0') {
    throw new Error(`Invalid JSON-RPC version: ${String((msg as Record<string, unknown>).jsonrpc)}`)
  }
  return msg
}