diff --git a/src/backend/clients/database/SqliteDatabaseClient.test.ts b/src/backend/clients/database/SqliteDatabaseClient.test.ts
index 1149ee751..a29c0d32b 100644
--- a/src/backend/clients/database/SqliteDatabaseClient.test.ts
+++ b/src/backend/clients/database/SqliteDatabaseClient.test.ts
@@ -27,7 +27,7 @@ import { DatabaseClientFactory } from './index.js';
import { SqliteDatabaseClient } from './SqliteDatabaseClient.js';
/** Highest schema version the migration table can reach. */
-const CURRENT_SCHEMA_VERSION = 95;
+const CURRENT_SCHEMA_VERSION = 96;
/**
* These suites migrate real files on disk. Idle they finish in well under a
@@ -724,6 +724,17 @@ describe('SqliteDatabaseClient — boot and migrations', { timeout: DISK_MIGRATI
]);
});
+ it('keeps notification.send_key unique', async () => {
+ const rows = await client.read('PRAGMA index_list(`notification`)');
+ expect(
+ rows.find((r) => r.name === 'idx_notification_send_key'),
+ ).toMatchObject({ unique: 1 });
+ const columns = await client.read(
+ 'PRAGMA index_info(`idx_notification_send_key`)',
+ );
+ expect(columns.map((c) => String(c.name))).toEqual(['send_key']);
+ });
+
it('leaves an already-migrated database untouched on a second boot', async () => {
const dir = mkdtempSync(join(tmpdir(), 'puter-sqlite-'));
const path = join(dir, 'nested', 'puter.sqlite');
diff --git a/src/backend/clients/database/SqliteDatabaseClient.ts b/src/backend/clients/database/SqliteDatabaseClient.ts
index b4b9f3f59..4a24f45b1 100644
--- a/src/backend/clients/database/SqliteDatabaseClient.ts
+++ b/src/backend/clients/database/SqliteDatabaseClient.ts
@@ -134,6 +134,7 @@ const AVAILABLE_MIGRATIONS: [number, string[]][] = [
[92, ['0097_user-to-user-holder-permission-index.sql']],
[93, ['0098_notification-groups.sql']],
[94, ['0099_kv-share-handle-notify.sql']],
+ [95, ['0100_notification-send-key.sql']],
];
export class SqliteDatabaseClient extends AbstractDatabaseClient {
diff --git a/src/backend/clients/database/migrations/mysql/mysql_mig_54.sql b/src/backend/clients/database/migrations/mysql/mysql_mig_54.sql
new file mode 100644
index 000000000..92e127be6
--- /dev/null
+++ b/src/backend/clients/database/migrations/mysql/mysql_mig_54.sql
@@ -0,0 +1,43 @@
+-- Copyright (C) 2024-present Puter Technologies Inc.
+--
+-- This file is part of Puter.
+--
+-- Puter is free software: you can redistribute it and/or modify
+-- it under the terms of the GNU Affero General Public License as published
+-- by the Free Software Foundation, either version 3 of the License, or
+-- (at your option) any later version.
+--
+-- This program is distributed in the hope that it will be useful,
+-- but WITHOUT ANY WARRANTY; without even the implied warranty of
+-- MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+-- GNU Affero General Public License for more details.
+--
+-- You should have received a copy of the GNU Affero General Public License
+-- along with this program. If not, see .
+
+-- `notification`.`send_key`. See sqlite/0100_notification-send-key.sql for the
+-- rationale. The key is a uuid, so ascii and compared byte for byte. No
+-- per-file applied-state tracking, so every step is guarded and a replay does
+-- nothing.
+
+CALL _puter_add_col('notification', 'send_key', '`send_key` char(36) CHARACTER SET ascii COLLATE ascii_bin DEFAULT NULL');
+
+DROP PROCEDURE IF EXISTS _puter_add_notification_send_key_index;
+DELIMITER //
+CREATE PROCEDURE _puter_add_notification_send_key_index()
+BEGIN
+ IF NOT EXISTS (
+ SELECT 1 FROM INFORMATION_SCHEMA.STATISTICS
+ WHERE TABLE_SCHEMA = DATABASE()
+ AND TABLE_NAME = 'notification'
+ AND INDEX_NAME = 'idx_notification_send_key'
+ ) THEN
+ ALTER TABLE `notification` ADD UNIQUE KEY `idx_notification_send_key`
+ (`send_key`);
+ END IF;
+END//
+DELIMITER ;
+
+CALL _puter_add_notification_send_key_index();
+
+DROP PROCEDURE IF EXISTS _puter_add_notification_send_key_index;
diff --git a/src/backend/clients/database/migrations/postgres/postgres_mig_43.sql b/src/backend/clients/database/migrations/postgres/postgres_mig_43.sql
new file mode 100644
index 000000000..02196c5f4
--- /dev/null
+++ b/src/backend/clients/database/migrations/postgres/postgres_mig_43.sql
@@ -0,0 +1,24 @@
+-- Copyright (C) 2024-present Puter Technologies Inc.
+--
+-- This file is part of Puter.
+--
+-- Puter is free software: you can redistribute it and/or modify
+-- it under the terms of the GNU Affero General Public License as published
+-- by the Free Software Foundation, either version 3 of the License, or
+-- (at your option) any later version.
+--
+-- This program is distributed in the hope that it will be useful,
+-- but WITHOUT ANY WARRANTY; without even the implied warranty of
+-- MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+-- GNU Affero General Public License for more details.
+--
+-- You should have received a copy of the GNU Affero General Public License
+-- along with this program. If not, see .
+
+-- notification.send_key. See sqlite/0100_notification-send-key.sql for the
+-- rationale. Idempotent via IF NOT EXISTS.
+
+ALTER TABLE notification ADD COLUMN IF NOT EXISTS send_key char(36) DEFAULT NULL;
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_notification_send_key
+ ON notification (send_key);
diff --git a/src/backend/clients/database/migrations/sqlite/0100_notification-send-key.sql b/src/backend/clients/database/migrations/sqlite/0100_notification-send-key.sql
new file mode 100644
index 000000000..c72494a9c
--- /dev/null
+++ b/src/backend/clients/database/migrations/sqlite/0100_notification-send-key.sql
@@ -0,0 +1,27 @@
+-- Copyright (C) 2024-present Puter Technologies Inc.
+--
+-- This file is part of Puter.
+--
+-- Puter is free software: you can redistribute it and/or modify
+-- it under the terms of the GNU Affero General Public License as published
+-- by the Free Software Foundation, either version 3 of the License, or
+-- (at your option) any later version.
+--
+-- This program is distributed in the hope that it will be useful,
+-- but WITHOUT ANY WARRANTY; without even the implied warranty of
+-- MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+-- GNU Affero General Public License for more details.
+--
+-- You should have received a copy of the GNU Affero General Public License
+-- along with this program. If not, see .
+
+-- `notification`.`send_key`: the key of the delayed send (a `deferred`
+-- schedule) that wrote this row or last replaced it, derived from the
+-- schedule and its slot. Unique, so one slot can never write two rows, and a
+-- send that crashed after writing finds its row again by it. Null on every row
+-- a delayed send never touched.
+
+ALTER TABLE `notification` ADD COLUMN `send_key` TEXT DEFAULT NULL;
+
+CREATE UNIQUE INDEX IF NOT EXISTS `idx_notification_send_key`
+ ON `notification` (`send_key`);
diff --git a/src/backend/drivers/notification/NotificationDriver.sendAt.test.ts b/src/backend/drivers/notification/NotificationDriver.sendAt.test.ts
new file mode 100644
index 000000000..797b3f26c
--- /dev/null
+++ b/src/backend/drivers/notification/NotificationDriver.sendAt.test.ts
@@ -0,0 +1,493 @@
+/*
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see .
+ */
+
+/**
+ * `notify({ sendAt })` without and with `services.schedules`, and the deferred
+ * mailbox write an extension runs inside its own transaction, fenced on a guard
+ * in a table of its own (a test table here).
+ */
+
+import { v4 as uuidv4 } from 'uuid';
+import {
+ afterAll,
+ afterEach,
+ beforeAll,
+ beforeEach,
+ describe,
+ expect,
+ it,
+ vi,
+} from 'vitest';
+import type { BatchEntry } from '../../clients/database/DatabaseClient.js';
+import { makeActor, type Actor } from '../../core/actor.js';
+import { runWithContext } from '../../core/context.js';
+import { isHttpError } from '../../core/http/HttpError.js';
+import type {
+ DeferredSendPlan,
+ ScheduleServiceContract,
+} from '../../core/schedules.js';
+import type { PuterServer } from '../../server.js';
+import { EVENTS_BACKGROUND_PERMISSION } from '../../services/events/authorization.js';
+import { setupTestServer } from '../../testUtil.js';
+import type { IConfig } from '../../types.js';
+import type {
+ DeferredWriteTx,
+ NotificationDriver,
+} from './NotificationDriver.js';
+
+vi.setConfig({ testTimeout: 120_000 });
+
+const BOOT_TIMEOUT_MS = 480_000;
+const TEST_TABLE = 'deferred_send_test';
+
+let server: PuterServer;
+let holder: Actor;
+let app: { id: number; uid: string };
+let saved: unknown;
+let nextId = 1;
+
+const db = () => server.clients.db;
+const driver = () =>
+ server.drivers.notifications as unknown as NotificationDriver;
+const slot = () => server.services as unknown as { schedules?: unknown };
+/** Put `schedules` in the slot, or leave the slot empty. */
+const setSchedules = (schedules: unknown): void => {
+ if (schedules === undefined) delete slot().schedules;
+ else slot().schedules = schedules;
+};
+const asApp = (): Actor => makeActor({ user: holder.user, app });
+
+const notify = (fields: Record): Promise =>
+ runWithContext({ actor: asApp() }, () =>
+ driver().create({
+ object: { value: { type: 'app.message', title: 'Hi', ...fields } },
+ }),
+ );
+
+const codeOf = async (run: Promise): Promise => {
+ try {
+ await run;
+ } catch (err) {
+ if (isHttpError(err)) return err.legacyCode;
+ throw err;
+ }
+ return undefined;
+};
+
+const mailbox = () =>
+ db().read(
+ 'SELECT * FROM `notification` WHERE `user_id` = ? AND `app_uid` = ? ' +
+ 'ORDER BY `id`',
+ [holder.user!.id, app.uid],
+ );
+
+/** A row's value: JSON text on sqlite, parsed on the other engines. */
+const valueOf = (row: Record) =>
+ typeof row.value === 'string' ? JSON.parse(row.value) : row.value;
+
+/** A send claimed under `token`, as an extension's own row would be. */
+const claimed = async (token: string): Promise => {
+ const id = nextId++;
+ await db().write(
+ `INSERT INTO \`${TEST_TABLE}\` (\`id\`, \`token\`, \`state\`) VALUES (?, ?, 'sending')`,
+ [id, token],
+ );
+ return id;
+};
+
+/** The transaction an extension wraps the write in: its row, then the mailbox. */
+const txFor = (id: number, token: string): DeferredWriteTx => ({
+ guard: {
+ table: TEST_TABLE,
+ where: "`id` = ? AND `token` = ? AND `state` = 'delivered'",
+ values: [id, token],
+ },
+ async commit(entries: readonly BatchEntry[]) {
+ await db().batchWrite([
+ {
+ statement:
+ `UPDATE \`${TEST_TABLE}\` SET \`state\` = 'delivered' ` +
+ "WHERE `id` = ? AND `token` = ? AND `state` = 'sending'",
+ values: [id, token],
+ },
+ ...entries,
+ ]);
+ const [row] = await db().pread(
+ `SELECT \`state\`, \`token\` FROM \`${TEST_TABLE}\` WHERE \`id\` = ?`,
+ [id],
+ );
+ return row?.state === 'delivered' && row.token === token;
+ },
+});
+
+const write = (
+ sendKey: string,
+ tx: DeferredWriteTx,
+ over: {
+ payload?: string | null;
+ appUid?: string;
+ signal?: AbortSignal;
+ } = {},
+) =>
+ driver().writeDeferredAppMessage(
+ {
+ holderUserId: holder.user!.id,
+ appUid: over.appUid ?? app.uid,
+ payload:
+ over.payload === undefined
+ ? JSON.stringify({ message: { title: 'Later' } })
+ : over.payload,
+ sendKey,
+ signal: over.signal ?? new AbortController().signal,
+ },
+ tx,
+ );
+
+const fakeSchedules = (plan: Partial) =>
+ ({
+ onSubscriptionsChanged: vi.fn(async () => {}),
+ createFor: vi.fn(),
+ list: vi.fn(),
+ get: vi.fn(),
+ remove: vi.fn(),
+ receivePeerChanges: vi.fn(),
+ prepareSendAt: vi.fn(async (): Promise => ({
+ sendAt: 0,
+ immediate: false,
+ holderUserId: holder.user!.id,
+ appUid: app.uid,
+ handlerDepth: 0,
+ quota: {
+ recurringPerUser: 1,
+ recurringPerApp: 1,
+ oneShotsPerUser: 1,
+ oneShotsPerApp: 1,
+ budgetPerUser: 1,
+ budgetPerApp: 1,
+ },
+ ...plan,
+ })),
+ createDeferredAppMessage: vi.fn(async (p: DeferredSendPlan) => ({
+ scheduleId: 'sched-1',
+ sendAt: p.sendAt,
+ })),
+ }) satisfies ScheduleServiceContract;
+
+beforeAll(async () => {
+ server = await setupTestServer({
+ events: { enabled: true },
+ } as IConfig);
+ const username = `sendat-${Math.random().toString(36).slice(2, 10)}`;
+ const user = await server.stores.user.create({
+ username,
+ uuid: uuidv4(),
+ password: null,
+ email: `${username}@test.local`,
+ requires_email_confirmation: false,
+ });
+ const row = (await server.stores.user.getById(user.id))!;
+ holder = makeActor({
+ user: {
+ id: row.id,
+ uuid: row.uuid,
+ username: row.username,
+ email: row.email ?? null,
+ email_confirmed: true,
+ } as Actor['user'],
+ });
+ const name = `sendat-app-${Math.random().toString(36).slice(2, 10)}`;
+ app = (await server.stores.app.create(
+ { name, title: name, index_url: `https://${name}.test/` },
+ { ownerUserId: row.id },
+ )) as { id: number; uid: string };
+ await server.services.permission.grantUserAppPermission(
+ holder,
+ app.uid,
+ EVENTS_BACKGROUND_PERMISSION,
+ );
+ await db().write(
+ `CREATE TABLE \`${TEST_TABLE}\` (\`id\` INTEGER PRIMARY KEY, ` +
+ '`token` VARCHAR(64) NOT NULL, `state` VARCHAR(16) NOT NULL)',
+ [],
+ );
+ saved = slot().schedules;
+}, BOOT_TIMEOUT_MS);
+
+beforeEach(async () => {
+ await db().write('DELETE FROM `notification` WHERE `app_uid` = ?', [
+ app.uid,
+ ]);
+});
+
+afterEach(() => {
+ setSchedules(saved);
+ vi.restoreAllMocks();
+});
+
+afterAll(async () => {
+ await server?.shutdown();
+});
+
+describe('notify({ sendAt }) without services.schedules', () => {
+ it('is refused events_disabled, even when due now, storing nothing', async () => {
+ setSchedules(undefined);
+ expect(await codeOf(notify({ sendAt: Date.now() }))).toBe(
+ 'events_disabled',
+ );
+ expect(await codeOf(notify({ sendAt: Date.now() + 3_600_000 }))).toBe(
+ 'events_disabled',
+ );
+ expect(await mailbox()).toEqual([]);
+ });
+
+ it('sends as before without sendAt', async () => {
+ setSchedules(undefined);
+ const sent = (await notify({ text: 'now' })) as { uid: string };
+ expect(sent.uid).toBeTruthy();
+ expect(await mailbox()).toHaveLength(1);
+ });
+});
+
+describe('notify({ sendAt }) with services.schedules', () => {
+ it('hands a later send to createDeferredAppMessage and answers its id', async () => {
+ const at = Date.now() + 3_600_000;
+ const schedules = fakeSchedules({ sendAt: at });
+ setSchedules(schedules);
+
+ await expect(notify({ sendAt: at, text: 'later' })).resolves.toEqual({
+ scheduled: { scheduleId: 'sched-1', sendAt: at },
+ });
+ expect(schedules.prepareSendAt).toHaveBeenCalledWith(
+ expect.objectContaining({ user: holder.user }),
+ at,
+ );
+ expect(schedules.createDeferredAppMessage).toHaveBeenCalledWith(
+ expect.objectContaining({ sendAt: at }),
+ {
+ message: expect.objectContaining({
+ title: 'Hi',
+ text: 'later',
+ }),
+ },
+ );
+ expect(await mailbox()).toEqual([]);
+ });
+
+ it('sends at once when the plan says it is due', async () => {
+ const schedules = fakeSchedules({ immediate: true });
+ setSchedules(schedules);
+
+ const sent = (await notify({ sendAt: Date.now() })) as {
+ uid: string;
+ };
+ expect(sent.uid).toBeTruthy();
+ expect(schedules.createDeferredAppMessage).not.toHaveBeenCalled();
+ expect(await mailbox()).toHaveLength(1);
+ });
+
+ it('passes a refusal from prepareSendAt through', async () => {
+ const schedules = fakeSchedules({});
+ schedules.prepareSendAt.mockRejectedValueOnce(
+ Object.assign(new Error('scoped'), {}),
+ );
+ setSchedules(schedules);
+ await expect(notify({ sendAt: Date.now() })).rejects.toThrow('scoped');
+ expect(await mailbox()).toEqual([]);
+ });
+});
+
+describe('writeDeferredAppMessage', () => {
+ it('writes one row keyed by the send, in the caller`s transaction', async () => {
+ const token = uuidv4();
+ const key = uuidv4();
+ const id = await claimed(token);
+
+ await expect(write(key, txFor(id, token))).resolves.toBe('delivered');
+ const rows = await mailbox();
+ expect(rows).toHaveLength(1);
+ expect(rows[0]).toMatchObject({ uid: key, send_key: key });
+ expect(valueOf(rows[0])).toMatchObject({ title: 'Later' });
+ });
+
+ it('writes nothing when the guard no longer holds: another claim took it', async () => {
+ const id = await claimed(uuidv4());
+ const key = uuidv4();
+
+ await expect(write(key, txFor(id, uuidv4()))).resolves.toBe('lost');
+ expect(await mailbox()).toEqual([]);
+ });
+
+ it('replaces the open row of its group, which takes the key', async () => {
+ const open = (await notify({ group: 'g', text: 'first' })) as {
+ uid: string;
+ };
+ const token = uuidv4();
+ const key = uuidv4();
+ const id = await claimed(token);
+
+ await expect(
+ write(key, txFor(id, token), {
+ payload: JSON.stringify({
+ message: { title: 'Later', group: 'g' },
+ }),
+ }),
+ ).resolves.toBe('delivered');
+ const rows = await mailbox();
+ expect(rows).toHaveLength(1);
+ expect(rows[0]).toMatchObject({ uid: open.uid, send_key: key });
+ expect(valueOf(rows[0])).toMatchObject({ title: 'Later' });
+ });
+
+ it('drops a payload it can`t read, and a sender whose app is gone', async () => {
+ const token = uuidv4();
+ const id = await claimed(token);
+ await expect(
+ write(uuidv4(), txFor(id, token), { payload: '{"nope":1}' }),
+ ).resolves.toBe('dropped');
+ await expect(
+ write(uuidv4(), txFor(id, token), { appUid: `app-${uuidv4()}` }),
+ ).resolves.toBe('dropped');
+ expect(await mailbox()).toEqual([]);
+ });
+
+ it('writes nothing once the caller`s claim is lost', async () => {
+ const token = uuidv4();
+ const id = await claimed(token);
+ const lost = new AbortController();
+ lost.abort();
+
+ await expect(
+ write(uuidv4(), txFor(id, token), { signal: lost.signal }),
+ ).rejects.toThrow();
+ expect(await mailbox()).toEqual([]);
+ });
+});
+
+/**
+ * A second claim taking a send over while the first one's transaction is open.
+ * Needs two writers at once with row locks; sqlite serialises writers, so this
+ * runs on postgres only.
+ */
+describe.runIf(process.env.PUTER_TEST_DB_ENGINE === 'postgres')(
+ 'a deferred write racing another writer',
+ () => {
+ /**
+ * Hold the first statement matching `match` on its connection, before
+ * or after it runs, until released.
+ */
+ const holdStatement = (match: RegExp, when: 'before' | 'after') => {
+ const client = db() as unknown as {
+ query: (
+ target: unknown,
+ sql: string,
+ params?: unknown[],
+ ) => Promise;
+ };
+ const original = client.query.bind(client);
+ let reached!: () => void;
+ let release!: () => void;
+ const atBarrier = new Promise((r) => (reached = r));
+ const released = new Promise((r) => (release = r));
+ let held = false;
+ vi.spyOn(client as never, 'query' as never).mockImplementation(
+ (async (target: unknown, sql: string, params?: unknown[]) => {
+ if (held || !match.test(sql))
+ return original(target, sql, params);
+ held = true;
+ if (when === 'before') {
+ reached();
+ await released;
+ return original(target, sql, params);
+ }
+ const result = await original(target, sql, params);
+ reached();
+ await released;
+ return result;
+ }) as never,
+ );
+ return { atBarrier, release };
+ };
+
+ const DELIVERED = /SET `state` = 'delivered'/;
+ const REPLACE = /`replaced_at` = \?, `send_key` = \? WHERE/;
+
+ /** Whether `promise` is still pending after `ms`. */
+ const stillPending = async (promise: Promise, ms: number) => {
+ let settled = false;
+ const watched = (async () => {
+ try {
+ await promise;
+ } finally {
+ settled = true;
+ }
+ })();
+ await new Promise((r) => setTimeout(r, ms));
+ void watched;
+ return !settled;
+ };
+
+ it('a transaction that opened before a takeover writes nothing', async () => {
+ const stale = uuidv4();
+ const id = await claimed(stale);
+ const key = uuidv4();
+ const gate = holdStatement(DELIVERED, 'before');
+ const first = write(key, txFor(id, stale));
+ await gate.atBarrier;
+
+ await db().write(
+ `UPDATE \`${TEST_TABLE}\` SET \`token\` = ? WHERE \`id\` = ?`,
+ [uuidv4(), id],
+ );
+ gate.release();
+ await expect(first).resolves.toBe('lost');
+ expect(
+ await server.stores.notification.getBySendKey(key),
+ ).toBeNull();
+ expect(await mailbox()).toEqual([]);
+ });
+
+ it('an immediate replacement waiting on the open transaction lands after it, without the key', async () => {
+ const open = (await notify({ group: 'g', text: 'first' })) as {
+ uid: string;
+ };
+ const token = uuidv4();
+ const id = await claimed(token);
+ const key = uuidv4();
+ const gate = holdStatement(REPLACE, 'after');
+ const deferred = write(key, txFor(id, token), {
+ payload: JSON.stringify({
+ message: { title: 'Later', group: 'g', text: 'later' },
+ }),
+ });
+ await gate.atBarrier;
+
+ const immediate = notify({ group: 'g', text: 'newer' });
+ // Past the group lock's wait, then blocked on the row.
+ expect(await stillPending(immediate, 2_500)).toBe(true);
+ gate.release();
+ await expect(deferred).resolves.toBe('delivered');
+ await immediate;
+
+ const rows = await mailbox();
+ expect(rows).toHaveLength(1);
+ expect(rows[0]).toMatchObject({ uid: open.uid, send_key: null });
+ expect(valueOf(rows[0]).text).toBe('newer');
+ });
+ },
+);
diff --git a/src/backend/drivers/notification/NotificationDriver.ts b/src/backend/drivers/notification/NotificationDriver.ts
index f75cd5435..791e07f04 100644
--- a/src/backend/drivers/notification/NotificationDriver.ts
+++ b/src/backend/drivers/notification/NotificationDriver.ts
@@ -51,6 +51,14 @@ import {
resolveNotificationWrite,
type NotificationTypeName,
} from '../../services/notification/notificationTypes.js';
+import {
+ schedulesUnavailable,
+ type DeferredAppMessage,
+ type DeferredSendPlan,
+ type ScheduleServiceContract,
+ type SqlGuard,
+} from '../../core/schedules.js';
+import type { BatchEntry } from '../../clients/database/DatabaseClient.js';
import { withRedisLock } from '../../util/redisLock.js';
import { PuterDriver } from '../types.js';
import type { Actor } from '../../core/actor.js';
@@ -67,6 +75,44 @@ const GROUP_LOCK_RETRY_MS = 25;
type AppMessageWindow = 'minute' | 'day';
+export type { SqlGuard };
+
+/** What a delayed app message's fire-time write needs from its row. */
+export interface DeferredWriteInput {
+ holderUserId: number;
+ appUid: string;
+ /** The stored {@link DeferredAppMessage}, as JSON text. */
+ payload: string | null;
+ /** The slot's key: the written row's uid or `send_key`, and its `sendId`. */
+ sendKey: string;
+ /** Aborted when the caller's claim is lost. */
+ signal: AbortSignal;
+}
+
+/**
+ * The caller's transaction the mailbox write commits in. Atomicity is the
+ * caller's to keep; the type can't enforce it.
+ */
+export interface DeferredWriteTx {
+ /**
+ * What every mailbox statement is fenced on: the caller's row marked
+ * delivered by this claim.
+ */
+ guard: SqlGuard;
+ /**
+ * Must be the whole delivery in one step: the caller's guarded transition
+ * of its row to delivered and `entries`, in that order, in one
+ * `batchWrite`, then a read of the row on the primary. True only when that
+ * read shows this claim holding the row as delivered. Splitting the
+ * transition from `entries` lets one slot write two rows.
+ */
+ commit(entries: readonly BatchEntry[]): Promise;
+}
+
+/** How a delayed send's write stood when it stopped. */
+export type DeferredWriteOutcome =
+ 'delivered' | 'dropped' | 'lost' | 'senderSuspended';
+
/** The mailbox slice a call named, in the terms the query takes it. */
interface MailboxScope {
audiences: readonly string[];
@@ -124,6 +170,21 @@ const parseNotifyTarget = (to: unknown): string => {
return target.handle;
};
+/** A stored delayed message, or `null` for one that can't be read back. */
+const parseDelayedAppMessage = (
+ payload: string | null,
+): DeferredAppMessage | null => {
+ try {
+ const stored = JSON.parse(payload ?? 'null') as DeferredAppMessage;
+ if (typeof stored?.message?.title !== 'string') return null;
+ if (stored.to !== undefined && typeof stored.to !== 'string')
+ return null;
+ return stored;
+ } catch {
+ return null;
+ }
+};
+
/** Whether a row belongs to the slice, as the same SQL scope selected it. */
const inScope = (
row: Record,
@@ -354,7 +415,7 @@ export class NotificationDriver extends PuterDriver {
object: Record,
value: Record,
): Promise {
- const { type, to, ...fields } = value;
+ const { type, to, sendAt, ...fields } = value;
if (type !== APP_MESSAGE_TYPE) {
throw typeNotAllowed(`Apps can create only ${APP_MESSAGE_TYPE}`);
}
@@ -370,6 +431,12 @@ export class NotificationDriver extends PuterDriver {
const handle = to === undefined ? undefined : parseNotifyTarget(to);
const message = parseAppMessage(fields);
assertValueFits(value);
+ const timing =
+ sendAt === undefined || sendAt === null
+ ? null
+ : await this.#schedules().prepareSendAt(actor, sendAt);
+ if (timing && !timing.immediate)
+ return this.#scheduleAppMessage(actor, message, handle, timing);
if (handle === undefined) {
const uid = await this.#sendAppMessage(
@@ -410,12 +477,23 @@ export class NotificationDriver extends PuterDriver {
appUid: string,
message: AppMessage,
): Promise {
- const send = () =>
- this.#replaceOrWriteAppMessage(recipientId, appUid, message);
+ return this.#oneSendPerGroup(recipientId, appUid, message, () =>
+ this.#replaceOrWriteAppMessage(recipientId, appUid, message),
+ );
+ }
+
+ /**
+ * One send per group at a time, so a burst into an empty group writes one
+ * row that the rest replace. A lock that is busy past the wait or
+ * unreachable still sends: the race costs a second row and its day slot.
+ */
+ #oneSendPerGroup(
+ recipientId: number,
+ appUid: string,
+ message: AppMessage,
+ send: () => Promise,
+ ): Promise {
if (!message.group) return send();
- // One send per group at a time, so a burst into an empty group writes
- // one row that the rest replace. A lock that is busy past the wait or
- // unreachable still sends: the race costs a second row and its day slot.
return withRedisLock(
this.clients.redis,
`notif:app-message:group:${recipientId}:${appUid}:${message.group}`,
@@ -430,6 +508,146 @@ export class NotificationDriver extends PuterDriver {
);
}
+ // -- Delayed sends -----------------------------------------------
+
+ #schedules(): ScheduleServiceContract {
+ const schedules = this.services.schedules;
+ if (!schedules) throw schedulesUnavailable();
+ return schedules;
+ }
+
+ /**
+ * Store an app's message to go out at `sendAt`. A recipient named through a
+ * share is checked now without spending its budget; every limit is spent
+ * when it goes, and the share is checked again then.
+ */
+ async #scheduleAppMessage(
+ actor: UserActor,
+ message: AppMessage,
+ handle: string | undefined,
+ plan: DeferredSendPlan,
+ ): Promise<{ scheduled: { scheduleId: string; sendAt: number } }> {
+ if (handle !== undefined)
+ await this.services.kvShare.resolveNotifyRecipient(actor, handle, {
+ spendBudget: false,
+ });
+ return {
+ scheduled: await this.#schedules().createDeferredAppMessage(
+ plan,
+ handle === undefined ? { message } : { message, to: handle },
+ ),
+ };
+ }
+
+ /**
+ * Check the sender and recipient as they stand now, spend the recipient's
+ * limits, and write the mailbox row through `tx`, fenced on its guard.
+ * Called by extensions. Refusals are the same `HttpError`s an immediate
+ * send throws.
+ */
+ async writeDeferredAppMessage(
+ input: DeferredWriteInput,
+ tx: DeferredWriteTx,
+ ): Promise {
+ const { holderUserId, sendKey, signal } = input;
+ const sender = await this.services.events.resolveScheduleActor(
+ holderUserId,
+ input.appUid,
+ );
+ if (!sender) {
+ // A lagging replica can miss an app the primary has: retried.
+ const there =
+ (await this.stores.app.uidExistsOnPrimary(input.appUid)) &&
+ (await this.stores.user.getById(holderUserId)) !== null;
+ if (there) throw new Error('the sender could not be resolved');
+ return 'dropped';
+ }
+ if (sender.user?.suspended) return 'senderSuspended';
+ const stored = parseDelayedAppMessage(input.payload);
+ if (!stored) return 'dropped';
+ const { message } = stored;
+
+ let recipientId = holderUserId;
+ let appUid = input.appUid;
+ if (stored.to !== undefined) {
+ const recipient =
+ await this.services.kvShare.resolveNotifyRecipient(
+ sender,
+ stored.to,
+ );
+ recipientId = recipient.recipientUserId;
+ appUid = recipient.appUid;
+ }
+
+ return this.#oneSendPerGroup(recipientId, appUid, message, async () => {
+ // Primary, as for an immediate send.
+ const open = message.group
+ ? await this.stores.notification.getUnseenByGroup(
+ recipientId,
+ appUid,
+ message.group,
+ { primary: true },
+ )
+ : null;
+ await this.#spendAppMessageLimits(
+ recipientId,
+ appUid,
+ open ? ['minute'] : ['minute', 'day'],
+ );
+ message.app = appMessageSender(
+ appUid,
+ await this.stores.app.getByUid(appUid),
+ );
+ // Another attempt may hold the lease now; hand the claim back.
+ if (signal.aborted) throw new Error('delayed send lost its lease');
+
+ const registered = resolveNotificationWrite(
+ APP_MESSAGE_TYPE,
+ appUid,
+ );
+ const delivered = await tx.commit(
+ this.stores.notification.sendEntries(
+ {
+ sendKey,
+ userId: recipientId,
+ value: { ...message, type: registered.type },
+ type: registered.type,
+ audience: registered.audience,
+ appUid,
+ group: message.group ?? null,
+ replaceUid: open ? String(open.uid) : null,
+ at: Date.now(),
+ },
+ tx.guard,
+ ),
+ );
+ if (!delivered) return 'lost';
+ if (open)
+ await this.#spendDayIfFresh(recipientId, appUid, open, sendKey);
+ return 'delivered';
+ });
+ }
+
+ /**
+ * A replacement that found its row shown since the lookup wrote a new one,
+ * which costs a day slot. Already written, so a full day is not a refusal.
+ */
+ async #spendDayIfFresh(
+ recipientId: number,
+ appUid: string,
+ open: Record,
+ sendKey: string,
+ ): Promise {
+ try {
+ const written =
+ await this.stores.notification.getBySendKey(sendKey);
+ if (written && written.uid !== open.uid)
+ await this.#spendAppMessageLimits(recipientId, appUid, ['day']);
+ } catch {
+ // The row is written; a slot not spent is the only loss.
+ }
+ }
+
async #replaceOrWriteAppMessage(
recipientId: number,
appUid: string,
diff --git a/src/backend/services/kvShare/KvShareService.ts b/src/backend/services/kvShare/KvShareService.ts
index 1daaec39b..4f565d9e6 100644
--- a/src/backend/services/kvShare/KvShareService.ts
+++ b/src/backend/services/kvShare/KvShareService.ts
@@ -1705,11 +1705,14 @@ export class KvShareService extends PuterService {
* The other party of a share, for a notification one party sends through
* its handle from the app whose data it is. Every check runs before the
* pair budget is spent. A handle that is unknown, retired, not the
- * caller's, or seen from another app reads as absent.
+ * caller's, or seen from another app reads as absent. `spendBudget: false`
+ * checks without spending, for a send that is only being scheduled. Called
+ * by extensions.
*/
async resolveNotifyRecipient(
actor: Actor,
handle: unknown,
+ { spendBudget = true }: { spendBudget?: boolean } = {},
): Promise {
const named = typeof handle === 'string' ? handle : '';
const senderId = actor.user?.id;
@@ -1742,7 +1745,8 @@ export class KvShareService extends PuterService {
if (!checked?.holds) throw unknownKvShareHandle(named);
if (checked.refuses) throw recipientNotAcceptingShares();
- await this.#spendNotifyBudget(senderId, recipientId, share.appUid);
+ if (spendBudget)
+ await this.#spendNotifyBudget(senderId, recipientId, share.appUid);
return {
recipientUserId: recipientId,
appUid: share.appUid,
diff --git a/src/backend/services/notification/NotificationService.test.ts b/src/backend/services/notification/NotificationService.test.ts
index 8c753dd91..5c5851c32 100644
--- a/src/backend/services/notification/NotificationService.test.ts
+++ b/src/backend/services/notification/NotificationService.test.ts
@@ -395,6 +395,36 @@ describe('NotificationService.notifyUpdate', () => {
written.stop();
});
+ it('sends the desktop rewrite before reading the row back, so a failed read still updates it', async () => {
+ const user = await makeUser();
+ const uid = await notifications.notify(
+ [user.id],
+ { title: 'one share' },
+ { type: 'share.received' },
+ );
+ const pushed = collect('outer.gui.notif.message');
+ const read = vi
+ .spyOn(server.stores.notification, 'getByUid')
+ .mockRejectedValueOnce(new Error('replica down'));
+ try {
+ await expect(
+ notifications.notifyUpdate(
+ uid,
+ user.id,
+ { title: 'two shares' },
+ { type: 'share.received' },
+ ),
+ ).rejects.toThrow('replica down');
+ expect(pushed.seen).toHaveLength(1);
+ expect((pushed.seen[0] as NotifPush).response.notification).toEqual(
+ { title: 'two shares', type: 'share.received' },
+ );
+ } finally {
+ read.mockRestore();
+ pushed.stop();
+ }
+ });
+
it('refuses to fold a type that does not group', async () => {
const user = await makeUser();
await expect(
diff --git a/src/backend/services/notification/NotificationService.ts b/src/backend/services/notification/NotificationService.ts
index 4174debb9..453d3b11f 100644
--- a/src/backend/services/notification/NotificationService.ts
+++ b/src/backend/services/notification/NotificationService.ts
@@ -181,25 +181,17 @@ export class NotificationService extends PuterService {
}
if (!opts.silent) {
- // Listeners get their own copy; the socket sends `payload`.
- this.clients.event.emit(
- 'notif.written',
+ await this.emitWritten(
{
userId,
uid,
type: registered.type,
audience: registered.audience,
appUid,
- value: structuredClone(payload),
- sendId: uuidv4(),
+ value: payload,
},
- {},
+ { sendId: uuidv4() },
);
- await this.#push(userId, uid, payload, {
- type: registered.type,
- audience: registered.audience,
- appUid,
- });
}
return uid;
}),
@@ -246,50 +238,100 @@ export class NotificationService extends PuterService {
if (!updated) return false;
if (opts.silent) return true;
+ // Before the read, so the desktop copy updates even if it fails.
const desktop = inAccountMailbox(registered.type);
- if (desktop) {
- // Same uid as the original: the client replaces what it is already
- // showing rather than stacking another copy. It goes straight to
- // the wire rather than through dispatch — the row was already
- // published, and republishing it under its own id is a duplicate
- // to anything deduplicating on `event.id`.
- this.#socket().message(userId, uid, payload);
- this.#markDelivered(uid, userId);
- }
+ if (desktop) this.#sendRewrite(userId, uid, payload);
const row = await this.stores.notification.getByUid(uid, { userId });
const app = notification.app as { uid?: unknown } | undefined;
const appUid =
row?.app_uid ?? (typeof app?.uid === 'string' ? app.uid : null);
- this.clients.event.emit(
- 'notif.written',
+ await this.emitWritten(
{
userId,
uid,
type: registered.type,
audience: registered.audience,
appUid,
- value: structuredClone(payload),
+ value: payload,
updated: true,
- sendId: uuidv4(),
+ replacedAt,
+ },
+ { sendId: uuidv4(), rewriteSent: desktop },
+ );
+ return true;
+ }
+
+ /**
+ * Deliver a notification already written: `notif.written` for listeners,
+ * then the live copy. `sendId` names this one send; a delayed send passes
+ * the same one each time it delivers, so a listener can tell a repeat. An
+ * `updated` row is an earlier notification rewritten in place, and
+ * `replacedAt` marks it a replacement. `handlerDepth` is the depth the
+ * write was made at, when no actor in context carries it. `rewriteSent`
+ * says the caller already sent a desktop rewrite. Called by extensions.
+ */
+ async emitWritten(
+ written: {
+ userId: number;
+ uid: string;
+ type: string;
+ audience: string;
+ appUid: string | null;
+ value: Record;
+ updated?: boolean;
+ replacedAt?: number;
+ },
+ opts: { sendId: string; handlerDepth?: number; rewriteSent?: boolean },
+ ): Promise {
+ const { userId, uid, value: payload, updated } = written;
+ const scope = {
+ type: written.type,
+ audience: written.audience,
+ appUid: written.appUid,
+ };
+ const desktop = inAccountMailbox(scope.type);
+ if (updated && desktop && !opts.rewriteSent)
+ this.#sendRewrite(userId, uid, payload);
+ // Listeners get their own copy; the socket sends `payload`.
+ this.clients.event.emit(
+ 'notif.written',
+ {
+ userId,
+ uid,
+ ...scope,
+ value: structuredClone(payload),
+ ...(updated ? { updated: true as const } : {}),
+ sendId: opts.sendId,
},
{},
);
- // A type kept off the desktop has dispatch as its only live path, so
- // the rewrite goes out there again, under the same uid.
- if (!desktop) {
+ // A rewrite of a type kept off the desktop has dispatch as its only
+ // live path, so it goes out there again, under the same uid.
+ if (!updated || !desktop)
await this.#push(
userId,
uid,
payload,
- {
- type: registered.type,
- audience: registered.audience,
- appUid,
- },
- replacedAt,
+ scope,
+ written.replacedAt,
+ opts.handlerDepth,
);
- }
- return true;
+ }
+
+ /**
+ * Same uid as the original: the client replaces what it is already showing
+ * rather than stacking another copy. It goes straight to the wire rather
+ * than through dispatch — the row was already published, and republishing
+ * it under its own id is a duplicate to anything deduplicating on
+ * `event.id`.
+ */
+ #sendRewrite(
+ userId: number,
+ uid: string,
+ payload: Record,
+ ): void {
+ this.#socket().message(userId, uid, payload);
+ this.#markDelivered(uid, userId);
}
/**
@@ -414,6 +456,7 @@ export class NotificationService extends PuterService {
payload: Record,
scope: { type: string; audience: string; appUid: string | null },
replacedAt?: number,
+ depth?: number,
): Promise {
const desktop = inAccountMailbox(scope.type);
if (!notificationsFoldInEnabled(this.config)) {
@@ -433,9 +476,8 @@ export class NotificationService extends PuterService {
return;
}
- const handlerDepth = handlerDepthOf(
- Context.get('actor') as Actor | undefined,
- );
+ const handlerDepth =
+ depth ?? handlerDepthOf(Context.get('actor') as Actor | undefined);
this.clients.event.emit(
'notif.created',
{
diff --git a/src/backend/stores/notification/NotificationStore.js b/src/backend/stores/notification/NotificationStore.js
index ffa3f0291..da8b2bf3e 100644
--- a/src/backend/stores/notification/NotificationStore.js
+++ b/src/backend/stores/notification/NotificationStore.js
@@ -33,6 +33,9 @@ import { PuterStore } from '../types';
* }} NotificationInsert
*/
+/** How long a probe that found no `send_key` column stands. */
+const SEND_KEY_REPROBE_MS = 60_000;
+
/**
* Fixed width and compared byte for byte on every engine, whatever `group`
* holds.
@@ -75,6 +78,15 @@ const sliceWhere = (userId, { audience, appUid, excludeTypes }) => {
};
export class NotificationStore extends PuterStore {
+ #sendKeyPresent = false;
+ #sendKeyProbedAt = Number.NEGATIVE_INFINITY;
+ /** @type {Promise | null} */
+ #sendKeyProbe = null;
+
+ async onServerStart() {
+ await this.#hasSendKey();
+ }
+
// -- Reads --------------------------------------------------------
/**
@@ -96,6 +108,20 @@ export class NotificationStore extends PuterStore {
return this.#normalizeRow(rows[0]) ?? null;
}
+ /**
+ * The row a delayed send wrote or last replaced, from the primary. Called
+ * by extensions.
+ *
+ * @param {string} sendKey
+ */
+ async getBySendKey(sendKey) {
+ const rows = await this.clients.db.pread(
+ 'SELECT * FROM `notification` WHERE `send_key` = ? LIMIT 1',
+ [sendKey],
+ );
+ return this.#normalizeRow(rows[0]) ?? null;
+ }
+
/**
* The newest row `appUid` wrote for the user under `group`, or null when it
* was shown or acknowledged: an older row is never picked up in its place.
@@ -341,6 +367,7 @@ export class NotificationStore extends PuterStore {
type,
group_key: key,
replaced_at: null,
+ send_key: null,
});
}
@@ -354,6 +381,82 @@ export class NotificationStore extends PuterStore {
return (await this.create(args)).uid;
}
+ /**
+ * The statements that write a delayed send's notification, for the caller's
+ * transaction. Each runs only while `guard` holds. With `replaceUid` the
+ * open row is replaced in place and takes `sendKey`; the insert, whose uid
+ * is `sendKey`, then runs only if that row was shown or dismissed in
+ * between. Called by extensions.
+ *
+ * @param {{
+ * sendKey: string;
+ * userId: number;
+ * value: unknown;
+ * type: string;
+ * audience: string;
+ * appUid: string | null;
+ * group?: string | null;
+ * replaceUid?: string | null;
+ * at: number;
+ * }} send
+ * `at` is epoch ms, the row's time either way.
+ * @param {{ table: string; where: string; values: unknown[] }} guard
+ * @returns {{ statement: string; values: unknown[] }[]}
+ */
+ sendEntries(send, guard) {
+ const db = this.clients.db;
+ const serialized = JSON.stringify(send.value ?? {});
+ const seconds = Math.floor(send.at / 1000);
+ const fence = `SELECT 1 FROM \`${guard.table}\` WHERE ${guard.where}`;
+ const entries = [];
+ if (send.replaceUid) {
+ entries.push({
+ statement:
+ 'UPDATE `notification` SET `value` = ?, `shown` = NULL, ' +
+ `\`created_at\` = ${db.timestampFromUnixSql()}, ` +
+ '`replaced_at` = ?, `send_key` = ? WHERE `uid` = ? ' +
+ 'AND `user_id` = ? AND `shown` IS NULL ' +
+ `AND \`acknowledged\` IS NULL AND EXISTS (${fence})`,
+ values: [
+ serialized,
+ seconds,
+ send.at,
+ send.sendKey,
+ send.replaceUid,
+ send.userId,
+ ...guard.values,
+ ],
+ });
+ }
+ // Postgres can't type a bare parameter in a SELECT list.
+ const param = (type) =>
+ db.case({ postgres: `CAST(? AS ${type})`, otherwise: '?' });
+ entries.push({
+ statement:
+ 'INSERT INTO `notification` (`uid`, `user_id`, `value`, `type`, ' +
+ '`audience`, `app_uid`, `group_key`, `created_at`, `send_key`) ' +
+ `SELECT ${param('TEXT')}, ${param('INTEGER')}, ${param('JSONB')}, ` +
+ `${param('TEXT')}, ${param('TEXT')}, ${param('TEXT')}, ` +
+ `${param('TEXT')}, ${db.timestampFromUnixSql()}, ${param('TEXT')} ` +
+ `FROM \`${guard.table}\` WHERE ${guard.where} AND NOT EXISTS ` +
+ '(SELECT 1 FROM `notification` WHERE `send_key` = ?)',
+ values: [
+ send.sendKey,
+ send.userId,
+ serialized,
+ send.type,
+ send.audience,
+ send.appUid,
+ groupKey(send.group),
+ seconds,
+ send.sendKey,
+ ...guard.values,
+ send.sendKey,
+ ],
+ });
+ return entries;
+ }
+
/**
* Rewrite a notification the recipient hasn't dismissed. `shown` is cleared
* with it, so changed wording goes out again on their next connect —
@@ -362,6 +465,10 @@ export class NotificationStore extends PuterStore {
* `onlyUnseen` also skips a row already shown. `replacedAt`, in epoch ms,
* records the rewrite as a replacement and dates the row to it.
*
+ * `send_key` is cleared with it whenever the column exists, on every node:
+ * the content is no longer what a delayed send wrote, so its recovery must
+ * not deliver it under that send's key.
+ *
* False when no such row exists, so callers can fall back to a fresh
* notification instead of dropping what they were reporting.
*
@@ -378,6 +485,7 @@ export class NotificationStore extends PuterStore {
const serialized =
typeof value === 'string' ? value : JSON.stringify(value ?? {});
const set = ['`value` = ?', '`shown` = NULL'];
+ if (await this.#hasSendKey()) set.push('`send_key` = NULL');
const params = [serialized];
if (replacedAt !== undefined) {
set.push(
@@ -478,6 +586,45 @@ export class NotificationStore extends PuterStore {
// -- Internals ----------------------------------------------------
+ /**
+ * Whether `notification.send_key` exists, asked of the primary. Present is
+ * kept for the process; absent is asked again at most once a minute, so a
+ * node started before the column was added clears it soon after.
+ *
+ * @returns {Promise}
+ */
+ async #hasSendKey() {
+ if (this.#sendKeyPresent) return true;
+ if (Date.now() - this.#sendKeyProbedAt < SEND_KEY_REPROBE_MS)
+ return false;
+ if (this.#sendKeyProbe) {
+ await this.#sendKeyProbe;
+ return this.#sendKeyPresent;
+ }
+ const probe = this.#probeSendKey();
+ this.#sendKeyProbe = probe;
+ try {
+ await probe;
+ } finally {
+ if (this.#sendKeyProbe === probe) this.#sendKeyProbe = null;
+ }
+ return this.#sendKeyPresent;
+ }
+
+ /** Never rejects: an error reads as no column. */
+ async #probeSendKey() {
+ try {
+ await this.clients.db.pread(
+ 'SELECT `send_key` FROM `notification` WHERE 1 = 0',
+ [],
+ );
+ this.#sendKeyPresent = true;
+ } catch {
+ this.#sendKeyPresent = false;
+ }
+ this.#sendKeyProbedAt = Date.now();
+ }
+
#normalizeRow(row) {
if (!row) return null;
if (typeof row.value === 'string') {
diff --git a/src/backend/stores/notification/NotificationStore.sendKey.test.ts b/src/backend/stores/notification/NotificationStore.sendKey.test.ts
new file mode 100644
index 000000000..cc8bffcf7
--- /dev/null
+++ b/src/backend/stores/notification/NotificationStore.sendKey.test.ts
@@ -0,0 +1,184 @@
+/*
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see .
+ */
+
+/**
+ * `send_key` is cleared on every rewrite once the column exists, whichever node
+ * rewrites and whatever its config: a row a delayed send wrote and another node
+ * replaced must not be delivered again under that send's key.
+ */
+
+import { v4 as uuidv4 } from 'uuid';
+import {
+ afterAll,
+ afterEach,
+ beforeAll,
+ describe,
+ expect,
+ it,
+ vi,
+} from 'vitest';
+import type { PuterServer } from '../../server.js';
+import { setupTestServer } from '../../testUtil.js';
+import type { IConfig } from '../../types.js';
+import { NotificationStore } from './NotificationStore.js';
+
+let server: PuterServer;
+let userId: number;
+
+/** A store on the same database, as another node with its own config. */
+const node = (config: Partial): NotificationStore =>
+ new NotificationStore(
+ config as IConfig,
+ server.clients as never,
+ server.stores as never,
+ );
+
+/** A row as a delayed send writes it, keyed `sendKey`. */
+const writeSent = async (store: NotificationStore): Promise => {
+ const sendKey = uuidv4();
+ await server.clients.db.batchWrite(
+ store.sendEntries(
+ {
+ sendKey,
+ userId,
+ value: { title: 'later' },
+ type: 'app-message',
+ audience: 'account',
+ appUid: null,
+ at: Date.now(),
+ },
+ { table: 'user', where: '`id` = ?', values: [userId] },
+ ),
+ );
+ return sendKey;
+};
+
+const sendKeyOf = async (uid: string): Promise => {
+ const [row] = await server.clients.db.pread(
+ 'SELECT `send_key` FROM `notification` WHERE `uid` = ?',
+ [uid],
+ );
+ return row?.send_key;
+};
+
+beforeAll(async () => {
+ server = await setupTestServer();
+ const username = `sendkey-${Math.random().toString(36).slice(2, 10)}`;
+ const user = await server.stores.user.create({
+ username,
+ uuid: uuidv4(),
+ password: null,
+ email: `${username}@test.local`,
+ });
+ userId = user.id;
+});
+
+afterAll(async () => {
+ await server?.shutdown();
+});
+
+afterEach(() => {
+ vi.restoreAllMocks();
+});
+
+describe('clearing send_key on a rewrite', () => {
+ it('clears it from a row another node wrote, whatever either config says', async () => {
+ const writer = node({ events: { enabled: true } } as Partial);
+ const rewriter = node({});
+ const sendKey = await writeSent(writer);
+ expect(await sendKeyOf(sendKey)).toBe(sendKey);
+
+ await expect(
+ rewriter.updateValue(sendKey, userId, { title: 'replaced' }),
+ ).resolves.toBe(true);
+ expect(await sendKeyOf(sendKey)).toBeNull();
+ });
+
+ it('probes once and keeps a present column for the process', async () => {
+ const store = node({});
+ const pread = vi.spyOn(server.clients.db, 'pread');
+ const first = await writeSent(store);
+ const second = await writeSent(store);
+
+ await store.updateValue(first, userId, { v: 1 });
+ await store.updateValue(second, userId, { v: 2 });
+
+ const probes = pread.mock.calls.filter(([sql]) =>
+ String(sql).includes('WHERE 1 = 0'),
+ );
+ expect(probes).toHaveLength(1);
+ expect(await sendKeyOf(first)).toBeNull();
+ expect(await sendKeyOf(second)).toBeNull();
+ });
+
+ it('rewrites without naming the column while it is absent, and clears it after the next probe once it exists', async () => {
+ const db = server.clients.db;
+ const store = node({});
+ const sendKey = await writeSent(store);
+ const write = vi.spyOn(db, 'write');
+ const realPread = db.pread.bind(db);
+ let columnExists = false;
+ vi.spyOn(db, 'pread').mockImplementation(async (sql, values) => {
+ if (!columnExists && String(sql).includes('WHERE 1 = 0'))
+ throw new Error('no such column: send_key');
+ return realPread(sql, values);
+ });
+ let now = Date.now();
+ vi.spyOn(Date, 'now').mockImplementation(() => now);
+ const rewriteSql = () =>
+ String(
+ write.mock.calls.findLast(([sql]) =>
+ String(sql).startsWith('UPDATE `notification` SET `value`'),
+ )?.[0],
+ );
+
+ await store.updateValue(sendKey, userId, { v: 1 });
+ expect(rewriteSql()).not.toContain('send_key');
+ expect(await sendKeyOf(sendKey)).toBe(sendKey);
+
+ // Within the minute the absent answer stands.
+ columnExists = true;
+ now += 30_000;
+ await store.updateValue(sendKey, userId, { v: 2 });
+ expect(rewriteSql()).not.toContain('send_key');
+ expect(await sendKeyOf(sendKey)).toBe(sendKey);
+
+ now += 31_000;
+ await store.updateValue(sendKey, userId, { v: 3 });
+ expect(rewriteSql()).toContain('`send_key` = NULL');
+ expect(await sendKeyOf(sendKey)).toBeNull();
+ });
+
+ it('shares one probe between rewrites that race it', async () => {
+ const db = server.clients.db;
+ const store = node({});
+ const keys = await Promise.all([writeSent(store), writeSent(store)]);
+ const pread = vi.spyOn(db, 'pread');
+
+ await Promise.all(
+ keys.map((uid) => store.updateValue(uid, userId, { v: 1 })),
+ );
+
+ const probes = pread.mock.calls.filter(([sql]) =>
+ String(sql).includes('WHERE 1 = 0'),
+ );
+ expect(probes).toHaveLength(1);
+ for (const uid of keys) expect(await sendKeyOf(uid)).toBeNull();
+ });
+});
diff --git a/src/backend/stores/notification/NotificationStore.test.js b/src/backend/stores/notification/NotificationStore.test.js
index 3742e37a9..afba9b0d6 100644
--- a/src/backend/stores/notification/NotificationStore.test.js
+++ b/src/backend/stores/notification/NotificationStore.test.js
@@ -346,7 +346,9 @@ describe('NotificationStore', () => {
const uids = (rows) => rows.map((r) => r.uid).sort();
expect(
- uids(await store.listScoped(u.id, { audience: 'developer', appUid })),
+ uids(
+ await store.listScoped(u.id, { audience: 'developer', appUid }),
+ ),
).toEqual([mine.uid]);
// `null` is the rows naming no app, not "any app".
expect(