Lines 95.14% 98 / 103
Functions and Methods 76.19% 16 / 21
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
WebsocketHaSession 95.14% 98 / 103 76.19% 16 / 21 40 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 open 93.33% 14 / 15 0.00% 0 / 1 2.00
 close 100.00% 3 / 3 100.00% 1 / 1 1
 isOpen 100.00% 1 / 1 100.00% 1 / 1 1
 isConnected 100.00% 1 / 1 100.00% 1 / 1 2
 snapshotStateCache 100.00% 3 / 3 100.00% 1 / 1 2
 snapshotRegistry 100.00% 4 / 4 100.00% 1 / 1 2
 refreshRegistry 87.50% 7 / 8 0.00% 0 / 1 4.03
 countEntities 100.00% 1 / 1 100.00% 1 / 1 1
 getTimeZone 100.00% 1 / 1 100.00% 1 / 1 1
 getHaVersion 0.00% 0 / 1 0.00% 0 / 1 2
 getHaUserId 100.00% 1 / 1 100.00% 1 / 1 1
 getLocation 100.00% 1 / 1 100.00% 1 / 1 1
 listEntityIds 100.00% 1 / 1 100.00% 1 / 1 1
 connectSubscribeAndSeed 100.00% 21 / 21 100.00% 1 / 1 2
 onStateChanged 100.00% 6 / 6 100.00% 1 / 1 2
 onEventFired 66.66% 2 / 3 0.00% 0 / 1 3.33
 onTriggerFired 100.00% 3 / 3 100.00% 1 / 1 4
 onTriggerRejected 100.00% 2 / 2 100.00% 1 / 1 1
 onDisconnected 91.66% 11 / 12 0.00% 0 / 1 4.01
 reconnectAfterOutage 100.00% 14 / 14 100.00% 1 / 1 3
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Broker;
6
7use Amp\DeferredCancellation;
8use DateTimeZone;
9use Psr\Log\LoggerInterface;
10use Stewart\Client\Exception\HaClientException;
11use Stewart\Client\HaClient;
12use Stewart\Client\HaSiteSettings;
13use Stewart\Contracts\Event\HaEvent;
14use Stewart\Contracts\State\StateChange;
15use Stewart\Contracts\Sun\GeoLocation;
16use Stewart\Contracts\Time\Clock;
17use Stewart\Contracts\Time\MonotonicTime;
18use Stewart\Contracts\Trigger\TriggerEvent;
19use Stewart\Runtime\Broker\Component\ComponentSession;
20use Stewart\Runtime\Broker\Exposure\ExposureLink;
21use Stewart\Runtime\Broker\Trigger\HaTriggerLink;
22use Stewart\Runtime\Ipc\Wire\EntityStatesFragment;
23use Stewart\Runtime\Ipc\Wire\RegistryFragment;
24use Stewart\Runtime\Registry\RegistryCache;
25use Stewart\Runtime\State\StateCache;
26use Stewart\Support\Wire\TriggerSpec;
27use Throwable;
28
29use function Amp\async;
30
31final class WebsocketHaSession implements HaSession
32{
33    private HaSessionListener $listener;
34
35    private int $revision = 0;
36
37    private bool $open = false;
38
39    private readonly DeferredCancellation $stop;
40
41    private bool $reconnecting = false;
42
43    private bool $establishing = false;
44
45    /** @var array<string, StateChange> */
46    private array $pendingChanges = [];
47
48    private ?HaSiteSettings $siteSettings = null;
49
50    private ?string $haUserId = null;
51
52    private ?StateCacheSnapshot $stateCacheSnapshot = null;
53
54    private ?RegistryCacheSnapshot $registryCacheSnapshot = null;
55
56    public function __construct(
57        private readonly HaClient $client,
58        private readonly Reconnector $reconnector,
59        private readonly LoggerInterface $logger,
60        private readonly Clock $clock,
61        private readonly StateCache $states,
62        private readonly RegistryCache $registry,
63        private readonly HaTriggerLink $triggers,
64        private readonly ComponentSession $componentSession,
65        private readonly ExposureLink $exposures,
66    ) {
67        $this->stop = new DeferredCancellation();
68    }
69
70    public function open(HaSessionListener $listener): void
71    {
72        $this->listener = $listener;
73        $this->open = true;
74        $this->triggers->onTriggerFired($this->onTriggerFired(...));
75        $this->triggers->onTriggerRejected($this->onTriggerRejected(...));
76        $this->exposures->onEntitySynced($listener->exposedEntitySynced(...));
77        $this->exposures->onEntityCommanded($listener->exposedEntityCommanded(...));
78
79        $this->reconnector->retryUntilConnected($this->connectSubscribeAndSeed(...), $this->stop->getCancellation());
80
81        if (!$this->isOpen()) {
82            return;
83        }
84
85        $this->client->onDisconnect($this->onDisconnected(...));
86
87        $this->logger->info('Connected to Home Assistant', [
88            'ha_version' => $this->client->getHaVersion(),
89            'entities' => $this->states->count(),
90            'time_zone' => $this->getTimeZone()->getName(),
91        ]);
92    }
93
94    public function close(): void
95    {
96        $this->open = false;
97        $this->stop->cancel();
98        $this->client->close();
99    }
100
101    private function isOpen(): bool
102    {
103        return $this->open;
104    }
105
106    public function isConnected(): bool
107    {
108        return $this->open && !$this->reconnecting;
109    }
110
111    public function snapshotStateCache(): StateCacheSnapshot
112    {
113        if ($this->stateCacheSnapshot?->revision !== $this->revision) {
114            $this->stateCacheSnapshot = new StateCacheSnapshot(EntityStatesFragment::fromCollection($this->states->getAllStates()), $this->revision);
115        }
116
117        return $this->stateCacheSnapshot;
118    }
119
120    public function snapshotRegistry(): RegistryCacheSnapshot
121    {
122        $revision = $this->registry->getRevision();
123
124        if ($this->registryCacheSnapshot?->revision !== $revision) {
125            $this->registryCacheSnapshot = new RegistryCacheSnapshot(RegistryFragment::fromRegistry($this->registry->getIndexedRegistry()), $revision);
126        }
127
128        return $this->registryCacheSnapshot;
129    }
130
131    public function refreshRegistry(): bool
132    {
133        if (!$this->isConnected() || $this->establishing) {
134            return false;
135        }
136
137        try {
138            $registry = $this->client->getRegistry();
139        } catch (HaClientException $e) {
140            $this->logger->warning('Could not refresh the Home Assistant registry; the previous one stays', ['exception' => $e]);
141
142            return false;
143        }
144
145        $this->registry->replaceAsNextRevision($registry);
146
147        return true;
148    }
149
150    public function countEntities(): int
151    {
152        return $this->states->count();
153    }
154
155    public function getTimeZone(): DateTimeZone
156    {
157        return $this->siteSettings->timeZone ?? new DateTimeZone('UTC');
158    }
159
160    public function getHaVersion(): ?string
161    {
162        return $this->client->getHaVersion();
163    }
164
165    public function getHaUserId(): ?string
166    {
167        return $this->haUserId;
168    }
169
170    public function getLocation(): ?GeoLocation
171    {
172        return $this->siteSettings?->location;
173    }
174
175    public function listEntityIds(): array
176    {
177        return $this->states->listEntityIds()->toStrings();
178    }
179
180    private function connectSubscribeAndSeed(): void
181    {
182        // Subscribe before seeding and buffer changes in between, so no change is lost.
183        $this->triggers->markLinkLost();
184        $this->componentSession->markLinkLost();
185        $this->client->close();
186        $this->client->connect();
187        $this->establishing = true;
188
189        try {
190            $this->client->subscribeAllEvents($this->onStateChanged(...), $this->onEventFired(...));
191            $this->triggers->resubscribeAll();
192            $this->componentSession->openSession();
193            $this->siteSettings ??= $this->client->getSiteSettings();
194            $this->haUserId ??= $this->client->getCurrentUserId();
195            $states = $this->client->getStates();
196            $registry = $this->client->getRegistry();
197            $this->client->flushEvents();
198
199            $this->establishing = false;
200            $this->registry->replaceAsNextRevision($registry);
201            $this->states->replaceAll($states);
202
203            foreach ($this->pendingChanges as $change) {
204                $this->states->applyChange($change);
205            }
206
207            ++$this->revision;
208        } finally {
209            $this->establishing = false;
210            $this->pendingChanges = [];
211        }
212    }
213
214    private function onStateChanged(StateChange $change): void
215    {
216        if ($this->establishing) {
217            $this->pendingChanges[$change->entityId->value] = $change;
218
219            return;
220        }
221
222        $this->states->applyChange($change);
223        ++$this->revision;
224
225        $this->listener->stateChanged($change);
226    }
227
228    private function onEventFired(HaEvent $event): void
229    {
230        // Dropped until the state cache is rebuilt; handlers would read pre-outage state.
231        if ($this->establishing || $this->reconnecting) {
232            return;
233        }
234
235        $this->listener->eventFired($event);
236    }
237
238    private function onTriggerFired(TriggerSpec $spec, TriggerEvent $event): void
239    {
240        if ($this->establishing || $this->reconnecting) {
241            return;
242        }
243
244        $this->listener->triggerFired($spec, $event->firedAt === null ? $event->withFiredAt($this->clock->getNow()) : $event);
245    }
246
247    private function onTriggerRejected(TriggerSpec $spec, string $reason): void
248    {
249        $this->logger->error('Home Assistant rejected a trigger subscription', ['platforms' => $spec->listPlatforms(), 'reason' => $reason]);
250
251        $this->listener->triggerRejected($spec, $reason);
252    }
253
254    private function onDisconnected(HaClientException $lost): void
255    {
256        if (!$this->open || $this->reconnecting) {
257            return;
258        }
259
260        $this->triggers->markLinkLost();
261        $this->componentSession->markLinkLost();
262
263        $this->reconnecting = true;
264        $startedAt = $this->clock->getMonotonicTime();
265
266        $this->listener->connectionLost($lost->getMessage(), $this->clock->getNow());
267
268        async(function () use ($startedAt): void {
269            try {
270                $this->reconnectAfterOutage($startedAt);
271            } catch (Throwable $e) {
272                $this->logger->error('Failed handling a Home Assistant reconnect', ['exception' => $e]);
273            }
274        })->ignore();
275    }
276
277    private function reconnectAfterOutage(MonotonicTime $startedAt): void
278    {
279        try {
280            $this->reconnector->retryUntilConnected($this->connectSubscribeAndSeed(...), $this->stop->getCancellation());
281        } catch (Throwable $e) {
282            $this->reconnecting = false;
283            $this->listener->connectionFailed($e);
284
285            return;
286        }
287
288        $this->reconnecting = false;
289
290        if (!$this->isOpen()) {
291            return;
292        }
293
294        $outage = $this->clock->getMonotonicTime()->elapsedSince($startedAt);
295
296        $this->logger->info('Reconnected to Home Assistant', [
297            'outage' => (string) $outage,
298            'entities' => $this->states->count(),
299        ]);
300
301        $this->listener->reconnected($outage);
302    }
303}