<commit_before>/*
 * 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.seaborne.delta.server.local;

import static org.seaborne.delta.DPConst.F_BASE;
import static org.seaborne.delta.DPConst.F_ID;
import static org.seaborne.delta.DPConst.F_URI;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import org.apache.jena.atlas.io.IO;
import org.apache.jena.atlas.json.JSON;
import org.apache.jena.atlas.json.JsonBuilder;
import org.apache.jena.atlas.json.JsonObject;
import org.apache.jena.atlas.lib.FileOps;
import org.apache.jena.atlas.lib.Pair;
import org.apache.jena.atlas.logging.FmtLog;
import org.apache.jena.tdb.base.file.Location;
import org.seaborne.delta.*;
import org.seaborne.delta.lib.IOX;
import org.seaborne.delta.lib.JSONX;
import org.seaborne.delta.lib.LibX;
import org.seaborne.delta.link.DeltaLink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** A local server.
 *  <p>
 *  This provides for several {@link DataSource} areas (one per managed patch set - i.e. one per dataset).
 *  {@code LocalServer} is responsible for server wide configuration and for the 
 *  {@link DataSource} lifecycle of create and delete.  
 *    
 * @see DeltaLinkLocal 
 * @see DataSource 
 *
 */
public class LocalServer {
    /* File system layout:
     *   Server Root
     *      delta.cfg
     *      /NAME ... per DataSource.
     *          autogenerated assembler?
     *          /patches
     *          /data -- TDB database (optional)
     *          disabled -- if this file is present, then the datasource is not accessible.  
     *          
     *  But also external databases 
     */
    
    private static Logger LOG = LoggerFactory.getLogger(LocalServer.class);
    
    private static Map<Location, LocalServer> servers = new ConcurrentHashMap<>();
    
    private final DataRegistry dataRegistry;
    private final Location serverRoot;
    private final LocalServerConfig serverConfig;
    private static AtomicInteger counter = new AtomicInteger(0);
    // Cache of known disabled data sources. (Not set of startup).
    private Set<Id> disabledDatasources = new HashSet<>();
    private Object lock = new Object();
    
    /** Attach to the runtime area for the server. Use "delta.cfg" as the configuration file name.  
     * @param serverRoot
     * @return LocalServer
     */
    public static LocalServer attach(Location serverRoot) {
        return attach(serverRoot, DPConst.SERVER_CONFIG); 
    }
    
    /** Attach to the runtime area for the server. 
     * Use "delta.cfg" as the configuration file name.
     * Use the directory fo the configuration file as the location.
     * @param confFile  Filename
     * @return LocalServer
     */
    public static LocalServer attach(String confFile) {
        LocalServerConfig conf = LocalServerConfig.create()
            .parse(confFile)
            .build();
        return attach(conf);
    }
    

    /** Attach to the runtime area for the server.
     * @param serverRoot
     * @param confFile  Filename: absolute filename, or relative to the server process.
     * @return LocalServer
     */
    public static LocalServer attach(Location serverRoot, String confFile) {
        confFile = LibX.resolve(serverRoot, confFile);
        LocalServerConfig conf = LocalServerConfig.create()
            .setLocation(serverRoot)
            .parse(confFile)
            .build();
        return attach(conf);
    }
    
    public static LocalServer attach(LocalServerConfig conf) {
        Objects.requireNonNull(conf, "Null for configuation");
        if ( conf.location == null )
            throw new DeltaConfigException("No location");
        
        DataRegistry dataRegistry = new DataRegistry("Server"+counter.incrementAndGet());
        Pair<List<Path>, List<Path>> pair = scanDirectory(conf.location, dataRegistry);
        List<Path> dataSources = pair.getLeft();
        List<Path> disabledDataSources = pair.getRight();
        
        dataSources.stream().forEach(p->LOG.info("Data source: "+p));
        dataSources.stream().forEach(p->LOG.info("Data source: "+p+" : Disabled"));
        
        for ( Path p : dataSources ) {
            DataSource ds = makeDataSource(p);
            dataRegistry.put(ds.getId(), ds);
        }
        return attach(conf, dataRegistry);
    }
    
    public static void release(LocalServer localServer) {
        Location key = localServer.serverConfig.location;
        if ( key != null )
            servers.remove(key);
    }
     
    /** Scan a directory for datasource areas.
     * These must have a file called   
     */
    private static Pair<List<Path>/*enabled*/, List<Path>/*disabled*/> scanDirectory(Location serverRoot, DataRegistry dataRegistry) {
        Path dir = Paths.get(serverRoot.getDirectoryPath());
        try { 
            List<Path> x = Files.list(dir)
                .filter(LocalServer::isDataSource)
                .collect(Collectors.toList());
            List<Path> enabled = x.stream()
                .filter(path -> isEnabled(path))
                .collect(Collectors.toList());
            List<Path> disabled = x.stream()
                .filter(path -> !isEnabled(path))
                .collect(Collectors.toList());
            return Pair.create(enabled, disabled);
        }
        catch (IOException e) {
            LOG.error("Exception while reading "+dir);
            IO.exception(e);
            return null;
        }
    }

