A work-in-progress chat bot for Streamplace with chat overlay functionality
Shoutouts, chatter caching in StreamplaceBot to avoid API calls, StreamplaceClient with its own credentialManager
+358
-105
+13
main.ts
+13
main.ts
···
1
1
import { App, staticFiles } from "fresh";
2
2
import { type State } from "./utils.ts";
3
+
import { streamplaceWS } from "./utils/websocket.ts";
4
+
import StreamplaceBot from "./utils/streamplaceBot.ts";
5
+
import { ATPROTO_PASSWORD, ATPROTO_USERNAME, PDS_HOST_URL } from "./env.ts";
3
6
4
7
export const app = new App<State>();
5
8
9
+
export const streamplaceBot = new StreamplaceBot(
10
+
"did:plc:o6xucog6fghiyrvp7pyqxcs3", //this is me, make more flexible later
11
+
{
12
+
username: ATPROTO_USERNAME!,
13
+
password: ATPROTO_PASSWORD!,
14
+
pdsHostUrl: PDS_HOST_URL!,
15
+
},
16
+
);
17
+
streamplaceBot.init();
18
+
streamplaceWS.start();
6
19
app.use(staticFiles());
7
20
8
21
// Include file-system based routes here
+123
-57
utils/atcuteUtils.ts
+123
-57
utils/atcuteUtils.ts
···
1
-
import { Client, CredentialManager, ok } from "@atcute/client";
1
+
import type {} from "@atcute/atproto";
2
+
import {
3
+
Client,
4
+
CredentialManager,
5
+
ok,
6
+
simpleFetchHandler,
7
+
} from "@atcute/client";
2
8
import type {} from "@atcute/atproto";
3
9
import {
4
10
CompositeHandleResolver,
···
7
13
} from "@atcute/identity-resolver";
8
14
import { Handle } from "@atcute/lexicons";
9
15
import { isHandle } from "@atcute/lexicons/syntax";
10
-
import { ATPROTO_PASSWORD, ATPROTO_USERNAME, PDS_HOST_URL } from "../env.ts";
11
16
12
-
// Store credential manager singleton to avoid multiple logins
13
-
let credentialManager: CredentialManager | null = null;
14
-
let rpcClient: Client | null = null;
17
+
// Configuration for initializing a Streamplace client
18
+
export interface StreamplaceClientConfig {
19
+
username: string;
20
+
password: string;
21
+
pdsHostUrl: string;
22
+
}
15
23
16
-
export async function initStreamplaceClient(): Promise<void> {
17
-
if (credentialManager !== null) return;
24
+
// Wrapper class for managing a single bot's credentials and RPC client
25
+
export class StreamplaceClient {
26
+
private credentialManager: CredentialManager;
27
+
private rpcClient: Client;
28
+
private config: StreamplaceClientConfig;
29
+
private initialized: boolean = false;
18
30
19
-
credentialManager = new CredentialManager({
20
-
service: PDS_HOST_URL!,
21
-
});
22
-
rpcClient = new Client({ handler: credentialManager });
31
+
constructor(config: StreamplaceClientConfig) {
32
+
this.config = config;
33
+
this.credentialManager = new CredentialManager({
34
+
service: config.pdsHostUrl,
35
+
});
36
+
this.rpcClient = new Client({ handler: this.credentialManager });
37
+
}
23
38
24
-
await credentialManager.login({
25
-
identifier: ATPROTO_USERNAME!,
26
-
password: ATPROTO_PASSWORD!,
27
-
});
39
+
// Initialize the client by logging in
40
+
async init(): Promise<void> {
41
+
if (this.initialized) return;
28
42
29
-
console.log("Streamplace client initialized successfully");
30
-
}
43
+
await this.credentialManager.login({
44
+
identifier: this.config.username,
45
+
password: this.config.password,
46
+
});
31
47
32
-
export async function createStreamplaceMessage(
33
-
text: string,
34
-
streamerDid: string,
35
-
replyTo?: {
36
-
rootUri: string;
37
-
rootCid: string;
38
-
parentUri: string;
39
-
parentCid: string;
40
-
},
41
-
): Promise<void> {
42
-
// Make sure client is initialized
43
-
if (!credentialManager || !rpcClient) {
44
-
await initStreamplaceClient();
48
+
this.initialized = true;
49
+
console.log(
50
+
`Streamplace client initialized for ${this.config.username}`,
51
+
);
45
52
}
46
53
47
-
// Create the message record according to the lexicon
48
-
const record = {
49
-
text: text, // Plain text - stream.place will handle richtext formatting
50
-
streamer: streamerDid,
51
-
reply: {},
52
-
createdAt: new Date().toISOString(),
53
-
};
54
+
// Ensure the client is initialized before use
55
+
private async ensureInitialized(): Promise<void> {
56
+
if (!this.initialized) {
57
+
await this.init();
58
+
}
59
+
}
54
60
55
-
// Add reply reference if provided
56
-
if (replyTo) {
57
-
record.reply = {
58
-
root: {
59
-
uri: replyTo.rootUri,
60
-
cid: replyTo.rootCid,
61
-
},
62
-
parent: {
63
-
uri: replyTo.parentUri,
64
-
cid: replyTo.parentCid,
65
-
},
61
+
// Create a chat message on Streamplace
62
+
async createMessage(
63
+
text: string,
64
+
streamerDid: string,
65
+
facets?: any,
66
+
replyTo?: {
67
+
rootUri: string;
68
+
rootCid: string;
69
+
parentUri: string;
70
+
parentCid: string;
71
+
},
72
+
): Promise<any> {
73
+
await this.ensureInitialized();
74
+
75
+
// Create the message record according to the lexicon
76
+
const record = {
77
+
text: text,
78
+
facets: facets,
79
+
streamer: streamerDid,
80
+
reply: {},
81
+
createdAt: new Date().toISOString(),
66
82
};
83
+
84
+
// Add reply reference if provided
85
+
if (replyTo) {
86
+
record.reply = {
87
+
root: {
88
+
uri: replyTo.rootUri,
89
+
cid: replyTo.rootCid,
90
+
},
91
+
parent: {
92
+
uri: replyTo.parentUri,
93
+
cid: replyTo.parentCid,
94
+
},
95
+
};
96
+
}
97
+
98
+
const result = await ok(
99
+
this.rpcClient.post("com.atproto.repo.createRecord", {
100
+
input: {
101
+
repo: this.credentialManager.session!.did,
102
+
collection: "place.stream.chat.message",
103
+
record: record,
104
+
},
105
+
}),
106
+
);
107
+
108
+
return result;
67
109
}
68
110
69
-
await ok(rpcClient!.post("com.atproto.repo.createRecord", {
70
-
input: {
71
-
repo: credentialManager!.session!.did,
72
-
collection: "place.stream.chat.message",
73
-
record: record,
74
-
},
75
-
}));
111
+
// Get the current session DID
112
+
getDid(): string | undefined {
113
+
return this.credentialManager.session?.did;
114
+
}
115
+
116
+
// Get shoutouts from a specific DID's repository
117
+
async getShoutouts(did: Did, pdsFallback?: string): Promise<any> {
118
+
await this.ensureInitialized();
119
+
120
+
const rpc = new Client({
121
+
// TODO: get dynamic PDS - for now using fallback or configured PDS
122
+
handler: simpleFetchHandler({
123
+
service: pdsFallback || this.config.pdsHostUrl,
124
+
}),
125
+
});
126
+
127
+
const shoutouts = await ok(
128
+
rpc.get("com.atproto.repo.listRecords", {
129
+
params: {
130
+
repo: did,
131
+
collection: "online.timtinkers.bot.shoutout",
132
+
limit: 100,
133
+
// TODO: cursor for pagination
134
+
reverse: false,
135
+
},
136
+
}),
137
+
);
138
+
139
+
return shoutouts;
140
+
}
76
141
}
77
142
78
-
// Handle resolver
143
+
// Shared handle resolver instance
79
144
const handleResolver = new CompositeHandleResolver({
80
145
strategy: "dns-first",
81
146
methods: {
···
86
151
},
87
152
});
88
153
89
-
export async function resolveHandle(handle: Handle) {
154
+
// Resolve a handle to a DID
155
+
export async function resolveHandle(handle: Handle): Promise<Did> {
90
156
if (!isHandle(handle)) {
91
157
throw new Error("Not a valid handle");
92
158
}
+211
-46
utils/streamplaceBot.ts
+211
-46
utils/streamplaceBot.ts
···
1
1
import RichtextBuilder from "@atcute/bluesky-richtext-builder";
2
-
import { createStreamplaceMessage, getShoutouts } from "./atcuteUtils.ts";
2
+
import { StreamplaceClient, StreamplaceClientConfig } from "./atcuteUtils.ts";
3
3
import { didResolver } from "./didResolver.ts";
4
-
import { ATPROTO_USERNAME } from "../env.ts";
5
4
6
5
export interface CommandHandler {
7
6
(message: JetstreamMessage, args: string[]): Promise<void> | void;
8
7
}
9
8
9
+
// Cached information about a chatter
10
+
export interface Chatter {
11
+
did: Did;
12
+
handle: Handle;
13
+
hasShoutout: boolean;
14
+
hasBeenGreeted: boolean;
15
+
}
16
+
17
+
// Shoutout record from the repository
18
+
interface ShoutoutRecord {
19
+
user: Did;
20
+
text: string;
21
+
facets?: any;
22
+
}
23
+
10
24
class StreamplaceBot {
11
-
private streamerDid: string;
25
+
private streamerDid: Did;
12
26
private commandPrefix: string;
13
27
private commands: Map<string, CommandHandler>;
14
28
private enabled: boolean;
15
-
private botDid: string | null = null;
29
+
private client: StreamplaceClient;
30
+
31
+
// Caching
32
+
private chatters: Map<Did, Chatter> = new Map();
33
+
private shoutouts: Map<Did, ShoutoutRecord> = new Map();
34
+
private shoutoutsLoaded: boolean = false;
16
35
17
36
/**
18
37
* Create a new StreamplaceBot
19
38
* @param streamerDid The DID of the streamer whose chat to respond in
39
+
* @param clientConfig Configuration for the bot's AT Protocol client
20
40
* @param commandPrefix The prefix that triggers bot commands (default: "!")
21
41
*/
22
-
constructor(streamerDid: string, commandPrefix = "!") {
42
+
constructor(
43
+
streamerDid: Did,
44
+
clientConfig: StreamplaceClientConfig,
45
+
commandPrefix = "!",
46
+
) {
23
47
this.streamerDid = streamerDid;
24
48
this.commandPrefix = commandPrefix;
25
49
this.commands = new Map();
26
50
this.enabled = true;
27
-
this.botDid = ATPROTO_USERNAME!;
51
+
this.client = new StreamplaceClient(clientConfig);
52
+
}
53
+
54
+
// Initialize the bot - must be called before use
55
+
async init(): Promise<void> {
56
+
// Initialize the AT Protocol client
57
+
await this.client.init();
58
+
59
+
// Load shoutouts into cache
60
+
await this.loadShoutouts();
28
61
29
62
// Register default commands
30
-
this.registerDefaultCommands();
63
+
await this.registerDefaultCommands();
64
+
65
+
console.log(
66
+
`StreamplaceBot initialized for streamer: ${this.streamerDid}`,
67
+
);
68
+
}
69
+
70
+
// Load all shoutouts for the streamer into the cache
71
+
private async loadShoutouts(): Promise<void> {
72
+
if (this.shoutoutsLoaded) return;
73
+
74
+
try {
75
+
const shoutoutsData = await this.client.getShoutouts(
76
+
this.streamerDid,
77
+
"https://pds.timtinkers.online",
78
+
);
79
+
80
+
// Cache all shoutouts
81
+
for (const record of shoutoutsData.records) {
82
+
const userDid = record.value.user as Did;
83
+
this.shoutouts.set(userDid, {
84
+
user: userDid,
85
+
text: record.value.text,
86
+
facets: record.value.facets,
87
+
});
88
+
89
+
// Also resolve and cache the handle for users with shoutouts
90
+
await this.getOrCacheChatter(userDid, true);
91
+
}
92
+
93
+
this.shoutoutsLoaded = true;
94
+
console.log(`Loaded ${this.shoutouts.size} shoutouts into cache`);
95
+
} catch (error) {
96
+
console.error("Error loading shoutouts:", error);
97
+
}
98
+
}
99
+
100
+
// Get a chatter from cache or fetch and cache their information
101
+
private async getOrCacheChatter(
102
+
did: Did,
103
+
hasShoutout: boolean = false,
104
+
): Promise<Chatter> {
105
+
// Check if already cached
106
+
let chatter = this.chatters.get(did);
107
+
108
+
if (!chatter) {
109
+
// Resolve handle and create new chatter entry
110
+
try {
111
+
const resolved = await didResolver.resolve(did);
112
+
chatter = {
113
+
did: did,
114
+
handle: resolved.handle as Handle,
115
+
hasShoutout: hasShoutout || this.shoutouts.has(did),
116
+
hasBeenGreeted: false,
117
+
};
118
+
this.chatters.set(did, chatter);
119
+
console.log(`Cached new chatter: ${chatter.handle}`);
120
+
} catch (error) {
121
+
console.error(`Error resolving handle for DID ${did}:`, error);
122
+
// Create minimal chatter entry even if resolution fails
123
+
chatter = {
124
+
did: did,
125
+
handle: did as Handle, // Fallback to DID
126
+
hasShoutout: hasShoutout || this.shoutouts.has(did),
127
+
hasBeenGreeted: false,
128
+
};
129
+
this.chatters.set(did, chatter);
130
+
}
131
+
} else if (hasShoutout && !chatter.hasShoutout) {
132
+
// Update shoutout status if needed
133
+
chatter.hasShoutout = true;
134
+
}
135
+
136
+
return chatter;
31
137
}
32
138
33
139
/**
···
40
146
const record = message.commit.record!;
41
147
const text = record.text.trim();
42
148
149
+
// Get or cache chatter information
150
+
const chatter = await this.getOrCacheChatter(message.did);
151
+
152
+
// Auto-greet first-time chatters with shoutout if they have one
153
+
if (!chatter.hasBeenGreeted) {
154
+
chatter.hasBeenGreeted = true;
155
+
156
+
if (chatter.hasShoutout) {
157
+
const shoutout = this.shoutouts.get(message.did);
158
+
if (shoutout) {
159
+
await this.sendMessage(shoutout.text, shoutout.facets);
160
+
}
161
+
}
162
+
}
163
+
43
164
// Check if message starts with command prefix
44
165
if (!text.startsWith(this.commandPrefix)) return;
45
166
···
52
173
const handler = this.commands.get(commandName);
53
174
if (handler) {
54
175
console.log(
55
-
`Executing command: ${commandName} from user: ${message.did}`,
176
+
`Executing command: ${commandName} from user: ${chatter.handle}`,
56
177
);
57
178
try {
58
179
await handler(message, args);
···
89
210
/**
90
211
* Send a message to the chat
91
212
* @param text Message text
213
+
* @param facets Optional facets for mentions, links, etc.
92
214
*/
93
215
async sendMessage(text: string, facets?: any): Promise<void> {
94
216
try {
95
-
const result = await createStreamplaceMessage(
217
+
const result = await this.client.createMessage(
96
218
text,
97
219
this.streamerDid,
98
220
facets,
99
221
);
100
-
// Store bot's DID if not already known (from the first message)
101
-
if (!this.botDid && result?.uri) {
102
-
const didMatch = result.uri.match(/at:\/\/(did:[^\/]+)/);
103
-
if (didMatch) {
104
-
this.botDid = didMatch[1];
105
-
console.log(`Bot DID identified as: ${this.botDid}`);
106
-
}
107
-
}
108
222
console.log(`Bot sent message: ${text}`);
109
223
} catch (error) {
110
224
console.error("Error sending bot message:", error);
111
225
}
112
226
}
113
227
114
-
// Get the bot's DID
115
-
getBotDid(): string | null {
116
-
return this.botDid;
228
+
/**
229
+
* Get the bot's DID
230
+
*/
231
+
getBotDid(): string | undefined {
232
+
return this.client.getDid();
117
233
}
118
234
119
-
// Get the streamer's DID
120
-
getStreamerDid(): string | null {
235
+
/**
236
+
* Get the streamer's DID
237
+
*/
238
+
getStreamerDid(): string {
121
239
return this.streamerDid;
122
240
}
123
241
124
-
// Register the default set of commands
125
-
private async registerDefaultCommands(): void {
126
-
// Get shoutouts
127
-
const shoutouts = await getShoutouts(this.streamerDid) as any;
242
+
/**
243
+
* Get cached chatter information
244
+
*/
245
+
getChatter(did: Did): Chatter | undefined {
246
+
return this.chatters.get(did);
247
+
}
128
248
249
+
/**
250
+
* Get all cached chatters
251
+
*/
252
+
getAllChatters(): Map<Did, Chatter> {
253
+
return new Map(this.chatters);
254
+
}
255
+
256
+
/**
257
+
* Clear the chatter cache
258
+
*/
259
+
clearChatterCache(): void {
260
+
this.chatters.clear();
261
+
console.log("Chatter cache cleared");
262
+
}
263
+
264
+
/**
265
+
* Reload shoutouts from the repository
266
+
*/
267
+
async reloadShoutouts(): Promise<void> {
268
+
this.shoutoutsLoaded = false;
269
+
this.shoutouts.clear();
270
+
await this.loadShoutouts();
271
+
}
272
+
273
+
// Register the default set of commands
274
+
private registerDefaultCommands(): void {
129
275
// Help command
130
276
this.registerCommand("commands", async (_message, _args) => {
131
277
const commandsList = Array.from(this.commands.keys())
···
135
281
await this.sendMessage(`Available commands: ${commandsList}`);
136
282
});
137
283
138
-
this.registerCommand("shoutout", async (message, args) => {
139
-
// TO-DO: caching
140
-
const senderHandle = await didResolver.resolve(message.did);
141
-
const shoutouteeDid = message.commit.record!.facets![0].features[0]!
142
-
.did as Did;
284
+
// Shoutout command
285
+
this.registerCommand("shoutout", async (message, _args) => {
286
+
if (!message.commit.record?.facets?.[0]?.features?.[0]?.did) {
287
+
await this.sendMessage(
288
+
"Please mention a user to give them a shoutout!",
289
+
);
290
+
return;
291
+
}
143
292
144
-
const foundRecord = shoutouts.records.find((record) =>
145
-
record.value.user === shoutouteeDid
293
+
const senderChatter = await this.getOrCacheChatter(message.did);
294
+
const shoutouteeDid = message.commit.record.facets[0].features[0]
295
+
.did as Did;
296
+
const shoutouteeChatter = await this.getOrCacheChatter(
297
+
shoutouteeDid,
146
298
);
147
-
if (foundRecord) {
148
-
await this.sendMessage(foundRecord.value.text, foundRecord.value.facets);
299
+
300
+
// Check if there's a custom shoutout
301
+
const customShoutout = this.shoutouts.get(shoutouteeDid);
302
+
303
+
if (customShoutout) {
304
+
await this.sendMessage(
305
+
customShoutout.text,
306
+
customShoutout.facets,
307
+
);
149
308
} else {
309
+
// Generic shoutout
150
310
const { text, facets } = new RichtextBuilder()
151
-
.addMention(`@${senderHandle.handle}`, message.did)
311
+
.addMention(`@${senderChatter.handle}`, message.did)
152
312
.addText(" gives ")
153
-
.addMention(`@${args[0].replace(/^@+/, "")}`, shoutouteeDid)
313
+
.addMention(`@${shoutouteeChatter.handle}`, shoutouteeDid)
154
314
.addText(" a shoutout!");
155
315
156
316
await this.sendMessage(text, facets);
···
158
318
});
159
319
160
320
// Hug command
161
-
this.registerCommand("hug", async (message, args) => {
162
-
// TO-DO: caching
163
-
const senderHandle = await didResolver.resolve(message.did);
164
-
const huggeeDid = message.commit.record!.facets![0].features[0]!
321
+
this.registerCommand("hug", async (message, _args) => {
322
+
if (!message.commit.record?.facets?.[0]?.features?.[0]?.did) {
323
+
await this.sendMessage(
324
+
"Please mention a user to give them a hug!",
325
+
);
326
+
return;
327
+
}
328
+
329
+
const senderChatter = await this.getOrCacheChatter(message.did);
330
+
const huggeeDid = message.commit.record.facets[0].features[0]
165
331
.did as Did;
332
+
const huggeeChatter = await this.getOrCacheChatter(huggeeDid);
166
333
167
334
const { text, facets } = new RichtextBuilder()
168
-
.addMention(`@${senderHandle.handle}`, message.did)
335
+
.addMention(`@${senderChatter.handle}`, message.did)
169
336
.addText(" gives ")
170
-
.addMention(`@${args[0].replace(/^@+/, "")}`, huggeeDid)
337
+
.addMention(`@${huggeeChatter.handle}`, huggeeDid)
171
338
.addText(" a big hug!");
172
339
173
340
await this.sendMessage(text, facets);
···
175
342
}
176
343
}
177
344
178
-
export const streamplaceBot = new StreamplaceBot(
179
-
"did:plc:o6xucog6fghiyrvp7pyqxcs3",
180
-
);
345
+
export default StreamplaceBot;
+11
-2
utils/websocket.ts
+11
-2
utils/websocket.ts
···
1
1
import { JETSTREAM_URL } from "../env.ts";
2
2
import { didResolver } from "./didResolver.ts";
3
+
import { streamplaceBot } from "../main.ts";
3
4
4
5
// Client subscription message
5
6
interface SubscriptionMessage {
···
8
9
}
9
10
10
11
// WebSocket service class
11
-
export class StreamplaceWebSocketService {
12
+
class StreamplaceWebSocketService {
12
13
private jetstreamWs: WebSocket | null = null;
13
14
private clients = new Map<WebSocket, Set<string>>(); // client -> subscribed streamers
14
15
private streamerClients = new Map<string, Set<WebSocket>>(); // streamer -> clients
···
52
53
if (
53
54
jetstreamMessage.kind === "commit" &&
54
55
jetstreamMessage.commit?.operation === "create" &&
55
-
jetstreamMessage.commit.record
56
+
jetstreamMessage.commit.record &&
57
+
// TODO: make flexible later
58
+
jetstreamMessage.commit.record.streamer ===
59
+
"did:plc:o6xucog6fghiyrvp7pyqxcs3"
56
60
) {
57
61
this.processJetstreamMessage(jetstreamMessage);
58
62
}
···
138
142
}
139
143
140
144
private async processJetstreamMessage(jetstreamMessage: JetstreamMessage) {
145
+
// Let bot handle it
146
+
streamplaceBot.processMessage(jetstreamMessage);
147
+
141
148
// Extract the record data
142
149
const record = jetstreamMessage.commit.record!;
143
150
const enrichedMessage = await this.enrichMessage(
···
187
194
};
188
195
}
189
196
}
197
+
198
+
export const streamplaceWS = new StreamplaceWebSocketService();