snapshot of https://github.com/raimohanska/ourboard.git (master, 2026-09-23; blobs > 131072B omitted)

backend/migrations/001_init.js

@@ -0,0 +1,16 @@
+exports.up = (pgm) => {
+    pgm.sql(`
+    CREATE TABLE IF NOT EXISTS board (id text PRIMARY KEY, name text NOT NULL);
+    ALTER TABLE board ADD COLUMN IF NOT EXISTS content JSONB NOT NULL;
+    ALTER TABLE board ADD COLUMN IF NOT EXISTS history JSONB NOT NULL default '[]';            
+    CREATE TABLE IF NOT EXISTS board_event (board_id text REFERENCES board(id), last_serial integer, events JSONB NOT NULL, PRIMARY KEY (board_id, last_serial));
+    ALTER TABLE board_event ALTER COLUMN last_serial SET NOT NULL;
+    ALTER TABLE board_event ALTER COLUMN board_id SET NOT NULL;
+    CREATE TABLE IF NOT EXISTS board_api_token (board_id text REFERENCES board(id), token TEXT NOT NULL);
+    CREATE TABLE IF NOT EXISTS app_user (id text PRIMARY KEY, email text NOT NULL);
+    CREATE TABLE IF NOT EXISTS user_board (user_id text REFERENCES app_user(id), board_id text REFERENCES board(id), last_opened TIMESTAMP NOT NULL, PRIMARY KEY (user_id, board_id));
+    ALTER TABLE board ADD COLUMN IF NOT EXISTS ws_host TEXT NULL;
+    ALTER TABLE board ADD COLUMN IF NOT EXISTS created_at TIMESTAMP NULL DEFAULT now();            
+    ALTER TABLE board_event ADD COLUMN IF NOT EXISTS saved_at TIMESTAMP NULL DEFAULT now();
+    `)
+}

backend/migrations/002_add_first_serial.js

@@ -0,0 +1,7 @@
+exports.up = (pgm) => {
+    pgm.sql(`
+    ALTER TABLE board_event ADD COLUMN IF NOT EXISTS first_serial int;
+    UPDATE board_event SET first_serial = COALESCE(CAST(events#>'{events, 0, serial}' as int), 0);
+    ALTER TABLE board_event ALTER COLUMN first_serial SET NOT NULL;
+    `)
+}

backend/migrations/003_refactor_access.sql

@@ -0,0 +1,27 @@
+alter table board add public_read boolean null;
+alter table board add public_write boolean null;
+create table board_access (
+  board_id text not null references board(id),
+  domain text null,
+  email text null,
+  access text null
+);
+
+with allow_json as (
+select id, jsonb_array_elements(content -> 'accessPolicy' -> 'allowList') as e
+from board),
+
+allow_entry as (
+	select id, e ->> 'domain' as domain, e ->> 'email' as email, e ->> 'access' as access
+	from allow_json
+)
+
+insert into board_access(board_id, domain, email, access) (
+  select ae.id as board_id, ae.domain, ae.email, ae.access
+  from allow_entry ae
+);
+
+update board
+set public_read = coalesce(content -> 'accessPolicy' ->> 'publicRead', 'false') :: boolean,
+    public_write = coalesce(content -> 'accessPolicy' ->> 'publicWrite', 'false') :: boolean
+where content -> 'accessPolicy' is not null

backend/migrations/004_public_boards_access.sql

@@ -0,0 +1,4 @@
+update board
+set public_read = 't',
+    public_write = 't'
+where content -> 'accessPolicy' is null

backend/migrations/005_uuid_extension.sql

@@ -0,0 +1 @@
+CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

backend/migrations/006_unique_email.sql

@@ -0,0 +1 @@
+ALTER TABLE app_user ADD CONSTRAINT unique_email UNIQUE (email);

backend/migrations/007_add_crdt_update.sql

@@ -0,0 +1 @@
+alter table board_event add column crdt_update bytea null;

backend/migrations/008_allow_crdt_only_bundle.sql

@@ -0,0 +1,2 @@
+alter table board_event alter column first_serial drop not null;
+alter table board_event add constraint first_serial_or_crdt check (crdt_update is not null OR first_serial is not null);

backend/migrations/009_drop_board_event_primary_key.sql

@@ -0,0 +1,2 @@
+ALTER TABLE board_event DROP CONSTRAINT board_event_pkey;
+CREATE INDEX board_event_board_index ON board_event (board_id);

backend/migrations/010_drop_history_column.sql

@@ -0,0 +1 @@
+alter table board drop column history;

backend/package.json

@@ -0,0 +1,74 @@
+{
+  "name": "rboard-backend",
+  "version": "1.0.0",
+  "main": "dist/index.js",
+  "license": "MIT",
+  "dependencies": {
+    "@types/cookies": "^0.7.7",
+    "@types/date-fns": "^2.6.0",
+    "@types/express-ws": "^3.0.0",
+    "@types/html-entities": "^1.2.16",
+    "@types/json-diff": "^0.5.0",
+    "@types/lodash": "^4.14.161",
+    "@types/node": "^14.6.2",
+    "@types/pg": "^7.14.4",
+    "@types/ramda": "^0.27.29",
+    "@types/tcp-port-used": "^1.0.0",
+    "@types/ws": "^7.4.0",
+    "aws-sdk": "^2.778.0",
+    "cookies": "^0.8.0",
+    "csv-writer": "^1.6.0",
+    "date-fns": "^2.17.0",
+    "dotenv": "^8.2.0",
+    "express": "^4.17.1",
+    "express-ws": "^4.0.0",
+    "fp-ts": "^2.9.5",
+    "google-auth-library": "^7.0.2",
+    "googleapis": "^110.0.0",
+    "html-entities": "^2.1.0",
+    "io-ts": "^2.2.15",
+    "io-ts-types": "^0.5.15",
+    "json-diff": "^0.5.4",
+    "jsonwebtoken": "^8.5.1",
+    "lodash": "^4.17.20",
+    "lonna": "^0.12.2",
+    "monocle-ts": "^2.3.7",
+    "newtype-ts": "^0.3.4",
+    "node-pg-migrate": "^6.0.0",
+
+    "pg": "^8.3.3",
+    "pg-query-stream": "^4.2.1",
+
+    "tcp-port-used": "^1.0.2",
+    "tsx": "3.13.0",
+    "typera-express": "^2.3.0",
+
+    "typescript": "^5.3",
+    "uWebSockets.js": "uNetworking/uWebSockets.js#v18.14.0",
+    "uuid": "^8.3.0",
+    "uuid4": "^2.0.2",
+    "y-protocols": "^1.0.6"
+  },
+  "scripts": {
+    "start": "node --enable-source-maps .",
+    "dev": "npm-run-all --parallel watch",
+    "build": "tsc && ncc build dist/backend/src/server.js -o dist",
+    "watch": "tsc-watch --onSuccess \"node --enable-source-maps dist/backend/src/server.js\" --preserveWatchOutput",
+    "watch-ts": "tsc-watch --preserveWatchOutput",
+
+
+    "lint": "echo 'No linting configured'"
+  },
+  "engines": {
+    "node": ">=14"
+  },
+  "devDependencies": {
+    "@types/express": "^4.17.7",
+    "@types/jsonwebtoken": "^8.5.0",
+    "@types/node-fetch": "^2.5.9",
+
+    "@types/uuid": "^8.3.0",
+    "@vercel/ncc": "^0.38.1",
+    "tsc-watch": "^4.2.9"
+  }
+}

backend/src/api/api-routes.ts

@@ -0,0 +1,22 @@
+import { router } from "typera-express"
+import { boardCreate } from "./board-create"
+import { boardCSVGet } from "./board-csv-get"
+import { boardGet } from "./board-get"
+import { boardHierarchyGet } from "./board-hierarchy-get"
+import { boardHistoryGet } from "./board-history-get"
+import { boardUpdate } from "./board-update"
+import { githubWebhook } from "./github-webhook"
+import { itemCreate } from "./item-create"
+import { itemCreateOrUpdate } from "./item-create-or-update"
+
+export default router(
+    boardGet,
+    boardHierarchyGet,
+    boardCSVGet,
+    boardCreate,
+    boardUpdate,
+    githubWebhook,
+    itemCreate,
+    itemCreateOrUpdate,
+    boardHistoryGet,
+)

backend/src/api/board-create.ts

@@ -0,0 +1,29 @@
+import * as t from "io-ts"
+import { NonEmptyString } from "io-ts-types"
+import { ok } from "typera-common/response"
+import { body } from "typera-express/parser"
+import {
+    Board,
+    BoardAccessPolicyCodec,
+    CrdtDisabled,
+    CrdtEnabled,
+    newBoard,
+    optional,
+} from "../../../common/src/domain"
+import { addBoard } from "../board-state"
+import { route } from "./utils"
+import { getConfig } from "../config"
+/**
+ * Creates a new board.
+ *
+ * @tags Board
+ */
+export const boardCreate = route
+    .post("/api/v1/board")
+    .use(body(t.type({ name: NonEmptyString, accessPolicy: BoardAccessPolicyCodec, crdt: optional(t.literal(true)) })))
+    .handler(async (request) => {
+        const crdt = (getConfig().crdt === "true" || request.body.crdt) ?? false
+        let board: Board = newBoard(request.body.name, crdt ? CrdtEnabled : CrdtDisabled, request.body.accessPolicy)
+        const boardWithHistory = await addBoard(board, true)
+        return ok({ id: boardWithHistory.board.id, accessToken: boardWithHistory.accessTokens[0] })
+    })

backend/src/api/board-csv-get.ts

@@ -0,0 +1,62 @@
+import { createArrayCsvStringifier } from "csv-writer"
+import _ from "lodash"
+import { ok } from "typera-common/response"
+import { Board, Container, Item, TextItem } from "../../../common/src/domain"
+import { apiTokenHeader, checkBoardAPIAccess, route } from "./utils"
+import { augmentBoardWithCRDT } from "../../../common/src/board-crdt-helper"
+import { yWebSocketServer } from "../board-yjs-server"
+
+/**
+ * Gets board current contents
+ *
+ * @tags Board
+ */
+export const boardCSVGet = route
+    .get("/api/v1/board/:boardId/csv")
+    .use(apiTokenHeader)
+    .handler((request) =>
+        checkBoardAPIAccess(request, async (boardState) => {
+            const board = augmentBoardWithCRDT(
+                await yWebSocketServer.docs.getYDocAndWaitForFetch(boardState.board.id),
+                boardState.board,
+            )
+            const textItemsWithParent = Object.values(board.items).filter(
+                (i) => i.containerId !== undefined && (i.type === "text" || i.type === "note"),
+            )
+            const textItemGroups = _.groupBy(textItemsWithParent, (i) => i.containerId)
+            const rows = Object.entries(textItemGroups).map(([parentId, textItems]) => {
+                const rowContainer = board.items[parentId]
+                return {
+                    parents: parentChain(board)(rowContainer),
+                    rowContainer,
+                    textItems,
+                } as Row
+            })
+            if (rows.length === 0) return csv(board, [])
+            const maxDepth = _.max(rows.map((r) => r.parents.length))!
+            const csvData = rows.map((r) => {
+                return [
+                    ...r.parents.map((c) => c.text),
+                    ..._.times(maxDepth - r.parents.length, () => ""),
+                    r.rowContainer.text,
+                    ...r.textItems.map((i) => i.text),
+                ]
+            })
+            return csv(board, csvData)
+        }),
+    )
+
+type Row = { parents: Container[]; rowContainer: Container; textItems: TextItem[] }
+
+const parentChain = (board: Board) => (item: Item): Container[] => {
+    if (!item.containerId) return []
+    const parent = board.items[item.containerId]
+    if (parent.type !== "container")
+        throw Error(`Parent item ${item.containerId} is of type ${parent.type}, expecting container`)
+    return [parent, ...parentChain(board)(parent)]
+}
+
+function csv(board: Board, rows: string[][]) {
+    const result = createArrayCsvStringifier({}).stringifyRecords(rows)
+    return ok(result, { "content-type": "text/csv", "content-disposition": `attachment; filename=${board.name}.csv` })
+}

backend/src/api/board-get.ts

@@ -0,0 +1,22 @@
+import { ok } from "typera-common/response"
+import { apiTokenHeader, checkBoardAPIAccess, route } from "./utils"
+import { yWebSocketServer } from "../board-yjs-server"
+import { augmentBoardWithCRDT } from "../../../common/src/board-crdt-helper"
+
+/**
+ * Gets board current contents
+ *
+ * @tags Board
+ */
+export const boardGet = route
+    .get("/api/v1/board/:boardId")
+    .use(apiTokenHeader)
+    .handler((request) =>
+        checkBoardAPIAccess(request, async (boardState) => {
+            const board = augmentBoardWithCRDT(
+                await yWebSocketServer.docs.getYDocAndWaitForFetch(boardState.board.id),
+                boardState.board,
+            )
+            return ok({ board })
+        }),
+    )

backend/src/api/board-hierarchy-get.ts

@@ -0,0 +1,34 @@
+import { ok } from "typera-common/response"
+import { Board, Item } from "../../../common/src/domain"
+import { apiTokenHeader, checkBoardAPIAccess, route } from "./utils"
+import { augmentBoardWithCRDT } from "../../../common/src/board-crdt-helper"
+import { yWebSocketServer } from "../board-yjs-server"
+
+/**
+ * Gets board current contents
+ *
+ * @tags Board
+ */
+export const boardHierarchyGet = route
+    .get("/api/v1/board/:boardId/hierarchy")
+    .use(apiTokenHeader)
+    .handler((request) =>
+        checkBoardAPIAccess(request, async (boardState) => {
+            const board = augmentBoardWithCRDT(
+                await yWebSocketServer.docs.getYDocAndWaitForFetch(boardState.board.id),
+                boardState.board,
+            )
+            return ok({ board: getBoardHierarchy(board) })
+        }),
+    )
+
+export type ItemHierarchy = Item & { children: ItemHierarchy[] }
+export function getBoardHierarchy(board: Board) {
+    const allItems = Object.values(board.items)
+    const rootItems = allItems.filter((i) => i.containerId === undefined).map(getItemHierarchy(allItems))
+    return { ...board, items: rootItems }
+}
+const getItemHierarchy = (items: Item[]) => (item: Item): ItemHierarchy => {
+    const children: ItemHierarchy[] = items.filter((i) => i.containerId === item.id).map(getItemHierarchy(items))
+    return { ...item, children }
+}

backend/src/api/board-history-get.ts

@@ -0,0 +1,47 @@
+import { ok, streamingBody } from "typera-common/response"
+import { getFullBoardHistory } from "../board-store"
+import { withDBClient } from "../db"
+import { apiTokenHeader, checkBoardAPIAccess, route } from "./utils"
+
+/**
+ * List the history of a board
+ *
+ * @tags Board
+ */
+export const boardHistoryGet = route
+    .get("/api/v1/board/:boardId/history")
+    .use(apiTokenHeader)
+    .handler((request) =>
+        checkBoardAPIAccess(request, async (board) => {
+            return ok(
+                streamingJSONBody("history", async (callback) => {
+                    await withDBClient(
+                        async (client) =>
+                            await getFullBoardHistory(board.board.id, client, (bundle) => bundle.forEach(callback)),
+                    )
+                }),
+            )
+        }),
+    )
+
+function streamingJSONBody(fieldName: string, generator: (callback: (item: any) => void) => Promise<void>) {
+    return streamingBody(async (stream) => {
+        // Due to memory concerns we fetch board histories from DB as chunks, so this API
+        // response must also be chunked
+        try {
+            stream.write(`{"${fieldName}":[`)
+            let chunksProcessed = 0
+            await generator((item) => {
+                let prefix = chunksProcessed === 0 ? "" : ","
+                stream.write(`${prefix}${JSON.stringify(item)}`)
+                chunksProcessed++
+            })
+            stream.write("]}")
+            stream.end()
+            //console.log(`Wrote ${chunksProcessed} chunks`)
+        } catch (e) {
+            console.error(`Error writing a streamed body: ${e}`)
+            stream.end()
+        }
+    })
+}

backend/src/api/board-update.ts

@@ -0,0 +1,29 @@
+import * as t from "io-ts"
+import { NonEmptyString } from "io-ts-types"
+import { ok } from "typera-common/response"
+import { body } from "typera-express/parser"
+import { BoardAccessPolicyCodec } from "../../../common/src/domain"
+import { apiTokenHeader, checkBoardAPIAccess, dispatchSystemAppEvent, route } from "./utils"
+import { renameBoardConvenienceColumnOnly, updateBoardAccessPolicy } from "../board-store"
+
+/**
+ * Changes board name and, optionally, access policy.
+ *
+ * @tags Board
+ */
+export const boardUpdate = route
+    .put("/api/v1/board/:boardId")
+    .use(apiTokenHeader, body(t.type({ name: NonEmptyString, accessPolicy: BoardAccessPolicyCodec })))
+    .handler((request) =>
+        checkBoardAPIAccess(request, async (board) => {
+            const { boardId } = request.routeParams
+            const { name, accessPolicy } = request.body
+            await renameBoardConvenienceColumnOnly(boardId, name)
+            dispatchSystemAppEvent(board, { action: "board.rename", boardId, name })
+            if (accessPolicy) {
+                await updateBoardAccessPolicy(boardId, accessPolicy)
+                dispatchSystemAppEvent(board, { action: "board.setAccessPolicy", boardId, accessPolicy })
+            }
+            return ok({ ok: true })
+        }),
+    )

backend/src/api/github-webhook.ts

@@ -0,0 +1,74 @@
+import { encode as htmlEncode } from "html-entities"
+import * as t from "io-ts"
+import { badRequest, internalServerError, ok } from "typera-common/response"
+import { body } from "typera-express/parser"
+import { RED, YELLOW } from "../../../common/src/colors"
+import { Note } from "../../../common/src/domain"
+import { getBoard } from "../board-state"
+import { addItem, dispatchSystemAppEvent, InvalidRequest, route } from "./utils"
+
+// TODO: require API_TOKEN header for github too!
+/**
+ * GitHub webhook
+ *
+ * @tags Webhooks
+ */
+export const githubWebhook = route
+    .post("/api/v1/webhook/github/:boardId")
+    .use(
+        body(
+            t.partial({
+                issue: t.type({
+                    html_url: t.string,
+                    title: t.string,
+                    number: t.number,
+                    state: t.string,
+                    labels: t.array(t.type({ name: t.string })),
+                }),
+            }),
+        ),
+    )
+    .handler(async (request) => {
+        try {
+            const boardId = request.routeParams.boardId
+            const body = request.body
+            const board = await getBoard(boardId)
+            if (!board) {
+                console.warn(`Github webhook call for unknown board ${boardId}`)
+                return ok()
+            }
+            if (body.issue) {
+                const url = body.issue.html_url
+                const title = body.issue.title
+                const number = body.issue.number.toString()
+                const state = body.issue.state
+                if (state !== "open") {
+                    console.log(`Github webhook call board ${boardId}: Item in ${state} state`)
+                } else {
+                    const linkStart = `<a href=${url}>`
+                    const linkHTML = `${linkStart}${htmlEncode(number)}</a> ${htmlEncode(title)}`
+                    const existingItem = Object.values(board.board.items).find(
+                        (i) => i.type === "note" && i.text.includes(url),
+                    ) as Note | undefined
+                    const isBug = body.issue.labels.some((l) => l.name === "bug")
+                    const color = isBug ? RED : YELLOW
+                    if (!existingItem) {
+                        console.log(`Github webhook call board ${boardId}: New item`)
+                        addItem(board, "note", linkHTML, color, "New issues")
+                    } else {
+                        console.log(`Github webhook call board ${boardId}: Item exists`)
+                        const updatedItem: Note = { ...existingItem, color }
+                        dispatchSystemAppEvent(board, { action: "item.update", boardId, items: [updatedItem] })
+                    }
+                }
+            }
+            return ok()
+        } catch (e) {
+            console.error(e)
+            if (e instanceof InvalidRequest) {
+                return badRequest(e.message)
+            } else {
+                return internalServerError()
+            }
+        }
+    })

