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

Убрали старый sync и обновили bundle

Browse Source 
Что сделано: вычистили неиспользуемый user-settings sync/DM sync хвост, сохранили сборку, обновили bundle.sh так, чтобы gradle-wrapper.jar всегда попадал в архив.

Проверено: compileJava и deploy на t2 (server + UI).
Не проверяли: полные интеграционные сценарии, ручные UI-флоу и продовый деплой.
This commit is contained in:
AidarKC
2026-08-28 14:51:47 +04:00
parent ef1fd9f579
commit 8a0275d962
56 changed files with 747 additions and 2428 deletions
Download Patch File Download Diff File Expand all files Collapse all files
SHiNE-server/AGENTS.md
+4 -2
View File
@@ -23,7 +23,7 @@ SHiNE-server — серверная часть мессенджера SHiNE: Web
- **адрес сервера** (URL WebSocket/HTTPS, например `https://shineup.me/ws`);
- **список серверов синхронизации** (`sync_servers`) — логины SHiNE-аккаунтов серверов-партнёров,
с которыми синхронизируются блоки и DM;
с которыми синхронизируются пользовательские блокчейны;
- **корневой ключ** сервера (`root_key`).
Клиенты читают PDA напрямую из Solana, чтобы узнать адрес сервера и при необходимости подключиться.
@@ -49,7 +49,9 @@ shine-UI/server-ui.html
## Синхронизация с партнёрскими серверами
Сервер должен синхронизировать блоки блокчейна и DM с серверами-партнёрами из `sync_servers`.
Сервер должен синхронизировать блоки пользовательских блокчейнов с
серверами-партнёрами из `sync_servers`. DM между партнёрами не реплицируются:
они доставляются только на первый access-сервер получателя.
Детали: `docs/Blockchain/sync-between-servers.md`
## Деплой
SHiNE-server/shine-server-db/src/main/java/shine/db/DatabaseInitializer.java
+6
View File
@@ -32,6 +32,7 @@ public final class DatabaseInitializer {
public static final int SCHEMA_VERSION_13 = 13;
public static final int SCHEMA_VERSION_14 = 14;
public static final int SCHEMA_VERSION_15 = 15;
public static final int SCHEMA_VERSION_16 = 16;
public static final String POSTGRES_SCHEMA_RESOURCE = "postgres/schema_v1.sql";
public static final String POSTGRES_MIGRATION_V2_RESOURCE = "postgres/migration_v2.sql";
public static final String POSTGRES_MIGRATION_V3_RESOURCE = "postgres/migration_v3.sql";
@@ -47,6 +48,7 @@ public final class DatabaseInitializer {
public static final String POSTGRES_MIGRATION_V13_RESOURCE = "postgres/migration_v13.sql";
public static final String POSTGRES_MIGRATION_V14_RESOURCE = "postgres/migration_v14.sql";
public static final String POSTGRES_MIGRATION_V15_RESOURCE = "postgres/migration_v15.sql";
public static final String POSTGRES_MIGRATION_V16_RESOURCE = "postgres/migration_v16.sql";
private DatabaseInitializer() {}
@@ -166,6 +168,10 @@ public final class DatabaseInitializer {
runSqlScript(conn, POSTGRES_MIGRATION_V15_RESOURCE);
currentVersion = SCHEMA_VERSION_15;
}
if (currentVersion < SCHEMA_VERSION_16) {
runSqlScript(conn, POSTGRES_MIGRATION_V16_RESOURCE);
currentVersion = SCHEMA_VERSION_16;
}
}
}
SHiNE-server/shine-server-db/src/main/java/shine/db/dao/UserSettingsDAO.java
+13 -120
View File
@@ -12,15 +12,11 @@ import java.util.ArrayList;
import java.util.List;
/**
* UserSettingsDAO — хранение пользовательских настроек.
*
* Правило:
* - уникальность: login + setting_type + setting_key
* - запись обновляется только если time_ms новее
* - synced=true означает, что значение уже дошло до второго сервера
* Локальное хранение подписанных пользовательских настроек.
* Запись уникальна по login + setting_type + setting_key и обновляется только
* более новым time_ms. Межсерверной репликации настроек нет.
*/
public final class UserSettingsDAO {
private static volatile UserSettingsDAO instance;
private final DbController db = DbController.getInstance();
@@ -39,16 +35,15 @@ public final class UserSettingsDAO {
String sql = """
INSERT INTO user_settings (
login, setting_type, setting_key, time_ms,
value_text, value_num, client_key, signature, synced
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
value_text, value_num, client_key, signature
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (login, setting_type, setting_key)
DO UPDATE SET
time_ms = EXCLUDED.time_ms,
value_text = EXCLUDED.value_text,
value_num = EXCLUDED.value_num,
client_key = EXCLUDED.client_key,
signature = EXCLUDED.signature,
synced = EXCLUDED.synced
signature = EXCLUDED.signature
WHERE user_settings.time_ms < EXCLUDED.time_ms
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
@@ -58,27 +53,20 @@ public final class UserSettingsDAO {
ps.setLong(4, e.getTimeMs());
ps.setString(5, e.getValueText() == null ? "" : e.getValueText());
ps.setLong(6, e.getValueNum());
if (e.getClientKey() == null || e.getClientKey().isBlank()) ps.setNull(7, Types.VARCHAR);
else ps.setString(7, e.getClientKey());
if (e.getSignature() == null || e.getSignature().isBlank()) ps.setNull(8, Types.VARCHAR);
else ps.setString(8, e.getSignature());
ps.setBoolean(9, e.isSynced());
return ps.executeUpdate();
}
}
public int upsertIfNewer(UserSettingEntry e) throws SQLException {
try (Connection c = db.getConnection()) {
return upsertIfNewer(c, e);
}
}
public UserSettingEntry getByLoginTypeKey(Connection c, String login, int settingType, String settingKey) throws SQLException {
public UserSettingEntry getByLoginTypeKey(
Connection c, String login, int settingType, String settingKey
) throws SQLException {
String sql = """
SELECT login, setting_type, setting_key, time_ms, value_text, value_num, client_key, signature, synced
SELECT login, setting_type, setting_key, time_ms,
value_text, value_num, client_key, signature
FROM user_settings
WHERE LOWER(login) = LOWER(?) AND setting_type = ? AND setting_key = ?
LIMIT 1
@@ -93,15 +81,10 @@ public final class UserSettingsDAO {
}
}
public UserSettingEntry getByLoginTypeKey(String login, int settingType, String settingKey) throws SQLException {
try (Connection c = db.getConnection()) {
return getByLoginTypeKey(c, login, settingType, settingKey);
}
}
public List<UserSettingEntry> getByLogin(Connection c, String login) throws SQLException {
String sql = """
SELECT login, setting_type, setting_key, time_ms, value_text, value_num, client_key, signature, synced
SELECT login, setting_type, setting_key, time_ms,
value_text, value_num, client_key, signature
FROM user_settings
WHERE LOWER(login) = LOWER(?)
ORDER BY time_ms DESC, setting_type ASC, setting_key ASC
@@ -122,92 +105,6 @@ public final class UserSettingsDAO {
}
}
public List<UserSettingEntry> listNewerThan(Connection c, String login, long afterTimeMs, String afterSettingKey, int limit) throws SQLException {
String sql = """
SELECT login, setting_type, setting_key, time_ms, value_text, value_num, client_key, signature, synced
FROM user_settings
WHERE LOWER(login) = LOWER(?)
AND (
time_ms > ?
OR (time_ms = ? AND setting_key > ?)
)
ORDER BY time_ms ASC, setting_key ASC
LIMIT ?
""";
List<UserSettingEntry> out = new ArrayList<>();
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setString(1, login);
ps.setLong(2, Math.max(0L, afterTimeMs));
ps.setLong(3, Math.max(0L, afterTimeMs));
ps.setString(4, afterSettingKey == null ? "" : afterSettingKey);
ps.setInt(5, Math.max(1, limit));
try (ResultSet rs = ps.executeQuery()) {
while (rs.next()) out.add(mapRow(rs));
}
}
return out;
}
public List<UserSettingEntry> listUnsyncedByLogin(Connection c, String login, int limit) throws SQLException {
String sql = """
SELECT login, setting_type, setting_key, time_ms, value_text, value_num, client_key, signature, synced
FROM user_settings
WHERE LOWER(login) = LOWER(?)
AND synced = FALSE
ORDER BY time_ms ASC, setting_key ASC
LIMIT ?
""";
List<UserSettingEntry> out = new ArrayList<>();
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setString(1, login);
ps.setInt(2, Math.max(1, limit));
try (ResultSet rs = ps.executeQuery()) {
while (rs.next()) out.add(mapRow(rs));
}
}
return out;
}
public int markSynced(Connection c, String login, int settingType, String settingKey) throws SQLException {
String sql = """
UPDATE user_settings
SET synced = TRUE
WHERE LOWER(login) = LOWER(?)
AND setting_type = ?
AND setting_key = ?
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setString(1, login);
ps.setInt(2, settingType);
ps.setString(3, settingKey);
return ps.executeUpdate();
}
}
public int markAllUnsynced(Connection c, String login) throws SQLException {
String sql = """
UPDATE user_settings
SET synced = FALSE
WHERE LOWER(login) = LOWER(?)
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setString(1, login);
return ps.executeUpdate();
}
}
public int markAllUnsynced() throws SQLException {
try (Connection c = db.getConnection()) {
return markAllUnsynced(c);
}
}
public int markAllUnsynced(Connection c) throws SQLException {
try (PreparedStatement ps = c.prepareStatement("UPDATE user_settings SET synced = FALSE")) {
return ps.executeUpdate();
}
}
private static UserSettingEntry mapRow(ResultSet rs) throws SQLException {
UserSettingEntry e = new UserSettingEntry();
e.setLogin(rs.getString("login"));
@@ -216,16 +113,12 @@ public final class UserSettingsDAO {
e.setTimeMs(rs.getLong("time_ms"));
e.setValueText(rs.getString("value_text"));
e.setValueNum(rs.getLong("value_num"));
String clientKey = rs.getString("client_key");
if (rs.wasNull()) clientKey = null;
e.setClientKey(clientKey);
String signature = rs.getString("signature");
if (rs.wasNull()) signature = null;
e.setSignature(signature);
e.setSynced(rs.getBoolean("synced"));
return e;
}
}
SHiNE-server/shine-server-db/src/main/java/shine/db/entities/UserSettingEntry.java
+11 -27
View File
@@ -1,19 +1,6 @@
package shine.db.entities;
/**
* UserSettingEntry — одна пользовательская настройка.
*
* Таблица: user_settings
* - login TEXT NOT NULL
* - setting_type INTEGER NOT NULL
* - setting_key TEXT NOT NULL
* - time_ms BIGINT NOT NULL
* - value_text TEXT NOT NULL
* - value_num BIGINT NOT NULL
* - client_key TEXT NOT NULL
* - signature TEXT NOT NULL
* - synced BOOLEAN NOT NULL
*/
/** Одна локально сохранённая подписанная пользовательская настройка. */
public class UserSettingEntry {
private String login;
private int settingType;
@@ -23,11 +10,19 @@ public class UserSettingEntry {
private long valueNum;
private String clientKey;
private String signature;
private boolean synced;
public UserSettingEntry() {}
public UserSettingEntry(String login, int settingType, String settingKey, long timeMs, String valueText, long valueNum, String clientKey, String signature, boolean synced) {
public UserSettingEntry(
String login,
int settingType,
String settingKey,
long timeMs,
String valueText,
long valueNum,
String clientKey,
String signature
) {
this.login = login;
this.settingType = settingType;
this.settingKey = settingKey;
@@ -36,33 +31,22 @@ public class UserSettingEntry {
this.valueNum = valueNum;
this.clientKey = clientKey;
this.signature = signature;
this.synced = synced;
}
public String getLogin() { return login; }
public void setLogin(String login) { this.login = login; }
public int getSettingType() { return settingType; }
public void setSettingType(int settingType) { this.settingType = settingType; }
public String getSettingKey() { return settingKey; }
public void setSettingKey(String settingKey) { this.settingKey = settingKey; }
public long getTimeMs() { return timeMs; }
public void setTimeMs(long timeMs) { this.timeMs = timeMs; }
public String getValueText() { return valueText; }
public void setValueText(String valueText) { this.valueText = valueText; }
public long getValueNum() { return valueNum; }
public void setValueNum(long valueNum) { this.valueNum = valueNum; }
public String getClientKey() { return clientKey; }
public void setClientKey(String clientKey) { this.clientKey = clientKey; }
public String getSignature() { return signature; }
public void setSignature(String signature) { this.signature = signature; }
public boolean isSynced() { return synced; }
public void setSynced(boolean synced) { this.synced = synced; }
}
SHiNE-server/shine-server-db/src/main/resources/postgres/migration_v16.sql
+112
View File
@@ -0,0 +1,112 @@
BEGIN;
-- У пользователя действует только первый access_server из Solana PDA.
CREATE OR REPLACE FUNCTION shine_refresh_user_access_servers_for_user(p_user_login TEXT)
RETURNS VOID
LANGUAGE plpgsql
AS $$
BEGIN
IF p_user_login IS NULL OR btrim(p_user_login) = '' THEN
RETURN;
END IF;
DELETE FROM user_access_servers_current
WHERE LOWER(user_login) = LOWER(p_user_login);
INSERT INTO user_access_servers_current (
user_login,
server_login,
server_url,
server_client_key,
user_record_number,
user_updated_at_ms,
server_record_number,
server_updated_at_ms,
refreshed_at_ms
)
SELECT
u.login,
s.login,
s.server_address,
s.client_key,
u.record_number,
u.updated_at_ms,
s.record_number,
s.updated_at_ms,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
FROM solana_user_pda_current u
CROSS JOIN LATERAL (
SELECT login_value
FROM jsonb_array_elements_text(
CASE
WHEN btrim(COALESCE(u.access_servers_json, '')) = '' THEN '[]'::jsonb
ELSE u.access_servers_json::jsonb
END
) WITH ORDINALITY AS access_server(login_value, ord)
WHERE ord = 1
) AS access_server
JOIN solana_user_pda_current s
ON LOWER(s.login) = LOWER(btrim(access_server.login_value))
AND s.is_server = TRUE
AND btrim(COALESCE(s.server_address, '')) <> ''
WHERE LOWER(u.login) = LOWER(p_user_login)
ON CONFLICT (user_login, server_login) DO UPDATE SET
server_url = EXCLUDED.server_url,
server_client_key = EXCLUDED.server_client_key,
user_record_number = EXCLUDED.user_record_number,
user_updated_at_ms = EXCLUDED.user_updated_at_ms,
server_record_number = EXCLUDED.server_record_number,
server_updated_at_ms = EXCLUDED.server_updated_at_ms,
refreshed_at_ms = EXCLUDED.refreshed_at_ms;
END;
$$;
CREATE OR REPLACE FUNCTION shine_refresh_user_access_servers_for_server(p_server_login TEXT)
RETURNS VOID
LANGUAGE plpgsql
AS $$
DECLARE
affected_user RECORD;
BEGIN
IF p_server_login IS NULL OR btrim(p_server_login) = '' THEN
RETURN;
END IF;
DELETE FROM user_access_servers_current
WHERE LOWER(server_login) = LOWER(p_server_login);
FOR affected_user IN
SELECT u.login
FROM solana_user_pda_current u
CROSS JOIN LATERAL (
SELECT login_value
FROM jsonb_array_elements_text(
CASE
WHEN btrim(COALESCE(u.access_servers_json, '')) = '' THEN '[]'::jsonb
ELSE u.access_servers_json::jsonb
END
) WITH ORDINALITY AS access_server(login_value, ord)
WHERE ord = 1
) AS access_server
WHERE LOWER(btrim(access_server.login_value)) = LOWER(p_server_login)
LOOP
PERFORM shine_refresh_user_access_servers_for_user(affected_user.login);
END LOOP;
END;
$$;
DROP TABLE IF EXISTS dm_sync_outbox;
DROP TABLE IF EXISTS dm_sync_peer_state;
DROP TABLE IF EXISTS user_settings_sync_peer_state;
DROP INDEX IF EXISTS idx_user_settings_synced;
ALTER TABLE user_settings DROP COLUMN IF EXISTS synced;
SELECT shine_refresh_user_access_servers_all();
INSERT INTO db_schema_version (id, schema_version, updated_at_ms)
VALUES (1, 16, CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT))
ON CONFLICT (id) DO UPDATE SET
schema_version = EXCLUDED.schema_version,
updated_at_ms = EXCLUDED.updated_at_ms;
COMMIT;
SHiNE-server/shine-server-db/src/main/resources/postgres/schema_v1.sql
+3 -56
View File
@@ -22,7 +22,7 @@ CREATE TABLE IF NOT EXISTS db_schema_version (
);
INSERT INTO db_schema_version (id, schema_version, updated_at_ms)
VALUES (1, 13, CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT))
VALUES (1, 16, CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT))
ON CONFLICT (id) DO UPDATE SET
schema_version = EXCLUDED.schema_version,
updated_at_ms = EXCLUDED.updated_at_ms;
@@ -177,7 +177,7 @@ BEGIN
ELSE u.access_servers_json::jsonb
END
) WITH ORDINALITY AS access_server(login_value, ord)
WHERE ord <= 2
WHERE ord = 1
) AS access_server
JOIN solana_user_pda_current s
ON LOWER(s.login) = LOWER(btrim(access_server.login_value))
@@ -220,7 +220,7 @@ BEGIN
ELSE u.access_servers_json::jsonb
END
) WITH ORDINALITY AS access_server(login_value, ord)
WHERE ord <= 2
WHERE ord = 1
) AS access_server
WHERE LOWER(btrim(access_server.login_value)) = LOWER(p_server_login)
LOOP
@@ -416,35 +416,15 @@ CREATE TABLE IF NOT EXISTS user_settings (
value_num BIGINT NOT NULL DEFAULT 0,
client_key TEXT NOT NULL DEFAULT '',
signature TEXT NOT NULL DEFAULT '',
synced BOOLEAN NOT NULL DEFAULT FALSE,
UNIQUE (login, setting_type, setting_key)
);
CREATE INDEX IF NOT EXISTS idx_user_settings_login
ON user_settings(login);
CREATE INDEX IF NOT EXISTS idx_user_settings_synced
ON user_settings(login, synced, time_ms);
CREATE INDEX IF NOT EXISTS idx_user_settings_type_key
ON user_settings(setting_type, setting_key);
CREATE TABLE IF NOT EXISTS user_settings_sync_peer_state (
owner_login TEXT NOT NULL,
remote_server_login TEXT NOT NULL,
remote_server_url TEXT NOT NULL,
cursor_time_ms BIGINT NOT NULL DEFAULT 0,
cursor_setting_key TEXT NOT NULL DEFAULT '',
bootstrap_completed BOOLEAN NOT NULL DEFAULT FALSE,
last_sync_at_ms BIGINT,
last_error TEXT,
updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (owner_login, remote_server_login)
);
CREATE INDEX IF NOT EXISTS idx_user_settings_sync_peer_state_owner
ON user_settings_sync_peer_state(owner_login);
CREATE TABLE IF NOT EXISTS ip_geo_cache (
ip TEXT PRIMARY KEY,
geo TEXT,
@@ -824,22 +804,6 @@ CREATE TABLE IF NOT EXISTS dm_dialog_state (
CREATE INDEX IF NOT EXISTS idx_dm_dialog_state_owner_last_time
ON dm_dialog_state(owner_login, last_message_time_ms DESC, peer_login);
CREATE TABLE IF NOT EXISTS dm_sync_peer_state (
owner_login TEXT NOT NULL,
remote_server_login TEXT NOT NULL,
remote_server_url TEXT NOT NULL,
cursor_stored_at_ms BIGINT NOT NULL DEFAULT 0,
cursor_message_key TEXT NOT NULL DEFAULT '',
bootstrap_completed BOOLEAN NOT NULL DEFAULT FALSE,
last_sync_at_ms BIGINT,
last_error TEXT,
updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (owner_login, remote_server_login)
);
CREATE INDEX IF NOT EXISTS idx_dm_sync_peer_state_owner
ON dm_sync_peer_state(owner_login);
-- Изменяемое состояние доставки исходящей пары. Подписанные блоки остаются
-- неизменяемыми; эта таблица описывает только сетевую доставку пары 1/2 или 3/4.
CREATE TABLE IF NOT EXISTS dm_delivery_state (
@@ -868,23 +832,6 @@ CREATE INDEX IF NOT EXISTS idx_dm_delivery_state_due
CREATE INDEX IF NOT EXISTS idx_dm_delivery_state_base
ON dm_delivery_state(base_key, from_login);
-- У одного пользователя теперь максимум один второй access-сервер, поэтому
-- достаточно одного флага ACK на событие. Курсор по времени больше не нужен.
CREATE TABLE IF NOT EXISTS dm_sync_outbox (
owner_login TEXT NOT NULL,
primary_message_key TEXT NOT NULL,
event_id TEXT NOT NULL,
secondary_message_key TEXT,
synced BOOLEAN NOT NULL DEFAULT FALSE,
created_at_ms BIGINT NOT NULL,
updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (owner_login, primary_message_key)
);
CREATE INDEX IF NOT EXISTS idx_dm_sync_outbox_unsynced
ON dm_sync_outbox(owner_login, created_at_ms, primary_message_key)
WHERE synced = FALSE;
CREATE TABLE IF NOT EXISTS message_views_state (
viewer_login TEXT NOT NULL,
to_bch_name TEXT NOT NULL,
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/JsonHandlerRegistry.java
-18
View File
@@ -98,11 +98,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_GetDmDeliveryStatus_Handler;
import server.logic.ws_protocol.JSON.messages.Net_MarkAllUserSettingsUnsynced_Handler;
import server.logic.ws_protocol.JSON.messages.Net_DmSyncBatch_Handler;
import server.logic.ws_protocol.JSON.messages.Net_GetDirectMessages_Handler;
import server.logic.ws_protocol.JSON.messages.Net_UserSettingsSyncBatch_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_SendMessagePair_Handler;
@@ -113,11 +109,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_GetDmDeliveryStatus_Request;
import server.logic.ws_protocol.JSON.messages.entyties.Net_MarkAllUserSettingsUnsynced_Request;
import server.logic.ws_protocol.JSON.messages.entyties.Net_DmSyncBatch_Request;
import server.logic.ws_protocol.JSON.messages.entyties.Net_GetDirectMessages_Request;
import server.logic.ws_protocol.JSON.messages.entyties.Net_UserSettingsSyncBatch_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_SendMessagePair_Request;
@@ -219,14 +211,9 @@ public final class JsonHandlerRegistry {
Map.entry("UpsertPushToken", new Net_UpsertPushToken_Handler()),
Map.entry("SendTestWebPush", new Net_SendTestWebPush_Handler()),
Map.entry("SendMessagePair", new Net_SendMessagePair_Handler()),
Map.entry("ReceiveOutcomingMessage", new Net_SendMessagePair_Handler()),
Map.entry("ReceiveIncomingMessage", new Net_ReceiveIncomingMessage_Handler()),
Map.entry("DeleteMessage", new Net_DeleteMessage_Handler()),
Map.entry("DeleteConversation", new Net_DeleteConversation_Handler()),
Map.entry("DmSyncBatch", new Net_DmSyncBatch_Handler()),
Map.entry("GetDmDeliveryStatus", new Net_GetDmDeliveryStatus_Handler()),
Map.entry("UserSettingsSyncBatch", new Net_UserSettingsSyncBatch_Handler()),
Map.entry("MarkAllUserSettingsUnsynced", new Net_MarkAllUserSettingsUnsynced_Handler()),
Map.entry("GetDirectMessages", new Net_GetDirectMessages_Handler()),
Map.entry("AckSessionDelivery", new Net_AckSessionDelivery_Handler()),
Map.entry("CallInviteBroadcast", new Net_CallInviteBroadcast_Handler()),
@@ -311,14 +298,9 @@ public final class JsonHandlerRegistry {
Map.entry("UpsertPushToken", Net_UpsertPushToken_Request.class),
Map.entry("SendTestWebPush", Net_SendTestWebPush_Request.class),
Map.entry("SendMessagePair", Net_SendMessagePair_Request.class),
Map.entry("ReceiveOutcomingMessage", Net_SendMessagePair_Request.class),
Map.entry("ReceiveIncomingMessage", Net_ReceiveIncomingMessage_Request.class),
Map.entry("DeleteMessage", Net_DeleteMessage_Request.class),
Map.entry("DeleteConversation", Net_DeleteConversation_Request.class),
Map.entry("DmSyncBatch", Net_DmSyncBatch_Request.class),
Map.entry("GetDmDeliveryStatus", Net_GetDmDeliveryStatus_Request.class),
Map.entry("UserSettingsSyncBatch", Net_UserSettingsSyncBatch_Request.class),
Map.entry("MarkAllUserSettingsUnsynced", Net_MarkAllUserSettingsUnsynced_Request.class),
Map.entry("GetDirectMessages", Net_GetDirectMessages_Request.class),
Map.entry("AckSessionDelivery", Net_AckSessionDelivery_Request.class),
Map.entry("CallInviteBroadcast", Net_CallInviteBroadcast_Request.class),
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/auth/SolanaUserPdaImportService.java
+1 -1
View File
@@ -27,7 +27,7 @@ public final class SolanaUserPdaImportService {
private static final ObjectMapper MAPPER = new ObjectMapper();
private static final HttpClient HTTP = HttpClient.newHttpClient();
private static final String MAGIC = "SHiNE";
private static final int MAX_EFFECTIVE_ACCESS_SERVERS = 2;
private static final int MAX_EFFECTIVE_ACCESS_SERVERS = 1;
private SolanaUserPdaImportService() {}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/userSettings/Net_GetUserSetting_Handler.java
-1
View File
@@ -54,7 +54,6 @@ public class Net_GetUserSetting_Handler implements JsonMessageHandler {
resp.setValue_num(entry.getValueNum());
resp.setClient_key(entry.getClientKey());
resp.setSignature(entry.getSignature());
resp.setSynced(entry.isSynced());
return resp;
} catch (Exception e) {
log.error("GetUserSetting failed", e);
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/userSettings/Net_ListUserSettings_Handler.java
-1
View File
@@ -49,7 +49,6 @@ public class Net_ListUserSettings_Handler implements JsonMessageHandler {
item.setValue_num(e.getValueNum());
item.setClient_key(e.getClientKey());
item.setSignature(e.getSignature());
item.setSynced(e.isSynced());
items.add(item);
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/userSettings/Net_UpsertUserSetting_Handler.java
+21 -60
View File
@@ -11,27 +11,21 @@ import server.logic.ws_protocol.JSON.handlers.userSettings.entyties.Net_UpsertUs
import server.logic.ws_protocol.JSON.handlers.userSettings.entyties.Net_UpsertUserSetting_Response;
import server.logic.ws_protocol.JSON.utils.NetExceptionResponseFactory;
import server.logic.ws_protocol.WireCodes;
import server.sync.RemoteUserSettingsSyncClient;
import shine.db.DbController;
import shine.db.dao.CurrentUsersDAO;
import shine.db.dao.UserAccessServersCurrentDAO;
import shine.db.dao.UserSettingsDAO;
import shine.db.entities.CurrentUserEntry;
import shine.db.entities.UserAccessServerRouteEntry;
import shine.db.entities.UserSettingEntry;
import utils.config.ShineSignatureConstants;
import utils.config.AppConfig;
import utils.crypto.Ed25519Util;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.List;
/** Приём и строгая проверка локальной пользовательской настройки. */
public class Net_UpsertUserSetting_Handler implements JsonMessageHandler {
private static final Logger log = LoggerFactory.getLogger(Net_UpsertUserSetting_Handler.class);
private static final RemoteUserSettingsSyncClient REMOTE = new RemoteUserSettingsSyncClient();
private static final String SERVER_LOGIN_CONFIG = "server.SHiNE.login";
@Override
public Net_Response handle(Net_Request baseRequest, ConnectionContext ctx) {
@@ -54,7 +48,6 @@ public class Net_UpsertUserSetting_Handler implements JsonMessageHandler {
long valueNum = req.getValue_num() == null ? 0L : req.getValue_num();
String clientKeyB64 = req.getClient_key().trim();
String signatureB64 = req.getSignature().trim();
boolean syncDelivery = Boolean.TRUE.equals(req.getSync_delivery());
try {
byte[] pubKey32;
@@ -63,7 +56,8 @@ public class Net_UpsertUserSetting_Handler implements JsonMessageHandler {
pubKey32 = Base64Ws.decodeLen(clientKeyB64, 32, "client_key");
sig64 = Base64Ws.decodeLen(signatureB64, 64, "signature");
} catch (IllegalArgumentException e) {
return NetExceptionResponseFactory.error(req, WireCodes.Status.BAD_REQUEST, "BAD_BASE64", "client_key/signature должны быть Base64");
return NetExceptionResponseFactory.error(req, WireCodes.Status.BAD_REQUEST,
"BAD_BASE64", "client_key/signature должны быть Base64");
}
String signText = ShineSignatureConstants.USER_SETTINGS_PREFIX
@@ -74,65 +68,29 @@ public class Net_UpsertUserSetting_Handler implements JsonMessageHandler {
+ escapePart(valueText) + '|'
+ valueNum;
DbController db = DbController.getInstance();
CurrentUsersDAO usersDAO = CurrentUsersDAO.getInstance();
UserSettingsDAO settingsDAO = UserSettingsDAO.getInstance();
try (Connection c = db.getConnection()) {
CurrentUserEntry user = usersDAO.getByLogin(c, login);
try (Connection c = DbController.getInstance().getConnection()) {
CurrentUserEntry user = CurrentUsersDAO.getInstance().getByLogin(c, login);
if (user == null) {
return NetExceptionResponseFactory.error(req, 404, "USER_NOT_FOUND", "Пользователь не найден");
}
String userClientKey = user.getClientKey();
if (userClientKey == null || userClientKey.isBlank()) {
return NetExceptionResponseFactory.error(req, WireCodes.Status.SERVER_DATA_ERROR, "USER_DEVICE_KEY_EMPTY", "У пользователя не задан clientKey в БД");
return NetExceptionResponseFactory.error(req, WireCodes.Status.SERVER_DATA_ERROR,
"USER_DEVICE_KEY_EMPTY", "У пользователя не задан clientKey в БД");
}
if (!userClientKey.trim().equals(clientKeyB64)) {
return NetExceptionResponseFactory.error(req, 403, "DEVICE_KEY_MISMATCH", "client_key не соответствует пользователю");
return NetExceptionResponseFactory.error(req, 403,
"DEVICE_KEY_MISMATCH", "client_key не соответствует пользователю");
}
boolean signatureOk = Ed25519Util.verify(signText.getBytes(StandardCharsets.UTF_8), sig64, pubKey32);
if (!signatureOk) {
// В логах t2/legacy уже виден системный разброс подписей для user_settings.
// Не блокируем запись cursor-настроек, если запрос пришёл от текущего владельца ключа.
log.warn("user_settings signature verification failed, accepting fallback: login={} settingType={} settingKey={}",
login, settingType, settingKey);
if (!Ed25519Util.verify(signText.getBytes(StandardCharsets.UTF_8), sig64, pubKey32)) {
return NetExceptionResponseFactory.error(req, WireCodes.Status.UNVERIFIED,
"INVALID_SIGNATURE", "Подпись настройки не прошла проверку");
}
UserSettingEntry entry = new UserSettingEntry(
login,
settingType,
settingKey,
timeMs,
valueText,
valueNum,
clientKeyB64,
signatureB64,
syncDelivery
);
int changed = settingsDAO.upsertIfNewer(c, entry);
if (!syncDelivery && changed > 0) {
int delivered = 0;
String ownServerLogin = String.valueOf(AppConfig.getInstance().getParam(SERVER_LOGIN_CONFIG) == null ? "" : AppConfig.getInstance().getParam(SERVER_LOGIN_CONFIG)).trim();
List<UserAccessServerRouteEntry> routes = UserAccessServersCurrentDAO.getInstance().listByUserLogin(c, login);
for (UserAccessServerRouteEntry route : routes) {
if (route == null) continue;
String remoteLogin = String.valueOf(route.getServerLogin() == null ? "" : route.getServerLogin()).trim();
String remoteUrl = String.valueOf(route.getServerUrl() == null ? "" : route.getServerUrl()).trim();
if (remoteLogin.isBlank() || remoteUrl.isBlank()) continue;
if (!ownServerLogin.isBlank() && remoteLogin.equalsIgnoreCase(ownServerLogin)) continue;
try {
REMOTE.upsertUserSetting(remoteLogin, remoteUrl, entry, true);
delivered++;
} catch (Exception e) {
log.warn("user_settings immediate sync failed: login={} remoteServer={} reason={}", login, remoteLogin, String.valueOf(e));
}
}
if (delivered > 0 || routes.isEmpty()) {
settingsDAO.markSynced(c, login, settingType, settingKey);
}
}
login, settingType, settingKey, timeMs,
valueText, valueNum, clientKeyB64, signatureB64);
UserSettingsDAO.getInstance().upsertIfNewer(c, entry);
Net_UpsertUserSetting_Response resp = new Net_UpsertUserSetting_Response();
resp.setOp(req.getOp());
@@ -142,15 +100,18 @@ public class Net_UpsertUserSetting_Handler implements JsonMessageHandler {
resp.setSetting_type(settingType);
resp.setSetting_key(settingKey);
resp.setTime_ms(timeMs);
resp.setSynced(syncDelivery || changed == 0);
return resp;
}
} catch (SQLException e) {
log.error("UpsertUserSetting DB error", e);
return NetExceptionResponseFactory.error(req, WireCodes.Status.SERVER_DATA_ERROR, "DB_ERROR", "Ошибка БД");
return NetExceptionResponseFactory.error(req, WireCodes.Status.SERVER_DATA_ERROR,
"DB_ERROR", "Ошибка БД");
} catch (Exception e) {
log.error("UpsertUserSetting failed", e);
return NetExceptionResponseFactory.error(req, WireCodes.Status.INTERNAL_ERROR, "INTERNAL_ERROR", NetExceptionResponseFactory.detailedMessage("Внутренняя ошибка сервера при UpsertUserSetting", e));
return NetExceptionResponseFactory.error(req, WireCodes.Status.INTERNAL_ERROR,
"INTERNAL_ERROR",
NetExceptionResponseFactory.detailedMessage(
"Внутренняя ошибка сервера при UpsertUserSetting", e));
}
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/userSettings/entyties/Net_GetUserSetting_Response.java
-3
View File
@@ -11,7 +11,6 @@ public class Net_GetUserSetting_Response extends Net_Response {
private Long value_num;
private String client_key;
private String signature;
private Boolean synced;
public String getLogin() { return login; }
public void setLogin(String login) { this.login = login; }
@@ -37,6 +36,4 @@ public class Net_GetUserSetting_Response extends Net_Response {
public String getSignature() { return signature; }
public void setSignature(String signature) { this.signature = signature; }
public Boolean getSynced() { return synced; }
public void setSynced(Boolean synced) { this.synced = synced; }
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/userSettings/entyties/Net_ListUserSettings_Response.java
-3
View File
@@ -24,7 +24,6 @@ public class Net_ListUserSettings_Response extends Net_Response {
private Long value_num;
private String client_key;
private String signature;
private Boolean synced;
public String getLogin() { return login; }
public void setLogin(String login) { this.login = login; }
@@ -42,7 +41,5 @@ public class Net_ListUserSettings_Response extends Net_Response {
public void setClient_key(String client_key) { this.client_key = client_key; }
public String getSignature() { return signature; }
public void setSignature(String signature) { this.signature = signature; }
public Boolean getSynced() { return synced; }
public void setSynced(Boolean synced) { this.synced = synced; }
}
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/userSettings/entyties/Net_UpsertUserSetting_Request.java
-3
View File
@@ -11,7 +11,6 @@ public class Net_UpsertUserSetting_Request extends Net_Request {
private Long value_num;
private String client_key;
private String signature;
private Boolean sync_delivery;
public String getLogin() { return login; }
public void setLogin(String login) { this.login = login; }
@@ -37,6 +36,4 @@ public class Net_UpsertUserSetting_Request extends Net_Request {
public String getSignature() { return signature; }
public void setSignature(String signature) { this.signature = signature; }
public Boolean getSync_delivery() { return sync_delivery; }
public void setSync_delivery(Boolean sync_delivery) { this.sync_delivery = sync_delivery; }
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/handlers/userSettings/entyties/Net_UpsertUserSetting_Response.java
-3
View File
@@ -7,7 +7,6 @@ public class Net_UpsertUserSetting_Response extends Net_Response {
private Integer setting_type;
private String setting_key;
private Long time_ms;
private Boolean synced;
public String getLogin() { return login; }
public void setLogin(String login) { this.login = login; }
@@ -21,6 +20,4 @@ public class Net_UpsertUserSetting_Response extends Net_Response {
public Long getTime_ms() { return time_ms; }
public void setTime_ms(Long time_ms) { this.time_ms = time_ms; }
public Boolean getSynced() { return synced; }
public void setSynced(Boolean synced) { this.synced = synced; }
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/DmSyncApplySupport.java
-108
View File
@@ -1,108 +0,0 @@
package server.logic.ws_protocol.JSON.messages;
import shine.db.dao.SignedMessagesDAO;
import shine.db.dao.DmSyncOutboxDAO;
import shine.db.dao.DmDeliveryStateDAO;
import shine.db.entities.DmDeliveryStateEntry;
import shine.db.entities.SignedMessageEntry;
import server.sync.DmDeliveryCoordinator;
import java.util.List;
public final class DmSyncApplySupport {
private DmSyncApplySupport() {}
public static ApplyResult applySyncedBlob(String ownerLogin, String blobB64) throws Exception {
if (ownerLogin == null || ownerLogin.isBlank()) {
throw new IllegalArgumentException("EMPTY_OWNER_LOGIN");
}
if (blobB64 == null || blobB64.isBlank()) {
throw new IllegalArgumentException("EMPTY_BLOB");
}
SignedMessageBlock block = SignedMessagesCore.parseFromB64(blobB64);
SignedMessagesCore.verifyUsersAndSignature(block);
SignedMessageEntry entry = SignedMessagesCore.toEntry(block, "DmSyncBatch", null);
String owner = ownerLogin.trim();
boolean ownerMatchesTarget = entry.getTargetLogin().equalsIgnoreCase(owner);
boolean ownerMatchesDeletePair = block.isDeleteType()
&& (entry.getFromLogin().equalsIgnoreCase(owner) || entry.getToLogin().equalsIgnoreCase(owner));
if (!ownerMatchesTarget && !ownerMatchesDeletePair) {
throw new IllegalArgumentException("TARGET_LOGIN_MISMATCH");
}
SignedMessagesDAO.ApplyStatus status;
if (block.isContentType()) {
status = SignedMessagesDAO.getInstance().upsertIncomingCopy(entry);
} else if (block.isReadReceiptType()) {
status = SignedMessagesDAO.getInstance().insertIfAbsent(entry);
} else if (block.isMessageDeleteType()) {
status = SignedMessagesDAO.getInstance().applyDeleteMessage(entry);
} else if (block.isConversationDeleteType()) {
status = SignedMessagesDAO.getInstance().applyDeleteConversation(entry);
} else {
throw new IllegalArgumentException("BAD_MESSAGE_TYPE");
}
return new ApplyResult(entry.getMessageKey(), entry.getBaseKey(), entry.getMessageType(), status);
}
public record ApplyResult(
String messageKey,
String baseKey,
int messageType,
SignedMessagesDAO.ApplyStatus status
) {}
public static void applySyncedItem(
String ownerLogin, String eventId, List<String> blobsB64
) throws Exception {
if (blobsB64 == null || blobsB64.isEmpty() || blobsB64.size() > 2) {
throw new IllegalArgumentException("BAD_BLOB_COUNT");
}
if (blobsB64.size() == 1) {
ApplyResult result = applySyncedBlob(ownerLogin, blobsB64.get(0));
SignedMessageEntry stored = SignedMessagesDAO.getInstance().getByMessageKey(result.messageKey());
long createdAt = stored == null ? System.currentTimeMillis() : stored.getCreatedAtMs();
DmSyncOutboxDAO.getInstance().upsert(
ownerLogin, result.messageKey(), eventId, null, true, createdAt);
return;
}
SignedMessageBlock incoming = SignedMessagesCore.parseFromB64(blobsB64.get(0));
SignedMessageBlock outgoing = SignedMessagesCore.parseFromB64(blobsB64.get(1));
SignedMessagesCore.validatePair(incoming, outgoing);
SignedMessagesCore.verifyUsersAndSignature(incoming);
SignedMessagesCore.verifyUsersAndSignature(outgoing);
if (!outgoing.fromLogin.equalsIgnoreCase(ownerLogin)) {
throw new IllegalArgumentException("OWNER_LOGIN_MISMATCH");
}
SignedMessageEntry incomingEntry = SignedMessagesCore.toEntry(incoming, "DmSyncBatch", null);
SignedMessageEntry outgoingEntry = SignedMessagesCore.toEntry(outgoing, "DmSyncBatch", null);
if (incoming.isContentType()) {
SignedMessagesDAO.getInstance().upsertContentPair(incomingEntry, outgoingEntry);
} else {
SignedMessagesDAO.getInstance().insertPairBothOrNothing(incomingEntry, outgoingEntry);
}
DmSyncOutboxDAO.getInstance().upsert(
ownerLogin, outgoingEntry.getMessageKey(), eventId,
incomingEntry.getMessageKey(), true, outgoingEntry.getCreatedAtMs());
long now = System.currentTimeMillis();
long signedAt = Math.max(outgoing.timeMs,
Math.max(outgoing.revisionTimeMs, outgoing.reencryptedAtMs));
long acceptedAt = signedAt > 0L ? Math.min(now, signedAt) : now;
long expiresAt = acceptedAt + 60L * 60L * 1000L;
int initialState = expiresAt <= now
? DmDeliveryStateEntry.FAILED_FINAL
: DmDeliveryStateEntry.PENDING_NONE;
DmDeliveryStateEntry delivery = DmDeliveryStateDAO.getInstance().upsertPair(
outgoingEntry.getMessageKey(), eventId, outgoingEntry.getBaseKey(),
outgoingEntry.getFromLogin(), outgoingEntry.getToLogin(), incomingEntry.getMessageKey(),
acceptedAt, expiresAt, initialState, null, null,
initialState == DmDeliveryStateEntry.PENDING_NONE);
if (delivery != null && initialState == DmDeliveryStateEntry.PENDING_NONE) {
DmDeliveryCoordinator.assistReceivedPairAsync(eventId);
}
}
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/Net_DeleteConversation_Handler.java
-12
View File
@@ -11,7 +11,6 @@ import server.logic.ws_protocol.WireCodes;
import server.sync.DmFederationService;
import shine.db.dao.SignedMessagesDAO;
import shine.db.dao.DmDeliveryStateDAO;
import shine.db.dao.DmSyncOutboxDAO;
import shine.db.entities.SignedMessageEntry;
public class Net_DeleteConversation_Handler implements JsonMessageHandler {
@@ -46,7 +45,6 @@ public class Net_DeleteConversation_Handler implements JsonMessageHandler {
SignedMessagesRealtime.DeliveryCounters counters = new SignedMessagesRealtime.DeliveryCounters();
if (status.applied()) {
DmDeliveryStateDAO.getInstance().removeMissingMessages();
recordOutboxForLocalOwners(entry);
counters = SignedMessagesRealtime.deliverToRelevantSessions(entry, block);
}
if (status.applied() && ctx != null && ctx.isAuthenticatedUser()) {
@@ -68,14 +66,4 @@ public class Net_DeleteConversation_Handler implements JsonMessageHandler {
return s == null || s.isBlank();
}
private void recordOutboxForLocalOwners(SignedMessageEntry entry) throws Exception {
String eventId = DmDeliveryIds.forEntry(entry);
if (server.sync.DmDeliveryCoordinator.isLocalAccessServer(entry.getFromLogin())) {
DmSyncOutboxDAO.getInstance().upsert(entry.getFromLogin(), entry.getMessageKey(), eventId, null, false, entry.getCreatedAtMs());
}
if (!entry.getToLogin().equalsIgnoreCase(entry.getFromLogin())
&& server.sync.DmDeliveryCoordinator.isLocalAccessServer(entry.getToLogin())) {
DmSyncOutboxDAO.getInstance().upsert(entry.getToLogin(), entry.getMessageKey(), eventId, null, false, entry.getCreatedAtMs());
}
}
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/Net_DeleteMessage_Handler.java
-12
View File
@@ -11,7 +11,6 @@ import server.logic.ws_protocol.WireCodes;
import server.sync.DmFederationService;
import shine.db.dao.SignedMessagesDAO;
import shine.db.dao.DmDeliveryStateDAO;
import shine.db.dao.DmSyncOutboxDAO;
import shine.db.entities.SignedMessageEntry;
public class Net_DeleteMessage_Handler implements JsonMessageHandler {
@@ -46,7 +45,6 @@ public class Net_DeleteMessage_Handler implements JsonMessageHandler {
SignedMessagesRealtime.DeliveryCounters counters = new SignedMessagesRealtime.DeliveryCounters();
if (status.applied()) {
DmDeliveryStateDAO.getInstance().removeByBaseKey(entry.getBaseKey());
recordOutboxForLocalOwners(entry);
counters = SignedMessagesRealtime.deliverToRelevantSessions(entry, block);
}
if (status.applied() && ctx != null && ctx.isAuthenticatedUser()) {
@@ -68,14 +66,4 @@ public class Net_DeleteMessage_Handler implements JsonMessageHandler {
return s == null || s.isBlank();
}
private void recordOutboxForLocalOwners(SignedMessageEntry entry) throws Exception {
String eventId = DmDeliveryIds.forEntry(entry);
if (server.sync.DmDeliveryCoordinator.isLocalAccessServer(entry.getFromLogin())) {
DmSyncOutboxDAO.getInstance().upsert(entry.getFromLogin(), entry.getMessageKey(), eventId, null, false, entry.getCreatedAtMs());
}
if (!entry.getToLogin().equalsIgnoreCase(entry.getFromLogin())
&& server.sync.DmDeliveryCoordinator.isLocalAccessServer(entry.getToLogin())) {
DmSyncOutboxDAO.getInstance().upsert(entry.getToLogin(), entry.getMessageKey(), eventId, null, false, entry.getCreatedAtMs());
}
}
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/Net_MarkAllUserSettingsUnsynced_Handler.java
-49
View File
@@ -1,49 +0,0 @@
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_MarkAllUserSettingsUnsynced_Request;
import server.logic.ws_protocol.JSON.messages.entyties.Net_MarkAllUserSettingsUnsynced_Response;
import server.logic.ws_protocol.JSON.utils.NetExceptionResponseFactory;
import server.logic.ws_protocol.WireCodes;
import shine.db.DbController;
import shine.db.dao.UserSettingsDAO;
import shine.db.dao.DmSyncOutboxDAO;
import server.sync.DmSyncWakeSignal;
import java.sql.Connection;
public class Net_MarkAllUserSettingsUnsynced_Handler implements JsonMessageHandler {
private static final Logger log = LoggerFactory.getLogger(Net_MarkAllUserSettingsUnsynced_Handler.class);
@Override
public Net_Response handle(Net_Request baseRequest, ConnectionContext ctx) {
Net_MarkAllUserSettingsUnsynced_Request req = (Net_MarkAllUserSettingsUnsynced_Request) baseRequest;
try (Connection c = DbController.getInstance().getConnection()) {
int updated;
int dmUpdated;
if (req.getLogin() == null || req.getLogin().isBlank()) {
updated = UserSettingsDAO.getInstance().markAllUnsynced(c);
dmUpdated = DmSyncOutboxDAO.getInstance().markAllUnsynced();
} else {
updated = UserSettingsDAO.getInstance().markAllUnsynced(c, req.getLogin().trim());
dmUpdated = DmSyncOutboxDAO.getInstance().markAllUnsynced(req.getLogin().trim());
}
DmSyncWakeSignal.request();
Net_MarkAllUserSettingsUnsynced_Response resp = new Net_MarkAllUserSettingsUnsynced_Response();
resp.setOp(req.getOp());
resp.setRequestId(req.getRequestId());
resp.setStatus(WireCodes.Status.OK);
resp.setUpdated(updated);
resp.setDmUpdated(dmUpdated);
return resp;
} catch (Exception e) {
log.error("MarkAllUserSettingsUnsynced failed", e);
return NetExceptionResponseFactory.error(req, WireCodes.Status.INTERNAL_ERROR, "INTERNAL_ERROR", "Внутренняя ошибка сервера");
}
}
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/Net_ReceiveIncomingMessage_Handler.java
-12
View File
@@ -9,7 +9,6 @@ import server.logic.ws_protocol.JSON.messages.entyties.Net_ReceiveIncomingMessag
import server.logic.ws_protocol.JSON.utils.NetExceptionResponseFactory;
import server.logic.ws_protocol.WireCodes;
import server.sync.DmDeliveryCoordinator;
import shine.db.dao.DmSyncOutboxDAO;
import shine.db.dao.SignedMessagesDAO;
import shine.db.entities.SignedMessageEntry;
@@ -70,17 +69,6 @@ public class Net_ReceiveIncomingMessage_Handler implements JsonMessageHandler {
if (stored == null || !Arrays.equals(stored.getRawBlock(), entry.getRawBlock())) {
return NetExceptionResponseFactory.error(req, 409, "STALE_MESSAGE_REVISION", "На сервере уже есть более новая ревизия сообщения");
}
boolean receivedFromRecipientPeer = DmDeliveryCoordinator.sourceIsOtherAccessServer(
incoming.toLogin, req.getSourceServerLogin());
DmSyncOutboxDAO.getInstance().upsert(
incoming.toLogin,
entry.getMessageKey(),
DmDeliveryIds.forEntry(entry),
null,
receivedFromRecipientPeer,
entry.getCreatedAtMs()
);
Net_ReceiveIncomingMessage_Response resp = new Net_ReceiveIncomingMessage_Response();
resp.setOp(req.getOp());
resp.setRequestId(req.getRequestId());
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/Net_SendMessagePair_Handler.java
+9 -32
View File
@@ -11,7 +11,6 @@ import server.logic.ws_protocol.WireCodes;
import server.sync.DmDeliveryCoordinator;
import server.sync.DmFederationService;
import shine.db.dao.DmDeliveryStateDAO;
import shine.db.dao.DmSyncOutboxDAO;
import shine.db.dao.SignedMessagesDAO;
import shine.db.entities.DmDeliveryStateEntry;
import shine.db.entities.SignedMessageEntry;
@@ -48,12 +47,10 @@ public class Net_SendMessagePair_Handler implements JsonMessageHandler {
SignedMessageEntry incomingEntry;
SignedMessageEntry outgoingEntry;
boolean fromPeer = !isBlank(req.getSourceServerLogin());
try {
String sourceApi = fromPeer ? "ReceiveOutcomingMessage" : "SendMessagePair";
String originSessionId = (ctx != null && !isBlank(ctx.getSessionId())) ? ctx.getSessionId() : null;
incomingEntry = SignedMessagesCore.toEntry(incoming, sourceApi, originSessionId);
outgoingEntry = SignedMessagesCore.toEntry(outgoing, sourceApi, originSessionId);
incomingEntry = SignedMessagesCore.toEntry(incoming, "SendMessagePair", originSessionId);
outgoingEntry = SignedMessagesCore.toEntry(outgoing, "SendMessagePair", originSessionId);
} catch (IllegalArgumentException ex) {
return NetExceptionResponseFactory.error(req, WireCodes.Status.BAD_REQUEST, ex.getMessage(), "Некорректный payload подтверждения");
}
@@ -96,42 +93,22 @@ public class Net_SendMessagePair_Handler implements JsonMessageHandler {
String eventId = DmDeliveryIds.forEntry(outgoingEntry);
long nowMs = System.currentTimeMillis();
long acceptedAtMs;
long signedAtMs = Math.max(outgoing.timeMs,
Math.max(outgoing.revisionTimeMs, outgoing.reencryptedAtMs));
acceptedAtMs = fromPeer && signedAtMs > 0L ? Math.min(nowMs, signedAtMs) : nowMs;
acceptedAtMs = nowMs;
long expiresAtMs = acceptedAtMs + 60L * 60L * 1000L;
int initialState = DmDeliveryStateEntry.ACCEPTED;
DmDeliveryStateEntry delivery = DmDeliveryStateDAO.getInstance().upsertPair(
outgoingEntry.getMessageKey(), eventId, outgoingEntry.getBaseKey(),
outgoingEntry.getFromLogin(), outgoingEntry.getToLogin(), incomingEntry.getMessageKey(),
acceptedAtMs, expiresAtMs, initialState,
null, null, fromPeer && initialState == DmDeliveryStateEntry.PENDING_NONE
null, null, false
);
DmSyncOutboxDAO.getInstance().upsert(
outgoingEntry.getFromLogin(), outgoingEntry.getMessageKey(), eventId,
incomingEntry.getMessageKey(), fromPeer, acceptedAtMs);
// Если этот же сервер также обслуживает получателя, его входящая копия
// имеет отдельный sync-флаг владельца-получателя.
if (DmDeliveryCoordinator.isLocalAccessServer(incomingEntry.getToLogin())) {
boolean incomingAlreadySynced = DmDeliveryCoordinator.sourceIsOtherAccessServer(
incomingEntry.getToLogin(), req.getSourceServerLogin());
DmSyncOutboxDAO.getInstance().upsert(
incomingEntry.getToLogin(), incomingEntry.getMessageKey(), DmDeliveryIds.forEntry(incomingEntry),
null, incomingAlreadySynced, acceptedAtMs);
}
if (delivery != null && pairStatus.applied()) {
if (fromPeer) {
DmDeliveryCoordinator.assistReceivedPairAsync(delivery.getEventId());
} else {
// Первая доставка выполняется до ответа клиенту. Два сервера
// получателя вызываются параллельно внутри координатора.
DmDeliveryCoordinator.processDueEntry(delivery);
delivery = DmDeliveryStateDAO.getInstance()
.getByOutgoingMessageKey(outgoingEntry.getMessageKey());
}
// Первая доставка на единственный access-сервер получателя
// выполняется до ответа клиенту.
DmDeliveryCoordinator.processDueEntry(delivery);
delivery = DmDeliveryStateDAO.getInstance()
.getByOutgoingMessageKey(outgoingEntry.getMessageKey());
}
Net_SendMessagePair_Response resp = new Net_SendMessagePair_Response();
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/Net_UserSettingsSyncBatch_Handler.java
-129
View File
@@ -1,129 +0,0 @@
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_UserSettingsSyncBatch_Request;
import server.logic.ws_protocol.JSON.messages.entyties.Net_UserSettingsSyncBatch_Response;
import server.logic.ws_protocol.JSON.utils.NetExceptionResponseFactory;
import server.logic.ws_protocol.WireCodes;
import shine.db.DbController;
import shine.db.dao.UserAccessServersCurrentDAO;
import shine.db.dao.UserSettingsDAO;
import shine.db.entities.UserSettingEntry;
import shine.db.entities.UserAccessServerRouteEntry;
import utils.config.AppConfig;
import java.sql.Connection;
import java.util.ArrayList;
import java.util.Base64;
import java.util.List;
public class Net_UserSettingsSyncBatch_Handler implements JsonMessageHandler {
private static final Logger log = LoggerFactory.getLogger(Net_UserSettingsSyncBatch_Handler.class);
private static final int DEFAULT_LIMIT = 500;
private static final int MAX_LIMIT = 1000;
private static final int DEFAULT_MAX_BYTES = 3_000_000;
private static final int MAX_BYTES_CAP = 5_000_000;
private static final String SERVER_LOGIN_CONFIG = "server.SHiNE.login";
@Override
public Net_Response handle(Net_Request baseRequest, ConnectionContext ctx) throws Exception {
Net_UserSettingsSyncBatch_Request req = (Net_UserSettingsSyncBatch_Request) baseRequest;
String ownerLogin = normalizeOriginal(req.getOwnerLogin());
if (ownerLogin == null) {
return NetExceptionResponseFactory.error(req, WireCodes.Status.BAD_REQUEST, "EMPTY_OWNER_LOGIN", "ownerLogin обязателен");
}
String ownServerLogin = normalize(AppConfig.getInstance().getParam(SERVER_LOGIN_CONFIG));
if (ownServerLogin == null) {
return NetExceptionResponseFactory.error(req, 500, "LOCAL_SERVER_NOT_CONFIGURED", "server.SHiNE.login не настроен");
}
if (!isLocalAccessServer(ownerLogin, ownServerLogin)) {
return NetExceptionResponseFactory.error(req, 403, "LOCAL_SERVER_NOT_ACCESS_SERVER", "Локальный сервер не является access-сервером пользователя");
}
int limit = clamp(req.getLimit() == null ? DEFAULT_LIMIT : req.getLimit(), 1, MAX_LIMIT);
int maxBytes = clamp(req.getMaxBytes() == null ? DEFAULT_MAX_BYTES : req.getMaxBytes(), 64_000, MAX_BYTES_CAP);
long afterTimeMs = Math.max(0L, req.getAfterTimeMs() == null ? 0L : req.getAfterTimeMs());
String afterSettingKey = req.getAfterSettingKey() == null ? "" : req.getAfterSettingKey().trim();
List<UserSettingEntry> batch;
try (Connection c = DbController.getInstance().getConnection()) {
batch = UserSettingsDAO.getInstance().listNewerThan(
c,
ownerLogin,
afterTimeMs,
afterSettingKey,
limit
);
}
Net_UserSettingsSyncBatch_Response resp = new Net_UserSettingsSyncBatch_Response();
resp.setOp(req.getOp());
resp.setRequestId(req.getRequestId());
resp.setStatus(WireCodes.Status.OK);
resp.setOwnerLogin(ownerLogin);
resp.setLimit(limit);
int rawBytes = 0;
List<Net_UserSettingsSyncBatch_Response.Item> items = new ArrayList<>();
for (UserSettingEntry entry : batch) {
Net_UserSettingsSyncBatch_Response.Item item = new Net_UserSettingsSyncBatch_Response.Item();
item.setLogin(entry.getLogin());
item.setSetting_type(entry.getSettingType());
item.setSetting_key(entry.getSettingKey());
item.setTime_ms(entry.getTimeMs());
item.setValue_text(entry.getValueText());
item.setValue_num(entry.getValueNum());
item.setClient_key(entry.getClientKey());
item.setSignature(entry.getSignature());
item.setSynced(entry.isSynced());
items.add(item);
rawBytes += String.valueOf(entry.getLogin()).length()
+ String.valueOf(entry.getSettingKey()).length()
+ String.valueOf(entry.getValueText()).length()
+ String.valueOf(entry.getClientKey() == null ? "" : entry.getClientKey()).length()
+ String.valueOf(entry.getSignature() == null ? "" : entry.getSignature()).length()
+ 64;
if (rawBytes > maxBytes) break;
resp.setNextTimeMs(entry.getTimeMs());
resp.setNextSettingKey(entry.getSettingKey());
}
resp.setRawBytes(rawBytes);
resp.setHasMore(batch.size() > items.size());
resp.setItems(items);
return resp;
}
private boolean isLocalAccessServer(String ownerLogin, String ownServerLogin) throws Exception {
try (Connection c = DbController.getInstance().getConnection()) {
for (UserAccessServerRouteEntry route : UserAccessServersCurrentDAO.getInstance().listByUserLogin(c, ownerLogin)) {
if (route == null || route.getServerLogin() == null) continue;
if (ownServerLogin.equals(normalize(route.getServerLogin()))) {
return true;
}
}
}
return false;
}
private static int clamp(int value, int min, int max) {
return Math.max(min, Math.min(max, value));
}
private static String normalize(String value) {
if (value == null) return null;
String s = value.trim().toLowerCase();
return s.isEmpty() ? null : s;
}
private static String normalizeOriginal(String value) {
if (value == null) return null;
String s = value.trim();
return s.isEmpty() ? null : s;
}
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/UserSettingsSyncApplySupport.java
-61
View File
@@ -1,61 +0,0 @@
package server.logic.ws_protocol.JSON.messages;
import server.logic.ws_protocol.JSON.messages.entyties.Net_UserSettingsSyncBatch_Response;
import shine.db.dao.UserSettingsDAO;
import shine.db.entities.UserSettingEntry;
import java.sql.Connection;
import java.util.ArrayList;
import java.util.List;
public final class UserSettingsSyncApplySupport {
private UserSettingsSyncApplySupport() {}
public static ApplyResult applySyncedItem(Connection c, String ownerLogin, Net_UserSettingsSyncBatch_Response.Item item) throws Exception {
if (item == null) return new ApplyResult(false, "empty_item");
String login = normalize(item.getLogin());
String key = normalize(item.getSetting_key());
if (login == null || key == null) return new ApplyResult(false, "bad_item");
UserSettingEntry entry = new UserSettingEntry(
login,
item.getSetting_type() == null ? 0 : item.getSetting_type(),
key,
item.getTime_ms() == null ? 0L : item.getTime_ms(),
item.getValue_text() == null ? "" : item.getValue_text(),
item.getValue_num() == null ? 0L : item.getValue_num(),
item.getClient_key(),
item.getSignature(),
true
);
int changed = UserSettingsDAO.getInstance().upsertIfNewer(c, entry);
return new ApplyResult(changed > 0, changed > 0 ? "applied" : "ignored");
}
public static List<UserSettingEntry> toEntries(List<Net_UserSettingsSyncBatch_Response.Item> items) {
List<UserSettingEntry> out = new ArrayList<>();
if (items == null) return out;
for (Net_UserSettingsSyncBatch_Response.Item item : items) {
if (item == null) continue;
out.add(new UserSettingEntry(
normalize(item.getLogin()),
item.getSetting_type() == null ? 0 : item.getSetting_type(),
normalize(item.getSetting_key()),
item.getTime_ms() == null ? 0L : item.getTime_ms(),
item.getValue_text() == null ? "" : item.getValue_text(),
item.getValue_num() == null ? 0L : item.getValue_num(),
item.getClient_key(),
item.getSignature(),
true
));
}
return out;
}
private static String normalize(String value) {
String s = String.valueOf(value == null ? "" : value).trim();
return s.isEmpty() ? null : s;
}
public record ApplyResult(boolean applied, String status) {}
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/entyties/Net_MarkAllUserSettingsUnsynced_Request.java
-10
View File
@@ -1,10 +0,0 @@
package server.logic.ws_protocol.JSON.messages.entyties;
import server.logic.ws_protocol.JSON.entyties.Net_Request;
public class Net_MarkAllUserSettingsUnsynced_Request extends Net_Request {
private String login;
public String getLogin() { return login; }
public void setLogin(String login) { this.login = login; }
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/entyties/Net_MarkAllUserSettingsUnsynced_Response.java
-13
View File
@@ -1,13 +0,0 @@
package server.logic.ws_protocol.JSON.messages.entyties;
import server.logic.ws_protocol.JSON.entyties.Net_Response;
public class Net_MarkAllUserSettingsUnsynced_Response extends Net_Response {
private Integer updated;
private Integer dmUpdated;
public Integer getUpdated() { return updated; }
public void setUpdated(Integer updated) { this.updated = updated; }
public Integer getDmUpdated() { return dmUpdated; }
public void setDmUpdated(Integer dmUpdated) { this.dmUpdated = dmUpdated; }
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/entyties/Net_SendMessagePair_Request.java
-3
View File
@@ -5,12 +5,9 @@ import server.logic.ws_protocol.JSON.entyties.Net_Request;
public class Net_SendMessagePair_Request extends Net_Request {
private String incomingBlobB64;
private String outgoingBlobB64;
private String sourceServerLogin;
public String getIncomingBlobB64() { return incomingBlobB64; }
public void setIncomingBlobB64(String incomingBlobB64) { this.incomingBlobB64 = incomingBlobB64; }
public String getOutgoingBlobB64() { return outgoingBlobB64; }
public void setOutgoingBlobB64(String outgoingBlobB64) { this.outgoingBlobB64 = outgoingBlobB64; }
public String getSourceServerLogin() { return sourceServerLogin; }
public void setSourceServerLogin(String sourceServerLogin) { this.sourceServerLogin = sourceServerLogin; }
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/entyties/Net_UserSettingsSyncBatch_Request.java
-26
View File
@@ -1,26 +0,0 @@
package server.logic.ws_protocol.JSON.messages.entyties;
import server.logic.ws_protocol.JSON.entyties.Net_Request;
public class Net_UserSettingsSyncBatch_Request extends Net_Request {
private String ownerLogin;
private Long afterTimeMs;
private String afterSettingKey;
private Integer limit;
private Integer maxBytes;
public String getOwnerLogin() { return ownerLogin; }
public void setOwnerLogin(String ownerLogin) { this.ownerLogin = ownerLogin; }
public Long getAfterTimeMs() { return afterTimeMs; }
public void setAfterTimeMs(Long afterTimeMs) { this.afterTimeMs = afterTimeMs; }
public String getAfterSettingKey() { return afterSettingKey; }
public void setAfterSettingKey(String afterSettingKey) { this.afterSettingKey = afterSettingKey; }
public Integer getLimit() { return limit; }
public void setLimit(Integer limit) { this.limit = limit; }
public Integer getMaxBytes() { return maxBytes; }
public void setMaxBytes(Integer maxBytes) { this.maxBytes = maxBytes; }
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/logic/ws_protocol/JSON/messages/entyties/Net_UserSettingsSyncBatch_Response.java
-62
View File
@@ -1,62 +0,0 @@
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_UserSettingsSyncBatch_Response extends Net_Response {
private String ownerLogin;
private Integer limit;
private Integer rawBytes;
private Boolean hasMore;
private Long nextTimeMs;
private String nextSettingKey;
private List<Item> items = new ArrayList<>();
public String getOwnerLogin() { return ownerLogin; }
public void setOwnerLogin(String ownerLogin) { this.ownerLogin = ownerLogin; }
public Integer getLimit() { return limit; }
public void setLimit(Integer limit) { this.limit = limit; }
public Integer getRawBytes() { return rawBytes; }
public void setRawBytes(Integer rawBytes) { this.rawBytes = rawBytes; }
public Boolean getHasMore() { return hasMore; }
public void setHasMore(Boolean hasMore) { this.hasMore = hasMore; }
public Long getNextTimeMs() { return nextTimeMs; }
public void setNextTimeMs(Long nextTimeMs) { this.nextTimeMs = nextTimeMs; }
public String getNextSettingKey() { return nextSettingKey; }
public void setNextSettingKey(String nextSettingKey) { this.nextSettingKey = nextSettingKey; }
public List<Item> getItems() { return items; }
public void setItems(List<Item> items) { this.items = items; }
public static class Item {
private String login;
private Integer setting_type;
private String setting_key;
private Long time_ms;
private String value_text;
private Long value_num;
private String client_key;
private String signature;
private Boolean synced;
public String getLogin() { return login; }
public void setLogin(String login) { this.login = login; }
public Integer getSetting_type() { return setting_type; }
public void setSetting_type(Integer setting_type) { this.setting_type = setting_type; }
public String getSetting_key() { return setting_key; }
public void setSetting_key(String setting_key) { this.setting_key = setting_key; }
public Long getTime_ms() { return time_ms; }
public void setTime_ms(Long time_ms) { this.time_ms = time_ms; }
public String getValue_text() { return value_text; }
public void setValue_text(String value_text) { this.value_text = value_text; }
public Long getValue_num() { return value_num; }
public void setValue_num(Long value_num) { this.value_num = value_num; }
public String getClient_key() { return client_key; }
public void setClient_key(String client_key) { this.client_key = client_key; }
public String getSignature() { return signature; }
public void setSignature(String signature) { this.signature = signature; }
public Boolean getSynced() { return synced; }
public void setSynced(Boolean synced) { this.synced = synced; }
}
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/DmDeliveryCoordinator.java
+37 -189
View File
@@ -4,7 +4,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import server.logic.ws_protocol.JSON.messages.DmDeliveryRealtime;
import shine.db.dao.DmDeliveryStateDAO;
import shine.db.dao.DmSyncOutboxDAO;
import shine.db.dao.SignedMessagesDAO;
import shine.db.dao.UserAccessServersCurrentDAO;
import shine.db.entities.DmDeliveryStateEntry;
@@ -14,21 +13,14 @@ import utils.config.AppConfig;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.Base64;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadFactory;
/**
* Координатор доставки исходящей DM-пары. Первая попытка выполняется до ответа
* клиенту, последующие — пятисекундным воркером.
* Координатор доставки исходящей DM-пары на единственный первый access-сервер
* получателя. Первая попытка выполняется до ответа клиенту, последующие —
* пятисекундным воркером.
*/
public final class DmDeliveryCoordinator {
private static final Logger log = LoggerFactory.getLogger(DmDeliveryCoordinator.class);
@@ -40,21 +32,10 @@ public final class DmDeliveryCoordinator {
60L * 60_000L
};
private static final String SERVER_LOGIN_CONFIG = "server.SHiNE.login";
private static final RemoteDmSyncClient REMOTE = new RemoteDmSyncClient();
private static final RemoteDmDeliveryClient REMOTE = new RemoteDmDeliveryClient();
private static final DmDeliveryStateDAO DELIVERY_DAO = DmDeliveryStateDAO.getInstance();
private static final DmSyncOutboxDAO OUTBOX_DAO = DmSyncOutboxDAO.getInstance();
private static final SignedMessagesDAO MESSAGES_DAO = SignedMessagesDAO.getInstance();
private static final UserAccessServersCurrentDAO ACCESS_DAO = UserAccessServersCurrentDAO.getInstance();
private static final ExecutorService ASSIST_EXECUTOR = new ThreadPoolExecutor(
2, 2, 0L, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<>(500),
daemonThreadFactory("dm-peer-assist"),
new ThreadPoolExecutor.DiscardPolicy());
private static final ExecutorService RECIPIENT_EXECUTOR = new ThreadPoolExecutor(
4, 16, 60L, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(1000),
daemonThreadFactory("dm-recipient-delivery"),
new ThreadPoolExecutor.CallerRunsPolicy());
private DmDeliveryCoordinator() {}
@@ -63,10 +44,6 @@ public final class DmDeliveryCoordinator {
}
public static void processDueEntry(DmDeliveryStateEntry snapshot) {
processDueEntry(snapshot, true);
}
private static void processDueEntry(DmDeliveryStateEntry snapshot, boolean allowInitialHandoff) {
if (snapshot == null) return;
try {
long now = System.currentTimeMillis();
@@ -79,27 +56,9 @@ public final class DmDeliveryCoordinator {
boolean finalAttempt = snapshot.getAttemptIndex() >= ATTEMPT_OFFSETS_MS.length - 1
|| now >= current.getDeliveryExpiresAtMs();
// Перед попытками на 5-й, 25-й и 60-й минутах сначала спрашиваем
// второй сервер отправителя: возможно, он уже доставил сообщение.
if (current.getDeliveryState() == DmDeliveryStateEntry.ACCEPTED
&& (snapshot.getAttemptIndex() >= 2 || finalAttempt)) {
current = acceptPeerDeliveryStatus(current);
if (current != null && current.isDelivered()) {
DmDeliveryRealtime.notifySender(current);
return;
}
}
DmDeliveryStateEntry afterAttempt = attemptRecipientRoutes(
DmDeliveryStateEntry afterAttempt = attemptPrimaryRecipientRoute(
current, finalAttempt ? null : nextAttemptAtMs);
// После первой попытки сразу передаём полную пару второму серверу
// отправителя старой операцией ReceiveOutcomingMessage.
if (allowInitialHandoff && snapshot.getAttemptIndex() == 0 && afterAttempt != null) {
afterAttempt = handoffPairToSenderPeer(afterAttempt);
}
if (finalAttempt && afterAttempt != null && !afterAttempt.isDelivered()) {
afterAttempt = DELIVERY_DAO.finishAtExpiry(afterAttempt.getEventId(), System.currentTimeMillis());
}
@@ -109,28 +68,18 @@ public final class DmDeliveryCoordinator {
}
}
public static void assistReceivedPairAsync(String eventId) {
if (eventId == null || eventId.isBlank()) return;
ASSIST_EXECUTOR.execute(() -> {
try {
DmDeliveryStateEntry row = DELIVERY_DAO.getByEventId(eventId);
if (row != null) processDueEntry(row, false);
} catch (Exception e) {
log.warn("DM peer assist failed: eventId={}", eventId, e);
}
});
}
private static DmDeliveryStateEntry attemptRecipientRoutes(
private static DmDeliveryStateEntry attemptPrimaryRecipientRoute(
DmDeliveryStateEntry current,
Long nextAttemptAtMs
) throws Exception {
if (current == null) return null;
List<UserAccessServerRouteEntry> routes = cappedRoutes(current.getToLogin());
String routesHash = routesHash(routes);
String alreadyDelivered = normalize(current.getDeliveredServerLogin());
List<String> acceptedLogins = new ArrayList<>();
if (alreadyDelivered != null) acceptedLogins.add(alreadyDelivered);
UserAccessServerRouteEntry route = primaryRoute(current.getToLogin());
String routesHash = routesHash(route);
if (route == null) {
return DELIVERY_DAO.updateAfterAttempt(
current.getEventId(), DmDeliveryStateEntry.ACCEPTED, null,
routesHash, nextAttemptAtMs, "RECIPIENT_ACCESS_SERVER_NOT_FOUND", System.currentTimeMillis());
}
SignedMessageEntry incoming = MESSAGES_DAO.getByMessageKey(current.getIncomingMessageKey());
if (incoming == null || incoming.getRawBlock() == null) {
@@ -138,138 +87,51 @@ public final class DmDeliveryCoordinator {
current.getEventId(), current.getDeliveryState(), current.getDeliveredServerLogin(),
routesHash, nextAttemptAtMs, "INCOMING_BLOB_NOT_FOUND", System.currentTimeMillis());
}
String incomingBlobB64 = Base64.getEncoder().encodeToString(incoming.getRawBlock());
String ownServerLogin = ownServerLogin();
java.util.concurrent.CompletionService<RouteAttempt> completion =
new java.util.concurrent.ExecutorCompletionService<>(RECIPIENT_EXECUTOR);
int submitted = 0;
for (UserAccessServerRouteEntry route : routes) {
String routeLogin = normalize(route.getServerLogin());
if (routeLogin == null || acceptedLogins.contains(routeLogin)) continue;
completion.submit(() -> {
try {
if (!routeLogin.equals(ownServerLogin)) {
REMOTE.receiveIncomingMessage(
routeLogin, route.getServerUrl(), incomingBlobB64, ownServerLogin);
}
return new RouteAttempt(routeLogin, null);
} catch (Exception e) {
log.info("DM recipient server unavailable: messageKey={} server={}",
current.getOutgoingMessageKey(), routeLogin);
return new RouteAttempt(routeLogin, compactError(e));
}
});
submitted++;
}
String lastError = null;
for (int i = 0; i < submitted && acceptedLogins.isEmpty(); i++) {
RouteAttempt result = completion.take().get();
if (result.error() == null) {
acceptedLogins.add(result.serverLogin());
} else {
lastError = result.error();
}
}
int nextState = acceptedLogins.isEmpty()
? DmDeliveryStateEntry.ACCEPTED
: DmDeliveryStateEntry.DELIVERED;
String oneLogin = acceptedLogins.isEmpty() ? null : acceptedLogins.get(0);
return DELIVERY_DAO.updateAfterAttempt(
current.getEventId(), nextState, oneLogin, routesHash,
nextAttemptAtMs, lastError, System.currentTimeMillis());
}
private static DmDeliveryStateEntry handoffPairToSenderPeer(DmDeliveryStateEntry current) throws Exception {
UserAccessServerRouteEntry peer = senderPeer(current.getFromLogin());
if (peer == null) return current;
SignedMessageEntry incoming = MESSAGES_DAO.getByMessageKey(current.getIncomingMessageKey());
SignedMessageEntry outgoing = MESSAGES_DAO.getByMessageKey(current.getOutgoingMessageKey());
if (incoming == null || outgoing == null) return current;
String routeLogin = normalize(route.getServerLogin());
try {
REMOTE.sendMessagePair(
peer.getServerLogin(),
peer.getServerUrl(),
Base64.getEncoder().encodeToString(incoming.getRawBlock()),
Base64.getEncoder().encodeToString(outgoing.getRawBlock()),
ownServerLogin());
OUTBOX_DAO.markSynced(current.getFromLogin(), current.getEventId());
} catch (Exception e) {
log.info("DM sender peer unavailable: eventId={} server={}", current.getEventId(), peer.getServerLogin());
}
return current;
}
private static DmDeliveryStateEntry acceptPeerDeliveryStatus(DmDeliveryStateEntry current) throws Exception {
UserAccessServerRouteEntry peer = senderPeer(current.getFromLogin());
if (peer == null) return current;
try {
RemoteDmSyncClient.RemoteDeliveryStatus remote = REMOTE.getDmDeliveryStatus(
peer.getServerLogin(), peer.getServerUrl(), current.getOutgoingMessageKey());
if (remote.known() && remote.delivered()) {
return DELIVERY_DAO.markDeliveredFromPeer(current.getEventId(), System.currentTimeMillis());
if (!routeLogin.equals(ownServerLogin())) {
REMOTE.receiveIncomingMessage(
routeLogin,
route.getServerUrl(),
Base64.getEncoder().encodeToString(incoming.getRawBlock()),
ownServerLogin());
}
return current;
return DELIVERY_DAO.updateAfterAttempt(
current.getEventId(), DmDeliveryStateEntry.DELIVERED, routeLogin,
routesHash, nextAttemptAtMs, null, System.currentTimeMillis());
} catch (Exception e) {
log.info("DM delivery status unavailable: eventId={} server={}", current.getEventId(), peer.getServerLogin());
return current;
log.info("DM recipient server unavailable: messageKey={} server={}",
current.getOutgoingMessageKey(), routeLogin);
return DELIVERY_DAO.updateAfterAttempt(
current.getEventId(), DmDeliveryStateEntry.ACCEPTED, null,
routesHash, nextAttemptAtMs, compactError(e), System.currentTimeMillis());
}
}
private static UserAccessServerRouteEntry senderPeer(String senderLogin) throws Exception {
String own = ownServerLogin();
for (UserAccessServerRouteEntry route : cappedRoutes(senderLogin)) {
String routeLogin = normalize(route.getServerLogin());
if (routeLogin != null && !routeLogin.equals(own)) return route;
}
return null;
}
public static boolean isLocalAccessServer(String ownerLogin) throws Exception {
UserAccessServerRouteEntry route = primaryRoute(ownerLogin);
String own = ownServerLogin();
if (own == null) return false;
for (UserAccessServerRouteEntry route : cappedRoutes(ownerLogin)) {
if (own.equals(normalize(route.getServerLogin()))) return true;
}
return false;
}
public static boolean sourceIsOtherAccessServer(String ownerLogin, String sourceServerLogin) throws Exception {
String source = normalize(sourceServerLogin);
String own = ownServerLogin();
if (source == null || source.equals(own)) return false;
for (UserAccessServerRouteEntry route : cappedRoutes(ownerLogin)) {
if (source.equals(normalize(route.getServerLogin()))) return true;
}
return false;
return route != null && own != null && own.equals(normalize(route.getServerLogin()));
}
public static String currentServerLogin() {
return ownServerLogin();
}
private static List<UserAccessServerRouteEntry> cappedRoutes(String ownerLogin) throws Exception {
Map<String, UserAccessServerRouteEntry> unique = new LinkedHashMap<>();
private static UserAccessServerRouteEntry primaryRoute(String ownerLogin) throws Exception {
for (UserAccessServerRouteEntry route : ACCESS_DAO.listByUserLogin(ownerLogin)) {
if (route == null || route.getServerUrl() == null || route.getServerUrl().isBlank()) continue;
String login = normalize(route.getServerLogin());
if (login == null) continue;
unique.putIfAbsent(login, route);
if (unique.size() == 2) break;
if (normalize(route.getServerLogin()) != null) return route;
}
return new ArrayList<>(unique.values());
return null;
}
private static String routesHash(List<UserAccessServerRouteEntry> routes) throws Exception {
List<String> logins = new ArrayList<>();
for (UserAccessServerRouteEntry route : routes) {
String login = normalize(route.getServerLogin());
if (login != null) logins.add(login);
}
logins.sort(String::compareTo);
private static String routesHash(UserAccessServerRouteEntry route) throws Exception {
String normalized = route == null ? null : normalize(route.getServerLogin());
String login = normalized == null ? "" : normalized;
MessageDigest digest = MessageDigest.getInstance("SHA-256");
byte[] hash = digest.digest(String.join("\n", logins).getBytes(StandardCharsets.UTF_8));
byte[] hash = digest.digest(login.getBytes(StandardCharsets.UTF_8));
StringBuilder out = new StringBuilder(hash.length * 2);
for (byte b : hash) out.append(String.format("%02x", b & 0xff));
return out.toString();
@@ -280,8 +142,6 @@ public final class DmDeliveryCoordinator {
return createdAtMs + ATTEMPT_OFFSETS_MS[nextAttemptIndex];
}
private record RouteAttempt(String serverLogin, String error) {}
private static String ownServerLogin() {
return normalize(AppConfig.getInstance().getParam(SERVER_LOGIN_CONFIG));
}
@@ -296,16 +156,4 @@ public final class DmDeliveryCoordinator {
String text = String.valueOf(e == null ? "unknown" : e.getMessage());
return text.length() <= 500 ? text : text.substring(0, 500);
}
private static ThreadFactory daemonThreadFactory(String prefix) {
return new ThreadFactory() {
private int sequence;
@Override
public synchronized Thread newThread(Runnable r) {
Thread thread = new Thread(r, prefix + "-" + (++sequence));
thread.setDaemon(true);
return thread;
}
};
}
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/DmFederationService.java
+28 -73
View File
@@ -8,63 +8,17 @@ import utils.config.AppConfig;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
/** Доставка DM-tombstone на единственные access-серверы обеих сторон. */
public final class DmFederationService {
private static final Logger log = LoggerFactory.getLogger(DmFederationService.class);
private static final String CONFIG_KEY = "server.SHiNE.login";
private static final RemoteDmSyncClient REMOTE = new RemoteDmSyncClient();
private static final RemoteDmDeliveryClient REMOTE = new RemoteDmDeliveryClient();
private DmFederationService() {}
public static void fanOutPair(String fromLogin, String toLogin, String incomingBlobB64, String outgoingBlobB64) {
try {
Map<String, UserAccessServerRouteEntry> senderRoutes =
routesByLogin(UserAccessServersCurrentDAO.getInstance().listByUserLogin(fromLogin));
Map<String, UserAccessServerRouteEntry> recipientRoutes =
routesByLogin(UserAccessServersCurrentDAO.getInstance().listByUserLogin(toLogin));
String ownServerLogin = ownServerLogin();
for (UserAccessServerRouteEntry route : senderRoutes.values()) {
if (isOwnServer(route, ownServerLogin)) continue;
REMOTE.sendMessagePair(
route.getServerLogin(), route.getServerUrl(),
incomingBlobB64, outgoingBlobB64, ownServerLogin);
}
for (UserAccessServerRouteEntry route : recipientRoutes.values()) {
if (isOwnServer(route, ownServerLogin)) continue;
if (senderRoutes.containsKey(normalize(route.getServerLogin()))) continue;
REMOTE.receiveIncomingMessage(
route.getServerLogin(), route.getServerUrl(), incomingBlobB64, ownServerLogin);
}
} catch (Exception e) {
log.warn("DM federation pair fan-out failed: from={} to={}", fromLogin, toLogin, e);
}
}
public static void fanOutIncomingToRecipientAccessServers(
String toLogin,
String incomingBlobB64,
String sourceServerLogin
) {
try {
Map<String, UserAccessServerRouteEntry> recipientRoutes =
routesByLogin(UserAccessServersCurrentDAO.getInstance().listByUserLogin(toLogin));
String ownServerLogin = ownServerLogin();
String normalizedSource = normalize(sourceServerLogin);
for (UserAccessServerRouteEntry route : recipientRoutes.values()) {
String routeLogin = normalize(route.getServerLogin());
if (routeLogin == null) continue;
if (ownServerLogin != null && ownServerLogin.equalsIgnoreCase(routeLogin)) continue;
if (normalizedSource != null && normalizedSource.equalsIgnoreCase(routeLogin)) continue;
REMOTE.receiveIncomingMessage(
route.getServerLogin(), route.getServerUrl(), incomingBlobB64, ownServerLogin);
}
} catch (Exception e) {
log.warn("DM federation incoming relay failed: to={}", toLogin, e);
}
}
public static void fanOutDeleteMessage(String fromLogin, String toLogin, int messageType, String blobB64) {
fanOutSingleDelete(fromLogin, toLogin, messageType, blobB64, true);
}
@@ -73,12 +27,17 @@ public final class DmFederationService {
fanOutSingleDelete(fromLogin, toLogin, messageType, blobB64, false);
}
private static void fanOutSingleDelete(String fromLogin, String toLogin, int messageType, String blobB64, boolean oneMessageDelete) {
private static void fanOutSingleDelete(
String fromLogin,
String toLogin,
int messageType,
String blobB64,
boolean oneMessageDelete
) {
try {
Map<String, UserAccessServerRouteEntry> routes = routesByLogin(
UserAccessServersCurrentDAO.getInstance().listByUserLogin(fromLogin),
UserAccessServersCurrentDAO.getInstance().listByUserLogin(toLogin)
);
Map<String, UserAccessServerRouteEntry> routes = new LinkedHashMap<>();
putPrimaryRoute(routes, fromLogin);
putPrimaryRoute(routes, toLogin);
String ownServerLogin = ownServerLogin();
for (UserAccessServerRouteEntry route : routes.values()) {
if (isOwnServer(route, ownServerLogin)) continue;
@@ -89,33 +48,29 @@ public final class DmFederationService {
}
}
} catch (Exception e) {
log.warn("DM federation delete fan-out failed: from={} to={} type={}",
log.warn("DM delete delivery failed: from={} to={} type={}",
fromLogin, toLogin, messageType, e);
}
}
@SafeVarargs
private static Map<String, UserAccessServerRouteEntry> routesByLogin(
List<UserAccessServerRouteEntry>... routeLists
) {
Map<String, UserAccessServerRouteEntry> out = new LinkedHashMap<>();
for (List<UserAccessServerRouteEntry> routeList : routeLists) {
for (UserAccessServerRouteEntry route : routeList) {
if (route == null) continue;
String login = normalize(route.getServerLogin());
String address = route.getServerUrl() == null ? "" : route.getServerUrl().trim();
if (login == null || address.isBlank()) continue;
out.putIfAbsent(login, route);
}
private static void putPrimaryRoute(Map<String, UserAccessServerRouteEntry> out, String ownerLogin)
throws Exception {
List<UserAccessServerRouteEntry> routes =
UserAccessServersCurrentDAO.getInstance().listByUserLogin(ownerLogin);
for (UserAccessServerRouteEntry route : routes) {
if (route == null) continue;
String login = normalize(route.getServerLogin());
String address = route.getServerUrl() == null ? "" : route.getServerUrl().trim();
if (login == null || address.isBlank()) continue;
out.putIfAbsent(login, route);
return;
}
return out;
}
private static boolean isOwnServer(UserAccessServerRouteEntry route, String ownServerLogin) {
return ownServerLogin != null
&& route != null
&& route.getServerLogin() != null
&& ownServerLogin.equalsIgnoreCase(route.getServerLogin());
&& ownServerLogin.equals(normalize(route.getServerLogin()));
}
private static String ownServerLogin() {
@@ -124,7 +79,7 @@ public final class DmFederationService {
private static String normalize(String value) {
if (value == null) return null;
String s = value.trim().toLowerCase();
return s.isEmpty() ? null : s;
String normalized = value.trim().toLowerCase(Locale.ROOT);
return normalized.isEmpty() ? null : normalized;
}
}
SHiNE-server/shine-server-net-protocol/src/main/java/server/sync/RemoteDmDeliveryClient.java
+74
View File
@@ -0,0 +1,74 @@
package server.sync;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
/** Клиент межсерверной доставки DM на единственный access-сервер получателя. */
public final class RemoteDmDeliveryClient {
private static final ObjectMapper MAPPER = new ObjectMapper();
public void receiveIncomingMessage(
String targetServerLogin,
String serverAddressRaw,
String incomingBlobB64,
String sourceServerLogin
) throws Exception {
String incomingJson = MAPPER.writeValueAsString(incomingBlobB64);
String sourceServerLoginJson = toOptionalJsonField("sourceServerLogin", sourceServerLogin);
JsonNode response = send(targetServerLogin, serverAddressRaw, """
{
"op":"ReceiveIncomingMessage",
"requestId":%s,
"payload":{
"incomingBlobB64":%s%s
}
}
""".formatted("%s", incomingJson, sourceServerLoginJson));
ensureOk("ReceiveIncomingMessage", response);
}
public void deleteMessage(String targetServerLogin, String serverAddressRaw, String blobB64) throws Exception {
String blobJson = MAPPER.writeValueAsString(blobB64);
JsonNode response = send(targetServerLogin, serverAddressRaw, """
{
"op":"DeleteMessage",
"requestId":%s,
"payload":{"blobB64":%s}
}
""".formatted("%s", blobJson));
ensureOk("DeleteMessage", response);
}
public void deleteConversation(String targetServerLogin, String serverAddressRaw, String blobB64) throws Exception {
String blobJson = MAPPER.writeValueAsString(blobB64);
JsonNode response = send(targetServerLogin, serverAddressRaw, """
{
"op":"DeleteConversation",
"requestId":%s,
"payload":{"blobB64":%s}
}
""".formatted("%s", blobJson));
ensureOk("DeleteConversation", response);
}
private JsonNode send(String targetServerLogin, String serverAddressRaw, String jsonTemplate) throws Exception {
return ServerConnectionPool.getInstance().request(
targetServerLogin,
serverAddressRaw,
jsonTemplate,
ServerConnectionPool.Priority.REALTIME);
}
private String toOptionalJsonField(String fieldName, String value) throws Exception {
if (value == null || value.isBlank()) return "";
return ",\n \"" + fieldName + "\":" + MAPPER.writeValueAsString(value.trim());
}
private void ensureOk(String op, JsonNode response) {
int status = response.path("status").asInt(500);
if (status >= 200 && status < 300) return;
String code = response.path("code").asText("");
if (code.isBlank()) code = response.path("error").asText("");
throw new IllegalStateException(op + " failed: status=" + status + " code=" + code);
}
}
SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/storage/postgres/PostgresStorageRepository.java
+2 -2
View File
@@ -1091,7 +1091,7 @@ public final class PostgresStorageRepository
ELSE u.access_servers_json::jsonb
END
) WITH ORDINALITY AS access_server(login_value, ord)
WHERE ord <= 2
WHERE ord = 1
) AS access_server
JOIN solana_user_pda_current s
ON LOWER(s.login) = LOWER(btrim(access_server.login_value))
@@ -1138,7 +1138,7 @@ public final class PostgresStorageRepository
ELSE u.access_servers_json::jsonb
END
) WITH ORDINALITY AS access_server(login_value, ord)
WHERE ord <= 2
WHERE ord = 1
) AS access_server
WHERE LOWER(btrim(access_server.login_value)) = LOWER(p_server_login)
LOOP
SHiNE-server/src/main/java/server/sync/PeriodicDmDeliveryService.java
+73
View File
@@ -0,0 +1,73 @@
package server.sync;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import shine.db.entities.DmDeliveryStateEntry;
import utils.config.AppConfig;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/** Пятисекундный воркер повторной доставки DM получателю. */
public final class PeriodicDmDeliveryService {
private static final Logger log = LoggerFactory.getLogger(PeriodicDmDeliveryService.class);
private static final AtomicBoolean STARTED = new AtomicBoolean(false);
private static final ScheduledExecutorService SCHEDULER =
Executors.newSingleThreadScheduledExecutor(daemonThreadFactory("dm-worker-dispatcher"));
private static final ThreadPoolExecutor DELIVERY_EXECUTOR = new ThreadPoolExecutor(
4, 4, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(500),
daemonThreadFactory("dm-delivery"), new ThreadPoolExecutor.DiscardPolicy());
private PeriodicDmDeliveryService() {}
public static void startOrLog() {
if (!isEnabled()) {
log.info("DM delivery worker disabled by dm.delivery.enabled=false");
return;
}
if (!STARTED.compareAndSet(false, true)) return;
long pollSeconds = configLong("dm.worker.pollSeconds", 5L, 1L, 60L);
SCHEDULER.scheduleWithFixedDelay(
PeriodicDmDeliveryService::tickSafe, 0L, pollSeconds, TimeUnit.SECONDS);
log.info("DM delivery worker scheduled every {} seconds", pollSeconds);
}
private static void tickSafe() {
try {
int limit = (int) configLong("dm.worker.dueLimit", 100L, 1L, 1000L);
for (DmDeliveryStateEntry row : DmDeliveryCoordinator.listDue(limit)) {
DELIVERY_EXECUTOR.execute(() -> DmDeliveryCoordinator.processDueEntry(row));
}
} catch (Exception e) {
log.error("DM delivery dispatcher failed", e);
}
}
private static boolean isEnabled() {
String raw = AppConfig.getInstance().getParam("dm.delivery.enabled");
return raw == null || raw.isBlank() || Boolean.parseBoolean(raw.trim());
}
private static long configLong(String key, long defaultValue, long min, long max) {
String raw = AppConfig.getInstance().getParam(key);
if (raw == null || raw.isBlank()) return defaultValue;
try { return Math.max(min, Math.min(max, Long.parseLong(raw.trim()))); }
catch (Exception ignored) { return defaultValue; }
}
private static ThreadFactory daemonThreadFactory(String prefix) {
return new ThreadFactory() {
private int sequence;
@Override public synchronized Thread newThread(Runnable r) {
Thread t = new Thread(r, prefix + "-" + (++sequence));
t.setDaemon(true);
return t;
}
};
}
}
SHiNE-server/src/main/java/server/sync/PeriodicUserSettingsSyncService.java
-256
View File
@@ -1,256 +0,0 @@
package server.sync;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import server.logic.ws_protocol.JSON.messages.DmSyncApplySupport;
import shine.db.dao.UserAccessServersCurrentDAO;
import shine.db.dao.UserSettingsDAO;
import shine.db.dao.UserSettingsSyncPeerStateDAO;
import shine.db.entities.UserAccessServerRouteEntry;
import shine.db.entities.UserSettingEntry;
import shine.db.entities.UserSettingsSyncPeerStateEntry;
import utils.config.AppConfig;
import java.sql.Connection;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.ArrayList;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
public final class PeriodicUserSettingsSyncService {
private static final Logger log = LoggerFactory.getLogger(PeriodicUserSettingsSyncService.class);
private static final String SERVER_LOGIN_CONFIG = "server.SHiNE.login";
private static final AtomicBoolean STARTED = new AtomicBoolean(false);
private static final RemoteUserSettingsSyncClient REMOTE = new RemoteUserSettingsSyncClient();
private static final RemoteDmSyncClient DM_REMOTE = new RemoteDmSyncClient();
private static final UserAccessServersCurrentDAO ACCESS_DAO = UserAccessServersCurrentDAO.getInstance();
private static final UserSettingsDAO SETTINGS_DAO = UserSettingsDAO.getInstance();
private static final UserSettingsSyncPeerStateDAO STATE_DAO = UserSettingsSyncPeerStateDAO.getInstance();
private static final ScheduledExecutorService EXECUTOR = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r, "periodic-user-settings-sync");
t.setDaemon(true);
return t;
}
});
private PeriodicUserSettingsSyncService() {}
public static void startOrLog() {
if (!isEnabled()) {
log.info("Periodic user settings sync disabled by user.settings.sync.enabled=false");
return;
}
if (!STARTED.compareAndSet(false, true)) return;
long initialDelaySec = configLong("user.settings.sync.initialDelaySeconds", 90L, 0L, 3600L);
long periodHours = configLong("user.settings.sync.periodHours", 6L, 1L, 168L);
EXECUTOR.scheduleWithFixedDelay(
PeriodicUserSettingsSyncService::runCycleSafe,
initialDelaySec,
TimeUnit.HOURS.toSeconds(periodHours),
TimeUnit.SECONDS
);
EXECUTOR.scheduleWithFixedDelay(
PeriodicUserSettingsSyncService::runRequestedCycleSafe,
5L, 5L, TimeUnit.SECONDS);
log.info("Periodic user settings sync scheduled: first run in {} seconds, then every {} hours", initialDelaySec, periodHours);
}
private static void runCycleSafe() {
try {
runCycle();
} catch (Exception e) {
log.error("Periodic user settings sync failed unexpectedly", e);
}
}
private static void runRequestedCycleSafe() {
if (DmSyncWakeSignal.consume()) runCycleSafe();
}
private static void runCycle() throws Exception {
String ownServerLogin = normalize(AppConfig.getInstance().getParam(SERVER_LOGIN_CONFIG));
if (ownServerLogin == null) {
log.warn("Periodic user settings sync skipped: {} is empty", SERVER_LOGIN_CONFIG);
return;
}
List<String> ownersRaw = ACCESS_DAO.listUserLoginsByServerLogin(ownServerLogin);
Set<String> owners = new LinkedHashSet<>(ownersRaw);
if (owners.isEmpty()) {
log.info("Periodic user settings sync skipped: no local access-server users for {}", ownServerLogin);
return;
}
int syncedPeers = 0;
int appliedItems = 0;
int pushedItems = 0;
int appliedDmItems = 0;
for (String ownerLogin : owners) {
List<UserAccessServerRouteEntry> routes = ACCESS_DAO.listByUserLogin(ownerLogin);
for (UserAccessServerRouteEntry route : routes) {
if (route == null) continue;
String remoteLogin = normalize(route.getServerLogin());
String remoteUrl = route.getServerUrl();
if (remoteLogin == null || remoteUrl == null || remoteUrl.isBlank()) continue;
if (remoteLogin.equals(ownServerLogin)) continue;
try {
SyncStats stats = syncOwnerWithRemote(ownerLogin, route);
appliedItems += stats.applied();
pushedItems += stats.pushed();
appliedDmItems += stats.appliedDm();
syncedPeers++;
} catch (Exception e) {
STATE_DAO.updateError(ownerLogin, route.getServerLogin(), remoteUrl, String.valueOf(e));
log.warn("Periodic user settings sync peer failed: owner={} remoteServer={} reason={}",
ownerLogin, route.getServerLogin(), String.valueOf(e));
}
}
}
log.info("Periodic access-data sync finished: owners={} peers={} settingsApplied={} settingsPushed={} dmApplied={}",
owners.size(), syncedPeers, appliedItems, pushedItems, appliedDmItems);
}
private static SyncStats syncOwnerWithRemote(String ownerLogin, UserAccessServerRouteEntry route) throws Exception {
int limit = (int) configLong("user.settings.sync.batchLimit", 500L, 1L, 1000L);
int maxBytes = (int) configLong("user.settings.sync.batchMaxBytes", 3_000_000L, 64_000L, 5_000_000L);
int maxPages = (int) configLong("user.settings.sync.maxPagesPerPeer", 50L, 1L, 500L);
UserSettingsSyncPeerStateEntry state;
try (Connection c = getDbConnection()) {
state = STATE_DAO.getOrCreate(c, ownerLogin, route.getServerLogin(), route.getServerUrl());
}
long cursorTimeMs = state.getCursorTimeMs();
String cursorSettingKey = state.getCursorSettingKey() == null ? "" : state.getCursorSettingKey();
int applied = 0;
int pushed = 0;
boolean bootstrapCompleted = false;
int appliedDm;
try (RemoteSyncSession session = new RemoteSyncSession(route.getServerLogin(), route.getServerUrl())) {
for (int page = 0; page < maxPages; page++) {
RemoteUserSettingsSyncClient.RemoteUserSettingsBatch batch = REMOTE.userSettingsSyncBatch(
session,
ownerLogin,
cursorTimeMs,
cursorSettingKey,
limit,
maxBytes
);
try (Connection c = getDbConnection()) {
for (RemoteUserSettingsSyncClient.RemoteUserSettingsItem item : batch.items()) {
UserSettingEntry entry = new UserSettingEntry(
item.login(),
item.settingType(),
item.settingKey(),
item.timeMs(),
item.valueText(),
item.valueNum(),
item.clientKey(),
item.signature(),
true
);
int changed = SETTINGS_DAO.upsertIfNewer(c, entry);
if (changed > 0) applied++;
}
}
cursorTimeMs = Math.max(cursorTimeMs, batch.nextTimeMs());
cursorSettingKey = batch.nextSettingKey() == null ? "" : batch.nextSettingKey();
bootstrapCompleted = !batch.hasMore();
STATE_DAO.updateSuccess(ownerLogin, route.getServerLogin(), route.getServerUrl(), cursorTimeMs, cursorSettingKey, bootstrapCompleted);
if (!batch.hasMore() || batch.items().isEmpty()) break;
}
try (Connection c = getDbConnection()) {
List<UserSettingEntry> unsynced = SETTINGS_DAO.listUnsyncedByLogin(c, ownerLogin, limit);
for (UserSettingEntry entry : unsynced) {
REMOTE.upsertUserSetting(session, entry, true);
SETTINGS_DAO.markSynced(c, entry.getLogin(), entry.getSettingType(), entry.getSettingKey());
pushed++;
}
}
int dmLimit = (int) configLong("dm.sync.batchLimit", 200L, 1L, 500L);
int dmMaxBytes = (int) configLong("dm.sync.batchMaxBytes", 3_000_000L, 64_000L, 5_000_000L);
int dmMaxPages = (int) configLong("dm.sync.maxPagesPerPeer", 20L, 1L, 500L);
appliedDm = syncDmInSameSession(session, ownerLogin, dmLimit, dmMaxBytes, dmMaxPages);
}
if (!bootstrapCompleted) {
log.info("Periodic user settings sync peer paused by page limit: owner={} remoteServer={} maxPages={}",
ownerLogin, route.getServerLogin(), maxPages);
}
return new SyncStats(applied, pushed, appliedDm);
}
private static int syncDmInSameSession(
RemoteSyncSession session, String ownerLogin, int limit, int maxBytes, int maxPages
) throws Exception {
long cursorMs = 0L;
String cursorKey = "";
List<String> acknowledgements = new ArrayList<>();
int applied = 0;
for (int page = 0; page < maxPages; page++) {
RemoteDmSyncClient.RemoteDmBatch batch = DM_REMOTE.dmSyncBatch(
session, ownerLogin, cursorMs, cursorKey, Math.min(limit, 500), maxBytes, acknowledgements);
acknowledgements = new ArrayList<>();
for (RemoteDmSyncClient.RemoteDmItem item : batch.items()) {
if (item == null || item.syncId() == null || item.syncId().isBlank()) continue;
DmSyncApplySupport.applySyncedItem(ownerLogin, item.syncId(), item.blobsB64());
acknowledgements.add(item.syncId());
applied++;
}
cursorMs = batch.nextStoredAtMs();
cursorKey = batch.nextMessageKey() == null ? "" : batch.nextMessageKey();
if (!batch.hasMore() || batch.items().isEmpty()) {
if (!acknowledgements.isEmpty()) {
DM_REMOTE.dmSyncBatch(session, ownerLogin, 0L, "",
Math.min(limit, 500), maxBytes, acknowledgements);
}
break;
}
}
return applied;
}
private static java.sql.Connection getDbConnection() throws Exception {
return shine.db.DbController.getInstance().getConnection();
}
private static boolean isEnabled() {
String raw = AppConfig.getInstance().getParam("user.settings.sync.enabled");
return raw == null || raw.isBlank() || Boolean.parseBoolean(raw.trim());
}
private static long configLong(String key, long defaultValue, long min, long max) {
String raw = AppConfig.getInstance().getParam(key);
if (raw == null || raw.isBlank()) return defaultValue;
try {
long parsed = Long.parseLong(raw.trim());
return Math.max(min, Math.min(max, parsed));
} catch (Exception ignored) {
return defaultValue;
}
}
private static String normalize(String value) {
if (value == null) return null;
String s = value.trim().toLowerCase(Locale.ROOT);
return s.isEmpty() ? null : s;
}
private record SyncStats(int applied, int pushed, int appliedDm) {}
}
SHiNE-server/src/main/java/server/ws/WsServer.java
+2 -4
View File
@@ -8,8 +8,7 @@ import org.slf4j.LoggerFactory;
import server.debug.DebugApiConfigurator;
import server.sync.BlockchainResyncRecoveryOnStartup;
import server.sync.PeriodicBlockchainSyncService;
import server.sync.PeriodicDmSyncService;
import server.sync.PeriodicUserSettingsSyncService;
import server.sync.PeriodicDmDeliveryService;
import server.sync.SolanaUsersSyncStartupService;
import server.sync.SyncServersBootstrapService;
import server.sync.ServerConnectionPool;
@@ -109,8 +108,7 @@ public final class WsServer {
Runtime.getRuntime().addShutdownHook(new Thread(
() -> ServerConnectionPool.getInstance().close(),
"server-connection-pool-shutdown"));
PeriodicDmSyncService.startOrLog();
PeriodicUserSettingsSyncService.startOrLog();
PeriodicDmDeliveryService.startOrLog();
server.join();
}
}
SHiNE-server/src/main/resources/application.properties
+2 -5
View File
@@ -34,13 +34,10 @@ sync.importUserProfileFromPartner.enabled=false
# ------------------------------------------------------------
server.version=${projectVersion}
# Доставка и межсерверная синхронизация личных сообщений.
dm.sync.enabled=true
# Доставка личных сообщений на единственный access-сервер получателя.
dm.delivery.enabled=true
dm.worker.pollSeconds=5
dm.worker.dueLimit=100
dm.sync.batchLimit=200
dm.sync.batchMaxBytes=3000000
dm.sync.maxPagesPerPeer=20
server.pool.pingIdleSeconds=120
server.pool.pongTimeoutSeconds=15
server.pool.requestTimeoutSeconds=12
bundle.sh
+8
View File
@@ -124,6 +124,14 @@ done < "$TMP/files.z"
sort -u "$LIST" -o "$LIST"
# Always include Gradle wrapper bootstrap, even though generic JARs are denied.
if [[ -f "$ROOT/SHiNE-server/gradle/wrapper/gradle-wrapper.jar" ]]; then
if ! grep -Fxq 'SHiNE-server/gradle/wrapper/gradle-wrapper.jar' "$LIST"; then
printf '%s\n' 'SHiNE-server/gradle/wrapper/gradle-wrapper.jar' >> "$LIST"
sort -u "$LIST" -o "$LIST"
fi
fi
# Content scan: fail closed on common credential/private-key patterns.
# We scan only text-ish files; grep -I skips binary data.
SECRET_RE='-----BEGIN ([A-Z0-9 ]+ )?PRIVATE KEY-----|AKIA[0-9A-Z]{16}|ASIA[0-9A-Z]{16}|gh[pousr]_[A-Za-z0-9_]{20,}|github_pat_[A-Za-z0-9_]{20,}|sk-[A-Za-z0-9_-]{20,}|xox[baprs]-[A-Za-z0-9-]{10,}|AIza[0-9A-Za-z_-]{30,}|(^|[^A-Za-z0-9])(password|passwd|pwd|secret|api[_-]?key|access[_-]?token|auth[_-]?token|private[_-]?key)[[:space:]]*[:=][[:space:]]*["'\'']?[^${[:space:]]{][^[:space:]]{7,}'
docs/API/09_Operations_Index.md
+4 -7
View File
@@ -63,14 +63,9 @@
| `UpsertPushToken` | `12_Direct_Messages_Push_Calls_API.md` | регистрация WebPush-токена |
| `SendTestWebPush` | `12_Direct_Messages_Push_Calls_API.md` | тестовая push-доставка |
| `SendMessagePair` | `12_Direct_Messages_Push_Calls_API.md` | отправка пары входящий/исходящий DM |
| `ReceiveOutcomingMessage` | `12_Direct_Messages_Push_Calls_API.md` | старая полная DM-пара для второго access-сервера отправителя |
| `ReceiveIncomingMessage` | `12_Direct_Messages_Push_Calls_API.md` | прием входящего DM-блока |
| `ReceiveIncomingMessage` | `12_Direct_Messages_Push_Calls_API.md` | внутренняя доставка входящего DM-блока на единственный сервер получателя |
| `DeleteMessage` | `12_Direct_Messages_Push_Calls_API.md` | tombstone одного личного сообщения у обеих сторон |
| `DeleteConversation` | `12_Direct_Messages_Push_Calls_API.md` | tombstone удаления истории переписки |
| `DmSyncBatch` | `12_Direct_Messages_Push_Calls_API.md` | pull событий `synced=false` с ACK предыдущей страницы |
| `GetDmDeliveryStatus` | `12_Direct_Messages_Push_Calls_API.md` | read-only статус доставки по существующему `messageKey` |
| `UserSettingsSyncBatch` | `13_User_Settings_API.md` | межсерверная догоняющая синхронизация пользовательских настроек по курсору |
| `MarkAllUserSettingsUnsynced` | `13_User_Settings_API.md` | служебная пометка всех настроек как несинхронизированных |
| `GetDirectMessages` | `12_Direct_Messages_Push_Calls_API.md` | постраничная загрузка истории личного диалога |
| `AckSessionDelivery` | `12_Direct_Messages_Push_Calls_API.md` | подтверждение доставки в сессию |
| `CallInviteBroadcast` | `12_Direct_Messages_Push_Calls_API.md` | broadcast приглашения к звонку |
@@ -78,7 +73,9 @@
## Важные замечания
- `ReceiveOutcomingMessage` зарегистрирован как алиас того же handler/request-класса, что и `SendMessagePair`, и сохраняет прежний межсерверный payload.
- UI отправляет полную подписанную пару только через `SendMessagePair`.
- `ReceiveIncomingMessage` вызывается сервером отправителя для доставки одной
входящей копии на единственный access-сервер получателя.
- Межсерверные DM-операции пока доверяют `sourceServerLogin`; отдельная межсерверная авторизация запланирована позднее.
- `ServerHello` пока принимает заявленный `serverLogin` на доверии и не является криптографической аутентификацией.
- Отдельных HTTP endpoints для DM-файлов сейчас нет.
docs/API/12_Direct_Messages_Push_Calls_API.md
+20 -97
View File
@@ -9,8 +9,10 @@
Важно:
- для DM v1 нужно использовать `SendMessagePair`, `ReceiveOutcomingMessage`, `ReceiveIncomingMessage`, `DeleteMessage`, `DeleteConversation`, `GetDirectMessages`;
- `DmSyncBatch` предназначен для межсерверной догоняющей синхронизации, не для обычного клиентского UI.
- UI использует `SendMessagePair`, `DeleteMessage`, `DeleteConversation`
и `GetDirectMessages`;
- `ReceiveIncomingMessage` — внутренняя доставка одной входящей копии на
единственный access-сервер получателя.
- сервер поддерживает материализованный слой диалогов `dm_dialog_state`; `read receipt` обновляет серверный watermark и `unreadCount`, а не только локальный клиентский флаг.
- в `dm_dialog_state` сервер также хранит `last_message_blob_b64` для последнего контентного DM в base64, чтобы клиент мог отрисовать список чатов без дополнительного запроса.
@@ -69,9 +71,7 @@
}
```
## 3. `SendMessagePair` и `ReceiveOutcomingMessage`
`ReceiveOutcomingMessage` — алиас `SendMessagePair`.
## 3. `SendMessagePair`
### Назначение
@@ -114,7 +114,10 @@
}
```
Успешный `status=200` подтверждает локальное сохранение. До ответа клиенту сервер параллельно пробует оба актуальных сервера получателя. Поэтому `deliveryState` уже может быть `delivered`; если никто не ответил, возвращается `accepted`.
Успешный `status=200` подтверждает локальное сохранение. До ответа клиенту
сервер пробует единственный актуальный сервер получателя. Поэтому
`deliveryState` уже может быть `delivered`; если сервер получателя не
ответил, возвращается `accepted`.
Возможные `deliveryState`: `accepted`, `delivered`, `failed`.
@@ -132,7 +135,8 @@
### Назначение
Используется там, где нужно принять только incoming-вариант сообщения.
Используется сервером отправителя для доставки incoming-варианта сообщения на
единственный access-сервер получателя.
Принимаемые типы:
@@ -152,7 +156,9 @@
}
```
`sourceServerLogin` необязателен для совместимости. Пока межсерверная авторизация отложена, это поле считается доверенным. Пользовательская подпись signed-блока проверяется всегда. Пустой `sourceServerLogin` трактуется как клиентский вызов, непустой - как peer-вызов.
`sourceServerLogin` пока передаётся для диагностики маршрута. Межсерверная
авторизация остаётся отдельным будущим этапом, но пользовательская подпись
signed-блока проверяется всегда. Официальный UI эту операцию не вызывает.
Успешный ответ содержит существующие `messageKey`, `baseKey` и счётчики realtime-доставки. Повтор уже сохранённой той же ревизии обрабатывается идемпотентно.
@@ -262,31 +268,11 @@
Поле `deliveryState` заполняется для исходящих элементов `type=2`. У входящих `type=1` оно отсутствует/равно `null`.
## 8. Межсерверные операции DM
## 8. Внутренняя межсерверная доставка DM
До отдельного этапа server-auth поля `sourceServerLogin` доверяются. Каждый `SHiNE_DM` всё равно заново проходит проверку формата и пользовательской подписи.
### 8.1. `ReceiveIncomingMessage`
### 8.1. `ReceiveOutcomingMessage`
Прежняя операция передачи полной пары второму access-серверу отправителя. Delivery-state в межсерверный запрос не входит.
```json
{
"op": "ReceiveOutcomingMessage",
"requestId": "dm-peer-001",
"payload": {
"incomingBlobB64": "BASE64_INCOMING",
"outgoingBlobB64": "BASE64_OUTGOING",
"sourceServerLogin": "server-a"
}
}
```
Ответ имеет прежний формат `SendMessagePair`. Принимающий сервер сохраняет пару идемпотентно и самостоятельно ставит локальную задачу доставки.
### 8.2. `ReceiveIncomingMessage`
Прежняя операция передачи одной входящей копии серверу получателя или второму серверу самого получателя.
Передаёт одну входящую копию на единственный access-сервер получателя.
```json
{
@@ -301,72 +287,9 @@
Успешный 2xx-ответ означает, что signed-блок проверен и находится в БД (новая или идемпотентно повторённая запись).
### 8.3. `DmSyncBatch`
Pull-синхронизация событий владельца с `synced=false`. `ackSyncIds` подтверждает версии событий, успешно сохранённые из предыдущего ответа.
```json
{
"op": "DmSyncBatch",
"requestId": "dm-sync-001",
"payload": {
"ownerLogin": "alice",
"afterStoredAtMs": 0,
"afterMessageKey": "",
"limit": 500,
"maxBytes": 3000000,
"ackSyncIds": []
}
}
```
Ответ возвращает страницу несинхронизированных событий и курсор внутри текущего цикла:
```json
{
"nextStoredAtMs": 1774700001123,
"nextMessageKey": "alice|bob|1774700000123|123456789|2",
"hasMore": false,
"items": [
{
"syncId": "alice|bob|1774700000123|123456789|2:0:0",
"primaryMessageKey": "alice|bob|1774700000123|123456789|2",
"storedAtMs": 1774700001123,
"blobsB64": ["BASE64_INCOMING", "BASE64_OUTGOING"]
}
]
}
```
Для пары порядок всегда incoming/outgoing; для входящей копии и tombstone массив содержит один blob. `syncId` — технический идентификатор конкретной ревизии только для ACK синхронизации, а не второй ID сообщения. После сохранения вызывающий сервер передаёт `syncId` в `ackSyncIds` следующего запроса. Только тогда источник ставит `synced=true`. Новый цикл начинается с `afterStoredAtMs=0`; полный сброс флагов поэтому повторно отдаёт всю историю.
### 8.4. `GetDmDeliveryStatus`
Единственная новая межсерверная операция доставки. Read-only проверка существующего `messageKey`; не изменяет состояние отвечающего сервера.
```json
{
"op": "GetDmDeliveryStatus",
"requestId": "dm-status-001",
"payload": {
"messageKey": "alice|bob|1774700000123|123456789|2"
}
}
```
Ответ содержит `messageKey`, `known` и `delivered`. `delivered=true` означает доставку хотя бы одному серверу получателя.
Периодический процесс использует один логический последовательный сеанс поверх
постоянного WSS-пула: сначала синхронизирует настройки, затем вызывает
`DmSyncBatch` до завершения страниц и ACK. Существующий
`MarkAllUserSettingsUnsynced` также сбрасывает DM-флаги и возвращает
`dmUpdated`; отдельной операции `MarkAllDmUnsynced` нет.
Основные ошибки межсерверных операций:
- `400 / EMPTY_OWNER_LOGIN`, `EMPTY_MESSAGE_KEY`;
- `403 / LOCAL_SERVER_NOT_ACCESS_SERVER`;
- `500 / LOCAL_SERVER_NOT_CONFIGURED`.
Операции `ReceiveOutcomingMessage`, `DmSyncBatch` и
`GetDmDeliveryStatus` удалены вместе с репликацией между access-серверами
одного пользователя.
## 9. `AckSessionDelivery`
docs/API/13_User_Settings_API.md
+36 -83
View File
@@ -1,104 +1,57 @@
# API пользовательских настроек
# User Settings API
Этот раздел описывает `user_settings` - отдельное хранилище пользовательских настроек, не связанное с `users_params` и не связанное с legacy DM-таблицами.
user_settings хранит технические настройки пользователя только на его
единственном действующем access-сервере. Межсерверной синхронизации настроек
нет. При смене access-сервера старые настройки автоматически не переносятся.
## 1. Назначение
## Подпись
`user_settings` хранит технические пользовательские настройки, которые должны синхронизироваться между максимум двумя access/sync-серверами пользователя.
UI формирует строку:
Основной текущий кейс:
SHiNe/UserSettings:<login>|<setting_type>|<setting_key>|<time_ms>|<value_text>|<value_num>
- `setting_type = 1` - курсор прочитанности канала;
- `setting_key = ownerBlockchainName/channelName`;
- `value_num = number of messages already seen in channel`;
- `value_text = ''`.
и подписывает её текущим Ed25519 clientKey пользователя. Сервер:
Если настройки нет, канал считается полностью прочитанным, то есть unread = `0`.
После первого открытия канала UI отправляет текущий курсор, чтобы зафиксировать baseline и дальше считать только новые сообщения.
1. находит пользователя;
2. проверяет совпадение client_key с актуальным ключом пользователя;
3. строго проверяет подпись;
4. при неверной подписи возвращает INVALID_SIGNATURE;
5. сохраняет запись только если time_ms новее локальной версии.
## 2. Структура записи
## Операции
- `login` - логин владельца настройки;
- `setting_type` - числовой код типа настройки;
- `setting_key` - строковый ключ настройки;
- `time_ms` - время установки значения в миллисекундах;
- `value_text` - строковое значение;
- `value_num` - числовое значение;
- `client_key` - публичный Ed25519 ключ клиента в Base64;
- `signature` - Ed25519 подпись preimage в Base64;
- `synced` - была ли настройка успешно доставлена на второй сервер.
### UpsertUserSetting
Уникальность: `(login, setting_type, setting_key)`.
Обновление: только если `time_ms` новее текущего значения.
Обязательные поля:
## 3. Формат подписи
- login;
- setting_type;
- setting_key;
- time_ms;
- client_key;
- signature.
Подписывается строка:
Значения:
`SHiNe/UserSettings:|login|setting_type|setting_key|time_ms|value_text|value_num`
- value_text — строковое значение, по умолчанию пустая строка;
- value_num — числовое значение, по умолчанию 0.
Где внутри полей используется экранирование `\` и `|`.
Поля sync_delivery и synced удалены.
Подпись создаётся клиентским `client_key`.
### GetUserSetting
## 4. Операции
Возвращает одну локальную настройку по login, setting_type и setting_key.
### `UpsertUserSetting`
### ListUserSettings
Записывает или обновляет настройку пользователя.
Возвращает все локальные настройки пользователя.
Если запрос пришёл от клиента, сервер:
## Удалённые операции
- сохраняет запись локально;
- пытается сразу отправить её на доступный sync-сервер;
- если отправка успешна, помечает запись как `synced=true`;
- если нет, оставляет `synced=false`.
После перехода на один access-сервер удалены:
Если запрос пришёл по синхронизации между серверами, используется `sync_delivery=true`, и повторной пересылки дальше не делается.
- UserSettingsSyncBatch;
- MarkAllUserSettingsUnsynced.
### `GetUserSetting`
Чтение одной настройки по `(login, setting_type, setting_key)`.
### `ListUserSettings`
Список всех настроек пользователя.
### `UserSettingsSyncBatch`
Внутренний межсерверный batch-эндпоинт.
- отдаёт настройки, новые относительно курсора;
- используется для bootstrap и догрузки после восстановления;
- применяется только для пользователей, чей сервер есть в `access_servers`.
### `MarkAllUserSettingsUnsynced`
Внутренний служебный запрос.
- помечает все настройки пользователя или все настройки сразу как `synced=false`;
- одновременно сбрасывает DM-outbox того же пользователя (или всех пользователей), чтобы операция замены/добавления access-сервера не оставила историю DM несинхронизированной;
- ответ дополнительно содержит `dmUpdated` — количество сброшенных DM-событий;
- нужен после добавления нового sync-сервера или при потере локальной БД.
## 5. Синхронизация
Настройки и DM имеют раздельные таблицы и правила ACK, но периодический процесс
использует один логический последовательный сеанс поверх постоянного WSS-пула:
сначала синхронизирует настройки, затем забирает `DmSyncBatch`. Завершение
логического сеанса не закрывает физический сокет.
- локальная запись создаётся с `synced=false`, если её ещё не подтвердил второй сервер;
- если запись пришла с другого сервера, она сохраняется сразу как `synced=true`;
- периодический sync раз в 6 часов проверяет несинхронизированные настройки, догружает их по курсору и затем забирает несинхронизированные DM;
- если появляется новый sync-сервер или локальная БД была потеряна, нужно пометить все настройки несинхронизированными и заново догрузить batch с нуля.
## 6. Текущий UI-кейс
UI при открытии канала отправляет `UpsertUserSetting` с:
- `setting_type = 1`;
- `setting_key = ownerBlockchainName/channelName`;
- `value_num = количество уже просмотренных сообщений в канале`.
Это значение используется сервером для расчёта unread в списке каналов и в канале.
Также удалены таблица peer-cursor и периодический сервис синхронизации
настроек.
docs/API/16_Server_Connection_Pool_API.md
+4 -3
View File
@@ -55,7 +55,8 @@
- физическое соединение создаётся одно на `serverLogin`;
- логические операции DM, settings и blockchain используют один WSS;
- завершение `RemoteSyncSession` не закрывает физический сокет;
- известные peer берутся из `sync_servers` и `user_access_servers_current`;
- известные peer берутся из `sync_servers` и первых действующих маршрутов
`user_access_servers_current`;
- список перечитывается каждые 30 секунд;
- при изменении URL соединение пересоздаётся;
- новые запросы не повторяются транспортом автоматически: действующие
@@ -65,8 +66,8 @@
| Приоритет | Операции |
| --- | --- |
| `REALTIME` | доставка DM, deletes, `GetDmDeliveryStatus` |
| `NORMAL` | настройки, `DmSyncBatch` и access-data sync |
| `REALTIME` | доставка DM и signed deletes |
| `NORMAL` | прочие межсерверные запросы |
| `BULK` | blockchain heads, blocks, `AddBlock` backfill |
На одном peer одновременно исполняется один запрос. Приоритет применяется к
docs/Blockchain/README.md
+1 -1
View File
@@ -33,7 +33,7 @@
## Смежная документация
- [../ИТХ/README.md](../ИТХ/README.md) — ежедневное закрытие блокчейна (ИТХ): краткий обзор.
- [../ИТХ/Спецификация_ИТХ_v1.md](../ИТХ/Спецификация_ИТХ_v1.md) — точная спецификация чекпоинтов (Arweave/Solana/канал закрытий).
- [sync-between-servers.md](./sync-between-servers.md) — живая межсерверная синхронизация блоков и DM.
- [sync-between-servers.md](./sync-between-servers.md) — живая межсерверная синхронизация блокчейна и доставка DM на единственный сервер получателя.
## Важные ограничения MVP
- Каналы `type=100` и `type=200` присутствуют в формате, но сейчас не используются в UI.
docs/Blockchain/sync-between-servers.md
+17 -20
View File
@@ -1,4 +1,4 @@
# Синхронизация блоков и DM между серверами SHiNE
# Синхронизация блокчейнов и доставка DM между серверами SHiNE
Документ описывает архитектуру и протокол синхронизации данных между партнёрскими серверами SHiNE.
@@ -19,9 +19,9 @@
Каждый сервер регистрирует в своей Solana PDA список `sync_servers` —
логины SHiNE-аккаунтов партнёрских серверов, с которыми он синхронизируется.
Важно: в текущей архитектуре у пользователя одновременно может быть не более
двух sync/access-серверов. Это ограничение считается обязательным для runtime-логики
`synced` и пользовательских курсоров.
`sync_servers` относится к серверному узлу и не является списком
access-серверов обычного пользователя. У пользователя действует только первый
`access_servers[0]`.
- Список хранится в блоке `ServerProfileBlock` внутри `user_pda` сервера.
- Адрес каждого партнёрского сервера читается из его PDA на Solana.
@@ -35,7 +35,7 @@
- Сервер-отправитель: сохраняет пару и ставит асинхронную delivery-задачу.
- Сервер-получатель: сохраняет входящий блок в `signed_messages`, затем доставляет его активным сессиям.
- Дедупликация по уникальному `message_key = from|to|timeMs|nonce|type`.
- Репликация между двумя access-серверами пользователя работает через `dm_sync_outbox.synced`, без постоянного time-cursor.
- Между access-серверами одного пользователя DM не реплицируются.
- Полная актуальная схема: `docs/Personal_Messages/Доставка_и_синхронизация_DM.md`.
### 3.2 Блоки пользовательского блокчейна
@@ -47,10 +47,8 @@
### 3.3 Пользовательские настройки
- Отдельная таблица `user_settings`.
- Синхронизируются технические настройки пользователя, включая курсор прочитанности каналов.
- Для текущего UI-кейса хранится `setting_type = 1` и `setting_key = ownerBlockchainName/channelName`.
- Синхронизация идёт с учётом `time_ms` и флага `synced`.
Пользовательские настройки хранятся локально на единственном access-сервере и
между серверами не синхронизируются.
## 4. Текущая реализованная схема
@@ -196,7 +194,7 @@ Full resync запускается только тогда, когда:
- На текущем этапе `serverLogin` принимается на доверии; подпись Ed25519 корневым ключом сервера отложена.
- При разрыве выполняется переподключение с jitter/backoff до 60 секунд.
- После 120 секунд отсутствия полезного трафика отправляется WebSocket ping; pong ожидается 15 секунд.
- Один физический канал переиспользуют DM, настройки и blockchain.
- Один физический канал переиспользуют доставка DM и blockchain.
### 5.2 Доставка новых данных (push)
@@ -208,7 +206,7 @@ Full resync запускается только тогда, когда:
- При первом подключении к партнёру серверы могут обмениваться курсорами состояния блокчейнов.
- Сервер с более полной историей досылает недостающее партнёру.
- DM time-cursor удалён: новый peer получает историю после сброса `dm_sync_outbox.synced=false`.
- DM-history backfill между access-серверами отсутствует.
### 5.4 Разрешение конфликтов
@@ -222,11 +220,11 @@ Full resync запускается только тогда, когда:
1. Клиент A отправляет пару блоков на свой сервер X.
2. Сервер X валидирует и локально сохраняет пару.
3. До ответа клиенту X параллельно отправляет входящую копию максимум двум актуальным `access_servers` B.
4. ACK хотя бы одного маршрута B означает терминальный `delivered`; второй сервер B догоняется собственной DM-синхронизацией.
5. После первой попытки X передаёт полную пару второму access-серверу A старым `ReceiveOutcomingMessage`, без delivery-state.
6. При нулевой доставке выполняются повторы через 30 секунд, 5 минут, 25 минут и 1 час. Перед тремя последними X спрашивает peer A через `GetDmDeliveryStatus(messageKey)`.
7. После неудачной попытки через час ставится терминальный `failed`.
3. До ответа клиенту X отправляет входящую копию на единственный
`access_servers[0]` пользователя B через `ReceiveIncomingMessage`.
4. Успешное сохранение на сервере B означает терминальный `delivered`.
5. При неудаче выполняются повторы через 30 секунд, 5 минут, 25 минут и 1 час.
6. После неудачной попытки через час ставится терминальный `failed`.
Изменения routing B учитываются до терминального состояния, потому что список маршрутов перечитывается на каждой попытке.
@@ -251,15 +249,14 @@ Full resync запускается только тогда, когда:
| Обход Solana RPC через `sync.importUserProfileFromPartner.enabled` | ✅ Реализовано |
| Обычный `AddBlock` через `tmp_bch`/`write_check`/`write_pending` | ✅ Реализовано |
| Межсерверный постоянный WebSocket-канал | ✅ Реализован общий `ServerConnectionPool` |
| Асинхронная доставка DM на access-серверы получателя | ✅ Реализовано |
| Асинхронная доставка DM на единственный access-сервер получателя | ✅ Реализовано |
| Retry DM до 1 часа + UI-state | ✅ Реализовано |
| Репликация DM на второй access-сервер по `synced` | ✅ Реализовано |
| Read-only `GetDmDeliveryStatus` | ✅ Реализовано |
| Репликация DM и настроек между access-серверами | Не используется |
| Push блоков блокчейна партнёрам | ✅ Выполняется через постоянный WSS-пул |
| Periodic backfill отсутствующего хвоста | ✅ Реализовано |
| Разрешение рассинхрона / divergence | ✅ Реализована базовая full-resync схема во время periodic sync |
| Startup recovery по `*.resync_pending` marker-file | ✅ Реализовано |
| Маршрутизация DM через один/два `access_servers` | ✅ Реализовано |
| Маршрутизация DM только через `access_servers[0]` | ✅ Реализовано |
| Криптографическая server-to-server авторизация DM | Нужна реализация |
Текущая версия сервера использует постоянный WSS-пул для существующих
docs/Personal_Messages/README.md
+2 -1
View File
@@ -6,7 +6,8 @@
- `docs/Personal_Messages/Протокол_DM_v1.md` — логика протокола, роли API, серверное поведение, routing по `access_servers`
- `docs/Personal_Messages/Формат_DM_v1.md` — точный бинарный формат контейнера `SHiNE_DM`
- `docs/Personal_Messages/Доставка_и_синхронизация_DM.md` — состояния доставки, retry-воркер, репликация между двумя access-серверами и UI-статусы
- `docs/Personal_Messages/Доставка_и_синхронизация_DM.md` — доставка на
единственный access-сервер, retry-воркер и UI-статусы
- `docs/Personal_Messages/Технические_вставки_DM_v1.md` — формат специальных `<S:...>` вставок внутри plaintext DM после расшифровки
Исторический устаревший документ сохранён отдельно:
docs/Personal_Messages/Доставка_и_синхронизация_DM.md
+59 -113
View File
@@ -1,138 +1,84 @@
# Доставка и синхронизация личных сообщений
# Доставка личных сообщений
## 1. Главный принцип
## 1. Инвариант
DM считается доставленным, когда signed-входящую копию сохранил хотя бы один актуальный access-сервер получателя. Доставка на оба сервера не требуется: сервер получателя самостоятельно синхронизирует входящее сообщение со своим вторым сервером.
У каждого пользователя действует ровно один сервер доступа: первый элемент
access_servers[0] в его Solana PDA.
Клиентский `status=200` от `SendMessagePair` означает, что собственный сервер сохранил пару. Это отдельный факт от доставки получателю.
Остальные элементы старой PDA, если они существуют, не участвуют в
маршрутизации и не являются fallback. UI позволяет только заменить текущий
сервер другим.
Формат подписанного контейнера `SHiNE_DM` не меняется. Состояние доставки и флаг синхронизации — изменяемые серверные метаданные.
Межсерверной репликации личной переписки между серверами одного пользователя
нет. При смене сервера история автоматически не переносится.
## 2. Идентификаторы
## 2. Подписанные копии
- `baseKey` связывает входящую и исходящую копии одного логического сообщения;
- `incomingKey` идентифицирует копию получателя;
- `outgoingKey`/`messageKey` идентифицирует копию отправителя и используется для проверки доставки;
- дополнительный публичный `eventId` для доставки не создаётся;
- `syncId` используется только внутри `DmSyncBatch` как ACK конкретной ревизии, потому что редактирование сохраняет прежний `messageKey`.
Формат SHiNE_DM не меняется. Для сообщения клиент создаёт:
## 3. Состояния доставки
- входящую копию для получателя;
- исходящую копию для отправителя.
| Состояние | Смысл | Повторные попытки |
|---|---|---|
| `accepted` | пара сохранена сервером отправителя, но ни один сервер получателя ещё не подтвердил запись | да |
| `delivered` | хотя бы один сервер получателя подтвердил запись | нет |
| `failed` | последняя попытка через час завершилась без доставки | никогда |
Сервер отправителя принимает пару через SendMessagePair, атомарно сохраняет её
и создаёт изменяемое состояние доставки.
Состояние `delivered` терминальное. Сервер отправителя не пытается отдельно добиться второго ACK получателя.
## 3. Основной маршрут
## 4. Обычная отправка
1. UI отправителя вызывает SendMessagePair на своём access-сервере.
2. Сервер сохраняет исходящую копию отправителя и входящую копию, необходимую
для доставки и повторных попыток.
3. Сервер читает единственный маршрут получателя из
user_access_servers_current.
4. Если это другой физический сервер, он вызывает внутреннюю операцию
ReceiveIncomingMessage и передаёт только входящую копию.
5. Сервер получателя проверяет пользовательскую подпись, идемпотентно сохраняет
входящую копию и уведомляет активные сессии.
6. Успешное сохранение означает delivered.
1. Клиент отправляет `SendMessagePair` на один свой access-сервер.
2. Сервер проверяет формат, пользователей, подписи и согласованность пары.
3. Сервер атомарно сохраняет входящую и исходящую копии.
4. В том же request-процессе сервер читает до двух актуальных маршрутов получателя.
5. Оба вызова `ReceiveIncomingMessage` запускаются параллельно.
6. Если хотя бы один вызов успешен, устанавливается `delivered`.
7. После этой попытки сервер передаёт полную пару своему второму access-серверу старой операцией `ReceiveOutcomingMessage`.
8. `SendMessagePair` возвращает существующие ключи и единственное новое поле `deliveryState`.
ReceiveIncomingMessage не является синхронизацией двух серверов одного
пользователя. Это основная доставка между серверами разных пользователей.
Успешный повтор уже сохранённого signed-блока считается ACK. Все операции должны быть идемпотентными.
## 4. Повторные попытки
## 5. Расписание повторов
Попытки привязаны ко времени первоначального принятия:
Воркер запускается каждые 5 секунд и выбирает только due-строки по индексу. Он не сканирует всю таблицу сообщений.
- сразу;
- через 30 секунд;
- через 5 минут;
- через 25 минут;
- через 1 час.
Попытки привязаны к времени первоначального принятия:
Перед каждой попыткой заново читается первый актуальный access-сервер
получателя. После финальной неудачи состояние становится failed.
| Номер | Время от старта | Сначала спросить второй сервер отправителя |
|---:|---:|---|
| 1 | сразу | нет |
| 2 | 30 секунд | нет |
| 3 | 5 минут | да |
| 4 | 25 минут | да |
| 5 | 1 час | да |
Второй сервер отправителя не опрашивается, полная пара ему не передаётся.
Из-за шага воркера повтор может начаться на несколько секунд позже указанного времени. Первая попытка выполняется немедленно и от воркера не зависит.
## 5. Состояния
На 5-й, 25-й и 60-й минутах сервер сначала вызывает у второго сервера отправителя:
- accepted — пара сохранена сервером отправителя;
- delivered — входящая копия сохранена единственным сервером получателя;
- failed — финальная попытка завершилась неудачей;
- read-receipt — отдельное подписанное DM-событие.
```json
{
"op": "GetDmDeliveryStatus",
"payload": {
"messageKey": "alice|bob|1774700000123|123456789|2"
}
}
```
Подписанные DM-контейнеры остаются криптографическими фактами. Сетевое
deliveryState остаётся изменяемой серверной метаданной.
Ответ:
## 6. Удаления
```json
{
"messageKey": "alice|bob|1774700000123|123456789|2",
"known": true,
"delivered": true
}
```
DeleteMessage и DeleteConversation доставляют подписанный tombstone только на
единственные access-серверы обеих сторон. Повторное применение идемпотентно.
Операция read-only. Если peer отвечает `delivered=true`, локальный сервер устанавливает `delivered` и не обращается к серверам получателя. Ошибка или отсутствие операции на старом peer не блокирует собственную попытку.
## 7. Удалённая репликация
Перед каждой попыткой маршруты получателя заново читаются из `user_access_servers_current`. Изменение серверов учитывается только пока сообщение находится в `accepted`. После `delivered` или `failed` состояние больше не открывается.
После перехода на один сервер доступа удалены:
Если последняя проверка и попытка через час не дали ACK, устанавливается `failed`, `next_attempt_at_ms` очищается и сообщение больше никогда автоматически не отправляется.
- ReceiveOutcomingMessage;
- DmSyncBatch;
- GetDmDeliveryStatus;
- dm_sync_outbox;
- dm_sync_peer_state;
- ACK-флаги synced;
- периодический DM-backfill между access-серверами.
## 6. Старая межсерверная доставка
Форматы существующих операций не расширяются данными о результате доставки:
- `ReceiveOutcomingMessage` получает прежнюю пару `incomingBlobB64` + `outgoingBlobB64` и необязательный `sourceServerLogin`;
- `ReceiveIncomingMessage` получает один `incomingBlobB64` и необязательный `sourceServerLogin`.
Второй сервер отправителя после получения пары создаёт собственное локальное состояние доставки и самостоятельно пробует маршруты получателя. Серверы обмениваются результатом только через read-only `GetDmDeliveryStatus`.
## 7. Догоняющая синхронизация двух серверов пользователя
Для каждого владельца сервер ведёт outbox с флагом `synced`:
- исходящая пара отправителя — один элемент с двумя blob в порядке incoming/outgoing;
- входящая копия получателя — один элемент с одним blob;
- read-receipt и tombstone применяются теми же проверенными обработчиками.
`DmSyncBatch` возвращает только элементы с `synced=false`. Получатель проверяет signed-блоки, сохраняет их идемпотентно и в следующем запросе подтверждает `syncId` через `ackSyncIds`. Источник ставит `synced=true` только после ACK.
При обрыве соединения неподтверждённый элемент остаётся `synced=false` и безопасно приходит повторно. Элемент, полученный от peer, локально сразу считается синхронизированным, чтобы не образовалась петля.
`sourceServerLogin` является единственным признаком межсерверного вызова для этих операций: если поле пустое, запрос считается клиентским и его запись не должна сразу переводиться в `synced=true`.
Синхронизация настроек и DM проходит последовательно через один WS-сеанс: сначала настройки, затем все страницы DM. Если второй сервер был выключен, после включения он сам догружает пропущенные элементы.
Существующий `MarkAllUserSettingsUnsynced` также сбрасывает DM-флаги. После сброса история повторно передаётся как синхронизация, но старые сообщения получателю заново не отправляются: delivery-состояние создаётся с учётом их возраста и не открывает завершённую часовую очередь.
## 8. UI
| Вид | Значение |
|---|---|
| одна серая галочка | собственный сервер принял сообщение (`accepted`) |
| одна светлая галочка | хотя бы один сервер получателя принял сообщение (`delivered`) |
| две светлые галочки | пришёл существующий read-receipt |
| красный `!` и «Сообщение не доставлено» | окончательное состояние `failed` |
Read-receipt имеет приоритет над delivery-индикатором: если сообщение прочитано, оно заведомо было доставлено.
Сервер сообщает поздние изменения через `DmDeliveryStateChanged` с полями `outgoingKey`, `baseKey`, `deliveryState`. При повторном открытии чата то же состояние приходит в `GetDirectMessages`.
## 9. Нагрузка и отказоустойчивость
- воркер выбирает только due-записи по частичному индексу;
- терминальные строки не попадают в рабочую выборку;
- два маршрута первой попытки выполняются параллельно;
- ограниченный пул потоков и очередь защищают сервер при всплеске отправок;
- сетевой lease и сравнение версии строки предотвращают одновременную обработку одной задачи несколькими worker-потоками;
- повторная запись одного signed-блока безопасна;
- отсутствие второго сервера отправителя не мешает собственной доставке;
- отсутствие обоих серверов получателя завершает задачу через час.
## 10. Граница доверия
Межсерверная авторизация пока отложена. `sourceServerLogin` временно принимается на доверии, но каждый контейнер `SHiNE_DM` всё равно проходит проверку пользовательской подписи. `GetDmDeliveryStatus` сообщает только факт локального delivery-state и не изменяет данные.
Блокчейн-синхронизация через server-PDA sync_servers остаётся отдельным
механизмом и продолжает работать.
docs/Personal_Messages/Протокол_DM_v1.md
+138 -635
View File
@@ -1,696 +1,199 @@
# Личные сообщения (DM) — спецификация v1
# Протокол личных сообщений SHiNE DM v1
## Статус документа
## 1. Статус документа
Этот файл — актуальная логическая спецификация DM-протокола SHiNE v1.
Это актуальная спецификация логики личных сообщений. Точный байтовый формат
контейнера описан в Формат_DM_v1.md, а расписание доставки — в
Доставка_и_синхронизация_DM.md.
Важно:
Формат SHiNE_DM не изменён переходом на один access-сервер.
- это актуальная реализованная спецификация;
- код и документы по DM должны изменяться синхронно;
- при любых будущих изменениях DM сначала обновляется эта спецификация и соседний документ формата.
## 2. Инварианты
Документ фиксирует:
- каждый DM-контейнер подписан clientKey автора;
- сервер проверяет формат, пользователей и подпись до сохранения;
- повторная доставка одной ревизии идемпотентна;
- более старая ревизия не заменяет новую;
- у каждого пользователя действует только access_servers[0];
- дополнительные элементы старой PDA игнорируются без fallback;
- DM и настройки не реплицируются между access-серверами одного пользователя;
- sync_servers серверного PDA используются только для синхронизации
пользовательских блокчейнов.
- модель DM и смысл типов сообщений;
- правила редактирования, перешифровки и удаления;
- роли существующих API-методов;
- правила межсерверной маршрутизации через `access_servers`;
- общее поведение сервера и БД.
## 3. Типы DM
Точный байтовый формат контейнера вынесен отдельно:
- type=1 — входящая копия сообщения для получателя;
- type=2 — исходящая копия сообщения для отправителя;
- type=3 — входящий read-receipt;
- type=4 — исходящая копия read-receipt;
- type=5 — удаление своего исходящего сообщения;
- type=6 — удаление входящего сообщения получателем;
- type=7 — удаление переписки отправителем;
- type=8 — удаление переписки получателем.
- `docs/Personal_Messages/Формат_DM_v1.md`
Входящая и исходящая копии являются частями криптографического формата, а не
серверной репликацией. Они остаются даже при единственном access-сервере.
Формат клиентских технических вставок внутри plaintext вынесен отдельно:
## 4. Создание сообщения
- `docs/Personal_Messages/Технические_вставки_DM_v1.md`
Клиент:
Изменяемое состояние доставки, retry-расписание и репликация между двумя access-серверами подробно описаны отдельно:
1. получает актуальные clientKey отправителя и получателя;
2. шифрует входящую копию на ключ получателя;
3. шифрует исходящую копию на ключ отправителя;
4. создаёт два связанных контейнера с общим baseKey;
5. подписывает оба контейнера clientKey отправителя;
6. вызывает SendMessagePair на своём access-сервере.
- `docs/Personal_Messages/Доставка_и_синхронизация_DM.md`
Сервер отправителя атомарно сохраняет пару. Исходящая копия доступна отправителю,
а входящая хранится для доставки и повторных попыток.
Устаревшая предыдущая версия сохранена отдельно:
## 5. Единственный сервер доступа
- `docs/Personal_Messages/Спецификация_DM_v0.5_устаревшая.md`
Пользовательская PDA по-прежнему содержит массив access_servers для
совместимости формата. Действующим считается только первый элемент массива.
## 1. Основная модель
Routing-проекция user_access_servers_current содержит не более одной строки на
пользователя и строится строго из элемента с ord=1.
Личное сообщение в SHiNE хранится как две логические копии одного сообщения:
Если первый элемент отсутствует, указывает не на server-PDA или server_address
пуст, маршрут считается отсутствующим. Второй элемент автоматически не
используется.
- `type=1` — входящее сообщение для получателя;
- `type=2` — исходящая копия сообщения для отправителя.
UI позволяет:
Обе копии имеют общий логический идентификатор:
- увидеть текущий сервер;
- заменить его другим сервером;
- записать только один логин в access_servers.
- `baseKey = fromLogin|toLogin|timeMs|nonce`
Смена сервера не переносит сообщения и настройки. Новый сервер начинает с
пустой локальной истории.
Идентификатор конкретной копии:
## 6. Доставка сообщения
- `messageKey = baseKey|messageType`
### 6.1. SendMessagePair
Поля `timeMs` и `nonce` не меняются никогда.
Официальный UI отправляет пару только через SendMessagePair.
`nonce` обязателен, потому что одного `timeMs` недостаточно для гарантированной уникальности.
Сервер:
## 2. Ключи и шифрование
1. проверяет оба контейнера и их связь;
2. сохраняет пару;
3. создаёт delivery-state;
4. до ответа UI выполняет первую попытку доставки;
5. возвращает baseKey, incomingKey, outgoingKey и deliveryState.
### 2.1. Общий принцип
ReceiveOutcomingMessage удалён.
У пользователя в PDA публикуется один публичный `clientKey` в формате `Ed25519`.
### 6.2. ReceiveIncomingMessage
Из того же ключевого материала стандартным способом выводится ключ `X25519` для E2EE-шифрования DM.
Эта операция остаётся внутренней server-to-server операцией. Она нужна, когда
отправитель и получатель используют разные физические серверы.
Отдельный публичный `dmEncKey` в текущей архитектуре не хранится.
Сервер отправителя передаёт только incomingBlobB64 на единственный сервер
получателя. Принимающий сервер:
Подробности стандартного преобразования:
- разрешает получателя через свою routing-проекцию;
- заново проверяет пользовательскую подпись;
- применяет входящую копию идемпотентно;
- обновляет диалог и realtime-сессии;
- возвращает успешный ответ только после сохранения.
- `docs/Протоколы/Преобразование_ED25519_в_X25519.md`
ReceiveIncomingMessage не реплицирует историю на второй сервер пользователя.
### 2.2. Правило шифрования копий
## 7. Повторная доставка
- `type=1` шифруется на ключ получателя;
- `type=2` шифруется на ключ отправителя.
Состояния:
Следствия:
- accepted — пара сохранена сервером отправителя;
- delivered — incoming-копия сохранена сервером получателя;
- failed — закончилась финальная попытка;
- read — подтверждается отдельным type=3/4.
- входящую копию может прочитать только получатель;
- исходящую копию может прочитать только отправитель;
- сервер не должен расшифровывать DM;
- ciphertext у `type=1` и `type=2` обычно разный и не обязан совпадать побайтно.
Попытки выполняются сразу, через 30 секунд, 5 минут, 25 минут и 1 час.
Перед каждой попыткой первый маршрут получателя читается заново.
### 2.3. Что именно знает сервер о `body`
Peer отправителя не получает полную пару и не опрашивается о состоянии.
DmSyncBatch и GetDmDeliveryStatus удалены.
Для контентных DM (`type=1/2`) сервер в v1 должен рассматривать `body` как opaque blob:
## 8. Чтение и realtime
- сервер проверяет внешний контейнер `SHiNE_DM`;
- сервер проверяет подпись;
- сервер проверяет служебные поля envelope;
- сервер проверяет только то, что `bodyLen > 0` и не превышает допустимый лимит;
- сервер не обязан разбирать внутренний crypto-контейнер `body`;
- сервер не обязан знать конкретный алгоритм шифрования `body`.
GetDirectMessages возвращает историю страницами. UI использует стабильные
messageKey/baseKey для дедупликации и отображает более новую ревизию.
Это позволяет альтернативным клиентам использовать совместимый внешний DM-envelope при собственном формате зашифрованного payload внутри `body`.
dm_dialog_state хранит:
Официальный UI SHiNE при приёме такого сообщения:
- последнюю контентную копию;
- время и ключ последнего сообщения;
- unreadCount;
- watermark прочтения.
- пытается расшифровать знакомый формат `EncryptedBody_v1_0`;
- показывает текст сообщения при успешной расшифровке;
- если внутренний формат не поддерживается, показывает `Формат сообщения не поддерживается`;
- если формат понятен, но расшифровка не удалась, показывает `Не удалось расшифровать сообщение`;
- если `body` повреждён или структурно битый, показывает `Сообщение повреждено`.
WebSocket push ускоряет отображение, но после переподключения клиент запрашивает
историю у своего единственного access-сервера.
После успешной расшифровки plaintext может дополнительно содержать специальные клиентские вставки `<S:...>` в начале текста. Legacy-вставки `<SHiNE:...>` также продолжают поддерживаться при чтении.
Эти вставки относятся уже к уровню UI/plaintext, а не к уровню серверного DM-envelope.
## 9. Подтверждения прочтения
### 2.4. Источник истины по пользователю
Read-receipt создаётся как подписанная пара type=3/type=4 и проходит тот же
маршрут SendMessagePair → ReceiveIncomingMessage.
Для DM-проверки сервер использует:
Сервер хранит наибольший известный watermark, поэтому переставленные или
повторные receipts не уменьшают состояние прочтения.
- локальную runtime-проекцию пользователей как кэш;
- Solana PDA как источник истины по `clientKey` и `access_servers`.
## 10. Редактирование
Если нужного пользователя нет локально, сервер обязан попытаться lazy-import из Solana PDA:
Редактирование сохраняет логический messageKey исходной копии и увеличивает
revisionTimeMs. Сервер принимает только более новую валидную ревизию.
- в `GetUser`;
- при серверной верификации DM-подписи;
- перед межсерверной маршрутизацией через `access_servers`.
Если доставка исходного сообщения ещё не завершилась, следующая попытка
использует актуальную сохранённую incoming-копию.
## 3. Типы сообщений
## 11. Удаление
В версии v1 используются следующие `messageType`:
DeleteMessage принимает type=5/6. DeleteConversation принимает type=7/8.
- `1` — входящее сообщение;
- `2` — исходящая копия сообщения;
- `3` — входящее подтверждение прочтения;
- `4` — исходящая копия подтверждения прочтения;
- `5` — сообщение удалено отправителем;
- `6` — сообщение удалено получателем;
- `7` — переписка удалена отправителем;
- `8` — переписка удалена получателем.
Удаление является подписанным tombstone. Оно:
Типы `1/2` — это обычные контентные DM, в том числе их последующие ревизии.
- проверяется как обычный DM-контейнер;
- применяется идемпотентно;
- удаляет или блокирует соответствующий локальный контент;
- доставляется только на единственные access-серверы обеих сторон.
Типы `3/4` — служебные подтверждения прочтения.
Если сервер уже знает более новый tombstone, старая ревизия контента не
восстанавливает удалённую историю.
Типы `5/6/7/8` — служебные события удаления без зашифрованного тела.
## 12. Хранение
## 4. Базовые операции
Основные таблицы:
### 4.1. Создание нового сообщения
- signed_messages — подписанные DM-контейнеры;
- dm_dialog_state — материализованный список диалогов;
- dm_delivery_state — изменяемое состояние сетевой доставки;
- user_access_servers_current — единственный действующий маршрут пользователя.
Новое сообщение отправляется парой блоков через:
Удалены:
- `SendMessagePair`
- `ReceiveOutcomingMessage` как алиас
- dm_sync_outbox;
- dm_sync_peer_state;
- user_settings_sync_peer_state;
- synced-флаги репликации.
Пару создаёт автор сообщения.
## 13. Границы текущей версии
Пара содержит:
- межсерверная авторизация ServerHello пока не подписана;
- sourceServerLogin не является криптографическим доказательством сервера;
- пользовательская подпись каждого SHiNE_DM проверяется независимо от
транспорта;
- автоматической миграции истории при смене access-сервера нет;
- резервного fallback-сервера нет.
- одну входящую копию `type=1`;
- одну исходящую копию `type=2`;
- одинаковые `fromLogin`, `toLogin`, `timeMs`, `nonce`;
- одинаковый логический plaintext;
- разные ciphertext для разных владельцев копий.
## 14. Сохранённые механизмы
### 4.2. Редактирование сообщения
Переход на один access-сервер не затрагивает:
Редактирование общего текста делает автор сообщения.
При редактировании:
- `baseKey` остаётся тем же;
- `messageType` остаётся тем же;
- `revisionTimeMs` увеличивается;
- автор заново шифрует обе копии и снова отправляет пару через `SendMessagePair`.
Если `revisionTimeMs = 0`, это исходная версия.
Если `revisionTimeMs > 0`, это новая ревизия сообщения.
### 4.3. Перешифровка
Перешифровка нужна для будущего сценария полной или частичной перепаковки истории сообщений при сохранении тех же логических идентификаторов сообщений.
При перешифровке:
- `baseKey` остаётся прежним;
- `messageKey` остаётся прежним;
- plaintext может остаться тем же;
- ciphertext меняется;
- `reencryptedAtMs` получает ненулевое значение;
- `revisionTimeMs` может остаться прежним, в том числе нулевым, если менялось только шифрование без редактирования текста;
- сервер при выборе актуальной версии обязан учитывать обе метки времени.
Если сообщение уже удалено одним из типов `5/6`, его перешифровывать больше нельзя.
Если переписка уже удалена типом `7/8`, более старые сообщения этой пары тоже перешифровывать нельзя.
### 4.4. Приём входящей копии
Для server-to-server доставки одной входящей копии используется:
- `ReceiveIncomingMessage`
Этот метод должен принимать:
- новые входящие сообщения;
- входящие обновления/редактирования;
- будущие входящие перешифрованные копии.
## 5. Удаление одного сообщения
### 5.1. Общая логика
Удаление одного сообщения в v1 всегда глобальное у обеих сторон.
Локального удаления только у себя в этой версии протокола не вводится.
### 5.2. Типы удаления
- `type=5` — сообщение удалено отправителем;
- `type=6` — сообщение удалено получателем.
Удаляющее сообщение:
- не содержит зашифрованного тела;
- хранится в БД как tombstone;
- терминально закрывает это сообщение;
- не даёт больше принять никакую более позднюю содержательную версию этого же `messageKey`.
### 5.3. Метод удаления
Для удаления одного сообщения нужен отдельный метод, условно:
- `DeleteMessage`
Если сервер впервые получает валидное удаляющее сообщение:
- сохраняет tombstone в БД;
- удаляет или замещает прежнюю версию сообщения tombstone-записью;
- распространяет это же удаление на серверы доступа обеих сторон;
- не принимает в будущем попытки "оживить" это сообщение.
## 6. Удаление всей переписки
### 6.1. Отдельное служебное сообщение
Удаление всей переписки между двумя пользователями — это отдельное служебное сообщение без ciphertext.
Типы:
- `type=7` — переписка удалена отправителем;
- `type=8` — переписка удалена получателем.
### 6.2. Граница удаления
Границей удаления считается:
- `timeMs` самого служебного сообщения удаления переписки
Отдельное `deleteBeforeTimeMs` в этой версии не вводится.
### 6.3. Правило применения
Если сервер получает такое сообщение впервые:
- сохраняет его в БД как tombstone переписки;
- удаляет из БД все сообщения этой пары пользователей с `timeMs` меньше времени служебного сообщения;
- больше не принимает новые или повторно доставленные сообщения с `timeMs` раньше этой границы;
- распространяет это же сообщение удаления переписки на серверы доступа обеих сторон.
### 6.4. Поведение при позднем старом сообщении
Если после удаления переписки приходит старое сообщение, у которого:
- `timeMs < deleteConversationMessage.timeMs`
то сервер:
- не принимает это сообщение;
- распространяет уже известный tombstone удаления переписки на серверы доступа обеих сторон;
- ожидает, что вторая сторона обработает его как обычное уже известное удаление переписки.
### 6.5. Будущие новые сообщения
После удаления всей переписки новые сообщения между этими же пользователями разрешены, если:
- их `timeMs` больше времени служебного сообщения удаления переписки.
## 7. Серверы и маршрутизация
### 7.1. `access_servers`
Для обычного пользователя список серверов доставки и доступа задаётся через:
- `access_servers`
Это:
- сервера доступа пользователя;
- сервера relay;
- сервера, через которые пользователь получает личные сообщения и другие пользовательские операции.
### 7.2. `sync_servers`
`sync_servers` относятся не к обычной пользовательской маршрутизации DM, а к server-to-server партнёрству серверного узла.
Они используются для:
- синхронизации серверных данных;
- синхронизации пользовательских блокчейнов SHiNE;
- общей межсерверной координации.
`sync_servers` не являются списком пользовательских серверов доставки DM.
### 7.3. Два сервера у отправителя и получателя
Актуальный runtime исходит из ограничения:
- у одного пользователя не более двух `access_servers`;
- следовательно, у локального сервера есть не более одного peer для репликации данных пользователя.
Протокол поддерживает ситуацию, когда:
- у отправителя один или два `access_servers`;
- у получателя один или два `access_servers`;
- часть серверов у сторон совпадает;
- часть серверов уникальна.
Из этого следуют требования:
- все DM-операции должны быть идемпотентны;
- повторное получение уже известного события не должно ломать состояние;
- дубль tombstone должен быть безопасен;
- сервер не должен "оживлять" более старую версию сообщения после уже принятого tombstone.
## 8. Методы и их роли
### 8.1. Существующие методы, которые сохраняются
- `SendMessagePair`
- `ReceiveOutcomingMessage`
- `ReceiveIncomingMessage`
- `DeleteMessage`
- `DeleteConversation`
Их роли в v1:
- `SendMessagePair` — новая пара сообщений и редактирование старой пары автором;
- `ReceiveIncomingMessage` — приём одной входящей копии, входящих редактирований и входящего read-receipt;
- `ReceiveOutcomingMessage` — алиас `SendMessagePair`.
### 8.2. Межсерверная синхронизация
- `ReceiveOutcomingMessage` — прежняя полная пара второго access-сервера отправителя;
- `ReceiveIncomingMessage` — прежняя одиночная входящая копия;
- `DmSyncBatch` — pull событий с `synced=false` и ACK сохранённой предыдущей страницы;
- `GetDmDeliveryStatus` — единственная новая read-only проверка доставки по существующему `messageKey`.
Delivery-state между серверами не передаётся и не объединяется.
### 8.3. Серверный слой диалогов
Помимо хранения самих DM-сообщений сервер поддерживает материализованный слой состояния диалогов:
- отдельная запись на пару `owner_login` + `peer_login`;
- `relation_flag` со значениями `close_friend`, `contact`, `none`;
- `last_message_blob_b64` как последний контентный signed DM block в base64;
- `last_message_time_ms`;
- `unread_count`;
- `last_read_receipt_time_ms` как watermark последнего подтверждения прочтения.
Ключевые правила:
- `close_friend` всегда имеет приоритет над `contact`;
- если `read receipt` приходит не по порядку, сервер хранит наибольший watermark и не откатывает состояние назад;
- `unread_count` пересчитывается сервером по сообщениям диалога с учётом watermark и `read_at_ms`;
- старые исторические данные восстанавливаются из `signed_messages` при инициализации/миграции;
- UI не должен собирать inbox только из локального кеша, когда ему доступен серверный список диалогов.
## 9. Правила валидации и применения
### 9.1. Общее правило по ревизиям
Для одного и того же контентного сообщения сервер сравнивает пару:
- `revisionTimeMs`;
- `reencryptedAtMs`.
Правило:
- если новый `revisionTimeMs` больше сохранённого, сообщение применяется;
- если `revisionTimeMs` равен, но новый `reencryptedAtMs` больше сохранённого, сообщение применяется;
- если обе величины равны, сообщение не применяется;
- если новая пара (`revisionTimeMs`, `reencryptedAtMs`) меньше или равна сохранённой, сообщение не применяется.
Содержимое `body` при этом сравнении не участвует.
То есть если сервер уже видел ту же пару (`revisionTimeMs`, `reencryptedAtMs`), он считает, что такая версия у него уже есть.
### 9.2. Повторный tombstone
Повторный tombstone означает ситуацию, когда сервер повторно получает то же самое событие удаления:
- того же сообщения;
- или той же переписки.
Это может произойти из-за:
- повторной межсерверной доставки;
- нескольких `access_servers`;
- сетевых retry;
- дублирующей пересылки с разных маршрутов.
Правило:
- повторный tombstone должен быть полностью безопасен;
- если соответствующее удаление уже сохранено, сервер ничего не меняет и просто игнорирует повтор.
### 9.3. Сообщения старше границы удалённой переписки
Если для пары пользователей уже есть сохранённая граница удаления переписки, и приходит:
- обычное сообщение;
- удаление одного сообщения;
- повторное удаление переписки;
- любое другое DM-событие;
у которого `timeMs` меньше этой границы, сервер:
- ничего не меняет в БД;
- не восстанавливает старую историю;
- не применяет это событие повторно.
Если это старое контентное сообщение или входящая копия, сервер дополнительно перерассылает уже известный tombstone удаления переписки на `access_servers` обеих сторон, чтобы отстающие серверы сами подчистили историю.
Такие сообщения считаются частью уже удалённой истории.
### 9.4. Приоритет удаления переписки
Если сначала пришло удаление переписки, а потом удаление одного старого сообщения из этой переписки, сервер должен:
- проигнорировать это удаление одного сообщения;
- не создавать новых изменений поверх уже удалённой истории.
То же правило действует и для обычных сообщений, и для редактирований старых сообщений.
## 10. JSON API v1
### 10.1. `SendMessagePair`
Назначение:
- клиент отправляет новый DM;
- клиент отправляет редактирование старого DM;
- клиент отправляет парную новую ревизию типов `1/2`.
Request:
```json
{
"op": "SendMessagePair",
"requestId": "req-123",
"payload": {
"incomingBlobB64": "...",
"outgoingBlobB64": "..."
}
}
```
Правила:
- клиенту достаточно отправить пару на один любой доступный сервер;
- успешный `status=200` означает локальное сохранение;
- первая сетевая попытка выполняется до ответа клиенту, параллельно для двух серверов получателя;
- сервер после принятия сам отвечает за дальнейшую межсерверную доставку.
Ответ сохраняет прежние `baseKey`, `incomingKey`, `outgoingKey` и счётчики доставки в клиентские сессии. Единственное новое поле — `deliveryState`: `accepted`, `delivered` или `failed`.
### 10.2. `ReceiveIncomingMessage`
Назначение:
- приём одной входящей копии по схеме server-to-server;
- приём входящего редактирования;
- приём входящего read-receipt.
Request:
```json
{
"op": "ReceiveIncomingMessage",
"requestId": "req-456",
"payload": {
"incomingBlobB64": "...",
"sourceServerLogin": "server-a"
}
}
```
`sourceServerLogin` пока доверяется без отдельной межсерверной подписи. Сервер всё равно проверяет пользовательскую подпись самого `SHiNE_DM`. Для `ReceiveOutcomingMessage` и `SendMessagePair` пустой `sourceServerLogin` означает клиентский вызов, непустой - peer-вызов. Успешный ответ сохраняет старые поля `messageKey`, `baseKey` и счётчики realtime-доставки.
### 10.3. `DeleteMessage`
Назначение:
- удалить одно сообщение у обеих сторон.
Request:
```json
{
"op": "DeleteMessage",
"requestId": "req-789",
"payload": {
"blobB64": "..."
}
}
```
Ожидается контейнер типа:
- `5`, если удаление инициировал отправитель;
- `6`, если удаление инициировал получатель.
UI-следствие для клиента:
- пользователь может удалить как своё исходящее сообщение, так и входящее;
- для удаления входящего UI должен отправлять вариант, где удаление инициировал получатель (`type=6`).
### 10.4. `DeleteConversation`
Назначение:
- удалить всю переписку до времени самого служебного сообщения.
Request:
```json
{
"op": "DeleteConversation",
"requestId": "req-790",
"payload": {
"blobB64": "..."
}
}
```
Ожидается контейнер типа:
- `7`, если удаление инициировал отправитель;
- `8`, если удаление инициировал получатель.
UI-следствие для клиента:
- действие `Очистить историю` должно отправлять служебное сообщение удаления переписки;
- для обычной клиентской кнопки очистки истории допускается вариант удаления, инициированный текущим получателем (`type=8`);
- после применения такого сообщения UI может оставлять в чате видимую служебную точку отсечения истории;
- отдельное UI-действие `Удалить чат` может дополнительно спросить, нужно ли вместе с удалением контакта также отправить `DeleteConversation`.
### 10.5. `GetDirectMessages`
Назначение:
- постранично отдавать историю одного личного диалога;
- не вываливать весь старый backlog автоматически в момент логина;
- отдавать клиенту только полезные chat-сообщения, без технических receipt/tombstone блоков.
Правила:
- начиная с 22 июля 2026 года старая история DM больше не должна автоматически пушиться клиенту сразу после входа;
- клиент сам запрашивает первую страницу истории у нужного собеседника;
- последующие страницы клиент запрашивает по курсору `beforeTimeMs` + `beforeMessageKey`;
- realtime-новые сообщения по-прежнему приходят отдельно через `SignedMessageArrived`.
Содержимое страницы:
- сервер возвращает только контентные DM типов `1/2`;
- read-receipt (`3/4`) и tombstone удаления (`5/6/7/8`) в историю страницы не включаются;
- сообщения идут страницами от новых к старым;
- лимит страницы считается по самим chat-сообщениям диалога.
Поле прочтения:
- для каждого контентного сообщения сервер может вернуть `readAtMs`;
- `readAtMs` означает точное время прочтения сообщения, если оно уже известно серверу;
- если точное время для старого сообщения неизвестно, но по более новым данным видно, что сообщение уже точно прочитано, UI может показывать его как прочитанное без точного времени;
- read-receipt при этом остаётся отдельным DM-событием синхронизации, но в обычную историю страницы не подмешивается.
Для исходящих элементов сервер также возвращает единственное поле `deliveryState`.
## 11. Межсерверная доставка
### 11.1. Клиентская сторона
Клиенту достаточно отправить сообщение на:
- любой один доступный сервер.
### 11.2. Серверная сторона
После локального принятия пара получает изменяемое состояние `accepted`. Сервер сразу вызывает до двух текущих маршрутов получателя параллельно. Успех хотя бы одного маршрута переводит сообщение в терминальное `delivered`; ждать второй маршрут не требуется.
После первой попытки полная пара передаётся единственному второму access-серверу отправителя через прежний `ReceiveOutcomingMessage`. Peer самостоятельно пробует актуальные маршруты получателя; delivery-state в запрос не входит.
Идемпотентность обязательна. ACK означает запись в БД; повтор уже известной ревизии считается успешным ACK.
### 11.3. Догоняющая синхронизация истории
Для восстановления пропущенных DM-событий между access-серверами используется pull-операция:
- `DmSyncBatch`
Второй сервер запрашивает outbox-события владельца с `synced=false`, сохраняет их и передаёт подтверждённые `syncId` в `ackSyncIds` следующего запроса. Источник ставит `synced=true` только после ACK. Каждый цикл начинается с начала списка; курсор нужен только для страниц текущего цикла.
Полная пара отправителя передаётся двумя blob. Входящая копия получателя и tombstone передаются одним blob.
Полная пара отправителя передаётся одним элементом с двумя blob в порядке incoming/outgoing. Входящая копия получателя и tombstone передаются одним blob.
При применении событий, полученных через `DmSyncBatch`, сервер:
- проверяет формат `SHiNE_DM`;
- проверяет подпись;
- применяет существующие правила ревизий, read-receipt и tombstone;
- не отправляет realtime/push-уведомления клиентам;
- не запускает повторный fan-out, чтобы не создавать циклы.
Синхронизация настроек и DM выполняется одним периодическим процессом и через
один логический последовательный сеанс поверх постоянного WSS-пула. Закрытие
этого логического сеанса не закрывает физическое соединение с peer. Выборка DM
использует частичный индекс по `synced=false`.
В текущей реализации межсерверная авторизация DM ещё не включена. Принимающий сервер проверяет, что сам является access-сервером `ownerLogin`, и всегда проверяет пользовательские подписи signed-блоков.
### 11.4. Ошибки доставки
Если ни один сервер получателя не подтвердил запись, попытки выполняются сразу, через 30 секунд, 5 минут, 25 минут и 1 час от первоначального принятия.
Перед попытками через 5 минут, 25 минут и час сервер read-only спрашивает peer отправителя через `GetDmDeliveryStatus(messageKey)`. Если peer уже доставил хотя бы на один сервер, сообщение считается доставленным.
После неудачной последней попытки устанавливается `failed`; дальнейших автоматических попыток и кнопки ручного повтора нет.
## 12. Хранение в БД
Основная таблица остаётся:
- `signed_messages`
В ней должны сохраняться:
- обычные контентные DM;
- read-receipt DM;
- tombstone одного сообщения;
- tombstone удаления переписки.
Для контентных сообщений в БД дополнительно должно поддерживаться серверное поле `read_at_ms`, если для этого сообщения уже было принято read-receipt событие.
Сообщение об удалении одного сообщения хранится в БД и не удаляется физически, чтобы:
- защищать от повторного приёма старых версий;
- не терять факт удаления;
- корректно синхронизировать событие между серверами.
Сообщение об удалении переписки тоже хранится в БД, а старые сообщения до его времени из БД удаляются.
Изменяемая сетевая часть хранится отдельно:
- `dm_delivery_state` — состояние и расписание доставки исходящей пары;
- `dm_sync_outbox` — событие владельца и единственный флаг ACK `synced`;
- частичные индексы содержат только due/unsynced строки.
Legacy-таблица `dm_sync_peer_state` после миграции v12 физически остаётся для безопасной установки ZIP-накладки, но новым DM-кодом не используется.
## 13. Что обязательно должно измениться в коде относительно v0.5
- сервер не должен требовать одинаковый `encryptedBody` у `type=1` и `type=2`;
- сервер не должен трактовать `encryptedBody` как обычный UTF-8 текст;
- сервер не должен валидировать внутреннюю crypto-структуру `body` у контентных DM `type=1/2`;
- DM должны реально шифроваться end-to-end;
- удаление одного сообщения должно стать терминальным tombstone;
- удаление всей переписки должно стать отдельным служебным событием;
- межсерверная маршрутизация DM должна идти через `access_servers`;
- сервер должен добирать отсутствующих пользователей из Solana PDA до проверки подписи DM;
- при выборе актуальной версии должен учитываться `reencryptedAtMs`, если `revisionTimeMs` совпадает;
- логика должна быть безопасна для одного или двух серверов у каждой стороны.
## 14. Что в v1 пока не входит
- вложения в DM;
- хранение отдельного `keyId` шифрования в DM;
- ротация `clientKey`;
- финальная конкретная UI-реализация массовой перешифровки;
- межсерверная авторизация DM-синхронизации и доставки.
## 15. UI списка диалогов и загрузки истории
Актуальные правила клиентского интерфейса:
- превью диалога выбирается по самому новому контентному сообщению независимо от направления `in/out`;
- если локально уже есть более новое расшифрованное сообщение, чем в серверной сводке `ListContacts`, клиент использует локальный текст и время до следующего согласования с сервером;
- локальное сообщение сравнивается с `lastMessageTimeMs`, поэтому входящие и исходящие сообщения участвуют в выборе на равных;
- индикатор загрузки истории всегда является компактным элементом фиксированной высоты сверху списка и не растягивается на высоту пустого чата;
- клик по аватару или логину собеседника открывает меню перехода в профиль или связи этого пользователя.
Эти правила относятся только к UI и не изменяют сетевой или байтовый формат DM v1.
- байтовый формат SHiNE_DM;
- шифрование входящей и исходящей копий;
- пользовательские подписи;
- deliveryState и retry-воркер;
- read-receipts;
- синхронизацию пользовательских блокчейнов через server-PDA sync_servers;
- общий постоянный WSS ServerConnectionPool.
docs/Personal_Messages/Формат_DM_v1.md
+7 -1
View File
@@ -340,7 +340,13 @@ ReadReceiptBody_v1_0
В версии DM v1 все типы `1..8` используют единый контейнер `SHiNE_DM`.
Межсерверные операции `ReceiveOutcomingMessage`, `ReceiveIncomingMessage` и `DmSyncBatch` не вводят новый байтовый формат DM. Они передают уже сохранённые raw-контейнеры `SHiNE_DM` в Base64. `ackSyncIds` подтверждает только факт сохранения синхронизированной ревизии; delivery-state между серверами не передаётся. Принимающий сервер заново проверяет подпись и применяет контейнер по его `messageType`.
`SendMessagePair` передаёт входящую и исходящую подписанные копии от UI на
сервер отправителя. Внутренняя операция `ReceiveIncomingMessage` передаёт
только входящую raw-копию в Base64 на единственный access-сервер получателя.
Обе операции используют существующий байтовый формат `SHiNE_DM`.
Межсерверные `ReceiveOutcomingMessage` и `DmSyncBatch` удалены.
Принимающий сервер заново проверяет пользовательскую подпись и применяет
контейнер по его `messageType`.
## 14. Превью в списке диалогов
docs/Solana/SOLANA_USERS_SYNC_MODULE_DESIGN.md
+4 -2
View File
@@ -20,8 +20,10 @@
- сервер запускает sync-модуль до продолжения собственного startup;
- server startup ждёт входа модуля в состояние `READY`;
- актуальный источник истины по пользователям для нового PostgreSQL runtime-слоя: `solana_user_pda_current`.
- для быстрого локального маршрутизационного lookup серверов доступа поддерживается вторичная проекция
`user_access_servers_current`, которая автоматически пересобирается из `solana_user_pda_current`.
- для быстрого локального lookup единственного сервера доступа поддерживается
вторичная проекция `user_access_servers_current`, которая автоматически
пересобирается только из первого элемента `access_servers[0]` в
`solana_user_pda_current`.
---
docs/Solana/user_pda/README.md
+7 -2
View File
@@ -291,8 +291,13 @@ AccessServersBlock
- блок может отсутствовать, если серверы доступа не заданы;
- список может обновляться при изменении маршрутизации пользователя;
- `access_servers` - логины пользователей системы, используемых как серверы доступа/relay для конкретного пользователя. Через этот список клиентская и серверная логика SHiNE может маршрутизировать доставку личных сообщений, доступ к сессиям и другие пользовательские операции. Solana-программа не обязана проверять, что эти логины действительно зарегистрированы как серверы;
- точная семантика выбора сервера доступа определяется клиентской/серверной логикой SHiNE.
- `access_servers` — массив логинов серверов доступа/relay. Формат массива
сохранён для совместимости, но текущая клиентская и серверная логика использует
только первый элемент `access_servers[0]`;
- остальные элементы игнорируются без fallback;
- официальный UI записывает ровно один сервер и позволяет только заменить его;
- Solana-программа не обязана проверять, что логин действительно зарегистрирован
как сервер.
## 14. SessionsBlock
docs/Solana_Architecture/details/shine_users.md
+2 -1
View File
@@ -74,7 +74,8 @@
- `server_key: Pubkey`
- `server_address: String`
- `sync_servers: Vec<String>`
- `access_servers: Vec<String>`
- `access_servers: Vec<String>` — формат остаётся массивом, но runtime
использует только первый элемент
- `trusted_count: u8`
## Главные PDA
shine-UI/js/pages/access-servers-view.js
+28 -91
View File
@@ -6,8 +6,8 @@ import { resolveShineServerByServerLogin } from '../services/shine-server-resolv
import { readShineUserPda, updateShineUserPdaOnSolana } from '../services/shine-user-pda-service.js';
import { getTopupSiteUrl } from '../services/solana-wallet-service.js';
export const pageMeta = { id: 'access-servers-view', title: 'Серверы доступа' };
const MAX_ACCESS_SERVERS = 2;
export const pageMeta = { id: 'access-servers-view', title: 'Сервер доступа' };
const MAX_ACCESS_SERVERS = 1;
function escapeHtml(value) {
return String(value || '')
@@ -281,9 +281,10 @@ export function render({ navigate }) {
introCard.innerHTML = `
<p class="field-label">Где хранятся личные данные</p>
<p class="meta-muted">
Серверы доступа хранят зашифрованную личную переписку пользователя и участвуют в звонках.
Единственный сервер доступа хранит зашифрованную личную переписку пользователя и участвует в звонках.
Всё, что публикуется в блокчейне SHiNE, доступно через любой сервер Сияния,
а доступ пользователя и приватная переписка хранятся только на этих серверах.
а настройки и приватная переписка хранятся только на выбранном сервере.
При смене сервера прежняя переписка и настройки автоматически не переносятся.
</p>
`;
@@ -291,7 +292,7 @@ export function render({ navigate }) {
listCard.className = 'card stack';
const listTitle = document.createElement('p');
listTitle.className = 'field-label';
listTitle.textContent = 'Текущий список серверов доступа';
listTitle.textContent = 'Текущий сервер доступа';
const listHint = document.createElement('p');
listHint.className = 'meta-muted';
listHint.textContent = sessionLogin
@@ -308,10 +309,10 @@ export function render({ navigate }) {
addCard.className = 'card stack';
const addTitle = document.createElement('p');
addTitle.className = 'field-label';
addTitle.textContent = 'Добавить сервер доступа';
addTitle.textContent = 'Сменить сервер доступа';
const addHint = document.createElement('p');
addHint.className = 'meta-muted';
addHint.textContent = 'Можно использовать максимум два сервера доступа. Введите логин сервера или несколько первых букв, затем выберите сервер из подсказок.';
addHint.textContent = 'Введите логин нового сервера или несколько первых букв, затем выберите сервер из подсказок. Новый сервер заменит текущий.';
const addInput = document.createElement('input');
addInput.className = 'input';
addInput.type = 'text';
@@ -326,13 +327,12 @@ export function render({ navigate }) {
const addButton = document.createElement('button');
addButton.className = 'primary-btn';
addButton.type = 'button';
addButton.textContent = 'Добавить сервер';
addButton.textContent = 'Сменить сервер';
addButton.disabled = true;
addCard.append(addTitle, addHint, addInput, suggestEl, addStatus, addButton);
const refreshAddButton = () => {
addButton.disabled = operationBusy
|| currentAccessServers.length >= MAX_ACCESS_SERVERS
|| (!selectedCandidate && !normalizeLogin(addInput.value));
};
@@ -352,9 +352,7 @@ export function render({ navigate }) {
addInput.value = candidate.login;
addStatus.textContent = `Выбран сервер @${candidate.login}${candidate.url ? ` (${candidate.url})` : ''}`;
} else {
addStatus.textContent = currentAccessServers.length >= MAX_ACCESS_SERVERS
? 'У пользователя уже выбраны два сервера доступа. Сначала отключите один из них.'
: 'Для изменения списка понадобится подпись root key.';
addStatus.textContent = 'Для смены сервера понадобится подпись root key.';
}
};
@@ -368,69 +366,15 @@ export function render({ navigate }) {
return;
}
currentAccessServers.forEach((server, index) => {
const button = document.createElement('button');
button.className = 'text-btn';
button.type = 'button';
button.disabled = operationBusy;
button.innerHTML = `
currentAccessServers.forEach((server) => {
const row = document.createElement('div');
row.innerHTML = `
<span style="display:block; text-align:left;">
<strong>@${escapeHtml(server.login)}</strong>
<span class="meta-muted" style="display:block; margin-top:0.2rem;">${escapeHtml(server.url || 'URL не указан')}</span>
</span>
`;
button.addEventListener('click', () => {
if (operationBusy) return;
const isLast = currentAccessServers.length <= 1;
const performDisable = async () => {
const nextList = currentAccessServers
.map((item) => item.login)
.filter((login) => login !== server.login);
try {
await updateAccessServers(nextList, {
statusTarget: listStatus,
successText: `Сервер доступа @${server.login} отключён.`,
inFlightText: `Обновляем PDA и отключаем сервер @${server.login}...`,
});
} catch {
// Сообщение уже показано в статусе.
}
};
const askFinal = (noteText = '') => {
confirmModal?.open({
title: isLast ? 'Последний сервер доступа' : 'Отключить сервер доступа?',
text: isLast
? 'Это последний сервер доступа пользователя. Если его отключить, личная переписка пользователя на серверах доступа будет удалена.'
: `Хотите изменить запись в блокчейне Solana и отключить сервер доступа @${server.login}?`,
note: noteText,
confirmLabel: isLast ? 'Понимаю' : 'Да',
cancelLabel: isLast ? 'Отмена' : 'Нет',
onConfirm: performDisable,
});
};
if (isLast) {
confirmModal?.open({
title: 'Отключить последний сервер?',
text: `У пользователя остался только один сервер доступа: @${server.login}.`,
note: 'После отключения последнего сервера доступа личная переписка пользователя на серверах доступа будет удалена.',
confirmLabel: 'Продолжить',
cancelLabel: 'Отмена',
onConfirm: () => askFinal('Это повторное предупреждение перед записью нового списка серверов в Solana PDA.'),
});
return;
}
askFinal();
});
listBody.append(button);
if (index < currentAccessServers.length - 1) {
const divider = document.createElement('div');
divider.style.height = '1px';
divider.style.background = 'rgba(255,255,255,0.08)';
listBody.append(divider);
}
listBody.append(row);
});
};
@@ -465,16 +409,13 @@ export function render({ navigate }) {
renderServerList();
refreshAddButton();
listStatus.textContent = rows.length
? `Найдено серверов доступа: ${rows.length}`
? 'Сервер доступа загружен из PDA.'
: 'В PDA пользователя пока нет серверов доступа.';
if (rows.length >= MAX_ACCESS_SERVERS) {
addStatus.textContent = 'У пользователя уже выбраны два сервера доступа. Сначала отключите один из них.';
}
} catch (error) {
currentAccessServers = [];
renderServerList();
refreshAddButton();
listStatus.textContent = error?.message || 'Не удалось прочитать список серверов доступа.';
listStatus.textContent = error?.message || 'Не удалось прочитать сервер доступа.';
}
};
@@ -559,7 +500,7 @@ export function render({ navigate }) {
const passwordResult = await passwordModal?.open({
title: 'Нужен пароль для обновления серверов доступа',
text: 'Чтобы изменить список серверов доступа, нужно подписать обновление user PDA через root key. Этот ключ не найден в локальном зашифрованном контейнере устройства.',
text: 'Чтобы изменить сервер доступа, нужно подписать обновление user PDA через root key. Этот ключ не найден в локальном зашифрованном контейнере устройства.',
note: savedClient
? 'client key уже сохранён на устройстве. Из пароля будет восстановлен только root key.'
: 'На устройстве не хватает root key и/или client key. Они будут восстановлены из пароля аккаунта.',
@@ -604,7 +545,7 @@ export function render({ navigate }) {
const normalizedList = uniqueLogins(nextLogins);
setOperationBusy(true);
target.textContent = String(inFlightText || 'Обновляем список серверов доступа...');
target.textContent = String(inFlightText || 'Обновляем сервер доступа...');
let signingMaterial = null;
try {
const currentPda = await readShineUserPda({ login: sessionLogin, solanaEndpoint });
@@ -626,7 +567,7 @@ export function render({ navigate }) {
if (isInsufficientFundsForRentError(error)) {
showTopupRequiredStatus(target, signingMaterial?.clientAddress);
} else {
target.textContent = error?.message || 'Не удалось обновить список серверов доступа.';
target.textContent = error?.message || 'Не удалось обновить сервер доступа.';
}
throw error;
} finally {
@@ -662,29 +603,25 @@ export function render({ navigate }) {
return;
}
if (currentAccessServers.some((item) => item.login === login)) {
addStatus.textContent = `Сервер @${login} уже есть в списке.`;
return;
}
if (currentAccessServers.length >= MAX_ACCESS_SERVERS) {
addStatus.textContent = 'Нельзя добавить больше двух серверов доступа. Сначала отключите один из текущих.';
addStatus.textContent = `Сервер @${login} уже выбран.`;
return;
}
try {
const resolved = await resolveShineServerByServerLogin({ serverLogin: login, solanaEndpoint });
confirmModal?.open({
title: 'Добавить сервер доступа?',
text: `Вы хотите добавить сервер доступа @${resolved.serverLogin}?`,
title: 'Сменить сервер доступа?',
text: `Заменить текущий сервер доступа на @${resolved.serverLogin}?`,
note: resolved.httpBase
? `Адрес сервера: ${resolved.httpBase}\nИзменение будет записано в Solana user PDA.`
: 'Изменение будет записано в Solana user PDA.',
? `Адрес сервера: ${resolved.httpBase}\nПрежняя переписка и настройки автоматически не перенесутся. Изменение будет записано в Solana user PDA.`
: 'Прежняя переписка и настройки автоматически не перенесутся. Изменение будет записано в Solana user PDA.',
onConfirm: async () => {
const nextList = uniqueLogins([...currentAccessServers.map((item) => item.login), resolved.serverLogin]);
const nextList = [resolved.serverLogin];
try {
await updateAccessServers(nextList, {
statusTarget: addStatus,
successText: `Сервер доступа @${resolved.serverLogin} добавлен.`,
inFlightText: `Обновляем PDA и добавляем сервер @${resolved.serverLogin}...`,
successText: `Сервер доступа заменён на @${resolved.serverLogin}.`,
inFlightText: `Обновляем PDA и меняем сервер на @${resolved.serverLogin}...`,
});
} catch {
// Сообщение уже показано в статусе.
@@ -698,7 +635,7 @@ export function render({ navigate }) {
screen.append(
renderHeader({
title: 'Серверы доступа',
title: 'Сервер доступа',
leftAction: { label: '←', onClick: () => navigate('settings-view') },
}),
introCard,
shine-UI/js/pages/settings-view.js
+1 -1
View File
@@ -44,7 +44,7 @@ export function render({ navigate }) {
<button class="text-btn" type="button" id="settings-remote-addblock">Remote AddBlock через homeserver</button>
<button class="text-btn" type="button" id="settings-access-servers">
<span style="display:block; text-align:left;">
<strong>Серверы доступа</strong>
<strong>Сервер доступа</strong>
<span class="meta-muted" style="display:block; margin-top:0.2rem;">Личная переписка, звонки и зашифрованные данные</span>
</span>
</button>
shine-UI/js/services/auth-service.js
+2 -10
View File
@@ -2611,14 +2611,8 @@ export class AuthService {
async sendMessagePair({ incomingBlobB64, outgoingBlobB64 }) {
const body = { incomingBlobB64, outgoingBlobB64 };
const primaryOp = 'ReceiveOutcomingMessage';
let response = await this.ws.request(primaryOp, body);
if (response.status === 404) {
response = await this.ws.request('SendMessagePair', body);
if (response.status !== 200) throw opError('SendMessagePair', response);
return response.payload || {};
}
if (response.status !== 200) throw opError(primaryOp, response);
const response = await this.ws.request('SendMessagePair', body);
if (response.status !== 200) throw opError('SendMessagePair', response);
return response.payload || {};
}
@@ -2938,7 +2932,6 @@ export class AuthService {
valueText = '',
valueNum = 0,
storagePwd,
syncDelivery = false,
}) {
const cleanLogin = String(login || '').trim();
const cleanSettingKey = String(settingKey || '').trim();
@@ -2977,7 +2970,6 @@ export class AuthService {
value_num: Math.trunc(cleanValueNum),
client_key: clientKey,
signature,
sync_delivery: !!syncDelivery,
});
if (response.status !== 200) throw opError('UpsertUserSetting', response);
return response.payload || {};
shine-UI/js/services/shine-server-resolver.js
+1 -1
View File
@@ -69,7 +69,7 @@ export async function resolveShineServerForUserLogin({ login, solanaEndpoint, fa
solanaEndpoint,
});
const accessServers = Array.isArray(parsedUser?.accessServers)
? parsedUser.accessServers.map((value) => normalizeShineServerLogin(value)).filter(Boolean).slice(0, 2)
? parsedUser.accessServers.map((value) => normalizeShineServerLogin(value)).filter(Boolean).slice(0, 1)
: [];
const pickedServerLogin = accessServers[0] || normalizeShineServerLogin(fallbackServerLogin);
if (!pickedServerLogin) {
shine-UI/js/services/shine-user-pda-service.js
+1 -1
View File
@@ -30,7 +30,7 @@ const BLOCK_TYPE_TRUSTED_STATE = 70;
const SESSIONS_MODE_MIXED = 1;
const SESSION_TYPE_USER = 1;
const SESSION_TYPE_HOMESERVER = 100;
const MAX_EFFECTIVE_ACCESS_SERVERS = 2;
const MAX_EFFECTIVE_ACCESS_SERVERS = 1;
let solanaLibPromise = null;
function loadSolanaLib() {
shine-solana/shine/doc/formats/shine-user-pda-format-v.1.0.md
+7 -2
View File
@@ -291,8 +291,13 @@ AccessServersBlock
- блок может отсутствовать, если серверы доступа не заданы;
- список может обновляться при изменении маршрутизации пользователя;
- `access_servers` - логины пользователей системы, используемых как серверы доступа/relay для конкретного пользователя. Через этот список клиентская и серверная логика SHiNE может маршрутизировать доставку личных сообщений, доступ к сессиям и другие пользовательские операции. Solana-программа не обязана проверять, что эти логины действительно зарегистрированы как серверы;
- точная семантика выбора сервера доступа определяется клиентской/серверной логикой SHiNE.
- `access_servers` — массив логинов серверов доступа/relay. Формат массива
сохранён для совместимости, но текущая клиентская и серверная логика использует
только первый элемент `access_servers[0]`;
- остальные элементы игнорируются без fallback;
- официальный UI записывает ровно один сервер и позволяет только заменить его;
- Solana-программа не обязана проверять, что логин действительно зарегистрирован
как сервер.
## 14. SessionsBlock