<commit_before>package peergos.shared.user;

import peergos.shared.*;
import peergos.shared.cbor.*;
import peergos.shared.user.fs.*;
import peergos.shared.util.*;

import java.nio.file.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.*;
import java.util.stream.*;

public class SharedWithCache {

    public enum Access { READ, WRITE }

    private static final String DIR_CACHE_FILENAME = "sharedWith.cbor";
    private static final String CACHE_BASE_NAME = "outbound";
    private static final Path CACHE_BASE = Paths.get(CapabilityStore.CAPABILITY_CACHE_DIR, CACHE_BASE_NAME);

    private final Function<Path, CompletableFuture<Optional<FileWrapper>>> retriever;
    private final Path cacheBase;
    private final String ourname;
    private final NetworkAccess network;
    private final Crypto crypto;

    public SharedWithCache(Function<Path, CompletableFuture<Optional<FileWrapper>>> retriever,
                           String ourname,
                           NetworkAccess network,
                           Crypto crypto) {
        this.retriever = retriever;
        this.cacheBase = Paths.get("/" + ourname).resolve(CACHE_BASE);
        this.ourname = ourname;
        this.network = network;
        this.crypto = crypto;
    }

    private static Path canonicalise(Path p) {
        return p.isAbsolute() ? p : Paths.get("/").resolve(p);
    }

    private CompletableFuture<Optional<SharedWithState>> retrieve(Path dir) {
        return retrieveWithFile(dir).thenApply(opt -> opt.map(p -> p.right));
    }

    private CompletableFuture<Optional<Pair<FileWrapper, SharedWithState>>> retrieveWithFile(Path dir) {
        return retriever.apply(cacheBase.resolve(dir).resolve(DIR_CACHE_FILENAME))
                .thenCompose(opt -> opt.isEmpty() ?
                        Futures.of(Optional.empty()) :
                        parseCacheFile(opt.get())
                                .thenApply(s -> new Pair<>(opt.get(), s))
                                .thenApply(Optional::of)
                );
    }

    /**
     *
     * @return root of cache
     */
    private CompletableFuture<FileWrapper> initializeCache() {
        return retriever.apply(Paths.get(ourname))
                .thenCompose(userRoot -> getOrMkdir(userRoot.get(), CapabilityStore.CAPABILITY_CACHE_DIR))
                .thenCompose(cacheRoot -> getOrMkdir(cacheRoot, CACHE_BASE_NAME)); //TODO build from outbound cap files
    }

    private CompletableFuture<FileWrapper> getOrMkdir(FileWrapper parent, String dirName) {
        return parent.getChild(dirName, crypto.hasher, network)
                .thenCompose(opt -> opt.isPresent() ?
                        Futures.of(opt.get()) :
                        parent.mkdir(dirName, network, true, crypto)
                                .thenCompose(p -> p.getChild(dirName, crypto.hasher, network))
                                .thenApply(Optional::get));
    }

    private CompletableFuture<FileWrapper> getOrMkdirs(FileWrapper parent, List<String> remaining) {
        if (remaining.isEmpty())
            return Futures.of(parent);
        return getOrMkdir(parent, remaining.get(0))
                .thenCompose(child -> getOrMkdirs(child, remaining.subList(1, remaining.size())));
    }

    private CompletableFuture<Pair<FileWrapper, SharedWithState>> retrieveWithFileOrCreate(Path dir) {
        return retriever.apply(cacheBase)
                .thenCompose(opt -> opt.isEmpty() ?
                        initializeCache() :
                        Futures.of(opt.get())
                ).thenCompose(cacheRoot -> getOrMkdirs(cacheRoot, toList(dir)))
                .thenCompose(parent -> parent.getChild(DIR_CACHE_FILENAME, crypto.hasher, network)
                        .thenCompose(fopt -> {
                            if (fopt.isPresent())
                                return parseCacheFile(fopt.get())
                                        .thenApply(c -> new Pair<>(fopt.get(), c));
                            SharedWithState empty = SharedWithState.empty();
                            byte[] raw = empty.serialize();
                            return parent.uploadOrReplaceFile(DIR_CACHE_FILENAME, AsyncReader.build(raw), raw.length,
                                    network, crypto, x -> {}, crypto.random.randomBytes(32))
                                    .thenCompose(updatedParent -> updatedParent.getChild(DIR_CACHE_FILENAME, crypto.hasher, network))
                                    .thenApply(copt -> new Pair<>(copt.get(), empty));
                        }));
    }