    /** Test for a valid data source - does not check "disabled" */
    private static boolean isDataSource(Path path) {
        if ( ! Files.isDirectory(path) ) 
            return false ;
        Path cfg = path.resolve(DPConst.DATA_CONFIG);
        if ( ! Files.exists(cfg) )
            return false ;
        if ( ! Files.isRegularFile(cfg) ) 
            LOG.warn("Data source configuration file name exists but is not a file: "+cfg);
        if ( ! Files.isReadable(cfg) )
            LOG.warn("Data source configuration file exists but is not readable: "+cfg);
        return true ;
    }
    
    private static boolean isEnabled(Path path) {
        Path disabled = path.resolve(DPConst.DISABLED);
        return ! Files.exists(disabled);
    }

    private static DataSource makeDataSource(Path dataSourceArea) {
        JsonObject sourceObj = JSON.read(dataSourceArea.resolve(DPConst.DATA_CONFIG).toString());
        String idStr = JSONX.getStrOrNull(sourceObj, F_ID) ;
        Id id = Id.fromString(idStr) ; 
        String uriStr = JSONX.getStrOrNull(sourceObj, F_URI) ;
        String baseStr = JSONX.getStrOrNull(sourceObj, F_BASE);
      
        Path patchesArea = dataSourceArea.resolve(DPConst.PATCHES);
        //FmtLog.info(LOG, "DataSource: id=%s, source=%s, patches=%s", id, dataSourceArea, patchesArea);
        
        // --> Path
        DataSource dataSource = DataSource.attach(id, uriStr, IOX.asLocation(dataSourceArea), IOX.asLocation(patchesArea));
        FmtLog.info(LOG, "DataSource: %s (%s)", dataSource, baseStr);
      
        return dataSource ;
    }
    
