diff --git a/SHiNE-server/shine-server-config/src/main/java/utils/config/AppConfig.java b/SHiNE-server/shine-server-config/src/main/java/utils/config/AppConfig.java index 0d53fe4c..027779b0 100644 --- a/SHiNE-server/shine-server-config/src/main/java/utils/config/AppConfig.java +++ b/SHiNE-server/shine-server-config/src/main/java/utils/config/AppConfig.java @@ -59,6 +59,8 @@ public final class AppConfig { public String getParam(String name) { String fromSystem = System.getProperty(name); if (fromSystem != null) return fromSystem; + String fromEnv = System.getenv(toEnvName(name)); + if (fromEnv != null && !fromEnv.isBlank()) return fromEnv.trim(); return properties.getProperty(name); } @@ -78,4 +80,11 @@ public final class AppConfig { String v = properties.getProperty(name); return v == null ? defaultValue : Boolean.parseBoolean(v); } + + private static String toEnvName(String name) { + return name + .replace('.', '_') + .replace('-', '_') + .toUpperCase(); + } } diff --git a/SHiNE-server/shine-server-solana-users-sync/build.gradle b/SHiNE-server/shine-server-solana-users-sync/build.gradle new file mode 100644 index 00000000..74314d62 --- /dev/null +++ b/SHiNE-server/shine-server-solana-users-sync/build.gradle @@ -0,0 +1,32 @@ +plugins { + id 'java' +} + +group = 'shine' +version = '1.0.0' + +java { + toolchain { + languageVersion = JavaLanguageVersion.of(17) + } +} + +repositories { + mavenCentral() +} + +dependencies { + implementation project(':shine-server-config') + + implementation 'com.squareup.okhttp3:okhttp:4.12.0' + implementation 'com.fasterxml.jackson.core:jackson-databind:2.17.2' + implementation 'org.postgresql:postgresql:42.7.7' + implementation 'org.bouncycastle:bcprov-jdk18on:1.78.1' + implementation 'org.slf4j:slf4j-api:2.0.16' + + testImplementation 'org.junit.jupiter:junit-jupiter:5.11.0' +} + +test { + useJUnitPlatform() +} diff --git a/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/codec/ShineUsersCodec.java b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/codec/ShineUsersCodec.java new file mode 100644 index 00000000..38597f90 --- /dev/null +++ b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/codec/ShineUsersCodec.java @@ -0,0 +1,1444 @@ +package sync.codec; + +import com.fasterxml.jackson.databind.JsonNode; +import sync.util.Base58Util; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Base64; +import java.util.HexFormat; +import java.util.List; + +public final class ShineUsersCodec { + + public static final String MAGIC = + "SHiNE"; + + public static final int IX_INIT_USERS_ECONOMY_CONFIG = + 1; + + public static final int IX_UPDATE_USERS_ECONOMY_CONFIG = + 2; + + public static final int IX_CREATE_USER_PDA = + 3; + + public static final int IX_UPDATE_USER_PDA = + 4; + + public static final long START_REGISTRATION_FEE_LAMPORTS = + 10_000_000L; + + public static final long START_LAMPORTS_PER_LIMIT_STEP = + 100_000L; + + public static final long START_BONUS_LIMIT = + 100_000L; + + private static final int BLOCK_TYPE_RECOVERY_KEY = + 0; + + private static final int BLOCK_TYPE_ROOT_KEY = + 1; + + private static final int BLOCK_TYPE_CLIENT_KEY = + 2; + + private static final int BLOCK_TYPE_BLOCKCHAIN_REGISTRY = + 3; + + private static final int BLOCK_TYPE_SERVER_PROFILE = + 30; + + private static final int BLOCK_TYPE_ACCESS_SERVERS = + 40; + + private static final int BLOCK_TYPE_SESSIONS = + 50; + + private static final int BLOCK_TYPE_TRUSTED_STATE = + 70; + + private static final int BLOCK_VERSION_0 = + 0; + + private static final int BLOCKCHAIN_TYPE_MAIN_USER = + 1; + + private static final byte[] ZERO_HASH = + new byte[32]; + + private ShineUsersCodec() { + } + + public static ParsedInstruction parseShineUsersInstruction( + JsonNode instructionNode, + String expectedProgramId + ) { + + if (instructionNode == null) { + return null; + } + + String programId = + instructionNode.path("programId") + .asText(""); + + if (!expectedProgramId.equals(programId)) { + return null; + } + + String dataBase58 = + instructionNode.path("data") + .asText(""); + + if (dataBase58.isBlank()) { + return new ParsedInstruction( + TxKind.OTHER, + false, + null, + null, + null, + null + ); + } + + byte[] data = + Base58Util.decode( + dataBase58 + ); + + Reader reader = + new Reader(data); + + int tag = + reader.readU8(); + + List accounts = + readInstructionAccounts( + instructionNode.path("accounts") + ); + + return switch (tag) { + case IX_INIT_USERS_ECONOMY_CONFIG -> + new ParsedInstruction( + TxKind.INIT_USERS_ECONOMY_CONFIG, + false, + null, + null, + EconomyConfigState.initial(), + null + ); + + case IX_UPDATE_USERS_ECONOMY_CONFIG -> + new ParsedInstruction( + TxKind.UPDATE_USERS_ECONOMY_CONFIG, + false, + null, + null, + new EconomyConfigState( + 1, + reader.readU64(), + reader.readU64(), + reader.readU64() + ), + null + ); + + case IX_CREATE_USER_PDA -> + parseCreateInstruction( + reader, + accounts + ); + + case IX_UPDATE_USER_PDA -> + parseUpdateInstruction( + reader, + accounts + ); + + default -> + new ParsedInstruction( + TxKind.OTHER, + false, + null, + null, + null, + null + ); + }; + } + + public static UserPdaSnapshot parseUserPdaAccount( + String pdaAddress, + long slot, + String rawDataBase64, + String lastTxSignature + ) { + + byte[] raw = + Base64.getDecoder() + .decode(rawDataBase64); + + if (raw.length < 9) { + throw new IllegalArgumentException( + "PDA record too short" + ); + } + + String magic = + new String( + raw, + 0, + 5, + StandardCharsets.UTF_8 + ); + + if (!MAGIC.equals(magic)) { + throw new IllegalArgumentException( + "Unsupported PDA magic" + ); + } + + int recordLen = + u16le( + raw, + 7 + ); + + if (recordLen > raw.length) { + throw new IllegalArgumentException( + "Invalid PDA record length" + ); + } + + byte[] useful = + new byte[recordLen]; + + System.arraycopy( + raw, + 0, + useful, + 0, + recordLen + ); + + Reader reader = + new Reader(useful); + + reader.skip(9); + + long createdAtMs = + reader.readU64(); + + long updatedAtMs = + reader.readU64(); + + int recordNumber = + (int) reader.readU32(); + + byte[] prevHash = + reader.readFixed(32); + + String login = + reader.readStringU8(); + + int blocksCount = + reader.readU8(); + + String recoveryKey = + null; + + String rootKey = + null; + + String clientKey = + null; + + String blockchainName = + null; + + String blockchainKey = + null; + + long paidLimitBytes = + 0L; + + long usedBytes = + 0L; + + int lastBlockNumber = + 0; + + String lastBlockHash = + ""; + + String lastBlockSignature = + ""; + + String arweaveTxId = + ""; + + boolean isServer = + false; + + int addressFormatType = + 0; + + int addressFormatVersion = + 0; + + String serverAddress = + ""; + + List syncServers = + new ArrayList<>(); + + List accessServers = + new ArrayList<>(); + + int sessionsMode = + 1; + + List sessions = + new ArrayList<>(); + + int trustedCount = + 0; + + for (int i = 0; i < blocksCount; i++) { + + int blockType = + reader.readU8(); + + int blockVersion = + reader.readU8(); + + if (blockVersion != BLOCK_VERSION_0) { + throw new IllegalArgumentException( + "Unsupported block version" + ); + } + + switch (blockType) { + + case BLOCK_TYPE_RECOVERY_KEY -> + recoveryKey = + Base58Util.encode( + reader.readFixed(32) + ); + + case BLOCK_TYPE_ROOT_KEY -> + rootKey = + Base58Util.encode( + reader.readFixed(32) + ); + + case BLOCK_TYPE_CLIENT_KEY -> + clientKey = + Base58Util.encode( + reader.readFixed(32) + ); + + case BLOCK_TYPE_BLOCKCHAIN_REGISTRY -> { + int count = + reader.readU8(); + + if (count != 1) { + throw new IllegalArgumentException( + "Unexpected blockchain count" + ); + } + + int blockchainType = + reader.readU8(); + + if (blockchainType != BLOCKCHAIN_TYPE_MAIN_USER) { + throw new IllegalArgumentException( + "Unsupported blockchain type" + ); + } + + blockchainName = + reader.readStringU8(); + + blockchainKey = + Base58Util.encode( + reader.readFixed(32) + ); + + paidLimitBytes = + reader.readU64(); + + usedBytes = + reader.readU64(); + + lastBlockNumber = + (int) reader.readU32(); + + lastBlockHash = + toHex( + reader.readFixed(32) + ); + + lastBlockSignature = + Base58Util.encode( + reader.readFixed(64) + ); + + int arweavePresent = + reader.readU8(); + + if (arweavePresent == 1) { + arweaveTxId = + reader.readStringU8(); + } else if (arweavePresent != 0) { + throw new IllegalArgumentException( + "Invalid arweave marker" + ); + } + } + + case BLOCK_TYPE_SERVER_PROFILE -> { + isServer = + reader.readU8() == 1; + + if (isServer) { + addressFormatType = + reader.readU8(); + addressFormatVersion = + reader.readU8(); + serverAddress = + reader.readStringU8(); + int syncCount = + reader.readU8(); + for (int j = 0; j < syncCount; j++) { + syncServers.add( + reader.readStringU8() + ); + } + } + } + + case BLOCK_TYPE_ACCESS_SERVERS -> { + int accessCount = + reader.readU8(); + for (int j = 0; j < accessCount; j++) { + accessServers.add( + reader.readStringU8() + ); + } + } + + case BLOCK_TYPE_SESSIONS -> { + sessionsMode = + reader.readU8(); + + int sessionCount = + reader.readU8(); + + for (int j = 0; j < sessionCount; j++) { + sessions.add( + new UserSessionSnapshot( + reader.readU8(), + reader.readU8(), + reader.readStringU8(), + Base58Util.encode( + reader.readFixed(32) + ) + ) + ); + } + } + + case BLOCK_TYPE_TRUSTED_STATE -> + trustedCount = + reader.readU8(); + + default -> + throw new IllegalArgumentException( + "Unsupported block type: " + blockType + ); + } + } + + String signature = + Base58Util.encode( + reader.readFixed(64) + ); + + return new UserPdaSnapshot( + pdaAddress, + login, + recordNumber, + slot, + lastTxSignature, + recoveryKey, + rootKey, + clientKey, + blockchainName, + blockchainKey, + paidLimitBytes, + usedBytes, + lastBlockNumber, + lastBlockHash, + lastBlockSignature, + arweaveTxId, + isServer, + addressFormatType, + addressFormatVersion, + serverAddress, + List.copyOf(syncServers), + List.copyOf(accessServers), + sessionsMode, + List.copyOf(sessions), + trustedCount, + createdAtMs, + updatedAtMs, + toHex(prevHash), + signature, + rawDataBase64 + ); + } + + public static UserPdaSnapshot buildCreateSnapshot( + UserPdaMutation mutation, + EconomyConfigState economyConfigState, + String txSignature, + long slot + ) { + + long paidLimitBytes = + economyConfigState.startBonusLimit() + + mutation.additionalLimit(); + + return buildSnapshot( + mutation, + 0, + slot, + txSignature, + mutation.createdAtMs(), + mutation.createdAtMs(), + toHex(ZERO_HASH), + paidLimitBytes + ); + } + + public static UserPdaSnapshot buildUpdateSnapshot( + UserPdaMutation mutation, + UserPdaSnapshot previous, + String txSignature, + long slot + ) { + + long paidLimitBytes = + previous.paidLimitBytes() + + mutation.additionalLimit(); + + return buildSnapshot( + mutation, + mutation.version(), + slot, + txSignature, + mutation.createdAtMs(), + mutation.updatedAtMs(), + toHex(mutation.prevHash()), + paidLimitBytes + ); + } + + public static String serializeSnapshotToBase64( + UserPdaSnapshot snapshot + ) { + + byte[] recoveryKey = + Base58Util.decode( + snapshot.recoveryKey() + ); + + byte[] rootKey = + Base58Util.decode( + snapshot.rootKey() + ); + + byte[] clientKey = + Base58Util.decode( + snapshot.clientKey() + ); + + byte[] blockchainKey = + Base58Util.decode( + snapshot.blockchainKey() + ); + + byte[] lastBlockHash = + fromHex( + snapshot.lastBlockHash() + ); + + byte[] lastBlockSignature = + Base58Util.decode( + snapshot.lastBlockSignature() + ); + + byte[] prevHash = + fromHex( + snapshot.prevRecordHash() + ); + + byte[] recordSignature = + Base58Util.decode( + snapshot.recordSignature() + ); + + List syncServers = + snapshot.syncServers(); + + List accessServers = + snapshot.accessServers(); + + List sessions = + snapshot.sessions(); + + byte[] loginBytes = + snapshot.login() + .getBytes(StandardCharsets.UTF_8); + + List out = + new ArrayList<>(); + + pushFixed(out, MAGIC.getBytes(StandardCharsets.UTF_8)); + out.add((byte) 1); + out.add((byte) 0); + pushU16(out, 0); + pushU64(out, snapshot.createdAtMs()); + pushU64(out, snapshot.updatedAtMs()); + pushU32(out, snapshot.recordNumber()); + pushFixed(out, prevHash); + pushStringU8(out, loginBytes); + + int blocksCount = + snapshot.isServer() + ? 8 + : 7; + + out.add((byte) blocksCount); + + out.add((byte) BLOCK_TYPE_RECOVERY_KEY); + out.add((byte) BLOCK_VERSION_0); + pushFixed(out, recoveryKey); + + out.add((byte) BLOCK_TYPE_ROOT_KEY); + out.add((byte) BLOCK_VERSION_0); + pushFixed(out, rootKey); + + out.add((byte) BLOCK_TYPE_CLIENT_KEY); + out.add((byte) BLOCK_VERSION_0); + pushFixed(out, clientKey); + + out.add((byte) BLOCK_TYPE_BLOCKCHAIN_REGISTRY); + out.add((byte) BLOCK_VERSION_0); + out.add((byte) 1); + out.add((byte) BLOCKCHAIN_TYPE_MAIN_USER); + pushString(out, snapshot.blockchainName()); + pushFixed(out, blockchainKey); + pushU64(out, snapshot.paidLimitBytes()); + pushU64(out, snapshot.usedBytes()); + pushU32(out, snapshot.lastBlockNumber()); + pushFixed(out, lastBlockHash); + pushFixed(out, lastBlockSignature); + + if (snapshot.arweaveTxId().isBlank()) { + out.add((byte) 0); + } else { + out.add((byte) 1); + pushString(out, snapshot.arweaveTxId()); + } + + if (snapshot.isServer()) { + out.add((byte) BLOCK_TYPE_SERVER_PROFILE); + out.add((byte) BLOCK_VERSION_0); + out.add((byte) 1); + out.add((byte) snapshot.addressFormatType()); + out.add((byte) snapshot.addressFormatVersion()); + pushString(out, snapshot.serverAddress()); + out.add((byte) syncServers.size()); + for (String syncServer : syncServers) { + pushString(out, syncServer); + } + } + + out.add((byte) BLOCK_TYPE_ACCESS_SERVERS); + out.add((byte) BLOCK_VERSION_0); + out.add((byte) accessServers.size()); + for (String accessServer : accessServers) { + pushString(out, accessServer); + } + + out.add((byte) BLOCK_TYPE_SESSIONS); + out.add((byte) BLOCK_VERSION_0); + out.add((byte) snapshot.sessionsMode()); + out.add((byte) sessions.size()); + for (UserSessionSnapshot session : sessions) { + out.add((byte) session.sessionType()); + out.add((byte) session.sessionVersion()); + pushString(out, session.sessionName()); + pushFixed( + out, + Base58Util.decode( + session.sessionPubKey() + ) + ); + } + + out.add((byte) BLOCK_TYPE_TRUSTED_STATE); + out.add((byte) BLOCK_VERSION_0); + out.add((byte) snapshot.trustedCount()); + + int recordLength = + out.size() + 64; + + byte[] recordLengthBytes = + new byte[]{ + (byte) (recordLength & 0xFF), + (byte) ((recordLength >> 8) & 0xFF) + }; + + out.set(7, recordLengthBytes[0]); + out.set(8, recordLengthBytes[1]); + + pushFixed(out, recordSignature); + + byte[] bytes = + new byte[out.size()]; + + for (int i = 0; i < out.size(); i++) { + bytes[i] = out.get(i); + } + + return Base64.getEncoder() + .encodeToString(bytes); + } + + private static ParsedInstruction parseCreateInstruction( + Reader reader, + List accounts + ) { + + CreateOrUpdateArgs args = + parseCreateArgs(reader); + + return new ParsedInstruction( + TxKind.CREATE_USER_PDA, + true, + accountAt( + accounts, + 1 + ), + args.login(), + null, + new UserPdaMutation( + true, + accountAt( + accounts, + 1 + ), + args.login(), + args.recoveryKey(), + args.rootKey(), + args.createdAtMs(), + args.createdAtMs(), + 0, + ZERO_HASH.clone(), + args.additionalLimit(), + args.fields(), + args.recordSignature() + ) + ); + } + + private static ParsedInstruction parseUpdateInstruction( + Reader reader, + List accounts + ) { + + UpdateArgs args = + parseUpdateArgs(reader); + + return new ParsedInstruction( + TxKind.UPDATE_USER_PDA, + true, + accountAt( + accounts, + 1 + ), + args.login(), + null, + new UserPdaMutation( + false, + accountAt( + accounts, + 1 + ), + args.login(), + args.recoveryKey(), + args.rootKey(), + args.createdAtMs(), + args.updatedAtMs(), + args.version(), + args.prevHash(), + args.additionalLimit(), + args.fields(), + args.recordSignature() + ) + ); + } + + private static CreateOrUpdateArgs parseCreateArgs( + Reader reader + ) { + + String login = + reader.readStringU8(); + + String recoveryKey = + Base58Util.encode( + reader.readFixed(32) + ); + + String rootKey = + Base58Util.encode( + reader.readFixed(32) + ); + + long createdAtMs = + reader.readU64(); + + long additionalLimit = + reader.readU64(); + + UserFields fields = + parseFields(reader); + + String recordSignature = + Base58Util.encode( + reader.readFixed(64) + ); + + return new CreateOrUpdateArgs( + login, + recoveryKey, + rootKey, + createdAtMs, + additionalLimit, + fields, + recordSignature + ); + } + + private static UpdateArgs parseUpdateArgs( + Reader reader + ) { + + String login = + reader.readStringU8(); + + String recoveryKey = + Base58Util.encode( + reader.readFixed(32) + ); + + String rootKey = + Base58Util.encode( + reader.readFixed(32) + ); + + long createdAtMs = + reader.readU64(); + + long updatedAtMs = + reader.readU64(); + + int version = + (int) reader.readU32(); + + byte[] prevHash = + reader.readFixed(32); + + long additionalLimit = + reader.readU64(); + + UserFields fields = + parseFields(reader); + + String recordSignature = + Base58Util.encode( + reader.readFixed(64) + ); + + return new UpdateArgs( + login, + recoveryKey, + rootKey, + createdAtMs, + updatedAtMs, + version, + prevHash, + additionalLimit, + fields, + recordSignature + ); + } + + private static UserFields parseFields( + Reader reader + ) { + + String clientKey = + Base58Util.encode( + reader.readFixed(32) + ); + + String blockchainKey = + Base58Util.encode( + reader.readFixed(32) + ); + + String blockchainName = + reader.readStringU8(); + + long usedBytes = + reader.readU64(); + + int lastBlockNumber = + (int) reader.readU32(); + + String lastBlockHash = + toHex( + reader.readFixed(32) + ); + + String lastBlockSignature = + Base58Util.encode( + reader.readFixed(64) + ); + + String arweaveTxId = + reader.readStringU8(); + + boolean isServer = + reader.readU8() == 1; + + int addressFormatType = + 0; + int addressFormatVersion = + 0; + String serverAddress = + ""; + List syncServers = + new ArrayList<>(); + + if (isServer) { + addressFormatType = + reader.readU8(); + addressFormatVersion = + reader.readU8(); + serverAddress = + reader.readStringU8(); + int syncCount = + reader.readU8(); + for (int i = 0; i < syncCount; i++) { + syncServers.add( + reader.readStringU8() + ); + } + } + + int accessCount = + reader.readU8(); + + List accessServers = + new ArrayList<>(); + + for (int i = 0; i < accessCount; i++) { + accessServers.add( + reader.readStringU8() + ); + } + + int sessionsMode = + reader.readU8(); + + int sessionsCount = + reader.readU8(); + + List sessions = + new ArrayList<>(); + + for (int i = 0; i < sessionsCount; i++) { + sessions.add( + new UserSessionSnapshot( + reader.readU8(), + reader.readU8(), + reader.readStringU8(), + Base58Util.encode( + reader.readFixed(32) + ) + ) + ); + } + + int trustedCount = + reader.readU8(); + + return new UserFields( + clientKey, + blockchainKey, + blockchainName, + usedBytes, + lastBlockNumber, + lastBlockHash, + lastBlockSignature, + arweaveTxId, + isServer, + addressFormatType, + addressFormatVersion, + serverAddress, + List.copyOf(syncServers), + List.copyOf(accessServers), + sessionsMode, + List.copyOf(sessions), + trustedCount + ); + } + + private static UserPdaSnapshot buildSnapshot( + UserPdaMutation mutation, + int recordNumber, + long slot, + String txSignature, + long createdAtMs, + long updatedAtMs, + String prevRecordHash, + long paidLimitBytes + ) { + + UserFields fields = + mutation.fields(); + + UserPdaSnapshot snapshot = + new UserPdaSnapshot( + mutation.pdaAddress(), + mutation.login(), + recordNumber, + slot, + txSignature, + mutation.recoveryKey(), + mutation.rootKey(), + fields.clientKey(), + fields.blockchainName(), + fields.blockchainKey(), + paidLimitBytes, + fields.usedBytes(), + fields.lastBlockNumber(), + fields.lastBlockHash(), + fields.lastBlockSignature(), + fields.arweaveTxId(), + fields.isServer(), + fields.addressFormatType(), + fields.addressFormatVersion(), + fields.serverAddress(), + fields.syncServers(), + fields.accessServers(), + fields.sessionsMode(), + fields.sessions(), + fields.trustedCount(), + createdAtMs, + updatedAtMs, + prevRecordHash, + mutation.recordSignature(), + "" + ); + + return snapshot.withRawDataBase64( + serializeSnapshotToBase64( + snapshot + ) + ); + } + + private static List readInstructionAccounts( + JsonNode accountsNode + ) { + + List accounts = + new ArrayList<>(); + + if (!accountsNode.isArray()) { + return accounts; + } + + for (JsonNode node : accountsNode) { + accounts.add( + node.asText("") + ); + } + + return accounts; + } + + private static String accountAt( + List accounts, + int index + ) { + + if (index < 0 + || index >= accounts.size()) { + return null; + } + + String value = + accounts.get(index); + + return value == null + || value.isBlank() + ? null + : value; + } + + private static String toHex( + byte[] bytes + ) { + return HexFormat.of() + .formatHex(bytes); + } + + private static byte[] fromHex( + String hex + ) { + if (hex == null + || hex.isBlank()) { + return new byte[0]; + } + + return HexFormat.of() + .parseHex(hex); + } + + private static int u16le( + byte[] data, + int offset + ) { + return (data[offset] & 0xFF) + | ((data[offset + 1] & 0xFF) << 8); + } + + private static void pushString( + List out, + String value + ) { + pushStringU8( + out, + value.getBytes(StandardCharsets.UTF_8) + ); + } + + private static void pushStringU8( + List out, + byte[] bytes + ) { + out.add((byte) bytes.length); + pushFixed( + out, + bytes + ); + } + + private static void pushFixed( + List out, + byte[] bytes + ) { + for (byte value : bytes) { + out.add(value); + } + } + + private static void pushU16( + List out, + int value + ) { + out.add((byte) (value & 0xFF)); + out.add((byte) ((value >> 8) & 0xFF)); + } + + private static void pushU32( + List out, + int value + ) { + out.add((byte) (value & 0xFF)); + out.add((byte) ((value >> 8) & 0xFF)); + out.add((byte) ((value >> 16) & 0xFF)); + out.add((byte) ((value >> 24) & 0xFF)); + } + + private static void pushU64( + List out, + long value + ) { + out.add((byte) (value & 0xFF)); + out.add((byte) ((value >> 8) & 0xFF)); + out.add((byte) ((value >> 16) & 0xFF)); + out.add((byte) ((value >> 24) & 0xFF)); + out.add((byte) ((value >> 32) & 0xFF)); + out.add((byte) ((value >> 40) & 0xFF)); + out.add((byte) ((value >> 48) & 0xFF)); + out.add((byte) ((value >> 56) & 0xFF)); + } + + private static final class Reader { + + private final byte[] data; + private int cursor; + + private Reader( + byte[] data + ) { + this.data = data; + this.cursor = 0; + } + + private int readU8() { + return data[cursor++] & 0xFF; + } + + private long readU32() { + long value = + (data[cursor] & 0xFFL) + | ((data[cursor + 1] & 0xFFL) << 8) + | ((data[cursor + 2] & 0xFFL) << 16) + | ((data[cursor + 3] & 0xFFL) << 24); + + cursor += 4; + return value; + } + + private long readU64() { + long value = 0L; + for (int i = 0; i < 8; i++) { + value |= (long) (data[cursor + i] & 0xFF) << (8 * i); + } + cursor += 8; + return value; + } + + private byte[] readFixed( + int length + ) { + byte[] out = + new byte[length]; + System.arraycopy( + data, + cursor, + out, + 0, + length + ); + cursor += length; + return out; + } + + private String readStringU8() { + int length = + readU8(); + String value = + new String( + data, + cursor, + length, + StandardCharsets.UTF_8 + ); + cursor += length; + return value; + } + + private void skip( + int bytes + ) { + cursor += bytes; + } + } + + public enum TxKind { + INIT_USERS_ECONOMY_CONFIG, + UPDATE_USERS_ECONOMY_CONFIG, + CREATE_USER_PDA, + UPDATE_USER_PDA, + OTHER + } + + public record ParsedInstruction( + TxKind kind, + boolean relevant, + String affectedPdaAddress, + String affectedLogin, + EconomyConfigState economyConfigState, + UserPdaMutation userPdaMutation + ) { + } + + public record EconomyConfigState( + int version, + long registrationFeeLamports, + long lamportsPerLimitStep, + long startBonusLimit + ) { + public static EconomyConfigState initial() { + return new EconomyConfigState( + 1, + START_REGISTRATION_FEE_LAMPORTS, + START_LAMPORTS_PER_LIMIT_STEP, + START_BONUS_LIMIT + ); + } + } + + public record UserSessionSnapshot( + int sessionType, + int sessionVersion, + String sessionName, + String sessionPubKey + ) { + } + + public record UserFields( + String clientKey, + String blockchainKey, + String blockchainName, + long usedBytes, + int lastBlockNumber, + String lastBlockHash, + String lastBlockSignature, + String arweaveTxId, + boolean isServer, + int addressFormatType, + int addressFormatVersion, + String serverAddress, + List syncServers, + List accessServers, + int sessionsMode, + List sessions, + int trustedCount + ) { + } + + private record CreateOrUpdateArgs( + String login, + String recoveryKey, + String rootKey, + long createdAtMs, + long additionalLimit, + UserFields fields, + String recordSignature + ) { + } + + private record UpdateArgs( + String login, + String recoveryKey, + String rootKey, + long createdAtMs, + long updatedAtMs, + int version, + byte[] prevHash, + long additionalLimit, + UserFields fields, + String recordSignature + ) { + } + + public record UserPdaMutation( + boolean create, + String pdaAddress, + String login, + String recoveryKey, + String rootKey, + long createdAtMs, + long updatedAtMs, + int version, + byte[] prevHash, + long additionalLimit, + UserFields fields, + String recordSignature + ) { + } + + public record UserPdaSnapshot( + String pdaAddress, + String login, + int recordNumber, + long slot, + String lastTxSignature, + String recoveryKey, + String rootKey, + String clientKey, + String blockchainName, + String blockchainKey, + long paidLimitBytes, + long usedBytes, + int lastBlockNumber, + String lastBlockHash, + String lastBlockSignature, + String arweaveTxId, + boolean isServer, + int addressFormatType, + int addressFormatVersion, + String serverAddress, + List syncServers, + List accessServers, + int sessionsMode, + List sessions, + int trustedCount, + long createdAtMs, + long updatedAtMs, + String prevRecordHash, + String recordSignature, + String rawDataBase64 + ) { + public UserPdaSnapshot withRawDataBase64( + String value + ) { + return new UserPdaSnapshot( + pdaAddress, + login, + recordNumber, + slot, + lastTxSignature, + recoveryKey, + rootKey, + clientKey, + blockchainName, + blockchainKey, + paidLimitBytes, + usedBytes, + lastBlockNumber, + lastBlockHash, + lastBlockSignature, + arweaveTxId, + isServer, + addressFormatType, + addressFormatVersion, + serverAddress, + syncServers, + accessServers, + sessionsMode, + sessions, + trustedCount, + createdAtMs, + updatedAtMs, + prevRecordHash, + recordSignature, + value + ); + } + } +} diff --git a/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/config/AppConfig.java b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/config/AppConfig.java new file mode 100644 index 00000000..9932f9c5 --- /dev/null +++ b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/config/AppConfig.java @@ -0,0 +1,234 @@ +package sync.config; + +import java.time.Duration; + +public record AppConfig( + String rpcUrl, + String websocketUrl, + String programId, + String databaseUrl, + String databaseUser, + String databasePassword, + Duration pollInterval, + String commitment +) { + + public static final String DEFAULT_PROGRAM_ID = + "SHiNEPr1APdAgNBteUyBXcNovaHctpSjUu8oH2ZJdN6"; + + public static final String FIXED_COMMITMENT = + "confirmed"; + + public static final String ENABLED_KEY = + "solana.users.sync.enabled"; + public static final String RPC_URL_KEY = + "solana.users.sync.rpcUrl"; + public static final String WEBSOCKET_URL_KEY = + "solana.users.sync.wsUrl"; + public static final String PROGRAM_ID_KEY = + "solana.users.sync.programId"; + public static final String DATABASE_URL_KEY = + "solana.users.sync.databaseUrl"; + public static final String DATABASE_USER_KEY = + "solana.users.sync.dbUser"; + public static final String DATABASE_PASSWORD_KEY = + "solana.users.sync.dbPassword"; + public static final String POLL_INTERVAL_KEY = + "solana.users.sync.pollIntervalSeconds"; + public static final String LEGACY_SOLANA_RPC_URL_KEY = + "solana.rpcUrl"; + + public static boolean isEnabled( + utils.config.AppConfig serverConfig + ) { + String value = + trimToNull( + serverConfig.getParam(ENABLED_KEY) + ); + + if (value == null) { + return false; + } + + return Boolean.parseBoolean(value); + } + + public static AppConfig fromServerConfig( + utils.config.AppConfig serverConfig + ) { + + String rpcUrl = + firstRequired( + serverConfig, + "Solana users sync RPC URL", + RPC_URL_KEY, + LEGACY_SOLANA_RPC_URL_KEY + ); + + String websocketUrl = + requireParam( + serverConfig, + WEBSOCKET_URL_KEY + ); + + String programId = + optionalParam( + serverConfig, + PROGRAM_ID_KEY + ); + + if (programId == null) { + programId = + DEFAULT_PROGRAM_ID; + } + + String databaseUrl = + requireParam( + serverConfig, + DATABASE_URL_KEY + ); + + String databaseUser = + requireParam( + serverConfig, + DATABASE_USER_KEY + ); + + String databasePassword = + requireParam( + serverConfig, + DATABASE_PASSWORD_KEY + ); + + long pollIntervalSeconds = + parsePositiveLong( + optionalParam( + serverConfig, + POLL_INTERVAL_KEY + ), + 300L, + POLL_INTERVAL_KEY + ); + + return new AppConfig( + rpcUrl, + websocketUrl, + programId, + databaseUrl, + databaseUser, + databasePassword, + Duration.ofSeconds( + pollIntervalSeconds + ), + FIXED_COMMITMENT + ); + } + + private static long parsePositiveLong( + String rawValue, + long defaultValue, + String envName + ) { + + if (rawValue == null) { + return defaultValue; + } + + try { + + long value = + Long.parseLong( + rawValue + ); + + if (value <= 0L) { + throw new IllegalArgumentException( + envName + " must be > 0" + ); + } + + return value; + + } catch (NumberFormatException exception) { + + throw new IllegalArgumentException( + envName + " must be a positive integer", + exception + ); + } + } + + private static String firstRequired( + utils.config.AppConfig serverConfig, + String humanName, + String... names + ) { + + for (String name : names) { + String value = + optionalParam( + serverConfig, + name + ); + + if (value != null) { + return value; + } + } + + throw new IllegalStateException( + "Missing required server config: " + + humanName + ); + } + + private static String requireParam( + utils.config.AppConfig serverConfig, + String name + ) { + + String value = + optionalParam( + serverConfig, + name + ); + + if (value == null) { + throw new IllegalStateException( + "Missing required server config: " + + name + ); + } + + return value; + } + + private static String optionalParam( + utils.config.AppConfig serverConfig, + String name + ) { + + String value = + serverConfig.getParam(name); + + return trimToNull(value); + } + + private static String trimToNull( + String value + ) { + + if (value == null) { + return null; + } + + String trimmed = + value.trim(); + + if (trimmed.isEmpty()) { + return null; + } + + return trimmed; + } +} diff --git a/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/model/ProgramAccountUpdate.java b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/model/ProgramAccountUpdate.java new file mode 100644 index 00000000..c86a6da3 --- /dev/null +++ b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/model/ProgramAccountUpdate.java @@ -0,0 +1,12 @@ +package sync.model; + +public record ProgramAccountUpdate( + String address, + String owner, + long lamports, + long slot, + String dataBase64, + boolean executable, + Long rentEpoch +) { +} diff --git a/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/model/SnapshotResult.java b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/model/SnapshotResult.java new file mode 100644 index 00000000..62f12d22 --- /dev/null +++ b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/model/SnapshotResult.java @@ -0,0 +1,9 @@ +package sync.model; + +import java.util.List; + +public record SnapshotResult( + long snapshotSlot, + List accounts +) { +} diff --git a/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/service/SolanaUsersSyncService.java b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/service/SolanaUsersSyncService.java new file mode 100644 index 00000000..5ef0b1c0 --- /dev/null +++ b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/service/SolanaUsersSyncService.java @@ -0,0 +1,822 @@ +package sync.service; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import sync.codec.ShineUsersCodec; +import sync.config.AppConfig; +import sync.model.ProgramAccountUpdate; +import sync.model.SnapshotResult; +import sync.source.AccountUpdateListener; +import sync.source.ConnectionListener; +import sync.source.rpc.SolanaRpcClient; +import sync.source.rpc.SolanaWebSocketClient; +import sync.storage.postgres.PostgresStorageRepository; +import sync.util.SolanaPdaUtil; + +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicBoolean; + +public final class SolanaUsersSyncService + implements AutoCloseable { + + private static final int FULL_SNAPSHOT_PROGRESS_STEP = + 100; + + private static final Logger log = + LoggerFactory.getLogger( + SolanaUsersSyncService.class + ); + + private final AppConfig config; + private final ObjectMapper mapper; + private final PostgresStorageRepository storage; + private final SolanaRpcClient rpcClient; + private final SolanaWebSocketClient webSocketClient; + private final ExecutorService syncExecutor; + private final ScheduledExecutorService pollScheduler; + private final CompletableFuture readyFuture = + new CompletableFuture<>(); + private final AtomicBoolean closed = + new AtomicBoolean(false); + private final AtomicBoolean syncRequested = + new AtomicBoolean(false); + private final AtomicBoolean syncWorkerScheduled = + new AtomicBoolean(false); + private final String economyConfigPda; + + private volatile boolean initialSyncCompleted = + false; + + public SolanaUsersSyncService( + AppConfig config + ) throws Exception { + + this.config = + config; + + this.mapper = + new ObjectMapper(); + + this.storage = + new PostgresStorageRepository( + config.databaseUrl(), + config.databaseUser(), + config.databasePassword(), + mapper + ); + + this.rpcClient = + new SolanaRpcClient( + config.rpcUrl(), + config.programId(), + config.commitment() + ); + + this.webSocketClient = + new SolanaWebSocketClient( + config.websocketUrl(), + config.programId(), + config.commitment() + ); + + this.syncExecutor = + Executors.newSingleThreadExecutor( + runnable -> { + Thread thread = + new Thread( + runnable, + "solana-users-sync-worker" + ); + thread.setDaemon(true); + return thread; + } + ); + + this.pollScheduler = + Executors.newSingleThreadScheduledExecutor( + runnable -> { + Thread thread = + new Thread( + runnable, + "solana-users-sync-periodic" + ); + thread.setDaemon(true); + return thread; + } + ); + + this.economyConfigPda = + SolanaPdaUtil.findProgramAddress( + List.of( + "shine_users_economy_config" + .getBytes(StandardCharsets.UTF_8) + ), + config.programId() + ); + } + + public void start() + throws Exception { + + log.info( + "Starting sync service. programId={} economyConfigPda={} pollInterval={}", + config.programId(), + economyConfigPda, + config.pollInterval() + ); + + storage.updateLifecycleState( + "STARTING", + false, + null, + null, + null + ); + + webSocketClient.start( + new AccountUpdateListener() { + @Override + public void onAccountUpdate( + ProgramAccountUpdate update + ) { + log.debug( + "Realtime notification received. address={} slot={}", + update.address(), + update.slot() + ); + requestSync( + "realtime" + ); + } + }, + new ConnectionListener() { + @Override + public void onConnected( + boolean firstConnection + ) { + log.info( + "Solana websocket connected. firstConnection={}", + firstConnection + ); + requestSync( + firstConnection + ? "initial-connect" + : "reconnect" + ); + } + + @Override + public void onDisconnected( + Throwable cause + ) { + log.warn( + "Solana websocket disconnected: {}", + cause == null + ? "unknown" + : cause.getMessage() + ); + } + } + ); + + long periodSeconds = + config.pollInterval() + .getSeconds(); + + pollScheduler.scheduleWithFixedDelay( + () -> requestSync( + "periodic" + ), + periodSeconds, + periodSeconds, + TimeUnit.SECONDS + ); + } + + public void awaitReady() + throws Exception { + readyFuture.get(); + } + + public boolean isReady() { + return readyFuture.isDone() + && !readyFuture.isCompletedExceptionally(); + } + + private void requestSync( + String reason + ) { + + if (closed.get()) { + return; + } + + syncRequested.set(true); + + if (!syncWorkerScheduled.compareAndSet( + false, + true + )) { + return; + } + + syncExecutor.submit( + () -> runSyncLoop(reason) + ); + } + + private void runSyncLoop( + String firstReason + ) { + + String reason = + firstReason; + + try { + + while (!closed.get()) { + + boolean shouldRun = + syncRequested.getAndSet(false); + + if (!shouldRun) { + return; + } + + performSync(reason); + reason = "coalesced"; + } + + } catch (Exception exception) { + + log.error( + "Sync loop failed", + exception + ); + + try { + storage.updateLifecycleState( + "FAILED", + false, + exception.getMessage(), + System.currentTimeMillis(), + null + ); + } catch (Exception storageException) { + log.error( + "Failed to persist sync failure state", + storageException + ); + } + + readyFuture.completeExceptionally( + exception + ); + + } finally { + + syncWorkerScheduled.set(false); + + if (syncRequested.get() + && !closed.get() + && syncWorkerScheduled.compareAndSet( + false, + true + )) { + syncExecutor.submit( + () -> runSyncLoop("rescheduled") + ); + } + } + } + + private void performSync( + String reason + ) throws Exception { + + long nowMs = + System.currentTimeMillis(); + + PostgresStorageRepository.SyncStateSnapshot state = + storage.loadState(); + + log.info( + "Starting history sync. reason={} lastSeenSignature={}", + reason, + state.lastSeenSignature() + ); + + storage.updateLifecycleState( + initialSyncCompleted + ? "SYNCING" + : "BOOTSTRAPPING", + false, + null, + nowMs, + null + ); + + SolanaRpcClient.SignatureFetchResult fetchResult = + rpcClient.getSignaturesForAddressSince( + economyConfigPda, + state.lastSeenSignature() + ); + + if (state.lastSeenSignature() != null + && !fetchResult.anchorFound()) { + + log.error( + "History anchor signature not found anymore: {}. Running current-state full snapshot fallback.", + state.lastSeenSignature() + ); + + runFullSnapshotFallback( + state, + fetchResult, + nowMs + ); + + markReadyAfterSync( + state, + nowMs + ); + + return; + } + + if (fetchResult.signatures().isEmpty()) { + + PostgresStorageRepository.SyncStateSnapshot newState = + new PostgresStorageRepository.SyncStateSnapshot( + "READY", + true, + nowMs, + nowMs, + state.lastSeenSignature(), + state.lastSeenSlot(), + state.lastRelevantSignature(), + state.lastRelevantSlot(), + null, + state.economyConfigState(), + nowMs + ); + + storage.applyHistoryBatch( + List.of(), + List.of(), + newState + ); + + log.info( + "History sync completed with no new transactions." + ); + + markReadyAfterSync( + newState, + nowMs + ); + + return; + } + + List chronologicalSignatures = + new ArrayList<>( + fetchResult.signatures() + ); + + Collections.reverse( + chronologicalSignatures + ); + + List envelopes = + new ArrayList<>(); + + Set updatePdaAddresses = + new LinkedHashSet<>(); + + for (SolanaRpcClient.SignatureRecord signatureRecord : chronologicalSignatures) { + + JsonNode transaction = + rpcClient.getTransactionJsonParsed( + signatureRecord.signature() + ); + + ParsedTxEnvelope envelope = + parseTransactionEnvelope( + signatureRecord, + transaction + ); + + envelopes.add( + envelope + ); + + if (envelope.parsedInstruction() != null + && envelope.parsedInstruction().kind() == ShineUsersCodec.TxKind.UPDATE_USER_PDA + && envelope.parsedInstruction().affectedPdaAddress() != null) { + updatePdaAddresses.add( + envelope.parsedInstruction() + .affectedPdaAddress() + ); + } + } + + Map currentSnapshots = + storage.getCurrentSnapshots( + updatePdaAddresses + ); + + List txEntries = + new ArrayList<>(); + + List snapshotsToPersist = + new ArrayList<>(); + + ShineUsersCodec.EconomyConfigState economyState = + state.economyConfigState(); + + String lastRelevantSignature = + state.lastRelevantSignature(); + + Long lastRelevantSlot = + state.lastRelevantSlot(); + + for (ParsedTxEnvelope envelope : envelopes) { + + ShineUsersCodec.ParsedInstruction parsedInstruction = + envelope.parsedInstruction(); + + String txKind = + parsedInstruction == null + ? "failed_or_unavailable" + : parsedInstruction.kind().name(); + + boolean relevant = + false; + + String affectedPdaAddress = + null; + + String affectedLogin = + null; + + if (parsedInstruction != null) { + + if (parsedInstruction.kind() == ShineUsersCodec.TxKind.INIT_USERS_ECONOMY_CONFIG + || parsedInstruction.kind() == ShineUsersCodec.TxKind.UPDATE_USERS_ECONOMY_CONFIG) { + economyState = + parsedInstruction.economyConfigState(); + } + + if (parsedInstruction.relevant() + && parsedInstruction.userPdaMutation() != null) { + + relevant = true; + affectedPdaAddress = parsedInstruction.affectedPdaAddress(); + affectedLogin = parsedInstruction.affectedLogin(); + + ShineUsersCodec.UserPdaSnapshot snapshot; + + if (parsedInstruction.kind() == ShineUsersCodec.TxKind.CREATE_USER_PDA) { + + if (economyState == null) { + economyState = + ShineUsersCodec.EconomyConfigState.initial(); + log.warn( + "Economy config state was absent while processing create tx {}. Falling back to initial constants.", + envelope.signatureRecord().signature() + ); + } + + snapshot = + ShineUsersCodec.buildCreateSnapshot( + parsedInstruction.userPdaMutation(), + economyState, + envelope.signatureRecord().signature(), + envelope.signatureRecord().slot() + ); + + } else { + + ShineUsersCodec.UserPdaSnapshot previous = + currentSnapshots.get( + affectedPdaAddress + ); + + if (previous == null) { + throw new IllegalStateException( + "Missing previous snapshot for update PDA " + + affectedPdaAddress + ); + } + + snapshot = + ShineUsersCodec.buildUpdateSnapshot( + parsedInstruction.userPdaMutation(), + previous, + envelope.signatureRecord().signature(), + envelope.signatureRecord().slot() + ); + } + + currentSnapshots.put( + snapshot.pdaAddress(), + snapshot + ); + + snapshotsToPersist.add( + snapshot + ); + + lastRelevantSignature = + envelope.signatureRecord().signature(); + + lastRelevantSlot = + envelope.signatureRecord().slot(); + } + } + + txEntries.add( + new PostgresStorageRepository.TxHistoryEntry( + envelope.signatureRecord().signature(), + envelope.signatureRecord().slot(), + envelope.signatureRecord().blockTime(), + txKind, + relevant, + affectedPdaAddress, + affectedLogin, + envelope.rawTransactionJson(), + nowMs + ) + ); + } + + SolanaRpcClient.SignatureRecord newestSeen = + fetchResult.signatures() + .get(0); + + PostgresStorageRepository.SyncStateSnapshot newState = + new PostgresStorageRepository.SyncStateSnapshot( + "READY", + true, + nowMs, + nowMs, + newestSeen.signature(), + newestSeen.slot(), + lastRelevantSignature, + lastRelevantSlot, + null, + economyState, + nowMs + ); + + storage.applyHistoryBatch( + txEntries, + snapshotsToPersist, + newState + ); + + log.info( + "History sync completed. txCount={} relevantCount={} latestSignature={}", + txEntries.size(), + snapshotsToPersist.size(), + newestSeen.signature() + ); + + markReadyAfterSync( + newState, + nowMs + ); + } + + private void runFullSnapshotFallback( + PostgresStorageRepository.SyncStateSnapshot state, + SolanaRpcClient.SignatureFetchResult fetchResult, + long nowMs + ) throws Exception { + + log.warn( + "Starting full snapshot fallback because incremental history anchor is unavailable." + ); + + SnapshotResult snapshotResult = + rpcClient.loadFullSnapshot(); + + log.info( + "Full snapshot downloaded. snapshotSlot={} rawAccounts={}", + snapshotResult.snapshotSlot(), + snapshotResult.accounts().size() + ); + + List currentSnapshots = + new ArrayList<>(); + + int processedAccounts = + 0; + + for (ProgramAccountUpdate account : snapshotResult.accounts()) { + + processedAccounts++; + + try { + currentSnapshots.add( + ShineUsersCodec.parseUserPdaAccount( + account.address(), + account.slot(), + account.dataBase64(), + state.lastSeenSignature() + ) + ); + } catch (Exception ignored) { + } + + if (processedAccounts == 1 + || processedAccounts % FULL_SNAPSHOT_PROGRESS_STEP == 0 + || processedAccounts == snapshotResult.accounts().size()) { + log.info( + "Full snapshot parse progress: {}/{} accounts, {} user PDA snapshots accepted.", + processedAccounts, + snapshotResult.accounts().size(), + currentSnapshots.size() + ); + } + } + + String newestSignature = + fetchResult.signatures().isEmpty() + ? state.lastSeenSignature() + : fetchResult.signatures().get(0).signature(); + + Long newestSlot = + fetchResult.signatures().isEmpty() + ? state.lastSeenSlot() + : fetchResult.signatures().get(0).slot(); + + PostgresStorageRepository.SyncStateSnapshot newState = + new PostgresStorageRepository.SyncStateSnapshot( + "READY", + true, + nowMs, + nowMs, + newestSignature, + newestSlot, + state.lastRelevantSignature(), + state.lastRelevantSlot(), + "history_anchor_missing_full_snapshot_fallback", + state.economyConfigState(), + nowMs + ); + + storage.replaceCurrentFromFullSnapshot( + currentSnapshots, + newState + ); + + log.warn( + "Full snapshot fallback completed. currentSnapshots={} newestSignature={} newestSlot={}", + currentSnapshots.size(), + newestSignature, + newestSlot + ); + } + + private void markReadyAfterSync( + PostgresStorageRepository.SyncStateSnapshot state, + long nowMs + ) throws Exception { + + initialSyncCompleted = true; + + if (!readyFuture.isDone()) { + readyFuture.complete(null); + log.info( + "Sync service entered READY state." + ); + } + + storage.updateLifecycleState( + "READY", + true, + state.lastError(), + nowMs, + nowMs + ); + } + + private ParsedTxEnvelope parseTransactionEnvelope( + SolanaRpcClient.SignatureRecord signatureRecord, + JsonNode transaction + ) { + + if (signatureRecord.failed()) { + return new ParsedTxEnvelope( + signatureRecord, + new ShineUsersCodec.ParsedInstruction( + ShineUsersCodec.TxKind.OTHER, + false, + null, + null, + null, + null + ), + transaction == null + ? "{\"failed\":true,\"error\":" + + String.valueOf(signatureRecord.errorJson()) + "}" + : transaction.toString() + ); + } + + if (transaction == null + || transaction.isNull()) { + return new ParsedTxEnvelope( + signatureRecord, + null, + "{\"transaction\":null}" + ); + } + + JsonNode instructions = + transaction.path("transaction") + .path("message") + .path("instructions"); + + if (instructions.isArray()) { + for (JsonNode instruction : instructions) { + ShineUsersCodec.ParsedInstruction parsedInstruction = + ShineUsersCodec.parseShineUsersInstruction( + instruction, + config.programId() + ); + + if (parsedInstruction != null) { + return new ParsedTxEnvelope( + signatureRecord, + parsedInstruction, + transaction.toString() + ); + } + } + } + + return new ParsedTxEnvelope( + signatureRecord, + new ShineUsersCodec.ParsedInstruction( + ShineUsersCodec.TxKind.OTHER, + false, + null, + null, + null, + null + ), + transaction.toString() + ); + } + + @Override + public void close() { + + if (!closed.compareAndSet( + false, + true + )) { + return; + } + + try { + pollScheduler.shutdownNow(); + } catch (Exception ignored) { + } + + try { + webSocketClient.close(); + } catch (Exception ignored) { + } + + try { + syncExecutor.shutdownNow(); + } catch (Exception ignored) { + } + + try { + rpcClient.close(); + } catch (Exception ignored) { + } + + try { + storage.close(); + } catch (Exception ignored) { + } + } + + private record ParsedTxEnvelope( + SolanaRpcClient.SignatureRecord signatureRecord, + ShineUsersCodec.ParsedInstruction parsedInstruction, + String rawTransactionJson + ) { + } +} diff --git a/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/source/AccountUpdateListener.java b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/source/AccountUpdateListener.java new file mode 100644 index 00000000..619ccb55 --- /dev/null +++ b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/source/AccountUpdateListener.java @@ -0,0 +1,11 @@ +package sync.source; + +import sync.model.ProgramAccountUpdate; + +@FunctionalInterface +public interface AccountUpdateListener { + + void onAccountUpdate( + ProgramAccountUpdate update + ); +} diff --git a/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/source/ConnectionListener.java b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/source/ConnectionListener.java new file mode 100644 index 00000000..4c8badce --- /dev/null +++ b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/source/ConnectionListener.java @@ -0,0 +1,12 @@ +package sync.source; + +public interface ConnectionListener { + + void onConnected( + boolean firstConnection + ); + + void onDisconnected( + Throwable cause + ); +} diff --git a/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/source/rpc/SolanaRpcClient.java b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/source/rpc/SolanaRpcClient.java new file mode 100644 index 00000000..bde598d8 --- /dev/null +++ b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/source/rpc/SolanaRpcClient.java @@ -0,0 +1,737 @@ +package sync.source.rpc; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import okhttp3.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import sync.model.ProgramAccountUpdate; +import sync.model.SnapshotResult; + +import java.io.IOException; +import java.time.Duration; +import java.util.*; + +public final class SolanaRpcClient + implements AutoCloseable { + + private static final Logger log = + LoggerFactory.getLogger( + SolanaRpcClient.class + ); + + private static final MediaType JSON = + MediaType.get("application/json"); + + private static final int SIGNATURE_PAGE_SIZE = + 1000; + + private static final int ACCOUNT_BATCH_SIZE = + 100; + + private final String rpcUrl; + private final String programId; + private final String commitment; + private final ObjectMapper mapper = + new ObjectMapper(); + private final OkHttpClient httpClient = + new OkHttpClient.Builder() + .callTimeout(Duration.ofMinutes(2)) + .build(); + + public SolanaRpcClient( + String rpcUrl, + String programId, + String commitment + ) { + this.rpcUrl = rpcUrl; + this.programId = programId; + this.commitment = commitment; + } + + public SnapshotResult loadFullSnapshot() + throws IOException { + + log.info( + "Requesting full snapshot via getProgramAccounts. programId={}", + programId + ); + + Map payload = + Map.of( + "jsonrpc", "2.0", + "id", 100, + "method", "getProgramAccounts", + "params", List.of( + programId, + Map.of( + "encoding", "base64", + "commitment", commitment, + "withContext", true + ) + ) + ); + + JsonNode root = + executeRpc(payload); + + log.info( + "Full snapshot RPC response received. Parsing account list..." + ); + + JsonNode result = + root.path("result"); + + long snapshotSlot = + result.path("context") + .path("slot") + .asLong(-1); + + if (snapshotSlot < 0) { + throw new IOException( + "Missing context.slot in getProgramAccounts" + ); + } + + JsonNode values = + result.path("value"); + + if (!values.isArray()) { + throw new IOException( + "Unexpected getProgramAccounts response" + ); + } + + List accounts = + new ArrayList<>(); + + for (JsonNode item : values) { + + JsonNode account = + item.path("account"); + + JsonNode data = + account.path("data"); + + if (!data.isArray() + || data.isEmpty()) { + continue; + } + + accounts.add( + parseAccount( + item.path("pubkey") + .asText(), + account, + snapshotSlot + ) + ); + } + + log.info( + "Full snapshot RPC parsed. accounts={} snapshotSlot={}", + accounts.size(), + snapshotSlot + ); + + return new SnapshotResult( + snapshotSlot, + accounts + ); + } + + public SignatureFetchResult getSignaturesForAddressSince( + String address, + String knownSignature + ) throws IOException { + + List signatures = + new ArrayList<>(); + + String before = + null; + + boolean anchorFound = + knownSignature == null; + + while (true) { + + Map options = + new LinkedHashMap<>(); + + options.put( + "limit", + SIGNATURE_PAGE_SIZE + ); + + options.put( + "commitment", + commitment + ); + + if (before != null) { + options.put( + "before", + before + ); + } + + Map payload = + Map.of( + "jsonrpc", "2.0", + "id", 103, + "method", "getSignaturesForAddress", + "params", List.of( + address, + options + ) + ); + + JsonNode root = + executeRpc(payload); + + JsonNode values = + root.path("result"); + + if (!values.isArray()) { + throw new IOException( + "Unexpected getSignaturesForAddress response" + ); + } + + if (values.isEmpty()) { + break; + } + + for (JsonNode item : values) { + + String signature = + item.path("signature") + .asText(""); + + if (signature.isBlank()) { + continue; + } + + if (knownSignature != null + && knownSignature.equals(signature)) { + anchorFound = true; + return new SignatureFetchResult( + signatures, + true + ); + } + + Long blockTime = + item.hasNonNull("blockTime") + ? item.get("blockTime").asLong() + : null; + + signatures.add( + new SignatureRecord( + signature, + item.path("slot") + .asLong(-1), + blockTime, + item.hasNonNull("err"), + item.path("err").toString() + ) + ); + } + + JsonNode last = + values.get(values.size() - 1); + + before = + last.path("signature") + .asText(""); + + if (before.isBlank() + || values.size() < SIGNATURE_PAGE_SIZE) { + break; + } + } + + return new SignatureFetchResult( + signatures, + anchorFound + ); + } + + public JsonNode getTransactionJsonParsed( + String signature + ) throws IOException { + + Map payload = + Map.of( + "jsonrpc", "2.0", + "id", 104, + "method", "getTransaction", + "params", List.of( + signature, + Map.of( + "encoding", + "jsonParsed", + "commitment", + commitment, + "maxSupportedTransactionVersion", + 0 + ) + ) + ); + + JsonNode root = + executeRpc(payload); + + JsonNode result = + root.get("result"); + + if (result == null + || result.isNull()) { + return null; + } + + return result; + } + + public long getCurrentSlot() + throws IOException { + + Map payload = + Map.of( + "jsonrpc", "2.0", + "id", 101, + "method", "getSlot", + "params", List.of( + Map.of( + "commitment", + commitment + ) + ) + ); + + JsonNode root = + executeRpc(payload); + + long slot = + root.path("result") + .asLong(-1); + + if (slot < 0) { + throw new IOException( + "Invalid getSlot response" + ); + } + + return slot; + } + + public AccountBatchResult getCurrentAccounts( + Collection addresses, + long recoverySlot + ) throws IOException { + + List updates = + new ArrayList<>(); + + List missingAddresses = + new ArrayList<>(); + + List addressList = + new ArrayList<>(addresses); + + for (int offset = 0; + offset < addressList.size(); + offset += ACCOUNT_BATCH_SIZE) { + + int end = + Math.min( + offset + ACCOUNT_BATCH_SIZE, + addressList.size() + ); + + List batch = + addressList.subList( + offset, + end + ); + + Map payload = + Map.of( + "jsonrpc", "2.0", + "id", 105, + "method", "getMultipleAccounts", + "params", List.of( + batch, + Map.of( + "encoding", + "base64", + "commitment", + commitment + ) + ) + ); + + JsonNode root = + executeRpc(payload); + + JsonNode values = + root.path("result") + .path("value"); + + if (!values.isArray()) { + throw new IOException( + "Unexpected getMultipleAccounts response" + ); + } + + if (values.size() != batch.size()) { + throw new IOException( + "Unexpected getMultipleAccounts account count" + ); + } + + for (int i = 0; i < batch.size(); i++) { + + String address = + batch.get(i); + + JsonNode account = + values.get(i); + + if (account == null + || account.isNull()) { + missingAddresses.add(address); + continue; + } + + String owner = + account.path("owner") + .asText(""); + + if (!programId.equals(owner)) { + continue; + } + + JsonNode data = + account.path("data"); + + if (!data.isArray() + || data.isEmpty()) { + throw new IOException( + "Unexpected account.data format for " + + address + ); + } + + updates.add( + parseAccount( + address, + account, + recoverySlot + ) + ); + } + } + + return new AccountBatchResult( + updates, + missingAddresses + ); + } + + public long getFirstAvailableBlock() + throws IOException { + + Map payload = + Map.of( + "jsonrpc", "2.0", + "id", 102, + "method", "getFirstAvailableBlock" + ); + + JsonNode root = + executeRpc(payload); + + return root.path("result") + .asLong(-1); + } + + public List getSignaturesAfterSlot( + long fromSlot, + long targetSlot + ) throws IOException { + + List signatures = + new ArrayList<>(); + + String before = + null; + + boolean lowerBoundaryReached = + false; + + while (!lowerBoundaryReached) { + + Map options = + new LinkedHashMap<>(); + + options.put("limit", SIGNATURE_PAGE_SIZE); + options.put("commitment", commitment); + + if (before != null) { + options.put("before", before); + } + + Map payload = + Map.of( + "jsonrpc", "2.0", + "id", 106, + "method", "getSignaturesForAddress", + "params", List.of( + programId, + options + ) + ); + + JsonNode root = + executeRpc(payload); + + JsonNode values = + root.path("result"); + + if (!values.isArray()) { + throw new IOException( + "Unexpected getSignaturesForAddress response" + ); + } + + if (values.isEmpty()) { + break; + } + + for (JsonNode item : values) { + + long slot = + item.path("slot") + .asLong(-1); + + if (slot < 0) { + continue; + } + + if (slot <= fromSlot) { + lowerBoundaryReached = true; + break; + } + + if (slot > targetSlot) { + continue; + } + + String signature = + item.path("signature") + .asText(""); + + if (signature.isBlank()) { + continue; + } + + JsonNode error = + item.get("err"); + + if (error != null + && !error.isNull()) { + continue; + } + + signatures.add( + new SignatureInfo( + signature, + slot + ) + ); + } + + JsonNode last = + values.get(values.size() - 1); + + before = + last.path("signature") + .asText(""); + + if (before.isBlank() + || values.size() < SIGNATURE_PAGE_SIZE) { + break; + } + } + + return signatures; + } + + public Set getTouchedAddresses( + List signatures + ) throws IOException { + + Set addresses = + new LinkedHashSet<>(); + + for (SignatureInfo signatureInfo : signatures) { + + JsonNode transaction = + getTransactionJsonParsed( + signatureInfo.signature() + ); + + if (transaction == null + || transaction.isNull()) { + throw new IOException( + "Transaction unavailable during recovery: " + + signatureInfo.signature() + ); + } + + JsonNode accountKeys = + transaction.path("transaction") + .path("message") + .path("accountKeys"); + + if (!accountKeys.isArray()) { + throw new IOException( + "Missing accountKeys for transaction: " + + signatureInfo.signature() + ); + } + + for (JsonNode keyNode : accountKeys) { + + String pubkey; + + if (keyNode.isTextual()) { + pubkey = keyNode.asText(); + } else { + pubkey = keyNode.path("pubkey") + .asText(""); + } + + if (!pubkey.isBlank()) { + addresses.add(pubkey); + } + } + } + + addresses.remove(programId); + return addresses; + } + + private ProgramAccountUpdate parseAccount( + String address, + JsonNode account, + long slot + ) { + + JsonNode data = + account.path("data"); + + return new ProgramAccountUpdate( + address, + account.path("owner") + .asText(), + account.path("lamports") + .asLong(), + slot, + data.get(0) + .asText(), + account.path("executable") + .asBoolean(false), + account.hasNonNull("rentEpoch") + ? account.get("rentEpoch").asLong() + : null + ); + } + + private JsonNode executeRpc( + Map payload + ) throws IOException { + + Request request = + new Request.Builder() + .url(rpcUrl) + .post( + RequestBody.create( + mapper.writeValueAsBytes(payload), + JSON + ) + ) + .build(); + + try (Response response = + httpClient.newCall(request).execute()) { + + if (!response.isSuccessful()) { + throw new IOException( + "Solana RPC HTTP error: " + response.code() + ); + } + + ResponseBody body = + response.body(); + + if (body == null) { + throw new IOException( + "Solana RPC returned empty body" + ); + } + + JsonNode root = + mapper.readTree( + body.string() + ); + + if (root.has("error")) { + throw new IOException( + "Solana RPC error: " + root.get("error") + ); + } + + return root; + } + } + + @Override + public void close() { + + httpClient.dispatcher() + .executorService() + .shutdown(); + httpClient.connectionPool() + .evictAll(); + } + + public record SignatureRecord( + String signature, + long slot, + Long blockTime, + boolean failed, + String errorJson + ) { + } + + public record SignatureFetchResult( + List signatures, + boolean anchorFound + ) { + } + + public record AccountBatchResult( + List updates, + List missingAddresses + ) { + } + + public record SignatureInfo( + String signature, + long slot + ) { + } +} diff --git a/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/source/rpc/SolanaWebSocketClient.java b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/source/rpc/SolanaWebSocketClient.java new file mode 100644 index 00000000..9320007e --- /dev/null +++ b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/source/rpc/SolanaWebSocketClient.java @@ -0,0 +1,542 @@ +package sync.source.rpc; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import okhttp3.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import sync.model.ProgramAccountUpdate; +import sync.source.AccountUpdateListener; +import sync.source.ConnectionListener; + +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +public final class SolanaWebSocketClient + extends WebSocketListener + implements AutoCloseable { + + private static final Logger log = + LoggerFactory.getLogger( + SolanaWebSocketClient.class + ); + + private final String websocketUrl; + private final String programId; + private final String commitment; + + private final ObjectMapper mapper = + new ObjectMapper(); + + private final OkHttpClient httpClient; + + private final ScheduledExecutorService scheduler; + + private final AtomicBoolean closed = + new AtomicBoolean(false); + + private final AtomicInteger reconnectAttempt = + new AtomicInteger(0); + + private final AtomicBoolean everSubscribed = + new AtomicBoolean(false); + + private volatile WebSocket webSocket; + + private volatile AccountUpdateListener + accountUpdateListener; + + private volatile ConnectionListener + connectionListener; + + public SolanaWebSocketClient( + String websocketUrl, + String programId, + String commitment + ) { + this.websocketUrl = websocketUrl; + this.programId = programId; + this.commitment = commitment; + + this.httpClient = + new OkHttpClient.Builder() + .readTimeout( + Duration.ZERO + ) + .pingInterval( + Duration.ofSeconds(20) + ) + .build(); + + this.scheduler = + Executors + .newSingleThreadScheduledExecutor( + runnable -> { + + Thread thread = + new Thread( + runnable, + "solana-rpc-reconnect" + ); + + thread.setDaemon( + true + ); + + return thread; + } + ); + } + + public void start( + AccountUpdateListener accountUpdateListener, + ConnectionListener connectionListener + ) { + + this.accountUpdateListener = + accountUpdateListener; + + this.connectionListener = + connectionListener; + + connect(); + } + + private void connect() { + + if (closed.get()) { + return; + } + + log.info( + "Connecting to Solana WebSocket: {}", + websocketUrl + ); + + Request request = + new Request.Builder() + .url( + websocketUrl + ) + .build(); + + this.webSocket = + httpClient.newWebSocket( + request, + this + ); + } + + @Override + public void onOpen( + WebSocket webSocket, + Response response + ) { + + reconnectAttempt.set( + 0 + ); + + log.info( + "WebSocket connected" + ); + + sendProgramSubscribe( + webSocket + ); + } + + private void sendProgramSubscribe( + WebSocket webSocket + ) { + + try { + + Map request = + Map.of( + "jsonrpc", + "2.0", + "id", + 1, + "method", + "programSubscribe", + "params", + List.of( + programId, + Map.of( + "encoding", + "base64", + "commitment", + commitment + ) + ) + ); + + String payload = + mapper.writeValueAsString( + request + ); + + if (!webSocket.send( + payload + )) { + throw new IllegalStateException( + "WebSocket rejected subscription request" + ); + } + + log.info( + "Subscription request sent for program: {}", + programId + ); + + } catch (Exception exception) { + + log.error( + "Failed to send subscription request", + exception + ); + + webSocket.cancel(); + } + } + + @Override + public void onMessage( + WebSocket webSocket, + String text + ) { + + try { + + JsonNode root = + mapper.readTree( + text + ); + + if ( + root.has("id") + && root.has("result") + && root.get("id").asInt() == 1 + ) { + + log.info( + "Subscribed successfully. subscriptionId={}", + root.get("result").asText() + ); + + boolean firstConnection = + everSubscribed + .compareAndSet( + false, + true + ); + + ConnectionListener listener = + connectionListener; + + if (listener != null) { + listener.onConnected( + firstConnection + ); + } + + return; + } + + if (root.has("error")) { + + log.error( + "Solana websocket RPC error: {}", + root.get("error") + ); + + return; + } + + if (!"programNotification" + .equals( + root + .path("method") + .asText() + )) { + return; + } + + ProgramAccountUpdate update = + parseProgramNotification( + root + ); + + log.debug( + "Account update. address={} slot={} base64Chars={}", + update.address(), + update.slot(), + update.dataBase64().length() + ); + + AccountUpdateListener listener = + accountUpdateListener; + + if (listener != null) { + listener.onAccountUpdate( + update + ); + } + + } catch (Exception exception) { + + log.error( + "Failed to process WebSocket message. raw={}", + text, + exception + ); + } + } + + private ProgramAccountUpdate parseProgramNotification( + JsonNode root + ) { + + JsonNode result = + root + .path("params") + .path("result"); + + JsonNode context = + result.path("context"); + + JsonNode value = + result.path("value"); + + JsonNode account = + value.path("account"); + + JsonNode data = + account.path("data"); + + if (!data.isArray() + || data.isEmpty()) { + throw new IllegalArgumentException( + "Unexpected account.data format" + ); + } + + return new ProgramAccountUpdate( + requiredText( + value, + "pubkey" + ), + requiredText( + account, + "owner" + ), + requiredLong( + account, + "lamports" + ), + requiredLong( + context, + "slot" + ), + data.get(0).asText(), + account.path( + "executable" + ).asBoolean(false), + account.hasNonNull( + "rentEpoch" + ) + ? account + .get("rentEpoch") + .asLong() + : null + ); + } + + private String requiredText( + JsonNode node, + String field + ) { + + JsonNode value = + node.get(field); + + if (value == null + || value.isNull() + || value.asText().isBlank()) { + throw new IllegalArgumentException( + "Missing field: " + + field + ); + } + + return value.asText(); + } + + private long requiredLong( + JsonNode node, + String field + ) { + + JsonNode value = + node.get(field); + + if (value == null + || !value.isNumber()) { + throw new IllegalArgumentException( + "Missing numeric field: " + + field + ); + } + + return value.asLong(); + } + + @Override + public void onClosed( + WebSocket webSocket, + int code, + String reason + ) { + + log.warn( + "WebSocket closed. code={} reason={}", + code, + reason + ); + + notifyDisconnected( + new IllegalStateException( + "WebSocket closed: " + + reason + ) + ); + + scheduleReconnect(); + } + + @Override + public void onFailure( + WebSocket webSocket, + Throwable throwable, + Response response + ) { + + log.error( + "WebSocket failure", + throwable + ); + + if (response != null) { + log.error( + "WebSocket HTTP status: {}", + response.code() + ); + } + + notifyDisconnected( + throwable + ); + + scheduleReconnect(); + } + + private void notifyDisconnected( + Throwable cause + ) { + + ConnectionListener listener = + connectionListener; + + if (listener != null) { + + try { + listener.onDisconnected( + cause + ); + } catch (Exception exception) { + log.error( + "Connection listener failed", + exception + ); + } + } + } + + private void scheduleReconnect() { + + if (closed.get()) { + return; + } + + int attempt = + reconnectAttempt + .incrementAndGet(); + + long delaySeconds = + switch ( + Math.min( + attempt, + 5 + ) + ) { + case 1 -> 1; + case 2 -> 2; + case 3 -> 5; + case 4 -> 10; + default -> 30; + }; + + log.warn( + "Reconnect scheduled in {} seconds (attempt {})", + delaySeconds, + attempt + ); + + scheduler.schedule( + this::connect, + delaySeconds, + TimeUnit.SECONDS + ); + } + + public void stop() { + close(); + } + + @Override + public void close() { + + if (!closed.compareAndSet( + false, + true + )) { + return; + } + + WebSocket socket = + this.webSocket; + + if (socket != null) { + socket.close( + 1000, + "Application shutdown" + ); + } + + scheduler.shutdownNow(); + + httpClient + .dispatcher() + .executorService() + .shutdown(); + + httpClient + .connectionPool() + .evictAll(); + } +} diff --git a/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/storage/postgres/PostgresStorageRepository.java b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/storage/postgres/PostgresStorageRepository.java new file mode 100644 index 00000000..52a3e43b --- /dev/null +++ b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/storage/postgres/PostgresStorageRepository.java @@ -0,0 +1,1076 @@ +package sync.storage.postgres; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import sync.codec.ShineUsersCodec; + +import java.sql.*; +import java.util.*; + +public final class PostgresStorageRepository + implements AutoCloseable { + + private static final Logger log = + LoggerFactory.getLogger( + PostgresStorageRepository.class + ); + + private final String databaseUrl; + private final String databaseUser; + private final String databasePassword; + private final ObjectMapper mapper; + + public PostgresStorageRepository( + String databaseUrl, + String databaseUser, + String databasePassword, + ObjectMapper mapper + ) throws Exception { + + this.databaseUrl = + databaseUrl; + this.databaseUser = + databaseUser; + this.databasePassword = + databasePassword; + this.mapper = + mapper; + + migrate(); + + log.info( + "PostgreSQL schema ready: {}", + databaseUrl + ); + } + + public synchronized SyncStateSnapshot loadState() + throws Exception { + + try (Connection connection = + openConnection(); + PreparedStatement statement = + connection.prepareStatement( + "SELECT " + + "status, ready, last_poll_at_ms, " + + "last_successful_poll_at_ms, " + + "last_seen_signature, last_seen_slot, " + + "last_relevant_signature, last_relevant_slot, " + + "last_error, economy_config_version, " + + "registration_fee_lamports, " + + "lamports_per_limit_step, " + + "start_bonus_limit, updated_at_ms " + + "FROM solana_sync_state WHERE id = 1" + ); + ResultSet resultSet = + statement.executeQuery()) { + + if (!resultSet.next()) { + return new SyncStateSnapshot( + "EMPTY", + false, + null, + null, + null, + null, + null, + null, + null, + null, + null + ); + } + + Integer economyVersion = + getNullableInt( + resultSet, + "economy_config_version" + ); + + ShineUsersCodec.EconomyConfigState economyConfigState = + null; + + if (economyVersion != null) { + economyConfigState = + new ShineUsersCodec.EconomyConfigState( + economyVersion, + resultSet.getLong( + "registration_fee_lamports" + ), + resultSet.getLong( + "lamports_per_limit_step" + ), + resultSet.getLong( + "start_bonus_limit" + ) + ); + } + + return new SyncStateSnapshot( + resultSet.getString("status"), + resultSet.getBoolean("ready"), + getNullableLong( + resultSet, + "last_poll_at_ms" + ), + getNullableLong( + resultSet, + "last_successful_poll_at_ms" + ), + resultSet.getString("last_seen_signature"), + getNullableLong( + resultSet, + "last_seen_slot" + ), + resultSet.getString("last_relevant_signature"), + getNullableLong( + resultSet, + "last_relevant_slot" + ), + resultSet.getString("last_error"), + economyConfigState, + getNullableLong( + resultSet, + "updated_at_ms" + ) + ); + } + } + + public synchronized boolean hasCurrentData() + throws Exception { + + try (Connection connection = + openConnection(); + PreparedStatement statement = + connection.prepareStatement( + "SELECT 1 FROM solana_user_pda_current LIMIT 1" + ); + ResultSet resultSet = + statement.executeQuery()) { + + return resultSet.next(); + } + } + + public synchronized Map getCurrentSnapshots( + Collection pdaAddresses + ) throws Exception { + + Map result = + new LinkedHashMap<>(); + + if (pdaAddresses == null + || pdaAddresses.isEmpty()) { + return result; + } + + String placeholders = + String.join( + ",", + Collections.nCopies( + pdaAddresses.size(), + "?" + ) + ); + + String sql = + "SELECT * FROM solana_user_pda_current " + + "WHERE pda_address IN (" + placeholders + ")"; + + try (Connection connection = + openConnection(); + PreparedStatement statement = + connection.prepareStatement(sql)) { + + int index = 1; + + for (String address : pdaAddresses) { + statement.setString( + index++, + address + ); + } + + try (ResultSet resultSet = + statement.executeQuery()) { + + while (resultSet.next()) { + ShineUsersCodec.UserPdaSnapshot snapshot = + mapSnapshot(resultSet); + result.put( + snapshot.pdaAddress(), + snapshot + ); + } + } + } + + return result; + } + + public synchronized void updateLifecycleState( + String status, + boolean ready, + String lastError, + Long lastPollAtMs, + Long lastSuccessfulPollAtMs + ) throws Exception { + + SyncStateSnapshot current = + loadState(); + + upsertSyncState( + new SyncStateSnapshot( + status, + ready, + lastPollAtMs != null + ? lastPollAtMs + : current.lastPollAtMs(), + lastSuccessfulPollAtMs != null + ? lastSuccessfulPollAtMs + : current.lastSuccessfulPollAtMs(), + current.lastSeenSignature(), + current.lastSeenSlot(), + current.lastRelevantSignature(), + current.lastRelevantSlot(), + lastError, + current.economyConfigState(), + System.currentTimeMillis() + ) + ); + } + + public synchronized void applyHistoryBatch( + List txEntries, + List snapshots, + SyncStateSnapshot newState + ) throws Exception { + + try (Connection connection = + openConnection()) { + + connection.setAutoCommit(false); + + try { + + upsertTxHistory( + connection, + txEntries + ); + + insertHistorySnapshots( + connection, + snapshots + ); + + upsertCurrentSnapshots( + connection, + snapshots + ); + + upsertSyncState( + connection, + newState + ); + + connection.commit(); + + } catch (Exception exception) { + + connection.rollback(); + throw exception; + } finally { + connection.setAutoCommit(true); + } + } + } + + public synchronized void replaceCurrentFromFullSnapshot( + List snapshots, + SyncStateSnapshot newState + ) throws Exception { + + try (Connection connection = + openConnection()) { + + connection.setAutoCommit(false); + + try (Statement statement = + connection.createStatement()) { + + statement.executeUpdate( + "TRUNCATE TABLE solana_user_pda_current" + ); + + upsertCurrentSnapshots( + connection, + snapshots + ); + + upsertSyncState( + connection, + newState + ); + + connection.commit(); + + } catch (Exception exception) { + + connection.rollback(); + throw exception; + } finally { + connection.setAutoCommit(true); + } + } + } + + private void upsertTxHistory( + Connection connection, + List txEntries + ) throws Exception { + + if (txEntries == null + || txEntries.isEmpty()) { + return; + } + + String sql = + "INSERT INTO solana_sync_tx_history (" + + "signature, slot, block_time, tx_kind, " + + "is_relevant, affected_pda_address, affected_login, " + + "raw_summary_json, processed_at_ms" + + ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) " + + "ON CONFLICT (signature) DO UPDATE SET " + + "slot = EXCLUDED.slot, " + + "block_time = EXCLUDED.block_time, " + + "tx_kind = EXCLUDED.tx_kind, " + + "is_relevant = EXCLUDED.is_relevant, " + + "affected_pda_address = EXCLUDED.affected_pda_address, " + + "affected_login = EXCLUDED.affected_login, " + + "raw_summary_json = EXCLUDED.raw_summary_json, " + + "processed_at_ms = EXCLUDED.processed_at_ms"; + + try (PreparedStatement statement = + connection.prepareStatement(sql)) { + + for (TxHistoryEntry entry : txEntries) { + + statement.setString( + 1, + entry.signature() + ); + statement.setLong( + 2, + entry.slot() + ); + if (entry.blockTime() == null) { + statement.setNull( + 3, + Types.BIGINT + ); + } else { + statement.setLong( + 3, + entry.blockTime() + ); + } + statement.setString( + 4, + entry.txKind() + ); + statement.setBoolean( + 5, + entry.relevant() + ); + statement.setString( + 6, + entry.affectedPdaAddress() + ); + statement.setString( + 7, + entry.affectedLogin() + ); + statement.setString( + 8, + entry.rawSummaryJson() + ); + statement.setLong( + 9, + entry.processedAtMs() + ); + statement.addBatch(); + } + + statement.executeBatch(); + } + } + + private void insertHistorySnapshots( + Connection connection, + List snapshots + ) throws Exception { + + if (snapshots == null + || snapshots.isEmpty()) { + return; + } + + String sql = + "INSERT INTO solana_user_pda_history (" + + "tx_signature, slot, block_time, pda_address, login, " + + "record_number, recovery_key, root_key, client_key, " + + "blockchain_name, blockchain_key, paid_limit_bytes, " + + "used_bytes, last_block_number, last_block_hash, " + + "last_block_signature, arweave_tx_id, is_server, " + + "address_format_type, address_format_version, " + + "server_address, sync_servers_json, access_servers_json, " + + "sessions_mode, sessions_json, trusted_count, " + + "created_at_ms, updated_at_ms, prev_record_hash, " + + "record_signature, raw_data_base64, saved_at_ms" + + ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " + + "ON CONFLICT (pda_address, record_number) DO NOTHING"; + + try (PreparedStatement statement = + connection.prepareStatement(sql)) { + + long nowMs = + System.currentTimeMillis(); + + for (ShineUsersCodec.UserPdaSnapshot snapshot : snapshots) { + + bindSnapshot( + statement, + snapshot, + nowMs + ); + statement.addBatch(); + } + + statement.executeBatch(); + } + } + + private void upsertCurrentSnapshots( + Connection connection, + List snapshots + ) throws Exception { + + if (snapshots == null + || snapshots.isEmpty()) { + return; + } + + String sql = + "INSERT INTO solana_user_pda_current (" + + "pda_address, login, record_number, slot, last_tx_signature, " + + "recovery_key, root_key, client_key, blockchain_name, " + + "blockchain_key, paid_limit_bytes, used_bytes, " + + "last_block_number, last_block_hash, last_block_signature, " + + "arweave_tx_id, is_server, address_format_type, " + + "address_format_version, server_address, sync_servers_json, " + + "access_servers_json, sessions_mode, sessions_json, " + + "trusted_count, created_at_ms, updated_at_ms, " + + "prev_record_hash, record_signature, raw_data_base64, " + + "first_seen_at_ms, last_synced_at_ms" + + ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " + + "ON CONFLICT (pda_address) DO UPDATE SET " + + "login = EXCLUDED.login, " + + "record_number = EXCLUDED.record_number, " + + "slot = EXCLUDED.slot, " + + "last_tx_signature = EXCLUDED.last_tx_signature, " + + "recovery_key = EXCLUDED.recovery_key, " + + "root_key = EXCLUDED.root_key, " + + "client_key = EXCLUDED.client_key, " + + "blockchain_name = EXCLUDED.blockchain_name, " + + "blockchain_key = EXCLUDED.blockchain_key, " + + "paid_limit_bytes = EXCLUDED.paid_limit_bytes, " + + "used_bytes = EXCLUDED.used_bytes, " + + "last_block_number = EXCLUDED.last_block_number, " + + "last_block_hash = EXCLUDED.last_block_hash, " + + "last_block_signature = EXCLUDED.last_block_signature, " + + "arweave_tx_id = EXCLUDED.arweave_tx_id, " + + "is_server = EXCLUDED.is_server, " + + "address_format_type = EXCLUDED.address_format_type, " + + "address_format_version = EXCLUDED.address_format_version, " + + "server_address = EXCLUDED.server_address, " + + "sync_servers_json = EXCLUDED.sync_servers_json, " + + "access_servers_json = EXCLUDED.access_servers_json, " + + "sessions_mode = EXCLUDED.sessions_mode, " + + "sessions_json = EXCLUDED.sessions_json, " + + "trusted_count = EXCLUDED.trusted_count, " + + "created_at_ms = EXCLUDED.created_at_ms, " + + "updated_at_ms = EXCLUDED.updated_at_ms, " + + "prev_record_hash = EXCLUDED.prev_record_hash, " + + "record_signature = EXCLUDED.record_signature, " + + "raw_data_base64 = EXCLUDED.raw_data_base64, " + + "last_synced_at_ms = EXCLUDED.last_synced_at_ms " + + "WHERE EXCLUDED.record_number >= solana_user_pda_current.record_number"; + + try (PreparedStatement statement = + connection.prepareStatement(sql)) { + + long nowMs = + System.currentTimeMillis(); + + for (ShineUsersCodec.UserPdaSnapshot snapshot : snapshots) { + + bindCurrentSnapshot( + statement, + snapshot, + nowMs + ); + statement.addBatch(); + } + + statement.executeBatch(); + } + } + + private void bindSnapshot( + PreparedStatement statement, + ShineUsersCodec.UserPdaSnapshot snapshot, + long nowMs + ) throws Exception { + + statement.setString( + 1, + snapshot.lastTxSignature() + ); + statement.setLong( + 2, + snapshot.slot() + ); + statement.setNull( + 3, + Types.BIGINT + ); + statement.setString( + 4, + snapshot.pdaAddress() + ); + statement.setString( + 5, + snapshot.login() + ); + statement.setInt( + 6, + snapshot.recordNumber() + ); + statement.setString( + 7, + snapshot.recoveryKey() + ); + statement.setString( + 8, + snapshot.rootKey() + ); + statement.setString( + 9, + snapshot.clientKey() + ); + statement.setString( + 10, + snapshot.blockchainName() + ); + statement.setString( + 11, + snapshot.blockchainKey() + ); + statement.setLong( + 12, + snapshot.paidLimitBytes() + ); + statement.setLong( + 13, + snapshot.usedBytes() + ); + statement.setInt( + 14, + snapshot.lastBlockNumber() + ); + statement.setString( + 15, + snapshot.lastBlockHash() + ); + statement.setString( + 16, + snapshot.lastBlockSignature() + ); + statement.setString( + 17, + snapshot.arweaveTxId() + ); + statement.setBoolean( + 18, + snapshot.isServer() + ); + statement.setInt( + 19, + snapshot.addressFormatType() + ); + statement.setInt( + 20, + snapshot.addressFormatVersion() + ); + statement.setString( + 21, + snapshot.serverAddress() + ); + statement.setString( + 22, + writeJson( + snapshot.syncServers() + ) + ); + statement.setString( + 23, + writeJson( + snapshot.accessServers() + ) + ); + statement.setInt( + 24, + snapshot.sessionsMode() + ); + statement.setString( + 25, + writeJson( + snapshot.sessions() + ) + ); + statement.setInt( + 26, + snapshot.trustedCount() + ); + statement.setLong( + 27, + snapshot.createdAtMs() + ); + statement.setLong( + 28, + snapshot.updatedAtMs() + ); + statement.setString( + 29, + snapshot.prevRecordHash() + ); + statement.setString( + 30, + snapshot.recordSignature() + ); + statement.setString( + 31, + snapshot.rawDataBase64() + ); + statement.setLong( + 32, + nowMs + ); + } + + private void bindCurrentSnapshot( + PreparedStatement statement, + ShineUsersCodec.UserPdaSnapshot snapshot, + long nowMs + ) throws Exception { + + statement.setString(1, snapshot.pdaAddress()); + statement.setString(2, snapshot.login()); + statement.setInt(3, snapshot.recordNumber()); + statement.setLong(4, snapshot.slot()); + statement.setString(5, snapshot.lastTxSignature()); + statement.setString(6, snapshot.recoveryKey()); + statement.setString(7, snapshot.rootKey()); + statement.setString(8, snapshot.clientKey()); + statement.setString(9, snapshot.blockchainName()); + statement.setString(10, snapshot.blockchainKey()); + statement.setLong(11, snapshot.paidLimitBytes()); + statement.setLong(12, snapshot.usedBytes()); + statement.setInt(13, snapshot.lastBlockNumber()); + statement.setString(14, snapshot.lastBlockHash()); + statement.setString(15, snapshot.lastBlockSignature()); + statement.setString(16, snapshot.arweaveTxId()); + statement.setBoolean(17, snapshot.isServer()); + statement.setInt(18, snapshot.addressFormatType()); + statement.setInt(19, snapshot.addressFormatVersion()); + statement.setString(20, snapshot.serverAddress()); + statement.setString(21, writeJson(snapshot.syncServers())); + statement.setString(22, writeJson(snapshot.accessServers())); + statement.setInt(23, snapshot.sessionsMode()); + statement.setString(24, writeJson(snapshot.sessions())); + statement.setInt(25, snapshot.trustedCount()); + statement.setLong(26, snapshot.createdAtMs()); + statement.setLong(27, snapshot.updatedAtMs()); + statement.setString(28, snapshot.prevRecordHash()); + statement.setString(29, snapshot.recordSignature()); + statement.setString(30, snapshot.rawDataBase64()); + statement.setLong(31, nowMs); + statement.setLong(32, nowMs); + } + + private ShineUsersCodec.UserPdaSnapshot mapSnapshot( + ResultSet resultSet + ) throws Exception { + + List syncServers = + Arrays.asList( + mapper.readValue( + resultSet.getString("sync_servers_json"), + String[].class + ) + ); + + List accessServers = + Arrays.asList( + mapper.readValue( + resultSet.getString("access_servers_json"), + String[].class + ) + ); + + List sessions = + Arrays.asList( + mapper.readValue( + resultSet.getString("sessions_json"), + ShineUsersCodec.UserSessionSnapshot[].class + ) + ); + + return new ShineUsersCodec.UserPdaSnapshot( + resultSet.getString("pda_address"), + resultSet.getString("login"), + resultSet.getInt("record_number"), + resultSet.getLong("slot"), + resultSet.getString("last_tx_signature"), + resultSet.getString("recovery_key"), + resultSet.getString("root_key"), + resultSet.getString("client_key"), + resultSet.getString("blockchain_name"), + resultSet.getString("blockchain_key"), + resultSet.getLong("paid_limit_bytes"), + resultSet.getLong("used_bytes"), + resultSet.getInt("last_block_number"), + resultSet.getString("last_block_hash"), + resultSet.getString("last_block_signature"), + resultSet.getString("arweave_tx_id"), + resultSet.getBoolean("is_server"), + resultSet.getInt("address_format_type"), + resultSet.getInt("address_format_version"), + resultSet.getString("server_address"), + List.copyOf(syncServers), + List.copyOf(accessServers), + resultSet.getInt("sessions_mode"), + List.copyOf(sessions), + resultSet.getInt("trusted_count"), + resultSet.getLong("created_at_ms"), + resultSet.getLong("updated_at_ms"), + resultSet.getString("prev_record_hash"), + resultSet.getString("record_signature"), + resultSet.getString("raw_data_base64") + ); + } + + private void upsertSyncState( + SyncStateSnapshot snapshot + ) throws Exception { + + try (Connection connection = + openConnection()) { + upsertSyncState( + connection, + snapshot + ); + } + } + + private void upsertSyncState( + Connection connection, + SyncStateSnapshot snapshot + ) throws Exception { + + String sql = + "INSERT INTO solana_sync_state (" + + "id, status, ready, last_poll_at_ms, " + + "last_successful_poll_at_ms, last_seen_signature, " + + "last_seen_slot, last_relevant_signature, " + + "last_relevant_slot, last_error, economy_config_version, " + + "registration_fee_lamports, lamports_per_limit_step, " + + "start_bonus_limit, updated_at_ms" + + ") VALUES (1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) " + + "ON CONFLICT (id) DO UPDATE SET " + + "status = EXCLUDED.status, " + + "ready = EXCLUDED.ready, " + + "last_poll_at_ms = EXCLUDED.last_poll_at_ms, " + + "last_successful_poll_at_ms = EXCLUDED.last_successful_poll_at_ms, " + + "last_seen_signature = EXCLUDED.last_seen_signature, " + + "last_seen_slot = EXCLUDED.last_seen_slot, " + + "last_relevant_signature = EXCLUDED.last_relevant_signature, " + + "last_relevant_slot = EXCLUDED.last_relevant_slot, " + + "last_error = EXCLUDED.last_error, " + + "economy_config_version = EXCLUDED.economy_config_version, " + + "registration_fee_lamports = EXCLUDED.registration_fee_lamports, " + + "lamports_per_limit_step = EXCLUDED.lamports_per_limit_step, " + + "start_bonus_limit = EXCLUDED.start_bonus_limit, " + + "updated_at_ms = EXCLUDED.updated_at_ms"; + + try (PreparedStatement statement = + connection.prepareStatement(sql)) { + + statement.setString(1, snapshot.status()); + statement.setBoolean(2, snapshot.ready()); + bindNullableLong(statement, 3, snapshot.lastPollAtMs()); + bindNullableLong(statement, 4, snapshot.lastSuccessfulPollAtMs()); + statement.setString(5, snapshot.lastSeenSignature()); + bindNullableLong(statement, 6, snapshot.lastSeenSlot()); + statement.setString(7, snapshot.lastRelevantSignature()); + bindNullableLong(statement, 8, snapshot.lastRelevantSlot()); + statement.setString(9, snapshot.lastError()); + + if (snapshot.economyConfigState() == null) { + statement.setNull(10, Types.INTEGER); + statement.setNull(11, Types.BIGINT); + statement.setNull(12, Types.BIGINT); + statement.setNull(13, Types.BIGINT); + } else { + statement.setInt(10, snapshot.economyConfigState().version()); + statement.setLong(11, snapshot.economyConfigState().registrationFeeLamports()); + statement.setLong(12, snapshot.economyConfigState().lamportsPerLimitStep()); + statement.setLong(13, snapshot.economyConfigState().startBonusLimit()); + } + + bindNullableLong(statement, 14, snapshot.updatedAtMs()); + statement.executeUpdate(); + } + } + + private void migrate() + throws Exception { + + try (Connection connection = + openConnection(); + Statement statement = + connection.createStatement()) { + + statement.executeUpdate( + "CREATE TABLE IF NOT EXISTS solana_sync_state (" + + "id INTEGER PRIMARY KEY CHECK (id = 1), " + + "status TEXT NOT NULL, " + + "ready BOOLEAN NOT NULL DEFAULT FALSE, " + + "last_poll_at_ms BIGINT, " + + "last_successful_poll_at_ms BIGINT, " + + "last_seen_signature TEXT, " + + "last_seen_slot BIGINT, " + + "last_relevant_signature TEXT, " + + "last_relevant_slot BIGINT, " + + "last_error TEXT, " + + "economy_config_version INTEGER, " + + "registration_fee_lamports BIGINT, " + + "lamports_per_limit_step BIGINT, " + + "start_bonus_limit BIGINT, " + + "updated_at_ms BIGINT" + + ")" + ); + + statement.executeUpdate( + "INSERT INTO solana_sync_state (id, status, ready) " + + "VALUES (1, 'EMPTY', FALSE) " + + "ON CONFLICT (id) DO NOTHING" + ); + + statement.executeUpdate( + "CREATE TABLE IF NOT EXISTS solana_sync_tx_history (" + + "signature TEXT PRIMARY KEY, " + + "slot BIGINT NOT NULL, " + + "block_time BIGINT, " + + "tx_kind TEXT NOT NULL, " + + "is_relevant BOOLEAN NOT NULL, " + + "affected_pda_address TEXT, " + + "affected_login TEXT, " + + "raw_summary_json TEXT NOT NULL, " + + "processed_at_ms BIGINT NOT NULL" + + ")" + ); + + statement.executeUpdate( + "CREATE INDEX IF NOT EXISTS idx_sync_tx_history_slot " + + "ON solana_sync_tx_history(slot)" + ); + statement.executeUpdate( + "CREATE INDEX IF NOT EXISTS idx_sync_tx_history_relevant " + + "ON solana_sync_tx_history(is_relevant)" + ); + statement.executeUpdate( + "CREATE INDEX IF NOT EXISTS idx_sync_tx_history_login " + + "ON solana_sync_tx_history(affected_login)" + ); + + statement.executeUpdate( + "CREATE TABLE IF NOT EXISTS solana_user_pda_current (" + + "pda_address TEXT PRIMARY KEY, " + + "login TEXT NOT NULL UNIQUE, " + + "record_number INTEGER NOT NULL, " + + "slot BIGINT NOT NULL, " + + "last_tx_signature TEXT NOT NULL, " + + "recovery_key TEXT NOT NULL, " + + "root_key TEXT NOT NULL, " + + "client_key TEXT NOT NULL, " + + "blockchain_name TEXT NOT NULL, " + + "blockchain_key TEXT NOT NULL, " + + "paid_limit_bytes BIGINT NOT NULL, " + + "used_bytes BIGINT NOT NULL, " + + "last_block_number INTEGER NOT NULL, " + + "last_block_hash TEXT NOT NULL, " + + "last_block_signature TEXT NOT NULL, " + + "arweave_tx_id TEXT NOT NULL, " + + "is_server BOOLEAN NOT NULL, " + + "address_format_type INTEGER NOT NULL, " + + "address_format_version INTEGER NOT NULL, " + + "server_address TEXT NOT NULL, " + + "sync_servers_json TEXT NOT NULL, " + + "access_servers_json TEXT NOT NULL, " + + "sessions_mode INTEGER NOT NULL, " + + "sessions_json TEXT NOT NULL, " + + "trusted_count INTEGER NOT NULL, " + + "created_at_ms BIGINT NOT NULL, " + + "updated_at_ms BIGINT NOT NULL, " + + "prev_record_hash TEXT NOT NULL, " + + "record_signature TEXT NOT NULL, " + + "raw_data_base64 TEXT NOT NULL, " + + "first_seen_at_ms BIGINT NOT NULL, " + + "last_synced_at_ms BIGINT NOT NULL" + + ")" + ); + + statement.executeUpdate( + "CREATE INDEX IF NOT EXISTS idx_user_pda_current_slot " + + "ON solana_user_pda_current(slot)" + ); + + statement.executeUpdate( + "CREATE TABLE IF NOT EXISTS solana_user_pda_history (" + + "id BIGSERIAL PRIMARY KEY, " + + "tx_signature TEXT NOT NULL, " + + "slot BIGINT NOT NULL, " + + "block_time BIGINT, " + + "pda_address TEXT NOT NULL, " + + "login TEXT NOT NULL, " + + "record_number INTEGER NOT NULL, " + + "recovery_key TEXT NOT NULL, " + + "root_key TEXT NOT NULL, " + + "client_key TEXT NOT NULL, " + + "blockchain_name TEXT NOT NULL, " + + "blockchain_key TEXT NOT NULL, " + + "paid_limit_bytes BIGINT NOT NULL, " + + "used_bytes BIGINT NOT NULL, " + + "last_block_number INTEGER NOT NULL, " + + "last_block_hash TEXT NOT NULL, " + + "last_block_signature TEXT NOT NULL, " + + "arweave_tx_id TEXT NOT NULL, " + + "is_server BOOLEAN NOT NULL, " + + "address_format_type INTEGER NOT NULL, " + + "address_format_version INTEGER NOT NULL, " + + "server_address TEXT NOT NULL, " + + "sync_servers_json TEXT NOT NULL, " + + "access_servers_json TEXT NOT NULL, " + + "sessions_mode INTEGER NOT NULL, " + + "sessions_json TEXT NOT NULL, " + + "trusted_count INTEGER NOT NULL, " + + "created_at_ms BIGINT NOT NULL, " + + "updated_at_ms BIGINT NOT NULL, " + + "prev_record_hash TEXT NOT NULL, " + + "record_signature TEXT NOT NULL, " + + "raw_data_base64 TEXT NOT NULL, " + + "saved_at_ms BIGINT NOT NULL, " + + "UNIQUE (pda_address, record_number)" + + ")" + ); + + statement.executeUpdate( + "CREATE INDEX IF NOT EXISTS idx_user_pda_history_login " + + "ON solana_user_pda_history(login)" + ); + statement.executeUpdate( + "CREATE INDEX IF NOT EXISTS idx_user_pda_history_slot " + + "ON solana_user_pda_history(slot)" + ); + } + } + + private Connection openConnection() + throws Exception { + return DriverManager.getConnection( + databaseUrl, + databaseUser, + databasePassword + ); + } + + private static void bindNullableLong( + PreparedStatement statement, + int index, + Long value + ) throws SQLException { + if (value == null) { + statement.setNull(index, Types.BIGINT); + } else { + statement.setLong(index, value); + } + } + + private static Long getNullableLong( + ResultSet resultSet, + String column + ) throws SQLException { + long value = resultSet.getLong(column); + return resultSet.wasNull() ? null : value; + } + + private static Integer getNullableInt( + ResultSet resultSet, + String column + ) throws SQLException { + int value = resultSet.getInt(column); + return resultSet.wasNull() ? null : value; + } + + private String writeJson( + Object value + ) throws JsonProcessingException { + return mapper.writeValueAsString(value); + } + + @Override + public void close() { + } + + public record TxHistoryEntry( + String signature, + long slot, + Long blockTime, + String txKind, + boolean relevant, + String affectedPdaAddress, + String affectedLogin, + String rawSummaryJson, + long processedAtMs + ) { + } + + public record SyncStateSnapshot( + String status, + boolean ready, + Long lastPollAtMs, + Long lastSuccessfulPollAtMs, + String lastSeenSignature, + Long lastSeenSlot, + String lastRelevantSignature, + Long lastRelevantSlot, + String lastError, + ShineUsersCodec.EconomyConfigState economyConfigState, + Long updatedAtMs + ) { + } +} diff --git a/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/util/Base58Util.java b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/util/Base58Util.java new file mode 100644 index 00000000..7b985555 --- /dev/null +++ b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/util/Base58Util.java @@ -0,0 +1,224 @@ +package sync.util; + +import java.nio.charset.StandardCharsets; +import java.util.Arrays; + +public final class Base58Util { + + private static final char[] ALPHABET = + "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz" + .toCharArray(); + + private static final int[] INDEXES = + new int[128]; + + static { + Arrays.fill( + INDEXES, + -1 + ); + + for (int i = 0; i < ALPHABET.length; i++) { + INDEXES[ALPHABET[i]] = i; + } + } + + private Base58Util() { + } + + public static String encode( + byte[] input + ) { + + if (input == null + || input.length == 0) { + return ""; + } + + byte[] copy = + Arrays.copyOf( + input, + input.length + ); + + int zeros = + 0; + + while (zeros < copy.length + && copy[zeros] == 0) { + zeros++; + } + + byte[] encoded = + new byte[copy.length * 2]; + + int outputStart = + encoded.length; + + int inputStart = + zeros; + + while (inputStart < copy.length) { + + int remainder = + divMod58( + copy, + inputStart + ); + + if (copy[inputStart] == 0) { + inputStart++; + } + + encoded[--outputStart] = + (byte) ALPHABET[remainder]; + } + + while (outputStart < encoded.length + && encoded[outputStart] == ALPHABET[0]) { + outputStart++; + } + + while (--zeros >= 0) { + encoded[--outputStart] = + (byte) ALPHABET[0]; + } + + return new String( + encoded, + outputStart, + encoded.length - outputStart, + StandardCharsets.US_ASCII + ); + } + + public static byte[] decode( + String input + ) { + + if (input == null + || input.isBlank()) { + return new byte[0]; + } + + char[] chars = + input.trim() + .toCharArray(); + + byte[] input58 = + new byte[chars.length]; + + for (int i = 0; i < chars.length; i++) { + + char c = chars[i]; + + if (c >= INDEXES.length + || INDEXES[c] < 0) { + throw new IllegalArgumentException( + "Invalid Base58 character: " + c + ); + } + + input58[i] = + (byte) INDEXES[c]; + } + + int zeros = + 0; + + while (zeros < input58.length + && input58[zeros] == 0) { + zeros++; + } + + byte[] decoded = + new byte[chars.length]; + + int outputStart = + decoded.length; + + int inputStart = + zeros; + + while (inputStart < input58.length) { + + int remainder = + divMod256( + input58, + inputStart + ); + + if (input58[inputStart] == 0) { + inputStart++; + } + + decoded[--outputStart] = + (byte) remainder; + } + + while (outputStart < decoded.length + && decoded[outputStart] == 0) { + outputStart++; + } + + return Arrays.copyOfRange( + decoded, + outputStart - zeros, + decoded.length + ); + } + + private static int divMod58( + byte[] number, + int startAt + ) { + + int remainder = + 0; + + for (int i = startAt; i < number.length; i++) { + + int digit256 = + number[i] & 0xFF; + + int temp = + remainder * 256 + + digit256; + + number[i] = + (byte) (temp / 58); + + remainder = + temp % 58; + } + + return remainder; + } + + private static int divMod256( + byte[] number58, + int startAt + ) { + + int remainder = + 0; + + for (int i = startAt; i < number58.length; i++) { + + int digit58 = + number58[i] & 0xFF; + + int temp = + remainder * 58 + + digit58; + + number58[i] = + (byte) (temp / 256); + + remainder = + temp % 256; + } + + return remainder; + } +} diff --git a/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/util/SolanaPdaUtil.java b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/util/SolanaPdaUtil.java new file mode 100644 index 00000000..af065063 --- /dev/null +++ b/SHiNE-server/shine-server-solana-users-sync/src/main/java/sync/util/SolanaPdaUtil.java @@ -0,0 +1,95 @@ +package sync.util; + +import org.bouncycastle.jcajce.provider.digest.SHA256; +import org.bouncycastle.math.ec.rfc8032.Ed25519; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; + +public final class SolanaPdaUtil { + + private static final byte[] PROGRAM_DERIVED_ADDRESS_DOMAIN = + "ProgramDerivedAddress" + .getBytes(StandardCharsets.UTF_8); + + private SolanaPdaUtil() { + } + + public static String findProgramAddress( + List seeds, + String programIdBase58 + ) { + + byte[] programId = + Base58Util.decode( + programIdBase58 + ); + + if (programId.length != 32) { + throw new IllegalArgumentException( + "Program id must decode to 32 bytes" + ); + } + + for (int bump = 255; bump >= 0; bump--) { + + List attemptSeeds = + new ArrayList<>( + seeds + ); + + attemptSeeds.add( + new byte[]{(byte) bump} + ); + + byte[] address = + createProgramAddress( + attemptSeeds, + programId + ); + + if (!isOnCurve(address)) { + return Base58Util.encode( + address + ); + } + } + + throw new IllegalStateException( + "Unable to find viable program address" + ); + } + + private static byte[] createProgramAddress( + List seeds, + byte[] programId + ) { + + SHA256.Digest digest = + new SHA256.Digest(); + + for (byte[] seed : seeds) { + digest.update(seed); + } + + digest.update(programId); + digest.update(PROGRAM_DERIVED_ADDRESS_DOMAIN); + + return digest.digest(); + } + + private static boolean isOnCurve( + byte[] publicKey + ) { + + if (publicKey.length != 32) { + return false; + } + + return Ed25519.validatePublicKeyFull( + publicKey, + 0 + ); + } +} diff --git a/SHiNE-server/src/main/java/server/sync/SolanaUsersSyncStartupService.java b/SHiNE-server/src/main/java/server/sync/SolanaUsersSyncStartupService.java new file mode 100644 index 00000000..491bbb1d --- /dev/null +++ b/SHiNE-server/src/main/java/server/sync/SolanaUsersSyncStartupService.java @@ -0,0 +1,85 @@ +package server.sync; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import sync.service.SolanaUsersSyncService; +import utils.config.AppConfig; + +public final class SolanaUsersSyncStartupService { + + private static final Logger log = + LoggerFactory.getLogger( + SolanaUsersSyncStartupService.class + ); + + private static volatile SolanaUsersSyncService service; + + private SolanaUsersSyncStartupService() { + } + + public static synchronized void startOrThrow() + throws Exception { + + if (service != null) { + return; + } + + AppConfig serverConfig = + AppConfig.getInstance(); + + if (!sync.config.AppConfig.isEnabled(serverConfig)) { + log.info( + "Solana users sync is disabled. Param {} is false or absent.", + sync.config.AppConfig.ENABLED_KEY + ); + return; + } + + sync.config.AppConfig syncConfig = + sync.config.AppConfig.fromServerConfig( + serverConfig + ); + + service = + new SolanaUsersSyncService( + syncConfig + ); + + Runtime.getRuntime() + .addShutdownHook( + new Thread( + SolanaUsersSyncStartupService::closeQuietly, + "solana-users-sync-shutdown" + ) + ); + + log.info( + "Starting Solana users sync before remaining server startup..." + ); + + service.start(); + service.awaitReady(); + + log.info( + "Solana users sync is READY. Server startup may continue." + ); + } + + public static synchronized void closeQuietly() { + + if (service == null) { + return; + } + + try { + service.close(); + } catch (Exception exception) { + log.warn( + "Failed to close Solana users sync service cleanly", + exception + ); + } finally { + service = null; + } + } +} diff --git a/SHiNE-server/src/main/java/server/ws/WsServer.java b/SHiNE-server/src/main/java/server/ws/WsServer.java index d318b2e2..56fc8a5a 100644 --- a/SHiNE-server/src/main/java/server/ws/WsServer.java +++ b/SHiNE-server/src/main/java/server/ws/WsServer.java @@ -8,6 +8,7 @@ import org.slf4j.LoggerFactory; import server.debug.DebugApiConfigurator; import server.sync.BlockchainResyncRecoveryOnStartup; import server.sync.PeriodicBlockchainSyncService; +import server.sync.SolanaUsersSyncStartupService; import server.sync.SyncServersBootstrapService; import utils.config.AppConfig; @@ -49,6 +50,11 @@ public final class WsServer { throw e; } + // ============================================================ + // 1.0) Синхронизация пользовательских Solana PDA + // ============================================================ + SolanaUsersSyncStartupService.startOrThrow(); + // ============================================================ // 1) Настройки порта // ============================================================ diff --git a/SHiNE-server/src/main/resources/application.properties b/SHiNE-server/src/main/resources/application.properties index b910adbd..1107feea 100644 --- a/SHiNE-server/src/main/resources/application.properties +++ b/SHiNE-server/src/main/resources/application.properties @@ -3,6 +3,14 @@ db.path=data/shine.sqlite server.SHiNE.login=shineupme solana.cluster=mainnet-beta solana.rpcUrl=https://api.mainnet-beta.solana.com +solana.users.sync.enabled=false +solana.users.sync.rpcUrl= +solana.users.sync.wsUrl= +solana.users.sync.programId=SHiNEPr1APdAgNBteUyBXcNovaHctpSjUu8oH2ZJdN6 +solana.users.sync.databaseUrl= +solana.users.sync.dbUser= +solana.users.sync.dbPassword= +solana.users.sync.pollIntervalSeconds=300 # ------------------------------------------------------------ # Межсерверная синхронизация: как создавать локальную запись пользователя, diff --git a/VERSION.properties b/VERSION.properties index aab16956..9ee98741 100644 --- a/VERSION.properties +++ b/VERSION.properties @@ -1,2 +1,2 @@ client.version=1.2.347 -server.version=1.2.318 +server.version=1.2.319 diff --git a/build.gradle b/build.gradle index 8014732a..f31b8758 100644 --- a/build.gradle +++ b/build.gradle @@ -81,6 +81,7 @@ dependencies { implementation project(':shine-server-net-protocol') // Модуль отвечающий за протокол (классы Net..Request/Response implementation project(':shine-server-net-server') // Хэндлеры для обработки сетевых запросов + implementation project(':shine-server-solana-users-sync') diff --git a/deploy/POSTGRESQL_SERVERS_STANDARD.md b/deploy/POSTGRESQL_SERVERS_STANDARD.md new file mode 100644 index 00000000..7fbfc28b --- /dev/null +++ b/deploy/POSTGRESQL_SERVERS_STANDARD.md @@ -0,0 +1,213 @@ +# PostgreSQL для серверов SHiNE + +Этот документ фиксирует целевой стандарт PostgreSQL, который должен использоваться на всех серверных контурах SHiNE после ухода от SQLite. + +Пока это не текущий production-state, а согласованная целевая схема, под которую можно писать миграции, deploy-скрипты и server-конфиги. + +## Цель + +На каждом сервере SHiNE должна быть отдельная локальная PostgreSQL-база: + +- запускается в Docker; +- использует `postgres:18`; +- хранит данные не внутри контейнера, а в примонтированной папке хоста; +- доступна только локально на сервере и из Docker-сети; +- не публикуется в интернет; +- автоматически перезапускается после reboot/crash; +- имеет healthcheck; +- использует отдельную БД приложения и отдельного пользователя приложения; +- пароли и секреты хранятся только в локальном `.env`/override-конфиге на сервере и не коммитятся в git. + +## Базовый стандарт + +- Версия PostgreSQL: `18.x` +- Docker image: `postgres:18` +- Имена по умолчанию: + - контейнер/сервис: `shine-postgres` + - база приложения: `shine_server_db` + - пользователь приложения: `shine_server` +- Политика перезапуска: `unless-stopped` +- Проверка готовности: `pg_isready` + +## Сетевой доступ + +PostgreSQL не должна быть доступна из интернета. + +Разрешённые варианты: + +- публиковать порт только на loopback: + - `127.0.0.1:5432:5432` +- либо не публиковать порт вообще, если клиент тоже живёт в Docker и ходит только по внутренней сети + +Запрещено: + +- `0.0.0.0:5432:5432` +- открытие `5432/tcp` через внешний firewall/NAT/public ingress + +Рекомендация по умолчанию для SHiNE: + +- использовать `127.0.0.1:5432:5432` + +Это даёт: + +- локальную диагностику через `psql` на самом сервере; +- отсутствие внешнего доступа из интернета; +- совместимость с приложением, если оно работает не в Docker. + +## Хранение данных + +Данные PostgreSQL должны храниться в локальной папке хоста, а не во внутреннем Docker volume контейнера. + +Причины: + +- проще бэкапить; +- проще переносить между серверами; +- проще контролировать место хранения; +- одинаковая схема для dev/test/production; +- ниже риск потерять данные при пересоздании контейнера. + +Рекомендуемая схема каталогов на сервере: + +Production: + +```text +/home/player/SHiNE/postgres/shine_server_db +``` + +Test/devnet: + +```text +/home/player/tX/postgres/shine_server_db +``` + +Если на одном сервере появится несколько баз SHiNE-сервисов, раскладывать их по отдельным каталогам: + +```text +/home/player/SHiNE/postgres/shine_server_db +/home/player/SHiNE/postgres/analytics_db +/home/player/SHiNE/postgres/other_service_db +``` + +## Пользователи и права + +Нужны два уровня доступа: + +- системный `postgres` superuser для администрирования; +- рабочий пользователь приложения `shine_server` для подключения самого SHiNE server. + +Требования: + +- приложение не должно работать под `postgres`; +- `shine_server` не должен быть `superuser`; +- `shine_server` должен владеть своей БД `shine_server_db`; +- для миграций по умолчанию использовать пользователя приложения, если ему хватает прав; +- административные операции выполнять отдельно под `postgres`. + +## Секреты + +В git нельзя хранить: + +- `.env` с реальными паролями; +- connection strings с паролями; +- SQL-файлы с зашитыми production/test паролями. + +Хранить на сервере локально, например: + +```text +/home/player/SHiNE/postgres/.env +``` + +или в другом root-only каталоге секретов хоста. + +Минимально нужны: + +- пароль `postgres` +- пароль `shine_server` + +Допустимо, но не рекомендуется, временно использовать одинаковый пароль для локального dev. Для production и постоянных test-серверов лучше разные пароли. + +## Пример целевого compose + +Ниже пример целевой схемы, которую можно брать за основу для серверов: + +```yaml +services: + shine-postgres: + image: postgres:18 + container_name: shine-postgres + restart: unless-stopped + env_file: + - .env + environment: + POSTGRES_DB: ${POSTGRES_SUPERUSER_DB} + POSTGRES_USER: ${POSTGRES_SUPERUSER} + POSTGRES_PASSWORD: ${POSTGRES_SUPERUSER_PASSWORD} + PGDATA: /var/lib/postgresql/data/pgdata + ports: + - "127.0.0.1:5432:5432" + volumes: + - ${SHINE_POSTGRES_DATA_DIR}:/var/lib/postgresql/data + - ./initdb:/docker-entrypoint-initdb.d:ro + healthcheck: + test: + - CMD-SHELL + - pg_isready -U "$${POSTGRES_SUPERUSER}" -d "$${POSTGRES_SUPERUSER_DB}" + interval: 10s + timeout: 5s + retries: 10 + start_period: 20s +``` + +## Рекомендуемые переменные окружения + +```dotenv +POSTGRES_SUPERUSER_DB=postgres +POSTGRES_SUPERUSER=postgres +POSTGRES_SUPERUSER_PASSWORD=change_me +SHINE_APP_DB=shine_server_db +SHINE_APP_USER=shine_server +SHINE_APP_PASSWORD=change_me_too +SHINE_POSTGRES_DATA_DIR=/home/player/SHiNE/postgres/shine_server_db +``` + +Для test/devnet путь адаптировать под конкретный контур: + +```dotenv +SHINE_POSTGRES_DATA_DIR=/home/player/t2/postgres/shine_server_db +``` + +## Подключение приложения + +SHiNE server должен подключаться именно к: + +- host: `127.0.0.1` +- port: `5432` +- db: `shine_server_db` +- user: `shine_server` + +Не подключать серверное приложение под `postgres`, кроме одноразовых административных операций вручную. + +## Бэкапы + +Так как данные лежат в bind-mount каталоге хоста, сама папка с данными должна попадать в серверную backup-стратегию. + +Но для PostgreSQL предпочтителен не только файловый backup, а как минимум один из вариантов: + +- регулярный `pg_dump` +- либо полноценный backup-скрипт с остановкой приложения/согласованным snapshot + +Минимальное требование: + +- перед серьёзными миграциями иметь свежий backup БД; +- перед production rollout новой серверной версии иметь проверяемый backup. + +## Для будущей миграции SHiNE + +При переводе SHiNE server с SQLite на PostgreSQL считать обязательным: + +- сначала поднять PostgreSQL по этому стандарту; +- затем добавить серверные конфиги подключения; +- затем прогнать миграции схемы; +- только потом переключать runtime приложения на PostgreSQL. + +До фактического rollout на конкретный сервер эта БД может ещё отсутствовать. Этот документ описывает не текущее наличие БД, а обязательный целевой стандарт для всех серверов SHiNE. diff --git a/deploy/README.md b/deploy/README.md index 78bd95ba..78ef3dd7 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -20,6 +20,8 @@ - `PRODUCTION_SERVERS.md` — production-контуры. - `TEST_SERVERS.md` — test/devnet-контуры. - `TURN_SERVERS.md` — TURN-серверы. +- `POSTGRESQL_SERVERS_STANDARD.md` — целевой стандарт PostgreSQL для всех серверов SHiNE. +- `SOLANA_USERS_SYNC_SERVER_SETUP.md` — интеграция синхронизации пользовательских Solana PDA в основной сервер. - `CONFIGURE_TURN_IN_SHINE.md` — как подключить TURN к SHiNE backend. - `SETUP_SERVER_FROM_ZERO.md` — настройка SHiNE-сервера и UI с нуля. - `SETUP_TURN_SERVER.md` — настройка TURN через Caddy/DNS/TLS. diff --git a/deploy/SOLANA_USERS_SYNC_SERVER_SETUP.md b/deploy/SOLANA_USERS_SYNC_SERVER_SETUP.md new file mode 100644 index 00000000..746f40d7 --- /dev/null +++ b/deploy/SOLANA_USERS_SYNC_SERVER_SETUP.md @@ -0,0 +1,116 @@ +# Интеграция синхронизации `shine_users` в основной сервер + +Этот документ описывает, что нужно для встраивания Solana sync-модуля пользовательских PDA в основной SHiNE-server. + +Основной архитектурный документ: + +- [docs/Solana/SOLANA_USERS_SYNC_MODULE_DESIGN.md](/home/ai/work/SHiNE/SHiNE-server-sha256/SHiNE-product/docs/Solana/SOLANA_USERS_SYNC_MODULE_DESIGN.md) + +## Что уже готово + +Отдельный модуль `sync-solana` уже умеет: + +- подключаться к Solana RPC и WebSocket; +- вычислять `users_economy_config_pda`; +- хранить checkpoint синхронизации в PostgreSQL; +- читать историю через `getSignaturesForAddress(users_economy_config_pda)`; +- поддерживать realtime через websocket; +- выполнять страховочный periodic poll раз в 5 минут; +- хранить: + - `solana_sync_state` + - `solana_sync_tx_history` + - `solana_user_pda_current` + - `solana_user_pda_history` +- блокировать дальнейший startup до входа в `READY`. + +## Что нужно перенести в основной сервер + +Из `sync-solana` в сервер нужно перенести рабочие классы: + +- `sync-solana/src/main/java/sync-solana/config/` +- `sync-solana/src/main/java/sync-solana/service/` +- `sync-solana/src/main/java/sync-solana/source/` +- `sync-solana/src/main/java/sync-solana/source/rpc/` +- `sync-solana/src/main/java/sync-solana/storage/postgres/` +- `sync-solana/src/main/java/sync-solana/codec/` +- `sync-solana/src/main/java/sync-solana/model/` +- `sync-solana/src/main/java/sync-solana/util/` + +`Main.java` нужен только как reference для bootstrap и как отдельный `main` в сервере уже не понадобится. + +Рекомендуемый вариант: + +- оформить это как отдельный Gradle submodule внутри `SHiNE-server`; +- запускать его из server startup как lifecycle-сервис. + +## Порядок запуска в сервере + +При старте основного сервера последовательность должна быть такой: + +1. прочитать общий server config; +2. создать Solana users sync service; +3. вызвать `start()`; +4. вызвать `awaitReady()`; +5. только после этого продолжать остальной startup сервера: + - синхронизацию с другими нодами; + - запуск WS/HTTP; + - остальную серверную инициализацию. + +Если Solana initial sync не дошёл до `READY`, startup сервера должен считаться неуспешным. + +## Переменные окружения сервера + +На сервере должны быть доступны: + +```text +SOLANA_RPC_URL= +SOLANA_WS_URL= +SOLANA_PROGRAM_ID=SHiNEPr1APdAgNBteUyBXcNovaHctpSjUu8oH2ZJdN6 +SYNC_POLL_INTERVAL_SECONDS=300 +``` + +Для PostgreSQL sync-модуль может использовать уже существующую server PostgreSQL-конфигурацию, если сервер уже предоставляет: + +```text +DATABASE_URL= +PGUSER= +PGPASSWORD= +``` + +Если в сервере используется другая схема конфигов, нужно сделать адаптер на уровне server config, а не менять саму логику sync. + +## Логи + +Sync-модуль должен писать в общие server logs через тот же `slf4j/logback`, что и основной сервер. + +Минимум, который должен быть виден в логах: + +- старт sync-модуля; +- вход в `READY`; +- realtime sync; +- periodic poll; +- reconnect websocket; +- fallback на full snapshot; +- ошибки RPC/WS/DB. + +## Что потребуется по deploy + +Отдельных deploy-скриптов для sync-модуля не требуется, если он встроен в основной server jar. + +По deploy нужно: + +- обновить server env/override-конфиг новыми переменными `SOLANA_*` и `SYNC_POLL_INTERVAL_SECONDS`; +- убедиться, что на сервере доступен PostgreSQL, в который модуль будет писать свои таблицы; +- при необходимости описать новые env в документации конкретного server-контура. + +## Что ещё проверить после интеграции + +После встраивания в основной сервер нужно отдельно проверить: + +- startup сервера с ожиданием `awaitReady()`; +- создание таблиц в server PostgreSQL; +- initial sync после пустой БД; +- restart recovery после уже существующего checkpoint; +- realtime update через websocket; +- periodic poll без новых транзакций; +- fallback на full snapshot при потере history anchor. diff --git a/docs/Solana/SOLANA_USERS_SYNC_MODULE_DESIGN.md b/docs/Solana/SOLANA_USERS_SYNC_MODULE_DESIGN.md new file mode 100644 index 00000000..35da6ea2 --- /dev/null +++ b/docs/Solana/SOLANA_USERS_SYNC_MODULE_DESIGN.md @@ -0,0 +1,550 @@ +# Модуль синхронизации `shine_users` из Solana + +## Назначение + +Этот модуль нужен для локальной серверной синхронизации всех пользовательских PDA программы `shine_users`. + +Цель модуля: + +- при старте получить актуальное состояние всех пользовательских PDA; +- дальше держать локальную копию в актуальном состоянии; +- хранить историю просмотренных транзакций синхронизации; +- хранить историю всех версий пользовательских PDA; +- уметь после рестарта продолжать синхронизацию без потери изменений; +- в будущем без большого переписывания встраиваться в основной SHiNE-сервер и блокировать дальнейший startup до входа в состояние `READY`. + +На текущем этапе модуль должен запускаться как отдельный Java-процесс со своим `main`, но внутренняя структура должна быть такой, чтобы потом его можно было перенести в сервер как обычный lifecycle-сервис. + +--- + +## Базовая идея + +Модуль использует два источника данных: + +1. realtime-поток через Solana WebSocket `programSubscribe` для программы `shine_users`; +2. историю транзакций через `getSignaturesForAddress(economy_config_pda)`. + +Ключевая договорённость: + +- каждая транзакция, которая создаёт или обновляет пользовательский `user_pda`, обязательно читает `users_economy_config_pda`; +- значит, история по `users_economy_config_pda` является полным журналом всех релевантных `create/update user_pda` транзакций; +- дополнительные транзакции, которые тоже читают или меняют `users_economy_config_pda`, допустимы и не мешают: модуль должен уметь распознавать, что такая транзакция не относится к изменению пользовательского PDA. + +`users_economy_config_pda` вычисляется из: + +- `program_id = SHiNEPr1APdAgNBteUyBXcNovaHctpSjUu8oH2ZJdN6` +- seed = `shine_users_economy_config` + +Источник в коде контракта: + +- `settings::USERS_ECONOMY_CONFIG_SEED = b"shine_users_economy_config"` +- `create_user_pda` и `update_user_pda` принимают `users_economy_config_pda` как обязательный аккаунт. + +--- + +## Что считается состоянием готовности + +Модуль должен иметь внутреннее состояние готовности `READY`. + +В `READY` он входит только после того, как: + +1. установлено websocket-подключение; +2. выполнена начальная актуализация истории; +3. локальная таблица актуальных PDA приведена в консистентное состояние; +4. таблица состояния синхронизации обновлена. + +Пока модуль не вошёл в `READY`, будущий сервер при встраивании не должен продолжать собственный startup. + +Для этого модуль должен поддерживать: + +- `start()` +- `awaitReady()` +- `isReady()` +- `close()` + +Для отдельного процесса `main` допустимо также печатать явный лог о входе в `READY`, но главным механизмом для будущего сервера должен быть Java API, а не парсинг логов. + +--- + +## Почему нельзя опираться только на WebSocket + +WebSocket `programSubscribe` хорош для realtime, но недостаточен как единственный источник истины: + +- если websocket временно оборвался, часть событий может быть пропущена; +- если соединение формально живо, но какое-то событие было потеряно, это не всегда можно заметить сразу; +- если в интервале не было транзакций, websocket по определению ничего не пришлёт. + +Поэтому нужен второй защитный контур: + +- периодический аудит истории по `users_economy_config_pda`; +- запуск такого аудита раз в 5 минут. + +--- + +## Общий lifecycle модуля + +### 1. Старт процесса + +При старте процесса модуль: + +1. читает конфиг; +2. инициализирует локальную PostgreSQL БД; +3. создаёт RPC-клиент; +4. создаёт WebSocket-клиент; +5. создаёт coordinator/service слой; +6. запускает realtime-подписку; +7. после подтверждённой подписки выполняет начальную актуализацию истории; +8. после успешной актуализации выставляет `READY`. + +### 2. Работа в фоне + +После входа в `READY` модуль одновременно: + +- принимает realtime-обновления по websocket; +- раз в 5 минут запускает проверку истории через RPC; +- пишет журнал транзакций; +- пишет историю версий пользовательских PDA; +- обновляет актуальный снимок пользовательских PDA. + +### 3. Реконнект + +Если websocket оборвался: + +1. запускается reconnect; +2. после успешного переподключения выполняется повторная актуализация истории; +3. модуль снова возвращается в нормальный режим. + +### 4. Остановка + +При остановке: + +- закрывается websocket; +- останавливаются фоновые scheduler/worker потоки; +- закрывается RPC-клиент; +- закрывается БД. + +--- + +## Источник истории: только по сигнатурам + +Основной checkpoint модуля должен строиться по сигнатурам транзакций, а не по времени. + +Хранить нужно: + +- последнюю просмотренную сигнатуру; +- последний просмотренный слот; +- последнюю релевантную сигнатуру; +- время последней успешной актуализации. + +Время хранится только как вспомогательная диагностика. Продолжение истории должно идти по сигнатурам и слотам. + +--- + +## Поведение начальной актуализации + +### Сценарий A: локальная БД пустая + +Если локальная БД ещё не содержит синхронизированных данных: + +1. выполняется первичная загрузка пользовательских PDA; +2. после этого фиксируется начальная точка истории; +3. модуль переходит в `READY`. + +Предпочтительный вариант: + +- использовать историю транзакций по `users_economy_config_pda` как основной механизм синхронизации; +- full snapshot всех program accounts остаётся аварийным fallback, а не штатным путём. + +### Сценарий B: локальная БД уже есть + +Если БД не пуста: + +1. берётся последняя обработанная сигнатура; +2. через `getSignaturesForAddress(users_economy_config_pda)` вытягивается история после неё; +3. новые транзакции разбираются и применяются; +4. после этого модуль входит в `READY`. + +### Сценарий C: новых транзакций не было + +Если после последней сигнатуры новых транзакций нет: + +- это не ошибка; +- модуль всё равно обновляет `last_poll_at` и `last_successful_poll_at`; +- фиксирует, что актуализация успешно проверена; +- может переходить в `READY`. + +--- + +## Периодическая актуализация раз в 5 минут + +Каждые 5 минут модуль должен: + +1. вызвать `getSignaturesForAddress(users_economy_config_pda)`; +2. получить новые сигнатуры после последней сохранённой точки; +3. сохранить все найденные транзакции в журнал истории; +4. выделить релевантные транзакции `create/update user_pda`; +5. для релевантных транзакций извлечь адрес PDA и логин; +6. подтянуть актуальное состояние затронутых PDA; +7. обновить: + - `solana_user_pda_current` + - `solana_user_pda_history` + - `solana_sync_state` + +Если новых транзакций нет: + +- модуль ничего не меняет в зеркале PDA; +- но помечает, что polling выполнен успешно и состояние истории актуализировано. + +--- + +## Что делать с нерелевантными транзакциями + +История должна храниться полностью, включая нерелевантные транзакции. + +Примеры нерелевантных транзакций: + +- `update_users_economy_config` +- служебные транзакции, где `economy_config_pda` присутствовал, но пользовательский `user_pda` не менялся + +Такие транзакции: + +- сохраняются в `solana_sync_tx_history`; +- помечаются `is_relevant = 0`; +- не приводят к обновлению пользовательских PDA. + +Это важно для аудита и отладки. + +--- + +## Как распознавать тип транзакции + +Для каждой транзакции из истории нужно определить её тип. + +Минимальный набор типов: + +- `create_user_pda` +- `update_user_pda` +- `update_users_economy_config` +- `init_users_economy_config` +- `other` + +Также для каждой транзакции нужно определять: + +- `is_relevant = 1`, если транзакция создаёт или обновляет пользовательский `user_pda`; +- `is_relevant = 0`, если это транзакция истории, но она не меняет пользовательские PDA. + +Для релевантных транзакций нужно дополнительно извлекать: + +- `affected_pda_address` +- `affected_login` + +Если транзакция затрагивает несколько пользовательских PDA, архитектура должна не запрещать хранить несколько связей, но на первом этапе можно исходить из одной пользовательской записи на одну транзакцию, если это соответствует текущему контракту. + +--- + +## Локальные таблицы + +Модуль должен использовать три основные таблицы. + +### 1. `solana_sync_state` + +Одна строка состояния синхронизации. + +Назначение: + +- хранить текущий checkpoint истории; +- хранить время последней успешной актуализации; +- хранить технический статус синка. + +Пример полей: + +- `id INTEGER PRIMARY KEY CHECK (id = 1)` +- `status TEXT NOT NULL` +- `ready INTEGER NOT NULL DEFAULT 0` +- `last_poll_at_ms INTEGER` +- `last_successful_poll_at_ms INTEGER` +- `last_seen_signature TEXT` +- `last_seen_slot INTEGER` +- `last_relevant_signature TEXT` +- `last_relevant_slot INTEGER` +- `last_error TEXT` +- `updated_at_ms INTEGER NOT NULL` + +### 2. `solana_sync_tx_history` + +Append-only журнал всех просмотренных транзакций по `users_economy_config_pda`. + +Назначение: + +- хранить полную историю опроса; +- фиксировать, какие tx были релевантны; +- хранить связь tx -> пользовательский PDA / login; +- упрощать аудит и диагностику. + +Пример полей: + +- `signature TEXT PRIMARY KEY` +- `slot INTEGER NOT NULL` +- `block_time INTEGER` +- `tx_kind TEXT NOT NULL` +- `is_relevant INTEGER NOT NULL` +- `affected_pda_address TEXT` +- `affected_login TEXT` +- `processed_at_ms INTEGER NOT NULL` +- `raw_summary_json TEXT NOT NULL` + +Индексы: + +- по `slot` +- по `is_relevant` +- по `affected_login` +- по `affected_pda_address` + +### 3. `solana_user_pda_current` + +Текущее актуальное состояние каждого пользовательского PDA. + +Назначение: + +- быстрый lookup текущих данных пользователя; +- будущая интеграция с сервером; +- опорная таблица для поиска текущих ключей и полей PDA. + +Пример полей: + +- `pda_address TEXT PRIMARY KEY` +- `login TEXT NOT NULL` +- `record_number INTEGER NOT NULL` +- `slot INTEGER NOT NULL` +- `last_tx_signature TEXT NOT NULL` +- `blockchain_name TEXT NOT NULL` +- `blockchain_key TEXT NOT NULL` +- `client_key TEXT NOT NULL` +- `paid_limit_bytes INTEGER NOT NULL` +- `used_bytes INTEGER NOT NULL` +- `last_block_number INTEGER NOT NULL` +- `last_block_hash TEXT` +- `arweave_tx_id TEXT` +- `is_server INTEGER NOT NULL DEFAULT 0` +- `server_address TEXT` +- `sync_servers_json TEXT NOT NULL DEFAULT '[]'` +- `access_servers_json TEXT NOT NULL DEFAULT '[]'` +- `sessions_json TEXT NOT NULL DEFAULT '[]'` +- `trusted_count INTEGER NOT NULL DEFAULT 0` +- `created_at_ms INTEGER NOT NULL` +- `updated_at_ms INTEGER NOT NULL` +- `raw_data_base64 TEXT NOT NULL` +- `first_seen_at_ms INTEGER NOT NULL` +- `last_synced_at_ms INTEGER NOT NULL` + +Индексы: + +- уникальный индекс на `login` +- индекс на `slot` +- индекс на `last_tx_signature` + +### 4. `solana_user_pda_history` + +Append-only история всех версий пользовательских PDA. + +Назначение: + +- хранить все старые публичные ключи и прочие поля прошлых версий; +- позволять видеть, когда и какая версия записи была актуальна; +- позволять разбирать изменения пользователя во времени. + +Пример полей: + +- `id INTEGER PRIMARY KEY` +- `tx_signature TEXT NOT NULL` +- `slot INTEGER NOT NULL` +- `block_time INTEGER` +- `pda_address TEXT NOT NULL` +- `login TEXT NOT NULL` +- `record_number INTEGER NOT NULL` +- `blockchain_name TEXT NOT NULL` +- `blockchain_key TEXT NOT NULL` +- `client_key TEXT NOT NULL` +- `paid_limit_bytes INTEGER NOT NULL` +- `used_bytes INTEGER NOT NULL` +- `last_block_number INTEGER NOT NULL` +- `last_block_hash TEXT` +- `arweave_tx_id TEXT` +- `is_server INTEGER NOT NULL DEFAULT 0` +- `server_address TEXT` +- `sync_servers_json TEXT NOT NULL DEFAULT '[]'` +- `access_servers_json TEXT NOT NULL DEFAULT '[]'` +- `sessions_json TEXT NOT NULL DEFAULT '[]'` +- `trusted_count INTEGER NOT NULL DEFAULT 0` +- `created_at_ms INTEGER NOT NULL` +- `updated_at_ms INTEGER NOT NULL` +- `raw_data_base64 TEXT NOT NULL` +- `saved_at_ms INTEGER NOT NULL` + +Индексы: + +- уникальный индекс на `(pda_address, record_number)` +- индекс на `login` +- индекс на `slot` +- индекс на `tx_signature` + +--- + +## Зачем нужны и `current`, и `history` + +Нужны обе таблицы: + +- `solana_user_pda_current` хранит только последнюю актуальную версию и удобна для быстрых запросов; +- `solana_user_pda_history` хранит все версии и нужна для расследований и просмотра старых публичных ключей. + +При обработке новой релевантной транзакции: + +1. новая версия всегда добавляется в `solana_user_pda_history`; +2. актуальная строка в `solana_user_pda_current` вставляется или обновляется. + +--- + +## Что именно считается “историей пользователя” + +История нужна не для секретных паролей, а для публичных данных PDA. + +В частности, история должна позволять видеть старые значения: + +- `client_key` +- `blockchain_key` +- `sessions` +- `access_servers` +- `sync_servers` +- `server_address` +- других публичных полей PDA + +Секретные приватные ключи или настоящие пользовательские пароли этот модуль не хранит и хранить не должен. + +--- + +## Логирование + +Модуль должен использовать нормальный logger, а не `System.out.println`. + +Требования к логированию: + +- совместимость с будущей интеграцией в общие логи сервера; +- явные уровни `INFO`, `WARN`, `ERROR`, `DEBUG`; +- короткие, но диагностичные сообщения; +- каждый важный переход состояния должен логироваться. + +Что обязательно логировать: + +- старт процесса; +- чтение конфига; +- вычисление `users_economy_config_pda`; +- старт websocket-подписки; +- успешную подписку; +- начало initial sync; +- окончание initial sync; +- вход в `READY`; +- periodic poll; +- число найденных сигнатур; +- число релевантных tx; +- число обновлённых PDA; +- реконнекты websocket; +- ошибки RPC/WS; +- переход в `FAILED`, если он будет. + +--- + +## Конфигурация + +В конфиге должны остаться только обязательные для постоянной эксплуатации параметры: + +```text +SOLANA_RPC_URL= +SOLANA_WS_URL= +SOLANA_PROGRAM_ID=SHiNEPr1APdAgNBteUyBXcNovaHctpSjUu8oH2ZJdN6 + +DATABASE_URL=jdbc:postgresql://127.0.0.1:5432/shine_server_db +PGUSER=shine_server +PGPASSWORD= +SYNC_POLL_INTERVAL_SECONDS=300 +``` + +Дополнительно: + +- `SOLANA_COMMITMENT` не нужен как параметр; +- commitment должен быть зафиксирован в коде как `confirmed`; +- `HELIUS_API_KEY` не нужен; +- `SOLANA_NETWORK` не нужен. + +--- + +## Поведение при ошибках + +### Ошибка periodic poll + +Если periodic poll не удался: + +- это логируется как `WARN` или `ERROR`; +- `solana_sync_state.last_error` обновляется; +- модуль не должен сразу завершаться, если reconnect/retry ещё возможны. + +### Ошибка initial sync + +Если initial sync не удался: + +- модуль не должен выставлять `READY`; +- отдельный процесс должен завершаться с ошибкой или оставаться в `FAILED`, в зависимости от выбранного runtime-режима; +- при встраивании в сервер основной startup должен считаться неуспешным. + +### Аварийный fallback + +Даже если основная логика опирается на историю по `users_economy_config_pda`, аварийный full snapshot всех `user_pda` можно оставить как последний защитный fallback на случай повреждённого или неполного RPC-ответа. + +Но это должен быть именно крайний защитный сценарий, а не штатный рабочий путь. + +--- + +## Почему хранить всю tx-историю полезно + +Полный журнал `solana_sync_tx_history` нужен не только для самого синка, но и для эксплуатации: + +- видно, что именно вернул RPC; +- видно, какие tx были признаны релевантными; +- видно, какие tx были проигнорированы и почему; +- можно поднимать старые кейсы без повторного запроса в Solana; +- упрощается аудит и отладка после инцидентов. + +--- + +## Что должно получиться в итоге + +В результате модуль должен работать так: + +1. стартует как отдельный Java-процесс; +2. подписывается на realtime-обновления `shine_users`; +3. делает начальную актуализацию через историю `users_economy_config_pda`; +4. входит в `READY`; +5. раз в 5 минут делает дополнительный audit истории; +6. хранит: + - состояние синка; + - полную историю просмотренных транзакций; + - актуальное зеркало всех пользовательских PDA; + - историю всех версий пользовательских PDA; +7. после будущего переноса в SHiNE-сервер может использоваться как блокирующий startup-модуль: + - сначала синхронизируется Solana; + - потом сервер продолжает запуск остальных подсистем. + +--- + +## Следующий шаг реализации + +После утверждения этого документа модуль нужно доработать в коде: + +1. упростить конфиг; +2. заменить текущее логирование на logger; +3. выделить lifecycle-сервис с `awaitReady()`; +4. реализовать вычисление `users_economy_config_pda`; +5. реализовать polling истории по сигнатурам; +6. добавить новые SQLite-таблицы; +7. добавить запись в `current` и `history`; +8. добавить periodic guard раз в 5 минут; +9. сохранить отдельный `main` для запуска как процесса. diff --git a/settings.gradle b/settings.gradle index 32854cbf..008cebfb 100644 --- a/settings.gradle +++ b/settings.gradle @@ -8,6 +8,7 @@ include 'shine-server-blockchain' include 'shine-server-db' include 'shine-server-net-protocol' include 'shine-server-net-server' +include 'shine-server-solana-users-sync' project(':shine-server-log').projectDir = file('SHiNE-server/shine-server-log') project(':shine-server-config').projectDir = file('SHiNE-server/shine-server-config') @@ -17,3 +18,4 @@ project(':shine-server-blockchain').projectDir = file('SHiNE-server/shine-server project(':shine-server-db').projectDir = file('SHiNE-server/shine-server-db') project(':shine-server-net-protocol').projectDir = file('SHiNE-server/shine-server-net-protocol') project(':shine-server-net-server').projectDir = file('SHiNE-server/shine-server-net-server') +project(':shine-server-solana-users-sync').projectDir = file('SHiNE-server/shine-server-solana-users-sync')