openvibers/openvibe.network Commits
Today's schedules reach followers: today.event.scheduled tells a streamer's followers once (low priority, three schedules a day), today.event.cancelled clears those notices (#395)
Changes
@@@ -292,6 +292,8 @@ their per-category email choice.| `deals.watch.matched` | Deals (`source: deals`) | `payload.recipient` (`usr_…`) | `DEAL_WATCH_MATCH`, `service`, normal || `trade.alert.triggered` | Trade (`source: trade`) | `subject.id` (`subject.type: user`) | `TRADE_ALERT`, `service`, high || `live.stream.started` | Live (`source: live`) | every follower of `payload.channel.subject`, plus everyone with `stream_live_all` on | `STREAM_LIVE`, `stream`, high; the Discord live alert after commit |++| `today.event.scheduled` | Today (`source: today`) | every follower of `subject.id`, for a public or followers-only stream 15 min to 14 days ahead; at most 3 announced schedules per streamer a day | `STREAM_SCHEDULED`, `stream`, low (the bell, no toast); the time in the stream's own zone |++| `today.event.cancelled` | Today | everyone told by the matching `today.event.scheduled` | those notices are dismissed |**Go-lives.** The followers are Network's own follow graph (see "Follow graph"): the active`user_follows` rows for `('channel', payload.channel.subject)`, read in the same transaction as the@@@ -22,6 +22,13 @@* A started event older than LIVE_STARTED_MAX_AGE (30 min; a replay, or Events catching up after an* outage) announces nothing: the stream may be long over.*++ * today.event.scheduled a streamer scheduled a stream on OpenVibe.Today (event.subject is the streamer;++ * payload.event_id the tev_ id). Their followers get one STREAM_SCHEDULED notice++ * each (low priority: the bell, no toast), for a public or followers-only stream++ * starting 15 minutes to 14 days ahead, at most SCHEDULED_PER_DAY announced events++ * per streamer in 24 hours.++ * today.event.cancelled the stream was cancelled: those notices are cleared (dismissed), so the bell never++ * points at a stream that is not happening.* <service>.usage.recorded a developer project's hourly usage rollup from Tools, Events or Media* (common.usage-recorded@1): kept per project and day for its dashboard* (server/developer/usage.js), never anyone's notification.@@@ -55,8 +62,12 @@ const CONSUMER = 'network-notifications';const INBOX_TABLE = 'network_event_inbox';const AUDIT_TOPICS = require('../admin/moderation-audit').TOPICS;const USAGE_TOPICS = require('../developer/usage').TOPICS;--const TOPICS = Object.freeze(['deals.watch.matched', 'coupons.watch.matched', 'trade.alert.triggered', 'live.stream.started', 'live.stream.ended', ...AUDIT_TOPICS, ...USAGE_TOPICS]);++const TOPICS = Object.freeze(['deals.watch.matched', 'coupons.watch.matched', 'trade.alert.triggered', 'live.stream.started', 'live.stream.ended', 'today.event.scheduled', 'today.event.cancelled', ...AUDIT_TOPICS, ...USAGE_TOPICS]);const LIVE_STARTED_MAX_AGE_MS = 30 * 60 * 1000;++const SCHEDULED_PER_DAY = 3;++const SCHEDULE_AHEAD_MIN_MS = 15 * 60 * 1000;++const SCHEDULE_AHEAD_MAX_MS = 14 * 24 * 60 * 60 * 1000;++const TODAY_EVENT_RE = /^tev_[0-9A-Za-z]{10,40}$/;const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;/** Plain, single-line, length-capped text: no markup or control characters reach the inbox. */@@@ -149,6 +160,34 @@ const HANDLERS = {},};++/** "Fri, Oct 30, 8:00 PM PDT" in the event's own zone (UTC when the zone is unknown). */++function whenText(iso, tz) {++ const d = new Date(iso);++ const fmt = (zone) => new Intl.DateTimeFormat('en-US', { timeZone: zone, weekday: 'short', month: 'short', day: 'numeric', hour: 'numeric', minute: '2-digit', timeZoneName: 'short' }).format(d);++ try { return fmt(tz || 'UTC'); } catch { return fmt('UTC'); }++}++++/** The facts of a today.event.scheduled envelope, or an 'ignored:*' outcome. No I/O. */++function todayScheduled(event, { now }) {++ if (event.source !== 'today') return 'ignored:source';++ const p = event.payload && typeof event.payload === 'object' ? event.payload : {};++ const subject = event.subject && event.subject.type === 'user' ? event.subject.id : null;++ if (!ids.isSubjectId('user', subject)) return 'ignored:subject';++ if (!TODAY_EVENT_RE.test(String(p.event_id || ''))) return 'ignored:payload';++ if (p.kind !== 'stream') return 'ignored:kind';++ if (p.status !== 'scheduled') return 'ignored:status';++ if (!['public', 'followers'].includes(p.visibility)) return 'ignored:visibility';++ const starts = Date.parse(p.starts_at || '');++ if (!Number.isFinite(starts)) return 'ignored:payload';++ if (starts - now < SCHEDULE_AHEAD_MIN_MS) return 'ignored:too_soon';++ if (starts - now > SCHEDULE_AHEAD_MAX_MS) return 'ignored:too_far';++ return {++ subject, eventId: String(p.event_id), title: clean(p.title, 120) || 'A stream', channel: clean(p.channel, 40) || null,++ startsAt: new Date(starts).toISOString(), when: whenText(starts, clean(p.tz, 64)),++ url: ownLink(`https://openvibe.today/e/${p.event_id}`) || null,++ };++}++/** The facts of a live.stream.started envelope, or an 'ignored:*' outcome. No I/O. */function liveStarted(event, { now, maxAgeMs }) {if (event.source !== 'live') return 'ignored:source';@@@ -184,6 +223,47 @@ async function createEventsConsumer({ db, notifications, secrets, discord = () =const userBySubject = db.prepare('SELECT id FROM users WHERE subject_id = ?');const streamerBySubject = db.prepare('SELECT id, username, display_name, avatar_url FROM users WHERE subject_id = ?');const followersOf = db.prepare("SELECT follower_subject FROM user_follows WHERE target_type = 'channel' AND target_id = ? AND active = 1");++ // Announced schedules per streamer in the last 24 hours (one notice per follower each, so distinct events).++ const scheduledToday = db.prepare(`SELECT COUNT(DISTINCT (rich_content::jsonb -> 'context' ->> 'today_event_id')) AS n FROM notifications++ WHERE type = 'STREAM_SCHEDULED' AND sender_id = ? AND created_at > ?`);++ const clearScheduled = db.prepare(`UPDATE notifications SET is_dismissed = 1++ WHERE type = 'STREAM_SCHEDULED' AND is_dismissed = 0 AND rich_content::jsonb -> 'context' ->> 'today_event_id' = ?`);++++ /** today.event.scheduled inside the inbox transaction: the daily window, then one notice per follower. */++ async function todayEventScheduled(event) {++ const v = todayScheduled(event, { now: now() });++ if (typeof v === 'string') return v;++ const streamer = await streamerBySubject.get(v.subject);++ if (!streamer) return 'ignored:channel';++ const since = new Date(now() - 24 * 60 * 60 * 1000).toISOString().replace('T', ' ').slice(0, 19);++ const n = Number((await scheduledToday.get(streamer.id, since))?.n) || 0;++ if (n >= SCHEDULED_PER_DAY) return { outcome: 'skipped:daily_limit', detail: { announced_today: n } };++ const followers = await followersOf.all(v.subject);++ const targets = new Set();++ let unresolved = 0;++ for (const f of followers) {++ const u = await userBySubject.get(f.follower_subject);++ if (u) targets.add(u.id); else unresolved++;++ }++ targets.delete(streamer.id);++ const name = streamer.display_name || streamer.username;++ const created = targets.size ? await notifications.createBulk([...targets], {++ type: 'STREAM_SCHEDULED', category: 'stream', priority: 'low', service: 'today',++ title: `${name} scheduled a stream`, message: `${v.title} · ${v.when}`,++ sender_id: streamer.id, sender_name: name, sender_avatar: streamer.avatar_url || null, url: v.url,++ rich_content: { service: 'today', chips: [v.when], context: { event_id: event.event_id, today_event_id: v.eventId, starts_at: v.startsAt, channel: v.channel } },++ }) : [];++ return { outcome: created.length ? 'notified' : targets.size ? 'suppressed:preference' : 'no-recipients', detail: { followers: followers.length, unresolved, targets: targets.size, notified: created.length } };++ }++++ /** today.event.cancelled: the stream's schedule notices leave every bell. */++ async function todayEventCancelled(event) {++ if (event.source !== 'today') return 'ignored:source';++ const id = event.payload && event.payload.event_id;++ if (!TODAY_EVENT_RE.test(String(id || ''))) return 'ignored:payload';++ const r = await clearScheduled.run(String(id));++ return { outcome: 'cleared', detail: { notices: r.changes || 0 } };++ }/** live.stream.started inside the inbox transaction: the announcement window, then one notification per person. */async function liveStreamStarted(event) {@@@ -228,6 +308,10 @@ async function createEventsConsumer({ db, notifications, secrets, discord = () =if (USAGE_TOPICS.includes(event.event_type)) return projectUsage ? projectUsage.record(event) : 'ignored:usage_off';// Creator analytics (WS-E task 6): each ended stream's totals, counts only.if (event.event_type === 'live.stream.ended') return require('../analytics/creators').record(db, event);++ if (event.event_type === 'today.event.scheduled' || event.event_type === 'today.event.cancelled') {++ const out = event.event_type === 'today.event.scheduled' ? await todayEventScheduled(event) : await todayEventCancelled(event);++ return typeof out === 'string' ? out : { outcome: out.outcome, detail: out.detail };++ }if (event.event_type === 'live.stream.started') {const out = await liveStreamStarted(event);if (typeof out === 'string') return out;@@@ -68,7 +68,7 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE(async () => {// The subscriptions this consumer is meant for, and nothing else.-- assert.deepStrictEqual(TOPICS, ['deals.watch.matched', 'coupons.watch.matched', 'trade.alert.triggered', 'live.stream.started', 'live.stream.ended', 'chat.moderation.action', 'live.moderation.action', 'community.moderation.action', 'space.moderation.action', 'tips.interaction.moderated', 'billing.staff.action',++ assert.deepStrictEqual(TOPICS, ['deals.watch.matched', 'coupons.watch.matched', 'trade.alert.triggered', 'live.stream.started', 'live.stream.ended', 'today.event.scheduled', 'today.event.cancelled', 'chat.moderation.action', 'live.moderation.action', 'community.moderation.action', 'space.moderation.action', 'tips.interaction.moderated', 'billing.staff.action',...['media', 'tools', 'games', 'wiki', 'blog', 'news', 'reviews', 'deals', 'coupons', 'trade', 'services'].map((svc) => `${svc}.moderation.action`),'tools.usage.recorded', 'events.usage.recorded', 'media.usage.recorded']);assert.deepStrictEqual(topicsFrom([]), TOPICS, 'scripts/subscribe-events.js subscribes every topic');@@@ -324,6 +324,56 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHEassert.strictEqual((await rows(7)).length, before + 1);}++ // ── today.event.scheduled: the streamer's followers, once, three schedules a day; cancelled clears them ──++ {++ let n = 0;++ const tev = () => `tev_01JTODAYTEST${String(++n).padStart(8, '0')}`;++ const todayEvent = (type, payload = {}, over = {}) => ({++ event_id: ids.newId('event'), event_type: type, version: 1, source: 'today',++ actor: { type: 'user', id: CAROL }, subject: { type: 'user', id: CAROL }, visibility: 'public', priority: 'normal',++ occurred_at: new Date(clock).toISOString(),++ payload: { event_id: tev(), kind: 'stream', channel: 'carol', title: 'Robot <b>build</b> part 2', starts_at: new Date(clock + 2 * 86400000).toISOString(),++ ends_at: null, tz: 'America/Los_Angeles', status: 'scheduled', visibility: 'public', url: null, seq: 0, ...payload },++ ...over,++ });++ const sched = async (uid) => await db.prepare("SELECT * FROM notifications WHERE user_id = ? AND type = 'STREAM_SCHEDULED' AND is_dismissed = 0 ORDER BY seq").all(uid);++ const e1 = todayEvent('today.event.scheduled');++ r = await post(e1, { now: clock });++ assert.strictEqual(r.body.outcome, 'notified', JSON.stringify(r.body));++ assert.deepStrictEqual(r.body.detail, { followers: 4, unresolved: 1, targets: 3, notified: 2 }, 'Dave and Erin; Frank muted streams; the stranger has no account');++ const [d1] = await sched(21);++ assert.strictEqual(d1.title, 'carol scheduled a stream');++ assert.match(d1.message, /^Robot .*build.* part 2 · \w{3}, \w{3} \d{1,2}, \d{1,2}:\d{2} [AP]M P[DS]T$/, 'the time in the stream\'s own zone');++ assert.ok(!/[<>]/.test(d1.message), 'no markup');++ assert.deepStrictEqual([d1.priority, d1.category, d1.service, d1.sender_id], ['low', 'stream', 'today', 20], 'the bell, no toast');++ assert.ok(d1.url === null || d1.url === `https://openvibe.today/e/${e1.payload.event_id}`, 'a link only once openvibe.today is open');++ assert.strictEqual(JSON.parse(d1.rich_content).context.today_event_id, e1.payload.event_id);++ assert.strictEqual((await sched(23)).length, 0, 'preferences apply');++ assert.strictEqual((await post(e1, { now: clock })).body.duplicate, true, 'a redelivery tells nobody again');++++ // Ignored: another source, a plain event, a stream too soon or too far, a cancelled status, a private one.++ assert.strictEqual((await post(todayEvent('today.event.scheduled', {}, { source: 'live' }), { now: clock })).body.outcome, 'ignored:source');++ assert.strictEqual((await post(todayEvent('today.event.scheduled', { kind: 'event' }), { now: clock })).body.outcome, 'ignored:kind');++ assert.strictEqual((await post(todayEvent('today.event.scheduled', { starts_at: new Date(clock + 5 * 60000).toISOString() }), { now: clock })).body.outcome, 'ignored:too_soon');++ assert.strictEqual((await post(todayEvent('today.event.scheduled', { starts_at: new Date(clock + 30 * 86400000).toISOString() }), { now: clock })).body.outcome, 'ignored:too_far');++ assert.strictEqual((await post(todayEvent('today.event.scheduled', { visibility: 'private' }), { now: clock })).body.outcome, 'ignored:visibility');++ assert.strictEqual((await post(todayEvent('today.event.scheduled', { event_id: 'evt_x' }), { now: clock })).body.outcome, 'ignored:payload');++++ // Three announced schedules a day per streamer; the fourth tells nobody.++ assert.strictEqual((await post(todayEvent('today.event.scheduled'), { now: clock })).body.outcome, 'notified');++ assert.strictEqual((await post(todayEvent('today.event.scheduled'), { now: clock })).body.outcome, 'notified');++ const fourth = await post(todayEvent('today.event.scheduled'), { now: clock });++ assert.deepStrictEqual([fourth.body.outcome, fourth.body.detail], ['skipped:daily_limit', { announced_today: 3 }]);++ assert.strictEqual((await sched(21)).length, 3);++++ // Cancelled: the stream's notices leave every bell; the others stay.++ const c = await post(todayEvent('today.event.cancelled', { event_id: e1.payload.event_id, status: 'cancelled' }), { now: clock });++ assert.deepStrictEqual([c.body.outcome, c.body.detail], ['cleared', { notices: 2 }]);++ assert.ok(!(await sched(21)).some((x) => JSON.parse(x.rich_content).context.today_event_id === e1.payload.event_id));++ assert.strictEqual((await sched(21)).length, 2);++ assert.strictEqual((await post(todayEvent('today.event.cancelled', { event_id: e1.payload.event_id }, { source: 'tips' }), { now: clock })).body.outcome, 'ignored:source');++ }++srv.close();console.log('events consumer: all checks passed');})().catch((err) => { console.error(err); process.exit(1); });
@@ -292,6 +292,8 @@ their per-category email choice.
@@ -292,6 +292,8 @@ their per-category email choice.
| `deals.watch.matched` | Deals (`source: deals`) | `payload.recipient` (`usr_…`) | `DEAL_WATCH_MATCH`, `service`, normal |
| `deals.watch.matched` | Deals (`source: deals`) | `payload.recipient` (`usr_…`) | `DEAL_WATCH_MATCH`, `service`, normal |
| `trade.alert.triggered` | Trade (`source: trade`) | `subject.id` (`subject.type: user`) | `TRADE_ALERT`, `service`, high |
| `trade.alert.triggered` | Trade (`source: trade`) | `subject.id` (`subject.type: user`) | `TRADE_ALERT`, `service`, high |
| `live.stream.started` | Live (`source: live`) | every follower of `payload.channel.subject`, plus everyone with `stream_live_all` on | `STREAM_LIVE`, `stream`, high; the Discord live alert after commit |
| `live.stream.started` | Live (`source: live`) | every follower of `payload.channel.subject`, plus everyone with `stream_live_all` on | `STREAM_LIVE`, `stream`, high; the Discord live alert after commit |
+| `today.event.scheduled` | Today (`source: today`) | every follower of `subject.id`, for a public or followers-only stream 15 min to 14 days ahead; at most 3 announced schedules per streamer a day | `STREAM_SCHEDULED`, `stream`, low (the bell, no toast); the time in the stream's own zone |
+| `today.event.cancelled` | Today | everyone told by the matching `today.event.scheduled` | those notices are dismissed |
**Go-lives.** The followers are Network's own follow graph (see "Follow graph"): the active
**Go-lives.** The followers are Network's own follow graph (see "Follow graph"): the active
`user_follows` rows for `('channel', payload.channel.subject)`, read in the same transaction as the
`user_follows` rows for `('channel', payload.channel.subject)`, read in the same transaction as the
@@ -22,6 +22,13 @@
@@ -22,6 +22,13 @@
* A started event older than LIVE_STARTED_MAX_AGE (30 min; a replay, or Events catching up after an
* A started event older than LIVE_STARTED_MAX_AGE (30 min; a replay, or Events catching up after an
* outage) announces nothing: the stream may be long over.
* outage) announces nothing: the stream may be long over.
*
*
+ * today.event.scheduled a streamer scheduled a stream on OpenVibe.Today (event.subject is the streamer;
+ * payload.event_id the tev_ id). Their followers get one STREAM_SCHEDULED notice
+ * each (low priority: the bell, no toast), for a public or followers-only stream
+ * starting 15 minutes to 14 days ahead, at most SCHEDULED_PER_DAY announced events
+ * per streamer in 24 hours.
+ * today.event.cancelled the stream was cancelled: those notices are cleared (dismissed), so the bell never
+ * points at a stream that is not happening.
* <service>.usage.recorded a developer project's hourly usage rollup from Tools, Events or Media
* <service>.usage.recorded a developer project's hourly usage rollup from Tools, Events or Media
* (common.usage-recorded@1): kept per project and day for its dashboard
* (common.usage-recorded@1): kept per project and day for its dashboard
* (server/developer/usage.js), never anyone's notification.
* (server/developer/usage.js), never anyone's notification.
@@ -55,8 +62,12 @@ const CONSUMER = 'network-notifications';
@@ -55,8 +62,12 @@ const CONSUMER = 'network-notifications';
const INBOX_TABLE = 'network_event_inbox';
const INBOX_TABLE = 'network_event_inbox';
const AUDIT_TOPICS = require('../admin/moderation-audit').TOPICS;
const AUDIT_TOPICS = require('../admin/moderation-audit').TOPICS;
const USAGE_TOPICS = require('../developer/usage').TOPICS;
const USAGE_TOPICS = require('../developer/usage').TOPICS;
-const TOPICS = Object.freeze(['deals.watch.matched', 'coupons.watch.matched', 'trade.alert.triggered', 'live.stream.started', 'live.stream.ended', ...AUDIT_TOPICS, ...USAGE_TOPICS]);
+const TOPICS = Object.freeze(['deals.watch.matched', 'coupons.watch.matched', 'trade.alert.triggered', 'live.stream.started', 'live.stream.ended', 'today.event.scheduled', 'today.event.cancelled', ...AUDIT_TOPICS, ...USAGE_TOPICS]);
const LIVE_STARTED_MAX_AGE_MS = 30 * 60 * 1000;
const LIVE_STARTED_MAX_AGE_MS = 30 * 60 * 1000;
+const SCHEDULED_PER_DAY = 3;
+const SCHEDULE_AHEAD_MIN_MS = 15 * 60 * 1000;
+const SCHEDULE_AHEAD_MAX_MS = 14 * 24 * 60 * 60 * 1000;
+const TODAY_EVENT_RE = /^tev_[0-9A-Za-z]{10,40}$/;
const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
/** Plain, single-line, length-capped text: no markup or control characters reach the inbox. */
/** Plain, single-line, length-capped text: no markup or control characters reach the inbox. */
@@ -149,6 +160,34 @@ const HANDLERS = {
@@ -149,6 +160,34 @@ const HANDLERS = {
},
},
};
};
+/** "Fri, Oct 30, 8:00 PM PDT" in the event's own zone (UTC when the zone is unknown). */
+function whenText(iso, tz) {
+ const d = new Date(iso);
+ const fmt = (zone) => new Intl.DateTimeFormat('en-US', { timeZone: zone, weekday: 'short', month: 'short', day: 'numeric', hour: 'numeric', minute: '2-digit', timeZoneName: 'short' }).format(d);
+ try { return fmt(tz || 'UTC'); } catch { return fmt('UTC'); }
+}
+
+/** The facts of a today.event.scheduled envelope, or an 'ignored:*' outcome. No I/O. */
+function todayScheduled(event, { now }) {
+ if (event.source !== 'today') return 'ignored:source';
+ const p = event.payload && typeof event.payload === 'object' ? event.payload : {};
+ const subject = event.subject && event.subject.type === 'user' ? event.subject.id : null;
+ if (!ids.isSubjectId('user', subject)) return 'ignored:subject';
+ if (!TODAY_EVENT_RE.test(String(p.event_id || ''))) return 'ignored:payload';
+ if (p.kind !== 'stream') return 'ignored:kind';
+ if (p.status !== 'scheduled') return 'ignored:status';
+ if (!['public', 'followers'].includes(p.visibility)) return 'ignored:visibility';
+ const starts = Date.parse(p.starts_at || '');
+ if (!Number.isFinite(starts)) return 'ignored:payload';
+ if (starts - now < SCHEDULE_AHEAD_MIN_MS) return 'ignored:too_soon';
+ if (starts - now > SCHEDULE_AHEAD_MAX_MS) return 'ignored:too_far';
+ return {
+ subject, eventId: String(p.event_id), title: clean(p.title, 120) || 'A stream', channel: clean(p.channel, 40) || null,
+ startsAt: new Date(starts).toISOString(), when: whenText(starts, clean(p.tz, 64)),
+ url: ownLink(`https://openvibe.today/e/${p.event_id}`) || null,
+ };
+}
+
/** The facts of a live.stream.started envelope, or an 'ignored:*' outcome. No I/O. */
/** The facts of a live.stream.started envelope, or an 'ignored:*' outcome. No I/O. */
function liveStarted(event, { now, maxAgeMs }) {
function liveStarted(event, { now, maxAgeMs }) {
if (event.source !== 'live') return 'ignored:source';
if (event.source !== 'live') return 'ignored:source';
@@ -184,6 +223,47 @@ async function createEventsConsumer({ db, notifications, secrets, discord = () =
@@ -184,6 +223,47 @@ async function createEventsConsumer({ db, notifications, secrets, discord = () =
const userBySubject = db.prepare('SELECT id FROM users WHERE subject_id = ?');
const userBySubject = db.prepare('SELECT id FROM users WHERE subject_id = ?');
const streamerBySubject = db.prepare('SELECT id, username, display_name, avatar_url FROM users WHERE subject_id = ?');
const streamerBySubject = db.prepare('SELECT id, username, display_name, avatar_url FROM users WHERE subject_id = ?');
const followersOf = db.prepare("SELECT follower_subject FROM user_follows WHERE target_type = 'channel' AND target_id = ? AND active = 1");
const followersOf = db.prepare("SELECT follower_subject FROM user_follows WHERE target_type = 'channel' AND target_id = ? AND active = 1");
+ // Announced schedules per streamer in the last 24 hours (one notice per follower each, so distinct events).
+ const scheduledToday = db.prepare(`SELECT COUNT(DISTINCT (rich_content::jsonb -> 'context' ->> 'today_event_id')) AS n FROM notifications
+ WHERE type = 'STREAM_SCHEDULED' AND sender_id = ? AND created_at > ?`);
+ const clearScheduled = db.prepare(`UPDATE notifications SET is_dismissed = 1
+ WHERE type = 'STREAM_SCHEDULED' AND is_dismissed = 0 AND rich_content::jsonb -> 'context' ->> 'today_event_id' = ?`);
+
+ /** today.event.scheduled inside the inbox transaction: the daily window, then one notice per follower. */
+ async function todayEventScheduled(event) {
+ const v = todayScheduled(event, { now: now() });
+ if (typeof v === 'string') return v;
+ const streamer = await streamerBySubject.get(v.subject);
+ if (!streamer) return 'ignored:channel';
+ const since = new Date(now() - 24 * 60 * 60 * 1000).toISOString().replace('T', ' ').slice(0, 19);
+ const n = Number((await scheduledToday.get(streamer.id, since))?.n) || 0;
+ if (n >= SCHEDULED_PER_DAY) return { outcome: 'skipped:daily_limit', detail: { announced_today: n } };
+ const followers = await followersOf.all(v.subject);
+ const targets = new Set();
+ let unresolved = 0;
+ for (const f of followers) {
+ const u = await userBySubject.get(f.follower_subject);
+ if (u) targets.add(u.id); else unresolved++;
+ }
+ targets.delete(streamer.id);
+ const name = streamer.display_name || streamer.username;
+ const created = targets.size ? await notifications.createBulk([...targets], {
+ type: 'STREAM_SCHEDULED', category: 'stream', priority: 'low', service: 'today',
+ title: `${name} scheduled a stream`, message: `${v.title} · ${v.when}`,
+ sender_id: streamer.id, sender_name: name, sender_avatar: streamer.avatar_url || null, url: v.url,
+ rich_content: { service: 'today', chips: [v.when], context: { event_id: event.event_id, today_event_id: v.eventId, starts_at: v.startsAt, channel: v.channel } },
+ }) : [];
+ return { outcome: created.length ? 'notified' : targets.size ? 'suppressed:preference' : 'no-recipients', detail: { followers: followers.length, unresolved, targets: targets.size, notified: created.length } };
+ }
+
+ /** today.event.cancelled: the stream's schedule notices leave every bell. */
+ async function todayEventCancelled(event) {
+ if (event.source !== 'today') return 'ignored:source';
+ const id = event.payload && event.payload.event_id;
+ if (!TODAY_EVENT_RE.test(String(id || ''))) return 'ignored:payload';
+ const r = await clearScheduled.run(String(id));
+ return { outcome: 'cleared', detail: { notices: r.changes || 0 } };
+ }
/** live.stream.started inside the inbox transaction: the announcement window, then one notification per person. */
/** live.stream.started inside the inbox transaction: the announcement window, then one notification per person. */
async function liveStreamStarted(event) {
async function liveStreamStarted(event) {
@@ -228,6 +308,10 @@ async function createEventsConsumer({ db, notifications, secrets, discord = () =
@@ -228,6 +308,10 @@ async function createEventsConsumer({ db, notifications, secrets, discord = () =
if (USAGE_TOPICS.includes(event.event_type)) return projectUsage ? projectUsage.record(event) : 'ignored:usage_off';
if (USAGE_TOPICS.includes(event.event_type)) return projectUsage ? projectUsage.record(event) : 'ignored:usage_off';
// Creator analytics (WS-E task 6): each ended stream's totals, counts only.
// Creator analytics (WS-E task 6): each ended stream's totals, counts only.
if (event.event_type === 'live.stream.ended') return require('../analytics/creators').record(db, event);
if (event.event_type === 'live.stream.ended') return require('../analytics/creators').record(db, event);
+ if (event.event_type === 'today.event.scheduled' || event.event_type === 'today.event.cancelled') {
+ const out = event.event_type === 'today.event.scheduled' ? await todayEventScheduled(event) : await todayEventCancelled(event);
+ return typeof out === 'string' ? out : { outcome: out.outcome, detail: out.detail };
+ }
if (event.event_type === 'live.stream.started') {
if (event.event_type === 'live.stream.started') {
const out = await liveStreamStarted(event);
const out = await liveStreamStarted(event);
if (typeof out === 'string') return out;
if (typeof out === 'string') return out;
@@ -68,7 +68,7 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE
@@ -68,7 +68,7 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE
(async () => {
(async () => {
// The subscriptions this consumer is meant for, and nothing else.
// The subscriptions this consumer is meant for, and nothing else.
- assert.deepStrictEqual(TOPICS, ['deals.watch.matched', 'coupons.watch.matched', 'trade.alert.triggered', 'live.stream.started', 'live.stream.ended', 'chat.moderation.action', 'live.moderation.action', 'community.moderation.action', 'space.moderation.action', 'tips.interaction.moderated', 'billing.staff.action',
+ assert.deepStrictEqual(TOPICS, ['deals.watch.matched', 'coupons.watch.matched', 'trade.alert.triggered', 'live.stream.started', 'live.stream.ended', 'today.event.scheduled', 'today.event.cancelled', 'chat.moderation.action', 'live.moderation.action', 'community.moderation.action', 'space.moderation.action', 'tips.interaction.moderated', 'billing.staff.action',
...['media', 'tools', 'games', 'wiki', 'blog', 'news', 'reviews', 'deals', 'coupons', 'trade', 'services'].map((svc) => `${svc}.moderation.action`),
...['media', 'tools', 'games', 'wiki', 'blog', 'news', 'reviews', 'deals', 'coupons', 'trade', 'services'].map((svc) => `${svc}.moderation.action`),
'tools.usage.recorded', 'events.usage.recorded', 'media.usage.recorded']);
'tools.usage.recorded', 'events.usage.recorded', 'media.usage.recorded']);
assert.deepStrictEqual(topicsFrom([]), TOPICS, 'scripts/subscribe-events.js subscribes every topic');
assert.deepStrictEqual(topicsFrom([]), TOPICS, 'scripts/subscribe-events.js subscribes every topic');
@@ -324,6 +324,56 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE
@@ -324,6 +324,56 @@ const rows = async (userId) => await db.prepare('SELECT * FROM notifications WHE
assert.strictEqual((await rows(7)).length, before + 1);
assert.strictEqual((await rows(7)).length, before + 1);
}
}
+ // ── today.event.scheduled: the streamer's followers, once, three schedules a day; cancelled clears them ──
+ {
+ let n = 0;
+ const tev = () => `tev_01JTODAYTEST${String(++n).padStart(8, '0')}`;
+ const todayEvent = (type, payload = {}, over = {}) => ({
+ event_id: ids.newId('event'), event_type: type, version: 1, source: 'today',
+ actor: { type: 'user', id: CAROL }, subject: { type: 'user', id: CAROL }, visibility: 'public', priority: 'normal',
+ occurred_at: new Date(clock).toISOString(),
+ payload: { event_id: tev(), kind: 'stream', channel: 'carol', title: 'Robot <b>build</b> part 2', starts_at: new Date(clock + 2 * 86400000).toISOString(),
+ ends_at: null, tz: 'America/Los_Angeles', status: 'scheduled', visibility: 'public', url: null, seq: 0, ...payload },
+ ...over,
+ });
+ const sched = async (uid) => await db.prepare("SELECT * FROM notifications WHERE user_id = ? AND type = 'STREAM_SCHEDULED' AND is_dismissed = 0 ORDER BY seq").all(uid);
+ const e1 = todayEvent('today.event.scheduled');
+ r = await post(e1, { now: clock });
+ assert.strictEqual(r.body.outcome, 'notified', JSON.stringify(r.body));
+ assert.deepStrictEqual(r.body.detail, { followers: 4, unresolved: 1, targets: 3, notified: 2 }, 'Dave and Erin; Frank muted streams; the stranger has no account');
+ const [d1] = await sched(21);
+ assert.strictEqual(d1.title, 'carol scheduled a stream');
+ assert.match(d1.message, /^Robot .*build.* part 2 · \w{3}, \w{3} \d{1,2}, \d{1,2}:\d{2} [AP]M P[DS]T$/, 'the time in the stream\'s own zone');
+ assert.ok(!/[<>]/.test(d1.message), 'no markup');
+ assert.deepStrictEqual([d1.priority, d1.category, d1.service, d1.sender_id], ['low', 'stream', 'today', 20], 'the bell, no toast');
+ assert.ok(d1.url === null || d1.url === `https://openvibe.today/e/${e1.payload.event_id}`, 'a link only once openvibe.today is open');
+ assert.strictEqual(JSON.parse(d1.rich_content).context.today_event_id, e1.payload.event_id);
+ assert.strictEqual((await sched(23)).length, 0, 'preferences apply');
+ assert.strictEqual((await post(e1, { now: clock })).body.duplicate, true, 'a redelivery tells nobody again');
+
+ // Ignored: another source, a plain event, a stream too soon or too far, a cancelled status, a private one.
+ assert.strictEqual((await post(todayEvent('today.event.scheduled', {}, { source: 'live' }), { now: clock })).body.outcome, 'ignored:source');
+ assert.strictEqual((await post(todayEvent('today.event.scheduled', { kind: 'event' }), { now: clock })).body.outcome, 'ignored:kind');
+ assert.strictEqual((await post(todayEvent('today.event.scheduled', { starts_at: new Date(clock + 5 * 60000).toISOString() }), { now: clock })).body.outcome, 'ignored:too_soon');
+ assert.strictEqual((await post(todayEvent('today.event.scheduled', { starts_at: new Date(clock + 30 * 86400000).toISOString() }), { now: clock })).body.outcome, 'ignored:too_far');
+ assert.strictEqual((await post(todayEvent('today.event.scheduled', { visibility: 'private' }), { now: clock })).body.outcome, 'ignored:visibility');
+ assert.strictEqual((await post(todayEvent('today.event.scheduled', { event_id: 'evt_x' }), { now: clock })).body.outcome, 'ignored:payload');
+
+ // Three announced schedules a day per streamer; the fourth tells nobody.
+ assert.strictEqual((await post(todayEvent('today.event.scheduled'), { now: clock })).body.outcome, 'notified');
+ assert.strictEqual((await post(todayEvent('today.event.scheduled'), { now: clock })).body.outcome, 'notified');
+ const fourth = await post(todayEvent('today.event.scheduled'), { now: clock });
+ assert.deepStrictEqual([fourth.body.outcome, fourth.body.detail], ['skipped:daily_limit', { announced_today: 3 }]);
+ assert.strictEqual((await sched(21)).length, 3);
+
+ // Cancelled: the stream's notices leave every bell; the others stay.
+ const c = await post(todayEvent('today.event.cancelled', { event_id: e1.payload.event_id, status: 'cancelled' }), { now: clock });
+ assert.deepStrictEqual([c.body.outcome, c.body.detail], ['cleared', { notices: 2 }]);
+ assert.ok(!(await sched(21)).some((x) => JSON.parse(x.rich_content).context.today_event_id === e1.payload.event_id));
+ assert.strictEqual((await sched(21)).length, 2);
+ assert.strictEqual((await post(todayEvent('today.event.cancelled', { event_id: e1.payload.event_id }, { source: 'tips' }), { now: clock })).body.outcome, 'ignored:source');
+ }
+
srv.close();
srv.close();
console.log('events consumer: all checks passed');
console.log('events consumer: all checks passed');
})().catch((err) => { console.error(err); process.exit(1); });
})().catch((err) => { console.error(err); process.exit(1); });