bluesky-groupchat-bot/index.mjs

import 'dotenv/config';
import {Bot, IncomingChatPreference, RichText} from "@skyware/bot";
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
});
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();
		var respond = text => sendMessage({conversationId: conversation.id, text});

		if (message.text.startsWith('/')) {
			let args = message.text.split(' ');
			let cmd = args[0].slice(1).toLowerCase();
			let commandList = `/list, /leave, /join, /invite, /ping`;
			switch(cmd) {
				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: ${commandList}`);
					return;
				default:
					await respond(`Only these commands are accepted: ${commandList}`);
					return;
			}
		}

		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');

		var text = message.text;
		var facets = message.facets || [];

		var linkedName = sender.displayName || sender.handle;
		var prefix = `${linkedName}: `;
		var prefixByteLength = new Blob([prefix]).size;

		facets = facets.map(f => {
			var facet = f.toRecord();
			facet.index.byteStart += prefixByteLength;
			facet.index.byteEnd += prefixByteLength;
			return facet;
		});

		text = prefix + text;
		facets.unshift({
			index: {
				byteStart: 0,
				byteEnd: new Blob([linkedName]).size
			},
			features: [{
				did: sender.did,
				"$type": "app.bsky.richtext.facet#mention"
			}]
		});

		var otherconvoids = memberlist.filter(x => x != conversation.id);
		var messages = otherconvoids.map(conversationId => ({
			conversationId,
			text,
			facets,
			embed: message.embed
		}));
		await sendMessages(messages, {resolveFacets: false});
	} catch (error) {
		console.error(error);
		if (conversation) sendMessage({conversationId: conversation.id, text: `${error.message}\ncause: ${error.cause.message}`}).catch(error => console.error(error));
	}
});






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;