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

@@ -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();
+    });
 });

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<typeof setInterval> | 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 },

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<number, string>();
+
+    const newSession = async (userId: number): Promise<string> => {
+        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);

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<PushDevice[]> {
         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<PushDeliveryTarget[]> {
         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<PushDevice> {
+        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<void> {
+    async touch(userId: number, uids: string[]): Promise<void> {
         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<void> {
+    async recordFailure(userId: number, uids: string[]): Promise<void> {
         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],
         );
     }