1import * as keyspacesstreams from "@distilled.cloud/aws/keyspacesstreams";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Binding from "../../Binding.ts";
5import * as Output from "../../Output.ts";
6import { isBindingHost } from "../Lambda/Function.ts";
7import type { Table } from "./Table.ts";
8import {
9 TableStreams,
10 type ListTableStreamsRequest,
11 type TableStreamsClient,
12} from "./TableStreams.ts";
13
14/**
15 * HTTP implementation of {@link TableStreams}: grants the CDC stream read
16 * actions (`cassandra:ListStreams` / `GetStream` / `GetShardIterator` /
17 * `GetRecords`) on the bound table and its streams, and calls the Keyspaces
18 * streams HTTP API with the function's IAM credentials.
19 */
20export const TableStreamsHttp = Layer.effect(
21 TableStreams,
22 Effect.gen(function* () {
23 const listStreams = yield* keyspacesstreams.listStreams;
24 const getStream = yield* keyspacesstreams.getStream;
25 const getShardIterator = yield* keyspacesstreams.getShardIterator;
26 const getRecords = yield* keyspacesstreams.getRecords;
27
28 return Effect.fn(function* <T extends Table>(table: T) {
29 const KeyspaceName = yield* table.keyspaceName;
30 const TableName = yield* table.tableName;
31 if (!globalThis.__ALCHEMY_RUNTIME__) {
32 const host = yield* Binding.Host;
33 if (isBindingHost(host)) {
34 yield* host.bind`Allow(${host}, AWS.Keyspaces.TableStreams(${table}))`(
35 {
36 policyStatements: [
37 {
38 Effect: "Allow",
39 Action: [
40 "cassandra:GetRecords",
41 "cassandra:GetShardIterator",
42 "cassandra:GetStream",
43 "cassandra:ListStreams",
44 ],
45 Resource: [
46 table.tableArn,
47 Output.interpolate`${table.tableArn}/stream/*`,
48 ],
49 },
50 ],
51 },
52 );
53 }
54 }
55 const logicalId = table.LogicalId;
56 const client: TableStreamsClient = {
57 listStreams: Effect.fn(
58 `AWS.Keyspaces.TableStreams.listStreams(${logicalId})`,
59 )(function* (request?: ListTableStreamsRequest) {
60 const keyspaceName = yield* KeyspaceName;
61 const tableName = yield* TableName;
62 return yield* listStreams({ ...request, keyspaceName, tableName });
63 }),
64 getStream: Effect.fn(
65 `AWS.Keyspaces.TableStreams.getStream(${logicalId})`,
66 )(function* (request: keyspacesstreams.GetStreamInput) {
67 return yield* getStream(request);
68 }),
69 getShardIterator: Effect.fn(
70 `AWS.Keyspaces.TableStreams.getShardIterator(${logicalId})`,
71 )(function* (request: keyspacesstreams.GetShardIteratorInput) {
72 return yield* getShardIterator(request);
73 }),
74 getRecords: Effect.fn(
75 `AWS.Keyspaces.TableStreams.getRecords(${logicalId})`,
76 )(function* (request: keyspacesstreams.GetRecordsInput) {
77 return yield* getRecords(request);
78 }),
79 };
80 return client;
81 });
82 }),
83);