1import * as Effect from "effect/Effect";
2import * as Binding from "../../Binding.ts";
3import { isBindingHost } from "../Lambda/Function.ts";
4import type { Stream } from "./Stream.ts";
5import type { StreamConsumer } from "./StreamConsumer.ts";
6
7/**
8 * Shared scaffolding for AWS Kinesis HTTP bindings.
9 *
10 * NOT exported from `index.ts` — every near-identical `{Op}Http.ts` in this
11 * service is a thin `Layer.effect(Cap, make…HttpBinding({ … }))` over one of
12 * the builders below. Everything except the operation, the IAM action list,
13 * and the injected identifier is boilerplate. Genuinely-different bindings
14 * (the batched `StreamSink`, the `Stream | StreamConsumer` polymorphic
15 * `ListTagsForResource`) stay bespoke.
16 */
17
18/**
19 * Build the impl Effect for an account-level operation (`ListStreams`,
20 * `DescribeLimits`, `DescribeAccountSettings`): the runtime callable passes
21 * the caller's request through unchanged and the deploy-time half grants
22 * `actions` on `*` (these Kinesis actions do not support resource-level
23 * permissions).
24 */
25export const makeKinesisAccountHttpBinding = <I, A, E, R>(options: {
26 /** Fully-qualified binding tag, e.g. `AWS.Kinesis.ListStreams`. */
27 tag: string;
28 /** The distilled operation, invoked with the caller's request as-is. */
29 operation: Effect.Effect<(input: I) => Effect.Effect<A, E>, never, R>;
30 /** IAM actions granted on `*`. */
31 actions: readonly string[];
32}) =>
33 Effect.gen(function* () {
34 const op = yield* options.operation;
35
36 return Effect.fn(function* () {
37 if (!globalThis.__ALCHEMY_RUNTIME__) {
38 const host = yield* Binding.Host;
39 if (isBindingHost(host)) {
40 yield* host.bind`Allow(${host}, ${options.tag}())`({
41 policyStatements: [
42 {
43 Effect: "Allow",
44 Action: [...options.actions],
45 Resource: ["*"],
46 },
47 ],
48 });
49 }
50 }
51 return Effect.fn(options.tag)(function* (request?: I) {
52 return yield* op((request ?? {}) as I);
53 });
54 });
55 });
56
57/**
58 * Build the impl Effect for a stream-scoped operation: the runtime callable
59 * injects the bound {@link Stream}'s identity under `key` (`StreamARN` for
60 * the describe/shard APIs, `StreamName` for the producer APIs, `ResourceARN`
61 * for the policy APIs) and the deploy-time half grants `actions` on the
62 * stream's ARN.
63 */
64export const makeStreamHttpBinding = <
65 K extends "StreamARN" | "StreamName" | "ResourceARN",
66 I extends { [P in K]?: string },
67 A,
68 E,
69 R,
70>(options: {
71 /** Fully-qualified binding tag, e.g. `AWS.Kinesis.DescribeStream`. */
72 tag: string;
73 /** The distilled operation; the stream identity is injected under `key`. */
74 operation: Effect.Effect<(input: I) => Effect.Effect<A, E>, never, R>;
75 /** IAM actions granted on the stream ARN. */
76 actions: readonly string[];
77 /** The request field the stream identity is injected under. */
78 key: K;
79}) =>
80 Effect.gen(function* () {
81 const op = yield* options.operation;
82
83 return Effect.fn(function* (stream: Stream) {
84 const identity = yield* options.key === "StreamName"
85 ? stream.streamName
86 : stream.streamArn;
87 if (!globalThis.__ALCHEMY_RUNTIME__) {
88 const host = yield* Binding.Host;
89 if (isBindingHost(host)) {
90 yield* host.bind`Allow(${host}, ${options.tag}(${stream}))`({
91 policyStatements: [
92 {
93 Effect: "Allow",
94 Action: [...options.actions],
95 Resource: [stream.streamArn],
96 },
97 ],
98 });
99 }
100 }
101 return Effect.fn(`${options.tag}(${stream.LogicalId})`)(function* (
102 request?: Omit<I, K>,
103 ) {
104 return yield* op({
105 ...request,
106 [options.key]: yield* identity,
107 } as I);
108 });
109 });
110 });
111
112/**
113 * Build the impl Effect for a consumer-scoped operation
114 * (`DescribeStreamConsumer`, `SubscribeToShard`): the runtime callable
115 * injects the bound {@link StreamConsumer}'s ARN as `ConsumerARN` and the
116 * deploy-time half grants `actions` on the consumer ARN.
117 */
118export const makeConsumerHttpBinding = <
119 I extends { ConsumerARN?: string },
120 A,
121 E,
122 R,
123>(options: {
124 /** Fully-qualified binding tag, e.g. `AWS.Kinesis.SubscribeToShard`. */
125 tag: string;
126 /** The distilled operation; `ConsumerARN` is injected from the consumer. */
127 operation: Effect.Effect<(input: I) => Effect.Effect<A, E>, never, R>;
128 /** IAM actions granted on the consumer ARN. */
129 actions: readonly string[];
130}) =>
131 Effect.gen(function* () {
132 const op = yield* options.operation;
133
134 return Effect.fn(function* (consumer: StreamConsumer) {
135 const ConsumerARN = yield* consumer.consumerArn;
136 if (!globalThis.__ALCHEMY_RUNTIME__) {
137 const host = yield* Binding.Host;
138 if (isBindingHost(host)) {
139 yield* host.bind`Allow(${host}, ${options.tag}(${consumer}))`({
140 policyStatements: [
141 {
142 Effect: "Allow",
143 Action: [...options.actions],
144 Resource: [consumer.consumerArn],
145 },
146 ],
147 });
148 }
149 }
150 return Effect.fn(`${options.tag}(${consumer.LogicalId})`)(function* (
151 request?: Omit<I, "ConsumerARN">,
152 ) {
153 return yield* op({
154 ...request,
155 ConsumerARN: yield* ConsumerARN,
156 } as I);
157 });
158 });
159 });