Rxdb

.env.development

@@ -10,6 +10,10 @@ VITE_FEEDBACK_URL=https://us-central1-em-search-test.cloudfunctions.net/sendFeed
 VITE_WEBSOCKET_HOST=localhost
 VITE_WEBSOCKET_PORT=3001
 
+# WEBRTC SERVER
+VITE_WEBRTC_HOST=localhost
+VITE_WEBRTC_PORT=3002
+
 # CAPACITOR
 # Local IP addresses should only be defined in .env.development.local so that they are not checked into version control.
 CAPACITOR_SERVER_URL=

.env.production

@@ -9,5 +9,9 @@ VITE_FEEDBACK_URL=https://us-central1-em-proto.cloudfunctions.net/sendFeedbackEm
 VITE_WEBSOCKET_HOST=em-staging-sa2jm.ondigitalocean.app
 VITE_WEBSOCKET_PORT=
 
+# WEBRTC SERVER
+VITE_WEBRTC_HOST=
+VITE_WEBRTC_PORT=
+
 # AI SERVER
 VITE_AI_URL=https://em-staging-sa2jm.ondigitalocean.app/ai

package.json

@@ -28,6 +28,7 @@
  "cap:sync:prod": "npm run build && NODE_ENV=production cap sync",
  "start": "vite --host --port 3000",
  "build": "vite build",
+ "build:premium": "node scripts/rxdb-premium.js",
  "clean": "rm -rf ./node_modules/.cache",
  "docs": "typedoc --options typedoc.json",
  "deploy:major": "npm run lint && npm version major && npm run build && firebase deploy --only hosting",
@@ -77,7 +78,6 @@
  "@capacitor/status-bar": "^4.1.1",
  "@emotion/react": "^11.11.4",
  "@emotion/styled": "^11.11.5",
- "@hocuspocus/provider": "^2.13.5",
  "@mui/material": "^5.15.20",
  "algoliasearch": "^4.24.0",
  "axios": "^1.7.2",
@@ -120,6 +120,8 @@
  "redux-devtools-extension": "^2.13.9",
  "redux-thunk": "^2.4.2",
  "reselect": "^5.1.1",
+ "rxdb": "^15.28.1",
+ "rxjs": "^7.8.1",
  "text-block-parser": "^1.1.1",
  "truncate-html": "^1.1.1",
  "ts-key-enum": "^2.0.12",
@@ -130,9 +132,7 @@
  "workbox-routing": "^7.1.0",
  "workbox-strategies": "^7.1.0",
  "xhtml-purifier": "^0.4.1",
