Lines 80.00% 48 / 60
Functions and Methods 50.00% 5 / 10
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
HaTriggerLink 80.00% 48 / 60 50.00% 5 / 10 28.61 0.00% 0 / 1
 __construct 100.00% 2 / 2 100.00% 1 / 1 1
 onTriggerFired 100.00% 1 / 1 100.00% 1 / 1 1
 onTriggerRejected 100.00% 1 / 1 100.00% 1 / 1 1
 subscribeTrigger 62.50% 10 / 16 0.00% 0 / 1 4.84
 unsubscribeTrigger 83.33% 5 / 6 0.00% 0 / 1 3.04
 resubscribeAll 100.00% 4 / 4 100.00% 1 / 1 2
 markLinkLost 100.00% 3 / 3 100.00% 1 / 1 2
 sendSubscribe 89.47% 17 / 19 0.00% 0 / 1 6.04
 deliverFired 66.66% 2 / 3 0.00% 0 / 1 2.15
 sendUnsubscribeInBackground 60.00% 3 / 5 0.00% 0 / 1 2.26
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Broker\Trigger;
6
7use Closure;
8use Psr\Log\LoggerInterface;
9use Stewart\Client\Exception\HaClientException;
10use Stewart\Client\HaClient;
11use Stewart\Contracts\Trigger\TriggerEvent;
12use Stewart\Support\Wire\TriggerSpec;
13use Throwable;
14
15use function Amp\async;
16
17// One Home Assistant subscription per spec; ids are per connection, so every connect re-issues them all.
18final class HaTriggerLink
19{
20    /** @var array<string, LiveTrigger> */
21    private array $triggersByKey = [];
22
23    private bool $linkReady = false;
24
25    private int $connectionGeneration = 0;
26
27    /** @var Closure(TriggerSpec, TriggerEvent): void */
28    private Closure $onFired;
29
30    /** @var Closure(TriggerSpec, string): void */
31    private Closure $onRejected;
32
33    public function __construct(
34        private readonly HaClient $client,
35        private readonly LoggerInterface $logger,
36    ) {
37        $this->onFired = static function (): void {};
38        $this->onRejected = static function (): void {};
39    }
40
41    /** @param Closure(TriggerSpec, TriggerEvent): void $handler */
42    public function onTriggerFired(Closure $handler): void
43    {
44        $this->onFired = $handler;
45    }
46
47    /** @param Closure(TriggerSpec, string): void $handler */
48    public function onTriggerRejected(Closure $handler): void
49    {
50        $this->onRejected = $handler;
51    }
52
53    public function subscribeTrigger(TriggerSpec $spec): void
54    {
55        $key = $spec->getSharingKey();
56
57        if (isset($this->triggersByKey[$key])) {
58            return;
59        }
60
61        $live = new LiveTrigger($spec);
62        $this->triggersByKey[$key] = $live;
63
64        if (!$this->linkReady) {
65            return;
66        }
67
68        $generation = $this->connectionGeneration;
69
70        async(function () use ($live, $generation): void {
71            try {
72                $this->sendSubscribe($live, $generation);
73            } catch (Throwable $e) {
74                $this->logger->warning('Home Assistant trigger subscription failed; it is retried after the next reconnect', [
75                    'platforms' => $live->spec->listPlatforms(),
76                    'exception' => $e,
77                ]);
78            }
79        })->ignore();
80    }
81
82    public function unsubscribeTrigger(TriggerSpec $spec): void
83    {
84        $key = $spec->getSharingKey();
85        $live = $this->triggersByKey[$key] ?? null;
86        unset($this->triggersByKey[$key]);
87
88        if ($live?->haSubscriptionId === null || !$this->linkReady) {
89            return;
90        }
91
92        $this->sendUnsubscribeInBackground($live->haSubscriptionId);
93    }
94
95    /** @throws HaClientException */
96    public function resubscribeAll(): void
97    {
98        $generation = ++$this->connectionGeneration;
99        $this->linkReady = true;
100
101        foreach ($this->triggersByKey as $live) {
102            $this->sendSubscribe($live, $generation);
103        }
104    }
105
106    public function markLinkLost(): void
107    {
108        $this->linkReady = false;
109
110        foreach ($this->triggersByKey as $live) {
111            $live->forgetHaSubscriptionId();
112        }
113    }
114
115    /** @throws HaClientException */
116    private function sendSubscribe(LiveTrigger $live, int $generation): void
117    {
118        $key = $live->spec->getSharingKey();
119
120        try {
121            $haSubscriptionId = $this->client->subscribeTrigger(
122                $live->spec->triggers,
123                $live->spec->variables,
124                fn(TriggerEvent $event) => $this->deliverFired($live, $event),
125            );
126        } catch (HaClientException $e) {
127            if (!$e->reason->isCommandRefusal()) {
128                throw $e;
129            }
130
131            if (($this->triggersByKey[$key] ?? null) === $live) {
132                unset($this->triggersByKey[$key]);
133                ($this->onRejected)($live->spec, $e->getMessage());
134            }
135
136            return;
137        }
138
139        if ($generation !== $this->connectionGeneration) {
140            return;
141        }
142
143        if (($this->triggersByKey[$key] ?? null) !== $live) {
144            $this->sendUnsubscribeInBackground($haSubscriptionId);
145
146            return;
147        }
148
149        $live->recordHaSubscriptionId($haSubscriptionId);
150    }
151
152    private function deliverFired(LiveTrigger $live, TriggerEvent $event): void
153    {
154        if (($this->triggersByKey[$live->spec->getSharingKey()] ?? null) !== $live) {
155            return;
156        }
157
158        ($this->onFired)($live->spec, $event);
159    }
160
161    private function sendUnsubscribeInBackground(int $haSubscriptionId): void
162    {
163        async(function () use ($haSubscriptionId): void {
164            try {
165                $this->client->unsubscribeEvents($haSubscriptionId);
166            } catch (Throwable $e) {
167                $this->logger->debug('Home Assistant trigger unsubscribe failed', ['subscription' => $haSubscriptionId, 'exception' => $e]);
168            }
169        })->ignore();
170    }
171}