providing password reset services for a long while: circa 2025

feat: add basic queue

+193
data/slack-queue.db

This is a binary file and will not be displayed.

+159
features/message-queue.ts
··· 1 + import { Database } from "bun:sqlite"; 2 + import type { Block, SlackAPIClient } from "slack-edge"; 3 + 4 + export interface SlackMessage { 5 + userId?: string; 6 + channelId?: string; 7 + blocks?: Block[]; 8 + text: string; 9 + timestamp?: number; 10 + status: "pending" | "sent" | "failed"; 11 + } 12 + 13 + export default class SlackMessageQueue { 14 + private db: Database; 15 + private slack: SlackAPIClient; 16 + private isProcessing = false; 17 + private batchSize = 50; 18 + 19 + constructor(slackClient: SlackAPIClient, dbPath = "slack-queue.db") { 20 + this.slack = slackClient; 21 + this.db = new Database(dbPath); 22 + this.initDatabase(); 23 + } 24 + 25 + private initDatabase() { 26 + this.db.run(` 27 + CREATE TABLE IF NOT EXISTS messages ( 28 + id INTEGER PRIMARY KEY AUTOINCREMENT, 29 + userId TEXT, 30 + channelId TEXT, 31 + blocks TEXT, 32 + text TEXT NOT NULL, 33 + timestamp INTEGER NOT NULL, 34 + status TEXT NOT NULL 35 + ) 36 + `); 37 + this.db.run("CREATE INDEX IF NOT EXISTS idx_status ON messages(status)"); 38 + } 39 + 40 + async enqueue(message: SlackMessage): Promise<void> { 41 + const stmt = this.db.prepare(` 42 + INSERT INTO messages (userId, channelId, blocks, text, timestamp, status) 43 + VALUES (?, ?, ?, ?, ?, ?) 44 + `); 45 + 46 + stmt.run( 47 + message.userId ?? null, 48 + message.channelId ?? null, 49 + JSON.stringify(message.blocks) ?? null, 50 + message.text, 51 + Date.now(), 52 + "pending", 53 + ); 54 + 55 + if (!this.isProcessing) { 56 + this.processQueue(); 57 + } 58 + } 59 + 60 + private async processQueue() { 61 + if (this.isProcessing) return; 62 + this.isProcessing = true; 63 + 64 + console.log("Processing queue"); 65 + 66 + try { 67 + while (true) { 68 + const messages = this.db 69 + .prepare( 70 + ` 71 + SELECT * FROM messages 72 + WHERE status = 'pending' 73 + LIMIT ? 74 + `, 75 + ) 76 + .all(this.batchSize) as (SlackMessage & { id: number })[]; 77 + 78 + console.log(messages); 79 + if (messages.length === 0) break; 80 + 81 + await Promise.all( 82 + messages.map(async (message) => { 83 + try { 84 + if (message.channelId) { 85 + await this.slack.chat.postMessage({ 86 + channel: message.channelId, 87 + blocks: 88 + JSON.parse(message.blocks as unknown as string) ?? 89 + undefined, 90 + text: message.text, 91 + }); 92 + 93 + console.log(res); 94 + } else if (message.userId) { 95 + await this.slack.chat.postMessage({ 96 + channel: message.userId, 97 + blocks: 98 + JSON.parse(message.blocks as unknown as string) ?? 99 + undefined, 100 + text: message.text, 101 + }); 102 + } 103 + 104 + console.log("Message sent successfully"); 105 + 106 + this.db 107 + .prepare( 108 + ` 109 + UPDATE messages 110 + SET status = 'sent' 111 + WHERE id = ? 112 + `, 113 + ) 114 + .run(message.id); 115 + } catch (error) { 116 + console.error("Error sending message", error); 117 + this.db 118 + .prepare( 119 + ` 120 + UPDATE messages 121 + SET status = 'failed' 122 + WHERE id = ? 123 + `, 124 + ) 125 + .run(message.id); 126 + } 127 + }), 128 + ); 129 + } 130 + } finally { 131 + this.isProcessing = false; 132 + } 133 + } 134 + 135 + async cleanup(olderThan: number = 7 * 24 * 60 * 60 * 1000) { 136 + const cutoff = Date.now() - olderThan; 137 + this.db 138 + .prepare( 139 + ` 140 + DELETE FROM messages 141 + WHERE timestamp < ? AND status != 'pending' 142 + `, 143 + ) 144 + .run(cutoff); 145 + } 146 + 147 + async queueLength() { 148 + const result = this.db 149 + .prepare( 150 + ` 151 + SELECT COUNT(*) as count 152 + FROM messages 153 + WHERE status = 'pending' 154 + `, 155 + ) 156 + .get(); 157 + return (result as { count: number }).count; 158 + } 159 + }
+34
index.ts
··· 1 1 import { SlackApp } from "slack-edge"; 2 + import SlackMessageQueue, { type SlackMessage } from "./features/message-queue"; 2 3 3 4 import * as features from "./features/index"; 4 5 ··· 33 34 }); 34 35 const slackClient = slackApp.client; 35 36 37 + const messageQueue = new SlackMessageQueue(slackClient, "data/slack-queue.db"); 38 + console.log(`👔 Message Queue Size: ${await messageQueue.queueLength()}`); 39 + 36 40 console.log(`⚒️ Loading ${Object.entries(features).length} features...`); 37 41 for (const [feature, handler] of Object.entries(features)) { 38 42 console.log(`📦 ${feature} loaded`); ··· 54 58 return new Response("OK"); 55 59 case "/slack": 56 60 return slackApp.run(request); 61 + case "/slack/message": { 62 + if ( 63 + request.headers.get("Authorization") !== 64 + `Bearer ${process.env.API_KEY}` 65 + ) { 66 + return new Response("Unauthorized", { status: 401 }); 67 + } 68 + 69 + const message: SlackMessage = await request.json(); 70 + const { userId, channelId, text } = message; 71 + 72 + if ((!userId && !channelId) || (userId && channelId) || !text) { 73 + return new Response( 74 + `Invalid fields: ${[ 75 + !userId && 76 + !channelId && 77 + "must provide either userId or channelId", 78 + userId && channelId && "cannot provide both userId and channelId", 79 + !text && "text is required", 80 + ] 81 + .filter(Boolean) 82 + .join(", ")}`, 83 + { status: 400 }, 84 + ); 85 + } 86 + 87 + await messageQueue.enqueue(message); 88 + 89 + return new Response(JSON.stringify({ ok: true }), { status: 200 }); 90 + } 57 91 default: 58 92 return new Response("404 Not Found", { status: 404 }); 59 93 }