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], ); }