113 lines
4.2 KiB
PHP
113 lines
4.2 KiB
PHP
<?php
|
|
|
|
namespace App\Jobs;
|
|
|
|
use App\Events\DeviceStateChanged;
|
|
use App\Jobs\Concerns\AppliesDeviceState;
|
|
use App\Models\Device;
|
|
use App\Support\Mqtt\ShellyNormalizer;
|
|
use App\Support\Mqtt\ShellyTopics;
|
|
use Illuminate\Contracts\Queue\ShouldQueue;
|
|
use Illuminate\Database\UniqueConstraintViolationException;
|
|
use Illuminate\Foundation\Queue\Queueable;
|
|
|
|
/**
|
|
* The single ingest path for Shelly status messages (H4). The MQTT callback only
|
|
* dispatches this (H2); here we resolve the device, upsert current state and broadcast.
|
|
*
|
|
* Devices auto-onboard: the first time a prefix publishes a recognizable component the device is
|
|
* created (Home-Assistant-style), so pointing a Shelly at the broker is all it takes — no manual
|
|
* assignment. Housekeeping topics (sys/wifi/cloud) never create a device.
|
|
*
|
|
* `observedAt` is the µs receive time from the listener. Because concurrent Horizon
|
|
* workers can finish out of order, state is only written when this message is newer than
|
|
* what is stored — an older message can never overwrite a newer one.
|
|
*/
|
|
class IngestShellyMessage implements ShouldQueue
|
|
{
|
|
use AppliesDeviceState;
|
|
use Queueable;
|
|
|
|
public function __construct(
|
|
public string $topic,
|
|
public string $payload,
|
|
public ?int $observedAt = null,
|
|
) {}
|
|
|
|
public function handle(): void
|
|
{
|
|
try {
|
|
\Illuminate\Support\Facades\Redis::connection()->incr('metrics:mqtt:count');
|
|
} catch (\Throwable) {
|
|
// metrics are best-effort
|
|
}
|
|
|
|
$parsed = ShellyTopics::parseStatus($this->topic);
|
|
if ($parsed === null) {
|
|
return;
|
|
}
|
|
|
|
[$prefix, $component] = $parsed;
|
|
|
|
$data = json_decode($this->payload, true);
|
|
if (! is_array($data)) {
|
|
return;
|
|
}
|
|
|
|
$updates = ShellyNormalizer::normalize($component, $data);
|
|
|
|
$device = Device::where('config->mqtt_prefix', $prefix)->first();
|
|
if ($device === null) {
|
|
// Auto-onboard on the first recognizable component; ignore sys/wifi/cloud noise.
|
|
if (! config('homeos.mqtt.auto_onboard') || ! $this->hasPrimaryType($updates)) {
|
|
return;
|
|
}
|
|
$device = $this->onboard($prefix);
|
|
}
|
|
|
|
$observedAt = $this->observedAt ?? (int) round(microtime(true) * 1_000_000);
|
|
$observedTime = \Illuminate\Support\Carbon::createFromTimestampMs(intdiv($observedAt, 1000));
|
|
|
|
// Presence reflects when the message was RECEIVED, and only ever advances. The guard
|
|
// lives in the WHERE clause (one atomic UPDATE), so concurrent workers can't rewind it
|
|
// via read-then-write — queue latency, retries or stale messages never mark it wrong.
|
|
Device::whereKey($device->id)
|
|
->where(fn ($q) => $q->whereNull('last_seen_at')->orWhere('last_seen_at', '<', $observedTime))
|
|
->update(['last_seen_at' => $observedTime, 'updated_at' => now()]);
|
|
|
|
foreach ($updates as $update) {
|
|
$entity = $device->entities()->firstOrCreate(
|
|
['key' => $update['key']],
|
|
['type' => $update['type']],
|
|
);
|
|
|
|
if ($this->applyState($entity->id, $update['state'], $observedAt)) {
|
|
DeviceStateChanged::dispatch($device->uuid, $entity->key, $update['state']);
|
|
}
|
|
}
|
|
}
|
|
|
|
/** @param array<int, array{type:string}> $updates */
|
|
private function hasPrimaryType(array $updates): bool
|
|
{
|
|
return array_intersect(array_column($updates, 'type'), ShellyNormalizer::PRIMARY_TYPES) !== [];
|
|
}
|
|
|
|
private function onboard(string $prefix): Device
|
|
{
|
|
try {
|
|
return Device::create([
|
|
'name' => $prefix, // user renames on the device detail page
|
|
'vendor' => 'Shelly',
|
|
'protocol' => 'mqtt',
|
|
'status' => 'active',
|
|
'last_seen_at' => now(),
|
|
'config' => ['mqtt_prefix' => $prefix, 'auto_onboarded' => true],
|
|
]);
|
|
} catch (UniqueConstraintViolationException) {
|
|
// A concurrent worker onboarded it first (devices_mqtt_prefix_unique) — use the winner.
|
|
return Device::where('config->mqtt_prefix', $prefix)->firstOrFail();
|
|
}
|
|
}
|
|
}
|