aidar/SHiNE-server
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Довести DM v1 до соответствия коду и документации

Browse Source
This commit is contained in:
AidarKC
2026-07-08 17:28:39 +04:00
parent faada9dcdb
commit 98a2631590
19 changed files with 283 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
SHiNE-server/shine-server-db/src/main/java/shine/db/dao/SignedMessagesV2DAO.java
+98 -32
View File
@@ -11,6 +11,17 @@ import java.util.ArrayList;
import java.util.List;
public final class SignedMessagesV2DAO {
public enum ApplyStatus {
APPLIED,
DUPLICATE_OR_OLDER,
BLOCKED_BY_MESSAGE_TOMBSTONE,
BLOCKED_BY_CONVERSATION_TOMBSTONE;
public boolean applied() {
return this == APPLIED;
}
}
private static volatile SignedMessagesV2DAO instance;
private final SqliteDbController db = SqliteDbController.getInstance();
@@ -25,8 +36,11 @@ public final class SignedMessagesV2DAO {
return instance;
}
public boolean insertIfAbsent(SignedMessageV2Entry e) throws Exception {
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;
}
String sql = """
INSERT OR IGNORE INTO signed_messages_v2 (
message_key, base_key, target_login, from_login, to_login,
@@ -37,7 +51,7 @@ public final class SignedMessagesV2DAO {
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
bindSignedMessage(ps, e);
return ps.executeUpdate() > 0;
return ps.executeUpdate() > 0 ? ApplyStatus.APPLIED : ApplyStatus.DUPLICATE_OR_OLDER;
}
}
}
@@ -67,25 +81,25 @@ public final class SignedMessagesV2DAO {
}
}
public boolean upsertContentPair(SignedMessageV2Entry incoming, SignedMessageV2Entry outgoing) throws Exception {
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 false;
return ApplyStatus.BLOCKED_BY_CONVERSATION_TOMBSTONE;
}
if (hasMessageDeleteTombstone(c, incoming.getBaseKey())) {
c.rollback();
return false;
return ApplyStatus.BLOCKED_BY_MESSAGE_TOMBSTONE;
}
Long currentRevision = getCurrentContentRevision(c, incoming.getBaseKey());
long nextRevision = incoming.getRevisionTimeMs();
if (currentRevision != null && nextRevision <= currentRevision) {
RevisionMarker currentMarker = getCurrentContentMarker(c, incoming.getBaseKey());
RevisionMarker nextMarker = RevisionMarker.of(incoming);
if (currentMarker != null && compareMarkers(nextMarker, currentMarker) <= 0) {
c.rollback();
return false;
return ApplyStatus.DUPLICATE_OR_OLDER;
}
upsertMessage(c, incoming);
@@ -94,7 +108,7 @@ public final class SignedMessagesV2DAO {
resetDeliveryRows(c, outgoing.getMessageKey());
c.commit();
return true;
return ApplyStatus.APPLIED;
} catch (Exception ex) {
try { c.rollback(); } catch (Exception ignored) {}
throw ex;
@@ -104,31 +118,31 @@ public final class SignedMessagesV2DAO {
}
}
public boolean upsertIncomingCopy(SignedMessageV2Entry incoming) throws Exception {
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 false;
return ApplyStatus.BLOCKED_BY_CONVERSATION_TOMBSTONE;
}
if (hasMessageDeleteTombstone(c, incoming.getBaseKey())) {
c.rollback();
return false;
return ApplyStatus.BLOCKED_BY_MESSAGE_TOMBSTONE;
}
Long currentRevision = getRevisionTimeMs(c, incoming.getMessageKey());
long nextRevision = incoming.getRevisionTimeMs();
if (currentRevision != null && nextRevision <= currentRevision) {
RevisionMarker currentMarker = getRevisionMarkerByMessageKey(c, incoming.getMessageKey());
RevisionMarker nextMarker = RevisionMarker.of(incoming);
if (currentMarker != null && compareMarkers(nextMarker, currentMarker) <= 0) {
c.rollback();
return false;
return ApplyStatus.DUPLICATE_OR_OLDER;
}
upsertMessage(c, incoming);
resetDeliveryRows(c, incoming.getMessageKey());
c.commit();
return true;
return ApplyStatus.APPLIED;
} catch (Exception ex) {
try { c.rollback(); } catch (Exception ignored) {}
throw ex;
@@ -138,18 +152,18 @@ public final class SignedMessagesV2DAO {
}
}
public boolean applyDeleteMessage(SignedMessageV2Entry tombstone) throws Exception {
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 false;
return ApplyStatus.BLOCKED_BY_CONVERSATION_TOMBSTONE;
}
if (hasMessageDeleteTombstone(c, tombstone.getBaseKey())) {
c.rollback();
return false;
return ApplyStatus.DUPLICATE_OR_OLDER;
}
deleteMessageContentAndReceipts(c, tombstone.getBaseKey());
@@ -157,7 +171,7 @@ public final class SignedMessagesV2DAO {
resetDeliveryRows(c, tombstone.getMessageKey());
c.commit();
return true;
return ApplyStatus.APPLIED;
} catch (Exception ex) {
try { c.rollback(); } catch (Exception ignored) {}
throw ex;
@@ -167,7 +181,7 @@ public final class SignedMessagesV2DAO {
}
}
public boolean applyDeleteConversation(SignedMessageV2Entry tombstone) throws Exception {
public ApplyStatus applyDeleteConversation(SignedMessageV2Entry tombstone) throws Exception {
try (Connection c = db.getConnection()) {
boolean prevAutoCommit = c.getAutoCommit();
c.setAutoCommit(false);
@@ -175,7 +189,7 @@ public final class SignedMessagesV2DAO {
Long currentBoundary = getLatestConversationDeleteBoundary(c, tombstone.getFromLogin(), tombstone.getToLogin());
if (currentBoundary != null && tombstone.getTimeMs() <= currentBoundary) {
c.rollback();
return false;
return ApplyStatus.DUPLICATE_OR_OLDER;
}
deleteConversationHistoryBefore(c, tombstone.getFromLogin(), tombstone.getToLogin(), tombstone.getTimeMs());
@@ -183,7 +197,7 @@ public final class SignedMessagesV2DAO {
resetDeliveryRows(c, tombstone.getMessageKey());
c.commit();
return true;
return ApplyStatus.APPLIED;
} catch (Exception ex) {
try { c.rollback(); } catch (Exception ignored) {}
throw ex;
@@ -214,6 +228,12 @@ public final class SignedMessagesV2DAO {
}
}
public SignedMessageV2Entry getLatestConversationDelete(String fromLogin, String toLogin) throws Exception {
try (Connection c = db.getConnection()) {
return getLatestConversationDelete(c, fromLogin, toLogin);
}
}
public void ensureDeliveryRow(String messageKey, String sessionId, long nowMs) throws Exception {
try (Connection c = db.getConnection()) {
String sql = """
@@ -338,30 +358,36 @@ public final class SignedMessagesV2DAO {
}
}
private Long getRevisionTimeMs(Connection c, String messageKey) throws SQLException {
String sql = "SELECT revision_time_ms FROM signed_messages_v2 WHERE message_key = ? LIMIT 1";
private RevisionMarker getRevisionMarkerByMessageKey(Connection c, String messageKey) throws SQLException {
String sql = """
SELECT revision_time_ms, reencrypted_at_ms
FROM signed_messages_v2
WHERE message_key = ?
LIMIT 1
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setString(1, messageKey);
try (ResultSet rs = ps.executeQuery()) {
if (!rs.next()) return null;
return rs.getLong(1);
return new RevisionMarker(rs.getLong(1), rs.getLong(2));
}
}
}
private Long getCurrentContentRevision(Connection c, String baseKey) throws SQLException {
private RevisionMarker getCurrentContentMarker(Connection c, String baseKey) throws SQLException {
String sql = """
SELECT MAX(revision_time_ms)
SELECT revision_time_ms, reencrypted_at_ms
FROM signed_messages_v2
WHERE base_key = ?
AND message_type IN (1, 2)
ORDER BY revision_time_ms DESC, reencrypted_at_ms DESC
LIMIT 1
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setString(1, baseKey);
try (ResultSet rs = ps.executeQuery()) {
if (!rs.next()) return null;
long value = rs.getLong(1);
return rs.wasNull() ? null : value;
return new RevisionMarker(rs.getLong(1), rs.getLong(2));
}
}
}
@@ -410,6 +436,34 @@ public final class SignedMessagesV2DAO {
}
}
private SignedMessageV2Entry getLatestConversationDelete(Connection c, String fromLogin, String toLogin) throws Exception {
String sql = """
SELECT
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
FROM signed_messages_v2
WHERE message_type IN (7, 8)
AND (
(from_login = ? COLLATE NOCASE AND to_login = ? COLLATE NOCASE)
OR (from_login = ? COLLATE NOCASE AND to_login = ? COLLATE NOCASE)
)
ORDER BY time_ms DESC
LIMIT 1
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setString(1, fromLogin);
ps.setString(2, toLogin);
ps.setString(3, toLogin);
ps.setString(4, fromLogin);
try (ResultSet rs = ps.executeQuery()) {
if (!rs.next()) return null;
return mapRow(rs);
}
}
}
private void deleteMessageContentAndReceipts(Connection c, String baseKey) throws SQLException {
deleteDeliveryRowsByMessageSelection(c, """
SELECT message_key
@@ -532,6 +586,12 @@ public final class SignedMessagesV2DAO {
return msg.contains("constraint") || msg.contains("unique") || msg.contains("primary key");
}
private int compareMarkers(RevisionMarker left, RevisionMarker right) {
int revisionCompare = Long.compare(left.revisionTimeMs, right.revisionTimeMs);
if (revisionCompare != 0) return revisionCompare;
return Long.compare(left.reencryptedAtMs, right.reencryptedAtMs);
}
private SignedMessageV2Entry mapRow(ResultSet rs) throws Exception {
SignedMessageV2Entry e = new SignedMessageV2Entry();
e.setMessageKey(rs.getString("message_key"));
@@ -553,4 +613,10 @@ public final class SignedMessagesV2DAO {
e.setReceiptRefType(rs.wasNull() ? null : maybeRefType);
return e;
}
private record RevisionMarker(long revisionTimeMs, long reencryptedAtMs) {
private static RevisionMarker of(SignedMessageV2Entry entry) {
return new RevisionMarker(entry.getRevisionTimeMs(), entry.getReencryptedAtMs());
}
}
}