    private List<String> toList(Path p) {
        return Arrays.asList(p.toString().split("/"));
    }

    private CompletableFuture<SharedWithState> parseCacheFile(FileWrapper cache) {
        return cache.getInputStream(network, crypto, x -> {})
                .thenCompose(in -> Serialize.readFully(in, cache.getSize()))
                .thenApply(CborObject::fromByteArray)
                .thenApply(SharedWithState::fromCbor);
    }

    public CompletableFuture<Map<Path, SharedWithState>> getAllDescendantShares(Path start) {
        return retriever.apply(cacheBase.resolve(start.getParent()))
                .thenCompose(opt -> {
                    if (opt.isEmpty())
                        return Futures.of(Collections.emptyMap());
                    FileWrapper parent = opt.get();
                    String filename = start.toFile().getName();
                    return parent.getChild(DIR_CACHE_FILENAME, crypto.hasher, network)
                            .thenCompose(fopt -> fopt.isEmpty() ?
                                    Futures.of(Collections.<Path, SharedWithState>emptyMap()) :
                                    parseCacheFile(fopt.get())
                                            .thenApply(c -> c.filter(filename)
                                                    .map(r -> Collections.singletonMap(start.getParent(), r))
                                                    .orElse(Collections.emptyMap()))
                            ).thenCompose(m -> parent.getChild(filename, crypto.hasher, network)
                                    .thenCompose(copt -> copt.isEmpty() ?
                                            Futures.of(m) :
                                            getAllDescendantSharesRecurse(copt.get(), start)
                                                    .thenApply(d -> merge(d, m))));
                });
    }

    private <K, V> Map<K, V> merge(Map<K, V> a, Map<K, V> b) {
        HashMap<K, V> res = new HashMap<>(a);
        res.putAll(b); // no key conflicts
        return res;
    }

