1import {
2 ClientCapabilities,
3 PlatformConnectionSource,
4 Persistence,
5} from "@t3tools/client-runtime/platform";
6import {
7 ConnectionBlockedError,
8 ConnectionTransientError,
9 Connectivity,
10 Wakeups,
11} from "@t3tools/client-runtime/connection";
12import { managedRelayAccountChanges, managedRelaySessionAtom } from "@t3tools/client-runtime/relay";
13import * as Clock from "effect/Clock";
14import * as Context from "effect/Context";
15import * as Effect from "effect/Effect";
16import * as Layer from "effect/Layer";
17import * as Option from "effect/Option";
18import * as Queue from "effect/Queue";
19import * as Stream from "effect/Stream";
20import Constants from "expo-constants";
21import * as Network from "expo-network";
22import { AppState } from "react-native";
23
24import { authClientMetadata } from "../lib/authClientMetadata";
25import * as Runtime from "../lib/runtime";
26import * as MobileStorage from "../persistence/mobile-storage";
27import { appAtomRegistry } from "../state/atom-registry";
28import { clearThreadOutboxEnvironment } from "../state/thread-outbox-removal";
29import { clearComposerDraftsEnvironment } from "../state/use-composer-drafts";
30import { clearThreadComposerErrorsForEnvironment } from "../state/thread-composer-error";
31import { mobileApplicationActiveWakeup } from "./app-state-wakeups";
32import * as ConnectionStorage from "./storage";
33
34function networkStatus(state: Network.NetworkState): "unknown" | "offline" | "online" {
35 if (state.isConnected === false) {
36 return "offline";
37 }
38 if (state.isConnected === true) {
39 return "online";
40 }
41 return "unknown";
42}
43
44const layerConnectivity = Connectivity.layer({
45 status: Effect.tryPromise({
46 try: () => Network.getNetworkStateAsync(),
47 catch: () => undefined,
48 }).pipe(
49 Effect.match({
50 onFailure: () => "unknown" as const,
51 onSuccess: networkStatus,
52 }),
53 ),
54 changes: Stream.callback((queue) =>
55 Effect.acquireRelease(
56 Effect.sync(() => {
57 let active = true;
58 const networkSubscription = Network.addNetworkStateListener((state) => {
59 Queue.offerUnsafe(queue, networkStatus(state));
60 });
61 const appStateSubscription = AppState.addEventListener("change", (state) => {
62 if (state !== "active") {
63 return;
64 }
65 void Network.getNetworkStateAsync()
66 .then((current) => {
67 if (active) {
68 Queue.offerUnsafe(queue, networkStatus(current));
69 }
70 })
71 .catch(() => undefined);
72 });
73 return {
74 close: () => {
75 active = false;
76 networkSubscription.remove();
77 appStateSubscription.remove();
78 },
79 };
80 }),
81 ({ close }) => Effect.sync(close),
82 ).pipe(Effect.asVoid),
83 ),
84});
85
86/** Wake connections on network path changes; online/offline misses Wi-Fi-to-cellular moves. */
87const networkPathChanges = Stream.callback<"network-changed">((queue) =>
88 Effect.acquireRelease(
89 Effect.sync(() => {
90 let active = true;
91 let previous: Network.NetworkStateType | undefined;
92 const record = (state: Network.NetworkState) => {
93 const type = state.isConnected === true ? state.type : undefined;
94 if (previous !== undefined && type !== undefined && type !== previous) {
95 Queue.offerUnsafe(queue, "network-changed");
96 }
97 previous = type ?? previous;
98 };
99 // Seed the current type: the listener reports only changes and would miss the first move.
100 void Network.getNetworkStateAsync()
101 .then((state) => {
102 if (active && previous === undefined && state.isConnected === true) {
103 previous = state.type;
104 }
105 })
106 .catch(() => undefined);
107 const subscription = Network.addNetworkStateListener(record);
108 return {
109 remove: () => {
110 active = false;
111 subscription.remove();
112 },
113 };
114 }),
115 (subscription) => Effect.sync(() => subscription.remove()),
116 ).pipe(Effect.asVoid),
117);
118
119const layerWakeups = Wakeups.layer({
120 changes: Stream.mergeAll(
121 [
122 Stream.callback<"application-active-probe" | "application-active-long">((queue) =>
123 Effect.acquireRelease(
124 Effect.gen(function* () {
125 const nowMs = yield* Clock.currentTimeMillis;
126 let backgroundedAtMs = AppState.currentState === "background" ? nowMs : null;
127 return AppState.addEventListener("change", (state) => {
128 if (state === "background") {
129 backgroundedAtMs = Date.now();
130 return;
131 }
132 if (state === "active") {
133 Queue.offerUnsafe(
134 queue,
135 mobileApplicationActiveWakeup(backgroundedAtMs, Date.now()),
136 );
137 backgroundedAtMs = null;
138 }
139 });
140 }),
141 (subscription) => Effect.sync(() => subscription.remove()),
142 ).pipe(Effect.asVoid),
143 ),
144 managedRelayAccountChanges(appAtomRegistry).pipe(
145 Stream.map(() => "credentials-changed" as const),
146 ),
147 networkPathChanges,
148 ],
149 { concurrency: "unbounded" },
150 ),
151});
152
153const layerCapabilities = Layer.effectContext(
154 Effect.gen(function* () {
155 const storage = yield* MobileStorage.MobileStorage;
156 return Context.make(
157 ClientCapabilities.CloudSession,
158 ClientCapabilities.CloudSession.of({
159 identity: Effect.sync(() =>
160 Option.fromNullishOr(appAtomRegistry.get(managedRelaySessionAtom)),
161 ),
162 clerkToken: Effect.gen(function* () {
163 const session = appAtomRegistry.get(managedRelaySessionAtom);
164 if (session === null) {
165 return yield* new ConnectionBlockedError({
166 reason: "authentication",
167 detail: "Sign in to T3 Connect to connect this environment.",
168 });
169 }
170 const token = yield* session.readClerkToken().pipe(
171 Effect.mapError(
172 (error) =>
173 new ConnectionTransientError({
174 reason: "network",
175 detail: error.message,
176 }),
177 ),
178 );
179 if (token === null) {
180 return yield* new ConnectionBlockedError({
181 reason: "authentication",
182 detail: "The T3 Connect session is unavailable.",
183 });
184 }
185 return token;
186 }),
187 }),
188 ).pipe(
189 Context.add(
190 ClientCapabilities.PrimaryEnvironmentAuth,
191 ClientCapabilities.PrimaryEnvironmentAuth.of({
192 bearerToken: Effect.succeedNone,
193 }),
194 ),
195 // The relay device id already persists in secure storage; analytics only sends its hash.
196 Context.add(ClientCapabilities.AnalyticsDevice, {
197 id: storage.loadOrCreateAgentAwarenessDeviceId.pipe(Effect.option),
198 }),
199 Context.add(
200 ClientCapabilities.RelayDeviceIdentity,
201 ClientCapabilities.RelayDeviceIdentity.of({
202 deviceId: storage.loadOrCreateAgentAwarenessDeviceId.pipe(
203 Effect.mapError(
204 (cause) =>
205 new ConnectionTransientError({
206 reason: "remote-unavailable",
207 detail: `Could not load the mobile device identity: ${String(cause)}`,
208 }),
209 ),
210 Effect.asSome,
211 ),
212 }),
213 ),
214 Context.add(
215 ClientCapabilities.ClientPresentation,
216 ClientCapabilities.ClientPresentation.of({
217 metadata: authClientMetadata(Constants.expoConfig?.version),
218 }),
219 ),
220 Context.add(
221 ClientCapabilities.SshEnvironmentGateway,
222 ClientCapabilities.SshEnvironmentGateway.of({
223 provision: Effect.fn("ClientCapabilities.SshEnvironmentGateway.provision")(() =>
224 Effect.fail(
225 new ConnectionBlockedError({
226 reason: "unsupported",
227 detail: "SSH environments are only available in the desktop app.",
228 }),
229 ),
230 ),
231 prepare: Effect.fn("ClientCapabilities.SshEnvironmentGateway.prepare")(() =>
232 Effect.fail(
233 new ConnectionBlockedError({
234 reason: "unsupported",
235 detail: "SSH environments are only available in the desktop app.",
236 }),
237 ),
238 ),
239 disconnect: Effect.fn("ClientCapabilities.SshEnvironmentGateway.disconnect")(
240 () => Effect.void,
241 ),
242 }),
243 ),
244 );
245 }),
246);
247
248const layerPlatformConnectionSource = Layer.succeed(
249 PlatformConnectionSource.PlatformConnectionSource,
250 PlatformConnectionSource.PlatformConnectionSource.of({
251 registrations: Stream.empty,
252 }),
253);
254
255const layerProvidedConnectionStorage = ConnectionStorage.layer.pipe(Layer.provide(Runtime.layer));
256const layerProvidedCapabilities = layerCapabilities.pipe(Layer.provide(Runtime.layer));
257
258const layerEnvironmentOwnedDataCleanup = Layer.succeed(
259 Persistence.EnvironmentOwnedDataCleanup,
260 Persistence.EnvironmentOwnedDataCleanup.of({
261 clear: Effect.fn("Persistence.EnvironmentOwnedDataCleanup.clear")((environmentId) =>
262 Effect.all(
263 [
264 Effect.promise(() => clearThreadOutboxEnvironment(environmentId)),
265 Effect.promise(() => clearComposerDraftsEnvironment(environmentId)),
266 Effect.sync(() => clearThreadComposerErrorsForEnvironment(environmentId)),
267 ],
268 { concurrency: "unbounded", discard: true },
269 ),
270 ),
271 }),
272);
273
274type ConnectionPlatformLayerSource =
275 | typeof layerProvidedConnectionStorage
276 | typeof Runtime.layer
277 | typeof layerConnectivity
278 | typeof layerWakeups
279 | typeof layerProvidedCapabilities
280 | typeof layerPlatformConnectionSource
281 | typeof layerEnvironmentOwnedDataCleanup;
282
283export const layer: Layer.Layer<
284 Layer.Success<ConnectionPlatformLayerSource>,
285 Layer.Error<ConnectionPlatformLayerSource>,
286 Layer.Services<ConnectionPlatformLayerSource>
287> = Layer.mergeAll(
288 layerProvidedConnectionStorage,
289 Runtime.layer,
290 layerConnectivity,
291 layerWakeups,
292 layerProvidedCapabilities,
293 layerPlatformConnectionSource,
294 layerEnvironmentOwnedDataCleanup,
295);