APervov/cloud
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

cloud creation

Browse Source
This commit is contained in:
Первов Артем
2026-06-17 09:36:58 +03:00
commit 123bd97018
46 changed files with 12285 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
.env.example Normal file
View File

@@ -0,0 +1,16 @@
PORT=3100
DATABASE_URL=postgres://postgres:postgres@localhost:5432/drill_cloud_v2
CORS_ALLOWED_ORIGINS=http://localhost:3000,http://localhost:5173
# Optional. When set, POST /ingest requires x-api-key: <value>.
INGEST_API_KEY=
# Pool sizing for node-postgres.
PG_POOL_MAX=20
# Maximum points accepted by one POST /ingest request.
INGEST_MAX_BATCH_SIZE=10000
# Graph API defaults.
HISTORY_TARGET_POINTS=2000
HISTORY_MAX_TARGET_POINTS=2000

8
.env.local.example Normal file
View File

@@ -0,0 +1,8 @@
PORT=3100
DATABASE_URL=postgres://postgres:postgres@localhost:5432/drill_cloud_v2
CORS_ALLOWED_ORIGINS=http://localhost:3000,http://localhost:5173
INGEST_API_KEY=dev-local-key
PG_POOL_MAX=20
INGEST_MAX_BATCH_SIZE=10000
HISTORY_TARGET_POINTS=2000
HISTORY_MAX_TARGET_POINTS=2000

36
.gitignore vendored Normal file
View File

@@ -0,0 +1,36 @@
# Dependencies
node_modules/
# Build output
dist/
coverage/
*.tsbuildinfo
# Environment files
.env
.env.*
!.env.example
!.env.local.example
# Logs
*.log
npm-debug.log*
yarn-debug.log*
yarn-error.log*
pnpm-debug.log*
# Runtime/dev artifacts
*.pid
*.seed
*.local
# IDE and OS files
.idea/
.vscode/
.DS_Store
Thumbs.db
# Tool caches
.eslintcache
.npm/
.cache/

19
Dockerfile Normal file
View File

@@ -0,0 +1,19 @@
FROM node:22-alpine AS deps
WORKDIR /app
COPY package*.json ./
RUN npm ci
FROM node:22-alpine AS build
WORKDIR /app
COPY --from=deps /app/node_modules ./node_modules
COPY . .
RUN npm run build
FROM node:22-alpine AS runtime
WORKDIR /app
ENV NODE_ENV=production
COPY package*.json ./
RUN npm ci --omit=dev
COPY --from=build /app/dist ./dist
COPY migrations ./migrations
CMD ["sh", "-c", "node --env-file-if-exists=.env dist/migrations/migrate.js && node --env-file-if-exists=.env dist/main.js"]

112
LOCAL_NO_DOCKER.md Normal file
View File

