0.4.6-beta.0

index.js

@@ -8,14 +8,13 @@ const banner = require('./scripts/banners/ufw.js');
 const { axiosService } = require('./scripts/services/axios.js');
 const { saveBufferToFile, loadBufferFromFile, sendBulkReport, BULK_REPORT_BUFFER } = require('./scripts/services/bulk.js');
 const { reportedIPs, loadReportedIPs, saveReportedIPs, isIPReportedRecently, markIPAsReported } = require('./scripts/services/cache.js');
+const ABUSE_STATE = require('./scripts/services/state.js');
 const { refreshServerIPs, getServerIPs } = require('./scripts/services/ipFetcher.js');
 const { repoSlug, repoUrl } = require('./scripts/repo.js');
 const isSpecialPurposeIP = require('./scripts/isSpecialPurposeIP.js');
 const logger = require('./scripts/logger.js');
 const config = require('./config.js');
 const { UFW_LOG_FILE, SERVER_ID, EXTENDED_LOGS, AUTO_UPDATE_ENABLED, AUTO_UPDATE_SCHEDULE, DISCORD_WEBHOOK_ENABLED, DISCORD_WEBHOOK_URL } = config.MAIN;
-
-const ABUSE_STATE = { isLimited: false, isBuffering: false, sentBulk: false };
 const RATE_LIMIT_LOG_INTERVAL = 10 * 60 * 1000;
 const BUFFER_STATS_INTERVAL = 5 * 60 * 1000;
 
@@ -178,19 +177,43 @@ const processLogLine = async (line, test = false) => {
 	}
 
 	// Watch
+	let incompleteLine = '';
+	let processing = Promise.resolve();
+
 	fileOffset = fs.statSync(UFW_LOG_FILE).size;
 	chokidar.watch(UFW_LOG_FILE, { persistent: true, ignoreInitial: true })
-		.on('change', path => {
-			const stats = fs.statSync(path);
+		.on('change', filePath => {
+			const stats = fs.statSync(filePath);
 			if (stats.size < fileOffset) {
+				incompleteLine = '';
 				fileOffset = 0;
 				logger.info('The file has been truncated, and the offset has been reset');
 			}
 
-			fs.createReadStream(path, { start: fileOffset, encoding: 'utf8' }).on('data', chunk => {
-				chunk.split('\n').filter(line => line.trim()).forEach(processLogLine);
-			}).on('end', () => {
-				fileOffset = stats.size;
+			const start = fileOffset;
+			fileOffset = stats.size;
+
+			processing = processing.then(async () => {
+				let data = '';
+				try {
+					await new Promise((resolve, reject) => {
+						fs.createReadStream(filePath, { start, encoding: 'utf8' })
+							.on('data', chunk => { data += chunk; })
+							.on('end', resolve)
+							.on('error', reject);
+					});
+				} catch (err) {
+					logger.error(`Failed to read log chunk: ${err.message}`);
+					return;
+				}
+
+				const text = incompleteLine + data;
+				const lines = text.split('\n');
+				incompleteLine = lines.pop();
+
+				for (const line of lines) {
+					if (line.trim()) await processLogLine(line);
+				}
 			});
 		});
 
@@ -203,4 +226,18 @@ const processLogLine = async (line, test = false) => {
 	process.send?.('ready');
 })();
 
+const gracefulShutdown = async signal => {
+	logger.info(`Received ${signal}, flushing pending writes...`);
+	try {
+		await saveBufferToFile();
+		await saveReportedIPs();
+	} catch (err) {
+		logger.error(`Error during shutdown flush: ${err.message}`);
+	}
+	process.exit(0);
+};
+
+process.on('SIGINT', () => gracefulShutdown('SIGINT'));
+process.on('SIGTERM', () => gracefulShutdown('SIGTERM'));
+
 module.exports = processLogLine;