1// @effect-diagnostics nodeBuiltinImport:off - Local developer verification reads private credential files.
2import * as NodeFSP from "node:fs/promises";
3import * as NodeRuntime from "@effect/platform-node/NodeRuntime";
4import * as NodeWS from "@effect/platform-node/NodeWS";
5import {
6 ORCHESTRATION_V2_WS_METHODS,
7 WS_METHODS,
8 WsRpcGroup,
9 type OrchestrationProjectShell,
10 type OrchestrationV2ThreadShell,
11} from "@t3tools/contracts";
12import type { RelayAgentActivityState } from "@t3tools/contracts/relay";
13import { projectThreadAwarenessV2 } from "@t3tools/shared/agentAwareness";
14import * as Cause from "effect/Cause";
15import * as Option from "effect/Option";
16import * as Clock from "effect/Clock";
17import * as Effect from "effect/Effect";
18import * as Layer from "effect/Layer";
19import * as Redacted from "effect/Redacted";
20import * as Schema from "effect/Schema";
21import * as Stream from "effect/Stream";
22import * as FetchHttpClient from "effect/http/FetchHttpClient";
23import * as RpcClient from "effect/rpc/RpcClient";
24import * as RpcSerialization from "effect/rpc/RpcSerialization";
25import * as Socket from "effect/socket/Socket";
26
27import * as RelayConfiguration from "../src/Config.ts";
28import { androidActivityData, fitFcmData } from "../src/agentActivity/fcmPayloads.ts";
29import * as WebCrypto from "../src/WebCrypto.ts";
30import * as FcmAssertionSigner from "../src/agentActivity/FcmAssertionSigner.ts";
31import * as FcmClient from "../src/agentActivity/FcmClient.ts";
32import * as FcmDeliveries from "../src/agentActivity/FcmDeliveries.ts";
33import { makeAggregateState } from "../src/agentActivity/agentActivityAggregate.ts";
34
35const Device = Schema.Struct({
36 token: Schema.NonEmptyString,
37 deviceId: Schema.NonEmptyString,
38 userId: Schema.NonEmptyString,
39 packageName: Schema.NonEmptyString,
40});
41const Connection = Schema.Struct({
42 wsUrl: Schema.NonEmptyString,
43 bearerToken: Schema.NonEmptyString,
44});
45class WatchReadFileError extends Schema.TaggedError<WatchReadFileError>()("WatchReadFileError", {
46 path: Schema.String,
47 cause: Schema.Defect(),
48}) {
49 override get message() {
50 return `Could not read ${this.path}`;
51 }
52}
53const readFile = (path: string) =>
54 Effect.tryPromise({
55 try: () => NodeFSP.readFile(path, "utf8"),
56 catch: (cause) => new WatchReadFileError({ path, cause }),
57 });
58const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
59class WatchUnregisteredDeviceError extends Schema.TaggedError<WatchUnregisteredDeviceError>()(
60 "WatchUnregisteredDeviceError",
61 {},
62) {
63 override get message() {
64 return "Device token is no longer registered";
65 }
66}
67
68class WatchStoppedError extends Schema.TaggedError<WatchStoppedError>()("WatchStoppedError", {
69 cause: Schema.Defect(),
70}) {
71 override get message() {
72 return "Android push watcher stopped. Check the private connection and Firebase configuration.";
73 }
74}
75
76const isWatchUnregisteredDeviceError = Schema.is(WatchUnregisteredDeviceError);
77
78const preferences = {
79 notificationsEnabled: true,
80 liveActivitiesEnabled: true,
81 notifyOnApproval: true,
82 notifyOnInput: true,
83 notifyOnCompletion: true,
84 notifyOnFailure: true,
85};
86
87const main = Effect.gen(function* () {
88 const [credentialsPath, devicePath, connectionPath] = process.argv.slice(2);
89 if (!credentialsPath || !devicePath || !connectionPath) {
90 return yield* Effect.logError(
91 "Usage: node scripts/android-push-watch.ts <service-account.json> <device.json> <connection.json>",
92 );
93 }
94 const credentials = yield* readFile(credentialsPath);
95 const device = yield* readFile(devicePath).pipe(
96 Effect.flatMap(Schema.decodeUnknownEffect(Schema.fromJsonString(Device))),
97 );
98 const connection = yield* readFile(connectionPath).pipe(
99 Effect.flatMap(Schema.decodeUnknownEffect(Schema.fromJsonString(Connection))),
100 );
101 const layerSocketConstructor = Layer.succeed(
102 Socket.WebSocketConstructor,
103 (url, protocols) =>
104 new NodeWS.WebSocket(url, protocols as string | string[] | undefined, {
105 headers: { authorization: `Bearer ${connection.bearerToken}` },
106 }) as unknown as globalThis.WebSocket,
107 );
108 const layerProtocol = RpcClient.layerProtocolSocket().pipe(
109 Layer.provide(
110 Socket.layerWebSocket(connection.wsUrl).pipe(Layer.provide(layerSocketConstructor)),
111 ),
112 Layer.provide(RpcSerialization.layerJson),
113 );
114 const layerFcm = FcmClient.layer.pipe(
115 Layer.provide(
116 FcmAssertionSigner.layer.pipe(
117 Layer.provide(Layer.succeed(WebCrypto.WebCrypto, { subtle: globalThis.crypto.subtle })),
118 ),
119 ),
120 Layer.provide(
121 Layer.mergeAll(
122 FetchHttpClient.layer,
123 Layer.succeed(RelayConfiguration.RelayConfiguration, {
124 relayIssuer: "http://localhost",
125 apns: null,
126 fcmServiceAccount: Redacted.make(credentials),
127 clerkSecretKey: Redacted.make(""),
128 clerkPublishableKey: "",
129 clerkJwtAudience: "",
130 apnsDeliveryJobSigningSecret: Redacted.make(""),
131 cloudMintPrivateKey: Redacted.make(""),
132 cloudMintPublicKey: "",
133 managedEndpointBaseDomain: undefined,
134 managedEndpointNamespace: undefined,
135 }),
136 ),
137 ),
138 );
139 yield* Effect.gen(function* () {
140 const rpc = yield* RpcClient.make(WsRpcGroup);
141 const sender = yield* FcmClient.FcmClient;
142 const config = yield* rpc[WS_METHODS.serverGetConfig]({});
143 const projects = new Map<string, OrchestrationProjectShell>();
144 const threads = new Map<string, OrchestrationV2ThreadShell>();
145 let states = new Map<string, RelayAgentActivityState>();
146 let previouslyActive = false;
147 yield* Effect.logInfo("Watching this paired environment for Android push verification.");
148 yield* rpc[ORCHESTRATION_V2_WS_METHODS.subscribeShell]({}).pipe(
149 Stream.runForEach(
150 Effect.fnUntraced(function* (item) {
151 switch (item.kind) {
152 case "synchronized":
153 return;
154 case "snapshot":
155 projects.clear();
156 threads.clear();
157 for (const project of item.snapshot.projects) projects.set(project.id, project);
158 for (const thread of item.snapshot.threads) threads.set(thread.id, thread);
159 break;
160 case "project.updated":
161 projects.set(item.project.id, item.project);
162 break;
163 case "project.removed":
164 projects.delete(item.projectId);
165 break;
166 case "thread.updated":
167 threads.set(item.thread.id, item.thread);
168 break;
169 case "thread.removed":
170 threads.delete(item.threadId);
171 break;
172 }
173 const next = new Map<string, RelayAgentActivityState>();
174 for (const thread of threads.values()) {
175 const project = projects.get(thread.projectId);
176 if (!project || thread.archivedAt) continue;
177 const state = projectThreadAwarenessV2({
178 environmentId: config.environment.environmentId,
179 project,
180 thread,
181 });
182 if (state) next.set(thread.id, state);
183 }
184 const state = item.kind === "thread.updated" ? next.get(item.thread.id) : undefined;
185 const previous = state ? states.get(state.threadId) : undefined;
186 const now = yield* Clock.currentTimeMillis;
187 const alert =
188 state && state.phase !== previous?.phase && item.kind !== "snapshot"
189 ? FcmDeliveries.androidAlertForState(state, preferences, now)
190 : null;
191 const aggregate = makeAggregateState({
192 activeStates: [...next.values()],
193 terminalState: null,
194 nowMs: now,
195 });
196 const active = (aggregate?.activeCount ?? 0) > 0;
197 const same = encodeJson([...next.values()]) === encodeJson([...states.values()]);
198 states = next;
199 if ((!active && !previouslyActive && !alert) || (same && !alert)) return;
200 const result = yield* sender.send({
201 token: device.token,
202 packageName: device.packageName,
203 alert: alert !== null,
204 data: fitFcmData({
205 t3_kind: "agent_activity",
206 device_id: device.deviceId,
207 user_id: device.userId,
208 updated_at: String(now),
209 ...androidActivityData(aggregate),
210 ...alert,
211 }),
212 });
213 if (result.unregistered) return yield* new WatchUnregisteredDeviceError({});
214 previouslyActive = active;
215 yield* Effect.logInfo(
216 `Android push accepted: ${state?.phase ?? (active ? "active" : "ended")}`,
217 );
218 }),
219 ),
220 );
221 }).pipe(Effect.provide(Layer.mergeAll(layerProtocol, layerFcm)));
222});
223
224NodeRuntime.runMain(
225 main.pipe(
226 Effect.scoped,
227 Effect.catchCause((cause) => {
228 const failure = Cause.findErrorOption(cause);
229 return Effect.fail(
230 Option.isSome(failure) && isWatchUnregisteredDeviceError(failure.value)
231 ? failure.value
232 : new WatchStoppedError({ cause }),
233 );
234 }),
235 // Socket errors can expose credential headers; retain the cause but print only the fixed message.
236 Effect.tapError((error) => Effect.logError(error.message)),
237 ),
238 { disableErrorReporting: true },
239);