@@ -0,0 +1,112 @@
# Local Run Without Docker
You can run the API without Docker. The application only needs Node.js and a PostgreSQL database with the TimescaleDB extension installed.
## Requirements
- Node.js 22+
- PostgreSQL 17 or 18
- TimescaleDB extension installed into that PostgreSQL instance
- `psql` available in your terminal
Official TimescaleDB docs for Windows describe installing PostgreSQL, running the TimescaleDB installer, restarting the PostgreSQL service, and enabling the extension with:
```sql
CREATE EXTENSION IF NOT EXISTS timescaledb;
```
Docs: https://docs.tigerdata.com/self-hosted/latest/install/installation-windows/
## 1. Create local database
Open PowerShell and run:
```powershell
createdb -h localhost -p 5432 -U postgres drill_cloud_v2
```
If `createdb` is not in PATH, add your PostgreSQL bin directory, for example:
```powershell
$env:Path += ";C:\Program Files\PostgreSQL\17\bin"
```
## 2. Enable TimescaleDB
```powershell
psql -h localhost -p 5432 -U postgres -d drill_cloud_v2 -c "CREATE EXTENSION IF NOT EXISTS timescaledb;"
```
Check it:
```powershell
psql -h localhost -p 5432 -U postgres -d drill_cloud_v2 -c "\dx"
```
You should see `timescaledb` in the extension list.
## 3. Configure app env
```powershell
Copy-Item .env.local.example .env
```
Edit `.env` if your local PostgreSQL password, port, or database name differs:
```text
DATABASE_URL=postgres://postgres:postgres@localhost:5432/drill_cloud_v2
```
## 4. Install and migrate
```powershell
npm install
npm run migrate
```
## 5. Start API
```powershell
npm run start:dev
```
The API starts on:
```text
http://localhost:3100
```
Health check:
```powershell
Invoke-RestMethod http://localhost:3100/health
```
## Useful local checks
Send one point:
```powershell
Invoke-RestMethod `
-Method Post `
-Uri http://localhost:3100/ingest `
-Headers @{ "x-api-key" = "dev-local-key" } `
-ContentType "application/json" `
-Body '[{"edge":"roman","tag":"test_tag","time":"2026-06-15T12:00:00Z","value":42}]'
```
Read current:
```powershell
Invoke-RestMethod "http://localhost:3100/current?edge=roman"
```
Read chart data:
```powershell
Invoke-RestMethod "http://localhost:3100/history?edge=roman&tags=test_tag&from=2026-06-15T00:00:00Z&to=2026-06-16T00:00:00Z"
```
## Note
The old remote database you gave me does not currently have TimescaleDB enabled. To run this v2 schema there, an admin user must install/allowlist TimescaleDB and run the migrations.

120
README.md Normal file
View File

@@ -0,0 +1,120 @@
# Drill Cloud v2
Optimized Drill Cloud backend for high-volume time-series reads on PostgreSQL + TimescaleDB.
## Why v2
The existing `history` table already has tens of millions of rows. The expensive path is not the number of tags, but dense time-series scans for charts. v2 stores raw points in a Timescale hypertable, keeps latest values in a small `current_values` table, and serves charts from raw data or continuous aggregates depending on the requested range.
Default chart budget is **2000 points per series**.
## Run locally
With Docker:
```bash
cp .env.example .env
docker compose up -d
npm install
npm run migrate
npm run start:dev
```
Default local database:
```text
postgres://postgres:postgres@localhost:5435/drill_cloud_v2
```
Without Docker, use a locally installed PostgreSQL + TimescaleDB and copy `.env.local.example` to `.env`.
See [LOCAL_NO_DOCKER.md](./LOCAL_NO_DOCKER.md).
## API
### Health
```http
GET /health
```
### Ingest array
```http
POST /ingest
Content-Type: application/json
x-api-key: <INGEST_API_KEY if configured>
[
{
"edge": "roman",
"tag": "BN1_24V_Supply_Fault",
"time": "2026-06-10T22:29:05.098Z",
"value": 1
}
]
```
`timestamp` as Unix milliseconds is also accepted instead of `time`.
### Ingest edge values
```http
POST /ingest/roman
Content-Type: application/json
{
"timestamp": 1781123345098,
"values": {
"BN1_24V_Supply_Fault": 1,
"Term_Drehz_IW_Motor": 1028
}
}
```
### Current values
```http
GET /current?edge=roman
GET /current?edge=roman&tags=tag1,tag2
```
### History for charts
```http
GET /history?edge=roman&tags=BN1_24V_Supply_Fault&from=2026-04-18T00:00:00Z&to=2026-04-26T00:00:00Z
```
Optional parameters:
- `targetPoints` - 100..2000, default 2000.
- `valueMode` - `avg`, `last`, `first`, `min`, `max` for aggregate layers.
- `tags` - comma-separated tag ids. If omitted, v2 uses `edge_tag`.
The response includes:
- `source`: `raw`, `1m`, `5m`, `1h`, `1d`, or `latest`.
- `series[].points[]`: compact chart points shaped as `{ "t": unixMs, "v": value }`.
- Aggregate points also include `first`, `min`, `max`, `avg`, `last`, `count`.
## Data model
Hot path tables:
- `history_points` - Timescale hypertable partitioned by `time`.
- `current_values` - latest value per `(edge_id, tag_id)`.
- `edge`, `tag`, `edge_tag` - catalog and series membership.
Continuous aggregates:
- `history_1m`
- `history_5m`
- `history_1h`
- `history_1d`
## Production notes
- Use TimescaleDB, not plain PostgreSQL, before running migrations.
- Keep `INGEST_API_KEY` set in production.
- Run chart queries with explicit `tags` whenever the UI knows selected series.
- For historical import from the old database, bulk copy into `history_points`, then refresh continuous aggregates.

21
docker-compose.yml Normal file
View File

@@ -0,0 +1,21 @@
services:
timescaledb:
container_name: cloud-timescaledb
image: timescale/timescaledb:latest-pg17
restart: unless-stopped
environment:
POSTGRES_DB: drill_cloud_v2
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
ports:
- "5437:5432"
volumes:
- drill-cloud-v2-timescale:/var/lib/postgresql/data
volumes:
drill-cloud-v2-timescale:
networks:
proxy:
external: true

24
eslint.config.mjs Normal file
View File

@@ -0,0 +1,24 @@
// @ts-check
import eslint from '@eslint/js';
import globals from 'globals';
import tseslint from 'typescript-eslint';
export default tseslint.config(
{
ignores: ['dist/**', 'node_modules/**', 'eslint.config.mjs'],
},
eslint.configs.recommended,
...tseslint.configs.recommended,
{
languageOptions: {
globals: {
...globals.node,
...globals.jest,
},
parserOptions: {
projectService: true,
tsconfigRootDir: import.meta.dirname,
},
},
},
);

199
migrations/001_init_timescale.sql Normal file
View File

@@ -0,0 +1,199 @@
CREATE EXTENSION IF NOT EXISTS timescaledb;
CREATE TABLE IF NOT EXISTS edge (
id text PRIMARY KEY,
name text NOT NULL,
parent_id text REFERENCES edge(id) ON DELETE RESTRICT,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS tag (
id text PRIMARY KEY,
name text NOT NULL,
tag_group text,
min_value double precision,
max_value double precision,
unit_of_measurement text,
comment text,
precision smallint,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS edge_tag (
edge_id text NOT NULL REFERENCES edge(id) ON DELETE CASCADE,
tag_id text NOT NULL REFERENCES tag(id) ON DELETE CASCADE,
created_at timestamptz NOT NULL DEFAULT now(),
PRIMARY KEY (edge_id, tag_id)
);
CREATE TABLE IF NOT EXISTS history_points (
time timestamptz NOT NULL,
edge_id text NOT NULL,
tag_id text NOT NULL,
value double precision NOT NULL,
received_at timestamptz NOT NULL DEFAULT now()
);
SELECT create_hypertable(
'history_points',
'time',
chunk_time_interval => INTERVAL '1 day',
if_not_exists => TRUE
);
CREATE INDEX IF NOT EXISTS history_points_edge_tag_time_desc_idx
ON history_points (edge_id, tag_id, time DESC)
INCLUDE (value);
CREATE INDEX IF NOT EXISTS history_points_time_brin_idx
ON history_points USING brin (time);
CREATE TABLE IF NOT EXISTS current_values (
edge_id text NOT NULL,
tag_id text NOT NULL,
value double precision NOT NULL,
time timestamptz NOT NULL,
updated_at timestamptz NOT NULL DEFAULT now(),
PRIMARY KEY (edge_id, tag_id)
);
CREATE INDEX IF NOT EXISTS current_values_updated_at_idx
ON current_values (updated_at DESC);
ALTER TABLE history_points SET (
timescaledb.compress,
timescaledb.compress_segmentby = 'edge_id,tag_id',
timescaledb.compress_orderby = 'time DESC'
);
SELECT add_compression_policy(
'history_points',
INTERVAL '7 days',
if_not_exists => TRUE
);
CREATE MATERIALIZED VIEW IF NOT EXISTS history_1m
WITH (timescaledb.continuous) AS
SELECT
time_bucket(INTERVAL '1 minute', time) AS bucket,
edge_id,
tag_id,
first(value, time) AS first_value,
min(value) AS min_value,
max(value) AS max_value,
avg(value) AS avg_value,
last(value, time) AS last_value,
count(*)::integer AS point_count
FROM history_points
GROUP BY bucket, edge_id, tag_id
WITH NO DATA;
CREATE MATERIALIZED VIEW IF NOT EXISTS history_5m
WITH (timescaledb.continuous) AS
SELECT
time_bucket(INTERVAL '5 minutes', time) AS bucket,
edge_id,
tag_id,
first(value, time) AS first_value,
min(value) AS min_value,
max(value) AS max_value,
avg(value) AS avg_value,
last(value, time) AS last_value,
count(*)::integer AS point_count
FROM history_points
GROUP BY bucket, edge_id, tag_id
WITH NO DATA;
CREATE MATERIALIZED VIEW IF NOT EXISTS history_1h
WITH (timescaledb.continuous) AS
SELECT
time_bucket(INTERVAL '1 hour', time) AS bucket,
edge_id,
tag_id,
first(value, time) AS first_value,
min(value) AS min_value,
max(value) AS max_value,
avg(value) AS avg_value,
last(value, time) AS last_value,
count(*)::integer AS point_count
FROM history_points
GROUP BY bucket, edge_id, tag_id
WITH NO DATA;
CREATE MATERIALIZED VIEW IF NOT EXISTS history_1d
WITH (timescaledb.continuous) AS
SELECT
time_bucket(INTERVAL '1 day', time) AS bucket,
edge_id,
tag_id,
first(value, time) AS first_value,
min(value) AS min_value,
max(value) AS max_value,
avg(value) AS avg_value,
last(value, time) AS last_value,
count(*)::integer AS point_count
FROM history_points
GROUP BY bucket, edge_id, tag_id
WITH NO DATA;
CREATE INDEX IF NOT EXISTS history_1m_edge_tag_bucket_idx
ON history_1m (edge_id, tag_id, bucket DESC);
CREATE INDEX IF NOT EXISTS history_5m_edge_tag_bucket_idx
ON history_5m (edge_id, tag_id, bucket DESC);
CREATE INDEX IF NOT EXISTS history_1h_edge_tag_bucket_idx
ON history_1h (edge_id, tag_id, bucket DESC);
CREATE INDEX IF NOT EXISTS history_1d_edge_tag_bucket_idx
ON history_1d (edge_id, tag_id, bucket DESC);
DO $$
BEGIN
PERFORM add_continuous_aggregate_policy(
'history_1m',
start_offset => INTERVAL '7 days',
end_offset => INTERVAL '1 minute',
schedule_interval => INTERVAL '1 minute'
);
EXCEPTION
WHEN duplicate_object THEN NULL;
END $$;
DO $$
BEGIN
PERFORM add_continuous_aggregate_policy(
'history_5m',
start_offset => INTERVAL '30 days',
end_offset => INTERVAL '5 minutes',
schedule_interval => INTERVAL '5 minutes'
);
EXCEPTION
WHEN duplicate_object THEN NULL;
END $$;
DO $$
BEGIN
PERFORM add_continuous_aggregate_policy(
'history_1h',
start_offset => INTERVAL '180 days',
end_offset => INTERVAL '1 hour',
schedule_interval => INTERVAL '15 minutes'
);
EXCEPTION
WHEN duplicate_object THEN NULL;
END $$;
DO $$
BEGIN
PERFORM add_continuous_aggregate_policy(
'history_1d',
start_offset => INTERVAL '5 years',
end_offset => INTERVAL '1 day',
schedule_interval => INTERVAL '1 hour'
);
EXCEPTION
WHEN duplicate_object THEN NULL;
END $$;

4
migrations/002_migration_tracking.sql Normal file
View File

@@ -0,0 +1,4 @@
CREATE TABLE IF NOT EXISTS app_migrations (
id text PRIMARY KEY,
applied_at timestamptz NOT NULL DEFAULT now()
);

8
nest-cli.json Normal file
View File

@@ -0,0 +1,8 @@
{
"$schema": "https://json.schemastore.org/nest-cli",
"collection": "@nestjs/schematics",
"sourceRoot": "src",
"compilerOptions": {
"deleteOutDir": true
}
}

10261
package-lock.json generated Normal file
View File

File diff suppressed because it is too large Load Diff

59
package.json Normal file
View File

@@ -0,0 +1,59 @@
{
"name": "drill-cloud-v2",
"version": "0.1.0",
"private": true,
"description": "Optimized Drill Cloud API on PostgreSQL + TimescaleDB.",
"license": "PERSONAL",
"scripts": {
"build": "nest build",
"start": "nest start",
"start:dev": "nest start --watch",
"start:prod": "node dist/main.js",
"migrate": "node --env-file-if-exists=.env -r ts-node/register -r tsconfig-paths/register src/migrations/migrate.ts",
"lint": "eslint \"src/**/*.ts\"",
"test": "jest --runInBand"
},
"dependencies": {
"@nestjs/common": "^11.0.1",
"@nestjs/config": "^4.0.2",
"@nestjs/core": "^11.0.1",
"@nestjs/platform-express": "^11.0.1",
"class-transformer": "^0.5.1",
"class-validator": "^0.14.2",
"compression": "^1.8.1",
"pg": "^8.13.1",
"reflect-metadata": "^0.2.2",
"rxjs": "^7.8.1"
},
"devDependencies": {
"@nestjs/cli": "^11.0.0",
"@types/compression": "^1.8.1",
"@types/express": "^5.0.0",
"@types/jest": "^30.0.0",
"@types/node": "^22.10.7",
"@types/pg": "^8.11.10",
"jest": "^30.0.0",
"ts-jest": "^29.2.5",
"ts-loader": "^9.5.2",
"ts-node": "^10.9.2",
"tsconfig-paths": "^4.2.0",
"typescript": "^5.7.3",
"typescript-eslint": "^8.20.0",
"@eslint/js": "^9.18.0",
"eslint": "^9.18.0",
"globals": "^16.0.0"
},
"jest": {
"moduleFileExtensions": [
"js",
"json",
"ts"
],
"rootDir": "src",
"testRegex": ".*\\.spec\\.ts$",
"transform": {
"^.+\\.(t|j)s$": "ts-jest"
},
"testEnvironment": "node"
}
}

21
src/app.module.ts Normal file
View File

@@ -0,0 +1,21 @@
import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config';
import { CurrentModule } from './current/current.module';
import { DbModule } from './db/db.module';
import { EdgeModule } from './edge/edge.module';
import { HealthController } from './health.controller';
import { HistoryModule } from './history/history.module';
import { IngestModule } from './ingest/ingest.module';
@Module({
imports: [
ConfigModule.forRoot({ isGlobal: true }),
DbModule,
IngestModule,
EdgeModule,
CurrentModule,
HistoryModule,
],
controllers: [HealthController],
})
export class AppModule {}

25
src/common/config-number.ts Normal file
View File

@@ -0,0 +1,25 @@
import { ConfigService } from '@nestjs/config';
// Читает числовой env-параметр с безопасным fallback, чтобы конфиг не ломал старт сервиса.
export function getNumberConfig(
config: ConfigService,
key: string,
fallback: number,
): number {
const value = config.get<string>(key);
if (value === undefined || value === '') {
return fallback;
}
const parsed = Number(value);
return Number.isFinite(parsed) ? parsed : fallback;
}
// Используется для лимитов и размеров пулов, где дробные значения не имеют смысла.
export function getIntegerConfig(
config: ConfigService,
key: string,
fallback: number,
): number {
return Math.floor(getNumberConfig(config, key, fallback));
}

23
src/common/ingest-api-key.guard.ts Normal file
View File

@@ -0,0 +1,23 @@
import { CanActivate, ExecutionContext, Injectable, UnauthorizedException } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import type { Request } from 'express';
@Injectable()
export class IngestApiKeyGuard implements CanActivate {
constructor(private readonly config: ConfigService) {}
canActivate(context: ExecutionContext): boolean {
const expectedKey = this.config.get<string>('INGEST_API_KEY');
if (!expectedKey) {
return true;
}
const request = context.switchToHttp().getRequest<Request>();
const actualKey = request.header('x-api-key');
if (actualKey !== expectedKey) {
throw new UnauthorizedException('Invalid ingest API key.');
}
return true;
}
}

11
src/common/normalize-text.ts Normal file
View File

@@ -0,0 +1,11 @@
import { BadRequestException } from '@nestjs/common';
// Единая нормализация обязательных строк: trim + понятная 400-ошибка для API.
export function normalizeRequiredText(value: string, fieldName: string): string {
const normalized = value.trim();
if (!normalized) {
throw new BadRequestException(`${fieldName} is required.`);
}
return normalized;
}

13
src/common/query-list.ts Normal file
View File

@@ -0,0 +1,13 @@
// Поддерживает both styles: ?tags=a,b и ?tags=a&tags=b; дубликаты убираются.
export function parseCommaSeparatedList(value: unknown): string[] | undefined {
if (value == null || value === '') {
return undefined;
}
const rawItems = Array.isArray(value)
? value.flatMap((item) => String(item).split(','))
: String(value).split(',');
const items = rawItems.map((item) => item.trim()).filter(Boolean);
return items.length ? Array.from(new Set(items)) : undefined;
}

41
src/current/current-events.service.ts Normal file
View File

@@ -0,0 +1,41 @@
import { Injectable } from '@nestjs/common';
import { Observable, Subject, filter, map } from 'rxjs';
export type CurrentEventItem = {
edge: string;
tag: string;
value: number;
time: Date;
updatedAt: Date;
};
export type CurrentEvent = {
edge: string;
items: CurrentEventItem[];
};
@Injectable()
export class CurrentEventsService {
private readonly events = new Subject<CurrentEvent>();
// Публикует latest-изменения после успешной записи ingest; подписчики получают только свой edge/tags.
publish(event: CurrentEvent): void {
if (event.items.length) {
this.events.next(event);
}
}
stream(edge: string, tags?: string[]): Observable<{ data: CurrentEvent }> {
const tagSet = tags?.length ? new Set(tags) : undefined;
return this.events.asObservable().pipe(
filter((event) => event.edge === edge),
map((event) => ({
...event,
items: tagSet ? event.items.filter((item) => tagSet.has(item.tag)) : event.items,
})),
filter((event) => event.items.length > 0),
map((event) => ({ data: event })),
);
}
}

27
src/current/current.controller.ts Normal file
View File

@@ -0,0 +1,27 @@
import { Controller, Get, Header, Query, Sse } from '@nestjs/common';
import { Observable } from 'rxjs';
import { CurrentEventsService } from './current-events.service';
import { CurrentService } from './current.service';
import { GetCurrentDto } from './dto/get-current.dto';
@Controller('current')
export class CurrentController {
constructor(
private readonly current: CurrentService,
private readonly currentEvents: CurrentEventsService,
) {}
@Get()
@Header('Cache-Control', 'no-store')
@Header('Pragma', 'no-cache')
@Header('Expires', '0')
findByEdge(@Query() query: GetCurrentDto) {
return this.current.findByEdge(query);
}
@Sse('events')
@Header('Cache-Control', 'no-store')
events(@Query() query: GetCurrentDto): Observable<MessageEvent> {
return this.currentEvents.stream(query.edge, query.tags) as unknown as Observable<MessageEvent>;
}
}

11
src/current/current.module.ts Normal file
View File

@@ -0,0 +1,11 @@
import { Module } from '@nestjs/common';
import { CurrentController } from './current.controller';
import { CurrentEventsService } from './current-events.service';
import { CurrentService } from './current.service';
@Module({
controllers: [CurrentController],
providers: [CurrentService, CurrentEventsService],
exports: [CurrentEventsService],
})
export class CurrentModule {}

57
src/current/current.service.ts Normal file
View File

@@ -0,0 +1,57 @@
import { Injectable } from '@nestjs/common';
import { normalizeRequiredText } from '../common/normalize-text';
import { DbService } from '../db/db.service';
import { GetCurrentDto } from './dto/get-current.dto';
type CurrentRow = {
edge_id: string;
tag_id: string;
value: number;
time: Date;
updated_at: Date;
};
type CurrentItem = {
edge: string;
tag: string;
value: number;
time: Date;
updatedAt: Date;
};
type CurrentResponse = {
edge: string;
items: CurrentItem[];
};
@Injectable()
export class CurrentService {
constructor(private readonly db: DbService) {}
// Возвращает latest-срез по edge; optional tags ограничивает ответ нужными сериями.
async findByEdge(query: GetCurrentDto): Promise<CurrentResponse> {
const edge = normalizeRequiredText(query.edge, 'edge');
const tags = query.tags ?? null;
const result = await this.db.query<CurrentRow>(
`
SELECT edge_id, tag_id, value, time, updated_at
FROM current_values
WHERE edge_id = $1
AND ($2::text[] IS NULL OR tag_id = ANY($2::text[]))
ORDER BY tag_id ASC
`,
[edge, tags],
);
return {
edge,
items: result.rows.map((row) => ({
edge: row.edge_id,
tag: row.tag_id,
value: row.value,
time: row.time,
updatedAt: row.updated_at,
})),
};
}
}

15
src/current/dto/get-current.dto.ts Normal file
View File

@@ -0,0 +1,15 @@
import { Transform } from 'class-transformer';
import { IsArray, IsNotEmpty, IsOptional, IsString } from 'class-validator';
import { parseCommaSeparatedList } from '../../common/query-list';
export class GetCurrentDto {
@IsNotEmpty()
@IsString()
edge!: string;
@IsOptional()
@Transform(({ value }) => parseCommaSeparatedList(value))
@IsArray()
@IsString({ each: true })
tags?: string[];
}

9
src/db/db.module.ts Normal file
View File

@@ -0,0 +1,9 @@
import { Global, Module } from '@nestjs/common';
import { DbService } from './db.service';
@Global()
@Module({
providers: [DbService],
exports: [DbService],
})
export class DbModule {}

83
src/db/db.service.ts Normal file
View File

@@ -0,0 +1,83 @@
import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { Pool, PoolClient, QueryResult, QueryResultRow } from 'pg';
import { getIntegerConfig } from '../common/config-number';
type DatabaseHealth = {
now: Date;
timescaledb_installed: boolean;
timescaledb_version: string | null;
};
@Injectable()
export class DbService implements OnModuleInit, OnModuleDestroy {
private pool!: Pool;
constructor(private readonly config: ConfigService) {}
// Инициализируем пул один раз на модуль и сразу проверяем доступность БД.
async onModuleInit(): Promise<void> {
const connectionString = this.config.get<string>('DATABASE_URL');
if (!connectionString) {
throw new Error('DATABASE_URL is required.');
}
this.pool = new Pool({
connectionString,
max: getIntegerConfig(this.config, 'PG_POOL_MAX', 20),
idleTimeoutMillis: 30_000,
connectionTimeoutMillis: 5_000,
statement_timeout: 60_000,
query_timeout: 60_000,
application_name: 'drill-cloud-v2',
});
await this.health();
}
async onModuleDestroy(): Promise<void> {
await this.pool?.end();
}
query<T extends QueryResultRow = QueryResultRow>(
text: string,
values: readonly unknown[] = [],
): Promise<QueryResult<T>> {
return this.pool.query<T>(text, [...values]);
}
// Общая транзакционная обертка: caller описывает операции, сервис отвечает за commit/rollback.
async withTransaction<T>(handler: (client: PoolClient) => Promise<T>): Promise<T> {
const client = await this.pool.connect();
try {
await client.query('BEGIN');
const result = await handler(client);
await client.query('COMMIT');
return result;
} catch (error) {
try {
await client.query('ROLLBACK');
} catch {
// Preserve the original transaction error.
}
throw error;
} finally {
client.release();
}
}
// Легкий probe для /health: проверяет связь с БД и установленный TimescaleDB.
async health(): Promise<DatabaseHealth> {
const result = await this.query<DatabaseHealth>(`
SELECT
now(),
ext.extversion IS NOT NULL AS timescaledb_installed,
ext.extversion AS timescaledb_version
FROM (SELECT 1) AS probe
LEFT JOIN pg_extension AS ext
ON ext.extname = 'timescaledb'
`);
return result.rows[0];
}
}

15
src/edge/edge.controller.ts Normal file
View File

@@ -0,0 +1,15 @@
import { Controller, Get, Header } from '@nestjs/common';
import { EdgeService } from './edge.service';
@Controller('edge')
export class EdgeController {
constructor(private readonly edge: EdgeService) {}
@Get()
@Header('Cache-Control', 'no-store')
@Header('Pragma', 'no-cache')
@Header('Expires', '0')
findAll() {
return this.edge.findAll();
}
}

9
src/edge/edge.module.ts Normal file
View File

@@ -0,0 +1,9 @@
import { Module } from '@nestjs/common';
import { EdgeController } from './edge.controller';
import { EdgeService } from './edge.service';
@Module({
controllers: [EdgeController],
providers: [EdgeService],
})
export class EdgeModule {}

71
src/edge/edge.service.ts Normal file
View File

@@ -0,0 +1,71 @@
import { Injectable } from '@nestjs/common';
import { DbService } from '../db/db.service';
type EdgeRow = {
id: string;
name: string;
tag_count: number;
current_tag_count: number;
live_tag_count: number;
last_data_at: Date | null;
created_at: Date;
updated_at: Date;
};
type EdgeItem = {
id: string;
name: string;
tagCount: number;
currentTagCount: number;
liveTagCount: number;
lastDataAt: Date | null;
createdAt: Date;
updatedAt: Date;
};
type EdgeResponse = {
items: EdgeItem[];
};
@Injectable()
export class EdgeService {
constructor(private readonly db: DbService) {}
// Возвращает каталог буровых с легкой сводкой по свежести данных для dashboard UI.
async findAll(): Promise<EdgeResponse> {
const result = await this.db.query<EdgeRow>(`
SELECT
e.id,
e.name,
count(DISTINCT et.tag_id)::integer AS tag_count,
count(DISTINCT cv.tag_id)::integer AS current_tag_count,
count(DISTINCT cv.tag_id) FILTER (
WHERE cv.time >= now() - INTERVAL '30 seconds'
)::integer AS live_tag_count,
max(cv.time) AS last_data_at,
e.created_at,
e.updated_at
FROM edge AS e
LEFT JOIN edge_tag AS et
ON et.edge_id = e.id
LEFT JOIN current_values AS cv
ON cv.edge_id = e.id
AND cv.tag_id = et.tag_id
GROUP BY e.id, e.name, e.created_at, e.updated_at
ORDER BY e.name ASC, e.id ASC
`);
return {
items: result.rows.map((row) => ({
id: row.id,
name: row.name,
tagCount: row.tag_count,
currentTagCount: row.current_tag_count,
liveTagCount: row.live_tag_count,
lastDataAt: row.last_data_at,
createdAt: row.created_at,
updatedAt: row.updated_at,
})),
};
}
}

16
src/health.controller.ts Normal file
View File

@@ -0,0 +1,16 @@
import { Controller, Get } from '@nestjs/common';
import { DbService } from './db/db.service';
@Controller()
export class HealthController {
constructor(private readonly db: DbService) {}
@Get('health')
async health() {
const database = await this.db.health();
return {
status: 'ok',
database,
};
}
}

50
src/history/dto/get-history.dto.ts Normal file
View File

@@ -0,0 +1,50 @@
import { Transform, Type } from 'class-transformer';
import {
ArrayMaxSize,
IsArray,
IsDateString,
IsIn,
IsNotEmpty,
IsNumber,
IsOptional,
IsString,
Max,
Min,
} from 'class-validator';
import { parseCommaSeparatedList } from '../../common/query-list';
const VALUE_MODES = ['avg', 'last', 'first', 'min', 'max'] as const;
export type ValueMode = (typeof VALUE_MODES)[number];
export class GetHistoryDto {
@IsNotEmpty()
@IsString()
edge!: string;
@IsOptional()
@Transform(({ value }) => parseCommaSeparatedList(value))
@IsArray()
@ArrayMaxSize(500)
@IsString({ each: true })
tags?: string[];
@IsOptional()
@IsDateString()
from?: string;
@IsOptional()
@IsDateString()
to?: string;
@IsOptional()
@Type(() => Number)
@IsNumber()
@Min(100)
@Max(2000)
targetPoints?: number;
@IsOptional()
@IsString()
@IsIn(VALUE_MODES)
valueMode?: ValueMode;
}

43
src/history/history-source.ts Normal file
View File

@@ -0,0 +1,43 @@
import { HistorySource } from './history.types';
export const RAW_HISTORY_SOURCE: HistorySource = {
name: 'raw',
table: 'history_points',
timeColumn: 'time',
intervalSeconds: 0,
};
export const AGGREGATE_HISTORY_SOURCES: HistorySource[] = [
{ name: '1m', table: 'history_1m', timeColumn: 'bucket', intervalSeconds: 60 },
{ name: '5m', table: 'history_5m', timeColumn: 'bucket', intervalSeconds: 300 },
{ name: '1h', table: 'history_1h', timeColumn: 'bucket', intervalSeconds: 3600 },
{ name: '1d', table: 'history_1d', timeColumn: 'bucket', intervalSeconds: 86400 },
];
// Подбирает самый дешевый слой данных под диапазон и целевое число точек на графике.
export function chooseHistorySource(
from: Date,
to: Date,
targetPoints: number,
): HistorySource {
const durationSeconds = Math.max((to.getTime() - from.getTime()) / 1000, 1);
const desiredSeconds = durationSeconds / targetPoints;
if (desiredSeconds < 60) {
return RAW_HISTORY_SOURCE;
}
if (desiredSeconds < 300) {
return AGGREGATE_HISTORY_SOURCES[0];
}
if (desiredSeconds < 3600) {
return AGGREGATE_HISTORY_SOURCES[1];
}
if (desiredSeconds < 86400) {
return AGGREGATE_HISTORY_SOURCES[2];
}
return AGGREGATE_HISTORY_SOURCES[3];
}

16
src/history/history.controller.ts Normal file
View File

@@ -0,0 +1,16 @@
import { Controller, Get, Header, Query } from '@nestjs/common';
import { GetHistoryDto } from './dto/get-history.dto';
import { HistoryService } from './history.service';
@Controller('history')
export class HistoryController {
constructor(private readonly history: HistoryService) {}
@Get()
@Header('Cache-Control', 'no-store')
@Header('Pragma', 'no-cache')
@Header('Expires', '0')
findSeries(@Query() query: GetHistoryDto) {
return this.history.findSeries(query);
}
}

61
src/history/history.mapper.ts Normal file
View File

@@ -0,0 +1,61 @@
import { ValueMode } from './dto/get-history.dto';
import { HistoryResponse, HistoryResponseSource, HistoryRow, HistorySeries } from './history.types';
type HistoryResponseOptions = {
edge: string;
source: HistoryResponseSource;
targetPoints: number;
rows?: HistoryRow[];
from?: Date;
to?: Date;
resolutionSeconds?: number | null;
valueMode?: ValueMode | 'raw';
};
// Собирает стабильный API response независимо от выбранного SQL-источника.
export function createHistoryResponse(options: HistoryResponseOptions): HistoryResponse {
return {
edge: options.edge,
source: options.source,
targetPoints: options.targetPoints,
series: mapHistoryRowsToSeries(options.rows ?? []),
from: options.from,
to: options.to,
resolutionSeconds: options.resolutionSeconds,
valueMode: options.valueMode,
};
}
// Группирует плоские строки БД в серии, удобные для frontend-графиков.
function mapHistoryRowsToSeries(rows: HistoryRow[]): HistorySeries[] {
const seriesMap = new Map<string, HistorySeries>();
for (const row of rows) {
const key = `${row.edge_id}\u0000${row.tag_id}`;
const series = seriesMap.get(key) ?? {
edge: row.edge_id,
tag: row.tag_id,
points: [],
};
const point = {
t: row.time.getTime(),
v: row.value,
...(row.point_count !== undefined
? {
first: row.first_value,
min: row.min_value,
max: row.max_value,
avg: row.avg_value,
last: row.last_value,
count: row.point_count,
}
: {}),
};
series.points.push(point);
seriesMap.set(key, series);
}
return Array.from(seriesMap.values());
}

10
src/history/history.module.ts Normal file
View File

@@ -0,0 +1,10 @@
import { Module } from '@nestjs/common';
import { HistoryController } from './history.controller';
import { HistoryRepository } from './history.repository';
import { HistoryService } from './history.service';
@Module({
controllers: [HistoryController],
providers: [HistoryRepository, HistoryService],
})
export class HistoryModule {}

186
src/history/history.repository.ts Normal file
View File

@@ -0,0 +1,186 @@
import { Injectable } from '@nestjs/common';
import { DbService } from '../db/db.service';
import { ValueMode } from './dto/get-history.dto';
import { HistoryRow, HistorySource } from './history.types';
const AGGREGATE_VALUE_COLUMNS: Record<ValueMode, string> = {
avg: 'avg_value',
last: 'last_value',
first: 'first_value',
min: 'min_value',
max: 'max_value',
};
const DOWNSAMPLE_PREDICATE = `
total <= $5
OR mod(rn - 1, greatest(ceil(total::numeric / $5)::integer, 1)) = 0
OR rn = total
`;
@Injectable()
export class HistoryRepository {
constructor(private readonly db: DbService) {}
// Берет последние N raw-точек по каждому tag без диапазона времени.
async findLatest(
edge: string,
tags: string[],
targetPoints: number,
): Promise<HistoryRow[]> {
const result = await this.db.query<HistoryRow>(
`
SELECT
$1::text AS edge_id,
latest.tag_id,
latest.time,
latest.value
FROM unnest($2::text[]) AS requested(tag_id)
CROSS JOIN LATERAL (
SELECT tag_id, time, value
FROM history_points
WHERE edge_id = $1
AND tag_id = requested.tag_id
ORDER BY time DESC
LIMIT $3
) AS latest
ORDER BY latest.tag_id ASC, latest.time ASC
`,
[edge, tags, targetPoints],
);
return result.rows;
}
// Читает raw-диапазон и прореживает его до targetPoints на каждую серию.
async findRawRange(
edge: string,
tags: string[],
from: Date,
to: Date,
targetPoints: number,
): Promise<HistoryRow[]> {
const result = await this.db.query<HistoryRow>(
`
WITH filtered AS (
SELECT edge_id, tag_id, time, value
FROM history_points
WHERE edge_id = $1
AND tag_id = ANY($2::text[])
AND time >= $3
AND time <= $4
),
ranked AS (
SELECT
edge_id,
tag_id,
time,
value,
row_number() OVER (PARTITION BY tag_id ORDER BY time ASC) AS rn,
count(*) OVER (PARTITION BY tag_id) AS total
FROM filtered
)
SELECT edge_id, tag_id, time, value
FROM ranked
WHERE ${DOWNSAMPLE_PREDICATE}
ORDER BY tag_id ASC, time ASC
`,
[edge, tags, from, to, targetPoints],
);
return result.rows;
}
// Читает continuous aggregate; valueMode выбирает колонку avg/last/min/max/first.
async findAggregateRange(
edge: string,
tags: string[],
from: Date,
to: Date,
targetPoints: number,
source: HistorySource,
valueMode: ValueMode,
): Promise<HistoryRow[]> {
const valueColumn = AGGREGATE_VALUE_COLUMNS[valueMode];
// Имена таблиц/колонок берутся только из whitelist-констант, не из пользовательского ввода.
const result = await this.db.query<HistoryRow>(
`
WITH filtered AS (
SELECT
edge_id,
tag_id,
${source.timeColumn} AS time,
first_value,
min_value,
max_value,
avg_value,
last_value,
point_count,
${valueColumn} AS value
FROM ${source.table}
WHERE edge_id = $1
AND tag_id = ANY($2::text[])
AND ${source.timeColumn} >= $3
AND ${source.timeColumn} <= $4
),
ranked AS (
SELECT
*,
row_number() OVER (PARTITION BY tag_id ORDER BY time ASC) AS rn,
count(*) OVER (PARTITION BY tag_id) AS total
FROM filtered
)
SELECT
edge_id,
tag_id,
time,
value,
first_value,
min_value,
max_value,
avg_value,
last_value,
point_count
FROM ranked
WHERE ${DOWNSAMPLE_PREDICATE}
ORDER BY tag_id ASC, time ASC
`,
[edge, tags, from, to, targetPoints],
);
return result.rows;
}
// Если tags не переданы, используем каталог edge_tag; fallback нужен для сырых импортов без связей.
async resolveTags(edge: string, requestedTags?: string[]): Promise<string[]> {
if (requestedTags?.length) {
return requestedTags;
}
const linked = await this.db.query<{ tag_id: string }>(
`
SELECT tag_id
FROM edge_tag
WHERE edge_id = $1
ORDER BY tag_id ASC
`,
[edge],
);
if (linked.rows.length) {
return linked.rows.map((row) => row.tag_id);
}
const distinct = await this.db.query<{ tag_id: string }>(
`
SELECT DISTINCT tag_id
FROM history_points
WHERE edge_id = $1
ORDER BY tag_id ASC
`,
[edge],
);
return distinct.rows.map((row) => row.tag_id);
}
}

92
src/history/history.service.ts Normal file
View File

@@ -0,0 +1,92 @@
import { BadRequestException, Injectable } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { getIntegerConfig } from '../common/config-number';
import { normalizeRequiredText } from '../common/normalize-text';
import { GetHistoryDto } from './dto/get-history.dto';
import { chooseHistorySource } from './history-source';
import { createHistoryResponse } from './history.mapper';
import { HistoryRepository } from './history.repository';
import { HistoryResponse } from './history.types';
@Injectable()
export class HistoryService {
constructor(
private readonly repository: HistoryRepository,
private readonly config: ConfigService,
) {}
// Use-case для графиков: выбирает latest/raw/aggregate и возвращает единый API-формат.
async findSeries(query: GetHistoryDto): Promise<HistoryResponse> {
const edge = normalizeRequiredText(query.edge, 'edge');
const targetPoints = this.getTargetPoints(query.targetPoints);
const tags = await this.repository.resolveTags(edge, query.tags);
if (!tags.length) {
return createHistoryResponse({
edge,
source: 'empty',
targetPoints,
});
}
if (!query.from && !query.to) {
const rows = await this.repository.findLatest(edge, tags, targetPoints);
return createHistoryResponse({
edge,
source: 'latest',
targetPoints,
rows,
});
}
const { from, to } = this.parseRange(query);
const source = chooseHistorySource(from, to, targetPoints);
const rows =
source.name === 'raw'
? await this.repository.findRawRange(edge, tags, from, to, targetPoints)
: await this.repository.findAggregateRange(
edge,
tags,
from,
to,
targetPoints,
source,
query.valueMode ?? 'avg',
);
return createHistoryResponse({
edge,
from,
to,
source: source.name,
resolutionSeconds: source.intervalSeconds || null,
targetPoints,
valueMode: source.name === 'raw' ? 'raw' : query.valueMode ?? 'avg',
rows,
});
}
// Частично заданный диапазон поддерживается: open-ended сторона получает безопасный default.
private parseRange(query: GetHistoryDto): { from: Date; to: Date } {
const from = query.from ? new Date(query.from) : new Date(0);
const to = query.to ? new Date(query.to) : new Date();
if (Number.isNaN(from.getTime()) || Number.isNaN(to.getTime())) {
throw new BadRequestException('Invalid from/to date.');
}
if (from > to) {
throw new BadRequestException('from cannot be greater than to.');
}
return { from, to };
}
// Ограничивает плотность ответа, чтобы API не возвращал больше точек, чем способен отрисовать UI.
private getTargetPoints(requested?: number): number {
const defaultValue = getIntegerConfig(this.config, 'HISTORY_TARGET_POINTS', 2000);
const maxValue = getIntegerConfig(this.config, 'HISTORY_MAX_TARGET_POINTS', 2000);
const value = requested ?? defaultValue;
return Math.min(Math.max(Math.floor(value), 100), maxValue);
}
}

53
src/history/history.types.ts Normal file
View File

@@ -0,0 +1,53 @@
import { ValueMode } from './dto/get-history.dto';
export type HistorySourceName = 'raw' | '1m' | '5m' | '1h' | '1d';
export type HistorySource = {
name: HistorySourceName;
table: string;
timeColumn: string;
intervalSeconds: number;
};
export type HistoryRow = {
edge_id: string;
tag_id: string;
time: Date;
value: number;
first_value?: number;
min_value?: number;
max_value?: number;
avg_value?: number;
last_value?: number;
point_count?: number;
};
export type HistoryPoint = {
t: number;
v: number;
first?: number;
min?: number;
max?: number;
avg?: number;
last?: number;
count?: number;
};
export type HistorySeries = {
edge: string;
tag: string;
points: HistoryPoint[];
};
export type HistoryResponseSource = HistorySourceName | 'empty' | 'latest';
export type HistoryResponse = {
edge: string;
source: HistoryResponseSource;
targetPoints: number;
series: HistorySeries[];
from?: Date;
to?: Date;
resolutionSeconds?: number | null;
valueMode?: ValueMode | 'raw';
};

16
src/ingest/dto/ingest-edge-values.dto.ts Normal file
View File

@@ -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>;
}

25
src/ingest/dto/ingest-point.dto.ts Normal file
View File

@@ -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;
}

32
src/ingest/ingest.controller.ts Normal file
View File

@@ -0,0 +1,32 @@
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);
}
@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);
}
}

12
src/ingest/ingest.module.ts Normal file
View File

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

234
src/ingest/ingest.service.ts Normal file
View File

@@ -0,0 +1,234 @@
import { BadRequestException, Injectable } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { getIntegerConfig } from '../common/config-number';
import { normalizeRequiredText } from '../common/normalize-text';
import { CurrentEventsService } from '../current/current-events.service';
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;
edges?: number;
tags?: number;
};
type CurrentValueRow = {
edge_id: string;
tag_id: string;
value: number;
time: Date;
updated_at: Date;
};
@Injectable()
export class IngestService {
constructor(
private readonly db: DbService,
private readonly config: ConfigService,
private readonly currentEvents: CurrentEventsService,
) {}
// Основной ingest pipeline: нормализация, справочники, raw history и latest values в одной транзакции.
async ingestPoints(points: IngestPointDto[]): Promise<IngestResult> {
if (!points.length) {
return { processed: 0 };
}
this.assertBatchSize(points.length);
const normalized = points.map((point) => this.normalizePoint(point));
const edges = Array.from(new Set(normalized.map((point) => point.edge)));
const tags = Array.from(new Set(normalized.map((point) => point.tag)));
const edgeTagPairs = Array.from(
new Map(normalized.map((point) => [`${point.edge}\u0000${point.tag}`, point])).values(),
);
await this.db.withTransaction(async (client) => {
await client.query(
`
INSERT INTO edge (id, name)
SELECT edge_id, edge_id
FROM unnest($1::text[]) AS edge_id
ON CONFLICT (id) DO NOTHING
`,
[edges],
);
await client.query(
`
INSERT INTO tag (id, name)
SELECT tag_id, tag_id
FROM unnest($1::text[]) AS tag_id
ON CONFLICT (id) DO NOTHING
`,
[tags],
);
await client.query(
`
INSERT INTO edge_tag (edge_id, tag_id)
SELECT edge_id, tag_id
FROM unnest($1::text[], $2::text[]) AS pair(edge_id, tag_id)
ON CONFLICT (edge_id, tag_id) DO NOTHING
`,
[edgeTagPairs.map((point) => point.edge), edgeTagPairs.map((point) => point.tag)],
);
await client.query(
`
INSERT INTO history_points (time, edge_id, tag_id, value)
SELECT time, edge_id, tag_id, value
FROM unnest(
$1::timestamptz[],
$2::text[],
$3::text[],
$4::double precision[]
) AS point(time, edge_id, tag_id, value)
`,
[
normalized.map((point) => point.time),
normalized.map((point) => point.edge),
normalized.map((point) => point.tag),
normalized.map((point) => point.value),
],
);
await client.query(
`
INSERT INTO current_values (edge_id, tag_id, value, time, updated_at)
SELECT DISTINCT ON (edge_id, tag_id)
edge_id,
tag_id,
value,
time,
now()
FROM unnest(
$1::text[],
$2::text[],
$3::double precision[],
$4::timestamptz[]
) AS point(edge_id, tag_id, value, time)
ORDER BY edge_id, tag_id, time DESC
ON CONFLICT (edge_id, tag_id) DO UPDATE
SET
value = EXCLUDED.value,
time = EXCLUDED.time,
updated_at = now()
WHERE EXCLUDED.time >= current_values.time
`,
[
normalized.map((point) => point.edge),
normalized.map((point) => point.tag),
normalized.map((point) => point.value),
normalized.map((point) => point.time),
],
);
});
await this.publishCurrentChanges(edgeTagPairs);
return {
processed: normalized.length,
edges: edges.length,
tags: tags.length,
};
}
private async publishCurrentChanges(points: NormalizedPoint[]): Promise<void> {
if (!points.length) {
return;
}
const result = await this.db.query<CurrentValueRow>(
`
SELECT cv.edge_id, cv.tag_id, cv.value, cv.time, cv.updated_at
FROM current_values AS cv
JOIN (
SELECT DISTINCT edge_id, tag_id
FROM unnest($1::text[], $2::text[]) AS pair(edge_id, tag_id)
) AS changed
ON changed.edge_id = cv.edge_id
AND changed.tag_id = cv.tag_id
`,
[points.map((point) => point.edge), points.map((point) => point.tag)],
);
const byEdge = new Map<string, CurrentValueRow[]>();
result.rows.forEach((row) => {
const rows = byEdge.get(row.edge_id) ?? [];
rows.push(row);
byEdge.set(row.edge_id, rows);
});
byEdge.forEach((rows, edge) => {
this.currentEvents.publish({
edge,
items: rows.map((row) => ({
edge: row.edge_id,
tag: row.tag_id,
value: row.value,
time: row.time,
updatedAt: row.updated_at,
})),
});
});
}
// Ограничивает размер одного запроса, чтобы один producer не занял память и соединение надолго.
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.`);
}
}
// Приводит входную точку к внутреннему формату и отсекает некорректные edge/tag/value/time.
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 time или Unix milliseconds; отсутствие валидного времени считается ошибкой данных.
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.');
}
}

