diff --git a/.gitignore b/.gitignore index c3aff6c7..3106583e 100644 --- a/.gitignore +++ b/.gitignore @@ -2,6 +2,7 @@ node_modules # Keep environment variables out of version control .env dist +.pnpm* .DS_Store /storage/** diff --git a/CHANGELOG.md b/CHANGELOG.md index 018ce257..c505f0f7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,14 +8,25 @@ The format follows semantic versioning principles. ### Added +- Added the `Broadcast` facade to `@arkstack/notifications`: a channel authorization registry (`Broadcast.channel('private-team.{teamId}', authorizer)`) used by every transport, `Broadcast.broadcast()` for arbitrary events, `Broadcast.handleAuth()` for custom auth routes, `Broadcast.clientConfig()`, and `Broadcast.routes()` registering `POST /realtime/auth` and `GET /realtime/config`. +- Added the `firebase-database` realtime transport (Firebase Realtime Database channels with per-user, time-limited grants checked by security rules, plus bounded `prune()`), the signal-only `polling` transport, and the `memory` transport for tests, on both server and client. +- Added device push alongside any streaming transport: `drivers.realtime.push: 'firebase'` pushes notifications sent to a user to their devices, with `push_tokens` for resolving tokens, `.push(false)` and `.devices(tokens)` per send, and the multicast outcome on `result.push`. +- Added self-hosted Pusher-protocol servers (soketi) through `transports.pusher.host` / `port` and the client's `wsHost` / `wsPort`. +- Added `RealtimeClient.onPoll()`, `RealtimeClient.deviceToken()` and `RealtimeClient.transportName` to `@arkstack/realtime`, with `createFirebaseDatabaseTransport`, `createPollingTransport`, `createMemoryTransport` and `emitMemoryEvent`. + ### Changed - `env()` now reads a value as a number only when the number says exactly what was written. `Number()` accepts hexadecimal, binary and octal literals, exponents, surrounding whitespace, and integers wider than a double can hold, and coercing those loses information silently. `'6379'` is still a number; `'0x1f'`, `'007'`, `'1e5'`, `' 12 '`, `'0b1010'`, `'Infinity'` and `'12345678901234567890'` are now the strings they were written as. ### Docs +- Documented realtime transports, channel authorization, device push, Firebase Realtime Database rules and pruning, polling, and testing with the memory transport in the notifications guide. + ### Fixed +- `UserNotificationCenter.send()` (and `Notification.db().broadcast()`) now broadcast on the user's channel. They previously sent to the user's push tokens as channels, so nothing was broadcast at all for users without tokens. Apps on the `firebase` transport still reach their users' devices without changes, because device push is on there unless `drivers.realtime.push` is `false`. +- `Notification.db().broadcast()` now returns the builder, so it chains as documented. +- The Pusher auth route registered by `registerAuthRoute()` now answers with HTTP 200 (pusher-js rejects any other status), registers on the H3 driver instead of the Express one, and authorizes through the channel registry. - Fixed `env()` destroying hexadecimal secrets. A `0x`-prefixed 32-byte key parses as a number, so it was silently replaced by a float keeping roughly its first thirteen digits, with nothing downstream able to tell. Private keys, hashes and identifiers read through `env()` now survive verbatim. ## [0.12.36] - 2026-06-20 diff --git a/docs/guide/notifications.md b/docs/guide/notifications.md index 95ffa5e5..7d0bc4b4 100644 --- a/docs/guide/notifications.md +++ b/docs/guide/notifications.md @@ -138,9 +138,7 @@ await Notification.db() .send('A new login was detected.', 'Security alert'); ``` -For this to work, your `User` model needs to have a `pushTokens` property/column, this can be a `string` and array of `string[]` or and object with a `token` property (`pushTokens.token`), an array of `pushTokens` object should also work (`{ pushTokens: { token: string } }[]`) - -The default behaviour for a missing `pushTokens` on the `User` model is no-op, so you don't have to worry about your app breaking. +The notification is stored, broadcast on the user's channel, and pushed to their devices when [device push](#device-push) is configured. Use `UserNotificationCenter` when you need to list, mark, or delete stored notifications: @@ -154,7 +152,7 @@ await UserNotificationCenter.delete(unread[0]); ## Realtime Notifications -The `realtime` driver broadcasts a notification to connected clients over [Pusher Channels](https://pusher.com/channels) or [Firebase Cloud Messaging](https://firebase.google.com/docs/cloud-messaging). Each user has their own channel (`${channel_prefix}${user.id}`, e.g. `user.7`). +The `realtime` driver broadcasts a notification to connected clients. Each user has their own channel (`${channel_prefix}${user.id}`, e.g. `user.7`). ```ts await Notification.realtime() @@ -167,7 +165,28 @@ await Notification.realtime() `.store()` is opt-in: when enabled the notification is written via `UserNotificationCenter` (giving it a real id and timestamps) **and** broadcast; otherwise it is broadcast only. Broadcast on an explicit channel with `.channel('team.updates')`, or change the client event name with `.event('alert')`. -`.channel()` (and `.recipient()`) also accept an **array**. For Pusher it fans out to multiple channels; for Firebase it is treated as a list of **device registration tokens** and delivered via a chunked multicast (500 tokens per call). The Firebase multicast returns `{ successCount, failureCount, invalidTokens }` — delete `invalidTokens` from your store, since FCM reports which are unregistered: +### Transports + +`drivers.realtime.transport` picks how events reach clients: + +| Transport | Delivers to | Server SDK | Client SDK | +| ------------------- | ------------------------------------ | ---------------- | ----------- | +| `pusher` | Pusher Channels, or soketi via `host` | `pusher` | `pusher-js` | +| `firebase` | FCM devices and topics | `firebase-admin` | `firebase` | +| `firebase-database` | Firebase Realtime Database channels | `firebase-admin` | `firebase` | +| `polling` | Nothing; clients poll on an interval | none | none | +| `memory` | An in-process buffer, for tests | none | none | + +The SDKs are optional peer dependencies; install only the ones you use: + +```sh +pnpm add pusher # pusher +pnpm add firebase-admin # firebase, firebase-database +``` + +To use a self-hosted Pusher-protocol server such as [soketi](https://soketi.app), set `host` and `port` (`PUSHER_HOST`, `PUSHER_PORT`) under `transports.pusher`. Clients receive the address through [the config route](#channel-authorization). + +`.channel()` (and `.recipient()`) also accept an **array**. For Pusher it fans out to multiple channels; for `firebase` it is treated as a list of **device registration tokens** and delivered via a chunked multicast (500 tokens per call). The multicast returns `{ successCount, failureCount, invalidTokens }`. Delete `invalidTokens` from your store, since FCM reports which are unregistered: ```ts await Notification.realtime({ transport: 'firebase' }) @@ -175,6 +194,67 @@ await Notification.realtime({ transport: 'firebase' }) .send('You have a new message'); ``` +### Broadcasting events + +A channel is just a channel. Use `Broadcast` to send your own event shapes over the configured transport, and pick them up with the client's `listen()`: + +```ts +import { Broadcast } from '@arkstack/notifications'; + +await Broadcast.broadcast( + [`private-order.${order.id}`, `private-team.${order.teamId}`], + 'order.updated', + { id: order.id, status: order.status }, +); +``` + +### Channel authorization + +Register who may join each private channel. `{name}` placeholders match one segment (they never span a dot) and are passed to the callback. The first matching rule decides, unmatched channels are denied, and the per-user notification channel is allowed for its own user unless you register a rule for it. + +```ts +// src/routes/realtime.ts +import { Broadcast } from '@arkstack/notifications'; + +Broadcast.channel('private-team.{teamId}', async (user, { teamId }) => { + return await TeamMember.where({ teamId, userId: user.id }).exists(); +}); + +await Broadcast.routes(); // POST /realtime/auth, GET /realtime/config +``` + +`Broadcast.routes()` registers two routes behind the framework `auth` middleware (pass `{ prefix, middleware }` to change them): + +- `POST /realtime/auth` authorizes the requested channel and answers in the active transport's terms: a Pusher signature, or a Firebase custom token for `firebase-database`. +- `GET /realtime/config` returns the public settings a client is created from. Secrets are never included. + +When a channel is joined by something other than the logged-in user, such as a guest holding a scoped token, call `Broadcast.handleAuth(subject, { socket_id, channel_name }, { uid })` from your own route. Rules receive whatever `subject` you pass. + +### Device push + +Set `drivers.realtime.push` to `'firebase'` to also push every notification sent to a user to their registered devices over FCM, whatever the streaming transport is. When `push` is unset it is on only if the transport is `firebase`; set it to `false` to turn it off there. Tokens are read from `user.pushTokens` (a string, `string[]`, `{ token }` or `{ token }[]`), or from your own resolver: + +```ts +realtime: { + transport: 'pusher', + push: 'firebase', + push_tokens: async (user) => + (await UserPushToken.where({ userId: user.id, isActive: true }).get()).all(), +}, +``` + +The send result carries the multicast outcome, so prune dead tokens there: + +```ts +const result = await Notification.realtime().recipient(user).send('Your payout has settled.'); + +if (result.push?.invalidTokens.length) { + await UserPushToken.whereIn('token', result.push.invalidTokens).delete(); +} +``` + +Skip the push for one send with `.push(false)`, or target specific devices with `.devices(tokens)`. On the client, `realtime.deviceToken()` returns the token to register; it needs `firebase.vapidKey` (`FIREBASE_VAPID_KEY`) and a `firebase-messaging-sw.js` service worker for background messages. + ### Delivery options FCM sends a data message at normal priority, and Android's Doze and App Standby may hold it until the next maintenance window. `.priority('high')` exempts it, and pairs naturally with a short TTL — FCM's own default is four weeks: @@ -226,12 +306,7 @@ import { supersede } from '@arkstack/realtime'; client.subscribe(channel, (n) => setItems((items) => supersede(items, n, 50))); ``` -Configure the transport in `src/config/notifications.ts` (`drivers.realtime`) and its credentials under `transports.pusher` / `transports.firebase`. The `pusher` / `firebase-admin` SDKs are optional, install only the one you use: - -```sh -pnpm add pusher # Pusher transport -pnpm add firebase-admin # Firebase transport -``` +Configure the transport in `src/config/notifications.ts` (`drivers.realtime`) and its credentials under `transports.*`. **Bringing your own transport.** `driverFactory` replaces the built-ins with anything implementing `RealtimeDriver` — a backend this package does not ship, or a fake in a test. Set it per send, or under `drivers.realtime` for the whole app; it takes precedence over `transport`. Do connection setup lazily inside the driver, as the bundled ones do. @@ -241,29 +316,90 @@ Notification.realtime({ driverFactory: () => new MyApnsDriver() }); `broadcast()` also accepts payloads that are not notifications, so an application can push its own event shapes over the same channel and pick them up with the client's `listen()`. +### Firebase Realtime Database + +The `firebase-database` transport writes each event to `{root}/{channel}/{key}` (default root `arkstack/channels`). When a client is authorized for a channel, the server writes a grant to `arkstack/grants/{uid}/{channel}` that expires after `grant_ttl` seconds (default 3600) and returns a custom token to sign in with. Deploy security rules that check the grant, and index `at`: + +```json +{ + "rules": { + "arkstack": { + "channels": { + "$channel": { + ".read": "auth != null && root.child('arkstack/grants/' + auth.uid + '/' + $channel).val() > now", + ".indexOn": ["at"] + } + } + } + } +} +``` + +Everything else stays unreadable, so the grants and the event index are never exposed. Channel names are stored URL-encoded with `.` written as `%2E`, so rules see the same key the server wrote. The `uid` must be usable as a database key; it defaults to the subscriber's `id`. + +Events are not deleted automatically. Schedule a prune; each call removes at most `limit` events: + +```ts +import { Broadcast, FirebaseDatabaseRealtimeDriver } from '@arkstack/notifications'; + +const driver = Broadcast.make('firebase-database') as FirebaseDatabaseRealtimeDriver; + +while ((await driver.prune(24 * 3600, 500)) > 0); +``` + +### Testing + +Set the transport to `memory` and read what was sent. The buffer lives on `globalThis`, so broadcasts made by code loaded separately from your tests are recorded too: + +```ts +import { MemoryRealtimeDriver } from '@arkstack/notifications'; + +beforeEach(() => MemoryRealtimeDriver.clear()); + +expect(MemoryRealtimeDriver.broadcasts()).toContainEqual( + expect.objectContaining({ channel: 'user.7', event: 'notification' }), +); +``` + **Firebase credentials** can be provided two ways. Point `admin_sdk_path` (`FIREBASE_ADMINSDK`, default `firebase-adminsdk.json`, resolved from the project root) at a downloaded service-account JSON file; if that file is absent, the driver falls back to the inline `project_id` / `client_email` / `private_key` values (`FIREBASE_PROJECT_ID` / `FIREBASE_CLIENT_EMAIL` / `FIREBASE_PRIVATE_KEY`). `app_name` (`FIREBASE_APP_NAME`, default your `APP_NAME`) names the Firebase Admin app instance so repeated broadcasts reuse it. ### Consuming on the client -Install `@arkstack/realtime` in your front-end and the matching client SDK (`pusher-js` or `firebase`): +Install `@arkstack/realtime` in your front-end and the matching client SDK (`pusher-js` or `firebase`). The simplest setup creates the client from the server's config route and adds your API's auth header: ```ts import { createRealtime } from '@arkstack/realtime'; +const headers = { Authorization: `Bearer ${token}` }; +const config = await fetch('/api/realtime/config', { headers }).then((r) => r.json()); + const realtime = createRealtime({ - transport: 'pusher', - pusher: { - key: import.meta.env.VITE_PUSHER_KEY, - cluster: 'mt1', - authEndpoint: '/broadcasting/auth', + ...config, + pusher: config.pusher && { ...config.pusher, apiBase: '/api', auth: { headers } }, + 'firebase-database': config['firebase-database'] && { + ...config['firebase-database'], + apiBase: '/api', + auth: { headers }, }, }); const unsubscribe = await realtime.forUser(user.id, (notification) => { console.log(notification.title, notification.description); }); + +// Your own events +await realtime.listen(`private-order.${order.id}`, 'order.updated', refresh); +``` + +Under `polling` nothing arrives on a channel. `onPoll()` runs a handler on the advertised interval only while polling, so one code path serves every transport: + +```ts +await realtime.listen(channel, 'order.updated', refresh); +await realtime.onPoll(refresh); ``` +For tests, use `transport: 'memory'` and feed subscribers with `emitMemoryEvent(channel, event, payload)`. + React and Vue bindings accumulate notifications for you (newest first): ::: code-group diff --git a/packages/notifications/src/Broadcast.ts b/packages/notifications/src/Broadcast.ts new file mode 100644 index 00000000..3eb78580 --- /dev/null +++ b/packages/notifications/src/Broadcast.ts @@ -0,0 +1,360 @@ +import type { + ChannelAuthRequest, + ChannelAuthorizer, + PushResult, + PushTokenLike, + RealtimeBroadcastPayload, + RealtimeClientConfig, + RealtimeDeliveryOptions, + RealtimeDriverName, + RealtimeDriverOptions, +} from './types' + +import { FirebaseDatabaseRealtimeDriver } from './drivers/realtime/FirebaseDatabaseRealtimeDriver' +import { FirebaseRealtimeDriver } from './drivers/realtime/FirebaseRealtimeDriver' +import { MemoryRealtimeDriver } from './drivers/realtime/MemoryRealtimeDriver' +import { PollingRealtimeDriver } from './drivers/realtime/PollingRealtimeDriver' +import { PusherRealtimeDriver } from './drivers/realtime/PusherRealtimeDriver' +import type { RealtimeDriver } from './Contracts/RealtimeDriver' +import { RequestException } from '@arkstack/common' +import { configure } from './config' + +type RealtimeSettings = { + transport?: RealtimeDriverName + push?: 'firebase' | false + push_tokens?: (user: any) => PushTokenLike[] | Promise + channel_prefix?: string + event?: string + driverFactory?: () => RealtimeDriver +} + +type ChannelRule = { + pattern: string + regex: RegExp + keys: string[] + authorizer: ChannelAuthorizer +} + +export type BroadcastRouteOptions = { + /** Route prefix (default `/realtime`). */ + prefix?: string + /** Extra middleware run before the framework `auth` middleware. */ + middleware?: unknown | unknown[] +} + +/** + * The server side of realtime: builds transports from config, broadcasts + * arbitrary events, and authorizes channel subscriptions against registered + * channel rules for every transport. + */ +export class Broadcast { + private static rules: ChannelRule[] = [] + + private static settings(): RealtimeSettings { + return (configure('drivers.realtime', {}) ?? {}) as RealtimeSettings + } + + /** + * The configured streaming transport name. + * + * @returns + */ + static transport(): RealtimeDriverName { + return this.settings().transport ?? 'pusher' + } + + /** + * The prefix of the per-user notification channel. + * + * @returns + */ + static channelPrefix(): string { + return this.settings().channel_prefix ?? 'user.' + } + + /** + * Register who may subscribe to channels matching `pattern`. `{name}` + * placeholders match one dot-free segment and are passed to the authorizer. + * A later registration of the same pattern replaces the earlier one. + * + * @param pattern e.g. `private-team.{teamId}` + * @param authorizer + * @returns + */ + static channel(pattern: string, authorizer: ChannelAuthorizer): typeof Broadcast { + const keys: string[] = [] + const source = pattern + .split(/(\{[^}]+\})/) + .map((part) => { + const key = /^\{([^}]+)\}$/.exec(part)?.[1] + + if (key) { + keys.push(key) + + return '([^.]+)' + } + + return part.replace(/[.*+?^${}()|[\]\\]/g, '\\$&') + }) + .join('') + + this.rules = this.rules.filter((rule) => rule.pattern !== pattern) + this.rules.push({ pattern, regex: new RegExp(`^${source}$`), keys, authorizer }) + + return this + } + + /** + * Remove every registered channel rule. + */ + static forgetChannels(): void { + this.rules = [] + } + + /** + * Whether `subject` may subscribe to `channel`. The first matching rule + * decides. The per-user notification channel is allowed for its own user + * unless a rule overrides it. Unmatched channels are denied. + * + * @param subject + * @param channel + * @returns + */ + static async authorize(subject: unknown, channel: string): Promise { + for (const rule of this.rules) { + const match = rule.regex.exec(channel) + + if (match) { + const params = Object.fromEntries(rule.keys.map((key, index) => [key, match[index + 1]])) + + return Boolean(await rule.authorizer(subject, params)) + } + } + + const id = (subject as { id?: unknown } | null)?.id + + return id !== undefined && id !== null && channel === `${this.channelPrefix()}${id}` + } + + /** + * Build a transport. Uses the configured `driverFactory` when there is one, + * otherwise the named or configured built-in. + * + * @param options + * @returns + */ + static driver(options: RealtimeDriverOptions = {}): RealtimeDriver { + const factory = options.driverFactory ?? this.settings().driverFactory + + if (factory) { + return factory() + } + + const transport = options.transport ?? this.transport() + + return this.make(transport, (options as Record)[transport] as Record) + } + + /** + * Build a named built-in transport from its `transports.*` config. + * + * @param transport + * @param overrides + * @returns + */ + static make(transport: RealtimeDriverName, overrides: Record = {}): RealtimeDriver { + const config = { + ...(configure(`transports.${transport}` as never, {}) as Record), + ...overrides, + } + + switch (transport) { + case 'firebase': + return new FirebaseRealtimeDriver(config) + case 'firebase-database': + return new FirebaseDatabaseRealtimeDriver(config) + case 'polling': + return new PollingRealtimeDriver(config) + case 'memory': + return new MemoryRealtimeDriver() + case 'pusher': + return new PusherRealtimeDriver(config) + default: + throw new Error(`Realtime: unknown transport "${transport}"`) + } + } + + /** + * Broadcast an arbitrary event on one or more channels over the configured + * transport. + * + * @param channel + * @param event + * @param payload + * @param delivery + * @returns + */ + static async broadcast( + channel: string | string[], + event: string, + payload: RealtimeBroadcastPayload, + delivery?: RealtimeDeliveryOptions, + ): Promise { + if (Array.isArray(channel) && channel.length === 0) { + return + } + + return await this.driver().broadcast(channel, event, payload, delivery) + } + + /** + * The device push transport, or null when push is off. When `push` is unset, + * apps streaming over `firebase` keep pushing to devices as they always have. + * + * @returns + */ + static pushDriver(): FirebaseRealtimeDriver | null { + const { push } = this.settings() + const enabled = push === 'firebase' || (push === undefined && this.transport() === 'firebase') + + return enabled ? this.make('firebase') as FirebaseRealtimeDriver : null + } + + /** + * A user's device tokens, from the configured `push_tokens` resolver or the + * user's `pushTokens` property. + * + * @param user + * @returns + */ + static async pushTokens(user: unknown): Promise { + const resolver = this.settings().push_tokens + const tokens = resolver + ? await resolver(user) + : (user as { pushTokens?: PushTokenLike | PushTokenLike[] } | null)?.pushTokens + + if (!tokens) return [] + + return (Array.isArray(tokens) ? tokens : [tokens]) + .map((token) => typeof token === 'string' ? token : token?.token) + .filter((token): token is string => !!token) + } + + /** + * Push an event to device tokens over the push transport. Returns null when + * push is off or there are no tokens. + * + * @param tokens + * @param event + * @param payload + * @param delivery + * @returns + */ + static async push( + tokens: string[], + event: string, + payload: RealtimeBroadcastPayload, + delivery?: RealtimeDeliveryOptions, + ): Promise { + const driver = this.pushDriver() + + if (!driver || tokens.length === 0) { + return null + } + + return await driver.broadcast(tokens, event, payload, delivery) as PushResult + } + + /** + * Authorize `subject` for the requested channel and return what the active + * transport needs: a Pusher signature, a Firebase custom token, or nothing. + * + * @param subject Usually the authenticated user. + * @param request The client's `socket_id` and `channel_name`. + * @param options `uid` overrides the Firebase uid (defaults to `subject.id`). + * @returns + */ + static async handleAuth( + subject: unknown, + request: ChannelAuthRequest, + options: { uid?: string, driver?: RealtimeDriver } = {}, + ): Promise { + const channel = request.channel_name + + RequestException.abortIf(!channel, 'Missing channel_name', 422) + RequestException.abortIf(!(await this.authorize(subject, channel!)), 'Channel access denied', 403) + + const driver = options.driver ?? this.driver() + + if (driver instanceof FirebaseDatabaseRealtimeDriver) { + return await driver.auth(options.uid ?? String((subject as { id?: unknown })?.id ?? ''), channel!) + } + + RequestException.abortIf(driver instanceof PusherRealtimeDriver && !request.socket_id, 'Missing socket_id', 422) + + return await driver.auth(request.socket_id ?? '', channel!) + } + + /** + * Public settings a `@arkstack/realtime` client is created from. + * + * @returns + */ + static clientConfig(): RealtimeClientConfig { + const transport = this.transport() + const settings = this.settings() + const config: RealtimeClientConfig = { + transport, + event: settings.event ?? 'notification', + channelPrefix: this.channelPrefix(), + } + + config[transport] = this.driver().clientConfig?.() ?? {} + + if (settings.push === 'firebase' && transport !== 'firebase') { + config.firebase = this.make('firebase').clientConfig?.() ?? {} + } + + return config + } + + /** + * Register `POST {prefix}/auth` and `GET {prefix}/config`, both behind the + * framework `auth` middleware. + * + * @param options + */ + static async routes(options: BroadcastRouteOptions = {}): Promise { + const prefix = options.prefix ?? '/realtime' + const drivers = ['@arkstack/driver-express', '@arkstack/driver-h3'] + + for (const path of drivers) { + try { + const { Router } = await import(path) + const { auth } = await import(`${path}/middlewares`) + const middleware = [options.middleware ?? []].flat().concat(auth) + + Router.post(`${prefix}/auth`, async ({ clearRequest, req }: any) => { + const result = await this.handleAuth(req?.user ?? clearRequest.user, { + socket_id: clearRequest.input('socket_id'), + channel_name: clearRequest.input('channel_name'), + }) + + // pusher-js only accepts a 200 from the authorizer. + return new Response(JSON.stringify(result ?? {}), { + status: 200, + headers: { 'content-type': 'application/json' }, + }) + }).middleware(middleware) + + Router.get(`${prefix}/config`, () => this.clientConfig()).middleware(middleware) + + return + } catch { + continue + } + } + + throw new Error('Realtime: no supported HTTP driver is installed to register routes on') + } +} diff --git a/packages/notifications/src/Contracts/RealtimeDriver.ts b/packages/notifications/src/Contracts/RealtimeDriver.ts index a8856638..8a68c4e2 100644 --- a/packages/notifications/src/Contracts/RealtimeDriver.ts +++ b/packages/notifications/src/Contracts/RealtimeDriver.ts @@ -1,6 +1,9 @@ import type { RealtimeBroadcastPayload, RealtimeDeliveryOptions, RealtimeDriverName } from '../types' +import type { FirebaseDatabaseRealtimeDriver } from '../drivers/realtime/FirebaseDatabaseRealtimeDriver' import type { FirebaseRealtimeDriver } from '../drivers/realtime/FirebaseRealtimeDriver' +import type { MemoryRealtimeDriver } from '../drivers/realtime/MemoryRealtimeDriver' +import type { PollingRealtimeDriver } from '../drivers/realtime/PollingRealtimeDriver' import type { PusherRealtimeDriver } from '../drivers/realtime/PusherRealtimeDriver' /** @@ -22,6 +25,10 @@ export interface RealtimeDriver { payload: RealtimeBroadcastPayload, delivery?: RealtimeDeliveryOptions, ): Promise + /** + * Sign or grant access to an already authorized channel. `socketId` is the + * Pusher socket id, or the subscriber's uid for transports without sockets. + */ auth( socketId: string, channel: string, @@ -32,10 +39,20 @@ export interface RealtimeDriver { middleware?: unknown | unknown[], channelPrefix?: string, ): Promise + /** + * Public connection settings a browser client needs. Must never include secrets. + */ + clientConfig?(): Record } export type RealtimeNotificationDriver = T extends 'firebase' ? FirebaseRealtimeDriver + : T extends 'firebase-database' + ? FirebaseDatabaseRealtimeDriver + : T extends 'polling' + ? PollingRealtimeDriver + : T extends 'memory' + ? MemoryRealtimeDriver : T extends 'pusher' ? PusherRealtimeDriver : RealtimeDriver diff --git a/packages/notifications/src/UserNotificationCenter.ts b/packages/notifications/src/UserNotificationCenter.ts index 85198380..a29910f1 100644 --- a/packages/notifications/src/UserNotificationCenter.ts +++ b/packages/notifications/src/UserNotificationCenter.ts @@ -5,7 +5,6 @@ import type { DbNotificationPayload } from './types' import { Notification } from './Notification' import type { User } from '@app/models/User' import type { UserNotification } from '@app/models/UserNotification' -import { UserPushToken } from './Contracts/UserPushToken' export class UserNotificationCenter { private static async getModel() { @@ -13,31 +12,27 @@ export class UserNotificationCenter { } /** - * Create a database notification then broadcast it using the configured realtime driver - * - * @param user - * @param payload + * Store a notification, broadcast it on the user's channel, and push it to + * their devices when `drivers.realtime.push` is configured. + * + * @param user + * @param payload + * @param devices Push to these tokens instead of the user's own. */ - static async send(user: User, payload: DbNotificationPayload, channel: string[] = []) { - return await Notification.realtime() + static async send(user: User, payload: DbNotificationPayload, devices?: string[]) { + const notification = Notification.realtime() .store() - .channel(this.resolvePushTokens(user.pushTokens ?? channel)) + .recipient(user) .meta(payload.meta) .type(payload.type) .subject(payload.title) .action(payload.actionText, payload.actionLink) - .send(payload.description, payload.title, undefined, payload) - - } - - private static resolvePushTokens(pushTokens?: string | string | UserPushToken | UserPushToken[]) { - if (!pushTokens) return [] - if (Array.isArray(pushTokens)) { - return pushTokens.map(e => typeof e === 'string' ? e : e.token) + if (devices?.length) { + notification.devices(devices) } - return [typeof pushTokens === 'string' ? pushTokens : pushTokens.token] + return await notification.send(payload.description, payload.title, undefined, payload) } /** diff --git a/packages/notifications/src/drivers/DbNotification.ts b/packages/notifications/src/drivers/DbNotification.ts index 8f8bfc6c..a0878bd6 100644 --- a/packages/notifications/src/drivers/DbNotification.ts +++ b/packages/notifications/src/drivers/DbNotification.ts @@ -55,9 +55,12 @@ export class DbNotification extends NotificationContract { * Broadcast the notification using the default realtime driver. * * @param realtime Set to false to disable realtime broadcast + * @returns */ - broadcast(realtime = true) { + broadcast(realtime = true): this { this.realtime = realtime + + return this } async create(user: User, payload: DbNotificationPayload) { diff --git a/packages/notifications/src/drivers/RealtimeNotification.ts b/packages/notifications/src/drivers/RealtimeNotification.ts index f6237487..6494bbfe 100644 --- a/packages/notifications/src/drivers/RealtimeNotification.ts +++ b/packages/notifications/src/drivers/RealtimeNotification.ts @@ -10,10 +10,9 @@ import type { RealtimeNotificationPayload, } from '../types' -import { FirebaseRealtimeDriver } from './realtime/FirebaseRealtimeDriver' +import { Broadcast } from '../Broadcast' import { NotificationContract } from '../Contracts/NotificationContract' -import { PusherRealtimeDriver } from './realtime/PusherRealtimeDriver' -import type { RealtimeDriver, RealtimeNotificationDriver } from '../Contracts/RealtimeDriver' +import type { RealtimeNotificationDriver } from '../Contracts/RealtimeDriver' import type { User } from '@app/models/User' import { UserNotificationCenter } from '../UserNotificationCenter' import { configure } from '../config' @@ -40,6 +39,8 @@ export class RealtimeNotification private shouldStore: boolean private tagName?: string private isRetraction = false + private shouldPush?: boolean + private deviceTokens?: string[] private deliveryOptions: RealtimeDeliveryOptions private payload: Partial = {} @@ -47,32 +48,17 @@ export class RealtimeNotification super() const driverConfig = configure('drivers.realtime', {}) as { - transport?: RealtimeDriverName - channel_prefix?: string event?: string store?: boolean delivery?: RealtimeDeliveryOptions - driverFactory?: () => RealtimeDriver } - const transport = options.transport ?? driverConfig?.transport ?? 'pusher' - const transportConfig = configure(`transports.${transport}` as never, {}) as Record this.channelName = options.channel this.eventName = options.event ?? driverConfig?.event ?? 'notification' - this.channelPrefix = driverConfig?.channel_prefix ?? 'user.' + this.channelPrefix = Broadcast.channelPrefix() this.shouldStore = options.store ?? driverConfig?.store ?? false this.deliveryOptions = { ...driverConfig?.delivery, ...options.delivery } - - // A supplied driver wins over `transport`: naming one of the built-ins is - // meaningless when the caller is bringing their own. - const factory = options.driverFactory ?? driverConfig?.driverFactory - - this.driver = (factory - ? factory() - : transport === 'firebase' - ? new FirebaseRealtimeDriver({ ...transportConfig, ...options.firebase }) - : new PusherRealtimeDriver({ ...transportConfig, ...options.pusher }) - ) as RealtimeNotificationDriver + this.driver = Broadcast.driver(options as RealtimeDriverOptions) as RealtimeNotificationDriver } from(_: string): this { @@ -93,12 +79,7 @@ export class RealtimeNotification * @returns */ transport(transport: T, options?: RealtimeDriverOptions[T]): this { - const transportConfig = configure(`transports.${transport}` as never, {}) as Record - - this.driver = (transport === 'firebase' - ? new FirebaseRealtimeDriver({ ...transportConfig, ...options as any }) - : new PusherRealtimeDriver({ ...transportConfig, ...options as any }) - ) as RealtimeNotificationDriver + this.driver = Broadcast.make(transport, options as Record) as RealtimeNotificationDriver return this } @@ -291,6 +272,41 @@ export class RealtimeNotification return this } + /** + * Also push this notification to the recipient's devices over the + * `drivers.realtime.push` transport. On by default for a `User` recipient + * when push is configured; pass `false` to skip it for this send. + * + * @param push + * @returns + */ + push(push = true): this { + this.shouldPush = push + + return this + } + + /** + * Push to these device tokens instead of the recipient's own. + * + * @param tokens + * @returns + */ + devices(tokens: string[]): this { + this.deviceTokens = tokens + this.shouldPush ??= true + + return this + } + + private async resolveDevices(): Promise { + if (this.shouldPush === false) return [] + if (this.deviceTokens) return this.deviceTokens + if (this.user) return await Broadcast.pushTokens(this.user) + + return [] + } + private resolveChannel(): string | string[] { if (this.channelName !== undefined) { return this.channelName @@ -358,6 +374,16 @@ export class RealtimeNotification if (channel && channel.length > 0) await this.driver.broadcast(channel, this.eventName, payload, this.deliveryOptions) - return { channel, event: this.eventName, payload, stored } + const devices = await this.resolveDevices() + const push = await Broadcast.push(devices, this.eventName, payload, this.deliveryOptions) + + return { + channel, + event: this.eventName, + payload, + stored, + ...(push ? { push } : {} + ) + } } } diff --git a/packages/notifications/src/drivers/realtime/FirebaseDatabaseRealtimeDriver.ts b/packages/notifications/src/drivers/realtime/FirebaseDatabaseRealtimeDriver.ts new file mode 100644 index 00000000..a26efaca --- /dev/null +++ b/packages/notifications/src/drivers/realtime/FirebaseDatabaseRealtimeDriver.ts @@ -0,0 +1,214 @@ +import type { FirebaseDatabaseTransportConfig, RealtimeBroadcastPayload, RealtimeDeliveryOptions } from '../../types' +import { firebaseAdminApp, firebaseWebConfig } from './firebaseAdmin' + +import type { RealtimeDriver } from '../../Contracts/RealtimeDriver' +import { env } from '@arkstack/common' + +/** The slice of the `firebase-admin` database API this driver uses. */ +interface DatabaseSnapshot { + forEach(callback: (child: { key: string | null, val(): any }) => boolean | void): boolean +} + +interface DatabaseQuery { + endAt(value: number): DatabaseQuery + limitToFirst(limit: number): DatabaseQuery + once(event: 'value'): Promise +} + +interface DatabaseReference extends DatabaseQuery { + push(): { key: string | null } + update(values: Record): Promise + orderByChild(path: string): DatabaseQuery +} + +interface Database { + ref(path?: string): DatabaseReference +} + +interface FirebaseAuth { + createCustomToken(uid: string, claims?: Record): Promise +} + +/** Characters Realtime Database keys cannot contain. */ +const FORBIDDEN_KEY = /[.#$[\]/]/ + +/** + * Encode a channel name into a single Realtime Database key. + * + * @param channel + * @returns + */ +export const firebaseChannelKey = (channel: string): string => encodeURIComponent(channel).replace(/\./g, '%2E') + +/** + * Broadcasts over [Firebase Realtime Database](https://firebase.google.com/docs/database). + * + * Each event is written to `{root}/{channel}/{key}` as `{ event, payload, at }`, + * and indexed under `{index_root}/{key}` so {@link prune} can clear old events in + * bounded batches. Clients sign in with a custom token from {@link auth}, which + * also writes a time-limited grant to `{grants_root}/{uid}/{channel}` that the + * database security rules check before allowing a read. + * + * `firebase-admin` is an optional peer dependency, imported lazily. + */ +export class FirebaseDatabaseRealtimeDriver implements RealtimeDriver { + private servicesPromise?: Promise<{ database: Database, auth: FirebaseAuth }> + + constructor(private options: FirebaseDatabaseTransportConfig = {}) { } + + private get root(): string { + return this.options.root ?? 'arkstack/channels' + } + + private get indexRoot(): string { + return `${this.root}-index` + } + + private get grantsRoot(): string { + return this.options.grants_root ?? 'arkstack/grants' + } + + private get databaseURL(): string { + return this.options.database_url ?? env('FIREBASE_DATABASE_URL', '') + } + + private services(): Promise<{ database: Database, auth: FirebaseAuth }> { + const databaseSpecifier = 'firebase-admin/database' + const authSpecifier = 'firebase-admin/auth' + + this.servicesPromise ??= (async () => { + const [instance, database, auth] = await Promise.all([ + firebaseAdminApp(this.options, '-database', { databaseURL: this.databaseURL }), + import(databaseSpecifier), + import(authSpecifier), + ]) + + return { database: database.getDatabase(instance), auth: auth.getAuth(instance) } + })() + + return this.servicesPromise + } + + /** + * Write the event to every channel in one multi-path update. + * + * `delivery` is accepted for contract parity and ignored. + * + * @param channel + * @param event + * @param payload + * @param _delivery + * @returns + */ + async broadcast( + channel: string | string[], + event: string, + payload: RealtimeBroadcastPayload, + _delivery?: RealtimeDeliveryOptions, + ): Promise { + const { database } = await this.services() + const channels = Array.isArray(channel) ? channel : [channel] + const at = { '.sv': 'timestamp' } + const updates: Record = {} + const keys: string[] = [] + + for (const name of channels) { + const key = database.ref(this.root).push().key! + const channelKey = firebaseChannelKey(name) + + updates[`${this.root}/${channelKey}/${key}`] = { event, payload, at } + updates[`${this.indexRoot}/${key}`] = { channel: channelKey, at } + keys.push(key) + } + + await database.ref().update(updates) + + return keys + } + + /** + * Grant `uid` read access to an already authorized channel and mint the + * custom token the client signs in with. + * + * @param uid + * @param channel + * @returns + */ + async auth(uid: string, channel: string): Promise<{ token: string, expires_at: number }> { + if (!uid || FORBIDDEN_KEY.test(uid)) { + throw new Error(`Realtime: "${uid}" cannot be used as a Firebase uid`) + } + + const { database, auth } = await this.services() + const expiresAt = Date.now() + (this.options.grant_ttl ?? 3600) * 1000 + + await database.ref().update({ + [`${this.grantsRoot}/${uid}/${firebaseChannelKey(channel)}`]: expiresAt, + }) + + return { token: await auth.createCustomToken(uid), expires_at: expiresAt } + } + + /** + * Delete events older than `maxAgeSeconds`, at most `limit` per call. Call it + * from a scheduled job until it returns 0. + * + * @param maxAgeSeconds + * @param limit + * @returns The number of events deleted. + */ + async prune(maxAgeSeconds: number, limit = 500): Promise { + const { database } = await this.services() + const cutoff = Date.now() - maxAgeSeconds * 1000 + const snapshot = await database.ref(this.indexRoot) + .orderByChild('at') + .endAt(cutoff) + .limitToFirst(limit) + .once('value') + + const updates: Record = {} + + snapshot.forEach((child) => { + const entry = child.val() as { channel?: string } | null + + if (child.key && entry?.channel) { + updates[`${this.root}/${entry.channel}/${child.key}`] = null + updates[`${this.indexRoot}/${child.key}`] = null + } + }) + + const deleted = Object.keys(updates).length / 2 + + if (deleted > 0) { + await database.ref().update(updates) + } + + return deleted + } + + /** + * Public settings a browser client needs to read channels. + * + * @returns + */ + clientConfig(): Record { + return { + ...firebaseWebConfig(this.options), + databaseURL: this.databaseURL, + root: this.root, + } + } + + /** + * Use `Broadcast.routes()`, which authorizes channels through the registry. + * + * @param _authEndpoint + * @param _middleware + * @param _channelPrefix + */ + async registerAuthRoute( + _authEndpoint?: string, + _middleware?: unknown | unknown[], + _channelPrefix?: string, + ): Promise { } +} diff --git a/packages/notifications/src/drivers/realtime/FirebaseRealtimeDriver.ts b/packages/notifications/src/drivers/realtime/FirebaseRealtimeDriver.ts index 089b36ad..63f10e4c 100644 --- a/packages/notifications/src/drivers/realtime/FirebaseRealtimeDriver.ts +++ b/packages/notifications/src/drivers/realtime/FirebaseRealtimeDriver.ts @@ -1,9 +1,7 @@ import type { FirebaseTransportConfig, RealtimeBroadcastPayload, RealtimeDeliveryOptions } from '../../types' -import { existsSync, readFileSync } from 'node:fs' import type { RealtimeDriver } from '../../Contracts/RealtimeDriver' -import { env } from '@arkstack/common' -import path from 'node:path' +import { firebaseAdminApp, firebaseWebConfig } from './firebaseAdmin' /** The slice of the `firebase-admin` messaging API this driver uses. */ interface FirebaseSendResponse { @@ -72,43 +70,13 @@ export class FirebaseRealtimeDriver implements RealtimeDriver { constructor(private options: FirebaseTransportConfig = {}) { } private messaging(): Promise { - const appSpecifier = 'firebase-admin/app' - const messagingSpecifier = 'firebase-admin/messaging' + const specifier = 'firebase-admin/messaging' this.messagingPromise ??= (async () => { - const [app, messaging] = await Promise.all([ - import(appSpecifier), - import(messagingSpecifier), - ]).catch(() => { - throw new Error( - 'The "firebase-admin" package is required for the Firebase realtime transport. Install it with `npm i firebase-admin`.', - ) - }) - - const adminsdk = path.join( - process.cwd(), - this.options.admin_sdk_path ?? env('FIREBASE_ADMINSDK', 'firebase-adminsdk.json') - ) - - let serviceAccount: Record | undefined - try { - if (existsSync(adminsdk)) - serviceAccount = JSON.parse(readFileSync(adminsdk, { encoding: 'utf-8' })) - } catch {/** */ } - - const credential = app.cert(serviceAccount ?? { - projectId: this.options.project_id ?? env('FIREBASE_PROJECT_ID', ''), - clientEmail: this.options.client_email ?? env('FIREBASE_CLIENT_EMAIL', ''), - // Env-stored keys keep literal "\n"; restore real newlines. - privateKey: (this.options.private_key ?? env('FIREBASE_PRIVATE_KEY', ''))?.replace(/\\n/g, '\n'), - }) - - // Reuse a named app so repeated broadcasts don't re-initialize. - const name = String( - this.options.app_name ?? env('FIREBASE_APP_NAME', 'arkstack-realtime') - ).replaceAll(' ', '-').toLowerCase() - const existing = app.getApps().find((a: { name: string }) => a.name === name) - const instance = existing ?? app.initializeApp({ credential }, name) + const [instance, messaging] = await Promise.all([ + firebaseAdminApp(this.options), + import(specifier), + ]) return messaging.getMessaging(instance) })() @@ -268,6 +236,16 @@ export class FirebaseRealtimeDriver implements RealtimeDriver { return Object.keys(block).length > 0 ? { [key]: block } : {} } + /** + * Public Firebase web settings for clients that receive messages or register + * device tokens. + * + * @returns + */ + clientConfig(): Record { + return firebaseWebConfig(this.options) + } + /** * Firebase channels do not require server-side authorization. * diff --git a/packages/notifications/src/drivers/realtime/MemoryRealtimeDriver.ts b/packages/notifications/src/drivers/realtime/MemoryRealtimeDriver.ts new file mode 100644 index 00000000..f53ec81c --- /dev/null +++ b/packages/notifications/src/drivers/realtime/MemoryRealtimeDriver.ts @@ -0,0 +1,68 @@ +import type { RealtimeBroadcastPayload, RealtimeDeliveryOptions } from '../../types' + +import type { RealtimeDriver } from '../../Contracts/RealtimeDriver' + +/** + * A broadcast recorded by {@link MemoryRealtimeDriver}. + */ +export interface RecordedBroadcast { + channel: string | string[] + event: string + payload: RealtimeBroadcastPayload + delivery?: RealtimeDeliveryOptions +} + +const BUFFER_KEY = Symbol.for('arkstack.realtime.memory') + +/** + * The recorded broadcasts. Kept on `globalThis` so code loaded through separate + * module graphs (an app's routes and its tests, say) shares one buffer. + * + * @returns + */ +const buffer = (): RecordedBroadcast[] => { + const store = globalThis as Record + + return (store[BUFFER_KEY] ??= []) +} + +/** + * Records every broadcast in process instead of sending it, for tests and local + * development. Read them with {@link MemoryRealtimeDriver.broadcasts}. + */ +export class MemoryRealtimeDriver implements RealtimeDriver { + /** + * Every broadcast recorded since the last {@link clear}. + * + * @returns + */ + static broadcasts(): RecordedBroadcast[] { + return buffer() + } + + /** + * Forget all recorded broadcasts. + */ + static clear(): void { + buffer().length = 0 + } + + async broadcast( + channel: string | string[], + event: string, + payload: RealtimeBroadcastPayload, + delivery?: RealtimeDeliveryOptions, + ): Promise { + buffer().push({ channel, event, payload, delivery }) + } + + async auth(_socketId: string, channel: string): Promise<{ auth: string }> { + return { auth: `memory:${channel}` } + } + + async registerAuthRoute(): Promise { } + + clientConfig(): Record { + return {} + } +} diff --git a/packages/notifications/src/drivers/realtime/PollingRealtimeDriver.ts b/packages/notifications/src/drivers/realtime/PollingRealtimeDriver.ts new file mode 100644 index 00000000..1cbaddf3 --- /dev/null +++ b/packages/notifications/src/drivers/realtime/PollingRealtimeDriver.ts @@ -0,0 +1,28 @@ +import type { PollingTransportConfig, RealtimeBroadcastPayload, RealtimeDeliveryOptions } from '../../types' + +import type { RealtimeDriver } from '../../Contracts/RealtimeDriver' + +/** + * A transport that pushes nothing. Clients are told to poll on an interval + * instead, so an app works without realtime infrastructure. + */ +export class PollingRealtimeDriver implements RealtimeDriver { + constructor(private options: PollingTransportConfig = {}) { } + + async broadcast( + _channel: string | string[], + _event: string, + _payload: RealtimeBroadcastPayload, + _delivery?: RealtimeDeliveryOptions, + ): Promise { } + + async auth(): Promise> { + return {} + } + + async registerAuthRoute(): Promise { } + + clientConfig(): Record { + return { interval: this.options.interval ?? 5000 } + } +} diff --git a/packages/notifications/src/drivers/realtime/PusherRealtimeDriver.ts b/packages/notifications/src/drivers/realtime/PusherRealtimeDriver.ts index f032bb4f..64148503 100644 --- a/packages/notifications/src/drivers/realtime/PusherRealtimeDriver.ts +++ b/packages/notifications/src/drivers/realtime/PusherRealtimeDriver.ts @@ -1,6 +1,8 @@ import type { PusherTransportConfig, RealtimeBroadcastPayload, RealtimeDeliveryOptions } from '../../types' import { RequestException, env } from '@arkstack/common' +import { Broadcast } from '../../Broadcast' + import type { RealtimeDriver } from '../../Contracts/RealtimeDriver' /** The slice of the `pusher` server SDK this driver uses. */ @@ -28,6 +30,8 @@ type PusherConstructor = new (options: { secret: string cluster: string useTLS?: boolean + host?: string + port?: string }) => PusherClient /** @@ -52,18 +56,57 @@ export class PusherRealtimeDriver implements RealtimeDriver { const Pusher = (mod.default ?? mod) as PusherConstructor + const { host, port } = this.server() + return new Pusher({ appId: this.options.app_id ?? env('PUSHER_APP_ID', ''), - key: this.options.key ?? env('PUSHER_KEY', ''), + key: this.key(), secret: this.options.secret ?? env('PUSHER_SECRET', ''), - cluster: this.options.cluster ?? env('PUSHER_CLUSTER', 'mt1'), - useTLS: this.options.use_tls ?? true, + cluster: this.cluster(), + useTLS: this.useTLS(), + ...(host ? { host, port: port ? String(port) : undefined } : {}), }) })() return this.clientPromise } + private key(): string { + return this.options.key ?? env('PUSHER_KEY', '') + } + + private cluster(): string { + return this.options.cluster ?? env('PUSHER_CLUSTER', 'mt1') + } + + private useTLS(): boolean { + return this.options.use_tls ?? true + } + + private server(): { host?: string, port?: number } { + const host = this.options.host ?? (env('PUSHER_HOST', '') || undefined) + const port = this.options.port ?? (Number(env('PUSHER_PORT', 0)) || undefined) + + return { host, port } + } + + /** + * Public settings a `pusher-js` client needs, including the self-hosted + * server address when `host` is set. + * + * @returns + */ + clientConfig(): Record { + const { host, port } = this.server() + + return { + key: this.key(), + cluster: this.cluster(), + forceTLS: this.useTLS(), + ...(host ? { wsHost: host, wsPort: port, wssPort: port } : {}), + } + } + /** * `delivery` is accepted for contract parity and ignored: Pusher pushes over a * connection the client already holds open, so there is no dozing radio to @@ -134,39 +177,36 @@ export class PusherRealtimeDriver implements RealtimeDriver { middleware?: unknown | unknown[], channelPrefix?: string ): Promise { - const client = await this.client() - const drivers = { - express: '@arkstack/driver-express', - h3: '@arkstack/driver-express' - } + const drivers = ['@arkstack/driver-express', '@arkstack/driver-h3'] - for (const [name, path] of Object.entries(drivers)) { - const midsPath = `${path}/middlewares` + for (const path of drivers) { try { const { Router } = await import(path) - const { auth } = await import(midsPath) - - const middlewares = new Set(Array.isArray(middleware) - ? middleware.concat(auth) - : [middleware, auth] - ) + const { auth } = await import(`${path}/middlewares`) - Router.post(authEndpoint, ({ clearRequest, req }: any) => { + Router.post(authEndpoint, async ({ clearRequest, req }: any) => { + const user = req?.user ?? clearRequest.user const channel = clearRequest.input('channel_name') const socketId = clearRequest.input('socket_id') - const expected = channelPrefix ?? - `${config('notifications.drivers.realtime.channel_prefix', 'user.')}${req.user?.id ?? clearRequest.user?.id}` + // An explicit channel limits access to exactly that channel. + const allowed = channelPrefix + ? channel === channelPrefix + : await Broadcast.authorize(user, channel) - RequestException.abortIf(channel !== expected, 'Channel access denied', 403) + RequestException.abortIf(!allowed, 'Channel access denied', 403) - return client.authorizeChannel(socketId, channel) - }).middleware(Array.from(middlewares)) + // pusher-js only accepts a 200 from the authorizer. + return new Response(JSON.stringify(await this.auth(socketId, channel)), { + status: 200, + headers: { 'content-type': 'application/json' }, + }) + }).middleware([middleware ?? []].flat().concat(auth)) - break + return } catch (e) { console.error( - `Failed to register auth route for ${name}: ${e instanceof Error ? e.message : e}` + `Failed to register auth route for ${path}: ${e instanceof Error ? e.message : e}` ) } } diff --git a/packages/notifications/src/drivers/realtime/firebaseAdmin.ts b/packages/notifications/src/drivers/realtime/firebaseAdmin.ts new file mode 100644 index 00000000..3b4bd98a --- /dev/null +++ b/packages/notifications/src/drivers/realtime/firebaseAdmin.ts @@ -0,0 +1,79 @@ +import { existsSync, readFileSync } from 'node:fs' + +import type { FirebaseTransportConfig } from '../../types' +import { env } from '@arkstack/common' +import path from 'node:path' + +/** The slice of `firebase-admin/app` the Firebase drivers use. */ +export interface FirebaseAdminAppModule { + cert(serviceAccount: Record): unknown + getApps(): Array<{ name: string }> + initializeApp(options: Record, name: string): unknown +} + +/** + * Initialize, or reuse, a named Firebase Admin app. + * + * Credentials come from the service account file at `admin_sdk_path` when it + * exists, otherwise from the inline `project_id` / `client_email` / `private_key`. + * + * @param options + * @param suffix Distinguishes apps that need different init options. + * @param extra Extra `initializeApp` options, such as `databaseURL`. + * @returns + */ +export const firebaseAdminApp = async ( + options: FirebaseTransportConfig, + suffix = '', + extra: Record = {}, +): Promise => { + const specifier = 'firebase-admin/app' + const app = await import(specifier).catch(() => { + throw new Error( + 'The "firebase-admin" package is required for the Firebase realtime transports. Install it with `npm i firebase-admin`.', + ) + }) as FirebaseAdminAppModule + + const adminsdk = path.join( + process.cwd(), + options.admin_sdk_path ?? env('FIREBASE_ADMINSDK', 'firebase-adminsdk.json') + ) + + let serviceAccount: Record | undefined + try { + if (existsSync(adminsdk)) + serviceAccount = JSON.parse(readFileSync(adminsdk, { encoding: 'utf-8' })) + } catch {/** */ } + + const credential = app.cert(serviceAccount ?? { + projectId: options.project_id ?? env('FIREBASE_PROJECT_ID', ''), + clientEmail: options.client_email ?? env('FIREBASE_CLIENT_EMAIL', ''), + // Env-stored keys keep literal "\n"; restore real newlines. + privateKey: (options.private_key ?? env('FIREBASE_PRIVATE_KEY', ''))?.replace(/\\n/g, '\n'), + }) + + // Reuse a named app so repeated broadcasts don't re-initialize. + const name = String(options.app_name ?? env('FIREBASE_APP_NAME', 'arkstack-realtime')) + .replaceAll(' ', '-') + .toLowerCase() + suffix + const existing = app.getApps().find((a) => a.name === name) + + return existing ?? app.initializeApp({ credential, ...extra }, name) +} + +/** + * The public web app settings a browser Firebase client is initialized with. + * + * @param options + * @returns + */ +export const firebaseWebConfig = ( + options: FirebaseTransportConfig, +): Record => ({ + apiKey: options.web?.api_key ?? env('FIREBASE_API_KEY', ''), + authDomain: options.web?.auth_domain ?? env('FIREBASE_AUTH_DOMAIN', ''), + projectId: options.project_id ?? env('FIREBASE_PROJECT_ID', ''), + appId: options.web?.app_id ?? env('FIREBASE_APP_ID', ''), + messagingSenderId: options.web?.messaging_sender_id ?? env('FIREBASE_MESSAGING_SENDER_ID', ''), + vapidKey: options.web?.vapid_key ?? env('FIREBASE_VAPID_KEY', ''), +}) diff --git a/packages/notifications/src/index.ts b/packages/notifications/src/index.ts index 1b44de65..6540d4ce 100644 --- a/packages/notifications/src/index.ts +++ b/packages/notifications/src/index.ts @@ -1,4 +1,5 @@ export * from './Notification' +export * from './Broadcast' export * from './UserNotificationCenter' export * from './config' export * from './Contracts/NotificationContract' @@ -12,5 +13,8 @@ export * from './drivers/sms/AfricasTalkingSmsDriver' export * from './drivers/sms/TwilioSmsDriver' export * from './drivers/realtime/PusherRealtimeDriver' export * from './drivers/realtime/FirebaseRealtimeDriver' +export * from './drivers/realtime/FirebaseDatabaseRealtimeDriver' +export * from './drivers/realtime/MemoryRealtimeDriver' +export * from './drivers/realtime/PollingRealtimeDriver' export * from './types' export * from './utils/template' diff --git a/packages/notifications/src/types.ts b/packages/notifications/src/types.ts index 0d78bff9..4d38aa03 100644 --- a/packages/notifications/src/types.ts +++ b/packages/notifications/src/types.ts @@ -16,7 +16,21 @@ export type NotificationData = Record export type SmsDriverName = 'africastalking' | 'twilio' -export type RealtimeDriverName = 'pusher' | 'firebase' +/** + * Realtime transports. + * + * - `pusher`: Pusher Channels, or any Pusher-protocol server such as soketi. + * - `firebase`: Firebase Cloud Messaging; delivers to devices, not channels. + * - `firebase-database`: Firebase Realtime Database channels. + * - `polling`: no push; clients are told to poll on an interval. + * - `memory`: records broadcasts in process, for tests. + */ +export type RealtimeDriverName = 'pusher' | 'firebase' | 'firebase-database' | 'polling' | 'memory' + +/** + * Transports that can deliver to device registration tokens. + */ +export type PushDriverName = 'firebase' export type NotificationChannel = 'mail' | 'sms' | 'db' | 'realtime' @@ -78,6 +92,11 @@ export type PusherTransportConfig = { secret?: string cluster?: string use_tls?: boolean + /** + * A self-hosted Pusher-protocol server (e.g. soketi). Overrides `cluster` addressing. + */ + host?: string + port?: number } export type FirebaseTransportConfig = { @@ -88,6 +107,43 @@ export type FirebaseTransportConfig = { admin_sdk_path?: string /** Delivery options applied to every send on this transport. */ delivery?: RealtimeDeliveryOptions + web?: FirebaseWebConfig +} + +/** + * Public Firebase web app settings handed to browser clients. Never secret. + */ +export type FirebaseWebConfig = { + api_key?: string + auth_domain?: string + app_id?: string + messaging_sender_id?: string + /** Web push VAPID key clients use to request a messaging token. */ + vapid_key?: string +} + +export type FirebaseDatabaseTransportConfig = Omit & { + database_url?: string + /** + * Database path that holds channel events (default `arkstack/channels`). + */ + root?: string + /** + * Database path that holds per-user channel grants (default `arkstack/grants`). + */ + grants_root?: string + /** + * How long, in seconds, a channel grant stays valid (default 3600). + */ + grant_ttl?: number + web?: FirebaseWebConfig +} + +export type PollingTransportConfig = { + /** + * How often, in milliseconds, clients should poll (default 5000). + */ + interval?: number } /** @@ -157,6 +213,9 @@ export type RealtimeDriverOptions RealtimeDriver pusher?: PusherTransportConfig firebase?: FirebaseTransportConfig + 'firebase-database'?: FirebaseDatabaseTransportConfig + polling?: PollingTransportConfig + memory?: Record } /** The notification payload delivered to realtime clients. */ @@ -197,12 +256,58 @@ export type RealtimeNotificationPayload = { */ export type RealtimeBroadcastPayload = RealtimeNotificationPayload | Record +/** + * The outcome of a device push: totals plus the tokens to prune. + */ +export type PushResult = { + successCount: number + failureCount: number + invalidTokens: string[] +} + /** The result of a realtime broadcast (plus the stored record when `store` is on). */ export type RealtimeBroadcastResult = { channel: string | string[] event: string payload: RealtimeNotificationPayload stored?: UserNotification + /** Present when the notification was also pushed to the recipient's devices. */ + push?: PushResult +} + +/** + A device registration token, or a record carrying one. + */ +export type PushTokenLike = string | { token: string } + +/** + Decides whether a subject (usually the authenticated user) may join a channel. + */ +export type ChannelAuthorizer = ( + subject: Subject, + params: Record, +) => boolean | Promise + +/** + * The body a realtime client posts to the channel auth endpoint. + */ +export type ChannelAuthRequest = { + socket_id?: string + channel_name?: string +} + +/** + * Public connection settings a `@arkstack/realtime` client is created from. + */ +export type RealtimeClientConfig = { + transport: RealtimeDriverName + event: string + channelPrefix: string + pusher?: Record + firebase?: Record + 'firebase-database'?: Record + polling?: Record + memory?: Record } export type DbNotificationType = 'transaction' | 'pocket' | 'family' | 'security' | 'promo' | 'bill' | 'goal' | string @@ -245,6 +350,17 @@ export interface NotificationConfig { } realtime?: { transport?: RealtimeDriverName + /** + * Also push notifications sent to a user to their registered devices + * over this transport. When unset, on only if `transport` is `firebase`; + * `false` turns it off. + */ + push?: PushDriverName | false + /** + * Resolve a user's device tokens for `push`. Defaults to reading + * `user.pushTokens`. + */ + push_tokens?: (user: any) => PushTokenLike[] | Promise /** Prefix for the per-user channel/topic (default `user.`). */ channel_prefix?: string /** Event name clients subscribe to (default `notification`). */ @@ -316,8 +432,10 @@ export interface NotificationConfig { secret: string; cluster: string; use_tls?: boolean + host?: string + port?: number } - firebase?: { + firebase?: ({ project_id: string; client_email: string; private_key: string; @@ -325,7 +443,12 @@ export interface NotificationConfig { } | { app_name?: string; admin_sdk_path: string + }) & { + delivery?: RealtimeDeliveryOptions + web?: FirebaseWebConfig } + 'firebase-database'?: FirebaseDatabaseTransportConfig + polling?: PollingTransportConfig } queue?: { name?: string, diff --git a/packages/notifications/tests/broadcast.test.ts b/packages/notifications/tests/broadcast.test.ts new file mode 100644 index 00000000..4a3587bb --- /dev/null +++ b/packages/notifications/tests/broadcast.test.ts @@ -0,0 +1,373 @@ +import { + Broadcast, + FirebaseDatabaseRealtimeDriver, + FirebaseRealtimeDriver, + MemoryRealtimeDriver, + Notification, + PollingRealtimeDriver, + PusherRealtimeDriver, + UserNotificationCenter, + firebaseChannelKey, +} from '../src' +import { afterEach, describe, expect, it, vi } from 'vitest' + +/** Override `drivers.realtime` for one test. */ +const withSettings = (settings: Record) => { + vi.spyOn(Broadcast as any, 'settings').mockReturnValue(settings) +} + +afterEach(() => { + vi.restoreAllMocks() + Broadcast.forgetChannels() + MemoryRealtimeDriver.clear() +}) + +describe('Broadcast channel registry', () => { + it('passes placeholder values to the authorizer', async () => { + const authorizer = vi.fn(() => true) + Broadcast.channel('private-team.{teamId}.project.{projectId}', authorizer) + + await expect(Broadcast.authorize({ id: 1 }, 'private-team.t-1.project.p-2')).resolves.toBe(true) + expect(authorizer).toHaveBeenCalledWith({ id: 1 }, { teamId: 't-1', projectId: 'p-2' }) + }) + + it('does not let a placeholder span a dot', async () => { + Broadcast.channel('private-team.{teamId}', () => true) + + await expect(Broadcast.authorize({ id: 1 }, 'private-team.t-1.project.p-2')).resolves.toBe(false) + }) + + it('denies a channel no rule matches', async () => { + await expect(Broadcast.authorize({ id: 1 }, 'private-anything')).resolves.toBe(false) + }) + + it('allows the per-user channel for its own user only', async () => { + await expect(Broadcast.authorize({ id: 7 }, 'user.7')).resolves.toBe(true) + await expect(Broadcast.authorize({ id: 8 }, 'user.7')).resolves.toBe(false) + await expect(Broadcast.authorize(null, 'user.7')).resolves.toBe(false) + }) + + it('lets a rule override the per-user channel', async () => { + Broadcast.channel('user.{id}', () => false) + + await expect(Broadcast.authorize({ id: 7 }, 'user.7')).resolves.toBe(false) + }) + + it('replaces an earlier rule for the same pattern', async () => { + Broadcast.channel('private-x.{id}', () => false) + Broadcast.channel('private-x.{id}', () => true) + + await expect(Broadcast.authorize({}, 'private-x.1')).resolves.toBe(true) + }) + + it('treats regex characters in a pattern literally', async () => { + Broadcast.channel('private-a+b.{id}', () => true) + + await expect(Broadcast.authorize({}, 'private-aab.1')).resolves.toBe(false) + await expect(Broadcast.authorize({}, 'private-a+b.1')).resolves.toBe(true) + }) +}) + +describe('Broadcast transports', () => { + it('builds every built-in transport', () => { + expect(Broadcast.make('pusher')).toBeInstanceOf(PusherRealtimeDriver) + expect(Broadcast.make('firebase')).toBeInstanceOf(FirebaseRealtimeDriver) + expect(Broadcast.make('firebase-database')).toBeInstanceOf(FirebaseDatabaseRealtimeDriver) + expect(Broadcast.make('polling')).toBeInstanceOf(PollingRealtimeDriver) + expect(Broadcast.make('memory')).toBeInstanceOf(MemoryRealtimeDriver) + }) + + it('rejects an unknown transport', () => { + expect(() => Broadcast.make('nope' as never)).toThrow(/unknown transport/) + }) + + it('uses the configured transport', () => { + withSettings({ transport: 'memory' }) + + expect(Broadcast.driver()).toBeInstanceOf(MemoryRealtimeDriver) + expect(Notification.realtime().driver).toBeInstanceOf(MemoryRealtimeDriver) + }) + + it('broadcasts arbitrary events and skips an empty channel list', async () => { + withSettings({ transport: 'memory' }) + + await Broadcast.broadcast(['a', 'b'], 'session.updated', { id: 's-1' }) + await Broadcast.broadcast([], 'ignored', {}) + + expect(MemoryRealtimeDriver.broadcasts()).toEqual([ + { channel: ['a', 'b'], event: 'session.updated', payload: { id: 's-1' }, delivery: undefined }, + ]) + }) + + it('records memory broadcasts on globalThis so separate module graphs share them', async () => { + await new MemoryRealtimeDriver().broadcast('c', 'e', { n: 1 }) + + const store = globalThis as Record + + expect(store[Symbol.for('arkstack.realtime.memory')]).toHaveLength(1) + }) + + it('tells polling clients how often to poll and pushes nothing', async () => { + const driver = new PollingRealtimeDriver({ interval: 2000 }) + + await expect(driver.broadcast('c', 'e', {})).resolves.toBeUndefined() + expect(driver.clientConfig()).toEqual({ interval: 2000 }) + }) + + it('hands pusher-js a self-hosted server address when host is set', () => { + const driver = new PusherRealtimeDriver({ key: 'k', cluster: 'eu', use_tls: false, host: 'soketi.test', port: 6001 }) + + expect(driver.clientConfig()).toEqual({ + key: 'k', cluster: 'eu', forceTLS: false, wsHost: 'soketi.test', wsPort: 6001, wssPort: 6001, + }) + }) + + it('omits the server address for hosted Pusher', () => { + const driver = new PusherRealtimeDriver({ key: 'k', cluster: 'eu' }) + + expect(driver.clientConfig()).toEqual({ key: 'k', cluster: 'eu', forceTLS: true }) + }) + + it('describes the active transport and the push transport for clients', () => { + withSettings({ transport: 'polling', push: 'firebase', channel_prefix: 'private-user.' }) + + const config = Broadcast.clientConfig() + + expect(config.transport).toBe('polling') + expect(config.channelPrefix).toBe('private-user.') + expect(config.event).toBe('notification') + expect(config.polling).toEqual({ interval: 5000 }) + expect(config.firebase).toEqual(expect.objectContaining({ vapidKey: expect.any(String) })) + }) +}) + +describe('Broadcast.handleAuth', () => { + it('requires a channel name', async () => { + await expect(Broadcast.handleAuth({ id: 1 }, {})).rejects.toThrow(/channel_name/) + }) + + it('denies a channel the subject is not authorized for', async () => { + await expect(Broadcast.handleAuth({ id: 1 }, { channel_name: 'user.2', socket_id: '1.1' })) + .rejects.toThrow(/denied/) + }) + + it('requires a socket id for Pusher', async () => { + const driver = new PusherRealtimeDriver() + + await expect(Broadcast.handleAuth({ id: 1 }, { channel_name: 'user.1' }, { driver })) + .rejects.toThrow(/socket_id/) + }) + + it('signs an authorized Pusher channel', async () => { + const driver = new PusherRealtimeDriver() + const auth = vi.spyOn(driver, 'auth').mockResolvedValue({ auth: 'key:sig' }) + + await expect(Broadcast.handleAuth({ id: 1 }, { channel_name: 'user.1', socket_id: '1.1' }, { driver })) + .resolves.toEqual({ auth: 'key:sig' }) + expect(auth).toHaveBeenCalledWith('1.1', 'user.1') + }) + + it('grants Firebase database channels to the subject uid', async () => { + const driver = new FirebaseDatabaseRealtimeDriver() + const auth = vi.spyOn(driver, 'auth').mockResolvedValue({ token: 't', expires_at: 1 }) + + await Broadcast.handleAuth({ id: 9 }, { channel_name: 'user.9' }, { driver }) + await Broadcast.handleAuth({ id: 9 }, { channel_name: 'user.9' }, { driver, uid: 'custom' }) + + expect(auth.mock.calls).toEqual([['9', 'user.9'], ['custom', 'user.9']]) + }) +}) + +describe('device push', () => { + const user = { id: 7, pushTokens: [{ token: 'a' }, 'b'] } as never + + const pushDriver = () => { + const driver = new FirebaseRealtimeDriver() + const broadcast = vi.spyOn(driver, 'broadcast') + .mockResolvedValue({ successCount: 1, failureCount: 1, invalidTokens: ['b'] }) + vi.spyOn(Broadcast, 'pushDriver').mockReturnValue(driver) + + return broadcast + } + + it('streams to the user channel and pushes to their devices', async () => { + withSettings({ transport: 'memory', push: 'firebase' }) + const broadcast = pushDriver() + + const result = await Notification.realtime().recipient(user).send('Hello', 'Hi') + + expect(MemoryRealtimeDriver.broadcasts()[0].channel).toBe('user.7') + expect(broadcast).toHaveBeenCalledWith(['a', 'b'], 'notification', expect.objectContaining({ title: 'Hi' }), {}) + expect(result.push).toEqual({ successCount: 1, failureCount: 1, invalidTokens: ['b'] }) + }) + + it('skips the push when asked', async () => { + withSettings({ transport: 'memory', push: 'firebase' }) + const broadcast = pushDriver() + + const result = await Notification.realtime().recipient(user).push(false).send('Hello') + + expect(broadcast).not.toHaveBeenCalled() + expect(result.push).toBeUndefined() + }) + + it('pushes to explicit devices instead of the user\'s own', async () => { + withSettings({ transport: 'memory', push: 'firebase' }) + const broadcast = pushDriver() + + await Notification.realtime().recipient(user).devices(['z']).send('Hello') + + expect(broadcast.mock.calls[0][0]).toEqual(['z']) + }) + + it('resolves tokens through the configured resolver', async () => { + withSettings({ push_tokens: async () => ['x', { token: 'y' }] }) + + await expect(Broadcast.pushTokens({ id: 1 })).resolves.toEqual(['x', 'y']) + }) + + it('keeps pushing to devices on the firebase transport when push is unset', () => { + withSettings({ transport: 'firebase' }) + + expect(Broadcast.pushDriver()).toBeInstanceOf(FirebaseRealtimeDriver) + }) + + it('lets push: false turn device push off on the firebase transport', () => { + withSettings({ transport: 'firebase', push: false }) + + expect(Broadcast.pushDriver()).toBeNull() + }) + + it('does nothing when push is off', async () => { + withSettings({ transport: 'memory' }) + + expect(Broadcast.pushDriver()).toBeNull() + await expect(Broadcast.push(['a'], 'e', {})).resolves.toBeNull() + }) + + it('chains broadcast() on a database notification', async () => { + withSettings({ transport: 'memory' }) + vi.spyOn(UserNotificationCenter, 'create').mockResolvedValue({ id: 5, readAt: null, createdAt: new Date() } as never) + + await Notification.db().broadcast().recipient({ id: 2 } as never).send('D', 'T') + + expect(MemoryRealtimeDriver.broadcasts()[0]).toMatchObject({ channel: 'user.2', payload: { title: 'T' } }) + }) + + it('still broadcasts a stored notification to a user with no devices', async () => { + withSettings({ transport: 'memory' }) + vi.spyOn(UserNotificationCenter, 'create').mockResolvedValue({ id: 3, readAt: null, createdAt: new Date() } as never) + + await UserNotificationCenter.send({ id: 4 } as never, { title: 'T', description: 'D' }) + + expect(MemoryRealtimeDriver.broadcasts()[0]).toMatchObject({ channel: 'user.4', payload: { id: '3' } }) + }) +}) + +describe('FirebaseDatabaseRealtimeDriver', () => { + const fakeServices = () => { + let next = 0 + const updates: Array> = [] + const indexed: Array<{ key: string, val: unknown }> = [] + const query = { + endAt: vi.fn(() => query), + limitToFirst: vi.fn(() => query), + once: vi.fn(async () => ({ + forEach: (cb: (child: { key: string, val(): unknown }) => void) => { + indexed.forEach((entry) => cb({ key: entry.key, val: () => entry.val })) + + return false + }, + })), + } + const database = { + ref: vi.fn(() => ({ + push: () => ({ key: `k${++next}` }), + update: vi.fn(async (values: Record) => { + updates.push(values) +}), + orderByChild: vi.fn(() => query), + })), + } + const auth = { createCustomToken: vi.fn(async (uid: string) => `token-for-${uid}`) } + + return { database, auth, updates, indexed, query } + } + + const withServices = (services: ReturnType, options = {}) => { + const driver = new FirebaseDatabaseRealtimeDriver(options); + (driver as any).servicesPromise = Promise.resolve(services) + + return driver + } + + it('encodes channel names into single database keys', () => { + expect(firebaseChannelKey('private-user.7')).toBe('private-user%2E7') + expect(firebaseChannelKey('a/b#c$d[e]')).toBe('a%2Fb%23c%24d%5Be%5D') + }) + + it('writes each channel event and its index entry in one update', async () => { + const services = fakeServices() + const driver = withServices(services) + + await driver.broadcast(['user.1', 'user.2'], 'notification', { id: 'n' }) + + expect(services.updates).toHaveLength(1) + expect(services.updates[0]).toEqual({ + 'arkstack/channels/user%2E1/k1': { event: 'notification', payload: { id: 'n' }, at: { '.sv': 'timestamp' } }, + 'arkstack/channels-index/k1': { channel: 'user%2E1', at: { '.sv': 'timestamp' } }, + 'arkstack/channels/user%2E2/k2': { event: 'notification', payload: { id: 'n' }, at: { '.sv': 'timestamp' } }, + 'arkstack/channels-index/k2': { channel: 'user%2E2', at: { '.sv': 'timestamp' } }, + }) + }) + + it('grants the channel until the TTL ends and mints a token', async () => { + vi.spyOn(Date, 'now').mockReturnValue(1_000) + const services = fakeServices() + const driver = withServices(services, { grant_ttl: 60 }) + + await expect(driver.auth('u-1', 'user.1')).resolves.toEqual({ token: 'token-for-u-1', expires_at: 61_000 }) + expect(services.updates[0]).toEqual({ 'arkstack/grants/u-1/user%2E1': 61_000 }) + }) + + it('rejects a uid that cannot be a database key', async () => { + const driver = withServices(fakeServices()) + + await expect(driver.auth('a.b', 'user.1')).rejects.toThrow(/uid/) + await expect(driver.auth('', 'user.1')).rejects.toThrow(/uid/) + }) + + it('prunes old events in a bounded batch', async () => { + vi.spyOn(Date, 'now').mockReturnValue(100_000) + const services = fakeServices() + services.indexed.push({ key: 'k1', val: { channel: 'user%2E1' } }, { key: 'k2', val: { channel: 'user%2E2' } }) + const driver = withServices(services) + + await expect(driver.prune(60, 2)).resolves.toBe(2) + expect(services.query.endAt).toHaveBeenCalledWith(40_000) + expect(services.query.limitToFirst).toHaveBeenCalledWith(2) + expect(services.updates[0]).toEqual({ + 'arkstack/channels/user%2E1/k1': null, + 'arkstack/channels-index/k1': null, + 'arkstack/channels/user%2E2/k2': null, + 'arkstack/channels-index/k2': null, + }) + }) + + it('skips the write when nothing is old enough to prune', async () => { + const services = fakeServices() + const driver = withServices(services) + + await expect(driver.prune(60)).resolves.toBe(0) + expect(services.updates).toHaveLength(0) + }) + + it('gives clients the database URL and channel root', () => { + const driver = new FirebaseDatabaseRealtimeDriver({ database_url: 'https://x.firebaseio.com', root: 'rt/channels' }) + + expect(driver.clientConfig()).toEqual(expect.objectContaining({ + databaseURL: 'https://x.firebaseio.com', + root: 'rt/channels', + })) + }) +}) diff --git a/packages/realtime/src/RealtimeClient.ts b/packages/realtime/src/RealtimeClient.ts index 61a5052c..eaced980 100644 --- a/packages/realtime/src/RealtimeClient.ts +++ b/packages/realtime/src/RealtimeClient.ts @@ -1,4 +1,11 @@ -import type { NotificationHandler, RealtimeConfig, RealtimeEventHandler, RealtimeSubscription, RealtimeTransport } from './types' +import type { + NotificationHandler, + RealtimeConfig, + RealtimeEventHandler, + RealtimeSubscription, + RealtimeTransport, + RealtimeTransportName, +} from './types' /** * Consumes Arkstack realtime notifications. Resolves a transport (Pusher, @@ -36,23 +43,50 @@ export class RealtimeClient { return await this.config.transportFactory() } - if (this.config.transport === 'firebase') { - if (!this.config.firebase) { - throw new Error('Realtime: `firebase` config is required for the Firebase transport') + switch (this.transportName) { + case 'firebase': { + const { createFirebaseTransport } = await import('./transports/firebase') + + return await createFirebaseTransport(this.required('firebase')) + } + case 'firebase-database': { + const { createFirebaseDatabaseTransport } = await import('./transports/firebase-database') + + return await createFirebaseDatabaseTransport(this.required('firebase-database')) + } + case 'polling': { + const { createPollingTransport } = await import('./transports/polling') + + return createPollingTransport(this.config.polling) } + case 'memory': { + const { createMemoryTransport } = await import('./transports/memory') - const { createFirebaseTransport } = await import('./transports/firebase') + return createMemoryTransport() + } + default: { + const { createPusherTransport } = await import('./transports/pusher') - return await createFirebaseTransport(this.config.firebase) + return await createPusherTransport(this.required('pusher')) + } } + } - if (!this.config.pusher) { - throw new Error('Realtime: `pusher` config is required for the Pusher transport') + private required(name: K): NonNullable { + const config = this.config[name] + + if (!config) { + throw new Error(`Realtime: \`${name}\` config is required for the ${name} transport`) } - const { createPusherTransport } = await import('./transports/pusher') + return config as NonNullable + } - return await createPusherTransport(this.config.pusher) + /** + * The configured transport name. + */ + get transportName(): RealtimeTransportName { + return this.config.transport ?? 'pusher' } /** @@ -146,6 +180,39 @@ export class RealtimeClient { return await this.subscribe(this.channelFor(userId), handler) } + /** + * Run `handler` on an interval while the transport is `polling`, so a view + * can refetch what it would otherwise receive live. Does nothing on push + * transports. Returns a function that stops it. + * + * @param handler + * @param interval Overrides the transport's advertised interval, in milliseconds. + * @returns + */ + async onPoll(handler: () => void, interval?: number): Promise<() => void> { + const transport = await this.transport() + + if (!transport.polling) { + return () => { } + } + + const timer = setInterval(handler, interval ?? transport.polling.interval) + + return () => clearInterval(timer) + } + + /** + * This device's push registration token, for registering with your API. + * Null when the transport cannot issue one. + * + * @returns + */ + async deviceToken(): Promise { + const transport = await this.transport() + + return (await transport.deviceToken?.()) ?? null + } + /** Tear down the underlying transport connection. */ async disconnect(): Promise { if (!this.transportPromise) { diff --git a/packages/realtime/src/index.ts b/packages/realtime/src/index.ts index ad278ed4..bfa4d293 100644 --- a/packages/realtime/src/index.ts +++ b/packages/realtime/src/index.ts @@ -1,5 +1,8 @@ export { RealtimeClient, createRealtime } from './RealtimeClient' export { createPusherTransport } from './transports/pusher' export { createFirebaseTransport } from './transports/firebase' +export { createFirebaseDatabaseTransport } from './transports/firebase-database' +export { createPollingTransport } from './transports/polling' +export { createMemoryTransport, emitMemoryEvent } from './transports/memory' export { supersede } from './supersede' export type * from './types' diff --git a/packages/realtime/src/transports/firebase-database.ts b/packages/realtime/src/transports/firebase-database.ts new file mode 100644 index 00000000..c5afd765 --- /dev/null +++ b/packages/realtime/src/transports/firebase-database.ts @@ -0,0 +1,183 @@ +import type { FirebaseDatabaseClientConfig, RealtimeEventHandler, RealtimeTransport } from '../types' + +/** The slices of the `firebase` SDK this transport uses. */ +interface FirebaseApp { + name: string +} + +interface FirebaseAppModule { + getApps(): FirebaseApp[] + initializeApp(options: Record, name: string): FirebaseApp +} + +interface FirebaseAuth { + currentUser: { uid: string } | null +} + +interface FirebaseAuthModule { + getAuth(app: FirebaseApp): FirebaseAuth + signInWithCustomToken(auth: FirebaseAuth, token: string): Promise + signOut(auth: FirebaseAuth): Promise +} + +interface DatabaseSnapshot { + val(): unknown +} + +interface FirebaseDatabaseModule { + getDatabase(app: FirebaseApp, url?: string): unknown + ref(database: unknown, path: string): unknown + query(reference: unknown, ...constraints: unknown[]): unknown + orderByChild(path: string): unknown + startAt(value: number): unknown + onValue(query: unknown, handler: (snapshot: DatabaseSnapshot) => void, options?: { onlyOnce?: boolean }): () => void + onChildAdded(query: unknown, handler: (snapshot: DatabaseSnapshot) => void): () => void +} + +interface ChannelEvent { + event?: string + payload?: unknown +} + +const APP_NAME = 'arkstack-realtime-database' + +/** + * Encode a channel name into a single Realtime Database key, matching the server. + * + * @param channel + * @returns + */ +const channelKey = (channel: string): string => encodeURIComponent(channel).replace(/\./g, '%2E') + +/** + * The `uid` claim of a Firebase custom token. + * + * @param token + * @returns + */ +const tokenUid = (token: string): string | null => { + try { + const segment = token.split('.')[1].replace(/-/g, '+').replace(/_/g, '/') + + return (JSON.parse(atob(segment)) as { uid?: string }).uid ?? null + } catch { + return null + } +} + +/** + * Realtime transport backed by [Firebase Realtime Database](https://firebase.google.com/docs/database/web/start). + * `firebase` is an optional peer dependency imported lazily. + * + * Each subscription first asks the API to authorize the channel, which grants + * this user read access and returns a custom token to sign in with. Only events + * written after the transport connected are delivered. + * + * @param config + * @returns + */ +export const createFirebaseDatabaseTransport = async ( + config: FirebaseDatabaseClientConfig, +): Promise => { + const appSpecifier = 'firebase/app' + const authSpecifier = 'firebase/auth' + const databaseSpecifier = 'firebase/database' + + const [appMod, authMod, databaseMod] = await Promise.all([ + import(appSpecifier) as Promise, + import(authSpecifier) as Promise, + import(databaseSpecifier) as Promise, + ]).catch(() => { + throw new Error( + 'The "firebase" package is required for the Firebase database transport. Install it with `npm i firebase`.', + ) + }) + + const app = appMod.getApps().find((existing) => existing.name === APP_NAME) ?? appMod.initializeApp({ + apiKey: config.apiKey, + authDomain: config.authDomain, + projectId: config.projectId, + appId: config.appId, + databaseURL: config.databaseURL, + }, APP_NAME) + + const auth = authMod.getAuth(app) + const database = databaseMod.getDatabase(app, config.databaseURL) + const root = config.root ?? 'arkstack/channels' + const endpoint = config.authEndpoint ?? `${config.apiBase}/realtime/auth` + const listeners = new Set<() => void>() + + const offset = await new Promise((resolve) => { + databaseMod.onValue( + databaseMod.ref(database, '.info/serverTimeOffset'), + (snapshot) => resolve(Number(snapshot.val()) || 0), + { onlyOnce: true }, + ) + }) + const since = Date.now() + offset + + // Sign-ins are chained so concurrent subscriptions do not race each other. + let signing: Promise = Promise.resolve() + + const authorize = async (channel: string): Promise => { + const response = await fetch(endpoint, { + method: 'POST', + headers: { + 'content-type': 'application/json', + accept: 'application/json', + ...config.auth?.headers, + }, + body: JSON.stringify({ channel_name: channel }), + }) + + if (!response.ok) { + throw new Error(`Realtime: channel "${channel}" was not authorized (${response.status})`) + } + + const { token } = await response.json() as { token: string } + + signing = signing.then(async () => { + if (auth.currentUser?.uid !== tokenUid(token)) { + await authMod.signInWithCustomToken(auth, token) + } + }) + + await signing + } + + return { + async subscribe(channel: string, event: string, handler: RealtimeEventHandler) { + await authorize(channel) + + const off = databaseMod.onChildAdded( + databaseMod.query( + databaseMod.ref(database, `${root}/${channelKey(channel)}`), + databaseMod.orderByChild('at'), + databaseMod.startAt(since), + ), + (snapshot) => { + const value = snapshot.val() as ChannelEvent | null + + if (value?.event === event) { + handler(value.payload) + } + }, + ) + + listeners.add(off) + + return { + channel, + unsubscribe() { + off() + listeners.delete(off) + }, + } + }, + async disconnect() { + listeners.forEach((off) => off()) + listeners.clear() + await authMod.signOut(auth) + }, + } +} diff --git a/packages/realtime/src/transports/firebase.ts b/packages/realtime/src/transports/firebase.ts index e90b41ca..2dd3ac77 100644 --- a/packages/realtime/src/transports/firebase.ts +++ b/packages/realtime/src/transports/firebase.ts @@ -126,6 +126,14 @@ export const createFirebaseTransport = async (config: FirebaseClientConfig): Pro await databaseApi.remove(eventReference) }, + async deviceToken() { + const getToken = messagingMod.getToken as ( + messaging: unknown, + options?: { vapidKey?: string }, + ) => Promise + + return (await getToken(messaging, config.vapidKey ? { vapidKey: config.vapidKey } : undefined)) || null + }, disconnect() { messageUnsubscribers.forEach((off) => off()) messageUnsubscribers.clear() diff --git a/packages/realtime/src/transports/memory.ts b/packages/realtime/src/transports/memory.ts new file mode 100644 index 00000000..c62e9a09 --- /dev/null +++ b/packages/realtime/src/transports/memory.ts @@ -0,0 +1,66 @@ +import type { RealtimeEventHandler, RealtimeTransport } from '../types' + +type Listener = { event: string, handler: RealtimeEventHandler } + +const BUS_KEY = Symbol.for('arkstack.realtime.client-memory') + +/** + * The in-process bus, kept on `globalThis` so every module copy shares it. + * + * @returns + */ +const bus = (): Map> => { + const store = globalThis as Record> | undefined> + + return (store[BUS_KEY] ??= new Map()) +} + +/** + * Deliver an event to every in-process subscriber of `channel`, as a server + * broadcast would. + * + * @param channel + * @param event + * @param payload + */ +export const emitMemoryEvent = (channel: string, event: string, payload: unknown): void => { + bus().get(channel)?.forEach((listener) => { + if (listener.event === event) listener.handler(payload) + }) +} + +/** + * An in-process transport for tests. Feed it with {@link emitMemoryEvent}. + * + * @returns + */ +export const createMemoryTransport = (): RealtimeTransport => { + const own = new Set<{ channel: string, listener: Listener }>() + + return { + subscribe(channel: string, event: string, handler: RealtimeEventHandler) { + const listener = { event, handler } + const listeners = bus().get(channel) ?? new Set() + const entry = { channel, listener } + + listeners.add(listener) + bus().set(channel, listeners) + own.add(entry) + + return { + channel, + unsubscribe() { + listeners.delete(listener) + own.delete(entry) + }, + } + }, + trigger(channel: string, event: string, payload: unknown) { + emitMemoryEvent(channel, event, payload) + }, + disconnect() { + own.forEach(({ channel, listener }) => bus().get(channel)?.delete(listener)) + own.clear() + }, + } +} diff --git a/packages/realtime/src/transports/polling.ts b/packages/realtime/src/transports/polling.ts new file mode 100644 index 00000000..561d6783 --- /dev/null +++ b/packages/realtime/src/transports/polling.ts @@ -0,0 +1,16 @@ +import type { PollingClientConfig, RealtimeTransport } from '../types' + +/** + * A transport that receives nothing. Subscriptions are accepted and stay + * silent; use {@link RealtimeClient.onPoll} to refetch on the advertised interval. + * + * @param config + * @returns + */ +export const createPollingTransport = (config: PollingClientConfig = {}): RealtimeTransport => ({ + polling: { interval: config.interval ?? 5000 }, + subscribe(channel: string) { + return { channel, unsubscribe() { } } + }, + disconnect() { }, +}) diff --git a/packages/realtime/src/transports/pusher.ts b/packages/realtime/src/transports/pusher.ts index 9975dd85..75ec8efe 100644 --- a/packages/realtime/src/transports/pusher.ts +++ b/packages/realtime/src/transports/pusher.ts @@ -39,6 +39,14 @@ export const createPusherTransport = async ( forceTLS: config.forceTLS ?? true, authEndpoint: config.authEndpoint ?? `${config.apiBase}/realtime/auth`, auth: config.auth, + ...(config.wsHost + ? { + wsHost: config.wsHost, + wsPort: config.wsPort, + wssPort: config.wssPort ?? config.wsPort, + enabledTransports: config.enabledTransports ?? ['ws', 'wss'], + } + : {}), }) const channels = new Map() diff --git a/packages/realtime/src/types.ts b/packages/realtime/src/types.ts index a239d0dd..986aa310 100644 --- a/packages/realtime/src/types.ts +++ b/packages/realtime/src/types.ts @@ -23,7 +23,16 @@ export interface RealtimeNotification { created_at: string } -export type RealtimeTransportName = 'pusher' | 'firebase' +/** + * Client transports, matching the server's `drivers.realtime.transport`. + * + * - `pusher`: Pusher Channels, or a self-hosted Pusher-protocol server via `wsHost`. + * - `firebase`: Firebase Cloud Messaging foreground messages. + * - `firebase-database`: Firebase Realtime Database channels. + * - `polling`: nothing is pushed; use {@link RealtimeClient.onPoll} to refetch. + * - `memory`: an in-process bus, for tests. + */ +export type RealtimeTransportName = 'pusher' | 'firebase' | 'firebase-database' | 'polling' | 'memory' export type NotificationHandler = (notification: RealtimeNotification) => void @@ -70,6 +79,14 @@ export interface RealtimeTransport { * Disconnect from a connected realtime channel */ disconnect(): void | Promise + /** + * This device's push registration token, when the transport can issue one. + */ + deviceToken?(): Promise + /** + * Set by transports that push nothing; clients should poll this often instead. + */ + polling?: { interval: number } } export interface PusherClientConfig { @@ -90,6 +107,11 @@ export interface PusherClientConfig { authEndpoint?: string auth?: { headers?: Record, params?: Record } forceTLS?: boolean + /** A self-hosted Pusher-protocol server such as soketi. */ + wsHost?: string + wsPort?: number + wssPort?: number + enabledTransports?: Array<'ws' | 'wss'> } export interface FirebaseClientConfig { @@ -105,6 +127,29 @@ export interface FirebaseClientConfig { vapidKey?: string } +export interface FirebaseDatabaseClientConfig { + apiKey: string + projectId: string + databaseURL: string + appId?: string + authDomain?: string + /** Database path that holds channel events (default `arkstack/channels`). */ + root?: string + /** + * Your API's base URL; channels are authorized at `{apiBase}/realtime/auth`. + * Ignored when {@link authEndpoint} is set. + */ + apiBase?: string + /** Endpoint that grants channel access and returns a Firebase custom token. */ + authEndpoint?: string + auth?: { headers?: Record } +} + +export interface PollingClientConfig { + /** How often, in milliseconds, {@link RealtimeClient.onPoll} handlers run (default 5000). */ + interval?: number +} + export interface RealtimeConfig { transport?: RealtimeTransportName /** Event name broadcasts are published under (default `notification`). */ @@ -113,6 +158,8 @@ export interface RealtimeConfig { channelPrefix?: string pusher?: PusherClientConfig firebase?: FirebaseClientConfig + 'firebase-database'?: FirebaseDatabaseClientConfig + polling?: PollingClientConfig /** Inject a transport directly — bypasses the built-ins (tests, custom backends). */ transportFactory?: () => RealtimeTransport | Promise } diff --git a/packages/realtime/tests/transports.test.ts b/packages/realtime/tests/transports.test.ts new file mode 100644 index 00000000..81bb6049 --- /dev/null +++ b/packages/realtime/tests/transports.test.ts @@ -0,0 +1,275 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { createFirebaseDatabaseTransport, createRealtime, emitMemoryEvent } from '../src' + +const fakes = vi.hoisted(() => ({ + pusherOptions: [] as Array>, + childListeners: new Map void>(), + queries: [] as unknown[][], + signIns: [] as string[], + currentUser: null as { uid: string } | null, + signedOut: 0, +})) + +vi.mock('pusher-js', () => ({ + default: class { + constructor(_key: string, options: Record) { + fakes.pusherOptions.push(options) + } + + subscribe() { + return { bind() { }, unbind() { }, trigger: () => true } + } + + unsubscribe() { } + + disconnect() { } + }, +})) + +vi.mock('firebase/app', () => ({ + initializeApp: vi.fn((_options: unknown, name = '[DEFAULT]') => ({ name })), + getApps: vi.fn(() => []), +})) + +vi.mock('firebase/messaging', () => ({ + getMessaging: vi.fn(() => ({})), + onMessage: vi.fn(() => vi.fn()), + getToken: vi.fn(async (_messaging: unknown, options?: { vapidKey?: string }) => `fcm-token:${options?.vapidKey}`), +})) + +vi.mock('firebase/auth', () => ({ + getAuth: vi.fn(() => ({ + get currentUser() { + return fakes.currentUser + }, + })), + signInWithCustomToken: vi.fn(async (_auth: unknown, token: string) => { + fakes.signIns.push(token) + fakes.currentUser = { uid: JSON.parse(atob(token.split('.')[1])).uid } + }), + signOut: vi.fn(async () => { + fakes.signedOut++ + fakes.currentUser = null + }), +})) + +vi.mock('firebase/database', () => ({ + getDatabase: vi.fn(() => ({})), + ref: vi.fn((_database: unknown, path: string) => ({ path })), + query: vi.fn((reference: { path: string }, ...constraints: unknown[]) => { + fakes.queries.push(constraints) + + return reference + }), + orderByChild: vi.fn((path: string) => ({ orderByChild: path })), + startAt: vi.fn((value: number) => ({ startAt: value })), + onValue: vi.fn((_reference: unknown, handler: (snapshot: { val(): unknown }) => void) => { + handler({ val: () => 250 }) + + return vi.fn() + }), + onChildAdded: vi.fn((reference: { path: string }, handler: (snapshot: { val(): unknown }) => void) => { + fakes.childListeners.set(reference.path, handler) + + return () => fakes.childListeners.delete(reference.path) + }), +})) + +/** A Firebase custom token whose payload carries `uid`. */ +const customToken = (uid: string) => `header.${btoa(JSON.stringify({ uid }))}.signature` + +afterEach(() => { + vi.useRealTimers() + vi.unstubAllGlobals() + fakes.pusherOptions.length = 0 + fakes.childListeners.clear() + fakes.queries.length = 0 + fakes.signIns.length = 0 + fakes.currentUser = null + fakes.signedOut = 0 +}) + +describe('polling transport', () => { + it('runs poll handlers on the advertised interval until stopped', async () => { + vi.useFakeTimers() + const client = createRealtime({ transport: 'polling', polling: { interval: 1000 } }) + const handler = vi.fn() + + const stop = await client.onPoll(handler) + vi.advanceTimersByTime(3000) + stop() + vi.advanceTimersByTime(3000) + + expect(handler).toHaveBeenCalledTimes(3) + }) + + it('accepts subscriptions that never fire', async () => { + const client = createRealtime({ transport: 'polling' }) + const handler = vi.fn() + + const off = await client.forUser(1, handler) + off() + + expect(handler).not.toHaveBeenCalled() + }) + + it('does not poll on push transports', async () => { + vi.useFakeTimers() + const client = createRealtime({ transport: 'memory' }) + const handler = vi.fn() + + await client.onPoll(handler, 10) + vi.advanceTimersByTime(100) + + expect(handler).not.toHaveBeenCalled() + }) +}) + +describe('memory transport', () => { + it('delivers emitted events to matching subscribers only', async () => { + const client = createRealtime({ transport: 'memory' }) + const notifications = vi.fn() + const custom = vi.fn() + + await client.forUser(5, notifications) + await client.listen('team.1', 'deploy', custom) + + emitMemoryEvent('user.5', 'notification', { id: 'n1' }) + emitMemoryEvent('user.5', 'other', { id: 'ignored' }) + emitMemoryEvent('team.1', 'deploy', { ok: true }) + emitMemoryEvent('team.2', 'deploy', { ok: false }) + + expect(notifications).toHaveBeenCalledWith({ id: 'n1' }) + expect(notifications).toHaveBeenCalledTimes(1) + expect(custom).toHaveBeenCalledWith({ ok: true }) + expect(custom).toHaveBeenCalledTimes(1) + }) + + it('stops delivering after disconnect', async () => { + const client = createRealtime({ transport: 'memory' }) + const handler = vi.fn() + + await client.forUser(6, handler) + await client.disconnect() + emitMemoryEvent('user.6', 'notification', {}) + + expect(handler).not.toHaveBeenCalled() + }) + + it('reports no device token', async () => { + await expect(createRealtime({ transport: 'memory' }).deviceToken()).resolves.toBeNull() + }) +}) + +describe('pusher transport', () => { + it('connects to a self-hosted server when wsHost is set', async () => { + const client = createRealtime({ + transport: 'pusher', + pusher: { key: 'k', apiBase: '/api', wsHost: 'soketi.test', wsPort: 6001, forceTLS: false }, + }) + + await client.forUser(1, vi.fn()) + + expect(fakes.pusherOptions[0]).toMatchObject({ + wsHost: 'soketi.test', + wsPort: 6001, + wssPort: 6001, + enabledTransports: ['ws', 'wss'], + authEndpoint: '/api/realtime/auth', + }) + }) + + it('leaves hosted Pusher addressing alone', async () => { + await createRealtime({ transport: 'pusher', pusher: { key: 'k', cluster: 'eu' } }).forUser(1, vi.fn()) + + expect(fakes.pusherOptions[0]).not.toHaveProperty('wsHost') + expect(fakes.pusherOptions[0].cluster).toBe('eu') + }) +}) + +describe('firebase messaging transport', () => { + it('issues a device token with the VAPID key', async () => { + const client = createRealtime({ + transport: 'firebase', + firebase: { apiKey: 'a', projectId: 'p', appId: 'i', messagingSenderId: 'm', vapidKey: 'v' }, + }) + + await expect(client.deviceToken()).resolves.toBe('fcm-token:v') + }) +}) + +describe('firebase database transport', () => { + const config = { + apiKey: 'a', + projectId: 'p', + databaseURL: 'https://p.firebaseio.com', + apiBase: '/api', + auth: { headers: { Authorization: 'Bearer t' } }, + } + + const stubAuth = (uid = 'u-1', ok = true) => { + const fetch = vi.fn(async () => ({ + ok, + status: ok ? 200 : 403, + json: async () => ({ token: customToken(uid) }), + })) + vi.stubGlobal('fetch', fetch) + + return fetch + } + + it('authorizes the channel, signs in once, and delivers matching events', async () => { + const fetch = stubAuth() + vi.spyOn(Date, 'now').mockReturnValue(10_000) + const transport = await createFirebaseDatabaseTransport(config) + const handler = vi.fn() + + await transport.subscribe('private-user.1', 'notification', handler) + await transport.subscribe('private-team.2', 'notification', vi.fn()) + + expect(fetch).toHaveBeenCalledWith('/api/realtime/auth', expect.objectContaining({ + method: 'POST', + body: JSON.stringify({ channel_name: 'private-user.1' }), + headers: expect.objectContaining({ Authorization: 'Bearer t' }), + })) + expect(fakes.signIns).toHaveLength(1) + // Only events written after connecting, on the server clock. + expect(fakes.queries[0]).toEqual([{ orderByChild: 'at' }, { startAt: 10_250 }]) + + const emit = fakes.childListeners.get('arkstack/channels/private-user%2E1')! + emit({ val: () => ({ event: 'notification', payload: { id: 'n1' } }) }) + emit({ val: () => ({ event: 'other', payload: { id: 'n2' } }) }) + + expect(handler).toHaveBeenCalledWith({ id: 'n1' }) + expect(handler).toHaveBeenCalledTimes(1) + }) + + it('signs in again when the token belongs to a different user', async () => { + stubAuth('u-2') + fakes.currentUser = { uid: 'u-1' } + const transport = await createFirebaseDatabaseTransport(config) + + await transport.subscribe('private-user.2', 'notification', vi.fn()) + + expect(fakes.signIns).toEqual([customToken('u-2')]) + }) + + it('rejects a channel the API refuses', async () => { + stubAuth('u-1', false) + const transport = await createFirebaseDatabaseTransport(config) + + await expect(transport.subscribe('private-user.9', 'notification', vi.fn())).rejects.toThrow(/not authorized \(403\)/) + expect(fakes.childListeners.size).toBe(0) + }) + + it('stops listening and signs out on disconnect', async () => { + stubAuth() + const transport = await createFirebaseDatabaseTransport(config) + + await transport.subscribe('private-user.1', 'notification', vi.fn()) + await transport.disconnect() + + expect(fakes.childListeners.size).toBe(0) + expect(fakes.signedOut).toBe(1) + }) +}) diff --git a/templates/express-inertia/src/config/notifications.ts b/templates/express-inertia/src/config/notifications.ts index 07cacb7b..14d8a10b 100644 --- a/templates/express-inertia/src/config/notifications.ts +++ b/templates/express-inertia/src/config/notifications.ts @@ -33,7 +33,10 @@ export default (): NotificationConfig => { table: 'user_notifications', }, realtime: { - transport: env('REALTIME_TRANSPORT', 'pusher') as 'pusher' | 'firebase', + /** + * pusher | firebase | firebase-database | polling | memory + */ + transport: env('REALTIME_TRANSPORT', 'pusher'), channel_prefix: env('REALTIME_CHANNEL_PREFIX', 'user.'), event: env('REALTIME_EVENT', 'notification'), store: env('REALTIME_STORE', false), @@ -116,6 +119,40 @@ export default (): NotificationConfig => { private_key: env('FIREBASE_PRIVATE_KEY', ''), app_name: env('APP_NAME', 'Arkstack'), admin_sdk_path: env('FIREBASE_ADMINSDK', 'firebase-adminsdk.json'), + web: { + api_key: env('FIREBASE_API_KEY', ''), + app_id: env('FIREBASE_APP_ID', ''), + messaging_sender_id: env('FIREBASE_MESSAGING_SENDER_ID', ''), + vapid_key: env('FIREBASE_VAPID_KEY', ''), + }, + }, + + /** + * Firebase Realtime Database Transport + * + * Realtime notifications will be broadcast over Realtime Database channels. + */ + 'firebase-database': { + project_id: env('FIREBASE_PROJECT_ID', ''), + client_email: env('FIREBASE_CLIENT_EMAIL', ''), + private_key: env('FIREBASE_PRIVATE_KEY', ''), + app_name: env('APP_NAME', 'Arkstack'), + admin_sdk_path: env('FIREBASE_ADMINSDK', 'firebase-adminsdk.json'), + database_url: env('FIREBASE_DATABASE_URL', ''), + web: { + api_key: env('FIREBASE_API_KEY', ''), + auth_domain: env('FIREBASE_AUTH_DOMAIN', ''), + app_id: env('FIREBASE_APP_ID', ''), + }, + }, + + /** + * Polling Realtime Transport + * + * Nothing is pushed; clients poll on this interval (ms) instead. + */ + polling: { + interval: env('REALTIME_POLL_INTERVAL', 5000), }, } } diff --git a/templates/express/src/config/notifications.ts b/templates/express/src/config/notifications.ts index 07cacb7b..5777b6b0 100644 --- a/templates/express/src/config/notifications.ts +++ b/templates/express/src/config/notifications.ts @@ -33,7 +33,14 @@ export default (): NotificationConfig => { table: 'user_notifications', }, realtime: { - transport: env('REALTIME_TRANSPORT', 'pusher') as 'pusher' | 'firebase', + /** + * pusher | firebase | firebase-database | polling | memory + */ + transport: env('REALTIME_TRANSPORT', 'pusher'), + /** + * Also push user notifications to their devices: 'firebase' or false. + */ + push: env('REALTIME_PUSH', false), channel_prefix: env('REALTIME_CHANNEL_PREFIX', 'user.'), event: env('REALTIME_EVENT', 'notification'), store: env('REALTIME_STORE', false), @@ -103,6 +110,9 @@ export default (): NotificationConfig => { secret: env('PUSHER_SECRET', ''), cluster: env('PUSHER_CLUSTER', 'mt1'), use_tls: env('PUSHER_USE_TLS', true), + // Set to use a self-hosted Pusher-protocol server such as soketi. + host: env('PUSHER_HOST', undefined), + port: env('PUSHER_PORT', undefined), }, /** @@ -116,6 +126,40 @@ export default (): NotificationConfig => { private_key: env('FIREBASE_PRIVATE_KEY', ''), app_name: env('APP_NAME', 'Arkstack'), admin_sdk_path: env('FIREBASE_ADMINSDK', 'firebase-adminsdk.json'), + web: { + api_key: env('FIREBASE_API_KEY', ''), + app_id: env('FIREBASE_APP_ID', ''), + messaging_sender_id: env('FIREBASE_MESSAGING_SENDER_ID', ''), + vapid_key: env('FIREBASE_VAPID_KEY', ''), + }, + }, + + /** + * Firebase Realtime Database Transport + * + * Realtime notifications will be broadcast over Realtime Database channels. + */ + 'firebase-database': { + project_id: env('FIREBASE_PROJECT_ID', ''), + client_email: env('FIREBASE_CLIENT_EMAIL', ''), + private_key: env('FIREBASE_PRIVATE_KEY', ''), + app_name: env('APP_NAME', 'Arkstack'), + admin_sdk_path: env('FIREBASE_ADMINSDK', 'firebase-adminsdk.json'), + database_url: env('FIREBASE_DATABASE_URL', ''), + web: { + api_key: env('FIREBASE_API_KEY', ''), + auth_domain: env('FIREBASE_AUTH_DOMAIN', ''), + app_id: env('FIREBASE_APP_ID', ''), + }, + }, + + /** + * Polling Realtime Transport + * + * Nothing is pushed; clients poll on this interval (ms) instead. + */ + polling: { + interval: env('REALTIME_POLL_INTERVAL', 5000), }, } } diff --git a/templates/h3/src/config/notifications.ts b/templates/h3/src/config/notifications.ts index 07cacb7b..14d8a10b 100644 --- a/templates/h3/src/config/notifications.ts +++ b/templates/h3/src/config/notifications.ts @@ -33,7 +33,10 @@ export default (): NotificationConfig => { table: 'user_notifications', }, realtime: { - transport: env('REALTIME_TRANSPORT', 'pusher') as 'pusher' | 'firebase', + /** + * pusher | firebase | firebase-database | polling | memory + */ + transport: env('REALTIME_TRANSPORT', 'pusher'), channel_prefix: env('REALTIME_CHANNEL_PREFIX', 'user.'), event: env('REALTIME_EVENT', 'notification'), store: env('REALTIME_STORE', false), @@ -116,6 +119,40 @@ export default (): NotificationConfig => { private_key: env('FIREBASE_PRIVATE_KEY', ''), app_name: env('APP_NAME', 'Arkstack'), admin_sdk_path: env('FIREBASE_ADMINSDK', 'firebase-adminsdk.json'), + web: { + api_key: env('FIREBASE_API_KEY', ''), + app_id: env('FIREBASE_APP_ID', ''), + messaging_sender_id: env('FIREBASE_MESSAGING_SENDER_ID', ''), + vapid_key: env('FIREBASE_VAPID_KEY', ''), + }, + }, + + /** + * Firebase Realtime Database Transport + * + * Realtime notifications will be broadcast over Realtime Database channels. + */ + 'firebase-database': { + project_id: env('FIREBASE_PROJECT_ID', ''), + client_email: env('FIREBASE_CLIENT_EMAIL', ''), + private_key: env('FIREBASE_PRIVATE_KEY', ''), + app_name: env('APP_NAME', 'Arkstack'), + admin_sdk_path: env('FIREBASE_ADMINSDK', 'firebase-adminsdk.json'), + database_url: env('FIREBASE_DATABASE_URL', ''), + web: { + api_key: env('FIREBASE_API_KEY', ''), + auth_domain: env('FIREBASE_AUTH_DOMAIN', ''), + app_id: env('FIREBASE_APP_ID', ''), + }, + }, + + /** + * Polling Realtime Transport + * + * Nothing is pushed; clients poll on this interval (ms) instead. + */ + polling: { + interval: env('REALTIME_POLL_INTERVAL', 5000), }, } }