1import * as kafka from "@distilled.cloud/gcp/managedkafka_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { GcpEnvironment } from "../Environment.ts";
10import type { Providers } from "../Providers.ts";
11import {
12 collectPages,
13 expandParent,
14 fieldMask,
15 fingerprint,
16 getTopic,
17 listAlchemyClusters,
18 normalizeLocation,
19 parseName,
20 replaceOnIdentity,
21 retryTransient,
22 rfc1035,
23 stringMapOf,
24 toPhysicalId,
25} from "./internal.ts";
26
27const DEFAULT_PARTITION_COUNT = 1;
28const DEFAULT_REPLICATION_FACTOR = 3;
29
30export type ClustersTopicProps = {
31 /**
32 * Parent cluster. Full name
33 * `projects/{project}/locations/{location}/clusters/{cluster}` or the
34 * cluster id (combined with `location`). Immutable — changing it
35 * replaces the topic.
36 */
37 cluster: string;
38 /**
39 * Region used when `cluster` is a bare id.
40 * @default the stack's GCP region (`GCP.Region`, else the profile region, else `us-central1`)
41 */
42 location?: string;
43 /**
44 * Topic id (the `{topic}` segment of `.../clusters/{cluster}/topics/{topic}`).
45 * If omitted, a unique RFC1035 name is generated. Immutable — changing
46 * it replaces the topic.
47 */
48 topicId?: string;
49 /**
50 * Partition count. Can only be increased, not decreased.
51 * @default 1
52 */
53 partitionCount?: number;
54 /**
55 * Replication factor. Immutable — changing it replaces the topic.
56 * @default 3
57 */
58 replicationFactor?: number;
59 /**
60 * Topic configs that override cluster defaults (`cleanup.policy`,
61 * `compression.type`, …).
62 */
63 configs?: Record<string, string>;
64};
65
66export type ClustersTopic = Resource<
67 "GCP.ManagedKafka.ClustersTopic",
68 ClustersTopicProps,
69 {
70 /** Full resource name `.../clusters/{cluster}/topics/{topic}`. */
71 name: string;
72 /** Topic id. */
73 topicId: string;
74 /** Parent cluster resource name. */
75 cluster: string;
76 /** Project id. */
77 project: string;
78 /** Location id. */
79 location: string;
80 /** Partition count. */
81 partitionCount: number;
82 /** Replication factor. */
83 replicationFactor: number;
84 /** Topic config overrides. */
85 configs: Record<string, string>;
86 },
87 never,
88 Providers
89>;
90
91/**
92 * A Kafka topic on a Managed Service for Apache Kafka cluster.
93 *
94 * Changing `topicId`, `cluster`, `location`, or `replicationFactor`
95 * replaces the topic. `partitionCount` can only increase. Configs update
96 * in place.
97 *
98 * ### Creating a Topic
99 * **Example:** Generated name
100 * ```typescript
101 * const topic = yield* GCP.ManagedKafka.ClustersTopic("Events", {
102 * cluster: cluster.name,
103 * });
104 * ```
105 *
106 * **Example:** Explicit id, partitions, and configs
107 * ```typescript
108 * const topic = yield* GCP.ManagedKafka.ClustersTopic("Events", {
109 * cluster: cluster.name,
110 * topicId: "order-events",
111 * partitionCount: 3,
112 * replicationFactor: 3,
113 * configs: { "cleanup.policy": "compact" },
114 * });
115 * ```
116 *
117 * @resource
118 * @category ManagedKafka
119 */
120export const ClustersTopic = Resource<ClustersTopic>(
121 "GCP.ManagedKafka.ClustersTopic",
122);
123
124export class ClustersTopicNotResolved extends Data.TaggedError(
125 "GCP.ManagedKafka.ClustersTopicNotResolved",
126)<{
127 name: string;
128}> {}
129
130const clusterOf = (cluster: string, project: string, location: string) =>
131 expandParent(cluster, project, location, "clusters");
132
133const resourceName = (cluster: string, topicId: string) =>
134 `${cluster}/topics/${topicId}`;
135
136const toAttrs = (topic: kafka.Topic, project: string) => {
137 const name = topic.name ?? "";
138 const parsed = parseName(name, "topics");
139 return {
140 name,
141 topicId: parsed.id,
142 cluster: parsed.parent,
143 project: parsed.project || project,
144 location: parsed.location,
145 partitionCount: topic.partitionCount ?? DEFAULT_PARTITION_COUNT,
146 replicationFactor: topic.replicationFactor ?? DEFAULT_REPLICATION_FACTOR,
147 configs: stringMapOf(topic.configs),
148 };
149};
150
151export const ClustersTopicProvider = () =>
152 Provider.succeed(ClustersTopic, {
153 stables: ["name", "topicId", "cluster", "project", "location"],
154
155 diff: Effect.fn(function* ({ news, olds, output }) {
156 if (!isResolved(news)) return undefined;
157 const env = yield* GcpEnvironment.current;
158 const location = normalizeLocation(
159 news.location ?? output?.location,
160 env.region,
161 );
162 const previousFactor =
163 olds?.replicationFactor ?? output?.replicationFactor;
164 const nextFactor = news.replicationFactor ?? previousFactor;
165 return replaceOnIdentity({
166 previousId: olds?.topicId ?? output?.topicId,
167 nextId: news.topicId
168 ? rfc1035(news.topicId, "topic")
169 : (olds?.topicId ?? output?.topicId),
170 previousParent: olds?.cluster ?? output?.cluster,
171 nextParent: clusterOf(news.cluster, env.project, location),
172 extra:
173 previousFactor !== undefined &&
174 nextFactor !== undefined &&
175 previousFactor !== nextFactor,
176 });
177 }),
178
179 read: Effect.fn(function* ({ id, olds, output }) {
180 const env = yield* GcpEnvironment.current;
181 const location = normalizeLocation(
182 olds?.location ?? output?.location,
183 env.region,
184 );
185 const topicId = yield* toPhysicalId(
186 id,
187 olds?.topicId,
188 output?.topicId,
189 "topic",
190 );
191 const cluster =
192 olds?.cluster !== undefined
193 ? clusterOf(olds.cluster, env.project, location)
194 : (output?.cluster ?? "");
195 const name =
196 output?.name ??
197 (cluster.length > 0 ? resourceName(cluster, topicId) : "");
198 const existing = yield* getTopic(name);
199 if (existing === undefined) return undefined;
200 const attrs = toAttrs(existing, env.project);
201 return output !== undefined || olds !== undefined
202 ? attrs
203 : Unowned(attrs);
204 }),
205
206 list: () =>
207 Effect.gen(function* () {
208 const env = yield* GcpEnvironment.current;
209 const clusters = yield* listAlchemyClusters(env.project, env.region);
210 const topics = yield* Effect.forEach(
211 clusters.filter((cluster) => (cluster.name ?? "").length > 0),
212 (cluster: kafka.Cluster) =>
213 collectPages(
214 kafka.listProjectsLocationsClustersTopics.pages({
215 parent: cluster.name!,
216 pageSize: 1000,
217 }),
218 (page) => page.topics,
219 ).pipe(
220 Effect.catchTag("NotFound", () =>
221 Effect.succeed([] as kafka.Topic[]),
222 ),
223 ),
224 { concurrency: 4 },
225 );
226 return topics.flat().map((topic) => toAttrs(topic, env.project));
227 }),
228
229 reconcile: Effect.fn(function* ({ id, news, output }) {
230 const env = yield* GcpEnvironment.current;
231 const location = normalizeLocation(
232 news.location ?? output?.location,
233 env.region,
234 );
235 const cluster = clusterOf(news.cluster, env.project, location);
236 const topicId = yield* toPhysicalId(
237 id,
238 news.topicId,
239 output?.topicId,
240 "topic",
241 );
242 const name = output?.name ?? resourceName(cluster, topicId);
243 const partitionCount = news.partitionCount ?? DEFAULT_PARTITION_COUNT;
244 const replicationFactor =
245 news.replicationFactor ?? DEFAULT_REPLICATION_FACTOR;
246 const configs = news.configs;
247
248 let current = yield* getTopic(name);
249
250 if (current === undefined) {
251 const created = yield* retryTransient(
252 kafka
253 .createProjectsLocationsClustersTopics({
254 parent: cluster,
255 topicId,
256 body: {
257 partitionCount,
258 replicationFactor,
259 configs,
260 },
261 })
262 .pipe(Effect.catchTag("Conflict", () => getTopic(name))),
263 );
264 current = created ?? undefined;
265 // A new topic reads back NotFound (UnknownTopicOrPartition) until its
266 // partitions are assigned; block until it is readable.
267 if (current !== undefined) {
268 current = yield* getTopic(name).pipe(
269 Effect.repeat({
270 schedule: Schedule.spaced("5 seconds"),
271 until: (topic) => topic !== undefined,
272 times: 24,
273 }),
274 );
275 }
276 }
277
278 if (current === undefined) {
279 return yield* new ClustersTopicNotResolved({ name });
280 }
281
282 const partitionsChanged =
283 (current.partitionCount ?? DEFAULT_PARTITION_COUNT) !== partitionCount;
284 const configsChanged =
285 fingerprint(stringMapOf(current.configs)) !==
286 fingerprint(stringMapOf(configs));
287
288 if (partitionsChanged || configsChanged) {
289 current = yield* kafka.patchProjectsLocationsClustersTopics({
290 name,
291 updateMask: fieldMask([
292 partitionsChanged && "partition_count",
293 configsChanged && "configs",
294 ]),
295 body: {
296 partitionCount,
297 configs,
298 },
299 });
300 }
301
302 return toAttrs(current, env.project);
303 }),
304
305 delete: Effect.fn(function* ({ output }) {
306 yield* kafka
307 .deleteProjectsLocationsClustersTopics({ name: output.name })
308 .pipe(Effect.catchTag("NotFound", () => Effect.void));
309 }),
310 });