1import * as Context from "effect/Context";
2import * as DateTime from "effect/DateTime";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Option from "effect/Option";
6import { and, eq, isNull } from "drizzle-orm";
7import * as Crypto from "effect/Crypto";
8import * as Schema from "effect/Schema";
9
10import * as RelayDb from "../db.ts";
11import { relayDeliveryAttempts } from "../persistence/schema.ts";
12
13export class DeliveryAttemptRecordPersistenceError extends Schema.TaggedError<DeliveryAttemptRecordPersistenceError>()(
14 "DeliveryAttemptRecordPersistenceError",
15 {
16 operation: Schema.Literals(["record", "claim-source-job", "complete-source-job"]),
17 sourceJobId: Schema.NullOr(Schema.String),
18 userId: Schema.NullOr(Schema.String),
19 environmentId: Schema.NullOr(Schema.String),
20 threadId: Schema.NullOr(Schema.String),
21 deviceId: Schema.NullOr(Schema.String),
22 kind: Schema.NullOr(Schema.String),
23 cause: Schema.Defect(),
24 },
25) {
26 override get message(): string {
27 return `Failed to persist APNs delivery attempt during ${this.operation}.`;
28 }
29}
30
31export interface DeliveryAttemptInput {
32 readonly userId: string | null;
33 readonly environmentId: string | null;
34 readonly threadId: string | null;
35 readonly deviceId: string | null;
36 readonly kind: string;
37 readonly sourceJobId?: string | null;
38 readonly token: string | null;
39 readonly apnsStatus?: number;
40 readonly apnsReason?: string;
41 readonly apnsId?: string | null;
42 readonly transportError?: string;
43}
44
45export interface DeliveryAttemptCompletionInput {
46 readonly sourceJobId: string;
47 readonly apnsStatus?: number;
48 readonly apnsReason?: string;
49 readonly apnsId?: string | null;
50 readonly transportError?: string;
51}
52
53export type DeliverySourceJobClaimResult = "claimed" | "completed" | "in_flight";
54
55export class DeliveryAttempts extends Context.Service<
56 DeliveryAttempts,
57 {
58 readonly record: (
59 input: DeliveryAttemptInput,
60 ) => Effect.Effect<void, DeliveryAttemptRecordPersistenceError>;
61 readonly claimSourceJob: (
62 input: DeliveryAttemptInput & { readonly sourceJobId: string },
63 ) => Effect.Effect<DeliverySourceJobClaimResult, DeliveryAttemptRecordPersistenceError>;
64 readonly completeSourceJob: (
65 input: DeliveryAttemptCompletionInput,
66 ) => Effect.Effect<void, DeliveryAttemptRecordPersistenceError>;
67 }
68>()("t3code-relay/agentActivity/DeliveryAttempts") {}
69
70const SOURCE_JOB_CLAIM_LEASE_MINUTES = 10;
71
72function insertValues(
73 input: DeliveryAttemptInput,
74 id: string,
75 createdAt: string,
76): typeof relayDeliveryAttempts.$inferInsert {
77 return {
78 id,
79 createdAt,
80 userId: input.userId,
81 environmentId: input.environmentId,
82 threadId: input.threadId,
83 deviceId: input.deviceId,
84 kind: input.kind,
85 sourceJobId: input.sourceJobId ?? null,
86 tokenSuffix: input.token?.slice(-8) ?? null,
87 apnsStatus: input.apnsStatus ?? null,
88 apnsReason: input.apnsReason ?? null,
89 apnsId: input.apnsId ?? null,
90 transportError: input.transportError ?? null,
91 };
92}
93
94export const make = Effect.gen(function* () {
95 const db = yield* RelayDb.RelayDb;
96 const crypto = yield* Crypto.Crypto;
97
98 const isExpiredClaim = (claimedAt: string | null, now: DateTime.DateTime) => {
99 if (claimedAt === null) {
100 return true;
101 }
102 return Option.match(DateTime.make(claimedAt), {
103 onNone: () => true,
104 onSome: (dateTime) =>
105 now.epochMilliseconds - dateTime.epochMilliseconds >=
106 SOURCE_JOB_CLAIM_LEASE_MINUTES * 60 * 1_000,
107 });
108 };
109
110 return DeliveryAttempts.of({
111 record: Effect.fn("relay.delivery_attempts.record")(function* (input) {
112 yield* Effect.annotateCurrentSpan({
113 "relay.delivery.kind": input.kind,
114 ...(input.sourceJobId ? { "relay.delivery.job_id": input.sourceJobId } : {}),
115 ...(input.deviceId ? { "relay.mobile.device_id": input.deviceId } : {}),
116 ...(input.environmentId ? { "relay.environment_id": input.environmentId } : {}),
117 ...(input.threadId ? { "relay.thread_id": input.threadId } : {}),
118 });
119 yield* Effect.gen(function* () {
120 const id = yield* crypto.randomUUIDv4();
121 const createdAt = DateTime.formatIso(yield* DateTime.now);
122 yield* db.insert(relayDeliveryAttempts).values(insertValues(input, id, createdAt));
123 }).pipe(
124 Effect.mapError(
125 (cause) =>
126 new DeliveryAttemptRecordPersistenceError({
127 operation: "record",
128 sourceJobId: input.sourceJobId ?? null,
129 userId: input.userId,
130 environmentId: input.environmentId,
131 threadId: input.threadId,
132 deviceId: input.deviceId,
133 kind: input.kind,
134 cause,
135 }),
136 ),
137 );
138 }),
139 claimSourceJob: Effect.fn("relay.delivery_attempts.claim_source_job")(function* (input) {
140 yield* Effect.annotateCurrentSpan({
141 "relay.delivery.kind": input.kind,
142 "relay.delivery.job_id": input.sourceJobId,
143 ...(input.deviceId ? { "relay.mobile.device_id": input.deviceId } : {}),
144 ...(input.environmentId ? { "relay.environment_id": input.environmentId } : {}),
145 ...(input.threadId ? { "relay.thread_id": input.threadId } : {}),
146 });
147 return yield* Effect.gen(function* () {
148 const id = yield* crypto.randomUUIDv4();
149 const now = yield* DateTime.now;
150 const createdAt = DateTime.formatIso(now);
151 const inserted = yield* db
152 .insert(relayDeliveryAttempts)
153 .values(insertValues(input, id, createdAt))
154 .onConflictDoNothing({ target: relayDeliveryAttempts.sourceJobId })
155 .returning({ id: relayDeliveryAttempts.id });
156 if (inserted.length > 0) {
157 return "claimed";
158 }
159
160 const existing = yield* db
161 .select({
162 createdAt: relayDeliveryAttempts.createdAt,
163 apnsStatus: relayDeliveryAttempts.apnsStatus,
164 apnsReason: relayDeliveryAttempts.apnsReason,
165 apnsId: relayDeliveryAttempts.apnsId,
166 transportError: relayDeliveryAttempts.transportError,
167 })
168 .from(relayDeliveryAttempts)
169 .where(eq(relayDeliveryAttempts.sourceJobId, input.sourceJobId))
170 .limit(1);
171 const row = existing[0];
172 if (!row) {
173 return "in_flight";
174 }
175 if (
176 row.apnsStatus !== null ||
177 row.apnsReason !== null ||
178 row.apnsId !== null ||
179 row.transportError !== null
180 ) {
181 return "completed";
182 }
183 if (!isExpiredClaim(row.createdAt, now)) {
184 return "in_flight";
185 }
186
187 const reclaimed = yield* db
188 .update(relayDeliveryAttempts)
189 .set({
190 createdAt,
191 })
192 .where(
193 and(
194 eq(relayDeliveryAttempts.sourceJobId, input.sourceJobId),
195 eq(relayDeliveryAttempts.createdAt, row.createdAt),
196 isNull(relayDeliveryAttempts.apnsStatus),
197 isNull(relayDeliveryAttempts.apnsReason),
198 isNull(relayDeliveryAttempts.apnsId),
199 isNull(relayDeliveryAttempts.transportError),
200 ),
201 )
202 .returning({ id: relayDeliveryAttempts.id });
203 return reclaimed.length > 0 ? "claimed" : "in_flight";
204 }).pipe(
205 Effect.mapError(
206 (cause) =>
207 new DeliveryAttemptRecordPersistenceError({
208 operation: "claim-source-job",
209 sourceJobId: input.sourceJobId,
210 userId: input.userId,
211 environmentId: input.environmentId,
212 threadId: input.threadId,
213 deviceId: input.deviceId,
214 kind: input.kind,
215 cause,
216 }),
217 ),
218 );
219 }),
220 completeSourceJob: Effect.fn("relay.delivery_attempts.complete_source_job")(function* (input) {
221 yield* Effect.annotateCurrentSpan({ "relay.delivery.job_id": input.sourceJobId });
222 const completedAt = DateTime.formatIso(yield* DateTime.now);
223 yield* db
224 .update(relayDeliveryAttempts)
225 .set({
226 createdAt: completedAt,
227 apnsStatus: input.apnsStatus ?? null,
228 apnsReason: input.apnsReason ?? null,
229 apnsId: input.apnsId ?? null,
230 transportError: input.transportError ?? null,
231 })
232 .where(eq(relayDeliveryAttempts.sourceJobId, input.sourceJobId))
233 .pipe(
234 Effect.mapError(
235 (cause) =>
236 new DeliveryAttemptRecordPersistenceError({
237 operation: "complete-source-job",
238 sourceJobId: input.sourceJobId,
239 userId: null,
240 environmentId: null,
241 threadId: null,
242 deviceId: null,
243 kind: null,
244 cause,
245 }),
246 ),
247 );
248 }),
249 });
250});
251
252export const layer = Layer.effect(DeliveryAttempts, make);