1import {
2 ServerSelfUpdateError,
3 type DesktopUpdateState,
4 type DesktopUpdateStatusReport,
5 type ServerSelfUpdateProgressStage,
6 type ServerSelfUpdateResult,
7} from "@t3tools/contracts";
8import * as Context from "effect/Context";
9import * as Crypto from "effect/Crypto";
10import * as Duration from "effect/Duration";
11import * as Effect from "effect/Effect";
12import * as Layer from "effect/Layer";
13import * as Option from "effect/Option";
14import * as Ref from "effect/Ref";
15import * as Result from "effect/Result";
16import * as Stream from "effect/Stream";
17
18import * as ServerConfig from "../config.ts";
19import * as DesktopTelemetryReceiver from "../resourceTelemetry/DesktopTelemetryReceiver.ts";
20
21/** Backstop for a desktop updater that hangs without ever reporting a
22 terminal outcome. Generous: it covers a slow download of a full build. */
23const DESKTOP_UPDATE_TIMEOUT = Duration.minutes(20);
24const DESKTOP_INSTALL_TIMEOUT = Duration.minutes(2);
25
26/** Progress stage a desktop update state maps to, or null when the state
27 carries no progress worth streaming. */
28function desktopUpdateProgressStage(
29 state: DesktopUpdateState,
30): ServerSelfUpdateProgressStage | null {
31 switch (state.status) {
32 case "checking":
33 case "available":
34 case "downloading":
35 return "downloading";
36 case "downloaded":
37 return "installing";
38 default:
39 return null;
40 }
41}
42
43export class DesktopAppUpdate extends Context.Service<
44 DesktopAppUpdate,
45 {
46 /** True when this server was spawned by a desktop app that can be
47 driven over the telemetry control channel. */
48 readonly available: boolean;
49 /** Checks and downloads through the desktop app, then returns a token
50 while this server is still connected. `commit` starts installation. */
51 readonly run: (
52 reportProgress: (
53 stage: ServerSelfUpdateProgressStage,
54 ) => Effect.Effect<void, ServerSelfUpdateError>,
55 ) => Effect.Effect<ServerSelfUpdateResult, ServerSelfUpdateError>;
56 /** Starts the prepared install. Success stops this server, so this effect
57 returns only when installation fails or times out. */
58 readonly commit: (
59 requestId: string,
60 onHandoffAccepted?: () => Effect.Effect<void>,
61 ) => Effect.Effect<never, ServerSelfUpdateError>;
62 }
63>()("t3/desktopUpdate/DesktopAppUpdate") {}
64
65export const make = Effect.fn("desktopUpdate.desktopAppUpdate.make")(function* () {
66 const config = yield* ServerConfig.ServerConfig;
67 const crypto = yield* Crypto.Crypto;
68 const receiver = yield* DesktopTelemetryReceiver.DesktopTelemetryReceiver;
69 const inFlight = yield* Ref.make(false);
70
71 const available = config.mode === "desktop" && config.desktopTelemetryControlFd !== undefined;
72 const failWith = (reason: string, cause?: unknown) =>
73 cause === undefined
74 ? new ServerSelfUpdateError({ reason })
75 : new ServerSelfUpdateError({ reason, cause });
76
77 const consumeReports = (
78 requestId: string,
79 changes: Stream.Stream<DesktopUpdateStatusReport>,
80 reportProgress: (
81 stage: ServerSelfUpdateProgressStage,
82 ) => Effect.Effect<void, ServerSelfUpdateError>,
83 ) =>
84 Effect.gen(function* () {
85 const lastStage = yield* Ref.make<ServerSelfUpdateProgressStage | null>(null);
86 const emitStage = (
87 stage: ServerSelfUpdateProgressStage | null,
88 ): Effect.Effect<void, ServerSelfUpdateError> =>
89 stage === null
90 ? Effect.void
91 : Ref.get(lastStage).pipe(
92 Effect.flatMap((previous) =>
93 previous === stage
94 ? Effect.void
95 : Ref.set(lastStage, stage).pipe(Effect.andThen(reportProgress(stage))),
96 ),
97 );
98
99 const terminal = yield* changes.pipe(
100 Stream.filter((report) => report.requestId === requestId),
101 Stream.mapEffect(
102 (
103 report,
104 ): Effect.Effect<Option.Option<DesktopUpdateStatusReport>, ServerSelfUpdateError> =>
105 report.outcome === undefined
106 ? emitStage(desktopUpdateProgressStage(report.state)).pipe(
107 Effect.as(Option.none<DesktopUpdateStatusReport>()),
108 )
109 : Effect.succeedSome(report),
110 ),
111 Stream.filterMap(
112 Option.match({
113 onNone: () => Result.failVoid,
114 onSome: Result.succeed,
115 }),
116 ),
117 Stream.runHead,
118 );
119 if (Option.isNone(terminal)) {
120 return yield* failWith("The desktop app stopped reporting its update.");
121 }
122
123 const report = terminal.value;
124 if (report.outcome === "ready-to-install") {
125 yield* emitStage("installing");
126 const targetVersion =
127 report.state.downloadedVersion ??
128 report.state.availableVersion ??
129 report.state.currentVersion;
130 yield* Effect.logInfo("Desktop app update prepared for install.", {
131 targetVersion,
132 });
133 yield* Ref.set(inFlight, false);
134 return { targetVersion, method: "desktop-app" as const, desktopUpdateToken: requestId };
135 }
136 if (report.outcome === "up-to-date") {
137 return yield* failWith(
138 `The T3 Code desktop app on this machine is already up to date on ${report.state.currentVersion}.`,
139 );
140 }
141 return yield* failWith(
142 report.reason ?? report.state.message ?? "The desktop app update failed.",
143 );
144 });
145
146 const run: DesktopAppUpdate["Service"]["run"] = Effect.fn("desktopUpdate.desktopAppUpdate.run")(
147 function* (reportProgress) {
148 if (!available) {
149 return yield* failWith(
150 "This server was not started by the T3 Code desktop app, so it cannot drive a desktop update.",
151 );
152 }
153 if (yield* Ref.getAndSet(inFlight, true)) {
154 return yield* failWith("A desktop app update is already in progress.");
155 }
156
157 return yield* Effect.scoped(
158 Effect.gen(function* () {
159 const requestId = yield* crypto
160 .randomUUIDv4()
161 .pipe(
162 Effect.mapError((error) =>
163 failWith("Could not generate a desktop update request id.", error),
164 ),
165 );
166 // Subscribe before sending the request so a fast first report
167 // cannot be missed.
168 const { changes } = yield* receiver.desktopUpdates;
169 yield* receiver
170 .requestDesktopUpdate(requestId)
171 .pipe(
172 Effect.mapError((error) =>
173 failWith("Could not reach the T3 Code desktop app on this machine.", error),
174 ),
175 );
176 return yield* consumeReports(requestId, changes, reportProgress).pipe(
177 Effect.onInterrupt(() => receiver.cancelDesktopUpdate(requestId).pipe(Effect.ignore)),
178 );
179 }),
180 ).pipe(
181 Effect.timeout(DESKTOP_UPDATE_TIMEOUT),
182 Effect.catchTags({
183 TimeoutError: () => failWith("The desktop app did not finish the update in time."),
184 }),
185 Effect.onError(() => Ref.set(inFlight, false)),
186 );
187 },
188 );
189
190 const commit: DesktopAppUpdate["Service"]["commit"] = Effect.fn(
191 "desktopUpdate.desktopAppUpdate.commit",
192 )(function* (requestId, onHandoffAccepted = () => Effect.void) {
193 if (!available) {
194 return yield* failWith("This server cannot commit a desktop app update.");
195 }
196 const terminal = yield* Effect.scoped(
197 Effect.gen(function* () {
198 const { latest, changes } = yield* receiver.desktopUpdates;
199 const reports = Option.match(latest, {
200 onNone: () => changes,
201 onSome: (report) => Stream.concat(Stream.make(report), changes),
202 });
203 yield* Effect.uninterruptible(
204 receiver.commitDesktopUpdate(requestId).pipe(
205 Effect.mapError((error) => failWith("Could not reach the T3 Code desktop app.", error)),
206 Effect.tap(() => onHandoffAccepted()),
207 ),
208 );
209 return yield* reports.pipe(
210 Stream.filter((report) => report.requestId === requestId && report.outcome === "failed"),
211 Stream.runHead,
212 Effect.timeout(DESKTOP_INSTALL_TIMEOUT),
213 Effect.catchTags({
214 TimeoutError: () =>
215 failWith("The desktop app did not report an install result in time."),
216 }),
217 );
218 }),
219 );
220 if (Option.isNone(terminal)) {
221 return yield* failWith("The desktop app stopped reporting the install.");
222 }
223 return yield* failWith(
224 terminal.value.reason ??
225 terminal.value.state.message ??
226 "The desktop app failed to install the update.",
227 );
228 });
229
230 return DesktopAppUpdate.of({ available, run, commit });
231});
232
233export const layer = Layer.effect(DesktopAppUpdate, make());