/*
 * 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.jena.tdb2.xloader;

import static org.apache.jena.tdb2.xloader.BulkLoaderX.async;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.jena.atlas.io.IO;
import org.apache.jena.atlas.iterator.IteratorSlotted;
import org.apache.jena.atlas.lib.*;
import org.apache.jena.atlas.logging.FmtLog;
import org.apache.jena.dboe.base.file.BinaryDataFile;
import org.apache.jena.dboe.base.file.BufferChannel;
import org.apache.jena.dboe.base.file.FileFactory;
import org.apache.jena.dboe.base.file.FileSet;
import org.apache.jena.dboe.base.record.Record;
import org.apache.jena.dboe.base.record.RecordFactory;
import org.apache.jena.dboe.sys.Names;
import org.apache.jena.dboe.trans.bplustree.BPlusTree;
import org.apache.jena.dboe.trans.bplustree.BPlusTreeParams;
import org.apache.jena.dboe.trans.bplustree.rewriter.BPlusTreeRewriter;
import org.apache.jena.graph.Node;
import org.apache.jena.graph.Triple;
import org.apache.jena.irix.IRIProvider;
import org.apache.jena.irix.SystemIRIx;
import org.apache.jena.riot.RDFParser;
import org.apache.jena.riot.system.StreamRDF;
import org.apache.jena.riot.thrift.RiotThriftException;
import org.apache.jena.riot.thrift.TRDF;
import org.apache.jena.riot.thrift.ThriftConvert;
import org.apache.jena.riot.thrift.wire.RDF_Term;
import org.apache.jena.sparql.core.DatasetGraph;
import org.apache.jena.sparql.core.Quad;
import org.apache.jena.tdb2.DatabaseMgr;
import org.apache.jena.tdb2.TDBException;
import org.apache.jena.tdb2.lib.NodeLib;
import org.apache.jena.system.progress.ProgressIterator;
import org.apache.jena.system.progress.ProgressMonitorOutput;
import org.apache.jena.system.progress.ProgressStreamRDF;
import org.apache.jena.tdb2.store.DatasetGraphTDB;
import org.apache.jena.tdb2.store.Hash;
import org.apache.jena.tdb2.store.NodeId;
import org.apache.jena.tdb2.store.NodeIdFactory;
import org.apache.jena.tdb2.store.nodetable.NodeTable;
import org.apache.jena.tdb2.store.nodetable.NodeTableTRDF;
import org.apache.jena.tdb2.store.nodetable.TReadAppendFileTransport;
import org.apache.jena.tdb2.sys.SystemTDB;
import org.apache.jena.tdb2.sys.TDBInternal;
import org.apache.thrift.TException;
import org.apache.thrift.TSerializer;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TTransport;
import org.slf4j.Logger;

public class ProcNodeTableBuilderX {

    /** Pair<triples, indexed nodes> */
    // [BULK} Output, not return.
    public static Pair<Long, Long> exec(Logger LOG, String DB, XLoaderFiles loaderFiles, List<String> datafiles, String sortArgs) {
        //Threads - 1 parser, 1 builder, 2 sort.
        // Excludes counting inlines.

        IRIProvider provider = SystemIRIx.getProvider();
        //SystemIRIx.setProvider(new IRIProviderAny());

        DatasetGraph dsg = DatabaseMgr.connectDatasetGraph(DB);
        DatasetGraphTDB dsgtdb = TDBInternal.getDatasetGraphTDB(dsg);
        NodeTable nt = dsgtdb.getTripleTable().getNodeTupleTable().getNodeTable();
        NodeTableTRDF nodeTable = (NodeTableTRDF)nt.baseNodeTable();

        OutputStream toSortOutputStream;
        InputStream fromSortInputStream;

        // ** Step 2: The sort
        Process proc2;
        try {
            //LOG.info("Step : external sort");
            String[] sortCmd =
                { "sort",
                    "--temporary-directory="+loaderFiles.TMPDIR, "--buffer-size=50%",
                    "--parallel=2", "--unique",
                    // Don't compress. There will be enough space because later we work on indexes.
                    //"--compress-program=/usr/bin/gzip",
                    "--key=1,1"
                };
            if ( sortArgs != null ) {}

            proc2 = new ProcessBuilder(sortCmd).start();
            // To process.
            toSortOutputStream = proc2.getOutputStream(); // Needs buffering
            // From process
            fromSortInputStream = proc2.getInputStream(); // Needs buffering

//            // Debug sort process.
//            InputStream fromSortErrortStream = proc2.getErrorStream();
//            IOUtils.copy(fromSortErrortStream, System.err);

        } catch (Exception ex) {
            throw new RuntimeException(ex);
        }

        // ** Step 1 : write intermediate file (hash, thrift bytes).
        AtomicLong countParseTicks = new AtomicLong(-1);
        AtomicLong countIndexedNodes = new AtomicLong(-1);

        long tickPoint = BulkLoaderX.DataTick;
        int superTick = BulkLoaderX.DataSuperTick;

        //LOG.info("Step : parse & send to external sort");
        Runnable task1 = ()->{
            ProgressMonitorOutput monitor = ProgressMonitorOutput.create(LOG, "Nodes", tickPoint, superTick);
            OutputStream output = IO.ensureBuffered(toSortOutputStream);
            // Counting.
            StreamRDF worker = new NodeHashTmpStream(output);
            ProgressStreamRDF stream = new ProgressStreamRDF(worker, monitor);
            monitor.start();
            String label = monitor.getLabel();
            datafiles.forEach( datafile->{
                String basename = FileOps.basename(datafile);
                monitor.setLabel(basename);
                stream.start();
                RDFParser.source(datafile).parse(stream);
                stream.finish();
            });
            monitor.finish();
            monitor.setLabel(label);

            IO.flush(output);
            IO.close(output);

            long x = monitor.getTime();

//            long x = timer.endTimer();
            long count = monitor.getTicks();
            countParseTicks.set(count);

            double xSec = x/1000.0;
            double rate = count/xSec;
            FmtLog.info(LOG, "== Parse: %s seconds : %,d triples/quads %,.0f TPS", Timer.timeStr(x), count, rate);
        };

        // [BULK] XXX AsyncParser.asyncParse(files, output)
        Thread thread1 = async(task1, "AsyncParser");

        //LOG.info("Step : read external sort, build node-data file, build index");
        Runnable task2 = ()->{
            Timer timer = new Timer();
            // Don't start timer until sort send something
            InputStream input = IO.ensureBuffered(fromSortInputStream);

            FileSet fileSet = new FileSet(dsgtdb.getLocation(), Names.nodeTableBaseName);
            BufferChannel blkState = FileFactory.createBufferChannel(fileSet, Names.extBptState);
            long idxTickPoint = BulkLoaderX.DataTick;
            int idxSuperTick = BulkLoaderX.DataSuperTick;
            ProgressMonitorOutput monitor = ProgressMonitorOutput.create(LOG, "Index", idxTickPoint, idxSuperTick);

            // SLOW

            // Library of tools!
            dsg.executeWrite(()->{
                BinaryDataFile objectFile = nodeTable.getData();
                Iterator<Record> rIter = records(LOG, input, objectFile);
                rIter = new ProgressIterator<>(rIter, monitor);
                // Record of (hash, nodeId)
                BPlusTree bpt1 = (BPlusTree)(nodeTable.getIndex());
                BPlusTreeParams bptParams = bpt1.getParams();
                RecordFactory factory = new RecordFactory(SystemTDB.LenNodeHash,  NodeId.SIZE);
                // Wait until something has been received from the sort step
                rIter.hasNext();
                monitor.start();
                // .. then start the timer. It is closed after the transaction finishes.
                timer.startTimer();

                BPlusTree bpt2 = BPlusTreeRewriter.packIntoBPlusTree(rIter,
                                                                     bptParams, factory, blkState,
                                                                     bpt1.getNodeManager().getBlockMgr(),
                                                                     bpt1.getRecordsMgr().getBlockMgr());
                bpt2.sync();
                bpt1.sync();
                objectFile.sync();
                monitor.finish();
            });
            blkState.sync();
            IO.close(input);
            long x = timer.endTimer();
            long count = monitor.getTicks();
            countIndexedNodes.set(count);
            String rateStr = BulkLoaderX.rateStr(count, x);
            FmtLog.info(LOG, "==  Index: %s seconds : %,d indexed RDF terms : %s PerSecond", Timer.timeStr(x), count, rateStr);

        };
        Thread thread2 = async(task2, "AsyncBuild");

        try {
            int x = proc2.waitFor();
            if ( x != 0 )
                FmtLog.error(LOG, "Sort RC = %d", x);
            else
                LOG.info("Sort finished");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        BulkLoaderX.waitFor(thread1);
        BulkLoaderX.waitFor(thread2);

        return Pair.create(countParseTicks.get(), countIndexedNodes.get());
    }

    // [BULK] XXX Make this a separate class.
    static Iterator<Record> records(Logger LOG, InputStream input, BinaryDataFile objectFile) {
        RecordFactory factory = new RecordFactory(SystemTDB.LenNodeHash,  NodeId.SIZE);
        byte[] bHash = new byte[SystemTDB.LenNodeHash];
        byte[] bbNodeId = new byte[NodeId.SIZE];
        RDF_Term term = new RDF_Term();
        // Better?
        TProtocol protocol;
        try {
            TTransport transport = new TReadAppendFileTransport(objectFile);
            if ( ! transport.isOpen() )
                transport.open();
            protocol = TRDF.protocol(transport);
        }
        catch (Exception ex) {
            throw new TDBException("NodeTableTRDF", ex);
        }

        return new IteratorSlotted<Record>() {
            long count = 0;
            @Override
            protected Record moveToNext() {
                return calc();
            }

            @Override
            protected boolean hasMore() {
                return true;
            }

            // One line of file encoded data to record.
            private Record calc() {
                count++;
                try {
                  // read hash.
                  for ( int i = 0 ; i < 16 ; i++ ) {
                      int x = hexRead(input);
                      if ( x < 0 )
                          return null;
                      bHash[i] = (byte)(x&0xFF);
                  }
                  char ch0 = (char)input.read(); // space.
                  byte[] key = bHash;

                  ByteArrayOutputStream bout = new ByteArrayOutputStream();
                  // Read de-hexer
                  for(;;) {
                      int v = hexRead(input);
                      if ( v < 0 )
                          break;
                      bout.write(v);
                  }
                  byte[] thrift = bout.toByteArray();
                  ThriftConvert.termFromBytes(term, thrift);
                  // write to nodes.dat -> NodeId
                  long x = objectFile.length();
                  NodeId nodeId = NodeIdFactory.createPtr(x);
                  objectFile.write(thrift);
                  Bytes.setLong(nodeId.getPtrLocation(), bbNodeId);
                  Record r = factory.create(key, bbNodeId);
                  return r;
              } catch (Exception ex) {
                  ex.printStackTrace();
                  return null;
              }
            }
        };
    }

    public static int hexRead(InputStream input) throws IOException {
        int c1 = input.read();
        if ( c1 < 0 )
            return -1;
        if ( c1 == '\n' || c1 == ' ')
            return -1;
        int c2 = input.read();
        int b1 = Hex.hexByteToInt(c1);
        int b2 = Hex.hexByteToInt(c2);
        int b = (b1<<4)|b2;
        return b;
    }


    public static void hexWrite(OutputStream output, int bits8) throws IOException {
        int x1 = (bits8>>4) & 0xF;
        int x2 = bits8 & 0xF;
        byte ch1 = Bytes.hexDigitsUC[x1];
        byte ch2 = Bytes.hexDigitsUC[x2];
        output.write(ch1);
        output.write(ch2);
    }


    static byte[] hashNode(Node node) {
        NodeLib.setHash(hash, node);
        return hash.getBytes();
    }

    private static Hash hash = new Hash(SystemTDB.LenNodeHash);

    //Cache needed to reduce duplicates
    /** Write the intermediate sort file */
    static class NodeHashTmpStream implements StreamRDF {

        private final OutputStream outputData;
        private CacheSet<Node> cache = CacheFactory.createCacheSet(100_000);

        NodeHashTmpStream(OutputStream outputFile) {
            this.outputData = outputFile;
        }

        @Override
        public void start() {}

        @Override
        public void triple(Triple triple) {
            node(triple.getSubject());
            node(triple.getPredicate());
            node(triple.getObject());
        }

        @Override
        public void quad(Quad quad) {
            node(quad.getGraph());
            node(quad.getSubject());
            node(quad.getPredicate());
            node(quad.getObject());
        }

        static TSerializer serializer;
        static {
            try {
                serializer = new TSerializer(new TCompactProtocol.Factory());
            }
            catch (TException e) {
                throw new RiotThriftException(e);
            }
        }

        private void node(Node node) {
            NodeId nid = NodeId.inline(node);
            if ( nid != null )
                return ;
            if ( cache.contains(node) )
                return;
            cache.add(node);
            // -- Hash of node
            //SystemTDB.LenNodeHash, SystemTDB.SizeOfNodeId
            NodeLib.setHash(hash, node);
            try {
                byte k[] = hash.getBytes();
                RDF_Term term = ThriftConvert.convert(node, false);
                byte[] tBytes = serializer.serialize(term);
                write(outputData, k);
                outputData.write(' ');
                write(outputData, tBytes);
                outputData.write('\n');

            } catch (TException e) {
                e.printStackTrace();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        private static void write(OutputStream outputData, byte[] bytes) throws IOException {
            for ( byte bits8 : bytes )
                hexWrite(outputData, bits8);
        }

        @Override
        public void base(String base) {}

        @Override
        public void prefix(String prefix, String iri) {}

        @Override
        public void finish() {
            IO.flush(outputData);
        }

    }
}