    private static LocalServer attach(LocalServerConfig config, DataRegistry dataRegistry) {
        Location loc = config.location;
        if ( ! loc.isMemUnique() ) {
            if ( servers.containsKey(loc) ) {
                LocalServer lServer = servers.get(loc);
                LocalServerConfig config2 = lServer.getConfig();
                if ( Objects.equals(config, config2) ) {<commit_msg>Ensure directory 'Patches' exists<commit_after>/*
 * 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.seaborne.delta.server.local;

import static org.seaborne.delta.DPConst.F_BASE;
import static org.seaborne.delta.DPConst.F_ID;
import static org.seaborne.delta.DPConst.F_URI;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import org.apache.jena.atlas.io.IO;
import org.apache.jena.atlas.json.JSON;
import org.apache.jena.atlas.json.JsonBuilder;
import org.apache.jena.atlas.json.JsonObject;
import org.apache.jena.atlas.lib.FileOps;
import org.apache.jena.atlas.lib.Pair;
import org.apache.jena.atlas.logging.FmtLog;
import org.apache.jena.tdb.base.file.Location;
import org.seaborne.delta.*;
import org.seaborne.delta.lib.IOX;
import org.seaborne.delta.lib.JSONX;
import org.seaborne.delta.lib.LibX;
import org.seaborne.delta.link.DeltaLink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** A local server.
 *  <p>
 *  This provides for several {@link DataSource} areas (one per managed patch set - i.e. one per dataset).
 *  {@code LocalServer} is responsible for server wide configuration and for the 
 *  {@link DataSource} lifecycle of create and delete.  
 *    
 * @see DeltaLinkLocal 
 * @see DataSource 
 *
 */
public class LocalServer {
    /* File system layout:
     *   Server Root
     *      delta.cfg
     *      /NAME ... per DataSource.
     *          autogenerated assembler?
     *          /patches
     *          /data -- TDB database (optional)
     *          disabled -- if this file is present, then the datasource is not accessible.  
     *          
     *  But also external databases 
     */
    
    private static Logger LOG = LoggerFactory.getLogger(LocalServer.class);
    
    private static Map<Location, LocalServer> servers = new ConcurrentHashMap<>();
    
    private final DataRegistry dataRegistry;
    private final Location serverRoot;
    private final LocalServerConfig serverConfig;
    private static AtomicInteger counter = new AtomicInteger(0);
    // Cache of known disabled data sources. (Not set of startup).
    private Set<Id> disabledDatasources = new HashSet<>();
    private Object lock = new Object();
    
    /** Attach to the runtime area for the server. Use "delta.cfg" as the configuration file name.  
     * @param serverRoot
     * @return LocalServer
     */
    public static LocalServer attach(Location serverRoot) {
        return attach(serverRoot, DPConst.SERVER_CONFIG); 
    }
    
    /** Attach to the runtime area for the server. 
     * Use "delta.cfg" as the configuration file name.
     * Use the directory fo the configuration file as the location.
     * @param confFile  Filename
     * @return LocalServer
     */
    public static LocalServer attach(String confFile) {
        LocalServerConfig conf = LocalServerConfig.create()
            .parse(confFile)
            .build();
        return attach(conf);
    }
    

    /** Attach to the runtime area for the server.
     * @param serverRoot
     * @param confFile  Filename: absolute filename, or relative to the server process.
     * @return LocalServer
     */
    public static LocalServer attach(Location serverRoot, String confFile) {
        confFile = LibX.resolve(serverRoot, confFile);
        LocalServerConfig conf = LocalServerConfig.create()
            .setLocation(serverRoot)
            .parse(confFile)
            .build();
        return attach(conf);
    }
    
    public static LocalServer attach(LocalServerConfig conf) {
        Objects.requireNonNull(conf, "Null for configuation");
        if ( conf.location == null )
            throw new DeltaConfigException("No location");
        
        DataRegistry dataRegistry = new DataRegistry("Server"+counter.incrementAndGet());
        Pair<List<Path>, List<Path>> pair = scanDirectory(conf.location, dataRegistry);
        List<Path> dataSources = pair.getLeft();
        List<Path> disabledDataSources = pair.getRight();
        
        dataSources.stream().forEach(p->LOG.info("Data source: "+p));
        dataSources.stream().forEach(p->LOG.info("Data source: "+p+" : Disabled"));
        
        for ( Path p : dataSources ) {
            DataSource ds = makeDataSource(p);
            dataRegistry.put(ds.getId(), ds);
        }
        return attach(conf, dataRegistry);
    }
    
    public static void release(LocalServer localServer) {
        Location key = localServer.serverConfig.location;
        if ( key != null )
            servers.remove(key);
    }
     
    /** Scan a directory for datasource areas.
     * These must have a file called   
     */
    private static Pair<List<Path>/*enabled*/, List<Path>/*disabled*/> scanDirectory(Location serverRoot, DataRegistry dataRegistry) {
        Path dir = Paths.get(serverRoot.getDirectoryPath());
        try { 
            List<Path> x = Files.list(dir)
                .filter(LocalServer::isDataSource)
                .collect(Collectors.toList());
            List<Path> enabled = x.stream()
                .filter(path -> isEnabled(path))
                .collect(Collectors.toList());
            List<Path> disabled = x.stream()
                .filter(path -> !isEnabled(path))
                .collect(Collectors.toList());
            return Pair.create(enabled, disabled);
        }
        catch (IOException e) {
            LOG.error("Exception while reading "+dir);
            IO.exception(e);
            return null;
        }
    }

    /** Test for a valid data source - does not check "disabled" */
    private static boolean isDataSource(Path path) {
        if ( ! Files.isDirectory(path) ) 
            return false ;
        Path cfg = path.resolve(DPConst.DATA_CONFIG);
        if ( ! Files.exists(cfg) )
            return false ;
        if ( ! Files.isRegularFile(cfg) ) 
            LOG.warn("Data source configuration file name exists but is not a file: "+cfg);
        if ( ! Files.isReadable(cfg) )
            LOG.warn("Data source configuration file exists but is not readable: "+cfg);
        return true ;
    }
    
    private static boolean isEnabled(Path path) {
        Path disabled = path.resolve(DPConst.DISABLED);
        return ! Files.exists(disabled);
    }

    private static DataSource makeDataSource(Path dataSourceArea) {
        JsonObject sourceObj = JSON.read(dataSourceArea.resolve(DPConst.DATA_CONFIG).toString());
        String idStr = JSONX.getStrOrNull(sourceObj, F_ID) ;
        Id id = Id.fromString(idStr) ; 
        String uriStr = JSONX.getStrOrNull(sourceObj, F_URI) ;
        String baseStr = JSONX.getStrOrNull(sourceObj, F_BASE);
      
        Path patchesArea = dataSourceArea.resolve(DPConst.PATCHES);
        FileOps.ensureDir(patchesArea.toString());
        //FmtLog.info(LOG, "DataSource: id=%s, source=%s, patches=%s", id, dataSourceArea, patchesArea);
        
        // --> Path
        DataSource dataSource = DataSource.attach(id, uriStr, IOX.asLocation(dataSourceArea), IOX.asLocation(patchesArea));
        FmtLog.info(LOG, "DataSource: %s (%s)", dataSource, baseStr);
      
        return dataSource ;
    }
    
    private static LocalServer attach(LocalServerConfig config, DataRegistry dataRegistry) {
        Location loc = config.location;
        if ( ! loc.isMemUnique() ) {
            if ( servers.containsKey(loc) ) {
                LocalServer lServer = servers.get(loc);
                LocalServerConfig config2 = lServer.getConfig();
                if ( Objects.equals(config, config2) ) {
{'fraction_non_alphanumeric': 0.10105848264443262, 'fraction_numerical': 0.0010643959552953698, 'mean_word_length': 2.6252947481243303, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 45, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '143959', 'n_tokens_mistral': 4666, 'n_tokens_neox': 4592, 'n_words': 1322}