24138607049

Committed 08 Apr 2026 01:42PM UTC coverage: 32.47% (-0.4%) from 32.824%

Build # 24138607049

Build Type

Pull #99

github

Committed by

web-flow

Commit Message

Merge 336546501 into 689a63b39

Pull Request Pull Request #99: Fix peer sync race with committed counter watermark

Coverage Stats

268 of 926 branches covered (28.94%)

Branch coverage included in aggregate %.

797 of 2354 relevant lines covered (33.86%)

5.7 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.99

src/main/java/com/knowledgepixels/registry/RegistryDB.java

package com.knowledgepixels.registry;

import com.knowledgepixels.registry.db.IndexInitializer;
import com.mongodb.MongoClient;
import com.mongodb.MongoNamespace;
import com.mongodb.ErrorCategory;
import com.mongodb.MongoWriteException;
import com.mongodb.ServerAddress;
import com.mongodb.client.ClientSession;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.CountOptions;
import com.mongodb.client.model.FindOneAndUpdateOptions;
import com.mongodb.client.model.ReturnDocument;
import com.mongodb.client.model.UpdateOptions;
import net.trustyuri.TrustyUriUtils;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.bson.types.Binary;
import org.eclipse.rdf4j.common.exception.RDF4JException;
import org.eclipse.rdf4j.model.IRI;
import org.eclipse.rdf4j.rio.RDFFormat;
import org.nanopub.MalformedNanopubException;
import org.nanopub.Nanopub;

import java.util.Calendar;
import org.nanopub.NanopubUtils;
import org.nanopub.extra.security.MalformedCryptoElementException;
import org.nanopub.extra.security.NanopubSignatureElement;
import org.nanopub.extra.security.SignatureUtils;
import org.nanopub.jelly.JellyUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.security.GeneralSecurityException;
import java.util.ArrayList;

import static com.mongodb.client.model.Indexes.ascending;

public class RegistryDB {

    private RegistryDB() {
    }

    private static final String REGISTRY_DB_NAME = Utils.getEnv("REGISTRY_DB_NAME", "nanopubRegistry");

    private static final Logger logger = LoggerFactory.getLogger(RegistryDB.class);

    private static MongoClient mongoClient;
    private static MongoDatabase mongoDB;

    /**
     * Returns the MongoDB database instance.
     *
     * @return the MongoDatabase instance
     */
    public static MongoDatabase getDB() {
        return mongoDB;
    }

    /**
     * Returns the MongoDB client instance.
     *
     * @return the MongoClient instance
     */
    public static MongoClient getClient() {
        return mongoClient;
    }

    /**
     * Returns the specified collection from the MongoDB database.
     *
     * @param name the name of the collection
     * @return the MongoCollection instance
     */
    public static MongoCollection<Document> collection(String name) {
        return mongoDB.getCollection(name);
    }

    /**
     * Initializes the MongoDB connection and sets up collections and indexes if not already initialized.
     */
    public static void init() {
        if (mongoClient != null) {
            logger.info("RegistryDB already initialized");
            return;
        }
        final String REGISTRY_DB_HOST = Utils.getEnv("REGISTRY_DB_HOST", "mongodb");
        final int REGISTRY_DB_PORT = Integer.parseInt(Utils.getEnv("REGISTRY_DB_PORT", String.valueOf(ServerAddress.defaultPort())));
        logger.info("Initializing RegistryDB connection to database '{}' at {}:{}", REGISTRY_DB_NAME, REGISTRY_DB_HOST, REGISTRY_DB_PORT);
        mongoClient = new MongoClient(REGISTRY_DB_HOST, REGISTRY_DB_PORT);
        mongoDB = mongoClient.getDatabase(REGISTRY_DB_NAME);

        try (ClientSession mongoSession = mongoClient.startSession()) {
            if (!isInitialized(mongoSession)) {
                IndexInitializer.initCollections(mongoSession);
            }
            initCounter(mongoSession);
        }
    }

    /**
     * Checks if the database has been initialized.
     *
     * @param mongoSession the MongoDB client session
     * @return true if initialized, false otherwise
     */
    public static boolean isInitialized(ClientSession mongoSession) {
        return getValue(mongoSession, Collection.SERVER_INFO.toString(), "setupId") != null;
    }

