diff --git a/.changeset/admin-operations.md b/.changeset/admin-operations.md new file mode 100644 index 000000000..32b35af74 --- /dev/null +++ b/.changeset/admin-operations.md @@ -0,0 +1,5 @@ +--- +"nostream": minor +--- + +Add admin Operations area with mirror and worker heartbeats, NIP-05 verification ops, storage counts, and dashboard view. diff --git a/resources/admin/assets/admin-ops.js b/resources/admin/assets/admin-ops.js new file mode 100644 index 000000000..10a429d82 --- /dev/null +++ b/resources/admin/assets/admin-ops.js @@ -0,0 +1,387 @@ +(() => { + const config = window.__ADMIN_DASHBOARD__ || { pathPrefix: '' } + const adminBase = `${config.pathPrefix || ''}/admin` + + const mirrorsBody = document.getElementById('ops-mirrors-body') + if (!mirrorsBody) { + return + } + + const elements = { + mirrorsSync: document.getElementById('ops-mirrors-sync'), + mirrorsBody, + workersSync: document.getElementById('ops-workers-sync'), + workersBody: document.getElementById('ops-workers-body'), + nip05Error: document.getElementById('ops-nip05-error'), + nip05Domain: document.getElementById('ops-nip05-domain'), + nip05Verified: document.getElementById('ops-nip05-verified'), + nip05Stale: document.getElementById('ops-nip05-stale'), + nip05MinFailures: document.getElementById('ops-nip05-min-failures'), + nip05Refresh: document.getElementById('ops-nip05-refresh'), + nip05Body: document.getElementById('ops-nip05-body'), + nip05Pagination: document.getElementById('ops-nip05-pagination'), + storageKinds: document.getElementById('ops-storage-kinds'), + storageSince: document.getElementById('ops-storage-since'), + storageUntil: document.getElementById('ops-storage-until'), + storageRun: document.getElementById('ops-storage-run'), + storageError: document.getElementById('ops-storage-error'), + storageBody: document.getElementById('ops-storage-body'), + } + + let pollTimer + let nip05Offset = 0 + const nip05Limit = 50 + + const formatTime = (iso) => { + if (!iso) { + return '—' + } + const date = new Date(iso) + if (Number.isNaN(date.getTime())) { + return iso + } + return date.toISOString().replace('T', ' ').replace(/\.\d{3}Z$/, ' UTC') + } + + const shortPubkey = (pubkey) => (pubkey.length > 16 ? `${pubkey.slice(0, 8)}…${pubkey.slice(-8)}` : pubkey) + + const escapeHtml = (value) => + String(value) + .replaceAll('&', '&') + .replaceAll('<', '<') + .replaceAll('>', '>') + .replaceAll('"', '"') + .replaceAll("'", ''') + + const escapeAttr = (value) => escapeHtml(value).replaceAll('`', '`') + + const signalBadge = (stale) => { + if (stale) { + return 'Stale' + } + return 'Live' + } + + const workerSignalBadge = (worker) => { + if (worker.monitored === false) { + return 'Unmonitored' + } + return signalBadge(worker.stale) + } + + const adminFetch = async (path, options = {}) => { + const response = await fetch(`${adminBase}${path}`, { + credentials: 'same-origin', + ...options, + }) + if (response.status === 401) { + throw new Error('Unauthorized') + } + return response + } + + const renderMirrors = (mirrors) => { + if (!Array.isArray(mirrors) || !mirrors.length) { + elements.mirrorsBody.innerHTML = + 'No mirrors configured.' + return + } + + elements.mirrorsBody.innerHTML = mirrors + .map((mirror) => { + const heartbeat = mirror.heartbeat + const state = heartbeat?.connectionState ?? 'unknown' + const reconnects = heartbeat?.reconnectCount ?? 0 + const lastEvent = formatTime(heartbeat?.lastEventAt) + const errorNote = heartbeat?.lastError + ? `${escapeHtml(heartbeat.lastError)}` + : '' + return ` + ${escapeHtml(mirror.index)} + ${escapeHtml(mirror.targetUrl)}${errorNote} + ${escapeHtml(state)} + ${escapeHtml(reconnects)} + ${escapeHtml(lastEvent)} + ${signalBadge(mirror.stale)} + ` + }) + .join('') + } + + const renderWorkers = (workers) => { + if (!Array.isArray(workers) || !workers.length) { + elements.workersBody.innerHTML = + 'No workers configured.' + return + } + + elements.workersBody.innerHTML = workers + .map((worker) => { + const heartbeat = worker.heartbeat + const finished = formatTime(heartbeat?.lastRunFinishedAt ?? heartbeat?.updatedAt) + const lastError = heartbeat?.lastError ?? '—' + return ` + ${escapeHtml(worker.label)} + ${escapeHtml(finished)} + ${escapeHtml(lastError)} + ${workerSignalBadge(worker)} + ` + }) + .join('') + } + + const renderNip05 = (rows, pagination) => { + if (!Array.isArray(rows) || !rows.length) { + elements.nip05Body.innerHTML = 'No matching verifications.' + } else { + elements.nip05Body.innerHTML = rows + .map( + (row) => ` + ${escapeHtml(shortPubkey(row.pubkey))} + ${escapeHtml(row.nip05)} + ${row.isVerified ? 'yes' : 'no'} + ${escapeHtml(row.failureCount)} + ${escapeHtml(formatTime(row.lastVerifiedAt))} + ${row.stale ? 'yes' : 'no'} + + `, + ) + .join('') + } + + if (elements.nip05Pagination && pagination) { + const { offset, limit, total } = pagination + elements.nip05Pagination.replaceChildren() + + const summary = document.createElement('span') + summary.textContent = `Showing ${offset + 1}–${Math.min(offset + limit, total)} of ${total}` + elements.nip05Pagination.append(summary) + + if (total > limit) { + const controls = document.createElement('span') + controls.className = 'ms-2' + + const previous = document.createElement('button') + previous.type = 'button' + previous.className = 'btn btn-console btn-sm me-1' + previous.textContent = 'Previous' + previous.disabled = offset <= 0 + previous.addEventListener('click', () => { + nip05Offset = Math.max(0, offset - limit) + void loadNip05() + }) + + const next = document.createElement('button') + next.type = 'button' + next.className = 'btn btn-console btn-sm' + next.textContent = 'Next' + next.disabled = offset + limit >= total + next.addEventListener('click', () => { + nip05Offset = offset + limit + void loadNip05() + }) + + controls.append(previous, next) + elements.nip05Pagination.append(controls) + } + } + } + + const renderStorage = (counts) => { + if (!Array.isArray(counts) || !counts.length) { + elements.storageBody.innerHTML = + 'No counts returned.' + return + } + + elements.storageBody.innerHTML = counts + .map( + (row) => ` + ${escapeHtml((row.kinds ?? []).join(', '))} + ${escapeHtml(row.count)} + `, + ) + .join('') + } + + const hideNip05Error = () => elements.nip05Error?.classList.add('d-none') + const showNip05Error = (message) => { + if (!elements.nip05Error) { + return + } + elements.nip05Error.textContent = message + elements.nip05Error.classList.remove('d-none') + } + + const hideStorageError = () => elements.storageError?.classList.add('d-none') + const showStorageError = (message) => { + if (!elements.storageError) { + return + } + elements.storageError.textContent = message + elements.storageError.classList.remove('d-none') + } + + const loadMirrors = async () => { + const response = await adminFetch('/ops/mirrors') + if (!response.ok) { + throw new Error('Failed to load mirrors') + } + const payload = await response.json() + renderMirrors(payload.mirrors) + if (elements.mirrorsSync) { + elements.mirrorsSync.innerHTML = `> mirrors: updated ${formatTime(new Date().toISOString())}` + } + } + + const loadWorkers = async () => { + const response = await adminFetch('/ops/workers') + if (!response.ok) { + throw new Error('Failed to load workers') + } + const payload = await response.json() + renderWorkers(payload.workers) + if (elements.workersSync) { + elements.workersSync.innerHTML = `> workers: updated ${formatTime(new Date().toISOString())}` + } + } + + const buildNip05Query = () => { + const params = new URLSearchParams() + params.set('limit', String(nip05Limit)) + params.set('offset', String(nip05Offset)) + + const domain = elements.nip05Domain?.value.trim() + if (domain) { + params.set('domain', domain) + } + + const verified = elements.nip05Verified?.value + if (verified) { + params.set('verified', verified) + } + + const stale = elements.nip05Stale?.value + if (stale) { + params.set('stale', stale) + } + + const minFailures = elements.nip05MinFailures?.value + if (minFailures !== undefined && minFailures !== '') { + params.set('minFailures', minFailures) + } + + return params.toString() + } + + const loadNip05 = async () => { + hideNip05Error() + const response = await adminFetch(`/ops/nip05-verifications?${buildNip05Query()}`) + const payload = await response.json().catch(() => ({})) + if (!response.ok) { + showNip05Error(payload.error ?? 'Failed to load NIP-05 verifications') + return + } + renderNip05(payload.verifications, payload.pagination) + } + + const loadStorage = async () => { + hideStorageError() + const kinds = elements.storageKinds?.value.trim() + if (!kinds) { + showStorageError('Enter at least one kind.') + return + } + + const params = new URLSearchParams({ kinds }) + const since = elements.storageSince?.value + const until = elements.storageUntil?.value + if (since) { + params.set('since', since) + } + if (until) { + params.set('until', until) + } + + const response = await adminFetch(`/ops/storage/counts?${params.toString()}`) + const payload = await response.json().catch(() => ({})) + if (!response.ok) { + showStorageError(payload.error ?? 'Storage count query failed') + return + } + renderStorage(payload.counts) + } + + const reverify = async (pubkey) => { + hideNip05Error() + const response = await adminFetch(`/ops/nip05-verifications/${encodeURIComponent(pubkey)}/reverify`, { + method: 'POST', + }) + const payload = await response.json().catch(() => ({})) + if (!response.ok) { + showNip05Error(payload.error ?? 'Re-verification failed') + return + } + await loadNip05() + } + + const refreshAll = async () => { + try { + await Promise.all([loadMirrors(), loadWorkers(), loadNip05()]) + } catch { + showNip05Error('Unable to refresh operations data.') + } + } + + const stopPolling = () => { + if (pollTimer) { + clearInterval(pollTimer) + pollTimer = undefined + } + } + + const startPolling = () => { + stopPolling() + pollTimer = setInterval(() => { + void refreshAll() + }, 30_000) + } + + const resetNip05Filters = () => { + nip05Offset = 0 + void loadNip05() + } + + elements.nip05Refresh?.addEventListener('click', resetNip05Filters) + elements.nip05Domain?.addEventListener('change', resetNip05Filters) + elements.nip05Verified?.addEventListener('change', resetNip05Filters) + elements.nip05Stale?.addEventListener('change', resetNip05Filters) + elements.nip05MinFailures?.addEventListener('change', resetNip05Filters) + + elements.storageRun?.addEventListener('click', () => { + void loadStorage() + }) + + elements.nip05Body?.addEventListener('click', (event) => { + const target = event.target + if (!(target instanceof HTMLElement)) { + return + } + const pubkey = target.getAttribute('data-reverify-pubkey') + if (!pubkey) { + return + } + void reverify(pubkey) + }) + + window.__ADMIN_OPS__ = { + load: () => { + nip05Offset = 0 + void refreshAll() + startPolling() + }, + stop: () => { + stopPolling() + }, + } +})() diff --git a/resources/admin/assets/dashboard.js b/resources/admin/assets/dashboard.js index 746ea5ad0..b008c7b36 100644 --- a/resources/admin/assets/dashboard.js +++ b/resources/admin/assets/dashboard.js @@ -345,6 +345,7 @@ unloadGrafanaFrames() stopMetricsStream() stopNetworkHealthPolling() + window.__ADMIN_OPS__?.stop?.() setNavOpen(false) loginPanel.classList.remove('d-none') dashboardPanel.classList.add('d-none') @@ -374,9 +375,8 @@ loadGrafanaFrames() startMetricsStream() - void refreshNetworkHealth() - void refreshNetworkHealthHistory() - startNetworkHealthPolling() + stopNetworkHealthPolling() + window.__ADMIN_OPS__?.stop?.() } const stopNetworkHealthPolling = () => { @@ -1485,16 +1485,33 @@ }) if (viewId === 'settings-view') { + stopNetworkHealthPolling() + window.__ADMIN_OPS__?.stop?.() + stopMetricsStream() void loadSettingsData() return } if (viewId === 'notifications-view') { + stopNetworkHealthPolling() + window.__ADMIN_OPS__?.stop?.() + stopMetricsStream() window.__ADMIN_NOTIFICATIONS__?.load?.() return } + if (viewId === 'operations-view') { + stopMetricsStream() + void refreshNetworkHealth() + void refreshNetworkHealthHistory() + startNetworkHealthPolling() + window.__ADMIN_OPS__?.load?.() + return + } + if (viewId === 'metrics-view') { + stopNetworkHealthPolling() + window.__ADMIN_OPS__?.stop?.() startMetricsStream() } } diff --git a/resources/admin/dashboard.html b/resources/admin/dashboard.html index c51206f42..2b514deac 100644 --- a/resources/admin/dashboard.html +++ b/resources/admin/dashboard.html @@ -65,6 +65,7 @@

Sign in

@@ -102,37 +103,6 @@

Health

-
-

Network Health

-

> probes: waiting for snapshot…

-
No external probe results yet. Enable NIP-66 relay monitoring to populate this panel.
-
-

HIST Recent probe runs

-
-
-
-
-
-

RUN Probe status

-

Pending

-
-
-
-
-

TGT Targets

-

0

-
-
-
-
-

TS Last probe run

-

—

-
-
-
-
-
-

Throughput

@@ -232,6 +202,170 @@

Charts

+
+
+

Mirror health

+

> mirrors: not loaded

+
+ + + + + + + + + + + + + + + +
IndexTargetStateReconnectsLast eventSignal
No mirrors configured.
+
+
+ +
+

Worker ops

+

> workers: not loaded

+
+ + + + + + + + + + + + + +
WorkerLast runLast errorSignal
No workers reported yet.
+
+
+ +
+

Tor / I2P & network probes

+

> probes: waiting for snapshot…

+
No external probe results yet. Enable NIP-66 relay monitoring to populate this panel.
+
+

HIST Recent probe runs

+
+
+
+
+
+

RUN Probe status

+

Pending

+
+
+
+
+

TGT Targets

+

0

+
+
+
+
+

TS Last probe run

+

—

+
+
+
+
+
+ +
+

NIP-05 verifications

+ +
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+ +
+
+ + + + + + + + + + + + + + + + +
PubkeyNIP-05VerifiedFailuresLast verifiedStale
No rows loaded.
+
+

+
+ +
+

Storage counts

+

Per-kind counts use NIP-45-style filters (comma-separated kinds, optional since/until unix seconds).

+
+
+ + +
+
+ + +
+
+ + +
+ +
+ +
+ + + + + + + + + + + +
KindsCount
Run a query to load counts.
+
+
+
+

Operator notifications

@@ -392,5 +526,6 @@

Preview diff

+ diff --git a/src/@types/admin-ops.ts b/src/@types/admin-ops.ts new file mode 100644 index 000000000..69b09d5ff --- /dev/null +++ b/src/@types/admin-ops.ts @@ -0,0 +1,60 @@ +export type AdminOpsConnectionState = 'connecting' | 'open' | 'closed' + +export interface AdminOpsMirrorHeartbeat { + index: string + targetUrl: string + connectionState: AdminOpsConnectionState + reconnectCount: number + lastError?: string + lastEventAt?: string + updatedAt: string +} + +export interface AdminOpsWorkerHeartbeat { + type: string + index?: string + lastRunStartedAt?: string + lastRunFinishedAt?: string + lastError?: string + updatedAt: string +} + +export interface AdminOpsConfiguredMirror { + index: string + targetUrl: string + enabled: boolean +} + +export interface AdminOpsConfiguredWorker { + type: string + index?: string + enabled: boolean + label: string +} + +export interface AdminOpsMirrorRow extends AdminOpsConfiguredMirror { + heartbeat?: AdminOpsMirrorHeartbeat + stale: boolean +} + +export interface AdminOpsWorkerRow extends AdminOpsConfiguredWorker { + heartbeat?: AdminOpsWorkerHeartbeat + monitored: boolean + stale: boolean +} + +export interface AdminNip05VerificationAdminRow { + pubkey: string + nip05: string + domain: string + isVerified: boolean + failureCount: number + lastVerifiedAt: string | null + lastCheckedAt: string + stale: boolean +} + +export interface AdminOpsStorageCountRow { + kinds: number[] + count: number +} diff --git a/src/@types/repositories.ts b/src/@types/repositories.ts index 3f9a5d7de..1846fd55d 100644 --- a/src/@types/repositories.ts +++ b/src/@types/repositories.ts @@ -44,7 +44,7 @@ export interface IEventRepository { upsert(event: Event): Promise upsertMany(events: Event[]): Promise findByFilters(filters: SubscriptionFilter[]): IQueryResult - countByFilters(filters: SubscriptionFilter[]): Promise + countByFilters(filters: SubscriptionFilter[], options?: { timeoutMs?: number }): Promise findReconciliationItems(filter: SubscriptionFilter, limit?: number): IQueryResult deleteByPubkeyAndIds(pubkey: Pubkey, ids: EventId[]): Promise deleteByPubkeyExceptKinds(pubkey: Pubkey, excludedKinds: number[]): Promise @@ -81,10 +81,21 @@ export interface IUserRepository { revokeAdmission(pubkey: Pubkey, client?: DatabaseClient): Promise } +export interface Nip05VerificationAdminListQuery { + domain?: string + verified?: boolean + stale?: boolean + minFailures?: number + verifyExpirationMs: number + limit: number + offset: number +} + export interface INip05VerificationRepository { findByPubkey(pubkey: Pubkey): Promise upsert(verification: Nip05Verification): Promise findPendingVerifications(updateFrequencyMs: number, maxFailures: number, limit: number): Promise + listForAdmin(query: Nip05VerificationAdminListQuery): Promise<{ rows: Nip05Verification[]; total: number }> deleteByPubkey(pubkey: Pubkey): Promise } diff --git a/src/app/maintenance-worker.ts b/src/app/maintenance-worker.ts index acd9114b1..6a4a6f04f 100644 --- a/src/app/maintenance-worker.ts +++ b/src/app/maintenance-worker.ts @@ -1,30 +1,31 @@ -import { - DEFAULT_NIP05_MAX_CONSECUTIVE_FAILURES, - DEFAULT_NIP05_VERIFY_UPDATE_FREQUENCY_MS, - Nip05VerificationOutcome, - verifyNip05Identifier, -} from '../utils/nip05' -import { IMaintenanceService, INotificationOutboxService, IPaymentsService } from '../@types/services' +import { FSWatcher } from 'fs' import { mergeDeepLeft, path, pipe } from 'ramda' +import { ICacheAdapter } from '../@types/adapters' import { IRunnable } from '../@types/base' - -import { createLogger } from '../factories/logger-factory' -import { PENDING_INVOICE_PAGE_SIZE } from '../services/payments-service' -import { delayMs } from '../utils/misc' +import { InvoiceStatus } from '../@types/invoice' +import { Nip05Verification } from '../@types/nip05' import { OperatorNotificationEventType } from '../@types/operator-notifications' import { INip05VerificationRepository, INotificationDeliveryLogRepository, INotificationOutboxRepository, } from '../@types/repositories' -import { InvoiceStatus } from '../@types/invoice' -import { isExpiredInvoice } from '../utils/invoice' -import { Nip05Verification } from '../@types/nip05' -import { FSWatcher } from 'fs' - +import { IMaintenanceService, INotificationOutboxService, IPaymentsService } from '../@types/services' import { Settings } from '../@types/settings' -import { SettingsStatic } from '../utils/settings' +import { createLogger } from '../factories/logger-factory' +import { PENDING_INVOICE_PAGE_SIZE } from '../services/payments-service' import { shutdownMetricsTelemetry } from '../telemetry/metrics' +import { sanitizeAdminOpsError } from '../utils/admin-ops-sanitize' +import { publishAdminOpsWorkerHeartbeatSafely } from '../utils/admin-ops-store' +import { isExpiredInvoice } from '../utils/invoice' +import { delayMs } from '../utils/misc' +import { + DEFAULT_NIP05_MAX_CONSECUTIVE_FAILURES, + DEFAULT_NIP05_VERIFY_UPDATE_FREQUENCY_MS, + Nip05VerificationOutcome, + verifyNip05Identifier, +} from '../utils/nip05' +import { SettingsStatic } from '../utils/settings' const UPDATE_INVOICE_INTERVAL = 60000 const NIP05_REVERIFICATION_BATCH_SIZE = 50 @@ -98,6 +99,7 @@ export class MaintenanceWorker implements IRunnable { private readonly notificationOutboxService: INotificationOutboxService, private readonly notificationDeliveryLogRepository: INotificationDeliveryLogRepository, private readonly notificationOutboxRepository: INotificationOutboxRepository, + private readonly opsCache?: ICacheAdapter, ) { this.process .on('SIGINT', this.onExit.bind(this)) @@ -133,6 +135,10 @@ export class MaintenanceWorker implements IRunnable { try { await this.onSchedule() } catch (error) { + await this.publishOpsHeartbeat({ + lastRunFinishedAt: new Date().toISOString(), + lastError: sanitizeAdminOpsError(error), + }) this.onError(error as Error) } finally { this.isRunning = false @@ -140,7 +146,26 @@ export class MaintenanceWorker implements IRunnable { }, UPDATE_INVOICE_INTERVAL) } + private async publishOpsHeartbeat(partial: { + lastRunStartedAt?: string + lastRunFinishedAt?: string + lastError?: string + }): Promise { + if (!this.opsCache) { + return + } + + await publishAdminOpsWorkerHeartbeatSafely(this.opsCache, { + type: 'maintenance', + updatedAt: new Date().toISOString(), + ...partial, + }) + } + private async onSchedule(): Promise { + const lastRunStartedAt = new Date().toISOString() + await this.publishOpsHeartbeat({ lastRunStartedAt }) + const currentSettings = this.settings() const clearOldEventsPromise = this.clearOldEventsSafely() @@ -149,6 +174,10 @@ export class MaintenanceWorker implements IRunnable { if (!path(['payments', 'enabled'], currentSettings)) { await clearOldEventsPromise + await this.publishOpsHeartbeat({ + lastRunStartedAt, + lastRunFinishedAt: new Date().toISOString(), + }) return } @@ -223,6 +252,10 @@ export class MaintenanceWorker implements IRunnable { } await clearOldEventsPromise + await this.publishOpsHeartbeat({ + lastRunStartedAt, + lastRunFinishedAt: new Date().toISOString(), + }) } private async processNotificationOutbox(): Promise { diff --git a/src/app/relay-monitor-worker.ts b/src/app/relay-monitor-worker.ts index e8a7b2f44..95aa2fda3 100644 --- a/src/app/relay-monitor-worker.ts +++ b/src/app/relay-monitor-worker.ts @@ -1,17 +1,23 @@ +import { ICacheAdapter } from '../@types/adapters' +import { AdminOpsWorkerHeartbeat } from '../@types/admin-ops' import { IRunnable } from '../@types/base' import { IRelayProbeSnapshotStore, RelayProbeRunSnapshot } from '../@types/relay-probe-snapshot' import { Settings } from '../@types/settings' import { createLogger } from '../factories/logger-factory' import { INip66EventPublisher } from '../services/nip66-event-publisher' import { shutdownMetricsTelemetry } from '../telemetry/metrics' -import { filterValidProbeTargets, resolveProbeTargets } from '../utils/relay-probe-targets' -import { deriveRelayProbeRunStatus, serializeProbeResults } from '../utils/relay-probe-snapshot' -import { resolvePublicProbeTargetKeys } from '../utils/relay-probe-targets' -import { RelayProbeHistoryStore } from '../utils/relay-probe-history' -import { getEffectiveProbeIntervalSeconds, getNip66HistoryRetention, getProbeIntervalMs } from '../utils/nip66-schedule' +import { ADMIN_OPS_HEARTBEAT_KEEPALIVE_MS, publishAdminOpsWorkerHeartbeatSafely } from '../utils/admin-ops-store' import { getMonitorPrivateKey } from '../utils/monitor-identity' +import { getEffectiveProbeIntervalSeconds, getNip66HistoryRetention, getProbeIntervalMs } from '../utils/nip66-schedule' import { runProbe } from '../utils/relay-probe' import { ProbeOptions, ProbeResult } from '../utils/relay-probe/types' +import { RelayProbeHistoryStore } from '../utils/relay-probe-history' +import { deriveRelayProbeRunStatus, serializeProbeResults } from '../utils/relay-probe-snapshot' +import { + filterValidProbeTargets, + resolveProbeTargets, + resolvePublicProbeTargetKeys, +} from '../utils/relay-probe-targets' const logger = createLogger('relay-monitor-worker') @@ -30,9 +36,16 @@ export const buildProbeOptions = (settings: Settings): ProbeOptions => { export { getProbeIntervalMs } from '../utils/nip66-schedule' +type RelayMonitorOpsHeartbeatFields = Pick< + AdminOpsWorkerHeartbeat, + 'lastRunStartedAt' | 'lastRunFinishedAt' | 'lastError' +> + export class RelayMonitorWorker implements IRunnable { private interval: NodeJS.Timeout | undefined + private opsKeepaliveInterval: NodeJS.Timeout | undefined private isRunning = false + private lastOpsWorkerHeartbeat: RelayMonitorOpsHeartbeatFields = {} public constructor( private readonly process: NodeJS.Process, @@ -41,6 +54,7 @@ export class RelayMonitorWorker implements IRunnable { private readonly probeRunner: RunProbeFn = runProbe, private readonly eventPublisher?: INip66EventPublisher, private readonly historyStore?: RelayProbeHistoryStore, + private readonly opsCache?: ICacheAdapter, ) { this.process .on('SIGINT', this.onExit.bind(this)) @@ -66,6 +80,24 @@ export class RelayMonitorWorker implements IRunnable { this.interval = setInterval(() => { void this.runScheduledProbes() }, intervalMs) + + if (this.opsCache) { + this.opsKeepaliveInterval = setInterval(() => { + this.publishOpsKeepalive() + }, ADMIN_OPS_HEARTBEAT_KEEPALIVE_MS) + } + } + + private publishOpsKeepalive(): void { + if (!this.opsCache || this.lastOpsWorkerHeartbeat.lastRunStartedAt === undefined) { + return + } + + void publishAdminOpsWorkerHeartbeatSafely(this.opsCache, { + type: 'relay-monitor', + updatedAt: new Date().toISOString(), + ...this.lastOpsWorkerHeartbeat, + }) } private async runScheduledProbes(): Promise { @@ -85,7 +117,30 @@ export class RelayMonitorWorker implements IRunnable { } } + private async publishOpsHeartbeat(partial: RelayMonitorOpsHeartbeatFields): Promise { + if (!this.opsCache) { + return + } + + this.lastOpsWorkerHeartbeat = { + ...this.lastOpsWorkerHeartbeat, + ...partial, + } + if ('lastError' in partial && partial.lastError === undefined) { + delete this.lastOpsWorkerHeartbeat.lastError + } + + await publishAdminOpsWorkerHeartbeatSafely(this.opsCache, { + type: 'relay-monitor', + updatedAt: new Date().toISOString(), + ...this.lastOpsWorkerHeartbeat, + }) + } + private async onSchedule(): Promise { + const lastRunStartedAt = new Date().toISOString() + await this.publishOpsHeartbeat({ lastRunStartedAt }) + const currentSettings = this.settings() if (!currentSettings.nip66?.enabled) { @@ -103,6 +158,11 @@ export class RelayMonitorWorker implements IRunnable { if (valid.length === 0) { logger.warn('no valid probe targets configured; skipping probe run') + await this.publishOpsHeartbeat({ + lastRunStartedAt, + lastRunFinishedAt: new Date().toISOString(), + lastError: 'No valid probe targets configured', + }) return } @@ -119,6 +179,11 @@ export class RelayMonitorWorker implements IRunnable { if (results.length === 0) { logger.warn('probe run produced no results') + await this.publishOpsHeartbeat({ + lastRunStartedAt, + lastRunFinishedAt: new Date().toISOString(), + lastError: 'Probe run produced no results', + }) return } @@ -154,6 +219,12 @@ export class RelayMonitorWorker implements IRunnable { logger.error('failed to publish NIP-66 events: %o', error) } } + + await this.publishOpsHeartbeat({ + lastRunStartedAt, + lastRunFinishedAt: new Date().toISOString(), + lastError: undefined, + }) } private onError(error: Error) { @@ -173,6 +244,7 @@ export class RelayMonitorWorker implements IRunnable { public close(callback?: () => void) { logger('closing') clearInterval(this.interval) + clearInterval(this.opsKeepaliveInterval) if (typeof callback === 'function') { callback() } diff --git a/src/app/static-mirroring-worker.ts b/src/app/static-mirroring-worker.ts index d725e404d..0e5fac5b3 100644 --- a/src/app/static-mirroring-worker.ts +++ b/src/app/static-mirroring-worker.ts @@ -1,10 +1,23 @@ -import { anyPass, map, mergeDeepRight, path } from 'ramda' -import { RawData, WebSocket } from 'ws' import cluster from 'cluster' import { randomUUID } from 'crypto' - -import { createRelayedEventMessage, createSubscriptionMessage } from '../utils/messages' +import { anyPass, map, mergeDeepRight, path } from 'ramda' +import { RawData, WebSocket } from 'ws' +import { ICacheAdapter } from '../@types/adapters' +import { AdminOpsConnectionState } from '../@types/admin-ops' +import { IRunnable } from '../@types/base' +import { Event, RelayedEvent } from '../@types/event' +import { OutgoingEventMessage } from '../@types/messages' +import { IEventRepository, IUserRepository } from '../@types/repositories' import { EventLimits, FeeSchedule, Mirror, Settings } from '../@types/settings' +import { WebSocketServerAdapterEvent } from '../constants/adapter' +import { EventExpirationTimeMetadataKey } from '../constants/base' +import { createLogger } from '../factories/logger-factory' +import { shutdownMetricsTelemetry } from '../telemetry/metrics' +import { + ADMIN_OPS_HEARTBEAT_KEEPALIVE_MS, + publishAdminOpsMirrorHeartbeatSafely, + publishAdminOpsWorkerHeartbeatSafely, +} from '../utils/admin-ops-store' import { getEventExpiration, getEventProofOfWork, @@ -20,28 +33,28 @@ import { isFileMessageEvent, isSealEvent, } from '../utils/event' -import { IEventRepository, IUserRepository } from '../@types/repositories' -import { createLogger } from '../factories/logger-factory' -import { Event } from '../@types/event' -import { EventExpirationTimeMetadataKey } from '../constants/base' -import { IRunnable } from '../@types/base' -import { OutgoingEventMessage } from '../@types/messages' -import { RelayedEvent } from '../@types/event' -import { WebSocketServerAdapterEvent } from '../constants/adapter' -import { shutdownMetricsTelemetry } from '../telemetry/metrics' +import { createRelayedEventMessage, createSubscriptionMessage } from '../utils/messages' const logger = createLogger('static-mirror-worker') export class StaticMirroringWorker implements IRunnable { private client: WebSocket | undefined + private opsKeepaliveInterval: NodeJS.Timeout | undefined private config: Mirror + private connectionState: AdminOpsConnectionState = 'connecting' + private reconnectCount = 0 + private lastError: string | undefined + private lastEventAt: string | undefined + private readonly mirrorIndex: string public constructor( private readonly eventRepository: IEventRepository, private readonly userRepository: IUserRepository, private readonly process: NodeJS.Process, private readonly settings: () => Settings, + private readonly opsCache?: ICacheAdapter, ) { + this.mirrorIndex = process.env.MIRROR_INDEX ?? '0' this.process .on('message', this.onMessage.bind(this)) .on('SIGINT', this.onExit.bind(this)) @@ -51,6 +64,29 @@ export class StaticMirroringWorker implements IRunnable { .on('unhandledRejection', this.onError.bind(this)) } + private async publishOpsHeartbeats(): Promise { + if (!this.opsCache || !this.config) { + return + } + + const updatedAt = new Date().toISOString() + await publishAdminOpsMirrorHeartbeatSafely(this.opsCache, { + index: this.mirrorIndex, + targetUrl: this.config.address, + connectionState: this.connectionState, + reconnectCount: this.reconnectCount, + lastError: this.lastError, + lastEventAt: this.lastEventAt, + updatedAt, + }) + await publishAdminOpsWorkerHeartbeatSafely(this.opsCache, { + type: 'static-mirroring', + index: this.mirrorIndex, + lastRunFinishedAt: updatedAt, + updatedAt, + }) + } + public run(): void { const currentSettings = this.settings() @@ -60,21 +96,35 @@ export class StaticMirroringWorker implements IRunnable { let since = Math.floor(Date.now() / 1000) - 60 * 10 + void this.publishOpsHeartbeats() + + if (this.opsCache) { + this.opsKeepaliveInterval = setInterval(() => { + void this.publishOpsHeartbeats() + }, ADMIN_OPS_HEARTBEAT_KEEPALIVE_MS) + } + const createMirror = (config: Mirror) => { const subscriptionId = `mirror-${randomUUID()}` logger('connecting to %s', config.address) + this.connectionState = 'connecting' + void this.publishOpsHeartbeats() - return new WebSocket(config.address, { timeout: 5000 }) - .on('open', function () { + const client = new WebSocket(config.address, { timeout: 5000 }) + + client + .on('open', () => { logger('connected to %s', config.address) + this.connectionState = 'open' + void this.publishOpsHeartbeats() if (Array.isArray(config.filters) && config.filters?.length) { const filters = config.filters.map((filter) => ({ ...filter, since })) logger('subscribing with %s: %o', subscriptionId, filters) - this.send(JSON.stringify(createSubscriptionMessage(subscriptionId, filters))) + client.send(JSON.stringify(createSubscriptionMessage(subscriptionId, filters))) } }) .on('message', async (raw: RawData) => { @@ -126,6 +176,8 @@ export class StaticMirroringWorker implements IRunnable { } since = Math.floor(Date.now() / 1000) - 30 + this.lastEventAt = new Date().toISOString() + void this.publishOpsHeartbeats() logger('%s >> local: %s', config.address, event.id) @@ -144,15 +196,22 @@ export class StaticMirroringWorker implements IRunnable { }) .on('close', (code, reason) => { logger(`disconnected (${code}): ${reason.toString()}`) + this.connectionState = 'closed' + this.reconnectCount += 1 + void this.publishOpsHeartbeats() setTimeout(() => { this.client.removeAllListeners() this.client = createMirror(config) }, 5000) }) - .on('error', function (error) { + .on('error', (error) => { logger('connection error: %o', error) + this.lastError = error instanceof Error ? error.message : String(error) + void this.publishOpsHeartbeats() }) + + return client } this.client = createMirror(this.config) @@ -347,6 +406,7 @@ export class StaticMirroringWorker implements IRunnable { public close(callback?: () => void) { logger('closing') + clearInterval(this.opsKeepaliveInterval) if (this.client) { this.client.terminate() } diff --git a/src/controllers/admin/get-ops-mirrors-controller.ts b/src/controllers/admin/get-ops-mirrors-controller.ts new file mode 100644 index 000000000..be81339c7 --- /dev/null +++ b/src/controllers/admin/get-ops-mirrors-controller.ts @@ -0,0 +1,21 @@ +import { Request, Response } from 'express' + +import { IController } from '../../@types/controllers' +import { Settings } from '../../@types/settings' +import { AdminOpsStore } from '../../utils/admin-ops-store' +import { listConfiguredMirrors, mergeMirrorRows } from '../../utils/admin-ops-config' +import { loadMergedSettings } from '../../utils/settings-config' + +export class GetAdminOpsMirrorsController implements IController { + public constructor( + private readonly opsStore: AdminOpsStore, + private readonly getSettings: () => Settings = loadMergedSettings, + ) {} + + public async handleRequest(_request: Request, response: Response): Promise { + const [heartbeats, settings] = await Promise.all([this.opsStore.listMirrorHeartbeats(), this.getSettings()]) + const mirrors = mergeMirrorRows(listConfiguredMirrors(settings), heartbeats) + + response.status(200).setHeader('content-type', 'application/json').send({ mirrors }) + } +} diff --git a/src/controllers/admin/get-ops-nip05-verifications-controller.ts b/src/controllers/admin/get-ops-nip05-verifications-controller.ts new file mode 100644 index 000000000..a273ff69c --- /dev/null +++ b/src/controllers/admin/get-ops-nip05-verifications-controller.ts @@ -0,0 +1,71 @@ +import { Request, Response } from 'express' + +import { IController } from '../../@types/controllers' +import { INip05VerificationRepository } from '../../@types/repositories' +import { Settings } from '../../@types/settings' +import { getNip05VerifyExpirationMs, toAdminNip05Row } from '../../utils/admin-ops-nip05' +import { loadMergedSettings } from '../../utils/settings-config' + +const DEFAULT_LIMIT = 50 +const MAX_LIMIT = 200 + +const parseBoolean = (value: unknown): boolean | undefined => { + if (value === undefined || value === '') { + return undefined + } + if (value === 'true' || value === true) { + return true + } + if (value === 'false' || value === false) { + return false + } + return undefined +} + +export class GetAdminOpsNip05VerificationsController implements IController { + public constructor( + private readonly nip05VerificationRepository: INip05VerificationRepository, + private readonly getSettings: () => Settings = loadMergedSettings, + ) {} + + public async handleRequest(request: Request, response: Response): Promise { + const settings = this.getSettings() + const verifyExpirationMs = getNip05VerifyExpirationMs(settings) + + const limitRaw = Number(request.query.limit ?? DEFAULT_LIMIT) + const offsetRaw = Number(request.query.offset ?? 0) + const limit = Number.isFinite(limitRaw) ? Math.min(Math.max(1, Math.floor(limitRaw)), MAX_LIMIT) : DEFAULT_LIMIT + const offset = Number.isFinite(offsetRaw) ? Math.max(0, Math.floor(offsetRaw)) : 0 + + const domain = typeof request.query.domain === 'string' ? request.query.domain.trim() : undefined + const verified = parseBoolean(request.query.verified) + const stale = parseBoolean(request.query.stale) + + const minFailuresRaw = request.query.minFailures ?? request.query.min_failures + const minFailures = + minFailuresRaw === undefined || minFailuresRaw === '' + ? undefined + : Number(minFailuresRaw) + + if (minFailures !== undefined && (!Number.isInteger(minFailures) || minFailures < 0)) { + response.status(400).setHeader('content-type', 'application/json').send({ error: 'minFailures must be a non-negative integer' }) + return + } + + const { rows, total } = await this.nip05VerificationRepository.listForAdmin({ + domain, + verified, + stale, + minFailures, + verifyExpirationMs, + limit, + offset, + }) + + response.status(200).setHeader('content-type', 'application/json').send({ + verifications: rows.map((row) => toAdminNip05Row(row, verifyExpirationMs)), + pagination: { limit, offset, total }, + verifyExpirationMs, + }) + } +} diff --git a/src/controllers/admin/get-ops-storage-counts-controller.ts b/src/controllers/admin/get-ops-storage-counts-controller.ts new file mode 100644 index 000000000..febadecf0 --- /dev/null +++ b/src/controllers/admin/get-ops-storage-counts-controller.ts @@ -0,0 +1,60 @@ +import { Request, Response } from 'express' + +import { IController } from '../../@types/controllers' +import { IEventRepository } from '../../@types/repositories' +import { createLogger } from '../../factories/logger-factory' +import { + ADMIN_OPS_STORAGE_QUERY_TIMEOUT_MS, + buildStorageCountFilters, + parseAdminOpsStorageQuery, +} from '../../utils/admin-ops-storage' + +const logger = createLogger('get-admin-ops-storage-counts-controller') + +const isStorageCountQueryTimeout = (error: unknown): boolean => + error instanceof Error && /query timeout|timed out/i.test(error.message) + +export class GetAdminOpsStorageCountsController implements IController { + public constructor(private readonly eventRepository: IEventRepository) {} + + public async handleRequest(request: Request, response: Response): Promise { + const parsed = parseAdminOpsStorageQuery(request.query.kinds, request.query.since, request.query.until) + if (parsed.error || !parsed.query) { + response + .status(400) + .setHeader('content-type', 'application/json') + .send({ error: parsed.error ?? 'Invalid query' }) + return + } + + const filters = buildStorageCountFilters(parsed.query) + + try { + const counts = await Promise.all( + filters.map(async (filter) => ({ + kinds: filter.kinds ?? [], + count: await this.eventRepository.countByFilters([filter], { + timeoutMs: ADMIN_OPS_STORAGE_QUERY_TIMEOUT_MS, + }), + })), + ) + + response.status(200).setHeader('content-type', 'application/json').send({ + since: parsed.query.since, + until: parsed.query.until, + counts, + }) + } catch (error) { + if (isStorageCountQueryTimeout(error)) { + response + .status(504) + .setHeader('content-type', 'application/json') + .send({ error: 'Storage count query timed out' }) + return + } + + logger.error('storage count query failed: %o', error) + response.status(500).setHeader('content-type', 'application/json').send({ error: 'Storage count query failed' }) + } + } +} diff --git a/src/controllers/admin/get-ops-workers-controller.ts b/src/controllers/admin/get-ops-workers-controller.ts new file mode 100644 index 000000000..f513bfd60 --- /dev/null +++ b/src/controllers/admin/get-ops-workers-controller.ts @@ -0,0 +1,21 @@ +import { Request, Response } from 'express' + +import { IController } from '../../@types/controllers' +import { Settings } from '../../@types/settings' +import { AdminOpsStore } from '../../utils/admin-ops-store' +import { listConfiguredWorkers, mergeWorkerRows } from '../../utils/admin-ops-config' +import { loadMergedSettings } from '../../utils/settings-config' + +export class GetAdminOpsWorkersController implements IController { + public constructor( + private readonly opsStore: AdminOpsStore, + private readonly getSettings: () => Settings = loadMergedSettings, + ) {} + + public async handleRequest(_request: Request, response: Response): Promise { + const [heartbeats, settings] = await Promise.all([this.opsStore.listWorkerHeartbeats(), this.getSettings()]) + const workers = mergeWorkerRows(listConfiguredWorkers(settings), heartbeats) + + response.status(200).setHeader('content-type', 'application/json').send({ workers }) + } +} diff --git a/src/controllers/admin/post-ops-nip05-reverify-controller.ts b/src/controllers/admin/post-ops-nip05-reverify-controller.ts new file mode 100644 index 000000000..ccc332a78 --- /dev/null +++ b/src/controllers/admin/post-ops-nip05-reverify-controller.ts @@ -0,0 +1,47 @@ +import { Request, Response } from 'express' + +import { IController } from '../../@types/controllers' +import { INip05VerificationRepository } from '../../@types/repositories' +import { Settings } from '../../@types/settings' +import { getNip05VerifyExpirationMs, toAdminNip05Row } from '../../utils/admin-ops-nip05' +import { reverifyNip05Verification } from '../../utils/nip05-reverification' +import { loadMergedSettings } from '../../utils/settings-config' +import { pubkeySchema } from '../../schemas/base-schema' + +export class PostAdminOpsNip05ReverifyController implements IController { + public constructor( + private readonly nip05VerificationRepository: INip05VerificationRepository, + private readonly getSettings: () => Settings = loadMergedSettings, + ) {} + + public async handleRequest(request: Request, response: Response): Promise { + const pubkeyResult = pubkeySchema.safeParse(request.params.pubkey) + if (!pubkeyResult.success) { + response.status(400).setHeader('content-type', 'application/json').send({ error: 'Invalid pubkey' }) + return + } + + const settings = this.getSettings() + if (!settings.nip05 || settings.nip05.mode === 'disabled') { + response.status(409).setHeader('content-type', 'application/json').send({ error: 'NIP-05 verification is disabled' }) + return + } + + const existing = await this.nip05VerificationRepository.findByPubkey(pubkeyResult.data) + if (!existing) { + response.status(404).setHeader('content-type', 'application/json').send({ error: 'Verification record not found' }) + return + } + + try { + const updated = await reverifyNip05Verification(existing, this.nip05VerificationRepository) + const verifyExpirationMs = getNip05VerifyExpirationMs(settings) + response + .status(200) + .setHeader('content-type', 'application/json') + .send({ verification: toAdminNip05Row(updated, verifyExpirationMs) }) + } catch { + response.status(502).setHeader('content-type', 'application/json').send({ error: 'Re-verification failed' }) + } + } +} diff --git a/src/factories/admin-ops-cache-adapter-factory.ts b/src/factories/admin-ops-cache-adapter-factory.ts new file mode 100644 index 000000000..72cfefbb0 --- /dev/null +++ b/src/factories/admin-ops-cache-adapter-factory.ts @@ -0,0 +1,11 @@ +import { RedisAdapter } from '../adapters/redis-adapter' +import { ICacheAdapter } from '../@types/adapters' +import { getCacheClient } from '../cache/client' + +export const createAdminOpsCacheAdapter = (): ICacheAdapter | undefined => { + try { + return new RedisAdapter(getCacheClient()) + } catch { + return undefined + } +} diff --git a/src/factories/controllers/get-admin-ops-mirrors-controller-factory.ts b/src/factories/controllers/get-admin-ops-mirrors-controller-factory.ts new file mode 100644 index 000000000..04af6ced3 --- /dev/null +++ b/src/factories/controllers/get-admin-ops-mirrors-controller-factory.ts @@ -0,0 +1,10 @@ +import { RedisAdapter } from '../../adapters/redis-adapter' +import { IController } from '../../@types/controllers' +import { getCacheClient } from '../../cache/client' +import { GetAdminOpsMirrorsController } from '../../controllers/admin/get-ops-mirrors-controller' +import { AdminOpsStore } from '../../utils/admin-ops-store' + +export const createGetAdminOpsMirrorsController = (): IController => { + const opsStore = new AdminOpsStore(new RedisAdapter(getCacheClient())) + return new GetAdminOpsMirrorsController(opsStore) +} diff --git a/src/factories/controllers/get-admin-ops-nip05-verifications-controller-factory.ts b/src/factories/controllers/get-admin-ops-nip05-verifications-controller-factory.ts new file mode 100644 index 000000000..d2b80c08f --- /dev/null +++ b/src/factories/controllers/get-admin-ops-nip05-verifications-controller-factory.ts @@ -0,0 +1,8 @@ +import { IController } from '../../@types/controllers' +import { getMasterDbClient } from '../../database/client' +import { GetAdminOpsNip05VerificationsController } from '../../controllers/admin/get-ops-nip05-verifications-controller' +import { Nip05VerificationRepository } from '../../repositories/nip05-verification-repository' + +export const createGetAdminOpsNip05VerificationsController = (): IController => { + return new GetAdminOpsNip05VerificationsController(new Nip05VerificationRepository(getMasterDbClient())) +} diff --git a/src/factories/controllers/get-admin-ops-storage-counts-controller-factory.ts b/src/factories/controllers/get-admin-ops-storage-counts-controller-factory.ts new file mode 100644 index 000000000..2955a0e97 --- /dev/null +++ b/src/factories/controllers/get-admin-ops-storage-counts-controller-factory.ts @@ -0,0 +1,9 @@ +import { IController } from '../../@types/controllers' +import { getMasterDbClient, getReadReplicaDbClient } from '../../database/client' +import { GetAdminOpsStorageCountsController } from '../../controllers/admin/get-ops-storage-counts-controller' +import { createEventRepository } from '../event-repository-factory' + +export const createGetAdminOpsStorageCountsController = (): IController => { + const eventRepository = createEventRepository(getMasterDbClient(), getReadReplicaDbClient()) + return new GetAdminOpsStorageCountsController(eventRepository) +} diff --git a/src/factories/controllers/get-admin-ops-workers-controller-factory.ts b/src/factories/controllers/get-admin-ops-workers-controller-factory.ts new file mode 100644 index 000000000..c34e0e25b --- /dev/null +++ b/src/factories/controllers/get-admin-ops-workers-controller-factory.ts @@ -0,0 +1,10 @@ +import { RedisAdapter } from '../../adapters/redis-adapter' +import { IController } from '../../@types/controllers' +import { getCacheClient } from '../../cache/client' +import { GetAdminOpsWorkersController } from '../../controllers/admin/get-ops-workers-controller' +import { AdminOpsStore } from '../../utils/admin-ops-store' + +export const createGetAdminOpsWorkersController = (): IController => { + const opsStore = new AdminOpsStore(new RedisAdapter(getCacheClient())) + return new GetAdminOpsWorkersController(opsStore) +} diff --git a/src/factories/controllers/post-admin-ops-nip05-reverify-controller-factory.ts b/src/factories/controllers/post-admin-ops-nip05-reverify-controller-factory.ts new file mode 100644 index 000000000..fe3343c7a --- /dev/null +++ b/src/factories/controllers/post-admin-ops-nip05-reverify-controller-factory.ts @@ -0,0 +1,8 @@ +import { IController } from '../../@types/controllers' +import { getMasterDbClient } from '../../database/client' +import { PostAdminOpsNip05ReverifyController } from '../../controllers/admin/post-ops-nip05-reverify-controller' +import { Nip05VerificationRepository } from '../../repositories/nip05-verification-repository' + +export const createPostAdminOpsNip05ReverifyController = (): IController => { + return new PostAdminOpsNip05ReverifyController(new Nip05VerificationRepository(getMasterDbClient())) +} diff --git a/src/factories/maintenance-worker-factory.ts b/src/factories/maintenance-worker-factory.ts index 27fd53115..5e280d8a5 100644 --- a/src/factories/maintenance-worker-factory.ts +++ b/src/factories/maintenance-worker-factory.ts @@ -1,3 +1,4 @@ +import { createAdminOpsCacheAdapter } from './admin-ops-cache-adapter-factory' import { createMaintenanceService } from './maintenance-service-factory' import { createNotificationOutboxService } from './notification-outbox-service-factory' import { createPaymentsService } from './payments-service-factory' @@ -22,5 +23,6 @@ export const maintenanceWorkerFactory = () => { createNotificationOutboxService(), notificationDeliveryLogRepository, notificationOutboxRepository, + createAdminOpsCacheAdapter(), ) } diff --git a/src/factories/relay-monitor-worker-factory.ts b/src/factories/relay-monitor-worker-factory.ts index 0dae13289..8f7d8d5f8 100644 --- a/src/factories/relay-monitor-worker-factory.ts +++ b/src/factories/relay-monitor-worker-factory.ts @@ -2,6 +2,7 @@ import { RedisAdapter } from '../adapters/redis-adapter' import { RelayMonitorWorker } from '../app/relay-monitor-worker' import { getCacheClient } from '../cache/client' import { getMasterDbClient, getReadReplicaDbClient } from '../database/client' +import { createAdminOpsCacheAdapter } from './admin-ops-cache-adapter-factory' import { createEventRepository } from './event-repository-factory' import { createSettings } from './settings-factory' import { Nip66EventPublisher } from '../services/nip66-event-publisher' @@ -16,5 +17,13 @@ export const relayMonitorWorkerFactory = () => { const eventRepository = createEventRepository(getMasterDbClient(), getReadReplicaDbClient()) const eventPublisher = new Nip66EventPublisher(eventRepository, cache) - return new RelayMonitorWorker(process, createSettings, snapshotStore, runProbe, eventPublisher, historyStore) + return new RelayMonitorWorker( + process, + createSettings, + snapshotStore, + runProbe, + eventPublisher, + historyStore, + createAdminOpsCacheAdapter(), + ) } diff --git a/src/factories/static-mirroring.worker-factory.ts b/src/factories/static-mirroring.worker-factory.ts index 5516848c5..e2420ac0f 100644 --- a/src/factories/static-mirroring.worker-factory.ts +++ b/src/factories/static-mirroring.worker-factory.ts @@ -1,4 +1,5 @@ import { getMasterDbClient, getReadReplicaDbClient } from '../database/client' +import { createAdminOpsCacheAdapter } from './admin-ops-cache-adapter-factory' import { createEventRepository } from './event-repository-factory' import { createSettings } from './settings-factory' import { StaticMirroringWorker } from '../app/static-mirroring-worker' @@ -9,6 +10,11 @@ export const staticMirroringWorkerFactory = () => { const readReplicaDbClient = getReadReplicaDbClient() const eventRepository = createEventRepository(dbClient, readReplicaDbClient) const userRepository = new UserRepository(dbClient, eventRepository) - - return new StaticMirroringWorker(eventRepository, userRepository, process, createSettings) + return new StaticMirroringWorker( + eventRepository, + userRepository, + process, + createSettings, + createAdminOpsCacheAdapter(), + ) } diff --git a/src/repositories/event-repository.ts b/src/repositories/event-repository.ts index 26431db60..f95de9445 100644 --- a/src/repositories/event-repository.ts +++ b/src/repositories/event-repository.ts @@ -101,7 +101,10 @@ export class EventRepository implements IEventRepository { return query } - public async countByFilters(filters: SubscriptionFilter[]): Promise { + public async countByFilters( + filters: SubscriptionFilter[], + options?: { timeoutMs?: number }, + ): Promise { logger('counting events for %o', filters) if (!Array.isArray(filters) || !filters.length) { @@ -129,7 +132,11 @@ export class EventRepository implements IEventRepository { const query = queries.length === 1 ? queries[0] : this.unionFilterQueries(queries) // Rows are already unique: EXISTS never repeats an event and UNION drops overlaps between filters. - const result = await this.readReplicaDbClient.from(query.as('matching_events')).count({ count: '*' }).first() + const countQuery = this.readReplicaDbClient.from(query.as('matching_events')).count({ count: '*' }) + if (typeof options?.timeoutMs === 'number') { + countQuery.timeout(options.timeoutMs, { cancel: true }) + } + const result = await countQuery.first() return Number(result?.count ?? 0) } diff --git a/src/repositories/nip05-verification-repository.ts b/src/repositories/nip05-verification-repository.ts index bf2665589..bafa2d92e 100644 --- a/src/repositories/nip05-verification-repository.ts +++ b/src/repositories/nip05-verification-repository.ts @@ -4,7 +4,7 @@ import { DatabaseClient, Pubkey } from '../@types/base' import { DBNip05Verification, Nip05Verification } from '../@types/nip05' import { fromBuffer, toBuffer } from '../utils/transform' import { createLogger } from '../factories/logger-factory' -import { INip05VerificationRepository } from '../@types/repositories' +import { INip05VerificationRepository, Nip05VerificationAdminListQuery } from '../@types/repositories' const logger = createLogger('nip05-verification-repository') @@ -97,4 +97,44 @@ export class Nip05VerificationRepository implements INip05VerificationRepository return this.dbClient('nip05_verifications').where('pubkey', toBuffer(pubkey)).delete() } + + public async listForAdmin(query: Nip05VerificationAdminListQuery): Promise<{ rows: Nip05Verification[]; total: number }> { + let builder = this.dbClient('nip05_verifications') + + if (query.domain) { + builder = builder.where('domain', query.domain) + } + + if (query.verified !== undefined) { + builder = builder.where('is_verified', query.verified) + } + + if (query.minFailures !== undefined) { + builder = builder.where('failure_count', '>=', query.minFailures) + } + + if (query.stale !== undefined) { + const cutoff = new Date(Date.now() - query.verifyExpirationMs) + if (query.stale) { + builder = builder.where((clause) => { + clause + .where('is_verified', false) + .orWhereNull('last_verified_at') + .orWhere('last_verified_at', '<', cutoff) + }) + } else { + builder = builder.where('is_verified', true).andWhere('last_verified_at', '>=', cutoff) + } + } + + const countRow = await builder.clone().count<{ count: string }>({ count: '*' }).first() + const total = Number(countRow?.count ?? 0) + + const rows = await builder + .orderBy('last_checked_at', 'desc') + .limit(query.limit) + .offset(query.offset) + + return { rows: rows.map(fromDBNip05Verification), total } + } } diff --git a/src/routes/admin/index.ts b/src/routes/admin/index.ts index 3f2740f27..0449675e5 100644 --- a/src/routes/admin/index.ts +++ b/src/routes/admin/index.ts @@ -17,6 +17,11 @@ import { createPostAdminLoginController } from '../../factories/controllers/post import { createPostAdminLogoutController } from '../../factories/controllers/post-admin-logout-controller-factory' import { createPostAdminSettingsRestoreController } from '../../factories/controllers/post-admin-settings-restore-controller-factory' import { createPostAdminSettingsValidateController } from '../../factories/controllers/post-admin-settings-validate-controller-factory' +import { createGetAdminOpsMirrorsController } from '../../factories/controllers/get-admin-ops-mirrors-controller-factory' +import { createGetAdminOpsWorkersController } from '../../factories/controllers/get-admin-ops-workers-controller-factory' +import { createGetAdminOpsNip05VerificationsController } from '../../factories/controllers/get-admin-ops-nip05-verifications-controller-factory' +import { createGetAdminOpsStorageCountsController } from '../../factories/controllers/get-admin-ops-storage-counts-controller-factory' +import { createPostAdminOpsNip05ReverifyController } from '../../factories/controllers/post-admin-ops-nip05-reverify-controller-factory' import { adminAuthGateMiddleware, adminAuthMiddleware } from '../../handlers/request-handlers/admin-auth-middleware' import { adminEnabledMiddleware } from '../../handlers/request-handlers/admin-enabled-middleware' import { adminJsonBodyMiddleware } from '../../handlers/request-handlers/admin-json-body-middleware' @@ -143,5 +148,35 @@ router.post( adminAuthMiddleware, withAdminController(createPostAdminNotificationTestController), ) +router.get( + '/ops/mirrors', + adminRateLimitMiddleware, + adminAuthMiddleware, + withAdminController(createGetAdminOpsMirrorsController), +) +router.get( + '/ops/workers', + adminRateLimitMiddleware, + adminAuthMiddleware, + withAdminController(createGetAdminOpsWorkersController), +) +router.get( + '/ops/nip05-verifications', + adminRateLimitMiddleware, + adminAuthMiddleware, + withAdminController(createGetAdminOpsNip05VerificationsController), +) +router.post( + '/ops/nip05-verifications/:pubkey/reverify', + adminRateLimitMiddleware, + adminAuthMiddleware, + withAdminController(createPostAdminOpsNip05ReverifyController), +) +router.get( + '/ops/storage/counts', + adminRateLimitMiddleware, + adminAuthMiddleware, + withAdminController(createGetAdminOpsStorageCountsController), +) export default router diff --git a/src/utils/admin-ops-config.ts b/src/utils/admin-ops-config.ts new file mode 100644 index 000000000..4e5d645db --- /dev/null +++ b/src/utils/admin-ops-config.ts @@ -0,0 +1,122 @@ +import { cpus } from 'os' + +import { + AdminOpsConfiguredMirror, + AdminOpsConfiguredWorker, + AdminOpsMirrorHeartbeat, + AdminOpsMirrorRow, + AdminOpsWorkerHeartbeat, + AdminOpsWorkerRow, +} from '../@types/admin-ops' +import { Settings } from '../@types/settings' +import { ADMIN_OPS_HEARTBEAT_TTL_SECONDS, isAdminOpsHeartbeatStale } from './admin-ops-store' + +const UNMONITORED_WORKER_TYPES = new Set(['worker', 'dvm-orchestrator']) + +export const isAdminOpsWorkerMonitored = (type: string): boolean => !UNMONITORED_WORKER_TYPES.has(type) + +export const listConfiguredMirrors = (settings: Settings): AdminOpsConfiguredMirror[] => { + const mirrors = settings.mirroring?.static + if (!Array.isArray(mirrors) || !mirrors.length) { + return [] + } + + return mirrors.map((mirror, index) => ({ + index: index.toString(), + targetUrl: mirror.address, + enabled: true, + })) +} + +export const listConfiguredWorkers = (settings: Settings): AdminOpsConfiguredWorker[] => { + const workers: AdminOpsConfiguredWorker[] = [] + + const workerCount = process.env.WORKER_COUNT + ? Number(process.env.WORKER_COUNT) + : settings.workers?.count || cpus().length + + for (let i = 0; i < workerCount; i++) { + workers.push({ + type: 'worker', + index: i.toString(), + enabled: true, + label: `Client worker ${i}`, + }) + } + + workers.push({ + type: 'maintenance', + enabled: true, + label: 'Maintenance', + }) + + const mirrors = listConfiguredMirrors(settings) + for (const mirror of mirrors) { + workers.push({ + type: 'static-mirroring', + index: mirror.index, + enabled: mirror.enabled, + label: `Static mirror ${mirror.index}`, + }) + } + + if (settings.nip66?.enabled) { + workers.push({ + type: 'relay-monitor', + enabled: true, + label: 'Relay monitor (NIP-66)', + }) + } + + const dvmWorkers = settings.dvm?.workers + if (Array.isArray(dvmWorkers) && dvmWorkers.length) { + for (let i = 0; i < dvmWorkers.length; i++) { + workers.push({ + type: 'dvm-orchestrator', + index: i.toString(), + enabled: true, + label: `DVM orchestrator ${i}`, + }) + } + } + + return workers +} + +const heartbeatKey = (type: string, index?: string) => + typeof index === 'string' && index.length > 0 ? `${type}:${index}` : type + +export const mergeMirrorRows = ( + configured: AdminOpsConfiguredMirror[], + heartbeats: AdminOpsMirrorHeartbeat[], +): AdminOpsMirrorRow[] => { + const byIndex = new Map(heartbeats.map((heartbeat) => [heartbeat.index, heartbeat])) + + return configured.map((mirror) => { + const heartbeat = byIndex.get(mirror.index) + return { + ...mirror, + heartbeat, + stale: !heartbeat || isAdminOpsHeartbeatStale(heartbeat.updatedAt, ADMIN_OPS_HEARTBEAT_TTL_SECONDS), + } + }) +} + +export const mergeWorkerRows = ( + configured: AdminOpsConfiguredWorker[], + heartbeats: AdminOpsWorkerHeartbeat[], +): AdminOpsWorkerRow[] => { + const byKey = new Map(heartbeats.map((heartbeat) => [heartbeatKey(heartbeat.type, heartbeat.index), heartbeat])) + + return configured.map((worker) => { + const heartbeat = byKey.get(heartbeatKey(worker.type, worker.index)) + const monitored = isAdminOpsWorkerMonitored(worker.type) + return { + ...worker, + heartbeat, + monitored, + stale: + monitored && (!heartbeat || isAdminOpsHeartbeatStale(heartbeat.updatedAt, ADMIN_OPS_HEARTBEAT_TTL_SECONDS)), + } + }) +} diff --git a/src/utils/admin-ops-nip05.ts b/src/utils/admin-ops-nip05.ts new file mode 100644 index 000000000..a92b347c3 --- /dev/null +++ b/src/utils/admin-ops-nip05.ts @@ -0,0 +1,32 @@ +import { Nip05Verification } from '../@types/nip05' +import { AdminNip05VerificationAdminRow } from '../@types/admin-ops' +import { DEFAULT_NIP05_VERIFY_EXPIRATION_MS } from './nip05' +import { Settings } from '../@types/settings' + +export const getNip05VerifyExpirationMs = (settings: Settings): number => + settings.nip05?.verifyExpiration ?? DEFAULT_NIP05_VERIFY_EXPIRATION_MS + +export const isNip05VerificationStale = ( + verification: Pick, + verifyExpirationMs: number, + now = Date.now(), +): boolean => { + if (!verification.isVerified || !verification.lastVerifiedAt) { + return true + } + return now - verification.lastVerifiedAt.getTime() > verifyExpirationMs +} + +export const toAdminNip05Row = ( + verification: Nip05Verification, + verifyExpirationMs: number, +): AdminNip05VerificationAdminRow => ({ + pubkey: verification.pubkey, + nip05: verification.nip05, + domain: verification.domain, + isVerified: verification.isVerified, + failureCount: verification.failureCount, + lastVerifiedAt: verification.lastVerifiedAt ? verification.lastVerifiedAt.toISOString() : null, + lastCheckedAt: verification.lastCheckedAt.toISOString(), + stale: isNip05VerificationStale(verification, verifyExpirationMs), +}) diff --git a/src/utils/admin-ops-sanitize.ts b/src/utils/admin-ops-sanitize.ts new file mode 100644 index 000000000..8a6b9e45b --- /dev/null +++ b/src/utils/admin-ops-sanitize.ts @@ -0,0 +1,26 @@ +const MAX_OPS_ERROR_LENGTH = 240 + +export const sanitizeAdminOpsError = (error: unknown): string | undefined => { + if (error === undefined || error === null) { + return undefined + } + + const message = + error instanceof Error + ? error.message + : typeof error === 'string' + ? error + : JSON.stringify(error) + + const redacted = message + .replace(/(wss?|https?):\/\/[^\s]+/gi, '[url]') + .replace(/Bearer\s+\S+/gi, 'Bearer [redacted]') + .replace(/\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b/g, '[email]') + .trim() + + if (!redacted) { + return undefined + } + + return redacted.length > MAX_OPS_ERROR_LENGTH ? `${redacted.slice(0, MAX_OPS_ERROR_LENGTH)}…` : redacted +} diff --git a/src/utils/admin-ops-storage.ts b/src/utils/admin-ops-storage.ts new file mode 100644 index 000000000..ca6c8dda4 --- /dev/null +++ b/src/utils/admin-ops-storage.ts @@ -0,0 +1,88 @@ +import { SubscriptionFilter } from '../@types/subscription' + +export const ADMIN_OPS_STORAGE_MAX_KINDS = 20 +export const ADMIN_OPS_STORAGE_MAX_WINDOW_SECONDS = 90 * 86_400 +export const ADMIN_OPS_STORAGE_QUERY_TIMEOUT_MS = 5000 + +export interface AdminOpsStorageQuery { + kinds: number[] + since?: number + until?: number +} + +export const parseAdminOpsStorageQuery = ( + rawKinds: unknown, + rawSince: unknown, + rawUntil: unknown, +): { query?: AdminOpsStorageQuery; error?: string } => { + if (typeof rawKinds !== 'string' || !rawKinds.trim()) { + return { error: 'kinds query parameter is required (comma-separated event kinds)' } + } + + const kinds = rawKinds + .split(',') + .map((part) => part.trim()) + .filter(Boolean) + .map((part) => Number(part)) + + if (!kinds.length || kinds.some((kind) => !Number.isInteger(kind) || kind < 0 || kind > 65535)) { + return { error: 'kinds must be a comma-separated list of integers between 0 and 65535' } + } + + if (kinds.length > ADMIN_OPS_STORAGE_MAX_KINDS) { + return { error: `at most ${ADMIN_OPS_STORAGE_MAX_KINDS} kinds are allowed per request` } + } + + const uniqueKinds = [...new Set(kinds)] + + let since: number | undefined + if (rawSince !== undefined && rawSince !== '') { + since = Number(rawSince) + if (!Number.isInteger(since) || since < 0) { + return { error: 'since must be a non-negative unix timestamp in seconds' } + } + } + + let until: number | undefined + if (rawUntil !== undefined && rawUntil !== '') { + until = Number(rawUntil) + if (!Number.isInteger(until) || until < 0) { + return { error: 'until must be a non-negative unix timestamp in seconds' } + } + } + + const now = Math.floor(Date.now() / 1000) + const effectiveUntil = until ?? now + const effectiveSince = since ?? effectiveUntil - 86_400 + + if (effectiveSince > effectiveUntil) { + return { error: 'since must be less than or equal to until' } + } + + if (effectiveUntil - effectiveSince > ADMIN_OPS_STORAGE_MAX_WINDOW_SECONDS) { + return { error: `time window must not exceed ${ADMIN_OPS_STORAGE_MAX_WINDOW_SECONDS} seconds` } + } + + return { + query: { + kinds: uniqueKinds, + since: effectiveSince, + until: effectiveUntil, + }, + } +} + +export const buildStorageCountFilters = (query: AdminOpsStorageQuery): SubscriptionFilter[] => { + const timeFilter: SubscriptionFilter = {} + if (typeof query.since === 'number') { + timeFilter.since = query.since + } + if (typeof query.until === 'number') { + timeFilter.until = query.until + } + + return query.kinds.map((kind) => ({ + ...timeFilter, + kinds: [kind], + })) +} diff --git a/src/utils/admin-ops-store.ts b/src/utils/admin-ops-store.ts new file mode 100644 index 000000000..00279c352 --- /dev/null +++ b/src/utils/admin-ops-store.ts @@ -0,0 +1,166 @@ +import { ICacheAdapter } from '../@types/adapters' +import { AdminOpsMirrorHeartbeat, AdminOpsWorkerHeartbeat } from '../@types/admin-ops' +import { createLogger } from '../factories/logger-factory' +import { sanitizeAdminOpsError } from './admin-ops-sanitize' + +const logger = createLogger('admin-ops-store') + +export const ADMIN_OPS_HEARTBEAT_INDEX_KEY = 'admin:ops:heartbeat:index' +export const ADMIN_OPS_HEARTBEAT_TTL_SECONDS = 120 +export const ADMIN_OPS_HEARTBEAT_KEEPALIVE_MS = 45_000 + +const mirrorKey = (index: string) => `admin:ops:mirror:${index}` +const workerKey = (type: string, index?: string) => + typeof index === 'string' && index.length > 0 ? `admin:ops:worker:${type}:${index}` : `admin:ops:worker:${type}` + +const indexMember = (category: 'mirror' | 'worker', id: string) => `${category}:${id}` + +export class AdminOpsStore { + public constructor(private readonly cache: ICacheAdapter) {} + + public async publishMirror(state: AdminOpsMirrorHeartbeat): Promise { + const key = mirrorKey(state.index) + const member = indexMember('mirror', state.index) + await this.cache.setKey(key, JSON.stringify(state), ADMIN_OPS_HEARTBEAT_TTL_SECONDS) + await this.cache.addToSet(ADMIN_OPS_HEARTBEAT_INDEX_KEY, [member]) + await this.cache.setKeyExpiry(ADMIN_OPS_HEARTBEAT_INDEX_KEY, ADMIN_OPS_HEARTBEAT_TTL_SECONDS * 4) + } + + public async publishWorker(state: AdminOpsWorkerHeartbeat): Promise { + const key = workerKey(state.type, state.index) + const member = indexMember('worker', typeof state.index === 'string' ? `${state.type}:${state.index}` : state.type) + await this.cache.setKey(key, JSON.stringify(state), ADMIN_OPS_HEARTBEAT_TTL_SECONDS) + await this.cache.addToSet(ADMIN_OPS_HEARTBEAT_INDEX_KEY, [member]) + await this.cache.setKeyExpiry(ADMIN_OPS_HEARTBEAT_INDEX_KEY, ADMIN_OPS_HEARTBEAT_TTL_SECONDS * 4) + } + + public async listMirrorHeartbeats(): Promise { + const members = await this.cache.getSetMembers(ADMIN_OPS_HEARTBEAT_INDEX_KEY) + const mirrorMembers = members.filter((member) => member.startsWith('mirror:')) + + const heartbeats: AdminOpsMirrorHeartbeat[] = [] + for (const member of mirrorMembers) { + const index = member.slice('mirror:'.length) + const raw = await this.cache.getKey(mirrorKey(index)) + if (!raw) { + continue + } + try { + heartbeats.push(JSON.parse(raw) as AdminOpsMirrorHeartbeat) + } catch { + // ignore corrupt payloads + } + } + + return heartbeats.sort((a, b) => a.index.localeCompare(b.index, undefined, { numeric: true })) + } + + public async listWorkerHeartbeats(): Promise { + const members = await this.cache.getSetMembers(ADMIN_OPS_HEARTBEAT_INDEX_KEY) + const workerMembers = members.filter((member) => member.startsWith('worker:')) + + const heartbeats: AdminOpsWorkerHeartbeat[] = [] + for (const member of workerMembers) { + const id = member.slice('worker:'.length) + const separator = id.indexOf(':') + const type = separator >= 0 ? id.slice(0, separator) : id + const index = separator >= 0 ? id.slice(separator + 1) : undefined + const raw = await this.cache.getKey(workerKey(type, index)) + if (!raw) { + continue + } + try { + heartbeats.push(JSON.parse(raw) as AdminOpsWorkerHeartbeat) + } catch { + // ignore corrupt payloads + } + } + + return heartbeats.sort((a, b) => { + const typeCompare = a.type.localeCompare(b.type) + if (typeCompare !== 0) { + return typeCompare + } + return (a.index ?? '').localeCompare(b.index ?? '', undefined, { numeric: true }) + }) + } +} + +let sharedStore: AdminOpsStore | undefined +let sharedStoreCache: ICacheAdapter | undefined + +export const getAdminOpsStore = (cache: ICacheAdapter): AdminOpsStore => { + if (!sharedStore || sharedStoreCache !== cache) { + sharedStore = new AdminOpsStore(cache) + sharedStoreCache = cache + } + return sharedStore +} + +export const resetAdminOpsStore = (): void => { + sharedStore = undefined + sharedStoreCache = undefined +} + +const sanitizeMirrorHeartbeat = (state: AdminOpsMirrorHeartbeat): AdminOpsMirrorHeartbeat => ({ + ...state, + lastError: state.lastError !== undefined ? sanitizeAdminOpsError(state.lastError) : undefined, +}) + +const sanitizeWorkerHeartbeat = (state: AdminOpsWorkerHeartbeat): AdminOpsWorkerHeartbeat => ({ + ...state, + lastError: state.lastError !== undefined ? sanitizeAdminOpsError(state.lastError) : undefined, +}) + +export const publishAdminOpsMirrorHeartbeat = async ( + cache: ICacheAdapter, + state: AdminOpsMirrorHeartbeat, +): Promise => { + await getAdminOpsStore(cache).publishMirror(sanitizeMirrorHeartbeat(state)) +} + +export const publishAdminOpsWorkerHeartbeat = async ( + cache: ICacheAdapter, + state: AdminOpsWorkerHeartbeat, +): Promise => { + await getAdminOpsStore(cache).publishWorker(sanitizeWorkerHeartbeat(state)) +} + +export const publishAdminOpsMirrorHeartbeatSafely = async ( + cache: ICacheAdapter | undefined, + state: AdminOpsMirrorHeartbeat, +): Promise => { + if (!cache) { + return + } + try { + await publishAdminOpsMirrorHeartbeat(cache, state) + } catch (error) { + logger.warn('failed to publish mirror ops heartbeat: %o', error) + } +} + +export const publishAdminOpsWorkerHeartbeatSafely = async ( + cache: ICacheAdapter | undefined, + state: AdminOpsWorkerHeartbeat, +): Promise => { + if (!cache) { + return + } + try { + await publishAdminOpsWorkerHeartbeat(cache, state) + } catch (error) { + logger.warn('failed to publish worker ops heartbeat: %o', error) + } +} + +export const isAdminOpsHeartbeatStale = (updatedAtIso: string | undefined, ttlSeconds: number): boolean => { + if (!updatedAtIso) { + return true + } + const updatedAt = Date.parse(updatedAtIso) + if (!Number.isFinite(updatedAt)) { + return true + } + return Date.now() - updatedAt > ttlSeconds * 1000 +} diff --git a/src/utils/nip05-reverification.ts b/src/utils/nip05-reverification.ts new file mode 100644 index 000000000..f8039c270 --- /dev/null +++ b/src/utils/nip05-reverification.ts @@ -0,0 +1,24 @@ +import { INip05VerificationRepository } from '../@types/repositories' +import { Nip05Verification } from '../@types/nip05' +import { applyReverificationOutcome } from '../app/maintenance-worker' +import { verifyNip05Identifier } from './nip05' + +export const reverifyNip05Verification = async ( + verification: Nip05Verification, + repository: INip05VerificationRepository, +): Promise => { + const outcome = await verifyNip05Identifier(verification.nip05, verification.pubkey) + const current = await repository.findByPubkey(verification.pubkey) + + if (!current) { + return verification + } + + if (current.nip05 !== verification.nip05) { + return current + } + + const updated = applyReverificationOutcome(current, outcome) + await repository.upsert(updated) + return updated +} diff --git a/test/unit/app/relay-monitor-worker.spec.ts b/test/unit/app/relay-monitor-worker.spec.ts index a8c2d84e7..ada537774 100644 --- a/test/unit/app/relay-monitor-worker.spec.ts +++ b/test/unit/app/relay-monitor-worker.spec.ts @@ -9,7 +9,9 @@ import { getProbeIntervalMs, RelayMonitorWorker, } from '../../../src/app/relay-monitor-worker' +import { ICacheAdapter } from '../../../src/@types/adapters' import { Settings } from '../../../src/@types/settings' +import { ADMIN_OPS_HEARTBEAT_KEEPALIVE_MS, resetAdminOpsStore } from '../../../src/utils/admin-ops-store' import * as metricsTelemetry from '../../../src/telemetry/metrics' import { ProbeResult } from '../../../src/utils/relay-probe/types' @@ -17,6 +19,8 @@ chai.use(sinonChai) const { expect } = chai +const flushAsyncWork = () => new Promise((resolve) => setImmediate(resolve)) + describe('RelayMonitorWorker', () => { let sandbox: Sinon.SinonSandbox let worker: RelayMonitorWorker @@ -51,6 +55,7 @@ describe('RelayMonitorWorker', () => { }) as ProbeResult beforeEach(() => { + resetAdminOpsStore() sandbox = Sinon.createSandbox() sandbox.stub(metricsTelemetry, 'shutdownMetricsTelemetry').resolves() @@ -116,8 +121,8 @@ describe('RelayMonitorWorker', () => { const clock = sandbox.useFakeTimers() worker.run() - await Promise.resolve() - await Promise.resolve() + await clock.tickAsync(0) + await clock.tickAsync(0) expect(probeRunner).to.have.been.calledOnceWith('wss://relay.example.com', buildProbeOptions(settingsState)) expect(snapshotStore.saveLatest).to.have.been.calledOnce @@ -128,6 +133,7 @@ describe('RelayMonitorWorker', () => { historyStore.append.resetHistory() await clock.tickAsync(60_000) + await clock.tickAsync(0) expect(probeRunner).to.have.been.calledOnce expect(snapshotStore.saveLatest).to.have.been.calledOnce @@ -193,14 +199,131 @@ describe('RelayMonitorWorker', () => { ) worker.run() - await Promise.resolve() - await Promise.resolve() - await Promise.resolve() + await flushAsyncWork() + await flushAsyncWork() expect(eventPublisher.publishAfterProbe).to.have.been.calledOnce expect(eventPublisher.publishAfterProbe.firstCall.args[0].status).to.equal('ok') }) + it('keepalive republishes the last probe heartbeat fields', async () => { + const workerPayloads: string[] = [] + const opsCache: ICacheAdapter = { + getKey: async () => '', + hasKey: async () => false, + setKey: async (key, value) => { + if (key.startsWith('admin:ops:worker:')) { + workerPayloads.push(value) + } + return true + }, + setKeyIfNotExists: async () => true, + addToSortedSet: async () => 0, + removeRangeByScoreFromSortedSet: async () => 0, + removeRangeByRankFromSortedSet: async () => 0, + getRangeFromSortedSet: async () => [], + setKeyExpiry: async () => {}, + deleteKey: async () => 0, + getHKey: async () => '', + setHKey: async () => true, + addToSet: async () => 1, + getSetMembers: async () => [], + eval: async () => null, + } + + settingsState.nip66!.targets = ['wss://relay.example.com'] + probeRunner.rejects(new Error('probe failed')) + + const clock = sandbox.useFakeTimers() + worker = new RelayMonitorWorker( + fakeProcess as unknown as NodeJS.Process, + settings, + snapshotStore, + probeRunner, + undefined, + historyStore as any, + opsCache, + ) + + worker.run() + await clock.tickAsync(0) + await clock.tickAsync(0) + + expect(workerPayloads.length).to.be.greaterThan(0) + const afterFailure = JSON.parse(workerPayloads[workerPayloads.length - 1]!) + expect(afterFailure.lastError).to.equal('Probe run produced no results') + + const countBeforeKeepalive = workerPayloads.length + await clock.tickAsync(ADMIN_OPS_HEARTBEAT_KEEPALIVE_MS) + await clock.tickAsync(0) + + expect(workerPayloads.length).to.be.greaterThan(countBeforeKeepalive) + const afterKeepalive = JSON.parse(workerPayloads[workerPayloads.length - 1]!) + expect(afterKeepalive.lastError).to.equal('Probe run produced no results') + expect(afterKeepalive.lastRunStartedAt).to.equal(afterFailure.lastRunStartedAt) + + clock.restore() + }) + + it('clears lastError after a later successful probe run', async () => { + const workerPayloads: string[] = [] + const opsCache: ICacheAdapter = { + getKey: async () => '', + hasKey: async () => false, + setKey: async (key, value) => { + if (key.startsWith('admin:ops:worker:')) { + workerPayloads.push(value) + } + return true + }, + setKeyIfNotExists: async () => true, + addToSortedSet: async () => 0, + removeRangeByScoreFromSortedSet: async () => 0, + removeRangeByRankFromSortedSet: async () => 0, + getRangeFromSortedSet: async () => [], + setKeyExpiry: async () => {}, + deleteKey: async () => 0, + getHKey: async () => '', + setHKey: async () => true, + addToSet: async () => 1, + getSetMembers: async () => [], + eval: async () => null, + } + + settingsState.nip66!.targets = ['wss://relay.example.com'] + probeRunner.onFirstCall().rejects(new Error('probe failed')) + probeRunner.onSecondCall().resolves(probeResult()) + + const clock = sandbox.useFakeTimers() + worker = new RelayMonitorWorker( + fakeProcess as unknown as NodeJS.Process, + settings, + snapshotStore, + probeRunner, + undefined, + historyStore as any, + opsCache, + ) + + worker.run() + await clock.tickAsync(0) + await clock.tickAsync(0) + + expect(workerPayloads.length).to.be.greaterThan(0) + const afterFailure = JSON.parse(workerPayloads[workerPayloads.length - 1]!) + expect(afterFailure.lastError).to.equal('Probe run produced no results') + + await clock.tickAsync(60_000) + await clock.tickAsync(0) + + expect(workerPayloads.length).to.be.greaterThan(1) + const afterSuccess = JSON.parse(workerPayloads[workerPayloads.length - 1]!) + expect(afterSuccess.lastError).to.equal(undefined) + expect(afterSuccess.lastRunFinishedAt).to.be.a('string') + + clock.restore() + }) + it('still publishes NIP-66 events when probe history append fails', async () => { const eventPublisher = { publishAfterProbe: sandbox.stub().resolves(), @@ -218,9 +341,8 @@ describe('RelayMonitorWorker', () => { ) worker.run() - await Promise.resolve() - await Promise.resolve() - await Promise.resolve() + await flushAsyncWork() + await flushAsyncWork() expect(eventPublisher.publishAfterProbe).to.have.been.calledOnce }) diff --git a/test/unit/utils/admin-ops-config.spec.ts b/test/unit/utils/admin-ops-config.spec.ts new file mode 100644 index 000000000..3a1915dbb --- /dev/null +++ b/test/unit/utils/admin-ops-config.spec.ts @@ -0,0 +1,43 @@ +import { expect } from 'chai' + +import { mergeMirrorRows, mergeWorkerRows } from '../../../src/utils/admin-ops-config' +import { ADMIN_OPS_HEARTBEAT_TTL_SECONDS, isAdminOpsHeartbeatStale } from '../../../src/utils/admin-ops-store' + +describe('admin-ops-config', () => { + it('marks unmonitored workers as not stale without a heartbeat', () => { + const rows = mergeWorkerRows( + [{ type: 'worker', index: '0', enabled: true, label: 'Client worker 0' }], + [], + ) + + expect(rows[0].monitored).to.equal(false) + expect(rows[0].stale).to.equal(false) + }) + + it('marks monitored workers stale when heartbeat is expired', () => { + const staleUpdatedAt = new Date(Date.now() - (ADMIN_OPS_HEARTBEAT_TTL_SECONDS + 5) * 1000).toISOString() + const rows = mergeWorkerRows( + [{ type: 'maintenance', enabled: true, label: 'Maintenance' }], + [{ type: 'maintenance', updatedAt: staleUpdatedAt }], + ) + + expect(rows[0].monitored).to.equal(true) + expect(rows[0].stale).to.equal(true) + }) + + it('merges mirror heartbeats and stale state', () => { + const freshUpdatedAt = new Date().toISOString() + const rows = mergeMirrorRows([{ index: '0', targetUrl: 'wss://mirror', enabled: true }], [ + { + index: '0', + targetUrl: 'wss://mirror', + connectionState: 'open', + reconnectCount: 0, + updatedAt: freshUpdatedAt, + }, + ]) + + expect(rows[0].stale).to.equal(false) + expect(isAdminOpsHeartbeatStale(freshUpdatedAt, ADMIN_OPS_HEARTBEAT_TTL_SECONDS)).to.equal(false) + }) +}) diff --git a/test/unit/utils/admin-ops-nip05.spec.ts b/test/unit/utils/admin-ops-nip05.spec.ts new file mode 100644 index 000000000..be0ad20bc --- /dev/null +++ b/test/unit/utils/admin-ops-nip05.spec.ts @@ -0,0 +1,35 @@ +import { expect } from 'chai' + +import { isNip05VerificationStale, toAdminNip05Row } from '../../../src/utils/admin-ops-nip05' + +describe('admin-ops-nip05', () => { + it('marks unverified rows as stale', () => { + const row = toAdminNip05Row( + { + pubkey: 'aa'.repeat(32), + nip05: 'user@example.com', + domain: 'example.com', + isVerified: false, + failureCount: 2, + lastVerifiedAt: null, + lastCheckedAt: new Date('2020-01-01T00:00:00.000Z'), + createdAt: new Date('2020-01-01T00:00:00.000Z'), + updatedAt: new Date('2020-01-01T00:00:00.000Z'), + }, + 86_400_000, + ) + + expect(row.stale).to.equal(true) + }) + + it('treats recently verified rows as fresh', () => { + const now = Date.now() + expect( + isNip05VerificationStale( + { isVerified: true, lastVerifiedAt: new Date(now - 1000) }, + 86_400_000, + now, + ), + ).to.equal(false) + }) +}) diff --git a/test/unit/utils/admin-ops-sanitize.spec.ts b/test/unit/utils/admin-ops-sanitize.spec.ts new file mode 100644 index 000000000..a4778e07c --- /dev/null +++ b/test/unit/utils/admin-ops-sanitize.spec.ts @@ -0,0 +1,19 @@ +import { expect } from 'chai' + +import { sanitizeAdminOpsError } from '../../../src/utils/admin-ops-sanitize' + +describe('sanitizeAdminOpsError', () => { + it('redacts urls and truncates long messages', () => { + const message = `failed wss://secret.relay.example/nostr Bearer token123 ${'x'.repeat(300)}` + const sanitized = sanitizeAdminOpsError(message) + + expect(sanitized).to.include('[url]') + expect(sanitized).to.include('Bearer [redacted]') + expect(sanitized!.length).to.be.at.most(241) + }) + + it('returns undefined for empty input', () => { + expect(sanitizeAdminOpsError(undefined)).to.equal(undefined) + expect(sanitizeAdminOpsError(' ')).to.equal(undefined) + }) +}) diff --git a/test/unit/utils/admin-ops-storage.spec.ts b/test/unit/utils/admin-ops-storage.spec.ts new file mode 100644 index 000000000..07197742e --- /dev/null +++ b/test/unit/utils/admin-ops-storage.spec.ts @@ -0,0 +1,26 @@ +import { expect } from 'chai' + +import { buildStorageCountFilters, parseAdminOpsStorageQuery } from '../../../src/utils/admin-ops-storage' + +describe('admin-ops-storage', () => { + it('requires kinds parameter', () => { + const result = parseAdminOpsStorageQuery(undefined, undefined, undefined) + expect(result.error).to.match(/kinds/) + }) + + it('parses kinds and builds per-kind filters', () => { + const parsed = parseAdminOpsStorageQuery('1,7,7', '100', '200') + expect(parsed.error).to.be.undefined + expect(parsed.query?.kinds).to.deep.equal([1, 7]) + + const filters = buildStorageCountFilters(parsed.query!) + expect(filters).to.have.length(2) + expect(filters[0]).to.deep.equal({ since: 100, until: 200, kinds: [1] }) + expect(filters[1]).to.deep.equal({ since: 100, until: 200, kinds: [7] }) + }) + + it('rejects invalid kind values', () => { + const result = parseAdminOpsStorageQuery('1,abc', undefined, undefined) + expect(result.error).to.match(/kinds/) + }) +}) diff --git a/test/unit/utils/admin-ops-store.spec.ts b/test/unit/utils/admin-ops-store.spec.ts new file mode 100644 index 000000000..9451677a5 --- /dev/null +++ b/test/unit/utils/admin-ops-store.spec.ts @@ -0,0 +1,58 @@ +import { expect } from 'chai' + +import { ICacheAdapter } from '../../../src/@types/adapters' +import { AdminOpsStore } from '../../../src/utils/admin-ops-store' + +const createRecordingCache = (): { cache: ICacheAdapter; workerPayloads: string[] } => { + const workerPayloads: string[] = [] + const cache: ICacheAdapter = { + getKey: async () => '', + hasKey: async () => false, + setKey: async (key, value) => { + if (key.startsWith('admin:ops:worker:')) { + workerPayloads.push(value) + } + return true + }, + setKeyIfNotExists: async () => true, + addToSortedSet: async () => 0, + removeRangeByScoreFromSortedSet: async () => 0, + removeRangeByRankFromSortedSet: async () => 0, + getRangeFromSortedSet: async () => [], + setKeyExpiry: async () => {}, + deleteKey: async () => 0, + getHKey: async () => '', + setHKey: async () => true, + addToSet: async () => 1, + getSetMembers: async () => [], + eval: async () => null, + } + + return { cache, workerPayloads } +} + +describe('AdminOpsStore', () => { + it('overwrites worker heartbeats with the latest publish payload', async () => { + const { cache, workerPayloads } = createRecordingCache() + const store = new AdminOpsStore(cache) + + await store.publishWorker({ + type: 'relay-monitor', + updatedAt: '2024-01-01T00:00:00.000Z', + lastRunFinishedAt: '2024-01-01T00:00:00.000Z', + lastError: 'probe failed', + }) + + await store.publishWorker({ + type: 'relay-monitor', + updatedAt: '2024-01-01T00:00:45.000Z', + lastRunFinishedAt: '2024-01-01T00:00:00.000Z', + lastError: 'probe failed', + }) + + expect(workerPayloads).to.have.length(2) + const latest = JSON.parse(workerPayloads[1]!) + expect(latest.lastError).to.equal('probe failed') + expect(latest.updatedAt).to.equal('2024-01-01T00:00:45.000Z') + }) +}) diff --git a/test/unit/utils/nip05-reverification.spec.ts b/test/unit/utils/nip05-reverification.spec.ts new file mode 100644 index 000000000..a38efdbc0 --- /dev/null +++ b/test/unit/utils/nip05-reverification.spec.ts @@ -0,0 +1,62 @@ +import { expect } from 'chai' +import Sinon from 'sinon' + +import { Nip05Verification } from '../../../src/@types/nip05' +import * as nip05 from '../../../src/utils/nip05' +import { reverifyNip05Verification } from '../../../src/utils/nip05-reverification' + +const baseVerification = (): Nip05Verification => ({ + pubkey: 'aa'.repeat(32), + nip05: 'user@example.com', + domain: 'example.com', + isVerified: false, + lastVerifiedAt: null, + lastCheckedAt: new Date('2024-01-01T00:00:00.000Z'), + failureCount: 1, + createdAt: new Date('2024-01-01T00:00:00.000Z'), + updatedAt: new Date('2024-01-01T00:00:00.000Z'), +}) + +describe('reverifyNip05Verification', () => { + let sandbox: Sinon.SinonSandbox + + beforeEach(() => { + sandbox = Sinon.createSandbox() + }) + + afterEach(() => { + sandbox.restore() + }) + + it('does not upsert when the nip05 identity changed during verification', async () => { + const verification = baseVerification() + const repository = { + findByPubkey: sandbox.stub().resolves({ + ...verification, + nip05: 'other@example.com', + }), + upsert: sandbox.stub().resolves(1), + } + + sandbox.stub(nip05, 'verifyNip05Identifier').resolves({ status: 'verified' }) + + const result = await reverifyNip05Verification(verification, repository as any) + + expect(result.nip05).to.equal('other@example.com') + expect(repository.upsert).to.not.have.been.called + }) + + it('upserts when the stored row still matches the requested nip05', async () => { + const verification = baseVerification() + const repository = { + findByPubkey: sandbox.stub().resolves({ ...verification }), + upsert: sandbox.stub().resolves(1), + } + + sandbox.stub(nip05, 'verifyNip05Identifier').resolves({ status: 'verified' }) + + await reverifyNip05Verification(verification, repository as any) + + expect(repository.upsert).to.have.been.calledOnce + }) +})