Add DM notification manager for seen state

js/services/dmNotificationManager.js

@@ -0,0 +1,214 @@
+import {
+  getConversation,
+  listConversations,
+  updateConversationFromMessage,
+  updateConversationOpenState,
+} from "../storage/dmDb.js";
+import { userLogger } from "../utils/logger.js";
+
+function sanitizeTimestamp(value) {
+  const numeric = Number(value);
+  if (!Number.isFinite(numeric) || numeric <= 0) {
+    return 0;
+  }
+  return Math.floor(numeric);
+}
+
+function normalizeConversationState(conversation) {
+  if (!conversation || typeof conversation !== "object") {
+    return {
+      conversation_id: "",
+      opened_until: 0,
+      downloaded_until: 0,
+      unseen_count: 0,
+    };
+  }
+
+  return {
+    conversation_id: conversation.conversation_id || "",
+    opened_until: sanitizeTimestamp(conversation.opened_until),
+    downloaded_until: sanitizeTimestamp(conversation.downloaded_until),
+    unseen_count: Number.isFinite(Number(conversation.unseen_count))
+      ? Number(conversation.unseen_count)
+      : 0,
+  };
+}
+
+export class DmNotificationManager {
+  constructor({ logger } = {}) {
+    this.logger = logger || userLogger;
+    this.conversationCache = new Map();
+    this.focusedConversationId = "";
+  }
+
+  getFocusedConversationId() {
+    return this.focusedConversationId;
+  }
+
+  setFocusedConversation(conversationId, isFocused = true) {
+    if (typeof conversationId !== "string") {
+      return;
+    }
+
+    const normalized = conversationId.trim();
+    if (!normalized) {
+      return;
+    }
+
+    if (isFocused) {
+      this.focusedConversationId = normalized;
+      return;
+    }
+
+    if (this.focusedConversationId === normalized) {
+      this.focusedConversationId = "";
+    }
+  }
+
+  async ensureConversationState(conversationId) {
+    if (typeof conversationId !== "string" || !conversationId.trim()) {
+      return null;
+    }
+
+    const normalized = conversationId.trim();
+    if (this.conversationCache.has(normalized)) {
+      return this.conversationCache.get(normalized);
+    }
+
+    try {
+      const stored = await getConversation(normalized);
+      const normalizedState = normalizeConversationState({
+        conversation_id: normalized,
+        ...stored,
+      });
+      this.conversationCache.set(normalized, normalizedState);
+      return normalizedState;
+    } catch (error) {
+      this.logger.warn("[dmNotifications] Failed to hydrate conversation", error);
+      const fallback = normalizeConversationState({ conversation_id: normalized });
+      this.conversationCache.set(normalized, fallback);
+      return fallback;
+    }
+  }
+
+  async recordMessage({ record, preview = "", direction = "" } = {}) {
+    if (!record || typeof record !== "object") {
+      return null;
+    }
+
+    const conversationId =
+      typeof record.conversation_id === "string"
+        ? record.conversation_id.trim()
+        : "";
+    if (!conversationId) {
+      return null;
+    }
+
+    const createdAt = sanitizeTimestamp(record.created_at);
+    const conversationState = await this.ensureConversationState(conversationId);
+    if (!conversationState) {
+      return null;
+    }
+
+    const isIncoming = direction === "incoming";
+    const isFocused = this.focusedConversationId === conversationId;
+    const shouldNotify =
+      isIncoming && !isFocused && createdAt > conversationState.opened_until;
+    const unseenDelta =
+      isIncoming && createdAt > conversationState.opened_until ? 1 : 0;
+
+    let updated = null;
+    try {
+      updated = await updateConversationFromMessage(record, {
+        preview,
+        unseenDelta,
+        downloadedUntil: createdAt,
+      });
+    } catch (error) {
+      this.logger.warn("[dmNotifications] Failed to update conversation", error);
+    }
+
+    const normalizedUpdate = normalizeConversationState({
+      conversation_id: conversationId,
+      ...updated,
+      opened_until: conversationState.opened_until,
+    });
+
+    this.conversationCache.set(conversationId, normalizedUpdate);
+
+    return {
+      conversation: normalizedUpdate,
+      shouldNotify,
+      isFocused,
+    };
+  }
+
+  async acknowledgeRenderedMessages({ conversationId, renderedUntil } = {}) {
+    if (typeof conversationId !== "string" || !conversationId.trim()) {
+      return null;
+    }
+
+    const normalized = conversationId.trim();
+    const conversationState = await this.ensureConversationState(normalized);
+    if (!conversationState) {
+      return null;
+    }
+
+    const nextOpenedUntil = Math.max(
+      conversationState.opened_until || 0,
+      sanitizeTimestamp(renderedUntil),
+    );
+
+    let updated = null;
+    try {
+      updated = await updateConversationOpenState(normalized, {
+        openedUntil: nextOpenedUntil,
+        unseenCount: 0,
+      });
+    } catch (error) {
+      this.logger.warn("[dmNotifications] Failed to update opened_until", error);
+    }
+
+    const normalizedUpdate = normalizeConversationState({
+      conversation_id: normalized,
+      ...conversationState,
+      ...updated,
+      opened_until: nextOpenedUntil,
+      unseen_count: 0,
+    });
+
+    this.conversationCache.set(normalized, normalizedUpdate);
+    return normalizedUpdate;
+  }
+
+  async listConversationSummaries() {
+    try {
+      const stored = await listConversations();
+      const summaries = stored.map((conversation) =>
+        normalizeConversationState(conversation),
+      );
+
+      for (const summary of summaries) {
+        if (summary.conversation_id) {
+          this.conversationCache.set(summary.conversation_id, summary);
+        }
+      }
+
+      return summaries;
+    } catch (error) {
+      this.logger.warn("[dmNotifications] Failed to list conversations", error);
+      return [];
+    }
+  }
+
+  getUnseenCount(conversationId) {
+    if (typeof conversationId !== "string" || !conversationId.trim()) {
+      return 0;
+    }
+
+    const cached = this.conversationCache.get(conversationId.trim());
+    return cached && Number.isFinite(Number(cached.unseen_count))
+      ? Number(cached.unseen_count)
+      : 0;
+  }
+}

