1/**
2 * Shared scaffolding for the OpenSearch domain data-plane bindings.
3 *
4 * An OpenSearch domain serves its search/index REST API under the domain's
5 * `endpoint`, authorized by the IAM method actions (`es:ESHttpGet`,
6 * `es:ESHttpPut`, …). There is no distilled operation for these — every
7 * request is a SigV4-signed HTTP call (service `"es"`) made with the host
8 * Function's own credentials.
9 *
10 * NOT exported from `index.ts` — only the per-capability contracts
11 * (`DomainRead` / `DomainWrite` / `DomainReadWrite`) and their `*Http`
12 * layers are public.
13 */
14import * as Credentials from "@distilled.cloud/aws/Credentials";
15import * as Region from "@distilled.cloud/aws/Region";
16import * as SigV4 from "@distilled.cloud/aws/SigV4";
17import * as Effect from "effect/Effect";
18import * as Redacted from "effect/Redacted";
19import * as Binding from "../../Binding.ts";
20import * as Output from "../../Output.ts";
21import { isBindingHost } from "../Lambda/Function.ts";
22import type {
23 BulkResponse,
24 CountResponse,
25 GetDocumentResponse,
26 RefreshOption,
27 SearchRequest,
28 SearchResponse,
29 WriteDocumentResponse,
30} from "./DataPlaneTypes.ts";
31import { OpenSearchApiError } from "./DataPlaneTypes.ts";
32import type { Domain } from "./Domain.ts";
33import type { ReadDomainClient } from "./DomainRead.ts";
34import type { WriteDomainClient } from "./DomainWrite.ts";
35
36/** One signed request against the domain's REST data plane. */
37export interface OpenSearchHttpRequest {
38 method: "GET" | "HEAD" | "POST" | "PUT" | "DELETE" | "PATCH";
39 /** Path relative to the domain endpoint, e.g. `"songs/_search"`. */
40 path: string;
41 /** Query-string parameters (undefined values are skipped). */
42 query?: Record<string, string | undefined>;
43 /** JSON body (sent as `application/json`). */
44 json?: unknown;
45 /** Raw NDJSON body (used by `_bulk`; sent as `application/x-ndjson`). */
46 ndjson?: string;
47 /** Statuses (besides 2xx) whose body is returned instead of failing. */
48 allowStatuses?: readonly number[];
49}
50
51/**
52 * Signs and sends one {@link OpenSearchHttpRequest} and returns the parsed
53 * JSON body (`undefined` for empty bodies, e.g. `HEAD` responses — check
54 * `status` instead).
55 */
56export type OpenSearchSend = (
57 request: OpenSearchHttpRequest,
58) => Effect.Effect<
59 { status: number; body: unknown },
60 OpenSearchApiError | Credentials.CredentialsError | SigV4.SigningError
61>;
62
63const refreshParam = (
64 refresh: RefreshOption | undefined,
65): Record<string, string | undefined> =>
66 refresh === undefined ? {} : { refresh: String(refresh) };
67
68/**
69 * Build the shared body of an OpenSearch data-plane `*Http` binding layer:
70 * resolve the ambient distilled `Credentials`/`Region` context once at layer
71 * construction, register the least-privilege IAM statement on the host
72 * Function at deploy time (`iamActions` on the domain and `domain/*`), and
73 * hand a signed-request `send` function to the capability-specific
74 * `makeClient`.
75 */
76export const makeOpenSearchDataPlaneBinding = <Client>(options: {
77 /** Capability name used in the binding SID and trace spans, e.g. `"DomainRead"`. */
78 name: string;
79 /** IAM actions the capability requires, e.g. `["es:ESHttpGet"]`. */
80 iamActions: string[];
81 /** Build the typed runtime client from the signed-request sender. */
82 makeClient: (send: OpenSearchSend) => Client;
83}) =>
84 Effect.gen(function* () {
85 const services = yield* Effect.context<
86 Credentials.Credentials | Region.Region
87 >();
88
89 return Effect.fn(function* (domain: Domain) {
90 const Endpoint = yield* domain.endpoint;
91 if (!globalThis.__ALCHEMY_RUNTIME__) {
92 const host = yield* Binding.Host;
93 if (isBindingHost(host)) {
94 yield* host.bind`Allow(${host}, AWS.OpenSearch.${options.name}(${domain}))`(
95 {
96 policyStatements: [
97 {
98 Effect: "Allow",
99 Action: options.iamActions,
100 Resource: [
101 domain.domainArn,
102 Output.interpolate`${domain.domainArn}/*`,
103 ],
104 },
105 ],
106 },
107 );
108 }
109 }
110
111 const send: OpenSearchSend = Effect.fn(
112 `AWS.OpenSearch.${options.name}(${domain.LogicalId})`,
113 )(function* (request: OpenSearchHttpRequest) {
114 const endpoint = yield* Endpoint;
115 if (endpoint === undefined) {
116 return yield* Effect.fail(
117 new OpenSearchApiError({
118 method: request.method,
119 path: request.path,
120 status: 0,
121 body: "domain has no endpoint (still provisioning, or VPC-only)",
122 }),
123 );
124 }
125
126 const url = new URL(
127 request.path.replace(/^\//, ""),
128 `https://${endpoint.replace(/^https?:\/\//, "")}/`,
129 );
130 for (const [key, value] of Object.entries(request.query ?? {})) {
131 if (value !== undefined) url.searchParams.set(key, value);
132 }
133
134 const headers: Record<string, string> = {};
135 let body: string | undefined;
136 if (request.json !== undefined) {
137 body = JSON.stringify(request.json);
138 headers["content-type"] = "application/json";
139 } else if (request.ndjson !== undefined) {
140 body = request.ndjson;
141 headers["content-type"] = "application/x-ndjson";
142 }
143
144 // Resolve credentials fresh per request (SSO/STS sessions rotate) and
145 // sign for the domain's own region, parsed from its endpoint.
146 const { credentials, region } = yield* Effect.gen(function* () {
147 const credentials = yield* yield* Credentials.Credentials;
148 const region =
149 /\.([a-z0-9-]+)\.(?:es|aos)\.amazonaws\.com$/.exec(
150 url.hostname,
151 )?.[1] ?? (yield* yield* Region.Region);
152 return { credentials, region };
153 }).pipe(Effect.provideContext(services));
154
155 const signed = yield* SigV4.sign({
156 method: request.method,
157 url: url.toString(),
158 headers,
159 body,
160 accessKeyId: Redacted.value(credentials.accessKeyId),
161 secretAccessKey: credentials.secretAccessKey,
162 sessionToken: credentials.sessionToken,
163 service: "es",
164 region,
165 allHeaders: true,
166 });
167
168 const toError = (status: number) => (cause: unknown) =>
169 new OpenSearchApiError({
170 method: request.method,
171 path: request.path,
172 status,
173 body: cause instanceof Error ? cause.message : String(cause),
174 });
175
176 const response = yield* Effect.tryPromise({
177 try: () =>
178 fetch(signed.url, {
179 method: signed.method,
180 headers: signed.headers,
181 body,
182 }),
183 catch: toError(0),
184 });
185 const text = yield* Effect.tryPromise({
186 try: () => response.text(),
187 catch: toError(response.status),
188 });
189
190 const ok =
191 (response.status >= 200 && response.status < 300) ||
192 (request.allowStatuses?.includes(response.status) ?? false);
193 if (!ok) {
194 return yield* Effect.fail(
195 new OpenSearchApiError({
196 method: request.method,
197 path: request.path,
198 status: response.status,
199 body: text,
200 }),
201 );
202 }
203 if (text.trim() === "") {
204 return { status: response.status, body: undefined };
205 }
206
207 const parsed = yield* Effect.try({
208 try: () => JSON.parse(text) as unknown,
209 catch: toError(response.status),
210 });
211 return { status: response.status, body: parsed };
212 });
213
214 return options.makeClient(send);
215 });
216 });
217
218const encodePath = (...segments: string[]): string =>
219 segments.map((segment) => encodeURIComponent(segment)).join("/");
220
221/** Build the read half of the domain data-plane client. */
222export const makeReadDomainClient = (
223 send: OpenSearchSend,
224): ReadDomainClient => ({
225 search: <TDoc>(request?: SearchRequest) =>
226 send({
227 method: "GET",
228 path:
229 request?.index !== undefined
230 ? `${encodeURIComponent(request.index)}/_search`
231 : "_search",
232 query: {
233 ...request?.query,
234 // The Query-DSL body travels in the `source` parameter so searches
235 // stay on es:ESHttpGet (POST would require the write IAM action).
236 ...(request?.body !== undefined
237 ? {
238 source: JSON.stringify(request.body),
239 source_content_type: "application/json",
240 }
241 : {}),
242 },
243 }).pipe(Effect.map(({ body }) => body as SearchResponse<TDoc>)),
244 count: (request) =>
245 send({
246 method: "GET",
247 path:
248 request?.index !== undefined
249 ? `${encodeURIComponent(request.index)}/_count`
250 : "_count",
251 query:
252 request?.body !== undefined
253 ? {
254 source: JSON.stringify(request.body),
255 source_content_type: "application/json",
256 }
257 : undefined,
258 }).pipe(Effect.map(({ body }) => body as CountResponse)),
259 getDocument: <TDoc>(index: string, id: string) =>
260 send({
261 method: "GET",
262 path: encodePath(index, "_doc", id),
263 allowStatuses: [404],
264 }).pipe(Effect.map(({ body }) => body as GetDocumentResponse<TDoc>)),
265 existsDocument: (index, id) =>
266 send({
267 method: "HEAD",
268 path: encodePath(index, "_doc", id),
269 allowStatuses: [404],
270 }).pipe(Effect.map(({ status }) => status === 200)),
271 get: (path, query) =>
272 send({ method: "GET", path, query }).pipe(Effect.map(({ body }) => body)),
273});
274
275/** Build the write half of the domain data-plane client. */
276export const makeWriteDomainClient = (
277 send: OpenSearchSend,
278): WriteDomainClient => ({
279 indexDocument: (index, document, options) =>
280 send(
281 options?.id !== undefined
282 ? {
283 method: "PUT",
284 path: encodePath(index, "_doc", options.id),
285 json: document,
286 query: refreshParam(options.refresh),
287 }
288 : {
289 method: "POST",
290 path: encodePath(index, "_doc"),
291 json: document,
292 query: refreshParam(options?.refresh),
293 },
294 ).pipe(Effect.map(({ body }) => body as WriteDocumentResponse)),
295 updateDocument: (index, id, body, options) =>
296 send({
297 method: "POST",
298 path: encodePath(index, "_update", id),
299 json: body,
300 query: refreshParam(options?.refresh),
301 }).pipe(Effect.map(({ body }) => body as WriteDocumentResponse)),
302 deleteDocument: (index, id, options) =>
303 send({
304 method: "DELETE",
305 path: encodePath(index, "_doc", id),
306 query: refreshParam(options?.refresh),
307 allowStatuses: [404],
308 }).pipe(Effect.map(({ body }) => body as WriteDocumentResponse)),
309 bulk: (operations, options) =>
310 send({
311 method: "POST",
312 path: "_bulk",
313 ndjson: `${operations.map((line) => JSON.stringify(line)).join("\n")}\n`,
314 query: refreshParam(options?.refresh),
315 }).pipe(Effect.map(({ body }) => body as BulkResponse)),
316});