diff --git a/alembic/versions/021_report_schedules.py b/alembic/versions/021_report_schedules.py new file mode 100644 index 0000000..1044d60 --- /dev/null +++ b/alembic/versions/021_report_schedules.py @@ -0,0 +1,58 @@ +"""Scheduled reports: a supplier risk digest emailed to members on a schedule. + +The digest lists the organization's new matches in the period: public records naming its +suppliers and disaster exposures, as recorded by the webhook detector in webhook_events +(migration 020), whether or not the organization uses webhooks. + +next_run_at drives the runner: a coordinator claims due schedules with FOR UPDATE SKIP +LOCKED and moves next_run_at forward before sending, so two replicas never send the same +run. Times are UTC. Recipients are checked against the organization's members when saved +and again when sent. Tenant table with forced row-level security. + +Revision ID: 021 +Revises: 020 +Create Date: 2026-10-23 +""" +from typing import Sequence, Union + +from alembic import op + +revision: str = "021" +down_revision: Union[str, None] = "020" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.execute(""" + CREATE TABLE report_schedules ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + organization_id uuid NOT NULL REFERENCES organizations(id), + name text NOT NULL CHECK (length(name) BETWEEN 1 AND 100), + kind text NOT NULL DEFAULT 'supplier_digest' CHECK (kind IN ('supplier_digest')), + frequency text NOT NULL CHECK (frequency IN ('daily', 'weekly', 'monthly')), + weekday smallint NULL CHECK (weekday BETWEEN 0 AND 6), + hour_utc smallint NOT NULL CHECK (hour_utc BETWEEN 0 AND 23), + recipients text[] NOT NULL CHECK (cardinality(recipients) BETWEEN 1 AND 20), + skip_empty boolean NOT NULL DEFAULT false, + active boolean NOT NULL DEFAULT true, + next_run_at timestamptz NOT NULL, + last_run_at timestamptz NULL, + last_status jsonb NULL, + created_by text NULL, + created_at timestamptz NOT NULL DEFAULT now(), + CONSTRAINT report_schedules_weekly_has_day CHECK (frequency <> 'weekly' OR weekday IS NOT NULL) + ) + """) + op.execute("CREATE INDEX ix_report_schedules_due ON report_schedules (organization_id, active, next_run_at)") + op.execute("ALTER TABLE report_schedules ENABLE ROW LEVEL SECURITY") + op.execute("ALTER TABLE report_schedules FORCE ROW LEVEL SECURITY") + op.execute(""" + CREATE POLICY report_schedules_org_isolation ON report_schedules + USING (organization_id = scirm_current_org()) + WITH CHECK (organization_id = scirm_current_org()) + """) + + +def downgrade() -> None: + op.execute("DROP TABLE IF EXISTS report_schedules") diff --git a/apps/api-gateway/main.py b/apps/api-gateway/main.py index e00593d..9d4e9ee 100644 --- a/apps/api-gateway/main.py +++ b/apps/api-gateway/main.py @@ -15,7 +15,7 @@ from uuid import uuid4 import httpx -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from fastapi import FastAPI, HTTPException, Depends, Request, WebSocket, WebSocketDisconnect from fastapi.security import HTTPBearer from pydantic import BaseModel, Field @@ -33,7 +33,7 @@ from libs.common.store import (API_KEY_PREFIX, AUDIT_CATEGORY_NAMES, FEEDBACK_VERDICTS, ROLES, EmailExists, LastAdmin, audit_category, audit_cursor, create_store, is_api_key) from libs.common import secret_box -from libs.integrations import webhooks +from libs.integrations import reports, webhooks # MFA seeds are encrypted at rest; outside development the key must be configured. secret_box.require_key() @@ -767,6 +767,124 @@ async def list_webhook_deliveries(webhook_id: str, limit: int = 20, token: str = limit=min(max(limit, 1), 100))} +class ReportScheduleRequest(BaseModel): + name: str = Field(..., min_length=1, max_length=100) + frequency: str + hour_utc: int = Field(..., ge=0, le=23) + weekday: Optional[int] = Field(None, ge=0, le=6) + recipients: List[str] = Field(..., min_length=1, max_length=20) + skip_empty: bool = False + + +class ReportScheduleUpdate(BaseModel): + name: Optional[str] = Field(None, min_length=1, max_length=100) + frequency: Optional[str] = None + hour_utc: Optional[int] = Field(None, ge=0, le=23) + weekday: Optional[int] = Field(None, ge=0, le=6) + recipients: Optional[List[str]] = Field(None, min_length=1, max_length=20) + skip_empty: Optional[bool] = None + active: Optional[bool] = None + + +async def _member_emails(org: str) -> set: + emails = {m["email"].lower() for m in await audit_store.list_members(org) if m.get("is_active")} + return emails | {email for email, info in DEV_USERS.items() if info["org_id"] == org} + + +async def _valid_recipients(org: str, recipients: List[str]) -> List[str]: + """Only active members of the organization: a schedule must not mail supplier data out.""" + wanted = list(dict.fromkeys(r.strip().lower() for r in recipients if r.strip())) + outside = [r for r in wanted if r not in await _member_emails(org)] + if not wanted or outside: + raise HTTPException(status_code=400, detail="Reports can only go to members of your organization" + + (f": not a member: {', '.join(outside)}" if outside else "")) + return wanted + + +def _schedule_timing(frequency: str, hour_utc: int, weekday: Optional[int]) -> Dict[str, Any]: + if frequency not in reports.FREQUENCIES: + raise HTTPException(status_code=400, detail=f"frequency must be one of: {', '.join(reports.FREQUENCIES)}") + if frequency == "weekly" and weekday is None: + raise HTTPException(status_code=400, detail="A weekly report needs a weekday (0 = Monday)") + weekday = weekday if frequency == "weekly" else None + nxt = reports.next_run(frequency, hour_utc, weekday, datetime.now(timezone.utc)) + return {"frequency": frequency, "hour_utc": hour_utc, "weekday": weekday, "next_run_at": nxt.isoformat()} + + +def _schedule_view(row: Dict[str, Any]) -> Dict[str, Any]: + return {**row, "when": reports.describe(row["frequency"], row["hour_utc"], row.get("weekday"))} + + +@app.get("/api/v1/org/reports") +async def list_report_schedules(token: str = Depends(security)): + user = await verify_token(token.credentials) + _require_admin(user, "manage scheduled reports") + rows = await audit_store.list_report_schedules(user.organization_id) + return {"schedules": [_schedule_view(r) for r in rows], "email_configured": reports.smtp_config() is not None, + "members": sorted(await _member_emails(user.organization_id))} + + +@app.post("/api/v1/org/reports") +async def create_report_schedule(body: ReportScheduleRequest, token: str = Depends(security)): + """A supplier risk digest emailed to members on a schedule (times in UTC).""" + user = await verify_token(token.credentials) + _require_admin(user, "manage scheduled reports") + fields = {"name": body.name.strip(), "skip_empty": body.skip_empty, + "recipients": await _valid_recipients(user.organization_id, body.recipients), + **_schedule_timing(body.frequency, body.hour_utc, body.weekday)} + row = await audit_store.create_report_schedule(user.organization_id, fields, user.email) + await _audit_event(user.organization_id, user.id, user.email, "report.scheduled", target=str(row["id"])) + return {"schedule": _schedule_view(row)} + + +@app.patch("/api/v1/org/reports/{schedule_id}") +async def update_report_schedule(schedule_id: str, body: ReportScheduleUpdate, token: str = Depends(security)): + user = await verify_token(token.credentials) + _require_admin(user, "manage scheduled reports") + current = next((s for s in await audit_store.list_report_schedules(user.organization_id) if s["id"] == schedule_id), + None) + if not current: + raise HTTPException(status_code=404, detail="Scheduled report not found") + fields: Dict[str, Any] = {k: v for k, v in body.model_dump(exclude_unset=True).items() + if k in ("name", "skip_empty", "active") and v is not None} + if body.recipients is not None: + fields["recipients"] = await _valid_recipients(user.organization_id, body.recipients) + if {"frequency", "hour_utc", "weekday"} & body.model_fields_set: + fields.update(_schedule_timing(body.frequency or current["frequency"], + body.hour_utc if body.hour_utc is not None else current["hour_utc"], + body.weekday if "weekday" in body.model_fields_set else current.get("weekday"))) + elif fields.get("active"): # resumed: next occurrence from now, not a missed one in the past + fields["next_run_at"] = reports.next_run(current["frequency"], current["hour_utc"], current.get("weekday"), + datetime.now(timezone.utc)).isoformat() + row = await audit_store.update_report_schedule(user.organization_id, schedule_id, fields) + return {"schedule": _schedule_view(row)} + + +@app.delete("/api/v1/org/reports/{schedule_id}") +async def delete_report_schedule(schedule_id: str, token: str = Depends(security)): + user = await verify_token(token.credentials) + _require_admin(user, "manage scheduled reports") + if not await audit_store.delete_report_schedule(user.organization_id, schedule_id): + raise HTTPException(status_code=404, detail="Scheduled report not found") + await _audit_event(user.organization_id, user.id, user.email, "report.deleted", target=schedule_id) + return {"deleted": True} + + +@app.post("/api/v1/org/reports/{schedule_id}/send-now") +async def send_report_now(schedule_id: str, token: str = Depends(security)): + """Queue this report for the next pass (within about a minute). Its regular schedule + continues afterwards.""" + user = await verify_token(token.credentials) + _require_admin(user, "manage scheduled reports") + if reports.smtp_config() is None: + raise HTTPException(status_code=409, detail="Email is not configured on this deployment (SMTP_HOST).") + row = await audit_store.update_report_schedule(user.organization_id, schedule_id, + {"next_run_at": await audit_store.db_now(), "active": True}) + if not row: + raise HTTPException(status_code=404, detail="Scheduled report not found") + return {"schedule": _schedule_view(row), "note": "Queued: it is sent within about a minute."} + + @app.post("/api/v1/org/invitations") async def create_invitation(body: InviteRequest, token: str = Depends(security)): """Invite someone to this organization. Returns the invitation link once. There is no diff --git a/apps/frontend/src/components/ReportsSettings.test.ts b/apps/frontend/src/components/ReportsSettings.test.ts new file mode 100644 index 0000000..f774962 --- /dev/null +++ b/apps/frontend/src/components/ReportsSettings.test.ts @@ -0,0 +1,14 @@ +import { describe, expect, it, vi } from 'vitest'; + +// The component module imports the API client, which needs a browser; only hourLabel is tested. +vi.mock('../services/api', () => ({ api: {} })); + +import { hourLabel } from './ReportsSettings'; + +describe('hourLabel', () => { + it('shows the UTC hour and the same moment in local time', () => { + const label = hourLabel(7, new Date('2026-10-23T12:00:00Z')); + expect(label.startsWith('07:00 UTC (')).toBe(true); + expect(label.endsWith('your time)')).toBe(true); + }); +}); diff --git a/apps/frontend/src/components/ReportsSettings.tsx b/apps/frontend/src/components/ReportsSettings.tsx new file mode 100644 index 0000000..b464f7b --- /dev/null +++ b/apps/frontend/src/components/ReportsSettings.tsx @@ -0,0 +1,182 @@ +import { FormEvent, useCallback, useEffect, useState } from 'react'; +import { useTranslation } from 'react-i18next'; +import { api, ReportFrequency, ReportSchedule } from '../services/api'; +import { utcDate } from '../services/time'; + +const WEEKDAYS = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']; + +/** "07:00 UTC (09:00 your time)" for an hour in UTC. */ +export function hourLabel(hourUtc: number, now: Date = new Date()): string { + const d = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate(), hourUtc)); + const local = d.toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' }); + return `${String(hourUtc).padStart(2, '0')}:00 UTC (${local} your time)`; +} + +function errorText(err: any, fallback: string): string { + return err?.response?.data?.detail || fallback; +} + +// A supplier risk digest emailed to members on a schedule (admins). +export default function ReportsSettings() { + const { t } = useTranslation(); + const [schedules, setSchedules] = useState([]); + const [members, setMembers] = useState([]); + const [emailOn, setEmailOn] = useState(true); + const [name, setName] = useState('Weekly supplier digest'); + const [frequency, setFrequency] = useState('weekly'); + const [weekday, setWeekday] = useState(0); + const [hour, setHour] = useState(7); + const [recipients, setRecipients] = useState([]); + const [skipEmpty, setSkipEmpty] = useState(false); + const [notice, setNotice] = useState(''); + const [error, setError] = useState(''); + + const load = useCallback(() => api.listReports().then((r) => { + setSchedules(r.schedules); + setMembers(r.members); + setEmailOn(r.email_configured); + }).catch((err) => setError(errorText(err, t('reports.loadFailed', 'Could not load the scheduled reports.')))), [t]); + useEffect(() => { load(); }, [load]); + + const create = async (e: FormEvent) => { + e.preventDefault(); + setError(''); + setNotice(''); + try { + await api.createReport({ name: name.trim(), frequency, hour_utc: hour, weekday: frequency === 'weekly' ? weekday : null, + recipients, skip_empty: skipEmpty }); + setRecipients([]); + await load(); + } catch (err) { + setError(errorText(err, t('reports.createFailed', 'Could not schedule the report.'))); + } + }; + + const act = async (s: ReportSchedule, action: 'send' | 'toggle' | 'delete') => { + setError(''); + setNotice(''); + try { + if (action === 'send') { + await api.sendReportNow(s.id); + setNotice(t('reports.queued', '"{{name}}" is queued and will be sent within about a minute.', { name: s.name })); + } else if (action === 'toggle') { + await api.updateReport(s.id, { active: !s.active }); + } else { + await api.deleteReport(s.id); + } + await load(); + } catch (err) { + setError(errorText(err, t('reports.actionFailed', 'That did not work.'))); + } + }; + + const toggleRecipient = (email: string) => + setRecipients((current) => (current.includes(email) ? current.filter((r) => r !== email) : [...current, email])); + + const field = 'px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-lg text-sm bg-white dark:bg-gray-800 dark:text-white'; + return ( +
+

