Skip to content

Commit 0d4dfa9

Browse files
jayleatonclaude
andcommitted
feat(server): phones register Expo push tokens with each server
Adds pushRegistrations.register/unregister RPCs (orchestration:read), a fork migration for push_registrations keyed by install id and bound to the calling session, and the pushNotifications capability. Rows of revoked or expired sessions are deleted before any delivery. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 6dec7c9 commit 0d4dfa9

13 files changed

Lines changed: 260 additions & 0 deletions

File tree

‎apps/server/src/environment/ServerEnvironment.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -288,6 +288,7 @@ export const make = Effect.gen(function* () {
288288
npmReleaseLookup: true,
289289
voiceTranscription: isSpeechPlatformSupported(hostPlatform, hostArchitecture, hostLibc),
290290
cloudVoiceTranscription: true,
291+
pushNotifications: true,
291292
...(serverSelfUpdate === null ? {} : { serverSelfUpdate }),
292293
...(serverInstallation === null ? {} : { serverInstallation }),
293294
...(serverSelfUpdate === "boot-service" || desktopAppUpdate

‎apps/server/src/observability/RpcInstrumentation.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,8 @@ const RPC_AGGREGATES = {
8383
[WS_METHODS.tailnetDevicesSetLimit]: "tailnetDevices",
8484
[WS_METHODS.tailnetDevicesLeave]: "tailnetDevices",
8585
[WS_METHODS.tailnetDevicesAllowJoin]: "tailnetDevices",
86+
[WS_METHODS.pushRegistrationsRegister]: "pushRegistrations",
87+
[WS_METHODS.pushRegistrationsUnregister]: "pushRegistrations",
8688
[WS_METHODS.subscribeTailnetDevices]: "tailnetDevices",
8789
[WS_METHODS.serverUpdateSettings]: "server",
8890
[WS_METHODS.serverGetAgentSkills]: "server",

‎apps/server/src/persistence/Migrations.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -110,6 +110,7 @@ import ForkMigration020 from "./Migrations/fork/020_RepairStringifiedModelSelect
110110
import ForkMigration021 from "./Migrations/fork/021_UpstreamScheduledTaskProfile.ts";
111111
import ForkMigration022 from "./Migrations/fork/022_NetworkContinuity.ts";
112112
import ForkMigration023 from "./Migrations/fork/023_ScheduledTaskCleanup.ts";
113+
import ForkMigration024 from "./Migrations/fork/024_PushRegistrations.ts";
113114

114115
const migrationEntries = [
115116
[1, "OrchestrationEvents", Migration0001],
@@ -223,6 +224,7 @@ const forkMigrationEntries = [
223224
[21, "UpstreamScheduledTaskProfile", ForkMigration021, 81],
224225
[22, "NetworkContinuity", ForkMigration022, 81],
225226
[23, "ScheduledTaskCleanup", ForkMigration023, 81],
227+
[24, "PushRegistrations", ForkMigration024, 81],
226228
] as const;
227229

228230
export const forkMigrationManifest = forkMigrationEntries.map(
Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
import * as Effect from "effect/Effect";
2+
import * as SqlClient from "effect/sql/SqlClient";
3+
4+
// Phones this server pushes to directly through Expo. A row only receives
5+
// pushes while its session is live, so revoking the device stops them.
6+
export default Effect.gen(function* () {
7+
const sql = yield* SqlClient.SqlClient;
8+
9+
yield* sql`
10+
CREATE TABLE push_registrations (
11+
install_id TEXT PRIMARY KEY,
12+
session_id TEXT NOT NULL,
13+
expo_push_token TEXT NOT NULL UNIQUE,
14+
label TEXT,
15+
updated_at TEXT NOT NULL
16+
)
17+
`;
18+
});
Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
import { assert, it } from "@effect/vitest";
2+
import { AuthSessionId } from "@t3tools/contracts";
3+
import * as Effect from "effect/Effect";
4+
import * as Layer from "effect/Layer";
5+
import * as SqlClient from "effect/sql/SqlClient";
6+
import * as TestClock from "effect/testing/TestClock";
7+
8+
import * as SqlitePersistence from "../persistence/Sqlite.ts";
9+
import * as PushRegistrations from "./PushRegistrations.ts";
10+
11+
const layer = PushRegistrations.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory));
12+
13+
const insertSession = Effect.fn(function* (
14+
sessionId: string,
15+
input: { readonly expiresAt: string; readonly revokedAt?: string },
16+
) {
17+
const sql = yield* SqlClient.SqlClient;
18+
yield* sql`
19+
INSERT INTO auth_sessions (session_id, subject, scopes, method, issued_at, expires_at, revoked_at)
20+
VALUES (${sessionId}, 'phone', '[]', 'bearer-session-token', '2026-01-01T00:00:00.000Z',
21+
${input.expiresAt}, ${input.revokedAt ?? null})
22+
`;
23+
});
24+
25+
const token = (id: string) => `ExponentPushToken[${id}]`;
26+
27+
it.layer(layer)("PushRegistrations", (it) => {
28+
it.effect("keeps one row per install and per token, and drops rows of dead sessions", () =>
29+
Effect.gen(function* () {
30+
yield* TestClock.setTime(Date.parse("2026-11-13T00:00:00.000Z"));
31+
const store = yield* PushRegistrations.PushRegistrations;
32+
const live = AuthSessionId.make("live");
33+
const renewed = AuthSessionId.make("renewed");
34+
yield* insertSession("live", { expiresAt: "2999-01-01T00:00:00.000Z" });
35+
yield* insertSession("renewed", { expiresAt: "2999-01-01T00:00:00.000Z" });
36+
yield* insertSession("revoked", {
37+
expiresAt: "2999-01-01T00:00:00.000Z",
38+
revokedAt: "2026-01-02T00:00:00.000Z",
39+
});
40+
yield* insertSession("expired", { expiresAt: "2000-01-01T00:00:00.000Z" });
41+
42+
// Re-registering the same install replaces its token and session.
43+
yield* store.register(live, { installId: "phone-a", expoPushToken: token("a1") });
44+
yield* store.register(renewed, { installId: "phone-a", expoPushToken: token("a2") });
45+
// A reinstall keeps the token under a new install id.
46+
yield* store.register(live, { installId: "phone-b", expoPushToken: token("b") });
47+
yield* store.register(live, { installId: "phone-b2", expoPushToken: token("b") });
48+
yield* store.register(AuthSessionId.make("revoked"), {
49+
installId: "phone-c",
50+
expoPushToken: token("c"),
51+
});
52+
yield* store.register(AuthSessionId.make("expired"), {
53+
installId: "phone-d",
54+
expoPushToken: token("d"),
55+
});
56+
57+
assert.deepStrictEqual(yield* store.listDeliverable, [
58+
{ installId: "phone-a", expoPushToken: token("a2") },
59+
{ installId: "phone-b2", expoPushToken: token("b") },
60+
]);
61+
const sql = yield* SqlClient.SqlClient;
62+
const remaining = yield* sql<{ id: string }>`
63+
SELECT install_id AS id FROM push_registrations ORDER BY install_id`;
64+
assert.deepStrictEqual(
65+
remaining.map((row) => row.id),
66+
["phone-a", "phone-b2"],
67+
);
68+
69+
yield* store.removeTokens([token("a2")]);
70+
yield* store.unregister("phone-b2");
71+
assert.deepStrictEqual(yield* store.listDeliverable, []);
72+
}),
73+
);
74+
});
Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
import {
2+
PushRegistrationError,
3+
type AuthSessionId,
4+
type PushRegistrationsRegisterInput,
5+
} from "@t3tools/contracts";
6+
import * as Context from "effect/Context";
7+
import * as DateTime from "effect/DateTime";
8+
import * as Effect from "effect/Effect";
9+
import * as Layer from "effect/Layer";
10+
import * as SqlClient from "effect/sql/SqlClient";
11+
12+
export interface PushTarget {
13+
readonly installId: string;
14+
readonly expoPushToken: string;
15+
}
16+
17+
export class PushRegistrations extends Context.Service<
18+
PushRegistrations,
19+
{
20+
readonly register: (
21+
sessionId: AuthSessionId,
22+
input: PushRegistrationsRegisterInput,
23+
) => Effect.Effect<void, PushRegistrationError>;
24+
readonly unregister: (installId: string) => Effect.Effect<void, PushRegistrationError>;
25+
/** Phones whose session is still live. Rows of revoked or expired sessions are deleted. */
26+
readonly listDeliverable: Effect.Effect<ReadonlyArray<PushTarget>, PushRegistrationError>;
27+
/** Expo reported these tokens as no longer registered on their device. */
28+
readonly removeTokens: (
29+
tokens: ReadonlyArray<string>,
30+
) => Effect.Effect<void, PushRegistrationError>;
31+
}
32+
>()("t3/push/PushRegistrations") {}
33+
34+
const toError = (cause: unknown) => new PushRegistrationError({ cause });
35+
36+
const make = Effect.gen(function* () {
37+
const sql = yield* SqlClient.SqlClient;
38+
39+
const register = Effect.fn("PushRegistrations.register")(function* (
40+
sessionId: AuthSessionId,
41+
input: PushRegistrationsRegisterInput,
42+
) {
43+
const now = DateTime.formatIso(yield* DateTime.now);
44+
yield* sql
45+
.withTransaction(
46+
Effect.gen(function* () {
47+
// A reinstall keeps the token but gets a new install id; one phone, one row.
48+
yield* sql`DELETE FROM push_registrations
49+
WHERE expo_push_token = ${input.expoPushToken} AND install_id <> ${input.installId}`;
50+
yield* sql`
51+
INSERT INTO push_registrations (install_id, session_id, expo_push_token, label, updated_at)
52+
VALUES (${input.installId}, ${sessionId}, ${input.expoPushToken}, ${input.label ?? null}, ${now})
53+
ON CONFLICT (install_id) DO UPDATE SET
54+
session_id = excluded.session_id,
55+
expo_push_token = excluded.expo_push_token,
56+
label = excluded.label,
57+
updated_at = excluded.updated_at
58+
`;
59+
}),
60+
)
61+
.pipe(Effect.mapError(toError));
62+
});
63+
64+
const unregister = Effect.fn("PushRegistrations.unregister")(function* (installId: string) {
65+
yield* sql`DELETE FROM push_registrations WHERE install_id = ${installId}`.pipe(
66+
Effect.mapError(toError),
67+
);
68+
});
69+
70+
const listDeliverable = Effect.gen(function* () {
71+
const now = DateTime.formatIso(yield* DateTime.now);
72+
yield* sql`
73+
DELETE FROM push_registrations
74+
WHERE session_id NOT IN (
75+
SELECT session_id FROM auth_sessions WHERE revoked_at IS NULL AND expires_at > ${now}
76+
)
77+
`;
78+
return yield* sql<PushTarget>`
79+
SELECT install_id AS "installId", expo_push_token AS "expoPushToken"
80+
FROM push_registrations
81+
ORDER BY install_id
82+
`;
83+
}).pipe(Effect.mapError(toError), Effect.withSpan("PushRegistrations.listDeliverable"));
84+
85+
const removeTokens = Effect.fn("PushRegistrations.removeTokens")(function* (
86+
tokens: ReadonlyArray<string>,
87+
) {
88+
if (tokens.length === 0) return;
89+
yield* sql`DELETE FROM push_registrations WHERE ${sql.in("expo_push_token", tokens)}`.pipe(
90+
Effect.mapError(toError),
91+
);
92+
});
93+
94+
return PushRegistrations.of({ register, unregister, listDeliverable, removeTokens });
95+
});
96+
97+
export const layer = Layer.effect(PushRegistrations, make);

‎apps/server/src/ws.test.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,7 @@ import * as McpAppRequests from "./mcpApps/McpAppRequests.ts";
7979
import * as ModelManifest from "./provider/ModelManifest.ts";
8080
import * as Orchestrator from "./orchestration-v2/Orchestrator.ts";
8181
import * as OtlpEndpointCheck from "./observability/OtlpEndpointCheck.ts";
82+
import * as PushRegistrations from "./push/PushRegistrations.ts";
8283
import * as PairingGrantStore from "./auth/PairingGrantStore.ts";
8384
import * as RpcAuthorization from "./auth/RpcAuthorization.ts";
8485
import * as PeerLinkRequests from "./peer/PeerLinkRequests.ts";
@@ -817,6 +818,7 @@ const layerUnusedConfigRpcServices = Layer.mergeAll(
817818
Layer.mock(OrchestrationEventStore.OrchestrationEventStore)({}),
818819
Layer.mock(Orchestrator.OrchestratorV2)({}),
819820
Layer.mock(OtlpEndpointCheck.OtlpEndpointCheck)({}),
821+
Layer.mock(PushRegistrations.PushRegistrations)({}),
820822
Layer.mock(PairingGrantStore.PairingGrantStore)({}),
821823
Layer.mock(PeerLinks.PeerLinks)({}),
822824
Layer.mock(TailnetDevices.TailnetDevices)({}),

‎apps/server/src/ws.ts‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import * as Orchestrator from "./orchestration-v2/Orchestrator.ts";
88
import * as PeerLinkRequests from "./peer/PeerLinkRequests.ts";
99
import * as PeerLinks from "./peer/PeerLinks.ts";
1010
import * as TailnetDevices from "./peer/TailnetDevices.ts";
11+
import * as PushRegistrations from "./push/PushRegistrations.ts";
1112

1213
import * as DateTime from "effect/DateTime";
1314
import * as Duration from "effect/Duration";
@@ -1581,6 +1582,7 @@ export const layerWsRpc = (
15811582
const sessions = yield* SessionStore.SessionStore;
15821583
const processDiagnostics = yield* ProcessDiagnostics.ProcessDiagnostics;
15831584
const triage = yield* TriageService.TriageService;
1585+
const pushRegistrations = yield* PushRegistrations.PushRegistrations;
15841586
const hostResources = yield* HostResources.HostResources;
15851587
const processResourceMonitor = yield* ProcessResourceMonitor.ProcessResourceMonitor;
15861588
const resourceTelemetry = yield* ResourceTelemetry.ResourceTelemetry;
@@ -2709,6 +2711,10 @@ export const layerWsRpc = (
27092711
[WS_METHODS.tailnetDevicesLeave]: () =>
27102712
tailnetDevices.leave({ sessionId: currentSession.sessionId }),
27112713
[WS_METHODS.tailnetDevicesAllowJoin]: () => tailnetDevices.allowJoin,
2714+
[WS_METHODS.pushRegistrationsRegister]: (input) =>
2715+
pushRegistrations.register(currentSessionId, input),
2716+
[WS_METHODS.pushRegistrationsUnregister]: (input) =>
2717+
pushRegistrations.unregister(input.installId),
27122718
[WS_METHODS.tailnetDevicesOpenDevice]: (input) =>
27132719
tailnetDevices.openDevice(input.environmentId, {
27142720
scopes: currentSession.scopes,
@@ -3581,6 +3587,7 @@ const provideWsRpcDependencies = <A, E, R>(
35813587
// mutation invalidates the HTTP diff cache that every client reads from.
35823588
Layer.provide(Layer.succeed(PullRequestService.PullRequestService, pullRequests)),
35833589
Layer.provide(OtlpEndpointCheck.layer),
3590+
Layer.provide(PushRegistrations.layer),
35843591
Layer.provide(
35853592
Layer.succeed(PullRequestReviewGuideService.PullRequestReviewGuideService, reviewGuides),
35863593
),

‎packages/contracts/src/clientRpcPermissions.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,9 @@ export const CLIENT_GUARDED_RPC_SCOPES = {
9191
[WS_METHODS.serverUpdateAgentGateway]: AuthAccessWriteScope,
9292
[WS_METHODS.serverRevealAgentGatewayToken]: AuthAccessWriteScope,
9393
[WS_METHODS.serverRotateAgentGatewayToken]: AuthAccessWriteScope,
94+
// A push carries only thread titles and states that a reader already sees.
95+
[WS_METHODS.pushRegistrationsRegister]: AuthOrchestrationReadScope,
96+
[WS_METHODS.pushRegistrationsUnregister]: AuthOrchestrationReadScope,
9497
// A session on another member, never past the caller's own grant; the server decides who may.
9598
[WS_METHODS.tailnetDevicesOpenDevice]: AuthOrchestrationReadScope,
9699
} as const;

‎packages/contracts/src/environment.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -204,6 +204,8 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({
204204
freshContext: Schema.optionalKey(Schema.Boolean),
205205
/** Server saves and runs scheduled task `threadMode`; older servers ignore it. */
206206
scheduledTaskThreadModes: Schema.optionalKey(Schema.Boolean),
207+
/** Server sends Expo pushes to phones that call pushRegistrations.register. */
208+
pushNotifications: Schema.optionalKey(Schema.Boolean),
207209
/** Server can transcribe 16 kHz mono PCM audio on the environment machine. */
208210
voiceTranscription: Schema.optionalKey(Schema.Boolean),
209211
cloudVoiceTranscription: Schema.optionalKey(Schema.Boolean),

0 commit comments

Comments
 (0)