/*
 * Copyright (c) "Neo4j"
 * Neo4j Sweden AB [http://neo4j.com]
 *
 * This file is part of Neo4j.
 *
 * 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.
 */
package apoc.util;

import static apoc.ApocConfig.apocConfig;
import static apoc.export.util.LimitedSizeInputStream.toLimitedIStream;
import static apoc.util.DateFormatUtil.getOrCreate;
import static java.net.HttpURLConnection.HTTP_NOT_MODIFIED;
import static org.eclipse.jetty.util.URIUtil.encodePath;
import static org.neo4j.configuration.GraphDatabaseSettings.SYSTEM_DATABASE_NAME;

import apoc.Pools;
import apoc.convert.ConvertUtils;
import apoc.export.util.CountingInputStream;
import apoc.export.util.ExportConfig;
import apoc.result.VirtualNode;
import apoc.result.VirtualRelationship;
import apoc.util.collection.Iterators;
import java.io.BufferedWriter;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStreamWriter;
import java.io.StringWriter;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.InvocationTargetException;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
import java.net.URLConnection;
import java.net.URLEncoder;
import java.time.format.DateTimeFormatter;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Base64;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.ListIterator;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Scanner;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import javax.lang.model.SourceVersion;
import org.apache.commons.compress.archivers.ArchiveEntry;
import org.apache.commons.compress.archivers.ArchiveInputStream;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.neo4j.configuration.Config;
import org.neo4j.configuration.GraphDatabaseSettings;
import org.neo4j.configuration.connectors.BoltConnector;
import org.neo4j.dbms.api.DatabaseManagementService;
import org.neo4j.graphdb.Entity;
import org.neo4j.graphdb.ExecutionPlanDescription;
import org.neo4j.graphdb.GraphDatabaseService;
import org.neo4j.graphdb.Label;
import org.neo4j.graphdb.Node;
import org.neo4j.graphdb.NotInTransactionException;
import org.neo4j.graphdb.QueryExecutionType;
import org.neo4j.graphdb.Relationship;
import org.neo4j.graphdb.RelationshipType;
import org.neo4j.graphdb.ResourceIterator;
import org.neo4j.graphdb.Result;
import org.neo4j.graphdb.ResultTransformer;
import org.neo4j.graphdb.Transaction;
import org.neo4j.graphdb.TransactionTerminatedException;
import org.neo4j.graphdb.schema.ConstraintType;
import org.neo4j.graphdb.schema.IndexDefinition;
import org.neo4j.graphdb.security.URLAccessChecker;
import org.neo4j.graphdb.security.URLAccessValidationError;
import org.neo4j.internal.kernel.api.procs.ProcedureCallContext;
import org.neo4j.internal.schema.ConstraintDescriptor;
import org.neo4j.kernel.api.QueryLanguage;
import org.neo4j.kernel.impl.coreapi.InternalTransaction;
import org.neo4j.kernel.impl.util.ValueUtils;
import org.neo4j.kernel.internal.GraphDatabaseAPI;
import org.neo4j.logging.Log;
import org.neo4j.logging.NullLog;
import org.neo4j.procedure.Mode;
import org.neo4j.procedure.TerminationGuard;
import org.neo4j.values.AnyValue;
import org.neo4j.values.storable.CoordinateReferenceSystem;
import org.neo4j.values.storable.PointValue;
import org.neo4j.values.storable.TextValue;
import org.neo4j.values.storable.Value;
import org.neo4j.values.storable.Values;

/**
 * @author mh
 * @since 24.04.16
 */
public class Util {

    public static final Label[] NO_LABELS = new Label[0];
    public static final String NODE_COUNT = "MATCH (n) RETURN count(*) as result";
    public static final String REL_COUNT = "MATCH ()-->() RETURN count(*) as result";
    public static final String ERROR_BYTES_OR_STRING = "Only byte[] or url String allowed";

    public static final int REDIRECT_LIMIT = 10;

    public static String INVALID_QUERY_MODE_ERROR = "This procedure allows for READ-only, non-schema based queries. "
            + "It is therefore not possible to perform writes or query the database with commands such as SHOW CONSTRAINTS/INDEXES.";

