fix(access-logs): harden ClickHouse pipeline, parse handshake errors, i18n

- Fix MV regex escaping, replaceAll for dates, explicit UTC end-to-end
- Batch dedup via insert token + dependent MV dedup; drop dead offset column
- Versioned MV (v3) with auto-drop of legacy views on ensureSchema
- Parse connection-error lines (from IP rejected <msg>) as rejected with source IP + outbound_tag=handshake-error; exclude empty dest from top lists
- Cache resolved CH config; remove leftover DuckDB/Parquet code and locale keys
- Add common.attention / common.restore locale keys (all languages)

Author: ddddeadispose

scripts/test-access-logs-clickhouse.js

@@ -74,6 +74,21 @@ async function offlineTests() {
     assert.ok(mvDdl.includes('replaceAll(ts_str'), 'date slashes replaced with replaceAll');
     // Unparsed lines must not land in 1970 (instantly TTL-dropped).
     assert.ok(mvDdl.includes("now('UTC')"), 'zero timestamps fall back to now()');
+    // Connection-level error lines are parsed via the fallback regex and tagged.
+    assert.ok(mvDdl.includes('handshake-error'), 'handshake-error fallback tag present');
+    assert.ok(mvDdl.includes('ne > 0'), 'error-line group participates in parse_ok');
+
+    // The fallback regex matches an error line but NOT as an access record.
+    const errLine = '2026/07/10 17:41:28.205208 from 95.24.24.226:9048 rejected proxy/vless/encoding: invalid request user id: 55837f55-c7ee-4533-b2a6-0ace8a266802';
+    assert.strictEqual(parseLikeMv(errLine).parse_ok, 0, 'error line is not a normal access record');
+    const errRe = new RegExp(clickhouse.CH_ERR_RE);
+    const em = errRe.exec(errLine);
+    assert.ok(em, 'error line matches fallback regex');
+    assert.strictEqual(em[2], '95.24.24.226:9048', 'fallback captures source');
+    assert.strictEqual(em[3], 'rejected', 'fallback captures action');
+    // A real access line must still be handled by the primary parser, not misrouted.
+    assert.ok(errRe.exec('2023/11/22 17:01:32 1.2.3.4:1122 accepted tcp:example.com:443 [in -> direct]'),
+        'fallback also matches normal lines (primary takes precedence in the MV)');
 
     // Retention clamps to sane bounds (0/NaN falls back to the 30-day default).
     assert.ok(clickhouse.schemaStatements(-5)[1].includes('INTERVAL 1 DAY'), 'retention floor');
@@ -152,6 +167,7 @@ async function onlineTests() {
     const batchId = 'test-batch-' + Date.now();
     await clickhouse.insertRaw([
         { node_id: 'n1', raw: '2023/11/22 17:01:32 1.2.3.4:1122 accepted tcp:example.com:443 [vless-in -> direct] email: 42' },
+        { node_id: 'n1', raw: '2023/11/22 17:01:33 from 9.9.9.9:5000 rejected proxy/vless/encoding: invalid request user id: abc' },
     ], batchId);
 
     // Give the MV a moment (insert is synchronous, but read is eventually there).
@@ -161,6 +177,14 @@ async function onlineTests() {
     assert.strictEqual(res.rows[0].network, 'tcp');
     assert.strictEqual(res.rows[0].dest_host, 'example.com');
 
+    // The connection-error line parsed via fallback: rejected, source set, tagged.
+    const errRes = await clickhouse.query(
+        "SELECT source_ip, action, outbound_tag, parse_ok FROM access_events WHERE source_ip = '9.9.9.9' LIMIT 1");
+    assert.ok(errRes.ok && errRes.rows.length >= 1, 'error line stored');
+    assert.strictEqual(errRes.rows[0].action, 'rejected', 'error line action');
+    assert.strictEqual(errRes.rows[0].outbound_tag, 'handshake-error', 'error line tagged');
+    assert.strictEqual(Number(errRes.rows[0].parse_ok), 1, 'error line counts as parsed');
+
     await clickhouse.truncate();
     void orig;
     console.log('  online: end-to-end OK');

src/locales/en.json

@@ -50,7 +50,9 @@
     "server": "Server",
     "ports": "Ports",
     "clear": "Clear",
-    "reset": "Reset"
+    "reset": "Reset",
+    "attention": "Attention",
+    "restore": "Restore"
   },
   "nav": {
     "dashboard": "Dashboard",

src/locales/ru.json

@@ -50,7 +50,9 @@
     "server": "Сервер",
     "ports": "Порты",
     "clear": "Очистить",
-    "reset": "Сбросить"
+    "reset": "Сбросить",
+    "attention": "Внимание",
+    "restore": "Восстановить"
   },
   "nav": {
     "dashboard": "Главная",

src/locales/zh-CN.json

@@ -50,7 +50,9 @@
     "server": "服务器",
     "ports": "端口",
     "clear": "清空",
-    "reset": "重置"
+    "reset": "重置",
+    "attention": "注意",
+    "restore": "恢复"
   },
   "nav": {
     "dashboard": "仪表盘",

src/services/accessLogs/clickhouseService.js

@@ -51,6 +51,17 @@ const CH_LINE_RE =
     '(?:\\s+email:\\s*(\\S+))?' +
     '\\s*$';
 
+// Fallback for connection-level error lines that are NOT access records, e.g.:
+//   2026/07/10 17:41:28 from 95.24.24.226:9048 rejected proxy/vless/encoding: invalid request user id: <uuid>
+// These carry a timestamp, source and action but no "tcp:/udp:" destination
+// (an attacker/scanner hitting an inbound with a bad UUID). Captures:
+//   1 ts, 2 src, 3 action. The tail (error text) stays in `raw`.
+const CH_ERR_RE =
+    '^(\\d{4}/\\d{2}/\\d{2} \\d{2}:\\d{2}:\\d{2}(?:\\.\\d+)?)\\s+' +
+    '(?:from\\s+)?' +
+    '(\\S+?)\\s+' +
+    '(accepted|rejected|blocked)(?:\\s|$)';
+
 // Escape a JS string for use inside a single-quoted ClickHouse SQL literal.
 // ClickHouse collapses unknown escapes ('\d' -> 'd'), so every backslash must
 // be doubled or the inlined regex silently loses all its character classes.
@@ -63,9 +74,9 @@ function sqlString(s) {
 // the stale one). The version is part of the MV name; ensureSchema drops any
 // older names listed here. Bump MV_VERSION whenever the MV definition changes
 // and append the previous name to LEGACY_MV_NAMES.
-const MV_VERSION = 2;
+const MV_VERSION = 3;
 const MV_NAME = `access_events_mv_v${MV_VERSION}`;
-const LEGACY_MV_NAMES = ['access_events_mv'];
+const LEGACY_MV_NAMES = ['access_events_mv', 'access_events_mv_v2'];
 
 // ── Config ────────────────────────────────────────────────────────────────
 
@@ -248,16 +259,21 @@ function schemaStatements(retentionDays) {
         SETTINGS non_replicated_deduplication_window = 1000`,
 
         // Parse raw -> structured on insert. Everything derives from `raw`.
-        // A line that does not match the regex (or carries a broken timestamp)
-        // still lands with parse_ok = 0 and event_time = now(), so no data is
-        // lost and it stays searchable by raw text within the retention window.
+        // Two shapes are recognised: (n) a normal access line and, as a fallback,
+        // (ne) a connection-level error line ("from IP rejected <msg>", no
+        // destination) which is tagged outbound_tag = 'handshake-error' so the
+        // action counters stay accurate and the attacking source IP is visible.
+        // A line matching neither still lands with parse_ok = 0 and
+        // event_time = now(), so nothing is lost and it stays searchable by raw.
         `CREATE MATERIALIZED VIEW IF NOT EXISTS ${MV_NAME} TO access_events AS
         WITH
             extractGroups(raw, '${sqlString(CH_LINE_RE)}') AS g,
             length(g) AS n,
-            if(n > 0, g[1], '') AS ts_str,
-            if(n > 0, g[2], '') AS src,
-            if(n > 0, g[3], '') AS act,
+            extractGroups(raw, '${sqlString(CH_ERR_RE)}') AS ge,
+            length(ge) AS ne,
+            if(n > 0, g[1], if(ne > 0, ge[1], '')) AS ts_str,
+            if(n > 0, g[2], if(ne > 0, ge[2], '')) AS src,
+            if(n > 0, g[3], if(ne > 0, ge[3], '')) AS act,
             if(n > 0, g[4], '') AS net,
             if(n > 0, g[5], '') AS dst,
             if(n > 0, g[6], '') AS route,
@@ -283,10 +299,10 @@ function schemaStatements(retentionDays) {
             dst_port AS dest_port,
             net AS network,
             in_tag AS inbound_tag,
-            out_tag AS outbound_tag,
+            if(n > 0, out_tag, if(ne > 0, 'handshake-error', '')) AS outbound_tag,
             act AS action,
             raw,
-            toUInt8(n > 0) AS parse_ok
+            toUInt8(n > 0 OR ne > 0) AS parse_ok
         FROM access_ingest`,
     ];
 }
@@ -392,6 +408,7 @@ async function truncate() {
 
 module.exports = {
     CH_LINE_RE,
+    CH_ERR_RE,
     readConfig,
     getClient,
     reset,

src/services/accessLogs/searchService.js

@@ -186,7 +186,7 @@ async function overview(filters = {}, opts = {}) {
 
     const topDestSql = `
         SELECT ${DEST} AS dest, count() AS hits
-        FROM access_events ${where}
+        FROM access_events ${where} ${andWhere} ${DEST} != ''
         GROUP BY dest ORDER BY hits DESC LIMIT ${topN}`;
 
     const topPortsSql = `
@@ -196,7 +196,7 @@ async function overview(filters = {}, opts = {}) {
 
     const topBlockedSql = `
         SELECT ${DEST} AS dest, count() AS hits
-        FROM access_events ${where} ${andWhere} action IN ('blocked','rejected')
+        FROM access_events ${where} ${andWhere} action IN ('blocked','rejected') AND ${DEST} != ''
         GROUP BY dest ORDER BY hits DESC LIMIT ${topN}`;
 
     const usersSql = `