1import * as vpclattice from "@distilled.cloud/aws/vpc-lattice";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import {
9 createInternalTags,
10 diffTags,
11 hasAlchemyTags,
12 tagRecord,
13} from "../../Tags.ts";
14import type { Providers } from "../Providers.ts";
15import { retryOnConflict } from "./internal.ts";
16
17/**
18 * Which traffic a service-network access log subscription captures.
19 */
20export type ServiceNetworkLogType = "SERVICE" | "RESOURCE";
21
22/** The delivery service encoded in a destination ARN (`arn:aws:<service>:…`). */
23const destinationType = (arn: string) => arn.split(":")[2] ?? "";
24
25/**
26 * Normalize a destination ARN for comparison — CloudWatch log-group ARNs
27 * round-trip through the API with a trailing `:*`.
28 */
29const normalizeDestinationArn = (arn: string) => arn.replace(/:\*$/, "");
30
31export interface AccessLogSubscriptionProps {
32 /**
33 * ID or ARN of the service network or lattice service whose traffic is
34 * logged. Immutable — changing it replaces the subscription.
35 */
36 resourceIdentifier: string;
37 /**
38 * ARN of the delivery destination: a CloudWatch log group, Firehose
39 * delivery stream, or S3 bucket. The destination can be updated in place,
40 * but changing the destination *type* (e.g. CloudWatch → S3) replaces the
41 * subscription.
42 */
43 destinationArn: string;
44 /**
45 * For service networks, whether to log traffic to lattice services or to
46 * shared VPC resources. Immutable — changing it replaces the subscription.
47 * @default "SERVICE"
48 */
49 serviceNetworkLogType?: ServiceNetworkLogType;
50 /**
51 * User-defined tags to apply to the subscription.
52 */
53 tags?: Record<string, string>;
54}
55
56export interface AccessLogSubscription extends Resource<
57 "AWS.VpcLattice.AccessLogSubscription",
58 AccessLogSubscriptionProps,
59 {
60 /**
61 * Service-assigned unique ID of the subscription.
62 */
63 accessLogSubscriptionId: string;
64 /**
65 * ARN of the subscription.
66 */
67 accessLogSubscriptionArn: string;
68 /**
69 * ID of the service network or service whose traffic is logged.
70 */
71 resourceId: string;
72 /**
73 * ARN of the service network or service whose traffic is logged.
74 */
75 resourceArn: string;
76 /**
77 * ARN of the delivery destination.
78 */
79 destinationArn: string;
80 /**
81 * Current tags reported for the subscription.
82 */
83 tags: Record<string, string>;
84 },
85 never,
86 Providers
87> {}
88
89/**
90 * An Amazon VPC Lattice access log subscription — delivers per-request
91 * access logs for a service network or lattice service to CloudWatch Logs,
92 * Kinesis Data Firehose, or S3.
93 *
94 * ### Creating Access Log Subscriptions
95 * **Example:** Log a Service Network to CloudWatch
96 * ```typescript
97 * const logs = yield* AccessLogSubscription("NetworkLogs", {
98 * resourceIdentifier: network.serviceNetworkId,
99 * destinationArn: logGroup.logGroupArn,
100 * });
101 * ```
102 *
103 * **Example:** Log a Service to S3
104 * ```typescript
105 * const logs = yield* AccessLogSubscription("ServiceLogs", {
106 * resourceIdentifier: service.serviceId,
107 * destinationArn: bucket.bucketArn,
108 * });
109 * ```
110 *
111 * @resource
112 */
113export const AccessLogSubscription = Resource<AccessLogSubscription>(
114 "AWS.VpcLattice.AccessLogSubscription",
115);
116
117export const AccessLogSubscriptionProvider = () =>
118 Provider.effect(
119 AccessLogSubscription,
120 Effect.gen(function* () {
121 const observe = (id: string) =>
122 vpclattice
123 .getAccessLogSubscription({ accessLogSubscriptionIdentifier: id })
124 .pipe(
125 Effect.catchTag("ResourceNotFoundException", () =>
126 Effect.succeed(undefined),
127 ),
128 );
129
130 // A resource can hold at most one subscription per destination type;
131 // recover the existing one for our destination's type.
132 const findByDestinationType = (
133 resourceIdentifier: string,
134 desiredDestinationArn: string,
135 ) =>
136 vpclattice.listAccessLogSubscriptions
137 .pages({ resourceIdentifier })
138 .pipe(
139 Stream.runCollect,
140 Effect.map((chunk) =>
141 Array.from(chunk)
142 .flatMap((page) => page.items ?? [])
143 .find(
144 (s) =>
145 destinationType(s.destinationArn) ===
146 destinationType(desiredDestinationArn),
147 ),
148 ),
149 Effect.catchTag("ResourceNotFoundException", () =>
150 Effect.succeed(undefined),
151 ),
152 );
153
154 const syncTags = Effect.fn(function* (
155 arn: string,
156 desiredTags: Record<string, string>,
157 ) {
158 const listed = yield* vpclattice.listTagsForResource({
159 resourceArn: arn,
160 });
161 const { removed, upsert } = diffTags(
162 tagRecord(listed.tags),
163 desiredTags,
164 );
165 if (upsert.length > 0) {
166 yield* vpclattice.tagResource({
167 resourceArn: arn,
168 tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
169 });
170 }
171 if (removed.length > 0) {
172 yield* vpclattice.untagResource({
173 resourceArn: arn,
174 tagKeys: removed,
175 });
176 }
177 });
178
179 return {
180 stables: [
181 "accessLogSubscriptionId",
182 "accessLogSubscriptionArn",
183 "resourceId",
184 "resourceArn",
185 ],
186 diff: Effect.fn(function* ({ olds, news }) {
187 if (!isResolved(news)) return;
188 if (
189 olds?.resourceIdentifier !== news.resourceIdentifier ||
190 (olds?.serviceNetworkLogType ?? undefined) !==
191 news.serviceNetworkLogType
192 ) {
193 return { action: "replace" } as const;
194 }
195 // The destination is mutable in place, but only within the same
196 // destination service (CloudWatch/Firehose/S3).
197 if (
198 olds !== undefined &&
199 destinationType(olds.destinationArn) !==
200 destinationType(news.destinationArn)
201 ) {
202 return { action: "replace" } as const;
203 }
204 }),
205 read: Effect.fn(function* ({ id, olds, output }) {
206 const subscription = output?.accessLogSubscriptionId
207 ? yield* observe(output.accessLogSubscriptionId)
208 : olds
209 ? yield* findByDestinationType(
210 olds.resourceIdentifier,
211 olds.destinationArn,
212 )
213 : undefined;
214 if (!subscription) return undefined;
215 const listed = yield* vpclattice.listTagsForResource({
216 resourceArn: subscription.arn,
217 });
218 const attrs = {
219 accessLogSubscriptionId: subscription.id,
220 accessLogSubscriptionArn: subscription.arn,
221 resourceId: subscription.resourceId,
222 resourceArn: subscription.resourceArn,
223 destinationArn: subscription.destinationArn,
224 tags: tagRecord(listed.tags),
225 };
226 return (yield* hasAlchemyTags(id, listed.tags))
227 ? attrs
228 : Unowned(attrs);
229 }),
230 reconcile: Effect.fn(function* ({ id, news, output, session }) {
231 const internalTags = yield* createInternalTags(id);
232 const desiredTags = { ...internalTags, ...news.tags };
233
234 // Observe — prefer the stable id cache, fall back to discovering
235 // the resource's subscription for our destination type. The create
236 // response carries the same identifying fields but no timestamps,
237 // so keep the local shape to the fields we actually use.
238 let subscription:
239 | {
240 id: string;
241 arn: string;
242 resourceId: string;
243 resourceArn: string;
244 destinationArn: string;
245 }
246 | undefined = output?.accessLogSubscriptionId
247 ? yield* observe(output.accessLogSubscriptionId)
248 : yield* findByDestinationType(
249 news.resourceIdentifier,
250 news.destinationArn,
251 );
252
253 // Ensure — create if missing. A ConflictException means a
254 // subscription for this destination type already exists (a race);
255 // recover it and converge below.
256 if (!subscription) {
257 const created = yield* retryOnConflict(
258 vpclattice.createAccessLogSubscription({
259 resourceIdentifier: news.resourceIdentifier,
260 destinationArn: news.destinationArn,
261 serviceNetworkLogType: news.serviceNetworkLogType,
262 tags: desiredTags,
263 }),
264 ).pipe(
265 Effect.catchTag("ConflictException", () =>
266 findByDestinationType(
267 news.resourceIdentifier,
268 news.destinationArn,
269 ),
270 ),
271 );
272 if (!created) {
273 return yield* Effect.fail(
274 new Error("Failed to create access log subscription"),
275 );
276 }
277 subscription = created;
278 }
279
280 // Sync destination — mutable within the same destination type.
281 if (
282 normalizeDestinationArn(subscription.destinationArn) !==
283 normalizeDestinationArn(news.destinationArn)
284 ) {
285 yield* vpclattice.updateAccessLogSubscription({
286 accessLogSubscriptionIdentifier: subscription.id,
287 destinationArn: news.destinationArn,
288 });
289 }
290
291 yield* syncTags(subscription.arn, desiredTags);
292
293 yield* session.note(subscription.arn);
294 return {
295 accessLogSubscriptionId: subscription.id,
296 accessLogSubscriptionArn: subscription.arn,
297 resourceId: subscription.resourceId,
298 resourceArn: subscription.resourceArn,
299 destinationArn: news.destinationArn,
300 tags: desiredTags,
301 };
302 }),
303 // Sub-resource: subscriptions are keyed by their service network /
304 // service and are removed with it, so nuke has nothing to enumerate.
305 list: () => Effect.succeed([] as AccessLogSubscription["Attributes"][]),
306 delete: Effect.fn(function* ({ output }) {
307 yield* retryOnConflict(
308 vpclattice.deleteAccessLogSubscription({
309 accessLogSubscriptionIdentifier: output.accessLogSubscriptionId,
310 }),
311 ).pipe(
312 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
313 );
314 }),
315 };
316 }),
317 );