117 lines
4.1 KiB
PHP
117 lines
4.1 KiB
PHP
<?php
|
|
|
|
namespace App\Jobs;
|
|
|
|
use App\Events\DeviceStateChanged;
|
|
use App\Jobs\Concerns\AppliesDeviceState;
|
|
use App\Models\Addon;
|
|
use App\Models\Device;
|
|
use App\Services\AddonService;
|
|
use App\Support\Mqtt\RingNormalizer;
|
|
use App\Support\Mqtt\RingTopics;
|
|
use Illuminate\Contracts\Queue\ShouldQueue;
|
|
use Illuminate\Database\UniqueConstraintViolationException;
|
|
use Illuminate\Foundation\Queue\Queueable;
|
|
use Illuminate\Support\Carbon;
|
|
use Illuminate\Support\Facades\Cache;
|
|
use Illuminate\Support\Str;
|
|
|
|
/**
|
|
* The single ingest path for ring-mqtt messages (H4). The MQTT callback only dispatches this
|
|
* (H2). Ring devices are cloud-bridged, so — unlike locally-discovered Shelly — they are
|
|
* auto-created here on first sight and flagged `cloud` for the "Cloud" badge (handoff §12).
|
|
*/
|
|
class IngestRingMessage 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
|
|
}
|
|
|
|
// The bridge's own health topic drives the add-on connection status.
|
|
if (RingTopics::isBridgeStatus($this->topic)) {
|
|
$online = strtolower(trim($this->payload)) === 'online';
|
|
app(AddonService::class)->markStatus('ring', $online ? 'connected' : 'error',
|
|
$online ? null : 'bridge offline');
|
|
|
|
return;
|
|
}
|
|
|
|
// Ignore everything unless the user actually installed Ring — stops a still-running
|
|
// bridge from recreating devices after an uninstall. Cached so motion floods stay cheap.
|
|
if (! $this->ringInstalled()) {
|
|
return;
|
|
}
|
|
|
|
$parsed = RingTopics::parseState($this->topic);
|
|
if ($parsed === null) {
|
|
return;
|
|
}
|
|
|
|
$updates = RingNormalizer::normalize($parsed['entity'], $this->payload);
|
|
if ($updates === []) {
|
|
return;
|
|
}
|
|
|
|
$observedAt = $this->observedAt ?? (int) round(microtime(true) * 1_000_000);
|
|
$observedTime = Carbon::createFromTimestampMs(intdiv($observedAt, 1000));
|
|
|
|
$device = $this->resolveDevice($parsed['ring_id'], $parsed['category']);
|
|
|
|
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']);
|
|
}
|
|
}
|
|
}
|
|
|
|
private function resolveDevice(string $ringId, string $category): Device
|
|
{
|
|
$existing = Device::where('config->ring_id', $ringId)->first();
|
|
if ($existing !== null) {
|
|
return $existing;
|
|
}
|
|
|
|
try {
|
|
return Device::create([
|
|
'name' => Str::headline($category).' · '.Str::upper(Str::substr($ringId, -4)),
|
|
'vendor' => 'Ring',
|
|
'protocol' => 'mqtt',
|
|
'status' => 'active',
|
|
'last_seen_at' => now(),
|
|
'config' => ['cloud' => true, 'ring_id' => $ringId, 'ring_category' => $category],
|
|
]);
|
|
} catch (UniqueConstraintViolationException) {
|
|
// A concurrent worker created it first (devices_ring_id_unique) — use the winner.
|
|
return Device::where('config->ring_id', $ringId)->firstOrFail();
|
|
}
|
|
}
|
|
|
|
private function ringInstalled(): bool
|
|
{
|
|
return Cache::remember(AddonService::installedCacheKey('ring'), 15,
|
|
fn () => Addon::where('key', 'ring')->where('installed', true)->exists());
|
|
}
|
|
}
|