perf(backend): batch digest and push writes, read in parallel, share parsers, delete dead code

packages/backend/src/platform/Crypto.ts

@@ -22,6 +22,18 @@ export const parseBase64Aes256GcmKey = <E>(raw: string, onError: (message: strin
 		return Effect.succeed(decoded)
 	})
 
+/** The same parse for `MAPLE_INGEST_KEY_ENCRYPTION_KEY`, with messages that name the variable. */
+export const parseIngestKeyEncryptionKey = <E>(raw: string, onError: (message: string) => E) =>
+	parseBase64Aes256GcmKey(raw, (message) =>
+		onError(
+			message === "Expected a non-empty base64 encryption key"
+				? "MAPLE_INGEST_KEY_ENCRYPTION_KEY is required"
+				: message === "Expected base64 for exactly 32 bytes"
+					? "MAPLE_INGEST_KEY_ENCRYPTION_KEY must be base64 for exactly 32 bytes"
+					: message,
+		),
+	)
+
 /**
  * Optional additional authenticated data. AAD is authenticated but not stored,
  * so it binds a ciphertext to the row it was written for: decryption only

packages/backend/src/platform/invocation-context.test.ts

@@ -1,6 +1,6 @@
 import { describe, expect, it } from "@effect/vitest"
 import { Context, Effect, Scope } from "effect"
-import { captureReusableContext } from "./invocation-context"
+import { withoutInvocationContext } from "./invocation-context"
 import { DrainRunOnInterrupt, PgConnectionScope, type PgConnectionScopeApi } from "./pg-connection-scope"
 
 class Greeting extends Context.Service<Greeting, { readonly text: string }>()("@maple/test/Greeting") {}
@@ -11,14 +11,15 @@ const scopeNamed = (name: string): PgConnectionScopeApi & { readonly name: strin
 	close: Effect.void,
 })
 
-describe("captureReusableContext", () => {
+describe("withoutInvocationContext", () => {
 	it.effect("keeps services but lets the later invocation's own scope, drain flag and Scope win", () =>
 		Effect.gen(function* () {
 			const constructionScope = scopeNamed("construction")
 			const requestScope = scopeNamed("request")
 			const constructionResource = yield* Scope.make()
 
-			const captured = yield* captureReusableContext<Greeting>().pipe(
+			const captured = yield* Effect.context<Greeting>().pipe(
+				Effect.map(withoutInvocationContext),
 				Effect.provideService(Greeting, { text: "hello" }),
 				Effect.provideService(PgConnectionScope, constructionScope),
 				Effect.provideService(DrainRunOnInterrupt, true),

packages/backend/src/platform/invocation-context.ts

@@ -1,4 +1,4 @@
-import { Context, Effect, Scope, Tracer } from "effect"
+import { Context, Scope, Tracer } from "effect"
 import { HttpServerRequest } from "effect/http"
 import { DrainRunOnInterrupt, PgConnectionScope } from "./pg-connection-scope"
 
@@ -17,6 +17,3 @@ export const withoutInvocationContext = <R>(context: Context.Context<R>) =>
 		Tracer.ParentSpan,
 		HttpServerRequest.HttpServerRequest,
 	)(context)
-
-/** `Effect.context`, safe to keep and provide on later invocations. */
-export const captureReusableContext = <R>() => Effect.map(Effect.context<R>(), withoutInvocationContext)

packages/backend/src/services/alerts/AnomalyDetectionService.ts

@@ -1547,11 +1547,9 @@ const make: Effect.Effect<
 		let openBudget = MAX_OPENS_PER_TICK
 
 		// Detector-state writes are accumulated here and flushed as one multi-row
-		// upsert after the loop, rather than one `dbExecute` per decision. Under
-		// `DatabasePgLive` each execute dials and tears down its own postgres.js
-		// client, so the handshake count is what costs, not the statement count —
-		// same trade as `error-tick-persistence.ts`. This loop averaged ~70 dials
-		// per org tick and peaked at 628.
+		// upsert after the loop, rather than one `dbExecute` per decision: that was
+		// a round trip per row, ~70 per org tick on average and 628 at peak. The
+		// invocation's pool is reused across executes, so it saves round trips, not dials.
 		//
 		// Keyed by `detectorKey` rather than a plain list: the consolidated-incident
 		// refcount below has to see this tick's own buffered writes, and last-wins

packages/backend/src/services/alerts/OnCallSchedulesService.ts

