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