1import { Endpoint } from "@distilled.cloud/aws";
2import * as TSQ from "@distilled.cloud/aws/timestream-query";
3import * as TSW from "@distilled.cloud/aws/timestream-write";
4import * as Effect from "effect/Effect";
5
6/**
7 * Amazon Timestream uses the **endpoint discovery** pattern: control-plane and
8 * data-plane requests must be routed to a cell-specific endpoint returned by
9 * `DescribeEndpoints`, not the static regional endpoint (which answers every
10 * operation other than `DescribeEndpoints` with `UnknownOperationException`).
11 *
12 * These helpers call `DescribeEndpoints` once, cache the discovered address for
13 * its advertised `CachePeriodInMinutes`, and wrap a distilled Timestream effect
14 * so it targets the discovered endpoint via the `Endpoint` service override.
15 */
16
17type Kind = "write" | "query";
18
19interface CachedEndpoint {
20 readonly url: string;
21 readonly expiresAt: number;
22}
23
24// Module-level cache keyed by discovery kind. `DescribeEndpoints` is a cheap
25// call but the API asks clients to respect the returned cache period; caching
26// also avoids an extra round-trip on every write/query.
27const cache = new Map<Kind, CachedEndpoint>();
28
29/**
30 * Discover (and cache) the cell-specific endpoint for `kind`, using the given
31 * `DescribeEndpoints` effect. Parameterized over the discovery effect so each
32 * caller carries only its own service module's error union — and so binding
33 * layers can pass an operation captured via yield-first (`yield* op`) whose
34 * calls are requirement-free.
35 */
36export const discover = <E, R>(
37 kind: Kind,
38 describe: Effect.Effect<
39 { Endpoints: { Address: string; CachePeriodInMinutes: number }[] },
40 E,
41 R
42 >,
43): Effect.Effect<string, E, R> =>
44 Effect.gen(function* () {
45 const now = Date.now();
46 const cached = cache.get(kind);
47 if (cached !== undefined && cached.expiresAt > now) {
48 return cached.url;
49 }
50 const response = yield* describe;
51 const endpoint = response.Endpoints[0];
52 const url = `https://${endpoint.Address}`;
53 cache.set(kind, {
54 url,
55 expiresAt: now + (endpoint.CachePeriodInMinutes ?? 5) * 60_000,
56 });
57 return url;
58 });
59
60/**
61 * Wrap a distilled Timestream effect so it targets the endpoint produced by
62 * `discovered` (see {@link discover}) via the `Endpoint` service override.
63 */
64export const withEndpoint =
65 <EDisc, RDisc>(discovered: Effect.Effect<string, EDisc, RDisc>) =>
66 <A, E, R>(
67 effect: Effect.Effect<A, E, R>,
68 ): Effect.Effect<A, E | EDisc, Exclude<R, Endpoint.Endpoint> | RDisc> =>
69 discovered.pipe(
70 Effect.flatMap((url) =>
71 effect.pipe(
72 Effect.provideService(Endpoint.Endpoint, Effect.succeed(url)),
73 ),
74 ),
75 );
76
77/**
78 * Route a distilled `timestream-write` effect through the discovered ingest
79 * endpoint. Adds `DescribeEndpoints`'s error union (including the synthetic
80 * `TimestreamNotOnboarded` tag) to the wrapped effect's errors.
81 */
82export const withWriteEndpoint = withEndpoint(
83 discover("write", TSW.describeEndpoints({})),
84);
85
86/**
87 * Route a distilled `timestream-query` effect through the discovered query
88 * endpoint.
89 */
90export const withQueryEndpoint = withEndpoint(
91 discover("query", TSQ.describeEndpoints({})),
92);