feat: Optimize network traffic with patch-based sync for Node.js server

package.json

@@ -12,6 +12,7 @@
     "buildsite": "vite build --outDir dist",
     "updatePlugin": "tsc public/pluginApi.ts",
     "runserver": "node server/node/server.cjs",
+    "runserver:patch": "node server/node/server.cjs --patch-sync",
     "sync": "node electron/sync",
     "electron": "node electron/dist/electron"
   },
@@ -56,6 +57,7 @@
     "eventsource-parser": "^1.1.2",
     "exifr": "^7.1.3",
     "express": "^4.18.2",
+    "fast-json-patch": "^3.1.1",
     "fflate": "^0.8.1",
     "gpt-3-encoder": "^1.1.4",
     "gpt3-tokenizer": "^1.1.5",

server/node/server.cjs

@@ -5,6 +5,9 @@
 const { existsSync, mkdirSync, readFileSync, writeFileSync } = require('fs');
 const fs = require('fs/promises')
 const crypto = require('crypto')
+const { applyPatch } = require('fast-json-patch')
+const { Packr, Unpackr, decode } = require('msgpackr')
+const fflate = require('fflate')
 app.use(express.static(path.join(process.cwd(), 'dist'), {index: false}));
 app.use(express.json({ limit: '50mb' }));
 app.use(express.raw({ type: 'application/octet-stream', limit: '50mb' }));
@@ -15,6 +18,24 @@
 
 let password = ''
 
+// Configuration flags for patch-based sync
+let enablePatchSync = process.env.RISU_PATCH_SYNC === '1' || process.argv.includes('--patch-sync')
+
+if (enablePatchSync) {
+    const [major, minor, patch] = process.version.slice(1).split('.').map(Number);
+    // v22.7.0, v23 and above have a bug with msgpackr that causes it to crash on encoding risu saves
+    if (major >= 23 || (major === 22 && minor === 7 && patch === 0)) {
+        console.log(`[Server] Detected problematic Node.js version ${process.version}. Disabling patch-based sync.`);
+        enablePatchSync = false;
+    }
+}
+
+// In-memory database cache for patch-based sync with versioning
+let dbCache = {}
+let dbVersions = {} // Track version numbers for each file
+let saveTimers = {}
+const SAVE_INTERVAL = 5000 // Save to disk after 5 seconds of inactivity
+
 const savePath = path.join(process.cwd(), "save")
 if(!existsSync(savePath)){
     mkdirSync(savePath)
@@ -29,6 +50,79 @@
     return hexRegex.test(str.toUpperCase().trim()) || str === '__password';
 }
 
