diff --git a/api/generated-schema.graphql b/api/generated-schema.graphql index b373873c11..0201eb59c5 100644 --- a/api/generated-schema.graphql +++ b/api/generated-schema.graphql @@ -3314,7 +3314,7 @@ type Query { vars: Vars! """Get information about all VMs on the system""" - vms: Vms! + vms: Vms parityHistory: [ParityCheck!]! array: UnraidArray! customization: Customization @@ -3326,7 +3326,7 @@ type Query { """Network interfaces""" networkInterfaces: [InfoNetworkInterface!]! - docker: Docker! + docker: Docker disks: [Disk!]! assignableDisks: [Disk!]! disk(id: PrefixedID!): Disk! diff --git a/api/scripts/qa/README.md b/api/scripts/qa/README.md new file mode 100644 index 0000000000..1abaf72478 --- /dev/null +++ b/api/scripts/qa/README.md @@ -0,0 +1,123 @@ +# Notification counter appliance test + +## Local burst and read-cost tests + +From the repository root, run: + +```sh +pnpm --filter ./api test src/unraid-api/graph/resolvers/notifications/notifications.load.spec.ts +CHOKIDAR_USEPOLLING=true pnpm --filter ./api test src/unraid-api/graph/resolvers/notifications/notifications.load.spec.ts +``` + +These tests use real temporary files, Chokidar events, notification readers, +and PubSub subscriptions. They cover creation of both native file copies, +content/severity changes, unread deletion revealing archive copies, and +archive-only creation. Fixtures include 1,000 unread plus 1,000 archive history +files and bursts of 24 or 100 notifications. + +Each `notification-load` record reports file-content read attempts, warning/alert +scan count, peak simultaneous scans, and elapsed time. Measurement starts after +hydration and ends after handlers and payload publications finish. Ground-truth +list queries run afterward and do not inflate the recorded event cost. Read +attempts include missing-copy reads; they are not physical disk I/O measurements. +Watcher metadata reads and other processes are outside this measurement. + +Assertions check final counters against full lists, final warning payloads, +deduplicated additions, no history recount, no reads of unrelated archived +payloads, and no more than two count reads per event plus the unread payload +scans. They require at most one active payload scan and at most four scans for +each large-history test burst. Archive-only bursts must trigger no warning/alert +scans. This limits the fixture's total read budget rather than only its counter +reads. Separate broadcast tests hold reads/publication open to prove that +overlapping requests combine, a mid-scan request produces a fresh final payload, +and read/publication failures do not block later requests. + +Before refresh scheduling, 100 changes with 1,000 unread history files caused +110,200 content-read attempts and 100 overlapping scans. With scheduling, the +same native-watcher workload used 1,300 attempts and one scan in a local run; +polling used 2,400 attempts and two sequential scans. Incoming events during a +scan can require a follow-up scan; there is no fixed scan-count guarantee for an +unbounded stream of events. Elapsed times remain diagnostic, not machine-dependent +CI limits. These local tests do not establish appliance latency or physical disk +I/O performance. Appliance acceptance remains a separate step. + +## Appliance setup + +`notification-counter.mjs` tests the installed API against Unraid's native +`/usr/local/emhttp/webGui/scripts/notify`. Run it inside a disposable, managed +QA guest after installing the production-shaped plugin for the selected commit. +The guest must have `/boot/config/qa-harness-ok` and Node.js with native `fetch`. + +The QA runner must establish the source commit and package digest before +installation. Extract the API entrypoint from that same package and calculate +its SHA-256. Supply that expected hash to the guest; do not calculate the +expected value from the guest's installed file. The script compares the +installed entrypoint with this hash and records the native notifier's hash. +This comparison covers the entrypoint. The runner remains responsible for +verifying the complete installed package and its revision. + +Supply these environment variables through the managed guest execution path: + +- `QA_GRAPHQL_URL`: authenticated local GraphQL endpoint for this guest. +- `QA_API_KEY`: temporary key with notification read, update, and delete permissions. +- `QA_API_ENTRYPOINT`: absolute installed API entrypoint path from the package. +- `QA_API_SHA256`: expected SHA-256 from the pinned package. +- `QA_SOURCE_REVISION`: full 40-character source commit for that package. +- `QA_REPORT_PATH`: fresh absolute path for the private JSON receipt. + +Run `node /absolute/guest/path/notification-counter.mjs`. Use the QA VM +provider's current lifecycle and execution contract. Do not run this script on +a production server. The runner must restore and release the guest even when +the test fails, and remove its temporary API key. + +The script uses `notify -i " 1"` to enable browser-file delivery only. +It does not change notification settings or invoke email and notification agents. +It creates uniquely named files and removes only those files in `finally`. +It verifies: + +- All three severities through the actual notifier, including both unread and + archive copies of each notification. +- Native archive actions, API restore, API archive over an existing copy, and API deletion. +- External severity changes and unread deletion that reveals the archive's severity. +- Native archive-only delivery and a burst of 24 native file writes. +- Matching cached counters, visible GraphQL lists, and independent disk counts + after every step and after cleanup. + +The script never calls `recalculateOverview`, so a watcher failure cannot be +hidden by a manual refresh. Each check allows 15 seconds for watcher delivery. +It stops if notification ground truth is ambiguous or a list reaches its +10,000-item cap. Existing notifications must have a supported importance value. +The receipt contains hashes, source revision, fixture counts, and check names; +it excludes credentials, the endpoint, and existing notification contents. + +A syntax check is not appliance acceptance. Only a receipt with `status=passed` +and `cleanup=complete`, plus the runner's package and VM lifecycle evidence, +proves that this appliance test passed. + +## Adversarial appliance pass + +Set `QA_ADVERSARIAL=true` and `QA_ALLOW_SERVICE_RESTART=true` to run the extended +pass. This requires a disposable guest, an isolated empty notification directory, +and native notification settings `normal=1`, `warning=1`, and `alert=1`. The runner +must prepare those settings before restarting the API, preserve the original +configuration, and restore it afterward. The temporary API key also needs +notification creation permission. With the current CLI, use an alphanumeric key +name, an explicit `GUEST` role, and the four explicit notification permissions. + +The extended pass checks API creation through the installed native script, +three authenticated GraphQL WebSocket subscriptions, content updates without +count changes, selected-ID and severity-filtered bulk archive/restore, all-item +archive/restore/deletion, persisted counts after service restart, subscription +reconnection, and recovery from a real startup directory failure without +another process restart. + +It also temporarily lowers the guest's inotify watch limit, creates a watched +subdirectory, requires a logged `ENOSPC`, restores the original limit in +`finally`, and proves that the same API process still counts native writes. +The runner must restore and release the VM on every outcome. This fault test +must never run on a shared or production guest. + +The receipt separately records whether the ID returned by native API creation +matches the actual disk record. A mismatch produces `passed_with_findings`, +which is not a clean regression verdict. The runner must report that finding +and establish whether it predates the candidate before deciding its scope. diff --git a/api/scripts/qa/notification-counter.mjs b/api/scripts/qa/notification-counter.mjs new file mode 100644 index 0000000000..f97b717965 --- /dev/null +++ b/api/scripts/qa/notification-counter.mjs @@ -0,0 +1,424 @@ +#!/usr/bin/env node +// Run inside a managed, disposable Unraid QA guest after installing the pinned plugin. +import assert from 'node:assert/strict'; +import { createHash, randomUUID } from 'node:crypto'; +import { execFileSync } from 'node:child_process'; +import { access, mkdir, readFile, readdir, rename, rm, unlink, writeFile } from 'node:fs/promises'; +import { createRequire } from 'node:module'; +import { isAbsolute, join } from 'node:path'; +import { setTimeout as delay } from 'node:timers/promises'; + +const required = (name) => { + assert(process.env[name], `Missing ${name}`); + return process.env[name]; +}; +const endpoint = required('QA_GRAPHQL_URL'); +const key = required('QA_API_KEY'); +const entrypoint = required('QA_API_ENTRYPOINT'); +const expectedHash = required('QA_API_SHA256'); +const revision = required('QA_SOURCE_REVISION'); +const reportPath = required('QA_REPORT_PATH'); +assert(/^[a-f0-9]{64}$/.test(expectedHash), 'Expected packaged API SHA-256'); +assert(/^[a-f0-9]{40}$/.test(revision), 'Expected exact source commit'); +assert(isAbsolute(entrypoint) && isAbsolute(reportPath), 'Use absolute guest paths'); +const notifier = '/usr/local/emhttp/webGui/scripts/notify'; +const prefix = `qa1859_${randomUUID().replaceAll('-', '')}`; +const checks = []; +const report = { revision, checks, status: 'running' }; +let notificationPath; +let stage = 'preflight'; +const hash = (bytes) => createHash('sha256').update(bytes).digest('hex'); +const counts = '{info warning alert total}'; +const overviewFields = `unread ${counts} archive ${counts}`; +const query = `query { notifications { + overview { ${overviewFields} } + unread: list(filter: {type: UNREAD, offset: 0, limit: 10000}) {id title subject description importance link} + archive: list(filter: {type: ARCHIVE, offset: 0, limit: 10000}) {id title subject description importance link} +} }`; + +async function graphql(query, variables = {}) { + const response = await fetch(endpoint, { + method: 'POST', + headers: { 'content-type': 'application/json', 'x-api-key': key }, + body: JSON.stringify({ query, variables }), + signal: AbortSignal.timeout(10000), + }); + assert.equal(response.status, 200, `GraphQL HTTP ${response.status}`); + const body = await response.json(); + assert(!body.errors?.length, `GraphQL errors: ${body.errors?.map((e) => e.message).join('; ')}`); + return body.data; +} + +const emptyCounts = () => ({ info: 0, warning: 0, alert: 0, total: 0 }); +function summarize(notifications) { + const result = emptyCounts(); + for (const notification of notifications) { + const importance = notification.importance.toLowerCase(); + assert(Object.hasOwn(result, importance) && importance !== 'total', 'Unknown importance'); + result[importance]++; + result.total++; + } + return result; +} +const localId = (id) => id.split(':').at(-1); +async function diskState() { + const state = {}; + for (const type of ['unread', 'archive']) { + state[type] = []; + for (const id of await readdir(join(notificationPath, type))) { + if (!id.endsWith('.notify')) continue; + const text = await readFile(join(notificationPath, type, id), 'utf8'); + const severity = text.match(/^importance\s*=\s*"?(normal|warning|alert)"?\s*$/m)?.[1]; + assert(severity, 'Disk fixture has unsupported importance; cannot establish ground truth'); + state[type].push({ id, importance: severity === 'normal' ? 'INFO' : severity.toUpperCase() }); + } + } + const unreadIds = new Set(state.unread.map((n) => n.id)); + state.archive = state.archive.filter((n) => !unreadIds.has(n.id)); + return state; +} + +async function verify(label, fixtureCounts) { + stage = label; + const deadline = Date.now() + 15000; + let lastError; + do { + try { + const disk = await diskState(); + const { notifications } = await graphql(query); + const fixtureOverview = {}; + for (const type of ['unread', 'archive']) { + const list = notifications[type]; + assert(list.length < 10000, 'List cap reached; cannot establish complete ground truth'); + assert.deepEqual(list.map((n) => localId(n.id)).sort(), disk[type].map((n) => n.id).sort()); + assert.deepEqual(notifications.overview[type], summarize(disk[type])); + assert.deepEqual(notifications.overview[type], summarize(list)); + fixtureOverview[type] = summarize(list.filter((n) => n.title.startsWith(prefix))); + assert.equal(fixtureOverview[type].total, fixtureCounts[type]); + } + checks.push({ label, fixtureOverview, diskMatchesListsAndCounters: true }); + console.log(`PASS ${label}`); + return notifications; + } catch (error) { + lastError = error; + await delay(150); + } + } while (Date.now() < deadline); + throw new Error(`${label}: ${lastError.message}`); +} + +function notify(event, severity, archiveOnly = false) { + execFileSync(notifier, [ + '-e', event, '-s', 'Notification counter QA', '-d', 'Disposable notification fixture', + '-i', `${severity} 1`, '-x', ...(archiveOnly ? ['-b'] : []), + ], { timeout: 10000, stdio: 'pipe' }); + return `${event}_ticket.notify`; +} +async function mutation(name, id) { + return graphql(`mutation($id: PrefixedID!) { ${name}(id: $id) { id } }`, { id }); +} +async function cleanup() { + if (!notificationPath) return; + for (const type of ['unread', 'archive']) { + for (const id of await readdir(join(notificationPath, type))) { + if (id.startsWith(prefix) && id.endsWith('.notify')) { + await unlink(join(notificationPath, type, id)); + } + } + } + await verify('fixture cleanup', { unread: 0, archive: 0 }); +} + +async function waitUntil(predicate, label, timeout = 15000) { + const deadline = Date.now() + timeout; + while (Date.now() < deadline) { + if (await predicate()) return; + await delay(100); + } + throw new Error(`Timed out: ${label}`); +} + +function apiProcess() { + const require = createRequire(entrypoint); + const list = JSON.parse(execFileSync(process.execPath, [require.resolve('pm2/bin/pm2'), 'jlist'], { + encoding: 'utf8', timeout: 10000, + env: { ...process.env, PM2_HOME: '/var/log/.pm2', PM2_DISCRETE_MODE: 'true' }, + stdio: ['ignore', 'pipe', 'pipe'], + })); + const app = list.find((app) => app.name === 'unraid-api' && app.pm2_env.pm_exec_path === entrypoint); + assert(app?.pid > 0, 'API process not running'); + return { pid: app.pid, restarts: app.pm2_env.restart_time }; +} + +async function observeSubscriptions() { + const require = createRequire(entrypoint); + const WebSocket = require('ws'); + const socket = new WebSocket(endpoint.replace(/^http/, 'ws'), 'graphql-transport-ws', { + headers: { 'x-api-key': key }, + }); + const messages = []; + let failure = false; + socket.on('error', () => { failure = true; }); + socket.on('message', (bytes) => { + try { messages.push(JSON.parse(bytes.toString())); } catch { failure = true; } + }); + socket.on('open', () => socket.send(JSON.stringify({ + type: 'connection_init', payload: { 'x-api-key': key }, + }))); + const until = async (predicate, label) => waitUntil(() => { + assert(!failure && !messages.some((m) => m.type === 'error' || m.payload?.errors), 'Subscription failed'); + return predicate(); + }, label); + try { + await until(() => messages.some((m) => m.type === 'connection_ack'), 'WebSocket acknowledgement'); + const fields = { + added: 'notificationAdded {id title subject importance}', + overview: `notificationsOverview {${overviewFields}}`, + warnings: 'notificationsWarningsAndAlerts {id title subject importance}', + }; + for (const [id, field] of Object.entries(fields)) { + socket.send(JSON.stringify({ id, type: 'subscribe', payload: { query: `subscription {${field}}` } })); + } + return { + messages, + until, + close: async () => { + socket.close(); + try { await waitUntil(() => socket.readyState === WebSocket.CLOSED, 'WebSocket close', 2000); } + finally { if (socket.readyState !== WebSocket.CLOSED) socket.terminate(); } + }, + }; + } catch (error) { + socket.terminate(); + throw error; + } +} + +async function adversarial() { + stage = 'adversarial preflight'; + assert.equal(process.env.QA_ALLOW_SERVICE_RESTART, 'true', 'Explicit disposable-guest restart permission required'); + const config = JSON.parse(execFileSync('/usr/bin/php', ['-r', + '$docroot="/usr/local/emhttp"; require_once "$docroot/webGui/include/Wrappers.php"; ' + + '$c=parse_plugin_cfg("dynamix",true); echo json_encode($c["notify"]);', + ], { encoding: 'utf8', timeout: 10000 })); + for (const severity of ['normal', 'warning', 'alert']) { + assert.equal(Number(config[severity]), 1, 'Native API creation requires browser-only notification settings'); + } + const baseline = await graphql(query); + assert.equal(baseline.notifications.unread.length + baseline.notifications.archive.length, 0, + 'Adversarial bulk tests require an isolated empty notification directory'); + let observer = await observeSubscriptions(); + try { + stage = 'native API creation and subscriptions'; + const title = `${prefix}_api`; + const { createNotification: created } = await graphql(`mutation($input: NotificationData!) { + createNotification(input: $input) {id title subject description importance link} + }`, { input: { title, subject: 'Quoted "subject" = value', description: 'Notification QA & content', + importance: 'WARNING', link: '/Settings/Notifications' } }); + const state = await verify('API creation through installed native notifier', { unread: 1, archive: 0 }); + const actual = state.unread.find((n) => n.title === title); + assert(actual, 'Created notification absent from list'); + // A native archive copy proves the API did not take its local-write fallback. + await access(join(notificationPath, 'archive', localId(actual.id))); + assert.equal(created.subject, 'Quoted "subject" = value'); + for (const field of ['title', 'subject', 'description', 'importance', 'link']) { + assert.equal(actual[field], created[field], 'Native file metadata differs from API creation response'); + } + report.nativeCreateReturnedIdMatchesDisk = localId(created.id) === localId(actual.id); + await observer.until(() => observer.messages.some((m) => m.id === 'added' + && m.payload?.data?.notificationAdded?.title === title), 'live notificationAdded'); + await observer.until(() => observer.messages.some((m) => m.id === 'overview' + && m.payload?.data?.notificationsOverview?.unread.warning === 1), 'live notificationsOverview'); + await observer.until(() => observer.messages.some((m) => m.id === 'warnings' + && m.payload?.data?.notificationsWarningsAndAlerts?.some((n) => n.title === title)), 'live warning subscription'); + checks.push({ label: 'three authenticated live GraphQL subscriptions', passed: true }); + const original = await readFile(join(notificationPath, 'unread', localId(actual.id)), 'utf8'); + await writeFile(join(notificationPath, 'unread', localId(actual.id)), + original.replace(/^subject=.*$/m, 'subject="Changed content"')); + await observer.until(() => observer.messages.some((m) => m.id === 'warnings' + && m.payload?.data?.notificationsWarningsAndAlerts?.some((n) => n.title === title + && n.subject === 'Changed content')), 'warning content update without count change'); + checks.push({ label: 'live content update at unchanged severity', passed: true }); + await mutation('archiveNotification', actual.id); + await verify('API-created notification can be archived by its listed ID', { unread: 0, archive: 1 }); + await observer.until(() => observer.messages.some((m) => m.id === 'warnings' + && m.payload?.data?.notificationsWarningsAndAlerts?.length === 0), 'warning removal subscription'); + await cleanup(); + + stage = 'bulk regression'; + const ids = ['normal', 'warning', 'alert'].map((s) => notify(`${prefix}_bulk_${s}`, s)); + await verify('bulk seed', { unread: 3, archive: 0 }); + await observer.until(() => ids.every((id) => observer.messages.some((m) => m.id === 'added' + && localId(m.payload?.data?.notificationAdded?.id ?? '') === id)), 'native bulk add events'); + for (const id of ids) { + const additions = observer.messages.filter((m) => m.id === 'added' + && localId(m.payload?.data?.notificationAdded?.id ?? '') === id); + assert.equal(additions.length, 1, 'Native add was lost or published more than once'); + } + checks.push({ label: 'one live notificationAdded per new native file', passed: true }); + notify(`${prefix}_bulk_normal`, 'normal'); + await verify('repeated native ticket creates no duplicate', { unread: 3, archive: 0 }); + await graphql(`mutation($ids: [PrefixedID!]!) {archiveNotifications(ids: $ids) {${overviewFields}}}`, { ids: ids.slice(0, 2) }); + await verify('bulk archive selected IDs', { unread: 1, archive: 2 }); + await graphql(`mutation($ids: [PrefixedID!]!) {unarchiveNotifications(ids: $ids) {${overviewFields}}}`, { ids: ids.slice(0, 2) }); + await verify('bulk restore selected IDs', { unread: 3, archive: 0 }); + await graphql(`mutation {archiveAll(importance: WARNING) {${overviewFields}}}`); + await verify('bulk archive by severity', { unread: 2, archive: 1 }); + await graphql(`mutation {unarchiveAll(importance: WARNING) {${overviewFields}}}`); + await verify('bulk restore by severity', { unread: 3, archive: 0 }); + await graphql(`mutation {archiveAll {${overviewFields}}}`); + await verify('archive all', { unread: 0, archive: 3 }); + await graphql(`mutation {unarchiveAll {${overviewFields}}}`); + await verify('restore all', { unread: 3, archive: 0 }); + await graphql(`mutation {archiveAll {${overviewFields}}}`); + await graphql(`mutation {deleteArchivedNotifications {${overviewFields}}}`); + await verify('delete all archived', { unread: 0, archive: 0 }); + await observer.close(); + observer = null; + + stage = 'restart persistence'; + const persistent = notify(`${prefix}_restart`, 'alert'); + await verify('before service restart', { unread: 1, archive: 0 }); + const before = apiProcess(); + execFileSync('/etc/rc.d/rc.unraid-api', ['restart'], { timeout: 60000, stdio: 'pipe' }); + await verify('rehydrates persisted native copies after restart', { unread: 1, archive: 0 }); + assert.notEqual(apiProcess().pid, before.pid, 'API process did not restart'); + observer = await observeSubscriptions(); + const reconnect = notify(`${prefix}_reconnect`, 'warning'); + await verify('new native write after restart', { unread: 2, archive: 0 }); + await observer.until(() => observer.messages.some((m) => m.id === 'added' + && localId(m.payload?.data?.notificationAdded?.id ?? '') === reconnect), 'subscription after restart'); + checks.push({ label: 'authenticated live subscription after restart', passed: true }); + await observer.close(); + observer = null; + + stage = 'startup filesystem failure'; + const unread = join(notificationPath, 'unread'); + const saved = join(notificationPath, `${prefix}_saved_unread`); + await rename(unread, saved); + await writeFile(unread, 'blocked directory'); + let startupError; + try { + execFileSync('/etc/rc.d/rc.unraid-api', ['restart'], { timeout: 60000, stdio: 'pipe' }); + const failedProcess = apiProcess(); + const response = await fetch(endpoint, { method: 'POST', + headers: { 'content-type': 'application/json', 'x-api-key': key }, + body: JSON.stringify({ query }), signal: AbortSignal.timeout(10000) }); + const body = await response.json(); + assert(body.errors?.length, 'Invalid directory did not expose an initialization error'); + await unlink(unread); + await rename(saved, unread); + await verify('recovers startup directory failure without another restart', { unread: 2, archive: 0 }); + assert.deepEqual(apiProcess(), failedProcess, 'Process restarted instead of recovering'); + } catch (error) { + startupError = error; + throw error; + } finally { + try { await access(saved); await rm(unread, { recursive: true, force: true }); await rename(saved, unread); } + catch (error) { + if (error.code !== 'ENOENT') { + report.startupCleanupErrorType = error.name; + if (!startupError) throw error; + } + } + } + await mutation('archiveNotification', persistent); + await verify('mutation after initialization recovery', { unread: 1, archive: 1 }); + + stage = 'runtime native watcher error'; + const limitPath = '/proc/sys/fs/inotify/max_user_watches'; + const originalLimit = await readFile(limitPath, 'utf8'); + const logPath = '/var/log/graphql-api.log'; + const oldLogLength = (await readFile(logPath)).length; + const running = apiProcess(); + const faultDirectory = join(notificationPath, `${prefix}_watch_fault`); + try { + await writeFile(limitPath, '1\n'); + await mkdir(faultDirectory); + await waitUntil(async () => (await readFile(logPath)).subarray(oldLogLength).toString('utf8').includes('ENOSPC'), + 'actual inotify resource exhaustion'); + } finally { + await writeFile(limitPath, originalLimit); + await rm(faultDirectory, { recursive: true, force: true }); + } + assert.deepEqual(apiProcess(), running, 'Watcher error restarted the API process'); + notify(`${prefix}_after_watch_error`, 'normal'); + await verify('counts follow real writes after logged ENOSPC without process restart', { unread: 2, archive: 1 }); + checks.push({ label: 'native inotify error logged; process and counts remain operational', passed: true }); + } finally { + await observer?.close(); + } +} + +// Write a fresh private report before mutations; never overwrite another run's evidence. +await writeFile(reportPath, JSON.stringify(report), { flag: 'wx', mode: 0o600 }); +try { + await access('/boot/config/qa-harness-ok'); + const apiBytes = await readFile(entrypoint); + report.apiSha256 = hash(apiBytes); + assert.equal(report.apiSha256, expectedHash, 'Installed API differs from pinned package'); + const script = await readFile(notifier, 'utf8'); + report.notifierSha256 = hash(script); + assert(script.includes('$overrule') && script.includes('strtok'), 'Notifier delivery override unavailable'); + // The native parser resolves defaults and persistent settings exactly as notify does. + const config = JSON.parse(execFileSync('/usr/bin/php', ['-r', + '$docroot="/usr/local/emhttp"; require_once "$docroot/webGui/include/Wrappers.php"; ' + + '$c=parse_plugin_cfg("dynamix",true); echo json_encode($c["notify"]["path"] ?? "/tmp/notifications");', + ], { timeout: 10000, encoding: 'utf8' })); + assert(typeof config === 'string' && isAbsolute(config), 'Invalid configured notification path'); + notificationPath = config; + await verify('baseline', { unread: 0, archive: 0 }); + const ids = ['normal', 'warning', 'alert'].map((severity) => notify(`${prefix}_${severity}`, severity)); + for (const id of ids) { + await access(join(notificationPath, 'unread', id)); + await access(join(notificationPath, 'archive', id)); + } + const created = await verify('native dual writes, all severities', { unread: 3, archive: 0 }); + assert.deepEqual(summarize(created.unread.filter((n) => n.title.startsWith(prefix))), + { info: 1, warning: 1, alert: 1, total: 3 }); + execFileSync(notifier, ['archive', ids[0]], { timeout: 10000, stdio: 'pipe' }); + await verify('native archive reveals existing archive copy', { unread: 2, archive: 1 }); + await mutation('unreadNotification', ids[0]); + await verify('API restore', { unread: 3, archive: 0 }); + await mutation('archiveNotification', ids[1]); + await verify('API archive replaces duplicate', { unread: 2, archive: 1 }); + await graphql(`mutation($id: PrefixedID!) { + deleteNotification(id: $id, type: ARCHIVE) { ${overviewFields} } + }`, { id: ids[1] }); + await verify('API archive deletion', { unread: 2, archive: 0 }); + const alertPath = join(notificationPath, 'unread', ids[2]); + const text = await readFile(alertPath, 'utf8'); + await writeFile(alertPath, text.replace(/^importance=.*$/m, 'importance="warning"')); + const changed = await verify('external severity change', { unread: 2, archive: 0 }); + assert.deepEqual(summarize(changed.unread.filter((n) => n.title.startsWith(prefix))), + { info: 1, warning: 1, alert: 0, total: 2 }); + await unlink(alertPath); + await verify('external unread deletion reveals archive severity', { unread: 1, archive: 1 }); + notify(`${prefix}_archive_only`, 'normal', true); + await verify('native archive-only write', { unread: 1, archive: 2 }); + for (let i = 0; i < 12; i++) notify(`${prefix}_burst_${i}`, ['normal', 'warning', 'alert'][i % 3]); + await verify('native burst of 24 file writes', { unread: 13, archive: 2 }); + if (process.env.QA_ADVERSARIAL === 'true') { + await cleanup(); + await adversarial(); + } + report.status = report.nativeCreateReturnedIdMatchesDisk === false ? 'passed_with_findings' : 'passed'; + if (report.status === 'passed_with_findings') process.exitCode = 1; +} catch (error) { + report.status = 'failed'; + // Never include credentials, endpoint, or existing notification contents in the receipt. + report.failedStage = stage; + report.errorType = error.name; + process.exitCode = 1; +} finally { + try { + await cleanup(); + report.cleanup = 'complete'; + } catch { + report.cleanup = 'failed'; + report.status = 'failed'; + process.exitCode = 1; + } + await writeFile(reportPath, `${JSON.stringify(report, null, 2)}\n`, { mode: 0o600 }); + console.log(`Result: ${report.status}; cleanup: ${report.cleanup}`); +} diff --git a/api/src/unraid-api/cli/generated/graphql.ts b/api/src/unraid-api/cli/generated/graphql.ts index 5119a999ad..777ffdc7cc 100644 --- a/api/src/unraid-api/cli/generated/graphql.ts +++ b/api/src/unraid-api/cli/generated/graphql.ts @@ -2383,7 +2383,7 @@ export type Query = { disk: Disk; disks: Array; display: InfoDisplay; - docker: Docker; + docker?: Maybe; flash: Flash; /** Get JSON Schema for API key creation form */ getApiKeyCreationFormSchema: ApiKeyFormSettings; @@ -2447,7 +2447,7 @@ export type Query = { validateOidcSession: OidcSessionValidation; vars: Vars; /** Get information about all VMs on the system */ - vms: Vms; + vms?: Maybe; }; diff --git a/api/src/unraid-api/graph/partial-results.spec.ts b/api/src/unraid-api/graph/partial-results.spec.ts new file mode 100644 index 0000000000..cf80f5dcdc --- /dev/null +++ b/api/src/unraid-api/graph/partial-results.spec.ts @@ -0,0 +1,141 @@ +import { ApolloDriver, ApolloDriverConfig } from '@nestjs/apollo'; +import { GraphQLModule, GraphQLSchemaHost, Query, Resolver } from '@nestjs/graphql'; +import { FastifyAdapter, NestFastifyApplication } from '@nestjs/platform-fastify'; +import { Test } from '@nestjs/testing'; +import { readFile } from 'fs/promises'; + +import { PrefixedID } from '@unraid/shared/prefixed-id-scalar.js'; +import { buildSchema } from 'graphql'; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { DockerResolver } from '@app/unraid-api/graph/resolvers/docker/docker.resolver.js'; +import { DockerService } from '@app/unraid-api/graph/resolvers/docker/docker.service.js'; +import { VmsResolver } from '@app/unraid-api/graph/resolvers/vms/vms.resolver.js'; +import { VmsService } from '@app/unraid-api/graph/resolvers/vms/vms.service.js'; +import { SubscriptionTrackerService } from '@app/unraid-api/graph/services/subscription-tracker.service.js'; + +@Resolver() +class ServerStatusResolver { + @Query(() => String) + serverStatus() { + return 'available'; + } +} + +type Result = { + data: Record | null; + errors?: { path: (string | number)[] }[]; +}; + +describe('Partial GraphQL results for unavailable services', () => { + let app: NestFastifyApplication; + const vms = { getDomains: vi.fn() }; + const docker = { getNetworks: vi.fn() }; + + beforeAll(async () => { + const module = await Test.createTestingModule({ + imports: [ + GraphQLModule.forRoot({ + driver: ApolloDriver, + autoSchemaFile: true, + }), + ], + providers: [ + PrefixedID, + ServerStatusResolver, + VmsResolver, + DockerResolver, + { provide: VmsService, useValue: vms }, + { provide: DockerService, useValue: docker }, + { + provide: SubscriptionTrackerService, + useValue: { registerTopic: vi.fn() }, + }, + ], + }) + .useMocker(() => ({})) + .compile(); + app = module.createNestApplication(new FastifyAdapter()); + await app.init(); + await app.getHttpAdapter().getInstance().ready(); + }); + + beforeEach(() => { + vms.getDomains.mockReset().mockResolvedValue([]); + docker.getNetworks.mockReset().mockResolvedValue([]); + }); + + afterAll(async () => app?.close()); + + async function execute(query: string): Promise { + const response = await app.inject({ method: 'POST', url: '/graphql', payload: { query } }); + expect(response.statusCode).toBe(200); + return response.json(); + } + + it('keeps the runtime and published service roots nullable', async () => { + const published = buildSchema(await readFile('generated-schema.graphql', 'utf8')); + const runtime = app.get(GraphQLSchemaHost).schema; + for (const schema of [published, runtime]) { + for (const root of ['vms', 'docker']) { + expect(schema.getQueryType()!.getFields()[root].type.toString()).toBe( + root === 'vms' ? 'Vms' : 'Docker' + ); + } + } + }); + + it.each(['domain', 'domains'])('preserves other roots when vms.%s fails', async (field) => { + vms.getDomains.mockRejectedValue(new Error('VMs are not available')); + const result = await execute( + `{ serverStatus vms { ${field} { id } } docker { networks { id } } }` + ); + expect(result.data).toEqual({ + serverStatus: 'available', + vms: { [field]: null }, + docker: { networks: [] }, + }); + expect(result.errors?.map((error) => error.path)).toEqual([['vms', field]]); + }); + + it('preserves other roots when Docker fails', async () => { + docker.getNetworks.mockRejectedValue(new Error('Docker is not available')); + const result = await execute( + '{ serverStatus vms { domains { id } } docker { networks { id } } }' + ); + expect(result.data).toEqual({ serverStatus: 'available', vms: { domains: [] }, docker: null }); + expect(result.errors?.map((error) => error.path)).toEqual([['docker', 'networks']]); + }); + + it('retains successful data and both errors when both services fail', async () => { + vms.getDomains.mockRejectedValue(new Error('VMs are not available')); + docker.getNetworks.mockRejectedValue(new Error('Docker is not available')); + const result = await execute( + '{ serverStatus guests: vms { domains { id } } containers: docker { networks { id } } }' + ); + expect(result.data).toEqual({ + serverStatus: 'available', + guests: { domains: null }, + containers: null, + }); + expect(result.errors?.map((error) => error.path)).toEqual( + expect.arrayContaining([ + ['guests', 'domains'], + ['containers', 'networks'], + ]) + ); + expect(result.errors).toHaveLength(2); + }); + + it('preserves successful empty service results', async () => { + const result = await execute( + '{ serverStatus vms { domains { id } } docker { networks { id } } }' + ); + expect(result.errors).toBeUndefined(); + expect(result.data).toEqual({ + serverStatus: 'available', + vms: { domains: [] }, + docker: { networks: [] }, + }); + }); +}); diff --git a/api/src/unraid-api/graph/resolvers/docker/docker.resolver.ts b/api/src/unraid-api/graph/resolvers/docker/docker.resolver.ts index e8d025c5f8..cc67450dc8 100644 --- a/api/src/unraid-api/graph/resolvers/docker/docker.resolver.ts +++ b/api/src/unraid-api/graph/resolvers/docker/docker.resolver.ts @@ -64,7 +64,7 @@ export class DockerResolver { action: AuthAction.READ_ANY, resource: Resource.DOCKER, }) - @Query(() => Docker) + @Query(() => Docker, { nullable: true }) public docker() { return { id: 'docker', diff --git a/api/src/unraid-api/graph/resolvers/notifications/loadNotificationsFile.test.ts b/api/src/unraid-api/graph/resolvers/notifications/loadNotificationsFile.test.ts index 09b3663e45..217db37827 100644 --- a/api/src/unraid-api/graph/resolvers/notifications/loadNotificationsFile.test.ts +++ b/api/src/unraid-api/graph/resolvers/notifications/loadNotificationsFile.test.ts @@ -2,6 +2,7 @@ import { ConfigService } from '@nestjs/config'; import { existsSync, mkdirSync, rmSync, writeFileSync } from 'fs'; +import { EventEmitter } from 'node:events'; import { tmpdir } from 'os'; import { join } from 'path'; @@ -15,14 +16,16 @@ import { } from '@app/unraid-api/graph/resolvers/notifications/notifications.model.js'; import { NotificationsService } from '@app/unraid-api/graph/resolvers/notifications/notifications.service.js'; -const { mockWatch } = vi.hoisted(() => { - const watcher = { - on: vi.fn().mockReturnThis(), - close: vi.fn().mockResolvedValue(undefined), - }; - +const { mockWatch } = await vi.hoisted(async () => { + const { EventEmitter } = await import('node:events'); return { - mockWatch: vi.fn(() => watcher), + mockWatch: vi.fn(() => { + const watcher = Object.assign(new EventEmitter(), { + close: vi.fn().mockResolvedValue(undefined), + }); + setImmediate(() => watcher.emit('ready')); + return watcher; + }), }; }); @@ -129,52 +132,34 @@ describe('NotificationsService - loadNotificationFile (minimal mocks)', () => { ); }); - it('replays buffered add events after overview hydration', async () => { - const bufferedPath = `${testNotificationsDir}/unread/buffered.notify`; - const hydratedService = createNotificationsService(); - const processNotificationAdd = vi.fn().mockResolvedValue(undefined); - const handleNotificationAdd = ( - Reflect.get(hydratedService, 'handleNotificationAdd') as (path: string) => Promise - ).bind(hydratedService); - - Reflect.set( - hydratedService, - 'ensureNotificationDirectories', - vi.fn().mockResolvedValue(undefined) - ); - Reflect.set(hydratedService, 'publishOverview', vi.fn().mockResolvedValue(undefined)); - Reflect.set(hydratedService, 'processNotificationAdd', processNotificationAdd); - Reflect.set( - hydratedService, - 'getNotificationsWatcher', - vi.fn().mockImplementation(async () => { - await handleNotificationAdd(bufferedPath); - return { - close: vi.fn().mockResolvedValue(undefined), - on: vi.fn().mockReturnThis(), - }; - }) - ); - Reflect.set( - hydratedService, - 'buildOverviewSnapshot', - vi.fn().mockResolvedValue({ - errorOccurred: false, - overview: { - unread: { alert: 0, info: 0, warning: 0, total: 0 }, - archive: { alert: 0, info: 0, warning: 0, total: 0 }, - }, - seenPaths: new Set(), - }) - ); - - await Reflect.get(hydratedService, 'initializeNotificationsState').call( - hydratedService, - testNotificationsDir, - true - ); - - expect(processNotificationAdd).toHaveBeenCalledWith(bufferedPath); + it('closes a watcher that fails before ready and retries on the next public read', async () => { + await Reflect.get(NotificationsService, 'watcher').close(); + Reflect.set(NotificationsService, 'watcher', null); + const failedWatcher = Object.assign(new EventEmitter(), { + close: vi.fn().mockResolvedValue(undefined), + }); + mockWatch.mockImplementationOnce(() => { + setImmediate(() => failedWatcher.emit('error', new Error('watch setup failed'))); + return failedWatcher; + }); + const recoveringService = createNotificationsService(); + await expect(Reflect.get(recoveringService, 'initialization')).rejects.toThrow(); + expect(failedWatcher.close).toHaveBeenCalledOnce(); + expect(Reflect.get(NotificationsService, 'watcher')).toBeNull(); + const attempts = mockWatch.mock.calls.length; + await expect( + Promise.all( + [0, 1].map(() => + recoveringService.getNotifications({ + type: NotificationType.UNREAD, + offset: 0, + limit: 10, + }) + ) + ) + ).resolves.toEqual([[], []]); + expect(mockWatch).toHaveBeenCalledTimes(attempts + 1); + expect(Reflect.get(NotificationsService, 'watcher')).not.toBe(failedWatcher); }); it('should load and validate a valid notification file', async () => { diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.broadcast.spec.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.broadcast.spec.ts new file mode 100644 index 0000000000..80a24987db --- /dev/null +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.broadcast.spec.ts @@ -0,0 +1,170 @@ +import { ConfigService } from '@nestjs/config'; +import { mkdir, mkdtemp, rm, writeFile } from 'fs/promises'; +import { tmpdir } from 'os'; +import { join } from 'path'; + +import type { FSWatcher } from 'chokidar'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { pubsub, PUBSUB_CHANNEL } from '@app/core/pubsub.js'; +import { + Notification, + NotificationImportance, +} from '@app/unraid-api/graph/resolvers/notifications/notifications.model.js'; +import { NotificationsService } from '@app/unraid-api/graph/resolvers/notifications/notifications.service.js'; + +describe.sequential('Warning and alert refresh scheduling', () => { + let basePath: string; + let service: NotificationsService; + let subscription: number; + let snapshots: Notification[][]; + const pending: Promise[] = []; + const gates: (() => void)[] = []; + + const write = (importance: string) => + writeFile( + join(basePath, 'unread', 'current.notify'), + `timestamp=1700000000\nevent=Test\nsubject=Test\ndescription=Test\nimportance=${importance}\n` + ); + function refresh(): Promise { + const result = Reflect.get(service, 'publishWarningsAndAlerts').call(service); + pending.push(result); + return result; + } + function gate() { + const result = Promise.withResolvers(); + gates.push(result.resolve); + return result; + } + + beforeEach(async () => { + basePath = await mkdtemp(join(tmpdir(), 'notification-broadcast-')); + await mkdir(join(basePath, 'unread')); + await mkdir(join(basePath, 'archive')); + await write('warning'); + service = new NotificationsService( + new ConfigService({ store: { dynamix: { notify: { path: basePath } } } }) + ); + await Reflect.get(service, 'initialization'); + const watcher: FSWatcher = Reflect.get(NotificationsService, 'watcher'); + await watcher.close(); + Reflect.set(NotificationsService, 'watcher', null); + snapshots = []; + subscription = await pubsub.subscribe( + PUBSUB_CHANNEL.NOTIFICATION_WARNINGS_AND_ALERTS, + (payload: { notificationsWarningsAndAlerts: Notification[] }) => + snapshots.push(payload.notificationsWarningsAndAlerts) + ); + }); + + afterEach(async () => { + gates.splice(0).forEach((release) => release()); + await Promise.allSettled(pending.splice(0)); + pubsub.unsubscribe(subscription); + vi.restoreAllMocks(); + await rm(basePath, { recursive: true, force: true }); + }); + + it('combines simultaneous refresh requests into one real payload read', async () => { + const scans = vi.spyOn(service, 'getWarningsAndAlerts'); + await Promise.all(Array.from({ length: 100 }, refresh)); + expect(scans).toHaveBeenCalledTimes(1); + expect(snapshots).toHaveLength(1); + expect(snapshots[0][0].importance).toBe(NotificationImportance.WARNING); + }); + + it('serializes a mid-read burst and publishes the latest disk state afterward', async () => { + const captured = gate(); + const release = gate(); + const read = service.getWarningsAndAlerts.bind(service); + let active = 0; + let peak = 0; + const scans = vi.spyOn(service, 'getWarningsAndAlerts').mockImplementation(async () => { + active++; + peak = Math.max(peak, active); + try { + const snapshot = await read(); + if (!snapshots.length) { + captured.resolve(); + await release.promise; + } + return snapshot; + } finally { + active--; + } + }); + refresh(); + await captured.promise; + await write('alert'); + Array.from({ length: 100 }, refresh); + release.resolve(); + await Promise.all(pending); + expect(peak).toBe(1); + expect(scans).toHaveBeenCalledTimes(2); + expect(snapshots.at(-1)?.[0].importance).toBe(NotificationImportance.ALERT); + }); + + it('keeps a request received during a failed scan and accepts later requests', async () => { + const captured = gate(); + const release = gate(); + const scans = vi.spyOn(service, 'getWarningsAndAlerts').mockImplementationOnce(async () => { + captured.resolve(); + await release.promise; + throw new Error('temporary notification read failure'); + }); + const logger = Reflect.get(service, 'logger'); + const errors = vi.spyOn(logger, 'error'); + refresh(); + await captured.promise; + await write('alert'); + refresh(); + release.resolve(); + await Promise.all(pending); + expect(scans).toHaveBeenCalledTimes(2); + expect(errors).toHaveBeenCalled(); + expect(snapshots.at(-1)?.[0].importance).toBe(NotificationImportance.ALERT); + await write('warning'); + await refresh(); + expect(scans).toHaveBeenCalledTimes(3); + expect(snapshots.at(-1)?.[0].importance).toBe(NotificationImportance.WARNING); + }); + + it('does not lose refresh requests received while publication is pending', async () => { + const captured = gate(); + const release = gate(); + const publish = pubsub.publish.bind(pubsub); + vi.spyOn(pubsub, 'publish').mockImplementationOnce(async (channel, payload) => { + captured.resolve(); + await release.promise; + return publish(channel, payload); + }); + const scans = vi.spyOn(service, 'getWarningsAndAlerts'); + refresh(); + await captured.promise; + await write('alert'); + Array.from({ length: 100 }, refresh); + release.resolve(); + await Promise.all(pending); + expect(scans).toHaveBeenCalledTimes(2); + expect(snapshots.at(-1)?.[0].importance).toBe(NotificationImportance.ALERT); + }); + + it.each(['read', 'publication'] as const)( + 'releases the scheduler after a %s failure with no pending request', + async (failure) => { + if (failure === 'read') { + vi.spyOn(service, 'getWarningsAndAlerts').mockRejectedValueOnce( + new Error('read failed') + ); + } else { + vi.spyOn(pubsub, 'publish').mockRejectedValueOnce(new Error('publication failed')); + } + await refresh(); + expect(snapshots).toEqual([]); + await write('alert'); + await refresh(); + expect(snapshots).toHaveLength(1); + expect(snapshots[0][0].importance).toBe(NotificationImportance.ALERT); + } + ); +}); diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.load.spec.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.load.spec.ts new file mode 100644 index 0000000000..89c589c349 --- /dev/null +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.load.spec.ts @@ -0,0 +1,224 @@ +import { ConfigService } from '@nestjs/config'; +import { copyFile, mkdir, mkdtemp, rm, unlink, writeFile } from 'fs/promises'; +import { tmpdir } from 'os'; +import { basename, join } from 'path'; + +import type { FSWatcher } from 'chokidar'; +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { pubsub, PUBSUB_CHANNEL } from '@app/core/pubsub.js'; +import { + Notification, + NotificationImportance, + NotificationType, +} from '@app/unraid-api/graph/resolvers/notifications/notifications.model.js'; +import { NotificationsService } from '@app/unraid-api/graph/resolvers/notifications/notifications.service.js'; + +const { UNREAD, ARCHIVE } = NotificationType; + +describe.sequential('Notification burst read costs', () => { + let basePath: string; + let service: NotificationsService; + let watcher: FSWatcher; + const subscriptions: number[] = []; + const handlers: Promise[] = []; + const publications: Promise[] = []; + + afterEach(async () => { + await watcher?.close(); + Reflect.set(NotificationsService, 'watcher', null); + await Promise.allSettled(handlers.splice(0)); + await Promise.allSettled(publications.splice(0)); + subscriptions.splice(0).forEach((id) => pubsub.unsubscribe(id)); + vi.restoreAllMocks(); + if (basePath) await rm(basePath, { recursive: true, force: true }); + }); + + for (const [historySize, burstSize] of [ + [0, 24], + [1000, 24], + [1000, 100], + ]) { + it.each(['create', 'change', 'delete', 'archive-only'] as const)( + `measures ${burstSize} %s events with ${historySize} unread and archived history files`, + async (operation) => { + basePath = await mkdtemp(join(tmpdir(), 'notification-load-')); + const path = (type: NotificationType, id: string) => + join(basePath, type.toLowerCase(), id); + const write = (type: NotificationType, id: string, importance = 'normal') => + writeFile( + path(type, id), + `timestamp=1700000000\nevent=${id}\nsubject=Load\ndescription=Load\nimportance=${importance}\n` + ); + await Promise.all([mkdir(path(UNREAD, '')), mkdir(path(ARCHIVE, ''))]); + for (const type of [UNREAD, ARCHIVE]) { + await Promise.all( + Array.from({ length: historySize }, (_, index) => + write(type, `${type.toLowerCase()}-history-${index}.notify`) + ) + ); + } + const ids = Array.from({ length: burstSize }, (_, index) => `burst-${index}.notify`); + if (operation === 'change' || operation === 'delete') { + await Promise.all( + ids.map(async (id) => { + await write(UNREAD, id); + await write(ARCHIVE, id, 'alert'); + }) + ); + } + + service = new NotificationsService( + new ConfigService({ store: { dynamix: { notify: { path: basePath } } } }) + ); + await Reflect.get(service, 'initialization'); + watcher = Reflect.get(NotificationsService, 'watcher'); + const reads: string[] = []; + const events: { kind: string; path: string }[] = []; + const failures: unknown[] = []; + const snapshots: Notification[][] = []; + const additions: Notification[] = []; + subscriptions.push( + await pubsub.subscribe( + PUBSUB_CHANNEL.NOTIFICATION_WARNINGS_AND_ALERTS, + (payload: { notificationsWarningsAndAlerts: Notification[] }) => + snapshots.push(payload.notificationsWarningsAndAlerts) + ), + await pubsub.subscribe( + PUBSUB_CHANNEL.NOTIFICATION_ADDED, + (payload: { notificationAdded: Notification }) => + additions.push(payload.notificationAdded) + ) + ); + const load = Reflect.get(service, 'loadNotificationFile').bind(service); + Reflect.set(service, 'loadNotificationFile', (file: string, type: NotificationType) => { + reads.push(file); + return load(file, type); + }); + const handle = Reflect.get(service, 'handleNotificationEvent').bind(service); + Reflect.set(service, 'handleNotificationEvent', (file: string, kind: string) => { + events.push({ kind, path: file }); + const pending = handle(file, kind).catch((error: unknown) => failures.push(error)); + handlers.push(pending); + return pending; + }); + const publish = Reflect.get(service, 'publishWarningsAndAlerts').bind(service); + Reflect.set(service, 'publishWarningsAndAlerts', () => { + const pending = publish(); + publications.push(pending); + return pending; + }); + let activeScans = 0; + let peakActiveScans = 0; + const readWarnings = service.getWarningsAndAlerts.bind(service); + const scans = vi.spyOn(service, 'getWarningsAndAlerts').mockImplementation(async () => { + activeScans++; + peakActiveScans = Math.max(peakActiveScans, activeScans); + try { + return await readWarnings(); + } finally { + activeScans--; + } + }); + const recount = vi.spyOn(service, 'recalculateOverview'); + const started = performance.now(); + await Promise.all( + ids.map(async (id) => { + switch (operation) { + case 'create': + await write(UNREAD, id, 'warning'); + await copyFile(path(UNREAD, id), path(ARCHIVE, id)); + break; + case 'change': + await write(UNREAD, id, 'warning'); + break; + case 'delete': + await unlink(path(UNREAD, id)); + break; + case 'archive-only': + await write(ARCHIVE, id, 'warning'); + break; + } + }) + ); + const expectedEvents = burstSize * (operation === 'create' ? 2 : 1); + await vi.waitFor(() => expect(events).toHaveLength(expectedEvents), { + timeout: 15000, + }); + await watcher.close(); + await Promise.all(handlers); + await Promise.all(publications); + const elapsedMs = performance.now() - started; + const measuredReads = [...reads]; + expect(failures).toEqual([]); + expect(recount).not.toHaveBeenCalled(); + expect( + measuredReads.filter((file) => basename(file).startsWith('archive-history-')) + ).toEqual([]); + const unreadEvents = events.filter((event) => event.path.includes('/unread/')); + expect(scans.mock.calls.length).toBeLessThanOrEqual(unreadEvents.length); + expect(peakActiveScans).toBeLessThanOrEqual(1); + if (historySize) expect(scans.mock.calls.length).toBeLessThanOrEqual(4); + const maximumUnread = historySize + (operation === 'archive-only' ? 0 : burstSize); + expect(measuredReads.length).toBeLessThanOrEqual( + 2 * expectedEvents + maximumUnread * scans.mock.calls.length + ); + const unreadTotal = + historySize + + (operation === 'delete' || operation === 'archive-only' ? 0 : burstSize); + const archiveTotal = + historySize + + (operation === 'delete' || operation === 'archive-only' ? burstSize : 0); + expect(service.getOverview()).toMatchObject({ + unread: { + total: unreadTotal, + warning: operation === 'create' || operation === 'change' ? burstSize : 0, + }, + archive: { total: archiveTotal, alert: operation === 'delete' ? burstSize : 0 }, + }); + if (unreadEvents.length) { + expect(snapshots.length).toBeGreaterThan(0); + const latest = snapshots.at(-1) ?? []; + expect(latest).toHaveLength(operation === 'delete' ? 0 : Math.min(burstSize, 50)); + for (const notification of latest) { + expect(ids).toContain(notification.id); + expect(notification.importance).toBe(NotificationImportance.WARNING); + } + } else { + expect(scans).not.toHaveBeenCalled(); + expect(snapshots).toEqual([]); + } + expect(additions).toHaveLength(operation === 'create' ? burstSize : 0); + expect(new Set(additions.map((notification) => notification.id)).size).toBe( + additions.length + ); + + // Ground-truth list reads are deliberately outside the measured event workload. + for (const type of [UNREAD, ARCHIVE]) { + const list = await service.getNotifications({ type, offset: 0, limit: 10000 }); + const counts = service.getOverview()[type.toLowerCase()]; + expect(counts.total).toBe(list.length); + for (const importance of Object.values(NotificationImportance)) { + expect(counts[importance.toLowerCase()]).toBe( + list.filter((notification) => notification.importance === importance).length + ); + } + } + console.info( + 'notification-load', + JSON.stringify({ + operation, + historySize, + burstSize, + events: expectedEvents, + fileReadAttempts: measuredReads.length, + warningScans: scans.mock.calls.length, + peakActiveScans, + elapsedMs: Math.round(elapsedMs), + }) + ); + }, + 60000 + ); + } +}); diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts new file mode 100644 index 0000000000..71afe29bd5 --- /dev/null +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.overview.spec.ts @@ -0,0 +1,653 @@ +import { ApolloDriver, ApolloDriverConfig } from '@nestjs/apollo'; +import { ConfigService } from '@nestjs/config'; +import { GraphQLModule } from '@nestjs/graphql'; +import { FastifyAdapter, NestFastifyApplication } from '@nestjs/platform-fastify'; +import { Test } from '@nestjs/testing'; +import { copyFile, mkdir, mkdtemp, readFile, rename, rm, unlink, writeFile } from 'fs/promises'; +import { tmpdir } from 'os'; +import { join } from 'path'; + +import type { FSWatcher } from 'chokidar'; +import { PrefixedID } from '@unraid/shared/prefixed-id-scalar.js'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { pubsub, PUBSUB_CHANNEL } from '@app/core/pubsub.js'; +import { + Notification, + NotificationImportance, + NotificationOverview, + NotificationType, +} from '@app/unraid-api/graph/resolvers/notifications/notifications.model.js'; +import { NotificationsResolver } from '@app/unraid-api/graph/resolvers/notifications/notifications.resolver.js'; +import { NotificationsService } from '@app/unraid-api/graph/resolvers/notifications/notifications.service.js'; + +const { UNREAD, ARCHIVE } = NotificationType; + +describe.sequential('Notification filesystem and subscription integration', () => { + let basePath: string; + let service: NotificationsService; + let config: ConfigService; + let watcher: FSWatcher; + let subscriptions: number[]; + let overviews: NotificationOverview[]; + let additions: Notification[]; + let warnings: Notification[][]; + let events: { event: string; path: string }[]; + const file = (type: NotificationType, id: string) => join(basePath, type.toLowerCase(), id); + + async function write(type: NotificationType, id: string, importance = 'normal') { + await writeFile( + file(type, id), + `timestamp=1700000000\nevent=Test\nsubject=Test\ndescription=Test\nimportance=${importance}\n` + ); + } + + async function start() { + subscriptions.push( + await pubsub.subscribe( + PUBSUB_CHANNEL.NOTIFICATION_OVERVIEW, + (payload: { notificationsOverview: NotificationOverview }) => + overviews.push(payload.notificationsOverview) + ) + ); + subscriptions.push( + await pubsub.subscribe( + PUBSUB_CHANNEL.NOTIFICATION_ADDED, + (payload: { notificationAdded: Notification }) => + additions.push(payload.notificationAdded) + ) + ); + subscriptions.push( + await pubsub.subscribe( + PUBSUB_CHANNEL.NOTIFICATION_WARNINGS_AND_ALERTS, + (payload: { notificationsWarningsAndAlerts: Notification[] }) => + warnings.push(payload.notificationsWarningsAndAlerts) + ) + ); + service = new NotificationsService(config); + await Reflect.get(service, 'initialization'); + watcher = Reflect.get(NotificationsService, 'watcher'); + watcher.on('all', (event, path) => events.push({ event, path })); + } + + async function expectCounts(unread: number, archive: number) { + await vi.waitFor( + async () => { + const overview = service.getOverview(); + expect(overview.unread.total).toBe(unread); + expect(overview.archive.total).toBe(archive); + for (const type of Object.values(NotificationType)) { + const notifications = await service.getNotifications({ + type, + offset: 0, + limit: 1000, + }); + const counts = overview[type.toLowerCase()]; + expect(counts.total).toBe(notifications.length); + for (const importance of Object.values(NotificationImportance)) { + expect(counts[importance.toLowerCase()]).toBe( + notifications.filter( + (notification) => notification.importance === importance + ).length + ); + } + } + }, + { timeout: 4000, interval: 20 } + ); + } + + async function expectEvent(event: string, type: NotificationType, id: string) { + await vi.waitFor(() => expect(events).toContainEqual({ event, path: file(type, id) }), { + timeout: 4000, + }); + events.splice( + events.findIndex((entry) => entry.event === event && entry.path === file(type, id)), + 1 + ); + await Reflect.get(service, 'overviewUpdate'); + } + + beforeEach(async () => { + basePath = await mkdtemp(join(tmpdir(), 'notification-integration-')); + await Promise.all( + Object.values(NotificationType).map((type) => mkdir(join(basePath, type.toLowerCase()))) + ); + config = new ConfigService({ store: { dynamix: { notify: { path: basePath } } } }); + subscriptions = []; + overviews = []; + additions = []; + warnings = []; + events = []; + }); + + afterEach(async () => { + const activeWatcher: FSWatcher | null = Reflect.get(NotificationsService, 'watcher'); + await activeWatcher?.close(); + Reflect.set(NotificationsService, 'watcher', null); + if (service) { + await Reflect.get(service, 'overviewUpdate'); + await Reflect.get(service, 'warningsUpdate'); + } + subscriptions.forEach((subscription) => pubsub.unsubscribe(subscription)); + vi.restoreAllMocks(); + vi.unstubAllGlobals(); + await rm(basePath, { recursive: true, force: true }); + }); + + it('hydrates all severities and excludes legacy archive copies', async () => { + for (const importance of ['normal', 'warning', 'alert']) { + await write(UNREAD, `${importance}.notify`, importance); + await write(ARCHIVE, `${importance}.notify`, importance); + await write(ARCHIVE, `archived-${importance}.notify`, importance); + } + await start(); + await expectCounts(3, 3); + expect(service.getOverview().archive).toEqual({ info: 1, warning: 1, alert: 1, total: 3 }); + expect(additions).toEqual([]); + await service.recalculateOverview(); + await expectCounts(3, 3); + }); + + it('keeps handling real file changes after a runtime watcher error', async () => { + await start(); + expect(() => watcher.emit('error', new Error('watch resource exhausted'))).not.toThrow(); + await write(UNREAD, 'after-error.notify', 'warning'); + await expectEvent('add', UNREAD, 'after-error.notify'); + await expectCounts(1, 0); + await service.archiveNotification({ id: 'after-error.notify' }); + await expectCounts(0, 1); + }); + + it('recovers a failed directory initialization on the next public request', async () => { + const blocked = join(basePath, 'blocked'); + await writeFile(blocked, 'not a directory'); + config.set('store.dynamix.notify.path', blocked); + service = new NotificationsService(config); + await expect(Reflect.get(service, 'initialization')).rejects.toThrow(); + await unlink(blocked); + await expect(service.getNotifications({ type: UNREAD, offset: 0, limit: 10 })).resolves.toEqual( + [] + ); + await writeFile( + join(blocked, 'unread', 'recovered.notify'), + 'timestamp=1700000000\nevent=Test\nsubject=Test\ndescription=Test\nimportance=alert\n' + ); + await vi.waitFor(() => expect(service.getOverview().unread.alert).toBe(1)); + }); + + it.each([UNREAD, ARCHIVE])( + 'handles legacy files written %s first without recounting', + async (first) => { + await start(); + const recount = vi.spyOn(service, 'recalculateOverview'); + const second = first === UNREAD ? ARCHIVE : UNREAD; + const initialOverview = structuredClone(overviews[0]); + await write(first, 'legacy.notify', 'warning'); + await expectEvent('add', first, 'legacy.notify'); + await expectCounts(first === UNREAD ? 1 : 0, first === ARCHIVE ? 1 : 0); + await copyFile(file(first, 'legacy.notify'), file(second, 'legacy.notify')); + await expectEvent('add', second, 'legacy.notify'); + await expectCounts(1, 0); + await vi.waitFor(() => expect(additions).toHaveLength(1)); + await vi.waitFor(() => expect(warnings.at(-1)).toHaveLength(1)); + expect(recount).not.toHaveBeenCalled(); + expect(overviews.at(-1)).toEqual(service.getOverview()); + expect(overviews[0]).toEqual(initialOverview); + } + ); + + it('creates a notification through the public API and publishes its real file data', async () => { + await start(); + const notification = await service.createNotification({ + title: 'Integration notification', + subject: 'Disk status', + description: 'Test warning', + importance: NotificationImportance.WARNING, + }); + expect(await readFile(file(UNREAD, notification.id), 'utf8')).toContain('importance=warning'); + await expectEvent('add', UNREAD, notification.id); + await expectCounts(1, 0); + await vi.waitFor(() => expect(additions).toHaveLength(1)); + expect(additions[0]).toMatchObject({ + id: notification.id, + subject: 'Disk status', + importance: NotificationImportance.WARNING, + }); + }); + + it('archives over a legacy copy, restores, and deletes without late watcher drift', async () => { + await write(UNREAD, 'legacy.notify', 'alert'); + await write(ARCHIVE, 'legacy.notify', 'normal'); + await start(); + const recount = vi.spyOn(service, 'recalculateOverview'); + const archived = await service.archiveNotification({ id: 'legacy.notify' }); + expect(archived.type).toBe(ARCHIVE); + expect(service.getOverview().archive.alert).toBe(1); + await expectEvent('unlink', UNREAD, 'legacy.notify'); + await expectCounts(0, 1); + await vi.waitFor(() => expect(warnings.at(-1)).toEqual([])); + const restored = await service.markAsUnread({ id: 'legacy.notify' }); + expect(restored.type).toBe(UNREAD); + expect(service.getOverview().unread.alert).toBe(1); + await expectEvent('unlink', ARCHIVE, 'legacy.notify'); + await expectEvent('add', UNREAD, 'legacy.notify'); + await expectCounts(1, 0); + await vi.waitFor(() => expect(additions).toHaveLength(1)); + expect(additions[0]).toMatchObject({ + id: 'legacy.notify', + type: UNREAD, + importance: NotificationImportance.ALERT, + }); + await service.deleteNotification({ id: 'legacy.notify', type: UNREAD }); + await expectEvent('unlink', UNREAD, 'legacy.notify'); + await expectCounts(0, 0); + await vi.waitFor(() => expect(warnings.at(-1)).toEqual([])); + expect(recount).not.toHaveBeenCalled(); + }); + + it('updates warning subscription content without changing counts or adding again', async () => { + await start(); + await write(UNREAD, 'content.notify', 'warning'); + await expectEvent('add', UNREAD, 'content.notify'); + await vi.waitFor(() => expect(additions).toHaveLength(1)); + await writeFile( + file(UNREAD, 'content.notify'), + 'timestamp=1700000000\nevent=Test\nsubject=Updated subject\ndescription=Updated description\nimportance=warning\n' + ); + await expectEvent('change', UNREAD, 'content.notify'); + await vi.waitFor(() => + expect(warnings.at(-1)?.[0]).toMatchObject({ + subject: 'Updated subject', + description: 'Updated description', + }) + ); + await expectCounts(1, 0); + expect(additions).toHaveLength(1); + }); + + it('tracks external overwrite renames and severity changes', async () => { + await write(UNREAD, 'external.notify', 'warning'); + await write(ARCHIVE, 'external.notify', 'normal'); + await start(); + await rename(file(UNREAD, 'external.notify'), file(ARCHIVE, 'external.notify')); + await expectEvent('unlink', UNREAD, 'external.notify'); + await expectCounts(0, 1); + expect(service.getOverview().archive.warning).toBe(1); + await write(ARCHIVE, 'external.notify', 'alert'); + await expectEvent('change', ARCHIVE, 'external.notify'); + await vi.waitFor(() => expect(service.getOverview().archive.alert).toBe(1)); + await expectCounts(0, 1); + await rename(file(ARCHIVE, 'external.notify'), file(UNREAD, 'external.notify')); + await expectEvent('add', UNREAD, 'external.notify'); + await expectCounts(1, 0); + }); + + it.each([UNREAD, ARCHIVE])('deletes a %s copy according to actual visibility', async (type) => { + await write(UNREAD, 'legacy.notify', 'warning'); + await write(ARCHIVE, 'legacy.notify', 'alert'); + await start(); + await service.deleteNotification({ id: 'legacy.notify', type }); + await expectEvent('unlink', type, 'legacy.notify'); + await expectCounts(type === UNREAD ? 0 : 1, type === UNREAD ? 1 : 0); + if (type === UNREAD) expect(service.getOverview().archive.alert).toBe(1); + }); + + it('handles external deletion of either legacy copy', async () => { + await write(UNREAD, 'legacy.notify', 'warning'); + await write(ARCHIVE, 'legacy.notify', 'alert'); + await start(); + await unlink(file(UNREAD, 'legacy.notify')); + await expectEvent('unlink', UNREAD, 'legacy.notify'); + await expectCounts(0, 1); + expect(service.getOverview().archive.alert).toBe(1); + await unlink(file(ARCHIVE, 'legacy.notify')); + await expectEvent('unlink', ARCHIVE, 'legacy.notify'); + await expectCounts(0, 0); + }); + + it('archives/restores by importance, archives all, then clears both categories', async () => { + for (const importance of ['normal', 'warning', 'alert']) { + await write(UNREAD, `${importance}.notify`, importance); + await write(ARCHIVE, `${importance}.notify`, importance); + } + await start(); + const recount = vi.spyOn(service, 'recalculateOverview'); + await service.archiveAll(NotificationImportance.WARNING); + await expectCounts(2, 1); + expect(service.getOverview().archive.warning).toBe(1); + await service.unarchiveAll(NotificationImportance.WARNING); + await expectCounts(3, 0); + await service.archiveAll(); + await expectCounts(0, 3); + await service.unarchiveAll(); + await expectCounts(3, 0); + await service.deleteAllNotifications(); + await expectCounts(0, 0); + expect(recount).not.toHaveBeenCalled(); + }); + + it('reveals hidden archive files after a bulk unread delete', async () => { + await write(UNREAD, 'legacy.notify', 'warning'); + await write(ARCHIVE, 'legacy.notify', 'alert'); + await start(); + await service.deleteNotifications(UNREAD); + await expectEvent('unlink', UNREAD, 'legacy.notify'); + await expectCounts(0, 1); + expect(service.getOverview().archive.alert).toBe(1); + }); + + it('reconciles cached IDs even when their unlink event has not arrived', async () => { + await write(UNREAD, 'removed.notify'); + await start(); + await watcher.close(); + await unlink(file(UNREAD, 'removed.notify')); + await service.deleteNotifications(UNREAD); + await expectCounts(0, 0); + }); + + it('handles a burst of legacy writes without full recounts or duplicate additions', async () => { + await start(); + const recount = vi.spyOn(service, 'recalculateOverview'); + const ids = Array.from({ length: 24 }, (_, index) => `burst-${index}.notify`); + await Promise.all( + ids.map(async (id) => { + await write(UNREAD, id, 'warning'); + await copyFile(file(UNREAD, id), file(ARCHIVE, id)); + }) + ); + await expectCounts(24, 0); + await vi.waitFor(() => expect(additions).toHaveLength(24)); + await vi.waitFor(() => expect(events.filter((entry) => entry.event === 'add')).toHaveLength(48)); + await Reflect.get(service, 'overviewUpdate'); + await expectCounts(24, 0); + expect(new Set(additions.map((notification) => notification.id)).size).toBe(24); + await vi.waitFor(() => expect(warnings.at(-1)).toHaveLength(1)); + expect(recount).not.toHaveBeenCalled(); + }); + + it('bounds event reads to affected files even with a large archived history', async () => { + await Promise.all( + Array.from({ length: 128 }, (_, index) => write(ARCHIVE, `history-${index}.notify`)) + ); + await start(); + const recount = vi.spyOn(service, 'recalculateOverview'); + const reads: string[] = []; + const load = Reflect.get(service, 'loadNotificationFile').bind(service); + Reflect.set(service, 'loadNotificationFile', (path: string, type: NotificationType) => { + reads.push(path); + return load(path, type); + }); + await write(UNREAD, 'new.notify'); + await copyFile(file(UNREAD, 'new.notify'), file(ARCHIVE, 'new.notify')); + await expectEvent('add', UNREAD, 'new.notify'); + await expectEvent('add', ARCHIVE, 'new.notify'); + await vi.waitFor(() => + expect(service.getOverview()).toMatchObject({ + unread: { total: 1 }, + archive: { total: 128 }, + }) + ); + expect(recount).not.toHaveBeenCalled(); + expect(reads.some((path) => path.includes('/history-'))).toBe(false); + await expectCounts(1, 128); + }); + + it('retains the previous counts when an explicit reconciliation cannot read a category', async () => { + await write(UNREAD, 'visible.notify'); + await start(); + await watcher.close(); + const unread = file(UNREAD, ''); + const saved = join(basePath, 'saved-unread'); + await rename(unread, saved); + try { + const before = service.getOverview(); + const result = await service.recalculateOverview(); + expect(result.error).toBe(true); + expect(service.getOverview()).toEqual(before); + } finally { + await rename(saved, unread); + } + await service.recalculateOverview(); + await expectCounts(1, 0); + }); + + it('keeps counts intact after a failed rename and allows a successful retry', async () => { + await write(UNREAD, 'blocked.notify', 'warning'); + await start(); + await mkdir(file(ARCHIVE, 'blocked.notify')); + await expect(service.archiveNotification({ id: 'blocked.notify' })).rejects.toThrow(); + await expectCounts(1, 0); + await rm(file(ARCHIVE, 'blocked.notify'), { recursive: true }); + await service.archiveNotification({ id: 'blocked.notify' }); + await expectEvent('unlink', UNREAD, 'blocked.notify'); + await expectCounts(0, 1); + }); + + it('replays real watcher changes after an in-flight full reconciliation', async () => { + await write(UNREAD, 'removed.notify'); + await start(); + const original = Reflect.get(service, 'buildOverviewSnapshot').bind(service); + const captured = Promise.withResolvers(); + const release = Promise.withResolvers(); + Reflect.set(service, 'buildOverviewSnapshot', async () => { + const result = await original(); + captured.resolve(); + await release.promise; + return result; + }); + const reconciliation = service.recalculateOverview(); + try { + await captured.promise; + await write(UNREAD, 'added.notify', 'alert'); + await unlink(file(UNREAD, 'removed.notify')); + await vi.waitFor( + () => { + expect(events).toContainEqual({ event: 'add', path: file(UNREAD, 'added.notify') }); + expect(events).toContainEqual({ + event: 'unlink', + path: file(UNREAD, 'removed.notify'), + }); + }, + { timeout: 4000 } + ); + } finally { + release.resolve(); + await reconciliation; + } + await Reflect.get(service, 'overviewUpdate'); + await expectCounts(1, 0); + expect(service.getOverview().unread.alert).toBe(1); + expect(overviews.at(-1)).toEqual(service.getOverview()); + }); + + it('replays actual adds and removals that arrive during the startup snapshot', async () => { + await write(UNREAD, 'old.notify'); + await write(ARCHIVE, 'old.notify', 'warning'); + const prototype = NotificationsService.prototype; + const snapshot = Reflect.get(prototype, 'buildOverviewSnapshot'); + const release = Promise.withResolvers(); + let captured = false; + Reflect.set(prototype, 'buildOverviewSnapshot', async function (this: NotificationsService) { + const result = await snapshot.call(this); + captured = true; + await release.promise; + return result; + }); + const starting = start(); + try { + await vi.waitFor(() => expect(captured).toBe(true)); + const activeWatcher: FSWatcher = Reflect.get(NotificationsService, 'watcher'); + const duringStartup: string[] = []; + activeWatcher.on('all', (event, path) => duringStartup.push(`${event}:${path}`)); + await write(UNREAD, 'during-startup.notify', 'alert'); + await unlink(file(UNREAD, 'old.notify')); + await vi.waitFor(() => { + expect(duringStartup).toContain(`add:${file(UNREAD, 'during-startup.notify')}`); + expect(duringStartup).toContain(`unlink:${file(UNREAD, 'old.notify')}`); + }); + await writeFile( + file(UNREAD, 'during-startup.notify'), + 'timestamp=1700000000\nevent=Test\nsubject=Startup update\ndescription=Test\nimportance=alert\n' + ); + await vi.waitFor(() => + expect(duringStartup).toContain(`change:${file(UNREAD, 'during-startup.notify')}`) + ); + } finally { + release.resolve(); + await starting; + Reflect.set(prototype, 'buildOverviewSnapshot', snapshot); + } + await expectCounts(1, 1); + expect(service.getOverview().unread.alert).toBe(1); + expect(service.getOverview().archive.warning).toBe(1); + await vi.waitFor(() => expect(additions).toHaveLength(1)); + expect(additions[0]).toMatchObject({ id: 'during-startup.notify', subject: 'Startup update' }); + }); + + it('rebuilds the index and watcher when the configured directory changes', async () => { + await write(UNREAD, 'old.notify'); + await start(); + const previousWatcher = watcher; + const next = join(basePath, 'alternate'); + await Promise.all( + Object.values(NotificationType).map((type) => + mkdir(join(next, type.toLowerCase()), { recursive: true }) + ) + ); + const nextFile = join(next, 'archive', 'new.notify'); + await writeFile( + nextFile, + 'timestamp=1700000000\nevent=Test\nsubject=Test\ndescription=Test\nimportance=alert\n' + ); + config.set('store.dynamix.notify.path', next); + await service.getNotifications({ type: ARCHIVE, offset: 0, limit: 100 }); + expect(Object.keys(previousWatcher.getWatched())).toHaveLength(0); + await expectCounts(0, 1); + expect(service.getOverview().archive.alert).toBe(1); + watcher = Reflect.get(NotificationsService, 'watcher'); + const removed = new Promise((resolve) => watcher.once('unlink', () => resolve())); + await unlink(nextFile); + await removed; + await expectCounts(0, 0); + }); + + it('recovers a failed startup replay on the next public request', async () => { + const prototype = NotificationsService.prototype; + const snapshot = Reflect.get(prototype, 'buildOverviewSnapshot'); + const handler = Reflect.get(prototype, 'handleNotificationEvent'); + const captured = Promise.withResolvers(); + const release = Promise.withResolvers(); + let failReplay = true; + Reflect.set(prototype, 'buildOverviewSnapshot', async function (this: NotificationsService) { + const result = await snapshot.call(this); + captured.resolve(); + await release.promise; + return result; + }); + Reflect.set( + prototype, + 'handleNotificationEvent', + async function (this: NotificationsService, path: string, event: string) { + if ( + failReplay && + path.endsWith('/replay.notify') && + !Reflect.get(this, 'isHydratingOverview') + ) { + failReplay = false; + throw new Error('replay read failed'); + } + return handler.call(this, path, event); + } + ); + const rejected = expect(start()).rejects.toThrow(); + try { + await captured.promise; + const activeWatcher: FSWatcher = Reflect.get(NotificationsService, 'watcher'); + const added = new Promise((resolve) => activeWatcher.once('add', () => resolve())); + await write(UNREAD, 'replay.notify', 'alert'); + await added; + release.resolve(); + await rejected; + expect(failReplay).toBe(false); + await expect( + service.getNotifications({ type: UNREAD, offset: 0, limit: 10 }) + ).resolves.toHaveLength(1); + await expectCounts(1, 0); + } finally { + release.resolve(); + Reflect.set(prototype, 'buildOverviewSnapshot', snapshot); + Reflect.set(prototype, 'handleNotificationEvent', handler); + } + }); + + it('returns matching lists and counts through actual GraphQL queries and mutations', async () => { + await write(UNREAD, 'graphql.notify', 'warning'); + await write(ARCHIVE, 'graphql.notify', 'warning'); + await start(); + vi.stubGlobal('getServerIdentifier', () => 'integration'); + const module = await Test.createTestingModule({ + imports: [ + GraphQLModule.forRoot({ + driver: ApolloDriver, + autoSchemaFile: true, + }), + ], + providers: [ + PrefixedID, + NotificationsResolver, + { provide: NotificationsService, useValue: service }, + ], + }).compile(); + const app = module.createNestApplication(new FastifyAdapter()); + try { + await app.init(); + await app.getHttpAdapter().getInstance().ready(); + const execute = async (query: string) => { + const response = await app.inject({ + method: 'POST', + url: '/graphql', + payload: { query }, + }); + expect(response.statusCode).toBe(200); + const result = response.json<{ data: unknown; errors?: unknown[] }>(); + expect(result.errors).toBeUndefined(); + return result.data; + }; + expect( + await execute( + '{ notifications { overview { unread { total } archive { total } } list(filter: {type: ARCHIVE, offset: 0, limit: 100}) { id } } }' + ) + ).toEqual({ + notifications: { overview: { unread: { total: 1 }, archive: { total: 0 } }, list: [] }, + }); + expect( + await execute( + 'mutation { archiveNotification(id: "integration:graphql.notify") { id type } }' + ) + ).toEqual({ archiveNotification: { id: 'integration:graphql.notify', type: ARCHIVE } }); + await expectEvent('unlink', UNREAD, 'graphql.notify'); + expect( + await execute( + '{ notifications { overview { unread { total } archive { total } } list(filter: {type: ARCHIVE, offset: 0, limit: 100}) { id importance } } }' + ) + ).toEqual({ + notifications: { + overview: { unread: { total: 0 }, archive: { total: 1 } }, + list: [ + { id: 'integration:graphql.notify', importance: NotificationImportance.WARNING }, + ], + }, + }); + expect( + await execute( + 'mutation { unreadNotification(id: "integration:graphql.notify") { type } }' + ) + ).toEqual({ unreadNotification: { type: UNREAD } }); + await expectEvent('add', UNREAD, 'graphql.notify'); + await expectCounts(1, 0); + } finally { + await app.close(); + } + }); +}); diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.spec.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.spec.ts index 14448be631..cc71feb67e 100644 --- a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.spec.ts +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.spec.ts @@ -58,7 +58,6 @@ async function disableNotificationsWatcher(service: NotificationsService) { } | null; await watcher?.close?.(); Reflect.set(NotificationsService, 'watcher', null); - Reflect.set(NotificationsService, 'overview', zeroOverview()); } } @@ -100,8 +99,9 @@ describe.sequential('NotificationsService', () => { // make sure each test is isolated (as much as possible) afterEach(async () => { - Reflect.set(NotificationsService, 'overview', zeroOverview()); await service.deleteAllNotifications(); + await Reflect.get(service, 'warningsUpdate'); + expect(service.getOverview()).toEqual(zeroOverview()); }); /**------------------------------------------------------------------------ diff --git a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts index 04834ecef8..35a3149d09 100644 --- a/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts +++ b/api/src/unraid-api/graph/resolvers/notifications/notifications.service.ts @@ -1,7 +1,7 @@ import { Injectable, Logger } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import { mkdir, readdir, readFile, rename, stat, unlink, writeFile } from 'fs/promises'; -import { basename, join } from 'path'; +import { basename, dirname, join } from 'path'; import type { Stats } from 'fs'; import { FSWatcher, watch } from 'chokidar'; @@ -33,6 +33,9 @@ import { validateObject } from '@app/unraid-api/graph/resolvers/validation.utils import { SortFn } from '@app/unraid-api/types/util.js'; import { batchProcess, formatDatetime, isFulfilled, isRejected, unraidTimestamp } from '@app/utils.js'; +type NotificationFileState = Partial>; +type NotificationFileEvent = 'add' | 'change' | 'unlink'; + @Injectable() export class NotificationsService { private static readonly FILE_IO_BATCH_SIZE = 32; @@ -43,8 +46,13 @@ export class NotificationsService { */ private path: string | null = null; private initialization: Promise | null = null; + private initializationFailed = false; + private overviewUpdate: Promise = Promise.resolve(); + private warningsUpdate: Promise | null = null; + private warningsUpdateRequested = false; + private notificationFiles = new Map(); private isHydratingOverview = false; - private pendingNotificationAdds = new Set(); + private pendingNotificationEvents = new Map(); private static overview: NotificationOverview = { unread: { @@ -94,29 +102,34 @@ export class NotificationsService { }; } + private async waitForInitialization() { + this.paths(); + if (this.initializationFailed) { + this.initialization = this.initializeNotificationsState(this.getConfiguredPath(), true); + } + await this.initialization; + } + private notificationPath(id: string, type: NotificationType): string { return resolveFileInDirectory(this.paths()[type], id); } private initializeNotificationsState(basePath: string, recreate = false) { + this.initializationFailed = false; const initialize = async () => { await this.ensureNotificationDirectories(basePath); - this.pendingNotificationAdds.clear(); + this.pendingNotificationEvents.clear(); this.isHydratingOverview = true; try { await this.getNotificationsWatcher(basePath, recreate); - const { errorOccurred, overview, seenPaths } = await this.buildOverviewSnapshot(); - NotificationsService.overview = overview; - await this.publishOverview(); - - const pendingAdds = [...this.pendingNotificationAdds].filter( - (path) => !seenPaths.has(path) - ); - this.pendingNotificationAdds.clear(); + const { error: errorOccurred } = await this.reconcileOverview(); + const pendingEvents = [...this.pendingNotificationEvents]; + this.pendingNotificationEvents.clear(); + this.isHydratingOverview = false; - for (const path of pendingAdds) { - await this.processNotificationAdd(path); + for (const [path, event] of pendingEvents) { + await this.handleNotificationEvent(path, event); } if (errorOccurred) { @@ -126,12 +139,17 @@ export class NotificationsService { } } finally { this.isHydratingOverview = false; - this.pendingNotificationAdds.clear(); + this.pendingNotificationEvents.clear(); } }; const previousInitialization = this.initialization ?? Promise.resolve(); - return previousInitialization.catch(() => undefined).then(initialize); + const initialization = previousInitialization.catch(() => undefined).then(initialize); + void initialization.catch((error) => { + if (this.initialization === initialization) this.initializationFailed = true; + this.logger.error(error); + }); + return initialization; } private async ensureNotificationDirectories(basePath: string) { @@ -159,56 +177,114 @@ export class NotificationsService { } await NotificationsService.watcher?.close().catch((e) => this.logger.error(e)); - NotificationsService.watcher = watch(basePath, { + const watcher = watch(basePath, { usePolling: CHOKIDAR_USEPOLLING, ignoreInitial: true, - }).on('add', (path) => { - void this.handleNotificationAdd(path).catch((e) => this.logger.error(e)); }); - - return NotificationsService.watcher; + NotificationsService.watcher = watcher; + watcher.on('error', (error) => this.logger.error(error)); + for (const event of ['add', 'change', 'unlink'] as const) { + watcher.on(event, (path) => { + void this.handleNotificationEvent(path, event).catch((e) => this.logger.error(e)); + }); + } + try { + await new Promise((resolve, reject) => { + watcher.once('error', reject); + watcher.once('ready', () => { + watcher.off('error', reject); + resolve(); + }); + }); + } catch (error) { + if (NotificationsService.watcher === watcher) NotificationsService.watcher = null; + await watcher.close().catch((closeError) => this.logger.error(closeError)); + throw error; + } + return watcher; } private isMissingFileError(error: unknown): error is NodeJS.ErrnoException { return error instanceof Error && 'code' in error && error.code === 'ENOENT'; } - private async handleNotificationAdd(path: string) { + private async handleNotificationEvent(path: string, event: NotificationFileEvent) { + const paths = this.paths(); + const type = Object.values(NotificationType).find((type) => dirname(path) === paths[type]); + if (!type) return; if (this.isHydratingOverview) { - this.pendingNotificationAdds.add(path); + const pending = this.pendingNotificationEvents.get(path); + this.pendingNotificationEvents.set( + path, + event === 'change' && pending === 'add' ? 'add' : event + ); return; } - await this.processNotificationAdd(path); + const { unread } = await this.synchronizeNotification(basename(path)); + if (type === NotificationType.UNREAD) { + if (event === 'add' && unread) { + await pubsub.publish(PUBSUB_CHANNEL.NOTIFICATION_ADDED, { notificationAdded: unread }); + } + void this.publishWarningsAndAlerts(); + } } - private async processNotificationAdd(path: string) { - // The path looks like /{notification base path}/{type}/{notification id} - const type = path.includes('/unread/') ? NotificationType.UNREAD : NotificationType.ARCHIVE; - // this.logger.debug(`Adding ${type} Notification: ${path}`); + private enqueueOverviewUpdate(action: () => Promise): Promise { + const update = this.overviewUpdate.catch(() => undefined).then(action); + this.overviewUpdate = update; + return update; + } - let notification: Notification; + private async readNotificationCopy(id: string, type: NotificationType) { try { - notification = await this.loadNotificationFile(path, NotificationType[type]); + return await this.loadNotificationFile(this.notificationPath(id, type), type); } catch (error) { - if (this.isMissingFileError(error)) { - this.logger.debug( - `[handleNotificationAdd] Notification disappeared before load: ${path}` - ); - return; - } + if (this.isMissingFileError(error)) return null; throw error; } + } - this.increment(notification.importance, NotificationsService.overview[type.toLowerCase()]); - - if (type === NotificationType.UNREAD) { - this.publishOverview(); - pubsub.publish(PUBSUB_CHANNEL.NOTIFICATION_ADDED, { - notificationAdded: notification, + private synchronizeNotification(id: string) { + return this.enqueueOverviewUpdate(async () => { + const [unread, archive] = await Promise.all([ + this.readNotificationCopy(id, NotificationType.UNREAD), + this.readNotificationCopy(id, NotificationType.ARCHIVE), + ]); + const changed = this.updateNotificationCounts(id, { + UNREAD: unread?.importance, + ARCHIVE: archive?.importance, }); - void this.publishWarningsAndAlerts(); + if (changed) await this.publishOverview(); + return { unread }; + }); + } + + private updateNotificationCounts(id: string, next: NotificationFileState): boolean { + const previous = this.notificationFiles.get(id) ?? {}; + let changed = false; + for (const type of Object.values(NotificationType)) { + const previousImportance = this.visibleImportance(previous, type); + const nextImportance = this.visibleImportance(next, type); + if (previousImportance === nextImportance) continue; + const counts = NotificationsService.overview[type.toLowerCase()]; + if (previousImportance) { + counts[previousImportance.toLowerCase()]--; + counts.total--; + } + if (nextImportance) this.increment(nextImportance, counts); + changed = true; } + if (next.UNREAD === undefined && next.ARCHIVE === undefined) { + this.notificationFiles.delete(id); + } else { + this.notificationFiles.set(id, next); + } + return changed; + } + + private visibleImportance(files: NotificationFileState, type: NotificationType) { + return type === NotificationType.UNREAD || files.UNREAD === undefined ? files[type] : undefined; } /** @@ -223,24 +299,38 @@ export class NotificationsService { return structuredClone(NotificationsService.overview); } - private publishOverview(overview = NotificationsService.overview) { + private publishOverview(overview = this.getOverview()) { return pubsub.publish(PUBSUB_CHANNEL.NOTIFICATION_OVERVIEW, { notificationsOverview: overview, }); } - private async publishWarningsAndAlerts() { - try { - const warningsAndAlerts = await this.getWarningsAndAlerts(); - await pubsub.publish(PUBSUB_CHANNEL.NOTIFICATION_WARNINGS_AND_ALERTS, { - notificationsWarningsAndAlerts: warningsAndAlerts, - }); - } catch (error) { - this.logger.error( - '[publishWarningsAndAlerts] Failed to broadcast warnings and alerts snapshot', - error as Error - ); - } + private publishWarningsAndAlerts(): Promise { + this.warningsUpdateRequested = true; + if (this.warningsUpdate) return this.warningsUpdate; + + this.warningsUpdate = Promise.resolve().then(async () => { + try { + while (this.warningsUpdateRequested) { + await this.overviewUpdate.catch(() => undefined); + this.warningsUpdateRequested = false; + try { + const warningsAndAlerts = await this.getWarningsAndAlerts(); + await pubsub.publish(PUBSUB_CHANNEL.NOTIFICATION_WARNINGS_AND_ALERTS, { + notificationsWarningsAndAlerts: warningsAndAlerts, + }); + } catch (error) { + this.logger.error( + '[publishWarningsAndAlerts] Failed to broadcast warnings and alerts snapshot', + error as Error + ); + } + } + } finally { + this.warningsUpdate = null; + } + }); + return this.warningsUpdate; } private increment(importance: NotificationImportance, collector: NotificationCounts) { @@ -248,16 +338,21 @@ export class NotificationsService { collector['total'] += 1; } - private decrement(importance: NotificationImportance, collector: NotificationCounts) { - collector[importance.toLowerCase()] -= 1; - collector['total'] -= 1; + public async recalculateOverview() { + await this.waitForInitialization(); + return this.reconcileOverview(); } - public async recalculateOverview() { - const { errorOccurred, overview } = await this.buildOverviewSnapshot(); - NotificationsService.overview = overview; - await this.publishOverview(); - return { error: errorOccurred, overview: this.getOverview() }; + private reconcileOverview() { + return this.enqueueOverviewUpdate(async () => { + const snapshot = await this.buildOverviewSnapshot(); + if (!snapshot.errorOccurred) { + this.notificationFiles = snapshot.files; + NotificationsService.overview = snapshot.overview; + await this.publishOverview(); + } + return { error: snapshot.errorOccurred, overview: this.getOverview() }; + }); } private async buildOverviewSnapshot() { @@ -275,29 +370,30 @@ export class NotificationsService { total: 0, }, }; - const seenPaths = new Set(); - - // todo - refactor this to be more memory efficient - // i.e. by using a lazy generator vs the current eager implementation - // - // recalculates stats for a particular notification type + const files = new Map(); const recalculate = async (type: NotificationType) => { - const ids = await this.listFilesInFolder(this.paths()[type]); - ids.forEach((id) => seenPaths.add(id)); - const [notifications] = await this.loadNotificationsFromPaths(ids, {}); - notifications.forEach((n) => this.increment(n.importance, overview[type.toLowerCase()])); + // Keep hidden archive copies in the index so removing unread can reveal them. + const paths = await this.listFilesInFolder(this.paths()[type]); + const [notifications, errors] = await this.loadNotificationsFromPaths(paths, { type }); + const failure = errors.find((error) => !this.isMissingFileError(error)); + if (failure) throw failure; + for (const notification of notifications) { + const copies = files.get(notification.id) ?? {}; + copies[type] = notification.importance; + files.set(notification.id, copies); + } }; - - const results = await batchProcess( - [NotificationType.ARCHIVE, NotificationType.UNREAD], - recalculate - ); - + const results = await batchProcess(Object.values(NotificationType), recalculate); if (results.errorOccurred) { - results.errors.forEach((e) => this.logger.error('[recalculateOverview] ' + e)); + results.errors.forEach((error) => this.logger.error('[recalculateOverview] ' + error)); } - - return { errorOccurred: results.errorOccurred, overview, seenPaths }; + for (const copies of files.values()) { + for (const type of Object.values(NotificationType)) { + const importance = this.visibleImportance(copies, type); + if (importance) this.increment(importance, overview[type.toLowerCase()]); + } + } + return { errorOccurred: results.errorOccurred, overview, files }; } /**------------------------------------------------------------------------ @@ -305,6 +401,7 @@ export class NotificationsService { *------------------------------------------------------------------------**/ public async createNotification(data: NotificationData): Promise { + await this.waitForInitialization(); const id: string = await this.makeNotificationId(data.title); const fileData = this.makeNotificationFileData(data); @@ -400,6 +497,7 @@ export class NotificationsService { *------------------------------------------------------------------------**/ public async deleteNotification({ id, type }: Pick) { + await this.waitForInitialization(); const path = this.notificationPath(id, type); // we don't want to update the overview stats if the deletion (unlink) fails @@ -408,8 +506,7 @@ export class NotificationsService { const notification = await this.loadNotificationFile(path, type); await unlink(path); - this.decrement(notification.importance, NotificationsService.overview[type.toLowerCase()]); - await this.publishOverview(); + await this.synchronizeNotification(id); if (type === NotificationType.UNREAD) { void this.publishWarningsAndAlerts(); } @@ -426,14 +523,14 @@ export class NotificationsService { * @remarks Ensures the notifications directory exists before emptying it */ public async deleteNotifications(type: NotificationType) { - await emptyDir(this.paths()[type]); - NotificationsService.overview[type.toLowerCase()] = { - alert: 0, - info: 0, - warning: 0, - total: 0, - }; - await this.publishOverview(); + await this.waitForInitialization(); + const folder = this.paths()[type]; + const ids = new Set(await readdir(folder)); + for (const [id, copies] of this.notificationFiles) { + if (copies[type] !== undefined) ids.add(id); + } + await emptyDir(folder); + await Promise.all([...ids].map((id) => this.synchronizeNotification(id))); if (type === NotificationType.UNREAD) { void this.publishWarningsAndAlerts(); } @@ -456,72 +553,19 @@ export class NotificationsService { * CRUD: Updating Notifications *------------------------------------------------------------------------**/ - /** - * Returns a function that: - * 1. moves a notification from one category to another. - * 2. updates stats overview - * 3. updates the stats snapshot if provided - * - * Note: the returned function implicitly triggers a pubsub event via `fs.rename`, - * which is expected to trigger `NOTIFICATION_ADDED` & `NOTIFICATION_OVERVIEW`. - * - * The published overview will include the update from this operation. - * - * @param params - * @returns lambda function - */ - private moveNotification(params: { - from: NotificationType; - to: NotificationType; - snapshot?: NotificationOverview; - }) { - const { from, to, snapshot } = params; - const fromStatKey = from.toLowerCase(); - const toStatKey = to.toLowerCase(); + private moveNotification(params: { from: NotificationType; to: NotificationType }) { + const { from, to } = params; return async (notification: Notification) => { const currentPath = this.notificationPath(notification.id, from); const targetPath = this.notificationPath(notification.id, to); - - /**----------------------- - * Event, PubSub, & Overview Update logic - * - * We assume `rename` kicks off 'unlink' and 'add' events - * in the chokidar file watcher (see `getNotificationsWatcher`). - * - * We assume the 'add' handler publishes to - * NOTIFICATION_ADDED & NOTIFICATION_OVERVIEW, - * and that no pubsub or overview updates occur upon 'unlink'. - * - * Thus, we explicitly update our state here via `decrement` and implicitly expect - * it to be updated (i.e. incremented & pubsub'd) via our filesystem changes. - * - * The reasons for this discrepancy are: - * - Backwards compatibility: not every notification will be created through this API, - * so we track state by watching the store (i.e. the file system). - * - * - Technical Limitations: By the time the unlink event fires, the notification file - * can no longer be read. This means we can only track overview totals accurately; - * to track other stats, we have to update them manually, prior to file deletion. - *------------------------**/ - this.decrement(notification.importance, NotificationsService.overview[fromStatKey]); - try { - await rename(currentPath, targetPath); - } catch (err) { - // revert our earlier decrement - // we do it this way (decrement -> try rename -> revert if error) to avoid - // a race condition between `rename` and `decrement` - this.increment(notification.importance, NotificationsService.overview[fromStatKey]); - throw err; - } - - if (snapshot) { - this.decrement(notification.importance, snapshot[fromStatKey]); - this.increment(notification.importance, snapshot[toStatKey]); - } + await rename(currentPath, targetPath); + // A legacy archive copy may already exist, so rename need not emit an add event. + await this.synchronizeNotification(notification.id); }; } public async archiveNotification({ id }: Pick): Promise { + await this.waitForInitialization(); const unreadPath = this.notificationPath(id, NotificationType.UNREAD); // We expect to only archive 'unread' notifications, but it's possible that the notification @@ -531,21 +575,10 @@ export class NotificationsService { throw new AppError(`Could not find notification in unreads: ${id}`, 404); } - /**----------------------- - * Why we use a snapshot - * - * An implicit update to `overview` creates a race condition: - * it might be missing changes from the 'add' event (i.e. incrementing the notification's new category). - * - * So, we use & modify a snapshot of the overview to make sure we're returning accurate - * data to the client. - *------------------------**/ - const snapshot = this.getOverview(); const notification = await this.loadNotificationFile(unreadPath, NotificationType.UNREAD); const moveToArchive = this.moveNotification({ from: NotificationType.UNREAD, to: NotificationType.ARCHIVE, - snapshot, }); await moveToArchive(notification); @@ -558,6 +591,7 @@ export class NotificationsService { } public async markAsUnread({ id }: Pick): Promise { + await this.waitForInitialization(); const archivePath = this.notificationPath(id, NotificationType.ARCHIVE); // the target notification might not be in the archive! if (!(await fileExists(archivePath))) { @@ -565,14 +599,10 @@ export class NotificationsService { throw new AppError(`Could not find notification in archive: ${id}`, 404); } - // we use a snapshot to provide an accurate overview update - // otherwise, we'd enter a race condition with the 'add' file watcher event handler - const snapshot = this.getOverview(); const notification = await this.loadNotificationFile(archivePath, NotificationType.ARCHIVE); const moveToUnread = this.moveNotification({ from: NotificationType.ARCHIVE, to: NotificationType.UNREAD, - snapshot, }); await moveToUnread(notification); @@ -584,6 +614,7 @@ export class NotificationsService { } public async archiveAll(importance?: NotificationImportance) { + await this.waitForInitialization(); const { UNREAD } = this.paths(); if (!importance) { @@ -591,21 +622,20 @@ export class NotificationsService { return { overview: NotificationsService.overview }; } - const overviewSnapshot = this.getOverview(); const unreads = await this.listFilesInFolder(UNREAD); const [notifications] = await this.loadNotificationsFromPaths(unreads, { importance }); const archive = this.moveNotification({ from: NotificationType.UNREAD, to: NotificationType.ARCHIVE, - snapshot: overviewSnapshot, }); const stats = await batchProcess(notifications, archive); void this.publishWarningsAndAlerts(); - return { ...stats, overview: overviewSnapshot }; + return { ...stats, overview: this.getOverview() }; } public async unarchiveAll(importance?: NotificationImportance) { + await this.waitForInitialization(); const { ARCHIVE } = this.paths(); if (!importance) { @@ -614,18 +644,16 @@ export class NotificationsService { return { overview: NotificationsService.overview }; } - const overviewSnapshot = this.getOverview(); const archives = await this.listFilesInFolder(ARCHIVE); const [notifications] = await this.loadNotificationsFromPaths(archives, { importance }); const unArchive = this.moveNotification({ from: NotificationType.ARCHIVE, to: NotificationType.UNREAD, - snapshot: overviewSnapshot, }); const stats = await batchProcess(notifications, unArchive); void this.publishWarningsAndAlerts(); - return { ...stats, overview: overviewSnapshot }; + return { ...stats, overview: this.getOverview() }; } /** @@ -666,26 +694,27 @@ export class NotificationsService { * @returns An array of all notifications in the system. */ public async getNotifications(filters: NotificationFilter): Promise { + await this.waitForInitialization(); this.logger.verbose('Getting Notifications'); const { type = NotificationType.UNREAD } = filters; - const { ARCHIVE, UNREAD } = this.paths(); - let files: string[]; + const files = await this.listNotificationFiles(type); + + const [notifications] = await this.loadNotificationsFromPaths(files, filters); + return notifications; + } + private async listNotificationFiles(type: NotificationType): Promise { + const { ARCHIVE, UNREAD } = this.paths(); if (type === NotificationType.UNREAD) { - files = await this.listFilesInFolder(UNREAD); - } else { - // Exclude notifications present in both unread & archive from archive. - //* this is necessary because the legacy script writes new notifications to both places. - //* this should be a temporary measure. - const unreads = new Set(await readdir(UNREAD)); - files = await this.listFilesInFolder(ARCHIVE, (archives) => { - return archives.filter((file) => !unreads.has(file)); - }); + return this.listFilesInFolder(UNREAD); } - const [notifications] = await this.loadNotificationsFromPaths(files, filters); - return notifications; + // The legacy notifier writes each new notification to both directories. + const unreads = new Set(await readdir(UNREAD)); + return this.listFilesInFolder(ARCHIVE, (archives) => + archives.filter((file) => !unreads.has(file)) + ); } /** @@ -990,6 +1019,7 @@ export class NotificationsService { *------------------------------------------------------------------------**/ private async loadUnreadNotifications(): Promise { + await this.waitForInitialization(); const { UNREAD } = this.paths(); const files = await this.listFilesInFolder(UNREAD); const [notifications] = await this.loadNotificationsFromPaths(files, { diff --git a/api/src/unraid-api/graph/resolvers/vms/vms.resolver.ts b/api/src/unraid-api/graph/resolvers/vms/vms.resolver.ts index 13f066ce7f..481d6fdf7d 100644 --- a/api/src/unraid-api/graph/resolvers/vms/vms.resolver.ts +++ b/api/src/unraid-api/graph/resolvers/vms/vms.resolver.ts @@ -10,7 +10,7 @@ import { VmsService } from '@app/unraid-api/graph/resolvers/vms/vms.service.js'; export class VmsResolver { constructor(private readonly vmsService: VmsService) {} - @Query(() => Vms, { description: 'Get information about all VMs on the system' }) + @Query(() => Vms, { nullable: true, description: 'Get information about all VMs on the system' }) @UsePermissions({ action: AuthAction.READ_ANY, resource: Resource.VMS, diff --git a/web/src/composables/gql/graphql.ts b/web/src/composables/gql/graphql.ts index e9be3b367b..ff17bb25e5 100644 --- a/web/src/composables/gql/graphql.ts +++ b/web/src/composables/gql/graphql.ts @@ -2390,7 +2390,7 @@ export type Query = { disk: Disk; disks: Array; display: InfoDisplay; - docker: Docker; + docker?: Maybe; flash: Flash; /** Get JSON Schema for API key creation form */ getApiKeyCreationFormSchema: ApiKeyFormSettings; @@ -2454,7 +2454,7 @@ export type Query = { validateOidcSession: OidcSessionValidation; vars: Vars; /** Get information about all VMs on the system */ - vms: Vms; + vms?: Maybe; }; @@ -3690,14 +3690,14 @@ export type SetThemeMutation = { __typename?: 'Mutation', customization: { __typ export type GetDockerContainerSizesQueryVariables = Exact<{ [key: string]: never; }>; -export type GetDockerContainerSizesQuery = { __typename?: 'Query', docker: { __typename?: 'Docker', id: string, containers: Array<{ __typename?: 'DockerContainer', id: string, names: Array, sizeRootFs?: number | null, sizeRw?: number | null, sizeLog?: number | null }> } }; +export type GetDockerContainerSizesQuery = { __typename?: 'Query', docker?: { __typename?: 'Docker', id: string, containers: Array<{ __typename?: 'DockerContainer', id: string, names: Array, sizeRootFs?: number | null, sizeRw?: number | null, sizeLog?: number | null }> } | null }; export type GetDockerContainersQueryVariables = Exact<{ skipCache?: InputMaybe; }>; -export type GetDockerContainersQuery = { __typename?: 'Query', docker: { __typename?: 'Docker', id: string, portConflicts: { __typename?: 'DockerPortConflicts', containerPorts: Array<{ __typename?: 'DockerContainerPortConflict', privatePort: number, type: ContainerPortType, containers: Array<{ __typename?: 'DockerPortConflictContainer', id: string, name: string }> }>, lanPorts: Array<{ __typename?: 'DockerLanPortConflict', lanIpPort: string, publicPort?: number | null, type: ContainerPortType, containers: Array<{ __typename?: 'DockerPortConflictContainer', id: string, name: string }> }> }, containers: Array<{ __typename?: 'DockerContainer', id: string, names: Array, state: ContainerState, status: string, image: string, created: number, lanIpPorts?: Array | null, autoStart: boolean, autoStartOrder?: number | null, autoStartWait?: number | null, networkSettings?: any | null, mounts?: Array | null, isOrphaned: boolean, projectUrl?: string | null, registryUrl?: string | null, supportUrl?: string | null, iconUrl?: string | null, webUiUrl?: string | null, shell?: string | null, tailscaleEnabled: boolean, ports: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }>, hostConfig?: { __typename?: 'ContainerHostConfig', networkMode: string } | null, templatePorts?: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }> | null }>, organizer: { __typename?: 'ResolvedOrganizerV1', version: number, views: Array<{ __typename?: 'ResolvedOrganizerView', id: string, name: string, rootId: string, prefs?: any | null, flatEntries: Array<{ __typename?: 'FlatOrganizerEntry', id: string, type: string, name: string, parentId?: string | null, depth: number, position: number, path: Array, hasChildren: boolean, childrenIds: Array, meta?: { __typename?: 'DockerContainer', id: string, names: Array, state: ContainerState, status: string, image: string, lanIpPorts?: Array | null, autoStart: boolean, autoStartWait?: number | null, networkSettings?: any | null, mounts?: Array | null, created: number, isUpdateAvailable?: boolean | null, isRebuildReady?: boolean | null, templatePath?: string | null, isOrphaned: boolean, projectUrl?: string | null, registryUrl?: string | null, supportUrl?: string | null, iconUrl?: string | null, webUiUrl?: string | null, shell?: string | null, tailscaleEnabled: boolean, ports: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }>, hostConfig?: { __typename?: 'ContainerHostConfig', networkMode: string } | null, templatePorts?: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }> | null } | null }> }> } } }; +export type GetDockerContainersQuery = { __typename?: 'Query', docker?: { __typename?: 'Docker', id: string, portConflicts: { __typename?: 'DockerPortConflicts', containerPorts: Array<{ __typename?: 'DockerContainerPortConflict', privatePort: number, type: ContainerPortType, containers: Array<{ __typename?: 'DockerPortConflictContainer', id: string, name: string }> }>, lanPorts: Array<{ __typename?: 'DockerLanPortConflict', lanIpPort: string, publicPort?: number | null, type: ContainerPortType, containers: Array<{ __typename?: 'DockerPortConflictContainer', id: string, name: string }> }> }, containers: Array<{ __typename?: 'DockerContainer', id: string, names: Array, state: ContainerState, status: string, image: string, created: number, lanIpPorts?: Array | null, autoStart: boolean, autoStartOrder?: number | null, autoStartWait?: number | null, networkSettings?: any | null, mounts?: Array | null, isOrphaned: boolean, projectUrl?: string | null, registryUrl?: string | null, supportUrl?: string | null, iconUrl?: string | null, webUiUrl?: string | null, shell?: string | null, tailscaleEnabled: boolean, ports: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }>, hostConfig?: { __typename?: 'ContainerHostConfig', networkMode: string } | null, templatePorts?: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }> | null }>, organizer: { __typename?: 'ResolvedOrganizerV1', version: number, views: Array<{ __typename?: 'ResolvedOrganizerView', id: string, name: string, rootId: string, prefs?: any | null, flatEntries: Array<{ __typename?: 'FlatOrganizerEntry', id: string, type: string, name: string, parentId?: string | null, depth: number, position: number, path: Array, hasChildren: boolean, childrenIds: Array, meta?: { __typename?: 'DockerContainer', id: string, names: Array, state: ContainerState, status: string, image: string, lanIpPorts?: Array | null, autoStart: boolean, autoStartWait?: number | null, networkSettings?: any | null, mounts?: Array | null, created: number, isUpdateAvailable?: boolean | null, isRebuildReady?: boolean | null, templatePath?: string | null, isOrphaned: boolean, projectUrl?: string | null, registryUrl?: string | null, supportUrl?: string | null, iconUrl?: string | null, webUiUrl?: string | null, shell?: string | null, tailscaleEnabled: boolean, ports: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }>, hostConfig?: { __typename?: 'ContainerHostConfig', networkMode: string } | null, templatePorts?: Array<{ __typename?: 'ContainerPort', privatePort?: number | null, publicPort?: number | null, type: ContainerPortType }> | null } | null }> }> } } | null }; export type CreateDockerFolderWithItemsMutationVariables = Exact<{ name: Scalars['String']['input']; @@ -3732,7 +3732,7 @@ export type GetDockerContainerLogsQueryVariables = Exact<{ }>; -export type GetDockerContainerLogsQuery = { __typename?: 'Query', docker: { __typename?: 'Docker', logs: { __typename?: 'DockerContainerLogs', containerId: string, cursor?: string | null, lines: Array<{ __typename?: 'DockerContainerLogLine', timestamp: string, message: string }> } } }; +export type GetDockerContainerLogsQuery = { __typename?: 'Query', docker?: { __typename?: 'Docker', logs: { __typename?: 'DockerContainerLogs', containerId: string, cursor?: string | null, lines: Array<{ __typename?: 'DockerContainerLogLine', timestamp: string, message: string }> } } | null }; export type MoveDockerEntriesToFolderMutationVariables = Exact<{ destinationFolderId: Scalars['String']['input']; @@ -3808,7 +3808,7 @@ export type GetContainerTailscaleStatusQueryVariables = Exact<{ }>; -export type GetContainerTailscaleStatusQuery = { __typename?: 'Query', docker: { __typename?: 'Docker', container?: { __typename?: 'DockerContainer', id: string, tailscaleStatus?: { __typename?: 'TailscaleStatus', online: boolean, version?: string | null, latestVersion?: string | null, updateAvailable: boolean, hostname?: string | null, dnsName?: string | null, relay?: string | null, relayName?: string | null, tailscaleIps?: Array | null, primaryRoutes?: Array | null, isExitNode: boolean, webUiUrl?: string | null, keyExpiry?: string | null, keyExpiryDays?: number | null, keyExpired: boolean, backendState?: string | null, authUrl?: string | null, exitNodeStatus?: { __typename?: 'TailscaleExitNodeStatus', online: boolean, tailscaleIps?: Array | null } | null } | null } | null } }; +export type GetContainerTailscaleStatusQuery = { __typename?: 'Query', docker?: { __typename?: 'Docker', container?: { __typename?: 'DockerContainer', id: string, tailscaleStatus?: { __typename?: 'TailscaleStatus', online: boolean, version?: string | null, latestVersion?: string | null, updateAvailable: boolean, hostname?: string | null, dnsName?: string | null, relay?: string | null, relayName?: string | null, tailscaleIps?: Array | null, primaryRoutes?: Array | null, isExitNode: boolean, webUiUrl?: string | null, keyExpiry?: string | null, keyExpiryDays?: number | null, keyExpired: boolean, backendState?: string | null, authUrl?: string | null, exitNodeStatus?: { __typename?: 'TailscaleExitNodeStatus', online: boolean, tailscaleIps?: Array | null } | null } | null } | null } | null }; export type UnpauseDockerContainerMutationVariables = Exact<{ id: Scalars['PrefixedID']['input'];