add passenger telemetry

backend/package.json

@@ -6,6 +6,7 @@
   "scripts": {
     "start": "node src/index.js",
     "dev": "nodemon src/index.js",
+    "test": "node --test",
     "migrate": "node src/utils/migrate.js",
     "create-admin": "node scripts/create-admin.js",
     "seed": "node src/utils/seed.js"

backend/src/controllers/passengerTelemetryController.js

@@ -0,0 +1,74 @@
+const { getPassengerTelemetryService } = require('../services/passengerTelemetryService');
+
+function sendError(res, error, fallbackMessage) {
+  if (error.status) {
+    return res.status(error.status).json({ error: error.message });
+  }
+
+  console.error(fallbackMessage, error);
+  return res.status(500).json({ error: fallbackMessage });
+}
+
+class PassengerTelemetryController {
+  constructor({
+    service = null,
+    broadcastToRoute = null
+  } = {}) {
+    this.service = service;
+    this.broadcastToRoute = broadcastToRoute;
+  }
+
+  getService() {
+    return this.service || getPassengerTelemetryService();
+  }
+
+  getBroadcastToRoute() {
+    if (this.broadcastToRoute) {
+      return this.broadcastToRoute;
+    }
+
+    return require('../services/realtimeService').broadcastToRoute;
+  }
+
+  async getRouteSnapshot(routeId) {
+    const telemetryService = require('../services/telemetryService');
+    return telemetryService.getRouteLiveVehicles(routeId);
+  }
+
+  async ingest(req, res) {
+    try {
+      const result = await this.getService().ingest(req.body);
+      const broadcastToRoute = this.getBroadcastToRoute();
+
+      if (result.routeId && (result.virtualVehicles || []).length > 0) {
+        const vehicles = await this.getRouteSnapshot(result.routeId).catch((error) => {
+          console.error('Failed to build passenger telemetry realtime snapshot', error);
+          return null;
+        });
+
+        if (vehicles) {
+          broadcastToRoute(result.routeId, {
+            type: 'snapshot:route',
+            routeId: result.routeId,
+            vehicles
+          });
+        } else {
+          for (const vehicle of result.virtualVehicles) {
+            broadcastToRoute(vehicle.route_id, {
+              type: 'vehicle:position',
+              routeId: vehicle.route_id,
+              vehicle
+            });
+          }
+        }
+      }
+
+      return res.status(202).json({ accepted: true });
+    } catch (error) {
+      return sendError(res, error, 'Failed to ingest passenger telemetry');
+    }
+  }
+}
+
+module.exports = new PassengerTelemetryController();
+module.exports.PassengerTelemetryController = PassengerTelemetryController;

backend/src/controllers/telemetryController.js

@@ -36,6 +36,19 @@ class TelemetryController {
         vehicle
       });
 
