Add ADS-B, AIS, and ALPR map layers with live CoT streaming.

Ingest aircraft and vessel tracks via OSINT feeds and tactical CoT, expose viewport-filtered SSE to the map, and add an OSM ALPR layer with tiled caching and performant marker sync.

server/utils/cotStore.js

@@ -1,66 +1,210 @@
 /**
- * In-memory CoT entity store: upsert by id, prune on read by TTL.
- * Single source of truth; getActiveEntities returns new objects (no mutation of returned refs).
+ * In-memory CoT store (TAK, ADS-B, AIS).
  */
 
 import { acquire } from './asyncLock.js'
-import { COT_ENTITY_TTL_MS } from './constants.js'
+import { COT_OSINT_TTL_MS, COT_TTL_MS } from './constants.js'
+import { inferSourceFromId, isInBbox, matchesLayerFilter } from './cotEntityUtils.js'
 
 const entities = new Map()
+/** @type {Set<(event: string, payload: unknown) => void>} */
+const listeners = new Set()
+
+/**
+ * @param {(event: string, payload: unknown) => void} fn
+ * @returns {() => void} Unsubscribe function.
+ */
+export function onCotChange(fn) {
+  listeners.add(fn)
+  return () => listeners.delete(fn)
+}
+
+function emitChange(event, payload) {
+  for (const fn of listeners) {
+    try {
+      fn(event, payload)
+    }
+    catch {
+      /* ignore listener errors */
+    }
+  }
+}
+
+function pickOptionalNumber(value, fallback) {
+  if (value === undefined || value === null) return fallback
+  const n = Number(value)
+  return Number.isFinite(n) ? n : fallback
+}
+
+/**
+ * @param {Record<string, unknown>} entity
+ */
+function toSnapshot(entity) {
+  return {
+    id: entity.id,
+    lat: entity.lat,
+    lng: entity.lng,
+    label: entity.label ?? entity.id,
+    type: entity.type ?? '',
+    source: entity.source ?? 'tak',
+    heading: entity.heading,
+    speed: entity.speed,
+    altitude: entity.altitude,
+    verticalRate: entity.verticalRate,
+    onGround: entity.onGround,
+    originCountry: entity.originCountry,
+    icao: entity.icao,
+    mmsi: entity.mmsi,
+    squawk: entity.squawk,
+    updatedAt: entity.updatedAt,
+  }
+}
+
+/**
+ * @param {Record<string, unknown>} entity
+ * @param {{ ttlMs?: number, osintTtlMs?: number }} opts
+ */
+function entityTtlMs(entity, opts) {
+  const source = entity.source
+  if (source === 'adsb' || source === 'ais') {
+    return opts.osintTtlMs ?? COT_OSINT_TTL_MS
+  }
+  return opts.ttlMs ?? COT_TTL_MS
+}
+
+function isEntityExpired(entity, now, opts) {
+  return now - entity.updatedAt > entityTtlMs(entity, opts)
+}
 
 /**
  * Upsert entity by id. Input is not mutated; stored value is a new object.
- * @param {{ id: string, lat: number, lng: number, label?: string, eventType?: string, type?: string }} parsed
+ * @param {{ id: string, lat: number, lng: number, label?: string, eventType?: string, type?: string, source?: string, heading?: number, speed?: number, altitude?: number }} parsed
+ * @param {{ silent?: boolean }} [options]
  */
