fix: stop duplicate notification emails

apps/alerting/alchemy.run.ts

@@ -64,9 +64,16 @@ export const createAlertingWorker = ({ stage, mapleDb }: CreateAlertingWorkerOpt
 				// Ref stages attach MAPLE_DB via worker.bind below.
 				...(mapleDb ? { MAPLE_DB: mapleDb } : {}),
 				AI_TRIAGE_WORKFLOW: aiTriageWorkflow,
-				EMAIL: Cloudflare.Email.SendEmail("email", {
-					allowedSenderAddresses: ["notifications@noreply.maple.dev"],
-				}),
+				// Production only: preview/stg workers run the same email crons against
+				// their own DB branches, so a binding here means every live stage sends
+				// its own copy of onboarding/digest/alert emails to real users.
+				...(stage.kind === "prd"
+					? {
+							EMAIL: Cloudflare.Email.SendEmail("email", {
+								allowedSenderAddresses: ["notifications@noreply.maple.dev"],
+							}),
+						}
+					: {}),
 				TINYBIRD_HOST: requireEnv("TINYBIRD_HOST"),
 				TINYBIRD_TOKEN: Redacted.make(requireEnv("TINYBIRD_TOKEN")),
 				// Alert-rule evaluation runs Tinybird-scoped raw SQL through

apps/api/alchemy.run.ts

@@ -143,9 +143,16 @@ export const createMapleApi = ({ stage, domains }: CreateMapleApiOptions) =>
 					simple: { limit: 600, period: 60 },
 				}),
 				API_V2_RATE_LIMIT_PARTITION: formatMapleStage(stage),
-				EMAIL: Cloudflare.Email.SendEmail("email", {
-					allowedSenderAddresses: ["notifications@noreply.maple.dev"],
-				}),
+				// Production only: preview/stg workers run the same email crons against
+				// their own DB branches, so a binding here means every live stage sends
+				// its own copy of onboarding/digest/alert emails to real users.
+				...(stage.kind === "prd"
+					? {
+							EMAIL: Cloudflare.Email.SendEmail("email", {
+								allowedSenderAddresses: ["notifications@noreply.maple.dev"],
+							}),
+						}
+					: {}),
 				TINYBIRD_HOST: requireEnv("TINYBIRD_HOST"),
 				TINYBIRD_TOKEN: Redacted.make(requireEnv("TINYBIRD_TOKEN")),
 				...optionalSecret("TINYBIRD_SIGNING_KEY"),

apps/api/src/lib/EmailService.ts

@@ -47,7 +47,14 @@ export class EmailService extends Context.Service<EmailService, EmailServiceShap
 			const workerEnv = yield* WorkerEnvironment
 			const binding = (workerEnv as Record<string, SendEmailBinding | undefined>).EMAIL
 
-			const isConfigured = binding !== undefined
+			// Real sends are production-only: preview/stg stages share real user data
+			// (branched DBs, Clerk members), so a live binding there would deliver
+			// duplicate copies of every cron-driven email. The alchemy configs no
+			// longer attach EMAIL outside prd; this guard covers any binding that
+			// still reaches a non-prod worker (wrangler dev, manual deploys).
+			const emailAllowed =
+				env.MAPLE_ENVIRONMENT === "production" || env.MAPLE_EMAIL_ALLOW_NONPROD === "true"
+			const isConfigured = binding !== undefined && emailAllowed
 
 			const send = Effect.fn("EmailService.send")(function* (
 				to: string,
@@ -67,6 +74,14 @@ export class EmailService extends Context.Service<EmailService, EmailServiceShap
 					)
 				}
 
