Software Engineering for Industrial Automation: IoT Dashboard with PHP

Why Industrial Automation Needs Software Engineers

Industrial automation is not just PLC programming and SCADA screens. Behind every factory floor sensor, vibration monitor, and temperature probe sits a data pipeline that must be reliable, queryable, and secure. Software engineers build that pipeline.

Consider the difference between a generic web dashboard and an industrial IoT monitoring system:

These constraints mean that generic web development patterns fail under industrial workloads. You cannot SELECT * FROM sensor_readings when the table has 500 million rows. You cannot poll for alerts every 60 seconds when a motor over-temperature event needs sub-second reaction. And you cannot store sensor data in MySQL and expect reasonable query performance past the first few million rows.

Mengapa Otomasi Industri Membutuhkan Insinyur Perangkat Lunak

Otomasi industri bukan hanya pemrograman PLC dan layar SCADA. Di balik setiap sensor lantai pabrik, monitor getaran, dan probe suhu terdapat pipeline data yang harus andal, dapat di-query, dan aman. Insinyur perangkat lunak membangun pipeline itu.

Pertimbangkan perbedaan antara dasbor web generik dan sistem pemantauan IoT industri:

Batasan ini berarti bahwa pola pengembangan web generik gagal di bawah beban kerja industri. Anda tidak dapat SELECT * FROM sensor_readings ketika tabel memiliki 500 juta baris. Anda tidak dapat melakukan polling peringatan setiap 60 detik ketika kejadian suhu berlebih motor membutuhkan reaksi sub-detik. Dan Anda tidak dapat menyimpan data sensor di MySQL dan mengharapkan performa query yang wajar setelah beberapa juta baris pertama.

System Architecture

A production-grade industrial IoT monitoring system follows a layered architecture. Each layer has a specific responsibility and failure mode.

Data Flow (Write Path)

Sensor → Modbus/OPC-UA → Edge Gateway → MQTT Publish
                                            │
                  ┌─────────────────────────┘
                  ▼
         MQTT Broker (topic: factory/line1/temperature)
                  │
                  ▼
         PHP Subscriber (validate → transform → enrich)
                  │
                  ▼
         InfluxDB Write (measurement: sensor_readings, tags: {sensor_id, unit}, fields: {value})

Data Flow (Read Path)

Dashboard Client → GET /api/dashboard/sensors/{sensor_id}/readings?range=1h
                                  │
                                  ▼
                       DashboardService (query builder)
                                  │
                                  ▼
                       InfluxDB Flux Query (aggregate, filter, downsample)
                                  │
                                  ▼
                       JSON Response → Chart (Chart.js / ECharts)

Every arrow in these diagrams represents an integration point that can fail. The PHP ingestion service must handle broker disconnects, malformed payloads, and InfluxDB write timeouts without losing data or crashing.

Arsitektur Sistem

Sistem pemantauan IoT industri tingkat produksi mengikuti arsitektur berlapis. Setiap lapisan memiliki tanggung jawab dan mode kegagalan spesifik.

Alur Data (Jalur Tulis)

Sensor → Modbus/OPC-UA → Gateway Edge → Publikasi MQTT
                                            │
                  ┌─────────────────────────┘
                  ▼
         Broker MQTT (topik: pabrik/line1/suhu)
                  │
                  ▼
         Subscriber PHP (validasi → transformasi → perkaya)
                  │
                  ▼
         InfluxDB Write (measurement: sensor_readings, tags: {sensor_id, unit}, fields: {value})

Alur Data (Jalur Baca)

Klien Dasbor → GET /api/dashboard/sensors/{sensor_id}/readings?range=1h
                                  │
                                  ▼
                       DashboardService (pembangun query)
                                  │
                                  ▼
                       InfluxDB Flux Query (agregasi, filter, downsample)
                                  │
                                  ▼
                       Respons JSON → Grafik (Chart.js / ECharts)

Setiap panah dalam diagram ini mewakili titik integrasi yang dapat gagal. Layanan ingesti PHP harus menangani pemutusan broker, payload yang salah format, dan timeout penulisan InfluxDB tanpa kehilangan data atau crash.

Setting Up the Stack

Before writing code, you need a running MQTT broker and InfluxDB instance. This tutorial assumes Docker for reproducibility.

1. Start Mosquitto MQTT Broker

docker run -d --name mosquitto \
  -p 1883:1883 \
  -p 9001:9001 \
  eclipse-mosquitto:2

2. Start InfluxDB v2

docker run -d --name influxdb \
  -p 8086:8086 \
  -e DOCKER_INFLUXDB_INIT_MODE=setup \
  -e DOCKER_INFLUXDB_INIT_USERNAME=admin \
  -e DOCKER_INFLUXDB_INIT_PASSWORD=admin1234 \
  -e DOCKER_INFLUXDB_INIT_ORG=se-polinema \
  -e DOCKER_INFLUXDB_INIT_BUCKET=sensor_data \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-super-secret-token \
  influxdb:2

3. Install PHP Dependencies

composer require php-mqtt/client guzzlehttp/guzzle

• php-mqtt/client — MQTT subscriber for ingesting sensor data
• guzzlehttp/guzzle — HTTP client for InfluxDB REST API

4. Create InfluxDB Bucket via API

curl -X POST "http://localhost:8086/api/v2/buckets" \
  -H "Authorization: Token my-super-secret-token" \
  -H "Content-Type: application/json" \
  -d '{"orgID": "<your-org-id>", "name": "sensor_data", "retentionRules": [{"type": "expire", "everySeconds": 7776000}]}'

This creates a bucket with a 90-day retention policy — raw sensor data older than 90 days is automatically dropped. Downsampled aggregates should live in a separate bucket with longer retention.

Menyiapkan Stack

Sebelum menulis kode, Anda memerlukan broker MQTT dan instance InfluxDB yang berjalan. Tutorial ini mengasumsikan Docker untuk reprodusibilitas.

1. Jalankan Broker MQTT Mosquitto

docker run -d --name mosquitto \
  -p 1883:1883 \
  -p 9001:9001 \
  eclipse-mosquitto:2

2. Jalankan InfluxDB v2

docker run -d --name influxdb \
  -p 8086:8086 \
  -e DOCKER_INFLUXDB_INIT_MODE=setup \
  -e DOCKER_INFLUXDB_INIT_USERNAME=admin \
  -e DOCKER_INFLUXDB_INIT_PASSWORD=admin1234 \
  -e DOCKER_INFLUXDB_INIT_ORG=se-polinema \
  -e DOCKER_INFLUXDB_INIT_BUCKET=sensor_data \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-super-secret-token \
  influxdb:2

3. Instal Dependensi PHP

composer require php-mqtt/client guzzlehttp/guzzle

• php-mqtt/client: Subscriber MQTT untuk ingesti data sensor
• guzzlehttp/guzzle: Klien HTTP untuk REST API InfluxDB

4. Buat Bucket InfluxDB via API

curl -X POST "http://localhost:8086/api/v2/buckets" \
  -H "Authorization: Token my-super-secret-token" \
  -H "Content-Type: application/json" \
  -d '{"orgID": "<id-org-anda>", "name": "sensor_data", "retentionRules": [{"type": "expire", "everySeconds": 7776000}]}'

