1import * as Lambda from "@/AWS/Lambda";
2import * as OpenSearch from "@/AWS/OpenSearch";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import { HttpServerRequest } from "effect/http/HttpServerRequest";
7import * as HttpServerResponse from "effect/http/HttpServerResponse";
8import path from "pathe";
9
10const main = path.resolve(import.meta.dirname, "data-plane-handler.ts");
11
12interface Song {
13 title: string;
14 plays: number;
15}
16
17export class OpenSearchDataPlaneFunction extends Lambda.Function<Lambda.Function>()(
18 "OpenSearchDataPlaneFunction",
19) {}
20
21/**
22 * Lambda fixture exercising the OpenSearch domain data-plane bindings
23 * (`DomainRead` / `DomainWrite` / `DomainReadWrite`) against a real domain.
24 * The domain has no resource-based access policy, so authorization is purely
25 * the identity-based `es:ESHttp*` grants the bindings attach — a missing
26 * grant surfaces as a 403 from the data plane.
27 */
28export default OpenSearchDataPlaneFunction.make(
29 {
30 main,
31 functionUrl: true,
32 timeout: Duration.seconds(30),
33 },
34 Effect.gen(function* () {
35 const domain = yield* OpenSearch.Domain("DataPlaneDomain", {
36 clusterConfig: { instanceType: "t3.small.search", instanceCount: 1 },
37 ebsOptions: { volumeType: "gp3", volumeSize: 10 },
38 });
39
40 const reader = yield* OpenSearch.DomainRead(domain);
41 const writer = yield* OpenSearch.DomainWrite(domain);
42 const client = yield* OpenSearch.DomainReadWrite(domain);
43
44 const bound = { reader, writer, client };
45
46 return {
47 fetch: Effect.gen(function* () {
48 const request = yield* HttpServerRequest;
49 const url = new URL(request.originalUrl);
50 const pathname = url.pathname;
51
52 if (request.method === "GET" && pathname === "/bindings") {
53 return yield* HttpServerResponse.json({
54 bound: Object.keys(bound),
55 });
56 }
57
58 if (request.method === "POST" && pathname === "/doc") {
59 // DomainWrite: PUT /{index}/_doc/{id} (es:ESHttpPut).
60 const response = yield* writer.indexDocument(
61 "songs",
62 { title: "The Wind Cries Mary", plays: 1 } satisfies Song,
63 { id: "1", refresh: true },
64 );
65 return yield* HttpServerResponse.json({ result: response.result });
66 }
67
68 if (request.method === "GET" && pathname === "/doc") {
69 // DomainRead: GET /{index}/_doc/{id} (es:ESHttpGet).
70 const doc = yield* reader.getDocument<Song>("songs", "1");
71 return yield* HttpServerResponse.json({
72 found: doc.found,
73 title: doc._source?.title,
74 });
75 }
76
77 if (request.method === "GET" && pathname === "/doc-missing") {
78 // A missing document is found:false, not an error.
79 const doc = yield* reader.getDocument<Song>("songs", "missing");
80 return yield* HttpServerResponse.json({ found: doc.found });
81 }
82
83 if (request.method === "GET" && pathname === "/exists") {
84 // DomainRead: HEAD /{index}/_doc/{id} (es:ESHttpHead).
85 const exists = yield* reader.existsDocument("songs", "1");
86 const missing = yield* reader.existsDocument("songs", "missing");
87 return yield* HttpServerResponse.json({ exists, missing });
88 }
89
90 if (request.method === "GET" && pathname === "/search") {
91 // DomainRead: Query DSL via the `source` param stays on ESHttpGet.
92 const result = yield* reader.search<Song>({
93 index: "songs",
94 body: { query: { match: { title: "wind" } } },
95 });
96 return yield* HttpServerResponse.json({
97 total: result.hits.total.value,
98 firstTitle: result.hits.hits[0]?._source.title,
99 });
100 }
101
102 if (request.method === "GET" && pathname === "/count") {
103 const result = yield* reader.count({ index: "songs" });
104 return yield* HttpServerResponse.json({ count: result.count });
105 }
106
107 if (request.method === "POST" && pathname === "/bulk") {
108 // DomainWrite: POST /_bulk with NDJSON (es:ESHttpPost).
109 const response = yield* writer.bulk(
110 [
111 { index: { _index: "songs", _id: "2" } },
112 { title: "Purple Haze", plays: 5 } satisfies Song,
113 ],
114 { refresh: true },
115 );
116 return yield* HttpServerResponse.json({
117 errors: response.errors,
118 items: response.items.length,
119 });
120 }
121
122 if (request.method === "POST" && pathname === "/update") {
123 const response = yield* writer.updateDocument(
124 "songs",
125 "1",
126 { doc: { plays: 2 } },
127 { refresh: true },
128 );
129 return yield* HttpServerResponse.json({ result: response.result });
130 }
131
132 if (request.method === "POST" && pathname === "/delete") {
133 // DomainWrite: DELETE (es:ESHttpDelete); deleting twice yields
134 // result "not_found" rather than an error.
135 const first = yield* writer.deleteDocument("songs", "2", {
136 refresh: true,
137 });
138 const second = yield* writer.deleteDocument("songs", "2", {
139 refresh: true,
140 });
141 return yield* HttpServerResponse.json({
142 first: first.result,
143 second: second.result,
144 });
145 }
146
147 if (request.method === "GET" && pathname === "/health") {
148 // DomainReadWrite raw escape hatch (es:ESHttp*).
149 const health = (yield* client.request("GET", "_cluster/health")) as {
150 status?: string;
151 };
152 return yield* HttpServerResponse.json({ status: health.status });
153 }
154
155 return yield* HttpServerResponse.json(
156 { error: "Not found", method: request.method, pathname },
157 { status: 404 },
158 );
159 }).pipe(Effect.orDie),
160 };
161 }).pipe(
162 Effect.provide(
163 Layer.mergeAll(
164 OpenSearch.DomainReadHttp,
165 OpenSearch.DomainWriteHttp,
166 OpenSearch.DomainReadWriteHttp,
167 ),
168 ),
169 ),
170);