diff --git a/README.md b/README.md index d664d301..0b3e9503 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,7 @@ Configuration is provided via environment variables: |HTTP_IP| IP Address for API requests from jambonz-api-server |no| |JAMBONES_GATHER_EARLY_HINTS_MATCH| if true and hints are provided, gather will opportunistically review interim transcripts if possible to reduce ASR latency |no| |JAMBONES_FREESWITCH| IP:port:secret for Freeswitch server (e.g. '127.0.0.1:8021:JambonzR0ck$'|yes| +|JAMBONES_HOLD_UNHOLD_EVENTS| if set, surface in-dialog hold/un-hold: emits a 'hold'/'unhold' event and, when sipRequestWithinDialogHook is configured, delivers it to the application. Disabled when unset|no| |JAMBONES_LOGLEVEL| log level for application, 'info' or 'debug'|no| |JAMBONES_MYSQL_HOST| mysql host|yes| |JAMBONES_MYSQL_USER| mysql username|yes| diff --git a/lib/config.js b/lib/config.js index c8c7fdf2..806d3834 100644 --- a/lib/config.js +++ b/lib/config.js @@ -71,6 +71,7 @@ const OTEL_EXPORTER_COLLECTOR_URL = process.env.OTEL_EXPORTER_COLLECTOR_URL; const JAMBONES_LOGLEVEL = process.env.JAMBONES_LOGLEVEL || 'info'; const JAMBONES_INJECT_CONTENT = process.env.JAMBONES_INJECT_CONTENT; +const JAMBONES_HOLD_UNHOLD_EVENTS = process.env.JAMBONES_HOLD_UNHOLD_EVENTS; const PORT = parseInt(process.env.HTTP_PORT, 10) || 3000; const HTTP_IP = process.env.HTTP_IP; @@ -170,6 +171,7 @@ module.exports = { JAMBONES_API_BASE_URL, JAMBONES_TIME_SERIES_HOST, JAMBONES_INJECT_CONTENT, + JAMBONES_HOLD_UNHOLD_EVENTS, JAMBONES_EAGERLY_PRE_CACHE_AUDIO, JAMBONES_ESL_LISTEN_ADDRESS, JAMBONES_SBCS, diff --git a/lib/session/call-session.js b/lib/session/call-session.js index 1469cd84..6499caff 100644 --- a/lib/session/call-session.js +++ b/lib/session/call-session.js @@ -29,6 +29,7 @@ const {parseUri} = require('drachtio-srf'); const { JAMBONES_INJECT_CONTENT, JAMBONES_EAGERLY_PRE_CACHE_AUDIO, + JAMBONES_HOLD_UNHOLD_EVENTS, AWS_REGION, } = require('../config'); const bent = require('bent'); @@ -38,6 +39,7 @@ const BADPRECONDITIONS = 'preconditions not met'; const CALLER_CANCELLED_ERR_MSG = 'Response not sent due to unknown transaction'; const { NonFatalTaskError} = require('../utils/error'); const { createMediaEndpoint } = require('../utils/media-endpoint'); +const { isOnhold } = require('../utils/sdp-utils'); const SttLatencyCalculator = require('../utils/stt-latency-calculator'); const sqlRetrieveQueueEventHook = `SELECT * FROM webhooks WHERE webhook_sid = @@ -2793,6 +2795,7 @@ Duration=${duration} ` const newSdp = await this.ep.modify(req.body); res.send(200, {body: newSdp}); this.logger.info({offer: req.body, answer: newSdp}, 'handling reINVITE'); + this._notifyHoldState(req); } } } @@ -2810,6 +2813,43 @@ Duration=${duration} ` } } + /** + * Detect a hold/un-hold transition on an in-dialog re-INVITE and notify the + * application (reusing the sipRequestWithinDialogHook channel). Also emits a + * local 'hold'/'unhold' event so in-process tasks (e.g. transcribe) can react. + * Opt-in: only runs when JAMBONES_HOLD_UNHOLD_EVENTS is set. + */ + _notifyHoldState(req) { + if (!JAMBONES_HOLD_UNHOLD_EVENTS) return; + const onHold = isOnhold(req.body); + const wasOnHold = !!this._onHold; + if (onHold === wasOnHold) return; // no state change -> nothing to do + this._onHold = onHold; + const event = onHold ? 'hold' : 'unhold'; + this.logger.info({event, callSid: this.callSid}, 'CallSession: hold state changed'); + + // in-process signal (used by the optional transcribe auto-pause) + this.emit(event); + + // notify the application over the in-dialog SIP hook, if configured. + // Delivered via requestor.request('verb:hook', ...) — the same channel + // _handleRefer uses — rather than currentTask.performHook, so the event is + // not dropped when the current task has ended (e.g. transcribe failed). + // callInfo.toJSON() is merged in manually so call_sid/account_sid are present. + if (this.sipRequestWithinDialogHook && this.requestor) { + const params = { + ...(this.callInfo.toJSON()), + sip_method: 'INVITE', + event, // 'hold' | 'unhold' + on_hold: onHold, + sip_body: req.body, + sip_headers: req.headers + }; + this.requestor.request('verb:hook', this.sipRequestWithinDialogHook, params) + .catch((err) => this.logger.error({err}, 'CallSession:_notifyHoldState - hook error')); + } + } + /** * Handle incoming REFER * @param {*} req diff --git a/lib/session/siprec-call-session.js b/lib/session/siprec-call-session.js index d796bdd4..43d61d9a 100644 --- a/lib/session/siprec-call-session.js +++ b/lib/session/siprec-call-session.js @@ -2,6 +2,7 @@ const InboundCallSession = require('./inbound-call-session'); const {createSipRecPayload} = require('../utils/siprec-utils'); const {CallStatus} = require('../utils/constants'); const {parseSiprecPayload} = require('../utils/siprec-utils'); +const {JAMBONES_HOLD_UNHOLD_EVENTS} = require('../config'); /** * @classdesc Subclass of InboundCallSession. This represents a CallSession that is * established for an inbound SIPREC call. @@ -33,6 +34,7 @@ class SipRecCallSession extends InboundCallSession { const combinedSdp = await createSipRecPayload(newSdp1, newSdp2, this.logger); res.send(200, {body: combinedSdp}); this.logger.info({offer: req.body, answer: combinedSdp}, 'SipRec handling reINVITE'); + this._notifySipRecReinvite(req, this.sdp1, this.sdp2); } else { this.logger.info('got SipRec reINVITE but no endpoint and media has not been released'); @@ -43,6 +45,43 @@ class SipRecCallSession extends InboundCallSession { } } + /** + * Notify the application of a SIPREC re-INVITE. SIPREC recording streams are + * sendonly by nature, so a stream flipping to a=inactive is treated as the + * (best-effort) hold signal; the full multipart body (SDP + rs-metadata) is + * always forwarded so the app can determine hold from the metadata itself. + * Delivered over the existing sipRequestWithinDialogHook channel; also emits + * a local 'hold'/'unhold' on transition (for optional transcribe auto-pause). + * Opt-in: only runs when JAMBONES_HOLD_UNHOLD_EVENTS is set. + */ + _notifySipRecReinvite(req, sdp1, sdp2) { + if (!JAMBONES_HOLD_UNHOLD_EVENTS) return; + const onHold = /a=inactive/.test(sdp1 || '') || /a=inactive/.test(sdp2 || ''); + const changed = onHold !== !!this._onHold; + this._onHold = onHold; + if (changed) { + const event = onHold ? 'hold' : 'unhold'; + this.logger.info({event, callSid: this.callSid}, 'SipRecCallSession: hold state changed'); + this.emit(event); + } + // Delivered via requestor.request('verb:hook', ...) — task-independent, so + // the event survives even if the current task has ended (e.g. transcribe + // failed on a speech-credential error). callInfo.toJSON() merged in manually. + if (this.sipRequestWithinDialogHook && this.requestor) { + const params = { + ...(this.callInfo.toJSON()), + sip_method: 'INVITE', + siprec: true, + event: changed ? (onHold ? 'hold' : 'unhold') : 'reinvite', + on_hold: onHold, + sip_body: req.body, // full multipart: SDP parts + rs-metadata XML + sip_headers: req.headers + }; + this.requestor.request('verb:hook', this.sipRequestWithinDialogHook, params) + .catch((err) => this.logger.error({err}, 'SipRecCallSession:_notifySipRecReinvite - hook error')); + } + } + async answerSipRecCall() { try { let remoteSdp = this.sdp1.replace(/sendonly/, 'sendrecv'); diff --git a/test/hold-unhold-test.js b/test/hold-unhold-test.js new file mode 100644 index 00000000..417f2f52 --- /dev/null +++ b/test/hold-unhold-test.js @@ -0,0 +1,176 @@ +const test = require('tape'); +const clearModule = require('clear-module'); + +// ENCRYPTION_SECRET is required when loading the session modules (encrypt-decrypt.js +// hashes it at require time). The npm `test` script sets it; guard here so the file +// can also be run standalone. +process.env.ENCRYPTION_SECRET = process.env.ENCRYPTION_SECRET || 'foobar'; + +// The feature is opt-in behind JAMBONES_HOLD_UNHOLD_EVENTS, which lib/config.js +// captures ONCE at load time. When this file runs inside the full suite (test/index.js), +// config.js and the session modules have already been required by earlier tests with the +// flag unset, so simply setting the env var here would be too late. loadWith() sets the +// flag and force-reloads config + the session modules so they re-read it, regardless of +// what loaded them first. +const ORIGINAL_FLAG = process.env.JAMBONES_HOLD_UNHOLD_EVENTS; +function loadWith(value) { + if (value === undefined) delete process.env.JAMBONES_HOLD_UNHOLD_EVENTS; + else process.env.JAMBONES_HOLD_UNHOLD_EVENTS = value; + clearModule('../lib/config'); + clearModule('../lib/session/call-session'); + clearModule('../lib/session/siprec-call-session'); + return { + CallSession: require('../lib/session/call-session'), + SipRecCallSession: require('../lib/session/siprec-call-session') + }; +} + +// enabled for the behavioural tests +const {CallSession, SipRecCallSession} = loadWith('1'); + +const noop = () => {}; + +// A minimal stand-in for a CallSession/SipRecCallSession instance, exposing only +// what _notifyHoldState / _notifySipRecReinvite touch. We invoke the real prototype +// methods against it so the hold-transition logic is exercised without spinning up +// drachtio/freeswitch. +function makeSession(overrides = {}) { + const emitted = []; + const hookCalls = []; + return { + _onHold: undefined, + callSid: 'call-sid-abc', + logger: {info: noop, error: noop, debug: noop}, + sipRequestWithinDialogHook: '/hold-events', + callInfo: {toJSON: () => ({call_sid: 'call-sid-abc', account_sid: 'acct-sid-xyz'})}, + requestor: { + request: (type, hook, params) => { + hookCalls.push({type, hook, params}); + return Promise.resolve(); + } + }, + emit: (event) => emitted.push(event), + emitted, + hookCalls, + ...overrides + }; +} + +const HOLD_SDP = 'v=0\r\no=- 1 1 IN IP4 1.1.1.1\r\ns=-\r\nc=IN IP4 1.1.1.1\r\nt=0 0\r\n' + + 'm=audio 4000 RTP/AVP 0\r\na=rtpmap:0 PCMU/8000\r\na=sendonly\r\n'; +const INACTIVE_SDP = HOLD_SDP.replace('a=sendonly', 'a=inactive'); +const ACTIVE_SDP = HOLD_SDP.replace('a=sendonly', 'a=sendrecv'); + +const notifyHoldState = CallSession.prototype._notifyHoldState; +const notifySipRecReinvite = SipRecCallSession.prototype._notifySipRecReinvite; + +test('CallSession._notifyHoldState - fires on hold transition', (t) => { + const s = makeSession(); + notifyHoldState.call(s, {body: HOLD_SDP, headers: {'call-id': 'x'}}); + + t.equal(s._onHold, true, 'tracks on-hold state'); + t.deepEqual(s.emitted, ['hold'], 'emits a single "hold" event'); + t.equal(s.hookCalls.length, 1, 'invokes the in-dialog hook once'); + + const {type, hook, params} = s.hookCalls[0]; + t.equal(type, 'verb:hook', 'delivered over the task-independent verb:hook channel'); + t.equal(hook, '/hold-events', 'uses the configured sipRequestWithinDialogHook'); + t.equal(params.event, 'hold', 'event = hold'); + t.equal(params.on_hold, true, 'on_hold = true'); + t.equal(params.sip_method, 'INVITE', 'sip_method = INVITE'); + t.equal(params.sip_body, HOLD_SDP, 'forwards the offered SDP'); + t.equal(params.call_sid, 'call-sid-abc', 'callInfo is merged in (call_sid present)'); + t.equal(params.account_sid, 'acct-sid-xyz', 'callInfo is merged in (account_sid present)'); + t.end(); +}); + +test('CallSession._notifyHoldState - a=inactive also counts as hold', (t) => { + const s = makeSession(); + notifyHoldState.call(s, {body: INACTIVE_SDP, headers: {}}); + t.equal(s._onHold, true, 'a=inactive treated as hold'); + t.deepEqual(s.emitted, ['hold'], 'emits "hold"'); + t.end(); +}); + +test('CallSession._notifyHoldState - no-op when hold state is unchanged', (t) => { + const s = makeSession({_onHold: true}); + notifyHoldState.call(s, {body: HOLD_SDP, headers: {}}); + t.equal(s.emitted.length, 0, 'does not re-emit when already on hold'); + t.equal(s.hookCalls.length, 0, 'does not re-invoke the hook when already on hold'); + t.end(); +}); + +test('CallSession._notifyHoldState - fires "unhold" on retrieve', (t) => { + const s = makeSession({_onHold: true}); + notifyHoldState.call(s, {body: ACTIVE_SDP, headers: {}}); + t.equal(s._onHold, false, 'clears on-hold state'); + t.deepEqual(s.emitted, ['unhold'], 'emits "unhold"'); + t.equal(s.hookCalls[0].params.event, 'unhold', 'hook event = unhold'); + t.equal(s.hookCalls[0].params.on_hold, false, 'on_hold = false'); + t.end(); +}); + +test('CallSession._notifyHoldState - still emits locally when no hook configured', (t) => { + const s = makeSession({sipRequestWithinDialogHook: undefined}); + notifyHoldState.call(s, {body: HOLD_SDP, headers: {}}); + t.deepEqual(s.emitted, ['hold'], 'local event still emitted'); + t.equal(s.hookCalls.length, 0, 'no hook invoked when unconfigured'); + t.end(); +}); + +test('SipRecCallSession._notifySipRecReinvite - detects hold via a=inactive on either stream', (t) => { + const s = makeSession(); + notifySipRecReinvite.call(s, {body: 'multipart-body', headers: {}}, ACTIVE_SDP, INACTIVE_SDP); + + t.equal(s._onHold, true, 'a=inactive on stream 2 => hold'); + t.deepEqual(s.emitted, ['hold'], 'emits "hold" on transition'); + t.equal(s.hookCalls.length, 1, 'invokes the hook'); + + const {params} = s.hookCalls[0]; + t.equal(params.siprec, true, 'marks payload as siprec'); + t.equal(params.event, 'hold', 'event = hold on transition'); + t.equal(params.on_hold, true, 'on_hold = true'); + t.equal(params.sip_body, 'multipart-body', 'forwards the full multipart body'); + t.end(); +}); + +test('SipRecCallSession._notifySipRecReinvite - non-transition reinvite still forwarded as event:reinvite', (t) => { + const s = makeSession({_onHold: false}); + notifySipRecReinvite.call(s, {body: 'multipart-body', headers: {}}, ACTIVE_SDP, ACTIVE_SDP); + + t.equal(s._onHold, false, 'stays off hold'); + t.equal(s.emitted.length, 0, 'no local event when state is unchanged'); + t.equal(s.hookCalls.length, 1, 'hook still invoked (SIPREC forwards every reinvite)'); + t.equal(s.hookCalls[0].params.event, 'reinvite', 'event = reinvite when there is no transition'); + t.end(); +}); + +test('SipRecCallSession._notifySipRecReinvite - emits "unhold" on retrieve transition', (t) => { + const s = makeSession({_onHold: true}); + notifySipRecReinvite.call(s, {body: 'multipart-body', headers: {}}, ACTIVE_SDP, ACTIVE_SDP); + + t.equal(s._onHold, false, 'clears on-hold state'); + t.deepEqual(s.emitted, ['unhold'], 'emits "unhold" on transition'); + t.equal(s.hookCalls[0].params.event, 'unhold', 'hook event = unhold'); + t.end(); +}); + +test('feature is inert unless JAMBONES_HOLD_UNHOLD_EVENTS is set', (t) => { + // reload the modules with the flag unset to exercise the disabled path + const {CallSession: CS, SipRecCallSession: SR} = loadWith(undefined); + try { + const s1 = makeSession(); + CS.prototype._notifyHoldState.call(s1, {body: HOLD_SDP, headers: {}}); + t.equal(s1.emitted.length, 0, 'CallSession: no event emitted when disabled'); + t.equal(s1.hookCalls.length, 0, 'CallSession: no hook invoked when disabled'); + + const s2 = makeSession(); + SR.prototype._notifySipRecReinvite.call(s2, {body: 'multipart-body', headers: {}}, ACTIVE_SDP, INACTIVE_SDP); + t.equal(s2.emitted.length, 0, 'SipRecCallSession: no event emitted when disabled'); + t.equal(s2.hookCalls.length, 0, 'SipRecCallSession: no hook invoked when disabled'); + } finally { + // restore the ambient flag + module cache for the rest of the suite + loadWith(ORIGINAL_FLAG); + } + t.end(); +}); diff --git a/test/index.js b/test/index.js index 3228a410..46f525a4 100644 --- a/test/index.js +++ b/test/index.js @@ -4,6 +4,7 @@ require('./ws-requestor-unit-test'); require('./http-requestor-retry-test'); require('./http-requestor-unit-test'); require('./unit-tests'); +require('./hold-unhold-test'); require('./docker_start'); require('./create-test-db'); require('./account-validation-tests');