From 93673e5786d589c2c197cbddce0f6e5de61a44f31f3651ca966325632820346b Mon Sep 17 00:00:00 2001 From: AidarKC Date: Wed, 22 Jul 2026 13:09:42 +0400 Subject: [PATCH] Improve direct message history and SQLite resilience --- .../java/shine/db/DatabaseInitializer.java | 1 + .../java/shine/db/SqliteDbController.java | 50 +- .../shine/db/dao/SignedMessagesV2DAO.java | 591 +++++++++++------- .../db/entities/SignedMessageV2Entry.java | 3 + .../ws_protocol/JSON/JsonHandlerRegistry.java | 4 + .../auth/Net_CreateAuthSession__Handler.java | 4 - .../auth/Net_SessionLogin_Handler.java | 4 - .../Net_GetDirectMessages_Handler.java | 98 +++ .../JSON/messages/SignedMessagesRealtime.java | 10 +- .../Net_GetDirectMessages_Request.java | 19 + .../Net_GetDirectMessages_Response.java | 71 +++ TODO/README.md | 1 + ...26-07-22_переход_с_sqlite_на_postgresql.md | 36 ++ docs/API/12_Direct_Messages_Push_Calls_API.md | 71 ++- shine-UI/js/app.js | 2 +- shine-UI/js/pages/chat-view.js | 134 +++- shine-UI/js/services/auth-service.js | 17 + shine-UI/js/state.js | 25 +- 18 files changed, 908 insertions(+), 233 deletions(-) create mode 100644 SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/Net_GetDirectMessages_Handler.java create mode 100644 SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/entyties/Net_GetDirectMessages_Request.java create mode 100644 SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/entyties/Net_GetDirectMessages_Response.java create mode 100644 TODO/medium/2026-07-22_переход_с_sqlite_на_postgresql.md diff --git a/SHiNE-server/shine-server-db/src/main/java/shine/db/DatabaseInitializer.java b/SHiNE-server/shine-server-db/src/main/java/shine/db/DatabaseInitializer.java index 81eeb4b0..d99c4706 100644 --- a/SHiNE-server/shine-server-db/src/main/java/shine/db/DatabaseInitializer.java +++ b/SHiNE-server/shine-server-db/src/main/java/shine/db/DatabaseInitializer.java @@ -641,6 +641,7 @@ public final class DatabaseInitializer { origin_session_id TEXT, receipt_ref_base_key TEXT, receipt_ref_type INTEGER, + read_at_ms INTEGER, FOREIGN KEY (from_login) REFERENCES solana_users(login), FOREIGN KEY (to_login) REFERENCES solana_users(login) ); diff --git a/SHiNE-server/shine-server-db/src/main/java/shine/db/SqliteDbController.java b/SHiNE-server/shine-server-db/src/main/java/shine/db/SqliteDbController.java index 6f39a7ba..c4cb4ab7 100644 --- a/SHiNE-server/shine-server-db/src/main/java/shine/db/SqliteDbController.java +++ b/SHiNE-server/shine-server-db/src/main/java/shine/db/SqliteDbController.java @@ -14,7 +14,7 @@ import java.sql.Statement; public final class SqliteDbController { private static volatile SqliteDbController instance; - private static final int LATEST_SCHEMA_VERSION = 11; + private static final int LATEST_SCHEMA_VERSION = 12; private final String jdbcUrl; @@ -94,6 +94,7 @@ public final class SqliteDbController { case 9 -> migrateToV9(); case 10 -> migrateToV10(); case 11 -> migrateToV11(); + case 12 -> migrateToV12(); default -> throw new RuntimeException("Unknown DB migration target version: " + targetVersion); } } @@ -329,6 +330,26 @@ public final class SqliteDbController { } } + private void migrateToV12() { + try (Connection c = DriverManager.getConnection(jdbcUrl); + Statement st = c.createStatement()) { + c.setAutoCommit(false); + try { + ensureSignedMessagesReadAtColumn(c, st); + backfillSignedMessagesReadAt(st); + setSchemaVersion(c, 12); + c.commit(); + } catch (Exception e) { + try { c.rollback(); } catch (Exception ignored) {} + throw new RuntimeException("DB migration to v12 failed", e); + } finally { + try { c.setAutoCommit(true); } catch (Exception ignored) {} + } + } catch (SQLException e) { + throw new RuntimeException("DB migration to v12 failed", e); + } + } + private static void ensureChat200StateTables(Statement st) throws SQLException { st.executeUpdate(""" CREATE TABLE IF NOT EXISTS chat200_state ( @@ -463,6 +484,33 @@ public final class SqliteDbController { } } + private static void ensureSignedMessagesReadAtColumn(Connection c, Statement st) throws SQLException { + if (!tableExists(c, "signed_messages_v2")) return; + if (!columnExists(c, "signed_messages_v2", "read_at_ms")) { + st.executeUpdate("ALTER TABLE signed_messages_v2 ADD COLUMN read_at_ms INTEGER"); + } + } + + private static void backfillSignedMessagesReadAt(Statement st) throws SQLException { + st.executeUpdate(""" + UPDATE signed_messages_v2 AS content + SET read_at_ms = ( + SELECT MIN(receipt.time_ms) + FROM signed_messages_v2 AS receipt + WHERE receipt.message_type IN (3, 4) + AND receipt.receipt_ref_base_key = content.base_key + ) + WHERE content.message_type IN (1, 2) + AND (content.read_at_ms IS NULL OR content.read_at_ms <= 0) + AND EXISTS ( + SELECT 1 + FROM signed_messages_v2 AS receipt + WHERE receipt.message_type IN (3, 4) + AND receipt.receipt_ref_base_key = content.base_key + ); + """); + } + /** * Временная одноразовая миграция на переходе к SHiNE_DM v1: * старые строки signed_messages_v2 больше не гарантированно совместимы diff --git a/SHiNE-server/shine-server-db/src/main/java/shine/db/dao/SignedMessagesV2DAO.java b/SHiNE-server/shine-server-db/src/main/java/shine/db/dao/SignedMessagesV2DAO.java index 530f4b4b..85af6e05 100644 --- a/SHiNE-server/shine-server-db/src/main/java/shine/db/dao/SignedMessagesV2DAO.java +++ b/SHiNE-server/shine-server-db/src/main/java/shine/db/dao/SignedMessagesV2DAO.java @@ -7,10 +7,14 @@ import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; +import java.sql.Statement; import java.util.ArrayList; import java.util.List; public final class SignedMessagesV2DAO { + private static final int SQLITE_BUSY_MAX_RETRIES = 6; + private static final long SQLITE_BUSY_RETRY_BASE_DELAY_MS = 40L; + public enum ApplyStatus { APPLIED, DUPLICATE_OR_OLDER, @@ -37,174 +41,195 @@ public final class SignedMessagesV2DAO { } public ApplyStatus insertIfAbsent(SignedMessageV2Entry e) throws Exception { - try (Connection c = db.getConnection()) { - if (isBlockedByConversationDelete(c, e.getFromLogin(), e.getToLogin(), e.getTimeMs())) { - return ApplyStatus.BLOCKED_BY_CONVERSATION_TOMBSTONE; + return withBusyRetry(() -> { + try (Connection c = db.getConnection()) { + if (isBlockedByConversationDelete(c, e.getFromLogin(), e.getToLogin(), e.getTimeMs())) { + return ApplyStatus.BLOCKED_BY_CONVERSATION_TOMBSTONE; + } + String sql = """ + INSERT OR IGNORE INTO signed_messages_v2 ( + message_key, base_key, target_login, from_login, to_login, + time_ms, nonce, message_type, revision_time_ms, reencrypted_at_ms, + raw_block, created_at_ms, source_api, origin_session_id, + receipt_ref_base_key, receipt_ref_type, read_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """; + try (PreparedStatement ps = c.prepareStatement(sql)) { + bindSignedMessage(ps, e); + ApplyStatus status = ps.executeUpdate() > 0 ? ApplyStatus.APPLIED : ApplyStatus.DUPLICATE_OR_OLDER; + if (status.applied()) { + markMessageReadByReceipt(c, e); + } + return status; + } } - String sql = """ - INSERT OR IGNORE INTO signed_messages_v2 ( - message_key, base_key, target_login, from_login, to_login, - time_ms, nonce, message_type, revision_time_ms, reencrypted_at_ms, - raw_block, created_at_ms, source_api, origin_session_id, - receipt_ref_base_key, receipt_ref_type - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """; - try (PreparedStatement ps = c.prepareStatement(sql)) { - bindSignedMessage(ps, e); - return ps.executeUpdate() > 0 ? ApplyStatus.APPLIED : ApplyStatus.DUPLICATE_OR_OLDER; - } - } + }); } public boolean insertPairBothOrNothing(SignedMessageV2Entry first, SignedMessageV2Entry second) throws Exception { - try (Connection c = db.getConnection()) { - boolean prevAutoCommit = c.getAutoCommit(); - c.setAutoCommit(false); - try { - int insertedFirst = insertStrict(c, first); - int insertedSecond = insertStrict(c, second); - if (insertedFirst == 1 && insertedSecond == 1) { - c.commit(); - return true; - } - c.rollback(); - return false; - } catch (SQLException sqlEx) { - try { c.rollback(); } catch (Exception ignored) {} - if (isConstraintViolation(sqlEx)) { + return withBusyRetry(() -> { + try (Connection c = db.getConnection()) { + boolean prevAutoCommit = c.getAutoCommit(); + c.setAutoCommit(false); + try { + int insertedFirst = insertStrict(c, first); + int insertedSecond = insertStrict(c, second); + if (insertedFirst == 1 && insertedSecond == 1) { + markMessageReadByReceipt(c, first); + markMessageReadByReceipt(c, second); + c.commit(); + return true; + } + c.rollback(); return false; + } catch (SQLException sqlEx) { + try { c.rollback(); } catch (Exception ignored) {} + if (isConstraintViolation(sqlEx)) { + return false; + } + throw sqlEx; + } finally { + c.setAutoCommit(prevAutoCommit); } - throw sqlEx; - } finally { - c.setAutoCommit(prevAutoCommit); } - } + }); } public ApplyStatus upsertContentPair(SignedMessageV2Entry incoming, SignedMessageV2Entry outgoing) throws Exception { - try (Connection c = db.getConnection()) { - boolean prevAutoCommit = c.getAutoCommit(); - c.setAutoCommit(false); - try { - if (isBlockedByConversationDelete(c, incoming.getFromLogin(), incoming.getToLogin(), incoming.getTimeMs())) { - c.rollback(); - return ApplyStatus.BLOCKED_BY_CONVERSATION_TOMBSTONE; - } - if (hasMessageDeleteTombstone(c, incoming.getBaseKey())) { - c.rollback(); - return ApplyStatus.BLOCKED_BY_MESSAGE_TOMBSTONE; - } + return withBusyRetry(() -> { + try (Connection c = db.getConnection()) { + boolean prevAutoCommit = c.getAutoCommit(); + c.setAutoCommit(false); + try { + if (isBlockedByConversationDelete(c, incoming.getFromLogin(), incoming.getToLogin(), incoming.getTimeMs())) { + c.rollback(); + return ApplyStatus.BLOCKED_BY_CONVERSATION_TOMBSTONE; + } + if (hasMessageDeleteTombstone(c, incoming.getBaseKey())) { + c.rollback(); + return ApplyStatus.BLOCKED_BY_MESSAGE_TOMBSTONE; + } - RevisionMarker currentMarker = getCurrentContentMarker(c, incoming.getBaseKey()); - RevisionMarker nextMarker = RevisionMarker.of(incoming); - if (currentMarker != null && compareMarkers(nextMarker, currentMarker) <= 0) { - c.rollback(); - return ApplyStatus.DUPLICATE_OR_OLDER; + RevisionMarker currentMarker = getCurrentContentMarker(c, incoming.getBaseKey()); + RevisionMarker nextMarker = RevisionMarker.of(incoming); + if (currentMarker != null && compareMarkers(nextMarker, currentMarker) <= 0) { + c.rollback(); + return ApplyStatus.DUPLICATE_OR_OLDER; + } + + upsertMessage(c, incoming); + upsertMessage(c, outgoing); + markMessageReadByReceipt(c, incoming); + markMessageReadByReceipt(c, outgoing); + resetDeliveryRows(c, incoming.getMessageKey()); + resetDeliveryRows(c, outgoing.getMessageKey()); + + c.commit(); + return ApplyStatus.APPLIED; + } catch (Exception ex) { + try { c.rollback(); } catch (Exception ignored) {} + throw ex; + } finally { + c.setAutoCommit(prevAutoCommit); } - - upsertMessage(c, incoming); - upsertMessage(c, outgoing); - resetDeliveryRows(c, incoming.getMessageKey()); - resetDeliveryRows(c, outgoing.getMessageKey()); - - c.commit(); - return ApplyStatus.APPLIED; - } catch (Exception ex) { - try { c.rollback(); } catch (Exception ignored) {} - throw ex; - } finally { - c.setAutoCommit(prevAutoCommit); } - } + }); } public ApplyStatus upsertIncomingCopy(SignedMessageV2Entry incoming) throws Exception { - try (Connection c = db.getConnection()) { - boolean prevAutoCommit = c.getAutoCommit(); - c.setAutoCommit(false); - try { - if (isBlockedByConversationDelete(c, incoming.getFromLogin(), incoming.getToLogin(), incoming.getTimeMs())) { - c.rollback(); - return ApplyStatus.BLOCKED_BY_CONVERSATION_TOMBSTONE; - } - if (hasMessageDeleteTombstone(c, incoming.getBaseKey())) { - c.rollback(); - return ApplyStatus.BLOCKED_BY_MESSAGE_TOMBSTONE; - } + return withBusyRetry(() -> { + try (Connection c = db.getConnection()) { + boolean prevAutoCommit = c.getAutoCommit(); + c.setAutoCommit(false); + try { + if (isBlockedByConversationDelete(c, incoming.getFromLogin(), incoming.getToLogin(), incoming.getTimeMs())) { + c.rollback(); + return ApplyStatus.BLOCKED_BY_CONVERSATION_TOMBSTONE; + } + if (hasMessageDeleteTombstone(c, incoming.getBaseKey())) { + c.rollback(); + return ApplyStatus.BLOCKED_BY_MESSAGE_TOMBSTONE; + } - RevisionMarker currentMarker = getRevisionMarkerByMessageKey(c, incoming.getMessageKey()); - RevisionMarker nextMarker = RevisionMarker.of(incoming); - if (currentMarker != null && compareMarkers(nextMarker, currentMarker) <= 0) { - c.rollback(); - return ApplyStatus.DUPLICATE_OR_OLDER; - } + RevisionMarker currentMarker = getRevisionMarkerByMessageKey(c, incoming.getMessageKey()); + RevisionMarker nextMarker = RevisionMarker.of(incoming); + if (currentMarker != null && compareMarkers(nextMarker, currentMarker) <= 0) { + c.rollback(); + return ApplyStatus.DUPLICATE_OR_OLDER; + } - upsertMessage(c, incoming); - resetDeliveryRows(c, incoming.getMessageKey()); - c.commit(); - return ApplyStatus.APPLIED; - } catch (Exception ex) { - try { c.rollback(); } catch (Exception ignored) {} - throw ex; - } finally { - c.setAutoCommit(prevAutoCommit); + upsertMessage(c, incoming); + markMessageReadByReceipt(c, incoming); + resetDeliveryRows(c, incoming.getMessageKey()); + c.commit(); + return ApplyStatus.APPLIED; + } catch (Exception ex) { + try { c.rollback(); } catch (Exception ignored) {} + throw ex; + } finally { + c.setAutoCommit(prevAutoCommit); + } } - } + }); } public ApplyStatus applyDeleteMessage(SignedMessageV2Entry tombstone) throws Exception { - try (Connection c = db.getConnection()) { - boolean prevAutoCommit = c.getAutoCommit(); - c.setAutoCommit(false); - try { - if (isBlockedByConversationDelete(c, tombstone.getFromLogin(), tombstone.getToLogin(), tombstone.getTimeMs())) { - c.rollback(); - return ApplyStatus.BLOCKED_BY_CONVERSATION_TOMBSTONE; - } - if (hasMessageDeleteTombstone(c, tombstone.getBaseKey())) { - c.rollback(); - return ApplyStatus.DUPLICATE_OR_OLDER; - } + return withBusyRetry(() -> { + try (Connection c = db.getConnection()) { + boolean prevAutoCommit = c.getAutoCommit(); + c.setAutoCommit(false); + try { + if (isBlockedByConversationDelete(c, tombstone.getFromLogin(), tombstone.getToLogin(), tombstone.getTimeMs())) { + c.rollback(); + return ApplyStatus.BLOCKED_BY_CONVERSATION_TOMBSTONE; + } + if (hasMessageDeleteTombstone(c, tombstone.getBaseKey())) { + c.rollback(); + return ApplyStatus.DUPLICATE_OR_OLDER; + } - deleteMessageContentAndReceipts(c, tombstone.getBaseKey()); - upsertMessage(c, tombstone); - resetDeliveryRows(c, tombstone.getMessageKey()); + deleteMessageContentAndReceipts(c, tombstone.getBaseKey()); + upsertMessage(c, tombstone); + resetDeliveryRows(c, tombstone.getMessageKey()); - c.commit(); - return ApplyStatus.APPLIED; - } catch (Exception ex) { - try { c.rollback(); } catch (Exception ignored) {} - throw ex; - } finally { - c.setAutoCommit(prevAutoCommit); + c.commit(); + return ApplyStatus.APPLIED; + } catch (Exception ex) { + try { c.rollback(); } catch (Exception ignored) {} + throw ex; + } finally { + c.setAutoCommit(prevAutoCommit); + } } - } + }); } public ApplyStatus applyDeleteConversation(SignedMessageV2Entry tombstone) throws Exception { - try (Connection c = db.getConnection()) { - boolean prevAutoCommit = c.getAutoCommit(); - c.setAutoCommit(false); - try { - Long currentBoundary = getLatestConversationDeleteBoundary(c, tombstone.getFromLogin(), tombstone.getToLogin()); - if (currentBoundary != null && tombstone.getTimeMs() <= currentBoundary) { - c.rollback(); - return ApplyStatus.DUPLICATE_OR_OLDER; + return withBusyRetry(() -> { + try (Connection c = db.getConnection()) { + boolean prevAutoCommit = c.getAutoCommit(); + c.setAutoCommit(false); + try { + Long currentBoundary = getLatestConversationDeleteBoundary(c, tombstone.getFromLogin(), tombstone.getToLogin()); + if (currentBoundary != null && tombstone.getTimeMs() <= currentBoundary) { + c.rollback(); + return ApplyStatus.DUPLICATE_OR_OLDER; + } + + deleteConversationHistoryBefore(c, tombstone.getFromLogin(), tombstone.getToLogin(), tombstone.getTimeMs()); + upsertMessage(c, tombstone); + resetDeliveryRows(c, tombstone.getMessageKey()); + + c.commit(); + return ApplyStatus.APPLIED; + } catch (Exception ex) { + try { c.rollback(); } catch (Exception ignored) {} + throw ex; + } finally { + c.setAutoCommit(prevAutoCommit); } - - deleteConversationHistoryBefore(c, tombstone.getFromLogin(), tombstone.getToLogin(), tombstone.getTimeMs()); - upsertMessage(c, tombstone); - resetDeliveryRows(c, tombstone.getMessageKey()); - - c.commit(); - return ApplyStatus.APPLIED; - } catch (Exception ex) { - try { c.rollback(); } catch (Exception ignored) {} - throw ex; - } finally { - c.setAutoCommit(prevAutoCommit); } - } + }); } public SignedMessageV2Entry getByMessageKey(String messageKey) throws Exception { @@ -214,7 +239,7 @@ public final class SignedMessagesV2DAO { message_key, base_key, target_login, from_login, to_login, time_ms, nonce, message_type, revision_time_ms, reencrypted_at_ms, raw_block, created_at_ms, source_api, origin_session_id, - receipt_ref_base_key, receipt_ref_type + receipt_ref_base_key, receipt_ref_type, read_at_ms FROM signed_messages_v2 WHERE message_key = ? """; @@ -235,90 +260,153 @@ public final class SignedMessagesV2DAO { } public void ensureDeliveryRow(String messageKey, String sessionId, long nowMs) throws Exception { - try (Connection c = db.getConnection()) { - String sql = """ - INSERT OR IGNORE INTO signed_message_session_delivery ( - message_key, session_id, delivered, delivered_at_ms, created_at_ms - ) VALUES (?, ?, 0, NULL, ?) - """; - try (PreparedStatement ps = c.prepareStatement(sql)) { - ps.setString(1, messageKey); - ps.setString(2, sessionId); - ps.setLong(3, nowMs); - ps.executeUpdate(); + ensureDeliveryRows(messageKey, List.of(sessionId), nowMs); + } + + public void ensureDeliveryRows(String messageKey, List sessionIds, long nowMs) throws Exception { + if (sessionIds == null || sessionIds.isEmpty()) return; + withBusyRetry(() -> { + try (Connection c = db.getConnection()) { + String sql = """ + INSERT OR IGNORE INTO signed_message_session_delivery ( + message_key, session_id, delivered, delivered_at_ms, created_at_ms + ) VALUES (?, ?, 0, NULL, ?) + """; + try (PreparedStatement ps = c.prepareStatement(sql)) { + for (String sessionId : sessionIds) { + if (sessionId == null || sessionId.isBlank()) continue; + ps.setString(1, messageKey); + ps.setString(2, sessionId); + ps.setLong(3, nowMs); + ps.addBatch(); + } + ps.executeBatch(); + } + return null; } - } + }); } public void markDelivered(String messageKey, String sessionId, long deliveredAtMs) throws Exception { - try (Connection c = db.getConnection()) { - String insertSql = """ - INSERT OR IGNORE INTO signed_message_session_delivery ( - message_key, session_id, delivered, delivered_at_ms, created_at_ms - ) VALUES (?, ?, 0, NULL, ?) - """; - try (PreparedStatement ps = c.prepareStatement(insertSql)) { - ps.setString(1, messageKey); - ps.setString(2, sessionId); - ps.setLong(3, deliveredAtMs); - ps.executeUpdate(); + withBusyRetry(() -> { + try (Connection c = db.getConnection()) { + String sql = """ + INSERT INTO signed_message_session_delivery ( + message_key, session_id, delivered, delivered_at_ms, created_at_ms + ) VALUES (?, ?, 1, ?, ?) + ON CONFLICT(message_key, session_id) DO UPDATE SET + delivered = 1, + delivered_at_ms = CASE + WHEN signed_message_session_delivery.delivered_at_ms IS NULL THEN excluded.delivered_at_ms + WHEN signed_message_session_delivery.delivered_at_ms > excluded.delivered_at_ms THEN excluded.delivered_at_ms + ELSE signed_message_session_delivery.delivered_at_ms + END + """; + try (PreparedStatement ps = c.prepareStatement(sql)) { + ps.setString(1, messageKey); + ps.setString(2, sessionId); + ps.setLong(3, deliveredAtMs); + ps.setLong(4, deliveredAtMs); + ps.executeUpdate(); + } + return null; } - - String updateSql = """ - UPDATE signed_message_session_delivery - SET delivered = 1, delivered_at_ms = ? - WHERE message_key = ? AND session_id = ? - """; - try (PreparedStatement ps = c.prepareStatement(updateSql)) { - ps.setLong(1, deliveredAtMs); - ps.setString(2, messageKey); - ps.setString(3, sessionId); - ps.executeUpdate(); - } - } + }); } public List listPendingForSession(String login, String sessionId) throws Exception { - try (Connection c = db.getConnection()) { - String fillSql = """ - INSERT OR IGNORE INTO signed_message_session_delivery ( - message_key, session_id, delivered, delivered_at_ms, created_at_ms - ) - SELECT m.message_key, ?, 0, NULL, ? - FROM signed_messages_v2 m - WHERE ( - (m.message_type IN (1, 3) AND m.to_login = ? COLLATE NOCASE) - OR (m.message_type IN (2, 4) AND m.from_login = ? COLLATE NOCASE) - OR (m.message_type IN (5, 6, 7, 8) - AND (m.from_login = ? COLLATE NOCASE OR m.to_login = ? COLLATE NOCASE)) - ) - """; - long now = System.currentTimeMillis(); - try (PreparedStatement ps = c.prepareStatement(fillSql)) { - ps.setString(1, sessionId); - ps.setLong(2, now); - ps.setString(3, login); - ps.setString(4, login); - ps.setString(5, login); - ps.setString(6, login); - ps.executeUpdate(); - } + return withBusyRetry(() -> { + try (Connection c = db.getConnection()) { + String fillSql = """ + INSERT OR IGNORE INTO signed_message_session_delivery ( + message_key, session_id, delivered, delivered_at_ms, created_at_ms + ) + SELECT m.message_key, ?, 0, NULL, ? + FROM signed_messages_v2 m + WHERE ( + (m.message_type IN (1, 3) AND m.to_login = ? COLLATE NOCASE) + OR (m.message_type IN (2, 4) AND m.from_login = ? COLLATE NOCASE) + OR (m.message_type IN (5, 6, 7, 8) + AND (m.from_login = ? COLLATE NOCASE OR m.to_login = ? COLLATE NOCASE)) + ) + """; + long now = System.currentTimeMillis(); + try (PreparedStatement ps = c.prepareStatement(fillSql)) { + ps.setString(1, sessionId); + ps.setLong(2, now); + ps.setString(3, login); + ps.setString(4, login); + ps.setString(5, login); + ps.setString(6, login); + ps.executeUpdate(); + } + String sql = """ + SELECT + m.message_key, m.base_key, m.target_login, m.from_login, m.to_login, + m.time_ms, m.nonce, m.message_type, m.revision_time_ms, m.reencrypted_at_ms, + m.raw_block, m.created_at_ms, m.source_api, m.origin_session_id, + m.receipt_ref_base_key, m.receipt_ref_type, m.read_at_ms + FROM signed_messages_v2 m + JOIN signed_message_session_delivery d + ON d.message_key = m.message_key + WHERE d.session_id = ? AND d.delivered = 0 + ORDER BY m.time_ms ASC, m.revision_time_ms ASC, m.reencrypted_at_ms ASC, m.created_at_ms ASC + """; + List out = new ArrayList<>(); + try (PreparedStatement ps = c.prepareStatement(sql)) { + ps.setString(1, sessionId); + try (ResultSet rs = ps.executeQuery()) { + while (rs.next()) out.add(mapRow(rs)); + } + } + return out; + } + }); + } + + public List listConversationPage( + String login, + String peerLogin, + long beforeTimeMs, + String beforeMessageKey, + int limit + ) throws Exception { + try (Connection c = db.getConnection()) { String sql = """ SELECT - m.message_key, m.base_key, m.target_login, m.from_login, m.to_login, - m.time_ms, m.nonce, m.message_type, m.revision_time_ms, m.reencrypted_at_ms, - m.raw_block, m.created_at_ms, m.source_api, m.origin_session_id, - m.receipt_ref_base_key, m.receipt_ref_type - FROM signed_messages_v2 m - JOIN signed_message_session_delivery d - ON d.message_key = m.message_key - WHERE d.session_id = ? AND d.delivered = 0 - ORDER BY m.time_ms ASC, m.revision_time_ms ASC, m.reencrypted_at_ms ASC, m.created_at_ms ASC + message_key, base_key, target_login, from_login, to_login, + time_ms, nonce, message_type, revision_time_ms, reencrypted_at_ms, + raw_block, created_at_ms, source_api, origin_session_id, + receipt_ref_base_key, receipt_ref_type, read_at_ms + FROM signed_messages_v2 + WHERE target_login = ? COLLATE NOCASE + AND message_type IN (1, 2) + AND ( + (from_login = ? COLLATE NOCASE AND to_login = ? COLLATE NOCASE) + OR (from_login = ? COLLATE NOCASE AND to_login = ? COLLATE NOCASE) + ) + AND ( + ? <= 0 + OR time_ms < ? + OR (time_ms = ? AND (? = '' OR message_key < ?)) + ) + ORDER BY time_ms DESC, message_key DESC + LIMIT ? """; List out = new ArrayList<>(); try (PreparedStatement ps = c.prepareStatement(sql)) { - ps.setString(1, sessionId); + ps.setString(1, login); + ps.setString(2, login); + ps.setString(3, peerLogin); + ps.setString(4, peerLogin); + ps.setString(5, login); + ps.setLong(6, beforeTimeMs); + ps.setLong(7, beforeTimeMs); + ps.setLong(8, beforeTimeMs); + ps.setString(9, beforeMessageKey == null ? "" : beforeMessageKey); + ps.setString(10, beforeMessageKey == null ? "" : beforeMessageKey); + ps.setInt(11, limit); try (ResultSet rs = ps.executeQuery()) { while (rs.next()) out.add(mapRow(rs)); } @@ -333,8 +421,8 @@ public final class SignedMessagesV2DAO { message_key, base_key, target_login, from_login, to_login, time_ms, nonce, message_type, revision_time_ms, reencrypted_at_ms, raw_block, created_at_ms, source_api, origin_session_id, - receipt_ref_base_key, receipt_ref_type - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + receipt_ref_base_key, receipt_ref_type, read_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(message_key) DO UPDATE SET base_key = excluded.base_key, target_login = excluded.target_login, @@ -350,7 +438,8 @@ public final class SignedMessagesV2DAO { source_api = excluded.source_api, origin_session_id = excluded.origin_session_id, receipt_ref_base_key = excluded.receipt_ref_base_key, - receipt_ref_type = excluded.receipt_ref_type + receipt_ref_type = excluded.receipt_ref_type, + read_at_ms = COALESCE(signed_messages_v2.read_at_ms, excluded.read_at_ms) """; try (PreparedStatement ps = c.prepareStatement(sql)) { bindSignedMessage(ps, e); @@ -358,6 +447,32 @@ public final class SignedMessagesV2DAO { } } + private void markMessageReadByReceipt(Connection c, SignedMessageV2Entry entry) throws SQLException { + if (entry == null) return; + int messageType = entry.getMessageType(); + if (messageType != 3 && messageType != 4) return; + String receiptRefBaseKey = String.valueOf(entry.getReceiptRefBaseKey() == null ? "" : entry.getReceiptRefBaseKey()).trim(); + if (receiptRefBaseKey.isEmpty()) return; + long readAtMs = entry.getTimeMs(); + if (readAtMs <= 0) return; + try (PreparedStatement ps = c.prepareStatement(""" + UPDATE signed_messages_v2 + SET read_at_ms = CASE + WHEN read_at_ms IS NULL OR read_at_ms <= 0 THEN ? + WHEN read_at_ms > ? THEN ? + ELSE read_at_ms + END + WHERE base_key = ? + AND message_type IN (1, 2) + """)) { + ps.setLong(1, readAtMs); + ps.setLong(2, readAtMs); + ps.setLong(3, readAtMs); + ps.setString(4, receiptRefBaseKey); + ps.executeUpdate(); + } + } + private RevisionMarker getRevisionMarkerByMessageKey(Connection c, String messageKey) throws SQLException { String sql = """ SELECT revision_time_ms, reencrypted_at_ms @@ -536,8 +651,8 @@ public final class SignedMessagesV2DAO { message_key, base_key, target_login, from_login, to_login, time_ms, nonce, message_type, revision_time_ms, reencrypted_at_ms, raw_block, created_at_ms, source_api, origin_session_id, - receipt_ref_base_key, receipt_ref_type - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + receipt_ref_base_key, receipt_ref_type, read_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """; try (PreparedStatement ps = c.prepareStatement(sql)) { bindSignedMessage(ps, e); @@ -563,6 +678,8 @@ public final class SignedMessagesV2DAO { ps.setString(15, e.getReceiptRefBaseKey()); if (e.getReceiptRefType() == null) ps.setObject(16, null); else ps.setInt(16, e.getReceiptRefType()); + if (e.getReadAtMs() == null) ps.setObject(17, null); + else ps.setLong(17, e.getReadAtMs()); } private void bindObjects(PreparedStatement ps, Object... bindValues) throws SQLException { @@ -586,6 +703,45 @@ public final class SignedMessagesV2DAO { return msg.contains("constraint") || msg.contains("unique") || msg.contains("primary key"); } + private boolean isBusyLock(SQLException ex) { + Throwable current = ex; + while (current != null) { + String msg = String.valueOf(current.getMessage()).toLowerCase(); + if (msg.contains("sqlite_busy") || msg.contains("database is locked") || msg.contains("database table is locked")) { + return true; + } + current = current.getCause(); + } + return false; + } + + private void sleepBeforeBusyRetry(int attempt) throws SQLException { + long delayMs = SQLITE_BUSY_RETRY_BASE_DELAY_MS * (1L << Math.min(attempt, 4)); + try { + Thread.sleep(delayMs); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + SQLException sqlEx = new SQLException("Interrupted while retrying SQLite busy lock", ie); + throw sqlEx; + } + } + + private T withBusyRetry(SqlWork work) throws Exception { + SQLException lastBusy = null; + for (int attempt = 0; attempt < SQLITE_BUSY_MAX_RETRIES; attempt++) { + try { + return work.run(); + } catch (SQLException ex) { + if (!isBusyLock(ex) || attempt >= SQLITE_BUSY_MAX_RETRIES - 1) { + throw ex; + } + lastBusy = ex; + sleepBeforeBusyRetry(attempt); + } + } + throw lastBusy == null ? new SQLException("SQLite busy retry failed") : lastBusy; + } + private int compareMarkers(RevisionMarker left, RevisionMarker right) { int revisionCompare = Long.compare(left.revisionTimeMs, right.revisionTimeMs); if (revisionCompare != 0) return revisionCompare; @@ -611,6 +767,8 @@ public final class SignedMessagesV2DAO { e.setReceiptRefBaseKey(rs.getString("receipt_ref_base_key")); int maybeRefType = rs.getInt("receipt_ref_type"); e.setReceiptRefType(rs.wasNull() ? null : maybeRefType); + long maybeReadAt = rs.getLong("read_at_ms"); + e.setReadAtMs(rs.wasNull() ? null : maybeReadAt); return e; } @@ -619,4 +777,9 @@ public final class SignedMessagesV2DAO { return new RevisionMarker(entry.getRevisionTimeMs(), entry.getReencryptedAtMs()); } } + + @FunctionalInterface + private interface SqlWork { + T run() throws Exception; + } } diff --git a/SHiNE-server/shine-server-db/src/main/java/shine/db/entities/SignedMessageV2Entry.java b/SHiNE-server/shine-server-db/src/main/java/shine/db/entities/SignedMessageV2Entry.java index 84e4bdfa..d19f2d63 100644 --- a/SHiNE-server/shine-server-db/src/main/java/shine/db/entities/SignedMessageV2Entry.java +++ b/SHiNE-server/shine-server-db/src/main/java/shine/db/entities/SignedMessageV2Entry.java @@ -17,6 +17,7 @@ public class SignedMessageV2Entry { private String originSessionId; private String receiptRefBaseKey; private Integer receiptRefType; + private Long readAtMs; public String getMessageKey() { return messageKey; } public void setMessageKey(String messageKey) { this.messageKey = messageKey; } @@ -50,4 +51,6 @@ public class SignedMessageV2Entry { public void setReceiptRefBaseKey(String receiptRefBaseKey) { this.receiptRefBaseKey = receiptRefBaseKey; } public Integer getReceiptRefType() { return receiptRefType; } public void setReceiptRefType(Integer receiptRefType) { this.receiptRefType = receiptRefType; } + public Long getReadAtMs() { return readAtMs; } + public void setReadAtMs(Long readAtMs) { this.readAtMs = readAtMs; } } diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/JsonHandlerRegistry.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/JsonHandlerRegistry.java index 24157745..9d2121aa 100644 --- a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/JsonHandlerRegistry.java +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/JsonHandlerRegistry.java @@ -91,6 +91,7 @@ import server.logic.ws_protocol.JSON.messages.Net_CallInviteBroadcast_Handler; import server.logic.ws_protocol.JSON.messages.Net_CallSignalToSession_Handler; import server.logic.ws_protocol.JSON.messages.Net_DeleteConversation_Handler; import server.logic.ws_protocol.JSON.messages.Net_DeleteMessage_Handler; +import server.logic.ws_protocol.JSON.messages.Net_GetDirectMessages_Handler; import server.logic.ws_protocol.JSON.messages.Net_SendSignal_Handler; import server.logic.ws_protocol.JSON.messages.Net_ReceiveIncomingMessage_Handler; import server.logic.ws_protocol.JSON.messages.Net_SendDirectMessage_Handler; @@ -102,6 +103,7 @@ import server.logic.ws_protocol.JSON.messages.entyties.Net_CallInviteBroadcast_R import server.logic.ws_protocol.JSON.messages.entyties.Net_CallSignalToSession_Request; import server.logic.ws_protocol.JSON.messages.entyties.Net_DeleteConversation_Request; import server.logic.ws_protocol.JSON.messages.entyties.Net_DeleteMessage_Request; +import server.logic.ws_protocol.JSON.messages.entyties.Net_GetDirectMessages_Request; import server.logic.ws_protocol.JSON.messages.entyties.Net_SendSignal_Request; import server.logic.ws_protocol.JSON.messages.entyties.Net_ReceiveIncomingMessage_Request; import server.logic.ws_protocol.JSON.messages.entyties.Net_SendDirectMessage_Request; @@ -200,6 +202,7 @@ public final class JsonHandlerRegistry { Map.entry("ReceiveIncomingMessage", new Net_ReceiveIncomingMessage_Handler()), Map.entry("DeleteMessage", new Net_DeleteMessage_Handler()), Map.entry("DeleteConversation", new Net_DeleteConversation_Handler()), + Map.entry("GetDirectMessages", new Net_GetDirectMessages_Handler()), Map.entry("AckSessionDelivery", new Net_AckSessionDelivery_Handler()), Map.entry("CallInviteBroadcast", new Net_CallInviteBroadcast_Handler()), Map.entry("CallSignalToSession", new Net_CallSignalToSession_Handler()), @@ -280,6 +283,7 @@ public final class JsonHandlerRegistry { Map.entry("ReceiveIncomingMessage", Net_ReceiveIncomingMessage_Request.class), Map.entry("DeleteMessage", Net_DeleteMessage_Request.class), Map.entry("DeleteConversation", Net_DeleteConversation_Request.class), + Map.entry("GetDirectMessages", Net_GetDirectMessages_Request.class), Map.entry("AckSessionDelivery", Net_AckSessionDelivery_Request.class), Map.entry("CallInviteBroadcast", Net_CallInviteBroadcast_Request.class), Map.entry("CallSignalToSession", Net_CallSignalToSession_Request.class), diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/auth/Net_CreateAuthSession__Handler.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/auth/Net_CreateAuthSession__Handler.java index 2e963c79..cebbfbde 100644 --- a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/auth/Net_CreateAuthSession__Handler.java +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/auth/Net_CreateAuthSession__Handler.java @@ -10,7 +10,6 @@ import server.logic.ws_protocol.JSON.entyties.Net_Response; import server.logic.ws_protocol.JSON.handlers.JsonMessageHandler; import server.logic.ws_protocol.JSON.handlers.auth.entyties.Net_CreateAuthSession_Request; import server.logic.ws_protocol.JSON.handlers.auth.entyties.Net_CreateAuthSession_Response; -import server.logic.ws_protocol.JSON.messages.SignedMessagesRealtime; import server.logic.ws_protocol.JSON.utils.AuthKeyUtils; import server.logic.ws_protocol.JSON.utils.NetExceptionResponseFactory; import server.logic.ws_protocol.WireCodes; @@ -51,8 +50,6 @@ public class Net_CreateAuthSession__Handler implements JsonMessageHandler { private static final Logger log = LoggerFactory.getLogger(Net_CreateAuthSession__Handler.class); private static final SecureRandom RANDOM = new SecureRandom(); private static final long CLOSE_AFTER_ERROR_DELAY_MS = 75L; - private static final long SIGNED_DM_BACKLOG_AFTER_AUTH_DELAY_MS = 250L; - public static final long ALLOWED_SKEW_MS = 30_000L; @Override @@ -424,7 +421,6 @@ public class Net_CreateAuthSession__Handler implements JsonMessageHandler { ctx.setAuthenticationStatus(ConnectionContext.AUTH_STATUS_USER); ActiveConnectionsRegistry.getInstance().register(ctx); - SignedMessagesRealtime.dispatchPendingForSessionAsync(ctx, SIGNED_DM_BACKLOG_AFTER_AUTH_DELAY_MS); // --- формируем ответ --- Net_CreateAuthSession_Response resp = new Net_CreateAuthSession_Response(); diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/auth/Net_SessionLogin_Handler.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/auth/Net_SessionLogin_Handler.java index bc4685e7..6a4f51a1 100644 --- a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/auth/Net_SessionLogin_Handler.java +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/auth/Net_SessionLogin_Handler.java @@ -10,7 +10,6 @@ import server.logic.ws_protocol.JSON.entyties.Net_Response; import server.logic.ws_protocol.JSON.handlers.JsonMessageHandler; import server.logic.ws_protocol.JSON.handlers.auth.entyties.Net_SessionLogin_Request; import server.logic.ws_protocol.JSON.handlers.auth.entyties.Net_SessionLogin_Response; -import server.logic.ws_protocol.JSON.messages.SignedMessagesRealtime; import server.logic.ws_protocol.JSON.utils.AuthKeyUtils; import server.logic.ws_protocol.JSON.utils.NetExceptionResponseFactory; import server.logic.ws_protocol.WireCodes; @@ -44,8 +43,6 @@ public class Net_SessionLogin_Handler implements JsonMessageHandler { private static final Logger log = LoggerFactory.getLogger(Net_SessionLogin_Handler.class); private static final long ALLOWED_SKEW_MS = 30_000L; - private static final long SIGNED_DM_BACKLOG_AFTER_AUTH_DELAY_MS = 250L; - @Override public Net_Response handle(Net_Request baseReq, ConnectionContext ctx) throws Exception { Net_SessionLogin_Request req = (Net_SessionLogin_Request) baseReq; @@ -302,7 +299,6 @@ public class Net_SessionLogin_Handler implements JsonMessageHandler { ctx.setAuthenticationStatus(ConnectionContext.AUTH_STATUS_USER); ActiveConnectionsRegistry.getInstance().register(ctx); - SignedMessagesRealtime.dispatchPendingForSessionAsync(ctx, SIGNED_DM_BACKLOG_AFTER_AUTH_DELAY_MS); // ответ Net_SessionLogin_Response resp = new Net_SessionLogin_Response(); diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/Net_GetDirectMessages_Handler.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/Net_GetDirectMessages_Handler.java new file mode 100644 index 00000000..19d7f5f9 --- /dev/null +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/Net_GetDirectMessages_Handler.java @@ -0,0 +1,98 @@ +package server.logic.ws_protocol.JSON.messages; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import server.logic.ws_protocol.JSON.ConnectionContext; +import server.logic.ws_protocol.JSON.entyties.Net_Request; +import server.logic.ws_protocol.JSON.entyties.Net_Response; +import server.logic.ws_protocol.JSON.handlers.JsonMessageHandler; +import server.logic.ws_protocol.JSON.messages.entyties.Net_GetDirectMessages_Request; +import server.logic.ws_protocol.JSON.messages.entyties.Net_GetDirectMessages_Response; +import server.logic.ws_protocol.JSON.utils.NetExceptionResponseFactory; +import server.logic.ws_protocol.WireCodes; +import shine.db.dao.SignedMessagesV2DAO; +import shine.db.entities.SignedMessageV2Entry; + +import java.util.ArrayList; +import java.util.Base64; +import java.util.List; + +public class Net_GetDirectMessages_Handler implements JsonMessageHandler { + private static final Logger log = LoggerFactory.getLogger(Net_GetDirectMessages_Handler.class); + private static final int DEFAULT_LIMIT = 50; + private static final int MAX_LIMIT = 200; + + @Override + public Net_Response handle(Net_Request baseRequest, ConnectionContext ctx) { + Net_GetDirectMessages_Request req = (Net_GetDirectMessages_Request) baseRequest; + if (ctx == null || !ctx.isAuthenticatedUser()) { + return NetExceptionResponseFactory.error(req, WireCodes.Status.UNVERIFIED, "NOT_AUTHENTICATED", "Требуется авторизация"); + } + if (req.getPeerLogin() == null || req.getPeerLogin().isBlank()) { + return NetExceptionResponseFactory.error(req, WireCodes.Status.BAD_REQUEST, "BAD_FIELDS", "peerLogin обязателен"); + } + + int limit = req.getLimit() == null ? DEFAULT_LIMIT : req.getLimit(); + if (limit <= 0 || limit > MAX_LIMIT) { + return NetExceptionResponseFactory.error(req, WireCodes.Status.BAD_REQUEST, "BAD_LIMIT", "limit должен быть в диапазоне 1.." + MAX_LIMIT); + } + + String login = ctx.getLogin().trim(); + String peerLogin = req.getPeerLogin().trim(); + long beforeTimeMs = req.getBeforeTimeMs() == null ? 0L : req.getBeforeTimeMs(); + String beforeMessageKey = req.getBeforeMessageKey() == null ? "" : req.getBeforeMessageKey().trim(); + + try { + List page = SignedMessagesV2DAO.getInstance().listConversationPage( + login, + peerLogin, + beforeTimeMs, + beforeMessageKey, + limit + 1 + ); + + boolean hasMore = page.size() > limit; + if (hasMore) { + page = new ArrayList<>(page.subList(0, limit)); + } + + Net_GetDirectMessages_Response resp = new Net_GetDirectMessages_Response(); + resp.setOp(req.getOp()); + resp.setRequestId(req.getRequestId()); + resp.setStatus(WireCodes.Status.OK); + resp.setLogin(login); + resp.setPeerLogin(peerLogin); + resp.setLimit(limit); + resp.setHasMore(hasMore); + + List items = new ArrayList<>(); + for (SignedMessageV2Entry entry : page) { + Net_GetDirectMessages_Response.MessageItem item = new Net_GetDirectMessages_Response.MessageItem(); + item.setMessageKey(entry.getMessageKey()); + item.setBaseKey(entry.getBaseKey()); + item.setFromLogin(entry.getFromLogin()); + item.setToLogin(entry.getToLogin()); + item.setMessageType(entry.getMessageType()); + item.setTimeMs(entry.getTimeMs()); + item.setNonce(entry.getNonce()); + item.setRevisionTimeMs(entry.getRevisionTimeMs()); + item.setReencryptedAtMs(entry.getReencryptedAtMs()); + item.setCreatedAtMs(entry.getCreatedAtMs()); + item.setReadAtMs(entry.getReadAtMs()); + item.setBlobB64(Base64.getEncoder().encodeToString(entry.getRawBlock())); + items.add(item); + } + resp.setMessages(items); + + if (hasMore && !items.isEmpty()) { + Net_GetDirectMessages_Response.MessageItem last = items.get(items.size() - 1); + resp.setNextBeforeTimeMs(last.getTimeMs()); + resp.setNextBeforeMessageKey(last.getMessageKey()); + } + return resp; + } catch (Exception e) { + log.error("GetDirectMessages failed for login={} peerLogin={}", login, peerLogin, e); + return NetExceptionResponseFactory.error(req, WireCodes.Status.INTERNAL_ERROR, "INTERNAL_ERROR", "Внутренняя ошибка сервера"); + } + } +} diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/SignedMessagesRealtime.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/SignedMessagesRealtime.java index b37804a1..2d681dc1 100644 --- a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/SignedMessagesRealtime.java +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/SignedMessagesRealtime.java @@ -50,12 +50,20 @@ public final class SignedMessagesRealtime { long now = System.currentTimeMillis(); for (String targetLogin : targetLoginsForMessage(message)) { List sessions = ActiveSessionsDAO.getInstance().getByLogin(targetLogin); + List sessionIdsToTrack = new ArrayList<>(); + for (ActiveSessionEntry s : sessions) { + String sessionId = s.getSessionId(); + if (excludeSessionId != null && excludeSessionId.equals(sessionId)) { + continue; + } + sessionIdsToTrack.add(sessionId); + } + SignedMessagesV2DAO.getInstance().ensureDeliveryRows(message.getMessageKey(), sessionIdsToTrack, now); for (ActiveSessionEntry s : sessions) { String sessionId = s.getSessionId(); if (excludeSessionId != null && excludeSessionId.equals(sessionId)) { continue; } - SignedMessagesV2DAO.getInstance().ensureDeliveryRow(message.getMessageKey(), sessionId, now); boolean deliveredOnline = sendEventToSessionIfOnline(sessionId, targetLogin, message, false); if (deliveredOnline) { counters.wsDelivered++; diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/entyties/Net_GetDirectMessages_Request.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/entyties/Net_GetDirectMessages_Request.java new file mode 100644 index 00000000..47f1da4c --- /dev/null +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/entyties/Net_GetDirectMessages_Request.java @@ -0,0 +1,19 @@ +package server.logic.ws_protocol.JSON.messages.entyties; + +import server.logic.ws_protocol.JSON.entyties.Net_Request; + +public class Net_GetDirectMessages_Request extends Net_Request { + private String peerLogin; + private Integer limit; + private Long beforeTimeMs; + private String beforeMessageKey; + + public String getPeerLogin() { return peerLogin; } + public void setPeerLogin(String peerLogin) { this.peerLogin = peerLogin; } + public Integer getLimit() { return limit; } + public void setLimit(Integer limit) { this.limit = limit; } + public Long getBeforeTimeMs() { return beforeTimeMs; } + public void setBeforeTimeMs(Long beforeTimeMs) { this.beforeTimeMs = beforeTimeMs; } + public String getBeforeMessageKey() { return beforeMessageKey; } + public void setBeforeMessageKey(String beforeMessageKey) { this.beforeMessageKey = beforeMessageKey; } +} diff --git a/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/entyties/Net_GetDirectMessages_Response.java b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/entyties/Net_GetDirectMessages_Response.java new file mode 100644 index 00000000..fdb914cb --- /dev/null +++ b/SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/entyties/Net_GetDirectMessages_Response.java @@ -0,0 +1,71 @@ +package server.logic.ws_protocol.JSON.messages.entyties; + +import server.logic.ws_protocol.JSON.entyties.Net_Response; + +import java.util.ArrayList; +import java.util.List; + +public class Net_GetDirectMessages_Response extends Net_Response { + private String login; + private String peerLogin; + private int limit; + private boolean hasMore; + private Long nextBeforeTimeMs; + private String nextBeforeMessageKey; + private List messages = new ArrayList<>(); + + public String getLogin() { return login; } + public void setLogin(String login) { this.login = login; } + public String getPeerLogin() { return peerLogin; } + public void setPeerLogin(String peerLogin) { this.peerLogin = peerLogin; } + public int getLimit() { return limit; } + public void setLimit(int limit) { this.limit = limit; } + public boolean isHasMore() { return hasMore; } + public void setHasMore(boolean hasMore) { this.hasMore = hasMore; } + public Long getNextBeforeTimeMs() { return nextBeforeTimeMs; } + public void setNextBeforeTimeMs(Long nextBeforeTimeMs) { this.nextBeforeTimeMs = nextBeforeTimeMs; } + public String getNextBeforeMessageKey() { return nextBeforeMessageKey; } + public void setNextBeforeMessageKey(String nextBeforeMessageKey) { this.nextBeforeMessageKey = nextBeforeMessageKey; } + public List getMessages() { return messages; } + public void setMessages(List messages) { this.messages = messages; } + + public static class MessageItem { + private String messageKey; + private String baseKey; + private String fromLogin; + private String toLogin; + private int messageType; + private long timeMs; + private long nonce; + private long revisionTimeMs; + private long reencryptedAtMs; + private long createdAtMs; + private Long readAtMs; + private String blobB64; + + public String getMessageKey() { return messageKey; } + public void setMessageKey(String messageKey) { this.messageKey = messageKey; } + public String getBaseKey() { return baseKey; } + public void setBaseKey(String baseKey) { this.baseKey = baseKey; } + public String getFromLogin() { return fromLogin; } + public void setFromLogin(String fromLogin) { this.fromLogin = fromLogin; } + public String getToLogin() { return toLogin; } + public void setToLogin(String toLogin) { this.toLogin = toLogin; } + public int getMessageType() { return messageType; } + public void setMessageType(int messageType) { this.messageType = messageType; } + public long getTimeMs() { return timeMs; } + public void setTimeMs(long timeMs) { this.timeMs = timeMs; } + public long getNonce() { return nonce; } + public void setNonce(long nonce) { this.nonce = nonce; } + public long getRevisionTimeMs() { return revisionTimeMs; } + public void setRevisionTimeMs(long revisionTimeMs) { this.revisionTimeMs = revisionTimeMs; } + public long getReencryptedAtMs() { return reencryptedAtMs; } + public void setReencryptedAtMs(long reencryptedAtMs) { this.reencryptedAtMs = reencryptedAtMs; } + public long getCreatedAtMs() { return createdAtMs; } + public void setCreatedAtMs(long createdAtMs) { this.createdAtMs = createdAtMs; } + public Long getReadAtMs() { return readAtMs; } + public void setReadAtMs(Long readAtMs) { this.readAtMs = readAtMs; } + public String getBlobB64() { return blobB64; } + public void setBlobB64(String blobB64) { this.blobB64 = blobB64; } + } +} diff --git a/TODO/README.md b/TODO/README.md index 5475728f..6d4c72ff 100644 --- a/TODO/README.md +++ b/TODO/README.md @@ -49,6 +49,7 @@ - `medium/2026-05-26_0029_esp32s3_file_storage.md` - ESP32S3 как личное файловое хранилище SHiNE для файлов переписок и вложений. - `medium/2026-06-02_сессионные_homeserver_в_pda.md` - несколько homeserver-ов пользователя как типизированные сессии в PDA с версией записи. - `medium/2026-06-03_подключение_других_устройств_через_qr.md` - довести подключение других устройств через QR: сейчас заготовка есть, но сценарий работает нестабильно и его нужно будет отдельно доделать. +- `medium/2026-07-22_переход_с_sqlite_на_postgresql.md` - подготовить перевод серверной БД с `SQLite` на `PostgreSQL` для более серьёзной конкурентной нагрузки и дальнейшего масштабирования. ### dao_запуск diff --git a/TODO/medium/2026-07-22_переход_с_sqlite_на_postgresql.md b/TODO/medium/2026-07-22_переход_с_sqlite_на_postgresql.md new file mode 100644 index 00000000..3be89265 --- /dev/null +++ b/TODO/medium/2026-07-22_переход_с_sqlite_на_postgresql.md @@ -0,0 +1,36 @@ +# Переход с SQLite на PostgreSQL + +## Зачем + +Текущая серверная база на `SQLite` удобна для простого односерверного режима, но она хуже подходит для большого числа параллельных записей, роста нагрузки и дальнейшего масштабирования сервера. + +`PostgreSQL` нужен как следующий уровень серверной БД для более надёжной конкурентной записи, более предсказуемой работы под нагрузкой и дальнейшего роста проекта. + +## Что сделать + +- Подготовить план переноса серверной БД с `SQLite` на `PostgreSQL`. +- Найти все места, где код завязан на особенности `SQLite`. +- Проверить все DAO и SQL-запросы на совместимость с `PostgreSQL`. +- Продумать схему миграции существующей production/test базы без потери данных. +- Отдельно проверить транзакции, `UPSERT`, индексы, case-insensitive сравнения и миграции схемы. +- После этого подготовить отдельный этап внедрения и переключения сервера. + +## Что уже есть в коде + +- Доступ к БД в основном проходит через DAO-слой, а не полностью размазан по проекту. +- Основная серверная логика уже разделена по модулям. +- Но SQL и миграции сейчас написаны под `SQLite` и потребуют отдельного прохода. + +## Откуда продолжать + +- Начать с инвентаризации всех DAO и схемы БД. +- После этого сделать отдельный документ с оценкой объёма работ по переносу. +- Затем решить, будет ли это: + - полный перевод сервера на `PostgreSQL`; + - или поддержка двух драйверов на переходный период. + +## Что потом обновить + +- Серверную документацию по БД и миграциям. +- Инструкции по локальному запуску сервера. +- Скрипты деплоя и настройки окружения. diff --git a/docs/API/12_Direct_Messages_Push_Calls_API.md b/docs/API/12_Direct_Messages_Push_Calls_API.md index 2a12d0a1..114b0d03 100644 --- a/docs/API/12_Direct_Messages_Push_Calls_API.md +++ b/docs/API/12_Direct_Messages_Push_Calls_API.md @@ -176,7 +176,72 @@ } ``` -## 7. `AckSessionDelivery` +## 7. `GetDirectMessages` + +Требует авторизации. Возвращает историю диалога с конкретным собеседником страницами. + +Важно: + +- начиная с 22 июля 2026 года сервер больше не высылает старую DM-историю автоматически при логине; +- после подключения клиент должен сам запросить первую страницу диалога; +- новые realtime-сообщения по-прежнему приходят событием `SignedMessageArrived`. +- в `GetDirectMessages` сервер отдаёт только обычные chat-сообщения (`type=1/2`), без read-receipt и delete/tombstone. + +### Запрос + +```json +{ + "op": "GetDirectMessages", + "requestId": "dm-history-001", + "payload": { + "peerLogin": "bob", + "limit": 50, + "beforeTimeMs": 1774700000123, + "beforeMessageKey": "alice|bob|1774700000123|123456789|1" + } +} +``` + +`beforeTimeMs` и `beforeMessageKey` необязательны. Если их нет, сервер вернёт самую новую страницу. + +### Успешный ответ + +```json +{ + "op": "GetDirectMessages", + "requestId": "dm-history-001", + "status": 200, + "ok": true, + "payload": { + "login": "alice", + "peerLogin": "bob", + "limit": 50, + "hasMore": true, + "nextBeforeTimeMs": 1774699999000, + "nextBeforeMessageKey": "alice|bob|1774699999000|123456780|2", + "messages": [ + { + "messageKey": "alice|bob|1774700000123|123456789|1", + "baseKey": "alice|bob|1774700000123|123456789", + "fromLogin": "alice", + "toLogin": "bob", + "messageType": 1, + "timeMs": 1774700000123, + "nonce": 123456789, + "revisionTimeMs": 0, + "reencryptedAtMs": 0, + "createdAtMs": 1774700001123, + "readAtMs": 1774700001456, + "blobB64": "BASE64_SIGNED_BLOCK" + } + ] + } +} +``` + +Для следующей страницы клиент должен передать `nextBeforeTimeMs` и `nextBeforeMessageKey` из предыдущего ответа. + +## 8. `AckSessionDelivery` Требует авторизации. Подтверждает доставку в текущую сессию. @@ -192,7 +257,7 @@ } ``` -## 8. Событие `SignedMessageArrived` +## 9. Событие `SignedMessageArrived` Сервер присылает его по WebSocket в активные сессии адресата. @@ -217,7 +282,7 @@ Для типов `5/6/7/8` событие тоже приходит в таком же конверте, но логика применения определяется `messageType` и бинарным `blobB64`. -## 9. `CallInviteBroadcast` +## 10. `CallInviteBroadcast` Требует авторизации. Шлёт приглашение к звонку в активные сессии `toLogin`. diff --git a/shine-UI/js/app.js b/shine-UI/js/app.js index 27f95d22..f24721a0 100644 --- a/shine-UI/js/app.js +++ b/shine-UI/js/app.js @@ -1203,7 +1203,7 @@ async function init() { } if (refBaseKey) { if (messageType === 3) { - markOutgoingReadByBaseKey(refBaseKey); + markOutgoingReadByBaseKey(refBaseKey, Number(parsed.timeMs || 0)); } else { if (markIncomingReadByBaseKey(refBaseKey)) { shouldRefreshToolbarUnread = true; diff --git a/shine-UI/js/pages/chat-view.js b/shine-UI/js/pages/chat-view.js index 7ecb9cd6..ce2e08b7 100644 --- a/shine-UI/js/pages/chat-view.js +++ b/shine-UI/js/pages/chat-view.js @@ -144,6 +144,7 @@ function openMessageActionsMenu({ anchorX = 0, anchorY = 0, messageText = '', + infoText = '', canReply = false, showReadAloud = true, canEdit = false, @@ -165,6 +166,7 @@ function openMessageActionsMenu({ ${showReadAloud ? '' : ''} ${canEdit ? '' : ''} ${canDelete ? '' : ''} + ${String(infoText || '').trim() ? `
${String(infoText || '').trim()}
` : ''} `; @@ -438,9 +440,27 @@ function buildCallStatusText(callSummary) { return 'Не дозвонился'; } -function resolveDeliveryStatus(msg) { +function resolveEffectiveReadState(messages, msg) { + if (msg?.from !== 'out') return { isRead: false, readAtMs: 0 }; + const explicitReadAtMs = Number(msg?.readAtMs || 0); + if (Number.isFinite(explicitReadAtMs) && explicitReadAtMs > 0) { + return { isRead: true, readAtMs: explicitReadAtMs }; + } + if (msg?.secondTick) { + return { isRead: true, readAtMs: 0 }; + } + const messageTimeMs = resolveMessageTimeMs(msg); + const hasNewerRead = (messages || []).some((row) => { + if (row?.from !== 'out') return false; + if (resolveMessageTimeMs(row) <= messageTimeMs) return false; + return Boolean(Number(row?.readAtMs || 0) > 0 || row?.secondTick); + }); + return { isRead: hasNewerRead, readAtMs: 0 }; +} + +function resolveDeliveryStatus(messages, msg) { if (msg?.from !== 'out') return ''; - if (msg?.secondTick) return '✓✓'; + if (resolveEffectiveReadState(messages, msg).isRead) return '✓✓'; if (msg?.firstTick) return '✓'; return '…'; } @@ -625,7 +645,7 @@ function renderLog(list, chatId, { onOpenActions, markAsRead = true, scrollMode timeNode.textContent = formatMessageTime(resolveMessageTimeMs(msg)); metaNode.append(timeNode); - const status = resolveDeliveryStatus(msg); + const status = resolveDeliveryStatus(messages, msg); if (status) { const statusNode = document.createElement('span'); statusNode.className = 'bubble-status'; @@ -684,6 +704,49 @@ function preserveComposerSelection(input, callback) { } } +async function mergeDirectMessagesPage(chatId, payloadMessages) { + const items = Array.isArray(payloadMessages) ? [...payloadMessages] : []; + items.sort((a, b) => Number(a?.timeMs || 0) - Number(b?.timeMs || 0)); + for (const item of items) { + const blobB64 = String(item?.blobB64 || '').trim(); + const messageKey = String(item?.messageKey || '').trim(); + if (!blobB64 || !messageKey) continue; + try { + const parsed = authService.parseSignedMessageBlob(blobB64); + let text = ''; + try { + const decrypted = await authService.decryptSignedMessageContent({ + parsed, + login: state.session.login, + storagePwd: state.session.storagePwdInMemory, + }); + text = String(decrypted?.text || ''); + } catch (error) { + text = `Нерасшифрованное сообщение (${error?.message || 'decrypt failed'})`; + } + addSignedMessageToChat({ + chatId, + messageKey, + baseKey: String(item?.baseKey || parsed?.baseKey || ''), + from: Number(parsed?.messageType || 0) === 2 ? 'out' : 'in', + text, + messageType: Number(parsed?.messageType || item?.messageType || 0), + unread: false, + readAtMs: Number(item?.readAtMs || 0), + rawBlobB64: blobB64, + revisionTimeMs: Number(item?.revisionTimeMs || parsed?.revisionTimeMs || 0), + }); + } catch (error) { + addAppLogEntry({ + level: 'warn', + source: 'dm-history', + message: 'Не удалось обработать сообщение из истории DM', + details: { chatId, messageKey, error: error?.message || 'unknown' }, + }); + } + } +} + function setChatKeyboardOpen(isOpen) { document.body.classList.toggle('chat-keyboard-open', !!isOpen); } @@ -703,6 +766,12 @@ export function render({ navigate, route }) { const isTextToSpeechReady = isTextToSpeechConfigured(state.entrySettings); const isKnownContact = (state.contacts || []).some((x) => String(x || '').toLowerCase() === String(chatId || '').toLowerCase()); const hasUnreadIncoming = getChatMessages(chatId).some((msg) => msg?.from === 'in' && msg?.unread); + let historyHasMore = true; + let historyLoading = false; + let historyNextBeforeTimeMs = 0; + let historyNextBeforeMessageKey = ''; + let historyBootstrapped = false; + let boundScrollContainer = null; const handleReadAloud = async (msg) => { if (!isTextToSpeechConfigured(state.entrySettings)) { @@ -1195,10 +1264,14 @@ export function render({ navigate, route }) { const handleOpenActions = (msg, event) => { const parsed = parseDmTechBlocks(String(msg?.text || '')); + const readState = resolveEffectiveReadState(getChatMessages(chatId), msg); openMessageActionsMenu({ anchorX: Number(event?.clientX || 0), anchorY: Number(event?.clientY || 0), messageText: parsed?.displayText || msg?.text || '', + infoText: (msg?.from === 'out' && readState.readAtMs > 0) + ? `Прочитано: ${formatMessageTime(readState.readAtMs)}` + : '', canReply: true, showReadAloud: isTextToSpeechReady, canEdit: msg?.from === 'out' && Number(msg?.messageType || 0) === 2 && !parsed?.callSummary, @@ -1227,6 +1300,55 @@ export function render({ navigate, route }) { }); }; + const loadHistoryPage = async ({ preserveScroll = false } = {}) => { + if (historyLoading) return; + if (historyBootstrapped && !historyHasMore) return; + historyLoading = true; + const scrollContainer = boundScrollContainer || log.closest('.screen-content') || wrap.parentElement || wrap; + const prevScrollTop = Number(scrollContainer?.scrollTop || 0); + const prevScrollHeight = Number(scrollContainer?.scrollHeight || 0); + try { + const payload = await authService.getDirectMessages({ + peerLogin: chatId, + limit: 50, + beforeTimeMs: historyBootstrapped ? historyNextBeforeTimeMs : 0, + beforeMessageKey: historyBootstrapped ? historyNextBeforeMessageKey : '', + }); + await mergeDirectMessagesPage(chatId, payload?.messages || []); + historyHasMore = Boolean(payload?.hasMore); + historyNextBeforeTimeMs = Number(payload?.nextBeforeTimeMs || 0); + historyNextBeforeMessageKey = String(payload?.nextBeforeMessageKey || '').trim(); + historyBootstrapped = true; + renderLog(log, chatId, { onOpenActions: handleOpenActions, markAsRead: false, scrollMode: preserveScroll ? 'preserve' : 'latest' }); + if (preserveScroll) { + window.requestAnimationFrame(() => { + const nextHeight = Number(scrollContainer?.scrollHeight || 0); + scrollContainer.scrollTop = Math.max(0, prevScrollTop + (nextHeight - prevScrollHeight)); + }); + } else { + window.requestAnimationFrame(() => scrollToLatestMessage(log)); + } + void sendReadReceiptsForVisible(chatId); + } catch (error) { + addAppLogEntry({ + level: 'warn', + source: 'dm-history', + message: 'Не удалось загрузить историю DM', + details: { chatId, error: error?.message || 'unknown' }, + }); + } finally { + historyLoading = false; + } + }; + + const handleHistoryScroll = () => { + const scrollContainer = boundScrollContainer || log.closest('.screen-content') || wrap.parentElement || wrap; + if (!scrollContainer) return; + if (Number(scrollContainer.scrollTop || 0) > 120) return; + if (!historyBootstrapped || !historyHasMore || historyLoading) return; + void loadHistoryPage({ preserveScroll: true }); + }; + editCancelBtn?.addEventListener('click', () => { if (activeEdit) { cancelEditMode({ restoreDraft: true }); @@ -1367,12 +1489,18 @@ export function render({ navigate, route }) { } }, 220); void sendReadReceiptsForVisible(chatId); + window.requestAnimationFrame(() => { + boundScrollContainer = log.closest('.screen-content') || wrap.parentElement || wrap; + boundScrollContainer?.addEventListener('scroll', handleHistoryScroll, { passive: true }); + void loadHistoryPage({ preserveScroll: false }); + }); screen.cleanup = () => { setChatKeyboardOpen(false); stopAllTwemojiAnimations(); window.removeEventListener('shine-chat-messages-updated', handleIncomingChatRefresh); window.visualViewport?.removeEventListener('resize', syncKeyboardUi); window.removeEventListener('resize', syncKeyboardUi); + boundScrollContainer?.removeEventListener('scroll', handleHistoryScroll); }; return screen; } diff --git a/shine-UI/js/services/auth-service.js b/shine-UI/js/services/auth-service.js index 029201ce..6bc03eeb 100644 --- a/shine-UI/js/services/auth-service.js +++ b/shine-UI/js/services/auth-service.js @@ -2579,6 +2579,23 @@ export class AuthService { }; } + async getDirectMessages({ peerLogin, limit = 50, beforeTimeMs = 0, beforeMessageKey = '' } = {}) { + const payload = { + peerLogin: String(peerLogin || '').trim(), + limit: Number(limit || 0), + }; + if (!payload.peerLogin) throw new Error('Не передан peerLogin'); + if (Number.isFinite(Number(beforeTimeMs)) && Number(beforeTimeMs) > 0) { + payload.beforeTimeMs = Math.trunc(Number(beforeTimeMs)); + } + if (String(beforeMessageKey || '').trim()) { + payload.beforeMessageKey = String(beforeMessageKey || '').trim(); + } + const response = await this.ws.request('GetDirectMessages', payload); + if (response.status !== 200) throw opError('GetDirectMessages', response); + return response.payload || {}; + } + async ackSessionDelivery(messageKey) { const response = await this.ws.request('AckSessionDelivery', { messageKey }); if (response.status !== 200) throw opError('AckSessionDelivery', response); diff --git a/shine-UI/js/state.js b/shine-UI/js/state.js index 9cc7242f..a26c0260 100644 --- a/shine-UI/js/state.js +++ b/shine-UI/js/state.js @@ -420,6 +420,7 @@ function persistMessageRecord(chatId, row) { unread: Boolean(row.unread), firstTick: Boolean(row.firstTick), secondTick: Boolean(row.secondTick), + readAtMs: Number(row.readAtMs || 0), readReceiptSent: Boolean(row.readReceiptSent), refBaseKey: String(row.refBaseKey || ''), ts: resolvedTs > 0 ? resolvedTs : Date.now(), @@ -455,6 +456,7 @@ export async function hydrateMessagesFromStore() { unread: Boolean(row.unread), firstTick: Boolean(row.firstTick), secondTick: Boolean(row.secondTick), + readAtMs: Number(row.readAtMs || 0), readReceiptSent: Boolean(row.readReceiptSent), refBaseKey: String(row.refBaseKey || ''), createdAtMs: Number(row.ts || 0), @@ -556,8 +558,9 @@ export function markOutgoingSent(tempId, { messageKey = '', baseKey = '' } = {}) }); } -export function markOutgoingReadByBaseKey(baseKey) { +export function markOutgoingReadByBaseKey(baseKey, readAtMs = 0) { if (!baseKey) return; + const normalizedReadAtMs = Number(readAtMs || 0); const keys = Object.keys(state.chats || {}); let matched = false; keys.forEach((chatId) => { @@ -567,6 +570,9 @@ export function markOutgoingReadByBaseKey(baseKey) { if (row.baseKey === baseKey) { matched = true; row.secondTick = true; + if (Number.isFinite(normalizedReadAtMs) && normalizedReadAtMs > 0) { + row.readAtMs = normalizedReadAtMs; + } persistMessageRecord(chatId, row); } }); @@ -574,7 +580,9 @@ export function markOutgoingReadByBaseKey(baseKey) { if (matched) { delete state.pendingOutgoingReadByBaseKey[baseKey]; } else { - state.pendingOutgoingReadByBaseKey[baseKey] = true; + state.pendingOutgoingReadByBaseKey[baseKey] = (Number.isFinite(normalizedReadAtMs) && normalizedReadAtMs > 0) + ? normalizedReadAtMs + : true; } return matched; } @@ -626,6 +634,7 @@ export function addSignedMessageToChat({ text = '', messageType = 1, unread = false, + readAtMs = 0, rawBlobB64 = '', refBaseKey = '', revisionTimeMs = 0, @@ -667,9 +676,14 @@ export function addSignedMessageToChat({ row.refBaseKey = String(refBaseKey || ''); row.firstTick = row.from === 'out'; row.secondTick = Boolean(existing?.secondTick); + row.readAtMs = Number(existing?.readAtMs || 0); row.readReceiptSent = Boolean(existing?.readReceiptSent); if (row.baseKey && row.from === 'out' && state.pendingOutgoingReadByBaseKey[row.baseKey]) { row.secondTick = true; + const pendingReadAtMs = Number(state.pendingOutgoingReadByBaseKey[row.baseKey] || 0); + if (Number.isFinite(pendingReadAtMs) && pendingReadAtMs > 0) { + row.readAtMs = pendingReadAtMs; + } delete state.pendingOutgoingReadByBaseKey[row.baseKey]; } if (row.baseKey && row.from === 'in' && state.pendingIncomingReadByBaseKey[row.baseKey]) { @@ -681,6 +695,13 @@ export function addSignedMessageToChat({ if (existingIndex < 0) { list.push(row); } + const nextReadAtMs = Number(readAtMs || 0); + if (Number.isFinite(nextReadAtMs) && nextReadAtMs > 0) { + row.readAtMs = nextReadAtMs; + if (row.from === 'out') { + row.secondTick = true; + } + } sortChatMessagesInPlace(normalizedChatId); persistMessageRecord(normalizedChatId, row); return true;