homeos/app/Jobs/IngestShellyMessage.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;

    /** Prefixes a device may never claim — they belong to internal/other-integration namespaces. */
    private const RESERVED_PREFIXES = ['homeos', 'ring', '$SYS'];

    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;

        // Reserved namespaces still match the broker's `+/status/#` wildcard; never let a device
        // masquerade as one of them (would create a bogus "homeos"/"ring" device).
        if (in_array($prefix, self::RESERVED_PREFIXES, true)) {
            return;
        }

        $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);
            if ($device === null) {
                return; // onboarding cap reached — drop rather than flood the DB
            }
        }

        // Apply per-device input roles: an input the user promoted to a window/door contact
        // becomes a contact entity (invert-aware). Config-driven, so this stays out of the driver.
        $updates = array_map(fn ($u) => $this->applyInputRole($device, $u), $updates);

        $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']],
            );

            // Keep the entity type in sync with its (config-driven) role — e.g. an input promoted
            // to a contact, or demoted back — so the correct renderer/query applies.
            if ($entity->type !== $update['type']) {
                $entity->forceFill(['type' => $update['type']])->save();
            }

            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) !== [];
    }

    /**
     * Promote a raw `input` update to a `contact` when the user has assigned that input a
     * window/door role on the device page (stored in device config `input_roles.<idx>`).
     *
     * @param  array{key:string,type:string,state:array<string,mixed>}  $update
     * @return array{key:string,type:string,state:array<string,mixed>}
     */
    private function applyInputRole(Device $device, array $update): array
    {
        if ($update['type'] !== 'input') {
            return $update;
        }

        $index = explode(':', $update['key'])[1] ?? '0';
        $role = data_get($device->config, "input_roles.{$index}");

        if (! is_array($role)) {
            return $update;
        }

        $on = (bool) ($update['state']['on'] ?? false);
        $open = ($role['invert'] ?? false) ? ! $on : $on;

        return [
            'key' => $update['key'],
            'type' => 'contact',
            'state' => ['open' => $open, 'position' => $open ? 'open' : 'closed', 'kind' => $role['kind'] ?? 'window'],
        ];
    }

    private function onboard(string $prefix): ?Device
    {
        // Cap auto-onboarding so a misbehaving/compromised device flooding distinct prefixes on
        // the shared account can't exhaust the DB. Assigned devices are unaffected.
        $cap = (int) config('homeos.mqtt.max_devices', 250);
        if ($cap > 0 && Device::count() >= $cap) {
            return null;
        }

        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();
        }
    }
}