backend/src/api/item-create-or-update.ts

@@ -0,0 +1,137 @@
+import * as t from "io-ts"
+import { NonEmptyString } from "io-ts-types"
+import _ from "lodash"
+import { ok } from "typera-common/response"
+import { body } from "typera-express/parser"
+import { Color, isNote, Note } from "../../../common/src/domain"
+import { ServerSideBoardState } from "../board-state"
+import {
+    addItem,
+    apiTokenHeader,
+    checkBoardAPIAccess,
+    dispatchSystemAppEvent,
+    findContainer,
+    getItemAttributesForContainer,
+    InvalidRequest,
+    route,
+} from "./utils"
+/**
+ * Creates a new item on given board or updates an existing one.
+ * If you want to add the item onto a specific area/container element on the board, you can
+ * find the id of the container by inspecting with your browser.
+ *
+ * @tags Board
+ */
+export const itemCreateOrUpdate = route
+    .put("/api/v1/board/:boardId/item/:itemId")
+    .use(
+        apiTokenHeader,
+        body(
+            t.intersection([
+                t.type({
+                    type: t.literal("note"),
+                    text: NonEmptyString,
+                    color: t.string,
+                }),
+                t.partial({
+                    container: t.string,
+                    x: t.number,
+                    y: t.number,
+                    width: t.number,
+                    height: t.number,
+                    replaceTextIfExists: t.boolean,
+                    replaceColorIfExists: t.boolean,
+                    replaceContainerIfExists: t.boolean,
+                }),
+            ]),
+        ),
+    )
+    .handler((request) =>
+        checkBoardAPIAccess(request, async (board) => {
+            const { itemId } = request.routeParams
+            let {
+                type,
+                text,
+                color,
+                container,
+                replaceTextIfExists,
+                replaceColorIfExists,
+                replaceContainerIfExists = true,
+                ...rest
+            } = request.body
+            console.log(`PUT item for board ${board.board.id} item ${itemId}: ${JSON.stringify(request.req.body)}`)
+            const existingItem = board.board.items[itemId]
+            if (existingItem) {
+                updateItem(
+                    board,
+                    rest.x ?? existingItem.x,
+                    rest.y ?? existingItem.y,
+                    type,
+                    text,
+                    color,
+                    container,
+                    rest.width ?? existingItem.width,
+                    rest.height ?? existingItem.height,
+                    itemId,
+                    replaceTextIfExists,
+                    replaceColorIfExists,
+                    replaceContainerIfExists,
+                )
+            } else {
+                console.log(`Adding new item`)
+                addItem(board, type, text, color, container, itemId, rest)
+            }
+            return ok({ ok: true })
+        }),
+    )
+
+function updateItem(
+    board: ServerSideBoardState,
+    x: number,
+    y: number,
+    type: "note",
+    text: string,
+    color: Color,
+    container: string | undefined,
+    width: number,
+    height: number,
+    itemId: string,
+    replaceTextIfExists: boolean | undefined,
+    replaceColorIfExists: boolean | undefined,
+    replaceContainerIfExists: boolean | undefined,
+) {
+    const existingItem = board.board.items[itemId]
+
+    if (!isNote(existingItem)) {
+        throw new InvalidRequest("Unexpected item type")
+    }
+
+    let updatedItem: Note = {
+        ...existingItem,
+        x: x !== undefined ? x : existingItem.x,
+        y: y !== undefined ? y : existingItem.y,
+        text: replaceTextIfExists !== false ? text : existingItem.text,
+        color: replaceColorIfExists !== false ? color || existingItem.color : existingItem.color,
+        width: width !== undefined ? width : existingItem.width,
+        height: height !== undefined ? height : existingItem.height,
+    }
+
+    if (container && replaceContainerIfExists !== false) {
+        const containerItem = findContainer(container, board.board)
+        const currentContainer = findContainer(existingItem.containerId, board.board)
+        const containerAttrs =
+            containerItem !== currentContainer ? getItemAttributesForContainer(container, board.board) : {}
+
+        updatedItem = {
+            ...updatedItem,
+            ...containerAttrs,
+        }
+    }
+
+    if (!_.isEqual(updatedItem, existingItem)) {
+        console.log(`Updating existing item`)
+        dispatchSystemAppEvent(board, { action: "item.update", boardId: board.board.id, items: [updatedItem] })
+    } else {
+        console.log(`Not updating: item not changed`)
+    }
+}

backend/src/api/item-create.ts

@@ -0,0 +1,41 @@
+import * as t from "io-ts"
+import { ok } from "typera-common/response"
+import { body } from "typera-express/parser"
+import { addItem, apiTokenHeader, checkBoardAPIAccess, route } from "./utils"
+
+/**
+ * Creates a new item on given board. If you want to add the item onto a
+ * specific area/container element on the board, you can find the id of the
+ * container by inspecting with your browser.
+ *
+ * @tags Board
+ */
+export const itemCreate = route
+    .post("/api/v1/board/:boardId/item")
+    .use(
+        apiTokenHeader,
+        body(
+            t.intersection([
+                t.type({
+                    type: t.literal("note"),
+                    text: t.string,
+                    color: t.string,
+                }),
+                t.partial({
+                    container: t.string,
+                    x: t.number,
+                    y: t.number,
+                    width: t.number,
+                    height: t.number,
+                }),
+            ]),
+        ),
+    )
+    .handler((request) =>
+        checkBoardAPIAccess(request, async (board) => {
+            const { type, text, color, container, ...rest } = request.body
+            console.log(`POST item for board ${board.board.id}: ${JSON.stringify(request.req.body)}`)
+            const item = addItem(board, type, text, color, container, undefined, rest)
+            return ok(item)
+        }),
+    )

backend/src/api/utils.ts

@@ -0,0 +1,124 @@
+import * as bodyParser from "body-parser"
+import * as t from "io-ts"
+import { badRequest, internalServerError, notFound } from "typera-common/response"
+import { applyMiddleware } from "typera-express"
+import { wrapNative } from "typera-express/middleware"
+import { headers } from "typera-express/parser"
+import { DEFAULT_NOTE_COLOR } from "../../../common/src/colors"
+import {
+    AppEvent,
+    Board,
+    BoardHistoryEntry,
+    Color,
+    Container,
+    EventUserInfo,
+    newISOTimeStamp,
+    newNote,
+    Note,
+    PersistableBoardItemEvent,
+} from "../../../common/src/domain"
+import { getBoard, ServerSideBoardState, updateBoards } from "../board-state"
+import { broadcastBoardEvent } from "../websocket-sessions"
+
+export const route = applyMiddleware(wrapNative(bodyParser.json()))
+export const apiTokenHeader = headers(t.partial({ API_TOKEN: t.string }))
+
+export async function checkBoardAPIAccess<T>(
+    request: { routeParams: { boardId: string }; headers: { API_TOKEN?: string | undefined } },
+    fn: (board: ServerSideBoardState) => Promise<T>,
+) {
+    const boardId = request.routeParams.boardId
+    const apiToken = request.headers.API_TOKEN
+    try {
+        const board = await getBoard(boardId)
+        if (!board) return notFound()
+        if (board.board.accessPolicy || board.accessTokens.length) {
+            if (!apiToken) {
+                return badRequest("API_TOKEN header is missing")
+            }
+            if (!board.accessTokens.some((t) => t === apiToken)) {
+                console.log(`API_TOKEN ${apiToken} not on list ${board.accessTokens}`)
+                return badRequest("Invalid API_TOKEN")
+            }
+        }
+        return await fn(board)
+    } catch (e) {
+        console.error(e)
+        if (e instanceof InvalidRequest) {
+            return badRequest(e.message)
+        } else {
+            return internalServerError()
+        }
+    }
+}
+
+export function findContainer(container: string | undefined, board: Board): Container | null {
+    if (container !== undefined) {
+        if (typeof container !== "string") {
+            throw new InvalidRequest("Expecting container to be undefined, or an id or name of an Container item")
+        }
+        const containerItem = Object.values(board.items).find(
+            (i) => i.type === "container" && (i.text.toLowerCase() === container.toLowerCase() || i.id === container),
+        )
+        if (!containerItem) {
+            throw new InvalidRequest(`Container "${container}" not found by id or name`)
+        }
+        return containerItem as Container
+    } else {
+        return null
+    }
+}
+
+export function getItemAttributesForContainer(container: string | undefined, board: Board) {
+    const containerItem = findContainer(container, board)
+    if (containerItem) {
+        return {
+            containerId: containerItem.id,
+            x: containerItem.x + 2,
+            y: containerItem.y + 2,
+        }
+    }
+    return {}
+}
+
+export function dispatchSystemAppEvent(board: ServerSideBoardState, appEvent: PersistableBoardItemEvent) {
+    const user: EventUserInfo = { userType: "system", nickname: "Github webhook" }
+    let historyEntry: BoardHistoryEntry = { ...appEvent, user, timestamp: newISOTimeStamp() }
+    console.log(JSON.stringify(historyEntry))
+    // TODO: refactor, this is the same sequence as done in connection-handler for messages from clients
+    const serial = updateBoards(board, historyEntry)
+    historyEntry = { ...historyEntry, serial }
+    broadcastBoardEvent(historyEntry)
+}
+
+export function addItem(
+    board: ServerSideBoardState,
+    type: "note",
+    text: string,
+    color: Color,
+    container: string | undefined,
+    itemId?: string,
+    partialParams?: Partial<{ x: number; y: number; width: number; height: number }>,
+) {
+    if (type !== "note") throw new InvalidRequest("Expecting type: note")
+    if (typeof text !== "string" || text.length === 0) throw new InvalidRequest("Expecting non zero-length text")
+
+    let itemAttributes: object = getItemAttributesForContainer(container, board.board)
+    if (itemId) itemAttributes = { ...itemAttributes, id: itemId }
+
+    // Merge partial parameters with existing attributes
+    if (partialParams) {
+        itemAttributes = { ...itemAttributes, ...partialParams }
+    }
+
+    const item: Note = { ...newNote(text, color || DEFAULT_NOTE_COLOR), ...itemAttributes }
+    const appEvent: AppEvent = { action: "item.add", boardId: board.board.id, items: [item], connections: [] }
+    dispatchSystemAppEvent(board, appEvent)
+    return item
+}
+
+export class InvalidRequest extends Error {
+    constructor(message: string) {
+        super(message)
+    }
+}

backend/src/board-event-handler.ts

@@ -0,0 +1,180 @@
+import {
+    AppEvent,
+    BoardHistoryEntry,
+    canWrite,
+    checkBoardAccess,
+    Id,
+    isBoardItemEvent,
+    isPersistableBoardItemEvent,
+    newISOTimeStamp,
+} from "../../common/src/domain"
+import { getBoard, maybeGetBoard, updateBoards } from "./board-state"
+import { getBoardInfo, renameBoardConvenienceColumnOnly, updateBoardAccessPolicy } from "./board-store"
+import { handleCommonEvent } from "./common-event-handler"
+import { MessageHandlerResult } from "./connection-handler"
+import { WS_HOST_DEFAULT, WS_HOST_LOCAL, WS_PROTOCOL } from "./host-config"
+import { obtainLock } from "./locker"
+import { associateUserWithBoard } from "./user-store"
+import { addSessionToBoard, broadcastBoardEvent, getSession } from "./websocket-sessions"
+import { toBuffer, WsWrapper } from "./ws-wrapper"
+
+export const handleBoardEvent = (allowedBoardId: Id | null, getSignedPutUrl: (key: string) => string) => async (
+    socket: WsWrapper,
+    appEvent: AppEvent,
+): Promise<MessageHandlerResult> => {
+    if (await handleCommonEvent(socket, appEvent)) return true
+    const session = getSession(socket)
+    if (!session) {
+        console.error("Session missing for socket " + socket.id)
+        return true
+    }
+    if (appEvent.action === "board.join") {
+        //await sleep(3000) // simulate latency
+
+        const boardInfo = await getBoardInfo(appEvent.boardId)
+        if (!boardInfo) {
+            console.warn(`Trying to join unknown board ${appEvent.boardId}`)
+            session.sendEvent({
+                action: "board.join.denied",
+                boardId: appEvent.boardId,
+                reason: "not found",
+            })
+            return true
+        }
+        const wsHost = boardInfo.ws_host ?? WS_HOST_DEFAULT
+
+        if (!allowedBoardId || appEvent.boardId !== allowedBoardId || !WS_HOST_LOCAL.includes(wsHost)) {
+            // Path - board id mismatch -> always redirect
+
+            const wsAddress = `${WS_PROTOCOL}://${wsHost}/socket/board/${appEvent.boardId}`
+            /* console.info(
+                    `Trying to join board ${appEvent.boardId} on socket for board ${allowedBoardId}, board host ${wsHost} local hostnames ${WS_HOST_LOCAL}`,
+            )*/
+
+            session.sendEvent({
+                action: "board.join.denied",
+                boardId: appEvent.boardId,
+                reason: "redirect",
+                wsAddress,
+            })
+            return true
+        }
+
+        let board = (await getBoard(appEvent.boardId))!
+
+        const accessPolicy = board.board.accessPolicy
+        const accessLevel = checkBoardAccess(accessPolicy, session.userInfo)
+        if (session.userInfo.userType === "authenticated") {
+            if (accessLevel === "none") {
+                console.warn("Access denied to board by user not on allowList")
+                session.sendEvent({
+                    action: "board.join.denied",
+                    boardId: appEvent.boardId,
+                    reason: "forbidden",
+                })
+                return true
+            } else {
+                await associateUserWithBoard(session.userInfo.userId, appEvent.boardId)
+            }
+        } else {
+            if (accessLevel === "none") {
+                console.warn("Access denied to board by anonymous user")
+                session.sendEvent({
+                    action: "board.join.denied",
+                    boardId: appEvent.boardId,
+                    reason: "unauthorized",
+                })
+                return true
+            }
+        }
+        await addSessionToBoard(board, socket, accessLevel, appEvent.initAtSerial)
+        return true
+    }
+
+    if (!session.boardSession) {
+        console.warn("Trying to send event to board without session", appEvent)
+        return true
+    }
+
+    if (isBoardItemEvent(appEvent)) {
+        const boardId = appEvent.boardId
+        const state = await getBoard(boardId)
+        if (!state) {
+            return true // Just ignoring for now, see above todo
+        }
+        if (!canWrite(session.boardSession.accessLevel)) {
+            console.warn("Trying to change read-only board")
+            return true
+        }
+        obtainLock(state.locks, appEvent, socket) // Allow even if was locked (offline use)
+        if (isPersistableBoardItemEvent(appEvent)) {
+            if (!session.isOnBoard(appEvent.boardId)) {
+                console.warn("Trying to send event to board without valid session")
+            } else {
+                let historyEntry: BoardHistoryEntry = {
+                    ...appEvent,
+                    user: session.userInfo,
+                    timestamp: newISOTimeStamp(),
+                }
+                try {
+                    const serial = updateBoards(state, historyEntry)
+                    historyEntry = { ...historyEntry, serial }
+                    broadcastBoardEvent(historyEntry, session)
+                    if (appEvent.action === "board.rename") {
+                        // special case: keeping name up to date as it's in a separate column
+                        await renameBoardConvenienceColumnOnly(appEvent.boardId, appEvent.name)
+                    }
+                    if (appEvent.action === "board.setAccessPolicy") {
+                        if (session.boardSession.accessLevel !== "admin") {
+                            console.warn("Trying to change access policy without admin access")
+                            return true
+                        }
+                        await updateBoardAccessPolicy(appEvent.boardId, appEvent.accessPolicy)
+                    }
+                    return { boardId, serial }
+                } catch (e) {
+                    console.warn(`Error applying event ${JSON.stringify(appEvent)}: ${e} -> forcing board refresh`)
+                    session.sendEvent({ action: "board.action.apply.failed" })
+                    return true
+                }
+            }
+        } else if (appEvent.action === "item.unlock") {
+            return true
+        }
+    } else {
+        switch (appEvent.action) {
+            case "cursor.move": {
+                const { boardId, position } = appEvent
+                const { x, y } = position
+                const state = maybeGetBoard(boardId)
+                if (state) {
+                    const session = getSession(socket)
+                    if (session && session.isOnBoard(appEvent.boardId)) {
+                        state.cursorPositions[socket.id] = { x, y, sessionId: socket.id }
+                        state.cursorsMoved = true
+                    }
+                }
+                return true
+            }
+            case "user.bringAllToMe": {
+                const session = getSession(socket)
+                const state = maybeGetBoard(appEvent.boardId)
+                if (session && state && session.isOnBoard(appEvent.boardId)) {
+                    if (session.sessionId !== appEvent.sessionId) {
+                        console.warn("Incorrect sessionId in user.bringAllToMe")
+                    } else {
+                        broadcastBoardEvent(appEvent, session)
+                    }
+                }
+                return true
+            }
+            case "asset.put.request": {
+                const { assetId } = appEvent
+                const signedUrl = getSignedPutUrl(assetId)
+                socket.send(toBuffer({ action: "asset.put.response", assetId, signedUrl }))
+                return true
+            }
+        }
+    }
+    return false
+}

backend/src/board-state.test.ts

@@ -0,0 +1,16 @@
+import { describe, expect, it } from "vitest"
+
+describe("board state iteration", () => {
+    it("is safe", () => {
+        // Checking that map value iteration is safe when deleteting items on the way
+        const boards = new Map<number, number>()
+        boards.set(1, 1)
+        boards.set(2, 2)
+        const results: number[] = []
+        for (let b of boards.values()) {
+            results.push(b)
+            boards.delete(b)
+        }
+        expect(results).toEqual([1, 2])
+    })
+})

backend/src/board-state.ts

