aiturk-hermes-ide/apps/desktop/electron/mobile-relay-client.ts

/**
 * mobile-relay-client.ts
 *
 * "Telefon bağlantısı" through the turkservis.online tunnel (contract:
 * workspace/aiturk-mobil-tunel-spec.md, v3). The IDE opens ONE outbound
 * WebSocket to the gateway. For every phone stream the gateway announces
 * (`open`), this client opens a PLAIN TCP connection to this IDE's own TLS
 * proxy on loopback (electron/mobile-lan-proxy.ts) and copies bytes both ways.
 *
 * It never terminates TLS and never looks at a byte: the TLS session is
 * between the phone (which pins the proxy's certificate) and the proxy. So the
 * gateway — and this file — only ever carry ciphertext, and every rule (device
 * token, forbidden list, root 404, WebSocket path list) stays in the proxy.
 *
 * Security contract (do not relax):
 *   - `hostToken` (`tsmr_…`) lives only in the injected credential store (the
 *     controller keeps it in the safeStorage vault). It is sent in the
 *     `Authorization` header of the host WebSocket and nowhere else: not in a
 *     log line, a status object or an error message.
 *   - Log lines are status changes only.
 *   - Errors are fixed codes.
 *
 * Everything that touches the outside is injected (WebSocket connector, fetch,
 * local TCP dial, timers, credential store) so the tests run against a fake
 * relay on 127.0.0.1 and a fake clock.
 */

export const MOBILE_RELAY_ORIGIN = 'https://turkservis.online'
export const MOBILE_RELAY_PATH = '/v1/aiturk/mobile-relay'
export const MOBILE_RELAY_FETCH_TIMEOUT_MS = 20_000
/** Largest payload of one binary frame, either direction (contract §1). */
export const RELAY_CHUNK_BYTES = 64 * 1024
/** Stream id prefix of a binary frame: big-endian uint32. */
export const RELAY_FRAME_HEADER_BYTES = 4
/** Pending bytes allowed per direction of one stream (contract §1 / §2). */
export const RELAY_BUFFER_LIMIT_BYTES = 1024 * 1024
/** Reconnect delays: 2 s, then growing to 30 s. */
export const RELAY_BACKOFF_MS: readonly number[] = [2000, 4000, 8000, 16_000, 30_000]
/** The gateway pings every 20 s and drops a silent host after 60 s; we give up a little later. */
export const RELAY_SILENCE_MS = 75_000
/** Concurrent phone streams (the gateway allows 16 per relay). */
export const RELAY_MAX_STREAMS = 16
/** Upper bound for one frame on the host socket; a data frame is at most 64 KiB + 4. */
export const RELAY_MAX_FRAME_BYTES = 1024 * 1024

const BUFFER_RESUME_BYTES = 256 * 1024
const BUFFER_POLL_MS = 25

const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/
const HOST_TOKEN_RE = /^tsmr_[A-Za-z0-9_-]{43}$/

export const relayHostUrl = (relayId: string) => `wss://turkservis.online${MOBILE_RELAY_PATH}/host/${relayId}`

/** The `r` of the QR and the `relay` of the rendezvous record: a lower-case uuid, nothing else. */
export function isRelayId(value: unknown): value is string {
  return typeof value === 'string' && UUID_RE.test(value)
}

export interface RelayCredentials {
  relayId: string
  hostToken: string
}

export function isRelayCredentials(value: unknown): value is RelayCredentials {
  const row = value as null | RelayCredentials

  return Boolean(row) && isRelayId(row.relayId) && typeof row.hostToken === 'string' && HOST_TOKEN_RE.test(row.hostToken)
}

/** Validates the `POST /sessions` answer: `{relayId, hostToken}` and nothing we would follow. */
export function parseRelaySession(data: unknown): null | RelayCredentials {
  const row = (data && typeof data === 'object' ? data : {}) as Record<string, unknown>
  const candidate = { relayId: typeof row.relayId === 'string' ? row.relayId.toLowerCase() : row.relayId, hostToken: row.hostToken }

  return isRelayCredentials(candidate) ? { relayId: candidate.relayId, hostToken: candidate.hostToken } : null
}

export interface RelayFetchResponse {
  status: number
  json: () => Promise<unknown>
}

