import express from "express";
import { createHash, createHmac, timingSafeEqual } from "node:crypto";
import { mkdirSync } from "node:fs";
import { join } from "node:path";
import { DatabaseSync } from "node:sqlite";
import Ajv from "ajv";
import addFormats from "ajv-formats";
const host = process.env.HOST?.trim() || "127.0.0.1";
const port = Number(process.env.PORT || "8787");
const signingSecret = process.env.SUPPIO_ACTION_SECRET?.trim();
const dataDirectory = process.env.DATA_DIR?.trim() || "./data";
const allowedDiscordUserIds = new Set(
(process.env.ALLOWED_DISCORD_USER_IDS || "")
.split(",")
.map((value) => value.trim())
.filter(Boolean)
);
if (!signingSecret) throw new Error("SUPPIO_ACTION_SECRET is required");
if (!Number.isInteger(port) || port < 1 || port > 65535) {
throw new Error("PORT must be a valid TCP port");
}
mkdirSync(dataDirectory, { recursive: true, mode: 0o700 });
const database = new DatabaseSync(join(dataDirectory, "action-results.sqlite"));
database.exec(`
PRAGMA journal_mode = WAL;
PRAGMA synchronous = FULL;
CREATE TABLE IF NOT EXISTS action_results (
action_id TEXT PRIMARY KEY,
request_hash TEXT NOT NULL,
state TEXT NOT NULL CHECK (state IN ('executing', 'completed')),
response_json TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS ticket_close_events (
event_id TEXT PRIMARY KEY,
request_hash TEXT NOT NULL,
payload_json TEXT NOT NULL,
status TEXT NOT NULL CHECK (status IN ('pending', 'completed')),
created_at TEXT NOT NULL,
completed_at TEXT
);
`);
const findResult = database.prepare(`
SELECT request_hash, state, response_json
FROM action_results
WHERE action_id = ?
`);
const claimResult = database.prepare(`
INSERT OR IGNORE INTO action_results (
action_id, request_hash, state, response_json, created_at, updated_at
) VALUES (?, ?, 'executing', NULL, ?, ?)
`);
const completeResult = database.prepare(`
UPDATE action_results
SET state = 'completed', response_json = ?, updated_at = ?
WHERE action_id = ? AND state = 'executing'
`);
const findTicketCloseEvent = database.prepare(`
SELECT request_hash, status
FROM ticket_close_events
WHERE event_id = ?
`);
const queueTicketCloseEvent = database.prepare(`
INSERT OR IGNORE INTO ticket_close_events (
event_id, request_hash, payload_json, status, created_at
) VALUES (?, ?, ?, 'pending', ?)
`);
const ajv = new Ajv({ allErrors: true, strict: true });
addFormats(ajv);
const changeCustomerPlanSchema = {
type: "object",
properties: {
customer_id: {
type: "string",
minLength: 1
},
plan: {
type: "string",
enum: ["starter", "plus", "pro"]
},
effective_at: {
type: ["string", "null"],
format: "date-time"
}
},
required: ["customer_id", "plan", "effective_at"],
additionalProperties: false
};
const handlers = new Map([
[
"change_customer_plan",
{
validate: ajv.compile(changeCustomerPlanSchema),
async authorize(event, arguments_) {
// Replace this global allowlist with your own lookup that proves this
// Discord user may change arguments_.customer_id.
return (
allowedDiscordUserIds.has(event.actor.user_id) &&
typeof arguments_.customer_id === "string"
);
},
async execute(_event, _arguments) {
// Replace this safe failure with one explicit application transaction.
// Return succeeded only after your transaction commits.
return {
status: "failed",
summary: "The plan-change handler is not connected yet.",
code: "handler_not_configured",
retryable: false
};
}
}
]
]);
const isObject = (value) =>
Boolean(value) && typeof value === "object" && !Array.isArray(value);
function validCustomerActionEnvelope(event) {
return (
isObject(event) &&
event.version === "1" &&
["customer_action.execute", "customer_action.test"].includes(event.event) &&
typeof event.action_id === "string" &&
/^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(
event.action_id
) &&
isObject(event.action) &&
typeof event.action.name === "string" &&
isObject(event.action.arguments) &&
isObject(event.actor) &&
event.actor.provider === "discord" &&
typeof event.actor.user_id === "string" &&
isObject(event.conversation) &&
["ticket", "forum"].includes(event.conversation.surface) &&
["guild_id", "channel_id", "message_id"].every(
(key) => typeof event.conversation[key] === "string"
) &&
isObject(event.agent) &&
typeof event.agent.id === "string" &&
typeof event.agent.workspace_id === "string"
);
}
function validTicketResolvedEnvelope(event) {
return (
isObject(event) &&
event.version === "1" &&
event.event === "ticket.resolved" &&
typeof event.event_id === "string" &&
/^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(
event.event_id
) &&
typeof event.occurred_at === "string" &&
Number.isFinite(Date.parse(event.occurred_at)) &&
["agent", "manual", "inactivity"].includes(event.source) &&
(event.reason === null || typeof event.reason === "string") &&
isObject(event.actor) &&
event.actor.provider === "discord" &&
(event.actor.user_id === null ||
typeof event.actor.user_id === "string") &&
isObject(event.conversation) &&
event.conversation.surface === "ticket" &&
["guild_id", "channel_id", "message_id"].every(
(key) => typeof event.conversation[key] === "string"
) &&
isObject(event.agent) &&
typeof event.agent.id === "string" &&
typeof event.agent.workspace_id === "string"
);
}
function validSignature(rawBody, timestamp, suppliedSignature) {
const timestampSeconds = Number(timestamp);
if (
!Number.isInteger(timestampSeconds) ||
Math.abs(Date.now() / 1000 - timestampSeconds) > 300
) {
return false;
}
const expectedSignature = `v1=${createHmac("sha256", signingSecret)
.update(timestamp, "utf8")
.update(".", "utf8")
.update(rawBody)
.digest("hex")}`;
const expected = Buffer.from(expectedSignature, "utf8");
const supplied = Buffer.from(suppliedSignature, "utf8");
return (
expected.length === supplied.length &&
timingSafeEqual(expected, supplied)
);
}
function protocolError(response, status, message) {
response.status(status).json({ error: message });
}
function knownFailure(summary, code) {
return {
status: "failed",
summary,
code,
retryable: false
};
}
function validExecutorResult(value) {
return (
isObject(value) &&
["succeeded", "failed"].includes(value.status) &&
typeof value.summary === "string" &&
value.summary.trim().length > 0 &&
value.summary.length <= 2000
);
}
function replayOrClaim(actionId, requestHash) {
const existing = findResult.get(actionId);
if (existing) {
if (existing.request_hash !== requestHash) {
return { kind: "conflict" };
}
if (existing.state === "completed" && existing.response_json) {
return {
kind: "replay",
result: JSON.parse(existing.response_json)
};
}
return {
kind: "replay",
result: knownFailure(
"A previous delivery may still be running or may have completed without recording its result.",
"execution_unconfirmed"
)
};
}
const now = new Date().toISOString();
const claimed = claimResult.run(actionId, requestHash, now, now);
if (claimed.changes === 1) return { kind: "claimed" };
// Another process won the INSERT OR IGNORE race. Read its durable row.
return replayOrClaim(actionId, requestHash);
}
function queueTicketClose(event, requestHash) {
const existing = findTicketCloseEvent.get(event.event_id);
if (existing) {
return existing.request_hash === requestHash
? { kind: "duplicate" }
: { kind: "conflict" };
}
const queued = queueTicketCloseEvent.run(
event.event_id,
requestHash,
JSON.stringify(event),
new Date().toISOString()
);
if (queued.changes === 1) return { kind: "queued" };
return queueTicketClose(event, requestHash);
}
const app = express();
app.get("/health", (_request, response) => {
response.json({ ok: true });
});
app.post(
"/suppio/actions",
express.raw({ type: "application/json", limit: "64kb" }),
async (request, response) => {
const rawBody = request.body;
if (!Buffer.isBuffer(rawBody)) {
protocolError(response, 415, "Content-Type must be application/json");
return;
}
const timestamp = request.get("X-Suppio-Timestamp") || "";
const signature = request.get("X-Suppio-Signature") || "";
if (!validSignature(rawBody, timestamp, signature)) {
protocolError(response, 401, "Invalid or expired signature");
return;
}
let event;
try {
event = JSON.parse(rawBody.toString("utf8"));
} catch {
protocolError(response, 400, "Invalid JSON");
return;
}
if (event?.event === "ticket.resolved") {
if (!validTicketResolvedEnvelope(event)) {
protocolError(response, 400, "Invalid ticket resolution event");
return;
}
const headerEventId = request.get("X-Suppio-Event-Id") || "";
if (headerEventId !== event.event_id) {
protocolError(response, 400, "Event ID mismatch");
return;
}
const requestHash = createHash("sha256").update(rawBody).digest("hex");
const queued = queueTicketClose(event, requestHash);
if (queued.kind === "conflict") {
protocolError(
response,
409,
"Event ID was reused with a different request"
);
return;
}
if (queued.kind === "queued") {
console.info("Ticket close work queued", {
eventId: event.event_id,
source: event.source,
guildId: event.conversation.guild_id,
channelId: event.conversation.channel_id
});
}
response.status(204).end();
return;
}
if (!validCustomerActionEnvelope(event)) {
protocolError(response, 400, "Invalid customer action event");
return;
}
const headerActionId = request.get("X-Suppio-Action-Id") || "";
if (headerActionId !== event.action_id) {
protocolError(response, 400, "Action ID mismatch");
return;
}
if (event.event === "customer_action.test") {
response.status(200).json({
status: "succeeded",
summary: "Connection verified."
});
return;
}
const handler = handlers.get(event.action.name);
if (!handler) {
response
.status(200)
.json(knownFailure("This action is not supported.", "unknown_action"));
return;
}
if (!handler.validate(event.action.arguments)) {
response.status(200).json(
knownFailure(
"The action arguments did not match the required schema.",
"invalid_arguments"
)
);
return;
}
if (!(await handler.authorize(event, event.action.arguments))) {
response
.status(200)
.json(knownFailure("You are not allowed to perform this action.", "forbidden"));
return;
}
const requestHash = createHash("sha256").update(rawBody).digest("hex");
const claim = replayOrClaim(event.action_id, requestHash);
if (claim.kind === "conflict") {
protocolError(response, 409, "Action ID was reused with a different request");
return;
}
if (claim.kind === "replay") {
response.status(200).json(claim.result);
return;
}
let result;
let serializedResult;
try {
result = await handler.execute(event, event.action.arguments);
if (!validExecutorResult(result)) {
throw new Error("Handler returned an invalid result");
}
serializedResult = JSON.stringify(result);
if (Buffer.byteLength(serializedResult, "utf8") > 64 * 1024) {
throw new Error("Handler result exceeded 64 KB");
}
} catch (error) {
console.error("Customer action handler failed", {
actionId: event.action_id,
actionName: event.action.name,
error
});
result = knownFailure(
"The requested action could not be completed.",
"action_failed"
);
serializedResult = JSON.stringify(result);
}
completeResult.run(
serializedResult,
new Date().toISOString(),
event.action_id
);
response.status(200).type("application/json").send(serializedResult);
}
);
app.use((error, _request, response, _next) => {
if (error?.type === "entity.too.large") {
protocolError(response, 413, "Request body is too large");
return;
}
console.error("Executor request failed", error);
protocolError(response, 500, "Executor request failed");
});
const server = app.listen(port, host, () => {
console.log(`Customer action executor listening on http://${host}:${port}`);
});
server.on("error", (error) => {
console.error("Customer action executor failed to start", error);
process.exitCode = 1;
});