Ini membuat bucket dengan kebijakan retensi 90 hari: data sensor mentah yang lebih tua dari 90 hari secara otomatis dihapus. Agregat yang di-downsample harus disimpan di bucket terpisah dengan retensi lebih panjang.

Project Structure

A modular monolith keeps bounded contexts separate — MQTT ingestion, time-series storage, and the dashboard API are three distinct modules that share nothing except the InfluxDB connection configuration.

src/
├── IoT/
│   ├── Domain/
│   │   ├── SensorReading.php        # Value object: sensor_id, metric, value, unit, timestamp
│   │   ├── AlertRule.php            # Value object: metric, operator, threshold, severity
│   │   ├── AlertEvent.php           # DTO: triggered alert with context
│   │   └── DashboardQuery.php       # DTO: aggregated query parameters
│   ├── Application/
│   │   ├── MQTTIngestionService.php       # Subscribes to MQTT, validates, writes to InfluxDB
│   │   ├── InfluxDBRepository.php         # Reads/writes to InfluxDB via HTTP API
│   │   ├── DashboardService.php           # Query builder for dashboard API
│   │   └── AlertService.php               # Threshold evaluation and alert triggering
│   └── Infrastructure/
│       ├── InfluxDBClient.php             # Guzzle wrapper for InfluxDB HTTP API
│       └── MQTTSubscriber.php             # php-mqtt/client wrapper with reconnect logic
└── config/
    └── influxdb.php                 # InfluxDB connection config (org, bucket, token, url)

This structure follows the ports and adapters pattern. The domain layer knows nothing about MQTT or HTTP — it only expresses industrial IoT concepts. The infrastructure layer handles the gritty details of network I/O.

Struktur Proyek

Monolit modular menjaga bounded context tetap terpisah: ingesti MQTT, penyimpanan time-series, dan API dasbor adalah tiga modul berbeda yang tidak berbagi apa pun kecuali konfigurasi koneksi InfluxDB.

src/
├── IoT/
│   ├── Domain/
│   │   ├── SensorReading.php        # Value object: sensor_id, metric, value, unit, timestamp
│   │   ├── AlertRule.php            # Value object: metric, operator, threshold, severity
│   │   ├── AlertEvent.php           # DTO: peringatan terpicu dengan konteks
│   │   └── DashboardQuery.php       # DTO: parameter query teragregasi
│   ├── Application/
│   │   ├── MQTTIngestionService.php       # Subscribe ke MQTT, validasi, tulis ke InfluxDB
│   │   ├── InfluxDBRepository.php         # Baca/tulis ke InfluxDB via HTTP API
│   │   ├── DashboardService.php           # Pembangun query untuk API dasbor
│   │   └── AlertService.php               # Evaluasi ambang dan pemicuan peringatan
│   └── Infrastructure/
│       ├── InfluxDBClient.php             # Wrapper Guzzle untuk HTTP API InfluxDB
│       └── MQTTSubscriber.php             # Wrapper php-mqtt/client dengan logika reconnect
└── config/
    └── influxdb.php                 # Konfigurasi koneksi InfluxDB (org, bucket, token, url)

Struktur ini mengikuti pola ports and adapters. Lapisan domain tidak tahu apa pun tentang MQTT atau HTTP: ia hanya mengekspresikan konsep IoT industri. Lapisan infrastruktur menangani detail rumit dari I/O jaringan.

Domain Layer: Value Objects and DTOs

SensorReading — The Core Domain Object

A sensor reading is an immutable measurement captured at a point in time. It is the atomic unit of industrial data.

<?php

declare(strict_types=1);

namespace App\IoT\Domain;

use DateTimeImmutable;

final readonly class SensorReading
{
    public function __construct(
        public string $sensorId,
        public string $metric,
        public float $value,
        public string $unit,
        public DateTimeImmutable $timestamp,
    ) {}

    public static function fromMQTTPayload(array $payload): self
    {
        return new self(
            sensorId: (string) $payload['sensor_id'],
            metric: (string) $payload['metric'],
            value: (float) $payload['value'],
            unit: (string) $payload['unit'],
            timestamp: isset($payload['timestamp'])
                ? new DateTimeImmutable($payload['timestamp'])
                : new DateTimeImmutable(),
        );
    }

    public function toInfluxDBLineProtocol(): string
    {
        $ts = $this->timestamp->getTimestamp() * 1_000_000_000;

        return sprintf(
            'sensor_readings,sensor_id=%s,metric=%s,unit=%s value=%.4f %d',
            $this->escapeTag($this->sensorId),
            $this->escapeTag($this->metric),
            $this->escapeTag($this->unit),
            $this->value,
            $ts,
        );
    }

    private function escapeTag(string $value): string
    {
        return str_replace([' ', ',', '='], ['\\ ', '\\,', '\\='], $value);
    }
}

Design decisions:

• fromMQTTPayload is the single factory method — all validation happens here, not in the constructor.
• toInfluxDBLineProtocol produces InfluxDB line protocol format directly. This avoids intermediate serialization and lets the infrastructure layer batch-write efficiently.
• escapeTag prevents injection in InfluxDB tag values — spaces, commas, and equals signs must be escaped per the line protocol spec.

AlertRule — Defining Monitoring Thresholds

<?php

declare(strict_types=1);

namespace App\IoT\Domain;

enum AlertOperator: string
{
    case GREATER_THAN = '>';
    case LESS_THAN = '<';
    case GREATER_THAN_OR_EQUAL = '>=';
    case LESS_THAN_OR_EQUAL = '<=';
    case EQUALS = '=';
}

enum AlertSeverity: string
{
    case INFO = 'info';
    case WARNING = 'warning';
    case CRITICAL = 'critical';
}

final readonly class AlertRule
{
    public function __construct(
        public string $id,
        public string $sensorId,
        public string $metric,
        public AlertOperator $operator,
        public float $threshold,
        public AlertSeverity $severity,
        public string $message,
    ) {}

    public function evaluate(SensorReading $reading): ?AlertEvent
    {
        if ($reading->sensorId !== $this->sensorId || $reading->metric !== $this->metric) {
            return null;
        }

        $triggered = match ($this->operator) {
            AlertOperator::GREATER_THAN => $reading->value > $this->threshold,
            AlertOperator::LESS_THAN => $reading->value < $this->threshold,
            AlertOperator::GREATER_THAN_OR_EQUAL => $reading->value >= $this->threshold,
            AlertOperator::LESS_THAN_OR_EQUAL => $reading->value <= $this->threshold,
            AlertOperator::EQUALS => abs($reading->value - $this->threshold) < PHP_FLOAT_EPSILON,
        };

        if (!$triggered) {
            return null;
        }

        return new AlertEvent(
            ruleId: $this->id,
            sensorId: $reading->sensorId,
            metric: $reading->metric,
            value: $reading->value,
            threshold: $this->threshold,
            severity: $this->severity,
            message: $this->message,
            timestamp: new \DateTimeImmutable(),
        );
    }
}

AlertEvent — The Result of a Triggered Rule

<?php

declare(strict_types=1);

namespace App\IoT\Domain;

use DateTimeImmutable;

