desk2/server/send.js

import * as math from './math';
import * as ser from './serializer';
import * as storage from './storage';
import * as config from './config';

import { MESSAGE, SESSION, EVENT } from './enums';
import { sessions, desks } from './storage';

function event_size(event) {
    let size = 4 + 4; // type + user_id

    switch (event.type) {
        case EVENT.PREDRAW:
        case EVENT.MOVE_CURSOR: {
            size += 4 * 2;
            break;
        }
   
        case EVENT.MOVE_CANVAS: {
            size += 4 * 2 + 4;
            break;
        }

        case EVENT.ZOOM_CANVAS: {
            size += 4 + 4 * 2;
            break;
        }

        case EVENT.USER_JOINED:
        case EVENT.LEAVE:
        case EVENT.CLEAR: {
            break;            
        }

        case EVENT.SET_COLOR: {
            size += 4;
            break;
        }

        case EVENT.SET_WIDTH: {
            size += 2;
            break;
        }

        case EVENT.STROKE: {
            size += 4 + 2 + 2 + 4; // stroke id + point count + width + color
            size += event.points.byteLength;
            size += math.round_to_pow2(event.pressures.byteLength, 4);
            break;
        }

        case EVENT.IMAGE: {
            size += 4 + 4 + 4 + 4 + 4; // file_id + x + y + width + height
            break;
        }

        case EVENT.IMAGE_MOVE: {
            size += 4 + 4 + 4; // file id + x + y
            break;
        }

        case EVENT.IMAGE_SCALE: {
            size += 4 + 4 + 4 + 4; // file_id + corner + x + y
            break;
        }

        case EVENT.UNDO:
        case EVENT.REDO: {
            break;
        }

        case EVENT.ERASER: {
            size += 4; // stroke id
            break;
        }

        default: {
            console.error(event.desk_id);
            console.error('fuck');
            console.trace();
            process.exit(1);
        }
    }

    return size;
}

function create_session(ws, desk_id) {
    const session = {
        id: math.crypto_random32(),
        desk_id: desk_id,
        state: SESSION.OPENED,
        sn: 0,
        lsn: 0,
        ws: ws,
        color: 0x00,
        width: 8,
    };

    storage.queries.insert_session.run({
        '$id': session.id,
        '$desk_id': desk_id
    });

    sessions[session.id] = session;

    return session;
}

export async function send_init(ws) {
    if (!ws) {
        return;
    }

    const session_id = ws.data.session_id;
    const desk_id = ws.data.desk_id;
    const desk = desks[desk_id];

    let opcode = MESSAGE.INIT;
    let size = 4 + 4 + 4 + 4 + 4 + 4; // opcode + user_id + lsn + event count + stroke count + user count + total_point_count
    let session = null;

    if (session_id in sessions && sessions[session_id].desk_id == desk_id) {
        session = sessions[session_id];
        size += 4 + 2; // color + width
    } else {
        size += 4; // session id
        opcode = MESSAGE.JOIN;
        session = create_session(ws, desk_id);
        ws.data.session_id = session.id;
    }

    session.desk_id = desk_id;
    session.ws = ws;
    session.sn = 0; // Always re-send everything on reconnect
    session.state = SESSION.OPENED;

    if (config.DEBUG_PRINT) console.log(`session ${session.id} opened`);

    let user_count = 0;

    for (const sid in sessions) {
        const other_session = sessions[sid];
        if (other_session.id !== session_id && other_session.desk_id === desk_id) {
            ++user_count;
            size += 4 + 4 + 2 + 1; // user id + color + width + online
        }
    }

    for (const event of desk.events) {
        size += event_size(event);
    }

    const s = ser.create(size);

    ser.u32(s, opcode);
    ser.u32(s, session.lsn);

    if (opcode === MESSAGE.JOIN) {
        ser.u32(s, session.id);
    } else {
        ser.u32(s, session.color);
        ser.u16(s, session.width);
    }

    ser.u32(s, desk.events.length);
    ser.u32(s, user_count);
    ser.u32(s, desk.total_points);

    for (const sid in sessions) {
        const other_session = sessions[sid];
        if (other_session.id !== session_id && other_session.desk_id === desk_id) {
            // console.log(other_session.id, other_session.color, other_session.width);
            ser.u32(s, other_session.id);
            ser.u32(s, other_session.color);
            ser.u16(s, other_session.width);
            ser.u8(s, other_session.state === SESSION.READY);
        }
    }

    ser.align(s, 4);

    for (const event of desk.events) {
        ser.event(s, event);
    }

    await ws.send(s.buffer);
}

export function send_ack(ws, lsn) {
    if (!ws) {
        return;
    }

    const size = 4 + 4; // opcode + lsn
    const s = ser.create(size);

    ser.u32(s, MESSAGE.ACK);
    ser.u32(s, lsn);

    if (config.DEBUG_PRINT) console.log(`ack ${lsn} out`);

    ws.send(s.buffer);
}

function send_fire(ws, event) {
    if (!ws) {
        return;
    }

    const s = ser.create(4 + 4 + event_size(event));

    ser.u32(s, MESSAGE.FIRE);
    ser.event(s, event);

    ws.send(s.buffer);
}

export function fire_event(from_session, event) {
    for (const sid in sessions) {
        const other = sessions[sid];

        if (other.id === from_session.id) {
            continue;
        }

        if (other.state !== SESSION.READY) {
            continue;
        }

        if (other.desk_id != from_session.desk_id) {
            continue;
        }

        if (event.type === EVENT.MOVE_CANVAS && other.follow !== from_session.id) {
            // Do not spam canvas move events to those who don't follow us
            continue;
        }

        send_fire(other.ws, event);
    }
}

async function sync_session(session_id) {
    if (!(session_id in sessions)) {
        return;
    }

    const session = sessions[session_id];
    const desk = desks[session.desk_id];

    if (!session.ws) {
        return;
    }

    if (session.state !== SESSION.READY) {
        return;
    }

    let size = 4 + 4 + 4; // opcode + sn + event count
    let count = desk.sn - session.sn;

    if (count === 0) {
        if (config.DEBUG_PRINT) console.log('client ACKed all events');
        return;
    }

    for (let i = count - 1; i >= 0; --i) {
        const event = desk.events[desk.events.length - 1 - i];
        size += event_size(event);
    }

    const s = ser.create(size);

    ser.u32(s, MESSAGE.SYN);
    ser.u32(s, desk.sn);
    ser.u32(s, count);

    for (let i = count - 1; i >= 0; --i) {
        const event = desk.events[desk.events.length - 1 - i];
        ser.event(s, event);
    }
    
    if (config.DEBUG_PRINT) console.log(`syn ${desk.sn} out`);

    await session.ws.send(s.buffer);
    
    if (session.sync_attempts < config.SYNC_MAX_ATTEMPTS) {
        session.sync_attempts += 1;
        session.sync_timer = setTimeout(() => sync_session(session_id), config.SYNC_TIMEOUT);
    }
}

export function sync_desk(desk_id) {
    for (const sid in sessions) {
        const session = sessions[sid];
        if (session.state === SESSION.READY && session.desk_id == desk_id) { // NOT ===, because might be string or int IDK
            if (session.sync_timer) {
                clearTimeout(session.sync_timer);
            }
            sync_session(sid);
        }
    }
}