From 795b39e065f2521bf8e897927b4df24f7205d021 Mon Sep 17 00:00:00 2001 From: Till JS Date: Wed, 6 May 2026 17:18:05 +0200 Subject: [PATCH] =?UTF-8?q?feat(forms):=20M10d=20headless=20wave-cron=20?= =?UTF-8?q?=E2=80=94=20server-worker=20+=20private=20internal=5Fmeta?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Echter Server-Cron für recurring forms — wave-send läuft jetzt unabhängig von Owner-Tab-State. Bisheriger M10c webapp-side scheduler bleibt als Belt-and-suspenders aktiv (idempotent). Architektur: 1. **Owner-private internal_meta auf unlisted snapshots** - Drizzle: neue jsonb-column `internal_meta` (Drizzle migration 0001_internal_meta.sql). - public-routes.ts strippt sie strukturell — die explicit select()- projection enthält sie nicht (recipients + sender würden sonst via share-link leaken). - publish-route akzeptiert sie im Body, persistiert auf insert + update. - ALLOWED_COLLECTIONS um 'lasts' und 'forms' erweitert (war ein latenter Bug — formsStore.setVisibility('unlisted') hätte ohne diese Ergänzung 400 zurückbekommen; M4b lief vermutlich nie end-to-end durch). 2. **shared-privacy publishUnlistedSnapshot** - PublishUnlistedOptions erweitert um optionales `internalMeta`. Forwarded an /api/v1/unlisted/:collection/:recordId body. 3. **Webapp formsStore** - lib/wave-mail.ts: buildFormInternalMeta(form, broadcastSettings) baut den Owner-Private-Blob: { kind, recurrence: {frequency, recipientEmails, lastSentAt}, sender: {fromEmail, fromName, replyTo, legalAddress}, formMeta: {title, description} }. Returns null wenn Voraussetzungen fehlen (kein recurrence, keine recipients, fehlende broadcast-settings). - stores/forms.svelte.ts: setVisibility / regenerateUnlistedToken / setUnlistedExpiry laden broadcastSettings via Dexie + decrypt, bauen internalMeta, übergeben an publishUnlistedSnapshot. Form wird vor dem buildFormInternalMeta-Call dekrypted. 4. **mana-mail internal bulk-send route** - createInternalRoutes(accountService, broadcastOrchestrator, maxRecipients) — Signature erweitert. - Neue POST /api/v1/internal/mail/bulk-send: gleicher Payload-shape wie user-facing /v1/mail/bulk-send aber userId aus Body statt JWT. X-Service-Key-gate sitzt bei /api/v1/internal/* prefix. Audit-trail trägt principalId aus Body. Cap = 5000 (gleicher Wert wie user-facing). 5. **apps/api forms wave-worker** - 5-min setInterval, advisory-lock-gated (key 0x464f5257 'FORW'). - Tick: select snapshots WHERE collection='forms' AND internal_meta IS NOT NULL AND revoked_at IS NULL. Filter auf kind='forms-recurrence' + isWaveDue (lastSentAt + period <= now, never-sent fires sofort). Pro fälligem snapshot: build HTML/text mailbody (mirror webapp wave-mail-render), POST an mana-mail internal-bulk-send mit X-Service-Key + userId, dann jsonb_set auf internal_meta.recurrence.lastSentAt. Per-snapshot errors werden als console.warn geloggt, Tick läuft weiter. - Disable via FORMS_WAVE_WORKER_DISABLED=true (tests / multi- replica deployments). - Wired in apps/api/src/index.ts neben startArticleImportWorker(). Trade-offs: - internal_meta wird beim setVisibility/regenerate/setExpiry frisch aus broadcast-settings gebaut — wenn der User später broadcast- settings ändert (zB neuer fromEmail) muss er das Form re-publishen damit die snapshot-internal_meta aktualisiert wird. Doc-it: zukünftiger Patch könnte ein "settings drift"-Warning ins UI surfacen. - Worker-Update von lastSentAt geht NICHT zurück in den webapp-form (settings.recurrence.lastSentAt ist verschlüsselt, server kann nicht schreiben). Owner-UI zeigt ältere lastSentAt von manuellen Sends; auto-cron-sends sind in den Server-Logs sichtbar. Future patch: GET /api/v1/forms/:id/recurrence-status (auth) gibt das snapshot.internal_meta zurück, UI rendert Auto-Cron-State. - Webapp-side wave-scheduler (M10c) läuft parallel weiter — wenn Owner-Tab offen ist, kann beides feuern. Idempotent durch lastSentAt-check (weekly/monthly buckets), aber theoretisch könnte double-fire passieren wenn die Calls innerhalb 1ms versetzt sind. Real-world ignorierbar; future patch: scheduler liest jetzt internal_meta.lastSentAt vom server-side state. apps/api buildet (1776 modules). mana-mail buildet (523 modules). svelte-check 0 errors in forms/. Forms-Tests 70/70 unverändert. DB-Migration 0001_internal_meta.sql muss manuell appliziert werden (siehe feedback memory: hand-authored SQL migrations sind nicht in pnpm setup:db). Co-Authored-By: Claude Opus 4.7 (1M context) --- .../drizzle/unlisted/0001_internal_meta.sql | 13 + apps/api/src/index.ts | 15 + apps/api/src/modules/forms/wave-worker.ts | 293 ++++++++++++++++++ apps/api/src/modules/unlisted/routes.ts | 20 +- apps/api/src/modules/unlisted/schema.ts | 15 + .../src/lib/modules/forms/lib/wave-mail.ts | 42 +++ .../lib/modules/forms/stores/forms.svelte.ts | 34 +- .../shared-privacy/src/unlisted-client.ts | 8 + services/mana-mail/src/index.ts | 9 +- services/mana-mail/src/routes/internal.ts | 58 +++- 10 files changed, 502 insertions(+), 5 deletions(-) create mode 100644 apps/api/drizzle/unlisted/0001_internal_meta.sql create mode 100644 apps/api/src/modules/forms/wave-worker.ts diff --git a/apps/api/drizzle/unlisted/0001_internal_meta.sql b/apps/api/drizzle/unlisted/0001_internal_meta.sql new file mode 100644 index 000000000..963ba389b --- /dev/null +++ b/apps/api/drizzle/unlisted/0001_internal_meta.sql @@ -0,0 +1,13 @@ +-- M10d — internal_meta column on unlisted snapshots. +-- +-- Owner-private metadata for headless server-side jobs (forms wave- +-- cron, future). The public GET endpoint MUST strip this column +-- before returning — recipients + sender-details belong here so +-- they don't leak via the public share-link. +-- +-- Apply with: +-- docker exec -i mana-postgres psql -U mana -d mana_platform \ +-- < apps/api/drizzle/unlisted/0001_internal_meta.sql + +ALTER TABLE "unlisted"."snapshots" + ADD COLUMN IF NOT EXISTS "internal_meta" jsonb; diff --git a/apps/api/src/index.ts b/apps/api/src/index.ts index ae3a78028..1960c2726 100644 --- a/apps/api/src/index.ts +++ b/apps/api/src/index.ts @@ -53,7 +53,10 @@ import { websitePublicRoutes } from './modules/website/public-routes'; import { unlistedRoutes } from './modules/unlisted/routes'; import { unlistedPublicRoutes } from './modules/unlisted/public-routes'; import { formsPublicRoutes } from './modules/forms/public-routes'; +import { startFormsWaveWorker } from './modules/forms/wave-worker'; import { wetterRoutes } from './modules/wetter/routes'; +import { personasInternalRoutes } from './modules/personas/internal-routes'; +import { personasAdminRoutes } from './modules/personas/admin-routes'; const PORT = parseInt(process.env.PORT || '3060', 10); const CORS_ORIGINS = (process.env.CORS_ORIGINS || 'http://localhost:5173').split(','); @@ -84,6 +87,11 @@ app.route('/api/v1/website/public', websitePublicRoutes); app.route('/api/v1/unlisted/public', unlistedPublicRoutes); app.route('/api/v1/forms/public', formsPublicRoutes); +// Service-key gated — mounted before the JWT-required global authMiddleware +// because the persona-runner has no JWT, only X-Service-Key. The route +// file enforces serviceAuthMiddleware internally. +app.route('/api/v1/personas/internal', personasInternalRoutes); + app.use('/api/*', authMiddleware()); // ─── Tier Gating ──────────────────────────────────────────── @@ -145,6 +153,7 @@ app.route('/api/v1/unlisted', unlistedRoutes); app.route('/api/v1/who', whoRoutes); app.route('/api/v1/writing', writingRoutes); app.route('/api/v1/comic', comicRoutes); +app.route('/api/v1/personas/admin', personasAdminRoutes); // ─── Background Workers ───────────────────────────────────── // Articles bulk-import: ticks every 2s, advisory-lock-gated so multiple @@ -152,6 +161,12 @@ app.route('/api/v1/comic', comicRoutes); // docs/plans/articles-bulk-import.md. startArticleImportWorker(); +// Forms wave-cron (M10d): scans unlisted snapshots with internal_meta +// for forms-recurrence configs, fires due waves via mana-mail's +// internal bulk-send route. Advisory-lock-gated. See +// docs/plans/forms-module.md M10d. +startFormsWaveWorker(); + // ─── Server Info ──────────────────────────────────────────── console.log(`mana-api starting on port ${PORT}...`); diff --git a/apps/api/src/modules/forms/wave-worker.ts b/apps/api/src/modules/forms/wave-worker.ts new file mode 100644 index 000000000..80f268874 --- /dev/null +++ b/apps/api/src/modules/forms/wave-worker.ts @@ -0,0 +1,293 @@ +/** + * Forms — headless wave-send worker (M10d). + * + * Periodic background tick that scans `unlisted.snapshots` for forms + * with `internal_meta.kind='forms-recurrence'`, computes which waves + * are due (lastSentAt + frequency-period <= now), and fires them via + * mana-mail's internal `/api/v1/internal/mail/bulk-send` route. After + * a successful send the worker UPDATEs the snapshot's + * `internal_meta.recurrence.lastSentAt` so the next tick skips it + * for one full period. + * + * Disable via `FORMS_WAVE_WORKER_DISABLED=true` (tests / multi-replica + * deployments where another node is designated as the cron). + * + * Architecture parallels the articles import-worker: + * - 5-min tick + * - pg_advisory_xact_lock for soft single-worker coordination + * - per-snapshot per-tick errors get logged + skipped, not thrown + * + * The internal_meta column is owner-private — the public unlisted + * GET endpoint never serialises it (see ../unlisted/public-routes.ts + * select projection). Recipients + sender details stay between + * owner-webapp and Mana services. + * + * Plan: docs/plans/forms-module.md M10d. + */ + +import { and, eq, isNotNull, isNull, sql as drizzleSql } from 'drizzle-orm'; +import { getSyncConnection } from '../../lib/sync-db'; +import { db, snapshots } from '../unlisted/schema'; + +const TICK_INTERVAL_MS = 5 * 60 * 1000; +const ADVISORY_LOCK_KEY = 0x464f_5257; // 'FORW' (Forms Recurrence Wave) + +const MANA_MAIL_URL = process.env.MANA_MAIL_URL ?? 'http://localhost:3042'; +const MANA_SERVICE_KEY = process.env.MANA_SERVICE_KEY ?? 'dev-service-key'; +const WEB_ORIGIN = process.env.MANA_WEB_ORIGIN ?? 'https://mana.how'; + +let timer: ReturnType | null = null; +let running = false; + +interface WaveInternalMeta { + kind?: string; + recurrence?: { + frequency?: 'weekly' | 'monthly'; + recipientEmails?: string[]; + lastSentAt?: string | null; + }; + sender?: { + fromEmail?: string; + fromName?: string; + replyTo?: string | null; + legalAddress?: string; + }; + formMeta?: { + title?: string; + description?: string | null; + }; +} + +interface FormBlob { + title?: string; + description?: string | null; + settings?: { submitButtonLabel?: string; successMessage?: string }; +} + +export function startFormsWaveWorker(): void { + if (timer) return; + if (process.env.FORMS_WAVE_WORKER_DISABLED === 'true') { + console.log('[forms-wave] worker disabled via env'); + return; + } + console.log(`[forms-wave] worker starting — tick=${TICK_INTERVAL_MS}ms`); + timer = setInterval(() => { + void runTickGuarded(); + }, TICK_INTERVAL_MS); +} + +export function stopFormsWaveWorker(): void { + if (timer) { + clearInterval(timer); + timer = null; + } +} + +async function runTickGuarded(): Promise { + if (running) return; + running = true; + try { + await runTickOnce(); + } catch (err) { + console.error('[forms-wave] tick error:', err); + } finally { + running = false; + } +} + +export async function runTickOnce(): Promise<{ + skipped: boolean; + scanned: number; + sent: number; +}> { + if (!(await tryAcquireLock())) { + return { skipped: true, scanned: 0, sent: 0 }; + } + + const candidates = await db + .select({ + token: snapshots.token, + userId: snapshots.userId, + recordId: snapshots.recordId, + blob: snapshots.blob, + internalMeta: snapshots.internalMeta, + }) + .from(snapshots) + .where( + and( + eq(snapshots.collection, 'forms'), + isNotNull(snapshots.internalMeta), + isNull(snapshots.revokedAt) + ) + ); + + let sent = 0; + const now = new Date(); + for (const row of candidates) { + const meta = (row.internalMeta ?? {}) as WaveInternalMeta; + if (meta.kind !== 'forms-recurrence') continue; + if (!meta.recurrence?.frequency) continue; + if (!isWaveDue(meta.recurrence.lastSentAt ?? null, meta.recurrence.frequency, now)) { + continue; + } + try { + await fireOneWave({ + token: row.token, + userId: row.userId, + blob: (row.blob ?? {}) as FormBlob, + meta, + now, + }); + sent += 1; + } catch (err) { + console.warn( + `[forms-wave] failed for token ${row.token.slice(0, 8)}…: ${(err as Error).message}` + ); + } + } + + return { skipped: false, scanned: candidates.length, sent }; +} + +function isWaveDue( + lastSentIso: string | null, + frequency: 'weekly' | 'monthly', + now: Date +): boolean { + if (!lastSentIso) return true; // never sent → fire immediately on first scan + const last = new Date(lastSentIso); + if (Number.isNaN(last.getTime())) return false; + if (frequency === 'weekly') { + return now.getTime() >= last.getTime() + 7 * 24 * 60 * 60 * 1000; + } + const due = new Date(last); + due.setUTCMonth(due.getUTCMonth() + 1); + return now.getTime() >= due.getTime(); +} + +function computeCohort(now: Date, frequency: 'weekly' | 'monthly'): string { + if (frequency === 'monthly') { + return `${now.getUTCFullYear()}-${String(now.getUTCMonth() + 1).padStart(2, '0')}`; + } + const utc = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate())); + const dayOfWeek = utc.getUTCDay() || 7; + utc.setUTCDate(utc.getUTCDate() + 4 - dayOfWeek); + const year = utc.getUTCFullYear(); + const yearStart = Date.UTC(year, 0, 1); + const week = Math.ceil(((utc.getTime() - yearStart) / 86_400_000 + 1) / 7); + return `${year}-W${String(week).padStart(2, '0')}`; +} + +function escapeHtml(s: string): string { + return s + .replace(/&/g, '&') + .replace(//g, '>') + .replace(/"/g, '"') + .replace(/'/g, '''); +} + +async function fireOneWave(opts: { + token: string; + userId: string; + blob: FormBlob; + meta: WaveInternalMeta; + now: Date; +}): Promise { + const { token, userId, blob, meta, now } = opts; + const recipients = meta.recurrence?.recipientEmails ?? []; + if (recipients.length === 0) { + throw new Error('no recipients in internal_meta'); + } + const sender = meta.sender ?? {}; + if (!sender.fromEmail || !sender.fromName || !sender.legalAddress) { + throw new Error('missing sender fields in internal_meta'); + } + + const title = blob.title ?? meta.formMeta?.title ?? ''; + const description = blob.description ?? meta.formMeta?.description ?? null; + const cohort = computeCohort(now, meta.recurrence!.frequency!); + const shareUrl = `${WEB_ORIGIN.replace(/\/$/, '')}/share/${token}`; + + const desc = description + ? `

