Improve Effect v4 quality across services and SDKs

apps/alerting/src/worker.ts

@@ -1,5 +1,5 @@
 import {
-	ANTICIPATED_ERROR_TAGS,
+	ANTICIPATED_ERROR_IDENTIFIERS,
 	AlertsService,
 	AnomalyDetectionService,
 	BucketCacheService,
@@ -38,7 +38,7 @@ const telemetry = MapleCloudflareSDK.make({
 	serviceName: "alerting",
 	serviceNamespace: "backend",
 	repositoryUrl: "https://github.com/Makisuo/maple",
-	anticipatedErrorTags: [...ANTICIPATED_ERROR_TAGS],
+	anticipatedErrorIdentifiers: [...ANTICIPATED_ERROR_IDENTIFIERS],
 })
 
 const buildLayer = (_env: Record<string, unknown>) => {

apps/api/src/alerting.ts

@@ -1,4 +1,4 @@
-export { ANTICIPATED_ERROR_TAGS } from "@maple/domain/anticipated-errors"
+export { ANTICIPATED_ERROR_IDENTIFIERS } from "@maple/domain/anticipated-errors"
 export { AlertRuntime, AlertsService } from "./services/AlertsService"
 export { AnomalyDetectionService } from "./services/AnomalyDetectionService"
 export { BucketCacheService } from "@maple/query-engine/caching"

apps/api/src/internal-rpc.test.ts

@@ -1,4 +1,4 @@
-import { describe, expect, it } from "vitest"
+import { describe, expect, it } from "@effect/vitest"
 import { Effect } from "effect"
 import type { InternalRpcInvalidInputError } from "@maple/domain/internal-rpc"
 import { callMcpToolRpc, submitDiagnosisRpc } from "./internal-rpc"
@@ -31,60 +31,60 @@ const unusedInvestigationService: InvestigationServiceShape = {
 }
 
 describe("internal RPC boundary", () => {
-	it("rejects invalid org IDs before MCP dispatch", async () => {
-		const error = await Effect.runPromise(
-			Effect.flip(
+	it.effect("rejects invalid org IDs before MCP dispatch", () =>
+		Effect.gen(function* () {
+			const error = yield* Effect.flip(
 				callMcpToolRpc({ orgId: " ", name: "inspect_trace", input: {} }) as Effect.Effect<
 					never,
 					InternalRpcInvalidInputError,
 					never
 				>,
-			),
-		)
-		expect(error._tag).toBe("@maple/internal-rpc/InvalidInputError")
-		expect(error.method).toBe("callMcpTool")
-	})
+			)
+			expect(error._tag).toBe("@maple/internal-rpc/InvalidInputError")
+			expect(error.method).toBe("callMcpTool")
+		}),
+	)
 
-	it("rejects invalid investigation IDs and model-produced reports", async () => {
-		for (const input of [
-			{ orgId: "org_1", investigationId: "not-a-uuid", report },
-			{ orgId: "org_1", investigationId, report: { summary: "incomplete" } },
-		]) {
-			const error = await Effect.runPromise(
-				Effect.flip(
+	it.effect("rejects invalid investigation IDs and model-produced reports", () =>
+		Effect.gen(function* () {
+			for (const input of [
+				{ orgId: "org_1", investigationId: "not-a-uuid", report },
+				{ orgId: "org_1", investigationId, report: { summary: "incomplete" } },
+			]) {
+				const error = yield* Effect.flip(
 					submitDiagnosisRpc(input).pipe(
 						Effect.provideService(InvestigationService, unusedInvestigationService),
 					),
-				),
-			)
-			expect(error._tag).toBe("@maple/internal-rpc/InvalidInputError")
-			if (error._tag !== "@maple/internal-rpc/InvalidInputError") {
-				throw new Error(`Expected invalid input, received ${error._tag}`)
+				)
+				expect(error._tag).toBe("@maple/internal-rpc/InvalidInputError")
+				if (error._tag !== "@maple/internal-rpc/InvalidInputError") {
+					throw new Error(`Expected invalid input, received ${error._tag}`)
+				}
+				expect(error.method).toBe("submitDiagnosis")
 			}
-			expect(error.method).toBe("submitDiagnosis")
-		}
-	})
+		}),
+	)
 
-	it("submits a decoded diagnosis to the org-scoped service", async () => {
-		const calls: Array<{ orgId: string; investigationId: string; summary: string }> = []
-		const expected = { id: investigationId, status: "diagnosed" } as never
-		const service: InvestigationServiceShape = {
-			...unusedInvestigationService,
-			submitDiagnosis: (orgId, id, request) =>
-				Effect.sync(() => {
-					calls.push({ orgId, investigationId: id, summary: request.report.summary })
-					return expected
-				}),
-		}
+	it.effect("submits a decoded diagnosis to the org-scoped service", () =>
+		Effect.gen(function* () {
+			const calls: Array<{ orgId: string; investigationId: string; summary: string }> = []
+			const expected = { id: investigationId, status: "diagnosed" } as never
+			const service: InvestigationServiceShape = {
+				...unusedInvestigationService,
+				submitDiagnosis: (orgId, id, request) =>
+					Effect.sync(() => {
+						calls.push({ orgId, investigationId: id, summary: request.report.summary })
+						return expected
+					}),
+			}
 
-		const result = await Effect.runPromise(
-			submitDiagnosisRpc({ orgId: "org_1", investigationId, report }).pipe(
+			const result = yield* submitDiagnosisRpc({ orgId: "org_1", investigationId, report }).pipe(
 				Effect.provideService(InvestigationService, service),
-			),
-		)
-		expect(result).toBe(expected)
-		expect(calls).toEqual([
-			{ orgId: "org_1", investigationId, summary: "Checkout latency doubled after deploy." },
-		])
-	})
+			)
+			expect(result).toBe(expected)
+			expect(calls).toEqual([
+				{ orgId: "org_1", investigationId, summary: "Checkout latency doubled after deploy." },
+			])
+		}),
+	)
 })

apps/api/src/mcp/dispatcher.test.ts

@@ -1,48 +1,50 @@
-import { describe, expect, it } from "vitest"
+import { describe, expect, it } from "@effect/vitest"
 import { Effect } from "effect"
 import type { InternalRpcToolNotFoundError } from "@maple/domain/internal-rpc"
 import { callMcpTool, listMcpTools } from "./dispatcher"
 import { mapleToolDefinitions, toInputSchema } from "./tools/registry"
 
 describe("MCP dispatcher", () => {
-	it("publishes the same names, descriptions, and schemas used by HTTP MCP", async () => {
-		const descriptors = await Effect.runPromise(listMcpTools)
-		expect(descriptors).toEqual(
-			mapleToolDefinitions.map((definition) => ({
-				name: definition.name,
-				description: definition.description,
-				inputSchema: toInputSchema(definition.schema),
-			})),
-		)
-	})
+	it.effect("publishes the same names, descriptions, and schemas used by HTTP MCP", () =>
+		Effect.gen(function* () {
+			const descriptors = yield* listMcpTools
+			expect(descriptors).toEqual(
+				mapleToolDefinitions.map((definition) => ({
+					name: definition.name,
+					description: definition.description,
+					inputSchema: toInputSchema(definition.schema),
+				})),
+			)
+		}),
+	)
 
-	it("returns MCP validation feedback for invalid model tool input", async () => {
-		const result = await Effect.runPromise(
-			callMcpTool("inspect_trace", {}) as unknown as Effect.Effect<
+	it.effect("returns MCP validation feedback for invalid model tool input", () =>
+		Effect.gen(function* () {
+			const result = yield* callMcpTool("inspect_trace", {}) as unknown as Effect.Effect<
 				{
 					readonly content: ReadonlyArray<{ readonly type: "text"; readonly text: string }>
 					readonly isError?: boolean
 				},
 				never,
 				never
-			>,
-		)
-		expect(result.isError).toBe(true)
-		expect(result.content[0]?.text).toContain("Invalid parameters")
-		expect(result.content[0]?.text).toContain("inspect_trace")
-	})
+			>
+			expect(result.isError).toBe(true)
+			expect(result.content[0]?.text).toContain("Invalid parameters")
+			expect(result.content[0]?.text).toContain("inspect_trace")
+		}),
+	)
 
-	it("fails unknown RPC tool names with a typed error", async () => {
-		const error = await Effect.runPromise(
-			Effect.flip(
+	it.effect("fails unknown RPC tool names with a typed error", () =>
+		Effect.gen(function* () {
+			const error = yield* Effect.flip(
 				callMcpTool("not_a_maple_tool", {}) as Effect.Effect<
 					never,
 					InternalRpcToolNotFoundError,
 					never
 				>,
-			),
-		)
-		expect(error._tag).toBe("@maple/internal-rpc/ToolNotFoundError")
-		expect(error.name).toBe("not_a_maple_tool")
-	})
+			)
+			expect(error._tag).toBe("@maple/internal-rpc/ToolNotFoundError")
+			expect(error.name).toBe("not_a_maple_tool")
+		}),
+	)
 })

apps/api/src/routes/scraper-internal.http.ts

@@ -117,7 +117,7 @@ export const ScraperInternalRouter = HttpRouter.use((router) =>
 				const denied = unauthorized(req)
 				if (denied) return denied
 
-				const rows = yield* service.listAllEnabled().pipe(Effect.catch(() => Effect.succeed([])))
+				const rows = yield* service.listAllEnabled()
 
 				// One public ingest key per org (lazily created on first use, like
 				// onboarding does). The scraper ingests with this key so scraped
@@ -128,12 +128,9 @@ export const ScraperInternalRouter = HttpRouter.use((router) =>
 					Effect.gen(function* () {
 						const cached = keyByOrg.get(orgId)
 						if (cached !== undefined) return cached
-						const key: string | null = yield* ingestKeys
+						const key: string = yield* ingestKeys
 							.getOrCreate(decodeOrgIdSync(orgId), SCRAPER_SYSTEM_USER)
-							.pipe(
-								Effect.map((keys): string | null => keys.publicKey),
-								Effect.catch(() => Effect.succeed(null)),
-							)
+							.pipe(Effect.map((keys) => keys.publicKey))
 						keyByOrg.set(orgId, key)
 						return key
 					})
@@ -171,7 +168,9 @@ export const ScraperInternalRouter = HttpRouter.use((router) =>
 									if (Option.isSome(target)) {
 										targets.push(target.value)
 									} else {
-										yield* Effect.logWarning("Skipping scrape sub-target (invalid row)").pipe(
+										yield* Effect.logWarning(
+											"Skipping scrape sub-target (invalid row)",
+										).pipe(
 											Effect.annotateLogs({
 												scrapeTargetId: row.id,
 												orgId: row.orgId,
@@ -196,7 +195,15 @@ export const ScraperInternalRouter = HttpRouter.use((router) =>
 				)
 
 				return yield* HttpServerResponse.json(targets)
-			}).pipe(Effect.withSpan("ScraperInternal.listTargets"))
+			}).pipe(
+				Effect.catch((error) =>
+					Effect.logError("Failed to build scraper target list").pipe(
+						Effect.annotateLogs({ error: error.message }),
+						Effect.as(errorText("Scraper target list unavailable", 503)),
+					),
+				),
+				Effect.withSpan("ScraperInternal.listTargets"),
+			)
 
 		const recordResults = (req: HttpServerRequest.HttpServerRequest) =>
 			Effect.gen(function* () {
@@ -209,18 +216,18 @@ export const ScraperInternalRouter = HttpRouter.use((router) =>
 				const results = yield* decodeScrapeResultsEffect(body.value).pipe(Effect.option)
 				if (Option.isNone(results)) return errorText("Invalid scrape results payload", 400)
 
-				yield* service
-					.recordScrapeResults(results.value)
-					.pipe(
-						Effect.catch((error) =>
-							Effect.logWarning("Failed to persist scrape results").pipe(
-								Effect.annotateLogs({ error: error.message }),
-							),
-						),
-					)
+				yield* service.recordScrapeResults(results.value)
 
 				return yield* HttpServerResponse.json({ recorded: results.value.length })
-			}).pipe(Effect.withSpan("ScraperInternal.recordResults"))
+			}).pipe(
+				Effect.catch((error) =>
+					Effect.logError("Failed to persist scrape results").pipe(
+						Effect.annotateLogs({ error: error.message }),
+						Effect.as(errorText("Scrape result persistence unavailable", 503)),
+					),
+				),
+				Effect.withSpan("ScraperInternal.recordResults"),
+			)
 
 		yield* router.add("GET", "/api/internal/scrape-targets", listTargets)
 		yield* router.add("POST", "/api/internal/scrape-results", recordResults)

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

@@ -1,7 +1,7 @@
 import type { AlertDestinationRow } from "@maple/db"
-import { AlertDeliveryError } from "@maple/domain/http"
+import { AlertDeliveryError, AlertDestinationId } from "@maple/domain/http"
 import { assert, describe, it } from "@effect/vitest"
-import { Effect } from "effect"
+import { Effect, Schema } from "effect"
 import {
 	buildAlertChatUrl,
 	buildDiscordEmbedsFromTemplate,
@@ -37,6 +37,7 @@ const baseContext: TemplateRenderContext = {
 
 const LINK = "https://web.localhost/alerts"
 const CHAT = "https://web.localhost/chat?mode=alert"
+const DESTINATION_ID = Schema.decodeUnknownSync(AlertDestinationId)("7c6b5a49-3821-4e0f-9d8c-7b6a59483726")
 
 /** Dispatch deps for non-email destinations — email sends must not happen. */
 const noEmailDeps: DispatchDeps = {
@@ -138,7 +139,7 @@ describe("buildDiscordEmbedsFromTemplate", () => {
 
 describe("dispatchDelivery", () => {
 	const destinationRow: AlertDestinationRow = {
-		id: "dest_1" as AlertDestinationRow["id"],
+		id: DESTINATION_ID,
 		orgId: "org_1" as AlertDestinationRow["orgId"],
 		name: "PagerDuty",
 		type: "pagerduty",

apps/api/src/services/AlertsService.ts

@@ -27,7 +27,6 @@ import {
 	AlertCheckDocument,
 	AlertChecksListResponse,
 	AlertCheckStatus as AlertCheckStatusSchema,
-	AlertEvaluationStatus as AlertEvaluationStatusSchema,
 	AlertIncidentDocument,
 	AlertIncidentsListResponse,
 	AlertIncidentStatus,
@@ -243,6 +242,8 @@ interface DeliveryAttemptFailure {
 }
 
 const MAX_DELIVERY_ATTEMPTS = 5
+const ALERT_TEST_DELIVERY_CONCURRENCY = 5
+const ALERT_CHECK_INGEST_CONCURRENCY = 4
 // Storm fuse: cap issue-hub upserts per scheduler tick so a pathological
 // group-by rule opening hundreds of incidents can't stall the per-minute tick.
 const ISSUE_UPSERTS_PER_TICK = 50
@@ -300,7 +301,6 @@ const decodeAlertDestinationTypeSync = Schema.decodeUnknownSync(AlertDestination
 const decodeAlertSeveritySync = Schema.decodeUnknownSync(AlertSeveritySchema)
 const decodeAlertSignalTypeSync = Schema.decodeUnknownSync(AlertSignalTypeSchema)
 const decodeAlertComparatorSync = Schema.decodeUnknownSync(AlertComparatorSchema)
-const decodeAlertEvaluationStatusSync = Schema.decodeUnknownSync(AlertEvaluationStatusSchema)
 const decodeAlertCheckStatusSync = Schema.decodeUnknownSync(AlertCheckStatusSchema)
 const decodeAlertIncidentTransitionSync = Schema.decodeUnknownSync(AlertIncidentTransitionSchema)
 const decodeAlertMetricTypeSync = Schema.decodeUnknownSync(AlertMetricTypeSchema)
@@ -2826,7 +2826,7 @@ export class AlertsService extends Context.Service<AlertsService, AlertsServiceS
 								linkUrl: resolveNotificationLinkUrl(normalized, null),
 								sentAtMs,
 							}),
-						{ concurrency: "unbounded" },
+						{ concurrency: ALERT_TEST_DELIVERY_CONCURRENCY },
 					)
 				}
 
@@ -3808,7 +3808,9 @@ export class AlertsService extends Context.Service<AlertsService, AlertsServiceS
 											isNotNull(alertIncidents.lastNotifiedAt),
 											gte(
 												alertIncidents.lastNotifiedAt,
-												new Date(timestamp - normalized.renotifyIntervalMinutes * 60_000),
+												new Date(
+													timestamp - normalized.renotifyIntervalMinutes * 60_000,
+												),
 											),
 										),
 									)
@@ -4748,7 +4750,7 @@ export class AlertsService extends Context.Service<AlertsService, AlertsServiceS
 									),
 									Effect.ignore,
 								),
-						{ concurrency: "unbounded" },
+						{ concurrency: ALERT_CHECK_INGEST_CONCURRENCY },
 					)
 				}
 

apps/api/src/services/NotificationDispatcher.ts

@@ -32,6 +32,7 @@ import { Env } from "../lib/Env"
  */
 
 const DELIVERY_TIMEOUT_MS = 15_000
+const NOTIFICATION_DELIVERY_CONCURRENCY = 5
 
 class NotificationDispatchError extends Data.TaggedError("@maple/api/services/NotificationDispatchError")<{
 	readonly message: string
@@ -249,7 +250,7 @@ const make: Effect.Effect<
 							"@maple/http/errors/AlertDeliveryError": () => Effect.succeed("failed" as const),
 						}),
 					),
-				{ concurrency: "unbounded" },
+				{ concurrency: NOTIFICATION_DELIVERY_CONCURRENCY },
 			)
 
 			return {