1import * as NodeSocketServer from "@effect/platform-node/NodeSocketServer";
2import {
3 AuthEnvironmentConnectScope,
4 AuthStandardClientScopes,
5 EnvironmentId,
6 ExecutionEnvironmentDescriptor,
7 HubError,
8 OutboundSourceMessage,
9 OutboundHubMessage,
10} from "@t3tools/contracts";
11import * as Context from "effect/Context";
12import * as Clock from "effect/Clock";
13import * as Crypto from "effect/Crypto";
14import * as Deferred from "effect/Deferred";
15import * as Effect from "effect/Effect";
16import * as Layer from "effect/Layer";
17import * as Option from "effect/Option";
18import * as Redacted from "effect/Redacted";
19import * as Schema from "effect/Schema";
20import * as Scope from "effect/Scope";
21import * as Semaphore from "effect/Semaphore";
22import * as Socket from "effect/socket/Socket";
23import type { AuthenticatedSession } from "../auth/EnvironmentAuth.ts";
24import * as HostedAuth from "../auth/HostedAuth.ts";
25import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
26import * as SessionStore from "../auth/SessionStore.ts";
27import { bridge } from "./OutboundSocket.ts";
28
29const encodeHub = Schema.encodeSync(Schema.fromJsonString(OutboundHubMessage));
30const decodeSource = Schema.decodeEffect(Schema.fromJsonString(OutboundSourceMessage));
31
32const Enrollment = Schema.Struct({
33 deviceId: Schema.String,
34 descriptor: ExecutionEnvironmentDescriptor,
35});
36const Stored = Schema.fromJsonString(Schema.Array(Enrollment));
37const decodeStored = Schema.decodeEffect(Stored);
38const encodeStored = Schema.encodeEffect(Stored);
39type Enrollment = typeof Enrollment.Type;
40type Peer = {
41 readonly session: AuthenticatedSession;
42 readonly url: string;
43 readonly send: (message: OutboundHubMessage) => Effect.Effect<void, Socket.SocketError>;
44 readonly closed: Deferred.Deferred<void>;
45 readonly grants: Map<string, Deferred.Deferred<string>>;
46 readonly credentials: Map<string, { readonly token: string; readonly expiresAt: number }>;
47};
48
49export class OutboundHub extends Context.Service<
50 OutboundHub,
51 {
52 readonly enrollments: Effect.Effect<ReadonlyArray<Enrollment>>;
53 readonly connect: (
54 id: EnvironmentId,
55 session: AuthenticatedSession,
56 ) => Effect.Effect<
57 {
58 readonly url: string;
59 readonly token: string;
60 readonly invalidated: Effect.Effect<void>;
61 },
62 HubError
63 >;
64 readonly remove: (id: EnvironmentId) => Effect.Effect<void, HubError>;
65 readonly accept: (
66 session: AuthenticatedSession,
67 socket: Socket.Socket,
68 ) => Effect.Effect<void, HubError, Scope.Scope>;
69 readonly channel: (
70 session: AuthenticatedSession,
71 id: string,
72 socket: Socket.Socket,
73 ) => Effect.Effect<void, HubError, Scope.Scope>;
74 }
75>()("t3/environment/OutboundHub") {}
76
77const isHubError = Schema.is(HubError);
78const failure = (cause: unknown) =>
79 isHubError(cause) ? cause : new HubError({ reason: "upstream", cause: Redacted.make(cause) });
80
81const make = Effect.gen(function* () {
82 const secrets = yield* ServerSecretStore.ServerSecretStore;
83 const hosted = yield* HostedAuth.HostedAuth;
84 const sessions = yield* SessionStore.SessionStore;
85 const crypto = yield* Crypto.Crypto;
86 const lock = yield* Semaphore.make(1);
87 const stored = yield* secrets.get("outbound-environments").pipe(Effect.mapError(failure));
88 let enrollments = Option.isSome(stored)
89 ? yield* decodeStored(new TextDecoder().decode(stored.value)).pipe(Effect.mapError(failure))
90 : [];
91 const peers = new Map<EnvironmentId, Peer>();
92 const channels = new Map<
93 string,
94 {
95 peer: Peer;
96 socket: Socket.Socket;
97 done: Deferred.Deferred<void>;
98 attached: Deferred.Deferred<void>;
99 claimed: boolean;
100 }
101 >();
102 const persist = (next: ReadonlyArray<Enrollment>) =>
103 encodeStored(next).pipe(
104 Effect.flatMap((json) =>
105 secrets.set("outbound-environments", new TextEncoder().encode(json)),
106 ),
107 Effect.tap(() =>
108 Effect.sync(() => {
109 enrollments = next;
110 }),
111 ),
112 Effect.mapError(failure),
113 );
114
115 const accept: OutboundHub["Service"]["accept"] = (session, socket) =>
116 Effect.gen(function* () {
117 if (!session.deviceId || !session.scopes.includes(AuthEnvironmentConnectScope)) {
118 return yield* new HubError({ reason: "credential" });
119 }
120 const deviceId = session.deviceId;
121 const reader = yield* socket.reader;
122 const writer = yield* socket.writer;
123 const writeLock = yield* Semaphore.make(1);
124 const send = (message: OutboundHubMessage) =>
125 writeLock.withPermit(writer.write(encodeHub(message)));
126 const read = reader.pull.pipe(
127 Effect.flatMap((frames) =>
128 Effect.forEach(frames, (frame) => {
129 if ((typeof frame === "string" ? Buffer.byteLength(frame) : frame.byteLength) > 65_536)
130 return Effect.fail(new HubError({ reason: "identity" }));
131 const text = typeof frame === "string" ? frame : new TextDecoder().decode(frame);
132 return decodeSource(text).pipe(Effect.mapError(failure));
133 }),
134 ),
135 Effect.timeout("60 seconds"),
136 );
137 const initial = yield* read.pipe(Effect.timeout("10 seconds"));
138 if (initial.length !== 1 || initial[0]?.type !== "register")
139 return yield* new HubError({ reason: "identity" });
140 const descriptor = initial[0].descriptor;
141 const id = descriptor.environmentId;
142 yield* hosted.bindEnvironment(session, id);
143 yield* lock.withPermit(
144 Effect.gen(function* () {
145 yield* persist([
146 ...enrollments.filter((value) => value.descriptor.environmentId !== id),
147 { deviceId, descriptor },
148 ]);
149 }),
150 );
151 const server = yield* NodeSocketServer.make({ host: "127.0.0.1", port: 0 });
152 if (server.address._tag !== "InetAddressV4")
153 return yield* new HubError({ reason: "upstream" });
154 const closed = yield* Deferred.make<void>();
155 const peer: Peer = {
156 session,
157 url: `http://127.0.0.1:${server.address.port}/`,
158 send,
159 closed,
160 grants: new Map(),
161 credentials: new Map(),
162 };
163 const previous = peers.get(id);
164 if (previous) yield* Deferred.succeed(previous.closed, undefined);
165 peers.set(id, peer);
166 yield* Effect.addFinalizer(() =>
167 Effect.gen(function* () {
168 if (peers.get(id) === peer) peers.delete(id);
169 yield* Deferred.succeed(closed, undefined);
170 for (const [key, channel] of channels)
171 if (channel.peer === peer) {
172 channels.delete(key);
173 yield* Deferred.succeed(channel.done, undefined);
174 }
175 }),
176 );
177 yield* server
178 .run((local) =>
179 Effect.scoped(
180 Effect.gen(function* () {
181 if (channels.size >= 512) return;
182 const channelId = yield* crypto.randomUUIDv4();
183 const done = yield* Deferred.make<void>();
184 const attached = yield* Deferred.make<void>();
185 channels.set(channelId, { peer, socket: local, done, attached, claimed: false });
186 yield* Effect.addFinalizer(() =>
187 Effect.sync(() => {
188 channels.delete(channelId);
189 }),
190 );
191 yield* send({ type: "connect", id: channelId });
192 yield* Deferred.await(attached).pipe(Effect.timeout("10 seconds"));
193 yield* Deferred.await(done);
194 }),
195 ).pipe(Effect.ignore),
196 )
197 .pipe(Effect.forkScoped);
198 yield* Effect.forever(
199 send({ type: "ping" }).pipe(Effect.andThen(Effect.sleep("20 seconds"))),
200 ).pipe(Effect.forkScoped);
201 const messages = Effect.forever(
202 read.pipe(
203 Effect.flatMap((messages) =>
204 Effect.forEach(
205 messages,
206 (message) => {
207 if (message.type !== "grant")
208 return message.type === "pong"
209 ? Effect.void
210 : Effect.fail(new HubError({ reason: "identity" }));
211 const pending = peer.grants.get(message.id);
212 return pending
213 ? Deferred.succeed(pending, message.token).pipe(Effect.asVoid)
214 : Effect.void;
215 },
216 { discard: true },
217 ),
218 ),
219 ),
220 );
221 yield* Effect.raceFirst(
222 messages,
223 Effect.raceFirst(Deferred.await(closed), sessions.awaitInvalidation(session.sessionId)),
224 );
225 }).pipe(Effect.scoped, Effect.mapError(failure));
226
227 return OutboundHub.of({
228 enrollments: Effect.sync(() => enrollments),
229 accept,
230 remove: (id) =>
231 lock.withPermit(
232 Effect.gen(function* () {
233 yield* persist(enrollments.filter((value) => value.descriptor.environmentId !== id));
234 const peer = peers.get(id);
235 if (peer) yield* Deferred.succeed(peer.closed, undefined);
236 }),
237 ),
238 connect: (id, session) =>
239 Effect.gen(function* () {
240 const peer = peers.get(id);
241 if (!peer) return yield* new HubError({ reason: "upstream" });
242 const scopes = AuthStandardClientScopes.filter((scope) => session.scopes.includes(scope));
243 if (scopes.length === 0) return yield* new HubError({ reason: "credential" });
244 const now = yield* Clock.currentTimeMillis;
245 for (const [key, credential] of peer.credentials)
246 if (credential.expiresAt <= now) peer.credentials.delete(key);
247 const key = `${session.sessionId}:${scopes.join(" ")}`;
248 const cached = peer.credentials.get(key);
249 if (cached)
250 return { url: peer.url, token: cached.token, invalidated: Deferred.await(peer.closed) };
251 if (peer.grants.size >= 128 || peer.credentials.size >= 512)
252 return yield* new HubError({ reason: "upstream" });
253 const grantId = yield* crypto.randomUUIDv4();
254 const grant = yield* Deferred.make<string>();
255 peer.grants.set(grantId, grant);
256 const token = yield* Effect.gen(function* () {
257 yield* peer.send({ type: "grant", id: grantId, scopes });
258 return yield* Effect.raceFirst(
259 Deferred.await(grant),
260 Deferred.await(peer.closed).pipe(
261 Effect.andThen(Effect.fail(new HubError({ reason: "upstream" }))),
262 ),
263 );
264 }).pipe(
265 Effect.timeout("10 seconds"),
266 Effect.ensuring(
267 Effect.sync(() => {
268 peer.grants.delete(grantId);
269 }),
270 ),
271 );
272 peer.credentials.set(key, { token, expiresAt: now + 60_000 });
273 return { url: peer.url, token, invalidated: Deferred.await(peer.closed) };
274 }).pipe(Effect.mapError(failure)),
275 channel: (session, id, socket) =>
276 Effect.gen(function* () {
277 const channel = channels.get(id);
278 if (
279 !channel ||
280 channel.claimed ||
281 session.deviceId !== channel.peer.session.deviceId ||
282 session.sessionId !== channel.peer.session.sessionId
283 )
284 return yield* new HubError({ reason: "credential" });
285 channel.claimed = true;
286 yield* Deferred.succeed(channel.attached, undefined);
287 yield* Effect.raceFirst(
288 bridge(channel.socket, socket),
289 Deferred.await(channel.peer.closed),
290 ).pipe(Effect.ensuring(Deferred.succeed(channel.done, undefined)), Effect.ignore);
291 }),
292 });
293});
294
295export const layer = Layer.effect(OutboundHub, make);