1import * as Lambda from "@/AWS/Lambda";
2import * as RedshiftData from "@/AWS/RedshiftData";
3import * as RedshiftServerless from "@/AWS/RedshiftServerless";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Schedule from "effect/Schedule";
7import { HttpServerRequest } from "effect/http/HttpServerRequest";
8import * as HttpServerResponse from "effect/http/HttpServerResponse";
9import path from "pathe";
10
11const main = path.resolve(import.meta.dirname, "data-api-handler.ts");
12
13export class RedshiftDataApiFunction extends Lambda.Function<Lambda.Function>()(
14 "RedshiftDataApiFunction",
15) {}
16
17/** Statement statuses that mean the run is over. */
18const isTerminal = (status: string | undefined): boolean =>
19 status === "FINISHED" || status === "FAILED" || status === "ABORTED";
20
21/**
22 * One route per Data API client behavior, all against the same serverless
23 * workgroup (distinct deterministic names from the RedshiftServerless suite
24 * so both gated suites can run side by side without fighting over physical
25 * names).
26 */
27export default RedshiftDataApiFunction.make(
28 {
29 main,
30 functionUrl: true,
31 // Data API statements are submitted then polled; a cold workgroup can
32 // take ~30s to serve its first statement — allow generous headroom.
33 timeout: Duration.seconds(240),
34 },
35 Effect.gen(function* () {
36 const namespace = yield* RedshiftServerless.Namespace("DataApiNamespace", {
37 namespaceName: "alchemy-test-rsd-ns",
38 dbName: "dev",
39 adminUsername: "alchemyadmin",
40 manageAdminPassword: true,
41 });
42 const workgroup = yield* RedshiftServerless.Workgroup("DataApiWorkgroup", {
43 workgroupName: "alchemy-test-rsd-wg",
44 namespaceName: namespace.namespaceName,
45 baseCapacity: 8,
46 publiclyAccessible: false,
47 });
48
49 const sql = yield* RedshiftData.Statements(workgroup, { database: "dev" });
50
51 // Poll a submitted statement until it reaches a terminal status
52 // (bounded: 45 * 2s = 90s).
53 const waitTerminal = (id: string) =>
54 sql.describe(id).pipe(
55 Effect.repeat({
56 schedule: Schedule.spaced("2 seconds"),
57 until: (r): boolean => isTerminal(r.Status),
58 times: 45,
59 }),
60 );
61
62 return {
63 fetch: Effect.gen(function* () {
64 const request = yield* HttpServerRequest;
65 const url = new URL(request.originalUrl);
66 const pathname = url.pathname;
67
68 if (request.method !== "GET") {
69 return yield* HttpServerResponse.json(
70 { error: "Not found", method: request.method, pathname },
71 { status: 404 },
72 );
73 }
74
75 // execute + describe + getResult via the composite query.
76 if (pathname === "/query") {
77 const result = yield* sql.query("SELECT 1 AS n");
78 return yield* HttpServerResponse.json({
79 columns: (result.ColumnMetadata ?? []).map((c) => c.name),
80 records: result.Records,
81 totalNumRows: result.TotalNumRows,
82 });
83 }
84
85 // executeBatch + describe (sub-statements) + getResult of a
86 // sub-statement.
87 if (pathname === "/batch") {
88 const submitted = yield* sql.executeBatch({
89 Sqls: ["SELECT 1 AS a", "SELECT 2 AS b"],
90 });
91 const described = yield* waitTerminal(submitted.Id!);
92 const subStatements = described.SubStatements ?? [];
93 const second = subStatements[1];
94 const result =
95 described.Status === "FINISHED" && second !== undefined
96 ? yield* sql.getResult(second.Id)
97 : undefined;
98 return yield* HttpServerResponse.json({
99 status: described.Status,
100 subStatementCount: subStatements.length,
101 secondRecords: result?.Records,
102 });
103 }
104
105 // listDatabases + listSchemas + listTables + describeTable.
106 if (pathname === "/metadata") {
107 const databases = yield* sql.listDatabases();
108 const schemas = yield* sql.listSchemas({
109 SchemaPattern: "pg_catalog",
110 });
111 const tables = yield* sql.listTables({
112 SchemaPattern: "pg_catalog",
113 TablePattern: "pg_class",
114 });
115 const table = yield* sql.describeTable({
116 Schema: "pg_catalog",
117 Table: "pg_class",
118 });
119 return yield* HttpServerResponse.json({
120 databases: databases.Databases,
121 schemas: schemas.Schemas,
122 tables: (tables.Tables ?? []).map((t) => t.name),
123 columnCount: (table.ColumnList ?? []).length,
124 });
125 }
126
127 // execute + listStatements (statement-owner scoped).
128 if (pathname === "/statements") {
129 const submitted = yield* sql.execute({ Sql: "SELECT 42 AS answer" });
130 yield* waitTerminal(submitted.Id!);
131 const listed = yield* sql.listStatements();
132 return yield* HttpServerResponse.json({
133 count: listed.Statements.length,
134 hasSubmitted: listed.Statements.some((s) => s.Id === submitted.Id),
135 });
136 }
137
138 // execute a heavy statement then cancel it. Cancellation races
139 // statement completion — a ValidationException means the statement
140 // already finished, which still proves the wiring end to end.
141 if (pathname === "/cancel") {
142 const submitted = yield* sql.execute({
143 Sql: "SELECT count(*) FROM pg_catalog.pg_attribute a CROSS JOIN pg_catalog.pg_attribute b CROSS JOIN pg_catalog.pg_attribute c",
144 });
145 const canceled = yield* sql.cancel(submitted.Id!).pipe(
146 Effect.map((r) => r.Status === true),
147 Effect.catchTag("ValidationException", () => Effect.succeed(false)),
148 );
149 return yield* HttpServerResponse.json({ canceled });
150 }
151
152 // execute with ResultFormat CSV + getResultV2.
153 if (pathname === "/result-v2") {
154 const submitted = yield* sql.execute({
155 Sql: "SELECT 7 AS n",
156 ResultFormat: "CSV",
157 });
158 const described = yield* waitTerminal(submitted.Id!);
159 const result = yield* sql.getResultV2(submitted.Id!);
160 return yield* HttpServerResponse.json({
161 status: described.Status,
162 resultFormat: result.ResultFormat,
163 records: result.Records,
164 });
165 }
166
167 return yield* HttpServerResponse.json(
168 { error: "Not found", method: request.method, pathname },
169 { status: 404 },
170 );
171 }).pipe(Effect.orDie),
172 };
173 }).pipe(Effect.provide(RedshiftData.StatementsHttp)),
174);