1import * as Alchemy from "alchemy";
2import * as Cloudflare from "alchemy/Cloudflare";
3import * as Crypto from "effect/Crypto";
4import * as Context from "effect/Context";
5import * as DateTime from "effect/DateTime";
6import * as Effect from "effect/Effect";
7import * as Layer from "effect/Layer";
8import * as Schema from "effect/Schema";
9
10import {
11 RelayDeliveryKind as RelayDeliveryKindSchema,
12 type RelayDeliveryResult,
13} from "@t3tools/contracts/relay";
14
15import {
16 sanitizeAgentActivityAggregateState,
17 sanitizeApnsNotificationPayload,
18} from "./agentActivityPayloads.ts";
19import {
20 expiresAtForJob,
21 makeApnsDeliveryJobPayload,
22 signApnsDeliveryJob,
23 type ApnsDeliveryJobPayload,
24 type SignedApnsDeliveryJob,
25} from "./apnsDeliveryJobs.ts";
26import * as RelayConfiguration from "../Config.ts";
27
28export class ApnsDeliveryQueueSendError extends Schema.TaggedError<ApnsDeliveryQueueSendError>()(
29 "ApnsDeliveryQueueSendError",
30 {
31 operation: Schema.Literals(["generate-job-id", "send"]),
32 jobId: Schema.NullOr(Schema.String),
33 kind: RelayDeliveryKindSchema,
34 userId: Schema.String,
35 deviceId: Schema.String,
36 cause: Schema.Defect(),
37 },
38) {
39 override get message(): string {
40 return `Failed to enqueue APNs ${this.kind.replaceAll("_", " ")} delivery during ${this.operation} for device ${this.deviceId}.`;
41 }
42}
43
44export type ApnsDeliveryQueueError = ApnsDeliveryQueueSendError;
45
46export class ApnsDeliveryQueueSender extends Context.Service<
47 ApnsDeliveryQueueSender,
48 {
49 readonly send: (
50 body: SignedApnsDeliveryJob,
51 ) => Effect.Effect<void, Cloudflare.Queues.SendError>;
52 }
53>()("t3code-relay/agentActivity/ApnsDeliveryQueue/ApnsDeliveryQueueSender") {}
54
55export class ApnsDeliveryQueue extends Context.Service<
56 ApnsDeliveryQueue,
57 {
58 readonly enqueueLiveActivity: (input: {
59 readonly kind: ApnsDeliveryJobPayload["kind"];
60 readonly userId: string;
61 readonly deviceId: string;
62 readonly token: string;
63 readonly bundleId?: string | null;
64 readonly apsEnvironment?: "sandbox" | "production" | null;
65 readonly aggregate: ApnsDeliveryJobPayload["aggregate"];
66 readonly alert?: ApnsDeliveryJobPayload["alert"];
67 }) => Effect.Effect<RelayDeliveryResult, ApnsDeliveryQueueError>;
68 readonly enqueuePushNotification: (input: {
69 readonly userId: string;
70 readonly deviceId: string;
71 readonly token: string;
72 readonly bundleId?: string | null;
73 readonly apsEnvironment?: "sandbox" | "production" | null;
74 readonly notification: NonNullable<ApnsDeliveryJobPayload["notification"]>;
75 }) => Effect.Effect<RelayDeliveryResult, ApnsDeliveryQueueError>;
76 }
77>()("t3code-relay/agentActivity/ApnsDeliveryQueue") {}
78
79export const make = Effect.gen(function* () {
80 const sender = yield* ApnsDeliveryQueueSender;
81 const crypto = yield* Crypto.Crypto;
82 const config = yield* RelayConfiguration.RelayConfiguration;
83
84 return ApnsDeliveryQueue.of({
85 enqueueLiveActivity: Effect.fn("relay.apns_delivery_queue.enqueue_live_activity")(
86 function* (input) {
87 yield* Effect.annotateCurrentSpan({
88 "relay.mobile.device_id": input.deviceId,
89 "relay.delivery.kind": input.kind,
90 });
91 const now = yield* DateTime.now;
92 const jobId = yield* crypto.randomUUIDv4().pipe(
93 Effect.mapError(
94 (cause) =>
95 new ApnsDeliveryQueueSendError({
96 operation: "generate-job-id",
97 jobId: null,
98 kind: input.kind,
99 userId: input.userId,
100 deviceId: input.deviceId,
101 cause,
102 }),
103 ),
104 );
105 yield* Effect.annotateCurrentSpan({ "relay.delivery.job_id": jobId });
106 const payload = makeApnsDeliveryJobPayload({
107 ...input,
108 aggregate:
109 input.aggregate === null ? null : sanitizeAgentActivityAggregateState(input.aggregate),
110 jobId,
111 createdAt: DateTime.formatIso(now),
112 expiresAt: expiresAtForJob(now.epochMilliseconds),
113 });
114 const signed = signApnsDeliveryJob({
115 secret: config.apnsDeliveryJobSigningSecret,
116 payload,
117 });
118 yield* sender.send(signed).pipe(
119 Effect.mapError(
120 (cause) =>
121 new ApnsDeliveryQueueSendError({
122 operation: "send",
123 jobId,
124 kind: input.kind,
125 userId: input.userId,
126 deviceId: input.deviceId,
127 cause,
128 }),
129 ),
130 );
131 return {
132 deviceId: input.deviceId,
133 kind: input.kind,
134 ok: true,
135 queued: true,
136 apnsStatus: null,
137 apnsReason: null,
138 apnsId: null,
139 };
140 },
141 ),
142 enqueuePushNotification: Effect.fn("relay.apns_delivery_queue.enqueue_push_notification")(
143 function* (input) {
144 yield* Effect.annotateCurrentSpan({
145 "relay.mobile.device_id": input.deviceId,
146 "relay.delivery.kind": "push_notification",
147 "relay.environment_id": input.notification.environmentId,
148 "relay.thread_id": input.notification.threadId,
149 });
150 const now = yield* DateTime.now;
151 const jobId = yield* crypto.randomUUIDv4().pipe(
152 Effect.mapError(
153 (cause) =>
154 new ApnsDeliveryQueueSendError({
155 operation: "generate-job-id",
156 jobId: null,
157 kind: "push_notification",
158 userId: input.userId,
159 deviceId: input.deviceId,
160 cause,
161 }),
162 ),
163 );
164 yield* Effect.annotateCurrentSpan({ "relay.delivery.job_id": jobId });
165 const payload = makeApnsDeliveryJobPayload({
166 kind: "push_notification",
167 userId: input.userId,
168 deviceId: input.deviceId,
169 token: input.token,
170 bundleId: input.bundleId,
171 apsEnvironment: input.apsEnvironment,
172 aggregate: null,
173 notification: sanitizeApnsNotificationPayload(input.notification),
174 jobId,
175 createdAt: DateTime.formatIso(now),
176 expiresAt: expiresAtForJob(now.epochMilliseconds),
177 });
178 const signed = signApnsDeliveryJob({
179 secret: config.apnsDeliveryJobSigningSecret,
180 payload,
181 });
182 yield* sender.send(signed).pipe(
183 Effect.mapError(
184 (cause) =>
185 new ApnsDeliveryQueueSendError({
186 operation: "send",
187 jobId,
188 kind: "push_notification",
189 userId: input.userId,
190 deviceId: input.deviceId,
191 cause,
192 }),
193 ),
194 );
195 return {
196 deviceId: input.deviceId,
197 kind: "push_notification" as const,
198 ok: true,
199 queued: true,
200 apnsStatus: null,
201 apnsReason: null,
202 apnsId: null,
203 };
204 },
205 ),
206 });
207});
208
209export const layer = Layer.effect(ApnsDeliveryQueue, make);
210
211export const layerCloudflareQueues = (
212 sender: Cloudflare.Queues.WriteQueueClient,
213 alchemyRuntimeContext: Alchemy.BaseRuntimeContext,
214) =>
215 layer.pipe(
216 Layer.provide(
217 Layer.succeed(
218 ApnsDeliveryQueueSender,
219 ApnsDeliveryQueueSender.of({
220 send: (body) =>
221 sender
222 .send(body)
223 .pipe(Effect.provideService(Alchemy.RuntimeContext, alchemyRuntimeContext)),
224 }),
225 ),
226 ),
227 );