@@ -0,0 +1,205 @@
+import { merge } from "lodash"
+import { boardReducer } from "../../common/src/board-reducer"
+import { Board, BoardCursorPositions, BoardHistoryEntry, Id } from "../../common/src/domain"
+import { sleep } from "../../common/src/sleep"
+import { createAccessToken, createBoard, fetchBoard, storeEventHistoryBundle } from "./board-store"
+import { quickCompactBoardHistory } from "./compact-history"
+import { Locks } from "./locker"
+import { UserSession, broadcastItemLocks, getBoardSessionCount, getSessionCount } from "./websocket-sessions"
+import * as Y from "yjs"
+import { inTransaction } from "./db"
+
+// A mutable state object for server side state
+export type ServerSideBoardState = {
+    ready: true
+    board: Board
+    recentEvents: BoardHistoryEntry[]
+    recentCrdtUpdate: Uint8Array | null
+    currentlyStoring: {
+        events: BoardHistoryEntry[]
+        crdtUpdate: Uint8Array | null
+    } | null
+    locks: ReturnType<typeof Locks>
+    cursorsMoved: boolean
+    cursorPositions: BoardCursorPositions
+    accessTokens: string[]
+    sessions: UserSession[]
+}
+
+export type ServerSideBoardStateInternal =
+    | ServerSideBoardState
+    | {
+          ready: false
+          fetch: Promise<ServerSideBoardState | null>
+      }
+
+let boards: Map<Id, ServerSideBoardStateInternal> = new Map()
+
+export async function getBoard(id: Id): Promise<ServerSideBoardState | null> {
+    let state = boards.get(id)
+    if (!state) {
+        console.log(`Loading board ${id} into memory`)
+        const fetchState = async () => {
+            const boardData = await fetchBoard(id)
+            if (!boardData) return null
+            const { board, accessTokens } = boardData
+            return {
+                ready: true,
+                board,
+                accessTokens,
+                recentEvents: [],
+                recentCrdtUpdate: null,
+                currentlyStoring: null,
+                locks: Locks((changedLocks) => broadcastItemLocks(id, changedLocks)),
+                cursorsMoved: false,
+                cursorPositions: {},
+                sessions: [],
+            } as ServerSideBoardState
+        }
+        const fetch = fetchState()
+        const temporaryState = {
+            ready: false as const,
+            fetch,
+        }
+        boards.set(id, temporaryState)
+        try {
+            const finalState = await fetch
+            if (!finalState) {
+                boards.delete(id)
+                return null
+            } else {
+                boards.set(id, finalState)
+                console.log(`Board loaded into memory: ${id}`)
+                return finalState
+            }
+        } catch (e) {
+            boards.delete(id)
+            // TODO: avoid retry loop
+            console.error(`Board load failed for board ${id}`)
+            throw e
+        }
+    } else if (!state.ready) {
+        return await state.fetch
+    } else {
+        return state
+    }
+}
+
+export function maybeGetBoard(id: Id): ServerSideBoardState | undefined {
+    const state = boards.get(id)
+    if (state?.ready) return state
+}
+
+export function updateBoards(boardState: ServerSideBoardState, appEvent: BoardHistoryEntry) {
+    const currentSerial = boardState.board.serial
+    const serial = currentSerial + 1
+    if (appEvent.serial !== undefined) {
+        throw Error("Event already has serial")
+    }
+    const eventWithSerial = { ...appEvent, serial }
+
+    const updatedBoard = boardReducer(boardState.board, eventWithSerial, { inplace: true, strictOnSerials: true })[0]
+
+    boardState.board = updatedBoard
+    boardState.recentEvents.push(eventWithSerial)
+    return serial
+}
+
+export function updateBoardCrdt(id: Id, crdtUpdate: Uint8Array) {
+    const boardState = maybeGetBoard(id)
+
+    if (!boardState) {
+        console.warn("CRDT update for board not loaded into memory", id)
+    } else {
+        boardState.recentCrdtUpdate = combineCrdtUpdates(boardState.recentCrdtUpdate, crdtUpdate)
+    }
+}
+
+export async function addBoard(board: Board, createToken?: boolean): Promise<ServerSideBoardState> {
+    await createBoard(board)
+    const accessTokens = createToken ? [await createAccessToken(board)] : []
+    const boardState = {
+        ready: true as const,
+        board,
+        serial: 0,
+        recentEvents: [],
+        recentCrdtUpdate: null,
+        currentlyStoring: null,
+        locks: Locks((changedLocks) => broadcastItemLocks(board.id, changedLocks)),
+        cursorsMoved: false,
+        cursorPositions: {},
+        accessTokens,
+        sessions: [],
+    }
+    boards.set(board.id, boardState)
+    return boardState
+}
+
+export function getActiveBoards() {
+    return [...boards.values()].filter((b) => b.ready) as ServerSideBoardState[]
+}
+
+let savingPromise: Promise<void> = saveBoards()
+
+async function saveBoards() {
+    await sleep(1000)
+    for (let state of boards.values()) {
+        if (state.ready) await saveBoardChanges(state)
+    }
+    savingPromise = saveBoards()
+}
+
+export async function awaitSavingChanges() {
+    await savingPromise
+}
+
+async function saveBoardChanges(state: ServerSideBoardState) {
+    if (state.recentEvents.length > 0 || state.recentCrdtUpdate !== null) {
+        if (state.currentlyStoring) {
+            throw Error("Invariant failed: storingEvents not empty")
+        }
+        const events = state.recentEvents.splice(0)
+        const crdtUpdate = state.recentCrdtUpdate
+        state.currentlyStoring = {
+            events,
+            crdtUpdate,
+        }
+        state.recentCrdtUpdate = null
+        console.log(
+            `Saving board ${state.board.id} at serial ${state.board.serial} with ${
+                state.currentlyStoring.events.length
+            } new events ${crdtUpdate ? "and CRDT update of size " + crdtUpdate.length : ""}`,
+        )
+        const lastSerial = state.board.serial
+        try {
+            await inTransaction((client) =>
+                storeEventHistoryBundle(state.board.id, events, lastSerial, crdtUpdate, client),
+            )
+        } catch (e) {
+            // Push event back to the head of save list for retrying later
+            state.recentEvents = [...state.currentlyStoring.events, ...state.recentEvents]
+            state.recentCrdtUpdate = merge(state.currentlyStoring.crdtUpdate, state.recentCrdtUpdate)
+            console.error("Board save failed for board", state.board.id, e)
+        }
+        state.currentlyStoring = null
+    }
+    if (state.recentEvents.length === 0 && getBoardSessionCount(state.board.id) === 0) {
+        console.log(`Purging board ${state.board.id} from memory`)
+        boards.delete(state.board.id)
+        await quickCompactBoardHistory(state.board.id)
+    }
+}
+
+export function combineCrdtUpdates(a: Uint8Array | null, b: Uint8Array | null) {
+    if (!a) return b
+    if (!b) return a
+    return Y.mergeUpdates([a, b])
+}
+
+export function getActiveBoardCount() {
+    return boards.size
+}
+
+setInterval(() => {
+    console.log("Statistics: active boards " + getActiveBoardCount() + ", sessions " + getSessionCount())
+}, 60000)

backend/src/board-store.ts

@@ -0,0 +1,481 @@
+import { PoolClient } from "pg"
+import QueryStream from "pg-query-stream"
+import * as uuid from "uuid"
+import { boardReducer } from "../../common/src/board-reducer"
+import { Board, BoardAccessPolicy, BoardHistoryEntry, Id, isBoardEmpty, Serial } from "../../common/src/domain"
+import { migrateBoard, migrateEvent, mkBootStrapEvent } from "../../common/src/migration"
+import { inTransaction, withDBClient } from "./db"
+import { assertNotNull } from "../../common/src/assertNotNull"
+
+export type BoardAndAccessTokens = {
+    board: Board
+    accessTokens: string[]
+}
+
+export type BoardInfo = {
+    id: Id
+    name: string
+    ws_host: string | null
+}
+
+export async function getBoardInfo(id: Id): Promise<BoardInfo | null> {
+    const result = await withDBClient((client) => client.query("SELECT id, name, ws_host FROM board WHERE id=$1", [id]))
+    return result.rows.length === 1 ? (result.rows[0] as BoardInfo) : null
+}
+
+const selectBoardQuery = `
+select id,
+    jsonb_set (content - 'accessPolicy', '{accessPolicy}', 
+    		cast(json_build_object(
+    			'allowList', (    			
+    				coalesce((
+					  select jsonb_agg(jsonb_strip_nulls(jsonb_build_object('domain', domain, 'access', access, 'email', email))) 
+					  from board_access
+					  where board_access.board_id = board.id
+					), '[]')    			
+    			), 
+    			'publicRead', public_read, 
+    			'publicWrite', public_write
+    		) as jsonb)) 
+    		as content
+from board
+where id=$1
+`
+
+export async function fetchBoard(id: Id): Promise<BoardAndAccessTokens | null> {
+    return await inTransaction(async (client) => {
+        const started = new Date().getTime()
+        const result = await client.query(selectBoardQuery, [id])
+        if (result.rows.length == 0) {
+            return null
+        } else {
+            const snapshot = result.rows[0].content as Board
+            if (
+                snapshot.accessPolicy &&
+                snapshot.accessPolicy.allowList.length === 0 &&
+                snapshot.accessPolicy.publicRead &&
+                snapshot.accessPolicy.publicWrite
+            ) {
+                // Effectively no access policy
+                delete snapshot.accessPolicy
+            }
+            let historyEventCount = 0
+            let lastSerial = 0
+            let board = snapshot
+
+            let i = 0
+            let rebuildingSnapshot = false
+            function updateBoardWithEventChunk(chunk: BoardHistoryEntry[]) {
+                if (chunk.length === 0) {
+                    return // CRDT-only bundle
+                }
+                board = chunk.reduce((b, e) => {
+                    i++
+                    if (e.action === "board.setAccessPolicy") {
+                        // Don't process access policy event when fetching board
+                        // Access policy may have been changed in the database after the event
+                        // And the board table status is considered the master
+                        return { ...b, serial: assertNotNull(e.serial) }
+                    }
+                    return boardReducer(b, e, { inplace: true, strictOnSerials: !rebuildingSnapshot })[0]
+                }, board)
+                historyEventCount += chunk.length
+                lastSerial = chunk[chunk.length - 1].serial ?? snapshot.serial
+            }
+
+            await getBoardHistory(id, snapshot.serial, updateBoardWithEventChunk).catch(async (error) => {
+                console.error(error.message)
+                console.error(
+                    `Error applying board history for snapshot update for board ${id}. Loop index ${i}. Rebooting snapshot. This may be a lossy operation.`,
+                )
+                i = 0
+                board = { ...snapshot, items: {}, connections: [] }
+                rebuildingSnapshot = true
+                try {
+                    await getFullBoardHistory(id, client, updateBoardWithEventChunk)
+                } catch (e) {
+                    console.error(`Unable to reboot snapshot, failing at loop index ${i}. Giving up.`)
+
+                    // TODO: this board cannot be repaired automatically. We should block usage, or it will be
+                    // and endless loop. Local dev, board ee803db1-f41a-43c6-9e39-83057faace60.
+
+                    throw e
+                }
+            })
+
+            const serial = (historyEventCount > 0 ? lastSerial : snapshot.serial) || 0
+            const elapsed = new Date().getTime() - started
+            console.log(
+                `Loaded board ${id} at serial ${serial} from snapshot at serial ${snapshot.serial} and ${historyEventCount} events after snapshot. Took ${elapsed}ms`,
+            )
+
+            if (historyEventCount > 1000 || rebuildingSnapshot /* time to create a new snapshot*/) {
+                console.log(
+                    `Saving snapshot for board ${id} at serial ${serial}/${snapshot.serial} with ${historyEventCount} new events`,
+                )
+                await saveBoardSnapshot(mkSnapshot(board, serial), client)
+            }
+            const accessTokens = (
+                await client.query("SELECT token FROM board_api_token WHERE board_id=$1", [id])
+            ).rows.map((row) => row.token)
+
+            return { board: { ...board, serial }, accessTokens }
+        }
+    })
+}
+
+export async function createBoard(board: Board): Promise<void> {
+    await inTransaction(async (client) => {
+        const result = await client.query("SELECT id FROM board WHERE id=$1", [board.id])
+        if (result.rows.length > 0) throw Error("Board already exists: " + board.id)
+        await client.query(`INSERT INTO board(id, name, content) VALUES ($1, $2, $3)`, [
+            board.id,
+            board.name,
+            mkSnapshot(board, 0),
+        ])
+
+        await updateAccessPolicy(board.id, board.accessPolicy, client)
+
+        if (!isBoardEmpty(board)) {
+            console.log(`Creating non-empty board ${board.id} -> bootstrapping history`)
+            const event = mkBootStrapEvent(board.id, board)
+            storeEventHistoryBundle(board.id, [event], event.serial!, null, client)
+        }
+    })
+}
+
+async function updateAccessPolicy(boardId: string, accessPolicy: BoardAccessPolicy, client: PoolClient): Promise<void> {
+    const publicRead = accessPolicy ? !!accessPolicy.publicRead : true
+    const publicWrite = accessPolicy ? !!accessPolicy.publicWrite : true
+    await client.query(`UPDATE board SET public_read=$1, public_write=$2 WHERE id=$3`, [
+        publicRead,
+        publicWrite,
+        boardId,
+    ])
+    await client.query(`DELETE FROM board_access WHERE board_id=$1`, [boardId])
+    if (accessPolicy) {
+        for (const entry of accessPolicy.allowList) {
+            const domain = "domain" in entry ? entry.domain : null
+            const email = "email" in entry ? entry.email : null
+            await client.query(`INSERT INTO BOARD_access (board_id, domain, email, access) VALUES ($1, $2, $3, $4)`, [
+                boardId,
+                domain,
+                email,
+                entry.access,
+            ])
+        }
+    }
+}
+
+// Updates the name column, which is there for admin convenience (not used by application logic)
+export async function renameBoardConvenienceColumnOnly(boardId: Id, name: string) {
+    await inTransaction(async (client) => {
+        const result = await client.query("SELECT content FROM board WHERE id=$1", [boardId])
+        if (result.rows.length !== 1) throw Error("Board not found: " + boardId)
+        await client.query("UPDATE board SET name=$1 WHERE id=$2", [name, boardId])
+    })
+}
+
+export async function updateBoardAccessPolicy(boardId: Id, accessPolicy: BoardAccessPolicy) {
+    await inTransaction(async (client) => {
+        const result = await client.query("SELECT content FROM board WHERE id=$1", [boardId])
+        if (result.rows.length !== 1) throw Error("Board not found: " + boardId)
+        await updateAccessPolicy(boardId, accessPolicy, client)
+    })
+}
+
+export async function createAccessToken(board: Board): Promise<string> {
+    const token = uuid.v4()
+    await inTransaction(async (client) =>
+        client.query("INSERT INTO board_api_token (board_id, token) VALUES ($1, $2)", [board.id, token]),
+    )
+    return token
+}
+
+type StreamingBoardEventCallback = (chunk: BoardHistoryEntry[]) => void
+
+// Due to memory concerns we fetch board histories from DB as chunks,
+// which are currently implemented as sort of a poor-man's observable
+function streamingBoardEventsQuery(text: string, values: any[], client: PoolClient, cb: StreamingBoardEventCallback) {
+    return new Promise((resolve, reject) => {
+        const query = new QueryStream(text, values)
+        const stream = client.query(query)
+        stream.on("error", reject)
+        stream.on("end", resolve)
+        stream.on("data", (row) => {
+            try {
+                const chunk = row.events?.events as BoardHistoryEntry[] | undefined
+
+                if (!chunk) {
+                    throw Error(`Unexpected DB row value ${chunk}`)
+                }
+
+                cb(chunk.map(migrateEvent))
+            } catch (error) {
+                console.error(error)
+                stream.destroy()
+                reject(error)
+            }
+        })
+    })
+}
+
+export function getFullBoardHistory(id: Id, client: PoolClient, cb: StreamingBoardEventCallback) {
+    return streamingBoardEventsQuery(
+        `
+        SELECT events 
+        FROM board_event 
+        WHERE board_id=$1 
+        ORDER BY last_serial, first_serial
+        `,
+        [id],
+        client,
+        cb,
+    )
+}
+
+export async function getBoardHistory(id: Id, afterSerial: Serial, cb: StreamingBoardEventCallback): Promise<void> {
+    await withDBClient(async (client) => {
+        let firstSerial = -1
+        let lastSerial = -1
+        let firstValidSerial = -1
+        await streamingBoardEventsQuery(
+            `
+            SELECT events 
+            FROM board_event 
+            WHERE board_id=$1 AND last_serial >= $2 
+            ORDER BY last_serial, first_serial
+            `,
+            [id, afterSerial],
+            client,
+            (chunk) => {
+                if (chunk.length === 0) {
+                    return // CRDT-only bundle
+                }
+                if (firstSerial === -1 && typeof chunk[0]?.serial === "number") {
+                    firstSerial = chunk[0]?.serial
+                }
+                lastSerial = chunk[chunk.length - 1].serial ?? -1
+
+                const validEventsAfter = chunk.filter((r) => r.serial! > afterSerial)
+                if (validEventsAfter.length === 0) {
+                    // Got chunk where no events have serial greater than the snapshot point -- discard it
+                    return
+                }
+
+                if (firstValidSerial === -1 && typeof validEventsAfter[0].serial === "number") {
+                    firstValidSerial = validEventsAfter[0].serial
+                }
+                cb(validEventsAfter)
+                return
+            },
+        )
+
+        // Client is up to date, ok
+        if (lastSerial === afterSerial) {
+            return
+        }
+
+        // Found continuous history, ok
+        if (firstValidSerial === afterSerial + 1) {
+            return
+        }
+
+        if (firstValidSerial === -1) {
+            if (afterSerial === 0) {
+                // Requesting from start, zero events found, is ok
+                return
+            }
+            // Client claims to be in the future, not ok
+            throw Error(
+                `Cannot find history to start after the requested serial ${afterSerial} for board ${id}. Seems like the requested serial is higher than currently stored in DB`,
+            )
+        }
+
+        // Found noncontinuous event timeline, not ok
+        throw Error(
+            `Cannot find history to start after the requested serial ${afterSerial} for board ${id}. Found history for ${firstValidSerial}..${lastSerial}`,
+        )
+    })
+}
+
+export function verifyContinuity(boardId: Id, init: Serial, ...histories: BoardHistoryEntry[][]) {
+    for (let history of histories) {
+        if (history.length > 0) {
+            if (!verifyTwoPoints(boardId, init, history[0].serial!)) {
+                return false
+            }
+            init = history[history.length - 1].serial!
+        }
+    }
+    return true
+}
+
+export function verifyEventArrayContinuity(boardId: Id, init: Serial, events: BoardHistoryEntry[]) {
+    for (let event of events) {
+        if (!verifyTwoPoints(boardId, init, event.serial!)) {
+            return false
+        }
+        init = event.serial!
+    }
+    return true
+}
+
+function verifyTwoPoints(boardId: Id, a: Serial, b: Serial) {
+    if (b !== a + 1) {
+        console.error(`History discontinuity: ${a} -> ${b} for board ${boardId}`)
+        return false
+    }
+    return true
+}
+
+function mkSnapshot(board: Board, serial: Serial) {
+    const { accessPolicy, ...result } = migrateBoard({ ...board, serial })
+    return result
+}
+
+export async function saveBoardSnapshot(board: Board, client: PoolClient) {
+    console.log(`Save board snapshot ${board.id} at serial ${board.serial}`)
+    client.query(
+        `
+    UPDATE board 
+    SET name=$2, content=$3 
+    WHERE id=$1`,
+        [board.id, board.name, board],
+    )
+}
+
+export async function storeEventHistoryBundle(
+    boardId: Id,
+    events: BoardHistoryEntry[],
+    lastSerial: Serial, // Needed in case events is empty
+    crdtUpdate: Uint8Array | null,
+    client: PoolClient,
+    savedAt = new Date(),
+) {
+    if (events.length === 0 && crdtUpdate === null) {
+        throw Error("Trying to store a bundle without events or crdtUpdate")
+    }
+    if (events[0]?.firstSerial !== undefined) {
+        throw Error("Assertion failed: folded events not expected on the server side.")
+    }
+    const firstSerial = events.length > 0 ? assertNotNull(events[0].serial) : null
+    if (events.length > 0 && events[events.length - 1].serial != lastSerial) {
+        throw Error("Serial mismatch between lastSerial and lastEvent.serial")
+    }
+    await client.query(
+        `
+        INSERT INTO board_event(board_id, first_serial, last_serial, events, crdt_update, saved_at) 
+        VALUES ($1, $2, $3, $4, $5, $6)
+        `,
+        [boardId, firstSerial, lastSerial, { events }, crdtUpdate, savedAt],
+    )
+}
+
+export async function storeCRDTOnlyEventHistoryBundle(
+    boardId: Id,
+    boardSerial: Serial,
+    crdtUpdate: Uint8Array | null,
+    client: PoolClient,
+    savedAt = new Date(),
+) {
+    await client.query(
+        `
+        INSERT INTO board_event(board_id, first_serial, last_serial, events, crdt_update, saved_at) 
+        VALUES ($1, $2, $3, $4, $5, $6)
+        `,
+        [boardId, null, boardSerial, null, crdtUpdate, savedAt],
+    )
+}
+
+export type BoardHistoryBundle = {
+    board_id: Id
+    last_serial: Serial
+    events: {
+        events: BoardHistoryEntry[]
+    }
+    crdt_update: Uint8Array | null
+}
+
+export async function getBoardHistoryBundlesWithLastSerialsBetween(
+    client: PoolClient,
+    id: Id,
+    lsMin: Serial,
+    lsMax: Serial,
+): Promise<BoardHistoryBundle[]> {
+    return (
+        await client.query(
+            `
+            SELECT board_id, last_serial, events, crdt_update 
+            FROM board_event 
+            WHERE board_id=$1 AND last_serial >= $2 AND last_serial <= $3 
+            ORDER BY last_serial, first_serial
+            `,
+            [id, lsMin, lsMax],
+        )
+    ).rows.map(migrateBundle)
+}
+
+export async function getBoardHistoryCrdtUpdates(client: PoolClient, id: Id): Promise<Uint8Array[]> {
+    return (
+        await client.query(
+            `
+            SELECT crdt_update 
+            FROM board_event 
+            WHERE board_id=$1 AND crdt_update IS NOT NULL 
+            ORDER BY last_serial, first_serial
+            `,
+            [id],
+        )
+    ).rows.map((row) => row.crdt_update)
+}
+
+function migrateBundle(b: BoardHistoryBundle): BoardHistoryBundle {
+    return { ...b, events: { ...b.events, events: b.events.events.map(migrateEvent) } }
+}
+
+export type BoardHistoryBundleMeta = {
+    board_id: Id
+    first_serial: Serial | null
+    last_serial: Serial
+    saved_at: Date
+}
+
+export async function getBoardHistoryBundleMetas(client: PoolClient, id: Id): Promise<BoardHistoryBundleMeta[]> {
+    return (
+        await client.query(
+            `
+            SELECT board_id, last_serial, first_serial, saved_at 
+            FROM board_event 
+            WHERE board_id=$1 
+            ORDER BY last_serial, first_serial
+            `,
+            [id],
+        )
+    ).rows
+}
+
+export function verifyContinuityFromMetas(boardId: Id, init: Serial, bundles: BoardHistoryBundleMeta[]) {
+    for (let bundle of bundles) {
+        if (!bundle.first_serial) {
+            // CRDT only bundle
+            if (bundle.last_serial !== init) {
+                console.error(
+                    `History discontinuity:  ${init} -> ${bundle.last_serial} for CRDT-only bundle for board ${boardId}`,
+                )
+                return false
+            }
+        } else {
+            if (!verifyTwoPoints(boardId, init, bundle.first_serial)) {
+                return false
+            }
+            init = bundle.last_serial
+        }
+    }
+    return true
+}
+
+export async function findAllBoards(client: PoolClient): Promise<Id[]> {
+    const result = await client.query("SELECT id FROM board")
+    return result.rows.map((row) => row.id)
+}

