From 05c13f36a4ff5e476241c65c09dff038fd411f0c 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?= Date: Mon, 22 Jun 2026 02:50:24 +0300 Subject: [PATCH] =?UTF-8?q?=D0=9E=D0=B1=D0=BD=D0=BE=D0=B2=D0=B8=D0=BB=20?= =?UTF-8?q?=D0=BF=D1=80=D0=B8=D0=B5=D0=BC=20=D0=B4=D0=B0=D0=BD=D0=BD=D1=8B?= =?UTF-8?q?=D1=85=20=D0=B8=D0=B7=20mqtt-ingest=20=D0=BD=D0=B0=20=D0=BF?= =?UTF-8?q?=D0=BE=D1=8D=D0=BB=D0=B5=D0=BC=D0=B5=D0=BD=D1=82=D0=BD=D1=8B?= =?UTF-8?q?=D0=B9.=20=D0=A3=D0=B1=D1=80=D0=B0=D0=BB=20batch=20=D1=81=D0=BE?= =?UTF-8?q?=D1=85=D1=80=D0=B0=D0=BD=D0=B5=D0=BD=D0=B8=D0=B5.=20=D0=A1?= =?UTF-8?q?=D0=BA=D0=BE=D1=80=D1=80=D0=B5=D0=BA=D1=82=D0=B8=D1=80=D0=BE?= =?UTF-8?q?=D0=B2=D0=B0=D0=BB=20update=20=D0=B2=20=D1=82=D0=B0=D0=B1=D0=BB?= =?UTF-8?q?=D0=B8=D1=86=D1=83=20current.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 18 ++++----- src/ingest/ingest.controller.ts | 13 +++---- src/ingest/ingest.service.ts | 67 +++++++++++---------------------- 3 files changed, 36 insertions(+), 62 deletions(-) diff --git a/README.md b/README.md index 0490676..03c4403 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ Lightweight Drill Cloud API for an existing PostgreSQL/TimescaleDB database. - No migrations. - No `tag-translation` module. - No alarm log writes. -- No automatic `current`, `edge`, or `tag` upserts during ingest. +- No automatic `edge` or `tag` upserts during ingest. ## Database @@ -36,17 +36,15 @@ Historical chart ranges use continuous aggregates: - `POST /ingest` - `POST /ingest/:edge` -`POST /ingest` accepts an array of points: +`POST /ingest` accepts one point: ```json -[ - { - "edge": "edge5", - "tag": "hook_weight_1", - "time": "2026-06-19T00:00:00Z", - "value": 12.34 - } -] +{ + "edge": "edge5", + "tag": "hook_weight_1", + "time": "2026-06-19T00:00:00Z", + "value": 12.34 +} ``` `POST /ingest/:edge` accepts a compact edge payload: diff --git a/src/ingest/ingest.controller.ts b/src/ingest/ingest.controller.ts index 8706690..02d25e4 100644 --- a/src/ingest/ingest.controller.ts +++ b/src/ingest/ingest.controller.ts @@ -1,4 +1,4 @@ -import { Body, Controller, Param, ParseArrayPipe, Post, UseGuards } from '@nestjs/common'; +import { Body, Controller, Param, Post, UseGuards } from '@nestjs/common'; import { IngestApiKeyGuard } from '../common/ingest-api-key.guard'; import { IngestEdgeValuesDto } from './dto/ingest-edge-values.dto'; import { IngestPointDto } from './dto/ingest-point.dto'; @@ -9,16 +9,13 @@ import { IngestService } from './ingest.service'; export class IngestController { constructor(private readonly ingest: IngestService) {} - /** Принимает массив сырых точек и сохраняет их как записи истории. */ + /** Принимает одну сырую точку и сохраняет ее как запись истории/current. */ @Post() - ingestPoints( - @Body(new ParseArrayPipe({ items: IngestPointDto })) - points: IngestPointDto[], - ) { - return this.ingest.ingestPoints(points); + ingestPoint(@Body() point: IngestPointDto) { + return this.ingest.ingestPoint(point); } - /** Принимает компактное тело запроса по edge и разворачивает каждое значение в точку истории. */ + /** Принимает компактное тело запроса по edge и сохраняет каждое значение построчно. */ @Post(':edge') ingestEdgeValues(@Param('edge') edge: string, @Body() body: IngestEdgeValuesDto) { const points = Object.entries(body.values).map(([tag, value]) => ({ diff --git a/src/ingest/ingest.service.ts b/src/ingest/ingest.service.ts index aa7a126..e44a44f 100644 --- a/src/ingest/ingest.service.ts +++ b/src/ingest/ingest.service.ts @@ -17,68 +17,47 @@ type IngestResult = { export class IngestService { constructor(private readonly db: DbService) {} - /** Проверяет батч, пишет сырую историю и обновляет последний снимок current для UI/SSE. */ - async ingestPoints(points: IngestPointDto[]): Promise { - if (!points.length) { - return { processed: 0 }; - } - - const normalized: NormalizedPoint[] = points.map((point) => ({ + /** Пишет одну сырую точку в history и обновляет текущий снимок current для UI/SSE. */ + async ingestPoint(point: IngestPointDto): Promise { + const normalized: NormalizedPoint = { edge: point.edge, tag: point.tag, value: point.value, time: new Date((point.time ?? point.timestamp) as string | number), - })); + }; await this.db.query( ` INSERT INTO history (timestamp, edge, tag, value) - SELECT time, edge, tag, value - FROM unnest( - $1::timestamp[], - $2::varchar[], - $3::varchar[], - $4::double precision[] - ) AS point(time, edge, tag, value) + VALUES ($1, $2, $3, $4) `, - [ - normalized.map((point) => point.time), - normalized.map((point) => point.edge), - normalized.map((point) => point.tag), - normalized.map((point) => point.value), - ], + [normalized.time, normalized.edge, normalized.tag, normalized.value], ); 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 + VALUES ($1, $2, $3, NOW(), NOW()) ON CONFLICT (edge, tag) DO UPDATE SET - value = EXCLUDED.value, - "updatedAt" = EXCLUDED."updatedAt" - WHERE current."updatedAt" <= EXCLUDED."updatedAt" + value = $3, + "updatedAt" = NOW() `, - [ - normalized.map((point) => point.time), - normalized.map((point) => point.edge), - normalized.map((point) => point.tag), - normalized.map((point) => point.value), - ], + [normalized.edge, normalized.tag, normalized.value], ); - return { processed: normalized.length }; + return { processed: 1 }; + } + + /** Сохраняет несколько точек без batch SQL: каждая точка проходит тот же линейный путь. */ + async ingestPoints(points: IngestPointDto[]): Promise { + let processed = 0; + + for (const point of points) { + const result = await this.ingestPoint(point); + processed += result.processed; + } + + return { processed }; } }