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

НЕ ПРОВЕРЕНО: DM-вложения, upload файлов и ревизии личных сообщений

Browse Source
This commit is contained in:
AidarKC
2026-06-18 11:46:58 +04:00
parent 2225c2d173
commit 92fd315505
25 changed files with 1910 additions and 516 deletions
Download Patch File Download Diff File Expand all files Collapse all files
SHiNE-server/shine-server-db/src/main/java/shine/db/DatabaseInitializer.java
+31
View File
@@ -618,6 +618,7 @@ public final class DatabaseInitializer {
time_ms INTEGER NOT NULL,
nonce INTEGER NOT NULL,
message_type INTEGER NOT NULL,
revision_time_ms INTEGER NOT NULL DEFAULT 0,
raw_block BLOB NOT NULL,
created_at_ms INTEGER NOT NULL,
source_api TEXT NOT NULL,
@@ -639,6 +640,36 @@ public final class DatabaseInitializer {
ON signed_messages_v2 (base_key, message_type);
""");
st.executeUpdate("""
CREATE TABLE IF NOT EXISTS dm_files (
file_hash_sha256 BLOB NOT NULL PRIMARY KEY,
file_size INTEGER NOT NULL,
ref_count INTEGER NOT NULL DEFAULT 0
);
""");
st.executeUpdate("""
CREATE TABLE IF NOT EXISTS dm_message_file_links (
message_key TEXT NOT NULL,
login TEXT NOT NULL,
file_hash_sha256 BLOB NOT NULL,
PRIMARY KEY (message_key, login, file_hash_sha256),
FOREIGN KEY (message_key) REFERENCES signed_messages_v2(message_key),
FOREIGN KEY (login) REFERENCES solana_users(login),
FOREIGN KEY (file_hash_sha256) REFERENCES dm_files(file_hash_sha256)
);
""");
st.executeUpdate("""
CREATE INDEX IF NOT EXISTS idx_dm_message_file_links_login
ON dm_message_file_links (login, file_hash_sha256);
""");
st.executeUpdate("""
CREATE INDEX IF NOT EXISTS idx_dm_message_file_links_message
ON dm_message_file_links (message_key);
""");
st.executeUpdate("""
CREATE UNIQUE INDEX IF NOT EXISTS uq_signed_messages_v2_receipt_incoming
ON signed_messages_v2 (target_login, receipt_ref_base_key)
SHiNE-server/shine-server-db/src/main/java/shine/db/SqliteDbController.java
+61 -1
View File
@@ -14,7 +14,7 @@ import java.sql.Statement;
public final class SqliteDbController {
private static volatile SqliteDbController instance;
private static final int LATEST_SCHEMA_VERSION = 5;
private static final int LATEST_SCHEMA_VERSION = 6;
private final String jdbcUrl;
@@ -88,6 +88,7 @@ public final class SqliteDbController {
case 3 -> migrateToV3();
case 4 -> migrateToV4();
case 5 -> migrateToV5();
case 6 -> migrateToV6();
default -> throw new RuntimeException("Unknown DB migration target version: " + targetVersion);
}
}
@@ -209,6 +210,26 @@ public final class SqliteDbController {
}
}
private void migrateToV6() {
try (Connection c = DriverManager.getConnection(jdbcUrl);
Statement st = c.createStatement()) {
c.setAutoCommit(false);
try {
ensureSignedMessagesRevisionColumn(c, st);
ensureDmFileTables(st);
setSchemaVersion(c, 6);
c.commit();
} catch (Exception e) {
try { c.rollback(); } catch (Exception ignored) {}
throw new RuntimeException("DB migration to v6 failed", e);
} finally {
try { c.setAutoCommit(true); } catch (Exception ignored) {}
}
} catch (SQLException e) {
throw new RuntimeException("DB migration to v6 failed", e);
}
}
private static void ensureChat200StateTables(Statement st) throws SQLException {
st.executeUpdate("""
CREATE TABLE IF NOT EXISTS chat200_state (
@@ -329,6 +350,45 @@ public final class SqliteDbController {
""");
}
private static void ensureSignedMessagesRevisionColumn(Connection c, Statement st) throws SQLException {
if (!tableExists(c, "signed_messages_v2")) return;
if (!columnExists(c, "signed_messages_v2", "revision_time_ms")) {
st.executeUpdate("ALTER TABLE signed_messages_v2 ADD COLUMN revision_time_ms INTEGER NOT NULL DEFAULT 0");
}
}
private static void ensureDmFileTables(Statement st) throws SQLException {
st.executeUpdate("""
CREATE TABLE IF NOT EXISTS dm_files (
file_hash_sha256 BLOB NOT NULL PRIMARY KEY,
file_size INTEGER NOT NULL,
ref_count INTEGER NOT NULL DEFAULT 0
);
""");
st.executeUpdate("""
CREATE TABLE IF NOT EXISTS dm_message_file_links (
message_key TEXT NOT NULL,
login TEXT NOT NULL,
file_hash_sha256 BLOB NOT NULL,
PRIMARY KEY (message_key, login, file_hash_sha256),
FOREIGN KEY (message_key) REFERENCES signed_messages_v2(message_key),
FOREIGN KEY (login) REFERENCES solana_users(login),
FOREIGN KEY (file_hash_sha256) REFERENCES dm_files(file_hash_sha256)
);
""");
st.executeUpdate("""
CREATE INDEX IF NOT EXISTS idx_dm_message_file_links_login
ON dm_message_file_links (login, file_hash_sha256);
""");
st.executeUpdate("""
CREATE INDEX IF NOT EXISTS idx_dm_message_file_links_message
ON dm_message_file_links (message_key);
""");
}
private static boolean columnExists(Connection c, String tableName, String columnName) throws SQLException {
try (Statement probe = c.createStatement();
ResultSet rs = probe.executeQuery("PRAGMA table_info(" + tableName + ")")) {
SHiNE-server/shine-server-db/src/main/java/shine/db/dao/SignedMessagesV2DAO.java
+300 -51
View File
@@ -1,6 +1,7 @@
package shine.db.dao;
import shine.db.SqliteDbController;
import shine.db.entities.DmFileRef;
import shine.db.entities.SignedMessageV2Entry;
import java.sql.Connection;
@@ -8,7 +9,10 @@ import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
public final class SignedMessagesV2DAO {
private static volatile SignedMessagesV2DAO instance;
@@ -30,35 +34,19 @@ public final class SignedMessagesV2DAO {
String sql = """
INSERT OR IGNORE INTO signed_messages_v2 (
message_key, base_key, target_login, from_login, to_login,
time_ms, nonce, message_type, raw_block, created_at_ms,
time_ms, nonce, message_type, revision_time_ms, raw_block, created_at_ms,
source_api, origin_session_id, receipt_ref_base_key, receipt_ref_type
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
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.setBytes(9, e.getRawBlock());
ps.setLong(10, e.getCreatedAtMs());
ps.setString(11, e.getSourceApi());
ps.setString(12, e.getOriginSessionId());
ps.setString(13, e.getReceiptRefBaseKey());
if (e.getReceiptRefType() == null) ps.setObject(14, null);
else ps.setInt(14, e.getReceiptRefType());
bindSignedMessage(ps, e);
return ps.executeUpdate() > 0;
}
}
}
/**
* Атомарная вставка пары блоков: либо вставляются оба, либо не вставляется ни один.
* Возвращает true только если обе записи добавлены в БД.
* Если хотя бы одна запись уже существует (или конфликтует по уникальности), возвращает false.
* Атомарная вставка пары блоков legacy/read-receipt: либо вставляются оба, либо не вставляется ни один.
*/
public boolean insertPairBothOrNothing(SignedMessageV2Entry first, SignedMessageV2Entry second) throws Exception {
try (Connection c = db.getConnection()) {
@@ -85,37 +73,97 @@ public final class SignedMessagesV2DAO {
}
}
private int insertStrict(Connection c, SignedMessageV2Entry e) throws SQLException {
String sql = """
INSERT INTO signed_messages_v2 (
message_key, base_key, target_login, from_login, to_login,
time_ms, nonce, message_type, raw_block, created_at_ms,
source_api, origin_session_id, receipt_ref_base_key, receipt_ref_type
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
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.setBytes(9, e.getRawBlock());
ps.setLong(10, e.getCreatedAtMs());
ps.setString(11, e.getSourceApi());
ps.setString(12, e.getOriginSessionId());
ps.setString(13, e.getReceiptRefBaseKey());
if (e.getReceiptRefType() == null) ps.setObject(14, null);
else ps.setInt(14, e.getReceiptRefType());
return ps.executeUpdate();
/**
* Атомарный upsert пары контентных DM с полной заменой файловых связей.
* Возвращает true, только если ревизия применена. Более старая или идентичная ревизия игнорируется.
*/
public boolean upsertContentPairReplaceFiles(
SignedMessageV2Entry incoming,
List<DmFileRef> incomingFiles,
SignedMessageV2Entry outgoing,
List<DmFileRef> outgoingFiles
) throws Exception {
try (Connection c = db.getConnection()) {
boolean prevAutoCommit = c.getAutoCommit();
c.setAutoCommit(false);
try {
if (!allFilesExist(c, incomingFiles) || !allFilesExist(c, outgoingFiles)) {
c.rollback();
return false;
}
Long currentIncomingRevision = getRevisionTimeMs(c, incoming.getMessageKey());
Long currentOutgoingRevision = getRevisionTimeMs(c, outgoing.getMessageKey());
long currentRevision = Math.max(
currentIncomingRevision != null ? currentIncomingRevision : Long.MIN_VALUE,
currentOutgoingRevision != null ? currentOutgoingRevision : Long.MIN_VALUE
);
long nextRevision = incoming.getRevisionTimeMs();
if (currentRevision != Long.MIN_VALUE && nextRevision < currentRevision) {
c.rollback();
return false;
}
if (currentRevision != Long.MIN_VALUE
&& nextRevision == currentRevision
&& hasSameRawBlock(c, incoming)
&& hasSameRawBlock(c, outgoing)) {
c.rollback();
return false;
}
replaceFileLinks(c, incoming.getMessageKey(), incoming.getTargetLogin(), incomingFiles);
replaceFileLinks(c, outgoing.getMessageKey(), outgoing.getTargetLogin(), outgoingFiles);
upsertMessage(c, incoming);
upsertMessage(c, outgoing);
resetDeliveryRows(c, incoming.getMessageKey());
resetDeliveryRows(c, outgoing.getMessageKey());
c.commit();
return true;
} catch (Exception ex) {
try { c.rollback(); } catch (Exception ignored) {}
throw ex;
} finally {
c.setAutoCommit(prevAutoCommit);
}
}
}
private boolean isConstraintViolation(SQLException ex) {
String msg = String.valueOf(ex.getMessage()).toLowerCase();
return msg.contains("constraint") || msg.contains("unique") || msg.contains("primary key");
public boolean fileExists(byte[] fileHash, long fileSize) throws Exception {
try (Connection c = db.getConnection()) {
return fileExists(c, fileHash, fileSize);
}
}
public boolean fileExistsByHash(byte[] fileHash) throws Exception {
try (Connection c = db.getConnection()) {
String sql = "SELECT 1 FROM dm_files WHERE file_hash_sha256 = ? LIMIT 1";
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setBytes(1, fileHash);
try (ResultSet rs = ps.executeQuery()) {
return rs.next();
}
}
}
}
public void registerFileIfAbsent(byte[] fileHash, long fileSize) throws Exception {
try (Connection c = db.getConnection()) {
String sql = """
INSERT INTO dm_files (file_hash_sha256, file_size, ref_count)
VALUES (?, ?, 0)
ON CONFLICT(file_hash_sha256) DO UPDATE SET
file_size = excluded.file_size
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setBytes(1, fileHash);
ps.setLong(2, fileSize);
ps.executeUpdate();
}
}
}
public SignedMessageV2Entry getByMessageKey(String messageKey) throws Exception {
@@ -123,7 +171,7 @@ public final class SignedMessagesV2DAO {
String sql = """
SELECT
message_key, base_key, target_login, from_login, to_login,
time_ms, nonce, message_type, raw_block, created_at_ms,
time_ms, nonce, message_type, revision_time_ms, raw_block, created_at_ms,
source_api, origin_session_id, receipt_ref_base_key, receipt_ref_type
FROM signed_messages_v2
WHERE message_key = ?
@@ -203,13 +251,13 @@ public final class SignedMessagesV2DAO {
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.raw_block, m.created_at_ms,
m.time_ms, m.nonce, m.message_type, m.revision_time_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.created_at_ms ASC
ORDER BY m.time_ms ASC, m.revision_time_ms ASC, m.created_at_ms ASC
""";
List<SignedMessageV2Entry> out = new ArrayList<>();
try (PreparedStatement ps = c.prepareStatement(sql)) {
@@ -222,6 +270,206 @@ public final class SignedMessagesV2DAO {
}
}
private void upsertMessage(Connection c, SignedMessageV2Entry e) throws SQLException {
String sql = """
INSERT INTO signed_messages_v2 (
message_key, base_key, target_login, from_login, to_login,
time_ms, nonce, message_type, revision_time_ms, raw_block, created_at_ms,
source_api, origin_session_id, receipt_ref_base_key, receipt_ref_type
) 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,
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
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
bindSignedMessage(ps, e);
ps.executeUpdate();
}
}
private void replaceFileLinks(Connection c, String messageKey, String login, List<DmFileRef> nextFiles) throws SQLException {
List<byte[]> oldHashes = listLinkedFileHashes(c, messageKey, login);
for (byte[] oldHash : oldHashes) {
adjustRefCount(c, oldHash, -1);
}
try (PreparedStatement ps = c.prepareStatement("""
DELETE FROM dm_message_file_links
WHERE message_key = ? AND login = ? COLLATE NOCASE
""")) {
ps.setString(1, messageKey);
ps.setString(2, login);
ps.executeUpdate();
}
if (nextFiles == null || nextFiles.isEmpty()) return;
Set<String> dedup = new HashSet<>();
for (DmFileRef ref : nextFiles) {
if (ref == null || ref.getFileHash() == null) continue;
String dedupKey = Arrays.toString(ref.getFileHash());
if (!dedup.add(dedupKey)) continue;
try (PreparedStatement ps = c.prepareStatement("""
INSERT OR IGNORE INTO dm_message_file_links (
message_key, login, file_hash_sha256
) VALUES (?, ?, ?)
""")) {
ps.setString(1, messageKey);
ps.setString(2, login);
ps.setBytes(3, ref.getFileHash());
int inserted = ps.executeUpdate();
if (inserted > 0) {
adjustRefCount(c, ref.getFileHash(), 1);
}
}
}
}
private List<byte[]> listLinkedFileHashes(Connection c, String messageKey, String login) throws SQLException {
String sql = """
SELECT file_hash_sha256
FROM dm_message_file_links
WHERE message_key = ? AND login = ? COLLATE NOCASE
""";
List<byte[]> out = new ArrayList<>();
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setString(1, messageKey);
ps.setString(2, login);
try (ResultSet rs = ps.executeQuery()) {
while (rs.next()) {
out.add(rs.getBytes(1));
}
}
}
return out;
}
private void adjustRefCount(Connection c, byte[] fileHash, int delta) throws SQLException {
String sql = """
UPDATE dm_files
SET ref_count = CASE
WHEN ref_count + ? < 0 THEN 0
ELSE ref_count + ?
END
WHERE file_hash_sha256 = ?
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setInt(1, delta);
ps.setInt(2, delta);
ps.setBytes(3, fileHash);
ps.executeUpdate();
}
}
private boolean allFilesExist(Connection c, List<DmFileRef> refs) throws SQLException {
if (refs == null) return true;
for (DmFileRef ref : refs) {
if (ref == null || ref.getFileHash() == null) return false;
if (!fileExists(c, ref.getFileHash(), ref.getFileSize())) return false;
}
return true;
}
private boolean fileExists(Connection c, byte[] fileHash, long fileSize) throws SQLException {
String sql = """
SELECT 1
FROM dm_files
WHERE file_hash_sha256 = ? AND file_size = ?
LIMIT 1
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setBytes(1, fileHash);
ps.setLong(2, fileSize);
try (ResultSet rs = ps.executeQuery()) {
return rs.next();
}
}
}
private Long getRevisionTimeMs(Connection c, String messageKey) throws SQLException {
String sql = "SELECT revision_time_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);
}
}
}
private boolean hasSameRawBlock(Connection c, SignedMessageV2Entry entry) throws SQLException {
String sql = "SELECT raw_block FROM signed_messages_v2 WHERE message_key = ? LIMIT 1";
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setString(1, entry.getMessageKey());
try (ResultSet rs = ps.executeQuery()) {
if (!rs.next()) return false;
return Arrays.equals(rs.getBytes(1), entry.getRawBlock());
}
}
}
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, SignedMessageV2Entry e) throws SQLException {
String sql = """
INSERT INTO signed_messages_v2 (
message_key, base_key, target_login, from_login, to_login,
time_ms, nonce, message_type, revision_time_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();
}
}
private void bindSignedMessage(PreparedStatement ps, SignedMessageV2Entry 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.setBytes(10, e.getRawBlock());
ps.setLong(11, e.getCreatedAtMs());
ps.setString(12, e.getSourceApi());
ps.setString(13, e.getOriginSessionId());
ps.setString(14, e.getReceiptRefBaseKey());
if (e.getReceiptRefType() == null) ps.setObject(15, null);
else ps.setInt(15, e.getReceiptRefType());
}
private boolean isConstraintViolation(SQLException ex) {
String msg = String.valueOf(ex.getMessage()).toLowerCase();
return msg.contains("constraint") || msg.contains("unique") || msg.contains("primary key");
}
private SignedMessageV2Entry mapRow(ResultSet rs) throws Exception {
SignedMessageV2Entry e = new SignedMessageV2Entry();
e.setMessageKey(rs.getString("message_key"));
@@ -232,6 +480,7 @@ public final class SignedMessagesV2DAO {
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.setRawBlock(rs.getBytes("raw_block"));
e.setCreatedAtMs(rs.getLong("created_at_ms"));
e.setSourceApi(rs.getString("source_api"));
SHiNE-server/shine-server-db/src/main/java/shine/db/entities/DmFileRef.java
+19
View File
@@ -0,0 +1,19 @@
package shine.db.entities;
public class DmFileRef {
private byte[] fileHash;
private long fileSize;
public DmFileRef() {
}
public DmFileRef(byte[] fileHash, long fileSize) {
this.fileHash = fileHash;
this.fileSize = fileSize;
}
public byte[] getFileHash() { return fileHash; }
public void setFileHash(byte[] fileHash) { this.fileHash = fileHash; }
public long getFileSize() { return fileSize; }
public void setFileSize(long fileSize) { this.fileSize = fileSize; }
}
SHiNE-server/shine-server-db/src/main/java/shine/db/entities/SignedMessageV2Entry.java
+3
View File
@@ -9,6 +9,7 @@ public class SignedMessageV2Entry {
private long timeMs;
private long nonce;
private int messageType;
private long revisionTimeMs;
private byte[] rawBlock;
private long createdAtMs;
private String sourceApi;
@@ -32,6 +33,8 @@ public class SignedMessageV2Entry {
public void setNonce(long nonce) { this.nonce = nonce; }
public int getMessageType() { return messageType; }
public void setMessageType(int messageType) { this.messageType = messageType; }
public long getRevisionTimeMs() { return revisionTimeMs; }
public void setRevisionTimeMs(long revisionTimeMs) { this.revisionTimeMs = revisionTimeMs; }
public byte[] getRawBlock() { return rawBlock; }
public void setRawBlock(byte[] rawBlock) { this.rawBlock = rawBlock; }
public long getCreatedAtMs() { return createdAtMs; }