package com.astrolabsoftware.FinkBrowser.Januser;

import com.Lomikel.Utils.Pair;
import com.Lomikel.Utils.LomikelException;
import com.Lomikel.Januser.GremlinRecipies;
import com.Lomikel.Januser.ModifyingGremlinClient;
import com.astrolabsoftware.FinkBrowser.HBaser.FinkHBaseClient;

// Tinker Pop
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.outV;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.V;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.fold;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.has;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.not;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.unfold;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.out;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.repeat;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.values;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.count;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.addV;
import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.label;
import static org.apache.tinkerpop.gremlin.process.traversal.P.within;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.Direction;

// Java
import java.lang.Math;
import java.util.Arrays;
import java.util.Iterator;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.HashMap;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.Set;
import java.util.HashSet;
import java.util.Date;
import java.text.SimpleDateFormat;

// Log4J
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.LogManager;

/** Fink-specific graph mutation and lifecycle recipes.
  *
  * <p>The package documentation defines the graph model, label-mirror
  * invariant, and transaction ownership shared by this class and the Groovy
  * traversal recipes.</p>
  * @opt attributes
  * @opt operations
  * @opt types
  * @opt visibility
  * @author <a href="mailto:Julius.Hrivnac@cern.ch">J.Hrivnac</a> */
// Client-dependent preconditions are checked by the operation that needs them.
public class FinkGremlinRecipies extends GremlinRecipies {
    
  /** Create and attach to {@link GraphTraversalSource}.
    * @param g The attached {@link GraphTraversalSource}. */
  public FinkGremlinRecipies(GraphTraversalSource g) {
    super(g);
    }
    
  /** Create and attach to {@link ModifyingGremlinClient}.
    * @param client The attached  {@link ModifyingGremlinClient}. */
  public FinkGremlinRecipies(ModifyingGremlinClient client) {
    super(client);
    }
    
  /*
   * The deprecated processOCol/fillOCol orchestration was intentionally
   * removed from the compiled API. It mixed source discovery (HBase/Fink
   * Portal), classification, and correlation generation in one operation.
   * Supported callers select object IDs explicitly, call classifySources or
   * classifySource, and invoke generateCorrelations as a separate lifecycle
   * step.
   */
    
  /** Classify <em>object</em> .
    * @param classifiers The {@link Classifier}s to be used.
    * @param oids        The {@link Set} of {@code objectId}s of <em>object</em> to be added.
    * @param hbaseUrl    The url of HBase with alerts as {@code ip:port:table:schema}.
    * @throws LomikelException If anything fails. */
  public void classifySources(Classifier[] classifiers,
                              Set<String>  oids,
                              String       hbaseUrl) throws LomikelException {
    int size = oids.size();
    int n = 0;
    long dt;
    double freq;
    long startTime = System.currentTimeMillis();
    // loop over all objects
    for (String oid : oids) {
      log.info(oid + " (" + n + " of " + size + "):");
      for (Classifier classifier : classifiers) {
        try {
          classifySource(classifier, oid, hbaseUrl);
          }
        catch (LomikelException e) {
          log.error("Cannot get classification for " + oid, e);
          }
        }
      n++;
      dt = (System.currentTimeMillis() - startTime) / 1000;
      freq = (double)n / (double)dt;
      log.info("\t\twith " + String.format("%.2f", freq) + " Hz");
      }
    }
    
   /** Classify <em>object</em>.
    * @param classifier The {@link Classifier} to be used.
    * @param objectId   The {@code objectId} of <em>object</em> to be added.
    * @param hbaseUrl   The url of HBase with alerts as {@code ip:port:table:schema}.
    * @throws LomikelException If anything fails. */
  public void classifySource(Classifier classifier,
                             String     objectId,
                             String     hbaseUrl) throws LomikelException {  
    fhclient(hbaseUrl);
    classifySource(classifier, objectId);
    }
   
