package pl.allegro.tech.hermes.consumers.supervisor.workload.weighted;

import org.agrona.concurrent.UnsafeBuffer;
import org.slf4j.Logger;
import pl.allegro.tech.hermes.api.SubscriptionName;
import pl.allegro.tech.hermes.consumers.subscription.id.SubscriptionId;
import pl.allegro.tech.hermes.consumers.subscription.id.SubscriptionIds;
import pl.allegro.tech.hermes.consumers.supervisor.workload.sbe.stubs.ConsumerLoadDecoder;
import pl.allegro.tech.hermes.consumers.supervisor.workload.sbe.stubs.ConsumerLoadDecoder.SubscriptionsDecoder;
import pl.allegro.tech.hermes.consumers.supervisor.workload.sbe.stubs.MessageHeaderDecoder;

import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

import static org.slf4j.LoggerFactory.getLogger;

class ConsumerNodeLoadDecoder {

    private static final Logger logger = getLogger(ConsumerNodeLoadDecoder.class);

    private final SubscriptionIds subscriptionIds;

    ConsumerNodeLoadDecoder(SubscriptionIds subscriptionIds) {
        this.subscriptionIds = subscriptionIds;
    }

    ConsumerNodeLoad decode(byte[] bytes) {
        MessageHeaderDecoder header = new MessageHeaderDecoder();
        ConsumerLoadDecoder body = new ConsumerLoadDecoder();

        UnsafeBuffer buffer = new UnsafeBuffer(bytes);
        header.wrap(buffer, 0);

        if (header.schemaId() != ConsumerLoadDecoder.SCHEMA_ID || header.templateId() != ConsumerLoadDecoder.TEMPLATE_ID) {
            logger.warn("Unable to decode consumer node load, schema or template id mismatch. "
                            + "Required by decoder: [schema id={}, template id={}], "
                            + "encoded in payload: [schema id={}, template id={}]",
                    ConsumerLoadDecoder.SCHEMA_ID, ConsumerLoadDecoder.TEMPLATE_ID,
                    header.schemaId(), header.templateId());
            return ConsumerNodeLoad.UNDEFINED;
        }

        body.wrap(buffer, header.encodedLength(), header.blockLength(), header.version());

        return new ConsumerNodeLoad(body.cpuUtilization(), decodeSubscriptionLoads(body));
    }

    private Map<SubscriptionName, SubscriptionLoad> decodeSubscriptionLoads(ConsumerLoadDecoder body) {
        Map<SubscriptionName, SubscriptionLoad> subscriptionLoads = new HashMap<>();
        for (SubscriptionsDecoder loadPerSubscriptionDecoder : body.subscriptions()) {
            long id = loadPerSubscriptionDecoder.id();
            Optional<SubscriptionId> subscriptionId = subscriptionIds.getSubscriptionId(id);
            if (subscriptionId.isPresent()) {
                double operationsPerSecond = loadPerSubscriptionDecoder.operationsPerSecond();
                SubscriptionLoad load = new SubscriptionLoad(operationsPerSecond);
                subscriptionLoads.put(subscriptionId.get().getSubscriptionName(), load);
            }
        }
        return subscriptionLoads;
    }
}