    /**
     * Renames a collection in the database. If the new collection name already exists, it will be dropped first.
     *
     * @param oldCollectionName the current name of the collection
     * @param newCollectionName the new name for the collection
     */
    public static void rename(String oldCollectionName, String newCollectionName) {
        // Designed as idempotent operation: calling multiple times has same effect as calling once
        if (hasCollection(oldCollectionName)) {
            if (hasCollection(newCollectionName)) {
                collection(newCollectionName).drop();
            }
            collection(oldCollectionName).renameCollection(new MongoNamespace(REGISTRY_DB_NAME, newCollectionName));
        }
    }

    /**
     * Checks if a collection with the given name exists in the database.
     *
     * @param collectionName the name of the collection to check
     * @return true if the collection exists, false otherwise
     */
    public static boolean hasCollection(String collectionName) {
        return mongoDB.listCollectionNames().into(new ArrayList<>()).contains(collectionName);
    }

    /**
     * Increases the trust state counter in the server info collection.
     *
     * @param mongoSession the MongoDB client session
     */
    public static void increaseStateCounter(ClientSession mongoSession) {
        MongoCursor<Document> cursor = collection(Collection.SERVER_INFO.toString()).find(mongoSession, new Document("_id", "trustStateCounter")).cursor();
        if (cursor.hasNext()) {
            long counter = cursor.next().getLong("value");
            collection(Collection.SERVER_INFO.toString()).updateOne(mongoSession, new Document("_id", "trustStateCounter"), new Document("$set", new Document("value", counter + 1)));
        } else {
            collection(Collection.SERVER_INFO.toString()).insertOne(mongoSession, new Document("_id", "trustStateCounter").append("value", 0L));
        }
    }

    /**
     * Checks if an element with the given name exists in the specified collection.
     *
     * @param mongoSession the MongoDB client session
     * @param collection   the name of the collection
     * @param elementName  the name of the element used as the _id field
     * @return true if the element exists, false otherwise
     */
    public static boolean has(ClientSession mongoSession, String collection, String elementName) {
        return has(mongoSession, collection, new Document("_id", elementName));
    }

    private static final CountOptions hasCountOptions = new CountOptions().limit(1);

    /**
     * Checks if any document matching the given filter exists in the specified collection.
     *
     * @param mongoSession the MongoDB client session
     * @param collection   the name of the collection
     * @param find         the filter to match documents
     * @return true if at least one matching document exists, false otherwise
     */
    public static boolean has(ClientSession mongoSession, String collection, Bson find) {
        return collection(collection).countDocuments(mongoSession, find, hasCountOptions) > 0;
    }

    /**
     * Retrieves a cursor for documents matching the given filter in the specified collection.
     *
     * @param mongoSession the MongoDB client session
     * @param collection   the name of the collection
     * @param find         the filter to match documents
     * @return a MongoCursor for the matching documents
     */
    public static MongoCursor<Document> get(ClientSession mongoSession, String collection, Bson find) {
        return collection(collection).find(mongoSession, find).cursor();
    }

    /**
     * Retrieves the value of an element with the given name from the specified collection.
     *
     * @param mongoSession the MongoDB client session
     * @param collection   the name of the collection
     * @param elementName  the name of the element used as the _id field
     * @return the value of the element, or null if not found
     */
    public static Object getValue(ClientSession mongoSession, String collection, String elementName) {
        logger.info("Getting value of element '{}' from collection '{}'", elementName, collection);
        Document d = collection(collection).find(mongoSession, new Document("_id", elementName)).first();
        if (d == null) {
            return null;
        }
        return d.get("value");
    }

    /**
     * Retrieves the boolean value of an element with the given name from the specified collection.
     *
     * @param mongoSession the MongoDB client session
     * @param collection   the name of the collection
     * @param elementName  the name of the element used as the _id field
     * @return the value of the element, or null if not found
     */
    public static boolean isSet(ClientSession mongoSession, String collection, String elementName) {
        Document d = collection(collection).find(mongoSession, new Document("_id", elementName)).first();
        if (d == null) {
            return false;
        }
        return d.getBoolean("value");
    }

    /**
     * Retrieves a single document matching the given filter from the specified collection.
     *
     * @param mongoSession the MongoDB client session
     * @param collection   the name of the collection
     * @param find         the filter to match the document
     * @return the matching document, or null if not found
     */
    public static Document getOne(ClientSession mongoSession, String collection, Bson find) {
        return collection(collection).find(mongoSession, find).first();
    }

