Max De Marzi

One of the often overlooked features in Neo4j is the “TransactionEventHandler” capabilities… better known in the database world as “Triggers“. When a transaction occurs, we can analyze that event and decide to take some action. To accomplish this, we’ll write a “Kernel Extension” ( a little different from the Unmanaged Extensions we’ve seen on this blog ) to tie in our trigger.

Imagine you are Special Agent Avery Ryan working at the FBI’s team of cyber crime investigators working to solve Internet-related murders, cyber-theft, hacking, sex offenses, blackmail, and any other crime deemed to be cyber-related within the FBI’s jurisdiction.

Your colleages ex-black-hat-hackers Raven Ramirez and Brody Nelson have built a social network of known suspects and their acquaintances. But former U.S. Marine Senior Special Agent Elijah Mundo can’t be bothered to learn Cypher, all he wants is an email when a new suspect is identified or a new connection to a known suspect is learned. So you assign hacking savant Daniel Krumitz to add this feature to your system.

Daniel looks at the TransactionEventHandler documentation and learns there are 3 hooks, beforeCommit, afterCommit and afterRollback that can be used:

 
beforeCommit(TransactionData data)
Invoked when a transaction is about to be committed.

afterCommit(TransactionData data, T state)
Invoked after the transaction has been committed successfully.

afterRollback(TransactionData data, T state)
Invoked after the transaction has been rolled back if committing the transaction failed for some reason.

Since we want to send an email after a new suspect is identified or connected, the afterCommit hook will be used. However we don’t want to put our logic here. Instead we’ll create an ExecutorService and a Runnable to handle it.

 
public class MyTransactionEventHandler implements TransactionEventHandler {

    public static GraphDatabaseService db;
    private static ExecutorService ex;

    public MyTransactionEventHandler(GraphDatabaseService graphDatabaseService, ExecutorService executor) {
        db = graphDatabaseService;
        ex = executor;
    }

    @Override
    public Object beforeCommit(TransactionData transactionData) throws Exception {
        return null;
    }

    @Override
    public void afterCommit(TransactionData transactionData, Object o) {
        ex.submit(new SuspectRunnable(transactionData, db));
    }

    @Override
    public void afterRollback(TransactionData transactionData, Object o) {

    }
}

Our SuspectRunnable is getting the database passed in as well as the transactionData which contains the nodes and relationships that were created, labels that were assigned, properties that were changed, etc. Let’s use these by starting a new transaction and checking if the newly created nodes have a Suspect label:

 
    @Override
    public void run() {
        try (Transaction tx = db.beginTx()) {
            Set<Node> suspects = new HashSet<>();
            for (Node node : td.createdNodes()) {
                if (node.hasLabel(Labels.Suspect)) {
                    suspects.add(node);
                    System.out.println("A new Suspect has been created!");
                }
            }

We are just printing a message to the “neo4j/data/log/console.log” file for our demo, but in a real application it would send an email, or go to a message queue or whatever you wanted it to do. So we capture any new Suspects, but we also need to capture if any nodes get assigned a Suspect label:

 
            for (LabelEntry labelEntry : td.assignedLabels()) {
                if (labelEntry.label().equals(Labels.Suspect) && !suspects.contains(labelEntry.node())) {
                    System.out.println("A new Suspect has been identified!");
                    suspects.add(labelEntry.node());
                }
            }

Now we need to identify any direct or indirect relationships to Suspects that may have been created. We’ll take the newly created Relationships and check if one of the nodes of the relationship is a Suspect. But we won’t stop there. We’ll use those nodes to traverse one more level out and see if we have a new indirect relationship to a Suspect:

 
            for (Relationship relationship : td.createdRelationships()) {
                if (relationship.isType(RelationshipTypes.KNOWS)) {
                    for (Node user : relationship.getNodes()) {
                        if (user.hasLabel(Labels.Suspect)) {
                            System.out.println("A new direct relationship to a Suspect has been created!");
                        }

                        for (Relationship knows : user.getRelationships(Direction.BOTH, 
                                                                        RelationshipTypes.KNOWS)) {
                            Node otherUser = knows.getOtherNode(user);
                            if (otherUser.hasLabel(Labels.Suspect) &&
                               !otherUser.equals(relationship.getOtherNode(user))) {
                                System.out.println("A new indirect relationship to a Suspect has been created!");
                            }
                        }
                    }
                }
            }

That’s it for our Runnable, now we need to hook this in to Neo4j by creating our KernelExtension that registers our transaction event handler. In our start method we will create our executor and handler. In our shutdown method (called when Neo4j shuts down) we will shutdown the executor and unregister the handler.

 
    @Override
    public Lifecycle newKernelExtension(final Dependencies dependencies) throws Throwable {
        return new LifecycleAdapter() {

            private MyTransactionEventHandler handler;
            private ExecutorService executor;

            @Override
            public void start() throws Throwable {
                executor = Executors.newFixedThreadPool(2);
                handler = new MyTransactionEventHandler(dependencies.getGraphDatabaseService(), executor);
                dependencies.getGraphDatabaseService().registerTransactionEventHandler(handler);
            }

            @Override
            public void shutdown() throws Throwable {
                executor.shutdown();
                dependencies.getGraphDatabaseService().unregisterTransactionEventHandler(handler);
            }
        };
    }

In order for Neo4j to pickup this class, we’ll need to create a “org.neo4j.kernel.extension.KernelExtensionFactory” file in the “resources/META-INF/services” directory with the entry “com.maxdemarzi.RegisterTransactionEventHandlerExtensionFactory”.

The full source code is available on github.

To deploy this to our server, we will build it:

 
mvn clean package

Then copy target/triggers-1.0.jar to the neo4j/plugins directory of our Neo4j server. We’ll start our server and we will tail the neo4j/data/log/console.log file. Let’s try a few queries:

 
CREATE (max:User {name:"Max"}) RETURN max;

Nothing happened because the node we created doesn’t trigger anything.

 
CREATE (al:Suspect {name:"Al Capone"}) RETURN al;

Results in “A new Suspect has been created!” appearing in our log file.

 
MATCH (max:User),(al:Suspect)
WHERE max.name = "Max" AND al.name = "Al Capone"
CREATE (max)-[r:KNOWS]->(al)
RETURN r;

Gives us “A new direct relationship to a Suspect has been created!”.

 
CREATE (monica:User {name:"Monica"}) RETURN monica;

Nothing happens.

 
MATCH (max:User),(monica:User)
WHERE max.name = "Max" AND monica.name = "Monica"
CREATE (max)-[r:KNOWS]->(monica)
RETURN r;

Gives us “A new indirect relationship to a Suspect has been created!” since Monica is now connected to Max who was already connected to our suspect Al Capone…and finally:

 
MATCH (monica:User)
WHERE monica.name = "Monica"
SET monica :Suspect
RETURN monica;

Shows us “A new Suspect has been identified!” since Monica went from being a User to a usual Suspect.

You can see adding Triggers to Neo4j gives you a whole new set of capabilities. The mechanism we used to add them to our server, Kernel Extensions are pretty powerful. They allow you to do anything the folks running Neo4j embedded can do, but in a neat little package that is easy to deploy.

Warning: It makes sense to me to use an ExecutorService to perform the action of the trigger if we are communicating with an external service or doing any kind of heavy operation and if it’s safe to lose the action of the trigger if anything were to happen.

However if you are performing any action on the data, like encrypting a sensitive field, using triggers for a custom constraint, or anything that MUST happen otherwise you risk data corruption, then you want to do it inline to be safe.