From f089f030495c3d9483329c2dff64cb7b32a47b90 Mon Sep 17 00:00:00 2001 From: Juan Castro Date: Fri, 23 Oct 2026 15:18:15 -0400 Subject: [PATCH] fix(push): deliver only to devices with an active session Delivery skips devices whose session is revoked, expired or gone; devices with no session still receive. A token that changes owner gets a new uid and created_at, and a register that keeps losing the token to another user ends in a 409. touch and recordFailure match the owner as well as the uid, listByUser reads from the primary, the stale sweep stops between batches once shutdown starts, and the user agent and label limits are exported. --- src/backend/services/push/PushService.test.ts | 12 ++ src/backend/services/push/PushService.ts | 3 + .../stores/push/PushDeviceStore.test.ts | 147 ++++++++++++++++-- src/backend/stores/push/PushDeviceStore.ts | 136 +++++++++++----- 4 files changed, 247 insertions(+), 51 deletions(-) diff --git a/src/backend/services/push/PushService.test.ts b/src/backend/services/push/PushService.test.ts index fff9a4b8e..a270c9a74 100644 --- a/src/backend/services/push/PushService.test.ts +++ b/src/backend/services/push/PushService.test.ts @@ -146,4 +146,16 @@ describe('PushService device cleanup', () => { expect(await devices().getByUid(stale.uid)).toBeNull(); expect(await devices().getByUid(recent.uid)).not.toBeNull(); }); + + // Last: it starts the service's shutdown. + it('stops sweeping once shutdown has started', async () => { + const user = await makeUser(); + const stale = await register(user.id, 'sess-shutdown'); + await setLastSeen(stale.uid, 1000); + + server.services.push.onServerPrepareShutdown(); + + expect(await server.services.push.sweepStaleDevices()).toBe(0); + expect(await devices().getByUid(stale.uid)).not.toBeNull(); + }); }); diff --git a/src/backend/services/push/PushService.ts b/src/backend/services/push/PushService.ts index 82ef9baa4..32578ea1c 100644 --- a/src/backend/services/push/PushService.ts +++ b/src/backend/services/push/PushService.ts @@ -29,6 +29,7 @@ const STALE_MAX_BATCHES = 50; /** Keeps push devices in step with their sessions; runs even with push off. */ export class PushService extends PuterService { #staleSweep: ReturnType | null = null; + #stopping = false; override onServerStart(): void { this.#armStaleSweep(); @@ -56,6 +57,7 @@ export class PushService extends PuterService { } override onServerPrepareShutdown(): void { + this.#stopping = true; if (this.#staleSweep) clearInterval(this.#staleSweep); this.#staleSweep = null; } @@ -68,6 +70,7 @@ export class PushService extends PuterService { let removed = 0; for (let pass = 0; pass < STALE_MAX_BATCHES; pass++) { + if (this.#stopping) break; const batch = await this.stores.pushDevice.removeUnseenSince( cutoff, { limit: STALE_BATCH_SIZE }, diff --git a/src/backend/stores/push/PushDeviceStore.test.ts b/src/backend/stores/push/PushDeviceStore.test.ts index c474a6553..b4a666b83 100644 --- a/src/backend/stores/push/PushDeviceStore.test.ts +++ b/src/backend/stores/push/PushDeviceStore.test.ts @@ -19,11 +19,13 @@ // Real database: PUTER_TEST_DB_ENGINE=postgres covers the postgres migration. -import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; import { PuterServer } from '../../server.js'; import { createTestUser, setupTestServer } from '../../testUtil.js'; import { hashPushToken, + PUSH_DEVICE_LABEL_MAX_LENGTH, + PUSH_DEVICE_USER_AGENT_MAX_LENGTH, type PushDeviceRegistration, } from './PushDeviceStore.js'; @@ -40,11 +42,21 @@ describe('PushDeviceStore', () => { const store = () => server.stores.pushDevice; + const sessionOf = new Map(); + + const newSession = async (userId: number): Promise => { + const row = (await server.stores.session.create(userId)) as { + uuid: string; + }; + return row.uuid; + }; + const makeUser = async () => { const username = `pd${Math.random().toString(36).slice(2, 9)}`; await createTestUser(server, { username, password: 'pw-test-1234' }); const user = await server.stores.user.getByUsername(username); if (!user) throw new Error('test user missing'); + sessionOf.set(user.id, await newSession(user.id)); return user; }; @@ -57,7 +69,7 @@ describe('PushDeviceStore', () => { ) => store().register({ userId, - sessionUid: 'session-a', + sessionUid: sessionOf.get(userId) ?? 'no-session', platform: 'web', token: freshToken(), ...overrides, @@ -89,7 +101,7 @@ describe('PushDeviceStore', () => { expect(device).toMatchObject({ userId: user.id, - sessionUid: 'session-a', + sessionUid: sessionOf.get(user.id), platform: 'web', appUid: null, userAgent: 'Firefox', @@ -131,10 +143,17 @@ describe('PushDeviceStore', () => { token, appUid: 'app-one', }); + await server.clients.db.write( + 'UPDATE `push_device` SET `created_at` = 1000 WHERE `uid` = ?', + [original.uid], + ); const moved = await register(next.id, { token, appUid: null }); - expect(moved.uid).toBe(original.uid); + expect(moved.uid).not.toBe(original.uid); + expect(moved.createdAt).toBeGreaterThan(1000); + expect(moved.sessionUid).toBe(sessionOf.get(next.id)); + expect(await store().getByUid(original.uid)).toBeNull(); expect(moved.userId).toBe(next.id); expect(moved.appUid).toBeNull(); expect(await store().listByUser(previous.id)).toEqual([]); @@ -143,7 +162,8 @@ describe('PushDeviceStore', () => { ).toEqual([]); expect( (await store().listByUser(next.id)).map((d) => d.uid), - ).toEqual([original.uid]); + ).toEqual([moved.uid]); + expect(await countByToken(token)).toBe(1); }); it('keeps one row when the same token registers twice at once', async () => { @@ -163,8 +183,61 @@ describe('PushDeviceStore', () => { userAgent: 'u'.repeat(2000), label: 'l'.repeat(200), }); - expect(device.userAgent).toHaveLength(512); - expect(device.label).toHaveLength(64); + expect(device.userAgent).toHaveLength( + PUSH_DEVICE_USER_AGENT_MAX_LENGTH, + ); + expect(device.label).toHaveLength(PUSH_DEVICE_LABEL_MAX_LENGTH); + }); + + // `otherId` takes `token` before each of the next `times` read-backs. + const interleave = (otherId: number, token: string, times: number) => { + const db = server.clients.db; + const pread = db.pread.bind(db); + let remaining = times; + let busy = false; + return vi.spyOn(db, 'pread').mockImplementation(async (q, p) => { + if (!busy && remaining > 0 && q.includes('`token_hash` = ?')) { + remaining--; + busy = true; + try { + await register(otherId, { token }); + } finally { + busy = false; + } + } + return pread(q, p); + }); + }; + + it('retries once when another user takes the token mid-register', async () => { + const user = await makeUser(); + const other = await makeUser(); + const token = freshToken(); + const spy = interleave(other.id, token, 1); + try { + const device = await register(user.id, { token }); + expect(device.userId).toBe(user.id); + } finally { + spy.mockRestore(); + } + expect(await store().listByUser(other.id)).toEqual([]); + expect(await countByToken(token)).toBe(1); + }); + + it('throws a conflict when another user keeps taking the token', async () => { + const user = await makeUser(); + const other = await makeUser(); + const token = freshToken(); + const spy = interleave(other.id, token, 2); + try { + await expect( + register(user.id, { token }), + ).rejects.toMatchObject({ statusCode: 409 }); + } finally { + spy.mockRestore(); + } + const [owned] = await store().listByUser(other.id); + expect(owned?.userId).toBe(other.id); }); }); @@ -205,6 +278,45 @@ describe('PushDeviceStore', () => { ); }); + it('skips devices whose session was revoked or has expired', async () => { + const user = await makeUser(); + const active = await register(user.id); + const revokedSession = await newSession(user.id); + const expiredSession = await newSession(user.id); + const revoked = await register(user.id, { + sessionUid: revokedSession, + }); + const expired = await register(user.id, { + sessionUid: expiredSession, + }); + const unknown = await register(user.id, { + sessionUid: 'no-such-session', + }); + const sessionless = await register(user.id); + await server.clients.db.write( + 'UPDATE `push_device` SET `session_uid` = NULL WHERE `uid` = ?', + [sessionless.uid], + ); + await server.stores.session.revokeCascade(revokedSession); + await server.clients.db.write( + 'UPDATE `sessions` SET `expires_at` = ? WHERE `uuid` = ?', + [Math.floor(Date.now() / 1000) - 60, expiredSession], + ); + + const uids = (await store().listForDelivery(user.id, null)).map( + (t) => t.uid, + ); + expect(uids.sort()).toEqual([active.uid, sessionless.uid].sort()); + expect(uids).not.toContain(revoked.uid); + expect(uids).not.toContain(expired.uid); + expect(uids).not.toContain(unknown.uid); + const revokedRows = await server.clients.db.read( + 'SELECT `revoked_at` FROM `sessions` WHERE `uuid` = ?', + [revokedSession], + ); + expect(revokedRows[0]?.revoked_at).not.toBeNull(); + }); + it('respects the limit, newest first', async () => { const user = await makeUser(); await register(user.id); @@ -269,14 +381,31 @@ describe('PushDeviceStore', () => { const device = await register(user.id); await setLastSeen(device.uid, 1000); - await store().recordFailure([device.uid, device.uid]); - await store().touch([device.uid]); + await store().recordFailure(user.id, [device.uid, device.uid]); + await store().touch(user.id, [device.uid]); const row = await store().getByUid(device.uid); expect(row?.failureCount).toBe(4); expect(row?.lastSeenAt).toBeGreaterThanOrEqual(device.lastSeenAt); }); + it('touches and counts failures only on the named owner’s rows', async () => { + const previous = await makeUser(); + const next = await makeUser(); + const token = freshToken(); + const old = await register(previous.id, { token }); + const moved = await register(next.id, { token }); + const earlier = moved.lastSeenAt - 100; + await setLastSeen(moved.uid, earlier); + + await store().recordFailure(previous.id, [old.uid, moved.uid]); + await store().touch(previous.id, [old.uid, moved.uid]); + + const row = await store().getByUid(moved.uid); + expect(row?.failureCount).toBe(3); + expect(row?.lastSeenAt).toBe(earlier); + }); + it('removes devices unseen since a cutoff, in batches', async () => { const user = await makeUser(); const old = await register(user.id); diff --git a/src/backend/stores/push/PushDeviceStore.ts b/src/backend/stores/push/PushDeviceStore.ts index 77dbcc007..c1f671795 100644 --- a/src/backend/stores/push/PushDeviceStore.ts +++ b/src/backend/stores/push/PushDeviceStore.ts @@ -19,6 +19,7 @@ import { createHash } from 'node:crypto'; import { v4 as uuidv4 } from 'uuid'; +import { HttpError } from '../../core/http/HttpError.js'; import { PuterStore } from '../types'; export type PushPlatform = 'web' | 'ios' | 'android'; @@ -53,13 +54,38 @@ export interface PushDeviceRegistration { label?: string | null; } -const USER_AGENT_MAX_LENGTH = 512; -const LABEL_MAX_LENGTH = 64; +export const PUSH_DEVICE_USER_AGENT_MAX_LENGTH = 512; +export const PUSH_DEVICE_LABEL_MAX_LENGTH = 64; const MAX_LIMIT = 1000; -const DEVICE_COLUMNS = - '`uid`, `user_id`, `session_uid`, `platform`, `app_uid`, `user_agent`, ' + - '`label`, `created_at`, `last_seen_at`, `failure_count`'; +const COLUMN_NAMES = [ + 'uid', + 'user_id', + 'session_uid', + 'platform', + 'app_uid', + 'user_agent', + 'label', + 'created_at', + 'last_seen_at', + 'failure_count', +]; +const DEVICE_COLUMNS = COLUMN_NAMES.map((c) => `\`${c}\``).join(', '); +const PREFIXED_DEVICE_COLUMNS = COLUMN_NAMES.map((c) => `d.\`${c}\``).join( + ', ', +); + +/** Columns a repeat registration overwrites, in bind order. */ +const REFRESHED_COLUMNS = [ + 'user_id', + 'session_uid', + 'platform', + 'app_uid', + 'user_agent', + 'label', + 'last_seen_at', + 'failure_count', +]; /** How a provider address is looked up; the address itself is never a key. */ export const hashPushToken = (token: string): string => @@ -121,7 +147,7 @@ export class PushDeviceStore extends PuterStore { ): Promise { const n = safeLimit(limit, 100); if (n === 0) return []; - const rows = await this.clients.db.read( + const rows = await this.clients.db.pread( `SELECT ${DEVICE_COLUMNS} FROM \`push_device\` WHERE \`user_id\` = ? ` + 'ORDER BY `id` DESC LIMIT ?', [userId, n], @@ -129,7 +155,7 @@ export class PushDeviceStore extends PuterStore { return rows.map(toDevice); } - /** Devices Puter registered, plus those `appUid` registered when given. */ + /** Puter's and `appUid`'s devices whose session, if any, is active. */ async listForDelivery( userId: number, appUid: string | null, @@ -137,12 +163,20 @@ export class PushDeviceStore extends PuterStore { ): Promise { const n = safeLimit(limit, 100); if (n === 0) return []; - const appClause = appUid ? ' OR `app_uid` = ?' : ''; + const appClause = appUid ? ' OR d.`app_uid` = ?' : ''; + const params: unknown[] = [userId]; + if (appUid) params.push(appUid); + params.push(nowSeconds(), n); const rows = await this.clients.db.read( - `SELECT ${DEVICE_COLUMNS}, \`token\` FROM \`push_device\` ` + - `WHERE \`user_id\` = ? AND (\`app_uid\` IS NULL${appClause}) ` + - 'ORDER BY `id` DESC LIMIT ?', - appUid ? [userId, appUid, n] : [userId, n], + `SELECT ${PREFIXED_DEVICE_COLUMNS}, d.\`token\` ` + + 'FROM `push_device` d LEFT JOIN `sessions` s ' + + 'ON s.`uuid` = d.`session_uid` AND s.`user_id` = d.`user_id` ' + + `WHERE d.\`user_id\` = ? AND (d.\`app_uid\` IS NULL${appClause}) ` + + 'AND (d.`session_uid` IS NULL OR (s.`uuid` IS NOT NULL ' + + 'AND s.`revoked_at` IS NULL ' + + 'AND (s.`expires_at` IS NULL OR s.`expires_at` > ?))) ' + + 'ORDER BY d.`id` DESC LIMIT ?', + params, ); return rows.map((row) => ({ ...toDevice(row), @@ -152,8 +186,18 @@ export class PushDeviceStore extends PuterStore { // -- Writes ------------------------------------------------------- - /** Insert, or move an existing token to this owner and reset failures. */ - async register({ + /** A token changing owner gets a new uid; 409 if it keeps changing. */ + async register(registration: PushDeviceRegistration): Promise { + for (let attempt = 0; attempt < 2; attempt++) { + const device = await this.#upsert(registration); + if (device.userId === registration.userId) return device; + } + throw new HttpError(409, 'This device is registered to another user', { + legacyCode: 'push_device_conflict', + }); + } + + async #upsert({ userId, sessionUid, platform, @@ -165,36 +209,44 @@ export class PushDeviceStore extends PuterStore { const db = this.clients.db; const tokenHash = hashPushToken(token); const now = nowSeconds(); - const updated = [ + const uid = uuidv4(); + const refreshed = [ userId, sessionUid, platform, appUid || null, - bounded(userAgent, USER_AGENT_MAX_LENGTH), - bounded(label, LABEL_MAX_LENGTH), + bounded(userAgent, PUSH_DEVICE_USER_AGENT_MAX_LENGTH), + bounded(label, PUSH_DEVICE_LABEL_MAX_LENGTH), now, 0, ]; - const upsert = db.upsertClause( - ['token_hash'], - [ - 'user_id', - 'session_uid', - 'platform', - 'app_uid', - 'user_agent', - 'label', - 'last_seen_at', - 'failure_count', - ], - ); + // Before `user_id`: mysql assigns left to right. + const sets = [ + '`uid` = CASE WHEN `push_device`.`user_id` = ? THEN `push_device`.`uid` ELSE ? END', + '`created_at` = CASE WHEN `push_device`.`user_id` = ? ' + + 'THEN `push_device`.`created_at` ELSE ? END', + ...REFRESHED_COLUMNS.map((c) => `\`${c}\` = ?`), + ].join(', '); + const conflict = db.case({ + mysql: `ON DUPLICATE KEY UPDATE ${sets}`, + otherwise: `ON CONFLICT(\`token_hash\`) DO UPDATE SET ${sets}`, + }); await db.write( 'INSERT INTO `push_device` (`uid`, `token`, `token_hash`, ' + - '`created_at`, `user_id`, `session_uid`, `platform`, ' + - '`app_uid`, `user_agent`, `label`, `last_seen_at`, ' + - '`failure_count`) ' + - `VALUES (?, ?, ?, ?, ${placeholders(updated)}) ${upsert}`, - [uuidv4(), token, tokenHash, now, ...updated, ...updated], + `\`created_at\`, ${REFRESHED_COLUMNS.map((c) => `\`${c}\``).join(', ')}) ` + + `VALUES (?, ?, ?, ?, ${placeholders(refreshed)}) ${conflict}`, + [ + uid, + token, + tokenHash, + now, + ...refreshed, + userId, + uid, + userId, + now, + ...refreshed, + ], ); const rows = await db.pread( `SELECT ${DEVICE_COLUMNS} FROM \`push_device\` WHERE \`token_hash\` = ? LIMIT 1`, @@ -238,23 +290,23 @@ export class PushDeviceStore extends PuterStore { return result.affectedRows; } - async touch(uids: string[]): Promise { + async touch(userId: number, uids: string[]): Promise { const list = unique(uids); if (list.length === 0) return; await this.clients.db.write( - 'UPDATE `push_device` SET `last_seen_at` = ? ' + - `WHERE \`uid\` IN (${placeholders(list)})`, - [nowSeconds(), ...list], + 'UPDATE `push_device` SET `last_seen_at` = ? WHERE `user_id` = ? ' + + `AND \`uid\` IN (${placeholders(list)})`, + [nowSeconds(), userId, ...list], ); } - async recordFailure(uids: string[]): Promise { + async recordFailure(userId: number, uids: string[]): Promise { const list = unique(uids); if (list.length === 0) return; await this.clients.db.write( 'UPDATE `push_device` SET `failure_count` = `failure_count` + 1 ' + - `WHERE \`uid\` IN (${placeholders(list)})`, - list, + `WHERE \`user_id\` = ? AND \`uid\` IN (${placeholders(list)})`, + [userId, ...list], ); }