SHA256
830 lines
36 KiB
Java
830 lines
36 KiB
Java
package shine.db.dao;
|
|
|
|
import shine.db.DbController;
|
|
import shine.db.entities.SignedMessageEntry;
|
|
|
|
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 SignedMessagesDAO {
|
|
|
|
public enum ApplyStatus {
|
|
APPLIED,
|
|
DUPLICATE_OR_OLDER,
|
|
BLOCKED_BY_MESSAGE_TOMBSTONE,
|
|
BLOCKED_BY_CONVERSATION_TOMBSTONE;
|
|
|
|
public boolean applied() {
|
|
return this == APPLIED;
|
|
}
|
|
}
|
|
|
|
private static volatile SignedMessagesDAO instance;
|
|
private final DbController db = DbController.getInstance();
|
|
|
|
private SignedMessagesDAO() {}
|
|
|
|
public static SignedMessagesDAO getInstance() {
|
|
if (instance == null) {
|
|
synchronized (SignedMessagesDAO.class) {
|
|
if (instance == null) instance = new SignedMessagesDAO();
|
|
}
|
|
}
|
|
return instance;
|
|
}
|
|
|
|
public ApplyStatus insertIfAbsent(SignedMessageEntry e) throws Exception {
|
|
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 INTO signed_messages (
|
|
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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT DO NOTHING
|
|
""";
|
|
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);
|
|
DmDialogStateDAO.getInstance().refreshFromEntry(c, e);
|
|
}
|
|
return status;
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
public boolean insertPairBothOrNothing(SignedMessageEntry first, SignedMessageEntry second) throws Exception {
|
|
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);
|
|
DmDialogStateDAO.getInstance().refreshConversationPair(c, first.getFromLogin(), first.getToLogin());
|
|
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);
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
public ApplyStatus upsertContentPair(SignedMessageEntry incoming, SignedMessageEntry outgoing) throws Exception {
|
|
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;
|
|
}
|
|
|
|
upsertMessage(c, incoming);
|
|
upsertMessage(c, outgoing);
|
|
markMessageReadByReceipt(c, incoming);
|
|
markMessageReadByReceipt(c, outgoing);
|
|
resetDeliveryRows(c, incoming.getMessageKey());
|
|
resetDeliveryRows(c, outgoing.getMessageKey());
|
|
DmDialogStateDAO.getInstance().refreshConversationPair(c, incoming.getFromLogin(), incoming.getToLogin());
|
|
|
|
c.commit();
|
|
return ApplyStatus.APPLIED;
|
|
} catch (Exception ex) {
|
|
try { c.rollback(); } catch (Exception ignored) {}
|
|
throw ex;
|
|
} finally {
|
|
c.setAutoCommit(prevAutoCommit);
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
public ApplyStatus upsertIncomingCopy(SignedMessageEntry incoming) throws Exception {
|
|
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;
|
|
}
|
|
|
|
upsertMessage(c, incoming);
|
|
markMessageReadByReceipt(c, incoming);
|
|
resetDeliveryRows(c, incoming.getMessageKey());
|
|
DmDialogStateDAO.getInstance().refreshConversationPair(c, incoming.getFromLogin(), incoming.getToLogin());
|
|
c.commit();
|
|
return ApplyStatus.APPLIED;
|
|
} catch (Exception ex) {
|
|
try { c.rollback(); } catch (Exception ignored) {}
|
|
throw ex;
|
|
} finally {
|
|
c.setAutoCommit(prevAutoCommit);
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
public ApplyStatus applyDeleteMessage(SignedMessageEntry tombstone) throws Exception {
|
|
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());
|
|
DmDialogStateDAO.getInstance().refreshConversationPair(c, tombstone.getFromLogin(), tombstone.getToLogin());
|
|
|
|
c.commit();
|
|
return ApplyStatus.APPLIED;
|
|
} catch (Exception ex) {
|
|
try { c.rollback(); } catch (Exception ignored) {}
|
|
throw ex;
|
|
} finally {
|
|
c.setAutoCommit(prevAutoCommit);
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
public ApplyStatus applyDeleteConversation(SignedMessageEntry tombstone) throws Exception {
|
|
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());
|
|
DmDialogStateDAO.getInstance().refreshConversationPair(c, tombstone.getFromLogin(), tombstone.getToLogin());
|
|
|
|
c.commit();
|
|
return ApplyStatus.APPLIED;
|
|
} catch (Exception ex) {
|
|
try { c.rollback(); } catch (Exception ignored) {}
|
|
throw ex;
|
|
} finally {
|
|
c.setAutoCommit(prevAutoCommit);
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
public SignedMessageEntry getByMessageKey(String messageKey) throws Exception {
|
|
try (Connection c = db.getConnection()) {
|
|
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, read_at_ms
|
|
FROM %s
|
|
WHERE message_key = ?
|
|
""".formatted(messagesTable());
|
|
try (PreparedStatement ps = c.prepareStatement(sql)) {
|
|
ps.setString(1, messageKey);
|
|
try (ResultSet rs = ps.executeQuery()) {
|
|
if (!rs.next()) return null;
|
|
return mapRow(rs);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
public SignedMessageEntry 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 {
|
|
ensureDeliveryRows(messageKey, List.of(sessionId), nowMs);
|
|
}
|
|
|
|
public void ensureDeliveryRows(String messageKey, List<String> sessionIds, long nowMs) throws Exception {
|
|
if (sessionIds == null || sessionIds.isEmpty()) return;
|
|
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 (?, ?, 0, NULL, ?)
|
|
ON CONFLICT DO NOTHING
|
|
""";
|
|
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 {
|
|
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;
|
|
}
|
|
});
|
|
}
|
|
|
|
public List<SignedMessageEntry> listPendingForSession(String login, String sessionId) throws Exception {
|
|
return withBusyRetry(() -> {
|
|
try (Connection c = db.getConnection()) {
|
|
String fillSql = """
|
|
INSERT INTO signed_message_session_delivery (
|
|
message_key, session_id, delivered, delivered_at_ms, created_at_ms
|
|
)
|
|
SELECT m.message_key, ?, 0, NULL, ?
|
|
FROM %s m
|
|
WHERE (
|
|
(m.message_type IN (1, 3) AND LOWER(m.to_login) = LOWER(?))
|
|
OR (m.message_type IN (2, 4) AND LOWER(m.from_login) = LOWER(?))
|
|
OR (m.message_type IN (5, 6, 7, 8)
|
|
AND (LOWER(m.from_login) = LOWER(?) OR LOWER(m.to_login) = LOWER(?)))
|
|
)
|
|
ON CONFLICT DO NOTHING
|
|
""".formatted(messagesTable());
|
|
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 %s 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
|
|
""".formatted(messagesTable());
|
|
List<SignedMessageEntry> 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<SignedMessageEntry> listConversationPage(
|
|
String login,
|
|
String peerLogin,
|
|
long beforeTimeMs,
|
|
String beforeMessageKey,
|
|
int limit
|
|
) throws Exception {
|
|
try (Connection c = db.getConnection()) {
|
|
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, read_at_ms
|
|
FROM %s
|
|
WHERE LOWER(target_login) = LOWER(?)
|
|
AND message_type IN (1, 2)
|
|
AND (
|
|
(LOWER(from_login) = LOWER(?) AND LOWER(to_login) = LOWER(?))
|
|
OR (LOWER(from_login) = LOWER(?) AND LOWER(to_login) = LOWER(?))
|
|
)
|
|
AND (
|
|
? <= 0
|
|
OR time_ms < ?
|
|
OR (time_ms = ? AND (? = '' OR message_key < ?))
|
|
)
|
|
ORDER BY time_ms DESC, message_key DESC
|
|
LIMIT ?
|
|
""".formatted(messagesTable());
|
|
List<SignedMessageEntry> out = new ArrayList<>();
|
|
try (PreparedStatement ps = c.prepareStatement(sql)) {
|
|
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));
|
|
}
|
|
}
|
|
return out;
|
|
}
|
|
}
|
|
|
|
public SyncBatch listSyncBatch(
|
|
String ownerLogin,
|
|
long afterStoredAtMs,
|
|
String afterMessageKey,
|
|
int limit,
|
|
int maxBytes
|
|
) throws Exception {
|
|
int safeLimit = Math.max(1, Math.min(limit, 500));
|
|
int sqlLimit = safeLimit + 1;
|
|
String safeAfterMessageKey = afterMessageKey == null ? "" : afterMessageKey;
|
|
try (Connection c = db.getConnection()) {
|
|
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, read_at_ms
|
|
FROM %s
|
|
WHERE (
|
|
LOWER(target_login) = LOWER(?)
|
|
OR (
|
|
message_type IN (5, 6, 7, 8)
|
|
AND (LOWER(from_login) = LOWER(?) OR LOWER(to_login) = LOWER(?))
|
|
)
|
|
)
|
|
AND (
|
|
created_at_ms > ?
|
|
OR (created_at_ms = ? AND (? = '' OR message_key > ?))
|
|
)
|
|
ORDER BY created_at_ms ASC, message_key ASC
|
|
LIMIT ?
|
|
""".formatted(messagesTable());
|
|
List<SignedMessageEntry> out = new ArrayList<>();
|
|
boolean hasMore = false;
|
|
int usedBytes = 0;
|
|
try (PreparedStatement ps = c.prepareStatement(sql)) {
|
|
ps.setString(1, ownerLogin);
|
|
ps.setString(2, ownerLogin);
|
|
ps.setString(3, ownerLogin);
|
|
ps.setLong(4, Math.max(0L, afterStoredAtMs));
|
|
ps.setLong(5, Math.max(0L, afterStoredAtMs));
|
|
ps.setString(6, safeAfterMessageKey);
|
|
ps.setString(7, safeAfterMessageKey);
|
|
ps.setInt(8, sqlLimit);
|
|
try (ResultSet rs = ps.executeQuery()) {
|
|
while (rs.next()) {
|
|
SignedMessageEntry row = mapRow(rs);
|
|
byte[] raw = row.getRawBlock();
|
|
int rowBytes = raw == null ? 0 : raw.length;
|
|
if (out.size() >= safeLimit) {
|
|
hasMore = true;
|
|
break;
|
|
}
|
|
if (!out.isEmpty() && maxBytes > 0 && usedBytes + rowBytes > maxBytes) {
|
|
hasMore = true;
|
|
break;
|
|
}
|
|
out.add(row);
|
|
usedBytes += rowBytes;
|
|
}
|
|
}
|
|
}
|
|
return new SyncBatch(out, hasMore, usedBytes);
|
|
}
|
|
}
|
|
|
|
private void upsertMessage(Connection c, SignedMessageEntry e) throws SQLException {
|
|
String sql = """
|
|
INSERT INTO %s (
|
|
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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(message_key) DO UPDATE SET
|
|
base_key = excluded.base_key,
|
|
target_login = excluded.target_login,
|
|
from_login = excluded.from_login,
|
|
to_login = excluded.to_login,
|
|
time_ms = excluded.time_ms,
|
|
nonce = excluded.nonce,
|
|
message_type = excluded.message_type,
|
|
revision_time_ms = excluded.revision_time_ms,
|
|
reencrypted_at_ms = excluded.reencrypted_at_ms,
|
|
raw_block = excluded.raw_block,
|
|
created_at_ms = excluded.created_at_ms,
|
|
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,
|
|
read_at_ms = COALESCE(%s.read_at_ms, excluded.read_at_ms)
|
|
""".formatted(messagesTable(), messagesTable());
|
|
try (PreparedStatement ps = c.prepareStatement(sql)) {
|
|
bindSignedMessage(ps, e);
|
|
ps.executeUpdate();
|
|
}
|
|
}
|
|
|
|
private void markMessageReadByReceipt(Connection c, SignedMessageEntry 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 %s
|
|
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)
|
|
""".formatted(messagesTable()))) {
|
|
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
|
|
FROM %s
|
|
WHERE message_key = ?
|
|
LIMIT 1
|
|
""".formatted(messagesTable());
|
|
try (PreparedStatement ps = c.prepareStatement(sql)) {
|
|
ps.setString(1, messageKey);
|
|
try (ResultSet rs = ps.executeQuery()) {
|
|
if (!rs.next()) return null;
|
|
return new RevisionMarker(rs.getLong(1), rs.getLong(2));
|
|
}
|
|
}
|
|
}
|
|
|
|
private RevisionMarker getCurrentContentMarker(Connection c, String baseKey) throws SQLException {
|
|
String sql = """
|
|
SELECT revision_time_ms, reencrypted_at_ms
|
|
FROM %s
|
|
WHERE base_key = ?
|
|
AND message_type IN (1, 2)
|
|
ORDER BY revision_time_ms DESC, reencrypted_at_ms DESC
|
|
LIMIT 1
|
|
""".formatted(messagesTable());
|
|
try (PreparedStatement ps = c.prepareStatement(sql)) {
|
|
ps.setString(1, baseKey);
|
|
try (ResultSet rs = ps.executeQuery()) {
|
|
if (!rs.next()) return null;
|
|
return new RevisionMarker(rs.getLong(1), rs.getLong(2));
|
|
}
|
|
}
|
|
}
|
|
|
|
private boolean hasMessageDeleteTombstone(Connection c, String baseKey) throws SQLException {
|
|
String sql = """
|
|
SELECT 1
|
|
FROM %s
|
|
WHERE base_key = ?
|
|
AND message_type IN (5, 6)
|
|
LIMIT 1
|
|
""".formatted(messagesTable());
|
|
try (PreparedStatement ps = c.prepareStatement(sql)) {
|
|
ps.setString(1, baseKey);
|
|
try (ResultSet rs = ps.executeQuery()) {
|
|
return rs.next();
|
|
}
|
|
}
|
|
}
|
|
|
|
private boolean isBlockedByConversationDelete(Connection c, String fromLogin, String toLogin, long timeMs) throws SQLException {
|
|
Long boundary = getLatestConversationDeleteBoundary(c, fromLogin, toLogin);
|
|
return boundary != null && timeMs < boundary;
|
|
}
|
|
|
|
private Long getLatestConversationDeleteBoundary(Connection c, String fromLogin, String toLogin) throws SQLException {
|
|
String sql = """
|
|
SELECT MAX(time_ms)
|
|
FROM %s
|
|
WHERE message_type IN (7, 8)
|
|
AND (
|
|
(LOWER(from_login) = LOWER(?) AND LOWER(to_login) = LOWER(?))
|
|
OR (LOWER(from_login) = LOWER(?) AND LOWER(to_login) = LOWER(?))
|
|
)
|
|
""".formatted(messagesTable());
|
|
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;
|
|
long value = rs.getLong(1);
|
|
return rs.wasNull() ? null : value;
|
|
}
|
|
}
|
|
}
|
|
|
|
private SignedMessageEntry 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, read_at_ms
|
|
FROM %s
|
|
WHERE message_type IN (7, 8)
|
|
AND (
|
|
(LOWER(from_login) = LOWER(?) AND LOWER(to_login) = LOWER(?))
|
|
OR (LOWER(from_login) = LOWER(?) AND LOWER(to_login) = LOWER(?))
|
|
)
|
|
ORDER BY time_ms DESC
|
|
LIMIT 1
|
|
""".formatted(messagesTable());
|
|
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
|
|
FROM %s
|
|
WHERE (base_key = ? AND message_type IN (1, 2))
|
|
OR (receipt_ref_base_key = ? AND message_type IN (3, 4))
|
|
""".formatted(messagesTable()), baseKey, baseKey);
|
|
|
|
try (PreparedStatement ps = c.prepareStatement("""
|
|
DELETE FROM %s
|
|
WHERE (base_key = ? AND message_type IN (1, 2))
|
|
OR (receipt_ref_base_key = ? AND message_type IN (3, 4))
|
|
""".formatted(messagesTable()))) {
|
|
ps.setString(1, baseKey);
|
|
ps.setString(2, baseKey);
|
|
ps.executeUpdate();
|
|
}
|
|
}
|
|
|
|
private void deleteConversationHistoryBefore(Connection c, String fromLogin, String toLogin, long boundaryTimeMs) throws SQLException {
|
|
deleteDeliveryRowsByMessageSelection(c, """
|
|
SELECT message_key
|
|
FROM %s
|
|
WHERE time_ms < ?
|
|
AND (
|
|
(LOWER(from_login) = LOWER(?) AND LOWER(to_login) = LOWER(?))
|
|
OR (LOWER(from_login) = LOWER(?) AND LOWER(to_login) = LOWER(?))
|
|
)
|
|
""".formatted(messagesTable()), boundaryTimeMs, fromLogin, toLogin, toLogin, fromLogin);
|
|
|
|
try (PreparedStatement ps = c.prepareStatement("""
|
|
DELETE FROM %s
|
|
WHERE time_ms < ?
|
|
AND (
|
|
(LOWER(from_login) = LOWER(?) AND LOWER(to_login) = LOWER(?))
|
|
OR (LOWER(from_login) = LOWER(?) AND LOWER(to_login) = LOWER(?))
|
|
)
|
|
""".formatted(messagesTable()))) {
|
|
ps.setLong(1, boundaryTimeMs);
|
|
ps.setString(2, fromLogin);
|
|
ps.setString(3, toLogin);
|
|
ps.setString(4, toLogin);
|
|
ps.setString(5, fromLogin);
|
|
ps.executeUpdate();
|
|
}
|
|
}
|
|
|
|
private void deleteDeliveryRowsByMessageSelection(Connection c, String messageKeySelectSql, Object... bindValues) throws SQLException {
|
|
String sql = "DELETE FROM signed_message_session_delivery WHERE message_key IN (" + messageKeySelectSql + ")";
|
|
try (PreparedStatement ps = c.prepareStatement(sql)) {
|
|
bindObjects(ps, bindValues);
|
|
ps.executeUpdate();
|
|
}
|
|
}
|
|
|
|
private void resetDeliveryRows(Connection c, String messageKey) throws SQLException {
|
|
try (PreparedStatement ps = c.prepareStatement("""
|
|
UPDATE signed_message_session_delivery
|
|
SET delivered = 0, delivered_at_ms = NULL
|
|
WHERE message_key = ?
|
|
""")) {
|
|
ps.setString(1, messageKey);
|
|
ps.executeUpdate();
|
|
}
|
|
}
|
|
|
|
private int insertStrict(Connection c, SignedMessageEntry e) throws SQLException {
|
|
String sql = """
|
|
INSERT INTO %s (
|
|
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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""".formatted(messagesTable());
|
|
try (PreparedStatement ps = c.prepareStatement(sql)) {
|
|
bindSignedMessage(ps, e);
|
|
return ps.executeUpdate();
|
|
}
|
|
}
|
|
|
|
private void bindSignedMessage(PreparedStatement ps, SignedMessageEntry e) throws SQLException {
|
|
ps.setString(1, e.getMessageKey());
|
|
ps.setString(2, e.getBaseKey());
|
|
ps.setString(3, e.getTargetLogin());
|
|
ps.setString(4, e.getFromLogin());
|
|
ps.setString(5, e.getToLogin());
|
|
ps.setLong(6, e.getTimeMs());
|
|
ps.setLong(7, e.getNonce());
|
|
ps.setInt(8, e.getMessageType());
|
|
ps.setLong(9, e.getRevisionTimeMs());
|
|
ps.setLong(10, e.getReencryptedAtMs());
|
|
ps.setBytes(11, e.getRawBlock());
|
|
ps.setLong(12, e.getCreatedAtMs());
|
|
ps.setString(13, e.getSourceApi());
|
|
ps.setString(14, e.getOriginSessionId());
|
|
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 {
|
|
for (int i = 0; i < bindValues.length; i++) {
|
|
Object value = bindValues[i];
|
|
int param = i + 1;
|
|
if (value instanceof String s) {
|
|
ps.setString(param, s);
|
|
} else if (value instanceof Long l) {
|
|
ps.setLong(param, l);
|
|
} else if (value instanceof Integer n) {
|
|
ps.setInt(param, n);
|
|
} else {
|
|
ps.setObject(param, value);
|
|
}
|
|
}
|
|
}
|
|
|
|
private boolean isConstraintViolation(SQLException ex) {
|
|
String msg = String.valueOf(ex.getMessage()).toLowerCase();
|
|
return msg.contains("constraint") || msg.contains("unique") || msg.contains("primary key");
|
|
}
|
|
|
|
private <T> T withBusyRetry(SqlWork<T> work) throws Exception {
|
|
return work.run();
|
|
}
|
|
|
|
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 String messagesTable() {
|
|
return "signed_messages";
|
|
}
|
|
|
|
private SignedMessageEntry mapRow(ResultSet rs) throws Exception {
|
|
SignedMessageEntry e = new SignedMessageEntry();
|
|
e.setMessageKey(rs.getString("message_key"));
|
|
e.setBaseKey(rs.getString("base_key"));
|
|
e.setTargetLogin(rs.getString("target_login"));
|
|
e.setFromLogin(rs.getString("from_login"));
|
|
e.setToLogin(rs.getString("to_login"));
|
|
e.setTimeMs(rs.getLong("time_ms"));
|
|
e.setNonce(rs.getLong("nonce"));
|
|
e.setMessageType(rs.getInt("message_type"));
|
|
e.setRevisionTimeMs(rs.getLong("revision_time_ms"));
|
|
e.setReencryptedAtMs(rs.getLong("reencrypted_at_ms"));
|
|
e.setRawBlock(rs.getBytes("raw_block"));
|
|
e.setCreatedAtMs(rs.getLong("created_at_ms"));
|
|
e.setSourceApi(rs.getString("source_api"));
|
|
e.setOriginSessionId(rs.getString("origin_session_id"));
|
|
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;
|
|
}
|
|
|
|
private record RevisionMarker(long revisionTimeMs, long reencryptedAtMs) {
|
|
private static RevisionMarker of(SignedMessageEntry entry) {
|
|
return new RevisionMarker(entry.getRevisionTimeMs(), entry.getReencryptedAtMs());
|
|
}
|
|
}
|
|
|
|
public record SyncBatch(List<SignedMessageEntry> items, boolean hasMore, int rawBytes) {}
|
|
|
|
@FunctionalInterface
|
|
private interface SqlWork<T> {
|
|
T run() throws Exception;
|
|
}
|
|
}
|