import { existsSync } from "node:fs"
import { resolve } from "node:path"
import * as AWS from "@alchemy.run/aws"
import * as Output from "alchemy/Output"
import * as Effect from "effect/Effect"
import { ecsSecrets, pgUrlRequireSsl, resolveAwsRegion } from "@maple/infra/aws"
import { cloudflareIpv4Ranges, issueRegionalCertificate, publishProxiedCname } from "@maple/infra/acm"
import type { MapleDbLogin, MapleRegion, MapleStackContext, MapleStage } from "@maple/infra/cloudflare"
import {
	REGION_INFRA,
	ReplayBlobs,
	resolveDeploymentEnvironment,
	resolveWorkerName,
} from "@maple/infra/cloudflare"
import { r2BucketCredentials } from "@maple/infra/r2-credentials"
// Only the primitives: these values feed ECS `env:` and Secrets Manager, not Worker bindings.
import { commitShaEnv, optionalPlain, requiredPlain } from "@maple/infra/env"
import { createIngestCollector } from "./alchemy/collector.ts"
import { blockedIngestKeysEnv, blockIngestKeysAtEdge } from "./alchemy/edge-block.ts"
import { gatewayTuningEnv } from "./alchemy/gateway-env.ts"

/** Compiled ahead of time by CI so the image build is a COPY. Absent on a dev machine. */
const PREBUILT_BINARY = "apps/ingest/dist/maple-ingest"
/**
 * Absolute on purpose: alchemy has changed how a relative `dockerfile` resolves
 * between releases, and each change broke the deploy (docs/infra.md).
 */
const PREBUILT_DOCKERFILE = resolve("apps/ingest/Dockerfile.prebuilt")

/** Port the gateway binds (`apps/ingest/Dockerfile` EXPOSEs the same). */
const INGEST_PORT = 3474
/** The gateway's OTLP/gRPC server (`INGEST_OTLP_GRPC_PORT`), reached through the ALB. */
const INGEST_GRPC_PORT = 4317
/** OTLP/gRPC export paths: `/opentelemetry.proto.collector.<signal>.v1.<Signal>Service/Export`. */
const OTLP_GRPC_PATH = "/opentelemetry.proto.collector.*"

const GIB = 1024 * 1024 * 1024

/**
 * Where each host type keeps the WAL, and its cap. c7gd hosts mount their NVMe instance
 * store (59 GB on a medium); a t4g has none, so the WAL shares the 30 GB root volume.
 * The per-lane budget is `INGEST_QUEUE_MAX_BYTES / (WAL_SHARDS * lanes)`.
 */
const WAL_STORAGE = {
	"c7gd.large": { disk: "instance-store", maxBytes: 48 * GIB },
	"c7gd.medium": { disk: "instance-store", maxBytes: 48 * GIB },
	"t4g.small": { disk: "root-volume", maxBytes: 4 * GIB },
} as const satisfies Record<
	MapleStackContext["profile"]["ingest"]["instanceType"],
	{ disk: "instance-store" | "root-volume"; maxBytes: number }
>

/**
 * Pinned rather than derived: the gateway defaults to `num_cpus * 2`, which
 * would reshape the on-disk WAL whenever the task size changes.
 */
const WAL_SHARDS = 4

/** Where the gateway keeps its WAL inside the container (the binary's default `INGEST_QUEUE_DIR`). */
const WAL_CONTAINER_DIR = "/var/lib/maple-ingest/wal"

/** The host's instance-store NVMe (mounted by `ec2UserData`) or a root-volume directory. */
const WAL_HOST_DIR = "/mnt/wal"

/**
 * Boot script for a gateway host (ECS-optimized AL2023). An instance-store host mounts the
 * NVMe at WAL_HOST_DIR and only THEN joins the cluster, so a host whose disk did not come
 * up never gets a task. The S3 tier covers the disk being wiped on replace.
 */
const ec2UserData = (clusterName: string, disk: "instance-store" | "root-volume") => `#!/bin/bash
set -euxo pipefail

mkdir -p ${WAL_HOST_DIR}
${
	disk === "instance-store"
		? `disk=$(ls /dev/disk/by-id/nvme-Amazon_EC2_NVMe_Instance_Storage_* | grep -v -- -part | head -n1)