backend/src/board-yjs-server.ts

@@ -0,0 +1,102 @@
+import expressWs from "express-ws"
+import * as Y from "yjs"
+import { updateBoardCrdt } from "./board-state"
+import { getBoardHistoryCrdtUpdates } from "./board-store"
+import { withDBClient } from "./db"
+import { getSessionIdFromCookies } from "./http-session"
+import { getSessionById } from "./websocket-sessions"
+import YWebSocketServer from "./y-websocket-server/YWebSocketServer"
+import * as WebSocket from "ws"
+import { canRead, canWrite } from "../../common/src/domain"
+
+const socketsBySessionId: Record<string, WebSocket[]> = {}
+
+export function closeYjsSocketsBySessionId(sessionId: string) {
+    const sockets = socketsBySessionId[sessionId]
+    if (sockets) {
+        for (const socket of sockets) {
+            socket.close()
+        }
+        delete socketsBySessionId[sessionId]
+        console.log(
+            `CLOSED ${sockets.length} y.js sockets by session id ${sessionId} - remaining sockets exist for ${
+                Object.keys(socketsBySessionId).length
+            } other sessions`,
+        )
+    }
+}
+
+export const yWebSocketServer = new YWebSocketServer({
+    persistence: {
+        bindState: async (docName, ydoc) => {
+            const boardId = docName
+            const updates = await withDBClient(async (client) => getBoardHistoryCrdtUpdates(client, boardId))
+
+            if (updates.length === 0) {
+                const initUpdate = Y.encodeStateAsUpdate(ydoc)
+                console.log(`Storing initial CRDT state to DB for board ${boardId}`)
+                updateBoardCrdt(boardId, initUpdate)
+            } else {
+                console.log(`Loaded ${updates.length} CRDT updates from DB for board ${boardId}`)
+                for (const update of updates) {
+                    Y.applyUpdate(ydoc, update)
+                }
+            }
+
+            ydoc.on("update", (update: Uint8Array, origin: any, doc: Y.Doc) => {
+                updateBoardCrdt(boardId, update)
+            })
+        },
+        writeState: async (docName, ydoc) => {
+            // TODO: needed?
+        },
+    },
+})
+
+export function BoardYJSServer(ws: expressWs.Instance, path: string) {
+    ws.app.ws(path, async (socket, req) => {
+        const boardId = req.params.boardId
+        const sessionId = getSessionIdFromCookies(req)
+        const session = sessionId ? getSessionById(sessionId) : undefined
+        if (
+            !sessionId ||
+            !session ||
+            !session.boardSession ||
+            session.boardSession.boardId !== boardId ||
+            !canRead(session.boardSession.accessLevel)
+        ) {
+            // TODO: implement read-only YJS connections
+            //console.warn("No session for YJS connection for board", boardId)
+            socket.close()
+            return
+        }
+        if (!socketsBySessionId[sessionId]) {
+            socketsBySessionId[sessionId] = []
+        }
+        socketsBySessionId[sessionId].push(socket)
+        console.log(
+            `OPENED y.js connection for session ${sessionId}. Now sockets exist for ${
+                Object.keys(socketsBySessionId).length
+            } sessions`,
+        )
+        socket.addEventListener("close", () => {
+            if (socketsBySessionId[sessionId]) {
+                socketsBySessionId[sessionId] = socketsBySessionId[sessionId].filter((s) => s !== socket)
+                if (socketsBySessionId[sessionId].length === 0) {
+                    delete socketsBySessionId[sessionId]
+                }
+                console.log(
+                    `CLOSED y.js connection. Now sockets exist for ${Object.keys(socketsBySessionId).length} sessions`,
+                )
+            }
+        })
+        const readOnly = !canWrite(session.boardSession.accessLevel)
+        const docName = boardId
+        try {
+            await yWebSocketServer.setupWSConnection(socket, docName, readOnly)
+        } catch (e) {
+            console.error("Error setting up YJS connection", e)
+            socket.close()
+        }
+    })
+}

backend/src/common-event-handler.ts

@@ -0,0 +1,114 @@
+import * as Y from "yjs"
+import { AppEvent, Board, CrdtEnabled, checkBoardAccess, defaultBoardSize } from "../../common/src/domain"
+import { addBoard } from "./board-state"
+import { fetchBoard } from "./board-store"
+import { yWebSocketServer } from "./board-yjs-server"
+import { MessageHandlerResult } from "./connection-handler"
+import { getAuthenticatedUserFromJWT } from "./http-session"
+import {
+    associateUserWithBoard,
+    dissociateUserWithBoard,
+    getUserAssociatedBoards,
+    getUserIdForEmail,
+} from "./user-store"
+import { getSession, logoutUser, setNicknameForSession, setVerifiedUserForSession } from "./websocket-sessions"
+import { WsWrapper, toBuffer } from "./ws-wrapper"
+
+export async function handleCommonEvent(socket: WsWrapper, appEvent: AppEvent): Promise<MessageHandlerResult> {
+    switch (appEvent.action) {
+        case "auth.login.jwt": {
+            const user = getAuthenticatedUserFromJWT(appEvent.jwt)
+            const session = getSession(socket)
+            if (session && user !== null) {
+                const userId = await getUserIdForEmail(user.email)
+                const userInfo = await setVerifiedUserForSession(user, session)
+                console.log(`${user.name} logged in`)
+                session.sendEvent({ action: "auth.login.response", success: true, userId })
+                if (session.boardSession) {
+                    await associateUserWithBoard(userId, session.boardSession.boardId)
+                }
+                session.sendEvent({
+                    action: "user.boards",
+                    email: user.email,
+                    boards: await getUserAssociatedBoards(userInfo),
+                })
+            } else if (session) {
+                session.sendEvent({ action: "auth.login.response", success: false })
+            }
+            return true
+        }
+        case "auth.logout": {
+            const session = getSession(socket)
+            if (session && session.userInfo.userType === "authenticated") {
+                logoutUser(appEvent, socket)
+                console.log(`${session.userInfo.name} logged out`)
+            }
+            socket.close()
+            return true
+        }
+        case "nickname.set": {
+            setNicknameForSession(appEvent, socket)
+            return true
+        }
+        case "board.associate": {
+            // TODO: maybe access check? Not security-wise necessary
+            const session = getSession(socket)
+            if (session) {
+                if (session.userInfo.userType !== "authenticated") {
+                    console.warn("Trying to associate board without authenticated user")
+                    return true
+                }
+                const userId = session.userInfo.userId
+                await associateUserWithBoard(userId, appEvent.boardId, appEvent.lastOpened)
+            }
+            return true
+        }
+        case "board.dissociate": {
+            const session = getSession(socket)
+            if (session) {
+                if (session.userInfo.userType !== "authenticated") {
+                    console.warn("Trying to dissociate board without authenticated user")
+                    return true
+                }
+                const userId = session.userInfo.userId
+                await dissociateUserWithBoard(userId, appEvent.boardId)
+            }
+            return true
+        }
+        case "board.add": {
+            const session = getSession(socket)
+            if (session) {
+                const { payload } = appEvent
+                let template: Board | null = null
+                if ("templateId" in payload && payload.templateId) {
+                    const aliased = process.env[`BOARD_ALIAS_${payload.templateId}`]
+                    const templateId = aliased || payload.templateId
+                    const found = await fetchBoard(templateId)
+                    if (found) {
+                        const accessLevel = checkBoardAccess(found.board.accessPolicy, session.userInfo)
+                        if (accessLevel === "none") {
+                            console.warn(`Trying to use board ${found.board.id} as template, without board permissions`)
+                            return true
+                        }
+                        template = { ...found.board, accessPolicy: undefined }
+                    } else {
+                        console.error(`Template ${payload.templateId}${aliased ? `(${templateId})` : ""} not found`)
+                    }
+                }
+                const board = { ...defaultBoardSize, items: {}, connections: [], ...template, ...payload, serial: 0 }
+                if (template && template.crdt === CrdtEnabled) {
+                    const templateDoc = await yWebSocketServer.docs.getYDocAndWaitForFetch(template.id)
+                    const newDoc = await yWebSocketServer.docs.getYDocAndWaitForFetch(board.id)
+                    Y.applyUpdate(newDoc, Y.encodeStateAsUpdate(templateDoc))
+                }
+                await addBoard(board)
+                socket.send(toBuffer({ action: "board.add.ack", boardId: board.id }))
+            }
+            return true
+        }
+        case "ping": {
+            return true
+        }
+    }
+    return false
+}

backend/src/compact-history.ts

@@ -0,0 +1,116 @@
+import { format } from "date-fns"
+import _ from "lodash"
+import { BoardHistoryEntry, Id } from "../../common/src/domain"
+import {
+    BoardHistoryBundleMeta,
+    getBoardHistoryBundleMetas,
+    getBoardHistoryBundlesWithLastSerialsBetween,
+    storeEventHistoryBundle,
+    verifyContinuity,
+    verifyContinuityFromMetas,
+    verifyEventArrayContinuity,
+} from "./board-store"
+import * as Y from "yjs"
+import { inTransaction } from "./db"
+
+function chunkBy<T>(arr: T[], shouldSplit: (a: T, b: T) => boolean) {
+    const result = []
+    let currentChunk = []
+    for (let i = 0; i < arr.length; i++) {
+        if (i > 0 && shouldSplit(arr[i - 1], arr[i])) {
+            result.push(currentChunk)
+            currentChunk = []
+        }
+        currentChunk.push(arr[i])
+    }
+    result.push(currentChunk)
+    return result
+}
+function getHour(b: BoardHistoryBundleMeta) {
+    return format(new Date(b.saved_at), "yyyy-MM-dd hh")
+}
+
+export async function quickCompactBoardHistory(id: Id): Promise<number> {
+    try {
+        return await inTransaction(async (client) => {
+            // Lock the board to prevent loading the board while compacting
+            await client.query("select 1 from board where id=$1 for update", [id])
+            const bundleMetas = await getBoardHistoryBundleMetas(client, id)
+            if (bundleMetas.length === 0) return 0
+            const consistent = verifyContinuityFromMetas(id, 0, bundleMetas)
+            if (consistent) {
+                // Group in one-hour bundles
+                //console.log("Grouped by date", groupedByHour)
+                const toCompact = chunkBy(
+                    bundleMetas,
+                    (a, b) => getHour(a) !== getHour(b) && a.last_serial !== b.last_serial,
+                ).filter((chunk) => chunk.length > 1)
+                let compactions = 0
+                for (let bs of toCompact) {
+                    const firstBundle = bs[0]
+                    const lastBundle = bs[bs.length - 1]
+                    console.log(
+                        `Compacting ${bs.length} bundles into one for board ${id}, containing serials ${firstBundle.first_serial}...${lastBundle.last_serial}`,
+                    )
+                    const lastSerial = lastBundle.last_serial
+                    const bundlesWithData = await getBoardHistoryBundlesWithLastSerialsBetween(
+                        client,
+                        id,
+                        firstBundle.last_serial,
+                        lastSerial,
+                    )
+                    const eventArrays = bundlesWithData.map((b) => b.events.events)
+                    const events: BoardHistoryEntry[] = eventArrays.flat()
+                    const crdtUpdates = bundlesWithData.flatMap((d) => (d.crdt_update ? [d.crdt_update] : []))
+                    const combinedCrdtUpdate = crdtUpdates.length ? Y.mergeUpdates(crdtUpdates) : null
+                    const initSerial = firstBundle.first_serial
+                        ? firstBundle.first_serial - 1
+                        : firstBundle.last_serial - 1
+                    const consistent =
+                        verifyContinuity(id, initSerial, ...eventArrays) &&
+                        verifyEventArrayContinuity(id, initSerial, events)
+                    if (consistent && bundlesWithData.length == bs.length) {
+                        // 1. delete existing bundles
+                        const deleteResult = await client.query(
+                            `DELETE FROM board_event where board_id=$1 and last_serial in (${bundlesWithData
+                                .map((b) => b.last_serial)
+                                .join(",")})`,
+                            [id],
+                        )
+                        if (deleteResult.rowCount != bs.length) {
+                            throw Error(
+                                `Unexpected rowcount when deleting on compaction: ${deleteResult.rowCount} for board ${id}`,
+                            )
+                        }
+                        // 2. store as a single bundle
+                        await storeEventHistoryBundle(
+                            id,
+                            events,
+                            lastSerial,
+                            combinedCrdtUpdate,
+                            client,
+                            lastBundle.saved_at,
+                        )
+                    } else {
+                        throw Error("Discontinuity detected in compacted history.")
+                    }
+                    compactions++
+                }
+                if (compactions > 0) {
+                } else {
+                    console.log(
+                        `Board ${id}: Verified ${bundleMetas.length} bundles containing ${
+                            bundleMetas[bundleMetas.length - 1].last_serial
+                        } events => no need to compact`,
+                    )
+                }
+                return compactions
+            } else {
+                throw Error("Discontinuity detected in bundle metadata.")
+            }
+        })
+    } catch (e) {
+        console.error(`Aborting compaction of board ${id} because of an error: ${e}`)
+        return 0
+    }
+}

backend/src/config.ts

@@ -0,0 +1,39 @@
+import path from "path"
+import fs from "fs"
+import { authProvider } from "./oauth"
+import * as t from "io-ts"
+import { optional } from "../../common/src/domain"
+import { decodeOrThrow } from "./decodeOrThrow"
+
+export type StorageBackend = Readonly<
+    { type: "LOCAL"; directory: string; assetStorageURL: string } | { type: "AWS"; assetStorageURL: string }
+>
+export type Config = Readonly<{ storageBackend: StorageBackend; authSupported: boolean; crdt: CrdtConfigString }>
+
+const CrdtConfigString = t.union([
+    t.literal("opt-in"),
+    t.literal("opt-in-authenticated"),
+    t.literal("true"),
+    t.literal("false"),
+])
+export type CrdtConfigString = t.TypeOf<typeof CrdtConfigString>
+
+export const getConfig = (): Config => {
+    const storageBackend: StorageBackend = process.env.AWS_ASSETS_BUCKET_URL
+        ? { type: "AWS", assetStorageURL: process.env.AWS_ASSETS_BUCKET_URL }
+        : { type: "LOCAL", directory: path.resolve("localfiles"), assetStorageURL: "/assets" }
+
+    if (storageBackend.type === "LOCAL") {
+        try {
+            fs.mkdirSync(storageBackend.directory)
+        } catch (e) {}
+    }
+
+    const crdt = decodeOrThrow(CrdtConfigString, process.env.COLLABORATIVE_EDITING ?? "opt-in")
+
+    return {
+        storageBackend,
+        authSupported: authProvider !== null,
+        crdt,
+    }
+}

backend/src/connection-handler.ts

@@ -0,0 +1,68 @@
+import { AppEvent, Id, Serial, EventWrapper } from "../../common/src/domain"
+import { getActiveBoards } from "./board-state"
+import { getConfig } from "./config"
+import { releaseLocksFor } from "./locker"
+import { broadcastCursorPositions, endSession, startSession } from "./websocket-sessions"
+import { WsWrapper, toBuffer } from "./ws-wrapper"
+
+export type ConnectionHandlerParams = Readonly<{
+    getSignedPutUrl: (key: string) => string
+}>
+
+export const connectionHandler = (socket: WsWrapper, handleMessage: MessageHandler) => {
+    startSession(socket)
+    const config = getConfig()
+    socket.send(
+        toBuffer({
+            action: "server.config",
+            assetStorageURL: config.storageBackend.assetStorageURL,
+            authSupported: config.authSupported,
+            crdt: config.crdt,
+        }),
+    )
+    socket.onError(() => {
+        socket.close()
+    })
+    socket.onMessage(async (o: object) => {
+        try {
+            let event = o as EventWrapper
+            let serialsToAck: Record<Id, Serial> = {}
+            for (const e of event.events) {
+                const serialAck = await handleMessage(socket, e)
+                if (serialAck === true) {
+                } else if (serialAck === false) {
+                    console.warn("Unhandled app-event message", e)
+                } else {
+                    serialsToAck[serialAck.boardId] = serialAck.serial
+                }
+            }
+            if (event.ackId) {
+                socket.send(toBuffer({ action: "ack", ackId: event.ackId, serials: serialsToAck }))
+            }
+        } catch (e) {
+            console.error("Error while handling event from client. Closing connection.", e)
+            socket.close()
+        }
+    })
+
+    socket.onClose(() => {
+        endSession(socket)
+        getActiveBoards().forEach((state) => {
+            delete state.cursorPositions[socket.id]
+            state.cursorsMoved = true
+        })
+        releaseLocksFor(socket)
+    })
+}
+
+setInterval(() => {
+    getActiveBoards().forEach((bh) => {
+        if (bh.cursorsMoved) {
+            broadcastCursorPositions(bh.board.id, bh.cursorPositions)
+            bh.cursorsMoved = false
+        }
+    })
+}, 100)
+
+export type MessageHandler = (socket: WsWrapper, appEvent: AppEvent) => Promise<MessageHandlerResult>
+export type MessageHandlerResult = { boardId: Id; serial: Serial } | boolean

