feat: add Redis TimeSeries helper with safe increment and auto-creation

pavelzotikov · pavelzotikov · commit dafb4cb46ff1 · 2025-11-05T12:00:56.000+03:00
diff --git a/package.json b/package.json
@@ -81,6 +81,7 @@
     "mongodb": "^3.7.3",
     "morgan": "^1.10.1",
     "prom-client": "^15.1.3",
+    "redis": "^4.7.0",
     "safe-regex": "^2.1.0",
     "ts-node-dev": "^2.0.0",
     "uuid": "^8.3.2"
diff --git a/src/models/eventsFactory.js b/src/models/eventsFactory.js
@@ -1,11 +1,13 @@
 import { getMidnightWithTimezoneOffset, getUTCMidnight } from '../utils/dates';
 import safe from 'safe-regex';
 import { createProjectEventsByIdLoader } from '../dataLoaders';
+import { Effect, sgr } from '../utils/ansi';
 
 const Factory = require('./modelFactory');
 const mongo = require('../mongo');
 const Event = require('../models/event');
 const { ObjectID } = require('mongodb');
+import RedisHelper from '../redisHelper';
 const { composeEventPayloadByRepetition } = require('../utils/merge');
 
 const MAX_DB_READ_BATCH_SIZE = Number(process.env.MAX_DB_READ_BATCH_SIZE);
