1import * as data from "@distilled.cloud/aws/redshift-data";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Schedule from "effect/Schedule";
5import * as Binding from "../../Binding.ts";
6import { isBindingHost } from "../Lambda/Function.ts";
7import type { Workgroup } from "../RedshiftServerless/Workgroup.ts";
8import {
9 RedshiftStatementFailed,
10 Statements,
11 type BatchExecuteStatementRequest,
12 type DescribeTableRequest,
13 type ExecuteStatementRequest,
14 type ListDatabasesRequest,
15 type ListSchemasRequest,
16 type ListStatementsRequest,
17 type ListTablesRequest,
18 type StatementsClient,
19 type StatementsOptions,
20} from "./Statements.ts";
21
22/** Statement statuses that mean the run is over. */
23const isTerminal = (status: string | undefined): boolean =>
24 status === "FINISHED" || status === "FAILED" || status === "ABORTED";
25
26export const StatementsHttp = Layer.effect(
27 Statements,
28 Effect.gen(function* () {
29 const executeStatement = yield* data.executeStatement;
30 const batchExecuteStatement = yield* data.batchExecuteStatement;
31 const describeStatement = yield* data.describeStatement;
32 const cancelStatement = yield* data.cancelStatement;
33 const getStatementResult = yield* data.getStatementResult;
34 const getStatementResultV2 = yield* data.getStatementResultV2;
35 const describeTableOp = yield* data.describeTable;
36 const listDatabasesOp = yield* data.listDatabases;
37 const listSchemasOp = yield* data.listSchemas;
38 const listTablesOp = yield* data.listTables;
39 const listStatementsOp = yield* data.listStatements;
40
41 return Effect.fn(function* (
42 workgroup: Workgroup,
43 options: StatementsOptions = {},
44 ) {
45 const workgroupName = yield* workgroup.workgroupName;
46 if (!globalThis.__ALCHEMY_RUNTIME__) {
47 const host = yield* Binding.Host;
48 if (isBindingHost(host)) {
49 yield* host.bind`Allow(${host}, AWS.RedshiftData.Statements(${workgroup}))`(
50 {
51 policyStatements: [
52 {
53 // Workgroup-scoped Data API actions (submit statements and
54 // read database metadata through the workgroup).
55 Effect: "Allow",
56 Action: [
57 "redshift-data:ExecuteStatement",
58 "redshift-data:BatchExecuteStatement",
59 "redshift-data:DescribeTable",
60 "redshift-data:ListDatabases",
61 "redshift-data:ListSchemas",
62 "redshift-data:ListTables",
63 ],
64 Resource: [workgroup.workgroupArn],
65 },
66 {
67 // Statement-scoped actions are authorized per-statement
68 // (owner condition), not by ARN.
69 Effect: "Allow",
70 Action: [
71 "redshift-data:DescribeStatement",
72 "redshift-data:GetStatementResult",
73 "redshift-data:GetStatementResultV2",
74 "redshift-data:CancelStatement",
75 "redshift-data:ListStatements",
76 ],
77 Resource: ["*"],
78 },
79 {
80 // IAM temporary-credential auth against the serverless
81 // workgroup (used when no SecretArn is supplied).
82 Effect: "Allow",
83 Action: ["redshift-serverless:GetCredentials"],
84 Resource: [workgroup.workgroupArn],
85 },
86 ...(options.secretArn
87 ? [
88 {
89 Effect: "Allow" as const,
90 Action: [
91 "secretsmanager:GetSecretValue",
92 "secretsmanager:DescribeSecret",
93 ],
94 Resource: [options.secretArn],
95 },
96 ]
97 : []),
98 ],
99 },
100 );
101 }
102 }
103
104 const database = options.database ?? "dev";
105
106 const execute = Effect.fn(
107 `AWS.RedshiftData.Statements.execute(${workgroup.LogicalId})`,
108 )(function* (request: ExecuteStatementRequest) {
109 return yield* executeStatement({
110 ...request,
111 WorkgroupName: yield* workgroupName,
112 Database: database,
113 SecretArn: options.secretArn,
114 DbUser: options.dbUser,
115 });
116 });
117
118 const executeBatch = Effect.fn(
119 `AWS.RedshiftData.Statements.executeBatch(${workgroup.LogicalId})`,
120 )(function* (request: BatchExecuteStatementRequest) {
121 return yield* batchExecuteStatement({
122 ...request,
123 WorkgroupName: yield* workgroupName,
124 Database: database,
125 SecretArn: options.secretArn,
126 DbUser: options.dbUser,
127 });
128 });
129
130 const describe = Effect.fn(
131 `AWS.RedshiftData.Statements.describe(${workgroup.LogicalId})`,
132 )(function* (id: string) {
133 return yield* describeStatement({ Id: id });
134 });
135
136 const cancel = Effect.fn(
137 `AWS.RedshiftData.Statements.cancel(${workgroup.LogicalId})`,
138 )(function* (id: string) {
139 return yield* cancelStatement({ Id: id });
140 });
141
142 const getResult = Effect.fn(
143 `AWS.RedshiftData.Statements.getResult(${workgroup.LogicalId})`,
144 )(function* (id: string, nextToken?: string) {
145 return yield* getStatementResult({ Id: id, NextToken: nextToken });
146 });
147
148 const getResultV2 = Effect.fn(
149 `AWS.RedshiftData.Statements.getResultV2(${workgroup.LogicalId})`,
150 )(function* (id: string, nextToken?: string) {
151 return yield* getStatementResultV2({ Id: id, NextToken: nextToken });
152 });
153
154 const describeTable = Effect.fn(
155 `AWS.RedshiftData.Statements.describeTable(${workgroup.LogicalId})`,
156 )(function* (request: DescribeTableRequest) {
157 return yield* describeTableOp({
158 ...request,
159 WorkgroupName: yield* workgroupName,
160 Database: database,
161 SecretArn: options.secretArn,
162 DbUser: options.dbUser,
163 });
164 });
165
166 const listDatabases = Effect.fn(
167 `AWS.RedshiftData.Statements.listDatabases(${workgroup.LogicalId})`,
168 )(function* (request: ListDatabasesRequest = {}) {
169 return yield* listDatabasesOp({
170 ...request,
171 WorkgroupName: yield* workgroupName,
172 Database: database,
173 SecretArn: options.secretArn,
174 DbUser: options.dbUser,
175 });
176 });
177
178 const listSchemas = Effect.fn(
179 `AWS.RedshiftData.Statements.listSchemas(${workgroup.LogicalId})`,
180 )(function* (request: ListSchemasRequest = {}) {
181 return yield* listSchemasOp({
182 ...request,
183 WorkgroupName: yield* workgroupName,
184 Database: database,
185 SecretArn: options.secretArn,
186 DbUser: options.dbUser,
187 });
188 });
189
190 const listTables = Effect.fn(
191 `AWS.RedshiftData.Statements.listTables(${workgroup.LogicalId})`,
192 )(function* (request: ListTablesRequest = {}) {
193 return yield* listTablesOp({
194 ...request,
195 WorkgroupName: yield* workgroupName,
196 Database: database,
197 SecretArn: options.secretArn,
198 DbUser: options.dbUser,
199 });
200 });
201
202 const listStatements = Effect.fn(
203 `AWS.RedshiftData.Statements.listStatements(${workgroup.LogicalId})`,
204 )(function* (request: ListStatementsRequest = {}) {
205 return yield* listStatementsOp({
206 ...request,
207 WorkgroupName: yield* workgroupName,
208 });
209 });
210
211 const query = Effect.fn(
212 `AWS.RedshiftData.Statements.query(${workgroup.LogicalId})`,
213 )(function* (sql: string, parameters?: data.SqlParameter[]) {
214 const submitted = yield* execute({ Sql: sql, Parameters: parameters });
215 const id = submitted.Id!;
216 // Poll until the statement reaches a terminal status. Redshift
217 // Serverless typically finishes simple statements in a few seconds;
218 // budget ~2.5 min (30 * 5s).
219 const described = yield* describe(id).pipe(
220 Effect.repeat({
221 schedule: Schedule.spaced("5 seconds"),
222 until: (r) => isTerminal(r.Status),
223 times: 30,
224 }),
225 );
226 if (described.Status !== "FINISHED") {
227 return yield* Effect.fail(
228 new RedshiftStatementFailed({
229 statementId: id,
230 status: described.Status ?? "UNKNOWN",
231 error: described.Error,
232 }),
233 );
234 }
235 return yield* getResult(id);
236 });
237
238 return {
239 execute,
240 executeBatch,
241 describe,
242 cancel,
243 getResult,
244 getResultV2,
245 describeTable,
246 listDatabases,
247 listSchemas,
248 listTables,
249 listStatements,
250 query,
251 } satisfies StatementsClient;
252 });
253 }),
254);