export type RelayFetch = (url: string, init: Record<string, unknown>) => Promise<RelayFetchResponse>

export type RelayFailureCode =
  | 'register_failed'
  | 'rate_limited'
  | 'vault_unavailable'
  | 'connect_failed'
  | 'token_rejected'
  | 'replaced'
  | 'busy'
  | 'timeout'

/** Wait before the single second try of a request whose first try never got an answer. */
export const MOBILE_NETWORK_RETRY_DELAY_MS = 500

/** An aborted / timed-out request already used its whole budget: it is not tried again. */
export function isTimeoutError(error: unknown): boolean {
  const name = (error as null | { name?: unknown })?.name

  return name === 'TimeoutError' || name === 'AbortError'
}

export interface NetworkRetryOptions {
  /** Injected wait (tests: fake clock). Default: a real timer. */
  sleep?: (ms: number) => Promise<void>
  retryDelayMs?: number
}

const realSleep = (ms: number) => new Promise<void>(resolve => setTimeout(resolve, ms))

/**
 * `POST /sessions`; a fixed failure code instead of any upstream text.
 *
 * A process that sat idle for a long time can fail its first request on a dead
 * pooled keep-alive connection. So a THROWN network error (no HTTP answer at
 * all) is tried once more after ~0.5 s. Any HTTP answer is final: no retry.
 */
export async function registerRelaySession(
  fetcher: RelayFetch,
  options: NetworkRetryOptions = {}
): Promise<RelayCredentials | { error: RelayFailureCode }> {
  const sleep = options.sleep ?? realSleep

  for (let attempt = 1; ; attempt++) {
    let response: RelayFetchResponse

    try {
      response = await fetcher(`${MOBILE_RELAY_ORIGIN}${MOBILE_RELAY_PATH}/sessions`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json', Accept: 'application/json' },
        body: '{}',
        redirect: 'error',
        signal: AbortSignal.timeout(MOBILE_RELAY_FETCH_TIMEOUT_MS)
      })
    } catch (error) {
      if (attempt >= 2 || isTimeoutError(error)) {
        return { error: 'register_failed' }
      }

      await sleep(options.retryDelayMs ?? MOBILE_NETWORK_RETRY_DELAY_MS)

      continue
    }

    try {
      if (response.status !== 201) {
        return { error: response.status === 429 ? 'rate_limited' : 'register_failed' }
      }

      return parseRelaySession(await response.json()) ?? { error: 'register_failed' }
    } catch {
      return { error: 'register_failed' }
    }
  }
}

// ── What the client needs from the outside ──────────────────────────────────

/** Host WebSocket (the `ws` package fits: it can send request headers). */
export interface RelaySocket {
  on: (event: string, listener: (...args: any[]) => void) => unknown
  /** A string is a text frame, a Buffer a binary frame. */
  send: (data: Buffer | string) => unknown
  close: (code?: number) => unknown
  terminate?: () => unknown
  /** Stop / continue reading from the gateway (back-pressure towards a slow local connection). */
  pause?: () => unknown
  resume?: () => unknown
  readonly bufferedAmount?: number
}

export type RelayConnect = (url: string, headers: Record<string, string>) => RelaySocket

export interface RelayWebSocketCtor {
  new (url: string, options?: Record<string, unknown>): any
}

/** Plain TCP connection to the local proxy (`net.Socket` fits). */
export interface RelayLocalSocket {
  on: (event: string, listener: (...args: any[]) => void) => unknown
  write: (data: Buffer) => boolean
  destroy: () => unknown
  pause: () => unknown
  resume: () => unknown
  readonly writableLength?: number
}

/** null = the proxy is not listening right now. */
export type RelayDial = () => null | RelayLocalSocket

/** One binary frame: stream id (big-endian uint32) + payload. */
export function encodeRelayFrame(id: number, payload: Buffer): Buffer {
  const frame = Buffer.allocUnsafe(RELAY_FRAME_HEADER_BYTES + payload.length)

  frame.writeUInt32BE(id, 0)
  payload.copy(frame, RELAY_FRAME_HEADER_BYTES)

  return frame
}

/** Pieces of at most `size` bytes, in order, without copying. */
export function splitRelayChunks(chunk: Buffer, size = RELAY_CHUNK_BYTES): Buffer[] {
  const parts: Buffer[] = []

  for (let offset = 0; offset < chunk.length; offset += size) {
    parts.push(chunk.subarray(offset, offset + size))
  }

  return parts
}