+      const routeVehicles = await telemetryService.getRouteLiveVehicles(vehicle.route_id).catch((error) => {
+        logger.error('Failed to build realtime snapshot after official telemetry ingest:', error);
+        return null;
+      });
+
+      if (routeVehicles) {
+        broadcastToRoute(vehicle.route_id, {
+          type: 'snapshot:route',
+          routeId: vehicle.route_id,
+          vehicles: routeVehicles
+        });
+      }
+
       return res.status(201).json({
         message: 'Telemetry ingested successfully',
         vehicle

backend/src/middleware/passengerTelemetryRateLimiter.js

@@ -0,0 +1,83 @@
+const { hashValue } = require('../services/passengerTelemetryAggregator');
+
+const buckets = new Map();
+
+function cleanupBucket(key, now) {
+  const bucket = buckets.get(key);
+
+  if (bucket && bucket.resetAt <= now) {
+    buckets.delete(key);
+  }
+}
+
+function incrementBucket(key, now, windowMs) {
+  cleanupBucket(key, now);
+
+  const existing = buckets.get(key);
+
+  if (existing) {
+    existing.count += 1;
+    return existing;
+  }
+
+  const bucket = {
+    count: 1,
+    resetAt: now + windowMs
+  };
+
+  buckets.set(key, bucket);
+  return bucket;
+}
+
+function normalizeKeyPart(value, fallback) {
+  if (value === undefined || value === null || value === '') {
+    return fallback;
+  }
+
+  return hashValue(String(value).trim());
+}
+
+function createPassengerTelemetryRateLimiter({
+  maxRequests = 60,
+  windowMs = 60 * 1000,
+  nowProvider = () => Date.now()
+} = {}) {
+  return (req, res, next) => {
+    const now = nowProvider();
+    const sessionId = req.body?.session_id ?? req.body?.sessionId;
+    const sessionKey = normalizeKeyPart(sessionId, 'missing-session');
+    const ipKey = normalizeKeyPart(req.ip || req.socket?.remoteAddress, 'unknown-ip');
+
+    const keys = [
+      `passenger-telemetry:session:${sessionKey}`,
+      `passenger-telemetry:ip:${ipKey}`
+    ];
+
+    if (req.apiKeyInfo?.id) {
+      keys.push(`passenger-telemetry:api-key:${normalizeKeyPart(req.apiKeyInfo.id, 'api-key')}`);
+    }
+
+    const exceeded = keys
+      .map((key) => incrementBucket(key, now, windowMs))
+      .find((bucket) => bucket.count > maxRequests);
+
+    if (exceeded) {
+      return res.status(429).json({
+        error: 'Too many passenger telemetry samples',
+        message: 'Rate limit exceeded. Please try again later.',
+        retryAfter: Math.ceil((exceeded.resetAt - now) / 1000)
+      });
+    }
+
+    return next();
+  };
+}
+
+function resetPassengerTelemetryRateLimitersForTesting() {
+  buckets.clear();
+}
+
+module.exports = {
+  createPassengerTelemetryRateLimiter,
+  resetPassengerTelemetryRateLimitersForTesting
+};

backend/src/routes/api.js

@@ -10,6 +10,8 @@ const holidaysController = require('../controllers/holidaysController');
 const etaController = require('../controllers/etaController');
 const syncController = require('../controllers/syncController');
 const telemetryController = require('../controllers/telemetryController');
+const passengerTelemetryController = require('../controllers/passengerTelemetryController');
+const { createPassengerTelemetryRateLimiter } = require('../middleware/passengerTelemetryRateLimiter');
 
 // Routes
 router.get('/routes', routesController.getAllRoutes.bind(routesController));
@@ -60,6 +62,11 @@ router.get('/eta/stop/:stopId', etaController.getStopEtas.bind(etaController));
 
 // Realtime GPS / Telemetry
 router.post('/telemetry/ingest', telemetryController.ingest.bind(telemetryController));
+router.post(
+  '/passenger-telemetry/ingest',
+  createPassengerTelemetryRateLimiter(),
+  passengerTelemetryController.ingest.bind(passengerTelemetryController)
+);
 router.get('/realtime/routes/:routeId/vehicles', telemetryController.getRouteLiveVehicles.bind(telemetryController));
 router.get('/realtime/vehicles/:vehicleId', telemetryController.getVehicleLiveState.bind(telemetryController));
 

backend/src/services/liveVehicleMapper.js

@@ -0,0 +1,32 @@
+function parseNullableNumber(value) {
+  if (value === undefined || value === null || value === '') {
+    return null;
+  }
+
+  const parsed = Number.parseFloat(value);
+  return Number.isFinite(parsed) ? parsed : null;
+}
+
+function decorateOfficialVehicle(row) {
+  return {
+    ...row,
+    source: 'official',
+    confidence: 1,
+    accuracy_m: parseNullableNumber(row.accuracy_meters),
+    is_estimated: false
+  };
+}
+
+function sortLiveVehicles(a, b) {
+  if (a.source !== b.source) {
+    return a.source === 'official' ? -1 : 1;
+  }
+
+  return new Date(b.last_seen_at).getTime() - new Date(a.last_seen_at).getTime();
+}
+
+module.exports = {
+  decorateOfficialVehicle,
+  parseNullableNumber,
+  sortLiveVehicles
+};

backend/src/services/passengerTelemetryAggregator.js

@@ -0,0 +1,503 @@
+const crypto = require('crypto');
+
+const { createHttpError, validatePassengerTelemetryPayload } = require('./passengerTelemetryValidator');
+
+const EARTH_RADIUS_M = 6371000;
+
+function clamp(value, min, max) {
+  return Math.min(max, Math.max(min, value));
+}
+
+function toRadians(value) {
+  return value * Math.PI / 180;
+}
+
+function hashValue(value) {
+  return crypto.createHash('sha256').update(String(value)).digest('hex');
+}
+
+function haversineDistanceMeters(a, b) {
+  const lat1 = toRadians(a.latitude);
+  const lat2 = toRadians(b.latitude);
+  const deltaLat = toRadians(b.latitude - a.latitude);
+  const deltaLon = toRadians(b.longitude - a.longitude);
+  const sinLat = Math.sin(deltaLat / 2);
+  const sinLon = Math.sin(deltaLon / 2);
+  const h = sinLat * sinLat + Math.cos(lat1) * Math.cos(lat2) * sinLon * sinLon;
+
+  return 2 * EARTH_RADIUS_M * Math.atan2(Math.sqrt(h), Math.sqrt(1 - h));
+}
+
+function bearingDegrees(a, b) {
+  const lat1 = toRadians(a.latitude);
+  const lat2 = toRadians(b.latitude);
+  const deltaLon = toRadians(b.longitude - a.longitude);
+  const y = Math.sin(deltaLon) * Math.cos(lat2);
+  const x = Math.cos(lat1) * Math.sin(lat2) -
+    Math.sin(lat1) * Math.cos(lat2) * Math.cos(deltaLon);
+
+  return (Math.atan2(y, x) * 180 / Math.PI + 360) % 360;
+}
+
+function headingDifferenceDegrees(a, b) {
+  const diff = Math.abs(a - b) % 360;
+  return diff > 180 ? 360 - diff : diff;
+}
+
+function toPlanar(point, referenceLatitude) {
+  return {
+    x: point.longitude * 111320 * Math.cos(toRadians(referenceLatitude)),
+    y: point.latitude * 110540
+  };
+}
+
+function distanceToSegment(point, start, end) {
+  const refLat = (point.latitude + start.latitude + end.latitude) / 3;
+  const p = toPlanar(point, refLat);
+  const a = toPlanar(start, refLat);
+  const b = toPlanar(end, refLat);
+  const dx = b.x - a.x;
+  const dy = b.y - a.y;
+  const lengthSquared = dx * dx + dy * dy;
+
+  if (lengthSquared === 0) {
+    return { distanceM: haversineDistanceMeters(point, start), fraction: 0 };
+  }
+
+  const fraction = clamp(((p.x - a.x) * dx + (p.y - a.y) * dy) / lengthSquared, 0, 1);
+  const x = a.x + fraction * dx;
+  const y = a.y + fraction * dy;
+
+  return {
+    distanceM: Math.sqrt((p.x - x) ** 2 + (p.y - y) ** 2),
+    fraction
+  };
+}
+
+function average(values) {
+  const finite = values.filter((value) => Number.isFinite(value));
+  return finite.length
+    ? finite.reduce((sum, value) => sum + value, 0) / finite.length
+    : null;
+}
+
+function normalizeRoute(row) {
+  const rawStops = Array.isArray(row.stops)
+    ? row.stops
+    : typeof row.stops === 'string'
+      ? JSON.parse(row.stops)
+      : [];
+
+  return {
+    id: Number.parseInt(row.id, 10),
+    route_number: row.route_number,
+    name: row.name,
+    type: row.type,
+    color: row.color,
+    stops: rawStops
+      .map((stop) => ({
+        stop_id: Number.parseInt(stop.stop_id ?? stop.id, 10),
+        sequence: Number.parseInt(stop.sequence, 10),
+        latitude: Number.parseFloat(stop.latitude),
+        longitude: Number.parseFloat(stop.longitude)
+      }))
+      .filter((stop) => (
+        Number.isInteger(stop.stop_id) &&
+        Number.isFinite(stop.latitude) &&
+        Number.isFinite(stop.longitude)
+      ))
+      .sort((a, b) => a.sequence - b.sequence)
+  };
+}
+
+function confidenceLabel(confidence) {
+  if (confidence >= 0.75) return 'HIGH';
+  if (confidence >= 0.5) return 'MEDIUM';
+  return 'LOW';
+}
+
+class PassengerTelemetryAggregator {
+  constructor({
+    pool,
+    nowProvider = () => new Date(),
+    ttlMs = 10 * 60 * 1000,
+    routeMatchMaxDistanceM = 200,
+    maxAggregationAccuracyM = 250,
+    poorAccuracyThresholdM = 100,
+    clusterDistanceM = 160,
+    clusterTimeWindowMs = 3 * 60 * 1000,
+    clusterSpeedWindowKmh = 25,
+    minDistinctSessions = 2,
+    minConfidence = 0.3,
+    officialSuppressionDistanceM = 300
+  }) {
+    if (!pool || typeof pool.query !== 'function') {
+      throw new Error('PassengerTelemetryAggregator requires a pool with query(sql, params)');
+    }
+
+    Object.assign(this, {
+      pool,
+      nowProvider,
+      ttlMs,
+      routeMatchMaxDistanceM,
+      maxAggregationAccuracyM,
+      poorAccuracyThresholdM,
+      clusterDistanceM,
+      clusterTimeWindowMs,
+      clusterSpeedWindowKmh,
+      minDistinctSessions,
+      minConfidence,
+      officialSuppressionDistanceM
+    });
+    this.samples = [];
+  }
+
+  async ingest(payload) {
+    const now = this.nowProvider();
+    this.prune(now);
+
+    const point = validatePassengerTelemetryPayload(payload, {
+      now,
+      maxAgeSeconds: Math.floor(this.ttlMs / 1000)
+    });
+    const match = await this.matchPointToRoute(point);
+
+    if (!match) {
+      return { accepted: true, matched: false, virtualVehicles: [] };
+    }
+
+    if (point.accuracyM !== null && point.accuracyM > this.maxAggregationAccuracyM) {
+      return {
+        accepted: true,
+        matched: true,
+        routeId: match.route.id,
+        ignoredReason: 'accuracy_too_low',
+        virtualVehicles: []
+      };
+    }
+
+    this.samples.push({
+      sample_id: crypto.randomUUID ? crypto.randomUUID() : hashValue(`${Date.now()}:${Math.random()}`),
+      session_key: hashValue(point.sessionId),
+      route_id: match.route.id,
+      route_number: match.route.route_number,
+      route_name: match.route.name,
+      route_type: match.route.type,
+      route_color: match.route.color,
+      latitude: point.latitude,
+      longitude: point.longitude,
+      accuracy_m: point.accuracyM,
+      speed_kmh: point.speedKmh,
+      heading: point.heading,
+      observed_at: point.observedAt,
+      received_at: now,
+      app_state: point.appState,
+      stop_id: point.stopId,
+      matched_distance_m: match.distanceM,
+      progress_m: match.progressM,
+      direction: this.resolveDirection(point.heading, match.bearing)
+    });
+
+    return {
+      accepted: true,
+      matched: true,
+      routeId: match.route.id,
+      virtualVehicles: await this.getVirtualVehiclesForRoute(match.route.id)
+    };
+  }
+
+  async getVirtualVehiclesForRoute(routeId, options = {}) {
+    const routeIdInt = Number.parseInt(routeId, 10);
+
+    if (!Number.isInteger(routeIdInt) || routeIdInt <= 0) {
+      throw createHttpError(400, 'routeId must be a positive integer');
+    }
+
+    const now = this.nowProvider();
+    this.prune(now);
+
+    const routeSamples = this.samples.filter((sample) => (
+      sample.route_id === routeIdInt &&
+      now.getTime() - sample.received_at.getTime() <= this.ttlMs &&
+      sample.matched_distance_m <= this.routeMatchMaxDistanceM &&
+      (sample.accuracy_m === null || sample.accuracy_m <= this.maxAggregationAccuracyM)
+    ));
+
+    if (routeSamples.length < this.minDistinctSessions) {
+      return [];
+    }
+
+    const officialVehicles = Array.isArray(options.officialVehicles) ? options.officialVehicles : [];
+
+    return this.clusterSamples(routeSamples)
+      .map((cluster) => this.buildVirtualVehicle(cluster, now))
+      .filter(Boolean)
+      .filter((vehicle) => !this.isSuppressedByOfficialVehicle(vehicle, officialVehicles));
+  }
+
+  prune(now = this.nowProvider()) {
+    const cutoffMs = now.getTime() - this.ttlMs;
+    this.samples = this.samples.filter((sample) => (
+      sample.received_at.getTime() >= cutoffMs &&
+      sample.observed_at.getTime() >= cutoffMs
+    ));
+  }
+
+  reset() {
+    this.samples = [];
+  }
+
+  async getCandidateRoutes(routeId) {
+    const params = [];
+    let whereClause = 'WHERE r.is_active = true';
+
+    if (routeId) {
+      params.push(routeId);
+      whereClause += ' AND r.id = $1';
+    }
+
+    const result = await this.pool.query(
+      `SELECT
+        r.id,
+        r.route_number,
+        r.name,
+        r.type,
+        r.color,
+        COALESCE(
+          json_agg(
+            json_build_object(
+              'stop_id', s.id,
+              'sequence', rs.sequence,
+              'latitude', ST_Y(s.location::geometry),
+              'longitude', ST_X(s.location::geometry)
+            )
+            ORDER BY rs.sequence
+          ) FILTER (WHERE s.id IS NOT NULL),
+          '[]'::json
+        ) AS stops
+      FROM routes r
+      LEFT JOIN route_stops rs ON rs.route_id = r.id
+      LEFT JOIN stops s ON s.id = rs.stop_id AND s.is_active = true
+      ${whereClause}
+      GROUP BY r.id, r.route_number, r.name, r.type, r.color
+      ORDER BY r.id`,
+      params
+    );
+
+    return result.rows.map(normalizeRoute);
+  }
+
+  async matchPointToRoute(point) {
+    const routes = await this.getCandidateRoutes(point.routeId);
+
+    if (point.routeId && routes.length === 0) {
+      throw createHttpError(404, 'Route not found');
+    }
+
+    const best = routes
+      .map((route) => ({ route, match: this.distanceToRoute(point, route) }))
+      .filter((entry) => entry.match)
+      .sort((a, b) => a.match.distanceM - b.match.distanceM)[0];
+
+    if (!best || best.match.distanceM > this.routeMatchMaxDistanceM) {
+      return null;
+    }
+
+    return { route: best.route, ...best.match };
+  }
+
+  distanceToRoute(point, route) {
+    if (!route.stops || route.stops.length === 0) {
+      return null;
+    }
+
+    if (route.stops.length === 1) {
+      return {
+        distanceM: haversineDistanceMeters(point, route.stops[0]),
+        progressM: 0,
+        bearing: point.heading ?? 0
+      };
+    }
+
+    let progressBeforeM = 0;
+    let best = null;
+
+    for (let index = 0; index < route.stops.length - 1; index += 1) {
+      const start = route.stops[index];
+      const end = route.stops[index + 1];
+      const segmentLengthM = haversineDistanceMeters(start, end);
+      const segment = distanceToSegment(point, start, end);
+      const candidate = {
+        distanceM: segment.distanceM,
+        progressM: progressBeforeM + segment.fraction * segmentLengthM,
+        bearing: bearingDegrees(start, end)
+      };
+
+      if (!best || candidate.distanceM < best.distanceM) {
+        best = candidate;
+      }
+
+      progressBeforeM += segmentLengthM;
+    }
+
+    return best;
+  }
+
+  resolveDirection(heading, routeBearing) {
+    if (!Number.isFinite(heading) || !Number.isFinite(routeBearing)) {
+      return 'unknown';
+    }
+
+    return headingDifferenceDegrees(heading, routeBearing) <= 90 ? 'forward' : 'backward';
+  }
+
+  clusterSamples(samples) {
+    const clusters = [];
+
+    for (const sample of [...samples].sort((a, b) => a.progress_m - b.progress_m)) {
+      const cluster = clusters.find((item) => this.sampleFitsCluster(sample, item));
+
+      if (!cluster) {
+        clusters.push({
+          direction: sample.direction,
+          progress_m: sample.progress_m,
+          observed_at: sample.observed_at,
+          speed_kmh: sample.speed_kmh,
+          samples: [sample]
+        });
+        continue;
+      }
+
+      cluster.samples.push(sample);
+      cluster.progress_m = average(cluster.samples.map((item) => item.progress_m));
+      cluster.observed_at = new Date(Math.max(...cluster.samples.map((item) => item.observed_at.getTime())));
+      cluster.speed_kmh = average(cluster.samples.map((item) => item.speed_kmh));
+    }
+
+    return clusters;
+  }
+
+  sampleFitsCluster(sample, cluster) {
+    const directionsMatch = sample.direction === cluster.direction ||
+      sample.direction === 'unknown' ||
+      cluster.direction === 'unknown';
+    const speedDiff = Number.isFinite(sample.speed_kmh) && Number.isFinite(cluster.speed_kmh)
+      ? Math.abs(sample.speed_kmh - cluster.speed_kmh)
+      : 0;
+
+    return directionsMatch &&
+      Math.abs(sample.progress_m - cluster.progress_m) <= this.clusterDistanceM &&
+      Math.abs(sample.observed_at.getTime() - cluster.observed_at.getTime()) <= this.clusterTimeWindowMs &&
+      speedDiff <= this.clusterSpeedWindowKmh;
+  }
+
+  buildVirtualVehicle(cluster, now) {
+    const samples = cluster.samples;
+    const distinctSessions = new Set(samples.map((sample) => sample.session_key));
+
+    if (distinctSessions.size < this.minDistinctSessions) {
+      return null;
+    }
+
+    const confidence = this.calculateConfidence(samples, distinctSessions.size, now);
+
+    if (confidence < this.minConfidence) {
+      return null;
+    }
+
+    const newest = samples.reduce((latest, sample) => (
+      sample.observed_at > latest.observed_at ? sample : latest
+    ), samples[0]);
+    const position = this.weightedPosition(samples);
+    const clusterId = hashValue(
+      `${newest.route_id}:${cluster.direction}:${Math.round(cluster.progress_m / this.clusterDistanceM)}`
+    ).slice(0, 12);
+    const speedKmh = average(samples.map((sample) => sample.speed_kmh));
+
+    return {
+      vehicle_id: `crowd:${newest.route_id}:${clusterId}`,
+      vehicle_number: null,
+      registration: null,
+      vehicle_type: newest.route_type,
+      capacity: null,
+      route_id: newest.route_id,
+      route_number: newest.route_number,
+      route_name: newest.route_name,
+      route_color: newest.route_color,
+      latitude: position.latitude,
+      longitude: position.longitude,
+      speed_kmh: speedKmh,
+      heading: average(samples.map((sample) => sample.heading)),
+      accuracy_meters: average(samples.map((sample) => sample.accuracy_m)),
+      accuracy_m: average(samples.map((sample) => sample.accuracy_m)),
+      source_type: 'passenger',
+      source_ref: null,
+      source: 'crowdsourced',
+      confidence,
+      confidence_label: confidenceLabel(confidence),
+      is_estimated: true,
+      last_seen_at: newest.observed_at.toISOString(),
+      updated_at: now.toISOString(),
+      is_online: true,
+      seconds_since_update: Math.max(0, Math.floor((now.getTime() - newest.observed_at.getTime()) / 1000)),
+      is_moving: Number.isFinite(speedKmh) ? speedKmh >= 1 : null
+    };
+  }
+
+  calculateConfidence(samples, distinctSessionCount, now) {
+    const avgAccuracy = average(samples.map((sample) => sample.accuracy_m)) ?? this.poorAccuracyThresholdM;
+    const avgDistance = average(samples.map((sample) => sample.matched_distance_m)) ?? this.routeMatchMaxDistanceM;
+    const newestObservedAt = Math.max(...samples.map((sample) => sample.observed_at.getTime()));
+    const sessionFactor = clamp((distinctSessionCount - 1) / 3, 0, 1);
+    const accuracyFactor = clamp(1 - (avgAccuracy / this.maxAggregationAccuracyM), 0, 1);
+    const distanceFactor = clamp(1 - (avgDistance / this.routeMatchMaxDistanceM), 0, 1);
+    const recencyFactor = clamp(1 - ((now.getTime() - newestObservedAt) / this.ttlMs), 0, 1);
+    const confidence = 0.2 + sessionFactor * 0.35 + accuracyFactor * 0.2 + distanceFactor * 0.15 + recencyFactor * 0.1;
+
+    return Number(clamp(confidence, 0, 0.95).toFixed(3));
+  }
+
+  weightedPosition(samples) {
+    let weightSum = 0;
+    let latSum = 0;
+    let lonSum = 0;
+
+    for (const sample of samples) {
+      const accuracy = Number.isFinite(sample.accuracy_m) && sample.accuracy_m > 0
+        ? sample.accuracy_m
+        : this.poorAccuracyThresholdM;
+      const weight = 1 / Math.max(accuracy, 5);
+      weightSum += weight;
+      latSum += sample.latitude * weight;
+      lonSum += sample.longitude * weight;
+    }
+
+    return {
+      latitude: Number((latSum / weightSum).toFixed(6)),
+      longitude: Number((lonSum / weightSum).toFixed(6))
+    };
+  }
+
+  isSuppressedByOfficialVehicle(vehicle, officialVehicles) {
+    return officialVehicles.some((official) => {
+      if (!official || official.is_online === false) return false;
+
+      const latitude = Number.parseFloat(official.latitude);
+      const longitude = Number.parseFloat(official.longitude);
+
+      if (!Number.isFinite(latitude) || !Number.isFinite(longitude)) return false;
+
+      return haversineDistanceMeters(
+        { latitude: vehicle.latitude, longitude: vehicle.longitude },
+        { latitude, longitude }
+      ) <= this.officialSuppressionDistanceM;
+    });
+  }
+}
+
+module.exports = {
+  PassengerTelemetryAggregator,
+  bearingDegrees,
+  haversineDistanceMeters,
+  hashValue
+};

backend/src/services/passengerTelemetryService.js

@@ -0,0 +1,21 @@
+const { PassengerTelemetryAggregator } = require('./passengerTelemetryAggregator');
+
+let singleton = null;
+
+function getPassengerTelemetryService() {
+  if (!singleton) {
+    const pool = require('../config/database');
+    singleton = new PassengerTelemetryAggregator({ pool });
+  }
+
+  return singleton;
+}
+
+function setPassengerTelemetryServiceForTesting(service) {
+  singleton = service;
+}
+
+module.exports = {
+  getPassengerTelemetryService,
+  setPassengerTelemetryServiceForTesting
+};

backend/src/services/passengerTelemetryValidator.js

@@ -0,0 +1,130 @@
+const DEFAULT_MAX_FUTURE_SECONDS = 120;
+const DEFAULT_MAX_AGE_SECONDS = 10 * 60;
+const MAX_REASONABLE_ACCURACY_M = 1000;
+const MAX_CITY_TRANSIT_SPEED_KMH = 120;
+
+function createHttpError(status, message) {
+  const error = new Error(message);
+  error.status = status;
+  return error;
+}
+
+function numberValue(value, fieldName, required = false) {
+  if (!required && (value === undefined || value === null || value === '')) {
+    return null;
+  }
+
+  const parsed = Number.parseFloat(value);
+  if (!Number.isFinite(parsed)) {
+    throw createHttpError(400, `${fieldName} must be a valid number`);
+  }
+
+  return parsed;
+}
+
+function positiveInteger(value, fieldName) {
+  if (value === undefined || value === null || value === '') {
+    return null;
+  }
+
+  const parsed = Number.parseInt(value, 10);
+  if (!Number.isInteger(parsed) || parsed <= 0) {
+    throw createHttpError(400, `${fieldName} must be a positive integer`);
+  }
+
+  return parsed;
+}
+
+function sessionIdValue(value) {
+  if (typeof value !== 'string' || !value.trim()) {
+    throw createHttpError(400, 'session_id must be a non-empty string');
+  }
+
+  const sessionId = value.trim();
+  if (sessionId.length > 128) {
+    throw createHttpError(400, 'session_id must be 128 characters or fewer');
+  }
+
+  return sessionId;
+}
+
+function observedAtValue(value, now, maxFutureSeconds, maxAgeSeconds) {
+  const observedAt = value ? new Date(value) : new Date(now);
+
+  if (Number.isNaN(observedAt.getTime())) {
+    throw createHttpError(400, 'observed_at must be a valid ISO datetime string');
+  }
+
+  const deltaMs = observedAt.getTime() - now.getTime();
+  if (deltaMs > maxFutureSeconds * 1000) {
+    throw createHttpError(400, 'observed_at is too far in the future');
+  }
+
+  if (deltaMs < -maxAgeSeconds * 1000) {
+    throw createHttpError(400, 'observed_at is too old');
+  }
+
+  return observedAt;
+}
+
+function validatePassengerTelemetryPayload(payload, options = {}) {
+  const now = options.now ? new Date(options.now) : new Date();
+  const sessionId = sessionIdValue(payload.session_id ?? payload.sessionId);
+  const latitude = numberValue(payload.latitude, 'latitude', true);
+  const longitude = numberValue(payload.longitude, 'longitude', true);
+
+  if (latitude < -90 || latitude > 90) {
+    throw createHttpError(400, 'latitude must be between -90 and 90');
+  }
+
+  if (longitude < -180 || longitude > 180) {
+    throw createHttpError(400, 'longitude must be between -180 and 180');
+  }
+
+  const accuracyM = numberValue(payload.accuracy_m ?? payload.accuracyMeters, 'accuracy_m');
+  if (accuracyM !== null && (accuracyM < 0 || accuracyM > MAX_REASONABLE_ACCURACY_M)) {
+    throw createHttpError(400, `accuracy_m must be between 0 and ${MAX_REASONABLE_ACCURACY_M}`);
+  }
+
+  const speedKmh = numberValue(payload.speed_kmh ?? payload.speedKmh, 'speed_kmh');
+  if (speedKmh !== null && (speedKmh < 0 || speedKmh > MAX_CITY_TRANSIT_SPEED_KMH)) {
+    throw createHttpError(400, `speed_kmh must be between 0 and ${MAX_CITY_TRANSIT_SPEED_KMH}`);
+  }
+
+  const heading = numberValue(payload.heading, 'heading');
+  if (heading !== null && (heading < 0 || heading > 360)) {
+    throw createHttpError(400, 'heading must be between 0 and 360');
+  }
+
+  const appState = payload.app_state ?? payload.appState ?? null;
+  if (appState !== null && appState !== undefined && !['foreground', 'background'].includes(appState)) {
+    throw createHttpError(400, 'app_state must be foreground or background');
+  }
+
+  return {
+    sessionId,
+    latitude,
+    longitude,
+    accuracyM,
+    speedKmh,
+    heading,
+    observedAt: observedAtValue(
+      payload.observed_at ?? payload.observedAt,
+      now,
+      options.maxFutureSeconds ?? DEFAULT_MAX_FUTURE_SECONDS,
+      options.maxAgeSeconds ?? DEFAULT_MAX_AGE_SECONDS
+    ),
+    routeId: positiveInteger(payload.route_id ?? payload.routeId, 'route_id'),
+    stopId: positiveInteger(payload.stop_id ?? payload.stopId, 'stop_id'),
+    appState: appState || null
+  };
+}
+
+module.exports = {
+  DEFAULT_MAX_AGE_SECONDS,
+  DEFAULT_MAX_FUTURE_SECONDS,
+  MAX_CITY_TRANSIT_SPEED_KMH,
+  MAX_REASONABLE_ACCURACY_M,
+  createHttpError,
+  validatePassengerTelemetryPayload
+};

backend/src/services/telemetryService.js

@@ -1,6 +1,8 @@
 const pool = require('../config/database');
 const logger = require('../config/logger');
 const { ensureRealtimeSchema } = require('./realtimeSchema');
+const { getPassengerTelemetryService } = require('./passengerTelemetryService');
+const { decorateOfficialVehicle, sortLiveVehicles } = require('./liveVehicleMapper');
 
 const DEFAULT_STALE_AFTER_SECONDS = 180;
 
@@ -243,7 +245,15 @@ class TelemetryService {
     `;
 
     const result = await pool.query(query, [routeIdInt, staleAfterSeconds]);
-    return result.rows;
+    const officialVehicles = result.rows.map(decorateOfficialVehicle);
+    const crowdsourcedVehicles = await getPassengerTelemetryService()
+      .getVirtualVehiclesForRoute(routeIdInt, { officialVehicles })
+      .catch((error) => {
+        logger.error(`Failed to build passenger telemetry vehicles for route ${routeIdInt}:`, error);
+        return [];
+      });
+
+    return [...officialVehicles, ...crowdsourcedVehicles].sort(sortLiveVehicles);
   }
 
   async getVehicleLiveState(vehicleId, staleAfterSeconds = DEFAULT_STALE_AFTER_SECONDS) {
@@ -293,7 +303,7 @@ class TelemetryService {
       throw createHttpError(404, 'Live state not found for this vehicle');
     }
 
-    return result.rows[0];
+    return decorateOfficialVehicle(result.rows[0]);
   }
 
   async getRoute(routeId) {

backend/test/passengerTelemetryAggregator.test.js

@@ -0,0 +1,142 @@
+const assert = require('node:assert/strict');
+const test = require('node:test');
+
+const { PassengerTelemetryAggregator } = require('../src/services/passengerTelemetryAggregator');
+
+const NOW = new Date('2026-05-20T12:30:00Z');
+const routeRows = [{
+  id: 1,
+  route_number: '12',
+  name: 'Route 12',
+  type: 'bus',
+  color: '#0066CC',
+  stops: [
+    { stop_id: 1, sequence: 1, latitude: 55.485, longitude: 28.768 },
+    { stop_id: 2, sequence: 2, latitude: 55.485, longitude: 28.778 }
+  ]
+}];
+
+function createPool(rows = routeRows) {
+  return {
+    async query(sql, params = []) {
+      if (params.length > 0) {
+        return { rows: rows.filter((row) => row.id === params[0]) };
+      }
+
+      return { rows };
+    }
+  };
+}
+
+function createAggregator(options = {}) {
+  return new PassengerTelemetryAggregator({
+    pool: createPool(options.rows),
+    nowProvider: () => NOW,
+    ...options
+  });
+}
+
+async function ingestCluster(aggregator, accuracyM = 18) {
+  await aggregator.ingest({
+    session_id: 'session-a',
+    latitude: 55.48502,
+    longitude: 28.7700,
+    accuracy_m: accuracyM,
+    speed_kmh: 31,
+    heading: 90,
+    observed_at: '2026-05-20T12:29:55Z',
+    route_id: 1
+  });
+
+  await aggregator.ingest({
+    session_id: 'session-b',
+    latitude: 55.48498,
+    longitude: 28.77015,
+    accuracy_m: accuracyM,
+    speed_kmh: 33,
+    heading: 92,
+    observed_at: '2026-05-20T12:29:58Z',
+    route_id: 1
+  });
+}
+
+test('one passenger point does not create a virtual vehicle', async () => {
+  const aggregator = createAggregator();
+  const result = await aggregator.ingest({
+    session_id: 'session-a',
+    latitude: 55.485,
+    longitude: 28.770,
+    accuracy_m: 18,
+    speed_kmh: 32,
+    heading: 90,
+    observed_at: '2026-05-20T12:29:55Z',
+    route_id: 1
+  });
+
+  assert.equal(result.accepted, true);
+  assert.equal(result.matched, true);
+  assert.deepEqual(result.virtualVehicles, []);
+  assert.deepEqual(await aggregator.getVirtualVehiclesForRoute(1), []);
+});
+
+test('nearby points from distinct sessions create a virtual crowdsourced vehicle', async () => {
+  const aggregator = createAggregator();
+  await ingestCluster(aggregator);
+  const vehicles = await aggregator.getVirtualVehiclesForRoute(1);
+
+  assert.equal(vehicles.length, 1);
+  assert.match(vehicles[0].vehicle_id, /^crowd:1:/);
+  assert.equal(vehicles[0].source, 'crowdsourced');
+  assert.equal(vehicles[0].source_ref, null);
+  assert.equal(vehicles[0].is_estimated, true);
+  assert.equal('session_id' in vehicles[0], false);
+  assert.equal('session_key' in vehicles[0], false);
+});
+
+test('poor accuracy lowers crowdsourced confidence', async () => {
+  const goodAggregator = createAggregator();
+  await ingestCluster(goodAggregator, 15);
+  const [goodVehicle] = await goodAggregator.getVirtualVehiclesForRoute(1);
+
+  const poorAggregator = createAggregator();
+  await ingestCluster(poorAggregator, 150);
+  const [poorVehicle] = await poorAggregator.getVirtualVehiclesForRoute(1);
+
+  assert.ok(goodVehicle.confidence > poorVehicle.confidence);
+  assert.equal(poorVehicle.source, 'crowdsourced');
+});
+
+test('points outside route geometry are accepted but ignored', async () => {
+  const aggregator = createAggregator();
+  const result = await aggregator.ingest({
+    session_id: 'session-a',
+    latitude: 56.0,
+    longitude: 29.5,
+    accuracy_m: 20,
+    observed_at: '2026-05-20T12:29:55Z',
+    route_id: 1
+  });
+
+  assert.equal(result.accepted, true);
+  assert.equal(result.matched, false);
+  assert.deepEqual(await aggregator.getVirtualVehiclesForRoute(1), []);
+});
+
+test('official telemetry suppresses nearby crowdsourced virtual vehicle', async () => {
+  const aggregator = createAggregator();
+  await ingestCluster(aggregator);
+  const [vehicle] = await aggregator.getVirtualVehiclesForRoute(1);
+
+  assert.ok(vehicle);
+
+  const suppressed = await aggregator.getVirtualVehiclesForRoute(1, {
+    officialVehicles: [{
+      vehicle_id: 99,
+      latitude: vehicle.latitude,
+      longitude: vehicle.longitude,
+      is_online: true
+    }]
+  });
+
+  assert.deepEqual(suppressed, []);
+});

backend/test/passengerTelemetryApi.test.js

@@ -0,0 +1,110 @@
+const assert = require('node:assert/strict');
+const test = require('node:test');
+
+const { PassengerTelemetryController } = require('../src/controllers/passengerTelemetryController');
+const { decorateOfficialVehicle } = require('../src/services/liveVehicleMapper');
+
+function createResponse() {
+  return {
+    statusCode: null,
+    body: null,
+    status(code) {
+      this.statusCode = code;
+      return this;
+    },
+    json(payload) {
+      this.body = payload;
+      return this;
+    }
+  };
+}
+
+test('POST passenger telemetry ingest returns accepted response and broadcasts virtual vehicles', async () => {
+  const broadcasts = [];
+  const service = {
+    async ingest(payload) {
+      assert.equal(payload.session_id, 'temporary-anonymous-id');
+      return {
+        accepted: true,
+        routeId: 12,
+        virtualVehicles: [
+          {
+            vehicle_id: 'crowd:12:abc123',
+            route_id: 12,
+            source: 'crowdsourced',
+            source_ref: null,
+            is_estimated: true
+          }
+        ]
+      };
+    }
+  };
+
+  const controller = new PassengerTelemetryController({
+    service,
+    broadcastToRoute: (routeId, message) => broadcasts.push({ routeId, message })
+  });
+  controller.getRouteSnapshot = async () => [
+    {
+      vehicle_id: 'crowd:12:abc123',
+      route_id: 12,
+      source: 'crowdsourced',
+      source_ref: null,
+      is_estimated: true
+    }
+  ];
+  const res = createResponse();
+
+  await controller.ingest({
+    body: {
+      session_id: 'temporary-anonymous-id',
+      latitude: 55.485,
+      longitude: 28.768
+    }
+  }, res);
+
+  assert.equal(res.statusCode, 202);
+  assert.deepEqual(res.body, { accepted: true });
+  assert.equal(broadcasts.length, 1);
+  assert.equal(broadcasts[0].routeId, 12);
+  assert.equal(broadcasts[0].message.type, 'snapshot:route');
+  assert.equal(broadcasts[0].message.vehicles[0].source, 'crowdsourced');
+  assert.equal('session_id' in broadcasts[0].message.vehicles[0], false);
+});
+
+test('passenger telemetry controller maps validation failures to 400', async () => {
+  const service = {
+    async ingest() {
+      const error = new Error('latitude must be between -90 and 90');
+      error.status = 400;
+      throw error;
+    }
+  };
+  const controller = new PassengerTelemetryController({
+    service,
+    broadcastToRoute: () => {}
+  });
+  const res = createResponse();
+
+  await controller.ingest({ body: {} }, res);
+
+  assert.equal(res.statusCode, 400);
+  assert.deepEqual(res.body, { error: 'latitude must be between -90 and 90' });
+});
+
+test('official live vehicle mapping adds optional realtime fields', () => {
+  const vehicle = decorateOfficialVehicle({
+    vehicle_id: 7,
+    vehicle_number: 'DEMO-7A',
+    route_id: 12,
+    latitude: '55.485',
+    longitude: '28.768',
+    accuracy_meters: '5.5',
+    last_seen_at: '2026-05-20T12:29:58Z'
+  });
+
+  assert.equal(vehicle.source, 'official');
+  assert.equal(vehicle.confidence, 1);
+  assert.equal(vehicle.accuracy_m, 5.5);
+  assert.equal(vehicle.is_estimated, false);
+});

backend/test/passengerTelemetryValidator.test.js

@@ -0,0 +1,70 @@
+const assert = require('node:assert/strict');
+const test = require('node:test');
+
+const { validatePassengerTelemetryPayload } = require('../src/services/passengerTelemetryValidator');
+
+const NOW = new Date('2026-05-20T12:30:00Z');
+
+test('validates and normalizes passenger telemetry payload', () => {
+  const payload = validatePassengerTelemetryPayload({
+    session_id: ' temporary-anonymous-id ',
+    latitude: '55.485',
+    longitude: '28.768',
+    accuracy_m: '18.5',
+    speed_kmh: '32',
+    heading: '91',
+    observed_at: '2026-05-20T12:29:30Z',
+    route_id: '12',
+    stop_id: '123',
+    app_state: 'foreground'
+  }, { now: NOW });
+
+  assert.equal(payload.sessionId, 'temporary-anonymous-id');
+  assert.equal(payload.latitude, 55.485);
+  assert.equal(payload.longitude, 28.768);
+  assert.equal(payload.accuracyM, 18.5);
+  assert.equal(payload.speedKmh, 32);
+  assert.equal(payload.heading, 91);
+  assert.equal(payload.routeId, 12);
+  assert.equal(payload.stopId, 123);
+  assert.equal(payload.appState, 'foreground');
+});
+
+test('rejects invalid latitude and longitude', () => {
+  assert.throws(() => validatePassengerTelemetryPayload({
+    session_id: 's1',
+    latitude: 91,
+    longitude: 28.768
+  }, { now: NOW }), /latitude must be between -90 and 90/);
+
+  assert.throws(() => validatePassengerTelemetryPayload({
+    session_id: 's1',
+    latitude: 55.485,
+    longitude: 181
+  }, { now: NOW }), /longitude must be between -180 and 180/);
+});
+
+test('rejects stale and far-future observations', () => {
+  assert.throws(() => validatePassengerTelemetryPayload({
+    session_id: 's1',
+    latitude: 55.485,
+    longitude: 28.768,
+    observed_at: '2026-05-20T12:10:00Z'
+  }, { now: NOW }), /observed_at is too old/);
+
+  assert.throws(() => validatePassengerTelemetryPayload({
+    session_id: 's1',
+    latitude: 55.485,
+    longitude: 28.768,
+    observed_at: '2026-05-20T12:40:00Z'
+  }, { now: NOW }), /observed_at is too far in the future/);
+});
+
+test('rejects implausible city transit speed', () => {
+  assert.throws(() => validatePassengerTelemetryPayload({
+    session_id: 's1',
+    latitude: 55.485,
+    longitude: 28.768,
+    speed_kmh: 180
+  }, { now: NOW }), /speed_kmh must be between 0 and 120/);
+});