an app to share curated trails
sidetrail.app
1import "dotenv/config";
2import WebSocket, { WebSocketServer } from "ws";
3import { createServer } from "http";
4
5const JETSTREAM_URL =
6 process.env.JETSTREAM_URL || "wss://jetstream2.us-east.bsky.network/subscribe";
7const PORT = parseInt(process.env.PORT || "3002", 10);
8const WANTED_COLLECTIONS = [
9 "app.sidetrail.trail",
10 "app.sidetrail.walk",
11 "app.sidetrail.completion",
12];
13
14// Track connected clients
15const clients = new Set<WebSocket>();
16
17// Jetstream connection state
18let jetstreamWs: WebSocket | null = null;
19let isDestroyed = false;
20
21function buildJetstreamUrl(): string {
22 const params = new URLSearchParams();
23 for (const collection of WANTED_COLLECTIONS) {
24 params.append("wantedCollections", collection);
25 }
26 return `${JETSTREAM_URL}?${params.toString()}`;
27}
28
29function broadcast(data: string): void {
30 for (const client of clients) {
31 if (client.readyState === WebSocket.OPEN) {
32 client.send(data);
33 }
34 }
35}
36
37function connectToJetstream(): void {
38 if (isDestroyed) return;
39
40 const url = buildJetstreamUrl();
41 console.log(`[Jetstream] Connecting to: ${url}`);
42
43 jetstreamWs = new WebSocket(url);
44
45 jetstreamWs.on("open", () => {
46 console.log("[Jetstream] Connected");
47 });
48
49 jetstreamWs.on("message", (data) => {
50 const message = data.toString();
51 const parsed = JSON.parse(message);
52 if (parsed.kind !== "commit") return;
53 broadcast(message);
54 });
55
56 jetstreamWs.on("error", (err) => {
57 console.error("[Jetstream] Error:", err.message);
58 });
59
60 jetstreamWs.on("close", (code, reason) => {
61 console.log(`[Jetstream] Closed (code: ${code}, reason: ${reason})`);
62 jetstreamWs = null;
63
64 if (!isDestroyed) {
65 console.log("[Jetstream] Reconnecting in 5s...");
66 setTimeout(connectToJetstream, 5000);
67 }
68 });
69}
70
71function shutdown(): void {
72 console.log("\nShutting down...");
73 isDestroyed = true;
74
75 if (jetstreamWs) {
76 jetstreamWs.close();
77 jetstreamWs = null;
78 }
79
80 for (const client of clients) {
81 client.close();
82 }
83 clients.clear();
84
85 process.exit(0);
86}
87
88// Create HTTP server for health checks
89const server = createServer((req, res) => {
90 if (req.url === "/health") {
91 res.writeHead(200, { "Content-Type": "application/json" });
92 res.end(
93 JSON.stringify({
94 status: "ok",
95 clients: clients.size,
96 jetstream: jetstreamWs?.readyState === WebSocket.OPEN ? "connected" : "disconnected",
97 }),
98 );
99 } else {
100 res.writeHead(404);
101 res.end("Not found");
102 }
103});
104
105// Create WebSocket server attached to HTTP server
106const wss = new WebSocketServer({ server });
107
108wss.on("connection", (ws, req) => {
109 const clientIp = req.socket.remoteAddress;
110 console.log(`[Client] Connected from ${clientIp} (total: ${clients.size + 1})`);
111
112 clients.add(ws);
113
114 ws.on("close", () => {
115 clients.delete(ws);
116 console.log(`[Client] Disconnected (total: ${clients.size})`);
117 });
118
119 ws.on("error", (err) => {
120 console.error(`[Client] Error:`, err.message);
121 clients.delete(ws);
122 });
123});
124
125// Start server
126server.listen(PORT, () => {
127 console.log(`[Server] WebSocket relay listening on port ${PORT}`);
128 console.log(`[Server] Health check: http://localhost:${PORT}/health`);
129
130 // Connect to Jetstream
131 connectToJetstream();
132});
133
134// Handle shutdown signals
135process.on("SIGINT", shutdown);
136process.on("SIGTERM", shutdown);