final readonly class AlertEvent
{
    public function __construct(
        public string $ruleId,
        public string $sensorId,
        public string $metric,
        public float $value,
        public float $threshold,
        public AlertSeverity $severity,
        public string $message,
        public DateTimeImmutable $timestamp,
    ) {}
}

DashboardQuery — Parameter Object for API Queries

<?php

declare(strict_types=1);

namespace App\IoT\Domain;

final readonly class DashboardQuery
{
    public function __construct(
        public string $sensorId,
        public ?string $metric = null,
        public string $range = '1h',
        public string $aggregation = 'mean',
        public string $window = '1m',
        public int $limit = 100,
    ) {}

    public static function fromArray(array $params): self
    {
        return new self(
            sensorId: (string) $params['sensor_id'],
            metric: $params['metric'] ?? null,
            range: (string) ($params['range'] ?? '1h'),
            aggregation: (string) ($params['aggregation'] ?? 'mean'),
            window: (string) ($params['window'] ?? '1m'),
            limit: (int) ($params['limit'] ?? 100),
        );
    }
}

Lapisan Domain: Value Object dan DTO

SensorReading: Objek Domain Inti

Pembacaan sensor adalah pengukuran immutable yang ditangkap pada suatu titik waktu. Ini adalah unit atomik dari data industri.

<?php

declare(strict_types=1);

namespace App\IoT\Domain;

use DateTimeImmutable;

final readonly class SensorReading
{
    public function __construct(
        public string $sensorId,
        public string $metric,
        public float $value,
        public string $unit,
        public DateTimeImmutable $timestamp,
    ) {}

    public static function fromMQTTPayload(array $payload): self
    {
        return new self(
            sensorId: (string) $payload['sensor_id'],
            metric: (string) $payload['metric'],
            value: (float) $payload['value'],
            unit: (string) $payload['unit'],
            timestamp: isset($payload['timestamp'])
                ? new DateTimeImmutable($payload['timestamp'])
                : new DateTimeImmutable(),
        );
    }

    public function toInfluxDBLineProtocol(): string
    {
        $ts = $this->timestamp->getTimestamp() * 1_000_000_000;

        return sprintf(
            'sensor_readings,sensor_id=%s,metric=%s,unit=%s value=%.4f %d',
            $this->escapeTag($this->sensorId),
            $this->escapeTag($this->metric),
            $this->escapeTag($this->unit),
            $this->value,
            $ts,
        );
    }

    private function escapeTag(string $value): string
    {
        return str_replace([' ', ',', '='], ['\\ ', '\\,', '\\='], $value);
    }
}

Keputusan desain:

• fromMQTTPayload adalah satu-satunya factory method: semua validasi terjadi di sini, bukan di constructor.
• toInfluxDBLineProtocol menghasilkan format line protocol InfluxDB secara langsung. Ini menghindari serialisasi perantara dan memungkinkan lapisan infrastruktur menulis batch secara efisien.
• escapeTag mencegah injeksi pada nilai tag InfluxDB: spasi, koma, dan tanda sama dengan harus di-escape sesuai spesifikasi line protocol.

AlertRule: Mendefinisikan Ambang Pemantauan

<?php

declare(strict_types=1);

namespace App\IoT\Domain;

enum AlertOperator: string
{
    case GREATER_THAN = '>';
    case LESS_THAN = '<';
    case GREATER_THAN_OR_EQUAL = '>=';
    case LESS_THAN_OR_EQUAL = '<=';
    case EQUALS = '=';
}

enum AlertSeverity: string
{
    case INFO = 'info';
    case WARNING = 'warning';
    case CRITICAL = 'critical';
}

final readonly class AlertRule
{
    public function __construct(
        public string $id,
        public string $sensorId,
        public string $metric,
        public AlertOperator $operator,
        public float $threshold,
        public AlertSeverity $severity,
        public string $message,
    ) {}

    public function evaluate(SensorReading $reading): ?AlertEvent
    {
        if ($reading->sensorId !== $this->sensorId || $reading->metric !== $this->metric) {
            return null;
        }

        $triggered = match ($this->operator) {
            AlertOperator::GREATER_THAN => $reading->value > $this->threshold,
            AlertOperator::LESS_THAN => $reading->value < $this->threshold,
            AlertOperator::GREATER_THAN_OR_EQUAL => $reading->value >= $this->threshold,
            AlertOperator::LESS_THAN_OR_EQUAL => $reading->value <= $this->threshold,
            AlertOperator::EQUALS => abs($reading->value - $this->threshold) < PHP_FLOAT_EPSILON,
        };

        if (!$triggered) {
            return null;
        }

        return new AlertEvent(
            ruleId: $this->id,
            sensorId: $reading->sensorId,
            metric: $reading->metric,
            value: $reading->value,
            threshold: $this->threshold,
            severity: $this->severity,
            message: $this->message,
            timestamp: new \DateTimeImmutable(),
        );
    }
}

AlertEvent: Hasil dari Aturan yang Terpicu

<?php

declare(strict_types=1);

namespace App\IoT\Domain;

use DateTimeImmutable;

final readonly class AlertEvent
{
    public function __construct(
        public string $ruleId,
        public string $sensorId,
        public string $metric,
        public float $value,
        public float $threshold,
        public AlertSeverity $severity,
        public string $message,
        public DateTimeImmutable $timestamp,
    ) {}
}

DashboardQuery: Parameter Object untuk Query API

<?php

declare(strict_types=1);

namespace App\IoT\Domain;

final readonly class DashboardQuery
{
    public function __construct(
        public string $sensorId,
        public ?string $metric = null,
        public string $range = '1h',
        public string $aggregation = 'mean',
        public string $window = '1m',
        public int $limit = 100,
    ) {}

    public static function fromArray(array $params): self
    {
        return new self(
            sensorId: (string) $params['sensor_id'],
            metric: $params['metric'] ?? null,
            range: (string) ($params['range'] ?? '1h'),
            aggregation: (string) ($params['aggregation'] ?? 'mean'),
            window: (string) ($params['window'] ?? '1m'),
            limit: (int) ($params['limit'] ?? 100),
        );
    }
}

Infrastructure Layer: Talking to InfluxDB and MQTT

InfluxDB HTTP Client

InfluxDB v2 exposes a REST API for both reads (Flux queries) and writes (line protocol). This client wraps Guzzle with sensible defaults for time-series workloads.

<?php

declare(strict_types=1);

namespace App\IoT\Infrastructure;

use GuzzleHttp\Client;
use GuzzleHttp\Exception\GuzzleException;
use RuntimeException;

final class InfluxDBClient
{
    private Client $http;
    private string $org;

    public function __construct(
        private string $url,
        private string $token,
        string $org,
        private string $bucket,
    ) {
        $this->http = new Client([
            'base_uri' => $this->url,
            'timeout' => 10,
            'connect_timeout' => 3,
            'headers' => [
                'Authorization' => "Token {$this->token}",
                'Content-Type' => 'text/plain; charset=utf-8',
            ],
        ]);
        $this->org = $org;
    }

