Tie points to locator serial numbers and store receiver telemetry

Adds devices/<serial>/log as a job-anchored MQTT ingest path: a locator
identifies itself by serial in the topic (auto-registered on first
sight, org resolved from the job in the payload) rather than by a
pre-provisioned broker credential. Device.serialNumber is now globally
unique so the bare topic segment is enough to resolve identity.
Locator lookup/auto-register logic is shared (LocatorRegistryService)
between this and the existing devices/<mqttUsername>/points path.

New /sim page (own header, outside the main app nav) simulates a
transmitter: pick an open job or create one, set a serial number and
telemetry defaults, and send points one at a time or on an interval
along a simulated walking path. It calls a new authenticated backend
endpoint (POST /orgs/:orgId/sim/publish) that publishes onto the real
broker rather than writing the DB directly, so the simulator exercises
the actual ingest pipeline end-to-end.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

backend/src/ingest/dto/mqtt-messages.dto.ts

@@ -139,6 +139,15 @@ export class MqttPointsMessageDto {
   points: MqttPointDto[];
 }
 
+// Single-reading log entry for a locator that publishes directly (or is
+// relayed) to devices/<serial>/log — the serial comes from the topic itself,
+// so identity is job-anchored rather than publisher-credential-anchored.
+export class MqttLogMessageDto extends MqttPointDto {
+  @IsString()
+  @IsNotEmpty()
+  jobId: string;
+}
+
 export class MqttJobMessageDto {
   @IsString()
   @IsNotEmpty()

backend/src/ingest/ingest-router.service.ts

@@ -1,6 +1,7 @@
 import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
 import { PrismaService } from '../prisma/prisma.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';
 
@@ -13,6 +14,7 @@ export class IngestRouterService implements OnModuleInit {
     private readonly mqttClient: MqttClientService,
     private readonly pointsIngest: PointsIngestService,
     private readonly jobsIngest: JobsIngestService,
+    private readonly logIngest: LogIngestService,
   ) {}
 
   onModuleInit() {
@@ -27,19 +29,26 @@ export class IngestRouterService implements OnModuleInit {
     // Raw log captures every message on devices/#, matched or not (audit/debug trail)
     await this.prisma.deviceEvent.create({ data: { topic, payload } });
 
-    const [root, username, ...rest] = topic.split('/');
+    const [root, idSegment, ...rest] = topic.split('/');
     const subtopic = rest.join('/');
-    if (root !== 'devices' || !username || subtopic === 'jobs/ack') {
+    if (root !== 'devices' || !idSegment || subtopic === 'jobs/ack') {
       return; // not device traffic, or our own ack echoed back
     }
 
-    const device = await this.prisma.device.findUnique({ where: { mqttUsername: username } });
+    // 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 "${username}" (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 "${username}" ignored`);
+      this.logger.warn(`Message from deactivated device "${idSegment}" ignored`);
       return;
     }
 
@@ -57,7 +66,7 @@ export class IngestRouterService implements OnModuleInit {
       case 'status':
         break; // lastSeenAt already stamped above
       default:
-        this.logger.debug(`Unhandled subtopic "${subtopic}" from ${username}`);
+        this.logger.debug(`Unhandled subtopic "${subtopic}" from ${idSegment}`);
     }
   }
 }

backend/src/ingest/ingest.module.ts

@@ -2,11 +2,21 @@ import { Module } from '@nestjs/common';
 import { RealtimeModule } from '../realtime/realtime.module';
 import { IngestRouterService } from './ingest-router.service';
 import { JobsIngestService } from './jobs-ingest.service';
+import { LocatorRegistryService } from './locator-registry.service';
+import { LogIngestService } from './log-ingest.service';
 import { MqttClientService } from './mqtt-client.service';
 import { PointsIngestService } from './points-ingest.service';
 
 @Module({
   imports: [RealtimeModule],
-  providers: [MqttClientService, IngestRouterService, PointsIngestService, JobsIngestService],
+  providers: [
+    MqttClientService,
+    IngestRouterService,
+    PointsIngestService,
+    JobsIngestService,
+    LogIngestService,
+    LocatorRegistryService,
+  ],
+  exports: [MqttClientService],
 })
 export class IngestModule {}

backend/src/ingest/locator-registry.service.ts

@@ -0,0 +1,37 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { Device } from '@prisma/client';
+import { PrismaService } from '../prisma/prisma.service';
+
+@Injectable()
+export class LocatorRegistryService {
+  private readonly logger = new Logger(LocatorRegistryService.name);
+
+  constructor(private readonly prisma: PrismaService) {}
+
+  // Serial numbers are globally unique, so a bare serial identifies a locator
+  // regardless of which org's data pipeline saw it first. Unknown serials are
+  // auto-registered under the given org so field data is never dropped.
+  async resolve(orgId: string, serial: string): Promise<Device> {
+    const existing = await this.prisma.device.findUnique({ where: { serialNumber: serial } });
+    if (existing) {
+      this.prisma.device
+        .update({ where: { id: existing.id }, data: { lastSeenAt: new Date() } })
+        .catch(() => undefined);
+      return existing;
+    }
+
+    this.logger.log(`Auto-registering locator serial ${serial} under org ${orgId}`);
+    try {
+      return await this.prisma.device.create({
+        data: { orgId, name: `Locator ${serial}`, serialNumber: serial, lastSeenAt: new Date() },
+      });
+    } catch {
+      // lost a concurrent-registration race; the row exists now
+      const raced = await this.prisma.device.findUnique({ where: { serialNumber: serial } });
+      if (!raced) {
+        throw new Error(`Failed to resolve or create locator ${serial}`);
+      }
+      return raced;
+    }
+  }
+}

backend/src/ingest/log-ingest.service.ts

@@ -0,0 +1,74 @@
+import { Injectable, Logger } from '@nestjs/common';
+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 { MqttLogMessageDto } from './dto/mqtt-messages.dto';
+import { LocatorRegistryService } from './locator-registry.service';
+
+@Injectable()
+export class LogIngestService {
+  private readonly logger = new Logger(LogIngestService.name);
+
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly realtime: RealtimeService,
+    private readonly locatorRegistry: LocatorRegistryService,
+  ) {}
+
+  // devices/<serial>/log carries the locator's identity in the topic itself
+  // (no publisher device needs to be pre-registered); the job named in the
+  // payload supplies the org, since the topic alone doesn't identify one.
+  async handle(serial: string, rawPayload: string) {
+    const msg = plainToInstance(MqttLogMessageDto, JSON.parse(rawPayload) as object);
+    const errors = await validate(msg, { whitelist: true });
+    if (errors.length > 0) {
+      this.logger.warn(`Invalid log message from serial ${serial}: ${errors}`);
+      return;
+    }
+
+    const job = await this.prisma.job.findUnique({ where: { id: msg.jobId } });
+    if (!job) {
+      this.logger.warn(`Unknown jobId ${msg.jobId} in log message from serial ${serial}`);
+      return;
+    }
+
+    const locator = await this.locatorRegistry.resolve(job.orgId, serial);
+
+    const point = await this.prisma.locatePoint.create({
+      data: {
+        jobId: job.id,
+        deviceId: locator.id,
+        lat: msg.lat,
+        lng: msg.lng,
+        altitude: msg.alt,
+        utilityType: msg.utility,
+        sequence: msg.seq,
+        fixType: msg.fix,
+        hAccuracy: msg.hAcc,
+        vAccuracy: msg.vAcc,
+        satellites: msg.sats,
+        hdop: msg.hdop,
+        depth: msg.depth,
+        frequencyHz: msg.freqHz,
+        currentMa: msg.currentMa,
+        signalDb: msg.signalDb,
+        gainDb: msg.gainDb,
+        locateMode: msg.mode,
+        phaseDeg: msg.phaseDeg,
+        compassDeg: msg.compassDeg,
+        distortionPct: msg.distortionPct,
+        recordedAt: new Date(msg.ts),
+        raw: msg as object,
+      },
+    });
+
+    this.realtime.publish(`job:${job.id}`, {
+      type: 'points',
+      jobId: job.id,
+      points: [toPointDto(point)],
+    });
+    this.logger.debug(`Stored 1 point from serial ${serial} for job ${job.ticketNumber}`);
+  }
+}

