feat(ga4): Google Analytics integration with delta reconciliation

apps/alerting/src/scheduled.test.ts

@@ -5,7 +5,7 @@ import { buildLayer, catchTickFailure, selectScheduledProgram, type ScheduledTic
 
 const cronCases = [
 	["*/5 * * * *", ["anomaly", "cloudflareAnalytics", "planetScale"]],
-	["*/15 * * * *", ["digest"]],
+	["*/15 * * * *", ["digest", "googleAnalytics"]],
 	["0 * * * *", ["serviceMapRollup"]],
 	["* * * * *", ["alert", "error", "escalation", "fixVerification"]],
 ] as const
@@ -27,6 +27,7 @@ describe("alerting Effect root", () => {
 				error: tick("error"),
 				escalation: tick("escalation"),
 				fixVerification: tick("fixVerification"),
+				googleAnalytics: tick("googleAnalytics"),
 				planetScale: tick("planetScale"),
 				serviceMapRollup: tick("serviceMapRollup"),
 			} satisfies ScheduledTickPrograms
@@ -56,6 +57,7 @@ describe("alerting Effect root", () => {
 				error: errorGate.await.pipe(Effect.andThen(record("error"))),
 				escalation: record("escalation"),
 				fixVerification: record("fixVerification"),
+				googleAnalytics: record("googleAnalytics"),
 				planetScale: record("planetScale"),
 				serviceMapRollup: record("serviceMapRollup"),
 			} satisfies ScheduledTickPrograms
@@ -81,6 +83,7 @@ describe("alerting Effect root", () => {
 			error: tick("error"),
 			escalation: tick("escalation"),
 			fixVerification: tick("fixVerification"),
+			googleAnalytics: tick("googleAnalytics"),
 			planetScale: tick("planetScale"),
 			serviceMapRollup: tick("serviceMapRollup"),
 		} satisfies ScheduledTickPrograms

apps/alerting/src/scheduled.ts

@@ -27,6 +27,8 @@ import {
 	ErrorsService,
 	EscalationService,
 	FixVerificationTickService,
+	GoogleAnalyticsOAuthService,
+	GoogleAnalyticsService,
 	HazelOAuthService,
 	layerPg,
 	NotificationDispatcher,
@@ -236,6 +238,12 @@ export const buildLayer = (env: AlertingWorkerEnv) => {
 		),
 	)
 
+	const GoogleAnalyticsOAuthServiceLive = GoogleAnalyticsOAuthService.layer.pipe(Layer.provide(BaseLive))
+
+	const GoogleAnalyticsServiceLive = GoogleAnalyticsService.layer.pipe(
+		Layer.provide(Layer.mergeAll(BaseLive, GoogleAnalyticsOAuthServiceLive, OrgIngestKeysServiceLive)),
+	)
+
 	const PlanetScaleOAuthServiceLive = PlanetScaleOAuthService.layer.pipe(Layer.provide(BaseLive))
 
 	const PlanetScaleServiceLive = PlanetScaleService.layer.pipe(
@@ -246,6 +254,7 @@ export const buildLayer = (env: AlertingWorkerEnv) => {
 		AlertsServiceLive,
 		AnomalyDetectionServiceLive,
 		CloudflareAnalyticsServiceLive,
+		GoogleAnalyticsServiceLive,
 		PlanetScaleServiceLive,
 		DigestServiceLive,
 		ErrorsServiceLive,
@@ -420,6 +429,21 @@ const cloudflareAnalyticsTick = makeTick(
 	}),
 )
 
+/**
+ * Runs on the 15-minute cron, not Cloudflare's 5-minute one. GA4 does not update fast enough to
+ * reward a tighter cadence, and every tick spends Data API quota tokens per property.
+ */
+const googleAnalyticsTick = makeTick(
+	GoogleAnalyticsService.use((analytics) => analytics.pollAllOrgs()),
+	"google_analytics",
+	(result) => ({
+		properties: result.properties,
+		rowsIngested: result.rowsIngested,
+		skipped: result.skipped,
+		failures: result.failures,
+	}),
+)
+
 const planetScaleTick = makeTick(
 	PlanetScaleService.use((planetscale) => planetscale.pollAllOrgs()),
 	"planetscale",
@@ -443,6 +467,7 @@ export interface ScheduledTickPrograms<R = never> {
 	readonly error: Effect.Effect<void, never, R>
 	readonly escalation: Effect.Effect<void, never, R>
 	readonly fixVerification: Effect.Effect<void, never, R>
+	readonly googleAnalytics: Effect.Effect<void, never, R>
 	readonly planetScale: Effect.Effect<void, never, R>
 	readonly serviceMapRollup: Effect.Effect<void, never, R>
 }
@@ -463,7 +488,9 @@ export const selectScheduledProgram = <R>(
 				discard: true,
 			}),
 		),
-		Match.when("*/15 * * * *", () => ticks.digest),
+		Match.when("*/15 * * * *", () =>
+			Effect.all([ticks.digest, ticks.googleAnalytics], { concurrency: 2, discard: true }),
+		),
 		Match.when("0 * * * *", () => ticks.serviceMapRollup),
 		Match.when("* * * * *", () =>
 			// `fixVerification` is chained onto `error` rather than listed beside it:
@@ -493,6 +520,7 @@ type ScheduledServices =
 	| ErrorsService
 	| EscalationService
 	| FixVerificationTickService
+	| GoogleAnalyticsService
 	| PlanetScaleService
 	| ServiceMapRollupService
 
@@ -504,6 +532,7 @@ export const scheduledTicks: ScheduledTickPrograms<ScheduledServices> = {
 	error: errorTick,
 	escalation: escalationTick,
 	fixVerification: fixVerificationTick,
+	googleAnalytics: googleAnalyticsTick,
 	planetScale: planetScaleTick,
 	serviceMapRollup: serviceMapRollupTick,
 }

apps/alerting/src/worker.ts

@@ -26,6 +26,7 @@ import {
 	appUrlsEnv,
 	authEnv,
 	cloudflareOAuthEnv,
+	googleAnalyticsOAuthEnv,
 	ingestKeyCryptoEnv,
 	merge,
 	optionalPlain,
@@ -105,12 +106,13 @@ const configuredEnv = (stage: MapleStage) =>
 		optionalSecret("AUTUMN_SECRET_KEY"),
 		optionalSecret("INTERNAL_SERVICE_TOKEN"),
 		// The alerting worker is where incidents open and resolve, so it is the one
-		// that sends push (platform/Apns.ts) — and it runs the Cloudflare analytics
-		// and PlanetScale inventory pollers, each of which resolves and refreshes
-		// per-org OAuth tokens with the same config the api worker uses.
+		// that sends push (platform/Apns.ts) — and it runs the Cloudflare analytics,
+		// PlanetScale inventory and Google Analytics pollers, each of which resolves
+		// and refreshes per-org OAuth tokens with the same config the api worker uses.
 		apnsEnv,
 		cloudflareOAuthEnv,
 		planetScaleOAuthEnv,
+		googleAnalyticsOAuthEnv,
 	)
 
 /**

apps/api/src/alerting.ts

@@ -9,6 +9,8 @@ export { BucketCacheService } from "@maple/query-engine/caching"
 export { CacheBackendLive } from "@/platform/CacheBackendLive"
 export { CloudflareAnalyticsService } from "./services/integrations/CloudflareAnalyticsService"
 export { CloudflareOAuthService } from "./services/auth/CloudflareOAuthService"
+export { GoogleAnalyticsService } from "./services/integrations/GoogleAnalyticsService"
+export { GoogleAnalyticsOAuthService } from "./services/auth/GoogleAnalyticsOAuthService"
 export { ErrorActorsService } from "./services/errors/ErrorActorsService"
 export { ErrorIssueReadModelsService } from "./services/errors/ErrorIssueReadModelsService"
 export { ErrorIssueWorkflowService } from "./services/errors/ErrorIssueWorkflowService"

apps/api/src/dashboard-templates/index.ts

@@ -23,6 +23,7 @@ import { kubernetesPodTemplate } from "./infrastructure/kubernetes-pod"
 import { kafkaTemplate } from "./messaging/kafka"
 import { natsTemplate } from "./messaging/nats"
 import { rabbitmqTemplate } from "./messaging/rabbitmq"
+import { googleAnalyticsTemplate } from "./product/google-analytics"
 import type { TemplateDefinition, TemplateMetadata, TemplatePreviewWidget } from "./types"
 
 export const DASHBOARD_TEMPLATES: ReadonlyArray<TemplateDefinition> = [
@@ -53,6 +54,8 @@ export const DASHBOARD_TEMPLATES: ReadonlyArray<TemplateDefinition> = [
 	kafkaTemplate,
 	natsTemplate,
 	rabbitmqTemplate,
+	// Product
+	googleAnalyticsTemplate,
 ]
 
 const TEMPLATE_BY_ID = new Map<string, TemplateDefinition>(DASHBOARD_TEMPLATES.map((t) => [t.id, t]))