/** @jsxImportSource https://esm.sh/react */ import { AtpAgent } from "https://esm.sh/@atproto/api@0.13.15"; import { AnimatePresence, motion } from "https://esm.sh/framer-motion"; import React, { useEffect, useMemo, useRef, useState } from "https://esm.sh/react"; import { createRoot } from "https://esm.sh/react-dom/client"; import { TagCloud } from "https://esm.sh/react-tagcloud@2.2.0"; const service = "https://public.api.bsky.app"; const agent = new AtpAgent({ service }); function App() { const [messages, setMessages] = useState([]); const [isActive, setIsActive] = useState(true); const [messageCount, setMessageCount] = useState(0); const [displayCount, setDisplayCount] = useState(20); const [messageRate, setMessageRate] = useState(1); const [filterType, setFilterType] = useState("all"); const wsRef = useRef(null); const messageQueue = useRef([]); const connectWebSocket = () => { wsRef.current = new WebSocket( "wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post", ); wsRef.current.onopen = () => { console.log("WebSocket connection established"); }; wsRef.current.onmessage = (event) => { const newMessage = JSON.parse(event.data); if (filterType === "all" || newMessage.commit?.record?.$type === filterType) { messageQueue.current.push(newMessage); } }; wsRef.current.onerror = (error) => { console.error("WebSocket error:", error); }; wsRef.current.onclose = () => { console.log("WebSocket connection closed"); }; }; useEffect(() => { if (isActive) { connectWebSocket(); const messageInterval = setInterval(async () => { if (messageQueue.current.length > 0) { const newMessage = messageQueue.current.shift(); const { data } = await agent.getProfile({ actor: newMessage.did }); newMessage.handle = data.handle; newMessage.avatar = data.avatar; setMessages((prevMessages) => [newMessage, ...prevMessages.slice(0, displayCount - 1)]); setMessageCount((prevCount) => prevCount + 1); } }, messageRate === "realtime" ? 0 : 1000 / messageRate); return () => { clearInterval(messageInterval); }; } return () => { if (wsRef.current) { wsRef.current.close(); } }; }, [isActive, displayCount, messageRate, filterType]); const handleStop = () => { setIsActive(false); if (wsRef.current) { wsRef.current.close(); } }; const handleRestart = () => { setIsActive(true); setMessages([]); setMessageCount(0); messageQueue.current = []; connectWebSocket(); }; const wordCloudData = useMemo(() => { const words = messages .flatMap((message) => message.commit?.record?.text?.split(/\s+/) || []) .filter((word) => word.length > 3) .reduce((acc, word) => { acc[word] = (acc[word] || 0) + 1; return acc; }, {}); return Object.entries(words) .map(([value, count]) => ({ value, count })) .sort((a, b) => b.count - a.count) .slice(0, 50); }, [messages]); return (

Bluesky Firehose

Messages to display: setDisplayCount(Number(e.target.value))} className="w-full px-3 py-2 bg-gray-700 rounded-md" min="1" max="100" />

Message rate:

Filter by type:

Total Messages: {messages.length}

Word Cloud

{messages.map((message) => (

{message.commit?.record?.text}

{message.handle}

))}

); } function client() { createRoot(document.getElementById("root")).render(); } if (typeof document !== "undefined") { client(); } export default async function server(request: Request): Promise { return new Response( ` Bluesky Firehose Tech Demo

`, { headers: { "content-type": "text/html", }, }, ); }

firehose

Versions

v0