mirror of
https://github.com/mauriceboe/TREK.git
synced 2026-11-14 00:01:03 +00:00
feat(native): a silent push keeps every watch on a trip current
A change to a trip's packing lists or to-dos (any writer: REST, MCP, plugin RPC) is noted from the realtime broadcast and, a couple of seconds later, each member with the iPhone app gets a silent push through the app relay, marked as a background push. The push carries no content: the woken app reads both lists of the trip its watch shows from the server with its own session and hands them to the watch. The app gains the remote-notification background mode for it.
This commit is contained in:
@@ -25,6 +25,14 @@ class AppDelegate: UIResponder, UIApplicationDelegate, UNUserNotificationCenterD
|
||||
TrekPush.shared.didFail(error)
|
||||
}
|
||||
|
||||
/// A silent push from the TREK server: a list of a trip changed, so the Apple
|
||||
/// Watch is brought up to date (TrekWatchRefresher). Nothing else arrives this way.
|
||||
func application(_ application: UIApplication, didReceiveRemoteNotification userInfo: [AnyHashable: Any],
|
||||
fetchCompletionHandler completionHandler: @escaping (UIBackgroundFetchResult) -> Void) {
|
||||
guard userInfo["trek"] as? String == "watch-sync" else { return completionHandler(.noData) }
|
||||
TrekWatchRefresher.refresh(completion: completionHandler)
|
||||
}
|
||||
|
||||
func userNotificationCenter(_ center: UNUserNotificationCenter, willPresent notification: UNNotification,
|
||||
withCompletionHandler completionHandler: @escaping (UNNotificationPresentationOptions) -> Void) {
|
||||
completionHandler([.banner, .list, .sound])
|
||||
|
||||
@@ -120,6 +120,10 @@
|
||||
</array>
|
||||
</dict>
|
||||
</array>
|
||||
<key>UIBackgroundModes</key>
|
||||
<array>
|
||||
<string>remote-notification</string>
|
||||
</array>
|
||||
<key>UIApplicationSceneManifest</key>
|
||||
<dict>
|
||||
<key>UIApplicationSupportsMultipleScenes</key>
|
||||
|
||||
@@ -94,6 +94,11 @@ final class TrekWatchLink: NSObject, WCSessionDelegate {
|
||||
return result
|
||||
}
|
||||
|
||||
/// The snapshot the watch has now, as last published.
|
||||
func storedSnapshot() -> Data? {
|
||||
queue.sync { UserDefaults.standard.data(forKey: snapshotKey) }
|
||||
}
|
||||
|
||||
// MARK: Ticks from the watch
|
||||
|
||||
/// The ticks queued so far, oldest first, and the queue emptied.
|
||||
@@ -203,7 +208,7 @@ enum TrekWatchWriter {
|
||||
|
||||
/// Called on the main queue, as is `completion`.
|
||||
static func write(_ action: [String: Any], completion: @escaping (Bool) -> Void) {
|
||||
guard let request = request(for: action) else {
|
||||
guard let request = Self.request(for: action) else {
|
||||
completion(false)
|
||||
return
|
||||
}
|
||||
@@ -215,24 +220,37 @@ enum TrekWatchWriter {
|
||||
task.end()
|
||||
}
|
||||
}
|
||||
perform(request) { _, ok in finish(ok) }
|
||||
}
|
||||
|
||||
/// Sends a request with the WebView's session (its cookies for the server).
|
||||
/// Called on the main queue; `completion` runs on a background queue.
|
||||
static func perform(_ request: URLRequest, completion: @escaping (Data?, Bool) -> Void) {
|
||||
WKWebsiteDataStore.default().httpCookieStore.getAllCookies { cookies in
|
||||
guard let host = request.url?.host else { return finish(false) }
|
||||
guard let host = request.url?.host else { return completion(nil, false) }
|
||||
let session = cookies.filter { cookie in
|
||||
let domain = cookie.domain.hasPrefix(".") ? String(cookie.domain.dropFirst()) : cookie.domain
|
||||
return host == domain || host.hasSuffix("." + domain)
|
||||
}
|
||||
guard !session.isEmpty else { return finish(false) }
|
||||
guard !session.isEmpty else { return completion(nil, false) }
|
||||
var signed = request
|
||||
for (field, value) in HTTPCookie.requestHeaderFields(with: session) {
|
||||
signed.setValue(value, forHTTPHeaderField: field)
|
||||
}
|
||||
URLSession.shared.dataTask(with: signed) { _, response, error in
|
||||
URLSession.shared.dataTask(with: signed) { data, response, error in
|
||||
let status = (response as? HTTPURLResponse)?.statusCode ?? 0
|
||||
finish(error == nil && (200..<300).contains(status))
|
||||
completion(data, error == nil && (200..<300).contains(status))
|
||||
}.resume()
|
||||
}
|
||||
}
|
||||
|
||||
/// The server's address for an API path, or nil without a server.
|
||||
static func url(_ path: [String]) -> URL? {
|
||||
guard let server = TrekServerStore.url, let base = URL(string: server),
|
||||
base.scheme == "https" || base.scheme == "http" else { return nil }
|
||||
return path.reduce(base.appendingPathComponent("api")) { $0.appendingPathComponent($1) }
|
||||
}
|
||||
|
||||
/// The write for a tick as the watch sends it, or nil for anything else.
|
||||
static func request(for action: [String: Any]) -> URLRequest? {
|
||||
guard let server = TrekServerStore.url, let base = URL(string: server),
|
||||
@@ -253,3 +271,129 @@ enum TrekWatchWriter {
|
||||
return request
|
||||
}
|
||||
}
|
||||
|
||||
/// Brings the watch's packing list and to-dos up to date while the app is in the
|
||||
/// background, woken by the server's silent push (WatchSyncPushService in TREK):
|
||||
/// reads both lists of the trip the watch shows from the server and puts them
|
||||
/// into the snapshot the watch has, built the way the UI builds them
|
||||
/// (watchPacking and watchTodos in client/src/native/watch.ts). Everything else
|
||||
/// in the snapshot stays as the UI last made it. In front, the UI does this
|
||||
/// itself as soon as the lists change, so nothing is done here then.
|
||||
enum TrekWatchRefresher {
|
||||
private static let maxName = 80
|
||||
private static let maxGroup = 40
|
||||
/// Below what WatchConnectivity carries in one application context.
|
||||
private static let maxBytes = 55_000
|
||||
|
||||
static func refresh(completion: @escaping (UIBackgroundFetchResult) -> Void) {
|
||||
guard TrekWatchLink.isAvailable, UIApplication.shared.applicationState != .active,
|
||||
let stored = TrekWatchLink.shared.storedSnapshot(),
|
||||
var snapshot = (try? JSONSerialization.jsonObject(with: stored)) as? [String: Any],
|
||||
let trip = snapshot["trip"] as? [String: Any], let tripId = trip["id"] as? Int,
|
||||
let packingURL = TrekWatchWriter.url(["trips", String(tripId), "packing"]),
|
||||
let todoURL = TrekWatchWriter.url(["trips", String(tripId), "todo"]) else {
|
||||
return completion(.noData)
|
||||
}
|
||||
let group = DispatchGroup()
|
||||
var packing: [[String: Any]]?
|
||||
var todos: [[String: Any]]?
|
||||
group.enter()
|
||||
items(at: packingURL) { list in
|
||||
packing = list
|
||||
group.leave()
|
||||
}
|
||||
group.enter()
|
||||
items(at: todoURL) { list in
|
||||
todos = list
|
||||
group.leave()
|
||||
}
|
||||
group.notify(queue: .main) {
|
||||
guard let packing, let todos else { return completion(.failed) }
|
||||
snapshot["packing"] = packingList(packing)
|
||||
snapshot["todos"] = todoList(todos)
|
||||
snapshot["generatedAt"] = ISO8601DateFormatter().string(from: Date())
|
||||
guard let data = fitted(snapshot) else { return completion(.failed) }
|
||||
TrekWatchLink.shared.publish(data)
|
||||
completion(.newData)
|
||||
}
|
||||
}
|
||||
|
||||
/// The `items` of a list endpoint, on the main queue; nil when it could not be read.
|
||||
private static func items(at url: URL, then completion: @escaping ([[String: Any]]?) -> Void) {
|
||||
var request = URLRequest(url: url, timeoutInterval: 10)
|
||||
request.httpShouldHandleCookies = false
|
||||
request.setValue("application/json", forHTTPHeaderField: "Accept")
|
||||
TrekWatchWriter.perform(request) { data, ok in
|
||||
var list: [[String: Any]]?
|
||||
if ok, let data, let body = (try? JSONSerialization.jsonObject(with: data)) as? [String: Any] {
|
||||
list = body["items"] as? [[String: Any]]
|
||||
}
|
||||
DispatchQueue.main.async { completion(list) }
|
||||
}
|
||||
}
|
||||
|
||||
private static func done(_ item: [String: Any]) -> Bool {
|
||||
(item["checked"] as? Bool) ?? ((item["checked"] as? Int).map { $0 != 0 } ?? false)
|
||||
}
|
||||
|
||||
private static func trimmed(_ value: Any?, _ max: Int) -> String? {
|
||||
guard let text = (value as? String)?.trimmingCharacters(in: .whitespacesAndNewlines), !text.isEmpty else { return nil }
|
||||
return String(text.prefix(max))
|
||||
}
|
||||
|
||||
private static func entry(_ item: [String: Any]) -> [String: Any]? {
|
||||
guard let id = item["id"] as? Int, let name = item["name"] as? String else { return nil }
|
||||
var out: [String: Any] = ["id": id, "name": String(name.prefix(maxName)), "done": done(item)]
|
||||
if let group = trimmed(item["category"], maxGroup) { out["group"] = group }
|
||||
return out
|
||||
}
|
||||
|
||||
/// Unpacked first, then packed, each in the list's own order.
|
||||
static func packingList(_ items: [[String: Any]]) -> Any {
|
||||
guard !items.isEmpty else { return NSNull() }
|
||||
let sorted = items.sorted {
|
||||
let (a, b) = (done($0) ? 1 : 0, done($1) ? 1 : 0)
|
||||
return a != b ? a < b : (($0["sort_order"] as? Int) ?? 0) < (($1["sort_order"] as? Int) ?? 0)
|
||||
}
|
||||
let packed = items.filter { done($0) }.count
|
||||
return ["items": sorted.compactMap(entry), "packed": packed, "total": items.count]
|
||||
}
|
||||
|
||||
/// Open ones first by due date (undated last) and priority, then the done ones.
|
||||
static func todoList(_ items: [[String: Any]]) -> Any {
|
||||
guard !items.isEmpty else { return NSNull() }
|
||||
let due = { (item: [String: Any]) -> String? in trimmed(item["due_date"], 10) }
|
||||
let sorted = items.sorted { a, b in
|
||||
if done(a) != done(b) { return !done(a) }
|
||||
if (due(a) == nil) != (due(b) == nil) { return due(a) != nil }
|
||||
if let da = due(a), let db = due(b), da != db { return da < db }
|
||||
let (pa, pb) = ((a["priority"] as? Int) ?? 0, (b["priority"] as? Int) ?? 0)
|
||||
if pa != pb { return pa > pb }
|
||||
return ((a["sort_order"] as? Int) ?? 0) < ((b["sort_order"] as? Int) ?? 0)
|
||||
}
|
||||
let rows = sorted.compactMap { item -> [String: Any]? in
|
||||
guard var row = entry(item) else { return nil }
|
||||
if let date = due(item) { row["due"] = date }
|
||||
if let priority = item["priority"] as? Int, priority > 0 { row["priority"] = min(3, priority) }
|
||||
return row
|
||||
}
|
||||
return ["items": rows, "open": items.filter { !done($0) }.count, "total": items.count]
|
||||
}
|
||||
|
||||
/// The snapshot as JSON within the size the watch takes: done items go first, as on the UI's side.
|
||||
private static func fitted(_ snapshot: [String: Any]) -> Data? {
|
||||
var object = snapshot
|
||||
// A list that is now empty leaves the snapshot, as the UI leaves it out.
|
||||
for key in ["packing", "todos"] where object[key] is NSNull { object.removeValue(forKey: key) }
|
||||
guard var data = try? JSONSerialization.data(withJSONObject: object) else { return nil }
|
||||
if data.count > maxBytes {
|
||||
for key in ["packing", "todos"] {
|
||||
guard var list = object[key] as? [String: Any], let items = list["items"] as? [[String: Any]] else { continue }
|
||||
list["items"] = items.filter { ($0["done"] as? Bool) != true }
|
||||
object[key] = list
|
||||
}
|
||||
data = (try? JSONSerialization.data(withJSONObject: object)) ?? data
|
||||
}
|
||||
return data
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import { Notifications } from '../../db/entities/Notifications.entity';
|
||||
import { PushSubscriptions } from '../../db/entities/PushSubscriptions.entity';
|
||||
import { Settings } from '../../db/entities/Settings.entity';
|
||||
import { TodoItems } from '../../db/entities/TodoItems.entity';
|
||||
import { TripMembers } from '../../db/entities/TripMembers.entity';
|
||||
import { Trips } from '../../db/entities/Trips.entity';
|
||||
import { AddonsModule } from '../addons/addons.module';
|
||||
import { SchedulingModule } from '../scheduling/scheduling.module';
|
||||
@@ -16,6 +17,7 @@ import { NotificationsService } from './notifications.service';
|
||||
import { PushSubscriptionsService } from './push/push-subscriptions.service';
|
||||
import { PushController } from './push/push.controller';
|
||||
import { VapidKeysService } from './push/vapid-keys.service';
|
||||
import { WatchSyncPushService } from './push/watch-sync-push.service';
|
||||
import { ReminderJobsService } from './reminder-jobs.service';
|
||||
import { StorageHealthNotifierService } from './storage-health-notifier.service';
|
||||
import { NtfyService } from './transports/ntfy.service';
|
||||
@@ -62,6 +64,7 @@ import { Module } from '@nestjs/common';
|
||||
Trips,
|
||||
TodoItems,
|
||||
PushSubscriptions,
|
||||
TripMembers,
|
||||
]),
|
||||
|
||||
AddonsModule,
|
||||
@@ -78,6 +81,7 @@ import { Module } from '@nestjs/common';
|
||||
WebPushService,
|
||||
VapidKeysService,
|
||||
PushSubscriptionsService,
|
||||
WatchSyncPushService,
|
||||
NotificationsMcp,
|
||||
ReminderJobsService,
|
||||
StorageHealthNotifierService,
|
||||
|
||||
@@ -0,0 +1,97 @@
|
||||
import { TripMembers } from '../../../db/entities/TripMembers.entity';
|
||||
import type { TripMembersRepository } from '../../../db/repositories/TripMembers.repository';
|
||||
import { logError } from '../../audit/audit-log.logger';
|
||||
import { withRequestContext } from '../../database/request-context';
|
||||
import { RealtimeService } from '../../realtime/realtime.service';
|
||||
import { WebPushService } from '../transports/web-push.service';
|
||||
import { MikroORM } from '@mikro-orm/core';
|
||||
import { InjectRepository } from '@mikro-orm/nestjs';
|
||||
import { Injectable, type OnModuleDestroy, type OnModuleInit } from '@nestjs/common';
|
||||
|
||||
/**
|
||||
* How long the changes of one trip are gathered into one sync. Ticking off a
|
||||
* list goes in runs, and Apple throttles background pushes to a device that
|
||||
* gets too many, so a run of ticks becomes one push a couple of seconds later.
|
||||
*/
|
||||
export const WATCH_SYNC_SETTLE_MS = 2000;
|
||||
|
||||
/** The lists the watch shows: the packing lists (items and bags) and the to-dos. */
|
||||
const LIST_EVENT = /^(packing|todo):/;
|
||||
|
||||
interface PendingSync {
|
||||
timer: ReturnType<typeof setTimeout>;
|
||||
/** A change everyone on the trip sees; otherwise only `users` (a private packing item). */
|
||||
everyone: boolean;
|
||||
users: Set<number>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Keeps the Apple Watches of a trip's members current while their iPhone app is
|
||||
* in the background. The watch only gets what its phone sends it, and a phone
|
||||
* app in the background hears nothing over the WebSocket, so a tick by someone
|
||||
* else used to reach a watch only once its phone app was next opened.
|
||||
*
|
||||
* Every trip broadcast of a list event (whoever made it, over REST, MCP or a
|
||||
* plugin's RPC) is noted here; a couple of seconds later each member with the
|
||||
* iPhone app gets a silent push (WebPushService.sendWatchSync), which wakes the
|
||||
* app to read the lists and pass them to its watch. The push carries no content
|
||||
* and no trip: the app reads what it needs from this server with its own
|
||||
* session. A private packing item syncs only its owner.
|
||||
*/
|
||||
@Injectable()
|
||||
export class WatchSyncPushService implements OnModuleInit, OnModuleDestroy {
|
||||
private readonly pending = new Map<string, PendingSync>();
|
||||
private stopListening: (() => void) | null = null;
|
||||
|
||||
constructor(
|
||||
private readonly realtime: RealtimeService,
|
||||
private readonly push: WebPushService,
|
||||
@InjectRepository(TripMembers) private readonly members: TripMembersRepository,
|
||||
private readonly orm: MikroORM,
|
||||
) {}
|
||||
|
||||
onModuleInit(): void {
|
||||
this.stopListening = this.realtime.onTripBroadcast((tripId, eventType, onlyUserId) =>
|
||||
this.note(tripId, eventType, onlyUserId),
|
||||
);
|
||||
}
|
||||
|
||||
onModuleDestroy(): void {
|
||||
this.stopListening?.();
|
||||
this.stopListening = null;
|
||||
for (const entry of this.pending.values()) clearTimeout(entry.timer);
|
||||
this.pending.clear();
|
||||
}
|
||||
|
||||
/** A broadcast went out: a list event joins its trip's next sync. */
|
||||
note(tripId: number | string, eventType: string, onlyUserId?: number): void {
|
||||
if (!LIST_EVENT.test(eventType)) return;
|
||||
const key = String(tripId);
|
||||
let entry = this.pending.get(key);
|
||||
if (!entry) {
|
||||
const timer = setTimeout(() => void this.flush(key), WATCH_SYNC_SETTLE_MS);
|
||||
// A sync still waiting never holds the process open on shutdown.
|
||||
timer.unref?.();
|
||||
entry = { timer, everyone: false, users: new Set() };
|
||||
this.pending.set(key, entry);
|
||||
}
|
||||
if (onlyUserId == null) entry.everyone = true;
|
||||
else entry.users.add(onlyUserId);
|
||||
}
|
||||
|
||||
/** Sends one trip's gathered sync. */
|
||||
async flush(key: string): Promise<void> {
|
||||
const entry = this.pending.get(key);
|
||||
if (!entry) return;
|
||||
this.pending.delete(key);
|
||||
clearTimeout(entry.timer);
|
||||
try {
|
||||
const users = entry.everyone
|
||||
? await withRequestContext(this.orm, () => this.members.rosterUserIds(key))
|
||||
: entry.users;
|
||||
await Promise.all([...users].map((userId) => this.push.sendWatchSync(userId)));
|
||||
} catch (err) {
|
||||
logError(`Watch sync failed trip=${key}: ${err instanceof Error ? err.message : err}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -48,7 +48,37 @@ export const FORBIDDEN_FAILURES_BEFORE_REMOVAL = 3;
|
||||
*/
|
||||
const COLLAPSING_EVENT = 'collab_message';
|
||||
|
||||
type Urgency = 'normal' | 'high';
|
||||
type Urgency = 'low' | 'normal' | 'high';
|
||||
|
||||
/**
|
||||
* The watch sync (WatchSyncPushService): a silent push to the TREK iPhone app
|
||||
* that a packing list or the to-dos of one of its trips changed, so the app can
|
||||
* refresh its Apple Watch while it is in the background. The relay sends it as
|
||||
* an APNs background push (content-available, no alert) when it carries this
|
||||
* header; the record itself says nothing beyond that.
|
||||
*/
|
||||
export const BACKGROUND_PUSH_HEADER = 'X-Trek-Push';
|
||||
const WATCH_SYNC_PAYLOAD = Buffer.from(JSON.stringify({ type: 'watch-sync' }), 'utf8');
|
||||
/** A refresh that arrives ten minutes late is stale; the next change sends another. */
|
||||
const WATCH_SYNC_TTL_SECONDS = 10 * 60;
|
||||
/** Later syncs replace a queued one at Apple: only the newest state matters. */
|
||||
const WATCH_SYNC_TOPIC = 'trek-watch-sync';
|
||||
|
||||
/** The relay's APNs endpoints, the only devices with a watch to refresh. */
|
||||
export function isAppleRelayEndpoint(endpoint: string): boolean {
|
||||
try {
|
||||
const url = new URL(endpoint);
|
||||
return url.protocol === 'https:' && url.hostname === 'push.liketrek.com' && url.pathname.startsWith('/v1/apns/');
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
interface DeliveryOptions {
|
||||
ttlSeconds?: number;
|
||||
/** Sent as a background push by the relay, never shown. */
|
||||
background?: boolean;
|
||||
}
|
||||
/** The events that are about a date, where arriving late defeats the point (RFC 8030 section 5.3). */
|
||||
const HIGH_URGENCY_EVENTS: ReadonlySet<string> = new Set(['trip_reminder', 'todo_due']);
|
||||
|
||||
@@ -169,6 +199,29 @@ export class WebPushService {
|
||||
return this.deliver(userId, buildPushPayload(msg), urgency, msg.event);
|
||||
}
|
||||
|
||||
/**
|
||||
* The watch sync to each of the user's iPhones (the relay's APNs endpoints):
|
||||
* true when at least one took it. Browsers and Android devices have no watch
|
||||
* and get nothing. Nothing is sent while the server has no usable key pair.
|
||||
*/
|
||||
async sendWatchSync(userId: number): Promise<boolean> {
|
||||
try {
|
||||
if (!(await this.isAvailable())) return false;
|
||||
const rows = (await this.subscriptions.listForUser(userId)).filter((row) => isAppleRelayEndpoint(row.endpoint));
|
||||
if (rows.length === 0) return false;
|
||||
const keys = await this.keys.getKeys();
|
||||
const sign = createVapidSigner(keys, this.keys.getSubject());
|
||||
const options: DeliveryOptions = { ttlSeconds: WATCH_SYNC_TTL_SECONDS, background: true };
|
||||
const results = await Promise.all(
|
||||
rows.map((row) => this.deliverTo(row, WATCH_SYNC_PAYLOAD, keys.publicKey, sign, 'low', 'watch_sync', options)),
|
||||
);
|
||||
return results.includes(true);
|
||||
} catch (err) {
|
||||
logError(`Web Push failed event=watch_sync user=${userId}: ${err instanceof Error ? err.message : err}`);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/** The Send test button: a short fixed message to every browser the user turned push on in. */
|
||||
async sendTest(userId: number): Promise<ChannelTestResult> {
|
||||
if (!(await this.isAvailable())) return { success: false, error: PUSH_UNAVAILABLE_ERROR };
|
||||
@@ -200,6 +253,7 @@ export class WebPushService {
|
||||
sign: VapidSigner,
|
||||
urgency: Urgency,
|
||||
event: string,
|
||||
options: DeliveryOptions = {},
|
||||
): Promise<boolean> {
|
||||
const host = hostOf(row.endpoint);
|
||||
if (row.vapid_public_key !== publicKey) {
|
||||
@@ -230,11 +284,12 @@ export class WebPushService {
|
||||
{
|
||||
method: 'POST',
|
||||
headers: {
|
||||
TTL: String(PUSH_TTL_SECONDS),
|
||||
TTL: String(options.ttlSeconds ?? PUSH_TTL_SECONDS),
|
||||
Urgency: urgency,
|
||||
'Content-Encoding': 'aes128gcm',
|
||||
'Content-Type': 'application/octet-stream',
|
||||
Authorization: sign(row.endpoint),
|
||||
...(options.background ? { [BACKGROUND_PUSH_HEADER]: 'background', Topic: WATCH_SYNC_TOPIC } : {}),
|
||||
},
|
||||
body: new Uint8Array(body),
|
||||
signal: AbortSignal.timeout(PUSH_TIMEOUT_MS),
|
||||
|
||||
@@ -17,8 +17,26 @@ import type { TrekWsPayload, TrekWsPluginEventName, TrekWsTripEventName, TrekWsU
|
||||
* The service is deliberately dependency-free so out-of-container code (the
|
||||
* no-Nest test harnesses) can construct it with a bare `new RealtimeService()`.
|
||||
*/
|
||||
/**
|
||||
* Told of every trip broadcast once it went out: the trip, the event and, for a
|
||||
* private item, the one user it went to. The watch sync (WatchSyncPushService)
|
||||
* listens here, so every writer that broadcasts (REST, MCP, plugin RPC) reaches
|
||||
* it without knowing of it.
|
||||
*/
|
||||
export type TripBroadcastListener = (tripId: number | string, eventType: string, onlyUserId?: number) => void;
|
||||
|
||||
@Injectable()
|
||||
export class RealtimeService {
|
||||
private readonly listeners = new Set<TripBroadcastListener>();
|
||||
|
||||
/** Registers a listener for trip broadcasts; returns the way to remove it. */
|
||||
onTripBroadcast(listener: TripBroadcastListener): () => void {
|
||||
this.listeners.add(listener);
|
||||
return () => {
|
||||
this.listeners.delete(listener);
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Broadcast an event to all sockets in a trip room. Mirrors the
|
||||
* ws-state.ts signature exactly: `excludeSid` is the X-Socket-Id
|
||||
@@ -59,6 +77,15 @@ export class RealtimeService {
|
||||
]
|
||||
): void {
|
||||
broadcast(...args);
|
||||
const [tripId, eventType, , , onlyUserId] = args;
|
||||
for (const listener of this.listeners) {
|
||||
// A listener is a side matter: it never costs the broadcast or the next listener.
|
||||
try {
|
||||
listener(tripId, eventType, onlyUserId);
|
||||
} catch {
|
||||
/* the listener logs its own failures */
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -0,0 +1,234 @@
|
||||
/**
|
||||
* The watch sync (WSYNC-*): list changes on a trip become one silent push per
|
||||
* member a moment later (WatchSyncPushService), sent to the iPhone app only and
|
||||
* marked for the relay as a background push (WebPushService.sendWatchSync), and
|
||||
* every trip broadcast reaches its listeners (RealtimeService.onTripBroadcast).
|
||||
*/
|
||||
import type { PushSubscriptionRow } from '../../../../src/db/repositories/PushSubscriptions.repository';
|
||||
import type { TripMembersRepository } from '../../../../src/db/repositories/TripMembers.repository';
|
||||
import type { PushSubscriptionsService } from '../../../../src/nest/notifications/push/push-subscriptions.service';
|
||||
import type { VapidKeysService } from '../../../../src/nest/notifications/push/vapid-keys.service';
|
||||
import {
|
||||
WATCH_SYNC_SETTLE_MS,
|
||||
WatchSyncPushService,
|
||||
} from '../../../../src/nest/notifications/push/watch-sync-push.service';
|
||||
import { generateVapidKeyPair } from '../../../../src/nest/notifications/push/web-push-crypto';
|
||||
import {
|
||||
BACKGROUND_PUSH_HEADER,
|
||||
WebPushService,
|
||||
isAppleRelayEndpoint,
|
||||
} from '../../../../src/nest/notifications/transports/web-push.service';
|
||||
import { RealtimeService } from '../../../../src/nest/realtime/realtime.service';
|
||||
import type { MikroORM } from '@mikro-orm/core';
|
||||
|
||||
import { createECDH, randomBytes } from 'node:crypto';
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
|
||||
const { logError, safeFetchFollow } = vi.hoisted(() => ({ logError: vi.fn(), safeFetchFollow: vi.fn() }));
|
||||
vi.mock('../../../../src/nest/audit/audit-log.logger', () => ({
|
||||
LOG_LEVEL: 'error',
|
||||
logInfo: vi.fn(),
|
||||
logDebug: vi.fn(),
|
||||
logError,
|
||||
logWarn: vi.fn(),
|
||||
}));
|
||||
vi.mock('../../../../src/utils/ssrfGuard', () => ({
|
||||
SsrfBlockedError: class SsrfBlockedError extends Error {},
|
||||
safeFetchFollow,
|
||||
}));
|
||||
// The repository's read runs inside a request context; the fake repository needs none.
|
||||
vi.mock('../../../../src/nest/database/request-context', () => ({
|
||||
withRequestContext: (_orm: unknown, fn: () => unknown) => fn(),
|
||||
}));
|
||||
// The socket registry has no rooms here; only the listeners matter.
|
||||
vi.mock('../../../../src/nest/realtime/ws-state', () => ({
|
||||
broadcast: vi.fn(),
|
||||
broadcastToUser: vi.fn(),
|
||||
getOnlineUserIds: () => new Set<number>(),
|
||||
}));
|
||||
|
||||
const APNS = 'https://push.liketrek.com/v1/apns/production/' + 'ab'.repeat(32);
|
||||
const FCM = 'https://push.liketrek.com/v1/fcm/' + 'x'.repeat(140);
|
||||
const BROWSER = 'https://fcm.googleapis.com/fcm/send/abc';
|
||||
|
||||
describe('WatchSyncPushService', () => {
|
||||
let realtime: RealtimeService;
|
||||
let sendWatchSync: ReturnType<typeof vi.fn>;
|
||||
let rosterUserIds: ReturnType<typeof vi.fn>;
|
||||
let service: WatchSyncPushService;
|
||||
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
logError.mockClear();
|
||||
realtime = new RealtimeService();
|
||||
sendWatchSync = vi.fn().mockResolvedValue(true);
|
||||
rosterUserIds = vi.fn().mockResolvedValue(new Set([1, 2, 3]));
|
||||
service = new WatchSyncPushService(
|
||||
realtime,
|
||||
{ sendWatchSync } as unknown as WebPushService,
|
||||
{ rosterUserIds } as unknown as TripMembersRepository,
|
||||
{} as MikroORM,
|
||||
);
|
||||
service.onModuleInit();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
service.onModuleDestroy();
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it('WSYNC-001: a run of list changes on a trip becomes one sync to every member, once it settled', async () => {
|
||||
realtime.broadcast(12, 'packing:updated', { item: {} } as never);
|
||||
realtime.broadcast(12, 'todo:updated', { item: {} } as never);
|
||||
realtime.broadcast(12, 'packing:deleted', { itemId: 4 } as never);
|
||||
expect(sendWatchSync).not.toHaveBeenCalled();
|
||||
|
||||
await vi.advanceTimersByTimeAsync(WATCH_SYNC_SETTLE_MS);
|
||||
expect(rosterUserIds).toHaveBeenCalledTimes(1);
|
||||
expect(rosterUserIds).toHaveBeenCalledWith('12');
|
||||
expect(sendWatchSync.mock.calls.map(([id]) => id).sort()).toEqual([1, 2, 3]);
|
||||
});
|
||||
|
||||
it('WSYNC-002: what the watch does not show sends nothing', async () => {
|
||||
realtime.broadcast(12, 'place:updated' as never, {} as never);
|
||||
realtime.broadcast(12, 'collab:message:created' as never, {} as never);
|
||||
await vi.advanceTimersByTimeAsync(WATCH_SYNC_SETTLE_MS);
|
||||
expect(sendWatchSync).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('WSYNC-003: a private packing item syncs its owner alone, unless the trip changed for everyone too', async () => {
|
||||
realtime.broadcast(12, 'packing:updated', { item: {} } as never, undefined, 2);
|
||||
await vi.advanceTimersByTimeAsync(WATCH_SYNC_SETTLE_MS);
|
||||
expect(rosterUserIds).not.toHaveBeenCalled();
|
||||
expect(sendWatchSync.mock.calls).toEqual([[2]]);
|
||||
|
||||
sendWatchSync.mockClear();
|
||||
realtime.broadcast(12, 'packing:updated', { item: {} } as never, undefined, 2);
|
||||
realtime.broadcast(12, 'todo:created', { item: {} } as never);
|
||||
await vi.advanceTimersByTimeAsync(WATCH_SYNC_SETTLE_MS);
|
||||
expect(sendWatchSync.mock.calls.map(([id]) => id).sort()).toEqual([1, 2, 3]);
|
||||
});
|
||||
|
||||
it('WSYNC-004: each trip gathers on its own', async () => {
|
||||
realtime.broadcast(12, 'todo:updated', { item: {} } as never);
|
||||
realtime.broadcast(13, 'todo:updated', { item: {} } as never);
|
||||
await vi.advanceTimersByTimeAsync(WATCH_SYNC_SETTLE_MS);
|
||||
expect(rosterUserIds.mock.calls).toEqual([['12'], ['13']]);
|
||||
});
|
||||
|
||||
it('WSYNC-005: a failed read is logged and never thrown into the broadcast', async () => {
|
||||
rosterUserIds.mockRejectedValue(new Error('db gone'));
|
||||
realtime.broadcast(12, 'todo:updated', { item: {} } as never);
|
||||
await vi.advanceTimersByTimeAsync(WATCH_SYNC_SETTLE_MS);
|
||||
expect(logError).toHaveBeenCalledWith(expect.stringContaining('Watch sync failed trip=12: db gone'));
|
||||
expect(sendWatchSync).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('WSYNC-006: shut down, it stops listening and drops what was waiting', async () => {
|
||||
realtime.broadcast(12, 'todo:updated', { item: {} } as never);
|
||||
service.onModuleDestroy();
|
||||
realtime.broadcast(13, 'todo:updated', { item: {} } as never);
|
||||
await vi.advanceTimersByTimeAsync(WATCH_SYNC_SETTLE_MS * 2);
|
||||
expect(sendWatchSync).not.toHaveBeenCalled();
|
||||
await service.flush('12');
|
||||
expect(sendWatchSync).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe('RealtimeService.onTripBroadcast', () => {
|
||||
it('WSYNC-007: listeners hear every trip broadcast, one failing listener costs nothing, and can be removed', () => {
|
||||
const realtime = new RealtimeService();
|
||||
const heard = vi.fn();
|
||||
realtime.onTripBroadcast(() => {
|
||||
throw new Error('broken listener');
|
||||
});
|
||||
const stop = realtime.onTripBroadcast(heard);
|
||||
expect(() => realtime.broadcast(7, 'packing:updated', { item: {} } as never, 'sock', 3)).not.toThrow();
|
||||
expect(heard).toHaveBeenCalledWith(7, 'packing:updated', 3);
|
||||
stop();
|
||||
realtime.broadcast(7, 'packing:updated', { item: {} } as never);
|
||||
expect(heard).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('WebPushService.sendWatchSync', () => {
|
||||
const keys = generateVapidKeyPair();
|
||||
|
||||
function row(id: number, endpoint: string): PushSubscriptionRow {
|
||||
const device = createECDH('prime256v1');
|
||||
device.generateKeys();
|
||||
return {
|
||||
id,
|
||||
user_id: 5,
|
||||
endpoint,
|
||||
p256dh: device.getPublicKey().toString('base64url'),
|
||||
auth: randomBytes(16).toString('base64url'),
|
||||
vapid_public_key: keys.publicKey,
|
||||
user_agent: null,
|
||||
created_at: '2026-11-08T00:00:00Z',
|
||||
last_success_at: null,
|
||||
failure_count: 0,
|
||||
};
|
||||
}
|
||||
|
||||
function service(rows: PushSubscriptionRow[], available = true) {
|
||||
const vapid = {
|
||||
isAvailable: vi.fn().mockResolvedValue(available),
|
||||
getKeys: vi.fn().mockResolvedValue({ ...keys, source: 'database' }),
|
||||
getSubject: () => 'mailto:admin@trek.example.test',
|
||||
} as unknown as VapidKeysService;
|
||||
const subscriptions = {
|
||||
listForUser: vi.fn().mockResolvedValue(rows),
|
||||
recordSuccess: vi.fn(),
|
||||
recordFailure: vi.fn().mockResolvedValue(1),
|
||||
deleteById: vi.fn(),
|
||||
} as unknown as PushSubscriptionsService;
|
||||
return new WebPushService(vapid, subscriptions);
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
safeFetchFollow.mockReset().mockResolvedValue(new Response(null, { status: 201 }));
|
||||
});
|
||||
|
||||
it('WSYNC-008: only the iPhone app gets it, as a short-lived background push the relay recognises', async () => {
|
||||
const sent = await service([row(1, APNS), row(2, FCM), row(3, BROWSER)]).sendWatchSync(5);
|
||||
expect(sent).toBe(true);
|
||||
expect(safeFetchFollow).toHaveBeenCalledTimes(1);
|
||||
const [endpoint, init] = safeFetchFollow.mock.calls[0];
|
||||
expect(endpoint).toBe(APNS);
|
||||
expect(init.headers).toMatchObject({
|
||||
[BACKGROUND_PUSH_HEADER]: 'background',
|
||||
Topic: 'trek-watch-sync',
|
||||
Urgency: 'low',
|
||||
TTL: '600',
|
||||
'Content-Encoding': 'aes128gcm',
|
||||
});
|
||||
});
|
||||
|
||||
it('WSYNC-009: a user without the iPhone app, or a server without keys, sends nothing', async () => {
|
||||
expect(await service([row(2, FCM), row(3, BROWSER)]).sendWatchSync(5)).toBe(false);
|
||||
expect(await service([row(1, APNS)], false).sendWatchSync(5)).toBe(false);
|
||||
expect(safeFetchFollow).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('WSYNC-010: an ordinary notification still goes out without the background mark', async () => {
|
||||
await service([row(1, APNS)]).sendToUser(5, {
|
||||
event: 'collab_message',
|
||||
title: 'Hi',
|
||||
body: 'there',
|
||||
navigateTarget: '/trips/1',
|
||||
url: 'https://trek.example.test/trips/1',
|
||||
});
|
||||
const [, init] = safeFetchFollow.mock.calls[0];
|
||||
expect(init.headers[BACKGROUND_PUSH_HEADER]).toBeUndefined();
|
||||
expect(init.headers.TTL).toBe(String(24 * 60 * 60));
|
||||
});
|
||||
|
||||
it("WSYNC-011: only the relay's APNs endpoints count as an iPhone", () => {
|
||||
expect(isAppleRelayEndpoint(APNS)).toBe(true);
|
||||
expect(isAppleRelayEndpoint(FCM)).toBe(false);
|
||||
expect(isAppleRelayEndpoint(BROWSER)).toBe(false);
|
||||
expect(isAppleRelayEndpoint('http://push.liketrek.com/v1/apns/production/ab')).toBe(false);
|
||||
expect(isAppleRelayEndpoint('not a url')).toBe(false);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user