decentralised sync engine
feat: shard forwarding
+65
-6
+29
-5
src/lib/handlers/connect.ts
+29
-5
src/lib/handlers/connect.ts
···
3
3
issuedLatticeTokens,
4
4
isValidSession,
5
5
} from "@/lib/sessions";
6
+
import { shardSessions } from "@/lib/state";
7
+
import type { ShardMessage } from "@/lib/types/messages";
6
8
import type { PreHandler, WsRouteHandler } from "@/lib/types/routes";
7
-
import { rawDataToString } from "@/lib/utils/ws";
9
+
import { stringToAtUri } from "@/lib/utils/atproto";
10
+
import { storeMessageInShard } from "@/lib/utils/gmstn";
11
+
import {
12
+
rawDataToString,
13
+
validateWsMessageType,
14
+
} from "@/lib/utils/ws/validate";
8
15
9
16
export const connectPreHandler: PreHandler = (req, reply, done) => {
10
17
const { query } = req;
···
65
72
return;
66
73
}
67
74
68
-
socket.on("message", (event) => {
69
-
const message = rawDataToString(event);
70
-
console.log(message)
71
-
})
75
+
socket.on("message", (rawData) => {
76
+
const event = rawDataToString(rawData);
77
+
78
+
const data: unknown = JSON.parse(event);
79
+
const validateTypeResult = validateWsMessageType(data);
80
+
if (!validateTypeResult.ok) return;
81
+
82
+
const { type: messageType } = validateTypeResult.data;
83
+
84
+
switch (messageType) {
85
+
case "shard/message": {
86
+
const shardMessage = validateTypeResult.data as ShardMessage;
87
+
const { channel } = shardMessage;
88
+
const atUriParseResult = stringToAtUri(channel);
89
+
if (!atUriParseResult.ok) return;
90
+
const { data: channelAtUri } = atUriParseResult;
91
+
92
+
storeMessageInShard({ channelAtUri, message: shardMessage });
93
+
}
94
+
}
95
+
});
72
96
};
+36
-1
src/lib/utils/gmstn.ts
+36
-1
src/lib/utils/gmstn.ts
···
1
1
import { shardSessions } from "@/lib/state";
2
-
import type { Did } from "@/lib/types/atproto";
2
+
import type { AtUri, Did } from "@/lib/types/atproto";
3
3
import type { ShardSessionInfo } from "@/lib/types/handshake";
4
+
import type { ShardMessage } from "@/lib/types/messages";
4
5
import { getEndpointFromDid } from "@/lib/utils/atproto";
5
6
import WebSocket from "ws";
6
7
···
22
23
shardSessions.set(sessionInfo, ws);
23
24
return ws;
24
25
};
26
+
27
+
export const storeMessageInShard = ({
28
+
channelAtUri,
29
+
message,
30
+
}: {
31
+
channelAtUri: AtUri;
32
+
message: ShardMessage;
33
+
}) => {
34
+
const sessionInfo = shardSessions
35
+
.keys()
36
+
.find((sessionInfo) =>
37
+
sessionInfo.allowedChannels.some(
38
+
(allowedChannel) => allowedChannel.rKey === channelAtUri.rKey,
39
+
),
40
+
);
41
+
if (!sessionInfo) return;
42
+
43
+
const shardSocket = shardSessions.get(sessionInfo);
44
+
if (!shardSocket) {
45
+
console.error(
46
+
"Could find session info object in map, but socket could not be retrieved from map. Race condition?",
47
+
);
48
+
return;
49
+
}
50
+
if (shardSocket.readyState === WebSocket.OPEN)
51
+
shardSocket.send(JSON.stringify(message));
52
+
53
+
console.log(
54
+
"Sent off message",
55
+
message,
56
+
"to shard located at",
57
+
shardSocket.url,
58
+
);
59
+
};