Hugging Face's logo

Spaces:
vish85521
/
enterprisecargo
Paused

App Files Community
enterprisecargo
File size: 2,955 Bytes
eeb3436
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
const { withTransaction } = require("../db/transaction");

function createTelemetryIngestService(deps) {
  const {
    pool,
    logger,
    runtimeState,
    telemetryRepository,
    assetRepository,
    alertEngineService,
    telemetryValidator,
  } = deps;

  async function handleIncomingTelemetry(topicInfo, payload) {
    const validation = telemetryValidator.validateAndNormalizeTelemetry(topicInfo, payload);

    if (!validation.valid) {
      runtimeState.markMqttMessageRejected(`invalid_payload:${validation.errors.join("|")}`);
      logger.warn("Telemetry payload rejected", {
        tenantCode: topicInfo.tenantCode,
        truckCode: topicInfo.truckCode,
        containerCode: topicInfo.containerCode,
        errors: validation.errors,
      });
      return;
    }

    const context = await assetRepository.resolveAssetContextByCodes(pool, {
      tenantCode: topicInfo.tenantCode,
      truckCode: topicInfo.truckCode,
      containerCode: topicInfo.containerCode,
    });

    if (!context) {
      runtimeState.markMqttMessageRejected("unknown_asset_reference");
      logger.warn("Telemetry rejected due to unknown tenant/truck/container mapping", {
        tenantCode: topicInfo.tenantCode,
        truckCode: topicInfo.truckCode,
        containerCode: topicInfo.containerCode,
      });
      return;
    }

    const receivedAt = new Date().toISOString();
    const normalized = validation.normalized;

    const telemetry = {
      tenantId: context.tenant_id,
      fleetId: context.fleet_id,
      truckId: context.truck_id,
      containerId: context.container_id,
      tripId: context.trip_id,
      gatewayDeviceId: null,
      sensorDeviceId: null,
      mqttTopic: topicInfo.topic,
      seq: normalized.seq,
      sourceTs: normalized.sourceTs,
      receivedAt,
      gpsLat: normalized.gpsLat,
      gpsLon: normalized.gpsLon,
      speedKph: normalized.speedKph,
      temperatureC: normalized.temperatureC,
      humidityPct: normalized.humidityPct,
      pressureHpa: normalized.pressureHpa,
      tiltDeg: normalized.tiltDeg,
      shock: normalized.shock,
      gasRaw: normalized.gasRaw,
      gasAlert: normalized.gasAlert,
      sdOk: normalized.sdOk,
      gpsFix: normalized.gpsFix,
      uplink: normalized.uplink,
      rawPayload: normalized.rawPayload,
    };

    await withTransaction(pool, async (client) => {
      await telemetryRepository.insertTelemetryHistory(client, telemetry);
      await telemetryRepository.upsertTelemetryLatest(client, telemetry);
      await alertEngineService.evaluateTelemetryInTransaction(client, {
        tenantId: context.tenant_id,
        fleetId: context.fleet_id,
        truckId: context.truck_id,
        containerId: context.container_id,
        tripId: context.trip_id,
      }, telemetry);
    });

    runtimeState.markMqttMessageAccepted();
  }

  return {
    handleIncomingTelemetry,
  };
}

module.exports = {
  createTelemetryIngestService,
};