decentralised sync engine
feat: websocket sessions yay
+66
-4
+14
-4
src/index.ts
+14
-4
src/index.ts
···
1
1
import { OWNER_DID, SERVER_PORT, SERVICE_DID } from "@/lib/env";
2
-
import { performHandshakes } from "@/lib/setup";
2
+
import { connectToShards, performHandshakes } from "@/lib/setup";
3
3
import { handshakeTokens, setRegistrationState } from "@/lib/state";
4
4
import type { AtUri } from "@/lib/types/atproto";
5
5
import { getRecordFromAtUri } from "@/lib/utils/atproto";
···
89
89
});
90
90
91
91
// TODO: probably move this to an `attachListeners` hook that attaches the listeners we want.
92
+
// least tested. will probably have nuances we need to work on in the future
92
93
attachLatticeRegistrationListener(prismWebsocket);
93
94
94
95
await performHandshakes(latticeAtUri);
96
+
97
+
const test = await connectToShards();
98
+
console.log("connected to", test.map((socket) => socket.url));
95
99
96
100
// TODO: change this to the actual WS sessions
97
101
const handshakeTokenEntries = handshakeTokens.entries().toArray();
98
102
99
103
if (handshakeTokenEntries.length === 0) {
100
-
console.warn("Warning: there are zero handshake tokens on this Lattice.");
101
-
console.warn("If you're hacking locally, you might want to make sure that there's a running Shard as well.")
104
+
console.warn(
105
+
"Warning: there are zero handshake tokens on this Lattice.",
106
+
);
107
+
console.warn(
108
+
"If you're hacking locally, you might want to make sure that there's a running Shard as well.",
109
+
);
102
110
// NOTE: might change in the future
103
-
console.warn("Channel records connecting a Lattice to a Shard are not supported now. Dev lattices must point to a dev Shard if both are running locally")
111
+
console.warn(
112
+
"Channel records connecting a Lattice to a Shard are not supported now. Dev lattices must point to a dev Shard if both are running locally",
113
+
);
104
114
}
105
115
};
106
116
+39
src/lib/setup.ts
+39
src/lib/setup.ts
···
8
8
} from "@/lib/utils/atproto";
9
9
import { getConstellationBacklink } from "@/lib/utils/constellation";
10
10
import { isDomain } from "@/lib/utils/domains";
11
+
import { connectToShard, getShardEndpointFromDid } from "@/lib/utils/gmstn";
11
12
import { initiateHandshakeTo } from "@/lib/utils/handshake";
12
13
13
14
export const performHandshakes = async (latticeAtUri: AtUri) => {
···
103
104
104
105
const channelAtUris = entry[1];
105
106
107
+
// FIXME: perf issue. we are awaiting each handshake to resolve before we make new ones
108
+
// this means that the handshakes are consecutive and not concurrent.
109
+
// stuff this into a Promise.all by mapping over the array instead
106
110
const handshakeResult = await initiateHandshakeTo({
107
111
did: shardDid,
108
112
channels: channelAtUris,
···
113
117
handshakeTokens.set(shardAtUri, sessionInfo);
114
118
}
115
119
};
120
+
121
+
export const connectToShards = async () => {
122
+
const shardSessions = handshakeTokens.entries();
123
+
const shardConnectionPromises = shardSessions
124
+
.map(async (session) => {
125
+
const atUri = session[0];
126
+
const { token } = session[1];
127
+
const rkey = atUri.rKey ?? "";
128
+
const shardDid = isDomain(rkey)
129
+
? `did:web:${encodeURIComponent(rkey)}`
130
+
: `did:plc:${rkey}`;
131
+
132
+
console.log(shardDid);
133
+
134
+
// TODO: again, implement proper did -> endpoint parsing here too.
135
+
// for now, we just assume did:web and construce a URL based on that.
136
+
// @ts-expect-error trust me bro it's a string
137
+
const shardUrlResult = await getShardEndpointFromDid(shardDid);
138
+
139
+
if (!shardUrlResult.ok) return;
140
+
141
+
return {
142
+
// TODO: xrpc and lexicon this endpoint
143
+
shardUrl: `${shardUrlResult.data.origin}/connect`,
144
+
sessionToken: token,
145
+
};
146
+
})
147
+
.toArray();
148
+
149
+
const shardConnectionRequests = await Promise.all(shardConnectionPromises);
150
+
151
+
return shardConnectionRequests
152
+
.filter((request) => request !== undefined)
153
+
.map((request) => connectToShard(request));
154
+
};
+13
src/lib/utils/gmstn.ts
+13
src/lib/utils/gmstn.ts
···
1
1
import type { Did } from "@/lib/types/atproto";
2
2
import { getEndpointFromDid } from "@/lib/utils/atproto";
3
+
import WebSocket from "ws";
3
4
4
5
export const getShardEndpointFromDid = async (did: Did) => {
5
6
return await getEndpointFromDid(did, "GemstoneShard");
6
7
};
8
+
9
+
export const connectToShard = ({
10
+
shardUrl,
11
+
sessionToken,
12
+
}: {
13
+
shardUrl: string;
14
+
sessionToken: string;
15
+
}) => {
16
+
const endpoint = new URL(shardUrl);
17
+
endpoint.searchParams.append("token", sessionToken);
18
+
return new WebSocket(endpoint);
19
+
};