pushstream-socketio/src/appv2.js

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 userId of userToDel) {
			const storage = new storageObj(userId);
			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;