SHA256
186 lines
10 KiB
Java
186 lines
10 KiB
Java
package server.archive;
|
|
|
|
import blockchain.Ans104DataItem;
|
|
import blockchain.BchBlockEntry;
|
|
import com.fasterxml.jackson.databind.JsonNode;
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
import org.slf4j.Logger;
|
|
import org.slf4j.LoggerFactory;
|
|
import server.logic.ws_protocol.JSON.handlers.blockchain.Net_AddBlock_Handler;
|
|
import shine.db.dao.ArweaveBlockImportDAO;
|
|
import shine.db.dao.BlockchainStateDAO;
|
|
import shine.db.dao.BlocksDAO;
|
|
import shine.db.dao.SolanaUserPdaCurrentDAO;
|
|
import shine.db.entities.BlockchainStateEntry;
|
|
import shine.db.entities.SolanaUserPdaCurrentEntry;
|
|
|
|
import java.io.IOException;
|
|
import java.net.URI;
|
|
import java.net.http.HttpClient;
|
|
import java.net.http.HttpRequest;
|
|
import java.net.http.HttpResponse;
|
|
import java.nio.charset.StandardCharsets;
|
|
import java.time.Duration;
|
|
import java.util.*;
|
|
|
|
/**
|
|
* Discovers App=test5590 child DataItems, extracts their exact serialized bytes
|
|
* from the root ANS-104 bundle and imports them through the normal AddBlock checks.
|
|
*/
|
|
public final class ArweaveBlockSyncService {
|
|
private static final Logger log = LoggerFactory.getLogger(ArweaveBlockSyncService.class);
|
|
private static final ObjectMapper MAPPER = new ObjectMapper();
|
|
private static final Base64.Decoder B64URL = Base64.getUrlDecoder();
|
|
|
|
private final ArweaveBlocksConfig cfg;
|
|
private final HttpClient http = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(20)).build();
|
|
private final ArweaveBlockImportDAO importDAO = ArweaveBlockImportDAO.getInstance();
|
|
private final BlocksDAO blocksDAO = BlocksDAO.getInstance();
|
|
private final SolanaUserPdaCurrentDAO usersDAO = SolanaUserPdaCurrentDAO.getInstance();
|
|
private final BlockchainStateDAO stateDAO = BlockchainStateDAO.getInstance();
|
|
private final Net_AddBlock_Handler addBlock = new Net_AddBlock_Handler();
|
|
|
|
public ArweaveBlockSyncService(ArweaveBlocksConfig cfg) { this.cfg = cfg; }
|
|
|
|
public void runCycle() throws Exception {
|
|
discover();
|
|
drainQueue();
|
|
}
|
|
|
|
private void discover() throws Exception {
|
|
long stored = importDAO.getLastBlockHeight();
|
|
long minHeight = Math.max(cfg.syncStartBlockHeight(), stored);
|
|
String cursor = null;
|
|
long maxHeight = minHeight;
|
|
Map<String, byte[]> rootCache = new HashMap<>();
|
|
int discovered = 0;
|
|
|
|
do {
|
|
JsonNode response = graphQlPage(minHeight, cursor);
|
|
JsonNode txs = response.path("data").path("transactions");
|
|
if (response.has("errors")) throw new IOException("Arweave GraphQL errors: " + response.path("errors"));
|
|
JsonNode edges = txs.path("edges");
|
|
if (!edges.isArray()) throw new IOException("Arweave GraphQL returned no transactions.edges");
|
|
|
|
String nextCursor = null;
|
|
for (JsonNode edge : edges) {
|
|
nextCursor = edge.path("cursor").asText(null);
|
|
JsonNode node = edge.path("node");
|
|
String itemIdText = node.path("id").asText("").trim();
|
|
String rootTx = node.path("bundledIn").path("id").asText("").trim();
|
|
long height = node.path("block").path("height").asLong(-1L);
|
|
if (itemIdText.isBlank() || rootTx.isBlank() || height < 0) continue;
|
|
maxHeight = Math.max(maxHeight, height);
|
|
byte[] id32;
|
|
try { id32 = B64URL.decode(itemIdText); }
|
|
catch (IllegalArgumentException bad) { continue; }
|
|
if (id32.length != 32 || blocksDAO.existsByDataItemId(id32) || importDAO.exists(id32)) continue;
|
|
|
|
byte[] bundle = rootCache.computeIfAbsent(rootTx, key -> {
|
|
try { return downloadRootBundle(key); }
|
|
catch (Exception e) { throw new RootDownloadRuntimeException(e); }
|
|
});
|
|
byte[] raw = Ans104Bundle.find(bundle, id32, Math.max(cfg.publishMaxItems() * 4, 100_000));
|
|
if (raw == null) throw new IOException("DataItem " + itemIdText + " not found in root bundle " + rootTx);
|
|
Ans104DataItem parsed = new Ans104DataItem(raw);
|
|
if (!parsed.hasTag(ArweaveBlocksConfig.TEST_TAG_NAME, ArweaveBlocksConfig.TEST_TAG_VALUE)) continue;
|
|
if (!parsed.verifySignature()) throw new IOException("Bad ANS-104 signature for " + itemIdText);
|
|
importDAO.enqueueIfMissing(id32, rootTx, height, raw, System.currentTimeMillis());
|
|
discovered++;
|
|
}
|
|
boolean hasNext = txs.path("pageInfo").path("hasNextPage").asBoolean(false);
|
|
cursor = hasNext ? nextCursor : null;
|
|
if (hasNext && (cursor == null || cursor.isBlank())) throw new IOException("GraphQL hasNextPage without cursor");
|
|
} while (cursor != null);
|
|
|
|
// Keep one-height overlap: the next query includes this height and deduplicates IDs.
|
|
importDAO.setLastBlockHeight(maxHeight, System.currentTimeMillis());
|
|
if (discovered > 0) log.info("Arweave discovery queued {} new SHiNE test DataItems through height {}", discovered, maxHeight);
|
|
}
|
|
|
|
private void drainQueue() throws Exception {
|
|
int passes = 0;
|
|
int importedTotal = 0;
|
|
boolean progress;
|
|
do {
|
|
progress = false;
|
|
List<ArweaveBlockImportDAO.QueueItem> pending = importDAO.listPending(cfg.syncQueueBatchSize());
|
|
if (pending.isEmpty()) break;
|
|
for (ArweaveBlockImportDAO.QueueItem q : pending) {
|
|
Ans104DataItem item;
|
|
BchBlockEntry block;
|
|
try {
|
|
item = new Ans104DataItem(q.rawDataItem());
|
|
if (!Arrays.equals(item.id32(), q.dataItemId())) throw new IllegalArgumentException("data_item_id mismatch");
|
|
if (!item.hasTag(ArweaveBlocksConfig.TEST_TAG_NAME, ArweaveBlocksConfig.TEST_TAG_VALUE)) throw new IllegalArgumentException("bad App tag");
|
|
if (!item.verifySignature()) throw new IllegalArgumentException("bad ANS-104 signature");
|
|
block = new BchBlockEntry(q.rawDataItem());
|
|
} catch (Exception e) {
|
|
importDAO.reject(q.dataItemId(), "invalid_data_item: " + e.getMessage(), System.currentTimeMillis());
|
|
continue;
|
|
}
|
|
|
|
SolanaUserPdaCurrentEntry user = usersDAO.getByBlockchainKey(item.owner32());
|
|
if (user == null || user.getBlockchainName() == null || user.getBlockchainName().isBlank()) {
|
|
importDAO.setPendingError(q.dataItemId(), "unknown_owner", System.currentTimeMillis());
|
|
continue;
|
|
}
|
|
ensureState(user);
|
|
Net_AddBlock_Handler.ArweaveImportResult result = addBlock.addBlockFromArweave(user.getBlockchainName(), q.rawDataItem());
|
|
if (result.ok()) {
|
|
importDAO.delete(q.dataItemId());
|
|
importedTotal++; progress = true;
|
|
continue;
|
|
}
|
|
String reason = result.reasonCode();
|
|
if ("bad_block_number".equals(reason) || "bad_prev_hash".equals(reason)
|
|
|| "prev_line_block_not_found".equals(reason) || "channel_not_found".equals(reason)
|
|
|| "chain_resync_in_progress".equals(reason)) {
|
|
importDAO.setPendingError(q.dataItemId(), reason, System.currentTimeMillis());
|
|
} else {
|
|
importDAO.reject(q.dataItemId(), reason, System.currentTimeMillis());
|
|
}
|
|
}
|
|
passes++;
|
|
} while (progress && passes < 8);
|
|
if (importedTotal > 0) log.info("Imported {} SHiNE test DataItems from Arweave", importedTotal);
|
|
}
|
|
|
|
private void ensureState(SolanaUserPdaCurrentEntry user) throws Exception {
|
|
BlockchainStateEntry existing = stateDAO.getByBlockchainName(user.getBlockchainName());
|
|
if (existing != null) return;
|
|
BlockchainStateEntry s = new BlockchainStateEntry();
|
|
s.setBlockchainName(user.getBlockchainName()); s.setLogin(user.getLogin()); s.setBlockchainKey(user.getBlockchainKey());
|
|
s.setSizeLimit(user.getPaidLimitBytes() > 0 ? user.getPaidLimitBytes() : 100_000L);
|
|
s.setFileSizeBytes(0L); s.setLastBlockNumber(-1); s.setLastBlockHash(null); s.setUpdatedAtMs(System.currentTimeMillis());
|
|
stateDAO.insertIfMissing(s);
|
|
}
|
|
|
|
private JsonNode graphQlPage(long minHeight, String cursor) throws Exception {
|
|
String after = cursor == null ? "null" : "\"" + escapeGraphQl(cursor) + "\"";
|
|
String query = "query { transactions(tags:[{name:\"App\",values:[\"test5590\"]}], block:{min:" + minHeight + "}, first:" + cfg.syncPageSize() + ", after:" + after + ", sort:HEIGHT_ASC) { pageInfo { hasNextPage } edges { cursor node { id bundledIn { id } block { height } } } } }";
|
|
String body = MAPPER.writeValueAsString(Map.of("query", query));
|
|
HttpRequest req = HttpRequest.newBuilder(URI.create(trim(cfg.syncGateway()) + "/graphql"))
|
|
.timeout(Duration.ofSeconds(60)).header("Content-Type","application/json").header("Accept","application/json")
|
|
.POST(HttpRequest.BodyPublishers.ofString(body, StandardCharsets.UTF_8)).build();
|
|
HttpResponse<String> resp = http.send(req, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8));
|
|
if (resp.statusCode() < 200 || resp.statusCode() >= 300) throw new IOException("GraphQL HTTP " + resp.statusCode());
|
|
return MAPPER.readTree(resp.body());
|
|
}
|
|
|
|
private byte[] downloadRootBundle(String txId) throws Exception {
|
|
HttpRequest req = HttpRequest.newBuilder(URI.create(trim(cfg.syncGateway()) + "/" + txId))
|
|
.timeout(Duration.ofMinutes(5)).GET().build();
|
|
HttpResponse<byte[]> resp = http.send(req, HttpResponse.BodyHandlers.ofByteArray());
|
|
if (resp.statusCode() < 200 || resp.statusCode() >= 300) throw new IOException("Arweave root HTTP " + resp.statusCode() + " tx=" + txId);
|
|
byte[] body = resp.body();
|
|
if (body == null || body.length == 0) throw new IOException("Empty Arweave root bundle " + txId);
|
|
if (body.length > cfg.syncMaxRootBundleBytes()) throw new IOException("Root bundle exceeds syncMaxRootBundleBytes: " + body.length);
|
|
return body;
|
|
}
|
|
|
|
private static String trim(String s){return String.valueOf(s==null?"":s).trim().replaceAll("/+$","");}
|
|
private static String escapeGraphQl(String s){return s.replace("\\","\\\\").replace("\"","\\\"");}
|
|
private static final class RootDownloadRuntimeException extends RuntimeException { RootDownloadRuntimeException(Throwable cause){super(cause);} }
|
|
}
|