    public static String labelString(Node n) {
        return joinLabels(n.getLabels(), ":");
    }

    public static String joinLabels(Iterable<Label> labels, String s) {
        return StreamSupport.stream(labels.spliterator(), false)
                .map(Label::name)
                .collect(Collectors.joining(s));
    }

    public static List<String> labelStrings(Node n) {
        return StreamSupport.stream(n.getLabels().spliterator(), false)
                .map(Label::name)
                .sorted()
                .collect(Collectors.toList());
    }

    public static Label[] labels(Object labelNames) {
        if (labelNames == null) return NO_LABELS;
        if (labelNames instanceof List) {
            Set names = new LinkedHashSet((List) labelNames); // Removing duplicates
            Label[] labels = new Label[names.size()];
            int i = 0;
            for (Object l : names) {
                if (l == null) continue;
                labels[i++] = Label.label(l.toString());
            }
            if (i <= labels.length) return Arrays.copyOf(labels, i);
            return labels;
        }
        return new Label[] {Label.label(labelNames.toString())};
    }

    public static Label[] getLabelsArray(Node node) {
        List<Label> labels = getLabels(node);
        return labels.toArray(new Label[0]);
    }

    public static List<Label> getLabels(Node node) {
        return getLabelsAsStream(node).collect(Collectors.toList());
    }

    private static Stream<Label> getLabelsAsStream(Node node) {
        return StreamSupport.stream(node.getLabels().spliterator(), false);
    }

    public static Stream<Object> stream(Object values) {
        return ConvertUtils.convertToList(values).stream();
    }

    public static Stream<Node> nodeStream(InternalTransaction tx, Object ids) {
        return stream(ids).map(id -> node(tx, id));
    }

    public static List<Node> nodeList(InternalTransaction tx, Object ids) {
        if (ids == null) return List.of();
        return stream(ids).map(id -> node(tx, id)).toList();
    }

    public static Node node(InternalTransaction tx, Object id) {
        if (id instanceof Node node) return rebind(tx, node);
        if (id instanceof Number nodeId)
            return tx.getNodeByElementId(tx.elementIdMapper().nodeElementId(nodeId.longValue()));
        if (id instanceof String elementId) return tx.getNodeByElementId(elementId);
        throw new RuntimeException("Can't convert " + id.getClass() + " to a Node");
    }

    public static Stream<Relationship> relsStream(InternalTransaction tx, Object ids) {
        return stream(ids).map(id -> relationship(tx, id));
    }

    public static Relationship relationship(InternalTransaction tx, Object id) {
        if (id instanceof Relationship rel) return rebind(tx, rel);
        if (id instanceof Number relId)
            return tx.getRelationshipByElementId(tx.elementIdMapper().relationshipElementId(relId.longValue()));
        if (id instanceof String elementId) return tx.getRelationshipByElementId(elementId);
        throw new RuntimeException("Can't convert " + id.getClass() + " to a Relationship");
    }

    public static <T> T retryInTx(
            Log log,
            GraphDatabaseService db,
            Function<Transaction, T> function,
            long retry,
            long maxRetries,
            Consumer<Long> callbackForRetry) {
        try (Transaction tx = db.beginTx()) {
            T result = function.apply(tx);
            tx.commit();
            return result;
        } catch (Exception e) {
            if (retry >= maxRetries) throw e;
            if (log != null) {
                log.warn("Retrying operation %d of %d", retry, maxRetries);
            }
            callbackForRetry.accept(retry);
            Util.sleep(100);
            return retryInTx(log, db, function, retry + 1, maxRetries, callbackForRetry);
        }
    }

    public static <T> Future<T> inTxFuture(
            Log log,
            ExecutorService pool,
            GraphDatabaseService db,
            Function<Transaction, T> function,
            long maxRetries,
            Consumer<Long> callbackForRetry,
            Consumer<Void> callbackAction) {
        try {
            return pool.submit(() -> {
                try {
                    return retryInTx(log, db, function, 0, maxRetries, callbackForRetry);
                } finally {
                    callbackAction.accept(null);
                }
            });
        } catch (Exception e) {
            throw new RuntimeException("Error executing in separate transaction", e);
        }
    }

