openvibers/openvibe.today Commits

Today: fill the agenda from the network's public services (#33)

OP OpenVibers committed 12 days ago · e3422719dc · 18 files, +1169 −15

Import Live streams that are on now, public Fun plans and IRL quests, Nager.Date holidays per country, and (when configured) public people-facing events from OpenVibe.Events. Each item is idempotent by source id, keeps its source link and time zone, and is retired when its source stops listing it. Followers' channels come first in a signed-in agenda; the Live events bus maps a channel to its Network subject.

Changes

ddiff --git a/.env.example b/.env.example
iindex ec74dc9..28e0df9 100644
---- a/.env.example
++++ b/.env.example
@@@ -40,6 +40,18 @@ TODAY_EVENTS_URL=
OV_LIVE_INTERNAL_URL=http://127.0.0.1:3000
#TODAY_LIVE_TIMEOUT_MS=2500
++# Imports: the public services that fill the agenda (server/imports/). The job runs at boot and every interval;
++# TODAY_IMPORTS=0 turns it off everywhere. Countries are Nager.Date codes; the Events pull stays off until
++# TODAY_IMPORTS_EVENTS_TOPICS names public people-facing topics (comma-separated).
++#TODAY_IMPORTS_INTERVAL_MS=900000
++#TODAY_IMPORTS_TIMEOUT_MS=8000
++#OV_LIVE_URL=https://openvibe.live
++#OV_FUN_URL=https://openvibe.fun
++#TODAY_HOLIDAY_COUNTRIES=US,GB,CA,DE,FR,ES,IT,NL,BE,AT,CH,SE,NO,DK,FI,PL,PT,IE,AU,NZ,JP,BR,IN,MX,CZ
++#OV_EVENTS_URL=https://openvibe.events
++#TODAY_IMPORTS_EVENTS_TOPICS=
++#TODAY_IMPORTS_EVENTS_SCOPE=events.event.read
++
# The daily brief at 06:00 UTC for subjects active in the last 30 days.
#TODAY_BRIEF_HOUR_UTC=6
#TODAY_BRIEF_ACTIVE_DAYS=30
ddiff --git a/README.md b/README.md
iindex 6e49fb5..c036438 100644
---- a/README.md
++++ b/README.md
@@@ -193,8 +193,18 @@ npm start # PORT 5100, loopback; DATABASE_URL or an embedded PGli
The tests boot the app with a mock Network and Events: events (validation, idempotency, cap, followers, cross-site,
transactional outbox), ICS (folding, escaping, cancellation window, token rotation), the consumer (signed, replay, bad
signature), the outbox (fixed id, backoff, restart), reminders (one notification, concurrency, cancelled), agenda
--(follows and Network down), brief (sections, Rundown failure, one per day), pages, sign-in, account export/deletion
--and discovery.
++(follows and Network down), imports (idempotent, time zones, source links, retirement, no private fields), brief
++(sections, Rundown failure, one per day), pages, sign-in, account export/deletion and discovery.
++
++## Imports
++
++`server/imports/` fills the public calendar from the services around the network: the streams on now on Live, public
++Fun plans and accepted IRL quests, Nager.Date holidays per country, and (when `TODAY_IMPORTS_EVENTS_TOPICS` names
++public people-facing topics) public events from OpenVibe.Events. Upcoming streams are not imported: the schedule card
++in Live's studio writes them into this service's own table. The job runs at boot and every
++`TODAY_IMPORTS_INTERVAL_MS`, is off under `TODAY_DRILL`, and `npm run imports:run` runs one pass by hand. Each row
++keeps its `source`/`source_id`, its source URL and its time zone; a second run updates the same row, and a source
++item that disappears retires its row.
## Deploy
ddiff --git a/migrations/0002_imports.sql b/migrations/0002_imports.sql
nnew file mode 100644
iindex 0000000..72ecaec
---- /dev/null
++++ b/migrations/0002_imports.sql
@@@ -0,0 +1,48 @@
++-- phase: expand
++-- OpenVibe.Today: the imports that fill the network's agenda from the public services around it. An imported
++-- event is an ordinary today_events row with the source it came from written on it, so the public API, the ICS
++-- feed and the agenda all read one table. The unique index on (source, source_id) is what makes an import
++-- idempotent: the same item from the same source is one row, updated in place.
++--
++-- today_imports is the ledger of which event each source item became. today_import_sources is the last result of
++-- each importer, so an operator can see a source that has stopped answering. today_channels remembers which
++-- Network subject owns which Live username: the public Live API names the channel, the events bus carries the
++-- subject, and the two together let a person's agenda put the channels they follow first.
++
++ALTER TABLE today_events DROP CONSTRAINT today_events_kind;
++ALTER TABLE today_events ADD CONSTRAINT today_events_kind
++ CHECK (kind IN ('stream', 'event', 'plan', 'quest', 'holiday', 'observance'));
++
++ALTER TABLE today_events ADD COLUMN source text COLLATE "C";
++ALTER TABLE today_events ADD COLUMN source_id text COLLATE "C";
++ALTER TABLE today_events ADD COLUMN source_url text;
++ALTER TABLE today_events ADD COLUMN source_kind text COLLATE "C";
++
++CREATE UNIQUE INDEX today_events_source_key
++ ON today_events (source, source_id)
++ WHERE source IS NOT NULL AND source_id IS NOT NULL;
++CREATE INDEX today_events_by_source ON today_events (source, starts_at);
++
++CREATE TABLE today_imports (
++ source text COLLATE "C" NOT NULL,
++ source_id text COLLATE "C" NOT NULL,
++ event_id text COLLATE "C" NOT NULL REFERENCES today_events(id) ON DELETE CASCADE,
++ seen_at timestamptz NOT NULL DEFAULT now(),
++ PRIMARY KEY (source, source_id)
++);
++
++CREATE TABLE today_import_sources (
++ source text COLLATE "C" PRIMARY KEY,
++ last_run_at timestamptz,
++ last_ok_at timestamptz,
++ last_error text,
++ imported integer NOT NULL DEFAULT 0,
++ retired integer NOT NULL DEFAULT 0,
++ updated_at timestamptz NOT NULL DEFAULT now()
++);
++
++CREATE TABLE today_channels (
++ username text COLLATE "C" PRIMARY KEY,
++ subject text COLLATE "C" NOT NULL,
++ updated_at timestamptz NOT NULL DEFAULT now()
++);
ddiff --git a/package.json b/package.json
iindex de6cb8c..231502f 100644
---- a/package.json
++++ b/package.json
@@@ -8,7 +8,8 @@
"start": "node server/index.js",
"dev": "NODE_ENV=development node server/index.js",
"test": "node test/run.js",
-- "test:pg": "TODAY_TEST_STORE=pg node test/run.js"
++ "test:pg": "TODAY_TEST_STORE=pg node test/run.js",
++ "imports:run": "node scripts/import-now.js"
},
"keywords": [
"openvibe"
ddiff --git a/scripts/import-now.js b/scripts/import-now.js
nnew file mode 100644
iindex 0000000..c7e7a36
---- /dev/null
++++ b/scripts/import-now.js
@@@ -0,0 +1,33 @@
++#!/usr/bin/env node
++'use strict';
++
++/**
++ * Run one import pass now (the boot job runs every TODAY_IMPORTS_INTERVAL_MS). Useful after a deploy, or for an
++ * operator checking a source:
++ *
++ * node scripts/import-now.js [live,fun,holidays,events]
++ *
++ * It refuses under NODE_ENV=production unless the process is already configured with DATABASE_URL; it uses the same
++ * loader as the service. The output is counts only.
++ */
++const { load } = require('../server/config');
++const { openStore } = require('../server/db');
++const { createImports, SOURCES } = require('../server/imports');
++
++(async () => {
++ const config = load(process.env);
++ if (config.drill) throw new Error('refusing to import under TODAY_DRILL');
++ const store = await openStore(config);
++ try {
++ const imports = createImports({ config, s: store, now: store.now });
++ const asked = process.argv.slice(2).flatMap((x) => x.split(',')).filter(Boolean);
++ const sources = asked.length ? asked : SOURCES;
++ const out = await imports.runOnce({ sources });
++ for (const [source, result] of Object.entries(out.sources || {})) {
++ console.log(`[import] ${source}: ${result.ok ? 'ok' : 'failed'} imported=${result.imported || 0} retired=${result.retired || 0}${result.error ? ` error=${result.error}` : ''}`);
++ }
++ process.exitCode = Object.values(out.sources || {}).every((r) => r.ok) ? 0 : 1;
++ } finally {
++ await store.close();
++ }
++})().catch((err) => { console.error(`[import] ${err && err.message ? err.message : err}`); process.exitCode = 1; });
ddiff --git a/server/agenda.js b/server/agenda.js
iindex 856b11a..6eddf40 100644
---- a/server/agenda.js
++++ b/server/agenda.js
@@@ -24,12 +24,16 @@ function createAgenda({ s, follows }) {
unavailable = true;
}
const rows = await db.many(
-- `SELECT * FROM today_events e
++ `SELECT e.*, (e.owner_subject = ANY($4::text[])
++ OR EXISTS (SELECT 1 FROM today_channels c WHERE LOWER(c.username) = LOWER(e.channel) AND c.subject = ANY($4::text[]))) AS followed
++ FROM today_events e
WHERE e.starts_at >= $1::timestamptz AND e.starts_at <= $2::timestamptz AND e.status <> 'cancelled'
AND (e.owner_subject = $3
OR EXISTS (SELECT 1 FROM today_reminders r WHERE r.subject = $3 AND r.event_id = e.id)
-- OR (e.owner_subject = ANY($4::text[]) AND e.visibility IN ('public', 'followers')))
-- ORDER BY e.starts_at ASC, e.id ASC`,
++ OR (e.owner_subject = ANY($4::text[]) AND e.visibility IN ('public', 'followers'))
++ OR EXISTS (SELECT 1 FROM today_channels c WHERE LOWER(c.username) = LOWER(e.channel)
++ AND c.subject = ANY($4::text[]) AND e.visibility IN ('public', 'followers')))
++ ORDER BY followed DESC, e.starts_at ASC, e.id ASC`,
[from, to, subject, followed],
);
const reminded = new Set((await db.many('SELECT event_id FROM today_reminders WHERE subject = $1', [subject])).map((r) => r.event_id));
@@@ -48,13 +52,17 @@ function createAgenda({ s, follows }) {
if (token) followed = (await follows.forSubject(subject, token)).follows;
else followed = follows.peek(subject);
return await db.many(
-- `SELECT * FROM today_events e
++ `SELECT e.*, (e.owner_subject = ANY($5::text[])
++ OR EXISTS (SELECT 1 FROM today_channels c WHERE LOWER(c.username) = LOWER(e.channel) AND c.subject = ANY($5::text[]))) AS followed
++ FROM today_events e
WHERE e.starts_at >= $1::timestamptz AND e.starts_at <= $2::timestamptz
AND (e.status <> 'cancelled' OR e.updated_at >= $3::timestamptz)
AND (e.owner_subject = $4
OR EXISTS (SELECT 1 FROM today_reminders r WHERE r.subject = $4 AND r.event_id = e.id)
-- OR (e.owner_subject = ANY($5::text[]) AND e.visibility IN ('public', 'followers')))
-- ORDER BY e.starts_at ASC, e.id ASC`,
++ OR (e.owner_subject = ANY($5::text[]) AND e.visibility IN ('public', 'followers'))
++ OR EXISTS (SELECT 1 FROM today_channels c WHERE LOWER(c.username) = LOWER(e.channel)
++ AND c.subject = ANY($5::text[]) AND e.visibility IN ('public', 'followers')))
++ ORDER BY followed DESC, e.starts_at ASC, e.id ASC`,
[from, to, new Date(now - 7 * 24 * 60 * 60_000).toISOString(), subject, followed],
);
}
ddiff --git a/server/app.js b/server/app.js
iindex 175b5df..e752e80 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -10,8 +10,8 @@
* /auth/* Network SSO with PKCE (auth/sso.js)
* /api/health, /api/ready, /release.json, /metrics (loopback only)
*
-- * The jobs (outbox publisher, reminders, daily brief) are made here but started by server/index.js (and never
-- * under TODAY_DRILL); tests drive them explicitly.
++ * The jobs (outbox publisher, import refresh, reminders, daily brief) are made here but started by server/index.js
++ * (and never under TODAY_DRILL); tests drive them explicitly.
*/
const path = require('path');
const express = require('express');
@@@ -40,6 +40,7 @@ const { createFeeds } = require('./feeds');
const { createFollows } = require('./identity/follows');
const { createAgenda } = require('./agenda');
const { createBrief } = require('./brief');
++const { createImports } = require('./imports');
const { createNotifier } = require('./notifications');
const { createReminders } = require('./reminders');
const { createNetworkSender } = require('openvibe-sdk/account-data');
@@@ -71,6 +72,7 @@ async function createApp(opts = {}) {
const notifier = createNotifier({ config, s, fetchImpl, log });
const reminders = createReminders({ config, s, events, notifier, log });
const publisher = createPublisher({ config, s, fetchImpl, log });
++ const imports = createImports({ config, s, fetchImpl, log, now: s.now });
// Account export and deletion (ADR-033). The sender posts to Network's internal export/deletion routes with
// this service's own client-credentials token; a test injects a stand-in through opts.accountSend.
@@@ -82,7 +84,7 @@ async function createApp(opts = {}) {
const ctx = {
config, s, keys, sso, principal, events, feeds, follows, agenda, brief, notifier, reminders, publisher,
-- accountData, accountSend, consumer, log,
++ imports, accountData, accountSend, consumer, log,
};
const app = express();
ddiff --git a/server/config.js b/server/config.js
iindex 31f018d..39a0114 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -71,6 +71,25 @@ function load(env = process.env) {
timeoutMs: Math.max(300, int(env.TODAY_LIVE_TIMEOUT_MS, 2500)),
},
++ // Imports from the public services around the network (server/imports/). Off under TODAY_DRILL and when
++ // TODAY_IMPORTS=0. Country codes are Nager.Date's; an empty events topic list leaves the Events pull off.
++ imports: {
++ enabled: env.TODAY_IMPORTS !== '0' && env.TODAY_IMPORTS !== 'false',
++ intervalMs: Math.max(60_000, int(env.TODAY_IMPORTS_INTERVAL_MS, 15 * 60_000)),
++ timeoutMs: Math.max(500, int(env.TODAY_IMPORTS_TIMEOUT_MS, 8000)),
++ live: { url: trim(env.OV_LIVE_URL || 'https://openvibe.live') },
++ fun: { url: trim(env.OV_FUN_URL || 'https://openvibe.fun') },
++ holidays: {
++ countries: String(env.TODAY_HOLIDAY_COUNTRIES || 'US,GB,CA,DE,FR,ES,IT,NL,BE,AT,CH,SE,NO,DK,FI,PL,PT,IE,AU,NZ,JP,BR,IN,MX,CZ')
++ .split(',').map((x) => x.trim().toUpperCase()).filter((x) => /^[A-Z]{2}$/.test(x)),
++ },
++ events: {
++ url: trim(env.OV_EVENTS_URL || env.TODAY_EVENTS_URL || ''),
++ topics: String(env.TODAY_IMPORTS_EVENTS_TOPICS || '').split(',').map((x) => x.trim()).filter(Boolean),
++ scope: env.TODAY_IMPORTS_EVENTS_SCOPE || 'events.event.read',
++ },
++ },
++
brief: {
hourUtc: Math.min(23, Math.max(0, int(env.TODAY_BRIEF_HOUR_UTC, 6))),
activeDays: Math.max(1, int(env.TODAY_BRIEF_ACTIVE_DAYS, 30)),
ddiff --git a/server/events/consumer.js b/server/events/consumer.js
iindex a9b0eb5..058ba56 100644
---- a/server/events/consumer.js
++++ b/server/events/consumer.js
@@@ -37,12 +37,31 @@ function createEventsConsumer({ config, s, accountData, accountSend = null, now
const payload = event.payload && typeof event.payload === 'object' ? event.payload : {};
const username = payload.channel && typeof payload.channel.username === 'string' ? payload.channel.username : null;
if (!username) return 'ignored:payload';
++ const subjectId = (payload.channel && payload.channel.subject && typeof payload.channel.subject.id === 'string' && payload.channel.subject.id)
++ || (event.actor && typeof event.actor.id === 'string' && event.actor.id) || null;
const atIso = new Date(now()).toISOString();
return await s.tx(async () => {
const inserted = await record(event.event_id, atIso);
if (!inserted) return { duplicate: true, outcome: null };
++ // The public Live list names a channel; the bus carries its Network subject. Remember the pair so a
++ // follower's agenda can put an imported "on now" row first (server/agenda.js).
++ if (subjectId && /^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(subjectId)) {
++ await db.exec(
++ `INSERT INTO today_channels (username, subject, updated_at) VALUES ($1, $2, $3::timestamptz)
++ ON CONFLICT (username) DO UPDATE SET subject = EXCLUDED.subject, updated_at = EXCLUDED.updated_at`,
++ [username, subjectId, atIso],
++ );
++ }
if (event.event_type === 'live.stream.started') {
const at = new Date(now()).toISOString();
++ // An on-now row imported from the public list already exists: give it the subject the bus knows.
++ if (subjectId) {
++ await db.exec(
++ `UPDATE today_events SET owner_subject = $1, seq = seq + 1, updated_at = $2::timestamptz
++ WHERE source = 'live' AND LOWER(channel) = LOWER($3) AND status = 'live' AND owner_subject <> $1`,
++ [subjectId, atIso, username],
++ );
++ }
const row = await db.maybe(
`SELECT * FROM today_events
WHERE LOWER(channel) = LOWER($1) AND status = 'scheduled'
@@@ -54,7 +73,15 @@ function createEventsConsumer({ config, s, accountData, accountSend = null, now
await db.exec("UPDATE today_events SET status = 'live', seq = seq + 1, updated_at = $1::timestamptz WHERE id = $2", [atIso, row.id]);
return { duplicate: false, outcome: 'live', event_id: row.id };
}
-- // live.stream.ended
++ // live.stream.ended: the imported on-now row ends with the stream it names (the public list catches up
++ // on its own too, but the bus is immediate).
++ const streamId = payload.stream_id == null ? null : String(payload.stream_id);
++ if (streamId) {
++ await db.exec(
++ "UPDATE today_events SET status = 'done', seq = seq + 1, updated_at = $1::timestamptz WHERE source = 'live' AND source_id = $2 AND status = 'live'",
++ [atIso, 'stream:' + streamId],
++ );
++ }
const row = await db.maybe(
"SELECT * FROM today_events WHERE LOWER(channel) = LOWER($1) AND status = 'live' ORDER BY starts_at DESC, updated_at DESC LIMIT 1",
[username],
ddiff --git a/server/events/store.js b/server/events/store.js
iindex 300e65e..f9ae405 100644
---- a/server/events/store.js
++++ b/server/events/store.js
@@@ -36,6 +36,10 @@ function wire(row) {
status: row.status,
visibility: row.visibility,
url: row.url || null,
++ source: row.source || null,
++ source_id: row.source_id || null,
++ source_url: row.source_url || null,
++ source_kind: row.source_kind || null,
seq: Number(row.seq) || 0,
created_at: iso(row.created_at),
updated_at: iso(row.updated_at),
@@@ -86,7 +90,9 @@ function createEventsStore({ s, log = console }) {
async function listPublic({ channel = null, from, to, limit = 50 } = {}) {
const args = [from, to, Math.min(100, Math.max(1, Number(limit) || 50))];
-- let where = "visibility = 'public' AND status <> 'cancelled' AND starts_at >= $1::timestamptz AND starts_at <= $2::timestamptz";
++ // A stream that is live now stays on the public list however long ago it started; everything else is
++ // bounded by the window.
++ let where = "visibility = 'public' AND status <> 'cancelled' AND ((starts_at >= $1::timestamptz AND starts_at <= $2::timestamptz) OR (status = 'live' AND starts_at <= $2::timestamptz))";
if (channel) { args.push(channel); where += ` AND LOWER(channel) = LOWER($${args.length})`; }
const rows = await db.many(
`SELECT * FROM today_events WHERE ${where} ORDER BY starts_at ASC, id ASC LIMIT $3`,
ddiff --git a/server/identity/account-data.js b/server/identity/account-data.js
iindex 56a21bb..3091b13 100644
---- a/server/identity/account-data.js
++++ b/server/identity/account-data.js
@@@ -20,6 +20,9 @@ const TABLES = [
{ table: 'today_feeds', subject: 'subject', value: (usr) => usr, file: 'feeds.json', columns: ['subject', 'created_at'], erase: 'delete' },
{ table: 'today_briefs', subject: 'subject', value: (usr) => usr, file: 'briefs.json', erase: 'delete' },
{ table: 'today_idempotency', subject: 'subject', value: (usr) => usr, file: 'idempotency.json', erase: 'delete' },
++ // The public Live username → Network subject map the import job uses to put a followed channel first. It is a
++ // public channel fact, not a private one, so it is erased with the person and not exported.
++ { table: 'today_channels', subject: 'subject', value: (usr) => usr, file: null, erase: 'delete' },
];
async function extraErase(t) {
ddiff --git a/server/imports/holidays.js b/server/imports/holidays.js
nnew file mode 100644
iindex 0000000..e83506c
---- /dev/null
++++ b/server/imports/holidays.js
@@@ -0,0 +1,124 @@
++'use strict';
++
++/**
++ * Holidays and observances from Nager.Date (https://date.nager.at): an open dataset with a public API and no key,
++ * one request per country and year. The answer is a list of
++ * { date, localName, name, countryCode, types: ['Public'|'Bank'|...] }.
++ *
++ * A holiday is a local day, not an instant: it starts at local midnight in the country's own zone and lasts until
++ * the next local midnight. The zone matters (a holiday in Berlin and one in New York start hours apart) and it moves
++ * with daylight saving, so both midnights are worked out for the date rather than assumed.
++ */
++
++/** The IANA zone a country's holidays are read in. Countries not listed are read in UTC. */
++const COUNTRY_ZONES = Object.freeze({
++ AT: 'Europe/Vienna', AU: 'Australia/Sydney', BE: 'Europe/Brussels', BR: 'America/Sao_Paulo',
++ CA: 'America/Toronto', CH: 'Europe/Zurich', CZ: 'Europe/Prague', DE: 'Europe/Berlin',
++ DK: 'Europe/Copenhagen', ES: 'Europe/Madrid', FI: 'Europe/Helsinki', FR: 'Europe/Paris',
++ GB: 'Europe/London', IE: 'Europe/Dublin', IN: 'Asia/Kolkata', IT: 'Europe/Rome',
++ JP: 'Asia/Tokyo', MX: 'America/Mexico_City', NL: 'Europe/Amsterdam', NO: 'Europe/Oslo',
++ NZ: 'Pacific/Auckland', PL: 'Europe/Warsaw', PT: 'Europe/Lisbon', SE: 'Europe/Stockholm',
++ US: 'America/New_York',
++});
++
++/** The zone a country's holidays are read in. */
++function zoneOf(country) {
++ return COUNTRY_ZONES[String(country || '').toUpperCase()] || 'UTC';
++}
++
++/** True when the runtime knows this IANA zone. */
++function isZone(zone) {
++ try {
++ new Intl.DateTimeFormat('en-US', { timeZone: zone });
++ return true;
++ } catch {
++ return false;
++ }
++}
++
++/** The offset of a zone at an instant, in milliseconds (east of UTC is positive). */
++function offsetAt(zone, ms) {
++ const parts = new Intl.DateTimeFormat('en-US', {
++ timeZone: zone, hourCycle: 'h23',
++ year: 'numeric', month: '2-digit', day: '2-digit',
++ hour: '2-digit', minute: '2-digit', second: '2-digit',
++ }).formatToParts(new Date(ms));
++ const at = {};
++ for (const p of parts) if (p.type !== 'literal') at[p.type] = p.value;
++ const asUtc = Date.UTC(Number(at.year), Number(at.month) - 1, Number(at.day), Number(at.hour) % 24, Number(at.minute), Number(at.second));
++ return asUtc - ms;
++}
++
++/**
++ * The UTC instant of local midnight on a calendar date in a zone. Two passes: the first guess uses the offset at
++ * UTC midnight, the second the offset at that guess, which is what settles the hours either side of a
++ * daylight-saving change.
++ */
++function zonedMidnight(date, zone) {
++ const [y, m, d] = String(date).split('-').map(Number);
++ if (!y || !m || !d) return null;
++ const naive = Date.UTC(y, m - 1, d, 0, 0, 0);
++ let guess = naive - offsetAt(zone, naive);
++ guess = naive - offsetAt(zone, guess);
++ const out = new Date(guess);
++ return Number.isNaN(out.getTime()) ? null : out;
++}
++
++/** The local calendar date after the one given, as YYYY-MM-DD. */
++function nextDate(date) {
++ const [y, m, d] = String(date).split('-').map(Number);
++ if (!y || !m || !d) return null;
++ return new Date(Date.UTC(y, m - 1, d + 1)).toISOString().slice(0, 10);
++}
++
++/** The locale-independent key that keeps the same observance one row when its display name changes. */
++function nameKey(name) {
++ return String(name || '')
++ .normalize('NFKD')
++ .replace(/[\u0300-\u036f]/g, '')
++ .toLowerCase()
++ .replace(/[^a-z0-9]+/g, '-')
++ .replace(/^-+|-+$/g, '')
++ .slice(0, 60) || 'holiday';
++}
++
++/**
++ * The events a country and year hold, as import records. An observance is kept apart from a holiday:
++ * both are on the calendar, but only one of them is a day off.
++ */
++function recordsFor(payload, { country, year, zone = zoneOf(country) } = {}) {
++ const list = Array.isArray(payload) ? payload : [];
++ const countryCode = String(country || '').toUpperCase();
++ const api = 'https://date.nager.at/api/v3/PublicHolidays/' + year + '/' + countryCode;
++ const out = [];
++ for (const item of list) {
++ if (!item || typeof item.date !== 'string' || !/^\d{4}-\d{2}-\d{2}$/.test(item.date)) continue;
++ const start = zonedMidnight(item.date, zone);
++ const end = zonedMidnight(nextDate(item.date), zone);
++ if (!start || !end) continue;
++ const types = Array.isArray(item.types) ? item.types.map((t) => String(t).toLowerCase()) : [String(item.type || '').toLowerCase()];
++ const observance = types.includes('observance') && !types.includes('public');
++ const name = String(item.localName || item.name || 'Holiday').trim().slice(0, 200) || 'Holiday';
++ const cc = String(item.countryCode || countryCode).toUpperCase();
++ out.push({
++ source: 'holidays',
++ source_id: 'holiday:' + cc + ':' + item.date + ':' + nameKey(name),
++ source_kind: observance ? 'observance' : 'holiday',
++ kind: observance ? 'observance' : 'holiday',
++ title: name,
++ description: (observance ? 'An observance in ' : 'A public holiday in ') + cc + ', from the Nager.Date open dataset.',
++ starts_at: start.toISOString(),
++ ends_at: end.toISOString(),
++ tz: zone,
++ url: api,
++ source_url: api,
++ owner_subject: 'sys:holidays',
++ channel: null,
++ status: 'scheduled',
++ visibility: 'public',
++ });
++ }
++ return out;
++}
++
++module.exports = { COUNTRY_ZONES, zoneOf, isZone, zonedMidnight, nextDate, recordsFor };
ddiff --git a/server/imports/index.js b/server/imports/index.js
nnew file mode 100644
iindex 0000000..152a36b
---- /dev/null
++++ b/server/imports/index.js
@@@ -0,0 +1,336 @@
++'use strict';
++
++/**
++ * Import jobs that fill Today from the public services around the network. Each source is read through its public
++ * API only, its items are normalised by server/imports/normalize.js (or holidays.js), and one row per (source,
++ * source_id) is written to today_events. The unique index on that pair is the idempotency: a second run updates the
++ * same row instead of adding one, and a changed source bumps seq (the ICS SEQUENCE) only once.
++ *
++ * live the streams on now from GET /api/streams; a stream that ended is marked done. Upcoming streams are
++ * not imported: the schedule card in Live's studio writes them into this service's own table.
++ * fun public group plans (GET /api/v1/plans) and accepted/live IRL quests (GET /api/v1/quests).
++ * holidays Nager.Date's open public-holiday API, per country, at local midnight in that country's zone.
++ * events public people-facing events pulled from OpenVibe.Events, off until TODAY_IMPORTS_EVENTS_TOPICS
++ * names the topics to read (release notes, deploys and telemetry are never events).
++ *
++ * Nothing here reads a token or a person's identity from a source. A source row that stops being public is retired:
++ * a stream is marked done, a plan or public event cancelled. The jobs are off under TODAY_DRILL and when
++ * TODAY_IMPORTS=0; tests call runOnce() themselves.
++ */
++const { serviceAuth } = require('openvibe-contracts');
++const { newEventId } = require('../db');
++const { liveStream, funPlan, funQuest, networkEvent } = require('./normalize');
++const { recordsFor, isZone } = require('./holidays');
++
++const KINDS = Object.freeze(['stream', 'event', 'plan', 'quest', 'holiday', 'observance']);
++const STATUSES = Object.freeze(['scheduled', 'live', 'done', 'cancelled']);
++const MAX_TITLE = 120;
++const MAX_DESCRIPTION = 1000;
++const MAX_SOURCE_ID = 200;
++const HOLIDAY_TTL_MS = 12 * 60 * 60_000;
++const PAGE_LIMIT = 50;
++const MAX_PAGES = 10;
++const SOURCES = Object.freeze(['live', 'fun', 'holidays', 'events']);
++
++const iso = (v) => (v == null ? null : (v instanceof Date ? v.toISOString() : String(v)));
++const same = (a, b) => (a == null && b == null) || (a != null && b != null && String(a) === String(b));
++const plain = (v, max) => String(v == null ? '' : v).trim().slice(0, max);
++const https = (v) => (typeof v === 'string' && /^https:\/\/[^\s"'<>]{1,300}$/.test(v) ? v : null);
++
++/** One normalised source record, checked against the columns Today owns. */
++function cleanRecord(record, { source, at }) {
++ if (!record || record.source !== source) return null;
++ const sourceId = plain(record.source_id, MAX_SOURCE_ID);
++ const title = plain(record.title, MAX_TITLE);
++ const starts = record.starts_at ? new Date(record.starts_at) : null;
++ if (!sourceId || !title || !starts || Number.isNaN(starts.getTime())) return null;
++ const ends = record.ends_at ? new Date(record.ends_at) : null;
++ if (ends && Number.isNaN(ends.getTime())) return null;
++ const kind = KINDS.includes(record.kind) ? record.kind : 'event';
++ const status = STATUSES.includes(record.status) ? record.status : 'scheduled';
++ const tz = typeof record.tz === 'string' && isZone(record.tz) ? record.tz : 'UTC';
++ const description = plain(record.description, MAX_DESCRIPTION) || null;
++ return {
++ source,
++ source_id: sourceId,
++ source_kind: plain(record.source_kind, 40) || kind,
++ kind,
++ title,
++ description,
++ starts_at: starts.toISOString(),
++ ends_at: ends ? ends.toISOString() : null,
++ tz,
++ status,
++ visibility: 'public',
++ url: https(record.url),
++ source_url: https(record.source_url) || https(record.url),
++ owner_subject: plain(record.owner_subject, 80) || 'sys:' + source,
++ channel: plain(record.channel, 64) || null,
++ updated_at: new Date(at).toISOString(),
++ };
++}
++
++function changed(row, r) {
++ return !same(iso(row.title), r.title)
++ || !same(row.description, r.description)
++ || !same(iso(row.starts_at), r.starts_at)
++ || !same(iso(row.ends_at), r.ends_at)
++ || !same(row.tz, r.tz)
++ || !same(row.status, r.status)
++ || !same(row.url, r.url)
++ || !same(row.source_url, r.source_url)
++ || !same(row.source_kind, r.source_kind)
++ || !same(row.owner_subject, r.owner_subject)
++ || !same(row.channel, r.channel);
++}
++
++function createImports({ config, s, fetchImpl = globalThis.fetch, log = console, now = s.now }) {
++ const db = s.db;
++ const holidays = new Map(); // `CC:year` -> { at, records }
++ const stats = { last: null, running: false, started: false };
++ let timer = null;
++ let timeout = null;
++
++ async function getJson(url, { headers = {}, fetch = fetchImpl } = {}) {
++ const res = await fetch(url, {
++ headers: { Accept: 'application/json', 'User-Agent': 'OpenVibe.Today (+https://openvibe.today)', ...headers },
++ signal: AbortSignal.timeout(config.imports.timeoutMs),
++ });
++ if (!res.ok) {
++ const err = new Error(`${new URL(url).host} answered ${res.status}`);
++ err.status = res.status;
++ throw err;
++ }
++ // Nager.Date answers 204 for a country it has no data for; that is an empty year, not a failure.
++ if (res.status === 204) return [];
++ const body = await res.json().catch(() => null);
++ if (!body || typeof body !== 'object') throw new Error(`${new URL(url).host} answered without JSON`);
++ return body;
++ }
++
++ async function channelSubjects() {
++ const rows = await db.many('SELECT username, subject FROM today_channels');
++ return new Map(rows.map((r) => [String(r.username).toLowerCase(), r.subject]));
++ }
++
++ /** Write a whole source's records in one transaction; returns the source ids that exist after it. */
++ async function applyRecords(source, records, at) {
++ const seen = [];
++ await s.tx(async () => {
++ for (const raw of records) {
++ const r = cleanRecord(raw, { source, at });
++ if (!r) continue;
++ seen.push(r.source_id);
++ const row = await db.maybe('SELECT * FROM today_events WHERE source = $1 AND source_id = $2', [source, r.source_id]);
++ let event = row;
++ if (!row) {
++ const id = newEventId();
++ event = await db.maybe(
++ `INSERT INTO today_events
++ (id, owner_subject, kind, channel, title, description, starts_at, ends_at, tz, status, visibility,
++ url, source, source_id, source_url, source_kind, seq, created_at, updated_at)
++ VALUES ($1, $2, $3, $4, $5, $6, $7::timestamptz, $8::timestamptz, $9, $10, 'public',
++ $11, $12, $13, $14, $15, 0, $16::timestamptz, $16::timestamptz)
++ ON CONFLICT (source, source_id) WHERE source IS NOT NULL AND source_id IS NOT NULL DO NOTHING
++ RETURNING *`,
++ [id, r.owner_subject, r.kind, r.channel, r.title, r.description, r.starts_at, r.ends_at, r.tz, r.status,
++ r.url, r.source, r.source_id, r.source_url, r.source_kind, r.updated_at],
++ );
++ if (!event) event = await db.maybe('SELECT * FROM today_events WHERE source = $1 AND source_id = $2', [source, r.source_id]);
++ } else if (changed(row, r)) {
++ event = await db.maybe(
++ `UPDATE today_events SET owner_subject = $2, kind = $3, channel = $4, title = $5, description = $6,
++ starts_at = $7::timestamptz, ends_at = $8::timestamptz, tz = $9, status = $10, url = $11,
++ source_url = $12, source_kind = $13, seq = seq + 1, updated_at = $14::timestamptz
++ WHERE id = $1 RETURNING *`,
++ [row.id, r.owner_subject, r.kind, r.channel, r.title, r.description, r.starts_at, r.ends_at, r.tz,
++ r.status, r.url, r.source_url, r.source_kind, r.updated_at],
++ );
++ }
++ if (event) {
++ await db.exec(
++ `INSERT INTO today_imports (source, source_id, event_id, seen_at) VALUES ($1, $2, $3, $4::timestamptz)
++ ON CONFLICT (source, source_id) DO UPDATE SET event_id = EXCLUDED.event_id, seen_at = EXCLUDED.seen_at`,
++ [source, r.source_id, event.id, r.updated_at],
++ );
++ }
++ }
++ });
++ return [...new Set(seen)];
++ }
++
++ /** Mark imported rows for this source that a successful read no longer lists. */
++ async function retire(source, seen, at, status) {
++ const when = new Date(at).toISOString();
++ const n = await db.exec(
++ `UPDATE today_events SET status = $1, seq = seq + 1, updated_at = $2::timestamptz
++ WHERE source = $3 AND NOT (source_id = ANY($4::text[])) AND status NOT IN ('done', 'cancelled')`,
++ [status, when, source, seen],
++ );
++ return Number(n) || 0;
++ }
++
++ async function runLive(at, fetch) {
++ const body = await getJson(`${config.imports.live.url}/api/streams`, { fetch });
++ if (!Array.isArray(body.streams)) throw new Error('Live answered without a streams list');
++ const list = body.streams;
++ const subjects = await channelSubjects();
++ const records = [];
++ for (const stream of list) {
++ const username = plain(stream && stream.channel && stream.channel.username, 64);
++ records.push(liveStream(stream, { subject: subjects.get(username.toLowerCase()) || null }));
++ }
++ const seen = await applyRecords('live', records, at);
++ return { ok: true, imported: seen.length, retired: await retire('live', seen, at, 'done') };
++ }
++
++ async function pages(fetch, base, key) {
++ const out = [];
++ let cursor = null;
++ for (let page = 0; page < MAX_PAGES; page++) {
++ const url = `${base}${base.includes('?') ? '&' : '?'}limit=${PAGE_LIMIT}${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ''}`;
++ const body = await getJson(url, { fetch });
++ if (!Array.isArray(body[key])) throw new Error(`${new URL(url).host} answered without a ${key} list`);
++ const rows = body[key];
++ out.push(...rows);
++ cursor = typeof body.next_cursor === 'string' && body.next_cursor ? body.next_cursor : null;
++ if (!cursor) break;
++ }
++ return out;
++ }
++
++ async function runFun(at, fetch) {
++ const plans = await pages(fetch, `${config.imports.fun.url}/api/v1/plans`, 'plans');
++ const planSeen = await applyRecords('fun', plans.map(funPlan), at);
++ const quests = await pages(fetch, `${config.imports.fun.url}/api/v1/quests`, 'quests');
++ const questSeen = await applyRecords('fun', quests.map(funQuest), at);
++ const seen = [...new Set([...planSeen, ...questSeen])];
++ return { ok: true, imported: seen.length, retired: await retire('fun', seen, at, 'cancelled') };
++ }
++
++ async function holidayRecords(at, fetch, { force = false } = {}) {
++ const year = new Date(at).getUTCFullYear();
++ const years = [year, year + 1];
++ const records = [];
++ let failed = 0;
++ let fetched = 0;
++ for (const country of config.imports.holidays.countries) {
++ const cc = String(country || '').trim().toUpperCase();
++ if (!/^[A-Z]{2}$/.test(cc)) continue;
++ for (const y of years) {
++ const key = `${cc}:${y}`;
++ const hit = holidays.get(key);
++ if (!force && hit && at - hit.at < HOLIDAY_TTL_MS) { records.push(...hit.records); continue; }
++ try {
++ const payload = await getJson(`https://date.nager.at/api/v3/PublicHolidays/${y}/${cc}`, { fetch });
++ const rows = recordsFor(payload, { country: cc, year: y });
++ holidays.set(key, { at, records: rows });
++ records.push(...rows);
++ fetched++;
++ } catch (err) {
++ failed++;
++ log.warn(`[OpenVibe.Today] holidays ${cc}/${y}: ${(err && err.message) || err}`);
++ }
++ }
++ }
++ if (failed && !fetched && !records.length) throw new Error(`Nager.Date did not answer for ${failed} country-years`);
++ return { records, failed };
++ }
++
++ async function runHolidays(at, fetch, opts) {
++ const { records, failed } = await holidayRecords(at, fetch, opts);
++ const seen = await applyRecords('holidays', records, at);
++ return { ok: true, imported: seen.length, retired: 0, ...(failed ? { failed } : {}) };
++ }
++
++ async function runEvents(at, fetch) {
++ const { topics, url, scope } = config.imports.events;
++ if (!topics.length || !url || !config.oauth.clientSecret) return { ok: true, imported: 0, retired: 0, skipped: 'not_configured' };
++ const tokens = serviceAuth.createTokenClient({
++ tokenUrl: `${config.networkInternalUrl}/oauth/token`,
++ clientId: config.oauth.clientId,
++ clientSecret: config.oauth.clientSecret,
++ audience: 'openvibe.events',
++ scope,
++ fetchImpl: fetch,
++ });
++ const body = await getJson(`${url.replace(/\/+$/, '')}/api/v1/events?topic=${encodeURIComponent(topics.join(','))}&limit=100`, { fetch,
++ headers: await tokens.authHeaders(),
++ });
++ const records = (Array.isArray(body.events) ? body.events : [])
++ .map((row) => networkEvent(row && row.event ? row.event : row))
++ .filter(Boolean);
++ const seen = await applyRecords('events', records, at);
++ return { ok: true, imported: seen.length, retired: 0 };
++ }
++
++ const runners = { live: runLive, fun: runFun, holidays: runHolidays, events: runEvents };
++
++ async function recordSource(source, result) {
++ const at = new Date(now()).toISOString();
++ await db.exec(
++ `INSERT INTO today_import_sources (source, last_run_at, last_ok_at, last_error, imported, retired, updated_at)
++ VALUES ($1, $2::timestamptz, $3::timestamptz, $4, $5, $6, $2::timestamptz)
++ ON CONFLICT (source) DO UPDATE SET last_run_at = EXCLUDED.last_run_at,
++ last_ok_at = CASE WHEN $3::timestamptz IS NULL THEN today_import_sources.last_ok_at ELSE EXCLUDED.last_ok_at END,
++ last_error = EXCLUDED.last_error, imported = EXCLUDED.imported, retired = EXCLUDED.retired,
++ updated_at = EXCLUDED.updated_at`,
++ [source, at, result.ok ? at : null, result.ok ? null : String(result.error || 'failed').slice(0, 500), result.imported || 0, result.retired || 0],
++ );
++ }
++
++ /**
++ * Run one import pass. `sources` selects a subset (tests do); a source that throws is recorded and never stops
++ * another source. Nothing is retired by a source that failed to read.
++ */
++ async function runOnce({ sources = SOURCES, fetchImpl: fetch = fetchImpl, force = false } = {}) {
++ if (stats.running) return { busy: true, at: new Date(now()).toISOString(), sources: {} };
++ stats.running = true;
++ const at = now();
++ const out = {};
++ try {
++ for (const source of sources) {
++ const runner = runners[source];
++ if (!runner) continue;
++ try {
++ out[source] = await runner(at, fetch, { force });
++ } catch (err) {
++ out[source] = { ok: false, imported: 0, retired: 0, error: (err && err.message) || String(err) };
++ log.warn(`[OpenVibe.Today] import ${source}: ${(err && err.message) || err}`);
++ }
++ await recordSource(source, out[source]);
++ }
++ stats.last = { at: new Date(at).toISOString(), sources: out };
++ return stats.last;
++ } finally {
++ stats.running = false;
++ }
++ }
++
++ function start() {
++ if (!config.imports.enabled || config.drill) return false;
++ if (stats.started) return true;
++ stats.started = true;
++ const tick = async () => {
++ try { await runOnce(); } catch (err) { log.warn('[OpenVibe.Today] imports:', (err && err.message) || err); }
++ };
++ timeout = setTimeout(tick, 10_000);
++ if (timeout.unref) timeout.unref();
++ timer = setInterval(tick, config.imports.intervalMs);
++ if (timer.unref) timer.unref();
++ log.log(`[OpenVibe.Today] imports every ${Math.round(config.imports.intervalMs / 60_000)} min: ${SOURCES.join(', ')}`);
++ return true;
++ }
++
++ function stop() {
++ if (timeout) clearTimeout(timeout);
++ if (timer) clearInterval(timer);
++ timeout = null;
++ timer = null;
++ stats.started = false;
++ }
++
++ return { runOnce, start, stop, stats, sources: SOURCES };
++}
++
++module.exports = { createImports, cleanRecord, changed, SOURCES };
ddiff --git a/server/imports/normalize.js b/server/imports/normalize.js
nnew file mode 100644
iindex 0000000..ef01715
---- /dev/null
++++ b/server/imports/normalize.js
@@@ -0,0 +1,201 @@
++'use strict';
++
++/**
++ * What a public source says, read as an event. Every function here is pure: it takes what a source returned and gives
++ * back the row Today wants, with the source's own id on it. Nothing is fetched and nothing is written here, so each
++ * reading can be tried on its own.
++ *
++ * Only public facts are read. A stream's title, channel name and start are public (they are on the public Live page);
++ * a plan's title and time are public (they are on the public Fun page); a person's identity is not, so an imported
++ * plan is owned by the network rather than by the person who wrote it. A source may return private fields next to the
++ * public ones; nothing here copies anything but the named public fields.
++ *
++ * OpenVibe.Live has no schedule store of its own: the schedule card in Live's studio writes to Today through this
++ * service's own API (plan 2026-10-28 §4.2), so an upcoming stream is already a today_events row and must not be
++ * imported back from Live. The live list below is only the streams that are on now.
++ */
++const { isZone } = require('./holidays');
++
++const LIVE_HOME = 'https://openvibe.live';
++const FUN_HOME = 'https://openvibe.fun';
++const EVENTS_HOME = 'https://openvibe.events';
++
++const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
++
++/**
++ * A source time as an instant. Live writes SQLite's ov_now() as `YYYY-MM-DD HH:MM:SS` in UTC with no zone, which
++ * JavaScript would otherwise read in the server's local zone; an ISO time with an offset is read as it is.
++ */
++function parseSourceTime(value) {
++ if (value instanceof Date) return Number.isNaN(value.getTime()) ? null : value;
++ if (typeof value === 'number' && Number.isFinite(value)) {
++ const d = new Date(value);
++ return Number.isNaN(d.getTime()) ? null : d;
++ }
++ if (typeof value !== 'string') return null;
++ const s = value.trim();
++ if (!s) return null;
++ let iso = s;
++ if (/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}(?::\d{2}(?:\.\d+)?)?$/.test(s)) iso = s.replace(' ', 'T') + 'Z';
++ else if (/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}(?::\d{2}(?:\.\d+)?)?$/.test(s)) iso = s + 'Z';
++ const d = new Date(iso);
++ return Number.isNaN(d.getTime()) ? null : d;
++}
++
++function plain(value, max) {
++ return String(value == null ? '' : value).trim().slice(0, max);
++}
++
++/** Only an https page is a source link; anything else falls back to the service's home. */
++function httpsUrl(value, fallback) {
++ return typeof value === 'string' && /^https:\/\/[^\s"'<>]{1,300}$/.test(value) ? value : fallback;
++}
++
++function zoneOr(value, fallback = 'UTC') {
++ return typeof value === 'string' && isZone(value) ? value : fallback;
++}
++
++/** The subject a channel is known by: the Network subject when the bus told us, the channel name otherwise. */
++function ownerOf(subject, username) {
++ if (typeof subject === 'string' && USER_SUBJECT.test(subject)) return subject;
++ return 'chan:' + String(username || 'unknown').slice(0, 64);
++}
++
++/**
++ * A stream that is on now. Live's public list says which streams are live; the row becomes an event that is live, so
++ * the agenda and the public list can say "on now" without asking Live again.
++ */
++function liveStream(stream, { subject = null } = {}) {
++ if (!stream || typeof stream !== 'object') return null;
++ const id = stream.id == null ? null : String(stream.id);
++ const channel = stream.channel && typeof stream.channel === 'object' ? stream.channel : {};
++ const username = plain(channel.username || stream.username, 64);
++ if (!id || !username) return null;
++ const started = parseSourceTime(stream.started_at || stream.created_at);
++ if (!started) return null;
++ const title = plain(stream.title, 200) || (username + ' is live');
++ const url = LIVE_HOME + '/@' + encodeURIComponent(username);
++ return {
++ source: 'live',
++ source_id: 'stream:' + id,
++ source_kind: 'on_now',
++ kind: 'stream',
++ title,
++ description: 'Live now on OpenVibe.Live.',
++ starts_at: started.toISOString(),
++ ends_at: null,
++ tz: 'UTC',
++ url,
++ source_url: url,
++ owner_subject: ownerOf(subject, username),
++ channel: username,
++ status: 'live',
++ visibility: 'public',
++ };
++}
++
++/**
++ * A group plan on OpenVibe.Fun. The plan's own timezone is kept, so a plan in Berlin starts at the hour the plan says
++ * it starts, whatever the reader's own clock says. Fun's own Today body gives a plan two hours (server/today.js);
++ * `closes_at` is when voting closed, not when the plan ends, so it is never used as ends_at.
++ */
++function funPlan(plan) {
++ if (!plan || typeof plan !== 'object') return null;
++ if (plan.visibility && plan.visibility !== 'public') return null;
++ const slug = plain(plan.slug, 80);
++ if (!slug) return null;
++ const starts = parseSourceTime(plan.starts_at);
++ if (!starts) return null;
++ const where = [plain(plan.city, 120), plain(plan.place_name, 120)].filter(Boolean).join(', ');
++ const url = FUN_HOME + '/plans/' + encodeURIComponent(slug);
++ return {
++ source: 'fun',
++ source_id: 'plan:' + slug,
++ source_kind: 'plan',
++ kind: 'plan',
++ title: plain(plan.title, 200) || 'A plan on OpenVibe.Fun',
++ description: [plain(plan.summary, 500), where].filter(Boolean).join(' — ').slice(0, 1000) || null,
++ starts_at: starts.toISOString(),
++ ends_at: new Date(starts.getTime() + 2 * 60 * 60_000).toISOString(),
++ tz: zoneOr(plan.timezone),
++ url,
++ source_url: url,
++ owner_subject: 'sys:fun',
++ channel: null,
++ status: 'scheduled',
++ visibility: 'public',
++ };
++}
++
++/**
++ * An IRL quest on OpenVibe.Fun. A quest has no clock of its own, so it enters the calendar at the moment it was
++ * accepted and stays live while the streamer is running it. Completed or verified quests are history, not an
++ * upcoming plan, and hidden/unlisted ones never appear in the public list.
++ */
++function funQuest(quest) {
++ if (!quest || typeof quest !== 'object') return null;
++ if (quest.visibility && quest.visibility !== 'public') return null;
++ const slug = plain(quest.slug, 80);
++ const status = String(quest.status || '').toLowerCase();
++ if (!slug || !['accepted', 'live'].includes(status)) return null;
++ const start = parseSourceTime(quest.accepted_at || quest.created_at);
++ if (!start) return null;
++ const where = [plain(quest.city, 120), plain(quest.place_name, 120)].filter(Boolean).join(', ');
++ const url = FUN_HOME + '/q/' + encodeURIComponent(slug);
++ return {
++ source: 'fun',
++ source_id: 'quest:' + slug,
++ source_kind: status === 'live' ? 'on_now' : 'quest',
++ kind: 'quest',
++ title: plain(quest.title, 200) || 'A quest on OpenVibe.Fun',
++ description: [plain(quest.brief, 500), where].filter(Boolean).join(' — ').slice(0, 1000) || null,
++ starts_at: start.toISOString(),
++ ends_at: null,
++ tz: 'UTC',
++ url,
++ source_url: url,
++ owner_subject: 'sys:fun',
++ channel: plain(quest.streamer_username, 64) || null,
++ status: status === 'live' ? 'live' : 'scheduled',
++ visibility: 'public',
++ };
++}
++
++/**
++ * A public event from the network's event bus. Release notes, deploys, migrations, builds and heartbeats are not
++ * events a person turns up to, so they are left out: only a public event with a title and a time is one.
++ */
++function networkEvent(event) {
++ if (!event || typeof event !== 'object') return null;
++ const type = String(event.event_type || '');
++ if (!type || /release|deploy|migration|build|rollout|heartbeat|analytics|usage|telemetry/i.test(type)) return null;
++ if (event.visibility && event.visibility !== 'public') return null;
++ const payload = event.payload && typeof event.payload === 'object' ? event.payload : {};
++ const title = plain(payload.title, 200);
++ const starts = parseSourceTime(payload.starts_at || payload.at || payload.when || event.occurred_at || event.timestamp);
++ const id = typeof event.event_id === 'string' ? event.event_id : null;
++ if (!title || !starts || !id) return null;
++ const url = httpsUrl(payload.url, EVENTS_HOME);
++ return {
++ source: 'events',
++ source_id: 'event:' + id,
++ source_kind: 'network_event',
++ kind: 'event',
++ title,
++ description: plain(payload.summary || payload.description, 1000) || null,
++ starts_at: starts.toISOString(),
++ ends_at: parseSourceTime(payload.ends_at) ? parseSourceTime(payload.ends_at).toISOString() : null,
++ tz: zoneOr(payload.tz),
++ url,
++ source_url: url,
++ owner_subject: 'sys:events',
++ channel: null,
++ status: 'scheduled',
++ visibility: 'public',
++ };
++}
++
++module.exports = {
++ liveStream, funPlan, funQuest, networkEvent, ownerOf, parseSourceTime, httpsUrl, zoneOr,
++ LIVE_HOME, FUN_HOME, EVENTS_HOME,
++};
ddiff --git a/server/index.js b/server/index.js
iindex 0db63f3..a4cb37c 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -19,6 +19,7 @@ function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }
() => { for (const t of timers) clearInterval(t); },
() => ctx.keys.client.stop(),
() => ctx.publisher.stop(),
++ () => ctx.imports.stop(),
() => ctx.reminders.stop(),
() => ctx.brief.stop(),
async () => { clearInterval(ctx.telemetryTick); if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); },
@@@ -49,6 +50,7 @@ async function start() {
// Jobs. Each refuses to start under TODAY_DRILL (or when it is not configured); tests call their runOnce().
ctx.publisher.start();
++ ctx.imports.start();
ctx.reminders.start();
ctx.brief.start();
ddiff --git a/test/agenda.test.js b/test/agenda.test.js
iindex 568eb2c..efd934f 100644
---- a/test/agenda.test.js
++++ b/test/agenda.test.js
@@@ -45,7 +45,12 @@ const { boot, check, done } = require('./helpers/boot');
assert.ok(!ids.includes(unfollowed.id), 'not followed and not reminded');
assert.strictEqual(b.events.find((e) => e.id === own.id).own, true);
assert.strictEqual(b.events.find((e) => e.id === reminded.id).reminded, true);
-- assert.deepStrictEqual(ids, [...ids].sort((x, y) => {
++ // Followed channels come first, then everything is soonest first. Carol's two events start after alice's
++ // own one, so this also proves the agenda is not merely sorting by starts_at.
++ const followedIds = new Set([followed.id, followedPrivate.id]);
++ assert.ok(ids.slice(0, 2).every((id) => followedIds.has(id)), `followed first: ${ids.join(', ')}`);
++ const rest = ids.slice(2);
++ assert.deepStrictEqual(rest, [...rest].sort((x, y) => {
const a = b.events.find((e) => e.id === x); const c = b.events.find((e) => e.id === y);
return new Date(a.starts_at) - new Date(c.starts_at);
}));
ddiff --git a/test/consumer.test.js b/test/consumer.test.js
iindex 5d7693d..31c2a30 100644
---- a/test/consumer.test.js
++++ b/test/consumer.test.js
@@@ -33,6 +33,24 @@ const ended = (id) => ({ event_id: 'evt_01M61F7R3WXM5T5ET9CRGNHEP2', event_type:
assert.strictEqual(await t.ctx.s.db.value('SELECT COUNT(*)::int AS n FROM today_processed WHERE event_id = $1', [started(id).event_id]), 1);
});
++ await check('the bus remembers the channel\'s Network subject and gives it to an imported on-now row', async () => {
++ await t.ctx.s.db.exec(
++ `INSERT INTO today_events (id, owner_subject, kind, channel, title, starts_at, tz, status, visibility, source, source_id, source_kind)
++ VALUES ('tev_01JZ0000000000000000001', 'chan:alice', 'stream', 'alice', 'Imported now', now(), 'UTC', 'live', 'public', 'live', 'stream:7', 'on_now')`,
++ );
++ const event = {
++ ...started(id),
++ event_id: 'evt_01M61F7R3WXM5T5ET9CRGNHEP3',
++ payload: { ...started(id).payload, channel: { username: 'alice', subject: { type: 'user', id: alice.subject } } },
++ };
++ const r = await post(bodyOf(event));
++ assert.strictEqual(r.status, 200, r.text);
++ const mapping = await t.ctx.s.db.maybe('SELECT * FROM today_channels WHERE username = $1', ['alice']);
++ assert.strictEqual(mapping.subject, alice.subject);
++ const imported = await t.ctx.s.db.maybe("SELECT * FROM today_events WHERE source = 'live' AND source_id = 'stream:7'");
++ assert.strictEqual(imported.owner_subject, alice.subject, 'the public-list row learns its subject');
++ });
++
await check('a replayed delivery changes nothing', async () => {
const b = bodyOf(started(id));
const r = await post(b);
ddiff --git a/test/imports.test.js b/test/imports.test.js
nnew file mode 100644
iindex 0000000..8c07078
---- /dev/null
++++ b/test/imports.test.js
@@@ -0,0 +1,299 @@
++'use strict';
++/**
++ * The import jobs: public Live streams on now, Fun plans and quests, Nager.Date holidays, and the source link every
++ * row keeps. The same source item is one row across runs, a changed item bumps seq once, a stream that ends is
++ * retired, no private field leaks, and a follower's agenda puts the channels they follow first.
++ */
++const assert = require('assert');
++const { boot, check, done } = require('./helpers/boot');
++const { networkEvent, funQuest, funPlan } = require('../server/imports/normalize');
++
++const NOW = Date.UTC(2026, 9, 31, 12, 0, 0);
++const iso = (ms) => new Date(ms).toISOString();
++
++const LIVE = {
++ streams: [
++ {
++ id: 42,
++ title: 'Night walk',
++ started_at: '2026-10-31 11:30:00',
++ channel: { username: 'goosely', display_name: 'Goose', email: '[email protected]' },
++ stream_key: 'live_should_never_be_stored',
++ is_nsfw: 0,
++ },
++ { id: 43, title: 'No channel', started_at: '2026-10-31 11:40:00', channel: {} },
++ ],
++};
++
++const PLANS = {
++ count: 1,
++ plans: [{
++ slug: 'berlin-picnic', title: 'Berlin picnic', summary: 'Bring bread and a blanket.',
++ city: 'Berlin', timezone: 'Europe/Berlin', starts_at: '2026-10-31T11:00:00.000Z',
++ closes_at: '2026-10-30T18:00:00.000Z', status: 'decided', visibility: 'public',
++ owner_username: 'carol', owner_subject: 'usr_01JZ0000000000000000000CAR',
++ }],
++ next_cursor: null,
++};
++
++const QUESTS = {
++ count: 2,
++ quests: [
++ { slug: 'find-the-cat', title: 'Find the cat', brief: 'The cat is somewhere in Alfama.',
++ status: 'accepted', accepted_at: '2026-10-31T13:00:00.000Z', city: 'Lisbon',
++ place_name: 'Alfama', streamer_username: 'goosely', visibility: 'public' },
++ { slug: 'already-done', title: 'Already done', status: 'completed', accepted_at: '2026-10-30T13:00:00.000Z', visibility: 'public' },
++ { slug: 'hidden-quest', title: 'Hidden', status: 'accepted', accepted_at: '2026-10-31T14:00:00.000Z', visibility: 'hidden' },
++ ],
++ next_cursor: null,
++};
++
++const HOLIDAYS = {
++ 'DE:2026': [
++ { date: '2026-12-25', localName: 'Erster Weihnachtstag', name: 'Christmas Day', countryCode: 'DE', types: ['Public'] },
++ { date: '2026-06-03', localName: 'Sommerfest', name: 'Summer day', countryCode: 'DE', types: ['Public'] },
++ ],
++ 'DE:2027': [],
++ 'US:2026': [{ date: '2026-12-25', localName: 'Christmas Day', name: 'Christmas Day', countryCode: 'US', types: ['Public'] }],
++ 'US:2027': [],
++};
++
++function fakeFetch({ live = LIVE, plans = PLANS, quests = QUESTS, failLive = false } = {}) {
++ return async (url) => {
++ const u = new URL(url);
++ if (u.hostname === 'openvibe.live' && u.pathname === '/api/streams') {
++ if (failLive) throw new Error('Live is restarting');
++ return { ok: true, status: 200, json: async () => live };
++ }
++ if (u.hostname === 'openvibe.fun' && u.pathname === '/api/v1/plans') return { ok: true, status: 200, json: async () => plans };
++ if (u.hostname === 'openvibe.fun' && u.pathname === '/api/v1/quests') return { ok: true, status: 200, json: async () => quests };
++ if (u.hostname === 'date.nager.at') {
++ const m = /^\/api\/v3\/PublicHolidays\/(\d{4})\/([A-Z]{2})$/.exec(u.pathname);
++ const rows = m ? HOLIDAYS[`${m[2]}:${m[1]}`] : null;
++ return rows ? { ok: true, status: 200, json: async () => rows } : { ok: false, status: 404, json: async () => ({}) };
++ }
++ return { ok: false, status: 404, json: async () => ({}) };
++ };
++}
++
++(async () => {
++ const t = await boot({ now: () => NOW, env: { TODAY_HOLIDAY_COUNTRIES: 'DE,US', TODAY_IMPORTS_EVENTS_TOPICS: 'network.meetup.*' } });
++ try {
++ const alice = t.network.addUser('alice');
++ const bob = t.network.addUser('bob');
++ const fetch = fakeFetch();
++
++ let liveId = null;
++ let planId = null;
++ let firstSeq = null;
++
++ await check('a first pass imports the on-now stream, the plan, the quest and the holidays', async () => {
++ const out = await t.ctx.imports.runOnce({ sources: ['live', 'fun', 'holidays'], fetchImpl: fetch, force: true });
++ assert.strictEqual(out.sources.live.ok, true, JSON.stringify(out.sources.live));
++ assert.strictEqual(out.sources.live.imported, 1, 'one valid live stream');
++ assert.strictEqual(out.sources.fun.imported, 2, 'one plan and one accepted quest');
++ assert.strictEqual(out.sources.holidays.imported, 3);
++
++ const r = await t.get(`/api/v1/events?from=2026-01-01T00:00:00Z&to=2027-01-01T00:00:00Z&limit=100`);
++ assert.strictEqual(r.status, 200, r.text);
++ const events = r.json().events;
++ assert.strictEqual(events.length, 6);
++ const bySource = (source, id) => events.find((e) => e.source === source && e.source_id === id);
++
++ const live = bySource('live', 'stream:42');
++ assert.ok(live, 'the live stream row');
++ liveId = live.id;
++ firstSeq = live.seq;
++ assert.strictEqual(live.kind, 'stream');
++ assert.strictEqual(live.status, 'live');
++ assert.strictEqual(live.starts_at, '2026-10-31T11:30:00.000Z', 'SQLite UTC time read as UTC');
++ assert.strictEqual(live.tz, 'UTC');
++ assert.strictEqual(live.channel, 'goosely');
++ assert.strictEqual(live.url, 'https://openvibe.live/@goosely');
++ assert.strictEqual(live.source_url, 'https://openvibe.live/@goosely');
++ assert.strictEqual(live.owner_subject, 'chan:goosely', 'no subject is invented from a public API');
++
++ const plan = bySource('fun', 'plan:berlin-picnic');
++ assert.ok(plan);
++ planId = plan.id;
++ assert.strictEqual(plan.kind, 'plan');
++ assert.strictEqual(plan.starts_at, '2026-10-31T11:00:00.000Z');
++ assert.strictEqual(plan.ends_at, '2026-10-31T13:00:00.000Z', 'two hours, not the voting close');
++ assert.strictEqual(plan.tz, 'Europe/Berlin');
++ assert.strictEqual(plan.url, 'https://openvibe.fun/plans/berlin-picnic');
++ assert.strictEqual(plan.source_url, plan.url);
++
++ const quest = bySource('fun', 'quest:find-the-cat');
++ assert.ok(quest);
++ assert.strictEqual(quest.kind, 'quest');
++ assert.strictEqual(quest.starts_at, '2026-10-31T13:00:00.000Z');
++ assert.strictEqual(quest.url, 'https://openvibe.fun/q/find-the-cat');
++ assert.ok(!bySource('fun', 'quest:already-done'), 'completed quests are history');
++ assert.ok(!bySource('fun', 'quest:hidden-quest'), 'hidden quests never enter');
++
++ const de = events.find((e) => e.source === 'holidays' && e.title === 'Erster Weihnachtstag');
++ assert.ok(de);
++ assert.strictEqual(de.kind, 'holiday');
++ assert.strictEqual(de.tz, 'Europe/Berlin');
++ assert.strictEqual(de.starts_at, '2026-12-24T23:00:00.000Z', 'local midnight in CET');
++ assert.strictEqual(de.ends_at, '2026-12-25T23:00:00.000Z');
++ assert.strictEqual(de.url, 'https://date.nager.at/api/v3/PublicHolidays/2026/DE');
++
++ const summer = events.find((e) => e.source === 'holidays' && e.title === 'Sommerfest');
++ assert.strictEqual(summer.starts_at, '2026-06-02T22:00:00.000Z', 'local midnight in CEST');
++
++ const us = events.find((e) => e.source === 'holidays' && e.title === 'Christmas Day');
++ assert.strictEqual(us.starts_at, '2026-12-25T05:00:00.000Z', 'New York midnight is 05:00 UTC');
++ assert.strictEqual(us.tz, 'America/New_York');
++ });
++
++ await check('a second pass changes nothing: same rows, same seq, no duplicates', async () => {
++ const before = await t.ctx.s.db.value('SELECT COUNT(*)::int AS n FROM today_events');
++ const out = await t.ctx.imports.runOnce({ sources: ['live', 'fun', 'holidays'], fetchImpl: fetch, force: true });
++ assert.strictEqual(out.sources.live.imported, 1);
++ assert.strictEqual(out.sources.fun.imported, 2);
++ assert.strictEqual(out.sources.holidays.imported, 3);
++ const after = await t.ctx.s.db.value('SELECT COUNT(*)::int AS n FROM today_events');
++ assert.strictEqual(after, before);
++ const row = await t.ctx.s.db.maybe('SELECT * FROM today_events WHERE source = $1 AND source_id = $2', ['live', 'stream:42']);
++ assert.strictEqual(row.id, liveId);
++ assert.strictEqual(Number(row.seq), Number(firstSeq), 'an unchanged row does not bump seq');
++ const dupes = await t.ctx.s.db.many('SELECT source, source_id, COUNT(*)::int AS n FROM today_events WHERE source IS NOT NULL GROUP BY 1, 2 HAVING COUNT(*) > 1');
++ assert.deepStrictEqual(dupes, []);
++ });
++
++ await check('a changed source title updates the same row and bumps seq once', async () => {
++ const changed = { streams: [{ ...LIVE.streams[0], title: 'Night walk (part 2)' }] };
++ const out = await t.ctx.imports.runOnce({ sources: ['live'], fetchImpl: fakeFetch({ live: changed }) });
++ assert.strictEqual(out.sources.live.imported, 1);
++ const row = await t.ctx.s.db.maybe('SELECT * FROM today_events WHERE source = $1 AND source_id = $2', ['live', 'stream:42']);
++ assert.strictEqual(row.id, liveId);
++ assert.strictEqual(row.title, 'Night walk (part 2)');
++ assert.strictEqual(Number(row.seq), Number(firstSeq) + 1);
++ assert.strictEqual(await t.ctx.s.db.value('SELECT COUNT(*)::int AS n FROM today_events WHERE source = $1', ['live']), 1);
++ });
++
++ await check('a failing source retires nothing and records the error', async () => {
++ const out = await t.ctx.imports.runOnce({ sources: ['live'], fetchImpl: fakeFetch({ failLive: true }) });
++ assert.strictEqual(out.sources.live.ok, false);
++ assert.match(out.sources.live.error, /restarting/);
++ const row = await t.ctx.s.db.maybe('SELECT * FROM today_events WHERE source = $1 AND source_id = $2', ['live', 'stream:42']);
++ assert.strictEqual(row.status, 'live', 'a failed read must not retire the stream');
++ const source = await t.ctx.s.db.maybe('SELECT * FROM today_import_sources WHERE source = $1', ['live']);
++ assert.match(source.last_error, /restarting/);
++ });
++
++ await check('a stream that is no longer listed is marked done and no longer says on now', async () => {
++ const out = await t.ctx.imports.runOnce({ sources: ['live'], fetchImpl: fakeFetch({ live: { streams: [] } }) });
++ assert.strictEqual(out.sources.live.retired, 1);
++ const row = await t.ctx.s.db.maybe('SELECT * FROM today_events WHERE id = $1', [liveId]);
++ assert.strictEqual(row.status, 'done');
++ // A done stream stays readable as history (calendars need it); it is not live any more.
++ const r = await t.get('/api/v1/events?from=2026-10-31T10:00:00Z&to=2026-10-31T20:00:00Z&limit=100');
++ const served = r.json().events.find((e) => e.id === liveId);
++ assert.ok(served);
++ assert.strictEqual(served.status, 'done');
++ });
++
++ await check('a stream that is live now stays on the public list however long ago it started', async () => {
++ await t.ctx.s.db.exec(
++ `INSERT INTO today_events (id, owner_subject, kind, channel, title, starts_at, tz, status, visibility)
++ VALUES ('tev_01JZ000000000000000000L1', 'chan:long', 'stream', 'long', 'A five-hour stream', $1::timestamptz, 'UTC', 'live', 'public')`,
++ [iso(NOW - 5 * 60 * 60_000)],
++ );
++ const r = await t.get('/api/v1/events?limit=100');
++ const served = r.json().events.find((e) => e.id === 'tev_01JZ000000000000000000L1');
++ assert.ok(served, 'default window still shows the live stream');
++ assert.strictEqual(served.status, 'live');
++ });
++
++ await check('nothing private from a source is stored or served', async () => {
++ const row = await t.ctx.s.db.maybe('SELECT * FROM today_events WHERE id = $1', [planId]);
++ const dump = JSON.stringify(row) + JSON.stringify(await t.get(`/api/v1/events/${planId}`).then((x) => x.json()));
++ assert.ok(!/goose@example\.test/.test(dump), 'no source email');
++ assert.ok(!/live_should_never_be_stored/.test(dump), 'no Live ingest key');
++ assert.ok(!/usr_01JZ0000000000000000000CAR/.test(dump), 'no plan owner subject');
++ assert.strictEqual(row.visibility, 'public');
++ assert.strictEqual(row.owner_subject, 'sys:fun');
++ });
++
++ await check('a follower\'s agenda puts their followed channel first, and later learns its subject', async () => {
++ // The public API only says "goosely"; the bus is what tells Today which usr_ owns it.
++ await t.ctx.s.db.exec(
++ 'INSERT INTO today_channels (username, subject, updated_at) VALUES ($1, $2, $3::timestamptz) ON CONFLICT (username) DO UPDATE SET subject = EXCLUDED.subject',
++ ['goosely', alice.subject, iso(NOW)],
++ );
++ t.network.setFollow(bob, alice.subject);
++ const out = await t.ctx.imports.runOnce({ sources: ['live'], fetchImpl: fakeFetch() });
++ assert.strictEqual(out.sources.live.imported, 1);
++ const row = await t.ctx.s.db.maybe('SELECT * FROM today_events WHERE source = $1 AND source_id = $2', ['live', 'stream:42']);
++ assert.strictEqual(row.status, 'live');
++ assert.strictEqual(row.owner_subject, alice.subject, 'the importer used the channel map');
++ const agenda = await t.get('/api/v1/me/agenda?days=30', { as: bob });
++ assert.strictEqual(agenda.status, 200, agenda.text);
++ const ids = agenda.json().events.map((e) => e.id);
++ assert.ok(ids.includes(row.id), 'the followed channel\'s stream is in the agenda');
++ assert.strictEqual(ids[0], row.id, `followed first, got ${ids.join(', ')}`);
++ });
++
++ await check('the optional Events pull imports a public people-facing event and drops the rest', async () => {
++ const eventsFetch = async (url, opts = {}) => {
++ const u = new URL(url);
++ if (u.pathname === '/oauth/token') {
++ return { ok: true, status: 200, json: async () => ({ access_token: 'tok_events_fixture', token_type: 'Bearer', expires_in: 300 }) };
++ }
++ if (u.pathname === '/api/v1/events') {
++ assert.strictEqual(opts.headers.Authorization, 'Bearer tok_events_fixture');
++ assert.strictEqual(u.searchParams.get('topic'), 'network.meetup.*');
++ return {
++ ok: true, status: 200,
++ json: async () => ({
++ events: [
++ { cursor: 'c1', event: { event_id: 'evt_01JZ000000000000000000M01', event_type: 'network.meetup.scheduled', visibility: 'public', payload: { title: 'Berlin meetup', starts_at: '2026-11-01T18:00:00Z', tz: 'Europe/Berlin', url: 'https://openvibe.network/@meetup' } } },
++ { cursor: 'c2', event: { event_id: 'evt_01JZ000000000000000000M02', event_type: 'live.release.deployed', visibility: 'public', payload: { title: 'v1.2.3', starts_at: '2026-11-01T18:00:00Z' } } },
++ { cursor: 'c3', event: { event_id: 'evt_01JZ000000000000000000M03', event_type: 'network.meetup.scheduled', visibility: 'internal', payload: { title: 'Private', starts_at: '2026-11-01T18:00:00Z' } } },
++ { cursor: 'c4', event: { event_id: 'evt_01JZ000000000000000000M04', event_type: 'network.meetup.scheduled', visibility: 'public', payload: { title: 'No time' } } },
++ ],
++ }),
++ };
++ }
++ throw new Error(`unexpected ${url}`);
++ };
++ const out = await t.ctx.imports.runOnce({ sources: ['events'], fetchImpl: eventsFetch });
++ assert.strictEqual(out.sources.events.ok, true, JSON.stringify(out.sources.events));
++ assert.strictEqual(out.sources.events.imported, 1);
++ const row = await t.ctx.s.db.maybe('SELECT * FROM today_events WHERE source = $1', ['events']);
++ assert.strictEqual(row.title, 'Berlin meetup');
++ assert.strictEqual(row.source_url, 'https://openvibe.network/@meetup');
++ assert.strictEqual(row.owner_subject, 'sys:events');
++ });
++
++ await check('the normalisers refuse non-public and non-people events', async () => {
++ assert.strictEqual(networkEvent({
++ event_id: 'evt_01JZ00000000000000000000', event_type: 'live.release.deployed', visibility: 'public',
++ payload: { title: 'v1.2.3', starts_at: '2026-11-01T12:00:00Z' },
++ }), null, 'a release note is not an event');
++ assert.strictEqual(funQuest({ slug: 'x', title: 'x', status: 'completed' }), null);
++ assert.strictEqual(funQuest({ slug: 'x', title: 'x', status: 'accepted', accepted_at: '2026-11-01T12:00:00Z', visibility: 'hidden' }), null);
++ assert.strictEqual(funPlan({ slug: 'x', title: 'x', starts_at: '2026-11-01T12:00:00Z', visibility: 'unlisted' }), null);
++ const meeting = networkEvent({
++ event_id: 'evt_01JZ00000000000000000001', event_type: 'network.meetup.scheduled', visibility: 'public',
++ payload: { title: 'Berlin meetup', starts_at: '2026-11-01T18:00:00Z', tz: 'Europe/Berlin', url: 'https://openvibe.network/@meetup' },
++ });
++ assert.ok(meeting);
++ assert.strictEqual(meeting.kind, 'event');
++ assert.strictEqual(meeting.source_url, 'https://openvibe.network/@meetup');
++ });
++
++ await check('the job refuses to start under drill', async () => {
++ const { createImports } = require('../server/imports');
++ const imports = createImports({ config: { ...t.config, drill: true }, s: t.ctx.s, fetchImpl: fetch });
++ assert.strictEqual(imports.start(), false);
++ const off = createImports({ config: { ...t.config, imports: { ...t.config.imports, enabled: false } }, s: t.ctx.s, fetchImpl: fetch });
++ assert.strictEqual(off.start(), false);
++ });
++ } finally {
++ await t.close();
++ }
++ done();
++})();