| 1 | import * as Effect from "effect/Effect"; |
| 2 | import * as Binding from "../../Binding.ts"; |
| 3 | import * as Output from "../../Output.ts"; |
| 4 | import { isBindingHost } from "../Lambda/Function.ts"; |
| 5 | import type { ServerlessCluster } from "./ServerlessCluster.ts"; |
| 6 | |
| 7 | /** |
| 8 | * Shared scaffolding for AWS MSK (Kafka) HTTP bindings. |
| 9 | * |
| 10 | * NOT exported from `index.ts` — every `{Op}Http.ts` in this service is a |
| 11 | * thin `Layer.effect(Cap, makeKafkaClusterHttpBinding({ … }))` over the |
| 12 | * builder below. Everything except the operation and the IAM action list is |
| 13 | * boilerplate. |
| 14 | */ |
| 15 | |
| 16 | /** |
| 17 | * The cluster ARN is `arn:…:cluster/name/uuid`. Topic resources live under a |
| 18 | * sibling `topic/` namespace scoped to the same cluster path; both the |
| 19 | * `kafka:*Topic*` control-plane actions and the `kafka-cluster:*` data-plane |
| 20 | * actions authorize against these ARNs. |
| 21 | */ |
| 22 | export const topicArnGlob = (clusterArn: string) => |
| 23 | `${clusterArn.replace(":cluster/", ":topic/")}/*`; |
| 24 | |
| 25 | /** Consumer-group ARNs under the cluster's `group/` namespace. */ |
| 26 | export const groupArnGlob = (clusterArn: string) => |
| 27 | `${clusterArn.replace(":cluster/", ":group/")}/*`; |
| 28 | |
| 29 | /** Transactional-id ARNs under the cluster's `transactional-id/` namespace. */ |
| 30 | export const transactionalIdArnGlob = (clusterArn: string) => |
| 31 | `${clusterArn.replace(":cluster/", ":transactional-id/")}/*`; |
| 32 | |
| 33 | /** |
| 34 | * Build the impl Effect for a cluster-scoped MSK control-plane operation: the |
| 35 | * runtime callable injects the bound {@link ServerlessCluster}'s ARN as |
| 36 | * `ClusterArn` and the deploy-time half grants `actions` on the cluster ARN |
| 37 | * (plus the cluster's `topic/` ARN namespace when `topicScoped` is set — |
| 38 | * topic-management actions authorize against topic ARNs). |
| 39 | */ |
| 40 | export const makeKafkaClusterHttpBinding = < |
| 41 | I extends { ClusterArn: string }, |
| 42 | A, |
| 43 | E, |
| 44 | R, |
| 45 | >(options: { |
| 46 | /** Fully-qualified binding tag, e.g. `AWS.Kafka.ListTopics`. */ |
| 47 | tag: string; |
| 48 | /** The distilled operation; `ClusterArn` is injected from the cluster. */ |
| 49 | operation: Effect.Effect<(input: I) => Effect.Effect<A, E>, never, R>; |
| 50 | /** IAM actions granted on the cluster ARN (and topic glob, if scoped). */ |
| 51 | actions: readonly string[]; |
| 52 | /** Also grant `actions` on the cluster's `topic/…/*` ARN namespace. */ |
| 53 | topicScoped?: boolean; |
| 54 | }) => |
| 55 | Effect.gen(function* () { |
| 56 | const op = yield* options.operation; |
| 57 | |
| 58 | return Effect.fn(function* (cluster: ServerlessCluster) { |
| 59 | const ClusterArn = yield* cluster.clusterArn; |
| 60 | if (!globalThis.__ALCHEMY_RUNTIME__) { |
| 61 | const host = yield* Binding.Host; |
| 62 | if (isBindingHost(host)) { |
| 63 | yield* host.bind`Allow(${host}, ${options.tag}(${cluster}))`({ |
| 64 | policyStatements: [ |
| 65 | { |
| 66 | Effect: "Allow", |
| 67 | Action: [...options.actions], |
| 68 | Resource: options.topicScoped |
| 69 | ? [ |
| 70 | Output.interpolate`${cluster.clusterArn}`, |
| 71 | cluster.clusterArn.pipe(Output.map(topicArnGlob)), |
| 72 | ] |
| 73 | : [Output.interpolate`${cluster.clusterArn}`], |
| 74 | }, |
| 75 | ], |
| 76 | }); |
| 77 | } |
| 78 | } |
| 79 | return Effect.fn(`${options.tag}(${cluster.LogicalId})`)(function* ( |
| 80 | request?: Omit<I, "ClusterArn">, |
| 81 | ) { |
| 82 | return yield* op({ |
| 83 | ...request, |
| 84 | ClusterArn: yield* ClusterArn, |
| 85 | } as I); |
| 86 | }); |
| 87 | }); |
| 88 | }); |