Merge pull request #27 from rsscloud/feat/dual-write-json-store

andrewshell · web-flow · commit 096f46ac9408 · 2026-03-18T12:32:45.000-05:00
feat: add dual-write JSON file store alongside MongoDB
diff --git a/app.js b/app.js
@@ -5,6 +5,7 @@ const config = require('./config'),
     express = require('express'),
     exphbs = require('express-handlebars'),
     getDayjs = require('./services/dayjs-wrapper'),
+    jsonStore = require('./services/json-store'),
     mongodb = require('./services/mongodb'),
     morgan = require('morgan'),
     removeExpiredSubscriptions = require('./services/remove-expired-subscriptions'),
@@ -67,10 +68,38 @@ app.use(express.static('public', {
 app.use(require('./controllers'));
 
 // Start server
+async function seedJsonStore() {
+    const db = mongodb.get('rsscloud');
+    const resources = await db.collection('resources').find({}).toArray();
+    const subscriptions = await db.collection('subscriptions').find({}).toArray();
+
+    for (const resource of resources) {
+        jsonStore.setResource(resource._id, resource);
+    }
+
+    for (const sub of subscriptions) {
+        jsonStore.setSubscriptions(sub._id, sub.pleaseNotify || []);
+    }
+
+    jsonStore.flush();
+}
+
+async function gracefulShutdown() {
+    jsonStore.shutdown();
+    await mongodb.closeAll();
+    process.exit();
+}
+
+process.on('SIGINT', gracefulShutdown);
+process.on('SIGTERM', gracefulShutdown);
+
 async function startServer() {
     await initializeDayjs();
     await mongodb.connect('rsscloud', config.mongodbUri);
 
+    jsonStore.initialize(config.dataFilePath);
+    await seedJsonStore();
+
     // Start cleanup scheduling
     scheduleCleanupTasks();
 
diff --git a/config.js b/config.js
@@ -21,5 +21,6 @@ module.exports = {
     maxResourceSize: getNumericConfig('MAX_RESOURCE_SIZE', 256000),
     ctSecsResourceExpire: getNumericConfig('CT_SECS_RESOURCE_EXPIRE', 90000),
     minSecsBetweenPings: getNumericConfig('MIN_SECS_BETWEEN_PINGS', 0),
-    requestTimeout: getNumericConfig('REQUEST_TIMEOUT', 4000)
+    requestTimeout: getNumericConfig('REQUEST_TIMEOUT', 4000),
+    dataFilePath: getConfig('DATA_FILE_PATH', './data/subscriptions.json')
 };
diff --git a/controllers/index.js b/controllers/index.js
@@ -1,4 +1,5 @@
 const express = require('express'),
+    jsonStore = require('../services/json-store'),
     router = new express.Router();
 
 router.use('/', require('./home'));
@@ -10,4 +11,9 @@ router.use('/pingForm', require('./ping-form'));
 router.use('/viewLog', require('./view-log'));
 router.use('/RPC2', require('./rpc2'));
 
+router.get('/subscriptions.json', (req, res) => {
+    res.set('Content-Type', 'application/json');
+    res.send(JSON.stringify(jsonStore.getData(), null, 2));
+});
+
 module.exports = router;
diff --git a/package.json b/package.json
@@ -4,8 +4,8 @@
   "description": "An rssCloud Server",
   "main": "app.js",
   "scripts": {
-    "start": "nodemon --use_strict ./app.js",
-    "client": "nodemon --use_strict ./client.js",
+    "start": "nodemon --use_strict --ignore data/ ./app.js",
+    "client": "nodemon --use_strict --ignore data/ ./client.js",
     "import-data": "node ./bin/import-data.js",
     "lint": "eslint --fix controllers/ services/ test/ *.js",
     "format": "prettier --write .",
diff --git a/services/init-resource.js b/services/init-resource.js
@@ -3,7 +3,6 @@ const getDayjs = require('./dayjs-wrapper');
 async function initResource(resource) {
     const dayjs = await getDayjs();
     const defaultResource = {
-        flDirty: true,
         lastSize: 0,
         lastHash: '',
         ctChecks: 0,
diff --git a/services/json-store.js b/services/json-store.js
@@ -0,0 +1,84 @@
+const fs = require('fs');
+const path = require('path');
+
+let data = {};
+let filePath = null;
+let flushTimer = null;
+let flushing = false;
+
+function initialize(dataFilePath, flushIntervalMs = 60000) {
+    filePath = dataFilePath;
+
+    const dir = path.dirname(filePath);
+    fs.mkdirSync(dir, { recursive: true });
+
+    if (fs.existsSync(filePath)) {
+        try {
+            data = JSON.parse(fs.readFileSync(filePath, 'utf8'));
+        } catch {
+            data = {};
+        }
+    }
+
+    flushTimer = setInterval(() => flush(), flushIntervalMs);
+}
+
+function setResource(feedUrl, resourceObj) {
+    if (!data[feedUrl]) {
+        data[feedUrl] = { resource: {}, subscribers: [] };
+    }
+    const clean = Object.assign({}, resourceObj);
+    delete clean._id;
+    data[feedUrl].resource = clean;
+}
+
+function setSubscriptions(feedUrl, pleaseNotifyArray) {
+    if (!data[feedUrl]) {
+        data[feedUrl] = { resource: {}, subscribers: [] };
+    }
+    data[feedUrl].subscribers = pleaseNotifyArray.map(sub => {
+        const clean = Object.assign({}, sub);
+        delete clean._id;
+        return clean;
+    });
+}
+
+function getData() {
+    return data;
+}
+
+function clear() {
+    data = {};
+}
+
+function flush() {
+    if (!filePath || flushing) {
+        return;
+    }
+    flushing = true;
+    try {
+        const tmpPath = filePath + '.tmp';
+        fs.writeFileSync(tmpPath, JSON.stringify(data, null, 2));
+        fs.renameSync(tmpPath, filePath);
+    } finally {
+        flushing = false;
+    }
+}
+
+function shutdown() {
+    if (flushTimer) {
+        clearInterval(flushTimer);
+        flushTimer = null;
+    }
+    flush();
+}
+
+module.exports = {
+    initialize,
+    setResource,
+    setSubscriptions,
+    getData,
+    clear,
+    flush,
+    shutdown
+};
diff --git a/services/mongodb.js b/services/mongodb.js
@@ -32,16 +32,6 @@ async function closeAll() {
     await Promise.all(Object.keys(state).map(close));
 }
 
-function cleanup() {
-    closeAll()
-        .finally(() => {
-            process.exit();
-        });
-}
-
-process.on('SIGINT', cleanup);
-process.on('SIGTERM', cleanup);
-
 module.exports = {
     connect,
     get,
diff --git a/services/notify-subscribers.js b/services/notify-subscribers.js
@@ -1,5 +1,6 @@
 const config = require('../config'),
     getDayjs = require('./dayjs-wrapper'),
+    jsonStore = require('./json-store'),
     logEvent = require('./log-event'),
     mongodb = require('./mongodb'),
     notifyOne = require('./notify-one');
@@ -22,6 +23,7 @@ async function upsertSubscriptions(subscriptions) {
             subscriptions,
             { upsert: true }
         );
+    jsonStore.setSubscriptions(subscriptions._id, subscriptions.pleaseNotify);
 }
 
 async function notifyOneSubscriber(resourceUrl, subscription) {
diff --git a/services/ping.js b/services/ping.js
@@ -4,6 +4,7 @@ const appMessage = require('./app-messages'),
     ErrorResponse = require('./error-response'),
     getDayjs = require('./dayjs-wrapper'),
     initResource = require('./init-resource'),
+    jsonStore = require('./json-store'),
     logEvent = require('./log-event'),
     mongodb = require('./mongodb'),
     notifySubscribers = require('./notify-subscribers');
@@ -56,13 +57,7 @@ async function checkForResourceChange(resource, resourceUrl, startticks) {
 
     const hash = md5Hash(body);
 
-    if (resource.lastHash !== hash) {
-        resource.flDirty = true;
-    } else if (resource.lastSize !== body.length) {
-        resource.flDirty = true;
-    } else {
-        resource.flDirty = false;
-    }
+    const changed = (resource.lastHash !== hash) || (resource.lastSize !== body.length);
 
     resource.lastHash = hash;
     resource.lastSize = body.length;
@@ -71,7 +66,7 @@ async function checkForResourceChange(resource, resourceUrl, startticks) {
         'Ping',
         {
             resourceUrl: resourceUrl,
-            changed: resource.flDirty,
+            changed: changed,
             hash: resource.lastHash,
             size: resource.lastSize,
             stats: {
@@ -81,6 +76,8 @@ async function checkForResourceChange(resource, resourceUrl, startticks) {
         },
         startticks
     );
+
+    return changed;
 }
 
 async function fetchResource(resourceUrl) {
@@ -101,10 +98,11 @@ async function upsertResource(resource) {
             resource,
             { upsert: true }
         );
+    jsonStore.setResource(resource._id, resource);
 }
 
-async function notifySubscribersIfDirty(resource, resourceUrl) {
-    if (resource.flDirty) {
+async function notifySubscribersIfDirty(changed, resource, resourceUrl) {
+    if (changed) {
         const dayjs = await getDayjs();
         resource.ctUpdates += 1;
         resource.whenLastUpdate = new Date(dayjs().utc().format());
@@ -120,8 +118,8 @@ async function ping(resourceUrl) {
         );
 
     await checkPingFrequency(resource);
-    await checkForResourceChange(resource, resourceUrl, startticks);
-    await notifySubscribersIfDirty(resource, resourceUrl);
+    const changed = await checkForResourceChange(resource, resourceUrl, startticks);
+    await notifySubscribersIfDirty(changed, resource, resourceUrl);
     await upsertResource(resource);
 
     return {
diff --git a/services/please-notify.js b/services/please-notify.js
@@ -3,6 +3,7 @@ const appMessages = require('./app-messages'),
     ErrorResponse = require('./error-response'),
     getDayjs = require('./dayjs-wrapper'),
     initSubscription = require('./init-subscription'),
+    jsonStore = require('./json-store'),
     logEvent = require('./log-event'),
     mongodb = require('./mongodb'),
     notifyOne = require('./notify-one'),
@@ -47,6 +48,7 @@ async function upsertSubscriptions(subscriptions) {
             subscriptions,
             { upsert: true }
         );
+    jsonStore.setSubscriptions(subscriptions._id, subscriptions.pleaseNotify);
 }
 
 async function notifyApiUrl(notifyProcedure, apiurl, protocol, resourceUrl, diffDomain) {
diff --git a/services/remove-expired-subscriptions.js b/services/remove-expired-subscriptions.js
@@ -1,5 +1,6 @@
 const mongodb = require('./mongodb');
 const getDayjs = require('./dayjs-wrapper');
+const jsonStore = require('./json-store');
 const config = require('../config');
 
 /**
@@ -49,13 +50,15 @@ async function removeExpiredSubscriptions() {
                 if (validSubscriptions.length === 0) {
                     // Remove entire document if no valid subscriptions remain
                     await collection.deleteOne({ _id: doc._id });
+                    jsonStore.setSubscriptions(doc._id, []);
                     documentsDeleted++;
                 } else {
                     // Update document with filtered subscriptions
                     await collection.updateOne(
                         { _id: doc._id },
                         { $set: { pleaseNotify: validSubscriptions } }
                     );
+                    jsonStore.setSubscriptions(doc._id, validSubscriptions);
                 }
             }
         }
diff --git a/test/mongodb.js b/test/mongodb.js
@@ -1,5 +1,6 @@
 const config = require('../config'),
     initSubscription = require('../services/init-subscription'),
+    jsonStore = require('../services/json-store'),
     mongodb = require('../services/mongodb');
 
 async function fetchSubscriptions(resourceUrl) {
@@ -69,5 +70,6 @@ module.exports = {
         await mongodb.get('rsscloud').collection('events').drop();
         await mongodb.get('rsscloud').collection('resources').drop();
         await mongodb.get('rsscloud').collection('subscriptions').drop();
+        jsonStore.clear();
     }
 };
