1import type {
2 DesktopTelemetryRequestDesktopUpdate,
3 DesktopUpdateRemoteOutcome,
4 DesktopUpdateState,
5} from "@t3tools/contracts";
6import * as Duration from "effect/Duration";
7import * as Clock from "effect/Clock";
8import * as Deferred from "effect/Deferred";
9import * as Effect from "effect/Effect";
10import * as Option from "effect/Option";
11import * as Queue from "effect/Queue";
12import * as Ref from "effect/Ref";
13import type * as Scope from "effect/Scope";
14import * as Stream from "effect/Stream";
15
16import * as DesktopObservability from "../app/DesktopObservability.ts";
17import * as DesktopTelemetryPublisher from "../telemetry/DesktopTelemetryPublisher.ts";
18import * as DesktopUpdates from "./DesktopUpdates.ts";
19import {
20 nextRemoteDesktopUpdateStep,
21 normalizeRemoteUpdateReason,
22 type RemoteDesktopUpdateAttempts,
23} from "./remoteUpdateFlow.ts";
24
25const { logInfo, logError } = DesktopObservability.makeComponentLogger("desktop-remote-updates");
26
27/** Pause before retrying an action the updater refused for a held reservation. */
28const ACTION_RETRY_DELAY = Duration.millis(250);
29const PREPARED_UPDATE_TTL = Duration.minutes(5);
30
31interface PreparedUpdate {
32 readonly requestId: string;
33 readonly downloadedVersion: string;
34 readonly status: "prepared" | "committing" | "failed";
35 readonly failureReason?: string;
36 readonly expiresAt: number;
37}
38
39type CommitClaim =
40 | { readonly _tag: "invalid" }
41 | { readonly _tag: "failed"; readonly prepared: PreparedUpdate }
42 | { readonly _tag: "committing" }
43 | { readonly _tag: "claimed"; readonly prepared: PreparedUpdate };
44
45/**
46 * Server-triggered desktop updates. Mirrors updater state, prepares downloads,
47 * cancels abandoned preparations, and commits installs after the remote client
48 * confirms that it received the preparation token.
49 */
50export const listen: Effect.Effect<
51 void,
52 never,
53 DesktopUpdates.DesktopUpdates | DesktopTelemetryPublisher.DesktopTelemetryPublisher | Scope.Scope
54> = Effect.gen(function* () {
55 const updates = yield* DesktopUpdates.DesktopUpdates;
56 const publisher = yield* DesktopTelemetryPublisher.DesktopTelemetryPublisher;
57 const activeRequestIdRef = yield* Ref.make(Option.none<string>());
58 const requestControlRef = yield* Ref.make<{
59 readonly active: Option.Option<{
60 readonly requestId: string;
61 readonly signal: Deferred.Deferred<void>;
62 }>;
63 readonly cancelledBeforeStart: ReadonlyArray<string>;
64 }>({ active: Option.none(), cancelledBeforeStart: [] });
65 const preparedUpdateRef = yield* Ref.make(Option.none<PreparedUpdate>());
66
67 const publishReport = (
68 state: DesktopUpdateState,
69 terminal?: { readonly outcome: DesktopUpdateRemoteOutcome; readonly reason?: string },
70 explicitRequestId?: string,
71 ): Effect.Effect<void> => {
72 const reason = normalizeRemoteUpdateReason(terminal?.reason);
73 return Ref.get(activeRequestIdRef).pipe(
74 Effect.flatMap((requestId) =>
75 publisher.publishUpdateReport({
76 version: 1,
77 type: "desktopUpdateStatus",
78 ...(explicitRequestId !== undefined
79 ? { requestId: explicitRequestId }
80 : Option.isSome(requestId)
81 ? { requestId: requestId.value }
82 : {}),
83 ...(terminal === undefined ? {} : { outcome: terminal.outcome }),
84 ...(reason ? { reason } : {}),
85 state,
86 }),
87 ),
88 );
89 };
90
91 const recordPreparedFailure = (requestId: string, reason: string) =>
92 Ref.modify(preparedUpdateRef, (prepared) => {
93 if (
94 Option.isNone(prepared) ||
95 prepared.value.requestId !== requestId ||
96 prepared.value.status === "failed"
97 ) {
98 return [false, prepared] as const;
99 }
100 return [
101 true,
102 Option.some({ ...prepared.value, status: "failed" as const, failureReason: reason }),
103 ] as const;
104 });
105
106 const clearActiveRequest = (requestId: string) =>
107 Ref.update(activeRequestIdRef, (active) =>
108 Option.isSome(active) && active.value === requestId ? Option.none() : active,
109 );
110
111 yield* Effect.scoped(
112 Effect.gen(function* () {
113 const { latest, changes } = yield* updates.subscribe;
114 yield* publishReport(latest);
115 yield* Stream.runForEach(changes, (state) =>
116 Effect.gen(function* () {
117 yield* publishReport(state);
118 const prepared = yield* Ref.get(preparedUpdateRef);
119 if (
120 Option.isSome(prepared) &&
121 prepared.value.status === "committing" &&
122 state.errorContext === "install"
123 ) {
124 const reason = state.message ?? "The desktop app failed to install the update.";
125 if (yield* recordPreparedFailure(prepared.value.requestId, reason)) {
126 yield* publishReport(state, { outcome: "failed", reason }, prepared.value.requestId);
127 }
128 }
129 }),
130 );
131 }),
132 ).pipe(Effect.forkScoped);
133
134 const handleRequest = (request: DesktopTelemetryRequestDesktopUpdate): Effect.Effect<void> =>
135 Effect.scoped(
136 Effect.gen(function* () {
137 const cancellation = yield* Deferred.make<void>();
138 const shouldStart = yield* Ref.modify(requestControlRef, (control) => {
139 if (control.cancelledBeforeStart.includes(request.requestId)) {
140 return [
141 false,
142 {
143 ...control,
144 cancelledBeforeStart: control.cancelledBeforeStart.filter(
145 (requestId) => requestId !== request.requestId,
146 ),
147 },
148 ] as const;
149 }
150 return [
151 true,
152 {
153 ...control,
154 active: Option.some({ requestId: request.requestId, signal: cancellation }),
155 },
156 ] as const;
157 });
158 if (!shouldStart) return;
159
160 const now = yield* Clock.currentTimeMillis;
161 const prepared = yield* Ref.modify(preparedUpdateRef, (current) =>
162 Option.isSome(current) &&
163 current.value.status === "prepared" &&
164 current.value.expiresAt <= now
165 ? ([Option.none<typeof current.value>(), Option.none()] as const)
166 : ([current, current] as const),
167 );
168 if (Option.isSome(prepared) && prepared.value.status !== "failed") {
169 yield* publishReport(
170 yield* updates.getState,
171 {
172 outcome: "failed",
173 reason: "A prepared desktop update is already in progress.",
174 },
175 request.requestId,
176 );
177 return;
178 }
179 yield* Ref.set(activeRequestIdRef, Option.some(request.requestId));
180 yield* logInfo("remote update requested", { requestId: request.requestId });
181 const { latest, changes } = yield* updates.subscribe;
182 const disabledReason = Option.getOrNull(yield* updates.disabledReason);
183 let attempts: RemoteDesktopUpdateAttempts = { checks: 0, downloads: 0 };
184 // The updater admits one action at a time. A state event can land
185 // while the action that produced it still holds the reservation
186 // (e.g. "available" before the check releases), so a forked action
187 // can be refused with no later state event to retry on. Rejected
188 // actions re-enqueue their state here after a short pause so the
189 // step runs again once the reservation is free.
190 const retries = yield* Queue.unbounded<DesktopUpdateState>();
191 const retryLater = (state: DesktopUpdateState) =>
192 Effect.sleep(ACTION_RETRY_DELAY).pipe(
193 Effect.andThen(Queue.offer(retries, state)),
194 Effect.asVoid,
195 Effect.forkScoped,
196 );
197
198 // Returns true when the run reached a terminal outcome.
199 const step = (state: DesktopUpdateState): Effect.Effect<boolean, never, Scope.Scope> =>
200 Effect.gen(function* () {
201 const next = nextRemoteDesktopUpdateStep(state, attempts, disabledReason);
202 switch (next.action) {
203 case "wait":
204 return false;
205 // Counters increment before the action so the state event the
206 // action produces already sees it; a refusal for a held
207 // reservation rolls the count back, since it was not a try.
208 case "check":
209 attempts = { ...attempts, checks: attempts.checks + 1 };
210 yield* updates.check("remote-update").pipe(
211 Effect.flatMap((result) => {
212 if (result.checked) return Effect.void;
213 attempts = { ...attempts, checks: attempts.checks - 1 };
214 return retryLater(state);
215 }),
216 Effect.forkScoped,
217 );
218 return false;
219 case "download":
220 attempts = { ...attempts, downloads: attempts.downloads + 1 };
221 yield* updates.download.pipe(
222 Effect.flatMap((result) => {
223 if (result.accepted) return Effect.void;
224 attempts = { ...attempts, downloads: attempts.downloads - 1 };
225 return retryLater(state);
226 }),
227 Effect.forkScoped,
228 );
229 return false;
230 case "install": {
231 if (state.installKind === "ui") {
232 yield* publishReport(state, { outcome: "up-to-date" }, request.requestId);
233 return true;
234 }
235 // The download event fires before its action releases the
236 // updater reservation. Wait until the prepared install can
237 // be committed by the client in a separate RPC.
238 if (yield* updates.isActionActive) {
239 yield* retryLater(state);
240 return false;
241 }
242 if (state.downloadedVersion === null) {
243 yield* publishReport(
244 state,
245 {
246 outcome: "failed",
247 reason: "The desktop app lost the downloaded update.",
248 },
249 request.requestId,
250 );
251 return true;
252 }
253 yield* Ref.set(
254 preparedUpdateRef,
255 Option.some({
256 requestId: request.requestId,
257 downloadedVersion: state.downloadedVersion,
258 status: "prepared",
259 expiresAt:
260 (yield* Clock.currentTimeMillis) + Duration.toMillis(PREPARED_UPDATE_TTL),
261 }),
262 );
263 yield* publishReport(state, { outcome: "ready-to-install" }, request.requestId);
264 yield* logInfo("remote update prepared", { requestId: request.requestId });
265 return true;
266 }
267 case "done":
268 yield* publishReport(
269 state,
270 {
271 outcome: next.outcome,
272 ...(next.reason === undefined ? {} : { reason: next.reason }),
273 },
274 request.requestId,
275 );
276 yield* logInfo("remote update finished", {
277 requestId: request.requestId,
278 outcome: next.outcome,
279 reason: next.reason ?? null,
280 });
281 return true;
282 }
283 });
284
285 yield* Effect.raceFirst(
286 Effect.gen(function* () {
287 if (yield* step(latest)) return;
288 yield* Stream.merge(changes, Stream.fromQueue(retries)).pipe(
289 Stream.mapEffect(step),
290 Stream.takeUntil((done) => done),
291 Stream.runDrain,
292 );
293 }),
294 Deferred.await(cancellation),
295 );
296 }),
297 ).pipe(
298 Effect.ensuring(
299 Effect.all(
300 [
301 clearActiveRequest(request.requestId),
302 Ref.update(requestControlRef, (control) => ({
303 ...control,
304 active:
305 Option.isSome(control.active) &&
306 control.active.value.requestId === request.requestId
307 ? Option.none()
308 : control.active,
309 })),
310 ],
311 { discard: true },
312 ),
313 ),
314 Effect.catchCause((cause) =>
315 logError("remote update request failed unexpectedly", {
316 requestId: request.requestId,
317 cause: String(cause),
318 }),
319 ),
320 );
321
322 // Sequential by construction: a second remote request queued mid-run is
323 // handled after the current one, when the state machine resolves it fast.
324 yield* Stream.runForEach(publisher.updateRequests, handleRequest).pipe(Effect.forkScoped);
325
326 yield* Stream.runForEach(publisher.updateCancellations, (cancellation) =>
327 Effect.gen(function* () {
328 const activeSignal = yield* Ref.modify(requestControlRef, (control) => {
329 if (
330 Option.isSome(control.active) &&
331 control.active.value.requestId === cancellation.requestId
332 ) {
333 return [Option.some(control.active.value.signal), control] as const;
334 }
335 return [
336 Option.none<Deferred.Deferred<void>>(),
337 {
338 ...control,
339 cancelledBeforeStart: [
340 ...control.cancelledBeforeStart.filter(
341 (requestId) => requestId !== cancellation.requestId,
342 ),
343 cancellation.requestId,
344 ],
345 },
346 ] as const;
347 });
348 if (Option.isSome(activeSignal)) {
349 yield* Deferred.succeed(activeSignal.value, undefined);
350 }
351 const matchedPrepared = yield* Ref.modify(preparedUpdateRef, (prepared) => {
352 if (Option.isNone(prepared) || prepared.value.requestId !== cancellation.requestId) {
353 return [false, prepared] as const;
354 }
355 return [true, prepared.value.status === "prepared" ? Option.none() : prepared] as const;
356 });
357 if (matchedPrepared) {
358 yield* Ref.update(requestControlRef, (control) => {
359 return {
360 ...control,
361 cancelledBeforeStart: control.cancelledBeforeStart.filter(
362 (requestId) => requestId !== cancellation.requestId,
363 ),
364 };
365 });
366 }
367 }),
368 ).pipe(Effect.forkScoped);
369
370 yield* Stream.runForEach(publisher.updateCommits, (commit) =>
371 Effect.gen(function* () {
372 const current = yield* updates.getState;
373 const now = yield* Clock.currentTimeMillis;
374 const claim = yield* Ref.modify(
375 preparedUpdateRef,
376 (prepared): readonly [CommitClaim, Option.Option<PreparedUpdate>] => {
377 if (
378 Option.isNone(prepared) ||
379 prepared.value.requestId !== commit.requestId ||
380 (prepared.value.status === "prepared" && prepared.value.expiresAt <= now)
381 ) {
382 return [{ _tag: "invalid" as const }, prepared] as const;
383 }
384 if (prepared.value.status === "failed") {
385 return [{ _tag: "failed" as const, prepared: prepared.value }, prepared] as const;
386 }
387 if (prepared.value.status === "committing") {
388 return [{ _tag: "committing" as const }, prepared] as const;
389 }
390 return [
391 { _tag: "claimed" as const, prepared: prepared.value },
392 Option.some({ ...prepared.value, status: "committing" as const }),
393 ] as const;
394 },
395 );
396 if (claim._tag === "invalid") {
397 yield* publishReport(
398 yield* updates.getState,
399 {
400 outcome: "failed",
401 reason: "This desktop update is no longer prepared.",
402 },
403 commit.requestId,
404 );
405 return;
406 }
407 if (claim._tag === "failed") {
408 yield* publishReport(
409 current,
410 {
411 outcome: "failed",
412 reason: claim.prepared.failureReason ?? "The desktop app failed to install the update.",
413 },
414 commit.requestId,
415 );
416 return;
417 }
418 if (claim._tag === "committing") {
419 return;
420 }
421 yield* Ref.set(activeRequestIdRef, Option.some(commit.requestId));
422 if (current.downloadedVersion !== claim.prepared.downloadedVersion) {
423 const reason = "This desktop update is no longer prepared.";
424 if (yield* recordPreparedFailure(commit.requestId, reason)) {
425 yield* publishReport(current, { outcome: "failed", reason }, commit.requestId);
426 }
427 return;
428 }
429 const result = yield* updates.installPrepared(claim.prepared.downloadedVersion);
430 if (
431 !result.accepted &&
432 result.state.downloadedVersion === claim.prepared.downloadedVersion &&
433 (yield* updates.isInstallActive)
434 ) {
435 // Another install (local, or an earlier remote request) already owns
436 // the shutdown and will relaunch the app on the same downloaded
437 // version. This commit joins it: no failure marker, and the client
438 // proves the handoff the same way, by transport loss then the
439 // target version on reconnect.
440 yield* logInfo("remote update commit joining an in-progress install", {
441 requestId: commit.requestId,
442 });
443 return;
444 }
445 if (!result.accepted || result.failed) {
446 const reason = result.state.message ?? "The desktop app could not start the install.";
447 if (yield* recordPreparedFailure(commit.requestId, reason)) {
448 yield* publishReport(result.state, { outcome: "failed", reason }, commit.requestId);
449 }
450 return;
451 }
452 // A successful install tears down this backend. Do not send a success
453 // marker from the old process: transport loss followed by the target
454 // version is the only proof that the handoff succeeded.
455 }).pipe(
456 Effect.ensuring(clearActiveRequest(commit.requestId)),
457 Effect.catchCause((cause) =>
458 logError("remote update commit failed unexpectedly", {
459 requestId: commit.requestId,
460 cause: String(cause),
461 }),
462 ),
463 ),
464 ).pipe(Effect.forkScoped);
465});