21060548866

Committed 16 Jan 2026 08:33AM UTC coverage: 11.504% (+0.4%) from 11.102%

Build # 21060548866

Build Type

push

github

Committed by

ashleycaselli

Commit Message

chore: add command for updating git submodule before testing

Run Details

58 of 588 branches covered (9.86%)

Branch coverage included in aggregate %.

215 of 1785 relevant lines covered (12.04%)

2.1 hits per line

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

20.07

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

package com.knowledgepixels.registry;

import com.mongodb.MongoClient;
import com.mongodb.MongoNamespace;
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.IndexOptions;
import com.mongodb.client.model.Indexes;
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 static com.mongodb.client.model.Indexes.*;

public class RegistryDB {

    private RegistryDB() {
    }

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

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

    private static MongoClient mongoClient;
    private static MongoDatabase mongoDB;

    public static MongoDatabase getDB() {
        return mongoDB;
    }

    public static MongoClient getClient() {
        return mongoClient;
    }

    public static MongoCollection<Document> collection(String name) {
        return mongoDB.getCollection(name);
    }

    private final static IndexOptions unique = new IndexOptions().unique(true);

    public static void init() {
        if (mongoClient != null) {
            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())));
        mongoClient = new MongoClient(REGISTRY_DB_HOST, REGISTRY_DB_PORT);
        mongoDB = mongoClient.getDatabase(REGISTRY_DB_NAME);

        try (ClientSession mongoSession = mongoClient.startSession()) {
            if (isInitialized(mongoSession)) {
                return;
            }

            final IndexOptions unique = new IndexOptions().unique(true);

            collection("tasks").createIndex(mongoSession, Indexes.descending("not-before"));

            collection(Collection.NANOPUBS.toString()).createIndex(mongoSession, ascending("fullId"), unique);
            collection(Collection.NANOPUBS.toString()).createIndex(mongoSession, descending("counter"), unique);
            collection(Collection.NANOPUBS.toString()).createIndex(mongoSession, ascending("pubkey"));

            collection("lists").createIndex(mongoSession, ascending("pubkey", "type"), unique);
            collection("lists").createIndex(mongoSession, ascending("status"));

            collection("listEntries").createIndex(mongoSession, ascending("np"));
            collection("listEntries").createIndex(mongoSession, ascending("pubkey", "type", "np"), unique);
            collection("listEntries").createIndex(mongoSession, compoundIndex(ascending("pubkey"), ascending("type"), descending("position")), unique);
            collection("listEntries").createIndex(mongoSession, ascending("pubkey", "type", "checksum"), unique);
            collection("listEntries").createIndex(mongoSession, ascending("invalidated"));

            collection("invalidations").createIndex(mongoSession, ascending("invalidatingNp"));
            collection("invalidations").createIndex(mongoSession, ascending("invalidatingPubkey"));
            collection("invalidations").createIndex(mongoSession, ascending("invalidatedNp"));
            collection("invalidations").createIndex(mongoSession, ascending("invalidatingPubkey", "invalidatedNp"));

            collection("trustEdges").createIndex(mongoSession, ascending("fromAgent"));
            collection("trustEdges").createIndex(mongoSession, ascending("fromPubkey"));
            collection("trustEdges").createIndex(mongoSession, ascending("toAgent"));
            collection("trustEdges").createIndex(mongoSession, ascending("toPubkey"));
            collection("trustEdges").createIndex(mongoSession, ascending("source"));
            collection("trustEdges").createIndex(mongoSession, ascending("fromAgent", "fromPubkey", "toAgent", "toPubkey", "source"), unique);
            collection("trustEdges").createIndex(mongoSession, ascending("invalidated"));

            collection("hashes").createIndex(mongoSession, ascending("hash"), unique);
            collection("hashes").createIndex(mongoSession, ascending("value"), unique);
        }
    }

    public static void initLoadingCollections(ClientSession mongoSession) {
        collection("endorsements_loading").createIndex(mongoSession, ascending("agent"));
        collection("endorsements_loading").createIndex(mongoSession, ascending("pubkey"));
        collection("endorsements_loading").createIndex(mongoSession, ascending("endorsedNanopub"));
        collection("endorsements_loading").createIndex(mongoSession, ascending("source"));
        collection("endorsements_loading").createIndex(mongoSession, ascending("status"));
        // zip: Hmm, not possible to set com.mongodb.client.model.WriteModel<T> here
        // I'd currently recommend to have a package with the DB related stuff
        // and therein a Custom Extension T of <Document>, where we could put that WriteModel<T>.

        collection("agents_loading").createIndex(mongoSession, ascending("agent"), unique);
        collection("agents_loading").createIndex(mongoSession, descending("accountCount"));
        collection("agents_loading").createIndex(mongoSession, descending("avgPathCount"));
        collection("agents_loading").createIndex(mongoSession, descending("totalRatio"));

        collection("accounts_loading").createIndex(mongoSession, ascending("agent"));
        collection("accounts_loading").createIndex(mongoSession, ascending("pubkey"));
        collection("accounts_loading").createIndex(mongoSession, ascending("agent", "pubkey"), unique);
        collection("accounts_loading").createIndex(mongoSession, ascending("type"));
        collection("accounts_loading").createIndex(mongoSession, ascending("status"));
        collection("accounts_loading").createIndex(mongoSession, descending("ratio"));
        collection("accounts_loading").createIndex(mongoSession, descending("pathCount"));

        collection("trustPaths_loading").createIndex(mongoSession, ascending("agent", "pubkey", "depth", "sorthash"), unique);
        collection("trustPaths_loading").createIndex(mongoSession, ascending("depth"));
        collection("trustPaths_loading").createIndex(mongoSession, descending("ratio"));
    }

    public static boolean isInitialized(ClientSession mongoSession) {
        return getValue(mongoSession, Collection.SERVER_INFO.toString(), "setupId") != null;
    }

    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));
        }
    }

    public static boolean hasCollection(String collectionName) {
        return mongoDB.listCollectionNames().into(new ArrayList<String>()).contains(collectionName);
    }

    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));
        }
    }

    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);

    public static boolean has(ClientSession mongoSession, String collection, Bson find) {
        return collection(collection).countDocuments(mongoSession, find, hasCountOptions) > 0;
    }

    public static MongoCursor<Document> get(ClientSession mongoSession, String collection, Bson find) {
        return collection(collection).find(mongoSession, find).cursor();
    }

    public static Object getValue(ClientSession mongoSession, String collection, String elementName) {
        Document d = collection(collection).find(mongoSession, new Document("_id", elementName)).first();
        if (d == null) return null;
        return d.get("value");
    }

    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");
    }

    public static Document getOne(ClientSession mongoSession, String collection, Bson find) {
        return collection(collection).find(mongoSession, find).first();
    }

    public static Document getOne(ClientSession mongoSession, String collection, Bson find, Bson sort) {
        return collection(collection).find(mongoSession, find).sort(sort).first();
    }

    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);
    }

    public static Object getMaxValue(ClientSession mongoSession, String collection, Bson find, String fieldName) {
        Document doc = collection(collection).find(mongoSession, find).projection(new Document(fieldName, 1)).sort(new Document(fieldName, -1)).first();
        if (doc == null) return null;
        return doc.get(fieldName);
    }

    public static Document getMaxValueDocument(ClientSession mongoSession, String collection, Bson find, String fieldName) {
        return collection(collection).find(mongoSession, find).sort(new Document(fieldName, -1)).first();
    }

    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));
        }
    }

    public static void insert(ClientSession mongoSession, String collection, Document doc) {
        collection(collection).insertOne(mongoSession, doc);
    }

    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));
    }

    public static void recordHash(ClientSession mongoSession, String value) {
        try {
            insert(mongoSession, "hashes", new Document("value", value).append("hash", Utils.getHash(value)));
        } catch (MongoWriteException e) {
            // Duplicate key error -- ignore it
            if (e.getError().getCode() != 11000) throw e;
        }
    }

    public static String unhash(String hash) {
        try (var c = collection("hashes").find(new Document("hash", hash)).cursor()) {
            if (c.hasNext()) return c.next().get("value").toString();
            return null;
        }
    }

    /**
     * Insert nanopub to the DB.
     */
    public static boolean loadNanopub(ClientSession mongoSession, Nanopub nanopub) {
        return loadNanopub(mongoSession, nanopub, null);
    }

    public static boolean loadNanopub(ClientSession mongoSession, Nanopub nanopub, String pubkeyHash, String... types) {
        if (nanopub.getTripleCount() > 1200) {
            log.info("Nanopub has too many triples ({}): {}", nanopub.getTripleCount(), nanopub.getUri());
            return false;
        }
        if (nanopub.getByteCount() > 1000000) {
            log.info("Nanopub is too large ({}): {}", nanopub.getByteCount(), nanopub.getUri());
            return false;
        }
        String pubkey = getPubkey(nanopub);
        if (pubkey == null) {
            log.info("Ignoring invalid nanopub: {}", nanopub.getUri());
            return false;
        }
        String ph = Utils.getHash(pubkey);
        if (pubkeyHash != null && !pubkeyHash.equals(ph)) {
            log.info("Ignoring nanopub with non-matching pubkey: {}", nanopub.getUri());
            return false;
        }
        recordHash(mongoSession, pubkey);

        String ac = TrustyUriUtils.getArtifactCode(nanopub.getUri().stringValue());
        if (ac == null) {
            // I don't think this ever happens, but checking here to be sure
            log.info("ERROR. Unexpected Trusty URI: {}", nanopub.getUri());
            return false;
        }
        if (has(mongoSession, Collection.NANOPUBS.toString(), ac)) {
            log.info("Already loaded: {}", nanopub.getUri());
        } else {
            Long counter = (Long) getMaxValue(mongoSession, Collection.NANOPUBS.toString(), "counter");
            if (counter == null) counter = 0l;
            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);
            }
            collection(Collection.NANOPUBS.toString()).insertOne(mongoSession, new Document("_id", ac).append("fullId", nanopub.getUri().stringValue()).append("counter", counter + 1).append("pubkey", ph).append("content", nanopubString).append("jelly", new Binary(jellyContent)));

            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)) {
                        addToList(mongoSession, nanopub, pubkeyHash, Utils.getTypeHash(mongoSession, t));
                    }
                }
            }
        }

        // 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));
        } catch (MongoWriteException e) {
            // Duplicate key error -- ignore it
            if (e.getError().getCode() != 11000) throw e;
        }

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

            Document doc = getMaxValueDocument(mongoSession, "listEntries", new Document("pubkey", pubkeyHash).append("type", typeHash), "position");
            long position;
            String checksum;
            if (doc == null) {
                position = 0l;
                checksum = NanopubUtils.updateXorChecksum(nanopub.getUri(), NanopubUtils.INIT_CHECKSUM);
            } else {
                position = doc.getLong("position") + 1;
                checksum = NanopubUtils.updateXorChecksum(nanopub.getUri(), doc.getString("checksum"));
            }
            collection("listEntries").insertOne(mongoSession, new Document("pubkey", pubkeyHash).append("type", typeHash).append("position", position).append("np", ac).append("checksum", checksum).append("invalidated", false));
        }
    }

    /**
     * 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) {
            log.error("Error in checking the signature of the nanopub {}", nanopub.getUri());
        }
        return null;
    }

    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);
    }

}

1	package com.knowledgepixels.registry;
2
3	import com.mongodb.MongoClient;
4	import com.mongodb.MongoNamespace;
5	import com.mongodb.MongoWriteException;
6	import com.mongodb.ServerAddress;
7	import com.mongodb.client.ClientSession;
8	import com.mongodb.client.MongoCollection;
9	import com.mongodb.client.MongoCursor;
10	import com.mongodb.client.MongoDatabase;
11	import com.mongodb.client.model.CountOptions;
12	import com.mongodb.client.model.IndexOptions;
13	import com.mongodb.client.model.Indexes;
14	import com.mongodb.client.model.UpdateOptions;
15	import net.trustyuri.TrustyUriUtils;
16	import org.bson.Document;
17	import org.bson.conversions.Bson;
18	import org.bson.types.Binary;
19	import org.eclipse.rdf4j.common.exception.RDF4JException;
20	import org.eclipse.rdf4j.model.IRI;
21	import org.eclipse.rdf4j.rio.RDFFormat;
22	import org.nanopub.MalformedNanopubException;
23	import org.nanopub.Nanopub;
24	import org.nanopub.NanopubUtils;
25	import org.nanopub.extra.security.MalformedCryptoElementException;
26	import org.nanopub.extra.security.NanopubSignatureElement;
27	import org.nanopub.extra.security.SignatureUtils;
28	import org.nanopub.jelly.JellyUtils;
29	import org.slf4j.Logger;
30	import org.slf4j.LoggerFactory;
31
32	import java.io.IOException;
33	import java.security.GeneralSecurityException;
34	import java.util.ArrayList;
35
36	import static com.mongodb.client.model.Indexes.*;
37
38	public class RegistryDB {
39
40	private RegistryDB() {
41	}
42
43	private static final String REGISTRY_DB_NAME = Utils.getEnv("REGISTRY_DB_NAME", "nanopubRegistry");	12✔
44
45	private static final Logger log = LoggerFactory.getLogger(RegistryDB.class);	9✔
46
47	private static MongoClient mongoClient;
48	private static MongoDatabase mongoDB;
49
50	public static MongoDatabase getDB() {
51	return mongoDB;	6✔
52	}
53
54	public static MongoClient getClient() {
55	return mongoClient;	6✔
56	}
57
58	public static MongoCollection<Document> collection(String name) {
59	return mongoDB.getCollection(name);	12✔
60	}
61
62	private final static IndexOptions unique = new IndexOptions().unique(true);	18✔
63
64	public static void init() {
65	if (mongoClient != null) {	6✔
66	return;	3✔
67	}
68	final String REGISTRY_DB_HOST = Utils.getEnv("REGISTRY_DB_HOST", "mongodb");	12✔
69	final int REGISTRY_DB_PORT = Integer.parseInt(Utils.getEnv("REGISTRY_DB_PORT", String.valueOf(ServerAddress.defaultPort())));	18✔
70	mongoClient = new MongoClient(REGISTRY_DB_HOST, REGISTRY_DB_PORT);	18✔
71	mongoDB = mongoClient.getDatabase(REGISTRY_DB_NAME);	12✔
72
73	try (ClientSession mongoSession = mongoClient.startSession()) {	9✔
74	if (isInitialized(mongoSession)) {	9!
75	return;	×
76	}
77
78	final IndexOptions unique = new IndexOptions().unique(true);	18✔
79
80	collection("tasks").createIndex(mongoSession, Indexes.descending("not-before"));	36✔
81
82	collection(Collection.NANOPUBS.toString()).createIndex(mongoSession, ascending("fullId"), unique);	42✔
83	collection(Collection.NANOPUBS.toString()).createIndex(mongoSession, descending("counter"), unique);	42✔
84	collection(Collection.NANOPUBS.toString()).createIndex(mongoSession, ascending("pubkey"));	39✔
85
86	collection("lists").createIndex(mongoSession, ascending("pubkey", "type"), unique);	51✔
87	collection("lists").createIndex(mongoSession, ascending("status"));	36✔
88
89	collection("listEntries").createIndex(mongoSession, ascending("np"));	36✔
90	collection("listEntries").createIndex(mongoSession, ascending("pubkey", "type", "np"), unique);	63✔
91	collection("listEntries").createIndex(mongoSession, compoundIndex(ascending("pubkey"), ascending("type"), descending("position")), unique);	117✔
92	collection("listEntries").createIndex(mongoSession, ascending("pubkey", "type", "checksum"), unique);	63✔
93	collection("listEntries").createIndex(mongoSession, ascending("invalidated"));	36✔
94
95	collection("invalidations").createIndex(mongoSession, ascending("invalidatingNp"));	36✔
96	collection("invalidations").createIndex(mongoSession, ascending("invalidatingPubkey"));	36✔
97	collection("invalidations").createIndex(mongoSession, ascending("invalidatedNp"));	36✔
98	collection("invalidations").createIndex(mongoSession, ascending("invalidatingPubkey", "invalidatedNp"));	48✔
99
100	collection("trustEdges").createIndex(mongoSession, ascending("fromAgent"));	36✔
101	collection("trustEdges").createIndex(mongoSession, ascending("fromPubkey"));	36✔
102	collection("trustEdges").createIndex(mongoSession, ascending("toAgent"));	36✔
103	collection("trustEdges").createIndex(mongoSession, ascending("toPubkey"));	36✔
104	collection("trustEdges").createIndex(mongoSession, ascending("source"));	36✔
105	collection("trustEdges").createIndex(mongoSession, ascending("fromAgent", "fromPubkey", "toAgent", "toPubkey", "source"), unique);	87✔
106	collection("trustEdges").createIndex(mongoSession, ascending("invalidated"));	36✔
107
108	collection("hashes").createIndex(mongoSession, ascending("hash"), unique);	39✔
109	collection("hashes").createIndex(mongoSession, ascending("value"), unique);	39✔
110	}	×
111	}	3✔
112
113	public static void initLoadingCollections(ClientSession mongoSession) {
114	collection("endorsements_loading").createIndex(mongoSession, ascending("agent"));	×
115	collection("endorsements_loading").createIndex(mongoSession, ascending("pubkey"));	×
116	collection("endorsements_loading").createIndex(mongoSession, ascending("endorsedNanopub"));	×
117	collection("endorsements_loading").createIndex(mongoSession, ascending("source"));	×
118	collection("endorsements_loading").createIndex(mongoSession, ascending("status"));	×
119	// zip: Hmm, not possible to set com.mongodb.client.model.WriteModel<T> here
120	// I'd currently recommend to have a package with the DB related stuff
121	// and therein a Custom Extension T of <Document>, where we could put that WriteModel<T>.
122
123	collection("agents_loading").createIndex(mongoSession, ascending("agent"), unique);	×
124	collection("agents_loading").createIndex(mongoSession, descending("accountCount"));	×
125	collection("agents_loading").createIndex(mongoSession, descending("avgPathCount"));	×
126	collection("agents_loading").createIndex(mongoSession, descending("totalRatio"));	×
127
128	collection("accounts_loading").createIndex(mongoSession, ascending("agent"));	×
129	collection("accounts_loading").createIndex(mongoSession, ascending("pubkey"));	×
130	collection("accounts_loading").createIndex(mongoSession, ascending("agent", "pubkey"), unique);	×
131	collection("accounts_loading").createIndex(mongoSession, ascending("type"));	×
132	collection("accounts_loading").createIndex(mongoSession, ascending("status"));	×
133	collection("accounts_loading").createIndex(mongoSession, descending("ratio"));	×
134	collection("accounts_loading").createIndex(mongoSession, descending("pathCount"));	×
135
136	collection("trustPaths_loading").createIndex(mongoSession, ascending("agent", "pubkey", "depth", "sorthash"), unique);	×
137	collection("trustPaths_loading").createIndex(mongoSession, ascending("depth"));	×
138	collection("trustPaths_loading").createIndex(mongoSession, descending("ratio"));	×
139	}	×
140
141	public static boolean isInitialized(ClientSession mongoSession) {
142	return getValue(mongoSession, Collection.SERVER_INFO.toString(), "setupId") != null;	24!
143	}
144
145	public static void rename(String oldCollectionName, String newCollectionName) {
146	// Designed as idempotent operation: calling multiple times has same effect as calling once
147	if (hasCollection(oldCollectionName)) {	×
148	if (hasCollection(newCollectionName)) {	×
149	collection(newCollectionName).drop();	×
150	}
151	collection(oldCollectionName).renameCollection(new MongoNamespace(REGISTRY_DB_NAME, newCollectionName));	×
152	}
153	}	×
154
155	public static boolean hasCollection(String collectionName) {
156	return mongoDB.listCollectionNames().into(new ArrayList<String>()).contains(collectionName);	×
157	}
158
159	public static void increaseStateCounter(ClientSession mongoSession) {
160	MongoCursor<Document> cursor = collection(Collection.SERVER_INFO.toString()).find(mongoSession, new Document("_id", "trustStateCounter")).cursor();	×
161	if (cursor.hasNext()) {	×
162	long counter = cursor.next().getLong("value");	×
163	collection(Collection.SERVER_INFO.toString()).updateOne(mongoSession, new Document("_id", "trustStateCounter"), new Document("$set", new Document("value", counter + 1)));	×
164	} else {	×
165	collection(Collection.SERVER_INFO.toString()).insertOne(mongoSession, new Document("_id", "trustStateCounter").append("value", 0l));	×
166	}
167	}	×
168
169	public static boolean has(ClientSession mongoSession, String collection, String elementName) {
170	return has(mongoSession, collection, new Document("_id", elementName));	×
171	}
172
173	private static final CountOptions hasCountOptions = new CountOptions().limit(1);	21✔
174
175	public static boolean has(ClientSession mongoSession, String collection, Bson find) {
176	return collection(collection).countDocuments(mongoSession, find, hasCountOptions) > 0;	×
177	}
178
179	public static MongoCursor<Document> get(ClientSession mongoSession, String collection, Bson find) {
180	return collection(collection).find(mongoSession, find).cursor();	×
181	}
182
183	public static Object getValue(ClientSession mongoSession, String collection, String elementName) {
184	Document d = collection(collection).find(mongoSession, new Document("_id", elementName)).first();	36✔
185	if (d == null) return null;	12!
186	return d.get("value");	×
187	}
188
189	public static boolean isSet(ClientSession mongoSession, String collection, String elementName) {
190	Document d = collection(collection).find(mongoSession, new Document("_id", elementName)).first();	×
191	if (d == null) return false;	×
192	return d.getBoolean("value");	×
193	}
194
195	public static Document getOne(ClientSession mongoSession, String collection, Bson find) {
196	return collection(collection).find(mongoSession, find).first();	×
197	}
198
199	public static Document getOne(ClientSession mongoSession, String collection, Bson find, Bson sort) {
200	return collection(collection).find(mongoSession, find).sort(sort).first();	×
201	}
202
203	public static Object getMaxValue(ClientSession mongoSession, String collection, String fieldName) {
204	Document doc = collection(collection).find(mongoSession).projection(new Document(fieldName, 1)).sort(new Document(fieldName, -1)).first();	×
205	if (doc == null) return null;	×
206	return doc.get(fieldName);	×
207	}
208
209	public static Object getMaxValue(ClientSession mongoSession, String collection, Bson find, String fieldName) {
210	Document doc = collection(collection).find(mongoSession, find).projection(new Document(fieldName, 1)).sort(new Document(fieldName, -1)).first();	×
211	if (doc == null) return null;	×
212	return doc.get(fieldName);	×
213	}
214
215	public static Document getMaxValueDocument(ClientSession mongoSession, String collection, Bson find, String fieldName) {
216	return collection(collection).find(mongoSession, find).sort(new Document(fieldName, -1)).first();	×
217	}
218
219	public static void set(ClientSession mongoSession, String collection, Document update) {
220	Bson find = new Document("_id", update.get("_id"));	×
221	MongoCursor<Document> cursor = collection(collection).find(mongoSession, find).cursor();	×
222	if (cursor.hasNext()) {	×
223	collection(collection).updateOne(mongoSession, find, new Document("$set", update));	×
224	}
225	}	×
226
227	public static void insert(ClientSession mongoSession, String collection, Document doc) {
228	collection(collection).insertOne(mongoSession, doc);	×
229	}	×
230
231	public static void setValue(ClientSession mongoSession, String collection, String elementId, Object value) {
232	collection(collection).updateOne(mongoSession, new Document("_id", elementId), new Document("$set", new Document("value", value)), new UpdateOptions().upsert(true));	×
233	}	×
234
235	public static void recordHash(ClientSession mongoSession, String value) {
236	try {
237	insert(mongoSession, "hashes", new Document("value", value).append("hash", Utils.getHash(value)));	×
238	} catch (MongoWriteException e) {	×
239	// Duplicate key error -- ignore it
240	if (e.getError().getCode() != 11000) throw e;	×
241	}	×
242	}	×
243
244	public static String unhash(String hash) {
245	try (var c = collection("hashes").find(new Document("hash", hash)).cursor()) {	×
246	if (c.hasNext()) return c.next().get("value").toString();	×
247	return null;	×
248	}	×
249	}
250
251	/**
252	* Insert nanopub to the DB.
253	*/
254	public static boolean loadNanopub(ClientSession mongoSession, Nanopub nanopub) {
255	return loadNanopub(mongoSession, nanopub, null);	×
256	}
257
258	public static boolean loadNanopub(ClientSession mongoSession, Nanopub nanopub, String pubkeyHash, String... types) {
259	if (nanopub.getTripleCount() > 1200) {	×
260	log.info("Nanopub has too many triples ({}): {}", nanopub.getTripleCount(), nanopub.getUri());	×
261	return false;	×
262	}
263	if (nanopub.getByteCount() > 1000000) {	×
264	log.info("Nanopub is too large ({}): {}", nanopub.getByteCount(), nanopub.getUri());	×
265	return false;	×
266	}
267	String pubkey = getPubkey(nanopub);	×
268	if (pubkey == null) {	×
269	log.info("Ignoring invalid nanopub: {}", nanopub.getUri());	×
270	return false;	×
271	}
272	String ph = Utils.getHash(pubkey);	×
273	if (pubkeyHash != null && !pubkeyHash.equals(ph)) {	×
274	log.info("Ignoring nanopub with non-matching pubkey: {}", nanopub.getUri());	×
275	return false;	×
276	}
277	recordHash(mongoSession, pubkey);	×
278
279	String ac = TrustyUriUtils.getArtifactCode(nanopub.getUri().stringValue());	×
280	if (ac == null) {	×
281	// I don't think this ever happens, but checking here to be sure
282	log.info("ERROR. Unexpected Trusty URI: {}", nanopub.getUri());	×
283	return false;	×
284	}
285	if (has(mongoSession, Collection.NANOPUBS.toString(), ac)) {	×
286	log.info("Already loaded: {}", nanopub.getUri());	×
287	} else {
288	Long counter = (Long) getMaxValue(mongoSession, Collection.NANOPUBS.toString(), "counter");	×
289	if (counter == null) counter = 0l;	×
290	String nanopubString;
291	byte[] jellyContent;
292	try {
293	nanopubString = NanopubUtils.writeToString(nanopub, RDFFormat.TRIG);	×
294	// Save the same thing in the Jelly format for faster loading
295	jellyContent = JellyUtils.writeNanopubForDB(nanopub);	×
296	} catch (IOException ex) {	×
297	throw new RuntimeException(ex);	×
298	}	×
299	collection(Collection.NANOPUBS.toString()).insertOne(mongoSession, new Document("_id", ac).append("fullId", nanopub.getUri().stringValue()).append("counter", counter + 1).append("pubkey", ph).append("content", nanopubString).append("jelly", new Binary(jellyContent)));	×
300
301	for (IRI invalidatedId : Utils.getInvalidatedNanopubIds(nanopub)) {	×
302	String invalidatedAc = TrustyUriUtils.getArtifactCode(invalidatedId.stringValue());	×
303	if (invalidatedAc == null) continue; // This should never happen; checking here just to be sure	×
304
305	// Add this nanopub also to all lists of invalidated nanopubs:
306	collection("invalidations").insertOne(mongoSession, new Document("invalidatingNp", ac).append("invalidatingPubkey", ph).append("invalidatedNp", invalidatedAc));	×
307	MongoCursor<Document> invalidatedEntries = collection("listEntries").find(mongoSession, new Document("np", invalidatedAc).append("pubkey", ph)).cursor();	×
308	while (invalidatedEntries.hasNext()) {	×
309	Document invalidatedEntry = invalidatedEntries.next();	×
310	addToList(mongoSession, nanopub, ph, invalidatedEntry.getString("type"));	×
311	}	×
312
313	collection("listEntries").updateMany(mongoSession, new Document("np", invalidatedAc).append("pubkey", ph), new Document("$set", new Document("invalidated", true)));	×
314	collection("trustEdges").updateMany(mongoSession, new Document("source", invalidatedAc), new Document("$set", new Document("invalidated", true)));	×
315	}	×
316	}
317
318	if (pubkeyHash != null) {	×
319	for (String type : types) {	×
320	// TODO Check if nanopub really has the type?
321	addToList(mongoSession, nanopub, pubkeyHash, Utils.getTypeHash(mongoSession, type));	×
322	if (type.equals("$")) {	×
323	for (IRI t : NanopubUtils.getTypes(nanopub)) {	×
324	addToList(mongoSession, nanopub, pubkeyHash, Utils.getTypeHash(mongoSession, t));	×
325	}	×
326	}
327	}
328	}
329
330	// Add the invalidating nanopubs also to the lists of this nanopub:
331	try (MongoCursor<Document> invalidations = collection("invalidations").find(mongoSession, new Document("invalidatedNp", ac).append("invalidatingPubkey", ph)).cursor()) {	×
332	if (invalidations.hasNext()) {	×
333	collection("listEntries").updateMany(mongoSession, new Document("np", ac).append("pubkey", ph), new Document("$set", new Document("invalidated", true)));	×
334	collection("trustEdges").updateMany(mongoSession, new Document("source", ac), new Document("$set", new Document("invalidated", true)));	×
335	}
336	while (invalidations.hasNext()) {	×
337	String iac = invalidations.next().getString("invalidatingNp");	×
338	try {
339	Document npDoc = collection(Collection.NANOPUBS.toString()).find(mongoSession, new Document("_id", iac)).projection(new Document("jelly", 1)).first();	×
340	Nanopub inp = JellyUtils.readFromDB(npDoc.get("jelly", Binary.class).getData());	×
341	for (IRI type : NanopubUtils.getTypes(inp)) {	×
342	addToList(mongoSession, inp, ph, Utils.getTypeHash(mongoSession, type));	×
343	}	×
344	} catch (RDF4JException \| MalformedNanopubException ex) {	×
345	ex.printStackTrace();	×
346	}	×
347	}	×
348
349	}
350
351	return true;	×
352	}
353
354	private static void addToList(ClientSession mongoSession, Nanopub nanopub, String pubkeyHash, String typeHash) {
355	String ac = TrustyUriUtils.getArtifactCode(nanopub.getUri().stringValue());	×
356	try {
357	insert(mongoSession, "lists", new Document("pubkey", pubkeyHash).append("type", typeHash));	×
358	} catch (MongoWriteException e) {	×
359	// Duplicate key error -- ignore it
360	if (e.getError().getCode() != 11000) throw e;	×
361	}	×
362
363	if (has(mongoSession, "listEntries", new Document("pubkey", pubkeyHash).append("type", typeHash).append("np", ac))) {	×
364	log.info("Already listed: {}", nanopub.getUri());	×
365	} else {
366
367	Document doc = getMaxValueDocument(mongoSession, "listEntries", new Document("pubkey", pubkeyHash).append("type", typeHash), "position");	×
368	long position;
369	String checksum;
370	if (doc == null) {	×
371	position = 0l;	×
372	checksum = NanopubUtils.updateXorChecksum(nanopub.getUri(), NanopubUtils.INIT_CHECKSUM);	×
373	} else {
374	position = doc.getLong("position") + 1;	×
375	checksum = NanopubUtils.updateXorChecksum(nanopub.getUri(), doc.getString("checksum"));	×
376	}
377	collection("listEntries").insertOne(mongoSession, new Document("pubkey", pubkeyHash).append("type", typeHash).append("position", position).append("np", ac).append("checksum", checksum).append("invalidated", false));	×
378	}
379	}	×
380
381	/**
382	* Returns the public key string of the Nanopub's signature, or null if not available or invalid.
383	*
384	* @param nanopub the nanopub to extract the public key from
385	* @return The public key string, or null if not available or invalid.
386	*/
387	public static String getPubkey(Nanopub nanopub) {
388	// TODO shouldn't this be moved to a utility class in nanopub-java? there is a similar method in NanopubElement class of nanodash
389	NanopubSignatureElement el;
390	try {
391	el = SignatureUtils.getSignatureElement(nanopub);	9✔
392	if (el != null && SignatureUtils.hasValidSignature(el) && el.getPublicKeyString() != null) {	24!
393	return el.getPublicKeyString();	9✔
394	}
395	} catch (MalformedCryptoElementException \| GeneralSecurityException ex) {	×
396	log.error("Error in checking the signature of the nanopub {}", nanopub.getUri());	×
397	}	3✔
398	return null;	6✔
399	}
400
401	public static String calculateTrustStateHash(ClientSession mongoSession) {
402	MongoCursor<Document> tp = collection("trustPaths_loading").find(mongoSession).sort(ascending("_id")).cursor();	×
403	// TODO Improve this so we don't create the full string just for calculating its hash:
404	String s = "";	×
405	while (tp.hasNext()) {	×
406	Document d = tp.next();	×
407	s += d.getString("_id") + " (" + d.getString("type") + ")\n";	×
408	}	×
409	return Utils.getHash(s);	×
410	}
411
412	}

knowledgepixels / nanopub-registry / 21060548866

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