1import * as dataexchange from "@distilled.cloud/aws/dataexchange";
2import * as Effect from "effect/Effect";
3import * as Result from "effect/Result";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
11import type { Providers } from "../Providers.ts";
12import { readDataExchangeTags, syncDataExchangeTags } from "./internal.ts";
13
14/**
15 * Server-side encryption for auto-exported revisions.
16 */
17export interface EventActionEncryption {
18 /**
19 * The encryption type: `aws:kms` (with `kmsKeyArn`) or `AES256`.
20 */
21 type: "aws:kms" | "AES256";
22 /**
23 * The ARN of the KMS key used to encrypt exported objects when `type` is
24 * `aws:kms`.
25 */
26 kmsKeyArn?: string;
27}
28
29/**
30 * The auto-export destination of an event action.
31 */
32export interface EventActionExportRevisionToS3 {
33 /**
34 * The S3 bucket auto-exported revisions are written to. The bucket policy
35 * must allow the AWS Data Exchange service principal to write to it.
36 */
37 bucket: string;
38 /**
39 * Pattern for naming exported objects, built from `${Revision.CreatedAt}`
40 * and `${Asset.Name}` variables.
41 * @default "${Asset.Name}"
42 */
43 keyPattern?: string;
44 /**
45 * Server-side encryption applied to exported objects.
46 */
47 encryption?: EventActionEncryption;
48}
49
50export interface EventActionProps {
51 /**
52 * The id of the ENTITLED data set whose `RevisionPublished` event triggers
53 * the action. Event actions can only be created for entitled data sets
54 * (subscriptions or accepted data grants), not for owned data sets.
55 * Immutable — changing it replaces the event action.
56 */
57 dataSetId: string;
58 /**
59 * Where to auto-export newly published revisions. Mutable in place.
60 */
61 exportRevisionToS3: EventActionExportRevisionToS3;
62 /**
63 * Tags to apply to the event action. Merged with internal Alchemy tags.
64 */
65 tags?: Record<string, string>;
66}
67
68export interface EventAction extends Resource<
69 "AWS.DataExchange.EventAction",
70 EventActionProps,
71 {
72 /**
73 * The unique identifier of the event action.
74 */
75 eventActionId: string;
76 /**
77 * The ARN of the event action.
78 */
79 eventActionArn: string;
80 /**
81 * The id of the entitled data set the event action watches.
82 */
83 dataSetId: string;
84 },
85 never,
86 Providers
87> {}
88
89/**
90 * An AWS Data Exchange event action — an auto-export rule that copies every
91 * newly published revision of an ENTITLED data set into your S3 bucket the
92 * moment the provider publishes it. This is the subscriber-side automation
93 * primitive: subscribe to a product (or accept a data grant), attach an
94 * event action, and fresh data lands in your bucket with no polling.
95 *
96 * Event actions require an entitled data set — creating one against an
97 * owned data set fails with a `ValidationException`.
98 *
99 * ### Auto-Exporting Entitled Data
100 * **Example:** Export new revisions to S3
101 * ```typescript
102 * import * as AWS from "alchemy/AWS";
103 *
104 * const landing = yield* AWS.S3.Bucket("Landing", {});
105 *
106 * const autoExport = yield* AWS.DataExchange.EventAction("AutoExport", {
107 * dataSetId: entitledDataSetId,
108 * exportRevisionToS3: { bucket: landing.bucketName },
109 * });
110 * ```
111 *
112 * **Example:** Encrypted export with a key pattern
113 * ```typescript
114 * const autoExport = yield* AWS.DataExchange.EventAction("AutoExport", {
115 * dataSetId: entitledDataSetId,
116 * exportRevisionToS3: {
117 * bucket: landing.bucketName,
118 * keyPattern: "prices/${Revision.CreatedAt}/${Asset.Name}",
119 * encryption: { type: "aws:kms", kmsKeyArn: key.keyArn },
120 * },
121 * });
122 * ```
123 *
124 * @resource
125 */
126export const EventAction = Resource<EventAction>(
127 "AWS.DataExchange.EventAction",
128);
129
130const toWireAction = (props: EventActionProps): dataexchange.Action => ({
131 ExportRevisionToS3: {
132 RevisionDestination: {
133 Bucket: props.exportRevisionToS3.bucket,
134 KeyPattern: props.exportRevisionToS3.keyPattern,
135 },
136 Encryption: props.exportRevisionToS3.encryption
137 ? {
138 Type: props.exportRevisionToS3.encryption.type,
139 KmsKeyArn: props.exportRevisionToS3.encryption.kmsKeyArn,
140 }
141 : undefined,
142 },
143});
144
145const sameAction = (
146 observed: dataexchange.Action | undefined,
147 desired: dataexchange.Action,
148): boolean => {
149 const a = observed?.ExportRevisionToS3;
150 const b = desired.ExportRevisionToS3;
151 return (
152 a !== undefined &&
153 b !== undefined &&
154 a.RevisionDestination.Bucket === b.RevisionDestination.Bucket &&
155 (a.RevisionDestination.KeyPattern ?? undefined) ===
156 (b.RevisionDestination.KeyPattern ?? undefined) &&
157 (a.Encryption?.Type ?? undefined) === (b.Encryption?.Type ?? undefined) &&
158 (a.Encryption?.KmsKeyArn ?? undefined) ===
159 (b.Encryption?.KmsKeyArn ?? undefined)
160 );
161};
162
163export const EventActionProvider = () =>
164 Provider.effect(
165 EventAction,
166 Effect.gen(function* () {
167 /** Get an event action by id; typed not-found → undefined. */
168 const getById = Effect.fn(function* (eventActionId: string) {
169 return yield* dataexchange
170 .getEventAction({ EventActionId: eventActionId })
171 .pipe(
172 Effect.catchTag("ResourceNotFoundException", () =>
173 Effect.succeed(undefined),
174 ),
175 );
176 });
177
178 /**
179 * Scan the data set's event actions for one carrying our ownership
180 * tags. Event action ids are server-generated, so recovery from a
181 * lost output (state-persistence failure) goes through per-action tag
182 * inspection.
183 */
184 const findByTags = Effect.fn(function* (id: string, dataSetId: string) {
185 return yield* dataexchange.listEventActions
186 .items({ EventSourceId: dataSetId })
187 .pipe(
188 Stream.mapEffect(
189 Effect.fn(function* (entry) {
190 const tags = yield* readDataExchangeTags(entry.Arn);
191 return (yield* hasAlchemyTags(id, tags)) ? entry : undefined;
192 }),
193 ),
194 Stream.filter((entry) => entry !== undefined),
195 Stream.runHead,
196 Effect.map((head) =>
197 head._tag === "Some" ? head.value : undefined,
198 ),
199 Effect.catchTag("ResourceNotFoundException", () =>
200 Effect.succeed(undefined),
201 ),
202 );
203 });
204
205 return {
206 stables: ["eventActionId", "eventActionArn", "dataSetId"],
207
208 read: Effect.fn(function* ({ id, olds, output }) {
209 const dataSetId = output?.dataSetId ?? olds?.dataSetId;
210 const eventAction = output?.eventActionId
211 ? yield* getById(output.eventActionId)
212 : dataSetId !== undefined
213 ? yield* findByTags(id, dataSetId)
214 : undefined;
215 if (eventAction === undefined) return undefined;
216 const attrs = {
217 eventActionId: eventAction.Id!,
218 eventActionArn: eventAction.Arn!,
219 dataSetId:
220 eventAction.Event?.RevisionPublished?.DataSetId ?? dataSetId!,
221 };
222 const tags = yield* readDataExchangeTags(attrs.eventActionArn);
223 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
224 }),
225
226 diff: Effect.fn(function* ({ news, olds }) {
227 if (!isResolved(news)) return undefined;
228 // The watched data set is the event action's identity.
229 if ((news.dataSetId as string) !== (olds.dataSetId as string)) {
230 return { action: "replace" } as const;
231 }
232 }),
233
234 reconcile: Effect.fn(function* ({ id, news, output, session }) {
235 const dataSetId = news.dataSetId as string;
236 const desiredAction = toWireAction(news);
237 const internalTags = yield* createInternalTags(id);
238 const desiredTags = { ...news.tags, ...internalTags };
239
240 // 1. Observe — output ids are only a cache; fall back to a
241 // tag-ownership scan of the data set's event actions.
242 let eventAction = output?.eventActionId
243 ? yield* getById(output.eventActionId)
244 : undefined;
245 if (eventAction === undefined) {
246 eventAction = yield* findByTags(id, dataSetId);
247 }
248
249 // 2. Ensure — create when missing.
250 if (eventAction === undefined) {
251 eventAction = yield* dataexchange.createEventAction({
252 Action: desiredAction,
253 Event: { RevisionPublished: { DataSetId: dataSetId } },
254 Tags: desiredTags,
255 });
256 }
257
258 // 3. Sync — the export destination is mutable in place. Only call
259 // the API on an actual delta.
260 if (!sameAction(eventAction.Action, desiredAction)) {
261 eventAction = yield* dataexchange.updateEventAction({
262 EventActionId: eventAction.Id!,
263 Action: desiredAction,
264 });
265 }
266
267 // 3b. Sync tags against OBSERVED cloud tags (adoption-safe).
268 yield* syncDataExchangeTags(eventAction.Arn!, desiredTags);
269
270 yield* session.note(eventAction.Arn!);
271 return {
272 eventActionId: eventAction.Id!,
273 eventActionArn: eventAction.Arn!,
274 dataSetId,
275 };
276 }),
277
278 delete: Effect.fn(function* ({ output }) {
279 yield* dataexchange
280 .deleteEventAction({ EventActionId: output.eventActionId })
281 .pipe(
282 // Deletion is idempotent — a missing event action is success.
283 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
284 Effect.retry({
285 while: (e): boolean => e._tag === "ThrottlingException",
286 schedule: Schedule.max([
287 Schedule.fixed("2 seconds"),
288 Schedule.recurs(5),
289 ]),
290 }),
291 );
292 }),
293
294 list: () =>
295 dataexchange.listEventActions.items({}).pipe(
296 // RevisionPublished is the only event type Data Exchange supports;
297 // entries without it carry no data set identity, so skip them
298 // rather than fabricate a dataSetId.
299 Stream.filterMap((entry) =>
300 entry.Event.RevisionPublished !== undefined
301 ? Result.succeed({
302 eventActionId: entry.Id,
303 eventActionArn: entry.Arn,
304 dataSetId: entry.Event.RevisionPublished.DataSetId,
305 })
306 : Result.failVoid,
307 ),
308 Stream.runCollect,
309 Effect.map((chunk) => Array.from(chunk)),
310 ),
311 };
312 }),
313 );