1import * as DataExchange from "@/AWS/DataExchange";
2import * as Lambda from "@/AWS/Lambda";
3import * as S3 from "@/AWS/S3";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Schedule from "effect/Schedule";
8import * as Stream from "effect/Stream";
9import { HttpServerRequest } from "effect/http/HttpServerRequest";
10import * as HttpServerResponse from "effect/http/HttpServerResponse";
11import path from "pathe";
12
13const main = path.resolve(import.meta.dirname, "handler.ts");
14
15const CSV = "date,price\n2026-07-14,42.5\n";
16const KEY = "prices.csv";
17
18export class DataExchangeTestFunction extends Lambda.Function<Lambda.Function>()(
19 "DataExchangeTestFunction",
20) {}
21
22export default DataExchangeTestFunction.make(
23 {
24 main,
25 functionUrl: true,
26 // The /import route runs a full import job (create → start → poll to
27 // COMPLETED), which takes tens of seconds.
28 timeout: Duration.minutes(2),
29 },
30 Effect.gen(function* () {
31 // The owned data set + draft revision the scoped bindings are bound to,
32 // and the bucket the import job reads from.
33 const dataSet = yield* DataExchange.DataSet("BindingDataSet", {
34 description: "alchemy dataexchange bindings fixture",
35 });
36 const revision = yield* DataExchange.Revision("BindingRevision", {
37 dataSetId: dataSet.dataSetId,
38 comment: "bindings fixture revision",
39 });
40 const bucket = yield* S3.Bucket("BindingBucket", { forceDestroy: true });
41
42 // Event source: subscribe the host to Data Exchange revision events.
43 // The deploy proves the EventBridge rule + invoke permission wiring.
44 yield* DataExchange.consumeDataSetEvents(
45 { kinds: ["revision-published", "revision-revoked"] },
46 (events) =>
47 Stream.runForEach(events, (event) =>
48 Effect.log(`dataexchange event: ${event["detail-type"]}`),
49 ),
50 );
51
52 // Data-set-scoped bindings.
53 const getDataSet = yield* DataExchange.GetDataSet(dataSet);
54 const listDataSetRevisions =
55 yield* DataExchange.ListDataSetRevisions(dataSet);
56 const sendDataSetNotification =
57 yield* DataExchange.SendDataSetNotification(dataSet);
58 // Revision-scoped bindings.
59 const getRevision = yield* DataExchange.GetRevision(revision);
60 const listRevisionAssets = yield* DataExchange.ListRevisionAssets(revision);
61 const getAsset = yield* DataExchange.GetAsset(revision);
62 // Account-scoped bindings.
63 const listDataSets = yield* DataExchange.ListDataSets();
64 const createJob = yield* DataExchange.CreateJob();
65 const startJob = yield* DataExchange.StartJob();
66 const getJob = yield* DataExchange.GetJob();
67 const listJobs = yield* DataExchange.ListJobs();
68 const listDataGrants = yield* DataExchange.ListDataGrants();
69 const listReceivedDataGrants = yield* DataExchange.ListReceivedDataGrants();
70 const listEventActions = yield* DataExchange.ListEventActions();
71 // S3 access for the import job: the Lambda writes the source object and
72 // the Data Exchange import job reads it with the caller's permissions.
73 const putObject = yield* S3.PutObject(bucket);
74 const getObject = yield* S3.GetObject(bucket);
75
76 const bound = {
77 getDataSet,
78 listDataSetRevisions,
79 sendDataSetNotification,
80 getRevision,
81 listRevisionAssets,
82 getAsset,
83 listDataSets,
84 createJob,
85 startJob,
86 getJob,
87 listJobs,
88 listDataGrants,
89 listReceivedDataGrants,
90 listEventActions,
91 putObject,
92 getObject,
93 };
94
95 const BucketName = yield* bucket.bucketName;
96
97 return {
98 fetch: Effect.gen(function* () {
99 const request = yield* HttpServerRequest;
100 const url = new URL(request.originalUrl);
101 const pathname = url.pathname;
102
103 if (request.method === "GET" && pathname === "/bindings") {
104 return yield* HttpServerResponse.json({
105 bound: Object.keys(bound),
106 });
107 }
108
109 // Data-set-scoped read: the DataSetId is injected from the binding.
110 if (request.method === "GET" && pathname === "/data-set") {
111 const detail = yield* getDataSet();
112 return yield* HttpServerResponse.json({
113 id: detail.Id,
114 name: detail.Name,
115 assetType: detail.AssetType,
116 origin: detail.Origin,
117 });
118 }
119
120 if (request.method === "GET" && pathname === "/revisions") {
121 const { Revisions } = yield* listDataSetRevisions();
122 return yield* HttpServerResponse.json({
123 ids: (Revisions ?? []).map((entry) => entry.Id),
124 });
125 }
126
127 // Revision-scoped read: DataSetId + RevisionId are injected.
128 if (request.method === "GET" && pathname === "/revision") {
129 const detail = yield* getRevision();
130 return yield* HttpServerResponse.json({
131 id: detail.Id,
132 finalized: detail.Finalized ?? false,
133 });
134 }
135
136 if (request.method === "GET" && pathname === "/assets") {
137 const { Assets } = yield* listRevisionAssets();
138 return yield* HttpServerResponse.json({
139 count: (Assets ?? []).length,
140 });
141 }
142
143 // Account-level list, filtered down to owned data sets.
144 if (request.method === "GET" && pathname === "/data-sets") {
145 const { DataSets } = yield* listDataSets({ Origin: "OWNED" });
146 return yield* HttpServerResponse.json({
147 ids: (DataSets ?? []).map((entry) => entry.Id),
148 });
149 }
150
151 // Provider-generated notification about the bound data set.
152 if (request.method === "POST" && pathname === "/notify") {
153 const result = yield* sendDataSetNotification({
154 Type: "DATA_UPDATE",
155 Comment: "alchemy bindings fixture notification",
156 }).pipe(Effect.result);
157 return yield* HttpServerResponse.json({
158 ok: result._tag === "Success",
159 error: result._tag === "Failure" ? result.failure._tag : undefined,
160 message:
161 result._tag === "Failure" ? String(result.failure) : undefined,
162 });
163 }
164
165 // Full import flow: put a CSV in S3, import it into the bound
166 // revision via a job, poll the job to COMPLETED, then read the
167 // imported asset back through the revision-scoped bindings. Typed
168 // failures are surfaced as JSON so the test can report the cause.
169 if (request.method === "POST" && pathname === "/import") {
170 const flow = Effect.gen(function* () {
171 yield* putObject({ Key: KEY, Body: CSV, ContentType: "text/csv" });
172
173 const dataSetDetail = yield* getDataSet();
174 const revisionDetail = yield* getRevision();
175 const job = yield* createJob({
176 Type: "IMPORT_ASSETS_FROM_S3",
177 Details: {
178 ImportAssetsFromS3: {
179 DataSetId: dataSetDetail.Id!,
180 RevisionId: revisionDetail.Id!,
181 AssetSources: [{ Bucket: yield* BucketName, Key: KEY }],
182 },
183 },
184 });
185 yield* startJob({ JobId: job.Id! });
186 const done = yield* getJob({ JobId: job.Id! }).pipe(
187 Effect.repeat({
188 schedule: Schedule.spaced("2 seconds"),
189 until: (j): boolean =>
190 j.State === "COMPLETED" ||
191 j.State === "ERROR" ||
192 j.State === "CANCELLED",
193 times: 40,
194 }),
195 );
196
197 const { Assets } = yield* listRevisionAssets();
198 const assetId = (Assets ?? [])[0]?.Id;
199 const asset =
200 assetId !== undefined
201 ? yield* getAsset({ AssetId: assetId })
202 : undefined;
203 return {
204 jobState: done.State,
205 jobErrors: done.Errors,
206 assetCount: (Assets ?? []).length,
207 assetName: asset?.Name,
208 };
209 });
210 const result = yield* flow.pipe(Effect.result);
211 return result._tag === "Success"
212 ? yield* HttpServerResponse.json(result.success)
213 : yield* HttpServerResponse.json({
214 error: result.failure._tag,
215 message: String(result.failure),
216 });
217 }
218
219 if (request.method === "GET" && pathname === "/jobs") {
220 const dataSetDetail = yield* getDataSet();
221 const { Jobs } = yield* listJobs({ DataSetId: dataSetDetail.Id! });
222 return yield* HttpServerResponse.json({
223 states: (Jobs ?? []).map((entry) => entry.State),
224 });
225 }
226
227 // Account-level data grant + event action enumerations.
228 if (request.method === "GET" && pathname === "/grants") {
229 const sent = yield* listDataGrants();
230 const received = yield* listReceivedDataGrants();
231 return yield* HttpServerResponse.json({
232 sent: (sent.DataGrantSummaries ?? []).length,
233 received: (received.DataGrantSummaries ?? []).length,
234 });
235 }
236
237 if (request.method === "GET" && pathname === "/event-actions") {
238 const { EventActions } = yield* listEventActions();
239 return yield* HttpServerResponse.json({
240 count: (EventActions ?? []).length,
241 });
242 }
243
244 return yield* HttpServerResponse.json(
245 { error: "Not found", method: request.method, pathname },
246 { status: 404 },
247 );
248 }).pipe(Effect.orDie),
249 };
250 }).pipe(
251 Effect.provide(
252 Layer.mergeAll(
253 Lambda.EventSource,
254 DataExchange.GetDataSetHttp,
255 DataExchange.ListDataSetRevisionsHttp,
256 DataExchange.SendDataSetNotificationHttp,
257 DataExchange.GetRevisionHttp,
258 DataExchange.ListRevisionAssetsHttp,
259 DataExchange.GetAssetHttp,
260 DataExchange.ListDataSetsHttp,
261 DataExchange.CreateJobHttp,
262 DataExchange.StartJobHttp,
263 DataExchange.GetJobHttp,
264 DataExchange.ListJobsHttp,
265 DataExchange.ListDataGrantsHttp,
266 DataExchange.ListReceivedDataGrantsHttp,
267 DataExchange.ListEventActionsHttp,
268 S3.PutObjectHttp,
269 S3.GetObjectHttp,
270 ),
271 ),
272 ),
273);