30893238389

Committed 04 Aug 2026 08:44AM UTC coverage: 83.415% (+51.1%) from 32.285%

Build # 30893238389

Build Type

Pull #123

github

Committed by

web-flow

Commit Message

Merge 84fad06c1 into e5ed6d462

Pull Request Pull Request #123: Add unit tests for existing functionality

Coverage Stats

865 of 1112 branches covered (77.79%)

Branch coverage included in aggregate %.

2706 of 3169 relevant lines covered (85.39%)

12.96 hits per line

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

71.13

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

package com.knowledgepixels.registry;

import com.knowledgepixels.registry.db.IndexInitializer;
import com.mongodb.*;
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 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 java.util.Calendar;
import java.util.HashSet;
import java.util.Set;

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 (database: {})", REGISTRY_DB_NAME);
            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()) {
            logger.debug("MongoDB client session started for initialization");
            if (!isInitialized(mongoSession)) {
                logger.info("Database '{}' not initialized; creating collections and indexes", REGISTRY_DB_NAME);
                IndexInitializer.initCollections(mongoSession);
            } else {
                logger.debug("Database '{}' already has setupId", REGISTRY_DB_NAME);
            }
            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) {
        boolean initialized = getValue(mongoSession, Collection.SERVER_INFO.toString(), "setupId") != null;
        logger.debug("isInitialized check for database '{}': {}", REGISTRY_DB_NAME, initialized);
        return initialized;
    }

    /**
     * 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) {
        boolean exists = mongoDB.listCollectionNames().into(new ArrayList<>()).contains(collectionName);
        logger.debug("Collection existence check for '{}': {}", collectionName, exists);
        return exists;
    }

    /**
     * Increases the trust state counter in the server info collection.
     *
     * @param mongoSession the MongoDB client session
     */
    public static void increaseStateCounter(ClientSession mongoSession) {
        try (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)));
                logger.debug("Incremented trustStateCounter from {} to {}", counter, counter + 1);
            } else {
                collection(Collection.SERVER_INFO.toString()).insertOne(mongoSession, new Document("_id", "trustStateCounter").append("value", 0L));
                logger.info("Initialized trustStateCounter to 0 in collection '{}'", Collection.SERVER_INFO);
            }
        }
    }

    /**
     * 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) {
        boolean found = collection(collection).countDocuments(mongoSession, find, hasCountOptions) > 0;
        logger.debug("Existence check in collection '{}' for filter {}: {}", collection, find, found);
        return found;
    }

    /**
     * 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) {
        logger.trace("Querying collection '{}' with filter {}", collection, 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.debug("Reading value of element '{}' from collection '{}'", elementName, collection);
        Document d = collection(collection).find(mongoSession, new Document("_id", elementName)).first();
        if (d == null) {
            logger.trace("Element '{}' not found in collection '{}'", elementName, collection);
            return null;
        }
        Object value = d.get("value");
        logger.debug("Found element '{}' in collection '{}' with value type {}", elementName, collection, value == null ? "null" : value.getClass().getSimpleName());
        return 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) {
            logger.trace("isSet: element '{}' not found in collection '{}'", elementName, collection);
            return false;
        }
        Boolean val = d.getBoolean("value");
        logger.debug("isSet: element '{}' in collection '{}' has boolean value {}", elementName, collection, val);
        return val;
    }

    /**
     * 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) {
        logger.trace("getOne from '{}' with filter {}", collection, 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) {
            logger.trace("getMaxValue: no documents in collection '{}' for field '{}'", collection, fieldName);
            return null;
        }
        Object val = doc.get(fieldName);
        logger.debug("getMaxValue: collection '{}' field '{}' max = {}", collection, fieldName, val);
        return val;
    }

    /**
     * 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) {
        logger.trace("getMaxValueDocument in '{}' with filter {} for field '{}'", collection, find, 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"));
        try (MongoCursor<Document> cursor = collection(collection).find(mongoSession, find).cursor()) {
            if (cursor.hasNext()) {
                collection(collection).updateOne(mongoSession, find, new Document("$set", update));
                logger.debug("Updated document with _id={} in collection '{}'", update.get("_id"), collection);
            } else {
                logger.debug("set: no existing document with _id={} in collection '{}'; update skipped", update.get("_id"), collection);
            }
        }
    }

    /**
     * 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);
        logger.debug("Inserted document into '{}' with _id={}", collection, doc.get("_id"));
    }

    /**
     * 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) {
        logger.debug("Setting value for element '{}' in collection '{}' (upsert)", elementId, collection);
        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));
            logger.debug("Recorded hash for value (hash={})", hash);
        } catch (MongoWriteException e) {
            // Concurrent upsert race: another thread inserted the same hash — safe to ignore
            if (e.getError().getCategory() != ErrorCategory.DUPLICATE_KEY) {
                logger.error("Failed to record hash for value (hash={}): {}", hash, e.getMessage(), e);
                throw e;
            }
            logger.debug("Concurrent insertion for hash {} detected; duplicate ignored", hash);
        }
    }

    /**
     * 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()) {
                String value = c.next().getString("value");
                logger.debug("Unhash found value for hash {}", hash);
                return value;
            }
            logger.debug("Unhash: no value found for hash {}", hash);
            return null;
        }
    }

    /**
     * Initializes the counter document to the current maximum counter value
     * in the nanopubs collection.
     * Uses $max to ensure the counter is 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));
        if (maxCounter != null) {
            logger.info("Nanopub counter resumed at {} (max found in DB)", effective);
        } else {
            logger.info("Nanopub counter initialized to 0 (no existing nanopubs found)");
        }
    }

    /**
     * Returns the next counter value for a nanopub via atomic increment.
     */
    private static long getNextCounter(ClientSession mongoSession) {
        Document result = collection("counters").findOneAndUpdate(mongoSession, new Document("_id", "nanopubs"), new Document("$inc", new Document("value", 1L)), new FindOneAndUpdateOptions().upsert(true).returnDocument(ReturnDocument.AFTER));
        return result.getLong("value");
    }

    /**
     * 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.warn("Ignoring nanopub {}: no valid public key / signature found", 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.error("Rejecting nanopub {}: triple count {} exceeds limit of 1200", nanopub.getUri(), nanopub.getTripleCount());
            return false;
        }
        if (nanopub.getByteCount() > 1000000) {
            logger.error("Rejecting nanopub {}: size {} bytes exceeds limit of 1000000", nanopub.getUri(), nanopub.getByteCount());
            return false;
        }
        Calendar creationTime;
        try {
            creationTime = nanopub.getCreationTime();
        } catch (Exception ex) {
            logger.warn("Nanopub {} has a malformed timestamp; proceeding without one", nanopub.getUri());
            creationTime = null;
        }
        if (creationTime != null && creationTime.getTimeInMillis() > System.currentTimeMillis() + 60000) {
            logger.error("Rejecting nanopub {}: timestamp {} is more than 60s in the future", nanopub.getUri(), creationTime.toInstant());
            return false;
        }
        String nanopubUriStr = nanopub.getUri().stringValue();
        for (IRI graphUri : nanopub.getGraphUris()) {
            if (!graphUri.stringValue().startsWith(nanopubUriStr)) {
                logger.error("Rejecting nanopub {}: graph URI {} does not start with the nanopub base URI", nanopub.getUri(), graphUri);
                return false;
            }
        }
        String ph = Utils.getHash(verifiedPubkey);
        if (pubkeyHash != null && !pubkeyHash.equals(ph)) {
            logger.error("Rejecting nanopub {}: provided pubkey hash {} does not match computed hash {}", nanopub.getUri(), pubkeyHash, ph);
            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.error("Rejecting nanopub {}: could not extract artifact code from Trusty URI", nanopub.getUri());
            return false;
        }
        if (has(mongoSession, Collection.NANOPUBS.toString(), ac)) {
            logger.debug("Skipping nanopub {}: already present in the database", 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) {
                logger.error("Failed to serialize nanopub {}: {}", nanopub.getUri(), ex.getMessage(), 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;
                logger.info("Loaded nanopub {} (counter: {}, pubkey hash: {})", nanopub.getUri(), counter, ph);
            } catch (MongoWriteException e) {
                if (e.getError().getCategory() != ErrorCategory.DUPLICATE_KEY) {
                    logger.error("Failed to insert nanopub {} (artifact {}): {}", nanopub.getUri(), ac, e.getMessage(), e);
                    throw e;
                }
                // Another thread inserted this nanopub concurrently — safe to skip
                logger.debug("Skipping nanopub {}: inserted concurrently by another thread", nanopub.getUri());
            }

            if (inserted) {
                for (IRI invalidatedId : Utils.getInvalidatedNanopubIds(nanopub)) {
                    String invalidatedAc = TrustyUriUtils.getArtifactCode(invalidatedId.stringValue());
                    if (invalidatedAc == null) {
                        logger.warn("Nanopub {} references invalidated nanopub {} with an unresolvable artifact code; skipping", nanopub.getUri(), invalidatedId);
                        continue;  // This should never happen; checking here just to be sure
                    }

                    // Add this nanopub also to all lists of invalidated nanopubs:
                    logger.debug("Nanopub {} invalidates {}; updating list entries and trust edges", nanopub.getUri(), invalidatedId);
                    collection("invalidations").insertOne(mongoSession, new Document("invalidatingNp", ac).append("invalidatingPubkey", ph).append("invalidatedNp", invalidatedAc));
                    try (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)));
                    logger.debug("Marked invalidated entries and trust edges for invalidated artifact {}", invalidatedAc);
                }
            }
        }

        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)));
                logger.debug("Marked existing list entries and trust edges for nanopub {} as invalidated due to invalidations", ac);
            }
            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) {
                    logger.error("Failed to load invalidating nanopub {} for invalidation record; skipping", iac, ex);
                }
            }

        }

        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));
            logger.debug("Ensured list document exists for pubkey={} type={}", pubkeyHash, typeHash);
        } catch (MongoWriteException e) {
            // Duplicate key error -- ignore it
            if (e.getError().getCategory() != ErrorCategory.DUPLICATE_KEY) {
                logger.error("Failed to create list document for pubkey={} type={}: {}", pubkeyHash, typeHash, e.getMessage(), e);
                throw e;
            }
            logger.trace("List document already existed for pubkey={} type={}", pubkeyHash, typeHash);
        }

        if (has(mongoSession, "listEntries", new Document("pubkey", pubkeyHash).append("type", typeHash).append("np", ac))) {
            logger.debug("Already listed: nanopub {} (artifact {}) for pubkey={} type={}", nanopub.getUri(), ac, pubkeyHash, typeHash);
        } 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));
                    logger.debug("Inserted list entry: pubkey={} type={} np={} position={} checksum={}", pubkeyHash, typeHash, ac, position, checksum);
                    break;
                } catch (MongoWriteException e) {
                    if (e.getError().getCategory() != ErrorCategory.DUPLICATE_KEY) {
                        logger.error("Failed to insert list entry for pubkey={} type={} np={}: {}", pubkeyHash, typeHash, ac, e.getMessage(), e);
                        throw e;
                    }
                    if (has(mongoSession, "listEntries", new Document("pubkey", pubkeyHash).append("type", typeHash).append("np", ac))) {
                        logger.debug("Concurrent insert detected and entry already exists for pubkey={} type={} np={}", pubkeyHash, typeHash, ac);
                        break; // Already listed by concurrent thread
                    }
                    if (attempt >= 100) {
                        logger.error("Failed to insert list entry after {} attempts for pubkey={} type={} np={}", attempt + 1, pubkeyHash, typeHash, ac);
                        throw new RuntimeException("Failed to insert list entry after " + (attempt + 1) + " attempts");
                    }
                    logger.debug("Retrying list entry insert (attempt {}) for pubkey={} type={} np={}", attempt + 1, pubkeyHash, typeHash, ac);
                }
            }
        }
    }

    /**
     * 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) {
            logger.trace("initListPositionIfNeeded: no action needed for pubkey={} type={}", pubkeyHash, typeHash);
            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)));
        logger.debug("Initialized maxPosition={} for list pubkey={} type={}", maxPos, pubkeyHash, typeHash);
    }

    /**
     * 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) {
            logger.debug("buildChecksumFallbacks: no entries for pubkey={} type={}", pubkeyHash, typeHash);
            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"));
            }
        }
        String result = sb.toString();
        logger.debug("buildChecksumFallbacks for pubkey={} type={} -> {}", pubkeyHash, typeHash, result);
        return result;
    }

    /**
     * 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) {
                logger.trace("Valid signature found for nanopub {}", nanopub.getUri());
                return el.getPublicKeyString();
            }
            logger.debug("No valid signature element or public key present for nanopub {}", nanopub.getUri());
        } catch (MalformedCryptoElementException | GeneralSecurityException ex) {
            logger.error("Failed to verify signature of nanopub {}: {}", nanopub.getUri(), ex.getMessage(), ex);
        }
        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) {
        // Accounts still in the 'toLoad' staging status are not yet servable and must not enter the
        // public trust state: a path's head account flips 'toLoad' -> 'loaded' between UPDATE cycles
        // (in LOAD_FULL) without changing any trust path, so if those paths were hashed, a leaf
        // account's promotion would never change the hash and never trigger a new snapshot, leaving
        // it published as 'toLoad' indefinitely (see issue #119). By excluding 'toLoad' paths here,
        // crossing out of 'toLoad' is itself a membership change -> hash change -> emission.
        Set<String> toLoadAccounts = new HashSet<>();
        try (MongoCursor<Document> ac = collection("accounts_loading")
                .find(mongoSession, new Document("status", EntryStatus.toLoad.getValue()))
                .projection(new Document("agent", 1).append("pubkey", 1))
                .cursor()) {
            while (ac.hasNext()) {
                Document a = ac.next();
                toLoadAccounts.add(a.getString("agent") + "|" + a.getString("pubkey"));
            }
        }

        String s;
        try (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:
            s = "";
            while (tp.hasNext()) {
                Document d = tp.next();
                if (toLoadAccounts.contains(d.getString("agent") + "|" + d.getString("pubkey"))) {
                    continue;
                }
                s += d.getString("_id") + " (" + d.getString("type") + ")\n";
            }
        }
        String hash = Utils.getHash(s);
        logger.debug("Calculated trust state hash: {}", hash);
        return hash;
    }

}

knowledgepixels / nanopub-registry / 30893238389

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