feat: initial implementation of chat in terminal from stream.place
This commit is contained in:
Stevan Freeborn committed
commit
689304656f
39 files changed
+4243
No files matched your search
+399
@@ -0,0 +1,399 @@
|
||||
#!/usr/bin/env node
|
||||
import http from "node:http";
|
||||
import open from "open";
|
||||
import os from 'os';
|
||||
import {
|
||||
buildAtprotoLoopbackClientMetadata,
|
||||
NodeOAuthClient,
|
||||
type NodeSavedSession,
|
||||
type NodeSavedState,
|
||||
} from "@atproto/oauth-client-node";
|
||||
import { Client } from "@atproto/lex";
|
||||
import * as place from "./lexicons/place.js";
|
||||
import WebSocket from "ws";
|
||||
import {
|
||||
CompositeDidDocumentResolver,
|
||||
PlcDidDocumentResolver,
|
||||
WebDidDocumentResolver,
|
||||
LocalActorResolver,
|
||||
WellKnownHandleResolver,
|
||||
DohJsonHandleResolver,
|
||||
CompositeHandleResolver,
|
||||
} from "@atcute/identity-resolver";
|
||||
import readline from "readline";
|
||||
import path from "node:path";
|
||||
import fs from "node:fs/promises";
|
||||
|
||||
const c = {
|
||||
gray: (str: string) => `\x1b[90m${str}\x1b[0m`,
|
||||
bold: (str: string) => `\x1b[1m${str}\x1b[22m`,
|
||||
cyan: (str: string) => `\x1b[36m${str}\x1b[39m`,
|
||||
green: (str: string) => `\x1b[32m${str}\x1b[39m`,
|
||||
magenta: (str: string) => `\x1b[35m${str}\x1b[39m`,
|
||||
yellow: (str: string) => `\x1b[33m${str}\x1b[39m`,
|
||||
red: (str: string) => `\x1b[31m${str}\x1b[39m`,
|
||||
};
|
||||
|
||||
const DEFAULT_JETSTREAM_URL =
|
||||
"wss://jetstream.us-east.bsky.network/xrpc/network.bsky.jetstream.subscribeEvents?collections=place.stream.chat.message";
|
||||
const CONFIG_DIR = path.join(os.homedir(), ".streamplace-term-chat");
|
||||
await fs.mkdir(CONFIG_DIR, { recursive: true });
|
||||
|
||||
const SESSION_FILE = path.join(CONFIG_DIR, ".session-store.json");
|
||||
const STATE_FILE = path.join(CONFIG_DIR, ".state-store.json");
|
||||
|
||||
async function loadJsonStore<T>(filePath: string): Promise<Map<string, T>> {
|
||||
try {
|
||||
const data = await fs.readFile(filePath, "utf-8");
|
||||
return new Map(Object.entries(JSON.parse(data)));
|
||||
} catch {
|
||||
return new Map();
|
||||
}
|
||||
}
|
||||
|
||||
async function saveJsonStore<T>(
|
||||
filePath: string,
|
||||
store: Map<string, T>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const obj = Object.fromEntries(store.entries());
|
||||
await fs.writeFile(filePath, JSON.stringify(obj, null, 2), "utf-8");
|
||||
} catch (err) {
|
||||
console.error(`Failed to save store to ${filePath}:`, err);
|
||||
}
|
||||
}
|
||||
|
||||
const stateStore = await loadJsonStore<NodeSavedState>(STATE_FILE);
|
||||
const sessionStore = await loadJsonStore<NodeSavedSession>(SESSION_FILE);
|
||||
|
||||
const handleResolver = new CompositeHandleResolver({
|
||||
methods: {
|
||||
dns: new DohJsonHandleResolver({
|
||||
dohUrl: "https://mozilla.cloudflare-dns.com/dns-query",
|
||||
}),
|
||||
http: new WellKnownHandleResolver(),
|
||||
},
|
||||
});
|
||||
|
||||
const didResolver = new CompositeDidDocumentResolver({
|
||||
methods: {
|
||||
plc: new PlcDidDocumentResolver(),
|
||||
web: new WebDidDocumentResolver(),
|
||||
},
|
||||
});
|
||||
|
||||
const actorResolver = new LocalActorResolver({
|
||||
handleResolver,
|
||||
didDocumentResolver: didResolver,
|
||||
});
|
||||
|
||||
const handleCache = new Map<string, string>();
|
||||
|
||||
async function resolveHandle(did: string): Promise<string> {
|
||||
const cached = handleCache.get(did);
|
||||
|
||||
if (cached) {
|
||||
return cached;
|
||||
}
|
||||
|
||||
try {
|
||||
const actor = await actorResolver.resolve(did as any);
|
||||
const identifier = actor.handle ?? actor.did;
|
||||
handleCache.set(did, identifier);
|
||||
return identifier;
|
||||
} catch {
|
||||
return did;
|
||||
}
|
||||
}
|
||||
|
||||
const oauthClient = new NodeOAuthClient({
|
||||
clientMetadata: buildAtprotoLoopbackClientMetadata({
|
||||
scope: "atproto repo:place.stream.chat.message?action=create",
|
||||
redirect_uris: ["http://127.0.0.1:3000/callback"],
|
||||
}),
|
||||
stateStore: {
|
||||
async get(key: string) {
|
||||
return stateStore.get(key);
|
||||
},
|
||||
async set(key: string, value: NodeSavedState) {
|
||||
stateStore.set(key, value);
|
||||
await saveJsonStore(STATE_FILE, stateStore);
|
||||
},
|
||||
async del(key: string) {
|
||||
stateStore.delete(key);
|
||||
await saveJsonStore(STATE_FILE, stateStore);
|
||||
},
|
||||
},
|
||||
sessionStore: {
|
||||
async get(key: string) {
|
||||
return sessionStore.get(key);
|
||||
},
|
||||
async set(key: string, value: NodeSavedSession) {
|
||||
sessionStore.set(key, value);
|
||||
await saveJsonStore(SESSION_FILE, sessionStore);
|
||||
},
|
||||
async del(key: string) {
|
||||
sessionStore.delete(key);
|
||||
await saveJsonStore(SESSION_FILE, sessionStore);
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
async function resolveToDid(input: string): Promise<string> {
|
||||
if (input.startsWith("did:")) {
|
||||
return input;
|
||||
}
|
||||
|
||||
const resolved = await actorResolver.resolve(input as any);
|
||||
return resolved.did;
|
||||
}
|
||||
|
||||
async function login(handleOrDid: string) {
|
||||
let userDid: string;
|
||||
|
||||
try {
|
||||
userDid = await resolveToDid(handleOrDid);
|
||||
} catch (err) {
|
||||
console.error(c.red(`Failed to resolve handle "${handleOrDid}":`), err);
|
||||
throw err;
|
||||
}
|
||||
|
||||
try {
|
||||
const session = await oauthClient.restore(userDid);
|
||||
|
||||
if (session) {
|
||||
console.log(
|
||||
c.green(`\nRestored existing session for @${handleOrDid} (${userDid})`),
|
||||
);
|
||||
return session;
|
||||
}
|
||||
} catch { }
|
||||
|
||||
console.log(c.gray(`\nInitiating browser login for @${handleOrDid}...`));
|
||||
|
||||
const authUrl = await oauthClient.authorize(handleOrDid, {
|
||||
scope: "atproto repo:place.stream.chat.message?action=create",
|
||||
});
|
||||
|
||||
const params = await new Promise<URLSearchParams>((resolve, reject) => {
|
||||
const server = http.createServer((req, res) => {
|
||||
if (!req.url) {
|
||||
return;
|
||||
}
|
||||
|
||||
const url = new URL(req.url, "http://127.0.0.1:3000");
|
||||
|
||||
if (url.pathname === "/callback") {
|
||||
res.writeHead(200, { "Content-Type": "text/html" });
|
||||
res.end(
|
||||
"<h1>Authorized! You can close this tab and return to your terminal.</h1>",
|
||||
);
|
||||
resolve(url.searchParams);
|
||||
server.close();
|
||||
}
|
||||
});
|
||||
|
||||
server.listen(3000, "127.0.0.1", () => {
|
||||
console.log(
|
||||
c.gray(
|
||||
"Listening on http://127.0.0.1:3000/callback for OAuth redirect...",
|
||||
),
|
||||
);
|
||||
open(authUrl.toString());
|
||||
});
|
||||
|
||||
server.on("error", reject);
|
||||
});
|
||||
|
||||
const { session } = await oauthClient.callback(params);
|
||||
return session;
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const args = process.argv.slice(2);
|
||||
const handleArg = args.find((arg) => !arg.startsWith("--"));
|
||||
const streamerFlagIndex = args.indexOf("--streamer");
|
||||
const streamerInput =
|
||||
streamerFlagIndex !== -1
|
||||
? args[streamerFlagIndex + 1]
|
||||
: process.env.STREAMER_DID;
|
||||
|
||||
if (!handleArg || args.includes("--help") || args.includes("-h")) {
|
||||
console.log(
|
||||
c.gray(`
|
||||
Streamplace Chat Client
|
||||
|
||||
Usage:
|
||||
npm start -- <your-handle> [--streamer <streamer-handle-or-did>]
|
||||
|
||||
Options:
|
||||
--streamer Specify target streamer handle/DID to view and post chat messages.
|
||||
Defaults to your own handle if omitted.
|
||||
|
||||
Examples:
|
||||
npm start -- alice.bsky.social
|
||||
npm start -- alice.bsky.social --streamer streamer.bsky.social
|
||||
`),
|
||||
);
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
const session = await login(handleArg);
|
||||
const client = new Client(session);
|
||||
|
||||
let targetStreamerDid = session.did as string;
|
||||
|
||||
if (streamerInput) {
|
||||
if (streamerInput.startsWith("did:")) {
|
||||
targetStreamerDid = streamerInput;
|
||||
} else {
|
||||
try {
|
||||
const resolved = await actorResolver.resolve(streamerInput as any);
|
||||
targetStreamerDid = resolved.did;
|
||||
} catch (err) {
|
||||
console.error(
|
||||
c.red(
|
||||
`Could not resolve streamer handle "${streamerInput}". Using user's DID.`,
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const connectedAsHandle = await resolveHandle(session.did);
|
||||
const targetStreamerHandle = await resolveHandle(targetStreamerDid);
|
||||
|
||||
console.clear();
|
||||
|
||||
console.log(
|
||||
c.green(c.bold("Connected as:")) +
|
||||
` @${connectedAsHandle} (${session.did})`,
|
||||
);
|
||||
console.log(
|
||||
c.yellow(c.bold("Chat room streamer:")) +
|
||||
` @${targetStreamerHandle} (${targetStreamerDid})`,
|
||||
);
|
||||
console.log(
|
||||
c.gray("Type your message and hit Enter. Commands: /quit, /help\n"),
|
||||
);
|
||||
|
||||
const rl = readline.createInterface({
|
||||
input: process.stdin,
|
||||
output: process.stdout,
|
||||
prompt: c.cyan(c.bold("> ")),
|
||||
});
|
||||
|
||||
let ws: WebSocket | null = null;
|
||||
let isConnecting = false;
|
||||
|
||||
function connectWs() {
|
||||
if (isConnecting) return;
|
||||
isConnecting = true;
|
||||
|
||||
ws = new WebSocket(process.env.JETSTREAM_URL || DEFAULT_JETSTREAM_URL);
|
||||
|
||||
ws.on("open", () => {
|
||||
isConnecting = false;
|
||||
rl.prompt();
|
||||
});
|
||||
|
||||
ws.on("message", async (msg) => {
|
||||
try {
|
||||
const jsonMsg = JSON.parse(msg.toString());
|
||||
|
||||
if (jsonMsg.payload?.record?.streamer !== targetStreamerDid) {
|
||||
return;
|
||||
}
|
||||
|
||||
const createdAt = new Date(jsonMsg.payload.record.createdAt);
|
||||
const timestamp = createdAt.toLocaleTimeString("en-US", {
|
||||
hour12: false,
|
||||
hour: "2-digit",
|
||||
minute: "2-digit",
|
||||
});
|
||||
|
||||
const senderDid = jsonMsg.payload.did;
|
||||
const senderIdentifier = await resolveHandle(senderDid);
|
||||
const messageText = jsonMsg.payload.record.text;
|
||||
const formattedLine = `${c.gray(`[${timestamp}]`)} ${c.magenta("[SP]")} ${c.cyan(
|
||||
c.bold(senderIdentifier),
|
||||
)}: ${messageText}`;
|
||||
|
||||
readline.clearLine(process.stdout, 0);
|
||||
readline.cursorTo(process.stdout, 0);
|
||||
console.log(formattedLine);
|
||||
rl.prompt(true);
|
||||
} catch { }
|
||||
});
|
||||
|
||||
ws.on("error", (err) => {
|
||||
readline.clearLine(process.stdout, 0);
|
||||
readline.cursorTo(process.stdout, 0);
|
||||
console.error(`[WebSocket Error] ${err.message}`);
|
||||
rl.prompt(true);
|
||||
});
|
||||
|
||||
ws.on("close", () => {
|
||||
isConnecting = false;
|
||||
setTimeout(connectWs, 3000);
|
||||
});
|
||||
}
|
||||
|
||||
connectWs();
|
||||
|
||||
rl.on("line", async (line) => {
|
||||
process.stdout.moveCursor(0, -1);
|
||||
process.stdout.clearLine(0);
|
||||
process.stdout.cursorTo(0);
|
||||
|
||||
const text = line.trim();
|
||||
|
||||
if (!text) {
|
||||
rl.prompt();
|
||||
return;
|
||||
}
|
||||
|
||||
if (text === "/quit" || text === "/exit") {
|
||||
rl.close();
|
||||
return;
|
||||
}
|
||||
|
||||
if (text === "/help") {
|
||||
console.log("Commands: /quit, /exit, /help");
|
||||
rl.prompt();
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await client.create(place.stream.chat.message as any, {
|
||||
streamer: targetStreamerDid,
|
||||
text: text,
|
||||
createdAt: new Date().toISOString(),
|
||||
});
|
||||
} catch (err) {
|
||||
if (err instanceof Error) {
|
||||
console.error(c.red(`Failed to send message: ${err.message}`));
|
||||
}
|
||||
}
|
||||
|
||||
rl.prompt();
|
||||
});
|
||||
|
||||
rl.on("close", () => {
|
||||
if (ws) {
|
||||
ws.removeAllListeners();
|
||||
ws.close();
|
||||
}
|
||||
|
||||
console.log(c.gray("\nExiting chat. Goodbye!"));
|
||||
process.exit(0);
|
||||
});
|
||||
}
|
||||
|
||||
main().catch((err) => {
|
||||
if (err instanceof Error) {
|
||||
console.error(c.red(`Fatal error: ${err.message}`));
|
||||
}
|
||||
|
||||
process.exit(1);
|
||||
});
|
||||
Reference in new issue
Block a user