decentralised sync engine
feat: broadcast incoming message to clients
+64
-11
+19
-3
src/lib/handlers/connect.ts
+19
-3
src/lib/handlers/connect.ts
···
3
3
issuedLatticeTokens,
4
4
isValidSession,
5
5
} from "@/lib/sessions";
6
-
import type { ShardMessage } from "@/lib/types/messages";
6
+
import { shardMessageSchema } from "@/lib/types/messages";
7
7
import type { PreHandler, WsRouteHandler } from "@/lib/types/routes";
8
8
import { stringToAtUri } from "@/lib/utils/atproto";
9
-
import { storeMessageInShard } from "@/lib/utils/gmstn";
9
+
import { sendToChannelClients, storeMessageInShard } from "@/lib/utils/gmstn";
10
10
import {
11
11
rawDataToString,
12
12
validateWsMessageType,
13
13
} from "@/lib/utils/ws/validate";
14
+
import { z } from "zod";
14
15
15
16
export const connectPreHandler: PreHandler = (req, reply, done) => {
16
17
const { query } = req;
···
82
83
83
84
switch (messageType) {
84
85
case "shard/message": {
85
-
const shardMessage = validateTypeResult.data as ShardMessage;
86
+
const {
87
+
success,
88
+
error,
89
+
data: shardMessage,
90
+
} = shardMessageSchema.safeParse(validateTypeResult.data);
91
+
if (!success) {
92
+
console.error(
93
+
"could not parse",
94
+
validateTypeResult.data,
95
+
"as a valid ShardMessage.",
96
+
);
97
+
console.error(z.treeifyError(error));
98
+
return;
99
+
}
86
100
const { channel } = shardMessage;
101
+
87
102
const atUriParseResult = stringToAtUri(channel);
88
103
if (!atUriParseResult.ok) return;
89
104
const { data: channelAtUri } = atUriParseResult;
90
105
106
+
sendToChannelClients({ channelAtUri, message: shardMessage });
91
107
storeMessageInShard({ channelAtUri, message: shardMessage });
92
108
}
93
109
}
+4
-4
src/lib/sessions.ts
+4
-4
src/lib/sessions.ts
···
78
78
return sessionInfo;
79
79
};
80
80
81
-
export const activeLatticeSessions = new Map<LatticeSessionInfo, WebSocket>();
81
+
export const clientSessions = new Map<LatticeSessionInfo, WebSocket>();
82
82
83
83
export const isValidSession = (sessionInfo: LatticeSessionInfo) => {
84
84
return (
···
99
99
} catch {
100
100
return { ok: false };
101
101
}
102
-
activeLatticeSessions.set(sessionInfo, socket);
102
+
clientSessions.set(sessionInfo, socket);
103
103
return { ok: true, data: { sessionSocket: socket } };
104
104
};
105
105
106
106
export const deleteSession = (
107
107
sessionInfo: LatticeSessionInfo,
108
108
): Result<undefined, undefined> => {
109
-
if (!activeLatticeSessions.has(sessionInfo)) return { ok: false };
109
+
if (!clientSessions.has(sessionInfo)) return { ok: false };
110
110
try {
111
-
activeLatticeSessions.delete(sessionInfo);
111
+
clientSessions.delete(sessionInfo);
112
112
} catch {
113
113
return { ok: false };
114
114
}
+41
-4
src/lib/utils/gmstn.ts
+41
-4
src/lib/utils/gmstn.ts
···
1
+
import { clientSessions } from "@/lib/sessions";
1
2
import { shardSessions } from "@/lib/state";
2
3
import type { AtUri, Did } from "@/lib/types/atproto";
3
4
import type { ShardSessionInfo } from "@/lib/types/handshake";
···
31
32
channelAtUri: AtUri;
32
33
message: ShardMessage;
33
34
}) => {
34
-
const sessionInfo = shardSessions
35
+
const shardSessionInfo = shardSessions
35
36
.keys()
36
37
.find((sessionInfo) =>
37
38
sessionInfo.allowedChannels.some(
38
39
(allowedChannel) => allowedChannel.rKey === channelAtUri.rKey,
39
40
),
40
41
);
41
-
if (!sessionInfo) return;
42
+
if (!shardSessionInfo) return;
42
43
43
-
const shardSocket = shardSessions.get(sessionInfo);
44
+
const shardSocket = shardSessions.get(shardSessionInfo);
44
45
if (!shardSocket) {
45
46
console.error(
46
47
"Could find session info object in map, but socket could not be retrieved from map. Race condition?",
47
48
);
48
49
return;
49
50
}
51
+
const messageToSendToShard = {
52
+
...message,
53
+
sessionToken: shardSessionInfo.token,
54
+
};
50
55
if (shardSocket.readyState === WebSocket.OPEN)
51
-
shardSocket.send(JSON.stringify(message));
56
+
shardSocket.send(JSON.stringify(messageToSendToShard));
52
57
53
58
console.log(
54
59
"Sent off message",
···
57
62
shardSocket.url,
58
63
);
59
64
};
65
+
66
+
export const sendToChannelClients = ({
67
+
channelAtUri,
68
+
message,
69
+
}: {
70
+
channelAtUri: AtUri;
71
+
message: ShardMessage;
72
+
}) => {
73
+
const sessions = clientSessions
74
+
.keys()
75
+
.filter((sessionInfo) =>
76
+
sessionInfo.allowedChannels.some(
77
+
(allowedChannel) => allowedChannel.rKey === channelAtUri.rKey,
78
+
),
79
+
);
80
+
81
+
const clientSockets = sessions
82
+
.map((session) => {
83
+
return clientSessions.get(session);
84
+
})
85
+
.filter((e) => e !== undefined);
86
+
87
+
clientSockets.forEach((clientSocket) => {
88
+
clientSocket.send(JSON.stringify(message));
89
+
console.log(
90
+
"Sent off message",
91
+
message,
92
+
"to clientSocket pointing to",
93
+
clientSocket.url,
94
+
);
95
+
});
96
+
};