<commit_before>package com.jivesoftware.os.wiki.miru.deployable.storage;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.jivesoftware.os.amza.api.BAInterner;
import com.jivesoftware.os.amza.api.PartitionClient;
import com.jivesoftware.os.amza.api.partition.Consistency;
import com.jivesoftware.os.amza.api.partition.Durability;
import com.jivesoftware.os.amza.api.partition.PartitionName;
import com.jivesoftware.os.amza.api.partition.PartitionProperties;
import com.jivesoftware.os.amza.api.stream.RowType;
import com.jivesoftware.os.amza.client.http.AmzaClientProvider;
import com.jivesoftware.os.amza.client.http.HttpPartitionClientFactory;
import com.jivesoftware.os.amza.client.http.HttpPartitionHostsProvider;
import com.jivesoftware.os.amza.client.http.RingHostHttpClientProvider;
import com.jivesoftware.os.miru.api.base.MiruTenantId;
import com.jivesoftware.os.mlogger.core.MetricLogger;
import com.jivesoftware.os.mlogger.core.MetricLoggerFactory;
import com.jivesoftware.os.routing.bird.http.client.HttpClient;
import com.jivesoftware.os.routing.bird.http.client.HttpClientException;
import com.jivesoftware.os.routing.bird.http.client.TenantAwareHttpClient;

import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.Executors;

/**
 *
 */
public class WikiMiruPayloadsAmza {

    private static final MetricLogger LOG = MetricLoggerFactory.getLogger();

    private final ObjectMapper mapper;
    private final AmzaClientProvider<HttpClient, HttpClientException> clientProvider;

    private final String nameSpace;
    private final PartitionProperties partitionProperties;
    private final long additionalSolverAfterNMillis = 1_000; //TODO expose to conf?
    private final long abandonLeaderSolutionAfterNMillis = 5_000; //TODO expose to conf?
    private final long abandonSolutionAfterNMillis = 30_000; //TODO expose to conf?

    public WikiMiruPayloadsAmza(String nameSpace,
        ObjectMapper mapper,
        TenantAwareHttpClient<String> httpClient,
        long awaitLeaderElectionForNMillis) {

        this.nameSpace = nameSpace;
        this.mapper = mapper;
        BAInterner interner = new BAInterner();

        this.clientProvider = new AmzaClientProvider<>(
            new HttpPartitionClientFactory(interner),
            new HttpPartitionHostsProvider(interner, httpClient, mapper),
            new RingHostHttpClientProvider(httpClient),
            Executors.newCachedThreadPool(), //TODO expose to conf?
            awaitLeaderElectionForNMillis,
            -1,
            -1);

        partitionProperties = new PartitionProperties(Durability.fsync_async,
            -1, -1, -1, -1, -1, -1, -1, -1,
            false,
            Consistency.quorum,
            true,
            true,
            false,
            RowType.snappy_primary,
            "lab",
            -1,
            null,
            -1,
            -1);
    }

    private PartitionName getPartitionName(MiruTenantId tenantId) {
        byte[] pname = (nameSpace + "-" + tenantId.toString() + "-wiki").getBytes(StandardCharsets.UTF_8);
        return new PartitionName(false, pname, pname);
    }

    public <T> void multiPut(MiruTenantId tenantId, List<KeyAndPayload<T>> timesAndPayloads) throws Exception {

        PartitionClient partition = clientProvider.getPartition(getPartitionName(tenantId), 3, partitionProperties);
        long now = System.currentTimeMillis();
        partition.commit(Consistency.leader_quorum,
            null,
            (stream) -> {
                for (KeyAndPayload<T> keyAndPayload : timesAndPayloads) {
                    stream.commit(keyAndPayload.key.getBytes(StandardCharsets.UTF_8), mapper.writeValueAsBytes(keyAndPayload.payload), now, false);
                }
                return true;
            },
            additionalSolverAfterNMillis, abandonSolutionAfterNMillis, Optional.empty());

    }

