diff --git a/CHANGELOG.md b/CHANGELOG.md index f9007e99..3a19436f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Added +- **Scheduled email digest reports, completed and enabled** (#154): the digest feature whose foundation landed in #209 (and shipped disabled since 1.0.0-beta) is now finished and active. The report grew from log volume only to five sections, each with a quiet empty state: log volume with period-over-period trend, top 5 services by error+critical count with delta vs the previous period (engine-agnostic via `reservoir.topValues`, so TimescaleDB/ClickHouse/MongoDB all work), new error groups first seen in the period (`error_groups.first_seen`), a security summary (windowed detection totals and top triggered Sigma rules read from raw `detection_events` to avoid continuous-aggregate staleness, plus a point-in-time open/investigating incident count), and monitor uptime (aggregated from `monitor_uptime_daily`, section omitted for orgs without monitors). Emails are now real HTML built on the shared `email-templates.ts` component library (inline CSS, escaped user-controlled strings, en-US number formatting) with the full report duplicated in the plaintext fallback, and the base template footer learned an `unsubscribeUrl` variant so digest recipients (who may have no account) get a one-click unsubscribe instead of the channel-settings link. Scheduling was redesigned: instead of one cron per organization registered at worker boot (which could never pick up config changes at runtime, because graphile-worker cron items are fixed once the runner starts), a single static hourly `digest-dispatch` cron sweeps enabled `digest_configs`, matches `delivery_hour`/`delivery_day_of_week` against the current UTC hour, and enqueues `digest-generation` jobs with hour-keyed dedup keys and a `last_sent_at` double-fire guard (migration 053), so config CRUD is live on both queue backends with no restart. New management surface: session-auth CRUD under `/api/v1/digests` (config upsert/delete, recipient add/remove/resubscribe with token rotation on resubscribe, membership for reads and owner/admin for writes, audit-logged as `digest.*`, recipients capped by a new `digests.max_recipients` capability with the canonical 4-case tests), a public `POST /api/v1/digests/unsubscribe` where the 32-byte random token is the credential (idempotent, returns a masked email), an "Email Digests" settings page (schedule card with UTC delivery time and recipient management) and a public `/unsubscribe` page consumed by the email footer link. The worker-side disable comment from the beta is gone: the dispatch cron registers at boot + ## [1.1.0] - 2026-07-08 ### Added diff --git a/packages/backend/migrations/053_digest_last_sent.sql b/packages/backend/migrations/053_digest_last_sent.sql new file mode 100644 index 00000000..89c81cb2 --- /dev/null +++ b/packages/backend/migrations/053_digest_last_sent.sql @@ -0,0 +1,7 @@ +-- ============================================================================ +-- Migration 053: Digest dispatch bookkeeping +-- Adds last_sent_at to digest_configs. The hourly digest-dispatch cron uses it +-- as a double-fire guard and operators can see when a digest last went out. +-- ============================================================================ + +ALTER TABLE digest_configs ADD COLUMN IF NOT EXISTS last_sent_at TIMESTAMPTZ; diff --git a/packages/backend/src/capabilities/registry.ts b/packages/backend/src/capabilities/registry.ts index 1d5ca1ca..ff671146 100644 --- a/packages/backend/src/capabilities/registry.ts +++ b/packages/backend/src/capabilities/registry.ts @@ -99,6 +99,11 @@ export const CAPABILITIES = { defaultLimit: null, description: 'Maximum inbound webhook receivers per organization', }, + 'digests.max_recipients': { + kind: 'limit', + defaultLimit: null, + description: 'Maximum digest email recipients per organization', + }, // Consumption quotas (OSS-permissive: null = unlimited). signal maps to #212 metering types. 'ingestion.max_bytes_monthly': { diff --git a/packages/backend/src/database/types.ts b/packages/backend/src/database/types.ts index 72c5e03a..d6c65c65 100644 --- a/packages/backend/src/database/types.ts +++ b/packages/backend/src/database/types.ts @@ -1138,6 +1138,7 @@ export interface DigestConfigsTable { delivery_hour: number; delivery_day_of_week: number | null; enabled: Generated; + last_sent_at: Timestamp | null; created_at: Generated; updated_at: Generated; } diff --git a/packages/backend/src/lib/email-templates.ts b/packages/backend/src/lib/email-templates.ts index fce0953b..dcba239a 100644 --- a/packages/backend/src/lib/email-templates.ts +++ b/packages/backend/src/lib/email-templates.ts @@ -97,10 +97,13 @@ const colors = { interface BaseEmailOptions { preheader?: string; // Preview text shown in email clients + // When set, the footer swaps the channel-settings link for a one-click + // unsubscribe link (used by digest emails, where recipients may have no account) + unsubscribeUrl?: string; } function baseTemplate(content: string, options: BaseEmailOptions = {}): string { - const { preheader } = options; + const { preheader, unsubscribeUrl } = options; const frontendUrl = getFrontendUrl(); const logoUrl = getLogoUrl(); @@ -143,7 +146,9 @@ function baseTemplate(content: string, options: BaseEmailOptions = {}): string { Sent by LogTide

- Manage notification settings + ${unsubscribeUrl + ? `Unsubscribe from these reports` + : `Manage notification settings`}

@@ -833,3 +838,254 @@ Manage notifications: ${frontendUrl}/dashboard/settings/channels return { html, text }; } + +// ============================================================================ +// DIGEST REPORT EMAIL (#154) +// ============================================================================ + +export interface DigestEmailData { + organizationName: string; + frequency: 'daily' | 'weekly'; + periodLabel: string; + logVolume: { + current: number; + previous: number; + trend: string; + }; + topErrorServices: Array<{ + service: string; + errorCount: number; + previousCount: number; + delta: number; + }>; + newErrorGroups: Array<{ + exceptionType: string; + exceptionMessage: string; + occurrenceCount: number; + language: string; + }>; + security: { + totalDetections: number; + topRules: Array<{ ruleTitle: string; severity: string; count: number }>; + openIncidents: number; + }; + uptime: { + monitorCount: number; + overallUptimePct: number; + worstMonitors: Array<{ name: string; uptimePct: number }>; + } | null; + unsubscribeUrl: string; + dashboardUrl: string; +} + +function digestSectionTitle(title: string): string { + return ` + +

+ ${escapeHtml(title)} +

+ + `; +} + +function digestEmptyLine(message: string): string { + return ` + +

${escapeHtml(message)}

