Merge remote-tracking branch 'origin/main' into pr-1969

# Conflicts:
#	docs/releases.md

docs/releases.md

@@ -2297,6 +2297,8 @@ and the [ntfy Android app](https://github.com/binwiederhier/ntfy-android/release
 **Bug fixes + maintenance:**
 
 * Admin API: return 4xx errors instead of a 500 for invalid input, and remove a deleted user's web push subscriptions
+* Web app: Deleted notifications no longer come back when a poll replays messages published in the same second as the last one ([#2098](https://github.com/binwiederhier/ntfy/issues/2098), thanks to [@mattford63](https://github.com/mattford63) and [@arminus](https://github.com/arminus) for reporting)
+* Web app: A message replayed after a reconnect keeps its read state and no longer notifies again, and an older version of a sequence no longer replaces a newer one ([#2090](https://github.com/binwiederhier/ntfy/issues/2090), thanks to [@tobykaspar79](https://github.com/tobykaspar79) for reporting)
 * Check the whole server config again when `base-url` or a `message-size-limit` above 4K is set; most checks after them were skipped, so an invalid config (e.g. `upstream-base-url` equal to `base-url`) now fails to start ([#1969](https://github.com/binwiederhier/ntfy/pull/1969), thanks to [@mikejgray](https://github.com/mikejgray) for the contribution)
 
 ### ntfy iOS app v1.8.0 (UNRELEASED)

web/e2e/notifications.spec.js

@@ -9,6 +9,31 @@ let topic;
 // Browser notifications the app showed, recorded by the init script in beforeEach
 const shown = (page) => page.evaluate(() => window.shownNotifications.map((n) => n.body));
 
+// Publishes messages within one second, right after a second starts. A since=<id> marker replays
+// every message from its second, so the earlier ones come back to a client that has the last one.
+const publishInOneSecond = async (request, topic, messages) => {
+  await new Promise((resolve) => setTimeout(resolve, 1000 - (Date.now() % 1000)));
+  const times = [];
+  for (const message of messages) {
+    times.push((await (await request.post(`/${topic}`, { data: message })).json()).time);
+  }
+  expect(new Set(times).size, `published in one second: ${times}`).toBe(1);
+};
+
+// Messages the topic's WebSocket received from the server, from now on
+const liveMessages = (page, topic) => {
+  const received = [];
+  page.on("websocket", (ws) => {
+    if (new URL(ws.url()).pathname.endsWith(`/${topic}/ws`)) {
+      ws.on("framereceived", (frame) => {
+        const event = JSON.parse(frame.payload);
+        if (event.event === "message") received.push(event.message);
+      });
+    }
+  });
+  return received;
+};
+
 test.beforeEach(async ({ page }) => {
   // Record browser notifications instead of showing them; headless Chrome can't show them anyway
   await page.addInitScript(() => {
@@ -268,6 +293,49 @@ test.describe("sequences", () => {
   });
 });
 
+test.describe("replays", () => {
+  test("deleted messages don't come back when a reconnect replays them", async ({ page, request }) => {
+    await publishInOneSecond(request, topic, ["first", "second", "third"]);
+    const items = page.getByRole("listitem", { name: "Notification" });
+    await expect(items).toHaveCount(3);
+    for (const message of ["first", "second", "third"]) {
+      await items.filter({ hasText: message }).getByRole("button", { name: "Delete" }).click();
+    }
+    await expect(items).toHaveCount(0);
+
+    // Reconnecting and the startup poll ask for since=<last>, and get the rest of its second again
+    const live = liveMessages(page, topic);
+    const polled = page.waitForResponse((r) => new URL(r.url()).pathname.endsWith(`/${topic}/json`));
+    await page.reload();
+    await page.clock.runFor(2000);
+    expect(await (await polled).text()).toContain("first");
+    await expect.poll(() => live).toContain("first");
+    await page.waitForTimeout(1000); // Let the app store what it got
+    await expect(items).toHaveCount(0);
+    await expect(page.getByText("You haven't received any notifications for this topic yet.")).toBeVisible();
+  });
+
+  test("a message replayed on reconnect stays read and doesn't notify again", async ({ page, request }) => {
+    await publishInOneSecond(request, topic, ["first", "second"]);
+    await expect(page.getByRole("listitem", { name: "Notification" })).toHaveCount(2);
+
+    // Read them by opening the topic, then reconnect from another one, so one that comes back unread is counted
+    const other = unique("other");
+    await subscribe(page, other);
+    await expect(page).toHaveTitle("(2) ntfy");
+    await page.getByRole("button", { name: topic, exact: true }).click();
+    await expect(page).toHaveTitle("ntfy");
+    await page.getByRole("button", { name: other, exact: true }).click();
+    const live = liveMessages(page, topic);
+    await page.reload();
+    await page.clock.runFor(2000);
+    await expect.poll(() => live).toContain("first");
+    await page.waitForTimeout(1000); // Let the app store what it got
+    await expect(page).toHaveTitle("ntfy");
+    expect(await shown(page)).toEqual([]);
+  });
+});
+
 test.describe("browser notifications", () => {
   test("a new message is shown", async ({ page, request }) => {
     await request.post(`/${topic}`, { data: "pop up" });

web/public/sw.js

@@ -64,6 +64,11 @@ const handlePushMessage = async (data) => {
     }),
   );
 
+  // Don't store a message again that the user deleted (e.g. delivered via WebSocket first)
+  if ((await db.notifications.get(message.id))?.deleted) {
+    return;
+  }
+
   // Delete existing notification with same sequence ID (if any)
   const sequenceId = message.sequence_id || message.id;
   if (sequenceId) {

web/src/app/SubscriptionManager.js

@@ -176,7 +176,7 @@ export class SubscriptionManager {
 
     return this.db.notifications
       .orderBy("time") // Sort by time
-      .filter((n) => n.subscriptionId === subscriptionId)
+      .filter((n) => n.subscriptionId === subscriptionId && !n.deleted)
       .reverse()
       .toArray();
   }
@@ -184,6 +184,7 @@ export class SubscriptionManager {
   async getAllNotifications() {
     return this.db.notifications
       .orderBy("time") // Efficient, see docs
+      .filter((n) => !n.deleted)
       .reverse()
       .toArray();
   }
@@ -200,9 +201,24 @@ export class SubscriptionManager {
       // Add notification to database; check and add in one transaction, so a poll storing the
       // same message concurrently (see addNotifications) can't slip in between
       const added = await this.db.transaction("rw", this.db.notifications, async () => {
-        if (await this.db.notifications.get(notification.id)) {
+        // A replayed message keeps its local state (e.g. read or deleted), but one that only a poll
+        // stored (no "new" marker, see addNotifications) wasn't delivered live yet, so it's new
+        const existing = await this.db.notifications.get(notification.id);
+        if (existing && existing.new === undefined) {
+          await this.db.notifications.update(notification.id, { new: 1 });
+          return true;
+        } else if (existing) {
           return false;
         }
+
+        // An update replaces the stored versions of its sequence, unless a newer one is stored
+        // already (e.g. a reconnect replayed an older version), or was deleted
+        const sequenceId = notification.sequence_id || notification.id;
+        const versions = await this.db.notifications.where({ subscriptionId, sequenceId }).toArray();
+        if (versions.some((v) => v.time > notification.time)) {
+          return false;
+        }
+        await this.db.notifications.bulkDelete(versions.map((v) => v.id));
         await this.db.notifications.add({
           ...messageWithSequenceId(notification),
           subscriptionId,
@@ -230,9 +246,9 @@ export class SubscriptionManager {
    * older versions of the same sequence; will not throw if they exist
    */
   async addNotifications(subscriptionId, notifications) {
-    await this.db.transaction("rw", this.db.notifications, async () => {
+    const added = await this.db.transaction("rw", this.db.notifications, async () => {
       // Skip notifications that are already stored (e.g. delivered via WebSocket while this poll
-      // was in flight), so overwriting them doesn't drop their "new" marker
+      // was in flight), so overwriting them doesn't drop their "new" marker, or deleted
       const existing = await this.db.notifications.bulkGet(notifications.map((n) => n.id));
       const notificationsWithSubscriptionId = notifications
         .filter((_, i) => !existing[i])
@@ -252,7 +268,14 @@ export class SubscriptionManager {
       const replaced = versions.flat().filter((v) => updates.some((n) => n.sequenceId === v.sequenceId));
       await this.db.notifications.bulkDelete(replaced.map((v) => v.id));
       await this.db.notifications.bulkPut(updates);
+      return notificationsWithSubscriptionId.length > 0;
     });
+
+    // Don't move the marker if the poll only replayed known messages: a replay of the marker's second
+    // ends with a message older than the marker, and moving back to it replays the marker again
+    if (!added) {
+      return;
+    }
     const lastNotificationId = notifications.at(-1).id;
     await this.db.subscriptions.update(subscriptionId, {
       last: lastNotificationId,
@@ -272,8 +295,13 @@ export class SubscriptionManager {
     return true;
   }
 
+  /**
+   * Marks a notification as deleted instead of removing it: the server replays the whole second of a
+   * since=<id> marker, so a poll can return it again, and the stored row keeps it from being re-added.
+   * Deleted notifications are pruned with the others.
+   */
   async deleteNotification(notificationId) {
-    await this.db.notifications.delete(notificationId);
+    await this.db.notifications.where({ id: notificationId }).modify({ deleted: 1, new: 0 });
   }
 
   async deleteNotificationBySequenceId(subscriptionId, sequenceId) {
@@ -281,7 +309,7 @@ export class SubscriptionManager {
   }
 
   async deleteNotifications(subscriptionId) {
-    await this.db.notifications.where({ subscriptionId }).delete();
+    await this.db.notifications.where({ subscriptionId }).modify({ deleted: 1, new: 0 });
   }
 
   async markNotificationRead(notificationId) {

web/src/app/SubscriptionManager.test.js

@@ -158,6 +158,37 @@ describe("SubscriptionManager.addNotification", () => {
     expect((await db.subscriptions.get(subscriptionId)).last).toBe("m2");
   });
 
+  it("replaces an older version of the same sequence", async () => {
+    const { db, manager } = await setup();
+    await manager.addNotification(subscriptionId, message("m1", { sequence_id: "s1", message: "Downloading" }));
+
+    expect(
+      await manager.addNotification(subscriptionId, message("m2", { sequence_id: "s1", message: "Download complete", time: 1700000001 })),
+    ).toBe(true);
+
+    expect((await db.notifications.toArray()).map((n) => [n.id, n.new])).toEqual([["m2", 1]]);
+  });
+
+  it("does not replace a newer version of the same sequence when the WebSocket replays an older one", async () => {
+    const { db, manager } = await setup();
+    await manager.addNotification(subscriptionId, message("m2", { sequence_id: "s1", message: "Download complete", time: 1700000001 }));
+
+    expect(await manager.addNotification(subscriptionId, message("m1", { sequence_id: "s1", message: "Downloading" }))).toBe(false);
+
+    expect((await db.notifications.toArray()).map((n) => n.message)).toEqual(["Download complete"]);
+    expect((await db.subscriptions.get(subscriptionId)).last).toBe("m2");
+  });
+
+  it("marks a message the poll stored as unread when the WebSocket delivers it", async () => {
+    const { db, manager } = await setup();
+    await manager.addNotifications(subscriptionId, [message("m1")]);
+
+    // The poll stored it before the live delivery arrived; the user hasn't seen it yet
+    expect(await manager.addNotification(subscriptionId, message("m1"))).toBe(true);
+
+    expect((await db.notifications.get("m1")).new).toBe(1);
+  });
+
   it("does not store delete and clear events", async () => {
     const { db, manager } = await setup();
     expect(await manager.addNotification(subscriptionId, message("d1", { event: "message_delete", sequence_id: "s1" }))).toBe(false);
@@ -233,3 +264,87 @@ describe("SubscriptionManager.addNotifications", () => {
     expect((await db.notifications.toArray()).map((n) => n.message)).toEqual(["Download complete"]);
   });
 });
+
+const visibleIds = async (manager) => (await manager.getNotifications(subscriptionId)).map((n) => n.id).sort();
+
+// The server replays the whole second of a since=<id> marker, so a poll returns messages the
+// client already has, or had and deleted (#2098)
+describe("SubscriptionManager deleted notifications", () => {
+  it("does not add a deleted message again when a poll replays it", async () => {
+    const { manager } = await setup();
+    await manager.addNotifications(subscriptionId, [message("m1"), message("m2"), message("m3")]);
+    await manager.deleteNotification("m1");
+    await manager.deleteNotification("m2");
+
+    await manager.addNotifications(subscriptionId, [message("m1"), message("m2")]);
+
+    expect(await visibleIds(manager)).toEqual(["m3"]);
+  });
+
+  it("does not add a deleted message again when the WebSocket replays it", async () => {
+    const { manager } = await setup();
+    await manager.addNotification(subscriptionId, message("m1"));
+    await manager.deleteNotification("m1");
+
+    expect(await manager.addNotification(subscriptionId, message("m1"))).toBe(false);
+
+    expect(await visibleIds(manager)).toEqual([]);
+  });
+
+  it("does not add older versions of a deleted sequence again, but newer ones", async () => {
+    const { manager } = await setup();
+    await manager.addNotifications(subscriptionId, [message("m2", { sequence_id: "s1", message: "Download complete", time: 1700000001 })]);
+    await manager.deleteNotification("m2");
+
+    await manager.addNotifications(subscriptionId, [message("m1", { sequence_id: "s1", message: "Downloading" })]);
+    expect(await visibleIds(manager)).toEqual([]);
+
+    await manager.addNotification(subscriptionId, message("m3", { sequence_id: "s1", message: "Download failed", time: 1700000002 }));
+    expect(await visibleIds(manager)).toEqual(["m3"]);
+  });
+
+  it("does not add messages again after deleting all of them", async () => {
+    const { manager } = await setup();
+    await manager.addNotifications(subscriptionId, [message("m1"), message("m2")]);
+    await manager.deleteNotifications(subscriptionId);
+
+    await manager.addNotifications(subscriptionId, [message("m1"), message("m2")]);
+
+    expect(await visibleIds(manager)).toEqual([]);
+  });
+
+  it("keeps the last message when a poll only replays known or deleted messages", async () => {
+    const { db, manager } = await setup();
+    await manager.addNotifications(subscriptionId, [message("m1"), message("m2"), message("m3")]);
+    await manager.deleteNotification("m1");
+
+    await manager.addNotifications(subscriptionId, [message("m1"), message("m2")]);
+
+    expect((await db.subscriptions.get(subscriptionId)).last).toBe("m3");
+  });
+
+  it("does not list deleted messages or count them as new", async () => {
+    const { manager } = await setup();
+    await manager.addNotification(subscriptionId, message("m1"));
+    await manager.addNotification(subscriptionId, message("m2"));
+    await manager.deleteNotification("m1");
+
+    expect((await manager.getAllNotifications()).map((n) => n.id)).toEqual(["m2"]);
+    expect((await manager.all()).find((s) => s.id === subscriptionId).new).toBe(1);
+  });
+
+  it("forgets deleted messages when the subscription is removed or they are pruned", async () => {
+    const { manager } = await setup();
+    await manager.addNotifications(subscriptionId, [message("old", { time: 1600000000 }), message("m1")]);
+    await manager.deleteNotifications(subscriptionId);
+
+    await manager.pruneNotifications(1650000000);
+    await manager.addNotifications(subscriptionId, [message("old", { time: 1600000000 })]);
+    expect(await visibleIds(manager)).toEqual(["old"]);
+
+    await manager.remove({ id: subscriptionId });
+    await manager.upsert(baseUrl, "mytopic");
+    await manager.addNotifications(subscriptionId, [message("m1")]);
+    expect(await visibleIds(manager)).toEqual(["m1"]);
+  });
+});

web/src/components/hooks.js

@@ -65,11 +65,8 @@ export const useConnectionListeners = (account, subscriptions, users, webPushTop
           await subscriptionManager.markNotificationReadBySequenceId(subscription.id, notification.sequence_id);
           await notifier.cancel(subscription, notification);
         } else {
-          // Regular message: delete existing and add new
-          const sequenceId = notification.sequence_id || notification.id;
-          if (sequenceId) {
-            await subscriptionManager.deleteNotificationBySequenceId(subscription.id, sequenceId);
-          }
+          // Regular message: add it, replacing older versions of its sequence; a replayed duplicate
+          // keeps its local state (e.g. read) and doesn't notify again
           const added = await subscriptionManager.addNotification(subscription.id, notification);
           if (added) {
             await subscriptionManager.notify(subscription.id, notification);