  /** Classify <em>object</em>.
    * @param classifier The {@link Classifier} to be used.
    * @param objectId   The {@code objectId} of <em>object</em> to be added.
    * @throws LomikelException If anything fails. */
  public void classifySource(Classifier classifier,
                             String     objectId) throws LomikelException {
    if (!supportsTransactions()) {
      throw new UnsupportedOperationException("Atomic classification requires rollback-capable transactions");
      }
    if (_classificationTransaction.get()) {
      throw new IllegalStateException("Nested classification transactions are not supported");
      }
    _classificationTransaction.set(true);
    try {
      Iterator<Vertex> objects = g().V().has("lbl", "object").
                                        has("objectId", objectId).
                                        limit(1);
      if (objects.hasNext()) {
        Vertex object = objects.next();
        // Cleanup intentionally follows the indexed lbl property. Registration
        // additionally requires the native label, so only cache an endpoint
        // that standalone registration would select as well.
        if ("object".equals(object.label())) {
          _classificationObjects.get().put(objectId, object);
          }
        List<Edge> edges = g().V(object).inE("deepcontains").
                              where(outV().
                                has("lbl",        "OCol").
                                has("survey",     classifier.survey()).
                                has("classifier", classifier.name()  ).
                                has("flavor",     classifier.flavor())).
                              toList();
        for (Edge edge : edges) {
          edge.remove();
          }
        }
      classifier.classify(this, objectId);
      commit();
      }
    catch (LomikelException | RuntimeException e) {
      try {
        rollback();
        }
      catch (RuntimeException rollbackFailure) {
        e.addSuppressed(rollbackFailure);
        }
      throw e;
      }
    finally {
      _classificationObjects.remove();
      _classificationTransaction.remove();
      }
    }
       
  /** Register  <em>object</em> in <em>OCol</em>.
    * @param classifier The {@link Classifier} to be used.
    * @param cls        The type (class) of <em>OCol</em> {@link Vertex}.
    *                   It will be created if not yet exists.
    * @param objectId   The objectId of the new <em>Source</em> {@link Vertex}.
    *                   It will be created if not yet exists.
    * @param weight     The weight of the connection.
    *                   Usualy the number of <em>Alerts</em> of this type. 
    * @param instancesS The <em>jd</em> of related <em>Alerts</em> as strings separated by comma.
    *                   Potential square brackets are removed.
    *                   May be {@code null} or empty.
    * @param weightsS   The corresponding per-alert weights, separated by comma. */
  public void registerOCol(Classifier classifier,
                           String     cls,
                           String     objectId,
                           double     weight,
                           String     instancesS,
                           String     weightsS) {   
    List<String> instances = new ArrayList<>();
    List<Double> weights   = new ArrayList<>();
    if (instancesS != null && !instancesS.trim().equals("")) {
      for (String instance : instancesS.replaceAll("\\[", "").replaceAll("]", "").split(",")) {
        instances.add(instance);
        }
      }
    if (weightsS != null && !weightsS.trim().equals("")) {
      for (String weighs : weightsS.replaceAll("\\[", "").replaceAll("]", "").split(",")) {
        weights.add(Double.valueOf(weighs.trim()));
        }
      }
    registerOCol(classifier, cls, objectId, weight, instances, weights);
    }
    
  /** Register <em>object</em> in <em>OCol</em>. Replace possible existing registration.
    * @param classifier The {@link Classifier} to be used.
    * @param cls        The type (class) of <em>OCol</em> {@link Vertex}.
    *                   It will be created if not yet exists.
    * @param objectId   The objectId of the new <em>Source</em> {@link Vertex}.
    *                   It will be created if not yet exists.
    * @param weight     The total weight of the connection.
    *                   Usualy the number of <em>Alerts</em> of this type. 
    * @param instances  The <em>jd</em> of related <em>Alerts</em>.
    * @param weights    The weights of related <em>Alerts</em>. */
  public void registerOCol(Classifier   classifier,
                           String       cls,
                           String       objectId,
                           double       weight,
                           List<String> instances,
                           List<Double> weights) { 
    validateWeight(weight, "aggregate weight");
    if (instances == null || weights == null || instances.size() != weights.size()) {
      throw new IllegalArgumentException("Instances and weights must have equal lengths");
      }
    for (Double instanceWeight : weights) {
      if (instanceWeight == null) {
        throw new IllegalArgumentException("Per-instance weight must not be null");
        }
      validateWeight(instanceWeight, "per-instance weight");
      }
    Map<String, Object> attributes = new HashMap<>();
    attributes.put("weight",    weight);
    attributes.put("instances", instances.toString().replaceFirst("\\[", "").replaceAll("]", ""));
    attributes.put("weights",   weights.toString().replaceFirst("\\[", "").replaceAll("]", ""));
    registerOCol(classifier, cls, objectId, attributes, true);
    }
    
