Build core domain: orgs, users, jobs, locate points, auth, live map

Replaces the device-events demo with the actual product:

- Prisma + PostGIS data layer (postgis/postgis:17-3.5). Lat/lng decimals
  are the source of truth; a generated geometry(Point,4326) column with
  a GIST index backs bbox queries. Migrations apply on container boot.
- JWT auth (bcryptjs + httpOnly cookie) with public registration that
  creates an org; per-org roles (ORG_ADMIN/MEMBER/VIEWER) enforced by
  guards on all /orgs/:orgId routes.
- Scoped API keys (X-API-Key, sha256-hashed, shown once) for
  programmatic access, manageable by org admins.
- REST API: jobs/tickets CRUD with filters, points query (time range,
  recordedAt cursor, bbox), members, devices, api-keys.
- MQTT ingest: devices publish to devices/{username}/points and /jobs;
  unknown tickets auto-create stub jobs (source=DEVICE); every message
  is raw-logged to device_events; acks on devices/{username}/jobs/ack.
  Broker gets a dedicated backend user; testuser is now a plain device.
- Realtime: plain-WS gateway at /api/ws (socket.io removed) with
  cookie auth and per-job channels feeding the map live.
- Next.js frontend: login/register, jobs list with filters, job detail
  with live Google map (APWA utility colors, polylines per run) behind
  a provider-neutral JobMap abstraction for a future Esri swap, and
  settings pages for members/devices/api-keys.
- Seed: Umagul org, admin user, testuser device, demo job with RTK
  points. Sample publisher updated to the new topic contract.

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

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

@@ -0,0 +1,100 @@
+import { GpsFixType, UtilityType } from '@prisma/client';
+import { Type } from 'class-transformer';
+import {
+  ArrayMaxSize,
+  ArrayMinSize,
+  IsArray,
+  IsDateString,
+  IsEnum,
+  IsInt,
+  IsNotEmpty,
+  IsNumber,
+  IsOptional,
+  IsString,
+  Max,
+  MaxLength,
+  Min,
+  ValidateNested,
+} from 'class-validator';
+
+export class MqttPointDto {
+  @IsNumber()
+  @Min(-90)
+  @Max(90)
+  lat: number;
+
+  @IsNumber()
+  @Min(-180)
+  @Max(180)
+  lng: number;
+
+  @IsOptional()
+  @IsNumber()
+  alt?: number;
+
+  @IsOptional()
+  @IsEnum(GpsFixType)
+  fix?: GpsFixType;
+
+  @IsOptional()
+  @IsNumber()
+  @Min(0)
+  hAcc?: number;
+
+  @IsOptional()
+  @IsNumber()
+  @Min(0)
+  depth?: number;
+
+  @IsOptional()
+  @IsEnum(UtilityType)
+  utility?: UtilityType;
+
+  @IsOptional()
+  @IsInt()
+  seq?: number;
+
+  @IsDateString()
+  ts: string;
+}
+
+export class MqttPointsMessageDto {
+  // Job is resolved by jobId when present, else by (device org, ticket)
+  @IsOptional()
+  @IsString()
+  jobId?: string;
+
+  @IsOptional()
+  @IsString()
+  @MaxLength(64)
+  ticket?: string;
+
+  @IsArray()
+  @ArrayMinSize(1)
+  @ArrayMaxSize(500)
+  @ValidateNested({ each: true })
+  @Type(() => MqttPointDto)
+  points: MqttPointDto[];
+}
+
+export class MqttJobMessageDto {
+  @IsString()
+  @IsNotEmpty()
+  @MaxLength(64)
+  ticket: string;
+
+  @IsOptional()
+  @IsString()
+  @MaxLength(200)
+  title?: string;
+
+  @IsOptional()
+  @IsString()
+  @MaxLength(4000)
+  description?: string;
+
+  @IsOptional()
+  @IsString()
+  @MaxLength(400)
+  address?: string;
+}

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