const validStreamId = (value: unknown): value is number => Number.isInteger(value) && (value as number) >= 0 && (value as number) <= 0xffffffff

// ── The client ──────────────────────────────────────────────────────────────

export type RelayStatus = 'off' | 'connecting' | 'connected' | 'failed'

export interface RelayState {
  status: RelayStatus
  /** Fixed code while `failed` (a retry is always scheduled), otherwise null. */
  code: null | RelayFailureCode
}

export interface RelayCredentialStore {
  /** null when there is no registration yet. Throwing = the vault is unusable. */
  load: () => Promise<null | RelayCredentials>
  /** `reissued` = this replaces a registration (rejected by the gateway, or renewed by the user). */
  save: (credentials: RelayCredentials, info: { reissued: boolean }) => Promise<void>
  clear: () => Promise<void>
}

export interface RelayClientDeps {
  connect: RelayConnect
  fetch: RelayFetch
  dial: RelayDial
  credentials: RelayCredentialStore
  setTimer?: (fn: () => void, ms: number) => unknown
  clearTimer?: (handle: unknown) => void
  /** Status changes only. */
  log?: (line: string) => void
  onState?: (state: RelayState) => void
}

interface Stream {
  local: RelayLocalSocket
  paused: boolean
}

export function createRelayClient(deps: RelayClientDeps) {
  const setTimer = deps.setTimer ?? ((fn, ms) => setTimeout(fn, ms))
  const clearTimer = deps.clearTimer ?? (handle => clearTimeout(handle as NodeJS.Timeout))

  let state: RelayState = { status: 'off', code: null }
  let wanted = false
  let generation = 0
  let attempt = 0
  let socket: null | RelaySocket = null
  let credentials: null | RelayCredentials = null
  /** The gateway rejected the stored registration; the next one replaces it. */
  let rejected = false
  let registering: null | Promise<RelayCredentials | { error: RelayFailureCode }> = null
  let retryTimer: unknown = null
  let silenceTimer: unknown = null
  const streams = new Map<number, Stream>()
  /** Same (injectable) clock as every other wait of this client. */
  const sleep = (ms: number) => new Promise<void>(resolve => void setTimer(resolve, ms))

  function setState(status: RelayStatus, code: null | RelayFailureCode = null) {
    if (state.status === status && state.code === code) {
      return
    }

    state = { status, code }
    deps.log?.(`relay ${status}${code ? `:${code}` : ''}`)

    try {
      deps.onState?.(state)
    } catch {
      // a broken listener must not break the relay
    }
  }

  function clearTimers() {
    if (retryTimer !== null) {
      clearTimer(retryTimer)
      retryTimer = null
    }

    if (silenceTimer !== null) {
      clearTimer(silenceTimer)
      silenceTimer = null
    }
  }

  /** Stored registration, or a new one. One request at a time. */
  function register(): Promise<RelayCredentials | { error: RelayFailureCode }> {
    if (credentials) {
      return Promise.resolve(credentials)
    }

    registering ??= (async (): Promise<RelayCredentials | { error: RelayFailureCode }> => {
      let stored: null | RelayCredentials

      try {
        stored = await deps.credentials.load()
      } catch {
        return { error: 'vault_unavailable' }
      }

      if (isRelayCredentials(stored)) {
        credentials = stored

        return stored
      }

      const fresh = await registerRelaySession(deps.fetch, { sleep })

      if ('error' in fresh) {
        return fresh
      }

      try {
        await deps.credentials.save(fresh, { reissued: rejected })
      } catch {
        return { error: 'vault_unavailable' }
      }

      rejected = false
      credentials = fresh

      return fresh
    })().finally(() => {
      registering = null
    })

    return registering
  }

  function dropStreams() {
    for (const stream of streams.values()) {
      stream.local.destroy()
    }

    streams.clear()
  }

  function retry(code: RelayFailureCode, delay?: number) {
    const wait = delay ?? RELAY_BACKOFF_MS[Math.min(attempt, RELAY_BACKOFF_MS.length - 1)]

    attempt++
    setState('failed', code)
    retryTimer = setTimer(() => {
      retryTimer = null
      void connect()
    }, wait)
  }

  async function connect() {
    if (!wanted || socket) {
      return
    }

    const mine = ++generation

    clearTimers()
    setState('connecting')

    const current = await register()

    if (!wanted || mine !== generation) {
      return
    }

    if ('error' in current) {
      retry(current.error)

      return
    }

    let ws: RelaySocket

    try {
      ws = deps.connect(relayHostUrl(current.relayId), { Authorization: `Bearer ${current.hostToken}` })
    } catch {
      retry('connect_failed')

      return
    }

    socket = ws

    const send = (data: Buffer | string) => {
      if (socket === ws) {
        try {
          ws.send(data)
        } catch {
          // the close event follows
        }
      }
    }

    const control = (value: Record<string, unknown>) => send(JSON.stringify(value))

    const watch = () => {
      if (silenceTimer !== null) {
        clearTimer(silenceTimer)
      }

      silenceTimer = setTimer(() => {
        silenceTimer = null

        if (socket === ws) {
          try {
            ;(ws.terminate ?? ws.close).call(ws)
          } catch {
            // already gone
          }

          down(1006, 'timeout')
        }
      }, RELAY_SILENCE_MS)
    }

    const down = (code: number, reason?: RelayFailureCode) => {
      if (socket !== ws) {
        return
      }

      socket = null
      clearTimers()
      dropStreams()

      if (!wanted) {
        setState('off')

        return
      }

      // 4401: the token was refused. 4404 / HTTP 404: the gateway no longer knows this relay
      // (unused for 30 days). Either way the registration is dead: take a new one.
      if (code === 4401 || code === 4404) {
        credentials = null
        rejected = true
        void deps.credentials
          .clear()
          .catch(() => undefined)
          .then(() => {
            if (wanted && socket === null && retryTimer === null && mine === generation) {
              retry('token_rejected')
            }
          })

        return
      }

      if (code === 4409) {
        // Another IDE instance holds this registration: wait 30 s instead of fighting it.
        retry('replaced', RELAY_BACKOFF_MS[RELAY_BACKOFF_MS.length - 1])

        return
      }

      // 4429: the relay service is full; the normal growing delay applies.
      retry(reason ?? (code === 4429 ? 'busy' : 'connect_failed'))
    }

    // Gateway → local back-pressure: while any local connection is behind, stop reading the host socket.
    const stalled = new Set<Stream>()

    const unstall = (stream: Stream) => {
      if (stalled.delete(stream) && stalled.size === 0 && socket === ws) {
        ws.resume?.()
      }
    }

    /** Ends one stream on both sides. `tell` = the gateway does not know yet. */
    const endStream = (id: number, stream: Stream, code: number, tell: boolean) => {
      if (streams.get(id) !== stream) {
        return
      }

      streams.delete(id)
      unstall(stream)
      stream.local.destroy()

      if (tell) {
        control({ t: 'close', id, code })
      }
    }

    const openStream = (id: number) => {
      if (streams.has(id)) {
        return
      }

      if (streams.size >= RELAY_MAX_STREAMS) {
        return control({ t: 'close', id, code: 4429 })
      }

      let local: null | RelayLocalSocket

      try {
        local = deps.dial()
      } catch {
        local = null
      }

      if (!local) {
        return control({ t: 'close', id, code: 4503 })
      }

      const stream: Stream = { local, paused: false }
      let failed = false

      const live = () => streams.get(id) === stream

      // Local → gateway back-pressure: stop reading the proxy while the host socket is behind.
      const drain = () => {
        if (!live() || socket !== ws) {
          return
        }

        if ((ws.bufferedAmount ?? 0) > BUFFER_RESUME_BYTES) {
          setTimer(drain, BUFFER_POLL_MS)

          return
        }

        stream.paused = false
        local.resume()
      }

      streams.set(id, stream)
      local.on('connect', () => {
        if (live()) {
          control({ t: 'opened', id })
        }
      })
      local.on('data', (chunk: Buffer) => {
        if (!live()) {
          return
        }

        // Bytes are forwarded exactly as read: no parsing, no TLS here.
        for (const part of splitRelayChunks(chunk)) {
          send(encodeRelayFrame(id, part))
        }

        if (!stream.paused && (ws.bufferedAmount ?? 0) > RELAY_BUFFER_LIMIT_BYTES) {
          stream.paused = true
          local.pause()
          setTimer(drain, BUFFER_POLL_MS)
        }
      })
      local.on('drain', () => unstall(stream))
      local.on('error', () => {
        failed = true
      })
      local.on('close', () => endStream(id, stream, failed ? 1011 : 1000, true))
    }

    ws.on('open', () => {
      if (socket === ws) {
        attempt = 0
        setState('connected')
        watch()
      }
    })
    ws.on('close', (code: number) => down(Number(code) || 1006))
    ws.on('error', () => down(1006))
    // A refusal at the HTTP level (before the WebSocket exists) means the same as the close codes.
    ws.on('unexpected-response', (req: { destroy?: () => void }, res: { statusCode?: number; resume?: () => void }) => {
      const status = Number(res?.statusCode)

      res?.resume?.()
      req?.destroy?.()
      down(status === 401 || status === 403 ? 4401 : status === 404 ? 4404 : status === 429 ? 4429 : 1006)
    })
    ws.on('message', (data: unknown, isBinary?: boolean) => {
      if (socket !== ws) {
        return
      }

      watch()

      if (isBinary) {
        const frame = Buffer.isBuffer(data) ? data : Array.isArray(data) ? Buffer.concat(data) : Buffer.from(data as ArrayBuffer)

        if (frame.length <= RELAY_FRAME_HEADER_BYTES) {
          return
        }

        const id = frame.readUInt32BE(0)
        const stream = streams.get(id)

        // Late or unknown ids are ignored.
        if (!stream) {
          return
        }

        stream.local.write(frame.subarray(RELAY_FRAME_HEADER_BYTES))

        const pending = stream.local.writableLength ?? 0

        // Gateway → local: a proxy that does not read must not make this process hold the data.
        // First stop reading the gateway; if the backlog still reaches 1 MiB, give the stream up.
        if (pending > RELAY_BUFFER_LIMIT_BYTES) {
          endStream(id, stream, 4408, true)
        } else if (pending > BUFFER_RESUME_BYTES && ws.pause && !stalled.has(stream)) {
          stalled.add(stream)
          ws.pause()
        }

        return
      }

      let message: any

      try {
        message = JSON.parse(typeof data === 'string' ? data : Buffer.isBuffer(data) ? data.toString('utf8') : String(data))
      } catch {
        return
      }

      if (!message || typeof message !== 'object') {
        return
      }

      if (message.t === 'ping') {
        return control({ t: 'pong' })
      }

      if (!validStreamId(message.id)) {
        return
      }

      if (message.t === 'open') {
        return openStream(message.id)
      }

      if (message.t === 'close') {
        const stream = streams.get(message.id)

        if (stream) {
          endStream(message.id, stream, 1000, false)
        }
      }
    })
  }

  const api = {
    state: (): RelayState => state,

    /** Open phone streams right now. */
    streams: (): number => streams.size,

    /**
     * Makes sure a registration exists (stored or new) and returns its id; null
     * when the gateway or the vault is unavailable. Does not open the socket.
     */
    async ensureRegistered(): Promise<null | string> {
      const current = await register()

      return 'error' in current ? null : current.relayId
    },

    /**
     * "Röle kimliğini yenile": take a NEW registration. The old one is replaced
     * only once the new one is stored; false = nothing changed.
     */
    async renew(): Promise<boolean> {
      const fresh = await registerRelaySession(deps.fetch, { sleep })

      if ('error' in fresh) {
        return false
      }

      try {
        await deps.credentials.save(fresh, { reissued: true })
      } catch {
        return false
      }

      credentials = fresh
      rejected = false

      if (wanted) {
        // Drop the socket that was authenticated with the old token and come back with the new one.
        api.stop()
        api.start()
      }

      return true
    },

    start() {
      if (wanted) {
        return
      }

      wanted = true
      attempt = 0
      void connect()
    },

    stop() {
      wanted = false
      generation++
      clearTimers()

      const ws = socket

      socket = null
      dropStreams()

      if (ws) {
        try {
          ws.close(1000)
        } catch {
          // already gone
        }
      }

      setState('off')
    }
  }

  return api
}

export type RelayClient = ReturnType<typeof createRelayClient>