decentralised sync engine
feat: client lattice connection
+117
-17
+2
src/index.ts
+2
src/index.ts
···
30
30
wsHandler,
31
31
method,
32
32
handler: httpHandler,
33
+
preHandler,
33
34
skipRegistrationCheckHttp,
34
35
skipRegistrationCheckWs,
35
36
} = route;
···
49
50
wsHandler: skipRegistrationCheckWs
50
51
? wsHandler
51
52
: wrapWsRegistrationCheck(wsHandler),
53
+
preHandler,
52
54
});
53
55
}
54
56
}
+72
src/lib/handlers/connect.ts
+72
src/lib/handlers/connect.ts
···
1
+
import {
2
+
createNewSession,
3
+
issuedLatticeTokens,
4
+
isValidSession,
5
+
} from "@/lib/sessions";
6
+
import type { PreHandler, WsRouteHandler } from "@/lib/types/routes";
7
+
import { rawDataToString } from "@/lib/utils/ws";
8
+
9
+
export const connectPreHandler: PreHandler = (req, reply, done) => {
10
+
const { query } = req;
11
+
if (!query) return;
12
+
if (!(typeof query === "object" && "token" in query)) {
13
+
reply.code(400).send("Provide token in query params");
14
+
return;
15
+
}
16
+
17
+
const sessionToken = query.token as string;
18
+
19
+
const sessionInfo = issuedLatticeTokens.get(sessionToken);
20
+
if (!sessionInfo) {
21
+
reply
22
+
.code(404)
23
+
.send(
24
+
"Session token could not resolve to existing session. retry?",
25
+
);
26
+
return;
27
+
}
28
+
29
+
if (!isValidSession(sessionInfo)) {
30
+
reply
31
+
.code(403)
32
+
.send(
33
+
"Session token resolved to session, but did not pass verification. this should not happen.",
34
+
);
35
+
return;
36
+
}
37
+
38
+
console.log(
39
+
"Found session:",
40
+
sessionInfo.id,
41
+
"from session token",
42
+
sessionToken,
43
+
);
44
+
done();
45
+
};
46
+
47
+
export const connectWsHandler: WsRouteHandler = (socket, req) => {
48
+
const { query } = req;
49
+
if (!query) return;
50
+
if (!(typeof query === "object" && "token" in query)) {
51
+
socket.close();
52
+
return;
53
+
}
54
+
const sessionToken = query.token as string;
55
+
56
+
const sessionInfo = issuedLatticeTokens.get(sessionToken);
57
+
if (!sessionInfo) {
58
+
socket.close();
59
+
return;
60
+
}
61
+
62
+
const sessionCreateResult = createNewSession({ sessionInfo, socket });
63
+
if (!sessionCreateResult.ok) {
64
+
socket.close();
65
+
return;
66
+
}
67
+
68
+
socket.on("message", (event) => {
69
+
const message = rawDataToString(event);
70
+
console.log(message)
71
+
})
72
+
};
+34
-17
src/lib/types/messages.ts
+34
-17
src/lib/types/messages.ts
···
1
-
import { didPlcSchema } from "@/lib/types/atproto";
1
+
import { didSchema } from "@/lib/types/atproto";
2
2
import { z } from "zod";
3
3
4
-
export const websocketMessageSchema = z.object({
5
-
type: z.union([z.literal("shard/message"), z.literal("shard/history")]),
6
-
});
7
-
4
+
export const websocketMessageSchema = z
5
+
.object({
6
+
type: z.union([
7
+
z.literal("shard/message"),
8
+
z.literal("shard/history"),
9
+
z.literal("shard/requestHistory"),
10
+
]),
11
+
})
12
+
.loose();
8
13
export type WebsocketMessage = z.infer<typeof websocketMessageSchema>;
9
14
10
-
export const shardMessageSchema = websocketMessageSchema.extend({
11
-
type: z.literal("shard/message"),
12
-
text: z.string(),
13
-
did: didPlcSchema,
14
-
timestamp: z.coerce.date(),
15
-
});
16
-
15
+
export const shardMessageSchema = websocketMessageSchema
16
+
.safeExtend({
17
+
type: z.literal("shard/message"),
18
+
channel: z.string(),
19
+
content: z.string(),
20
+
sentBy: didSchema,
21
+
sentAt: z.coerce.date(),
22
+
})
23
+
.strict();
17
24
export type ShardMessage = z.infer<typeof shardMessageSchema>;
18
25
19
-
export const historyMessageSchema = websocketMessageSchema.extend({
20
-
type: z.literal("shard/history"),
21
-
messages: z.optional(z.array(shardMessageSchema)),
22
-
});
26
+
export const historyMessageSchema = websocketMessageSchema
27
+
.safeExtend({
28
+
type: z.literal("shard/history"),
29
+
messages: z.optional(z.array(shardMessageSchema)),
30
+
channel: z.string(),
31
+
})
32
+
.strict();
33
+
export type HistoryMessage = z.infer<typeof historyMessageSchema>;
23
34
24
-
export type HistoryMessage = z.infer<typeof historyMessageSchema>;
35
+
export const requestHistoryMessageSchema = websocketMessageSchema
36
+
.safeExtend({
37
+
type: z.literal("shard/requestHistory"),
38
+
channel: z.string(),
39
+
})
40
+
.strict();
41
+
export type RequestHistoryMessage = z.infer<typeof requestHistoryMessageSchema>;
+7
src/routes/connect/route.ts
+7
src/routes/connect/route.ts
+2
src/routes/index.ts
+2
src/routes/index.ts
···
1
1
import type { Route, WsRoute } from "@/lib/types/routes";
2
+
import { connectRoute } from "@/routes/connect/route";
2
3
import { didWebDocRoute } from "@/routes/dot-well-known/did-dot-json/route";
3
4
import { handshakeRoute } from "@/routes/handshake/route";
4
5
import { indexRoute } from "@/routes/route";
···
7
8
"/": indexRoute,
8
9
"/.well-known/did.json": didWebDocRoute,
9
10
"/handshake": handshakeRoute,
11
+
"/connect": connectRoute,
10
12
};