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 NodeSocket from "@effect/platform-node/NodeSocket";
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});
45const readFile = (path: string) => Effect.tryPromise(() => NodeFSP.readFile(path, "utf8"));
46const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
47class WatchUnregisteredDeviceError extends Schema.TaggedError<WatchUnregisteredDeviceError>()(
48 "WatchUnregisteredDeviceError",
49 {},
50) {
51 override get message() {
52 return "Device token is no longer registered";
53 }
54}
55
56class WatchStoppedError extends Schema.TaggedError<WatchStoppedError>()("WatchStoppedError", {
57 cause: Schema.Defect(),
58}) {
59 override get message() {
60 return "Android push watcher stopped. Check the private connection and Firebase configuration.";
61 }
62}
63
64const isWatchUnregisteredDeviceError = Schema.is(WatchUnregisteredDeviceError);
65
66const preferences = {
67 notificationsEnabled: true,
68 liveActivitiesEnabled: true,
69 notifyOnApproval: true,
70 notifyOnInput: true,
71 notifyOnCompletion: true,
72 notifyOnFailure: true,
73};
74
75const main = Effect.gen(function* () {
76 const [credentialsPath, devicePath, connectionPath] = process.argv.slice(2);
77 if (!credentialsPath || !devicePath || !connectionPath) {
78 return yield* Effect.logError(
79 "Usage: node scripts/android-push-watch.ts <service-account.json> <device.json> <connection.json>",
80 );
81 }
82 const credentials = yield* readFile(credentialsPath);
83 const device = yield* readFile(devicePath).pipe(
84 Effect.flatMap(Schema.decodeUnknownEffect(Schema.fromJsonString(Device))),
85 );
86 const connection = yield* readFile(connectionPath).pipe(
87 Effect.flatMap(Schema.decodeUnknownEffect(Schema.fromJsonString(Connection))),
88 );
89 const layerSocketConstructor = Layer.succeed(
90 Socket.WebSocketConstructor,
91 // Socket.makeWebSocket only ever passes its `protocols` option here.
92 (url, protocols) =>
93 new NodeSocket.NodeWS.WebSocket(url, protocols as string | string[] | undefined, {
94 headers: { authorization: `Bearer ${connection.bearerToken}` },
95 }) as unknown as globalThis.WebSocket,
96 );
97 const layerProtocol = RpcClient.layerProtocolSocket().pipe(
98 Layer.provide(
99 Socket.layerWebSocket(connection.wsUrl).pipe(Layer.provide(layerSocketConstructor)),
100 ),
101 Layer.provide(RpcSerialization.layerJson),
102 );
103 const layerFcm = FcmClient.layer.pipe(
104 Layer.provide(
105 FcmAssertionSigner.layer.pipe(
106 Layer.provide(Layer.succeed(WebCrypto.WebCrypto, { subtle: globalThis.crypto.subtle })),
107 ),
108 ),
109 Layer.provide(
110 Layer.mergeAll(
111 FetchHttpClient.layer,
112 Layer.succeed(RelayConfiguration.RelayConfiguration, {
113 relayIssuer: "http://localhost",
114 apns: null,
115 fcmServiceAccount: Redacted.make(credentials),
116 clerkSecretKey: Redacted.make(""),
117 clerkPublishableKey: "",
118 clerkJwtAudience: "",
119 apnsDeliveryJobSigningSecret: Redacted.make(""),
120 cloudMintPrivateKey: Redacted.make(""),
121 cloudMintPublicKey: "",
122 managedEndpointBaseDomain: undefined,
123 managedEndpointNamespace: undefined,
124 }),
125 ),
126 ),
127 );
128 yield* Effect.gen(function* () {
129 const rpc = yield* RpcClient.make(WsRpcGroup);
130 const sender = yield* FcmClient.FcmClient;
131 const config = yield* rpc[WS_METHODS.serverGetConfig]({});
132 const projects = new Map<string, OrchestrationProjectShell>();
133 const threads = new Map<string, OrchestrationV2ThreadShell>();
134 let states = new Map<string, RelayAgentActivityState>();
135 let previouslyActive = false;
136 yield* Effect.logInfo("Watching this paired environment for Android push verification.");
137 yield* rpc[ORCHESTRATION_V2_WS_METHODS.subscribeShell]({}).pipe(
138 Stream.runForEach(
139 Effect.fnUntraced(function* (item) {
140 switch (item.kind) {
141 case "synchronized":
142 return;
143 case "snapshot":
144 projects.clear();
145 threads.clear();
146 for (const project of item.snapshot.projects) projects.set(project.id, project);
147 for (const thread of item.snapshot.threads) threads.set(thread.id, thread);
148 break;
149 case "project.updated":
150 projects.set(item.project.id, item.project);
151 break;
152 case "project.removed":
153 projects.delete(item.projectId);
154 break;
155 case "thread.updated":
156 threads.set(item.thread.id, item.thread);
157 break;
158 case "thread.removed":
159 threads.delete(item.threadId);
160 break;
161 }
162 const next = new Map<string, RelayAgentActivityState>();
163 for (const thread of threads.values()) {
164 const project = projects.get(thread.projectId);
165 if (!project || thread.archivedAt) continue;
166 const state = projectThreadAwarenessV2({
167 environmentId: config.environment.environmentId,
168 project,
169 thread,
170 });
171 if (state) next.set(thread.id, state);
172 }
173 const state = item.kind === "thread.updated" ? next.get(item.thread.id) : undefined;
174 const previous = state ? states.get(state.threadId) : undefined;
175 // A fresh subscription restores ongoing work without announcing old completions.
176 const now = yield* Clock.currentTimeMillis;
177 const alert =
178 state && state.phase !== previous?.phase && item.kind !== "snapshot"
179 ? FcmDeliveries.androidAlertForState(state, preferences, now)
180 : null;
181 const aggregate = makeAggregateState({
182 activeStates: [...next.values()],
183 terminalState: null,
184 nowMs: now,
185 });
186 const active = (aggregate?.activeCount ?? 0) > 0;
187 const same = encodeJson([...next.values()]) === encodeJson([...states.values()]);
188 states = next;
189 if ((!active && !previouslyActive && !alert) || (same && !alert)) return;
190 const result = yield* sender.send({
191 token: device.token,
192 packageName: device.packageName,
193 alert: alert !== null,
194 data: fitFcmData({
195 t3_kind: "agent_activity",
196 device_id: device.deviceId,
197 user_id: device.userId,
198 updated_at: String(now),
199 ...androidActivityData(aggregate),
200 ...alert,
201 }),
202 });
203 if (result.unregistered) return yield* new WatchUnregisteredDeviceError({});
204 previouslyActive = active;
205 yield* Effect.logInfo(
206 `Android push accepted: ${state?.phase ?? (active ? "active" : "ended")}`,
207 );
208 }),
209 ),
210 );
211 }).pipe(Effect.provide(Layer.mergeAll(layerProtocol, layerFcm)));
212});
213
214NodeRuntime.runMain(
215 main.pipe(
216 Effect.scoped,
217 Effect.catchCause((cause) => {
218 const failure = Cause.findErrorOption(cause);
219 return Effect.fail(
220 Option.isSome(failure) && isWatchUnregisteredDeviceError(failure.value)
221 ? failure.value
222 : new WatchStoppedError({ cause }),
223 );
224 }),
225 // Socket failures may contain credential-bearing request headers. Keep the
226 // cause on the error, but print only the fixed message at the CLI boundary.
227 Effect.tapError((error) => Effect.logError(error.message)),
228 ),
229 { disableErrorReporting: true },
230);