feat(ga4): Google Analytics integration with delta reconciliation

apps/ai/src/runtime/graph-boundaries.test.ts

@@ -44,6 +44,7 @@ describe("AI runtime graph boundaries", () => {
 		for (const routeOnlyService of [
 			"DailySpendService",
 			"CloudflareAnalyticsService",
+			"GoogleAnalyticsService",
 			"AnomalyDetectionService",
 			"AiTriageService",
 			"DigestService",

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

@@ -14,6 +14,7 @@ import { Env } from "@maple/backend/platform/Env"
 import { ErrorsService } from "@maple/backend/services/errors/ErrorsService"
 import { EscalationService } from "@maple/backend/services/alerts/EscalationService"
 import { FixVerificationTickService } from "@maple/backend/services/errors/FixVerificationTickService"
+import { GoogleAnalyticsService } from "@maple/backend/services/integrations/GoogleAnalyticsService"
 import { layerPg } from "@maple/backend/platform/DatabasePgLive"
 import { PullRequestLookupLive } from "@maple/backend/services/errors/pull-request-lookup-live"
 import { PlanetScaleService } from "@maple/backend/services/integrations/PlanetScaleService"
@@ -37,6 +38,7 @@ export const buildLayer = (env: AlertingWorkerEnv) =>
 		AlertsService.layer,
 		AnomalyDetectionService.layer,
 		CloudflareAnalyticsService.layer,
+		GoogleAnalyticsService.layer,
 		PlanetScaleService.layer,
 		DigestService.layer,
 		ErrorsService.layer,
@@ -210,6 +212,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",
@@ -233,6 +250,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>
 }
@@ -253,7 +271,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:
@@ -283,6 +303,7 @@ type ScheduledServices =
 	| ErrorsService
 	| EscalationService
 	| FixVerificationTickService
+	| GoogleAnalyticsService
 	| PlanetScaleService
 	| ServiceMapRollupService
 
@@ -294,6 +315,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,
@@ -94,12 +95,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/resources/env.ts

@@ -17,6 +17,7 @@ import {
 	authEnv,
 	cloudflareOAuthEnv,
 	derived,
+	googleAnalyticsOAuthEnv,
 	ingestKeyCryptoEnv,
 	merge,
 	optionalPlain,
@@ -107,4 +108,5 @@ export const apiConfiguredEnv = (stage: MapleStage, domains: MapleDomains) =>
 		optionalPlain("GITHUB_API_BASE_URL"),
 		cloudflareOAuthEnv,
 		planetScaleOAuthEnv,
+		googleAnalyticsOAuthEnv,
 	)

apps/api/src/routes/v1/integrations.http.ts

@@ -54,6 +54,10 @@ import {
 } from "@maple/backend/services/integrations/cloudflare-analytics/queries"
 import { PlanetScaleConnectionService } from "@maple/backend/services/integrations/PlanetScaleConnectionService"
 import { PlanetScaleService } from "@maple/backend/services/integrations/PlanetScaleService"
+import {
+	GOOGLE_ANALYTICS_CALLBACK_PATH,
+	GoogleAnalyticsOAuthService,
+} from "@maple/backend/services/auth/GoogleAnalyticsOAuthService"
 import {
 	PLANETSCALE_CALLBACK_PATH,
 	PlanetScaleOAuthService,
@@ -75,6 +79,7 @@ const HAZEL_MESSAGE_TYPE = "maple:integration:hazel"
 const GITHUB_MESSAGE_TYPE = "maple:integration:github"
 const CLOUDFLARE_MESSAGE_TYPE = "maple:integration:cloudflare"
 const PLANETSCALE_MESSAGE_TYPE = "maple:integration:planetscale"
+const GOOGLE_ANALYTICS_MESSAGE_TYPE = "maple:integration:google-analytics"
 
 /**
  * How long `cloudflarePrime` spends on the post-connect poll. Long enough for zone discovery plus
@@ -826,6 +831,7 @@ export const IntegrationsCallbackRouter = HttpRouter.use((router) =>
 		const cloudflareAnalytics = yield* CloudflareAnalyticsService
 		const planetscaleOAuth = yield* PlanetScaleOAuthService
 		const planetscaleConnection = yield* PlanetScaleConnectionService
+		const googleAnalyticsOAuth = yield* GoogleAnalyticsOAuthService
 		const env = yield* Env
 
 		const dashboardTargetOrigin = resolveDashboardTargetOrigin(env.MAPLE_APP_BASE_URL)
@@ -850,6 +856,14 @@ export const IntegrationsCallbackRouter = HttpRouter.use((router) =>
 				messageType: CLOUDFLARE_MESSAGE_TYPE,
 				label: "Cloudflare",
 			})
+		const googleAnalyticsCallbackPage = (params: Omit<CallbackPageParams, "targetOrigin">) =>
+			renderCallbackPage({
+				...params,
+				targetOrigin: dashboardTargetOrigin,
+				messageType: GOOGLE_ANALYTICS_MESSAGE_TYPE,
+				label: "Google Analytics",
+			})
+
 		const planetscaleCallbackPage = (params: Omit<CallbackPageParams, "targetOrigin">) =>
 			renderCallbackPage({
 				...params,
@@ -1271,5 +1285,83 @@ export const IntegrationsCallbackRouter = HttpRouter.use((router) =>
 		})
 
 		yield* router.add("GET", PLANETSCALE_CALLBACK_PATH, handlePlanetScale)
+
+		const googleAnalyticsErrorPage = (message: string) =>
+			htmlResponse(googleAnalyticsCallbackPage({ status: "error", message, returnTo: null }), 400)
+
+		const handleGoogleAnalytics = Effect.fn("integrations.googleAnalyticsOAuthCallback")(function* (
+			req: HttpServerRequest.HttpServerRequest,
+		) {
+			const urlOption = Option.liftThrowable(() => new URL(req.url, "http://localhost"))()
+			if (Option.isNone(urlOption)) {
+				return googleAnalyticsErrorPage("Malformed callback URL")
+			}
+			const url = urlOption.value
+			const code = url.searchParams.get("code")
+			const state = url.searchParams.get("state")
+			const oauthError = url.searchParams.get("error")
+
+			if (oauthError) {
+				// Google's own codes are terse; `access_denied` is the one users actually hit,
+				// by closing the consent screen.
+				return googleAnalyticsErrorPage(
+					oauthError === "access_denied"
+						? "Google sign-in was cancelled — the connection wasn't authorized."
+						: `Google returned an error (${oauthError})`,
+				)
+			}
+
+			if (!code || !state) {
+				return googleAnalyticsErrorPage("Missing code or state in callback")
+			}
+
+			return yield* googleAnalyticsOAuth.completeConnect(code, state).pipe(
+				// The first collection is NOT run here. It takes tens of seconds on a grant with
+				// several properties, and the popup would sit blank for all of it; the dashboard
+				// calls `prime` from the tab that stays open instead.
+				// Collector state is deliberately preserved across a reconnect — the ledger is what
+				// stops the restatement window's hours being emitted twice. See the note above
+				// `GoogleAnalyticsService`.
+				// The callback page reduces failures to short human copy — make sure the real
+				// cause still lands in the server log for diagnosis.
+				Effect.tapError((error) =>
+					Effect.logError("Google Analytics OAuth completeConnect failed", {
+						tag: error._tag,
+						message: error.message,
+					}),
+				),
+				Effect.map((result) =>
+					htmlResponse(
+						googleAnalyticsCallbackPage({
+							status: "success",
+							message:
+								"Google Analytics connected. You can close this window and return to Maple.",
+							returnTo: result.returnTo,
+						}),
+					),
+				),
+				Effect.catchTags({
+					// Validation/upstream messages are our own sanitized strings — and for this
+					// provider they carry the two refusals a user can actually act on: a grant
+					// with no refresh token, and one that reaches no GA4 property.
+					"@maple/http/errors/IntegrationsValidationError": (error) =>
+						Effect.succeed(googleAnalyticsErrorPage(error.message)),
+					"@maple/http/errors/IntegrationsUpstreamError": (error) =>
+						Effect.succeed(googleAnalyticsErrorPage(error.message)),
+					"@maple/http/errors/IntegrationsRevokedError": () =>
+						Effect.succeed(
+							googleAnalyticsErrorPage(
+								"Google rejected the authorization — reconnect and try again",
+							),
+						),
+					"@maple/http/errors/IntegrationsPersistenceError": () =>
+						Effect.succeed(
+							googleAnalyticsErrorPage("Failed to complete Google Analytics connection"),
+						),
+				}),
+			)
+		})
+
+		yield* router.add("GET", GOOGLE_ANALYTICS_CALLBACK_PATH, handleGoogleAnalytics)
 	}),
 )

apps/api/src/routes/v2/alchemy-provider.integration.test.ts

@@ -56,6 +56,7 @@ import {
 	ApiV2RateLimiterAllowAllLayer,
 	ConfigResourceServiceStubsLayer,
 	makeWarehouseServiceStub,
+	GoogleAnalyticsServiceStubsLayer,
 	PlanetScaleServiceStubsLayer,
 	SlackIntegrationServiceStubLayer,
 	TelemetryServiceStubsLayer,
@@ -190,6 +191,7 @@ const makeHarness = () => {
 		Layer.provide(V2TransportErrorBoundaryLive),
 		Layer.provide(SlackIntegrationServiceStubLayer),
 		Layer.provide(PlanetScaleServiceStubsLayer),
+		Layer.provide(GoogleAnalyticsServiceStubsLayer),
 		Layer.provideMerge(ApiAuthorizationV2Layer),
 		Layer.provideMerge(AuditLogService.layerMemory),
 		Layer.provideMerge(ApiV2RateLimiterAllowAllLayer),

apps/api/src/routes/v2/alerts.http.test.ts

@@ -42,6 +42,7 @@ import {
 	ApiV2RateLimiterAllowAllLayer,
 	ConfigResourceServiceStubsLayer,
 	makeWarehouseServiceStub,
+	GoogleAnalyticsServiceStubsLayer,
 	PlanetScaleServiceStubsLayer,
 	SlackIntegrationServiceStubLayer,
 	TelemetryServiceStubsLayer,
@@ -178,6 +179,7 @@ const makeHarness = (
 		Layer.provide(V2TransportErrorBoundaryLive),
 		Layer.provide(SlackIntegrationServiceStubLayer),
 		Layer.provide(PlanetScaleServiceStubsLayer),
+		Layer.provide(GoogleAnalyticsServiceStubsLayer),
 		Layer.provideMerge(ApiAuthorizationV2Layer),
 		Layer.provideMerge(AuditLogService.layerMemory),
 		Layer.provideMerge(ApiV2RateLimiterAllowAllLayer),

apps/api/src/routes/v2/api-keys.http.test.ts

@@ -19,6 +19,7 @@ import {
 	AlertsServiceStubLayer,
 	AllV2GroupLayersLive,
 	ConfigResourceServiceStubsLayer,
+	GoogleAnalyticsServiceStubsLayer,
 	PlanetScaleServiceStubsLayer,
 	SlackIntegrationServiceStubLayer,
 	TelemetryServiceStubsLayer,
@@ -64,6 +65,7 @@ const makeHarness = (checkRateLimit: RateLimiterApi["check"] = () => Effect.succ
 		Layer.provide(V2TransportErrorBoundaryLive),
 		Layer.provide(SlackIntegrationServiceStubLayer),
 		Layer.provide(PlanetScaleServiceStubsLayer),
+		Layer.provide(GoogleAnalyticsServiceStubsLayer),
 		Layer.provide(AlertsServiceStubLayer),
 		Layer.provide(ConfigResourceServiceStubsLayer),
 		Layer.provide(TelemetryServiceStubsLayer),

apps/api/src/routes/v2/config-resources.http.test.ts

@@ -27,6 +27,7 @@ import {
 	ApiV2RateLimiterAllowAllLayer,
 	makeWarehouseServiceStub,
 	Phase1ResourceStubsLayer,
+	GoogleAnalyticsServiceStubsLayer,
 	PlanetScaleServiceStubsLayer,
 	SlackIntegrationServiceStubLayer,
 	SetupAuditServiceStubLayer,
@@ -114,6 +115,7 @@ const makeHarness = () => {
 		Layer.provide(V2TransportErrorBoundaryLive),
 		Layer.provide(SlackIntegrationServiceStubLayer),
 		Layer.provide(PlanetScaleServiceStubsLayer),
+		Layer.provide(GoogleAnalyticsServiceStubsLayer),
 		Layer.provide(AlertsServiceStubLayer),
 		Layer.provide(Phase1ResourceStubsLayer),
 		Layer.provide(SetupAuditServiceStubLayer),

apps/api/src/routes/v2/dashboards.http.test.ts

@@ -24,6 +24,7 @@ import {
 	AllV2GroupLayersLive,
 	ApiV2RateLimiterAllowAllLayer,
 	ConfigResourceServiceStubsLayer,
+	GoogleAnalyticsServiceStubsLayer,
 	PlanetScaleServiceStubsLayer,
 	SlackIntegrationServiceStubLayer,
 	TelemetryServiceStubsLayer,
@@ -68,6 +69,7 @@ const makeHarness = () => {
 		Layer.provide(V2TransportErrorBoundaryLive),
 		Layer.provide(SlackIntegrationServiceStubLayer),
 		Layer.provide(PlanetScaleServiceStubsLayer),
+		Layer.provide(GoogleAnalyticsServiceStubsLayer),
 		Layer.provide(AlertsServiceStubLayer),
 		Layer.provide(ConfigResourceServiceStubsLayer),
 		Layer.provide(TelemetryServiceStubsLayer),

apps/api/src/routes/v2/integrations.http.test.ts

@@ -49,6 +49,7 @@ import {
 	AllV2GroupLayersLive,
 	ApiV2RateLimiterAllowAllLayer,
 	ConfigResourceServiceStubsLayer,
+	GoogleAnalyticsServiceStubsLayer,
 	TelemetryServiceStubsLayer,
 } from "./v2-test-support"
 
@@ -170,6 +171,7 @@ const makeHarness = (slack: Partial<SlackIntegrationServiceApi> = {}, planetscal
 		Layer.provide(V2TransportErrorBoundaryLive),
 		Layer.provide(slackServiceLayer(slack)),
 		Layer.provide(planetscaleServiceLayer(planetscale)),
+		Layer.provide(GoogleAnalyticsServiceStubsLayer),
 		Layer.provide(AlertsServiceStubLayer),
 		Layer.provide(ConfigResourceServiceStubsLayer),
 		Layer.provide(TelemetryServiceStubsLayer),