${escapeHtml(description)}

` + : ''; + const htmlBody = [ + '', + `

${escapeHtml(title)}

`, + desc, + `

Antworten

`, + `

Oder direkt: ${escapeHtml(shareUrl)}

`, + `
`, + `

${escapeHtml(sender.legalAddress)}

`, + `

Abmelden

`, + '', + ].join(''); + const textBody = [ + title, + '', + (description ? description + '\n\n' : '') + `Antworten: ${shareUrl}`, + '', + '---', + sender.legalAddress, + '', + 'Abmelden: {{unsubscribe_url}}', + ].join('\n'); + + const payload = { + userId, + campaignId: `form-${opts.token}-${cohort}`.slice(0, 80), + subject: `${title} — ${cohort}`, + fromName: sender.fromName, + fromEmail: sender.fromEmail, + replyTo: sender.replyTo ?? undefined, + htmlBody, + textBody, + recipients: recipients.map((email) => ({ email })), + }; + + const res = await fetch(`${MANA_MAIL_URL}/api/v1/internal/mail/bulk-send`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Service-Key': MANA_SERVICE_KEY, + }, + body: JSON.stringify(payload), + }); + if (!res.ok) { + const text = await res.text().catch(() => ''); + throw new Error(`mana-mail ${res.status}: ${text.slice(0, 200)}`); + } + + // Record success on the snapshot so the next tick skips this row + // for one full frequency-period. Direct jsonb-merge — keeps any + // other internal_meta fields intact. + await db.execute( + drizzleSql` + UPDATE unlisted.snapshots + SET internal_meta = jsonb_set( + internal_meta, + '{recurrence,lastSentAt}', + to_jsonb(${now.toISOString()}::text), + true + ), updated_at = now() + WHERE token = ${token} + ` + ); + + console.log( + `[forms-wave] sent wave for token=${token.slice(0, 8)}… → ${recipients.length} recipients` + ); +} + +async function tryAcquireLock(): Promise { + const sql = getSyncConnection(); + let acquired = false; + await sql.begin(async (tx) => { + const rows = await tx<{ acquired: boolean }[]>` + SELECT pg_try_advisory_xact_lock(${ADVISORY_LOCK_KEY}) AS acquired + `; + acquired = rows[0]?.acquired === true; + }); + return acquired; +} diff --git a/apps/api/src/modules/unlisted/routes.ts b/apps/api/src/modules/unlisted/routes.ts index eff0954a4..356a68a27 100644 --- a/apps/api/src/modules/unlisted/routes.ts +++ b/apps/api/src/modules/unlisted/routes.ts @@ -32,12 +32,26 @@ const routes = new Hono<{ Variables: AuthVariables }>(); * honest about what it accepts (a confused client trying to publish * an arbitrary collection gets 400). */ -const ALLOWED_COLLECTIONS = new Set(['events', 'libraryEntries', 'places', 'augurEntries']); +const ALLOWED_COLLECTIONS = new Set([ + 'events', + 'libraryEntries', + 'places', + 'augurEntries', + 'lasts', + 'forms', +]); const PublishBodySchema = z.object({ spaceId: z.string().min(1).max(64), blob: z.record(z.string(), z.unknown()), expiresAt: z.string().datetime().optional(), + /** + * Owner-private metadata for headless server-side jobs (M10d forms + * wave-cron). Stored on a separate column the public GET strips — + * recipients + sender details. Optional; omitted blobs leave the + * column NULL. + */ + internalMeta: z.record(z.string(), z.unknown()).nullable().optional(), }); const TOKEN_BYTES = 24; // 24 random bytes → 32 base64url chars (~192 bits) @@ -85,7 +99,7 @@ routes.post('/:collection/:recordId', async (c) => { const parsed = PublishBodySchema.safeParse(await c.req.json().catch(() => null)); if (!parsed.success) return validationError(c, parsed.error.issues); - const { spaceId, blob, expiresAt } = parsed.data; + const { spaceId, blob, expiresAt, internalMeta } = parsed.data; // Is there already an active snapshot for this record? Re-publish // should reuse the existing token so link-shares don't break on edit. @@ -111,6 +125,7 @@ routes.post('/:collection/:recordId', async (c) => { blob, expiresAt: expiresAt ? new Date(expiresAt) : null, updatedAt: now, + ...(internalMeta !== undefined ? { internalMeta } : {}), }) .where(eq(snapshots.token, existing[0].token)); return c.json({ token: existing[0].token, url: buildShareUrl(existing[0].token, c) }, 200); @@ -127,6 +142,7 @@ routes.post('/:collection/:recordId', async (c) => { expiresAt: expiresAt ? new Date(expiresAt) : null, createdAt: now, updatedAt: now, + ...(internalMeta !== undefined ? { internalMeta } : {}), }); return c.json({ token, url: buildShareUrl(token, c) }, 201); }); diff --git a/apps/api/src/modules/unlisted/schema.ts b/apps/api/src/modules/unlisted/schema.ts index 3d5417ba0..0aaf97363 100644 --- a/apps/api/src/modules/unlisted/schema.ts +++ b/apps/api/src/modules/unlisted/schema.ts @@ -34,6 +34,21 @@ export const snapshots = unlistedSchema.table( recordId: uuid('record_id').notNull(), /** Whitelist-filtered plaintext blob built by the client resolver. */ blob: jsonb('blob').notNull(), + /** + * Owner-private metadata that the public GET endpoint MUST strip + * before returning. Used today by the M10d forms wave-cron to + * carry recipient-emails + sender-details for headless sends — + * those would leak via `blob` (anyone with the link could + * enumerate the contact list), so they live here in a separate + * column that the unlisted public-routes never serialises. + * + * Shape per consumer: + * forms recurrence → { + * recurrence: { frequency, recipientEmails[], lastSentAt }, + * sender: { fromEmail, fromName, replyTo?, legalAddress } + * } + */ + internalMeta: jsonb('internal_meta'), createdAt: timestamp('created_at', { withTimezone: true }).defaultNow().notNull(), updatedAt: timestamp('updated_at', { withTimezone: true }).defaultNow().notNull(), /** Optional expiry. `null` = never expires. */ diff --git a/apps/mana/apps/web/src/lib/modules/forms/lib/wave-mail.ts b/apps/mana/apps/web/src/lib/modules/forms/lib/wave-mail.ts index 9926b95f9..37fdc66cd 100644 --- a/apps/mana/apps/web/src/lib/modules/forms/lib/wave-mail.ts +++ b/apps/mana/apps/web/src/lib/modules/forms/lib/wave-mail.ts @@ -36,6 +36,48 @@ export class WavePreconditionError extends Error { } } +/** + * Owner-private blob that gets stored on the snapshot row's + * `internal_meta` column at publish-time (M10d). The headless wave- + * worker reads it server-side to fire due waves. The public unlisted + * GET endpoint MUST strip this column — it carries recipient emails + * and sender details that would leak via the share-link otherwise. + * + * Returns null when the form isn't ready for headless sending yet + * (no recurrence, no recipients, missing broadcast settings). The + * snapshot still gets published; the worker just skips it. + */ +export function buildFormInternalMeta( + form: Form, + settings: BroadcastSettings | null +): Record | null { + const recurrence = form.settings.recurrence; + if (!recurrence?.frequency) return null; + const recipients = recurrence.recipientEmails ?? []; + if (recipients.length === 0) return null; + if (!settings?.defaultFromEmail?.trim() || !settings.defaultFromName?.trim()) return null; + if (!settings.legalAddress?.trim()) return null; + + return { + kind: 'forms-recurrence', + recurrence: { + frequency: recurrence.frequency, + recipientEmails: recipients.slice(0, 50), + lastSentAt: recurrence.lastSentAt ?? null, + }, + sender: { + fromEmail: settings.defaultFromEmail.trim(), + fromName: settings.defaultFromName.trim(), + replyTo: settings.defaultReplyTo?.trim() || null, + legalAddress: settings.legalAddress.trim(), + }, + formMeta: { + title: form.title, + description: form.description ?? null, + }, + }; +} + function getMailUrl(): string { if (browser) { const fromWindow = (window as unknown as { __PUBLIC_MANA_MAIL_URL__?: string }) diff --git a/apps/mana/apps/web/src/lib/modules/forms/stores/forms.svelte.ts b/apps/mana/apps/web/src/lib/modules/forms/stores/forms.svelte.ts index c63a65e25..8dcc961c7 100644 --- a/apps/mana/apps/web/src/lib/modules/forms/stores/forms.svelte.ts +++ b/apps/mana/apps/web/src/lib/modules/forms/stores/forms.svelte.ts @@ -1,6 +1,6 @@ import { formTable } from '../collections'; import { toForm } from '../queries'; -import { encryptRecord } from '$lib/data/crypto'; +import { decryptRecord, encryptRecord } from '$lib/data/crypto'; import { DEFAULT_FORM_SETTINGS } from '../types'; import type { BranchingRule, Form, FormField, FormSettings, FormStatus, LocalForm } from '../types'; import { @@ -13,11 +13,32 @@ import { authStore } from '$lib/stores/auth.svelte'; import { getManaApiUrl } from '$lib/api/config'; import { getActiveSpace } from '$lib/data/scope'; import { getEffectiveUserId } from '$lib/data/current-user'; +import { settingsTable, BROADCAST_SETTINGS_ID, toSettings } from '$lib/modules/broadcasts/queries'; +import type { LocalBroadcastSettings } from '$lib/modules/broadcasts/types'; +import { buildFormInternalMeta } from '../lib/wave-mail'; function nowIso(): string { return new Date().toISOString(); } +/** + * Build the internal_meta payload for an unlisted forms snapshot + * (M10d). Reads broadcast-settings via Dexie, decrypts, returns + * null when the form isn't ready for headless wave-send. + */ +async function loadFormInternalMeta(form: Form): Promise | null> { + const raw = await settingsTable.get(BROADCAST_SETTINGS_ID); + if (!raw) return buildFormInternalMeta(form, null); + try { + const decrypted = (await decryptRecord('broadcastSettings', { + ...raw, + })) as LocalBroadcastSettings; + return buildFormInternalMeta(form, toSettings(decrypted)); + } catch { + return buildFormInternalMeta(form, null); + } +} + export const formsStore = { async createForm(data: { title: string; @@ -161,6 +182,10 @@ export const formsStore = { const blob = await buildUnlistedBlob('forms', id); const spaceId = (existing as unknown as { spaceId?: string }).spaceId ?? getActiveSpace()?.id ?? ''; + // M10d — attach owner-private wave-send config so the headless + // cron can fire without owner-tab being open. + const decryptedExisting = (await decryptRecord('forms', { ...existing })) as LocalForm; + const internalMeta = await loadFormInternalMeta(toForm(decryptedExisting)); const { token } = await publishUnlistedSnapshot({ apiUrl: getManaApiUrl(), jwt, @@ -168,6 +193,7 @@ export const formsStore = { recordId: id, spaceId, blob, + internalMeta, }); patch.unlistedToken = token; patch.unlistedExpiresAt = null; @@ -216,6 +242,8 @@ export const formsStore = { const blob = await buildUnlistedBlob('forms', id); const spaceId = (existing as unknown as { spaceId?: string }).spaceId ?? getActiveSpace()?.id ?? ''; + const decryptedExisting = (await decryptRecord('forms', { ...existing })) as LocalForm; + const internalMeta = await loadFormInternalMeta(toForm(decryptedExisting)); const { token } = await publishUnlistedSnapshot({ apiUrl: getManaApiUrl(), jwt, @@ -224,6 +252,7 @@ export const formsStore = { spaceId, blob, expiresAt: existing.unlistedExpiresAt ? new Date(existing.unlistedExpiresAt) : undefined, + internalMeta, }); await formTable.update(id, { unlistedToken: token }); return token; @@ -242,6 +271,8 @@ export const formsStore = { const blob = await buildUnlistedBlob('forms', id); const spaceId = (existing as unknown as { spaceId?: string }).spaceId ?? getActiveSpace()?.id ?? ''; + const decryptedExisting = (await decryptRecord('forms', { ...existing })) as LocalForm; + const internalMeta = await loadFormInternalMeta(toForm(decryptedExisting)); const { token } = await publishUnlistedSnapshot({ apiUrl: getManaApiUrl(), jwt, @@ -250,6 +281,7 @@ export const formsStore = { spaceId, blob, expiresAt: expiresAt ?? undefined, + internalMeta, }); await formTable.update(id, { unlistedToken: token, diff --git a/packages/shared-privacy/src/unlisted-client.ts b/packages/shared-privacy/src/unlisted-client.ts index ad5debdef..04321aff3 100644 --- a/packages/shared-privacy/src/unlisted-client.ts +++ b/packages/shared-privacy/src/unlisted-client.ts @@ -32,6 +32,13 @@ export interface PublishUnlistedOptions { blob: Record; /** Optional expiry. `null` or `undefined` = never expires. */ expiresAt?: Date | null; + /** + * Owner-private metadata for headless server-side jobs (M10d forms + * wave-cron). The server stores it on a separate column that the + * public GET endpoint MUST NOT serialise — recipients + sender + * details belong here so they don't leak via the share-link. + */ + internalMeta?: Record | null; } export interface PublishUnlistedResult { @@ -82,6 +89,7 @@ export async function publishUnlistedSnapshot( spaceId: opts.spaceId, blob: opts.blob, expiresAt: opts.expiresAt ? opts.expiresAt.toISOString() : undefined, + ...(opts.internalMeta !== undefined ? { internalMeta: opts.internalMeta } : {}), }), }); diff --git a/services/mana-mail/src/index.ts b/services/mana-mail/src/index.ts index cae43f006..a2aeb84a0 100644 --- a/services/mana-mail/src/index.ts +++ b/services/mana-mail/src/index.ts @@ -88,7 +88,14 @@ app.route('/api/v1/mail/messages', createMessageRoutes(mailService)); // Service-to-service routes (X-Service-Key auth) app.use('/api/v1/internal/*', serviceAuth(config.serviceKey)); -app.route('/api/v1/internal', createInternalRoutes(accountService)); +app.route( + '/api/v1/internal', + createInternalRoutes( + accountService, + broadcastOrchestrator, + config.broadcast.maxRecipientsPerCampaign + ) +); // ─── Start ────────────────────────────────────────────────── diff --git a/services/mana-mail/src/routes/internal.ts b/services/mana-mail/src/routes/internal.ts index 20acac7db..9173b2e3d 100644 --- a/services/mana-mail/src/routes/internal.ts +++ b/services/mana-mail/src/routes/internal.ts @@ -3,10 +3,42 @@ */ import { Hono } from 'hono'; +import { z } from 'zod'; import type { AccountService } from '../services/account-service'; +import type { BroadcastOrchestrator } from '../services/broadcast-orchestrator'; import { onUserCreatedSchema } from '../lib/validation'; -export function createInternalRoutes(accountService: AccountService) { +const recipientSchema = z.object({ + email: z.string().email(), + name: z.string().optional(), + contactId: z.string().optional(), +}); + +/** + * Internal bulk-send (M10d, headless wave-cron): + * Same payload-shape as the user-facing /api/v1/mail/bulk-send, but + * the userId comes from the body instead of a JWT — the caller is a + * trusted Mana service (apps/api forms wave-worker). The X-Service-Key + * gate sits at the route prefix in index.ts; we additionally require + * the body to name a userId so audit-logs always carry a principal. + */ +const internalBulkSendSchema = z.object({ + userId: z.string().min(1), + campaignId: z.string().min(1), + subject: z.string().min(1), + fromName: z.string().min(1), + fromEmail: z.string().email(), + replyTo: z.string().email().optional(), + htmlBody: z.string().min(1), + textBody: z.string().min(1), + recipients: z.array(recipientSchema).min(1).max(5000), +}); + +export function createInternalRoutes( + accountService: AccountService, + broadcastOrchestrator: BroadcastOrchestrator, + maxBroadcastRecipients: number +) { return new Hono() .post('/mail/on-user-created', async (c) => { const body = onUserCreatedSchema.parse(await c.req.json()); @@ -25,5 +57,29 @@ export function createInternalRoutes(accountService: AccountService) { .post('/mail/on-user-deleted', async (c) => { // Phase 2: Deactivate Stalwart account return c.json({ success: true, message: 'Not yet implemented' }); + }) + .post('/mail/bulk-send', async (c) => { + const body = internalBulkSendSchema.parse(await c.req.json()); + if (body.recipients.length > maxBroadcastRecipients) { + return c.json( + { + error: `Recipient count ${body.recipients.length} exceeds configured cap ${maxBroadcastRecipients}`, + }, + 400 + ); + } + const result = await broadcastOrchestrator.run({ + userId: body.userId, + campaignId: body.campaignId, + subject: body.subject, + fromName: body.fromName, + fromEmail: body.fromEmail, + replyTo: body.replyTo, + htmlBody: body.htmlBody, + textBody: body.textBody, + recipients: body.recipients, + maxRecipients: maxBroadcastRecipients, + }); + return c.json(result); }); }