1import * as Effect from "effect/Effect";
2import * as Binding from "../../Binding.ts";
3import * as Output from "../../Output.ts";
4import { isBindingHost } from "../Lambda/Function.ts";
5import {
6 groupArnGlob,
7 topicArnGlob,
8 transactionalIdArnGlob,
9} from "./BindingHttp.ts";
10import { connectEnvPrefix } from "./Connect.ts";
11import type { ServerlessCluster } from "./ServerlessCluster.ts";
12
13/**
14 * Shared scaffolding for the MSK connect bindings.
15 *
16 * NOT exported from `index.ts` — `ConnectReadHttp.ts` / `ConnectWriteHttp.ts`
17 * / `ConnectReadWriteHttp.ts` are thin
18 * `Layer.effect(Cap, makeKafkaConnectHttpBinding({ … }))` calls over the
19 * builder below. Only the IAM action sets differ per access level.
20 */
21
22/** Cluster-level actions every MSK IAM-auth client needs. */
23export const KAFKA_CONNECT_ACTIONS = [
24 "kafka-cluster:Connect",
25 "kafka-cluster:DescribeCluster",
26 "kafka-cluster:DescribeClusterDynamicConfiguration",
27] as const;
28
29/** Topic-level consumer actions. */
30export const KAFKA_READ_TOPIC_ACTIONS = [
31 "kafka-cluster:DescribeTopic",
32 "kafka-cluster:ReadData",
33] as const;
34
35/** Consumer-group actions (join a group, commit offsets). */
36export const KAFKA_READ_GROUP_ACTIONS = [
37 "kafka-cluster:DescribeGroup",
38 "kafka-cluster:AlterGroup",
39] as const;
40
41/** Topic-level producer actions. */
42export const KAFKA_WRITE_TOPIC_ACTIONS = [
43 "kafka-cluster:DescribeTopic",
44 "kafka-cluster:WriteData",
45] as const;
46
47/** Cluster-level producer actions (idempotent producers). */
48export const KAFKA_WRITE_CLUSTER_ACTIONS = [
49 "kafka-cluster:WriteDataIdempotently",
50] as const;
51
52/** Transactional-id actions (transactional producers). */
53export const KAFKA_WRITE_TRANSACTION_ACTIONS = [
54 "kafka-cluster:DescribeTransactionalId",
55 "kafka-cluster:AlterTransactionalId",
56] as const;
57
58/**
59 * Build the impl Effect for a connect binding. At deploy time it grants the
60 * given action sets on the cluster / topic / group / transactional-id ARN
61 * namespaces and publishes the SASL/IAM bootstrap endpoint as
62 * `MSK_{LOGICAL_ID}_{BROKERS,ARN}` environment variables on the host
63 * Function; at runtime it resolves the same values into a typed connection
64 * descriptor.
65 */
66export const makeKafkaConnectHttpBinding = (options: {
67 /** Fully-qualified binding tag, e.g. `AWS.Kafka.ConnectReadWrite`. */
68 tag: string;
69 /** IAM actions granted on the cluster ARN. */
70 clusterActions: readonly string[];
71 /** IAM actions granted on the cluster's `topic/…/*` ARNs. */
72 topicActions: readonly string[];
73 /** IAM actions granted on the cluster's `group/…/*` ARNs. */
74 groupActions?: readonly string[];
75 /** IAM actions granted on the cluster's `transactional-id/…/*` ARNs. */
76 transactionActions?: readonly string[];
77}) =>
78 Effect.gen(function* () {
79 return Effect.fn(function* (cluster: ServerlessCluster) {
80 const BootstrapServers = yield* cluster.bootstrapBrokerStringSaslIam;
81 const ClusterArn = yield* cluster.clusterArn;
82
83 if (!globalThis.__ALCHEMY_RUNTIME__) {
84 const host = yield* Binding.Host;
85 if (isBindingHost(host)) {
86 const prefix = connectEnvPrefix(cluster.LogicalId);
87 yield* host.bind`Allow(${host}, ${options.tag}(${cluster}))`({
88 env: {
89 [`${prefix}_BROKERS`]: cluster.bootstrapBrokerStringSaslIam,
90 [`${prefix}_ARN`]: cluster.clusterArn,
91 },
92 policyStatements: [
93 {
94 Effect: "Allow",
95 Action: [...options.clusterActions],
96 Resource: [Output.interpolate`${cluster.clusterArn}`],
97 },
98 {
99 Effect: "Allow",
100 Action: [...options.topicActions],
101 Resource: [cluster.clusterArn.pipe(Output.map(topicArnGlob))],
102 },
103 ...(options.groupActions?.length
104 ? [
105 {
106 Effect: "Allow" as const,
107 Action: [...options.groupActions],
108 Resource: [
109 cluster.clusterArn.pipe(Output.map(groupArnGlob)),
110 ],
111 },
112 ]
113 : []),
114 ...(options.transactionActions?.length
115 ? [
116 {
117 Effect: "Allow" as const,
118 Action: [...options.transactionActions],
119 Resource: [
120 cluster.clusterArn.pipe(
121 Output.map(transactionalIdArnGlob),
122 ),
123 ],
124 },
125 ]
126 : []),
127 ],
128 });
129 }
130 }
131
132 return Effect.gen(function* () {
133 const bootstrapServers = yield* BootstrapServers;
134 const clusterArn = yield* ClusterArn;
135 if (!bootstrapServers) {
136 return yield* Effect.die(
137 `MSK SASL/IAM bootstrap brokers for '${cluster.LogicalId}' are not available yet`,
138 );
139 }
140 return {
141 bootstrapServers,
142 brokers: bootstrapServers.split(","),
143 clusterArn,
144 authentication: "iam" as const,
145 };
146 });
147 });
148 });