1import * as ivschat from "@distilled.cloud/aws/ivschat";
2import * as AWS from "@/AWS";
3import * as Core from "@/Test/Core";
4import * as Test from "@/Test/Alchemy";
5import { describe, expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9import * as HttpClient from "effect/http/HttpClient";
10import * as HttpClientRequest from "effect/http/HttpClientRequest";
11import WebSocket from "ws";
12
13import IVSChatTestFunctionLive, {
14 IVSChatTestFunction,
15} from "./fixtures/handler.ts";
16
17const testOptions = { providers: AWS.providers() };
18const { test, beforeAll, afterAll } = Test.make(testOptions);
19const sharedStack = Core.scratchStack(testOptions, "IVSChatBindings");
20
21const readinessPolicy = Schedule.max([
22 Schedule.fixed("2 seconds"),
23 Schedule.recurs(60),
24]);
25
26let baseUrl: string;
27
28class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
29 readonly status: number;
30 readonly body: string;
31}> {}
32
33class WsFailure extends Data.TaggedError("WsFailure")<{
34 readonly reason: string;
35}> {}
36
37const post = (path: string) =>
38 HttpClient.execute(HttpClientRequest.post(`${baseUrl}${path}`)).pipe(
39 Effect.flatMap((response) =>
40 response.status >= 500
41 ? response.text.pipe(
42 Effect.flatMap((body) =>
43 Effect.fail(
44 new TransientUpstream({ status: response.status, body }),
45 ),
46 ),
47 )
48 : Effect.succeed(response),
49 ),
50 Effect.retry({
51 while: (e) => e._tag === "TransientUpstream",
52 schedule: Schedule.max([
53 Schedule.exponential("1 second"),
54 Schedule.recurs(5),
55 ]),
56 }),
57 );
58
59describe(
60 "IVSChat Bindings",
61 {
62 tags: [
63 "provider:aws",
64 "provider:aws:ivschat",
65 "provider:aws:lambda",
66 "live",
67 ],
68 },
69 () => {
70 beforeAll(
71 Effect.gen(function* () {
72 yield* Effect.logInfo(
73 "IVSChat test setup: destroying previous resources",
74 );
75 yield* sharedStack.destroy();
76
77 yield* Effect.logInfo("IVSChat test setup: deploying fixture");
78 const { functionUrl } = yield* sharedStack.deploy(
79 Effect.gen(function* () {
80 return yield* IVSChatTestFunction;
81 }).pipe(Effect.provide(IVSChatTestFunctionLive)),
82 );
83
84 expect(functionUrl).toBeTruthy();
85 baseUrl = functionUrl!.replace(/\/+$/, "");
86 const readinessUrl = `${baseUrl}/ping`;
87
88 yield* Effect.logInfo(
89 `IVSChat test setup: probing readiness at ${readinessUrl}`,
90 );
91 yield* HttpClient.get(readinessUrl).pipe(
92 Effect.flatMap((response) =>
93 response.status === 200
94 ? Effect.succeed(response)
95 : Effect.fail(
96 new Error(`Function not ready: ${response.status}`),
97 ),
98 ),
99 Effect.tapError((error) =>
100 Effect.logWarning(
101 `IVSChat test setup: fixture not ready yet (${String(error)})`,
102 ),
103 ),
104 Effect.retry({ schedule: readinessPolicy }),
105 );
106 }),
107 { timeout: 240_000 },
108 );
109
110 afterAll(sharedStack.destroy(), { timeout: 120_000 });
111
112 describe("IVSChat.CreateChatToken", () => {
113 test.provider(
114 "mints a redacted chat token honoring sessionDuration",
115 (_stack) =>
116 Effect.gen(function* () {
117 const response = (yield* post("/token").pipe(
118 Effect.flatMap((r) => r.json),
119 )) as {
120 tokenLength: number;
121 tokenIsRedacted: boolean;
122 tokenExpirationTime?: string;
123 sessionExpirationTime?: string;
124 };
125
126 expect(response.tokenLength).toBeGreaterThan(0);
127 // SensitiveString in distilled decodes to a Redacted value.
128 expect(response.tokenIsRedacted).toBe(true);
129 expect(response.tokenExpirationTime).toBeTruthy();
130 expect(response.sessionExpirationTime).toBeTruthy();
131
132 // sessionDuration: "30 minutes" must reach the wire as
133 // sessionDurationInMinutes: 30 (the API default is 60).
134 const sessionMinutes =
135 (new Date(response.sessionExpirationTime!).getTime() -
136 Date.now()) /
137 60_000;
138 expect(sessionMinutes).toBeGreaterThan(20);
139 expect(sessionMinutes).toBeLessThan(40);
140 }),
141 { timeout: 120_000 },
142 );
143 });
144
145 describe("IVSChat.SendEvent", () => {
146 test.provider(
147 "broadcasts an application event to the room",
148 (_stack) =>
149 Effect.gen(function* () {
150 const response = (yield* post("/send-event").pipe(
151 Effect.flatMap((r) => r.json),
152 )) as { id?: string };
153
154 expect(typeof response.id).toBe("string");
155 expect(response.id!.length).toBeGreaterThan(0);
156 }),
157 { timeout: 120_000 },
158 );
159 });
160
161 describe("IVSChat.DeleteMessage", () => {
162 test.provider(
163 "broadcasts a DELETEMESSAGE moderation event",
164 (_stack) =>
165 Effect.gen(function* () {
166 const response = (yield* post("/delete-message").pipe(
167 Effect.flatMap((r) => r.json),
168 )) as { deleted?: string };
169
170 expect(typeof response.deleted).toBe("string");
171 expect(response.deleted!.length).toBeGreaterThan(0);
172 }),
173 { timeout: 120_000 },
174 );
175 });
176
177 describe("IVSChat.RoomMessageReviewEventSource", () => {
178 /**
179 * The IVS Chat messaging wire format is PascalCase — send
180 * `{Action, Content, RequestId}`, receive `{Type, Content, ErrorCode,
181 * ErrorMessage}` frames.
182 */
183 interface ChatFrame {
184 Type: string;
185 Content?: string;
186 ErrorCode?: number;
187 ErrorMessage?: string;
188 }
189
190 /**
191 * Connect to the room's WebSocket messaging endpoint with a chat token,
192 * send one message, and resolve with the first frame the room sends
193 * back for it (`MESSAGE` when the review handler allows, `ERROR` when
194 * it denies). Encapsulated in `Effect.callback` so fiber interruption
195 * closes the socket.
196 */
197 const wsSendMessage = (
198 endpoint: string,
199 token: string,
200 content: string,
201 ) =>
202 Effect.callback<ChatFrame, WsFailure>((resume, signal) => {
203 const socket = new WebSocket(endpoint, token);
204 let settled = false;
205 const settle = (effect: Effect.Effect<ChatFrame, WsFailure>) => {
206 if (settled) return;
207 settled = true;
208 try {
209 socket.close();
210 } catch {
211 // already closed
212 }
213 resume(effect);
214 };
215 signal.addEventListener("abort", () =>
216 settle(Effect.fail(new WsFailure({ reason: "interrupted" }))),
217 );
218 socket.on("open", () =>
219 socket.send(
220 JSON.stringify({
221 Action: "SEND_MESSAGE",
222 RequestId: "alchemy-review-test",
223 Content: content,
224 }),
225 ),
226 );
227 socket.on("message", (data) => {
228 let frame: ChatFrame;
229 try {
230 frame = JSON.parse(String(data));
231 } catch {
232 return; // ignore unparsable frames
233 }
234 // MESSAGE (delivered) and ERROR (denied) both settle the probe;
235 // ignore unrelated frames (e.g. other EVENTs).
236 if (frame.Type === "MESSAGE" || frame.Type === "ERROR") {
237 settle(Effect.succeed(frame));
238 }
239 });
240 socket.on("error", (error) =>
241 settle(Effect.fail(new WsFailure({ reason: String(error) }))),
242 );
243 socket.on("close", (code) =>
244 settle(
245 Effect.fail(new WsFailure({ reason: `closed early (${code})` })),
246 ),
247 );
248 }).pipe(
249 Effect.timeout("15 seconds"),
250 Effect.catchTag("TimeoutError", () =>
251 Effect.fail(new WsFailure({ reason: "no frame within 15s" })),
252 ),
253 );
254
255 const wsInfo = Effect.gen(function* () {
256 const info = (yield* post("/ws-info").pipe(
257 Effect.flatMap((r) => r.json),
258 )) as { token?: string; endpoint?: string; roomArn?: string };
259 expect(info.token).toBeTruthy();
260 expect(info.endpoint).toContain("wss://edge.ivschat.");
261 return info as { token: string; endpoint: string; roomArn: string };
262 });
263
264 test.provider(
265 "associates the handler + invoke permission on deploy",
266 (_stack) =>
267 Effect.gen(function* () {
268 const { roomArn } = yield* wsInfo;
269 // Out-of-band: the deployed room must carry the fixture Lambda as
270 // its messageReviewHandler (set through the binding contract).
271 const room = yield* ivschat.getRoom({ identifier: roomArn });
272 expect(room.messageReviewHandler?.uri).toContain(":function:");
273 expect(room.messageReviewHandler?.fallbackResult ?? "ALLOW").toBe(
274 "ALLOW",
275 );
276 }),
277 { timeout: 120_000 },
278 );
279
280 test.provider(
281 "review handler modifies allowed messages",
282 (_stack) =>
283 Effect.gen(function* () {
284 // The handler association + invoke Permission propagate shortly
285 // after deploy; until then IVS Chat's fallbackResult (ALLOW)
286 // delivers the ORIGINAL content — treat that as retryable.
287 const frame = yield* Effect.gen(function* () {
288 const { token, endpoint } = yield* wsInfo;
289 const received = yield* wsSendMessage(
290 endpoint,
291 token,
292 "hello moderators",
293 );
294 if (
295 received.Type !== "MESSAGE" ||
296 !received.Content?.includes("[reviewed]")
297 ) {
298 return yield* Effect.fail(
299 new WsFailure({
300 reason: `not yet reviewed: ${JSON.stringify(received)}`,
301 }),
302 );
303 }
304 return received;
305 }).pipe(
306 Effect.tapError((e) =>
307 Effect.logInfo(
308 `review-allow attempt failed: ${e._tag === "WsFailure" ? e.reason : String(e)}`,
309 ),
310 ),
311 Effect.retry({
312 schedule: Schedule.max([
313 Schedule.exponential("2 seconds"),
314 Schedule.recurs(5),
315 ]),
316 }),
317 );
318
319 expect(frame.Type).toBe("MESSAGE");
320 expect(frame.Content).toBe("hello moderators [reviewed]");
321 }),
322 { timeout: 120_000 },
323 );
324
325 test.provider(
326 "review handler denies flagged messages with a 406",
327 (_stack) =>
328 Effect.gen(function* () {
329 const frame = yield* Effect.gen(function* () {
330 const { token, endpoint } = yield* wsInfo;
331 const received = yield* wsSendMessage(
332 endpoint,
333 token,
334 "please deny-me now",
335 );
336 if (received.Type !== "ERROR") {
337 return yield* Effect.fail(
338 new WsFailure({
339 reason: `not yet denied: ${JSON.stringify(received)}`,
340 }),
341 );
342 }
343 return received;
344 }).pipe(
345 Effect.tapError((e) =>
346 Effect.logInfo(
347 `review-deny attempt failed: ${e._tag === "WsFailure" ? e.reason : String(e)}`,
348 ),
349 ),
350 Effect.retry({
351 schedule: Schedule.max([
352 Schedule.exponential("2 seconds"),
353 Schedule.recurs(5),
354 ]),
355 }),
356 );
357
358 expect(frame.Type).toBe("ERROR");
359 expect(frame.ErrorCode).toBe(406);
360 expect(frame.ErrorMessage).toContain("alchemy-moderated");
361 }),
362 { timeout: 120_000 },
363 );
364 });
365
366 describe("IVSChat.DisconnectUser", () => {
367 test.provider(
368 "disconnects a user's room connections",
369 (_stack) =>
370 Effect.gen(function* () {
371 const response = (yield* post("/disconnect-user").pipe(
372 Effect.flatMap((r) => r.json),
373 )) as { ok?: boolean };
374
375 expect(response.ok).toBe(true);
376 }),
377 { timeout: 120_000 },
378 );
379 });
380 },
381);