backend/src/db.ts

@@ -0,0 +1,72 @@
+import pg, { PoolClient } from "pg"
+import process from "process"
+import migrate from "node-pg-migrate"
+
+const DATABASE_URL = process.env.DATABASE_URL ?? "postgres://r-board:secret@127.0.0.1:13338/r-board"
+const DATABASE_SSL_ENABLED = process.env.DATABASE_SSL_ENABLED === "true"
+
+const pgConfig = {
+    connectionString: DATABASE_URL,
+    ssl: DATABASE_SSL_ENABLED
+        ? {
+              rejectUnauthorized: false,
+          }
+        : undefined,
+}
+const connectionPool = new pg.Pool(pgConfig)
+
+export function closeConnectionPool() {
+    connectionPool.end()
+}
+
+export async function initDB(backendDir: string = ".") {
+    console.log("Running database migrations")
+    await inTransaction((client) =>
+        migrate({
+            count: 100000,
+            databaseUrl: DATABASE_URL,
+            migrationsTable: "pgmigrations",
+            dir: `${backendDir}/migrations`,
+            direction: "up",
+            dbClient: client,
+        }),
+    )
+    console.log("Completed database migrations")
+
+    return {
+        onEvent: async (eventNames: string[], cb: (n: pg.Notification) => any) => {
+            const client = await connectionPool.connect()
+            eventNames.map((e) => client.query(`LISTEN ${e}`))
+            client.on("notification", cb)
+        },
+    }
+}
+
+export async function withDBClient<T>(f: (client: PoolClient) => Promise<T>): Promise<T> {
+    const client = await connectionPool.connect()
+    try {
+        await client.query("BEGIN;SET SESSION CHARACTERISTICS AS TRANSACTION READ ONLY;")
+        return await f(client)
+    } finally {
+        await client.query("ROLLBACK;")
+        client.release()
+    }
+}
+
+export async function inTransaction<T>(f: (client: PoolClient) => Promise<T>): Promise<T> {
+    const client = await connectionPool.connect()
+    try {
+        await client.query(`
+            BEGIN;
+            SET SESSION CHARACTERISTICS AS TRANSACTION READ WRITE;
+        `)
+        const result = await f(client)
+        await client.query("COMMIT;")
+        return result
+    } catch (e) {
+        await client.query("ROLLBACK;")
+        throw e
+    } finally {
+        client.release()
+    }
+}

backend/src/decodeOrThrow.ts

@@ -0,0 +1,21 @@
+import * as t from "io-ts"
+import { Left, isLeft, left } from "fp-ts/lib/Either"
+import { PathReporter } from "io-ts/lib/PathReporter"
+
+export function decodeOrThrow<T>(codec: t.Type<T, any>, input: any): T {
+    const validationResult = codec.decode(input)
+    if (isLeft(validationResult)) {
+        throw new ValidationError(validationResult)
+    }
+    return validationResult.right
+}
+
+class ValidationError extends Error {
+    constructor(errors: Left<t.Errors>) {
+        super(report_(errors.left))
+    }
+}
+
+function report_(errors: t.Errors) {
+    return PathReporter.report(left(errors)).join("\n")
+}

backend/src/env.ts

@@ -0,0 +1,7 @@
+import process from "process"
+
+export function getEnv(name: string): string {
+    const value = process.env[name]
+    if (!value) throw new Error("Missing ENV: " + name)
+    return value
+}

backend/src/expiring-map.ts

@@ -0,0 +1,53 @@
+export function AutoExpiringMap<V extends any>(ttlSeconds: number) {
+    const timers = new Map<string | number, NodeJS.Timeout | undefined>()
+    const data: Record<string, V> = {}
+    const listeners: ((v: Record<string, V>) => any)[] = []
+    const proxy = new Proxy(data, {
+        set(target, key, value) {
+            if (typeof key === "symbol") return false
+            target[key] = value
+            setExpiryTimer(key)
+            listeners.forEach((l) => l(target))
+            return true
+        },
+        deleteProperty(target, key) {
+            if (typeof key === "symbol") return false
+            const didDelete = delete target[key]
+            if (!didDelete) {
+                return false
+            }
+            listeners.forEach((l) => l(target))
+            return true
+        },
+    })
+
+    const setExpiryTimer = (key: string | number) => {
+        if (timers.has(key)) {
+            clearTimeout(timers.get(key)!)
+        }
+
+        timers.set(
+            key,
+            setTimeout(() => {
+                timers.delete(key)
+                delete proxy[key]
+            }, ttlSeconds * 1000),
+        )
+    }
+
+    const autoExpiringMap = {
+        get: (key: string) => proxy[key],
+        has: (key: string) => !!proxy[key],
+        entries: () => Object.entries(proxy),
+        delete: (key: string) => delete proxy[key],
+        set: (key: string, value: any) => {
+            proxy[key] = value
+        },
+        onChange: (fn: (v: Record<string, V>) => any) => {
+            listeners.push(fn)
+            return autoExpiringMap
+        },
+    }
+
+    return autoExpiringMap
+}

backend/src/express-server.ts

@@ -0,0 +1,162 @@
+import dotenv from "dotenv"
+import express from "express"
+import expressWs from "express-ws"
+import fs from "fs"
+import * as Http from "http"
+import * as Https from "https"
+import * as path from "path"
+
+import apiRoutes from "./api/api-routes"
+import { handleBoardEvent } from "./board-event-handler"
+import { BoardYJSServer } from "./board-yjs-server"
+import { getConfig } from "./config"
+import { connectionHandler } from "./connection-handler"
+import { getEnv } from "./env"
+import { authProvider, setupAuth } from "./oauth"
+
+import { possiblyRequireAuth } from "./require-auth"
+import { createGetSignedPutUrl } from "./storage"
+import { WsWrapper } from "./ws-wrapper"
+import Cookies from "cookies"
+import { removeAuthenticatedUser, setAuthenticatedUser } from "./http-session"
+
+dotenv.config()
+
+export const startExpressServer = (httpPort?: number, httpsPort?: number): (() => void) => {
+    const config = getConfig()
+
+    const app = express()
+
+    if (authProvider) {
+        setupAuth(app, authProvider)
+    } else {
+        app.get("/logout", async (req, res) => {
+            removeAuthenticatedUser(req, res)
+            res.redirect("/")
+        })
+    }
+    app.get("/test-callback", async (req, res) => {
+        const cookies = new Cookies(req, res)
+        const returnTo = cookies.get("returnTo") || "/"
+        setAuthenticatedUser(req, res, { domain: null, email: "ourboardtester@test.com", name: "Ourboard tester" })
+        res.redirect(returnTo)
+    })
+
+    possiblyRequireAuth(app)
+
+    app.use("/", express.static("../frontend/dist"))
+    app.use("/", express.static("../frontend/public"))
+
+    if (config.storageBackend.type === "LOCAL") {
+        const localDirectory = config.storageBackend.directory
+        app.put("/assets/:id", (req, res) => {
+            if (!req.params.id) {
+                return res.sendStatus(400)
+            }
+
+            const w = fs.createWriteStream(localDirectory + "/" + req.params.id)
+
+            req.pipe(w)
+
+            req.on("end", () => {
+                !res.headersSent && res.sendStatus(200)
+            })
+
+            w.on("error", () => {
+                res.sendStatus(500)
+            })
+        })
+        app.use("/assets", express.static(localDirectory))
+    }
+
+    app.get("/assets/external", (req, res) => {
+        const src = req.query.src
+        if (typeof src !== "string" || ["http://", "https://"].every((prefix) => !src.startsWith(prefix)))
+            return res.send(400)
+        const protocol = src.startsWith("https://") ? Https : Http
+
+        protocol
+            .request(src, (upstreamResponse) => {
+                res.writeHead(upstreamResponse.statusCode!, upstreamResponse.headers)
+                upstreamResponse
+                    .pipe(res, {
+                        end: true,
+                    })
+                    .on("error", (err) => res.status(500).send(err.message))
+            })
+            .end()
+    })
+
+    app.get("/b/:boardId", async (req, res) => {
+        res.sendFile(path.resolve("../frontend/dist/index.html"))
+    })
+
+    app.use(apiRoutes.handler())
+
+    let stop = () => {}
+
+    if (httpPort) {
+        const http = new Http.Server(app)
+        startWs(http, app)
+        http.listen(httpPort, () => {
+            console.log("Listening HTTP on port " + httpPort)
+        })
+        const prevStop = stop
+        stop = () => {
+            prevStop()
+            http.close()
+        }
+    }
+
+    if (httpsPort) {
+        let https = new Https.Server(
+            {
+                cert: fs.readFileSync(getEnv("HTTPS_CERT_FILE")),
+                key: fs.readFileSync(getEnv("HTTPS_KEY_FILE")),
+            },
+            app,
+        )
+        startWs(https, app)
+        https.listen(httpsPort, () => {
+            console.log("Listening HTTPS on port " + httpsPort)
+        })
+        const prevStop = stop
+        stop = () => {
+            prevStop()
+            https.close()
+        }
+    }
+
+    const redirectURL = process.env.REDIRECT_URL
+    if (redirectURL) {
+        app.get("*", function (req, res, next) {
+            if (req.headers["x-forwarded-proto"] !== "https") {
+                res.redirect(redirectURL)
+            } else {
+                next()
+            }
+        })
+    }
+    return stop
+}
+
+function startWs(http: any, app: express.Express) {
+    const ws: expressWs.Instance = expressWs(app, http)
+
+    const signedPutUrl = createGetSignedPutUrl(getConfig().storageBackend)
+
+    ws.app.ws("/socket/lobby", (socket, req) => {
+        connectionHandler(WsWrapper(socket), handleBoardEvent(null, signedPutUrl))
+    })
+    ws.app.ws("/socket/board/:boardId", (socket, req) => {
+        const boardId = req.params.boardId
+        connectionHandler(WsWrapper(socket), handleBoardEvent(boardId, signedPutUrl))
+    })
+
+    BoardYJSServer(ws, "/socket/yjs/board/:boardId/")
+
+    ws.app.ws("*", (socket, req) => {
+        console.warn(`Unexpected WS connection: ${req.url} `)
+        socket.close()
+    })
+}

backend/src/generic-oidc-auth.ts

@@ -0,0 +1,123 @@
+import { Request, Response } from "express"
+import * as t from "io-ts"
+import JWT from "jsonwebtoken"
+import { OAuthAuthenticatedUser } from "../../common/src/authenticated-user"
+import { optional } from "../../common/src/domain"
+import { decodeOrThrow } from "./decodeOrThrow"
+import { getEnv } from "./env"
+import { ROOT_URL } from "./host-config"
+import { AuthProvider } from "./oauth"
+import { REQUIRE_AUTH } from "./require-auth"
+
+type GenericOAuthConfig = {
+    OIDC_CONFIG_URL: string
+    OIDC_CLIENT_ID: string
+    OIDC_CLIENT_SECRET: string
+    OIDC_LOGOUT?: string
+}
+
+export const genericOIDCConfig: GenericOAuthConfig | null = process.env.OIDC_CONFIG_URL
+    ? {
+          OIDC_CONFIG_URL: getEnv("OIDC_CONFIG_URL"),
+          OIDC_CLIENT_ID: getEnv("OIDC_CLIENT_ID"),
+          OIDC_CLIENT_SECRET: getEnv("OIDC_CLIENT_SECRET"),
+          OIDC_LOGOUT: process.env.OIDC_LOGOUT,
+      }
+    : null
+
+export function GenericOIDCAuthProvider(config: GenericOAuthConfig): AuthProvider {
+    console.log(`Setting up generic OAuth authentication using client id ${config.OIDC_CLIENT_ID}`)
+
+    const callbackUrl = `${ROOT_URL}/google-callback`
+
+    const openIdConfiguration = (async () => {
+        const response = await fetch(config.OIDC_CONFIG_URL)
+        return decodeOrThrow(OpenIdConfiguration, await response.json())
+    })()
+
+    async function getAccountFromCode(code: string): Promise<OAuthAuthenticatedUser> {
+        const response = await fetch((await openIdConfiguration).token_endpoint, {
+            method: "POST",
+            headers: {
+                "content-type": "application/x-www-form-urlencoded",
+            },
+            body: `grant_type=authorization_code&code=${encodeURIComponent(code)}&client_id=${encodeURIComponent(
+                config.OIDC_CLIENT_ID,
+            )}&client_secret=${config.OIDC_CLIENT_SECRET}&redirect_uri=${callbackUrl}`,
+        })
+
+        const body = await response.json()
+
+        const idToken = JWT.decode(body.id_token)
+        //console.log(JSON.stringify(idToken, null, 2))
+        const user = decodeOrThrow(IdToken, idToken)
+        return {
+            email: user.email,
+            name: "name" in user ? user.name : user.preferred_username,
+            picture: user.picture ?? undefined,
+            domain: user.hd ?? null,
+        }
+    }
+
+    async function getAuthPageURL() {
+        const scopes = "email openid profile"
+        const state = "TODO"
+        const redirectUri = callbackUrl
+        return `${(await openIdConfiguration).authorization_endpoint}?scope=${encodeURIComponent(
+            scopes,
+        )}&response_type=code&state=${encodeURIComponent(state)}&redirect_uri=${encodeURIComponent(
+            redirectUri,
+        )}&client_id=${config.OIDC_CLIENT_ID}`
+    }
+
+    const shouldHandleLogout = REQUIRE_AUTH || config.OIDC_LOGOUT
+
+    async function getLogoutUrl(): Promise<string> {
+        if (config.OIDC_LOGOUT && config.OIDC_LOGOUT !== "true") {
+            return config.OIDC_LOGOUT
+        }
+
+        const logoutUrl = (await openIdConfiguration).end_session_endpoint
+
+        if (!logoutUrl) {
+            throw Error(
+                `OIDC configuration at ${config.OIDC_CONFIG_URL} does not specify end_session_endpoint. Use OIDC_LOGOUT environment variable to define the logout endpoint explicitly.`,
+            )
+        }
+
+        return logoutUrl
+    }
+
+    const logout = shouldHandleLogout
+        ? async (req: Request, res: Response) => {
+              res.redirect(await getLogoutUrl())
+          }
+        : undefined
+
+    return {
+        getAccountFromCode,
+        getAuthPageURL,
+        logout,
+    }
+}
+
+const OpenIdConfiguration = t.type({
+    authorization_endpoint: t.string,
+    token_endpoint: t.string,
+    end_session_endpoint: optional(t.string),
+})
+
+const IdToken = t.union([
+    t.type({
+        email: t.string,
+        name: t.string,
+        picture: optional(t.string),
+        hd: optional(t.string),
+    }),
+    t.type({
+        email: t.string,
+        preferred_username: t.string,
+        picture: optional(t.string),
+        hd: optional(t.string),
+    }),
+])

backend/src/github-webhook/example-payload.json