@@ -309,24 +309,28 @@ export class OnCallSchedulesService extends Context.Service<
 			const nowMs = yield* runtime.now
 			const [from, to] = nowWindow(nowMs)
 			const [rows, nearby, active, chains] = yield* dbExecute((db) =>
-				Effect.all([
-					db.run(
-						PG.from(OnCallSchedules)
-							.select()
-							.where(($) => [$.orgId.eq(orgId)])
-							.orderBy(["name", "asc"])
-							.limit(MAX_ONCALL_SCHEDULES_PER_ORG),
-					),
-					overlappingOverrides(db, orgId, from, to),
-					// The org-wide window above is capped, so the overrides that decide who is on
-					// call now are read on their own: at most the upcoming cap per schedule.
-					db.run(
-						PG.from(OnCallOverrides)
-							.select()
-							.where(($) => [$.orgId.eq(orgId), $.startsAt.lte(nowMs), $.endsAt.gt(nowMs)]),
-					),
-					chainIdsBySchedule(db, orgId),
-				]),
+				Effect.all(
+					[
+						db.run(
+							PG.from(OnCallSchedules)
+								.select()
+								.where(($) => [$.orgId.eq(orgId)])
+								.orderBy(["name", "asc"])
+								.limit(MAX_ONCALL_SCHEDULES_PER_ORG),
+						),
+						overlappingOverrides(db, orgId, from, to),
+						// The org-wide window above is capped, so the overrides that decide who is on
+						// call now are read on their own: at most the upcoming cap per schedule.
+						db.run(
+							PG.from(OnCallOverrides)
+								.select()
+								.where(($) => [$.orgId.eq(orgId), $.startsAt.lte(nowMs), $.endsAt.gt(nowMs)]),
+						),
+						chainIdsBySchedule(db, orgId),
+					],
+					// Four independent reads, so they share the pool instead of queueing.
+					{ concurrency: "unbounded" },
+				),
 			)
 			const overrides = Arr.dedupeWith([...active, ...nearby], (a, b) => a.id === b.id)
 			yield* Effect.annotateCurrentSpan("result.rowCount", rows.length)

packages/backend/src/services/alerts/alert-escalation-runs.ts

@@ -328,27 +328,31 @@ export const loadStepRecipients = (
 		const scheduleIds = Arr.dedupe(
 			Arr.flatMap(step.targets, (target) => (target.type === "schedule" ? [target.scheduleId] : [])),
 		)
-		const [schedules, overrides] = yield* Effect.all([
-			scheduleIds.length === 0
-				? Effect.succeed([])
-				: db.run(
-						PG.from(OnCallSchedules)
-							.select("id", "timeZone", "layersJson")
-							.where(($) => [$.orgId.eq(orgId), $.id.in_(...scheduleIds)]),
-					),
-			scheduleIds.length === 0
-				? Effect.succeed([])
-				: db.run(
-						PG.from(OnCallOverrides)
-							.select("id", "scheduleId", "userId", "startsAt", "endsAt", "createdAt")
-							.where(($) => [
-								$.orgId.eq(orgId),
-								$.scheduleId.in_(...scheduleIds),
-								$.endsAt.gt(nowMs - ONCALL_LOOKAROUND_MS),
-								$.startsAt.lt(nowMs + ONCALL_LOOKAROUND_MS),
-							]),
-					),
-		])
+		// Both pairs below are independent reads: they share the pool instead of queueing.
+		const [schedules, overrides] = yield* Effect.all(
+			[
+				scheduleIds.length === 0
+					? Effect.succeed([])
+					: db.run(
+							PG.from(OnCallSchedules)
+								.select("id", "timeZone", "layersJson")
+								.where(($) => [$.orgId.eq(orgId), $.id.in_(...scheduleIds)]),
+						),
+				scheduleIds.length === 0
+					? Effect.succeed([])
+					: db.run(
+							PG.from(OnCallOverrides)
+								.select("id", "scheduleId", "userId", "startsAt", "endsAt", "createdAt")
+								.where(($) => [
+									$.orgId.eq(orgId),
+									$.scheduleId.in_(...scheduleIds),
+									$.endsAt.gt(nowMs - ONCALL_LOOKAROUND_MS),
+									$.startsAt.lt(nowMs + ONCALL_LOOKAROUND_MS),
+								]),
+						),
+			],
+			{ concurrency: "unbounded" },
+		)
 		const shifts = Arr.map(scheduleIds, (scheduleId) => {
 			const schedule = Arr.findFirst(schedules, (row) => row.id === scheduleId)
 			if (Option.isNone(schedule)) return { scheduleId, userId: null }
@@ -368,19 +372,22 @@ export const loadStepRecipients = (
 		)
 		const base = { userIds, gapScheduleIds, onCall, missingUserIds: [] }
 		if (userIds.length === 0) return { ...base, recipients: [] } satisfies StepRecipients
-		const [identities, workspaces] = yield* Effect.all([
-			db.run(
-				PG.from(ChatIdentities)
-					.select("id", "userId", "connector", "externalUserId", "createdAt")
-					.where(($) => [$.orgId.eq(orgId), $.userId.in_(...userIds)])
-					.orderBy(["createdAt", "asc"], ["id", "asc"]),
-			),
-			db.run(
-				PG.from(ChatWorkspaces)
-					.select("connector")
-					.where(($) => [$.orgId.eq(orgId)]),
-			),
-		])
+		const [identities, workspaces] = yield* Effect.all(
+			[
+				db.run(
+					PG.from(ChatIdentities)
+						.select("id", "userId", "connector", "externalUserId", "createdAt")
+						.where(($) => [$.orgId.eq(orgId), $.userId.in_(...userIds)])
+						.orderBy(["createdAt", "asc"], ["id", "asc"]),
+				),
+				db.run(
+					PG.from(ChatWorkspaces)
+						.select("connector")
+						.where(($) => [$.orgId.eq(orgId)]),
+				),
+			],
+			{ concurrency: "unbounded" },
+		)
 		const linked = new Set(Arr.map(workspaces, (row) => row.connector))
 		// A connector that cannot open a DM, or with no workspace to send through, pages nobody.
 		const chatRecipients = Arr.flatMap(identities, (identity): ReadonlyArray<MemberRecipient> =>