    public static <T> Future<T> inTxFuture(
            ExecutorService pool, GraphDatabaseService db, Function<Transaction, T> function) {
        return inTxFuture(null, pool, db, function, 0, _ignored -> {}, _ignored -> {});
    }

    public static <T> T inTx(GraphDatabaseService db, Pools pools, Function<Transaction, T> function) {
        try {
            return inTxFuture(pools.getDefaultExecutorService(), db, function).get();
        } catch (RuntimeException e) {
            throw e;
        } catch (Exception e) {
            // TODO: consider dropping the exception to logs
            throw new RuntimeException("Error executing in separate transaction: " + e.getMessage(), e);
        }
    }

    public static <T> T inThread(Pools pools, Callable<T> callable) {
        try {
            return inFuture(pools, callable).get();
        } catch (Exception e) {
            throw new RuntimeException("Error executing in separate thread: " + e.getMessage(), e);
        }
    }

    public static <T> Future<T> inFuture(Pools pools, Callable<T> callable) {
        return pools.getDefaultExecutorService().submit(callable);
    }

    public static Double toDouble(Object value) {
        if (value == null) return null;
        if (value instanceof Number) return ((Number) value).doubleValue();
        try {
            return Double.parseDouble(value.toString());
        } catch (NumberFormatException e) {
            return null;
        }
    }

    public static Map<String, Object> subMap(Map<String, ?> params, String prefix) {
        Map<String, Object> config = new HashMap<>(10);
        int len = prefix.length() + (prefix.isEmpty() || prefix.endsWith(".") ? 0 : 1);
        for (Map.Entry<String, ?> entry : params.entrySet()) {
            String key = entry.getKey();
            if (key.startsWith(prefix)) {
                config.put(key.substring(len), entry.getValue());
            }
        }
        return config;
    }

    public static Long toLong(Object value) {
        if (value == null) return null;
        if (value instanceof Number) return ((Number) value).longValue();
        try {
            String s = value.toString();
            if (s.contains(".")) return (long) Double.parseDouble(s);
            return Long.parseLong(s);
        } catch (NumberFormatException e) {
            return null;
        }
    }

    public static Integer toInteger(Object value) {
        if (value == null) return null;
        if (value instanceof Number) return ((Number) value).intValue();
        try {
            String s = value.toString();
            if (s.contains(".")) return (int) Double.parseDouble(s);
            return Integer.parseInt(value.toString());
        } catch (NumberFormatException e) {
            return null;
        }
    }

    public static URLConnection openUrlConnection(URL src, Map<String, Object> headers) throws IOException {
        URLConnection con = src.openConnection();
        con.setRequestProperty("User-Agent", "APOC Procedures for Neo4j");
        if (con instanceof HttpURLConnection) {
            HttpURLConnection http = (HttpURLConnection) con;
            http.setInstanceFollowRedirects(false);
            if (headers != null) {
                Object method = headers.get("method");
                if (method != null) {
                    http.setRequestMethod(method.toString());
                    http.setChunkedStreamingMode(1024 * 1024);
                }
                headers.forEach((k, v) -> con.setRequestProperty(k, v == null ? "" : v.toString()));
            }
        }

        con.setConnectTimeout(apocConfig().getInt("apoc.http.timeout.connect", 10_000));
        con.setReadTimeout(apocConfig().getInt("apoc.http.timeout.read", 60_000));
        return con;
    }

    public static boolean isRedirect(HttpURLConnection con) throws IOException {
        int responseCode = con.getResponseCode();
        boolean isRedirectCode =
                responseCode >= 300 && responseCode <= 307 && responseCode != 306 && responseCode != HTTP_NOT_MODIFIED;
        if (isRedirectCode) {
            URL location = new URL(con.getHeaderField("Location"));
            String oldProtocol = con.getURL().getProtocol();
            String protocol = location.getProtocol();
            if (!protocol.equals(oldProtocol)
                    && !protocol.startsWith(oldProtocol)) { // we allow http -> https redirect and similar
                throw new RuntimeException("The redirect URI has a different protocol: " + location.toString());
            }
        }
        return isRedirectCode;
    }

