111 lines
4.1 KiB
PHP
111 lines
4.1 KiB
PHP
<?php
|
|
|
|
namespace App\Jobs;
|
|
|
|
use App\Events\DeviceStateChanged;
|
|
use App\Models\Device;
|
|
use App\Models\DeviceState;
|
|
use App\Models\Entity;
|
|
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.
|
|
*
|
|
* `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 Queueable;
|
|
|
|
public function __construct(
|
|
public string $topic,
|
|
public string $payload,
|
|
public ?int $observedAt = null,
|
|
) {}
|
|
|
|
public function handle(): void
|
|
{
|
|
$parsed = ShellyTopics::parseStatus($this->topic);
|
|
if ($parsed === null) {
|
|
return;
|
|
}
|
|
|
|
[$prefix, $component] = $parsed;
|
|
|
|
// Unknown devices are handled by discovery (Phase 4), not silently created here.
|
|
$device = Device::where('config->mqtt_prefix', $prefix)->first();
|
|
if ($device === null) {
|
|
return;
|
|
}
|
|
|
|
$data = json_decode($this->payload, true);
|
|
if (! is_array($data)) {
|
|
return;
|
|
}
|
|
|
|
$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 advances — queue latency,
|
|
// retries or stale messages can never refresh (or rewind) last_seen incorrectly.
|
|
if ($device->last_seen_at === null || $observedTime->gt($device->last_seen_at)) {
|
|
$device->forceFill(['last_seen_at' => $observedTime])->save();
|
|
}
|
|
|
|
foreach (ShellyNormalizer::normalize($component, $data) 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']);
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Monotonic, race-safe upsert. Returns true when this message's state was applied,
|
|
* false when it was stale (an equal-or-newer state is already stored).
|
|
*/
|
|
private function applyState(int $entityId, array $state, int $observedAt): bool
|
|
{
|
|
$encoded = json_encode($state);
|
|
|
|
// Overwrite an existing row only when this message is newer.
|
|
$updated = DeviceState::query()
|
|
->where('entity_id', $entityId)
|
|
->where(fn ($q) => $q->whereNull('observed_at')->orWhere('observed_at', '<', $observedAt))
|
|
->update(['state' => $encoded, 'observed_at' => $observedAt, 'updated_at' => now()]);
|
|
|
|
if ($updated > 0) {
|
|
return true;
|
|
}
|
|
|
|
// No row yet → insert (unique(entity_id) guards against a concurrent insert).
|
|
if (! DeviceState::where('entity_id', $entityId)->exists()) {
|
|
try {
|
|
DeviceState::create(['entity_id' => $entityId, 'state' => $state, 'observed_at' => $observedAt]);
|
|
|
|
return true;
|
|
} catch (UniqueConstraintViolationException) {
|
|
// Lost the insert race — fall through and retry the conditional update.
|
|
return DeviceState::query()
|
|
->where('entity_id', $entityId)
|
|
->where(fn ($q) => $q->whereNull('observed_at')->orWhere('observed_at', '<', $observedAt))
|
|
->update(['state' => $encoded, 'observed_at' => $observedAt, 'updated_at' => now()]) > 0;
|
|
}
|
|
}
|
|
|
|
// Existing row is equal-or-newer → this message is stale.
|
|
return false;
|
|
}
|
|
}
|