openvibers/openvibe.watch Commits
Actions: a fired condition notifies the owner (plan T18 step 4); fix repeat firing (#31)
Changes
@@@ -96,9 +96,9 @@ when checked from the site, fetching nothing.Check-now from the page respects the API's limits and its refusals: a second check while one runs ora check of a paused watch is the API's 409, said in words; a kind without a carrier is the API's 422;and the manual-check and write budgets (10/120 and 30/600 per person, a minute/hour) are enforced--with a friendly page. "Notify me" stores the notification action; a fired condition still emits--`watch.watch.triggered` through OpenVibe.Network as today (actions are step 4 and are not dispatched--yet, which the form says on the page).++with a friendly page. "Notify me" stores the notification action: when the condition fires the owner gets one++notification on OpenVibe (through OpenVibe.Network), and the watch page lists what each firing did under "What it did++when it fired".## The watch model@@@ -164,8 +164,23 @@ manual `POST …/check` never overlaps a scheduled one. One checkA condition that fires is recorded on the check run (`state: condition_met`, `triggers: 1`) andemits `watch.watch.triggered`; it fires again only for a value that is new or for a new run of--holding, never once per check while nothing moves. **Actions are step 4 of the plan**: a fired--condition records the trigger and emits the event, and nothing else is dispatched yet.++holding, never once per check while nothing moves (the last firing is compared by when it finished: the holding trail++starts at the observation the firing run recorded).++++## Actions++++A fired condition queues the watch's actions in the same transaction as the check run that fired++(`server/actions.js`, table `action_runs`): each action once per trigger. The action worker (every++`WATCH_ACTIONS_TICK_MS`, up to `WATCH_ACTIONS_BATCH` at a time, off with `WATCH_ACTIONS=off`) delivers them:++++| kind | what happens |++|---|---|++| `notification` | the owner gets one OpenVibe notification through OpenVibe.Network (`openvibe-sdk/notifications`, type `WATCH_TRIGGERED`): the watch's name, the value (and the value before), and a link to the watch page. A refusal retries after 1, 5 and 30 minutes, 2 and 6 hours, then `failed`; an owner Network does not know (a deleted account) is `skipped`; a person who switched the category off is Network's decision and counts as `sent`. A watch owned by an agent or a service has no one to notify (`skipped`). Needs the Network OAuth client secret and the grants `identity.subject.resolve` + `network.notifications.push` |++| `event`, `webhook`, `actor-task`, `codes-run`, `run-job` | recorded `unavailable` (`watch.action_unavailable`): their dispatchers are later steps, and the watch page says so |++++Rows are claimed with a short lease (FOR UPDATE SKIP LOCKED), so two processes never deliver one row at once.++Network keeps no idempotency record for a push, so a lost answer can repeat a notification: delivery is at least once.++`/api/ready` reports the queue (`actions: { enabled, queued, failed, sent }`).Nothing is invented: a value the source did not state is `null`, a failed check is a recorded state,and the validators/body hash are stored only after a body was read successfully, so an unreadable@@@ -383,8 +398,8 @@ Called elsewhere, as the service principal `watch` (client credentials from Open422), cross-site writes refused, and hostile names and values escaped on every pageNot yet demonstrated: a check against a real site from a deployed service (nothing is deployed--yet), the event/webhook carriers (step 5), the Run rung (step 6), Node probes (step 7) and the--actions (step 4) — each answers `watch.carrier_unavailable` and fetches nothing today.++yet), the event/webhook carriers (step 5), the Run rung (step 6) and Node probes (step 7) — each answers++`watch.carrier_unavailable` and fetches nothing today — and the action kinds beyond notifications.## Deploy@@@ -8,7 +8,7 @@"port": 4730,"domain": "openvibe.watch (a public server-rendered site + a loopback-only API; vhost in deploy/nginx; README \"Public site\" and \"Public host\")","contracts": "openvibe-contracts v0.129.0",-- "sdk": "openvibe-sdk v0.38.0",++ "sdk": "openvibe-sdk v0.39.0","shared": "openvibe-shared v3.0.0","capabilities": ["watch.watch.read",nnew file mode 100644@@@ -0,0 +1,26 @@++-- phase: expand++-- OpenVibe.Watch: the actions a fired condition sets in motion (plan T18 step 4, server/actions.js). One row per action++-- of the watch per trigger, written in the same transaction as the check run that fired, so a trigger queues each of++-- its actions exactly once; the action worker delivers the queued ones (notifications today) and records the outcome.++-- request holds what will be delivered (for a notification: network.notification-push-request@1 fields, never a++-- token or a secret).++++CREATE TABLE action_runs (++ rid bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,++ watch_id text COLLATE "C" NOT NULL,++ check_run_id text COLLATE "C" NOT NULL, -- ckr_<ULID>: the run that fired++ action_index integer NOT NULL, -- its place in the watch's action list++ kind text COLLATE "C" NOT NULL, -- notification | event | webhook | actor-task | codes-run | run-job++ owner_sub text COLLATE "C" NOT NULL,++ request jsonb NOT NULL DEFAULT '{}',++ state text COLLATE "C" NOT NULL DEFAULT 'queued'++ CHECK (state IN ('queued', 'sent', 'skipped', 'failed', 'unavailable')),++ attempts integer NOT NULL DEFAULT 0,++ next_attempt_at bigint NOT NULL DEFAULT 0,++ last_error text COLLATE "C",++ created_at bigint NOT NULL,++ finished_at bigint,++ UNIQUE (check_run_id, action_index)++);++CREATE INDEX action_runs_due ON action_runs (next_attempt_at, rid) WHERE state = 'queued';++CREATE INDEX action_runs_watch ON action_runs (watch_id, rid DESC);@@@ -13,7 +13,7 @@"express": "^4.21.2","fast-xml-parser": "^4.5.3","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.39.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.0","pg": "^8.23.0"},@@@ -785,9 +785,9 @@}},"node_modules/openvibe-sdk": {-- "version": "0.38.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.0",-- "integrity": "sha512-/Z+1kB5l4Hff0E1O1Zo+DcXaZX93wrmISOPP75sB7NqVUZVeKuJgjYu6vBwG8ufJjlk0vD3ZJ0ntBLI/3IKdpg==",++ "version": "0.39.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.39.0",++ "integrity": "sha512-unJTjjNZszufkl8GbCu9sMnVa/9qWi2sO7cMz9YZp3jLjkznZRCSEcTMNt18gkABY/eruV4BRzoWoOK5iP8J9g==","license": "MIT","engines": {"node": ">=20"@@@ -25,7 +25,7 @@"express": "^4.21.2","fast-xml-parser": "^4.5.3","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.39.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.0","pg": "^8.23.0"},@@@ -31,6 +31,7 @@ async function extraErase(t, subjects, counts) {const mine = '(SELECT id FROM watches WHERE owner_sub = ANY($1::text[]))';counts.add(counts.erased, 'observations', await t.exec(`DELETE FROM observations WHERE watch_id IN ${mine}`, [subjects]));counts.add(counts.erased, 'check_runs', await t.exec(`DELETE FROM check_runs WHERE watch_id IN ${mine}`, [subjects]));++ counts.add(counts.erased, 'action_runs', await t.exec(`DELETE FROM action_runs WHERE watch_id IN ${mine}`, [subjects]));counts.add(counts.erased, 'watch_endpoint_state', await t.exec(`DELETE FROM watch_endpoint_state WHERE watch_id IN ${mine}`, [subjects]));counts.add(counts.erased, 'watches', await t.exec('DELETE FROM watches WHERE owner_sub = ANY($1::text[])', [subjects]));}nnew file mode 100644@@@ -0,0 +1,173 @@++'use strict';++/**++ * Watch actions (plan T18 step 4): what a fired condition sets in motion.++ *++ * const actions = createActions({ db, config, notifications, now, log });++ * await actions.queue({ watch, run, owner, value, previousValue, observedAt }) inside settle()'s transaction++ * actions.start(); actions.stop(); await actions.tick(); await actions.status(); await actions.recent(watchId)++ *++ * queue() writes one action_runs row per action of the watch, in the SAME transaction as the check run that fired, so++ * a trigger queues each of its actions exactly once (UNIQUE (check_run_id, action_index)) and never without its run.++ *++ * Delivery, by kind:++ * notification the owner gets one OpenVibe notification through Network (openvibe-sdk/notifications, type++ * WATCH_TRIGGERED): the watch's name, the value and a link to the watch page. A person who switched++ * the category off is Network's decision and counts as sent; an owner Network does not know (a++ * deleted account) is `skipped`. A refusal retries after 1, 5 and 30 minutes, 2 and 6 hours, then++ * `failed`. Network keeps no idempotency record for a push, so a lost answer can repeat one: at least++ * once. A watch owned by an agent or a service has no one to notify: `skipped`.++ * event, webhook, actor-task, codes-run, run-job++ * `unavailable` with `watch.action_unavailable`: their dispatchers are later steps of the plan, and++ * the watch page says so rather than pretending they ran.++ *++ * The worker claims due rows with a short lease (FOR UPDATE SKIP LOCKED, next_attempt_at moved past the lease), so++ * two processes never deliver one row at the same time, and a process that dies mid-delivery leaves the row to the++ * next tick after the lease. Without the Network OAuth client secret there is no token to push with: rows wait and++ * status() says delivery is off.++ */++++const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++const BACKOFF_MS = [60_000, 5 * 60_000, 30 * 60_000, 2 * 3600_000, 6 * 3600_000];++const LEASE_MS = 2 * 60_000;++const TITLE_MAX = 120;++const MESSAGE_MAX = 300;++const LATER = new Set(['event', 'webhook', 'actor-task', 'codes-run', 'run-job']);++++const clip = (s, n) => { const t = String(s == null ? '' : s).replace(/\s+/g, ' ').trim(); return t.length > n ? `${t.slice(0, n - 1)}…` : t; };++++/** A value as a person reads it in one line. */++function valueText(v) {++ if (v == null) return 'no value';++ if (typeof v === 'string') return v;++ if (typeof v === 'number' || typeof v === 'boolean') return String(v);++ try { return JSON.stringify(v); } catch { return String(v); }++}++++/** The notification a fired watch sends its owner (network.notification-push-request@1 fields). */++function notificationFor({ config, watch, run, action, value, previousValue, observedAt }) {++ const now = valueText(value);++ const before = previousValue === undefined || previousValue === null ? null : valueText(previousValue);++ return {++ type: 'WATCH_TRIGGERED',++ service: 'watch',++ category: action.category || 'service',++ priority: action.priority || 'normal',++ icon: '👁',++ title: clip(`“${watch.name}” fired`, TITLE_MAX),++ message: clip(before !== null && before !== now ? `${now} (was ${before})` : now, MESSAGE_MAX),++ url: `${config.baseUrl}/watches/${encodeURIComponent(watch.id)}`,++ rich_content: { watch_id: watch.id, check_run_id: run.id, observed_at: observedAt ? new Date(observedAt).toISOString() : null },++ };++}++++function createActions({ db, config, notifications = null, now = () => Date.now(), log = console }) {++ const enabled = Boolean(notifications) && config.actions.enabled;++ let timer = null;++ let busy = null;++++ const st = {++ insert: db.prepare(`INSERT INTO action_runs (watch_id, check_run_id, action_index, kind, owner_sub, request, state, last_error, created_at, finished_at)++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (check_run_id, action_index) DO NOTHING`),++ claim: db.prepare(`SELECT rid, watch_id, kind, owner_sub, request, attempts FROM action_runs++ WHERE state = 'queued' AND next_attempt_at <= ? ORDER BY next_attempt_at, rid LIMIT ? FOR UPDATE SKIP LOCKED`),++ lease: db.prepare('UPDATE action_runs SET attempts = attempts + 1, next_attempt_at = ? WHERE rid = ?'),++ finish: db.prepare('UPDATE action_runs SET state = ?, last_error = ?, finished_at = ? WHERE rid = ?'),++ retry: db.prepare('UPDATE action_runs SET next_attempt_at = ?, last_error = ? WHERE rid = ?'),++ counts: db.prepare(`SELECT COUNT(*) FILTER (WHERE state = 'queued') AS queued, COUNT(*) FILTER (WHERE state = 'failed') AS failed,++ COUNT(*) FILTER (WHERE state = 'sent') AS sent FROM action_runs`),++ recent: db.prepare(`SELECT check_run_id, action_index, kind, state, attempts, last_error, created_at, finished_at FROM action_runs++ WHERE watch_id = ? ORDER BY rid DESC LIMIT ?`),++ };++++ /** One row per action of the watch, for the run that fired. Call inside the run's transaction. */++ async function queue({ watch, run, owner, value, previousValue, observedAt }) {++ const list = Array.isArray(watch.action) ? watch.action : [];++ const at = now();++ for (let i = 0; i < list.length; i++) {++ const action = list[i] || {};++ const kind = String(action.kind || '');++ let state = 'queued', error = null, request = {};++ if (kind === 'notification') {++ if (USER_SUBJECT.test(String(owner || ''))) request = notificationFor({ config, watch, run, action, value, previousValue, observedAt });++ else { state = 'skipped'; error = 'watch.action_no_person: the watch belongs to an agent or a service, not a person'; }++ } else if (LATER.has(kind)) {++ state = 'unavailable'; error = `watch.action_unavailable: ${kind} actions are a later step (plan T18)`;++ } else {++ state = 'unavailable'; error = `watch.action_unavailable: unknown action kind ${kind || '(none)'}`;++ }++ await st.insert.run(watch.id, run.id, i, kind || 'unknown', String(owner || ''), JSON.stringify(request), state, error, at, state === 'queued' ? null : at);++ }++ }++++ async function deliver(row) {++ const request = typeof row.request === 'string' ? JSON.parse(row.request) : row.request;++ if (row.kind !== 'notification') return { state: 'unavailable', error: `watch.action_unavailable: ${row.kind}` };++ const out = await notifications.push({ subjectId: row.owner_sub, ...request });++ if (out.sent) return { state: 'sent', error: out.skipped ? 'delivered; the person switched this category off' : null };++ return { state: 'skipped', error: 'watch.action_owner_unknown: Network does not know this owner (a deleted account?)' };++ }++++ /** One pass: claim due rows under a lease, deliver each, record the outcome. Returns counts. */++ async function tick() {++ if (!enabled) return { enabled: false };++ if (busy) return await busy;++ busy = (async () => {++ const out = { sent: 0, skipped: 0, retried: 0, failed: 0 };++ const at = now();++ const rows = await db.tx(async () => {++ const due = await st.claim.all(at, config.actions.batch);++ for (const r of due) await st.lease.run(at + LEASE_MS, r.rid);++ return due;++ });++ for (const r of rows) {++ const attempt = Number(r.attempts) + 1;++ try {++ const res = await deliver(r);++ await st.finish.run(res.state, res.error, now(), r.rid);++ if (res.state === 'sent') out.sent++; else out.skipped++;++ } catch (err) {++ const why = String((err && (err.code || err.message)) || err).slice(0, 300);++ if (attempt >= BACKOFF_MS.length + 1) {++ await st.finish.run('failed', why, now(), r.rid);++ out.failed++;++ log.warn(`[actions] ${r.watch_id}: ${r.kind} gave up after ${attempt} attempts: ${why}`);++ } else {++ await st.retry.run(now() + BACKOFF_MS[attempt - 1], why, r.rid);++ out.retried++;++ }++ }++ }++ return out;++ })();++ try { return await busy; } finally { busy = null; }++ }++++ function start() {++ if (timer || !enabled) return;++ timer = setInterval(() => { tick().catch((err) => log.warn(`[actions] tick: ${err.message}`)); }, config.actions.tickMs);++ timer.unref?.();++ }++ async function stop() {++ if (timer) clearInterval(timer);++ timer = null;++ if (busy) await busy.catch(() => {});++ }++++ async function status() {++ const c = await st.counts.get();++ return { enabled, queued: Number(c.queued) || 0, failed: Number(c.failed) || 0, sent: Number(c.sent) || 0 };++ }++++ async function recent(watchId, limit = 10) {++ return (await st.recent.all(watchId, limit)).map((r) => ({++ check_run_id: r.check_run_id, kind: r.kind, state: r.state, attempts: Number(r.attempts),++ error: r.last_error || null,++ created_at: new Date(Number(r.created_at)).toISOString(),++ finished_at: r.finished_at == null ? null : new Date(Number(r.finished_at)).toISOString(),++ }));++ }++++ return { queue, tick, start, stop, status, recent, enabled };++}++++module.exports = { createActions, notificationFor, BACKOFF_MS };@@@ -58,7 +58,7 @@ const CSP = "default-src 'self'; script-src 'self' 'unsafe-inline' https://openv+ "connect-src 'self' https://openvibe.network https://openvibe.events https://cloudflareinsights.com; "+ "frame-src 'self' https://openvibe.network; frame-ancestors 'none'; object-src 'none'; base-uri 'self'; form-action 'self' https://openvibe.network";--function createApp({ config, db, registry, check, observations, scheduler, auth, outbox, now, log = console, limitsNow = null, sessions, keyStore, siteLimits, fetchImpl = globalThis.fetch, accountData = null, accountSend = null }) {++function createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions = null, now, log = console, limitsNow = null, sessions, keyStore, siteLimits, fetchImpl = globalThis.fetch, accountData = null, accountSend = null }) {const app = express();app.disable('x-powered-by');app.set('trust proxy', config.trustProxy != null ? config.trustProxy : 'loopback');@@@ -94,7 +94,7 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,// Readiness (openvibe-shared/ready): 503 only when the database fails; the Network key and the// check worker (running, queue keeping up) are optional and degrade it (see observability.js).-- const readiness = createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, now, release: release.release });++ const readiness = createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, actions, now, release: release.release });app.get('/api/ready', readiness.handler);// GET /release.json (ADR-016) and POST /release-metrics (open tabs' update reports into /metrics).release.mount(app, { registry: metrics.registry });@@@ -139,7 +139,7 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,if (!req.path.startsWith('/api') && !req.path.startsWith('/auth')) res.setHeader('Content-Security-Policy', CSP);next();});-- app.use(createPageRoutes({ config, db, registry, check, observations, sessions, siteLimits, log }));++ app.use(createPageRoutes({ config, db, registry, check, observations, sessions, siteLimits, actions, log }));app.use((req, res) => send(res, 404, {viewer: req.viewer, config, path: req.originalUrl, title: 'Not found',@@@ -12,7 +12,7 @@* updated with what the site said (ETag, Last-Modified, status, body hash)* 5 extract → value; record an observation (value, hashes, changed, condition, retention)* 6 evaluate the condition; a condition that MET writes check_runs(triggers = 1, state-- * 'condition_met') and emits watch.watch.triggered. Actions are step 4 of the plan and are++ * 'condition_met'), emits watch.watch.triggered and queues the watch's actions (server/actions.js). Formerly* NOT dispatched here: a met condition records the trigger and emits the event, nothing else* 7 arm next_due_at (+ jitter, or exponential backoff after a failure, capped) and not_before* (Retry-After)@@@ -34,7 +34,7 @@ const FAILURE_STATES = new Set(['http_error', 'timeout', 'parse_error', 'rate_liconst PERSON_OWNER = /^(usr|agt)_[0-9A-HJKMNP-TV-Z]{26}$/;const USER_OWNER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;--function createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, now = () => Date.now(), log = console, ids = contractIds }) {++function createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions = null, now = () => Date.now(), log = console, ids = contractIds }) {const inflight = new Set();const st = {@@@ -55,7 +55,7 @@ function createCheck({ db, registry, config, carriers, extract, condition, obserrun: db.prepare('SELECT * FROM check_runs WHERE id = ?'),// The last check that fired the condition: a repeat needs a value that is new or a new run// of holding, so a satisfied watch does not fire once per check while nothing moves.-- lastTrigger: db.prepare('SELECT id, started_at FROM check_runs WHERE watch_id = ? AND triggers > 0 ORDER BY rid DESC LIMIT 1'),++ lastTrigger: db.prepare('SELECT id, started_at, finished_at FROM check_runs WHERE watch_id = ? AND triggers > 0 ORDER BY rid DESC LIMIT 1'),};const boolOf = (v) => Boolean(Number(v));@@@ -161,7 +161,7 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser* the events and the watch's re-armed runtime state all commit together, or none of them do.* `failures` is the consecutive failure count (0 when the check succeeded).*/-- async function settle({ watch, run, failures, successAt = null, notBefore = 0, events = [], observation = null, endpoint = null }) {++ async function settle({ watch, run, failures, successAt = null, notBefore = 0, events = [], observation = null, endpoint = null, fired = null }) {const armed = nextDue(watch, run.finished_at, { failures, maxBackoffMs: config.worker.maxBackoffMs });await db.tx(async () => {if (observation) await observations.insert(observation);@@@ -172,6 +172,8 @@ function createCheck({ db, registry, config, carriers, extract, condition, obserfailures, next_due: armed, not_before: notBefore,});for (const e of events || []) await outbox.emit(envelope(e, config.serviceId, now));++ // A run that fired queues the watch's actions in this same commit (server/actions.js): once per trigger.++ if (fired && actions) await actions.queue({ watch, run, ...fired });});outbox.kick().catch(() => { });return { run, observation, met: run.triggers > 0, failures };@@@ -312,7 +314,10 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser// value that is new (changed) or for a new run of holding (the trail began after the last// firing) — never once per check while nothing moves.const lastFired = await st.lastTrigger.get(watch.id);-- const repeat = verdict.met && !verdict.changed && lastFired != null && verdict.since != null && lastFired.started_at >= verdict.since;++ // The trail's `since` is the observed_at of its first holding observation, which the run that recorded it stamps++ // with its finish time; the firing run's start is always earlier, so the last firing is compared by when it++ // finished (comparing its start made every check while the condition held fire again).++ const repeat = verdict.met && !verdict.changed && lastFired != null && verdict.since != null && Number(lastFired.finished_at) >= verdict.since;const met = verdict.met && !repeat;if (verdict.error && !patternError) patternError = verdict.error;@@@ -353,7 +358,8 @@ function createCheck({ db, registry, config, carriers, extract, condition, obserwatch, owner: row.owner_sub, observation, value, met, run,firedOn, firedAt: finishedAt, previousValue, failures: 0,});-- const result = await settle({ watch, run, failures: 0, successAt: finishedAt, events, observation, endpoint });++ const fired = met ? { owner: row.owner_sub, value, previousValue, observedAt: firedOn ? Number(firedOn.observed_at) : finishedAt } : null;++ const result = await settle({ watch, run, failures: 0, successAt: finishedAt, events, observation, endpoint, fired });return {...result, met, value, verdict: { ...verdict, met, repeat },observation: observation ? observations.view(observation) : null,@@@ -79,6 +79,14 @@ function load(env = process.env) {secrets: String(env.WATCH_EVENTS_SECRET || '').split(',').map((x) => x.trim()).filter(Boolean),},++ // Actions (server/actions.js): every WATCH_ACTIONS_TICK_MS deliver up to WATCH_ACTIONS_BATCH queued actions of fired++ // watches. Notifications need the Network OAuth client secret (a token for network.notifications.push).++ actions: {++ enabled: env.WATCH_ACTIONS !== 'off',++ tickMs: int(env.WATCH_ACTIONS_TICK_MS, 5000),++ batch: int(env.WATCH_ACTIONS_BATCH, 20),++ },++// The check worker: every WATCH_TICK_MS, start checks for active watches whose turn has come.worker: {enabled: env.WATCH_WORKER !== 'off',@@@ -74,9 +74,8 @@ const OPS = {};const NEEDS_VALUE = new Set(['gt', 'gte', 'lt', 'lte', 'eq', 'ne', 'contains', 'matches']);const NOTIFY_ACTION = [{ kind: 'notification', category: 'service', priority: 'normal' }];--// With notifications off the watch still needs an action (the contract requires one); actions are--// step 4 of the plan and none is dispatched yet, so this records the person's intent without changing--// what happens today (see how-it-works and the note on the form).++// With notifications off the watch still needs an action (the contract requires one): an event, whose dispatcher is a++// later step, so a quiet watch records its trigger and tells nobody (server/actions.js records it as unavailable).const QUIET_ACTION = [{ kind: 'event', event_type: 'watch.watch.triggered' }];const INT = (v, d) => { const n = parseInt(v, 10); return Number.isFinite(n) ? n : d; };@@@ -92,7 +91,7 @@ function sameOrigin(req, baseUrl) {}function createPageRoutes(ctx) {-- const { config, db, registry, check, observations, sessions, log = console } = ctx;++ const { config, db, registry, check, observations, sessions, actions = null, log = console } = ctx;const r = express.Router();const formBody = express.urlencoded({ extended: false, limit: '32kb' });@@@ -323,10 +322,10 @@ ${extractionBlock}${conditionBlock}<fieldset><legend>How often, and how to tell you</legend>${field('Cadence', cadence, { name: 'cadence', error: errors.cadence, hint: 'Watch never checks more often than this, and can wait longer when the site asks it to.' })}--${checkbox('Notify me when it fires', { name: 'notify', checked: fields.notify === 'on', hint: 'A fired condition becomes a WATCH_TRIGGERED notification through OpenVibe.Network, as today.' })}++${checkbox('Notify me when it fires', { name: 'notify', checked: fields.notify === 'on', hint: 'When the condition fires you get one notification on OpenVibe (your notifications on any OpenVibe site), with the value and a link here.' })}</fieldset><p class="form-actions"><button class="btn primary" type="submit">${submit}</button> <a href="/watches">Cancel</a></p>--<p class="muted small">Actions are step 4 of the plan: a fired condition records its trigger and emits <code>watch.watch.triggered</code> today, and no other action is dispatched yet — turning notifications off does not silence that until actions run.</p>++<p class="muted small">Notifications are delivered today. Events, webhooks and tasks for OpenVibe.Actor, Codes and Run are later steps: a watch that asks for one records it as not available yet.</p></form>`;}@@@ -517,6 +516,7 @@ ${watchForm({ template, fields, errors, action: '/watches/new', submit: 'Createconst latest = await observations.latest(row.id);const obs = await observations.list(row.id, { before: ridBefore(req.query.obs_before), limit: obsLimit(req.query.obs_limit) });const runs = await check.list(row.id, { before: ridBefore(req.query.checks_before), limit: obsLimit(req.query.checks_limit) });++ const acts = actions ? await actions.recent(row.id, 10) : [];const obsRows = obs.map((o) => [time(new Date(Number(o.observed_at)).toISOString()),@@@ -536,6 +536,14 @@ ${watchForm({ template, fields, errors, action: '/watches/new', submit: 'Create];});++ const ACT_WORDS = { queued: ['waiting to send', 'info'], sent: ['sent', 'ok'], skipped: ['not sent', 'warn'], failed: ['failed', 'bad'], unavailable: ['not available yet', 'warn'] };++ const actRows = acts.map((a) => [++ time(a.created_at),++ a.kind === 'notification' ? 'Notification' : html`<code>${a.kind}</code>`,++ badge((ACT_WORDS[a.state] || [a.state])[0], (ACT_WORDS[a.state] || [null, 'info'])[1]),++ a.error ? html`<span class="muted small">${a.error}</span>` : (a.attempts > 1 ? html`<span class="muted small">${a.attempts} attempts</span>` : ''),++ ]);++const ran = String(req.query.ran || '');const actionForms = html`<div class="actions">${v.status === 'active'@@@ -567,6 +575,8 @@ ${actionForms}<section aria-labelledby="h-obs"><h2 id="h-obs">Observation history</h2>${table(['When', 'Value', 'Changed', 'Condition'], obsRows, { empty: 'No observations yet. Watch records one when a check reads a value.' })}<nav class="pager">${nextObs ? html`<a class="btn" href="${nextObs}">Older observations</a>` : html`<span class="muted">That is every observation kept.</span>`}</nav></section>++<section aria-labelledby="h-acts"><h2 id="h-acts">What it did when it fired</h2>++${table(['When', 'Action', 'Result', 'Detail'], actRows, { empty: 'Nothing yet: the condition has not fired since actions began (2026-10-10).' })}</section><section aria-labelledby="h-runs"><h2 id="h-runs">Check runs</h2><p class="muted small">Every check, whatever it found. A skipped kind, a site error and a wait are recorded states, not silence.</p>${table(['When', 'State', 'HTTP', 'Detail', 'Read', 'Fired'], runRows, { empty: 'No checks yet.' })}@@@ -23,7 +23,10 @@ const { createCheck } = require('./check');const { createScheduler } = require('./scheduler');const { createServiceOutbox } = require('openvibe-sdk/events');const { validate } = require('openvibe-contracts');--const { jwksClient } = require('openvibe-sdk/auth');++const { jwksClient, createServiceTokenClient } = require('openvibe-sdk/auth');++const { createClient } = require('openvibe-sdk/core');++const { createNotificationsClient } = require('openvibe-sdk/notifications');++const { createActions } = require('./actions');const { gracefulStop } = require('openvibe-sdk/service');const { createAuth } = require('./auth');const { createSessions } = require('./sessions');@@@ -33,7 +36,7 @@ const { createApp } = require('./app');const accountDataLib = require('./account-data');const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');--async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null } = {}) {++async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null } = {}) {config = config || load();// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.const db = givenDb || await openDb(config, { log });@@@ -54,7 +57,16 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchconst spacer = createSpacer({ hostMinIntervalMs: config.fetch.hostMinIntervalMs });// Every carrier goes out through the spacer, so two watches over one host never hammer it.const carriers = createCarriers({ fetcher: spacedFetcher(fetcher, spacer), config, log });-- const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, now, log });++ // Actions of fired watches (plan T18 step 4): notifications go to the owner through Network with Watch's own++ // client-credentials token; without the client secret they wait (actions.status() says delivery is off).++ const notifications = config.oauth.clientSecret++ ? createNotificationsClient(createClient({++ tokenProvider: createServiceTokenClient({ network: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }),++ baseUrls: { network: config.networkInternalUrl }, fetch: fetchImpl,++ }))++ : null;++ const actions = createActions({ db, config, notifications: givenNotifications || notifications, now, log });++ const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, now, log });const scheduler = createScheduler({ db, check, config, now, log });const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });const auth = createAuth({ config, log });@@@ -69,13 +81,14 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchconst accountSend = givenSend || (config.oauth.clientSecret? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }): null);-- const app = createApp({ config, db, registry, check, observations, scheduler, auth, outbox, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend });++ const app = createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend });// One JWKS client for the process (the SDK shares it with verifyUserToken): refresh in the background// on an unref'd timer, keeping the last good keys through a Network outage.keys.start();const keyLoaded = keys.keys().catch(() => null);outbox.start();if (config.worker.enabled) scheduler.start();++ actions.start();const pruneTimer = setInterval(async () => {try { await observations.prune(); } catch (err) { log.error(`[observations] prune: ${err.message}`); }try { await sessions.prune(); } catch (err) { log.error(`[sessions] prune: ${err.message}`); }@@@ -105,6 +118,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchif (subscriptions) subscriptions.stop();keys.stop();await scheduler.stop();++ await actions.stop();await outbox.stop();if (server) {server.closeAllConnections?.();@@@ -113,7 +127,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetchif (!givenDb) await db.close();}-- return { config, db, registry, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };++ return { config, db, registry, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, actions, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };}/**@@@ -48,7 +48,7 @@ function readers(db, { startedAt }) {};}--function createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, now, release = null }) {++function createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, actions = null, now, release = null }) {const read = readers(db, { startedAt: now() });const iso = (v) => (v == null ? null : new Date(v).toISOString());return createReadiness({@@@ -98,6 +98,7 @@ function createWatchReadiness({ db, config, registry, check, observations, schedchecks_in_flight: check.inflight().length,observations: dbOk ? await read.observations() : null,outbox: dbOk ? await outbox.status() : null,++ actions: dbOk && actions ? await actions.status() : null,};},});nnew file mode 100644@@@ -0,0 +1,154 @@++'use strict';++/**++ * Actions (plan T18 step 4): a condition that fires queues the watch's actions in the run's own transaction, once per++ * trigger; the worker delivers a notification to the owner through Network (openvibe-sdk/notifications), retries a++ * refusal with backoff, skips an owner Network does not know and an owner that is not a person, and records the kinds++ * whose dispatchers are later steps as unavailable. The watch page says what happened.++ */++const assert = require('assert');++const { boot, site, suite, newUser, newAgent } = require('./helpers');++++const t = suite('actions');++let svc;++let web;++let price = '19.99';++const pushes = [];++let mode = 'ok';++const notifications = {++ async push(req) {++ pushes.push(req);++ if (mode === 'fail') { const e = new Error('Network answered 503'); e.status = 503; e.code = 'notifications.unavailable'; throw e; }++ if (mode === 'unknown') return { sent: false, reason: 'unknown_subject' };++ return { sent: true, skipped: mode === 'muted' };++ },++};++++const person = { sub: newUser() };++const agent = { sub: newAgent() };++const def = (overrides = {}) => ({++ name: 'Widget price',++ source: { kind: 'http', url: `${web.origin}/price`, format: null },++ cadence: { every_sec: 900 },++ extraction: { kind: 'text' },++ condition: { op: 'lt', value: 10 },++ action: [{ kind: 'notification', priority: 'high' }],++ ...overrides,++});++const create = async (principal, overrides = {}) => await svc.registry.create(def(overrides), principal);++const check = async (id) => await svc.check.run(id, { trigger: 'schedule' });++const acts = async (id) => await svc.db.prepare('SELECT * FROM action_runs WHERE watch_id = ? ORDER BY rid').all(id);++const setPrice = async (v) => { await fetch(`${web.origin}/set?v=${v}`); };++/** Fire a watch: one check at 19.99 (nothing), then the price drops below 10. */++async function fire(w) {++ await setPrice('19.99');++ await check(w.id);++ await setPrice('7.50');++ const out = await check(w.id);++ assert.strictEqual(out.run.state, 'condition_met');++ return out;++}++++t('boot', async () => {++ web = await site({++ '/price': () => ({ headers: { 'content-type': 'text/plain' }, body: price }),++ '/set': (req, _res, ctx) => { price = ctx.url.searchParams.get('v') || price; return { body: 'ok' }; },++ });++ svc = await boot({ notifications });++});++++t('a check that does not fire queues nothing; one that fires queues each action once, with its run', async () => {++ mode = 'ok'; pushes.length = 0;++ const w = await create(person, { action: [{ kind: 'notification', priority: 'high' }, { kind: 'webhook', url: 'https://hooks.example.org/w', secret_env: 'WATCH_HOOK_TEST' }] });++ await setPrice('19.99');++ await check(w.id);++ assert.deepStrictEqual(await acts(w.id), [], 'no trigger, no action');++ await setPrice('7.50');++ const out = await check(w.id);++ const rows = await acts(w.id);++ assert.deepStrictEqual(rows.map((r) => [r.kind, r.state, r.check_run_id, Number(r.action_index)]),++ [['notification', 'queued', out.run.id, 0], ['webhook', 'unavailable', out.run.id, 1]]);++ assert.match(rows[1].last_error, /watch\.action_unavailable/);++++ const r = await svc.actions.tick();++ assert.deepStrictEqual(r, { sent: 1, skipped: 0, retried: 0, failed: 0 });++ assert.strictEqual(pushes.length, 1);++ const p = pushes[0];++ assert.strictEqual(p.subjectId, person.sub);++ assert.deepStrictEqual([p.type, p.service, p.category, p.priority], ['WATCH_TRIGGERED', 'watch', 'service', 'high']);++ assert.strictEqual(p.title, '“Widget price” fired');++ assert.strictEqual(p.message, '7.50 (was 19.99)');++ assert.strictEqual(p.url, `https://openvibe.watch/watches/${w.id}`);++ assert.strictEqual(p.rich_content.check_run_id, out.run.id);++ assert.strictEqual((await acts(w.id))[0].state, 'sent');++++ await svc.actions.tick();++ assert.strictEqual(pushes.length, 1, 'a sent action is never sent again');++ await check(w.id); // 7.50 still holds: no new trigger, no new action++ assert.strictEqual((await acts(w.id)).length, 2);++});++++t('a refusal waits with backoff; the next tick does not retry early; a later success sends it', async () => {++ pushes.length = 0; mode = 'fail';++ const w = await create(person);++ await fire(w);++ let r = await svc.actions.tick();++ assert.deepStrictEqual(r, { sent: 0, skipped: 0, retried: 1, failed: 0 });++ let [row] = await acts(w.id);++ assert.strictEqual(row.state, 'queued');++ assert.strictEqual(Number(row.attempts), 1);++ assert.match(row.last_error, /notifications\.unavailable/);++ assert.ok(Number(row.next_attempt_at) > Date.now() + 50_000, 'the first retry waits about a minute');++ r = await svc.actions.tick();++ assert.strictEqual(pushes.length, 1, 'not due yet: no second push');++ mode = 'ok';++ await svc.db.prepare('UPDATE action_runs SET next_attempt_at = 0 WHERE watch_id = ?').run(w.id);++ r = await svc.actions.tick();++ assert.deepStrictEqual(r, { sent: 1, skipped: 0, retried: 0, failed: 0 });++ [row] = await acts(w.id);++ assert.deepStrictEqual([row.state, Number(row.attempts)], ['sent', 2]);++});++++t('an owner Network does not know is skipped; a muted category is delivered; an agent-owned watch notifies nobody', async () => {++ pushes.length = 0; mode = 'unknown';++ const gone = await create(person, { name: 'gone' });++ await fire(gone);++ await svc.actions.tick();++ assert.strictEqual((await acts(gone.id))[0].state, 'skipped');++ assert.match((await acts(gone.id))[0].last_error, /owner_unknown/);++++ mode = 'muted';++ const muted = await create(person, { name: 'muted' });++ await fire(muted);++ await svc.actions.tick();++ assert.strictEqual((await acts(muted.id))[0].state, 'sent', 'Network decided: the person switched the category off');++++ mode = 'ok';++ const before = pushes.length;++ const bot = await create(agent, { name: 'agent watch' });++ await fire(bot);++ const [row] = await acts(bot.id);++ assert.deepStrictEqual([row.state, /action_no_person/.test(row.last_error)], ['skipped', true]);++ await svc.actions.tick();++ assert.strictEqual(pushes.length, before, 'nothing is pushed for an agent');++});++++t('readiness counts the queue, and the watch page says what the watch did when it fired', async () => {++ const st = await svc.actions.status();++ assert.strictEqual(st.enabled, true);++ assert.ok(st.sent >= 3 && st.queued === 0 && st.failed === 0, JSON.stringify(st));++ const w = await create(person, { name: 'page check' });++ mode = 'ok';++ await fire(w);++ await svc.actions.tick();++ const cookie = await svc.signIn({ subject: person.sub, username: 'ana' });++ const res = await fetch(`${svc.base}/watches/${w.id}`, { headers: { cookie } });++ const body = await res.text();++ assert.strictEqual(res.status, 200);++ assert.match(body, /What it did when it fired/);++ assert.match(body, /Notification/);++ assert.match(body, /class="badge ok">sent</);++});++++t('done', async () => { await web.close(); await svc.stop(); });++++t.run();@@@ -117,6 +117,11 @@ t('a condition that fires emits watch.watch.triggered with the observation thatassert.deepStrictEqual([Number(last.changed), Number(last.condition_met)], [1, 1]);valid('watch.observation@1', svc.observations.view(last));valid('watch.check-run@1', svc.check.runView((await runs(w.id)).pop()));++++ // While 7.50 keeps holding, the next checks do not fire again (the trail began with the run that fired).++ const again = await check(w.id);++ assert.deepStrictEqual([again.run.triggers, again.met], [0, false], 'a condition that keeps holding fires once');++ assert.strictEqual((await envelopes('watch.watch.triggered')).filter((e) => e.payload.watch_id === w.id).length, 1);});t('a failing check is a recorded state and a watch.check.failed event, and it backs off', async () => {@@@ -49,7 +49,7 @@ const newAgent = () => require('openvibe-contracts').ids.newId('agent');* the caller says otherwise (a test of the SSRF guard boots with allowPrivate: false).* log: Watch's logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).*/--async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null } = {}) {++async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null } = {}) {// A stub Network JWKS serving the generated signing key, so the SDK's JWKS client has something real// to fetch and verify against (no test touches the internet). Its URL is overridable via env.const jwksSite = await site({ '/api/.well-known/jwks': () => ({ body: JSON.stringify({ keys: [publicJwk] }) }) });@@@ -73,11 +73,13 @@ async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, lWATCH_HOST_MIN_INTERVAL_MS: '0',WATCH_FETCH_TIMEOUT_MS: '1500',WATCH_TICK_MS: '50',++ // Actions are delivered when a test calls svc.actions.tick(), never by a background timer mid-test.++ WATCH_ACTIONS_TICK_MS: '3600000',...env,});// One database per boot (PGlite, or WATCH_TEST_STORE=pg: the containers), dropped when the boot stops.const testdb = await require('./db').testDb();-- const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(now ? { now } : {}) });++ const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(now ? { now } : {}) });const base = `http://127.0.0.1:${h.server.address().port}`;/** A session cookie for a person, as /auth/callback would have set it. */
@@ -96,9 +96,9 @@ when checked from the site, fetching nothing.
@@ -96,9 +96,9 @@ when checked from the site, fetching nothing.
Check-now from the page respects the API's limits and its refusals: a second check while one runs or
Check-now from the page respects the API's limits and its refusals: a second check while one runs or
a check of a paused watch is the API's 409, said in words; a kind without a carrier is the API's 422;
a check of a paused watch is the API's 409, said in words; a kind without a carrier is the API's 422;
and the manual-check and write budgets (10/120 and 30/600 per person, a minute/hour) are enforced
and the manual-check and write budgets (10/120 and 30/600 per person, a minute/hour) are enforced
-with a friendly page. "Notify me" stores the notification action; a fired condition still emits
-`watch.watch.triggered` through OpenVibe.Network as today (actions are step 4 and are not dispatched
-yet, which the form says on the page).
+with a friendly page. "Notify me" stores the notification action: when the condition fires the owner gets one
+notification on OpenVibe (through OpenVibe.Network), and the watch page lists what each firing did under "What it did
+when it fired".
## The watch model
## The watch model
@@ -164,8 +164,23 @@ manual `POST …/check` never overlaps a scheduled one. One check
@@ -164,8 +164,23 @@ manual `POST …/check` never overlaps a scheduled one. One check
A condition that fires is recorded on the check run (`state: condition_met`, `triggers: 1`) and
A condition that fires is recorded on the check run (`state: condition_met`, `triggers: 1`) and
emits `watch.watch.triggered`; it fires again only for a value that is new or for a new run of
emits `watch.watch.triggered`; it fires again only for a value that is new or for a new run of
-holding, never once per check while nothing moves. **Actions are step 4 of the plan**: a fired
-condition records the trigger and emits the event, and nothing else is dispatched yet.
+holding, never once per check while nothing moves (the last firing is compared by when it finished: the holding trail
+starts at the observation the firing run recorded).
+
+## Actions
+
+A fired condition queues the watch's actions in the same transaction as the check run that fired
+(`server/actions.js`, table `action_runs`): each action once per trigger. The action worker (every
+`WATCH_ACTIONS_TICK_MS`, up to `WATCH_ACTIONS_BATCH` at a time, off with `WATCH_ACTIONS=off`) delivers them:
+
+| kind | what happens |
+|---|---|
+| `notification` | the owner gets one OpenVibe notification through OpenVibe.Network (`openvibe-sdk/notifications`, type `WATCH_TRIGGERED`): the watch's name, the value (and the value before), and a link to the watch page. A refusal retries after 1, 5 and 30 minutes, 2 and 6 hours, then `failed`; an owner Network does not know (a deleted account) is `skipped`; a person who switched the category off is Network's decision and counts as `sent`. A watch owned by an agent or a service has no one to notify (`skipped`). Needs the Network OAuth client secret and the grants `identity.subject.resolve` + `network.notifications.push` |
+| `event`, `webhook`, `actor-task`, `codes-run`, `run-job` | recorded `unavailable` (`watch.action_unavailable`): their dispatchers are later steps, and the watch page says so |
+
+Rows are claimed with a short lease (FOR UPDATE SKIP LOCKED), so two processes never deliver one row at once.
+Network keeps no idempotency record for a push, so a lost answer can repeat a notification: delivery is at least once.
+`/api/ready` reports the queue (`actions: { enabled, queued, failed, sent }`).
Nothing is invented: a value the source did not state is `null`, a failed check is a recorded state,
Nothing is invented: a value the source did not state is `null`, a failed check is a recorded state,
and the validators/body hash are stored only after a body was read successfully, so an unreadable
and the validators/body hash are stored only after a body was read successfully, so an unreadable
@@ -383,8 +398,8 @@ Called elsewhere, as the service principal `watch` (client credentials from Open
@@ -383,8 +398,8 @@ Called elsewhere, as the service principal `watch` (client credentials from Open
422), cross-site writes refused, and hostile names and values escaped on every page
422), cross-site writes refused, and hostile names and values escaped on every page
Not yet demonstrated: a check against a real site from a deployed service (nothing is deployed
Not yet demonstrated: a check against a real site from a deployed service (nothing is deployed
-yet), the event/webhook carriers (step 5), the Run rung (step 6), Node probes (step 7) and the
-actions (step 4) — each answers `watch.carrier_unavailable` and fetches nothing today.
+yet), the event/webhook carriers (step 5), the Run rung (step 6) and Node probes (step 7) — each answers
+`watch.carrier_unavailable` and fetches nothing today — and the action kinds beyond notifications.
## Deploy
## Deploy
@@ -8,7 +8,7 @@
@@ -8,7 +8,7 @@
"port": 4730,
"port": 4730,
"domain": "openvibe.watch (a public server-rendered site + a loopback-only API; vhost in deploy/nginx; README \"Public site\" and \"Public host\")",
"domain": "openvibe.watch (a public server-rendered site + a loopback-only API; vhost in deploy/nginx; README \"Public site\" and \"Public host\")",
"contracts": "openvibe-contracts v0.129.0",
"contracts": "openvibe-contracts v0.129.0",
- "sdk": "openvibe-sdk v0.38.0",
+ "sdk": "openvibe-sdk v0.39.0",
"shared": "openvibe-shared v3.0.0",
"shared": "openvibe-shared v3.0.0",
"capabilities": [
"capabilities": [
"watch.watch.read",
"watch.watch.read",
new file mode 100644
new file mode 100644
@@ -0,0 +1,26 @@
@@ -0,0 +1,26 @@
+-- phase: expand
+-- OpenVibe.Watch: the actions a fired condition sets in motion (plan T18 step 4, server/actions.js). One row per action
+-- of the watch per trigger, written in the same transaction as the check run that fired, so a trigger queues each of
+-- its actions exactly once; the action worker delivers the queued ones (notifications today) and records the outcome.
+-- request holds what will be delivered (for a notification: network.notification-push-request@1 fields, never a
+-- token or a secret).
+
+CREATE TABLE action_runs (
+ rid bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+ watch_id text COLLATE "C" NOT NULL,
+ check_run_id text COLLATE "C" NOT NULL, -- ckr_<ULID>: the run that fired
+ action_index integer NOT NULL, -- its place in the watch's action list
+ kind text COLLATE "C" NOT NULL, -- notification | event | webhook | actor-task | codes-run | run-job
+ owner_sub text COLLATE "C" NOT NULL,
+ request jsonb NOT NULL DEFAULT '{}',
+ state text COLLATE "C" NOT NULL DEFAULT 'queued'
+ CHECK (state IN ('queued', 'sent', 'skipped', 'failed', 'unavailable')),
+ attempts integer NOT NULL DEFAULT 0,
+ next_attempt_at bigint NOT NULL DEFAULT 0,
+ last_error text COLLATE "C",
+ created_at bigint NOT NULL,
+ finished_at bigint,
+ UNIQUE (check_run_id, action_index)
+);
+CREATE INDEX action_runs_due ON action_runs (next_attempt_at, rid) WHERE state = 'queued';
+CREATE INDEX action_runs_watch ON action_runs (watch_id, rid DESC);
@@ -13,7 +13,7 @@
@@ -13,7 +13,7 @@
"express": "^4.21.2",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
"fast-xml-parser": "^4.5.3",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.39.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
},
},
@@ -785,9 +785,9 @@
@@ -785,9 +785,9 @@
}
}
},
},
"node_modules/openvibe-sdk": {
"node_modules/openvibe-sdk": {
- "version": "0.38.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.0",
- "integrity": "sha512-/Z+1kB5l4Hff0E1O1Zo+DcXaZX93wrmISOPP75sB7NqVUZVeKuJgjYu6vBwG8ufJjlk0vD3ZJ0ntBLI/3IKdpg==",
+ "version": "0.39.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.39.0",
+ "integrity": "sha512-unJTjjNZszufkl8GbCu9sMnVa/9qWi2sO7cMz9YZp3jLjkznZRCSEcTMNt18gkABY/eruV4BRzoWoOK5iP8J9g==",
"license": "MIT",
"license": "MIT",
"engines": {
"engines": {
"node": ">=20"
"node": ">=20"
@@ -25,7 +25,7 @@
@@ -25,7 +25,7 @@
"express": "^4.21.2",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
"fast-xml-parser": "^4.5.3",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",
- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.39.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.0",
"pg": "^8.23.0"
"pg": "^8.23.0"
},
},
@@ -31,6 +31,7 @@ async function extraErase(t, subjects, counts) {
@@ -31,6 +31,7 @@ async function extraErase(t, subjects, counts) {
const mine = '(SELECT id FROM watches WHERE owner_sub = ANY($1::text[]))';
const mine = '(SELECT id FROM watches WHERE owner_sub = ANY($1::text[]))';
counts.add(counts.erased, 'observations', await t.exec(`DELETE FROM observations WHERE watch_id IN ${mine}`, [subjects]));
counts.add(counts.erased, 'observations', await t.exec(`DELETE FROM observations WHERE watch_id IN ${mine}`, [subjects]));
counts.add(counts.erased, 'check_runs', await t.exec(`DELETE FROM check_runs WHERE watch_id IN ${mine}`, [subjects]));
counts.add(counts.erased, 'check_runs', await t.exec(`DELETE FROM check_runs WHERE watch_id IN ${mine}`, [subjects]));
+ counts.add(counts.erased, 'action_runs', await t.exec(`DELETE FROM action_runs WHERE watch_id IN ${mine}`, [subjects]));
counts.add(counts.erased, 'watch_endpoint_state', await t.exec(`DELETE FROM watch_endpoint_state WHERE watch_id IN ${mine}`, [subjects]));
counts.add(counts.erased, 'watch_endpoint_state', await t.exec(`DELETE FROM watch_endpoint_state WHERE watch_id IN ${mine}`, [subjects]));
counts.add(counts.erased, 'watches', await t.exec('DELETE FROM watches WHERE owner_sub = ANY($1::text[])', [subjects]));
counts.add(counts.erased, 'watches', await t.exec('DELETE FROM watches WHERE owner_sub = ANY($1::text[])', [subjects]));
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,173 @@
@@ -0,0 +1,173 @@
+'use strict';
+/**
+ * Watch actions (plan T18 step 4): what a fired condition sets in motion.
+ *
+ * const actions = createActions({ db, config, notifications, now, log });
+ * await actions.queue({ watch, run, owner, value, previousValue, observedAt }) inside settle()'s transaction
+ * actions.start(); actions.stop(); await actions.tick(); await actions.status(); await actions.recent(watchId)
+ *
+ * queue() writes one action_runs row per action of the watch, in the SAME transaction as the check run that fired, so
+ * a trigger queues each of its actions exactly once (UNIQUE (check_run_id, action_index)) and never without its run.
+ *
+ * Delivery, by kind:
+ * notification the owner gets one OpenVibe notification through Network (openvibe-sdk/notifications, type
+ * WATCH_TRIGGERED): the watch's name, the value and a link to the watch page. A person who switched
+ * the category off is Network's decision and counts as sent; an owner Network does not know (a
+ * deleted account) is `skipped`. A refusal retries after 1, 5 and 30 minutes, 2 and 6 hours, then
+ * `failed`. Network keeps no idempotency record for a push, so a lost answer can repeat one: at least
+ * once. A watch owned by an agent or a service has no one to notify: `skipped`.
+ * event, webhook, actor-task, codes-run, run-job
+ * `unavailable` with `watch.action_unavailable`: their dispatchers are later steps of the plan, and
+ * the watch page says so rather than pretending they ran.
+ *
+ * The worker claims due rows with a short lease (FOR UPDATE SKIP LOCKED, next_attempt_at moved past the lease), so
+ * two processes never deliver one row at the same time, and a process that dies mid-delivery leaves the row to the
+ * next tick after the lease. Without the Network OAuth client secret there is no token to push with: rows wait and
+ * status() says delivery is off.
+ */
+
+const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+const BACKOFF_MS = [60_000, 5 * 60_000, 30 * 60_000, 2 * 3600_000, 6 * 3600_000];
+const LEASE_MS = 2 * 60_000;
+const TITLE_MAX = 120;
+const MESSAGE_MAX = 300;
+const LATER = new Set(['event', 'webhook', 'actor-task', 'codes-run', 'run-job']);
+
+const clip = (s, n) => { const t = String(s == null ? '' : s).replace(/\s+/g, ' ').trim(); return t.length > n ? `${t.slice(0, n - 1)}…` : t; };
+
+/** A value as a person reads it in one line. */
+function valueText(v) {
+ if (v == null) return 'no value';
+ if (typeof v === 'string') return v;
+ if (typeof v === 'number' || typeof v === 'boolean') return String(v);
+ try { return JSON.stringify(v); } catch { return String(v); }
+}
+
+/** The notification a fired watch sends its owner (network.notification-push-request@1 fields). */
+function notificationFor({ config, watch, run, action, value, previousValue, observedAt }) {
+ const now = valueText(value);
+ const before = previousValue === undefined || previousValue === null ? null : valueText(previousValue);
+ return {
+ type: 'WATCH_TRIGGERED',
+ service: 'watch',
+ category: action.category || 'service',
+ priority: action.priority || 'normal',
+ icon: '👁',
+ title: clip(`“${watch.name}” fired`, TITLE_MAX),
+ message: clip(before !== null && before !== now ? `${now} (was ${before})` : now, MESSAGE_MAX),
+ url: `${config.baseUrl}/watches/${encodeURIComponent(watch.id)}`,
+ rich_content: { watch_id: watch.id, check_run_id: run.id, observed_at: observedAt ? new Date(observedAt).toISOString() : null },
+ };
+}
+
+function createActions({ db, config, notifications = null, now = () => Date.now(), log = console }) {
+ const enabled = Boolean(notifications) && config.actions.enabled;
+ let timer = null;
+ let busy = null;
+
+ const st = {
+ insert: db.prepare(`INSERT INTO action_runs (watch_id, check_run_id, action_index, kind, owner_sub, request, state, last_error, created_at, finished_at)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (check_run_id, action_index) DO NOTHING`),
+ claim: db.prepare(`SELECT rid, watch_id, kind, owner_sub, request, attempts FROM action_runs
+ WHERE state = 'queued' AND next_attempt_at <= ? ORDER BY next_attempt_at, rid LIMIT ? FOR UPDATE SKIP LOCKED`),
+ lease: db.prepare('UPDATE action_runs SET attempts = attempts + 1, next_attempt_at = ? WHERE rid = ?'),
+ finish: db.prepare('UPDATE action_runs SET state = ?, last_error = ?, finished_at = ? WHERE rid = ?'),
+ retry: db.prepare('UPDATE action_runs SET next_attempt_at = ?, last_error = ? WHERE rid = ?'),
+ counts: db.prepare(`SELECT COUNT(*) FILTER (WHERE state = 'queued') AS queued, COUNT(*) FILTER (WHERE state = 'failed') AS failed,
+ COUNT(*) FILTER (WHERE state = 'sent') AS sent FROM action_runs`),
+ recent: db.prepare(`SELECT check_run_id, action_index, kind, state, attempts, last_error, created_at, finished_at FROM action_runs
+ WHERE watch_id = ? ORDER BY rid DESC LIMIT ?`),
+ };
+
+ /** One row per action of the watch, for the run that fired. Call inside the run's transaction. */
+ async function queue({ watch, run, owner, value, previousValue, observedAt }) {
+ const list = Array.isArray(watch.action) ? watch.action : [];
+ const at = now();
+ for (let i = 0; i < list.length; i++) {
+ const action = list[i] || {};
+ const kind = String(action.kind || '');
+ let state = 'queued', error = null, request = {};
+ if (kind === 'notification') {
+ if (USER_SUBJECT.test(String(owner || ''))) request = notificationFor({ config, watch, run, action, value, previousValue, observedAt });
+ else { state = 'skipped'; error = 'watch.action_no_person: the watch belongs to an agent or a service, not a person'; }
+ } else if (LATER.has(kind)) {
+ state = 'unavailable'; error = `watch.action_unavailable: ${kind} actions are a later step (plan T18)`;
+ } else {
+ state = 'unavailable'; error = `watch.action_unavailable: unknown action kind ${kind || '(none)'}`;
+ }
+ await st.insert.run(watch.id, run.id, i, kind || 'unknown', String(owner || ''), JSON.stringify(request), state, error, at, state === 'queued' ? null : at);
+ }
+ }
+
+ async function deliver(row) {
+ const request = typeof row.request === 'string' ? JSON.parse(row.request) : row.request;
+ if (row.kind !== 'notification') return { state: 'unavailable', error: `watch.action_unavailable: ${row.kind}` };
+ const out = await notifications.push({ subjectId: row.owner_sub, ...request });
+ if (out.sent) return { state: 'sent', error: out.skipped ? 'delivered; the person switched this category off' : null };
+ return { state: 'skipped', error: 'watch.action_owner_unknown: Network does not know this owner (a deleted account?)' };
+ }
+
+ /** One pass: claim due rows under a lease, deliver each, record the outcome. Returns counts. */
+ async function tick() {
+ if (!enabled) return { enabled: false };
+ if (busy) return await busy;
+ busy = (async () => {
+ const out = { sent: 0, skipped: 0, retried: 0, failed: 0 };
+ const at = now();
+ const rows = await db.tx(async () => {
+ const due = await st.claim.all(at, config.actions.batch);
+ for (const r of due) await st.lease.run(at + LEASE_MS, r.rid);
+ return due;
+ });
+ for (const r of rows) {
+ const attempt = Number(r.attempts) + 1;
+ try {
+ const res = await deliver(r);
+ await st.finish.run(res.state, res.error, now(), r.rid);
+ if (res.state === 'sent') out.sent++; else out.skipped++;
+ } catch (err) {
+ const why = String((err && (err.code || err.message)) || err).slice(0, 300);
+ if (attempt >= BACKOFF_MS.length + 1) {
+ await st.finish.run('failed', why, now(), r.rid);
+ out.failed++;
+ log.warn(`[actions] ${r.watch_id}: ${r.kind} gave up after ${attempt} attempts: ${why}`);
+ } else {
+ await st.retry.run(now() + BACKOFF_MS[attempt - 1], why, r.rid);
+ out.retried++;
+ }
+ }
+ }
+ return out;
+ })();
+ try { return await busy; } finally { busy = null; }
+ }
+
+ function start() {
+ if (timer || !enabled) return;
+ timer = setInterval(() => { tick().catch((err) => log.warn(`[actions] tick: ${err.message}`)); }, config.actions.tickMs);
+ timer.unref?.();
+ }
+ async function stop() {
+ if (timer) clearInterval(timer);
+ timer = null;
+ if (busy) await busy.catch(() => {});
+ }
+
+ async function status() {
+ const c = await st.counts.get();
+ return { enabled, queued: Number(c.queued) || 0, failed: Number(c.failed) || 0, sent: Number(c.sent) || 0 };
+ }
+
+ async function recent(watchId, limit = 10) {
+ return (await st.recent.all(watchId, limit)).map((r) => ({
+ check_run_id: r.check_run_id, kind: r.kind, state: r.state, attempts: Number(r.attempts),
+ error: r.last_error || null,
+ created_at: new Date(Number(r.created_at)).toISOString(),
+ finished_at: r.finished_at == null ? null : new Date(Number(r.finished_at)).toISOString(),
+ }));
+ }
+
+ return { queue, tick, start, stop, status, recent, enabled };
+}
+
+module.exports = { createActions, notificationFor, BACKOFF_MS };
@@ -58,7 +58,7 @@ const CSP = "default-src 'self'; script-src 'self' 'unsafe-inline' https://openv
@@ -58,7 +58,7 @@ const CSP = "default-src 'self'; script-src 'self' 'unsafe-inline' https://openv
+ "connect-src 'self' https://openvibe.network https://openvibe.events https://cloudflareinsights.com; "
+ "connect-src 'self' https://openvibe.network https://openvibe.events https://cloudflareinsights.com; "
+ "frame-src 'self' https://openvibe.network; frame-ancestors 'none'; object-src 'none'; base-uri 'self'; form-action 'self' https://openvibe.network";
+ "frame-src 'self' https://openvibe.network; frame-ancestors 'none'; object-src 'none'; base-uri 'self'; form-action 'self' https://openvibe.network";
-function createApp({ config, db, registry, check, observations, scheduler, auth, outbox, now, log = console, limitsNow = null, sessions, keyStore, siteLimits, fetchImpl = globalThis.fetch, accountData = null, accountSend = null }) {
+function createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions = null, now, log = console, limitsNow = null, sessions, keyStore, siteLimits, fetchImpl = globalThis.fetch, accountData = null, accountSend = null }) {
const app = express();
const app = express();
app.disable('x-powered-by');
app.disable('x-powered-by');
app.set('trust proxy', config.trustProxy != null ? config.trustProxy : 'loopback');
app.set('trust proxy', config.trustProxy != null ? config.trustProxy : 'loopback');
@@ -94,7 +94,7 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
@@ -94,7 +94,7 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
// Readiness (openvibe-shared/ready): 503 only when the database fails; the Network key and the
// Readiness (openvibe-shared/ready): 503 only when the database fails; the Network key and the
// check worker (running, queue keeping up) are optional and degrade it (see observability.js).
// check worker (running, queue keeping up) are optional and degrade it (see observability.js).
- const readiness = createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, now, release: release.release });
+ const readiness = createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, actions, now, release: release.release });
app.get('/api/ready', readiness.handler);
app.get('/api/ready', readiness.handler);
// GET /release.json (ADR-016) and POST /release-metrics (open tabs' update reports into /metrics).
// GET /release.json (ADR-016) and POST /release-metrics (open tabs' update reports into /metrics).
release.mount(app, { registry: metrics.registry });
release.mount(app, { registry: metrics.registry });
@@ -139,7 +139,7 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
@@ -139,7 +139,7 @@ function createApp({ config, db, registry, check, observations, scheduler, auth,
if (!req.path.startsWith('/api') && !req.path.startsWith('/auth')) res.setHeader('Content-Security-Policy', CSP);
if (!req.path.startsWith('/api') && !req.path.startsWith('/auth')) res.setHeader('Content-Security-Policy', CSP);
next();
next();
});
});
- app.use(createPageRoutes({ config, db, registry, check, observations, sessions, siteLimits, log }));
+ app.use(createPageRoutes({ config, db, registry, check, observations, sessions, siteLimits, actions, log }));
app.use((req, res) => send(res, 404, {
app.use((req, res) => send(res, 404, {
viewer: req.viewer, config, path: req.originalUrl, title: 'Not found',
viewer: req.viewer, config, path: req.originalUrl, title: 'Not found',
@@ -12,7 +12,7 @@
@@ -12,7 +12,7 @@
* updated with what the site said (ETag, Last-Modified, status, body hash)
* updated with what the site said (ETag, Last-Modified, status, body hash)
* 5 extract → value; record an observation (value, hashes, changed, condition, retention)
* 5 extract → value; record an observation (value, hashes, changed, condition, retention)
* 6 evaluate the condition; a condition that MET writes check_runs(triggers = 1, state
* 6 evaluate the condition; a condition that MET writes check_runs(triggers = 1, state
- * 'condition_met') and emits watch.watch.triggered. Actions are step 4 of the plan and are
+ * 'condition_met'), emits watch.watch.triggered and queues the watch's actions (server/actions.js). Formerly
* NOT dispatched here: a met condition records the trigger and emits the event, nothing else
* NOT dispatched here: a met condition records the trigger and emits the event, nothing else
* 7 arm next_due_at (+ jitter, or exponential backoff after a failure, capped) and not_before
* 7 arm next_due_at (+ jitter, or exponential backoff after a failure, capped) and not_before
* (Retry-After)
* (Retry-After)
@@ -34,7 +34,7 @@ const FAILURE_STATES = new Set(['http_error', 'timeout', 'parse_error', 'rate_li
@@ -34,7 +34,7 @@ const FAILURE_STATES = new Set(['http_error', 'timeout', 'parse_error', 'rate_li
const PERSON_OWNER = /^(usr|agt)_[0-9A-HJKMNP-TV-Z]{26}$/;
const PERSON_OWNER = /^(usr|agt)_[0-9A-HJKMNP-TV-Z]{26}$/;
const USER_OWNER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const USER_OWNER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
-function createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, now = () => Date.now(), log = console, ids = contractIds }) {
+function createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions = null, now = () => Date.now(), log = console, ids = contractIds }) {
const inflight = new Set();
const inflight = new Set();
const st = {
const st = {
@@ -55,7 +55,7 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
@@ -55,7 +55,7 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
run: db.prepare('SELECT * FROM check_runs WHERE id = ?'),
run: db.prepare('SELECT * FROM check_runs WHERE id = ?'),
// The last check that fired the condition: a repeat needs a value that is new or a new run
// The last check that fired the condition: a repeat needs a value that is new or a new run
// of holding, so a satisfied watch does not fire once per check while nothing moves.
// of holding, so a satisfied watch does not fire once per check while nothing moves.
- lastTrigger: db.prepare('SELECT id, started_at FROM check_runs WHERE watch_id = ? AND triggers > 0 ORDER BY rid DESC LIMIT 1'),
+ lastTrigger: db.prepare('SELECT id, started_at, finished_at FROM check_runs WHERE watch_id = ? AND triggers > 0 ORDER BY rid DESC LIMIT 1'),
};
};
const boolOf = (v) => Boolean(Number(v));
const boolOf = (v) => Boolean(Number(v));
@@ -161,7 +161,7 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
@@ -161,7 +161,7 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
* the events and the watch's re-armed runtime state all commit together, or none of them do.
* the events and the watch's re-armed runtime state all commit together, or none of them do.
* `failures` is the consecutive failure count (0 when the check succeeded).
* `failures` is the consecutive failure count (0 when the check succeeded).
*/
*/
- async function settle({ watch, run, failures, successAt = null, notBefore = 0, events = [], observation = null, endpoint = null }) {
+ async function settle({ watch, run, failures, successAt = null, notBefore = 0, events = [], observation = null, endpoint = null, fired = null }) {
const armed = nextDue(watch, run.finished_at, { failures, maxBackoffMs: config.worker.maxBackoffMs });
const armed = nextDue(watch, run.finished_at, { failures, maxBackoffMs: config.worker.maxBackoffMs });
await db.tx(async () => {
await db.tx(async () => {
if (observation) await observations.insert(observation);
if (observation) await observations.insert(observation);
@@ -172,6 +172,8 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
@@ -172,6 +172,8 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
failures, next_due: armed, not_before: notBefore,
failures, next_due: armed, not_before: notBefore,
});
});
for (const e of events || []) await outbox.emit(envelope(e, config.serviceId, now));
for (const e of events || []) await outbox.emit(envelope(e, config.serviceId, now));
+ // A run that fired queues the watch's actions in this same commit (server/actions.js): once per trigger.
+ if (fired && actions) await actions.queue({ watch, run, ...fired });
});
});
outbox.kick().catch(() => { });
outbox.kick().catch(() => { });
return { run, observation, met: run.triggers > 0, failures };
return { run, observation, met: run.triggers > 0, failures };
@@ -312,7 +314,10 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
@@ -312,7 +314,10 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
// value that is new (changed) or for a new run of holding (the trail began after the last
// value that is new (changed) or for a new run of holding (the trail began after the last
// firing) — never once per check while nothing moves.
// firing) — never once per check while nothing moves.
const lastFired = await st.lastTrigger.get(watch.id);
const lastFired = await st.lastTrigger.get(watch.id);
- const repeat = verdict.met && !verdict.changed && lastFired != null && verdict.since != null && lastFired.started_at >= verdict.since;
+ // The trail's `since` is the observed_at of its first holding observation, which the run that recorded it stamps
+ // with its finish time; the firing run's start is always earlier, so the last firing is compared by when it
+ // finished (comparing its start made every check while the condition held fire again).
+ const repeat = verdict.met && !verdict.changed && lastFired != null && verdict.since != null && Number(lastFired.finished_at) >= verdict.since;
const met = verdict.met && !repeat;
const met = verdict.met && !repeat;
if (verdict.error && !patternError) patternError = verdict.error;
if (verdict.error && !patternError) patternError = verdict.error;
@@ -353,7 +358,8 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
@@ -353,7 +358,8 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
watch, owner: row.owner_sub, observation, value, met, run,
watch, owner: row.owner_sub, observation, value, met, run,
firedOn, firedAt: finishedAt, previousValue, failures: 0,
firedOn, firedAt: finishedAt, previousValue, failures: 0,
});
});
- const result = await settle({ watch, run, failures: 0, successAt: finishedAt, events, observation, endpoint });
+ const fired = met ? { owner: row.owner_sub, value, previousValue, observedAt: firedOn ? Number(firedOn.observed_at) : finishedAt } : null;
+ const result = await settle({ watch, run, failures: 0, successAt: finishedAt, events, observation, endpoint, fired });
return {
return {
...result, met, value, verdict: { ...verdict, met, repeat },
...result, met, value, verdict: { ...verdict, met, repeat },
observation: observation ? observations.view(observation) : null,
observation: observation ? observations.view(observation) : null,
@@ -79,6 +79,14 @@ function load(env = process.env) {
@@ -79,6 +79,14 @@ function load(env = process.env) {
secrets: String(env.WATCH_EVENTS_SECRET || '').split(',').map((x) => x.trim()).filter(Boolean),
secrets: String(env.WATCH_EVENTS_SECRET || '').split(',').map((x) => x.trim()).filter(Boolean),
},
},
+ // Actions (server/actions.js): every WATCH_ACTIONS_TICK_MS deliver up to WATCH_ACTIONS_BATCH queued actions of fired
+ // watches. Notifications need the Network OAuth client secret (a token for network.notifications.push).
+ actions: {
+ enabled: env.WATCH_ACTIONS !== 'off',
+ tickMs: int(env.WATCH_ACTIONS_TICK_MS, 5000),
+ batch: int(env.WATCH_ACTIONS_BATCH, 20),
+ },
+
// The check worker: every WATCH_TICK_MS, start checks for active watches whose turn has come.
// The check worker: every WATCH_TICK_MS, start checks for active watches whose turn has come.
worker: {
worker: {
enabled: env.WATCH_WORKER !== 'off',
enabled: env.WATCH_WORKER !== 'off',
@@ -74,9 +74,8 @@ const OPS = {
@@ -74,9 +74,8 @@ const OPS = {
};
};
const NEEDS_VALUE = new Set(['gt', 'gte', 'lt', 'lte', 'eq', 'ne', 'contains', 'matches']);
const NEEDS_VALUE = new Set(['gt', 'gte', 'lt', 'lte', 'eq', 'ne', 'contains', 'matches']);
const NOTIFY_ACTION = [{ kind: 'notification', category: 'service', priority: 'normal' }];
const NOTIFY_ACTION = [{ kind: 'notification', category: 'service', priority: 'normal' }];
-// With notifications off the watch still needs an action (the contract requires one); actions are
-// step 4 of the plan and none is dispatched yet, so this records the person's intent without changing
-// what happens today (see how-it-works and the note on the form).
+// With notifications off the watch still needs an action (the contract requires one): an event, whose dispatcher is a
+// later step, so a quiet watch records its trigger and tells nobody (server/actions.js records it as unavailable).
const QUIET_ACTION = [{ kind: 'event', event_type: 'watch.watch.triggered' }];
const QUIET_ACTION = [{ kind: 'event', event_type: 'watch.watch.triggered' }];
const INT = (v, d) => { const n = parseInt(v, 10); return Number.isFinite(n) ? n : d; };
const INT = (v, d) => { const n = parseInt(v, 10); return Number.isFinite(n) ? n : d; };
@@ -92,7 +91,7 @@ function sameOrigin(req, baseUrl) {
@@ -92,7 +91,7 @@ function sameOrigin(req, baseUrl) {
}
}
function createPageRoutes(ctx) {
function createPageRoutes(ctx) {
- const { config, db, registry, check, observations, sessions, log = console } = ctx;
+ const { config, db, registry, check, observations, sessions, actions = null, log = console } = ctx;
const r = express.Router();
const r = express.Router();
const formBody = express.urlencoded({ extended: false, limit: '32kb' });
const formBody = express.urlencoded({ extended: false, limit: '32kb' });
@@ -323,10 +322,10 @@ ${extractionBlock}
@@ -323,10 +322,10 @@ ${extractionBlock}
${conditionBlock}
${conditionBlock}
<fieldset><legend>How often, and how to tell you</legend>
<fieldset><legend>How often, and how to tell you</legend>
${field('Cadence', cadence, { name: 'cadence', error: errors.cadence, hint: 'Watch never checks more often than this, and can wait longer when the site asks it to.' })}
${field('Cadence', cadence, { name: 'cadence', error: errors.cadence, hint: 'Watch never checks more often than this, and can wait longer when the site asks it to.' })}
-${checkbox('Notify me when it fires', { name: 'notify', checked: fields.notify === 'on', hint: 'A fired condition becomes a WATCH_TRIGGERED notification through OpenVibe.Network, as today.' })}
+${checkbox('Notify me when it fires', { name: 'notify', checked: fields.notify === 'on', hint: 'When the condition fires you get one notification on OpenVibe (your notifications on any OpenVibe site), with the value and a link here.' })}
</fieldset>
</fieldset>
<p class="form-actions"><button class="btn primary" type="submit">${submit}</button> <a href="/watches">Cancel</a></p>
<p class="form-actions"><button class="btn primary" type="submit">${submit}</button> <a href="/watches">Cancel</a></p>
-<p class="muted small">Actions are step 4 of the plan: a fired condition records its trigger and emits <code>watch.watch.triggered</code> today, and no other action is dispatched yet — turning notifications off does not silence that until actions run.</p>
+<p class="muted small">Notifications are delivered today. Events, webhooks and tasks for OpenVibe.Actor, Codes and Run are later steps: a watch that asks for one records it as not available yet.</p>
</form>`;
</form>`;
}
}
@@ -517,6 +516,7 @@ ${watchForm({ template, fields, errors, action: '/watches/new', submit: 'Create
@@ -517,6 +516,7 @@ ${watchForm({ template, fields, errors, action: '/watches/new', submit: 'Create
const latest = await observations.latest(row.id);
const latest = await observations.latest(row.id);
const obs = await observations.list(row.id, { before: ridBefore(req.query.obs_before), limit: obsLimit(req.query.obs_limit) });
const obs = await observations.list(row.id, { before: ridBefore(req.query.obs_before), limit: obsLimit(req.query.obs_limit) });
const runs = await check.list(row.id, { before: ridBefore(req.query.checks_before), limit: obsLimit(req.query.checks_limit) });
const runs = await check.list(row.id, { before: ridBefore(req.query.checks_before), limit: obsLimit(req.query.checks_limit) });
+ const acts = actions ? await actions.recent(row.id, 10) : [];
const obsRows = obs.map((o) => [
const obsRows = obs.map((o) => [
time(new Date(Number(o.observed_at)).toISOString()),
time(new Date(Number(o.observed_at)).toISOString()),
@@ -536,6 +536,14 @@ ${watchForm({ template, fields, errors, action: '/watches/new', submit: 'Create
@@ -536,6 +536,14 @@ ${watchForm({ template, fields, errors, action: '/watches/new', submit: 'Create
];
];
});
});
+ const ACT_WORDS = { queued: ['waiting to send', 'info'], sent: ['sent', 'ok'], skipped: ['not sent', 'warn'], failed: ['failed', 'bad'], unavailable: ['not available yet', 'warn'] };
+ const actRows = acts.map((a) => [
+ time(a.created_at),
+ a.kind === 'notification' ? 'Notification' : html`<code>${a.kind}</code>`,
+ badge((ACT_WORDS[a.state] || [a.state])[0], (ACT_WORDS[a.state] || [null, 'info'])[1]),
+ a.error ? html`<span class="muted small">${a.error}</span>` : (a.attempts > 1 ? html`<span class="muted small">${a.attempts} attempts</span>` : ''),
+ ]);
+
const ran = String(req.query.ran || '');
const ran = String(req.query.ran || '');
const actionForms = html`<div class="actions">
const actionForms = html`<div class="actions">
${v.status === 'active'
${v.status === 'active'
@@ -567,6 +575,8 @@ ${actionForms}
@@ -567,6 +575,8 @@ ${actionForms}
<section aria-labelledby="h-obs"><h2 id="h-obs">Observation history</h2>
<section aria-labelledby="h-obs"><h2 id="h-obs">Observation history</h2>
${table(['When', 'Value', 'Changed', 'Condition'], obsRows, { empty: 'No observations yet. Watch records one when a check reads a value.' })}
${table(['When', 'Value', 'Changed', 'Condition'], obsRows, { empty: 'No observations yet. Watch records one when a check reads a value.' })}
<nav class="pager">${nextObs ? html`<a class="btn" href="${nextObs}">Older observations</a>` : html`<span class="muted">That is every observation kept.</span>`}</nav></section>
<nav class="pager">${nextObs ? html`<a class="btn" href="${nextObs}">Older observations</a>` : html`<span class="muted">That is every observation kept.</span>`}</nav></section>
+<section aria-labelledby="h-acts"><h2 id="h-acts">What it did when it fired</h2>
+${table(['When', 'Action', 'Result', 'Detail'], actRows, { empty: 'Nothing yet: the condition has not fired since actions began (2026-10-10).' })}</section>
<section aria-labelledby="h-runs"><h2 id="h-runs">Check runs</h2>
<section aria-labelledby="h-runs"><h2 id="h-runs">Check runs</h2>
<p class="muted small">Every check, whatever it found. A skipped kind, a site error and a wait are recorded states, not silence.</p>
<p class="muted small">Every check, whatever it found. A skipped kind, a site error and a wait are recorded states, not silence.</p>
${table(['When', 'State', 'HTTP', 'Detail', 'Read', 'Fired'], runRows, { empty: 'No checks yet.' })}
${table(['When', 'State', 'HTTP', 'Detail', 'Read', 'Fired'], runRows, { empty: 'No checks yet.' })}
@@ -23,7 +23,10 @@ const { createCheck } = require('./check');
@@ -23,7 +23,10 @@ const { createCheck } = require('./check');
const { createScheduler } = require('./scheduler');
const { createScheduler } = require('./scheduler');
const { createServiceOutbox } = require('openvibe-sdk/events');
const { createServiceOutbox } = require('openvibe-sdk/events');
const { validate } = require('openvibe-contracts');
const { validate } = require('openvibe-contracts');
-const { jwksClient } = require('openvibe-sdk/auth');
+const { jwksClient, createServiceTokenClient } = require('openvibe-sdk/auth');
+const { createClient } = require('openvibe-sdk/core');
+const { createNotificationsClient } = require('openvibe-sdk/notifications');
+const { createActions } = require('./actions');
const { gracefulStop } = require('openvibe-sdk/service');
const { gracefulStop } = require('openvibe-sdk/service');
const { createAuth } = require('./auth');
const { createAuth } = require('./auth');
const { createSessions } = require('./sessions');
const { createSessions } = require('./sessions');
@@ -33,7 +36,7 @@ const { createApp } = require('./app');
@@ -33,7 +36,7 @@ const { createApp } = require('./app');
const accountDataLib = require('./account-data');
const accountDataLib = require('./account-data');
const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');
const { createNetworkSender, startSubscriptions } = require('openvibe-sdk/account-data');
-async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null } = {}) {
+async function start({ config, db: givenDb = null, now = () => Date.now(), fetchImpl = globalThis.fetch, lookupImpl, log = console, listen = true, limitsNow = null, accountSend: givenSend = null, notifications: givenNotifications = null } = {}) {
config = config || load();
config = config || load();
// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
// PostgreSQL (ADR-035): opened and migrated here unless the caller (a test) hands in a migrated handle.
const db = givenDb || await openDb(config, { log });
const db = givenDb || await openDb(config, { log });
@@ -54,7 +57,16 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -54,7 +57,16 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
const spacer = createSpacer({ hostMinIntervalMs: config.fetch.hostMinIntervalMs });
const spacer = createSpacer({ hostMinIntervalMs: config.fetch.hostMinIntervalMs });
// Every carrier goes out through the spacer, so two watches over one host never hammer it.
// Every carrier goes out through the spacer, so two watches over one host never hammer it.
const carriers = createCarriers({ fetcher: spacedFetcher(fetcher, spacer), config, log });
const carriers = createCarriers({ fetcher: spacedFetcher(fetcher, spacer), config, log });
- const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, now, log });
+ // Actions of fired watches (plan T18 step 4): notifications go to the owner through Network with Watch's own
+ // client-credentials token; without the client secret they wait (actions.status() says delivery is off).
+ const notifications = config.oauth.clientSecret
+ ? createNotificationsClient(createClient({
+ tokenProvider: createServiceTokenClient({ network: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl }),
+ baseUrls: { network: config.networkInternalUrl }, fetch: fetchImpl,
+ }))
+ : null;
+ const actions = createActions({ db, config, notifications: givenNotifications || notifications, now, log });
+ const check = createCheck({ db, registry, config, carriers, extract, condition, observations, outbox, actions, now, log });
const scheduler = createScheduler({ db, check, config, now, log });
const scheduler = createScheduler({ db, check, config, now, log });
const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });
const keys = jwksClient(config.jwksUrl, { fetch: fetchImpl, log });
const auth = createAuth({ config, log });
const auth = createAuth({ config, log });
@@ -69,13 +81,14 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -69,13 +81,14 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
const accountSend = givenSend || (config.oauth.clientSecret
const accountSend = givenSend || (config.oauth.clientSecret
? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl })
: null);
: null);
- const app = createApp({ config, db, registry, check, observations, scheduler, auth, outbox, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend });
+ const app = createApp({ config, db, registry, check, observations, scheduler, auth, outbox, actions, now, log, limitsNow, sessions, keyStore, siteLimits, accountData, accountSend });
// One JWKS client for the process (the SDK shares it with verifyUserToken): refresh in the background
// One JWKS client for the process (the SDK shares it with verifyUserToken): refresh in the background
// on an unref'd timer, keeping the last good keys through a Network outage.
// on an unref'd timer, keeping the last good keys through a Network outage.
keys.start();
keys.start();
const keyLoaded = keys.keys().catch(() => null);
const keyLoaded = keys.keys().catch(() => null);
outbox.start();
outbox.start();
if (config.worker.enabled) scheduler.start();
if (config.worker.enabled) scheduler.start();
+ actions.start();
const pruneTimer = setInterval(async () => {
const pruneTimer = setInterval(async () => {
try { await observations.prune(); } catch (err) { log.error(`[observations] prune: ${err.message}`); }
try { await observations.prune(); } catch (err) { log.error(`[observations] prune: ${err.message}`); }
try { await sessions.prune(); } catch (err) { log.error(`[sessions] prune: ${err.message}`); }
try { await sessions.prune(); } catch (err) { log.error(`[sessions] prune: ${err.message}`); }
@@ -105,6 +118,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -105,6 +118,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
if (subscriptions) subscriptions.stop();
if (subscriptions) subscriptions.stop();
keys.stop();
keys.stop();
await scheduler.stop();
await scheduler.stop();
+ await actions.stop();
await outbox.stop();
await outbox.stop();
if (server) {
if (server) {
server.closeAllConnections?.();
server.closeAllConnections?.();
@@ -113,7 +127,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -113,7 +127,7 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
if (!givenDb) await db.close();
if (!givenDb) await db.close();
}
}
- return { config, db, registry, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };
+ return { config, db, registry, observations, guard, fetcher, spacer, carriers, extract, condition, check, scheduler, outbox, actions, keys, keyLoaded, auth, sessions, keyStore, siteLimits, accountData, app, server, close };
}
}
/**
/**
@@ -48,7 +48,7 @@ function readers(db, { startedAt }) {
@@ -48,7 +48,7 @@ function readers(db, { startedAt }) {
};
};
}
}
-function createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, now, release = null }) {
+function createWatchReadiness({ db, config, registry, check, observations, scheduler, outbox, actions = null, now, release = null }) {
const read = readers(db, { startedAt: now() });
const read = readers(db, { startedAt: now() });
const iso = (v) => (v == null ? null : new Date(v).toISOString());
const iso = (v) => (v == null ? null : new Date(v).toISOString());
return createReadiness({
return createReadiness({
@@ -98,6 +98,7 @@ function createWatchReadiness({ db, config, registry, check, observations, sched
@@ -98,6 +98,7 @@ function createWatchReadiness({ db, config, registry, check, observations, sched
checks_in_flight: check.inflight().length,
checks_in_flight: check.inflight().length,
observations: dbOk ? await read.observations() : null,
observations: dbOk ? await read.observations() : null,
outbox: dbOk ? await outbox.status() : null,
outbox: dbOk ? await outbox.status() : null,
+ actions: dbOk && actions ? await actions.status() : null,
};
};
},
},
});
});
new file mode 100644
new file mode 100644
@@ -0,0 +1,154 @@
@@ -0,0 +1,154 @@
+'use strict';
+/**
+ * Actions (plan T18 step 4): a condition that fires queues the watch's actions in the run's own transaction, once per
+ * trigger; the worker delivers a notification to the owner through Network (openvibe-sdk/notifications), retries a
+ * refusal with backoff, skips an owner Network does not know and an owner that is not a person, and records the kinds
+ * whose dispatchers are later steps as unavailable. The watch page says what happened.
+ */
+const assert = require('assert');
+const { boot, site, suite, newUser, newAgent } = require('./helpers');
+
+const t = suite('actions');
+let svc;
+let web;
+let price = '19.99';
+const pushes = [];
+let mode = 'ok';
+const notifications = {
+ async push(req) {
+ pushes.push(req);
+ if (mode === 'fail') { const e = new Error('Network answered 503'); e.status = 503; e.code = 'notifications.unavailable'; throw e; }
+ if (mode === 'unknown') return { sent: false, reason: 'unknown_subject' };
+ return { sent: true, skipped: mode === 'muted' };
+ },
+};
+
+const person = { sub: newUser() };
+const agent = { sub: newAgent() };
+const def = (overrides = {}) => ({
+ name: 'Widget price',
+ source: { kind: 'http', url: `${web.origin}/price`, format: null },
+ cadence: { every_sec: 900 },
+ extraction: { kind: 'text' },
+ condition: { op: 'lt', value: 10 },
+ action: [{ kind: 'notification', priority: 'high' }],
+ ...overrides,
+});
+const create = async (principal, overrides = {}) => await svc.registry.create(def(overrides), principal);
+const check = async (id) => await svc.check.run(id, { trigger: 'schedule' });
+const acts = async (id) => await svc.db.prepare('SELECT * FROM action_runs WHERE watch_id = ? ORDER BY rid').all(id);
+const setPrice = async (v) => { await fetch(`${web.origin}/set?v=${v}`); };
+/** Fire a watch: one check at 19.99 (nothing), then the price drops below 10. */
+async function fire(w) {
+ await setPrice('19.99');
+ await check(w.id);
+ await setPrice('7.50');
+ const out = await check(w.id);
+ assert.strictEqual(out.run.state, 'condition_met');
+ return out;
+}
+
+t('boot', async () => {
+ web = await site({
+ '/price': () => ({ headers: { 'content-type': 'text/plain' }, body: price }),
+ '/set': (req, _res, ctx) => { price = ctx.url.searchParams.get('v') || price; return { body: 'ok' }; },
+ });
+ svc = await boot({ notifications });
+});
+
+t('a check that does not fire queues nothing; one that fires queues each action once, with its run', async () => {
+ mode = 'ok'; pushes.length = 0;
+ const w = await create(person, { action: [{ kind: 'notification', priority: 'high' }, { kind: 'webhook', url: 'https://hooks.example.org/w', secret_env: 'WATCH_HOOK_TEST' }] });
+ await setPrice('19.99');
+ await check(w.id);
+ assert.deepStrictEqual(await acts(w.id), [], 'no trigger, no action');
+ await setPrice('7.50');
+ const out = await check(w.id);
+ const rows = await acts(w.id);
+ assert.deepStrictEqual(rows.map((r) => [r.kind, r.state, r.check_run_id, Number(r.action_index)]),
+ [['notification', 'queued', out.run.id, 0], ['webhook', 'unavailable', out.run.id, 1]]);
+ assert.match(rows[1].last_error, /watch\.action_unavailable/);
+
+ const r = await svc.actions.tick();
+ assert.deepStrictEqual(r, { sent: 1, skipped: 0, retried: 0, failed: 0 });
+ assert.strictEqual(pushes.length, 1);
+ const p = pushes[0];
+ assert.strictEqual(p.subjectId, person.sub);
+ assert.deepStrictEqual([p.type, p.service, p.category, p.priority], ['WATCH_TRIGGERED', 'watch', 'service', 'high']);
+ assert.strictEqual(p.title, '“Widget price” fired');
+ assert.strictEqual(p.message, '7.50 (was 19.99)');
+ assert.strictEqual(p.url, `https://openvibe.watch/watches/${w.id}`);
+ assert.strictEqual(p.rich_content.check_run_id, out.run.id);
+ assert.strictEqual((await acts(w.id))[0].state, 'sent');
+
+ await svc.actions.tick();
+ assert.strictEqual(pushes.length, 1, 'a sent action is never sent again');
+ await check(w.id); // 7.50 still holds: no new trigger, no new action
+ assert.strictEqual((await acts(w.id)).length, 2);
+});
+
+t('a refusal waits with backoff; the next tick does not retry early; a later success sends it', async () => {
+ pushes.length = 0; mode = 'fail';
+ const w = await create(person);
+ await fire(w);
+ let r = await svc.actions.tick();
+ assert.deepStrictEqual(r, { sent: 0, skipped: 0, retried: 1, failed: 0 });
+ let [row] = await acts(w.id);
+ assert.strictEqual(row.state, 'queued');
+ assert.strictEqual(Number(row.attempts), 1);
+ assert.match(row.last_error, /notifications\.unavailable/);
+ assert.ok(Number(row.next_attempt_at) > Date.now() + 50_000, 'the first retry waits about a minute');
+ r = await svc.actions.tick();
+ assert.strictEqual(pushes.length, 1, 'not due yet: no second push');
+ mode = 'ok';
+ await svc.db.prepare('UPDATE action_runs SET next_attempt_at = 0 WHERE watch_id = ?').run(w.id);
+ r = await svc.actions.tick();
+ assert.deepStrictEqual(r, { sent: 1, skipped: 0, retried: 0, failed: 0 });
+ [row] = await acts(w.id);
+ assert.deepStrictEqual([row.state, Number(row.attempts)], ['sent', 2]);
+});
+
+t('an owner Network does not know is skipped; a muted category is delivered; an agent-owned watch notifies nobody', async () => {
+ pushes.length = 0; mode = 'unknown';
+ const gone = await create(person, { name: 'gone' });
+ await fire(gone);
+ await svc.actions.tick();
+ assert.strictEqual((await acts(gone.id))[0].state, 'skipped');
+ assert.match((await acts(gone.id))[0].last_error, /owner_unknown/);
+
+ mode = 'muted';
+ const muted = await create(person, { name: 'muted' });
+ await fire(muted);
+ await svc.actions.tick();
+ assert.strictEqual((await acts(muted.id))[0].state, 'sent', 'Network decided: the person switched the category off');
+
+ mode = 'ok';
+ const before = pushes.length;
+ const bot = await create(agent, { name: 'agent watch' });
+ await fire(bot);
+ const [row] = await acts(bot.id);
+ assert.deepStrictEqual([row.state, /action_no_person/.test(row.last_error)], ['skipped', true]);
+ await svc.actions.tick();
+ assert.strictEqual(pushes.length, before, 'nothing is pushed for an agent');
+});
+
+t('readiness counts the queue, and the watch page says what the watch did when it fired', async () => {
+ const st = await svc.actions.status();
+ assert.strictEqual(st.enabled, true);
+ assert.ok(st.sent >= 3 && st.queued === 0 && st.failed === 0, JSON.stringify(st));
+ const w = await create(person, { name: 'page check' });
+ mode = 'ok';
+ await fire(w);
+ await svc.actions.tick();
+ const cookie = await svc.signIn({ subject: person.sub, username: 'ana' });
+ const res = await fetch(`${svc.base}/watches/${w.id}`, { headers: { cookie } });
+ const body = await res.text();
+ assert.strictEqual(res.status, 200);
+ assert.match(body, /What it did when it fired/);
+ assert.match(body, /Notification/);
+ assert.match(body, /class="badge ok">sent</);
+});
+
+t('done', async () => { await web.close(); await svc.stop(); });
+
+t.run();
@@ -117,6 +117,11 @@ t('a condition that fires emits watch.watch.triggered with the observation that
@@ -117,6 +117,11 @@ t('a condition that fires emits watch.watch.triggered with the observation that
assert.deepStrictEqual([Number(last.changed), Number(last.condition_met)], [1, 1]);
assert.deepStrictEqual([Number(last.changed), Number(last.condition_met)], [1, 1]);
valid('watch.observation@1', svc.observations.view(last));
valid('watch.observation@1', svc.observations.view(last));
valid('watch.check-run@1', svc.check.runView((await runs(w.id)).pop()));
valid('watch.check-run@1', svc.check.runView((await runs(w.id)).pop()));
+
+ // While 7.50 keeps holding, the next checks do not fire again (the trail began with the run that fired).
+ const again = await check(w.id);
+ assert.deepStrictEqual([again.run.triggers, again.met], [0, false], 'a condition that keeps holding fires once');
+ assert.strictEqual((await envelopes('watch.watch.triggered')).filter((e) => e.payload.watch_id === w.id).length, 1);
});
});
t('a failing check is a recorded state and a watch.check.failed event, and it backs off', async () => {
t('a failing check is a recorded state and a watch.check.failed event, and it backs off', async () => {
@@ -49,7 +49,7 @@ const newAgent = () => require('openvibe-contracts').ids.newId('agent');
@@ -49,7 +49,7 @@ const newAgent = () => require('openvibe-contracts').ids.newId('agent');
* the caller says otherwise (a test of the SSRF guard boots with allowPrivate: false).
* the caller says otherwise (a test of the SSRF guard boots with allowPrivate: false).
* log: Watch's logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).
* log: Watch's logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).
*/
*/
-async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null } = {}) {
+async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, limitsNow = null, allowPrivate = true, accountSend = null, notifications = null } = {}) {
// A stub Network JWKS serving the generated signing key, so the SDK's JWKS client has something real
// A stub Network JWKS serving the generated signing key, so the SDK's JWKS client has something real
// to fetch and verify against (no test touches the internet). Its URL is overridable via env.
// to fetch and verify against (no test touches the internet). Its URL is overridable via env.
const jwksSite = await site({ '/api/.well-known/jwks': () => ({ body: JSON.stringify({ keys: [publicJwk] }) }) });
const jwksSite = await site({ '/api/.well-known/jwks': () => ({ body: JSON.stringify({ keys: [publicJwk] }) }) });
@@ -73,11 +73,13 @@ async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, l
@@ -73,11 +73,13 @@ async function boot({ env = {}, worker = 'off', lookupImpl, now, log = silent, l
WATCH_HOST_MIN_INTERVAL_MS: '0',
WATCH_HOST_MIN_INTERVAL_MS: '0',
WATCH_FETCH_TIMEOUT_MS: '1500',
WATCH_FETCH_TIMEOUT_MS: '1500',
WATCH_TICK_MS: '50',
WATCH_TICK_MS: '50',
+ // Actions are delivered when a test calls svc.actions.tick(), never by a background timer mid-test.
+ WATCH_ACTIONS_TICK_MS: '3600000',
...env,
...env,
});
});
// One database per boot (PGlite, or WATCH_TEST_STORE=pg: the containers), dropped when the boot stops.
// One database per boot (PGlite, or WATCH_TEST_STORE=pg: the containers), dropped when the boot stops.
const testdb = await require('./db').testDb();
const testdb = await require('./db').testDb();
- const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(now ? { now } : {}) });
+ const h = await start({ config, db: testdb.db, log, lookupImpl, limitsNow, accountSend, ...(notifications ? { notifications } : {}), ...(now ? { now } : {}) });
const base = `http://127.0.0.1:${h.server.address().port}`;
const base = `http://127.0.0.1:${h.server.address().port}`;
/** A session cookie for a person, as /auth/callback would have set it. */
/** A session cookie for a person, as /auth/callback would have set it. */