// Copyright (c) 2020 by Juliusz Chroboczek.

// Permission is hereby granted, free of charge, to any person obtaining a copy

// of this software and associated documentation files (the "Software"), to deal

// in the Software without restriction, including without limitation the rights

// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell

// copies of the Software, and to permit persons to whom the Software is

// furnished to do so, subject to the following conditions:

//

// The above copyright notice and this permission notice shall be included in

// all copies or substantial portions of the Software.

//

// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR

// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,

// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE

// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER

// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,

// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN

// THE SOFTWARE.

'use strict';

/**

* toHex formats an array as a hexadecimal string.

*

* @param {number[]|Uint8Array} array - the array to format

* @returns {string} - the hexadecimal representation of array

*/

function toHex(array) {

let a = new Uint8Array(array);

function hex(x) {

let h = x.toString(16);

if(h.length < 2)

h = '0' + h;

return h;

}

return a.reduce((x, y) => x + hex(y), '');

}

/**

* newRandomId returns a random string of 32 hex digits (16 bytes).

*

* @returns {string}

*/

function newRandomId() {

let a = new Uint8Array(16);

crypto.getRandomValues(a);

return toHex(a);

}

let localIdCounter = 0;

/**

* newLocalId returns a string that is unique in this session.

*

* @returns {string}

*/

function newLocalId() {

let id = `${localIdCounter}`

localIdCounter++;

return id;

}

/**

* @typedef {Object} user

* @property {string} username

* @property {Array<string>} permissions

* @property {Object<string,any>} data

* @property {Object<string,Object<string,boolean>>} streams

*/

/**

* ServerConnection encapsulates a websocket connection to the server and

* all the associated streams.

* @constructor

*/

function ServerConnection() {

/**

* The id of this connection.

*

* @type {string}

* @const

*/

this.id = newRandomId();

/**

* The group that we have joined, or null if we haven't joined yet.

*

* @type {string}

*/

this.group = null;

/**

* The username we joined as.

*

* @type {string}

*/

this.username = null;

/**

* The set of users in this group, including ourself.

*

* @type {Object<string,user>}

*/

this.users = {};

/**

* The underlying websocket.

*

* @type {WebSocket}

*/

this.socket = null;

/**

* The negotiated protocol version.

*

* @type {string}

*/

this.version = null;

/**

* The set of all up streams, indexed by their id.

*

* @type {Object<string,Stream>}

*/

this.up = {};

/**

* The set of all down streams, indexed by their id.

*

* @type {Object<string,Stream>}

*/

this.down = {};

/**

* The ICE configuration used by all associated streams.

*

* @type {RTCConfiguration}

*/

this.rtcConfiguration = null;

/**

* The permissions granted to this connection.

*

* @type {Array<string>}

*/

this.permissions = [];

/**

* userdata is a convenient place to attach data to a ServerConnection.

* It is not used by the library.

*

* @type {Object<unknown,unknown>}

*/

this.userdata = {};

/**

* The time at which we last received a message from the server.

*

* @type {number}

*/

this.lastServerMessage = null;

/**

* The interval handler which checks for liveness.

*

* @type {number}

*/

this.pingHandler = null;

/* Callbacks */

/**

* onconnected is called when the connection has been established

*

* @type{(this: ServerConnection) => void}

*/

this.onconnected = null;

/**

* onerror is called whenever a fatal error occurs. The stream will

* then be closed, and onclose called normally.

*

* @type{(this: ServerConnection, error: unknown) => void}

*/

this.onerror = null;

/**

* onclose is called when the connection is closed

*

* @type{(this: ServerConnection, code: number, reason: string) => void}

*/

this.onclose = null;

/**

* onpeerconnection is called before we establish a new peer connection.

* It may either return null, or a new RTCConfiguration that overrides

* the value obtained from the server.

*

* @type{(this: ServerConnection) => RTCConfiguration}

*/

this.onpeerconnection = null;

/**

* onuser is called whenever a user in the group changes. The users

* array has already been updated.

*

* @type{(this: ServerConnection, id: string, kind: string) => void}

*/

this.onuser = null;

/**

* onjoined is called whenever we join or leave a group or whenever the

* permissions we have in a group change.

*

* kind is one of 'join', 'fail', 'change' or 'leave'.

*

* @type{(this: ServerConnection, kind: string, group: string, permissions: Array<string>, status: Object<string,any>, data: Object<string,any>, error: string, message: string) => void}

*/

this.onjoined = null;

/**

* ondownstream is called whenever a new down stream is added. It

* should set up the stream's callbacks; actually setting up the UI

* should be done in the stream's ondowntrack callback.

*

* @type{(this: ServerConnection, stream: Stream) => void}

*/

this.ondownstream = null;

/**

* onchat is called whenever a new chat message is received.

*

* @type {(this: ServerConnection, id: string, source: string, dest: string, username: string, time: Date, privileged: boolean, history: boolean, kind: string, message: string) => void}

*/

this.onchat = null;

/**

* onusermessage is called when an application-specific message is

* received. Id is null when the message originated at the server,

* a user-id otherwise.

*

* 'kind' is typically one of 'error', 'warning', 'info' or 'mute'. If

* 'id' is non-null, 'privileged' indicates whether the message was

* sent by an operator.

*

* @type {(this: ServerConnection, id: string, dest: string, username: string, time: Date, privileged: boolean, kind: string, error: string, message: unknown) => void}

*/

this.onusermessage = null;

/**

* The set of files currently being transferred.

*

* @type {Object<string,TransferredFile>}

*/

this.transferredFiles = {};

/**

* onfiletransfer is called whenever a peer offers a file transfer.

*

* If the transfer is accepted, it should set up the file transfer

* callbacks and return immediately. It may also throw an exception

* in order to reject the file transfer.

*

* @type {(this: ServerConnection, f: TransferredFile) => void}

*/

this.onfiletransfer = null;

}

