Refactoring

index.ts

@@ -1,9 +1,10 @@
 import { api } from "misskey-js";
+import Stream from "./misskey-js/streaming";
+import type { Stream as MisskeyStream } from "misskey-js";
+import type { Note } from "misskey-js/entities.js";
+
 import OpenAI from "openai";
 import type { ChatCompletionMessageParam } from "openai/resources/index.js";
-import type { Stream as MisskeyStream } from "misskey-js";
-import Stream from "./misskey-js/streaming";
-import type { Note } from "misskey-js/entities.js";
 
 const misskey = new api.APIClient({
   origin: Bun.env["MISSKEY_ORIGIN"] || "https://misskey.cannorin.net",
@@ -18,15 +19,17 @@ const openai = new OpenAI({
 
 // #region util
 
-/** shuffle the array */
-function randomize<T>(array: T[]) {
-  for (let i = array.length - 1; i > 0; i--) {
-    const r = Math.floor(Math.random() * (i + 1));
-    const tmp = array[i] as T;
-    array[i] = array[r] as T;
-    array[r] = tmp;
+/** pick random N elements from array.
+ *  just shuffle it if N is unspecified.
+ *  the original array remains unmodified. */
+function sample<T>(arr: T[], n: number = arr.length): T[] {
+  if (n > arr.length) throw new Error("sample: N out of range");
+  const copy = [...arr];
+  for (let i = 0; i < n; i++) {
+    const j = i + Math.floor(Math.random() * (copy.length - i));
+    [copy[i], copy[j]] = [copy[j] as T, copy[i] as T];
   }
-  return array;
+  return copy.slice(0, n);
 }
 
 /** sleep for N milliseconds */
@@ -48,7 +51,7 @@ async function getNotes() {
       limit: 100,
     });
     // exclude bot notes
-    return randomize(timeline.filter((p) => !p.user.isBot)).slice(0, count);
+    return sample(timeline.filter((p) => !p.user.isBot), count);
   }
 
   // randomly sample N global notes
@@ -59,7 +62,7 @@ async function getNotes() {
       limit: 100,
     });
     // exclude bot notes and replies
-    return randomize(timeline.filter((p) => !p.user.isBot && !p.reply)).slice(0, count);
+    return sample(timeline.filter((p) => !p.user.isBot && !p.reply), count);
   }
 
   // randomly sample N notes of mine
@@ -69,7 +72,7 @@ async function getNotes() {
       limit: 100,
       withRenotes: false,
     });
-    return randomize(notes).slice(0, count);
+    return sample(notes, count);
   }
 
   const notes = await Promise.all([
@@ -77,14 +80,14 @@ async function getNotes() {
     getGlobalNotes(10),
     getMyNotes(2),
   ]);
-  return randomize(notes.flat());
+  return sample(notes.flat());
 }
 
 /** fetch the whole reply tree */
-async function expandReplyTree(note: Note, acc: Note[] = []): Promise<Note[]> {
-  if (!note.reply) return [...acc, note];
+async function expandReplyTree(note: Note, acc: Note[] = [], cutoff = 5) {
+  if (!note.reply || cutoff < 1) return [...acc, note];
   const reply = await misskey.request("notes/show", { noteId: note.reply.id });
-  return await expandReplyTree(reply, [...acc, note]);
+  return await expandReplyTree(reply, [...acc, note], cutoff - 1);
 }
 
 /** convert a note to a chat message */
@@ -195,6 +198,14 @@ const jobs: Job[] = [];
 let stream: MisskeyStream;
 let channel: ReturnType<typeof stream.useChannel<"main">>;
 
+/** dispose stream for recreation */
+function disposeStream() {
+  channel.removeAllListeners();
+  channel.dispose();
+  stream.removeAllListeners();
+  stream.close();
+}
+
 /** connect to streaming API and add handlers */
 function initializeStream() {
   stream = new Stream(
@@ -210,10 +221,9 @@ function initializeStream() {
     console.log("* connected");
   });
 
-  // reconnect automatically
+  // notify when disconnected (it will reconnect automatically)
   stream.on("_disconnected_", () => {
-    console.log("* disconnected, reconnecting");
-    initializeStream();
+    console.log("* disconnected");
   });
 
   // push a reply job when receiving a mention
@@ -238,7 +248,6 @@ initializeStream();
 async function runJob() {
   while (true) {
     const job = jobs.pop();
-
     if (job) {
       console.log(`* pop: ${job.type}`);
       try {
@@ -248,7 +257,6 @@ async function runJob() {
         console.log(`* error: ${e}`);
       }
     }
-
     await sleep(1000); // 1sec
   }
 }
@@ -257,13 +265,11 @@ async function runJob() {
 async function pushJob() {
   while (true) {
     const now = new Date(Date.now());
-
     // push a post job every 15 minutes (XX:00, XX:15, XX:30, XX:45)
     if (now.getMinutes() % 15 < Number.EPSILON && !jobs.some((job) => job.type === "post")) {
       console.log("* push: post");
       jobs.push({ type: "post" });
     }
-
     await sleep(60 * 1000); // 1min
   }
 }
@@ -271,9 +277,13 @@ async function pushJob() {
 
 async function main() {
   try {
-    await Promise.all([runJob(), pushJob()]);
-  } catch (e) {
-    console.error(e);
+    try {
+      await Promise.all([runJob(), pushJob()]);
+    } catch (e) {
+      console.error(e);
+    }
+  } finally {
+    disposeStream();
   }
 }
 await main();