From c99602eb53a64afd62a97e86f8900a432ef2680e Mon Sep 17 00:00:00 2001 From: Ryjek Date: Thu, 2 Jul 2026 11:38:21 +0200 Subject: [PATCH] initail --- .dockerignore | 2 + .gitignore | 11 +- Dockerfile | 12 ++- app.js | 82 -------------- database/.gitkeep | 0 docker-compose.yml | 5 +- package.json | 19 ++++ public/index.html | 8 +- public/test.php | 7 +- src/app.js | 34 ++++++ src/appv2.js | 259 +++++++++++++++++++++++++++++++++++++++++++++ src/myDatabase.js | 64 +++++++++++ src/storage.js | 110 +++++++++++++++++++ storage/.gitkeep | 0 14 files changed, 520 insertions(+), 93 deletions(-) create mode 100644 .dockerignore delete mode 100644 app.js create mode 100644 database/.gitkeep create mode 100644 package.json create mode 100644 src/app.js create mode 100644 src/appv2.js create mode 100644 src/myDatabase.js create mode 100644 src/storage.js create mode 100644 storage/.gitkeep diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..25c8fdb --- /dev/null +++ b/.dockerignore @@ -0,0 +1,2 @@ +node_modules +package-lock.json \ No newline at end of file diff --git a/.gitignore b/.gitignore index 723ef36..5005b28 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,10 @@ -.idea \ No newline at end of file +.idea +node_modules +package-lock.json + + + +database +!database/.gitkeep +storage +!storage/.gitkeep \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index 7eb0a57..ccb37ec 100644 --- a/Dockerfile +++ b/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 -RUN npm install express +COPY ./src/* /srv/app +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 diff --git a/app.js b/app.js deleted file mode 100644 index fbf5314..0000000 --- a/app.js +++ /dev/null @@ -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('') - 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); diff --git a/database/.gitkeep b/database/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/docker-compose.yml b/docker-compose.yml index 45388ad..b889eff 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -12,8 +12,9 @@ services: ports: - 8201:8080 volumes: - - ./app.js:/srv/app.js - - ./public:/srv/public + - ./src:/srv/app + - ./storage:/storage + - ./database:/database php: image: docker.git.tlfactory.pl/apps/php-apache:7.4.20 ports: diff --git a/package.json b/package.json new file mode 100644 index 0000000..e52ef3a --- /dev/null +++ b/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" + } +} diff --git a/public/index.html b/public/index.html index 2edd437..3ca54dd 100644 --- a/public/index.html +++ b/public/index.html @@ -3,13 +3,17 @@ diff --git a/public/test.php b/public/test.php index 384016a..bb22d43 100644 --- a/public/test.php +++ b/public/test.php @@ -1,5 +1,6 @@ $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 "
$result
"; +//echo "
$result
"; +//8839716547:AAHNDG5ezuBxKHVaFCIivq2hWrNjFxDMHMs +// 234706670 \ No newline at end of file diff --git a/src/app.js b/src/app.js new file mode 100644 index 0000000..9d5e720 --- /dev/null +++ b/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('') + 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(); \ No newline at end of file diff --git a/src/appv2.js b/src/appv2.js new file mode 100644 index 0000000..bfbcbe7 --- /dev/null +++ b/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; \ No newline at end of file diff --git a/src/myDatabase.js b/src/myDatabase.js new file mode 100644 index 0000000..0fc1071 --- /dev/null +++ b/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; \ No newline at end of file diff --git a/src/storage.js b/src/storage.js new file mode 100644 index 0000000..1702166 --- /dev/null +++ b/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; \ No newline at end of file diff --git a/storage/.gitkeep b/storage/.gitkeep new file mode 100644 index 0000000..e69de29