/**

* @typedef {Object} message

* @property {string} type

* @property {Array<string>} [version]

* @property {string} [kind]

* @property {string} [error]

* @property {string} [id]

* @property {string} [replace]

* @property {string} [source]

* @property {string} [dest]

* @property {string} [username]

* @property {string} [password]

* @property {string} [token]

* @property {boolean} [privileged]

* @property {Array<string>} [permissions]

* @property {Object<string,any>} [status]

* @property {Object<string,any>} [data]

* @property {string} [group]

* @property {unknown} [value]

* @property {boolean} [noecho]

* @property {string|number} [time]

* @property {string} [sdp]

* @property {RTCIceCandidate} [candidate]

* @property {string} [label]

* @property {Object<string,Array<string>>|Array<string>} [request]

* @property {Object<string,any>} [rtcConfiguration]

*/

/**

* close forcibly closes a server connection. The onclose callback will

* be called when the connection is effectively closed.

*/

ServerConnection.prototype.close = function() {

this.socket && this.socket.close(1000, 'Close requested by client');

this.socket = null;

};

/**

* error forcibly closes a server connection and invokes the onerror

* callback. The onclose callback will be invoked when the connection

* is effectively closed.

*

* @param {any} e

*/

ServerConnection.prototype.error = function(e) {

if(this.onerror)

this.onerror.call(this, e);

this.close();

};

/**

* send sends a message to the server.

* @param {message} m - the message to send.

*/

ServerConnection.prototype.send = function(m) {

if(!this.socket || this.socket.readyState !== this.socket.OPEN) {

// send on a closed socket doesn't throw

throw(new Error('Connection is not open'));

}

return this.socket.send(JSON.stringify(m));

};

/**

* connect connects to the server.

*

* @param {string} url - The URL to connect to.

* @function

*/