    /**
     * Write line protocol data to InfluxDB.
     * Accepts a newline-separated string of line protocol records.
     */
    public function write(string $lineProtocol): void
    {
        try {
            $this->http->post('/api/v2/write', [
                'query' => [
                    'org' => $this->org,
                    'bucket' => $this->bucket,
                    'precision' => 'ns',
                ],
                'body' => $lineProtocol,
            ]);
        } catch (GuzzleException $e) {
            throw new RuntimeException(
                'InfluxDB write failed: ' . $e->getMessage(),
                previous: $e,
            );
        }
    }

    /**
     * Execute a Flux query and return parsed results.
     */
    public function query(string $fluxQuery): array
    {
        try {
            $response = $this->http->post('/api/v2/query', [
                'headers' => [
                    'Content-Type' => 'application/vnd.flux',
                    'Accept' => 'application/csv',
                ],
                'query' => ['org' => $this->org],
                'body' => $fluxQuery,
            ]);

            return $this->parseCSVResponse((string) $response->getBody());
        } catch (GuzzleException $e) {
            throw new RuntimeException(
                'InfluxDB query failed: ' . $e->getMessage(),
                previous: $e,
            );
        }
    }

    /**
     * Check connectivity and bucket existence.
     */
    public function healthCheck(): bool
    {
        try {
            $response = $this->http->get('/health');
            $body = json_decode((string) $response->getBody(), true);

            return ($body['status'] ?? '') === 'pass';
        } catch (GuzzleException) {
            return false;
        }
    }

    /**
     * Parse InfluxDB CSV-annotated response into associative array.
     */
    private function parseCSVResponse(string $csv): array
    {
        $lines = explode("\n", trim($csv));
        $results = [];

        if (count($lines) < 2) {
            return $results;
        }

        // First line: annotation headers (#group, #datatype, #default)
        // Skip annotation lines that start with #
        $dataLines = array_filter($lines, fn(string $line): bool => !str_starts_with($line, '#'));
        $dataLines = array_values($dataLines);

        if (empty($dataLines)) {
            return $results;
        }

        $headers = str_getcsv(array_shift($dataLines));

        foreach ($dataLines as $line) {
            $row = str_getcsv($line);
            if (count($row) < count($headers)) {
                continue;
            }
            $results[] = array_combine($headers, $row);
        }

        return $results;
    }
}

MQTT Subscriber with Reconnect Logic

<?php

declare(strict_types=1);

namespace App\IoT\Infrastructure;

use PhpMqtt\Client\MqttClient;
use PhpMqtt\Client\ConnectionSettings;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;

final class MQTTSubscriber
{
    private MqttClient $client;
    private int $reconnectDelay = 5;

    public function __construct(
        private string $brokerHost,
        private int $brokerPort,
        private string $clientId,
        private ?LoggerInterface $logger = null,
    ) {
        $this->logger = $logger ?? new NullLogger();
    }

    /**
     * Subscribe to a topic and process incoming messages.
     * Automatically reconnects on connection loss.
     *
     * @param callable(SensorReading): void $onMessage
     */
    public function subscribe(string $topic, callable $onMessage): void
    {
        $settings = (new ConnectionSettings())
            ->setKeepAliveInterval(60)
            ->setConnectTimeout(10)
            ->setReconnectAutomatically(true)
            ->setMaxReconnectAttempts(10)
            ->setDelayBetweenReconnectAttempts($this->reconnectDelay * 1000);

        $this->client = new MqttClient(
            $this->brokerHost,
            $this->brokerPort,
            $this->clientId,
        );

        $this->client->registerLoopEventHandler(function () {
            $this->logger?->info('MQTT subscriber loop running', [
                'client_id' => $this->clientId,
            ]);
        });

        $this->logger?->info('Connecting to MQTT broker', [
            'host' => $this->brokerHost,
            'port' => $this->brokerPort,
        ]);

        $this->client->connect($settings, true);

        $this->logger?->info('Subscribing to topic', ['topic' => $topic]);
        $this->client->subscribe($topic, function (string $topic, string $message) use ($onMessage): void {
            $payload = json_decode($message, true);

            if (!is_array($payload)) {
                $this->logger?->warning('Received invalid JSON payload', [
                    'topic' => $topic,
                    'message' => $message,
                ]);
                return;
            }

            try {
                $reading = \App\IoT\Domain\SensorReading::fromMQTTPayload($payload);
                $onMessage($reading);
            } catch (\Throwable $e) {
                $this->logger?->error('Failed to process sensor reading', [
                    'topic' => $topic,
                    'error' => $e->getMessage(),
                ]);
            }
        }, 0);

        $this->client->loop(true);
    }

    public function disconnect(): void
    {
        if (isset($this->client)) {
            $this->client->disconnect();
        }
    }
}

Note: The php-mqtt/client package provides a clean abstraction over the MQTT protocol. In production, you may want to run the subscriber as a long-lived PHP process using Swoole, RoadRunner, or a systemd service with supervisor for automatic restarts.

Lapisan Infrastruktur: Berbicara dengan InfluxDB dan MQTT

Klien HTTP InfluxDB

InfluxDB v2 mengekspos REST API untuk pembacaan (query Flux) dan penulisan (line protocol). Klien ini membungkus Guzzle dengan default yang masuk akal untuk beban kerja time-series.

<?php

declare(strict_types=1);

namespace App\IoT\Infrastructure;

use GuzzleHttp\Client;
use GuzzleHttp\Exception\GuzzleException;
use RuntimeException;

final class InfluxDBClient
{
    private Client $http;
    private string $org;

    public function __construct(
        private string $url,
        private string $token,
        string $org,
        private string $bucket,
    ) {
        $this->http = new Client([
            'base_uri' => $this->url,
            'timeout' => 10,
            'connect_timeout' => 3,
            'headers' => [
                'Authorization' => "Token {$this->token}",
                'Content-Type' => 'text/plain; charset=utf-8',
            ],
        ]);
        $this->org = $org;
    }

    /**
     * Tulis data line protocol ke InfluxDB.
     * Menerima string line protocol yang dipisahkan newline.
     */
    public function write(string $lineProtocol): void
    {
        try {
            $this->http->post('/api/v2/write', [
                'query' => [
                    'org' => $this->org,
                    'bucket' => $this->bucket,
                    'precision' => 'ns',
                ],
                'body' => $lineProtocol,
            ]);
        } catch (GuzzleException $e) {
            throw new RuntimeException(
                'Penulisan InfluxDB gagal: ' . $e->getMessage(),
                previous: $e,
            );
        }
    }

    /**
     * Jalankan query Flux dan kembalikan hasil yang di-parse.
     */
    public function query(string $fluxQuery): array
    {
        try {
            $response = $this->http->post('/api/v2/query', [
                'headers' => [
                    'Content-Type' => 'application/vnd.flux',
                    'Accept' => 'application/csv',
                ],
                'query' => ['org' => $this->org],
                'body' => $fluxQuery,
            ]);

            return $this->parseCSVResponse((string) $response->getBody());
        } catch (GuzzleException $e) {
            throw new RuntimeException(
                'Query InfluxDB gagal: ' . $e->getMessage(),
                previous: $e,
            );
        }
    }

    /**
     * Periksa konektivitas dan keberadaan bucket.
     */
    public function healthCheck(): bool
    {
        try {
            $response = $this->http->get('/health');
            $body = json_decode((string) $response->getBody(), true);

            return ($body['status'] ?? '') === 'pass';
        } catch (GuzzleException) {
            return false;
        }
    }

