tram/sync-solana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

синхронизация с солана

Browse Source
This commit is contained in:
tram
2026-07-21 13:19:40 +03:00
commit 02a849089c
22 changed files with 3100 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+21
View File
@@ -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
.gitignore
+7
View File
@@ -0,0 +1,7 @@
target/
data/*.db
data/*.db-shm
data/*.db-wal
.env
.idea/
*.iml
PATCH_NOTES.md
+23
View File
@@ -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<Void> subscribed = new CompletableFuture<>();
2. Метод:
public CompletableFuture<Void> subscribedFuture() {
return subscribed;
}
3. В обработчике успешной подписки после лога:
subscribed.complete(null);
README.md
+83
View File
@@ -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.
pom.xml
+56
View File
@@ -0,0 +1,56 @@
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.example</groupId>
<artifactId>helius-sync-test</artifactId>
<version>1.0.0</version>
<properties>
<maven.compiler.release>17</maven.compiler.release>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<okhttp.version>4.12.0</okhttp.version>
<jackson.version>2.17.2</jackson.version>
<sqlite.version>3.46.0.0</sqlite.version>
</properties>
<dependencies>
<dependency>
<groupId>com.squareup.okhttp3</groupId>
<artifactId>okhttp</artifactId>
<version>${okhttp.version}</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>${jackson.version}</version>
</dependency>
<dependency>
<groupId>org.xerial</groupId>
<artifactId>sqlite-jdbc</artifactId>
<version>${sqlite.version}</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.13.0</version>
<configuration>
<release>17</release>
</configuration>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>exec-maven-plugin</artifactId>
<version>3.4.1</version>
<configuration>
<mainClass>sync.Main</mainClass>
</configuration>
</plugin>
</plugins>
</build>
</project>
src/main/java/sync/Main.java
+148
View File
@@ -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
);
}
}
}
src/main/java/sync/config/AppConfig.java
+136
View File
@@ -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();
}
}
src/main/java/sync/core/SyncCheckpoint.java
+9
View File
@@ -0,0 +1,9 @@
package sync.core;
public record SyncCheckpoint(
SyncState state,
Long lastObservedSlot,
Long lastReconciledSlot,
Long lastFullSyncSlot
) {
}
src/main/java/sync/core/SyncCoordinator.java
+379
View File
@@ -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<String> 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) {
}
}
}
src/main/java/sync/core/SyncMode.java
+8
View File
@@ -0,0 +1,8 @@
package sync.core;
public enum SyncMode {
FULL_BOOTSTRAP,
REALTIME_SYNC,
INCREMENTAL_RECOVERY,
FULL_RECONCILIATION
}
src/main/java/sync/core/SyncState.java
+10
View File
@@ -0,0 +1,10 @@
package sync.core;
public enum SyncState {
EMPTY,
BOOTSTRAPPING,
LIVE,
RECOVERING,
RECONCILING,
FAILED
}
src/main/java/sync/model/ProgramAccountUpdate.java
+12
View File
@@ -0,0 +1,12 @@
package sync.model;
public record ProgramAccountUpdate(
String address,
String owner,
long lamports,
long slot,
String dataBase64,
boolean executable,
Long rentEpoch
) {
}
src/main/java/sync/model/RecoveryResult.java
+44
View File
@@ -0,0 +1,44 @@
package sync.model;
import java.util.List;
public record RecoveryResult(
boolean complete,
long fromSlot,
long toSlot,
List<ProgramAccountUpdate> updates,
List<String> missingAddresses,
String fallbackReason
) {
public static RecoveryResult complete(
long fromSlot,
long toSlot,
List<ProgramAccountUpdate> updates,
List<String> 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
);
}
}
src/main/java/sync/model/SnapshotResult.java
+9
View File
@@ -0,0 +1,9 @@
package sync.model;
import java.util.List;
public record SnapshotResult(
long snapshotSlot,
List<ProgramAccountUpdate> accounts
) {
}
src/main/java/sync/source/AccountUpdateListener.java
+11
View File
@@ -0,0 +1,11 @@
package sync.source;
import sync.model.ProgramAccountUpdate;
@FunctionalInterface
public interface AccountUpdateListener {
void onAccountUpdate(
ProgramAccountUpdate update
);
}
src/main/java/sync/source/ConnectionListener.java
+12
View File
@@ -0,0 +1,12 @@
package sync.source;
public interface ConnectionListener {
void onConnected(
boolean firstConnection
);
void onDisconnected(
Throwable cause
);
}
src/main/java/sync/source/SolanaDataSource.java
+25
View File
@@ -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;
}
src/main/java/sync/source/rpc/RpcSolanaDataSource.java
+241
View File
@@ -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<SolanaRpcClient.SignatureInfo>
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<String> 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();
}
}
src/main/java/sync/source/rpc/SolanaRpcClient.java
+716
View File
@@ -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<String, Object> 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<ProgramAccountUpdate> 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<String, Object> 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<String, Object> payload =
Map.of(
"jsonrpc", "2.0",
"id", 102,
"method", "getFirstAvailableBlock"
);
JsonNode root =
executeRpc(payload);
return root
.path("result")
.asLong(-1);
}
public List<SignatureInfo> getSignaturesAfterSlot(
long fromSlot,
long targetSlot
) throws IOException {
List<SignatureInfo> signatures =
new ArrayList<>();
String before =
null;
boolean lowerBoundaryReached =
false;
while (!lowerBoundaryReached) {
Map<String, Object> options =
new LinkedHashMap<>();
options.put(
"limit",
SIGNATURE_PAGE_SIZE
);
options.put(
"commitment",
commitment
);
if (before != null) {
options.put(
"before",
before
);
}
Map<String, Object> 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<String> getTouchedAddresses(
List<SignatureInfo> signatures
) throws IOException {
Set<String> 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<String> addresses,
long recoverySlot
) throws IOException {
List<ProgramAccountUpdate> updates =
new ArrayList<>();
List<String> missingAddresses =
new ArrayList<>();
List<String> 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<String> batch =
addressList.subList(
offset,
end
);
Map<String, Object> 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<String, Object> 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<String, Object> 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<ProgramAccountUpdate> updates,
List<String> missingAddresses
) {
}
@Override
public void close() {
httpClient
.dispatcher()
.executorService()
.shutdown();
httpClient
.connectionPool()
.evictAll();
}
}
src/main/java/sync/source/rpc/SolanaWebSocketClient.java
+541
View File
@@ -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<String, Object> 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();
}
}
src/main/java/sync/storage/StorageRepository.java
+57
View File
@@ -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<ProgramAccountUpdate> updates
) throws Exception;
void deleteAccount(
String address
) throws Exception;
void deleteAccounts(
Iterable<String> addresses
) throws Exception;
boolean accountExists(
String address
) throws Exception;
List<String> findExistingAddresses(
Iterable<String> 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;
}
src/main/java/sync/storage/sqlite/SQLiteStorageRepository.java
+552
View File
@@ -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<ProgramAccountUpdate> 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<String> 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<String> findExistingAddresses(
Iterable<String> addresses
) throws SQLException {
List<String> 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();
}
}