+// Encoding/decoding functions for RisuSave format
+const packr = new Packr({ useRecords: false });
+const unpackr = new Unpackr({ int64AsType: 'number', useRecords: false });
+
+const magicHeader = new Uint8Array([0, 82, 73, 83, 85, 83, 65, 86, 69, 0, 7]); 
+const magicCompressedHeader = new Uint8Array([0, 82, 73, 83, 85, 83, 65, 86, 69, 0, 8]);
+
+function checkHeader(data) {
+    let header = 'raw';
+
+    if (data.length < magicHeader.length) {
+        return 'none';
+    }
+
+    for (let i = 0; i < magicHeader.length; i++) {
+        if (data[i] !== magicHeader[i]) {
+            header = 'none';
+            break;
+        }
+    }
+
+    if (header === 'none') {
+        header = 'compressed';
+        for (let i = 0; i < magicCompressedHeader.length; i++) {
+            if (data[i] !== magicCompressedHeader[i]) {
+                header = 'none';
+                break;
+            }
+        }
+    }
+
+    return header;
+}
+
+async function decodeRisuSaveServer(data) {
+    try {
+        switch(checkHeader(data)){
+            case "compressed":
+                data = data.slice(magicCompressedHeader.length)
+                return decode(fflate.decompressSync(data))
+            case "raw":
+                data = data.slice(magicHeader.length)
+                return unpackr.decode(data)
+        }
+        return unpackr.decode(data)
+    }
+    catch (error) {
+        try {
+            console.log('risudecode')
+            const risuSaveHeader = new Uint8Array(Buffer.from("\u0000\u0000RISU",'utf-8'))
+            const realData = data.subarray(risuSaveHeader.length)
+            const dec = unpackr.decode(realData)
+            return dec   
+        } catch (error) {
+            const buf = Buffer.from(fflate.decompressSync(Buffer.from(data)))
+            try {
+                return JSON.parse(buf.toString('utf-8'))                            
+            } catch (error) {
+                return unpackr.decode(buf)
+            }
+        }
+    }
+}
+
+async function encodeRisuSaveServer(data) {
+    // Encode to legacy format (no compression for simplicity)
+    const encoded = packr.encode(data);
+    const result = new Uint8Array(encoded.length + magicHeader.length);
+    result.set(magicHeader, 0);
+    result.set(encoded, magicHeader.length);
+    return result;
+}
+
 app.get('/', async (req, res, next) => {
 
     const clientIP = req.headers['x-forwarded-for'] || req.ip || req.socket.remoteAddress || 'Unknown IP';
@@ -39,7 +133,7 @@
         const mainIndex = await fs.readFile(path.join(process.cwd(), 'dist', 'index.html'))
         const root = htmlparser.parse(mainIndex)
         const head = root.querySelector('head')
-        head.innerHTML = `<script>globalThis.__NODE__ = true</script>` + head.innerHTML
+        head.innerHTML = `<script>globalThis.__NODE__ = true; globalThis.__PATCH_SYNC__ = ${enablePatchSync}</script>` + head.innerHTML
 
         res.send(root.toString())
     } catch (error) {
@@ -212,26 +306,26 @@
         res.send({status: 'unset'})
     }
     else if(req.headers['risu-auth']  === password){
         res.send({status:'correct'})
     }
     else{
         res.send({status:'incorrect'})
     }
 })
 
 app.post('/api/crypto', async (req, res) => {
     try {
         const hash = crypto.createHash('sha256')
         hash.update(Buffer.from(req.body.data, 'utf-8'))
         res.send(hash.digest('hex'))
     } catch (error) {
         next(error)
     }
 })
 
 
 app.post('/api/set_password', async (req, res) => {
     if(password === ''){
         password = req.body.password
         writeFileSync(passwordPath, password, 'utf-8')
     }
@@ -262,6 +356,26 @@
         return;
     }
     try {
+        const fullPath = path.join(savePath, filePath);
+
+        // Stop any pending save timer for this file
+        if (saveTimers[filePath]) {
+            clearTimeout(saveTimers[filePath]);
+            delete saveTimers[filePath];
+        }
+
+        // write to disk if available in cache
+        if (dbCache[filePath]) {
+            const decodedFilePath = Buffer.from(filePath, 'hex').toString('utf-8');
+            let dataToSave = await encodeRisuSaveServer(dbCache[filePath]);
+            await fs.writeFile(fullPath, dataToSave);
+        }
+
+        // Clear cache and reset version after read operation
+        if (dbCache[filePath]) delete dbCache[filePath];
+        dbVersions[filePath] = 0;
+
+        // read from disk
         if(!existsSync(path.join(savePath, filePath))){
             res.send();
         }
@@ -315,9 +429,10 @@
         return
     }
     try {
-        const data = (await fs.readdir(path.join(savePath))).map((v) => {
-            return Buffer.from(v, 'hex').toString('utf-8')
-        })
+        const data = (await fs.readdir(path.join(savePath)))
+            .map((v) => { return Buffer.from(v, 'hex').toString('utf-8') })
+            .filter((v) => { return v.startsWith(req.headers['key-prefix'].trim()) })
+
         res.send({
             success: true,
             content: data
@@ -352,6 +467,15 @@
 
     try {
         await fs.writeFile(path.join(savePath, filePath), fileContent);
+        // Clear cache for this file since it was directly written
+        if (dbCache[filePath]) delete dbCache[filePath];
+        // Clear any pending save timer for this file
+        if (saveTimers[filePath]) {
+            clearTimeout(saveTimers[filePath]);
+            delete saveTimers[filePath];
+        }
+        // Reset version to 0 after direct write
+        dbVersions[filePath] = 0;
         res.send({
             success: true
         });
@@ -360,6 +484,116 @@
     }
 });
 
+app.post('/api/patch', async (req, res, next) => {
+    // Check if patch sync is enabled
+    if (!enablePatchSync) {
+        res.status(404).send({
+            error: 'Patch sync is not enabled'
+        });
+        return;
+    }
+
+    if(req.headers['risu-auth'].trim() !== password.trim()){
+        console.log('incorrect')
+        res.status(400).send({
+            error:'Password Incorrect'
+        });
+        return
+    }
+    const filePath = req.headers['file-path'];
+    const patch = req.body.patch;
+    const clientVersion = parseInt(req.body.expectedVersion) || 0;
+
+    if (!filePath || !patch) {
+        res.status(400).send({
+            error:'File path and patch required'
+        });
+        return;
+    }
+    if(!isHex(filePath)){
+        res.status(400).send({
+            error:'Invaild Path'
+        });
+        return;
+    }
+
+    try {
+        const decodedFilePath = Buffer.from(filePath, 'hex').toString('utf-8');
+
+        // Initialize version if not exists
+        if (!dbVersions[filePath]) dbVersions[filePath] = 0;
+
+        // Check version mismatch
+        const serverVersion = dbVersions[filePath];
+        if (clientVersion !== serverVersion) {
+            console.log(`[Patch] Version mismatch for ${decodedFilePath}: client=${clientVersion}, server=${serverVersion}`);
+            res.status(409).send({
+                error: 'Version mismatch',
+            });
+            return;
+        }
+
+        // Load database into memory if not already cached
+        if (!dbCache[filePath]) {
+            const fullPath = path.join(savePath, filePath);
+            if (existsSync(fullPath)) {
+                const fileContent = await fs.readFile(fullPath);
+                dbCache[filePath] = await decodeRisuSaveServer(fileContent);
+            } 
+            else {
+                dbCache[filePath] = {};
+            }
+        }
+
+        // Apply patch to in-memory database
+        const result = applyPatch(dbCache[filePath], patch, true);
+
+        // Increment version after successful patch
+        dbVersions[filePath]++;
+        const newVersion = dbVersions[filePath];
+
+        // Schedule save to disk (debounced)
+        if (saveTimers[filePath]) {
+            clearTimeout(saveTimers[filePath]);
+        }
+        saveTimers[filePath] = setTimeout(async () => {
+            try {
+                const fullPath = path.join(savePath, filePath);
+                let dataToSave = await encodeRisuSaveServer(dbCache[filePath]);
+                await fs.writeFile(fullPath, dataToSave);
+                // Create backup for database files after successful save
+                if (decodedFilePath.includes('database/database.bin')) {
+                    try {
+                        const timestamp = Math.floor(Date.now() / 100).toString();
+                        const backupFileName = `database/dbbackup-${timestamp}.bin`;
+                        const backupFilePath = Buffer.from(backupFileName, 'utf-8').toString('hex');
+                        const backupFullPath = path.join(savePath, backupFilePath);
+                        // Create backup using the same data that was just saved
+                        await fs.writeFile(backupFullPath, dataToSave);
+                    } catch (backupError) {
+                        console.error(`[Patch] Error creating backup:`, backupError);
+                    }
+                }
+            } catch (error) {
+                console.error(`[Patch] Error saving ${filePath}:`, error);
+            } finally {
+                delete saveTimers[filePath];
+            }
+        }, SAVE_INTERVAL);
+
+        res.send({
+            success: true,
+            appliedOperations: result.length,
+            newVersion: newVersion
+        });
+    } catch (error) {
+        console.error(`[Patch] Error applying patch to ${filePath}:`, error);
+        res.status(500).send({
+            error: 'Patch application failed: ' + (error && error.message ? error.message : error)
+        });
+    }
+});
+
 async function getHttpsOptions() {
 
     const keyPath = path.join(sslPath, 'server.key');
@@ -388,19 +622,19 @@
     try {
 
         const port = process.env.PORT || 6001;
-        const httpsOptions = await getHttpsOptions();
-
-        if (httpsOptions) {
+        const httpsOptions = await getHttpsOptions();        if (httpsOptions) {
             // HTTPS
             https.createServer(httpsOptions, app).listen(port, () => {
                 console.log("[Server] HTTPS server is running.");
                 console.log(`[Server] https://localhost:${port}/`);
+                console.log(`[Server] Patch sync: ${enablePatchSync ? 'ENABLED' : 'DISABLED'}`);
             });
         } else {
             // HTTP
             app.listen(port, () => {
                 console.log("[Server] HTTP server is running.");
                 console.log(`[Server] http://localhost:${port}/`);
+                console.log(`[Server] Patch sync: ${enablePatchSync ? 'ENABLED' : 'DISABLED'}`);
             });
         }
     } catch (error) {