feat: auto-provision Cloudflare Workers Observability trace/log destinations

apps/api/src/app.ts

@@ -47,6 +47,7 @@ import { ApiAuthorizationLayer } from "./services/ApiAuthorizationLayer"
 import { InternalServiceAuthorizationLayer } from "./services/InternalServiceAuthorizationLayer"
 import { CloudflareAnalyticsService } from "./services/CloudflareAnalyticsService"
 import { CloudflareOAuthService } from "./services/CloudflareOAuthService"
+import { CloudflareObservabilityService } from "./services/CloudflareObservabilityService"
 import { DashboardPersistenceService } from "./services/DashboardPersistenceService"
 import { DemoService } from "./services/DemoService"
 import { DigestService } from "./services/DigestService"
@@ -115,6 +116,10 @@ const CloudflareAnalyticsServiceLive = CloudflareAnalyticsService.layer.pipe(
 	Layer.provideMerge(Layer.mergeAll(CoreServicesLive, WarehouseQueryServiceLive)),
 )
 
+const CloudflareObservabilityServiceLive = CloudflareObservabilityService.layer.pipe(
+	Layer.provideMerge(CoreServicesLive),
+)
+
 const DemoServiceLive = DemoService.layer.pipe(
 	Layer.provideMerge(Layer.mergeAll(CoreServicesLive, WarehouseQueryServiceLive)),
 )