backend/src/ingest/points-ingest.service.ts

@@ -6,6 +6,7 @@ import { toPointDto } from '../points/points.service';
 import { PrismaService } from '../prisma/prisma.service';
 import { RealtimeService } from '../realtime/realtime.service';
 import { MqttPointsMessageDto } from './dto/mqtt-messages.dto';
+import { LocatorRegistryService } from './locator-registry.service';
 
 @Injectable()
 export class PointsIngestService {
@@ -14,6 +15,7 @@ export class PointsIngestService {
   constructor(
     private readonly prisma: PrismaService,
     private readonly realtime: RealtimeService,
+    private readonly locatorRegistry: LocatorRegistryService,
   ) {}
 
   async handle(device: Device, rawPayload: string) {
@@ -33,7 +35,9 @@ export class PointsIngestService {
       return;
     }
 
-    const locator = await this.resolveLocator(device, msg.serial);
+    const locator = !msg.serial || msg.serial === device.serialNumber
+      ? device
+      : await this.locatorRegistry.resolve(device.orgId, msg.serial);
 
     const points = await this.prisma.locatePoint.createManyAndReturn({
       data: msg.points.map((p) => ({
@@ -71,42 +75,6 @@ export class PointsIngestService {
     this.logger.debug(`Stored ${points.length} points for job ${job.ticketNumber}`);
   }
 
-  // Points are attributed to the locator receiver named by its serial number.
-  // The publishing MQTT credential (a phone/gateway, possibly relaying for
-  // several locators) only establishes the org; unknown serials are
-  // auto-registered so field data is never dropped.
-  private async resolveLocator(publisher: Device, serial: string | undefined): Promise<Device> {
-    if (!serial || serial === publisher.serialNumber) {
-      return publisher;
-    }
-    const existing = await this.prisma.device.findUnique({
-      where: { orgId_serialNumber: { orgId: publisher.orgId, serialNumber: serial } },
-    });
-    if (existing) {
-      this.prisma.device
-        .update({ where: { id: existing.id }, data: { lastSeenAt: new Date() } })
-        .catch(() => undefined);
-      return existing;
-    }
-    this.logger.log(`Auto-registering locator serial ${serial} (via ${publisher.mqttUsername})`);
-    try {
-      return await this.prisma.device.create({
-        data: {
-          orgId: publisher.orgId,
-          name: `Locator ${serial}`,
-          serialNumber: serial,
-          lastSeenAt: new Date(),
-        },
-      });
-    } catch {
-      // lost a concurrent-registration race; the row exists now
-      const raced = await this.prisma.device.findUnique({
-        where: { orgId_serialNumber: { orgId: publisher.orgId, serialNumber: serial } },
-      });
-      return raced ?? publisher;
-    }
-  }
-
   private async resolveJob(device: Device, msg: MqttPointsMessageDto): Promise<Job | null> {
     if (msg.jobId) {
       const job = await this.prisma.job.findFirst({