ServerConnection.prototype.connect = function(url) {

let sc = this;

if(sc.socket)

throw new Error("Attempting to connect stale connection");

sc.socket = new WebSocket(url);

this.pingHandler = setInterval(e => {

if(!sc.lastServerMessage) {

sc.error(new Error('Timeout'));

return;

}

let d = new Date().valueOf() - sc.lastServerMessage;

if(d > 65000) {

sc.error(new Error('Timeout'));

return;

}

if(sc.version && d >= 15000)

sc.send({type: 'ping'});

}, 10000);

this.socket.onerror = function(e) {

if(sc.onerror)

sc.onerror.call(sc, new Error('Socket error: ' + e));

};

this.socket.onopen = function(e) {

try {

sc.send({

type: 'handshake',

version: ['2'],

id: sc.id,

});

} catch(e) {

sc.error(e);

return;

}

};

this.socket.onclose = function(e) {

sc.permissions = [];

for(let id in sc.up) {

let c = sc.up[id];

c.close();

}

for(let id in sc.down) {

let c = sc.down[id];

c.close();

}

for(let id in sc.users) {

delete(sc.users[id]);

if(sc.onuser)

sc.onuser.call(sc, id, 'delete');

}

if(sc.group && sc.onjoined)

sc.onjoined.call(sc, 'leave', sc.group, [], {}, {}, '', '');

sc.group = null;

sc.username = null;

if(sc.pingHandler) {

clearInterval(sc.pingHandler);

sc.pingHandler = null;

}

if(sc.onclose)

sc.onclose.call(sc, e.code, e.reason);

};

this.socket.onmessage = function(e) {

let m;

try {

m = JSON.parse(e.data);

} catch(e) {

sc.error(e);

return;

}

if(m.type !== 'handshake' && !sc.version) {

sc.error(new Error("Server didn't send handshake"));

return;

}

sc.lastServerMessage = new Date().valueOf();

switch(m.type) {

case 'handshake': {

if((m.version instanceof Array) && m.version.includes('2')) {

sc.version = '2';

} else {

sc.version = null;

sc.error(new Error(`Unknown protocol version ${m.version}`));

return;

}

if(sc.onconnected)

sc.onconnected.call(sc);

break;

}

case 'offer':

sc.gotOffer(m.id, m.label, m.source, m.username,

m.sdp, m.replace);

break;

case 'answer':

sc.gotAnswer(m.id, m.sdp);

break;

case 'renegotiate':

sc.gotRenegotiate(m.id);

break;

case 'close':

sc.gotClose(m.id);

break;

case 'abort':

sc.gotAbort(m.id);

break;

case 'ice':

sc.gotRemoteIce(m.id, m.candidate);

break;

case 'joined':

if(m.kind === 'leave' || m.kind === 'fail') {

for(let id in sc.users) {

delete(sc.users[id]);

if(sc.onuser)

sc.onuser.call(sc, id, 'delete');

}

sc.username = null;

sc.permissions = [];

sc.rtcConfiguration = null;

} else if(m.kind === 'join' || m.kind == 'change') {

if(m.kind === 'join' && sc.group) {

throw new Error('Joined multiple groups');

} else if(m.kind === 'change' && m.group != sc.group) {

console.warn('join(change) for inconsistent group');

break;

}

sc.group = m.group;

sc.username = m.username;

sc.permissions = m.permissions || [];

sc.rtcConfiguration = m.rtcConfiguration || null;

}

if(sc.onjoined)

sc.onjoined.call(sc, m.kind, m.group,

m.permissions || [],

m.status, m.data,

m.error || null, m.value || null);

break;

case 'user':

switch(m.kind) {

case 'add':

if(m.id in sc.users)

console.warn(`Duplicate user ${m.id} ${m.username}`);

sc.users[m.id] = {

username: m.username,

permissions: m.permissions || [],

data: m.data || {},

streams: {},

};

break;

case 'change':

if(!(m.id in sc.users)) {

console.warn(`Unknown user ${m.id} ${m.username}`);

sc.users[m.id] = {

username: m.username,

permissions: m.permissions || [],

data: m.data || {},

streams: {},

};

} else {

sc.users[m.id].username = m.username;

sc.users[m.id].permissions = m.permissions || [];

sc.users[m.id].data = m.data || {};

}

break;

case 'delete':

if(!(m.id in sc.users))

console.warn(`Unknown user ${m.id} ${m.username}`);

for(let t in sc.transferredFiles) {

let f = sc.transferredFiles[t];

if(f.userid === m.id)

f.fail('user has gone away');

}

delete(sc.users[m.id]);

break;

default:

console.warn(`Unknown user action ${m.kind}`);

return;

}

if(sc.onuser)

sc.onuser.call(sc, m.id, m.kind);

break;

case 'chat':

case 'chathistory':

if(sc.onchat)

sc.onchat.call(

sc, m.id, m.source, m.dest, m.username, parseTime(m.time),

m.privileged, m.type === 'chathistory', m.kind,

'' + m.value,

);

break;

case 'usermessage':

if(m.kind === 'filetransfer')

sc.fileTransfer(m.source, m.username, m.value);

else if(sc.onusermessage)

sc.onusermessage.call(

sc, m.source, m.dest, m.username, parseTime(m.time),

m.privileged, m.kind, m.error, m.value,

);

break;

case 'ping':

sc.send({

type: 'pong',

});

break;

case 'pong':

/* nothing */

break;

default:

console.warn('Unexpected server message', m.type);

return;

}

};

};

