diff --git a/src/services/api/handlers.ts b/src/services/api/handlers.ts index 3a70720..2279ccf 100644 --- a/src/services/api/handlers.ts +++ b/src/services/api/handlers.ts @@ -31,6 +31,7 @@ import { type ContactChannel, PreferencesSchema, type NotificationRequestedPayload, + type NotificationDispatchedPayload, type NotificationTarget, buildStreamEvent, CreateWorkflowSchema, @@ -38,6 +39,7 @@ import { STREAMS, PUBSUB_CHANNELS, } from "@/contracts/index.js"; +import { renderWithTemplate, type TemplateCache } from "@/templates/index.js"; import { readJsonBody, sendJson, sendNoContent, sendValidationError } from "./http.js"; import type { RouteContext } from "./router.js"; import { globalEmitter, getPriorityBucket, normaliseTarget } from "@/shared/index.js"; @@ -66,6 +68,8 @@ export interface Deps { logger: Logger; redis: RedisClient; producers: Record; + outboundProducers?: Record; + templateCache?: TemplateCache; userRepo: UserRepository; contactRepo: ContactRepository; templateRepo: TemplateRepository; @@ -349,9 +353,9 @@ export function createHandlers(deps: Deps) { // Resolve the (unresolved) target. Inline users are synced first. let targets: NotificationTarget[] = []; + const inlineUsersToSync: InlineUserLike[] = []; if (body.user !== undefined) { const users = Array.isArray(body.user) ? body.user : [body.user]; - const inlineUsersToSync: InlineUserLike[] = []; for (const u of users) { if (typeof u === "string") { targets.push({ type: "user", userId: u }); @@ -377,6 +381,100 @@ export function createHandlers(deps: Deps) { const baseNotificationId = (req.headers["x-idempotency-key"] as string) || randomUUID(); const priority = body.priority ?? "normal"; + + // ─── Approach B: Transactional Direct Fast-Path ───────────────────────── + // If critical priority, unscheduled, and direct contact addresses are provided + // via inline users, bypass Enricher and Decision streams and publish directly + // to OUTBOUND_CRITICAL for immediate delivery. + if ( + priority === "critical" && + !body.sendAt && + deps.outboundProducers?.critical && + inlineUsersToSync.length > 0 && + inlineUsersToSync.length === targets.length + ) { + const requestedChannels = + body.channels && body.channels.length > 0 ? body.channels : (["email"] as ContactChannel[]); + + const dbTemplate = deps.templateCache + ? await deps.templateCache.getCachedTemplate(ctx.projectId!, body.template) + : await deps.templateRepo.findById(ctx.projectId!, body.template); + + const rendered = renderWithTemplate(dbTemplate, body.data ?? {}); + const fastPathEvents: any[] = []; + let lastFastNotificationId = ""; + + for (let idx = 0; idx < inlineUsersToSync.length; idx++) { + const u = inlineUsersToSync[idx]!; + const notificationId = + targets.length > 1 ? `${baseNotificationId}-${idx}` : baseNotificationId; + lastFastNotificationId = notificationId; + + const contacts = contactsOf(u); + const activeContacts = contacts.filter((c) => requestedChannels.includes(c.channel)); + + for (const c of activeContacts) { + const taskPayload: NotificationDispatchedPayload = { + projectId: ctx.projectId!, + taskId: `${notificationId}:${c.target || randomUUID()}`, + enrichedEventId: notificationId, + recipientId: u.id, + channel: c.channel, + priority: "critical", + templateId: body.template, + templateVariables: body.data ?? {}, + aiPrompts: body.aiPrompts, + renderedContent: rendered, + destination: c.target, + deliveryOptions: { + maxAttempts: 3, + timeoutMs: 10_000, + }, + campaignId: body.campaign, + }; + + fastPathEvents.push( + buildStreamEvent( + "notification.dispatched", + taskPayload as Record, + "api", + notificationId, + ), + ); + } + } + + if (fastPathEvents.length > 0) { + const { messageIds: fastMessageIds } = + await deps.outboundProducers.critical.publishBatch(fastPathEvents); + deps.logger.info( + { count: fastPathEvents.length, priority: "critical" }, + "notification direct fast-path dispatched", + ); + metrics.messagesPublished.inc( + { channel: "api", priority: "critical" }, + fastPathEvents.length, + ); + + if (targets.length === 1) { + sendJson(res, 202, { + messageId: fastMessageIds[0], + notificationId: lastFastNotificationId, + target: targets[0], + ...(body.campaign ? { campaign: body.campaign } : {}), + }); + } else { + sendJson(res, 202, { + batchSize: targets.length, + messageIds: fastMessageIds, + notificationIdsBase: baseNotificationId, + ...(body.campaign ? { campaign: body.campaign } : {}), + }); + } + return; + } + } + const p = getPriorityBucket(priority); const producer = deps.producers[p] ?? deps.producers.normal!; diff --git a/src/services/api/main.ts b/src/services/api/main.ts index 398652e..558fa90 100644 --- a/src/services/api/main.ts +++ b/src/services/api/main.ts @@ -14,6 +14,7 @@ import { AdminUserRepository, } from "@/repositories/index.js"; import { STREAMS } from "@/contracts/index.js"; +import { TemplateCache } from "@/templates/index.js"; import { readJsonBody, readRawBody, sendJson, HttpError } from "./http.js"; import { Router } from "./router.js"; import { createHandlers } from "./handlers.js"; @@ -258,13 +259,28 @@ export async function startApiServer() { } } + const outboundProducers = { + critical: new StreamProducer({ + redis: redis.native, + stream: STREAMS.OUTBOUND_CRITICAL, + logger, + }), + normal: new StreamProducer({ redis: redis.native, stream: STREAMS.OUTBOUND_NORMAL, logger }), + low: new StreamProducer({ redis: redis.native, stream: STREAMS.OUTBOUND_LOW, logger }), + }; + + const templateRepo = new TemplateRepository(db); + const templateCache = new TemplateCache(templateRepo); + deps = { logger, redis, producers, + outboundProducers, userRepo: new UserRepository(db), contactRepo: new ContactRepository(db), - templateRepo: new TemplateRepository(db), + templateRepo, + templateCache, projectRepo: new ProjectRepository(db), workflowRepo: new WorkflowRepository(db), segmentRepo: new SegmentRepository(db), diff --git a/src/services/enricher/main.ts b/src/services/enricher/main.ts index d8b4f9d..331a532 100644 --- a/src/services/enricher/main.ts +++ b/src/services/enricher/main.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import { loadEnv, readBaseConfig } from "@/index.js"; import { createLogger } from "@/index.js"; import { RedisClient } from "@/index.js"; @@ -16,18 +17,25 @@ import { buildStreamEvent, type NotificationCreatedPayload, type NotificationEnrichedPayload, + type NotificationDispatchedPayload, } from "@/index.js"; import { type StreamName } from "@/contracts/streams.js"; import { IdempotencyGuard } from "@/index.js"; import { createDatabase } from "@/db/index.js"; +import { scheduledPayloads } from "@/db/schema.js"; import { UserRepository, PreferenceRepository, TemplateRepository, ContactRepository, } from "@/index.js"; -import { TemplateCache } from "@/templates/index.js"; -import { getPriorityBucket, type WorkerOptions } from "@/shared/index.js"; +import { TemplateCache, renderWithTemplate } from "@/templates/index.js"; +import { isInQuietHours } from "@/services/engine/main.js"; +import { + getPriorityBucket, + globalEmitter as defaultGlobalEmitter, + type WorkerOptions, +} from "@/shared/index.js"; import { startHealthReporter } from "@/workers/index.js"; // ─── Bootstrap ───────────────────────────────────────────────────────────── @@ -47,20 +55,28 @@ let healthInterval: NodeJS.Timeout | null = null; export interface EnricherWorkerOptions extends WorkerOptions { producers: any; + outboundProducers?: any; + scheduledProducer?: any; idempotency: any; userRepo: any; prefRepo: any; contactRepo: any; templateCache: TemplateCache; + db?: any; + globalEmitter?: any; } export class EnricherWorker extends BaseWorker { private readonly producers: any; + private readonly outboundProducers?: any; + private readonly scheduledProducer?: any; private readonly idempotency: any; private readonly userRepo: any; private readonly prefRepo: any; private readonly contactRepo: any; private readonly templateCache: TemplateCache; + private readonly db?: any; + private readonly globalEmitter?: any; private userBatch: { projectId: string; @@ -125,11 +141,15 @@ export class EnricherWorker extends BaseWorker { constructor(options: EnricherWorkerOptions) { super(options); this.producers = options.producers; + this.outboundProducers = options.outboundProducers; + this.scheduledProducer = options.scheduledProducer; this.idempotency = options.idempotency; this.userRepo = options.userRepo; this.prefRepo = options.prefRepo; this.contactRepo = options.contactRepo; this.templateCache = options.templateCache; + this.db = options.db; + this.globalEmitter = options.globalEmitter ?? defaultGlobalEmitter; this.flushTimer = setInterval(() => void this.flushWorkerBuffers(), 5); } @@ -453,21 +473,133 @@ export class EnricherWorker extends BaseWorker { enrichedPayload.priority = msgPriority; - batchedEvents[p].push( - buildStreamEvent( - "notification.enriched", - enrichedPayload as Record, - "enricher", - event.metadata.traceId, - ), - ); + if (this.outboundProducers) { + // Topic opt-out check + const isOptedOut = topics.some((t) => profile.preferences?.topics?.[t] === false); + if (isOptedOut) { + this.logger.info( + { recipientId: profile.userId, channel }, + "user opted out of topic — dropping", + ); + this.globalEmitter?.emit("notification:skipped", { + projectId: raw.projectId, + eventId: event.id, + recipientId: profile.userId, + reason: "user_opted_out", + }); + continue; + } + + // Channel disabled check + if (profile.preferences?.channels?.[channel] === false) { + this.logger.info( + { recipientId: profile.userId, channel }, + "user disabled channel — dropping", + ); + this.globalEmitter?.emit("notification:skipped", { + projectId: raw.projectId, + eventId: event.id, + recipientId: profile.userId, + reason: "channel_disabled", + }); + continue; + } + + // Quiet hours check + let scheduledAt = raw.scheduledAt; + const qh = profile.preferences?.quietHours; + if (qh && qh.length > 0 && msgPriority !== "critical") { + const qhResult = isInQuietHours(profile.timezone ?? "UTC", qh); + if (qhResult.inQuietHours && qhResult.nextActiveTime) { + scheduledAt = qhResult.nextActiveTime.toISOString(); + } + } + + const rendered = renderWithTemplate(template, raw.data ?? {}); + const now = Date.now(); + const isScheduled = scheduledAt && new Date(scheduledAt).getTime() > now; + const taskId = `${event.id}:${destination || randomUUID()}`; + + const taskPayload: NotificationDispatchedPayload = { + projectId: raw.projectId, + taskId, + enrichedEventId: event.id, + recipientId: profile.userId, + channel: channel as any, + priority: msgPriority, + templateId: raw.templateId, + templateVariables: raw.data ?? {}, + aiPrompts: raw.aiPrompts, + recipient: enrichedPayload.recipient, + renderedContent: rendered, + destination, + deliveryOptions: { + maxAttempts: 3, + timeoutMs: 10_000, + }, + fallbackChain: fallbackChain?.length ? fallbackChain : undefined, + campaignId: raw.campaignId, + }; + + if (isScheduled && this.scheduledProducer) { + if (this.db) { + await this.db + .insert(scheduledPayloads) + .values({ + taskId, + payload: { + ...taskPayload, + scheduledAt, + }, + }) + .onConflictDoNothing() + .catch((err: any) => + this.logger.error({ err, taskId }, "failed to cache scheduled payload"), + ); + } + + const scheduledEnvelope = buildStreamEvent( + "notification.scheduled", + { + projectId: raw.projectId, + enrichedEventId: event.id, + taskId, + scheduledAt, + }, + "enricher", + event.metadata.traceId, + ); + await this.scheduledProducer.publish(scheduledEnvelope); + continue; + } + + batchedEvents[p].push( + buildStreamEvent( + "notification.dispatched", + taskPayload as Record, + "enricher", + event.metadata.traceId, + ), + ); + } else { + batchedEvents[p].push( + buildStreamEvent( + "notification.enriched", + enrichedPayload as Record, + "enricher", + event.metadata.traceId, + ), + ); + } } } } for (const p of ["critical", "normal", "low"] as const) { if (batchedEvents[p].length > 0) { - const producer = this.producers[p] ?? this.producers.normal; + const producer = this.outboundProducers + ? (this.outboundProducers[p] ?? this.outboundProducers.normal) + : (this.producers[p] ?? this.producers.normal); for (const ev of batchedEvents[p]) { publishPromises.push( new Promise((resolve, reject) => { @@ -540,6 +672,22 @@ export async function startEnricherWorker() { low: new StreamProducer({ redis: redis.native, stream: STREAMS.ENRICHED_LOW, logger }), }; + const outboundProducers = { + critical: new StreamProducer({ + redis: redis.native, + stream: STREAMS.OUTBOUND_CRITICAL, + logger, + }), + normal: new StreamProducer({ redis: redis.native, stream: STREAMS.OUTBOUND_NORMAL, logger }), + low: new StreamProducer({ redis: redis.native, stream: STREAMS.OUTBOUND_LOW, logger }), + }; + + const scheduledProducer = new StreamProducer({ + redis: redis.native, + stream: STREAMS.SCHEDULED, + logger, + }); + const idempotency = new IdempotencyGuard({ redis: redis.native, keyPrefix: "notif:processed:enricher", @@ -560,8 +708,6 @@ export async function startEnricherWorker() { // 4. Load all stored preferences for this user // 5. Publish notification.enriched to ENRICHED stream - // 5. Publish notification.enriched to ENRICHED stream - worker = new EnricherWorker({ consumer, pendingScanner, @@ -569,11 +715,14 @@ export async function startEnricherWorker() { maxRetriesBeforeDlq: 5, concurrency: config.WORKER_CONCURRENCY, producers, + outboundProducers, + scheduledProducer, idempotency, userRepo, prefRepo, contactRepo, templateCache, + db, }); // ─── Health check interval ────────────────────────────────────────────────── diff --git a/tests/api.test.ts b/tests/api.test.ts index 10ba74e..c5cdda5 100644 --- a/tests/api.test.ts +++ b/tests/api.test.ts @@ -508,6 +508,56 @@ describe("API Handlers", () => { expect(body.notificationIdsBase).toBeDefined(); }); + it("dispatches directly to outboundProducers when priority is critical and recipient is inline (Approach B transactional fast-path)", async () => { + const mockOutboundProducers = { + critical: { + publishBatch: vi.fn().mockResolvedValue({ messageIds: ["fast-msg-1"] }), + }, + }; + + const fastDeps: any = { + ...deps, + outboundProducers: mockOutboundProducers, + templateCache: { + getCachedTemplate: vi.fn().mockResolvedValue({ + id: "urgent-alert", + content: { subject: "Security Alert: {{code}}" }, + }), + }, + }; + const fastHandlers = createHandlers(fastDeps); + + const req = createMockReq({ + user: { + id: "usr_fast_1", + email: "security@example.com", + }, + template: "urgent-alert", + priority: "critical", + channels: ["email"], + data: { code: "123456" }, + }); + const res = createMockRes(); + + await fastHandlers.notify(req, res, { + projectId: "test_project_id", + params: {}, + query: new URLSearchParams(), + } as any); + + expect(res.statusCode).toBe(202); + // Bypasses standard inbound producers + expect(deps.producers.critical!.publishBatch).not.toHaveBeenCalled(); + // Directly hits outbound critical producer + expect(mockOutboundProducers.critical.publishBatch).toHaveBeenCalledTimes(1); + + const publishedArg = (mockOutboundProducers.critical.publishBatch as any).mock.calls[0][0][0]; + expect(publishedArg.type).toBe("notification.dispatched"); + expect(publishedArg.payload.destination).toBe("security@example.com"); + expect(publishedArg.payload.channel).toBe("email"); + expect(publishedArg.payload.renderedContent.content.subject).toBe("Security Alert: 123456"); + }); + describe("New CRUD Endpoints", () => { it("listWorkflows", async () => { const res = createMockRes(); diff --git a/tests/workers.test.ts b/tests/workers.test.ts index 1df6d9c..2761fef 100644 --- a/tests/workers.test.ts +++ b/tests/workers.test.ts @@ -800,6 +800,72 @@ describe("EnricherWorker", () => { expect(channels).toContain("slack"); expect(channels).toContain("webhook"); }); + + it("dispatches directly to outboundProducers when configured (Approach A pipeline collapse)", async () => { + const mockOutboundProducers = { + normal: { + publishBatch: vi.fn().mockResolvedValue({ messageIds: ["out-1"], eventIds: ["evt-out-1"] }), + }, + }; + + const fastWorker = new EnricherWorker({ + consumer: mockConsumer as any, + pendingScanner: {} as any, + logger: mockLogger as any, + maxRetriesBeforeDlq: 5, + producers: mockProducers, + outboundProducers: mockOutboundProducers, + idempotency: mockIdempotency, + userRepo: mockUserRepo, + prefRepo: mockPrefRepo, + contactRepo: { + findActiveByUserIds: vi + .fn() + .mockResolvedValue( + new Map([ + ["usr-1", [{ channel: "email", target: "alice@example.com", enabled: true }]], + ]), + ), + } as any, + templateCache: { + getCachedTemplate: vi.fn().mockResolvedValue({ + id: "welcome", + content: { subject: "Welcome {{name}}" }, + }), + } as any, + }); + + const msg = { + id: "msg-collapse-1", + event: { + id: "evt-collapse-1", + type: "notification.requested", + timestamp: new Date().toISOString(), + metadata: { traceId: "trace-collapse" }, + payload: { + projectId: "123e4567-e89b-12d3-a456-426614174000", + target: { type: "user", userId: "usr-1" }, + channels: ["email"], + templateId: "welcome", + data: { name: "Alice" }, + }, + }, + }; + + await fastWorker.process(msg as any); + + // Verify it bypassed intermediate mockProducers and published directly to mockOutboundProducers + expect(mockProducers.normal.publishBatch).not.toHaveBeenCalled(); + expect(mockOutboundProducers.normal.publishBatch).toHaveBeenCalledTimes(1); + + const publishedBatch = (mockOutboundProducers.normal.publishBatch as any).mock.calls[0][0]; + expect(publishedBatch).toHaveLength(1); + const dispatched = publishedBatch[0]; + expect(dispatched.type).toBe("notification.dispatched"); + expect(dispatched.payload.channel).toBe("email"); + expect(dispatched.payload.destination).toBe("alice@example.com"); + expect(dispatched.payload.renderedContent.content.subject).toBe("Welcome Alice"); + }); }); describe("DeliveryWorker", () => {