    private static void writePayload(URLConnection con, String payload) throws IOException {
        if (payload == null) return;
        con.setDoOutput(true);
        BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(con.getOutputStream(), "UTF-8"));
        writer.write(payload);
        writer.close();
    }

    private static String handleRedirect(URLConnection con, String url) throws IOException {
        if (!(con instanceof HttpURLConnection)) return url;
        if (!isRedirect(((HttpURLConnection) con))) return url;
        return con.getHeaderField("Location");
    }

    public static CountingInputStream openInputStream(
            Object input,
            Map<String, Object> headers,
            String payload,
            String compressionAlgo,
            URLAccessChecker urlAccessChecker)
            throws IOException, URISyntaxException, URLAccessValidationError {
        if (input instanceof String) {
            String urlAddress = (String) input;
            final ArchiveType archiveType = ArchiveType.from(urlAddress);
            if (archiveType.isArchive()) {
                return getStreamCompressedFile(urlAddress, headers, payload, archiveType, urlAccessChecker);
            }

            StreamConnection sc = getStreamConnection(urlAddress, headers, payload, urlAccessChecker);
            return sc.toCountingInputStream(compressionAlgo);
        } else if (input instanceof byte[]) {
            return FileUtils.getInputStreamFromBinary((byte[]) input, compressionAlgo);
        } else {
            throw new RuntimeException(ERROR_BYTES_OR_STRING);
        }
    }

    private static CountingInputStream getStreamCompressedFile(
            String urlAddress,
            Map<String, Object> headers,
            String payload,
            ArchiveType archiveType,
            URLAccessChecker urlAccessChecker)
            throws IOException, URISyntaxException, URLAccessValidationError {
        StreamConnection sc;
        InputStream stream;
        String[] tokens = urlAddress.split("!");
        urlAddress = tokens[0];
        String zipFileName;
        if (tokens.length == 2) {
            zipFileName = tokens[1];
            sc = getStreamConnection(urlAddress, headers, payload, urlAccessChecker);
            stream = getFileStreamIntoCompressedFile(sc.getInputStream(), zipFileName, archiveType);
            stream = toLimitedIStream(stream, sc.getLength());
        } else throw new IllegalArgumentException("filename can't be null or empty");

        return new CountingInputStream(stream, sc.getLength());
    }

    public static StreamConnection getStreamConnection(
            String urlAddress, Map<String, Object> headers, String payload, URLAccessChecker urlAccessChecker)
            throws IOException, URISyntaxException, URLAccessValidationError {
        return FileUtils.getStreamConnection(
                FileUtils.from(urlAddress), urlAddress, headers, payload, urlAccessChecker);
    }

    private static InputStream getFileStreamIntoCompressedFile(InputStream is, String fileName, ArchiveType archiveType)
            throws IOException {
        try (ArchiveInputStream archive = archiveType.getInputStream(is)) {
            ArchiveEntry archiveEntry;

            while ((archiveEntry = archive.getNextEntry()) != null) {
                if (!archiveEntry.isDirectory() && archiveEntry.getName().equals(fileName)) {
                    return new ByteArrayInputStream(IOUtils.toByteArray(archive));
                }
            }
        }

        return null;
    }

    public static StreamConnection readHttpInputStream(
            String urlAddress,
            Map<String, Object> headers,
            String payload,
            int redirectLimit,
            URLAccessChecker urlAccessChecker)
            throws IOException {
        URL url = apocConfig().checkAllowedUrlAndPinToIP(urlAddress, urlAccessChecker);
        URLConnection con = openUrlConnection(url, headers);
        writePayload(con, payload);
        String newUrl = handleRedirect(con, urlAddress);
        if (newUrl != null && !urlAddress.equals(newUrl)) {
            con.getInputStream().close();
            if (redirectLimit == 0) {
                throw new IOException("Redirect limit exceeded");
            }
            return readHttpInputStream(newUrl, headers, payload, --redirectLimit, urlAccessChecker);
        }

        return new StreamConnection.UrlStreamConnection(con);
    }

    public static boolean toBoolean(Object value) {
        if ((value == null
                || value instanceof Number && (((Number) value).longValue()) == 0L
                || value instanceof String
                        && (value.equals("")
                                || ((String) value).equalsIgnoreCase("false")
                                || ((String) value).equalsIgnoreCase("no")
                                || ((String) value).equalsIgnoreCase("0"))
                || value instanceof Boolean && value.equals(false))) {
            return false;
        }
        return true;
    }

    public static String toString(Object string) {
        return string == null ? null : string.toString();
    }

    public static String encodeUrlComponent(String value) {
        try {
            return URLEncoder.encode(value, "UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException("Unsupported character set utf-8");
        }
    }

    public static String toJson(Object value) {
        try {
            return JsonUtil.OBJECT_MAPPER.writeValueAsString(value);
        } catch (IOException e) {
            throw new RuntimeException("Can't convert " + value + " to JSON");
        }
    }

    public static <T> T fromJson(String value, Class<T> type) {
        try {
            return JsonUtil.OBJECT_MAPPER.readValue(value, type);
        } catch (IOException e) {
            throw new RuntimeException("Can't convert " + value + " from JSON");
        }
    }

    public static Stream<List<Object>> partitionSubList(List<Object> data, int partitions) {
        return partitionSubList(data, partitions, null);
    }

    public static Stream<List<Object>> partitionSubList(List<Object> data, int partitions, List<Object> tombstone) {
        if (partitions == 0) partitions = 1;
        List<Object> list = new ArrayList<>(data);
        int total = list.size();
        int batchSize = Math.max((int) Math.ceil((double) total / partitions), 1);
        Stream<List<Object>> stream = IntStream.range(0, partitions)
                .parallel()
                .mapToObj((part) ->
                        list.subList(Math.min(part * batchSize, total), Math.min((part + 1) * batchSize, total)))
                .filter(partition -> !partition.isEmpty());
        return tombstone == null ? stream : Stream.concat(stream, Stream.of(tombstone));
    }

    public static Long runNumericQuery(Transaction tx, String query, Map<String, Object> params) {
        if (params == null) params = Collections.emptyMap();
        try (ResourceIterator<Long> it = tx.execute(query, params).<Long>columnAs("result")) {
            return it.next();
        }
    }

    public static long nodeCount(Transaction tx) {
        return runNumericQuery(tx, NODE_COUNT, null);
    }

    public static long relCount(Transaction tx) {
        return runNumericQuery(tx, REL_COUNT, null);
    }

    public static String readResourceFile(String name) {
        InputStream is = Util.class.getClassLoader().getResourceAsStream(name);
        return new Scanner(is).useDelimiter("\\Z").next();
    }

    @SuppressWarnings("unchecked")
    public static Map<String, Object> readMap(String value) {
        try {
            return JsonUtil.OBJECT_MAPPER.readValue(value, Map.class);
        } catch (IOException e) {
            throw new RuntimeException("Couldn't read as JSON " + value);
        }
    }

    public static <T> List<T> take(Iterator<T> iterator, int batchsize) {
        List<T> result = new ArrayList<>(batchsize);
        while (iterator.hasNext() && batchsize-- > 0) {
            result.add(iterator.next());
        }
        return result;
    }

    public static Map<String, Object> merge(Map<String, Object> first, Map<String, Object> second) {
        if (second == null || second.isEmpty()) return first == null ? Collections.EMPTY_MAP : first;
        if (first == null || first.isEmpty()) return second == null ? Collections.EMPTY_MAP : second;
        Map<String, Object> combined = new HashMap<>(first);
        combined.putAll(second);
        return combined;
    }

    public static <T> Map<String, T> map(T... values) {
        Map<String, T> map = new LinkedHashMap<>();
        for (int i = 0; i < values.length; i += 2) {
            if (values[i] == null) continue;
            map.put(values[i].toString(), values[i + 1]);
        }
        return map;
    }

    public static <T> Map<String, T> map(List<T> pairs) {
        Map<String, T> res = new LinkedHashMap<>(pairs.size() / 2);
        Iterator<T> it = pairs.iterator();
        while (it.hasNext()) {
            Object key = it.next();
            T value = it.next();
            if (key != null) res.put(key.toString(), value);
        }
        return res;
    }

    public static <T, R> List<R> map(Stream<T> stream, Function<T, R> mapper) {
        return stream.map(mapper).collect(Collectors.toList());
    }

    public static <T, R> List<R> map(Collection<T> collection, Function<T, R> mapper) {
        return map(collection.stream(), mapper);
    }

    public static Map<String, Object> mapFromLists(List<String> keys, List<Object> values) {
        if (keys == null || values == null || keys.size() != values.size())
            throw new RuntimeException("keys and values lists have to be not null and of same size");
        if (keys.isEmpty()) return Collections.emptyMap();
        if (keys.size() == 1) return Collections.singletonMap(keys.get(0), values.get(0));
        ListIterator<Object> it = values.listIterator();
        Map<String, Object> res = new LinkedHashMap<>(keys.size());
        for (String key : keys) {
            res.put(key, it.next());
        }
        return res;
    }

    public static Map<String, Object> mapFromPairs(List<List<Object>> pairs) {
        if (pairs.isEmpty()) return Collections.emptyMap();
        Map<String, Object> map = new LinkedHashMap<>(pairs.size());
        for (List<Object> pair : pairs) {
            if (pair.isEmpty()) continue;
            Object key = pair.get(0);
            if (key == null) continue;
            Object value = pair.size() >= 2 ? pair.get(1) : null;
            map.put(key.toString(), value);
        }
        return map;
    }

    public static String cleanUrl(String url) {
        try {
            URL source = new URL(url);
            String file = source.getFile();
            if (source.getRef() != null) file += "#" + source.getRef();
            return new URL(source.getProtocol(), source.getHost(), source.getPort(), file).toString();
        } catch (MalformedURLException mfu) {
            return String.format("invalid URL (%s)", url);
        }
    }

    public static <T> T getFuture(Future<T> f, Map<String, Long> errorMessages, AtomicInteger errors, T errorValue) {
        try {
            T t = f.get();
            return t;
        } catch (Exception e) {
            errors.incrementAndGet();
            errorMessages.compute(e.getMessage(), (s, i) -> i == null ? 1 : i + 1);
            return errorValue;
        }
    }

    public static <T> T getFutureOrCancel(
            Future<T> f, Map<String, Long> errorMessages, AtomicInteger errors, T errorValue) {
        try {
            if (f.isDone()) return f.get();
            else {
                f.cancel(false);
                errors.incrementAndGet();
            }
        } catch (Exception e) {
            errors.incrementAndGet();
            errorMessages.compute(e.getMessage(), (s, i) -> i == null ? 1 : i + 1);
        }
        return errorValue;
    }

    public static void logErrors(String message, Map<String, Long> errors, Log log) {
        if (!errors.isEmpty()) {
            log.warn(message);
            errors.forEach((k, v) -> log.warn("%d times: %s", v, k));
        }
    }

    public static void sleep(int millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            /* ignore */
        }
    }

    public static String quote(String var) {
        return SourceVersion.isIdentifier(var) && !var.contains("$") ? var : '`' + var + '`';
    }

    private static final String ESCAPED_UNICODE_BACKTICK = "\\u0060";

    private static final Pattern PATTERN_ESCAPED_4DIGIT_UNICODE = Pattern.compile("\\\\u+(\\p{XDigit}{4})");
    private static final Pattern PATTERN_LABEL_AND_TYPE_QUOTATION = Pattern.compile("(?<!`)`(?:`{2})*(?!`)");

    private static final List<String[]> SUPPORTED_ESCAPE_CHARS = Collections.unmodifiableList(Arrays.asList(
            new String[] {"\\b", "\b"},
            new String[] {"\\f", "\f"},
            new String[] {"\\n", "\n"},
            new String[] {"\\r", "\r"},
            new String[] {"\\t", "\t"},
            new String[] {"\\`", "``"}));

    /**
     * Sanitizes the given input to be used as a valid schema name
     *
     * @param value The value to sanitize
     * @return A value that is safe to be used in string concatenation, an empty optional indicates a value that cannot be safely quoted
     */
    public static String sanitize(String value) {
        return sanitize(value, false);
    }

    /**
     * Sanitizes the given input to be used as a valid schema name
     *
     * @param value The value to sanitize
     * @param addQuotes If quotation should be added
     * @return A value that is safe to be used in string concatenation, an empty optional indicates a value that cannot be safely quoted
     */
    public static String sanitize(String value, boolean addQuotes) {
        if (value == null || value.isEmpty()) {
            return value;
        }

        // Replace escaped chars
        for (String[] pair : SUPPORTED_ESCAPE_CHARS) {
            value = value.replace(pair[0], pair[1]);
        }
        value = value.replace(ESCAPED_UNICODE_BACKTICK, "`");

        // Replace escaped octal hex
        // Excluding the support for 6 digit literals, as this contradicts the overall example in CIP-59r
        Matcher matcher = PATTERN_ESCAPED_4DIGIT_UNICODE.matcher(value);
        StringBuffer sb = new StringBuffer();
        while (matcher.find()) {
            String replacement = Character.toString((char) Integer.parseInt(matcher.group(1), 16));
            matcher.appendReplacement(sb, Matcher.quoteReplacement(replacement));
        }

        matcher.appendTail(sb);
        value = sb.toString();

        value = value.replace("\\u", "\\u005C\\u0075");

        matcher = PATTERN_LABEL_AND_TYPE_QUOTATION.matcher(value);
        value = matcher.replaceAll("`$0");
        value = value.replace("\\\\", "\\");

        if (!addQuotes) {
            return value;
        }

        return String.format(Locale.ENGLISH, "`%s`", value);
    }

    public static String param(String var) {
        return var.charAt(0) == '$' ? var : '$' + quote(var);
    }

    public static String withMapping(Stream<String> columns, Function<String, String> withMapping) {
        String with = columns.map(withMapping).collect(Collectors.joining(","));
        return with.isEmpty() ? with : " WITH " + with + " ";
    }

    public static boolean isWriteableInstance(GraphDatabaseAPI db, Transaction systemTx) {
        var socketAddress = db.getDependencyResolver()
                .resolveDependency(Config.class)
                .get(BoltConnector.advertised_address)
                .toString();
        String query = "SHOW DATABASE $databaseName WHERE address = $socketAddress";
        final Map<String, Object> params = Map.of("databaseName", db.databaseName(), "socketAddress", socketAddress);
        if (systemTx == null) {
            GraphDatabaseService systemDb = db.getDependencyResolver()
                    .resolveDependency(DatabaseManagementService.class)
                    .database(SYSTEM_DATABASE_NAME);
            return systemDb.executeTransactionally(query, params, Util::getSingleWriter);
        } else {
            final Result result = systemTx.execute(query, params);
            return getSingleWriter(result);
        }
    }

    public static boolean isWriteableInstance(GraphDatabaseAPI db) {
        return isWriteableInstance(db, null);
    }

    private static Boolean getSingleWriter(Result result) {
        return Iterators.single(result.columnAs("writer"));
    }

    /**
     * Given a context related to the procedure invocation this method checks if the transaction is terminated in some way
     *
     * @param db
     * @return
     *
     */
    public static boolean transactionIsTerminated(TerminationGuard db) {
        try {
            db.check();
            return false;
        } catch (TransactionTerminatedException | NotInTransactionException tge) {
            return true;
        }
    }

    public static void waitForFutures(List<Future> futures) {
        for (Future future : futures) {
            try {
                if (future != null) future.get();
            } catch (InterruptedException | ExecutionException e) {
                // ignore
            }
        }
    }

    public static void removeFinished(List<Future> futures) {
        if (futures.size() > 25) {
            futures.removeIf(Future::isDone);
        }
    }

    public static void close(AutoCloseable closeable, Consumer<Exception> onError) {
        try {
            if (closeable != null) closeable.close();
