Sensor Service

Implementation reference for the sensor-service. For its architectural role, see Sensor Service Architecture; for the domain model, see Telemetry Ingestion.

A Node.js / Express service in strict TypeScript. It ingests sensor readings from constrained hardware, persists them, publishes them for live distribution, and manages per-building thresholds. Its design answers two pressures: never make a sensor wait, and never edit the core to add a sensor type.

Source Layout

graph TD
    subgraph Controllers
        IC["ingestionController.ts"]
        RC["readController.ts"]
        TC["thresholdController.ts"]
        WC["writerController.ts"]
        AC["actionController.ts"]
    end
    subgraph "Kernel — Microkernel"
        K["sensorKernel.ts\nregistry: type to module"]
    end
    subgraph "Strategy Modules"
        BASE["BaseSensorModule.ts\nTemplate Method"]
        TM["TemperatureModule.ts"]
        PM["PeopleCountModule.ts"]
        AM["AirQualityModule.ts"]
    end
    subgraph Services
        MS["*ModuleService.ts\nper-metric persistence"]
        AS["ActionService.ts"]
    end
    IC --> K
    RC --> K
    WC --> K
    TC --> MS
    AC --> AS
    K --> BASE
    BASE --> TM
    BASE --> PM
    BASE --> AM
    TM --> MS
    PM --> MS
    AM --> MS
    AC --> AJ["actions.json\ndata-driven routing"]

Configuration

The temperature module additionally calls the notification-service push endpoint when a reading breaches its bound.

The Microkernel

SensorKernel is a pure dispatch registry. Modules are registered once at startup; request handling is then a lookup with no branching on sensor type in the core.

class SensorKernel {
  register(module: ISensorModule): this  // throws on duplicate type
  resolve(type: string): ISensorModule | undefined
  getRegisteredTypes(): readonly string[]
}

Duplicate registration throws synchronously — it is a configuration bug and should crash at boot, not silently overwrite.

Ingestion: Fast Path / Slow Path

The critical constraint is that hardware clients are released immediately. The controller validates synchronously, flushes the response, and only then processes.

sequenceDiagram
    autonumber
    participant HW as Sensor
    participant IC as IngestionController
    participant K as SensorKernel
    participant M as Module
    participant DB as MongoDB
    participant R as Redis
    HW->>IC: POST /ingest { type, ...data }
    IC->>K: resolve(type)
    alt unknown type
        IC-->>HW: 404
    else known
        IC->>M: validate(data)   (pure, synchronous)
        alt invalid
            IC-->>HW: 422 + errors
        else valid
            IC-->>HW: 202 Accepted
            Note over IC: setImmediate — response is flushed first
            IC->>M: process(data)
            M->>DB: persist
            M->>R: PUBLISH telemetry:raw
        end
    end

• A missing/blank type is 400; an unregistered type is 404; a failed validation is 422 with structured details.
• setImmediate defers module.process() until after the 202 is on the wire, so persistence and broker latency never affect the sensor’s round-trip.
• Background failures are caught and logged; they never reach the (already closed) connection.

The Module Contract

ISensorModule and ValidationResult

validate() is pure and synchronous (no I/O) and returns an immutable ValidationResult rather than throwing, so the controller can return machine-readable errors before processing. process() is impure and runs only after a successful validate().

BaseSensorModule — Template Method

Concrete modules extend BaseSensorModule<T> rather than implementing ISensorModule directly. The base seals the ingestion pipeline:

async process(payload: unknown): Promise<void> {
  await this.persist(payload);                         // child-specific
  this.publishTelemetry(this.buildTelemetryEvent(payload)); // base-owned
}

• The order is fixed — persist, then publish — so a module can never publish a telemetry:raw event before the reading is queryable from the read path, nor forget the publish.
• publishTelemetry writes to telemetry:raw fire-and-forget (failures logged), keeping ingestion latency independent of Redis.
• getLatest is shared (most-recent document per room); the threshold methods default to no-ops and are overridden only by modules that support thresholds.

A child therefore implements only what is unique: type, model, validate, persist, buildTelemetryEvent, getAllLatest, getDashboardData.

Lifecycle methods: create and apply

Two contract methods extend a module beyond pure ingestion:

• create(buildingId, roomId, sensorType, sensorId) — registers a physical sensor. The base class supplies the default: it inserts a Sensors document, so no module needs to reimplement it.
• apply() — the hook a module overrides to actuate a received command against its hardware. It defaults to a no-op in the base class and is currently still a stub in TemperatureModule; the action path (below) presently dispatches through actions.json rather than through apply().

Data Models

Signals

Each metric is its own MongoDB time-series collection. The engine groups readings from the same building into compressed buckets internally; Mongoose reads and writes plain documents — the bucketing is transparent.

All three schemas share the following time-series options:

• timeField: "createdAt" — the Date axis; stamped automatically on insert via default: Date.now.
• metaField: "building" — the series identifier; readings from the same building are co-located in the same bucket.
• granularity: "seconds" — bucket windows sized to the sensor polling rate.
• expireAfterSeconds: 7 776 000 (90 days) — retention enforced by the collection, no separate TTL index needed.

Two secondary indexes support dashboard queries that filter on the measurement timestamp (Unix epoch from the hardware) rather than the createdAt time axis:

• { building: 1, timestamp: -1 } — building-wide range scans.
• { building: 1, roomId: 1, timestamp: -1 } — per-room point and range queries.

BuildingThreshold

One document per building (buildingId unique). It holds building-level default bounds and an array of per-room overrides, each with optional temperature { minTemp, maxTemp }, peopleCount { maxPeople }, and airQuality { maxAqi, maxCo2 }.

Sensor

The registry of physical sensors, written by the module create() path: { buildingId, roomId, sensorId, sensorType }. A plain (non-time-series) collection. sensorId is unique.

Action

The shape of a dispatched command (actionSignal.ts): { buildingId, roomId, sensorId, sensorType, timestamp, actionName, actionArguments }, available for persisting/auditing actions sent to a sensor.

HTTP API

Thresholds

Thresholds are static, human-defined configuration. They live in a single shared BuildingThreshold document with nested slices per metric. When a domain service updates its slice it uses scoped dot-notation so it cannot corrupt a neighbour’s configuration:

findOneAndUpdate({ buildingId }, { $set: { "temperature.maxTemp": 25 } })

A room-level bound, when present, overrides the building default for that room during evaluation.

Read-Path Caching

getAllLatest(buildingId) — which drives the initial dashboard load — runs a MongoDB aggregation. To collapse duplicate work when many users open the same building, results are cached in Redis under sensor:latest:{type}:{buildingId} with a 10-second TTL. The live telemetry path bypasses the cache entirely: every write publishes to telemetry:raw, so the frontend receives real-time updates via the socket without waiting for cache expiry.

Sensor Registry & Actions

Beyond ingesting readings, the service now tracks which sensors exist and can dispatch commands back to them. These are the write and command counterparts to the read/ingest paths.

Registration

POST /sensor registers a physical sensor. writerController resolves the module by sensorType and calls module.create(), which the base class implements by inserting a Sensor document. As with ingestion, the handler validates the type synchronously and returns 202 Accepted without waiting on the write. Registered sensors are then listable via the GET /sensors/... endpoints and are what the frontend’s room-sensor panel renders.

Actions: a data-driven command path

POST /executeAction sends a command — e.g. increase, decrease — to the actuator API that fronts a sensor. Unlike ingestion, this path is synchronous: it awaits the downstream call and reports the real outcome (200 on success, 502 when the endpoint errors or is unreachable, 404 when no route matches). The frontend supplies { sensorId, actionName, actionArguments }.

Routing is entirely driven by actions.json, so onboarding a new sensor or action is a config edit, never a code change. The table is keyed actionName → sensorId → endpoint:

{
  "increase": {
    "123": {
      "url": "http://gateway/simulator/tp/control/receive",
      "method": "POST",
      "arguments": { "0": "value" }
    }
  }
}

• url / method — the downstream API to call for that (action, sensor) pair.
• arguments maps each positional index of the frontend’s actionArguments array to the field name the target API expects. { "0": "value" } with actionArguments: ["1"] produces the request body { "value": "1" }. This lets different sensors name the same positional argument differently.

The file is resolved relative to the compiled module (the package root, independent of the working directory) and re-read on every request, so edits take effect without a restart. ActionService.getSensorId(roomId, buildingId) is available to resolve a sensor id from its location when a caller does not pass one directly.

sequenceDiagram
    autonumber
    participant UI as Frontend
    participant AC as actionController
    participant J as actions.json
    participant API as Actuator API
    UI->>AC: POST /executeAction { sensorId, actionName, actionArguments }
    AC->>J: lookup actionName → sensorId
    alt no route
        AC-->>UI: 404
    else route found
        AC->>AC: map positional args to named fields
        AC->>API: method url { ...mappedArgs }
        alt ok
            API-->>AC: 2xx
            AC-->>UI: 200 accepted
        else error / unreachable
            AC-->>UI: 502
        end
    end

Extension Guide

Adding a sensor type requires no change to the controller or kernel:

1. Create a module extending BaseSensorModule<T> and implement the unique members (type, model, validate, persist, buildTelemetryEvent, read methods).
2. Register it during boot:

const kernel = new SensorKernel()
  .register(new TemperatureModule())
  .register(new PeopleCountModule())
  .register(new AirQualityModule())
  .register(new LightLevelModule()); // new type

The base class supplies the telemetry:raw publish and the shared getLatest, so a new module only writes what is genuinely specific to it.