- "y-indexeddb": "https://github.com/raineorshine/y-indexeddb#y-indexeddb-multiplex",
- "yallist": "^5.0.0",
- "yjs": "^13.6.17"
+ "yallist": "^5.0.0"
  },
  "devDependencies": {
  "@babel/core": "^7.24.7",

rxdb-server/.gitignore

@@ -0,0 +1,2 @@
+/node_modules
+/build

rxdb-server/package.json

@@ -0,0 +1,27 @@
+{
+ "name": "rxdb-server",
+ "version": "1.0.0",
+ "main": "main",
+ "scripts": {
+ "build": "rm -rf build ; npx tsc",
+ "kill": "pm2 kill",
+ "start": "pm2 start pm2.config.js",
+ "logs": "pm2 logs",
+ "monit": "pm2 monit",
+ "restart": "pm2 restart pm2.config.js",
+ "status": "pm2 status pm2.config.js",
+ "stop": "pm2 stop pm2.config.js"
+ },
+ "keywords": [],
+ "author": "",
+ "license": "ISC",
+ "description": "",
+ "devDependencies": {
+ "pm2": "^5.4.2",
+ "typescript": "^5.5.3"
+ },
+ "dependencies": {
+ "rxdb": "15.27.0",
+ "ws": "^8.18.0"
+ }
+}

rxdb-server/pm2.config.js

@@ -0,0 +1,13 @@
+// pm2 configuration
+// https://pm2.keymetrics.io/docs/usage/application-declaration/
+
+module.exports = {
+ apps: [
+ {
+ name: 'server',
+ script: './build/index.js',
+ time: true,
+ wait_ready: true,
+ },
+ ],
+}

rxdb-server/src/index.ts

@@ -0,0 +1,172 @@
+import { getFromMapOrCreate, promiseWait, randomCouchString } from 'rxdb'
+import { type PeerMessage, SIMPLE_PEER_PING_INTERVAL } from 'rxdb/plugins/replication-webrtc'
+import type { ServerOptions, WebSocket } from 'ws'
+
+const PORT = 3002 // TODO - Get from env
+
+export const PEER_ID_LENGTH = 12
+export type ServerPeer = {
+ id: string
+ socket: WebSocket
+ rooms: Set<string>
+ lastPing: number
+}
+
+/**
+ * Starts a WebRTC signaling server
+ * that can be used in tests.
+ */
+export async function startSignalingServerSimplePeer(serverOptions: ServerOptions) {
+ const { WebSocketServer } = await import('ws')
+ const wss = new WebSocketServer(serverOptions)
+
+ const peerById = new Map<string, ServerPeer>()
+ const peersByRoom = new Map<string, Set<string>>()
+
+ let serverClosed = false
+ wss.on('close', () => {
+ serverClosed = true
+ peerById.clear()
+ peersByRoom.clear()
+ })
+
+ /**
+ * Clients can disconnect without telling that to the
+ * server. Therefore we have to automatically disconnect clients that
+ * have not send a ping message in the last 2 minutes.
+ */
+ ;(async () => {
+ // eslint-disable-next-line no-unmodified-loop-condition
+ while (!serverClosed) {
+ await promiseWait(1000 * 5)
+ const minTime = Date.now() - SIMPLE_PEER_PING_INTERVAL
+ Array.from(peerById.values()).forEach(peer => {
+ if (peer.lastPing < minTime) {
+ disconnectSocket(peer.id, 'no ping for 2 minutes')
+ }
+ })
+ }
+ })()
+
+ /** Disconnect a peer. */
+ function disconnectSocket(peerId: string, reason: string) {
+ const peer = peerById.get(peerId)
+ if (peer) {
+ peer.socket.close && peer.socket.close(undefined, reason)
+ peer.rooms.forEach(roomId => {
+ const room = peersByRoom.get(roomId)
+ room?.delete(peerId)
+ if (room && room.size === 0) {
+ peersByRoom.delete(roomId)
+ }
+ })
+ }
+ peerById.delete(peerId)
+ }
+
+ wss.on('connection', function (ws) {
+ /**
+ * PeerID is created by the server to prevent malicious
+ * actors from falsy claiming other peoples ids.
+ */
+ const peerId = randomCouchString(PEER_ID_LENGTH)
+ const peer: ServerPeer = {
+ id: peerId,
+ socket: ws,
+ rooms: new Set(),
+ lastPing: Date.now(),
+ }
+ peerById.set(peerId, peer)
+
+ sendMessage(ws, { type: 'init', yourPeerId: peerId })
+
+ ws.on('error', err => {
+ console.error('SERVER ERROR:', err)
+ disconnectSocket(peerId, 'socket errored')
+ })
+ ws.on('close', () => {
+ disconnectSocket(peerId, 'socket disconnected')
+ })
+
+ ws.on('message', msgEvent => {
+ peer.lastPing = Date.now()
+ const message = JSON.parse(msgEvent.toString())
+ const type = message.type
+ switch (type) {
+ case 'join': {
+ const roomId = message.room
+ if (!validateIdString(roomId) || !validateIdString(peerId)) {
+ disconnectSocket(peerId, 'invalid ids')
+ return
+ }
+
+ if (peer.rooms.has(peerId)) {
+ return
+ }
+ peer.rooms.add(roomId)
+
+ const room = getFromMapOrCreate(peersByRoom, message.room, () => new Set())
+
+ room.add(peerId)
+
+ // tell everyone about new room state
+ room.forEach(otherPeerId => {
+ const otherPeer = peerById.get(otherPeerId)
+ if (otherPeer) {
+ sendMessage(otherPeer.socket, {
+ type: 'joined',
+ otherPeerIds: Array.from(room),
+ })
+ }
+ })
+ break
+ }
+ case 'signal': {
+ if (message.senderPeerId !== peerId) {
+ disconnectSocket(peerId, 'spoofed sender')
+ return
+ }
+ const receiver = peerById.get(message.receiverPeerId)
+ if (receiver) {
+ sendMessage(receiver.socket, message)
+ }
+ break
+ }
+ case 'ping':
+ break
+ default:
+ disconnectSocket(peerId, 'unknown message type ' + type)
+ }
+ })
+ })
+
+ return {
+ port: serverOptions.port,
+ server: wss,
+ localUrl: 'ws://localhost:' + serverOptions.port,
+ }
+}
+
+/** Send a message to a peer. */
+function sendMessage(ws: WebSocket, message: PeerMessage) {
+ const msgString = JSON.stringify(message)
+ ws.send(msgString)
+}
+
+/** Validate that a string is a valid id. */
+function validateIdString(roomId: string): boolean {
+ if (typeof roomId === 'string' && roomId.length > 5 && roomId.length < 100) {
+ return true
+ } else {
+ return false
+ }
+}
+
+/** Main. */
+async function main() {
+ await startSignalingServerSimplePeer({
+ port: PORT,
+ })
+}
+
+main()

rxdb-server/tsconfig.json

@@ -0,0 +1,9 @@
+{
+ "extends": "../tsconfig.json",
+ "include": ["src/index.ts"],
+ "compilerOptions": {
+ "module": "commonjs",
+ "noEmit": false,
+ "outDir": "build"
+ }
+}