From ed5fe399ebc178bfa65f35a344213237a96c437f Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 10 Nov 2026 02:27:10 +0100 Subject: [PATCH 1/2] perf(backend): batch digest and push writes, read in parallel, purge fix verifications with the org - DigestService: the daily membership sweep wrote one upsert per member. It now writes multi-row upserts of 500, deduped by (org, user) with the last entry winning as before, because Postgres rejects an upsert that touches a row twice in one statement. Clerk member lists are read four orgs at a time instead of one. - MobileDevicesService.markPushed: one UPDATE for the whole id list instead of one per device. - OnCallSchedulesService.list, AnomalyAlertSource.observe and loadStepRecipients: independent reads outside a transaction now run together on the invocation's pool instead of one after another. - OrganizationService: error_issue_verifications is purged on org deletion, like the issue tables it points at. A leftover waiting row was selected by the verification tick on every run and skipped, taking a slot of the per-tick limit for good. --- .../services/alerts/OnCallSchedulesService.ts | 40 +++--- .../services/alerts/alert-escalation-runs.ts | 75 ++++++----- .../alerts/anomaly/AnomalyAlertSource.ts | 90 +++++++------- .../src/services/digest/DigestService.test.ts | 84 +++++++++++++ .../src/services/digest/DigestService.ts | 116 ++++++++++-------- .../org/OrganizationService.delete.test.ts | 41 +++++++ .../src/services/org/OrganizationService.ts | 5 +- .../push/MobileDevicesService.test.ts | 78 ++++++++++++ .../src/services/push/MobileDevicesService.ts | 17 +-- 9 files changed, 387 insertions(+), 159 deletions(-) create mode 100644 packages/backend/src/services/push/MobileDevicesService.test.ts diff --git a/packages/backend/src/services/alerts/OnCallSchedulesService.ts b/packages/backend/src/services/alerts/OnCallSchedulesService.ts index cb7b4c01fe..56081c7a40 100644 --- a/packages/backend/src/services/alerts/OnCallSchedulesService.ts +++ b/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) diff --git a/packages/backend/src/services/alerts/alert-escalation-runs.ts b/packages/backend/src/services/alerts/alert-escalation-runs.ts index a174623d24..a1a7c887c1 100644 --- a/packages/backend/src/services/alerts/alert-escalation-runs.ts +++ b/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 => diff --git a/packages/backend/src/services/alerts/anomaly/AnomalyAlertSource.ts b/packages/backend/src/services/alerts/anomaly/AnomalyAlertSource.ts index 4349cba9a7..ff560bb83f 100644 --- a/packages/backend/src/services/alerts/anomaly/AnomalyAlertSource.ts +++ b/packages/backend/src/services/alerts/anomaly/AnomalyAlertSource.ts @@ -129,49 +129,53 @@ export class AnomalyAlertSource extends Context.Service - Effect.all({ - settings: detectionSettings(db, orgId), - rules: db.run( - PG.from(AlertRules) - .select( - "id", - "severity", - "thresholdsJson", - "serviceNamesJson", - "environmentsJson", - "anomalySignalsJson", - ) - .where(($) => [ - $.orgId.eq(orgId), - $.signalType.eq("anomaly"), - $.enabled.eq(true), - ]), - ), - anomalies: db.run( - PG.from(AnomalyIncidents) - .select( - "id", - "signalType", - "serviceName", - "deploymentEnv", - "severity", - "lastObservedValue", - "baselineMedian", - "errorIssueId", - ) - .where(($) => [$.orgId.eq(orgId), $.status.eq("open")]), - ), - // Every anomaly rule's open incidents, so an anomaly stays on the rule paging it. - openAlertIncidents: db.run( - PG.from(AlertIncidents) - .select("ruleId", "groupKey") - .where(($) => [ - $.orgId.eq(orgId), - $.signalType.eq("anomaly"), - $.status.eq("open"), - ]), - ), - }), + Effect.all( + { + settings: detectionSettings(db, orgId), + rules: db.run( + PG.from(AlertRules) + .select( + "id", + "severity", + "thresholdsJson", + "serviceNamesJson", + "environmentsJson", + "anomalySignalsJson", + ) + .where(($) => [ + $.orgId.eq(orgId), + $.signalType.eq("anomaly"), + $.enabled.eq(true), + ]), + ), + anomalies: db.run( + PG.from(AnomalyIncidents) + .select( + "id", + "signalType", + "serviceName", + "deploymentEnv", + "severity", + "lastObservedValue", + "baselineMedian", + "errorIssueId", + ) + .where(($) => [$.orgId.eq(orgId), $.status.eq("open")]), + ), + // Every anomaly rule's open incidents, so an anomaly stays on the rule paging it. + openAlertIncidents: db.run( + PG.from(AlertIncidents) + .select("ruleId", "groupKey") + .where(($) => [ + $.orgId.eq(orgId), + $.signalType.eq("anomaly"), + $.status.eq("open"), + ]), + ), + }, + // Four independent reads, so they share the pool instead of queueing. + { concurrency: "unbounded" }, + ), ) const candidates = Arr.map(rules, (rule) => ({ id: rule.id, diff --git a/packages/backend/src/services/digest/DigestService.test.ts b/packages/backend/src/services/digest/DigestService.test.ts index f4e8f39b0c..e689b8645d 100644 --- a/packages/backend/src/services/digest/DigestService.test.ts +++ b/packages/backend/src/services/digest/DigestService.test.ts @@ -923,3 +923,87 @@ describe("DigestService — subscriber opt-out", () => { }).pipe(Effect.provide(layer)) }) }) + +describe("DigestService.reconcileSubscriptions", () => { + const subsOf = (orgId: OrgId) => + Effect.gen(function* () { + const database = yield* Database + return yield* database.execute((db) => + db.run( + PG.from(DigestSubscriptions) + .select() + .where(($) => [$.orgId.eq(orgId)]), + ), + ) + }) + + it.effect("subscribes every member of every org, past one statement's worth of rows", () => { + const { layer } = makeHarness() + const otherOrg = OrgId.make("org_digest_other") + const member = (orgId: OrgId, index: number) => ({ + orgId, + userId: UserId.make(`user-${index}`), + email: `member-${index}@example.com`, + }) + return Effect.gen(function* () { + yield* TestClock.setTime(TICK_MS) + const digest = yield* DigestService + + yield* digest.reconcileSubscriptions([ + ...Arr.makeBy(520, (index) => member(ORG_ID, index)), + ...Arr.makeBy(3, (index) => member(otherOrg, index)), + ]) + + const first = yield* subsOf(ORG_ID) + assert.strictEqual(new Set(Arr.map(first, (row) => row.userId)).size, 520) + assert.strictEqual(new Set(Arr.map(first, (row) => row.id)).size, 520) + assert.isTrue( + Arr.every( + first, + (row) => + row.enabled && + row.dayOfWeek === 1 && + row.timezone === "UTC" && + row.createdAt === TICK_MS, + ), + ) + assert.deepStrictEqual(Arr.map(yield* subsOf(otherOrg), (row) => row.email).sort(), [ + "member-0@example.com", + "member-1@example.com", + "member-2@example.com", + ]) + }).pipe(Effect.provide(layer)) + }) + + it.effect("updates a stored email and keeps one row for a member Clerk lists twice", () => { + const { layer } = makeHarness() + return Effect.gen(function* () { + yield* TestClock.setTime(TICK_MS) + const digest = yield* DigestService + const existing = yield* seedSub({ + email: "old@example.com", + dayOfWeek: 4, + timezone: "Europe/Berlin", + }) + const existingUser = UserId.make(`user-${existing}`) + const repeated = UserId.make("user-repeated") + + yield* digest.reconcileSubscriptions([ + { orgId: ORG_ID, userId: repeated, email: "first@example.com" }, + { orgId: ORG_ID, userId: existingUser, email: "new@example.com" }, + { orgId: ORG_ID, userId: repeated, email: "second@example.com" }, + ]) + + // The sweep owns the email and nothing else on a row that already exists. + const kept = yield* getSub(existing) + assert.strictEqual(kept.email, "new@example.com") + assert.strictEqual(kept.dayOfWeek, 4) + assert.strictEqual(kept.timezone, "Europe/Berlin") + const rows = Arr.filter(yield* subsOf(ORG_ID), (row) => row.userId === repeated) + assert.deepStrictEqual( + Arr.map(rows, (row) => row.email), + ["second@example.com"], + ) + }).pipe(Effect.provide(layer)) + }) +}) diff --git a/packages/backend/src/services/digest/DigestService.ts b/packages/backend/src/services/digest/DigestService.ts index 57bbbcfeb6..04ff3efb9f 100644 --- a/packages/backend/src/services/digest/DigestService.ts +++ b/packages/backend/src/services/digest/DigestService.ts @@ -879,6 +879,11 @@ export class DigestService extends Context.Service()("@maple/back // holds starts empty on every tick. const SYNC_WINDOW_MS = 15 * 60 * 1000 const CLERK_PAGE_SIZE = 100 + // Orgs whose member lists are read at once: under a Worker's six open connections + // and well clear of Clerk's rate limit. + const CLERK_ORG_CONCURRENCY = 4 + // Nine bound parameters per row, far under Postgres's 65535 per statement. + const MEMBERSHIP_UPSERT_CHUNK = 500 const paginateClerk = ( spanName: string, @@ -913,32 +918,35 @@ export class DigestService extends Context.Service()("@maple/back "Failed to list Clerk organizations", ) - const perOrgMemberships = yield* Effect.forEach(orgs, (org) => - Effect.gen(function* () { - const members = yield* paginateClerk( - "Clerk.organizations.getOrganizationMembershipList", - { orgId: org.id }, - (params) => - clerk.organizations.getOrganizationMembershipList({ - organizationId: org.id, - ...params, - }), - `Failed to list Clerk members for org ${org.id}`, - ) + const perOrgMemberships = yield* Effect.forEach( + orgs, + (org) => + Effect.gen(function* () { + const members = yield* paginateClerk( + "Clerk.organizations.getOrganizationMembershipList", + { orgId: org.id }, + (params) => + clerk.organizations.getOrganizationMembershipList({ + organizationId: org.id, + ...params, + }), + `Failed to list Clerk members for org ${org.id}`, + ) - return Arr.flatMap(members, (member) => { - const memberEmail = member.publicUserData?.identifier - const memberUserId = member.publicUserData?.userId - if (!memberEmail || !memberUserId) return [] - return [ - { - orgId: OrgId.make(org.id), - userId: UserId.make(memberUserId), - email: memberEmail, - }, - ] - }) - }), + return Arr.flatMap(members, (member) => { + const memberEmail = member.publicUserData?.identifier + const memberUserId = member.publicUserData?.userId + if (!memberEmail || !memberUserId) return [] + return [ + { + orgId: OrgId.make(org.id), + userId: UserId.make(memberUserId), + email: memberEmail, + }, + ] + }) + }), + { concurrency: CLERK_ORG_CONCURRENCY }, ) return perOrgMemberships.flat() @@ -949,40 +957,44 @@ export class DigestService extends Context.Service()("@maple/back ) { const now = yield* Clock.currentTimeMillis + // One row per (org, user), the last one winning as it did row by row: Postgres + // rejects an upsert that touches the same row twice in one statement. + const members = Arr.fromIterable( + new Map(Arr.map(clerkMemberships, (m) => [`${m.orgId}:${m.userId}`, m] as const)).values(), + ) + const rows = yield* Effect.forEach(members, (m) => + Effect.map(random.uuidV4(), (id) => ({ + id, + orgId: m.orgId, + userId: m.userId, + email: m.email, + enabled: true, + dayOfWeek: 1, + timezone: "UTC", + createdAt: now, + updatedAt: now, + })), + ) // Upsert all current Clerk members (re-enables returning members, updates // email). `enabled` is recomputed from the stored opt-out rather than // forced true: a member who turned the digest off stays off, while one // this sweep disabled when they left the org comes back enabled. yield* Effect.forEach( - clerkMemberships, - (m) => - Effect.flatMap(random.uuidV4(), (id) => - dbExecute((db) => - db.run( - PG.insertInto(DigestSubscriptions) - .values({ - id, - orgId: m.orgId, - userId: m.userId, - email: m.email, - enabled: true, - dayOfWeek: 1, - timezone: "UTC", - createdAt: now, + Arr.chunksOf(rows, MEMBERSHIP_UPSERT_CHUNK), + (chunk) => + dbExecute((db) => + db.run( + PG.insertInto(DigestSubscriptions) + .values(chunk) + .onConflictDoUpdate({ + target: ["orgId", "userId"], + set: ($, excluded) => ({ + email: excluded.email, + enabled: PG.asBoolean($.optedOutAt.isNull()), + webAnalyticsEnabled: PG.asBoolean($.webAnalyticsOptedOutAt.isNull()), updatedAt: now, - }) - .onConflictDoUpdate({ - target: ["orgId", "userId"], - set: ($) => ({ - email: m.email, - enabled: PG.asBoolean($.optedOutAt.isNull()), - webAnalyticsEnabled: PG.asBoolean( - $.webAnalyticsOptedOutAt.isNull(), - ), - updatedAt: now, - }), }), - ), + }), ), ), { discard: true }, diff --git a/packages/backend/src/services/org/OrganizationService.delete.test.ts b/packages/backend/src/services/org/OrganizationService.delete.test.ts index c614617262..27bced6431 100644 --- a/packages/backend/src/services/org/OrganizationService.delete.test.ts +++ b/packages/backend/src/services/org/OrganizationService.delete.test.ts @@ -1,7 +1,12 @@ +import { randomUUID } from "node:crypto" import { afterEach, assert, describe, it } from "@effect/vitest" import { ConfigProvider, Effect, Exit, Layer, Option, Schema, Array as Arr, Record as Rec } from "effect" import { FetchHttpClient } from "effect/http" +import * as PG from "@maple-dev/effect-orm/postgres" +import { ErrorIssueVerifications } from "@maple/db/tables" import { OrganizationForbiddenError, OrgId, RoleName } from "@maple/domain/http" +import { ErrorIssueId, ErrorIssuePullRequestId, ErrorIssueVerificationId } from "@maple/domain/primitives" +import { Database } from "@maple/backend/platform/DatabaseLive" import { Env } from "@maple/backend/platform/Env" import { cleanupTestDbs, createTestDb, executeSql, type TestDb } from "@maple/backend/platform/test-pglite" import { countOrgRows, ORG_A, ORG_B, seedOrgRow } from "@maple/backend/platform/test-two-orgs" @@ -124,4 +129,40 @@ describe("OrganizationService.delete", () => { assert.strictEqual(yield* countOrgRows(testDb, "org_billing_links", "billing_org_id", ORG_A), 0) }).pipe(Effect.provide(makeLayer(testDb))) }) + + it.effect("leaves the verification tick nothing due for the deleted org", () => { + const testDb = createTestDb(trackedDbs) + const NOW_MS = Date.UTC(2026, 6, 6, 9, 0, 0) + const waiting = (orgId: OrgId) => ({ + id: ErrorIssueVerificationId.make(randomUUID()), + orgId, + issueId: ErrorIssueId.make(randomUUID()), + pullRequestId: ErrorIssuePullRequestId.make(randomUUID()), + mergedAt: NOW_MS - 7_200_000, + verifyAfter: NOW_MS - 3_600_000, + createdAt: NOW_MS - 7_200_000, + updatedAt: NOW_MS - 7_200_000, + }) + return Effect.gen(function* () { + const database = yield* Database + yield* database.execute((db) => + db.run(PG.insertInto(ErrorIssueVerifications).values([waiting(ORG_A), waiting(ORG_B)])), + ) + const organizations = yield* OrganizationService + yield* organizations.delete(ORG_A, [role("org:admin")]) + + // What the tick selects: every org's waiting rows whose window has closed. + const due = yield* database.execute((db) => + db.run( + PG.from(ErrorIssueVerifications) + .select("orgId") + .where(($) => [$.status.eq("waiting"), $.verifyAfter.lte(NOW_MS)]), + ), + ) + assert.deepStrictEqual( + Arr.map(due, (row) => row.orgId), + [ORG_B], + ) + }).pipe(Effect.provide(Layer.merge(makeLayer(testDb), testDb.layer))) + }) }) diff --git a/packages/backend/src/services/org/OrganizationService.ts b/packages/backend/src/services/org/OrganizationService.ts index 4cc7493826..541a9925b3 100644 --- a/packages/backend/src/services/org/OrganizationService.ts +++ b/packages/backend/src/services/org/OrganizationService.ts @@ -61,6 +61,7 @@ import { ErrorIssueEvents, ErrorIssueFingerprints, ErrorIssueStates, + ErrorIssueVerifications, ErrorIssues, ErrorNotificationPolicies, LiveActivities, @@ -214,6 +215,9 @@ const ORG_SCOPED_TABLES: ReadonlyArray = [ ErrorIncidents, // The aliases go with the issues they point at. ErrorIssueFingerprints, + // A `waiting` row left behind is due work: the verification tick scans every org's and + // would pick it up on each run, for an issue that is purged below. + ErrorIssueVerifications, ErrorIssues, ErrorNotificationPolicies, Actors, @@ -315,7 +319,6 @@ export const UNPURGED_ORG_SCOPED_TABLES = [ "CloudflareZoneSettings", "ErrorFingerprintCandidates", "ErrorIssuePullRequests", - "ErrorIssueVerifications", "ErrorNotificationDeliveries", "ErrorTickStates", "Investigations", diff --git a/packages/backend/src/services/push/MobileDevicesService.test.ts b/packages/backend/src/services/push/MobileDevicesService.test.ts new file mode 100644 index 0000000000..06051196e6 --- /dev/null +++ b/packages/backend/src/services/push/MobileDevicesService.test.ts @@ -0,0 +1,78 @@ +import { afterEach, assert, describe, it } from "@effect/vitest" +import * as PG from "@maple-dev/effect-orm/postgres" +import { MobileDevices } from "@maple/db/tables" +import { OrgId, UserId } from "@maple/domain/primitives" +import { Array as Arr, Effect, Layer } from "effect" +import { TestClock } from "effect/testing" +import { Database } from "@maple/backend/platform/DatabaseLive" +import { cleanupTestDbs, createTestDb, type TestDb } from "@maple/backend/platform/test-pglite" +import { MobileDevicesService } from "./MobileDevicesService" + +const createdDbs: TestDb[] = [] + +afterEach(() => cleanupTestDbs(createdDbs)) + +const ORG = OrgId.make("org_push_test") +const OTHER_ORG = OrgId.make("org_push_other") +const USER = UserId.make("user_push_test") +const REGISTERED_MS = Date.UTC(2026, 6, 6, 9, 0, 0) +const PUSHED_MS = REGISTERED_MS + 60_000 + +const makeLayer = () => MobileDevicesService.layer.pipe(Layer.provideMerge(createTestDb(createdDbs).layer)) + +const register = (orgId: OrgId, token: string) => + Effect.gen(function* () { + const devices = yield* MobileDevicesService + return yield* devices.register(orgId, USER, { + platform: "ios", + token, + environment: "production", + bundleId: "com.maple.mobile", + }) + }) + +/** `lastPushedAt` by token, read from the table: the service does not expose it. */ +const lastPushedByToken = Effect.gen(function* () { + const database = yield* Database + const rows = yield* database.execute((db) => + db.run(PG.from(MobileDevices).select("token", "lastPushedAt").orderBy(["token", "asc"])), + ) + return Arr.map(rows, (row) => [row.token, row.lastPushedAt] as const) +}) + +describe("MobileDevicesService.markPushed", () => { + it.effect("stamps every named device of the org and no other row", () => + Effect.gen(function* () { + yield* TestClock.setTime(REGISTERED_MS) + const devices = yield* MobileDevicesService + const first = yield* register(ORG, "token-a") + const second = yield* register(ORG, "token-b") + yield* register(ORG, "token-c") + const foreign = yield* register(OTHER_ORG, "token-d") + + yield* TestClock.setTime(PUSHED_MS) + // The other org's device id is named too: the write stays inside the org. + yield* devices.markPushed(ORG, [first.id, second.id, foreign.id]) + + assert.deepStrictEqual(yield* lastPushedByToken, [ + ["token-a", PUSHED_MS], + ["token-b", PUSHED_MS], + ["token-c", null], + ["token-d", null], + ]) + }).pipe(Effect.provide(makeLayer())), + ) + + it.effect("writes nothing for an empty list", () => + Effect.gen(function* () { + yield* TestClock.setTime(REGISTERED_MS) + const devices = yield* MobileDevicesService + yield* register(ORG, "token-a") + + yield* TestClock.setTime(PUSHED_MS) + yield* devices.markPushed(ORG, []) + + assert.deepStrictEqual(yield* lastPushedByToken, [["token-a", null]]) + }).pipe(Effect.provide(makeLayer())), + ) +}) diff --git a/packages/backend/src/services/push/MobileDevicesService.ts b/packages/backend/src/services/push/MobileDevicesService.ts index 97b48843b2..3aeaafbaf2 100644 --- a/packages/backend/src/services/push/MobileDevicesService.ts +++ b/packages/backend/src/services/push/MobileDevicesService.ts @@ -313,17 +313,12 @@ export class MobileDevicesService extends Context.Service - dbExecute((db) => - db.run( - PG.update(MobileDevices) - .set({ lastPushedAt: now }) - .where(($) => [$.orgId.eq(orgId), $.id.eq(id)]), - ), - ), - { discard: true }, + yield* dbExecute((db) => + db.run( + PG.update(MobileDevices) + .set({ lastPushedAt: now }) + .where(($) => [$.orgId.eq(orgId), PG.inList($.id, ids)]), + ), ) }) From 43fd814b95ef6f2f03d17823142451f95dcd4971 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Tue, 10 Nov 2026 02:27:18 +0100 Subject: [PATCH 2/2] chore(backend): share the bearer token and ingest key parsers, delete dead code - getBearerToken had four identical copies, one per auth layer. It lives in services/auth/bearer-token.ts now. - parseEncryptionKey had three copies differing only in the error they build. platform/Crypto.ts exports parseIngestKeyEncryptionKey, which takes the error constructor. - Deleted services/alerts/alert-templates (and its test): no production importer in any workspace, the dashboard and the MCP tool keep their own. - Deleted the unreferenced GithubApiPullRequestFile and GithubReviewThread types, and captureReusableContext, which only its test called. That test now exercises withoutInvocationContext, the function production uses. - Two comments said every execute dials its own Postgres client. The invocation's pool is reused, so they now say what batching saves. --- packages/backend/src/platform/Crypto.ts | 12 + .../src/platform/invocation-context.test.ts | 7 +- .../src/platform/invocation-context.ts | 5 +- .../alerts/AnomalyDetectionService.ts | 8 +- .../services/alerts/alert-templates/index.ts | 17 - .../alert-templates/planetscale.test.ts | 102 ------ .../alerts/alert-templates/planetscale.ts | 309 ------------------ .../services/alerts/alert-templates/types.ts | 67 ---- .../src/services/auth/ApiAuthorizationLive.ts | 11 +- .../services/auth/ApiAuthorizationV2Live.ts | 9 +- .../services/auth/SessionAuthorizationLive.ts | 9 +- .../auth/UserSessionAuthorizationLive.ts | 9 +- .../backend/src/services/auth/bearer-token.ts | 10 + .../integrations/ScrapeTargetsService.ts | 16 +- .../integrations/scrape-check-retention.ts | 6 +- .../vcs/vendor/github/GithubAppClient.ts | 7 - .../org/OrgClickHouseSettingsService.ts | 18 +- .../src/services/org/OrgIngestKeysService.ts | 16 +- 18 files changed, 49 insertions(+), 589 deletions(-) delete mode 100644 packages/backend/src/services/alerts/alert-templates/index.ts delete mode 100644 packages/backend/src/services/alerts/alert-templates/planetscale.test.ts delete mode 100644 packages/backend/src/services/alerts/alert-templates/planetscale.ts delete mode 100644 packages/backend/src/services/alerts/alert-templates/types.ts create mode 100644 packages/backend/src/services/auth/bearer-token.ts diff --git a/packages/backend/src/platform/Crypto.ts b/packages/backend/src/platform/Crypto.ts index 328ca850f1..b153ee9a29 100644 --- a/packages/backend/src/platform/Crypto.ts +++ b/packages/backend/src/platform/Crypto.ts @@ -22,6 +22,18 @@ export const parseBase64Aes256GcmKey = (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 = (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 diff --git a/packages/backend/src/platform/invocation-context.test.ts b/packages/backend/src/platform/invocation-context.test.ts index 0e9cbb7c77..21c2ea92e1 100644 --- a/packages/backend/src/platform/invocation-context.test.ts +++ b/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()("@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().pipe( + const captured = yield* Effect.context().pipe( + Effect.map(withoutInvocationContext), Effect.provideService(Greeting, { text: "hello" }), Effect.provideService(PgConnectionScope, constructionScope), Effect.provideService(DrainRunOnInterrupt, true), diff --git a/packages/backend/src/platform/invocation-context.ts b/packages/backend/src/platform/invocation-context.ts index 29d74bfb91..3bc2f055bf 100644 --- a/packages/backend/src/platform/invocation-context.ts +++ b/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 = (context: Context.Context) => Tracer.ParentSpan, HttpServerRequest.HttpServerRequest, )(context) - -/** `Effect.context`, safe to keep and provide on later invocations. */ -export const captureReusableContext = () => Effect.map(Effect.context(), withoutInvocationContext) diff --git a/packages/backend/src/services/alerts/AnomalyDetectionService.ts b/packages/backend/src/services/alerts/AnomalyDetectionService.ts index 54a109eaac..96fe9c95fd 100644 --- a/packages/backend/src/services/alerts/AnomalyDetectionService.ts +++ b/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 diff --git a/packages/backend/src/services/alerts/alert-templates/index.ts b/packages/backend/src/services/alerts/alert-templates/index.ts deleted file mode 100644 index 74057fe973..0000000000 --- a/packages/backend/src/services/alerts/alert-templates/index.ts +++ /dev/null @@ -1,17 +0,0 @@ -import { Array as Arr, Option } from "effect" -import { PLANETSCALE_ALERT_TEMPLATES } from "./planetscale" -import type { AlertTemplateDefinition } from "./types" - -export type { AlertTemplateDefinition } from "./types" - -/** - * Every ready-made alert rule Maple can offer. - * - * One flat registry rather than per-integration lookups: the alerts UI lists - * them together, and a template's `requirement` is what decides whether it is - * offered, not which file it lives in. - */ -export const ALERT_TEMPLATES: ReadonlyArray = [...PLANETSCALE_ALERT_TEMPLATES] - -export const getAlertTemplate = (id: string): AlertTemplateDefinition | undefined => - Option.getOrUndefined(Arr.findFirst(ALERT_TEMPLATES, (template) => template.id === id)) diff --git a/packages/backend/src/services/alerts/alert-templates/planetscale.test.ts b/packages/backend/src/services/alerts/alert-templates/planetscale.test.ts deleted file mode 100644 index 6789e54077..0000000000 --- a/packages/backend/src/services/alerts/alert-templates/planetscale.test.ts +++ /dev/null @@ -1,102 +0,0 @@ -import { assert, describe, it } from "@effect/vitest" -import { AlertRuleUpsertRequest, MAX_ALERT_WINDOW_MINUTES } from "@maple/domain/http" -import { Schema, Array as Arr } from "effect" -import { ALERT_TEMPLATES, getAlertTemplate } from "./index" - -const DESTINATIONS = ["11111111-1111-4111-8111-111111111111"] - -const decode = Schema.decodeUnknownSync(AlertRuleUpsertRequest) - -const buildAll = (values: Record = {}) => - Arr.flatMap(ALERT_TEMPLATES, (template) => - Arr.map(template.build({ values, destinationIds: DESTINATIONS, enabled: true }), (rule) => ({ - templateId: template.id, - rule, - })), - ) - -describe("PlanetScale alert templates", () => { - it("produces rules that round-trip through the upsert schema", () => { - // This is the test that catches a template whose rule would only fail at - // creation time, in front of the operator who clicked "Apply". - for (const { templateId, rule } of buildAll({ threshold: "800", database: "main-db" })) { - assert.doesNotThrow(() => decode(rule), `${templateId} produced an invalid rule`) - } - }) - - it("stays within the alert window cap", () => { - for (const { templateId, rule } of buildAll({ threshold: "800" })) { - assert.isAtMost( - rule.windowMinutes, - MAX_ALERT_WINDOW_MINUTES, - `${templateId} exceeds the window cap`, - ) - } - }) - - it("tags every rule with its template so 'already applied' needs no new column", () => { - for (const { templateId, rule } of buildAll({ threshold: "800" })) { - assert.include(rule.tags ?? [], `template:${templateId}`) - assert.include(rule.tags ?? [], "planetscale") - } - }) - - it("emits one rule per product for the metrics that differ by engine", () => { - // MySQL and Postgres report replication lag under different names; an - // operator asking for "replica lag" means both. - const lag = getAlertTemplate("ps-replica-lag")! - const rules = lag.build({ values: {}, destinationIds: DESTINATIONS, enabled: true }) - assert.strictEqual(rules.length, 2) - const metrics = Arr.map(rules, (rule) => - rule.queryBuilderDraft !== null && rule.queryBuilderDraft !== undefined - ? (rule.queryBuilderDraft as { metricName?: string }).metricName - : undefined, - ) - assert.includeMembers(metrics, [ - "planetscale_mysql_replica_lag_seconds", - "planetscale_postgres_replica_lag_seconds", - ]) - }) - - it("groups by branch rather than filtering on it", () => { - // Metrics queries honor exactly one attr.* equality, so a database+branch - // filter would silently drop the branch and alert on every branch. - const lag = getAlertTemplate("ps-replica-lag")! - const [rule] = lag.build({ - values: { database: "main-db" }, - destinationIds: DESTINATIONS, - enabled: true, - }) - const draft = rule.queryBuilderDraft as { whereClause: string; groupBy: string[] } - assert.strictEqual(draft.whereClause, "attr.planetscale_database_name = 'main-db'") - assert.deepStrictEqual(draft.groupBy, ["attr.planetscale_branch_name"]) - }) - - it("watches every database when none is named", () => { - const lag = getAlertTemplate("ps-replica-lag")! - const [rule] = lag.build({ values: {}, destinationIds: DESTINATIONS, enabled: true }) - assert.strictEqual((rule.queryBuilderDraft as { whereClause: string }).whereClause, "") - }) - - it("escapes a quote in a database name rather than breaking the where clause", () => { - const lag = getAlertTemplate("ps-replica-lag")! - const [rule] = lag.build({ - values: { database: "it's-db" }, - destinationIds: DESTINATIONS, - enabled: true, - }) - assert.strictEqual( - (rule.queryBuilderDraft as { whereClause: string }).whereClause, - "attr.planetscale_database_name = 'it''s-db'", - ) - }) - - it("falls back to the documented default when a numeric parameter is unparseable", () => { - const [rule] = getAlertTemplate("ps-replica-lag")!.build({ - values: { thresholdSeconds: "not-a-number" }, - destinationIds: DESTINATIONS, - enabled: true, - }) - assert.strictEqual(rule.threshold, 30) - }) -}) diff --git a/packages/backend/src/services/alerts/alert-templates/planetscale.ts b/packages/backend/src/services/alerts/alert-templates/planetscale.ts deleted file mode 100644 index 9935143448..0000000000 --- a/packages/backend/src/services/alerts/alert-templates/planetscale.ts +++ /dev/null @@ -1,309 +0,0 @@ -import { AlertRuleUpsertRequest } from "@maple/domain/http" -import { - PLANETSCALE_STORAGE_RUNWAY_SQL, - PLANETSCALE_STORAGE_USED_SQL, -} from "@maple/query-engine-integrations" -import { Schema, Array as Arr, String as Str } from "effect" -import type { AlertTemplateDefinition } from "./types" -import { templateTag } from "./types" - -/** - * Alert templates for the signals a PlanetScale database can page someone about. - * - * Branch OOM is deliberately absent: it arrives as a webhook, already opens a - * `kind="integration"` triage issue, and already reaches people through the - * error-notification policy. Duplicating it as a metric rule would double-notify - * on an event that isn't a metric. It gets a timeline marker instead. - */ - -const decodeDestinationIds = Schema.decodeUnknownSync(AlertRuleUpsertRequest.fields.destinationIds) - -const numericParam = (values: Readonly>, key: string, fallback: number) => { - const parsed = Number(values[key]) - return Number.isFinite(parsed) ? parsed : fallback -} - -const DATABASE_PARAM = { - key: "database", - label: "Database", - description: "Leave blank to watch every PlanetScale database in the organization.", - required: false, - placeholder: "main-db", -} as const - -/** - * Metrics queries honor exactly ONE `attr.*` equality filter — a second is - * silently dropped. So a rule can't say "this database AND this branch"; it says - * "this database" and groups by branch, which evaluates per branch and is the - * honest version of the same intent. - */ -const whereFor = (database: string | undefined) => - database === undefined || database === "" - ? "" - : `attr.planetscale_database_name = '${Str.replace(/'/g, "''")(database)}'` - -const metricsDraft = (input: { - readonly name: string - readonly metricName: string - readonly aggregation: string - readonly database: string | undefined -}) => ({ - id: input.metricName, - name: input.name, - enabled: true, - hidden: false, - dataSource: "metrics" as const, - signalSource: "default" as const, - metricName: input.metricName, - metricType: "gauge" as const, - isMonotonic: false, - whereClause: whereFor(input.database), - aggregation: input.aggregation, - stepInterval: "auto", - orderByDirection: "desc" as const, - addOns: { groupBy: true, having: false, orderBy: false, limit: false, legend: false }, - // Per branch, since the single-equality limit means the branch can't be a filter. - groupBy: ["attr.planetscale_branch_name"], - having: "", - orderBy: "", - limit: "", - legend: "", -}) - -const REQUIREMENT = { - kind: "integration", - integrationId: "planetscale", - label: "PlanetScale", - missing: "PlanetScale not connected", -} as const - -/** - * MySQL/Vitess and Postgres expose the same concept under different metric - * names, and an operator asking for "replica lag" means both. This is why - * `build` returns an array. - */ -const REPLICA_LAG_METRICS = [ - { metric: "planetscale_mysql_replica_lag_seconds", label: "MySQL" }, - { metric: "planetscale_postgres_replica_lag_seconds", label: "Postgres" }, -] as const - -const CONNECTION_METRICS = [ - { metric: "planetscale_edge_active_connections", label: "MySQL" }, - { metric: "planetscale_edge_postgres_active_connections", label: "Postgres" }, - // Connections without a service token are polled from the metrics API under this name. - { metric: "planetscale_api_connections", label: "API" }, -] as const - -const suffix = (database: string | undefined) => - database === undefined || database === "" ? "" : ` · ${database}` - -export const PLANETSCALE_ALERT_TEMPLATES: ReadonlyArray = [ - { - id: "ps-replica-lag", - name: "Replica lag", - description: "Warns when a branch's replicas fall behind, before reads start returning stale rows.", - category: "infrastructure", - tags: ["planetscale", "database", "replication"], - requirement: REQUIREMENT, - requiredMetricPrefixes: ["planetscale_mysql_replica_lag", "planetscale_postgres_replica_lag"], - parameters: [ - DATABASE_PARAM, - { - key: "thresholdSeconds", - label: "Lag threshold (seconds)", - description: "Fires when the worst replica in a branch exceeds this.", - required: false, - defaultValue: "30", - }, - ], - build: ({ values, destinationIds, enabled }) => - Arr.map( - REPLICA_LAG_METRICS, - ({ metric, label }) => - new AlertRuleUpsertRequest({ - name: `PlanetScale replica lag (${label})${suffix(values.database)}`, - enabled, - severity: "warning", - signalType: "builder_query", - comparator: "gt", - threshold: numericParam(values, "thresholdSeconds", 30), - windowMinutes: 5, - // Two consecutive windows: replication catches up on its own - // constantly, and a single spike is not an incident. - consecutiveBreachesRequired: 2, - tags: ["planetscale", templateTag("ps-replica-lag")], - queryBuilderDraft: metricsDraft({ - name: "Replica lag", - metricName: metric, - aggregation: "max", - database: values.database, - }), - destinationIds: decodeDestinationIds(destinationIds), - }), - ), - }, - { - id: "ps-connection-ceiling", - name: "Connection ceiling", - description: "Warns when active connections approach the limit your plan allows.", - category: "infrastructure", - tags: ["planetscale", "database", "connections"], - requirement: REQUIREMENT, - requiredMetricPrefixes: ["planetscale_"], - parameters: [ - DATABASE_PARAM, - { - key: "threshold", - label: "Connection threshold", - // Required with no default on purpose: the ceiling is a property of - // the PlanetScale plan and appears in no metric, so guessing one - // would ship an alert that fires at a number nobody chose. - description: - "Your plan's connection limit, minus the headroom you want. Not derivable from the metrics.", - required: true, - placeholder: "800", - }, - ], - build: ({ values, destinationIds, enabled }) => - Arr.map( - CONNECTION_METRICS, - ({ metric, label }) => - new AlertRuleUpsertRequest({ - name: `PlanetScale connections (${label})${suffix(values.database)}`, - enabled, - severity: "warning", - signalType: "builder_query", - comparator: "gt", - threshold: numericParam(values, "threshold", 0), - windowMinutes: 5, - consecutiveBreachesRequired: 2, - tags: ["planetscale", templateTag("ps-connection-ceiling")], - queryBuilderDraft: metricsDraft({ - name: "Active connections", - metricName: metric, - aggregation: "max", - database: values.database, - }), - destinationIds: decodeDestinationIds(destinationIds), - }), - ), - }, - { - id: "ps-storage-used", - name: "Storage used", - description: "Warns when a branch's volume crosses a usage threshold.", - category: "infrastructure", - tags: ["planetscale", "database", "storage"], - requirement: REQUIREMENT, - requiredMetricPrefixes: ["planetscale_volume_"], - parameters: [ - { - key: "thresholdPercent", - label: "Used threshold (%)", - description: "Fires when the worst branch's volume crosses this.", - required: false, - defaultValue: "85", - }, - ], - build: ({ values, destinationIds, enabled }) => [ - new AlertRuleUpsertRequest({ - name: "PlanetScale storage used", - enabled, - severity: "warning", - // raw_query, not builder_query: this is a ratio of two metrics, which - // the metrics builder cannot express. - signalType: "raw_query", - comparator: "gt", - threshold: numericParam(values, "thresholdPercent", 85), - windowMinutes: 15, - rawQuerySql: PLANETSCALE_STORAGE_USED_SQL, - rawQueryReducer: "max", - tags: ["planetscale", templateTag("ps-storage-used")], - destinationIds: decodeDestinationIds(destinationIds), - }), - ], - }, - { - id: "ps-storage-runway", - name: "Storage runway", - description: - "Fires when a branch's volume is on track to fill within N days, from the slope of its free space.", - category: "infrastructure", - tags: ["planetscale", "database", "storage", "forecast"], - requirement: REQUIREMENT, - requiredMetricPrefixes: ["planetscale_volume_available_bytes"], - parameters: [ - { - key: "thresholdDays", - label: "Runway threshold (days)", - description: "Fires when the projected days-until-full drops below this.", - required: false, - defaultValue: "14", - }, - ], - build: ({ values, destinationIds, enabled }) => [ - new AlertRuleUpsertRequest({ - name: "PlanetScale storage runway", - enabled, - severity: "critical", - signalType: "raw_query", - comparator: "lt", - threshold: numericParam(values, "thresholdDays", 14), - // A day of history is what makes the slope meaningful. This is exactly - // MAX_ALERT_WINDOW_MINUTES — there is no headroom here, and the cap - // should not be raised to buy some. - windowMinutes: 1440, - rawQuerySql: PLANETSCALE_STORAGE_RUNWAY_SQL, - // min: the branch with the least runway is the one that matters. - rawQueryReducer: "min", - tags: ["planetscale", templateTag("ps-storage-runway")], - destinationIds: decodeDestinationIds(destinationIds), - }), - ], - }, - { - id: "ps-pod-saturation", - name: "Pod saturation", - description: "Warns when a branch's pods sit near their CPU or memory limit.", - category: "infrastructure", - tags: ["planetscale", "database", "saturation"], - requirement: REQUIREMENT, - requiredMetricPrefixes: ["planetscale_pods_"], - parameters: [ - DATABASE_PARAM, - { - key: "thresholdPercent", - label: "Utilization threshold (%)", - description: "Fires when peak utilization stays above this.", - required: false, - defaultValue: "90", - }, - ], - build: ({ values, destinationIds, enabled }) => - Arr.map( - [ - { metric: "planetscale_pods_cpu_util_percentages", label: "CPU" }, - { metric: "planetscale_pods_mem_util_percentages", label: "memory" }, - ] as const, - ({ metric, label }) => - new AlertRuleUpsertRequest({ - name: `PlanetScale pod ${label}${suffix(values.database)}`, - enabled, - severity: "warning", - signalType: "builder_query", - comparator: "gt", - threshold: numericParam(values, "thresholdPercent", 90), - windowMinutes: 10, - consecutiveBreachesRequired: 2, - tags: ["planetscale", templateTag("ps-pod-saturation")], - queryBuilderDraft: metricsDraft({ - name: `Pod ${label}`, - metricName: metric, - aggregation: "max", - database: values.database, - }), - destinationIds: decodeDestinationIds(destinationIds), - }), - ), - }, -] diff --git a/packages/backend/src/services/alerts/alert-templates/types.ts b/packages/backend/src/services/alerts/alert-templates/types.ts deleted file mode 100644 index 928dace5cc..0000000000 --- a/packages/backend/src/services/alerts/alert-templates/types.ts +++ /dev/null @@ -1,67 +0,0 @@ -import type { AlertRuleUpsertRequest } from "@maple/domain/http" - -/** - * Ready-made alert rules for signals an integration knows about but a generic - * rule builder can't suggest. - * - * These are templates over the EXISTING rule machinery, deliberately not a new - * signal type. `builder_query` compiles through the same path as a dashboard - * chart and `raw_query` covers everything expressible as ClickHouse SQL, so a - * new signal type would mean new evaluation, new state keying, new incident - * rendering and a new public contract — for zero new capability. - * - * Applying a template goes through the normal AlertsService upsert, so - * name-uniqueness, compile validation and the txid response come for free. - */ - -export interface AlertTemplateParameter { - readonly key: string - readonly label: string - readonly description: string - readonly required: boolean - /** Absent for free-text; present for numeric thresholds the operator must pick. */ - readonly defaultValue?: string - readonly placeholder?: string -} - -export interface AlertTemplateRequirement { - /** What must be connected. Shown next to the template name when it isn't. */ - readonly kind: "integration" - readonly integrationId: string - readonly label: string - /** Row-sized reason the template is unavailable, e.g. "PlanetScale not connected". */ - readonly missing: string -} - -export interface AlertTemplateDefinition { - readonly id: string - readonly name: string - readonly description: string - readonly category: "infrastructure" - readonly tags: ReadonlyArray - readonly requirement: AlertTemplateRequirement - /** - * Metric-name prefixes the org must already be ingesting. The UI greys the - * template out rather than letting someone create a rule that can only ever - * report no-data. - */ - readonly requiredMetricPrefixes: ReadonlyArray - readonly parameters: ReadonlyArray - /** - * Returns an ARRAY because one operator intent can need several rules: the - * MySQL and Postgres products expose replication lag under different metric - * names, and a "replica lag" alert has to cover both. - */ - readonly build: (params: { - readonly values: Readonly> - readonly destinationIds: ReadonlyArray - readonly enabled: boolean - }) => ReadonlyArray -} - -/** - * Tag that records which template produced a rule, so "already applied" needs no - * new column. Rule tags are capped at 32 characters, which is why template ids - * are short (`ps-storage-runway`) while their user-facing `name` is not. - */ -export const templateTag = (templateId: string) => `template:${templateId}` diff --git a/packages/backend/src/services/auth/ApiAuthorizationLive.ts b/packages/backend/src/services/auth/ApiAuthorizationLive.ts index 363b0e8d19..d2595d5579 100644 --- a/packages/backend/src/services/auth/ApiAuthorizationLive.ts +++ b/packages/backend/src/services/auth/ApiAuthorizationLive.ts @@ -1,6 +1,6 @@ import { HttpServerRequest } from "effect/http" import { CurrentTenant, RoleName, UnauthorizedError } from "@maple/domain/http" -import { Effect, Layer, Option, Schema, String as Str } from "effect" +import { Effect, Layer, Option, Schema } from "effect" import { ApiKeysService } from "@maple/backend/services/org/ApiKeysService" import { makeResolveTenant } from "./AuthService" import { annotateAuthSpan } from "@maple/backend/services/auth/auth-span" @@ -10,18 +10,11 @@ import { recordApiDenial } from "@maple/backend/services/auth/audit-denial" import { withAuditedRead } from "@maple/backend/services/audit/audit-access" import { Env } from "@maple/backend/platform/Env" import { OrganizationRegionService } from "@maple/backend/services/org/OrganizationRegionService" +import { getBearerToken } from "./bearer-token" const decodeRoleNameSync = Schema.decodeUnknownSync(RoleName) const apiKeyDefaultRoles = [decodeRoleNameSync("root")] as const -const getBearerToken = (headers: Record): string | undefined => { - const header = headers["authorization"] ?? headers["Authorization"] - if (!header) return undefined - const [scheme, token] = Str.split(" ")(header) - if (!scheme || !token || Str.toLowerCase(scheme) !== "bearer") return undefined - return token -} - export const ApiAuthorizationLive = Layer.effect( CurrentTenant.Authorization, Effect.gen(function* () { diff --git a/packages/backend/src/services/auth/ApiAuthorizationV2Live.ts b/packages/backend/src/services/auth/ApiAuthorizationV2Live.ts index ec4c5d9efc..00e7ad0db6 100644 --- a/packages/backend/src/services/auth/ApiAuthorizationV2Live.ts +++ b/packages/backend/src/services/auth/ApiAuthorizationV2Live.ts @@ -32,18 +32,11 @@ import { apiV2RateLimitKey, deviceRateLimitKey, } from "./ApiV2RateLimiter" +import { getBearerToken } from "./bearer-token" const decodeRoleNameSync = Schema.decodeUnknownSync(RoleName) const apiKeyDefaultRoles = [decodeRoleNameSync("root")] as const -const getBearerToken = (headers: Record): string | undefined => { - const header = headers["authorization"] ?? headers["Authorization"] - if (!header) return undefined - const [scheme, token] = Str.split(" ")(header) - if (!scheme || !token || Str.toLowerCase(scheme) !== "bearer") return undefined - return token -} - /** The access token of an RFC 9449 `Authorization: DPoP ` request. */ const getDpopToken = (headers: Record): string | undefined => { const [scheme, token] = Str.split(" ")(headers["authorization"] ?? "") diff --git a/packages/backend/src/services/auth/SessionAuthorizationLive.ts b/packages/backend/src/services/auth/SessionAuthorizationLive.ts index 9248b7fc4e..c5a8610abf 100644 --- a/packages/backend/src/services/auth/SessionAuthorizationLive.ts +++ b/packages/backend/src/services/auth/SessionAuthorizationLive.ts @@ -9,14 +9,7 @@ import { AuditLogService } from "@maple/backend/services/audit/AuditLogService" import { withAuditedRead } from "@maple/backend/services/audit/audit-access" import { Env } from "@maple/backend/platform/Env" import { OrganizationRegionService } from "@maple/backend/services/org/OrganizationRegionService" - -const getBearerToken = (headers: Record): string | undefined => { - const header = headers["authorization"] ?? headers["Authorization"] - if (!header) return undefined - const [scheme, token] = Str.split(" ")(header) - if (!scheme || !token || Str.toLowerCase(scheme) !== "bearer") return undefined - return token -} +import { getBearerToken } from "./bearer-token" /** * Authorization for the internal API: Clerk sessions only. diff --git a/packages/backend/src/services/auth/UserSessionAuthorizationLive.ts b/packages/backend/src/services/auth/UserSessionAuthorizationLive.ts index 1ebeab2879..b3e25742b3 100644 --- a/packages/backend/src/services/auth/UserSessionAuthorizationLive.ts +++ b/packages/backend/src/services/auth/UserSessionAuthorizationLive.ts @@ -4,14 +4,7 @@ import { makeResolveClerkUser } from "@maple/auth" import { CurrentTenant } from "@maple/domain/http" import { Effect, Layer, String as Str } from "effect" import { Env } from "@maple/backend/platform/Env" - -const getBearerToken = (headers: Record): string | undefined => { - const header = headers["authorization"] ?? headers["Authorization"] - if (!header) return undefined - const [scheme, token] = Str.split(" ")(header) - if (!scheme || !token || Str.toLowerCase(scheme) !== "bearer") return undefined - return token -} +import { getBearerToken } from "./bearer-token" /** * Authorization for the few requests made before the caller has an organization. A Clerk session diff --git a/packages/backend/src/services/auth/bearer-token.ts b/packages/backend/src/services/auth/bearer-token.ts new file mode 100644 index 0000000000..9ee2b4e63a --- /dev/null +++ b/packages/backend/src/services/auth/bearer-token.ts @@ -0,0 +1,10 @@ +import { String as Str } from "effect" + +/** The token of an `Authorization: Bearer ` header; undefined for any other shape. */ +export const getBearerToken = (headers: Record): string | undefined => { + const header = headers["authorization"] ?? headers["Authorization"] + if (!header) return undefined + const [scheme, token] = Str.split(" ")(header) + if (!scheme || !token || Str.toLowerCase(scheme) !== "bearer") return undefined + return token +} diff --git a/packages/backend/src/services/integrations/ScrapeTargetsService.ts b/packages/backend/src/services/integrations/ScrapeTargetsService.ts index 352f8e40a2..56901f0333 100644 --- a/packages/backend/src/services/integrations/ScrapeTargetsService.ts +++ b/packages/backend/src/services/integrations/ScrapeTargetsService.ts @@ -48,7 +48,7 @@ import { import { HttpClient, HttpClientRequest } from "effect/http" import { encryptAes256Gcm, - parseBase64Aes256GcmKey, + parseIngestKeyEncryptionKey, type EncryptedValue, } from "@maple/backend/platform/Crypto" import { forkRequestScoped } from "@maple/backend/platform/fork-request-scoped" @@ -329,17 +329,6 @@ const normalizeBranchPatterns = (patterns: ReadonlyArray | undefined): s return [...seen] } -const parseEncryptionKey = (raw: string): Effect.Effect => - parseBase64Aes256GcmKey(raw, (message) => - toEncryptionError( - 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, - ), - ) - const encryptCredentials = ( plaintext: string, encryptionKey: Buffer, @@ -644,8 +633,9 @@ export class ScrapeTargetsService extends Context.Service { /** * Apply retention to the given targets. * - * Every statement runs inside ONE `execute`: under `DatabasePgLive` each call - * dials and tears down its own postgres.js client, so the handshake count is - * what costs, not the statement count. + * Every statement runs inside ONE `execute`, so the prune is one span. The + * invocation's pool is reused across executes (`DatabasePgLive`); only an entry + * point with no connection scope still dials per execute. */ export const pruneChecksForTargets = Effect.fn("ScrapeCheckRetention.pruneForTargets")(function* ( targets: ReadonlyArray, diff --git a/packages/backend/src/services/integrations/vcs/vendor/github/GithubAppClient.ts b/packages/backend/src/services/integrations/vcs/vendor/github/GithubAppClient.ts index 86986d71b0..44936cf8e5 100644 --- a/packages/backend/src/services/integrations/vcs/vendor/github/GithubAppClient.ts +++ b/packages/backend/src/services/integrations/vcs/vendor/github/GithubAppClient.ts @@ -322,7 +322,6 @@ const GithubApiPullRequestFileSchema = Schema.Struct({ deletions: Schema.Number, patch: Schema.optionalKey(Schema.String), }) -export type GithubApiPullRequestFile = Schema.Schema.Type const GithubApiPullRequestFileList = Schema.Array(GithubApiPullRequestFileSchema) const GithubApiPullRequestCommitList = Schema.Array( @@ -479,12 +478,6 @@ const GithubReviewThreadsResponse = Schema.Struct({ errors: GithubGraphqlErrors, }) const GithubMutationResponse = Schema.Struct({ errors: GithubGraphqlErrors }) -export type GithubReviewThread = NonNullable< - NonNullable< - NonNullable["data"]>["repository"] - >["pullRequest"] ->["reviewThreads"]["nodes"][number] - export type GithubApiDiscussionComment = Schema.Schema.Type[number] const GithubApiCheckRunSchema = Schema.Struct({ diff --git a/packages/backend/src/services/org/OrgClickHouseSettingsService.ts b/packages/backend/src/services/org/OrgClickHouseSettingsService.ts index afd4f806bf..0b68f581b6 100644 --- a/packages/backend/src/services/org/OrgClickHouseSettingsService.ts +++ b/packages/backend/src/services/org/OrgClickHouseSettingsService.ts @@ -57,7 +57,7 @@ import { FetchHttpClient, HttpClient, HttpClientRequest } from "effect/http" import { decryptAes256Gcm, encryptAes256Gcm, - parseBase64Aes256GcmKey, + parseIngestKeyEncryptionKey, type EncryptedValue, } from "@maple/backend/platform/Crypto" import { SchemaApplyWorkflow } from "@maple/backend/platform/bindings" @@ -418,17 +418,6 @@ const STALE_APPLY_RUN_MS = 30 * 60_000 const toEncryptionError = (message: string) => new OrgClickHouseSettingsEncryptionError({ message }) -const parseEncryptionKey = (raw: string): Effect.Effect => - parseBase64Aes256GcmKey(raw, (message) => - toEncryptionError( - 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, - ), - ) - const encryptToken = ( plaintext: string, encryptionKey: Buffer, @@ -928,7 +917,10 @@ export class OrgClickHouseSettingsService extends Context.Service< const dbExecute = makeDbExecute(database, "OrgClickHouseSettingsService", toPersistenceError) const env = yield* Env const httpClient = yield* HttpClient.HttpClient - const encryptionKey = yield* parseEncryptionKey(Redacted.value(env.MAPLE_INGEST_KEY_ENCRYPTION_KEY)) + const encryptionKey = yield* parseIngestKeyEncryptionKey( + Redacted.value(env.MAPLE_INGEST_KEY_ENCRYPTION_KEY), + toEncryptionError, + ) // Dev-only way past a per-org BYO row; the environment gate keeps it off deploys. const ignoreOrgClickHouse = env.MAPLE_ENVIRONMENT === "development" && env.MAPLE_IGNORE_ORG_CLICKHOUSE // The background schema-apply Workflow, bound only by the api Worker that hosts it. diff --git a/packages/backend/src/services/org/OrgIngestKeysService.ts b/packages/backend/src/services/org/OrgIngestKeysService.ts index 21e2552436..f9052f4d22 100644 --- a/packages/backend/src/services/org/OrgIngestKeysService.ts +++ b/packages/backend/src/services/org/OrgIngestKeysService.ts @@ -33,7 +33,7 @@ import { import { decryptAes256Gcm, encryptAes256Gcm, - parseBase64Aes256GcmKey, + parseIngestKeyEncryptionKey, type EncryptedValue, } from "@maple/backend/platform/Crypto" import { Database } from "@maple/backend/platform/DatabaseLive" @@ -65,17 +65,6 @@ const decodeIsoDateTimeStringSync = Schema.decodeUnknownSync(IsoDateTimeString) // provides `SharedIngestKeysMemo`. const ORG_INGEST_KEYS_MEMO_TTL_MS = 300_000 -const parseEncryptionKey = (raw: string): Effect.Effect => - parseBase64Aes256GcmKey(raw, (message) => - toEncryptionError( - 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, - ), - ) - const parseLookupHmacKey = (raw: string): Effect.Effect => parseIngestKeyLookupHmacKey(raw, toEncryptionError) @@ -108,8 +97,9 @@ export class OrgIngestKeysService extends Context.Service( const ingestKeysMemo = (yield* SharedIngestKeysMemo) ?? (yield* makeExpiringStore()) const env = yield* Env - const encryptionKey = yield* parseEncryptionKey( + const encryptionKey = yield* parseIngestKeyEncryptionKey( Redacted.value(env.MAPLE_INGEST_KEY_ENCRYPTION_KEY), + toEncryptionError, ) const lookupHmacKey = yield* parseLookupHmacKey( Redacted.value(env.MAPLE_INGEST_KEY_LOOKUP_HMAC_KEY),