import { ClientPresentation, CloudSession, EnvironmentOwnedDataCleanup, PlatformConnectionSource, PrimaryEnvironmentAuth, RelayDeviceIdentity, SshEnvironmentGateway, } from "@t3tools/client-runtime/platform"; import { ConnectionBlockedError, ConnectionTransientError, Connectivity, mapRemoteEnvironmentError, PrimaryConnectionRegistration, PrimaryConnectionTarget, Wakeups, } from "@t3tools/client-runtime/connection"; import { fetchRemoteEnvironmentDescriptor } from "@t3tools/client-runtime/environment"; import { managedRelayAccountChanges, managedRelaySessionAtom } from "@t3tools/client-runtime/relay"; import { EnvironmentRpcRequestObserver } from "@t3tools/client-runtime/rpc"; import { AuthStandardClientScopes, type DesktopBridge, type DesktopSshEnvironmentTarget, } from "@t3tools/contracts"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Schedule from "effect/Schedule"; import * as Stream from "effect/Stream"; import { readDesktopPrimaryBearerToken } from "../environments/primary/desktopAuth"; import { primaryEnvironmentHttpLayer } from "../environments/primary/httpLayer"; import { readPrimaryEnvironmentTarget } from "../environments/primary/target"; import { clearComposerDraftsEnvironment } from "../composerDraftStore"; import { isHostedStaticApp } from "../hostedPairing"; import { appAtomRegistry } from "../rpc/atomRegistry"; import { acknowledgeRpcRequest, trackRpcRequestSent } from "../rpc/requestLatencyState"; import { connectionStorageLayer } from "./storage"; let nextObservedRpcRequestId = 0; function currentNetworkStatus(): "unknown" | "offline" | "online" { if (typeof navigator === "undefined") { return "unknown"; } return navigator.onLine ? "online" : "offline"; } const connectivityLayer = Connectivity.layer({ status: Effect.sync(currentNetworkStatus), changes: Stream.callback((queue) => Effect.acquireRelease( Effect.sync(() => { const online = () => Queue.offerUnsafe(queue, "online"); const offline = () => Queue.offerUnsafe(queue, "offline"); window.addEventListener("online", online); window.addEventListener("offline", offline); return { online, offline }; }), ({ online, offline }) => Effect.sync(() => { window.removeEventListener("online", online); window.removeEventListener("offline", offline); }), ).pipe(Effect.asVoid), ), }); const wakeupsLayer = Wakeups.layer({ changes: Stream.merge( Stream.callback<"application-active">((queue) => Effect.acquireRelease( Effect.sync(() => { const listener = () => { if (document.visibilityState === "visible") { Queue.offerUnsafe(queue, "application-active"); } }; document.addEventListener("visibilitychange", listener); return listener; }), (listener) => Effect.sync(() => { document.removeEventListener("visibilitychange", listener); }), ).pipe(Effect.asVoid), ), managedRelayAccountChanges(appAtomRegistry).pipe( Stream.map(() => "credentials-changed" as const), ), ), }); function clientMetadata() { const desktop = window.desktopBridge !== undefined; const platform = navigator.platform.trim(); return { label: desktop ? "T3 Code Desktop" : "T3 Code Web", deviceType: "desktop" as const, ...(platform === "" ? {} : { os: platform }), }; } function sshPreparationError(cause: unknown) { const message = cause instanceof Error ? cause.message : String(cause); if (message.toLowerCase().includes("cancel")) { return new ConnectionBlockedError({ reason: "authentication", detail: message, }); } return new ConnectionTransientError({ reason: "remote-unavailable", detail: `Could not prepare the SSH environment: ${message}`, }); } export const provisionDesktopSshEnvironment = Effect.fn( "web.connectionPlatform.ssh.provisionDesktop", )(function* (bridge: DesktopBridge, target: DesktopSshEnvironmentTarget) { const bootstrap = yield* Effect.tryPromise({ try: () => bridge.ensureSshEnvironment(target, { issuePairingToken: true, }), catch: sshPreparationError, }); const pairingToken = bootstrap.pairingToken; if (pairingToken === null) { return yield* new ConnectionBlockedError({ reason: "authentication", detail: "The SSH environment did not issue a pairing credential.", }); } const descriptor = yield* Effect.tryPromise({ try: () => bridge.fetchSshEnvironmentDescriptor(bootstrap.httpBaseUrl), catch: sshPreparationError, }); const access = yield* Effect.tryPromise({ try: () => bridge.bootstrapSshBearerSession(bootstrap.httpBaseUrl, pairingToken), catch: sshPreparationError, }); return { environmentId: descriptor.environmentId, label: descriptor.label, bootstrap, bearerToken: access.access_token, }; }); const capabilitiesLayer = Layer.effectContext( Effect.sync(() => { const presentation = ClientPresentation.of({ metadata: clientMetadata(), scopes: AuthStandardClientScopes, }); const cloudSession = CloudSession.of({ clerkToken: Effect.gen(function* () { const session = appAtomRegistry.get(managedRelaySessionAtom); if (session === null) { return yield* new ConnectionBlockedError({ reason: "authentication", detail: "Sign in to T3 Cloud to connect this environment.", }); } const token = yield* session.readClerkToken().pipe( Effect.mapError( (error) => new ConnectionTransientError({ reason: "network", detail: error.message, }), ), ); return yield* Option.match(token, { onNone: () => Effect.fail( new ConnectionBlockedError({ reason: "authentication", detail: "The T3 Cloud session is unavailable.", }), ), onSome: Effect.succeed, }); }), }); const identity = RelayDeviceIdentity.of({ deviceId: Effect.succeed(Option.none()), }); const primaryAuth = PrimaryEnvironmentAuth.of({ bearerToken: Effect.tryPromise({ try: readDesktopPrimaryBearerToken, catch: (cause) => new ConnectionTransientError({ reason: "remote-unavailable", detail: `Could not load the desktop primary credential: ${String(cause)}`, }), }).pipe(Effect.map(Option.fromNullishOr)), }); const ssh = SshEnvironmentGateway.of({ provision: Effect.fn("web.connectionPlatform.ssh.provision")(function* (target) { const bridge = window.desktopBridge; if (bridge === undefined) { return yield* new ConnectionBlockedError({ reason: "unsupported", detail: "SSH environments are only available in the desktop app.", }); } return yield* provisionDesktopSshEnvironment(bridge, target); }), prepare: Effect.fn("web.connectionPlatform.ssh.prepare")(function* (input) { const bridge = window.desktopBridge; if (bridge === undefined) { return yield* new ConnectionBlockedError({ reason: "unsupported", detail: "SSH environments are only available in the desktop app.", }); } const bootstrap = yield* Effect.tryPromise({ try: () => bridge.ensureSshEnvironment(input.target, { issuePairingToken: true, }), catch: sshPreparationError, }); if (bootstrap.pairingToken === null) { return yield* new ConnectionBlockedError({ reason: "authentication", detail: "The SSH environment did not issue a pairing credential.", }); } const access = yield* Effect.tryPromise({ try: () => bridge.bootstrapSshBearerSession(bootstrap.httpBaseUrl, bootstrap.pairingToken!), catch: sshPreparationError, }); return { bootstrap, bearerToken: access.access_token, }; }), disconnect: Effect.fn("web.connectionPlatform.ssh.disconnect")(function* (target) { const bridge = window.desktopBridge; if (bridge === undefined) { return; } yield* Effect.tryPromise({ try: () => bridge.disconnectSshEnvironment(target), catch: (cause) => new ConnectionTransientError({ reason: "remote-unavailable", detail: `Could not disconnect the SSH environment: ${String(cause)}`, }), }); }), }); return Context.make(CloudSession, cloudSession).pipe( Context.add(PrimaryEnvironmentAuth, primaryAuth), Context.add(RelayDeviceIdentity, identity), Context.add(ClientPresentation, presentation), Context.add(SshEnvironmentGateway, ssh), ); }), ); const loadPrimaryConnectionRegistration = Effect.fn( "web.connectionPlatform.loadPrimaryConnectionRegistration", )(function* () { const resolved = readPrimaryEnvironmentTarget(); if (resolved === null) { return yield* new ConnectionBlockedError({ reason: "configuration", detail: "Unable to resolve the primary environment endpoint.", }); } const descriptor = yield* fetchRemoteEnvironmentDescriptor({ httpBaseUrl: resolved.target.httpBaseUrl, }).pipe(Effect.provide(primaryEnvironmentHttpLayer), Effect.mapError(mapRemoteEnvironmentError)); return new PrimaryConnectionRegistration({ target: new PrimaryConnectionTarget({ environmentId: descriptor.environmentId, label: descriptor.label, httpBaseUrl: resolved.target.httpBaseUrl, wsBaseUrl: resolved.target.wsBaseUrl, }), }); }); const primaryRegistrationRetrySchedule = Schedule.exponential("1 second").pipe( Schedule.either(Schedule.spaced("16 seconds")), ); const platformConnectionSourceLayer = Layer.sync(PlatformConnectionSource, () => { if (isHostedStaticApp()) { return PlatformConnectionSource.of({ registrations: Stream.empty, }); } return PlatformConnectionSource.of({ registrations: Stream.fromEffect(loadPrimaryConnectionRegistration()).pipe( Stream.tapError((error) => Effect.logWarning("Could not discover the primary environment.", { error, }), ), Stream.retry(primaryRegistrationRetrySchedule), Stream.catchCause(() => Stream.empty), ), }); }); const environmentOwnedDataCleanupLayer = Layer.succeed( EnvironmentOwnedDataCleanup, EnvironmentOwnedDataCleanup.of({ clear: (environmentId) => Effect.sync(() => { clearComposerDraftsEnvironment(environmentId); }), }), ); const rpcRequestObserverLayer = Layer.succeed( EnvironmentRpcRequestObserver, EnvironmentRpcRequestObserver.of({ observe: ({ environmentId, method }) => Effect.sync(() => { nextObservedRpcRequestId += 1; const requestId = `${environmentId}:${nextObservedRpcRequestId}`; trackRpcRequestSent(requestId, `${method} ยท ${environmentId}`); return Effect.sync(() => { acknowledgeRpcRequest(requestId); }); }), }), ); type ConnectionPlatformLayerSource = | typeof connectionStorageLayer | typeof connectivityLayer | typeof wakeupsLayer | typeof capabilitiesLayer | typeof platformConnectionSourceLayer | typeof environmentOwnedDataCleanupLayer | typeof rpcRequestObserverLayer; export const connectionPlatformLayer: Layer.Layer< Layer.Success, Layer.Error, Layer.Services > = Layer.mergeAll( connectionStorageLayer, connectivityLayer, wakeupsLayer, capabilitiesLayer, platformConnectionSourceLayer, environmentOwnedDataCleanupLayer, rpcRequestObserverLayer, );