mkfs.xfs -f "$disk"
echo "UUID=$(blkid -s UUID -o value "$disk") ${WAL_HOST_DIR} xfs noatime,nofail 0 2" >> /etc/fstab
mount ${WAL_HOST_DIR}`
		: ""
}

cat >> /etc/ecs/ecs.config <<'CONFIG'
ECS_CLUSTER=${clusterName}
ECS_ENABLE_TASK_IAM_ROLE_NETWORK_HOST=true
ECS_CONTAINER_STOP_TIMEOUT=120s
CONFIG
`

export interface CreateMapleIngestOptions extends Pick<MapleStackContext, "stage" | "region" | "profile"> {
	/** The public hostname (`domains.ingest`). Every stage that deploys a fleet has one. */
	hostname: string
	/** The declared database's gateway login; a stage without one reads `MAPLE_INGEST_PG_URL` instead. */
	dbLogin?: MapleDbLogin
}

/**
 * The gateway's write credentials for the replay payload store
 * (`ReplayBlobs`), or `undefined` on a stage that
 * keeps payloads inline (`profile.deploys.replayBlobs`).
 */
const replayBlobWriterCredentials = (stage: MapleStage, region: MapleRegion, enabled: boolean) =>
	Effect.gen(function* () {
		if (!enabled) return undefined
		// Yielded so the token is ordered behind the bucket.
		yield* ReplayBlobs
		const bucket = resolveWorkerName("replay-blobs", stage, region)
		const credentials = yield* r2BucketCredentials({
			id: "replay-blobs-writer",
			tokenName: `${bucket}-writer`,
			bucketName: bucket,
			jurisdiction: REGION_INFRA[region].storageJurisdiction,
			permissions: ["Workers R2 Storage Bucket Item Write"],
		})
		return { ...credentials, bucket }
	})

/**
 * The Rust OTLP gateway (`apps/ingest`) on ECS, one fleet per `MapleRegion`:
 * ARM64 EC2 hosts with the WAL on local NVMe, behind a public ALB, plus an OTel
 * collector for the gateway's own telemetry (`./alchemy/collector.ts`).
 */