    /**
     * Retrieves the maximum value of a specified field from the documents in the given collection.
     *
     * @param mongoSession the MongoDB client session
     * @param collection   the name of the collection
     * @param fieldName    the field for which to find the maximum value
     * @return the maximum value of the specified field, or null if no documents exist
     */
    public static Object getMaxValue(ClientSession mongoSession, String collection, String fieldName) {
        Document doc = collection(collection).find(mongoSession).projection(new Document(fieldName, 1)).sort(new Document(fieldName, -1)).first();
        if (doc == null) {
            return null;
        }
        return doc.get(fieldName);
    }

    /**
     * Retrieves the document with the maximum value of a specified field from the documents matching the given filter in the specified collection.
     *
     * @param mongoSession the MongoDB client session
     * @param collection   the name of the collection
     * @param find         the filter to match documents
     * @param fieldName    the field for which to find the maximum value
     * @return the document with the maximum value of the specified field, or null if no matching documents exist
     */
    public static Document getMaxValueDocument(ClientSession mongoSession, String collection, Bson find, String fieldName) {
        return collection(collection).find(mongoSession, find).sort(new Document(fieldName, -1)).first();
    }

    /**
     * Sets or updates a document in the specified collection.
     *
     * @param mongoSession the MongoDB client session
     * @param collection   the name of the collection
     * @param update       the document to set or update (must contain an _id field)
     */
    public static void set(ClientSession mongoSession, String collection, Document update) {
        Bson find = new Document("_id", update.get("_id"));
        MongoCursor<Document> cursor = collection(collection).find(mongoSession, find).cursor();
        if (cursor.hasNext()) {
            collection(collection).updateOne(mongoSession, find, new Document("$set", update));
        }
    }

    /**
     * Inserts a document into the specified collection.
     *
     * @param mongoSession the MongoDB client session
     * @param collection   the name of the collection
     * @param doc          the document to insert
     */
    public static void insert(ClientSession mongoSession, String collection, Document doc) {
        collection(collection).insertOne(mongoSession, doc);
    }

    /**
     * Sets the value of an element with the given name in the specified collection.
     * If the element does not exist, it will be created.
     *
     * @param mongoSession the MongoDB client session
     * @param collection   the name of the collection
     * @param elementId    the name of the element used as the _id field
     * @param value        the value to set
     */
    public static void setValue(ClientSession mongoSession, String collection, String elementId, Object value) {
        collection(collection).updateOne(mongoSession, new Document("_id", elementId), new Document("$set", new Document("value", value)), new UpdateOptions().upsert(true));
    }

    /**
     * Records the hash of a given value in the "hashes" collection.
     * Uses upsert to avoid expensive exception-based duplicate handling.
     *
     * @param mongoSession the MongoDB client session
     * @param value        the value to hash and record
     */
    public static void recordHash(ClientSession mongoSession, String value) {
        String hash = Utils.getHash(value);
        try {
            collection("hashes").updateOne(mongoSession,
                    new Document("value", value),
                    new Document("$setOnInsert", new Document("value", value).append("hash", hash)),
                    new UpdateOptions().upsert(true));
        } catch (MongoWriteException e) {
            // Concurrent upsert race: another thread inserted the same hash — safe to ignore
            if (e.getError().getCategory() != ErrorCategory.DUPLICATE_KEY) throw e;
        }
    }

    /**
     * Retrieves the original value corresponding to a given hash from the "hashes" collection.
     *
     * @param hash the hash to look up
     * @return the original value, or null if not found
     */
    public static String unhash(String hash) {
        try (var c = collection("hashes").find(new Document("hash", hash)).cursor()) {
            if (c.hasNext()) {
                return c.next().getString("value");
            }
            return null;
        }
    }

    private static final int COUNTER_BATCH_SIZE = Integer.parseInt(
            Utils.getEnv("REGISTRY_COUNTER_BATCH_SIZE", "20"));

    /**
     * Thread-local batch range for counter allocation: [nextToUse, endExclusive).
     * Each thread claims a batch of COUNTER_BATCH_SIZE counter values from MongoDB,
     * then hands them out locally with zero contention until the batch is exhausted.
     */
    private static final ThreadLocal<long[]> counterRange = new ThreadLocal<>();

