authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-06-19 21:13:01-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-06-19 21:13:01-07:00
logd5a72d5be105d16a95df6dc18ad08f0b0cdb7efb
treee7528f84cd0c7ca83bfa613caff790db2707760e
parent742dd7af0dcb54c32774f87fd354105726c7a9db
signature Signed by PGP key B5690EEEBB952194

[codex] align relay agent activity Effect services (#3179)

Co-authored-by: codex <codex@users.noreply.github.com>

13 files changed, 320 insertions(+), 285 deletions(-)

infra/relay/src/agentActivity/AgentActivityPublisher.test.ts+17-14
...@@ -41,8 +41,8 @@ function target(deviceId: string): LiveActivities.TargetRow {...@@ -41,8 +41,8 @@ function target(deviceId: string): LiveActivities.TargetRow {
41}41}
4242
43function makeLiveActivities(43function makeLiveActivities(
44 overrides: Partial<LiveActivities.LiveActivitiesShape> = {},44 overrides: Partial<LiveActivities.LiveActivities["Service"]> = {},
45): LiveActivities.LiveActivitiesShape {45): LiveActivities.LiveActivities["Service"] {
46 return {46 return {
47 register: () => Effect.void,47 register: () => Effect.void,
48 listTargets: () => Effect.succeed([]),48 listTargets: () => Effect.succeed([]),
...@@ -55,8 +55,8 @@ function makeLiveActivities(...@@ -55,8 +55,8 @@ function makeLiveActivities(
55}55}
5656
57function makeAgentActivityRows(57function makeAgentActivityRows(
58 overrides: Partial<AgentActivityRows.AgentActivityRowsShape> = {},58 overrides: Partial<AgentActivityRows.AgentActivityRows["Service"]> = {},
59): AgentActivityRows.AgentActivityRowsShape {59): AgentActivityRows.AgentActivityRows["Service"] {
60 return {60 return {
61 upsert: () => Effect.void,61 upsert: () => Effect.void,
62 remove: () => Effect.void,62 remove: () => Effect.void,
...@@ -88,8 +88,8 @@ function makeEnvironmentLinks(...@@ -88,8 +88,8 @@ function makeEnvironmentLinks(
88}88}
8989
90function makeApnsDeliveries(90function makeApnsDeliveries(
91 overrides: Partial<ApnsDeliveries.ApnsDeliveriesShape> = {},91 overrides: Partial<ApnsDeliveries.ApnsDeliveries["Service"]> = {},
92): ApnsDeliveries.ApnsDeliveriesShape {92): ApnsDeliveries.ApnsDeliveries["Service"] {
93 return {93 return {
94 sendForTarget: () => Effect.succeed(null),94 sendForTarget: () => Effect.succeed(null),
95 sendPushNotificationForTarget: () => Effect.succeed(null),95 sendPushNotificationForTarget: () => Effect.succeed(null),
...@@ -133,7 +133,8 @@ describe("AgentActivityPublisher", () => {...@@ -133,7 +133,8 @@ describe("AgentActivityPublisher", () => {
133 remote_start_queued_at: null,133 remote_start_queued_at: null,
134 remote_started_at: "1970-01-01T00:00:01.000Z",134 remote_started_at: "1970-01-01T00:00:01.000Z",
135 };135 };
136 const sent: Array<Parameters<ApnsDeliveries.ApnsDeliveriesShape["sendForTarget"]>[0]> = [];136 const sent: Array<Parameters<ApnsDeliveries.ApnsDeliveries["Service"]["sendForTarget"]>[0]> =
137 [];
137 const deliveryResult: RelayDeliveryResult = {138 const deliveryResult: RelayDeliveryResult = {
138 deviceId: "device-1",139 deviceId: "device-1",
139 kind: "live_activity_update",140 kind: "live_activity_update",
...@@ -211,7 +212,8 @@ describe("AgentActivityPublisher", () => {...@@ -211,7 +212,8 @@ describe("AgentActivityPublisher", () => {
211 readonly environmentId: string;212 readonly environmentId: string;
212 readonly environmentPublicKey: string;213 readonly environmentPublicKey: string;
213 }> = [];214 }> = [];
214 const upserts: Array<Parameters<AgentActivityRows.AgentActivityRowsShape["upsert"]>[0]> = [];215 const upserts: Array<Parameters<AgentActivityRows.AgentActivityRows["Service"]["upsert"]>[0]> =
216 [];
215217
216 return Effect.gen(function* () {218 return Effect.gen(function* () {
217 const result = yield* Effect.gen(function* () {219 const result = yield* Effect.gen(function* () {
...@@ -302,9 +304,10 @@ describe("AgentActivityPublisher", () => {...@@ -302,9 +304,10 @@ describe("AgentActivityPublisher", () => {
302 updatedAt: "1970-01-01T00:00:10.000Z",304 updatedAt: "1970-01-01T00:00:10.000Z",
303 };305 };
304 const sentAggregates: Array<306 const sentAggregates: Array<
305 Parameters<ApnsDeliveries.ApnsDeliveriesShape["sendForTarget"]>[0]307 Parameters<ApnsDeliveries.ApnsDeliveries["Service"]["sendForTarget"]>[0]
306 > = [];308 > = [];
307 const removals: Array<Parameters<AgentActivityRows.AgentActivityRowsShape["remove"]>[0]> = [];309 const removals: Array<Parameters<AgentActivityRows.AgentActivityRows["Service"]["remove"]>[0]> =
310 [];
308311
309 return Effect.gen(function* () {312 return Effect.gen(function* () {
310 const result = yield* Effect.gen(function* () {313 const result = yield* Effect.gen(function* () {
...@@ -405,10 +408,10 @@ describe("AgentActivityPublisher", () => {...@@ -405,10 +408,10 @@ describe("AgentActivityPublisher", () => {
405 headline: "Needs input",408 headline: "Needs input",
406 };409 };
407 const liveAggregates: Array<410 const liveAggregates: Array<
408 Parameters<ApnsDeliveries.ApnsDeliveriesShape["sendForTarget"]>[0]411 Parameters<ApnsDeliveries.ApnsDeliveries["Service"]["sendForTarget"]>[0]
409 > = [];412 > = [];
410 const pushAggregates: Array<413 const pushAggregates: Array<
411 Parameters<ApnsDeliveries.ApnsDeliveriesShape["sendPushNotificationForTarget"]>[0]414 Parameters<ApnsDeliveries.ApnsDeliveries["Service"]["sendPushNotificationForTarget"]>[0]
412 > = [];415 > = [];
413416
414 return Effect.gen(function* () {417 return Effect.gen(function* () {
...@@ -517,10 +520,10 @@ describe("AgentActivityPublisher", () => {...@@ -517,10 +520,10 @@ describe("AgentActivityPublisher", () => {
517 headline: "Needs approval",520 headline: "Needs approval",
518 };521 };
519 const liveAggregates: Array<522 const liveAggregates: Array<
520 Parameters<ApnsDeliveries.ApnsDeliveriesShape["sendForTarget"]>[0]523 Parameters<ApnsDeliveries.ApnsDeliveries["Service"]["sendForTarget"]>[0]
521 > = [];524 > = [];
522 const pushAggregates: Array<525 const pushAggregates: Array<
523 Parameters<ApnsDeliveries.ApnsDeliveriesShape["sendPushNotificationForTarget"]>[0]526 Parameters<ApnsDeliveries.ApnsDeliveries["Service"]["sendPushNotificationForTarget"]>[0]
524 > = [];527 > = [];
525528
526 return Effect.gen(function* () {529 return Effect.gen(function* () {
infra/relay/src/agentActivity/AgentActivityPublisher.ts+12-14
...@@ -23,22 +23,20 @@ export type AgentActivityPublishError =...@@ -23,22 +23,20 @@ export type AgentActivityPublishError =
23 | LiveActivities.LiveActivityTargetListPersistenceError23 | LiveActivities.LiveActivityTargetListPersistenceError
24 | ApnsDeliveries.ApnsDeliveryError;24 | ApnsDeliveries.ApnsDeliveryError;
2525
26export interface AgentActivityPublisherShape {
27 readonly publish: (input: {
28 readonly environmentId: string;
29 readonly environmentPublicKey: string;
30 readonly threadId: string;
31 readonly state: RelayAgentActivityState | null;
32 }) => Effect.Effect<RelayPublishResponse, AgentActivityPublishError>;
33 readonly replayForLiveActivityRegistration: (input: {
34 readonly userId: string;
35 readonly deviceId: string;
36 }) => Effect.Effect<RelayDeliveryResult | null, AgentActivityPublishError>;
37}
38
39export class AgentActivityPublisher extends Context.Service<26export class AgentActivityPublisher extends Context.Service<
40 AgentActivityPublisher,27 AgentActivityPublisher,
41 AgentActivityPublisherShape28 {
29 readonly publish: (input: {
30 readonly environmentId: string;
31 readonly environmentPublicKey: string;
32 readonly threadId: string;
33 readonly state: RelayAgentActivityState | null;
34 }) => Effect.Effect<RelayPublishResponse, AgentActivityPublishError>;
35 readonly replayForLiveActivityRegistration: (input: {
36 readonly userId: string;
37 readonly deviceId: string;
38 }) => Effect.Effect<RelayDeliveryResult | null, AgentActivityPublishError>;
39 }
42>()("t3code-relay/agentActivity/AgentActivityPublisher") {}40>()("t3code-relay/agentActivity/AgentActivityPublisher") {}
4341
44const make = Effect.gen(function* () {42const make = Effect.gen(function* () {
infra/relay/src/agentActivity/AgentActivityRows.ts+22-20
...@@ -3,7 +3,7 @@ import { RelayAgentActivityState as RelayAgentActivityStateSchema } from "@t3too...@@ -3,7 +3,7 @@ import { RelayAgentActivityState as RelayAgentActivityStateSchema } from "@t3too
3import * as Context from "effect/Context";3import * as Context from "effect/Context";
4import * as DateTime from "effect/DateTime";4import * as DateTime from "effect/DateTime";
5import * as Effect from "effect/Effect";5import * as Effect from "effect/Effect";
6import { cast } from "effect/Function";6import * as Function from "effect/Function";
7import * as Layer from "effect/Layer";7import * as Layer from "effect/Layer";
8import * as Option from "effect/Option";8import * as Option from "effect/Option";
9import * as Schema from "effect/Schema";9import * as Schema from "effect/Schema";
...@@ -39,24 +39,26 @@ export class AgentActivityRowListPersistenceError extends Schema.TaggedErrorClas...@@ -39,24 +39,26 @@ export class AgentActivityRowListPersistenceError extends Schema.TaggedErrorClas
39 }39 }
40}40}
4141
42export interface AgentActivityRowsShape {42export class AgentActivityRows extends Context.Service<
43 readonly upsert: (input: {43 AgentActivityRows,
44 readonly environmentPublicKey: string;44 {
45 readonly state: RelayAgentActivityState;45 readonly upsert: (input: {
46 }) => Effect.Effect<void, AgentActivityRowUpsertPersistenceError>;46 readonly environmentPublicKey: string;
47 readonly remove: (input: {47 readonly state: RelayAgentActivityState;
48 readonly environmentId: string;48 }) => Effect.Effect<void, AgentActivityRowUpsertPersistenceError>;
49 readonly environmentPublicKey: string;49 readonly remove: (input: {
50 readonly threadId: string;50 readonly environmentId: string;
51 }) => Effect.Effect<void, AgentActivityRowDeletePersistenceError>;51 readonly environmentPublicKey: string;
52 readonly listForUser: (input: {52 readonly threadId: string;
53 readonly userId: string;53 }) => Effect.Effect<void, AgentActivityRowDeletePersistenceError>;
54 }) => Effect.Effect<ReadonlyArray<RelayAgentActivityState>, AgentActivityRowListPersistenceError>;54 readonly listForUser: (input: {
55}55 readonly userId: string;
5656 }) => Effect.Effect<
57export class AgentActivityRows extends Context.Service<AgentActivityRows, AgentActivityRowsShape>()(57 ReadonlyArray<RelayAgentActivityState>,
58 "t3code-relay/agentActivity/AgentActivityRows",58 AgentActivityRowListPersistenceError
59) {}59 >;
60 }
61>()("t3code-relay/agentActivity/AgentActivityRows") {}
6062
61const decodeJsonString = Schema.decodeEffect(Schema.UnknownFromJsonString);63const decodeJsonString = Schema.decodeEffect(Schema.UnknownFromJsonString);
62const encodeJsonValue = Schema.encodeEffect(Schema.UnknownFromJsonString);64const encodeJsonValue = Schema.encodeEffect(Schema.UnknownFromJsonString);
...@@ -82,7 +84,7 @@ const make = Effect.gen(function* () {...@@ -82,7 +84,7 @@ const make = Effect.gen(function* () {
82 const now = yield* DateTime.now;84 const now = yield* DateTime.now;
83 const stateJson = yield* encodeRelayAgentActivityStateJson(input.state).pipe(85 const stateJson = yield* encodeRelayAgentActivityStateJson(input.state).pipe(
84 Effect.flatMap(decodeJsonString),86 Effect.flatMap(decodeJsonString),
85 Effect.map(cast<unknown, RelayAgentActivityState>),87 Effect.map(Function.cast<unknown, RelayAgentActivityState>),
86 );88 );
87 yield* db89 yield* db
88 .insert(relayAgentActivityRows)90 .insert(relayAgentActivityRows)
infra/relay/src/agentActivity/ApnsClient.ts+56-54
...@@ -8,8 +8,10 @@ import * as Layer from "effect/Layer";...@@ -8,8 +8,10 @@ import * as Layer from "effect/Layer";
8import * as Option from "effect/Option";8import * as Option from "effect/Option";
9import * as Redacted from "effect/Redacted";9import * as Redacted from "effect/Redacted";
10import * as Schema from "effect/Schema";10import * as Schema from "effect/Schema";
11import { Headers, HttpClient, HttpClientRequest } from "effect/unstable/http";11import * as Headers from "effect/unstable/http/Headers";
12import * as RelayConfiguration from "../Config.ts";12import * as HttpClient from "effect/unstable/http/HttpClient";
13import * as HttpClientRequest from "effect/unstable/http/HttpClientRequest";
14import type * as RelayConfiguration from "../Config.ts";
13import type { ApnsNotificationPayload } from "./apnsDeliveryJobs.ts";15import type { ApnsNotificationPayload } from "./apnsDeliveryJobs.ts";
1416
15const LIVE_ACTIVITY_NAME = "AgentActivity";17const LIVE_ACTIVITY_NAME = "AgentActivity";
...@@ -231,29 +233,28 @@ function apnsReasonFromBody(body: string): string | undefined {...@@ -231,29 +233,28 @@ function apnsReasonFromBody(body: string): string | undefined {
231 });233 });
232}234}
233235
234export interface ApnsClientShape {236export class ApnsClient extends Context.Service<
235 readonly makeLiveActivityRequest: typeof makeLiveActivityRequest;237 ApnsClient,
236 readonly makePushNotificationRequest: typeof makePushNotificationRequest;238 {
237 readonly sendLiveActivityRequest: (input: {239 readonly makeLiveActivityRequest: typeof makeLiveActivityRequest;
238 readonly credentials: RelayConfiguration.ApnsCredentials;240 readonly makePushNotificationRequest: typeof makePushNotificationRequest;
239 readonly request: ApnsLiveActivityRequest;241 readonly sendLiveActivityRequest: (input: {
240 readonly issuedAtUnixSeconds: number;242 readonly credentials: RelayConfiguration.ApnsCredentials;
241 }) => Effect.Effect<ApnsDeliveryResult, ApnsError>;243 readonly request: ApnsLiveActivityRequest;
242 readonly sendPushNotificationRequest: (input: {244 readonly issuedAtUnixSeconds: number;
243 readonly credentials: RelayConfiguration.ApnsCredentials;245 }) => Effect.Effect<ApnsDeliveryResult, ApnsError>;
244 readonly request: ApnsPushNotificationRequest;246 readonly sendPushNotificationRequest: (input: {
245 readonly issuedAtUnixSeconds: number;247 readonly credentials: RelayConfiguration.ApnsCredentials;
246 }) => Effect.Effect<ApnsDeliveryResult, ApnsError>;248 readonly request: ApnsPushNotificationRequest;
247}249 readonly issuedAtUnixSeconds: number;
248250 }) => Effect.Effect<ApnsDeliveryResult, ApnsError>;
249export class ApnsClient extends Context.Service<ApnsClient, ApnsClientShape>()(251 }
250 "t3code-relay/agentActivity/ApnsClient",252>()("t3code-relay/agentActivity/ApnsClient") {}
251) {}
252253
253const make = Effect.gen(function* () {254const make = Effect.gen(function* () {
254 const httpClient = yield* HttpClient.HttpClient;255 const httpClient = yield* HttpClient.HttpClient;
255256
256 const sendLiveActivityRequest: ApnsClientShape["sendLiveActivityRequest"] = Effect.fn(257 const sendLiveActivityRequest: ApnsClient["Service"]["sendLiveActivityRequest"] = Effect.fn(
257 "relay.apns.send_live_activity_request",258 "relay.apns.send_live_activity_request",
258 )(function* (input) {259 )(function* (input) {
259 yield* Effect.annotateCurrentSpan({ "relay.apns.event": input.request.event });260 yield* Effect.annotateCurrentSpan({ "relay.apns.event": input.request.event });
...@@ -288,40 +289,41 @@ const make = Effect.gen(function* () {...@@ -288,40 +289,41 @@ const make = Effect.gen(function* () {
288 };289 };
289 });290 });
290291
291 const sendPushNotificationRequest: ApnsClientShape["sendPushNotificationRequest"] = Effect.fn(292 const sendPushNotificationRequest: ApnsClient["Service"]["sendPushNotificationRequest"] =
292 "relay.apns.send_push_notification_request",293 Effect.fn("relay.apns.send_push_notification_request")(function* (input) {
293 )(function* (input) {294 yield* Effect.annotateCurrentSpan({ "relay.apns.event": "push_notification" });
294 yield* Effect.annotateCurrentSpan({ "relay.apns.event": "push_notification" });295 const jwt = yield* makeApnsJwt({
295 const jwt = yield* makeApnsJwt({296 ...input.credentials,
296 ...input.credentials,297 issuedAtUnixSeconds: input.issuedAtUnixSeconds,
297 issuedAtUnixSeconds: input.issuedAtUnixSeconds,298 });
299 const host =
300 input.credentials.environment === "production"
301 ? "https://api.push.apple.com"
302 : "https://api.sandbox.push.apple.com";
303 const response = yield* HttpClientRequest.post(
304 `${host}/3/device/${input.request.token}`,
305 ).pipe(
306 HttpClientRequest.setHeaders({
307 authorization: `bearer ${jwt}`,
308 "apns-priority": input.request.priority,
309 "apns-push-type": "alert",
310 "apns-topic": input.credentials.bundleId,
311 }),
312 HttpClientRequest.bodyJson(input.request.payload),
313 Effect.flatMap(httpClient.execute),
314 Effect.mapError((cause) => new ApnsHttpRequestError({ cause })),
315 );
316 const responseText = yield* response.text.pipe(
317 Effect.mapError((cause) => new ApnsHttpRequestError({ cause })),
318 );
319 const reason = apnsReasonFromBody(responseText);
320 return {
321 ok: response.status >= 200 && response.status < 300,
322 status: response.status,
323 ...(reason === undefined ? {} : { reason }),
324 apnsId: Option.getOrNull(Headers.get(response.headers, "apns-id")),
325 };
298 });326 });
299 const host =
300 input.credentials.environment === "production"
301 ? "https://api.push.apple.com"
302 : "https://api.sandbox.push.apple.com";
303 const response = yield* HttpClientRequest.post(`${host}/3/device/${input.request.token}`).pipe(
304 HttpClientRequest.setHeaders({
305 authorization: `bearer ${jwt}`,
306 "apns-priority": input.request.priority,
307 "apns-push-type": "alert",
308 "apns-topic": input.credentials.bundleId,
309 }),
310 HttpClientRequest.bodyJson(input.request.payload),
311 Effect.flatMap(httpClient.execute),
312 Effect.mapError((cause) => new ApnsHttpRequestError({ cause })),
313 );
314 const responseText = yield* response.text.pipe(
315 Effect.mapError((cause) => new ApnsHttpRequestError({ cause })),
316 );
317 const reason = apnsReasonFromBody(responseText);
318 return {
319 ok: response.status >= 200 && response.status < 300,
320 status: response.status,
321 ...(reason === undefined ? {} : { reason }),
322 apnsId: Option.getOrNull(Headers.get(response.headers, "apns-id")),
323 };
324 });
325327
326 return ApnsClient.of({328 return ApnsClient.of({
327 makeLiveActivityRequest,329 makeLiveActivityRequest,
infra/relay/src/agentActivity/ApnsDeliveries.test.ts+10-10
...@@ -141,16 +141,16 @@ function makeLayer(input: {...@@ -141,16 +141,16 @@ function makeLayer(input: {
141 readonly sourceJobClaims?: ReadonlyMap<string, DeliveryAttempts.DeliverySourceJobClaimResult>;141 readonly sourceJobClaims?: ReadonlyMap<string, DeliveryAttempts.DeliverySourceJobClaimResult>;
142 readonly queuedJobs?: Array<SignedApnsDeliveryJob>;142 readonly queuedJobs?: Array<SignedApnsDeliveryJob>;
143 readonly queuedStarts?: Array<143 readonly queuedStarts?: Array<
144 Parameters<LiveActivities.LiveActivitiesShape["markStartQueued"]>[0]144 Parameters<LiveActivities.LiveActivities["Service"]["markStartQueued"]>[0]
145 >;145 >;
146 readonly clearedStarts?: Array<146 readonly clearedStarts?: Array<
147 Parameters<LiveActivities.LiveActivitiesShape["clearStartQueued"]>[0]147 Parameters<LiveActivities.LiveActivities["Service"]["clearStartQueued"]>[0]
148 >;148 >;
149 readonly markedDeliveries?: Array<149 readonly markedDeliveries?: Array<
150 Parameters<LiveActivities.LiveActivitiesShape["markDelivery"]>[0]150 Parameters<LiveActivities.LiveActivities["Service"]["markDelivery"]>[0]
151 >;151 >;
152 readonly invalidatedTokens?: Array<152 readonly invalidatedTokens?: Array<
153 Parameters<LiveActivities.LiveActivitiesShape["invalidateDeliveryToken"]>[0]153 Parameters<LiveActivities.LiveActivities["Service"]["invalidateDeliveryToken"]>[0]
154 >;154 >;
155 readonly currentTargets?: ReadonlyArray<LiveActivities.TargetRow>;155 readonly currentTargets?: ReadonlyArray<LiveActivities.TargetRow>;
156 readonly config?: RelayConfiguration.RelayConfiguration["Service"];156 readonly config?: RelayConfiguration.RelayConfiguration["Service"];
...@@ -227,10 +227,10 @@ describe("ApnsDeliveries", () => {...@@ -227,10 +227,10 @@ describe("ApnsDeliveries", () => {
227 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];227 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
228 const queuedJobs: Array<SignedApnsDeliveryJob> = [];228 const queuedJobs: Array<SignedApnsDeliveryJob> = [];
229 const queuedStarts: Array<229 const queuedStarts: Array<
230 Parameters<LiveActivities.LiveActivitiesShape["markStartQueued"]>[0]230 Parameters<LiveActivities.LiveActivities["Service"]["markStartQueued"]>[0]
231 > = [];231 > = [];
232 const markedDeliveries: Array<232 const markedDeliveries: Array<
233 Parameters<LiveActivities.LiveActivitiesShape["markDelivery"]>[0]233 Parameters<LiveActivities.LiveActivities["Service"]["markDelivery"]>[0]
234 > = [];234 > = [];
235235
236 return Effect.gen(function* () {236 return Effect.gen(function* () {
...@@ -933,7 +933,7 @@ describe("ApnsDeliveries", () => {...@@ -933,7 +933,7 @@ describe("ApnsDeliveries", () => {
933 it.effect("invalidates dead device push tokens after permanent APNs alert failures", () => {933 it.effect("invalidates dead device push tokens after permanent APNs alert failures", () => {
934 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];934 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
935 const invalidatedTokens: Array<935 const invalidatedTokens: Array<
936 Parameters<LiveActivities.LiveActivitiesShape["invalidateDeliveryToken"]>[0]936 Parameters<LiveActivities.LiveActivities["Service"]["invalidateDeliveryToken"]>[0]
937 > = [];937 > = [];
938 const payload = makeApnsDeliveryJobPayload({938 const payload = makeApnsDeliveryJobPayload({
939 kind: "push_notification",939 kind: "push_notification",
...@@ -1000,7 +1000,7 @@ describe("ApnsDeliveries", () => {...@@ -1000,7 +1000,7 @@ describe("ApnsDeliveries", () => {
1000 it.effect("clears queued start state when a start job fails in APNs", () => {1000 it.effect("clears queued start state when a start job fails in APNs", () => {
1001 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];1001 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
1002 const clearedStarts: Array<1002 const clearedStarts: Array<
1003 Parameters<LiveActivities.LiveActivitiesShape["clearStartQueued"]>[0]1003 Parameters<LiveActivities.LiveActivities["Service"]["clearStartQueued"]>[0]
1004 > = [];1004 > = [];
1005 const payload = makeApnsDeliveryJobPayload({1005 const payload = makeApnsDeliveryJobPayload({
1006 kind: "live_activity_start",1006 kind: "live_activity_start",
...@@ -1035,7 +1035,7 @@ describe("ApnsDeliveries", () => {...@@ -1035,7 +1035,7 @@ describe("ApnsDeliveries", () => {
1035 it.effect("invalidates dead push-to-start tokens after permanent APNs start failures", () => {1035 it.effect("invalidates dead push-to-start tokens after permanent APNs start failures", () => {
1036 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];1036 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
1037 const invalidatedTokens: Array<1037 const invalidatedTokens: Array<
1038 Parameters<LiveActivities.LiveActivitiesShape["invalidateDeliveryToken"]>[0]1038 Parameters<LiveActivities.LiveActivities["Service"]["invalidateDeliveryToken"]>[0]
1039 > = [];1039 > = [];
1040 const payload = makeApnsDeliveryJobPayload({1040 const payload = makeApnsDeliveryJobPayload({
1041 kind: "live_activity_start",1041 kind: "live_activity_start",
...@@ -1082,7 +1082,7 @@ describe("ApnsDeliveries", () => {...@@ -1082,7 +1082,7 @@ describe("ApnsDeliveries", () => {
1082 it.effect("invalidates dead Live Activity tokens after APNs unregisters them", () => {1082 it.effect("invalidates dead Live Activity tokens after APNs unregisters them", () => {
1083 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];1083 const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
1084 const invalidatedTokens: Array<1084 const invalidatedTokens: Array<
1085 Parameters<LiveActivities.LiveActivitiesShape["invalidateDeliveryToken"]>[0]1085 Parameters<LiveActivities.LiveActivities["Service"]["invalidateDeliveryToken"]>[0]
1086 > = [];1086 > = [];
1087 const payload = makeApnsDeliveryJobPayload({1087 const payload = makeApnsDeliveryJobPayload({
1088 kind: "live_activity_update",1088 kind: "live_activity_update",
infra/relay/src/agentActivity/ApnsDeliveries.ts+33-34
...@@ -356,7 +356,7 @@ export type SendLiveActivityDeliveryInput =...@@ -356,7 +356,7 @@ export type SendLiveActivityDeliveryInput =
356 });356 });
357357
358function makeLiveActivityDeliveryRequest(358function makeLiveActivityDeliveryRequest(
359 apns: Apns.ApnsClientShape,359 apns: Apns.ApnsClient["Service"],
360 input: SendLiveActivityDeliveryInput,360 input: SendLiveActivityDeliveryInput,
361 now: DateTime.DateTime,361 now: DateTime.DateTime,
362) {362) {
...@@ -391,33 +391,32 @@ function makeLiveActivityDeliveryRequest(...@@ -391,33 +391,32 @@ function makeLiveActivityDeliveryRequest(
391 }391 }
392}392}
393393
394export interface ApnsDeliveriesShape {394export class ApnsDeliveries extends Context.Service<
395 readonly sendForTarget: (input: {395 ApnsDeliveries,
396 readonly target: LiveActivities.TargetRow;396 {
397 readonly aggregate: RelayAgentActivityAggregateState | null;397 readonly sendForTarget: (input: {
398 readonly nowMs: number;398 readonly target: LiveActivities.TargetRow;
399 }) => Effect.Effect<RelayDeliveryResult | null, ApnsDeliveryError>;399 readonly aggregate: RelayAgentActivityAggregateState | null;
400 readonly sendPushNotificationForTarget: (input: {400 readonly nowMs: number;
401 readonly target: LiveActivities.TargetRow;401 }) => Effect.Effect<RelayDeliveryResult | null, ApnsDeliveryError>;
402 readonly aggregate: RelayAgentActivityAggregateState | null;402 readonly sendPushNotificationForTarget: (input: {
403 }) => Effect.Effect<RelayDeliveryResult | null, ApnsDeliveryError>;403 readonly target: LiveActivities.TargetRow;
404 readonly sendLiveActivity: (404 readonly aggregate: RelayAgentActivityAggregateState | null;
405 input: SendLiveActivityDeliveryInput,405 }) => Effect.Effect<RelayDeliveryResult | null, ApnsDeliveryError>;
406 ) => Effect.Effect<RelayDeliveryResult, ApnsDeliveryError>;406 readonly sendLiveActivity: (
407 readonly processSignedJob: (407 input: SendLiveActivityDeliveryInput,
408 body: unknown,408 ) => Effect.Effect<RelayDeliveryResult, ApnsDeliveryError>;
409 ) => Effect.Effect<RelayDeliveryResult, ApnsDeliveryError>;409 readonly processSignedJob: (
410 readonly sendPushNotification: (input: {410 body: unknown,
411 readonly target: LiveActivityDeliveryTarget;411 ) => Effect.Effect<RelayDeliveryResult, ApnsDeliveryError>;
412 readonly token: string;412 readonly sendPushNotification: (input: {
413 readonly sourceJobId?: string | null;413 readonly target: LiveActivityDeliveryTarget;
414 readonly notification: ApnsNotificationPayload;414 readonly token: string;
415 }) => Effect.Effect<RelayDeliveryResult, ApnsDeliveryError>;415 readonly sourceJobId?: string | null;
416}416 readonly notification: ApnsNotificationPayload;
417417 }) => Effect.Effect<RelayDeliveryResult, ApnsDeliveryError>;
418export class ApnsDeliveries extends Context.Service<ApnsDeliveries, ApnsDeliveriesShape>()(418 }
419 "t3code-relay/agentActivity/ApnsDeliveries",419>()("t3code-relay/agentActivity/ApnsDeliveries") {}
420) {}
421420
422const make = Effect.gen(function* () {421const make = Effect.gen(function* () {
423 const attempts = yield* DeliveryAttempts.DeliveryAttempts;422 const attempts = yield* DeliveryAttempts.DeliveryAttempts;
...@@ -442,7 +441,7 @@ const make = Effect.gen(function* () {...@@ -442,7 +441,7 @@ const make = Effect.gen(function* () {
442 );441 );
443 });442 });
444443
445 const sendLiveActivity: ApnsDeliveriesShape["sendLiveActivity"] = Effect.fn(444 const sendLiveActivity: ApnsDeliveries["Service"]["sendLiveActivity"] = Effect.fn(
446 "relay.apns_deliveries.send_live_activity",445 "relay.apns_deliveries.send_live_activity",
447 )(function* (input) {446 )(function* (input) {
448 yield* Effect.annotateCurrentSpan({447 yield* Effect.annotateCurrentSpan({
...@@ -550,7 +549,7 @@ const make = Effect.gen(function* () {...@@ -550,7 +549,7 @@ const make = Effect.gen(function* () {
550 };549 };
551 });550 });
552551
553 const sendPushNotification: ApnsDeliveriesShape["sendPushNotification"] = Effect.fn(552 const sendPushNotification: ApnsDeliveries["Service"]["sendPushNotification"] = Effect.fn(
554 "relay.apns_deliveries.send_push_notification",553 "relay.apns_deliveries.send_push_notification",
555 )(function* (input) {554 )(function* (input) {
556 yield* Effect.annotateCurrentSpan({555 yield* Effect.annotateCurrentSpan({
...@@ -654,14 +653,14 @@ const make = Effect.gen(function* () {...@@ -654,14 +653,14 @@ const make = Effect.gen(function* () {
654 };653 };
655 });654 });
656655
657 const processSignedJob: ApnsDeliveriesShape["processSignedJob"] = Effect.fn(656 const processSignedJob: ApnsDeliveries["Service"]["processSignedJob"] = Effect.fn(
658 "relay.apns_deliveries.process_signed_job",657 "relay.apns_deliveries.process_signed_job",
659 )(function* (body) {658 )(function* (body) {
660 const signedJob = yield* decodeSignedApnsDeliveryJob(body).pipe(659 const signedJob = yield* decodeSignedApnsDeliveryJob(body).pipe(
661 Effect.mapError(660 Effect.mapError(
662 () =>661 () =>
663 new ApnsDeliveryJobInvalid({662 new ApnsDeliveryJobInvalid({
664 message: "Invalid APNs delivery queue job.",663 reason: "invalid-queue-payload",
665 }),664 }),
666 ),665 ),
667 );666 );
...@@ -686,7 +685,7 @@ const make = Effect.gen(function* () {...@@ -686,7 +685,7 @@ const make = Effect.gen(function* () {
686 if (payload.aggregate === null) {685 if (payload.aggregate === null) {
687 return Effect.fail(686 return Effect.fail(
688 new ApnsDeliveryJobInvalid({687 new ApnsDeliveryJobInvalid({
689 message: "Live Activity start/update jobs require an aggregate.",688 reason: "missing-live-activity-aggregate",
690 }),689 }),
691 );690 );
692 }691 }
...@@ -715,7 +714,7 @@ const make = Effect.gen(function* () {...@@ -715,7 +714,7 @@ const make = Effect.gen(function* () {
715 if (payload.notification === null) {714 if (payload.notification === null) {
716 return Effect.fail(715 return Effect.fail(
717 new ApnsDeliveryJobInvalid({716 new ApnsDeliveryJobInvalid({
718 message: "Push notification jobs require a notification payload.",717 reason: "missing-push-notification",
719 }),718 }),
720 );719 );
721 }720 }
infra/relay/src/agentActivity/ApnsDeliveryQueue.ts+21-24
...@@ -33,34 +33,31 @@ export class ApnsDeliveryQueueSendError extends Schema.TaggedErrorClass<ApnsDeli...@@ -33,34 +33,31 @@ export class ApnsDeliveryQueueSendError extends Schema.TaggedErrorClass<ApnsDeli
3333
34export type ApnsDeliveryQueueError = ApnsDeliveryQueueSendError;34export type ApnsDeliveryQueueError = ApnsDeliveryQueueSendError;
3535
36export interface ApnsDeliveryQueueSenderShape {
37 readonly send: (body: SignedApnsDeliveryJob) => Effect.Effect<void, ApnsDeliveryQueueSendError>;
38}
39
40export class ApnsDeliveryQueueSender extends Context.Service<36export class ApnsDeliveryQueueSender extends Context.Service<
41 ApnsDeliveryQueueSender,37 ApnsDeliveryQueueSender,
42 ApnsDeliveryQueueSenderShape38 {
39 readonly send: (body: SignedApnsDeliveryJob) => Effect.Effect<void, ApnsDeliveryQueueSendError>;
40 }
43>()("t3code-relay/agentActivity/ApnsDeliveryQueue/ApnsDeliveryQueueSender") {}41>()("t3code-relay/agentActivity/ApnsDeliveryQueue/ApnsDeliveryQueueSender") {}
4442
45export interface ApnsDeliveryQueueShape {43export class ApnsDeliveryQueue extends Context.Service<
46 readonly enqueueLiveActivity: (input: {44 ApnsDeliveryQueue,
47 readonly kind: ApnsDeliveryJobPayload["kind"];45 {
48 readonly userId: string;46 readonly enqueueLiveActivity: (input: {
49 readonly deviceId: string;47 readonly kind: ApnsDeliveryJobPayload["kind"];
50 readonly token: string;48 readonly userId: string;
51 readonly aggregate: ApnsDeliveryJobPayload["aggregate"];49 readonly deviceId: string;
52 }) => Effect.Effect<RelayDeliveryResult, ApnsDeliveryQueueError>;50 readonly token: string;
53 readonly enqueuePushNotification: (input: {51 readonly aggregate: ApnsDeliveryJobPayload["aggregate"];
54 readonly userId: string;52 }) => Effect.Effect<RelayDeliveryResult, ApnsDeliveryQueueError>;
55 readonly deviceId: string;53 readonly enqueuePushNotification: (input: {
56 readonly token: string;54 readonly userId: string;
57 readonly notification: NonNullable<ApnsDeliveryJobPayload["notification"]>;55 readonly deviceId: string;
58 }) => Effect.Effect<RelayDeliveryResult, ApnsDeliveryQueueError>;56 readonly token: string;
59}57 readonly notification: NonNullable<ApnsDeliveryJobPayload["notification"]>;
6058 }) => Effect.Effect<RelayDeliveryResult, ApnsDeliveryQueueError>;
61export class ApnsDeliveryQueue extends Context.Service<ApnsDeliveryQueue, ApnsDeliveryQueueShape>()(59 }
62 "t3code-relay/agentActivity/ApnsDeliveryQueue",60>()("t3code-relay/agentActivity/ApnsDeliveryQueue") {}
63) {}
6461
65const make = Effect.gen(function* () {62const make = Effect.gen(function* () {
66 const sender = yield* ApnsDeliveryQueueSender;63 const sender = yield* ApnsDeliveryQueueSender;
infra/relay/src/agentActivity/DeliveryAttempts.ts+14-15
...@@ -43,21 +43,20 @@ export interface DeliveryAttemptCompletionInput {...@@ -43,21 +43,20 @@ export interface DeliveryAttemptCompletionInput {
4343
44export type DeliverySourceJobClaimResult = "claimed" | "completed" | "in_flight";44export type DeliverySourceJobClaimResult = "claimed" | "completed" | "in_flight";
4545
46export interface DeliveryAttemptsShape {46export class DeliveryAttempts extends Context.Service<
47 readonly record: (47 DeliveryAttempts,
48 input: DeliveryAttemptInput,48 {
49 ) => Effect.Effect<void, DeliveryAttemptRecordPersistenceError>;49 readonly record: (
50 readonly claimSourceJob: (50 input: DeliveryAttemptInput,
51 input: DeliveryAttemptInput & { readonly sourceJobId: string },51 ) => Effect.Effect<void, DeliveryAttemptRecordPersistenceError>;
52 ) => Effect.Effect<DeliverySourceJobClaimResult, DeliveryAttemptRecordPersistenceError>;52 readonly claimSourceJob: (
53 readonly completeSourceJob: (53 input: DeliveryAttemptInput & { readonly sourceJobId: string },
54 input: DeliveryAttemptCompletionInput,54 ) => Effect.Effect<DeliverySourceJobClaimResult, DeliveryAttemptRecordPersistenceError>;
55 ) => Effect.Effect<void, DeliveryAttemptRecordPersistenceError>;55 readonly completeSourceJob: (
56}56 input: DeliveryAttemptCompletionInput,
5757 ) => Effect.Effect<void, DeliveryAttemptRecordPersistenceError>;
58export class DeliveryAttempts extends Context.Service<DeliveryAttempts, DeliveryAttemptsShape>()(58 }
59 "t3code-relay/agentActivity/DeliveryAttempts",59>()("t3code-relay/agentActivity/DeliveryAttempts") {}
60) {}
6160
62const SOURCE_JOB_CLAIM_LEASE_MINUTES = 10;61const SOURCE_JOB_CLAIM_LEASE_MINUTES = 10;
6362
infra/relay/src/agentActivity/Devices.ts+16-17
...@@ -40,23 +40,22 @@ export class DeviceListPersistenceError extends Schema.TaggedErrorClass<DeviceLi...@@ -40,23 +40,22 @@ export class DeviceListPersistenceError extends Schema.TaggedErrorClass<DeviceLi
40 }40 }
41}41}
4242
43export interface DevicesShape {43export class Devices extends Context.Service<
44 readonly register: (input: {44 Devices,
45 readonly userId: string;45 {
46 readonly registration: RelayDeviceRegistrationRequest;46 readonly register: (input: {
47 }) => Effect.Effect<void, DeviceRegistrationPersistenceError>;47 readonly userId: string;
48 readonly unregister: (input: {48 readonly registration: RelayDeviceRegistrationRequest;
49 readonly userId: string;49 }) => Effect.Effect<void, DeviceRegistrationPersistenceError>;
50 readonly deviceId: string;50 readonly unregister: (input: {
51 }) => Effect.Effect<void, DeviceUnregistrationPersistenceError>;51 readonly userId: string;
52 readonly listForUser: (input: {52 readonly deviceId: string;
53 readonly userId: string;53 }) => Effect.Effect<void, DeviceUnregistrationPersistenceError>;
54 }) => Effect.Effect<ReadonlyArray<RelayClientDeviceRecord>, DeviceListPersistenceError>;54 readonly listForUser: (input: {
55}55 readonly userId: string;
5656 }) => Effect.Effect<ReadonlyArray<RelayClientDeviceRecord>, DeviceListPersistenceError>;
57export class Devices extends Context.Service<Devices, DevicesShape>()(57 }
58 "t3code-relay/agentActivity/Devices",58>()("t3code-relay/agentActivity/Devices") {}
59) {}
6059
61const make = Effect.gen(function* () {60const make = Effect.gen(function* () {
62 const db = yield* RelayDb.RelayDb;61 const db = yield* RelayDb.RelayDb;
infra/relay/src/agentActivity/LiveActivities.ts+36-37
...@@ -7,7 +7,7 @@ import { RelayAgentActivityAggregateState as RelayAgentActivityAggregateStateSch...@@ -7,7 +7,7 @@ import { RelayAgentActivityAggregateState as RelayAgentActivityAggregateStateSch
7import * as Context from "effect/Context";7import * as Context from "effect/Context";
8import * as DateTime from "effect/DateTime";8import * as DateTime from "effect/DateTime";
9import * as Effect from "effect/Effect";9import * as Effect from "effect/Effect";
10import { cast } from "effect/Function";10import * as Function from "effect/Function";
11import * as Layer from "effect/Layer";11import * as Layer from "effect/Layer";
12import * as Schema from "effect/Schema";12import * as Schema from "effect/Schema";
13import { and, eq, sql } from "drizzle-orm";13import { and, eq, sql } from "drizzle-orm";
...@@ -64,41 +64,40 @@ export interface LiveActivityRow {...@@ -64,41 +64,40 @@ export interface LiveActivityRow {
6464
65export type TargetRow = DeviceRow & LiveActivityRow;65export type TargetRow = DeviceRow & LiveActivityRow;
6666
67export interface LiveActivitiesShape {67export class LiveActivities extends Context.Service<
68 readonly register: (input: {68 LiveActivities,
69 readonly userId: string;69 {
70 readonly registration: RelayLiveActivityRegistrationRequest;70 readonly register: (input: {
71 }) => Effect.Effect<void, LiveActivityRegistrationPersistenceError>;71 readonly userId: string;
72 readonly listTargets: (input: {72 readonly registration: RelayLiveActivityRegistrationRequest;
73 readonly userId: string;73 }) => Effect.Effect<void, LiveActivityRegistrationPersistenceError>;
74 }) => Effect.Effect<ReadonlyArray<TargetRow>, LiveActivityTargetListPersistenceError>;74 readonly listTargets: (input: {
75 readonly markDelivery: (input: {75 readonly userId: string;
76 readonly userId: string;76 }) => Effect.Effect<ReadonlyArray<TargetRow>, LiveActivityTargetListPersistenceError>;
77 readonly deviceId: string;77 readonly markDelivery: (input: {
78 readonly kind: RelayDeliveryKind;78 readonly userId: string;
79 readonly aggregate: RelayAgentActivityAggregateState | null;79 readonly deviceId: string;
80 readonly deliveredAt: string;80 readonly kind: RelayDeliveryKind;
81 }) => Effect.Effect<void, LiveActivityDeliveryMarkPersistenceError>;81 readonly aggregate: RelayAgentActivityAggregateState | null;
82 readonly markStartQueued: (input: {82 readonly deliveredAt: string;
83 readonly userId: string;83 }) => Effect.Effect<void, LiveActivityDeliveryMarkPersistenceError>;
84 readonly deviceId: string;84 readonly markStartQueued: (input: {
85 readonly queuedAt: string;85 readonly userId: string;
86 }) => Effect.Effect<void, LiveActivityDeliveryMarkPersistenceError>;86 readonly deviceId: string;
87 readonly clearStartQueued: (input: {87 readonly queuedAt: string;
88 readonly userId: string;88 }) => Effect.Effect<void, LiveActivityDeliveryMarkPersistenceError>;
89 readonly deviceId: string;89 readonly clearStartQueued: (input: {
90 }) => Effect.Effect<void, LiveActivityDeliveryMarkPersistenceError>;90 readonly userId: string;
91 readonly invalidateDeliveryToken: (input: {91 readonly deviceId: string;
92 readonly userId: string;92 }) => Effect.Effect<void, LiveActivityDeliveryMarkPersistenceError>;
93 readonly deviceId: string;93 readonly invalidateDeliveryToken: (input: {
94 readonly kind: RelayDeliveryKind;94 readonly userId: string;
95 readonly invalidatedAt: string;95 readonly deviceId: string;
96 }) => Effect.Effect<void, LiveActivityDeliveryMarkPersistenceError>;96 readonly kind: RelayDeliveryKind;
97}97 readonly invalidatedAt: string;
9898 }) => Effect.Effect<void, LiveActivityDeliveryMarkPersistenceError>;
99export class LiveActivities extends Context.Service<LiveActivities, LiveActivitiesShape>()(99 }
100 "t3code-relay/agentActivity/LiveActivities",100>()("t3code-relay/agentActivity/LiveActivities") {}
101) {}
102101
103const decodeJsonString = Schema.decodeEffect(Schema.UnknownFromJsonString);102const decodeJsonString = Schema.decodeEffect(Schema.UnknownFromJsonString);
104const encodeJsonValue = Schema.encodeEffect(Schema.UnknownFromJsonString);103const encodeJsonValue = Schema.encodeEffect(Schema.UnknownFromJsonString);
...@@ -223,7 +222,7 @@ const make = Effect.gen(function* () {...@@ -223,7 +222,7 @@ const make = Effect.gen(function* () {
223 ? null222 ? null
224 : yield* encodeRelayAgentActivityAggregateStateJson(input.aggregate).pipe(223 : yield* encodeRelayAgentActivityAggregateStateJson(input.aggregate).pipe(
225 Effect.flatMap(decodeJsonString),224 Effect.flatMap(decodeJsonString),
226 Effect.map(cast<unknown, RelayAgentActivityAggregateState>),225 Effect.map(Function.cast<unknown, RelayAgentActivityAggregateState>),
227 );226 );
228227
229 yield* db228 yield* db
infra/relay/src/agentActivity/MobileRegistrations.test.ts+21-18
...@@ -38,7 +38,9 @@ const device: RelayDeviceRegistrationRequest = {...@@ -38,7 +38,9 @@ const device: RelayDeviceRegistrationRequest = {
38 },38 },
39};39};
4040
41function makeDevices(overrides: Partial<Devices.DevicesShape> = {}): Devices.DevicesShape {41function makeDevices(
42 overrides: Partial<Devices.Devices["Service"]> = {},
43): Devices.Devices["Service"] {
42 return {44 return {
43 register: () => Effect.void,45 register: () => Effect.void,
44 unregister: () => Effect.void,46 unregister: () => Effect.void,
...@@ -48,8 +50,8 @@ function makeDevices(overrides: Partial<Devices.DevicesShape> = {}): Devices.Dev...@@ -48,8 +50,8 @@ function makeDevices(overrides: Partial<Devices.DevicesShape> = {}): Devices.Dev
48}50}
4951
50function makeLiveActivities(52function makeLiveActivities(
51 overrides: Partial<LiveActivities.LiveActivitiesShape> = {},53 overrides: Partial<LiveActivities.LiveActivities["Service"]> = {},
52): LiveActivities.LiveActivitiesShape {54): LiveActivities.LiveActivities["Service"] {
53 return {55 return {
54 register: () => Effect.void,56 register: () => Effect.void,
55 listTargets: () => Effect.succeed([]),57 listTargets: () => Effect.succeed([]),
...@@ -62,8 +64,8 @@ function makeLiveActivities(...@@ -62,8 +64,8 @@ function makeLiveActivities(
62}64}
6365
64function makeAgentActivityRows(66function makeAgentActivityRows(
65 overrides: Partial<AgentActivityRows.AgentActivityRowsShape> = {},67 overrides: Partial<AgentActivityRows.AgentActivityRows["Service"]> = {},
66): AgentActivityRows.AgentActivityRowsShape {68): AgentActivityRows.AgentActivityRows["Service"] {
67 return {69 return {
68 upsert: () => Effect.void,70 upsert: () => Effect.void,
69 remove: () => Effect.void,71 remove: () => Effect.void,
...@@ -108,8 +110,8 @@ function makeEnvironmentLinks(...@@ -108,8 +110,8 @@ function makeEnvironmentLinks(
108}110}
109111
110function makeDeliveryAttempts(112function makeDeliveryAttempts(
111 overrides: Partial<DeliveryAttempts.DeliveryAttemptsShape> = {},113 overrides: Partial<DeliveryAttempts.DeliveryAttempts["Service"]> = {},
112): DeliveryAttempts.DeliveryAttemptsShape {114): DeliveryAttempts.DeliveryAttempts["Service"] {
113 return {115 return {
114 record: () => Effect.void,116 record: () => Effect.void,
115 claimSourceJob: () => Effect.succeed("claimed"),117 claimSourceJob: () => Effect.succeed("claimed"),
...@@ -138,8 +140,8 @@ const config = RelayConfiguration.RelayConfiguration.of({...@@ -138,8 +140,8 @@ const config = RelayConfiguration.RelayConfiguration.of({
138});140});
139141
140function makeRegistrationReplayLayer(input: {142function makeRegistrationReplayLayer(input: {
141 readonly devices: Devices.DevicesShape;143 readonly devices: Devices.Devices["Service"];
142 readonly liveActivities: LiveActivities.LiveActivitiesShape;144 readonly liveActivities: LiveActivities.LiveActivities["Service"];
143 readonly queuedJobs: Array<SignedApnsDeliveryJob>;145 readonly queuedJobs: Array<SignedApnsDeliveryJob>;
144}) {146}) {
145 return MobileRegistrations.layer.pipe(147 return MobileRegistrations.layer.pipe(
...@@ -167,8 +169,8 @@ function makeRegistrationReplayLayer(input: {...@@ -167,8 +169,8 @@ function makeRegistrationReplayLayer(input: {
167}169}
168170
169function makeAgentActivityPublisher(171function makeAgentActivityPublisher(
170 overrides: Partial<AgentActivityPublisher.AgentActivityPublisherShape> = {},172 overrides: Partial<AgentActivityPublisher.AgentActivityPublisher["Service"]> = {},
171): AgentActivityPublisher.AgentActivityPublisherShape {173): AgentActivityPublisher.AgentActivityPublisher["Service"] {
172 return {174 return {
173 publish: () => Effect.succeed({ ok: true, deliveries: [] }),175 publish: () => Effect.succeed({ ok: true, deliveries: [] }),
174 replayForLiveActivityRegistration: () => Effect.succeed(null),176 replayForLiveActivityRegistration: () => Effect.succeed(null),
...@@ -178,10 +180,10 @@ function makeAgentActivityPublisher(...@@ -178,10 +180,10 @@ function makeAgentActivityPublisher(
178180
179describe("MobileRegistrations", () => {181describe("MobileRegistrations", () => {
180 it.effect("registers devices through the device persistence service", () => {182 it.effect("registers devices through the device persistence service", () => {
181 let registered: Parameters<Devices.DevicesShape["register"]>[0] | null = null;183 let registered: Parameters<Devices.Devices["Service"]["register"]>[0] | null = null;
182 let replayed:184 let replayed:
183 | Parameters<185 | Parameters<
184 AgentActivityPublisher.AgentActivityPublisherShape["replayForLiveActivityRegistration"]186 AgentActivityPublisher.AgentActivityPublisher["Service"]["replayForLiveActivityRegistration"]
185 >[0]187 >[0]
186 | null = null;188 | null = null;
187189
...@@ -263,7 +265,7 @@ describe("MobileRegistrations", () => {...@@ -263,7 +265,7 @@ describe("MobileRegistrations", () => {
263 });265 });
264266
265 it.effect("unregisters the current user's device", () => {267 it.effect("unregisters the current user's device", () => {
266 let unregistered: Parameters<Devices.DevicesShape["unregister"]>[0] | null = null;268 let unregistered: Parameters<Devices.Devices["Service"]["unregister"]>[0] | null = null;
267269
268 return Effect.gen(function* () {270 return Effect.gen(function* () {
269 const result = yield* Effect.gen(function* () {271 const result = yield* Effect.gen(function* () {
...@@ -310,10 +312,11 @@ describe("MobileRegistrations", () => {...@@ -310,10 +312,11 @@ describe("MobileRegistrations", () => {
310 deviceId: "device-1" as const,312 deviceId: "device-1" as const,
311 activityPushToken: "activity-token" as const,313 activityPushToken: "activity-token" as const,
312 };314 };
313 let registered: Parameters<LiveActivities.LiveActivitiesShape["register"]>[0] | null = null;315 let registered: Parameters<LiveActivities.LiveActivities["Service"]["register"]>[0] | null =
316 null;
314 let replayed:317 let replayed:
315 | Parameters<318 | Parameters<
316 AgentActivityPublisher.AgentActivityPublisherShape["replayForLiveActivityRegistration"]319 AgentActivityPublisher.AgentActivityPublisher["Service"]["replayForLiveActivityRegistration"]
317 >[0]320 >[0]
318 | null = null;321 | null = null;
319322
...@@ -372,9 +375,9 @@ describe("MobileRegistrations", () => {...@@ -372,9 +375,9 @@ describe("MobileRegistrations", () => {
372 () => {375 () => {
373 const queuedJobs: Array<SignedApnsDeliveryJob> = [];376 const queuedJobs: Array<SignedApnsDeliveryJob> = [];
374 const queuedStarts: Array<377 const queuedStarts: Array<
375 Parameters<LiveActivities.LiveActivitiesShape["markStartQueued"]>[0]378 Parameters<LiveActivities.LiveActivities["Service"]["markStartQueued"]>[0]
376 > = [];379 > = [];
377 const registeredDevices: Array<Parameters<Devices.DevicesShape["register"]>[0]> = [];380 const registeredDevices: Array<Parameters<Devices.Devices["Service"]["register"]>[0]> = [];
378 const devices = makeDevices({381 const devices = makeDevices({
379 register: (input) =>382 register: (input) =>
380 Effect.sync(() => {383 Effect.sync(() => {
infra/relay/src/agentActivity/MobileRegistrations.ts+14-16
...@@ -15,24 +15,22 @@ export type MobileRegistrationError =...@@ -15,24 +15,22 @@ export type MobileRegistrationError =
15 | Devices.DeviceUnregistrationPersistenceError15 | Devices.DeviceUnregistrationPersistenceError
16 | LiveActivities.LiveActivityRegistrationPersistenceError;16 | LiveActivities.LiveActivityRegistrationPersistenceError;
1717
18export interface MobileRegistrationsShape {
19 readonly registerDevice: (input: {
20 readonly userId: string;
21 readonly payload: RelayDeviceRegistrationRequest;
22 }) => Effect.Effect<{ readonly ok: true }, MobileRegistrationError>;
23 readonly registerLiveActivity: (input: {
24 readonly userId: string;
25 readonly payload: RelayLiveActivityRegistrationRequest;
26 }) => Effect.Effect<{ readonly ok: true }, MobileRegistrationError>;
27 readonly unregisterDevice: (input: {
28 readonly userId: string;
29 readonly deviceId: string;
30 }) => Effect.Effect<{ readonly ok: true }, MobileRegistrationError>;
31}
32
33export class MobileRegistrations extends Context.Service<18export class MobileRegistrations extends Context.Service<
34 MobileRegistrations,19 MobileRegistrations,
35 MobileRegistrationsShape20 {
21 readonly registerDevice: (input: {
22 readonly userId: string;
23 readonly payload: RelayDeviceRegistrationRequest;
24 }) => Effect.Effect<{ readonly ok: true }, MobileRegistrationError>;
25 readonly registerLiveActivity: (input: {
26 readonly userId: string;
27 readonly payload: RelayLiveActivityRegistrationRequest;
28 }) => Effect.Effect<{ readonly ok: true }, MobileRegistrationError>;
29 readonly unregisterDevice: (input: {
30 readonly userId: string;
31 readonly deviceId: string;
32 }) => Effect.Effect<{ readonly ok: true }, MobileRegistrationError>;
33 }
36>()("t3code-relay/agentActivity/MobileRegistrations") {}34>()("t3code-relay/agentActivity/MobileRegistrations") {}
3735
38const make = Effect.gen(function* () {36const make = Effect.gen(function* () {
infra/relay/src/agentActivity/apnsDeliveryJobs.ts+48-12
...@@ -52,9 +52,45 @@ export type SignedApnsDeliveryJob = typeof SignedApnsDeliveryJob.Type;...@@ -52,9 +52,45 @@ export type SignedApnsDeliveryJob = typeof SignedApnsDeliveryJob.Type;
52export class ApnsDeliveryJobInvalid extends Schema.TaggedErrorClass<ApnsDeliveryJobInvalid>()(52export class ApnsDeliveryJobInvalid extends Schema.TaggedErrorClass<ApnsDeliveryJobInvalid>()(
53 "ApnsDeliveryJobInvalid",53 "ApnsDeliveryJobInvalid",
54 {54 {
55 message: Schema.String,55 reason: Schema.Literals([
56 "invalid-queue-payload",
57 "missing-live-activity-aggregate",
58 "unexpected-live-activity-notification",
59 "missing-push-notification",
60 "unexpected-push-notification-aggregate",
61 "invalid-created-at",
62 "invalid-expires-at",
63 "invalid-time-window",
64 "time-window-too-long",
65 "invalid-signature",
66 ]),
56 },67 },
57) {}68) {
69 override get message(): string {
70 switch (this.reason) {
71 case "invalid-queue-payload":
72 return "Invalid APNs delivery queue job.";
73 case "missing-live-activity-aggregate":
74 return "Live Activity start/update jobs require an aggregate.";
75 case "unexpected-live-activity-notification":
76 return "Live Activity jobs must not carry push notification payloads.";
77 case "missing-push-notification":
78 return "Push notification jobs require a notification payload.";
79 case "unexpected-push-notification-aggregate":
80 return "Push notification jobs must not carry aggregate state.";
81 case "invalid-created-at":
82 return "Invalid APNs delivery job creation time.";
83 case "invalid-expires-at":
84 return "Invalid APNs delivery job expiry.";
85 case "invalid-time-window":
86 return "Invalid APNs delivery job time window.";
87 case "time-window-too-long":
88 return "APNs delivery job time window is too long.";
89 case "invalid-signature":
90 return "Invalid APNs delivery job signature.";
91 }
92 }
93}
5894
59export class ApnsDeliveryJobExpired extends Schema.TaggedErrorClass<ApnsDeliveryJobExpired>()(95export class ApnsDeliveryJobExpired extends Schema.TaggedErrorClass<ApnsDeliveryJobExpired>()(
60 "ApnsDeliveryJobExpired",96 "ApnsDeliveryJobExpired",
...@@ -106,31 +142,31 @@ function validatePayloadShape(payload: ApnsDeliveryJobPayload): ApnsDeliveryJobI...@@ -106,31 +142,31 @@ function validatePayloadShape(payload: ApnsDeliveryJobPayload): ApnsDeliveryJobI
106 case "live_activity_update":142 case "live_activity_update":
107 if (payload.aggregate === null) {143 if (payload.aggregate === null) {
108 return new ApnsDeliveryJobInvalid({144 return new ApnsDeliveryJobInvalid({
109 message: "Live Activity start/update jobs require an aggregate.",145 reason: "missing-live-activity-aggregate",
110 });146 });
111 }147 }
112 if (payload.notification !== null) {148 if (payload.notification !== null) {
113 return new ApnsDeliveryJobInvalid({149 return new ApnsDeliveryJobInvalid({
114 message: "Live Activity jobs must not carry push notification payloads.",150 reason: "unexpected-live-activity-notification",
115 });151 });
116 }152 }
117 return null;153 return null;
118 case "live_activity_end":154 case "live_activity_end":
119 if (payload.notification !== null) {155 if (payload.notification !== null) {
120 return new ApnsDeliveryJobInvalid({156 return new ApnsDeliveryJobInvalid({
121 message: "Live Activity jobs must not carry push notification payloads.",157 reason: "unexpected-live-activity-notification",
122 });158 });
123 }159 }
124 return null;160 return null;
125 case "push_notification":161 case "push_notification":
126 if (payload.notification === null) {162 if (payload.notification === null) {
127 return new ApnsDeliveryJobInvalid({163 return new ApnsDeliveryJobInvalid({
128 message: "Push notification jobs require a notification payload.",164 reason: "missing-push-notification",
129 });165 });
130 }166 }
131 if (payload.aggregate !== null) {167 if (payload.aggregate !== null) {
132 return new ApnsDeliveryJobInvalid({168 return new ApnsDeliveryJobInvalid({
133 message: "Push notification jobs must not carry aggregate state.",169 reason: "unexpected-push-notification-aggregate",
134 });170 });
135 }171 }
136 return null;172 return null;
...@@ -177,19 +213,19 @@ export function verifySignedApnsDeliveryJob(input: {...@@ -177,19 +213,19 @@ export function verifySignedApnsDeliveryJob(input: {
177 }213 }
178 const createdAt = DateTime.make(input.job.payload.createdAt);214 const createdAt = DateTime.make(input.job.payload.createdAt);
179 if (Option.isNone(createdAt)) {215 if (Option.isNone(createdAt)) {
180 return new ApnsDeliveryJobInvalid({ message: "Invalid APNs delivery job creation time." });216 return new ApnsDeliveryJobInvalid({ reason: "invalid-created-at" });
181 }217 }
182 const expiresAt = DateTime.make(input.job.payload.expiresAt);218 const expiresAt = DateTime.make(input.job.payload.expiresAt);
183 if (Option.isNone(expiresAt)) {219 if (Option.isNone(expiresAt)) {
184 return new ApnsDeliveryJobInvalid({ message: "Invalid APNs delivery job expiry." });220 return new ApnsDeliveryJobInvalid({ reason: "invalid-expires-at" });
185 }221 }
186 const createdAtMs = createdAt.value.epochMilliseconds;222 const createdAtMs = createdAt.value.epochMilliseconds;
187 const expiresAtMs = expiresAt.value.epochMilliseconds;223 const expiresAtMs = expiresAt.value.epochMilliseconds;
188 if (expiresAtMs <= createdAtMs) {224 if (expiresAtMs <= createdAtMs) {
189 return new ApnsDeliveryJobInvalid({ message: "Invalid APNs delivery job time window." });225 return new ApnsDeliveryJobInvalid({ reason: "invalid-time-window" });
190 }226 }
191 if (expiresAtMs - createdAtMs > MAX_JOB_AGE_MS) {227 if (expiresAtMs - createdAtMs > MAX_JOB_AGE_MS) {
192 return new ApnsDeliveryJobInvalid({ message: "APNs delivery job time window is too long." });228 return new ApnsDeliveryJobInvalid({ reason: "time-window-too-long" });
193 }229 }
194 if (expiresAtMs <= input.nowMs) {230 if (expiresAtMs <= input.nowMs) {
195 return new ApnsDeliveryJobExpired({ expiresAt: input.job.payload.expiresAt });231 return new ApnsDeliveryJobExpired({ expiresAt: input.job.payload.expiresAt });
...@@ -199,7 +235,7 @@ export function verifySignedApnsDeliveryJob(input: {...@@ -199,7 +235,7 @@ export function verifySignedApnsDeliveryJob(input: {
199 payload: input.job.payload,235 payload: input.job.payload,
200 });236 });
201 if (!timingSafeEqualBase64Url(input.job.signature, expected)) {237 if (!timingSafeEqualBase64Url(input.job.signature, expected)) {
202 return new ApnsDeliveryJobInvalid({ message: "Invalid APNs delivery job signature." });238 return new ApnsDeliveryJobInvalid({ reason: "invalid-signature" });
203 }239 }
204 return input.job.payload;240 return input.job.payload;
205}241}