1/**
2 * Shared scaffolding for Amazon OpenSearch Ingestion (OSIS) HTTP bindings.
3 *
4 * NOT exported from `index.ts` — every thin `{Op}Http.ts` in this service is
5 * a `Layer.effect(Cap, make…HttpBinding({ … }))` over one of the builders
6 * below. Everything except the operation, the injected identifier, and the
7 * IAM action list is boilerplate:
8 *
9 * - Pipeline-scoped operations (`osis:GetPipeline`, `osis:StartPipeline`,
10 * `osis:StopPipeline`, …) inject the bound {@link Pipeline}'s name or ARN
11 * into the request and are granted on the pipeline ARN.
12 * - Account-level operations (`osis:ValidatePipeline`,
13 * `osis:ListPipelineBlueprints`, …) take the caller's request as-is and
14 * are granted on `*` — they are not scoped to a single pipeline resource.
15 * - The `osis:Ingest` data plane has no distilled operation — it is a
16 * SigV4-signed HTTP POST (service `"osis"`) against the pipeline's ingest
17 * endpoint, made with the host Function's own credentials.
18 */
19import * as Credentials from "@distilled.cloud/aws/Credentials";
20import * as Region from "@distilled.cloud/aws/Region";
21import * as SigV4 from "@distilled.cloud/aws/SigV4";
22import * as Data from "effect/Data";
23import * as Effect from "effect/Effect";
24import * as Redacted from "effect/Redacted";
25import * as Binding from "../../Binding.ts";
26import type * as Output from "../../Output.ts";
27import { isBindingHost } from "../Lambda/Function.ts";
28import type { Pipeline } from "./Pipeline.ts";
29
30/**
31 * Build the impl Effect for an OSIS operation scoped to a {@link Pipeline}:
32 * the deploy-time half grants `actions` on the bound pipeline's ARN, and the
33 * runtime half injects the pipeline's identifier (name or ARN) as
34 * `requestKey` into every request.
35 */
36export const makeOsisPipelineHttpBinding = <
37 I extends object,
38 K extends keyof I & string,
39 A,
40 E,
41 R,
42>(options: {
43 /** Fully-qualified binding tag, e.g. `AWS.OSIS.StartPipeline`. */
44 tag: string;
45 /** The distilled operation; `requestKey` is injected from the pipeline. */
46 operation: Effect.Effect<(input: I) => Effect.Effect<A, E>, never, R>;
47 /** IAM actions granted on the pipeline ARN. */
48 actions: readonly string[];
49 /** The request field the resolved identifier is injected as. */
50 requestKey: K;
51 /** Resolve the injected identifier from the bound pipeline. */
52 identifier: (pipeline: Pipeline) => Output.Output<string, never>;
53}) =>
54 Effect.gen(function* () {
55 const op = yield* options.operation;
56
57 return Effect.fn(function* (pipeline: Pipeline) {
58 const Identifier = yield* options.identifier(pipeline);
59 if (!globalThis.__ALCHEMY_RUNTIME__) {
60 const host = yield* Binding.Host;
61 if (isBindingHost(host)) {
62 yield* host.bind`Allow(${host}, ${options.tag}(${pipeline}))`({
63 policyStatements: [
64 {
65 Effect: "Allow",
66 Action: [...options.actions],
67 Resource: [pipeline.pipelineArn],
68 },
69 ],
70 });
71 }
72 }
73 return Effect.fn(`${options.tag}(${pipeline.LogicalId})`)(function* (
74 request?: Omit<I, K>,
75 ) {
76 return yield* op({
77 ...request,
78 [options.requestKey]: yield* Identifier,
79 } as I);
80 });
81 });
82 });
83
84/**
85 * Build the impl Effect for an account-level OSIS operation (config
86 * validation, blueprint catalog reads, endpoint-connection listing). The
87 * deploy-time half grants `actions` on `*` — these operations are not scoped
88 * to a single pipeline resource.
89 */
90export const makeOsisAccountHttpBinding = <I, A, E, R>(options: {
91 /** Fully-qualified binding tag, e.g. `AWS.OSIS.ValidatePipeline`. */
92 tag: string;
93 /** The distilled operation, invoked with the caller's request as-is. */
94 operation: Effect.Effect<(input: I) => Effect.Effect<A, E>, never, R>;
95 /** IAM actions granted on `*`. */
96 actions: readonly string[];
97}) =>
98 Effect.gen(function* () {
99 const op = yield* options.operation;
100
101 return Effect.fn(function* () {
102 if (!globalThis.__ALCHEMY_RUNTIME__) {
103 const host = yield* Binding.Host;
104 if (isBindingHost(host)) {
105 yield* host.bind`Allow(${host}, ${options.tag}())`({
106 policyStatements: [
107 {
108 Effect: "Allow",
109 Action: [...options.actions],
110 Resource: ["*"],
111 },
112 ],
113 });
114 }
115 }
116 return Effect.fn(options.tag)(function* (request?: I) {
117 return yield* op((request ?? {}) as I);
118 });
119 });
120 });
121
122/**
123 * A failed `osis:Ingest` request against a pipeline's ingest endpoint —
124 * carries the HTTP status and response body returned by Data Prepper.
125 */
126export class PipelineIngestError extends Data.TaggedError(
127 "OsisPipelineIngestError",
128)<{
129 readonly pipelineName: string;
130 /** Request path on the ingest endpoint, e.g. `/logs/ingest`. */
131 readonly path: string;
132 /** HTTP status; `0` when the request never reached the endpoint. */
133 readonly status: number;
134 readonly body: string;
135}> {}
136
137/** A single `osis:Ingest` request against the pipeline's ingest endpoint. */
138export interface IngestRequest {
139 /**
140 * Path of the pipeline's HTTP source, e.g. `/logs/ingest` — must match the
141 * `path` configured on the `http` (or `otel_*`) source in the pipeline
142 * configuration body.
143 */
144 path: string;
145 /**
146 * Events to ingest, JSON-serialized as the request body. Data Prepper's
147 * `http` source expects a JSON array of event objects.
148 */
149 events: ReadonlyArray<unknown>;
150}
151
152/**
153 * Build the impl Effect for the `osis:Ingest` data plane: the deploy-time
154 * half grants `osis:Ingest` on the bound pipeline's ARN, and the runtime half
155 * signs (SigV4, service `"osis"`) and POSTs each batch of events to the
156 * pipeline's ingest endpoint with the host Function's own credentials.
157 */
158export const makeOsisIngestBinding = Effect.gen(function* () {
159 const services = yield* Effect.context<
160 Credentials.Credentials | Region.Region
161 >();
162
163 return Effect.fn(function* (pipeline: Pipeline) {
164 const PipelineName = yield* pipeline.pipelineName;
165 const IngestEndpointUrls = yield* pipeline.ingestEndpointUrls;
166 if (!globalThis.__ALCHEMY_RUNTIME__) {
167 const host = yield* Binding.Host;
168 if (isBindingHost(host)) {
169 yield* host.bind`Allow(${host}, AWS.OSIS.Ingest(${pipeline}))`({
170 policyStatements: [
171 {
172 Effect: "Allow",
173 Action: ["osis:Ingest"],
174 Resource: [pipeline.pipelineArn],
175 },
176 ],
177 });
178 }
179 }
180
181 return Effect.fn(`AWS.OSIS.Ingest(${pipeline.LogicalId})`)(function* (
182 request: IngestRequest,
183 ) {
184 const pipelineName = yield* PipelineName;
185 const endpoints = yield* IngestEndpointUrls;
186 const endpoint = endpoints?.[0];
187 const fail = (status: number, body: string) =>
188 Effect.fail(
189 new PipelineIngestError({
190 pipelineName,
191 path: request.path,
192 status,
193 body,
194 }),
195 );
196 if (endpoint === undefined) {
197 return yield* fail(0, "pipeline has no ingestEndpointUrls");
198 }
199
200 // Ingest endpoint URLs are bare hostnames
201 // (`{name}-{id}.{region}.osis.amazonaws.com`); sign for the endpoint's
202 // own region, parsed from the hostname.
203 const base = endpoint.startsWith("https://")
204 ? endpoint
205 : `https://${endpoint}`;
206 const url = new URL(
207 request.path.startsWith("/") ? request.path : `/${request.path}`,
208 base,
209 );
210
211 // Resolve credentials fresh per request (STS sessions rotate).
212 const { credentials, region } = yield* Effect.gen(function* () {
213 const credentials = yield* yield* Credentials.Credentials;
214 const region =
215 /\.([a-z0-9-]+)\.osis\.amazonaws\.com$/.exec(url.hostname)?.[1] ??
216 (yield* yield* Region.Region);
217 return { credentials, region };
218 }).pipe(Effect.provideContext(services));
219
220 const body = JSON.stringify(request.events);
221 const signed = yield* SigV4.sign({
222 method: "POST",
223 url: url.toString(),
224 headers: { "content-type": "application/json" },
225 body,
226 accessKeyId: Redacted.value(credentials.accessKeyId),
227 secretAccessKey: credentials.secretAccessKey,
228 sessionToken: credentials.sessionToken,
229 service: "osis",
230 region,
231 allHeaders: true,
232 });
233
234 const response = yield* Effect.tryPromise({
235 try: () =>
236 fetch(signed.url, {
237 method: signed.method,
238 headers: signed.headers,
239 body,
240 }),
241 catch: (cause) =>
242 new PipelineIngestError({
243 pipelineName,
244 path: request.path,
245 status: 0,
246 body: cause instanceof Error ? cause.message : String(cause),
247 }),
248 });
249 const text = yield* Effect.tryPromise({
250 try: () => response.text(),
251 catch: (cause) =>
252 new PipelineIngestError({
253 pipelineName,
254 path: request.path,
255 status: response.status,
256 body: cause instanceof Error ? cause.message : String(cause),
257 }),
258 });
259 if (response.status < 200 || response.status >= 300) {
260 return yield* fail(response.status, text);
261 }
262 });
263 });
264});