export const createMapleIngest = ({ stage, region, hostname, profile, dbLogin }: CreateMapleIngestOptions) =>
	Effect.gen(function* () {
		const replayBlobs = yield* replayBlobWriterCredentials(stage, region, profile.deploys.replayBlobs)
		const { desiredCount, scaling, instanceType, taskSize, collectorTaskSize, selfTraceSampleRatio } =
			profile.ingest
		const name = (base: string) => resolveWorkerName(base, stage, region)
		const tags = { Service: "maple-ingest", Region: region }

		// Public subnets and NO NAT gateway: NAT's per-GB processing fee would
		// outweigh compute and egress combined for a service that pushes telemetry out.
		const network = yield* AWS.EC2.Network("ingest-network", {
			cidrBlock: REGION_INFRA[region].ingestCidrBlock,
			availabilityZones: 2,
			nat: "none",
			gatewayEndpoints: ["s3"],
			tags,
		})

		// The ALB terminates TLS on 443 behind Cloudflare's proxy and admits only Cloudflare's
		// edge: that is what makes `Cf-IPCountry` trustworthy. The group's `description` must not
		// change: AWS treats it as immutable (a replace).
		const albSecurityGroup = yield* AWS.EC2.SecurityGroup("ingest-alb-sg", {
			vpcId: network.vpcId,
			groupName: name("ingest-alb"),
			description: "Maple OTLP ingest - public HTTPS to the load balancer",
			ingress: (yield* cloudflareIpv4Ranges).map((cidr) => ({
				ipProtocol: "tcp",
				fromPort: 443,
				toPort: 443,
				cidrIpv4: cidr,
				description: "OTLP over HTTPS from Cloudflare",
			})),
		})

		// Host networking: an awsvpc task on EC2 cannot take a public IP, and without
		// one it has no egress in a VPC with no NAT. The ALB then targets instances,
		// which needs the `usesAwsvpc` change in `patches/alchemy@*.patch`.
		const clusterName = name("ingest")

		// The hosts carry public IPs, so this rule is what keeps plaintext OTLP
		// from reaching a host directly instead of through the ALB.
		const instanceSecurityGroup = yield* AWS.EC2.SecurityGroup("ingest-ec2-sg", {
			vpcId: network.vpcId,
			groupName: name("ingest-ec2"),
			description: "Maple OTLP ingest gateway hosts",
			ingress: [
				{
					ipProtocol: "tcp",
					fromPort: INGEST_PORT,
					toPort: INGEST_PORT,
					referencedGroupId: albSecurityGroup.groupId,
					description: "ALB to gateway",
				},
				{
					ipProtocol: "tcp",
					fromPort: INGEST_GRPC_PORT,
					toPort: INGEST_GRPC_PORT,
					referencedGroupId: albSecurityGroup.groupId,
					description: "ALB to gateway OTLP/gRPC",
				},
			],
		})

		// ECS agent registration, ECR pulls and logs, plus Session Manager instead of SSH.
		// The gateway itself gets the TASK role, not this one.
		const instanceRole = yield* AWS.IAM.Role("ingest-ec2-instance-role", {
			roleName: name("ingest-ec2-instance"),
			assumeRolePolicyDocument: {
				Version: "2012-10-17",
				Statement: [
					{
						Effect: "Allow",
						Principal: { Service: "ec2.amazonaws.com" },
						Action: ["sts:AssumeRole"],
					},
				],
			},
			managedPolicyArns: [
				"arn:aws:iam::aws:policy/service-role/AmazonEC2ContainerServiceforEC2Role",
				"arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore",
			],
			tags,
		})
		const instanceProfile = yield* AWS.IAM.InstanceProfile("ingest-ec2-instance-profile", {
			instanceProfileName: name("ingest-ec2-instance"),
			roleName: instanceRole.roleName,
		})

		// Newest ECS-optimized AL2023 arm64 image. Every ingest rollout launches fresh hosts
		// (the old task holds the port), so AMI patching rides the ingest changes.
		const imageId = AWS.EC2.getAmi({
			owners: ["amazon"],
			name: ["al2023-ami-ecs-hvm-*-kernel-6.1-arm64"],
			architecture: "arm64",
		}).ImageId.as<string>()

		const launchTemplate = yield* AWS.AutoScaling.LaunchTemplate("ingest-ec2-launch-template", {
			launchTemplateName: name("ingest-ec2"),
			imageId,
			instanceType,
			securityGroupIds: [instanceSecurityGroup.groupId],
			instanceProfileName: instanceProfile.instanceProfileName,
			associatePublicIpAddress: true,
			userData: ec2UserData(clusterName, WAL_STORAGE[instanceType].disk),
			tags,
		})

		// ECS managed scaling owns the instance count (the patch keeps a redeploy from
		// resetting it to `minSize`). Max is doubled so a rolling deploy can overlap hosts.
		const maxTasks = scaling?.max ?? desiredCount
		const autoScalingGroup = yield* AWS.AutoScaling.AutoScalingGroup("ingest-ec2-asg", {
			autoScalingGroupName: name("ingest-ec2"),
			launchTemplate,
			subnetIds: network.publicSubnetIds,
			minSize: 0,
			maxSize: maxTasks * 2,
			healthCheckType: "EC2",
			healthCheckGracePeriod: "2 minutes",
			// ECS stamps this tag on adoption, and alchemy converges tags to the declared set.
			tags: { ...tags, AmazonECSManaged: "" },
		})

		// Managed draining: a scale-in drains the host's task first, and its SIGTERM
		// path empties the WAL (shutdown drain, then the S3 tier).
		const capacityProvider = yield* AWS.ECS.CapacityProvider("ingest-ec2-capacity", {
			name: name("ingest-ec2"),
			autoScalingGroupArn: autoScalingGroup.autoScalingGroupArn,
			managedScaling: {
				status: "ENABLED",
				targetCapacity: 100,
				minimumScalingStepSize: 1,
				maximumScalingStepSize: 2,
				instanceWarmupPeriod: 120,
			},
			managedTerminationProtection: "DISABLED",
			managedDraining: "ENABLED",
			tags,
		})

		const cluster = yield* AWS.ECS.Cluster("ingest-cluster", {
			clusterName,
			capacityProviders: [capacityProvider.name],
			tags,
		})

		const secret = ecsSecrets(name("ingest"), tags)

		const tinybirdTokenValue = yield* requiredPlain("TINYBIRD_TOKEN")
		const tinybirdToken = yield* secret("tinybird-token", tinybirdTokenValue)
		// NOT `MAPLE_PG_URL`: that is the migration admin's URL. The gateway reads
		// ingest keys through the pooler as its own role.
		const pgUrl = dbLogin
			? yield* secret("maple-pg-url", pgUrlRequireSsl(dbLogin.url))
			: yield* requiredPlain("MAPLE_INGEST_PG_URL").pipe(
					Effect.flatMap((url) => secret("maple-pg-url", url)),
				)
		const keyEncryptionKeyValue = yield* requiredPlain("MAPLE_INGEST_KEY_ENCRYPTION_KEY")
		const keyEncryptionKey = yield* secret("ingest-key-encryption-key", keyEncryptionKeyValue)
		// Only set while a key rotation is in flight (docs/secret-encryption.md).
		const { MAPLE_INGEST_KEY_DECRYPTION_KEYS: decryptionKeysValue } = yield* optionalPlain(
			"MAPLE_INGEST_KEY_DECRYPTION_KEYS",
		)
		const keyDecryptionKeys = decryptionKeysValue
			? yield* secret("ingest-key-decryption-keys", decryptionKeysValue)
			: undefined
		const keyLookupHmacKeyValue = yield* requiredPlain("MAPLE_INGEST_KEY_LOOKUP_HMAC_KEY")
		const keyLookupHmacKey = yield* secret("ingest-key-lookup-hmac-key", keyLookupHmacKeyValue)

		// Autumn absent means billing enforcement is dark.
		const { AUTUMN_SECRET_KEY: autumnKey } = yield* optionalPlain("AUTUMN_SECRET_KEY")
		const autumnSecret = autumnKey ? yield* secret("autumn-secret-key", autumnKey) : undefined

		// The access key id is not secret, but it only exists once the token does and
		// `env` takes plan-time strings only.
		const replayR2Secret = replayBlobs
			? yield* secret("replay-r2-secret-access-key", replayBlobs.secretAccessKey)
			: undefined
		const replayR2AccessKeyId = replayBlobs
			? yield* secret("replay-r2-access-key-id", replayBlobs.accessKeyId)
			: undefined

		// prd only, or one preview that forces it (`profile.deploys.collector`).
		const collectorEndpoint = profile.deploys.collector
			? yield* createIngestCollector({
					stage,
					region,
					tags,
					network,
					cluster,
					gatewaySecurityGroupId: instanceSecurityGroup.groupId,
					tinybirdTokenArn: tinybirdToken.secretArn,
					taskSize: collectorTaskSize,
				})
			: undefined

		const issuedCertificateArn = yield* issueRegionalCertificate({
			id: "ingest-cert",
			hostname,
			region: resolveAwsRegion(region),
			tags,
		})

		// Durability tier for the WAL (`apps/ingest/src/wal_store.rs`): sealed,
		// unexported segments, claimed by the next task if their owner dies.
		// Named up front so the env var below is a plain string.
		const walBucketName = name("ingest-wal")
		const walSegments = yield* AWS.S3.Bucket("ingest-wal-segments", {
			bucketName: walBucketName,
			// Backstop for segments whose delete was lost.
			lifecycleRules: [
				{
					ID: "expire-wal-segments",
					Status: "Enabled",
					Filter: { Prefix: "wal/" },
					Expiration: { Days: 7 },
					AbortIncompleteMultipartUpload: { DaysAfterInitiation: 1 },
				},
			],
			publicAccessBlock: {
				blockPublicAcls: true,
				blockPublicPolicy: true,
				ignorePublicAcls: true,
				restrictPublicBuckets: true,
			},
			encryption: { sseAlgorithm: "AES256" },
			forceDestroy: true,
			tags,
		})

		// Its own logical id: alchemy keys resources by id alone, and a repeat id
		// silently returns the first registration. Keep these policies without a
		// `description`: changing one replaces the policy under the same name, and the
		// patched delete detaches the old one from the live task role.
		const walSegmentsPolicy = yield* AWS.IAM.Policy("ingest-wal-segments-access", {
			policyName: name("ingest-wal-segments"),
			policyDocument: {
				Version: "2012-10-17",
				Statement: [
					{
						Effect: "Allow",
						Action: ["s3:GetObject", "s3:PutObject", "s3:DeleteObject"],
						Resource: [Output.map(walSegments.bucketArn, (arn) => `${arn}/wal/*`)],
					},
					{
						Effect: "Allow",
						Action: ["s3:ListBucket"],
						Resource: [walSegments.bucketArn],
						Condition: { StringLike: { "s3:prefix": ["wal/*"] } },
					},
				],
			},
		})

		// The gateway marks its own task scale-in-protected while the WAL holds
		// backlog (`apps/ingest/src/task_protection.rs`).
		const taskProtectionPolicy = yield* AWS.IAM.Policy("ingest-task-protection", {
			policyName: name("ingest-task-protection"),
			policyDocument: {
				Version: "2012-10-17",
				Statement: [
					{
						Effect: "Allow",
						Action: ["ecs:UpdateTaskProtection"],
						Resource: [
							Output.map(
								cluster.clusterArn,
								(arn) => `${arn.replace(":cluster/", ":task/")}/*`,
							),
						],
					},
				],
			},
		})

		// One task per host, bound to the host's port and NVMe. A rolling deploy brings
		// up a fresh host for the new task (the port is taken) and drains the old one.
		// `securityGroups` reaches only the ALB here: the host's own group admits it.
		const service = yield* AWS.ECS.Service("ingest-ec2", {
			cluster,
			serviceName: name("ingest-ec2"),
			taskRoleManagedPolicyArns: [taskProtectionPolicy.policyArn, walSegmentsPolicy.policyArn],

			// Rebuilt only when the context hash changes. CI leaves a prebuilt binary
			// at dist/ (fast COPY build); without it this falls back to a source build.
			context: "apps/ingest",
			...(existsSync(PREBUILT_BINARY) ? { dockerfile: PREBUILT_DOCKERFILE } : undefined),
			// The docker build platform is derived from this, so the binary must be
			// aarch64 too: a mismatch only shows up as `exec format error` at runtime.
			runtimePlatform: { cpuArchitecture: "ARM64", operatingSystemFamily: "LINUX" },
			networkMode: "host",
			requiresCompatibilities: ["EC2"],
			capacityProviderStrategy: [{ capacityProvider: capacityProvider.name, weight: 1 }],
			placementConstraints: [{ type: "distinctInstance" }],
			cpu: taskSize.cpu,
			memory: taskSize.memory,
			volumes: [{ name: "wal", host: { sourcePath: WAL_HOST_DIR } }],
			container: {
				// The shutdown drain (`INGEST_SHUTDOWN_DRAIN_SECS`, default 90) must fit inside this.
				stopTimeout: 120,
				mountPoints: [{ sourceVolume: "wal", containerPath: WAL_CONTAINER_DIR }],
				// Replaces the mapping alchemy derives from `port`; host mode needs hostPort = containerPort.
				portMappings: [INGEST_PORT, INGEST_GRPC_PORT].map((port) => ({
					containerPort: port,
					hostPort: port,
					protocol: "tcp",
				})),
			},

			desiredCount,
			// alchemy stops pinning desiredCount while `scaling` is set.
			...(scaling ? { scaling } : undefined),
			vpcId: network.vpcId,
			subnets: network.publicSubnetIds,
			securityGroups: [albSecurityGroup.groupId],

			// The conditionless rule is the listener's default action, the same ALB, listener
			// and target group `true` composes. OTLP/gRPC paths go to a gRPC target group.
			loadBalancer: {
				rules: [
					{},
					{
						path: OTLP_GRPC_PATH,
						forward: `${INGEST_GRPC_PORT}/http`,
						protocolVersion: "GRPC",
					},
				],
			},
			// `port` is the CONTAINER port; the listener defaults to 443 with a
			// certificate. Do not set `listenerPort`: Cloudflare cannot proxy to 3474.
			port: INGEST_PORT,
			healthCheckPath: "/health",
			certificateArn: issuedCertificateArn,
			// Covers the startup Postgres probe, which exits the process on failure.
			healthCheckGracePeriod: "60 seconds",
			// Old tasks stay scale-in protected for up to 15 minutes while the WAL drains,
			// which outlasts alchemy's 10-minute default.
			deploymentStabilizationTimeout: "25 minutes",

			logging: { retention: "30 days" },

			secrets: {
				TINYBIRD_TOKEN: tinybirdToken.secretArn,
				MAPLE_PG_URL: pgUrl.secretArn,
				MAPLE_INGEST_KEY_ENCRYPTION_KEY: keyEncryptionKey.secretArn,
				...(keyDecryptionKeys
					? { MAPLE_INGEST_KEY_DECRYPTION_KEYS: keyDecryptionKeys.secretArn }
					: undefined),
				MAPLE_INGEST_KEY_LOOKUP_HMAC_KEY: keyLookupHmacKey.secretArn,
				...(autumnSecret ? { AUTUMN_SECRET_KEY: autumnSecret.secretArn } : undefined),
				...(replayR2Secret && replayR2AccessKeyId
					? {
							INGEST_REPLAY_R2_SECRET_ACCESS_KEY: replayR2Secret.secretArn,
							INGEST_REPLAY_R2_ACCESS_KEY_ID: replayR2AccessKeyId.secretArn,
						}
					: undefined),
			},

			// `optionalPlain` returns a Config: always `yield*` it before spreading, or the
			// variable is silently dropped. The `satisfies` below turns that into a type error.
			env: {
				INGEST_PORT: String(INGEST_PORT),
				INGEST_OTLP_GRPC_PORT: String(INGEST_GRPC_PORT),
				MAPLE_ENVIRONMENT: resolveDeploymentEnvironment(stage),
				// Lets the unknown-key 401 name the other region's ingest URL.
				...(stage.kind === "prd" && { MAPLE_REGION: region }),
				TINYBIRD_HOST: yield* requiredPlain("TINYBIRD_HOST"),
				INGEST_KEY_STORE_BACKEND: "postgres",
				// A replaced role changes the task definition, so the fleet rolls onto the
				// new secret before alchemy deletes the old role.
				...(dbLogin && { MAPLE_PG_ROLE_ID: dbLogin.id }),

				// Trust `Cf-IPCountry` (fills `session_replays.Country`): the ALB admits only Cloudflare.
				MAPLE_INGEST_TRUST_PROXY_GEO: "true",

				INGEST_QUEUE_MAX_BYTES: String(WAL_STORAGE[instanceType].maxBytes),
				INGEST_WAL_SHARDS: String(WAL_SHARDS),
				// The task role signs these requests; the S3 gateway endpoint keeps them private.
				INGEST_WAL_S3_BUCKET: walBucketName,
				INGEST_WAL_S3_REGION: resolveAwsRegion(region),

				...(replayBlobs
					? {
							INGEST_REPLAY_R2_ENDPOINT: replayBlobs.endpoint,
							INGEST_REPLAY_R2_BUCKET: replayBlobs.bucket,
							...(yield* optionalPlain("INGEST_REPLAY_R2_REGION", "auto")),
						}
					: undefined),

				// The gateway's own telemetry (and customer OTLP in forward/dual mode)
				// goes to the in-VPC collector when this stage runs one.
				...(collectorEndpoint
					? { INGEST_FORWARD_OTLP_ENDPOINT: collectorEndpoint }
					: yield* optionalPlain("INGEST_FORWARD_OTLP_ENDPOINT")),
				...(selfTraceSampleRatio
					? { INGEST_SELF_TRACE_SAMPLE_RATIO: selfTraceSampleRatio }
					: yield* optionalPlain("INGEST_SELF_TRACE_SAMPLE_RATIO")),
				// Source token rps limits are split across this many tasks. The scaling
				// floor, so a scaled-out fleet admits more rather than throttling below the limit.
				INGEST_RATE_LIMIT_FLEET_SIZE: String(scaling?.min ?? desiredCount),
				...(yield* blockedIngestKeysEnv),
				// The org Maple's own telemetry is filed under. Required, no fallback.
				MAPLE_INTERNAL_ORG_ID: yield* requiredPlain("MAPLE_INTERNAL_ORG_ID"),
				...(yield* gatewayTuningEnv),
				// `INGEST_COMMIT_SHA` (prd) is the image's own revision, so a deploy that leaves
				// `apps/ingest` alone does not change the task definition and roll the hosts.
				...(yield* commitShaEnv("INGEST_COMMIT_SHA")),
			} satisfies Record<string, string | Output.Output<string>>,

			tags,
		})

		// The public name, proxied through Cloudflare to the ALB.
		yield* publishProxiedCname({ id: "ingest-public-cname", hostname, serviceUrl: service.url })

		if (stage.kind === "prd" && region === "us") yield* blockIngestKeysAtEdge(stage)

		return {
			// The ALB itself, behind the public hostname.
			serviceUrl: service.url,
			// Shared with `apps/electric`: two `AWS.EC2.Network`s in one stack fight
			// over the internet gateway.
			network,
			// Resolvable only inside the VPC; surfaced so a preview's logs say where the gateway points.
			collectorEndpoint,
		}
	})