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