+				if (!emailAllowed) {
+					return yield* Effect.fail(
+						new EmailDeliveryError({
+							message: `Email suppressed: sends are disabled in ${env.MAPLE_ENVIRONMENT} (set MAPLE_EMAIL_ALLOW_NONPROD=true to override)`,
+						}),
+					)
+				}
+
 				const result = yield* Effect.tryPromise({
 					try: () =>
 						binding.send({

apps/api/src/lib/Env.ts

@@ -26,6 +26,10 @@ export interface EnvShape {
 	readonly MAPLE_INGEST_KEY_LOOKUP_HMAC_KEY: Redacted.Redacted<string>
 	readonly MAPLE_INGEST_PUBLIC_URL: string
 	readonly MAPLE_APP_BASE_URL: string
+	/** Deployment environment (`production`, `staging`, `pr-<n>`, `development`) — set by alchemy from the stage. */
+	readonly MAPLE_ENVIRONMENT: string
+	/** Escape hatch: allow real email sends outside production (e.g. a dedicated stg test run). */
+	readonly MAPLE_EMAIL_ALLOW_NONPROD: string
 	readonly CLERK_SECRET_KEY: Option.Option<Redacted.Redacted<string>>
 	readonly CLERK_PUBLISHABLE_KEY: Option.Option<string>
 	readonly CLERK_JWT_KEY: Option.Option<Redacted.Redacted<string>>
@@ -100,6 +104,8 @@ const envConfig = Config.all({
 	MAPLE_INGEST_KEY_LOOKUP_HMAC_KEY: Config.redacted("MAPLE_INGEST_KEY_LOOKUP_HMAC_KEY"),
 	MAPLE_INGEST_PUBLIC_URL: stringWithDefault("MAPLE_INGEST_PUBLIC_URL", "http://127.0.0.1:3474"),
 	MAPLE_APP_BASE_URL: stringWithDefault("MAPLE_APP_BASE_URL", "http://127.0.0.1:3471"),
+	MAPLE_ENVIRONMENT: stringWithDefault("MAPLE_ENVIRONMENT", "development"),
+	MAPLE_EMAIL_ALLOW_NONPROD: stringWithDefault("MAPLE_EMAIL_ALLOW_NONPROD", "false"),
 	CLERK_SECRET_KEY: optionalRedacted("CLERK_SECRET_KEY"),
 	CLERK_PUBLISHABLE_KEY: optionalString("CLERK_PUBLISHABLE_KEY"),
 	CLERK_JWT_KEY: optionalRedacted("CLERK_JWT_KEY"),

apps/api/src/services/AlertDeliveryDispatch.test.ts

@@ -260,4 +260,54 @@ describe("dispatchDelivery", () => {
 			assert.include(error.message, "EMAIL binding is missing")
 		}),
 	)
+
+	it.effect("email: succeeds with annotation when only some members fail", () =>
+		Effect.gen(function* () {
+			const sent: string[] = []
+			const deps: DispatchDeps = {
+				sendEmail: (to) =>
+					to === "oncall@acme.test"
+						? Effect.fail(
+								new AlertDeliveryError({
+									message: "mailbox unavailable",
+									destinationType: "email",
+								}),
+							)
+						: Effect.sync(() => {
+								sent.push(to)
+							}),
+			}
+
+			const result = yield* dispatchDelivery(emailContext, "{}", failingFetch, 5_000, LINK, CHAT, deps)
+
+			assert.deepStrictEqual(sent, ["ops@acme.test"])
+			assert.include(result.providerMessage, "Emailed 1 of 2 members")
+			assert.include(result.providerMessage, "oncall@acme.test")
+			assert.include(result.providerMessage, "mailbox unavailable")
+		}),
+	)
+
+	it.effect("email: fails with the first member error when every member fails", () =>
+		Effect.gen(function* () {
+			const deps: DispatchDeps = {
+				sendEmail: () =>
+					Effect.fail(
+						new AlertDeliveryError({
+							message: "Cloudflare Email send timed out after 15s",
+							destinationType: "email",
+						}),
+					),
+			}
+
+			const error = yield* Effect.flip(
+				dispatchDelivery(emailContext, "{}", failingFetch, 5_000, LINK, CHAT, deps),
+			)
+
+			assert.instanceOf(error, AlertDeliveryError)
+			assert.include(error.message, "failed for all 2 members")
+			// The verbatim member error must survive aggregation so retryability
+			// classification (timeout detection) keeps working upstream.
+			assert.include(error.message, "timed out")
+		}),
+	)
 })

apps/api/src/services/AlertDeliveryDispatch.ts

@@ -881,12 +881,49 @@ export const dispatchDelivery = (
 				email: (config) =>
 					Effect.gen(function* () {
 						const { subject, html } = yield* buildAlertEmailContent(context, linkUrl, chatUrl)
-						yield* Effect.forEach(
-							config.members,
-							(member) => deps.sendEmail(member.email, subject, html),
-							{ discard: true },
+						const outcomes = yield* Effect.forEach(config.members, (member) =>
+							deps.sendEmail(member.email, subject, html).pipe(
+								Effect.match({
+									onSuccess: () => ({ member, error: null as string | null }),
+									onFailure: (error) => ({ member, error: error.message }),
+								}),
+							),
 						)
+						const failures = outcomes.filter((o) => o.error != null)
 						const count = config.members.length
+
+						if (failures.length === count && count > 0) {
+							// Nobody received the email — safe to fail retryable; the retry
+							// re-sends to members who all got nothing. Keep the first failure
+							// message verbatim so timeout classification survives aggregation.
+							return yield* Effect.fail(
+								makeDeliveryError(
+									`Email delivery failed for all ${count} member${count === 1 ? "" : "s"}: ${failures[0]!.error}`,
+									"email",
+								),
+							)
+						}
+
+						if (failures.length > 0) {
+							// Partial success is terminal: there is no per-member attempt
+							// state, so retrying the event would re-email the members who
+							// already received it. Report success and surface the failures.
+							yield* Effect.logWarning("Alert email delivered to a subset of members").pipe(
+								Effect.annotateLogs({
+									failedCount: failures.length,
+									memberCount: count,
+									firstError: failures[0]!.error,
+								}),
+							)
+							return {
+								providerMessage: `Emailed ${count - failures.length} of ${count} members; failed: ${failures
+									.map((f) => `${f.member.email} (${f.error})`)
+									.join(", ")}`,
+								providerReference: null,
+								responseCode: null,
+							} as DispatchResult
+						}
+
 						return {
 							providerMessage: `Emailed ${count} member${count === 1 ? "" : "s"}`,
 							providerReference: null,

apps/api/src/services/AlertsService.test.ts

@@ -706,6 +706,161 @@ describe("AlertsService", () => {
 		}).pipe(Effect.provide(makeLayer(testDb, makeWarehouseStub(state), { fetch: okFetch })))
 	})
 
+	it.effect("suppresses trigger and resolve notifications while an incident flaps", () => {
+		const testDb = createTestDb(trackedDbs)
+		const breachedRows = [
+			{
+				count: 200,
+				avgDuration: 40,
+				p50Duration: 20,
+				p95Duration: 120,
+				p99Duration: 240,
+				errorRate: 10,
+				satisfiedCount: 180,
+				toleratingCount: 10,
+				apdexScore: 0.925,
+			},
+		] as ReadonlyArray<Record<string, unknown>>
+		const healthyRows = [
+			{
+				count: 200,
+				avgDuration: 20,
+				p50Duration: 10,
+				p95Duration: 80,
+				p99Duration: 160,
+				errorRate: 0.5,
+				satisfiedCount: 195,
+				toleratingCount: 3,
+				apdexScore: 0.9825,
+			},
+		] as ReadonlyArray<Record<string, unknown>>
+		const state = { tracesAggregateRows: breachedRows }
+
+		return Effect.gen(function* () {
+			yield* TestClock.setTime(DEFAULT_CLOCK_EPOCH_MS)
+			const alerts = yield* AlertsService
+			const orgId = asOrgId("org_flap")
+			const userId = asUserId("user_flap")
+			const destination = yield* createWebhookDestination(alerts, orgId, userId)
+			yield* createErrorRateRule(alerts, orgId, userId, destination.id)
+
+			const tick = Effect.gen(function* () {
+				yield* alerts.runSchedulerTick()
+				yield* TestClock.adjust(Duration.minutes(1))
+			})
+
+			// Flap 1: open (trigger delivered) then resolve (resolve delivered).
+			yield* tick
+			yield* tick
+			state.tracesAggregateRows = healthyRows
+			yield* tick
+			yield* tick
+
+			// Flap 2 within the renotify interval: incident opens again, but both
+			// its trigger and its resolve notifications are suppressed.
+			state.tracesAggregateRows = breachedRows
+			yield* tick
+			yield* tick
+			state.tracesAggregateRows = healthyRows
+			yield* tick
+			yield* tick
+
+			const incidents = yield* alerts.listIncidents(orgId)
+			const events = yield* alerts.listDeliveryEvents(orgId)
+
+			assert.lengthOf(incidents.incidents, 2)
+			assert.isTrue(incidents.incidents.every((incident) => incident.status === "resolved"))
+			// Only the first flap emailed: one trigger + one resolve, nothing for flap 2.
+			assert.deepStrictEqual(
+				events.events.map((event: { eventType: string }) => event.eventType).sort(),
+				["resolve", "trigger"],
+			)
+		}).pipe(Effect.provide(makeLayer(testDb, makeWarehouseStub(state), { fetch: okFetch })))
+	})
+
+	it.effect("queues at most one renotify per interval while deliveries keep failing", () => {
+		const testDb = createTestDb(trackedDbs)
+		const state = {
+			tracesAggregateRows: [
+				{
+					count: 200,
+					avgDuration: 40,
+					p50Duration: 20,
+					p95Duration: 120,
+					p99Duration: 240,
+					errorRate: 10,
+					satisfiedCount: 180,
+					toleratingCount: 10,
+					apdexScore: 0.925,
+				},
+			] as ReadonlyArray<Record<string, unknown>>,
+		}
+		const failingFetch: typeof fetch = (async () =>
+			new Response("boom", { status: 500 })) as unknown as typeof fetch
+
+		return Effect.gen(function* () {
+			yield* TestClock.setTime(DEFAULT_CLOCK_EPOCH_MS)
+			const alerts = yield* AlertsService
+			const orgId = asOrgId("org_renotify_gate")
+			const userId = asUserId("user_renotify_gate")
+			const destination = yield* createWebhookDestination(alerts, orgId, userId)
+			yield* alerts.createRule(
+				orgId,
+				userId,
+				adminRoles,
+				new AlertRuleUpsertRequest({
+					name: "Checkout error rate",
+					severity: "critical",
+					enabled: true,
+					serviceNames: ["checkout"],
+					signalType: "error_rate",
+					comparator: "gt",
+					threshold: 5,
+					windowMinutes: 5,
+					minimumSampleCount: 10,
+					consecutiveBreachesRequired: 2,
+					consecutiveHealthyRequired: 2,
+					renotifyIntervalMinutes: 5,
+					destinationIds: [destination.id],
+				}),
+			)
+
+			// Trigger opens on the second tick; every delivery attempt fails with a
+			// retryable 500 for the whole test, so lastNotifiedAt can only advance
+			// via the queue-time gate.
+			for (let minute = 0; minute < 9; minute += 1) {
+				yield* alerts.runSchedulerTick()
+				yield* TestClock.adjust(Duration.minutes(1))
+			}
+
+			const events = yield* alerts.listDeliveryEvents(orgId)
+			const renotifyFirstAttempts = events.events.filter(
+				(event: { eventType: string; attemptNumber: number }) =>
+					event.eventType === "renotify" && event.attemptNumber === 1,
+			)
+			// Trigger at t=1min → renotify due at t=6min. Ticks at 7/8 min must NOT
+			// re-queue a fresh renotify chain even though nothing was delivered.
+			assert.lengthOf(renotifyFirstAttempts, 1)
+
+			const incidents = yield* alerts.listIncidents(orgId)
+			assert.lengthOf(incidents.incidents, 1)
+			assert.isNotNull(incidents.incidents[0]?.lastNotifiedAt)
+
+			// One interval after the queue-time advance, a second renotify chain starts.
+			for (let minute = 0; minute < 3; minute += 1) {
+				yield* alerts.runSchedulerTick()
+				yield* TestClock.adjust(Duration.minutes(1))
+			}
+			const eventsAfter = yield* alerts.listDeliveryEvents(orgId)
+			const renotifyChains = new Set(
+				eventsAfter.events
+					.filter((event: { eventType: string }) => event.eventType === "renotify")
+					.map((event: { deliveryKey: string }) => event.deliveryKey.split(":").at(-1)),
+			)
+			assert.strictEqual(renotifyChains.size, 2)
+		}).pipe(Effect.provide(makeLayer(testDb, makeWarehouseStub(state), { fetch: failingFetch })))
+	})
+
 	it.effect("skips unchanged alert_rule_states writes and refreshes on the 5-minute heartbeat", () => {
 		const testDb = createTestDb(trackedDbs)
 		// Healthy from the start: errorRate 1 stays below the threshold of 5.