/*
 *  Licensed 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.
 *
 *  See the NOTICE file distributed with this work for additional
 *  information regarding copyright ownership.
 */

package org.seaborne.delta.integration;

import static java.util.concurrent.TimeUnit.SECONDS;
import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.seaborne.delta.systemtest.Matrix.deltaServerURL1;
import static org.seaborne.delta.systemtest.Matrix.deltaServerURL2;

import java.util.function.Supplier;

import org.apache.jena.atlas.lib.Lib;
import org.apache.jena.graph.Node;
import org.apache.jena.query.QueryExecution;
import org.apache.jena.rdfconnection.RDFConnection;
import org.apache.jena.sparql.sse.SSE;
import org.junit.*;
import org.seaborne.delta.Id;
import org.seaborne.delta.client.DeltaLinkHTTP;
import org.seaborne.delta.client.DeltaLinkSwitchable;
import org.seaborne.delta.link.DeltaLink;
import org.seaborne.delta.systemtest.Matrix;

@FixMethodOrder(org.junit.runners.MethodSorters.NAME_ASCENDING)
public class TestDeltaZkFuseki {

    private static Node g1 = SSE.parseNode(":g1"); 
    private static Node g2 = SSE.parseNode("_:g2"); 
    private static Node s1 = SSE.parseNode(":s1"); 
    private static Node s2 = SSE.parseNode("_:s2"); 
    private static Node p1 = SSE.parseNode(":p1"); 
    private static Node p2 = SSE.parseNode(":p2"); 
    private static Node o1 = SSE.parseNode("<http://example/o1>");
    private static Node o2 = SSE.parseNode("123");
    
    @BeforeClass public static void beforeClass() { DeltaLinkSwitchable.silentSwitchOver = true; } 
    @AfterClass public static void afterClass()   { DeltaLinkSwitchable.silentSwitchOver = false; } 
    
    @Before public void before() { Matrix.setup(); }
    @After  public void after()  { Matrix.teardown(); }

    // Fuseki switch over tests.
    // Two DeltaServers, one Zookeeper.
    
    // Things only happen on DeltaConnection.sync.

    // Single fuseki server, two Delta servers 
    @Test
    public void t1_fusekiOne() {
        DeltaLink dLink1 = DeltaLinkHTTP.connect(deltaServerURL1);
        DeltaLink dLink2 = DeltaLinkHTTP.connect(deltaServerURL2);
        dLink1.ping();
        dLink2.ping();

        assertFalse(dLink1.existsByName("ABC"));
        Id logABC = dLink1.newDataSource("ABC", "http://example/abc");
        Lib.sleep(500);
        assertTrue(dLink1.existsByName("ABC"));
        assertTrue(dLink2.existsByName("ABC"));

        int fusekiPort = Matrix.choosePort();
        DeltaLinkSwitchable dLinkFuseki = (DeltaLinkSwitchable)Matrix.setupFuseki("ABC", "target/Zone", fusekiPort, deltaServerURL1, deltaServerURL2);
        String fusekiURL = "http://localhost:"+fusekiPort+"/ABC";
        RDFConnection conn = RDFConnection.connect(fusekiURL);

        conn.update("PREFIX : <http://delta/> INSERT DATA { :s :p :o }");
        assertCount("Count, before failover", conn, 1);

        if ( true )
            Matrix.deltaServer1.stop();
        else
            dLinkFuseki.switchover();

        assertCount("Count, after failover", conn, 1);
        conn.update("PREFIX : <http://delta/> INSERT DATA { :s :p :o2 }");
        assertCount(conn, 2);
    }

    @Test
    public void t2_fuseki2() {
        DeltaLink dLink1 = DeltaLinkHTTP.connect(deltaServerURL1);
        DeltaLink dLink2 = DeltaLinkHTTP.connect(deltaServerURL2);
        dLink1.ping();
        dLink2.ping();
        
        if ( dLink1.existsByName("ABC") ) {
            System.err.println("Exists");
        }
        
        Id log2 = dLink2.newDataSource("ABC", "http://example/abc");
        // Updaing the system is asynchronous
        await().atMost(5, SECONDS).until(()->dLink1.exists(log2));
        
        int fusekiPort1 = Matrix.choosePort();
        int fusekiPort2 = Matrix.choosePort();

        
        // Set primary delta server.
        // Retry needed?
        Matrix.setupFuseki("ABC", "target/Zone1", fusekiPort1, deltaServerURL1, deltaServerURL2);
        Matrix.setupFuseki("ABC", "target/Zone2", fusekiPort2, deltaServerURL2, deltaServerURL1);
        
        String fusekiURL1 = "http://localhost:"+fusekiPort1+"/ABC";
        String fusekiURL2 = "http://localhost:"+fusekiPort1+"/ABC";
        
//        System.err.printf("Fuseki servers %s , %s\n", fusekiURL1, fusekiURL2);
//        System.err.printf("Delta servers %s , %s\n", deltaServerURL1, deltaServerURL2);
        
        RDFConnection conn1 = RDFConnection.connect(fusekiURL1);
        RDFConnection conn2 = RDFConnection.connect(fusekiURL2);
        
        //boolean b = conn1.queryAsk("ASK{}");
        
        conn1.update("PREFIX : <http://delta/> INSERT DATA { :s :p :o }");
        assertCount(conn1, 1);
        assertCount(conn2, 1);
        
        Matrix.deltaServer1.stop();
        
        conn2.update("PREFIX : <http://delta/> INSERT DATA { :s :p :o2 }");

        assertCount(conn1, 2);
        assertCount(conn2, 2);

    }
    
    private static void assertCount(String msg, RDFConnection conn, long expected) {
        long x = count(conn);
        assertEquals(msg, expected, x);
    }
    private static void assertCount(RDFConnection conn, long expected) {
        long x = count(conn);
        assertEquals(expected, x);
    }
    
    private static long count(RDFConnection conn) {
        try ( QueryExecution qExec = conn.query("SELECT (count(*) AS ?count) { ?s ?p ?o }") ) {
            return qExec.execSelect().next().getLiteral("count").getLong();
        }
    }
    static <X> X retry(Supplier<X> action, int numRetries) {
        X rtn = null;
        int attempt = 0 ;
        while(rtn == null && attempt < numRetries ) {
            if ( attempt > 0 ) {
                System.err.println("Retry: "+attempt);
                // Short pause to let thread switching happen.
                Lib.sleep(100);
            }
            attempt++;
            rtn = action.get();
        }
        return rtn;
    }
        
}