    public <T> T get(MiruTenantId tenantId, String k, Class<T> payloadClass) throws Exception {

        PartitionClient partition = clientProvider.getPartition(getPartitionName(tenantId), 3, partitionProperties);
        T[] t = (T[]) new Object[1];
        partition.get(Consistency.none,<commit_msg>move to quorum writes for wiki payloads<commit_after>package com.jivesoftware.os.wiki.miru.deployable.storage;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.jivesoftware.os.amza.api.BAInterner;
import com.jivesoftware.os.amza.api.PartitionClient;
import com.jivesoftware.os.amza.api.partition.Consistency;
import com.jivesoftware.os.amza.api.partition.Durability;
import com.jivesoftware.os.amza.api.partition.PartitionName;
import com.jivesoftware.os.amza.api.partition.PartitionProperties;
import com.jivesoftware.os.amza.api.stream.RowType;
import com.jivesoftware.os.amza.client.http.AmzaClientProvider;
import com.jivesoftware.os.amza.client.http.HttpPartitionClientFactory;
import com.jivesoftware.os.amza.client.http.HttpPartitionHostsProvider;
import com.jivesoftware.os.amza.client.http.RingHostHttpClientProvider;
import com.jivesoftware.os.miru.api.base.MiruTenantId;
import com.jivesoftware.os.mlogger.core.MetricLogger;
import com.jivesoftware.os.mlogger.core.MetricLoggerFactory;
import com.jivesoftware.os.routing.bird.http.client.HttpClient;
import com.jivesoftware.os.routing.bird.http.client.HttpClientException;
import com.jivesoftware.os.routing.bird.http.client.TenantAwareHttpClient;

import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.Executors;

/**
 *
 */
public class WikiMiruPayloadsAmza {

    private static final MetricLogger LOG = MetricLoggerFactory.getLogger();

    private final ObjectMapper mapper;
    private final AmzaClientProvider<HttpClient, HttpClientException> clientProvider;

    private final String nameSpace;
    private final PartitionProperties partitionProperties;
    private final long additionalSolverAfterNMillis = 1_000; //TODO expose to conf?
    private final long abandonLeaderSolutionAfterNMillis = 5_000; //TODO expose to conf?
    private final long abandonSolutionAfterNMillis = 30_000; //TODO expose to conf?

    public WikiMiruPayloadsAmza(String nameSpace,
        ObjectMapper mapper,
        TenantAwareHttpClient<String> httpClient,
        long awaitLeaderElectionForNMillis) {

        this.nameSpace = nameSpace;
        this.mapper = mapper;
        BAInterner interner = new BAInterner();

        this.clientProvider = new AmzaClientProvider<>(
            new HttpPartitionClientFactory(interner),
            new HttpPartitionHostsProvider(interner, httpClient, mapper),
            new RingHostHttpClientProvider(httpClient),
            Executors.newCachedThreadPool(), //TODO expose to conf?
            awaitLeaderElectionForNMillis,
            -1,
            -1);

        partitionProperties = new PartitionProperties(Durability.fsync_async,
            -1, -1, -1, -1, -1, -1, -1, -1,
            false,
            Consistency.none,
            true,
            true,
            false,
            RowType.snappy_primary,
            "lab",
            -1,
            null,
            -1,
            -1);
    }

    private PartitionName getPartitionName(MiruTenantId tenantId) {
        byte[] pname = (nameSpace + "-" + tenantId.toString() + "-wiki").getBytes(StandardCharsets.UTF_8);
        return new PartitionName(false, pname, pname);
    }

    public <T> void multiPut(MiruTenantId tenantId, List<KeyAndPayload<T>> timesAndPayloads) throws Exception {

        PartitionClient partition = clientProvider.getPartition(getPartitionName(tenantId), 3, partitionProperties);
        long now = System.currentTimeMillis();
        partition.commit(Consistency.quorum,
            null,
            (stream) -> {
                for (KeyAndPayload<T> keyAndPayload : timesAndPayloads) {
                    stream.commit(keyAndPayload.key.getBytes(StandardCharsets.UTF_8), mapper.writeValueAsBytes(keyAndPayload.payload), now, false);
                }
                return true;
            },
            additionalSolverAfterNMillis, abandonSolutionAfterNMillis, Optional.empty());

    }

    public <T> T get(MiruTenantId tenantId, String k, Class<T> payloadClass) throws Exception {

        PartitionClient partition = clientProvider.getPartition(getPartitionName(tenantId), 3, partitionProperties);
        T[] t = (T[]) new Object[1];
        partition.get(Consistency.none,
{'fraction_non_alphanumeric': 0.0940690604565791, 'fraction_numerical': 0.007112538717448664, 'mean_word_length': 3.3611805902951475, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2283215', 'n_tokens_mistral': 2614, 'n_tokens_neox': 2551, 'n_words': 469}