initail

.dockerignore

@@ -0,0 +1,2 @@
+node_modules
+package-lock.json

.gitignore

@@ -1 +1,10 @@
-.idea
+.idea
+node_modules
+package-lock.json
+
+
+
+database
+!database/.gitkeep
+storage
+!storage/.gitkeep

Dockerfile

@@ -1,13 +1,17 @@
 FROM node:22-alpine
 
 RUN mkdir /srv/public/
+Run mkdir /srv/app
+RUN mkdir /storage
+RUN mkdir /database
 
 WORKDIR /srv/
 
-RUN npm install  socket.io
+COPY ./src/* /srv/app
-RUN npm install  express
+COPY ./package.json /srv/package.json
+
+RUN npm install
 
-COPY ./app.js /srv/app.js
 COPY ./public/index.html /srv/public/index.html
 
-CMD node /srv/app.js
+CMD node app/app.js

app.js

@@ -1,82 +0,0 @@
-const PORT = process.env.PORT || 8080;
-const PREFIX = process.env.PREFIX || '/ws';
-const DEV = process.env.DEV || false;
-console.log('Socket.io listen on: http://localhost:' + PORT + '' + PREFIX + '/socket.io');
-console.log('Publisher listen on: http://localhost:' + PORT + '' + PREFIX + '/pub/$roomID');
-if (DEV !== false) {
-    console.log('System is on DEV mode');
-}
-
-const app = require('express')();
-const server = require('http').createServer(app);
-var io;
-
-if (DEV !== false) {
-    io = require('socket.io')(server, {path: PREFIX + '/socket.io', cors: {origin: '*',}});
-} else {
-    io = require('socket.io')(server, {path: PREFIX + '/socket.io'});
-}
-
-const bodyParser = require('body-parser');
-
-
-var jsonParser = bodyParser.json();
-const queue = {};
-
-app.get('/test', function (req, res) {
-  //res.send('<script src="/socket.io/socket.io.js"></script><script>  const socket = io("http://localhost");</script>')
-  res.sendFile(__dirname + '/public/index.html')
-});
-
-app.post('/pub/:roomId', jsonParser, function (req,res) {
-	const room = req.params.roomId;
-    console.log('publish to ' + room + ' data:');
-    console.log(req.body);
-	if (typeof queue[room] !== 'undefined') {
-		queue[room].push(req.body);
-	} else {
-		io.to(room).emit('data',req.body);
-	}
-	console.log(queue);
-    res.sendStatus(200);
-});
-
-function takeFromQueue(room) {
-	if (typeof queue[room] == 'undefined') {
-		return;
-	}
-	const d = queue[room].shift();
-	if (typeof d === 'undefined') {
-		delete queue[room];
-		console.log("Queue delivered. Remove obj.")
-		return;
-	}
-	io.to(room).emit('data',d);
-
-	setTimeout(()=>{
-		takeFromQueue(room);
-	}, 500);
-}
-
-io.on('connection',  function(socket) { 
-    console.log('new user ' + socket.id);
-	console.log("_________________________")
-    if (typeof socket.handshake.query.id !== 'undefined') {
-		const room = socket.handshake.query.id;
-		socket.join(room);
-		console.log('user ' + socket.id + ' joined to room ' + room);
-		takeFromQueue(room);
-	}
-    socket.on('disconnect', () => {
-		const room = socket.handshake.query.id;
-		console.log('user ' + socket.id + ' disconnected');
-		if (typeof queue[room] === "undefined") {
-			queue[room]=[];
-	    }
-		console.log('create queue: ' + room);
-    });
-});
-
-
-
-server.listen(PORT);

docker-compose.yml

@@ -12,8 +12,9 @@ services:
     ports:
       - 8201:8080
     volumes:
-      - ./app.js:/srv/app.js
+      - ./src:/srv/app
-      - ./public:/srv/public
+      - ./storage:/storage
+      - ./database:/database
   php:
     image: docker.git.tlfactory.pl/apps/php-apache:7.4.20
     ports:

package.json

@@ -0,0 +1,19 @@
+{
+  "name": "push-stream-channels",
+  "version": "1.0.0",
+  "description": "",
+  "main": "client.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "repository": {
+    "type": "git",
+    "url": "git@git.tlfactory.pl:apps/push-stream-channels.git"
+  },
+  "private": true,
+  "dependencies": {
+    "better-sqlite3": "^12.11.1",
+    "express": "^5.2.1",
+    "socket.io": "^4.8.3"
+  }
+}

public/index.html

@@ -3,13 +3,17 @@
 <script>
     const socket = io("http://localhost:8201", {
     path: '/ws/socket.io',
-    query: 'id=channel'
+    query: 'id=1000_channel'
     });
-    socket.on('data', (arg) => {
+    socket.on('data', (arg, callback) => {
 		const doc = document.getElementById('test');
 		const div = document.createElement('div');
 		div.innerHTML = JSON.stringify(arg);
 		doc.append(div);
+
+		callback({
+			received: true
+        });
     });
 </script>
 

public/test.php

@@ -1,5 +1,6 @@
 <?php
-$ch = curl_init('http://push:8080/pub/channel');
+$uid = mt_rand(100,400000);
+$ch = curl_init('http://push:8080/pub/'. $uid .'_channel');
 # Setup request to send json via POST.
 $payload = json_encode( array( "customer"=> $argv[1]) );
 curl_setopt( $ch, CURLOPT_POSTFIELDS, $payload );
@@ -11,4 +12,6 @@ $result = curl_exec($ch);
 var_export($result);
 curl_close($ch);
 # Print response.
-echo "<pre>$result</pre>";
+//echo "<pre>$result</pre>";
+//8839716547:AAHNDG5ezuBxKHVaFCIivq2hWrNjFxDMHMs
+// 234706670

src/app.js

@@ -0,0 +1,34 @@
+const PORT = process.env.PORT || 8080;
+const PREFIX = process.env.PREFIX || '/ws';
+const DEV = process.env.DEV || false;
+const appObj = require("./appv2");
+
+const bodyParser = require('body-parser');
+const jsonParser = bodyParser.json();
+
+console.log('Socket.io listen on: http://localhost:' + PORT + '' + PREFIX + '/socket.io');
+console.log('Publisher listen on: http://localhost:' + PORT + '' + PREFIX + '/pub/$roomID');
+if (DEV !== false) {
+    console.log('System is on DEV mode');
+}
+
+const appv2 = new appObj(PORT, PREFIX, DEV);
+appv2.init();
+appv2.app.get('/test', function (req, res) {
+  //res.send('<script src="/socket.io/socket.io.js"></script><script>  const socket = io("http://localhost");</script>')
+  res.sendFile(__dirname + '/public/index.html')
+});
+
+appv2.app.post('/pub/:roomId', jsonParser, (req,res) => appv2.onPostPub(req,res));
+appv2.io.on('connection', (socket) => appv2.onClientConnect(socket));
+appv2.server.listen(PORT);
+
+
+setInterval(() => {
+    // appv2.clearQueue();
+    // appv2.clearQueueDb();
+},10000);
+
+appv2.clearQueueDb();
+
+// appv2.clearQueueFiles();

src/appv2.js

@@ -0,0 +1,259 @@
+const storageObj = require("./storage");
+const socket = require('socket.io');
+const http = require('http');
+const express = require('express');
+const fsp = require('fs/promises');
+const myDb = require("./myDatabase");
+
+class Appv2 {
+	io;
+	app;
+	server;
+	connections = {};
+	queue = {};
+	queueSettings = {
+		0: {
+			createdAt: new Date(),
+		}
+	};
+
+	constructor(port,prefix,dev) {
+		this.port = port;
+		this.prefix = prefix;
+		this.dev = dev;
+	}
+
+	init() {
+		this._takeIO();
+	}
+	get server() {
+		return this.server;
+	}
+	get app() {
+		return this.app;
+	}
+
+	get io() {
+		return this.io;
+	}
+
+	get timeToClearQueue() {
+		return 10;
+	}
+
+	_takeIO() {
+		this.app = express();
+		this.server = http.createServer(this.app);
+		if (this.dev !== false) {
+			this.io = socket(this.server, {path: this.prefix + '/socket.io', cors: {origin: '*',}});
+		} else {
+			this.io = socket(this.server, {path: this.prefix + '/socket.io'});
+		}
+	}
+
+	postResponse(res, code) {
+		if(typeof res === 'undefined') {
+			return;
+		}
+		res.sendStatus(code);
+	}
+
+	onPostPub(req,res) {
+		const room = req.params.roomId;
+		const th = this;
+		console.log('publish to ' + room + ' data:');
+		console.log(req.body);
+		const userId = room.split("_")[0];
+
+		if (typeof this.connections[userId] === 'undefined') {
+			th.addToQueue(userId, req.body);
+			this.postResponse(res, 404);
+			return;
+		}
+		this.connections[userId].timeout(100).emit('data', req.body, (err, response)=> {
+			let success = true;
+			console.log(response);
+			if(typeof response !== 'undefined') {
+				if (!response['received']) {
+					success = false;
+				}
+			}
+			if (err){
+				success = false;
+			}
+			if(success){
+				th.postResponse(res,200);
+			} else {
+				th.addToQueue(userId, req.body);
+				th.postResponse(res, 404);
+			}
+		});
+	}
+
+	onClientConnect(socket) {
+		console.log('new user ' + socket.id);
+		console.log("_________________________")
+		if (typeof socket.handshake.query.id !== 'undefined') {
+			const room = socket.handshake.query.id;
+			const userId = room.split("_")[0];
+			this.connections[userId] = socket;
+			this.tryResolveQueueStorage(userId, room).then(r => console.log(r));
+			console.log('user ' + socket.id + ' joined to room ' + room);
+		}
+		socket.on('disconnect', () => {
+			const room = socket.handshake.query.id;
+			console.log('user ' + socket.id + ' disconnected');
+		});
+	}
+
+	async addToQueue(userId, data) {
+		const storage = new storageObj(userId);
+		await storage.addQueue(data);
+
+
+
+		// if (typeof this.queue[userId] === 'undefined') {
+		// 	this.queue[userId] = [];
+		// }
+		// if (typeof this.queueSettings[userId] === 'undefined') {
+		// 	this.queueSettings[userId] = {
+		// 		createdAt: Math.round(Date.now() / 1000),
+		// 		lastMessageAt: Math.round(Date.now() / 1000),
+		// 	};
+		// } else {
+		// 	this.queueSettings[userId].lastMessageAt = Math.round(Date.now() / 1000);
+		// }
+		// this.queue[userId].push(data);
+	}
+
+	async tryResolveQueueStorage(userId, room) {
+		const storage = new storageObj(userId);
+		for (const line of await storage.readQueueAll()) {
+			const tmp = {
+				params:{
+					roomId: room
+				},
+				body: line
+			};
+			this.onPostPub(tmp);
+		}
+		await storage.removeQueue();
+		return "success deliver query";
+	}
+
+	tryResolveQueue(userId, room) {
+		if(typeof this.queue[userId] === 'undefined') {
+			return;
+		}
+		const data = this.queue[userId].shift();
+		const th = this;
+
+		if(typeof data === 'undefined') {
+			delete this.queue[userId];
+			return;
+		}
+		const tmp = {
+			params:{
+				roomId: room
+			},
+			body: data
+		};
+		this.onPostPub(tmp);
+		setTimeout(()=>{
+			th.tryResolveQueue(userId, room);
+		},1000);
+	}
+
+	async clearQueue() {
+		const nowTime = Math.round(Date.now() / 1000);
+
+		for (const userId in this.queue) {
+			const settings = this.queueSettings[userId];
+			if (typeof settings === 'undefined') {
+				console.log("Empty settings for user " + userId);
+				continue;
+			}
+			if (nowTime - settings.createdAt >= this.timeToClearQueue) {
+				delete this.queue[userId];
+				delete this.queueSettings[userId];
+
+				const storage = new storageObj(userId);
+				storage.removeQueue();
+				console.log("Delete queue for user " + userId);
+			}
+		}
+	}
+
+	async clearQueueFiles() {
+		const startTimeToDebug = Date.now();
+		const baseDir = "/storage";
+		const nowTime = Math.round(Date.now() / 1000);
+		const items = await fsp.readdir(baseDir, { withFileTypes: true });
+		const userIds = [];
+		for (const item of items) {
+			if (item.isDirectory()) {
+				const dir = baseDir + "/" + item.name
+				const users = await fsp.readdir(dir, { withFileTypes: true });
+				for (const user of users) {
+					const userId = user.name.split(".")[0];
+					if (userId.indexOf("config") === -1) {
+						userIds.push(userId);
+					}
+				}
+			}
+		}
+		const afterReadDirectories = Date.now();
+		console.log((afterReadDirectories - startTimeToDebug) / 1000 + " seconds");
+		for (const uid of userIds) {
+			const storage = new storageObj(uid);
+			const config = await storage.readConfig();
+			if (!config || nowTime - config.createdAt >= this.timeToClearQueue) {
+				await storage.removeQueue();
+			}
+		}
+		console.log((Date.now() - afterReadDirectories) / 1000 + " seconds");
+	}
+
+	async clearQueueDb() {
+		const startTimeToDebug = Date.now();
+		const db = new myDb();
+		const users = db.takeAllToClear();
+		if (!users) {
+			return;
+		}
+		let count = 0;
+		const userToDel = [];
+		const currentTime = Math.round(Date.now() / 1000);
+
+		for (const user of users) {
+			const date = new Date(user.created_at.replace(" ", "T"));
+			const time = Math.round( date.getTime()/ 1000);
+
+			if (currentTime - time > this.timeToClearQueue) {
+				userToDel.push(user.id);
+			}
+		}
+		const afterFetchFromDb = Date.now();
+		console.log("Fetch from db: "+(Date.now() - startTimeToDebug) / 1000 + " seconds");
+
+		for (const user of users) {
+			const storage = new storageObj(user.id);
+			await storage.removeQueue();
+			count++;
+		}
+
+		const afterall = Date.now();
+		console.log("Files delete: " + (afterall - afterFetchFromDb) / 1000 + " seconds");
+		console.log("Success in: " + (afterall - startTimeToDebug) / 1000 + " seconds");
+
+		console.log("Success clear queue: " + count);
+
+
+
+		// push-1  | 9.759 seconds
+		// push-1  | Success clear queue: 5369
+	}
+}
+
+
+module.exports = Appv2;

src/myDatabase.js

@@ -0,0 +1,64 @@
+const Database = require("better-sqlite3");
+
+class MyDatabase {
+	dbObj;
+	constructor() {
+		this.dbObj = new Database("/database/users.db");
+		this.dbObj.exec(`
+			CREATE TABLE IF NOT EXISTS users (
+				id INTEGER PRIMARY KEY,
+				created_at DATETIME DEFAULT (datetime('now')),
+				resolve_at DATETIME DEFAULT null,
+				deleted INTEGER DEFAULT 0
+			);
+		`);
+	}
+
+	createUser(userId) {
+		const user = this.getUser(userId);
+		const now = new Date().toISOString();
+
+		if (!user) {
+			console.log("No user with id " + userId);
+			this.dbObj.prepare("INSERT INTO users (id) VALUES (?)").run(userId);
+		} else {
+			if (user.deleted === 1) {
+				console.log("User with id deleted " + userId);
+				this.dbObj.prepare("UPDATE users SET created_at = datetime('now'), deleted = 0 WHERE id = ?").run(userId);
+			}
+		}
+	}
+
+	getUser(userId) {
+		const user = this.dbObj.prepare("SELECT * FROM users WHERE id = ?").get(userId);
+		if (typeof user === "undefined") {
+			return false;
+		}
+		return user;
+	}
+	deleteUser(userId) {
+		// this.dbObj.prepare("DELETE FROM users WHERE id = ?").run(userId);
+		this.dbObj.prepare("UPDATE users SET deleted=1, resolve_at=datetime('now') WHERE id = ?").run(userId);
+	}
+
+	takeAllToClear() {
+		const all = this.dbObj.prepare("SELECT * FROM users WHERE deleted = 0").all();
+		if (typeof all === "undefined" || all.length < 1) {
+			return false;
+		}
+		return all;
+	}
+
+
+	clearDeleted() {
+		this.dbObj.prepare("DELETE FROM users WHERE deleted = 1");
+	}
+	countAll() {
+		return this.dbObj.prepare("SELECT COUNT(*) as count FROM users").get();
+	}
+
+
+
+}
+
+module.exports = MyDatabase;

src/storage.js

@@ -0,0 +1,110 @@
+const fs = require('fs');
+const fsp = require('fs/promises');
+const readline = require('readline');
+const readLine = require("node:readline");
+const myDb = require('./myDatabase');
+class Storage {
+	userId;
+	dbObj;
+
+	constructor(userId) {
+		this.dbObj = new myDb();
+		this.userId = userId;
+	}
+
+	get getDir() {
+		return "/storage/" + Math.floor(this.userId / 100);
+	}
+	get fileName() {
+		return this.getDir + "/" + this.userId + ".jsonl";
+	}
+	get fileNameConfig() {
+		return this.getDir + "/" + this.userId + "-config.jsonl";
+	}
+
+	async fileExists() {
+		let isExist = false;
+		try {
+			await fsp.access(this.fileName);
+			isExist = true;
+		} catch (_) {
+		}
+
+		return isExist;
+	}
+
+	async addQueue(data) {
+		let toFile = "";
+		if (!await this.fileExists()) {
+			await fsp.mkdir(this.getDir, { recursive: true });
+			// await fsp.appendFile(this.fileNameConfig, JSON.stringify({
+			// 	createdAt: Math.floor(Date.now() / 1000),
+			// }), 'utf8');
+			this.dbObj.createUser(this.userId);
+		}
+
+		toFile += JSON.stringify(data);
+		await fsp.appendFile(this.fileName, toFile + "\n", 'utf8');
+	}
+
+	async removeQueue() {
+		try	{
+			await fsp.unlink(this.fileName);
+		} catch(err) {
+			console.log(err);
+		}
+		this.dbObj.deleteUser(this.userId);
+	}
+
+	readQueueLine(callback) {
+		const readFile = readline.createInterface({
+			input: fs.createReadStream(this.fileName)
+		});
+
+		// for await (const line of readFile) {
+		// 	yield line;
+		// }
+		if (typeof callback !== 'function') {
+			return;
+		}
+
+		readFile.on('line', (line) => {
+			const obj = JSON.parse(line);
+			callback(obj);
+		});
+	}
+	async readQueueAll() {
+		const all = await fsp.readFile(this.fileName,"utf-8");
+		return all.split("\n")
+			.filter(line => line.trim())
+			.map(line => JSON.parse(line));
+	}
+
+	async readConfig() {
+		const readFile = readLine.createInterface({
+			input: fs.createReadStream(this.fileNameConfig),
+		});
+		for await (const line of readFile) {
+			readFile.close();
+			return JSON.parse(line);
+		}
+
+		return null;
+	}
+
+	async readOneLine() {
+		const readFile = readLine.createInterface({
+			input: fs.createReadStream(this.fileName),
+		});
+
+		for await (const line of readFile) {
+			readFile.close();
+			return JSON.parse(line);
+		}
+
+		return null;
+	}
+
+}
+
+module.exports = Storage;