@@ -192,6 +197,7 @@ const VcsServicesLive = Layer.mergeAll(
 export const MainLive = Layer.mergeAll(
 	CoreServicesLive,
 	CloudflareAnalyticsServiceLive,
+	CloudflareObservabilityServiceLive,
 	WarehouseQueryServiceLive,
 	EdgeCacheServiceLive,
 	QueryEngineServiceLive,

apps/api/src/lib/CloudflareApi.ts

@@ -189,6 +189,126 @@ export const listWorkerScripts: (
 	return scripts.flatMap((script) => (script.id == null || script.id === "" ? [] : [script.id]))
 })
 
+export interface CloudflareObservabilityDestinationApi {
+	readonly slug: string
+	readonly name: string
+	readonly dataset: string
+	readonly enabled: boolean
+	readonly url: string
+	readonly destinationConf: string
+	readonly logpushJob: number | null
+	readonly scripts: ReadonlyArray<string>
+	readonly headers: Record<string, unknown>
+	readonly jobStatus: {
+		readonly errorMessage: string
+		readonly lastComplete: string
+		readonly lastError: string
+	} | null
+}
+
+// Workers Observability destinations are account-level resources; a connected account is unlikely
+// to have more than a handful, but cap the stream to avoid a pathological response fan-out.
+const MAX_OBSERVABILITY_DESTINATIONS = 100
+
+/**
+ * List Workers Observability telemetry destinations for the account.
+ */
+export const listObservabilityDestinations: (
+	accessToken: string,
+	accountId: string,
+	apiBaseUrl?: string,
+) => Effect.Effect<ReadonlyArray<CloudflareObservabilityDestinationApi>, CloudflareApiError, never> = Effect.fn(
+	"CloudflareApi.listObservabilityDestinations",
+)(function* (accessToken: string, accountId: string, apiBaseUrl?: string) {
+	yield* Effect.annotateCurrentSpan("maple.cloudflare.account_id", accountId)
+	const destinations = yield* runMapped(
+		accessToken,
+		Workers.listObservabilityDestinations
+			.items({ accountId })
+			.pipe(Stream.take(MAX_OBSERVABILITY_DESTINATIONS), Stream.runCollect),
+		apiBaseUrl,
+	)
+	yield* Effect.annotateCurrentSpan("maple.cloudflare.observability_destination_count", destinations.length)
+	return destinations.map((destination): CloudflareObservabilityDestinationApi => ({
+		slug: destination.slug,
+		name: destination.name,
+		dataset: destination.configuration.logpushDataset,
+		enabled: destination.enabled,
+		url: destination.configuration.url,
+		destinationConf: destination.configuration.destinationConf,
+		logpushJob: null,
+		scripts: destination.scripts,
+		headers: destination.configuration.headers,
+		jobStatus: destination.configuration.jobStatus ?? null,
+	}))
+})
+
+/**
+ * Create a Workers Observability telemetry destination pointing at an OTLP HTTP endpoint.
+ */
+export const createObservabilityDestination: (
+	accessToken: string,
+	accountId: string,
+	name: string,
+	dataset: string,
+	url: string,
+	headers: Record<string, string>,
+	apiBaseUrl?: string,
+) => Effect.Effect<CloudflareObservabilityDestinationApi, CloudflareApiError, never> = Effect.fn(
+	"CloudflareApi.createObservabilityDestination",
+)(function* (
+	accessToken: string,
+	accountId: string,
+	name: string,
+	dataset: string,
+	url: string,
+	headers: Record<string, string>,
+	apiBaseUrl?: string,
+) {
+	yield* Effect.annotateCurrentSpan("maple.cloudflare.account_id", accountId)
+	const destination = yield* runMapped(
+		accessToken,
+		Workers.createObservabilityDestination({
+			accountId,
+			configuration: { headers, logpushDataset: dataset, type: "logpush", url },
+			enabled: true,
+			name,
+		}),
+		apiBaseUrl,
+	)
+	return {
+		slug: destination.slug,
+		name: destination.name,
+		dataset: destination.configuration.logpushDataset,
+		enabled: destination.enabled,
+		url: destination.configuration.url,
+		destinationConf: destination.configuration.destinationConf,
+		logpushJob: destination.configuration.logpushJob,
+		scripts: destination.scripts,
+		headers, // caller-supplied; the response does not round-trip headers.
+		jobStatus: null,
+	} satisfies CloudflareObservabilityDestinationApi
+})
+
+/**
+ * Delete a Workers Observability telemetry destination by slug.
+ */
+export const deleteObservabilityDestination: (
+	accessToken: string,
+	accountId: string,
+	slug: string,
+	apiBaseUrl?: string,
+) => Effect.Effect<void, CloudflareApiError, never> = Effect.fn("CloudflareApi.deleteObservabilityDestination")(
+	function* (accessToken: string, accountId: string, slug: string, apiBaseUrl?: string) {
+		yield* Effect.annotateCurrentSpan("maple.cloudflare.account_id", accountId)
+		yield* runMapped(
+			accessToken,
+			Workers.deleteObservabilityDestination({ accountId, slug }),
+			apiBaseUrl,
+		)
+	},
+)
+
 // ---------------------------------------------------------------------------
 // GraphQL Analytics (the raw escape hatch the module doc-comment anticipates)
 // ---------------------------------------------------------------------------

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

@@ -2,6 +2,7 @@ import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstab
 import { HttpApiBuilder } from "effect/unstable/httpapi"
 import {
 	CloudflareDisconnectResponse,
+	CloudflareIntegrationStatus,
 	CloudflareStartConnectResponse,
 	CloudflareTopTrafficResponse,
 	CloudflareTopTrafficRow,
@@ -35,6 +36,7 @@ import { Env } from "../lib/Env"
 import { graphqlQuery } from "../lib/CloudflareApi"
 import { CloudflareAnalyticsService } from "../services/CloudflareAnalyticsService"
 import { CloudflareOAuthService } from "../services/CloudflareOAuthService"
+import { CloudflareObservabilityService } from "../services/CloudflareObservabilityService"
 import { abrCount } from "../services/cloudflare-analytics/mapping"
 import {
 	decodeTopTrafficResponse,
@@ -50,6 +52,9 @@ import { requireAdmin as requireAdminRole } from "../lib/auth"
 
 const asExternalUserId = Schema.decodeUnknownSync(ExternalUserId)
 const asUserId = Schema.decodeUnknownSync(UserId)
+const CLOUDFLARE_OBSERVABILITY_SYSTEM_USER_ID = Schema.decodeUnknownSync(UserId)(
+	"system-cloudflare-observability",
+)
 
 const HAZEL_CALLBACK_PATH = "/api/integrations/hazel/callback"
 const GITHUB_CALLBACK_PATH = "/api/integrations/github/callback"
@@ -97,6 +102,7 @@ export const HttpIntegrationsLive = HttpApiBuilder.group(MapleApi, "integrations
 		const vcsCommits = yield* VcsCommitService
 		const cloudflare = yield* CloudflareOAuthService
 		const cloudflareAnalytics = yield* CloudflareAnalyticsService
+		const cloudflareObservability = yield* CloudflareObservabilityService
 		const database = yield* Database
 		const edgeCache = yield* EdgeCacheService
 		const env = yield* Env
@@ -176,7 +182,12 @@ export const HttpIntegrationsLive = HttpApiBuilder.group(MapleApi, "integrations
 				.handle("cloudflareStatus", () =>
 					Effect.gen(function* () {
 						const tenant = yield* CurrentTenant.Context
-						return yield* cloudflareAnalytics.getIntegrationStatus(tenant.orgId)
+						const base = yield* cloudflareAnalytics.getIntegrationStatus(tenant.orgId)
+						const destinations = yield* cloudflareObservability.getDestinations(tenant.orgId)
+						return new CloudflareIntegrationStatus({
+							...base,
+							observabilityDestinations: destinations,
+						})
 					}),
 				)
 				.handle("cloudflareUsage", () =>
@@ -346,6 +357,8 @@ export const HttpIntegrationsLive = HttpApiBuilder.group(MapleApi, "integrations
 					Effect.gen(function* () {
 						const tenant = yield* CurrentTenant.Context
 						yield* requireAdmin(tenant.roles)
+						// Clean up Cloudflare-side observability destinations while the OAuth token is still valid.
+						yield* cloudflareObservability.deleteDestinations(tenant.orgId)
 						const result = yield* cloudflare.disconnect(tenant.orgId)
 						return new CloudflareDisconnectResponse(result)
 					}),
@@ -603,6 +616,7 @@ export const IntegrationsCallbackRouter = HttpRouter.use((router) =>
 		const github = yield* GithubConnectService
 		const cloudflare = yield* CloudflareOAuthService
 		const cloudflareAnalytics = yield* CloudflareAnalyticsService
+		const cloudflareObservability = yield* CloudflareObservabilityService
 		const env = yield* Env
 
 		const dashboardTargetOrigin = resolveDashboardTargetOrigin(env.MAPLE_APP_BASE_URL)
@@ -896,6 +910,14 @@ export const IntegrationsCallbackRouter = HttpRouter.use((router) =>
 							),
 						),
 					),
+					// Auto-provision trace/log Workers Observability destinations while the user is still
+					// on the callback page so the dashboard card is immediately up to date.
+					Effect.tap((result) =>
+						cloudflareObservability.ensureDestinations(
+							result.orgId,
+							CLOUDFLARE_OBSERVABILITY_SYSTEM_USER_ID,
+						),
+					),
 					Effect.map((result) =>
 						htmlResponse(
 							cloudflareCallbackPage({

apps/api/src/services/CloudflareAnalyticsService.ts

@@ -72,6 +72,7 @@ import { Env } from "../lib/Env"
 import { dateToMs } from "../lib/time"
 import { WarehouseQueryService } from "../lib/WarehouseQueryService"
 import { CloudflareOAuthService } from "./CloudflareOAuthService"
+import { hasObservabilityScopes } from "./CloudflareObservabilityService"
 import { OrgClickHouseSettingsService } from "./OrgClickHouseSettingsService"
 import { OrgIngestKeysService } from "./OrgIngestKeysService"
 import {
@@ -2044,8 +2045,10 @@ export class CloudflareAnalyticsService extends Context.Service<
 					connectedByUserId: null,
 					scope: null,
 					analyticsCapable: false,
+					observabilityCapable: false,
 					zones: [],
 					workers: null,
+					observabilityDestinations: [],
 				})
 			}
 			const analytics = yield* getStatus(orgId)
@@ -2056,8 +2059,10 @@ export class CloudflareAnalyticsService extends Context.Service<
 				connectedByUserId: decodeUserIdSync(connection.connectedByUserId),
 				scope: connection.scope,
 				analyticsCapable: hasAnalyticsScopes(connection.scope),
+				observabilityCapable: hasObservabilityScopes(connection.scope),
 				zones: analytics.zones.map((zone) => new CloudflareAnalyticsZoneStatus(zone)),
 				workers: analytics.workers ? new CloudflareAnalyticsWorkersStatus(analytics.workers) : null,
+				observabilityDestinations: [],
 			})
 		})