TypeScript 55.4%
Python 43.2%
SQL 1.2%
1import { createHash } from "node:crypto";2import type { FastifyInstance } from "fastify";3import { z } from "zod";4import { assertUrlAllowed, newId, UrlPolicyError } from "@websensor/core";5import { getConnector } from "@websensor/connectors";6import { db, sql, textArray } from "@websensor/db";7import { config } from "./config";8import { EVENT_SELECT, eventConditions, listEvents } from "./queries";910/**11 * Owner-scoped routes (anonymous owner token, phase 1): watchlists, alert rules (+ webhook12 * channel), notifications, bookmarks, saved views and custom URL monitors (spec §42–45, §81, §100).13 */14export function ownerToken(headers: Record<string, unknown>): string | null {15 const t = headers["x-websensor-owner"];16 return typeof t === "string" && /^[A-Za-z0-9_-]{16,80}$/.test(t) ? t : null;17}1819const WATCH_KINDS = ["entity", "source", "keyword", "category", "url", "event_type", "country", "group"] as const;2021export async function registerUserRoutes(app: FastifyInstance): Promise<void> {22 const requireOwner = (req: { headers: Record<string, unknown> }, reply: { status: (n: number) => { send: (b: unknown) => unknown } }): string | null => {23 const owner = ownerToken(req.headers);24 if (!owner) reply.status(401).send({ error: "owner_token_required", detail: "Send an X-WebSensor-Owner header (16–80 URL-safe characters)." });25 return owner;26 };2728 // ---- Watchlists -----------------------------------------------------------------------------29 app.get("/api/v1/watchlists", async (req, reply) => {30 const owner = requireOwner(req, reply);31 if (!owner) return;32 const rows = await db.execute<Record<string, unknown>>(sql`select w.id, w.name, w.created_at, coalesce((select json_agg(json_build_object('kind', i.kind, 'value', i.value, 'added_at', i.added_at) order by i.added_at) from watchlist_items i where i.watchlist_id = w.id), '[]'::json) as items from watchlists w where owner_token = ${owner} order by created_at`);33 return { items: rows.rows };34 });35 const wlBody = z.object({ name: z.string().min(1).max(80).default("My watchlist"), items: z.array(z.object({ kind: z.enum(WATCH_KINDS), value: z.string().min(1).max(300) })).max(300).default([]) });36 app.post("/api/v1/watchlists", async (req, reply) => {37 const owner = requireOwner(req, reply);38 if (!owner) return;39 const body = wlBody.parse(req.body ?? {});40 const count = (await db.execute<{ n: string }>(sql`select count(*)::text as n from watchlists where owner_token = ${owner}`)).rows[0]?.n;41 if (Number(count) >= 20) return reply.status(429).send({ error: "too_many_watchlists" });42 const id = newId("wl");43 await db.execute(sql`insert into watchlists (id, owner_token, name) values (${id}, ${owner}, ${body.name})`);44 for (const it of body.items) await db.execute(sql`insert into watchlist_items (watchlist_id, kind, value) values (${id}, ${it.kind}, ${it.value}) on conflict do nothing`);45 return { id, name: body.name, items: body.items };46 });47 app.put<{ Params: { id: string } }>("/api/v1/watchlists/:id", async (req, reply) => {48 const owner = requireOwner(req, reply);49 if (!owner) return;50 const w = (await db.execute<{ id: string }>(sql`select id from watchlists where id = ${req.params.id} and owner_token = ${owner}`)).rows[0];51 if (!w) return reply.status(404).send({ error: "not_found" });52 const body = wlBody.partial().parse(req.body ?? {});53 if (body.name) await db.execute(sql`update watchlists set name = ${body.name} where id = ${w.id}`);54 if (body.items) {55 await db.execute(sql`delete from watchlist_items where watchlist_id = ${w.id}`);56 for (const it of body.items) await db.execute(sql`insert into watchlist_items (watchlist_id, kind, value) values (${w.id}, ${it.kind}, ${it.value}) on conflict do nothing`);57 }58 return { ok: true };59 });60 app.delete<{ Params: { id: string } }>("/api/v1/watchlists/:id", async (req, reply) => {61 const owner = requireOwner(req, reply);62 if (!owner) return;63 await db.execute(sql`delete from watchlists where id = ${req.params.id} and owner_token = ${owner}`);64 return { ok: true };65 });66 app.get<{ Params: { id: string }; Querystring: { limit?: string } }>("/api/v1/watchlists/:id/events", async (req, reply) => {67 const owner = requireOwner(req, reply);68 if (!owner) return;69 const items = (await db.execute<{ kind: string; value: string }>(sql`select i.kind, i.value from watchlist_items i join watchlists w on w.id = i.watchlist_id where w.id = ${req.params.id} and w.owner_token = ${owner}`)).rows;70 if (!items.length) return reply.send({ items: [] });71 const by = (k: string): string[] => items.filter((i) => i.kind === k).map((i) => i.value);72 const conds = [] as ReturnType<typeof sql>[];73 const ents = by("entity");74 const srcs = by("source");75 const cats = by("category");76 const types = by("event_type");77 const countries = by("country");78 const urls = by("url");79 if (ents.length) conds.push(sql`exists (select 1 from event_entities x where x.event_id = e.id and x.entity_id = any(${textArray(ents)}))`);80 if (srcs.length) conds.push(sql`e.source_id = any(${textArray(srcs)})`);81 if (cats.length) conds.push(sql`e.categories && ${textArray(cats)}`);82 if (types.length) conds.push(sql`e.event_type = any(${textArray(types)})`);83 if (countries.length) conds.push(sql`e.country = any(${textArray(countries.map((c) => c.toUpperCase()))})`);84 for (const u of urls) conds.push(sql`(e.url = ${u} or e.url like ${u.replace(/\/$/, "") + "/%"})`);85 for (const k of by("keyword")) conds.push(sql`(e.title ilike ${"%" + k + "%"} or e.summary ilike ${"%" + k + "%"})`);86 if (!conds.length) return { items: [] };87 const rows = await db.execute<Record<string, unknown>>(sql`select ${EVENT_SELECT} from events e join sources s on s.id = e.source_id join sensors sen on sen.id = e.sensor_id where (s.kind = 'registry' or s.owner_token = ${owner}) and (${sql.join(conds, sql` or `)}) order by e.detected_at desc limit ${Math.min(200, Number(req.query.limit ?? 50))}`);88 return { items: rows.rows };89 });9091 // ---- Alerts -----------------------------------------------------------------------------------92 const ruleSchema = z.object({93 importance_min: z.number().min(0).max(100).optional(),94 signal_min: z.number().min(0).max(100).optional(),95 event_types: z.array(z.string().max(60)).max(60).optional(),96 groups: z.array(z.string().max(30)).max(12).optional(),97 entities: z.array(z.string().max(120)).max(100).optional(),98 sources: z.array(z.string().max(120)).max(100).optional(),99 keywords: z.array(z.string().max(80)).max(50).optional(),100 categories: z.array(z.string().max(40)).max(40).optional(),101 countries: z.array(z.string().max(3)).max(60).optional(),102 silent_only: z.boolean().optional(),103 first_party_only: z.boolean().optional(),104 confirmed_only: z.boolean().optional(),105 });106 const alertBody = z.object({ name: z.string().min(1).max(80), rule: ruleSchema, channel: z.enum(["web", "webhook"]).default("web"), channel_config: z.object({ url: z.string().url().max(500).optional(), secret: z.string().max(200).optional() }).default({}) });107 app.get("/api/v1/alerts", async (req, reply) => {108 const owner = requireOwner(req, reply);109 if (!owner) return;110 const rows = await db.execute<Record<string, unknown>>(sql`select a.id, a.name, a.rule, a.channel, a.channel_config - 'secret' as channel_config, a.enabled, a.created_at, a.last_fired_at, a.fired_count, (select count(*) from notifications n where n.alert_id = a.id and n.created_at >= now() - interval '24 hours')::int as fired_24h from alerts a where owner_token = ${owner} order by created_at`);111 return { items: rows.rows, channels: { web: "active", webhook: "active", email: "planned", slack: "planned", discord: "planned", telegram: "planned", push: "planned" } };112 });113 app.post("/api/v1/alerts", async (req, reply) => {114 const owner = requireOwner(req, reply);115 if (!owner) return;116 const body = alertBody.parse(req.body ?? {});117 if (body.channel === "webhook") {118 if (!body.channel_config.url) return reply.status(400).send({ error: "webhook_url_required" });119 try {120 const u = await assertUrlAllowed(body.channel_config.url);121 if (u.url.protocol !== "https:") return reply.status(400).send({ error: "webhook_https_required" });122 } catch (e) {123 return reply.status(400).send({ error: "webhook_url_rejected", detail: e instanceof UrlPolicyError ? e.message : String(e) });124 }125 }126 const count = (await db.execute<{ n: string }>(sql`select count(*)::text as n from alerts where owner_token = ${owner}`)).rows[0]?.n;127 if (Number(count) >= 50) return reply.status(429).send({ error: "too_many_alerts" });128 const id = newId("alr");129 await db.execute(sql`insert into alerts (id, owner_token, name, rule, channel, channel_config) values (${id}, ${owner}, ${body.name}, ${JSON.stringify(body.rule)}::jsonb, ${body.channel}, ${JSON.stringify(body.channel_config)}::jsonb)`);130 return { id, name: body.name, rule: body.rule, channel: body.channel };131 });132 app.patch<{ Params: { id: string } }>("/api/v1/alerts/:id", async (req, reply) => {133 const owner = requireOwner(req, reply);134 if (!owner) return;135 const body = z.object({ enabled: z.boolean().optional(), name: z.string().min(1).max(80).optional() }).parse(req.body ?? {});136 if (body.enabled !== undefined) await db.execute(sql`update alerts set enabled = ${body.enabled} where id = ${req.params.id} and owner_token = ${owner}`);137 if (body.name) await db.execute(sql`update alerts set name = ${body.name} where id = ${req.params.id} and owner_token = ${owner}`);138 return { ok: true };139 });140 app.delete<{ Params: { id: string } }>("/api/v1/alerts/:id", async (req, reply) => {141 const owner = requireOwner(req, reply);142 if (!owner) return;143 await db.execute(sql`delete from alerts where id = ${req.params.id} and owner_token = ${owner}`);144 return { ok: true };145 });146 app.get<{ Querystring: { limit?: string; unread?: string } }>("/api/v1/notifications", async (req, reply) => {147 const owner = requireOwner(req, reply);148 if (!owner) return;149 const rows = await db.execute<Record<string, unknown>>(sql`150 select n.id, n.alert_id, a.name as alert_name, n.event_id, n.channel, n.status, n.created_at, n.read_at, n.delivered_at, n.error,151 json_build_object('id', e.id, 'slug', e.slug, 'title', e.title, 'importance', e.importance, 'signal_score', e.signal_score, 'event_type', e.event_type, 'silent_change', e.silent_change, 'detected_at', e.detected_at, 'source', json_build_object('id', s.id, 'name', s.name)) as event152 from notifications n join alerts a on a.id = n.alert_id join events e on e.id = n.event_id join sources s on s.id = e.source_id153 where a.owner_token = ${owner} ${req.query.unread === "1" ? sql`and n.read_at is null` : sql``} order by n.created_at desc limit ${Math.min(200, Number(req.query.limit ?? 50))}`);154 const unread = (await db.execute<{ n: string }>(sql`select count(*)::text as n from notifications n join alerts a on a.id = n.alert_id where a.owner_token = ${owner} and n.read_at is null`)).rows[0]?.n;155 return { items: rows.rows, unread: Number(unread ?? 0) };156 });157 app.post("/api/v1/notifications/read", async (req, reply) => {158 const owner = requireOwner(req, reply);159 if (!owner) return;160 const body = z.object({ ids: z.array(z.number().int()).max(500).optional() }).parse(req.body ?? {});161 if (body.ids?.length) await db.execute(sql`update notifications n set read_at = now() from alerts a where a.id = n.alert_id and a.owner_token = ${owner} and n.id = any(${sql.raw("array[" + body.ids.map((i) => Number(i)).join(",") + "]::bigint[]")})`);162 else await db.execute(sql`update notifications n set read_at = now() from alerts a where a.id = n.alert_id and a.owner_token = ${owner} and n.read_at is null`);163 return { ok: true };164 });165166 // ---- Bookmarks ---------------------------------------------------------------------------------167 app.get<{ Querystring: { limit?: string } }>("/api/v1/bookmarks", async (req, reply) => {168 const owner = requireOwner(req, reply);169 if (!owner) return;170 const rows = await db.execute<Record<string, unknown>>(sql`select b.created_at as bookmarked_at, b.note, x.* from bookmarks b join lateral (select ${EVENT_SELECT} from events e join sources s on s.id = e.source_id join sensors sen on sen.id = e.sensor_id where e.id = b.event_id) x on true where b.owner_token = ${owner} order by b.created_at desc limit ${Math.min(500, Number(req.query.limit ?? 100))}`);171 return { items: rows.rows };172 });173 app.post("/api/v1/bookmarks", async (req, reply) => {174 const owner = requireOwner(req, reply);175 if (!owner) return;176 const body = z.object({ event_id: z.string().min(4).max(80), note: z.string().max(500).optional() }).parse(req.body ?? {});177 const ev = (await db.execute<{ id: string }>(sql`select id from events where id = ${body.event_id} or slug = ${body.event_id}`)).rows[0];178 if (!ev) return reply.status(404).send({ error: "event_not_found" });179 const count = (await db.execute<{ n: string }>(sql`select count(*)::text as n from bookmarks where owner_token = ${owner}`)).rows[0]?.n;180 if (Number(count) >= 2000) return reply.status(429).send({ error: "too_many_bookmarks" });181 await db.execute(sql`insert into bookmarks (owner_token, event_id, note) values (${owner}, ${ev.id}, ${body.note ?? null}) on conflict (owner_token, event_id) do update set note = coalesce(excluded.note, bookmarks.note)`);182 return { ok: true, event_id: ev.id };183 });184 app.delete<{ Params: { id: string } }>("/api/v1/bookmarks/:id", async (req, reply) => {185 const owner = requireOwner(req, reply);186 if (!owner) return;187 await db.execute(sql`delete from bookmarks where owner_token = ${owner} and (event_id = ${req.params.id} or event_id = (select id from events where slug = ${req.params.id}))`);188 return { ok: true };189 });190 app.get("/api/v1/bookmarks/ids", async (req, reply) => {191 const owner = requireOwner(req, reply);192 if (!owner) return;193 const rows = await db.execute<{ event_id: string }>(sql`select event_id from bookmarks where owner_token = ${owner}`);194 return { ids: rows.rows.map((r) => r.event_id) };195 });196197 // ---- Saved views -------------------------------------------------------------------------------198 app.get("/api/v1/views", async (req, reply) => {199 const owner = requireOwner(req, reply);200 if (!owner) return;201 return { items: (await db.execute<Record<string, unknown>>(sql`select id, name, query, created_at from saved_views where owner_token = ${owner} order by created_at`)).rows };202 });203 app.post("/api/v1/views", async (req, reply) => {204 const owner = requireOwner(req, reply);205 if (!owner) return;206 const body = z.object({ name: z.string().min(1).max(60), query: z.string().min(1).max(600) }).parse(req.body ?? {});207 const count = (await db.execute<{ n: string }>(sql`select count(*)::text as n from saved_views where owner_token = ${owner}`)).rows[0]?.n;208 if (Number(count) >= 50) return reply.status(429).send({ error: "too_many_views" });209 const id = newId("view");210 await db.execute(sql`insert into saved_views (id, owner_token, name, query) values (${id}, ${owner}, ${body.name}, ${body.query})`);211 return { id, ...body };212 });213 app.delete<{ Params: { id: string } }>("/api/v1/views/:id", async (req, reply) => {214 const owner = requireOwner(req, reply);215 if (!owner) return;216 await db.execute(sql`delete from saved_views where id = ${req.params.id} and owner_token = ${owner}`);217 return { ok: true };218 });219220 // ---- Custom URL monitors (spec §45, §87) -----------------------------------------------------------221 const monitorBody = z.object({222 url: z.string().url().max(2000),223 name: z.string().min(1).max(80).optional(),224 /** hourly | daily (maps to tier C / D) */225 frequency: z.enum(["hourly", "daily"]).default("hourly"),226 /** low = only big changes, normal, high = every meaningful change */227 sensitivity: z.enum(["low", "normal", "high"]).default("normal"),228 selector: z.string().max(200).optional(),229 keywords: z.array(z.string().min(1).max(60)).max(20).optional(),230 });231 const ownerSourceId = (owner: string): string => `custom-${createHash("sha256").update(owner).digest("hex").slice(0, 12)}`;232233 app.get("/api/v1/monitors", async (req, reply) => {234 const owner = requireOwner(req, reply);235 if (!owner) return;236 const sid = ownerSourceId(owner);237 const rows = await db.execute<Record<string, unknown>>(sql`select id, name, url, tier, health, status, enabled, config, next_check_at, last_check_at, last_change_at, last_event_at, last_status, last_error, total_runs, raw_changes, meaningful_changes, created_at from sensors where source_id = ${sid} order by created_at`);238 return { items: rows.rows, limit: config.monitorsPerOwner, source_id: sid };239 });240 app.post("/api/v1/monitors", async (req, reply) => {241 const owner = requireOwner(req, reply);242 if (!owner) return;243 const body = monitorBody.parse(req.body ?? {});244 let host: string;245 try {246 const u = await assertUrlAllowed(body.url);247 if (!["http:", "https:"].includes(u.url.protocol)) return reply.status(400).send({ error: "scheme_not_allowed" });248 host = u.url.hostname;249 } catch (e) {250 return reply.status(400).send({ error: "url_rejected", detail: e instanceof UrlPolicyError ? e.message : "URL not allowed" });251 }252 const sid = ownerSourceId(owner);253 const n = Number((await db.execute<{ n: string }>(sql`select count(*)::text as n from sensors where source_id = ${sid} and enabled`)).rows[0]?.n ?? 0);254 if (n >= config.monitorsPerOwner) return reply.status(429).send({ error: "monitor_limit_reached", limit: config.monitorsPerOwner });255 // Sensor auto-test (spec §72): fetch + normalize before activation; baseline is taken by the engine.256 const sensorId = `${sid}_${newId("m").slice(2)}`;257 const cfg: Record<string, unknown> = { custom: true, sensitivity: body.sensitivity, ...(body.selector ? { selector: body.selector } : {}), ...(body.keywords?.length ? { keywords: body.keywords } : {}) };258 const tier = body.frequency === "hourly" ? "C" : "D";259 const connector = getConnector("http");260 const endpoint = { id: sensorId, sourceId: sid, name: body.name ?? host, url: body.url, type: "HTML" as const, tier: tier as "C" | "D", connector: "http", config: cfg, etag: null, lastModified: null, state: null };261 const obs = await connector.fetch(endpoint);262 if (obs.error) return reply.status(422).send({ error: "fetch_failed", detail: `${obs.error.code}: ${obs.error.message}` });263 if (obs.meta.status >= 400) return reply.status(422).send({ error: "http_error", status: obs.meta.status });264 let norm;265 try {266 norm = await connector.normalize(endpoint, obs);267 } catch (e) {268 return reply.status(422).send({ error: "unparseable", detail: (e as Error).message });269 }270 if ((norm.text ?? "").length < 40 && norm.mode === "text") return reply.status(422).send({ error: "thin_content", detail: "The page has almost no server-rendered text (client-side app?)." });271 await db.execute(sql`insert into sources (id, name, domain, homepage, description, categories, tier, importance_weight, enabled, llm_enabled, kind, owner_token, first_party, notes)272 values (${sid}, ${"Custom monitors"}, ${"custom.websensor.io"}, ${null}, ${"Private URL monitors"}, ${"{custom}"}::text[], 'C', 0.5, true, false, 'custom', ${owner}, true, 'owner-scoped custom monitors') on conflict (id) do nothing`);273 await db.execute(sql`insert into sensors (id, source_id, name, url, type, connector, tier, importance_weight, config, priority, status, validated_at, enabled, next_check_at)274 values (${sensorId}, ${sid}, ${body.name ?? host}, ${body.url}, 'HTML', 'http', ${tier}, ${body.sensitivity === "high" ? 1.3 : body.sensitivity === "low" ? 0.7 : 1}, ${JSON.stringify(cfg)}::jsonb, 3, 'VALIDATED', now(), true, now())`);275 return { id: sensorId, source_id: sid, url: body.url, tier, status: "VALIDATED", test: { http_status: obs.meta.status, content_type: obs.meta.contentType, bytes: obs.meta.contentLength, title: norm.title ?? null, mode: norm.mode, extraction_confidence: norm.extractionConfidence } };276 });277 app.delete<{ Params: { id: string } }>("/api/v1/monitors/:id", async (req, reply) => {278 const owner = requireOwner(req, reply);279 if (!owner) return;280 const sid = ownerSourceId(owner);281 await db.execute(sql`delete from sensors where id = ${req.params.id} and source_id = ${sid}`);282 return { ok: true };283 });284 app.get<{ Params: { id: string }; Querystring: { limit?: string } }>("/api/v1/monitors/:id/events", async (req, reply) => {285 const owner = requireOwner(req, reply);286 if (!owner) return;287 const sid = ownerSourceId(owner);288 const sensor = (await db.execute<{ id: string }>(sql`select id from sensors where id = ${req.params.id} and source_id = ${sid}`)).rows[0];289 if (!sensor) return reply.status(404).send({ error: "not_found" });290 const [events, changes] = await Promise.all([291 listEvents({ sensor: sensor.id, limit: Math.min(100, Number(req.query.limit ?? 30)), includeCustom: true }),292 db.execute<Record<string, unknown>>(sql`select id, detected_at, kind, signal, change_class, field_changes, meaningful, event_id, old_snapshot_id, new_snapshot_id from changes where sensor_id = ${sensor.id} order by detected_at desc limit 30`).then((r) => r.rows),293 ]);294 return { events: events.items, changes };295 });296297 // keep eventConditions referenced for owner-scoped searches later298 void eventConditions;299}300