    /**
     * Initializes the counter documents to the current maximum counter value
     * in the nanopubs collection.
     * Uses $max to ensure the counters are never decreased. Safe to call on every startup.
     */
    private static void initCounter(ClientSession mongoSession) {
        Long maxCounter = (Long) getMaxValue(mongoSession, Collection.NANOPUBS.toString(), "counter");
        long effective = maxCounter != null ? maxCounter : 0L;
        collection("counters").updateOne(mongoSession,
                new Document("_id", "nanopubs"),
                new Document("$max", new Document("value", effective)),
                new UpdateOptions().upsert(true));
        collection("counters").updateOne(mongoSession,
                new Document("_id", "nanopubs-committed"),
                new Document("$max", new Document("value", effective)),
                new UpdateOptions().upsert(true));
        logger.info("Counter initialized to {}", effective);
    }

    /**
     * Returns the next counter value for a nanopub, using thread-local batch
     * allocation to reduce MongoDB contention.
     */
    private static long getNextCounter(ClientSession mongoSession) {
        long[] range = counterRange.get();
        if (range != null && range[0] < range[1]) {
            return range[0]++;
        }
        Document result = collection("counters").findOneAndUpdate(
                mongoSession,
                new Document("_id", "nanopubs"),
                new Document("$inc", new Document("value", (long) COUNTER_BATCH_SIZE)),
                new FindOneAndUpdateOptions().upsert(true).returnDocument(ReturnDocument.AFTER));
        long batchEnd = result.getLong("value");
        long batchStart = batchEnd - COUNTER_BATCH_SIZE + 1;
        counterRange.set(new long[]{batchStart + 1, batchEnd + 1});
        return batchStart;
    }

    /**
     * Updates the committed counter to the current max(counter) from nanopubs.
     * Call this after parallel loading completes to make new nanopubs visible to peers.
     */
    public static void updateCommittedCounter(ClientSession mongoSession) {
        Long maxCounter = (Long) getMaxValue(mongoSession, Collection.NANOPUBS.toString(), "counter");
        if (maxCounter != null) {
            collection("counters").updateOne(mongoSession,
                    new Document("_id", "nanopubs-committed"),
                    new Document("$max", new Document("value", maxCounter)),
                    new UpdateOptions().upsert(true));
        }
    }

    /**
     * Returns the committed counter value — the highest counter below which all
     * nanopubs are guaranteed to be inserted. Safe for peers to sync up to.
     */
    public static long getCommittedCounter(ClientSession mongoSession) {
        Document doc = collection("counters").find(mongoSession, new Document("_id", "nanopubs-committed")).first();
        return doc != null && doc.get("value") != null ? doc.getLong("value") : 0L;
    }

    /**
     * Loads a nanopublication into the database.
     *
     * @param mongoSession the MongoDB client session
     * @param nanopub      the nanopublication to load
     */
    public static boolean loadNanopub(ClientSession mongoSession, Nanopub nanopub) {
        return loadNanopub(mongoSession, nanopub, null);
    }

    /**
     * Loads a nanopublication into the database, optionally filtering by public key hash and types.
     *
     * @param mongoSession the MongoDB client session
     * @param nanopub      the nanopublication to load
     * @param pubkeyHash   the public key hash to filter by (can be null)
     * @param types        the types to filter by (can be empty)
     * @return true if the nanopublication was loaded, false otherwise
     */
    public static boolean loadNanopub(ClientSession mongoSession, Nanopub nanopub, String pubkeyHash, String... types) {
        String pubkey = getPubkey(nanopub);
        if (pubkey == null) {
            logger.info("Ignoring invalid nanopub: {}", nanopub.getUri());
            return false;
        }
        return loadNanopubVerified(mongoSession, nanopub, pubkey, pubkeyHash, types);
    }