@@ -0,0 +1,239 @@
+{
+    "action": "assigned",
+    "issue": {
+        "url": "https://api.github.com/repos/raimohanska/r-board/issues/129",
+        "repository_url": "https://api.github.com/repos/raimohanska/r-board",
+        "labels_url": "https://api.github.com/repos/raimohanska/r-board/issues/129/labels{/name}",
+        "comments_url": "https://api.github.com/repos/raimohanska/r-board/issues/129/comments",
+        "events_url": "https://api.github.com/repos/raimohanska/r-board/issues/129/events",
+        "html_url": "https://github.com/raimohanska/r-board/issues/129",
+        "id": 810436967,
+        "node_id": "MDU6SXNzdWU4MTA0MzY5Njc=",
+        "number": 129,
+        "title": "Github issues integration",
+        "user": {
+            "login": "raimohanska",
+            "id": 292964,
+            "node_id": "MDQ6VXNlcjI5Mjk2NA==",
+            "avatar_url": "https://avatars.githubusercontent.com/u/292964?v=4",
+            "gravatar_id": "",
+            "url": "https://api.github.com/users/raimohanska",
+            "html_url": "https://github.com/raimohanska",
+            "followers_url": "https://api.github.com/users/raimohanska/followers",
+            "following_url": "https://api.github.com/users/raimohanska/following{/other_user}",
+            "gists_url": "https://api.github.com/users/raimohanska/gists{/gist_id}",
+            "starred_url": "https://api.github.com/users/raimohanska/starred{/owner}{/repo}",
+            "subscriptions_url": "https://api.github.com/users/raimohanska/subscriptions",
+            "organizations_url": "https://api.github.com/users/raimohanska/orgs",
+            "repos_url": "https://api.github.com/users/raimohanska/repos",
+            "events_url": "https://api.github.com/users/raimohanska/events{/privacy}",
+            "received_events_url": "https://api.github.com/users/raimohanska/received_events",
+            "type": "User",
+            "site_admin": false
+        },
+        "labels": [
+            {
+                "id": 2438308116,
+                "node_id": "MDU6TGFiZWwyNDM4MzA4MTE2",
+                "url": "https://api.github.com/repos/raimohanska/r-board/labels/enhancement",
+                "name": "bug",
+                "color": "a2eeef",
+                "default": true,
+                "description": "Bug"
+            }
+        ],
+        "state": "open",
+        "locked": false,
+        "assignee": {
+            "login": "raimohanska",
+            "id": 292964,
+            "node_id": "MDQ6VXNlcjI5Mjk2NA==",
+            "avatar_url": "https://avatars.githubusercontent.com/u/292964?v=4",
+            "gravatar_id": "",
+            "url": "https://api.github.com/users/raimohanska",
+            "html_url": "https://github.com/raimohanska",
+            "followers_url": "https://api.github.com/users/raimohanska/followers",
+            "following_url": "https://api.github.com/users/raimohanska/following{/other_user}",
+            "gists_url": "https://api.github.com/users/raimohanska/gists{/gist_id}",
+            "starred_url": "https://api.github.com/users/raimohanska/starred{/owner}{/repo}",
+            "subscriptions_url": "https://api.github.com/users/raimohanska/subscriptions",
+            "organizations_url": "https://api.github.com/users/raimohanska/orgs",
+            "repos_url": "https://api.github.com/users/raimohanska/repos",
+            "events_url": "https://api.github.com/users/raimohanska/events{/privacy}",
+            "received_events_url": "https://api.github.com/users/raimohanska/received_events",
+            "type": "User",
+            "site_admin": false
+        },
+        "assignees": [
+            {
+                "login": "raimohanska",
+                "id": 292964,
+                "node_id": "MDQ6VXNlcjI5Mjk2NA==",
+                "avatar_url": "https://avatars.githubusercontent.com/u/292964?v=4",
+                "gravatar_id": "",
+                "url": "https://api.github.com/users/raimohanska",
+                "html_url": "https://github.com/raimohanska",
+                "followers_url": "https://api.github.com/users/raimohanska/followers",
+                "following_url": "https://api.github.com/users/raimohanska/following{/other_user}",
+                "gists_url": "https://api.github.com/users/raimohanska/gists{/gist_id}",
+                "starred_url": "https://api.github.com/users/raimohanska/starred{/owner}{/repo}",
+                "subscriptions_url": "https://api.github.com/users/raimohanska/subscriptions",
+                "organizations_url": "https://api.github.com/users/raimohanska/orgs",
+                "repos_url": "https://api.github.com/users/raimohanska/repos",
+                "events_url": "https://api.github.com/users/raimohanska/events{/privacy}",
+                "received_events_url": "https://api.github.com/users/raimohanska/received_events",
+                "type": "User",
+                "site_admin": false
+            }
+        ],
+        "milestone": null,
+        "comments": 0,
+        "created_at": "2021-02-17T18:39:11Z",
+        "updated_at": "2021-02-17T19:08:21Z",
+        "closed_at": null,
+        "author_association": "OWNER",
+        "active_lock_reason": null,
+        "body": "",
+        "performed_via_github_app": null
+    },
+    "assignee": {
+        "login": "raimohanska",
+        "id": 292964,
+        "node_id": "MDQ6VXNlcjI5Mjk2NA==",
+        "avatar_url": "https://avatars.githubusercontent.com/u/292964?v=4",
+        "gravatar_id": "",
+        "url": "https://api.github.com/users/raimohanska",
+        "html_url": "https://github.com/raimohanska",
+        "followers_url": "https://api.github.com/users/raimohanska/followers",
+        "following_url": "https://api.github.com/users/raimohanska/following{/other_user}",
+        "gists_url": "https://api.github.com/users/raimohanska/gists{/gist_id}",
+        "starred_url": "https://api.github.com/users/raimohanska/starred{/owner}{/repo}",
+        "subscriptions_url": "https://api.github.com/users/raimohanska/subscriptions",
+        "organizations_url": "https://api.github.com/users/raimohanska/orgs",
+        "repos_url": "https://api.github.com/users/raimohanska/repos",
+        "events_url": "https://api.github.com/users/raimohanska/events{/privacy}",
+        "received_events_url": "https://api.github.com/users/raimohanska/received_events",
+        "type": "User",
+        "site_admin": false
+    },
+    "repository": {
+        "id": 305431036,
+        "node_id": "MDEwOlJlcG9zaXRvcnkzMDU0MzEwMzY=",
+        "name": "r-board",
+        "full_name": "raimohanska/r-board",
+        "private": false,
+        "owner": {
+            "login": "raimohanska",
+            "id": 292964,
+            "node_id": "MDQ6VXNlcjI5Mjk2NA==",
+            "avatar_url": "https://avatars.githubusercontent.com/u/292964?v=4",
+            "gravatar_id": "",
+            "url": "https://api.github.com/users/raimohanska",
+            "html_url": "https://github.com/raimohanska",
+            "followers_url": "https://api.github.com/users/raimohanska/followers",
+            "following_url": "https://api.github.com/users/raimohanska/following{/other_user}",
+            "gists_url": "https://api.github.com/users/raimohanska/gists{/gist_id}",
+            "starred_url": "https://api.github.com/users/raimohanska/starred{/owner}{/repo}",
+            "subscriptions_url": "https://api.github.com/users/raimohanska/subscriptions",
+            "organizations_url": "https://api.github.com/users/raimohanska/orgs",
+            "repos_url": "https://api.github.com/users/raimohanska/repos",
+            "events_url": "https://api.github.com/users/raimohanska/events{/privacy}",
+            "received_events_url": "https://api.github.com/users/raimohanska/received_events",
+            "type": "User",
+            "site_admin": false
+        },
+        "html_url": "https://github.com/raimohanska/r-board",
+        "description": "An online whiteboard",
+        "fork": false,
+        "url": "https://api.github.com/repos/raimohanska/r-board",
+        "forks_url": "https://api.github.com/repos/raimohanska/r-board/forks",
+        "keys_url": "https://api.github.com/repos/raimohanska/r-board/keys{/key_id}",
+        "collaborators_url": "https://api.github.com/repos/raimohanska/r-board/collaborators{/collaborator}",
+        "teams_url": "https://api.github.com/repos/raimohanska/r-board/teams",
+        "hooks_url": "https://api.github.com/repos/raimohanska/r-board/hooks",
+        "issue_events_url": "https://api.github.com/repos/raimohanska/r-board/issues/events{/number}",
+        "events_url": "https://api.github.com/repos/raimohanska/r-board/events",
+        "assignees_url": "https://api.github.com/repos/raimohanska/r-board/assignees{/user}",
+        "branches_url": "https://api.github.com/repos/raimohanska/r-board/branches{/branch}",
+        "tags_url": "https://api.github.com/repos/raimohanska/r-board/tags",
+        "blobs_url": "https://api.github.com/repos/raimohanska/r-board/git/blobs{/sha}",
+        "git_tags_url": "https://api.github.com/repos/raimohanska/r-board/git/tags{/sha}",
+        "git_refs_url": "https://api.github.com/repos/raimohanska/r-board/git/refs{/sha}",
+        "trees_url": "https://api.github.com/repos/raimohanska/r-board/git/trees{/sha}",
+        "statuses_url": "https://api.github.com/repos/raimohanska/r-board/statuses/{sha}",
+        "languages_url": "https://api.github.com/repos/raimohanska/r-board/languages",
+        "stargazers_url": "https://api.github.com/repos/raimohanska/r-board/stargazers",
+        "contributors_url": "https://api.github.com/repos/raimohanska/r-board/contributors",
+        "subscribers_url": "https://api.github.com/repos/raimohanska/r-board/subscribers",
+        "subscription_url": "https://api.github.com/repos/raimohanska/r-board/subscription",
+        "commits_url": "https://api.github.com/repos/raimohanska/r-board/commits{/sha}",
+        "git_commits_url": "https://api.github.com/repos/raimohanska/r-board/git/commits{/sha}",
+        "comments_url": "https://api.github.com/repos/raimohanska/r-board/comments{/number}",
+        "issue_comment_url": "https://api.github.com/repos/raimohanska/r-board/issues/comments{/number}",
+        "contents_url": "https://api.github.com/repos/raimohanska/r-board/contents/{+path}",
+        "compare_url": "https://api.github.com/repos/raimohanska/r-board/compare/{base}...{head}",
+        "merges_url": "https://api.github.com/repos/raimohanska/r-board/merges",
+        "archive_url": "https://api.github.com/repos/raimohanska/r-board/{archive_format}{/ref}",
+        "downloads_url": "https://api.github.com/repos/raimohanska/r-board/downloads",
+        "issues_url": "https://api.github.com/repos/raimohanska/r-board/issues{/number}",
+        "pulls_url": "https://api.github.com/repos/raimohanska/r-board/pulls{/number}",
+        "milestones_url": "https://api.github.com/repos/raimohanska/r-board/milestones{/number}",
+        "notifications_url": "https://api.github.com/repos/raimohanska/r-board/notifications{?since,all,participating}",
+        "labels_url": "https://api.github.com/repos/raimohanska/r-board/labels{/name}",
+        "releases_url": "https://api.github.com/repos/raimohanska/r-board/releases{/id}",
+        "deployments_url": "https://api.github.com/repos/raimohanska/r-board/deployments",
+        "created_at": "2020-10-19T15:33:11Z",
+        "updated_at": "2021-02-17T18:39:38Z",
+        "pushed_at": "2021-02-17T18:31:01Z",
+        "git_url": "git://github.com/raimohanska/r-board.git",
+        "ssh_url": "git@github.com:raimohanska/r-board.git",
+        "clone_url": "https://github.com/raimohanska/r-board.git",
+        "svn_url": "https://github.com/raimohanska/r-board",
+        "homepage": null,
+        "size": 1137,
+        "stargazers_count": 4,
+        "watchers_count": 4,
+        "language": "TypeScript",
+        "has_issues": true,
+        "has_projects": true,
+        "has_downloads": true,
+        "has_wiki": true,
+        "has_pages": false,
+        "forks_count": 2,
+        "mirror_url": null,
+        "archived": false,
+        "disabled": false,
+        "open_issues_count": 33,
+        "license": {
+            "key": "other",
+            "name": "Other",
+            "spdx_id": "NOASSERTION",
+            "url": null,
+            "node_id": "MDc6TGljZW5zZTA="
+        },
+        "forks": 2,
+        "open_issues": 33,
+        "watchers": 4,
+        "default_branch": "master"
+    },
+    "sender": {
+        "login": "raimohanska",
+        "id": 292964,
+        "node_id": "MDQ6VXNlcjI5Mjk2NA==",
+        "avatar_url": "https://avatars.githubusercontent.com/u/292964?v=4",
+        "gravatar_id": "",
+        "url": "https://api.github.com/users/raimohanska",
+        "html_url": "https://github.com/raimohanska",
+        "followers_url": "https://api.github.com/users/raimohanska/followers",
+        "following_url": "https://api.github.com/users/raimohanska/following{/other_user}",
+        "gists_url": "https://api.github.com/users/raimohanska/gists{/gist_id}",
+        "starred_url": "https://api.github.com/users/raimohanska/starred{/owner}{/repo}",
+        "subscriptions_url": "https://api.github.com/users/raimohanska/subscriptions",
+        "organizations_url": "https://api.github.com/users/raimohanska/orgs",
+        "repos_url": "https://api.github.com/users/raimohanska/repos",
+        "events_url": "https://api.github.com/users/raimohanska/events{/privacy}",
+        "received_events_url": "https://api.github.com/users/raimohanska/received_events",
+        "type": "User",
+        "site_admin": false
+    }
+}

backend/src/google-auth.ts

@@ -0,0 +1,70 @@
+import { google } from "googleapis"
+import { OAuthAuthenticatedUser } from "../../common/src/authenticated-user"
+import { assertNotNull } from "../../common/src/assertNotNull"
+import { getEnv } from "./env"
+import { AuthProvider } from "./oauth"
+import { ROOT_URL } from "./host-config"
+import { decodeOrThrow } from "./decodeOrThrow"
+import * as T from "io-ts"
+import { optional } from "../../common/src/domain"
+import JWT from "jsonwebtoken"
+
+type GoogleConfig = {
+    clientID: string
+    clientSecret: string
+    callbackURL: string
+}
+
+export const googleConfig: GoogleConfig | null = process.env.GOOGLE_OAUTH_CLIENT_ID
+    ? {
+          clientID: getEnv("GOOGLE_OAUTH_CLIENT_ID"),
+          clientSecret: getEnv("GOOGLE_OAUTH_CLIENT_SECRET"),
+          callbackURL: `${ROOT_URL}/google-callback`,
+      }
+    : null
+
+export const GoogleAuthProvider = (googleConfig: GoogleConfig): AuthProvider => {
+    console.log(`Setting up Google authentication using client ID ${googleConfig.clientID}`)
+
+    const googleScopes = ["email", "https://www.googleapis.com/auth/userinfo.profile"]
+
+    function googleOAUTH2() {
+        if (!googleConfig.clientID || !googleConfig.clientSecret)
+            throw new Error("Missing environment variables for Google OAuth")
+        return new google.auth.OAuth2(googleConfig.clientID, googleConfig.clientSecret, googleConfig.callbackURL)
+    }
+
+    async function getAuthPageURL() {
+        return googleOAUTH2().generateAuthUrl({
+            scope: googleScopes,
+            prompt: "select_account",
+        })
+    }
+
+    const IdToken = T.strict({
+        hd: optional(T.string),
+        email: T.string,
+        email_verified: T.boolean,
+        name: T.string,
+        picture: optional(T.string),
+    })
+
+    async function getAccountFromCode(code: string): Promise<OAuthAuthenticatedUser> {
+        const auth = googleOAUTH2()
+        const data = await auth.getToken(code)
+        const idToken = decodeOrThrow(IdToken, JWT.decode(assertNotNull(data.tokens.id_token)))
+        const email = idToken.email
+
+        return {
+            name: idToken.name,
+            email,
+            picture: idToken.picture ?? undefined,
+            domain: idToken.hd ?? null,
+        }
+    }
+
+    return {
+        getAuthPageURL,
+        getAccountFromCode,
+    }
+}

backend/src/host-config.ts

@@ -0,0 +1,6 @@
+export const ROOT_URL = process.env.ROOT_URL ?? "http://localhost:1337"
+export const ROOT_HOST = new URL(ROOT_URL).host
+export const ROOT_PROTOCOL = new URL(ROOT_URL).protocol
+export const WS_HOST_LOCAL = (process.env.WS_HOST_LOCAL ?? ROOT_HOST).split(",")
+export const WS_HOST_DEFAULT = process.env.WS_HOST_DEFAULT ?? ROOT_HOST
+export const WS_PROTOCOL = process.env.WS_PROTOCOL ?? (ROOT_PROTOCOL.startsWith("https") ? "wss" : "ws")

backend/src/http-session.ts

@@ -0,0 +1,53 @@
+import Cookies from "cookies"
+import { IncomingMessage, ServerResponse } from "http"
+import JWT from "jsonwebtoken"
+import { getEnv } from "./env"
+import { OAuthAuthenticatedUser } from "../../common/src/authenticated-user"
+import { ISOTimeStamp, newISOTimeStamp } from "../../common/src/domain"
+
+const secret = getEnv("SESSION_SIGNING_SECRET")
+
+export type LoginInfo = OAuthAuthenticatedUser & {
+    timestamp: ISOTimeStamp | undefined
+}
+
+export function getSessionIdFromCookies(req: IncomingMessage): string | null {
+    return new Cookies(req, null as any).get("sessionId") ?? null
+}
+
+// Get / set authenticated user stored in cookies
+export function getAuthenticatedUser(req: IncomingMessage): LoginInfo | null {
+    const userCookie = new Cookies(req, null as any).get("user")
+    if (userCookie) {
+        return getAuthenticatedUserFromJWT(userCookie)
+    }
+    return null
+}
+
+export function getAuthenticatedUserFromJWT(jwt: string): LoginInfo | null {
+    try {
+        JWT.verify(jwt, secret)
+        const loginInfo = JWT.decode(jwt) as LoginInfo
+        if (loginInfo.domain === undefined) {
+            console.log("Rejecting legacy token without domain")
+            return null
+        }
+        return loginInfo
+    } catch (e) {
+        console.warn("Token verification failed", jwt, e)
+    }
+    return null
+}
+
+export function setAuthenticatedUser(req: IncomingMessage, res: ServerResponse, userInfo: OAuthAuthenticatedUser) {
+    const loginInfo: LoginInfo = { ...userInfo, timestamp: newISOTimeStamp() }
+    const jwt = JWT.sign(loginInfo, secret)
+    new Cookies(req, res).set("user", jwt, {
+        maxAge: 365 * 24 * 3600 * 1000,
+        httpOnly: false,
+    }) // Max 365 days expiration
+}
+
+export function removeAuthenticatedUser(req: IncomingMessage, res: ServerResponse) {
+    new Cookies(req, res).set("user", "", { maxAge: 0, httpOnly: true })
+}

backend/src/locker.ts

@@ -0,0 +1,58 @@
+import { Id, BoardItemEvent, isPersistableBoardItemEvent, getItemIds, ItemLocks } from "../../common/src/domain"
+import { getActiveBoards, ServerSideBoardState } from "./board-state"
+import { AutoExpiringMap } from "./expiring-map"
+import { WsWrapper } from "./ws-wrapper"
+
+const LOCK_TTL_SECONDS = 10
+
+export function Locks(onChange: (locks: ItemLocks) => any) {
+    const locks = AutoExpiringMap<string>(LOCK_TTL_SECONDS).onChange(onChange)
+
+    return {
+        lockItem: (itemId: Id, sessionId: Id) => {
+            if (locks.has(itemId) && locks.get(itemId) !== sessionId) {
+                return false
+            }
+
+            locks.set(itemId, sessionId)
+            return true
+        },
+        unlockItem: (itemId: Id, sessionId: Id) => {
+            if (locks.get(itemId) === sessionId) {
+                return locks.delete(itemId)
+            }
+
+            return false
+        },
+        delete: (itemId: string) => locks.delete(itemId),
+        entries: () => locks.entries(),
+    }
+}
+
+export function obtainLock(locks: ServerSideBoardState["locks"], e: BoardItemEvent, socket: WsWrapper) {
+    if (isPersistableBoardItemEvent(e)) {
+        const itemIds = getItemIds(e)
+        // Since we are operating on multiple items at a time, locking must succeed for all of them
+        // for the action to succeed
+        return itemIds.every((id) => locks.lockItem(id, socket.id))
+    } else {
+        const { itemId, action } = e
+        switch (action) {
+            case "item.lock":
+                return locks.lockItem(itemId, socket.id)
+            case "item.unlock":
+                return locks.unlockItem(itemId, socket.id)
+        }
+    }
+}
+
+export function releaseLocksFor(socket: WsWrapper) {
+    getActiveBoards().forEach((state) => {
+        const locks = state.locks
+        for (const [itemId, sessionId] of locks.entries()) {
+            if (socket.id === sessionId) {
+                locks.delete(itemId)
+            }
+        }
+    })
+}

backend/src/oauth.ts

@@ -0,0 +1,60 @@
+import Cookies from "cookies"
+import { Express, Request, Response } from "express"
+import { OAuthAuthenticatedUser } from "../../common/src/authenticated-user"
+import { removeAuthenticatedUser, setAuthenticatedUser } from "./http-session"
+import { GoogleAuthProvider, googleConfig } from "./google-auth"
+import { GenericOIDCAuthProvider, genericOIDCConfig } from "./generic-oidc-auth"
+
+export interface AuthProvider {
+    getAuthPageURL: () => Promise<string>
+    getAccountFromCode: (code: string) => Promise<OAuthAuthenticatedUser>
+    logout?: (req: Request, res: Response) => Promise<void>
+}
+
+export function setupAuth(app: Express, provider: AuthProvider) {
+    app.get("/login", async (req, res) => {
+        new Cookies(req, res).set("returnTo", parseReturnPath(req), {
+            maxAge: 24 * 3600 * 1000,
+            httpOnly: true,
+        }) // Max 24 hours
+        const authUrl = await provider.getAuthPageURL()
+        res.setHeader("content-type", "text/html")
+        res.send(`Signing in...<script>document.location='${authUrl}'</script>`)
+    })
+
+    app.get("/logout", async (req, res) => {
+        removeAuthenticatedUser(req, res)
+        if (provider.logout) {
+            await provider.logout(req, res)
+        } else {
+            res.redirect(parseReturnPath(req))
+        }
+    })
+
+    app.get("/google-callback", async (req, res) => {
+        const code = (req.query?.code as string) || ""
+        const cookies = new Cookies(req, res)
+        const returnTo = cookies.get("returnTo") || "/"
+        cookies.set("returnTo", "", { maxAge: 0, httpOnly: true })
+        console.log("Verifying google auth", code)
+        try {
+            const userInfo = await provider.getAccountFromCode(code)
+            console.log("Found", userInfo)
+            setAuthenticatedUser(req, res, userInfo)
+            res.redirect(returnTo)
+        } catch (e) {
+            console.error(e)
+            res.status(500).send("Internal error")
+        }
+    })
+
+    function parseReturnPath(req: Request) {
+        return (req.query.returnTo as string) || "/"
+    }
+}
+
+export const authProvider: AuthProvider | null = googleConfig
+    ? GoogleAuthProvider(googleConfig)
+    : genericOIDCConfig
+    ? GenericOIDCAuthProvider(genericOIDCConfig)
+    : null