/**

* Protocol version 1 uses integers for dates, later versions use dates in

* ISO 8601 format. This function takes a date in either format and

* returns a Date object.

*

* @param {string|number} value

* @returns {Date}

*/

function parseTime(value) {

if(!value)

return null;

try {

return new Date(value);

} catch(e) {

console.warn(`Couldn't parse ${value}:`, e);

return null;

}

}

/**

* join requests to join a group. The onjoined callback will be called

* when we've effectively joined.

*

* @param {string} group - The name of the group to join.

* @param {string} username - the username to join as.

* @param {string|Object} credentials - password or authServer.

* @param {Object<string,any>} [data] - the initial associated data.

*/

ServerConnection.prototype.join = async function(group, username, credentials, data) {

let m = {

type: 'join',

kind: 'join',

group: group,

};

if(typeof username !== 'undefined' && username !== null)

m.username = username;

if((typeof credentials) === 'string') {

m.password = credentials;

} else {

switch(credentials.type) {

case 'password':

m.password = credentials.password;

break;

case 'token':

m.token = credentials.token;

break;

case 'authServer':

let r = await fetch(credentials.authServer, {

method: "POST",

headers: {

"Content-Type": "application/json",

},

body: JSON.stringify({

location: credentials.location,

username: username,

password: credentials.password,

}),

});

if(!r.ok)

throw new Error(

`The authorisation server said ${r.status} ${r.statusText}`,

);

if(r.status === 204) {

// no data, fallback to password auth

m.password = credentials.password;

break;

}

let ctype = r.headers.get("Content-Type");

if(!ctype)

throw new Error(

"The authorisation server didn't return a content type",

);

let semi = ctype.indexOf(";");

if(semi >= 0)

ctype = ctype.slice(0, semi);

ctype = ctype.trim();

switch(ctype.toLowerCase()) {

case 'application/jwt':

let data = await r.text();

if(!data)

throw new Error(

"The authorisation server returned empty token",

);

m.token = data;

break;

default:

throw new Error(`The authorisation server returned ${ctype}`);

break;

}

break;

default:

throw new Error(`Unknown credentials type ${credentials.type}`);

}

}

if(data)

m.data = data;

this.send(m);

};

/**

* leave leaves a group. The onjoined callback will be called when we've

* effectively left.

*

* @param {string} group - The name of the group to join.

*/

ServerConnection.prototype.leave = function(group) {

this.send({

type: 'join',

kind: 'leave',

group: group,

});

};

