| ... | @@ -12,10 +12,19 @@ import { relayDeliveryAttempts } from "../persistence/schema.ts"; | ... | @@ -12,10 +12,19 @@ import { relayDeliveryAttempts } from "../persistence/schema.ts"; |
| 12 | | 12 | |
| 13 | export class DeliveryAttemptRecordPersistenceError extends Schema.TaggedErrorClass<DeliveryAttemptRecordPersistenceError>()( | 13 | export class DeliveryAttemptRecordPersistenceError extends Schema.TaggedErrorClass<DeliveryAttemptRecordPersistenceError>()( |
| 14 | "DeliveryAttemptRecordPersistenceError", | 14 | "DeliveryAttemptRecordPersistenceError", |
| 15 | { cause: Schema.Defect() }, | 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 | }, |
| 16 | ) { | 25 | ) { |
| 17 | override get message(): string { | 26 | override get message(): string { |
| 18 | return "Failed to persist APNs delivery attempt"; | 27 | return `Failed to persist APNs delivery attempt during ${this.operation}.`; |
| 19 | } | 28 | } |
| 20 | } | 29 | } |
| 21 | | 30 | |
| ... | @@ -99,30 +108,43 @@ export const make = Effect.gen(function* () { | ... | @@ -99,30 +108,43 @@ export const make = Effect.gen(function* () { |
| 99 | }; | 108 | }; |
| 100 | | 109 | |
| 101 | return DeliveryAttempts.of({ | 110 | return DeliveryAttempts.of({ |
| 102 | record: Effect.fn("relay.delivery_attempts.record")( | 111 | record: Effect.fn("relay.delivery_attempts.record")(function* (input) { |
| 103 | function* (input) { | 112 | yield* Effect.annotateCurrentSpan({ |
| 104 | yield* Effect.annotateCurrentSpan({ | 113 | "relay.delivery.kind": input.kind, |
| 105 | "relay.delivery.kind": input.kind, | 114 | ...(input.sourceJobId ? { "relay.delivery.job_id": input.sourceJobId } : {}), |
| 106 | ...(input.sourceJobId ? { "relay.delivery.job_id": input.sourceJobId } : {}), | 115 | ...(input.deviceId ? { "relay.mobile.device_id": input.deviceId } : {}), |
| 107 | ...(input.deviceId ? { "relay.mobile.device_id": input.deviceId } : {}), | 116 | ...(input.environmentId ? { "relay.environment_id": input.environmentId } : {}), |
| 108 | ...(input.environmentId ? { "relay.environment_id": input.environmentId } : {}), | 117 | ...(input.threadId ? { "relay.thread_id": input.threadId } : {}), |
| 109 | ...(input.threadId ? { "relay.thread_id": input.threadId } : {}), | 118 | }); |
| 110 | }); | 119 | yield* Effect.gen(function* () { |
| 111 | const id = yield* crypto.randomUUIDv4; | 120 | const id = yield* crypto.randomUUIDv4; |
| 112 | const createdAt = DateTime.formatIso(yield* DateTime.now); | 121 | const createdAt = DateTime.formatIso(yield* DateTime.now); |
| 113 | yield* db.insert(relayDeliveryAttempts).values(insertValues(input, id, createdAt)); | 122 | yield* db.insert(relayDeliveryAttempts).values(insertValues(input, id, createdAt)); |
| 114 | }, | 123 | }).pipe( |
| 115 | Effect.mapError((cause) => new DeliveryAttemptRecordPersistenceError({ cause })), | 124 | Effect.mapError( |
| 116 | ), | 125 | (cause) => |
| 117 | claimSourceJob: Effect.fn("relay.delivery_attempts.claim_source_job")( | 126 | new DeliveryAttemptRecordPersistenceError({ |
| 118 | function* (input) { | 127 | operation: "record", |
| 119 | yield* Effect.annotateCurrentSpan({ | 128 | sourceJobId: input.sourceJobId ?? null, |
| 120 | "relay.delivery.kind": input.kind, | 129 | userId: input.userId, |
| 121 | "relay.delivery.job_id": input.sourceJobId, | 130 | environmentId: input.environmentId, |
| 122 | ...(input.deviceId ? { "relay.mobile.device_id": input.deviceId } : {}), | 131 | threadId: input.threadId, |
| 123 | ...(input.environmentId ? { "relay.environment_id": input.environmentId } : {}), | 132 | deviceId: input.deviceId, |
| 124 | ...(input.threadId ? { "relay.thread_id": input.threadId } : {}), | 133 | kind: input.kind, |
| 125 | }); | 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* () { |
| 126 | const id = yield* crypto.randomUUIDv4; | 148 | const id = yield* crypto.randomUUIDv4; |
| 127 | const now = yield* DateTime.now; | 149 | const now = yield* DateTime.now; |
| 128 | const createdAt = DateTime.formatIso(now); | 150 | const createdAt = DateTime.formatIso(now); |
| ... | @@ -179,26 +201,51 @@ export const make = Effect.gen(function* () { | ... | @@ -179,26 +201,51 @@ export const make = Effect.gen(function* () { |
| 179 | ) | 201 | ) |
| 180 | .returning({ id: relayDeliveryAttempts.id }); | 202 | .returning({ id: relayDeliveryAttempts.id }); |
| 181 | return reclaimed.length > 0 ? "claimed" : "in_flight"; | 203 | return reclaimed.length > 0 ? "claimed" : "in_flight"; |
| 182 | }, | 204 | }).pipe( |
| 183 | Effect.mapError((cause) => new DeliveryAttemptRecordPersistenceError({ cause })), | 205 | Effect.mapError( |
| 184 | ), | 206 | (cause) => |
| 185 | completeSourceJob: Effect.fn("relay.delivery_attempts.complete_source_job")( | 207 | new DeliveryAttemptRecordPersistenceError({ |
| 186 | function* (input) { | 208 | operation: "claim-source-job", |
| 187 | yield* Effect.annotateCurrentSpan({ "relay.delivery.job_id": input.sourceJobId }); | 209 | sourceJobId: input.sourceJobId, |
| 188 | const completedAt = DateTime.formatIso(yield* DateTime.now); | 210 | userId: input.userId, |
| 189 | yield* db | 211 | environmentId: input.environmentId, |
| 190 | .update(relayDeliveryAttempts) | 212 | threadId: input.threadId, |
| 191 | .set({ | 213 | deviceId: input.deviceId, |
| 192 | createdAt: completedAt, | 214 | kind: input.kind, |
| 193 | apnsStatus: input.apnsStatus ?? null, | 215 | cause, |
| 194 | apnsReason: input.apnsReason ?? null, | 216 | }), |
| 195 | apnsId: input.apnsId ?? null, | 217 | ), |
| 196 | transportError: input.transportError ?? null, | 218 | ); |
| 197 | }) | 219 | }), |
| 198 | .where(eq(relayDeliveryAttempts.sourceJobId, input.sourceJobId)); | 220 | completeSourceJob: Effect.fn("relay.delivery_attempts.complete_source_job")(function* (input) { |
| 199 | }, | 221 | yield* Effect.annotateCurrentSpan({ "relay.delivery.job_id": input.sourceJobId }); |
| 200 | Effect.mapError((cause) => new DeliveryAttemptRecordPersistenceError({ cause })), | 222 | const completedAt = DateTime.formatIso(yield* DateTime.now); |
| 201 | ), | 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 | }), |
| 202 | }); | 249 | }); |
| 203 | }); | 250 | }); |
| 204 | | 251 | |