1import * as personalize from "@distilled.cloud/aws/personalize";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import * as Stream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
11import type { Providers } from "../Providers.ts";
12import { readPersonalizeTags, syncPersonalizeTags } from "./internal.ts";
13
14export interface EventTrackerProps {
15 /**
16 * Name of the event tracker. If omitted, a unique name is generated from
17 * the app, stage, and logical ID. Changing the name replaces the tracker.
18 */
19 name?: string;
20 /**
21 * ARN of the dataset group the event tracker records events for. The
22 * dataset group should contain an Interactions dataset (events are recorded
23 * into it). Only one event tracker can exist per dataset group. Immutable —
24 * changing it replaces the tracker.
25 */
26 datasetGroupArn: string;
27 /**
28 * User-defined tags for the event tracker.
29 */
30 tags?: Record<string, string>;
31}
32
33export interface EventTracker extends Resource<
34 "AWS.Personalize.EventTracker",
35 EventTrackerProps,
36 {
37 /**
38 * ARN of the event tracker.
39 */
40 eventTrackerArn: string;
41 /**
42 * The tracking ID passed as `trackingId` to the PutEvents /
43 * PutActionInteractions data-plane operations.
44 */
45 trackingId: string;
46 /**
47 * Name of the event tracker.
48 */
49 name: string;
50 /**
51 * Event tracker status (e.g. `ACTIVE`, `CREATE PENDING`).
52 */
53 status: string;
54 /**
55 * ARN of the dataset group the tracker records events for.
56 */
57 datasetGroupArn: string;
58 },
59 never,
60 Providers
61> {}
62
63/**
64 * An Amazon Personalize event tracker — the ingestion endpoint for streaming
65 * interaction events into a dataset group's Interactions dataset. Creating a
66 * tracker yields a `trackingId` that the {@link PutEvents} data-plane binding
67 * uses to record events in real time.
68 *
69 * ### Creating an Event Tracker
70 * **Example:** Track Events for a Dataset Group
71 * ```typescript
72 * const tracker = yield* Personalize.EventTracker("Tracker", {
73 * datasetGroupArn: group.datasetGroupArn,
74 * });
75 * ```
76 *
77 * ### Streaming Events
78 * **Example:** Record Click Events from a Lambda
79 * ```typescript
80 * // init
81 * const putEvents = yield* Personalize.PutEvents(tracker);
82 *
83 * // runtime
84 * yield* putEvents({
85 * sessionId: "session-1",
86 * userId: "user-1",
87 * eventList: [{ eventType: "click", itemId: "item-42", sentAt: new Date() }],
88 * });
89 * ```
90 *
91 * @resource
92 */
93export const EventTracker = Resource<EventTracker>(
94 "AWS.Personalize.EventTracker",
95);
96
97export const EventTrackerProvider = () =>
98 Provider.effect(
99 EventTracker,
100 Effect.gen(function* () {
101 const createName = Effect.fn(function* (
102 id: string,
103 props: EventTrackerProps,
104 ) {
105 return props.name ?? (yield* createPhysicalName({ id, maxLength: 63 }));
106 });
107
108 const describe = Effect.fn(function* (eventTrackerArn: string) {
109 const response = yield* personalize
110 .describeEventTracker({ eventTrackerArn })
111 .pipe(
112 Effect.catchTag("ResourceNotFoundException", () =>
113 Effect.succeed(undefined),
114 ),
115 );
116 return response?.eventTracker;
117 });
118
119 /** Poll until the tracker reaches ACTIVE; fail fast on CREATE FAILED. */
120 const waitActive = Effect.fn(function* (eventTrackerArn: string) {
121 const tracker = yield* describe(eventTrackerArn).pipe(
122 Effect.repeat({
123 schedule: Schedule.max([
124 Schedule.fixed("2 seconds"),
125 Schedule.recurs(40),
126 ]),
127 until: (t) =>
128 t?.status === "ACTIVE" || (t?.status ?? "").includes("FAILED"),
129 }),
130 );
131 if (tracker?.status !== "ACTIVE") {
132 return yield* Effect.fail(
133 new Error(
134 `Personalize event tracker ${eventTrackerArn} did not become ACTIVE (status: ${tracker?.status})`,
135 ),
136 );
137 }
138 return tracker;
139 });
140
141 /** Find an existing tracker's ARN by name within its dataset group. */
142 const findArnByName = Effect.fn(function* (
143 name: string,
144 datasetGroupArn: string,
145 ) {
146 const pages = yield* personalize.listEventTrackers
147 .pages({ datasetGroupArn })
148 .pipe(Stream.runCollect);
149 return Array.from(pages)
150 .flatMap((page) => page.eventTrackers ?? [])
151 .find((summary) => summary.name === name)?.eventTrackerArn;
152 });
153
154 const toAttrs = (tracker: personalize.EventTracker) => ({
155 eventTrackerArn: tracker.eventTrackerArn!,
156 trackingId: tracker.trackingId!,
157 name: tracker.name!,
158 status: tracker.status!,
159 datasetGroupArn: tracker.datasetGroupArn!,
160 });
161
162 return {
163 stables: ["eventTrackerArn", "trackingId", "name"],
164
165 diff: Effect.fn(function* ({ id, olds, news }) {
166 if (!isResolved(news)) return undefined;
167 const oldName = yield* createName(id, olds);
168 const newName = yield* createName(id, news);
169 if (
170 oldName !== newName ||
171 (olds.datasetGroupArn ?? undefined) !==
172 (news.datasetGroupArn ?? undefined)
173 ) {
174 return { action: "replace" } as const;
175 }
176 }),
177
178 read: Effect.fn(function* ({ id, output }) {
179 if (!output?.eventTrackerArn) return undefined;
180 const tracker = yield* describe(output.eventTrackerArn);
181 if (tracker === undefined) return undefined;
182 const attrs = toAttrs(tracker);
183 const tags = yield* readPersonalizeTags(tracker.eventTrackerArn!);
184 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
185 }),
186
187 reconcile: Effect.fn(function* ({ id, news, output, session }) {
188 const name = yield* createName(id, news);
189 const internalTags = yield* createInternalTags(id);
190 const desiredTags = { ...internalTags, ...news.tags };
191
192 // 1. Observe — cloud state is authoritative; output is an ARN cache.
193 let tracker =
194 output?.eventTrackerArn !== undefined
195 ? yield* describe(output.eventTrackerArn)
196 : undefined;
197
198 // 2. Ensure — create if missing, then wait for ACTIVE. A crashed
199 // prior run may have left a same-named tracker behind with no
200 // persisted state — adopt it by name and converge.
201 if (tracker === undefined) {
202 const arn = yield* personalize
203 .createEventTracker({
204 name,
205 datasetGroupArn: news.datasetGroupArn,
206 tags: Object.entries(desiredTags).map(([tagKey, tagValue]) => ({
207 tagKey,
208 tagValue,
209 })),
210 })
211 .pipe(
212 Effect.map((created) => created.eventTrackerArn!),
213 Effect.catchTag("ResourceAlreadyExistsException", (error) =>
214 findArnByName(name, news.datasetGroupArn).pipe(
215 Effect.flatMap((existing) =>
216 existing === undefined
217 ? Effect.fail(error)
218 : Effect.succeed(existing),
219 ),
220 ),
221 ),
222 );
223 tracker = yield* waitActive(arn);
224 }
225
226 // 3. Sync tags — diff against OBSERVED cloud tags (no-op after a
227 // fresh create; converges an adopted leftover).
228 yield* syncPersonalizeTags(tracker.eventTrackerArn!, desiredTags);
229
230 yield* session.note(tracker.eventTrackerArn!);
231 return toAttrs(tracker);
232 }),
233
234 delete: Effect.fn(function* ({ output }) {
235 yield* personalize
236 .deleteEventTracker({ eventTrackerArn: output.eventTrackerArn })
237 .pipe(
238 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
239 Effect.retry({
240 while: (e) => e._tag === "ResourceInUseException",
241 schedule: Schedule.max([
242 Schedule.fixed("3 seconds"),
243 Schedule.recurs(20),
244 ]),
245 }),
246 );
247 // Deletion is asynchronous — wait until the tracker is actually gone
248 // so its dataset group can delete without ResourceInUse churn.
249 const remaining = yield* describe(output.eventTrackerArn).pipe(
250 Effect.repeat({
251 schedule: Schedule.max([
252 Schedule.fixed("3 seconds"),
253 Schedule.recurs(40),
254 ]),
255 until: (tracker): boolean => tracker === undefined,
256 }),
257 );
258 if (remaining !== undefined) {
259 return yield* Effect.fail(
260 new Error(
261 `Personalize event tracker ${output.eventTrackerArn} was not deleted (status: ${remaining.status})`,
262 ),
263 );
264 }
265 }),
266
267 list: () =>
268 personalize.listEventTrackers.pages({}).pipe(
269 Stream.runCollect,
270 Effect.map((chunk) =>
271 Array.from(chunk).flatMap((page) => page.eventTrackers ?? []),
272 ),
273 Effect.flatMap(
274 Effect.forEach(
275 (summary) =>
276 describe(summary.eventTrackerArn!).pipe(
277 Effect.map((t) => (t ? toAttrs(t) : undefined)),
278 ),
279 { concurrency: 4 },
280 ),
281 ),
282 Effect.map((items) => items.filter((item) => item !== undefined)),
283 ),
284 };
285 }),
286 );