    /**
     * Parse respons CSV InfluxDB menjadi array asosiatif.
     */
    private function parseCSVResponse(string $csv): array
    {
        $lines = explode("\n", trim($csv));
        $results = [];

        if (count($lines) < 2) {
            return $results;
        }

        $dataLines = array_filter($lines, fn(string $line): bool => !str_starts_with($line, '#'));
        $dataLines = array_values($dataLines);

        if (empty($dataLines)) {
            return $results;
        }

        $headers = str_getcsv(array_shift($dataLines));

        foreach ($dataLines as $line) {
            $row = str_getcsv($line);
            if (count($row) < count($headers)) {
                continue;
            }
            $results[] = array_combine($headers, $row);
        }

        return $results;
    }
}

Subscriber MQTT dengan Logika Reconnect

<?php

declare(strict_types=1);

namespace App\IoT\Infrastructure;

use PhpMqtt\Client\MqttClient;
use PhpMqtt\Client\ConnectionSettings;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;

final class MQTTSubscriber
{
    private MqttClient $client;
    private int $reconnectDelay = 5;

    public function __construct(
        private string $brokerHost,
        private int $brokerPort,
        private string $clientId,
        private ?LoggerInterface $logger = null,
    ) {
        $this->logger = $logger ?? new NullLogger();
    }

    /**
     * Subscribe ke topik dan proses pesan masuk.
     * Otomatis reconnect saat koneksi terputus.
     *
     * @param callable(SensorReading): void $onMessage
     */
    public function subscribe(string $topic, callable $onMessage): void
    {
        $settings = (new ConnectionSettings())
            ->setKeepAliveInterval(60)
            ->setConnectTimeout(10)
            ->setReconnectAutomatically(true)
            ->setMaxReconnectAttempts(10)
            ->setDelayBetweenReconnectAttempts($this->reconnectDelay * 1000);

        $this->client = new MqttClient(
            $this->brokerHost,
            $this->brokerPort,
            $this->clientId,
        );

        $this->client->registerLoopEventHandler(function () {
            $this->logger?->info('Loop subscriber MQTT berjalan', [
                'client_id' => $this->clientId,
            ]);
        });

        $this->logger?->info('Menghubungkan ke broker MQTT', [
            'host' => $this->brokerHost,
            'port' => $this->brokerPort,
        ]);

        $this->client->connect($settings, true);

        $this->logger?->info('Subscribe ke topik', ['topic' => $topic]);
        $this->client->subscribe($topic, function (string $topic, string $message) use ($onMessage): void {
            $payload = json_decode($message, true);

            if (!is_array($payload)) {
                $this->logger?->warning('Menerima payload JSON tidak valid', [
                    'topic' => $topic,
                    'message' => $message,
                ]);
                return;
            }

            try {
                $reading = \App\IoT\Domain\SensorReading::fromMQTTPayload($payload);
                $onMessage($reading);
            } catch (\Throwable $e) {
                $this->logger?->error('Gagal memproses pembacaan sensor', [
                    'topic' => $topic,
                    'error' => $e->getMessage(),
                ]);
            }
        }, 0);

        $this->client->loop(true);
    }

    public function disconnect(): void
    {
        if (isset($this->client)) {
            $this->client->disconnect();
        }
    }
}

Catatan: Paket php-mqtt/client menyediakan abstraksi bersih di atas protokol MQTT. Dalam produksi, Anda mungkin ingin menjalankan subscriber sebagai proses PHP long-lived menggunakan Swoole, RoadRunner, atau layanan systemd dengan supervisor untuk restart otomatis.

Application Layer: Wiring Everything Together

InfluxDBRepository — Mediating Between Domain and Infrastructure

<?php

declare(strict_types=1);

namespace App\IoT\Application;

use App\IoT\Domain\SensorReading;
use App\IoT\Domain\DashboardQuery;
use App\IoT\Infrastructure\InfluxDBClient;

final class InfluxDBRepository
{
    public function __construct(
        private InfluxDBClient $client,
    ) {}

    /**
     * Write a single sensor reading to InfluxDB.
     */
    public function storeReading(SensorReading $reading): void
    {
        $this->client->write($reading->toInfluxDBLineProtocol() . "\n");
    }

    /**
     * Batch-write multiple readings for efficiency.
     */
    public function storeReadings(array $readings): void
    {
        $lines = array_map(
            fn(SensorReading $r): string => $r->toInfluxDBLineProtocol(),
            $readings,
        );

        $this->client->write(implode("\n", $lines) . "\n");
    }

    /**
     * Query aggregated sensor data for dashboard display.
     */
    public function queryDashboard(DashboardQuery $query): array
    {
        $metricFilter = $query->metric !== null
            ? sprintf('|> filter(fn: (r) => r["metric"] == "%s")', $query->metric)
            : '';

        $fluxQuery = <<<FLUX
from(bucket: "sensor_data")
  |> range(start: -{$query->range})
  |> filter(fn: (r) => r["_measurement"] == "sensor_readings")
  |> filter(fn: (r) => r["sensor_id"] == "{$query->sensorId}")
  {$metricFilter}
  |> filter(fn: (r) => r["_field"] == "value")
  |> aggregateWindow(every: {$query->window}, fn: {$query->aggregation}, createEmpty: true)
  |> yield(name: "{$query->aggregation}")
FLUX;

        return $this->client->query($fluxQuery);
    }

    /**
     * Query the most recent reading for each metric of a sensor.
     */
    public function latestReadings(string $sensorId): array
    {
        $fluxQuery = <<<FLUX
from(bucket: "sensor_data")
  |> range(start: -1h)
  |> filter(fn: (r) => r["_measurement"] == "sensor_readings")
  |> filter(fn: (r) => r["sensor_id"] == "{$sensorId}")
  |> filter(fn: (r) => r["_field"] == "value")
  |> last()
  |> group(columns: ["metric"])
FLUX;

        return $this->client->query($fluxQuery);
    }
}

MQTTIngestionService — The Ingestion Orchestrator

<?php

declare(strict_types=1);

namespace App\IoT\Application;

use App\IoT\Domain\SensorReading;
use App\IoT\Infrastructure\MQTTSubscriber;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;

final class MQTTIngestionService
{
    /** @var SensorReading[] */
    private array $buffer = [];
    private int $batchSize;

    public function __construct(
        private MQTTSubscriber $subscriber,
        private InfluxDBRepository $repository,
        private AlertService $alertService,
        ?int $batchSize = null,
        private ?LoggerInterface $logger = null,
    ) {
        $this->batchSize = $batchSize ?? 100;
        $this->logger = $logger ?? new NullLogger();
    }

    /**
     * Start the MQTT ingestion loop.
     * Subscribes to a wildcard topic and processes all sensor messages.
     */
    public function start(string $topic): void
    {
        $this->logger?->info('Starting MQTT ingestion service', [
            'topic' => $topic,
            'batch_size' => $this->batchSize,
        ]);

        $this->subscriber->subscribe($topic, function (SensorReading $reading): void {
            $this->handleReading($reading);
        });
    }

