1import type * as rdsdata from "@distilled.cloud/aws/rds-data";
2import type { RuntimeContext } from "alchemy";
3import * as AWS from "alchemy/AWS";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Schedule from "effect/Schedule";
7import type * as Scope from "effect/Scope";
8import {
9 Database,
10 type DatabaseService,
11 type DirectDatabase,
12} from "./Database.ts";
13import { BetterAuthMigrationError } from "./Errors.ts";
14
15/**
16 * Minimal shape of the RDS Data API surface the Kysely dialect drives —
17 * promise-based so the same dialect serves the runtime (backed by the
18 * `AWS.RDSData.*` bindings) and deploy-time migrations (backed by
19 * distilled with ambient credentials).
20 *
21 * @internal
22 */
23export interface DataApiExecutor {
24 readonly execute: (request: {
25 sql: string;
26 parameters?: rdsdata.SqlParameter[];
27 includeResultMetadata?: boolean;
28 transactionId?: string;
29 }) => Promise<DataApiResponse>;
30 readonly begin: () => Promise<{ transactionId?: string }>;
31 readonly commit: (transactionId: string) => Promise<unknown>;
32 readonly rollback: (transactionId: string) => Promise<unknown>;
33}
34
35interface DataApiResponse {
36 records?: rdsdata.Field[][];
37 columnMetadata?: rdsdata.ColumnMetadata[];
38 numberOfRecordsUpdated?: number;
39}
40
41/**
42 * Build a Kysely dialect over the RDS Data API.
43 *
44 * Postgres-flavoured, with `$n` placeholders rewritten to the Data API's
45 * named `:n` parameters. Streaming is unsupported (the Data API is
46 * request/response).
47 *
48 * @internal
49 */
50export const makeDataApiDialect = (
51 executor: DataApiExecutor,
52): Effect.Effect<import("kysely").Dialect> =>
53 Effect.promise(async () => {
54 const {
55 CastNode,
56 DataTypeNode,
57 OperationNodeTransformer,
58 PostgresAdapter,
59 PostgresIntrospector,
60 PostgresQueryCompiler,
61 ReferenceNode,
62 } = await import("kysely");
63
64 class DataApiIntrospectionTransformer extends OperationNodeTransformer {
65 protected override transformAlias(node: import("kysely").AliasNode) {
66 const alias = super.transformAlias(node);
67 const reference = alias.node;
68 if (
69 ReferenceNode.is(reference) &&
70 reference.column.kind === "ColumnNode" &&
71 reference.column.column.name === "relkind"
72 ) {
73 // The Data API cannot return PostgreSQL's internal "char" type.
74 return {
75 ...alias,
76 node: CastNode.create(reference, DataTypeNode.create("text")),
77 };
78 }
79 return alias;
80 }
81 }
82
83 const introspectionTransformer = new DataApiIntrospectionTransformer();
84 const introspectionPlugin: import("kysely").KyselyPlugin = {
85 transformQuery: ({ node }) =>
86 introspectionTransformer.transformNode(node),
87 transformResult: async ({ result }) => result,
88 };
89
90 class DataApiQueryCompiler extends PostgresQueryCompiler {
91 protected override getCurrentParameterPlaceholder(): string {
92 return `:${this.numParameters}`;
93 }
94 }
95
96 class DataApiConnection {
97 transactionId: string | undefined;
98
99 async executeQuery(compiledQuery: {
100 sql: string;
101 parameters: ReadonlyArray<unknown>;
102 }) {
103 const response = await executor.execute({
104 sql: compiledQuery.sql,
105 parameters: compiledQuery.parameters.map((value, index) =>
106 AWS.RDSData.toSqlParameter(`${index + 1}`, value),
107 ),
108 includeResultMetadata: true,
109 ...(this.transactionId === undefined
110 ? {}
111 : { transactionId: this.transactionId }),
112 });
113 return {
114 rows: AWS.RDSData.toRows(response) as never[],
115 numAffectedRows: BigInt(response.numberOfRecordsUpdated ?? 0),
116 };
117 }
118
119 async *streamQuery(): AsyncIterableIterator<never> {
120 throw new Error("RDS Data API does not support streaming queries");
121 }
122 }
123
124 class DataApiDriver {
125 async init() {}
126 async acquireConnection() {
127 return new DataApiConnection();
128 }
129 async beginTransaction(connection: DataApiConnection) {
130 const { transactionId } = await executor.begin();
131 connection.transactionId = transactionId;
132 }
133 async commitTransaction(connection: DataApiConnection) {
134 if (connection.transactionId !== undefined) {
135 await executor.commit(connection.transactionId);
136 connection.transactionId = undefined;
137 }
138 }
139 async rollbackTransaction(connection: DataApiConnection) {
140 if (connection.transactionId !== undefined) {
141 await executor.rollback(connection.transactionId);
142 connection.transactionId = undefined;
143 }
144 }
145 async releaseConnection() {}
146 async destroy() {}
147 }
148
149 return {
150 createAdapter: () => new PostgresAdapter(),
151 createDriver: () => new DataApiDriver(),
152 createQueryCompiler: () => new DataApiQueryCompiler(),
153 createIntrospector: (db: import("kysely").Kysely<unknown>) =>
154 new PostgresIntrospector(db.withPlugin(introspectionPlugin)),
155 } as unknown as import("kysely").Dialect;
156 });
157
158export interface AuroraDataApiOptions {
159 /**
160 * Secrets Manager secret holding the database credentials. Required when
161 * passing a bare `DBCluster`; defaults to the composite's `secret` when
162 * passing an `AWS.RDS.Aurora` result.
163 */
164 readonly secret?: AWS.SecretsManager.Secret;
165 /** Database name inside the cluster. */
166 readonly database?: string;
167 /**
168 * Deploy-time automatic migration (over the Data API with ambient
169 * credentials). `false` disables.
170 * @default enabled
171 */
172 readonly migrate?: false;
173}
174
175/** The `AWS.RDS.Aurora` composite pieces this layer consumes. */
176interface AuroraLike {
177 readonly cluster: AWS.RDS.DBCluster;
178 readonly secret: AWS.SecretsManager.Secret;
179 readonly writer: AWS.RDS.DBInstance;
180}
181
182const isAuroraLike = (value: object): value is AuroraLike =>
183 "cluster" in value && "secret" in value && "writer" in value;
184
185/**
186 * A freshly-created cluster answers `DatabaseNotFoundException` until its
187 * writer instance registers, and a Serverless v2 cluster answers
188 * `DatabaseResumingException` while scaling from zero — both transient.
189 * Bounded per the speed doctrine (~1 minute of backoff).
190 */
191const transientRetry = <A, E extends { _tag: string }, R>(
192 effect: Effect.Effect<A, E, R>,
193): Effect.Effect<A, E, R> =>
194 effect.pipe(
195 Effect.retry({
196 while: (error) =>
197 error._tag === "DatabaseResumingException" ||
198 error._tag === "DatabaseNotFoundException",
199 schedule: Schedule.exponential("2 seconds", 1.5),
200 times: 8,
201 }),
202 );
203
204/**
205 * Aurora (RDS Data API) database layer for Better Auth — the optimal
206 * Lambda → Aurora pairing: SQL over HTTPS with IAM auth, no VPC
207 * attachment, no `pg`, no connection pooling concerns.
208 *
209 * Runtime access flows through the `AWS.RDSData.*` bindings, which grant
210 * the host `rds-data:*` + `secretsmanager:GetSecretValue` IAM and inject
211 * the cluster/secret ARNs. Requires the cluster to have the Data API
212 * enabled (`AWS.RDS.Aurora` enables it by default).
213 *
214 *
215 * ### Lambda with an Aurora-backed BetterAuth
216 * Pass the `AWS.RDS.Aurora` composite directly — the layer wires the
217 * cluster, credentials secret, and the writer-instance dependency (so
218 * deploy-time migrations wait for the cluster to be queryable) from one
219 * value.
220 * **Example:** Function URL serving auth over the Data API
221 * ```typescript
222 * import { BetterAuth } from "@alchemy.run/better-auth";
223 * import { AuroraDataApi } from "@alchemy.run/better-auth/AuroraDataApi";
224 * import * as AWS from "alchemy/AWS";
225 *
226 * export const Db = AWS.RDS.Aurora("AuthDb", { subnetIds, securityGroupIds });
227 *
228 * export default AuthFunction.make(
229 * { main, url: true, memorySize: 512 },
230 * Effect.gen(function* () {
231 * const auth = yield* BetterAuth({ emailAndPassword: { enabled: true } });
232 * return { fetch: ... };
233 * }).pipe(Effect.provide(AuroraDataApi(Db, { database: "postgres" }))),
234 * );
235 * ```
236 *
237 * ### Serverless v2 scale-from-zero
238 * A paused cluster answers `DatabaseResumingException` while waking; the
239 * layer retries the transient window with bounded backoff at both deploy
240 * and runtime.
241 * **Example:** Bare cluster + explicit secret
242 * ```typescript
243 * AuroraDataApi(cluster, { secret, database: "auth" })
244 * ```
245 *
246 * @layer
247 * @provides BetterAuth.Database
248 * @peer kysely
249 * @peer @distilled.cloud/aws
250 * @product Aurora
251 */
252export const AuroraDataApi = (
253 cluster:
254 | AWS.RDS.DBCluster
255 | AuroraLike
256 | Effect.Effect<AWS.RDS.DBCluster | AuroraLike, never, any>,
257 options?: AuroraDataApiOptions,
258) =>
259 Layer.effect(
260 Database,
261 Effect.gen(function* () {
262 const source = Effect.isEffect(cluster)
263 ? yield* cluster as Effect.Effect<AWS.RDS.DBCluster | AuroraLike>
264 : cluster;
265 const composite = isAuroraLike(source) ? source : undefined;
266 const db = composite?.cluster ?? (source as AWS.RDS.DBCluster);
267 const secret = options?.secret ?? composite?.secret;
268 if (secret === undefined) {
269 return yield* Effect.die(
270 new Error(
271 "AuroraDataApi: pass the AWS.RDS.Aurora composite, or a DBCluster together with `options.secret`.",
272 ),
273 );
274 }
275 const bindingOptions = {
276 secret,
277 ...(options?.database === undefined
278 ? {}
279 : { database: options.database }),
280 };
281 const executeStatement = yield* AWS.RDSData.ExecuteStatement(
282 db,
283 bindingOptions,
284 );
285 const beginTransaction = yield* AWS.RDSData.BeginTransaction(
286 db,
287 bindingOptions,
288 );
289 const commitTransaction = yield* AWS.RDSData.CommitTransaction(
290 db,
291 bindingOptions,
292 );
293 const rollbackTransaction = yield* AWS.RDSData.RollbackTransaction(
294 db,
295 bindingOptions,
296 );
297
298 const runtime = Effect.gen(function* () {
299 const context = yield* Effect.context<RuntimeContext>();
300 const run = <A, E>(
301 effect: Effect.Effect<A, E, RuntimeContext>,
302 ): Promise<A> =>
303 Effect.runPromise(
304 effect.pipe(Effect.provideContext(context)) as Effect.Effect<A, E>,
305 );
306 const dialect = yield* makeDataApiDialect({
307 execute: (request) =>
308 run(
309 transientRetry(
310 executeStatement(
311 request as AWS.RDSData.ExecuteStatementRequest,
312 ),
313 ),
314 ),
315 begin: () => run(transientRetry(beginTransaction())),
316 commit: (transactionId) => run(commitTransaction({ transactionId })),
317 rollback: (transactionId) =>
318 run(rollbackTransaction({ transactionId })),
319 });
320 return { dialect, type: "postgres" as const };
321 });
322
323 const service: DatabaseService = {
324 provider: "postgres",
325 runtime: runtime as DatabaseService["runtime"],
326 };
327
328 // Deploy-time migrations call the Data API through distilled with the
329 // ambient stack credentials (mirroring D1's deploy-time HTTP client).
330 // DCE'd from runtime bundles.
331 if (!globalThis.__ALCHEMY_RUNTIME__ && options?.migrate !== false) {
332 return {
333 ...service,
334 migrate: {
335 identity: {
336 clusterArn: db.dbClusterArn,
337 // depend on the WRITER instance too: a cluster with no
338 // registered instance answers DatabaseNotFoundException
339 ...(composite === undefined
340 ? {}
341 : { writerArn: composite.writer.dbInstanceArn }),
342 } as Record<string, unknown>,
343 connect: Effect.gen(function* () {
344 // Init half — capture the ARNs as Action dependencies.
345 const clusterArn = yield* db.dbClusterArn;
346 const secretArn = yield* secret.secretArn;
347 if (composite !== undefined) {
348 // capture-only: the migration must wait for the writer
349 yield* composite.writer.dbInstanceArn;
350 }
351 const ambient = yield* Effect.context<never>();
352 // Apply half — Data API dialect over distilled.
353 return Effect.gen(function* () {
354 const resourceArn = yield* clusterArn;
355 const resolvedSecretArn = yield* secretArn;
356 const rdsdata = yield* Effect.promise(
357 () => import("@distilled.cloud/aws/rds-data"),
358 );
359 const base = {
360 resourceArn,
361 secretArn: resolvedSecretArn,
362 ...(options?.database === undefined
363 ? {}
364 : { database: options.database }),
365 };
366 const run = <A>(effect: Effect.Effect<A, any, any>) =>
367 Effect.runPromise(
368 effect.pipe(
369 Effect.provideContext(ambient),
370 ) as Effect.Effect<A>,
371 );
372 const dialect = yield* makeDataApiDialect({
373 execute: (request) =>
374 run(
375 transientRetry(
376 rdsdata.executeStatement({
377 ...base,
378 ...request,
379 } as never),
380 ),
381 ),
382 begin: () =>
383 run(transientRetry(rdsdata.beginTransaction(base))),
384 commit: (transactionId) =>
385 run(
386 rdsdata.commitTransaction({
387 resourceArn,
388 secretArn: resolvedSecretArn,
389 transactionId,
390 }),
391 ),
392 rollback: (transactionId) =>
393 run(
394 rdsdata.rollbackTransaction({
395 resourceArn,
396 secretArn: resolvedSecretArn,
397 transactionId,
398 }),
399 ),
400 });
401 return { dialect, type: "postgres" } as DirectDatabase;
402 }).pipe(
403 Effect.catchDefect((cause: unknown) =>
404 Effect.fail(
405 new BetterAuthMigrationError({
406 message:
407 "Failed to reach the Aurora Data API for Better Auth schema migrations",
408 cause,
409 }),
410 ),
411 ),
412 );
413 }) as Effect.Effect<
414 Effect.Effect<
415 DirectDatabase,
416 BetterAuthMigrationError,
417 Scope.Scope
418 >,
419 never,
420 RuntimeContext
421 >,
422 },
423 } satisfies DatabaseService;
424 }
425 return service;
426 }),
427 ).pipe(
428 Layer.provide(AWS.RDSData.ExecuteStatementHttp),
429 Layer.provide(AWS.RDSData.BeginTransactionHttp),
430 Layer.provide(AWS.RDSData.CommitTransactionHttp),
431 Layer.provide(AWS.RDSData.RollbackTransactionHttp),
432 );