1import { makeAggregateState } from "./agentActivityAggregate.ts";
2export {
3 makeAggregateState,
4 TERMINAL_AGENT_ACTIVITY_DISPLAY_TTL_MS,
5} from "./agentActivityAggregate.ts";
6import type {
7 RelayAgentActivityState,
8 RelayDeliveryResult,
9 RelayPublishResponse,
10} from "@t3tools/contracts/relay";
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 }
48>()("t3code-relay/agentActivity/AgentActivityPublisher") {}
49
50export const make = Effect.gen(function* () {
51 const rows = yield* AgentActivityRows.AgentActivityRows;
52 const links = yield* EnvironmentLinks.EnvironmentLinks;
53 const liveActivities = yield* LiveActivities.LiveActivities;
54 const apnsDeliveries = yield* ApnsDeliveries.ApnsDeliveries;
55 const fcmDeliveries = yield* FcmDeliveries.FcmDeliveries;
56
57 const publishForDeliveryUser = Effect.fnUntraced(function* (input: {
58 readonly deliveryUser: EnvironmentLinks.AgentAwarenessDeliveryUserRecord;
59 readonly state: RelayAgentActivityState | null;
60 readonly nowMs: number;
61 }) {
62 const activeStates = input.deliveryUser.liveActivitiesEnabled
63 ? yield* rows.listForUser({ userId: input.deliveryUser.userId })
64 : [];
65 const liveActivityAggregate = input.deliveryUser.liveActivitiesEnabled
66 ? makeAggregateState({
67 activeStates,
68 terminalState: input.state && isTerminalPhase(input.state) ? input.state : null,
69 nowMs: input.nowMs,
70 })
71 : null;
72 const notificationOnlyAggregate =
73 input.deliveryUser.notificationsEnabled &&
74 !input.deliveryUser.liveActivitiesEnabled &&
75 input.state !== null
76 ? makeAggregateState({
77 activeStates: isTerminalPhase(input.state) ? [] : [input.state],
78 terminalState: isTerminalPhase(input.state) ? input.state : null,
79 nowMs: input.nowMs,
80 })
81 : null;
82 const targets = yield* liveActivities.listTargets({ userId: input.deliveryUser.userId });
83 const deliveriesByTarget = yield* Effect.forEach(
84 targets,
85 Effect.fnUntraced(function* (target) {
86 if (target.platform === "android") {
87 return [yield* fcmDeliveries.enqueue({ target, state: input.state })];
88 }
89 return yield* Effect.all(
90 [
91 apnsDeliveries.sendForTarget({
92 target,
93 aggregate: liveActivityAggregate,
94 nowMs: input.nowMs,
95 }),
96 notificationOnlyAggregate === null
97 ? Effect.succeed(null)
98 : apnsDeliveries.sendPushNotificationForTarget({
99 target,
100 aggregate: notificationOnlyAggregate,
101 }),
102 ],
103 { concurrency: 2 },
104 );
105 }),
106 { concurrency: 4 },
107 );
108 return deliveriesByTarget.flat();
109 });
110
111 return AgentActivityPublisher.of({
112 replayForLiveActivityRegistration: Effect.fn(
113 "relay.agent_activity_publisher.replay_for_live_activity_registration",
114 )(function* (input) {
115 yield* Effect.annotateCurrentSpan({
116 "relay.mobile.device_id": input.deviceId,
117 "relay.operation": "replayForLiveActivityRegistration",
118 });
119 const { activeStates, targets } = yield* Effect.all(
120 {
121 activeStates: rows.listForUser({ userId: input.userId }),
122 targets: liveActivities.listTargets({ userId: input.userId }),
123 },
124 { concurrency: 2 },
125 );
126 const target = targets.find((row) => row.device_id === input.deviceId) ?? null;
127 if (target === null) {
128 return null;
129 }
130 if (target.platform === "android") {
131 return yield* fcmDeliveries.enqueue({ target, state: null, replay: true });
132 }
133 const now = yield* DateTime.now;
134 const aggregate = makeAggregateState({
135 activeStates,
136 terminalState: null,
137 nowMs: now.epochMilliseconds,
138 });
139 return yield* apnsDeliveries.sendForTarget({
140 target,
141 aggregate,
142 nowMs: now.epochMilliseconds,
143 replay: true,
144 });
145 }),
146 publish: Effect.fn("relay.agent_activity_publisher.publish")(function* (input) {
147 yield* Effect.annotateCurrentSpan({
148 "relay.environment_id": input.environmentId,
149 "relay.thread_id": input.threadId,
150 "relay.agent_activity.phase": input.state?.phase ?? "deleted",
151 });
152 if (input.state) {
153 // Terminal states are persisted too (pruned by the cron after they
154 // age out) so a thread that finishes while other agents are active
155 // stays visible as Done/Failed in subsequent aggregates instead of
156 // silently vanishing from the Live Activity.
157 yield* rows.upsert({
158 environmentPublicKey: input.environmentPublicKey,
159 state: input.state,
160 });
161 } else {
162 yield* rows.remove({
163 environmentId: input.environmentId,
164 environmentPublicKey: input.environmentPublicKey,
165 threadId: input.threadId,
166 });
167 }
168
169 const deliveryUsers = yield* links.listDeliveryUsersForEnvironment({
170 environmentId: input.environmentId,
171 environmentPublicKey: input.environmentPublicKey,
172 });
173 const now = yield* DateTime.now;
174 const deliveriesByUser = yield* Effect.forEach(
175 deliveryUsers,
176 (deliveryUser) =>
177 publishForDeliveryUser({
178 deliveryUser,
179 state: input.state,
180 nowMs: now.epochMilliseconds,
181 }),
182 { concurrency: 4 },
183 );
184 const deliveries = deliveriesByUser.flat();
185 return {
186 ok: true,
187 deliveries: deliveries.filter(
188 (delivery): delivery is RelayDeliveryResult => delivery !== null,
189 ),
190 };
191 }),
192 });
193});
194
195export const layer = Layer.effect(AgentActivityPublisher, make);