1import * as athena from "@distilled.cloud/aws/athena";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { isResolved } from "../../Diff.ts";
5import { createPhysicalName } from "../../PhysicalName.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import type { Providers } from "../Providers.ts";
9
10export interface PreparedStatementProps {
11 /**
12 * Name of the prepared statement. If omitted, a unique name is generated.
13 * Must start with a letter or underscore and contain only letters, digits,
14 * and underscores. Changing this replaces the prepared statement.
15 */
16 statementName?: string;
17 /**
18 * The SQL statement text, with `?` positional parameters (e.g.
19 * `SELECT * FROM t WHERE id = ?`). Updatable in place.
20 */
21 queryStatement: string;
22 /**
23 * Workgroup the statement is saved in. Changing this replaces the prepared
24 * statement.
25 * @default "primary"
26 */
27 workGroup?: string;
28 /**
29 * Optional human-readable description. Updatable in place.
30 */
31 description?: string;
32}
33
34export interface PreparedStatement extends Resource<
35 "AWS.Athena.PreparedStatement",
36 PreparedStatementProps,
37 {
38 /**
39 * Name of the prepared statement (its identity within the workgroup).
40 */
41 statementName: string;
42 /**
43 * Workgroup the statement is saved in.
44 */
45 workGroup: string;
46 /**
47 * The SQL statement text.
48 */
49 queryStatement: string;
50 /**
51 * Description of the prepared statement.
52 */
53 description: string | undefined;
54 },
55 never,
56 Providers
57> {}
58
59/**
60 * An Amazon Athena prepared statement — a named, parameterized SQL statement
61 * saved in a workgroup and run with `EXECUTE name USING ...`. Identity is the
62 * `(workGroup, statementName)` pair; the `queryStatement` and `description`
63 * are updatable in place.
64 *
65 * ### Preparing Statements
66 * **Example:** Save a parameterized statement in a workgroup
67 * ```typescript
68 * const stmt = yield* AWS.Athena.PreparedStatement("TopN", {
69 * workGroup: wg.workGroupName,
70 * queryStatement: "SELECT * FROM analytics.orders WHERE customer_id = ?",
71 * });
72 * // then at runtime: EXECUTE <stmt.statementName> USING 'customer-123'
73 * ```
74 *
75 * @resource
76 */
77export const PreparedStatement = Resource<PreparedStatement>(
78 "AWS.Athena.PreparedStatement",
79);
80
81const toAttributes = (ps: athena.PreparedStatement) => ({
82 statementName: ps.StatementName!,
83 workGroup: ps.WorkGroupName ?? "primary",
84 queryStatement: ps.QueryStatement ?? "",
85 description: ps.Description,
86});
87
88export const PreparedStatementProvider = () =>
89 Provider.effect(
90 PreparedStatement,
91 Effect.gen(function* () {
92 // Statement names must match [a-zA-Z_][a-zA-Z0-9_@:]* — generate the
93 // engine name and fold every other character to `_`.
94 const toName = (id: string, props: PreparedStatementProps) =>
95 props.statementName
96 ? Effect.succeed(props.statementName)
97 : createPhysicalName({ id, maxLength: 128 }).pipe(
98 Effect.map((n) => n.replace(/[^a-zA-Z0-9_]/g, "_")),
99 Effect.map((n) => (/^[a-zA-Z_]/.test(n) ? n : `_${n}`)),
100 );
101
102 const getOne = (workGroup: string, statementName: string) =>
103 athena
104 .getPreparedStatement({
105 WorkGroup: workGroup,
106 StatementName: statementName,
107 })
108 .pipe(
109 Effect.map((res) => res.PreparedStatement),
110 Effect.catchTag(
111 ["ResourceNotFoundException", "WorkGroupNotFound"],
112 () => Effect.succeed(undefined),
113 ),
114 );
115
116 return {
117 stables: ["statementName", "workGroup"],
118 diff: Effect.fn(function* ({ id, olds, news }) {
119 if (!isResolved(news)) return;
120 if (
121 (yield* toName(id, olds ?? { queryStatement: "" })) !==
122 (yield* toName(id, news))
123 ) {
124 return { action: "replace" } as const;
125 }
126 if (
127 (olds?.workGroup ?? "primary") !== (news.workGroup ?? "primary")
128 ) {
129 return { action: "replace" } as const;
130 }
131 }),
132 read: Effect.fn(function* ({ id, olds, output }) {
133 const workGroup = output?.workGroup ?? olds?.workGroup ?? "primary";
134 const name =
135 output?.statementName ??
136 (yield* toName(id, olds ?? { queryStatement: "" }));
137 const ps = yield* getOne(workGroup, name);
138 return ps ? toAttributes(ps) : undefined;
139 }),
140 list: () =>
141 Effect.gen(function* () {
142 const wgPages = yield* athena.listWorkGroups
143 .pages({})
144 .pipe(Stream.runCollect);
145 const workGroups = Array.from(wgPages)
146 .flatMap((page) => page.WorkGroups ?? [])
147 .flatMap((wg) => (wg.Name ? [wg.Name] : []));
148 const attrs: ReturnType<typeof toAttributes>[] = [];
149 for (const wg of workGroups) {
150 const namePages = yield* athena.listPreparedStatements
151 .pages({ WorkGroup: wg })
152 .pipe(Stream.runCollect);
153 const names = Array.from(namePages)
154 .flatMap((page) => page.PreparedStatements ?? [])
155 .flatMap((ps) => (ps.StatementName ? [ps.StatementName] : []));
156 for (let i = 0; i < names.length; i += 256) {
157 const res = yield* athena.batchGetPreparedStatement({
158 WorkGroup: wg,
159 PreparedStatementNames: names.slice(i, i + 256),
160 });
161 for (const ps of res.PreparedStatements ?? []) {
162 if (ps.StatementName) {
163 attrs.push(toAttributes({ ...ps, WorkGroupName: wg }));
164 }
165 }
166 }
167 }
168 return attrs;
169 }),
170 reconcile: Effect.fn(function* ({ id, news, output, session }) {
171 const workGroup = news.workGroup ?? "primary";
172 const name = output?.statementName ?? (yield* toName(id, news));
173
174 // Observe — cloud state is authoritative.
175 const ps = yield* getOne(workGroup, name);
176
177 if (!ps) {
178 // Ensure — create if missing. CreatePreparedStatement upserts on
179 // name collision within the workgroup, so a crash-and-retry race
180 // converges rather than erroring.
181 yield* athena.createPreparedStatement({
182 WorkGroup: workGroup,
183 StatementName: name,
184 QueryStatement: news.queryStatement,
185 Description: news.description,
186 });
187 } else if (
188 ps.QueryStatement !== news.queryStatement ||
189 (news.description !== undefined &&
190 ps.Description !== news.description)
191 ) {
192 // Sync — statement text and description are updatable in place.
193 yield* athena.updatePreparedStatement({
194 WorkGroup: workGroup,
195 StatementName: name,
196 QueryStatement: news.queryStatement,
197 Description: news.description,
198 });
199 }
200
201 yield* session.note(`${workGroup}/${name}`);
202 const final = yield* getOne(workGroup, name);
203 return final
204 ? toAttributes(final)
205 : {
206 statementName: name,
207 workGroup,
208 queryStatement: news.queryStatement,
209 description: news.description,
210 };
211 }),
212 delete: Effect.fn(function* ({ output }) {
213 // Idempotent — the statement (or its whole workgroup) may already be
214 // gone; both surface as typed not-found tags.
215 yield* athena
216 .deletePreparedStatement({
217 WorkGroup: output.workGroup,
218 StatementName: output.statementName,
219 })
220 .pipe(
221 Effect.catchTag(
222 ["ResourceNotFoundException", "WorkGroupNotFound"],
223 () => Effect.void,
224 ),
225 );
226 }),
227 };
228 }),
229 );