1import * as logging from "@distilled.cloud/gcp/logging_v2";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
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 { createInternalLabels, hasAlchemyLabels } from "../Labels.ts";
11import type { Providers } from "../Providers.ts";
12import type { SinkBigQueryOptions, SinkExclusion } from "./Sink.ts";
13import {
14 billingAccountIdOf,
15 billingAccountParent,
16 encodeDescription,
17 hasOwnershipMarker,
18 lastSegment,
19 lookupProjectBillingAccountId,
20 parseDescription,
21 resolveBillingAccountId,
22 toPhysicalId,
23} from "./internal.ts";
24
25export type BillingSinkExclusion = SinkExclusion;
26export type BillingSinkBigQueryOptions = SinkBigQueryOptions;
27
28export type BillingSinkProps = {
29 /**
30 * Sink id (the `{sink}` segment of
31 * `billingAccounts/{billingAccount}/sinks/{sink}`). If omitted, a unique
32 * name is generated from the stack, stage, and logical id. Limited to
33 * 100 characters: letters, digits, underscores, hyphens, periods; first
34 * character must be alphanumeric. Immutable — changing it replaces the
35 * sink.
36 */
37 sinkId?: string;
38 /**
39 * Billing account id (`XXXXXX-XXXXXX-XXXXXX` or
40 * `billingAccounts/{id}`). If omitted, Alchemy uses the billing
41 * account linked to the current project. Immutable — changing it
42 * replaces the sink.
43 */
44 billingAccountId?: string;
45 /**
46 * Export destination. One of:
47 * `storage.googleapis.com/[GCS_BUCKET]`,
48 * `bigquery.googleapis.com/projects/[PROJECT_ID]/datasets/[DATASET]`,
49 * `pubsub.googleapis.com/projects/[PROJECT_ID]/topics/[TOPIC_ID]`,
50 * `logging.googleapis.com/projects/[PROJECT_ID]`,
51 * `logging.googleapis.com/billingAccounts/[BILLING_ACCOUNT_ID]/locations/[LOCATION_ID]/buckets/[BUCKET_ID]`.
52 */
53 destination: string;
54 /**
55 * Advanced logs filter. Only matching entries in the sink's parent are
56 * exported. Empty or omitted exports every entry.
57 */
58 filter?: string;
59 /**
60 * Human-readable description (max 8000 characters). Logging sinks have
61 * no labels field, so Alchemy ownership is stored in a `[alchemy …]`
62 * prefix and stripped from attributes.
63 */
64 description?: string;
65 /**
66 * When true, the sink exists but does not export any log entries.
67 * @default false
68 */
69 disabled?: boolean;
70 /**
71 * Exclusion filters. An entry matching `filter` and any exclusion is
72 * not exported.
73 */
74 exclusions?: BillingSinkExclusion[];
75 /**
76 * BigQuery-specific export options. Ignored unless `destination` is a
77 * BigQuery dataset.
78 */
79 bigqueryOptions?: BillingSinkBigQueryOptions;
80 /**
81 * When true, Cloud Logging assigns a unique writer service account as
82 * `writerIdentity`. Billing-account sinks typically use a service
83 * agent.
84 */
85 uniqueWriterIdentity?: boolean;
86 /**
87 * Caller-provided writer service account
88 * (`serviceAccount:some@email`). Only valid when routing to a log
89 * bucket in a different project.
90 */
91 customWriterIdentity?: string;
92};
93
94export type BillingSink = Resource<
95 "GCP.Logging.BillingSink",
96 BillingSinkProps,
97 {
98 /** Full resource name `billingAccounts/{billingAccount}/sinks/{sink}`. */
99 name: string;
100 /** Sink id (last path segment). */
101 sinkId: string;
102 /** Billing account id. */
103 billingAccountId: string;
104 /** Export destination. */
105 destination: string;
106 /** Advanced logs filter, if set. */
107 filter: string | undefined;
108 /** User description with the Alchemy ownership prefix stripped. */
109 description: string | undefined;
110 /** Whether the sink is disabled. */
111 disabled: boolean;
112 /** Exclusion filters. */
113 exclusions: BillingSinkExclusion[];
114 /** BigQuery export options, if set. */
115 bigqueryOptions: BillingSinkBigQueryOptions | undefined;
116 /**
117 * IAM identity Cloud Logging uses to write to `destination`.
118 */
119 writerIdentity: string | undefined;
120 /** RFC3339 creation timestamp. */
121 createTime: string | undefined;
122 /** RFC3339 last-update timestamp. */
123 updateTime: string | undefined;
124 },
125 never,
126 Providers
127>;
128
129/**
130 * A Cloud Logging sink that routes billing-account log entries to a
131 * destination.
132 *
133 * Logging sinks have no labels field, so Alchemy stamps ownership into
134 * the description for `list` / nuke. Name is identity — changing `sinkId`
135 * or `billingAccountId` replaces the sink. Filter, destination,
136 * description, disabled flag, exclusions, and BigQuery options update in
137 * place.
138 *
139 * ### Creating a Billing Sink
140 * **Example:** Route errors to the `_Default` log bucket
141 * ```typescript
142 * const sink = yield* GCP.Logging.BillingSink("Errors", {
143 * destination:
144 * "logging.googleapis.com/billingAccounts/AAAAAA-BBBBBB-CCCCCC/locations/global/buckets/_Default",
145 * filter: "severity>=ERROR",
146 * });
147 * ```
148 *
149 * **Example:** Named sink with a description
150 * ```typescript
151 * const sink = yield* GCP.Logging.BillingSink("Errors", {
152 * sinkId: "app-errors",
153 * billingAccountId: "AAAAAA-BBBBBB-CCCCCC",
154 * destination:
155 * "logging.googleapis.com/billingAccounts/AAAAAA-BBBBBB-CCCCCC/locations/global/buckets/_Default",
156 * filter: "severity>=ERROR",
157 * description: "application errors",
158 * });
159 * ```
160 *
161 * ### Updating a Billing Sink
162 * **Example:** Change the filter and add an exclusion
163 * ```typescript
164 * const sink = yield* GCP.Logging.BillingSink("Errors", {
165 * destination:
166 * "logging.googleapis.com/billingAccounts/AAAAAA-BBBBBB-CCCCCC/locations/global/buckets/_Default",
167 * filter: "severity>=WARNING",
168 * description: "warnings and errors",
169 * exclusions: [
170 * {
171 * name: "drop-healthchecks",
172 * filter: 'httpRequest.requestUrl="/healthz"',
173 * },
174 * ],
175 * });
176 * ```
177 *
178 * @resource
179 * @category Logging
180 */
181export const BillingSink = Resource<BillingSink>("GCP.Logging.BillingSink");
182
183export class BillingSinkNotResolved extends Data.TaggedError(
184 "GCP.Logging.BillingSinkNotResolved",
185)<{
186 name: string;
187}> {}
188
189const resourceName = (billingAccountId: string, sinkId: string) =>
190 `${billingAccountParent(billingAccountId)}/sinks/${sinkId}`;
191
192const sinkIdOf = (sink: logging.LogSink) => {
193 const raw = sink.name ?? sink.resourceName ?? "";
194 return raw.includes("/") ? lastSegment(raw) : raw;
195};
196
197const billingAccountOfName = (name: string, fallback: string) => {
198 const match = name.match(/^billingAccounts\/([^/]+)\//);
199 return match?.[1] ?? fallback;
200};
201
202const exclusionsOf = (
203 list:
204 | readonly logging.LogExclusion[]
205 | readonly BillingSinkExclusion[]
206 | undefined,
207): BillingSinkExclusion[] =>
208 (list ?? [])
209 .map((exclusion) => ({
210 name: exclusion.name ?? "",
211 filter: exclusion.filter ?? "",
212 description: exclusion.description,
213 disabled: exclusion.disabled === true,
214 }))
215 .sort((left, right) => left.name.localeCompare(right.name));
216
217const sameExclusions = (
218 left:
219 | readonly logging.LogExclusion[]
220 | readonly BillingSinkExclusion[]
221 | undefined,
222 right:
223 | readonly logging.LogExclusion[]
224 | readonly BillingSinkExclusion[]
225 | undefined,
226) => JSON.stringify(exclusionsOf(left)) === JSON.stringify(exclusionsOf(right));
227
228const toExclusionsBody = (
229 list: readonly BillingSinkExclusion[] | undefined,
230): logging.LogExclusion[] | undefined => {
231 if (list === undefined) return undefined;
232 return list.map((exclusion) => ({
233 name: exclusion.name,
234 filter: exclusion.filter,
235 description: exclusion.description,
236 disabled: exclusion.disabled === true ? true : undefined,
237 }));
238};
239
240const toAttrs = (sink: logging.LogSink, billingAccountId: string) => {
241 const sinkId = sinkIdOf(sink);
242 const parsed = parseDescription(sink.description);
243 const bq = sink.bigqueryOptions;
244 const account = billingAccountOfName(
245 sink.resourceName ?? sink.name ?? "",
246 billingAccountId,
247 );
248 return {
249 name: sink.resourceName ?? (sinkId ? resourceName(account, sinkId) : ""),
250 sinkId,
251 billingAccountId: account,
252 destination: sink.destination ?? "",
253 filter: sink.filter,
254 description: parsed.description,
255 disabled: sink.disabled === true,
256 exclusions: exclusionsOf(sink.exclusions),
257 bigqueryOptions:
258 bq?.usePartitionedTables !== undefined
259 ? { usePartitionedTables: bq.usePartitionedTables }
260 : undefined,
261 writerIdentity: sink.writerIdentity,
262 createTime: sink.createTime,
263 updateTime: sink.updateTime,
264 };
265};
266
267const getByName = (name: string) =>
268 logging
269 .getBillingAccountsSinks({ sinkName: name })
270 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
271
272const toCreateBody = (
273 sinkId: string,
274 props: BillingSinkProps,
275 description: string,
276): logging.LogSink => ({
277 name: sinkId,
278 destination: props.destination,
279 filter: props.filter,
280 description,
281 disabled: props.disabled === true ? true : undefined,
282 exclusions: toExclusionsBody(props.exclusions),
283 bigqueryOptions: props.bigqueryOptions,
284});
285
286export const BillingSinkProvider = () =>
287 Provider.succeed(BillingSink, {
288 stables: ["name", "sinkId", "billingAccountId", "createTime"],
289
290 diff: Effect.fn(function* ({ news, olds, output }) {
291 if (!isResolved(news)) return undefined;
292 const previous = olds?.sinkId ?? output?.sinkId;
293 const idChanged =
294 previous !== undefined &&
295 news.sinkId !== undefined &&
296 news.sinkId !== previous;
297 const previousAccount =
298 olds?.billingAccountId ?? output?.billingAccountId;
299 const accountChanged =
300 previousAccount !== undefined &&
301 news.billingAccountId !== undefined &&
302 billingAccountIdOf(news.billingAccountId) !==
303 billingAccountIdOf(previousAccount);
304 if (!idChanged && !accountChanged) return undefined;
305 return { action: "replace" as const, deleteFirst: false };
306 }),
307
308 read: Effect.fn(function* ({ id, olds, output }) {
309 const billingAccountId = yield* resolveBillingAccountId(
310 olds?.billingAccountId,
311 output?.billingAccountId,
312 );
313 const sinkId = yield* toPhysicalId(id, olds?.sinkId, output?.sinkId, "s");
314 const name = output?.name ?? resourceName(billingAccountId, sinkId);
315 const existing = yield* getByName(name);
316 if (existing === undefined) return undefined;
317 const attrs = toAttrs(existing, billingAccountId);
318 const { labels } = parseDescription(existing.description);
319 return (yield* hasAlchemyLabels(id, labels)) ? attrs : Unowned(attrs);
320 }),
321
322 list: () =>
323 Effect.gen(function* () {
324 const env = yield* GcpEnvironment.current;
325 const billingAccountId = yield* lookupProjectBillingAccountId(
326 env.project,
327 );
328 if (billingAccountId === undefined) return [];
329 return yield* logging.listBillingAccountsSinks
330 .pages({
331 parent: billingAccountParent(billingAccountId),
332 pageSize: 1000,
333 })
334 .pipe(
335 Stream.flatMap((page) => Stream.fromIterable(page.sinks ?? [])),
336 Stream.filter((sink) => hasOwnershipMarker(sink.description)),
337 Stream.map((sink) => toAttrs(sink, billingAccountId)),
338 Stream.runCollect,
339 Effect.map((chunk) => Array.from(chunk)),
340 Effect.catchTag("NotFound", () =>
341 Effect.succeed([] as BillingSink["Attributes"][]),
342 ),
343 );
344 }),
345
346 reconcile: Effect.fn(function* ({ id, news, output }) {
347 const billingAccountId = yield* resolveBillingAccountId(
348 news.billingAccountId,
349 output?.billingAccountId,
350 );
351 const sinkId = yield* toPhysicalId(id, news.sinkId, output?.sinkId, "s");
352 const name = resourceName(billingAccountId, sinkId);
353 const ownership = yield* createInternalLabels(id);
354 const desiredDescription = encodeDescription(ownership, news.description);
355
356 let current = yield* getByName(output?.name ?? name);
357
358 if (current === undefined) {
359 const created = yield* logging
360 .createBillingAccountsSinks({
361 parent: billingAccountParent(billingAccountId),
362 uniqueWriterIdentity: news.uniqueWriterIdentity,
363 customWriterIdentity: news.customWriterIdentity,
364 body: toCreateBody(sinkId, news, desiredDescription),
365 })
366 .pipe(Effect.catchTag("Conflict", () => getByName(name)));
367 current = created ?? undefined;
368 }
369
370 if (current === undefined) {
371 return yield* new BillingSinkNotResolved({ name });
372 }
373
374 const desiredDisabled = news.disabled === true;
375 const desiredBq = news.bigqueryOptions?.usePartitionedTables === true;
376 const observedBq = current.bigqueryOptions?.usePartitionedTables === true;
377
378 const destinationChanged =
379 (current.destination ?? "") !== news.destination;
380 const filterChanged = (current.filter ?? "") !== (news.filter ?? "");
381 const descriptionChanged =
382 (current.description ?? "") !== desiredDescription;
383 const disabledChanged = (current.disabled === true) !== desiredDisabled;
384 const exclusionsChanged = !sameExclusions(
385 current.exclusions,
386 news.exclusions,
387 );
388 const bqChanged =
389 news.bigqueryOptions !== undefined && desiredBq !== observedBq;
390
391 const updateMask = [
392 destinationChanged ? "destination" : undefined,
393 filterChanged ? "filter" : undefined,
394 descriptionChanged ? "description" : undefined,
395 disabledChanged ? "disabled" : undefined,
396 exclusionsChanged ? "exclusions" : undefined,
397 bqChanged ? "bigqueryOptions.usePartitionedTables" : undefined,
398 ].filter((field): field is string => field !== undefined);
399
400 if (updateMask.length > 0) {
401 const body: logging.LogSink = {
402 name: sinkId,
403 destination: news.destination,
404 filter: news.filter ?? "",
405 description: desiredDescription,
406 disabled: desiredDisabled,
407 };
408 if (exclusionsChanged) {
409 body.exclusions = toExclusionsBody(news.exclusions) ?? [];
410 }
411 if (bqChanged) {
412 body.bigqueryOptions = {
413 usePartitionedTables: desiredBq,
414 };
415 }
416 current = yield* logging.patchBillingAccountsSinks({
417 sinkName: current.resourceName ?? name,
418 uniqueWriterIdentity: news.uniqueWriterIdentity,
419 customWriterIdentity: news.customWriterIdentity,
420 updateMask: updateMask.join(","),
421 body,
422 });
423 }
424
425 return toAttrs(current, billingAccountId);
426 }),
427
428 delete: Effect.fn(function* ({ output }) {
429 yield* logging
430 .deleteBillingAccountsSinks({ sinkName: output.name })
431 .pipe(Effect.catchTag("NotFound", () => Effect.void));
432 }),
433 });