feat(dashboard): send room messages from web (#33)
This commit is contained in:
@@ -22,6 +22,7 @@ export {
|
||||
getOpenWorkItem,
|
||||
getOpenWorkItemForChat,
|
||||
getRecentChatMessages,
|
||||
hasMessage,
|
||||
hasRecentRestartAnnouncement,
|
||||
markWorkItemDelivered,
|
||||
markWorkItemDeliveryRetry,
|
||||
|
||||
@@ -95,6 +95,17 @@ export function getAllChatsFromDatabase(database: Database): ChatInfo[] {
|
||||
.all() as ChatInfo[];
|
||||
}
|
||||
|
||||
export function hasMessageInDatabase(
|
||||
database: Database,
|
||||
chatJid: string,
|
||||
id: string,
|
||||
): boolean {
|
||||
const row = database
|
||||
.prepare('SELECT 1 FROM messages WHERE chat_jid = ? AND id = ? LIMIT 1')
|
||||
.get(chatJid, id);
|
||||
return !!row;
|
||||
}
|
||||
|
||||
export function storeMessageInDatabase(
|
||||
database: Database,
|
||||
msg: NewMessage,
|
||||
|
||||
@@ -26,6 +26,7 @@ import {
|
||||
getNewMessagesBySeqFromDatabase,
|
||||
getNewMessagesFromDatabase,
|
||||
getRecentChatMessagesFromDatabase,
|
||||
hasMessageInDatabase,
|
||||
hasRecentRestartAnnouncementInDatabase,
|
||||
storeChatMetadataInDatabase,
|
||||
storeMessageInDatabase,
|
||||
@@ -141,6 +142,10 @@ export function storeMessage(msg: NewMessage): void {
|
||||
storeMessageInDatabase(requireDatabase(), msg);
|
||||
}
|
||||
|
||||
export function hasMessage(chatJid: string, id: string): boolean {
|
||||
return hasMessageInDatabase(requireDatabase(), chatJid, id);
|
||||
}
|
||||
|
||||
export function getNewMessages(
|
||||
jids: string[],
|
||||
lastTimestamp: string,
|
||||
|
||||
@@ -521,7 +521,12 @@ async function main(): Promise<void> {
|
||||
},
|
||||
purgeOnStart: true,
|
||||
});
|
||||
webDashboardServer = startWebDashboardServer(WEB_DASHBOARD);
|
||||
webDashboardServer = startWebDashboardServer({
|
||||
...WEB_DASHBOARD,
|
||||
getRoomBindings: runtimeState.getRoomBindings,
|
||||
enqueueMessageCheck: (chatJid, groupFolder) =>
|
||||
queue.enqueueMessageCheck(chatJid, resolveGroupIpcPath(groupFolder)),
|
||||
});
|
||||
|
||||
leaseRecoveryTimer = setInterval(() => {
|
||||
const failover = getGlobalFailoverInfo();
|
||||
|
||||
@@ -5,7 +5,12 @@ import path from 'path';
|
||||
import { afterEach, describe, expect, it } from 'vitest';
|
||||
|
||||
import type { StatusSnapshot } from './status-dashboard.js';
|
||||
import type { PairedTask, ScheduledTask } from './types.js';
|
||||
import type {
|
||||
NewMessage,
|
||||
PairedTask,
|
||||
RegisteredGroup,
|
||||
ScheduledTask,
|
||||
} from './types.js';
|
||||
import { createWebDashboardHandler } from './web-dashboard-server.js';
|
||||
|
||||
const tempDirs: string[] = [];
|
||||
@@ -344,6 +349,263 @@ describe('web dashboard server handler', () => {
|
||||
expect(wrongMethod.status).toBe(405);
|
||||
});
|
||||
|
||||
it('injects room messages and queues room work from the web dashboard', async () => {
|
||||
const messages: NewMessage[] = [];
|
||||
const metadata: Array<{
|
||||
chatJid: string;
|
||||
timestamp: string;
|
||||
name?: string;
|
||||
channel?: string;
|
||||
isGroup?: boolean;
|
||||
}> = [];
|
||||
const queued: Array<{ chatJid: string; groupFolder: string }> = [];
|
||||
const rooms: Record<string, RegisteredGroup> = {
|
||||
'dc:ops': {
|
||||
name: '#ops',
|
||||
folder: 'ops-room',
|
||||
added_at: '2026-04-26T05:00:00.000Z',
|
||||
},
|
||||
};
|
||||
const handler = createWebDashboardHandler({
|
||||
readStatusSnapshots: () => [],
|
||||
getTasks: () => [],
|
||||
getPairedTasks: () => [],
|
||||
getRoomBindings: () => rooms,
|
||||
storeChatMetadata: (chatJid, timestamp, name, channel, isGroup) => {
|
||||
metadata.push({ chatJid, timestamp, name, channel, isGroup });
|
||||
},
|
||||
storeMessage: (message) => {
|
||||
messages.push(message);
|
||||
},
|
||||
hasMessage: () => false,
|
||||
enqueueMessageCheck: (chatJid, groupFolder) => {
|
||||
queued.push({ chatJid, groupFolder });
|
||||
},
|
||||
now: () => '2026-04-26T05:10:00.000Z',
|
||||
});
|
||||
|
||||
const response = await handler(
|
||||
new Request(
|
||||
`http://localhost/api/rooms/${encodeURIComponent('dc:ops')}/messages`,
|
||||
{
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body: JSON.stringify({
|
||||
requestId: 'room-compose-1',
|
||||
text: ' run a dashboard check ',
|
||||
}),
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
await expect(response.json()).resolves.toMatchObject({
|
||||
ok: true,
|
||||
queued: true,
|
||||
});
|
||||
expect(metadata).toEqual([
|
||||
{
|
||||
chatJid: 'dc:ops',
|
||||
timestamp: '2026-04-26T05:10:00.000Z',
|
||||
name: '#ops',
|
||||
channel: 'web-dashboard',
|
||||
isGroup: true,
|
||||
},
|
||||
]);
|
||||
expect(messages).toHaveLength(1);
|
||||
expect(messages[0]).toMatchObject({
|
||||
chat_jid: 'dc:ops',
|
||||
sender: 'web-dashboard',
|
||||
sender_name: 'Web Dashboard',
|
||||
content: 'run a dashboard check',
|
||||
timestamp: '2026-04-26T05:10:00.000Z',
|
||||
is_from_me: false,
|
||||
is_bot_message: false,
|
||||
message_source_kind: 'ipc_injected_human',
|
||||
});
|
||||
expect(messages[0]?.id).toBe('web-room-compose-1');
|
||||
expect(queued).toEqual([{ chatJid: 'dc:ops', groupFolder: 'ops-room' }]);
|
||||
});
|
||||
|
||||
it('deduplicates repeated room message request ids', async () => {
|
||||
const messages: NewMessage[] = [];
|
||||
const queued: Array<{ chatJid: string; groupFolder: string }> = [];
|
||||
const existing = new Set<string>();
|
||||
const handler = createWebDashboardHandler({
|
||||
readStatusSnapshots: () => [],
|
||||
getTasks: () => [],
|
||||
getPairedTasks: () => [],
|
||||
getRoomBindings: () => ({
|
||||
'dc:ops': {
|
||||
name: '#ops',
|
||||
folder: 'ops-room',
|
||||
added_at: '2026-04-26T05:00:00.000Z',
|
||||
},
|
||||
}),
|
||||
storeChatMetadata: () => undefined,
|
||||
storeMessage: (message) => {
|
||||
messages.push(message);
|
||||
existing.add(`${message.chat_jid}:${message.id}`);
|
||||
},
|
||||
hasMessage: (chatJid, id) => existing.has(`${chatJid}:${id}`),
|
||||
enqueueMessageCheck: (chatJid, groupFolder) => {
|
||||
queued.push({ chatJid, groupFolder });
|
||||
},
|
||||
});
|
||||
const request = () =>
|
||||
new Request(
|
||||
`http://localhost/api/rooms/${encodeURIComponent('dc:ops')}/messages`,
|
||||
{
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body: JSON.stringify({
|
||||
requestId: 'same-submit',
|
||||
text: 'repeat me',
|
||||
}),
|
||||
},
|
||||
);
|
||||
|
||||
const first = await handler(request());
|
||||
const second = await handler(request());
|
||||
|
||||
expect(first.status).toBe(200);
|
||||
expect(second.status).toBe(200);
|
||||
await expect(first.json()).resolves.toMatchObject({
|
||||
id: 'web-same-submit',
|
||||
queued: true,
|
||||
});
|
||||
await expect(second.json()).resolves.toMatchObject({
|
||||
id: 'web-same-submit',
|
||||
queued: false,
|
||||
duplicate: true,
|
||||
});
|
||||
expect(messages).toHaveLength(1);
|
||||
expect(queued).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('deduplicates existing room message request ids after restart', async () => {
|
||||
const handler = createWebDashboardHandler({
|
||||
readStatusSnapshots: () => [],
|
||||
getTasks: () => [],
|
||||
getPairedTasks: () => [],
|
||||
getRoomBindings: () => ({
|
||||
'dc:ops': {
|
||||
name: '#ops',
|
||||
folder: 'ops-room',
|
||||
added_at: '2026-04-26T05:00:00.000Z',
|
||||
},
|
||||
}),
|
||||
storeChatMetadata: () => {
|
||||
throw new Error('storeChatMetadata should not be called');
|
||||
},
|
||||
storeMessage: () => {
|
||||
throw new Error('storeMessage should not be called');
|
||||
},
|
||||
hasMessage: (chatJid, id) =>
|
||||
chatJid === 'dc:ops' && id === 'web-previous-submit',
|
||||
enqueueMessageCheck: () => {
|
||||
throw new Error('enqueueMessageCheck should not be called');
|
||||
},
|
||||
});
|
||||
|
||||
const response = await handler(
|
||||
new Request(
|
||||
`http://localhost/api/rooms/${encodeURIComponent('dc:ops')}/messages`,
|
||||
{
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body: JSON.stringify({
|
||||
requestId: 'previous-submit',
|
||||
text: 'repeat after restart',
|
||||
}),
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
await expect(response.json()).resolves.toMatchObject({
|
||||
id: 'web-previous-submit',
|
||||
queued: false,
|
||||
duplicate: true,
|
||||
});
|
||||
});
|
||||
|
||||
it('rejects invalid room message requests', async () => {
|
||||
const handler = createWebDashboardHandler({
|
||||
readStatusSnapshots: () => [],
|
||||
getTasks: () => [],
|
||||
getPairedTasks: () => [],
|
||||
getRoomBindings: () => ({
|
||||
'dc:ops': {
|
||||
name: '#ops',
|
||||
folder: 'ops-room',
|
||||
added_at: '2026-04-26T05:00:00.000Z',
|
||||
},
|
||||
}),
|
||||
storeMessage: () => {
|
||||
throw new Error('storeMessage should not be called');
|
||||
},
|
||||
enqueueMessageCheck: () => {
|
||||
throw new Error('enqueueMessageCheck should not be called');
|
||||
},
|
||||
});
|
||||
|
||||
const empty = await handler(
|
||||
new Request(
|
||||
`http://localhost/api/rooms/${encodeURIComponent('dc:ops')}/messages`,
|
||||
{
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body: JSON.stringify({ text: ' ' }),
|
||||
},
|
||||
),
|
||||
);
|
||||
expect(empty.status).toBe(400);
|
||||
|
||||
const missing = await handler(
|
||||
new Request(
|
||||
`http://localhost/api/rooms/${encodeURIComponent('dc:missing')}/messages`,
|
||||
{
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body: JSON.stringify({ text: 'hello' }),
|
||||
},
|
||||
),
|
||||
);
|
||||
expect(missing.status).toBe(404);
|
||||
|
||||
const wrongMethod = await handler(
|
||||
new Request(
|
||||
`http://localhost/api/rooms/${encodeURIComponent('dc:ops')}/messages`,
|
||||
{
|
||||
method: 'GET',
|
||||
},
|
||||
),
|
||||
);
|
||||
expect(wrongMethod.status).toBe(405);
|
||||
});
|
||||
|
||||
it('returns 503 when room message injection is not wired', async () => {
|
||||
const handler = createWebDashboardHandler({
|
||||
readStatusSnapshots: () => [],
|
||||
getTasks: () => [],
|
||||
getPairedTasks: () => [],
|
||||
});
|
||||
|
||||
const response = await handler(
|
||||
new Request(
|
||||
`http://localhost/api/rooms/${encodeURIComponent('dc:ops')}/messages`,
|
||||
{
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body: JSON.stringify({ text: 'hello' }),
|
||||
},
|
||||
),
|
||||
);
|
||||
|
||||
expect(response.status).toBe(503);
|
||||
});
|
||||
|
||||
it('serves Vite static assets and falls back to index for SPA routes', async () => {
|
||||
const staticDir = fs.mkdtempSync(
|
||||
path.join(os.tmpdir(), 'ejclaw-dashboard-'),
|
||||
|
||||
@@ -7,6 +7,9 @@ import {
|
||||
getAllOpenPairedTasks,
|
||||
getAllTasks,
|
||||
getTaskById,
|
||||
hasMessage,
|
||||
storeChatMetadata,
|
||||
storeMessage,
|
||||
updateTask,
|
||||
} from './db.js';
|
||||
import { logger } from './logger.js';
|
||||
@@ -14,13 +17,19 @@ import {
|
||||
readStatusSnapshots,
|
||||
type StatusSnapshot,
|
||||
} from './status-dashboard.js';
|
||||
import type { PairedTask, ScheduledTask } from './types.js';
|
||||
import type {
|
||||
NewMessage,
|
||||
PairedTask,
|
||||
RegisteredGroup,
|
||||
ScheduledTask,
|
||||
} from './types.js';
|
||||
import {
|
||||
buildWebDashboardOverview,
|
||||
sanitizeScheduledTask,
|
||||
} from './web-dashboard-data.js';
|
||||
|
||||
const DEFAULT_STATUS_MAX_AGE_MS = 10 * 60 * 1000;
|
||||
const ROOM_MESSAGE_ID_CACHE_LIMIT = 500;
|
||||
|
||||
export interface WebDashboardHandlerOptions {
|
||||
staticDir?: string;
|
||||
@@ -34,6 +43,17 @@ export interface WebDashboardHandlerOptions {
|
||||
) => void;
|
||||
deleteTask?: (id: string) => void;
|
||||
getPairedTasks?: () => PairedTask[];
|
||||
getRoomBindings?: () => Record<string, RegisteredGroup>;
|
||||
storeChatMetadata?: (
|
||||
chatJid: string,
|
||||
timestamp: string,
|
||||
name?: string,
|
||||
channel?: string,
|
||||
isGroup?: boolean,
|
||||
) => void;
|
||||
storeMessage?: (message: NewMessage) => void;
|
||||
hasMessage?: (chatJid: string, id: string) => boolean;
|
||||
enqueueMessageCheck?: (chatJid: string, groupFolder: string) => void;
|
||||
now?: () => string;
|
||||
}
|
||||
|
||||
@@ -125,6 +145,16 @@ function parseTaskActionPath(pathname: string): string | null {
|
||||
}
|
||||
}
|
||||
|
||||
function parseRoomMessagePath(pathname: string): string | null {
|
||||
const match = pathname.match(/^\/api\/rooms\/([^/]+)\/messages$/);
|
||||
if (!match) return null;
|
||||
try {
|
||||
return decodeURIComponent(match[1]);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function isTaskAction(value: unknown): value is TaskAction {
|
||||
return value === 'pause' || value === 'resume' || value === 'cancel';
|
||||
}
|
||||
@@ -138,6 +168,42 @@ async function readTaskAction(request: Request): Promise<TaskAction | null> {
|
||||
}
|
||||
}
|
||||
|
||||
function sanitizeRoomMessageRequestId(value: unknown): string | null {
|
||||
if (typeof value !== 'string') return null;
|
||||
const trimmed = value.trim();
|
||||
if (!trimmed) return null;
|
||||
const safe = trimmed.replace(/[^A-Za-z0-9._:-]/g, '-').slice(0, 120);
|
||||
return safe || null;
|
||||
}
|
||||
|
||||
async function readRoomMessageBody(
|
||||
request: Request,
|
||||
): Promise<{ text: string; requestId: string | null } | null> {
|
||||
try {
|
||||
const body = (await request.json()) as {
|
||||
text?: unknown;
|
||||
requestId?: unknown;
|
||||
};
|
||||
if (typeof body.text !== 'string') return null;
|
||||
const text = body.text.trim();
|
||||
if (!text) return null;
|
||||
return {
|
||||
text: text.length > 8000 ? text.slice(0, 8000) : text,
|
||||
requestId: sanitizeRoomMessageRequestId(body.requestId),
|
||||
};
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function makeWebMessageId(): string {
|
||||
return `web-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
|
||||
}
|
||||
|
||||
function makeWebMessageIdFromRequest(requestId: string | null): string {
|
||||
return requestId ? `web-${requestId}` : makeWebMessageId();
|
||||
}
|
||||
|
||||
export function createWebDashboardHandler(
|
||||
opts: WebDashboardHandlerOptions = {},
|
||||
): (request: Request) => Response | Promise<Response> {
|
||||
@@ -147,11 +213,30 @@ export function createWebDashboardHandler(
|
||||
const mutateTask = opts.updateTask ?? updateTask;
|
||||
const removeTask = opts.deleteTask ?? deleteTask;
|
||||
const loadPairedTasks = opts.getPairedTasks ?? getAllOpenPairedTasks;
|
||||
const loadRoomBindings = opts.getRoomBindings;
|
||||
const writeChatMetadata = opts.storeChatMetadata ?? storeChatMetadata;
|
||||
const writeMessage = opts.storeMessage ?? storeMessage;
|
||||
const messageExists = opts.hasMessage ?? hasMessage;
|
||||
const enqueueMessageCheck = opts.enqueueMessageCheck;
|
||||
const statusMaxAgeMs = opts.statusMaxAgeMs ?? DEFAULT_STATUS_MAX_AGE_MS;
|
||||
const seenRoomMessageIds: string[] = [];
|
||||
const seenRoomMessageIdSet = new Set<string>();
|
||||
|
||||
function rememberRoomMessageId(id: string): boolean {
|
||||
if (seenRoomMessageIdSet.has(id)) return false;
|
||||
seenRoomMessageIdSet.add(id);
|
||||
seenRoomMessageIds.push(id);
|
||||
if (seenRoomMessageIds.length > ROOM_MESSAGE_ID_CACHE_LIMIT) {
|
||||
const oldest = seenRoomMessageIds.shift();
|
||||
if (oldest) seenRoomMessageIdSet.delete(oldest);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
return async (request: Request): Promise<Response> => {
|
||||
const url = new URL(request.url);
|
||||
const actionTaskId = parseTaskActionPath(url.pathname);
|
||||
const messageRoomJid = parseRoomMessagePath(url.pathname);
|
||||
|
||||
if (actionTaskId) {
|
||||
if (request.method !== 'POST') {
|
||||
@@ -192,6 +277,64 @@ export function createWebDashboardHandler(
|
||||
});
|
||||
}
|
||||
|
||||
if (messageRoomJid) {
|
||||
if (request.method !== 'POST') {
|
||||
return jsonResponse({ error: 'Method not allowed' }, { status: 405 });
|
||||
}
|
||||
|
||||
if (!loadRoomBindings || !enqueueMessageCheck) {
|
||||
return jsonResponse(
|
||||
{ error: 'Room message injection is not configured' },
|
||||
{ status: 503 },
|
||||
);
|
||||
}
|
||||
|
||||
const body = await readRoomMessageBody(request);
|
||||
if (!body) {
|
||||
return jsonResponse(
|
||||
{ error: 'Message text is required' },
|
||||
{ status: 400 },
|
||||
);
|
||||
}
|
||||
|
||||
const binding = loadRoomBindings()[messageRoomJid];
|
||||
if (!binding) {
|
||||
return jsonResponse({ error: 'Room not found' }, { status: 404 });
|
||||
}
|
||||
|
||||
const timestamp = opts.now?.() ?? new Date().toISOString();
|
||||
const id = makeWebMessageIdFromRequest(body.requestId);
|
||||
if (body.requestId && messageExists(messageRoomJid, id)) {
|
||||
return jsonResponse({ ok: true, id, queued: false, duplicate: true });
|
||||
}
|
||||
|
||||
if (!rememberRoomMessageId(`${messageRoomJid}:${id}`)) {
|
||||
return jsonResponse({ ok: true, id, queued: false, duplicate: true });
|
||||
}
|
||||
|
||||
writeChatMetadata(
|
||||
messageRoomJid,
|
||||
timestamp,
|
||||
binding.name,
|
||||
'web-dashboard',
|
||||
true,
|
||||
);
|
||||
writeMessage({
|
||||
id,
|
||||
chat_jid: messageRoomJid,
|
||||
sender: 'web-dashboard',
|
||||
sender_name: 'Web Dashboard',
|
||||
content: body.text,
|
||||
timestamp,
|
||||
is_from_me: false,
|
||||
is_bot_message: false,
|
||||
message_source_kind: 'ipc_injected_human',
|
||||
});
|
||||
enqueueMessageCheck(messageRoomJid, binding.folder);
|
||||
|
||||
return jsonResponse({ ok: true, id, queued: true });
|
||||
}
|
||||
|
||||
if (request.method !== 'GET' && request.method !== 'HEAD') {
|
||||
return jsonResponse({ error: 'Method not allowed' }, { status: 405 });
|
||||
}
|
||||
@@ -242,6 +385,8 @@ export function startWebDashboardServer(
|
||||
host?: string;
|
||||
port?: number;
|
||||
staticDir?: string;
|
||||
getRoomBindings?: () => Record<string, RegisteredGroup>;
|
||||
enqueueMessageCheck?: (chatJid: string, groupFolder: string) => void;
|
||||
} = {},
|
||||
): StartedWebDashboardServer | null {
|
||||
const enabled = opts.enabled ?? WEB_DASHBOARD.enabled;
|
||||
@@ -253,7 +398,11 @@ export function startWebDashboardServer(
|
||||
const server = Bun.serve({
|
||||
hostname: host,
|
||||
port,
|
||||
fetch: createWebDashboardHandler({ staticDir }),
|
||||
fetch: createWebDashboardHandler({
|
||||
staticDir,
|
||||
getRoomBindings: opts.getRoomBindings,
|
||||
enqueueMessageCheck: opts.enqueueMessageCheck,
|
||||
}),
|
||||
});
|
||||
const url = `http://${host}:${server.port}`;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user