<commit_before>package peergos.shared.storage;

import peergos.shared.*;
import peergos.shared.cbor.*;
import peergos.shared.corenode.*;
import peergos.shared.crypto.hash.*;
import peergos.shared.io.ipfs.cid.*;
import peergos.shared.io.ipfs.multibase.binary.*;
import peergos.shared.io.ipfs.multihash.*;
import peergos.shared.user.*;
import peergos.shared.user.fs.*;
import peergos.shared.util.*;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.*;

public class DirectS3BlockStore implements ContentAddressedStorage {

    private final boolean directWrites, publicReads, authedReads;
    private final Optional<String> basePublicReadUrl;
    private final Optional<String> baseAuthedUrl;
    private final HttpPoster direct;
    private final ContentAddressedStorage fallback;
    private final Multihash nodeId;
    private final LRUCache<PublicKeyHash, Multihash> storageNodeByOwner = new LRUCache<>(100);
    private final CoreNode core;

    public DirectS3BlockStore(BlockStoreProperties blockStoreProperties,
                              HttpPoster direct,
                              ContentAddressedStorage fallback,
                              Multihash nodeId,
                              CoreNode core) {
        this.directWrites = blockStoreProperties.directWrites;
        this.publicReads = blockStoreProperties.publicReads;
        this.authedReads = blockStoreProperties.authedReads;
        this.basePublicReadUrl = blockStoreProperties.basePublicReadUrl;
        this.baseAuthedUrl = blockStoreProperties.baseAuthedUrl;
        this.direct = direct;
        this.fallback = fallback;
        this.nodeId = nodeId;
        this.core = core;
    }

    @Override
    public CompletableFuture<BlockStoreProperties> blockStoreProperties() {
        return Futures.of(new BlockStoreProperties(directWrites, publicReads, authedReads, basePublicReadUrl, baseAuthedUrl));
    }

    @Override
    public ContentAddressedStorage directToOrigin() {
        return fallback;
    }

    public static String hashToKey(Multihash hash) {
        // To be compatible with IPFS we use the same scheme here, the cid bytes encoded as uppercase base32
        String padded = new Base32().encodeAsString(hash.toBytes());
        int padStart = padded.indexOf("=");
        return padStart > 0 ? padded.substring(0, padStart) : padded;
    }

    public static Multihash keyToHash(String keyFileName) {
        // To be compatible with IPFS we use the same scheme here, the cid bytes encoded as uppercase base32
        byte[] decoded = new Base32().decode(keyFileName);
        return Cid.cast(decoded);
    }

    @Override
    public CompletableFuture<Multihash> id() {
        return fallback.id();
    }

    @Override
    public CompletableFuture<TransactionId> startTransaction(PublicKeyHash owner) {
        return fallback.startTransaction(owner);
    }

    @Override
    public CompletableFuture<Boolean> closeTransaction(PublicKeyHash owner, TransactionId tid) {
        return fallback.closeTransaction(owner, tid);
    }

    @Override
    public CompletableFuture<List<Multihash>> put(PublicKeyHash owner,
                                                  PublicKeyHash writer,
                                                  List<byte[]> signedHashes,
                                                  List<byte[]> blocks,
                                                  TransactionId tid) {
        return onOwnersNode(owner).thenCompose(ownersNode -> {
            if (ownersNode && directWrites) {
                CompletableFuture<List<Multihash>> res = new CompletableFuture<>();
                fallback.authWrites(owner, writer, signedHashes, blocks.stream().map(x -> x.length).collect(Collectors.toList()), false, tid)
                        .thenCompose(preAuthed -> {
                            List<CompletableFuture<Multihash>> futures = new ArrayList<>();
                            for (int i = 0; i < blocks.size(); i++) {
                                PresignedUrl url = preAuthed.get(i);
                                Multihash targetName = keyToHash(url.base.substring(url.base.lastIndexOf("/") + 1));
                                futures.add(direct.put(url.base, blocks.get(i), url.fields)
                                        .thenApply(x -> targetName));
                            }
                            return Futures.combineAllInOrder(futures);
                        }).thenApply(res::complete)
                        .exceptionally(res::completeExceptionally);
                return res;
            }
            return fallback.put(owner, writer, signedHashes, blocks, tid);
        });
    }

    private CompletableFuture<Boolean> onOwnersNode(PublicKeyHash owner) {
        Multihash cached = storageNodeByOwner.get(owner);
        if (cached != null)
            return Futures.of(cached.equals(nodeId));
        return core.getUsername(owner)
                .thenCompose(user -> core.getChain(user))
                .thenApply(chain -> {
                    List<Multihash> storageProviders = chain.get(chain.size() - 1).claim.storageProviders;
                    Multihash mainNode = storageProviders.get(0);
                    storageNodeByOwner.put(owner, mainNode);
                    System.out.println("Are we on owner's node? " + mainNode + " == " + nodeId);
                    return mainNode.equals(nodeId);
                });
    }