backend/src/professions.ts

@@ -0,0 +1,72 @@
+export const professions = [
+    "Accountant",
+    "Actress",
+    "Architect",
+    "Astronomer",
+    "Author",
+    "Baker",
+    "Bricklayer",
+    "Bus driver",
+    "Butcher",
+    "Carpenter",
+    "Chef",
+    "Cleaner",
+    "Coder",
+    "Consultant",
+    "Dentist",
+    "Designer",
+    "DevOps specialist",
+    "Doctor",
+    "Dustman",
+    "Electrician",
+    "Engineer",
+    "Factory worker",
+    "Farmer",
+    "Fire fighter",
+    "Fisherman",
+    "Florist",
+    "Gardener",
+    "Hairdresser",
+    "Journalist",
+    "Judge",
+    "Lawyer",
+    "Lecturer",
+    "Librarian",
+    "Lifeguard",
+    "Mechanic",
+    "Model",
+    "Newsreader",
+    "Nurse",
+    "Optician",
+    "Painter",
+    "Pharmacist",
+    "Photographer",
+    "Physician",
+    "Pilot",
+    "Plumber",
+    "Pointy-haired boss",
+    "Politician",
+    "Postman",
+    "Product Owner",
+    "Real estate agent",
+    "Receptionist",
+    "Scientist",
+    "Scrum master",
+    "Secretary",
+    "Shop assistant",
+    "Tailor",
+    "Taxi driver",
+    "Teacher",
+    "Tester",
+    "Translator",
+    "Traffic warden",
+    "Travel agent",
+    "Vet",
+    "UNIX guru",
+    "Waiter/Waitress",
+    "Window cleaner",
+]
+
+export function randomProfession() {
+    return professions[Math.floor(Math.random() * professions.length)]
+}

backend/src/require-auth.ts

@@ -0,0 +1,17 @@
+import { Express, Request, Response, NextFunction } from "express"
+import { getAuthenticatedUser } from "./http-session"
+
+export const REQUIRE_AUTH = process.env.REQUIRE_AUTH === "true"
+
+export function possiblyRequireAuth(app: Express) {
+    if (REQUIRE_AUTH) {
+        // Require authentication for all resources except the URLs bound by setupAuth above
+        app.use("/", (req: Request, res: Response, next: NextFunction) => {
+            if (!getAuthenticatedUser(req)) {
+                res.redirect("/login")
+            } else {
+                next()
+            }
+        })
+    }
+}

backend/src/s3.ts

@@ -0,0 +1,26 @@
+import * as AWS from "aws-sdk"
+
+const s3Config = {
+    region: "eu-north-1",
+    apiVersion: "2006-03-01",
+    signatureVersion: "v4",
+}
+
+let s3Instance: AWS.S3 | null = null
+
+export const s3 = () => {
+    s3Instance = s3Instance || new AWS.S3(s3Config)
+    return s3Instance
+}
+
+export function getSignedPutUrl(Key: string) {
+    const signedUrlExpireSeconds = 60 * 5
+
+    const url = s3().getSignedUrl("putObject", {
+        Bucket: "r-board-assets",
+        Key,
+        Expires: signedUrlExpireSeconds,
+    })
+
+    return url
+}

backend/src/server.ts

@@ -0,0 +1,60 @@
+import dotenv from "dotenv"
+dotenv.config()
+
+import * as Http from "http"
+import { exampleBoard } from "../../common/src/domain"
+import { awaitSavingChanges } from "./board-state"
+import { createBoard, fetchBoard } from "./board-store"
+import { initDB } from "./db"
+import { startExpressServer } from "./express-server"
+import { terminateSessions } from "./websocket-sessions"
+
+let stopServer: (() => void) | null = null
+
+async function shutdown() {
+    console.log("Shutdown initiated. Closing sockets.")
+    if (stopServer) stopServer()
+    terminateSessions()
+    console.log("Shutdown in progress. Waiting for all changes to be saved...")
+    await awaitSavingChanges()
+    console.log("Shutdown complete. Exiting process.")
+    process.exit(0)
+}
+
+process.on("SIGTERM", () => {
+    console.log("Received SIGTERM. Initiating shutdown.")
+    shutdown()
+})
+
+const PORT = parseInt(process.env.PORT || "1337")
+const HTTPS_PORT = process.env.HTTPS_PORT ? parseInt(process.env.HTTPS_PORT) : undefined
+const BIND_UWEBSOCKETS_TO_PORT = process.env.BIND_UWEBSOCKETS_TO_PORT === "true"
+if (BIND_UWEBSOCKETS_TO_PORT && process.env.UWEBSOCKETS_PORT) {
+    throw Error("Cannot have both UWEBSOCKETS_PORT and BIND_UWEBSOCKETS_TO_PORT envs")
+}
+const HTTP_PORT = BIND_UWEBSOCKETS_TO_PORT ? null : PORT
+const UWEBSOCKETS_PORT = BIND_UWEBSOCKETS_TO_PORT
+    ? PORT
+    : process.env.UWEBSOCKETS_PORT
+    ? parseInt(process.env.UWEBSOCKETS_PORT)
+    : null
+
+initDB()
+    .then(async () => {
+        if (!(await fetchBoard("default"))) {
+            await createBoard(exampleBoard)
+        }
+    })
+    .then(() => {
+        if (HTTP_PORT) {
+            stopServer = startExpressServer(HTTP_PORT, HTTPS_PORT)
+        }
+        if (UWEBSOCKETS_PORT) {
+            import("./uwebsockets-server").then((uwebsockets) => {
+                uwebsockets.startUWebSocketsServer(UWEBSOCKETS_PORT)
+            })
+        }
+    })
+    .catch((e) => {
+        console.error(e)
+    })

backend/src/storage.ts

@@ -0,0 +1,9 @@
+import { getSignedPutUrl as s3GetSignedPutUrl } from "./s3"
+import { StorageBackend } from "./config"
+
+function localFSGetSignedPutUrl(Key: string): string {
+    return "/assets/" + Key
+}
+
+export const createGetSignedPutUrl = (storageBackend: StorageBackend): ((key: string) => string) =>
+    storageBackend.type === "AWS" ? s3GetSignedPutUrl : localFSGetSignedPutUrl

backend/src/tools/wait-for-db.ts

@@ -0,0 +1,11 @@
+import TcpPortUsed from "tcp-port-used"
+const port = 13338
+;(async function () {
+    console.log(`Waiting for DB to bind port ${port}...`)
+    try {
+        await TcpPortUsed.waitUntilUsed(port, 100, 10000)
+    } catch {
+        console.error("Timed out waiting for DB")
+        process.exit(1)
+    }
+})()

backend/src/user-store.ts

@@ -0,0 +1,62 @@
+import { inTransaction, withDBClient } from "./db"
+import * as uuid from "uuid"
+import { EventUserInfo, Id, RecentBoard, EventUserInfoAuthenticated, ISOTimeStamp } from "../../common/src/domain"
+import { uniqBy } from "lodash"
+
+export function getUserIdForEmail(email: string): Promise<string> {
+    return inTransaction(async (client) => {
+        let id: string | undefined = (await client.query("SELECT id FROM app_user WHERE email=$1", [email])).rows[0]?.id
+        if (!id) {
+            id = uuid.v4()
+            await client.query("INSERT INTO app_user (id, email) VALUES ($1, $2);", [id, email])
+        }
+        return id
+    })
+}
+
+export async function associateUserWithBoard(
+    userId: string,
+    boardId: Id,
+    lastOpened: ISOTimeStamp = new Date().toISOString(),
+) {
+    try {
+        await inTransaction(async (client) => {
+            await client.query(
+                `INSERT INTO user_board (user_id, board_id, last_opened) values ($1, $2, $3) 
+                 ON CONFLICT (user_id, board_id) DO UPDATE SET last_opened=EXCLUDED.last_opened`,
+                [userId, boardId, lastOpened],
+            )
+        })
+    } catch (e) {
+        console.error(`Failed to associate user ${userId} with board ${boardId}`)
+    }
+}
+
+export async function dissociateUserWithBoard(userId: string, boardId: Id) {
+    try {
+        await inTransaction(async (client) => {
+            await client.query(`DELETE FROM user_board WHERE user_id=$1 and board_id=$2`, [userId, boardId])
+        })
+    } catch (e) {
+        console.error(`Failed to dissociate user ${userId} with board ${boardId}`)
+    }
+}
+
+export async function getUserAssociatedBoards(user: EventUserInfoAuthenticated): Promise<RecentBoard[]> {
+    const rows = (
+        await withDBClient((client) =>
+            client.query(
+                "SELECT b.id, b.name, ub.last_opened FROM user_board ub JOIN board b on (ub.board_id = b.id) WHERE ub.user_id = $1",
+                [user.userId],
+            ),
+        )
+    ).rows
+    return rows.map((r) => {
+        return {
+            id: r.id,
+            name: r.name,
+            userEmail: user.email,
+            opened: r.last_opened.toISOString(),
+        }
+    })
+}

backend/src/uwebsockets-server.ts

@@ -0,0 +1,105 @@
+import uws from "uWebSockets.js"
+import { EventFromServer } from "../../common/src/domain"
+import * as uuid from "uuid"
+import { connectionHandler, MessageHandler } from "./connection-handler"
+import { handleBoardEvent } from "./board-event-handler"
+import { createGetSignedPutUrl } from "./storage"
+import { getConfig } from "./config"
+import * as L from "lonna"
+import { handleCommonEvent } from "./common-event-handler"
+
+export const WsWrapper = (ws: uws.WebSocket) => {
+    const errorE = L.bus<void>()
+    const closeE = L.bus<void>()
+    const msgE = L.bus<object>()
+
+    const onError = (f: () => void) => {
+        errorE.forEach(f)
+    }
+    const onMessage = (f: (msg: object) => void) => {
+        msgE.forEach(f)
+    }
+    const onClose = (f: () => void) => {
+        closeE.forEach(f)
+    }
+    return {
+        send: (buffer: Buffer) => {
+            try {
+                ws.send(buffer, false)
+            } catch (e) {
+                ws.close()
+            }
+        },
+        onError,
+        onMessage,
+        onClose,
+        id: uuid.v4(),
+        close: () => {
+            ws.close()
+            closeE.push()
+        },
+        errorE,
+        closeE,
+        msgE,
+    }
+}
+type WsWrapper = ReturnType<typeof WsWrapper>
+
+type WsUserData = {
+    handler: MessageHandler
+}
+
+export function startUWebSocketsServer(port: number) {
+    const config = getConfig()
+    const app = uws.App()
+
+    const sockets = new Map<uws.WebSocket, WsWrapper>()
+    const textDecoder = new TextDecoder()
+
+    const signedPutUrl = createGetSignedPutUrl(config.storageBackend)
+    mountWs("/socket/lobby", () => handleBoardEvent(null, signedPutUrl))
+    mountWs("/socket/board/:boardId", (req) => handleBoardEvent(req.getParameter(0), signedPutUrl))
+
+    app.get("/", (res) => res.writeStatus("200 OK").end("Sorry, we only serve websocket clients here."))
+
+    function mountWs(path: string, f: (req: uws.HttpRequest) => MessageHandler) {
+        app.ws(path, {
+            upgrade: (res, req, context) => {
+                res.upgrade(
+                    {
+                        handler: f(req),
+                    } as WsUserData,
+                    req.getHeader("sec-websocket-key"),
+                    req.getHeader("sec-websocket-protocol"),
+                    req.getHeader("sec-websocket-extensions"),
+                    // 3 headers are used to setup websocket
+                    context,
+                )
+            },
+            open: (ws) => {
+                const wrapper = WsWrapper(ws)
+                const handler = (ws as WsUserData & uws.WebSocket).handler
+                sockets.set(ws, wrapper)
+                connectionHandler(wrapper, handler)
+            },
+            message: (ws, message, isBinary) => {
+                if (isBinary) throw Error("Binary message")
+                const object = JSON.parse(textDecoder.decode(message))
+                const wrapper = sockets.get(ws)
+                if (!wrapper) {
+                    throw Error("Wrapper not found for socket " + ws)
+                }
+                wrapper.msgE.push(object)
+            },
+            close: (ws) => {
+                const wrapper = sockets.get(ws)
+                if (wrapper) {
+                    wrapper.closeE.push()
+                }
+            },
+        })
+    }
+    app.listen(port, () => {
+        console.log("uWebSockets listening on " + port)
+    })
+}

backend/src/websocket-sessions.ts

@@ -0,0 +1,347 @@
+import { OAuthAuthenticatedUser } from "../../common/src/authenticated-user"
+import {
+    AccessLevel,
+    AckJoinBoard,
+    AuthLogout,
+    BoardHistoryEntry,
+    CURSOR_POSITIONS_ACTION_TYPE,
+    EventFromServer,
+    EventUserInfoAuthenticated,
+    Id,
+    ItemLocks,
+    JoinedBoard,
+    Serial,
+    SessionUserInfo,
+    SetNickname,
+    UnidentifiedUserInfo,
+    UserCursorPosition,
+    UserInfoUpdate,
+    getBoardAttributes,
+    isBoardHistoryEntry,
+} from "../../common/src/domain"
+import { ServerSideBoardState, maybeGetBoard } from "./board-state"
+import { getBoardHistory } from "./board-store"
+import { closeYjsSocketsBySessionId } from "./board-yjs-server"
+import { randomProfession } from "./professions"
+import { getUserIdForEmail } from "./user-store"
+import { WsWrapper, toBuffer } from "./ws-wrapper"
+
+export type UserSession = {
+    readonly sessionId: Id
+    boardSession: UserSessionBoardEntry | null
+    userInfo: SessionUserInfo
+    sendEvent: (event: EventFromServer) => void
+    isOnBoard: (boardId: Id) => boolean
+    close(): void
+}
+
+export type UserSessionBoardEntry = {
+    boardId: Id
+    status: "ready" | "buffering"
+    accessLevel: AccessLevel
+    bufferedEvents: BoardHistoryEntry[]
+}
+
+/*
+socket: WsWrapper
+    boards: Id[]
+    userInfo: EventUserInfo
+    */
+export type SocketId = string
+
+const sessions: Record<SocketId, UserSession> = {}
+
+const everyoneOnTheBoard = (boardId: string) => {
+    const boardState = maybeGetBoard(boardId)
+    if (!boardState) {
+        console.warn(`Trying to send to a board not in memory: ${boardId}`)
+        return []
+    }
+    return boardState.sessions
+}
+const sendTo = (recipients: UserSession[], message: EventFromServer) => {
+    recipients.forEach((c) => c.sendEvent(message))
+}
+const everyoneElseOnTheSameBoard = (boardId: Id, session?: UserSession) =>
+    everyoneOnTheBoard(boardId).filter((s) => s !== session)
+
+export function startSession(socket: WsWrapper) {
+    sessions[socket.id] = userSession(socket)
+}
+
+function userSession(socket: WsWrapper): UserSession {
+    const sessionId = socket.id
+
+    function sendEvent(event: EventFromServer) {
+        if (isBoardHistoryEntry(event)) {
+            const entry = session.boardSession
+            if (!entry) throw Error("Board " + event.boardId + " not found for session " + sessionId)
+            if (entry.status === "buffering") {
+                entry.bufferedEvents.push(event)
+                return
+            }
+        }
+        socket.send(toBuffer(event))
+    }
+    const session: UserSession = {
+        sessionId,
+        userInfo: anonymousUser("Anonymous " + randomProfession()),
+        boardSession: null,
+        sendEvent,
+        isOnBoard: (boardId: Id) => session.boardSession != null && session.boardSession.boardId === boardId,
+        close: () => socket.close(),
+    }
+    sessions[socket.id] = session
+    return session
+}
+
+function anonymousUser(nickname: string): UnidentifiedUserInfo {
+    return { userType: "unidentified", nickname }
+}
+
+export function endSession(socket: WsWrapper) {
+    const sessionId = socket.id
+    const session = sessions[sessionId]
+    if (!session) {
+        console.warn(`Ending non-existing session ${sessionId}`)
+        return
+    }
+    if (session.boardSession) {
+        const boardState = maybeGetBoard(session.boardSession.boardId)
+        if (boardState) {
+            boardState.sessions = boardState.sessions.filter((s) => s.sessionId !== sessionId)
+            broadcastBoardEvent({ action: "board.left", boardId: boardState.board.id, sessionId })
+        } else {
+            console.warn(`Board state not found when ending session: ${session.boardSession.boardId}`)
+        }
+    }
+    delete sessions[socket.id]
+    closeYjsSocketsBySessionId(sessionId)
+}
+export function getBoardSessionCount(id: Id) {
+    return everyoneOnTheBoard(id).length
+}
+export function getSession(socket: WsWrapper): UserSession | undefined {
+    return getSessionById(socket.id)
+}
+
+export function getSessionById(sessionId: string): UserSession | undefined {
+    return sessions[sessionId]
+}
+
+export function terminateSessions() {
+    Object.values(sessions).forEach((session) => session.close())
+}
+
+export async function addSessionToBoard(
+    boardState: ServerSideBoardState,
+    origin: WsWrapper,
+    accessLevel: AccessLevel,
+    initAtSerial?: Serial,
+): Promise<void> {
+    const session = sessions[origin.id]
+    if (!session) throw new Error("No session found for socket " + origin.id)
+    const boardId = boardState.board.id
+    if (!boardState.sessions.includes(session)) {
+        boardState.sessions = [...boardState.sessions, session]
+    }
+    const initDiff = initAtSerial && boardState.board.serial - initAtSerial
+    if (initDiff && initDiff > Object.keys(boardState.board.items).length) {
+        console.log(`Sending fresh board state for board ${boardId} instead of diff (${initDiff} events to sync)`)
+        initAsNew(session, boardId, accessLevel, boardState)
+    } else if (initAtSerial) {
+        const entry: UserSessionBoardEntry = { boardId, status: "buffering", accessLevel, bufferedEvents: [] }
+        // 1. Add session to the board with "buffering" status, to collect all events that were meant to be sent during this async initialization
+        session.boardSession = entry
+        try {
+            const boardAttributes = getBoardAttributes(boardState.board, session.userInfo)
+
+            //console.log(`Starting session at ${initAtSerial}`)
+            // 2. capture all board events that haven't yet been flushed to the DB
+            const inMemoryEvents = (boardState.currentlyStoring?.events ?? [])
+                .concat(boardState.recentEvents)
+                .filter((e) => e.serial! > initAtSerial)
+
+            // 3. Fetch events from DB as chunks
+            // IMPORTANT NOTE: this is the only await here and must remain so, as the logic here depends on everything else being synchronous.
+            console.log(`Loading board history for board ${boardState.board.id} session at serial ${initAtSerial}`)
+            let first = true
+            await getBoardHistory(boardState.board.id, initAtSerial, (chunk) => {
+                // Send a chunk of events with done: false, so that client knows to wait for more
+                session.sendEvent({
+                    action: "board.init.diff",
+                    first,
+                    last: false,
+                    boardAttributes,
+                    recentEvents: chunk,
+                    initAtSerial,
+                    accessLevel,
+                })
+                first = false
+            })
+
+            console.log(`Got board history for board ${boardState.board.id} session at serial ${initAtSerial}`)
+
+            // 4. Send the last chunk containing both the inMemoryEvents and the buffered events (done: true)
+            // In memory events: not yet flushed to DB when query was made
+            // Buffered events: events that occurred after the in memory events were captured
+            session.sendEvent({
+                action: "board.init.diff",
+                boardAttributes,
+                first,
+                last: true,
+                recentEvents: [...inMemoryEvents, ...entry.bufferedEvents],
+                initAtSerial,
+                accessLevel,
+            })
+
+            // 5. Set the client to "ready" status so that new events will be flushed
+            entry.status = "ready"
+            entry.bufferedEvents = []
+        } catch (e) {
+            console.warn(
+                `Failed to bootstrap client on board ${boardId} at serial ${initAtSerial}. Sending full state.`,
+            )
+            entry.status = "ready"
+            entry.bufferedEvents = []
+            session.sendEvent({
+                action: "board.init",
+                board: boardState.board,
+                accessLevel,
+            })
+        }
+    } else {
+        initAsNew(session, boardId, accessLevel, boardState)
+    }
+
+    // TODO SECURITY: don't reveal authenticated emails to unidentified users on same board
+    // TODO: what to include in joined events? Not just nickname, as we want to show who's identified (beside the cursor)
+
+    session.sendEvent({
+        action: "board.join.ack",
+        boardId: boardState.board.id,
+        sessionId: session.sessionId,
+        nickname: session.userInfo.nickname,
+    } as AckJoinBoard)
+
+    // Notify new user of existing users
+    everyoneOnTheBoard(boardState.board.id).forEach((s) => {
+        session.sendEvent({
+            action: "board.joined",
+            boardId: boardState.board.id,
+            sessionId: s.sessionId,
+            ...s.userInfo,
+        } as JoinedBoard)
+    })
+
+    // Notify existing users of new user
+    broadcastJoinEvent(boardState.board.id, session)
+}
+
+function initAsNew(session: UserSession, boardId: string, accessLevel: AccessLevel, boardState: ServerSideBoardState) {
+    session.boardSession = { boardId, status: "ready", accessLevel, bufferedEvents: [] }
+    session.sendEvent({
+        action: "board.init",
+        board: boardState.board,
+        accessLevel,
+    })
+}
+
+export function setNicknameForSession(event: SetNickname, origin: WsWrapper) {
+    const session = getSession(origin)
+    if (!session) {
+        console.warn(`Session not found: ${origin.id}`)
+        return
+    }
+
+    session.userInfo =
+        session.userInfo.userType === "unidentified"
+            ? anonymousUser(event.nickname)
+            : { ...session.userInfo, nickname: event.nickname }
+    const updateInfo: UserInfoUpdate = {
+        action: "userinfo.set",
+        sessionId: session.sessionId,
+        ...session.userInfo,
+    }
+    if (session.boardSession) {
+        sendTo(everyoneOnTheBoard(session.boardSession.boardId), updateInfo)
+    }
+}
+
+export async function setVerifiedUserForSession(
+    event: OAuthAuthenticatedUser,
+    session: UserSession,
+): Promise<EventUserInfoAuthenticated> {
+    const userId = await getUserIdForEmail(event.email)
+    session.userInfo = {
+        userType: "authenticated",
+        nickname: event.name,
+        name: event.name,
+        email: event.email,
+        picture: event.picture,
+        domain: event.domain,
+        userId,
+    }
+    if (session.boardSession) {
+        // TODO SECURITY: don't reveal authenticated emails to unidentified users on same board
+        sendTo(everyoneElseOnTheSameBoard(session.boardSession.boardId, session), {
+            action: "user.login",
+            email: event.email,
+            name: event.name,
+            picture: event.picture,
+        })
+    }
+    return session.userInfo
+}
+
+export function logoutUser(event: AuthLogout, origin: WsWrapper) {
+    const session = getSession(origin)
+    if (!session) {
+        console.warn("Session not found for socket " + origin.id)
+    } else {
+        session.userInfo = { userType: "unidentified", nickname: session.userInfo.nickname }
+    }
+}
+
+export function broadcastBoardEvent(event: EventFromServer & { boardId: string }, origin?: UserSession) {
+    //console.log("Broadcast", event.action, "to", everyoneElseOnTheSameBoard(event.boardId, origin).length)
+    sendTo(everyoneElseOnTheSameBoard(event.boardId, origin), event)
+}
+
+export function broadcastJoinEvent(boardId: Id, session: UserSession) {
+    sendTo(everyoneElseOnTheSameBoard(boardId, session), {
+        action: "board.joined",
+        boardId,
+        sessionId: session.sessionId,
+        ...session.userInfo,
+    } as JoinedBoard)
+}
+
+export function broadcastCursorPositions(boardId: Id, positions: Record<Id, UserCursorPosition>) {
+    sendTo(everyoneOnTheBoard(boardId), { action: CURSOR_POSITIONS_ACTION_TYPE, p: positions })
+}
+
+const BROADCAST_DEBOUNCE_MS = 20
+
+// Debounce by 20ms per board id, otherwise every item interaction (e.g. drag on 10 items, one event each) broadcasts locks
+export const broadcastItemLocks = (() => {
+    let timeouts: Record<Id, NodeJS.Timeout | undefined> = {}
+    const hasActiveTimer = (boardId: string) => timeouts[boardId] !== undefined
+
+    return function _broadcastItemLocks(boardId: string, locks: ItemLocks) {
+        if (hasActiveTimer(boardId)) {
+            return
+        }
+        timeouts[boardId] = setTimeout(() => {
+            const boardState = maybeGetBoard(boardId)
+            if (boardState) {
+                sendTo(boardState.sessions, { action: "board.locks", boardId, locks })
+            }
+            timeouts[boardId] = undefined
+        }, BROADCAST_DEBOUNCE_MS)
+    }
+})()
+
+export function getSessionCount() {
+    return Object.values(sessions).length
+}

