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