1import * as GCP from "alchemy/GCP";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import * as HttpServerResponse from "effect/http/HttpServerResponse";
6import {
7 DeadOrderEvents,
8 decodeOrderEvent,
9 emailObjectFor,
10 OrderEvents,
11 Outbox,
12 type OrderEvent,
13} from "./resources.ts";
14
15/** How many times Pub/Sub tries a message before dead-lettering it. */
16export const MAX_DELIVERY_ATTEMPTS = 5;
17
18class UndeliverableEmail extends Data.TaggedError("UndeliverableEmail")<{
19 to: string;
20}> {}
21
22/**
23 * Stand-in for a mail provider. `.invalid` is reserved (RFC 2606) and
24 * never resolves, so mail to it always fails — a message this consumer
25 * can never process, which is what the dead-letter topic is for.
26 */
27interface SentEmail {
28 to: string;
29 subject: string;
30 text: string;
31 sentAt: string;
32}
33
34const send = (
35 event: OrderEvent,
36): Effect.Effect<SentEmail, UndeliverableEmail> =>
37 event.email.endsWith(".invalid")
38 ? Effect.fail(new UndeliverableEmail({ to: event.email }))
39 : Effect.succeed({
40 to: event.email,
41 subject: `Order ${event.orderId} confirmed`,
42 text: `Thanks! We received your order totalling $${event.total.toFixed(2)}.`,
43 sentAt: new Date().toISOString(),
44 });
45
46/**
47 * Sends an order confirmation for every `order.created` event.
48 *
49 * The `filter` is applied by Pub/Sub on the subscription, so other event
50 * types are never delivered here at all — this service is not woken up
51 * (or billed) for them.
52 *
53 * A failed send fails the push; Pub/Sub retries it, and after
54 * {@link MAX_DELIVERY_ATTEMPTS} attempts forwards it to
55 * `DeadOrderEvents` instead of retrying forever.
56 */
57export default class Email extends GCP.Function<Email>()(
58 "Email",
59 {
60 main: import.meta.url,
61 },
62 Effect.gen(function* () {
63 const outbox = yield* GCP.Storage.WriteBucket(Outbox);
64
65 yield* GCP.PubSub.consumeTopicMessages(
66 OrderEvents,
67 {
68 filter: 'attributes.type = "order.created"',
69 deadLetter: {
70 topic: yield* DeadOrderEvents,
71 maxDeliveryAttempts: MAX_DELIVERY_ATTEMPTS,
72 },
73 },
74 (messages) =>
75 messages.pipe(
76 Stream.runForEach(({ message, deliveryAttempt }) =>
77 Effect.gen(function* () {
78 const event = yield* decodeOrderEvent(message.data);
79 const email = yield* send(event).pipe(
80 Effect.tapError((error) =>
81 Effect.logWarning(
82 `email: cannot send to ${error.to} (attempt ${deliveryAttempt ?? 1})`,
83 ),
84 ),
85 );
86 // Keyed by event id, so a redelivery overwrites instead of
87 // recording a second email.
88 yield* outbox.put(
89 emailObjectFor(event.eventId),
90 JSON.stringify(email),
91 {
92 contentType: "application/json",
93 metadata: { orderId: event.orderId, to: email.to },
94 },
95 );
96 yield* Effect.log(`email: confirmed ${event.orderId}`);
97 }),
98 ),
99 // A failure answers the push with a 500, so Pub/Sub redelivers.
100 Effect.orDie,
101 ),
102 );
103
104 return {
105 fetch: Effect.succeed(HttpServerResponse.text("ok")),
106 };
107 }).pipe(
108 Effect.provide([GCP.Run.TopicEventSource, GCP.Storage.WriteBucketHttp]),
109 ),
110) {}