    public CompletableFuture<Map<Path, SharedWithState>> getAllDescendantSharesRecurse(FileWrapper f, Path toUs) {<commit_msg>Fix path resolve issue with absolute paths<commit_after>package peergos.shared.user;

import peergos.shared.*;
import peergos.shared.cbor.*;
import peergos.shared.user.fs.*;
import peergos.shared.util.*;

import java.nio.file.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.*;
import java.util.stream.*;

public class SharedWithCache {

    public enum Access { READ, WRITE }

    private static final String DIR_CACHE_FILENAME = "sharedWith.cbor";
    private static final String CACHE_BASE_NAME = "outbound";
    private static final Path CACHE_BASE = Paths.get(CapabilityStore.CAPABILITY_CACHE_DIR, CACHE_BASE_NAME);

    private final Function<Path, CompletableFuture<Optional<FileWrapper>>> retriever;
    private final Path cacheBase;
    private final String ourname;
    private final NetworkAccess network;
    private final Crypto crypto;

    public SharedWithCache(Function<Path, CompletableFuture<Optional<FileWrapper>>> retriever,
                           String ourname,
                           NetworkAccess network,
                           Crypto crypto) {
        this.retriever = retriever;
        this.cacheBase = Paths.get("/" + ourname).resolve(CACHE_BASE);
        this.ourname = ourname;
        this.network = network;
        this.crypto = crypto;
    }

    private static Path canonicalise(Path p) {
        return p.isAbsolute() ? p : Paths.get("/").resolve(p);
    }

    private static Path toRelative(Path p) {
        return p.isAbsolute() ? Paths.get(p.toString().substring(1)) : p;
    }

    private CompletableFuture<Optional<SharedWithState>> retrieve(Path dir) {
        return retrieveWithFile(dir).thenApply(opt -> opt.map(p -> p.right));
    }

    private CompletableFuture<Optional<Pair<FileWrapper, SharedWithState>>> retrieveWithFile(Path dir) {
        return retriever.apply(cacheBase.resolve(toRelative(dir)).resolve(DIR_CACHE_FILENAME))
                .thenCompose(opt -> opt.isEmpty() ?
                        Futures.of(Optional.empty()) :
                        parseCacheFile(opt.get())
                                .thenApply(s -> new Pair<>(opt.get(), s))
                                .thenApply(Optional::of)
                );
    }

    /**
     *
     * @return root of cache
     */
    private CompletableFuture<FileWrapper> initializeCache() {
        return retriever.apply(Paths.get(ourname))
                .thenCompose(userRoot -> getOrMkdir(userRoot.get(), CapabilityStore.CAPABILITY_CACHE_DIR))
                .thenCompose(cacheRoot -> getOrMkdir(cacheRoot, CACHE_BASE_NAME)); //TODO build from outbound cap files
    }

    private CompletableFuture<FileWrapper> getOrMkdir(FileWrapper parent, String dirName) {
        return parent.getChild(dirName, crypto.hasher, network)
                .thenCompose(opt -> opt.isPresent() ?
                        Futures.of(opt.get()) :
                        parent.mkdir(dirName, network, true, crypto)
                                .thenCompose(p -> p.getChild(dirName, crypto.hasher, network))
                                .thenApply(Optional::get));
    }

    private CompletableFuture<FileWrapper> getOrMkdirs(FileWrapper parent, List<String> remaining) {
        if (remaining.isEmpty())
            return Futures.of(parent);
        return getOrMkdir(parent, remaining.get(0))
                .thenCompose(child -> getOrMkdirs(child, remaining.subList(1, remaining.size())));
    }

    private CompletableFuture<Pair<FileWrapper, SharedWithState>> retrieveWithFileOrCreate(Path dir) {
        return retriever.apply(cacheBase)
                .thenCompose(opt -> opt.isEmpty() ?
                        initializeCache() :
                        Futures.of(opt.get())
                ).thenCompose(cacheRoot -> getOrMkdirs(cacheRoot, toList(dir)))
                .thenCompose(parent -> parent.getChild(DIR_CACHE_FILENAME, crypto.hasher, network)
                        .thenCompose(fopt -> {
                            if (fopt.isPresent())
                                return parseCacheFile(fopt.get())
                                        .thenApply(c -> new Pair<>(fopt.get(), c));
                            SharedWithState empty = SharedWithState.empty();
                            byte[] raw = empty.serialize();
                            return parent.uploadOrReplaceFile(DIR_CACHE_FILENAME, AsyncReader.build(raw), raw.length,
                                    network, crypto, x -> {}, crypto.random.randomBytes(32))
                                    .thenCompose(updatedParent -> updatedParent.getChild(DIR_CACHE_FILENAME, crypto.hasher, network))
                                    .thenApply(copt -> new Pair<>(copt.get(), empty));
                        }));
    }

    private List<String> toList(Path p) {
        return Arrays.asList(p.toString().split("/"));
    }

    private CompletableFuture<SharedWithState> parseCacheFile(FileWrapper cache) {
        return cache.getInputStream(network, crypto, x -> {})
                .thenCompose(in -> Serialize.readFully(in, cache.getSize()))
                .thenApply(CborObject::fromByteArray)
                .thenApply(SharedWithState::fromCbor);
    }

    public CompletableFuture<Map<Path, SharedWithState>> getAllDescendantShares(Path start) {
        return retriever.apply(cacheBase.resolve(toRelative(start.getParent())))
                .thenCompose(opt -> {
                    if (opt.isEmpty())
                        return Futures.of(Collections.emptyMap());
                    FileWrapper parent = opt.get();
                    String filename = start.toFile().getName();
                    return parent.getChild(DIR_CACHE_FILENAME, crypto.hasher, network)
                            .thenCompose(fopt -> fopt.isEmpty() ?
                                    Futures.of(Collections.<Path, SharedWithState>emptyMap()) :
                                    parseCacheFile(fopt.get())
                                            .thenApply(c -> c.filter(filename)
                                                    .map(r -> Collections.singletonMap(start.getParent(), r))
                                                    .orElse(Collections.emptyMap()))
                            ).thenCompose(m -> parent.getChild(filename, crypto.hasher, network)
                                    .thenCompose(copt -> copt.isEmpty() ?
                                            Futures.of(m) :
                                            getAllDescendantSharesRecurse(copt.get(), start)
                                                    .thenApply(d -> merge(d, m))));
                });
    }

    private <K, V> Map<K, V> merge(Map<K, V> a, Map<K, V> b) {
        HashMap<K, V> res = new HashMap<>(a);
        res.putAll(b); // no key conflicts
        return res;
    }

    public CompletableFuture<Map<Path, SharedWithState>> getAllDescendantSharesRecurse(FileWrapper f, Path toUs) {
{'fraction_non_alphanumeric': 0.10964179321368887, 'fraction_numerical': 0.0006539271961055002, 'mean_word_length': 1.835015447991761, 'pattern_counts': {'":': 0, '<': 69, '<?xml version=': 0, '>': 121, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 38, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5516937', 'n_tokens_mistral': 3671, 'n_tokens_neox': 3537, 'n_words': 663}