@@ -69,6 +71,12 @@ const MAX_DB_READ_BATCH_SIZE = Number(process.env.MAX_DB_READ_BATCH_SIZE);
  * Factory Class for Event's Model
  */
 class EventsFactory extends Factory {
+  /**
+  /**
+   * Redis helper instance for modifying data through redis
+   */
+  redis = new RedisHelper();
+
   /**
    * Event types with collections where they stored
    * @return {{EVENTS: string, DAILY_EVENTS: string, REPETITIONS: string, RELEASES: string}}
@@ -94,6 +102,8 @@ class EventsFactory extends Factory {
       throw new Error('Can not construct Event model, because projectId is not provided');
     }
 
+    this.redis.initialize();
+
     this.projectId = projectId;
     this.eventsDataLoader = createProjectEventsByIdLoader(mongo.databases.events, this.projectId);
   }
@@ -392,6 +402,20 @@ class EventsFactory extends Factory {
     };
   }
 
+  async getChartData(hours = 24, timezoneOffset = 0, projectId = '', groupHash = '') {
+    try {
+      const redisData = await this.redis.getChartDataFromRedis(hours, timezoneOffset, projectId, groupHash);
+
+      if (redisData && redisData.length > 0) {
+        return redisData;
+      }
+
+      return this.findChartData(days = hours, timezoneOffset, groupHash);
+    } catch (err) {
+      return this.findChartData(days = hours, timezoneOffset, groupHash);
+    }
+  }
+
   /**
    * Fetch timestamps and total count of errors (or target error) for each day since
    *
diff --git a/src/redisHelper.ts b/src/redisHelper.ts
@@ -0,0 +1,138 @@
+import HawkCatcher from '@hawk.so/nodejs';
+import { createClient, RedisClientType } from 'redis';
+import { Effect, sgr } from './utils/ansi';
+
+/**
+ * Helper class for working with Redis
+ */
+export default class RedisHelper {
+  /**
+   * TTL for lock records in Redis (in seconds)
+   */
+  private static readonly LOCK_TTL = 10;
+
+  /**
+   * Redis client instance
+   */
+  private readonly redisClient!: RedisClientType;
+
+  /**
+   * Constructor
+   * Initializes the Redis client and sets up error handling
+   */
+  constructor() {
+    try {
+      this.redisClient = createClient({ url: process.env.REDIS_URL });
+
+      this.redisClient.on('error', (error) => {
+        console.error('[Redis] Client error:', error);
+        if (error) {
+          HawkCatcher.send(error);
+        }
+      });
+    } catch (error) {
+      console.error('[Redis] Error creating client:', error);
+    }
+  }
+
+  /**
+   * Connect to Redis
+   */
+  public async initialize(): Promise<void> {
+    try {
+      await this.redisClient.connect();
+      console.log('[Redis] Connected successfully');
+    } catch (error) {
+      console.error('[Redis] Connection failed:', error);
+      HawkCatcher.send(error as Error);
+    }
+  }
+
+  /**
+   * Close Redis client
+   */
+  public async close(): Promise<void> {
+    if (this.redisClient.isOpen) {
+      await this.redisClient.quit();
+      console.log('[Redis] Connection closed');
+    }
+  }
+  
+  public async getChartDataFromRedis(
+    hours: number,                // количество интервалов (часов или дней)
+    timezoneOffset = 0,
+    projectId = '',
+    groupHash = ''
+  ): Promise<{ timestamp: number; count: number }[]> {
+    if (!this.redisClient.isOpen) {
+      throw new Error('Redis client not connected');
+    }
+  
+    const key = groupHash
+      ? `ts:events:${groupHash}:hourly`
+      : projectId
+      ? `ts:events:${projectId}:hourly`
+      : `ts:events:hourly`;
+  
+    const now = Date.now();
+  
+    // определяем начало выборки
+    const fromDate = new Date(now);
+    fromDate.setMinutes(0, 0, 0);
+    fromDate.setMilliseconds(fromDate.getMilliseconds() - (hours * 60 * 60 * 1000));
+    const from = fromDate.getTime();
+  
+    let result: [string, string][] = [];
+    try {
+      result = (await this.redisClient.sendCommand([
+        'TS.RANGE',
+        key,
+        from.toString(),
+        now.toString(),
+      ])) as [string, string][] | [];
+    } catch (err: any) {
+      if (err.message.includes('TSDB: the key does not exist')) {
+        console.warn(`[Redis] Key ${key} does not exist, returning zeroed data`);
+        result = [];
+      } else {
+        throw err;
+      }
+    }
+
+    console.log(groupHash, result)
+  
+    // агрегируем события по интервалу
+    const dataPoints: { [ts: number]: number } = {};
+    for (const [tsStr] of result) {
+      const tsMs = Number(tsStr);
+      const date = new Date(tsMs);
+  
+      let intervalStart: number;
+      date.setMinutes(0, 0, 0);
+      intervalStart = Date.UTC(date.getUTCFullYear(), date.getUTCMonth(), date.getUTCDate(), date.getUTCHours());
+  
+      const intervalWithOffset = intervalStart + timezoneOffset * 60 * 1000;
+      
+      dataPoints[intervalWithOffset] = (dataPoints[intervalWithOffset] || 0) + 1;
+    }
+  
+    // заполняем пропущенные интервалы нулями
+    const filled: { timestamp: number; count: number }[] = [];
+    const nowDate = new Date(now);
+  
+    for (let i = 0; i < hours; i++) {
+      const date = new Date(nowDate);
+  
+      date.setHours(date.getHours() - i, 0, 0, 0);
+      var intervalStart = Date.UTC(date.getUTCFullYear(), date.getUTCMonth(), date.getUTCDate(), date.getUTCHours());
+  
+      const intervalWithOffset = intervalStart + timezoneOffset * 60 * 1000;
+      filled.push({
+        timestamp: Math.floor(intervalWithOffset / 1000),
+        count: dataPoints[intervalWithOffset] || 0,
+      });
+    }
+  
+    return filled.sort((a, b) => a.timestamp - b.timestamp);
+  }
+}
diff --git a/src/resolvers/event.js b/src/resolvers/event.js
@@ -86,7 +86,7 @@ module.exports = {
     async chartData({ projectId, groupHash }, { days, timezoneOffset }, context) {
       const factory = getEventsFactory(context, projectId);
 
-      return factory.findChartData(days, timezoneOffset, groupHash);
+      return factory.getChartData(days, timezoneOffset, projectId, groupHash);
     },
 
     /**
diff --git a/src/resolvers/project.js b/src/resolvers/project.js
@@ -468,7 +468,7 @@ module.exports = {
     async chartData(project, { days, timezoneOffset }, context) {
       const factory = getEventsFactory(context, project._id);
 
-      return factory.findChartData(days, timezoneOffset);
+      return factory.getChartData(days, timezoneOffset, project._id);
     },
 
     /**
diff --git a/yarn.lock b/yarn.lock
@@ -788,6 +788,40 @@
   resolved "https://registry.yarnpkg.com/@protobufjs/utf8/-/utf8-1.1.0.tgz#a777360b5b39a1a2e5106f8e858f2fd2d060c570"
   integrity sha512-Vvn3zZrhQZkkBE8LSuW3em98c0FwgO4nxzv6OdSxPKJIEKY2bGbHn+mhGIPerzI4twdxaP8/0+06HBpwf345Lw==
 
+"@redis/bloom@1.2.0":
+  version "1.2.0"
+  resolved "https://registry.yarnpkg.com/@redis/bloom/-/bloom-1.2.0.tgz#d3fd6d3c0af3ef92f26767b56414a370c7b63b71"
+  integrity sha512-HG2DFjYKbpNmVXsa0keLHp/3leGJz1mjh09f2RLGGLQZzSHpkmZWuwJbAvo3QcRY8p80m5+ZdXZdYOSBLlp7Cg==
+
+"@redis/client@1.6.1":
+  version "1.6.1"
+  resolved "https://registry.yarnpkg.com/@redis/client/-/client-1.6.1.tgz#c4636b7cb34e96008a988409b7e787364ae761a2"
+  integrity sha512-/KCsg3xSlR+nCK8/8ZYSknYxvXHwubJrU82F3Lm1Fp6789VQ0/3RJKfsmRXjqfaTA++23CvC3hqmqe/2GEt6Kw==
+  dependencies:
+    cluster-key-slot "1.1.2"
+    generic-pool "3.9.0"
+    yallist "4.0.0"
+
+"@redis/graph@1.1.1":
+  version "1.1.1"
+  resolved "https://registry.yarnpkg.com/@redis/graph/-/graph-1.1.1.tgz#8c10df2df7f7d02741866751764031a957a170ea"
+  integrity sha512-FEMTcTHZozZciLRl6GiiIB4zGm5z5F3F6a6FZCyrfxdKOhFlGkiAqlexWMBzCi4DcRoyiOsuLfW+cjlGWyExOw==
+
+"@redis/json@1.0.7":
+  version "1.0.7"
+  resolved "https://registry.yarnpkg.com/@redis/json/-/json-1.0.7.tgz#016257fcd933c4cbcb9c49cde8a0961375c6893b"
+  integrity sha512-6UyXfjVaTBTJtKNG4/9Z8PSpKE6XgSyEb8iwaqDcy+uKrd/DGYHTWkUdnQDyzm727V7p21WUMhsqz5oy65kPcQ==
+
+"@redis/search@1.2.0":
+  version "1.2.0"
+  resolved "https://registry.yarnpkg.com/@redis/search/-/search-1.2.0.tgz#50976fd3f31168f585666f7922dde111c74567b8"
+  integrity sha512-tYoDBbtqOVigEDMAcTGsRlMycIIjwMCgD8eR2t0NANeQmgK/lvxNAvYyb6bZDD4frHRhIHkJu2TBRvB0ERkOmw==
+
+"@redis/time-series@1.1.0":
+  version "1.1.0"
+  resolved "https://registry.yarnpkg.com/@redis/time-series/-/time-series-1.1.0.tgz#cba454c05ec201bd5547aaf55286d44682ac8eb5"
+  integrity sha512-c1Q99M5ljsIuc4YdaCwfUEXsofakb9c8+Zse2qxTadu8TalLXuAESzLvFAvNVbkmSlvlzIQOLpBCmWI9wTOt+g==
+
 "@shelf/jest-mongodb@^1.2.2":
   version "1.3.4"
   resolved "https://registry.yarnpkg.com/@shelf/jest-mongodb/-/jest-mongodb-1.3.4.tgz#200bac386cf513bed2d41952b1857689f0b88f31"
@@ -2138,6 +2172,11 @@ cloudpayments@^6.0.1:
     object-hash "^2.2.0"
     qs "^6.10.1"
 
+cluster-key-slot@1.1.2:
+  version "1.1.2"
+  resolved "https://registry.yarnpkg.com/cluster-key-slot/-/cluster-key-slot-1.1.2.tgz#88ddaa46906e303b5de30d3153b7d9fe0a0c19ac"
+  integrity sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==
+
 co@^4.6.0:
   version "4.6.0"
   resolved "https://registry.yarnpkg.com/co/-/co-4.6.0.tgz#6ea6bdf3d853ae54ccb8e47bfa0bf3f9031fb184"
@@ -3326,6 +3365,11 @@ gauge@^3.0.0:
     strip-ansi "^6.0.1"
     wide-align "^1.1.2"
 
+generic-pool@3.9.0:
+  version "3.9.0"
+  resolved "https://registry.yarnpkg.com/generic-pool/-/generic-pool-3.9.0.tgz#36f4a678e963f4fdb8707eab050823abc4e8f5e4"
+  integrity sha512-hymDOu5B53XvN4QT9dBmZxPX4CWhBPPLguTZ9MMFeFa/Kg0xWVfylOVNlJji/E7yTZWFd/q9GO5TxDLq156D7g==
+
 gensync@^1.0.0-beta.2:
   version "1.0.0-beta.2"
   resolved "https://registry.yarnpkg.com/gensync/-/gensync-1.0.0-beta.2.tgz#32a6ee76c3d7f52d46b2b1ae5d93fea8580a25e0"
@@ -5718,6 +5762,18 @@ readdirp@~3.6.0:
   dependencies:
     picomatch "^2.2.1"
 
+redis@^4.7.0:
+  version "4.7.1"
+  resolved "https://registry.yarnpkg.com/redis/-/redis-4.7.1.tgz#08588a30936be0e7ad9c0f3e1ac6a85ccaf73e94"
+  integrity sha512-S1bJDnqLftzHXHP8JsT5II/CtHWQrASX5K96REjWjlmWKrviSOLWmM7QnRLstAWsu1VBBV1ffV6DzCvxNP0UJQ==
+  dependencies:
+    "@redis/bloom" "1.2.0"
+    "@redis/client" "1.6.1"
+    "@redis/graph" "1.1.1"
+    "@redis/json" "1.0.7"
+    "@redis/search" "1.2.0"
+    "@redis/time-series" "1.1.0"
+
 regex-not@^1.0.0, regex-not@^1.0.2:
   version "1.0.2"
   resolved "https://registry.yarnpkg.com/regex-not/-/regex-not-1.0.2.tgz#1f4ece27e00b0b65e0247a6810e6a85d83a5752c"
@@ -7107,7 +7163,7 @@ y18n@^4.0.0:
   resolved "https://registry.yarnpkg.com/y18n/-/y18n-4.0.3.tgz#b5f259c82cd6e336921efd7bfd8bf560de9eeedf"
   integrity sha512-JKhqTOwSrqNA1NY5lSztJ1GrBiUodLMmIZuLiDaMRJ+itFd+ABVE8XBjOvIWL+rSqNDC74LCSFmlb/U4UZ4hJQ==
 
-yallist@^4.0.0:
+yallist@4.0.0, yallist@^4.0.0:
   version "4.0.0"
   resolved "https://registry.yarnpkg.com/yallist/-/yallist-4.0.0.tgz#9bb92790d9c0effec63be73519e11a35019a3a72"
   integrity sha512-3wdGidZyq5PB084XLES5TpOSRA3wjXAlIWMhum2kRcv/41Sn2emQ0dycQW4uZXLejwKvg6EsvbdlVL+FYEct7A==
