/** * MessageStore - SQLite-based persistence for the messaging system. * * Messages are stored in the `messages` table with indexed lookups * for inbox/outbox/conversation queries. * * Follows the same patterns as ChatStore: * - EventEmitter for change notifications * - SQLite for structured data storage (synchronous) * - JSON columns for optional metadata */ import { EventEmitter } from "node:events"; import { randomUUID } from "node:crypto"; import type { Database } from "./db.js"; import { fromJson, toJsonNullable } from "./db.js"; import { createLogger } from "./logger.js"; import { DASHBOARD_USER_ID, normalizeMessageParticipant, validateMessageMetadata, type Message, type MessageCreateInput, type MessageFilter, type MessageType, type Mailbox, type ParticipantType } from "./types.js"; const messageStoreLog = createLogger("message-store"); // ── Event Types ───────────────────────────────────────────────────── /** Events emitted by MessageStore */ export interface MessageStoreEvents { /** Emitted when a new message is created and sent */ "message:sent": [message: Message]; /** Emitted when a message is received by a participant */ "message:received": [message: Message]; /** Emitted when a message is marked as read */ "message:read": [message: Message]; /** Emitted when a message is deleted */ "message:deleted": [messageId: string]; } // ── Row Interfaces ─────────────────────────────────────────────────── /** Database row shape for the messages table. */ interface MessageRow { id: string; fromId: string; fromType: string; toId: string; toType: string; content: string; type: string; read: number; metadata: string | null; createdAt: string; updatedAt: string; } // ── Options Types ──────────────────────────────────────────────────── /** Options for MessageStore constructor */ export interface MessageStoreOptions { /** Optional hook invoked when a message is addressed to an agent */ onMessageToAgent?: (message: Message) => void; } // ── MessageStore Class ─────────────────────────────────────────────── /** * MessageStore manages messages between agents, users, and the system. * Uses SQLite for persistent storage with efficient indexed queries. */ export class MessageStore extends EventEmitter { private onMessageToAgent?: (message: Message) => void; // Prepared statements for frequently-run queries private stmtInsert!: ReturnType; private stmtGetById!: ReturnType; private stmtUpdateRead!: ReturnType; private stmtDelete!: ReturnType; constructor( private db: Database, options?: MessageStoreOptions, ) { super(); this.setMaxListeners(100); this.onMessageToAgent = options?.onMessageToAgent; // Prepare frequently-run statements this.stmtInsert = this.db.prepare(` INSERT INTO messages (id, fromId, fromType, toId, toType, content, type, read, metadata, createdAt, updatedAt) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `); this.stmtGetById = this.db.prepare(` SELECT * FROM messages WHERE id = ? `); this.stmtUpdateRead = this.db.prepare(` UPDATE messages SET read = 1, updatedAt = ? WHERE id = ? `); this.stmtDelete = this.db.prepare(` DELETE FROM messages WHERE id = ? `); } // ── Row-to-Object Converters ─────────────────────────────────────── /** * Convert a database row to a Message object. */ private rowToMessage(row: MessageRow): Message { return { id: row.id, fromId: row.fromId, fromType: row.fromType as ParticipantType, toId: row.toId, toType: row.toType as ParticipantType, content: row.content, type: row.type as MessageType, read: row.read === 1, metadata: fromJson(row.metadata), createdAt: row.createdAt, updatedAt: row.updatedAt, }; } // ── Public API ──────────────────────────────────────────────────── /** * Create and store a new message. * @param input - Message creation parameters * @returns The created message */ sendMessage(input: MessageCreateInput): Message { validateMessageMetadata(input.metadata); const now = new Date().toISOString(); const messageId = `msg-${randomUUID().slice(0, 8)}`; const from = normalizeMessageParticipant(input.fromId ?? "system", input.fromType ?? "system"); const to = normalizeMessageParticipant(input.toId, input.toType); const message: Message = { id: messageId, fromId: from.id, fromType: from.type, toId: to.id, toType: to.type, content: input.content, type: input.type, read: false, metadata: input.metadata, createdAt: now, updatedAt: now, }; this.stmtInsert.run( message.id, message.fromId, message.fromType, message.toId, message.toType, message.content, message.type, message.read ? 1 : 0, toJsonNullable(message.metadata), message.createdAt, message.updatedAt, ); this.db.bumpLastModified(); messageStoreLog.log(`MessageStore emitting message:sent id=${message.id} type=${message.type} fromId=${message.fromId} toId=${message.toId}`); this.emit("message:sent", message); this.emit("message:received", message); if (message.toType === "agent" && this.onMessageToAgent) { this.onMessageToAgent(message); } return message; } /** * Get a single message by ID. * @param id - The message ID * @returns The message, or null if not found */ getMessage(id: string): Message | null { const row = this.stmtGetById.get(id) as unknown as MessageRow | undefined; if (!row) return null; return this.rowToMessage(row); } /** * Get inbox messages for a participant (messages where they are the recipient). * @param ownerId - The participant ID * @param ownerType - The participant type * @param filter - Optional filter criteria * @returns Array of messages (newest first) */ getInbox( ownerId: string, ownerType: ParticipantType, filter?: MessageFilter, ): Message[] { return this.queryMessagesByParticipant("to", ownerId, ownerType, filter); } /** * Get outbox messages for a participant (messages they sent). * @param ownerId - The participant ID * @param ownerType - The participant type * @param filter - Optional filter criteria * @returns Array of messages (newest first) */ getOutbox( ownerId: string, ownerType: ParticipantType, filter?: MessageFilter, ): Message[] { return this.queryMessagesByParticipant("from", ownerId, ownerType, filter); } private getParticipantIdsForLookup(ownerId: string, ownerType: ParticipantType): string[] { if (ownerType === "user" && ownerId === DASHBOARD_USER_ID) { return [DASHBOARD_USER_ID, "user:dashboard", "User: user:dashboard"]; } return [ownerId]; } private queryMessagesByParticipant( direction: "to" | "from", ownerId: string, ownerType: ParticipantType, filter?: MessageFilter, ): Message[] { const idCol = direction === "to" ? "toId" : "fromId"; const typeCol = direction === "to" ? "toType" : "fromType"; const participantIds = this.getParticipantIdsForLookup(ownerId, ownerType); const idPredicate = participantIds.length === 1 ? `${idCol} = ?` : `${idCol} IN (${participantIds.map(() => "?").join(", ")})`; const whereClauses: string[] = [idPredicate, `${typeCol} = ?`]; const params: (string | number)[] = [...participantIds, ownerType]; if (filter?.type) { whereClauses.push("type = ?"); params.push(filter.type); } if (filter?.read !== undefined) { whereClauses.push("read = ?"); params.push(filter.read ? 1 : 0); } const whereSql = whereClauses.join(" AND "); const limit = filter?.limit ?? 100; const offset = filter?.offset ?? 0; const rows = this.db.prepare(` SELECT * FROM messages WHERE ${whereSql} ORDER BY createdAt DESC, rowid DESC LIMIT ? OFFSET ? `).all(...params, limit, offset); return (rows as unknown as MessageRow[]).map((row) => this.rowToMessage(row)); } /** * Mark a message as read. * @param messageId - The message ID * @returns The updated message * @throws Error if message not found */ markAsRead(messageId: string): Message { // First check if the message exists const existing = this.getMessage(messageId); if (!existing) { throw new Error(`Message ${messageId} not found`); } if (existing.read) return existing; const now = new Date().toISOString(); this.stmtUpdateRead.run(now, messageId); this.db.bumpLastModified(); const updated = this.getMessage(messageId); this.emit("message:read", updated!); return updated!; } /** * Mark all inbox messages as read for a participant. * @param ownerId - The participant ID * @param ownerType - The participant type * @returns Number of messages marked as read */ markAllAsRead( ownerId: string, ownerType: ParticipantType, ): number { const now = new Date().toISOString(); const participantIds = this.getParticipantIdsForLookup(ownerId, ownerType); const toIdPredicate = participantIds.length === 1 ? "toId = ?" : `toId IN (${participantIds.map(() => "?").join(", ")})`; // Get count of unread messages before updating const unreadRow = this.db.prepare(` SELECT COUNT(*) as count FROM messages WHERE ${toIdPredicate} AND toType = ? AND read = 0 `).get(...participantIds, ownerType) as { count: number } | undefined; const count = unreadRow?.count ?? 0; // Mark all as read this.db.prepare(` UPDATE messages SET read = 1, updatedAt = ? WHERE ${toIdPredicate} AND toType = ? AND read = 0 `).run(now, ...participantIds, ownerType); this.db.bumpLastModified(); return count; } /** * Delete a message by ID. * @param id - The message ID * @throws Error if message not found */ deleteMessage(id: string): void { // First check if the message exists const existing = this.getMessage(id); if (!existing) { throw new Error(`Message ${id} not found`); } this.stmtDelete.run(id); this.db.bumpLastModified(); this.emit("message:deleted", id); } /** * Get all messages between two participants (conversation view). * @param participantA - First participant * @param participantB - Second participant * @returns Array of messages (oldest first for conversation ordering) */ getConversation( participantA: { id: string; type: ParticipantType }, participantB: { id: string; type: ParticipantType }, ): Message[] { const participantAIds = this.getParticipantIdsForLookup(participantA.id, participantA.type); const participantBIds = this.getParticipantIdsForLookup(participantB.id, participantB.type); const participantAFromPredicate = participantAIds.length === 1 ? "fromId = ?" : `fromId IN (${participantAIds.map(() => "?").join(", ")})`; const participantAToPredicate = participantAIds.length === 1 ? "toId = ?" : `toId IN (${participantAIds.map(() => "?").join(", ")})`; const participantBFromPredicate = participantBIds.length === 1 ? "fromId = ?" : `fromId IN (${participantBIds.map(() => "?").join(", ")})`; const participantBToPredicate = participantBIds.length === 1 ? "toId = ?" : `toId IN (${participantBIds.map(() => "?").join(", ")})`; // Find messages where either participant is sender or receiver const rows = this.db.prepare(` SELECT * FROM messages WHERE ( (${participantAFromPredicate} AND fromType = ? AND ${participantBToPredicate} AND toType = ?) OR (${participantBFromPredicate} AND fromType = ? AND ${participantAToPredicate} AND toType = ?) ) ORDER BY createdAt ASC `).all( ...participantAIds, participantA.type, ...participantBIds, participantB.type, ...participantBIds, participantB.type, ...participantAIds, participantA.type, ); return (rows as unknown as MessageRow[]).map((row) => this.rowToMessage(row)); } /** * Get mailbox summary for a participant. * @param ownerId - The participant ID * @param ownerType - The participant type * @returns Mailbox summary with unread count and last message */ getMailbox( ownerId: string, ownerType: ParticipantType, ): Mailbox { const participantIds = this.getParticipantIdsForLookup(ownerId, ownerType); const toIdPredicate = participantIds.length === 1 ? "toId = ?" : `toId IN (${participantIds.map(() => "?").join(", ")})`; const unreadRow = this.db.prepare(` SELECT COUNT(*) as count FROM messages WHERE ${toIdPredicate} AND toType = ? AND read = 0 `).get(...participantIds, ownerType) as { count: number } | undefined; const unreadCount = unreadRow?.count ?? 0; const lastRow = this.db.prepare(` SELECT * FROM messages WHERE ${toIdPredicate} AND toType = ? ORDER BY createdAt DESC, rowid DESC LIMIT 1 `).get(...participantIds, ownerType) as unknown as MessageRow | undefined; const lastMessage = lastRow ? this.rowToMessage(lastRow) : undefined; return { ownerId, ownerType, unreadCount, lastMessage, }; } /** * Get all agent-to-agent messages across all agents. * @returns Array of messages (newest first) */ getAllAgentToAgentMessages(): Message[] { const rows = this.db.prepare(` SELECT * FROM messages WHERE type = ? ORDER BY createdAt DESC, rowid DESC `).all("agent-to-agent"); return (rows as unknown as MessageRow[]).map((row) => this.rowToMessage(row)); } /** * Get unread count across all agent-to-agent messages. */ getUnreadAgentToAgentCount(): number { const row = this.db.prepare(` SELECT COUNT(*) as count FROM messages WHERE type = ? AND read = 0 `).get("agent-to-agent") as { count: number } | undefined; return row?.count ?? 0; } /** * Set or update the hook used when messages are sent to agents. */ setMessageToAgentHook(hook: (message: Message) => void): void { this.onMessageToAgent = hook; } }