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 $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.`). * * @param array{key:string,type:string,state:array} $update * @return array{key:string,type:string,state:array} */ 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; // Keep the RAW input level (`on`) in the state so the device page can re-derive open/closed // when the user flips inversion or the role, without the displayed value having been baked in. return [ 'key' => $update['key'], 'type' => 'contact', 'state' => ['open' => $open, 'position' => $open ? 'open' : 'closed', 'kind' => $role['kind'] ?? 'window', 'on' => $on], ]; } 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(); } } }