1import { HubError } from "@t3tools/contracts";
2import * as Effect from "effect/Effect";
3import * as Semaphore from "effect/Semaphore";
4import * as Socket from "effect/socket/Socket";
5
6const CHUNK_BYTES = 64 * 1024;
7const WINDOW_CHUNKS = 16;
8
9/** ACKs follow TCP drain, bounding WebSocket's otherwise unbounded send buffer. */
10export const bridge = (tcp: Socket.Socket, tunnel: Socket.Socket) =>
11 Effect.gen(function* () {
12 const tcpReader = yield* tcp.reader;
13 const tunnelReader = yield* tunnel.reader;
14 const tcpWriter = yield* tcp.writer;
15 const tunnelWriter = yield* tunnel.writer;
16 const credits = yield* Semaphore.make(WINDOW_CHUNKS);
17 const writes = yield* Semaphore.make(1);
18 const pending = new Set<number>();
19 let sequence = 0;
20 let expected = 0;
21
22 const send = tcpReader.pull.pipe(
23 Effect.flatMap((chunks) =>
24 Effect.forEach(
25 chunks,
26 (chunk) =>
27 Effect.gen(function* () {
28 const bytes = typeof chunk === "string" ? new TextEncoder().encode(chunk) : chunk;
29 for (let offset = 0; offset < bytes.byteLength; offset += CHUNK_BYTES) {
30 yield* credits.take(1);
31 if (sequence > 0xffff_ffff) return yield* new HubError({ reason: "upstream" });
32 const id = sequence++;
33 const payload = bytes.subarray(offset, offset + CHUNK_BYTES);
34 const frame = new Uint8Array(5 + payload.byteLength);
35 new DataView(frame.buffer).setUint32(1, id);
36 frame.set(payload, 5);
37 pending.add(id);
38 yield* writes.withPermit(tunnelWriter.write(frame));
39 }
40 }),
41 { discard: true },
42 ),
43 ),
44 Effect.forever,
45 );
46 const receive = tunnelReader.pull.pipe(
47 Effect.flatMap((frames) =>
48 Effect.forEach(
49 frames,
50 (frame) =>
51 Effect.gen(function* () {
52 if (
53 typeof frame === "string" ||
54 frame.byteLength < 5 ||
55 frame.byteLength > CHUNK_BYTES + 5
56 ) {
57 return yield* new HubError({ reason: "identity" });
58 }
59 const id = new DataView(frame.buffer, frame.byteOffset, frame.byteLength).getUint32(
60 1,
61 );
62 if (frame[0] === 1) {
63 if (frame.byteLength !== 5 || !pending.delete(id))
64 return yield* new HubError({ reason: "identity" });
65 yield* credits.release(1);
66 return;
67 }
68 if (frame[0] !== 0 || frame.byteLength === 5 || id !== expected++) {
69 return yield* new HubError({ reason: "identity" });
70 }
71 yield* tcpWriter.write(frame.subarray(5));
72 const ack = new Uint8Array(5);
73 ack[0] = 1;
74 new DataView(ack.buffer).setUint32(1, id);
75 yield* writes.withPermit(tunnelWriter.write(ack));
76 }),
77 { discard: true },
78 ),
79 ),
80 Effect.forever,
81 );
82 return yield* Effect.raceFirst(send, receive);
83 });