    /**
     * Process a single sensor reading: buffer, validate, store, alert.
     */
    private function handleReading(SensorReading $reading): void
    {
        $this->logger?->debug('Received sensor reading', [
            'sensor_id' => $reading->sensorId,
            'metric' => $reading->metric,
            'value' => $reading->value,
            'unit' => $reading->unit,
        ]);

        $this->buffer[] = $reading;

        if (count($this->buffer) >= $this->batchSize) {
            $this->flush();
        }

        $alertEvent = $this->alertService->evaluate($reading);
        if ($alertEvent !== null) {
            $this->logger?->warning('Alert triggered', [
                'rule_id' => $alertEvent->ruleId,
                'severity' => $alertEvent->severity->value,
                'message' => $alertEvent->message,
                'value' => $alertEvent->value,
                'threshold' => $alertEvent->threshold,
            ]);

            $this->alertService->dispatch($alertEvent);
        }
    }

    /**
     * Flush buffered readings to InfluxDB in a single batch write.
     */
    public function flush(): void
    {
        if (empty($this->buffer)) {
            return;
        }

        $count = count($this->buffer);

        try {
            $this->repository->storeReadings($this->buffer);
            $this->logger?->info('Flushed sensor readings to InfluxDB', [
                'count' => $count,
            ]);
        } catch (\Throwable $e) {
            $this->logger?->error('Failed to flush readings — data may be lost', [
                'count' => $count,
                'error' => $e->getMessage(),
            ]);

            throw $e;
        } finally {
            $this->buffer = [];
        }
    }
}

Key design decisions in the ingestion service:

• Batching — Individual HTTP writes to InfluxDB are expensive. Accumulate readings in memory and flush every N records (default 100). This reduces HTTP round-trips by 100x.
• Alert evaluation is inline — After every reading, the alert service checks registered rules. This keeps latency low and avoids a separate polling loop.
• Buffer is lost on crash — If the PHP process dies before flush(), buffered readings are lost. This is acceptable for most monitoring use cases, but for regulatory data, write to a persistent queue (Redis Streams, Kafka) before ingesting.

AlertService — Evaluating Rules and Dispatching Alerts

<?php

declare(strict_types=1);

namespace App\IoT\Application;

use App\IoT\Domain\AlertRule;
use App\IoT\Domain\AlertEvent;
use App\IoT\Domain\SensorReading;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;

final class AlertService
{
    /** @var AlertRule[] */
    private array $rules = [];

    /** @var callable[] */
    private array $handlers = [];

    public function __construct(
        private ?LoggerInterface $logger = null,
    ) {
        $this->logger = $logger ?? new NullLogger();
    }

    /**
     * Register an alert rule.
     */
    public function registerRule(AlertRule $rule): void
    {
        $this->rules[] = $rule;
        $this->logger?->info('Alert rule registered', [
            'rule_id' => $rule->id,
            'sensor_id' => $rule->sensorId,
            'metric' => $rule->metric,
            'threshold' => $rule->threshold,
            'severity' => $rule->severity->value,
        ]);
    }

    /**
     * Register an alert handler (email, webhook, SMS, etc.).
     *
     * @param callable(AlertEvent): void $handler
     */
    public function registerHandler(callable $handler): void
    {
        $this->handlers[] = $handler;
    }

    /**
     * Evaluate all registered rules against a sensor reading.
     * Returns the first triggered alert, or null if no rule fires.
     */
    public function evaluate(SensorReading $reading): ?AlertEvent
    {
        foreach ($this->rules as $rule) {
            $event = $rule->evaluate($reading);
            if ($event !== null) {
                return $event;
            }
        }

        return null;
    }

    /**
     * Dispatch an alert event to all registered handlers.
     */
    public function dispatch(AlertEvent $event): void
    {
        foreach ($this->handlers as $handler) {
            try {
                $handler($event);
            } catch (\Throwable $e) {
                $this->logger?->error('Alert handler failed', [
                    'rule_id' => $event->ruleId,
                    'error' => $e->getMessage(),
                ]);
            }
        }
    }
}

DashboardService — Building Dashboard API Responses

<?php

declare(strict_types=1);

namespace App\IoT\Application;

use App\IoT\Domain\DashboardQuery;

final class DashboardService
{
    public function __construct(
        private InfluxDBRepository $repository,
    ) {}

    /**
     * Get aggregated sensor readings for a time-range chart.
     */
    public function getSensorReadings(DashboardQuery $query): array
    {
        $raw = $this->repository->queryDashboard($query);

        return $this->formatChartData($raw);
    }

    /**
     * Get latest sensor values for a live dashboard widget.
     */
    public function getSensorStatus(string $sensorId): array
    {
        $raw = $this->repository->latestReadings($sensorId);

        return array_map(function (array $row): array {
            return [
                'metric' => $row['metric'] ?? 'unknown',
                'value' => (float) ($row['_value'] ?? 0),
                'unit' => $row['unit'] ?? '',
                'timestamp' => $row['_time'] ?? '',
            ];
        }, $raw);
    }

    /**
     * Get all active alerts for display on the dashboard.
     */
    public function getActiveAlerts(): array
    {
        // In a real system, active alerts would be stored in Redis or a relational DB.
        // This is a placeholder that queries recent critical readings from InfluxDB.
        $query = new DashboardQuery(
            sensorId: '.*', // In real code, query all sensors
            range: '10m',
        );

        return $this->repository->queryDashboard($query);
    }

    /**
     * Format raw InfluxDB CSV data into chart-friendly JSON.
     */
    private function formatChartData(array $raw): array
    {
        $labels = [];
        $values = [];

        foreach ($raw as $row) {
            $labels[] = $row['_time'] ?? '';
            $values[] = (float) ($row['_value'] ?? 0);
        }

        return [
            'labels' => $labels,
            'values' => $values,
            'count' => count($raw),
        ];
    }
}

Lapisan Aplikasi: Menyatukan Semuanya

InfluxDBRepository: Menjembatani Domain dan Infrastruktur

<?php

declare(strict_types=1);

namespace App\IoT\Application;

use App\IoT\Domain\SensorReading;
use App\IoT\Domain\DashboardQuery;
use App\IoT\Infrastructure\InfluxDBClient;

final class InfluxDBRepository
{
    public function __construct(
        private InfluxDBClient $client,
    ) {}

    /**
     * Tulis satu pembacaan sensor ke InfluxDB.
     */
    public function storeReading(SensorReading $reading): void
    {
        $this->client->write($reading->toInfluxDBLineProtocol() . "\n");
    }

    /**
     * Tulis batch beberapa pembacaan untuk efisiensi.
     */
    public function storeReadings(array $readings): void
    {
        $lines = array_map(
            fn(SensorReading $r): string => $r->toInfluxDBLineProtocol(),
            $readings,
        );

        $this->client->write(implode("\n", $lines) . "\n");
    }

    /**
     * Query data sensor teragregasi untuk tampilan dasbor.
     */
    public function queryDashboard(DashboardQuery $query): array
    {
        $metricFilter = $query->metric !== null
            ? sprintf('|> filter(fn: (r) => r["metric"] == "%s")', $query->metric)
            : '';

        $fluxQuery = <<<FLUX
from(bucket: "sensor_data")
  |> range(start: -{$query->range})
  |> filter(fn: (r) => r["_measurement"] == "sensor_readings")
  |> filter(fn: (r) => r["sensor_id"] == "{$query->sensorId}")
  {$metricFilter}
  |> filter(fn: (r) => r["_field"] == "value")
  |> aggregateWindow(every: {$query->window}, fn: {$query->aggregation}, createEmpty: true)
  |> yield(name: "{$query->aggregation}")
FLUX;

        return $this->client->query($fluxQuery);
    }

