128 lines
4.5 KiB
TypeScript
128 lines
4.5 KiB
TypeScript
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';
|
|
import { PointsIngestService } from './points-ingest.service';
|
|
|
|
@Injectable()
|
|
export class IngestRouterService implements OnModuleInit {
|
|
private readonly logger = new Logger(IngestRouterService.name);
|
|
|
|
constructor(
|
|
private readonly prisma: PrismaService,
|
|
private readonly mqttClient: MqttClientService,
|
|
private readonly pointsIngest: PointsIngestService,
|
|
private readonly jobsIngest: JobsIngestService,
|
|
private readonly logIngest: LogIngestService,
|
|
private readonly appLogIngest: AppLogIngestService,
|
|
) {}
|
|
|
|
onModuleInit() {
|
|
this.mqttClient.onMessage((topic, payload) => {
|
|
this.route(topic, payload.toString()).catch((err) =>
|
|
this.logger.error(`Failed to process ${topic}: ${err.message}`),
|
|
);
|
|
});
|
|
}
|
|
|
|
private async route(topic: string, payload: string) {
|
|
const segments = topic.split('/');
|
|
const root = segments[0];
|
|
|
|
// 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
|
|
}
|
|
|
|
// devices/<serial>/log identifies the locator by serial in the topic
|
|
// itself (job in the payload supplies org) — no publisher device lookup.
|
|
if (subtopic === 'log') {
|
|
await this.logIngest.handle(idSegment, payload);
|
|
return;
|
|
}
|
|
|
|
const device = await this.prisma.device.findUnique({
|
|
where: { mqttUsername: idSegment },
|
|
});
|
|
if (!device) {
|
|
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`,
|
|
);
|
|
return;
|
|
}
|
|
|
|
this.prisma.device
|
|
.update({ where: { id: device.id }, data: { lastSeenAt: new Date() } })
|
|
.catch(() => undefined);
|
|
|
|
switch (subtopic) {
|
|
case 'points':
|
|
await this.pointsIngest.handle(device, payload);
|
|
break;
|
|
case 'jobs':
|
|
await this.jobsIngest.handle(device, payload);
|
|
break;
|
|
case 'status':
|
|
break; // lastSeenAt already stamped above
|
|
default:
|
|
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})`,
|
|
);
|
|
}
|
|
}
|