diff --git a/backend/src/ingest/dto/mqtt-messages.dto.ts b/backend/src/ingest/dto/mqtt-messages.dto.ts index 58dd709..ac99449 100644 --- a/backend/src/ingest/dto/mqtt-messages.dto.ts +++ b/backend/src/ingest/dto/mqtt-messages.dto.ts @@ -6,6 +6,7 @@ import { IsArray, IsDateString, IsEnum, + IsIn, IsInt, IsNotEmpty, IsNumber, @@ -139,10 +140,18 @@ export class MqttPointsMessageDto { points: MqttPointDto[]; } -// Single-reading log entry for a locator that publishes directly (or is -// relayed) to devices//log — the serial comes from the topic itself, -// so identity is job-anchored rather than publisher-credential-anchored. +export const MQTT_LOG_MESSAGE_TYPES = ['log', 'status'] as const; +export type MqttLogMessageType = (typeof MQTT_LOG_MESSAGE_TYPES)[number]; + +// Single reading for a locator that publishes directly (or is relayed) to +// devices//log — the serial comes from the topic itself, so identity +// is job-anchored rather than publisher-credential-anchored. `type` decides +// what happens to it: "log" persists a LocatePoint; "status" is an ephemeral +// current-position update, broadcast live but never written to the DB. export class MqttLogMessageDto extends MqttPointDto { + @IsIn(MQTT_LOG_MESSAGE_TYPES) + type: MqttLogMessageType; + @IsString() @IsNotEmpty() jobId: string; diff --git a/backend/src/ingest/log-ingest.service.ts b/backend/src/ingest/log-ingest.service.ts index 5f28851..9e613de 100644 --- a/backend/src/ingest/log-ingest.service.ts +++ b/backend/src/ingest/log-ingest.service.ts @@ -20,6 +20,9 @@ export class LogIngestService { // devices//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. + // `type` decides what happens to the reading: "log" persists a LocatePoint; + // "status" is an ephemeral current-position update, broadcast live but + // never written to locate_points. async handle(serial: string, rawPayload: string) { const msg = plainToInstance(MqttLogMessageDto, JSON.parse(rawPayload) as object); const errors = await validate(msg, { whitelist: true }); @@ -36,6 +39,35 @@ export class LogIngestService { const locator = await this.locatorRegistry.resolve(job.orgId, serial); + if (msg.type === 'status') { + this.realtime.publish(`job:${job.id}`, { + type: 'status', + jobId: job.id, + deviceId: locator.id, + serial, + lat: msg.lat, + lng: msg.lng, + altitude: msg.alt ?? null, + fixType: msg.fix ?? 'NONE', + utilityType: msg.utility ?? 'UNKNOWN', + hAccuracy: msg.hAcc ?? null, + vAccuracy: msg.vAcc ?? null, + satellites: msg.sats ?? null, + hdop: msg.hdop ?? null, + depth: msg.depth ?? null, + frequencyHz: msg.freqHz ?? null, + currentMa: msg.currentMa ?? null, + signalDb: msg.signalDb ?? null, + gainDb: msg.gainDb ?? null, + locateMode: msg.mode ?? null, + phaseDeg: msg.phaseDeg ?? null, + compassDeg: msg.compassDeg ?? null, + distortionPct: msg.distortionPct ?? null, + recordedAt: msg.ts, + }); + return; + } + const point = await this.prisma.locatePoint.create({ data: { jobId: job.id, diff --git a/backend/src/sim/dto/sim.dto.ts b/backend/src/sim/dto/sim.dto.ts index 87a0dec..bd96689 100644 --- a/backend/src/sim/dto/sim.dto.ts +++ b/backend/src/sim/dto/sim.dto.ts @@ -1,17 +1,12 @@ import { IsNotEmpty, IsString, MaxLength } from 'class-validator'; -import { MqttPointDto } from '../../ingest/dto/mqtt-messages.dto'; +import { MqttLogMessageDto } from '../../ingest/dto/mqtt-messages.dto'; -// What the simulator UI sends the backend. Mirrors a devices//log -// payload plus the two fields that live in the topic/routing rather than the -// wire payload itself (serial, jobId) so this DTO doubles as validation for -// both the REST call and the outgoing MQTT message. -export class SimPublishPointDto extends MqttPointDto { +// What the simulator UI sends the backend: a devices//log payload +// (type: "log" persists a point, "status" is a live-only position update) +// plus "serial", which lives in the topic rather than the wire payload. +export class SimPublishPointDto extends MqttLogMessageDto { @IsString() @IsNotEmpty() @MaxLength(64) serial: string; - - @IsString() - @IsNotEmpty() - jobId: string; } diff --git a/web/components/map/google/GoogleJobMap.tsx b/web/components/map/google/GoogleJobMap.tsx index b0e666e..0f6c4bc 100644 --- a/web/components/map/google/GoogleJobMap.tsx +++ b/web/components/map/google/GoogleJobMap.tsx @@ -1,9 +1,10 @@ /// import { APIProvider, InfoWindow, Map as GoogleMap, useMap } from '@vis.gl/react-google-maps'; import { useEffect, useRef, useState } from 'react'; -import { JobMapProps, MapPoint, pointDetailRows, UTILITY_COLORS } from '../types'; +import { JobMapProps, LiveStatus, liveStatusToMapPoint, MapPoint, pointDetailRows, UTILITY_COLORS } from '../types'; const API_KEY = process.env.NEXT_PUBLIC_GOOGLE_MAPS_API_KEY || ''; +const LIVE_COLOR = '#1a73e8'; const DEFAULT_CENTER = { lat: 39.5, lng: -98.35 }; // continental US const DEFAULT_ZOOM = 4; @@ -95,6 +96,80 @@ function PointsLayer({ return null; } +// Renders the transmitter's current position as a "blue dot" (marker + soft +// accuracy halo), updated in place as new status messages arrive. +function LiveMarker({ status, onSelect }: { status: LiveStatus | null; onSelect: (point: MapPoint) => void }) { + const map = useMap(); + const markerRef = useRef(null); + const haloRef = useRef(null); + const statusRef = useRef(status); + statusRef.current = status; + const onSelectRef = useRef(onSelect); + onSelectRef.current = onSelect; + + useEffect(() => { + if (!map) { + return; + } + if (!status) { + markerRef.current?.setMap(null); + haloRef.current?.setMap(null); + markerRef.current = null; + haloRef.current = null; + return; + } + + const position = { lat: status.lat, lng: status.lng }; + + if (!markerRef.current) { + markerRef.current = new google.maps.Marker({ + map, + position, + zIndex: 999, + cursor: 'pointer', + icon: { + path: google.maps.SymbolPath.CIRCLE, + scale: 8, + fillColor: LIVE_COLOR, + fillOpacity: 1, + strokeColor: '#ffffff', + strokeWeight: 2, + }, + }); + markerRef.current.addListener('click', () => { + if (statusRef.current) { + onSelectRef.current(liveStatusToMapPoint(statusRef.current)); + } + }); + haloRef.current = new google.maps.Circle({ + map, + center: position, + radius: status.hAccuracy ?? 5, + fillColor: LIVE_COLOR, + fillOpacity: 0.15, + strokeColor: LIVE_COLOR, + strokeOpacity: 0.3, + strokeWeight: 1, + clickable: false, + }); + } else { + markerRef.current.setPosition(position); + haloRef.current?.setCenter(position); + haloRef.current?.setRadius(status.hAccuracy ?? 5); + } + }, [map, status]); + + useEffect( + () => () => { + markerRef.current?.setMap(null); + haloRef.current?.setMap(null); + }, + [], + ); + + return null; +} + function PointDetails({ point, onClose }: { point: MapPoint; onClose: () => void }) { return ( @@ -114,7 +189,7 @@ function PointDetails({ point, onClose }: { point: MapPoint; onClose: () => void ); } -export default function GoogleJobMap({ points, fitBounds = true, heightPx = 480 }: JobMapProps) { +export default function GoogleJobMap({ points, liveStatus = null, fitBounds = true, heightPx = 480 }: JobMapProps) { const [selected, setSelected] = useState(null); if (!API_KEY) { @@ -136,8 +211,17 @@ export default function GoogleJobMap({ points, fitBounds = true, heightPx = 480 } // A point can be re-fetched (new object identity) between renders; keep the - // InfoWindow anchored to the latest copy of the same point by id. - const selectedCurrent = selected ? (points.find((p) => p.id === selected.id) ?? selected) : null; + // InfoWindow anchored to the latest copy of the same point by id. The live + // marker's "point" is synthesized fresh from liveStatus each time. + const selectedCurrent = (() => { + if (!selected) { + return null; + } + if (selected.id.startsWith('live-')) { + return liveStatus ? liveStatusToMapPoint(liveStatus) : null; + } + return points.find((p) => p.id === selected.id) ?? selected; + })(); return (
@@ -150,6 +234,7 @@ export default function GoogleJobMap({ points, fitBounds = true, heightPx = 480 onClick={() => setSelected(null)} > + {selectedCurrent && setSelected(null)} />} diff --git a/web/components/map/types.ts b/web/components/map/types.ts index f678799..f360e68 100644 --- a/web/components/map/types.ts +++ b/web/components/map/types.ts @@ -67,8 +67,41 @@ export function pointDetailRows(p: MapPoint): Array<[string, string]> { return rows; } +// A transmitter's current position, pushed live over the job's realtime +// channel. Never persisted as a LocatePoint — just the latest "where is it +// right now," replaced in place as new status updates arrive. +export interface LiveStatus { + deviceId: string; + serial: string; + lat: number; + lng: number; + altitude: number | null; + utilityType: string; + fixType: string; + hAccuracy: number | null; + vAccuracy: number | null; + satellites: number | null; + hdop: number | null; + depth: number | null; + frequencyHz: number | null; + currentMa: number | null; + signalDb: number | null; + gainDb: number | null; + locateMode: string | null; + phaseDeg: number | null; + compassDeg: number | null; + distortionPct: number | null; + recordedAt: string; +} + +export function liveStatusToMapPoint(s: LiveStatus): MapPoint { + return { ...s, id: `live-${s.serial}`, sequence: null }; +} + export interface JobMapProps { points: MapPoint[]; + // Live current position of a transmitter, shown as a distinct marker + liveStatus?: LiveStatus | null; // Pan/zoom to fit all points whenever their count changes fitBounds?: boolean; heightPx?: number; diff --git a/web/lib/use-job-stream.ts b/web/lib/use-job-stream.ts index b8c9639..29e64dc 100644 --- a/web/lib/use-job-stream.ts +++ b/web/lib/use-job-stream.ts @@ -1,11 +1,18 @@ import { useEffect, useRef } from 'react'; -import type { MapPoint } from '../components/map/types'; +import type { LiveStatus, MapPoint } from '../components/map/types'; -// Subscribes to live points for a job over the backend WebSocket. -// Reconnects with capped exponential backoff; resubscribes on reconnect. -export function useJobStream(jobId: string | null, onPoints: (points: MapPoint[]) => void) { - const handlerRef = useRef(onPoints); - handlerRef.current = onPoints; +// Subscribes to live points + live position status for a job over the +// backend WebSocket. Reconnects with capped exponential backoff; resubscribes +// on reconnect. +export function useJobStream( + jobId: string | null, + onPoints: (points: MapPoint[]) => void, + onStatus?: (status: LiveStatus) => void, +) { + const pointsRef = useRef(onPoints); + pointsRef.current = onPoints; + const statusRef = useRef(onStatus); + statusRef.current = onStatus; useEffect(() => { if (!jobId) { @@ -29,8 +36,13 @@ export function useJobStream(jobId: string | null, onPoints: (points: MapPoint[] socket.onmessage = (event) => { try { const msg = JSON.parse(event.data); - if (msg.type === 'points' && msg.jobId === jobId) { - handlerRef.current(msg.points); + if (msg.jobId !== jobId) { + return; + } + if (msg.type === 'points') { + pointsRef.current(msg.points); + } else if (msg.type === 'status') { + statusRef.current?.(msg as LiveStatus); } } catch { // ignore malformed frames diff --git a/web/pages/jobs/[jobId].tsx b/web/pages/jobs/[jobId].tsx index c7b9674..afe80a4 100644 --- a/web/pages/jobs/[jobId].tsx +++ b/web/pages/jobs/[jobId].tsx @@ -2,7 +2,7 @@ import { useRouter } from 'next/router'; import { useCallback, useEffect, useState } from 'react'; import Layout from '../../components/Layout'; import JobMap from '../../components/map/JobMap'; -import { pointSummary, type MapPoint } from '../../components/map/types'; +import { pointSummary, type LiveStatus, type MapPoint } from '../../components/map/types'; import { api } from '../../lib/api'; import { useRequireAuth } from '../../lib/auth-context'; import { useJobStream } from '../../lib/use-job-stream'; @@ -32,6 +32,7 @@ export default function JobDetailPage() { const [job, setJob] = useState(null); const [points, setPoints] = useState([]); const [live, setLive] = useState(0); + const [liveStatus, setLiveStatus] = useState(null); const [error, setError] = useState(null); useEffect(() => { @@ -57,7 +58,9 @@ export default function JobDetailPage() { setLive((n) => n + incoming.length); }, []); - useJobStream(jobId, onLivePoints); + const onStatus = useCallback((status: LiveStatus) => setLiveStatus(status), []); + + useJobStream(jobId, onLivePoints, onStatus); const updateStatus = async (status: string) => { if (!orgId || !jobId) { @@ -108,9 +111,14 @@ export default function JobDetailPage() { {points.length > 0 && ( latest: {pointSummary(points[points.length - 1])} )} + {liveStatus && ( + + ● transmitter {liveStatus.serial} live ({new Date(liveStatus.recordedAt).toLocaleTimeString()}) + + )}
- + )} diff --git a/web/pages/sim/index.tsx b/web/pages/sim/index.tsx index cd2679c..f5c8f93 100644 --- a/web/pages/sim/index.tsx +++ b/web/pages/sim/index.tsx @@ -14,11 +14,14 @@ interface JobOption { interface SentEntry { at: string; seq: number; + type: MessageType; lat: number; lng: number; depth: number; } +type MessageType = 'log' | 'status'; + const UTILITIES = ['GAS', 'WATER', 'ELECTRIC', 'SEWER', 'TELECOM', 'CATV', 'FIBER', 'STEAM', 'UNKNOWN']; const FIX_TYPES = ['FIXED_RTK', 'FLOAT_RTK', 'DGPS', 'AUTONOMOUS', 'NONE']; const LOCATE_MODES = ['PEAK', 'NULL', 'BROAD_PEAK', 'SONDE']; @@ -47,6 +50,7 @@ export default function SimulatorPage() { const [newTicket, setNewTicket] = useState(''); const [newTitle, setNewTitle] = useState(''); + const [messageType, setMessageType] = useState('log'); const [serial, setSerial] = useState(''); const [utility, setUtility] = useState('GAS'); const [fixType, setFixType] = useState('FIXED_RTK'); @@ -131,6 +135,7 @@ export default function SimulatorPage() { const pointDepth = Math.max(0.1, jitter(depth, 0.15)); try { await api.post(`/api/orgs/${orgId}/sim/publish`, { + type: messageType, serial, jobId, lat: pos.lat, @@ -153,7 +158,12 @@ export default function SimulatorPage() { ts: new Date().toISOString(), }); setCount(seq); - setSent((prev) => [{ at: new Date().toLocaleTimeString(), seq, lat: pos.lat, lng: pos.lng, depth: pointDepth }, ...prev].slice(0, 25)); + setSent((prev) => + [ + { at: new Date().toLocaleTimeString(), seq, type: messageType, lat: pos.lat, lng: pos.lng, depth: pointDepth }, + ...prev, + ].slice(0, 25), + ); setError(null); } catch (err: any) { setError(err.message); @@ -210,6 +220,18 @@ export default function SimulatorPage() {

{error &&

{error}

} +
+ Message type + + +
+
Job {!creatingJob ? ( @@ -371,10 +393,10 @@ export default function SimulatorPage() {