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 { createPhysicalName } from "../../PhysicalName.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 * The type of assets a data set can hold. Immutable — changing it replaces
16 * the data set.
17 */
18export type DataSetAssetType =
19 | "S3_SNAPSHOT"
20 | "REDSHIFT_DATA_SHARE"
21 | "API_GATEWAY_API"
22 | "S3_DATA_ACCESS"
23 | "LAKE_FORMATION_DATA_PERMISSION";
24
25export interface DataSetProps {
26 /**
27 * Name of the data set. If omitted, a unique name is generated from the
28 * app, stage, and logical ID. The name is mutable — changing it updates the
29 * data set in place.
30 */
31 name?: string;
32 /**
33 * The type of assets the data set holds. `S3_SNAPSHOT` (static files
34 * snapshotted from S3) is the most common.
35 * Immutable — changing it replaces the data set.
36 * @default "S3_SNAPSHOT"
37 */
38 assetType?: DataSetAssetType;
39 /**
40 * A description of the data set, shown to subscribers on AWS Data Exchange.
41 * @default the data set name
42 */
43 description?: string;
44 /**
45 * Tags to apply to the data set. Merged with internal Alchemy tags.
46 */
47 tags?: Record<string, string>;
48}
49
50export interface DataSet extends Resource<
51 "AWS.DataExchange.DataSet",
52 DataSetProps,
53 {
54 /**
55 * The unique identifier of the data set.
56 */
57 dataSetId: string;
58 /**
59 * The ARN of the data set.
60 */
61 dataSetArn: string;
62 /**
63 * The name of the data set.
64 */
65 name: string;
66 /**
67 * The type of assets the data set holds.
68 */
69 assetType: string;
70 /**
71 * The origin of the data set — `OWNED` for data sets created by this
72 * account, `ENTITLED` for data sets obtained through a subscription.
73 */
74 origin: string;
75 },
76 never,
77 Providers
78> {}
79
80/**
81 * An AWS Data Exchange data set — the top-level container that data providers
82 * publish revisions of data into. An owned data set holds revisions, each of
83 * which holds assets (e.g. S3 snapshot files) that subscribers receive.
84 *
85 * ### Creating Data Sets
86 * **Example:** Basic S3-snapshot data set
87 * ```typescript
88 * import * as AWS from "alchemy/AWS";
89 *
90 * const dataSet = yield* AWS.DataExchange.DataSet("Prices", {
91 * description: "Daily commodity price snapshots",
92 * });
93 * ```
94 *
95 * **Example:** Named data set with tags
96 * ```typescript
97 * const dataSet = yield* AWS.DataExchange.DataSet("Prices", {
98 * name: "commodity-prices",
99 * assetType: "S3_SNAPSHOT",
100 * description: "Daily commodity price snapshots",
101 * tags: { team: "data" },
102 * });
103 * ```
104 *
105 * ### Publishing Revisions
106 * **Example:** Add a revision to a data set
107 * ```typescript
108 * const revision = yield* AWS.DataExchange.Revision("PricesV1", {
109 * dataSetId: dataSet.dataSetId,
110 * comment: "Initial snapshot",
111 * });
112 * ```
113 *
114 * @resource
115 */
116export const DataSet = Resource<DataSet>("AWS.DataExchange.DataSet");
117
118export const DataSetProvider = () =>
119 Provider.effect(
120 DataSet,
121 Effect.gen(function* () {
122 const createName = Effect.fn(function* (id: string, props: DataSetProps) {
123 return props.name ?? (yield* createPhysicalName({ id, maxLength: 98 }));
124 });
125
126 /** Get a data set by id; typed not-found → undefined. */
127 const getById = Effect.fn(function* (dataSetId: string) {
128 return yield* dataexchange
129 .getDataSet({ DataSetId: dataSetId })
130 .pipe(
131 Effect.catchTag("ResourceNotFoundException", () =>
132 Effect.succeed(undefined),
133 ),
134 );
135 });
136
137 /**
138 * Scan owned data sets for one with the given name. Data set ids are
139 * server-generated, so recovery from a lost output (state-persistence
140 * failure) goes through the deterministic physical name.
141 */
142 const findByName = Effect.fn(function* (name: string) {
143 const head = yield* dataexchange.listDataSets
144 .items({ Origin: "OWNED" })
145 .pipe(
146 Stream.filter((dataSet) => dataSet.Name === name),
147 Stream.runHead,
148 );
149 if (head._tag === "None") return undefined;
150 return yield* getById(head.value.Id);
151 });
152
153 const toAttrs = (dataSet: {
154 Id?: string;
155 Arn?: string;
156 Name?: string;
157 AssetType?: string;
158 Origin?: string;
159 }): DataSet["Attributes"] => ({
160 dataSetId: dataSet.Id!,
161 dataSetArn: dataSet.Arn!,
162 name: dataSet.Name!,
163 assetType: dataSet.AssetType!,
164 origin: dataSet.Origin!,
165 });
166
167 return {
168 stables: ["dataSetId", "dataSetArn", "assetType", "origin"],
169
170 read: Effect.fn(function* ({ id, olds, output }) {
171 const dataSet = output?.dataSetId
172 ? yield* getById(output.dataSetId)
173 : yield* findByName(yield* createName(id, olds ?? {}));
174 if (dataSet === undefined) return undefined;
175 const attrs = toAttrs(dataSet);
176 const tags = yield* readDataExchangeTags(attrs.dataSetArn);
177 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
178 }),
179
180 diff: Effect.fn(function* ({ news, olds }) {
181 if (!isResolved(news)) return undefined;
182 // The asset type is immutable; name and description update in place.
183 if (
184 (news.assetType ?? "S3_SNAPSHOT") !==
185 (olds.assetType ?? "S3_SNAPSHOT")
186 ) {
187 return { action: "replace" } as const;
188 }
189 }),
190
191 reconcile: Effect.fn(function* ({ id, news, output, session }) {
192 const name = yield* createName(id, news);
193 const assetType = news.assetType ?? "S3_SNAPSHOT";
194 const description = news.description ?? name;
195 const internalTags = yield* createInternalTags(id);
196 const desiredTags = { ...news.tags, ...internalTags };
197
198 // 1. Observe — the output id is only a cache; fall back to the
199 // deterministic name when it's missing or stale.
200 let dataSet = output?.dataSetId
201 ? yield* getById(output.dataSetId)
202 : undefined;
203 if (dataSet === undefined) {
204 dataSet = yield* findByName(name);
205 }
206
207 // 2. Ensure — create when missing. Data set names are not unique in
208 // DataExchange, so there is no AlreadyExists race to tolerate.
209 if (dataSet === undefined) {
210 dataSet = yield* dataexchange.createDataSet({
211 AssetType: assetType,
212 Name: name,
213 Description: description,
214 Tags: desiredTags,
215 });
216 }
217
218 // 3. Sync — name and description are mutable in place.
219 if (dataSet.Name !== name || dataSet.Description !== description) {
220 dataSet = yield* dataexchange.updateDataSet({
221 DataSetId: dataSet.Id!,
222 Name: name,
223 Description: description,
224 });
225 }
226
227 // 3b. Sync tags against OBSERVED cloud tags (adoption-safe).
228 yield* syncDataExchangeTags(dataSet.Arn!, desiredTags);
229
230 yield* session.note(dataSet.Arn!);
231 return {
232 dataSetId: dataSet.Id!,
233 dataSetArn: dataSet.Arn!,
234 name,
235 assetType: dataSet.AssetType ?? assetType,
236 origin: dataSet.Origin ?? "OWNED",
237 };
238 }),
239
240 delete: Effect.fn(function* ({ output }) {
241 yield* dataexchange
242 .deleteDataSet({ DataSetId: output.dataSetId })
243 .pipe(
244 // A revision-deletion race can transiently reject the delete;
245 // the engine deletes revisions first, so retry briefly.
246 Effect.retry({
247 while: (e) => e._tag === "ConflictException",
248 schedule: Schedule.max([
249 Schedule.fixed("2 seconds"),
250 Schedule.recurs(10),
251 ]),
252 }),
253 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
254 );
255 }),
256
257 list: () =>
258 dataexchange.listDataSets.items({}).pipe(
259 Stream.map((entry) => ({
260 dataSetId: entry.Id,
261 dataSetArn: entry.Arn,
262 name: entry.Name,
263 assetType: entry.AssetType,
264 origin: entry.Origin,
265 })),
266 Stream.runCollect,
267 Effect.map((chunk) => Array.from(chunk)),
268 ),
269 };
270 }),
271 );