    @Override
    public CompletableFuture<List<Multihash>> putRaw(PublicKeyHash owner,
                                                     PublicKeyHash writer,
                                                     List<byte[]> signatures,
                                                     List<byte[]> blocks,<commit_msg>Add useful error message for empty chain in direct s3 blockstore<commit_after>package peergos.shared.storage;

import peergos.shared.*;
import peergos.shared.cbor.*;
import peergos.shared.corenode.*;
import peergos.shared.crypto.hash.*;
import peergos.shared.io.ipfs.cid.*;
import peergos.shared.io.ipfs.multibase.binary.*;
import peergos.shared.io.ipfs.multihash.*;
import peergos.shared.user.*;
import peergos.shared.user.fs.*;
import peergos.shared.util.*;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.*;

public class DirectS3BlockStore implements ContentAddressedStorage {

    private final boolean directWrites, publicReads, authedReads;
    private final Optional<String> basePublicReadUrl;
    private final Optional<String> baseAuthedUrl;
    private final HttpPoster direct;
    private final ContentAddressedStorage fallback;
    private final Multihash nodeId;
    private final LRUCache<PublicKeyHash, Multihash> storageNodeByOwner = new LRUCache<>(100);
    private final CoreNode core;

    public DirectS3BlockStore(BlockStoreProperties blockStoreProperties,
                              HttpPoster direct,
                              ContentAddressedStorage fallback,
                              Multihash nodeId,
                              CoreNode core) {
        this.directWrites = blockStoreProperties.directWrites;
        this.publicReads = blockStoreProperties.publicReads;
        this.authedReads = blockStoreProperties.authedReads;
        this.basePublicReadUrl = blockStoreProperties.basePublicReadUrl;
        this.baseAuthedUrl = blockStoreProperties.baseAuthedUrl;
        this.direct = direct;
        this.fallback = fallback;
        this.nodeId = nodeId;
        this.core = core;
    }

    @Override
    public CompletableFuture<BlockStoreProperties> blockStoreProperties() {
        return Futures.of(new BlockStoreProperties(directWrites, publicReads, authedReads, basePublicReadUrl, baseAuthedUrl));
    }

    @Override
    public ContentAddressedStorage directToOrigin() {
        return fallback;
    }

    public static String hashToKey(Multihash hash) {
        // To be compatible with IPFS we use the same scheme here, the cid bytes encoded as uppercase base32
        String padded = new Base32().encodeAsString(hash.toBytes());
        int padStart = padded.indexOf("=");
        return padStart > 0 ? padded.substring(0, padStart) : padded;
    }

    public static Multihash keyToHash(String keyFileName) {
        // To be compatible with IPFS we use the same scheme here, the cid bytes encoded as uppercase base32
        byte[] decoded = new Base32().decode(keyFileName);
        return Cid.cast(decoded);
    }

    @Override
    public CompletableFuture<Multihash> id() {
        return fallback.id();
    }

    @Override
    public CompletableFuture<TransactionId> startTransaction(PublicKeyHash owner) {
        return fallback.startTransaction(owner);
    }

    @Override
    public CompletableFuture<Boolean> closeTransaction(PublicKeyHash owner, TransactionId tid) {
        return fallback.closeTransaction(owner, tid);
    }

    @Override
    public CompletableFuture<List<Multihash>> put(PublicKeyHash owner,
                                                  PublicKeyHash writer,
                                                  List<byte[]> signedHashes,
                                                  List<byte[]> blocks,
                                                  TransactionId tid) {
        return onOwnersNode(owner).thenCompose(ownersNode -> {
            if (ownersNode && directWrites) {
                CompletableFuture<List<Multihash>> res = new CompletableFuture<>();
                fallback.authWrites(owner, writer, signedHashes, blocks.stream().map(x -> x.length).collect(Collectors.toList()), false, tid)
                        .thenCompose(preAuthed -> {
                            List<CompletableFuture<Multihash>> futures = new ArrayList<>();
                            for (int i = 0; i < blocks.size(); i++) {
                                PresignedUrl url = preAuthed.get(i);
                                Multihash targetName = keyToHash(url.base.substring(url.base.lastIndexOf("/") + 1));
                                futures.add(direct.put(url.base, blocks.get(i), url.fields)
                                        .thenApply(x -> targetName));
                            }
                            return Futures.combineAllInOrder(futures);
                        }).thenApply(res::complete)
                        .exceptionally(res::completeExceptionally);
                return res;
            }
            return fallback.put(owner, writer, signedHashes, blocks, tid);
        });
    }

    private CompletableFuture<Boolean> onOwnersNode(PublicKeyHash owner) {
        Multihash cached = storageNodeByOwner.get(owner);
        if (cached != null)
            return Futures.of(cached.equals(nodeId));
        return core.getUsername(owner)
                .thenCompose(user -> core.getChain(user)
                        .thenApply(chain -> {
                            if (chain.isEmpty())
                                throw new IllegalStateException("Empty chain returned for " + user);
                            List<Multihash> storageProviders = chain.get(chain.size() - 1).claim.storageProviders;
                            Multihash mainNode = storageProviders.get(0);
                            storageNodeByOwner.put(owner, mainNode);
                            System.out.println("Are we on owner's node? " + mainNode + " == " + nodeId);
                            return mainNode.equals(nodeId);
                        }));
    }

    @Override
    public CompletableFuture<List<Multihash>> putRaw(PublicKeyHash owner,
                                                     PublicKeyHash writer,
                                                     List<byte[]> signatures,
                                                     List<byte[]> blocks,
{'fraction_non_alphanumeric': 0.08849632321866284, 'fraction_numerical': 0.003296424647113515, 'mean_word_length': 1.9706251569168969, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 65, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 30, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6809998', 'n_tokens_mistral': 3000, 'n_tokens_neox': 2918, 'n_words': 648}