Merge branch 'fix-vault-styles' of https://github.com/naturalcrit/homebrewery into fix-vault-styles

.gitignore

@@ -7,6 +7,7 @@ storage
 build/*
 config/local.*
 config/docker.*
+config/dev.*
 
 todo.md
 startDB.bat

client/homebrew/pages/sharePage/sharePage.jsx

@@ -53,21 +53,25 @@ const SharePage = (props)=>{
 		document.addEventListener('keydown', handleControlKeys);
 		fetchThemeBundle(undefined, setThemeBundle, currentBrew.renderer, currentBrew.theme);
 
-		// listen for changes in the brew version
-		const eventSource = new EventSource('/stream');
-		eventSource.addEventListener('message', (evt)=>{
-			const messageData = JSON.parse(evt.data);
+		let eventSource;
+		// Start a stream if the brew has been updated in the last month
+		if(Date.parse(props.brew?.updatedAt) > (new Date).setDate((new Date).getDate() - 30)){
+			// listen for changes in the brew version
+			eventSource = new EventSource(`/api/stream/${props.brew.shareId}`);
 
-			if(messageData.eventType == 'brewUpdated'){
-				if(messageData.shareId == currentBrew.shareId && messageData.version != currentBrew.version) {
+			eventSource.addEventListener('brewUpdated', (evt)=>{
+				const messageData = JSON.parse(evt.data);
+
+				if(messageData.version != currentBrew.version) {
 					console.log('should fetch brew');
 					fetchUpdatedBrew();
 				}
-			}
-		});
+			});
+		}
 
 		return ()=>{
 			document.removeEventListener('keydown', handleControlKeys);
+			eventSource?.close();
 		};
 	}, []);
 

server/app.js

@@ -20,6 +20,7 @@ import api from './homebrew.api.js';
 const { homebrewApi, getBrew, getCSS } = api;
 import adminApi                    from './admin.api.js';
 import vaultApi                    from './vault.api.js';
+import streamApi                   from './stream.api.js';
 import pageRoutes from './page-routes.js';
 
 import serveCompressedStaticAssets from './static-assets.mv.js';
@@ -32,7 +33,6 @@ import bodyParser         from 'body-parser';
 import cookieParser       from 'cookie-parser';
 import forceSSL           from './forcessl.mw.js';
 
-import Stream from './eventStreamSource.js';
 import dbCheck            from './middleware/dbCheck.js';
 
 import cors from 'cors';
@@ -115,6 +115,7 @@ export default async function createApp(vite) {
 	app.use(homebrewApi);
 	app.use(adminApi(vite));
 	app.use(vaultApi);
+	app.use(streamApi);
 
 	String.prototype.replaceAll = function(s, r){return this.split(s).join(r);};
 
@@ -194,26 +195,6 @@ export default async function createApp(vite) {
 		}
 	});
 
-	// Create Event Stream source for pages to listen to
-	app.get('/stream', (req, res)=>{
-		res.writeHead(200, {
-			'Content-Type'     : 'text/event-stream',
-			'Cache-Control'    : 'no-cache',
-			'Connection'       : 'keep-alive',
-			'Content-Encoding' : 'none'
-		});
-
-		Stream.on('sendUpdate', (event, data)=>{
-			if(isLocalEnvironment) { console.log('Event:', event, '\nData:', data); }
-			res.write(`data: ${JSON.stringify({ ...data, eventType: event })}\n\n`);
-		});
-	});
-
-	// After Stream starts, send initStream event
-	setTimeout(()=>{
-		Stream.emit('sendUpdate', 'initStream', { time: new Date });
-	}, 1000);
-
 
 	// Local only
 	if(isLocalEnvironment){

server/config.js

@@ -4,4 +4,5 @@ export default nconf
 	.argv()
 	.env({ lowerCase: true })
 	.file('environment', { file: `config/${process.env.NODE_ENV}.json` })
+	.file('development', { file: 'config/dev.json' })
 	.file('defaults', { file: 'config/default.json' });

server/eventStreamSource.js

@@ -1,9 +1,94 @@
 import { EventEmitter } from 'events';
+import _ from 'lodash';
+import { nanoid } from 'nanoid';
+
+import config from './config.js';
+
+// DEBUG
+const DEBUG = {
+	showEvents          : config.get('showStreamEvents'),
+	showSubscriberCount : config.get('showStreamSubscriberCount')
+};
+
+// Delays
+const REPORT_DELAY = config.get('REPORT_DELAY') || 30 * 60 * 1000; //ms
+const HEARTBEAT_DELAY = config.get('HEARTBEAT_DELAY') || 20 * 1000; //ms
+
 
 const Stream = new EventEmitter;
 
+// Create array of stream subscribers
+let subscribers = [];
+
+// Listener functions
+const subscribe = function(shareId, res){
+	const id = nanoid(24); // NanoID is assumed globally unique;
+	const subscriber = {
+		id,
+		shareId,
+		stream : res
+	};
+
+	subscribers.push(subscriber);
+
+	Stream.emit('sendUpdate', 'subscribe', { id: subscriber.id, shareId: subscriber.shareId });
+
+	return subscriber.id;
+};
+
+const unsubscribe = function(id){
+	subscribers
+		.filter((sub)=>{return sub.id == id;})
+		.forEach((sub)=>{
+			// Unsubscribe event - not currently used
+			// Stream.emit('sendUpdate', 'unsubscribe', { id: sub.id, shareId: sub.shareId });
+			sub?.stream?.status(200).end();
+		});
+
+	subscribers = subscribers.filter((sub)=>{ return sub.id != id; });
+};
+
+// Create heartbeat
+const heartbeatId = config.get('node_env') != 'test' ?
+	setInterval(()=>{
+		if(DEBUG.showEvents){ console.log('STREAM HEARTBEAT @ ', new Date); };
+		subscribers
+			.forEach((sub)=>{
+				sub?.stream?.write(`:\n\n`);
+			});
+	}, HEARTBEAT_DELAY)
+	:
+	undefined;
+
+Stream.on('close', ()=>{
+	console.log('STREAM CLOSE');
+	clearInterval(heartbeatId);
+});
+
+// Create global sendUpdate listener
+Stream.on('sendUpdate', (event, data)=>{
+	if(DEBUG.showEvents) console.log('Event:', event, '\nData:', data);
+	subscribers
+		.filter((sub)=>{
+			if(data?.id) return data.id == sub.id;
+			return data.shareId == sub.shareId;
+		})
+		.forEach((sub)=>{
+			sub?.stream?.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
+		});
+});
+
+// DEBUG: Report subscriber count periodically
+if(DEBUG.showSubscriberCount){
+	const reportIntervalId = setInterval(()=>{
+		console.log('Subscriber count:', subscribers.length, ' @ ', new Date);
+	}, REPORT_DELAY);
+}
+
 export default {
 	emit : function(event) {return Stream.emit(event, ...([...arguments].slice(1)));},    // Arguments doesn't work for arrow functions
 	on   : (event, listener)=>{return Stream.on(event, listener);},
-	off  : (event, listener)=>{return Stream.off(event, listener);}
+	off  : (event, listener)=>{return Stream.off(event, listener);},
+	subscribe,
+	unsubscribe
 };

server/stream.api.js

@@ -0,0 +1,23 @@
+import express from 'express';
+
+import Stream from './eventStreamSource.js';
+
+const app = express.Router();
+
+// Create Event Stream source for pages to listen to
+app.get('/api/stream/:id', (req, res)=>{
+	res.writeHead(200, {
+		'Content-Type'     : 'text/event-stream',
+		'Cache-Control'    : 'no-cache',
+		'Connection'       : 'keep-alive',
+		'Content-Encoding' : 'none'
+	});
+
+	const subId = Stream.subscribe(req.params.id, res);
+
+	res.on('close', ()=>{
+		Stream.unsubscribe(subId);
+	});
+});
+
+export default app;