  /** Register <em>object</em> in <em>OCol</em>.
    * @param classifier The {@link Classifier} to be used.
    * @param cls        The type (class) of <em>OCol</em> {@link Vertex}.
    *                   It will be created if not yet exists.
    * @param objectId   The objectId of the new <em>Source</em> {@link Vertex}.
    *                   It will be created if not yet exists.
    * @param attributes The additional {@link Edge} attributes.    
    * @param replace    Whether to replace existing resistration. */
  public synchronized void registerOCol(Classifier          classifier,
                                        String              cls,
                                        String              objectId,
                                        Map<String, ?> attributes,
                                        boolean             replace) {
    Object weightValue = attributes == null ? null : attributes.get("weight");
    double weight;
    if (weightValue instanceof Number) {
      weight = ((Number)weightValue).doubleValue();
      }
    else if (weightValue instanceof String) {
      try {
        weight = Double.parseDouble(((String)weightValue).trim());
        }
      catch (NumberFormatException e) {
        throw new IllegalArgumentException("Registration weight must be numeric", e);
        }
      }
    else {
      throw new IllegalArgumentException("Registration weight must be numeric");
      }
    validateWeight(weight, "aggregate weight");
    Map<String, Object> validatedAttributes = new HashMap<>();
    validatedAttributes.putAll(attributes);
    validatedAttributes.remove("lbl");
    validatedAttributes.put("weight", weight);
    boolean rollbackOnFailure = !_classificationTransaction.get() && supportsTransactions();
    try {
    //log.info("\tregistering " + objectId + " as " + classifier + " / " + cls + " with attributes " + attributes + ", replace = " + replace);
    log.info("\tregistering " + objectId + " as " + classifier + " / " + cls + " with weight = " + weight + ", replace = " + replace);
    String importDate = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date());
    Vertex ocol = g().V().has("lbl",        "OCol"             ).
                          has("survey",     classifier.survey()).
                          has("classifier", classifier.name()  ).
                          has("flavor",     classifier.flavor()).
                          has("cls",        cls                ).
                          filter(label().is("OCol")).
                          fold().
                          coalesce(unfold(), 
                                  addV("OCol").
                                  property("lbl",        "OCol").
                                  property("survey",     classifier.survey()).
                                  property("classifier", classifier.name()  ).
                                  property("flavor",     classifier.flavor()).
                                  property("cls",        cls                )).
                         next();
    boolean classificationTransaction = _classificationTransaction.get();
    Vertex s = classificationTransaction ? _classificationObjects.get().get(objectId) : null;
    boolean cacheObject = false;
    if (s == null) {
      s = g().V().has("lbl",      "object").
                  has("objectId", objectId).
                  filter(label().is("object")).
                  fold().
                  coalesce(unfold(),
                           addV("object").
                           property("lbl",      "object").
                           property("objectId", objectId)).
                  next();
      cacheObject = classificationTransaction;
      }
    s.property("importDate", importDate);
    if (cacheObject) {
      _classificationObjects.get().put(objectId, s);
      }
    if (replace) {
      replaceRegistrationEdge(ocol, s, validatedAttributes);
      }
    else {
      createRegistrationEdge(ocol, s, validatedAttributes);
      }
    if (!_classificationTransaction.get()) {
      commit();
      }
      }
    catch (RuntimeException e) {
      if (rollbackOnFailure) {
        try {
          rollback();
          }
        catch (RuntimeException rollbackFailure) {
          e.addSuppressed(rollbackFailure);
          }
        }
      throw e;
      }
    }

  /** Replace all parallel registrations with one fully prepared edge. */
  private void replaceRegistrationEdge(Vertex              ocol,
                                       Vertex              object,
                                       Map<String, Object> attributes) {
    List<Edge> existingEdges = getEdge(ocol, object, "deepcontains");
    if (supportsTransactions()) {
      for (Edge edge : existingEdges) {
        edge.remove();
        }
      createRegistrationEdge(ocol, object, attributes);
      return;
      }

    Edge replacement = createRegistrationEdge(ocol, object, attributes);
    try {
      for (Edge edge : existingEdges) {
        edge.remove();
        }
      }
    catch (RuntimeException failure) {
      try {
        replacement.remove();
        }
      catch (RuntimeException cleanupFailure) {
        failure.addSuppressed(cleanupFailure);
        }
      throw failure;
      }
    }

  /** Create a complete registration edge or remove its partial state on failure. */
  private Edge createRegistrationEdge(Vertex              ocol,
                                      Vertex              object,
                                      Map<String, Object> attributes) {
    Edge edge = ocol.addEdge("deepcontains", object);
    try {
      edge.property("lbl", "deepcontains");
      for (Map.Entry<String, Object> attribute : attributes.entrySet()) {
        edge.property(attribute.getKey(), attribute.getValue());
        }
      return edge;
      }
    catch (RuntimeException failure) {
      try {
        edge.remove();
        }
      catch (RuntimeException cleanupFailure) {
        failure.addSuppressed(cleanupFailure);
        }
      throw failure;
      }
    }

  /** Validate scientific registration weights before graph mutation. */
  private static void validateWeight(double weight, String description) {
    if (!Double.isFinite(weight) || weight < 0.0) {
      throw new IllegalArgumentException("Invalid " + description + ": " + weight);
      }
    }
   
  /** Clean tree under <em>OCol</em>.
    * Drop alerts. Alerts are dropped even if they have other
    * {@link Edge}s.
    * @param classifier The {@link Classifier} to be used.
    *                   All its <em>flavors</em> are handled.
    * @param cls        The type (class) of <em>OCol</em>.
    * @throws LomikelException If anything goes wrong. */
  public void cleanOCol(Classifier classifier,
                        String     cls) throws LomikelException {
    log.info("Cleaning " + cls + " OCol");
    g().V().has("lbl",        "OCol").
            has("survey",     classifier.survey()).
            has("classifier", classifier.name()  ).
            has("flavor",     classifier.flavor()).
            has("cls",        cls                ).
            out("deepcontains").
            out().
            drop().
            iterate();
    commit();
    }
        
  /** Generate <em>overlaps</em> Edges between <em>OCol</em>.
    * Possibly between two {@link Classifier}s.
    * @param classifiers The {@link Classifier}s to be used. */
  public void generateCorrelations(Classifier... classifiers) {
    if (!supportsTransactions()) {
      throw new UnsupportedOperationException("Atomic correlation regeneration requires rollback-capable transactions");
      }
    try {
      generateCorrelationsInTransaction(classifiers);
      }
    catch (RuntimeException e) {
      try {
        rollback();
        }
      catch (RuntimeException rollbackFailure) {
        e.addSuppressed(rollbackFailure);
        }
      throw e;
      }
    }

  /** Generate correlations inside one caller-owned transaction. */
  private void generateCorrelationsInTransaction(Classifier... classifiers) {
    log.info("Generating correlations for OCol of " + Arrays.asList(classifiers));
    // Clean all correlations 
    g().E().has("lbl", "overlaps").
            drop().
            iterate();
    Set<String> classifierScopes = new HashSet<>();
    List<Vertex> scopedOCols = new ArrayList<>();
    List<Vertex> malformedScopedOCols = new ArrayList<>();
    for (Classifier classifier : classifiers) {
      classifierScopes.add(correlationScope(classifier.survey(), classifier.name(), classifier.flavor()));
      scopedOCols.addAll(g().V().has("lbl",        "OCol"             ).
                                has("survey",     classifier.survey()).
                                has("classifier", classifier.name()  ).
                                has("flavor",     classifier.flavor()).
                                has("cls").
                                toList());
      malformedScopedOCols.addAll(g().V().has("lbl",        "OCol"             ).
                                          has("survey",     classifier.survey()).
                                          has("classifier", classifier.name()  ).
                                          has("flavor",     classifier.flavor()).
                                          not(has("cls")).
                                          toList());
      }
    // Accumulate correlations and sizes
    Map<OCol, Double>             weights0 = new HashMap<>(); // cls -> weight (for one objext)
    Map<Pair<OCol, OCol>, Double> corrS    = new HashMap<>(); // [cls1, cls2] -> weight (for all object between OCol-OCol)
    Map<OCol, Double>             sizeS    = new HashMap<>(); // cls -> total (for all objects of OCol)
    SortedSet<OCol>               types0   = new TreeSet<>(); // [cls] (for one object)
    SortedSet<OCol>               types    = new TreeSet<>(); // [cls] (for all objects)
    Vertex object;
    Iterator<Edge> deepcontainsIt;
    Edge deepcontains;
    Edge overlaps;
    double weight;
    double weight1;
    double weight2;
    double cor;
    Vertex ocol;
    Vertex ocol1;
    Vertex ocol2;
    OCol cls;
    Pair<OCol, OCol> rel;
    Map<OCol, Vertex> uniqueOCols = new HashMap<>();
    Set<OCol> duplicateOCols = new HashSet<>();
    for (Vertex scopedOCol : scopedOCols) {
      cls = new OCol(scopedOCol);
      if (uniqueOCols.putIfAbsent(cls, scopedOCol) != null) {
        duplicateOCols.add(cls);
        }
      }
    // Loop over objets and accumulated weights to each object
    GraphTraversal<Vertex, Vertex> objectT = g().V().has("lbl", "object");
    while (objectT.hasNext()) {
      weights0.clear();
      types0.clear();
      object = objectT.next();
      deepcontainsIt = object.edges(Direction.IN, "deepcontains");
      // Get weights only from OCols in the requested classifier scope.
      while (deepcontainsIt.hasNext()) {
        deepcontains = deepcontainsIt.next();
        ocol1 = deepcontains.outVertex();
        if (!ocol1.property("lbl").isPresent() ||
            !"OCol".equals(ocol1.property("lbl").value()) ||
            !ocol1.property("survey").isPresent() ||
            !ocol1.property("classifier").isPresent() ||
            !ocol1.property("flavor").isPresent() ||
            !ocol1.property("cls").isPresent()) {
          continue;
          }
        if (!classifierScopes.contains(correlationScope(ocol1.property("survey").value().toString(),
                                                         ocol1.property("classifier").value().toString(),
                                                         ocol1.property("flavor").value().toString()))) {
          continue;
          }
        if (!deepcontains.property("weight").isPresent()) {
          throw new IllegalArgumentException("Scoped deepcontains edge has no weight");
          }
        try {
          weight = Double.parseDouble(deepcontains.property("weight").value().toString());
          }
        catch (NumberFormatException e) {
          throw new IllegalArgumentException("Scoped deepcontains edge has a non-numeric weight", e);
          }
        if (!Double.isFinite(weight) || weight < 0.0) {
          throw new IllegalArgumentException("Scoped deepcontains edge has an invalid weight: " + weight);
          }
        cls = new OCol(ocol1);
        types0.add(cls);
        types.add(cls);
        weights0.put(cls, weight);
        }
      // Double loop over accumulated weights and fill weights between OCols
      for (OCol cls1 : types0) {
        weight1 = weights0.get(cls1);
        for (OCol cls2 : types0) {
          weight2 = weights0.get(cls2);
          rel = Pair.of(cls1, cls2);
          // OCol-OCol
          if (!corrS.containsKey(rel)) {
            corrS.put(rel, 0.0);
            }
          cor = corrS.get(rel);
          //corrS.put(rel, cor + 1.0);
          corrS.put(rel, cor + Math.sqrt(weight1 * weight2));
          }
        }
      }
    // Fill total sizes
    double sizeS0;
    for (OCol cls1 : types) {
      sizeS0 = 0.0;
      for (OCol cls2 : types) {
        if (corrS.containsKey(Pair.of(cls1, cls2))) {
          sizeS0 += corrS.get(Pair.of(cls1, cls2));
          }
        }
      sizeS.put(cls1, sizeS0);
      }
    // Mutate only after all scoped correlation inputs have been validated.
    for (Vertex malformedScopedOCol : malformedScopedOCols) {
      g().V(malformedScopedOCol).drop().iterate();
      }
    // Preserve overlaps crossing the requested-scope boundary; those require
    // both classifier scopes to be regenerated together.
    Set<Object> scopedIds = new HashSet<>();
    for (Vertex scopedOCol : scopedOCols) {
      scopedIds.add(scopedOCol.id());
      }
    Set<Object> internalOverlapIds = new HashSet<>();
    for (Vertex scopedOCol : scopedOCols) {
      Iterator<Edge> overlapsIt = scopedOCol.edges(Direction.BOTH, "overlaps");
      while (overlapsIt.hasNext()) {
        Edge oldOverlap = overlapsIt.next();
        if (scopedIds.contains(oldOverlap.outVertex().id()) &&
            scopedIds.contains(oldOverlap.inVertex().id())) {
          internalOverlapIds.add(oldOverlap.id());
          }
        }
      }
    if (!internalOverlapIds.isEmpty()) {
      g().E(internalOverlapIds.toArray()).drop().iterate();
      }
    // Create overlaps
    int ns = 0;
    // Loop over OCol-OCol and create overlaps Edge OCol-OCol if non empty 
    OCol cls1;
    OCol cls2;
    for (Map.Entry<Pair<OCol, OCol>, Double> entry : corrS.entrySet()) {
      rel = entry.getKey();
      cls1 = rel.first();
      cls2 = rel.second();
      weight = entry.getValue();
      ocol1 = duplicateOCols.contains(cls1) ? findOCol(cls1) : uniqueOCols.get(cls1);
      ocol2 = duplicateOCols.contains(cls2) ? findOCol(cls2) : uniqueOCols.get(cls2);
      overlaps = ocol1.addEdge("overlaps", ocol2);
      overlaps.property("lbl",          "overlaps");
      overlaps.property("intersection", weight);
      overlaps.property("sizeIn",       sizeS.get(cls1));
      overlaps.property("sizeOut",      sizeS.get(cls2));
      ns++;
      }
    commit();
    log.info("" + ns + " object-object correlations generated");
    }

  /** Find the endpoint selected by the existing indexed OCol lookup.
    * @param cls The logical OCol identity.
    * @return The first matching physical OCol endpoint. */
  protected Vertex findOCol(OCol cls) {
    return g().V().has("lbl",        "OCol"          ).
                   has("survey",     cls.survey()    ).
                   has("classifier", cls.classifier()).
                   has("flavor",     cls.flavor()    ).
                   has("cls",        cls.cls()       ).
                   next();
    }
    
  /** Create a new {@link FinkHBaseClient}. Singleton when url unchanged.
    * @param hbaseUrl The HBase url as {@code ip:port:table[:schema]}.
    * @return         The corresponding {@link FinkHBaseClient}, created and initialised if needed.
    * @throws LomikelException If cannot be created. */
  public synchronized FinkHBaseClient fhclient(String hbaseUrl) throws LomikelException {
    if (hbaseUrl == null) {
      return _fhclient;
      }
    if (hbaseUrl.equals(_fhclientUrl) && _fhclient != null) {
      return _fhclient;
      }
    String[] url = hbaseUrl.split(":", -1);
    if ((url.length != 3 && url.length != 4) ||
        url[0].isEmpty() || url[1].isEmpty() || url[2].isEmpty()) {
      throw new LomikelException("Invalid HBase URL (expected ip:port:table[:schema]): " + hbaseUrl);
      }
    String ip     = url[0];
    String port   = url[1];
    String table  = url[2];
    String schema = url.length == 4 ? url[3] : "";
    FinkHBaseClient candidate = null;
    try {
      candidate = new FinkHBaseClient(ip, port);
      candidate.connect(table, schema);
      }
    catch (RuntimeException | LomikelException failure) {
      if (candidate != null) {
        candidate.close();
        }
      if (failure instanceof LomikelException) {
        throw (LomikelException)failure;
        }
      throw new LomikelException("Cannot initialise FinkHBaseClient for " + hbaseUrl, failure);
      }
    FinkHBaseClient previous = _fhclient;
    _fhclient = candidate;
    _fhclientUrl = hbaseUrl;
    if (previous != null) {
      previous.close();
      }
    return _fhclient;
    }
    
  /** Get existing {@link FinkHBaseClient}.
    * @return The corresponding {@link FinkHBaseClient}.
    * @throws LomikelException If not yet created. */
  public FinkHBaseClient fhclient() throws LomikelException {
    if (_fhclient == null) {
      throw new LomikelException("FinkHBaseClient not initialised");
      }
    return _fhclient;
    }
    
  /** Give HBase url.
    * @return The HBase url as {@code ip:port:table[:schema]}. */
  public String hbaseUrl() {
    return _fhclientUrl;
    }
    
  /** Build an unambiguous identity for a correlation-generation scope. */
  private static String correlationScope(String survey,
                                         String classifier,
                                         String flavor) {
    return survey + "\u0000" + classifier + "\u0000" + flavor;
    }

  private FinkHBaseClient _fhclient;
  
  private String _fhclientUrl;

  /** Whether this thread's registration participates in a classification transaction. */
  private ThreadLocal<Boolean> _classificationTransaction = ThreadLocal.withInitial(() -> false);

  /** Objects already selected or created by this thread's classification transaction. */
  private ThreadLocal<Map<String, Vertex>> _classificationObjects = ThreadLocal.withInitial(HashMap::new);
   
 
  private static String FINK_OBJECTS_WS = "https://api.ztf.fink-portal.org/api/v1/objects";
  private static String FINK_LATESTS_WS = "https://api.ztf.fink-portal.org/api/v1/latests";
  private static String FINK_ANOMALY_WS = "https://api.ztf.fink-portal.org/api/v1/anomaly";

  /** Logging . */
  private static Logger log = LogManager.getLogger(FinkGremlinRecipies.class);

  }