first commit

src/ingest/dto/ingest-edge-values.dto.ts

@@ -0,0 +1,16 @@
+import { Type } from 'class-transformer';
+import { IsDateString, IsNumber, IsObject, IsOptional } from 'class-validator';
+
+export class IngestEdgeValuesDto {
+  @IsOptional()
+  @IsDateString()
+  time?: string;
+
+  @IsOptional()
+  @Type(() => Number)
+  @IsNumber()
+  timestamp?: number;
+
+  @IsObject()
+  values!: Record<string, number>;
+}

src/ingest/dto/ingest-point.dto.ts

@@ -0,0 +1,25 @@
+import { Type } from 'class-transformer';
+import { IsDateString, IsNotEmpty, IsNumber, IsOptional, IsString } from 'class-validator';
+
+export class IngestPointDto {
+  @IsNotEmpty()
+  @IsString()
+  edge!: string;
+
+  @IsNotEmpty()
+  @IsString()
+  tag!: string;
+
+  @IsOptional()
+  @IsDateString()
+  time?: string;
+
+  @IsOptional()
+  @Type(() => Number)
+  @IsNumber()
+  timestamp?: number;
+
+  @Type(() => Number)
+  @IsNumber()
+  value!: number;
+}

src/ingest/ingest.controller.ts

@@ -0,0 +1,34 @@
+import { Body, Controller, Param, ParseArrayPipe, 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';
+import { IngestService } from './ingest.service';
+
+@Controller('ingest')
+@UseGuards(IngestApiKeyGuard)
+export class IngestController {
+  constructor(private readonly ingest: IngestService) {}
+
+  /** Принимает массив сырых точек и сохраняет их как записи истории. */
+  @Post()
+  ingestPoints(
+    @Body(new ParseArrayPipe({ items: IngestPointDto }))
+    points: IngestPointDto[],
+  ) {
+    return this.ingest.ingestPoints(points);
+  }
+
+  /** Принимает компактное тело запроса по edge и разворачивает каждое значение в точку истории. */
+  @Post(':edge')
+  ingestEdgeValues(@Param('edge') edge: string, @Body() body: IngestEdgeValuesDto) {
+    const points = Object.entries(body.values).map(([tag, value]) => ({
+      edge,
+      tag,
+      value: Number(value),
+      time: body.time,
+      timestamp: body.timestamp,
+    }));
+
+    return this.ingest.ingestPoints(points);
+  }
+}

src/ingest/ingest.module.ts

@@ -0,0 +1,10 @@
+import { Module } from '@nestjs/common';
+import { IngestApiKeyGuard } from '../common/ingest-api-key.guard';
+import { IngestController } from './ingest.controller';
+import { IngestService } from './ingest.service';
+
+@Module({
+  controllers: [IngestController],
+  providers: [IngestService, IngestApiKeyGuard],
+})
+export class IngestModule {}

src/ingest/ingest.service.ts

@@ -0,0 +1,108 @@
+import { BadRequestException, Injectable } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import { getIntegerConfig } from '../common/config-number';
+import { normalizeRequiredText } from '../common/normalize-text';
+import { DbService } from '../db/db.service';
+import { IngestPointDto } from './dto/ingest-point.dto';
+
+const DEFAULT_MAX_BATCH_SIZE = 10_000;
+
+type NormalizedPoint = {
+  time: Date;
+  edge: string;
+  tag: string;
+  value: number;
+};
+
+type IngestResult = {
+  processed: number;
+};
+
+@Injectable()
+export class IngestService {
+  constructor(
+    private readonly db: DbService,
+    private readonly config: ConfigService,
+  ) {}
+
+  /** Проверяет батч и записывает сырые точки в существующую таблицу history. */
+  async ingestPoints(points: IngestPointDto[]): Promise<IngestResult> {
+    if (!points.length) {
+      return { processed: 0 };
+    }
+
+    this.assertBatchSize(points.length);
+
+    const normalized = points.map((point) => this.normalizePoint(point));
+
+    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)
+      `,
+      [
+        normalized.map((point) => point.time),
+        normalized.map((point) => point.edge),
+        normalized.map((point) => point.tag),
+        normalized.map((point) => point.value),
+      ],
+    );
+
+    return { processed: normalized.length };
+  }
+
+  /** Защищает API от слишком больших ingest-запросов, занимающих соединение с БД надолго. */
+  private assertBatchSize(size: number): void {
+    const maxBatchSize = Math.max(
+      getIntegerConfig(this.config, 'INGEST_MAX_BATCH_SIZE', DEFAULT_MAX_BATCH_SIZE),
+      1,
+    );
+
+    if (size > maxBatchSize) {
+      throw new BadRequestException(`Batch size must not exceed ${maxBatchSize} points.`);
+    }
+  }
+
+  /** Приводит входной DTO к нормализованному виду для вставки в history. */
+  private normalizePoint(point: IngestPointDto): NormalizedPoint {
+    const edge = normalizeRequiredText(point.edge, 'edge');
+    const tag = normalizeRequiredText(point.tag, 'tag');
+    const time = this.parsePointTime(point);
+
+    if (!Number.isFinite(point.value)) {
+      throw new BadRequestException('value must be a finite number.');
+    }
+
+    return {
+      edge,
+      tag,
+      value: point.value,
+      time,
+    };
+  }
+
+  /** Принимает ISO-время или миллисекунды Unix и отклоняет точки без валидного времени. */
+  private parsePointTime(point: IngestPointDto): Date {
+    if (point.time) {
+      const date = new Date(point.time);
+      if (!Number.isNaN(date.getTime())) {
+        return date;
+      }
+    }
+
+    if (point.timestamp !== undefined) {
+      const date = new Date(point.timestamp);
+      if (!Number.isNaN(date.getTime())) {
+        return date;
+      }
+    }
+
+    throw new BadRequestException('Each point must contain valid time or timestamp.');
+  }
+}