1import * as NodeSocket from "@effect/platform-node/NodeSocket";
2import {
3 AuthStandardClientScopes,
4 HubError,
5 OutboundHubMessage,
6 type AuthSessionId,
7 type HostedSourceConnection,
8 OutboundSourceMessage,
9} from "@t3tools/contracts";
10import * as DateTime from "effect/DateTime";
11import * as Duration from "effect/Duration";
12import * as Effect from "effect/Effect";
13import * as Layer from "effect/Layer";
14import * as Option from "effect/Option";
15import * as Schedule from "effect/Schedule";
16import * as Schema from "effect/Schema";
17import * as Semaphore from "effect/Semaphore";
18import * as Stream from "effect/Stream";
19import { HttpServer } from "effect/http";
20import * as Socket from "effect/socket/Socket";
21import * as SessionStore from "../auth/SessionStore.ts";
22import * as ServerActivation from "../serverActivation.ts";
23import { bridge } from "./OutboundSocket.ts";
24import * as ServerEnvironment from "./ServerEnvironment.ts";
25import * as SourceConnection from "./SourceConnection.ts";
26
27const encodeSource = Schema.encodeSync(Schema.fromJsonString(OutboundSourceMessage));
28const decodeHub = Schema.decodeEffect(Schema.fromJsonString(OutboundHubMessage));
29
30export const layer = Layer.effectDiscard(
31 Effect.gen(function* () {
32 const connection = yield* SourceConnection.SourceConnection;
33 const sessions = yield* SessionStore.SessionStore;
34 const environment = yield* ServerEnvironment.ServerEnvironment;
35 const server = yield* HttpServer.HttpServer;
36 const address = server.address;
37 if (address._tag !== "InetAddressV4" && address._tag !== "InetAddressV6") return;
38
39 const connect = (target: HostedSourceConnection) =>
40 Effect.scoped(
41 Effect.gen(function* () {
42 const now = yield* DateTime.now;
43 const remaining = target.expiresAt.epochMilliseconds - now.epochMilliseconds;
44 if (remaining <= 0) return yield* Effect.never;
45 const url = new URL("api/hub/outbound", target.serverUrl);
46 url.protocol = "wss:";
47 const socket = yield* Socket.makeWebSocket(url.href, {
48 headers: { authorization: target.authorization },
49 openTimeout: "10 seconds",
50 highWaterMark: 1024 * 1024,
51 }).pipe(Effect.provide(NodeSocket.layerWebSocketConstructorWS));
52 const reader = yield* socket.reader;
53 const writer = yield* socket.writer;
54 const lock = yield* Semaphore.make(1);
55 const send = (message: OutboundSourceMessage) =>
56 lock.withPermit(writer.write(encodeSource(message)));
57 const grants = new Map<AuthSessionId, number>();
58 const channels = yield* Semaphore.make(512);
59 yield* Effect.addFinalizer(() =>
60 Effect.gen(function* () {
61 yield* connection.setConnected(false);
62 yield* Effect.forEach(
63 grants.keys(),
64 (id) => sessions.revoke(id).pipe(Effect.ignore),
65 { concurrency: 8, discard: true },
66 );
67 }),
68 );
69 yield* send({ type: "register", descriptor: yield* environment.getDescriptor });
70 const run = reader.pull.pipe(
71 Effect.timeout("60 seconds"),
72 Effect.flatMap((frames) =>
73 Effect.forEach(
74 frames,
75 (frame) =>
76 Effect.gen(function* () {
77 if (
78 (typeof frame === "string" ? Buffer.byteLength(frame) : frame.byteLength) >
79 65_536
80 )
81 return yield* new HubError({ reason: "identity" });
82 const text =
83 typeof frame === "string" ? frame : new TextDecoder().decode(frame);
84 const message = yield* decodeHub(text);
85 yield* connection.setConnected(true);
86 if (message.type === "ping") return yield* send({ type: "pong" });
87 if (message.type === "grant") {
88 if (
89 message.scopes.length === 0 ||
90 message.scopes.some(
91 (scope) => !AuthStandardClientScopes.some((allowed) => allowed === scope),
92 )
93 ) {
94 return yield* new HubError({ reason: "credential" });
95 }
96 const now = yield* DateTime.now;
97 for (const [id, expiresAt] of grants)
98 if (expiresAt <= now.epochMilliseconds) grants.delete(id);
99 if (grants.size >= 1_024) return yield* new HubError({ reason: "upstream" });
100 const issued = yield* sessions.issue({
101 subject: "self-hosted-outbound",
102 method: "bearer-access-token",
103 scopes: [...new Set(message.scopes)],
104 ttl: Duration.millis(
105 Math.min(
106 120_000,
107 target.expiresAt.epochMilliseconds - now.epochMilliseconds,
108 ),
109 ),
110 });
111 grants.set(issued.sessionId, issued.expiresAt.epochMilliseconds);
112 return yield* send({ type: "grant", id: message.id, token: issued.token });
113 }
114 yield* Effect.scoped(
115 Effect.gen(function* () {
116 const channelUrl = new URL(
117 `api/hub/outbound/channel/${encodeURIComponent(message.id)}`,
118 target.serverUrl,
119 );
120 channelUrl.protocol = "wss:";
121 const channel = yield* Socket.makeWebSocket(channelUrl.href, {
122 headers: { authorization: target.authorization },
123 openTimeout: "10 seconds",
124 highWaterMark: 1024 * 1024,
125 }).pipe(Effect.provide(NodeSocket.layerWebSocketConstructorWS));
126 const local = yield* NodeSocket.makeNet({
127 host: "127.0.0.1",
128 port: address.port,
129 openTimeout: "10 seconds",
130 });
131 return yield* bridge(local, channel);
132 }),
133 ).pipe(
134 channels.withPermitsIfAvailable(1),
135 Effect.ignore,
136 Effect.forkScoped({ startImmediately: true }),
137 );
138 }),
139 { discard: true },
140 ),
141 ),
142 Effect.forever,
143 );
144 yield* Effect.raceFirst(run, Effect.sleep(Duration.millis(remaining)));
145 }),
146 );
147
148 yield* ServerActivation.forkParked(
149 connection.changes.pipe(
150 Stream.switchMap((target) =>
151 Option.isNone(target)
152 ? Stream.never
153 : Stream.fromEffect(
154 connect(target.value).pipe(
155 Effect.retry(
156 Schedule.exponential("1 second").pipe(
157 Schedule.modifyDelay(({ duration }) =>
158 Effect.succeed(Duration.min(duration, Duration.seconds(30))),
159 ),
160 ),
161 ),
162 ),
163 ),
164 ),
165 Stream.runDrain,
166 ),
167 );
168 }),
169);