Compare commits
2 Commits
9eefe50401
...
8bcc12ec96
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8bcc12ec96 | ||
|
|
daa3407b9d |
@@ -8,14 +8,16 @@ describes from a usage/dashboard perspective.
|
||||
## Overview
|
||||
|
||||
The broker (`eclipse-mosquitto`, service `mqtt` / container `ul-hub-mqtt`)
|
||||
exposes four listeners, each with a different trust model:
|
||||
exposes three listeners, each with a different trust model:
|
||||
|
||||
| Port | Protocol | Auth | Who it's for |
|
||||
|------|----------|------|---------------|
|
||||
| `1883` | MQTT (plaintext) | username/password | internal services (e.g. the Laravel subscriber, Python publisher) |
|
||||
| `9001` (mapped to host `9005`) | MQTT over WebSocket | none (anonymous) | browser clients (dashboard) |
|
||||
| `8883` | MQTT over TLS | **client certificate** | field devices |
|
||||
| `8884` | MQTT over TLS | username/password (server cert only) | administrators |
|
||||
| `8884` | MQTT over TLS | username/password (server cert only) | scoped app clients and administrators |
|
||||
|
||||
The former anonymous WebSocket listener on `9001` is disabled. The current web portal receives
|
||||
live updates from the backend rather than connecting directly to Mosquitto.
|
||||
|
||||
Device authentication happens on **port 8883**. A device presents a client
|
||||
certificate signed by the app's own Certificate Authority (CA); Mosquitto
|
||||
@@ -123,6 +125,10 @@ since they all share `acl_file /mosquitto/config/devices.acl`:
|
||||
# Certificate CN becomes the MQTT username — restrict each device to its own namespace.
|
||||
pattern readwrite devices/%u/#
|
||||
|
||||
# App username is orgId; app clients may publish durable points and read acks only in that org.
|
||||
pattern write ul/%u/app/+/log/points
|
||||
pattern read ul/%u/app/+/ack
|
||||
|
||||
# Admin: <username>
|
||||
user <username>
|
||||
topic readwrite #
|
||||
@@ -169,9 +175,9 @@ Managed via the same `/certificates` page:
|
||||
| Client | Listener | Auth | Can publish/subscribe |
|
||||
|---|---|---|---|
|
||||
| Field device (cert CN = serial) | 8883 (TLS) | client cert | `devices/<serial>/#` only |
|
||||
| App (username = orgId) | 8884 (TLS) | scoped per-org username/password | publish `ul/<orgId>/app/+/log/points`; read `ul/<orgId>/app/+/ack` |
|
||||
| Internal service (e.g. subscriber) | 1883 | username/password | depends on ACL entry for that username — none defined by default beyond `devices/%u/#`, so a plain username with no matching device row is effectively scoped to `devices/<username>/#` too, unless added as an admin |
|
||||
| Admin | 8884 (TLS) or 1883 | username/password | `#` and `$SYS/#` (everything) |
|
||||
| Browser dashboard | 9001/9005 (WebSocket) | anonymous | no ACL applied — `allow_anonymous true`, so effectively unrestricted; treat this listener as untrusted/read-only in front-end code |
|
||||
|
||||
## Known gaps
|
||||
|
||||
@@ -180,5 +186,5 @@ Managed via the same `/certificates` page:
|
||||
needs to be production-grade.
|
||||
- No automatic reload of `mosquitto.conf`/ACL/passwd changes — every
|
||||
provisioning action requires a manual `docker compose restart mqtt`.
|
||||
- The WebSocket listener (9001) is fully anonymous with no ACL, so anything
|
||||
reachable on port 9005 should be treated as public.
|
||||
- Sprint 2's per-org app credential permits ack visibility across clients in the same org. This
|
||||
accepted interim limitation is removed when OIDC-derived per-client broker identity lands.
|
||||
|
||||
16
backend/jest.config.js
Normal file
16
backend/jest.config.js
Normal file
@@ -0,0 +1,16 @@
|
||||
/**
|
||||
* Jest config for the UlHub backend (NestJS).
|
||||
* ts-jest transpiles the TypeScript sources; decorators/metadata are honored via tsconfig.
|
||||
* Specs live next to sources as *.spec.ts (unit) — e2e would go under test/ with a separate config.
|
||||
*/
|
||||
module.exports = {
|
||||
moduleFileExtensions: ['js', 'json', 'ts'],
|
||||
rootDir: 'src',
|
||||
testRegex: '.*\\.spec\\.ts$',
|
||||
transform: {
|
||||
'^.+\\.ts$': ['ts-jest', { tsconfig: '<rootDir>/../tsconfig.json' }],
|
||||
},
|
||||
collectCoverageFrom: ['**/*.(t|j)s'],
|
||||
coverageDirectory: '../coverage',
|
||||
testEnvironment: 'node',
|
||||
};
|
||||
3527
backend/package-lock.json
generated
3527
backend/package-lock.json
generated
File diff suppressed because it is too large
Load Diff
@@ -8,7 +8,9 @@
|
||||
"build": "nest build",
|
||||
"format": "prettier --write \"src/**/*.ts\"",
|
||||
"postinstall": "prisma generate",
|
||||
"db:seed": "ts-node prisma/seed.ts"
|
||||
"db:seed": "ts-node prisma/seed.ts",
|
||||
"test": "jest",
|
||||
"test:ci": "jest --ci --runInBand"
|
||||
},
|
||||
"prisma": {
|
||||
"seed": "ts-node prisma/seed.ts"
|
||||
@@ -44,9 +46,12 @@
|
||||
"@types/cookie-parser": "^1.4.7",
|
||||
"@types/express": "^4.17.17",
|
||||
"@types/node": "^20.11.0",
|
||||
"@types/jest": "^29.5.12",
|
||||
"@types/passport-jwt": "^4.0.1",
|
||||
"@types/ws": "^8.5.10",
|
||||
"jest": "^29.7.0",
|
||||
"prettier": "^3.0.0",
|
||||
"ts-jest": "^29.2.5",
|
||||
"prisma": "^6.10.0",
|
||||
"ts-node": "^10.9.1",
|
||||
"typescript": "^5.5.0"
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
-- App-path ingest idempotency + provenance (Sprint 2, S2-a; SRS-SYN-2/7, §3.4.2).
|
||||
-- Adds the UUIDv7 idempotency key and forward-compat provenance tags to locate_points.
|
||||
|
||||
-- CreateEnum
|
||||
CREATE TYPE "PointOrigin" AS ENUM ('APP', 'LOCATOR', 'MAGLINK');
|
||||
|
||||
-- CreateEnum
|
||||
CREATE TYPE "PointUploadPath" AS ENUM ('APP_MQTT', 'DEVICE_MQTT', 'REST_BATCH');
|
||||
|
||||
-- AlterTable
|
||||
ALTER TABLE "locate_points" ADD COLUMN "pointId" TEXT,
|
||||
ADD COLUMN "origin" "PointOrigin" NOT NULL DEFAULT 'LOCATOR',
|
||||
ADD COLUMN "uploadPath" "PointUploadPath",
|
||||
ADD COLUMN "originClientId" TEXT,
|
||||
ADD COLUMN "createdAt" TIMESTAMPTZ(6);
|
||||
|
||||
-- CreateIndex
|
||||
CREATE UNIQUE INDEX "locate_points_pointId_key" ON "locate_points"("pointId");
|
||||
|
||||
-- CreateIndex
|
||||
CREATE INDEX "locate_points_jobId_createdAt_idx" ON "locate_points"("jobId", "createdAt");
|
||||
@@ -53,6 +53,22 @@ enum LocateMode {
|
||||
SONDE
|
||||
}
|
||||
|
||||
// Which producer a point originated from (telemetry-schema.md `origin`). Tagged now
|
||||
// for SYN-6 path arbitration; no arbitration LOGIC runs yet (deferred, sprint-2 scope).
|
||||
enum PointOrigin {
|
||||
APP
|
||||
LOCATOR
|
||||
MAGLINK
|
||||
}
|
||||
|
||||
// Which transport carried the point to the cloud (telemetry-schema.md `uploadPath`).
|
||||
// Same forward-compat tagging as PointOrigin.
|
||||
enum PointUploadPath {
|
||||
APP_MQTT
|
||||
DEVICE_MQTT
|
||||
REST_BATCH
|
||||
}
|
||||
|
||||
model Organization {
|
||||
id String @id @default(cuid())
|
||||
name String
|
||||
@@ -185,6 +201,25 @@ model LocatePoint {
|
||||
id BigInt @id @default(autoincrement())
|
||||
jobId String
|
||||
deviceId String?
|
||||
|
||||
// Client-generated UUIDv7, the idempotency key across every upload path
|
||||
// (app MQTT, device-direct MQTT, REST batch — SRS §3.4.2 / telemetry-schema.md).
|
||||
// A replayed UUID is ingested exactly once (unique constraint below). Nullable:
|
||||
// the legacy device-direct path (LogIngest/PointsIngest) predates it and does not
|
||||
// supply one yet; the app path (SYN-2) always does.
|
||||
pointId String? @unique
|
||||
|
||||
// Provenance tags (telemetry-schema.md). Set on the app path now; carried for
|
||||
// forward-compat with SYN-6 arbitration, which does not run yet.
|
||||
origin PointOrigin @default(LOCATOR)
|
||||
uploadPath PointUploadPath?
|
||||
originClientId String?
|
||||
|
||||
// Event time asserted by the producer (record creation, telemetry-schema.md
|
||||
// `createdAt`). Ingest orders and dedups by this, NOT by arrival (`receivedAt`),
|
||||
// so replays and out-of-order delivery converge to the same stored ordering.
|
||||
createdAt DateTime? @db.Timestamptz(6)
|
||||
|
||||
lat Decimal @db.Decimal(10, 8)
|
||||
lng Decimal @db.Decimal(11, 8)
|
||||
altitude Decimal? @db.Decimal(8, 3)
|
||||
@@ -217,7 +252,11 @@ model LocatePoint {
|
||||
job Job @relation(fields: [jobId], references: [id], onDelete: Cascade)
|
||||
device Device? @relation(fields: [deviceId], references: [id], onDelete: SetNull)
|
||||
|
||||
// (jobId, recordedAt) serves GPS-time reads; (jobId, createdAt) serves the
|
||||
// event-time ordering the app path and portal read use (SRS-SYN-2). pointId is
|
||||
// already uniquely indexed above — that index also backs the dedup lookup.
|
||||
@@index([jobId, recordedAt])
|
||||
@@index([jobId, createdAt])
|
||||
@@map("locate_points")
|
||||
}
|
||||
|
||||
|
||||
27
backend/src/app.service.spec.ts
Normal file
27
backend/src/app.service.spec.ts
Normal file
@@ -0,0 +1,27 @@
|
||||
import { Test, TestingModule } from '@nestjs/testing';
|
||||
import { AppService } from './app.service';
|
||||
|
||||
// Smoke test for the QA gate (S1-f, trace: SRS §6 gate / NFR-8).
|
||||
// Exercises Nest DI wiring end-to-end for a pure service so a green run proves the
|
||||
// toolchain (ts-jest + @nestjs/testing) is functional. Real coverage grows from here.
|
||||
describe('AppService (smoke)', () => {
|
||||
let service: AppService;
|
||||
|
||||
beforeEach(async () => {
|
||||
const moduleRef: TestingModule = await Test.createTestingModule({
|
||||
providers: [AppService],
|
||||
}).compile();
|
||||
service = moduleRef.get<AppService>(AppService);
|
||||
});
|
||||
|
||||
it('is resolvable from the DI container', () => {
|
||||
expect(service).toBeDefined();
|
||||
});
|
||||
|
||||
it('returns the API welcome payload', () => {
|
||||
expect(service.getHello()).toEqual({
|
||||
message: 'Welcome to UlHub API',
|
||||
docs: 'GET /api',
|
||||
});
|
||||
});
|
||||
});
|
||||
315
backend/src/ingest/app-log-ingest.service.spec.ts
Normal file
315
backend/src/ingest/app-log-ingest.service.spec.ts
Normal file
@@ -0,0 +1,315 @@
|
||||
import { Test } from '@nestjs/testing';
|
||||
import { readFileSync } from 'fs';
|
||||
import { resolve } from 'path';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { RealtimeService } from '../realtime/realtime.service';
|
||||
import { AppLogAck } from './dto/app-messages.dto';
|
||||
import { AppLogIngestService } from './app-log-ingest.service';
|
||||
import { MqttClientService } from './mqtt-client.service';
|
||||
|
||||
// Unit coverage for the durable app-log ingest path (S2-a, SRS-SYN-2/7, §3.4.2).
|
||||
// Prisma is an in-memory fake keyed on pointId so idempotency is exercised for real
|
||||
// (a replayed UUID must be stored exactly once), without a database. The broker is a
|
||||
// spy so we can assert the application-level ack shape and topic.
|
||||
describe('AppLogIngestService', () => {
|
||||
const ORG = 'org_alpha';
|
||||
const CLIENT = 'app_client_1';
|
||||
const JOB = { id: 'job_1', orgId: ORG, ticketNumber: 'T-1' };
|
||||
|
||||
let service: AppLogIngestService;
|
||||
let published: { topic: string; payload: AppLogAck }[];
|
||||
let store: Map<string, { pointId: string; jobId: string; createdAt: Date }>;
|
||||
|
||||
const point = (
|
||||
pointId: string,
|
||||
createdAt: string,
|
||||
extra: Record<string, unknown> = {},
|
||||
) => ({
|
||||
pointId,
|
||||
createdAt,
|
||||
origin: 'APP',
|
||||
uploadPath: 'APP_MQTT',
|
||||
lat: 40,
|
||||
lng: -80,
|
||||
ts: createdAt,
|
||||
qualityFlag: 'IN_SPEC',
|
||||
...extra,
|
||||
});
|
||||
|
||||
const msg = (
|
||||
points: object[],
|
||||
job: { jobId?: string; ticket?: string } = { jobId: JOB.id },
|
||||
) => JSON.stringify({ schemaVersion: '1', ...job, points });
|
||||
|
||||
const lastAck = () => published[published.length - 1].payload;
|
||||
|
||||
beforeEach(async () => {
|
||||
published = [];
|
||||
store = new Map();
|
||||
|
||||
const prisma = {
|
||||
organization: {
|
||||
findUnique: jest.fn(({ where }: { where: { id: string } }) =>
|
||||
Promise.resolve(where.id === ORG ? { id: ORG } : null),
|
||||
),
|
||||
},
|
||||
job: {
|
||||
findFirst: jest.fn(
|
||||
({ where }: { where: { id: string; orgId: string } }) =>
|
||||
Promise.resolve(
|
||||
where.id === JOB.id && where.orgId === ORG ? JOB : null,
|
||||
),
|
||||
),
|
||||
findUnique: jest.fn(() => Promise.resolve(null)),
|
||||
create: jest.fn(() => Promise.resolve(JOB)),
|
||||
},
|
||||
locatePoint: {
|
||||
findMany: jest.fn(
|
||||
({ where }: { where: { pointId: { in: string[] } } }) =>
|
||||
Promise.resolve(
|
||||
where.pointId.in
|
||||
.filter((id) => store.has(id))
|
||||
.map((pointId) => ({ pointId })),
|
||||
),
|
||||
),
|
||||
createManyAndReturn: jest.fn(
|
||||
({
|
||||
data,
|
||||
}: {
|
||||
data: { pointId: string; jobId: string; createdAt: Date }[];
|
||||
}) => {
|
||||
const inserted: unknown[] = [];
|
||||
for (const row of data) {
|
||||
if (store.has(row.pointId)) {
|
||||
continue; // skipDuplicates
|
||||
}
|
||||
store.set(row.pointId, row);
|
||||
inserted.push({
|
||||
...row,
|
||||
id: BigInt(store.size),
|
||||
lat: 40,
|
||||
lng: -80,
|
||||
altitude: null,
|
||||
utilityType: 'UNKNOWN',
|
||||
sequence: null,
|
||||
fixType: 'NONE',
|
||||
hAccuracy: null,
|
||||
vAccuracy: null,
|
||||
satellites: null,
|
||||
hdop: null,
|
||||
depth: null,
|
||||
frequencyHz: null,
|
||||
currentMa: null,
|
||||
signalDb: null,
|
||||
gainDb: null,
|
||||
locateMode: null,
|
||||
phaseDeg: null,
|
||||
compassDeg: null,
|
||||
distortionPct: null,
|
||||
recordedAt: row.createdAt,
|
||||
receivedAt: row.createdAt,
|
||||
origin: 'APP',
|
||||
uploadPath: 'APP_MQTT',
|
||||
originClientId: CLIENT,
|
||||
deviceId: null,
|
||||
});
|
||||
}
|
||||
return Promise.resolve(inserted);
|
||||
},
|
||||
),
|
||||
},
|
||||
};
|
||||
|
||||
const mqtt = {
|
||||
publish: jest.fn((topic: string, payload: AppLogAck) =>
|
||||
published.push({ topic, payload }),
|
||||
),
|
||||
};
|
||||
|
||||
const moduleRef = await Test.createTestingModule({
|
||||
providers: [
|
||||
AppLogIngestService,
|
||||
{ provide: PrismaService, useValue: prisma },
|
||||
{ provide: RealtimeService, useValue: { publish: jest.fn() } },
|
||||
{ provide: MqttClientService, useValue: mqtt },
|
||||
],
|
||||
}).compile();
|
||||
|
||||
service = moduleRef.get(AppLogIngestService);
|
||||
});
|
||||
|
||||
it('ingests a fresh point once and acks it accepted on the client namespace', async () => {
|
||||
await service.handle(
|
||||
ORG,
|
||||
CLIENT,
|
||||
msg([
|
||||
point('018f1a00-0000-7000-8000-000000000001', '2026-08-21T10:00:00Z'),
|
||||
]),
|
||||
);
|
||||
|
||||
expect(store.size).toBe(1);
|
||||
expect(published[0].topic).toBe(`ul/${ORG}/app/${CLIENT}/ack`);
|
||||
const ack = lastAck();
|
||||
expect(ack.results).toEqual([
|
||||
{ pointId: '018f1a00-0000-7000-8000-000000000001', outcome: 'ACCEPTED' },
|
||||
]);
|
||||
});
|
||||
|
||||
it('consumes the shared app fixture and emits the shared ack fixture', async () => {
|
||||
const fixtureRoot = resolve(
|
||||
__dirname,
|
||||
'../../../../meta/contracts/fixtures',
|
||||
);
|
||||
const publishFixture = readFileSync(
|
||||
resolve(fixtureRoot, 'app-log-points-v1.json'),
|
||||
'utf8',
|
||||
);
|
||||
const expectedAck = JSON.parse(
|
||||
readFileSync(resolve(fixtureRoot, 'app-log-ack-v1.json'), 'utf8'),
|
||||
);
|
||||
|
||||
await service.handle(ORG, CLIENT, publishFixture);
|
||||
|
||||
expect(store.size).toBe(1);
|
||||
expect(lastAck()).toEqual(expectedAck);
|
||||
});
|
||||
|
||||
it('ingests a replayed UUID exactly once (idempotency) and acks it as duplicate', async () => {
|
||||
const p = point(
|
||||
'018f1a00-0000-7000-8000-000000000002',
|
||||
'2026-08-21T10:00:00Z',
|
||||
);
|
||||
await service.handle(ORG, CLIENT, msg([p]));
|
||||
await service.handle(ORG, CLIENT, msg([p])); // replay
|
||||
|
||||
expect(store.size).toBe(1); // stored exactly once
|
||||
const ack = lastAck();
|
||||
expect(ack.results).toEqual([
|
||||
{ pointId: '018f1a00-0000-7000-8000-000000000002', outcome: 'DUPLICATE' },
|
||||
]);
|
||||
});
|
||||
|
||||
it('orders a batch by event time, not array order', async () => {
|
||||
const later = point(
|
||||
'018f1a00-0000-7000-8000-00000000000a',
|
||||
'2026-08-21T10:05:00Z',
|
||||
);
|
||||
const earlier = point(
|
||||
'018f1a00-0000-7000-8000-00000000000b',
|
||||
'2026-08-21T10:01:00Z',
|
||||
);
|
||||
await service.handle(ORG, CLIENT, msg([later, earlier]));
|
||||
|
||||
const rows = [...store.values()];
|
||||
expect(rows.map((r) => r.pointId)).toEqual([
|
||||
earlier.pointId,
|
||||
later.pointId,
|
||||
]);
|
||||
});
|
||||
|
||||
it('rejects a point whose jobId belongs to another org (cross-namespace backstop)', async () => {
|
||||
await service.handle(
|
||||
ORG,
|
||||
CLIENT,
|
||||
msg(
|
||||
[point('018f1a00-0000-7000-8000-00000000000c', '2026-08-21T10:00:00Z')],
|
||||
{ jobId: 'job_of_org_beta' },
|
||||
),
|
||||
);
|
||||
|
||||
expect(store.size).toBe(0);
|
||||
const ack = lastAck();
|
||||
expect(ack.results).toEqual([
|
||||
{
|
||||
pointId: '018f1a00-0000-7000-8000-00000000000c',
|
||||
outcome: 'REJECTED',
|
||||
reasonCode: 'UNKNOWN_JOB_OR_WRONG_ORG',
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('drops (no ack) a message on an unknown org namespace', async () => {
|
||||
await service.handle(
|
||||
'org_ghost',
|
||||
CLIENT,
|
||||
msg([
|
||||
point('018f1a00-0000-7000-8000-00000000000d', '2026-08-21T10:00:00Z'),
|
||||
]),
|
||||
);
|
||||
expect(published).toHaveLength(0);
|
||||
expect(store.size).toBe(0);
|
||||
});
|
||||
|
||||
it('does not emit an ambiguous ack when malformed JSON has no usable pointId', async () => {
|
||||
await service.handle(ORG, CLIENT, '{ not json');
|
||||
expect(published).toHaveLength(0);
|
||||
});
|
||||
|
||||
it('rejects a schema-invalid point but names its UUID in the ack', async () => {
|
||||
// Missing required createdAt + bad lat → validation fails; pointId still surfaced.
|
||||
await service.handle(
|
||||
ORG,
|
||||
CLIENT,
|
||||
msg([
|
||||
{
|
||||
pointId: '018f1a00-0000-7000-8000-00000000000e',
|
||||
lat: 999,
|
||||
lng: 0,
|
||||
ts: '2026-08-21T10:00:00Z',
|
||||
},
|
||||
]),
|
||||
);
|
||||
const ack = lastAck();
|
||||
expect(store.size).toBe(0);
|
||||
expect(ack.results).toEqual([
|
||||
{
|
||||
pointId: '018f1a00-0000-7000-8000-00000000000e',
|
||||
outcome: 'REJECTED',
|
||||
reasonCode: 'VALIDATION_ERROR',
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('accepts a valid point while rejecting a malformed neighbor in the same batch', async () => {
|
||||
const bad = point(
|
||||
'018f1a00-0000-7000-8000-00000000000f',
|
||||
'2026-08-21T10:00:00Z',
|
||||
{ lat: 999 },
|
||||
);
|
||||
const good = point(
|
||||
'018f1a00-0000-7000-8000-000000000010',
|
||||
'2026-08-21T10:00:01Z',
|
||||
);
|
||||
await service.handle(ORG, CLIENT, msg([bad, good]));
|
||||
|
||||
expect(store.size).toBe(1);
|
||||
expect(lastAck().results).toEqual(
|
||||
expect.arrayContaining([
|
||||
{
|
||||
pointId: bad.pointId,
|
||||
outcome: 'REJECTED',
|
||||
reasonCode: 'VALIDATION_ERROR',
|
||||
},
|
||||
{ pointId: good.pointId, outcome: 'ACCEPTED' },
|
||||
]),
|
||||
);
|
||||
});
|
||||
|
||||
it('rejects a non-v7 UUID even when it is otherwise a valid UUID', async () => {
|
||||
const uuidV4 = '550e8400-e29b-41d4-a716-446655440000';
|
||||
await service.handle(
|
||||
ORG,
|
||||
CLIENT,
|
||||
msg([point(uuidV4, '2026-08-21T10:00:00Z')]),
|
||||
);
|
||||
|
||||
expect(store.size).toBe(0);
|
||||
expect(lastAck().results).toEqual([
|
||||
{
|
||||
pointId: uuidV4,
|
||||
outcome: 'REJECTED',
|
||||
reasonCode: 'POINT_ID_NOT_UUIDV7',
|
||||
},
|
||||
]);
|
||||
});
|
||||
});
|
||||
277
backend/src/ingest/app-log-ingest.service.ts
Normal file
277
backend/src/ingest/app-log-ingest.service.ts
Normal file
@@ -0,0 +1,277 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
import { GpsFixType, Prisma } from '@prisma/client';
|
||||
import { plainToInstance } from 'class-transformer';
|
||||
import { validate } from 'class-validator';
|
||||
import { toPointDto } from '../points/points.service';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { RealtimeService } from '../realtime/realtime.service';
|
||||
import {
|
||||
AppFixType,
|
||||
AppLogAck,
|
||||
AppLogAckResult,
|
||||
AppLogPointDto,
|
||||
AppLogPointsMessageDto,
|
||||
} from './dto/app-messages.dto';
|
||||
import { MqttClientService } from './mqtt-client.service';
|
||||
|
||||
const SCHEMA_VERSION = '1';
|
||||
const UUID_V7 =
|
||||
/^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
|
||||
|
||||
// Durable app-originated ingest for ul/{orgId}/app/{clientId}/log/points. The exact v1
|
||||
// publish/ack envelopes are frozen in meta/contracts/telemetry-schema.md.
|
||||
@Injectable()
|
||||
export class AppLogIngestService {
|
||||
private readonly logger = new Logger(AppLogIngestService.name);
|
||||
|
||||
constructor(
|
||||
private readonly prisma: PrismaService,
|
||||
private readonly realtime: RealtimeService,
|
||||
private readonly mqttClient: MqttClientService,
|
||||
) {}
|
||||
|
||||
async handle(
|
||||
orgId: string,
|
||||
clientId: string,
|
||||
rawPayload: string,
|
||||
): Promise<void> {
|
||||
const org = await this.prisma.organization.findUnique({
|
||||
where: { id: orgId },
|
||||
select: { id: true },
|
||||
});
|
||||
if (!org) {
|
||||
this.logger.warn(
|
||||
`app-log for unknown org "${orgId}" (client ${clientId}) — dropped`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(rawPayload);
|
||||
} catch {
|
||||
// There is no trustworthy pointId to acknowledge. Raw app payloads are redacted by the
|
||||
// router; retain/log on the app side rather than sending a null/ambiguous acknowledgement.
|
||||
this.logger.warn(
|
||||
`invalid JSON from app ${orgId}/${clientId} — no acknowledgement emitted`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const msg = plainToInstance(AppLogPointsMessageDto, parsed as object);
|
||||
const rawPoints = Array.isArray((parsed as { points?: unknown[] })?.points)
|
||||
? (parsed as { points: unknown[] }).points
|
||||
: [];
|
||||
const usableIds = rawPoints
|
||||
.map((value) =>
|
||||
typeof (value as { pointId?: unknown })?.pointId === 'string'
|
||||
? (value as { pointId: string }).pointId
|
||||
: null,
|
||||
)
|
||||
.filter((value): value is string => value !== null);
|
||||
|
||||
const hasJobId = typeof msg.jobId === 'string' && msg.jobId.length > 0;
|
||||
const hasTicket = typeof msg.ticket === 'string' && msg.ticket.length > 0;
|
||||
const envelopeValid =
|
||||
msg.schemaVersion === SCHEMA_VERSION &&
|
||||
rawPoints.length > 0 &&
|
||||
rawPoints.length <= 500 &&
|
||||
hasJobId !== hasTicket;
|
||||
if (!envelopeValid) {
|
||||
this.publishAck(
|
||||
orgId,
|
||||
clientId,
|
||||
usableIds.map((pointId) => ({
|
||||
pointId,
|
||||
outcome: 'REJECTED',
|
||||
reasonCode: 'ENVELOPE_VALIDATION_ERROR',
|
||||
})),
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
// Validate independently so one malformed capture does not discard valid captures in the
|
||||
// same batch (LOG-7). Invalid UUID strings can still be named in a rejection; missing IDs
|
||||
// cannot be safely correlated and therefore remain unacknowledged.
|
||||
const results: AppLogAckResult[] = [];
|
||||
const candidates: AppLogPointDto[] = [];
|
||||
for (const rawPoint of rawPoints) {
|
||||
const point = plainToInstance(AppLogPointDto, rawPoint as object);
|
||||
const pointId = typeof point.pointId === 'string' ? point.pointId : null;
|
||||
const errors = await validate(point, { whitelist: true });
|
||||
if (!pointId) {
|
||||
continue;
|
||||
}
|
||||
if (errors.length > 0 || !UUID_V7.test(pointId)) {
|
||||
results.push({
|
||||
pointId,
|
||||
outcome: 'REJECTED',
|
||||
reasonCode: UUID_V7.test(pointId)
|
||||
? 'VALIDATION_ERROR'
|
||||
: 'POINT_ID_NOT_UUIDV7',
|
||||
});
|
||||
continue;
|
||||
}
|
||||
candidates.push(point);
|
||||
}
|
||||
|
||||
const job = await this.resolveJob(orgId, clientId, msg);
|
||||
if (!job) {
|
||||
results.push(
|
||||
...candidates.map((point) => ({
|
||||
pointId: point.pointId,
|
||||
outcome: 'REJECTED' as const,
|
||||
reasonCode: 'UNKNOWN_JOB_OR_WRONG_ORG',
|
||||
})),
|
||||
);
|
||||
this.publishAck(orgId, clientId, results);
|
||||
return;
|
||||
}
|
||||
|
||||
// Collapse duplicates inside this delivery, then use producer event time rather than
|
||||
// arrival/array order. The first valid occurrence wins; repeated IDs are safe to release.
|
||||
const byPointId = new Map<string, AppLogPointDto>();
|
||||
for (const point of candidates) {
|
||||
if (byPointId.has(point.pointId)) {
|
||||
results.push({
|
||||
pointId: point.pointId,
|
||||
outcome: 'DUPLICATE',
|
||||
reasonCode: 'DUPLICATE_IN_BATCH',
|
||||
});
|
||||
} else {
|
||||
byPointId.set(point.pointId, point);
|
||||
}
|
||||
}
|
||||
const ordered = [...byPointId.values()].sort(
|
||||
(a, b) => +new Date(a.createdAt) - +new Date(b.createdAt),
|
||||
);
|
||||
|
||||
const seen = ordered.length
|
||||
? await this.prisma.locatePoint.findMany({
|
||||
where: { pointId: { in: ordered.map((point) => point.pointId) } },
|
||||
select: { pointId: true },
|
||||
})
|
||||
: [];
|
||||
const seenIds = new Set(seen.map((row) => row.pointId));
|
||||
const fresh = ordered.filter((point) => !seenIds.has(point.pointId));
|
||||
for (const point of ordered.filter((candidate) =>
|
||||
seenIds.has(candidate.pointId),
|
||||
)) {
|
||||
results.push({ pointId: point.pointId, outcome: 'DUPLICATE' });
|
||||
}
|
||||
|
||||
let stored: Awaited<
|
||||
ReturnType<typeof this.prisma.locatePoint.createManyAndReturn>
|
||||
> = [];
|
||||
if (fresh.length > 0) {
|
||||
stored = await this.prisma.locatePoint.createManyAndReturn({
|
||||
data: fresh.map((point) => ({
|
||||
jobId: job.id,
|
||||
deviceId: null,
|
||||
pointId: point.pointId,
|
||||
origin: 'APP' as const,
|
||||
uploadPath: 'APP_MQTT' as const,
|
||||
originClientId: clientId,
|
||||
createdAt: new Date(point.createdAt),
|
||||
lat: point.lat,
|
||||
lng: point.lng,
|
||||
altitude: point.alt,
|
||||
utilityType: point.utility,
|
||||
sequence: point.seq,
|
||||
fixType: this.toStoredFix(point.fix),
|
||||
hAccuracy: point.hAcc,
|
||||
vAccuracy: point.vAcc,
|
||||
satellites: point.sats,
|
||||
hdop: point.hdop,
|
||||
depth: point.depth,
|
||||
frequencyHz: point.freqHz,
|
||||
currentMa: point.currentMa,
|
||||
signalDb: point.signalDb,
|
||||
gainDb: point.gainDb,
|
||||
locateMode: point.mode,
|
||||
phaseDeg: point.phaseDeg,
|
||||
compassDeg: point.compassDeg,
|
||||
distortionPct: point.distortionPct,
|
||||
recordedAt: new Date(point.ts),
|
||||
raw: point as unknown as Prisma.InputJsonValue,
|
||||
})),
|
||||
skipDuplicates: true,
|
||||
});
|
||||
}
|
||||
|
||||
const storedIds = new Set(stored.map((row) => row.pointId));
|
||||
for (const point of fresh) {
|
||||
results.push({
|
||||
pointId: point.pointId,
|
||||
outcome: storedIds.has(point.pointId) ? 'ACCEPTED' : 'DUPLICATE',
|
||||
});
|
||||
}
|
||||
|
||||
if (stored.length > 0) {
|
||||
this.realtime.publish(`job:${job.id}`, {
|
||||
type: 'points',
|
||||
jobId: job.id,
|
||||
points: stored.map(toPointDto),
|
||||
});
|
||||
}
|
||||
this.publishAck(orgId, clientId, results);
|
||||
}
|
||||
|
||||
private async resolveJob(
|
||||
orgId: string,
|
||||
clientId: string,
|
||||
msg: AppLogPointsMessageDto,
|
||||
) {
|
||||
if (msg.jobId) {
|
||||
const job = await this.prisma.job.findFirst({
|
||||
where: { id: msg.jobId, orgId },
|
||||
});
|
||||
if (!job)
|
||||
this.logger.warn(
|
||||
`app-log ${orgId}/${clientId} referenced a foreign/unknown job ${msg.jobId}`,
|
||||
);
|
||||
return job;
|
||||
}
|
||||
const existing = await this.prisma.job.findUnique({
|
||||
where: { orgId_ticketNumber: { orgId, ticketNumber: msg.ticket! } },
|
||||
});
|
||||
if (existing) return existing;
|
||||
return this.prisma.job.create({
|
||||
data: {
|
||||
orgId,
|
||||
ticketNumber: msg.ticket!,
|
||||
title: `Ticket ${msg.ticket} (app-created)`,
|
||||
status: 'IN_PROGRESS',
|
||||
source: 'WEB',
|
||||
startedAt: new Date(),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
private toStoredFix(fix?: AppFixType): GpsFixType | undefined {
|
||||
switch (fix) {
|
||||
case 'FLOAT':
|
||||
return GpsFixType.FLOAT_RTK;
|
||||
case 'FIXED':
|
||||
return GpsFixType.FIXED_RTK;
|
||||
case 'NO_FIX':
|
||||
return GpsFixType.NONE;
|
||||
case 'AUTONOMOUS':
|
||||
return GpsFixType.AUTONOMOUS;
|
||||
case 'DGPS':
|
||||
return GpsFixType.DGPS;
|
||||
default:
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
private publishAck(
|
||||
orgId: string,
|
||||
clientId: string,
|
||||
results: AppLogAckResult[],
|
||||
): void {
|
||||
if (results.length === 0) return;
|
||||
const ack: AppLogAck = { schemaVersion: SCHEMA_VERSION, results };
|
||||
this.mqttClient.publish(`ul/${orgId}/app/${clientId}/ack`, ack);
|
||||
}
|
||||
}
|
||||
189
backend/src/ingest/dto/app-messages.dto.ts
Normal file
189
backend/src/ingest/dto/app-messages.dto.ts
Normal file
@@ -0,0 +1,189 @@
|
||||
import { LocateMode, UtilityType } from '@prisma/client';
|
||||
import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
|
||||
import { Type } from 'class-transformer';
|
||||
import {
|
||||
ArrayMaxSize,
|
||||
ArrayMinSize,
|
||||
IsArray,
|
||||
IsDateString,
|
||||
IsEnum,
|
||||
IsIn,
|
||||
IsInt,
|
||||
IsNumber,
|
||||
IsOptional,
|
||||
IsString,
|
||||
IsUUID,
|
||||
Max,
|
||||
MaxLength,
|
||||
Min,
|
||||
ValidateNested,
|
||||
} from 'class-validator';
|
||||
|
||||
export const APP_FIX_TYPES = [
|
||||
'AUTONOMOUS',
|
||||
'DGPS',
|
||||
'FLOAT',
|
||||
'FIXED',
|
||||
'NO_FIX',
|
||||
] as const;
|
||||
export type AppFixType = (typeof APP_FIX_TYPES)[number];
|
||||
|
||||
// Frozen app MQTT wire profile v1 from meta/contracts/telemetry-schema.md. This is deliberately
|
||||
// separate from MqttPointDto: the app wire uses the normative fix tokens FLOAT/FIXED/NO_FIX,
|
||||
// while the legacy device path uses Prisma's FLOAT_RTK/FIXED_RTK/NONE storage tokens.
|
||||
export class AppLogPointDto {
|
||||
@ApiProperty({
|
||||
format: 'uuid',
|
||||
description: 'Client-generated UUIDv7 idempotency key.',
|
||||
})
|
||||
@IsUUID()
|
||||
pointId: string;
|
||||
|
||||
@ApiProperty({ format: 'date-time' })
|
||||
@IsDateString()
|
||||
createdAt: string;
|
||||
|
||||
@IsIn(['APP'])
|
||||
origin: 'APP';
|
||||
|
||||
@IsIn(['APP_MQTT'])
|
||||
uploadPath: 'APP_MQTT';
|
||||
|
||||
@IsNumber()
|
||||
@Min(-90)
|
||||
@Max(90)
|
||||
lat: number;
|
||||
|
||||
@IsNumber()
|
||||
@Min(-180)
|
||||
@Max(180)
|
||||
lng: number;
|
||||
|
||||
@IsOptional()
|
||||
@IsNumber()
|
||||
alt?: number;
|
||||
|
||||
@IsDateString()
|
||||
ts: string;
|
||||
|
||||
@IsOptional()
|
||||
@IsIn(APP_FIX_TYPES)
|
||||
fix?: AppFixType;
|
||||
|
||||
@IsOptional()
|
||||
@IsNumber()
|
||||
@Min(0)
|
||||
hAcc?: number;
|
||||
|
||||
@IsOptional()
|
||||
@IsNumber()
|
||||
@Min(0)
|
||||
vAcc?: number;
|
||||
|
||||
@IsOptional()
|
||||
@IsInt()
|
||||
@Min(0)
|
||||
sats?: number;
|
||||
|
||||
@IsOptional()
|
||||
@IsNumber()
|
||||
@Min(0)
|
||||
hdop?: number;
|
||||
|
||||
@IsOptional()
|
||||
@IsNumber()
|
||||
@Min(0)
|
||||
depth?: number;
|
||||
|
||||
@IsOptional()
|
||||
@IsInt()
|
||||
@Min(0)
|
||||
freqHz?: number;
|
||||
|
||||
@IsOptional()
|
||||
@IsNumber()
|
||||
@Min(0)
|
||||
currentMa?: number;
|
||||
|
||||
@IsOptional()
|
||||
@IsNumber()
|
||||
signalDb?: number;
|
||||
|
||||
@IsOptional()
|
||||
@IsNumber()
|
||||
gainDb?: number;
|
||||
|
||||
@IsOptional()
|
||||
@IsEnum(LocateMode)
|
||||
mode?: LocateMode;
|
||||
|
||||
@IsOptional()
|
||||
@IsNumber()
|
||||
phaseDeg?: number;
|
||||
|
||||
@IsOptional()
|
||||
@IsNumber()
|
||||
@Min(0)
|
||||
@Max(360)
|
||||
compassDeg?: number;
|
||||
|
||||
@IsOptional()
|
||||
@IsNumber()
|
||||
@Min(0)
|
||||
@Max(100)
|
||||
distortionPct?: number;
|
||||
|
||||
@IsOptional()
|
||||
@IsEnum(UtilityType)
|
||||
utility?: UtilityType;
|
||||
|
||||
@IsOptional()
|
||||
@IsInt()
|
||||
seq?: number;
|
||||
|
||||
@IsIn(['IN_SPEC', 'OUT_OF_SPEC', 'MANUAL', 'PPK_CORRECTED', 'WAIVED', 'NONCOMPLIANT'])
|
||||
qualityFlag: 'IN_SPEC' | 'OUT_OF_SPEC' | 'MANUAL' | 'PPK_CORRECTED' | 'WAIVED' | 'NONCOMPLIANT';
|
||||
}
|
||||
|
||||
export class AppLogPointsMessageDto {
|
||||
@ApiPropertyOptional({ description: 'MQTT app-log wire profile version.' })
|
||||
@IsString()
|
||||
@MaxLength(16)
|
||||
schemaVersion: string;
|
||||
|
||||
@ApiPropertyOptional({
|
||||
description: 'Job id; exactly one of jobId or ticket is required.',
|
||||
})
|
||||
@IsOptional()
|
||||
@IsString()
|
||||
jobId?: string;
|
||||
|
||||
@ApiPropertyOptional({
|
||||
description: 'Ticket number; exactly one of jobId or ticket is required.',
|
||||
})
|
||||
@IsOptional()
|
||||
@IsString()
|
||||
@MaxLength(64)
|
||||
ticket?: string;
|
||||
|
||||
@ApiProperty({ type: [AppLogPointDto] })
|
||||
@IsArray()
|
||||
@ArrayMinSize(1)
|
||||
@ArrayMaxSize(500)
|
||||
@ValidateNested({ each: true })
|
||||
@Type(() => AppLogPointDto)
|
||||
points: AppLogPointDto[];
|
||||
}
|
||||
|
||||
export type AppLogAckOutcome = 'ACCEPTED' | 'DUPLICATE' | 'REJECTED';
|
||||
|
||||
export interface AppLogAckResult {
|
||||
pointId: string;
|
||||
outcome: AppLogAckOutcome;
|
||||
reasonCode?: string;
|
||||
}
|
||||
|
||||
export interface AppLogAck {
|
||||
schemaVersion: string;
|
||||
results: AppLogAckResult[];
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { AppLogIngestService } from './app-log-ingest.service';
|
||||
import { JobsIngestService } from './jobs-ingest.service';
|
||||
import { LogIngestService } from './log-ingest.service';
|
||||
import { MqttClientService } from './mqtt-client.service';
|
||||
@@ -15,6 +16,7 @@ export class IngestRouterService implements OnModuleInit {
|
||||
private readonly pointsIngest: PointsIngestService,
|
||||
private readonly jobsIngest: JobsIngestService,
|
||||
private readonly logIngest: LogIngestService,
|
||||
private readonly appLogIngest: AppLogIngestService,
|
||||
) {}
|
||||
|
||||
onModuleInit() {
|
||||
@@ -26,10 +28,32 @@ export class IngestRouterService implements OnModuleInit {
|
||||
}
|
||||
|
||||
private async route(topic: string, payload: string) {
|
||||
// Raw log captures every message on devices/#, matched or not (audit/debug trail)
|
||||
await this.prisma.deviceEvent.create({ data: { topic, payload } });
|
||||
const segments = topic.split('/');
|
||||
const root = segments[0];
|
||||
|
||||
const [root, idSegment, ...rest] = topic.split('/');
|
||||
// App point payloads contain personal/location data. Preserve routing evidence without
|
||||
// creating an unmanaged second copy of the telemetry in device_events (S2 security H3).
|
||||
const auditPayload =
|
||||
root === 'ul' &&
|
||||
segments[2] === 'app' &&
|
||||
segments.slice(4).join('/') === 'log/points'
|
||||
? JSON.stringify({
|
||||
redacted: true,
|
||||
bytes: Buffer.byteLength(payload, 'utf8'),
|
||||
})
|
||||
: payload;
|
||||
await this.prisma.deviceEvent.create({
|
||||
data: { topic, payload: auditPayload },
|
||||
});
|
||||
|
||||
// App / device MQTT namespace: ul/{orgId}/{clientClass}/{clientId}/{subtopic...}
|
||||
// (SRS §3.4.2). Sprint 2 handles the durable app-log path; other clientClasses and
|
||||
// subtopics are represented but not yet ingested here.
|
||||
if (root === 'ul') {
|
||||
return this.routeUl(segments, payload);
|
||||
}
|
||||
|
||||
const [, idSegment, ...rest] = segments;
|
||||
const subtopic = rest.join('/');
|
||||
if (root !== 'devices' || !idSegment || subtopic === 'jobs/ack') {
|
||||
return; // not device traffic, or our own ack echoed back
|
||||
@@ -42,13 +66,19 @@ export class IngestRouterService implements OnModuleInit {
|
||||
return;
|
||||
}
|
||||
|
||||
const device = await this.prisma.device.findUnique({ where: { mqttUsername: idSegment } });
|
||||
const device = await this.prisma.device.findUnique({
|
||||
where: { mqttUsername: idSegment },
|
||||
});
|
||||
if (!device) {
|
||||
this.logger.warn(`Message from unregistered device username "${idSegment}" (raw-logged only)`);
|
||||
this.logger.warn(
|
||||
`Message from unregistered device username "${idSegment}" (raw-logged only)`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (!device.isActive) {
|
||||
this.logger.warn(`Message from deactivated device "${idSegment}" ignored`);
|
||||
this.logger.warn(
|
||||
`Message from deactivated device "${idSegment}" ignored`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -69,4 +99,29 @@ export class IngestRouterService implements OnModuleInit {
|
||||
this.logger.debug(`Unhandled subtopic "${subtopic}" from ${idSegment}`);
|
||||
}
|
||||
}
|
||||
|
||||
// ul/{orgId}/{clientClass}/{clientId}/{subtopic...} — SRS §3.4.2 topic scheme.
|
||||
// The namespace itself carries tenant (orgId) + publisher (clientId) identity; the
|
||||
// broker ACL confines a publisher to its own namespace, and AppLogIngest re-checks
|
||||
// tenant scope on the payload's job as a backstop. Only the durable app-log path is
|
||||
// ingested this sprint.
|
||||
private async routeUl(segments: string[], payload: string) {
|
||||
const [, orgId, clientClass, clientId, ...rest] = segments;
|
||||
const subtopic = rest.join('/');
|
||||
if (!orgId || !clientClass || !clientId) {
|
||||
this.logger.warn(`Malformed ul topic "${segments.join('/')}" — ignored`);
|
||||
return;
|
||||
}
|
||||
// Our own cloud→publisher acks are echoed back to us on the same subscription; skip.
|
||||
if (subtopic === 'ack') {
|
||||
return;
|
||||
}
|
||||
if (clientClass === 'app' && subtopic === 'log/points') {
|
||||
await this.appLogIngest.handle(orgId, clientId, payload);
|
||||
return;
|
||||
}
|
||||
this.logger.debug(
|
||||
`Unhandled ul path "${clientClass}/${subtopic}" (org ${orgId}, client ${clientId})`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
import { RealtimeModule } from '../realtime/realtime.module';
|
||||
import { AppLogIngestService } from './app-log-ingest.service';
|
||||
import { IngestRouterService } from './ingest-router.service';
|
||||
import { JobsIngestService } from './jobs-ingest.service';
|
||||
import { LocatorRegistryService } from './locator-registry.service';
|
||||
@@ -15,6 +16,7 @@ import { PointsIngestService } from './points-ingest.service';
|
||||
PointsIngestService,
|
||||
JobsIngestService,
|
||||
LogIngestService,
|
||||
AppLogIngestService,
|
||||
LocatorRegistryService,
|
||||
],
|
||||
exports: [MqttClientService],
|
||||
|
||||
@@ -1,4 +1,9 @@
|
||||
import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
|
||||
import {
|
||||
Injectable,
|
||||
Logger,
|
||||
OnModuleDestroy,
|
||||
OnModuleInit,
|
||||
} from '@nestjs/common';
|
||||
import { connect, MqttClient } from 'mqtt';
|
||||
|
||||
export type MqttMessageListener = (topic: string, payload: Buffer) => void;
|
||||
@@ -12,19 +17,29 @@ export class MqttClientService implements OnModuleInit, OnModuleDestroy {
|
||||
onModuleInit() {
|
||||
const host = process.env.MQTT_HOST || 'mosquitto';
|
||||
const port = Number(process.env.MQTT_PORT || 1883);
|
||||
const username = process.env.MQTT_USERNAME;
|
||||
const password = process.env.MQTT_PASSWORD;
|
||||
if (!username || !password) {
|
||||
throw new Error(
|
||||
'MQTT_USERNAME and MQTT_PASSWORD are required; refusing to use a default broker credential.',
|
||||
);
|
||||
}
|
||||
this.client = connect(`mqtt://${host}:${port}`, {
|
||||
username: process.env.MQTT_USERNAME || 'backend',
|
||||
password: process.env.MQTT_PASSWORD || 'backendpass',
|
||||
username,
|
||||
password,
|
||||
clientId: `ulhub-backend-${Math.random().toString(16).slice(2)}`,
|
||||
});
|
||||
|
||||
this.client.on('connect', () => {
|
||||
this.logger.log(`Connected to MQTT broker at ${host}:${port}`);
|
||||
this.client!.subscribe('devices/#', (err) => {
|
||||
// devices/# = legacy device-direct namespace; ul/# = SRS §3.4.2 topic scheme
|
||||
// (app + device-direct). QoS 1 so the broker redelivers durable log records the
|
||||
// backend missed while disconnected (broker loss ≠ capture loss, SRS-SYN-1/7).
|
||||
this.client!.subscribe(['devices/#', 'ul/#'], { qos: 1 }, (err) => {
|
||||
if (err) {
|
||||
this.logger.error('Failed to subscribe to devices/#', err);
|
||||
this.logger.error('Failed to subscribe to devices/# + ul/#', err);
|
||||
} else {
|
||||
this.logger.log('Subscribed to devices/#');
|
||||
this.logger.log('Subscribed to devices/# and ul/#');
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@@ -6,8 +6,13 @@ import { CreatePointDto, QueryPointsDto } from './dto/points.dto';
|
||||
// JSON-safe shape: BigInt id -> string, Decimal -> number
|
||||
export interface PointDto {
|
||||
id: string;
|
||||
pointId: string | null;
|
||||
jobId: string;
|
||||
deviceId: string | null;
|
||||
origin: string;
|
||||
uploadPath: string | null;
|
||||
originClientId: string | null;
|
||||
createdAt: Date | null;
|
||||
lat: number;
|
||||
lng: number;
|
||||
altitude: number | null;
|
||||
@@ -38,8 +43,13 @@ function num(value: unknown): number | null {
|
||||
export function toPointDto(p: LocatePoint): PointDto {
|
||||
return {
|
||||
id: p.id.toString(),
|
||||
pointId: p.pointId,
|
||||
jobId: p.jobId,
|
||||
deviceId: p.deviceId,
|
||||
origin: p.origin,
|
||||
uploadPath: p.uploadPath,
|
||||
originClientId: p.originClientId,
|
||||
createdAt: p.createdAt,
|
||||
lat: Number(p.lat),
|
||||
lng: Number(p.lng),
|
||||
altitude: num(p.altitude),
|
||||
|
||||
@@ -13,9 +13,12 @@ services:
|
||||
ports:
|
||||
- "1883:1883"
|
||||
- "8883:8883"
|
||||
- "9001:9001"
|
||||
- "8884:8884"
|
||||
volumes:
|
||||
- ./mosquitto:/mosquitto
|
||||
# Publicly trusted server identity for app MQTTS. The device listener still
|
||||
# validates client certificates against /mosquitto/certs/ca.crt.
|
||||
- /etc/letsencrypt:/etc/letsencrypt:ro
|
||||
|
||||
backend:
|
||||
build:
|
||||
@@ -37,7 +40,7 @@ services:
|
||||
MQTT_HOST: mosquitto
|
||||
MQTT_PORT: 1883
|
||||
MQTT_USERNAME: ${MQTT_BACKEND_USERNAME:-backend}
|
||||
MQTT_PASSWORD: ${MQTT_BACKEND_PASSWORD:-backendpass}
|
||||
MQTT_PASSWORD: ${MQTT_BACKEND_PASSWORD:?set MQTT_BACKEND_PASSWORD}
|
||||
MQTT_CERTS_DIR: /mosquitto-certs
|
||||
MQTT_TLS_PORT: 8883
|
||||
# Must match the CN the broker's server cert was provisioned for
|
||||
@@ -81,4 +84,4 @@ services:
|
||||
- postgres
|
||||
|
||||
volumes:
|
||||
postgres-data:
|
||||
postgres-data:
|
||||
|
||||
@@ -12,10 +12,22 @@ topic readwrite #
|
||||
topic readwrite $SYS/#
|
||||
|
||||
# Backend service: reads all device traffic, writes job acks back to devices,
|
||||
# and publishes on behalf of the /sim simulator tool (devices/<serial>/log)
|
||||
# and publishes on behalf of the /sim simulator tool (devices/<serial>/log).
|
||||
# Also ingests the SRS §3.4.2 ul/# namespace (app + device-direct) and writes the
|
||||
# application-level acks back on ul/{orgId}/{clientClass}/{clientId}/ack.
|
||||
user backend
|
||||
topic read devices/#
|
||||
topic write devices/+/jobs/ack
|
||||
topic write devices/+/log
|
||||
topic read ul/#
|
||||
topic write ul/+/+/+/ack
|
||||
|
||||
# --- App-client namespace confinement (S2-a / SRS §3.4.2 Topic scheme) --------------
|
||||
# The interim app credential uses username == orgId, so %u confines it to that tenant.
|
||||
# The '+' clientId wildcard is the accepted Sprint 2 tradeoff: clients within one org share
|
||||
# the credential and can see that org's ack topics until OIDC-derived per-client identity lands.
|
||||
# Publish/read permissions are narrower than readwrite '#': only durable point input and acks.
|
||||
pattern write ul/%u/app/+/log/points
|
||||
pattern read ul/%u/app/+/ack
|
||||
|
||||
# testuser is a demo *device*: only the per-device pattern rule above applies
|
||||
|
||||
@@ -6,11 +6,6 @@ password_file /mosquitto/config/passwd
|
||||
acl_file /mosquitto/config/devices.acl
|
||||
allow_anonymous false
|
||||
|
||||
# WebSocket — browser clients, no authentication required
|
||||
listener 9001 0.0.0.0
|
||||
protocol websockets
|
||||
allow_anonymous true
|
||||
|
||||
# TLS MQTT — devices authenticate with client certificates (port 8883)
|
||||
# require_certificate true forces client cert; cert CN becomes the MQTT username.
|
||||
# ACL restricts each device to devices/<serial_number>/#
|
||||
@@ -19,20 +14,19 @@ allow_anonymous true
|
||||
# since there's no config/cert hot-reload.
|
||||
listener 8883 0.0.0.0
|
||||
cafile /mosquitto/certs/ca.crt
|
||||
certfile /mosquitto/certs/server.crt
|
||||
keyfile /mosquitto/certs/server.key
|
||||
certfile /etc/letsencrypt/live/dev.hub.umagul.net/fullchain.pem
|
||||
keyfile /etc/letsencrypt/live/dev.hub.umagul.net/privkey.pem
|
||||
require_certificate true
|
||||
use_identity_as_username true
|
||||
allow_anonymous false
|
||||
acl_file /mosquitto/config/devices.acl
|
||||
|
||||
# TLS MQTT — admin access via username/password, no client cert required (port 8884)
|
||||
# Connect with CA cert for server verification, then username/password.
|
||||
# listener 8884 0.0.0.0
|
||||
# cafile /mosquitto/certs/ca.crt
|
||||
# certfile /mosquitto/certs/server.crt
|
||||
# keyfile /mosquitto/certs/server.key
|
||||
# require_certificate false
|
||||
# password_file /mosquitto/config/passwd
|
||||
# allow_anonymous false
|
||||
# acl_file /mosquitto/config/devices.acl
|
||||
# TLS MQTT — app/admin username+password access (port 8884). App usernames are orgIds;
|
||||
# devices.acl confines them to ul/{orgId}/app/... . No anonymous listener is exposed.
|
||||
listener 8884 0.0.0.0
|
||||
certfile /etc/letsencrypt/live/dev.hub.umagul.net/fullchain.pem
|
||||
keyfile /etc/letsencrypt/live/dev.hub.umagul.net/privkey.pem
|
||||
require_certificate false
|
||||
password_file /mosquitto/config/passwd
|
||||
allow_anonymous false
|
||||
acl_file /mosquitto/config/devices.acl
|
||||
|
||||
@@ -6,18 +6,6 @@ server {
|
||||
root /var/www/certbot;
|
||||
}
|
||||
|
||||
location /mqtt {
|
||||
# proxy MQTT over WebSocket (strip the /mqtt prefix so Mosquitto sees /)
|
||||
proxy_pass http://127.0.0.1:9001/;
|
||||
proxy_http_version 1.1;
|
||||
proxy_set_header Upgrade $http_upgrade;
|
||||
proxy_set_header Connection "upgrade";
|
||||
proxy_set_header Host $host;
|
||||
proxy_set_header X-Real-IP $remote_addr;
|
||||
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
|
||||
proxy_read_timeout 300s;
|
||||
}
|
||||
|
||||
location / {
|
||||
return 301 https://$host$request_uri;
|
||||
}
|
||||
@@ -33,14 +21,8 @@ server {
|
||||
client_max_body_size 20m;
|
||||
|
||||
location /mqtt {
|
||||
proxy_pass http://127.0.0.1:9001/;
|
||||
proxy_http_version 1.1;
|
||||
proxy_set_header Upgrade $http_upgrade;
|
||||
proxy_set_header Connection "upgrade";
|
||||
proxy_set_header Host $host;
|
||||
proxy_set_header X-Real-IP $remote_addr;
|
||||
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
|
||||
proxy_read_timeout 300s;
|
||||
# Anonymous MQTT-over-WebSocket was removed for Sprint 2 security criterion B1.
|
||||
return 410;
|
||||
}
|
||||
|
||||
location / {
|
||||
|
||||
Reference in New Issue
Block a user