feat: split web and api into separate apps
This commit is contained in:
M1
@@ -1,52 +0,0 @@
|
||||
/// Internal endpoints used by the Rust monitor runner.
|
||||
/// Protected by MONITOR_TOKEN — not exposed to users.
|
||||
|
||||
import { Elysia } from "elysia";
|
||||
import sql from "../db";
|
||||
|
||||
export async function pruneOldPings(retentionDays = 90) {
|
||||
const result = await sql`DELETE FROM pings WHERE checked_at < now() - ${retentionDays + ' days'}::interval`;
|
||||
return result.count;
|
||||
}
|
||||
|
||||
// Run retention cleanup every hour
|
||||
setInterval(() => {
|
||||
const days = Number(process.env.PING_RETENTION_DAYS ?? 90);
|
||||
pruneOldPings(days).catch((err) => console.error("Retention cleanup failed:", err));
|
||||
}, 60 * 60 * 1000);
|
||||
|
||||
export const internal = new Elysia({ prefix: "/internal", detail: { hide: true } })
|
||||
.derive(({ headers, error }) => {
|
||||
if (headers["x-monitor-token"] !== process.env.MONITOR_TOKEN)
|
||||
return error(401, { error: "Unauthorized" });
|
||||
return {};
|
||||
})
|
||||
|
||||
// Returns monitors that are due for a check.
|
||||
// scheduled_at = last_checked_at + interval_s (ideal fire time), so jitter = actual_start - scheduled_at
|
||||
.get("/due", async () => {
|
||||
const monitors = await sql`
|
||||
SELECT m.id, m.url, m.method, m.request_headers, m.request_body, m.timeout_ms, m.interval_s, m.query,
|
||||
CASE
|
||||
WHEN last.checked_at IS NULL THEN now()
|
||||
ELSE last.checked_at + (m.interval_s || ' seconds')::interval
|
||||
END AS scheduled_at
|
||||
FROM monitors m
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT checked_at FROM pings
|
||||
WHERE monitor_id = m.id
|
||||
ORDER BY checked_at DESC LIMIT 1
|
||||
) last ON true
|
||||
WHERE m.enabled = true
|
||||
AND (last.checked_at IS NULL
|
||||
OR last.checked_at < now() - (m.interval_s || ' seconds')::interval)
|
||||
`;
|
||||
return monitors;
|
||||
})
|
||||
|
||||
// Manual retention cleanup trigger
|
||||
.post("/prune", async () => {
|
||||
const days = Number(process.env.PING_RETENTION_DAYS ?? 90);
|
||||
const deleted = await pruneOldPings(days);
|
||||
return { deleted, retention_days: days };
|
||||
});
|
||||
@@ -1,118 +0,0 @@
|
||||
import { Elysia, t } from "elysia";
|
||||
import { requireAuth } from "./auth";
|
||||
import sql from "../db";
|
||||
import { validateMonitorUrl } from "../utils/ssrf";
|
||||
|
||||
const MonitorBody = t.Object({
|
||||
name: t.String({ description: "Human-readable name" }),
|
||||
url: t.String({ format: "uri", description: "URL to check" }),
|
||||
method: t.Optional(t.String({ default: "GET", description: "HTTP method: GET, POST, PUT, PATCH, DELETE, HEAD" })),
|
||||
request_headers: t.Optional(t.Any({ description: "Request headers as key-value object" })),
|
||||
request_body: t.Optional(t.Nullable(t.String({ description: "Request body for POST/PUT/PATCH" }))),
|
||||
timeout_ms: t.Optional(t.Number({ minimum: 1000, maximum: 60000, default: 30000, description: "Request timeout in ms" })),
|
||||
interval_s: t.Optional(t.Number({ minimum: 1, default: 60, description: "Check interval in seconds" })),
|
||||
query: t.Optional(t.Any({ description: "PingQL query — filter conditions for up/down" })),
|
||||
});
|
||||
|
||||
export const monitors = new Elysia({ prefix: "/monitors" })
|
||||
.use(requireAuth)
|
||||
|
||||
// List monitors
|
||||
.get("/", async ({ accountId }) => {
|
||||
return sql`SELECT * FROM monitors WHERE account_id = ${accountId} ORDER BY created_at DESC`;
|
||||
}, { detail: { summary: "List monitors", tags: ["monitors"] } })
|
||||
|
||||
// Create monitor
|
||||
.post("/", async ({ accountId, body, error }) => {
|
||||
// SSRF protection
|
||||
const ssrfError = await validateMonitorUrl(body.url);
|
||||
if (ssrfError) return error(400, { error: ssrfError });
|
||||
|
||||
const [monitor] = await sql`
|
||||
INSERT INTO monitors (account_id, name, url, method, request_headers, request_body, timeout_ms, interval_s, query)
|
||||
VALUES (
|
||||
${accountId}, ${body.name}, ${body.url},
|
||||
${(body.method ?? 'GET').toUpperCase()},
|
||||
${body.request_headers ? sql.json(body.request_headers) : null},
|
||||
${body.request_body ?? null},
|
||||
${body.timeout_ms ?? 30000},
|
||||
${body.interval_s ?? 60},
|
||||
${body.query ? sql.json(body.query) : null}
|
||||
)
|
||||
RETURNING *
|
||||
`;
|
||||
return monitor;
|
||||
}, { body: MonitorBody, detail: { summary: "Create monitor", tags: ["monitors"] } })
|
||||
|
||||
// Get monitor + recent status
|
||||
.get("/:id", async ({ accountId, params, error }) => {
|
||||
const [monitor] = await sql`
|
||||
SELECT * FROM monitors WHERE id = ${params.id} AND account_id = ${accountId}
|
||||
`;
|
||||
if (!monitor) return error(404, { error: "Not found" });
|
||||
|
||||
const results = await sql`
|
||||
SELECT * FROM pings WHERE monitor_id = ${params.id}
|
||||
ORDER BY checked_at DESC LIMIT 100
|
||||
`;
|
||||
return { ...monitor, results };
|
||||
}, { detail: { summary: "Get monitor with results", tags: ["monitors"] } })
|
||||
|
||||
// Update monitor
|
||||
.patch("/:id", async ({ accountId, params, body, error }) => {
|
||||
// SSRF protection on URL change
|
||||
if (body.url) {
|
||||
const ssrfError = await validateMonitorUrl(body.url);
|
||||
if (ssrfError) return error(400, { error: ssrfError });
|
||||
}
|
||||
|
||||
const [monitor] = await sql`
|
||||
UPDATE monitors SET
|
||||
name = COALESCE(${body.name ?? null}, name),
|
||||
url = COALESCE(${body.url ?? null}, url),
|
||||
method = COALESCE(${body.method ? body.method.toUpperCase() : null}, method),
|
||||
request_headers = COALESCE(${body.request_headers ? sql.json(body.request_headers) : null}, request_headers),
|
||||
request_body = COALESCE(${body.request_body ?? null}, request_body),
|
||||
timeout_ms = COALESCE(${body.timeout_ms ?? null}, timeout_ms),
|
||||
interval_s = COALESCE(${body.interval_s ?? null}, interval_s),
|
||||
query = COALESCE(${body.query ? sql.json(body.query) : null}, query)
|
||||
WHERE id = ${params.id} AND account_id = ${accountId}
|
||||
RETURNING *
|
||||
`;
|
||||
if (!monitor) return error(404, { error: "Not found" });
|
||||
return monitor;
|
||||
}, { body: t.Partial(MonitorBody), detail: { summary: "Update monitor", tags: ["monitors"] } })
|
||||
|
||||
// Delete monitor
|
||||
.delete("/:id", async ({ accountId, params, error }) => {
|
||||
const [deleted] = await sql`
|
||||
DELETE FROM monitors WHERE id = ${params.id} AND account_id = ${accountId} RETURNING id
|
||||
`;
|
||||
if (!deleted) return error(404, { error: "Not found" });
|
||||
return { deleted: true };
|
||||
}, { detail: { summary: "Delete monitor", tags: ["monitors"] } })
|
||||
|
||||
// Toggle enabled
|
||||
.post("/:id/toggle", async ({ accountId, params, error }) => {
|
||||
const [monitor] = await sql`
|
||||
UPDATE monitors SET enabled = NOT enabled
|
||||
WHERE id = ${params.id} AND account_id = ${accountId}
|
||||
RETURNING id, enabled
|
||||
`;
|
||||
if (!monitor) return error(404, { error: "Not found" });
|
||||
return monitor;
|
||||
}, { detail: { summary: "Toggle monitor on/off", tags: ["monitors"] } })
|
||||
|
||||
// Check history
|
||||
.get("/:id/pings", async ({ accountId, params, query, error }) => {
|
||||
const [monitor] = await sql`
|
||||
SELECT id FROM monitors WHERE id = ${params.id} AND account_id = ${accountId}
|
||||
`;
|
||||
if (!monitor) return error(404, { error: "Not found" });
|
||||
const limit = Math.min(Number(query.limit ?? 100), 1000);
|
||||
return sql`
|
||||
SELECT * FROM pings
|
||||
WHERE monitor_id = ${params.id}
|
||||
ORDER BY checked_at DESC LIMIT ${limit}
|
||||
`;
|
||||
}, { detail: { summary: "Get ping history", tags: ["monitors"] } });
|
||||
@@ -1,114 +0,0 @@
|
||||
import { Elysia, t } from "elysia";
|
||||
import sql from "../db";
|
||||
import { resolveKey } from "./auth";
|
||||
|
||||
// ── SSE bus ───────────────────────────────────────────────────────────────────
|
||||
type SSEController = ReadableStreamDefaultController<Uint8Array>;
|
||||
const bus = new Map<string, Set<SSEController>>(); // keyed by accountId
|
||||
const enc = new TextEncoder();
|
||||
|
||||
function publish(accountId: string, data: object) {
|
||||
const subs = bus.get(accountId);
|
||||
if (!subs?.size) return;
|
||||
const msg = enc.encode(`data: ${JSON.stringify(data)}\n\n`);
|
||||
for (const ctrl of subs) {
|
||||
try { ctrl.enqueue(msg); } catch { subs.delete(ctrl); }
|
||||
}
|
||||
}
|
||||
|
||||
function makeSSEStream(accountId: string): Response {
|
||||
let ctrl: SSEController;
|
||||
let heartbeat: Timer;
|
||||
const stream = new ReadableStream<Uint8Array>({
|
||||
start(c) {
|
||||
ctrl = c;
|
||||
if (!bus.has(accountId)) bus.set(accountId, new Set());
|
||||
bus.get(accountId)!.add(ctrl);
|
||||
ctrl.enqueue(enc.encode(": connected\n\n"));
|
||||
heartbeat = setInterval(() => {
|
||||
try { ctrl.enqueue(enc.encode(": heartbeat\n\n")); } catch { clearInterval(heartbeat); }
|
||||
}, 10_000);
|
||||
},
|
||||
cancel() {
|
||||
clearInterval(heartbeat);
|
||||
bus.get(accountId)?.delete(ctrl);
|
||||
if (bus.get(accountId)?.size === 0) bus.delete(accountId);
|
||||
},
|
||||
});
|
||||
return new Response(stream, {
|
||||
headers: {
|
||||
"Content-Type": "text/event-stream",
|
||||
"Cache-Control": "no-cache",
|
||||
"Connection": "keep-alive",
|
||||
"X-Accel-Buffering": "no",
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
// ── Routes ────────────────────────────────────────────────────────────────────
|
||||
export const ingest = new Elysia()
|
||||
|
||||
// Internal: called by Rust monitor runner
|
||||
.post("/internal/ingest", async ({ body, headers, error }) => {
|
||||
const token = headers["x-monitor-token"];
|
||||
if (token !== process.env.MONITOR_TOKEN) return error(401, { error: "Unauthorized" });
|
||||
|
||||
const meta = body.meta ? { ...body.meta } : {};
|
||||
if (body.cert_expiry_days != null) meta.cert_expiry_days = body.cert_expiry_days;
|
||||
|
||||
const scheduledAt = body.scheduled_at ? new Date(body.scheduled_at) : null;
|
||||
const jitterMs = body.jitter_ms ?? null;
|
||||
|
||||
const [ping] = await sql`
|
||||
INSERT INTO pings (monitor_id, scheduled_at, jitter_ms, status_code, latency_ms, up, error, meta)
|
||||
VALUES (
|
||||
${body.monitor_id},
|
||||
${scheduledAt},
|
||||
${jitterMs},
|
||||
${body.status_code ?? null},
|
||||
${body.latency_ms ?? null},
|
||||
${body.up},
|
||||
${body.error ?? null},
|
||||
${Object.keys(meta).length > 0 ? sql.json(meta) : null}
|
||||
)
|
||||
RETURNING *
|
||||
`;
|
||||
|
||||
// Look up account and publish to account-level bus
|
||||
const [monitor] = await sql`SELECT account_id FROM monitors WHERE id = ${body.monitor_id}`;
|
||||
if (monitor) publish(monitor.account_id, ping);
|
||||
|
||||
return { ok: true };
|
||||
}, {
|
||||
body: t.Object({
|
||||
monitor_id: t.String(),
|
||||
scheduled_at: t.Optional(t.Nullable(t.String())),
|
||||
jitter_ms: t.Optional(t.Nullable(t.Number())),
|
||||
status_code: t.Optional(t.Number()),
|
||||
latency_ms: t.Optional(t.Number()),
|
||||
up: t.Boolean(),
|
||||
error: t.Optional(t.Nullable(t.String())),
|
||||
cert_expiry_days: t.Optional(t.Nullable(t.Number())),
|
||||
meta: t.Optional(t.Any()),
|
||||
}),
|
||||
detail: { hide: true },
|
||||
})
|
||||
|
||||
// SSE: single stream for all of the account's monitors
|
||||
.get("/account/stream", async ({ headers, cookie }) => {
|
||||
const authHeader = headers["authorization"] ?? "";
|
||||
const bearer = authHeader.match(/^bearer\s+(.+)$/i)?.[1]?.trim();
|
||||
const key = bearer ?? cookie?.pingql_key?.value;
|
||||
|
||||
if (!key) return new Response(JSON.stringify({ error: "Unauthorized" }), { status: 401 });
|
||||
|
||||
const resolved = await resolveKey(key);
|
||||
if (!resolved) return new Response(JSON.stringify({ error: "Unauthorized" }), { status: 401 });
|
||||
|
||||
const limit = Number(process.env.MAX_SSE_PER_ACCOUNT ?? 10);
|
||||
if ((bus.get(resolved.accountId)?.size ?? 0) >= limit) {
|
||||
return new Response(JSON.stringify({ error: "Too many connections" }), { status: 429 });
|
||||
}
|
||||
|
||||
return makeSSEStream(resolved.accountId);
|
||||
}, { detail: { hide: true } });
|
||||
Reference in New Issue
Block a user