    /**
     * Loads a nanopublication with a pre-verified public key, skipping signature verification.
     * Use this when the caller has already verified the signature via getPubkey().
     */
    static boolean loadNanopubVerified(ClientSession mongoSession, Nanopub nanopub, String verifiedPubkey, String pubkeyHash, String... types) {
        if (nanopub.getTripleCount() > 1200) {
            logger.info("Nanopub has too many triples ({}): {}", nanopub.getTripleCount(), nanopub.getUri());
            return false;
        }
        if (nanopub.getByteCount() > 1000000) {
            logger.info("Nanopub is too large ({}): {}", nanopub.getByteCount(), nanopub.getUri());
            return false;
        }
        Calendar creationTime;
        try {
            creationTime = nanopub.getCreationTime();
        } catch (Exception ex) {
            logger.info("Nanopub has malformed timestamp, treating as no timestamp: {}", nanopub.getUri());
            creationTime = null;
        }
        if (creationTime != null && creationTime.getTimeInMillis() > System.currentTimeMillis() + 60000) {
            logger.info("Nanopub has a future timestamp: {}", nanopub.getUri());
            return false;
        }
        String nanopubUriStr = nanopub.getUri().stringValue();
        for (IRI graphUri : nanopub.getGraphUris()) {
            if (!graphUri.stringValue().startsWith(nanopubUriStr)) {
                logger.info("Nanopub has graph URI not matching base URI: {}", nanopub.getUri());
                return false;
            }
        }
        String ph = Utils.getHash(verifiedPubkey);
        if (pubkeyHash != null && !pubkeyHash.equals(ph)) {
            logger.info("Ignoring nanopub with non-matching pubkey: {}", nanopub.getUri());
            return false;
        }
        recordHash(mongoSession, verifiedPubkey);

        String ac = TrustyUriUtils.getArtifactCode(nanopub.getUri().stringValue());
        if (ac == null) {
            // I don't think this ever happens, but checking here to be sure
            logger.info("ERROR. Unexpected Trusty URI: {}", nanopub.getUri());
            return false;
        }
        if (has(mongoSession, Collection.NANOPUBS.toString(), ac)) {
            logger.debug("Already loaded: {}", nanopub.getUri());
        } else {
            String nanopubString;
            byte[] jellyContent;
            try {
                nanopubString = NanopubUtils.writeToString(nanopub, RDFFormat.TRIG);
                // Save the same thing in the Jelly format for faster loading
                jellyContent = JellyUtils.writeNanopubForDB(nanopub);
            } catch (IOException ex) {
                throw new RuntimeException(ex);
            }
            long counter = getNextCounter(mongoSession);
            boolean inserted = false;
            try {
                collection(Collection.NANOPUBS.toString()).insertOne(mongoSession, new Document("_id", ac).append("fullId", nanopub.getUri().stringValue()).append("counter", counter).append("pubkey", ph).append("content", nanopubString).append("jelly", new Binary(jellyContent)));
                inserted = true;
            } catch (MongoWriteException e) {
                if (e.getError().getCategory() != ErrorCategory.DUPLICATE_KEY) throw e;
                // Another thread inserted this nanopub concurrently — safe to skip
                logger.debug("Already loaded (concurrent): {}", nanopub.getUri());
            }

            if (inserted) {
                for (IRI invalidatedId : Utils.getInvalidatedNanopubIds(nanopub)) {
                    String invalidatedAc = TrustyUriUtils.getArtifactCode(invalidatedId.stringValue());
                    if (invalidatedAc == null) continue;  // This should never happen; checking here just to be sure

                    // Add this nanopub also to all lists of invalidated nanopubs:
                    collection("invalidations").insertOne(mongoSession, new Document("invalidatingNp", ac).append("invalidatingPubkey", ph).append("invalidatedNp", invalidatedAc));
                    MongoCursor<Document> invalidatedEntries = collection("listEntries").find(mongoSession, new Document("np", invalidatedAc).append("pubkey", ph)).cursor();
                    while (invalidatedEntries.hasNext()) {
                        Document invalidatedEntry = invalidatedEntries.next();
                        addToList(mongoSession, nanopub, ph, invalidatedEntry.getString("type"));
                    }

                    collection("listEntries").updateMany(mongoSession, new Document("np", invalidatedAc).append("pubkey", ph), new Document("$set", new Document("invalidated", true)));
                    collection("trustEdges").updateMany(mongoSession, new Document("source", invalidatedAc), new Document("$set", new Document("invalidated", true)));
                }
            }
        }

        if (pubkeyHash != null) {
            for (String type : types) {
                // TODO Check if nanopub really has the type?
                addToList(mongoSession, nanopub, pubkeyHash, Utils.getTypeHash(mongoSession, type));
                if (type.equals("$")) {
                    for (IRI t : NanopubUtils.getTypes(nanopub)) {
                        String th = Utils.getTypeHash(mongoSession, t);
                        if (CoverageFilter.isCoveredType(th)) {
                            addToList(mongoSession, nanopub, pubkeyHash, th);
                        }
                    }
                }
            }
        }

        // Add the invalidating nanopubs also to the lists of this nanopub:
        try (MongoCursor<Document> invalidations = collection("invalidations").find(mongoSession, new Document("invalidatedNp", ac).append("invalidatingPubkey", ph)).cursor()) {
            if (invalidations.hasNext()) {
                collection("listEntries").updateMany(mongoSession, new Document("np", ac).append("pubkey", ph), new Document("$set", new Document("invalidated", true)));
                collection("trustEdges").updateMany(mongoSession, new Document("source", ac), new Document("$set", new Document("invalidated", true)));
            }
            while (invalidations.hasNext()) {
                String iac = invalidations.next().getString("invalidatingNp");
                try {
                    Document npDoc = collection(Collection.NANOPUBS.toString()).find(mongoSession, new Document("_id", iac)).projection(new Document("jelly", 1)).first();
                    Nanopub inp = JellyUtils.readFromDB(npDoc.get("jelly", Binary.class).getData());
                    for (IRI type : NanopubUtils.getTypes(inp)) {
                        addToList(mongoSession, inp, ph, Utils.getTypeHash(mongoSession, type));
                    }
                } catch (RDF4JException | MalformedNanopubException ex) {
                    ex.printStackTrace();
                }
            }

        }

        return true;
    }