34
src/main.ts Normal file
View File

@@ -0,0 +1,34 @@
import { ValidationPipe } from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import compression from 'compression';
import { AppModule } from './app.module';
async function bootstrap() {
const app = await NestFactory.create(AppModule);
app.use(compression());
const corsAllowedOrigins = process.env.CORS_ALLOWED_ORIGINS
? process.env.CORS_ALLOWED_ORIGINS.split(',').map((origin) => origin.trim()).filter(Boolean)
: true;
// CORS настраивается через env, чтобы один build работал локально и в окружениях деплоя.
app.enableCors({
origin: corsAllowedOrigins,
credentials: true,
methods: ['GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS'],
allowedHeaders: ['Content-Type', 'Authorization', 'x-api-key'],
});
// Whitelist защищает API от лишних полей в DTO и делает контракты запросов строгими.
app.useGlobalPipes(
new ValidationPipe({
transform: true,
whitelist: true,
forbidNonWhitelisted: true,
}),
);
await app.listen(process.env.PORT ?? 3100);
}
void bootstrap();

62
src/migrations/migrate.ts Normal file
View File

@@ -0,0 +1,62 @@
import { readdir, readFile } from 'node:fs/promises';
import { join } from 'node:path';
import { Pool } from 'pg';
async function main() {
const connectionString = process.env.DATABASE_URL;
if (!connectionString) {
throw new Error('DATABASE_URL is required.');
}
const pool = new Pool({
connectionString,
application_name: 'drill-cloud-v2-migrate',
});
let lockAcquired = false;
try {
// Защищает от параллельного применения миграций несколькими инстансами API.
await pool.query('SELECT pg_advisory_lock(hashtext($1))', ['drill-cloud-v2:migrate']);
lockAcquired = true;
await pool.query(`
CREATE TABLE IF NOT EXISTS app_migrations (
id text PRIMARY KEY,
applied_at timestamptz NOT NULL DEFAULT now()
)
`);
const migrationsDir = join(process.cwd(), 'migrations');
// Порядок определяется именем файла: 001_..., 002_..., и т.д.
const files = (await readdir(migrationsDir))
.filter((file) => file.endsWith('.sql'))
.sort((left, right) => left.localeCompare(right));
for (const file of files) {
const applied = await pool.query<{ exists: boolean }>(
'SELECT EXISTS (SELECT 1 FROM app_migrations WHERE id = $1)',
[file],
);
if (applied.rows[0]?.exists) {
console.log(`skip ${file}`);
continue;
}
const sql = await readFile(join(migrationsDir, file), 'utf8');
console.log(`apply ${file}`);
await pool.query(sql);
await pool.query('INSERT INTO app_migrations (id) VALUES ($1)', [file]);
}
} finally {
if (lockAcquired) {
await pool.query('SELECT pg_advisory_unlock(hashtext($1))', ['drill-cloud-v2:migrate']);
}
await pool.end();
}
}
void main().catch((error) => {
console.error(error);
process.exit(1);
});

4
tsconfig.build.json Normal file
View File

@@ -0,0 +1,4 @@
{
"extends": "./tsconfig.json",
"exclude": ["node_modules", "dist", "test", "**/*spec.ts"]
}

21
tsconfig.json Normal file
View File

@@ -0,0 +1,21 @@
{
"compilerOptions": {
"module": "commonjs",
"moduleResolution": "node",
"esModuleInterop": true,
"declaration": true,
"removeComments": true,
"emitDecoratorMetadata": true,
"experimentalDecorators": true,
"allowSyntheticDefaultImports": true,
"target": "ES2023",
"sourceMap": true,
"outDir": "./dist",
"baseUrl": "./",
"incremental": true,
"skipLibCheck": true,
"strict": true,
"forceConsistentCasingInFileNames": true,
"noFallthroughCasesInSwitch": true
}
}