Initial commit

backend/src/services/realtimeSchema.js

@@ -0,0 +1,30 @@
+const fs = require('fs').promises;
+const path = require('path');
+
+const pool = require('../config/database');
+const logger = require('../config/logger');
+
+let ensurePromise = null;
+
+async function ensureRealtimeSchema() {
+  if (ensurePromise) {
+    return ensurePromise;
+  }
+
+  ensurePromise = (async () => {
+    const migrationPath = path.join(__dirname, '../../migrations/005_realtime_gps.sql');
+    const sql = await fs.readFile(migrationPath, 'utf8');
+
+    await pool.query(sql);
+    logger.info('Realtime GPS schema ensured');
+  })().catch((error) => {
+    ensurePromise = null;
+    throw error;
+  });
+
+  return ensurePromise;
+}
+
+module.exports = {
+  ensureRealtimeSchema
+};

backend/src/services/realtimeService.js

@@ -0,0 +1,304 @@
+const WebSocket = require('ws');
+
+const logger = require('../config/logger');
+const telemetryService = require('./telemetryService');
+const { verifyJwtToken, getApiKeyInfo } = require('../middleware/auth');
+
+const routeSubscriptions = new Map();
+
+let webSocketServer = null;
+let heartbeatInterval = null;
+
+function sendMessage(ws, payload) {
+  if (!ws || ws.readyState !== WebSocket.OPEN) {
+    return;
+  }
+
+  ws.send(JSON.stringify(payload));
+}
+
+function addRouteSubscription(ws, routeId) {
+  const routeKey = String(routeId);
+
+  if (!routeSubscriptions.has(routeKey)) {
+    routeSubscriptions.set(routeKey, new Set());
+  }
+
+  routeSubscriptions.get(routeKey).add(ws);
+  ws.subscriptions.add(routeKey);
+}
+
+function removeRouteSubscription(ws, routeId) {
+  const routeKey = String(routeId);
+  const subscribers = routeSubscriptions.get(routeKey);
+
+  if (!subscribers) {
+    return;
+  }
+
+  subscribers.delete(ws);
+  ws.subscriptions.delete(routeKey);
+
+  if (subscribers.size === 0) {
+    routeSubscriptions.delete(routeKey);
+  }
+}
+
+function cleanupSocket(ws) {
+  if (!ws.subscriptions) {
+    return;
+  }
+
+  for (const routeId of ws.subscriptions) {
+    removeRouteSubscription(ws, routeId);
+  }
+}
+
+async function handleAuthMessage(ws, payload) {
+  const token = typeof payload.token === 'string' ? payload.token.trim() : '';
+  const apiKey = typeof payload.apiKey === 'string' ? payload.apiKey.trim() : '';
+
+  if (!token && !apiKey) {
+    sendMessage(ws, {
+      type: 'error',
+      message: 'Provide token or apiKey in auth message'
+    });
+    return;
+  }
+
+  if (token) {
+    const user = verifyJwtToken(token);
+
+    if (!user) {
+      sendMessage(ws, {
+        type: 'error',
+        message: 'Invalid or expired token'
+      });
+      return;
+    }
+
+    ws.auth = {
+      type: 'jwt',
+      user
+    };
+
+    sendMessage(ws, {
+      type: 'auth:ok',
+      authType: 'jwt',
+      user: {
+        id: user.id,
+        username: user.username,
+        role: user.role
+      }
+    });
+
+    return;
+  }
+
+  const apiKeyInfo = await getApiKeyInfo(apiKey, true);
+
+  if (!apiKeyInfo) {
+    sendMessage(ws, {
+      type: 'error',
+      message: 'Invalid API key'
+    });
+    return;
+  }
+
+  ws.auth = {
+    type: 'apiKey',
+    apiKeyInfo
+  };
+
+  sendMessage(ws, {
+    type: 'auth:ok',
+    authType: 'apiKey',
+    appName: apiKeyInfo.app_name
+  });
+}
+
+async function handleSubscribeRoute(ws, payload) {
+  if (!ws.auth) {
+    sendMessage(ws, {
+      type: 'error',
+      message: 'Authenticate first'
+    });
+    return;
+  }
+
+  const routeId = Number.parseInt(payload.routeId, 10);
+
+  if (!Number.isInteger(routeId) || routeId <= 0) {
+    sendMessage(ws, {
+      type: 'error',
+      message: 'routeId must be a positive integer'
+    });
+    return;
+  }
+
+  addRouteSubscription(ws, routeId);
+
+  const vehicles = await telemetryService.getRouteLiveVehicles(routeId).catch((error) => {
+    logger.error('Failed to build realtime route snapshot:', error);
+    return [];
+  });
+
+  sendMessage(ws, {
+    type: 'subscription:ok',
+    routeId
+  });
+
+  sendMessage(ws, {
+    type: 'snapshot:route',
+    routeId,
+    vehicles
+  });
+}
+
+function handleUnsubscribeRoute(ws, payload) {
+  const routeId = Number.parseInt(payload.routeId, 10);
+
+  if (!Number.isInteger(routeId) || routeId <= 0) {
+    sendMessage(ws, {
+      type: 'error',
+      message: 'routeId must be a positive integer'
+    });
+    return;
+  }
+
+  removeRouteSubscription(ws, routeId);
+
+  sendMessage(ws, {
+    type: 'subscription:removed',
+    routeId
+  });
+}
+
+function broadcastToRoute(routeId, payload) {
+  const subscribers = routeSubscriptions.get(String(routeId));
+
+  if (!subscribers || subscribers.size === 0) {
+    return;
+  }
+
+  for (const ws of subscribers) {
+    sendMessage(ws, payload);
+  }
+}
+
+function attachRealtimeServer(server) {
+  if (webSocketServer) {
+    return webSocketServer;
+  }
+
+  webSocketServer = new WebSocket.Server({
+    server,
+    path: '/ws/realtime'
+  });
+
+  webSocketServer.on('connection', (ws, request) => {
+    ws.isAlive = true;
+    ws.subscriptions = new Set();
+    ws.auth = null;
+
+    logger.info(`Realtime websocket connected from ${request.socket.remoteAddress}`);
+
+    sendMessage(ws, {
+      type: 'welcome',
+      message: 'Realtime GPS websocket connected'
+    });
+
+    ws.on('pong', () => {
+      ws.isAlive = true;
+    });
+
+    ws.on('message', async (rawMessage) => {
+      try {
+        const payload = JSON.parse(rawMessage.toString());
+
+        switch (payload.type) {
+          case 'auth':
+            await handleAuthMessage(ws, payload);
+            break;
+          case 'subscribe:route':
+            await handleSubscribeRoute(ws, payload);
+            break;
+          case 'unsubscribe:route':
+            handleUnsubscribeRoute(ws, payload);
+            break;
+          case 'ping':
+            sendMessage(ws, { type: 'pong', timestamp: new Date().toISOString() });
+            break;
+          default:
+            sendMessage(ws, {
+              type: 'error',
+              message: `Unsupported message type: ${payload.type}`
+            });
+        }
+      } catch (error) {
+        logger.error('Realtime websocket message error:', error);
+        sendMessage(ws, {
+          type: 'error',
+          message: 'Invalid realtime message payload'
+        });
+      }
+    });
+
+    ws.on('close', () => {
+      cleanupSocket(ws);
+      logger.info('Realtime websocket disconnected');
+    });
+
+    ws.on('error', (error) => {
+      logger.error('Realtime websocket error:', error);
+      cleanupSocket(ws);
+    });
+  });
+
+  heartbeatInterval = setInterval(() => {
+    if (!webSocketServer) {
+      return;
+    }
+
+    webSocketServer.clients.forEach((ws) => {
+      if (ws.isAlive === false) {
+        cleanupSocket(ws);
+        return ws.terminate();
+      }
+
+      ws.isAlive = false;
+      ws.ping();
+    });
+  }, 30000);
+
+  webSocketServer.on('close', () => {
+    if (heartbeatInterval) {
+      clearInterval(heartbeatInterval);
+      heartbeatInterval = null;
+    }
+  });
+
+  logger.info('Realtime websocket server attached at /ws/realtime');
+
+  return webSocketServer;
+}
+
+function closeRealtimeServer() {
+  if (!webSocketServer) {
+    return;
+  }
+
+  if (heartbeatInterval) {
+    clearInterval(heartbeatInterval);
+    heartbeatInterval = null;
+  }
+
+  webSocketServer.close();
+  webSocketServer = null;
+}
+
+module.exports = {
+  attachRealtimeServer,
+  broadcastToRoute,
+  closeRealtimeServer
+};

