1import * as dataexchange from "@distilled.cloud/aws/dataexchange";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
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 { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
10import type { Providers } from "../Providers.ts";
11import { readDataExchangeTags, syncDataExchangeTags } from "./internal.ts";
12
13export interface RevisionProps {
14 /**
15 * The unique identifier of the data set the revision belongs to.
16 * Changing it replaces the revision.
17 */
18 dataSetId: string;
19 /**
20 * An optional comment about the revision (up to 16,348 characters).
21 * Mutable in place.
22 */
23 comment?: string;
24 /**
25 * Whether the revision is finalized. A revision can only be finalized once
26 * it contains at least one asset (imported via a DataExchange job) —
27 * finalizing an empty revision fails with a `ValidationException`.
28 * @default false
29 */
30 finalized?: boolean;
31 /**
32 * Tags to apply to the revision. Merged with internal Alchemy tags.
33 */
34 tags?: Record<string, string>;
35}
36
37export interface Revision extends Resource<
38 "AWS.DataExchange.Revision",
39 RevisionProps,
40 {
41 /**
42 * The unique identifier of the revision.
43 */
44 revisionId: string;
45 /**
46 * The ARN of the revision.
47 */
48 revisionArn: string;
49 /**
50 * The unique identifier of the data set the revision belongs to.
51 */
52 dataSetId: string;
53 /**
54 * Whether the revision is finalized.
55 */
56 finalized: boolean;
57 },
58 never,
59 Providers
60> {}
61
62/**
63 * A revision of an AWS Data Exchange data set — a versioned container of
64 * assets. Providers import assets into a revision (via DataExchange jobs) and
65 * then finalize it to make the snapshot available to subscribers.
66 *
67 * ### Creating Revisions
68 * **Example:** Revision with a comment
69 * ```typescript
70 * import * as AWS from "alchemy/AWS";
71 *
72 * const dataSet = yield* AWS.DataExchange.DataSet("Prices", {
73 * description: "Daily commodity price snapshots",
74 * });
75 *
76 * const revision = yield* AWS.DataExchange.Revision("PricesV1", {
77 * dataSetId: dataSet.dataSetId,
78 * comment: "Initial snapshot",
79 * });
80 * ```
81 *
82 * ### Finalizing
83 * Once assets have been imported into the revision (via a DataExchange
84 * import job), flip `finalized` to publish it. Finalizing an empty revision
85 * fails.
86 *
87 * **Example:** Finalize a revision that has assets
88 * ```typescript
89 * const revision = yield* AWS.DataExchange.Revision("PricesV1", {
90 * dataSetId: dataSet.dataSetId,
91 * comment: "Initial snapshot",
92 * finalized: true,
93 * });
94 * ```
95 *
96 * @resource
97 */
98export const Revision = Resource<Revision>("AWS.DataExchange.Revision");
99
100export const RevisionProvider = () =>
101 Provider.effect(
102 Revision,
103 Effect.gen(function* () {
104 /** Get a revision by ids; typed not-found → undefined. */
105 const getById = Effect.fn(function* (
106 dataSetId: string,
107 revisionId: string,
108 ) {
109 return yield* dataexchange
110 .getRevision({ DataSetId: dataSetId, RevisionId: revisionId })
111 .pipe(
112 Effect.catchTag("ResourceNotFoundException", () =>
113 Effect.succeed(undefined),
114 ),
115 );
116 });
117
118 /**
119 * Scan a data set's revisions for one carrying our ownership tags.
120 * Revision ids are server-generated, so recovery from a lost output
121 * (state-persistence failure) goes through per-revision tag inspection.
122 */
123 const findByTags = Effect.fn(function* (id: string, dataSetId: string) {
124 return yield* dataexchange.listDataSetRevisions
125 .items({ DataSetId: dataSetId })
126 .pipe(
127 Stream.mapEffect(
128 Effect.fn(function* (entry) {
129 const tags = yield* readDataExchangeTags(entry.Arn);
130 return (yield* hasAlchemyTags(id, tags)) ? entry : undefined;
131 }),
132 ),
133 Stream.filter((entry) => entry !== undefined),
134 Stream.runHead,
135 Effect.map((head) =>
136 head._tag === "Some" ? head.value : undefined,
137 ),
138 Effect.catchTag("ResourceNotFoundException", () =>
139 Effect.succeed(undefined),
140 ),
141 );
142 });
143
144 return {
145 stables: ["revisionId", "revisionArn", "dataSetId"],
146
147 read: Effect.fn(function* ({ id, olds, output }) {
148 const dataSetId = output?.dataSetId ?? olds?.dataSetId;
149 if (dataSetId === undefined) return undefined;
150 const revision = output?.revisionId
151 ? yield* getById(dataSetId, output.revisionId)
152 : yield* findByTags(id, dataSetId);
153 if (revision === undefined) return undefined;
154 const attrs = {
155 revisionId: revision.Id!,
156 revisionArn: revision.Arn!,
157 dataSetId,
158 finalized: revision.Finalized ?? false,
159 };
160 const tags = yield* readDataExchangeTags(attrs.revisionArn);
161 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
162 }),
163
164 diff: Effect.fn(function* ({ news, olds }) {
165 if (!isResolved(news)) return undefined;
166 if ((news.dataSetId as string) !== (olds.dataSetId as string)) {
167 return { action: "replace" } as const;
168 }
169 }),
170
171 reconcile: Effect.fn(function* ({ id, news, output, session }) {
172 const dataSetId = news.dataSetId as string;
173 const internalTags = yield* createInternalTags(id);
174 const desiredTags = { ...news.tags, ...internalTags };
175 const desiredFinalized = news.finalized ?? false;
176
177 // 1. Observe — output ids are only a cache; fall back to a
178 // tag-ownership scan of the data set's revisions.
179 let revision = output?.revisionId
180 ? yield* getById(dataSetId, output.revisionId)
181 : undefined;
182 if (revision === undefined) {
183 revision = yield* findByTags(id, dataSetId);
184 }
185
186 // 2. Ensure — create when missing.
187 if (revision === undefined) {
188 revision = yield* dataexchange.createRevision({
189 DataSetId: dataSetId,
190 Comment: news.comment,
191 Tags: desiredTags,
192 });
193 }
194
195 // 3. Sync — comment and finalized are mutable in one call. Only
196 // call the API on an actual delta.
197 if (
198 (revision.Comment ?? undefined) !== (news.comment ?? undefined) ||
199 (revision.Finalized ?? false) !== desiredFinalized
200 ) {
201 revision = yield* dataexchange.updateRevision({
202 DataSetId: dataSetId,
203 RevisionId: revision.Id!,
204 Comment: news.comment,
205 Finalized: desiredFinalized,
206 });
207 }
208
209 // 3b. Sync tags against OBSERVED cloud tags (adoption-safe).
210 yield* syncDataExchangeTags(revision.Arn!, desiredTags);
211
212 yield* session.note(revision.Arn!);
213 return {
214 revisionId: revision.Id!,
215 revisionArn: revision.Arn!,
216 dataSetId,
217 finalized: revision.Finalized ?? false,
218 };
219 }),
220
221 delete: Effect.fn(function* ({ output }) {
222 yield* dataexchange
223 .deleteRevision({
224 DataSetId: output.dataSetId,
225 RevisionId: output.revisionId,
226 })
227 .pipe(
228 // Asset deletions inside the revision can transiently conflict.
229 Effect.retry({
230 while: (e) => e._tag === "ConflictException",
231 schedule: Schedule.max([
232 Schedule.fixed("2 seconds"),
233 Schedule.recurs(10),
234 ]),
235 }),
236 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
237 );
238 }),
239
240 // Revisions are sub-resources keyed by their parent data set; there
241 // is no account-level enumeration into a parentless Attributes shape.
242 list: () => Effect.succeed([]),
243 };
244 }),
245 );