/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.rya.indexing.pcj.fluo.app.util;

import java.util.HashSet;
import java.util.Set;

import org.eclipse.rdf4j.query.algebra.Filter;
import org.eclipse.rdf4j.query.algebra.SingletonSet;
import org.eclipse.rdf4j.query.algebra.evaluation.function.Function;
import org.eclipse.rdf4j.query.algebra.evaluation.function.FunctionRegistry;
import org.eclipse.rdf4j.query.algebra.helpers.AbstractQueryModelVisitor;
import org.eclipse.rdf4j.query.parser.ParsedQuery;
import org.eclipse.rdf4j.query.parser.ParsedTupleQuery;
import org.eclipse.rdf4j.query.parser.sparql.SPARQLParser;
import org.eclipse.rdf4j.queryrender.sparql.SPARQLQueryRenderer;

/**
 * Class for creating a String representation a given Filter, and for
 * converting the String representation of the Filter back to the Filter.
 *
 */
public class FilterSerializer {

    private static final SPARQLQueryRenderer renderer = new SPARQLQueryRenderer();
    private static final SPARQLParser parser = new SPARQLParser();
    
    /**
     * Converts a {@link Filter} to a SPARQL query containing only the SPARQL representation
     * of the Filter along with a Select clause that return all variables.  The argument of the
     * Filter is replaced by a {@link SingletonSet} so that the body of the SPARQL query consists of only a
     * single Filter clause.  
     * @param filter - Filter to be serialized
     * @return - SPARQL String containing a single Filter clause that represents the serialized Filter
     * @throws FilterParseException
     */
    public static String serialize(Filter filter) throws FilterParseException {
        Filter clone = filter.clone();
        clone.setArg(new SingletonSet());
        try {
            return removeAngularBracketsFromNonUriFunctions(renderer.render(new ParsedTupleQuery(clone)));
        } catch (Exception e) {
            throw new FilterParseException("Unable to parse Filter.", e);
        }
    }
    
    /**
     * Converts a SPARQL query consisting of a single Filter clause back to a Filter.
     * @param sparql - SPARQL query representing a Filter
     * @return - parsed Filter included in the SPARQL query
     * @throws FilterParseException
     */
    public static Filter deserialize(String sparql) throws FilterParseException {
        
        try {
            ParsedQuery pq = parser.parseQuery(sparql, null);
            FilterVisitor visitor = new FilterVisitor();
            pq.getTupleExpr().visit(visitor);
            Set<Filter> filters = visitor.getFilters();
            
            if(filters.size() != 1) {
                throw new FilterParseException("Filter String must contain only one Filter.");
            }
            
            return filters.iterator().next();
            
        } catch (Exception e) {
            throw new FilterParseException("Unable to parse Filter.", e);
        }
    }
    
    public static class FilterVisitor extends AbstractQueryModelVisitor<RuntimeException> {

        private Set<Filter> filters;
        
        public FilterVisitor() {
            filters = new HashSet<>();
        }

        public Set<Filter> getFilters() {
            return filters;
        }

        public void meet(Filter node) {
            filters.add(node);
        }
        
    }
    
    /**
     * There are a number of Functions in the FunctionRegistry whose getURI() method does not return a valid URI (NOW()
     * is one such method). The SPARQLQueryRender adds angular brackets to the result returned by
     * {@link Function#getURI()} by default, which leads to a MalformedQueryException when the SPARQLParser attempts to
     * parse the SPARQL created by the renderer. Therefore, a call to serialize and then deserialize for a Filter
     * containing a Function that returns an invalid URI will generate an exception. This method removes the angular
     * brackets from the result returned by {@link Function#getURI()} if it is not a valid URI so that the parser will
     * parse it.
     * 
     * @param query - query generated by query renderer
     * @return - String with angular brackets removed from around all invalid Function URI
     */
    private static String removeAngularBracketsFromNonUriFunctions(String query) {
        FunctionRegistry registry = FunctionRegistry.getInstance();
        for(String key: registry.getKeys()) {
            if (key.indexOf(':') < 0) {
                query = query.replace("<"+key.trim()+">", key);
            }
        }
        return query;
    }
    
    public static class FilterParseException extends Exception {

        private static final long serialVersionUID = 1L;
        
        /**
         * Constructs an instance of {@link FilterParseException}.
         *
         * @param message - Explains why this exception is being thrown.
         */
        public FilterParseException(final String message) {
            super(message);
        }

        /**
         * Constructs an instance of {@link FilterParseException}.
         *
         * @param message - Explains why this exception is being thrown.
         * @param cause - The exception that caused this one to be thrown.
         */
        public FilterParseException(final String message, final Throwable t) {
            super(message, t);
        }
    }
    
}