    private static void addToList(ClientSession mongoSession, Nanopub nanopub, String pubkeyHash, String typeHash) {
        String ac = TrustyUriUtils.getArtifactCode(nanopub.getUri().stringValue());
        try {
            insert(mongoSession, "lists", new Document("pubkey", pubkeyHash).append("type", typeHash)
                    .append("maxPosition", -1L));
        } catch (MongoWriteException e) {
            // Duplicate key error -- ignore it
            if (e.getError().getCategory() != ErrorCategory.DUPLICATE_KEY) throw e;
        }

        if (has(mongoSession, "listEntries", new Document("pubkey", pubkeyHash).append("type", typeHash).append("np", ac))) {
            logger.debug("Already listed: {}", nanopub.getUri());
        } else {
            initListPositionIfNeeded(mongoSession, pubkeyHash, typeHash);

            for (int attempt = 0; ; attempt++) {
                // Atomically claim next position
                Document updated = collection("lists").findOneAndUpdate(mongoSession,
                        new Document("pubkey", pubkeyHash).append("type", typeHash),
                        new Document("$inc", new Document("maxPosition", 1L)),
                        new FindOneAndUpdateOptions().returnDocument(ReturnDocument.AFTER));
                long position = updated.getLong("maxPosition");

                // Get checksum from previous entry by exact position lookup (O(1) index hit)
                String checksum;
                if (position == 0) {
                    checksum = NanopubUtils.updateXorChecksum(nanopub.getUri(), NanopubUtils.INIT_CHECKSUM);
                } else {
                    Document prevEntry = collection("listEntries").find(mongoSession,
                            new Document("pubkey", pubkeyHash).append("type", typeHash)
                                    .append("position", position - 1)).first();
                    String prevChecksum = (prevEntry != null) ? prevEntry.getString("checksum") : null;
                    if (prevChecksum == null) {
                        // Rare: previous entry not yet inserted by concurrent thread; fall back to sorted query
                        Document maxDoc = getMaxValueDocument(mongoSession, "listEntries",
                                new Document("pubkey", pubkeyHash).append("type", typeHash), "position");
                        prevChecksum = (maxDoc != null) ? maxDoc.getString("checksum") : NanopubUtils.INIT_CHECKSUM;
                    }
                    checksum = NanopubUtils.updateXorChecksum(nanopub.getUri(), prevChecksum);
                }

                try {
                    collection("listEntries").insertOne(mongoSession, new Document("pubkey", pubkeyHash)
                            .append("type", typeHash).append("position", position).append("np", ac)
                            .append("checksum", checksum).append("invalidated", false));
                    break;
                } catch (MongoWriteException e) {
                    if (e.getError().getCategory() != ErrorCategory.DUPLICATE_KEY) throw e;
                    if (has(mongoSession, "listEntries", new Document("pubkey", pubkeyHash)
                            .append("type", typeHash).append("np", ac))) {
                        break; // Already listed by concurrent thread
                    }
                    if (attempt >= 100) {
                        throw new RuntimeException("Failed to insert list entry after " + (attempt + 1) + " attempts");
                    }
                }
            }
        }
    }

