refactor: new design — sidecar services, moderator Gateway client, tool execute API
- Replace standalone no-reply-api Docker service with unified sidecar (services/main.mjs) that routes /no-reply/* and /moderator/* and starts/stops with openclaw-gateway - Add moderator Discord Gateway client (services/moderator/index.mjs) for real-time MESSAGE_CREATE push instead of polling; notifies plugin via HTTP callback - Add plugin HTTP routes (plugin/web/dirigent-api.ts) for moderator → plugin callbacks (wake-from-dormant, interrupt tail-match) - Fix tool registration format: AgentTool requires execute: not handler:; factory form for tools needing ctx - Rename no-reply-process.ts → sidecar-process.ts, startNoReplyApi → startSideCar - Remove dead config fields from openclaw.plugin.json (humanList, agentList, listMode, channelPoliciesFile, endSymbols, waitIdentifier, multiMessage*, bypassUserIds, etc.) - Rename noReplyPort → sideCarPort - Remove docker-compose.yml, dev-up/down scripts, package-plugin.mjs, test-no-reply-api.mjs - Update install.mjs: clean dist before build, copy services/, drop dead config writes - Update README, Makefile, smoke script for new architecture Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
514
services/moderator/index.mjs
Normal file
514
services/moderator/index.mjs
Normal file
@@ -0,0 +1,514 @@
|
||||
/**
|
||||
* Moderator bot service.
|
||||
*
|
||||
* Exports createModeratorService(config) returning { httpHandler(req, res), stop() }.
|
||||
*
|
||||
* Responsibilities:
|
||||
* - Discord Gateway WS with intents GUILD_MESSAGES (512) | MESSAGE_CONTENT (32768)
|
||||
* - On MESSAGE_CREATE dispatch: notify plugin API
|
||||
* - HTTP sub-handler for /health, /me, /send, /delete-message, /create-channel, /guilds, /channels/:guildId
|
||||
*/
|
||||
|
||||
import { URL as NodeURL } from "node:url";
|
||||
|
||||
const DISCORD_API = "https://discord.com/api/v10";
|
||||
const GATEWAY_URL = "wss://gateway.discord.gg/?v=10&encoding=json";
|
||||
const MAX_RECONNECT_DELAY_MS = 60_000;
|
||||
const INTENTS = 512 | 32768; // GUILD_MESSAGES | MESSAGE_CONTENT
|
||||
|
||||
// ── Helpers ────────────────────────────────────────────────────────────────────
|
||||
|
||||
function sendJson(res, status, payload) {
|
||||
res.writeHead(status, { "Content-Type": "application/json; charset=utf-8" });
|
||||
res.end(JSON.stringify(payload));
|
||||
}
|
||||
|
||||
function readBody(req) {
|
||||
return new Promise((resolve, reject) => {
|
||||
let body = "";
|
||||
req.on("data", (chunk) => {
|
||||
body += chunk;
|
||||
if (body.length > 1_000_000) {
|
||||
req.destroy();
|
||||
reject(new Error("body too large"));
|
||||
}
|
||||
});
|
||||
req.on("end", () => {
|
||||
try {
|
||||
resolve(body ? JSON.parse(body) : {});
|
||||
} catch {
|
||||
reject(new Error("invalid_json"));
|
||||
}
|
||||
});
|
||||
req.on("error", reject);
|
||||
});
|
||||
}
|
||||
|
||||
function getBotUserIdFromToken(token) {
|
||||
try {
|
||||
const segment = token.split(".")[0];
|
||||
const padded = segment + "=".repeat((4 - (segment.length % 4)) % 4);
|
||||
return Buffer.from(padded, "base64").toString("utf8");
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
// ── Discord REST helpers ───────────────────────────────────────────────────────
|
||||
|
||||
async function discordGet(token, path) {
|
||||
const r = await fetch(`${DISCORD_API}${path}`, {
|
||||
headers: { Authorization: `Bot ${token}` },
|
||||
});
|
||||
if (!r.ok) {
|
||||
const text = await r.text().catch(() => "");
|
||||
throw new Error(`Discord GET ${path} failed (${r.status}): ${text}`);
|
||||
}
|
||||
return r.json();
|
||||
}
|
||||
|
||||
async function discordPost(token, path, body) {
|
||||
const r = await fetch(`${DISCORD_API}${path}`, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
Authorization: `Bot ${token}`,
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
body: JSON.stringify(body),
|
||||
});
|
||||
return { ok: r.ok, status: r.status, data: await r.json().catch(() => null) };
|
||||
}
|
||||
|
||||
async function discordDelete(token, path) {
|
||||
const r = await fetch(`${DISCORD_API}${path}`, {
|
||||
method: "DELETE",
|
||||
headers: { Authorization: `Bot ${token}` },
|
||||
});
|
||||
return { ok: r.ok, status: r.status };
|
||||
}
|
||||
|
||||
// ── Gateway connection ─────────────────────────────────────────────────────────
|
||||
|
||||
function createGatewayConnection(token, onMessage, log) {
|
||||
let ws = null;
|
||||
let heartbeatTimer = null;
|
||||
let heartbeatAcked = true;
|
||||
let lastSequence = null;
|
||||
let sessionId = null;
|
||||
let resumeUrl = null;
|
||||
let reconnectTimer = null;
|
||||
let reconnectAttempts = 0;
|
||||
let destroyed = false;
|
||||
|
||||
function sendPayload(data) {
|
||||
if (ws?.readyState === 1 /* OPEN */) {
|
||||
ws.send(JSON.stringify(data));
|
||||
}
|
||||
}
|
||||
|
||||
function stopHeartbeat() {
|
||||
if (heartbeatTimer) {
|
||||
clearInterval(heartbeatTimer);
|
||||
clearTimeout(heartbeatTimer);
|
||||
heartbeatTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
function startHeartbeat(intervalMs) {
|
||||
stopHeartbeat();
|
||||
heartbeatAcked = true;
|
||||
|
||||
const jitter = Math.floor(Math.random() * intervalMs);
|
||||
const firstTimer = setTimeout(() => {
|
||||
if (destroyed) return;
|
||||
if (!heartbeatAcked) {
|
||||
ws?.close(4000, "missed heartbeat ack");
|
||||
return;
|
||||
}
|
||||
heartbeatAcked = false;
|
||||
sendPayload({ op: 1, d: lastSequence });
|
||||
|
||||
heartbeatTimer = setInterval(() => {
|
||||
if (destroyed) return;
|
||||
if (!heartbeatAcked) {
|
||||
ws?.close(4000, "missed heartbeat ack");
|
||||
return;
|
||||
}
|
||||
heartbeatAcked = false;
|
||||
sendPayload({ op: 1, d: lastSequence });
|
||||
}, intervalMs);
|
||||
}, jitter);
|
||||
|
||||
heartbeatTimer = firstTimer;
|
||||
}
|
||||
|
||||
function cleanup() {
|
||||
stopHeartbeat();
|
||||
if (ws) {
|
||||
ws.onopen = null;
|
||||
ws.onmessage = null;
|
||||
ws.onclose = null;
|
||||
ws.onerror = null;
|
||||
try { ws.close(1000); } catch { /* ignore */ }
|
||||
ws = null;
|
||||
}
|
||||
}
|
||||
|
||||
function scheduleReconnect(resume) {
|
||||
if (destroyed) return;
|
||||
if (reconnectTimer) clearTimeout(reconnectTimer);
|
||||
|
||||
reconnectAttempts++;
|
||||
const baseDelay = Math.min(1000 * Math.pow(2, reconnectAttempts), MAX_RECONNECT_DELAY_MS);
|
||||
const delay = baseDelay + Math.random() * 1000;
|
||||
|
||||
log.info(`dirigent-moderator: reconnect in ${Math.round(delay)}ms (attempt ${reconnectAttempts})`);
|
||||
|
||||
reconnectTimer = setTimeout(() => {
|
||||
reconnectTimer = null;
|
||||
connect(resume);
|
||||
}, delay);
|
||||
}
|
||||
|
||||
function connect(isResume = false) {
|
||||
if (destroyed) return;
|
||||
|
||||
cleanup();
|
||||
|
||||
const url = isResume && resumeUrl ? resumeUrl : GATEWAY_URL;
|
||||
|
||||
try {
|
||||
ws = new WebSocket(url);
|
||||
} catch (err) {
|
||||
log.warn(`dirigent-moderator: ws constructor failed: ${String(err)}`);
|
||||
scheduleReconnect(false);
|
||||
return;
|
||||
}
|
||||
|
||||
const currentWs = ws;
|
||||
|
||||
ws.onopen = () => {
|
||||
if (currentWs !== ws || destroyed) return;
|
||||
reconnectAttempts = 0;
|
||||
|
||||
if (isResume && sessionId) {
|
||||
sendPayload({
|
||||
op: 6,
|
||||
d: { token, session_id: sessionId, seq: lastSequence },
|
||||
});
|
||||
} else {
|
||||
sendPayload({
|
||||
op: 2,
|
||||
d: {
|
||||
token,
|
||||
intents: INTENTS,
|
||||
properties: {
|
||||
os: "linux",
|
||||
browser: "dirigent",
|
||||
device: "dirigent",
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
ws.onmessage = (evt) => {
|
||||
if (currentWs !== ws || destroyed) return;
|
||||
|
||||
try {
|
||||
const msg = JSON.parse(typeof evt.data === "string" ? evt.data : String(evt.data));
|
||||
const { op, t, s, d } = msg;
|
||||
|
||||
if (s != null) lastSequence = s;
|
||||
|
||||
switch (op) {
|
||||
case 10: // Hello
|
||||
startHeartbeat(d.heartbeat_interval);
|
||||
break;
|
||||
case 11: // Heartbeat ACK
|
||||
heartbeatAcked = true;
|
||||
break;
|
||||
case 1: // Heartbeat request
|
||||
sendPayload({ op: 1, d: lastSequence });
|
||||
break;
|
||||
case 0: // Dispatch
|
||||
if (t === "READY") {
|
||||
sessionId = d.session_id;
|
||||
resumeUrl = d.resume_gateway_url;
|
||||
log.info("dirigent-moderator: connected and ready");
|
||||
} else if (t === "RESUMED") {
|
||||
log.info("dirigent-moderator: session resumed");
|
||||
} else if (t === "MESSAGE_CREATE") {
|
||||
onMessage(d);
|
||||
}
|
||||
break;
|
||||
case 7: // Reconnect
|
||||
log.info("dirigent-moderator: reconnect requested by Discord");
|
||||
cleanup();
|
||||
scheduleReconnect(true);
|
||||
break;
|
||||
case 9: // Invalid Session
|
||||
log.warn(`dirigent-moderator: invalid session, resumable=${d}`);
|
||||
cleanup();
|
||||
sessionId = d ? sessionId : null;
|
||||
setTimeout(() => {
|
||||
if (!destroyed) connect(!!d && !!sessionId);
|
||||
}, 3000 + Math.random() * 2000);
|
||||
break;
|
||||
}
|
||||
} catch {
|
||||
// ignore parse errors
|
||||
}
|
||||
};
|
||||
|
||||
ws.onclose = (evt) => {
|
||||
if (currentWs !== ws) return;
|
||||
stopHeartbeat();
|
||||
if (destroyed) return;
|
||||
|
||||
const code = evt.code;
|
||||
|
||||
if (code === 4004) {
|
||||
log.warn("dirigent-moderator: token invalid (4004), stopping");
|
||||
return;
|
||||
}
|
||||
if (code === 4010 || code === 4011 || code === 4013 || code === 4014) {
|
||||
log.warn(`dirigent-moderator: fatal close (${code}), re-identifying`);
|
||||
sessionId = null;
|
||||
scheduleReconnect(false);
|
||||
return;
|
||||
}
|
||||
|
||||
log.info(`dirigent-moderator: disconnected (code=${code}), will reconnect`);
|
||||
const canResume = !!sessionId && code !== 4012;
|
||||
scheduleReconnect(canResume);
|
||||
};
|
||||
|
||||
ws.onerror = () => {
|
||||
// onclose will fire after this
|
||||
};
|
||||
}
|
||||
|
||||
// Start initial connection
|
||||
connect(false);
|
||||
|
||||
return {
|
||||
stop() {
|
||||
destroyed = true;
|
||||
if (reconnectTimer) {
|
||||
clearTimeout(reconnectTimer);
|
||||
reconnectTimer = null;
|
||||
}
|
||||
cleanup();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// ── HTTP route handler ─────────────────────────────────────────────────────────
|
||||
|
||||
function createHttpHandler(token, botUserId, log) {
|
||||
return async function httpHandler(req, res) {
|
||||
const url = req.url ?? "/";
|
||||
|
||||
// GET /health
|
||||
if (req.method === "GET" && url === "/health") {
|
||||
return sendJson(res, 200, { ok: true, botId: botUserId });
|
||||
}
|
||||
|
||||
// GET /me
|
||||
if (req.method === "GET" && url === "/me") {
|
||||
try {
|
||||
const data = await discordGet(token, "/users/@me");
|
||||
return sendJson(res, 200, { id: data.id, username: data.username });
|
||||
} catch (err) {
|
||||
return sendJson(res, 500, { ok: false, error: String(err) });
|
||||
}
|
||||
}
|
||||
|
||||
// GET /guilds
|
||||
if (req.method === "GET" && url === "/guilds") {
|
||||
try {
|
||||
const guilds = await discordGet(token, "/users/@me/guilds");
|
||||
const ADMIN = 8n;
|
||||
const adminGuilds = guilds
|
||||
.filter((g) => (BigInt(g.permissions ?? "0") & ADMIN) === ADMIN)
|
||||
.map((g) => ({ id: g.id, name: g.name }));
|
||||
return sendJson(res, 200, { guilds: adminGuilds });
|
||||
} catch (err) {
|
||||
return sendJson(res, 500, { ok: false, error: String(err) });
|
||||
}
|
||||
}
|
||||
|
||||
// GET /channels/:guildId
|
||||
const channelsMatch = url.match(/^\/channels\/(\d+)$/);
|
||||
if (req.method === "GET" && channelsMatch) {
|
||||
const guildId = channelsMatch[1];
|
||||
try {
|
||||
const channels = await discordGet(token, `/guilds/${guildId}/channels`);
|
||||
return sendJson(res, 200, {
|
||||
channels: channels
|
||||
.filter((c) => c.type === 0)
|
||||
.map((c) => ({ id: c.id, name: c.name, type: c.type })),
|
||||
});
|
||||
} catch (err) {
|
||||
return sendJson(res, 500, { ok: false, error: String(err) });
|
||||
}
|
||||
}
|
||||
|
||||
// POST /send
|
||||
if (req.method === "POST" && url === "/send") {
|
||||
let body;
|
||||
try {
|
||||
body = await readBody(req);
|
||||
} catch (err) {
|
||||
return sendJson(res, 400, { ok: false, error: String(err) });
|
||||
}
|
||||
const { channelId, content } = body;
|
||||
if (!channelId || !content) {
|
||||
return sendJson(res, 400, { ok: false, error: "channelId and content required" });
|
||||
}
|
||||
try {
|
||||
const result = await discordPost(token, `/channels/${channelId}/messages`, { content });
|
||||
if (!result.ok) {
|
||||
return sendJson(res, result.status, { ok: false, error: `Discord API error ${result.status}` });
|
||||
}
|
||||
return sendJson(res, 200, { ok: true, messageId: result.data?.id });
|
||||
} catch (err) {
|
||||
return sendJson(res, 500, { ok: false, error: String(err) });
|
||||
}
|
||||
}
|
||||
|
||||
// POST /delete-message
|
||||
if (req.method === "POST" && url === "/delete-message") {
|
||||
let body;
|
||||
try {
|
||||
body = await readBody(req);
|
||||
} catch (err) {
|
||||
return sendJson(res, 400, { ok: false, error: String(err) });
|
||||
}
|
||||
const { channelId, messageId } = body;
|
||||
if (!channelId || !messageId) {
|
||||
return sendJson(res, 400, { ok: false, error: "channelId and messageId required" });
|
||||
}
|
||||
try {
|
||||
const result = await discordDelete(token, `/channels/${channelId}/messages/${messageId}`);
|
||||
return sendJson(res, 200, { ok: result.ok });
|
||||
} catch (err) {
|
||||
return sendJson(res, 500, { ok: false, error: String(err) });
|
||||
}
|
||||
}
|
||||
|
||||
// POST /create-channel
|
||||
if (req.method === "POST" && url === "/create-channel") {
|
||||
let body;
|
||||
try {
|
||||
body = await readBody(req);
|
||||
} catch (err) {
|
||||
return sendJson(res, 400, { ok: false, error: String(err) });
|
||||
}
|
||||
const { guildId, name, permissionOverwrites = [] } = body;
|
||||
if (!guildId || !name) {
|
||||
return sendJson(res, 400, { ok: false, error: "guildId and name required" });
|
||||
}
|
||||
try {
|
||||
const result = await discordPost(token, `/guilds/${guildId}/channels`, {
|
||||
name,
|
||||
type: 0,
|
||||
permission_overwrites: permissionOverwrites,
|
||||
});
|
||||
if (!result.ok) {
|
||||
return sendJson(res, result.status, { ok: false, error: `Discord API error ${result.status}` });
|
||||
}
|
||||
return sendJson(res, 200, { ok: true, channelId: result.data?.id });
|
||||
} catch (err) {
|
||||
return sendJson(res, 500, { ok: false, error: String(err) });
|
||||
}
|
||||
}
|
||||
|
||||
return sendJson(res, 404, { error: "not_found" });
|
||||
};
|
||||
}
|
||||
|
||||
// ── Plugin notification ────────────────────────────────────────────────────────
|
||||
|
||||
function createNotifyPlugin(pluginApiUrl, pluginApiToken, log) {
|
||||
return function notifyPlugin(message) {
|
||||
const body = JSON.stringify({
|
||||
channelId: message.channel_id,
|
||||
messageId: message.id,
|
||||
senderId: message.author?.id,
|
||||
guildId: message.guild_id,
|
||||
content: message.content,
|
||||
});
|
||||
|
||||
const headers = {
|
||||
"Content-Type": "application/json",
|
||||
};
|
||||
if (pluginApiToken) {
|
||||
headers["Authorization"] = `Bearer ${pluginApiToken}`;
|
||||
}
|
||||
|
||||
fetch(`${pluginApiUrl}/dirigent/api/moderator/message`, {
|
||||
method: "POST",
|
||||
headers,
|
||||
body,
|
||||
}).catch((err) => {
|
||||
log.warn(`dirigent-moderator: notify plugin failed: ${String(err)}`);
|
||||
});
|
||||
};
|
||||
}
|
||||
|
||||
// ── Public API ─────────────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Create the moderator service.
|
||||
*
|
||||
* @param {object} config
|
||||
* @param {string} config.token - Discord bot token
|
||||
* @param {string} config.pluginApiUrl - e.g. "http://127.0.0.1:18789"
|
||||
* @param {string} [config.pluginApiToken] - bearer token for plugin API
|
||||
* @param {string} [config.scheduleIdentifier] - e.g. "➡️"
|
||||
* @param {boolean} [config.debugMode]
|
||||
* @returns {{ httpHandler: Function, stop: Function }}
|
||||
*/
|
||||
export function createModeratorService(config) {
|
||||
const { token, pluginApiUrl, pluginApiToken = "", scheduleIdentifier = "➡️", debugMode = false } = config;
|
||||
|
||||
const log = {
|
||||
info: (msg) => console.log(`[dirigent-moderator] ${msg}`),
|
||||
warn: (msg) => console.warn(`[dirigent-moderator] WARN ${msg}`),
|
||||
};
|
||||
|
||||
if (debugMode) {
|
||||
log.info(`debug mode enabled, scheduleIdentifier=${scheduleIdentifier}`);
|
||||
}
|
||||
|
||||
// Decode bot user ID from token
|
||||
const botUserId = getBotUserIdFromToken(token);
|
||||
log.info(`bot user id decoded: ${botUserId ?? "(unknown)"}`);
|
||||
|
||||
// Plugin notify callback (fire-and-forget)
|
||||
const notifyPlugin = createNotifyPlugin(pluginApiUrl, pluginApiToken, log);
|
||||
|
||||
// Gateway connection
|
||||
const gateway = createGatewayConnection(
|
||||
token,
|
||||
(message) => {
|
||||
// Skip bot's own messages
|
||||
if (message.author?.id === botUserId) return;
|
||||
notifyPlugin(message);
|
||||
},
|
||||
log,
|
||||
);
|
||||
|
||||
// HTTP handler (caller strips /moderator prefix)
|
||||
const httpHandler = createHttpHandler(token, botUserId, log);
|
||||
|
||||
return {
|
||||
httpHandler,
|
||||
stop() {
|
||||
log.info("stopping moderator service");
|
||||
gateway.stop();
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user