From 02a849089c35ef4626b1d7f5e59b918411954b7a Mon Sep 17 00:00:00 2001 From: peradas Date: Tue, 21 Jul 2026 13:19:40 +0300 Subject: [PATCH] =?UTF-8?q?=D1=81=D0=B8=D0=BD=D1=85=D1=80=D0=BE=D0=BD?= =?UTF-8?q?=D0=B8=D0=B7=D0=B0=D1=86=D0=B8=D1=8F=20=D1=81=20=D1=81=D0=BE?= =?UTF-8?q?=D0=BB=D0=B0=D0=BD=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 21 + .gitignore | 7 + PATCH_NOTES.md | 23 + README.md | 83 ++ pom.xml | 56 ++ src/main/java/sync/Main.java | 148 ++++ src/main/java/sync/config/AppConfig.java | 136 ++++ src/main/java/sync/core/SyncCheckpoint.java | 9 + src/main/java/sync/core/SyncCoordinator.java | 379 +++++++++ src/main/java/sync/core/SyncMode.java | 8 + src/main/java/sync/core/SyncState.java | 10 + .../java/sync/model/ProgramAccountUpdate.java | 12 + src/main/java/sync/model/RecoveryResult.java | 44 ++ src/main/java/sync/model/SnapshotResult.java | 9 + .../sync/source/AccountUpdateListener.java | 11 + .../java/sync/source/ConnectionListener.java | 12 + .../java/sync/source/SolanaDataSource.java | 25 + .../sync/source/rpc/RpcSolanaDataSource.java | 241 ++++++ .../java/sync/source/rpc/SolanaRpcClient.java | 716 ++++++++++++++++++ .../source/rpc/SolanaWebSocketClient.java | 541 +++++++++++++ .../java/sync/storage/StorageRepository.java | 57 ++ .../sqlite/SQLiteStorageRepository.java | 552 ++++++++++++++ 22 files changed, 3100 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 PATCH_NOTES.md create mode 100644 README.md create mode 100644 pom.xml create mode 100644 src/main/java/sync/Main.java create mode 100644 src/main/java/sync/config/AppConfig.java create mode 100644 src/main/java/sync/core/SyncCheckpoint.java create mode 100644 src/main/java/sync/core/SyncCoordinator.java create mode 100644 src/main/java/sync/core/SyncMode.java create mode 100644 src/main/java/sync/core/SyncState.java create mode 100644 src/main/java/sync/model/ProgramAccountUpdate.java create mode 100644 src/main/java/sync/model/RecoveryResult.java create mode 100644 src/main/java/sync/model/SnapshotResult.java create mode 100644 src/main/java/sync/source/AccountUpdateListener.java create mode 100644 src/main/java/sync/source/ConnectionListener.java create mode 100644 src/main/java/sync/source/SolanaDataSource.java create mode 100644 src/main/java/sync/source/rpc/RpcSolanaDataSource.java create mode 100644 src/main/java/sync/source/rpc/SolanaRpcClient.java create mode 100644 src/main/java/sync/source/rpc/SolanaWebSocketClient.java create mode 100644 src/main/java/sync/storage/StorageRepository.java create mode 100644 src/main/java/sync/storage/sqlite/SQLiteStorageRepository.java diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..9316478 --- /dev/null +++ b/.env.example @@ -0,0 +1,21 @@ +# Universal Solana RPC configuration. +# +# Helius example: +# SOLANA_RPC_URL=https://devnet.helius-rpc.com/?api-key=YOUR_KEY +# SOLANA_WS_URL=wss://devnet.helius-rpc.com/?api-key=YOUR_KEY +# +# Standard/self-hosted Solana RPC example: +# SOLANA_RPC_URL=http://127.0.0.1:8899 +# SOLANA_WS_URL=ws://127.0.0.1:8900 + +SOLANA_RPC_URL= +SOLANA_WS_URL= + +SOLANA_PROGRAM_ID=put_your_program_id_here +SOLANA_COMMITMENT=confirmed +SQLITE_PATH=./data/solana.db + +# Temporary backward-compatible Helius fallback. +# These are used only when SOLANA_RPC_URL / SOLANA_WS_URL are absent. +HELIUS_API_KEY= +SOLANA_NETWORK=devnet diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f154394 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +target/ +data/*.db +data/*.db-shm +data/*.db-wal +.env +.idea/ +*.iml diff --git a/PATCH_NOTES.md b/PATCH_NOTES.md new file mode 100644 index 0000000..09a99b9 --- /dev/null +++ b/PATCH_NOTES.md @@ -0,0 +1,23 @@ +Нужно также добавить в существующий AppConfig: + +public String rpcUrl() { + String host = switch (network) { + case "devnet" -> "https://devnet.helius-rpc.com/"; + case "mainnet" -> "https://mainnet.helius-rpc.com/"; + default -> throw new IllegalStateException("Unsupported network: " + network); + }; + return host + "?api-key=" + heliusApiKey; +} + +И в HeliusWebSocketClient: + +1. Поле: +private final CompletableFuture subscribed = new CompletableFuture<>(); + +2. Метод: +public CompletableFuture subscribedFuture() { + return subscribed; +} + +3. В обработчике успешной подписки после лога: +subscribed.complete(null); diff --git a/README.md b/README.md new file mode 100644 index 0000000..d88ccaf --- /dev/null +++ b/README.md @@ -0,0 +1,83 @@ +# Helius PDA Sync Test + +Минимальное Java 21 приложение для проверки realtime-синхронизации: + +`Solana Program -> Helius WebSocket -> programSubscribe -> SQLite` + +## Что делает + +- подключается к Helius WebSocket; +- подписывается через `programSubscribe` на изменения аккаунтов программы; +- сохраняет каждый полученный аккаунт в SQLite; +- обновляет запись только если новый `slot` не старее сохранённого; +- автоматически переподключается после обрыва соединения. + +На этом этапе приложение хранит raw `base64` данные PDA и не декодирует Anchor/Borsh. + +## Требования + +- Java 21+ +- Maven 3.9+ + +## Настройка + +```bash +export HELIUS_API_KEY="your_key" +export SOLANA_PROGRAM_ID="your_program_id" +export SQLITE_PATH="./data/solana.db" +export SOLANA_COMMITMENT="confirmed" +``` + +## Запуск + +```bash +mvn clean compile exec:java +``` + +Ожидаемый лог: + +```text +Database initialized: ... +Connecting to Helius WebSocket... +WebSocket connected +Subscription request sent for program: ... +Subscribed successfully. Subscription id: ... +``` + +После создания или изменения PDA: + +```text +Account update: ... +Saved account ... +``` + +## Проверка SQLite + +```bash +sqlite3 data/solana.db +``` + +```sql +.headers on +.mode column + +SELECT + address, + owner, + lamports, + slot, + length(data_base64) AS base64_length, + updated_at +FROM program_accounts +ORDER BY slot DESC; +``` + +## Первый тест + +1. Запусти приложение. +2. Дождись `Subscribed successfully`. +3. Создай или измени PDA через контракт. +4. Убедись, что пришёл `programNotification`. +5. Проверь запись в SQLite. + +Следующий шаг после проверки realtime: добавить `getProgramAccounts` на старте и декодирование конкретного PDA по IDL. diff --git a/pom.xml b/pom.xml new file mode 100644 index 0000000..862753f --- /dev/null +++ b/pom.xml @@ -0,0 +1,56 @@ + + 4.0.0 + + com.example + helius-sync-test + 1.0.0 + + + 17 + UTF-8 + 4.12.0 + 2.17.2 + 3.46.0.0 + + + + + com.squareup.okhttp3 + okhttp + ${okhttp.version} + + + com.fasterxml.jackson.core + jackson-databind + ${jackson.version} + + + org.xerial + sqlite-jdbc + ${sqlite.version} + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + 3.13.0 + + 17 + + + + org.codehaus.mojo + exec-maven-plugin + 3.4.1 + + sync.Main + + + + + diff --git a/src/main/java/sync/Main.java b/src/main/java/sync/Main.java new file mode 100644 index 0000000..fb54222 --- /dev/null +++ b/src/main/java/sync/Main.java @@ -0,0 +1,148 @@ +package sync; + +import sync.config.AppConfig; +import sync.core.SyncCoordinator; +import sync.core.SyncMode; +import sync.source.ConnectionListener; +import sync.source.SolanaDataSource; +import sync.source.rpc.RpcSolanaDataSource; +import sync.storage.StorageRepository; +import sync.storage.sqlite.SQLiteStorageRepository; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; + +public final class Main { + + private Main() { + } + + public static void main( + String[] args + ) { + + try { + + AppConfig config = + AppConfig + .fromEnvironment(); + + StorageRepository storage = + new SQLiteStorageRepository( + config.sqlitePath() + ); + + SolanaDataSource dataSource = + new RpcSolanaDataSource( + config.rpcUrl(), + config.websocketUrl(), + config.programId(), + config.commitment() + ); + + SyncCoordinator syncCoordinator = + new SyncCoordinator( + dataSource, + storage + ); + + Runtime + .getRuntime() + .addShutdownHook( + new Thread( + () -> { + + System.out.println( + "Shutting down..." + ); + + try { + + dataSource.close(); + + storage.close(); + + } catch (Exception exception) { + + System.err.println( + "Shutdown error: " + + exception.getMessage() + ); + } + } + ) + ); + + dataSource.startRealtime( + + syncCoordinator + ::handleRealtimeUpdate, + + new ConnectionListener() { + + @Override + public void onConnected( + boolean firstConnection + ) { + + if (firstConnection) { + + CompletableFuture + .runAsync( + syncCoordinator + ::start + ); + + } else { + + CompletableFuture + .runAsync( + () -> + syncCoordinator + .runMode( + SyncMode + .INCREMENTAL_RECOVERY + ) + ); + } + } + + @Override + public void onDisconnected( + Throwable cause + ) { + + System.err.println( + "Solana realtime disconnected: " + + cause.getMessage() + ); + } + } + ); + + new CountDownLatch( + 1 + ).await(); + + } catch ( + InterruptedException exception + ) { + + Thread + .currentThread() + .interrupt(); + + } catch (Exception exception) { + + System.err.println( + "Application startup failed:" + ); + + exception.printStackTrace(); + + System.exit( + 1 + ); + } + } +} diff --git a/src/main/java/sync/config/AppConfig.java b/src/main/java/sync/config/AppConfig.java new file mode 100644 index 0000000..841185e --- /dev/null +++ b/src/main/java/sync/config/AppConfig.java @@ -0,0 +1,136 @@ +package sync.config; + +import java.nio.file.Path; + +public record AppConfig( + String rpcUrl, + String websocketUrl, + String programId, + Path sqlitePath, + String commitment +) { + public static AppConfig fromEnvironment() { + String programId = requireEnv("SOLANA_PROGRAM_ID"); + String sqlitePath = System.getenv().getOrDefault( + "SQLITE_PATH", + "./data/solana.db" + ); + String commitment = System.getenv().getOrDefault( + "SOLANA_COMMITMENT", + "confirmed" + ); + + if (!commitment.equals("processed") + && !commitment.equals("confirmed") + && !commitment.equals("finalized")) { + throw new IllegalArgumentException( + "SOLANA_COMMITMENT must be processed, confirmed or finalized" + ); + } + + String rpcUrl = optionalEnv("SOLANA_RPC_URL"); + String websocketUrl = optionalEnv("SOLANA_WS_URL"); + + // Backward-compatible fallback for the current Helius-based .env. + if (rpcUrl == null || websocketUrl == null) { + String heliusApiKey = optionalEnv("HELIUS_API_KEY"); + String network = System.getenv().getOrDefault( + "SOLANA_NETWORK", + "devnet" + ); + + if (heliusApiKey == null) { + throw new IllegalStateException( + "Set SOLANA_RPC_URL and SOLANA_WS_URL, " + + "or provide HELIUS_API_KEY for the compatibility fallback" + ); + } + + if (!network.equals("devnet") && !network.equals("mainnet")) { + throw new IllegalArgumentException( + "SOLANA_NETWORK must be devnet or mainnet" + ); + } + + if (rpcUrl == null) { + rpcUrl = heliusRpcUrl( + network, + heliusApiKey + ); + } + + if (websocketUrl == null) { + websocketUrl = heliusWebSocketUrl( + network, + heliusApiKey + ); + } + } + + return new AppConfig( + rpcUrl, + websocketUrl, + programId, + Path.of(sqlitePath), + commitment + ); + } + + private static String heliusRpcUrl( + String network, + String apiKey + ) { + return switch (network) { + case "devnet" -> + "https://devnet.helius-rpc.com/?api-key=" + apiKey; + case "mainnet" -> + "https://mainnet.helius-rpc.com/?api-key=" + apiKey; + default -> + throw new IllegalStateException( + "Unsupported network: " + network + ); + }; + } + + private static String heliusWebSocketUrl( + String network, + String apiKey + ) { + return switch (network) { + case "devnet" -> + "wss://devnet.helius-rpc.com/?api-key=" + apiKey; + case "mainnet" -> + "wss://mainnet.helius-rpc.com/?api-key=" + apiKey; + default -> + throw new IllegalStateException( + "Unsupported network: " + network + ); + }; + } + + private static String requireEnv( + String name + ) { + String value = optionalEnv(name); + + if (value == null) { + throw new IllegalStateException( + "Missing required environment variable: " + name + ); + } + + return value; + } + + private static String optionalEnv( + String name + ) { + String value = System.getenv(name); + + if (value == null || value.isBlank()) { + return null; + } + + return value.trim(); + } +} diff --git a/src/main/java/sync/core/SyncCheckpoint.java b/src/main/java/sync/core/SyncCheckpoint.java new file mode 100644 index 0000000..f4eb772 --- /dev/null +++ b/src/main/java/sync/core/SyncCheckpoint.java @@ -0,0 +1,9 @@ +package sync.core; + +public record SyncCheckpoint( + SyncState state, + Long lastObservedSlot, + Long lastReconciledSlot, + Long lastFullSyncSlot +) { +} \ No newline at end of file diff --git a/src/main/java/sync/core/SyncCoordinator.java b/src/main/java/sync/core/SyncCoordinator.java new file mode 100644 index 0000000..94fdbb9 --- /dev/null +++ b/src/main/java/sync/core/SyncCoordinator.java @@ -0,0 +1,379 @@ +package sync.core; + +import sync.model.ProgramAccountUpdate; +import sync.model.RecoveryResult; +import sync.model.SnapshotResult; +import sync.source.SolanaDataSource; +import sync.storage.StorageRepository; + +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; + +public final class SyncCoordinator { + + private final SolanaDataSource + dataSource; + + private final StorageRepository + storage; + + private final AtomicBoolean modeRunning = + new AtomicBoolean(false); + + public SyncCoordinator( + SolanaDataSource dataSource, + StorageRepository storage + ) { + + this.dataSource = + dataSource; + + this.storage = + storage; + } + + public void start() { + + try { + + boolean hasAccounts = + storage.hasAccounts(); + + SyncCheckpoint checkpoint = + storage + .getSyncCheckpoint(); + + if (!hasAccounts) { + + runMode( + SyncMode.FULL_BOOTSTRAP + ); + + return; + } + + if ( + checkpoint + .lastReconciledSlot() + != null + ) { + + runMode( + SyncMode.INCREMENTAL_RECOVERY + ); + + return; + } + + runMode( + SyncMode.FULL_RECONCILIATION + ); + + } catch (Exception exception) { + + markFailed(); + + throw new RuntimeException( + "Sync coordinator failed", + exception + ); + } + } + + public void handleRealtimeUpdate( + ProgramAccountUpdate update + ) { + + try { + + storage.upsertAccount( + update + ); + + /* + * Realtime advances only the observed boundary. + * + * A WebSocket event at slot 5000 does not prove + * that every relevant event before 5000 arrived. + */ + storage.updateLastObservedSlot( + update.slot() + ); + + } catch (Exception exception) { + + markFailed(); + + throw new RuntimeException( + "Realtime update failed", + exception + ); + } + } + + public void runMode( + SyncMode mode + ) { + + if (!modeRunning.compareAndSet( + false, + true + )) { + + System.out.println( + "Sync mode skipped because another mode is running: " + + mode + ); + + return; + } + + try { + + switch (mode) { + + case FULL_BOOTSTRAP -> + fullBootstrap(); + + case REALTIME_SYNC -> + realtimeSync(); + + case INCREMENTAL_RECOVERY -> + incrementalRecovery(); + + case FULL_RECONCILIATION -> + fullReconciliation(); + } + + } catch (Exception exception) { + + markFailed(); + + throw new RuntimeException( + "Sync mode failed: " + + mode, + exception + ); + + } finally { + + modeRunning.set( + false + ); + } + } + + private void fullBootstrap() + throws Exception { + + System.out.println( + "Sync mode: FULL_BOOTSTRAP" + ); + + storage.updateSyncState( + SyncState.BOOTSTRAPPING + ); + + SnapshotResult snapshot = + dataSource + .loadFullSnapshot(); + + storage.upsertAccounts( + snapshot.accounts() + ); + + storage.updateLastObservedSlot( + snapshot.snapshotSlot() + ); + + storage.updateLastReconciledSlot( + snapshot.snapshotSlot() + ); + + storage.updateLastFullSyncSlot( + snapshot.snapshotSlot() + ); + + storage.updateSyncState( + SyncState.LIVE + ); + + System.out.println( + "FULL_BOOTSTRAP finished" + ); + } + + private void realtimeSync() + throws Exception { + + System.out.println( + "Sync mode: REALTIME_SYNC" + ); + + storage.updateSyncState( + SyncState.LIVE + ); + } + + private void incrementalRecovery() + throws Exception { + + System.out.println( + "Sync mode: INCREMENTAL_RECOVERY" + ); + + storage.updateSyncState( + SyncState.RECOVERING + ); + + SyncCheckpoint checkpoint = + storage + .getSyncCheckpoint(); + + Long lastReconciledSlot = + checkpoint + .lastReconciledSlot(); + + if (lastReconciledSlot == null) { + + System.out.println( + "No reconciled checkpoint available. " + + "Falling back to full reconciliation." + ); + + fullReconciliation(); + + return; + } + + RecoveryResult recovery = + dataSource + .recoverFrom( + lastReconciledSlot + ); + + if (!recovery.complete()) { + + System.out.println( + "Incremental recovery incomplete. reason=" + + recovery.fallbackReason() + + ". Falling back to full reconciliation." + ); + + fullReconciliation(); + + return; + } + + /* + * Apply current states of all program-owned accounts + * discovered during recovery. + */ + storage.upsertAccounts( + recovery.updates() + ); + + /* + * missingAddresses means only: + * + * "this touched address no longer exists on-chain". + * + * It may have been an unrelated temporary account. + * + * Only addresses already present in our local + * program_accounts projection are safe to delete. + */ + List confirmedDeletedAddresses = + storage.findExistingAddresses( + recovery.missingAddresses() + ); + + storage.deleteAccounts( + confirmedDeletedAddresses + ); + + /* + * Only now can we advance the trusted continuous + * recovery boundary. + */ + storage.updateLastReconciledSlot( + recovery.toSlot() + ); + + /* + * Do not lower lastObservedSlot. + * + * updateLastObservedSlot() uses MAX semantics, + * so realtime may already have moved it further. + */ + storage.updateLastObservedSlot( + recovery.toSlot() + ); + + storage.updateSyncState( + SyncState.LIVE + ); + + System.out.printf( + "INCREMENTAL_RECOVERY finished. " + + "from=%d to=%d updates=%d missing=%d deleted=%d%n", + recovery.fromSlot(), + recovery.toSlot(), + recovery.updates().size(), + recovery.missingAddresses().size(), + confirmedDeletedAddresses.size() + ); + } + + private void fullReconciliation() + throws Exception { + + System.out.println( + "Sync mode: FULL_RECONCILIATION" + ); + + storage.updateSyncState( + SyncState.RECONCILING + ); + + SnapshotResult snapshot = + dataSource + .loadFullSnapshot(); + + storage.upsertAccounts( + snapshot.accounts() + ); + + storage.updateLastObservedSlot( + snapshot.snapshotSlot() + ); + + storage.updateLastReconciledSlot( + snapshot.snapshotSlot() + ); + + storage.updateLastFullSyncSlot( + snapshot.snapshotSlot() + ); + + storage.updateSyncState( + SyncState.LIVE + ); + + System.out.println( + "FULL_RECONCILIATION finished" + ); + } + + private void markFailed() { + + try { + + storage.updateSyncState( + SyncState.FAILED + ); + + } catch (Exception ignored) { + } + } +} \ No newline at end of file diff --git a/src/main/java/sync/core/SyncMode.java b/src/main/java/sync/core/SyncMode.java new file mode 100644 index 0000000..2f5264e --- /dev/null +++ b/src/main/java/sync/core/SyncMode.java @@ -0,0 +1,8 @@ +package sync.core; + +public enum SyncMode { + FULL_BOOTSTRAP, + REALTIME_SYNC, + INCREMENTAL_RECOVERY, + FULL_RECONCILIATION +} diff --git a/src/main/java/sync/core/SyncState.java b/src/main/java/sync/core/SyncState.java new file mode 100644 index 0000000..6f92dfb --- /dev/null +++ b/src/main/java/sync/core/SyncState.java @@ -0,0 +1,10 @@ +package sync.core; + +public enum SyncState { + EMPTY, + BOOTSTRAPPING, + LIVE, + RECOVERING, + RECONCILING, + FAILED +} diff --git a/src/main/java/sync/model/ProgramAccountUpdate.java b/src/main/java/sync/model/ProgramAccountUpdate.java new file mode 100644 index 0000000..8b4c00c --- /dev/null +++ b/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/src/main/java/sync/model/RecoveryResult.java b/src/main/java/sync/model/RecoveryResult.java new file mode 100644 index 0000000..de605e2 --- /dev/null +++ b/src/main/java/sync/model/RecoveryResult.java @@ -0,0 +1,44 @@ +package sync.model; + +import java.util.List; + +public record RecoveryResult( + boolean complete, + long fromSlot, + long toSlot, + List updates, + List missingAddresses, + String fallbackReason +) { + + public static RecoveryResult complete( + long fromSlot, + long toSlot, + List updates, + List missingAddresses + ) { + return new RecoveryResult( + true, + fromSlot, + toSlot, + updates, + missingAddresses, + null + ); + } + + public static RecoveryResult incomplete( + long fromSlot, + long toSlot, + String fallbackReason + ) { + return new RecoveryResult( + false, + fromSlot, + toSlot, + List.of(), + List.of(), + fallbackReason + ); + } +} \ No newline at end of file diff --git a/src/main/java/sync/model/SnapshotResult.java b/src/main/java/sync/model/SnapshotResult.java new file mode 100644 index 0000000..83a917e --- /dev/null +++ b/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/src/main/java/sync/source/AccountUpdateListener.java b/src/main/java/sync/source/AccountUpdateListener.java new file mode 100644 index 0000000..e3a5932 --- /dev/null +++ b/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/src/main/java/sync/source/ConnectionListener.java b/src/main/java/sync/source/ConnectionListener.java new file mode 100644 index 0000000..c152aff --- /dev/null +++ b/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/src/main/java/sync/source/SolanaDataSource.java b/src/main/java/sync/source/SolanaDataSource.java new file mode 100644 index 0000000..9d48481 --- /dev/null +++ b/src/main/java/sync/source/SolanaDataSource.java @@ -0,0 +1,25 @@ +package sync.source; + +import sync.model.RecoveryResult; +import sync.model.SnapshotResult; + +public interface SolanaDataSource + extends AutoCloseable { + + SnapshotResult loadFullSnapshot() + throws Exception; + + void startRealtime( + AccountUpdateListener accountUpdateListener, + ConnectionListener connectionListener + ) throws Exception; + + void stopRealtime(); + + RecoveryResult recoverFrom( + long lastProcessedSlot + ) throws Exception; + + long getCurrentSlot() + throws Exception; +} diff --git a/src/main/java/sync/source/rpc/RpcSolanaDataSource.java b/src/main/java/sync/source/rpc/RpcSolanaDataSource.java new file mode 100644 index 0000000..be584ae --- /dev/null +++ b/src/main/java/sync/source/rpc/RpcSolanaDataSource.java @@ -0,0 +1,241 @@ +package sync.source.rpc; + +import sync.model.RecoveryResult; +import sync.model.SnapshotResult; +import sync.source.AccountUpdateListener; +import sync.source.ConnectionListener; +import sync.source.SolanaDataSource; + +import java.io.IOException; +import java.util.List; +import java.util.Set; + +public final class RpcSolanaDataSource + implements SolanaDataSource { + + private final SolanaRpcClient + rpcClient; + + private final SolanaWebSocketClient + webSocketClient; + + public RpcSolanaDataSource( + String rpcUrl, + String websocketUrl, + String programId, + String commitment + ) { + + this.rpcClient = + new SolanaRpcClient( + rpcUrl, + programId, + commitment + ); + + this.webSocketClient = + new SolanaWebSocketClient( + websocketUrl, + programId, + commitment + ); + } + + @Override + public SnapshotResult loadFullSnapshot() + throws Exception { + + return rpcClient + .loadFullSnapshot(); + } + + @Override + public void startRealtime( + AccountUpdateListener accountUpdateListener, + ConnectionListener connectionListener + ) { + + webSocketClient.start( + accountUpdateListener, + connectionListener + ); + } + + @Override + public void stopRealtime() { + + webSocketClient.stop(); + } + + @Override + public RecoveryResult recoverFrom( + long lastReconciledSlot + ) throws Exception { + + /* + * Fix the upper boundary at the beginning. + * + * Realtime is already active and will handle + * anything newer than targetSlot. + */ + long targetSlot = + rpcClient + .getCurrentSlot(); + + System.out.printf( + "Incremental recovery requested. fromSlot=%d targetSlot=%d%n", + lastReconciledSlot, + targetSlot + ); + + if (targetSlot + <= lastReconciledSlot) { + + return RecoveryResult.complete( + lastReconciledSlot, + targetSlot, + List.of(), + List.of() + ); + } + + long firstAvailableBlock = + rpcClient + .getFirstAvailableBlock(); + + if (firstAvailableBlock < 0) { + + return RecoveryResult.incomplete( + lastReconciledSlot, + targetSlot, + "cannot_resolve_first_available_block" + ); + } + + /* + * The RPC node has already pruned the checkpoint + * range. Incremental recovery cannot be proven. + */ + if (lastReconciledSlot + < firstAvailableBlock) { + + return RecoveryResult.incomplete( + lastReconciledSlot, + targetSlot, + "checkpoint_older_than_rpc_history" + ); + } + + try { + + List + signatures = + rpcClient + .getSignaturesAfterSlot( + lastReconciledSlot, + targetSlot + ); + + System.out.printf( + "Recovery transactions found: %d%n", + signatures.size() + ); + + if (signatures.isEmpty()) { + + return RecoveryResult.complete( + lastReconciledSlot, + targetSlot, + List.of(), + List.of() + ); + } + + Set touchedAddresses = + rpcClient + .getTouchedAddresses( + signatures + ); + + System.out.printf( + "Recovery unique touched addresses: %d%n", + touchedAddresses.size() + ); + + if (touchedAddresses.isEmpty()) { + + return RecoveryResult.complete( + lastReconciledSlot, + targetSlot, + List.of(), + List.of() + ); + } + + SolanaRpcClient.AccountBatchResult + accountBatchResult = + rpcClient + .getCurrentAccounts( + touchedAddresses, + targetSlot + ); + + System.out.printf( + "Recovery state resolved: updates=%d missing=%d%n", + accountBatchResult + .updates() + .size(), + accountBatchResult + .missingAddresses() + .size() + ); + + return RecoveryResult.complete( + lastReconciledSlot, + targetSlot, + accountBatchResult + .updates(), + accountBatchResult + .missingAddresses() + ); + + } catch (IOException exception) { + + /* + * Missing transaction history, unsupported + * transaction format or unavailable RPC history + * means we cannot claim a complete recovery. + * + * Coordinator will fall back to a full snapshot. + */ + + System.err.println( + "Incremental recovery RPC failure: " + + exception.getMessage() + ); + + return RecoveryResult.incomplete( + lastReconciledSlot, + targetSlot, + "rpc_history_recovery_failed: " + + exception.getMessage() + ); + } + } + + @Override + public long getCurrentSlot() + throws Exception { + + return rpcClient + .getCurrentSlot(); + } + + @Override + public void close() { + + webSocketClient.close(); + + rpcClient.close(); + } +} \ No newline at end of file diff --git a/src/main/java/sync/source/rpc/SolanaRpcClient.java b/src/main/java/sync/source/rpc/SolanaRpcClient.java new file mode 100644 index 0000000..84c1c0d --- /dev/null +++ b/src/main/java/sync/source/rpc/SolanaRpcClient.java @@ -0,0 +1,716 @@ +package sync.source.rpc; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import okhttp3.*; +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 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 { + + System.out.println( + "Full snapshot loading started..." + ); + + 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); + + JsonNode result = + root.path("result"); + + long snapshotSlot = + result + .path("context") + .path("slot") + .asLong(-1); + + if (snapshotSlot < 0) { + throw new IOException( + "Missing context.slot" + ); + } + + 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 + ) + ); + } + + System.out.printf( + "Full snapshot loaded. accounts=%d snapshotSlot=%d%n", + accounts.size(), + snapshotSlot + ); + + return new SnapshotResult( + snapshotSlot, + accounts + ); + } + + 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 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", 103, + "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; + } + + /* + * We reached the already reconciled range. + */ + if (slot <= fromSlot) { + + lowerBoundaryReached = + true; + + break; + } + + /* + * Realtime is already running. + * + * Recovery covers only: + * + * (fromSlot, targetSlot] + */ + if (slot > targetSlot) { + continue; + } + + String signature = + item.path("signature") + .asText(); + + if (signature.isBlank()) { + continue; + } + + /* + * Failed transactions cannot have produced + * successful account state changes. + */ + JsonNode error = + item.get("err"); + + if (error != null + && !error.isNull()) { + + continue; + } + + signatures.add( + new SignatureInfo( + signature, + slot + ) + ); + } + + if (lowerBoundaryReached) { + break; + } + + JsonNode last = + values.get( + values.size() - 1 + ); + + before = + last.path("signature") + .asText(); + + if (before.isBlank()) { + break; + } + + if (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 = + getTransaction( + signatureInfo.signature() + ); + + if (transaction == null + || transaction.isNull()) { + + /* + * If a transaction that belongs to the recovery + * range cannot be read, we cannot claim that + * recovery is complete. + */ + throw new IOException( + "Transaction unavailable during recovery: " + + signatureInfo.signature() + ); + } + + JsonNode message = + transaction + .path("transaction") + .path("message"); + + JsonNode accountKeys = + 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 + ); + } + } + } + + /* + * Do not treat the executable program account + * itself as a PDA candidate. + */ + addresses.remove( + programId + ); + + return addresses; + } + + 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( + "getMultipleAccounts returned unexpected account count" + ); + } + + for ( + int i = 0; + i < batch.size(); + i++ + ) { + + String address = + batch.get(i); + + JsonNode account = + values.get(i); + + /* + * The source only reports that the account + * does not currently exist. + * + * It does NOT decide whether the local + * storage should delete anything. + */ + if (account == null + || account.isNull()) { + + missingAddresses.add( + address + ); + + continue; + } + + String owner = + account.path("owner") + .asText(""); + + /* + * Only accounts currently owned by our + * program belong in the projection. + */ + 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 + ); + } + + private JsonNode getTransaction( + 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; + } + + 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; + } + } + + public record SignatureInfo( + String signature, + long slot + ) { + } + + public record AccountBatchResult( + List updates, + List missingAddresses + ) { + } + + @Override + public void close() { + + httpClient + .dispatcher() + .executorService() + .shutdown(); + + httpClient + .connectionPool() + .evictAll(); + } +} \ No newline at end of file diff --git a/src/main/java/sync/source/rpc/SolanaWebSocketClient.java b/src/main/java/sync/source/rpc/SolanaWebSocketClient.java new file mode 100644 index 0000000..f9c7563 --- /dev/null +++ b/src/main/java/sync/source/rpc/SolanaWebSocketClient.java @@ -0,0 +1,541 @@ +package sync.source.rpc; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import okhttp3.*; +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 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; + } + + System.out.println( + "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 + ); + + System.out.println( + "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" + ); + } + + System.out.println( + "Subscription request sent for program: " + + programId + ); + + } catch (Exception exception) { + + System.err.println( + "Failed to send subscription request: " + + exception.getMessage() + ); + + 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 + ) { + + System.out.println( + "Subscribed successfully. Subscription id: " + + root + .get("result") + .asText() + ); + + boolean firstConnection = + everSubscribed + .compareAndSet( + false, + true + ); + + ConnectionListener listener = + connectionListener; + + if (listener != null) { + listener.onConnected( + firstConnection + ); + } + + return; + } + + if (root.has("error")) { + + System.err.println( + "Solana RPC error: " + + root.get("error") + ); + + return; + } + + if (!"programNotification" + .equals( + root + .path("method") + .asText() + )) { + return; + } + + ProgramAccountUpdate update = + parseProgramNotification( + root + ); + + System.out.printf( + "Account update: %s, slot=%d, base64Chars=%d%n", + update.address(), + update.slot(), + update.dataBase64().length() + ); + + AccountUpdateListener listener = + accountUpdateListener; + + if (listener != null) { + listener.onAccountUpdate( + update + ); + } + + } catch (Exception exception) { + + System.err.println( + "Failed to process WebSocket message: " + + exception.getMessage() + ); + + System.err.println( + "Raw message: " + + text + ); + } + } + + 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 + ) { + + System.out.printf( + "WebSocket closed. code=%d reason=%s%n", + code, + reason + ); + + notifyDisconnected( + new IllegalStateException( + "WebSocket closed: " + + reason + ) + ); + + scheduleReconnect(); + } + + @Override + public void onFailure( + WebSocket webSocket, + Throwable throwable, + Response response + ) { + + System.err.println( + "WebSocket failure: " + + throwable.getMessage() + ); + + if (response != null) { + System.err.println( + "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) { + System.err.println( + "Connection listener failed: " + + exception.getMessage() + ); + } + } + } + + 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; + }; + + System.out.printf( + "Reconnect scheduled in %d seconds (attempt %d)%n", + 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/src/main/java/sync/storage/StorageRepository.java b/src/main/java/sync/storage/StorageRepository.java new file mode 100644 index 0000000..6efa959 --- /dev/null +++ b/src/main/java/sync/storage/StorageRepository.java @@ -0,0 +1,57 @@ +package sync.storage; + +import sync.core.SyncCheckpoint; +import sync.core.SyncState; +import sync.model.ProgramAccountUpdate; + +import java.util.List; + +public interface StorageRepository + extends AutoCloseable { + + boolean hasAccounts() + throws Exception; + + void upsertAccount( + ProgramAccountUpdate update + ) throws Exception; + + void upsertAccounts( + Iterable updates + ) throws Exception; + + void deleteAccount( + String address + ) throws Exception; + + void deleteAccounts( + Iterable addresses + ) throws Exception; + + boolean accountExists( + String address + ) throws Exception; + + List findExistingAddresses( + Iterable addresses + ) throws Exception; + + SyncCheckpoint getSyncCheckpoint() + throws Exception; + + void updateSyncState( + SyncState state + ) throws Exception; + + void updateLastObservedSlot( + long slot + ) throws Exception; + + void updateLastReconciledSlot( + long slot + ) throws Exception; + + void updateLastFullSyncSlot( + long slot + ) throws Exception; +} \ No newline at end of file diff --git a/src/main/java/sync/storage/sqlite/SQLiteStorageRepository.java b/src/main/java/sync/storage/sqlite/SQLiteStorageRepository.java new file mode 100644 index 0000000..4cb9eff --- /dev/null +++ b/src/main/java/sync/storage/sqlite/SQLiteStorageRepository.java @@ -0,0 +1,552 @@ +package sync.storage.sqlite; + +import sync.core.SyncCheckpoint; +import sync.core.SyncState; +import sync.model.ProgramAccountUpdate; +import sync.storage.StorageRepository; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.*; +import java.util.ArrayList; +import java.util.List; + +public final class SQLiteStorageRepository + implements StorageRepository { + + private final Connection connection; + + public SQLiteStorageRepository( + Path databasePath + ) throws SQLException, IOException { + + Path absolutePath = + databasePath.toAbsolutePath(); + + Path parent = + absolutePath.getParent(); + + if (parent != null) { + Files.createDirectories(parent); + } + + this.connection = + DriverManager.getConnection( + "jdbc:sqlite:" + absolutePath + ); + + configure(); + migrate(); + + System.out.println( + "Database initialized: " + absolutePath + ); + } + + private void configure() + throws SQLException { + + try (Statement statement = + connection.createStatement()) { + + statement.execute( + "PRAGMA journal_mode=WAL" + ); + + statement.execute( + "PRAGMA synchronous=NORMAL" + ); + + statement.execute( + "PRAGMA busy_timeout=5000" + ); + } + } + + private void migrate() + throws SQLException { + + try (Statement statement = + connection.createStatement()) { + + statement.execute( + "CREATE TABLE IF NOT EXISTS program_accounts (" + + "address TEXT PRIMARY KEY," + + "owner TEXT NOT NULL," + + "lamports INTEGER NOT NULL," + + "slot INTEGER NOT NULL," + + "data_base64 TEXT NOT NULL," + + "executable INTEGER NOT NULL," + + "rent_epoch INTEGER," + + "updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP" + + ")" + ); + + statement.execute( + "CREATE INDEX IF NOT EXISTS idx_program_accounts_slot " + + "ON program_accounts(slot)" + ); + + statement.execute( + "CREATE TABLE IF NOT EXISTS sync_state (" + + "id INTEGER PRIMARY KEY CHECK (id = 1)," + + "state TEXT NOT NULL," + + "last_observed_slot INTEGER," + + "last_reconciled_slot INTEGER," + + "last_full_sync_slot INTEGER," + + "last_sync_at TEXT," + + "updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP" + + ")" + ); + + statement.execute( + "INSERT OR IGNORE INTO sync_state " + + "(id, state) VALUES (1, 'EMPTY')" + ); + + ensureColumnExists( + statement, + "sync_state", + "last_observed_slot", + "INTEGER" + ); + + ensureColumnExists( + statement, + "sync_state", + "last_reconciled_slot", + "INTEGER" + ); + + ensureColumnExists( + statement, + "sync_state", + "last_full_sync_slot", + "INTEGER" + ); + + if (columnExists( + statement, + "sync_state", + "last_processed_slot" + )) { + + statement.execute( + "UPDATE sync_state SET " + + "last_observed_slot = COALESCE(" + + "last_observed_slot, last_processed_slot), " + + "last_reconciled_slot = COALESCE(" + + "last_reconciled_slot, last_processed_slot)" + ); + } + } + } + + private void ensureColumnExists( + Statement statement, + String table, + String column, + String type + ) throws SQLException { + + if (!columnExists( + statement, + table, + column + )) { + + statement.execute( + "ALTER TABLE " + + table + + " ADD COLUMN " + + column + + " " + + type + ); + } + } + + private boolean columnExists( + Statement statement, + String table, + String column + ) throws SQLException { + + try (ResultSet resultSet = + statement.executeQuery( + "PRAGMA table_info(" + + table + + ")" + )) { + + while (resultSet.next()) { + + if (column.equals( + resultSet.getString("name") + )) { + return true; + } + } + } + + return false; + } + + @Override + public synchronized void upsertAccount( + ProgramAccountUpdate update + ) throws SQLException { + + String sql = + "INSERT INTO program_accounts (" + + "address, owner, lamports, slot, " + + "data_base64, executable, rent_epoch, updated_at" + + ") VALUES (?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) " + + + "ON CONFLICT(address) DO UPDATE SET " + + "owner = excluded.owner, " + + "lamports = excluded.lamports, " + + "slot = excluded.slot, " + + "data_base64 = excluded.data_base64, " + + "executable = excluded.executable, " + + "rent_epoch = excluded.rent_epoch, " + + "updated_at = CURRENT_TIMESTAMP " + + + "WHERE excluded.slot >= program_accounts.slot"; + + try (PreparedStatement statement = + connection.prepareStatement(sql)) { + + statement.setString(1, update.address()); + statement.setString(2, update.owner()); + statement.setLong(3, update.lamports()); + statement.setLong(4, update.slot()); + statement.setString(5, update.dataBase64()); + statement.setInt( + 6, + update.executable() + ? 1 + : 0 + ); + + if (update.rentEpoch() == null) { + statement.setNull( + 7, + Types.BIGINT + ); + } else { + statement.setLong( + 7, + update.rentEpoch() + ); + } + + statement.executeUpdate(); + } + } + + @Override + public synchronized void upsertAccounts( + Iterable updates + ) throws Exception { + + for (ProgramAccountUpdate update : updates) { + upsertAccount(update); + } + } + + @Override + public synchronized void deleteAccount( + String address + ) throws SQLException { + + try (PreparedStatement statement = + connection.prepareStatement( + "DELETE FROM program_accounts " + + "WHERE address = ?" + )) { + + statement.setString( + 1, + address + ); + + statement.executeUpdate(); + } + } + + @Override + public synchronized void deleteAccounts( + Iterable addresses + ) throws Exception { + + for (String address : addresses) { + deleteAccount(address); + } + } + + @Override + public synchronized boolean accountExists( + String address + ) throws SQLException { + + try (PreparedStatement statement = + connection.prepareStatement( + "SELECT 1 " + + "FROM program_accounts " + + "WHERE address = ? " + + "LIMIT 1" + )) { + + statement.setString( + 1, + address + ); + + try (ResultSet resultSet = + statement.executeQuery()) { + + return resultSet.next(); + } + } + } + + @Override + public synchronized List findExistingAddresses( + Iterable addresses + ) throws SQLException { + + List existing = + new ArrayList<>(); + + try (PreparedStatement statement = + connection.prepareStatement( + "SELECT 1 " + + "FROM program_accounts " + + "WHERE address = ? " + + "LIMIT 1" + )) { + + for (String address : addresses) { + + statement.setString( + 1, + address + ); + + try (ResultSet resultSet = + statement.executeQuery()) { + + if (resultSet.next()) { + existing.add( + address + ); + } + } + } + } + + return existing; + } + + @Override + public synchronized boolean hasAccounts() + throws SQLException { + + try ( + Statement statement = + connection.createStatement(); + + ResultSet resultSet = + statement.executeQuery( + "SELECT EXISTS(" + + "SELECT 1 " + + "FROM program_accounts " + + "LIMIT 1" + + ")" + ) + ) { + + return resultSet.next() + && resultSet.getInt(1) == 1; + } + } + + @Override + public synchronized SyncCheckpoint getSyncCheckpoint() + throws SQLException { + + try ( + Statement statement = + connection.createStatement(); + + ResultSet resultSet = + statement.executeQuery( + "SELECT " + + "state, " + + "last_observed_slot, " + + "last_reconciled_slot, " + + "last_full_sync_slot " + + "FROM sync_state " + + "WHERE id = 1" + ) + ) { + + if (!resultSet.next()) { + + return new SyncCheckpoint( + SyncState.EMPTY, + null, + null, + null + ); + } + + return new SyncCheckpoint( + SyncState.valueOf( + resultSet.getString( + "state" + ) + ), + getNullableLong( + resultSet, + "last_observed_slot" + ), + getNullableLong( + resultSet, + "last_reconciled_slot" + ), + getNullableLong( + resultSet, + "last_full_sync_slot" + ) + ); + } + } + + private Long getNullableLong( + ResultSet resultSet, + String column + ) throws SQLException { + + long value = + resultSet.getLong( + column + ); + + if (resultSet.wasNull()) { + return null; + } + + return value; + } + + @Override + public synchronized void updateSyncState( + SyncState state + ) throws SQLException { + + try (PreparedStatement statement = + connection.prepareStatement( + "UPDATE sync_state " + + "SET state = ?, " + + "updated_at = CURRENT_TIMESTAMP " + + "WHERE id = 1" + )) { + + statement.setString( + 1, + state.name() + ); + + statement.executeUpdate(); + } + } + + @Override + public synchronized void updateLastObservedSlot( + long slot + ) throws SQLException { + + updateMaxSlot( + "last_observed_slot", + slot + ); + } + + @Override + public synchronized void updateLastReconciledSlot( + long slot + ) throws SQLException { + + updateMaxSlot( + "last_reconciled_slot", + slot + ); + } + + private void updateMaxSlot( + String column, + long slot + ) throws SQLException { + + String sql = + "UPDATE sync_state " + + "SET " + column + " = " + + "CASE " + + "WHEN " + column + " IS NULL " + + "OR ? > " + column + " " + + "THEN ? " + + "ELSE " + column + " " + + "END, " + + "last_sync_at = CURRENT_TIMESTAMP, " + + "updated_at = CURRENT_TIMESTAMP " + + "WHERE id = 1"; + + try (PreparedStatement statement = + connection.prepareStatement(sql)) { + + statement.setLong( + 1, + slot + ); + + statement.setLong( + 2, + slot + ); + + statement.executeUpdate(); + } + } + + @Override + public synchronized void updateLastFullSyncSlot( + long slot + ) throws SQLException { + + try (PreparedStatement statement = + connection.prepareStatement( + "UPDATE sync_state " + + "SET last_full_sync_slot = ?, " + + "updated_at = CURRENT_TIMESTAMP " + + "WHERE id = 1" + )) { + + statement.setLong( + 1, + slot + ); + + statement.executeUpdate(); + } + } + + @Override + public void close() + throws SQLException { + + connection.close(); + } +} \ No newline at end of file