/** * Minimal Asterisk Manager Interface (AMI) client over a raw TCP socket. * * Protocol: text packets of `Key: Value\r\n` lines terminated by an empty * line. The first packet is the banner (`Asterisk Call Manager/x.y`), then * responses (`Response: Success|Error`, matched to our `ActionID`) and events * (`Event: …`) interleave on the same socket. * * Deliberately tiny: one socket per call session (see callTracker.ts) — no * reconnect logic, no shared connection, no global state. Every wait has a * timeout so a stalled PBX can never leave a promise pending forever. */ import { createConnection, type Socket } from 'node:net' import { EventEmitter } from 'node:events' import { randomUUID } from 'node:crypto' export type AmiPacket = Record export interface AmiClientOptions { host: string port: number username: string secret: string /** TCP connect timeout (ms). Default 5000. */ connectTimeoutMs?: number /** Login response timeout (ms). Default 5000. */ loginTimeoutMs?: number } export interface AmiClient { /** Send an action and wait for the matching response packet (by ActionID). * Pass `actionId` to correlate later events (e.g. OriginateResponse) yourself. */ send: (action: string, headers?: Record, timeoutMs?: number, actionId?: string) => Promise /** Register an event listener (`Event:` packets). Returns an unsubscribe fn. */ onEvent: (fn: (evt: AmiPacket) => void) => () => void /** Fires once when the socket is gone (error or remote close). */ onClosed: (fn: (reason: string) => void) => void /** Graceful logoff + destroy. Safe to call multiple times. */ close: () => void readonly closed: boolean } /** Returns the first value of a (possibly repeated) header. */ export const amiValue = (pkt: AmiPacket | null | undefined, key: string): string => { const v = pkt?.[key] if (Array.isArray(v)) return v[0] ?? '' return v ?? '' } /** Returns all values of a (possibly repeated) header. */ export const amiValues = (pkt: AmiPacket | null | undefined, key: string): string[] => { const v = pkt?.[key] if (Array.isArray(v)) return v return v != null ? [v] : [] } /** * Reads a channel variable exposed via manager.conf `channelvars=…`, in the * formats Asterisk has used: * `ChanVariable: NAME=value` (Asterisk ≥ 12, one header per var) * `ChanVariable(NAME): value` (older builds) * Two-channel events (DialBegin/DialEnd/…) render the destination side as * `DestChanVariable`; pass prefix 'Dest' for that. */ export const amiChanVariable = (pkt: AmiPacket, name: string, prefix: '' | 'Dest' = ''): string | null => { const key = `${prefix}ChanVariable` const direct = pkt[`${key}(${name})`] if (direct != null) return Array.isArray(direct) ? (direct[0] ?? '') : direct for (const line of amiValues(pkt, key)) { const idx = line.indexOf('=') if (idx > 0 && line.slice(0, idx) === name) return line.slice(idx + 1) } return null } const parsePacket = (raw: string): AmiPacket => { const pkt: AmiPacket = {} for (const line of raw.split('\r\n')) { const idx = line.indexOf(':') if (idx <= 0) continue const key = line.slice(0, idx).trim() const value = line.slice(idx + 1).trim() const existing = pkt[key] if (existing === undefined) pkt[key] = value else if (Array.isArray(existing)) existing.push(value) else pkt[key] = [existing, value] } return pkt } export const connectAmi = (opts: AmiClientOptions): Promise => { return new Promise((resolve, reject) => { const connectTimeoutMs = opts.connectTimeoutMs ?? 5000 const loginTimeoutMs = opts.loginTimeoutMs ?? 5000 const emitter = new EventEmitter() emitter.setMaxListeners(0) const pending = new Map void; reject: (e: Error) => void; timer: NodeJS.Timeout }>() let buffer = '' let bannerSeen = false let closed = false let settled = false let closeReason = '' const socket: Socket = createConnection({ host: opts.host, port: opts.port }) socket.setNoDelay(true) socket.setKeepAlive(true, 15000) const failAllPending = (err: Error) => { for (const [id, p] of pending) { clearTimeout(p.timer) p.reject(err) pending.delete(id) } } const markClosed = (reason: string) => { if (closed) return closed = true closeReason = reason failAllPending(new Error(`AMI connection closed (${reason})`)) emitter.emit('closed', reason) if (!settled) { settled = true reject(new Error(`AMI ${reason}`)) } } const connectTimer = setTimeout(() => { if (!settled) { socket.destroy() markClosed(`connect timeout after ${connectTimeoutMs} ms`) } }, connectTimeoutMs) connectTimer.unref?.() const writeAction = (action: string, headers: Record, actionId: string) => { let out = `Action: ${action}\r\nActionID: ${actionId}\r\n` for (const [k, v] of Object.entries(headers)) { if (v === undefined || v === null || v === '') continue // Header values must be single-line; strip CR/LF defensively. out += `${k}: ${String(v).replace(/[\r\n]+/g, ' ')}\r\n` } out += '\r\n' socket.write(out) } const send = (action: string, headers: Record = {}, timeoutMs = 10000, explicitActionId?: string): Promise => { if (closed) return Promise.reject(new Error(`AMI connection closed (${closeReason})`)) const actionId = explicitActionId || randomUUID() return new Promise((res, rej) => { const timer = setTimeout(() => { pending.delete(actionId) rej(new Error(`AMI ${action}: no response within ${timeoutMs} ms`)) }, timeoutMs) timer.unref?.() pending.set(actionId, { resolve: res, reject: rej, timer }) try { writeAction(action, headers, actionId) } catch (err: any) { clearTimeout(timer) pending.delete(actionId) rej(err) } }) } const handlePacket = (pkt: AmiPacket) => { const actionId = amiValue(pkt, 'ActionID') if (pkt.Event === undefined && pkt.Response !== undefined && actionId && pending.has(actionId)) { const p = pending.get(actionId)! clearTimeout(p.timer) pending.delete(actionId) p.resolve(pkt) return } if (pkt.Event !== undefined) emitter.emit('event', pkt) } socket.on('data', (chunk) => { buffer += chunk.toString('utf8') if (!bannerSeen) { const nl = buffer.indexOf('\r\n') if (nl === -1) return bannerSeen = true buffer = buffer.slice(nl + 2) } let sep = buffer.indexOf('\r\n\r\n') while (sep !== -1) { const raw = buffer.slice(0, sep) buffer = buffer.slice(sep + 4) if (raw.trim()) { try { handlePacket(parsePacket(raw)) } catch (err: any) { console.warn('[ami] packet handling error:', err?.message || err) } } sep = buffer.indexOf('\r\n\r\n') } }) socket.on('error', (err) => markClosed(`socket error: ${err.message}`)) socket.on('close', () => markClosed(closeReason || 'remote close')) socket.on('connect', async () => { clearTimeout(connectTimer) try { const res = await send('Login', { Username: opts.username, Secret: opts.secret, Events: 'on' }, loginTimeoutMs) if (amiValue(res, 'Response') !== 'Success') { throw new Error(`AMI login failed: ${amiValue(res, 'Message') || 'unknown error'}`) } settled = true resolve(client) } catch (err: any) { settled = true socket.destroy() closed = true reject(err instanceof Error ? err : new Error(String(err))) } }) const client: AmiClient = { send, onEvent: (fn) => { emitter.on('event', fn) return () => emitter.off('event', fn) }, onClosed: (fn) => { emitter.once('closed', fn) }, close: () => { if (closed) return try { socket.write('Action: Logoff\r\n\r\n') } catch { /* ignore */ } setTimeout(() => { try { socket.destroy() } catch { /* ignore */ } }, 200) markClosed('closed by client') }, get closed() { return closed } } }) }