A work-in-progress chat bot for Streamplace with chat overlay functionality
Initial working websocket and ChatOverlay component
+343
-295
+2
deno.json
+2
deno.json
···
12
12
"imports": {
13
13
"@atcute/atproto": "npm:@atcute/atproto@^3.1.4",
14
14
"@atcute/client": "npm:@atcute/client@^4.0.3",
15
+
"@atcute/identity-resolver": "npm:@atcute/identity-resolver@^1.1.3",
16
+
"@atcute/lexicons": "npm:@atcute/lexicons@^1.1.1",
15
17
"@std/dotenv": "jsr:@std/dotenv@^0.225.5",
16
18
"fresh": "jsr:@fresh/core@^2.1.1",
17
19
"preact": "npm:preact@^10.27.2",
+19
deno.lock
+19
deno.lock
···
23
23
"jsr:@std/uuid@^1.0.7": "1.0.9",
24
24
"npm:@atcute/atproto@^3.1.4": "3.1.4",
25
25
"npm:@atcute/client@^4.0.3": "4.0.3",
26
+
"npm:@atcute/identity-resolver@^1.1.3": "1.1.3_@atcute+identity@1.1.0",
27
+
"npm:@atcute/lexicons@^1.1.1": "1.1.1",
26
28
"npm:@opentelemetry/api@^1.9.0": "1.9.0",
27
29
"npm:@preact/signals@^2.2.1": "2.3.1_preact@10.27.2",
28
30
"npm:@preact/signals@^2.3.1": "2.3.1_preact@10.27.2",
···
147
149
"@atcute/lexicons"
148
150
]
149
151
},
152
+
"@atcute/identity-resolver@1.1.3_@atcute+identity@1.1.0": {
153
+
"integrity": "sha512-KZgGgg99CWaV7Df3+h3X/WMrDzTPQVfsaoIVbTNLx2B56BvCL2EmaxPSVw/7BFUJMZHlVU4rtoEB4lyvNyMswA==",
154
+
"dependencies": [
155
+
"@atcute/identity",
156
+
"@atcute/lexicons",
157
+
"@atcute/util-fetch",
158
+
"@badrap/valita"
159
+
]
160
+
},
150
161
"@atcute/identity@1.1.0": {
151
162
"integrity": "sha512-6vRvRqJatDB+JUQsb+UswYmtBGQnSZcqC3a2y6H5DB/v5KcIh+6nFFtc17G0+3W9rxdk7k9M4KkgkdKf/YDNoQ==",
152
163
"dependencies": [
···
158
169
"integrity": "sha512-k6qy5p3j9fJJ6ekaMPfEfp3ni4TW/XNuH9ZmsuwC0fi0tOjp+Fa8ZQakHwnqOzFt/cVBfGcmYE/lKNAbeTjgUg==",
159
170
"dependencies": [
160
171
"esm-env"
172
+
]
173
+
},
174
+
"@atcute/util-fetch@1.0.2": {
175
+
"integrity": "sha512-6kOwQzaoPu7ss0EwvgSOPVg/JIJdozbXva5d/g99bBg8RaSX8se//INOyr/nby/GwrpxCIZU9Wzm+pl7xerI0Q==",
176
+
"dependencies": [
177
+
"@badrap/valita"
161
178
]
162
179
},
163
180
"@badrap/valita@0.4.6": {
···
371
388
"jsr:@std/dotenv@~0.225.5",
372
389
"npm:@atcute/atproto@^3.1.4",
373
390
"npm:@atcute/client@^4.0.3",
391
+
"npm:@atcute/identity-resolver@^1.1.3",
392
+
"npm:@atcute/lexicons@^1.1.1",
374
393
"npm:@preact/signals@^2.3.1",
375
394
"npm:preact@^10.27.2"
376
395
]
+19
-22
islands/ChatOverlay.tsx
+19
-22
islands/ChatOverlay.tsx
···
1
1
import { useEffect, useRef, useState } from "preact/hooks";
2
2
3
-
interface MessageLabel {
4
-
type: string;
5
-
value: string;
6
-
}
7
-
8
3
interface ChatOverlayProps {
9
4
wsUrl: string;
10
-
streamerDid: string;
5
+
streamerHandle: Handle;
11
6
maxMessages?: number;
12
7
}
13
8
14
9
export function ChatOverlay({
15
10
wsUrl,
16
-
streamerDid,
11
+
streamerHandle,
17
12
maxMessages = 50,
18
13
}: ChatOverlayProps) {
19
14
const [messages, setMessages] = useState<EnrichedChatMessage[]>([]);
20
15
const [connected, setConnected] = useState(false);
16
+
const [isClient, setIsClient] = useState(false);
21
17
const wsRef = useRef<WebSocket | null>(null);
22
18
const messagesEndRef = useRef<HTMLDivElement>(null);
23
19
20
+
// This effect runs only on the client after hydration
24
21
useEffect(() => {
22
+
setIsClient(true);
23
+
}, []);
24
+
25
+
// This effect runs only when we're on the client
26
+
useEffect(() => {
27
+
if (!isClient) {
28
+
return;
29
+
}
30
+
25
31
const ws = new WebSocket(wsUrl);
26
32
wsRef.current = ws;
27
33
28
34
ws.onopen = () => {
29
-
console.log("Connected to WebSocket");
35
+
console.log(`🟢 WebSocket CONNECTED to: ${wsUrl}`);
30
36
setConnected(true);
31
-
32
-
// Subscribe to streamer
33
-
ws.send(JSON.stringify({
34
-
type: "subscribe",
35
-
streamer: streamerDid,
36
-
}));
37
37
};
38
38
39
39
ws.onmessage = (event) => {
···
48
48
49
49
setMessages((prev) => {
50
50
const newMessages = [...prev, message];
51
-
// Keep only last N messages
52
51
return newMessages.slice(-maxMessages);
53
52
});
54
53
}
···
68
67
69
68
return () => {
70
69
if (ws.readyState === WebSocket.OPEN) {
71
-
ws.send(JSON.stringify({
72
-
type: "unsubscribe",
73
-
streamer: streamerDid,
74
-
}));
70
+
ws.close();
75
71
}
76
-
ws.close();
77
72
};
78
-
}, [wsUrl, streamerDid]);
73
+
}, [isClient, wsUrl, streamerHandle, maxMessages]);
79
74
80
75
// Auto-scroll to bottom on new messages
81
76
useEffect(() => {
82
-
messagesEndRef.current?.scrollIntoView({ behavior: "smooth" });
77
+
if (messages.length > 0) {
78
+
messagesEndRef.current?.scrollIntoView({ behavior: "smooth" });
79
+
}
83
80
}, [messages]);
84
81
85
82
return (
-79
main.ts
-79
main.ts
···
1
1
import { App, staticFiles } from "fresh";
2
2
import { type State } from "./utils.ts";
3
-
import { StreamplaceWebSocketService } from "./utils/websocket.ts";
4
3
5
4
export const app = new App<State>();
6
5
7
6
app.use(staticFiles());
8
-
9
-
// Start WebSocket service
10
-
// const wsService = new StreamplaceWebSocketService(8080);
11
-
// wsService.start();
12
-
// console.log("WebSocket service started on port 8080");
13
-
14
-
const clients = new Map<WebSocket, Set<string>>();
15
-
const streamerClients = new Map<string, Set<WebSocket>>();
16
-
let jetstreamWs: WebSocket | null = null;
17
-
18
-
function connectToJetstream() {
19
-
const jetstreamUrl = "wss://jetstream1.us-east.bsky.network/subscribe?wantedCollections=place.stream.chat.message";
20
-
jetstreamWs = new WebSocket(jetstreamUrl);
21
-
22
-
jetstreamWs.onopen = () => {
23
-
console.log("Connected to Jetstream");
24
-
};
25
-
26
-
jetstreamWs.onmessage = (event) => {
27
-
try {
28
-
const jetstreamMessage = JSON.parse(event.data);
29
-
30
-
if (
31
-
jetstreamMessage.kind === "commit" &&
32
-
jetstreamMessage.commit?.operation === "create" &&
33
-
jetstreamMessage.commit.record
34
-
) {
35
-
console.log(jetstreamMessage)
36
-
const record = jetstreamMessage.commit.record;
37
-
const enrichedMessage = {
38
-
id: jetstreamMessage.commit.cid,
39
-
text: record.text,
40
-
author: {
41
-
did: jetstreamMessage.did,
42
-
handle: jetstreamMessage.did,
43
-
},
44
-
timestamp: new Date(record.createdAt),
45
-
color: { red: 128, green: 128, blue: 128 },
46
-
facets: record.facets,
47
-
isReply: !!record.reply,
48
-
streamer: record.streamer,
49
-
};
50
-
51
-
const streamer = record.streamer;
52
-
const subscribedClients = streamerClients.get(streamer);
53
-
54
-
if (subscribedClients && subscribedClients.size > 0) {
55
-
const messageJson = JSON.stringify({
56
-
type: "chat_message",
57
-
data: enrichedMessage,
58
-
});
59
-
60
-
for (const client of subscribedClients) {
61
-
if (client.readyState === WebSocket.OPEN) {
62
-
client.send(messageJson);
63
-
}
64
-
}
65
-
66
-
console.log(
67
-
`Sent message to ${subscribedClients.size} clients for streamer ${streamer}`,
68
-
);
69
-
}
70
-
}
71
-
} catch (error) {
72
-
console.error("Error processing jetstream message:", error);
73
-
}
74
-
};
75
-
76
-
jetstreamWs.onclose = () => {
77
-
console.log("Jetstream connection closed, attempting to reconnect...");
78
-
setTimeout(() => connectToJetstream(), 5000);
79
-
};
80
-
}
81
-
82
-
connectToJetstream();
83
-
84
-
// Export these so the ws route can access them
85
-
export { clients, streamerClients };
86
7
87
8
// Include file-system based routes here
88
9
app.fsRoutes();
-17
routes/[streamerDid].tsx
-17
routes/[streamerDid].tsx
···
1
-
import { PageProps } from "fresh";
2
-
import { ChatOverlay } from "../islands/ChatOverlay.tsx";
3
-
4
-
export default function ChatPage(props: PageProps) {
5
-
const streamerDid = props.params.streamerDid;
6
-
7
-
return (
8
-
<div>
9
-
<h1>Chat Overlay</h1>
10
-
<ChatOverlay
11
-
wsUrl="ws://localhost:8000/ws"
12
-
streamerDid={streamerDid}
13
-
maxMessages={50}
14
-
/>
15
-
</div>
16
-
);
17
-
}
+58
routes/api/websocket/[streamerHandle].ts
+58
routes/api/websocket/[streamerHandle].ts
···
1
+
import { define } from "../../../utils.ts";
2
+
import { StreamplaceWebSocketService } from "../../../utils/websocket.ts";
3
+
import { resolveHandle } from "../../../utils/atcuteUtils.ts";
4
+
5
+
// Global WebSocket service instance
6
+
let wsService: StreamplaceWebSocketService | null = null;
7
+
8
+
function getWebSocketService() {
9
+
if (!wsService) {
10
+
wsService = new StreamplaceWebSocketService();
11
+
wsService.start();
12
+
}
13
+
return wsService;
14
+
}
15
+
16
+
export const handler = define.handlers({
17
+
async GET(ctx) {
18
+
const { streamerHandle } = ctx.params;
19
+
20
+
if (ctx.req.headers.get("upgrade") !== "websocket") {
21
+
return new Response("WebSocket connections only", { status: 426 });
22
+
}
23
+
24
+
const { socket, response } = Deno.upgradeWebSocket(ctx.req);
25
+
const wsService = getWebSocketService();
26
+
const streamerDid = await resolveHandle(streamerHandle as Handle);
27
+
28
+
socket.onopen = () => {
29
+
console.log(`New client connected for streamer: ${streamerHandle}`);
30
+
// Store the streamerDid with the socket
31
+
(socket as any).streamerDid = streamerDid;
32
+
wsService.handleNewConnection(socket, streamerDid);
33
+
};
34
+
35
+
socket.onmessage = (event) => {
36
+
try {
37
+
const message = JSON.parse(event.data);
38
+
wsService.handleClientMessage(socket, message);
39
+
} catch (error) {
40
+
console.error("Error handling client message:", error);
41
+
socket.send(
42
+
JSON.stringify({ error: "Invalid message format" }),
43
+
);
44
+
}
45
+
};
46
+
47
+
socket.onclose = () => {
48
+
console.log(`Client disconnected for streamer: ${streamerDid}`);
49
+
wsService.removeClient(socket);
50
+
};
51
+
52
+
socket.onerror = (error) => {
53
+
console.error("WebSocket error:", error);
54
+
};
55
+
56
+
return response;
57
+
},
58
+
});
+21
routes/chat/[streamerHandle].tsx
+21
routes/chat/[streamerHandle].tsx
···
1
+
import { PageProps } from "fresh";
2
+
import { ChatOverlay } from "../../islands/ChatOverlay.tsx";
3
+
4
+
export default function ChatPage(props: PageProps) {
5
+
const { streamerHandle } = props.params;
6
+
const wsUrl = typeof window !== "undefined"
7
+
? `${
8
+
location.protocol === "https:" ? "wss:" : "ws:"
9
+
}//${location.host}/api/websocket/${streamerHandle}`
10
+
: `ws://localhost:8000/api/websocket/${streamerHandle}`;
11
+
12
+
return (
13
+
<div>
14
+
<ChatOverlay
15
+
wsUrl={wsUrl}
16
+
streamerHandle={streamerHandle as Handle}
17
+
maxMessages={50}
18
+
/>
19
+
</div>
20
+
);
21
+
}
-84
routes/ws.ts
-84
routes/ws.ts
···
1
-
import { define } from "../utils.ts";
2
-
import { clients, streamerClients } from "../main.ts";
3
-
4
-
export const handler = define.handlers({
5
-
GET(ctx) {
6
-
const { socket, response } = Deno.upgradeWebSocket(ctx.req);
7
-
8
-
socket.onopen = () => {
9
-
console.log("New client connected");
10
-
clients.set(socket, new Set());
11
-
};
12
-
13
-
socket.onmessage = (event) => {
14
-
try {
15
-
const message = JSON.parse(event.data);
16
-
const clientStreamers = clients.get(socket);
17
-
if (!clientStreamers) return;
18
-
19
-
switch (message.type) {
20
-
case "subscribe": {
21
-
clientStreamers.add(message.streamer);
22
-
if (!streamerClients.has(message.streamer)) {
23
-
streamerClients.set(message.streamer, new Set());
24
-
}
25
-
streamerClients.get(message.streamer)!.add(socket);
26
-
console.log(
27
-
`Client subscribed to streamer: ${message.streamer}`,
28
-
);
29
-
socket.send(JSON.stringify({
30
-
type: "subscribed",
31
-
streamer: message.streamer,
32
-
}));
33
-
break;
34
-
}
35
-
36
-
case "unsubscribe": {
37
-
clientStreamers.delete(message.streamer);
38
-
const streamerClientSet = streamerClients.get(
39
-
message.streamer,
40
-
);
41
-
if (streamerClientSet) {
42
-
streamerClientSet.delete(socket);
43
-
if (streamerClientSet.size === 0) {
44
-
streamerClients.delete(message.streamer);
45
-
}
46
-
}
47
-
console.log(
48
-
`Client unsubscribed from streamer: ${message.streamer}`,
49
-
);
50
-
socket.send(JSON.stringify({
51
-
type: "unsubscribed",
52
-
streamer: message.streamer,
53
-
}));
54
-
break;
55
-
}
56
-
}
57
-
} catch (error) {
58
-
console.error("Error handling client message:", error);
59
-
socket.send(
60
-
JSON.stringify({ error: "Invalid message format" }),
61
-
);
62
-
}
63
-
};
64
-
65
-
socket.onclose = () => {
66
-
console.log("Client disconnected");
67
-
const clientStreamers = clients.get(socket);
68
-
if (clientStreamers) {
69
-
for (const streamer of clientStreamers) {
70
-
const streamerClientSet = streamerClients.get(streamer);
71
-
if (streamerClientSet) {
72
-
streamerClientSet.delete(socket);
73
-
if (streamerClientSet.size === 0) {
74
-
streamerClients.delete(streamer);
75
-
}
76
-
}
77
-
}
78
-
}
79
-
clients.delete(socket);
80
-
};
81
-
82
-
return response;
83
-
},
84
-
});
-19
static/logo.svg
-19
static/logo.svg
···
1
-
<svg width="40" height="40" fill="none" xmlns="http://www.w3.org/2000/svg">
2
-
<path
3
-
d="M34.092 8.845C38.929 20.652 34.092 27 30 30.5c1 3.5-2.986 4.222-4.5 2.5-4.457 1.537-13.512 1.487-20-5C2 24.5 4.73 16.714 14 11.5c8-4.5 16-7 20.092-2.655Z"
4
-
fill="#FFDB1E"
5
-
/>
6
-
<path
7
-
d="M14 11.5c6.848-4.497 15.025-6.38 18.368-3.47C37.5 12.5 21.5 22.612 15.5 25c-6.5 2.587-3 8.5-6.5 8.5-3 0-2.5-4-5.183-7.75C2.232 23.535 6.16 16.648 14 11.5Z"
8
-
fill="#fff"
9
-
stroke="#FFDB1E"
10
-
/>
11
-
<path
12
-
d="M28.535 8.772c4.645 1.25-.365 5.695-4.303 8.536-3.732 2.692-6.606 4.21-7.923 4.83-.366.173-1.617-2.252-1.617-1 0 .417-.7 2.238-.934 2.326-1.365.512-4.223 1.29-5.835 1.29-3.491 0-1.923-4.754 3.014-9.122.892-.789 1.478-.645 2.283-.645-.537-.773-.534-.917.403-1.546C17.79 10.64 23 8.77 25.212 8.42c.366.014.82.35.82.629.41-.14 2.095-.388 2.503-.278Z"
13
-
fill="#FFE600"
14
-
/>
15
-
<path
16
-
d="M14.297 16.49c.985-.747 1.644-1.01 2.099-2.526.566.121.841-.08 1.29-.701.324.466 1.657.608 2.453.701-.715.451-1.057.852-1.452 2.106-1.464-.611-3.167-.302-4.39.42Z"
17
-
fill="#fff"
18
-
/>
19
-
</svg>
+1
-1
streamplaceBot.ts
utils/streamplaceBot.ts
+1
-1
streamplaceBot.ts
utils/streamplaceBot.ts
+153
utils/didResolver.ts
+153
utils/didResolver.ts
···
1
+
interface PlcDirectoryResponse {
2
+
"@context": string[];
3
+
id: Did;
4
+
alsoKnownAs?: string[];
5
+
verificationMethod: unknown[];
6
+
service: Array<{
7
+
id: string;
8
+
type: string;
9
+
serviceEndpoint: string;
10
+
}>;
11
+
}
12
+
13
+
interface ChatProfileResponse {
14
+
uri: string;
15
+
cid: string;
16
+
value: {
17
+
$type: "place.stream.chat.profile";
18
+
color: {
19
+
red: number;
20
+
green: number;
21
+
blue: number;
22
+
};
23
+
};
24
+
}
25
+
26
+
interface UserProfile {
27
+
handle: Handle;
28
+
pdsEndpoint: string;
29
+
color?: {
30
+
red: number;
31
+
green: number;
32
+
blue: number;
33
+
};
34
+
}
35
+
36
+
const DEFAULT_COLOR = {
37
+
red: 128,
38
+
green: 128,
39
+
blue: 128,
40
+
};
41
+
42
+
class DidToHandleResolver {
43
+
private cache = new Map<Did, UserProfile>();
44
+
private pendingRequests = new Map<Did, Promise<UserProfile>>();
45
+
private readonly plcDirectoryUrl = "https://plc.directory";
46
+
47
+
async resolve(did: Did): Promise<UserProfile> {
48
+
// Check cache first
49
+
const cached = this.cache.get(did);
50
+
if (cached) {
51
+
return cached;
52
+
}
53
+
54
+
// Check if there's already a pending request for this DID
55
+
const pending = this.pendingRequests.get(did);
56
+
if (pending) {
57
+
return pending;
58
+
}
59
+
60
+
// Create new request
61
+
const request = this.fetchProfile(did);
62
+
this.pendingRequests.set(did, request);
63
+
64
+
try {
65
+
const profile = await request;
66
+
this.cache.set(did, profile);
67
+
return profile;
68
+
} finally {
69
+
this.pendingRequests.delete(did);
70
+
}
71
+
}
72
+
73
+
private async fetchProfile(did: Did): Promise<UserProfile> {
74
+
// First, get the DID document from PLC directory
75
+
const plcResponse = await fetch(
76
+
`${this.plcDirectoryUrl}/${encodeURIComponent(did)}`,
77
+
);
78
+
79
+
if (!plcResponse.ok) {
80
+
throw new Error(
81
+
`Failed to resolve DID ${did}: ${plcResponse.statusText}`,
82
+
);
83
+
}
84
+
85
+
const plcData: PlcDirectoryResponse = await plcResponse.json();
86
+
87
+
// Extract handle from alsoKnownAs
88
+
if (!plcData.alsoKnownAs || plcData.alsoKnownAs.length === 0) {
89
+
throw new Error(`No handle found for DID ${did}`);
90
+
}
91
+
92
+
const handle = plcData.alsoKnownAs[0].replace(/^at:\/\//, "") as Handle;
93
+
94
+
// Find the PDS endpoint
95
+
const pdsService = plcData.service.find((s) =>
96
+
s.type === "AtprotoPersonalDataServer"
97
+
);
98
+
99
+
if (!pdsService) {
100
+
throw new Error(`No PDS service found for DID ${did}`);
101
+
}
102
+
103
+
const pdsEndpoint = pdsService.serviceEndpoint;
104
+
105
+
// Now fetch the chat profile
106
+
let color = DEFAULT_COLOR;
107
+
108
+
try {
109
+
const profileUrl = new URL(
110
+
`${pdsEndpoint}/xrpc/com.atproto.repo.getRecord`,
111
+
);
112
+
profileUrl.searchParams.set("repo", did);
113
+
profileUrl.searchParams.set(
114
+
"collection",
115
+
"place.stream.chat.profile",
116
+
);
117
+
profileUrl.searchParams.set("rkey", "self");
118
+
119
+
const profileResponse = await fetch(profileUrl.toString());
120
+
121
+
if (profileResponse.ok) {
122
+
const profileData: ChatProfileResponse = await profileResponse
123
+
.json();
124
+
if (profileData.value?.color) {
125
+
color = profileData.value.color;
126
+
}
127
+
}
128
+
// If profile doesn't exist or fetch fails, we'll use the default color
129
+
} catch (error) {
130
+
console.warn(`Failed to fetch chat profile for ${did}:`, error);
131
+
// Continue with default color
132
+
}
133
+
134
+
return {
135
+
handle,
136
+
pdsEndpoint,
137
+
color,
138
+
};
139
+
}
140
+
141
+
// Optional: Clear cache if needed
142
+
clearCache() {
143
+
this.cache.clear();
144
+
}
145
+
146
+
// Optional: Get cache size for monitoring
147
+
getCacheSize(): number {
148
+
return this.cache.size;
149
+
}
150
+
}
151
+
152
+
// Export singleton instance
153
+
export const didResolver = new DidToHandleResolver();
+11
-8
utils/globals.d.ts
+11
-8
utils/globals.d.ts
···
1
+
declare type Did = import("@atcute/lexicons").Did;
2
+
declare type Handle = import("@atcute/lexicons").Handle;
3
+
1
4
// Raw chat message as received from stream.place WebSocket
2
5
interface RawChatMessage {
3
6
$type: "place.stream.chat.defs#messageView";
4
7
author: {
5
-
did: string;
6
-
handle: string;
8
+
did: Did;
9
+
handle: Handle;
7
10
};
8
11
chatProfile: {
9
12
$type: "place.stream.chat.profile";
···
37
40
38
41
// Raw chat message as received from Jetstream WebSocket
39
42
interface JetstreamMessage {
40
-
did: string;
43
+
did: Did;
41
44
time_us: number;
42
45
kind: "commit";
43
46
commit: {
···
48
51
record?: {
49
52
$type: "place.stream.chat.message";
50
53
createdAt: string;
51
-
streamer: string; // This is a DID
54
+
streamer: Did;
52
55
text: string;
53
56
facets?: Array<{
54
57
features: unknown[];
···
79
82
id: string;
80
83
text: string;
81
84
author: {
82
-
did: string;
83
-
handle: string;
84
-
displayName?: string;
85
+
did: Did;
86
+
handle: Handle;
87
+
displayName?: string; // Streamplace doesn't have display names yet but might one day
85
88
};
86
-
streamer: string;
89
+
streamer: Did;
87
90
timestamp: Date; // Parsed createdAt
88
91
color: {
89
92
red: number;
+26
utils/streamplaceUtils.ts
utils/atcuteUtils.ts
+26
utils/streamplaceUtils.ts
utils/atcuteUtils.ts
···
1
1
import { Client, CredentialManager, ok } from "@atcute/client";
2
2
import type {} from "@atcute/atproto";
3
+
import {
4
+
CompositeHandleResolver,
5
+
DohJsonHandleResolver,
6
+
WellKnownHandleResolver,
7
+
} from "@atcute/identity-resolver";
8
+
import { Handle } from "@atcute/lexicons";
9
+
import { isHandle } from "@atcute/lexicons/syntax";
3
10
import { ATPROTO_PASSWORD, ATPROTO_USERNAME, PDS_HOST_URL } from "../env.ts";
4
11
5
12
// Store credential manager singleton to avoid multiple logins
···
67
74
},
68
75
}));
69
76
}
77
+
78
+
// Handle resolver
79
+
const handleResolver = new CompositeHandleResolver({
80
+
strategy: "dns-first",
81
+
methods: {
82
+
dns: new DohJsonHandleResolver({
83
+
dohUrl: "https://dns.google/resolve?",
84
+
}),
85
+
http: new WellKnownHandleResolver(),
86
+
},
87
+
});
88
+
89
+
export async function resolveHandle(handle: Handle) {
90
+
if (!isHandle(handle)) {
91
+
throw new Error("Not a valid handle");
92
+
}
93
+
94
+
return await handleResolver.resolve(handle);
95
+
}
+33
-65
utils/websocket.ts
+33
-65
utils/websocket.ts
···
1
1
import { JETSTREAM_URL } from "../env.ts";
2
+
import { didResolver } from "./didResolver.ts";
2
3
3
4
// Client subscription message
4
5
interface SubscriptionMessage {
···
12
13
private clients = new Map<WebSocket, Set<string>>(); // client -> subscribed streamers
13
14
private streamerClients = new Map<string, Set<WebSocket>>(); // streamer -> clients
14
15
15
-
constructor(private port: number = 8080) {}
16
+
constructor() {}
16
17
17
18
start() {
18
-
console.log(`Starting WebSocket service on port ${this.port}`);
19
-
20
-
// Connect to Jetstream
19
+
console.log("Starting WebSocket service");
21
20
this.connectToJetstream();
21
+
}
22
+
23
+
// Handle new connection with pre-defined streamer DID
24
+
handleNewConnection(client: WebSocket, streamerDid: Did) {
25
+
// Initialize client with empty subscriptions
26
+
this.clients.set(client, new Set());
22
27
23
-
// Start HTTP server for WebSocket upgrades
24
-
this.startWebSocketServer();
28
+
// Auto-subscribe to the streamer from the route parameter
29
+
this.handleClientMessage(client, {
30
+
type: "subscribe",
31
+
streamer: streamerDid,
32
+
});
25
33
}
26
34
27
35
private connectToJetstream() {
···
46
54
jetstreamMessage.commit?.operation === "create" &&
47
55
jetstreamMessage.commit.record
48
56
) {
49
-
console.log(jetstreamMessage); //debug
50
57
this.processJetstreamMessage(jetstreamMessage);
51
58
}
52
59
} catch (error) {
···
66
73
};
67
74
}
68
75
69
-
private startWebSocketServer() {
70
-
Deno.serve({ port: this.port }, (req) => {
71
-
const { socket, response } = Deno.upgradeWebSocket(req);
72
-
73
-
socket.onopen = () => {
74
-
console.log("New client connected");
75
-
this.clients.set(socket, new Set());
76
-
};
77
-
78
-
socket.onmessage = (event) => {
79
-
try {
80
-
const message: SubscriptionMessage = JSON.parse(event.data);
81
-
this.handleClientMessage(socket, message);
82
-
} catch (error) {
83
-
console.error("Error handling client message:", error);
84
-
socket.send(
85
-
JSON.stringify({ error: "Invalid message format" }),
86
-
);
87
-
}
88
-
};
89
-
90
-
socket.onclose = () => {
91
-
console.log("Client disconnected");
92
-
this.removeClient(socket);
93
-
};
94
-
95
-
return response;
96
-
});
97
-
}
98
-
99
-
private handleClientMessage(
100
-
client: WebSocket,
101
-
message: SubscriptionMessage,
102
-
) {
76
+
handleClientMessage(client: WebSocket, message: SubscriptionMessage) {
103
77
const clientStreamers = this.clients.get(client);
104
78
if (!clientStreamers) return;
105
79
106
80
switch (message.type) {
107
-
case "subscribe": { // Add streamer to client's subscription list
81
+
case "subscribe": {
108
82
clientStreamers.add(message.streamer);
109
83
110
-
// Add client to streamer's client list
111
84
if (!this.streamerClients.has(message.streamer)) {
112
85
this.streamerClients.set(message.streamer, new Set());
113
86
}
···
123
96
break;
124
97
}
125
98
126
-
case "unsubscribe": { // Remove streamer from client's subscription list
99
+
case "unsubscribe": {
127
100
clientStreamers.delete(message.streamer);
128
101
129
-
// Remove client from streamer's client list
130
102
const streamerClientSet = this.streamerClients.get(
131
103
message.streamer,
132
104
);
···
149
121
}
150
122
}
151
123
152
-
private removeClient(client: WebSocket) {
124
+
removeClient(client: WebSocket) {
153
125
const clientStreamers = this.clients.get(client);
154
126
if (clientStreamers) {
155
-
// Remove client from all streamer client sets
156
127
for (const streamer of clientStreamers) {
157
128
const streamerClientSet = this.streamerClients.get(streamer);
158
129
if (streamerClientSet) {
···
166
137
this.clients.delete(client);
167
138
}
168
139
169
-
private processJetstreamMessage(jetstreamMessage: JetstreamMessage) {
140
+
private async processJetstreamMessage(jetstreamMessage: JetstreamMessage) {
170
141
// Extract the record data
171
142
const record = jetstreamMessage.commit.record!;
172
-
const enrichedMessage = this.enrichMessage(jetstreamMessage, record);
143
+
const enrichedMessage = await this.enrichMessage(
144
+
jetstreamMessage,
145
+
record,
146
+
);
173
147
174
148
// Get clients subscribed to this streamer
175
-
const streamer = record.streamer; // This is a DID
149
+
const streamer = record.streamer;
176
150
const subscribedClients = this.streamerClients.get(streamer);
177
151
178
152
if (subscribedClients && subscribedClients.size > 0) {
···
187
161
client.send(messageJson);
188
162
}
189
163
}
190
-
191
-
console.log(
192
-
`Sent message to ${subscribedClients.size} clients for streamer ${streamer}`,
193
-
);
194
164
}
195
165
}
196
166
197
-
private enrichMessage(
167
+
private async enrichMessage(
198
168
jetstreamMessage: JetstreamMessage,
199
169
record: NonNullable<JetstreamMessage["commit"]["record"]>,
200
-
): EnrichedChatMessage {
170
+
): Promise<EnrichedChatMessage> {
201
171
// Placeholder enrichment - just convert the format for now
172
+
const profile = await didResolver.resolve(jetstreamMessage.did);
173
+
202
174
return {
203
175
id: jetstreamMessage.commit.cid,
204
176
text: record.text,
205
177
author: {
206
-
did: jetstreamMessage.did, // Message author's DID
207
-
handle: jetstreamMessage.did, // TODO: Resolve DID to handle
208
-
// displayName would be enriched here by looking up profile
178
+
did: jetstreamMessage.did,
179
+
handle: profile.handle,
180
+
// displayName if implemented
209
181
},
210
182
timestamp: new Date(record.createdAt),
211
-
color: {
212
-
red: 128, // TODO: Get from user's chat profile
213
-
green: 128,
214
-
blue: 128,
215
-
},
183
+
color: profile.color!,
216
184
facets: record.facets,
217
185
isReply: !!record.reply,
218
-
streamer: record.streamer, // The streamer's DID
186
+
streamer: record.streamer,
219
187
};
220
188
}
221
189
}