+ + `; +} + +function digestTable(headers: string[], rows: string[][]): string { + const headerCells = headers + .map( + (h, i) => + `${escapeHtml(h)}` + ) + .join(''); + const bodyRows = rows + .map( + (cells) => + `${cells + .map( + (c, i) => + `${escapeHtml(c)}` + ) + .join('')}` + ) + .join(''); + + return ` + + + ${headerCells} + ${bodyRows} +
+ + `; +} + +function formatDelta(delta: number): string { + if (delta > 0) return `+${delta.toLocaleString('en-US')}`; + return delta.toLocaleString('en-US'); +} + +export function generateDigestEmail(data: DigestEmailData): { html: string; text: string } { + const frequencyLabel = data.frequency === 'daily' ? 'Daily' : 'Weekly'; + const title = `${frequencyLabel} Digest`; + const quiet = data.logVolume.current === 0 && data.logVolume.previous === 0; + + const logVolumeSection = quiet + ? alertBox('No activity during this period. Your systems have been quiet.', 'info') + : infoBox([ + { label: 'Total Logs', value: data.logVolume.current.toLocaleString('en-US') }, + { label: 'Trend', value: data.logVolume.trend }, + { label: 'Previous Period', value: data.logVolume.previous.toLocaleString('en-US') }, + ]); + + const topServicesSection = + data.topErrorServices.length > 0 + ? digestTable( + ['Service', 'Errors', 'Previous', 'Delta'], + data.topErrorServices.map((s) => [ + s.service, + s.errorCount.toLocaleString('en-US'), + s.previousCount.toLocaleString('en-US'), + formatDelta(s.delta), + ]) + ) + : digestEmptyLine('No errors recorded in this period.'); + + const errorGroupsSection = + data.newErrorGroups.length > 0 + ? digestTable( + ['Error', 'Language', 'Occurrences'], + data.newErrorGroups.map((g) => [ + `${g.exceptionType}${g.exceptionMessage ? ': ' + truncate(g.exceptionMessage, 60) : ''}`, + g.language, + g.occurrenceCount.toLocaleString('en-US'), + ]) + ) + : digestEmptyLine('No new error groups appeared in this period.'); + + const securityRows = [ + { label: 'Detections', value: data.security.totalDetections.toLocaleString('en-US') }, + { label: 'Open Incidents', value: data.security.openIncidents.toLocaleString('en-US') }, + ]; + const securitySection = + data.security.topRules.length > 0 + ? infoBox(securityRows) + + digestTable( + ['Rule', 'Severity', 'Detections'], + data.security.topRules.map((r) => [r.ruleTitle, r.severity, r.count.toLocaleString('en-US')]) + ) + : infoBox(securityRows); + + const uptimeSection = data.uptime + ? digestSectionTitle('Uptime') + + infoBox([ + { label: 'Overall', value: `${data.uptime.overallUptimePct.toLocaleString('en-US')}%` }, + { label: 'Monitors', value: data.uptime.monitorCount.toLocaleString('en-US') }, + ]) + + (data.uptime.worstMonitors.length > 0 + ? digestTable( + ['Monitor', 'Uptime'], + data.uptime.worstMonitors.map((m) => [m.name, `${m.uptimePct.toLocaleString('en-US')}%`]) + ) + : '') + : ''; + + const html = baseTemplate( + card(` + ${header(title, { text: data.frequency, color: colors.info })} + ${divider()} + ${subtitle(`${data.organizationName} - ${data.periodLabel}`)} + ${timestamp()} + ${digestSectionTitle('Log Volume')} + ${logVolumeSection} + ${digestSectionTitle('Top Services by Errors')} + ${topServicesSection} + ${digestSectionTitle('New Error Groups')} + ${errorGroupsSection} + ${digestSectionTitle('Security')} + ${securitySection} + ${uptimeSection} + ${cta('View Dashboard', data.dashboardUrl)} + `), + { + preheader: quiet + ? `Quiet period for ${data.organizationName}` + : `${data.logVolume.current.toLocaleString('en-US')} logs, ${data.security.totalDetections.toLocaleString('en-US')} detections for ${data.organizationName}`, + unsubscribeUrl: data.unsubscribeUrl, + } + ); + + const lines: string[] = []; + lines.push(`LogTide ${frequencyLabel} Digest`); + lines.push(`Organization: ${data.organizationName}`); + lines.push(`Period: ${data.periodLabel}`); + lines.push(''); + lines.push('='.repeat(50)); + lines.push(''); + lines.push('LOG VOLUME'); + lines.push('-'.repeat(10)); + if (quiet) { + lines.push('No activity during this period.'); + lines.push('Your systems have been quiet.'); + } else { + lines.push(`Total logs: ${data.logVolume.current.toLocaleString('en-US')}`); + lines.push(`Trend: ${data.logVolume.trend}`); + lines.push(`Previous period: ${data.logVolume.previous.toLocaleString('en-US')}`); + } + lines.push(''); + lines.push('TOP SERVICES BY ERRORS'); + lines.push('-'.repeat(22)); + if (data.topErrorServices.length > 0) { + for (const s of data.topErrorServices) { + lines.push( + `${s.service}: ${s.errorCount.toLocaleString('en-US')} errors (previous: ${s.previousCount.toLocaleString('en-US')}, delta: ${formatDelta(s.delta)})` + ); + } + } else { + lines.push('No errors recorded in this period.'); + } + lines.push(''); + lines.push('NEW ERROR GROUPS'); + lines.push('-'.repeat(16)); + if (data.newErrorGroups.length > 0) { + for (const g of data.newErrorGroups) { + const message = g.exceptionMessage ? `: ${truncate(g.exceptionMessage, 80)}` : ''; + lines.push(`${g.exceptionType}${message} (${g.language}, ${g.occurrenceCount.toLocaleString('en-US')} occurrences)`); + } + } else { + lines.push('No new error groups appeared in this period.'); + } + lines.push(''); + lines.push('SECURITY'); + lines.push('-'.repeat(8)); + lines.push(`Detections: ${data.security.totalDetections.toLocaleString('en-US')}`); + lines.push(`Open incidents: ${data.security.openIncidents.toLocaleString('en-US')}`); + for (const r of data.security.topRules) { + lines.push(`${r.ruleTitle} [${r.severity}]: ${r.count.toLocaleString('en-US')} detections`); + } + if (data.uptime) { + lines.push(''); + lines.push('UPTIME'); + lines.push('-'.repeat(6)); + lines.push(`Overall: ${data.uptime.overallUptimePct.toLocaleString('en-US')}% across ${data.uptime.monitorCount.toLocaleString('en-US')} monitor(s)`); + for (const m of data.uptime.worstMonitors) { + lines.push(`${m.name}: ${m.uptimePct.toLocaleString('en-US')}%`); + } + } + lines.push(''); + lines.push(`View your dashboard: ${data.dashboardUrl}`); + lines.push(''); + lines.push('To unsubscribe from these reports, click:'); + lines.push(data.unsubscribeUrl); + lines.push(''); + lines.push('--'); + lines.push('Sent by LogTide'); + + return { html, text: lines.join('\n') }; +} + diff --git a/packages/backend/src/modules/audit-log/actions.ts b/packages/backend/src/modules/audit-log/actions.ts index c6dca7fc..72e88839 100644 --- a/packages/backend/src/modules/audit-log/actions.ts +++ b/packages/backend/src/modules/audit-log/actions.ts @@ -65,6 +65,12 @@ export const AUDIT_ACTIONS = { 'channel.created': 'config_change', 'channel.updated': 'config_change', 'channel.deleted': 'config_change', + // email digest reports (#154) + 'digest.config_updated': 'config_change', + 'digest.config_deleted': 'config_change', + 'digest.recipient_added': 'config_change', + 'digest.recipient_removed': 'config_change', + 'digest.recipient_resubscribed': 'config_change', // webhooks 'webhook.delivery_replayed': 'config_change', // auth diff --git a/packages/backend/src/modules/digests/generator.ts b/packages/backend/src/modules/digests/generator.ts index 518a2781..67b35447 100644 --- a/packages/backend/src/modules/digests/generator.ts +++ b/packages/backend/src/modules/digests/generator.ts @@ -1,7 +1,9 @@ /** * Digest Generator Service * - * Generates and sends email digest reports summarizing log activity over a period. + * Generates and sends email digest reports summarizing organization activity + * over a period: log volume, top services by errors, new error groups, + * security detections and monitor uptime. */ import nodemailer from 'nodemailer'; @@ -9,12 +11,40 @@ import { db } from '../../database/connection.js'; import { reservoir } from '../../database/reservoir.js'; import { hub } from '@logtide/core'; import { config } from '../../config/index.js'; +import { generateDigestEmail } from '../../lib/email-templates.js'; import type { DigestJobPayload } from './scheduler.js'; -interface LogVolumeStats { - currentPeriodCount: number; - previousPeriodCount: number; - trend: string; +export interface DigestReportData { + organizationName: string; + frequency: 'daily' | 'weekly'; + periodLabel: string; + logVolume: { + current: number; + previous: number; + trend: string; + }; + topErrorServices: Array<{ + service: string; + errorCount: number; + previousCount: number; + delta: number; + }>; + newErrorGroups: Array<{ + exceptionType: string; + exceptionMessage: string; + occurrenceCount: number; + language: string; + }>; + security: { + totalDetections: number; + topRules: Array<{ ruleTitle: string; severity: string; count: number }>; + openIncidents: number; + }; + uptime: { + monitorCount: number; + overallUptimePct: number; + worstMonitors: Array<{ name: string; uptimePct: number }>; + } | null; } interface DigestRecipient { @@ -22,6 +52,14 @@ interface DigestRecipient { unsubscribe_token: string; } +interface Period { + from: Date; + to: Date; + previousFrom: Date; + previousTo: Date; +} + +const ERROR_LEVELS = ['error', 'critical'] as const; let emailTransporter: nodemailer.Transporter | null = null; @@ -53,14 +91,13 @@ function getEmailTransporter(): nodemailer.Transporter | null { } export class DigestGeneratorService { - + async generateAndSendDigest(payload: DigestJobPayload): Promise { const { organizationId, digestConfigId, frequency } = payload; hub.captureLog('info', `[DigestGenerator] Generating ${frequency} digest for org ${organizationId}`); try { - const organization = await db .selectFrom('organizations') .select(['name']) @@ -71,7 +108,6 @@ export class DigestGeneratorService { throw new Error(`Organization ${organizationId} not found`); } - const recipients = await this.fetchRecipients(organizationId, digestConfigId); if (recipients.length === 0) { @@ -79,16 +115,9 @@ export class DigestGeneratorService { return; } - - const stats = await this.calculateLogVolume(organizationId, frequency); + const report = await this.buildReportData(organizationId, organization.name, frequency); - - await this.sendDigestEmails( - recipients, - organization.name, - frequency, - stats - ); + await this.sendDigestEmails(recipients, report); hub.captureLog('info', `[DigestGenerator] Digest sent to ${recipients.length} recipient(s) for org ${organizationId}`); } catch (error: any) { @@ -97,92 +126,284 @@ export class DigestGeneratorService { } } - - private async fetchRecipients( + /** + * Compute all report sections for the period. Sections run sequentially: + * this is a background cron path where simplicity beats latency. + */ + async buildReportData( organizationId: string, - digestConfigId: string - ): Promise { - const recipients = await db - .selectFrom('digest_recipients') - .select(['email', 'unsubscribe_token']) + organizationName: string, + frequency: 'daily' | 'weekly' + ): Promise { + const period = this.buildPeriod(frequency); + + const projects = await db + .selectFrom('projects') + .select(['id']) .where('organization_id', '=', organizationId) - .where('digest_config_id', '=', digestConfigId) - .where('subscribed', '=', true) .execute(); + const projectIds = projects.map((p) => p.id); - return recipients; + const logVolume = await this.calculateLogVolume(projectIds, period); + const topErrorServices = await this.calculateTopErrorServices(projectIds, period); + const newErrorGroups = await this.calculateNewErrorGroups(organizationId, period); + const security = await this.calculateSecuritySummary(organizationId, period); + const uptime = await this.calculateUptimeSummary(organizationId, period); + + return { + organizationName, + frequency, + periodLabel: frequency === 'daily' ? 'last 24 hours' : 'last 7 days', + logVolume, + topErrorServices, + newErrorGroups, + security, + uptime, + }; + } + + private buildPeriod(frequency: 'daily' | 'weekly'): Period { + // Sliding window relative to execution time + const hoursInPeriod = frequency === 'daily' ? 24 : 168; + const now = new Date(); + const from = new Date(now.getTime() - hoursInPeriod * 60 * 60 * 1000); + const previousFrom = new Date(now.getTime() - hoursInPeriod * 2 * 60 * 60 * 1000); + + return { from, to: now, previousFrom, previousTo: from }; } private async calculateLogVolume( + projectIds: string[], + period: Period + ): Promise { + if (projectIds.length === 0) { + return { current: 0, previous: 0, trend: this.calculateTrend(0, 0) }; + } + + const currentResult = await reservoir.count({ + projectId: projectIds, + from: period.from, + to: period.to, + toExclusive: true, + }); + + const previousResult = await reservoir.count({ + projectId: projectIds, + from: period.previousFrom, + to: period.previousTo, + toExclusive: true, + }); + + return { + current: currentResult.count, + previous: previousResult.count, + trend: this.calculateTrend(currentResult.count, previousResult.count), + }; + } + + /** + * Top 5 services by error+critical log count, with the delta against the + * previous period. Engine-agnostic through reservoir.topValues. + */ + private async calculateTopErrorServices( + projectIds: string[], + period: Period + ): Promise { + if (projectIds.length === 0) { + return []; + } + + const current = await reservoir.topValues({ + field: 'service', + projectId: projectIds, + level: [...ERROR_LEVELS], + from: period.from, + to: period.to, + limit: 5, + }); + + if (current.values.length === 0) { + return []; + } + + // Wide limit so services that dropped out of the top 5 still resolve + const previous = await reservoir.topValues({ + field: 'service', + projectId: projectIds, + level: [...ERROR_LEVELS], + from: period.previousFrom, + to: period.previousTo, + limit: 100, + }); + + const previousByService = new Map(previous.values.map((v) => [v.value, v.count])); + + return current.values.map((v) => { + const previousCount = previousByService.get(v.value) ?? 0; + return { + service: v.value, + errorCount: v.count, + previousCount, + delta: v.count - previousCount, + }; + }); + } + + /** + * Error groups whose first occurrence falls inside the period. + */ + private async calculateNewErrorGroups( organizationId: string, - frequency: 'daily' | 'weekly' - ): Promise { - // Uses a 24-hour sliding window relative to execution time - const hoursInPeriod = frequency === 'daily' ? 24 : 168; // 7 days = 168 hours + period: Period + ): Promise { + const groups = await db + .selectFrom('error_groups') + .select(['exception_type', 'exception_message', 'occurrence_count', 'language']) + .where('organization_id', '=', organizationId) + .where('first_seen', '>=', period.from) + .where('first_seen', '<', period.to) + .orderBy('occurrence_count', 'desc') + .limit(10) + .execute(); - const now = new Date(); - const currentPeriodStart = new Date(now.getTime() - hoursInPeriod * 60 * 60 * 1000); - const previousPeriodStart = new Date(now.getTime() - hoursInPeriod * 2 * 60 * 60 * 1000); - const previousPeriodEnd = currentPeriodStart; + return groups.map((g) => ({ + exceptionType: g.exception_type, + exceptionMessage: g.exception_message ?? '', + occurrenceCount: g.occurrence_count, + language: g.language, + })); + } - let currentPeriodCount = 0; - let previousPeriodCount = 0; + /** + * Security summary: windowed detection totals and top triggered Sigma rules + * (raw detection_events, so the most recent hours are never stale like the + * continuous aggregates), plus a point-in-time open incident count. + */ + private async calculateSecuritySummary( + organizationId: string, + period: Period + ): Promise { + const totalRow = await db + .selectFrom('detection_events') + .select((eb) => eb.fn.countAll().as('count')) + .where('organization_id', '=', organizationId) + .where('time', '>=', period.from) + .where('time', '<', period.to) + .executeTakeFirst(); - try { - - // organization-level metrics we must count logs for all projects that - // belong to the organization and pass those project ids to the reservoir - const projects = await db - .selectFrom('projects') - .select(['id']) - .where('organization_id', '=', organizationId) - .execute(); - - const projectIds = projects.map((p) => p.id); - - // If the organization has no projects, return zero counts explicitly - if (projectIds.length === 0) { - const trend = this.calculateTrend(0, 0); - return { - currentPeriodCount: 0, - previousPeriodCount: 0, - trend, - }; - } else { - const currentPeriodResult = await reservoir.count({ - projectId: projectIds, - from: currentPeriodStart, - to: now, - toExclusive: true, - }); - currentPeriodCount = currentPeriodResult.count; + const topRules = await db + .selectFrom('detection_events') + .select((eb) => ['rule_title', 'severity', eb.fn.countAll().as('count')] as const) + .where('organization_id', '=', organizationId) + .where('time', '>=', period.from) + .where('time', '<', period.to) + .groupBy(['rule_title', 'severity']) + .orderBy('count', 'desc') + .limit(5) + .execute(); + + const openRow = await db + .selectFrom('incidents') + .select((eb) => eb.fn.countAll().as('count')) + .where('organization_id', '=', organizationId) + .where('status', 'in', ['open', 'investigating']) + .executeTakeFirst(); + + return { + totalDetections: Number(totalRow?.count ?? 0), + topRules: topRules.map((r) => ({ + ruleTitle: r.rule_title, + severity: r.severity, + count: Number(r.count), + })), + openIncidents: Number(openRow?.count ?? 0), + }; + } + + /** + * Uptime summary from monitor_uptime_daily. Returns null when the org has + * no enabled monitors so the email can skip the section entirely. + * Daily buckets only partially overlap a sliding 24h window; that + * approximation is acceptable for a digest. + */ + private async calculateUptimeSummary( + organizationId: string, + period: Period + ): Promise { + const monitors = await db + .selectFrom('monitors') + .select(['id', 'name']) + .where('organization_id', '=', organizationId) + .where('enabled', '=', true) + .execute(); + + if (monitors.length === 0) { + return null; + } - const previousPeriodResult = await reservoir.count({ - projectId: projectIds, - from: previousPeriodStart, - to: previousPeriodEnd, - toExclusive: true, + const bucketFrom = new Date(period.from); + bucketFrom.setUTCHours(0, 0, 0, 0); + + const rows = await db + .selectFrom('monitor_uptime_daily') + .select((eb) => [ + 'monitor_id', + eb.fn.sum('successful_checks').as('successful'), + eb.fn.sum('total_checks').as('total'), + ]) + .where('organization_id', '=', organizationId) + .where('bucket', '>=', bucketFrom) + .groupBy('monitor_id') + .execute(); + + const statsByMonitor = new Map(rows.map((r) => [r.monitor_id, r])); + + let overallSuccessful = 0; + let overallTotal = 0; + const perMonitor: Array<{ name: string; uptimePct: number }> = []; + + for (const monitor of monitors) { + const stats = statsByMonitor.get(monitor.id); + const successful = Number(stats?.successful ?? 0); + const total = Number(stats?.total ?? 0); + overallSuccessful += successful; + overallTotal += total; + + if (total > 0) { + perMonitor.push({ + name: monitor.name, + uptimePct: Math.round((successful / total) * 10000) / 100, }); - previousPeriodCount = previousPeriodResult.count; } - } catch (error: unknown) { - const message = error instanceof Error ? error.message : String(error); - hub.captureLog('error', `[DigestGenerator] Log volume count failed for org ${organizationId}: ${message}`, { error }); - throw error; } - const trend = this.calculateTrend(currentPeriodCount, previousPeriodCount); + const overallUptimePct = + overallTotal > 0 ? Math.round((overallSuccessful / overallTotal) * 10000) / 100 : 100; + + perMonitor.sort((a, b) => a.uptimePct - b.uptimePct); return { - currentPeriodCount, - previousPeriodCount, - trend, + monitorCount: monitors.length, + overallUptimePct, + worstMonitors: perMonitor.slice(0, 5), }; } - - //current and previous counts - + private async fetchRecipients( + organizationId: string, + digestConfigId: string + ): Promise { + const recipients = await db + .selectFrom('digest_recipients') + .select(['email', 'unsubscribe_token']) + .where('organization_id', '=', organizationId) + .where('digest_config_id', '=', digestConfigId) + .where('subscribed', '=', true) + .execute(); + + return recipients; + } + private calculateTrend(current: number, previous: number): string { if (previous === 0 && current === 0) { return 'no change'; @@ -204,12 +425,9 @@ export class DigestGeneratorService { } } - //mail private async sendDigestEmails( recipients: DigestRecipient[], - organizationName: string, - frequency: 'daily' | 'weekly', - stats: LogVolumeStats + report: DigestReportData ): Promise { const transporter = getEmailTransporter(); @@ -217,22 +435,22 @@ export class DigestGeneratorService { throw new Error('Email transporter not configured'); } - const subject = `[LogTide Digest] ${frequency === 'daily' ? 'Daily' : 'Weekly'} Report - ${organizationName}`; + const subject = `[LogTide Digest] ${report.frequency === 'daily' ? 'Daily' : 'Weekly'} Report - ${report.organizationName}`; + const frontendUrl = this.getFrontendUrl(); - const emailPromises = recipients.map(async (recipient) => { - const text = this.generatePlaintextEmail( - organizationName, - frequency, - stats, - recipient.unsubscribe_token - ); + const { html, text } = generateDigestEmail({ + ...report, + unsubscribeUrl: `${frontendUrl}/unsubscribe?token=${recipient.unsubscribe_token}`, + dashboardUrl: `${frontendUrl}/dashboard`, + }); await transporter.sendMail({ from: `"LogTide" <${config.SMTP_FROM || config.SMTP_USER}>`, to: recipient.email, subject, text, + html, }); hub.captureLog('info', `[DigestGenerator] Email sent to ${recipient.email}`); @@ -249,53 +467,6 @@ export class DigestGeneratorService { } } - //email template - private generatePlaintextEmail( - organizationName: string, - frequency: 'daily' | 'weekly', - stats: LogVolumeStats, - unsubscribeToken: string - ): string { - const period = frequency === 'daily' ? 'last 24 hours' : 'last 7 days'; - const frontendUrl = this.getFrontendUrl(); - const unsubscribeUrl = `${frontendUrl}/unsubscribe?token=${unsubscribeToken}`; - - let content = `LogTide ${frequency === 'daily' ? 'Daily' : 'Weekly'} Digest\n`; - content += `Organization: ${organizationName}\n`; - content += `Period: ${period}\n`; - content += `\n`; - content += `━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n`; - content += `\n`; - - if (stats.currentPeriodCount === 0 && stats.previousPeriodCount === 0) { - content += ` Log Volume\n`; - content += `\n`; - content += `No activity during this period.\n`; - content += `Your systems have been quiet.\n`; - } else { - content += ` Log Volume\n`; - content += `\n`; - // Fixed locale: digest content must not depend on the server's locale - content += `Total logs: ${stats.currentPeriodCount.toLocaleString('en-US')}\n`; - content += `Trend: ${stats.trend}\n`; - content += `Previous period: ${stats.previousPeriodCount.toLocaleString('en-US')}\n`; - } - - content += `\n`; - content += `━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n`; - content += `\n`; - content += `View your dashboard: ${frontendUrl}\n`; - content += `\n`; - content += `To unsubscribe from these reports, click:\n`; - content += `${unsubscribeUrl}\n`; - content += `\n`; - content += `—\n`; - content += `LogTide - observability for your infrastructure\n`; - - return content; - } - - private getFrontendUrl(): string { return config.FRONTEND_URL || 'http://localhost:3000'; } diff --git a/packages/backend/src/modules/digests/index.ts b/packages/backend/src/modules/digests/index.ts new file mode 100644 index 00000000..ec616a70 --- /dev/null +++ b/packages/backend/src/modules/digests/index.ts @@ -0,0 +1,5 @@ +export * from './service.js'; +export * from './routes.js'; +export * from './public-routes.js'; +export * from './scheduler.js'; +export * from './generator.js'; diff --git a/packages/backend/src/modules/digests/public-routes.ts b/packages/backend/src/modules/digests/public-routes.ts new file mode 100644 index 00000000..7a24c244 --- /dev/null +++ b/packages/backend/src/modules/digests/public-routes.ts @@ -0,0 +1,45 @@ +/** + * Public Digest Routes (#154) + * + * One-click unsubscribe consumed by the link in digest email footers. No + * authentication: the 32-byte random unsubscribe token is the credential + * (unique index on digest_recipients.unsubscribe_token, not enumerable). + */ + +import type { FastifyInstance } from 'fastify'; +import { z } from 'zod'; +import { digestsService } from './service.js'; + +const unsubscribeSchema = z.object({ + token: z.string().min(1).max(200), +}); + +/** + * Mask an email for the unsubscribe confirmation page: g***@example.com + */ +function maskEmail(email: string): string { + const [local, domain] = email.split('@'); + if (!domain) return '***'; + const visible = local.slice(0, 1); + return `${visible}***@${domain}`; +} + +export async function digestsPublicRoutes(fastify: FastifyInstance) { + /** + * POST /api/v1/digests/unsubscribe + * Marks the recipient behind the token as unsubscribed. Idempotent. + */ + fastify.post('/unsubscribe', async (request, reply) => { + const parsed = unsubscribeSchema.safeParse(request.body); + if (!parsed.success) { + return reply.status(400).send({ error: 'Invalid unsubscribe request' }); + } + + const result = await digestsService.unsubscribeByToken(parsed.data.token); + if (!result) { + return reply.status(404).send({ error: 'Invalid or expired unsubscribe link' }); + } + + return reply.send({ success: true, email: maskEmail(result.email) }); + }); +} diff --git a/packages/backend/src/modules/digests/routes.ts b/packages/backend/src/modules/digests/routes.ts new file mode 100644 index 00000000..d28a8794 --- /dev/null +++ b/packages/backend/src/modules/digests/routes.ts @@ -0,0 +1,320 @@ +/** + * Digest Configuration API Routes (#154) + * + * Session-authenticated CRUD for the per-organization digest config and its + * recipient list. Reads need membership; writes need owner/admin. + */ + +import type { FastifyInstance } from 'fastify'; +import { z } from 'zod'; +import { digestsService } from './service.js'; +import { authenticate } from '../auth/middleware.js'; +import { OrganizationsService } from '../organizations/service.js'; +import { context } from '@logtide/shared/context'; +import { assertWithinLimit, withLimitLock } from '../../capabilities/index.js'; +import { CapabilityError } from '../../capabilities/errors.js'; +import { auditLogService } from '../audit-log/service.js'; + +const organizationsService = new OrganizationsService(); + +// ============================================================================ +// VALIDATION SCHEMAS +// ============================================================================ + +const orgQuerySchema = z.object({ + organizationId: z.string().uuid('organizationId must be a valid uuid'), +}); + +const configBodySchema = z + .object({ + frequency: z.enum(['daily', 'weekly']), + deliveryHour: z.number().int().min(0).max(23), + deliveryDayOfWeek: z.number().int().min(0).max(6).nullish(), + enabled: z.boolean(), + }) + .refine((body) => body.frequency !== 'weekly' || body.deliveryDayOfWeek != null, { + message: 'deliveryDayOfWeek is required for weekly digests', + path: ['deliveryDayOfWeek'], + }); + +const addRecipientSchema = z.object({ + email: z.string().email('Must be a valid email address'), +}); + +const recipientParamsSchema = z.object({ + id: z.string().uuid(), +}); + +// ============================================================================ +// HELPERS +// ============================================================================ + +async function checkOrganizationMembership( + userId: string, + organizationId: string +): Promise { + const organizations = await organizationsService.getUserOrganizations(userId); + return organizations.some((org) => org.id === organizationId); +} + +async function checkAdminRole(userId: string, organizationId: string): Promise { + return organizationsService.isOwnerOrAdmin(organizationId, userId); +} + +// ============================================================================ +// ROUTES +// ============================================================================ + +export async function digestsRoutes(fastify: FastifyInstance) { + // All routes require authentication + fastify.addHook('onRequest', authenticate); + + /** + * GET /api/v1/digests/config + * Current digest config + recipients for an organization (members) + */ + fastify.get('/config', async (request: any, reply) => { + try { + const { organizationId } = orgQuerySchema.parse(request.query); + + const isMember = await checkOrganizationMembership(request.user.id, organizationId); + if (!isMember) { + return reply.status(403).send({ error: 'Not a member of this organization' }); + } + + const config = await digestsService.getConfig(organizationId); + const recipients = await digestsService.listRecipients(organizationId); + + return reply.send({ config: config ?? null, recipients }); + } catch (error) { + if (error instanceof z.ZodError) { + return reply.status(400).send({ error: 'Validation error', details: error.errors }); + } + console.error(error, 'Failed to get digest config'); + return reply.status(500).send({ error: 'Failed to get digest config' }); + } + }); + + /** + * PUT /api/v1/digests/config + * Create or update the organization digest config (admin only) + */ + fastify.put('/config', async (request: any, reply) => { + try { + const { organizationId } = orgQuerySchema.parse(request.query); + const body = configBodySchema.parse(request.body); + + const isMember = await checkOrganizationMembership(request.user.id, organizationId); + if (!isMember) { + return reply.status(403).send({ error: 'Not a member of this organization' }); + } + + const isAdmin = await checkAdminRole(request.user.id, organizationId); + if (!isAdmin) { + return reply.status(403).send({ error: 'Only admins can configure digests' }); + } + + const config = await digestsService.upsertConfig(organizationId, { + frequency: body.frequency, + deliveryHour: body.deliveryHour, + deliveryDayOfWeek: body.deliveryDayOfWeek, + enabled: body.enabled, + }); + + await auditLogService.record({ + action: 'digest.config_updated', + target: { type: 'digest_config', id: config.id }, + organizationId, + metadata: { + frequency: config.frequency, + deliveryHour: config.delivery_hour, + enabled: config.enabled, + }, + }); + + return reply.send({ config }); + } catch (error) { + if (error instanceof z.ZodError) { + return reply.status(400).send({ error: 'Validation error', details: error.errors }); + } + console.error(error, 'Failed to save digest config'); + return reply.status(500).send({ error: 'Failed to save digest config' }); + } + }); + + /** + * DELETE /api/v1/digests/config + * Remove the digest config (and its recipients via cascade) (admin only) + */ + fastify.delete('/config', async (request: any, reply) => { + try { + const { organizationId } = orgQuerySchema.parse(request.query); + + const isMember = await checkOrganizationMembership(request.user.id, organizationId); + if (!isMember) { + return reply.status(403).send({ error: 'Not a member of this organization' }); + } + + const isAdmin = await checkAdminRole(request.user.id, organizationId); + if (!isAdmin) { + return reply.status(403).send({ error: 'Only admins can configure digests' }); + } + + const deleted = await digestsService.deleteConfig(organizationId); + if (!deleted) { + return reply.status(404).send({ error: 'Digest config not found' }); + } + + await auditLogService.record({ + action: 'digest.config_deleted', + target: { type: 'digest_config', id: organizationId }, + organizationId, + }); + + return reply.status(204).send(); + } catch (error) { + if (error instanceof z.ZodError) { + return reply.status(400).send({ error: 'Validation error', details: error.errors }); + } + console.error(error, 'Failed to delete digest config'); + return reply.status(500).send({ error: 'Failed to delete digest config' }); + } + }); + + /** + * POST /api/v1/digests/recipients + * Add a recipient (admin only, capability-limited) + */ + fastify.post('/recipients', async (request: any, reply) => { + try { + const { organizationId } = orgQuerySchema.parse(request.query); + const body = addRecipientSchema.parse(request.body); + + const isMember = await checkOrganizationMembership(request.user.id, organizationId); + if (!isMember) { + return reply.status(403).send({ error: 'Not a member of this organization' }); + } + + const isAdmin = await checkAdminRole(request.user.id, organizationId); + if (!isAdmin) { + return reply.status(403).send({ error: 'Only admins can manage digest recipients' }); + } + + const recipient = await withLimitLock(organizationId, 'digests.max_recipients', async () => { + await context.runAsSystem('digests:recipient-limit-check', async () => { + await context.with({ organizationId }, async () => { + const count = await digestsService.countRecipients(organizationId); + await assertWithinLimit('digests.max_recipients', count); + }); + }); + + return digestsService.addRecipient(organizationId, body.email); + }); + + await auditLogService.record({ + action: 'digest.recipient_added', + target: { type: 'digest_recipient', id: recipient.id }, + organizationId, + metadata: { email: recipient.email }, + }); + + return reply.status(201).send({ recipient }); + } catch (error) { + if (error instanceof CapabilityError) { + throw error; + } + if (error instanceof z.ZodError) { + return reply.status(400).send({ error: 'Validation error', details: error.errors }); + } + if (error instanceof Error && error.message.includes('not configured')) { + return reply.status(409).send({ error: 'Configure the digest before adding recipients' }); + } + if (error instanceof Error && (error.message.includes('unique') || error.message.includes('duplicate'))) { + return reply.status(409).send({ error: 'This email is already a recipient' }); + } + console.error(error, 'Failed to add digest recipient'); + return reply.status(500).send({ error: 'Failed to add recipient' }); + } + }); + + /** + * DELETE /api/v1/digests/recipients/:id + * Remove a recipient (admin only) + */ + fastify.delete('/recipients/:id', async (request: any, reply) => { + try { + const { organizationId } = orgQuerySchema.parse(request.query); + const { id } = recipientParamsSchema.parse(request.params); + + const isMember = await checkOrganizationMembership(request.user.id, organizationId); + if (!isMember) { + return reply.status(403).send({ error: 'Not a member of this organization' }); + } + + const isAdmin = await checkAdminRole(request.user.id, organizationId); + if (!isAdmin) { + return reply.status(403).send({ error: 'Only admins can manage digest recipients' }); + } + + const removed = await digestsService.removeRecipient(organizationId, id); + if (!removed) { + return reply.status(404).send({ error: 'Recipient not found' }); + } + + await auditLogService.record({ + action: 'digest.recipient_removed', + target: { type: 'digest_recipient', id }, + organizationId, + }); + + return reply.status(204).send(); + } catch (error) { + if (error instanceof z.ZodError) { + return reply.status(400).send({ error: 'Validation error', details: error.errors }); + } + console.error(error, 'Failed to remove digest recipient'); + return reply.status(500).send({ error: 'Failed to remove recipient' }); + } + }); + + /** + * POST /api/v1/digests/recipients/:id/resubscribe + * Re-subscribe a recipient who opted out (admin only) + */ + fastify.post('/recipients/:id/resubscribe', async (request: any, reply) => { + try { + const { organizationId } = orgQuerySchema.parse(request.query); + const { id } = recipientParamsSchema.parse(request.params); + + const isMember = await checkOrganizationMembership(request.user.id, organizationId); + if (!isMember) { + return reply.status(403).send({ error: 'Not a member of this organization' }); + } + + const isAdmin = await checkAdminRole(request.user.id, organizationId); + if (!isAdmin) { + return reply.status(403).send({ error: 'Only admins can manage digest recipients' }); + } + + const recipient = await digestsService.resubscribeRecipient(organizationId, id); + if (!recipient) { + return reply.status(404).send({ error: 'Recipient not found' }); + } + + await auditLogService.record({ + action: 'digest.recipient_resubscribed', + target: { type: 'digest_recipient', id }, + organizationId, + metadata: { email: recipient.email }, + }); + + return reply.send({ recipient }); + } catch (error) { + if (error instanceof z.ZodError) { + return reply.status(400).send({ error: 'Validation error', details: error.errors }); + } + console.error(error, 'Failed to resubscribe digest recipient'); + return reply.status(500).send({ error: 'Failed to resubscribe recipient' }); + } + }); +} diff --git a/packages/backend/src/modules/digests/scheduler.ts b/packages/backend/src/modules/digests/scheduler.ts index 2dc163ac..540e3ba7 100644 --- a/packages/backend/src/modules/digests/scheduler.ts +++ b/packages/backend/src/modules/digests/scheduler.ts @@ -1,17 +1,21 @@ /** * Digest Scheduler * - * Reads all active digest configs from the database and registers them as - * repeating cron jobs via the queue abstraction. Called once at worker startup. + * Registers a single static hourly cron job ("digest-dispatch") at worker + * startup. The dispatch job (src/queue/jobs/digest-dispatch.ts) scans the + * active digest configs each hour and enqueues digest-generation jobs for the + * ones due at that UTC hour. * - * Both BullMQ (Redis) and graphile-worker (PostgreSQL) backends are supported - * through the ICronRegistry interface — this service never knows which is active. + * Earlier versions registered one cron job per organization at boot, but + * graphile-worker cron items cannot change while the runner is up, so config + * CRUD would have required a worker restart. A static dispatch cron plus a + * due-check keeps schedule changes live on both queue backends (BullMQ and + * graphile-worker) through the ICronRegistry interface - this service never + * knows which is active. */ -import { db } from '../../database/connection.js'; import { getCronRegistry } from '../../queue/queue-factory.js'; import { hub } from '@logtide/core'; -import type { CronJobDefinition } from '../../queue/abstractions/types.js'; export interface DigestJobPayload { organizationId: string; @@ -19,58 +23,24 @@ export interface DigestJobPayload { frequency: 'daily' | 'weekly'; } -export class DigestScheduler { - /** - * Register all active digest configs as repeating cron jobs. - */ - async registerAllDigests(): Promise { - const configs = await db - .selectFrom('digest_configs') - .select(['id', 'organization_id', 'frequency', 'delivery_hour', 'delivery_day_of_week']) - .where('enabled', '=', true) - .execute(); - - if (configs.length === 0) { - hub.captureLog('info', '[DigestScheduler] No active digest configs found'); - return; - } - - const items: CronJobDefinition[] = configs.map((config) => ({ - task: 'digest-generation', - cronExpression: this.buildCronExpression( - config.frequency as 'daily' | 'weekly', - config.delivery_hour, - config.delivery_day_of_week - ), - payload: { - organizationId: config.organization_id, - digestConfigId: config.id, - frequency: config.frequency, - } satisfies DigestJobPayload, - // Stable identifier per org — prevents duplicate schedules on restart - identifier: `digest:${config.organization_id}`, - })); - - await getCronRegistry('digest-generation').registerCronJobs(items); - hub.captureLog('info', `[DigestScheduler] Registered ${items.length} digest schedule(s)`); - } +export const DIGEST_DISPATCH_CRON = '0 * * * *'; // every hour on the hour +export class DigestScheduler { /** - * Build a standard 5-field cron expression from a digest config. - * - * Daily: "0 8 * * *" — every day at delivery_hour - * Weekly: "0 8 * * 1" — every week on delivery_day_of_week + * Register the hourly digest-dispatch cron job. Called once at worker boot. */ - private buildCronExpression( - frequency: 'daily' | 'weekly', - deliveryHour: number, - deliveryDayOfWeek: number | null - ): string { - if (frequency === 'daily') { - return `0 ${deliveryHour} * * *`; - } - // Weekly — delivery_day_of_week is guaranteed non-null by the DB constraint - return `0 ${deliveryHour} * * ${deliveryDayOfWeek}`; + async registerDispatchCron(): Promise { + await getCronRegistry('digest-dispatch').registerCronJobs([ + { + task: 'digest-dispatch', + cronExpression: DIGEST_DISPATCH_CRON, + payload: {}, + // Stable identifier - prevents duplicate schedules on restart + identifier: 'digest-dispatch', + }, + ]); + + hub.captureLog('info', '[DigestScheduler] Registered hourly digest dispatch cron'); } } diff --git a/packages/backend/src/modules/digests/service.ts b/packages/backend/src/modules/digests/service.ts new file mode 100644 index 00000000..30f99ca3 --- /dev/null +++ b/packages/backend/src/modules/digests/service.ts @@ -0,0 +1,164 @@ +/** + * Digests Service + * + * CRUD for digest_configs (one per organization) and digest_recipients, + * plus token-based unsubscribe used by the public endpoint. + */ + +import crypto from 'crypto'; +import { db } from '../../database/connection.js'; +import type { DigestFrequency } from '../../database/types.js'; + +export interface DigestConfigInput { + frequency: DigestFrequency; + deliveryHour: number; + deliveryDayOfWeek?: number | null; + enabled: boolean; +} + +export class DigestsService { + async getConfig(organizationId: string) { + return db + .selectFrom('digest_configs') + .select([ + 'id', + 'frequency', + 'delivery_hour', + 'delivery_day_of_week', + 'enabled', + 'last_sent_at', + 'created_at', + 'updated_at', + ]) + .where('organization_id', '=', organizationId) + .executeTakeFirst(); + } + + async upsertConfig(organizationId: string, input: DigestConfigInput) { + const values = { + organization_id: organizationId, + frequency: input.frequency, + delivery_hour: input.deliveryHour, + delivery_day_of_week: input.frequency === 'weekly' ? input.deliveryDayOfWeek ?? null : null, + enabled: input.enabled, + }; + + return db + .insertInto('digest_configs') + .values(values) + .onConflict((oc) => + oc.column('organization_id').doUpdateSet({ + frequency: values.frequency, + delivery_hour: values.delivery_hour, + delivery_day_of_week: values.delivery_day_of_week, + enabled: values.enabled, + updated_at: new Date(), + }) + ) + .returning([ + 'id', + 'frequency', + 'delivery_hour', + 'delivery_day_of_week', + 'enabled', + 'last_sent_at', + 'created_at', + 'updated_at', + ]) + .executeTakeFirstOrThrow(); + } + + async deleteConfig(organizationId: string): Promise { + const result = await db + .deleteFrom('digest_configs') + .where('organization_id', '=', organizationId) + .executeTakeFirst(); + return result.numDeletedRows > 0n; + } + + async listRecipients(organizationId: string) { + return db + .selectFrom('digest_recipients') + .select(['id', 'email', 'user_id', 'subscribed', 'created_at']) + .where('organization_id', '=', organizationId) + .orderBy('created_at', 'asc') + .execute(); + } + + async countRecipients(organizationId: string): Promise { + const row = await db + .selectFrom('digest_recipients') + .select((eb) => eb.fn.countAll().as('count')) + .where('organization_id', '=', organizationId) + .executeTakeFirst(); + return Number(row?.count ?? 0); + } + + /** + * Add a recipient to the organization's digest. The digest config must + * exist first; recipients hang off it. + */ + async addRecipient(organizationId: string, email: string, userId?: string | null) { + const config = await this.getConfig(organizationId); + if (!config) { + throw new Error('Digest is not configured for this organization'); + } + + return db + .insertInto('digest_recipients') + .values({ + organization_id: organizationId, + digest_config_id: config.id, + email, + user_id: userId ?? null, + unsubscribe_token: crypto.randomBytes(32).toString('base64url'), + }) + .returning(['id', 'email', 'user_id', 'subscribed', 'created_at']) + .executeTakeFirstOrThrow(); + } + + async removeRecipient(organizationId: string, recipientId: string): Promise { + const result = await db + .deleteFrom('digest_recipients') + .where('organization_id', '=', organizationId) + .where('id', '=', recipientId) + .executeTakeFirst(); + return result.numDeletedRows > 0n; + } + + /** + * Re-subscribe a recipient who unsubscribed. Rotates the unsubscribe token + * so links in previously delivered emails cannot flip the state back. + */ + async resubscribeRecipient(organizationId: string, recipientId: string) { + return db + .updateTable('digest_recipients') + .set({ + subscribed: true, + unsubscribe_token: crypto.randomBytes(32).toString('base64url'), + updated_at: new Date(), + }) + .where('organization_id', '=', organizationId) + .where('id', '=', recipientId) + .returning(['id', 'email', 'user_id', 'subscribed', 'created_at']) + .executeTakeFirst(); + } + + /** + * One-click unsubscribe from an email link. The opaque token IS the + * credential; no authentication involved. + */ + async unsubscribeByToken(token: string): Promise<{ email: string } | null> { + // tenant-scope-ok: public unsubscribe path, the unique random token is the credential + const row = await db + .updateTable('digest_recipients') + .set({ subscribed: false, updated_at: new Date() }) + .where('unsubscribe_token', '=', token) + .returning(['email']) + .executeTakeFirst(); + + return row ?? null; + } +} + +export const digestsService = new DigestsService(); diff --git a/packages/backend/src/queue/jobs/digest-dispatch.ts b/packages/backend/src/queue/jobs/digest-dispatch.ts new file mode 100644 index 00000000..9adb8a4b --- /dev/null +++ b/packages/backend/src/queue/jobs/digest-dispatch.ts @@ -0,0 +1,111 @@ +/** + * Digest Dispatch Job Processor + * + * Runs hourly (single static cron registered at worker boot). Scans the active + * digest configs, decides which ones are due at the current UTC hour and + * enqueues one digest-generation job per due config. + * + * This indirection exists because graphile-worker cron items are fixed at + * runner start: per-org cron jobs cannot be added or removed while the worker + * is running, so config CRUD would need a worker restart to take effect. A + * single static dispatch cron plus a due-check keeps schedule changes live on + * both queue backends. + */ + +import { db } from '../../database/connection.js'; +import { createQueue } from '../connection.js'; +import { hub } from '@logtide/core'; +import type { IJob } from '../abstractions/types.js'; +import type { DigestJobPayload } from '../../modules/digests/scheduler.js'; + +interface DispatchableConfig { + frequency: 'daily' | 'weekly'; + delivery_hour: number; + delivery_day_of_week: number | null; + last_sent_at: Date | null; +} + +// A digest fires at most once per delivery slot; anything sent less than two +// hours ago is a duplicate trigger (worker restart, overlapping cron tick). +const DOUBLE_FIRE_GUARD_MS = 2 * 60 * 60 * 1000; + +export function isConfigDue(config: DispatchableConfig, now: Date): boolean { + if (now.getUTCHours() !== config.delivery_hour) { + return false; + } + + if (config.frequency === 'weekly' && now.getUTCDay() !== config.delivery_day_of_week) { + return false; + } + + if ( + config.last_sent_at && + now.getTime() - new Date(config.last_sent_at).getTime() < DOUBLE_FIRE_GUARD_MS + ) { + return false; + } + + return true; +} + +export async function processDigestDispatch(_job: IJob>): Promise { + const now = new Date(); + + // tenant-scope-ok: system dispatch cron sweeps every org's digest config by design + const configs = await db + .selectFrom('digest_configs') + .select([ + 'id', + 'organization_id', + 'frequency', + 'delivery_hour', + 'delivery_day_of_week', + 'last_sent_at', + ]) + .where('enabled', '=', true) + .execute(); + + const due = configs.filter((config) => + isConfigDue( + { + frequency: config.frequency, + delivery_hour: config.delivery_hour, + delivery_day_of_week: config.delivery_day_of_week, + last_sent_at: config.last_sent_at ? new Date(config.last_sent_at as unknown as string) : null, + }, + now + ) + ); + + if (due.length === 0) { + return; + } + + hub.captureLog('info', `[DigestDispatch] ${due.length} digest(s) due at ${now.toISOString()}`); + + const queue = createQueue('digest-generation'); + const hourKey = now.toISOString().slice(0, 13); // YYYY-MM-DDTHH (UTC) + + for (const config of due) { + await queue.add( + 'digest-generation', + { + organizationId: config.organization_id, + digestConfigId: config.id, + frequency: config.frequency, + }, + { + // Hour-keyed so a duplicate dispatch within the same slot dedupes + jobKey: `digest:${config.id}:${hourKey}`, + removeOnComplete: true, + } + ); + + // tenant-scope-ok: id comes from the enabled-configs sweep above; system cron has no org context + await db + .updateTable('digest_configs') + .set({ last_sent_at: now }) + .where('id', '=', config.id) + .execute(); + } +} diff --git a/packages/backend/src/server.ts b/packages/backend/src/server.ts index 1733622a..ccc9f346 100644 --- a/packages/backend/src/server.ts +++ b/packages/backend/src/server.ts @@ -50,6 +50,8 @@ import { auditLogRoutes, auditLogService } from './modules/audit-log/index.js'; import { bootstrapService } from './modules/bootstrap/index.js'; import { runDataAvailabilityBackfill } from './modules/projects/data-availability-backfill.js'; import { notificationChannelsRoutes } from './modules/notification-channels/index.js'; +import { digestsRoutes } from './modules/digests/routes.js'; +import { digestsPublicRoutes } from './modules/digests/public-routes.js'; import { webhookDeliveriesRoutes } from './modules/webhooks/routes.js'; import internalLoggingPlugin from './plugins/internal-logging-plugin.js'; import { initializeInternalLogging, shutdownInternalLogging } from './utils/internal-logger.js'; @@ -182,6 +184,9 @@ export async function build(opts = {}) { await fastify.register(projectsRoutes, { prefix: '/api/v1/projects' }); await fastify.register(notificationsRoutes, { prefix: '/api/v1/notifications' }); await fastify.register(notificationChannelsRoutes, { prefix: '/api/v1/notification-channels' }); + await fastify.register(digestsRoutes, { prefix: '/api/v1/digests' }); + // Public one-click unsubscribe: the URL token is the credential, no auth (#154) + await fastify.register(digestsPublicRoutes, { prefix: '/api/v1/digests' }); await fastify.register(webhookDeliveriesRoutes, { prefix: '/api/v1/webhooks' }); await fastify.register(onboardingRoutes, { prefix: '/api/v1/onboarding' }); await fastify.register(alertsRoutes, { prefix: '/api/v1/alerts' }); diff --git a/packages/backend/src/tests/lib/email-templates.test.ts b/packages/backend/src/tests/lib/email-templates.test.ts index 0cb9692e..ff9b25df 100644 --- a/packages/backend/src/tests/lib/email-templates.test.ts +++ b/packages/backend/src/tests/lib/email-templates.test.ts @@ -18,6 +18,8 @@ import { generateIncidentEmail, generateSigmaDetectionEmail, generateInvitationEmail, + generateDigestEmail, + type DigestEmailData, } from '../../lib/email-templates.js'; describe('Email Templates - Helpers', () => { @@ -752,3 +754,120 @@ describe('Email Templates - Common Elements', () => { expect(result.html).toContain(''); }); }); + +describe('Email Templates - Digest', () => { + const baseData: DigestEmailData = { + organizationName: 'Acme Corp', + frequency: 'daily', + periodLabel: 'last 24 hours', + logVolume: { current: 15000, previous: 12000, trend: '+3000 (+25.0%)' }, + topErrorServices: [ + { service: 'api-', errorCount: 50, previousCount: 20, delta: 30 }, + { service: 'web', errorCount: 10, previousCount: 15, delta: -5 }, + ], + newErrorGroups: [ + { + exceptionType: 'TypeError', + exceptionMessage: 'Cannot read properties of ', + occurrenceCount: 42, + language: 'nodejs', + }, + ], + security: { + totalDetections: 11, + topRules: [{ ruleTitle: 'Suspicious ', severity: 'high', count: 7 }], + openIncidents: 2, + }, + uptime: { + monitorCount: 3, + overallUptimePct: 99.5, + worstMonitors: [{ name: 'API Health', uptimePct: 98.5 }], + }, + unsubscribeUrl: 'https://app.logtide.dev/unsubscribe?token=tok_123', + dashboardUrl: 'https://app.logtide.dev/dashboard', + }; + + it('renders every section in the HTML body', () => { + const result = generateDigestEmail(baseData); + + expect(result.html).toContain(''); + expect(result.html).toContain('Acme Corp'); + expect(result.html).toContain('15,000'); + expect(result.html).toContain('+3000 (+25.0%)'); + expect(result.html).toContain('TypeError'); + expect(result.html).toContain('42'); + expect(result.html).toContain('Suspicious'); + expect(result.html).toContain('99.5'); + expect(result.html).toContain('98.5'); + expect(result.html).toContain('last 24 hours'); + }); + + it('escapes user-controlled values in HTML', () => { + const result = generateDigestEmail(baseData); + + expect(result.html).not.toContain('api-'); + expect(result.html).toContain('api-<gateway>'); + expect(result.html).not.toContain('Suspicious '); + expect(result.html).toContain('Suspicious <Login>'); + expect(result.html).not.toContain('Cannot read properties of '); + }); + + it('uses the unsubscribe link in the footer instead of channel settings', () => { + const result = generateDigestEmail(baseData); + + expect(result.html).toContain('https://app.logtide.dev/unsubscribe?token=tok_123'); + expect(result.html).toContain('Unsubscribe'); + expect(result.html).not.toContain('/dashboard/settings/channels'); + }); + + it('carries all sections in the plaintext fallback', () => { + const result = generateDigestEmail(baseData); + + expect(result.text).toContain('Acme Corp'); + expect(result.text).toContain('Total logs: 15,000'); + expect(result.text).toContain('Trend: +3000 (+25.0%)'); + expect(result.text).toContain('Previous period: 12,000'); + expect(result.text).toContain('api-'); + expect(result.text).toContain('TypeError'); + expect(result.text).toContain('Suspicious '); + expect(result.text).toContain('Open incidents: 2'); + expect(result.text).toContain('99.5'); + expect(result.text).toContain('https://app.logtide.dev/unsubscribe?token=tok_123'); + }); + + it('shows the quiet-period message when there was no log activity', () => { + const quiet: DigestEmailData = { + ...baseData, + logVolume: { current: 0, previous: 0, trend: 'no change' }, + topErrorServices: [], + newErrorGroups: [], + security: { totalDetections: 0, topRules: [], openIncidents: 0 }, + uptime: null, + }; + + const result = generateDigestEmail(quiet); + + expect(result.html).toContain('No activity during this period'); + expect(result.text).toContain('No activity during this period'); + expect(result.text).toContain('quiet'); + }); + + it('omits the uptime section when uptime is null', () => { + const noUptime: DigestEmailData = { ...baseData, uptime: null }; + + const result = generateDigestEmail(noUptime); + + expect(result.html).not.toContain('Uptime'); + expect(result.text).not.toContain('Uptime'); + }); + + it('labels weekly digests as weekly', () => { + const weekly: DigestEmailData = { ...baseData, frequency: 'weekly', periodLabel: 'last 7 days' }; + + const result = generateDigestEmail(weekly); + + expect(result.html).toContain('Weekly Digest'); + expect(result.text).toContain('Weekly Digest'); + expect(result.text).toContain('last 7 days'); + }); +}); diff --git a/packages/backend/src/tests/modules/capabilities/digest-recipients-limit.test.ts b/packages/backend/src/tests/modules/capabilities/digest-recipients-limit.test.ts new file mode 100644 index 00000000..de8ee9ab --- /dev/null +++ b/packages/backend/src/tests/modules/capabilities/digest-recipients-limit.test.ts @@ -0,0 +1,154 @@ +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; +import Fastify, { FastifyInstance } from 'fastify'; +import crypto from 'crypto'; +import { db } from '../../../database/index.js'; +import { digestsRoutes } from '../../../modules/digests/routes.js'; +import { contextPlugin } from '../../../context/index.js'; +import { capabilities } from '../../../capabilities/index.js'; +import { createTestContext } from '../../helpers/factories.js'; + +async function createTestSession(userId: string) { + const token = crypto.randomBytes(32).toString('hex'); + await db + .insertInto('sessions') + .values({ user_id: userId, token, expires_at: new Date(Date.now() + 24 * 60 * 60 * 1000) }) + .execute(); + return token; +} + +async function insertConfig(organizationId: string) { + return db + .insertInto('digest_configs') + .values({ organization_id: organizationId, frequency: 'daily', delivery_hour: 8 }) + .returningAll() + .executeTakeFirstOrThrow(); +} + +async function insertRecipient(organizationId: string, configId: string, email: string) { + return db + .insertInto('digest_recipients') + .values({ + organization_id: organizationId, + digest_config_id: configId, + email, + unsubscribe_token: crypto.randomBytes(32).toString('base64url'), + }) + .returningAll() + .executeTakeFirstOrThrow(); +} + +describe('digests.max_recipients enforcement', () => { + let app: FastifyInstance; + let orgId: string; + let userId: string; + let token: string; + let configId: string; + + beforeAll(async () => { + app = Fastify(); + await app.register(contextPlugin); + await app.register(digestsRoutes, { prefix: '/api/v1/digests' }); + await app.ready(); + }); + + afterAll(async () => { + await app.close(); + }); + + beforeEach(async () => { + await db.deleteFrom('organization_entitlements').execute(); + await db.deleteFrom('digest_recipients').execute(); + await db.deleteFrom('digest_configs').execute(); + await db.deleteFrom('sessions').execute(); + await db.deleteFrom('organization_members').execute(); + await db.deleteFrom('projects').execute(); + await db.deleteFrom('organizations').execute(); + await db.deleteFrom('users').execute(); + + const ctx = await createTestContext(); + orgId = ctx.organization.id; + userId = ctx.user.id; + token = await createTestSession(userId); + const config = await insertConfig(orgId); + configId = config.id; + capabilities.invalidate(orgId); + }); + + // Case 1: at limit => blocked + it('blocks adding a recipient when the org limit is reached', async () => { + await db + .insertInto('organization_entitlements') + .values({ organization_id: orgId, capability: 'digests.max_recipients', enabled: null, limit_value: 1 }) + .execute(); + capabilities.invalidate(orgId); + + await insertRecipient(orgId, configId, 'first@example.com'); + + const res = await app.inject({ + method: 'POST', + url: `/api/v1/digests/recipients?organizationId=${orgId}`, + headers: { Authorization: `Bearer ${token}` }, + payload: { email: 'second@example.com' }, + }); + + expect(res.statusCode).toBe(403); + const body = JSON.parse(res.body); + expect(body.code).toBe('capability.digests.max_recipients.limit_reached'); + }); + + // Case 2: under limit => passes + it('allows adding a recipient when under the limit', async () => { + await db + .insertInto('organization_entitlements') + .values({ organization_id: orgId, capability: 'digests.max_recipients', enabled: null, limit_value: 2 }) + .execute(); + capabilities.invalidate(orgId); + + await insertRecipient(orgId, configId, 'first@example.com'); + + const res = await app.inject({ + method: 'POST', + url: `/api/v1/digests/recipients?organizationId=${orgId}`, + headers: { Authorization: `Bearer ${token}` }, + payload: { email: 'second@example.com' }, + }); + + expect(res.statusCode).toBe(201); + }); + + // Case 3: unlimited default (no entitlement row) => passes + it('allows adding a recipient when no limit is configured (unlimited default)', async () => { + const res = await app.inject({ + method: 'POST', + url: `/api/v1/digests/recipients?organizationId=${orgId}`, + headers: { Authorization: `Bearer ${token}` }, + payload: { email: 'anyone@example.com' }, + }); + + expect(res.statusCode).toBe(201); + }); + + // Case 4: org isolation => org A at limit does not block org B + it('does not block org B when org A is at the limit', async () => { + await db + .insertInto('organization_entitlements') + .values({ organization_id: orgId, capability: 'digests.max_recipients', enabled: null, limit_value: 1 }) + .execute(); + capabilities.invalidate(orgId); + await insertRecipient(orgId, configId, 'first@example.com'); + + const ctxB = await createTestContext(); + const tokenB = await createTestSession(ctxB.user.id); + await insertConfig(ctxB.organization.id); + capabilities.invalidate(ctxB.organization.id); + + const res = await app.inject({ + method: 'POST', + url: `/api/v1/digests/recipients?organizationId=${ctxB.organization.id}`, + headers: { Authorization: `Bearer ${tokenB}` }, + payload: { email: 'orgb@example.com' }, + }); + + expect(res.statusCode).toBe(201); + }); +}); diff --git a/packages/backend/src/tests/modules/digests/generator-integration.test.ts b/packages/backend/src/tests/modules/digests/generator-integration.test.ts index 3e0e3b40..0e5af4cd 100644 --- a/packages/backend/src/tests/modules/digests/generator-integration.test.ts +++ b/packages/backend/src/tests/modules/digests/generator-integration.test.ts @@ -132,7 +132,178 @@ describe('DigestGeneratorService Integration', () => { expect(trendMatch, 'Email should contain "Trend" metric').toBeTruthy(); expect(trendMatch![1]).toBe(`+${expectedDelta}`); - + expect(trendMatch![2]).toBe(`+${expectedPercentChange}`); }); + + it('should include error groups, security and uptime sections in the digest', async () => { + const user = await db + .insertInto('users') + .values({ + email: 'sections@example.com', + password_hash: 'test_hash', + name: 'Sections User', + }) + .returning('id') + .executeTakeFirstOrThrow(); + + const org = await db + .insertInto('organizations') + .values({ + name: 'Sections Org', + slug: 'sections-org', + owner_id: user.id, + }) + .returningAll() + .executeTakeFirstOrThrow(); + + const project = await db + .insertInto('projects') + .values({ + organization_id: org.id, + name: 'Sections Project', + slug: 'sections-project', + user_id: user.id, + }) + .returning('id') + .executeTakeFirstOrThrow(); + + const config = await db + .insertInto('digest_configs') + .values({ + organization_id: org.id, + frequency: 'daily', + delivery_hour: 8, + }) + .returning('id') + .executeTakeFirstOrThrow(); + + await db + .insertInto('digest_recipients') + .values({ + organization_id: org.id, + digest_config_id: config.id, + email: 'sections-digest@example.com', + unsubscribe_token: 'sections_token', + }) + .execute(); + + const now = new Date(); + const twoHoursAgo = new Date(now.getTime() - 2 * 60 * 60 * 1000); + + // New error group first seen inside the period + await db + .insertInto('error_groups') + .values({ + organization_id: org.id, + project_id: project.id, + fingerprint: 'fp_sections_1', + exception_type: 'TypeError', + exception_message: 'Cannot read properties of undefined', + language: 'nodejs', + occurrence_count: 42, + first_seen: twoHoursAgo, + last_seen: now, + }) + .execute(); + + // Sigma rule + detections inside the period + const rule = await db + .insertInto('sigma_rules') + .values({ + organization_id: org.id, + title: 'Suspicious Login Burst', + level: 'high', + status: 'stable', + logsource: JSON.stringify({ product: 'test' }), + detection: JSON.stringify({ condition: 'selection' }), + }) + .returning('id') + .executeTakeFirstOrThrow(); + + await db + .insertInto('detection_events') + .values([1, 2, 3].map(() => ({ + time: twoHoursAgo, + organization_id: org.id, + project_id: project.id, + sigma_rule_id: rule.id, + log_id: crypto.randomUUID(), + severity: 'high', + rule_title: 'Suspicious Login Burst', + service: 'auth-service', + log_level: 'warn', + log_message: 'failed login', + }))) + .execute(); + + // Open incident + await db + .insertInto('incidents') + .values({ + organization_id: org.id, + project_id: project.id, + title: 'Brute force attempt', + severity: 'high', + status: 'open', + }) + .execute(); + + // Monitor with mixed up/down results in the period + const monitor = await db + .insertInto('monitors') + .values({ + organization_id: org.id, + project_id: project.id, + name: 'API Health', + type: 'http', + target: 'https://example.com/health', + }) + .returning('id') + .executeTakeFirstOrThrow(); + + const results = []; + for (let i = 0; i < 10; i++) { + results.push({ + time: new Date(twoHoursAgo.getTime() + i * 60 * 1000), + monitor_id: monitor.id, + organization_id: org.id, + project_id: project.id, + status: i === 0 ? ('down' as const) : ('up' as const), + }); + } + await db.insertInto('monitor_results').values(results).execute(); + + await generator.generateAndSendDigest({ + organizationId: org.id, + digestConfigId: config.id, + frequency: 'daily', + }); + + expect(mockSendMail).toHaveBeenCalledTimes(1); + const emailCall = mockSendMail.mock.calls[0][0]; + const text = emailCall.text as string; + const html = emailCall.html as string; + + // New error group section + expect(text).toContain('TypeError'); + expect(text).toContain('Cannot read properties of undefined'); + expect(text).toContain('42'); + + // Security section (3 detections, top rule, 1 open incident) + expect(text).toContain('Detections: 3'); + expect(text).toContain('Suspicious Login Burst'); + expect(text).toContain('Open incidents: 1'); + + // Uptime section (9 up / 10 checks = 90%) + expect(text).toContain('UPTIME'); + expect(text).toContain('API Health'); + expect(text).toContain('90%'); + + // HTML version carries the same content and the unsubscribe link + expect(html).toContain(''); + expect(html).toContain('TypeError'); + expect(html).toContain('Suspicious Login Burst'); + expect(html).toContain('unsubscribe?token=sections_token'); + }); }); diff --git a/packages/backend/src/tests/modules/digests/generator.test.ts b/packages/backend/src/tests/modules/digests/generator.test.ts index f11a94ba..b0a7619a 100644 --- a/packages/backend/src/tests/modules/digests/generator.test.ts +++ b/packages/backend/src/tests/modules/digests/generator.test.ts @@ -5,6 +5,7 @@ import type { DigestJobPayload } from '../../../modules/digests/scheduler.js'; vi.mock('../../../database/reservoir.js', () => ({ reservoir: { count: vi.fn(), + topValues: vi.fn(), }, })); @@ -19,6 +20,9 @@ vi.mock('../../../database/connection.js', () => ({ selectFrom: vi.fn().mockReturnThis(), select: vi.fn().mockReturnThis(), where: vi.fn().mockReturnThis(), + groupBy: vi.fn().mockReturnThis(), + orderBy: vi.fn().mockReturnThis(), + limit: vi.fn().mockReturnThis(), execute: vi.fn(), executeTakeFirst: vi.fn(), } @@ -60,9 +64,11 @@ describe('DigestGeneratorService', () => { const { db } = await import('../../../database/connection.js'); mockDb = db; - mockDb.execute.mockReset(); - mockDb.executeTakeFirst.mockReset(); - mockReservoir.count.mockReset(); + // Defaults keep the section queries harmless; tests override with *Once. + mockDb.execute.mockReset().mockResolvedValue([]); + mockDb.executeTakeFirst.mockReset().mockResolvedValue({ count: 0 }); + mockReservoir.count.mockReset().mockResolvedValue({ count: 0 }); + mockReservoir.topValues.mockReset().mockResolvedValue({ values: [] }); mockSendMail.mockReset().mockResolvedValue({ messageId: 'test-message-id' }); generator = new DigestGeneratorService(); @@ -73,6 +79,11 @@ describe('DigestGeneratorService', () => { }); describe('generateAndSendDigest', () => { + /** + * Call order with the shared fluent mock: + * executeTakeFirst: org lookup, then per-section counts (default {count: 0}) + * execute: recipients, projects, then per-section lists (default []) + */ it('should generate and send daily digest with log volume', async () => { const payload: DigestJobPayload = { organizationId: 'org_1', @@ -80,7 +91,6 @@ describe('DigestGeneratorService', () => { frequency: 'daily', }; - // organization lookup mockDb.executeTakeFirst.mockResolvedValueOnce({ name: 'Test Organization', }); @@ -97,25 +107,16 @@ describe('DigestGeneratorService', () => { }, ]); - // Mock projects query + // projects query mockDb.execute.mockResolvedValueOnce([{ id: 'project_1' }]); - const { reservoir } = await import('../../../database/reservoir.js'); - - (reservoir.count as any).mockResolvedValueOnce({ - count: 15000, - }); - - (reservoir.count as any).mockResolvedValueOnce({ - count: 12000, - }); + mockReservoir.count.mockResolvedValueOnce({ count: 15000 }); + mockReservoir.count.mockResolvedValueOnce({ count: 12000 }); await generator.generateAndSendDigest(payload); - // Verify emails were sent expect(mockSendMail).toHaveBeenCalledTimes(2); - // Check first email const firstCall = mockSendMail.mock.calls[0][0]; expect(firstCall.to).toBe('user1@test.com'); expect(firstCall.subject).toContain('Daily Report'); @@ -124,7 +125,6 @@ describe('DigestGeneratorService', () => { expect(firstCall.text).toContain('+3000 (+25.0%)'); expect(firstCall.text).toContain('token_1'); - // Check second email const secondCall = mockSendMail.mock.calls[1][0]; expect(secondCall.to).toBe('user2@test.com'); expect(secondCall.text).toContain('token_2'); @@ -148,18 +148,10 @@ describe('DigestGeneratorService', () => { }, ]); - // Mock projects query mockDb.execute.mockResolvedValueOnce([{ id: 'project_1' }]); - const { reservoir } = await import('../../../database/reservoir.js'); - - (reservoir.count as any).mockResolvedValueOnce({ - count: 100000, - }); - - (reservoir.count as any).mockResolvedValueOnce({ - count: 95000, - }); + mockReservoir.count.mockResolvedValueOnce({ count: 100000 }); + mockReservoir.count.mockResolvedValueOnce({ count: 95000 }); await generator.generateAndSendDigest(payload); @@ -188,13 +180,8 @@ describe('DigestGeneratorService', () => { }, ]); - // Mock projects query mockDb.execute.mockResolvedValueOnce([{ id: 'project_1' }]); - const { reservoir } = await import('../../../database/reservoir.js'); - (reservoir.count as any).mockResolvedValueOnce({ count: 0 }); - (reservoir.count as any).mockResolvedValueOnce({ count: 0 }); - await generator.generateAndSendDigest(payload); expect(mockSendMail).toHaveBeenCalledTimes(1); @@ -221,12 +208,10 @@ describe('DigestGeneratorService', () => { }, ]); - // Mock projects query mockDb.execute.mockResolvedValueOnce([{ id: 'project_1' }]); - const { reservoir } = await import('../../../database/reservoir.js'); - (reservoir.count as any).mockResolvedValueOnce({ count: 8000 }); - (reservoir.count as any).mockResolvedValueOnce({ count: 10000 }); + mockReservoir.count.mockResolvedValueOnce({ count: 8000 }); + mockReservoir.count.mockResolvedValueOnce({ count: 10000 }); await generator.generateAndSendDigest(payload); @@ -254,12 +239,10 @@ describe('DigestGeneratorService', () => { }, ]); - // Mock projects query mockDb.execute.mockResolvedValueOnce([{ id: 'project_1' }]); - const { reservoir } = await import('../../../database/reservoir.js'); - (reservoir.count as any).mockResolvedValueOnce({ count: 5000 }); - (reservoir.count as any).mockResolvedValueOnce({ count: 0 }); + mockReservoir.count.mockResolvedValueOnce({ count: 5000 }); + mockReservoir.count.mockResolvedValueOnce({ count: 0 }); await generator.generateAndSendDigest(payload); @@ -285,7 +268,6 @@ describe('DigestGeneratorService', () => { await generator.generateAndSendDigest(payload); - // Should not send any emails expect(mockSendMail).not.toHaveBeenCalled(); }); @@ -313,6 +295,9 @@ describe('DigestGeneratorService', () => { expect(mockSendMail).toHaveBeenCalledTimes(1); const emailCall = mockSendMail.mock.calls[0][0]; expect(emailCall.text).toContain('No activity during this period'); + // No projects: the reservoir must never be queried + expect(mockReservoir.count).not.toHaveBeenCalled(); + expect(mockReservoir.topValues).not.toHaveBeenCalled(); }); it('should throw error if organization not found', async () => { @@ -322,7 +307,6 @@ describe('DigestGeneratorService', () => { frequency: 'daily', }; - // Organization not found mockDb.executeTakeFirst.mockResolvedValueOnce(undefined); await expect(generator.generateAndSendDigest(payload)).rejects.toThrow( @@ -351,9 +335,8 @@ describe('DigestGeneratorService', () => { ]); mockDb.execute.mockResolvedValueOnce([{ id: 'project_1' }]); - const { reservoir } = await import('../../../database/reservoir.js'); - (reservoir.count as any).mockResolvedValueOnce({ count: 1000 }); - (reservoir.count as any).mockResolvedValueOnce({ count: 900 }); + mockReservoir.count.mockResolvedValueOnce({ count: 1000 }); + mockReservoir.count.mockResolvedValueOnce({ count: 900 }); await generator.generateAndSendDigest(payload); @@ -364,4 +347,151 @@ describe('DigestGeneratorService', () => { expect(emailCall.text).toContain('https://app.logtide.com/unsubscribe?token=secure_token_abc123'); }); }); + + describe('buildReportData', () => { + /** + * Call order inside buildReportData with the shared fluent mock: + * execute: projects, error_groups, top rules, monitors, uptime rows + * executeTakeFirst: detection total, open incidents + * reservoir: count x2, topValues x2 (previous period skipped when current is empty) + */ + it('computes top error services with delta vs previous period', async () => { + mockDb.execute.mockResolvedValueOnce([{ id: 'project_1' }]); // projects + + mockReservoir.topValues.mockResolvedValueOnce({ + values: [ + { value: 'api', count: 50 }, + { value: 'web', count: 10 }, + ], + }); + mockReservoir.topValues.mockResolvedValueOnce({ + values: [{ value: 'api', count: 20 }], + }); + + const report = await generator.buildReportData('org_1', 'Test Org', 'daily'); + + expect(report.topErrorServices).toEqual([ + { service: 'api', errorCount: 50, previousCount: 20, delta: 30 }, + { service: 'web', errorCount: 10, previousCount: 0, delta: 10 }, + ]); + + // error+critical levels, top 5 for current period + const firstCall = mockReservoir.topValues.mock.calls[0][0]; + expect(firstCall.field).toBe('service'); + expect(firstCall.level).toEqual(['error', 'critical']); + expect(firstCall.limit).toBe(5); + }); + + it('skips the previous-period query when there are no current errors', async () => { + mockDb.execute.mockResolvedValueOnce([{ id: 'project_1' }]); + + const report = await generator.buildReportData('org_1', 'Test Org', 'daily'); + + expect(report.topErrorServices).toEqual([]); + expect(mockReservoir.topValues).toHaveBeenCalledTimes(1); + }); + + it('collects new error groups first seen in the period', async () => { + mockDb.execute.mockResolvedValueOnce([{ id: 'project_1' }]); // projects + mockDb.execute.mockResolvedValueOnce([ + { + exception_type: 'TypeError', + exception_message: 'Cannot read properties of undefined', + occurrence_count: 42, + language: 'nodejs', + }, + { + exception_type: 'ValueError', + exception_message: null, + occurrence_count: 3, + language: 'python', + }, + ]); // error_groups + + const report = await generator.buildReportData('org_1', 'Test Org', 'daily'); + + expect(report.newErrorGroups).toEqual([ + { + exceptionType: 'TypeError', + exceptionMessage: 'Cannot read properties of undefined', + occurrenceCount: 42, + language: 'nodejs', + }, + { + exceptionType: 'ValueError', + exceptionMessage: '', + occurrenceCount: 3, + language: 'python', + }, + ]); + }); + + it('computes the security summary', async () => { + mockDb.execute.mockResolvedValueOnce([{ id: 'project_1' }]); // projects + mockDb.execute.mockResolvedValueOnce([]); // error_groups + mockDb.execute.mockResolvedValueOnce([ + { rule_title: 'Suspicious Login', severity: 'high', count: 7 }, + { rule_title: 'Port Scan', severity: 'medium', count: 4 }, + ]); // top rules + + mockDb.executeTakeFirst.mockResolvedValueOnce({ count: 11 }); // detections total + mockDb.executeTakeFirst.mockResolvedValueOnce({ count: 2 }); // open incidents + + const report = await generator.buildReportData('org_1', 'Test Org', 'weekly'); + + expect(report.security).toEqual({ + totalDetections: 11, + topRules: [ + { ruleTitle: 'Suspicious Login', severity: 'high', count: 7 }, + { ruleTitle: 'Port Scan', severity: 'medium', count: 4 }, + ], + openIncidents: 2, + }); + }); + + it('computes uptime summary from monitor daily buckets', async () => { + mockDb.execute.mockResolvedValueOnce([{ id: 'project_1' }]); // projects + mockDb.execute.mockResolvedValueOnce([]); // error_groups + mockDb.execute.mockResolvedValueOnce([]); // top rules + mockDb.execute.mockResolvedValueOnce([ + { id: 'mon_1', name: 'API Health' }, + { id: 'mon_2', name: 'Landing Page' }, + ]); // monitors + mockDb.execute.mockResolvedValueOnce([ + { monitor_id: 'mon_1', successful: 95, total: 100 }, + { monitor_id: 'mon_2', successful: 100, total: 100 }, + ]); // uptime rows + + const report = await generator.buildReportData('org_1', 'Test Org', 'daily'); + + expect(report.uptime).toEqual({ + monitorCount: 2, + overallUptimePct: 97.5, + worstMonitors: [ + { name: 'API Health', uptimePct: 95 }, + { name: 'Landing Page', uptimePct: 100 }, + ], + }); + }); + + it('returns null uptime when the org has no enabled monitors', async () => { + mockDb.execute.mockResolvedValueOnce([{ id: 'project_1' }]); // projects + // error_groups, top rules, monitors all fall through to the [] default + + const report = await generator.buildReportData('org_1', 'Test Org', 'daily'); + + expect(report.uptime).toBeNull(); + }); + + it('labels the period by frequency', async () => { + mockDb.execute.mockResolvedValueOnce([]); // projects (none) + + const daily = await generator.buildReportData('org_1', 'Test Org', 'daily'); + expect(daily.periodLabel).toBe('last 24 hours'); + + mockDb.execute.mockResolvedValueOnce([]); + const weekly = await generator.buildReportData('org_1', 'Test Org', 'weekly'); + expect(weekly.periodLabel).toBe('last 7 days'); + }); + }); }); diff --git a/packages/backend/src/tests/modules/digests/routes.test.ts b/packages/backend/src/tests/modules/digests/routes.test.ts new file mode 100644 index 00000000..7729c74f --- /dev/null +++ b/packages/backend/src/tests/modules/digests/routes.test.ts @@ -0,0 +1,447 @@ +import { describe, it, expect, beforeEach, afterAll, beforeAll } from 'vitest'; +import Fastify, { FastifyInstance } from 'fastify'; +import crypto from 'crypto'; +import { db } from '../../../database/index.js'; +import { digestsRoutes } from '../../../modules/digests/routes.js'; +import { digestsPublicRoutes } from '../../../modules/digests/public-routes.js'; +import { createTestContext, createTestUser } from '../../helpers/factories.js'; + +async function createTestSession(userId: string) { + const token = crypto.randomBytes(32).toString('hex'); + await db + .insertInto('sessions') + .values({ user_id: userId, token, expires_at: new Date(Date.now() + 24 * 60 * 60 * 1000) }) + .execute(); + return token; +} + +describe('Digests Routes', () => { + let app: FastifyInstance; + let authToken: string; + let testUser: any; + let testOrganization: any; + + beforeAll(async () => { + app = Fastify(); + await app.register(digestsRoutes, { prefix: '/api/v1/digests' }); + await app.register(digestsPublicRoutes, { prefix: '/api/v1/digests' }); + await app.ready(); + }); + + afterAll(async () => { + await app.close(); + }); + + beforeEach(async () => { + await db.deleteFrom('digest_recipients').execute(); + await db.deleteFrom('digest_configs').execute(); + await db.deleteFrom('organization_members').execute(); + await db.deleteFrom('projects').execute(); + await db.deleteFrom('organizations').execute(); + await db.deleteFrom('sessions').execute(); + await db.deleteFrom('users').execute(); + + const context = await createTestContext(); + testUser = context.user; + testOrganization = context.organization; + + authToken = await createTestSession(testUser.id); + }); + + async function insertConfig(organizationId: string, overrides: Record = {}) { + return db + .insertInto('digest_configs') + .values({ + organization_id: organizationId, + frequency: 'daily', + delivery_hour: 8, + ...overrides, + }) + .returningAll() + .executeTakeFirstOrThrow(); + } + + async function insertRecipient( + organizationId: string, + configId: string, + email = 'someone@example.com' + ) { + return db + .insertInto('digest_recipients') + .values({ + organization_id: organizationId, + digest_config_id: configId, + email, + unsubscribe_token: crypto.randomBytes(32).toString('base64url'), + }) + .returningAll() + .executeTakeFirstOrThrow(); + } + + describe('GET /api/v1/digests/config', () => { + it('returns null config when the org has none', async () => { + const res = await app.inject({ + method: 'GET', + url: `/api/v1/digests/config?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + }); + + expect(res.statusCode).toBe(200); + const body = JSON.parse(res.body); + expect(body.config).toBeNull(); + expect(body.recipients).toEqual([]); + }); + + it('returns the config with its recipients', async () => { + const config = await insertConfig(testOrganization.id); + await insertRecipient(testOrganization.id, config.id); + + const res = await app.inject({ + method: 'GET', + url: `/api/v1/digests/config?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + }); + + expect(res.statusCode).toBe(200); + const body = JSON.parse(res.body); + expect(body.config.frequency).toBe('daily'); + expect(body.config.delivery_hour).toBe(8); + expect(body.recipients).toHaveLength(1); + expect(body.recipients[0].email).toBe('someone@example.com'); + // The unsubscribe token must never leak through the management API + expect(body.recipients[0].unsubscribe_token).toBeUndefined(); + }); + + it('rejects non-members', async () => { + const outsider = await createTestUser({ email: 'outsider@example.com' }); + const outsiderToken = await createTestSession(outsider.id); + + const res = await app.inject({ + method: 'GET', + url: `/api/v1/digests/config?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${outsiderToken}` }, + }); + + expect(res.statusCode).toBe(403); + }); + + it('rejects unauthenticated requests', async () => { + const res = await app.inject({ + method: 'GET', + url: `/api/v1/digests/config?organizationId=${testOrganization.id}`, + }); + + expect(res.statusCode).toBe(401); + }); + }); + + describe('PUT /api/v1/digests/config', () => { + it('creates the config on first save', async () => { + const res = await app.inject({ + method: 'PUT', + url: `/api/v1/digests/config?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + payload: { frequency: 'daily', deliveryHour: 9, enabled: true }, + }); + + expect(res.statusCode).toBe(200); + const body = JSON.parse(res.body); + expect(body.config.frequency).toBe('daily'); + expect(body.config.delivery_hour).toBe(9); + expect(body.config.enabled).toBe(true); + }); + + it('updates the existing config (upsert)', async () => { + await insertConfig(testOrganization.id, { delivery_hour: 8 }); + + const res = await app.inject({ + method: 'PUT', + url: `/api/v1/digests/config?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + payload: { frequency: 'weekly', deliveryHour: 18, deliveryDayOfWeek: 1, enabled: false }, + }); + + expect(res.statusCode).toBe(200); + const body = JSON.parse(res.body); + expect(body.config.frequency).toBe('weekly'); + expect(body.config.delivery_hour).toBe(18); + expect(body.config.delivery_day_of_week).toBe(1); + expect(body.config.enabled).toBe(false); + + const rows = await db.selectFrom('digest_configs').selectAll().execute(); + expect(rows).toHaveLength(1); + }); + + it('rejects weekly frequency without a day of week', async () => { + const res = await app.inject({ + method: 'PUT', + url: `/api/v1/digests/config?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + payload: { frequency: 'weekly', deliveryHour: 8, enabled: true }, + }); + + expect(res.statusCode).toBe(400); + }); + + it('rejects an out-of-range delivery hour', async () => { + const res = await app.inject({ + method: 'PUT', + url: `/api/v1/digests/config?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + payload: { frequency: 'daily', deliveryHour: 24, enabled: true }, + }); + + expect(res.statusCode).toBe(400); + }); + + it('rejects members without admin role', async () => { + const member = await createTestUser({ email: 'member@example.com' }); + await db + .insertInto('organization_members') + .values({ organization_id: testOrganization.id, user_id: member.id, role: 'member' }) + .execute(); + const memberToken = await createTestSession(member.id); + + const res = await app.inject({ + method: 'PUT', + url: `/api/v1/digests/config?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${memberToken}` }, + payload: { frequency: 'daily', deliveryHour: 8, enabled: true }, + }); + + expect(res.statusCode).toBe(403); + }); + }); + + describe('DELETE /api/v1/digests/config', () => { + it('deletes the config and cascades to recipients', async () => { + const config = await insertConfig(testOrganization.id); + await insertRecipient(testOrganization.id, config.id); + + const res = await app.inject({ + method: 'DELETE', + url: `/api/v1/digests/config?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + }); + + expect(res.statusCode).toBe(204); + const recipients = await db.selectFrom('digest_recipients').selectAll().execute(); + expect(recipients).toHaveLength(0); + }); + + it('returns 404 when there is no config', async () => { + const res = await app.inject({ + method: 'DELETE', + url: `/api/v1/digests/config?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + }); + + expect(res.statusCode).toBe(404); + }); + }); + + describe('POST /api/v1/digests/recipients', () => { + it('adds a recipient with a generated unsubscribe token', async () => { + await insertConfig(testOrganization.id); + + const res = await app.inject({ + method: 'POST', + url: `/api/v1/digests/recipients?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + payload: { email: 'new@example.com' }, + }); + + expect(res.statusCode).toBe(201); + const body = JSON.parse(res.body); + expect(body.recipient.email).toBe('new@example.com'); + expect(body.recipient.subscribed).toBe(true); + expect(body.recipient.unsubscribe_token).toBeUndefined(); + + const row = await db + .selectFrom('digest_recipients') + .selectAll() + .where('email', '=', 'new@example.com') + .executeTakeFirstOrThrow(); + expect(row.unsubscribe_token.length).toBeGreaterThanOrEqual(40); + }); + + it('rejects adding a recipient before the digest is configured', async () => { + const res = await app.inject({ + method: 'POST', + url: `/api/v1/digests/recipients?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + payload: { email: 'new@example.com' }, + }); + + expect(res.statusCode).toBe(409); + }); + + it('rejects duplicate emails', async () => { + const config = await insertConfig(testOrganization.id); + await insertRecipient(testOrganization.id, config.id, 'dup@example.com'); + + const res = await app.inject({ + method: 'POST', + url: `/api/v1/digests/recipients?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + payload: { email: 'dup@example.com' }, + }); + + expect(res.statusCode).toBe(409); + }); + + it('rejects invalid emails', async () => { + await insertConfig(testOrganization.id); + + const res = await app.inject({ + method: 'POST', + url: `/api/v1/digests/recipients?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + payload: { email: 'not-an-email' }, + }); + + expect(res.statusCode).toBe(400); + }); + }); + + describe('DELETE /api/v1/digests/recipients/:id', () => { + it('removes a recipient', async () => { + const config = await insertConfig(testOrganization.id); + const recipient = await insertRecipient(testOrganization.id, config.id); + + const res = await app.inject({ + method: 'DELETE', + url: `/api/v1/digests/recipients/${recipient.id}?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + }); + + expect(res.statusCode).toBe(204); + }); + + it('cannot remove a recipient of another organization', async () => { + // org B with its own digest + recipient + const ctxB = await createTestContext(); + const configB = await insertConfig(ctxB.organization.id); + const recipientB = await insertRecipient(ctxB.organization.id, configB.id, 'orgb@example.com'); + + // org A admin tries to delete org B's recipient through their own org scope + const res = await app.inject({ + method: 'DELETE', + url: `/api/v1/digests/recipients/${recipientB.id}?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + }); + + expect(res.statusCode).toBe(404); + const stillThere = await db + .selectFrom('digest_recipients') + .selectAll() + .where('id', '=', recipientB.id) + .executeTakeFirst(); + expect(stillThere).toBeDefined(); + }); + }); + + describe('POST /api/v1/digests/recipients/:id/resubscribe', () => { + it('resubscribes and rotates the unsubscribe token', async () => { + const config = await insertConfig(testOrganization.id); + const recipient = await insertRecipient(testOrganization.id, config.id); + await db + .updateTable('digest_recipients') + .set({ subscribed: false }) + .where('id', '=', recipient.id) + .execute(); + + const res = await app.inject({ + method: 'POST', + url: `/api/v1/digests/recipients/${recipient.id}/resubscribe?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + }); + + expect(res.statusCode).toBe(200); + const body = JSON.parse(res.body); + expect(body.recipient.subscribed).toBe(true); + + const row = await db + .selectFrom('digest_recipients') + .selectAll() + .where('id', '=', recipient.id) + .executeTakeFirstOrThrow(); + expect(row.subscribed).toBe(true); + expect(row.unsubscribe_token).not.toBe(recipient.unsubscribe_token); + }); + + it('returns 404 for unknown recipients', async () => { + const res = await app.inject({ + method: 'POST', + url: `/api/v1/digests/recipients/${crypto.randomUUID()}/resubscribe?organizationId=${testOrganization.id}`, + headers: { Authorization: `Bearer ${authToken}` }, + }); + + expect(res.statusCode).toBe(404); + }); + }); + + describe('POST /api/v1/digests/unsubscribe (public)', () => { + it('unsubscribes by token without authentication', async () => { + const config = await insertConfig(testOrganization.id); + const recipient = await insertRecipient(testOrganization.id, config.id, 'giuseppe@example.com'); + + const res = await app.inject({ + method: 'POST', + url: '/api/v1/digests/unsubscribe', + payload: { token: recipient.unsubscribe_token }, + }); + + expect(res.statusCode).toBe(200); + const body = JSON.parse(res.body); + expect(body.success).toBe(true); + expect(body.email).toBe('g***@example.com'); + + const row = await db + .selectFrom('digest_recipients') + .selectAll() + .where('id', '=', recipient.id) + .executeTakeFirstOrThrow(); + expect(row.subscribed).toBe(false); + }); + + it('is idempotent', async () => { + const config = await insertConfig(testOrganization.id); + const recipient = await insertRecipient(testOrganization.id, config.id); + + const first = await app.inject({ + method: 'POST', + url: '/api/v1/digests/unsubscribe', + payload: { token: recipient.unsubscribe_token }, + }); + const second = await app.inject({ + method: 'POST', + url: '/api/v1/digests/unsubscribe', + payload: { token: recipient.unsubscribe_token }, + }); + + expect(first.statusCode).toBe(200); + expect(second.statusCode).toBe(200); + }); + + it('returns 404 for unknown tokens', async () => { + const res = await app.inject({ + method: 'POST', + url: '/api/v1/digests/unsubscribe', + payload: { token: 'nope' }, + }); + + expect(res.statusCode).toBe(404); + }); + + it('rejects requests without a token', async () => { + const res = await app.inject({ + method: 'POST', + url: '/api/v1/digests/unsubscribe', + payload: {}, + }); + + expect(res.statusCode).toBe(400); + }); + }); +}); diff --git a/packages/backend/src/tests/modules/digests/scheduler.test.ts b/packages/backend/src/tests/modules/digests/scheduler.test.ts index e237dbaf..86891e09 100644 --- a/packages/backend/src/tests/modules/digests/scheduler.test.ts +++ b/packages/backend/src/tests/modules/digests/scheduler.test.ts @@ -1,6 +1,5 @@ import { describe, it, expect, vi, beforeEach } from 'vitest'; -import { DigestScheduler, digestScheduler } from '../../../modules/digests/scheduler.js'; -import { db } from '../../../database/connection.js'; +import { DigestScheduler, DIGEST_DISPATCH_CRON } from '../../../modules/digests/scheduler.js'; import { getCronRegistry } from '../../../queue/queue-factory.js'; vi.mock('@logtide/core', () => ({ @@ -9,23 +8,11 @@ vi.mock('@logtide/core', () => ({ }, })); - -vi.mock('../../../database/connection.js', () => { - return { - db: { - selectFrom: vi.fn().mockReturnThis(), - select: vi.fn().mockReturnThis(), - where: vi.fn().mockReturnThis(), - execute: vi.fn(), - }, - }; -}); - - +const mockRegisterCronJobs = vi.fn().mockResolvedValue(undefined); vi.mock('../../../queue/queue-factory.js', () => { return { getCronRegistry: vi.fn().mockReturnValue({ - registerCronJobs: vi.fn().mockResolvedValue(undefined), + registerCronJobs: (...args: unknown[]) => mockRegisterCronJobs(...args), }), }; }); @@ -35,73 +22,26 @@ describe('DigestScheduler', () => { vi.clearAllMocks(); }); - describe('registerAllDigests', () => { - it('should log and return early if no active configs are found', async () => { - const { hub } = await import('@logtide/core'); - ((db as any).execute as ReturnType).mockResolvedValue([]); - - const scheduler = new DigestScheduler(); - await scheduler.registerAllDigests(); - - expect(hub.captureLog).toHaveBeenCalledWith('info', '[DigestScheduler] No active digest configs found'); - expect(getCronRegistry('digest-generation').registerCronJobs).not.toHaveBeenCalled(); - }); - - it('should correctly register a daily and a weekly cron job', async () => { - const mockConfigs = [ - { - id: 'conf_1', - organization_id: 'org_1', - frequency: 'daily', - delivery_hour: 8, - delivery_day_of_week: null - }, - { - id: 'conf_2', - organization_id: 'org_2', - frequency: 'weekly', - delivery_hour: 14, - delivery_day_of_week: 1 - } - ]; - - ((db as any).execute as ReturnType).mockResolvedValue(mockConfigs); - const { hub } = await import('@logtide/core'); - + describe('registerDispatchCron', () => { + it('registers a single hourly dispatch cron job', async () => { const scheduler = new DigestScheduler(); - await scheduler.registerAllDigests(); - - const expectedItems = [ - { - task: 'digest-generation', - cronExpression: '0 8 * * *', - payload: { - organizationId: 'org_1', - digestConfigId: 'conf_1', - frequency: 'daily' - }, - identifier: 'digest:org_1' - }, - { - task: 'digest-generation', - cronExpression: '0 14 * * 1', - payload: { - organizationId: 'org_2', - digestConfigId: 'conf_2', - frequency: 'weekly' - }, - identifier: 'digest:org_2' - } - ]; - - expect(getCronRegistry('digest-generation').registerCronJobs).toHaveBeenCalledWith(expectedItems); - expect(hub.captureLog).toHaveBeenCalledWith('info', '[DigestScheduler] Registered 2 digest schedule(s)'); + await scheduler.registerDispatchCron(); + + expect(getCronRegistry).toHaveBeenCalledWith('digest-dispatch'); + expect(mockRegisterCronJobs).toHaveBeenCalledTimes(1); + + const items = mockRegisterCronJobs.mock.calls[0][0]; + expect(items).toHaveLength(1); + expect(items[0]).toEqual({ + task: 'digest-dispatch', + cronExpression: DIGEST_DISPATCH_CRON, + payload: {}, + identifier: 'digest-dispatch', + }); }); - }); - describe('export instance', () => { - it('should export a singleton instance', () => { - expect(digestScheduler).toBeInstanceOf(DigestScheduler); + it('uses an hourly cron expression', () => { + expect(DIGEST_DISPATCH_CRON).toBe('0 * * * *'); }); }); }); diff --git a/packages/backend/src/tests/queue/jobs/digest-dispatch.test.ts b/packages/backend/src/tests/queue/jobs/digest-dispatch.test.ts new file mode 100644 index 00000000..9bead0e0 --- /dev/null +++ b/packages/backend/src/tests/queue/jobs/digest-dispatch.test.ts @@ -0,0 +1,142 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import type { IJob } from '../../../queue/abstractions/types.js'; + +vi.mock('@logtide/core', () => ({ + hub: { + captureLog: vi.fn(), + }, +})); + +const mockAdd = vi.fn().mockResolvedValue({ id: 'job_1' }); +vi.mock('../../../queue/connection.js', () => ({ + createQueue: vi.fn(() => ({ + add: mockAdd, + })), +})); + +const mockExecute = vi.fn(); +const mockUpdateExecute = vi.fn().mockResolvedValue(undefined); +vi.mock('../../../database/connection.js', () => ({ + db: { + selectFrom: vi.fn().mockReturnThis(), + select: vi.fn().mockReturnThis(), + updateTable: vi.fn().mockReturnThis(), + set: vi.fn().mockReturnThis(), + where: vi.fn().mockReturnThis(), + execute: (...args: unknown[]) => mockExecute(...args), + }, +})); + +import { isConfigDue, processDigestDispatch } from '../../../queue/jobs/digest-dispatch.js'; +import { db } from '../../../database/connection.js'; + +function makeJob(): IJob> { + return { id: 'dispatch_1', name: 'digest-dispatch', data: {} }; +} + +describe('isConfigDue', () => { + // 2026-07-13 is a Monday; 08:30 UTC + const mondayAt8 = new Date('2026-07-13T08:30:00.000Z'); + + const baseDaily = { + frequency: 'daily' as const, + delivery_hour: 8, + delivery_day_of_week: null, + last_sent_at: null, + }; + + it('daily config is due when the current UTC hour matches', () => { + expect(isConfigDue(baseDaily, mondayAt8)).toBe(true); + }); + + it('daily config is not due on a different hour', () => { + expect(isConfigDue({ ...baseDaily, delivery_hour: 9 }, mondayAt8)).toBe(false); + }); + + it('weekly config requires matching UTC day of week', () => { + const weekly = { ...baseDaily, frequency: 'weekly' as const, delivery_day_of_week: 1 }; + expect(isConfigDue(weekly, mondayAt8)).toBe(true); + expect(isConfigDue({ ...weekly, delivery_day_of_week: 2 }, mondayAt8)).toBe(false); + }); + + it('is not due again when last_sent_at is within the double-fire guard', () => { + const sentRecently = { ...baseDaily, last_sent_at: new Date('2026-07-13T08:05:00.000Z') }; + expect(isConfigDue(sentRecently, mondayAt8)).toBe(false); + }); + + it('is due when last_sent_at is older than the guard window', () => { + const sentYesterday = { ...baseDaily, last_sent_at: new Date('2026-07-12T08:30:00.000Z') }; + expect(isConfigDue(sentYesterday, mondayAt8)).toBe(true); + }); +}); + +describe('processDigestDispatch', () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.useRealTimers(); + }); + + it('enqueues generation jobs only for due configs and stamps last_sent_at', async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-07-13T08:00:30.000Z')); + + mockExecute.mockResolvedValueOnce([ + { + id: 'conf_due', + organization_id: 'org_1', + frequency: 'daily', + delivery_hour: 8, + delivery_day_of_week: null, + last_sent_at: null, + }, + { + id: 'conf_not_due', + organization_id: 'org_2', + frequency: 'daily', + delivery_hour: 20, + delivery_day_of_week: null, + last_sent_at: null, + }, + ]); + // updateTable(...).execute() resolves via the same mockExecute + mockExecute.mockResolvedValue(undefined); + + await processDigestDispatch(makeJob()); + + expect(mockAdd).toHaveBeenCalledTimes(1); + const [jobName, payload, options] = mockAdd.mock.calls[0]; + expect(jobName).toBe('digest-generation'); + expect(payload).toEqual({ + organizationId: 'org_1', + digestConfigId: 'conf_due', + frequency: 'daily', + }); + // hour-keyed jobKey dedupes double enqueues within the same hour + expect(options.jobKey).toBe('digest:conf_due:2026-07-13T08'); + + expect(db.updateTable).toHaveBeenCalledWith('digest_configs'); + vi.useRealTimers(); + }); + + it('does nothing when no config is due', async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-07-13T03:00:00.000Z')); + + mockExecute.mockResolvedValueOnce([ + { + id: 'conf_1', + organization_id: 'org_1', + frequency: 'daily', + delivery_hour: 8, + delivery_day_of_week: null, + last_sent_at: null, + }, + ]); + + await processDigestDispatch(makeJob()); + + expect(mockAdd).not.toHaveBeenCalled(); + expect(db.updateTable).not.toHaveBeenCalled(); + vi.useRealTimers(); + }); +}); diff --git a/packages/backend/src/worker.ts b/packages/backend/src/worker.ts index e7efc061..b58efb5b 100644 --- a/packages/backend/src/worker.ts +++ b/packages/backend/src/worker.ts @@ -25,8 +25,8 @@ import { enrichmentService } from './modules/siem/enrichment-service.js'; import { retentionService } from './modules/retention/index.js'; import { auditLogService } from './modules/audit-log/index.js'; import { sigmaSyncService } from './modules/sigma/sync-service.js'; -// Email digest feature is incomplete and disabled for 1.0.0-beta (see #154 below). -// import { digestScheduler } from './modules/digests/scheduler.js'; +import { digestScheduler } from './modules/digests/scheduler.js'; +import { processDigestDispatch } from './queue/jobs/digest-dispatch.js'; import { initializeWorkerLogging, shutdownInternalLogging, isInternalLoggingEnabled } from './utils/internal-logger.js'; import { hub } from '@logtide/core'; import { reservoir, reservoirReady } from './database/reservoir.js'; @@ -92,12 +92,13 @@ const pipelineWorker = createWorker('log-pipeline', async (j const digestWorker = createWorker('digest-generation', async (job) => { await processDigestGeneration(job); }); -// The email digest reports feature (#154) is incomplete: it only summarizes log -// volume (no error groups, security or uptime sections, no HTML layout) and has no -// UI to configure it. The scheduler is intentionally left unregistered so no partial -// digests are ever sent in the 1.0.0-beta release. The worker, service, database -// schema and tests stay in place so the remaining scope can be finished under #154. -// await digestScheduler.registerAllDigests(); +// Hourly dispatch cron scans digest configs and enqueues due digests (#154). +// A single static cron (instead of one per org) keeps config CRUD live on both +// queue backends: graphile-worker cron items cannot change while the runner is up. +const digestDispatchWorker = createWorker>('digest-dispatch', async (job) => { + await processDigestDispatch(job); +}); +await digestScheduler.registerDispatchCron(); // Create worker for outbound webhook delivery (#218) const webhookDeliveryWorker = createWorker('webhook-delivery', async (job) => { @@ -335,6 +336,15 @@ digestWorker.on('failed', (job, err) => { } }); +digestDispatchWorker.on('failed', (job, err) => { + if (isInternalLoggingEnabled()) { + hub.captureLog('error', `Digest dispatch job failed: ${err.message}`, { + error: { name: err.name, message: err.message, stack: err.stack }, + jobId: job?.id, + }); + } +}); + webhookDeliveryWorker.on('failed', (job, err) => { console.error(`Webhook delivery job ${job?.id} failed:`, err); if (isInternalLoggingEnabled()) { @@ -912,6 +922,7 @@ async function gracefulShutdown(signal: string) { await monitorNotificationWorker.close(); await pipelineWorker.close(); await digestWorker.close(); + await digestDispatchWorker.close(); await webhookDeliveryWorker.close(); await receiverEventsWorker.close(); console.log('[Worker] Workers closed'); diff --git a/packages/frontend/src/lib/api/digests.ts b/packages/frontend/src/lib/api/digests.ts new file mode 100644 index 00000000..8d3c112b --- /dev/null +++ b/packages/frontend/src/lib/api/digests.ts @@ -0,0 +1,122 @@ +import { getApiUrl } from '$lib/config'; +import { getAuthToken } from '$lib/utils/auth'; + +export type DigestFrequency = 'daily' | 'weekly'; + +export interface DigestConfig { + id: string; + frequency: DigestFrequency; + delivery_hour: number; + delivery_day_of_week: number | null; + enabled: boolean; + last_sent_at: string | null; + created_at: string; + updated_at: string; +} + +export interface DigestRecipient { + id: string; + email: string; + user_id: string | null; + subscribed: boolean; + created_at: string; +} + +export interface DigestConfigInput { + frequency: DigestFrequency; + deliveryHour: number; + deliveryDayOfWeek?: number | null; + enabled: boolean; +} + +async function fetchWithAuth(url: string, options: RequestInit = {}): Promise { + const token = getAuthToken(); + const headers: HeadersInit = { + ...(options.body ? { 'Content-Type': 'application/json' } : {}), + ...(token ? { Authorization: `Bearer ${token}` } : {}), + ...(options.headers || {}), + }; + + return fetch(url, { + ...options, + headers, + credentials: 'include', + }); +} + +async function unwrapError(response: Response, fallback: string): Promise { + const error = await response.json().catch(() => ({ error: fallback })); + throw new Error(error.error || fallback); +} + +export const digestsAPI = { + async getConfig( + organizationId: string + ): Promise<{ config: DigestConfig | null; recipients: DigestRecipient[] }> { + const response = await fetchWithAuth( + `${getApiUrl()}/api/v1/digests/config?organizationId=${organizationId}` + ); + if (!response.ok) await unwrapError(response, 'Failed to load digest settings'); + return response.json(); + }, + + async saveConfig(organizationId: string, input: DigestConfigInput): Promise { + const response = await fetchWithAuth( + `${getApiUrl()}/api/v1/digests/config?organizationId=${organizationId}`, + { method: 'PUT', body: JSON.stringify(input) } + ); + if (!response.ok) await unwrapError(response, 'Failed to save digest settings'); + return (await response.json()).config; + }, + + async deleteConfig(organizationId: string): Promise { + const response = await fetchWithAuth( + `${getApiUrl()}/api/v1/digests/config?organizationId=${organizationId}`, + { method: 'DELETE' } + ); + if (!response.ok) await unwrapError(response, 'Failed to delete digest settings'); + }, + + async addRecipient(organizationId: string, email: string): Promise { + const response = await fetchWithAuth( + `${getApiUrl()}/api/v1/digests/recipients?organizationId=${organizationId}`, + { method: 'POST', body: JSON.stringify({ email }) } + ); + if (!response.ok) await unwrapError(response, 'Failed to add recipient'); + return (await response.json()).recipient; + }, + + async removeRecipient(organizationId: string, recipientId: string): Promise { + const response = await fetchWithAuth( + `${getApiUrl()}/api/v1/digests/recipients/${recipientId}?organizationId=${organizationId}`, + { method: 'DELETE' } + ); + if (!response.ok) await unwrapError(response, 'Failed to remove recipient'); + }, + + async resubscribeRecipient( + organizationId: string, + recipientId: string + ): Promise { + const response = await fetchWithAuth( + `${getApiUrl()}/api/v1/digests/recipients/${recipientId}/resubscribe?organizationId=${organizationId}`, + { method: 'POST' } + ); + if (!response.ok) await unwrapError(response, 'Failed to resubscribe recipient'); + return (await response.json()).recipient; + }, + + /** + * Public one-click unsubscribe. No auth: the token from the email link is + * the credential. + */ + async unsubscribe(token: string): Promise<{ success: boolean; email: string }> { + const response = await fetch(`${getApiUrl()}/api/v1/digests/unsubscribe`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ token }), + }); + if (!response.ok) await unwrapError(response, 'Invalid or expired unsubscribe link'); + return response.json(); + }, +}; diff --git a/packages/frontend/src/routes/dashboard/settings/+layout.svelte b/packages/frontend/src/routes/dashboard/settings/+layout.svelte index 803130dc..bee1d353 100644 --- a/packages/frontend/src/routes/dashboard/settings/+layout.svelte +++ b/packages/frontend/src/routes/dashboard/settings/+layout.svelte @@ -15,6 +15,7 @@ import ChevronRight from '@lucide/svelte/icons/chevron-right'; import BarChart3 from '@lucide/svelte/icons/bar-chart-3'; import Send from '@lucide/svelte/icons/send'; + import Newspaper from '@lucide/svelte/icons/newspaper'; interface Props { children?: import('svelte').Snippet; @@ -79,6 +80,7 @@ label: 'Notifications', items: [ { label: 'Channels', href: '/dashboard/settings/channels', icon: BellRing }, + { label: 'Email Digests', href: '/dashboard/settings/digests', icon: Newspaper }, { label: 'Webhook Deliveries', href: '/dashboard/settings/webhooks', icon: Send }, ], }, diff --git a/packages/frontend/src/routes/dashboard/settings/digests/+page.svelte b/packages/frontend/src/routes/dashboard/settings/digests/+page.svelte new file mode 100644 index 00000000..f71dcd21 --- /dev/null +++ b/packages/frontend/src/routes/dashboard/settings/digests/+page.svelte @@ -0,0 +1,349 @@ + + + + Email Digests - Settings - LogTide + + +
+
+

Email Digests

+

+ Periodic email summaries of log activity, errors, security detections and uptime. +

+
+ + {#if loading} +

Loading digest settings...

+ {:else} + + + Schedule + + When the digest report is generated and sent. Times are in UTC. + + + +
{ + e.preventDefault(); + saveConfig(); + }} + class="space-y-4" + > +
+
+ +

+ Subscribed recipients receive the report on the schedule below. +

+
+ (enabled = v)} disabled={!canManage || saving} /> +
+ + + +
+
+ + +
+ +
+ + +
+ + {#if frequency === 'weekly'} +
+ + +
+ {/if} +
+ + {#if config?.last_sent_at} +

+ Last sent: {new Date(config.last_sent_at).toLocaleString()} +

+ {/if} + + + +
+ + + + + Recipients + + Who receives the digest. Each email includes its own one-click unsubscribe link. + + + + {#if !config} +

+ Save the schedule first, then add recipients. +

+ {:else} + {#if canManage} +
{ + e.preventDefault(); + addRecipient(); + }} + class="flex gap-2" + > + + +
+ {/if} + + {#if recipients.length === 0} +

No recipients yet.

+ {:else} +
+ {#each recipients as recipient (recipient.id)} +
+
+ + {recipient.email} + {#if recipient.subscribed} + Subscribed + {:else} + Unsubscribed + {/if} +
+ {#if canManage} +
+ {#if !recipient.subscribed} + + {/if} + +
+ {/if} +
+ {/each} +
+ {/if} + {/if} + + + {/if} +
diff --git a/packages/frontend/src/routes/unsubscribe/+page.svelte b/packages/frontend/src/routes/unsubscribe/+page.svelte new file mode 100644 index 00000000..b9497c5b --- /dev/null +++ b/packages/frontend/src/routes/unsubscribe/+page.svelte @@ -0,0 +1,76 @@ + + + + Unsubscribe - LogTide + + +
+ + +
+ {#if status === 'working'} + + {:else if status === 'done'} + + {:else} + + {/if} +
+ + {#if status === 'working'} + Unsubscribing... + {:else if status === 'done'} + You are unsubscribed + {:else} + Something went wrong + {/if} + + + {#if status === 'working'} + One moment while we update your preferences. + {:else if status === 'done'} + {maskedEmail} will no longer receive digest reports from this organization. + {:else} + {errorMessage} + {/if} + + + + {#if status === 'done'} + Changed your mind? An organization admin can resubscribe you from the digest settings. + {:else if status === 'error'} + If you keep receiving digests, ask an organization admin to remove your address. + {/if} + + +