@@ -0,0 +1,63 @@
+import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
+import { PrismaService } from '../prisma/prisma.service';
+import { JobsIngestService } from './jobs-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,
+  ) {}
+
+  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) {
+    // 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 subtopic = rest.join('/');
+    if (root !== 'devices' || !username || subtopic === 'jobs/ack') {
+      return; // not device traffic, or our own ack echoed back
+    }
+
+    const device = await this.prisma.device.findUnique({ where: { mqttUsername: username } });
+    if (!device) {
+      this.logger.warn(`Message from unregistered device username "${username}" (raw-logged only)`);
+      return;
+    }
+    if (!device.isActive) {
+      this.logger.warn(`Message from deactivated device "${username}" 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 ${username}`);
+    }
+  }
+}

backend/src/ingest/ingest.module.ts

@@ -0,0 +1,12 @@
+import { Module } from '@nestjs/common';
+import { RealtimeModule } from '../realtime/realtime.module';
+import { IngestRouterService } from './ingest-router.service';
+import { JobsIngestService } from './jobs-ingest.service';
+import { MqttClientService } from './mqtt-client.service';
+import { PointsIngestService } from './points-ingest.service';
+
+@Module({
+  imports: [RealtimeModule],
+  providers: [MqttClientService, IngestRouterService, PointsIngestService, JobsIngestService],
+})
+export class IngestModule {}

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

@@ -0,0 +1,60 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { Device } from '@prisma/client';
+import { plainToInstance } from 'class-transformer';
+import { validate } from 'class-validator';
+import { PrismaService } from '../prisma/prisma.service';
+import { MqttJobMessageDto } from './dto/mqtt-messages.dto';
+import { MqttClientService } from './mqtt-client.service';
+
+@Injectable()
+export class JobsIngestService {
+  private readonly logger = new Logger(JobsIngestService.name);
+
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly mqttClient: MqttClientService,
+  ) {}
+
+  async handle(device: Device, rawPayload: string) {
+    let msg: MqttJobMessageDto;
+    try {
+      msg = plainToInstance(MqttJobMessageDto, JSON.parse(rawPayload) as object);
+    } catch {
+      this.ack(device, { ticket: null, status: 'error', reason: 'invalid JSON' });
+      return;
+    }
+
+    const errors = await validate(msg, { whitelist: true });
+    if (errors.length > 0) {
+      this.ack(device, { ticket: msg.ticket ?? null, status: 'error', reason: 'validation failed' });
+      this.logger.warn(`Invalid job message from ${device.mqttUsername}: ${errors}`);
+      return;
+    }
+
+    const existing = await this.prisma.job.findUnique({
+      where: { orgId_ticketNumber: { orgId: device.orgId, ticketNumber: msg.ticket } },
+    });
+    if (existing) {
+      this.ack(device, { ticket: msg.ticket, jobId: existing.id, status: 'exists' });
+      return;
+    }
+
+    const job = await this.prisma.job.create({
+      data: {
+        orgId: device.orgId,
+        ticketNumber: msg.ticket,
+        title: msg.title || `Ticket ${msg.ticket}`,
+        description: msg.description,
+        address: msg.address,
+        source: 'DEVICE',
+        createdByDeviceId: device.id,
+      },
+    });
+    this.ack(device, { ticket: msg.ticket, jobId: job.id, status: 'created' });
+    this.logger.log(`Device ${device.mqttUsername} created job ${job.ticketNumber}`);
+  }
+
+  private ack(device: Device, payload: object) {
+    this.mqttClient.publish(`devices/${device.mqttUsername}/jobs/ack`, payload);
+  }
+}

backend/src/ingest/mqtt-client.service.ts

