streamplace-term-chat/src/index.ts

#!/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.
      --silent    Suppress the information messages printed above the chat and prompt.

    Examples:
      npm start -- alice.bsky.social
      npm start -- alice.bsky.social --streamer streamer.bsky.social
      npm start -- alice.bsky.social --silent
    `),
    );
    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();

  if (args.includes("--silent") === false) {
    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);
});