backend/src/ws-wrapper.ts

@@ -0,0 +1,48 @@
+import * as WebSocket from "ws"
+import * as uuid from "uuid"
+import { EventFromServer } from "../../common/src/domain"
+
+export const WsWrapper = (ws: WebSocket) => {
+    const onError = (f: () => void) => {
+        ws.addEventListener("error", f)
+    }
+    const onMessage = (f: (msg: object) => void) => {
+        ws.addEventListener("message", (msg: any) => {
+            try {
+                f(JSON.parse(msg.data))
+            } catch (e) {
+                console.error("Error in WsWrapper/onMessage. Closing connection.", e)
+                ws.close()
+            }
+        })
+    }
+    const onClose = (f: () => void) => {
+        ws.addEventListener("close", f)
+    }
+    return {
+        send: (buffer: Buffer) => {
+            try {
+                ws.send(buffer, { binary: false })
+            } catch (e) {
+                ws.close()
+            }
+        },
+        onError,
+        onMessage,
+        onClose,
+        id: uuid.v4(),
+        close: () => ws.close(),
+    }
+}
+export type WsWrapper = ReturnType<typeof WsWrapper>
+
+type CachedBuffer = { msg: EventFromServer; buffer: Buffer }
+let cachedBuffer: CachedBuffer | null = null
+export function toBuffer(msg: EventFromServer) {
+    // We cache the latest buffer to avoid creating a new buffer for every message.
+    if (cachedBuffer && cachedBuffer.msg === msg) {
+        return cachedBuffer.buffer
+    }
+    cachedBuffer = { msg, buffer: Buffer.from(JSON.stringify(msg)) }
+    return cachedBuffer.buffer
+}

backend/src/y-websocket-server/Docs.ts

@@ -0,0 +1,58 @@
+import { WSSharedDoc } from "./WSSharedDoc"
+import { Persistence } from "./Persistence"
+
+export interface DocsOptions {
+    persistence?: Persistence
+    gc?: boolean
+}
+
+interface DocState {
+    doc: WSSharedDoc
+    fetchPromise: Promise<void>
+}
+
+export class Docs {
+    readonly docs = new Map<string, DocState>()
+    readonly persistence: Persistence | null
+    readonly gc: boolean
+
+    constructor(options: DocsOptions = {}) {
+        this.persistence = options.persistence || null
+        this.gc = options.gc ?? true
+    }
+
+    /**
+     * Gets a Y.Doc by name, whether in memory or on disk
+     */
+    getYDoc(docname: string): WSSharedDoc {
+        return this.getDocState(docname).doc
+    }
+
+    async getYDocAndWaitForFetch(docname: string): Promise<WSSharedDoc> {
+        const state = this.getDocState(docname)
+        await state.fetchPromise
+        return state.doc
+    }
+
+    private getDocState(docname: string): DocState {
+        let state = this.docs.get(docname)
+        if (!state) {
+            const doc = new WSSharedDoc(this, docname)
+            console.log(`Loading document ${doc.name} into memory`)
+            doc.gc = this.gc
+            if (this.persistence !== null) {
+                void this.persistence.bindState(docname, doc)
+            }
+            const fetchPromise =
+                this.persistence !== null ? this.persistence.bindState(docname, doc) : Promise.resolve()
+            state = { doc, fetchPromise }
+            this.docs.set(docname, state)
+        }
+        return state
+    }
+
+    deleteYDoc(doc: WSSharedDoc) {
+        console.log(`Purging document ${doc.name} from memory`)
+        this.docs.delete(doc.name)
+    }
+}

backend/src/y-websocket-server/Persistence.ts

@@ -0,0 +1,25 @@
+import * as Y from "yjs"
+
+export interface Persistence {
+    bindState: (docName: string, ydoc: Y.Doc) => Promise<void>
+    writeState: (docName: string, ydoc: Y.Doc) => Promise<any>
+}
+
+export function createLevelDbPersistence(persistenceDir: string): Persistence {
+    console.info('Persisting documents to "' + persistenceDir + '"')
+    // @ts-ignore
+    const LeveldbPersistence = require("y-leveldb").LeveldbPersistence
+    const ldb = new LeveldbPersistence(persistenceDir)
+    return {
+        bindState: async (docName, ydoc) => {
+            const persistedYdoc = await ldb.getYDoc(docName)
+            const newUpdates = Y.encodeStateAsUpdate(ydoc)
+            ldb.storeUpdate(docName, newUpdates)
+            Y.applyUpdate(ydoc, Y.encodeStateAsUpdate(persistedYdoc))
+            ydoc.on("update", (update) => {
+                ldb.storeUpdate(docName, update)
+            })
+        },
+        writeState: async (docName, ydoc) => {},
+    }
+}

backend/src/y-websocket-server/Protocol.ts

@@ -0,0 +1,2 @@
+export const messageSync = 0
+export const messageAwareness = 1

backend/src/y-websocket-server/WSSharedDoc.ts

@@ -0,0 +1,105 @@
+import * as Y from "yjs"
+
+import * as awarenessProtocol from "y-protocols/awareness"
+import * as syncProtocol from "y-protocols/sync"
+
+import * as encoding from "lib0/encoding"
+import * as WebSocket from "ws"
+import { Docs } from "./Docs"
+import { messageAwareness, messageSync } from "./Protocol"
+
+export const wsReadyStateConnecting = 0
+export const wsReadyStateOpen = 1
+export const wsReadyStateClosing = 2
+export const wsReadyStateClosed = 3
+
+export class WSSharedDoc extends Y.Doc {
+    private docs: Docs
+    readonly name: string
+    private conns: Map<WebSocket, Set<number>> = new Map<WebSocket, Set<number>>()
+    readonly awareness = new awarenessProtocol.Awareness(this)
+
+    constructor(docs: Docs, name: string) {
+        super({ gc: docs.gc })
+        this.docs = docs
+        this.name = name
+        this.awareness.setLocalState(null)
+
+        const awarenessChangeHandler = (
+            { added, updated, removed }: { added: number[]; updated: number[]; removed: number[] },
+            conn: WebSocket,
+        ) => {
+            const changedClients = added.concat(updated, removed)
+            if (conn !== null) {
+                const connControlledIDs = this.conns.get(conn)
+                if (connControlledIDs !== undefined) {
+                    added.forEach((clientID) => {
+                        connControlledIDs.add(clientID)
+                    })
+                    removed.forEach((clientID) => {
+                        connControlledIDs.delete(clientID)
+                    })
+                }
+            }
+            const encoder = encoding.createEncoder()
+            encoding.writeVarUint(encoder, messageAwareness)
+            encoding.writeVarUint8Array(
+                encoder,
+                awarenessProtocol.encodeAwarenessUpdate(this.awareness, changedClients),
+            )
+            const buff = encoding.toUint8Array(encoder)
+            this.conns.forEach((_, c) => {
+                this.send(c, buff)
+            })
+        }
+        this.awareness.on("update", awarenessChangeHandler)
+
+        const updateHandler = (update: Uint8Array, origin: any, doc: WSSharedDoc) => {
+            const encoder = encoding.createEncoder()
+            encoding.writeVarUint(encoder, messageSync)
+            syncProtocol.writeUpdate(encoder, update)
+            const message = encoding.toUint8Array(encoder)
+            doc.conns.forEach((_, conn) => this.send(conn, message))
+        }
+
+        this.on("update", updateHandler)
+    }
+
+    send(conn: WebSocket, m: Uint8Array) {
+        if (conn.readyState !== wsReadyStateConnecting && conn.readyState !== wsReadyStateOpen) {
+            this.closeConn(conn)
+        }
+        try {
+            conn.send(m, (err: any) => {
+                err != null && this.closeConn(conn)
+            })
+        } catch (e) {
+            console.error("Failed to send message to client. Closing connection.", e)
+            this.closeConn(conn)
+        }
+    }
+
+    closeConn(conn: WebSocket) {
+        if (this.conns.has(conn)) {
+            const controlledIds = this.conns.get(conn)!
+            this.conns.delete(conn)
+            awarenessProtocol.removeAwarenessStates(this.awareness, Array.from(controlledIds), null)
+            if (this.conns.size === 0 && this.docs.persistence !== null) {
+                // if persisted, we store state and destroy ydocument
+                this.docs.persistence.writeState(this.name, this).then(() => {
+                    this.destroy()
+                })
+                this.docs.deleteYDoc(this)
+            }
+        }
+        conn.close()
+    }
+
+    addConnection(conn: WebSocket) {
+        this.conns.set(conn, new Set())
+    }
+
+    hasConnection(conn: WebSocket) {
+        return this.conns.has(conn)
+    }
+}

backend/src/y-websocket-server/YWebSocketServer.ts

@@ -0,0 +1,131 @@
+import * as awarenessProtocol from "y-protocols/awareness"
+import * as syncProtocol from "y-protocols/sync"
+
+import * as decoding from "lib0/decoding"
+import * as encoding from "lib0/encoding"
+import * as WebSocket from "ws"
+import { Docs, DocsOptions } from "./Docs"
+import { messageAwareness, messageSync } from "./Protocol"
+import { WSSharedDoc } from "./WSSharedDoc"
+
+const pingTimeout = 30000
+
+export default class YWebSocketServer {
+    docs: Docs
+    constructor(options?: DocsOptions) {
+        this.docs = new Docs(options)
+    }
+
+    async setupWSConnection(conn: WebSocket, docName: string, readOnly: boolean) {
+        conn.binaryType = "arraybuffer"
+        // get doc, initialize if it does not exist yet
+        const doc = this.docs.getYDoc(docName)
+
+        console.log(`YJS connection established for ${docName}`)
+
+        doc.addConnection(conn)
+        // listen and reply to events
+        conn.on("message", (message: ArrayBuffer) => messageListener(conn, doc, readOnly, new Uint8Array(message)))
+
+        // Check if connection is still alive
+        let pongReceived = true
+        const pingInterval = setInterval(() => {
+            if (!pongReceived) {
+                if (doc.hasConnection(conn)) {
+                    doc.closeConn(conn)
+                }
+                clearInterval(pingInterval)
+            } else if (doc.hasConnection(conn)) {
+                pongReceived = false
+                try {
+                    conn.ping()
+                } catch (e) {
+                    doc.closeConn(conn)
+                    clearInterval(pingInterval)
+                }
+            }
+        }, pingTimeout)
+        conn.on("close", () => {
+            doc.closeConn(conn)
+            clearInterval(pingInterval)
+        })
+        conn.on("pong", () => {
+            pongReceived = true
+        })
+        // put the following in a variables in a block so the interval handlers don't keep in in
+        // scope
+        {
+            // send sync step 1
+            const encoder = encoding.createEncoder()
+            encoding.writeVarUint(encoder, messageSync)
+            syncProtocol.writeSyncStep1(encoder, doc)
+            doc.send(conn, encoding.toUint8Array(encoder))
+            const awarenessStates = doc.awareness.getStates()
+            if (awarenessStates.size > 0) {
+                const encoder = encoding.createEncoder()
+                encoding.writeVarUint(encoder, messageAwareness)
+                encoding.writeVarUint8Array(
+                    encoder,
+                    awarenessProtocol.encodeAwarenessUpdate(doc.awareness, Array.from(awarenessStates.keys())),
+                )
+                doc.send(conn, encoding.toUint8Array(encoder))
+            }
+        }
+    }
+}
+
+// Read-only implementation found at https://discuss.yjs.dev/t/read-only-or-one-way-only-sync/135/3
+const readSyncMessage = (
+    decoder: decoding.Decoder,
+    encoder: encoding.Encoder,
+    doc: WSSharedDoc,
+    readOnly = false,
+    transactionOrigin: any,
+) => {
+    const messageType = decoding.readVarUint(decoder)
+    switch (messageType) {
+        case syncProtocol.messageYjsSyncStep1:
+            syncProtocol.readSyncStep1(decoder, encoder, doc)
+            break
+        case syncProtocol.messageYjsSyncStep2:
+            if (!readOnly) syncProtocol.readSyncStep2(decoder, doc, transactionOrigin)
+            break
+        case syncProtocol.messageYjsUpdate:
+            if (!readOnly) syncProtocol.readUpdate(decoder, doc, transactionOrigin)
+            break
+        default:
+            throw new Error("Unknown message type")
+    }
+    return messageType
+}
+
+const messageListener = (conn: WebSocket, doc: WSSharedDoc, readOnly: boolean, message: Uint8Array) => {
+    try {
+        const encoder = encoding.createEncoder()
+        const decoder = decoding.createDecoder(message)
+        const messageType = decoding.readVarUint(decoder)
+        switch (messageType) {
+            case messageSync:
+                encoding.writeVarUint(encoder, messageSync)
+                readSyncMessage(decoder, encoder, doc, readOnly, conn)
+
+                // If the `encoder` only contains the type of reply message and no
+                // message, there is no need to send the message. When `encoder` only
+                // contains the type of reply, its length is 1.
+                if (encoding.length(encoder) > 1) {
+                    doc.send(conn, encoding.toUint8Array(encoder))
+                }
+                break
+            case messageAwareness: {
+                awarenessProtocol.applyAwarenessUpdate(doc.awareness, decoding.readVarUint8Array(decoder), conn)
+                break
+            }
+            default: {
+                console.warn("Unexpected message type" + messageType)
+            }
+        }
+    } catch (err) {
+        console.error(err)
+        doc.emit("error", [err])
+    }
+}

backend/tsconfig.json

@@ -0,0 +1,9 @@
+{
+    "extends": "../tsconfig",
+    "compilerOptions": {
+        "module": "commonjs",
+        "outDir": "./dist",
+        "rootDir": "..",
+        "sourceMap": true
+    }
+}