{t('reports.title', 'Scheduled reports')}

+

+ {t('reports.hint', 'A supplier risk digest by email: new public records naming your suppliers and new disaster exposures since the last report. Only members of your organization can receive it.')} +

+ {!emailOn && ( +

+ {t('reports.noEmail', 'Email is not configured on this deployment (SMTP), so reports cannot be sent yet. You can still set them up.')} +

+ )} + {error &&

{error}

} + {notice &&

{notice}

} + +
+
+ + + {frequency === 'weekly' && ( + + )} + +
+
+

{t('reports.recipients', 'Send to')}

+
+ {members.map((m) => ( + + ))} +
+
+
+ + +
+
+ + {schedules.length > 0 && ( + + + + + + + + + + {schedules.map((s) => ( + + + + + + + ))} + +
{t('reports.report', 'Report')}{t('reports.next', 'Next')}{t('reports.last', 'Last')} +
+

{s.name}

+

{s.when} · {s.recipients.join(', ')}{!s.active && ` · ${t('reports.paused', 'paused')}`}

+ 		{s.active ? utcDate(s.next_run_at).toLocaleString() : '–'} + {s.last_status ? ( + + {s.last_status.ok + ? (s.last_status.sent ? t('reports.sent', 'Sent ({{n}} items)', { n: s.last_status.events ?? 0 }) : t('reports.notNeeded', 'Nothing new, not sent')) + : s.last_status.error} + {utcDate(s.last_status.at).toLocaleString()} + + ) : {t('reports.never', 'not yet')}} + + + + +
+ )} +
+ ); +} diff --git a/apps/frontend/src/pages/AuditLogPage.tsx b/apps/frontend/src/pages/AuditLogPage.tsx index dae52f1..bedfc40 100644 --- a/apps/frontend/src/pages/AuditLogPage.tsx +++ b/apps/frontend/src/pages/AuditLogPage.tsx @@ -17,6 +17,8 @@ const NAMED: Record = { 'api_key.refused': 'API key refused', 'webhook.created': 'Added a webhook', 'webhook.deleted': 'Deleted a webhook', + 'report.scheduled': 'Scheduled a report', + 'report.deleted': 'Deleted a scheduled report', 'integrations.sap.save': 'Saved the SAP connection', 'integrations.sap.test': 'Tested the SAP connection', 'integrations.sap.delete': 'Removed the SAP connection', @@ -34,6 +36,10 @@ const ROUTES: Record = { 'PATCH /api/v1/org/webhooks/{webhook_id}': 'Paused, resumed or changed a webhook', 'DELETE /api/v1/org/webhooks/{webhook_id}': 'Deleted a webhook', 'POST /api/v1/org/webhooks/{webhook_id}/test': 'Sent a test webhook', + 'POST /api/v1/org/reports': 'Scheduled a report', + 'PATCH /api/v1/org/reports/{schedule_id}': 'Paused, resumed or changed a scheduled report', + 'DELETE /api/v1/org/reports/{schedule_id}': 'Deleted a scheduled report', + 'POST /api/v1/org/reports/{schedule_id}/send-now': 'Sent a report now', 'POST /api/v1/org/members/{member_id}/mfa/reset': "Reset a teammate's two-step sign-in", 'POST /api/v1/auth/mfa/setup': 'Started two-step sign-in setup', 'POST /api/v1/auth/mfa/confirm': 'Confirmed two-step sign-in', diff --git a/apps/frontend/src/pages/IntegrationsPage.tsx b/apps/frontend/src/pages/IntegrationsPage.tsx index 840e828..3b66aeb 100644 --- a/apps/frontend/src/pages/IntegrationsPage.tsx +++ b/apps/frontend/src/pages/IntegrationsPage.tsx @@ -6,6 +6,7 @@ import { useAuth } from '../contexts/AuthContext'; import { utcDate } from '../services/time'; import ApiKeysSettings from '../components/ApiKeysSettings'; import WebhooksSettings from '../components/WebhooksSettings'; +import ReportsSettings from '../components/ReportsSettings'; type Tone = 'ok' | 'warn' | 'bad' | 'off'; const TONES: Record = { @@ -128,6 +129,7 @@ export default function IntegrationsPage() { + ); } diff --git a/apps/frontend/src/services/api.ts b/apps/frontend/src/services/api.ts index a2fcccf..0bbdba5 100644 --- a/apps/frontend/src/services/api.ts +++ b/apps/frontend/src/services/api.ts @@ -344,6 +344,29 @@ export interface WebhookDelivery { created_at: string; } +export type ReportFrequency = 'daily' | 'weekly' | 'monthly'; + +export interface ReportScheduleInput { + name: string; + frequency: ReportFrequency; + hour_utc: number; + /** 0 = Monday; weekly only. */ + weekday?: number | null; + recipients: string[]; + skip_empty?: boolean; +} + +export interface ReportSchedule extends ReportScheduleInput { + id: string; + active: boolean; + /** e.g. "Every Monday at 07:00 UTC" */ + when: string; + next_run_at: string; + last_run_at: string | null; + last_status: { ok: boolean; sent: boolean; events?: number; recipients?: number; error?: string; detail?: string; at: string } | null; + created_by: string | null; +} + export type AuditCategory = 'sign_in' | 'access' | 'integrations' | 'changes'; export interface AuditFilters { @@ -603,6 +626,27 @@ export const api = { return (await apiClient.get(`/api/v1/org/webhooks/${encodeURIComponent(id)}/deliveries`)).data; }, + // Scheduled reports (admins) + async listReports(): Promise<{ schedules: ReportSchedule[]; email_configured: boolean; members: string[] }> { + return (await apiClient.get('/api/v1/org/reports')).data; + }, + + async createReport(body: ReportScheduleInput): Promise { + return (await apiClient.post('/api/v1/org/reports', body)).data.schedule; + }, + + async updateReport(id: string, changes: Partial & { active?: boolean }): Promise { + return (await apiClient.patch(`/api/v1/org/reports/${encodeURIComponent(id)}`, changes)).data.schedule; + }, + + async deleteReport(id: string): Promise { + await apiClient.delete(`/api/v1/org/reports/${encodeURIComponent(id)}`); + }, + + async sendReportNow(id: string): Promise { + await apiClient.post(`/api/v1/org/reports/${encodeURIComponent(id)}/send-now`); + }, + // Audit log (admins and auditors) async listAudit(filters: AuditFilters & { limit?: number; before?: string }): Promise { return (await apiClient.get('/api/v1/audit', { params: auditParams(filters) })).data; diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index dffbd5f..9dc0c7a 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -84,6 +84,12 @@ services: - REVIEWER_SERVICE_URL=http://reviewer:8000 - DISCOVERY_SERVICE_URL=http://discovery:8000 - INGESTION_SERVICE_URL=http://ingestion:8000 + # Scheduled report emails go to Mailpit (http://127.0.0.1:8025), never out of the machine. + - SMTP_HOST=mailpit + - SMTP_PORT=1025 + - SMTP_STARTTLS=false + - SMTP_FROM=SCIRM + - PUBLIC_APP_URL=http://localhost:3000 - LOG_LEVEL=DEBUG depends_on: migrate: @@ -394,6 +400,11 @@ services: - ENVIRONMENT=development - SAMPLE_FEATURES=${SAMPLE_FEATURES:-} - CORS_ALLOWED_ORIGINS=http://localhost:3000,http://127.0.0.1:3000,http://localhost:8000 + # Scheduled report emails go to Mailpit (http://127.0.0.1:8025), never out of the machine. + - SMTP_HOST=mailpit + - SMTP_PORT=1025 + - SMTP_STARTTLS=false + - SMTP_FROM=SCIRM - LOG_LEVEL=DEBUG depends_on: migrate: @@ -416,6 +427,12 @@ services: - ./libs:/app/libs command: uvicorn main:app --app-dir apps/api-gateway --host 0.0.0.0 --port 8000 --reload --reload-dir apps/api-gateway --reload-dir libs + # Catches every email the stack sends (scheduled reports). Web UI on 127.0.0.1:8025. + mailpit: + image: axllent/mailpit:v1.21 + ports: + - "127.0.0.1:8025:8025" + frontend: build: context: ./apps/frontend diff --git a/docs/scheduled-reports.md b/docs/scheduled-reports.md new file mode 100644 index 0000000..2439224 --- /dev/null +++ b/docs/scheduled-reports.md @@ -0,0 +1,60 @@ +# Scheduled reports + +A **supplier risk digest** by email, daily, weekly or monthly. It lists what is new since +the previous report: +- **public records naming your suppliers**: FDA recalls, shortages, warning letters and + import alerts, one line per recall event, with a link to the source +- **disaster exposures**: GDACS alerts near a supplier's site, or in its country when its + sites are unknown + +It uses the same matching rules as the supplier panel and the [webhooks](webhooks.md); the +detector records these events for every organization whether or not it uses webhooks. +Each email says that these are records matched by name and alerts matched by location, +not findings about a supplier. + +## Setting up + +An administrator opens **Administration → Integrations → Scheduled reports**: +- **Name** and **how often**: daily, weekly on a chosen day, or monthly on the 1st. +- **Time** in UTC. The list shows each hour in your local time too. +- **Recipients**: members of your organization only. They are checked when the schedule is + saved and again when the report is sent, so someone who leaves stops receiving it. +- **Don't send when there is nothing new**: otherwise a short "nothing new" email is sent. + +**Send now** sends the report within about a minute, covering the period since the last +report; the regular schedule continues afterwards. **Pause** stops it. Resuming picks up at +the next scheduled time, without catching up on missed ones. Each schedule shows its next +run and the result of the last one. Creating and deleting schedules are recorded in the +[audit log](audit-log.md). + +The first report covers the previous day, week or month. Events are only recorded from +the time the webhook detector first ran for the organization, so a new deployment's +first report can be short. + +## Email settings + +The coordinator sends the email; the gateway only reads whether email is configured. + +| Variable | Default | | +|---|---|---| +| `SMTP_HOST` | (unset) | Without it nothing is sent, and the page says so. | +| `SMTP_PORT` | `587` | | +| `SMTP_USERNAME`, `SMTP_PASSWORD` | (unset) | Login, if the server needs one. | +| `SMTP_STARTTLS` | `true` | | +| `SMTP_FROM` | `SCIRM ` | Use an address your mail service may send from (SPF/DKIM). | +| `PUBLIC_APP_URL` | (unset) | Link back to SCIRM in the email. | + +In staging these come from the `scirm-secrets` keys `smtp-host`, `smtp-port`, +`smtp-username`, `smtp-password` and `smtp-from`. The keys are optional: without them the +pods still start. + +**Development.** The Compose stack includes [Mailpit](https://mailpit.axllent.org/), which +catches every email. Read it at ; nothing leaves your machine. + +## Running + +The coordinator checks for due reports every 30 seconds (`WEBHOOK_TICK_SECONDS`). +- **No double sends.** A due schedule is moved forward before it is sent (claimed with + `FOR UPDATE SKIP LOCKED`), so several coordinator replicas never send it twice. +- **Failures.** A failed send (SMTP error, no member recipients left) is shown on the + schedule, and the next run happens as normal. diff --git a/infra/k8s/staging/all-services.yml b/infra/k8s/staging/all-services.yml index 850bc4a..133f27e 100644 --- a/infra/k8s/staging/all-services.yml +++ b/infra/k8s/staging/all-services.yml @@ -398,6 +398,37 @@ spec: secretKeyRef: name: scirm-secrets key: mfa-encryption-key + # Scheduled report emails (optional: without smtp-host nothing is sent and the UI says so). + - name: SMTP_HOST + valueFrom: + secretKeyRef: + name: scirm-secrets + key: smtp-host + optional: true + - name: SMTP_PORT + valueFrom: + secretKeyRef: + name: scirm-secrets + key: smtp-port + optional: true + - name: SMTP_USERNAME + valueFrom: + secretKeyRef: + name: scirm-secrets + key: smtp-username + optional: true + - name: SMTP_PASSWORD + valueFrom: + secretKeyRef: + name: scirm-secrets + key: smtp-password + optional: true + - name: SMTP_FROM + valueFrom: + secretKeyRef: + name: scirm-secrets + key: smtp-from + optional: true # Users who set up two-step sign-in are asked for a code (the default; explicit here). - name: MFA_ENABLED value: "true" diff --git a/infra/k8s/staging/coordinator-deployment.yml b/infra/k8s/staging/coordinator-deployment.yml index 85ff914..eff2a61 100644 --- a/infra/k8s/staging/coordinator-deployment.yml +++ b/infra/k8s/staging/coordinator-deployment.yml @@ -56,6 +56,40 @@ spec: secretKeyRef: name: scirm-secrets key: mfa-encryption-key + # Scheduled report emails (optional: without smtp-host nothing is sent and the UI says so). + - name: SMTP_HOST + valueFrom: + secretKeyRef: + name: scirm-secrets + key: smtp-host + optional: true + - name: SMTP_PORT + valueFrom: + secretKeyRef: + name: scirm-secrets + key: smtp-port + optional: true + - name: SMTP_USERNAME + valueFrom: + secretKeyRef: + name: scirm-secrets + key: smtp-username + optional: true + - name: SMTP_PASSWORD + valueFrom: + secretKeyRef: + name: scirm-secrets + key: smtp-password + optional: true + - name: SMTP_FROM + valueFrom: + secretKeyRef: + name: scirm-secrets + key: smtp-from + optional: true + # Links back to SCIRM in report emails. + - name: PUBLIC_APP_URL + value: "REPLACE_WITH_STAGING_APP_URL" - name: PLANNER_SERVICE_URL value: "http://planner:8000" - name: RESEARCHER_SERVICE_URL diff --git a/infra/k8s/staging/secrets.example.yml b/infra/k8s/staging/secrets.example.yml index 019ff09..1048032 100644 --- a/infra/k8s/staging/secrets.example.yml +++ b/infra/k8s/staging/secrets.example.yml @@ -37,6 +37,13 @@ stringData: # Losing it means every user with MFA must set it up again, and SAP passwords must be re-entered. mfa-encryption-key: "REPLACE_WITH_ACTUAL_MFA_ENCRYPTION_KEY" encryption-key: "REPLACE_WITH_ACTUAL_ENCRYPTION_KEY" + + # Email for scheduled reports (optional; leave these out to disable sending). + smtp-host: "REPLACE_WITH_SMTP_HOST" + smtp-port: "587" + smtp-username: "REPLACE_WITH_SMTP_USERNAME" + smtp-password: "REPLACE_WITH_SMTP_PASSWORD" + smtp-from: "SCIRM " # Monitoring grafana-admin-password: "REPLACE_WITH_ACTUAL_GRAFANA_PASSWORD" diff --git a/libs/common/store.py b/libs/common/store.py index 50c08ad..d3c401d 100644 --- a/libs/common/store.py +++ b/libs/common/store.py @@ -114,6 +114,11 @@ def api_key_hash(key: str) -> str: "delivered_at", "created_at") +_REPORT_PUBLIC = ("id", "name", "kind", "frequency", "weekday", "hour_utc", "recipients", "skip_empty", "active", + "next_run_at", "last_run_at", "last_status", "created_by", "created_at") +_REPORT_EDITABLE = ("name", "frequency", "weekday", "hour_utc", "recipients", "skip_empty", "active", "next_run_at") + + def _webhook_bound(org: str, webhook_id: str) -> str: """What a webhook's encrypted secret is bound to: it opens only for this webhook.""" return f"webhook:{org}:{webhook_id}" @@ -246,6 +251,7 @@ def __init__(self, seed_demo: bool = True): self.webhook_events: Dict[str, Dict[str, Any]] = {} self.webhook_deliveries: Dict[str, Dict[str, Any]] = {} self.notification_state: Dict[str, str] = {} + self.report_schedules: Dict[str, Dict[str, Any]] = {} async def close(self) -> None: return None @@ -883,6 +889,65 @@ async def finish_delivery(self, org: str, delivery_id: str, result: Dict[str, An hook["last_delivery"] = {"at": await self.db_now(), "ok": bool(result.get("ok")), "status_code": result.get("status_code"), "error": d["last_error"]} + # --- scheduled reports -------------------------------------------------- + def _own_report(self, org: str, schedule_id: str) -> Optional[Dict[str, Any]]: + row = self.report_schedules.get(str(schedule_id)) + return row if row and row["organization_id"] == org else None + + async def create_report_schedule(self, org: str, fields: Dict[str, Any], created_by: Optional[str]) -> Dict[str, Any]: + row = {"id": str(uuid.uuid4()), "organization_id": org, "kind": "supplier_digest", "skip_empty": False, + "active": True, "weekday": None, "last_run_at": None, "last_status": None, "created_by": created_by, + "created_at": await self.db_now(), **{k: fields[k] for k in _REPORT_EDITABLE if k in fields}} + self.report_schedules[row["id"]] = row + return {k: row.get(k) for k in _REPORT_PUBLIC} + + async def list_report_schedules(self, org: str) -> List[Dict[str, Any]]: + rows = sorted((r for r in self.report_schedules.values() if r["organization_id"] == org), + key=lambda r: r["created_at"]) + return [{k: r.get(k) for k in _REPORT_PUBLIC} for r in rows] + + async def update_report_schedule(self, org: str, schedule_id: str, fields: Dict[str, Any]) -> Optional[Dict[str, Any]]: + row = self._own_report(org, schedule_id) + if not row: + return None + row.update({k: v for k, v in fields.items() if k in _REPORT_EDITABLE}) + return {k: row.get(k) for k in _REPORT_PUBLIC} + + async def delete_report_schedule(self, org: str, schedule_id: str) -> bool: + if not self._own_report(org, schedule_id): + return False + del self.report_schedules[str(schedule_id)] + return True + + async def claim_due_reports(self, org: str, lease_seconds: int = 600) -> List[Dict[str, Any]]: + from datetime import timedelta + now = datetime.now(timezone.utc) + due = [r for r in self.report_schedules.values() if r["organization_id"] == org and r["active"] + and datetime.fromisoformat(r["next_run_at"]) <= now] + for r in due: + r["next_run_at"] = _utc_iso(now + timedelta(seconds=lease_seconds)) + return [{k: r.get(k) for k in _REPORT_PUBLIC} for r in due] + + async def finish_report(self, org: str, schedule_id: str, next_run_at: str, last_run_at: Optional[str], + status: Dict[str, Any]) -> None: + row = self._own_report(org, schedule_id) + if row: + row.update(next_run_at=next_run_at, last_status=status) + if last_run_at: + row["last_run_at"] = last_run_at + + async def list_events(self, org: str, since: str, until: str) -> List[Dict[str, Any]]: + """The organization's new-match events created in (since, until], oldest first. The + next report starts where this one ended, so a boundary event is in exactly one.""" + lo, hi = datetime.fromisoformat(since), datetime.fromisoformat(until) + rows = [e for e in self.webhook_events.values() if e["organization_id"] == org and e["type"] != "ping" + and lo < datetime.fromisoformat(e["created_at"]) <= hi] + return [{k: e[k] for k in ("id", "type", "payload", "created_at")} + for e in sorted(rows, key=lambda e: e["created_at"])] + + async def organization_name(self, org: str) -> str: + return (self.orgs.get(org) or {}).get("name") or "your organization" + # --- audit -------------------------------------------------------------- async def record_audit(self, org: Optional[str], entry: Dict[str, Any]) -> None: self.audit.append({**entry, "id": str(uuid.uuid4()), "organization_id": org, "created_at": _now()}) @@ -2103,11 +2168,99 @@ async def finish_delivery(self, org: str, delivery_id: str, result: Dict[str, An "WHERE id = (SELECT webhook_id FROM webhook_deliveries WHERE id = CAST(:i AS uuid))", {"ok": ok, "code": result.get("status_code"), "err": error, "i": delivery_id})]) + # --- scheduled reports -------------------------------------------------- + _REPORT_COLUMNS = ", ".join(_REPORT_PUBLIC) + + @staticmethod + def _report_row(row: Dict[str, Any]) -> Dict[str, Any]: + out = _jsonable(row) + out["recipients"] = list(out.get("recipients") or []) + if isinstance(out.get("last_status"), str): + out["last_status"] = json.loads(out["last_status"]) + return out + + @staticmethod + def _report_params(fields: Dict[str, Any]) -> Dict[str, Any]: + params = {k: fields.get(k) for k in _REPORT_EDITABLE} + if params.get("next_run_at"): + params["next_run_at"] = datetime.fromisoformat(params["next_run_at"]) + if params.get("recipients") is not None: + params["recipients"] = list(params["recipients"]) + return params + + async def create_report_schedule(self, org: str, fields: Dict[str, Any], created_by: Optional[str]) -> Dict[str, Any]: + p = self._report_params(fields) + rows = await self._all(org, ( + "INSERT INTO report_schedules (organization_id, name, frequency, weekday, hour_utc, recipients, skip_empty, " + "next_run_at, created_by) VALUES (CAST(:org AS uuid), :name, :frequency, :weekday, :hour_utc, " + "CAST(:recipients AS text[]), COALESCE(:skip_empty, false), :next_run_at, :by) " + f"RETURNING {self._REPORT_COLUMNS}"), + {**{k: p[k] for k in ("name", "frequency", "weekday", "hour_utc", "recipients", "skip_empty", "next_run_at")}, + "org": org, "by": created_by}) + return self._report_row(rows[0]) + + async def list_report_schedules(self, org: str) -> List[Dict[str, Any]]: + rows = await self._all(org, f"SELECT {self._REPORT_COLUMNS} FROM report_schedules ORDER BY created_at") + return [self._report_row(r) for r in rows] + + async def update_report_schedule(self, org: str, schedule_id: str, fields: Dict[str, Any]) -> Optional[Dict[str, Any]]: + if not _is_uuid(schedule_id): + return None + sets = [k for k in _REPORT_EDITABLE if k in fields] + if not sets: + rows = await self._all(org, f"SELECT {self._REPORT_COLUMNS} FROM report_schedules WHERE id = CAST(:i AS uuid)", + {"i": schedule_id}) + return self._report_row(rows[0]) if rows else None + assignments = ", ".join(f"{k} = CAST(:{k} AS text[])" if k == "recipients" else f"{k} = :{k}" for k in sets) + p = self._report_params(fields) + rows = await self._all(org, ( + f"UPDATE report_schedules SET {assignments} WHERE id = CAST(:i AS uuid) RETURNING {self._REPORT_COLUMNS}"), + {**{k: p[k] for k in sets}, "i": schedule_id}) + return self._report_row(rows[0]) if rows else None + + async def delete_report_schedule(self, org: str, schedule_id: str) -> bool: + if not _is_uuid(schedule_id): + return False + return bool(await self._all(org, "DELETE FROM report_schedules WHERE id = CAST(:i AS uuid) RETURNING id", + {"i": schedule_id})) + + async def claim_due_reports(self, org: str, lease_seconds: int = 600) -> List[Dict[str, Any]]: + """Due active schedules, moved to a short lease first (FOR UPDATE SKIP LOCKED), so a + second worker skips them while this one sends.""" + rows = await self._all(org, ( + "WITH due AS (SELECT id FROM report_schedules WHERE active AND next_run_at <= now() " + "ORDER BY next_run_at FOR UPDATE SKIP LOCKED) " + "UPDATE report_schedules r SET next_run_at = now() + make_interval(secs => :lease) FROM due " + f"WHERE r.id = due.id RETURNING {', '.join('r.' + c for c in _REPORT_PUBLIC)}"), {"lease": lease_seconds}) + return [self._report_row(r) for r in rows] + + async def finish_report(self, org: str, schedule_id: str, next_run_at: str, last_run_at: Optional[str], + status: Dict[str, Any]) -> None: + await self._exec(org, [( + "UPDATE report_schedules SET next_run_at = :n, last_run_at = COALESCE(:l, last_run_at), " + "last_status = CAST(:s AS jsonb) WHERE id = CAST(:i AS uuid)", + {"n": datetime.fromisoformat(next_run_at), "l": datetime.fromisoformat(last_run_at) if last_run_at else None, + "s": json.dumps(status, default=str), "i": schedule_id})]) + + async def list_events(self, org: str, since: str, until: str) -> List[Dict[str, Any]]: + """The organization's new-match events created in (since, until], oldest first. The + next report starts where this one ended, so a boundary event is in exactly one.""" + rows = await self._all(org, ( + "SELECT id, type, payload, created_at FROM webhook_events WHERE type <> 'ping' " + "AND created_at > :since AND created_at <= :until ORDER BY created_at LIMIT 5000"), + {"since": datetime.fromisoformat(since), "until": datetime.fromisoformat(until)}) + return [{"id": str(r["id"]), "type": r["type"], "created_at": _utc_iso(r["created_at"]), + "payload": json.loads(r["payload"]) if isinstance(r["payload"], str) else r["payload"]} for r in rows] + + async def organization_name(self, org: str) -> str: + rows = await self._all(None, "SELECT name FROM organizations WHERE id = CAST(:o AS uuid)", {"o": org}) + return rows[0]["name"] if rows else "your organization" + async def list_audit(self, org: str, limit: int = 100, filters: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: """Newest first. filters: category, actor, failed, since, until (ISO), before (cursor).""" where, params = _audit_where(filters or {}) - sql =("SELECT * FROM audit_logs" + (" WHERE " + " AND ".join(where) if where else "") + sql = ("SELECT * FROM audit_logs" + (" WHERE " + " AND ".join(where) if where else "") + " ORDER BY created_at DESC, id DESC LIMIT :limit") rows = await self._all(org, sql, {**params, "limit": limit}) return [_jsonable(r) for r in rows] diff --git a/libs/integrations/reports.py b/libs/integrations/reports.py new file mode 100644 index 0000000..f903bd2 --- /dev/null +++ b/libs/integrations/reports.py @@ -0,0 +1,201 @@ +""" +Scheduled reports: a supplier risk digest emailed to an organization's members. + +The digest covers the period since the previous run (or one day, week or month for the +first): new public records naming the organization's suppliers (record.matched) and new +disaster exposures (exposure.detected), as recorded in webhook_events by the webhook +detector. Each record links to its source; the email links back to SCIRM when +PUBLIC_APP_URL is set. + +Email goes out over SMTP (SMTP_HOST, SMTP_PORT, SMTP_USERNAME, SMTP_PASSWORD, SMTP_FROM, +SMTP_STARTTLS). Without SMTP_HOST nothing is sent and the schedule shows why. Recipients +must be active members of the organization, checked when saved and again when sent, so a +schedule cannot be used to send supplier data outside the organization. +""" + +import asyncio +import html +import os +import smtplib +import ssl +from datetime import datetime, timedelta, timezone +from email.message import EmailMessage +from typing import Any, Dict, List, Optional, Tuple + +FREQUENCIES = ("daily", "weekly", "monthly") +PERIOD = {"daily": timedelta(days=1), "weekly": timedelta(days=7), "monthly": timedelta(days=31)} +WEEKDAYS = ("Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday") +MAX_ROWS = 200 +LEASE = timedelta(minutes=10) + + +def smtp_config() -> Optional[Dict[str, Any]]: + host = os.getenv("SMTP_HOST", "").strip() + if not host: + return None + return {"host": host, "port": int(os.getenv("SMTP_PORT", "587")), "username": os.getenv("SMTP_USERNAME", ""), + "password": os.getenv("SMTP_PASSWORD", ""), "sender": os.getenv("SMTP_FROM", "SCIRM "), + "starttls": os.getenv("SMTP_STARTTLS", "true").lower() == "true"} + + +def next_run(frequency: str, hour_utc: int, weekday: Optional[int], after: datetime) -> datetime: + """The first scheduled time strictly after `after` (UTC): daily at hour_utc, weekly on + weekday (0 = Monday) at hour_utc, monthly on the 1st at hour_utc.""" + after = after.astimezone(timezone.utc) + if frequency == "daily": + candidate = after.replace(hour=hour_utc, minute=0, second=0, microsecond=0) + return candidate if candidate > after else candidate + timedelta(days=1) + if frequency == "weekly": + candidate = after.replace(hour=hour_utc, minute=0, second=0, microsecond=0) + candidate += timedelta(days=(int(weekday or 0) - candidate.weekday()) % 7) + return candidate if candidate > after else candidate + timedelta(days=7) + if frequency == "monthly": + candidate = after.replace(day=1, hour=hour_utc, minute=0, second=0, microsecond=0) + if candidate > after: + return candidate + year, month = (candidate.year + 1, 1) if candidate.month == 12 else (candidate.year, candidate.month + 1) + return candidate.replace(year=year, month=month) + raise ValueError(f"unknown frequency {frequency!r}") + + +def describe(frequency: str, hour_utc: int, weekday: Optional[int]) -> str: + when = f"{hour_utc:02d}:00 UTC" + if frequency == "daily": + return f"Every day at {when}" + if frequency == "weekly": + return f"Every {WEEKDAYS[int(weekday or 0)]} at {when}" + return f"On the 1st of every month at {when}" + + +def _row(event: Dict[str, Any]) -> Dict[str, Any]: + data = event.get("payload") or {} + record, supplier, match = data.get("record") or {}, data.get("supplier") or {}, data.get("match") or {} + where = "" + if match.get("site"): + site = match["site"] + where = f"{site.get('distance_km', '?')} km from {site.get('site') or 'a site'}" + elif match.get("match_type") == "country_exposure": + where = f"in {match.get('matched_on')}" + return {"type": event["type"], "supplier": supplier.get("name") or "", "title": record.get("title") or "", + "source": record.get("source") or "", "severity": record.get("severity") or "", + "url": record.get("url") or "", "published": (record.get("published_at") or "")[:10], + "count": data.get("record_count") or 1, "where": where} + + +def build_digest(org_name: str, events: List[Dict[str, Any]], since: datetime, until: datetime, + app_url: str = "") -> Tuple[str, str, str]: + """(subject, plain text, HTML) of the supplier risk digest.""" + rows = [_row(e) for e in events if e["type"] in ("record.matched", "exposure.detected")] + records = [r for r in rows if r["type"] == "record.matched"] + exposures = [r for r in rows if r["type"] == "exposure.detected"] + period = f"{since.astimezone(timezone.utc):%Y-%m-%d %H:%M} to {until.astimezone(timezone.utc):%Y-%m-%d %H:%M} UTC" + if rows: + subject = (f"SCIRM digest for {org_name}: {len(records)} new record(s) naming your suppliers, " + f"{len(exposures)} disaster exposure(s)") + else: + subject = f"SCIRM digest for {org_name}: nothing new" + text = [f"Supplier risk digest for {org_name}", period, ""] + if not rows: + text.append("No new public records named your suppliers, and no new disaster alerts were near them.") + for title, group in (("New public records naming your suppliers", records), + ("New disaster exposures", exposures)): + if group: + text += [f"{title} ({len(group)})", ""] + for r in group[:MAX_ROWS]: + extra = f" ({r['count']} records)" if r["count"] > 1 else "" + text.append(f"- {r['supplier']}: {r['title']}{extra} [{r['severity']}] {r['where']} {r['url']}".rstrip()) + if len(group) > MAX_ROWS: + text.append(f"... and {len(group) - MAX_ROWS} more in SCIRM.") + text.append("") + text += ["These are public records matched by name, and alerts matched by location. They are not findings " + "about your suppliers; check the source record.", ""] + if app_url: + text.append(f"Open SCIRM: {app_url}/suppliers") + + e = html.escape + + cell = "{}" + + def record_cell(r: Dict[str, Any]) -> str: + title = e(r["title"]) + if r["url"]: + title = f"{title}" + if r["count"] > 1: + title += f" ({r['count']} records)" + return title + + def table(group: List[Dict[str, Any]]) -> str: + head = "".join(f"{h}" + for h in ("Supplier", "Record", "Severity", "Where", "Published")) + body = "".join("" + "".join(cell.format(v) for v in ( + e(r["supplier"]), record_cell(r), e(r["severity"]), e(r["where"]), e(r["published"]))) + "" + for r in group[:MAX_ROWS]) + more = f"

... and {len(group) - MAX_ROWS} more in SCIRM.

" if len(group) > MAX_ROWS else "" + return (f"{head}" + f"{body}
{more}") + + parts = [f"

Supplier risk digest for {e(org_name)}

", + f"

{e(period)}

"] + if not rows: + parts.append("

No new public records named your suppliers, and no new " + "disaster alerts were near them.

") + for title, group in (("New public records naming your suppliers", records), ("New disaster exposures", exposures)): + if group: + parts.append(f"

{e(title)} ({len(group)})

{table(group)}") + parts.append("

These are public records matched by " + "name, and alerts matched by location. They are not findings about your suppliers; check the source " + "record.

") + if app_url: + parts.append(f"

Open SCIRM

") + return subject, "\n".join(text), "
" + "".join(parts) + "
" + + +def _send_sync(config: Dict[str, Any], recipients: List[str], subject: str, text: str, body_html: str) -> None: + message = EmailMessage() + message["From"], message["To"], message["Subject"] = config["sender"], ", ".join(recipients), subject + message.set_content(text) + message.add_alternative(body_html, subtype="html") + with smtplib.SMTP(config["host"], config["port"], timeout=30) as smtp: + if config["starttls"]: + smtp.starttls(context=ssl.create_default_context()) + if config["username"]: + smtp.login(config["username"], config["password"]) + smtp.send_message(message) + + +async def send_email(recipients: List[str], subject: str, text: str, body_html: str) -> None: + config = smtp_config() + if config is None: + raise RuntimeError("Email is not configured (SMTP_HOST).") + await asyncio.to_thread(_send_sync, config, recipients, subject, text, body_html) + + +async def run_due(store: Any, org: str, app_url: str = "") -> List[Dict[str, Any]]: + """Send this organization's due reports. Each claimed schedule has already been moved to + a short lease, so another replica skips it; afterwards it is set to its next run.""" + results = [] + for schedule in await store.claim_due_reports(org, lease_seconds=int(LEASE.total_seconds())): + # Database time: events are stamped by the database, and the window is (since, now]. + now = datetime.fromisoformat(await store.db_now()) + last = schedule.get("last_run_at") + since = datetime.fromisoformat(last) if last else now - PERIOD[schedule["frequency"]] + status: Dict[str, Any] + try: + members = {m["email"].lower() for m in await store.list_members(org) if m.get("is_active")} + recipients = [r for r in schedule["recipients"] if r.lower() in members] + if not recipients: + raise RuntimeError("None of the recipients is an active member of the organization any more.") + events = await store.list_events(org, since.isoformat(), now.isoformat()) + if schedule.get("skip_empty") and not events: + status = {"ok": True, "sent": False, "events": 0, "detail": "Nothing new; not sent (skip empty)."} + else: + subject, text, body_html = build_digest(await store.organization_name(org), events, since, now, app_url) + await send_email(recipients, subject, text, body_html) + status = {"ok": True, "sent": True, "events": len(events), "recipients": len(recipients)} + except Exception as exc: # recorded on the schedule; the next run still happens + status = {"ok": False, "sent": False, "error": str(exc)[:300] or type(exc).__name__} + nxt = next_run(schedule["frequency"], schedule["hour_utc"], schedule.get("weekday"), now) + await store.finish_report(org, schedule["id"], nxt.isoformat(), now.isoformat() if status["ok"] else None, + {**status, "at": now.isoformat()}) + results.append({**status, "schedule_id": schedule["id"]}) + return results diff --git a/mkdocs.yml b/mkdocs.yml index d7af935..85928e3 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -22,6 +22,7 @@ nav: - Audit log: audit-log.md - Public API: public-api.md - Webhooks: webhooks.md + - Scheduled reports: scheduled-reports.md - Development: - Branching: development/branching.md - Operations: diff --git a/services/coordinator/main.py b/services/coordinator/main.py index 4a6b52d..5c49d1f 100644 --- a/services/coordinator/main.py +++ b/services/coordinator/main.py @@ -32,7 +32,7 @@ from libs.common.security import setup_cors from libs.common.store import create_store from libs.common.tenancy import internal_headers, tenant_org -from libs.integrations import sap_import, sap_odata, sap_spend, webhooks +from libs.integrations import reports, sap_import, sap_odata, sap_spend, webhooks from libs.intel import geo # --------------------------------------------------------------------------- @@ -54,6 +54,8 @@ WEBHOOKS_ENABLED = os.getenv("WEBHOOKS_ENABLED", "true").lower() == "true" WEBHOOK_TICK_SECONDS = int(os.getenv("WEBHOOK_TICK_SECONDS", "30")) WEBHOOK_DETECT_SECONDS = int(os.getenv("WEBHOOK_DETECT_SECONDS", "300")) +# Links in report emails (none when unset). +PUBLIC_APP_URL = os.getenv("PUBLIC_APP_URL", "").rstrip("/") LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO") logger = structlog.get_logger() @@ -172,7 +174,7 @@ async def webhook_tick(detect_now: bool) -> Dict[str, int]: """One pass over every organization: send due deliveries and, when detect_now, turn newly seen public records into events (libs/integrations/webhooks.py). One organization's failure never stops the others.""" - totals = {"events": 0, "delivered": 0, "failed": 0} + totals = {"events": 0, "delivered": 0, "failed": 0, "reports": 0, "reports_failed": 0} for org_id in await store.list_org_ids(): try: if detect_now: @@ -181,6 +183,11 @@ async def webhook_tick(detect_now: bool) -> Dict[str, int]: totals["delivered" if result["ok"] else "failed"] += 1 except Exception as exc: logger.error("Webhook pass failed", org_id=org_id, error=type(exc).__name__) + try: # scheduled reports read the events above (libs/integrations/reports.py) + for result in await reports.run_due(store, org_id, PUBLIC_APP_URL): + totals["reports" if result["ok"] else "reports_failed"] += 1 + except Exception as exc: + logger.error("Report pass failed", org_id=org_id, error=type(exc).__name__) return totals diff --git a/tests/db/test_database.py b/tests/db/test_database.py index 4fdbc72..dad82b7 100644 --- a/tests/db/test_database.py +++ b/tests/db/test_database.py @@ -319,9 +319,9 @@ def test_only_read_scope_and_hashes_are_accepted(self, db): assert isinstance(self._key(db, db["a"], "not-a-hash"), psycopg2.errors.CheckViolation) -def test_webhook_tables_are_tenant_scoped_with_forced_rls(db): - """Migration 020: all four tables force row-level security; another org sees nothing.""" - tables = ("webhooks", "webhook_events", "webhook_deliveries", "notification_state") +def test_webhook_and_report_tables_are_tenant_scoped_with_forced_rls(db): + """Migrations 020 and 021: these tables force row-level security; another org sees nothing.""" + tables = ("webhooks", "webhook_events", "webhook_deliveries", "notification_state", "report_schedules") with db["owner"].cursor() as cur: cur.execute("SELECT relname, relrowsecurity, relforcerowsecurity FROM pg_class WHERE relname = ANY(%s)", (list(tables),)) diff --git a/tests/security/test_reports.py b/tests/security/test_reports.py new file mode 100644 index 0000000..6e867af --- /dev/null +++ b/tests/security/test_reports.py @@ -0,0 +1,183 @@ +""" +Scheduled reports: schedule arithmetic, the digest's content and escaping, admin-only API +with recipients limited to members, and the runner (lease, member re-check, skip empty). +Runs on both backends, like the rest of the security suite. +""" + +import asyncio +import uuid +from datetime import datetime, timedelta, timezone + +import pytest + +from libs.integrations import reports + + +def bearer(token: str) -> dict: + return {"Authorization": f"Bearer {token}"} + + +def at(text: str) -> datetime: + return datetime.fromisoformat(text).replace(tzinfo=timezone.utc) + + +class TestNextRun: + def test_daily(self): + assert reports.next_run("daily", 7, None, at("2026-10-23T06:59:00")) == at("2026-10-23T07:00:00") + assert reports.next_run("daily", 7, None, at("2026-10-23T07:00:00")) == at("2026-10-24T07:00:00") + + def test_weekly(self): # 2026-10-23 is a Friday (weekday 4) + assert reports.next_run("weekly", 8, 0, at("2026-10-23T10:00:00")) == at("2026-10-26T08:00:00") + assert reports.next_run("weekly", 8, 4, at("2026-10-23T07:00:00")) == at("2026-10-23T08:00:00") + assert reports.next_run("weekly", 8, 4, at("2026-10-23T09:00:00")) == at("2026-10-30T08:00:00") + + def test_monthly_rolls_over_the_year(self): + assert reports.next_run("monthly", 6, None, at("2026-12-01T06:00:00")) == at("2027-01-01T06:00:00") + assert reports.next_run("monthly", 6, None, at("2026-11-30T23:00:00")) == at("2026-12-01T06:00:00") + + +def _event(etype="record.matched", supplier="Zentrix Pharma", title="Recall: tablets", **extra): + payload = {"supplier": {"name": supplier}, "match": {"match_type": "name_exact"}, + "record": {"title": title, "url": "https://www.fda.gov/x", "severity": "high", + "published_at": "2026-10-20", "source": "fda_recalls"}, "record_count": 1, **extra} + return {"id": str(uuid.uuid4()), "type": etype, "payload": payload, "created_at": "2026-10-22T00:00:00+00:00"} + + +class TestDigest: + def test_counts_and_content(self): + subject, text, body = reports.build_digest("Acme", [_event(), _event("exposure.detected", "Osaka KK", "Earthquake", + match={"site": {"site": "Plant 2", "distance_km": 14}})], + at("2026-10-16T00:00:00"), at("2026-10-23T00:00:00"), "https://scirm.example") + assert "1 new record" in subject and "1 disaster exposure" in subject + assert "Zentrix Pharma" in text and "14 km from Plant 2" in text and "https://scirm.example/suppliers" in body + + def test_untrusted_text_is_escaped(self): + _, _, body = reports.build_digest("Org", [_event(supplier="", title="a\">")], + at("2026-10-16T00:00:00"), at("2026-10-23T00:00:00")) + assert "