-export async function updateFromCot(parsed) {
+export async function updateFromCot(parsed, options = {}) {
   if (!parsed || typeof parsed.id !== 'string') return
   const lat = Number(parsed.lat)
   const lng = Number(parsed.lng)
   if (!Number.isFinite(lat) || !Number.isFinite(lng)) return
 
+  let snapshot = null
   await acquire(`cot-${parsed.id}`, async () => {
     const now = Date.now()
     const existing = entities.get(parsed.id)
     const label = typeof parsed.label === 'string' ? parsed.label : (existing?.label ?? parsed.id)
-    const type = typeof parsed.eventType === 'string' ? parsed.eventType : (typeof parsed.type === 'string' ? parsed.type : (existing?.type ?? ''))
+    const type = typeof parsed.eventType === 'string'
+      ? parsed.eventType
+      : (typeof parsed.type === 'string' ? parsed.type : (existing?.type ?? ''))
+    const explicitSource = parsed.source
+    const source = explicitSource === 'adsb' || explicitSource === 'ais' || explicitSource === 'tak'
+      ? explicitSource
+      : inferSourceFromId(parsed.id)
 
-    entities.set(parsed.id, {
+    const stored = {
       id: parsed.id,
       lat,
       lng,
       label,
       type,
+      source,
+      heading: pickOptionalNumber(parsed.heading, existing?.heading),
+      speed: pickOptionalNumber(parsed.speed, existing?.speed),
+      altitude: pickOptionalNumber(parsed.altitude, existing?.altitude),
+      verticalRate: pickOptionalNumber(parsed.verticalRate, existing?.verticalRate),
+      onGround: typeof parsed.onGround === 'boolean' ? parsed.onGround : existing?.onGround,
+      originCountry: typeof parsed.originCountry === 'string' ? parsed.originCountry : existing?.originCountry,
+      icao: typeof parsed.icao === 'string' ? parsed.icao : existing?.icao,
+      mmsi: typeof parsed.mmsi === 'string' ? parsed.mmsi : existing?.mmsi,
+      squawk: typeof parsed.squawk === 'string' ? parsed.squawk : existing?.squawk,
       updatedAt: now,
-    })
+    }
+    entities.set(parsed.id, stored)
+    snapshot = toSnapshot(stored)
+  })
+
+  if (snapshot && !options.silent) emitChange('update', { entity: snapshot })
+}
+
+/**
+ * @param {number} now
+ * @param {{ ttlMs?: number, osintTtlMs?: number }} opts
+ */
+function pruneExpired(now, opts) {
+  const expired = []
+  for (const entity of entities.values()) {
+    if (isEntityExpired(entity, now, opts)) expired.push(entity.id)
+  }
+  for (const id of expired) {
+    entities.delete(id)
+    emitChange('remove', { id })
+  }
+}
+
+/**
+ * @param {{ ttlMs?: number, osintTtlMs?: number }} [opts]
+ */
+export async function pruneStaleEntities(opts = {}) {
+  const ttlMs = opts.ttlMs ?? COT_TTL_MS
+  const osintTtlMs = opts.osintTtlMs ?? COT_OSINT_TTL_MS
+  await acquire('cot-prune', async () => {
+    pruneExpired(Date.now(), { ttlMs, osintTtlMs })
   })
 }
 
 /**
- * Active entities (updated within ttlMs). Prunes expired. Returns new array of new objects.
- * @param {number} [ttlMs]
- * @returns {Promise<Array<{ id: string, lat: number, lng: number, label: string, type: string, updatedAt: number }>>} Snapshot of active entities.
+ * @param {Record<string, unknown>} entity
+ * @param {{ west: number, south: number, east: number, north: number } | null | undefined} bbox
+ * @param {boolean} takFilterBbox
  */
-export async function getActiveEntities(ttlMs = COT_ENTITY_TTL_MS) {
+function passesBboxFilter(entity, bbox, takFilterBbox) {
+  if (!bbox) return true
+  const inBox = isInBbox(entity, bbox)
+  if (entity.source === 'tak' && !takFilterBbox) return true
+  return inBox
+}
+
+/**
+ * Active entities (updated within ttlMs). Prunes expired. Returns new array of new objects.
+ * @param {{ ttlMs?: number, osintTtlMs?: number }} [opts]
+ */
+export async function getActiveEntities(opts = {}) {
+  const ttlMs = opts.ttlMs ?? COT_TTL_MS
+  const osintTtlMs = opts.osintTtlMs ?? COT_OSINT_TTL_MS
   return acquire('cot-prune', async () => {
     const now = Date.now()
+    pruneExpired(now, { ttlMs, osintTtlMs })
+    return [...entities.values()].map(toSnapshot)
+  })
+}
+
+/**
+ * Active entities filtered by viewport bbox and layer set.
+ * @param {{ west: number, south: number, east: number, north: number } | null} bbox
+ * @param {{ ttlMs?: number, osintTtlMs?: number, layers?: Set<string>, takFilterBbox?: boolean, maxEntities?: number }} [opts]
+ */
+export async function getActiveEntitiesInBbox(bbox, opts = {}) {
+  const ttlMs = opts.ttlMs ?? COT_TTL_MS
+  const osintTtlMs = opts.osintTtlMs ?? COT_OSINT_TTL_MS
+  const layers = opts.layers
+  const takFilterBbox = Boolean(opts.takFilterBbox)
+  const ttlOpts = { ttlMs, osintTtlMs }
+
+  return acquire('cot-prune', async () => {
+    const now = Date.now()
+    pruneExpired(now, ttlOpts)
     const active = []
-    const expired = []
     for (const entity of entities.values()) {
-      if (now - entity.updatedAt <= ttlMs) {
-        active.push({
-          id: entity.id,
-          lat: entity.lat,
-          lng: entity.lng,
-          label: entity.label ?? entity.id,
-          type: entity.type ?? '',
-          updatedAt: entity.updatedAt,
-        })
-      }
-      else {
-        expired.push(entity.id)
-      }
+      if (isEntityExpired(entity, now, ttlOpts)) continue
+      const snap = toSnapshot(entity)
+      if (!passesBboxFilter(snap, bbox, takFilterBbox)) continue
+      if (!matchesLayerFilter(layers, snap)) continue
+      active.push(snap)
+    }
+    const maxEntities = opts.maxEntities
+    if (maxEntities != null && active.length > maxEntities) {
+      active.sort((a, b) => (b.updatedAt ?? 0) - (a.updatedAt ?? 0))
+      active.length = maxEntities
     }
-    for (const id of expired) entities.delete(id)
     return active
   })
 }