    /**
     * Lazily initializes the maxPosition field on a lists document for lists
     * created before this field existed. Uses a one-time sorted query, then
     * all subsequent calls use the atomic counter.
     */
    private static void initListPositionIfNeeded(ClientSession mongoSession, String pubkeyHash, String typeHash) {
        Document listDoc = collection("lists").find(mongoSession,
                new Document("pubkey", pubkeyHash).append("type", typeHash)).first();
        if (listDoc == null || listDoc.get("maxPosition") != null) return;

        Document maxDoc = getMaxValueDocument(mongoSession, "listEntries",
                new Document("pubkey", pubkeyHash).append("type", typeHash), "position");
        long maxPos = (maxDoc != null) ? maxDoc.getLong("position") : -1L;

        // Conditional update: only set if maxPosition still doesn't exist (race-safe)
        collection("lists").updateOne(mongoSession,
                new Document("pubkey", pubkeyHash).append("type", typeHash)
                        .append("maxPosition", new Document("$exists", false)),
                new Document("$set", new Document("maxPosition", maxPos)));
    }

    /**
     * Builds a comma-separated list of checksums at geometric positions for a given list,
     * for use with the afterChecksums parameter during peer sync.
     * Returns checksums at positions: max, max-10, max-100, max-1000, max-10000, ...
     * Returns null if the list has no entries.
     */
    public static String buildChecksumFallbacks(ClientSession mongoSession, String pubkeyHash, String typeHash) {
        Document maxDoc = getMaxValueDocument(mongoSession, "listEntries",
                new Document("pubkey", pubkeyHash).append("type", typeHash), "position");
        if (maxDoc == null) return null;

        long maxPosition = maxDoc.getLong("position");
        StringBuilder sb = new StringBuilder();
        sb.append(maxDoc.getString("checksum"));

        for (long offset = 10; offset <= maxPosition; offset *= 10) {
            long targetPos = maxPosition - offset;
            Document entry = collection("listEntries").find(mongoSession,
                    new Document("pubkey", pubkeyHash).append("type", typeHash)
                            .append("position", targetPos)).first();
            if (entry != null) {
                sb.append(",").append(entry.getString("checksum"));
            }
        }
        return sb.toString();
    }

    /**
     * Returns the public key string of the Nanopub's signature, or null if not available or invalid.
     *
     * @param nanopub the nanopub to extract the public key from
     * @return The public key string, or null if not available or invalid.
     */
    public static String getPubkey(Nanopub nanopub) {
        // TODO shouldn't this be moved to a utility class in nanopub-java? there is a similar method in NanopubElement class of nanodash
        NanopubSignatureElement el;
        try {
            el = SignatureUtils.getSignatureElement(nanopub);
            if (el != null && SignatureUtils.hasValidSignature(el) && el.getPublicKeyString() != null) {
                return el.getPublicKeyString();
            }
        } catch (MalformedCryptoElementException | GeneralSecurityException ex) {
            logger.error("Error in checking the signature of the nanopub {}", nanopub.getUri());
        }
        return null;
    }

    /**
     * Calculates a hash representing the current state of the trust paths in the loading collection.
     *
     * @param mongoSession the MongoDB client session
     * @return the calculated trust state hash
     */
    public static String calculateTrustStateHash(ClientSession mongoSession) {
        MongoCursor<Document> tp = collection("trustPaths_loading").find(mongoSession).sort(ascending("_id")).cursor();
        // TODO Improve this so we don't create the full string just for calculating its hash:
        String s = "";
        while (tp.hasNext()) {
            Document d = tp.next();
            s += d.getString("_id") + " (" + d.getString("type") + ")\n";
        }
        return Utils.getHash(s);
    }

}

knowledgepixels / nanopub-registry / 24138607049

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous