diff --git a/.env.local.example b/.env.local.example index a150dda42..3e12c3f2e 100644 --- a/.env.local.example +++ b/.env.local.example @@ -65,6 +65,13 @@ INTEGRATION_ENCRYPTION_KEY_VERSION=1 # T3CODE_BOT_TOKEN= # AI_MODEL= +# iOS push notifications: an APNs auth key (Apple Developer > Keys, with APNs enabled). The key is +# the .p8 file's PEM, or its base64 body. Local workerd cannot reach APNs (no HTTP/2), so pushes +# fail there; they work on deployed stages. +# APNS_KEY_ID= +# APNS_TEAM_ID= +# APNS_PRIVATE_KEY= + # Maple's OAuth tokens for the public API v1 # CLERK_SECRET_KEY= diff --git a/apps/backend/src/rpc/handlers/push-devices.ts b/apps/backend/src/rpc/handlers/push-devices.ts new file mode 100644 index 000000000..e95874894 --- /dev/null +++ b/apps/backend/src/rpc/handlers/push-devices.ts @@ -0,0 +1,60 @@ +/** + * Runs in the caller's `UserObject`, which holds their push devices (`worker/objects/user/push.ts`). + * The api Worker routes every call to the caller's own object; the handlers refuse any other. + * Registering also claims the token in the directory, which takes it from a previous owner + * (`worker/objects/user/push-ownership.ts`). + */ +import { Database } from "@hazel/db" +import { CurrentUser, ErrorUtils, policy, withRemapDbErrors } from "@hazel/domain" +import { PushDeviceRpcs } from "@hazel/domain/rpc" +import { Effect } from "effect" +import { currentObject } from "../../lib/current-object" +import { registerDevice, unregisterDevice } from "../../worker/objects/user/push" +import { claimToken, releaseToken } from "../../worker/objects/user/push-ownership" + +/** The actor owns the `UserObject` the call runs in. */ +const ownObject = (action: string) => + ErrorUtils.refailUnauthorized( + "PushDevice", + action, + )( + policy("PushDevice", action, (actor) => + Effect.map( + currentObject, + (object) => + object._tag === "Some" && object.value.kind === "user" && object.value.id === actor.id, + ), + ), + ) + +export const PushDeviceRpcLive = PushDeviceRpcs.toLayer( + Effect.gen(function* () { + const db = yield* Database.Database + + return { + "pushDevice.register": (payload) => + db + .transaction( + Effect.gen(function* () { + yield* ownObject("register") + const user = yield* CurrentUser.Context + const registeredAt = yield* registerDevice(payload) + yield* claimToken(user.id, payload.token, registeredAt) + }), + ) + .pipe(withRemapDbErrors("PushDevice", "create")), + + "pushDevice.unregister": ({ token }) => + db + .transaction( + Effect.gen(function* () { + yield* ownObject("unregister") + const user = yield* CurrentUser.Context + yield* unregisterDevice(token) + yield* releaseToken(user.id, token) + }), + ) + .pipe(withRemapDbErrors("PushDevice", "delete")), + } + }), +) diff --git a/apps/backend/src/rpc/routing.ts b/apps/backend/src/rpc/routing.ts index a97239602..e9257b469 100644 --- a/apps/backend/src/rpc/routing.ts +++ b/apps/backend/src/rpc/routing.ts @@ -217,6 +217,8 @@ export const routes: RouteTable = { "userPresenceStatus.update": user, "userPresenceStatus.heartbeat": user, "userPresenceStatus.clearStatus": user, + "pushDevice.register": user, + "pushDevice.unregister": user, // OrgObject // The org's search index, filtered by the caller's `channel_access` there. diff --git a/apps/backend/src/rpc/server.ts b/apps/backend/src/rpc/server.ts index 86bf0d302..e0fc2f15a 100644 --- a/apps/backend/src/rpc/server.ts +++ b/apps/backend/src/rpc/server.ts @@ -21,6 +21,7 @@ import { OrganizationMemberRpcs, OrganizationRpcs, PinnedMessageRpcs, + PushDeviceRpcs, TypingIndicatorRpcs, UserPresenceStatusRpcs, UserRpcs, @@ -47,6 +48,7 @@ import { NotificationRpcLive } from "./handlers/notifications" import { OrganizationMemberRpcLive } from "./handlers/organization-members" import { OrganizationRpcLive } from "./handlers/organizations" import { PinnedMessageRpcLive } from "./handlers/pinned-messages" +import { PushDeviceRpcLive } from "./handlers/push-devices" import { TypingIndicatorRpcLive } from "./handlers/typing-indicators" import { UserPresenceStatusRpcLive } from "./handlers/user-presence-status" import { UserRpcLive } from "./handlers/users" @@ -77,6 +79,7 @@ const BaseRpcs = MessageRpcs.merge( OrganizationMemberRpcs, UserRpcs, UserPresenceStatusRpcs, + PushDeviceRpcs, ).merge( ChannelRpcs, ChannelMemberRpcs, @@ -110,6 +113,7 @@ const rpcGroups = [ ["OrganizationMemberRpcs", OrganizationMemberRpcs], ["UserRpcs", UserRpcs], ["UserPresenceStatusRpcs", UserPresenceStatusRpcs], + ["PushDeviceRpcs", PushDeviceRpcs], ["ChannelRpcs", ChannelRpcs], ["ChannelMemberRpcs", ChannelMemberRpcs], ["ChannelSectionRpcs", ChannelSectionRpcs], @@ -152,6 +156,7 @@ export const RpcHandlersLive = Layer.empty Layer.provideMerge(OrganizationMemberRpcLive), Layer.provideMerge(UserRpcLive), Layer.provideMerge(UserPresenceStatusRpcLive), + Layer.provideMerge(PushDeviceRpcLive), ) .pipe( Layer.provideMerge(ChannelRpcLive), diff --git a/apps/backend/src/services/apns.test.ts b/apps/backend/src/services/apns.test.ts new file mode 100644 index 000000000..0f47f96a8 --- /dev/null +++ b/apps/backend/src/services/apns.test.ts @@ -0,0 +1,212 @@ +import { ConfigProvider, Effect, Layer } from "effect" +import { HttpClient, HttpClientError, HttpClientResponse } from "effect/http" +import type { HttpClientRequest } from "effect/http" +import { describe, expect, it } from "vitest" +import { ApnsClient, type ApnsNotification, importApnsKey, pkcs8Of, signProviderToken } from "./apns" + +const decoder = new TextDecoder() + +const fromBase64Url = (value: string) => + Uint8Array.from(atob(value.replace(/-/g, "+").replace(/_/g, "/")), (char) => char.charCodeAt(0)) + +const json = (part: string) => JSON.parse(decoder.decode(fromBase64Url(part))) + +const generateKey = async () => { + const pair = (await crypto.subtle.generateKey({ name: "ECDSA", namedCurve: "P-256" }, true, [ + "sign", + "verify", + ])) as CryptoKeyPair + const pkcs8 = new Uint8Array(await crypto.subtle.exportKey("pkcs8", pair.privateKey)) + const base64 = btoa(String.fromCharCode(...pkcs8)) + const pem = `-----BEGIN PRIVATE KEY-----\n${base64.match(/.{1,64}/g)!.join("\n")}\n-----END PRIVATE KEY-----` + return { pair, base64, pem } +} + +const notification: ApnsNotification = { + environment: "development", + token: "ab".repeat(32), + topic: "app.hazel.ios", + expiration: 1_900_000_000, + collapseId: "n-1", + payload: { + aps: { + alert: { title: "Ada · #general", body: "hi" }, + sound: "default", + badge: 2, + "thread-id": "c-1", + }, + hazel: { path: "/acme/chat/c-1", channelId: "c-1", notificationId: "n-1" }, + }, +} + +describe("APNs provider token", () => { + it("signs an ES256 JWT the public key verifies", async () => { + const { pair, pem } = await generateKey() + const key = await importApnsKey(pem) + const token = await signProviderToken({ + key, + keyId: "KEY123", + teamId: "TEAM456", + nowMs: 1_800_000_000_500, + }) + const [header, claims, signature] = token.split(".") + expect(json(header!)).toEqual({ alg: "ES256", kid: "KEY123" }) + expect(json(claims!)).toEqual({ iss: "TEAM456", iat: 1_800_000_000 }) + const raw = fromBase64Url(signature!) + // JWS ES256: the raw r || s, 64 bytes. + expect(raw.length).toBe(64) + const valid = await crypto.subtle.verify( + { name: "ECDSA", hash: "SHA-256" }, + pair.publicKey, + raw, + new TextEncoder().encode(`${header}.${claims}`), + ) + expect(valid).toBe(true) + }) + + it("reads the key as PEM, PEM with escaped newlines, or bare base64", async () => { + const { base64, pem } = await generateKey() + const bytes = pkcs8Of(base64) + expect(pkcs8Of(pem)).toEqual(bytes) + expect(pkcs8Of(pem.replace(/\n/g, "\\n"))).toEqual(bytes) + }) +}) + +describe("ApnsClient", () => { + const clientWith = async (answer: (request: HttpClientRequest.HttpClientRequest) => Response) => { + const { pem } = await generateKey() + const requests: Array = [] + const http = Layer.succeed( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.sync(() => { + requests.push(request) + return HttpClientResponse.fromWeb(request, answer(request)) + }), + ), + ) + const config = ConfigProvider.layer( + ConfigProvider.fromUnknown({ + APNS_KEY_ID: "KEY123", + APNS_TEAM_ID: `TEAM-${crypto.randomUUID()}`, + APNS_PRIVATE_KEY: pem, + }), + ) + const layer = ApnsClient.layer.pipe(Layer.provide(Layer.mergeAll(http, config))) + return { layer, requests } + } + + it("posts to the environment's host with the APNs headers, reusing one provider token", async () => { + const { layer, requests } = await clientWith(() => new Response(null, { status: 200 })) + const results = await Effect.runPromise( + Effect.gen(function* () { + const apns = yield* ApnsClient + expect(apns.enabled).toBe(true) + const first = yield* apns.send(notification) + const second = yield* apns.send({ ...notification, environment: "production" }) + return [first, second] + }).pipe(Effect.provide(layer)), + ) + expect(results).toEqual([ + { status: 200, reason: null }, + { status: 200, reason: null }, + ]) + expect(requests.map((request) => request.url)).toEqual([ + `https://api.sandbox.push.apple.com/3/device/${notification.token}`, + `https://api.push.apple.com/3/device/${notification.token}`, + ]) + const headers = requests[0]!.headers + expect(headers["apns-topic"]).toBe("app.hazel.ios") + expect(headers["apns-push-type"]).toBe("alert") + expect(headers["apns-priority"]).toBe("10") + expect(headers["apns-expiration"]).toBe("1900000000") + expect(headers["apns-collapse-id"]).toBe("n-1") + // No trace propagation to Apple. + expect(headers.traceparent).toBeUndefined() + expect(headers.b3).toBeUndefined() + expect(headers.authorization).toMatch(/^bearer [\w-]+\.[\w-]+\.[\w-]+$/) + expect(requests[1]!.headers.authorization).toBe(headers.authorization) + const body = requests[0]!.body + expect(body._tag).toBe("Uint8Array") + if (body._tag === "Uint8Array") { + expect(JSON.parse(decoder.decode(body.body))).toEqual(notification.payload) + } + }) + + it("reads the reason of a refusal", async () => { + const { layer } = await clientWith( + () => new Response(JSON.stringify({ reason: "Unregistered", timestamp: 1 }), { status: 410 }), + ) + const result = await Effect.runPromise( + ApnsClient.use((apns) => apns.send(notification)).pipe(Effect.provide(layer)), + ) + expect(result).toEqual({ status: 410, reason: "Unregistered" }) + }) + + it("re-signs the provider token once APNs calls it expired", async () => { + let answers = [403, 200] + const { layer, requests } = await clientWith(() => { + const status = answers[0]! + answers = answers.slice(1) + return status === 403 + ? new Response(JSON.stringify({ reason: "ExpiredProviderToken" }), { status }) + : new Response(null, { status }) + }) + const results = await Effect.runPromise( + ApnsClient.use((apns) => Effect.all([apns.send(notification), apns.send(notification)])).pipe( + Effect.provide(layer), + ), + ) + expect(results).toEqual([ + { status: 403, reason: "ExpiredProviderToken" }, + { status: 200, reason: null }, + ]) + expect(requests[1]!.headers.authorization).not.toBe(requests[0]!.headers.authorization) + }) + + it("fails a request that never reached APNs without naming the device", async () => { + const { pem } = await generateKey() + const http = Layer.succeed( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.fail( + new HttpClientError.HttpClientError({ + reason: new HttpClientError.TransportError({ request, description: "no HTTP/2" }), + }), + ), + ), + ) + const config = ConfigProvider.layer( + ConfigProvider.fromUnknown({ APNS_KEY_ID: "K", APNS_TEAM_ID: "T", APNS_PRIVATE_KEY: pem }), + ) + const error = await Effect.runPromise( + ApnsClient.use((apns) => Effect.flip(apns.send(notification))).pipe( + Effect.provide(ApnsClient.layer.pipe(Layer.provide(Layer.mergeAll(http, config)))), + ), + ) + expect(error._tag).toBe("ApnsRequestError") + expect(error.message).toBe("APNs request failed (TransportError)") + expect(JSON.stringify(error)).not.toContain(notification.token) + }) + + it("is off without config", async () => { + const enabled = await Effect.runPromise( + ApnsClient.use((apns) => Effect.succeed(apns.enabled)).pipe( + Effect.provide( + ApnsClient.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed( + HttpClient.HttpClient, + HttpClient.make(() => Effect.die("no request expected")), + ), + ConfigProvider.layer(ConfigProvider.fromUnknown({})), + ), + ), + ), + ), + ), + ) + expect(enabled).toBe(false) + }) +}) diff --git a/apps/backend/src/services/apns.ts b/apps/backend/src/services/apns.ts new file mode 100644 index 000000000..c4f9d4d4b --- /dev/null +++ b/apps/backend/src/services/apns.ts @@ -0,0 +1,273 @@ +/** + * Apple Push Notification service: one HTTP/2 request per device (`POST /3/device/`), + * authenticated with a provider token, an ES256 JWT signed with the team's `.p8` key. Called by + * the `UserObject` from its outbox (`worker/objects/user/push.ts`). + * + * Config (all optional; any one unset disables push, logged once per isolate): + * - `APNS_KEY_ID`: the key's id (the JWT's `kid`). + * - `APNS_TEAM_ID`: the Apple developer team (the JWT's `iss`). + * - `APNS_PRIVATE_KEY`: the `.p8` key, as PEM or as its base64 PKCS#8 body. + * + * APNs refuses a provider token refreshed more than once per 20 minutes and one older than an + * hour, so each isolate signs one per key and reuses it for `PROVIDER_TOKEN_TTL_MS`. + * + * Local workerd has no HTTP/2 to APNs: there the request fails as `ApnsRequestError`. + */ +import { Clock, Config, Context, Effect, Layer, Option, Redacted, Schema } from "effect" +import { HttpClient, HttpClientRequest } from "effect/http" + +export type ApnsEnvironment = "development" | "production" + +export const APNS_HOSTS = { + production: "https://api.push.apple.com", + development: "https://api.sandbox.push.apple.com", +} satisfies Record + +/** One request, answer read: well under the outbox's 10 s delivery timeout. */ +export const APNS_REQUEST_TIMEOUT_MS = 8_000 + +/** How long a signed provider token is reused. */ +export const PROVIDER_TOKEN_TTL_MS = 50 * 60 * 1000 + +/** The JSON body APNs delivers to the app. */ +export interface ApnsPayload { + readonly aps: { + readonly alert: { readonly title: string; readonly body: string } + readonly sound: "default" + readonly badge: number + readonly "thread-id": string + } + readonly hazel: { + readonly path?: string + readonly channelId: string + readonly notificationId: string + } +} + +export interface ApnsNotification { + readonly environment: ApnsEnvironment + readonly token: string + /** The app's bundle id. */ + readonly topic: string + /** Epoch seconds after which APNs stops trying. */ + readonly expiration: number + /** Copies with one collapse id show as one on the phone (a retried send). */ + readonly collapseId: string + readonly payload: ApnsPayload +} + +/** APNs's answer: the status and, on a failure, its `reason`. */ +export interface ApnsResult { + readonly status: number + readonly reason: string | null +} + +/** + * APNs was not reached (network, no HTTP/2, a timeout). No cause: the HTTP client's errors carry + * the URL, which names the device token, and this ends up in logs and the outbox's `last_error`. + */ +export class ApnsRequestError extends Schema.TaggedError()("ApnsRequestError", { + message: Schema.String, +}) {} + +/** APNs answered with a failure worth retrying (429, 5xx, a stale provider token). */ +export class ApnsRetryableError extends Schema.TaggedError()("ApnsRetryableError", { + status: Schema.Number, + reason: Schema.NullOr(Schema.String), +}) {} + +export interface ApnsSender { + /** False when the `APNS_*` config is missing or the key does not import. */ + readonly enabled: boolean + /** One push. On `ExpiredProviderToken` the next send signs a new provider token. */ + readonly send: (notification: ApnsNotification) => Effect.Effect +} + +// PROVIDER TOKEN + +const encoder = new TextEncoder() + +const base64Url = (bytes: Uint8Array | ArrayBuffer) => + btoa(String.fromCharCode(...new Uint8Array(bytes))) + .replace(/\+/g, "-") + .replace(/\//g, "_") + .replace(/=+$/, "") + +const base64UrlJson = (value: unknown) => base64Url(encoder.encode(JSON.stringify(value))) + +/** The `.p8` key's PKCS#8 bytes, from PEM (escaped newlines too) or the bare base64 body. */ +export const pkcs8Of = (key: string): Uint8Array => { + const body = key + .replace(/\\n/g, "\n") + .replace(/-----(BEGIN|END) [A-Z ]+-----/g, "") + .replace(/\s+/g, "") + return Uint8Array.from(atob(body), (char) => char.charCodeAt(0)) +} + +export const importApnsKey = (key: string): Promise => + crypto.subtle.importKey("pkcs8", pkcs8Of(key), { name: "ECDSA", namedCurve: "P-256" }, false, ["sign"]) + +/** + * The provider token: `{alg: ES256, kid}.{iss, iat}`, signed. WebCrypto's ECDSA signature is + * already the raw `r || s` JWS wants. + */ +export const signProviderToken = async (input: { + readonly key: CryptoKey + readonly keyId: string + readonly teamId: string + readonly nowMs: number +}): Promise => { + const signingInput = `${base64UrlJson({ alg: "ES256", kid: input.keyId })}.${base64UrlJson({ + iss: input.teamId, + iat: Math.floor(input.nowMs / 1000), + })}` + const signature = await crypto.subtle.sign( + { name: "ECDSA", hash: "SHA-256" }, + input.key, + encoder.encode(signingInput), + ) + return `${signingInput}.${base64Url(signature)}` +} + +/** Signed tokens by `teamId.keyId`, shared by every object in the isolate. */ +const providerTokens = new Map() + +let loggedDisabled = false + +const ReasonBody = Schema.fromJsonString(Schema.Struct({ reason: Schema.optional(Schema.String) })) + +const disabled: ApnsSender = { + enabled: false, + send: () => Effect.fail(new ApnsRequestError({ message: "APNs is not configured" })), +} + +export class ApnsClient extends Context.Service()("@hazel/backend/ApnsClient") { + /** APNs over the ambient `HttpClient`; the `APNS_*` config is read once. */ + static readonly layer = Layer.effect( + this, + Effect.gen(function* () { + const client = yield* HttpClient.HttpClient + const config = yield* Effect.all({ + keyId: Config.option(Config.String("APNS_KEY_ID")), + teamId: Config.option(Config.String("APNS_TEAM_ID")), + privateKey: Config.option(Config.Redacted("APNS_PRIVATE_KEY")), + }).pipe( + Effect.catchTag("ConfigError", () => + Effect.succeed({ + keyId: Option.none(), + teamId: Option.none(), + privateKey: Option.none(), + }), + ), + ) + if ( + Option.isNone(config.keyId) || + Option.isNone(config.teamId) || + Option.isNone(config.privateKey) + ) { + if (!loggedDisabled) { + loggedDisabled = true + yield* Effect.logInfo( + "APNS_KEY_ID, APNS_TEAM_ID or APNS_PRIVATE_KEY is not set; push is off", + ) + } + return disabled + } + const keyId = config.keyId.value.trim() + const teamId = config.teamId.value.trim() + const rawKey = Redacted.value(config.privateKey.value) + const key = yield* Effect.tryPromise({ + try: () => importApnsKey(rawKey), + catch: () => + new ApnsRequestError({ message: "APNS_PRIVATE_KEY does not import; push is off" }), + }).pipe( + Effect.catchTag("ApnsRequestError", (error) => + Effect.logError(error.message).pipe(Effect.as(undefined)), + ), + ) + if (key === undefined) return disabled + const cacheKey = `${teamId}.${keyId}` + + const providerToken = Effect.gen(function* () { + const nowMs = yield* Clock.currentTimeMillis + const cached = providerTokens.get(cacheKey) + if (cached !== undefined && nowMs - cached.issuedAtMs < PROVIDER_TOKEN_TTL_MS) + return cached.token + const token = yield* Effect.tryPromise({ + try: () => signProviderToken({ key, keyId, teamId, nowMs }), + catch: () => new ApnsRequestError({ message: "Could not sign the provider token" }), + }) + providerTokens.set(cacheKey, { token, issuedAtMs: nowMs }) + return token + }) + + const send = Effect.fn("ApnsClient.send")(function* (notification: ApnsNotification) { + // The peer by name (the service map's edge); the URL, which holds the token, is never recorded. + yield* Effect.annotateCurrentSpan({ + "apns.environment": notification.environment, + "peer.service": "apns", + "server.address": new URL(APNS_HOSTS[notification.environment]).host, + }) + const token = yield* providerToken + const request = HttpClientRequest.post( + `${APNS_HOSTS[notification.environment]}/3/device/${notification.token}`, + ).pipe( + HttpClientRequest.setHeaders({ + authorization: `bearer ${token}`, + "apns-topic": notification.topic, + "apns-push-type": "alert", + "apns-priority": "10", + "apns-expiration": String(notification.expiration), + "apns-collapse-id": notification.collapseId, + }), + HttpClientRequest.bodyJsonUnsafe(notification.payload), + ) + const result = yield* client.execute(request).pipe( + Effect.flatMap((response) => + response.status === 200 + ? Effect.succeed({ status: 200, reason: null }) + : // A failure's body is `{"reason": "BadDeviceToken"}`; anything else has no reason. + response.text.pipe( + Effect.flatMap(Schema.decodeUnknownEffect(ReasonBody)), + Effect.map((body): ApnsResult => ({ + status: response.status, + reason: body.reason ?? null, + })), + Effect.catchTag("SchemaError", () => + Effect.succeed({ status: response.status, reason: null }), + ), + ), + ), + Effect.timeout(APNS_REQUEST_TIMEOUT_MS), + // The client's own span would record the URL (the device token), and its trace + // headers have no business at Apple: this span is the request's. + Effect.provideService(HttpClient.TracerDisabledWhen, () => true), + Effect.catchTags({ + HttpClientError: (error) => + Effect.fail( + new ApnsRequestError({ + message: `APNs request failed (${error.reason._tag})`, + }), + ), + TimeoutError: () => + Effect.fail(new ApnsRequestError({ message: "APNs request timed out" })), + }), + ) + // Compare and delete: of several sends refused with the same token, one re-signs. + if (result.status === 403 && result.reason === "ExpiredProviderToken") { + if (providerTokens.get(cacheKey)?.token === token) providerTokens.delete(cacheKey) + } + yield* Effect.annotateCurrentSpan({ + "apns.status": result.status, + "apns.reason": result.reason ?? "", + }) + return result + }) + + return { enabled: true, send } satisfies ApnsSender + }), + ) + + /** Push off (tools and tests that never send). */ + static readonly layerDisabled = Layer.succeed(this, disabled) +} diff --git a/apps/backend/src/services/notifications/notification-service.ts b/apps/backend/src/services/notifications/notification-service.ts index 08f834496..9dc8d1ef2 100644 --- a/apps/backend/src/services/notifications/notification-service.ts +++ b/apps/backend/src/services/notifications/notification-service.ts @@ -4,11 +4,12 @@ * (`./notification-rules.ts`), bumps each recipient's `notificationCount` and sends the * notification row to their `UserObject` (`user.createNotification`) through the outbox. */ -import { Database, SQLite, schema } from "@hazel/db" +import { Database, type DatabaseError, SQLite, schema } from "@hazel/db" import type { OrganizationId, OrganizationMemberId, UserId } from "@hazel/schema" import { Array, Clock, Context, Effect, Layer } from "effect" import { enqueueEffect } from "../../worker/objects/effects/registry" import { MessageNotificationEffects } from "../../worker/objects/user/message-notifications" +import type { PushContent } from "../../worker/objects/user/push" import { decide, defaultLevel, @@ -169,10 +170,23 @@ export class NotificationService extends Context.Service()( /** Notify the recipients of `activity`: badge counts here, rows in their `UserObject`s. */ const notify = Effect.fn("NotificationService.notify")(function* ( organizationId: OrganizationId, - activity: MessageActivity & { readonly messageId: schema.Message["id"] }, + activity: MessageActivity & { + readonly messageId: schema.Message["id"] + /** + * What a recipient's push says; absent: not pushed. Run only when some recipient's + * notification is not silent (silent ones are never pushed). + */ + readonly push?: + | Effect.Effect + | undefined + }, ) { const notified = yield* recipients(organizationId, activity) if (notified.length === 0) return notified + const push = + activity.push !== undefined && notified.some((recipient) => !recipient.silent) + ? yield* activity.push + : undefined yield* Effect.forEach( Array.chunksOf( notified.map((recipient) => recipient.channelMemberId), @@ -198,6 +212,8 @@ export class NotificationService extends Context.Service()( messageId: activity.messageId, reason: recipient.reason, silent: recipient.silent, + // A silent notification shows no system notification. + push: recipient.silent ? undefined : push, }, { id: `notify:${activity.messageId}:${recipient.userId}` }, ), diff --git a/apps/backend/src/worker/env.ts b/apps/backend/src/worker/env.ts index a41c81cec..18e40c571 100644 --- a/apps/backend/src/worker/env.ts +++ b/apps/backend/src/worker/env.ts @@ -81,4 +81,8 @@ export const apiEnv = ({ stage, urls }: HazelStackContext) => optionalSecret("KLIPY_API_KEY"), // Thread naming (services/thread-namer.ts); unset disables it. optionalSecret("OPENROUTER_API_KEY"), + // iOS push (services/apns.ts): an APNs auth key (.p8); any one unset disables push. + optionalSecret("APNS_KEY_ID"), + optionalSecret("APNS_TEAM_ID"), + optionalSecret("APNS_PRIVATE_KEY"), ) diff --git a/apps/backend/src/worker/objects/activate.ts b/apps/backend/src/worker/objects/activate.ts index 6ce6574c0..edf58a7e0 100644 --- a/apps/backend/src/worker/objects/activate.ts +++ b/apps/backend/src/worker/objects/activate.ts @@ -14,6 +14,7 @@ import * as Cloudflare from "@alchemy.run/cloudflare" import { RuntimeContext } from "alchemy/RuntimeContext" import { Cause, Effect, Exit, Layer, Scope } from "effect" import { FetchHttpClient, HttpServerRequest, HttpServerResponse } from "effect/http" +import { ApnsClient } from "../../services/apns" import { BotGatewayTransport } from "../../services/bot-gateway-transport" import { Directory } from "../../services/directory" import { HtmlRenders } from "../../services/html-renders" @@ -76,6 +77,9 @@ export const activateObject = (kind: ObjectKind) => Layer.provide(Layer.mergeAll(workerEnvLayer(env), htmlRenderStoreLive(env))), Layer.orDie, ), + ApnsClient.layer.pipe( + Layer.provide(Layer.mergeAll(FetchHttpClient.layer, workerEnvLayer(env))), + ), ), }).pipe( Effect.provideService(Scope.Scope, instanceScope), diff --git a/apps/backend/src/worker/objects/channel/message-events.ts b/apps/backend/src/worker/objects/channel/message-events.ts index f0b268a6f..03ea41265 100644 --- a/apps/backend/src/worker/objects/channel/message-events.ts +++ b/apps/backend/src/worker/objects/channel/message-events.ts @@ -5,7 +5,8 @@ * write and are delivered after it. * * - bots: `botGateway.messageEvent` (message create, update, delete); - * - notifications: `org.messageCreated` to the channel's `OrgObject` (`../org/message-notifications.ts`); + * - notifications: `org.messageCreated` to the channel's `OrgObject` (`../org/message-notifications.ts`), + * with the content its push shows; `org.messageDeleted` once it is deleted, which drops them; * - thread naming: `ai.nameThread` once a thread still called "Thread" has more than 3 replies. * - search: `org.indexMessages` (message create, update, delete), `./search-index.ts`. * @@ -41,6 +42,9 @@ export type MessageEventHook = ( event: MessageEvent, ) => Effect.Effect +/** How much of a message's content `org.messageCreated` carries: a push shows far less. */ +const NOTIFICATION_CONTENT_MAX = 2000 + /** Threads are named once they have more replies than this. */ export const THREAD_NAMING_THRESHOLD = 3 @@ -75,12 +79,33 @@ const botGatewayHook: MessageEventHook = (event) => const notificationsHook: MessageEventHook = (event) => Effect.gen(function* () { - if (event._tag !== "MessageCreated") return + if (!("message" in event)) return const { message } = event + const deleted = + event._tag === "MessageDeleted" || (event._tag === "MessageUpdated" && message.deletedAt !== null) + if (event._tag !== "MessageCreated" && !deleted) return const organizationId = yield* channelOrganization if (Option.isNone(organizationId)) return + if (deleted) { + yield* enqueueEffect( + MessageNotificationOrgEffects.messageDeleted, + { kind: "org", id: organizationId.value }, + { channelId: message.channelId, messageId: message.id }, + { id: `unnotify:${message.id}` }, + ) + return + } const db = yield* Database.Database const mentions = parseMentions(message.content) + // Only a message with no text needs its attachments counted ("Sent an attachment"). + const attachments = + message.content.trim() !== "" + ? Option.none() + : yield* db.first( + SQLite.from(schema.attachmentsTable) + .select(() => ({ count: SQLite.count() })) + .where(($) => [$.messageId.eq(message.id), $.deletedAt.isNull()]), + ) const replyTo = message.replyToMessageId === null ? Option.none() @@ -100,6 +125,11 @@ const notificationsHook: MessageEventHook = (event) => mentionedUserIds: mentions.userMentions, notifyAll: mentions.hasChannelMention || mentions.hasHereMention, replyToAuthorId: Option.getOrNull(Option.map(replyTo, (row) => row.authorId)), + content: Array.from(message.content).slice(0, NOTIFICATION_CONTENT_MAX).join(""), + attachmentCount: Option.match(attachments, { + onNone: () => 0, + onSome: (row) => Number(row.count), + }), }, ) }) diff --git a/apps/backend/src/worker/objects/effects/deliver.ts b/apps/backend/src/worker/objects/effects/deliver.ts index c43762288..0e4ba2844 100644 --- a/apps/backend/src/worker/objects/effects/deliver.ts +++ b/apps/backend/src/worker/objects/effects/deliver.ts @@ -1,13 +1,14 @@ /** * Outbox delivery and receipt. `deliverOutboxRow` is every object's `deliver`: object targets get - * a DO RPC `applyEffect`; directory, bot-gateway, ai, chat-sync and storage effects run right here, - * in the sender. + * a DO RPC `applyEffect`; directory, bot-gateway, ai, chat-sync, storage and push effects run right + * here, in the sender. * `applyReceivedEffect` is the receiving object's side: decode, run once per key, report. */ import { Database } from "@hazel/db" import { OutboxDeliveryError, type OutboxRow, withInbox } from "@hazel/sync-do" import { Cause, Effect, Option, Schema } from "effect" import type { SqlClient } from "effect/sql/SqlClient" +import type { ApnsClient } from "../../../services/apns" import type { BotGatewayTransport } from "../../../services/bot-gateway-transport" import type { Directory } from "../../../services/directory" import type { ObjectKind } from "../kinds" @@ -24,6 +25,7 @@ import { customersReceivers, CustomersEffects, customersSenderRefusal } from "./ import { directoryReceivers, DirectoryEffects } from "./directory-effects" import { connectSenderRefusal } from "../org/connect" import { searchSenderRefusal } from "../org/message-search" +import { notificationCountSenderRefusal } from "../org/notification-count" import { orgReceivers, OrgEffects } from "./org-effects" import { type AnyEffectKind, @@ -34,6 +36,8 @@ import { EffectSenderRefusedError, parseEffectTarget, } from "./registry" +import { pushReceivers, PushEffects } from "../user/push" +import { PushOwnershipUserEffects } from "../user/push-ownership" import { storageReceivers, StorageEffects } from "./storage-effects" import { userReceivers, UserEffects } from "./user-effects" @@ -50,14 +54,15 @@ const side = (kinds: Record, receivers: Record> = { +export const objectReceivers: Record> = { org: side(OrgEffects, orgReceivers), channel: side(ChannelEffects, channelReceivers), user: side(UserEffects, userReceivers), customers: side(CustomersEffects, customersReceivers), } -const directorySide = side(DirectoryEffects, directoryReceivers) +// SqlClient: the sender's own outbox, for a directory step that passes work on (`push-ownership.ts`). +const directorySide = side(DirectoryEffects, directoryReceivers) const botGatewaySide = side( BotGatewayEffects, botGatewayReceivers, @@ -73,6 +78,11 @@ const chatSyncSide = side(StorageEffects, storageReceivers) +const pushSide = side( + PushEffects, + pushReceivers, +) + const failure = (message: string, retryable = true) => new OutboxDeliveryError({ message, retryable }) const decode = (kind: AnyEffectKind, payload: unknown) => @@ -93,12 +103,22 @@ type SenderGuard = ( const senderGuards: Partial> = { channel: channelSenderRefusal, customers: customersSenderRefusal, + // Only a `UserObject`'s claim of a device token makes another user's object forget it. + user: (kind, _payload, sender) => + Effect.succeed( + kind === PushOwnershipUserEffects.forgetPushDevice.name && sender.kind !== "user" + ? Option.some(`${kind} sent by ${sender.kind}`) + : Option.none(), + ), // Each check refuses only the kinds it owns; the first refusal wins. org: (kind, payload, sender) => connectSenderRefusal(kind, payload, sender).pipe( Effect.flatMap((refusal) => Option.isSome(refusal) ? Effect.succeed(refusal) : searchSenderRefusal(kind, payload, sender), ), + Effect.map((refusal) => + Option.isSome(refusal) ? refusal : notificationCountSenderRefusal(kind, payload, sender), + ), ), } @@ -152,6 +172,7 @@ export const deliverOutboxRow = ( | ThreadNamer | ChatSyncProviders | HtmlRenders + | ApnsClient > => Effect.gen(function* () { const target = parseEffectTarget(row.target) @@ -161,6 +182,7 @@ export const deliverOutboxRow = ( if (target.kind === "ai") return yield* runLocal(aiSide, row, sender) if (target.kind === "chat-sync") return yield* runLocal(chatSyncSide, row, sender) if (target.kind === "storage") return yield* runLocal(storageSide, row, sender) + if (target.kind === "push") return yield* runLocal(pushSide, row, sender) const stubs = yield* ObjectStubs const result = yield* stubs .get({ kind: target.kind, id: target.id! }) @@ -190,7 +212,7 @@ export const deliverOutboxRow = ( export const applyReceivedEffect = ( objectKind: ObjectKind, request: ApplyEffectRequest, -): Effect.Effect => +): Effect.Effect => Effect.gen(function* () { const { kinds, receivers } = objectReceivers[objectKind] const kind = kinds[request.kind] diff --git a/apps/backend/src/worker/objects/effects/directory-effects.ts b/apps/backend/src/worker/objects/effects/directory-effects.ts index 8b2bc1f76..0016c30c0 100644 --- a/apps/backend/src/worker/objects/effects/directory-effects.ts +++ b/apps/backend/src/worker/objects/effects/directory-effects.ts @@ -5,10 +5,12 @@ import { SQLite, schema } from "@hazel/db" import { ChannelId, ChannelWebhookId, OrganizationId, UserId } from "@hazel/schema" import { Effect, Schema } from "effect" +import type { SqlClient } from "effect/sql/SqlClient" import { Directory, type DirectoryShape } from "../../../services/directory" import { DirectoryWireEffects, directoryWireReceivers } from "../directory-wire" import { ConnectDirectoryEffects, connectDirectoryReceivers } from "../connect-wire" import { UserProjection } from "../org/projections" +import { PushDirectoryEffects, pushDirectoryReceivers } from "../user/push-ownership" import { ChatSyncDirectoryEffects } from "../../../services/chat-sync/chat-sync-effects" import { chatSyncDirectoryReceivers } from "../../../services/chat-sync/chat-sync-index" import { CurrentEffectSender, defineEffect, type EffectReceivers, requireOrgSender } from "./registry" @@ -73,6 +75,7 @@ export const DirectoryEffects = { ...DirectoryWireEffects, ...ChatSyncDirectoryEffects, ...ConnectDirectoryEffects, + ...PushDirectoryEffects, } /** Index rows an org owns are written only by that org's own `OrgObject` (its outbox). */ @@ -133,4 +136,5 @@ export const directoryReceivers = { ...directoryWireReceivers, ...chatSyncDirectoryReceivers, ...connectDirectoryReceivers, -} satisfies EffectReceivers + ...pushDirectoryReceivers, +} satisfies EffectReceivers diff --git a/apps/backend/src/worker/objects/effects/org-effects.ts b/apps/backend/src/worker/objects/effects/org-effects.ts index e3e55fc05..aa178f982 100644 --- a/apps/backend/src/worker/objects/effects/org-effects.ts +++ b/apps/backend/src/worker/objects/effects/org-effects.ts @@ -25,6 +25,7 @@ import { } from "../org/projections" import { channelRenameReceivers, ChannelRenameEffects } from "../org/channel-rename" import { messageNotificationOrgReceivers, MessageNotificationOrgEffects } from "../org/message-notifications" +import { notificationCountOrgReceivers, NotificationCountOrgEffects } from "../org/notification-count" import { DirectoryWireOrgEffects } from "../directory-wire" import { upsertBotProjection } from "../org/bot-installs" import { acceptInvitation } from "../org/invitation-accept" @@ -44,6 +45,7 @@ export const OrgEffects = { /** Drop a user's copy (and presence) unless they are still a member or participant here. */ removeUserProjection: defineEffect("org.removeUserProjection", "org", RemoveUserProjectionPayload), ...MessageNotificationOrgEffects, + ...NotificationCountOrgEffects, ...ChannelRenameEffects, ...DirectoryWireOrgEffects, /** Chat sync: gateway ingest, channel-provisioned link settings (`services/chat-sync`). */ @@ -59,6 +61,7 @@ export const orgReceivers = { "org.upsertPresenceProjection": upsertPresenceProjection, "org.removeUserProjection": removeUserProjection, ...messageNotificationOrgReceivers, + ...notificationCountOrgReceivers, ...channelRenameReceivers, // Called lazily: these modules import the effect registries back. "org.upsertBotProjection": (payload) => upsertBotProjection(payload), diff --git a/apps/backend/src/worker/objects/effects/registry.ts b/apps/backend/src/worker/objects/effects/registry.ts index f31a45235..d82e9e97f 100644 --- a/apps/backend/src/worker/objects/effects/registry.ts +++ b/apps/backend/src/worker/objects/effects/registry.ts @@ -23,10 +23,18 @@ import type { ObjectKind } from "../kinds" /** * Where an effect is applied: an object, the directory (D1), the bot gateway, `ai` (a model * call made by the sending object itself, `ai-effects.ts`), `chat-sync` (a provider API call - * made by the sending `ChannelObject`, `services/chat-sync/chat-sync-outbound.ts`), or `storage` - * (HTML render bucket cleanup by the sender, `storage-effects.ts`). + * made by the sending `ChannelObject`, `services/chat-sync/chat-sync-outbound.ts`), `storage` + * (HTML render bucket cleanup by the sender, `storage-effects.ts`), or `push` (an APNs request + * made by the sending `UserObject`, `../user/push.ts`; its id only names its own outbox queue). */ -export type EffectTargetKind = ObjectKind | "directory" | "bot-gateway" | "ai" | "chat-sync" | "storage" +export type EffectTargetKind = + | ObjectKind + | "directory" + | "bot-gateway" + | "ai" + | "chat-sync" + | "storage" + | "push" export interface EffectKind { readonly name: Name @@ -47,7 +55,7 @@ export const defineEffect = < ): EffectKind => ({ name, target, payload }) /** An effect's target as the outbox row stores it: `channel:`, `directory`, `bot-gateway`. */ -export type EffectTarget = Target extends ObjectKind +export type EffectTarget = Target extends ObjectKind | "push" ? { readonly kind: Target; readonly id: string } : { readonly kind: Target } @@ -70,7 +78,9 @@ export const parseEffectTarget = ( if (separator === -1) return undefined const kind = value.slice(0, separator) const id = value.slice(separator + 1) - if (kind !== "org" && kind !== "channel" && kind !== "user" && kind !== "customers") return undefined + if (kind !== "org" && kind !== "channel" && kind !== "user" && kind !== "customers" && kind !== "push") { + return undefined + } return id.length > 0 ? { kind, id } : undefined } diff --git a/apps/backend/src/worker/objects/effects/user-effects.ts b/apps/backend/src/worker/objects/effects/user-effects.ts index 9fee3bd39..64f92a9c0 100644 --- a/apps/backend/src/worker/objects/effects/user-effects.ts +++ b/apps/backend/src/worker/objects/effects/user-effects.ts @@ -3,9 +3,11 @@ import { Database, SQLite, schema } from "@hazel/db" import { ChannelId, OrganizationId } from "@hazel/schema" import { Effect, Schema } from "effect" import type { SqlClient } from "effect/sql/SqlClient" +import type { ApnsClient } from "../../../services/apns" import { markOffline, projectPresenceTo } from "../user/presence" import { PresenceEffects } from "../user/presence-kinds" import { MessageNotificationEffects, messageNotificationReceivers } from "../user/message-notifications" +import { PushOwnershipUserEffects, pushOwnershipUserReceivers } from "../user/push-ownership" import { defineEffect, type EffectReceivers } from "./registry" /** The user joined (`deletedAt: null`) or left an organization: which `OrgObject`s the client opens. */ @@ -26,6 +28,7 @@ export const UserEffects = { ), ...PresenceEffects, ...MessageNotificationEffects, + ...PushOwnershipUserEffects, } export const userReceivers = { @@ -59,4 +62,5 @@ export const userReceivers = { ), "user.markOffline": (payload) => Effect.asVoid(markOffline(payload.userId)), ...messageNotificationReceivers, -} satisfies EffectReceivers + ...pushOwnershipUserReceivers, +} satisfies EffectReceivers diff --git a/apps/backend/src/worker/objects/org/message-notifications.ts b/apps/backend/src/worker/objects/org/message-notifications.ts index 3541a48f3..f4ad3d135 100644 --- a/apps/backend/src/worker/objects/org/message-notifications.ts +++ b/apps/backend/src/worker/objects/org/message-notifications.ts @@ -8,13 +8,16 @@ * levels (all, mentions, nothing), mutes, presence, do not disturb and quiet hours. * * Each recipient's `notificationCount` goes up here; the notification row itself is the - * recipient's `UserObject`'s (`user.createNotification`). + * recipient's `UserObject`'s (`user.createNotification`), with what its push says + * (`./push-content.ts`) when the channel sent the content. */ import { Database, SQLite, schema } from "@hazel/db" -import { ChannelId, type ConnectConversationId, MessageId, UserId } from "@hazel/schema" +import { ChannelId, type ConnectConversationId, MessageId, type OrganizationId, UserId } from "@hazel/schema" import { Effect, Option, Schema } from "effect" import { NotificationService } from "../../../services/notifications/notification-service" import { enqueueEffect, defineEffect } from "../effects/registry" +import { MessageNotificationEffects } from "../user/message-notifications" +import { loadPushContent } from "./push-content" export const MessageCreatedPayload = Schema.Struct({ channelId: ChannelId, @@ -26,11 +29,23 @@ export const MessageCreatedPayload = Schema.Struct({ notifyAll: Schema.Boolean, /** The author of the message this one replies to. */ replyToAuthorId: Schema.NullOr(UserId), + /** For the push (shortened). Optional: deliveries enqueued before push have none. */ + content: Schema.optional(Schema.String), + /** The message's attachments, counted when it has no text. */ + attachmentCount: Schema.optional(Schema.Number), }) export type MessageCreatedPayload = typeof MessageCreatedPayload.Type +/** A message deleted in a channel: its notifications go too, so no push shows its text. */ +export const MessageDeletedPayload = Schema.Struct({ + channelId: ChannelId, + messageId: MessageId, +}) +export type MessageDeletedPayload = typeof MessageDeletedPayload.Type + export const MessageNotificationOrgEffects = { messageCreated: defineEffect("org.messageCreated", "org", MessageCreatedPayload), + messageDeleted: defineEffect("org.messageDeleted", "org", MessageDeletedPayload), } const localChannel = (channelId: ChannelId) => @@ -43,66 +58,76 @@ const localChannel = (channelId: ChannelId) => ), ) -/** Host: send the message's notifications to every guest org of the conversation. */ -const forwardToGuests = Effect.fn("OrgEffects.forwardToGuests")(function* ( +/** Host: run `forward` for every guest org's mount of the conversation. */ +type Guest = { readonly organizationId: OrganizationId; readonly channelId: ChannelId } + +const forEachGuest = ( conversationId: ConnectConversationId, - payload: MessageCreatedPayload, -) { - const db = yield* Database.Database - const guests = yield* db.run( - SQLite.from(schema.connectConversationChannelsTable) - .select("organizationId", "channelId") - .where(($) => [ - $.conversationId.eq(conversationId), - $.role.eq("guest"), - $.isActive.eq(true), - $.deletedAt.isNull(), - ]), - ) - yield* Effect.forEach( - guests, - (guest) => - enqueueEffect( - MessageNotificationOrgEffects.messageCreated, - { kind: "org", id: guest.organizationId }, - { ...payload, channelId: guest.channelId }, - { id: `connect-notify:${payload.messageId}:${guest.organizationId}` }, - ), - { discard: true }, - ) -}) + forward: (guest: Guest) => Effect.Effect, +) => + Effect.gen(function* () { + const db = yield* Database.Database + const guests = yield* db.run( + SQLite.from(schema.connectConversationChannelsTable) + .select("organizationId", "channelId") + .where(($) => [ + $.conversationId.eq(conversationId), + $.role.eq("guest"), + $.isActive.eq(true), + $.deletedAt.isNull(), + ]), + ) + yield* Effect.forEach(guests, forward, { discard: true }) + }).pipe(Effect.withSpan("OrgEffects.forEachGuest")) + +/** + * The channel whose members a message in `channelId` concerns here. A Connect guest's message names + * its own mount: on the host side (the host channel, or a guest's mount copy with no local channel + * of that id) it is the host channel, and `forward` passes the event on to every guest org. + */ +const notifiedChannel = ( + channelId: ChannelId, + forward: (guest: Guest) => Effect.Effect, +) => + Effect.gen(function* () { + const db = yield* Database.Database + const mount = yield* db.first( + SQLite.from(schema.connectConversationChannelsTable) + .select("conversationId", "role") + .where(($) => [$.channelId.eq(channelId), $.isActive.eq(true), $.deletedAt.isNull()]) + .limit(1), + ) + const conversation = Option.isNone(mount) + ? Option.none() + : yield* db.first( + SQLite.from(schema.connectConversationsTable) + .select("id", "hostChannelId") + .where(($) => [$.id.eq(mount.value.conversationId), $.deletedAt.isNull()]) + .limit(1), + ) + const hosted = + Option.isSome(mount) && + Option.isSome(conversation) && + (mount.value.role === "host" || Option.isNone(yield* localChannel(channelId))) + if (!hosted) return yield* localChannel(channelId) + yield* forEachGuest(conversation.value.id, forward) + return yield* localChannel(conversation.value.hostChannelId) + }).pipe(Effect.withSpan("OrgEffects.notifiedChannel")) export const messageCreated = Effect.fn("OrgEffects.messageCreated")(function* ( input: MessageCreatedPayload, ) { - let payload = input - const db = yield* Database.Database - // A Connect guest's message names its own mount: notify the host channel's members here. - const mount = yield* db.first( - SQLite.from(schema.connectConversationChannelsTable) - .select("conversationId", "role") - .where(($) => [$.channelId.eq(payload.channelId), $.isActive.eq(true), $.deletedAt.isNull()]) - .limit(1), + const channel = yield* notifiedChannel(input.channelId, (guest) => + enqueueEffect( + MessageNotificationOrgEffects.messageCreated, + { kind: "org", id: guest.organizationId }, + { ...input, channelId: guest.channelId }, + { id: `connect-notify:${input.messageId}:${guest.organizationId}` }, + ), ) - const conversation = Option.isNone(mount) - ? Option.none() - : yield* db.first( - SQLite.from(schema.connectConversationsTable) - .select("id", "hostChannelId") - .where(($) => [$.id.eq(mount.value.conversationId), $.deletedAt.isNull()]) - .limit(1), - ) - // Host side: the host channel, or a guest's mount copy (no local channel of that id). - const hosted = - Option.isSome(mount) && - Option.isSome(conversation) && - (mount.value.role === "host" || Option.isNone(yield* localChannel(payload.channelId))) - if (hosted) { - yield* forwardToGuests(conversation.value.id, payload) - payload = { ...payload, channelId: conversation.value.hostChannelId } - } - const channel = yield* localChannel(payload.channelId) if (channel._tag === "None") return + const payload = { ...input, channelId: channel.value.id } + const content = payload.content yield* NotificationService.use((service) => service.notify(channel.value.organizationId, { @@ -114,10 +139,69 @@ export const messageCreated = Effect.fn("OrgEffects.messageCreated")(function* ( mentionedUserIds: payload.mentionedUserIds, notifyAll: payload.notifyAll, replyToAuthorId: payload.replyToAuthorId, + // Worked out only when some recipient's notification is not silent. + push: + content === undefined + ? undefined + : loadPushContent({ + organizationId: channel.value.organizationId, + channelId: payload.channelId, + authorId: payload.authorId, + mentionedUserIds: payload.mentionedUserIds, + content, + attachmentCount: payload.attachmentCount ?? 0, + }), }), ).pipe(Effect.provide(NotificationService.layer)) }) +/** + * Drop the message's notifications in the `UserObject`s that may hold one: the channel's members + * with unread notifications there (`notificationCount`, reset when they read it). Each that held + * one tells the org back (`org.notificationsRemoved`), so the unread count comes down too. + */ +export const messageDeleted = Effect.fn("OrgEffects.messageDeleted")(function* ( + input: MessageDeletedPayload, +) { + const channel = yield* notifiedChannel(input.channelId, (guest) => + enqueueEffect( + MessageNotificationOrgEffects.messageDeleted, + { kind: "org", id: guest.organizationId }, + { ...input, channelId: guest.channelId }, + { id: `connect-unnotify:${input.messageId}:${guest.organizationId}` }, + ), + ) + if (channel._tag === "None") return + const db = yield* Database.Database + const members = yield* db.run( + SQLite.from(schema.channelMembersTable) + .select("userId") + .where(($) => [ + $.channelId.eq(channel.value.id), + // Members who left since keep their notifications: theirs go too. + $.notificationCount.gt(0), + ]), + ) + yield* Effect.forEach( + members, + ({ userId }) => + enqueueEffect( + MessageNotificationEffects.deleteMessageNotifications, + { kind: "user", id: userId }, + { + messageIds: [input.messageId], + organizationId: channel.value.organizationId, + channelId: channel.value.id, + userId, + }, + { id: `unnotify:${input.messageId}:${userId}` }, + ), + { discard: true }, + ) + yield* Effect.annotateCurrentSpan({ "notification.recipients": members.length }) +}) + export const messageNotificationOrgReceivers = { "org.messageCreated": messageCreated, + "org.messageDeleted": messageDeleted, } diff --git a/apps/backend/src/worker/objects/org/notification-count.test.ts b/apps/backend/src/worker/objects/org/notification-count.test.ts new file mode 100644 index 000000000..0408f0dba --- /dev/null +++ b/apps/backend/src/worker/objects/org/notification-count.test.ts @@ -0,0 +1,43 @@ +import type { ChannelId, OrganizationId, UserId } from "@hazel/schema" +import { Effect, type Scope } from "effect" +import { describe, expect, it } from "vitest" +import { orgTarget } from "../kinds" +import { makeTestObjects } from "../testing" + +const run = (effect: Effect.Effect) => + Effect.runPromise(Effect.scoped(effect) as Effect.Effect) + +describe("org.notificationsRemoved", () => { + it.each([ + [5, 3, 2], + [2, 3, 0], + [3, 3, 0], + ])("count %i, %i removed: %i", (count, removed, expected) => + run( + Effect.gen(function* () { + const objects = yield* makeTestObjects + const orgId = crypto.randomUUID() as OrganizationId + const org = yield* objects.get(orgTarget(orgId)) + const channelId = crypto.randomUUID() as ChannelId + const userId = crypto.randomUUID() as UserId + org.storage.db + .prepare( + `INSERT INTO channel_members (channelId, userId, notificationCount) VALUES (?, ?, ?)`, + ) + .run(channelId, userId, count) + const applied = yield* org.applyEffect({ + objectId: orgId, + kind: "org.notificationsRemoved", + payload: { channelId, userId, count: removed }, + key: "removed", + sender: { kind: "user", id: userId }, + }) + expect(applied._tag).toBe("Applied") + const row = org.storage.db + .prepare(`SELECT notificationCount FROM channel_members WHERE userId = ?`) + .get(userId) as { notificationCount: number } + expect(row.notificationCount).toBe(expected) + }), + ), + ) +}) diff --git a/apps/backend/src/worker/objects/org/notification-count.ts b/apps/backend/src/worker/objects/org/notification-count.ts new file mode 100644 index 000000000..8768c30dc --- /dev/null +++ b/apps/backend/src/worker/objects/org/notification-count.ts @@ -0,0 +1,59 @@ +/** + * `org.notificationsRemoved`: a member's `UserObject` dropped `count` of their notifications in a + * channel without them reading it (the message was deleted), so the channel's unread count + * (`channel_members.notificationCount`) comes down by as many, never below 0 (reading the + * channel may have reset it first). Only that user's own object may send it. + */ +import { Database, SQLite, schema } from "@hazel/db" +import { ChannelId, UserId } from "@hazel/schema" +import { Effect, Option, Schema } from "effect" +import { defineEffect, type EffectSender } from "../effects/registry" + +export const NotificationsRemovedPayload = Schema.Struct({ + channelId: ChannelId, + userId: UserId, + count: Schema.Int.check(Schema.isGreaterThan(0)), +}) +export type NotificationsRemovedPayload = typeof NotificationsRemovedPayload.Type + +export const NotificationCountOrgEffects = { + notificationsRemoved: defineEffect("org.notificationsRemoved", "org", NotificationsRemovedPayload), +} + +export const notificationsRemoved = Effect.fn("OrgEffects.notificationsRemoved")(function* ( + payload: NotificationsRemovedPayload, +) { + const db = yield* Database.Database + // max(0, count - removed), in this order: a count too low to subtract from goes to 0 first, + // so the subtraction below never sees a count it already lowered. + yield* db.run( + SQLite.update(schema.channelMembersTable) + .set({ notificationCount: 0 }) + .where(($) => [ + $.channelId.eq(payload.channelId), + $.userId.eq(payload.userId), + $.notificationCount.lt(payload.count), + $.notificationCount.gt(0), + ]), + ) + yield* db.run( + SQLite.update(schema.channelMembersTable) + .set(($) => ({ notificationCount: $.notificationCount.add(-payload.count) })) + .where(($) => [ + $.channelId.eq(payload.channelId), + $.userId.eq(payload.userId), + $.notificationCount.gte(payload.count), + ]), + ) +}) + +export const notificationCountOrgReceivers = { + "org.notificationsRemoved": notificationsRemoved, +} + +/** `org.notificationsRemoved` only from the `UserObject` of the user it names. */ +export const notificationCountSenderRefusal = (kind: string, payload: unknown, sender: EffectSender) => + kind !== NotificationCountOrgEffects.notificationsRemoved.name || + (sender.kind === "user" && sender.id === (payload as NotificationsRemovedPayload).userId) + ? Option.none() + : Option.some(`${kind} must come from the named user's object`) diff --git a/apps/backend/src/worker/objects/org/push-content.test.ts b/apps/backend/src/worker/objects/org/push-content.test.ts new file mode 100644 index 000000000..915e8a755 --- /dev/null +++ b/apps/backend/src/worker/objects/org/push-content.test.ts @@ -0,0 +1,53 @@ +import { describe, expect, it } from "vitest" +import { displayNameOf, PUSH_BODY_MAX, pushBody, pushPath, pushTitle } from "./push-content" +import type { ChannelId } from "@hazel/schema" + +const names = new Map([["u-1", "Grace Hopper"]]) + +describe("push content", () => { + it("titles a channel, a DM, a group DM and a thread", () => { + expect(pushTitle("Ada", { type: "public", name: "general" }, undefined)).toBe("Ada · #general") + expect(pushTitle("Ada", { type: "private", name: "core" }, undefined)).toBe("Ada · #core") + expect(pushTitle("Ada", { type: "single", name: "Ada, Grace" }, undefined)).toBe("Ada") + expect(pushTitle("Ada", { type: "direct", name: "Ada, Grace, Alan" }, undefined)).toBe( + "Ada · Ada, Grace, Alan", + ) + expect(pushTitle("Ada", { type: "direct", name: "" }, undefined)).toBe("Ada") + expect( + pushTitle("Ada", { type: "thread", name: "Thread" }, { type: "public", name: "general" }), + ).toBe("Ada · thread in #general") + expect(pushTitle("Ada", { type: "thread", name: "Thread" }, { type: "single", name: "x" })).toBe( + "Ada · thread", + ) + }) + + it("names users, with a fallback", () => { + expect(displayNameOf({ firstName: "Ada", lastName: "Lovelace" })).toBe("Ada Lovelace") + expect(displayNameOf({ firstName: " ", lastName: "" })).toBe("Someone") + expect(displayNameOf(undefined)).toBe("Someone") + }) + + it("turns markup into plain text", () => { + expect(pushBody("hey @[userId:u-1] and @[userId:u-2]", 0, names)).toBe( + "hey @Grace Hopper and @someone", + ) + expect(pushBody("@[directive:channel] **standup**\n\n> in 5", 0, names)).toBe("@channel standup in 5") + expect(pushBody("see [the doc](https://x.test) and `code`", 0, names)).toBe("see the doc and code") + expect(pushBody("```ts\nconst a = 1\n```", 0, names)).toBe("const a = 1") + }) + + it("shortens long messages and describes attachment-only ones", () => { + const body = pushBody("a".repeat(500), 0, names) + expect(body.length).toBe(PUSH_BODY_MAX) + expect(body.endsWith("…")).toBe(true) + const emoji = pushBody("😀".repeat(300), 0, names) + expect(emoji).toBe(`${"😀".repeat(PUSH_BODY_MAX - 1)}…`) + expect(pushBody(" ", 2, names)).toBe("Sent an attachment") + expect(pushBody("", 0, names)).toBe("Sent a message") + }) + + it("links to the channel's route", () => { + expect(pushPath("acme", "c-1" as ChannelId)).toBe("/acme/chat/c-1") + expect(pushPath(null, "c-1" as ChannelId)).toBeUndefined() + }) +}) diff --git a/apps/backend/src/worker/objects/org/push-content.ts b/apps/backend/src/worker/objects/org/push-content.ts new file mode 100644 index 000000000..962003cf5 --- /dev/null +++ b/apps/backend/src/worker/objects/org/push-content.ts @@ -0,0 +1,135 @@ +/** + * What a message's push says (`../user/push.ts`), worked out by the channel's `OrgObject`, which + * knows the channel names, the org slug and the members' names (its `users` projection): + * + * - title: `Ada Lovelace · #general`; a 1:1 DM just `Ada Lovelace`; a group DM + * `Ada Lovelace · `; a thread `Ada Lovelace · thread in #general`. + * - body: the message as plain text (mentions as `@Name`), at most `PUSH_BODY_MAX` characters; + * `Sent an attachment` for a message with only attachments. + * - path: `//chat/` (a thread is a channel with its own route). + */ +import { Database, SQLite, schema } from "@hazel/db" +import type { ChannelId, OrganizationId, UserId } from "@hazel/schema" +import { Effect, Option } from "effect" +import type { PushContent } from "../user/push" + +export const PUSH_BODY_MAX = 180 + +/** Mentioned users whose names are looked up (Durable Object SQLite binds at most 100 parameters). */ +const MAX_NAMED_USERS = 50 + +type ChannelType = schema.Channel["type"] + +interface ChannelInfo { + readonly type: ChannelType + readonly name: string +} + +export const displayNameOf = (user: { readonly firstName: string; readonly lastName: string } | undefined) => + user === undefined ? "Someone" : `${user.firstName} ${user.lastName}`.trim() || "Someone" + +const isConversation = (type: ChannelType) => type === "public" || type === "private" + +export const pushTitle = (author: string, channel: ChannelInfo, parent: ChannelInfo | undefined): string => { + switch (channel.type) { + case "single": + return author + case "direct": + return channel.name.trim() === "" ? author : `${author} · ${channel.name}` + case "thread": + return parent !== undefined && isConversation(parent.type) + ? `${author} · thread in #${parent.name}` + : `${author} · thread` + case "public": + case "private": + return `${author} · #${channel.name}` + } +} + +/** Message markup as plain text: mentions named, markdown markers and line breaks dropped. */ +export const plainTextOf = (content: string, names: ReadonlyMap): string => + content + .replace(/@\[userId:([^\]]+)\]/g, (_, id: string) => `@${names.get(id) ?? "someone"}`) + .replace(/@\[directive:([^\]]+)\]/g, "@$1") + .replace(/!?\[([^\]]*)\]\([^)]*\)/g, "$1") + .replace(/```[a-z0-9-]*/gi, "") + .replace(/`/g, "") + .replace(/\*\*|__|~~|\|\|/g, "") + .replace(/^\s*(>|#{1,6})\s?/gm, "") + .replace(/\s+/g, " ") + .trim() + +/** At most `max` code points, so an emoji or a surrogate pair is never cut in half. */ +const truncate = (text: string, max: number) => { + const points = Array.from(text) + return points.length <= max + ? text + : `${points + .slice(0, max - 1) + .join("") + .trimEnd()}…` +} + +export const pushBody = (content: string, attachmentCount: number, names: ReadonlyMap) => { + const text = plainTextOf(content, names) + if (text !== "") return truncate(text, PUSH_BODY_MAX) + return attachmentCount > 0 ? "Sent an attachment" : "Sent a message" +} + +export const pushPath = (orgSlug: string | null, channelId: ChannelId) => + orgSlug === null || orgSlug === "" ? undefined : `/${orgSlug}/chat/${channelId}` + +/** The push for a message in `channelId` (a channel of this org), from what the org holds. */ +export const loadPushContent = Effect.fn("OrgObject.pushContent")(function* (input: { + readonly organizationId: OrganizationId + readonly channelId: ChannelId + readonly authorId: UserId + readonly mentionedUserIds: ReadonlyArray + readonly content: string + readonly attachmentCount: number +}) { + const db = yield* Database.Database + const channel = yield* db.first( + SQLite.from(schema.channelsTable) + .select("type", "name", "parentChannelId") + .where(($) => [$.id.eq(input.channelId)]) + .limit(1), + ) + if (Option.isNone(channel)) return undefined + const parentChannelId = channel.value.parentChannelId + const parent = + parentChannelId === null + ? Option.none() + : yield* db.first( + SQLite.from(schema.channelsTable) + .select("type", "name") + .where(($) => [$.id.eq(parentChannelId)]) + .limit(1), + ) + const organization = yield* db.first( + SQLite.from(schema.organizationsTable) + .select("slug") + .where(($) => [$.id.eq(input.organizationId)]) + .limit(1), + ) + const userIds = [...new Set([input.authorId, ...input.mentionedUserIds])].slice(0, MAX_NAMED_USERS) + const users = yield* db.run( + SQLite.from(schema.usersTable) + .select("id", "firstName", "lastName") + .where(($) => [$.id.in_(...userIds)]), + ) + const byId = new Map(users.map((user) => [user.id as string, user])) + const names = new Map(users.map((user) => [user.id as string, displayNameOf(user)])) + return { + title: pushTitle( + displayNameOf(byId.get(input.authorId)), + channel.value, + Option.getOrUndefined(parent), + ), + body: pushBody(input.content, input.attachmentCount, names), + ...Option.match( + Option.flatMap(organization, (row) => Option.fromNullishOr(pushPath(row.slug, input.channelId))), + { onNone: () => ({}), onSome: (path) => ({ path }) }, + ), + } satisfies PushContent +}) diff --git a/apps/backend/src/worker/objects/push-notifications.test.ts b/apps/backend/src/worker/objects/push-notifications.test.ts new file mode 100644 index 000000000..837080651 --- /dev/null +++ b/apps/backend/src/worker/objects/push-notifications.test.ts @@ -0,0 +1,681 @@ +/** + * Push notifications end to end on node SQLite: a message in a `ChannelObject` notifies through + * the `OrgObject` into the recipient's `UserObject`, which pushes to its devices (a fake + * `ApnsClient`, `./testing.ts`) at once while the user is idle, or once deferred and still unread. + */ +import { CurrentUser, UnauthorizedError } from "@hazel/domain" +import type { ChannelId, MessageId, OrganizationId, UserId } from "@hazel/schema" +import { Cause, Effect, Exit, Option, type Scope } from "effect" +import { describe, expect, it } from "vitest" +import { SQLite, schema } from "@hazel/db" +import { decodeExit, encodePayload, type HazelRpcTag, rpcFor } from "../../rpc/dispatch" +import { Directory } from "../../services/directory" +import { channelTarget, orgTarget, userTarget } from "./kinds" +import { encodeInvokePrincipal } from "./principal" +import { makeTestObjects, makeTestObjectsWith, type TestObject, type TestObjects } from "./testing" + +const uuid = () => crypto.randomUUID() + +const userOf = (id: string) => + new CurrentUser.Schema({ + id: id as UserId, + role: "member", + email: `${id}@example.com`, + isOnboarded: true, + timezone: null, + settings: null, + }) + +const call = (object: TestObject, objectId: string, tag: HazelRpcTag, payload: unknown, user: string) => + Effect.gen(function* () { + const rpc = yield* rpcFor(tag) + const response = yield* object.invoke({ + objectId, + tag, + payload: yield* encodePayload(rpc, payload), + principal: yield* encodeInvokePrincipal({ user: userOf(user), botScopes: null }), + }) + return yield* decodeExit(rpc, structuredClone(response).exit) + }) + +const ok = (exit: Exit.Exit) => { + if (Exit.isFailure(exit)) throw new Error(Cause.pretty(exit.cause)) + return exit.value as any +} + +const run = (effect: Effect.Effect) => + Effect.runPromise(Effect.scoped(effect) as Effect.Effect) + +const TOKEN = "a1".repeat(32) +const BUNDLE = "app.hazel.ios" + +/** An org `acme` with Ada (the author) and Alice in `#general`, viewers delivered. */ +const makeWorld = (objects: TestObjects) => + Effect.gen(function* () { + const orgId = uuid() as OrganizationId + const org = yield* objects.get(orgTarget(orgId)) + const [ada, alice] = [uuid(), uuid()] as Array + const db = org.storage.db + db.prepare(`INSERT INTO organizations (id, name, slug) VALUES (?, 'Acme', 'acme')`).run(orgId) + for (const [userId, first, last] of [ + [ada!, "Ada", "Lovelace"], + [alice!, "Alice", "Liddell"], + ] as const) { + db.prepare( + `INSERT INTO organization_members (id, organizationId, userId, role) VALUES (?, ?, ?, 'member')`, + ).run(uuid(), orgId, userId) + db.prepare( + `INSERT INTO users (id, externalId, email, firstName, lastName) VALUES (?, ?, ?, ?, ?)`, + ).run(userId, `ext-${userId}`, `${userId}@example.com`, first, last) + } + const channelId = uuid() as ChannelId + db.prepare( + `INSERT INTO channels (id, name, type, organizationId) VALUES (?, 'general', 'public', ?)`, + ).run(channelId, orgId) + for (const userId of [ada!, alice!]) { + db.prepare(`INSERT INTO channel_members (channelId, userId) VALUES (?, ?)`).run(channelId, userId) + } + const channel = yield* objects.get(channelTarget(channelId)) + const applied = yield* channel.applyEffect({ + objectId: channelId, + kind: "channel.setViewers", + payload: { + organizationId: orgId, + conversationId: null, + viewers: [ada!, alice!].map((userId) => ({ userId, organizationId: orgId, role: "member" })), + bots: [], + }, + key: `viewers:${channelId}`, + sender: { kind: "org", id: orgId }, + }) + expect(applied._tag).toBe("Applied") + const aliceObject = yield* objects.get(userTarget(alice!)) + return { objects, orgId, org, ada: ada!, alice: alice!, channelId, channel, aliceObject } + }) + +type World = Effect.Success> + +const register = (world: World, token = TOKEN, environment: "development" | "production" = "production") => + Effect.map( + call( + world.aliceObject, + world.alice, + "pushDevice.register", + { token, environment, bundleId: BUNDLE }, + world.alice, + ), + ok, + ) + +/** Ada mentions Alice in #general. */ +const mentionAlice = (world: World, text = "lunch?") => + Effect.map( + call( + world.channel, + world.channelId, + "message.create", + { + channelId: world.channelId, + content: `@[userId:${world.alice}] ${text}`, + embeds: null, + replyToMessageId: null, + threadChannelId: null, + authorId: world.ada, + deletedAt: null, + }, + world.ada, + ), + (exit) => ok(exit).data as { id: MessageId }, + ) + +const devicesOf = (object: TestObject) => + object.storage.db.prepare(`SELECT token, environment, bundleId FROM push_devices`).all() as Array<{ + token: string + environment: string + bundleId: string + }> + +const notificationsOf = (object: TestObject) => + object.storage.db.prepare(`SELECT id FROM notifications`).all() as Array<{ id: string }> + +const pendingPushes = (object: TestObject) => + ( + object.storage.db + .prepare(`SELECT count(*) AS n FROM _outbox WHERE kind = 'push.send' AND status = 'pending'`) + .get() as { n: number } + ).n + +/** Make deferred outbox rows due and run the object's alarm. */ +const runAlarm = (object: TestObject) => + Effect.gen(function* () { + object.storage.db.exec(`UPDATE _outbox SET next_at = 0 WHERE status = 'pending'`) + yield* object.alarm + }) + +describe("push devices", () => { + it("register is an upsert and unregister forgets the token, both idempotent", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + yield* register(world, TOKEN, "development") + yield* register(world, TOKEN, "production") + expect(devicesOf(world.aliceObject)).toEqual([ + { token: TOKEN, environment: "production", bundleId: BUNDLE }, + ]) + const unregister = () => + Effect.map( + call( + world.aliceObject, + world.alice, + "pushDevice.unregister", + { token: TOKEN }, + world.alice, + ), + ok, + ) + yield* unregister() + yield* unregister() + expect(devicesOf(world.aliceObject)).toEqual([]) + }), + )) + + it("refuses a call in someone else's UserObject", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + const exit = yield* call( + world.aliceObject, + world.alice, + "pushDevice.register", + { token: TOKEN, environment: "production", bundleId: BUNDLE }, + world.ada, + ) + expect( + Exit.isFailure(exit) && + Option.match(Cause.findErrorOption(exit.cause), { + onNone: () => false, + onSome: (error) => error instanceof UnauthorizedError, + }), + ).toBe(true) + expect(devicesOf(world.aliceObject)).toEqual([]) + }), + )) +}) + +describe("push notifications", () => { + it("pushes a mention to an idle user's device: title, body, badge, path, topic", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + yield* register(world) + yield* mentionAlice(world) + + const [notification] = notificationsOf(world.aliceObject) + expect(world.objects.pushes).toEqual([ + { + environment: "production", + token: TOKEN, + topic: BUNDLE, + expiration: expect.any(Number), + collapseId: expect.any(String), + payload: { + aps: { + alert: { title: "Ada Lovelace · #general", body: "@Alice Liddell lunch?" }, + sound: "default", + badge: 1, + "thread-id": world.channelId, + }, + hazel: { + path: `/acme/chat/${world.channelId}`, + channelId: world.channelId, + notificationId: notification!.id, + }, + }, + }, + ]) + expect(pendingPushes(world.aliceObject)).toBe(0) + }), + )) + + it("does not push without devices, or when APNs is not configured", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + yield* mentionAlice(world) + expect(notificationsOf(world.aliceObject).length).toBe(1) + expect(world.objects.pushes).toEqual([]) + + const off = yield* makeWorld(yield* makeTestObjectsWith({ apns: false })) + yield* register(off) + yield* mentionAlice(off) + expect(notificationsOf(off.aliceObject).length).toBe(1) + expect(off.objects.pushes).toEqual([]) + expect(pendingPushes(off.aliceObject)).toBe(0) + }), + )) + + it("does not push a silent notification, or one from before push (no content)", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + yield* register(world) + const deliver = (key: string, payload: Record) => + world.aliceObject.applyEffect({ + objectId: world.alice, + kind: "user.createNotification", + payload: { + memberId: uuid(), + channelId: world.channelId, + messageId: uuid(), + ...payload, + }, + key, + sender: { kind: "org", id: world.orgId }, + }) + const push = { title: "Ada Lovelace · #general", body: "hi" } + expect((yield* deliver("silent", { reason: "mention", silent: true, push }))._tag).toBe( + "Applied", + ) + // The payload an older OrgObject enqueued: no push fields at all. + expect((yield* deliver("old", { reason: "mention", silent: false }))._tag).toBe("Applied") + expect((yield* deliver("oldest", {}))._tag).toBe("Applied") + expect(notificationsOf(world.aliceObject).length).toBe(3) + expect(world.objects.pushes).toEqual([]) + expect(pendingPushes(world.aliceObject)).toBe(0) + }), + )) + + it("defers while the user is online elsewhere, then pushes if still unread", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + yield* register(world) + ok( + yield* call( + world.aliceObject, + world.alice, + "userPresenceStatus.heartbeat", + {}, + world.alice, + ), + ) + yield* mentionAlice(world) + expect(notificationsOf(world.aliceObject).length).toBe(1) + expect(world.objects.pushes).toEqual([]) + expect(pendingPushes(world.aliceObject)).toBe(1) + + yield* runAlarm(world.aliceObject) + expect(world.objects.pushes.length).toBe(1) + expect(world.objects.pushes[0]!.payload.aps.alert.body).toBe("@Alice Liddell lunch?") + expect(pendingPushes(world.aliceObject)).toBe(0) + }), + )) + + it("drops a deferred push once the user read the channel", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + yield* register(world) + ok( + yield* call( + world.aliceObject, + world.alice, + "userPresenceStatus.heartbeat", + {}, + world.alice, + ), + ) + yield* mentionAlice(world) + expect(pendingPushes(world.aliceObject)).toBe(1) + + // Alice reads #general on her laptop (`channelMember.clearNotifications` in the org). + const cleared = yield* world.aliceObject.applyEffect({ + objectId: world.alice, + kind: "user.clearChannelNotifications", + payload: { channelId: world.channelId }, + key: `clear:${world.channelId}`, + sender: { kind: "org", id: world.orgId }, + }) + expect(cleared._tag).toBe("Applied") + expect(notificationsOf(world.aliceObject)).toEqual([]) + + yield* runAlarm(world.aliceObject) + expect(world.objects.pushes).toEqual([]) + expect(pendingPushes(world.aliceObject)).toBe(0) + }), + )) + + it("forgets a device APNs calls unregistered; retries a 503", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + yield* register(world) + world.objects.setApnsResult({ status: 503, reason: "ServiceUnavailable" }) + yield* mentionAlice(world) + expect(world.objects.pushes.length).toBe(1) + // Retryable: the row stays for the alarm, the device stays. + expect(pendingPushes(world.aliceObject)).toBe(1) + expect(devicesOf(world.aliceObject).length).toBe(1) + + world.objects.setApnsResult({ status: 410, reason: "Unregistered" }) + yield* runAlarm(world.aliceObject) + expect(world.objects.pushes.length).toBe(2) + expect(pendingPushes(world.aliceObject)).toBe(0) + expect(devicesOf(world.aliceObject)).toEqual([]) + }), + )) + + it("stops retrying once the notification is older than APNs's expiry", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + yield* register(world) + world.objects.setApnsResult({ status: 503, reason: "ServiceUnavailable" }) + yield* mentionAlice(world) + expect(world.objects.pushes.length).toBe(1) + + world.aliceObject.storage.db + .prepare(`UPDATE notifications SET createdAt = ?`) + .run(new Date(Date.now() - 25 * 60 * 60 * 1000).toISOString()) + yield* runAlarm(world.aliceObject) + expect(world.objects.pushes.length).toBe(1) + expect(pendingPushes(world.aliceObject)).toBe(0) + }), + )) + + it("retries an expired provider token; drops a push APNs refuses the key for", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + yield* register(world) + world.objects.setApnsResult({ status: 403, reason: "ExpiredProviderToken" }) + yield* mentionAlice(world) + expect(pendingPushes(world.aliceObject)).toBe(1) + + world.objects.setApnsResult({ status: 200, reason: null }) + yield* runAlarm(world.aliceObject) + expect(world.objects.pushes.length).toBe(2) + expect(pendingPushes(world.aliceObject)).toBe(0) + + world.objects.setApnsResult({ status: 403, reason: "InvalidProviderToken" }) + yield* mentionAlice(world, "again") + expect(world.objects.pushes.length).toBe(3) + expect(pendingPushes(world.aliceObject)).toBe(0) + expect(devicesOf(world.aliceObject).length).toBe(1) + }), + )) + + it("collapses by notification, expires a day after it, and badges only unread sounds", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + yield* register(world) + // A silent notification already waiting: not in the badge. + yield* world.aliceObject.applyEffect({ + objectId: world.alice, + kind: "user.createNotification", + payload: { + memberId: uuid(), + channelId: world.channelId, + messageId: uuid(), + silent: true, + }, + key: "silent", + sender: { kind: "org", id: world.orgId }, + }) + yield* mentionAlice(world) + const created = world.aliceObject.storage.db + .prepare(`SELECT id, createdAt FROM notifications WHERE silent = 0`) + .get() as { id: string; createdAt: string } + const [push] = world.objects.pushes + expect(push!.payload.aps.badge).toBe(1) + expect(push!.collapseId).toBe(created.id) + const createdMs = Date.parse(`${created.createdAt.replace(" ", "T").replace(/Z?$/, "Z")}`) + expect(push!.expiration).toBe(Math.floor((createdMs + 24 * 60 * 60 * 1000) / 1000)) + }), + )) + + it("keeps at most 10 devices, dropping the least recently registered", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + const tokens = Array.from({ length: 11 }, (_, i) => + i.toString(16).padStart(2, "0").repeat(32), + ) + for (const [i, token] of tokens.entries()) { + yield* register(world, token) + // Distinct registration times, oldest first. + world.aliceObject.storage.db + .prepare(`UPDATE push_devices SET updatedAt = ? WHERE token = ?`) + .run(new Date(Date.UTC(2026, 0, 1, 0, i)).toISOString(), token) + } + const left = devicesOf(world.aliceObject).map((device) => device.token) + expect(left.length).toBe(10) + expect(left).not.toContain(tokens[0]) + }), + )) + + it("drops a deferred push once its message is deleted", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + yield* register(world) + ok( + yield* call( + world.aliceObject, + world.alice, + "userPresenceStatus.heartbeat", + {}, + world.alice, + ), + ) + const sent = yield* mentionAlice(world) + expect(pendingPushes(world.aliceObject)).toBe(1) + const unreadCount = () => + ( + world.org.storage.db + .prepare( + `SELECT notificationCount FROM channel_members WHERE userId = ? AND channelId = ?`, + ) + .get(world.alice, world.channelId) as { notificationCount: number } + ).notificationCount + expect(unreadCount()).toBe(1) + + ok( + yield* call( + world.channel, + world.channelId, + "message.delete", + { id: sent.id, channelId: world.channelId }, + world.ada, + ), + ) + expect(notificationsOf(world.aliceObject)).toEqual([]) + + yield* runAlarm(world.aliceObject) + expect(world.objects.pushes).toEqual([]) + expect(pendingPushes(world.aliceObject)).toBe(0) + // The channel's unread count comes down with the notification. + expect(unreadCount()).toBe(0) + }), + )) + it("holds back a deferred push if the user turned on do not disturb meanwhile", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + yield* register(world) + ok( + yield* call( + world.aliceObject, + world.alice, + "userPresenceStatus.heartbeat", + {}, + world.alice, + ), + ) + yield* mentionAlice(world) + expect(pendingPushes(world.aliceObject)).toBe(1) + + ok( + yield* call( + world.aliceObject, + world.alice, + "userPresenceStatus.update", + { status: "dnd" }, + world.alice, + ), + ) + yield* runAlarm(world.aliceObject) + expect(world.objects.pushes).toEqual([]) + expect(pendingPushes(world.aliceObject)).toBe(0) + expect(notificationsOf(world.aliceObject).length).toBe(1) + }), + )) +}) + +describe("push device ownership", () => { + const ownerOf = (world: World, token: string) => + Directory.use(({ db }) => + db.run( + SQLite.from(schema.pushDeviceTokensTable) + .select("userId") + .where(($) => [$.token.eq(token)]), + ), + ).pipe( + Effect.provideContext(world.objects.directory), + Effect.map((rows) => rows.map((row) => row.userId)), + ) + + it("does not send to a token another user claimed before that user's forget arrives", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + yield* register(world) + expect(yield* ownerOf(world, TOKEN)).toEqual([world.alice]) + // Ada's claim landed in the directory; the forget to Alice is still in Ada's outbox. + yield* Directory.use(({ db }) => + db.run( + SQLite.update(schema.pushDeviceTokensTable) + .set({ userId: world.ada, updatedAt: new Date(Date.now() + 60_000) }) + .where(($) => [$.token.eq(TOKEN)]), + ), + ).pipe(Effect.provideContext(world.objects.directory)) + + yield* mentionAlice(world) + expect(notificationsOf(world.aliceObject).length).toBe(1) + expect(world.objects.pushes).toEqual([]) + expect(devicesOf(world.aliceObject)).toEqual([]) + expect(pendingPushes(world.aliceObject)).toBe(0) + }), + )) + + it("sends while its own claim is still on its way (an older directory row)", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + yield* Directory.use(({ db }) => + db.run( + SQLite.insertInto(schema.pushDeviceTokensTable).values({ + token: TOKEN, + userId: world.ada, + updatedAt: new Date(Date.now() - 60_000), + }), + ), + ).pipe(Effect.provideContext(world.objects.directory)) + // Alice's object has the row; her claim has not reached the directory. + world.aliceObject.storage.db + .prepare( + `INSERT INTO push_devices (token, environment, bundleId, updatedAt) VALUES (?, 'production', ?, ?)`, + ) + .run(TOKEN, BUNDLE, new Date().toISOString()) + yield* mentionAlice(world) + expect(world.objects.pushes.length).toBe(1) + }), + )) + + it("a token registered by another user leaves the previous owner's object", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + yield* register(world) + expect(yield* ownerOf(world, TOKEN)).toEqual([world.alice]) + + // Ada signs in on Alice's phone; Alice's unregister never landed. + const adaObject = yield* world.objects.get(userTarget(world.ada)) + ok( + yield* call( + adaObject, + world.ada, + "pushDevice.register", + { token: TOKEN, environment: "production", bundleId: BUNDLE }, + world.ada, + ), + ) + expect(yield* ownerOf(world, TOKEN)).toEqual([world.ada]) + // The forget goes out from Ada's outbox. + yield* runAlarm(adaObject) + expect(devicesOf(world.aliceObject)).toEqual([]) + expect(devicesOf(adaObject).map((device) => device.token)).toEqual([TOKEN]) + + yield* mentionAlice(world) + expect(notificationsOf(world.aliceObject).length).toBe(1) + expect(world.objects.pushes).toEqual([]) + }), + )) + + it("an older claim never takes a token back; unregister releases only one's own", () => + run( + Effect.gen(function* () { + const world = yield* makeWorld(yield* makeTestObjects) + const adaObject = yield* world.objects.get(userTarget(world.ada)) + ok( + yield* call( + adaObject, + world.ada, + "pushDevice.register", + { token: TOKEN, environment: "production", bundleId: BUNDLE }, + world.ada, + ), + ) + // Ada's registration is newer than Alice's claim (her claim arrives late): Alice's + // object forgets the token itself, and the directory keeps Ada. + yield* Directory.use(({ db }) => + db.run( + SQLite.update(schema.pushDeviceTokensTable) + .set({ updatedAt: new Date(Date.now() + 60 * 60 * 1000) }) + .where(($) => [$.token.eq(TOKEN)]), + ), + ).pipe(Effect.provideContext(world.objects.directory)) + yield* register(world) + yield* runAlarm(world.aliceObject) + expect(devicesOf(world.aliceObject)).toEqual([]) + expect(yield* ownerOf(world, TOKEN)).toEqual([world.ada]) + expect(devicesOf(adaObject).length).toBe(1) + + // A forget from anything but a UserObject is refused. + const refused = yield* adaObject.applyEffect({ + objectId: world.ada, + kind: "user.forgetPushDevice", + payload: { token: TOKEN, claimedAt: new Date().toISOString() }, + key: "forged", + sender: { kind: "org", id: world.orgId }, + }) + expect(refused._tag).toBe("Rejected") + expect(devicesOf(adaObject).length).toBe(1) + + // Alice unregistering a token Ada holds leaves Ada's directory row. + ok( + yield* call( + world.aliceObject, + world.alice, + "pushDevice.unregister", + { token: TOKEN }, + world.alice, + ), + ) + expect(yield* ownerOf(world, TOKEN)).toEqual([world.ada]) + ok(yield* call(adaObject, world.ada, "pushDevice.unregister", { token: TOKEN }, world.ada)) + expect(yield* ownerOf(world, TOKEN)).toEqual([]) + }), + )) +}) diff --git a/apps/backend/src/worker/objects/runtime.ts b/apps/backend/src/worker/objects/runtime.ts index 657aa1e9c..b39a56ad5 100644 --- a/apps/backend/src/worker/objects/runtime.ts +++ b/apps/backend/src/worker/objects/runtime.ts @@ -28,6 +28,7 @@ import { makeRpcDispatcher, rpcFor, } from "../../rpc/dispatch" +import type { ApnsClient } from "../../services/apns" import type { BotGatewayTransport } from "../../services/bot-gateway-transport" import type { Directory } from "../../services/directory" import type { HtmlRenders } from "../../services/html-renders" @@ -75,6 +76,7 @@ export type ObjectPlatform = | ThreadNamer | ChatSyncProviders | HtmlRenders + | ApnsClient export interface ObjectRuntimeOptions { readonly kind: ObjectKind diff --git a/apps/backend/src/worker/objects/testing.ts b/apps/backend/src/worker/objects/testing.ts index f120b2e60..3116478df 100644 --- a/apps/backend/src/worker/objects/testing.ts +++ b/apps/backend/src/worker/objects/testing.ts @@ -11,6 +11,7 @@ import type { BotId } from "@hazel/schema" import { type FakeStorage, makeFakeState, makeFakeStorage, sqliteClientLayer } from "@hazel/sync-do/testing" import { ConfigProvider, Context, Deferred, Effect, type Exit, Layer, Option, Scope } from "effect" import { Persistence } from "effect/persistence" +import { ApnsClient, type ApnsNotification, type ApnsResult } from "../../services/apns" import { BotGatewayTransport } from "../../services/bot-gateway-transport" import { Directory } from "../../services/directory" import { HtmlRenderStore, HtmlRenders } from "../../services/html-renders" @@ -82,6 +83,11 @@ export interface TestObjects { readonly setThreadName: (name: string) => void /** The HTML render pages (`HtmlRenderStore`), by key. */ readonly bucket: Map + /** What the objects sent APNs, in order. */ + readonly pushes: Array + /** APNs's answer to the next pushes (default 200). */ + readonly setApnsResult: (result: ApnsResult) => void + /** Run `tag` (decoded payload) on the directory path, as the api Worker does for D1 methods. */ readonly invokeDirectory: ( tag: string, @@ -93,6 +99,8 @@ export interface TestObjects { export interface TestObjectsOptions { /** Each kind's migrations, in place of its store's (`storeMigrations`). */ readonly migrations?: Partial>> + /** APNs configured (default true): the fake records pushes. */ + readonly apns?: boolean } export const makeTestObjectsWith = (options: TestObjectsOptions = {}) => @@ -207,7 +215,18 @@ export const makeTestObjectsWith = (options: TestObjectsOptions = {}) => Layer.provide(ConfigProvider.layer(ConfigProvider.fromUnknown(TEST_CONFIG))), Layer.orDie, ) + const pushes: Array = [] + let apnsResult: ApnsResult = { status: 200, reason: null } + const apns = Layer.succeed(ApnsClient, { + enabled: options.apns ?? true, + send: (notification) => + Effect.sync(() => { + pushes.push(structuredClone(notification)) + return apnsResult + }), + }) const platform = Layer.mergeAll( + apns, Layer.succeedContext(directory), Layer.succeed(ObjectStubs, stubs), transport, @@ -285,6 +304,10 @@ export const makeTestObjectsWith = (options: TestObjectsOptions = {}) => return { get, bucket, + pushes, + setApnsResult: (result) => { + apnsResult = result + }, invokeDirectory: (tag, payload, principal) => isolated( Effect.flatMap(directoryPath, (dispatcher) => diff --git a/apps/backend/src/worker/objects/user/message-notifications.ts b/apps/backend/src/worker/objects/user/message-notifications.ts index dfba90475..026e59e5f 100644 --- a/apps/backend/src/worker/objects/user/message-notifications.ts +++ b/apps/backend/src/worker/objects/user/message-notifications.ts @@ -2,12 +2,15 @@ * `user.createNotification`: a message notification for the `UserObject`'s owner, from the * channel's `OrgObject` (`../org/message-notifications.ts`). `memberId` is the owner's * `organization_members` id in that org, as the client's notifications collection expects. + * `push` is what the owner's phones show, if it is pushed (`./push.ts`). */ import { Database, SQLite, schema } from "@hazel/db" -import { ChannelId, MessageId, OrganizationMemberId } from "@hazel/schema" +import { ChannelId, MessageId, OrganizationId, OrganizationMemberId, UserId } from "@hazel/schema" import { Effect, Schema } from "effect" import { NotificationReasons } from "../../../services/notifications/notification-rules" -import { defineEffect } from "../effects/registry" +import { defineEffect, enqueueEffect } from "../effects/registry" +import { NotificationCountOrgEffects } from "../org/notification-count" +import { PushContent, schedulePush } from "./push" export const CreateNotificationPayload = Schema.Struct({ memberId: OrganizationMemberId, @@ -16,30 +19,75 @@ export const CreateNotificationPayload = Schema.Struct({ /** Optional: deliveries enqueued before reasons have none. */ reason: Schema.optional(Schema.Literals(NotificationReasons)), silent: Schema.optional(Schema.Boolean), + /** Optional: deliveries enqueued before push have none, and are not pushed. */ + push: Schema.optional(PushContent), }) export type CreateNotificationPayload = typeof CreateNotificationPayload.Type +/** Messages deleted: their notifications go, so a deferred or retrying push finds nothing to send. */ +export const DeleteMessageNotificationsPayload = Schema.Struct({ + messageIds: Schema.Array(MessageId).check(Schema.isMaxLength(90)), + /** The channel's org, told how many came off its unread count. */ + organizationId: OrganizationId, + channelId: ChannelId, + /** This object's owner (the org's guard holds the reply to it). */ + userId: UserId, +}) + export const MessageNotificationEffects = { createNotification: defineEffect("user.createNotification", "user", CreateNotificationPayload), + deleteMessageNotifications: defineEffect( + "user.deleteMessageNotifications", + "user", + DeleteMessageNotificationsPayload, + ), } export const createNotification = Effect.fn("UserEffects.createNotification")(function* ( payload: CreateNotificationPayload, ) { const db = yield* Database.Database - yield* db.run( - SQLite.insertInto(schema.notificationsTable).values({ - memberId: payload.memberId, - targetedResourceId: payload.channelId, - targetedResourceType: "channel", - resourceId: payload.messageId, - resourceType: "message", - reason: payload.reason ?? null, - silent: payload.silent ?? false, - }), + const [created] = yield* db.run( + SQLite.insertInto(schema.notificationsTable) + .values({ + memberId: payload.memberId, + targetedResourceId: payload.channelId, + targetedResourceType: "channel", + resourceId: payload.messageId, + resourceType: "message", + reason: payload.reason ?? null, + silent: payload.silent ?? false, + }) + .returning("id"), + ) + if (created === undefined) return + yield* schedulePush({ + notificationId: created.id, + channelId: payload.channelId, + silent: payload.silent ?? false, + content: payload.push, + }) +}) + +export const deleteMessageNotifications = Effect.fn("UserEffects.deleteMessageNotifications")(function* ( + payload: typeof DeleteMessageNotificationsPayload.Type, +) { + if (payload.messageIds.length === 0) return + const db = yield* Database.Database + const removed = yield* db.run( + SQLite.deleteFrom(schema.notificationsTable) + .where(($) => [$.resourceId.in_(...payload.messageIds), $.resourceType.eq("message")]) + .returning("id"), + ) + if (removed.length === 0) return + yield* enqueueEffect( + NotificationCountOrgEffects.notificationsRemoved, + { kind: "org", id: payload.organizationId }, + { channelId: payload.channelId, userId: payload.userId, count: removed.length }, ) }) export const messageNotificationReceivers = { "user.createNotification": createNotification, + "user.deleteMessageNotifications": deleteMessageNotifications, } diff --git a/apps/backend/src/worker/objects/user/push-ownership.ts b/apps/backend/src/worker/objects/user/push-ownership.ts new file mode 100644 index 000000000..d1b9c78d0 --- /dev/null +++ b/apps/backend/src/worker/objects/user/push-ownership.ts @@ -0,0 +1,190 @@ +/** + * Who a device token belongs to. Each user's `UserObject` keeps its own `push_devices`, so a + * phone that signs into another account (without the first one's `pushDevice.unregister` ever + * landing) would keep getting the first user's pushes. The directory's `push_device_tokens` + * (D1) names the token's current owner, and a new owner's claim takes it from the previous one: + * + * 1. `pushDevice.register` (in the caller's `UserObject`) writes the local row with `updatedAt` + * = now and enqueues `directory.claimPushToken {token, userId, claimedAt}` in the same + * transaction. + * 2. The claim runs in that object's outbox (directory effects run in the sender). It reads the + * token's row; if another user holds it with an older `updatedAt`, it enqueues + * `user.forgetPushDevice {token, claimedAt}` to that user's `UserObject`, then upserts the row + * (an older claim never overwrites a newer one). A claim older than the row it finds (a late + * retry) makes its own object forget the token instead. Each step is one statement and the + * forget's outbox id is fixed, so a retried claim repeats nothing. + * 3. The previous owner's object deletes its row only if it is no newer than the claim, so a + * stale forget never removes a newer registration. + * + * `pushDevice.unregister` releases the directory row if it still names the caller. + * + * Until the forget lands, every send checks the directory too (`push.send`, `./push.ts`): a token + * the directory gives another user at or after the local registration is not sent to, and its + * local row goes. That also settles two users claiming an unowned token at the same moment (both + * read no owner, so neither forgets the other): the earlier claimant's row can never send, and is + * deleted on its next attempt. + */ +import { Database, SQLite, schema } from "@hazel/db" +import { UserId } from "@hazel/schema" +import { Effect, Option, Schema } from "effect" +import type { SqlClient } from "effect/sql/SqlClient" +import { Directory } from "../../../services/directory" +import { + CurrentEffectSender, + defineEffect, + EffectSenderRefusedError, + enqueueEffect, + type EffectReceivers, +} from "../effects/registry" + +export const ClaimPushTokenPayload = Schema.Struct({ + token: Schema.String, + userId: UserId, + /** The local row's `updatedAt`, ISO. */ + claimedAt: Schema.String, +}) + +export const ReleasePushTokenPayload = Schema.Struct({ token: Schema.String, userId: UserId }) + +export const ForgetPushDevicePayload = Schema.Struct({ + token: Schema.String, + /** Only a row registered at or before this (ISO) is forgotten. */ + claimedAt: Schema.String, +}) + +export const PushDirectoryEffects = { + claimPushToken: defineEffect("directory.claimPushToken", "directory", ClaimPushTokenPayload), + releasePushToken: defineEffect("directory.releasePushToken", "directory", ReleasePushTokenPayload), +} + +export const PushOwnershipUserEffects = { + forgetPushDevice: defineEffect("user.forgetPushDevice", "user", ForgetPushDevicePayload), +} + +/** Only the `UserObject` of `userId` claims or releases its tokens. */ +const requireUserSender = (kind: string, userId: UserId) => + Effect.flatMap(CurrentEffectSender, (sender) => + sender.kind === "user" && sender.id === userId + ? Effect.void + : Effect.fail( + new EffectSenderRefusedError({ + kind, + message: `${kind} for user ${userId} sent by ${sender.kind}${"id" in sender ? `:${sender.id}` : ""}`, + }), + ), + ) + +/** A short FNV-1a hash of `token`: tells tokens apart in an outbox id without naming them. */ +const tokenTag = (token: string) => { + let hash = 0x811c9dc5 + for (let index = 0; index < token.length; index++) { + hash = Math.imul(hash ^ token.charCodeAt(index), 0x01000193) + } + return (hash >>> 0).toString(16).padStart(8, "0") +} + +/** `userId`'s `UserObject` forgets `token` if registered no later than `claimedAt`. */ +const forget = (userId: UserId, token: string, claimedAt: Date, claimant: UserId) => + enqueueEffect( + PushOwnershipUserEffects.forgetPushDevice, + { kind: "user", id: userId }, + { token, claimedAt: claimedAt.toISOString() }, + // Fixed per claim (a retried claim enqueues nothing new), apart per token (two tokens + // claimed in one millisecond). The token itself stays out of the id. + { id: `push-forget:${claimant}:${userId}:${claimedAt.getTime()}:${tokenTag(token)}` }, + ) + +/** The token's owner in the directory, and since when. */ +export const tokenOwner = (token: string) => + Effect.flatMap(Directory, ({ db }) => + db.first( + SQLite.from(schema.pushDeviceTokensTable) + .select("userId", "updatedAt") + .where(($) => [$.token.eq(token)]) + .limit(1), + ), + ) + +const claimPushToken = Effect.fn("Directory.claimPushToken")(function* ( + payload: typeof ClaimPushTokenPayload.Type, +) { + yield* requireUserSender("directory.claimPushToken", payload.userId) + const { db } = yield* Directory + const claimedAt = new Date(payload.claimedAt) + const previous = yield* tokenOwner(payload.token) + if (Option.isSome(previous) && previous.value.userId !== payload.userId) { + if (previous.value.updatedAt.getTime() > claimedAt.getTime()) { + // Someone registered the token after this claim: it is theirs. + yield* forget(payload.userId, payload.token, previous.value.updatedAt, payload.userId) + yield* Effect.annotateCurrentSpan({ "push.claim": "stale" }) + return + } + yield* forget(previous.value.userId, payload.token, claimedAt, payload.userId) + } + const written = yield* db.run( + SQLite.insertInto(schema.pushDeviceTokensTable) + .values({ token: payload.token, userId: payload.userId, updatedAt: claimedAt }) + .onConflictDoUpdate({ + target: ["token"], + set: { userId: payload.userId, updatedAt: claimedAt }, + where: ($) => $.updatedAt.lte(claimedAt), + }) + .returning("token"), + ) + if (written.length === 0) { + // A newer claim landed between the read and the write. + const current = yield* tokenOwner(payload.token) + if (Option.isSome(current) && current.value.userId !== payload.userId) { + yield* forget(payload.userId, payload.token, current.value.updatedAt, payload.userId) + } + } + yield* Effect.annotateCurrentSpan({ + "push.claim": Option.isSome(previous) && previous.value.userId !== payload.userId ? "moved" : "kept", + }) +}) + +const releasePushToken = Effect.fn("Directory.releasePushToken")(function* ( + payload: typeof ReleasePushTokenPayload.Type, +) { + yield* requireUserSender("directory.releasePushToken", payload.userId) + const { db } = yield* Directory + yield* db.run( + SQLite.deleteFrom(schema.pushDeviceTokensTable).where(($) => [ + $.token.eq(payload.token), + $.userId.eq(payload.userId), + ]), + ) +}) + +export const pushDirectoryReceivers = { + "directory.claimPushToken": claimPushToken, + "directory.releasePushToken": releasePushToken, +} satisfies EffectReceivers + +const forgetPushDevice = Effect.fn("UserObject.forgetPushDevice")(function* ( + payload: typeof ForgetPushDevicePayload.Type, +) { + const db = yield* Database.Database + const forgotten = yield* db.run( + SQLite.deleteFrom(schema.pushDevicesTable) + .where(($) => [$.token.eq(payload.token), $.updatedAt.lte(new Date(payload.claimedAt))]) + .returning("token"), + ) + yield* Effect.annotateCurrentSpan({ "push.forgotten": forgotten.length }) +}) + +export const pushOwnershipUserReceivers = { + "user.forgetPushDevice": forgetPushDevice, +} satisfies EffectReceivers + +/** Claim `token` for `userId` (inside the registration's transaction). */ +export const claimToken = (userId: UserId, token: string, claimedAt: Date) => + enqueueEffect( + PushDirectoryEffects.claimPushToken, + { kind: "directory" }, + { token, userId, claimedAt: claimedAt.toISOString() }, + ) + +/** Release `token` if the directory still names `userId` (inside the unregistration's transaction). */ +export const releaseToken = (userId: UserId, token: string) => + enqueueEffect(PushDirectoryEffects.releasePushToken, { kind: "directory" }, { token, userId }) diff --git a/apps/backend/src/worker/objects/user/push.test.ts b/apps/backend/src/worker/objects/user/push.test.ts new file mode 100644 index 000000000..b70e11d0a --- /dev/null +++ b/apps/backend/src/worker/objects/user/push.test.ts @@ -0,0 +1,61 @@ +import { describe, expect, it } from "vitest" +import { apnsOutcome, decidePush, PUSH_IDLE_DELAY_MS, type PushInput } from "./push" +import { HEARTBEAT_TIMEOUT_MS } from "./presence-timing" + +const NOW = 1_800_000_000_000 + +const input = (overrides: Partial = {}): PushInput => ({ + enabled: true, + hasContent: true, + silent: false, + devices: 1, + presence: undefined, + nowMs: NOW, + ...overrides, +}) + +const seen = (status: NonNullable["status"], agoMs = 0) => ({ + status, + lastSeenAt: new Date(NOW - agoMs), +}) + +describe("decidePush", () => { + it("pushes at once to a user never seen, offline or away", () => { + expect(decidePush(input())).toEqual({ _tag: "Now" }) + expect(decidePush(input({ presence: seen("offline") }))).toEqual({ _tag: "Now" }) + expect(decidePush(input({ presence: seen("away") }))).toEqual({ _tag: "Now" }) + }) + + it("defers while the user is online or busy on another client", () => { + const deferred = { _tag: "Defer", delayMs: PUSH_IDLE_DELAY_MS } + expect(decidePush(input({ presence: seen("online", 1_000) }))).toEqual(deferred) + expect(decidePush(input({ presence: seen("busy", 1_000) }))).toEqual(deferred) + }) + + it("pushes at once when the online status is stale (no heartbeat for the timeout)", () => { + expect(decidePush(input({ presence: seen("online", HEARTBEAT_TIMEOUT_MS) }))).toEqual({ _tag: "Now" }) + }) + + it("never pushes on dnd, silent, without devices, content or config", () => { + expect(decidePush(input({ presence: seen("dnd") }))).toEqual({ _tag: "Skip", reason: "dnd" }) + expect(decidePush(input({ silent: true }))).toEqual({ _tag: "Skip", reason: "silent" }) + expect(decidePush(input({ devices: 0 }))).toEqual({ _tag: "Skip", reason: "no-devices" }) + expect(decidePush(input({ hasContent: false }))).toEqual({ _tag: "Skip", reason: "no-content" }) + expect(decidePush(input({ enabled: false }))).toEqual({ _tag: "Skip", reason: "disabled" }) + }) +}) + +describe("apnsOutcome", () => { + it("sorts APNs's answers", () => { + expect(apnsOutcome({ status: 200, reason: null })).toBe("Delivered") + expect(apnsOutcome({ status: 410, reason: "Unregistered" })).toBe("Unregistered") + expect(apnsOutcome({ status: 400, reason: "BadDeviceToken" })).toBe("Unregistered") + expect(apnsOutcome({ status: 400, reason: "DeviceTokenNotForTopic" })).toBe("Unregistered") + expect(apnsOutcome({ status: 403, reason: "ExpiredProviderToken" })).toBe("ProviderToken") + expect(apnsOutcome({ status: 403, reason: "InvalidProviderToken" })).toBe("Misconfigured") + expect(apnsOutcome({ status: 429, reason: "TooManyRequests" })).toBe("Retry") + expect(apnsOutcome({ status: 503, reason: "ServiceUnavailable" })).toBe("Retry") + expect(apnsOutcome({ status: 400, reason: "PayloadTooLarge" })).toBe("Rejected") + expect(apnsOutcome({ status: 403, reason: "BadCertificate" })).toBe("Rejected") + }) +}) diff --git a/apps/backend/src/worker/objects/user/push.ts b/apps/backend/src/worker/objects/user/push.ts new file mode 100644 index 000000000..33bc81b85 --- /dev/null +++ b/apps/backend/src/worker/objects/user/push.ts @@ -0,0 +1,333 @@ +/** + * Push notifications inside the `UserObject`: the owner's devices (`push_devices`), whether a new + * notification is pushed (`decidePush`), and the `push.send` outbox effect that sends it to APNs + * (`services/apns.ts`), one row per device. + * + * A push goes out only while the user is idle elsewhere: offline or away (or never seen) is + * pushed at once; online or busy on some client is deferred `PUSH_IDLE_DELAY_MS`, an outbox row + * the alarm delivers. Every send first checks that the notification still exists and is unread + * (reading the channel deletes it, `user.clearChannelNotifications`), so a deferred push the + * user already saw on another client is dropped. Silent notifications (do not disturb, quiet + * hours) and `dnd` are never pushed. + * + * Each row has its own target (`push:.`): a retrying device never holds back + * another device or a later notification behind the outbox's per-target order. + */ +import { Database, SQLite, schema } from "@hazel/db" +import { ChannelId, NotificationId } from "@hazel/schema" +import { Clock, Effect, Option, Schema } from "effect" +import { ApnsClient, ApnsRetryableError, type ApnsResult } from "../../../services/apns" +import { Directory } from "../../../services/directory" +import { CurrentEffectSender, defineEffect, enqueueEffect, type EffectReceivers } from "../effects/registry" +import { tokenOwner } from "./push-ownership" +import { HEARTBEAT_TIMEOUT_MS } from "./presence-timing" + +/** How long a notification waits while the user is active elsewhere. */ +export const PUSH_IDLE_DELAY_MS = 60_000 + +/** How long APNs keeps trying an undelivered push. */ +export const PUSH_EXPIRATION_MS = 24 * 60 * 60 * 1000 + +/** What a push says, and the app path it opens. */ +export const PushContent = Schema.Struct({ + title: Schema.String, + body: Schema.String, + /** `//chat/`; absent when the org has no slug. */ + path: Schema.optional(Schema.String), +}) +export type PushContent = typeof PushContent.Type + +export const PushSendPayload = Schema.Struct({ + notificationId: NotificationId, + channelId: ChannelId, + token: Schema.String, + ...PushContent.fields, +}) +export type PushSendPayload = typeof PushSendPayload.Type + +export const PushEffects = { + send: defineEffect("push.send", "push", PushSendPayload), +} + +// DECISION + +export type PushDecision = + | { readonly _tag: "Skip"; readonly reason: "disabled" | "no-content" | "silent" | "no-devices" | "dnd" } + | { readonly _tag: "Now" } + | { readonly _tag: "Defer"; readonly delayMs: number } + +export interface PushInput { + readonly enabled: boolean + readonly hasContent: boolean + readonly silent: boolean + readonly devices: number + /** The owner's presence row; undefined: never seen. */ + readonly presence: + | { readonly status: schema.UserPresenceStatus["status"]; readonly lastSeenAt: Date } + | undefined + readonly nowMs: number +} + +/** Whether a new notification is pushed, now or after `PUSH_IDLE_DELAY_MS`. Pure. */ +export const decidePush = (input: PushInput): PushDecision => { + if (!input.enabled) return { _tag: "Skip", reason: "disabled" } + if (!input.hasContent) return { _tag: "Skip", reason: "no-content" } + if (input.silent) return { _tag: "Skip", reason: "silent" } + if (input.devices === 0) return { _tag: "Skip", reason: "no-devices" } + const { presence } = input + if (presence === undefined) return { _tag: "Now" } + switch (presence.status) { + case "dnd": + return { _tag: "Skip", reason: "dnd" } + case "offline": + case "away": + return { _tag: "Now" } + case "online": + case "busy": + // No heartbeat for the timeout: gone, even if the alarm has not marked it offline yet. + return presence.lastSeenAt.getTime() + HEARTBEAT_TIMEOUT_MS <= input.nowMs + ? { _tag: "Now" } + : { _tag: "Defer", delayMs: PUSH_IDLE_DELAY_MS } + } +} + +/** What to do with APNs's answer. Pure. */ +export type ApnsOutcome = + | "Delivered" + /** The token is dead: forget the device. */ + | "Unregistered" + /** The provider token expired: the client signs a new one, retry. */ + | "ProviderToken" + /** The provider token is wrong (key, key id or team): no retry fixes that. */ + | "Misconfigured" + | "Retry" + /** Any other refusal: log and drop (retrying would not help). */ + | "Rejected" + +const DEAD_TOKEN_REASONS = new Set(["BadDeviceToken", "DeviceTokenNotForTopic", "Unregistered"]) + +export const apnsOutcome = ({ status, reason }: ApnsResult): ApnsOutcome => { + if (status === 200) return "Delivered" + if (status === 410) return "Unregistered" + if (status === 400 && reason !== null && DEAD_TOKEN_REASONS.has(reason)) return "Unregistered" + if (status === 403 && reason === "ExpiredProviderToken") return "ProviderToken" + if (status === 403 && reason === "InvalidProviderToken") return "Misconfigured" + if (status === 429 || status >= 500) return "Retry" + return "Rejected" +} + +// DEVICES + +/** A user's devices past this many: the least recently registered go. */ +export const MAX_PUSH_DEVICES = 10 + +/** Register (or refresh) a device; returns its `updatedAt`. */ +export const registerDevice = Effect.fn("UserObject.registerDevice")(function* (device: { + readonly token: string + readonly environment: schema.PushDevice["environment"] + readonly bundleId: string +}) { + const db = yield* Database.Database + const updatedAt = new Date(yield* Clock.currentTimeMillis) + yield* db.run( + SQLite.insertInto(schema.pushDevicesTable) + .values({ ...device, updatedAt }) + .onConflictDoUpdate({ + target: ["token"], + set: { environment: device.environment, bundleId: device.bundleId, updatedAt }, + }), + ) + yield* db.query( + Database.sql`DELETE FROM push_devices WHERE token NOT IN (SELECT token FROM push_devices ORDER BY updatedAt DESC LIMIT ${MAX_PUSH_DEVICES})`, + ) + return updatedAt +}) + +/** Forget a device; with `registeredBy`, only if it was not registered again after that. */ +export const unregisterDevice = (token: string, registeredBy?: Date) => + Effect.flatMap(Database.Database, (db) => + db.run( + SQLite.deleteFrom(schema.pushDevicesTable).where(($) => [ + $.token.eq(token), + ...(registeredBy === undefined ? [] : [$.updatedAt.lte(registeredBy)]), + ]), + ), + ) + +// SCHEDULING + +/** + * Decide on a new notification's push and enqueue one `push.send` per device (deferred while the + * user is active elsewhere). Inside the notification's transaction. + */ +export const schedulePush = Effect.fn("UserObject.pushDecision")(function* (input: { + readonly notificationId: NotificationId + readonly channelId: ChannelId + readonly silent: boolean + readonly content: PushContent | undefined +}) { + const apns = yield* ApnsClient + const db = yield* Database.Database + const devices = yield* db.run(SQLite.from(schema.pushDevicesTable).select("token")) + // The store holds only its owner's presence row. + const presence = yield* db.first( + SQLite.from(schema.userPresenceStatusTable).select("status", "lastSeenAt").limit(1), + ) + const decision = decidePush({ + enabled: apns.enabled, + hasContent: input.content !== undefined, + silent: input.silent, + devices: devices.length, + presence: Option.getOrUndefined(presence), + nowMs: yield* Clock.currentTimeMillis, + }) + yield* Effect.annotateCurrentSpan({ + "push.decision": decision._tag, + "push.skip_reason": decision._tag === "Skip" ? decision.reason : "", + "push.devices": devices.length, + "push.presence": Option.match(presence, { onNone: () => "none", onSome: (row) => row.status }), + }) + if (decision._tag === "Skip" || input.content === undefined) return decision + const content = input.content + yield* Effect.forEach( + devices, + (device, index) => + enqueueEffect( + PushEffects.send, + { kind: "push", id: `${input.notificationId}.${index}` }, + { + notificationId: input.notificationId, + channelId: input.channelId, + token: device.token, + ...content, + }, + decision._tag === "Defer" ? { delayMs: decision.delayMs } : undefined, + ), + { discard: true }, + ).pipe(Effect.withSpan("UserObject.enqueuePush", { attributes: { "push.devices": devices.length } })) + return decision +}) + +// SENDING + +/** + * `push.send`: one device, if the notification is still unread, the user is not on do not + * disturb now, and the token is still theirs. Runs in the sender's outbox: `CurrentEffectSender` + * is this `UserObject`, with its stored id, so it names the owner. + */ +const sendPush = Effect.fn("UserObject.sendPush")(function* (payload: PushSendPayload) { + const db = yield* Database.Database + const apns = yield* ApnsClient + const notification = yield* db.first( + SQLite.from(schema.notificationsTable) + .select("readAt", "createdAt") + .where(($) => [$.id.eq(payload.notificationId)]) + .limit(1), + ) + if (Option.isNone(notification) || notification.value.readAt !== null) { + yield* Effect.annotateCurrentSpan({ "push.skipped": "read" }) + return + } + const nowMs = yield* Clock.currentTimeMillis + // A send still retrying past APNs's own expiry would only show stale news. + if (notification.value.createdAt.getTime() + PUSH_EXPIRATION_MS <= nowMs) { + yield* Effect.annotateCurrentSpan({ "push.skipped": "expired" }) + return + } + // Deferred while they were online: do not disturb set since then holds it back. + const presence = yield* db.first(SQLite.from(schema.userPresenceStatusTable).select("status").limit(1)) + if (Option.isSome(presence) && presence.value.status === "dnd") { + yield* Effect.annotateCurrentSpan({ "push.skipped": "dnd" }) + return + } + const device = yield* db.first( + SQLite.from(schema.pushDevicesTable) + .select("environment", "bundleId", "updatedAt") + .where(($) => [$.token.eq(payload.token)]) + .limit(1), + ) + if (Option.isNone(device)) { + yield* Effect.annotateCurrentSpan({ "push.skipped": "unregistered" }) + return + } + if (!apns.enabled) { + yield* Effect.annotateCurrentSpan({ "push.skipped": "disabled" }) + return + } + // Another user claimed the token at or after our registration, and its forget has not landed + // yet: the phone is theirs now. (A directory row older than ours is our own claim pending.) + const sender = yield* CurrentEffectSender + const owner = yield* tokenOwner(payload.token) + if ( + sender.kind === "user" && + Option.isSome(owner) && + owner.value.userId !== sender.id && + owner.value.updatedAt.getTime() >= device.value.updatedAt.getTime() + ) { + yield* unregisterDevice(payload.token, owner.value.updatedAt) + yield* Effect.annotateCurrentSpan({ "push.skipped": "not-owner" }) + return + } + // The badge: what would make a sound, unread. + const unread = yield* db.first( + SQLite.from(schema.notificationsTable) + .select(() => ({ count: SQLite.count() })) + .where(($) => [$.readAt.isNull(), $.silent.eq(false)]), + ) + const sentAt = new Date(nowMs) + const result = yield* apns.send({ + environment: device.value.environment, + token: payload.token, + topic: device.value.bundleId, + expiration: Math.floor((notification.value.createdAt.getTime() + PUSH_EXPIRATION_MS) / 1000), + // A retry after a lost answer replaces the first copy on the phone. + collapseId: payload.notificationId, + payload: { + aps: { + alert: { title: payload.title, body: payload.body }, + sound: "default", + badge: Option.match(unread, { onNone: () => 0, onSome: (row) => Number(row.count) }), + "thread-id": payload.channelId, + }, + hazel: { + path: payload.path, + channelId: payload.channelId, + notificationId: payload.notificationId, + }, + }, + }) + const outcome = apnsOutcome(result) + yield* Effect.annotateCurrentSpan({ + "apns.environment": device.value.environment, + "apns.status": result.status, + "apns.reason": result.reason ?? "", + "push.outcome": outcome, + }) + switch (outcome) { + case "Delivered": + return + case "Unregistered": + // Not a registration that came in while this send was out. + yield* unregisterDevice(payload.token, sentAt) + return + case "ProviderToken": + return yield* new ApnsRetryableError({ status: result.status, reason: result.reason }) + case "Misconfigured": + yield* Effect.logError( + "APNs refused the provider token: check APNS_KEY_ID, APNS_TEAM_ID and APNS_PRIVATE_KEY", + ) + return + case "Retry": + return yield* new ApnsRetryableError({ status: result.status, reason: result.reason }) + case "Rejected": + yield* Effect.logWarning("APNs refused a push", { status: result.status, reason: result.reason }) + return + } +}) + +export const pushReceivers = { + "push.send": sendPush, +} satisfies EffectReceivers< + typeof PushEffects, + Database.Database | ApnsClient | Directory | CurrentEffectSender +> diff --git a/apps/native/scripts/build-ios.ts b/apps/native/scripts/build-ios.ts index 0cdb9345b..2db2e7b9b 100644 --- a/apps/native/scripts/build-ios.ts +++ b/apps/native/scripts/build-ios.ts @@ -22,7 +22,9 @@ const immediate = await Bun.build({ minify: true, }) if (!immediate.success) throw new AggregateError(immediate.logs, "src/ios-immediate.ts did not build") -const banner = await immediate.outputs[0]!.text() +const [immediateOutput] = immediate.outputs +if (immediateOutput === undefined) throw new Error("src/ios-immediate.ts built no output") +const banner = await immediateOutput.text() const result = await Bun.build({ banner, diff --git a/apps/web-foldkit/src/rpc.ts b/apps/web-foldkit/src/rpc.ts index f8a5451c9..91b49e5c0 100644 --- a/apps/web-foldkit/src/rpc.ts +++ b/apps/web-foldkit/src/rpc.ts @@ -23,6 +23,7 @@ import { OrganizationMemberRpcs, OrganizationRpcs, PinnedMessageRpcs, + PushDeviceRpcs, RssSubscriptionRpcs, TypingIndicatorRpcs, UserPresenceStatusRpcs, @@ -57,7 +58,7 @@ const AllRpcs = MessageRpcs.merge( BotRpcs, ConnectShareRpcs, InvitationRpcs, -).merge(ChatSyncRpcs, CustomerRpcs) +).merge(ChatSyncRpcs, CustomerRpcs, PushDeviceRpcs) const makeClient = RpcClient.make(AllRpcs, { flatten: true }) diff --git a/packages/db/migrations/directory/0004_push_device_tokens/migration.sql b/packages/db/migrations/directory/0004_push_device_tokens/migration.sql new file mode 100644 index 000000000..196bb738a --- /dev/null +++ b/packages/db/migrations/directory/0004_push_device_tokens/migration.sql @@ -0,0 +1,7 @@ +CREATE TABLE IF NOT EXISTS "push_device_tokens" ( + "token" text PRIMARY KEY NOT NULL, + "userId" text NOT NULL, + "updatedAt" text NOT NULL +); +--> statement-breakpoint +CREATE INDEX IF NOT EXISTS "push_device_tokens_user_id_idx" ON "push_device_tokens" ("userId"); diff --git a/packages/db/snapshots/directory.json b/packages/db/snapshots/directory.json index 1abf477e1..d2c167913 100644 --- a/packages/db/snapshots/directory.json +++ b/packages/db/snapshots/directory.json @@ -1,5 +1,5 @@ { - "migration": "0003_chat_sync_installations", + "migration": "0004_push_device_tokens", "entities": [ { "kind": "table", @@ -125,6 +125,13 @@ "columns": ["id"] } }, + { + "kind": "table", + "name": "push_device_tokens", + "primaryKey": { + "columns": ["token"] + } + }, { "kind": "table", "name": "users", @@ -1739,6 +1746,33 @@ "notNull": false, "default": null }, + { + "kind": "column", + "table": "push_device_tokens", + "name": "token", + "position": 0, + "type": "text", + "notNull": true, + "default": null + }, + { + "kind": "column", + "table": "push_device_tokens", + "name": "userId", + "position": 1, + "type": "text", + "notNull": true, + "default": null + }, + { + "kind": "column", + "table": "push_device_tokens", + "name": "updatedAt", + "position": 2, + "type": "text", + "notNull": true, + "default": null + }, { "kind": "column", "table": "users", @@ -2235,6 +2269,13 @@ "unique": true, "columns": ["\"tokenHash\""] }, + { + "kind": "index", + "table": "push_device_tokens", + "name": "push_device_tokens_user_id_idx", + "unique": false, + "columns": ["\"userId\""] + }, { "kind": "index", "table": "users", diff --git a/packages/db/snapshots/user.json b/packages/db/snapshots/user.json index 885981acd..b3ff46906 100644 --- a/packages/db/snapshots/user.json +++ b/packages/db/snapshots/user.json @@ -1,5 +1,5 @@ { - "migration": "0002_notification_levels", + "migration": "0003_push_devices", "entities": [ { "kind": "table", @@ -8,6 +8,13 @@ "columns": ["id"] } }, + { + "kind": "table", + "name": "push_devices", + "primaryKey": { + "columns": ["token"] + } + }, { "kind": "table", "name": "user_organizations", @@ -112,6 +119,51 @@ "notNull": true, "default": "false" }, + { + "kind": "column", + "table": "push_devices", + "name": "token", + "position": 0, + "type": "text", + "notNull": true, + "default": null + }, + { + "kind": "column", + "table": "push_devices", + "name": "environment", + "position": 1, + "type": "text", + "notNull": true, + "default": null + }, + { + "kind": "column", + "table": "push_devices", + "name": "bundleId", + "position": 2, + "type": "text", + "notNull": true, + "default": null + }, + { + "kind": "column", + "table": "push_devices", + "name": "createdAt", + "position": 3, + "type": "text", + "notNull": true, + "default": "(strftime('%Y-%m-%d %H:%M:%f', 'now'))" + }, + { + "kind": "column", + "table": "push_devices", + "name": "updatedAt", + "position": 4, + "type": "text", + "notNull": true, + "default": "(strftime('%Y-%m-%d %H:%M:%f', 'now'))" + }, { "kind": "column", "table": "user_organizations", diff --git a/packages/db/src/migrations.generated.ts b/packages/db/src/migrations.generated.ts index 1b8c416f1..023b2764b 100644 --- a/packages/db/src/migrations.generated.ts +++ b/packages/db/src/migrations.generated.ts @@ -89,6 +89,13 @@ export const generatedMigrations: Record +export type NewPushDevice = SQLite.InsertRowOf + +/** + * Directory (D1): which user a device token belongs to now. A phone signed into another account + * moves its token: the new owner's claim makes the previous owner's `UserObject` forget it + * (`apps/backend/src/worker/objects/user/push-ownership.ts`). + */ +export const pushDeviceTokensTable = SQLite.table("push_device_tokens", { + columns: { + token: SQLite.text, + userId: uuid(UserId), + /** When the owner registered it; an older claim never overwrites a newer one. */ + updatedAt: timestamptz, + }, + primaryKey: ["token"], + indexes: [SQLite.index("push_device_tokens_user_id_idx", ["userId"])], +}) +export type PushDeviceToken = SQLite.SelectRowOf diff --git a/packages/domain/src/rpc/index.ts b/packages/domain/src/rpc/index.ts index 29dd38ad5..8753c00d4 100644 --- a/packages/domain/src/rpc/index.ts +++ b/packages/domain/src/rpc/index.ts @@ -41,6 +41,8 @@ export * from "./organizations" export * as Organizations from "./organizations" export * from "./pinned-messages" export * as PinnedMessages from "./pinned-messages" +export * from "./push-devices" +export * as PushDevices from "./push-devices" export * from "./rss-subscriptions" export * as RssSubscriptions from "./rss-subscriptions" export * from "./typing-indicators" diff --git a/packages/domain/src/rpc/push-devices.ts b/packages/domain/src/rpc/push-devices.ts new file mode 100644 index 000000000..040b389fe --- /dev/null +++ b/packages/domain/src/rpc/push-devices.ts @@ -0,0 +1,41 @@ +import { Rpc, RpcGroup } from "effect/rpc" +import { Schema } from "effect" +import { InternalServerError, UnauthorizedError } from "../errors" +import { AuthMiddleware } from "./middleware" +import { RequiredScopes } from "../scopes/required-scopes" + +/** An APNs device token: the hex of the bytes iOS hands `didRegisterForRemoteNotifications`. */ +export const ApnsDeviceToken = Schema.String.check(Schema.isPattern(/^[0-9a-f]{16,200}$/)) + +/** Which APNs host the token belongs to: a debug or simulator build's `development`, else `production`. */ +export const ApnsEnvironment = Schema.Literals(["development", "production"]) +export type ApnsEnvironment = typeof ApnsEnvironment.Type + +/** + * Push devices: the current user's phones that get a system notification for each of the user's + * notifications (`UserObject`, `user/push.ts`). Both run in the user's own `UserObject`. + */ +export class PushDeviceRpcs extends RpcGroup.make( + /** Registers (or refreshes) this device's token; idempotent. */ + Rpc.make("pushDevice.register", { + payload: Schema.Struct({ + token: ApnsDeviceToken, + environment: ApnsEnvironment, + /** The app's bundle identifier, the APNs topic. */ + bundleId: Schema.String.check(Schema.isMaxLength(255)), + }), + success: Schema.Void, + error: Schema.Union([UnauthorizedError, InternalServerError]), + }) + .annotate(RequiredScopes, ["notifications:write"]) + .middleware(AuthMiddleware), + + /** Forgets this device's token (sign-out); a token not registered is no error. */ + Rpc.make("pushDevice.unregister", { + payload: Schema.Struct({ token: ApnsDeviceToken }), + success: Schema.Void, + error: Schema.Union([UnauthorizedError, InternalServerError]), + }) + .annotate(RequiredScopes, ["notifications:write"]) + .middleware(AuthMiddleware), +) {} diff --git a/packages/domain/src/scopes/action-names.ts b/packages/domain/src/scopes/action-names.ts index d2b8506d0..f1db73e33 100644 --- a/packages/domain/src/scopes/action-names.ts +++ b/packages/domain/src/scopes/action-names.ts @@ -16,6 +16,7 @@ import type { OrganizationMemberRpcs, OrganizationRpcs, PinnedMessageRpcs, + PushDeviceRpcs, RssSubscriptionRpcs, TypingIndicatorRpcs, UserPresenceStatusRpcs, @@ -49,6 +50,7 @@ export type RpcActionName = | ActionOf | ActionOf | ActionOf + | ActionOf | ActionOf | ActionOf | ActionOf diff --git a/packages/domain/src/scopes/all-scopes-map.ts b/packages/domain/src/scopes/all-scopes-map.ts index f27ffc3a8..f26518117 100644 --- a/packages/domain/src/scopes/all-scopes-map.ts +++ b/packages/domain/src/scopes/all-scopes-map.ts @@ -17,6 +17,7 @@ import { OrganizationMemberRpcs, OrganizationRpcs, PinnedMessageRpcs, + PushDeviceRpcs, RssSubscriptionRpcs, TypingIndicatorRpcs, UserPresenceStatusRpcs, @@ -42,6 +43,7 @@ export const RPC_SCOPE_MAP: ScopeMap = mergeScopeMaps( scopeMapFromRpcGroup(UserRpcs.requests), scopeMapFromRpcGroup(NotificationRpcs.requests), scopeMapFromRpcGroup(PinnedMessageRpcs.requests), + scopeMapFromRpcGroup(PushDeviceRpcs.requests), scopeMapFromRpcGroup(TypingIndicatorRpcs.requests), scopeMapFromRpcGroup(UserPresenceStatusRpcs.requests), scopeMapFromRpcGroup(AttachmentRpcs.requests), diff --git a/turbo.json b/turbo.json index 90db18eb0..76a9e5eb2 100644 --- a/turbo.json +++ b/turbo.json @@ -32,7 +32,10 @@ "OTEL_BASE_URL", "MAPLE_INGEST_KEY", "COMMIT_SHA", - "OPENROUTER_API_KEY" + "OPENROUTER_API_KEY", + "APNS_KEY_ID", + "APNS_TEAM_ID", + "APNS_PRIVATE_KEY" ], "globalPassThroughEnv": ["*"], "ui": "tui",