    /**
     * Query pembacaan terbaru untuk setiap metrik dari sebuah sensor.
     */
    public function latestReadings(string $sensorId): array
    {
        $fluxQuery = <<<FLUX
from(bucket: "sensor_data")
  |> range(start: -1h)
  |> filter(fn: (r) => r["_measurement"] == "sensor_readings")
  |> filter(fn: (r) => r["sensor_id"] == "{$sensorId}")
  |> filter(fn: (r) => r["_field"] == "value")
  |> last()
  |> group(columns: ["metric"])
FLUX;

        return $this->client->query($fluxQuery);
    }
}

MQTTIngestionService: Orkestrator Ingesti

<?php

declare(strict_types=1);

namespace App\IoT\Application;

use App\IoT\Domain\SensorReading;
use App\IoT\Infrastructure\MQTTSubscriber;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;

final class MQTTIngestionService
{
    /** @var SensorReading[] */
    private array $buffer = [];
    private int $batchSize;

    public function __construct(
        private MQTTSubscriber $subscriber,
        private InfluxDBRepository $repository,
        private AlertService $alertService,
        ?int $batchSize = null,
        private ?LoggerInterface $logger = null,
    ) {
        $this->batchSize = $batchSize ?? 100;
        $this->logger = $logger ?? new NullLogger();
    }

    /**
     * Mulai loop ingesti MQTT.
     * Subscribe ke topik wildcard dan proses semua pesan sensor.
     */
    public function start(string $topic): void
    {
        $this->logger?->info('Memulai layanan ingesti MQTT', [
            'topic' => $topic,
            'batch_size' => $this->batchSize,
        ]);

        $this->subscriber->subscribe($topic, function (SensorReading $reading): void {
            $this->handleReading($reading);
        });
    }

    /**
     * Proses satu pembacaan sensor: buffer, validasi, simpan, peringatan.
     */
    private function handleReading(SensorReading $reading): void
    {
        $this->logger?->debug('Menerima pembacaan sensor', [
            'sensor_id' => $reading->sensorId,
            'metric' => $reading->metric,
            'value' => $reading->value,
            'unit' => $reading->unit,
        ]);

        $this->buffer[] = $reading;

        if (count($this->buffer) >= $this->batchSize) {
            $this->flush();
        }

        $alertEvent = $this->alertService->evaluate($reading);
        if ($alertEvent !== null) {
            $this->logger?->warning('Peringatan terpicu', [
                'rule_id' => $alertEvent->ruleId,
                'severity' => $alertEvent->severity->value,
                'message' => $alertEvent->message,
                'value' => $alertEvent->value,
                'threshold' => $alertEvent->threshold,
            ]);

            $this->alertService->dispatch($alertEvent);
        }
    }

    /**
     * Kirim pembacaan yang di-buffer ke InfluxDB dalam satu batch write.
     */
    public function flush(): void
    {
        if (empty($this->buffer)) {
            return;
        }

        $count = count($this->buffer);

        try {
            $this->repository->storeReadings($this->buffer);
            $this->logger?->info('Mengirim pembacaan sensor ke InfluxDB', [
                'count' => $count,
            ]);
        } catch (\Throwable $e) {
            $this->logger?->error('Gagal mengirim pembacaan — data mungkin hilang', [
                'count' => $count,
                'error' => $e->getMessage(),
            ]);

            throw $e;
        } finally {
            $this->buffer = [];
        }
    }
}

Keputusan desain kunci dalam layanan ingesti:

• Batching: Penulisan HTTP individual ke InfluxDB mahal. Akumulasi pembacaan di memori dan kirim setiap N record (default 100). Ini mengurangi round-trip HTTP hingga 100x.
• Evaluasi peringatan dilakukan inline: Setelah setiap pembacaan, layanan peringatan memeriksa aturan terdaftar. Ini menjaga latensi tetap rendah dan menghindari loop polling terpisah.
• Buffer hilang saat crash: Jika proses PHP mati sebelum flush(), pembacaan yang di-buffer hilang. Ini dapat diterima untuk sebagian besar kasus pemantauan, tetapi untuk data regulasi, tulis ke antrean persisten (Redis Streams, Kafka) sebelum ingesti.

AlertService: Mengevaluasi Aturan dan Mengirim Peringatan

<?php

declare(strict_types=1);

namespace App\IoT\Application;

use App\IoT\Domain\AlertRule;
use App\IoT\Domain\AlertEvent;
use App\IoT\Domain\SensorReading;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;

final class AlertService
{
    /** @var AlertRule[] */
    private array $rules = [];

    /** @var callable[] */
    private array $handlers = [];

    public function __construct(
        private ?LoggerInterface $logger = null,
    ) {
        $this->logger = $logger ?? new NullLogger();
    }

    /**
     * Daftarkan aturan peringatan.
     */
    public function registerRule(AlertRule $rule): void
    {
        $this->rules[] = $rule;
        $this->logger?->info('Aturan peringatan terdaftar', [
            'rule_id' => $rule->id,
            'sensor_id' => $rule->sensorId,
            'metric' => $rule->metric,
            'threshold' => $rule->threshold,
            'severity' => $rule->severity->value,
        ]);
    }

    /**
     * Daftarkan handler peringatan (email, webhook, SMS, dll.).
     *
     * @param callable(AlertEvent): void $handler
     */
    public function registerHandler(callable $handler): void
    {
        $this->handlers[] = $handler;
    }

    /**
     * Evaluasi semua aturan terdaftar terhadap pembacaan sensor.
     * Mengembalikan peringatan pertama yang terpicu, atau null jika tidak ada aturan yang menyala.
     */
    public function evaluate(SensorReading $reading): ?AlertEvent
    {
        foreach ($this->rules as $rule) {
            $event = $rule->evaluate($reading);
            if ($event !== null) {
                return $event;
            }
        }

        return null;
    }

    /**
     * Kirim kejadian peringatan ke semua handler terdaftar.
     */
    public function dispatch(AlertEvent $event): void
    {
        foreach ($this->handlers as $handler) {
            try {
                $handler($event);
            } catch (\Throwable $e) {
                $this->logger?->error('Handler peringatan gagal', [
                    'rule_id' => $event->ruleId,
                    'error' => $e->getMessage(),
                ]);
            }
        }
    }
}

DashboardService: Membangun Respons API Dasbor

<?php

declare(strict_types=1);

namespace App\IoT\Application;

use App\IoT\Domain\DashboardQuery;

final class DashboardService
{
    public function __construct(
        private InfluxDBRepository $repository,
    ) {}

    /**
     * Dapatkan pembacaan sensor teragregasi untuk grafik rentang waktu.
     */
    public function getSensorReadings(DashboardQuery $query): array
    {
        $raw = $this->repository->queryDashboard($query);

        return $this->formatChartData($raw);
    }

