openvibers/openvibe.network Commits
T2: async-conversion fixes the PGlite suite could not see; async route rejections answer 500
Changes
@@@ -230,7 +230,7 @@ function createAdminRoutes(db, notificationService, emailService, requireAuth) {// GET /api/admin/email — get current email config (owner-only: exposes key metadata)router.get('/email', requireOwner, async (req, res) => {try {-- const status = emailService.getStatus();++ const status = await emailService.getStatus();res.json({ ok: true, email: status, metrics: await getEmailMetrics() });} catch (err) {res.status(500).json({ ok: false, error: err.message });@@@ -265,14 +265,14 @@ function createAdminRoutes(db, notificationService, emailService, requireAuth) {await collect();if (Object.keys(changes.set).length) await (await siteConfig.forDb(db)).change(changes, { actor: siteConfig.actorOf(req.user), reason: String(req.body.reason || 'email settings').slice(0, 300) });-- emailService.reload();++ await emailService.reload();// Auditawait db.prepare('INSERT INTO audit_log (user_id, action, details) VALUES (?, ?, ?)').run(req.user.id, 'email_config_update', JSON.stringify({ from_email }));-- res.json({ ok: true, email: emailService.getStatus(), ...(skipped.length ? { skipped, note: 'set in the environment; not saved to the database' } : {}) });++ res.json({ ok: true, email: await emailService.getStatus(), ...(skipped.length ? { skipped, note: 'set in the environment; not saved to the database' } : {}) });} catch (err) {if (err && err.status && err.code) return res.status(err.status).json({ ok: false, error: err.message, code: err.code, errors: err.errors });res.status(500).json({ ok: false, error: err.message });@@@ -591,7 +591,7 @@ function createAdminRoutes(db, notificationService, emailService, requireAuth) {if (search) {users = await db.prepare(`SELECT id, username, display_name, email, role, is_banned, created_at, last_seen-- FROM users WHERE username LIKE ? OR display_name LIKE ? OR email LIKE ?++ FROM users WHERE username ILIKE ? OR display_name ILIKE ? OR email ILIKE ?ORDER BY created_at DESC LIMIT ? OFFSET ?`).all(`%${search}%`, `%${search}%`, `%${search}%`, limit, offset);} else {@@@ -847,7 +847,7 @@ function createAdminRoutes(db, notificationService, emailService, requireAuth) {const unreadCount = (await db.prepare('SELECT COUNT(*) as cnt FROM notifications WHERE is_read = 0').get()).cnt;const anonCount = (await db.prepare('SELECT COUNT(*) as cnt FROM anon_users').get()).cnt;const sessionCount = (await db.prepare('SELECT COUNT(*) as cnt FROM user_sessions WHERE is_active = 1').get()).cnt;-- const emailStatus = emailService.getStatus();++ const emailStatus = await emailService.getStatus();res.json({ok: true,@@@ -47,10 +47,11 @@ function createStaffApi({ db, requireAuth, guard }) {const bearer = String(req.headers.authorization || '').replace(/^Bearer\s+/i, '');let sub = '';try { sub = JSON.parse(Buffer.from(bearer.split('.')[1] || '', 'base64url').toString('utf8')).sub || ''; } catch { /* not a JWT */ }-- if (String(sub).startsWith('svc:')) return serviceGuard(req, res, async () => await list(req, res));-- return requireAuth(req, res, async () => {++ // The guards call back after this handler returned: a failure in list() goes to next(err).++ if (String(sub).startsWith('svc:')) return serviceGuard(req, res, () => list(req, res).catch(next));++ return requireAuth(req, res, () => {if (!staff.can(staffClaims(req.user), 'staff.moderation.logs')) return res.status(403).json({ error: 'forbidden', detail: 'staff.moderation.logs required' });-- return await list(req, res);++ return list(req, res).catch(next);});});return r;@@@ -101,7 +101,7 @@ async function viewerOf(req) {function router({ fullGuard }) {const r = express.Router();r.use(http.middleware());-- r.get('/:creator/analytics', async (req, res) => {++ r.get('/:creator/analytics', async (req, res, next) => {const db = req.app.locals.db;const key = String(req.params.creator || '');const u = SUBJECT_RE.test(key) ? await db.prepare('SELECT subject_id, is_anon FROM users WHERE subject_id = ?').get(key)@@@ -111,7 +111,8 @@ function router({ fullGuard }) {const viewer = await viewerOf(req);if (viewer) return await answer(viewer.subject_id === u.subject_id);const h = String(req.headers.authorization || '');-- if (h.startsWith('Bearer ')) return fullGuard(req, res, async () => await answer(true)); // a service token: full, or refused++ // The guard calls back after this handler returned: a failure there goes to next(err) (server/async-routes.js).++ if (h.startsWith('Bearer ')) return fullGuard(req, res, () => answer(true).catch(next)); // a service token: full, or refusedreturn await answer(false);});return r;nnew file mode 100644@@@ -0,0 +1,50 @@++'use strict';++++// ═══════════════════════════════════════════════════════════════++// Async route handlers on Express 4 (plan T2).++//++// Every query is async on PostgreSQL, so most handlers are async functions. Express 4 ignores what a++// handler returns: a rejected promise (a failed query, say) was an unhandled rejection, which ends a++// Node 22 process, and the request hung. With this installed a rejection goes to next(err), exactly++// where a synchronous throw went before (Express's own error handling answers 500), as Express 5 does.++// Installed once, before the first request, by server/index.js.++// ═══════════════════════════════════════════════════════════════++++const Layer = require('express/lib/router/layer');++++const INSTALLED = Symbol.for('openvibe.network.async-routes');++++function forward(next) {++ return (err) => next(err || new Error('the route handler rejected without a reason'));++}++++function install() {++ if (Layer.prototype[INSTALLED]) return;++ Layer.prototype[INSTALLED] = true;++++ Layer.prototype.handle_request = function handle(req, res, next) {++ const fn = this.handle;++ if (fn.length > 3) return next(); // an error handler: not for a request++ try {++ const r = fn(req, res, next);++ if (r && typeof r.then === 'function') r.then(undefined, forward(next));++ } catch (err) {++ next(err);++ }++ };++++ Layer.prototype.handle_error = function handleError(error, req, res, next) {++ const fn = this.handle;++ if (fn.length !== 4) return next(error); // not an error handler++ try {++ const r = fn(error, req, res, next);++ if (r && typeof r.then === 'function') r.then(undefined, forward(next));++ } catch (err) {++ next(err);++ }++ };++}++++install();++++module.exports = { install };@@@ -128,7 +128,7 @@ function routes(requireAuth) {const r = await consumeToken(db, String(req.body?.token || ''));if (!r.ok) return res.status(400).json({ ok: false, error: r.error });try {-- req.app.locals.notificationService?.create({ user_id: r.user.id, type: 'EMAIL_VERIFIED', title: 'Email verified', message: `${r.user.email} is confirmed. Go-live alerts from streamers you follow will now reach your inbox.`, service: 'network', url: 'https://openvibe.network/notifications' });++ await req.app.locals.notificationService?.create({ user_id: r.user.id, type: 'EMAIL_VERIFIED', title: 'Email verified', message: `${r.user.email} is confirmed. Go-live alerts from streamers you follow will now reach your inbox.`, service: 'network', url: 'https://openvibe.network/notifications' });} catch { /* */ }res.json({ ok: true, email: r.user.email });});@@@ -15,7 +15,7 @@ try {class DiscordService {/**-- * @param {object} db — better-sqlite3 Database instance (with getSetting helper)++ * @param {object} db — the openvibe-sdk/db handle (with the async getSetting helper)*/constructor(db) {this._db = db;@@@ -36,7 +36,7 @@ class DiscordService {async init() {if (!Discord) return;-- const token = this._getSetting('discord_bot_token');++ const token = await this._getSetting('discord_bot_token');if (!token) {console.log('[Discord] No bot token configured — bot disabled');return;@@@ -87,13 +87,13 @@ class DiscordService {// ─── Status ────────────────────────────────────────────-- getStatus() {++ async getStatus() {return {available: !!Discord,connected: this._ready,botTag: this._client?.user?.tag || null,-- guildId: this._getSetting('discord_guild_id') || null,-- alertsChannelId: this._getSetting('discord_alerts_channel_id') || null,++ guildId: (await this._getSetting('discord_guild_id')) || null,++ alertsChannelId: (await this._getSetting('discord_alerts_channel_id')) || null,};}@@@ -114,14 +114,14 @@ class DiscordService {async sendLiveAlert(streamer, stream) {if (!this.isReady()) return { sent: false, reason: 'bot_not_connected' };-- const channelId = this._getSetting('discord_alerts_channel_id');++ const channelId = await this._getSetting('discord_alerts_channel_id');if (!channelId) return { sent: false, reason: 'no_alerts_channel' };// Dedupe check — one alert per streamer per cooldown window-- const cooldownMin = parseInt(this._getSetting('discord_dedupe_minutes') || '15', 10);++ const cooldownMin = parseInt((await this._getSetting('discord_dedupe_minutes')) || '15', 10);const cooldownMs = Math.max(cooldownMin, 1) * 60 * 1000;const key = streamer.username.toLowerCase();-- const lastAlert = await this._liveAlertCooldowns.get(key);++ const lastAlert = this._liveAlertCooldowns.get(key);if (lastAlert && (Date.now() - lastAlert) < cooldownMs) {return { sent: false, reason: 'cooldown', remaining_ms: cooldownMs - (Date.now() - lastAlert) };}@@@ -153,7 +153,7 @@ class DiscordService {}// Custom message template support-- const customTemplate = this._getSetting('discord_alert_message');++ const customTemplate = await this._getSetting('discord_alert_message');let content = null;if (customTemplate) {content = customTemplate@@@ -182,7 +182,7 @@ class DiscordService {async sendSystemAlert(alert) {if (!this.isReady()) return { sent: false, reason: 'bot_not_connected' };-- const channelId = this._getSetting('discord_system_channel_id');++ const channelId = await this._getSetting('discord_system_channel_id');if (!channelId) return { sent: false, reason: 'no_system_channel' };const colors = { info: 0x3b82f6, warning: 0xf59e0b, error: 0xef4444 };@@@ -16,7 +16,7 @@ module.exports = function createDiscordRoutes(db, discordService, requireAuth, r/** GET /api/admin/discord — bot status + current config */router.get('/', async (req, res) => {-- const status = discordService.getStatus();++ const status = await discordService.getStatus();const settings = {discord_bot_token: await db.getSetting('discord_bot_token') ? '••••••••' : '',discord_guild_id: await db.getSetting('discord_guild_id') || '',@@@ -85,7 +85,7 @@ module.exports = function createDiscordRoutes(db, discordService, requireAuth, r}}-- res.json({ ok: true, status: discordService.getStatus(), ...(skipped.length ? { skipped, note: 'set in the environment; not saved to the database' } : {}) });++ res.json({ ok: true, status: await discordService.getStatus(), ...(skipped.length ? { skipped, note: 'set in the environment; not saved to the database' } : {}) });});/** POST /api/admin/discord/test — send a test alert to the configured channel */@@@ -111,7 +111,7 @@ module.exports = function createDiscordRoutes(db, discordService, requireAuth, rrouter.post('/reinit', async (req, res) => {try {await discordService.reinit();-- res.json({ ok: true, status: discordService.getStatus() });++ res.json({ ok: true, status: await discordService.getStatus() });} catch (err) {res.status(500).json({ ok: false, error: err.message });}@@@ -295,7 +295,7 @@ function routers({ requireAuth, followsGuard }) {const viewer = await viewerOf(req);if (viewer && viewer.subject_id === t.subject_id) return await answer();if (viewer) return http.sendProblem(res, 403, 'follows.not_yours', { detail: 'only the channel owner sees who follows it', ctx: req.ov });-- return followsGuard(req, res, answer);++ return followsGuard(req, res, () => answer().catch(next)); // called back after this handler returned});// ADR-030 step 4: a first-party product (network.follows.write, checked by the mount) records a follow@@@ -133,7 +133,7 @@ async function write(db, subjectId, namespace, data, { writer, expectedRevision,.run(subjectId, namespace, ns.version, revision, JSON.stringify(data), by);const record = await read(db, subjectId, namespace);assertValid('modules.module-record@1', record);-- moduleEvents.record(db, {++ await moduleEvents.record(db, {subjectId, namespace, namespaceOwner: ownerOf(namespace), schemaVersion: ns.version, revision,change: cur ? 'updated' : 'created', reason: 'write', before: cur ? JSON.parse(cur.data) : {}, after: data,actor: actorOf(subjectId, writer), ctx,@@@ -153,7 +153,7 @@ async function removeRow(db, subjectId, namespace, { actor, reason, mergedInto,if (!row) return 0;await db.prepare('DELETE FROM user_modules WHERE subject_id = ? AND namespace = ?').run(subjectId, namespace);const revision = await nextRevision(db, subjectId, namespace, row.revision);-- moduleEvents.record(db, {++ await moduleEvents.record(db, {subjectId, namespace, namespaceOwner: ownerOf(namespace) || 'network', schemaVersion: row.version, revision,change: 'deleted', reason, before: JSON.parse(row.data), after: {}, actor, mergedInto, ctx,});@@@ -218,7 +218,7 @@ async function onSubjectMerged(db, { from, into, ctx } = {}) {const revision = await nextRevision(db, into, row.namespace, 0);await db.prepare(`INSERT INTO user_modules (subject_id, namespace, version, revision, data, updated_at, updated_by)VALUES (?, ?, ?, ?, ?, ov_now(), 'svc:network')`).run(into, row.namespace, row.version, revision, row.data);-- moduleEvents.record(db, {++ await moduleEvents.record(db, {subjectId: into, namespace: row.namespace, namespaceOwner: ownerOf(row.namespace) || 'network', schemaVersion: row.version, revision,change: 'created', reason: 'subject_merged', before: {}, after: JSON.parse(row.data), actor: system, mergedFrom: from, ctx,});@@@ -236,7 +236,7 @@ async function onSubjectMerged(db, { from, into, ctx } = {}) {const revision = await nextRevision(db, into, row.namespace, mine.revision);await db.prepare("UPDATE user_modules SET data = ?, revision = ?, updated_at = ov_now(), updated_by = 'svc:network' WHERE subject_id = ? AND namespace = ?").run(JSON.stringify(next), revision, into, row.namespace);-- moduleEvents.record(db, {++ await moduleEvents.record(db, {subjectId: into, namespace: row.namespace, namespaceOwner: ownerOf(row.namespace) || 'network', schemaVersion: mine.version, revision,change: 'updated', reason: 'subject_merged', before: a, after: next, actor: system, mergedFrom: from, ctx,});@@@ -8,6 +8,8 @@// ═══════════════════════════════════════════════════════════════const express = require('express');++// Async route handlers: a rejection reaches Express's error handling, as a throw did (server/async-routes.js, plan T2).++require('./async-routes');const cors = require('cors');const helmet = require('helmet');const cookieParser = require('cookie-parser');@@@ -382,6 +384,7 @@ console.log(`[Secrets] ${await require('./secrets').summary(db)}`);// ── Initialize Services ──────────────────────────────────────const notificationService = new NotificationService(db);const emailService = new EmailService(db);++await emailService.ready; // its settings are read from the database before the first requestapp.locals.notificationService = notificationService;app.locals.emailService = emailService;// live.stream.started: the followers are read from Live (its follow graph) with Network's own service token.@@@ -579,7 +582,7 @@ app.use('/api/v1/projects', rateLimit({ windowMs: 60_000, max: 60 }), require('.await require('./developer/event-relay').startRelay(db, { eventsUrl: config.eventsInternalUrl, privateKey, issuer: config.jwt.issuer });// network.user.updated (WS-B task 2): profile, role and ban changes, recorded by triggers on users and relayed// through the same outbox (server/identity/profile-events.js).--require('./identity/profile-events').start(db);++await require('./identity/profile-events').start(db);require('./identity/grants-admin').start(db);// Notification API (authenticated users)@@@ -869,10 +872,10 @@ const server = app.listen(config.port, config.host, () => {// ── Periodic Maintenance ─────────────────────────────────// Clean expired notifications every hour-- timers.push(setInterval(() => notificationService.maintenance(), 60 * 60 * 1000));++ timers.push(setInterval(() => notificationService.maintenance().catch((e) => console.warn('[Notifications] maintenance failed:', e.message)), 60 * 60 * 1000));// Process email queue every 2 minutes-- timers.push(setInterval(() => emailService.processQueue(notificationService), 2 * 60 * 1000));++ timers.push(setInterval(() => emailService.processQueue(notificationService).catch((e) => console.warn('[Email] queue failed:', e.message)), 2 * 60 * 1000));// Raw analytics retention (ADR-021) is scheduled by the PostgreSQL tracker itself: events older// than 30 days go, in bounded batches; hourly/daily rollups stay.@@@ -900,8 +903,10 @@ const server = app.listen(config.port, config.host, () => {// Clean expired sessions dailytimers.push(setInterval(async () => {-- const cleaned = (await db.prepare("DELETE FROM user_sessions WHERE expires_at < datetime('now') OR is_active = 0").run()).changes;-- if (cleaned > 0) console.log(`[Sessions] Cleaned ${cleaned} expired sessions`);++ try {++ const cleaned = (await db.prepare("DELETE FROM user_sessions WHERE expires_at < datetime('now') OR is_active = 0").run()).changes;++ if (cleaned > 0) console.log(`[Sessions] Cleaned ${cleaned} expired sessions`);++ } catch (e) { console.warn('[Sessions] cleanup failed:', e.message); }}, 24 * 60 * 60 * 1000));});@@@ -88,7 +88,7 @@ router.post('/link-account', principals.guard('identity.subject.resolve', { ownAconst me = await db.prepare('SELECT avatar_url, display_name, username FROM users WHERE id = ?').get(user_id);if (me) {const svc = req.app.locals.avatarService;-- if (svc && avatar_url && !me.avatar_url) svc.fromSite({ user_id, avatar_url, origin: service });++ if (svc && avatar_url && !me.avatar_url) await svc.fromSite({ user_id, avatar_url, origin: service });const name = String(display_name || '').trim().slice(0, 60);// Same rule the profile form enforces: a display name only re-cases the username.if (name && name !== me.display_name && name.toLowerCase() === String(me.username || '').toLowerCase()) await db.prepare('UPDATE users SET display_name = ? WHERE id = ?').run(name, user_id);@@@ -384,7 +384,7 @@ router.post('/operator/alerts', principals.guard('network.operator.alert'), asyn// ── Mark Notifications Read by Type ──────────────────────────// POST /internal/notifications/mark-read// Body: { user_id, type, url_pattern? }--router.post('/notifications/mark-read', principals.guard('network.notifications.push'), (req, res) => {++router.post('/notifications/mark-read', principals.guard('network.notifications.push'), async (req, res) => {const notifService = req.app.locals.notificationService;if (!notifService) return res.status(503).json({ error: 'Notification service unavailable' });@@@ -392,7 +392,7 @@ router.post('/notifications/mark-read', principals.guard('network.notifications.if (!user_id || !type) return res.status(400).json({ error: 'user_id and type required' });try {-- const changes = notifService.markReadByType(parseInt(user_id), type, url_pattern || null);++ const changes = await notifService.markReadByType(parseInt(user_id), type, url_pattern || null);res.json({ ok: true, marked: changes });} catch (err) {console.error('[Internal] Mark read by type error:', err);@@@ -61,8 +61,10 @@ class EmailService {INSERT INTO email_delivery_log (email_type, recipient, subject, status, error_message, user_id, notification_id, metadata)VALUES (?, ?, ?, ?, ?, ?, ?, ?)`);-- this._migrateFromSES();-- this._loadConfig();++ // Settings are read from the database (async): `ready` settles once they are loaded. Every send and++ // getStatus() waits for it; the boot awaits it too. A rejection is kept for those awaiters, not left unhandled.++ this.ready = (async () => { await this._migrateFromSES(); await this._loadConfig(); })();++ this.ready.catch(() => {});}/**@@@ -102,9 +104,11 @@ class EmailService {}/** Reload config (call after admin changes settings). */-- reload() {++ async reload() {this._apiKey = null;-- this._loadConfig();++ this.ready = this._loadConfig();++ this.ready.catch(() => {});++ await this.ready;}get isEnabled() { return this._enabled && !!this._apiKey; }@@@ -138,10 +142,11 @@ class EmailService {async _sendEmail({ to, subject, htmlBody, textBody, emailType = 'generic', userId = null, notificationId = null, metadata = null, fromEmail = null }) {if (!to) return false;++ await this.ready;if (!this.isEnabled) {-- const reason = this.diagnose() || 'Email service is disabled or not configured';-- this._recordDelivery({++ const reason = (await this.diagnose()) || 'Email service is disabled or not configured';++ await this._recordDelivery({emailType,recipient: to,subject,@@@ -164,7 +169,7 @@ class EmailService {html: htmlBody,text: textBody,});-- this._recordDelivery({++ await this._recordDelivery({emailType,recipient: to,subject,@@@ -175,7 +180,7 @@ class EmailService {});return true;} catch (err) {-- this._recordDelivery({++ await this._recordDelivery({emailType,recipient: to,subject,@@@ -198,7 +203,7 @@ class EmailService {async sendNotificationEmail({ to, username, subject, notification }) {const htmlBody = this._buildEmailHtml({ username, notification });const textBody = this._buildEmailText({ username, notification });-- const fromEmail = this.getFromEmail(notification.service);++ const fromEmail = await this.getFromEmail(notification.service);const sent = await this._sendEmail({to,subject: subject || notification.title || 'Notification',@@@ -262,6 +267,7 @@ class EmailService {* @param {import('./notification-service').NotificationService} notifService*/async processQueue(notifService) {++ await this.ready;if (!this.isEnabled) return;const pending = await notifService.getPendingEmails();@@@ -271,11 +277,11 @@ class EmailService {let skipped = 0;for (const notif of pending) {-- if (!notifService.shouldEmail(notif)) {++ if (!(await notifService.shouldEmail(notif))) {await notifService.markEmailed(notif.id);continue;}-- const why = notifService.emailGuard ? notifService.emailGuard(notif) : null;++ const why = notifService.emailGuard ? await notifService.emailGuard(notif) : null;if (why) {// Caps: leave the row un-emailed only if the cap may lift (it won't for stale/dup).if (why === 'stale' || why === 'dup') await notifService.markEmailed(notif.id);@@@ -412,7 +418,7 @@ body { margin: 0; padding: 0; background: #1a1a24; font-family: -apple-system, B].join('\n');}-- async _escapeHtml(value) {++ _escapeHtml(value) {return String(value ?? '').replace(/&/g, '&').replace(/</g, '<')@@@ -425,8 +431,9 @@ body { margin: 0; padding: 0; background: #1a1a24; font-family: -apple-system, B* Get current email service configuration status (for admin panel).*/async getStatus() {++ await this.ready;const apiKey = await this.db.getSetting('resend_api_key') || '';-- const issue = this.diagnose();++ const issue = await this.diagnose();const secrets = require('../secrets');return {enabled: this._enabled,@@@ -452,7 +459,8 @@ body { margin: 0; padding: 0; background: #1a1a24; font-family: -apple-system, B* Throws with a descriptive error if email service is not ready.*/async sendTestEmail(to) {-- const issue = this.diagnose();++ await this.ready;++ const issue = await this.diagnose();if (issue) throw new Error(issue);const subject = '🔥 OpenVibe — Test Email';@@@ -485,9 +485,9 @@ class NotificationService {* Run periodic maintenance (call from setInterval in main server).*/async maintenance() {-- const expired = this.cleanExpired();++ const expired = await this.cleanExpired();const maxAge = await this.db.getSetting?.('notification_max_age_days') || 90;-- const old = this.cleanOld(maxAge);++ const old = await this.cleanOld(maxAge);if (expired + old > 0) {console.log(`[Notifications] Cleaned ${expired} expired + ${old} old notifications`);}@@@ -87,9 +87,9 @@ function createNetworkReadiness({ db, getKeys, release, ecosystem = null, discorif (discordService && typeof discordService.isReady === 'function') {checks.push({name: 'discord_bot', required: false,-- check: () => {++ check: async () => {let configured = false;-- try { configured = !!discordService._getSetting('discord_bot_token'); } catch { configured = false; }++ try { configured = !!(await discordService._getSetting('discord_bot_token')); } catch { configured = false; }if (!configured) return skip('not configured'); // verified nothing: skipped, never ok (WS-Q task 7)return discordService.isReady() || 'bot token configured but the bot is not connected';},@@@ -107,7 +107,7 @@ async function receive(db, body, { notify, revise = null, now = Date.now(), envconst kept = kind === 'reminded' ? notices(prev) : [];for (const uid of users) {try {-- const created = notify(uid, n);++ const created = await notify(uid, n);if (created) out.notified += 1;if (created && created.id) kept.push([uid, created.id]);} catch (e) { console.warn('[operator-alerts] notify failed:', e.message); }@@@ -120,7 +120,7 @@ async function receive(db, body, { notify, revise = null, now = Date.now(), env? { title: `Resolved: ${a.name}`.slice(0, 120), icon: '✅', is_read: 1, message: `${a.summary} (resolved after ${ago(now - (prev.opened_at || now))}${prev.reopened ? `; it fired ${prev.reopened + 1} times` : ''}).`.slice(0, 500) }: { title: `Alert: ${a.name}`.slice(0, 120), icon: critical ? '🚨' : '⚠️', is_read: critical ? 0 : 1, message: `${a.summary}. ${message(a)} Fired again ${(prev.reopened || 0) + 1} time(s) within 12 h of resolving.`.slice(0, 500) };for (const [uid, id] of notices(prev)) {-- try { revise(uid, id, fields); } catch (e) { console.warn('[operator-alerts] revise failed:', e.message); }++ try { await revise(uid, id, fields); } catch (e) { console.warn('[operator-alerts] revise failed:', e.message); }}}return out;nnew file mode 100644@@@ -0,0 +1,96 @@++'use strict';++// What the async conversion (plan T2) must not break, where the database handle is not the point: a rejected++// async route handler answers 500 and the process lives (server/async-routes.js); the email service awaits its++// settings, from-address, preferences and guards (a promise is always truthy, and '[object Promise]' is not an++// address); the Discord status reads its settings. The database is a stub here: no query runs.++const assert = require('assert');++const express = require('express');++require('../server/async-routes');++const { EmailService } = require('../server/notifications/email-service');++const { DiscordService } = require('../server/discord/discord-service');++++const quiet = async (fn) => { const l = console.log, w = console.warn, e = console.error; console.log = console.warn = console.error = () => {}; try { return await fn(); } finally { console.log = l; console.warn = w; console.error = e; } };++const stubDb = (settings) => ({++ prepare: () => ({ run: async () => ({ changes: 1 }), get: async () => null, all: async () => [] }),++ getSetting: async (k) => (k in settings ? settings[k] : null),++});++++(async () => {++ // ── Express 4 + async handlers ──++ let unhandled = null;++ const onUnhandled = (e) => { unhandled = e; };++ process.on('unhandledRejection', onUnhandled);++ const app = express();++ app.set('env', 'test'); // the default error handler answers 500 without logging the stack++ app.get('/rejects', async () => { await null; throw new Error('the database is down'); });++ app.get('/throws', () => { throw new Error('sync'); });++ app.get('/ok', async (req, res) => { await null; res.json({ ok: true }); });++ app.get('/passes', async (req, res, next) => { await null; next(); });++ app.get('/passes', (req, res) => res.send('next route'));++ const handled = express();++ handled.set('env', 'test');++ handled.get('/x', async () => { throw new Error('to the error handler'); });++ handled.use(async (err, req, res, next) => { await null; res.status(503).json({ caught: err.message }); });++ const listen = (a) => new Promise((r) => { const s = a.listen(0, '127.0.0.1', () => r(s)); });++ const s1 = await listen(app); const s2 = await listen(handled);++ const get = async (s, p) => { const r = await fetch(`http://127.0.0.1:${s.address().port}${p}`); return { status: r.status, text: await r.text() }; };++ try {++ assert.strictEqual((await get(s1, '/rejects')).status, 500, 'a rejected async handler answers 500');++ assert.strictEqual((await get(s1, '/throws')).status, 500, 'a synchronous throw still answers 500');++ assert.deepStrictEqual(await get(s1, '/ok'), { status: 200, text: '{"ok":true}' });++ assert.deepStrictEqual(await get(s1, '/passes'), { status: 200, text: 'next route' }, 'next() from an async handler still reaches the next route');++ assert.deepStrictEqual(await get(s2, '/x'), { status: 503, text: '{"caught":"to the error handler"}' }, 'the rejection reaches the app error handler');++ await new Promise((r) => setImmediate(r));++ assert.strictEqual(unhandled, null, 'no unhandled rejection');++ } finally { s1.close(); s2.close(); process.off('unhandledRejection', onUnhandled); }++ console.log('async route handlers: rejections answer 500');++++ // ── Email service ──++ const email = new EmailService(stubDb({ email_enabled: true, resend_api_key: 're_test_key', email_from_address: '[email protected]', email_from_live: '[email protected]' }));++ await quiet(() => email.ready);++ assert.strictEqual(email.isEnabled, true, 'the settings are loaded once ready settles');++ assert.strictEqual(email._escapeHtml('<b>"x"</b>'), '<b>"x"</b>', 'escaping is synchronous');++ const sent = [];++ email._sendEmail = async (o) => { sent.push(o); return true; };++ await email.sendVerificationEmail({ to: '[email protected]', username: '<Ann>', verifyUrl: 'https://openvibe.network/verify-email?token=t' });++ assert.ok(!/\[object Promise\]/.test(sent[0].htmlBody) && /Hey <Ann>,/.test(sent[0].htmlBody), 'the verification email is escaped text');++ await email.sendNotificationEmail({ to: '[email protected]', username: 'ann', notification: { title: 'T', message: 'M', service: 'live', priority: 'normal' } });++ assert.strictEqual(sent[1].fromEmail, '[email protected]', 'the per-service from-address is a string');++ assert.ok(!/\[object Promise\]/.test(sent[1].htmlBody));++ const status = await email.getStatus();++ assert.strictEqual(status.issue, null, 'getStatus awaits diagnose()');++ assert.strictEqual(status.ready, true);++++ // processQueue: an async shouldEmail() of false and an async guard skip, only the rest is sent.++ const marked = [];++ const n = (id, extra = {}) => ({ id, email: `u${id}@example.test`, username: `u${id}`, title: 'T', type: 'SYSTEM', priority: 'critical', ...extra });++ const notif = {++ getPendingEmails: async () => [n(1), n(2, { no: true }), n(3, { guard: 'stale' }), n(4, { guard: 'user-cap' })],++ shouldEmail: async (x) => !x.no,++ emailGuard: async (x) => x.guard || null,++ markEmailed: async (id) => { marked.push(id); },++ };++ sent.length = 0;++ await quiet(() => email.processQueue(notif));++ assert.deepStrictEqual(sent.map((o) => o.to), ['[email protected]'], 'only the notification that passes both checks is sent');++ assert.deepStrictEqual(marked.sort(), [1, 2, 3], 'sent, opted out and stale are marked; a lifted cap may still send later');++++ // Disabled: the failure reason recorded is diagnose()'s text, not a promise.++ const off = new EmailService(stubDb({}));++ await quiet(() => off.ready);++ const logged = [];++ off._recordDelivery = async (o) => { logged.push(o); };++ assert.strictEqual(await off._sendEmail({ to: '[email protected]', subject: 's', htmlBody: '', textBody: '' }), false);++ assert.strictEqual(typeof logged[0].errorMessage, 'string');++ assert.match(logged[0].errorMessage, /Email is disabled/);++ console.log('email service: settings, from-address, preferences and guards are awaited');++++ // ── Discord status ──++ const discord = new DiscordService(stubDb({ discord_guild_id: 'g1', discord_alerts_channel_id: 'c1' }));++ const ds = await discord.getStatus();++ assert.strictEqual(ds.guildId, 'g1');++ assert.strictEqual(ds.alertsChannelId, 'c1');++ assert.strictEqual(ds.connected, false);++ console.log('discord service: getStatus reads its settings');++ console.log('async conversion: all checks passed');++})().catch((err) => { console.error(err); process.exit(1); });@@@ -35,7 +35,8 @@ app.locals.config = { jwt: { issuer: 'https://openvibe.network', accessTokenExpiapp.locals.privateKey = keys.privateKey;app.locals.publicKey = keys.publicKey;const revised = [];--app.locals.notificationService = { create: (n) => { sent.push(n); return { id: `n${sent.length}` }; }, revise: (id, uid, f) => { revised.push({ id, uid, ...f }); return true; } };++// Async, like NotificationService.create/revise on PostgreSQL: the notification ids kept for the revisions come from the awaited result.++app.locals.notificationService = { create: async (n) => { sent.push(n); return { id: `n${sent.length}` }; }, revise: async (id, uid, f) => { revised.push({ id, uid, ...f }); return true; } };app.use('/oauth', require('../server/auth/oauth-routes'));app.use('/internal', require('../server/internal/routes'));const server = http.createServer(app);@@@ -27,7 +27,7 @@ const freePort = () => new Promise((resolve) => { const s = net.createServer();const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-net-budget-'));const port = await freePort();const child = spawn(process.execPath, [path.join(__dirname, '..', 'server', 'index.js')], {-- env: { ...process.env, PORT: String(port), HOST: '127.0.0.1', DB_PATH: path.join(dir, 'network.db'), NODE_ENV: 'test' },++ env: { ...process.env, PORT: String(port), HOST: '127.0.0.1', PGLITE_DIR: path.join(dir, 'pglite'), NODE_ENV: 'test' },stdio: ['ignore', 'ignore', 'pipe'],});let stderr = '';@@@ -18,7 +18,7 @@ const freePort = () => new Promise((resolve) => { const s = net.createServer();const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-net-stop-'));const port = await freePort();const child = spawn(process.execPath, [path.join(__dirname, '..', 'server', 'index.js')], {-- env: { ...process.env, PORT: String(port), HOST: '127.0.0.1', DB_PATH: path.join(dir, 'network.db'), NODE_ENV: 'test' },++ env: { ...process.env, PORT: String(port), HOST: '127.0.0.1', PGLITE_DIR: path.join(dir, 'pglite'), NODE_ENV: 'test' },stdio: ['ignore', 'pipe', 'pipe'],});let out = '';
@@ -230,7 +230,7 @@ function createAdminRoutes(db, notificationService, emailService, requireAuth) {
@@ -230,7 +230,7 @@ function createAdminRoutes(db, notificationService, emailService, requireAuth) {
// GET /api/admin/email — get current email config (owner-only: exposes key metadata)
// GET /api/admin/email — get current email config (owner-only: exposes key metadata)
router.get('/email', requireOwner, async (req, res) => {
router.get('/email', requireOwner, async (req, res) => {
try {
try {
- const status = emailService.getStatus();
+ const status = await emailService.getStatus();
res.json({ ok: true, email: status, metrics: await getEmailMetrics() });
res.json({ ok: true, email: status, metrics: await getEmailMetrics() });
} catch (err) {
} catch (err) {
res.status(500).json({ ok: false, error: err.message });
res.status(500).json({ ok: false, error: err.message });
@@ -265,14 +265,14 @@ function createAdminRoutes(db, notificationService, emailService, requireAuth) {
@@ -265,14 +265,14 @@ function createAdminRoutes(db, notificationService, emailService, requireAuth) {
await collect();
await collect();
if (Object.keys(changes.set).length) await (await siteConfig.forDb(db)).change(changes, { actor: siteConfig.actorOf(req.user), reason: String(req.body.reason || 'email settings').slice(0, 300) });
if (Object.keys(changes.set).length) await (await siteConfig.forDb(db)).change(changes, { actor: siteConfig.actorOf(req.user), reason: String(req.body.reason || 'email settings').slice(0, 300) });
- emailService.reload();
+ await emailService.reload();
// Audit
// Audit
await db.prepare('INSERT INTO audit_log (user_id, action, details) VALUES (?, ?, ?)').run(
await db.prepare('INSERT INTO audit_log (user_id, action, details) VALUES (?, ?, ?)').run(
req.user.id, 'email_config_update', JSON.stringify({ from_email })
req.user.id, 'email_config_update', JSON.stringify({ from_email })
);
);
- res.json({ ok: true, email: emailService.getStatus(), ...(skipped.length ? { skipped, note: 'set in the environment; not saved to the database' } : {}) });
+ res.json({ ok: true, email: await emailService.getStatus(), ...(skipped.length ? { skipped, note: 'set in the environment; not saved to the database' } : {}) });
} catch (err) {
} catch (err) {
if (err && err.status && err.code) return res.status(err.status).json({ ok: false, error: err.message, code: err.code, errors: err.errors });
if (err && err.status && err.code) return res.status(err.status).json({ ok: false, error: err.message, code: err.code, errors: err.errors });
res.status(500).json({ ok: false, error: err.message });
res.status(500).json({ ok: false, error: err.message });
@@ -591,7 +591,7 @@ function createAdminRoutes(db, notificationService, emailService, requireAuth) {
@@ -591,7 +591,7 @@ function createAdminRoutes(db, notificationService, emailService, requireAuth) {
if (search) {
if (search) {
users = await db.prepare(`
users = await db.prepare(`
SELECT id, username, display_name, email, role, is_banned, created_at, last_seen
SELECT id, username, display_name, email, role, is_banned, created_at, last_seen
- FROM users WHERE username LIKE ? OR display_name LIKE ? OR email LIKE ?
+ FROM users WHERE username ILIKE ? OR display_name ILIKE ? OR email ILIKE ?
ORDER BY created_at DESC LIMIT ? OFFSET ?
ORDER BY created_at DESC LIMIT ? OFFSET ?
`).all(`%${search}%`, `%${search}%`, `%${search}%`, limit, offset);
`).all(`%${search}%`, `%${search}%`, `%${search}%`, limit, offset);
} else {
} else {
@@ -847,7 +847,7 @@ function createAdminRoutes(db, notificationService, emailService, requireAuth) {
@@ -847,7 +847,7 @@ function createAdminRoutes(db, notificationService, emailService, requireAuth) {
const unreadCount = (await db.prepare('SELECT COUNT(*) as cnt FROM notifications WHERE is_read = 0').get()).cnt;
const unreadCount = (await db.prepare('SELECT COUNT(*) as cnt FROM notifications WHERE is_read = 0').get()).cnt;
const anonCount = (await db.prepare('SELECT COUNT(*) as cnt FROM anon_users').get()).cnt;
const anonCount = (await db.prepare('SELECT COUNT(*) as cnt FROM anon_users').get()).cnt;
const sessionCount = (await db.prepare('SELECT COUNT(*) as cnt FROM user_sessions WHERE is_active = 1').get()).cnt;
const sessionCount = (await db.prepare('SELECT COUNT(*) as cnt FROM user_sessions WHERE is_active = 1').get()).cnt;
- const emailStatus = emailService.getStatus();
+ const emailStatus = await emailService.getStatus();
res.json({
res.json({
ok: true,
ok: true,
@@ -47,10 +47,11 @@ function createStaffApi({ db, requireAuth, guard }) {
@@ -47,10 +47,11 @@ function createStaffApi({ db, requireAuth, guard }) {
const bearer = String(req.headers.authorization || '').replace(/^Bearer\s+/i, '');
const bearer = String(req.headers.authorization || '').replace(/^Bearer\s+/i, '');
let sub = '';
let sub = '';
try { sub = JSON.parse(Buffer.from(bearer.split('.')[1] || '', 'base64url').toString('utf8')).sub || ''; } catch { /* not a JWT */ }
try { sub = JSON.parse(Buffer.from(bearer.split('.')[1] || '', 'base64url').toString('utf8')).sub || ''; } catch { /* not a JWT */ }
- if (String(sub).startsWith('svc:')) return serviceGuard(req, res, async () => await list(req, res));
- return requireAuth(req, res, async () => {
+ // The guards call back after this handler returned: a failure in list() goes to next(err).
+ if (String(sub).startsWith('svc:')) return serviceGuard(req, res, () => list(req, res).catch(next));
+ return requireAuth(req, res, () => {
if (!staff.can(staffClaims(req.user), 'staff.moderation.logs')) return res.status(403).json({ error: 'forbidden', detail: 'staff.moderation.logs required' });
if (!staff.can(staffClaims(req.user), 'staff.moderation.logs')) return res.status(403).json({ error: 'forbidden', detail: 'staff.moderation.logs required' });
- return await list(req, res);
+ return list(req, res).catch(next);
});
});
});
});
return r;
return r;
@@ -101,7 +101,7 @@ async function viewerOf(req) {
@@ -101,7 +101,7 @@ async function viewerOf(req) {
function router({ fullGuard }) {
function router({ fullGuard }) {
const r = express.Router();
const r = express.Router();
r.use(http.middleware());
r.use(http.middleware());
- r.get('/:creator/analytics', async (req, res) => {
+ r.get('/:creator/analytics', async (req, res, next) => {
const db = req.app.locals.db;
const db = req.app.locals.db;
const key = String(req.params.creator || '');
const key = String(req.params.creator || '');
const u = SUBJECT_RE.test(key) ? await db.prepare('SELECT subject_id, is_anon FROM users WHERE subject_id = ?').get(key)
const u = SUBJECT_RE.test(key) ? await db.prepare('SELECT subject_id, is_anon FROM users WHERE subject_id = ?').get(key)
@@ -111,7 +111,8 @@ function router({ fullGuard }) {
@@ -111,7 +111,8 @@ function router({ fullGuard }) {
const viewer = await viewerOf(req);
const viewer = await viewerOf(req);
if (viewer) return await answer(viewer.subject_id === u.subject_id);
if (viewer) return await answer(viewer.subject_id === u.subject_id);
const h = String(req.headers.authorization || '');
const h = String(req.headers.authorization || '');
- if (h.startsWith('Bearer ')) return fullGuard(req, res, async () => await answer(true)); // a service token: full, or refused
+ // The guard calls back after this handler returned: a failure there goes to next(err) (server/async-routes.js).
+ if (h.startsWith('Bearer ')) return fullGuard(req, res, () => answer(true).catch(next)); // a service token: full, or refused
return await answer(false);
return await answer(false);
});
});
return r;
return r;
new file mode 100644
new file mode 100644
@@ -0,0 +1,50 @@
@@ -0,0 +1,50 @@
+'use strict';
+
+// ═══════════════════════════════════════════════════════════════
+// Async route handlers on Express 4 (plan T2).
+//
+// Every query is async on PostgreSQL, so most handlers are async functions. Express 4 ignores what a
+// handler returns: a rejected promise (a failed query, say) was an unhandled rejection, which ends a
+// Node 22 process, and the request hung. With this installed a rejection goes to next(err), exactly
+// where a synchronous throw went before (Express's own error handling answers 500), as Express 5 does.
+// Installed once, before the first request, by server/index.js.
+// ═══════════════════════════════════════════════════════════════
+
+const Layer = require('express/lib/router/layer');
+
+const INSTALLED = Symbol.for('openvibe.network.async-routes');
+
+function forward(next) {
+ return (err) => next(err || new Error('the route handler rejected without a reason'));
+}
+
+function install() {
+ if (Layer.prototype[INSTALLED]) return;
+ Layer.prototype[INSTALLED] = true;
+
+ Layer.prototype.handle_request = function handle(req, res, next) {
+ const fn = this.handle;
+ if (fn.length > 3) return next(); // an error handler: not for a request
+ try {
+ const r = fn(req, res, next);
+ if (r && typeof r.then === 'function') r.then(undefined, forward(next));
+ } catch (err) {
+ next(err);
+ }
+ };
+
+ Layer.prototype.handle_error = function handleError(error, req, res, next) {
+ const fn = this.handle;
+ if (fn.length !== 4) return next(error); // not an error handler
+ try {
+ const r = fn(error, req, res, next);
+ if (r && typeof r.then === 'function') r.then(undefined, forward(next));
+ } catch (err) {
+ next(err);
+ }
+ };
+}
+
+install();
+
+module.exports = { install };
@@ -128,7 +128,7 @@ function routes(requireAuth) {
@@ -128,7 +128,7 @@ function routes(requireAuth) {
const r = await consumeToken(db, String(req.body?.token || ''));
const r = await consumeToken(db, String(req.body?.token || ''));
if (!r.ok) return res.status(400).json({ ok: false, error: r.error });
if (!r.ok) return res.status(400).json({ ok: false, error: r.error });
try {
try {
- req.app.locals.notificationService?.create({ user_id: r.user.id, type: 'EMAIL_VERIFIED', title: 'Email verified', message: `${r.user.email} is confirmed. Go-live alerts from streamers you follow will now reach your inbox.`, service: 'network', url: 'https://openvibe.network/notifications' });
+ await req.app.locals.notificationService?.create({ user_id: r.user.id, type: 'EMAIL_VERIFIED', title: 'Email verified', message: `${r.user.email} is confirmed. Go-live alerts from streamers you follow will now reach your inbox.`, service: 'network', url: 'https://openvibe.network/notifications' });
} catch { /* */ }
} catch { /* */ }
res.json({ ok: true, email: r.user.email });
res.json({ ok: true, email: r.user.email });
});
});
@@ -15,7 +15,7 @@ try {
@@ -15,7 +15,7 @@ try {
class DiscordService {
class DiscordService {
/**
/**
- * @param {object} db — better-sqlite3 Database instance (with getSetting helper)
+ * @param {object} db — the openvibe-sdk/db handle (with the async getSetting helper)
*/
*/
constructor(db) {
constructor(db) {
this._db = db;
this._db = db;
@@ -36,7 +36,7 @@ class DiscordService {
@@ -36,7 +36,7 @@ class DiscordService {
async init() {
async init() {
if (!Discord) return;
if (!Discord) return;
- const token = this._getSetting('discord_bot_token');
+ const token = await this._getSetting('discord_bot_token');
if (!token) {
if (!token) {
console.log('[Discord] No bot token configured — bot disabled');
console.log('[Discord] No bot token configured — bot disabled');
return;
return;
@@ -87,13 +87,13 @@ class DiscordService {
@@ -87,13 +87,13 @@ class DiscordService {
// ─── Status ────────────────────────────────────────────
// ─── Status ────────────────────────────────────────────
- getStatus() {
+ async getStatus() {
return {
return {
available: !!Discord,
available: !!Discord,
connected: this._ready,
connected: this._ready,
botTag: this._client?.user?.tag || null,
botTag: this._client?.user?.tag || null,
- guildId: this._getSetting('discord_guild_id') || null,
- alertsChannelId: this._getSetting('discord_alerts_channel_id') || null,
+ guildId: (await this._getSetting('discord_guild_id')) || null,
+ alertsChannelId: (await this._getSetting('discord_alerts_channel_id')) || null,
};
};
}
}
@@ -114,14 +114,14 @@ class DiscordService {
@@ -114,14 +114,14 @@ class DiscordService {
async sendLiveAlert(streamer, stream) {
async sendLiveAlert(streamer, stream) {
if (!this.isReady()) return { sent: false, reason: 'bot_not_connected' };
if (!this.isReady()) return { sent: false, reason: 'bot_not_connected' };
- const channelId = this._getSetting('discord_alerts_channel_id');
+ const channelId = await this._getSetting('discord_alerts_channel_id');
if (!channelId) return { sent: false, reason: 'no_alerts_channel' };
if (!channelId) return { sent: false, reason: 'no_alerts_channel' };
// Dedupe check — one alert per streamer per cooldown window
// Dedupe check — one alert per streamer per cooldown window
- const cooldownMin = parseInt(this._getSetting('discord_dedupe_minutes') || '15', 10);
+ const cooldownMin = parseInt((await this._getSetting('discord_dedupe_minutes')) || '15', 10);
const cooldownMs = Math.max(cooldownMin, 1) * 60 * 1000;
const cooldownMs = Math.max(cooldownMin, 1) * 60 * 1000;
const key = streamer.username.toLowerCase();
const key = streamer.username.toLowerCase();
- const lastAlert = await this._liveAlertCooldowns.get(key);
+ const lastAlert = this._liveAlertCooldowns.get(key);
if (lastAlert && (Date.now() - lastAlert) < cooldownMs) {
if (lastAlert && (Date.now() - lastAlert) < cooldownMs) {
return { sent: false, reason: 'cooldown', remaining_ms: cooldownMs - (Date.now() - lastAlert) };
return { sent: false, reason: 'cooldown', remaining_ms: cooldownMs - (Date.now() - lastAlert) };
}
}
@@ -153,7 +153,7 @@ class DiscordService {
@@ -153,7 +153,7 @@ class DiscordService {
}
}
// Custom message template support
// Custom message template support
- const customTemplate = this._getSetting('discord_alert_message');
+ const customTemplate = await this._getSetting('discord_alert_message');
let content = null;
let content = null;
if (customTemplate) {
if (customTemplate) {
content = customTemplate
content = customTemplate
@@ -182,7 +182,7 @@ class DiscordService {
@@ -182,7 +182,7 @@ class DiscordService {
async sendSystemAlert(alert) {
async sendSystemAlert(alert) {
if (!this.isReady()) return { sent: false, reason: 'bot_not_connected' };
if (!this.isReady()) return { sent: false, reason: 'bot_not_connected' };
- const channelId = this._getSetting('discord_system_channel_id');
+ const channelId = await this._getSetting('discord_system_channel_id');
if (!channelId) return { sent: false, reason: 'no_system_channel' };
if (!channelId) return { sent: false, reason: 'no_system_channel' };
const colors = { info: 0x3b82f6, warning: 0xf59e0b, error: 0xef4444 };
const colors = { info: 0x3b82f6, warning: 0xf59e0b, error: 0xef4444 };
@@ -16,7 +16,7 @@ module.exports = function createDiscordRoutes(db, discordService, requireAuth, r
@@ -16,7 +16,7 @@ module.exports = function createDiscordRoutes(db, discordService, requireAuth, r
/** GET /api/admin/discord — bot status + current config */
/** GET /api/admin/discord — bot status + current config */
router.get('/', async (req, res) => {
router.get('/', async (req, res) => {
- const status = discordService.getStatus();
+ const status = await discordService.getStatus();
const settings = {
const settings = {
discord_bot_token: await db.getSetting('discord_bot_token') ? '••••••••' : '',
discord_bot_token: await db.getSetting('discord_bot_token') ? '••••••••' : '',
discord_guild_id: await db.getSetting('discord_guild_id') || '',
discord_guild_id: await db.getSetting('discord_guild_id') || '',
@@ -85,7 +85,7 @@ module.exports = function createDiscordRoutes(db, discordService, requireAuth, r
@@ -85,7 +85,7 @@ module.exports = function createDiscordRoutes(db, discordService, requireAuth, r
}
}
}
}
- res.json({ ok: true, status: discordService.getStatus(), ...(skipped.length ? { skipped, note: 'set in the environment; not saved to the database' } : {}) });
+ res.json({ ok: true, status: await discordService.getStatus(), ...(skipped.length ? { skipped, note: 'set in the environment; not saved to the database' } : {}) });
});
});
/** POST /api/admin/discord/test — send a test alert to the configured channel */
/** POST /api/admin/discord/test — send a test alert to the configured channel */
@@ -111,7 +111,7 @@ module.exports = function createDiscordRoutes(db, discordService, requireAuth, r
@@ -111,7 +111,7 @@ module.exports = function createDiscordRoutes(db, discordService, requireAuth, r
router.post('/reinit', async (req, res) => {
router.post('/reinit', async (req, res) => {
try {
try {
await discordService.reinit();
await discordService.reinit();
- res.json({ ok: true, status: discordService.getStatus() });
+ res.json({ ok: true, status: await discordService.getStatus() });
} catch (err) {
} catch (err) {
res.status(500).json({ ok: false, error: err.message });
res.status(500).json({ ok: false, error: err.message });
}
}
@@ -295,7 +295,7 @@ function routers({ requireAuth, followsGuard }) {
@@ -295,7 +295,7 @@ function routers({ requireAuth, followsGuard }) {
const viewer = await viewerOf(req);
const viewer = await viewerOf(req);
if (viewer && viewer.subject_id === t.subject_id) return await answer();
if (viewer && viewer.subject_id === t.subject_id) return await answer();
if (viewer) return http.sendProblem(res, 403, 'follows.not_yours', { detail: 'only the channel owner sees who follows it', ctx: req.ov });
if (viewer) return http.sendProblem(res, 403, 'follows.not_yours', { detail: 'only the channel owner sees who follows it', ctx: req.ov });
- return followsGuard(req, res, answer);
+ return followsGuard(req, res, () => answer().catch(next)); // called back after this handler returned
});
});
// ADR-030 step 4: a first-party product (network.follows.write, checked by the mount) records a follow
// ADR-030 step 4: a first-party product (network.follows.write, checked by the mount) records a follow
@@ -133,7 +133,7 @@ async function write(db, subjectId, namespace, data, { writer, expectedRevision,
@@ -133,7 +133,7 @@ async function write(db, subjectId, namespace, data, { writer, expectedRevision,
.run(subjectId, namespace, ns.version, revision, JSON.stringify(data), by);
.run(subjectId, namespace, ns.version, revision, JSON.stringify(data), by);
const record = await read(db, subjectId, namespace);
const record = await read(db, subjectId, namespace);
assertValid('modules.module-record@1', record);
assertValid('modules.module-record@1', record);
- moduleEvents.record(db, {
+ await moduleEvents.record(db, {
subjectId, namespace, namespaceOwner: ownerOf(namespace), schemaVersion: ns.version, revision,
subjectId, namespace, namespaceOwner: ownerOf(namespace), schemaVersion: ns.version, revision,
change: cur ? 'updated' : 'created', reason: 'write', before: cur ? JSON.parse(cur.data) : {}, after: data,
change: cur ? 'updated' : 'created', reason: 'write', before: cur ? JSON.parse(cur.data) : {}, after: data,
actor: actorOf(subjectId, writer), ctx,
actor: actorOf(subjectId, writer), ctx,
@@ -153,7 +153,7 @@ async function removeRow(db, subjectId, namespace, { actor, reason, mergedInto,
@@ -153,7 +153,7 @@ async function removeRow(db, subjectId, namespace, { actor, reason, mergedInto,
if (!row) return 0;
if (!row) return 0;
await db.prepare('DELETE FROM user_modules WHERE subject_id = ? AND namespace = ?').run(subjectId, namespace);
await db.prepare('DELETE FROM user_modules WHERE subject_id = ? AND namespace = ?').run(subjectId, namespace);
const revision = await nextRevision(db, subjectId, namespace, row.revision);
const revision = await nextRevision(db, subjectId, namespace, row.revision);
- moduleEvents.record(db, {
+ await moduleEvents.record(db, {
subjectId, namespace, namespaceOwner: ownerOf(namespace) || 'network', schemaVersion: row.version, revision,
subjectId, namespace, namespaceOwner: ownerOf(namespace) || 'network', schemaVersion: row.version, revision,
change: 'deleted', reason, before: JSON.parse(row.data), after: {}, actor, mergedInto, ctx,
change: 'deleted', reason, before: JSON.parse(row.data), after: {}, actor, mergedInto, ctx,
});
});
@@ -218,7 +218,7 @@ async function onSubjectMerged(db, { from, into, ctx } = {}) {
@@ -218,7 +218,7 @@ async function onSubjectMerged(db, { from, into, ctx } = {}) {
const revision = await nextRevision(db, into, row.namespace, 0);
const revision = await nextRevision(db, into, row.namespace, 0);
await db.prepare(`INSERT INTO user_modules (subject_id, namespace, version, revision, data, updated_at, updated_by)
await db.prepare(`INSERT INTO user_modules (subject_id, namespace, version, revision, data, updated_at, updated_by)
VALUES (?, ?, ?, ?, ?, ov_now(), 'svc:network')`).run(into, row.namespace, row.version, revision, row.data);
VALUES (?, ?, ?, ?, ?, ov_now(), 'svc:network')`).run(into, row.namespace, row.version, revision, row.data);
- moduleEvents.record(db, {
+ await moduleEvents.record(db, {
subjectId: into, namespace: row.namespace, namespaceOwner: ownerOf(row.namespace) || 'network', schemaVersion: row.version, revision,
subjectId: into, namespace: row.namespace, namespaceOwner: ownerOf(row.namespace) || 'network', schemaVersion: row.version, revision,
change: 'created', reason: 'subject_merged', before: {}, after: JSON.parse(row.data), actor: system, mergedFrom: from, ctx,
change: 'created', reason: 'subject_merged', before: {}, after: JSON.parse(row.data), actor: system, mergedFrom: from, ctx,
});
});
@@ -236,7 +236,7 @@ async function onSubjectMerged(db, { from, into, ctx } = {}) {
@@ -236,7 +236,7 @@ async function onSubjectMerged(db, { from, into, ctx } = {}) {
const revision = await nextRevision(db, into, row.namespace, mine.revision);
const revision = await nextRevision(db, into, row.namespace, mine.revision);
await db.prepare("UPDATE user_modules SET data = ?, revision = ?, updated_at = ov_now(), updated_by = 'svc:network' WHERE subject_id = ? AND namespace = ?")
await db.prepare("UPDATE user_modules SET data = ?, revision = ?, updated_at = ov_now(), updated_by = 'svc:network' WHERE subject_id = ? AND namespace = ?")
.run(JSON.stringify(next), revision, into, row.namespace);
.run(JSON.stringify(next), revision, into, row.namespace);
- moduleEvents.record(db, {
+ await moduleEvents.record(db, {
subjectId: into, namespace: row.namespace, namespaceOwner: ownerOf(row.namespace) || 'network', schemaVersion: mine.version, revision,
subjectId: into, namespace: row.namespace, namespaceOwner: ownerOf(row.namespace) || 'network', schemaVersion: mine.version, revision,
change: 'updated', reason: 'subject_merged', before: a, after: next, actor: system, mergedFrom: from, ctx,
change: 'updated', reason: 'subject_merged', before: a, after: next, actor: system, mergedFrom: from, ctx,
});
});
@@ -8,6 +8,8 @@
@@ -8,6 +8,8 @@
// ═══════════════════════════════════════════════════════════════
// ═══════════════════════════════════════════════════════════════
const express = require('express');
const express = require('express');
+// Async route handlers: a rejection reaches Express's error handling, as a throw did (server/async-routes.js, plan T2).
+require('./async-routes');
const cors = require('cors');
const cors = require('cors');
const helmet = require('helmet');
const helmet = require('helmet');
const cookieParser = require('cookie-parser');
const cookieParser = require('cookie-parser');
@@ -382,6 +384,7 @@ console.log(`[Secrets] ${await require('./secrets').summary(db)}`);
@@ -382,6 +384,7 @@ console.log(`[Secrets] ${await require('./secrets').summary(db)}`);
// ── Initialize Services ──────────────────────────────────────
// ── Initialize Services ──────────────────────────────────────
const notificationService = new NotificationService(db);
const notificationService = new NotificationService(db);
const emailService = new EmailService(db);
const emailService = new EmailService(db);
+await emailService.ready; // its settings are read from the database before the first request
app.locals.notificationService = notificationService;
app.locals.notificationService = notificationService;
app.locals.emailService = emailService;
app.locals.emailService = emailService;
// live.stream.started: the followers are read from Live (its follow graph) with Network's own service token.
// live.stream.started: the followers are read from Live (its follow graph) with Network's own service token.
@@ -579,7 +582,7 @@ app.use('/api/v1/projects', rateLimit({ windowMs: 60_000, max: 60 }), require('.
@@ -579,7 +582,7 @@ app.use('/api/v1/projects', rateLimit({ windowMs: 60_000, max: 60 }), require('.
await require('./developer/event-relay').startRelay(db, { eventsUrl: config.eventsInternalUrl, privateKey, issuer: config.jwt.issuer });
await require('./developer/event-relay').startRelay(db, { eventsUrl: config.eventsInternalUrl, privateKey, issuer: config.jwt.issuer });
// network.user.updated (WS-B task 2): profile, role and ban changes, recorded by triggers on users and relayed
// network.user.updated (WS-B task 2): profile, role and ban changes, recorded by triggers on users and relayed
// through the same outbox (server/identity/profile-events.js).
// through the same outbox (server/identity/profile-events.js).
-require('./identity/profile-events').start(db);
+await require('./identity/profile-events').start(db);
require('./identity/grants-admin').start(db);
require('./identity/grants-admin').start(db);
// Notification API (authenticated users)
// Notification API (authenticated users)
@@ -869,10 +872,10 @@ const server = app.listen(config.port, config.host, () => {
@@ -869,10 +872,10 @@ const server = app.listen(config.port, config.host, () => {
// ── Periodic Maintenance ─────────────────────────────────
// ── Periodic Maintenance ─────────────────────────────────
// Clean expired notifications every hour
// Clean expired notifications every hour
- timers.push(setInterval(() => notificationService.maintenance(), 60 * 60 * 1000));
+ timers.push(setInterval(() => notificationService.maintenance().catch((e) => console.warn('[Notifications] maintenance failed:', e.message)), 60 * 60 * 1000));
// Process email queue every 2 minutes
// Process email queue every 2 minutes
- timers.push(setInterval(() => emailService.processQueue(notificationService), 2 * 60 * 1000));
+ timers.push(setInterval(() => emailService.processQueue(notificationService).catch((e) => console.warn('[Email] queue failed:', e.message)), 2 * 60 * 1000));
// Raw analytics retention (ADR-021) is scheduled by the PostgreSQL tracker itself: events older
// Raw analytics retention (ADR-021) is scheduled by the PostgreSQL tracker itself: events older
// than 30 days go, in bounded batches; hourly/daily rollups stay.
// than 30 days go, in bounded batches; hourly/daily rollups stay.
@@ -900,8 +903,10 @@ const server = app.listen(config.port, config.host, () => {
@@ -900,8 +903,10 @@ const server = app.listen(config.port, config.host, () => {
// Clean expired sessions daily
// Clean expired sessions daily
timers.push(setInterval(async () => {
timers.push(setInterval(async () => {
- const cleaned = (await db.prepare("DELETE FROM user_sessions WHERE expires_at < datetime('now') OR is_active = 0").run()).changes;
- if (cleaned > 0) console.log(`[Sessions] Cleaned ${cleaned} expired sessions`);
+ try {
+ const cleaned = (await db.prepare("DELETE FROM user_sessions WHERE expires_at < datetime('now') OR is_active = 0").run()).changes;
+ if (cleaned > 0) console.log(`[Sessions] Cleaned ${cleaned} expired sessions`);
+ } catch (e) { console.warn('[Sessions] cleanup failed:', e.message); }
}, 24 * 60 * 60 * 1000));
}, 24 * 60 * 60 * 1000));
});
});
@@ -88,7 +88,7 @@ router.post('/link-account', principals.guard('identity.subject.resolve', { ownA
@@ -88,7 +88,7 @@ router.post('/link-account', principals.guard('identity.subject.resolve', { ownA
const me = await db.prepare('SELECT avatar_url, display_name, username FROM users WHERE id = ?').get(user_id);
const me = await db.prepare('SELECT avatar_url, display_name, username FROM users WHERE id = ?').get(user_id);
if (me) {
if (me) {
const svc = req.app.locals.avatarService;
const svc = req.app.locals.avatarService;
- if (svc && avatar_url && !me.avatar_url) svc.fromSite({ user_id, avatar_url, origin: service });
+ if (svc && avatar_url && !me.avatar_url) await svc.fromSite({ user_id, avatar_url, origin: service });
const name = String(display_name || '').trim().slice(0, 60);
const name = String(display_name || '').trim().slice(0, 60);
// Same rule the profile form enforces: a display name only re-cases the username.
// Same rule the profile form enforces: a display name only re-cases the username.
if (name && name !== me.display_name && name.toLowerCase() === String(me.username || '').toLowerCase()) await db.prepare('UPDATE users SET display_name = ? WHERE id = ?').run(name, user_id);
if (name && name !== me.display_name && name.toLowerCase() === String(me.username || '').toLowerCase()) await db.prepare('UPDATE users SET display_name = ? WHERE id = ?').run(name, user_id);
@@ -384,7 +384,7 @@ router.post('/operator/alerts', principals.guard('network.operator.alert'), asyn
@@ -384,7 +384,7 @@ router.post('/operator/alerts', principals.guard('network.operator.alert'), asyn
// ── Mark Notifications Read by Type ──────────────────────────
// ── Mark Notifications Read by Type ──────────────────────────
// POST /internal/notifications/mark-read
// POST /internal/notifications/mark-read
// Body: { user_id, type, url_pattern? }
// Body: { user_id, type, url_pattern? }
-router.post('/notifications/mark-read', principals.guard('network.notifications.push'), (req, res) => {
+router.post('/notifications/mark-read', principals.guard('network.notifications.push'), async (req, res) => {
const notifService = req.app.locals.notificationService;
const notifService = req.app.locals.notificationService;
if (!notifService) return res.status(503).json({ error: 'Notification service unavailable' });
if (!notifService) return res.status(503).json({ error: 'Notification service unavailable' });
@@ -392,7 +392,7 @@ router.post('/notifications/mark-read', principals.guard('network.notifications.
@@ -392,7 +392,7 @@ router.post('/notifications/mark-read', principals.guard('network.notifications.
if (!user_id || !type) return res.status(400).json({ error: 'user_id and type required' });
if (!user_id || !type) return res.status(400).json({ error: 'user_id and type required' });
try {
try {
- const changes = notifService.markReadByType(parseInt(user_id), type, url_pattern || null);
+ const changes = await notifService.markReadByType(parseInt(user_id), type, url_pattern || null);
res.json({ ok: true, marked: changes });
res.json({ ok: true, marked: changes });
} catch (err) {
} catch (err) {
console.error('[Internal] Mark read by type error:', err);
console.error('[Internal] Mark read by type error:', err);
@@ -61,8 +61,10 @@ class EmailService {
@@ -61,8 +61,10 @@ class EmailService {
INSERT INTO email_delivery_log (email_type, recipient, subject, status, error_message, user_id, notification_id, metadata)
INSERT INTO email_delivery_log (email_type, recipient, subject, status, error_message, user_id, notification_id, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
`);
`);
- this._migrateFromSES();
- this._loadConfig();
+ // Settings are read from the database (async): `ready` settles once they are loaded. Every send and
+ // getStatus() waits for it; the boot awaits it too. A rejection is kept for those awaiters, not left unhandled.
+ this.ready = (async () => { await this._migrateFromSES(); await this._loadConfig(); })();
+ this.ready.catch(() => {});
}
}
/**
/**
@@ -102,9 +104,11 @@ class EmailService {
@@ -102,9 +104,11 @@ class EmailService {
}
}
/** Reload config (call after admin changes settings). */
/** Reload config (call after admin changes settings). */
- reload() {
+ async reload() {
this._apiKey = null;
this._apiKey = null;
- this._loadConfig();
+ this.ready = this._loadConfig();
+ this.ready.catch(() => {});
+ await this.ready;
}
}
get isEnabled() { return this._enabled && !!this._apiKey; }
get isEnabled() { return this._enabled && !!this._apiKey; }
@@ -138,10 +142,11 @@ class EmailService {
@@ -138,10 +142,11 @@ class EmailService {
async _sendEmail({ to, subject, htmlBody, textBody, emailType = 'generic', userId = null, notificationId = null, metadata = null, fromEmail = null }) {
async _sendEmail({ to, subject, htmlBody, textBody, emailType = 'generic', userId = null, notificationId = null, metadata = null, fromEmail = null }) {
if (!to) return false;
if (!to) return false;
+ await this.ready;
if (!this.isEnabled) {
if (!this.isEnabled) {
- const reason = this.diagnose() || 'Email service is disabled or not configured';
- this._recordDelivery({
+ const reason = (await this.diagnose()) || 'Email service is disabled or not configured';
+ await this._recordDelivery({
emailType,
emailType,
recipient: to,
recipient: to,
subject,
subject,
@@ -164,7 +169,7 @@ class EmailService {
@@ -164,7 +169,7 @@ class EmailService {
html: htmlBody,
html: htmlBody,
text: textBody,
text: textBody,
});
});
- this._recordDelivery({
+ await this._recordDelivery({
emailType,
emailType,
recipient: to,
recipient: to,
subject,
subject,
@@ -175,7 +180,7 @@ class EmailService {
@@ -175,7 +180,7 @@ class EmailService {
});
});
return true;
return true;
} catch (err) {
} catch (err) {
- this._recordDelivery({
+ await this._recordDelivery({
emailType,
emailType,
recipient: to,
recipient: to,
subject,
subject,
@@ -198,7 +203,7 @@ class EmailService {
@@ -198,7 +203,7 @@ class EmailService {
async sendNotificationEmail({ to, username, subject, notification }) {
async sendNotificationEmail({ to, username, subject, notification }) {
const htmlBody = this._buildEmailHtml({ username, notification });
const htmlBody = this._buildEmailHtml({ username, notification });
const textBody = this._buildEmailText({ username, notification });
const textBody = this._buildEmailText({ username, notification });
- const fromEmail = this.getFromEmail(notification.service);
+ const fromEmail = await this.getFromEmail(notification.service);
const sent = await this._sendEmail({
const sent = await this._sendEmail({
to,
to,
subject: subject || notification.title || 'Notification',
subject: subject || notification.title || 'Notification',
@@ -262,6 +267,7 @@ class EmailService {
@@ -262,6 +267,7 @@ class EmailService {
* @param {import('./notification-service').NotificationService} notifService
* @param {import('./notification-service').NotificationService} notifService
*/
*/
async processQueue(notifService) {
async processQueue(notifService) {
+ await this.ready;
if (!this.isEnabled) return;
if (!this.isEnabled) return;
const pending = await notifService.getPendingEmails();
const pending = await notifService.getPendingEmails();
@@ -271,11 +277,11 @@ class EmailService {
@@ -271,11 +277,11 @@ class EmailService {
let skipped = 0;
let skipped = 0;
for (const notif of pending) {
for (const notif of pending) {
- if (!notifService.shouldEmail(notif)) {
+ if (!(await notifService.shouldEmail(notif))) {
await notifService.markEmailed(notif.id);
await notifService.markEmailed(notif.id);
continue;
continue;
}
}
- const why = notifService.emailGuard ? notifService.emailGuard(notif) : null;
+ const why = notifService.emailGuard ? await notifService.emailGuard(notif) : null;
if (why) {
if (why) {
// Caps: leave the row un-emailed only if the cap may lift (it won't for stale/dup).
// Caps: leave the row un-emailed only if the cap may lift (it won't for stale/dup).
if (why === 'stale' || why === 'dup') await notifService.markEmailed(notif.id);
if (why === 'stale' || why === 'dup') await notifService.markEmailed(notif.id);
@@ -412,7 +418,7 @@ body { margin: 0; padding: 0; background: #1a1a24; font-family: -apple-system, B
@@ -412,7 +418,7 @@ body { margin: 0; padding: 0; background: #1a1a24; font-family: -apple-system, B
].join('\n');
].join('\n');
}
}
- async _escapeHtml(value) {
+ _escapeHtml(value) {
return String(value ?? '')
return String(value ?? '')
.replace(/&/g, '&')
.replace(/&/g, '&')
.replace(/</g, '<')
.replace(/</g, '<')
@@ -425,8 +431,9 @@ body { margin: 0; padding: 0; background: #1a1a24; font-family: -apple-system, B
@@ -425,8 +431,9 @@ body { margin: 0; padding: 0; background: #1a1a24; font-family: -apple-system, B
* Get current email service configuration status (for admin panel).
* Get current email service configuration status (for admin panel).
*/
*/
async getStatus() {
async getStatus() {
+ await this.ready;
const apiKey = await this.db.getSetting('resend_api_key') || '';
const apiKey = await this.db.getSetting('resend_api_key') || '';
- const issue = this.diagnose();
+ const issue = await this.diagnose();
const secrets = require('../secrets');
const secrets = require('../secrets');
return {
return {
enabled: this._enabled,
enabled: this._enabled,
@@ -452,7 +459,8 @@ body { margin: 0; padding: 0; background: #1a1a24; font-family: -apple-system, B
@@ -452,7 +459,8 @@ body { margin: 0; padding: 0; background: #1a1a24; font-family: -apple-system, B
* Throws with a descriptive error if email service is not ready.
* Throws with a descriptive error if email service is not ready.
*/
*/
async sendTestEmail(to) {
async sendTestEmail(to) {
- const issue = this.diagnose();
+ await this.ready;
+ const issue = await this.diagnose();
if (issue) throw new Error(issue);
if (issue) throw new Error(issue);
const subject = '🔥 OpenVibe — Test Email';
const subject = '🔥 OpenVibe — Test Email';
@@ -485,9 +485,9 @@ class NotificationService {
@@ -485,9 +485,9 @@ class NotificationService {
* Run periodic maintenance (call from setInterval in main server).
* Run periodic maintenance (call from setInterval in main server).
*/
*/
async maintenance() {
async maintenance() {
- const expired = this.cleanExpired();
+ const expired = await this.cleanExpired();
const maxAge = await this.db.getSetting?.('notification_max_age_days') || 90;
const maxAge = await this.db.getSetting?.('notification_max_age_days') || 90;
- const old = this.cleanOld(maxAge);
+ const old = await this.cleanOld(maxAge);
if (expired + old > 0) {
if (expired + old > 0) {
console.log(`[Notifications] Cleaned ${expired} expired + ${old} old notifications`);
console.log(`[Notifications] Cleaned ${expired} expired + ${old} old notifications`);
}
}
@@ -87,9 +87,9 @@ function createNetworkReadiness({ db, getKeys, release, ecosystem = null, discor
@@ -87,9 +87,9 @@ function createNetworkReadiness({ db, getKeys, release, ecosystem = null, discor
if (discordService && typeof discordService.isReady === 'function') {
if (discordService && typeof discordService.isReady === 'function') {
checks.push({
checks.push({
name: 'discord_bot', required: false,
name: 'discord_bot', required: false,
- check: () => {
+ check: async () => {
let configured = false;
let configured = false;
- try { configured = !!discordService._getSetting('discord_bot_token'); } catch { configured = false; }
+ try { configured = !!(await discordService._getSetting('discord_bot_token')); } catch { configured = false; }
if (!configured) return skip('not configured'); // verified nothing: skipped, never ok (WS-Q task 7)
if (!configured) return skip('not configured'); // verified nothing: skipped, never ok (WS-Q task 7)
return discordService.isReady() || 'bot token configured but the bot is not connected';
return discordService.isReady() || 'bot token configured but the bot is not connected';
},
},
@@ -107,7 +107,7 @@ async function receive(db, body, { notify, revise = null, now = Date.now(), env
@@ -107,7 +107,7 @@ async function receive(db, body, { notify, revise = null, now = Date.now(), env
const kept = kind === 'reminded' ? notices(prev) : [];
const kept = kind === 'reminded' ? notices(prev) : [];
for (const uid of users) {
for (const uid of users) {
try {
try {
- const created = notify(uid, n);
+ const created = await notify(uid, n);
if (created) out.notified += 1;
if (created) out.notified += 1;
if (created && created.id) kept.push([uid, created.id]);
if (created && created.id) kept.push([uid, created.id]);
} catch (e) { console.warn('[operator-alerts] notify failed:', e.message); }
} catch (e) { console.warn('[operator-alerts] notify failed:', e.message); }
@@ -120,7 +120,7 @@ async function receive(db, body, { notify, revise = null, now = Date.now(), env
@@ -120,7 +120,7 @@ async function receive(db, body, { notify, revise = null, now = Date.now(), env
? { title: `Resolved: ${a.name}`.slice(0, 120), icon: '✅', is_read: 1, message: `${a.summary} (resolved after ${ago(now - (prev.opened_at || now))}${prev.reopened ? `; it fired ${prev.reopened + 1} times` : ''}).`.slice(0, 500) }
? { title: `Resolved: ${a.name}`.slice(0, 120), icon: '✅', is_read: 1, message: `${a.summary} (resolved after ${ago(now - (prev.opened_at || now))}${prev.reopened ? `; it fired ${prev.reopened + 1} times` : ''}).`.slice(0, 500) }
: { title: `Alert: ${a.name}`.slice(0, 120), icon: critical ? '🚨' : '⚠️', is_read: critical ? 0 : 1, message: `${a.summary}. ${message(a)} Fired again ${(prev.reopened || 0) + 1} time(s) within 12 h of resolving.`.slice(0, 500) };
: { title: `Alert: ${a.name}`.slice(0, 120), icon: critical ? '🚨' : '⚠️', is_read: critical ? 0 : 1, message: `${a.summary}. ${message(a)} Fired again ${(prev.reopened || 0) + 1} time(s) within 12 h of resolving.`.slice(0, 500) };
for (const [uid, id] of notices(prev)) {
for (const [uid, id] of notices(prev)) {
- try { revise(uid, id, fields); } catch (e) { console.warn('[operator-alerts] revise failed:', e.message); }
+ try { await revise(uid, id, fields); } catch (e) { console.warn('[operator-alerts] revise failed:', e.message); }
}
}
}
}
return out;
return out;
new file mode 100644
new file mode 100644
@@ -0,0 +1,96 @@
@@ -0,0 +1,96 @@
+'use strict';
+// What the async conversion (plan T2) must not break, where the database handle is not the point: a rejected
+// async route handler answers 500 and the process lives (server/async-routes.js); the email service awaits its
+// settings, from-address, preferences and guards (a promise is always truthy, and '[object Promise]' is not an
+// address); the Discord status reads its settings. The database is a stub here: no query runs.
+const assert = require('assert');
+const express = require('express');
+require('../server/async-routes');
+const { EmailService } = require('../server/notifications/email-service');
+const { DiscordService } = require('../server/discord/discord-service');
+
+const quiet = async (fn) => { const l = console.log, w = console.warn, e = console.error; console.log = console.warn = console.error = () => {}; try { return await fn(); } finally { console.log = l; console.warn = w; console.error = e; } };
+const stubDb = (settings) => ({
+ prepare: () => ({ run: async () => ({ changes: 1 }), get: async () => null, all: async () => [] }),
+ getSetting: async (k) => (k in settings ? settings[k] : null),
+});
+
+(async () => {
+ // ── Express 4 + async handlers ──
+ let unhandled = null;
+ const onUnhandled = (e) => { unhandled = e; };
+ process.on('unhandledRejection', onUnhandled);
+ const app = express();
+ app.set('env', 'test'); // the default error handler answers 500 without logging the stack
+ app.get('/rejects', async () => { await null; throw new Error('the database is down'); });
+ app.get('/throws', () => { throw new Error('sync'); });
+ app.get('/ok', async (req, res) => { await null; res.json({ ok: true }); });
+ app.get('/passes', async (req, res, next) => { await null; next(); });
+ app.get('/passes', (req, res) => res.send('next route'));
+ const handled = express();
+ handled.set('env', 'test');
+ handled.get('/x', async () => { throw new Error('to the error handler'); });
+ handled.use(async (err, req, res, next) => { await null; res.status(503).json({ caught: err.message }); });
+ const listen = (a) => new Promise((r) => { const s = a.listen(0, '127.0.0.1', () => r(s)); });
+ const s1 = await listen(app); const s2 = await listen(handled);
+ const get = async (s, p) => { const r = await fetch(`http://127.0.0.1:${s.address().port}${p}`); return { status: r.status, text: await r.text() }; };
+ try {
+ assert.strictEqual((await get(s1, '/rejects')).status, 500, 'a rejected async handler answers 500');
+ assert.strictEqual((await get(s1, '/throws')).status, 500, 'a synchronous throw still answers 500');
+ assert.deepStrictEqual(await get(s1, '/ok'), { status: 200, text: '{"ok":true}' });
+ assert.deepStrictEqual(await get(s1, '/passes'), { status: 200, text: 'next route' }, 'next() from an async handler still reaches the next route');
+ assert.deepStrictEqual(await get(s2, '/x'), { status: 503, text: '{"caught":"to the error handler"}' }, 'the rejection reaches the app error handler');
+ await new Promise((r) => setImmediate(r));
+ assert.strictEqual(unhandled, null, 'no unhandled rejection');
+ } finally { s1.close(); s2.close(); process.off('unhandledRejection', onUnhandled); }
+ console.log('async route handlers: rejections answer 500');
+
+ // ── Email service ──
+ const email = new EmailService(stubDb({ email_enabled: true, resend_api_key: 're_test_key', email_from_address: '[email protected]', email_from_live: '[email protected]' }));
+ await quiet(() => email.ready);
+ assert.strictEqual(email.isEnabled, true, 'the settings are loaded once ready settles');
+ assert.strictEqual(email._escapeHtml('<b>"x"</b>'), '<b>"x"</b>', 'escaping is synchronous');
+ const sent = [];
+ email._sendEmail = async (o) => { sent.push(o); return true; };
+ await email.sendVerificationEmail({ to: '[email protected]', username: '<Ann>', verifyUrl: 'https://openvibe.network/verify-email?token=t' });
+ assert.ok(!/\[object Promise\]/.test(sent[0].htmlBody) && /Hey <Ann>,/.test(sent[0].htmlBody), 'the verification email is escaped text');
+ await email.sendNotificationEmail({ to: '[email protected]', username: 'ann', notification: { title: 'T', message: 'M', service: 'live', priority: 'normal' } });
+ assert.strictEqual(sent[1].fromEmail, '[email protected]', 'the per-service from-address is a string');
+ assert.ok(!/\[object Promise\]/.test(sent[1].htmlBody));
+ const status = await email.getStatus();
+ assert.strictEqual(status.issue, null, 'getStatus awaits diagnose()');
+ assert.strictEqual(status.ready, true);
+
+ // processQueue: an async shouldEmail() of false and an async guard skip, only the rest is sent.
+ const marked = [];
+ const n = (id, extra = {}) => ({ id, email: `u${id}@example.test`, username: `u${id}`, title: 'T', type: 'SYSTEM', priority: 'critical', ...extra });
+ const notif = {
+ getPendingEmails: async () => [n(1), n(2, { no: true }), n(3, { guard: 'stale' }), n(4, { guard: 'user-cap' })],
+ shouldEmail: async (x) => !x.no,
+ emailGuard: async (x) => x.guard || null,
+ markEmailed: async (id) => { marked.push(id); },
+ };
+ sent.length = 0;
+ await quiet(() => email.processQueue(notif));
+ assert.deepStrictEqual(sent.map((o) => o.to), ['[email protected]'], 'only the notification that passes both checks is sent');
+ assert.deepStrictEqual(marked.sort(), [1, 2, 3], 'sent, opted out and stale are marked; a lifted cap may still send later');
+
+ // Disabled: the failure reason recorded is diagnose()'s text, not a promise.
+ const off = new EmailService(stubDb({}));
+ await quiet(() => off.ready);
+ const logged = [];
+ off._recordDelivery = async (o) => { logged.push(o); };
+ assert.strictEqual(await off._sendEmail({ to: '[email protected]', subject: 's', htmlBody: '', textBody: '' }), false);
+ assert.strictEqual(typeof logged[0].errorMessage, 'string');
+ assert.match(logged[0].errorMessage, /Email is disabled/);
+ console.log('email service: settings, from-address, preferences and guards are awaited');
+
+ // ── Discord status ──
+ const discord = new DiscordService(stubDb({ discord_guild_id: 'g1', discord_alerts_channel_id: 'c1' }));
+ const ds = await discord.getStatus();
+ assert.strictEqual(ds.guildId, 'g1');
+ assert.strictEqual(ds.alertsChannelId, 'c1');
+ assert.strictEqual(ds.connected, false);
+ console.log('discord service: getStatus reads its settings');
+ console.log('async conversion: all checks passed');
+})().catch((err) => { console.error(err); process.exit(1); });
@@ -35,7 +35,8 @@ app.locals.config = { jwt: { issuer: 'https://openvibe.network', accessTokenExpi
@@ -35,7 +35,8 @@ app.locals.config = { jwt: { issuer: 'https://openvibe.network', accessTokenExpi
app.locals.privateKey = keys.privateKey;
app.locals.privateKey = keys.privateKey;
app.locals.publicKey = keys.publicKey;
app.locals.publicKey = keys.publicKey;
const revised = [];
const revised = [];
-app.locals.notificationService = { create: (n) => { sent.push(n); return { id: `n${sent.length}` }; }, revise: (id, uid, f) => { revised.push({ id, uid, ...f }); return true; } };
+// Async, like NotificationService.create/revise on PostgreSQL: the notification ids kept for the revisions come from the awaited result.
+app.locals.notificationService = { create: async (n) => { sent.push(n); return { id: `n${sent.length}` }; }, revise: async (id, uid, f) => { revised.push({ id, uid, ...f }); return true; } };
app.use('/oauth', require('../server/auth/oauth-routes'));
app.use('/oauth', require('../server/auth/oauth-routes'));
app.use('/internal', require('../server/internal/routes'));
app.use('/internal', require('../server/internal/routes'));
const server = http.createServer(app);
const server = http.createServer(app);
@@ -27,7 +27,7 @@ const freePort = () => new Promise((resolve) => { const s = net.createServer();
@@ -27,7 +27,7 @@ const freePort = () => new Promise((resolve) => { const s = net.createServer();
const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-net-budget-'));
const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-net-budget-'));
const port = await freePort();
const port = await freePort();
const child = spawn(process.execPath, [path.join(__dirname, '..', 'server', 'index.js')], {
const child = spawn(process.execPath, [path.join(__dirname, '..', 'server', 'index.js')], {
- env: { ...process.env, PORT: String(port), HOST: '127.0.0.1', DB_PATH: path.join(dir, 'network.db'), NODE_ENV: 'test' },
+ env: { ...process.env, PORT: String(port), HOST: '127.0.0.1', PGLITE_DIR: path.join(dir, 'pglite'), NODE_ENV: 'test' },
stdio: ['ignore', 'ignore', 'pipe'],
stdio: ['ignore', 'ignore', 'pipe'],
});
});
let stderr = '';
let stderr = '';
@@ -18,7 +18,7 @@ const freePort = () => new Promise((resolve) => { const s = net.createServer();
@@ -18,7 +18,7 @@ const freePort = () => new Promise((resolve) => { const s = net.createServer();
const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-net-stop-'));
const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-net-stop-'));
const port = await freePort();
const port = await freePort();
const child = spawn(process.execPath, [path.join(__dirname, '..', 'server', 'index.js')], {
const child = spawn(process.execPath, [path.join(__dirname, '..', 'server', 'index.js')], {
- env: { ...process.env, PORT: String(port), HOST: '127.0.0.1', DB_PATH: path.join(dir, 'network.db'), NODE_ENV: 'test' },
+ env: { ...process.env, PORT: String(port), HOST: '127.0.0.1', PGLITE_DIR: path.join(dir, 'pglite'), NODE_ENV: 'test' },
stdio: ['ignore', 'pipe', 'pipe'],
stdio: ['ignore', 'pipe', 'pipe'],
});
});
let out = '';
let out = '';