Lines 78.19% 104 / 133
Functions and Methods 64.28% 18 / 28
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
ExposureLink 78.19% 104 / 133 64.28% 18 / 28 113.54 0.00% 0 / 1
 __construct 100.00% 3 / 3 100.00% 1 / 1 1
 onEntitySynced 100.00% 1 / 1 100.00% 1 / 1 1
 onEntityCommanded 100.00% 1 / 1 100.00% 1 / 1 1
 onSessionLost 0.00% 0 / 1 0.00% 0 / 1 2
 receiveCommand 100.00% 11 / 11 100.00% 1 / 1 3
 acceptCommand 100.00% 4 / 4 100.00% 1 / 1 2
 rejectCommand 100.00% 1 / 1 100.00% 1 / 1 1
 exposeEntity 100.00% 9 / 9 100.00% 1 / 1 3
 reconfigureEntity 100.00% 10 / 10 100.00% 1 / 1 3
 updateEntity 88.88% 8 / 9 0.00% 0 / 1 3.01
 removeEntity 36.36% 4 / 11 0.00% 0 / 1 5.32
 orphanExposuresOfApp 100.00% 1 / 1 100.00% 1 / 1 1
 orphanExposuresOfWorker 100.00% 1 / 1 100.00% 1 / 1 1
 countExposuresOfApp 100.00% 1 / 1 100.00% 1 / 1 1
 findAddressByEntityId 100.00% 2 / 2 100.00% 1 / 1 1
 listKeptAddresses 0.00% 0 / 1 0.00% 0 / 1 2
 replayAll 55.55% 5 / 9 0.00% 0 / 1 9.16
 assertComponentUsable 100.00% 5 / 5 100.00% 1 / 1 5
 sendUpsert 81.81% 9 / 11 0.00% 0 / 1 3.05
 throwIfAppCaused 75.00% 3 / 4 0.00% 0 / 1 4.25
 handleUpdateFailure 66.66% 6 / 9 0.00% 0 / 1 3.33
 reportIfSessionLost 100.00% 2 / 2 100.00% 1 / 1 2
 orphanExposures 100.00% 7 / 7 100.00% 1 / 1 6
 sendUnavailable 33.33% 3 / 9 0.00% 0 / 1 8.74
 sendCommandAnswer 50.00% 3 / 6 0.00% 0 / 1 4.12
 syncOwner 100.00% 2 / 2 100.00% 1 / 1 2
 findLive 100.00% 1 / 1 100.00% 1 / 1 1
 createAddress 100.00% 1 / 1 100.00% 1 / 1 1
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Broker\Exposure;
6
7use Closure;
8use Psr\Log\LoggerInterface;
9use Stewart\Client\Component\Collection\ExposedEntityAddressCollection;
10use Stewart\Client\Component\ComponentCommand;
11use Stewart\Client\Component\ComponentCommandAnswer;
12use Stewart\Client\Component\ComponentErrorCode;
13use Stewart\Client\Component\ExposedEntityAddress;
14use Stewart\Client\Component\ExposedEntityDefinition;
15use Stewart\Client\Exception\HaClientException;
16use Stewart\Client\HaClient;
17use Stewart\Contracts\App\AppId;
18use Stewart\Contracts\Entity\EntityId;
19use Stewart\Contracts\Exception\ExposureError;
20use Stewart\Contracts\Exception\ExposureException;
21use Stewart\Contracts\Exposure\ExposedEntityKey;
22use Stewart\Contracts\Exposure\ExposedEntitySnapshot;
23use Stewart\Contracts\Exposure\ExposedStateChange;
24use Stewart\Runtime\Broker\Component\ComponentTracker;
25use Stewart\Runtime\Config\ExposeConfig;
26use Stewart\Runtime\Lifecycle\ComponentState;
27use Stewart\Runtime\Model\WorkerId;
28
29use function Amp\async;
30
31// Keeps every exposure for the next session; a gone app's entities stay, unavailable, until it exposes them again.
32final class ExposureLink
33{
34    /** @var array<string, LiveExposure> */
35    private array $exposuresByAddress = [];
36
37    /** @var Closure(ExposedEntitySync): void */
38    private Closure $onSynced;
39
40    /** @var Closure(ExposedEntityCommand): void */
41    private Closure $onCommanded;
42
43    /** @var Closure(): void */
44    private Closure $onSessionLost;
45
46    public function __construct(
47        private readonly HaClient $client,
48        private readonly LoggerInterface $logger,
49        private readonly ComponentTracker $tracker,
50        private readonly ExposeConfig $expose,
51    ) {
52        $this->onSynced = static function (): void {};
53        $this->onCommanded = static function (): void {};
54        $this->onSessionLost = static function (): void {};
55    }
56
57    /** @param Closure(ExposedEntitySync): void $handler */
58    public function onEntitySynced(Closure $handler): void
59    {
60        $this->onSynced = $handler;
61    }
62
63    /** @param Closure(ExposedEntityCommand): void $handler */
64    public function onEntityCommanded(Closure $handler): void
65    {
66        $this->onCommanded = $handler;
67    }
68
69    /** @param Closure(): void $handler */
70    public function onSessionLost(Closure $handler): void
71    {
72        $this->onSessionLost = $handler;
73    }
74
75    public function receiveCommand(ComponentCommand $command): void
76    {
77        $live = $this->findLive($command->appId, $command->key);
78
79        if ($live === null) {
80            $reason = \sprintf('Entity %s of app %s is not exposed by a running app.', $command->key, $command->appId);
81            $this->sendCommandAnswer(ComponentCommandAnswer::reject($command->commandId, $reason));
82
83            return;
84        }
85
86        $exposedCommand = $command->readExposedCommand($live->definition->platform);
87
88        if ($exposedCommand === null) {
89            $reason = \sprintf('Entity %s of app %s does not take %s commands.', $command->key, $command->appId, $command->action->value);
90            $this->sendCommandAnswer(ComponentCommandAnswer::reject($command->commandId, $reason));
91
92            return;
93        }
94
95        ($this->onCommanded)(new ExposedEntityCommand($command->commandId, $live->owner, $command->appId, $command->key, $exposedCommand));
96    }
97
98    public function acceptCommand(ExposedEntityCommand $command): void
99    {
100        $requestedState = $command->command->getRequestedState();
101
102        if ($requestedState !== null) {
103            $this->findLive($command->appId, $command->key)?->recordChange(new ExposedStateChange($requestedState));
104        }
105
106        $this->sendCommandAnswer(ComponentCommandAnswer::accept($command->commandId));
107    }
108
109    public function rejectCommand(ExposedEntityCommand $command, string $reason): void
110    {
111        $this->sendCommandAnswer(ComponentCommandAnswer::reject($command->commandId, $reason));
112    }
113
114    /** @throws ExposureException */
115    public function exposeEntity(
116        WorkerId $owner,
117        AppId $appId,
118        ExposedEntityKey $key,
119        ExposedEntityDefinition $definition,
120        ExposedStateChange $change,
121    ): ?ExposedEntitySnapshot {
122        $this->assertComponentUsable($key);
123
124        $live = new LiveExposure($owner, $this->createAddress($appId, $key), $definition, new ExposedStateChange(available: true)->withLaterChange($change));
125        $this->exposuresByAddress[$live->address->describe()] = $live;
126
127        if (!$this->tracker->isActive()) {
128            return null;
129        }
130
131        try {
132            return $this->sendUpsert($live, $live->definition, $live->buildUpsertChange());
133        } catch (ExposureException $e) {
134            unset($this->exposuresByAddress[$live->address->describe()]);
135
136            throw $e;
137        }
138    }
139
140    /** @throws ExposureException */
141    public function reconfigureEntity(AppId $appId, ExposedEntityKey $key, ExposedEntityDefinition $definition): ?ExposedEntitySnapshot
142    {
143        $live = $this->findLive($appId, $key) ?? throw ExposureException::removed($key);
144        $definition = $live->definition->withConfigOf($definition);
145
146        if (!$this->tracker->isActive()) {
147            $live->recordDefinition($definition);
148
149            return null;
150        }
151
152        $snapshot = $this->sendUpsert($live, $definition, new ExposedStateChange());
153        $live->recordDefinition($definition);
154
155        if ($snapshot !== null) {
156            $live->recordChange(new ExposedStateChange($snapshot->state));
157        }
158
159        return $snapshot;
160    }
161
162    /** @throws ExposureException */
163    public function updateEntity(AppId $appId, ExposedEntityKey $key, ExposedStateChange $change): void
164    {
165        $live = $this->findLive($appId, $key) ?? throw ExposureException::removed($key);
166
167        if (!$this->tracker->isActive()) {
168            $live->recordChange($change);
169
170            return;
171        }
172
173        try {
174            $this->client->updateExposedEntityState($live->address, $change);
175        } catch (HaClientException $e) {
176            $this->handleUpdateFailure($live, $change, $e);
177
178            return;
179        }
180
181        $live->recordChange($change);
182    }
183
184    public function removeEntity(AppId $appId, ExposedEntityKey $key): void
185    {
186        $live = $this->findLive($appId, $key);
187
188        if ($live === null) {
189            return;
190        }
191
192        unset($this->exposuresByAddress[$live->address->describe()]);
193
194        try {
195            $this->client->removeExposedEntity($live->address);
196        } catch (HaClientException $e) {
197            $this->logger->warning('Could not remove an exposed entity from Home Assistant; it stays there until removed by hand', [
198                'entity' => $live->address->describe(),
199                'exception' => $e,
200            ]);
201            $this->reportIfSessionLost($e);
202        }
203    }
204
205    public function orphanExposuresOfApp(AppId $appId): void
206    {
207        $this->orphanExposures(static fn(LiveExposure $live): bool => $live->address->appId->equals($appId));
208    }
209
210    public function orphanExposuresOfWorker(WorkerId $workerId): void
211    {
212        $this->orphanExposures(static fn(LiveExposure $live): bool => $live->isOwnedBy($workerId));
213    }
214
215    public function countExposuresOfApp(AppId $appId): int
216    {
217        return \count(array_filter($this->exposuresByAddress, static fn(LiveExposure $live): bool => $live->address->appId->equals($appId)));
218    }
219
220    public function findAddressByEntityId(EntityId $entityId): ?ExposedEntityAddress
221    {
222        $live = array_find($this->exposuresByAddress, static fn(LiveExposure $live): bool => $live->entityId?->equals($entityId) === true);
223
224        return $live?->address;
225    }
226
227    public function listKeptAddresses(): ExposedEntityAddressCollection
228    {
229        return ExposedEntityAddressCollection::fromList(array_map(static fn(LiveExposure $live) => $live->address, array_values($this->exposuresByAddress)));
230    }
231
232    /** @throws HaClientException */
233    public function replayAll(): void
234    {
235        if (!$this->tracker->isActive()) {
236            return;
237        }
238
239        foreach ($this->exposuresByAddress as $live) {
240            try {
241                $snapshot = $this->sendUpsert($live, $live->definition, $live->buildUpsertChange());
242            } catch (ExposureException $e) {
243                $this->logger->error('Home Assistant refused an exposed entity again', ['entity' => $live->address->describe(), 'exception' => $e]);
244
245                continue;
246            }
247
248            if ($snapshot !== null && $this->findLive($live->address->appId, $live->address->key) === $live) {
249                $this->syncOwner($live, $snapshot);
250            }
251        }
252    }
253
254    /** @throws ExposureException */
255    private function assertComponentUsable(ExposedEntityKey $key): void
256    {
257        match ($this->tracker->detection->state) {
258            ComponentState::Missing => throw ExposureException::createForReason(ExposureError::ComponentMissing, key: $key->value),
259            ComponentState::ProtocolMismatch => throw ExposureException::createForReason(ExposureError::ProtocolMismatch, key: $key->value),
260            ComponentState::Refused => throw ExposureException::createForReason(ExposureError::ComponentRefused, key: $key->value),
261            ComponentState::Unchecked, ComponentState::Replaced, ComponentState::Active => null,
262        };
263    }
264
265    /** @throws ExposureException */
266    private function sendUpsert(LiveExposure $live, ExposedEntityDefinition $definition, ExposedStateChange $change): ?ExposedEntitySnapshot
267    {
268        try {
269            if ($definition->isAction()) {
270                $this->client->upsertExposedAction($live->address, $definition);
271
272                return null;
273            }
274
275            $snapshot = $this->client->upsertExposedEntity($live->address, $definition, $change);
276        } catch (HaClientException $e) {
277            $this->throwIfAppCaused($e);
278            $this->logger->debug('Exposed entity waits for the next stewart integration session', ['entity' => $live->address->describe(), 'exception' => $e]);
279            $this->reportIfSessionLost($e);
280
281            return null;
282        }
283
284        $live->recordEntityId($snapshot->entityId);
285
286        return $snapshot;
287    }
288
289    /** @throws ExposureException */
290    private function throwIfAppCaused(HaClientException $e): void
291    {
292        match (ComponentErrorCode::tryFromException($e)) {
293            ComponentErrorCode::InvalidConfig, ComponentErrorCode::ActionTaken => throw ExposureException::configInvalid($e->findDetail() ?? $e->getMessage()),
294            ComponentErrorCode::InvalidState => throw ExposureException::stateInvalid($e->findDetail() ?? $e->getMessage()),
295            default => null,
296        };
297    }
298
299    /** @throws ExposureException */
300    private function handleUpdateFailure(LiveExposure $live, ExposedStateChange $change, HaClientException $e): void
301    {
302        $this->throwIfAppCaused($e);
303        $live->recordChange($change);
304
305        if (ComponentErrorCode::tryFromException($e) !== ComponentErrorCode::NotFound) {
306            $this->logger->debug('Exposed entity state waits for the next stewart integration session', ['entity' => $live->address->describe(), 'exception' => $e]);
307            $this->reportIfSessionLost($e);
308
309            return;
310        }
311
312        $snapshot = $this->sendUpsert($live, $live->definition, $live->buildUpsertChange());
313
314        if ($snapshot !== null) {
315            $this->syncOwner($live, $snapshot);
316        }
317    }
318
319    private function reportIfSessionLost(HaClientException $e): void
320    {
321        if (ComponentErrorCode::tryFromException($e) === ComponentErrorCode::NoSession) {
322            ($this->onSessionLost)();
323        }
324    }
325
326    /** @param Closure(LiveExposure): bool $belongs */
327    private function orphanExposures(Closure $belongs): void
328    {
329        $orphaned = [];
330
331        foreach ($this->exposuresByAddress as $live) {
332            if (!$live->isOrphaned() && $belongs($live)) {
333                $live->markOrphaned();
334                $orphaned[] = $live;
335            }
336        }
337
338        if ($orphaned !== [] && $this->tracker->isActive()) {
339            async($this->sendUnavailable(...), $orphaned);
340        }
341    }
342
343    /** @param list<LiveExposure> $orphaned */
344    private function sendUnavailable(array $orphaned): void
345    {
346        foreach ($orphaned as $live) {
347            // An action keeps its service; the broker refuses its calls while it is orphaned.
348            if ($live->definition->isAction()) {
349                continue;
350            }
351
352            try {
353                $this->client->updateExposedEntityState($live->address, new ExposedStateChange(available: false));
354            } catch (HaClientException $e) {
355                $this->logger->debug('Orphaned exposed entity turns unavailable with the next stewart integration session', [
356                    'entity' => $live->address->describe(),
357                    'exception' => $e,
358                ]);
359                $this->reportIfSessionLost($e);
360            }
361        }
362    }
363
364    private function sendCommandAnswer(ComponentCommandAnswer $answer): void
365    {
366        async(function () use ($answer): void {
367            try {
368                if (!$this->client->answerComponentCommand($answer)) {
369                    $this->logger->debug('Home Assistant stopped waiting for an exposed entity command before it was answered', ['command' => $answer->commandId]);
370                }
371            } catch (HaClientException $e) {
372                $this->logger->debug('Could not answer an exposed entity command; Home Assistant times it out', ['command' => $answer->commandId, 'exception' => $e]);
373            }
374        })->ignore();
375    }
376
377    private function syncOwner(LiveExposure $live, ExposedEntitySnapshot $snapshot): void
378    {
379        if ($live->owner !== null) {
380            ($this->onSynced)(new ExposedEntitySync($live->owner, $live->address->appId, $live->address->key, $snapshot));
381        }
382    }
383
384    private function findLive(AppId $appId, ExposedEntityKey $key): ?LiveExposure
385    {
386        return $this->exposuresByAddress[$this->createAddress($appId, $key)->describe()] ?? null;
387    }
388
389    private function createAddress(AppId $appId, ExposedEntityKey $key): ExposedEntityAddress
390    {
391        return new ExposedEntityAddress($this->expose->instance, $appId, $key);
392    }
393}