message queue

.gitignore

@@ -1,3 +1,4 @@
-chatlog.csv
+*.csv
 node_modules
 .env
+memberlist.json

index.mjs

@@ -1,11 +1,37 @@
 import 'dotenv/config';
 import {Bot, IncomingChatPreference, RichText} from "@skyware/bot";
-import { appendFileSync } from "fs";
+import fetchRetry from "fetch-retry";
+import { appendFileSync, readFileSync, writeFileSync } from "fs";
+
+var _fetch = fetchRetry(global.fetch);
+global.fetch = function fetch(url, options) {
+	//console.debug("fetch", String(url));
+	return _fetch(url, options);
+}
+
+
+
+
+var memberlist = []; // list of conversation ids
+try {
+	memberlist = JSON.parse(readFileSync("memberlist.json", "utf8"));
+} catch (error) {}
+
+function saveMemberlist() {
+	writeFileSync("memberlist.json", JSON.stringify(memberlist));
+}
+
+
+
 
 var bot = new Bot({
 	emitEvents: false,
 	emitChatEvents: true
 });
+bot.on("error", error => {
+	console.error("bot error", error);
+});
+
 await bot.login({
 	identifier: process.env.IDENTIFIER,
 	password: process.env.PASSWORD
@@ -14,36 +40,76 @@ await bot.setChatPreference(IncomingChatPreference.All);
 
 
 bot.on("message", async message => {
+	console.debug(new Date().toISOString(), message.senderDid, message.text);
 	try {
 		var sender = await message.getSender();
 		var conversation = await message.getConversation();
-		if (!conversation) { console.error("missing conversation"); }
+		var respond = text => sendMessage({conversationId: conversation.id, text});
 
 		if (message.text.startsWith('/')) {
-			let respond = text => conversation.sendMessage({text});
-			let cmd = message.text.split(' ')[0].slice(1).toLowerCase();
+			let args = message.text.split(' ');
+			let cmd = args[0].slice(1).toLowerCase();
+			let commandList = `/list, /leave, /join, /invite, /ping`;
 			switch(cmd) {
-				case "list":
-					let {conversations} = await bot.listConversations();
-					await respond(`${conversations.length} members in group chat: ${conversations.map(c => '@' + c.members.find(m => m.did != bot.profile.did).handle).join(', ')}`);
-					return;
-				case "leave":
-					await respond(`You left the group chat and will no longer receive messages! Send any message to join again, or send /join to re-join silently.`);
-					await conversation.leave();
-					return;
-				case "join": 
-					await respond(`Welcome to the group chat!`);
-					return;
 				case "ping":
 					await respond("pong");
 					return;
+				case "list":
+					let memberConversations = await Promise.all(memberlist.map(convoid => bot.getConversation(convoid)));
+					await respond(`${memberConversations.length} members in group chat: ${memberConversations.map(c => {
+						let handle = c.members.find(m => m.did != bot.profile.did).handle;
+						if (handle != "missing.invalid") { // work around https://github.com/skyware-js/bot/issues/19
+							handle = `@${handle}`;
+						}
+						return handle;
+					}).join(', ')}`);
+					return;
+				case "leave":
+					broadcast(new RichText().addMention(sender.displayName || sender.handle || message.senderDid, message.senderDid).addText(" left the group chat.")).catch(console.error);
+					memberlist = memberlist.filter(convoid => convoid != conversation.id);
+					saveMemberlist();
+					await respond(`You left the group chat. Rejoin at any time with /join`);
+					return;
+				case "join":
+					if (memberlist.includes(conversation.id)) {
+						await respond(`You're already in the group chat.`);
+					} else {
+						broadcast(new RichText().addMention(sender.displayName || sender.handle || message.senderDid, message.senderDid).addText(" joined the group chat.")).catch(console.error);
+						memberlist.push(conversation.id);
+						saveMemberlist();
+						await respond(`Welcome to the group chat; you are now chatting with ${memberlist.length} other people. Use /list to see them.`);
+					}
+					return;
+				case "invite":
+					if (!memberlist.includes(conversation.id)) {
+						await respond(`You are not allowed to invite people to the group chat when you are not in the group chat yourself. Use /join first.`);
+						return;
+					}
+					let handle = args[1];
+					if (!handle) {
+						await respond(`Usage: /invite <handle>`);
+						return;
+					}
+					handle = handle.replace('@','');
+					let profile = await bot.getProfile(handle);
+					await profile.sendMessage({text: `@${sender.handle} invited you to the group chat! Use /join to accept!`});
+					await broadcast(new RichText().addMention(sender.displayName || sender.handle || message.senderDid, message.senderDid).addText(" invited ").addMention(profile.displayName||profile.handle,profile.did).addText(" to the group chat."));
+					return;
 				case "help":
-					await respond(`Commands: /list, /leave, /join, /ping`);
+					await respond(`Commands: ${commandList}`);
+					return;
+				default:
+					await respond(`Only these commands are accepted: ${commandList}`);
 					return;
 			}
 		}
 
-		var logline = `${message.sentAt.toISOString()},${message.senderDid},${sender.displayName},${sender.handle},${message.text}`;
+		if (!memberlist.includes(conversation.id)) {
+			await respond("Use /join to join the group chat.");
+			return;
+		}
+
+		var logline = `${message.sentAt.toISOString()},${message.senderDid},${sender.displayName},${sender.handle},${JSON.stringify(message.text)}`;
 		console.log(logline);
 		appendFileSync(`chatlog.csv`, logline + '\n');
 
@@ -73,20 +139,17 @@ bot.on("message", async message => {
 			}]
 		});
 
-		var {conversations} = await bot.listConversations();
-		var otherConservations = conversations.filter(c => c.id != conversation?.id);
-		//otherConservations = otherConservations.filter(c => c.members.length > 1); // doesn't work, bsky doesnt let you know when they leave?
-
-		var messages = otherConservations.map(c => ({
-			conversationId: c.id,
+		var otherconvoids = memberlist.filter(x => x != conversation.id);
+		var messages = otherconvoids.map(conversationId => ({
+			conversationId,
 			text,
-			facets
+			facets,
+			embed: message.embed
 		}));
-
-		await bot.sendMessages(messages, {resolveFacets: false});
+		await sendMessages(messages, {resolveFacets: false});
 	} catch (error) {
 		console.error(error);
-		conversation?.sendMessage({text: error.message}).catch(error => console.error(error));
+		if (conversation) sendMessage({conversationId: conversation.id, text: `${error.message}\ncause: ${error.cause.message}`}).catch(error => console.error(error));
 	}
 });
 
@@ -94,4 +157,63 @@ bot.on("message", async message => {
 
 
 
+
+async function broadcast(text) {
+	var logline = `${new Date().toISOString()},,,,${JSON.stringify(text.text||text)}`;
+	console.log(logline);
+	appendFileSync(`chatlog.csv`, logline + '\n');
+	var messages = memberlist.map(conversationId => ({
+		conversationId,
+		text
+	}));
+	await sendMessages(messages);
+}
+
+
+// presere order
+var queue = [];
+var ratelimited = false;
+function sendMessages(messages, options) {
+	queue.push([messages, options]);
+	if (!ratelimited) return flushQueue();
+}
+
+async function flushQueue() {
+	if (!queue.length) return;
+	var [messages, options] = queue.shift();
+	try {
+		await bot.sendMessages(messages, options);
+		ratelimited = false;
+	} catch (error) {
+		if (error.cause?.kind == "RateLimitExceeded") {
+			console.warn("ratelimited");
+			ratelimited = true;
+			queue.unshift([messages, options]);
+		} else throw error;
+	}
+}
+
+(function flushQueueLoop() {
+	flushQueue().catch(console.error).then(() => {
+		setTimeout(flushQueueLoop, 10000);
+	});
+})();
+
+
+// no preserve order
+async function sendMessage(message, options) {
+	try {
+		return await bot.sendMessage(message, options);
+	} catch (error) {
+		if (error.cause?.kind == "RateLimitExceeded") {
+			console.warn("ratelimited");
+			await new Promise(r => setTimeout(r, 10000));
+			return await sendMessage(message, options);
+		} else throw error;
+	}
+}
+
+
+
+
 global.bot = bot;

package-lock.json

@@ -5,8 +5,9 @@
   "packages": {
     "": {
       "dependencies": {
-        "@skyware/bot": "^0.3.7-beta.1",
-        "dotenv": "^16.4.5"
+        "@skyware/bot": "^0.3.7",
+        "dotenv": "^16.4.5",
+        "fetch-retry": "^6.0.0"
       }
     },
     "node_modules/@atcute/base32": {
@@ -93,9 +94,9 @@
       "optional": true
     },
     "node_modules/@skyware/bot": {
-      "version": "0.3.7-beta.1",
-      "resolved": "https://registry.npmjs.org/@skyware/bot/-/bot-0.3.7-beta.1.tgz",
-      "integrity": "sha512-flH5DzHKBuLfSSJsHguh3ZcsTaillUo5sq1llrXEU6xY9Btwe2B97dvlvTNdDOHJeyXse5Mlwyq5CE4aeF/xHw==",
+      "version": "0.3.7",
+      "resolved": "https://registry.npmjs.org/@skyware/bot/-/bot-0.3.7.tgz",
+      "integrity": "sha512-27k7r4/YA+h8FobXokMa3iv8fm/850a7NmajOHP3/CaMbCYicXXBDnAIY4kIPP58Zl/A/JvdcFLpQBpfFkiQHQ==",
       "dependencies": {
         "@atcute/bluesky": "^1.0.7",
         "@atcute/bluesky-richtext-builder": "^1.0.1",
@@ -156,6 +157,11 @@
         "url": "https://github.com/sponsors/mysticatea"
       }
     },
+    "node_modules/fetch-retry": {
+      "version": "6.0.0",
+      "resolved": "https://registry.npmjs.org/fetch-retry/-/fetch-retry-6.0.0.tgz",
+      "integrity": "sha512-BUFj1aMubgib37I3v4q78fYo63Po7t4HUPTpQ6/QE6yK6cIQrP+W43FYToeTEyg5m2Y7eFUtijUuAv/PDlWuag=="
+    },
     "node_modules/partysocket": {
       "version": "1.0.2",
       "resolved": "https://registry.npmjs.org/partysocket/-/partysocket-1.0.2.tgz",

package.json

@@ -1,6 +1,7 @@
 {
   "dependencies": {
-    "@skyware/bot": "^0.3.7-beta.1",
-    "dotenv": "^16.4.5"
+    "@skyware/bot": "^0.3.7",
+    "dotenv": "^16.4.5",
+    "fetch-retry": "^6.0.0"
   }
 }