/**

* request sets the list of requested tracks

*

* @param {Object<string,Array<string>>} what

* - A dictionary that maps labels to a sequence of 'audio', 'video'

* or 'video-low. An entry with an empty label '' provides the default.

*/

ServerConnection.prototype.request = function(what) {

this.send({

type: 'request',

request: what,

});

};

/**

* findByLocalId finds an active connection with the given localId.

* It returns null if none was find.

*

* @param {string} localId

* @returns {Stream}

*/

ServerConnection.prototype.findByLocalId = function(localId) {

if(!localId)

return null;

let sc = this;

for(let id in sc.up) {

let s = sc.up[id];

if(s.localId === localId)

return s;

}

return null;

}

/**

* getRTCConfiguration returns the RTCConfiguration that should be used

* with this peer connection. This usually comes from the server, but may

* be overridden by the onpeerconnection callback.

*

* @returns {RTCConfiguration}

*/

ServerConnection.prototype.getRTCConfiguration = function() {

if(this.onpeerconnection) {

let conf = this.onpeerconnection.call(this);

if(conf !== null)

return conf;

}

return this.rtcConfiguration;

}

/**

* newUpStream requests the creation of a new up stream.

*

* @param {string} [localId]

* - The local id of the stream to create. If a stream already exists with

* the same local id, it is replaced with the new stream.

* @returns {Stream}

*/

ServerConnection.prototype.newUpStream = function(localId) {

let sc = this;

let id = newRandomId();

if(sc.up[id])

throw new Error('Eek!');

if(typeof RTCPeerConnection === 'undefined')

throw new Error("This browser doesn't support WebRTC");

let pc = new RTCPeerConnection(sc.getRTCConfiguration());

if(!pc)

throw new Error("Couldn't create peer connection");

let oldId = null;

if(localId) {

let old = sc.findByLocalId(localId);

oldId = old && old.id;

if(old)

old.close(true);

}

let c = new Stream(this, id, localId || newLocalId(), pc, true);

if(oldId)

c.replace = oldId;

sc.up[id] = c;

pc.onnegotiationneeded = async e => {

await c.negotiate();

};

pc.onicecandidate = e => {

if(!e.candidate)

return;

c.gotLocalIce(e.candidate);

};

pc.oniceconnectionstatechange = e => {

if(c.onstatus)

c.onstatus.call(c, pc.iceConnectionState);

if(pc.iceConnectionState === 'failed')

c.restartIce();

};

pc.ontrack = console.error;

return c;

};

/**

* chat sends a chat message to the server. The server will normally echo

* the message back to the client.

*

* @param {string} kind

* - The kind of message, either '', 'me' or an application-specific type.

* @param {string} dest - The id to send the message to, empty for broadcast.

* @param {string} value - The text of the message.

*/

ServerConnection.prototype.chat = function(kind, dest, value) {

this.send({

type: 'chat',

source: this.id,

dest: dest,

username: this.username,

kind: kind,

value: value,

});

};

/**

* userAction sends a request to act on a user.

*

* @param {string} kind - One of "op", "unop", "kick", "present", "unpresent".

* @param {string} dest - The id of the user to act upon.

* @param {any} [value] - An action-dependent parameter.

*/

ServerConnection.prototype.userAction = function(kind, dest, value) {

this.send({

type: 'useraction',

source: this.id,

dest: dest,

username: this.username,

kind: kind,

value: value,

});

};

/**

* userMessage sends an application-specific message to a user.

* This is similar to a chat message, but is not saved in the chat history.

*

* @param {string} kind - The kind of application-specific message.

* @param {string} dest - The id to send the message to, empty for broadcast.

* @param {unknown} [value] - An optional parameter.

* @param {boolean} [noecho] - If set, don't echo back the message to the sender.

*/

ServerConnection.prototype.userMessage = function(kind, dest, value, noecho) {

this.send({

type: 'usermessage',

source: this.id,

dest: dest,

username: this.username,

kind: kind,

value: value,

noecho: noecho,

});

};