js/services/nostrService.js

@@ -9,10 +9,8 @@ import {
   saveDirectMessageSnapshot,
   clearDirectMessageSnapshot,
 } from "../directMessagesStore.js";
-import {
-  updateConversationFromMessage,
-  writeMessages,
-} from "../storage/dmDb.js";
+import { writeMessages } from "../storage/dmDb.js";
+import { DmNotificationManager } from "./dmNotificationManager.js";
 import {
   getVideosMap as getStoredVideosMap,
   setVideosMap as setStoredVideosMap,
@@ -488,6 +486,9 @@ export class NostrService {
     this.dmSubscription = null;
     this.dmActorPubkey = null;
     this.dmHydratedFromSnapshot = false;
+    this.dmNotificationManager = new DmNotificationManager({
+      logger: userLogger,
+    });
     try {
       if (this.moderationService && typeof this.moderationService.setNostrClient === "function") {
         this.moderationService.setNostrClient(this.nostrClient);
@@ -612,13 +613,23 @@ export class NostrService {
 
     try {
       await writeMessages(record);
-      await updateConversationFromMessage(record, {
+      const notificationResult = await this.dmNotificationManager.recordMessage({
+        record,
         preview: extractSnapshotPreview(message),
-        unseenDelta: message?.direction === "incoming" ? 1 : 0,
-        downloadedUntil: sanitizeSnapshotTimestamp(
-          message?.timestamp ?? message?.event?.created_at ?? record.created_at,
-        ),
+        direction: message?.direction || "",
       });
+      if (notificationResult?.conversation) {
+        this.emit("directMessages:conversationUpdated", {
+          conversation: notificationResult.conversation,
+        });
+      }
+      if (notificationResult?.shouldNotify) {
+        this.emit("directMessages:notification", {
+          conversationId: record.conversation_id,
+          message: record,
+          unseenCount: notificationResult.conversation?.unseen_count || 0,
+        });
+      }
       return record;
     } catch (error) {
       userLogger.warn("[nostrService] Failed to persist DM record", error);
@@ -626,6 +637,25 @@ export class NostrService {
     }
   }
 
+  setFocusedDirectMessageConversation(conversationId, isFocused = true) {
+    this.dmNotificationManager.setFocusedConversation(conversationId, isFocused);
+  }
+
+  async acknowledgeRenderedDirectMessages(conversationId, renderedUntil) {
+    return this.dmNotificationManager.acknowledgeRenderedMessages({
+      conversationId,
+      renderedUntil,
+    });
+  }
+
+  getDirectMessageUnseenCount(conversationId) {
+    return this.dmNotificationManager.getUnseenCount(conversationId);
+  }
+
+  async listDirectMessageConversationSummaries() {
+    return this.dmNotificationManager.listConversationSummaries();
+  }
+
   applyDirectMessage(message, { reason = "update", event = null } = {}) {
     if (!message || message.ok !== true) {
       return;

js/storage/dmDb.js

@@ -1,7 +1,7 @@
 import { devLogger, userLogger } from "../utils/logger.js";
 
 const DB_NAME = "bitvidDm";
-const DB_VERSION = 1;
+const DB_VERSION = 2;
 
 const STORES = Object.freeze({
   CONVERSATIONS: "conversations",
@@ -57,6 +57,28 @@ const MIGRATIONS = [
       }
     },
   },
+  {
+    version: 2,
+    migrate: (db, transaction) => {
+      if (!db.objectStoreNames.contains(STORES.CONVERSATIONS)) {
+        return;
+      }
+
+      const store = transaction
+        ? transaction.objectStore(STORES.CONVERSATIONS)
+        : db.transaction(STORES.CONVERSATIONS, "readwrite").objectStore(
+            STORES.CONVERSATIONS,
+          );
+      if (!store.indexNames.contains("opened_until")) {
+        store.createIndex("opened_until", "opened_until", { unique: false });
+      }
+      if (!store.indexNames.contains("downloaded_until")) {
+        store.createIndex("downloaded_until", "downloaded_until", {
+          unique: false,
+        });
+      }
+    },
+  },
 ];
 
 function isIndexedDbAvailable() {
@@ -67,10 +89,10 @@ function isIndexedDbAvailable() {
   }
 }
 
-function applyMigrations(db, oldVersion, newVersion) {
+function applyMigrations(db, oldVersion, newVersion, transaction) {
   for (const migration of MIGRATIONS) {
     if (migration.version > oldVersion && migration.version <= newVersion) {
-      migration.migrate(db);
+      migration.migrate(db, transaction);
     }
   }
 }
@@ -88,7 +110,9 @@ function openDmDb() {
       const db = request.result;
       const oldVersion = event.oldVersion ?? 0;
       const newVersion = event.newVersion ?? DB_VERSION;
-      applyMigrations(db, oldVersion, newVersion);
+      const upgradeTransaction =
+        request.transaction || event.target?.transaction || null;
+      applyMigrations(db, oldVersion, newVersion, upgradeTransaction);
     };
 
     request.onsuccess = () => resolve(request.result);
@@ -191,6 +215,7 @@ function normalizeConversation(raw) {
     downloaded_until: sanitizeTimestamp(
       raw.downloaded_until ?? raw.downloadedUntil,
     ),
+    opened_until: sanitizeTimestamp(raw.opened_until ?? raw.openedUntil),
     last_message_preview:
       typeof raw.last_message_preview === "string"
         ? raw.last_message_preview
@@ -362,6 +387,7 @@ export async function updateConversationFromMessage(message, options = {}) {
   const downloadedUntil = sanitizeTimestamp(
     options.downloadedUntil ?? normalized.created_at,
   );
+  const openedUntil = sanitizeTimestamp(options.openedUntil);
 
   try {
     const db = await openDmDb();
@@ -387,6 +413,7 @@ export async function updateConversationFromMessage(message, options = {}) {
         existing.downloaded_until || 0,
         downloadedUntil,
       ),
+      opened_until: Math.max(existing.opened_until || 0, openedUntil),
       last_message_preview: preview,
       unseen_count: Math.max(
         0,
@@ -403,6 +430,87 @@ export async function updateConversationFromMessage(message, options = {}) {
   }
 }
 
+export async function updateConversationOpenState(
+  conversationId,
+  { openedUntil, unseenCount } = {},
+) {
+  if (typeof conversationId !== "string" || !conversationId.trim()) {
+    return null;
+  }
+
+  const normalizedId = conversationId.trim();
+  const nextOpenedUntil = sanitizeTimestamp(openedUntil);
+  const nextUnseenCount = Number.isFinite(Number(unseenCount))
+    ? Math.max(0, Number(unseenCount))
+    : null;
+
+  try {
+    const db = await openDmDb();
+    if (!db) {
+      return null;
+    }
+
+    const tx = db.transaction(STORES.CONVERSATIONS, "readwrite");
+    const store = tx.objectStore(STORES.CONVERSATIONS);
+    const existing =
+      (await runRequest(store.get(normalizedId))) ||
+      normalizeConversation({ conversation_id: normalizedId });
+
+    if (!existing) {
+      return null;
+    }
+
+    const next = {
+      ...existing,
+      opened_until: Math.max(existing.opened_until || 0, nextOpenedUntil),
+      unseen_count:
+        nextUnseenCount === null
+          ? existing.unseen_count
+          : nextUnseenCount,
+    };
+
+    store.put(next);
+    await finalizeTransaction(tx);
+    return next;
+  } catch (error) {
+    userLogger.warn("[dmDb] Failed to update conversation open state", error);
+    return null;
+  }
+}
+
+export async function listConversations() {
+  try {
+    const db = await openDmDb();
+    if (!db) {
+      return [];
+    }
+
+    const tx = db.transaction(STORES.CONVERSATIONS, "readonly");
+    const store = tx.objectStore(STORES.CONVERSATIONS);
+    const results = [];
+    const request = store.openCursor();
+
+    return await new Promise((resolve) => {
+      request.onsuccess = () => {
+        const cursor = request.result;
+        if (!cursor) {
+          resolve(results);
+          return;
+        }
+        results.push(cursor.value);
+        cursor.continue();
+      };
+      request.onerror = () => {
+        userLogger.warn("[dmDb] Failed to list conversations", request.error);
+        resolve(results);
+      };
+    });
+  } catch (error) {
+    userLogger.warn("[dmDb] Failed to list conversations", error);
+    return [];
+  }
+}
+
 export function describeDmDbSchema() {
   return {
     name: DB_NAME,