Compare commits

...
2 Commits
Author SHA256 Message Date
AidarKC 01321f5200 Добавили счётчик подписчиков в каналах и подписки пользователя 2026-08-26 11:11:46 +04:00
AidarKC 3a5851939e Починить межсерверную репликацию личных сообщений
Репликация личных сообщений между серверами теперь работает корректно.
2026-08-25 19:31:38 +04:00
49 changed files with 2233 additions and 557 deletions
@@ -30,6 +30,8 @@ public final class DatabaseInitializer {
public static final int SCHEMA_VERSION_11 = 11;
public static final int SCHEMA_VERSION_12 = 12;
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 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";
@@ -43,6 +45,8 @@ public final class DatabaseInitializer {
public static final String POSTGRES_MIGRATION_V11_RESOURCE = "postgres/migration_v11.sql";
public static final String POSTGRES_MIGRATION_V12_RESOURCE = "postgres/migration_v12.sql";
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";
private DatabaseInitializer() {}
@@ -154,6 +158,14 @@ public final class DatabaseInitializer {
runSqlScript(conn, POSTGRES_MIGRATION_V13_RESOURCE);
currentVersion = SCHEMA_VERSION_13;
}
if (currentVersion < SCHEMA_VERSION_14) {
runSqlScript(conn, POSTGRES_MIGRATION_V14_RESOURCE);
currentVersion = SCHEMA_VERSION_14;
}
if (currentVersion < SCHEMA_VERSION_15) {
runSqlScript(conn, POSTGRES_MIGRATION_V15_RESOURCE);
currentVersion = SCHEMA_VERSION_15;
}
}
}
@@ -99,6 +99,8 @@ public final class BlockchainResyncCleanupDAO {
int deletedBlocks = deleteBlocksForChain(c, blockchainName);
int deletedBlockchainState = deleteBlockchainStateForChain(c, blockchainName);
rebuildStatsState(c);
c.commit();
return new CleanupResult(
@@ -366,6 +368,100 @@ public final class BlockchainResyncCleanupDAO {
""", blockchainName);
}
private void rebuildStatsState(Connection c) throws SQLException {
try (PreparedStatement truncate = c.prepareStatement("""
TRUNCATE TABLE user_stats_state, channel_stats_state
""")) {
truncate.executeUpdate();
}
try (PreparedStatement ps = c.prepareStatement("""
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
)
SELECT
u.login,
COALESCE(own.owned_public_channels_count, 0),
COALESCE(fu.following_users_count, 0),
COALESCE(fc.following_channels_count, 0),
COALESCE(cf.close_friends_count, 0),
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
FROM solana_user_pda_current u
LEFT JOIN (
SELECT owner_login, COUNT(*)::INTEGER AS owned_public_channels_count
FROM channel_names_state
WHERE channel_type_code = 1
GROUP BY owner_login
) own ON LOWER(own.owner_login) = LOWER(u.login)
LEFT JOIN (
SELECT login, COUNT(*)::INTEGER AS following_users_count
FROM connections_state
WHERE rel_type = 30
AND to_block_number = 0
GROUP BY login
) fu ON LOWER(fu.login) = LOWER(u.login)
LEFT JOIN (
SELECT cs.login, COUNT(*)::INTEGER AS following_channels_count
FROM connections_state cs
JOIN channel_names_state cn
ON cn.owner_bch_name = cs.to_bch_name
AND cn.channel_root_block_number = cs.to_block_number
AND cn.channel_root_block_hash = cs.to_block_hash
WHERE cs.rel_type = 30
AND cn.channel_type_code = 1
GROUP BY cs.login
) fc ON LOWER(fc.login) = LOWER(u.login)
LEFT JOIN (
SELECT login, COUNT(*)::INTEGER AS close_friends_count
FROM connections_state
WHERE rel_type = 10
GROUP BY login
) cf ON LOWER(cf.login) = LOWER(u.login)
""")) {
ps.executeUpdate();
}
try (PreparedStatement ps = c.prepareStatement("""
INSERT INTO channel_stats_state (
owner_bch_name,
channel_root_block_number,
channel_root_block_hash,
owner_login,
channel_type_code,
subscribers_count,
updated_at_ms
)
SELECT
cn.owner_bch_name,
cn.channel_root_block_number,
cn.channel_root_block_hash,
cn.owner_login,
cn.channel_type_code,
COUNT(DISTINCT cs.login)::INTEGER AS subscribers_count,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
FROM channel_names_state cn
LEFT JOIN connections_state cs
ON cs.rel_type = 30
AND cs.to_bch_name = cn.owner_bch_name
AND cs.to_block_number = cn.channel_root_block_number
AND cs.to_block_hash = cn.channel_root_block_hash
WHERE cn.channel_type_code = 1
GROUP BY
cn.owner_bch_name,
cn.channel_root_block_number,
cn.channel_root_block_hash,
cn.owner_login,
cn.channel_type_code
""")) {
ps.executeUpdate();
}
}
private int executeDelete(Connection c, String sql, String value) throws SQLException {
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setString(1, value);
@@ -75,6 +75,35 @@ public final class UserAccessServersCurrentDAO {
return result;
}
/**
* Возвращает уникальные физические серверы из текущей routing-проекции.
* Используется транспортным пулом: одно WSS-соединение держится на сервер,
* а не на каждого пользователя этого сервера.
*/
public List<UserAccessServerRouteEntry> listDistinctServers() throws SQLException {
String sql = """
SELECT DISTINCT ON (LOWER(server_login))
server_login, server_url
FROM user_access_servers_current
WHERE server_login IS NOT NULL AND BTRIM(server_login) <> ''
AND server_url IS NOT NULL AND BTRIM(server_url) <> ''
ORDER BY LOWER(server_login), server_login, server_url
""";
List<UserAccessServerRouteEntry> result = new ArrayList<>();
try (Connection c = db.getConnection();
PreparedStatement ps = c.prepareStatement(sql);
ResultSet rs = ps.executeQuery()) {
while (rs.next()) {
UserAccessServerRouteEntry entry = new UserAccessServerRouteEntry();
entry.setUserLogin("");
entry.setServerLogin(rs.getString("server_login"));
entry.setServerUrl(rs.getString("server_url"));
result.add(entry);
}
}
return result;
}
private UserAccessServerRouteEntry mapRow(ResultSet rs) throws SQLException {
UserAccessServerRouteEntry entry = new UserAccessServerRouteEntry();
entry.setUserLogin(rs.getString("user_login"));
@@ -0,0 +1,620 @@
BEGIN;
CREATE TABLE IF NOT EXISTS user_stats_state (
login TEXT PRIMARY KEY REFERENCES solana_user_pda_current(login) ON DELETE CASCADE,
owned_public_channels_count INTEGER NOT NULL DEFAULT 0 CHECK (owned_public_channels_count >= 0),
following_users_count INTEGER NOT NULL DEFAULT 0 CHECK (following_users_count >= 0),
following_channels_count INTEGER NOT NULL DEFAULT 0 CHECK (following_channels_count >= 0),
close_friends_count INTEGER NOT NULL DEFAULT 0 CHECK (close_friends_count >= 0),
updated_at_ms BIGINT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_user_stats_state_following_users_count
ON user_stats_state (following_users_count);
CREATE TABLE IF NOT EXISTS channel_stats_state (
owner_bch_name TEXT NOT NULL,
channel_root_block_number INTEGER NOT NULL CHECK (channel_root_block_number >= 0),
channel_root_block_hash BYTEA NOT NULL,
owner_login TEXT NOT NULL,
channel_type_code INTEGER NOT NULL DEFAULT 1,
subscribers_count INTEGER NOT NULL DEFAULT 0 CHECK (subscribers_count >= 0),
updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (owner_bch_name, channel_root_block_number, channel_root_block_hash)
);
CREATE INDEX IF NOT EXISTS idx_channel_stats_state_owner_login
ON channel_stats_state (owner_login);
CREATE OR REPLACE FUNCTION shine_user_stats_state_ai()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
DECLARE
now_ms BIGINT;
BEGIN
now_ms := CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT);
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
0,
0,
now_ms
)
ON CONFLICT (login) DO NOTHING;
RETURN NEW;
END;
$$;
CREATE OR REPLACE FUNCTION shine_channel_names_state_stats_ai()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
DECLARE
now_ms BIGINT;
public_subscribers_count INTEGER;
BEGIN
now_ms := CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT);
IF NEW.owner_login IS NULL OR btrim(NEW.owner_login) = '' THEN
RETURN NEW;
END IF;
IF NEW.channel_type_code = 1 THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.owner_login,
1,
0,
0,
0,
now_ms
)
ON CONFLICT (login) DO UPDATE SET
owned_public_channels_count = user_stats_state.owned_public_channels_count + 1,
updated_at_ms = EXCLUDED.updated_at_ms;
SELECT COUNT(*)::INTEGER
INTO public_subscribers_count
FROM connections_state cs
JOIN channel_names_state cn
ON cn.owner_bch_name = cs.to_bch_name
AND cn.channel_root_block_number = cs.to_block_number
AND cn.channel_root_block_hash = cs.to_block_hash
WHERE cs.rel_type = 30
AND cn.owner_bch_name = NEW.owner_bch_name
AND cn.channel_root_block_number = NEW.channel_root_block_number
AND cn.channel_root_block_hash = NEW.channel_root_block_hash
AND cn.channel_type_code = 1;
INSERT INTO channel_stats_state (
owner_bch_name,
channel_root_block_number,
channel_root_block_hash,
owner_login,
channel_type_code,
subscribers_count,
updated_at_ms
) VALUES (
NEW.owner_bch_name,
NEW.channel_root_block_number,
NEW.channel_root_block_hash,
NEW.owner_login,
NEW.channel_type_code,
COALESCE(public_subscribers_count, 0),
now_ms
)
ON CONFLICT (owner_bch_name, channel_root_block_number, channel_root_block_hash) DO UPDATE SET
owner_login = EXCLUDED.owner_login,
channel_type_code = EXCLUDED.channel_type_code,
subscribers_count = EXCLUDED.subscribers_count,
updated_at_ms = EXCLUDED.updated_at_ms;
ELSE
WITH pending AS (
SELECT cs.login, COUNT(*)::INTEGER AS cnt
FROM connections_state cs
WHERE cs.rel_type = 30
AND cs.to_bch_name = NEW.owner_bch_name
AND cs.to_block_number = NEW.channel_root_block_number
AND cs.to_block_hash = NEW.channel_root_block_hash
GROUP BY cs.login
)
UPDATE user_stats_state us
SET following_channels_count = GREATEST(0, us.following_channels_count - pending.cnt),
updated_at_ms = now_ms
FROM pending
WHERE us.login = pending.login;
END IF;
RETURN NEW;
END;
$$;
CREATE OR REPLACE FUNCTION shine_blocks_connection_state_ai()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
DECLARE
resolved_login TEXT;
positive_rel_type INTEGER;
existed_before BOOLEAN;
target_channel_type INTEGER;
BEGIN
IF NEW.msg_type <> 3 THEN
RETURN NEW;
END IF;
resolved_login := shine_resolve_login(NEW.to_login, NEW.to_bch_name);
IF NEW.msg_sub_type IN (10, 20, 30, 40, 50, 52, 54, 60, 70, 74) THEN
IF resolved_login IS NULL OR NEW.to_bch_name IS NULL THEN
RETURN NEW;
END IF;
SELECT EXISTS (
SELECT 1
FROM connections_state
WHERE login = NEW.login
AND rel_type = NEW.msg_sub_type
AND to_login = resolved_login
AND to_bch_name = NEW.to_bch_name
AND to_block_number = COALESCE(NEW.to_block_number, 0)
AND to_block_hash = COALESCE(NEW.to_block_hash, decode(repeat('00', 32), 'hex'))
)
INTO existed_before;
IF NOT existed_before THEN
IF NEW.msg_sub_type = 10 THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
0,
1,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
close_friends_count = user_stats_state.close_friends_count + 1,
updated_at_ms = EXCLUDED.updated_at_ms;
ELSIF NEW.msg_sub_type = 30 THEN
IF NEW.to_block_number IS NULL THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
1,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_channels_count = user_stats_state.following_channels_count + 1,
updated_at_ms = EXCLUDED.updated_at_ms;
ELSIF NEW.to_block_number = 0 THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
1,
0,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_users_count = user_stats_state.following_users_count + 1,
updated_at_ms = EXCLUDED.updated_at_ms;
ELSE
SELECT cn.channel_type_code
INTO target_channel_type
FROM channel_names_state cn
WHERE cn.owner_bch_name = NEW.to_bch_name
AND cn.channel_root_block_number = NEW.to_block_number
AND cn.channel_root_block_hash = NEW.to_block_hash
LIMIT 1;
IF target_channel_type = 1 THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
1,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_channels_count = user_stats_state.following_channels_count + 1,
updated_at_ms = EXCLUDED.updated_at_ms;
INSERT INTO channel_stats_state (
owner_bch_name,
channel_root_block_number,
channel_root_block_hash,
owner_login,
channel_type_code,
subscribers_count,
updated_at_ms
) VALUES (
NEW.to_bch_name,
NEW.to_block_number,
NEW.to_block_hash,
resolved_login,
target_channel_type,
1,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (owner_bch_name, channel_root_block_number, channel_root_block_hash) DO UPDATE SET
owner_login = EXCLUDED.owner_login,
channel_type_code = EXCLUDED.channel_type_code,
subscribers_count = channel_stats_state.subscribers_count + 1,
updated_at_ms = EXCLUDED.updated_at_ms;
ELSIF target_channel_type IS NULL THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
1,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_channels_count = user_stats_state.following_channels_count + 1,
updated_at_ms = EXCLUDED.updated_at_ms;
END IF;
END IF;
END IF;
END IF;
DELETE FROM connections_state
WHERE login = NEW.login
AND rel_type = NEW.msg_sub_type
AND to_login = resolved_login
AND to_bch_name = NEW.to_bch_name
AND to_block_number = COALESCE(NEW.to_block_number, 0)
AND to_block_hash = COALESCE(NEW.to_block_hash, decode(repeat('00', 32), 'hex'));
INSERT INTO connections_state (
login, rel_type, to_login, to_bch_name, to_block_number, to_block_hash
) VALUES (
NEW.login,
NEW.msg_sub_type,
resolved_login,
NEW.to_bch_name,
COALESCE(NEW.to_block_number, 0),
COALESCE(NEW.to_block_hash, decode(repeat('00', 32), 'hex'))
);
RETURN NEW;
END IF;
positive_rel_type := CASE NEW.msg_sub_type
WHEN 11 THEN 10
WHEN 21 THEN 20
WHEN 31 THEN 30
WHEN 41 THEN 40
WHEN 51 THEN 50
WHEN 53 THEN 52
WHEN 55 THEN 54
WHEN 61 THEN 60
WHEN 71 THEN 70
WHEN 75 THEN 74
ELSE NULL
END;
IF positive_rel_type IS NULL OR resolved_login IS NULL THEN
RETURN NEW;
END IF;
SELECT EXISTS (
SELECT 1
FROM connections_state
WHERE login = NEW.login
AND rel_type = positive_rel_type
AND to_login = resolved_login
AND to_bch_name = NEW.to_bch_name
AND to_block_number = COALESCE(NEW.to_block_number, 0)
AND to_block_hash = COALESCE(NEW.to_block_hash, decode(repeat('00', 32), 'hex'))
)
INTO existed_before;
IF NOT existed_before THEN
RETURN NEW;
END IF;
IF positive_rel_type = 10 THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
0,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
close_friends_count = GREATEST(0, user_stats_state.close_friends_count - 1),
updated_at_ms = EXCLUDED.updated_at_ms;
ELSIF positive_rel_type = 30 THEN
IF NEW.to_block_number IS NULL THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
0,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_channels_count = GREATEST(0, user_stats_state.following_channels_count - 1),
updated_at_ms = EXCLUDED.updated_at_ms;
ELSIF NEW.to_block_number = 0 THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
0,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_users_count = GREATEST(0, user_stats_state.following_users_count - 1),
updated_at_ms = EXCLUDED.updated_at_ms;
ELSE
SELECT cn.channel_type_code
INTO target_channel_type
FROM channel_names_state cn
WHERE cn.owner_bch_name = NEW.to_bch_name
AND cn.channel_root_block_number = NEW.to_block_number
AND cn.channel_root_block_hash = NEW.to_block_hash
LIMIT 1;
IF target_channel_type = 1 THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
0,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_channels_count = GREATEST(0, user_stats_state.following_channels_count - 1),
updated_at_ms = EXCLUDED.updated_at_ms;
INSERT INTO channel_stats_state (
owner_bch_name,
channel_root_block_number,
channel_root_block_hash,
owner_login,
channel_type_code,
subscribers_count,
updated_at_ms
) VALUES (
NEW.to_bch_name,
NEW.to_block_number,
NEW.to_block_hash,
resolved_login,
target_channel_type,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (owner_bch_name, channel_root_block_number, channel_root_block_hash) DO UPDATE SET
owner_login = EXCLUDED.owner_login,
channel_type_code = EXCLUDED.channel_type_code,
subscribers_count = GREATEST(0, channel_stats_state.subscribers_count - 1),
updated_at_ms = EXCLUDED.updated_at_ms;
ELSIF target_channel_type IS NULL THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
0,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_channels_count = GREATEST(0, user_stats_state.following_channels_count - 1),
updated_at_ms = EXCLUDED.updated_at_ms;
END IF;
END IF;
END IF;
DELETE FROM connections_state
WHERE login = NEW.login
AND rel_type = positive_rel_type
AND to_login = resolved_login
AND to_bch_name = NEW.to_bch_name
AND to_block_number = COALESCE(NEW.to_block_number, 0)
AND to_block_hash = COALESCE(NEW.to_block_hash, decode(repeat('00', 32), 'hex'));
RETURN NEW;
END;
$$;
DROP TRIGGER IF EXISTS trg_user_stats_state_ai ON solana_user_pda_current;
CREATE TRIGGER trg_user_stats_state_ai
AFTER INSERT ON solana_user_pda_current
FOR EACH ROW
EXECUTE FUNCTION shine_user_stats_state_ai();
DROP TRIGGER IF EXISTS trg_channel_names_state_stats_ai ON channel_names_state;
CREATE TRIGGER trg_channel_names_state_stats_ai
AFTER INSERT ON channel_names_state
FOR EACH ROW
EXECUTE FUNCTION shine_channel_names_state_stats_ai();
DROP TRIGGER IF EXISTS trg_blocks_connection_state_ai ON blocks;
CREATE TRIGGER trg_blocks_connection_state_ai
AFTER INSERT ON blocks
FOR EACH ROW
EXECUTE FUNCTION shine_blocks_connection_state_ai();
TRUNCATE TABLE user_stats_state, channel_stats_state;
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
)
SELECT
u.login,
COALESCE(own.owned_public_channels_count, 0) AS owned_public_channels_count,
COALESCE(fu.following_users_count, 0) AS following_users_count,
COALESCE(fc.following_channels_count, 0) AS following_channels_count,
COALESCE(cf.close_friends_count, 0) AS close_friends_count,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT) AS updated_at_ms
FROM solana_user_pda_current u
LEFT JOIN (
SELECT owner_login, COUNT(*)::INTEGER AS owned_public_channels_count
FROM channel_names_state
WHERE channel_type_code = 1
GROUP BY owner_login
) own ON LOWER(own.owner_login) = LOWER(u.login)
LEFT JOIN (
SELECT login, COUNT(*)::INTEGER AS following_users_count
FROM connections_state
WHERE rel_type = 30
AND to_block_number = 0
GROUP BY login
) fu ON LOWER(fu.login) = LOWER(u.login)
LEFT JOIN (
SELECT cs.login, COUNT(*)::INTEGER AS following_channels_count
FROM connections_state cs
JOIN channel_names_state cn
ON cn.owner_bch_name = cs.to_bch_name
AND cn.channel_root_block_number = cs.to_block_number
AND cn.channel_root_block_hash = cs.to_block_hash
WHERE cs.rel_type = 30
AND cn.channel_type_code = 1
GROUP BY cs.login
) fc ON LOWER(fc.login) = LOWER(u.login)
LEFT JOIN (
SELECT login, COUNT(*)::INTEGER AS close_friends_count
FROM connections_state
WHERE rel_type = 10
GROUP BY login
) cf ON LOWER(cf.login) = LOWER(u.login);
INSERT INTO channel_stats_state (
owner_bch_name,
channel_root_block_number,
channel_root_block_hash,
owner_login,
channel_type_code,
subscribers_count,
updated_at_ms
)
SELECT
cn.owner_bch_name,
cn.channel_root_block_number,
cn.channel_root_block_hash,
cn.owner_login,
cn.channel_type_code,
COUNT(DISTINCT cs.login)::INTEGER AS subscribers_count,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT) AS updated_at_ms
FROM channel_names_state cn
LEFT JOIN connections_state cs
ON cs.rel_type = 30
AND cs.to_bch_name = cn.owner_bch_name
AND cs.to_block_number = cn.channel_root_block_number
AND cs.to_block_hash = cn.channel_root_block_hash
WHERE cn.channel_type_code = 1
GROUP BY
cn.owner_bch_name,
cn.channel_root_block_number,
cn.channel_root_block_hash,
cn.owner_login,
cn.channel_type_code;
INSERT INTO db_schema_version (id, schema_version, updated_at_ms)
VALUES (1, 14, 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;
@@ -0,0 +1,105 @@
BEGIN;
CREATE OR REPLACE FUNCTION shine_channel_names_state_stats_ai()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
DECLARE
now_ms BIGINT;
public_subscribers_count INTEGER;
BEGIN
now_ms := CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT);
IF NEW.owner_login IS NULL OR btrim(NEW.owner_login) = '' THEN
RETURN NEW;
END IF;
IF NEW.channel_type_code = 1 THEN
IF EXISTS (
SELECT 1
FROM solana_user_pda_current su
WHERE su.login = NEW.owner_login
) THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.owner_login,
1,
0,
0,
0,
now_ms
)
ON CONFLICT (login) DO UPDATE SET
owned_public_channels_count = user_stats_state.owned_public_channels_count + 1,
updated_at_ms = EXCLUDED.updated_at_ms;
END IF;
SELECT COUNT(*)::INTEGER
INTO public_subscribers_count
FROM connections_state cs
JOIN channel_names_state cn
ON cn.owner_bch_name = cs.to_bch_name
AND cn.channel_root_block_number = cs.to_block_number
AND cn.channel_root_block_hash = cs.to_block_hash
WHERE cs.rel_type = 30
AND cn.owner_bch_name = NEW.owner_bch_name
AND cn.channel_root_block_number = NEW.channel_root_block_number
AND cn.channel_root_block_hash = NEW.channel_root_block_hash
AND cn.channel_type_code = 1;
INSERT INTO channel_stats_state (
owner_bch_name,
channel_root_block_number,
channel_root_block_hash,
owner_login,
channel_type_code,
subscribers_count,
updated_at_ms
) VALUES (
NEW.owner_bch_name,
NEW.channel_root_block_number,
NEW.channel_root_block_hash,
NEW.owner_login,
NEW.channel_type_code,
COALESCE(public_subscribers_count, 0),
now_ms
)
ON CONFLICT (owner_bch_name, channel_root_block_number, channel_root_block_hash) DO UPDATE SET
owner_login = EXCLUDED.owner_login,
channel_type_code = EXCLUDED.channel_type_code,
subscribers_count = EXCLUDED.subscribers_count,
updated_at_ms = EXCLUDED.updated_at_ms;
ELSE
WITH pending AS (
SELECT cs.login, COUNT(*)::INTEGER AS cnt
FROM connections_state cs
WHERE cs.rel_type = 30
AND cs.to_bch_name = NEW.owner_bch_name
AND cs.to_block_number = NEW.channel_root_block_number
AND cs.to_block_hash = NEW.channel_root_block_hash
GROUP BY cs.login
)
UPDATE user_stats_state us
SET following_channels_count = GREATEST(0, us.following_channels_count - pending.cnt),
updated_at_ms = now_ms
FROM pending
WHERE us.login = pending.login;
END IF;
RETURN NEW;
END;
$$;
INSERT INTO db_schema_version (id, schema_version, updated_at_ms)
VALUES (1, 15, 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;
@@ -291,6 +291,12 @@ AFTER TRUNCATE ON solana_user_pda_current
FOR EACH STATEMENT
EXECUTE FUNCTION trg_refresh_user_access_servers_from_user_pda_truncate();
DROP TRIGGER IF EXISTS trg_user_stats_state_ai ON solana_user_pda_current;
CREATE TRIGGER trg_user_stats_state_ai
AFTER INSERT ON solana_user_pda_current
FOR EACH ROW
EXECUTE FUNCTION shine_user_stats_state_ai();
CREATE TABLE IF NOT EXISTS solana_user_pda_history (
id BIGSERIAL PRIMARY KEY,
tx_signature TEXT NOT NULL,
@@ -605,6 +611,32 @@ CREATE UNIQUE INDEX IF NOT EXISTS uq_channel_names_state_target
CREATE INDEX IF NOT EXISTS idx_channel_names_state_owner
ON channel_names_state (owner_login, owner_bch_name);
CREATE TABLE IF NOT EXISTS user_stats_state (
login TEXT PRIMARY KEY REFERENCES solana_user_pda_current(login) ON DELETE CASCADE,
owned_public_channels_count INTEGER NOT NULL DEFAULT 0 CHECK (owned_public_channels_count >= 0),
following_users_count INTEGER NOT NULL DEFAULT 0 CHECK (following_users_count >= 0),
following_channels_count INTEGER NOT NULL DEFAULT 0 CHECK (following_channels_count >= 0),
close_friends_count INTEGER NOT NULL DEFAULT 0 CHECK (close_friends_count >= 0),
updated_at_ms BIGINT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_user_stats_state_following_users_count
ON user_stats_state (following_users_count);
CREATE TABLE IF NOT EXISTS channel_stats_state (
owner_bch_name TEXT NOT NULL,
channel_root_block_number INTEGER NOT NULL CHECK (channel_root_block_number >= 0),
channel_root_block_hash BYTEA NOT NULL,
owner_login TEXT NOT NULL,
channel_type_code INTEGER NOT NULL DEFAULT 1,
subscribers_count INTEGER NOT NULL DEFAULT 0 CHECK (subscribers_count >= 0),
updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (owner_bch_name, channel_root_block_number, channel_root_block_hash)
);
CREATE INDEX IF NOT EXISTS idx_channel_stats_state_owner_login
ON channel_stats_state (owner_login);
CREATE TABLE IF NOT EXISTS chat200_state (
owner_login TEXT NOT NULL,
owner_bch_name TEXT NOT NULL,
@@ -921,6 +953,132 @@ BEGIN
END;
$$;
CREATE OR REPLACE FUNCTION shine_user_stats_state_ai()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
DECLARE
now_ms BIGINT;
BEGIN
now_ms := CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT);
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
0,
0,
now_ms
)
ON CONFLICT (login) DO NOTHING;
RETURN NEW;
END;
$$;
CREATE OR REPLACE FUNCTION shine_channel_names_state_stats_ai()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
DECLARE
now_ms BIGINT;
public_subscribers_count INTEGER;
BEGIN
now_ms := CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT);
IF NEW.owner_login IS NULL OR btrim(NEW.owner_login) = '' THEN
RETURN NEW;
END IF;
IF NEW.channel_type_code = 1 THEN
IF EXISTS (
SELECT 1
FROM solana_user_pda_current su
WHERE su.login = NEW.owner_login
) THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.owner_login,
1,
0,
0,
0,
now_ms
)
ON CONFLICT (login) DO UPDATE SET
owned_public_channels_count = user_stats_state.owned_public_channels_count + 1,
updated_at_ms = EXCLUDED.updated_at_ms;
END IF;
SELECT COUNT(*)::INTEGER
INTO public_subscribers_count
FROM connections_state cs
JOIN channel_names_state cn
ON cn.owner_bch_name = cs.to_bch_name
AND cn.channel_root_block_number = cs.to_block_number
AND cn.channel_root_block_hash = cs.to_block_hash
WHERE cs.rel_type = 30
AND cn.owner_bch_name = NEW.owner_bch_name
AND cn.channel_root_block_number = NEW.channel_root_block_number
AND cn.channel_root_block_hash = NEW.channel_root_block_hash
AND cn.channel_type_code = 1;
INSERT INTO channel_stats_state (
owner_bch_name,
channel_root_block_number,
channel_root_block_hash,
owner_login,
channel_type_code,
subscribers_count,
updated_at_ms
) VALUES (
NEW.owner_bch_name,
NEW.channel_root_block_number,
NEW.channel_root_block_hash,
NEW.owner_login,
NEW.channel_type_code,
COALESCE(public_subscribers_count, 0),
now_ms
)
ON CONFLICT (owner_bch_name, channel_root_block_number, channel_root_block_hash) DO UPDATE SET
owner_login = EXCLUDED.owner_login,
channel_type_code = EXCLUDED.channel_type_code,
subscribers_count = EXCLUDED.subscribers_count,
updated_at_ms = EXCLUDED.updated_at_ms;
ELSE
WITH pending AS (
SELECT cs.login, COUNT(*)::INTEGER AS cnt
FROM connections_state cs
WHERE cs.rel_type = 30
AND cs.to_bch_name = NEW.owner_bch_name
AND cs.to_block_number = NEW.channel_root_block_number
AND cs.to_block_hash = NEW.channel_root_block_hash
GROUP BY cs.login
)
UPDATE user_stats_state us
SET following_channels_count = GREATEST(0, us.following_channels_count - pending.cnt),
updated_at_ms = now_ms
FROM pending
WHERE us.login = pending.login;
END IF;
RETURN NEW;
END;
$$;
CREATE OR REPLACE FUNCTION shine_blocks_line_integrity_bi()
RETURNS TRIGGER
LANGUAGE plpgsql
@@ -999,6 +1157,8 @@ AS $$
DECLARE
resolved_login TEXT;
positive_rel_type INTEGER;
existed_before BOOLEAN;
target_channel_type INTEGER;
BEGIN
IF NEW.msg_type <> 3 THEN
RETURN NEW;
@@ -1011,10 +1171,159 @@ BEGIN
RETURN NEW;
END IF;
SELECT EXISTS (
SELECT 1
FROM connections_state
WHERE login = NEW.login
AND rel_type = NEW.msg_sub_type
AND to_login = resolved_login
AND to_bch_name = NEW.to_bch_name
AND to_block_number = COALESCE(NEW.to_block_number, 0)
AND to_block_hash = COALESCE(NEW.to_block_hash, decode(repeat('00', 32), 'hex'))
)
INTO existed_before;
IF NOT existed_before THEN
IF NEW.msg_sub_type = 10 THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
0,
1,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
close_friends_count = user_stats_state.close_friends_count + 1,
updated_at_ms = EXCLUDED.updated_at_ms;
ELSIF NEW.msg_sub_type = 30 THEN
IF NEW.to_block_number IS NULL THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
1,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_channels_count = user_stats_state.following_channels_count + 1,
updated_at_ms = EXCLUDED.updated_at_ms;
ELSIF NEW.to_block_number = 0 THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
1,
0,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_users_count = user_stats_state.following_users_count + 1,
updated_at_ms = EXCLUDED.updated_at_ms;
ELSE
SELECT cn.channel_type_code
INTO target_channel_type
FROM channel_names_state cn
WHERE cn.owner_bch_name = NEW.to_bch_name
AND cn.channel_root_block_number = NEW.to_block_number
AND cn.channel_root_block_hash = NEW.to_block_hash
LIMIT 1;
IF target_channel_type = 1 THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
1,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_channels_count = user_stats_state.following_channels_count + 1,
updated_at_ms = EXCLUDED.updated_at_ms;
INSERT INTO channel_stats_state (
owner_bch_name,
channel_root_block_number,
channel_root_block_hash,
owner_login,
channel_type_code,
subscribers_count,
updated_at_ms
) VALUES (
NEW.to_bch_name,
NEW.to_block_number,
NEW.to_block_hash,
resolved_login,
target_channel_type,
1,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (owner_bch_name, channel_root_block_number, channel_root_block_hash) DO UPDATE SET
owner_login = EXCLUDED.owner_login,
channel_type_code = EXCLUDED.channel_type_code,
subscribers_count = channel_stats_state.subscribers_count + 1,
updated_at_ms = EXCLUDED.updated_at_ms;
ELSIF target_channel_type IS NULL THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
1,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_channels_count = user_stats_state.following_channels_count + 1,
updated_at_ms = EXCLUDED.updated_at_ms;
END IF;
END IF;
END IF;
END IF;
DELETE FROM connections_state
WHERE login = NEW.login
AND rel_type = NEW.msg_sub_type
AND to_login = resolved_login;
AND to_login = resolved_login
AND to_bch_name = NEW.to_bch_name
AND to_block_number = COALESCE(NEW.to_block_number, 0)
AND to_block_hash = COALESCE(NEW.to_block_hash, decode(repeat('00', 32), 'hex'));
INSERT INTO connections_state (
login, rel_type, to_login, to_bch_name, to_block_number, to_block_hash
@@ -1048,10 +1357,161 @@ BEGIN
RETURN NEW;
END IF;
SELECT EXISTS (
SELECT 1
FROM connections_state
WHERE login = NEW.login
AND rel_type = positive_rel_type
AND to_login = resolved_login
AND to_bch_name = NEW.to_bch_name
AND to_block_number = COALESCE(NEW.to_block_number, 0)
AND to_block_hash = COALESCE(NEW.to_block_hash, decode(repeat('00', 32), 'hex'))
)
INTO existed_before;
IF NOT existed_before THEN
RETURN NEW;
END IF;
IF positive_rel_type = 10 THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
0,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
close_friends_count = GREATEST(0, user_stats_state.close_friends_count - 1),
updated_at_ms = EXCLUDED.updated_at_ms;
ELSIF positive_rel_type = 30 THEN
IF NEW.to_block_number IS NULL THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
0,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_channels_count = GREATEST(0, user_stats_state.following_channels_count - 1),
updated_at_ms = EXCLUDED.updated_at_ms;
ELSIF NEW.to_block_number = 0 THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
0,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_users_count = GREATEST(0, user_stats_state.following_users_count - 1),
updated_at_ms = EXCLUDED.updated_at_ms;
ELSE
SELECT cn.channel_type_code
INTO target_channel_type
FROM channel_names_state cn
WHERE cn.owner_bch_name = NEW.to_bch_name
AND cn.channel_root_block_number = NEW.to_block_number
AND cn.channel_root_block_hash = NEW.to_block_hash
LIMIT 1;
IF target_channel_type = 1 THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
0,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_channels_count = GREATEST(0, user_stats_state.following_channels_count - 1),
updated_at_ms = EXCLUDED.updated_at_ms;
INSERT INTO channel_stats_state (
owner_bch_name,
channel_root_block_number,
channel_root_block_hash,
owner_login,
channel_type_code,
subscribers_count,
updated_at_ms
) VALUES (
NEW.to_bch_name,
NEW.to_block_number,
NEW.to_block_hash,
resolved_login,
target_channel_type,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (owner_bch_name, channel_root_block_number, channel_root_block_hash) DO UPDATE SET
owner_login = EXCLUDED.owner_login,
channel_type_code = EXCLUDED.channel_type_code,
subscribers_count = GREATEST(0, channel_stats_state.subscribers_count - 1),
updated_at_ms = EXCLUDED.updated_at_ms;
ELSIF target_channel_type IS NULL THEN
INSERT INTO user_stats_state (
login,
owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count,
updated_at_ms
) VALUES (
NEW.login,
0,
0,
0,
0,
CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)
)
ON CONFLICT (login) DO UPDATE SET
following_channels_count = GREATEST(0, user_stats_state.following_channels_count - 1),
updated_at_ms = EXCLUDED.updated_at_ms;
END IF;
END IF;
END IF;
DELETE FROM connections_state
WHERE login = NEW.login
AND rel_type = positive_rel_type
AND to_login = resolved_login;
AND to_login = resolved_login
AND to_bch_name = NEW.to_bch_name
AND to_block_number = COALESCE(NEW.to_block_number, 0)
AND to_block_hash = COALESCE(NEW.to_block_hash, decode(repeat('00', 32), 'hex'));
RETURN NEW;
END;
@@ -1231,4 +1691,10 @@ AFTER INSERT ON blocks
FOR EACH ROW
EXECUTE FUNCTION shine_blocks_edit_apply_ai();
DROP TRIGGER IF EXISTS trg_channel_names_state_stats_ai ON channel_names_state;
CREATE TRIGGER trg_channel_names_state_stats_ai
AFTER INSERT ON channel_names_state
FOR EACH ROW
EXECUTE FUNCTION shine_channel_names_state_stats_ai();
COMMIT;
@@ -4,6 +4,8 @@ import org.eclipse.jetty.websocket.api.Session;
import shine.db.entities.CurrentUserEntry;
import shine.db.entities.ActiveSessionEntry;
import java.util.List;
/**
* ConnectionContext контекст состояния одного WebSocket-соединения.
* Живёт ровно столько же, сколько живёт подключение.
@@ -77,6 +79,12 @@ public class ConnectionContext {
*/
private Session wsSession;
/** Временная server-to-server роль, заявленная через ServerHello. */
private boolean serverConnection;
private String remoteServerLogin;
private int remoteServerProtocolVersion;
private List<String> remoteServerCapabilities = List.of();
// --- WebSocket Session ---
public Session getWsSession() {
@@ -87,6 +95,20 @@ public class ConnectionContext {
this.wsSession = wsSession;
}
public boolean isServerConnection() { return serverConnection; }
public void setServerConnection(boolean serverConnection) { this.serverConnection = serverConnection; }
public String getRemoteServerLogin() { return remoteServerLogin; }
public void setRemoteServerLogin(String remoteServerLogin) { this.remoteServerLogin = remoteServerLogin; }
public int getRemoteServerProtocolVersion() { return remoteServerProtocolVersion; }
public void setRemoteServerProtocolVersion(int value) { this.remoteServerProtocolVersion = value; }
public List<String> getRemoteServerCapabilities() { return remoteServerCapabilities; }
public void setRemoteServerCapabilities(List<String> capabilities) {
this.remoteServerCapabilities = capabilities == null ? List.of() : List.copyOf(capabilities);
}
// --- SolanaUser / ActiveSession ---
public CurrentUserEntry getCurrentUser() {
@@ -188,6 +210,10 @@ public class ConnectionContext {
authenticationStatus = AUTH_STATUS_NONE;
wsSession = null;
serverConnection = false;
remoteServerLogin = null;
remoteServerProtocolVersion = 0;
remoteServerCapabilities = List.of();
}
@Override
@@ -133,6 +133,7 @@ import server.logic.ws_protocol.JSON.handlers.system.Net_ClientDebugLog_Handler;
import server.logic.ws_protocol.JSON.handlers.system.Net_ListBlockchainHeads_Handler;
import server.logic.ws_protocol.JSON.handlers.system.Net_CallDeliveryReport_Handler;
import server.logic.ws_protocol.JSON.handlers.system.Net_Ping_Handler;
import server.logic.ws_protocol.JSON.handlers.system.Net_ServerHello_Handler;
import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_CallDeliveryReport_Request;
import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_ClientErrorLog_Request;
import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_ClientDebugLog_Request;
@@ -141,6 +142,7 @@ import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_GetServerInfo_
import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_GetSyncUserProfile_Request;
import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_ListBlockchainHeads_Request;
import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_Ping_Request;
import server.logic.ws_protocol.JSON.handlers.system.entyties.Net_ServerHello_Request;
import java.util.Map;
@@ -232,6 +234,7 @@ public final class JsonHandlerRegistry {
Map.entry("SendSignal", new Net_SendSignal_Handler()),
// --- system ---
Map.entry("ServerHello", new Net_ServerHello_Handler()),
Map.entry("Ping", new Net_Ping_Handler()),
Map.entry("GetServerInfo", new Net_GetServerInfo_Handler()),
Map.entry("ListBlockchainHeads", new Net_ListBlockchainHeads_Handler()),
@@ -323,6 +326,7 @@ public final class JsonHandlerRegistry {
Map.entry("SendSignal", Net_SendSignal_Request.class),
// --- system ---
Map.entry("ServerHello", Net_ServerHello_Request.class),
Map.entry("Ping", Net_Ping_Request.class),
Map.entry("GetServerInfo", Net_GetServerInfo_Request.class),
Map.entry("ListBlockchainHeads", Net_ListBlockchainHeads_Request.class),
@@ -16,6 +16,8 @@ import utils.blockchain.BlockchainNameUtil;
import blockchain.body.CreateChannelBody;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.List;
@@ -65,6 +67,7 @@ public class Net_GetChannelMessages_Handler implements JsonMessageHandler {
channel.setMetaUpdatedAtMs(meta.metaUpdatedAtMs);
channel.setChannelTypeCode(meta.channelTypeCode);
channel.setChannelTypeVersion(meta.channelTypeVersion);
channel.setSubscribersCount(loadSubscribersCount(c, ownerBch, lineCode, meta.channelTypeCode));
Net_GetChannelMessages_Response.BlockRef rootRef = new Net_GetChannelMessages_Response.BlockRef();
rootRef.setBlockNumber(lineCode);
rootRef.setBlockHash(req.getChannel().getChannelRootBlockHash());
@@ -180,4 +183,28 @@ public class Net_GetChannelMessages_Handler implements JsonMessageHandler {
return NetExceptionResponseFactory.error(req, WireCodes.Status.INTERNAL_ERROR, "internal_error", "Внутренняя ошибка сервера");
}
}
private static int loadSubscribersCount(Connection c, String ownerBch, int rootNumber, int channelTypeCode) {
if (channelTypeCode != (CreateChannelBody.CHANNEL_TYPE_PUBLIC & 0xFFFF)) {
return 0;
}
String sql = """
SELECT subscribers_count
FROM channel_stats_state
WHERE owner_bch_name = ?
AND channel_root_block_number = ?
LIMIT 1
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setString(1, ownerBch);
ps.setInt(2, rootNumber);
try (ResultSet rs = ps.executeQuery()) {
if (!rs.next()) return 0;
return Math.max(0, rs.getInt("subscribers_count"));
}
} catch (Exception e) {
log.warn("GetChannelMessages: не удалось загрузить subscribers_count для {}#{}", ownerBch, rootNumber, e);
return 0;
}
}
}
@@ -32,6 +32,7 @@ public class Net_GetChannelMessages_Response extends Net_Response {
private Long metaUpdatedAtMs;
private Integer channelTypeCode;
private Integer channelTypeVersion;
private Integer subscribersCount;
private BlockRef channelRoot;
public String getOwnerLogin() { return ownerLogin; }
@@ -67,6 +68,9 @@ public class Net_GetChannelMessages_Response extends Net_Response {
public Integer getChannelTypeVersion() { return channelTypeVersion; }
public void setChannelTypeVersion(Integer channelTypeVersion) { this.channelTypeVersion = channelTypeVersion; }
public Integer getSubscribersCount() { return subscribersCount; }
public void setSubscribersCount(Integer subscribersCount) { this.subscribersCount = subscribersCount; }
public BlockRef getChannelRoot() { return channelRoot; }
public void setChannelRoot(BlockRef channelRoot) { this.channelRoot = channelRoot; }
}
@@ -11,11 +11,15 @@ import server.logic.ws_protocol.JSON.handlers.tempToTest.entyties.Net_GetUser_Re
import server.logic.ws_protocol.JSON.utils.NetExceptionResponseFactory;
import server.logic.ws_protocol.WireCodes;
import shine.db.dao.BlockchainStateDAO;
import shine.db.DbController;
import shine.db.dao.CurrentUsersDAO;
import shine.db.entities.BlockchainStateEntry;
import shine.db.entities.CurrentUserEntry;
import java.sql.SQLException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Arrays;
public class Net_GetUser_Handler implements JsonMessageHandler {
@@ -63,6 +67,7 @@ public class Net_GetUser_Handler implements JsonMessageHandler {
resp.setSolanaKey(u.getSolanaKey());
resp.setBlockchainKey(u.getBlockchainKey());
resp.setClientKey(u.getClientKey());
loadUserStats(resp, u.getLogin());
// Возвращаем актуальный курсор блокчейна и, если запись состояния потеряна,
// автоматически восстанавливаем её для существующего пользователя.
@@ -125,4 +130,35 @@ public class Net_GetUser_Handler implements JsonMessageHandler {
}
return new String(out);
}
private static void loadUserStats(Net_GetUser_Response resp, String login) {
resp.setOwnedPublicChannelsCount(0);
resp.setFollowingUsersCount(0);
resp.setFollowingChannelsCount(0);
resp.setCloseFriendsCount(0);
String sql = """
SELECT owned_public_channels_count,
following_users_count,
following_channels_count,
close_friends_count
FROM user_stats_state
WHERE login = ?
LIMIT 1
""";
try (Connection c = DbController.getInstance().getConnection();
PreparedStatement ps = c.prepareStatement(sql)) {
ps.setString(1, login);
try (ResultSet rs = ps.executeQuery()) {
if (!rs.next()) {
return;
}
resp.setOwnedPublicChannelsCount(rs.getInt("owned_public_channels_count"));
resp.setFollowingUsersCount(rs.getInt("following_users_count"));
resp.setFollowingChannelsCount(rs.getInt("following_channels_count"));
resp.setCloseFriendsCount(rs.getInt("close_friends_count"));
}
} catch (Exception e) {
log.warn("GetUser: не удалось загрузить статистику для login={}", login, e);
}
}
}
@@ -43,6 +43,10 @@ public class Net_GetUser_Response extends Net_Response {
private String serverLastGlobalHash;
private Long serverBlockchainSizeBytes;
private Long serverBlockchainSizeLimitBytes;
private Integer ownedPublicChannelsCount;
private Integer followingUsersCount;
private Integer followingChannelsCount;
private Integer closeFriendsCount;
public Boolean getExists() { return exists; }
public void setExists(Boolean exists) { this.exists = exists; }
@@ -74,4 +78,16 @@ public class Net_GetUser_Response extends Net_Response {
public Long getServerBlockchainSizeLimitBytes() { return serverBlockchainSizeLimitBytes; }
public void setServerBlockchainSizeLimitBytes(Long serverBlockchainSizeLimitBytes) { this.serverBlockchainSizeLimitBytes = serverBlockchainSizeLimitBytes; }
public Integer getOwnedPublicChannelsCount() { return ownedPublicChannelsCount; }
public void setOwnedPublicChannelsCount(Integer ownedPublicChannelsCount) { this.ownedPublicChannelsCount = ownedPublicChannelsCount; }
public Integer getFollowingUsersCount() { return followingUsersCount; }
public void setFollowingUsersCount(Integer followingUsersCount) { this.followingUsersCount = followingUsersCount; }
public Integer getFollowingChannelsCount() { return followingChannelsCount; }
public void setFollowingChannelsCount(Integer followingChannelsCount) { this.followingChannelsCount = followingChannelsCount; }
public Integer getCloseFriendsCount() { return closeFriendsCount; }
public void setCloseFriendsCount(Integer closeFriendsCount) { this.closeFriendsCount = closeFriendsCount; }
}
@@ -123,7 +123,7 @@ public class Net_UpsertUserSetting_Handler implements JsonMessageHandler {
if (remoteLogin.isBlank() || remoteUrl.isBlank()) continue;
if (!ownServerLogin.isBlank() && remoteLogin.equalsIgnoreCase(ownServerLogin)) continue;
try {
REMOTE.upsertUserSetting(remoteUrl, entry, true);
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));
@@ -11,27 +11,19 @@ import shine.db.entities.BlockEntry;
import shine.db.entities.SyncServerEntry;
import utils.blockchain.BlockchainNameUtil;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.WebSocket;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.List;
import java.util.Locale;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
/**
* Фоновая one-shot репликация AddBlock на серверы из локальной таблицы sync_servers.
* Фоновая репликация AddBlock через общий постоянный WSS-пул на серверы
* из локальной таблицы sync_servers.
*/
public final class AddBlockSyncService {
private static final Logger log = LoggerFactory.getLogger(AddBlockSyncService.class);
private static final ObjectMapper MAPPER = new ObjectMapper();
private static final HttpClient HTTP = HttpClient.newBuilder()
.connectTimeout(Duration.ofSeconds(6))
.build();
private static final ExecutorService EXECUTOR = new ThreadPoolExecutor(
1,
Math.max(2, Runtime.getRuntime().availableProcessors()),
@@ -100,14 +92,14 @@ public final class AddBlockSyncService {
}
private void replicateToPartner(SyncServerEntry partner, String blockchainName, int blockNumber, BlockEntry currentBlock) throws Exception {
String wsUrl = buildWsUrl(partner.getServerAddress());
String wsUrl = ServerConnectionPool.buildWsUrl(partner.getServerAddress());
if (wsUrl == null) {
log.warn("AddBlock sync skipped: invalid server_address for partner login={} address={}",
partner.getLogin(), partner.getServerAddress());
return;
}
AddBlockPushResult firstTry = pushBlock(wsUrl, blockchainName, currentBlock);
AddBlockPushResult firstTry = pushBlock(partner, blockchainName, currentBlock);
if (firstTry.ok()) {
log.info("AddBlock sync ok: partner={} blockchainName={} blockNumber={}",
partner.getLogin(), blockchainName, blockNumber);
@@ -142,7 +134,7 @@ public final class AddBlockSyncService {
}
for (BlockEntry blockEntry : missingBlocks) {
AddBlockPushResult backfillResult = pushBlock(wsUrl, blockchainName, blockEntry);
AddBlockPushResult backfillResult = pushBlock(partner, blockchainName, blockEntry);
if (!backfillResult.ok()) {
log.warn("AddBlock sync backfill failed: partner={} blockchainName={} blockNumber={} code={}",
partner.getLogin(), blockchainName, blockEntry.getBlockNumber(), backfillResult.code());
@@ -154,8 +146,8 @@ public final class AddBlockSyncService {
partner.getLogin(), blockchainName, fromBlockNumber, blockNumber);
}
private AddBlockPushResult pushBlock(String wsUrl, String blockchainName, BlockEntry blockEntry) throws Exception {
JsonNode response = sendAddBlock(wsUrl, blockchainName, blockEntry);
private AddBlockPushResult pushBlock(SyncServerEntry partner, String blockchainName, BlockEntry blockEntry) throws Exception {
JsonNode response = sendAddBlock(partner, blockchainName, blockEntry);
int status = response.path("status").asInt(500);
if (status >= 200 && status < 300) {
return AddBlockPushResult.success();
@@ -172,31 +164,16 @@ public final class AddBlockSyncService {
return new AddBlockPushResult(false, status, code, serverLastGlobalNumber, serverLastGlobalHash);
}
private JsonNode sendAddBlock(String wsUrl, String blockchainName, BlockEntry blockEntry) throws Exception {
CompletableFuture<String> responseFuture = new CompletableFuture<>();
CountDownLatch openLatch = new CountDownLatch(1);
SyncWsListener listener = new SyncWsListener(responseFuture, openLatch);
WebSocket webSocket = HTTP.newWebSocketBuilder()
.connectTimeout(Duration.ofSeconds(6))
.buildAsync(URI.create(wsUrl), listener)
.get(8, TimeUnit.SECONDS);
if (!openLatch.await(8, TimeUnit.SECONDS)) {
tryAbort(webSocket);
throw new TimeoutException("WS open timeout");
}
String requestId = "sync-" + UUID.randomUUID();
String json = buildAddBlockJson(requestId, blockchainName, blockEntry);
webSocket.sendText(json, true).get(8, TimeUnit.SECONDS);
String responseJson = responseFuture.get(12, TimeUnit.SECONDS);
tryAbort(webSocket);
return MAPPER.readTree(responseJson);
private JsonNode sendAddBlock(SyncServerEntry partner, String blockchainName, BlockEntry blockEntry) throws Exception {
String jsonTemplate = buildAddBlockJsonTemplate(blockchainName, blockEntry);
return ServerConnectionPool.getInstance().request(
partner.getLogin(),
partner.getServerAddress(),
jsonTemplate,
ServerConnectionPool.Priority.BULK);
}
private String buildAddBlockJson(String requestId, String blockchainName, BlockEntry blockEntry) throws Exception {
private String buildAddBlockJsonTemplate(String blockchainName, BlockEntry blockEntry) throws Exception {
String prevHashHex = blockEntry.getBlockNumber() <= 0
? ""
: toHex(extractPrevHash32(blockEntry.getBlockBytes()));
@@ -205,8 +182,6 @@ public final class AddBlockSyncService {
String safeBlockchainName = MAPPER.writeValueAsString(blockchainName);
String safePrevHashHex = MAPPER.writeValueAsString(prevHashHex);
String safeBlockBytes = MAPPER.writeValueAsString(blockBytesB64);
String safeRequestId = MAPPER.writeValueAsString(requestId);
return """
{
"op":"AddBlock",
@@ -218,7 +193,7 @@ public final class AddBlockSyncService {
"blockBytesB64":%s
}
}
""".formatted(safeRequestId, safeBlockchainName, blockEntry.getBlockNumber(), safePrevHashHex, safeBlockBytes);
""".formatted("%s", safeBlockchainName, blockEntry.getBlockNumber(), safePrevHashHex, safeBlockBytes);
}
private static byte[] extractPrevHash32(byte[] blockBytes) {
@@ -240,32 +215,6 @@ public final class AddBlockSyncService {
return s.isEmpty() ? null : s;
}
private static String buildWsUrl(String serverAddressRaw) {
String host = normalizeHostLike(serverAddressRaw);
if (host == null) return null;
return "wss://" + host + "/ws";
}
private static String normalizeHostLike(String value) {
if (value == null) return null;
String raw = value.trim();
if (raw.isEmpty()) return null;
try {
String withScheme = raw.matches("^[a-zA-Z]+://.*$") ? raw : "https://" + raw;
URI uri = URI.create(withScheme);
String host = uri.getHost();
if (host == null || host.isBlank()) return null;
return host.trim().toLowerCase(Locale.ROOT);
} catch (Exception e) {
String cleaned = raw
.replaceFirst("^[a-zA-Z]+://", "")
.replaceFirst("/.*$", "")
.trim()
.toLowerCase(Locale.ROOT);
return cleaned.isEmpty() ? null : cleaned;
}
}
private static String toHex(byte[] bytes) {
if (bytes == null) return "";
StringBuilder sb = new StringBuilder(bytes.length * 2);
@@ -276,17 +225,6 @@ public final class AddBlockSyncService {
return sb.toString();
}
private static void tryAbort(WebSocket webSocket) {
try {
webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "ok");
} catch (Exception ignored) {
}
try {
webSocket.abort();
} catch (Exception ignored) {
}
}
private record AddBlockPushResult(
boolean ok,
int status,
@@ -309,52 +247,4 @@ public final class AddBlockSyncService {
}
}
private static final class SyncWsListener implements WebSocket.Listener {
private final CompletableFuture<String> responseFuture;
private final CountDownLatch openLatch;
private final StringBuilder textBuffer = new StringBuilder();
private SyncWsListener(CompletableFuture<String> responseFuture, CountDownLatch openLatch) {
this.responseFuture = responseFuture;
this.openLatch = openLatch;
}
@Override
public void onOpen(WebSocket webSocket) {
openLatch.countDown();
webSocket.request(1);
}
@Override
public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
textBuffer.append(data);
if (last && !responseFuture.isDone()) {
responseFuture.complete(textBuffer.toString());
}
webSocket.request(1);
return CompletableFuture.completedFuture(null);
}
@Override
public CompletionStage<?> onBinary(WebSocket webSocket, ByteBuffer data, boolean last) {
webSocket.request(1);
return CompletableFuture.completedFuture(null);
}
@Override
public CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason) {
if (!responseFuture.isDone()) {
responseFuture.completeExceptionally(new IllegalStateException("WS closed before response: " + statusCode + " " + reason));
}
return CompletableFuture.completedFuture(null);
}
@Override
public void onError(WebSocket webSocket, Throwable error) {
if (!responseFuture.isDone()) {
responseFuture.completeExceptionally(error);
}
openLatch.countDown();
}
}
}
@@ -149,7 +149,8 @@ public final class DmDeliveryCoordinator {
completion.submit(() -> {
try {
if (!routeLogin.equals(ownServerLogin)) {
REMOTE.receiveIncomingMessage(route.getServerUrl(), incomingBlobB64, ownServerLogin);
REMOTE.receiveIncomingMessage(
routeLogin, route.getServerUrl(), incomingBlobB64, ownServerLogin);
}
return new RouteAttempt(routeLogin, null);
} catch (Exception e) {
@@ -188,6 +189,7 @@ public final class DmDeliveryCoordinator {
if (incoming == null || outgoing == null) return current;
try {
REMOTE.sendMessagePair(
peer.getServerLogin(),
peer.getServerUrl(),
Base64.getEncoder().encodeToString(incoming.getRawBlock()),
Base64.getEncoder().encodeToString(outgoing.getRawBlock()),
@@ -204,7 +206,7 @@ public final class DmDeliveryCoordinator {
if (peer == null) return current;
try {
RemoteDmSyncClient.RemoteDeliveryStatus remote = REMOTE.getDmDeliveryStatus(
peer.getServerUrl(), current.getOutgoingMessageKey());
peer.getServerLogin(), peer.getServerUrl(), current.getOutgoingMessageKey());
if (remote.known() && remote.delivered()) {
return DELIVERY_DAO.markDeliveredFromPeer(current.getEventId(), System.currentTimeMillis());
}
@@ -27,12 +27,15 @@ public final class DmFederationService {
String ownServerLogin = ownServerLogin();
for (UserAccessServerRouteEntry route : senderRoutes.values()) {
if (isOwnServer(route, ownServerLogin)) continue;
REMOTE.sendMessagePair(route.getServerUrl(), incomingBlobB64, outgoingBlobB64, ownServerLogin);
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.getServerUrl(), incomingBlobB64, ownServerLogin);
REMOTE.receiveIncomingMessage(
route.getServerLogin(), route.getServerUrl(), incomingBlobB64, ownServerLogin);
}
} catch (Exception e) {
log.warn("DM federation pair fan-out failed: from={} to={}", fromLogin, toLogin, e);
@@ -54,7 +57,8 @@ public final class DmFederationService {
if (routeLogin == null) continue;
if (ownServerLogin != null && ownServerLogin.equalsIgnoreCase(routeLogin)) continue;
if (normalizedSource != null && normalizedSource.equalsIgnoreCase(routeLogin)) continue;
REMOTE.receiveIncomingMessage(route.getServerUrl(), incomingBlobB64, ownServerLogin);
REMOTE.receiveIncomingMessage(
route.getServerLogin(), route.getServerUrl(), incomingBlobB64, ownServerLogin);
}
} catch (Exception e) {
log.warn("DM federation incoming relay failed: to={}", toLogin, e);
@@ -79,9 +83,9 @@ public final class DmFederationService {
for (UserAccessServerRouteEntry route : routes.values()) {
if (isOwnServer(route, ownServerLogin)) continue;
if (oneMessageDelete) {
REMOTE.deleteMessage(route.getServerUrl(), blobB64);
REMOTE.deleteMessage(route.getServerLogin(), route.getServerUrl(), blobB64);
} else {
REMOTE.deleteConversation(route.getServerUrl(), blobB64);
REMOTE.deleteConversation(route.getServerLogin(), route.getServerUrl(), blobB64);
}
}
} catch (Exception e) {
@@ -2,37 +2,22 @@ package server.sync;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.WebSocket;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/**
* Минимальный клиент для межсерверных JSON-op запросов по WSS.
*/
public final class RemoteBlockchainSyncClient {
private static final Logger log = LoggerFactory.getLogger(RemoteBlockchainSyncClient.class);
private static final ObjectMapper MAPPER = new ObjectMapper();
private static final HttpClient HTTP = HttpClient.newBuilder()
.connectTimeout(Duration.ofSeconds(6))
.build();
public List<RemoteBlockchainHead> listBlockchainHeads(String serverAddressRaw) throws Exception {
JsonNode response = send(serverAddressRaw, """
return listBlockchainHeads(null, serverAddressRaw);
}
public List<RemoteBlockchainHead> listBlockchainHeads(String serverLogin, String serverAddressRaw) throws Exception {
JsonNode response = send(serverLogin, serverAddressRaw, """
{
"op":"ListBlockchainHeads",
"requestId":%s,
@@ -62,8 +47,12 @@ public final class RemoteBlockchainSyncClient {
}
public RemoteSyncUserProfile getSyncUserProfile(String serverAddressRaw, String login) throws Exception {
return getSyncUserProfile(null, serverAddressRaw, login);
}
public RemoteSyncUserProfile getSyncUserProfile(String serverLogin, String serverAddressRaw, String login) throws Exception {
String safeLogin = MAPPER.writeValueAsString(login);
JsonNode response = send(serverAddressRaw, """
JsonNode response = send(serverLogin, serverAddressRaw, """
{
"op":"GetSyncUserProfile",
"requestId":%s,
@@ -96,9 +85,17 @@ public final class RemoteBlockchainSyncClient {
);
}
public RemoteBlockchainBlock getBlockchainBlock(String serverAddressRaw, String blockchainName, int blockNumber) throws Exception {
public RemoteBlockchainBlock getBlockchainBlock(
String serverAddressRaw, String blockchainName, int blockNumber
) throws Exception {
return getBlockchainBlock(null, serverAddressRaw, blockchainName, blockNumber);
}
public RemoteBlockchainBlock getBlockchainBlock(
String serverLogin, String serverAddressRaw, String blockchainName, int blockNumber
) throws Exception {
String safeBlockchainName = MAPPER.writeValueAsString(blockchainName);
JsonNode response = send(serverAddressRaw, """
JsonNode response = send(serverLogin, serverAddressRaw, """
{
"op":"GetBlockchainBlock",
"requestId":%s,
@@ -126,32 +123,12 @@ public final class RemoteBlockchainSyncClient {
);
}
private JsonNode send(String serverAddressRaw, String jsonTemplate) throws Exception {
String requestId = MAPPER.writeValueAsString("sync-" + UUID.randomUUID());
String json = jsonTemplate.formatted(requestId);
String wsUrl = buildWsUrl(serverAddressRaw);
if (wsUrl == null) {
throw new IllegalArgumentException("Invalid server address: " + serverAddressRaw);
}
CompletableFuture<String> responseFuture = new CompletableFuture<>();
CountDownLatch openLatch = new CountDownLatch(1);
SyncWsListener listener = new SyncWsListener(responseFuture, openLatch);
WebSocket webSocket = HTTP.newWebSocketBuilder()
.connectTimeout(Duration.ofSeconds(6))
.buildAsync(URI.create(wsUrl), listener)
.get(8, TimeUnit.SECONDS);
if (!openLatch.await(8, TimeUnit.SECONDS)) {
tryAbort(webSocket);
throw new TimeoutException("WS open timeout");
}
webSocket.sendText(json, true).get(8, TimeUnit.SECONDS);
String responseJson = responseFuture.get(12, TimeUnit.SECONDS);
tryAbort(webSocket);
return MAPPER.readTree(responseJson);
private JsonNode send(String serverLogin, String serverAddressRaw, String jsonTemplate) throws Exception {
return ServerConnectionPool.getInstance().request(
serverLogin,
serverAddressRaw,
jsonTemplate,
ServerConnectionPool.Priority.BULK);
}
private static String errorCode(JsonNode response) {
@@ -161,40 +138,7 @@ public final class RemoteBlockchainSyncClient {
}
static String buildWsUrl(String serverAddressRaw) {
String host = normalizeHostLike(serverAddressRaw);
if (host == null) return null;
return "wss://" + host + "/ws";
}
private static String normalizeHostLike(String value) {
if (value == null) return null;
String raw = value.trim();
if (raw.isEmpty()) return null;
try {
String withScheme = raw.matches("^[a-zA-Z]+://.*$") ? raw : "https://" + raw;
URI uri = URI.create(withScheme);
String host = uri.getHost();
if (host == null || host.isBlank()) return null;
return host.trim().toLowerCase(Locale.ROOT);
} catch (Exception e) {
String cleaned = raw
.replaceFirst("^[a-zA-Z]+://", "")
.replaceFirst("/.*$", "")
.trim()
.toLowerCase(Locale.ROOT);
return cleaned.isEmpty() ? null : cleaned;
}
}
private static void tryAbort(WebSocket webSocket) {
try {
webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "ok");
} catch (Exception ignored) {
}
try {
webSocket.abort();
} catch (Exception ignored) {
}
return ServerConnectionPool.buildWsUrl(serverAddressRaw);
}
public record RemoteBlockchainHead(
@@ -220,53 +164,4 @@ public final class RemoteBlockchainSyncClient {
long blockchainSizeLimitBytes
) {}
private static final class SyncWsListener implements WebSocket.Listener {
private final CompletableFuture<String> responseFuture;
private final CountDownLatch openLatch;
private final StringBuilder textBuffer = new StringBuilder();
private SyncWsListener(CompletableFuture<String> responseFuture, CountDownLatch openLatch) {
this.responseFuture = responseFuture;
this.openLatch = openLatch;
}
@Override
public void onOpen(WebSocket webSocket) {
openLatch.countDown();
webSocket.request(1);
}
@Override
public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
textBuffer.append(data);
if (last && !responseFuture.isDone()) {
responseFuture.complete(textBuffer.toString());
}
webSocket.request(1);
return CompletableFuture.completedFuture(null);
}
@Override
public CompletionStage<?> onBinary(WebSocket webSocket, ByteBuffer data, boolean last) {
webSocket.request(1);
return CompletableFuture.completedFuture(null);
}
@Override
public CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason) {
if (!responseFuture.isDone()) {
responseFuture.completeExceptionally(new IllegalStateException("WS closed before response: " + statusCode + " " + reason));
}
return CompletableFuture.completedFuture(null);
}
@Override
public void onError(WebSocket webSocket, Throwable error) {
log.warn("Remote sync websocket error: {}", String.valueOf(error));
if (!responseFuture.isDone()) {
responseFuture.completeExceptionally(error);
}
openLatch.countDown();
}
}
}
@@ -3,37 +3,33 @@ package server.sync;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.WebSocket;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/** Клиент стабильных межсерверных DM-операций. */
public final class RemoteDmSyncClient {
private static final ObjectMapper MAPPER = new ObjectMapper();
private static final HttpClient HTTP = HttpClient.newBuilder()
.connectTimeout(Duration.ofSeconds(6))
.build();
public void sendMessagePair(
String serverAddressRaw,
String incomingBlobB64,
String outgoingBlobB64,
String sourceServerLogin
) throws Exception {
sendMessagePair(null, serverAddressRaw, incomingBlobB64, outgoingBlobB64, sourceServerLogin);
}
public void sendMessagePair(
String targetServerLogin,
String serverAddressRaw,
String incomingBlobB64,
String outgoingBlobB64,
String sourceServerLogin
) throws Exception {
String incomingJson = MAPPER.writeValueAsString(incomingBlobB64);
String outgoingJson = MAPPER.writeValueAsString(outgoingBlobB64);
String sourceServerLoginJson = toOptionalJsonField("sourceServerLogin", sourceServerLogin);
JsonNode response = send(serverAddressRaw, """
JsonNode response = send(targetServerLogin, serverAddressRaw, """
{
"op":"ReceiveOutcomingMessage",
"requestId":%s,
@@ -50,10 +46,19 @@ public final class RemoteDmSyncClient {
String serverAddressRaw,
String incomingBlobB64,
String sourceServerLogin
) throws Exception {
receiveIncomingMessage(null, serverAddressRaw, incomingBlobB64, sourceServerLogin);
}
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(serverAddressRaw, """
JsonNode response = send(targetServerLogin, serverAddressRaw, """
{
"op":"ReceiveIncomingMessage",
"requestId":%s,
@@ -66,8 +71,14 @@ public final class RemoteDmSyncClient {
}
public RemoteDeliveryStatus getDmDeliveryStatus(String serverAddressRaw, String messageKey) throws Exception {
return getDmDeliveryStatus(null, serverAddressRaw, messageKey);
}
public RemoteDeliveryStatus getDmDeliveryStatus(
String targetServerLogin, String serverAddressRaw, String messageKey
) throws Exception {
String messageKeyJson = MAPPER.writeValueAsString(messageKey);
JsonNode response = send(serverAddressRaw, """
JsonNode response = send(targetServerLogin, serverAddressRaw, """
{
"op":"GetDmDeliveryStatus",
"requestId":%s,
@@ -94,7 +105,21 @@ public final class RemoteDmSyncClient {
int maxBytes,
List<String> ackSyncIds
) throws Exception {
try (RemoteSyncSession session = new RemoteSyncSession(serverAddressRaw)) {
return dmSyncBatch(null, serverAddressRaw, ownerLogin, afterStoredAtMs,
afterMessageKey, limit, maxBytes, ackSyncIds);
}
public RemoteDmBatch dmSyncBatch(
String targetServerLogin,
String serverAddressRaw,
String ownerLogin,
long afterStoredAtMs,
String afterMessageKey,
int limit,
int maxBytes,
List<String> ackSyncIds
) throws Exception {
try (RemoteSyncSession session = new RemoteSyncSession(targetServerLogin, serverAddressRaw)) {
return dmSyncBatch(session, ownerLogin, afterStoredAtMs, afterMessageKey,
limit, maxBytes, ackSyncIds);
}
@@ -156,8 +181,12 @@ public final class RemoteDmSyncClient {
}
public void deleteMessage(String serverAddressRaw, String blobB64) throws Exception {
deleteMessage(null, serverAddressRaw, blobB64);
}
public void deleteMessage(String targetServerLogin, String serverAddressRaw, String blobB64) throws Exception {
String blobJson = MAPPER.writeValueAsString(blobB64);
JsonNode response = send(serverAddressRaw, """
JsonNode response = send(targetServerLogin, serverAddressRaw, """
{
"op":"DeleteMessage",
"requestId":%s,
@@ -168,8 +197,12 @@ public final class RemoteDmSyncClient {
}
public void deleteConversation(String serverAddressRaw, String blobB64) throws Exception {
deleteConversation(null, serverAddressRaw, blobB64);
}
public void deleteConversation(String targetServerLogin, String serverAddressRaw, String blobB64) throws Exception {
String blobJson = MAPPER.writeValueAsString(blobB64);
JsonNode response = send(serverAddressRaw, """
JsonNode response = send(targetServerLogin, serverAddressRaw, """
{
"op":"DeleteConversation",
"requestId":%s,
@@ -179,27 +212,12 @@ public final class RemoteDmSyncClient {
ensureOk("DeleteConversation", response);
}
private JsonNode send(String serverAddressRaw, String jsonTemplate) throws Exception {
String requestId = MAPPER.writeValueAsString("dm-sync-" + UUID.randomUUID());
String json = jsonTemplate.formatted(requestId);
String wsUrl = RemoteBlockchainSyncClient.buildWsUrl(serverAddressRaw);
if (wsUrl == null) throw new IllegalArgumentException("Invalid server address: " + serverAddressRaw);
CompletableFuture<String> responseFuture = new CompletableFuture<>();
CountDownLatch openLatch = new CountDownLatch(1);
SyncWsListener listener = new SyncWsListener(responseFuture, openLatch);
WebSocket webSocket = HTTP.newWebSocketBuilder()
.connectTimeout(Duration.ofSeconds(6))
.buildAsync(URI.create(wsUrl), listener)
.get(8, TimeUnit.SECONDS);
if (!openLatch.await(8, TimeUnit.SECONDS)) {
tryAbort(webSocket);
throw new TimeoutException("WS open timeout");
}
webSocket.sendText(json, true).get(8, TimeUnit.SECONDS);
String responseJson = responseFuture.get(12, TimeUnit.SECONDS);
tryAbort(webSocket);
return MAPPER.readTree(responseJson);
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 {
@@ -215,60 +233,10 @@ public final class RemoteDmSyncClient {
throw new IllegalStateException(op + " failed: status=" + status + " code=" + code);
}
private static void tryAbort(WebSocket webSocket) {
try { webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "ok"); } catch (Exception ignored) {}
try { webSocket.abort(); } catch (Exception ignored) {}
}
public record RemoteDeliveryStatus(String messageKey, boolean known, boolean delivered) {}
public record RemoteDmBatch(long nextStoredAtMs, String nextMessageKey, boolean hasMore, List<RemoteDmItem> items) {}
public record RemoteDmItem(
String syncId, String primaryMessageKey, long storedAtMs, List<String> blobsB64
) {}
private static final class SyncWsListener implements WebSocket.Listener {
private final CompletableFuture<String> responseFuture;
private final CountDownLatch openLatch;
private final StringBuilder textBuffer = new StringBuilder();
private SyncWsListener(CompletableFuture<String> responseFuture, CountDownLatch openLatch) {
this.responseFuture = responseFuture;
this.openLatch = openLatch;
}
@Override
public void onOpen(WebSocket webSocket) {
openLatch.countDown();
webSocket.request(1);
}
@Override
public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
textBuffer.append(data);
if (last && !responseFuture.isDone()) responseFuture.complete(textBuffer.toString());
webSocket.request(1);
return CompletableFuture.completedFuture(null);
}
@Override
public CompletionStage<?> onBinary(WebSocket webSocket, ByteBuffer data, boolean last) {
webSocket.request(1);
return CompletableFuture.completedFuture(null);
}
@Override
public CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason) {
if (!responseFuture.isDone()) {
responseFuture.completeExceptionally(
new IllegalStateException("WS closed before response: " + statusCode + " " + reason));
}
return CompletableFuture.completedFuture(null);
}
@Override
public void onError(WebSocket webSocket, Throwable error) {
if (!responseFuture.isDone()) responseFuture.completeExceptionally(error);
openLatch.countDown();
}
}
}
@@ -1,70 +1,35 @@
package server.sync;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.WebSocket;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/** Один последовательный WS-сеанс для синхронизации всех данных access-сервера. */
/**
* Логический последовательный сеанс поверх общего постоянного WSS-пула.
* close() больше не закрывает физическое соединение с сервером.
*/
public final class RemoteSyncSession implements AutoCloseable {
private static final ObjectMapper MAPPER = new ObjectMapper();
private static final HttpClient HTTP = HttpClient.newBuilder()
.connectTimeout(Duration.ofSeconds(6)).build();
private final LinkedBlockingQueue<String> responses = new LinkedBlockingQueue<>();
private final WebSocket webSocket;
private final String serverLogin;
private final String serverAddress;
/** Совместимый конструктор: при отсутствии логина пул использует адрес как ключ peer. */
public RemoteSyncSession(String serverAddressRaw) throws Exception {
String wsUrl = RemoteBlockchainSyncClient.buildWsUrl(serverAddressRaw);
if (wsUrl == null) throw new IllegalArgumentException("Invalid server address: " + serverAddressRaw);
Listener listener = new Listener(responses);
webSocket = HTTP.newWebSocketBuilder().connectTimeout(Duration.ofSeconds(6))
.buildAsync(URI.create(wsUrl), listener).get(8, TimeUnit.SECONDS);
this(null, serverAddressRaw);
}
public RemoteSyncSession(String serverLogin, String serverAddressRaw) {
this.serverLogin = serverLogin;
this.serverAddress = serverAddressRaw;
}
public synchronized JsonNode send(String jsonTemplate) throws Exception {
String requestId = MAPPER.writeValueAsString("access-sync-" + UUID.randomUUID());
webSocket.sendText(jsonTemplate.formatted(requestId), true).get(8, TimeUnit.SECONDS);
String json = responses.poll(12, TimeUnit.SECONDS);
if (json == null) throw new TimeoutException("WS response timeout");
return MAPPER.readTree(json);
return ServerConnectionPool.getInstance().request(
serverLogin,
serverAddress,
jsonTemplate,
ServerConnectionPool.Priority.NORMAL);
}
@Override
public void close() {
try { webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "ok"); } catch (Exception ignored) {}
}
private static final class Listener implements WebSocket.Listener {
private final LinkedBlockingQueue<String> responses;
private final StringBuilder text = new StringBuilder();
private Listener(LinkedBlockingQueue<String> responses) { this.responses = responses; }
@Override public void onOpen(WebSocket ws) { ws.request(1); }
@Override public CompletionStage<?> onText(WebSocket ws, CharSequence data, boolean last) {
text.append(data);
if (last) {
responses.offer(text.toString());
text.setLength(0);
}
ws.request(1);
return CompletableFuture.completedFuture(null);
}
@Override public CompletionStage<?> onBinary(WebSocket ws, ByteBuffer data, boolean last) {
ws.request(1);
return CompletableFuture.completedFuture(null);
}
@Override public void onError(WebSocket ws, Throwable error) {
responses.offer("{\"status\":500,\"code\":\"WS_ERROR\"}");
}
// Физическое соединение принадлежит ServerConnectionPool.
}
}
@@ -4,27 +4,22 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import shine.db.entities.UserSettingEntry;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.WebSocket;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
public final class RemoteUserSettingsSyncClient {
private static final ObjectMapper MAPPER = new ObjectMapper();
private static final HttpClient HTTP = HttpClient.newBuilder()
.connectTimeout(Duration.ofSeconds(6))
.build();
public void upsertUserSetting(String serverAddressRaw, UserSettingEntry entry, boolean syncDelivery) throws Exception {
try (RemoteSyncSession session = new RemoteSyncSession(serverAddressRaw)) {
public void upsertUserSetting(
String serverAddressRaw, UserSettingEntry entry, boolean syncDelivery
) throws Exception {
upsertUserSetting(null, serverAddressRaw, entry, syncDelivery);
}
public void upsertUserSetting(
String targetServerLogin, String serverAddressRaw, UserSettingEntry entry, boolean syncDelivery
) throws Exception {
try (RemoteSyncSession session = new RemoteSyncSession(targetServerLogin, serverAddressRaw)) {
upsertUserSetting(session, entry, syncDelivery);
}
}
@@ -69,7 +64,20 @@ public final class RemoteUserSettingsSyncClient {
int limit,
int maxBytes
) throws Exception {
try (RemoteSyncSession session = new RemoteSyncSession(serverAddressRaw)) {
return userSettingsSyncBatch(null, serverAddressRaw, ownerLogin,
afterTimeMs, afterSettingKey, limit, maxBytes);
}
public RemoteUserSettingsBatch userSettingsSyncBatch(
String targetServerLogin,
String serverAddressRaw,
String ownerLogin,
long afterTimeMs,
String afterSettingKey,
int limit,
int maxBytes
) throws Exception {
try (RemoteSyncSession session = new RemoteSyncSession(targetServerLogin, serverAddressRaw)) {
return userSettingsSyncBatch(session, ownerLogin, afterTimeMs, afterSettingKey, limit, maxBytes);
}
}
@@ -129,34 +137,6 @@ public final class RemoteUserSettingsSyncClient {
);
}
private JsonNode send(String serverAddressRaw, String jsonTemplate) throws Exception {
String requestId = MAPPER.writeValueAsString("user-settings-sync-" + UUID.randomUUID());
String json = jsonTemplate.formatted(requestId);
String wsUrl = RemoteBlockchainSyncClient.buildWsUrl(serverAddressRaw);
if (wsUrl == null) {
throw new IllegalArgumentException("Invalid server address: " + serverAddressRaw);
}
CompletableFuture<String> responseFuture = new CompletableFuture<>();
CountDownLatch openLatch = new CountDownLatch(1);
SyncWsListener listener = new SyncWsListener(responseFuture, openLatch);
WebSocket webSocket = HTTP.newWebSocketBuilder()
.connectTimeout(Duration.ofSeconds(6))
.buildAsync(URI.create(wsUrl), listener)
.get(8, TimeUnit.SECONDS);
if (!openLatch.await(8, TimeUnit.SECONDS)) {
tryAbort(webSocket);
throw new TimeoutException("WS open timeout");
}
webSocket.sendText(json, true).get(8, TimeUnit.SECONDS);
String responseJson = responseFuture.get(12, TimeUnit.SECONDS);
tryAbort(webSocket);
return MAPPER.readTree(responseJson);
}
private void ensureOk(String op, JsonNode response) {
int status = response.path("status").asInt(500);
if (status >= 200 && status < 300) return;
@@ -165,17 +145,6 @@ public final class RemoteUserSettingsSyncClient {
throw new IllegalStateException(op + " failed: status=" + status + " code=" + code);
}
private static void tryAbort(WebSocket webSocket) {
try {
webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "ok");
} catch (Exception ignored) {
}
try {
webSocket.abort();
} catch (Exception ignored) {
}
}
public record RemoteUserSettingsBatch(
long nextTimeMs,
String nextSettingKey,
@@ -194,35 +163,4 @@ public final class RemoteUserSettingsSyncClient {
String signature
) {}
private static final class SyncWsListener implements WebSocket.Listener {
private final CompletableFuture<String> responseFuture;
private final CountDownLatch openLatch;
private final StringBuilder textBuffer = new StringBuilder();
private SyncWsListener(CompletableFuture<String> responseFuture, CountDownLatch openLatch) {
this.responseFuture = responseFuture;
this.openLatch = openLatch;
}
@Override
public void onOpen(WebSocket webSocket) {
openLatch.countDown();
webSocket.request(1);
}
@Override
public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
textBuffer.append(data);
if (last && !responseFuture.isDone()) {
responseFuture.complete(textBuffer.toString());
}
webSocket.request(1);
return CompletableFuture.completedFuture(null);
}
@Override
public void onError(WebSocket webSocket, Throwable error) {
responseFuture.completeExceptionally(error);
}
}
}
@@ -67,7 +67,8 @@ public final class BlockchainResyncRecoveryOnStartup {
blockchainName, partnerLogin, partnerAddress);
try {
List<RemoteBlockchainSyncClient.RemoteBlockchainHead> heads = REMOTE.listBlockchainHeads(partnerAddress);
List<RemoteBlockchainSyncClient.RemoteBlockchainHead> heads =
REMOTE.listBlockchainHeads(partnerLogin, partnerAddress);
RemoteBlockchainSyncClient.RemoteBlockchainHead remoteHead = heads.stream()
.filter(h -> h != null && blockchainName.equals(h.blockchainName()))
.findFirst()
@@ -102,7 +102,7 @@ public final class PeriodicBlockchainSyncService {
if (partnerLogin == null) return;
List<RemoteBlockchainSyncClient.RemoteBlockchainHead> remoteHeads =
REMOTE.listBlockchainHeads(partner.getServerAddress());
REMOTE.listBlockchainHeads(partner.getLogin(), partner.getServerAddress());
for (RemoteBlockchainSyncClient.RemoteBlockchainHead remoteHead : remoteHeads) {
if (remoteHead == null || remoteHead.blockchainName() == null || remoteHead.blockchainName().isBlank()) {
@@ -170,7 +170,8 @@ public final class PeriodicBlockchainSyncService {
int fromBlockNumber = Math.max(localLast + 1, 0);
for (int blockNumber = fromBlockNumber; blockNumber <= remoteHead.lastBlockNumber(); blockNumber++) {
RemoteBlockchainSyncClient.RemoteBlockchainBlock remoteBlock =
REMOTE.getBlockchainBlock(partner.getServerAddress(), remoteHead.blockchainName(), blockNumber);
REMOTE.getBlockchainBlock(
partner.getLogin(), partner.getServerAddress(), remoteHead.blockchainName(), blockNumber);
if (remoteBlock == null) {
log.warn("Periodic blockchain sync: remote block not found. partner={} blockchainName={} blockNumber={}",
partnerLogin, remoteHead.blockchainName(), blockNumber);
@@ -284,7 +285,8 @@ public final class PeriodicBlockchainSyncService {
String localPrevHash = "";
for (int blockNumber = 0; blockNumber <= remoteHead.lastBlockNumber(); blockNumber++) {
RemoteBlockchainSyncClient.RemoteBlockchainBlock remoteBlock =
REMOTE.getBlockchainBlock(partner.getServerAddress(), blockchainName, blockNumber);
REMOTE.getBlockchainBlock(
partner.getLogin(), partner.getServerAddress(), blockchainName, blockNumber);
if (remoteBlock == null) {
log.warn("Blockchain resync: remote block not found. partner={} blockchainName={} blockNumber={}",
partnerLogin, blockchainName, blockNumber);
@@ -137,7 +137,7 @@ public final class PeriodicUserSettingsSyncService {
boolean bootstrapCompleted = false;
int appliedDm;
try (RemoteSyncSession session = new RemoteSyncSession(route.getServerUrl())) {
try (RemoteSyncSession session = new RemoteSyncSession(route.getServerLogin(), route.getServerUrl())) {
for (int page = 0; page < maxPages; page++) {
RemoteUserSettingsSyncClient.RemoteUserSettingsBatch batch = REMOTE.userSettingsSyncBatch(
session,
@@ -83,3 +83,11 @@ WebSocket-эндпоинт для одного соединения.
- `WsServer` = сервер, который слушает порт и вешает `/ws`.
- `BlockchainWsEndpoint` = обработчик одного WebSocket-подключения, мост между сетью и логикой.
## Постоянные server-to-server соединения
Исходящие межсерверные JSON-запросы проходят через
`server.sync.ServerConnectionPool`. Пул держит одно постоянное исходящее WSS-
соединение на `serverLogin`, отправляет `ServerHello`, поддерживает канал через
WebSocket ping/pong и переподключает его с backoff. DM, settings и blockchain
используют это соединение с разными приоритетами, не меняя свою бизнес-логику.
@@ -12,6 +12,7 @@ import server.sync.PeriodicDmSyncService;
import server.sync.PeriodicUserSettingsSyncService;
import server.sync.SolanaUsersSyncStartupService;
import server.sync.SyncServersBootstrapService;
import server.sync.ServerConnectionPool;
import utils.config.AppConfig;
import java.time.Duration;
@@ -104,6 +105,10 @@ public final class WsServer {
server.start();
log.info("✅ WS сервер запущен на ws://localhost:{}/ws", port);
ServerConnectionPool.getInstance().startOrLog();
Runtime.getRuntime().addShutdownHook(new Thread(
() -> ServerConnectionPool.getInstance().close(),
"server-connection-pool-shutdown"));
PeriodicDmSyncService.startOrLog();
PeriodicUserSettingsSyncService.startOrLog();
server.join();
@@ -41,6 +41,12 @@ 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
server.pool.connectTimeoutSeconds=15
server.pool.callerTimeoutSeconds=35
server.pool.maxQueuePerPeer=2000
server.info.url=
server.info.physicalRegion=
server.info.description=
@@ -5,6 +5,7 @@ import blockchain.body.ConnectionBody;
import blockchain.body.CreateChannelBody;
import blockchain.body.HeaderBody;
import blockchain.body.TextBody;
import shine.db.DbController;
import test.it.blockchain.AddBlockSender;
import test.it.blockchain.ChainState;
import test.it.utils.TestConfig;
@@ -13,6 +14,10 @@ import test.it.utils.log.TestResult;
import test.it.utils.ws.WsSession;
import java.time.Duration;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import static org.junit.jupiter.api.Assertions.*;
@@ -117,6 +122,27 @@ public class IT_03_AddBlock_NoAuth {
st1.registerTextChannelRoot(newsRootBlock, newsRootHash);
}
// CREATE_CHANNEL "Updates" второй публичный канал того же владельца
int updatesRootBlock;
byte[] updatesRootHash;
{
var ln = st1.nextLineByType(ChainState.TYPE_TECH);
sender1.send(new CreateChannelBody(
0,
ln.prevLineNumber, ln.prevLineHash32, ln.thisLineNumber,
"Updates",
"",
CreateChannelBody.CHANNEL_TYPE_PUBLIC,
CreateChannelBody.CHANNEL_TYPE_VERSION_DEFAULT
), t);
updatesRootBlock = st1.lastBlockNumber();
updatesRootHash = st1.getHash32(updatesRootBlock);
assertNotNull(updatesRootHash);
st1.registerTextChannelRoot(updatesRootBlock, updatesRootHash);
}
// POST #0 в канал "News"
int newsPost0Block;
byte[] newsPost0Hash;
@@ -188,7 +214,21 @@ public class IT_03_AddBlock_NoAuth {
bch1, newsRootBlock, newsRootHash,
"U2 follows U1 channel 'News' (target=U1 CREATE_CHANNEL root)", t);
// 3) FRIEND взаимно (на HEADER)
// 3) U2 подписался на второй канал U1 "Updates"
sendConnection(sender2, st2, MsgSubType.CONNECTION_FOLLOW,
bch1, updatesRootBlock, updatesRootHash,
"U2 follows U1 channel 'Updates' (target=U1 CREATE_CHANNEL root)", t);
assertEquals(2, countConnectionsByOwner(u2, u1),
"U2 должен иметь две отдельные записи подписки на каналы U1");
assertEquals(2, countFollowingChannels(u2),
"following_channels_count должен учитывать два разных канала одного владельца");
assertEquals(1, countSubscribers(bch1, newsRootBlock, newsRootHash),
"У канала News должен быть 1 подписчик");
assertEquals(1, countSubscribers(bch1, updatesRootBlock, updatesRootHash),
"У канала Updates должен быть 1 подписчик");
// 4) FRIEND взаимно (на HEADER)
sendConnection(sender1, st1, MsgSubType.CONNECTION_CLOSE_FRIEND,
bch2, u2HeaderBlock, u2HeaderHash,
"U1 -> U2: FRIEND", t);
@@ -197,7 +237,7 @@ public class IT_03_AddBlock_NoAuth {
bch1, u1HeaderBlock, u1HeaderHash,
"U2 -> U1: FRIEND", t);
// 4) CONTACT несколько
// 5) CONTACT несколько
sendConnection(sender1, st1, MsgSubType.CONNECTION_CONTACT,
bch2, u2HeaderBlock, u2HeaderHash,
"U1 -> U2: CONTACT", t);
@@ -236,7 +276,21 @@ public class IT_03_AddBlock_NoAuth {
bch3, u3HeaderBlock, u3HeaderHash,
"U1 -> U3: CONTACT", t);
// 5) U1 убирает U2 из контактов (UNCONTACT)
// 6) U2 отписывается только от News
sendConnection(sender2, st2, MsgSubType.CONNECTION_UNFOLLOW,
bch1, newsRootBlock, newsRootHash,
"U2 unfollows U1 channel 'News'", t);
assertEquals(1, countConnectionsByOwner(u2, u1),
"После отписки от одного канала должна остаться одна запись подписки");
assertEquals(1, countFollowingChannels(u2),
"following_channels_count должен уменьшиться ровно на 1");
assertEquals(0, countSubscribers(bch1, newsRootBlock, newsRootHash),
"После отписки от News у канала должен остаться 0 подписчиков");
assertEquals(1, countSubscribers(bch1, updatesRootBlock, updatesRootHash),
"Подписка на Updates должна остаться");
// 7) U1 убирает U2 из контактов (UNCONTACT)
sendConnection(sender1, st1, MsgSubType.CONNECTION_UNCONTACT,
bch2, u2HeaderBlock, u2HeaderHash,
"U1 -> U2: UNCONTACT", t);
@@ -288,4 +342,58 @@ public class IT_03_AddBlock_NoAuth {
toBlockHash32
), timeout);
}
private static int countConnectionsByOwner(String login, String ownerLogin) {
return queryInt("""
SELECT COUNT(*)
FROM connections_state
WHERE login = ?
AND rel_type = 30
AND to_login = ?
""", login, ownerLogin);
}
private static int countFollowingChannels(String login) {
return queryInt("""
SELECT following_channels_count
FROM user_stats_state
WHERE login = ?
""", login);
}
private static int countSubscribers(String ownerBlockchainName, int rootBlockNumber, byte[] rootBlockHash) {
return queryInt("""
SELECT COALESCE(subscribers_count, 0)
FROM channel_stats_state
WHERE owner_bch_name = ?
AND channel_root_block_number = ?
AND channel_root_block_hash = ?
""", ownerBlockchainName, rootBlockNumber, rootBlockHash);
}
private static int queryInt(String sql, Object... params) {
try (Connection c = DbController.getInstance().getConnection();
PreparedStatement ps = c.prepareStatement(sql)) {
for (int i = 0; i < params.length; i++) {
Object p = params[i];
if (p instanceof String s) {
ps.setString(i + 1, s);
} else if (p instanceof Integer n) {
ps.setInt(i + 1, n);
} else if (p instanceof byte[] bytes) {
ps.setBytes(i + 1, bytes);
} else {
throw new IllegalArgumentException("Unsupported SQL param type: " + (p == null ? "null" : p.getClass()));
}
}
try (ResultSet rs = ps.executeQuery()) {
if (!rs.next()) {
throw new IllegalStateException("Query returned no rows: " + sql);
}
return rs.getInt(1);
}
} catch (SQLException e) {
throw new RuntimeException("DB query failed: " + sql, e);
}
}
}
+2 -2
View File
@@ -1,3 +1,3 @@
backup.schema.version=2
backup.full.version=3
last.full.backup.date=2026-08-08
backup.full.version=4
last.full.backup.date=2026-08-25
+9 -1
View File
@@ -57,7 +57,11 @@
"serverLastGlobalNumber": 128,
"serverLastGlobalHash": "4f...ab",
"serverBlockchainSizeBytes": 45212,
"serverBlockchainSizeLimitBytes": 100000
"serverBlockchainSizeLimitBytes": 100000,
"ownedPublicChannelsCount": 3,
"followingUsersCount": 18,
"followingChannelsCount": 27,
"closeFriendsCount": 4
}
}
```
@@ -68,6 +72,10 @@
- `serverLastGlobalHash` — hash последнего блока (hex-строка 64 символа);
- `serverBlockchainSizeBytes` — текущий размер пользовательского блокчейна на сервере в байтах;
- `serverBlockchainSizeLimitBytes` — текущий лимит размера блокчейна на сервере в байтах;
- `ownedPublicChannelsCount` — количество публичных каналов, владельцем которых является пользователь;
- `followingUsersCount` — количество пользователей, на которых подписан пользователь;
- `followingChannelsCount` — количество публичных каналов, на которые подписан пользователь;
- `closeFriendsCount` — количество близких друзей пользователя.
### Успешный ответ: пользователя нет
+1
View File
@@ -158,6 +158,7 @@
"avaSha256": "0123...",
"avaSize": 248193,
"metaUpdatedAtMs": 1760000000000,
"subscribersCount": 128,
"channelRoot": { "blockNumber": 123, "blockHash": "..." }
},
"metaEvents": [
+2
View File
@@ -32,6 +32,7 @@
| `CloseActiveSession` | `03_Session_Management_API.md` | закрытие активной сессии |
| `AddBlock` | `04_Add_Block_to_Blockchain_API.md` | добавление блока в блокчейн |
| `GetBlockchainBlock` | `04_Add_Block_to_Blockchain_API.md` | чтение одного блока блокчейна |
| `ServerHello` | `16_Server_Connection_Pool_API.md` | объявление server-to-server соединения и возможностей peer без криптографической проверки |
| `Ping` | `05_Technical_Requests_API.md` | keep-alive |
| `GetServerInfo` | `05_Technical_Requests_API.md` | публичная информация о сервере |
| `ListBlockchainHeads` | `05_Technical_Requests_API.md` | список heads всех локальных блокчейнов |
@@ -79,6 +80,7 @@
- `ReceiveOutcomingMessage` зарегистрирован как алиас того же handler/request-класса, что и `SendMessagePair`, и сохраняет прежний межсерверный payload.
- Межсерверные DM-операции пока доверяют `sourceServerLogin`; отдельная межсерверная авторизация запланирована позднее.
- `ServerHello` пока принимает заявленный `serverLogin` на доверии и не является криптографической аутентификацией.
- Отдельных HTTP endpoints для DM-файлов сейчас нет.
- Классы `Net_MarkChannelMessagesSeen_*` существуют в коде, но операция `MarkChannelMessagesSeen` не зарегистрирована в `JsonHandlerRegistry`, поэтому в публичный список API не входит.
- HTTP debug endpoints из `src/main/java/server/debug/` не входят в этот индекс WebSocket `op`; они описаны отдельно в `13_HTTP_Debug_API.md`.
+4 -1
View File
@@ -83,7 +83,10 @@
## 5. Синхронизация
Настройки и DM имеют раздельные таблицы и правила ACK, но периодический процесс открывает один последовательный WS-сеанс с peer: сначала синхронизирует настройки, затем забирает `DmSyncBatch`.
Настройки и DM имеют раздельные таблицы и правила ACK, но периодический процесс
использует один логический последовательный сеанс поверх постоянного WSS-пула:
сначала синхронизирует настройки, затем забирает `DmSyncBatch`. Завершение
логического сеанса не закрывает физический сокет.
- локальная запись создаётся с `synced=false`, если её ещё не подтвердил второй сервер;
- если запись пришла с другого сервера, она сохраняется сразу как `synced=true`;
+13
View File
@@ -1,5 +1,18 @@
# История изменений документации блокчейна
## 2026-08-25 19:45:18 +0400
- Базовый коммит-ориентир: `3a58519`.
- Добавлены runtime-агрегаты статистики:
- `user_stats_state` хранит `owned_public_channels_count`, `following_users_count`, `following_channels_count`, `close_friends_count`;
- `channel_stats_state` хранит `subscribers_count` для публичных каналов.
- Обновление статистики выполняется атомарно вместе с записью блока в той же SQL-транзакции `AddBlock`.
- После полного resync статистика пересобирается из текущих `connections_state` и `channel_names_state`, чтобы старые данные продолжали работать.
## 2026-08-25 19:59:00 +0400
- Базовый коммит-ориентир: `working tree`.
- Триггер `channel_names_state` для статистики сделан устойчивым к старым данным, где канал уже есть, а владельца ещё нет в `solana_user_pda_current`.
- Добавлена совместимая миграция `v15`, чтобы существующие базы перестали падать на старых каналах при старте сервера.
## 2026-08-12 13:00:00 +0400
- Базовый коммит-ориентир: `working tree`.
- Канонический текстовый формат служебных тегов сокращён:
+2
View File
@@ -42,4 +42,6 @@
## Обязательное сопровождение
- При любом изменении формата/правил блокчейна в коде документы этого каталога обновляются в том же наборе изменений.
- Обычный `AddBlock` сейчас пишет через `<blockchainName>.tmp_bch`, `<blockchainName>.write_check` и `<blockchainName>.write_pending`; эта схема и `BlockchainTmpRecoveryOnStartup` должны быть описаны в актуальной документации по синхронизации и recovery.
- Для runtime-агрегатов статистики `user_stats_state` и `channel_stats_state` действует тот же принцип derived state: они обновляются вместе с `AddBlock` и полностью пересобираются при full resync.
- Если в старых данных есть канал владельца, которого ещё нет в `solana_user_pda_current`, сервер не падает: `channel_stats_state` всё равно обновляется, а `user_stats_state` создаётся только после появления пользователя в Solana PDA.
- Каждое обновление документов фиксируется в `CHANGELOG.md` с датой/временем и хэшем коммита-основания.
+14 -8
View File
@@ -183,16 +183,20 @@ Full resync запускается только тогда, когда:
Настройка влияет именно на этап подготовки отсутствующей локальной цепочки во время periodic sync.
## 5. Возможное развитие server-to-server транспорта
## 5. Реализованный постоянный server-to-server транспорт
Этот раздел не описывает текущую DM-доставку. DM уже использует короткие one-shot WebSocket-вызовы, indexed outbox и ACK. Ниже остаётся возможное развитие постоянного транспорта и server-auth.
Этот раздел не меняет текущую семантику DM, settings и blockchain. Он описывает
единый постоянный WSS-транспорт, через который выполняются уже существующие операции.
### 5.1 Межсерверное соединение
- Серверы устанавливают постоянное WebSocket-соединение друг с другом.
- Серверы устанавливают постоянное исходящее WebSocket-соединение друг с другом.
- Адрес партнёра определяется по `server_address` из его Solana PDA.
- Аутентификация: подпись Ed25519 корневым ключом сервера (`root_key` из PDA).
- При разрыве — переподключение с экспоненциальным backoff.
- После подключения отправляется `ServerHello` с `serverLogin`, версией протокола и capabilities.
- На текущем этапе `serverLogin` принимается на доверии; подпись Ed25519 корневым ключом сервера отложена.
- При разрыве выполняется переподключение с jitter/backoff до 60 секунд.
- После 120 секунд отсутствия полезного трафика отправляется WebSocket ping; pong ожидается 15 секунд.
- Один физический канал переиспользуют DM, настройки и blockchain.
### 5.2 Доставка новых данных (push)
@@ -246,19 +250,21 @@ Full resync запускается только тогда, когда:
| Плановый blockchain sync при старте + каждые 12 часов | ✅ Реализовано |
| Обход Solana RPC через `sync.importUserProfileFromPartner.enabled` | ✅ Реализовано |
| Обычный `AddBlock` через `tmp_bch`/`write_check`/`write_pending` | ✅ Реализовано |
| Межсерверный постоянный WebSocket-канал | Нужна реализация |
| Межсерверный постоянный WebSocket-канал | ✅ Реализован общий `ServerConnectionPool` |
| Асинхронная доставка DM на access-серверы получателя | ✅ Реализовано |
| Retry DM до 1 часа + UI-state | ✅ Реализовано |
| Репликация DM на второй access-сервер по `synced` | ✅ Реализовано |
| Read-only `GetDmDeliveryStatus` | ✅ Реализовано |
| Push блоков блокчейна партнёрам | ✅ Реализована базовая one-shot версия |
| Push блоков блокчейна партнёрам | ✅ Выполняется через постоянный WSS-пул |
| Periodic backfill отсутствующего хвоста | ✅ Реализовано |
| Разрешение рассинхрона / divergence | ✅ Реализована базовая full-resync схема во время periodic sync |
| Startup recovery по `*.resync_pending` marker-file | ✅ Реализовано |
| Маршрутизация DM через один/два `access_servers` | ✅ Реализовано |
| Криптографическая server-to-server авторизация DM | Нужна реализация |
Текущая версия сервера умеет синхронизацию блокчейнов и DM. Постоянные server-to-server соединения не требуются для текущей one-shot WS-реализации; отдельной будущей задачей остаётся криптографическая авторизация DM-вызовов.
Текущая версия сервера использует постоянный WSS-пул для существующих
server-to-server JSON-операций. Отдельной будущей задачей остаётся
криптографическая авторизация server-to-server вызовов.
Следующие отдельные шаги после текущего этапа:
- отдельно проверить full-resync и startup-recovery на реальном тестовом прогоне после ручного удаления БД/файлов.
File diff suppressed because one or more lines are too long

After

Width:  |  Height:  |  Size: 20 KiB

+3
View File
@@ -93,6 +93,7 @@ import * as chatView from './pages/chat-view.js?v=202608221218';
import * as userProfileView from './pages/user-profile-view.js';
import * as channelsList from './pages/channels-list.js?v=202608221218';
import * as channelView from './pages/channel-view.js';
import * as channelAboutView from './pages/channel-about-view.js';
import * as channelThreadView from './pages/channel-thread-view.js';
import * as addChannelView from './pages/add-channel-view.js';
import * as addPersonalPublicChatView from './pages/add-personal-public-chat-view.js';
@@ -155,6 +156,7 @@ const routes = {
user: userProfileView,
'channels-list': channelsList,
'channel-view': channelView,
'channel-about-view': channelAboutView,
'channel-thread-view': channelThreadView,
'add-channel-view': addChannelView,
'add-personal-public-chat-view': addPersonalPublicChatView,
@@ -212,6 +214,7 @@ const GUEST_ALLOWED_PAGES = new Set([
'network-view',
'channels-list',
'channel-view',
'channel-about-view',
'channel-thread-view',
'user',
'contact-search-view',
+10
View File
@@ -0,0 +1,10 @@
export const SHINE_CONNECTIONS_LOGO_SRC = '/assets/SHiNE_connections_blue.svg';
export function createShineConnectionsLogo({ className = '' } = {}) {
const img = document.createElement('img');
img.src = SHINE_CONNECTIONS_LOGO_SRC;
img.alt = '';
img.setAttribute('aria-hidden', 'true');
img.className = String(className || '').trim();
return img;
}
+2 -1
View File
@@ -1,6 +1,7 @@
import { resolveToolbarActive } from '../router.js';
import { state } from '../state.js';
import { openAuthRequiredModal } from '../services/auth-required-modal.js';
import { SHINE_CONNECTIONS_LOGO_SRC } from './shine-logo.js';
// iconImg — путь к неоновой PNG (если есть, рисуем картинку вместо эмодзи); glow — цвет доп.свечения
// активной/нажатой вкладки (var --tab-glow); hero — «герой»-вкладка (крупнее/ярче, всегда светится).
@@ -8,7 +9,7 @@ import { openAuthRequiredModal } from '../services/auth-required-modal.js';
const ITEMS = [
{ pageId: 'messages-list', label: 'Личные', icon: '💬', iconImg: '/assets/icon_lichnye.png', glow: 'rgba(0, 229, 255, .6)' },
{ pageId: 'channels-list', label: 'Каналы', icon: '📢', iconImg: '/assets/icon_kanaly.png', glow: 'rgba(0, 229, 255, .6)' },
{ pageId: 'network-view', label: 'Связи', icon: '🕸', iconImg: '/assets/icon_svyazi.png', glow: 'rgba(0, 229, 255, .6)', hero: true },
{ pageId: 'network-view', label: 'Связи', icon: '🕸', iconImg: SHINE_CONNECTIONS_LOGO_SRC, glow: 'rgba(0, 229, 255, .6)', hero: true },
{ pageId: 'notifications-view', label: 'Уведомления', icon: '🔔', iconImg: '/assets/icon_uvedomleniya.png', glow: 'rgba(0, 229, 255, .6)' },
{ pageId: 'profile-view', label: 'Профиль', icon: '👤', iconImg: '/assets/icon_profil.png', glow: 'rgba(0, 229, 255, .6)' },
];
+182
View File
@@ -0,0 +1,182 @@
import { renderHeader } from '../components/header.js';
import { authService, state } from '../state.js';
import { navigateBack } from '../router.js';
import { makeShineChannelRootRoute, makeShineChannelShortRoute } from '../services/shine-routes.js';
import { showToast } from '../services/channels-ux.js';
import { toUserMessage } from '../services/ui-error-texts.js';
export const pageMeta = { id: 'channel-about-view', title: 'О канале' };
function escapeHtml(text) {
return String(text || '')
.replaceAll('&', '&amp;')
.replaceAll('<', '&lt;')
.replaceAll('>', '&gt;')
.replaceAll('"', '&quot;')
.replaceAll("'", '&#39;');
}
function normalizeHash(hash) {
const normalized = String(hash || '').trim().toLowerCase();
return normalized || '0';
}
function toSafeInt(value) {
const parsed = Number(value);
return Number.isFinite(parsed) ? parsed : null;
}
function findCachedChannel(ownerBlockchainName, channelRootBlockNumber, channelRootBlockHash) {
const ownerBch = String(ownerBlockchainName || '').trim().toLowerCase();
const rootNo = Number(channelRootBlockNumber);
const rootHash = normalizeHash(channelRootBlockHash);
const rows = Object.values(state.channelsIndex || {});
return rows.find((row) => (
String(row?.channel?.ownerBlockchainName || '').trim().toLowerCase() === ownerBch
&& Number(row?.channel?.channelRoot?.blockNumber) === rootNo
&& normalizeHash(row?.channel?.channelRoot?.blockHash) === rootHash
)) || null;
}
function buildChannelLink(route) {
if (!route) return '';
const url = new URL(window.location.href);
url.pathname = `/${String(route).replace(/^\/+/, '')}`;
url.hash = '';
return url.toString();
}
function statsText(value) {
return Number.isFinite(Number(value)) ? String(Math.max(0, Number(value))) : '0';
}
export function render({ navigate, route, chrome }) {
const ownerBlockchainName = String(route?.params?.ownerBlockchainName || '').trim();
const channelRootBlockNumber = toSafeInt(route?.params?.channelRootBlockNumber);
const channelRootBlockHash = normalizeHash(route?.params?.channelRootBlockHash);
const channelRoute = makeShineChannelRootRoute({
ownerBlockchainName,
channelRootBlockNumber,
channelRootBlockHash,
});
const screen = document.createElement('section');
screen.className = 'stack channels-screen channels-screen--channel-about';
const topbar = renderHeader({
title: 'О канале',
leftAction: {
label: '←',
onClick: () => {
if (window.history.length > 1) {
navigateBack();
return;
}
if (channelRoute) navigate(channelRoute);
},
ariaLabel: 'Назад',
title: 'Назад',
},
});
chrome?.setTopbar(topbar);
const card = document.createElement('div');
card.className = 'card stack channel-about-card';
card.innerHTML = `
<div class="stack" id="channel-about-content">
<div class="meta-muted">Загрузка данных канала</div>
</div>
`;
const footer = document.createElement('div');
footer.className = 'meta-muted screen-footer';
footer.textContent = 'О канале (channel-about-view)';
screen.append(card, footer);
const renderContent = (channel) => {
const cleanName = String(channel?.displayTitle || channel?.displayName || channel?.channelName || 'Канал').trim();
const ownerName = String(channel?.ownerLogin || channel?.ownerName || 'автор').trim();
const description = String(channel?.channelDescription || channel?.description || '').trim();
const subscribersCount = Number(channel?.subscribersCount || 0);
const aboutRoute = makeShineChannelRootRoute({
ownerBlockchainName: channel?.ownerBlockchainName || ownerBlockchainName,
channelRootBlockNumber: channel?.channelRoot?.blockNumber ?? channelRootBlockNumber,
channelRootBlockHash: channel?.channelRoot?.blockHash ?? channelRootBlockHash,
});
const channelLinkRoute = makeShineChannelShortRoute({
ownerBlockchainName: channel?.ownerBlockchainName || ownerBlockchainName,
channelName: channel?.channelName || '',
});
const channelLink = buildChannelLink(channelLinkRoute);
const changedAtMs = Number(channel?.metaUpdatedAtMs || 0);
const changedAtLabel = changedAtMs ? new Date(changedAtMs).toLocaleString('ru-RU') : '—';
const avatarState = String(channel?.avaAr || '').trim() ? 'Установлен' : 'Не установлен';
const content = card.querySelector('#channel-about-content');
if (!content) return;
content.innerHTML = `
<div class="channel-profile-modal-head">
<h2 class="modal-title">${escapeHtml(cleanName)}</h2>
</div>
<div class="channel-meta-details-grid">
<span>Дата</span><strong>${escapeHtml(changedAtLabel)}</strong>
<span>Владелец</span><strong>${escapeHtml(ownerName)}</strong>
<span>Подписчиков</span><strong>${escapeHtml(statsText(subscribersCount))}</strong>
<span>Системное имя</span><code>${escapeHtml(String(channel?.channelName || '').trim() || 'channel')}</code>
<span>Название</span><strong>${escapeHtml(cleanName)}</strong>
<span>Описание</span><span style="white-space: pre-wrap;">${escapeHtml(description || 'Описание не задано.')}</span>
<span>Аватар</span><span>${escapeHtml(avatarState)}</span>
<span>Ссылка</span><span><a href="${escapeHtml(channelLink)}">${escapeHtml(channelLink)}</a></span>
</div>
<div class="form-actions-grid">
<button class="secondary-btn" type="button" id="channel-about-open">Открыть канал</button>
<button class="secondary-btn" type="button" id="channel-about-copy">Скопировать ссылку</button>
</div>
`;
content.querySelector('#channel-about-open')?.addEventListener('click', () => {
if (!channelLinkRoute) return;
navigate(channelLinkRoute);
});
content.querySelector('#channel-about-copy')?.addEventListener('click', async () => {
if (!channelLink) return;
try {
await navigator.clipboard.writeText(channelLink);
showToast('Ссылка скопирована');
} catch (error) {
showToast(toUserMessage(error, 'Не удалось скопировать ссылку'), { kind: 'error' });
}
});
};
const cached = findCachedChannel(ownerBlockchainName, channelRootBlockNumber, channelRootBlockHash);
if (cached?.channel) {
renderContent({
...cached.channel,
subscribersCount: cached.channel.subscribersCount ?? cached.subscribersCount ?? 0,
});
return screen;
}
void (async () => {
try {
const payload = await authService.getChannelMessages({
ownerBlockchainName,
channelRootBlockNumber,
channelRootBlockHash,
}, 1, 'asc', String(state.session.login || '').trim());
renderContent(payload?.channel || {});
} catch (error) {
const content = card.querySelector('#channel-about-content');
if (content) {
content.innerHTML = `
<div class="meta-muted">Не удалось загрузить данные канала.</div>
<div class="meta-muted">${escapeHtml(toUserMessage(error, 'Проверьте подключение к серверу и повторите попытку.'))}</div>
`;
}
}
})();
return screen;
}
+6 -6
View File
@@ -30,6 +30,7 @@ import {
extractLoginFromBlockchainName,
makeProfileRoute,
makeShineMessageRoute,
makeShineChannelAboutRoute,
} from '../services/shine-routes.js';
import { parseDmTechBlocks } from '../services/dm-tech-blocks.js';
@@ -2550,13 +2551,12 @@ export function render({ navigate, route, chrome }) {
channelHeaderButton.disabled = false;
channelHeaderButton.onclick = (event) => {
animatePress(event.currentTarget);
openAboutChannelModal(apiData.channel, {
canEdit: apiData?.isOwnChannel === true && !apiData?.isDiary && !isStoriesChannel(apiData?.channel),
onEdit: () => openEditChannelModal({
channel: apiData.channel,
onSave: onEditChannelMeta,
}),
const aboutRoute = makeShineChannelAboutRoute({
ownerBlockchainName: apiData?.channel?.ownerBlockchainName || apiData?.selector?.ownerBlockchainName || '',
channelRootBlockNumber: apiData?.selector?.channelRootBlockNumber ?? apiData?.channel?.channelRoot?.blockNumber ?? '',
channelRootBlockHash: apiData?.selector?.channelRootBlockHash ?? apiData?.channel?.channelRoot?.blockHash ?? '',
});
if (aboutRoute) navigate(aboutRoute);
};
}
if (channelEntrypointButton) {
+3 -3
View File
@@ -12,6 +12,7 @@ import {
writeChannelNotificationsState,
} from '../services/channels-ux.js';
import { makeShineChannelRoute } from '../services/shine-routes.js';
import { makeShineChannelShortRoute } from '../services/shine-routes.js';
import { renderAvatar } from '../components/avatar-image.js';
import { createOverflowDots } from '../components/overflow-dots.js';
import { parseDmTechBlocks } from '../services/dm-tech-blocks.js';
@@ -64,10 +65,9 @@ function buildChannelRouteFromSummary(summary, fallbackId) {
const ownerBch = String(summary?.channel?.ownerBlockchainName || '').trim();
const ownerLogin = String(summary?.channel?.ownerLogin || '').trim();
const channelName = String(summary?.channel?.channelName || '').trim();
return makeShineChannelRoute({
ownerLogin,
return makeShineChannelShortRoute({
ownerBlockchainName: ownerBch,
channelName: channelName || fallbackId,
channelName: channelName || fallbackId || ownerLogin,
});
}
+31 -12
View File
@@ -1,6 +1,7 @@
import { renderHeader } from '../components/header.js';
import { renderUserAvatar } from '../components/avatar-image.js';
import { createOverflowDots } from '../components/overflow-dots.js';
import { createShineConnectionsLogo } from '../components/shine-logo.js';
import { directMessages } from '../mock-data.js';
import {
addAppLogEntry,
@@ -30,11 +31,12 @@ import { isTextToSpeechConfigured, speakTextBySettings } from '../services/speec
import { showToast } from '../services/channels-ux.js';
import { buildDmReplyTechBlock, parseDmTechBlocks, sanitizeUserDmTextForSend } from '../services/dm-tech-blocks.js';
import { loadProfileSnapshot } from '../services/user-profile-params.js';
import { makeProfileLinksRoute, makeProfileRoute } from '../services/shine-routes.js';
export const pageMeta = { id: 'chat-view', title: 'Чат' };
const CONVERSATION_CLEAR_NOTICE_TEXT = 'История переписки очищена с этого места';
function createChatHeaderParts(login) {
function createChatHeaderParts(login, navigate) {
const cleanLogin = String(login || '').trim() || 'unknown';
const avatarSlot = document.createElement('span');
@@ -47,6 +49,13 @@ function createChatHeaderParts(login) {
});
avatarSlot.append(initialAvatar);
const avatarButton = document.createElement('button');
avatarButton.type = 'button';
avatarButton.className = 'chat-header-avatar-btn';
avatarButton.title = `Профиль ${cleanLogin}`;
avatarButton.setAttribute('aria-label', `Открыть профиль ${cleanLogin}`);
avatarButton.append(avatarSlot);
const loginEl = document.createElement('span');
loginEl.className = 'chat-header-login';
loginEl.setAttribute('role', 'heading');
@@ -75,7 +84,23 @@ function createChatHeaderParts(login) {
})
.catch(() => {});
return { centerNode: loginEl, avatarSlot };
const connectionsButton = document.createElement('button');
connectionsButton.type = 'button';
connectionsButton.className = 'icon-btn chat-header-icon-btn chat-header-connections-btn';
connectionsButton.title = `Связи ${cleanLogin}`;
connectionsButton.setAttribute('aria-label', `Открыть связи ${cleanLogin}`);
connectionsButton.append(createShineConnectionsLogo({ className: 'chat-header-connections-logo' }));
connectionsButton.addEventListener('click', () => {
if (!cleanLogin || cleanLogin === 'unknown') return;
navigate(makeProfileLinksRoute(cleanLogin));
});
avatarButton.addEventListener('click', () => {
if (!cleanLogin || cleanLogin === 'unknown') return;
navigate(makeProfileRoute(cleanLogin));
});
return { centerNode: loginEl, avatarButton, connectionsButton };
}
function truncatePreviewText(value, maxLen = 72) {
@@ -293,7 +318,6 @@ function openChatActionsMenu({
anchorY = 0,
onCall,
onVideoCall,
onInstantVideoCall,
onClearHistory,
onDeleteChat,
}) {
@@ -305,8 +329,7 @@ function openChatActionsMenu({
<div class="dm-floating-menu-layer" id="chat-header-actions-layer">
<div class="modal-card stack dm-message-actions-menu dm-message-actions-popover" id="${menuId}">
<button class="secondary-btn dm-message-action-btn" type="button" id="chat-menu-call">Звонок</button>
<button class="secondary-btn dm-message-action-btn" type="button" id="chat-menu-video-call">Звонок с поддержкой видео</button>
<button class="secondary-btn dm-message-action-btn" type="button" id="chat-menu-instant-video-call">Видеозвонок</button>
<button class="secondary-btn dm-message-action-btn" type="button" id="chat-menu-video-call">Видеозвонок</button>
<button class="secondary-btn dm-message-action-btn" type="button" id="chat-menu-clear-history">Очистить историю</button>
<button class="secondary-btn dm-message-action-btn dm-message-action-btn--danger" type="button" id="chat-menu-delete-chat">Удалить чат</button>
</div>
@@ -358,10 +381,6 @@ function openChatActionsMenu({
close();
if (typeof onVideoCall === 'function') await onVideoCall();
});
root.querySelector('#chat-menu-instant-video-call')?.addEventListener('click', async () => {
close();
if (typeof onInstantVideoCall === 'function') await onInstantVideoCall();
});
root.querySelector('#chat-menu-clear-history')?.addEventListener('click', async () => {
close();
if (typeof onClearHistory === 'function') await onClearHistory();
@@ -993,7 +1012,7 @@ export function render({ navigate, route, chrome }) {
const log = document.createElement('div');
log.className = 'messages-log dm-messages-log';
const chatHeaderParts = createChatHeaderParts(chatId);
const chatHeaderParts = createChatHeaderParts(chatId, navigate);
const chatHeader = renderHeader({
centerNode: chatHeaderParts.centerNode,
leftAction: { label: '←', onClick: () => navigate('messages-list') },
@@ -1016,7 +1035,6 @@ export function render({ navigate, route, chrome }) {
anchorY: Number(event?.currentTarget?.getBoundingClientRect?.().bottom || event?.clientY || 0),
onCall: () => handleStartCall('audio'),
onVideoCall: () => handleStartCall('video'),
onInstantVideoCall: () => handleStartCall('instant_video'),
onClearHistory: async () => {
openChatConfirmModal({
title: 'Очистить историю?',
@@ -1064,7 +1082,8 @@ export function render({ navigate, route, chrome }) {
},
],
});
chatHeader.querySelector('.header-left')?.append(chatHeaderParts.avatarSlot);
const chatHeaderLeft = chatHeader.querySelector('.header-left');
chatHeaderLeft?.append(chatHeaderParts.connectionsButton, chatHeaderParts.avatarButton);
chrome?.setTopbar(chatHeader);
if (!isKnownContact) {
+5 -5
View File
@@ -8,6 +8,7 @@ import {
} from '../state.js';
import { renderUserAvatar } from '../components/avatar-image.js';
import { createOverflowDots } from '../components/overflow-dots.js';
import { createShineConnectionsLogo } from '../components/shine-logo.js';
import { loadProfileSnapshot } from '../services/user-profile-params.js';
import { parseDmTechBlocks } from '../services/dm-tech-blocks.js';
@@ -167,15 +168,11 @@ function compareChatRows(a, b) {
export function render({ navigate, chrome }) {
const screen = document.createElement('section');
screen.className = 'stack dm-screen dm-list-screen';
const login = String(state.session.login || '').trim();
const head = document.createElement('header');
head.className = 'dm-head';
head.innerHTML = `
<div class="dm-head-brand">
<div class="dm-head-hex">${(login[0] || 'A').toUpperCase()}</div>
<div class="dm-head-id">
<span class="dm-head-name"></span>
</div>
<span class="dm-head-logo-wrap" aria-hidden="true"></span>
</div>
<h1 class="dm-head-title">Чаты</h1>
<div class="dm-head-menu-wrap">
@@ -191,6 +188,9 @@ export function render({ navigate, chrome }) {
</div>
</div>
`;
head.querySelector('.dm-head-logo-wrap')?.append(
createShineConnectionsLogo({ className: 'dm-head-logo' }),
);
const menuButton = head.querySelector('.dm-head-menu-btn');
menuButton?.append(createOverflowDots());
+39 -2
View File
@@ -1,5 +1,5 @@
import { profile } from '../mock-data.js';
import { state } from '../state.js';
import { authService, state } from '../state.js';
import {
PROFILE_GENDER_FEMALE,
PROFILE_GENDER_MALE,
@@ -223,6 +223,12 @@ export function render({ navigate, chrome }) {
let currentToggles = [];
let currentGender = 'unknown';
let currentAvatar = { value: '', source: '', txId: '', sha256Hex: '', timeMs: 0 };
let currentStats = {
ownedPublicChannelsCount: 0,
followingUsersCount: 0,
followingChannelsCount: 0,
closeFriendsCount: 0,
};
function syncIdentity() {
if (!identityEl) return;
@@ -269,6 +275,21 @@ export function render({ navigate, chrome }) {
updateToggleButton(shineBtn, 'Сияющий', shine.enabled);
}
function renderStats() {
const stats = [
{ label: 'Собственные публичные каналы', value: currentStats.ownedPublicChannelsCount },
{ label: 'Подписки на пользователей', value: currentStats.followingUsersCount },
{ label: 'Подписки на каналы', value: currentStats.followingChannelsCount },
{ label: 'Близкие друзья', value: currentStats.closeFriendsCount },
];
stats.forEach((stat) => {
const row = document.createElement('div');
row.className = 'card profile-param-item row';
row.innerHTML = `<div class="profile-param-value"><b>${escapeHtml(stat.label)}</b>: ${escapeHtml(String(Number(stat.value || 0)))}</div>`;
listWrap.append(row);
});
}
officialBtn?.classList.add('profile-badge-trigger');
shineBtn?.classList.add('profile-badge-trigger');
officialBtn?.addEventListener('click', () => {
@@ -286,6 +307,7 @@ export function render({ navigate, chrome }) {
function renderFields(fields) {
listWrap.innerHTML = '';
renderStats();
fields.forEach((field) => {
const row = document.createElement('div');
row.className = 'card profile-param-item row';
@@ -317,6 +339,12 @@ export function render({ navigate, chrome }) {
];
currentGender = 'unknown';
currentAvatar = { value: '', source: '', txId: '', sha256Hex: '', timeMs: 0 };
currentStats = {
ownedPublicChannelsCount: 0,
followingUsersCount: 0,
followingChannelsCount: 0,
closeFriendsCount: 0,
};
syncIdentity();
updateAvatarUi();
updateTogglesUi();
@@ -325,11 +353,20 @@ export function render({ navigate, chrome }) {
}
try {
const snapshot = await loadProfileSnapshot(login);
const [snapshot, user] = await Promise.all([
loadProfileSnapshot(login),
authService.getUser(login).catch(() => ({})),
]);
currentFields = Array.isArray(snapshot.fields) ? snapshot.fields : [];
currentToggles = Array.isArray(snapshot.toggles) ? snapshot.toggles : [];
currentGender = snapshot.gender || 'unknown';
currentAvatar = snapshot.avatar || { value: '', source: '', txId: '', sha256Hex: '', timeMs: 0 };
currentStats = {
ownedPublicChannelsCount: Number(user?.ownedPublicChannelsCount || 0),
followingUsersCount: Number(user?.followingUsersCount || 0),
followingChannelsCount: Number(user?.followingChannelsCount || 0),
closeFriendsCount: Number(user?.closeFriendsCount || 0),
};
syncIdentity();
updateAvatarUi();
updateTogglesUi();
+45 -1
View File
@@ -26,6 +26,7 @@ const PRETTY_PATHS = new Map([
['add-channel-view', 'channels/new'],
['add-personal-public-chat-view', 'channels/new-public-chat'],
['channel-view', 'channel'],
['channel-about-view', 'channel/about'],
['channel-thread-view', 'thread'],
['network-view', 'network'],
['notifications-view', 'notifications'],
@@ -52,6 +53,10 @@ const PRETTY_PATHS = new Map([
['remote-addblock-session-view', 'remote-addblock-session'],
]);
function looksLikeBlockchainName(value) {
return /^.+-\d+$/.test(String(value || '').trim());
}
export const PRE_AUTH_PAGES = [
'start-view',
'entry-settings-view',
@@ -162,6 +167,34 @@ export function parseRouteFromPath(pathname = '') {
}
}
if (segments.length >= 2 && looksLikeBlockchainName(segments[0])) {
const ownerBlockchainName = decodePart(segments[0]);
const channelName = decodePart(segments[1] || '');
const sub = decodePart(segments[2] || '').toLowerCase();
if (ownerBlockchainName && channelName) {
if (sub === 'about') {
return {
pageId: 'channel-about-view',
params: {
ownerBlockchainName,
channelRootBlockNumber: '',
channelRootBlockHash: '',
channelId: '',
channelName,
},
};
}
return {
pageId: 'channel-view',
params: {
ownerBlockchainName,
channelName,
channelId: '',
},
};
}
}
if (pageId === 'chat' || pageId === 'chat-view') {
return { pageId: 'chat-view', params: { chatId: dynamicId ? decodeURIComponent(dynamicId) : '' } };
}
@@ -225,6 +258,17 @@ export function parseRouteFromPath(pathname = '') {
}
if (pageId === 'channel') {
if (segments.length >= 5 && decodePart(segments[4] || '').toLowerCase() === 'about') {
return {
pageId: 'channel-about-view',
params: {
ownerBlockchainName: decodePart(segments[1]),
channelRootBlockNumber: segments[2] || '',
channelRootBlockHash: segments[3] || '',
channelId: '',
},
};
}
if (segments.length >= 4) {
return {
pageId: 'channel-view',
@@ -408,7 +452,7 @@ export function resolveToolbarActive(pageId) {
pageId === 'solana-users-init-view'
) return 'profile-view';
if (pageId === 'chat-view' || pageId === 'contact-search-view') return 'messages-list';
if (pageId === 'channel-view' || pageId === 'channel-thread-view' || pageId === 'add-channel-view' || pageId === 'add-personal-public-chat-view') return 'channels-list';
if (pageId === 'channel-view' || pageId === 'channel-about-view' || pageId === 'channel-thread-view' || pageId === 'add-channel-view' || pageId === 'add-personal-public-chat-view') return 'channels-list';
if (pageId === 'user') return 'messages-list';
return 'profile-view';
}
+2 -2
View File
@@ -153,7 +153,7 @@ function getCallTitleText(mode) {
return 'Видеозвонок';
}
if (normalizeCallMode(mode) === CALL_MODE_VIDEO) {
return 'Звонок с поддержкой видео';
return 'Видеозвонок';
}
return 'Звонок';
}
@@ -164,7 +164,7 @@ function getIncomingCallStatusText(peerLogin, mode) {
return `Входящий видеозвонок от ${name}`;
}
if (isVideoCallMode(mode)) {
return `Вам звонит ${name} (звонок с поддержкой видео)`;
return `Входящий видеозвонок от ${name}`;
}
return `Вам звонит ${name}`;
}
+20
View File
@@ -51,6 +51,26 @@ export function makeShineChannelRoute({ ownerLogin = '', ownerBlockchainName = '
return `${SHINE_ROUTE_SEGMENT}/${encodeRoutePart(ownerBch)}/${encodeRoutePart(chName)}`;
}
export function makeShineChannelRootRoute({ ownerBlockchainName = '', channelRootBlockNumber = '', channelRootBlockHash = '' }) {
const ownerBch = String(ownerBlockchainName || '').trim();
const rootNo = String(channelRootBlockNumber || '').trim();
const rootHash = String(channelRootBlockHash || '').trim();
if (!ownerBch || !rootNo || !rootHash) return '';
return `channel/${encodeRoutePart(ownerBch)}/${encodeRoutePart(rootNo)}/${encodeRoutePart(rootHash)}`;
}
export function makeShineChannelAboutRoute({ ownerBlockchainName = '', channelRootBlockNumber = '', channelRootBlockHash = '' }) {
const base = makeShineChannelRootRoute({ ownerBlockchainName, channelRootBlockNumber, channelRootBlockHash });
return base ? `${base}/about` : '';
}
export function makeShineChannelShortRoute({ ownerBlockchainName = '', channelName = '' }) {
const ownerBch = String(ownerBlockchainName || '').trim();
const chName = String(channelName || '').trim();
if (!ownerBch || !chName) return '';
return `${encodeRoutePart(ownerBch)}/${encodeRoutePart(chName)}`;
}
export function makeShineMessageRoute({ ownerLogin = '', messageBlockchainName = '', messageBlockNumber = '' }) {
const msgBch = String(messageBlockchainName || '').trim();
const msgNo = String(messageBlockNumber || '').trim();
+75 -6
View File
@@ -5927,6 +5927,14 @@ textarea.input {
transform: translateY(-40%);
}
.channels-screen--channel-about .screen-footer {
margin-top: auto;
padding: 8px 4px 0;
text-align: center;
font-size: 12px;
letter-spacing: 0.02em;
}
.entrypoint-history-list {
display: grid;
gap: 10px;
@@ -6184,12 +6192,13 @@ textarea.input {
background: linear-gradient(180deg, rgba(10,12,18,0.82), rgba(10,12,18,0.0));
}
.dm-head-brand { display: flex; align-items: center; gap: 8px; min-width: 0; }
.dm-head-hex {
width: 32px; height: 32px; flex: 0 0 auto; display: grid; place-items: center;
font-weight: 700; font-size: 15px; color: #1a1205;
background: linear-gradient(150deg, #F0B82E, #D49F22);
clip-path: polygon(50% 0%, 100% 25%, 100% 75%, 50% 100%, 0% 75%, 0% 25%);
box-shadow: 0 0 14px rgba(240, 184, 46, 0.35);
.dm-head-logo-wrap { display: inline-flex; align-items: center; justify-content: center; width: 38px; height: 38px; flex: 0 0 auto; }
.dm-head-logo {
width: 36px;
height: 36px;
object-fit: contain;
display: block;
filter: drop-shadow(0 0 7px rgba(71, 196, 255, 0.38));
}
.dm-head-id { min-width: 0; display: grid; }
.dm-head-name { font-size: 15px; font-weight: 600; color: var(--text); overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
@@ -9100,6 +9109,66 @@ body.chat-topbar-overlay .app-shell.keyboard-open .scroll-to-bottom-btn {
gap: 0;
}
.chat-header-connections-btn,
.chat-header-avatar-btn {
padding: 0;
min-width: 0;
min-height: 0;
border: 0;
background: transparent;
box-shadow: none;
transform: none;
}
.chat-header-connections-btn:hover,
.chat-header-connections-btn:active,
.chat-header-connections-btn:focus,
.chat-header-connections-btn:focus-visible,
.chat-header-avatar-btn:hover,
.chat-header-avatar-btn:active,
.chat-header-avatar-btn:focus,
.chat-header-avatar-btn:focus-visible {
border: 0;
background: transparent;
box-shadow: none;
transform: none;
}
.chat-header-connections-btn {
width: 38px;
height: 38px;
border-radius: 999px;
display: inline-flex;
align-items: center;
justify-content: center;
background: rgba(8, 17, 31, 0.65);
border: 1px solid rgba(97, 198, 255, 0.26);
box-shadow: 0 0 0 1px rgba(97, 198, 255, 0.08), 0 0 12px rgba(64, 172, 255, 0.18);
}
.chat-header-connections-btn:hover,
.chat-header-connections-btn:focus-visible {
border-color: rgba(111, 208, 255, 0.38);
box-shadow: 0 0 0 1px rgba(111, 208, 255, 0.16), 0 0 16px rgba(64, 172, 255, 0.28);
}
.chat-header-connections-logo {
width: 24px;
height: 24px;
object-fit: contain;
display: block;
}
.chat-header-avatar-btn {
display: inline-flex;
align-items: center;
justify-content: center;
}
.chat-header-avatar-btn .chat-header-avatar-slot {
display: inline-flex;
}
/* Пузырям сообщений не нужна отдельная обводка: направление читается по фону и геометрии. */
.dm-chat-screen .bubble,
.dm-chat-screen .bubble.in,