feat(S2-a): ingest durable app MQTT points securely
This commit is contained in:
Brent Perteet
@@ -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")
|
||||
}
|
||||
|
||||
|
||||
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),
|
||||
|
||||
Reference in New Issue
Block a user