/**

* groupAction sends a request to act on the current group.

*

* @param {string} kind

* @param {any} [data]

*/

ServerConnection.prototype.groupAction = function(kind, data) {

this.send({

type: 'groupaction',

source: this.id,

kind: kind,

username: this.username,

value: data,

});

};

/**

* gotOffer is called when we receive an offer from the server. Don't call this.

*

* @param {string} id

* @param {string} label

* @param {string} source

* @param {string} username

* @param {string} sdp

* @param {string} replace

* @function

*/

ServerConnection.prototype.gotOffer = async function(id, label, source, username, sdp, replace) {

let sc = this;

if(sc.up[id]) {

console.error("Duplicate connection id");

sc.send({

type: 'abort',

id: id,

});

return;

}

let oldLocalId = null;

if(replace) {

let old = sc.down[replace];

if(old) {

oldLocalId = old.localId;

old.close(true);

} else

console.error("Replacing unknown stream");

}

let c = sc.down[id];

if(c && oldLocalId)

console.error("Replacing duplicate stream");

if(!c) {

let pc;

try {

pc = new RTCPeerConnection(sc.getRTCConfiguration());

} catch(e) {

console.error(e);

sc.send({

type: 'abort',

id: id,

});

return;

}

c = new Stream(this, id, oldLocalId || newLocalId(), pc, false);

c.label = label;

sc.down[id] = c;

c.pc.onicecandidate = function(e) {

if(!e.candidate)

return;

c.gotLocalIce(e.candidate);

};

pc.oniceconnectionstatechange = e => {

if(c.onstatus)

c.onstatus.call(c, pc.iceConnectionState);

if(pc.iceConnectionState === 'failed') {

sc.send({

type: 'renegotiate',

id: id,

});

}

};

c.pc.ontrack = function(e) {

if(e.streams.length < 1) {

console.error("Got track with no stream");

return;

}

c.stream = e.streams[0];

let changed = recomputeUserStreams(sc, source);

if(c.ondowntrack) {

c.ondowntrack.call(

c, e.track, e.transceiver, e.streams[0],

);

}

if(changed && sc.onuser)

sc.onuser.call(sc, source, "change");

};

}

c.source = source;

c.username = username;

if(sc.ondownstream)

sc.ondownstream.call(sc, c);

try {

await c.pc.setRemoteDescription({

type: 'offer',

sdp: sdp,

});

await c.flushRemoteIceCandidates();

let answer = await c.pc.createAnswer();

if(!answer)

throw new Error("Didn't create answer");

await c.pc.setLocalDescription(answer);

this.send({

type: 'answer',

id: id,

sdp: c.pc.localDescription.sdp,

});

} catch(e) {

try {

if(c.onerror)

c.onerror.call(c, e);

} finally {

c.abort();

}

return;

}

c.localDescriptionSent = true;

c.flushLocalIceCandidates();

if(c.onnegotiationcompleted)

c.onnegotiationcompleted.call(c);

};

/**

* gotAnswer is called when we receive an answer from the server. Don't

* call this.

*

* @param {string} id

* @param {string} sdp

* @function

*/

ServerConnection.prototype.gotAnswer = async function(id, sdp) {

let c = this.up[id];

if(!c)

throw new Error('unknown up stream');

try {

await c.pc.setRemoteDescription({

type: 'answer',

sdp: sdp,

});

} catch(e) {

try {

if(c.onerror)

c.onerror.call(c, e);

} finally {

c.close();

}

return;

}

await c.flushRemoteIceCandidates();

if(c.onnegotiationcompleted)

c.onnegotiationcompleted.call(c);

};

/**

* gotRenegotiate is called when we receive a renegotiation request from

* the server. Don't call this.

*

* @param {string} id

* @function

*/

ServerConnection.prototype.gotRenegotiate = function(id) {

let c = this.up[id];

if(!c)

throw new Error('unknown up stream');

c.restartIce();

};

/**

* gotClose is called when we receive a close request from the server.