Open History to see the last commit
1import {
2 makeAggregateState,
3 TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS,
4} from "./agentActivityAggregate.ts";
5import type {
6 RelayAgentActivityState,
7 RelayDeliveryResult,
8 RelayPublishResponse,
9} from "@t3tools/contracts/relay";
10import * as Arr from "effect/Array";
11import * as Context from "effect/Context";
12import * as DateTime from "effect/DateTime";
13import * as Effect from "effect/Effect";
14import * as Layer from "effect/Layer";
15
16import { isTerminalPhase } from "./agentActivityPayloads.ts";
17
18export { isExpiredAgentActivityState } from "./agentActivityPayloads.ts";
19import * as AgentActivityRows from "./AgentActivityRows.ts";
20import * as EnvironmentLinks from "../environments/EnvironmentLinks.ts";
21import * as LiveActivities from "./LiveActivities.ts";
22import * as ApnsDeliveries from "./ApnsDeliveries.ts";
23import * as FcmDeliveries from "./FcmDeliveries.ts";
24
25export type AgentActivityPublishError =
26 | FcmDeliveries.FcmDeliveryError
27 | AgentActivityRows.AgentActivityRowUpsertPersistenceError
28 | AgentActivityRows.AgentActivityRowDeletePersistenceError
29 | AgentActivityRows.AgentActivityRowListPersistenceError
30 | EnvironmentLinks.EnvironmentLinkUserListPersistenceError
31 | LiveActivities.LiveActivityTargetListPersistenceError
32 | ApnsDeliveries.ApnsDeliveryError;
33
34export class AgentActivityPublisher extends Context.Service<
35 AgentActivityPublisher,
36 {
37 readonly publish: (input: {
38 readonly environmentId: string;
39 readonly environmentPublicKey: string;
40 readonly threadId: string;
41 readonly state: RelayAgentActivityState | null;
42 }) => Effect.Effect<RelayPublishResponse, AgentActivityPublishError>;
43 readonly replayForLiveActivityRegistration: (input: {
44 readonly userId: string;
45 readonly deviceId: string;
46 }) => Effect.Effect<RelayDeliveryResult | null, AgentActivityPublishError>;
47 readonly endIdleLiveActivities: Effect.Effect<
48 void,
49 LiveActivities.LiveActivityIdleTargetListPersistenceError
50 >;
51 }
52>()("t3code-relay/agentActivity/AgentActivityPublisher") {}
53
54const make = Effect.gen(function* () {
55 const rows = yield* AgentActivityRows.AgentActivityRows;
56 const links = yield* EnvironmentLinks.EnvironmentLinks;
57 const liveActivities = yield* LiveActivities.LiveActivities;
58 const apnsDeliveries = yield* ApnsDeliveries.ApnsDeliveries;
59 const fcmDeliveries = yield* FcmDeliveries.FcmDeliveries;
60
61 const publishForDeliveryUser = Effect.fnUntraced(function* (input: {
62 readonly deliveryUser: EnvironmentLinks.AgentAwarenessDeliveryUserRecord;
63 readonly state: RelayAgentActivityState | null;
64 readonly nowMs: number;
65 }) {
66 const targets = yield* liveActivities.listTargets({ userId: input.deliveryUser.userId });
67 // Only iOS targets with a running Live Activity read the aggregate, and few users have one,
68 // so skip the per-user activity query otherwise.
69 const liveActivityAggregate =
70 input.deliveryUser.liveActivitiesEnabled &&
71 targets.some((target) => target.platform !== "android" && target.activity_push_token)
72 ? makeAggregateState({
73 activeStates: yield* rows.listForUser({ userId: input.deliveryUser.userId }),
74 terminalState: input.state && isTerminalPhase(input.state) ? input.state : null,
75 nowMs: input.nowMs,
76 })
77 : null;
78 return yield* Effect.forEach(
79 targets,
80 Effect.fnUntraced(function* (target) {
81 if (target.platform === "android") {
82 return yield* fcmDeliveries.enqueue({ target, state: input.state });
83 }
84 return yield* apnsDeliveries.sendForTarget({
85 target,
86 aggregate: liveActivityAggregate,
87 state: input.deliveryUser.notificationsEnabled ? input.state : null,
88 nowMs: input.nowMs,
89 });
90 }),
91 { concurrency: 4 },
92 );
93 });
94
95 return AgentActivityPublisher.of({
96 replayForLiveActivityRegistration: Effect.fn(
97 "relay.agent_activity_publisher.replay_for_live_activity_registration",
98 )(function* (input) {
99 yield* Effect.annotateCurrentSpan({
100 "relay.mobile.device_id": input.deviceId,
101 "relay.operation": "replayForLiveActivityRegistration",
102 });
103 const { activeStates, targets } = yield* Effect.all(
104 {
105 activeStates: rows.listForUser({ userId: input.userId }),
106 targets: liveActivities.listTargets({ userId: input.userId }),
107 },
108 { concurrency: 2 },
109 );
110 const target = targets.find((row) => row.device_id === input.deviceId) ?? null;
111 if (target === null) {
112 return null;
113 }
114 if (target.platform === "android") {
115 return yield* fcmDeliveries.enqueue({ target, state: null, replay: true });
116 }
117 const now = yield* DateTime.now;
118 const aggregate = makeAggregateState({
119 activeStates,
120 terminalState: null,
121 nowMs: now.epochMilliseconds,
122 });
123 return yield* apnsDeliveries.sendForTarget({
124 target,
125 aggregate,
126 nowMs: now.epochMilliseconds,
127 replay: true,
128 });
129 }),
130 // An idle card receives no more publishes to clear its expired finished rows.
131 endIdleLiveActivities: Effect.gen(function* () {
132 const now = yield* DateTime.now;
133 const idleTargets = yield* liveActivities.listIdleArmedTargets({
134 deliveredBefore: DateTime.formatIso(
135 DateTime.subtract(now, { milliseconds: TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS }),
136 ),
137 });
138 yield* Effect.annotateCurrentSpan({ "relay.live_activities.idle_count": idleTargets.length });
139 yield* Effect.forEach(
140 Object.entries(Arr.groupBy(idleTargets, (target) => target.user_id)),
141 ([userId, candidates]) =>
142 Effect.gen(function* () {
143 const activeStates = yield* rows.listForUser({ userId });
144 const aggregate = makeAggregateState({
145 activeStates,
146 terminalState: null,
147 nowMs: now.epochMilliseconds,
148 });
149 if (aggregate !== null) return;
150 const targets = yield* liveActivities.listTargets({ userId });
151 const idleDeviceIds = new Set(candidates.map((target) => target.device_id));
152 yield* Effect.forEach(
153 targets.filter((target) => idleDeviceIds.has(target.device_id)),
154 (target) =>
155 apnsDeliveries
156 .sendForTarget({
157 target,
158 aggregate: null,
159 nowMs: now.epochMilliseconds,
160 replay: true,
161 })
162 .pipe(
163 Effect.catch((error) =>
164 Effect.logWarning("Failed to end idle Live Activity", {
165 deviceId: target.device_id,
166 error,
167 }),
168 ),
169 ),
170 { discard: true },
171 );
172 }).pipe(
173 Effect.catch((error) =>
174 Effect.logWarning("Failed to list idle Live Activity content or targets", {
175 userId,
176 error,
177 }),
178 ),
179 ),
180 { concurrency: 4, discard: true },
181 );
182 }).pipe(Effect.withSpan("relay.agent_activity_publisher.end_idle_live_activities")),
183 publish: Effect.fn("relay.agent_activity_publisher.publish")(function* (input) {
184 yield* Effect.annotateCurrentSpan({
185 "relay.environment_id": input.environmentId,
186 "relay.thread_id": input.threadId,
187 "relay.agent_activity.phase": input.state?.phase ?? "deleted",
188 });
189 if (input.state) {
190 // Retain terminal rows until cron pruning so completed threads remain visible alongside active ones.
191 yield* rows.upsert({
192 environmentPublicKey: input.environmentPublicKey,
193 state: input.state,
194 });
195 } else {
196 yield* rows.remove({
197 environmentId: input.environmentId,
198 environmentPublicKey: input.environmentPublicKey,
199 threadId: input.threadId,
200 });
201 }
202
203 const deliveryUsers = yield* links.listDeliveryUsersForEnvironment({
204 environmentId: input.environmentId,
205 environmentPublicKey: input.environmentPublicKey,
206 });
207 const now = yield* DateTime.now;
208 const deliveriesByUser = yield* Effect.forEach(
209 deliveryUsers,
210 (deliveryUser) =>
211 publishForDeliveryUser({
212 deliveryUser,
213 state: input.state,
214 nowMs: now.epochMilliseconds,
215 }),
216 { concurrency: 4 },
217 );
218 const deliveries = deliveriesByUser.flat();
219 return {
220 ok: true,
221 deliveries: deliveries.filter(
222 (delivery): delivery is RelayDeliveryResult => delivery !== null,
223 ),
224 };
225 }),
226 });
227});
228
229export const layer = Layer.effect(AgentActivityPublisher, make);