From f5ed7fa9beb9e5d2ed64cb7b5e4380d488aeee71 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=D0=9F=D0=B5=D1=80=D0=B2=D0=BE=D0=B2=20=D0=90=D1=80=D1=82?=
 =?UTF-8?q?=D0=B5=D0=BC?= <apervov@alabuga.ru>
Date: Fri, 19 Jun 2026 09:10:49 +0300
Subject: [PATCH] add back set to current values.

---
 src/ingest/ingest.service.ts | 32 +++++++++++++++++++++++++++++++-
 1 file changed, 31 insertions(+), 1 deletion(-)

diff --git a/src/ingest/ingest.service.ts b/src/ingest/ingest.service.ts
index 06a1810..a6d2a37 100644
--- a/src/ingest/ingest.service.ts
+++ b/src/ingest/ingest.service.ts
@@ -25,7 +25,7 @@ export class IngestService {
     private readonly config: ConfigService,
   ) {}
 
-  /** Проверяет батч и записывает сырые точки в существующую таблицу history. */
+  /** Проверяет батч, пишет сырую историю и обновляет последний снимок current для UI/SSE. */
   async ingestPoints(points: IngestPointDto[]): Promise<IngestResult> {
     if (!points.length) {
       return { processed: 0 };
@@ -54,6 +54,36 @@ export class IngestService {
       ],
     );
 
+    await this.db.query(
+      `
+        INSERT INTO current (edge, tag, value, "createdAt", "updatedAt")
+        SELECT DISTINCT ON (edge, tag)
+          edge,
+          tag,
+          value,
+          time AS "createdAt",
+          time AS "updatedAt"
+        FROM unnest(
+          $1::timestamp[],
+          $2::varchar[],
+          $3::varchar[],
+          $4::double precision[]
+        ) AS point(time, edge, tag, value)
+        ORDER BY edge ASC, tag ASC, time DESC
+        ON CONFLICT (edge, tag)
+        DO UPDATE SET
+          value = EXCLUDED.value,
+          "updatedAt" = EXCLUDED."updatedAt"
+        WHERE current."updatedAt" <= EXCLUDED."updatedAt"
+      `,
+      [
+        normalized.map((point) => point.time),
+        normalized.map((point) => point.edge),
+        normalized.map((point) => point.tag),
+        normalized.map((point) => point.value),
+      ],
+    );
+
     return { processed: normalized.length };
   }