backend/src/services/telemetryService.js

@@ -0,0 +1,386 @@
+const pool = require('../config/database');
+const logger = require('../config/logger');
+const { ensureRealtimeSchema } = require('./realtimeSchema');
+
+const DEFAULT_STALE_AFTER_SECONDS = 180;
+
+function createHttpError(status, message) {
+  const error = new Error(message);
+  error.status = status;
+  return error;
+}
+
+function parseCoordinate(value, fieldName) {
+  const parsed = Number.parseFloat(value);
+
+  if (!Number.isFinite(parsed)) {
+    throw createHttpError(400, `${fieldName} must be a valid number`);
+  }
+
+  return parsed;
+}
+
+function parseOptionalNumber(value, fieldName) {
+  if (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;
+}
+
+class TelemetryService {
+  async ingestTelemetry(payload) {
+    await ensureRealtimeSchema();
+
+    const lat = parseCoordinate(payload.lat, 'lat');
+    const lon = parseCoordinate(payload.lon, 'lon');
+
+    if (lat < -90 || lat > 90) {
+      throw createHttpError(400, 'lat must be between -90 and 90');
+    }
+
+    if (lon < -180 || lon > 180) {
+      throw createHttpError(400, 'lon must be between -180 and 180');
+    }
+
+    const speedKmh = parseOptionalNumber(payload.speedKmh, 'speedKmh');
+    const heading = parseOptionalNumber(payload.heading, 'heading');
+    const accuracyMeters = parseOptionalNumber(payload.accuracyMeters, 'accuracyMeters');
+
+    if (heading !== null && (heading < 0 || heading > 360)) {
+      throw createHttpError(400, 'heading must be between 0 and 360');
+    }
+
+    if (!payload.vehicleId && !payload.vehicleNumber) {
+      throw createHttpError(400, 'vehicleId or vehicleNumber is required');
+    }
+
+    const route = await this.getRoute(payload.routeId);
+    const vehicle = await this.resolveVehicle({
+      vehicleId: payload.vehicleId,
+      vehicleNumber: payload.vehicleNumber,
+      vehicleType: payload.vehicleType,
+      route
+    });
+
+    const effectiveRouteId = route?.id || vehicle.route_id;
+
+    if (!effectiveRouteId) {
+      throw createHttpError(400, 'routeId is required when the vehicle is not already assigned to a route');
+    }
+
+    if (vehicle.route_id !== effectiveRouteId) {
+      await pool.query(
+        'UPDATE vehicles SET route_id = $1, updated_at = CURRENT_TIMESTAMP WHERE id = $2',
+        [effectiveRouteId, vehicle.id]
+      );
+    }
+
+    const recordedAt = payload.recordedAt ? new Date(payload.recordedAt) : new Date();
+
+    if (Number.isNaN(recordedAt.getTime())) {
+      throw createHttpError(400, 'recordedAt must be a valid ISO datetime string');
+    }
+
+    const sourceType = payload.sourceType || 'manual';
+    const sourceRef = payload.sourceRef || null;
+    const rawPayload = payload.rawPayload || payload;
+
+    await pool.query(
+      `INSERT INTO telemetry_samples (
+        vehicle_id,
+        route_id,
+        source_type,
+        source_ref,
+        location,
+        speed_kmh,
+        heading,
+        accuracy_meters,
+        recorded_at,
+        raw_payload
+      ) VALUES (
+        $1,
+        $2,
+        $3,
+        $4,
+        ST_SetSRID(ST_MakePoint($6, $5), 4326)::geography,
+        $7,
+        $8,
+        $9,
+        $10,
+        $11::jsonb
+      )`,
+      [
+        vehicle.id,
+        effectiveRouteId,
+        sourceType,
+        sourceRef,
+        lat,
+        lon,
+        speedKmh,
+        heading,
+        accuracyMeters,
+        recordedAt.toISOString(),
+        JSON.stringify(rawPayload)
+      ]
+    );
+
+    await pool.query(
+      `INSERT INTO vehicle_live_state (
+        vehicle_id,
+        route_id,
+        location,
+        speed_kmh,
+        heading,
+        accuracy_meters,
+        source_type,
+        source_ref,
+        last_seen_at
+      ) VALUES (
+        $1,
+        $2,
+        ST_SetSRID(ST_MakePoint($4, $3), 4326)::geography,
+        $5,
+        $6,
+        $7,
+        $8,
+        $9,
+        $10
+      )
+      ON CONFLICT (vehicle_id) DO UPDATE SET
+        route_id = EXCLUDED.route_id,
+        location = EXCLUDED.location,
+        speed_kmh = EXCLUDED.speed_kmh,
+        heading = EXCLUDED.heading,
+        accuracy_meters = EXCLUDED.accuracy_meters,
+        source_type = EXCLUDED.source_type,
+        source_ref = EXCLUDED.source_ref,
+        last_seen_at = EXCLUDED.last_seen_at,
+        updated_at = CURRENT_TIMESTAMP`,
+      [
+        vehicle.id,
+        effectiveRouteId,
+        lat,
+        lon,
+        speedKmh,
+        heading,
+        accuracyMeters,
+        sourceType,
+        sourceRef,
+        recordedAt.toISOString()
+      ]
+    );
+
+    const liveState = await this.getVehicleLiveState(vehicle.id, DEFAULT_STALE_AFTER_SECONDS);
+
+    logger.info(`Telemetry ingested for vehicle ${vehicle.id} on route ${effectiveRouteId}`);
+
+    return liveState;
+  }
+
+  async getRouteLiveVehicles(routeId, options = {}) {
+    await ensureRealtimeSchema();
+
+    const routeIdInt = Number.parseInt(routeId, 10);
+
+    if (!Number.isInteger(routeIdInt) || routeIdInt <= 0) {
+      throw createHttpError(400, 'routeId must be a positive integer');
+    }
+
+    const staleAfterSeconds = Number.parseInt(
+      options.staleAfterSeconds ?? DEFAULT_STALE_AFTER_SECONDS,
+      10
+    );
+    const includeStale = Boolean(options.includeStale);
+
+    let query = `
+      SELECT
+        vls.vehicle_id,
+        v.vehicle_number,
+        v.registration,
+        v.type AS vehicle_type,
+        v.capacity,
+        vls.route_id,
+        r.route_number,
+        r.name AS route_name,
+        r.color AS route_color,
+        ST_Y(vls.location::geometry) AS latitude,
+        ST_X(vls.location::geometry) AS longitude,
+        vls.speed_kmh,
+        vls.heading,
+        vls.accuracy_meters,
+        vls.source_type,
+        vls.source_ref,
+        vls.last_seen_at,
+        vls.updated_at,
+        CASE
+          WHEN vls.last_seen_at >= CURRENT_TIMESTAMP - ($2 * INTERVAL '1 second') THEN true
+          ELSE false
+        END AS is_online,
+        GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - vls.last_seen_at))))::int AS seconds_since_update,
+        CASE WHEN COALESCE(vls.speed_kmh, 0) >= 1 THEN true ELSE false END AS is_moving
+      FROM vehicle_live_state vls
+      JOIN vehicles v ON v.id = vls.vehicle_id
+      JOIN routes r ON r.id = vls.route_id
+      WHERE vls.route_id = $1
+        AND v.is_active = true
+    `;
+
+    if (!includeStale) {
+      query += `
+        AND vls.last_seen_at >= CURRENT_TIMESTAMP - ($2 * INTERVAL '1 second')
+      `;
+    }
+
+    query += `
+      ORDER BY vls.last_seen_at DESC, v.vehicle_number ASC
+    `;
+
+    const result = await pool.query(query, [routeIdInt, staleAfterSeconds]);
+    return result.rows;
+  }
+
+  async getVehicleLiveState(vehicleId, staleAfterSeconds = DEFAULT_STALE_AFTER_SECONDS) {
+    await ensureRealtimeSchema();
+
+    const vehicleIdInt = Number.parseInt(vehicleId, 10);
+
+    if (!Number.isInteger(vehicleIdInt) || vehicleIdInt <= 0) {
+      throw createHttpError(400, 'vehicleId must be a positive integer');
+    }
+
+    const result = await pool.query(
+      `SELECT
+        vls.vehicle_id,
+        v.vehicle_number,
+        v.registration,
+        v.type AS vehicle_type,
+        v.capacity,
+        vls.route_id,
+        r.route_number,
+        r.name AS route_name,
+        r.color AS route_color,
+        ST_Y(vls.location::geometry) AS latitude,
+        ST_X(vls.location::geometry) AS longitude,
+        vls.speed_kmh,
+        vls.heading,
+        vls.accuracy_meters,
+        vls.source_type,
+        vls.source_ref,
+        vls.last_seen_at,
+        vls.updated_at,
+        CASE
+          WHEN vls.last_seen_at >= CURRENT_TIMESTAMP - ($2 * INTERVAL '1 second') THEN true
+          ELSE false
+        END AS is_online,
+        GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - vls.last_seen_at))))::int AS seconds_since_update,
+        CASE WHEN COALESCE(vls.speed_kmh, 0) >= 1 THEN true ELSE false END AS is_moving
+      FROM vehicle_live_state vls
+      JOIN vehicles v ON v.id = vls.vehicle_id
+      JOIN routes r ON r.id = vls.route_id
+      WHERE vls.vehicle_id = $1
+      LIMIT 1`,
+      [vehicleIdInt, staleAfterSeconds]
+    );
+
+    if (result.rows.length === 0) {
+      throw createHttpError(404, 'Live state not found for this vehicle');
+    }
+
+    return result.rows[0];
+  }
+
+  async getRoute(routeId) {
+    if (routeId === undefined || routeId === null || routeId === '') {
+      return null;
+    }
+
+    const routeIdInt = Number.parseInt(routeId, 10);
+
+    if (!Number.isInteger(routeIdInt) || routeIdInt <= 0) {
+      throw createHttpError(400, 'routeId must be a positive integer');
+    }
+
+    const result = await pool.query(
+      'SELECT * FROM routes WHERE id = $1 AND is_active = true',
+      [routeIdInt]
+    );
+
+    if (result.rows.length === 0) {
+      throw createHttpError(404, 'Route not found');
+    }
+
+    return result.rows[0];
+  }
+
+  async resolveVehicle({ vehicleId, vehicleNumber, vehicleType, route }) {
+    if (vehicleId !== undefined && vehicleId !== null && vehicleId !== '') {
+      const vehicleIdInt = Number.parseInt(vehicleId, 10);
+
+      if (!Number.isInteger(vehicleIdInt) || vehicleIdInt <= 0) {
+        throw createHttpError(400, 'vehicleId must be a positive integer');
+      }
+
+      const result = await pool.query(
+        'SELECT * FROM vehicles WHERE id = $1',
+        [vehicleIdInt]
+      );
+
+      if (result.rows.length === 0) {
+        throw createHttpError(404, 'Vehicle not found');
+      }
+
+      if (!result.rows[0].is_active) {
+        throw createHttpError(400, 'Vehicle is inactive');
+      }
+
+      return result.rows[0];
+    }
+
+    const cleanedVehicleNumber = typeof vehicleNumber === 'string'
+      ? vehicleNumber.trim()
+      : '';
+
+    if (!cleanedVehicleNumber) {
+      throw createHttpError(400, 'vehicleNumber must be a non-empty string');
+    }
+
+    const existing = await pool.query(
+      'SELECT * FROM vehicles WHERE vehicle_number = $1 LIMIT 1',
+      [cleanedVehicleNumber]
+    );
+
+    if (existing.rows.length > 0) {
+      if (!existing.rows[0].is_active) {
+        throw createHttpError(400, 'Vehicle is inactive');
+      }
+
+      return existing.rows[0];
+    }
+
+    if (!route) {
+      throw createHttpError(400, 'routeId is required to auto-create a vehicle by vehicleNumber');
+    }
+
+    const resolvedType = vehicleType || route.type || 'bus';
+
+    const created = await pool.query(
+      `INSERT INTO vehicles (vehicle_number, type, route_id, is_active)
+       VALUES ($1, $2, $3, true)
+       RETURNING *`,
+      [cleanedVehicleNumber, resolvedType, route.id]
+    );
+
+    logger.info(`Auto-created vehicle ${cleanedVehicleNumber} for route ${route.id}`);
+
+    return created.rows[0];
+  }
+}
+
+module.exports = new TelemetryService();