63 lines
1.7 KiB
PHP
63 lines
1.7 KiB
PHP
<?php
|
|
|
|
namespace App\Jobs;
|
|
|
|
use App\Events\DeviceStateChanged;
|
|
use App\Models\Device;
|
|
use App\Models\DeviceState;
|
|
use App\Support\Mqtt\ShellyNormalizer;
|
|
use App\Support\Mqtt\ShellyTopics;
|
|
use Illuminate\Contracts\Queue\ShouldQueue;
|
|
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.
|
|
*/
|
|
class IngestShellyMessage implements ShouldQueue
|
|
{
|
|
use Queueable;
|
|
|
|
public function __construct(
|
|
public string $topic,
|
|
public string $payload,
|
|
) {}
|
|
|
|
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;
|
|
}
|
|
|
|
$device->forceFill(['last_seen_at' => now()])->save();
|
|
|
|
foreach (ShellyNormalizer::normalize($component, $data) as $update) {
|
|
$entity = $device->entities()->firstOrCreate(
|
|
['key' => $update['key']],
|
|
['type' => $update['type']],
|
|
);
|
|
|
|
DeviceState::updateOrCreate(
|
|
['entity_id' => $entity->id],
|
|
['state' => $update['state']],
|
|
);
|
|
|
|
DeviceStateChanged::dispatch($device->uuid, $entity->key, $update['state']);
|
|
}
|
|
}
|
|
}
|