@@ -0,0 +1,59 @@
+import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
+import { connect, MqttClient } from 'mqtt';
+
+export type MqttMessageListener = (topic: string, payload: Buffer) => void;
+
+@Injectable()
+export class MqttClientService implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(MqttClientService.name);
+  private readonly listeners = new Set<MqttMessageListener>();
+  private client: MqttClient | null = null;
+
+  onModuleInit() {
+    const host = process.env.MQTT_HOST || 'mosquitto';
+    const port = Number(process.env.MQTT_PORT || 1883);
+    this.client = connect(`mqtt://${host}:${port}`, {
+      username: process.env.MQTT_USERNAME || 'backend',
+      password: process.env.MQTT_PASSWORD || 'backendpass',
+      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) => {
+        if (err) {
+          this.logger.error('Failed to subscribe to devices/#', err);
+        } else {
+          this.logger.log('Subscribed to devices/#');
+        }
+      });
+    });
+
+    this.client.on('message', (topic, payload) => {
+      for (const listener of this.listeners) {
+        listener(topic, payload);
+      }
+    });
+
+    this.client.on('error', (error) => {
+      this.logger.error(`MQTT client error: ${error.message}`);
+    });
+  }
+
+  async onModuleDestroy() {
+    await this.client?.endAsync();
+  }
+
+  onMessage(listener: MqttMessageListener): () => void {
+    this.listeners.add(listener);
+    return () => this.listeners.delete(listener);
+  }
+
+  publish(topic: string, payload: object) {
+    this.client?.publish(topic, JSON.stringify(payload), { qos: 1 }, (err) => {
+      if (err) {
+        this.logger.error(`Failed to publish to ${topic}: ${err.message}`);
+      }
+    });
+  }
+}

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

@@ -0,0 +1,95 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { Device, Job } 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 { MqttPointsMessageDto } from './dto/mqtt-messages.dto';
+
+@Injectable()
+export class PointsIngestService {
+  private readonly logger = new Logger(PointsIngestService.name);
+
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly realtime: RealtimeService,
+  ) {}
+
+  async handle(device: Device, rawPayload: string) {
+    const msg = plainToInstance(MqttPointsMessageDto, JSON.parse(rawPayload) as object);
+    const errors = await validate(msg, { whitelist: true });
+    if (errors.length > 0) {
+      this.logger.warn(`Invalid points message from ${device.mqttUsername}: ${errors}`);
+      return;
+    }
+    if (!msg.jobId && !msg.ticket) {
+      this.logger.warn(`Points message from ${device.mqttUsername} has neither jobId nor ticket`);
+      return;
+    }
+
+    const job = await this.resolveJob(device, msg);
+    if (!job) {
+      return;
+    }
+
+    const points = await this.prisma.locatePoint.createManyAndReturn({
+      data: msg.points.map((p) => ({
+        jobId: job.id,
+        deviceId: device.id,
+        lat: p.lat,
+        lng: p.lng,
+        altitude: p.alt,
+        fixType: p.fix,
+        hAccuracy: p.hAcc,
+        depth: p.depth,
+        utilityType: p.utility,
+        sequence: p.seq,
+        recordedAt: new Date(p.ts),
+        raw: p as object,
+      })),
+    });
+
+    this.realtime.publish(`job:${job.id}`, {
+      type: 'points',
+      jobId: job.id,
+      points: points.map(toPointDto),
+    });
+    this.logger.debug(`Stored ${points.length} points for job ${job.ticketNumber}`);
+  }
+
+  private async resolveJob(device: Device, msg: MqttPointsMessageDto): Promise<Job | null> {
+    if (msg.jobId) {
+      const job = await this.prisma.job.findFirst({
+        where: { id: msg.jobId, orgId: device.orgId },
+      });
+      if (!job) {
+        this.logger.warn(`Unknown jobId ${msg.jobId} from device ${device.mqttUsername}`);
+      }
+      return job;
+    }
+
+    const ticket = msg.ticket!;
+    const existing = await this.prisma.job.findUnique({
+      where: { orgId_ticketNumber: { orgId: device.orgId, ticketNumber: ticket } },
+    });
+    if (existing) {
+      return existing;
+    }
+
+    // Field-first workflow: points arrive before anyone opened the ticket in the web
+    // app. Create a stub so the data is never lost; metadata gets filled in later.
+    this.logger.log(`Auto-creating stub job for unknown ticket ${ticket} (device ${device.mqttUsername})`);
+    return this.prisma.job.create({
+      data: {
+        orgId: device.orgId,
+        ticketNumber: ticket,
+        title: `Ticket ${ticket} (device-created)`,
+        status: 'IN_PROGRESS',
+        source: 'DEVICE',
+        createdByDeviceId: device.id,
+        startedAt: new Date(),
+      },
+    });
+  }
+}