    /**
     * Dapatkan nilai sensor terbaru untuk widget dasbor langsung.
     */
    public function getSensorStatus(string $sensorId): array
    {
        $raw = $this->repository->latestReadings($sensorId);

        return array_map(function (array $row): array {
            return [
                'metric' => $row['metric'] ?? 'tidak diketahui',
                'value' => (float) ($row['_value'] ?? 0),
                'unit' => $row['unit'] ?? '',
                'timestamp' => $row['_time'] ?? '',
            ];
        }, $raw);
    }

    /**
     * Dapatkan semua peringatan aktif untuk ditampilkan di dasbor.
     */
    public function getActiveAlerts(): array
    {
        $query = new DashboardQuery(
            sensorId: '.*',
            range: '10m',
        );

        return $this->repository->queryDashboard($query);
    }

    /**
     * Format data CSV InfluxDB mentah menjadi JSON ramah grafik.
     */
    private function formatChartData(array $raw): array
    {
        $labels = [];
        $values = [];

        foreach ($raw as $row) {
            $labels[] = $row['_time'] ?? '';
            $values[] = (float) ($row['_value'] ?? 0);
        }

        return [
            'labels' => $labels,
            'values' => $values,
            'count' => count($raw),
        ];
    }
}

Building the Dashboard API Endpoint

The dashboard API exposes sensor data to frontend charts. Here is a minimal PHP router that ties everything together.

<?php

declare(strict_types=1);

namespace App\IoT\UI;

use App\IoT\Domain\DashboardQuery;
use App\IoT\Application\DashboardService;
use App\IoT\Application\MQTTIngestionService;
use App\IoT\Application\InfluxDBRepository;
use App\IoT\Application\AlertService;

/**
 * Minimal dashboard API router.
 * In production, use a framework (Laravel, Slim) or the PSR-7/PSR-15 stack.
 */
final class DashboardController
{
    public function __construct(
        private DashboardService $dashboardService,
        private MQTTIngestionService $ingestionService,
        private AlertService $alertService,
    ) {}

    /**
     * GET /api/dashboard/sensors/{sensorId}/readings
     *
     * Query parameters:
     *   - metric   (string, optional)  — filter by metric name
     *   - range    (string, default 1h) — time range (1h, 6h, 24h, 7d)
     *   - window   (string, default 1m) — aggregation window
     *   - aggregation (string, default mean) — mean, median, max, min, sum
     *
     * Response:
     * {
     *   "sensor_id": "temp-001",
     *   "metric": "temperature",
     *   "range": "1h",
     *   "data": { "labels": [...], "values": [...] },
     *   "count": 60
     * }
     */
    public function readings(string $sensorId, array $params): array
    {
        $params['sensor_id'] = $sensorId;
        $query = DashboardQuery::fromArray($params);

        $data = $this->dashboardService->getSensorReadings($query);

        return [
            'sensor_id' => $sensorId,
            'metric' => $params['metric'] ?? 'all',
            'range' => $query->range,
            'data' => $data,
            'count' => $data['count'],
        ];
    }

    /**
     * GET /api/dashboard/sensors/{sensorId}/status
     *
     * Response:
     * {
     *   "sensor_id": "temp-001",
     *   "latest": [
     *     { "metric": "temperature", "value": 72.5, "unit": "celsius", "timestamp": "..." },
     *     { "metric": "humidity", "value": 45.2, "unit": "percent", "timestamp": "..." }
     *   ]
     * }
     */
    public function status(string $sensorId): array
    {
        $latest = $this->dashboardService->getSensorStatus($sensorId);

        return [
            'sensor_id' => $sensorId,
            'latest' => $latest,
        ];
    }

    /**
     * GET /api/dashboard/alerts
     *
     * Response:
     * {
     *   "alerts": [
     *     { "rule_id": "...", "severity": "critical", "message": "...", "timestamp": "..." }
     *   ],
     *   "count": 2
     * }
     */
    public function alerts(): array
    {
        $alerts = $this->dashboardService->getActiveAlerts();

        return [
            'alerts' => $alerts,
            'count' => count($alerts),
        ];
    }
}

Wiring It All Together: Entry Point

<?php

declare(strict_types=1);

use App\IoT\Infrastructure\InfluxDBClient;
use App\IoT\Infrastructure\MQTTSubscriber;
use App\IoT\Application\InfluxDBRepository;
use App\IoT\Application\MQTTIngestionService;
use App\IoT\Application\DashboardService;
use App\IoT\Application\AlertService;
use App\IoT\Domain\AlertRule;
use App\IoT\Domain\AlertOperator;
use App\IoT\Domain\AlertSeverity;
use App\IoT\Domain\AlertEvent;
use App\IoT\UI\DashboardController;

require_once __DIR__ . '/vendor/autoload.php';

// --- Infrastructure wiring ---

$influxDB = new InfluxDBClient(
    url: 'http://localhost:8086',
    token: 'my-super-secret-token',
    org: 'se-polinema',
    bucket: 'sensor_data',
);

$subscriber = new MQTTSubscriber(
    brokerHost: 'localhost',
    brokerPort: 1883,
    clientId: 'se-polinema-ingestion-' . getmypid(),
);

// --- Application wiring ---

$repository = new InfluxDBRepository($influxDB);
$alertService = new AlertService();

// Register alert rules
$alertService->registerRule(new AlertRule(
    id: 'temp-critical',
    sensorId: 'temp-001',
    metric: 'temperature',
    operator: AlertOperator::GREATER_THAN,
    threshold: 80.0,
    severity: AlertSeverity::CRITICAL,
    message: 'Motor temperature exceeds safe operating limit!',
));

$alertService->registerRule(new AlertRule(
    id: 'vib-warning',
    sensorId: 'vib-002',
    metric: 'vibration',
    operator: AlertOperator::GREATER_THAN_OR_EQUAL,
    threshold: 7.5,
    severity: AlertSeverity::WARNING,
    message: 'Vibration level approaching maintenance threshold.',
));

// Register alert handler — log to stdout
$alertService->registerHandler(function (AlertEvent $event): void {
    echo sprintf(
        "[ALERT %s] [%s] %s | value=%.2f threshold=%.2f\n",
        strtoupper($event->severity->value),
        $event->timestamp->format('Y-m-d H:i:s'),
        $event->message,
        $event->value,
        $event->threshold,
    );
});

$ingestionService = new MQTTIngestionService(
    subscriber: $subscriber,
    repository: $repository,
    alertService: $alertService,
    batchSize: 100,
);

$dashboardService = new DashboardService($repository);
$dashboard = new DashboardController($dashboardService, $ingestionService, $alertService);

// --- Run: Choose one mode ---

$mode = $argv[1] ?? 'ingest';

if ($mode === 'ingest') {
    // Start MQTT ingestion — runs indefinitely
    echo "Starting MQTT ingestion...\n";
    $ingestionService->start('factory/+/+');
}

if ($mode === 'server') {
    // Minimal built-in server for dashboard API
    // Run: php entrypoint.php server
    echo "Starting dashboard API server on http://localhost:8080\n";
    echo "Endpoints:\n";
    echo "  GET /api/dashboard/sensors/{id}/readings\n";
    echo "  GET /api/dashboard/sensors/{id}/status\n";
    echo "  GET /api/dashboard/alerts\n";
}