Blueskyのfirehoseを取得するスクリプト

firehose.ts

/*
よくある投稿がダーッと流れてくるアレができるスクリプトです。

Denoで実行することを想定しています。
2024/10/21 現在動くコードです。
Blueskyはライブラリの変更が激しいのでこのスクリプトも参考程度に読むことをお勧めします。
*/

import { cborDecodeMulti } from "npm:@atproto/common";
import { AtpAgent } from "npm:@atproto/api";
import { is } from "jsr:@core/unknownutil";

const host = "bsky.network";
const nsid = "com.atproto.sync.subscribeRepos";

const agent = new AtpAgent({
  service: "https://bsky.social",
});

// AtURLを生成する関数
// TODO: ちゃんとした関数を使いたい
const atUri = (repo: string, path: string) => `at://${repo}/${path}`;

await agent.login({
  identifier: "HANDLE",
  password: "PASSWORD",
});

const ws = new WebSocket(
  `wss://${host}/xrpc/${nsid}`,
);

ws.onopen = (_event) => {
  console.log("Connected!");
};

// 投稿のURLを保持する配列
const uris: string[] = [];

ws.onmessage = async (event) => {
  try {
    // 多分Blobが返ってくる本当は条件分岐した方が良い
    const blob = event.data as Blob;
    const data = await blob.bytes();

    // いともたやすく行なわれる型情報の握り潰し
    const [header, payload] = cborDecodeMulti(data) as any;

    if (header["op"] == 1) {
      // regular message
      const t = header["t"];

      if (!payload) return;

      if (t) {
        if (is.Undefined(payload.ops)) return;
        if (is.Undefined(payload.ops[0])) return;
        if (is.Null(payload.ops[0].cid)) return;
        if (is.Null(payload.ops[0].action != "create")) return;
        if (payload.ops[0].path.indexOf("app.bsky.feed.post") == -1) return;

        uris.push(atUri(payload.repo, payload.ops[0].path));

        if (uris.length === 24) {
          const posts = await fetchPost(uris);
          uris.length = 0;

          posts.forEach((post) => {
            console.log(post.record.text);
            console.log(
              "-----------------------------------------------------------------",
            );
          });
        }
      } else {
        console.error("Error!");
      }
    }
  } catch (e) {
    // いともたすく行なわれる適当なエラーハンドリング
    console.log(e);
  }
};

async function fetchPost(uris: string[]) {
  /*
  getPostエンドポイントを使用して投稿の本文を取得する
  日本語に設定された投稿以外は除外される
  TODO: 投稿の除外処理を分離する
  */

  const resp = await agent.app.bsky.feed.getPosts({ uris: uris });

  if (!is.Array(resp.data.posts)) return [];

  const postContents = resp.data.posts.filter((post) => {
    // 欠損値を除外
    // TODO: なぜか型エラーが発生するのでなんとかする
    if (is.Undefined(post)) return false;
    if (is.Undefined(post.record.text)) return false;
    if (is.Undefined(post.record.langs)) return false;
    // 日本語に設定さていない投稿を除外
    if (!post.record.langs.includes("ja")) return false;

    return post;
  });

  return postContents;
}