/*
 * Copyright 2014-2016 Groupon, Inc
 * Copyright 2014-2016 The Billing Project, LLC
 *
 * The Billing Project 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.killbill.billing.invoice.usage;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.annotation.Nullable;

import org.joda.time.LocalDate;
import org.killbill.billing.callcontext.InternalTenantContext;
import org.killbill.billing.catalog.api.BillingMode;
import org.killbill.billing.catalog.api.CatalogApiException;
import org.killbill.billing.catalog.api.Currency;
import org.killbill.billing.catalog.api.Limit;
import org.killbill.billing.catalog.api.Tier;
import org.killbill.billing.catalog.api.TieredBlock;
import org.killbill.billing.catalog.api.Usage;
import org.killbill.billing.catalog.api.UsageType;
import org.killbill.billing.invoice.api.InvoiceItem;
import org.killbill.billing.invoice.api.InvoiceItemType;
import org.killbill.billing.invoice.generator.BillingIntervalDetail;
import org.killbill.billing.invoice.model.UsageInvoiceItem;
import org.killbill.billing.invoice.usage.details.UsageCapacityInArrearDetail;
import org.killbill.billing.invoice.usage.details.UsageConsumableInArrearDetail;
import org.killbill.billing.invoice.usage.details.UsageConsumableInArrearTierUnitDetail;
import org.killbill.billing.invoice.usage.details.UsageInArrearDetail;
import org.killbill.billing.invoice.usage.details.UsageInArrearDetailInitializer;
import org.killbill.billing.invoice.usage.details.UsageInArrearTierUnitDetail;
import org.killbill.billing.junction.BillingEvent;
import org.killbill.billing.usage.RawUsage;
import org.killbill.billing.usage.api.RolledUpUnit;
import org.killbill.billing.usage.api.RolledUpUsage;
import org.killbill.billing.util.config.definition.InvoiceConfig.UsageDetailMode;
import org.killbill.billing.util.jackson.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;

import static org.killbill.billing.invoice.usage.UsageUtils.getCapacityInArrearTier;
import static org.killbill.billing.invoice.usage.UsageUtils.getCapacityInArrearUnitTypes;
import static org.killbill.billing.invoice.usage.UsageUtils.getConsumableInArrearTieredBlocks;
import static org.killbill.billing.invoice.usage.UsageUtils.getConsumableInArrearUnitTypes;

/**
 * There is one such class per subscriptionId, matching a given in arrear/consumable usage section and
 * referenced through a contiguous list of billing events.
 */
public class ContiguousIntervalUsageInArrear {

    private static final Logger log = LoggerFactory.getLogger(ContiguousIntervalUsageInArrear.class);

    private final List<LocalDate> transitionTimes;
    private final List<BillingEvent> billingEvents;

    private final Usage usage;
    private final Set<String> unitTypes;
    private final List<RawUsage> rawSubscriptionUsage;
    private final LocalDate targetDate;
    private final UUID accountId;
    private final UUID invoiceId;
    private final AtomicBoolean isBuilt;
    private final LocalDate rawUsageStartDate;
    private final InternalTenantContext internalTenantContext;
    private final UsageDetailMode usageDetailMode;
    private static final ObjectMapper objectMapper = new ObjectMapper();

    public ContiguousIntervalUsageInArrear(final Usage usage,
                                           final UUID accountId,
                                           final UUID invoiceId,
                                           final List<RawUsage> rawSubscriptionUsage,
                                           final LocalDate targetDate,
                                           final LocalDate rawUsageStartDate,
                                           final UsageDetailMode usageDetailMode,
                                           final InternalTenantContext internalTenantContext) {
        this.usage = usage;
        this.accountId = accountId;
        this.invoiceId = invoiceId;
        this.unitTypes = usage.getUsageType() == UsageType.CAPACITY ? getCapacityInArrearUnitTypes(usage) : getConsumableInArrearUnitTypes(usage);
        this.rawSubscriptionUsage = rawSubscriptionUsage;
        this.targetDate = targetDate;
        this.rawUsageStartDate = rawUsageStartDate;
        this.internalTenantContext = internalTenantContext;
        this.billingEvents = Lists.newLinkedList();
        this.transitionTimes = Lists.newLinkedList();
        this.isBuilt = new AtomicBoolean(false);
        this.usageDetailMode = usageDetailMode;
    }

    /**
     * Builds the transitionTimes associated to that usage section. Those are determined based on billing events for when to start and when to stop,
     * the per usage billingPeriod and finally the targetDate.
     * <p/>
     * Those transition dates define the well defined billing granularity periods that should be billed for that specific usage section.
     *
     * @param closedInterval whether there was a last billing event referencing the usage section or whether this is ongoing and
     *                       then targetDate will define the endDate.
     */
    public ContiguousIntervalUsageInArrear build(final boolean closedInterval) {

        Preconditions.checkState(!isBuilt.get());
        Preconditions.checkState((!closedInterval && billingEvents.size() >= 1) ||
                                 (closedInterval && billingEvents.size() >= 2));

        final LocalDate startDate = internalTenantContext.toLocalDate(billingEvents.get(0).getEffectiveDate());
        if (targetDate.isBefore(startDate)) {
            return this;
        }
        final LocalDate endDate = closedInterval ? internalTenantContext.toLocalDate(billingEvents.get(billingEvents.size() - 1).getEffectiveDate()) : targetDate;

        final BillingIntervalDetail bid = new BillingIntervalDetail(startDate, endDate, targetDate, getBCD(), usage.getBillingPeriod(), usage.getBillingMode());

        int numberOfPeriod = 0;
        // First billingCycleDate prior startDate
        LocalDate nextBillCycleDate = bid.getFutureBillingDateFor(numberOfPeriod);
        if (startDate.compareTo(rawUsageStartDate) >= 0) {
            transitionTimes.add(startDate);
        }
        while (!nextBillCycleDate.isAfter(endDate)) {
            if (nextBillCycleDate.isAfter(startDate)) {
                if (nextBillCycleDate.compareTo(rawUsageStartDate) >= 0) {
                    transitionTimes.add(nextBillCycleDate);
                }
            }
            numberOfPeriod++;
            nextBillCycleDate = bid.getFutureBillingDateFor(numberOfPeriod);
        }
        if (closedInterval &&
            transitionTimes.size() > 0 &&
            endDate.isAfter(transitionTimes.get(transitionTimes.size() - 1))) {
            transitionTimes.add(endDate);
        }
        isBuilt.set(true);
        return this;
    }

    /**
     * Compute the missing usage invoice items based on what should be billed and what has been billed ($ amount comparison).
     *
     * @param existingUsage existing on disk usage items for the subscription
     * @throws CatalogApiException
     */
    public UsageInArrearItemsAndNextNotificationDate computeMissingItemsAndNextNotificationDate(final List<InvoiceItem> existingUsage) throws CatalogApiException {

        Preconditions.checkState(isBuilt.get());

        if (transitionTimes.size() < 2) {
            return new UsageInArrearItemsAndNextNotificationDate(ImmutableList.<InvoiceItem>of(), computeNextNotificationDate());
        }

        final List<InvoiceItem> result = Lists.newLinkedList();

        // We start by generating 'marker' USAGE items with $0 that will allow to correctly insert the next notification for when there is no USAGE to bill.
        // Those will be removed by the invoicing code later so as to not end up with superfluous $0 items
        LocalDate prevDate = null;
        for (final LocalDate curDate : transitionTimes) {
            if (prevDate != null) {
                final InvoiceItem item = new UsageInvoiceItem(invoiceId, accountId, getBundleId(), getSubscriptionId(), getPlanName(),
                                                              getPhaseName(), usage.getName(), prevDate, curDate, BigDecimal.ZERO, getCurrency());
                result.add(item);
            }
            prevDate = curDate;
        }

        final List<RolledUpUsage> allUsage = getRolledUpUsage();
        // Each RolledUpUsage 'ru' is for a specific time period and across all units
        for (final RolledUpUsage ru : allUsage) {

            //
            // Previously billed items:
            //
            // 1. Retrieves current price amount billed for that period of time (and usage section)
            final Iterable<InvoiceItem> billedItems = getBilledItems(ru.getStart(), ru.getEnd(), existingUsage);
            // 2. Verify whether previously built items have the item_details section
            final boolean areAllBilledItemsWithDetails = areAllBilledItemsWithDetails(billedItems);
            // 3. Computes total billed usage amount
            final BigDecimal billedUsage = computeBilledUsage(billedItems);

            final List<RolledUpUnit> rolledUpUnits = ru.getRolledUpUnits();

            UsageInArrearDetail toBeBilledUsageDetails = new UsageInArrearDetailInitializer();
            BigDecimal toBeBilledUsage;
            if (usage.getUsageType() == UsageType.CAPACITY) {
                // In CAPACITY mode we build across units
                toBeBilledUsageDetails = computeToBeBilledCapacityInArrear(rolledUpUnits);
            } else /* UsageType.CONSUMABLE */ {

                final Map<String, List<UsageConsumableInArrearTierUnitDetail>> previousUnitsUsage;
                if (areAllBilledItemsWithDetails) {
                    previousUnitsUsage = new HashMap<String, List<UsageConsumableInArrearTierUnitDetail>>();
                    for (RolledUpUnit cur : rolledUpUnits) {
                        final List<UsageConsumableInArrearTierUnitDetail> usageInArrearDetailForUnitType = getUsageConsumableInArrearDetailForUnitType(billedItems, cur.getUnitType());
                        previousUnitsUsage.put(cur.getUnitType(), usageInArrearDetailForUnitType);
                    }
                } else {
                    previousUnitsUsage = ImmutableMap.of();
                }

                // In CONSUMABLE mode we build for each unit independently, hence *for* loop
                final List<UsageConsumableInArrearTierUnitDetail> usageConsumableInArrearTierUnitDetails = new ArrayList<UsageConsumableInArrearTierUnitDetail>();
                for (final RolledUpUnit cur : rolledUpUnits) {
                    if (!unitTypes.contains(cur.getUnitType())) {
                        log.warn("ContiguousIntervalConsumableInArrear is skipping unitType " + cur.getUnitType());
                        continue;
                    }
                    final List<UsageConsumableInArrearTierUnitDetail> previousUsage = previousUnitsUsage.containsKey(cur.getUnitType()) ? previousUnitsUsage.get(cur.getUnitType()) : ImmutableList.<UsageConsumableInArrearTierUnitDetail>of();

                    usageConsumableInArrearTierUnitDetails.addAll(computeToBeBilledConsumableInArrear(cur, previousUsage, areAllBilledItemsWithDetails));
                }
                toBeBilledUsageDetails = new UsageConsumableInArrearDetail(usageConsumableInArrearTierUnitDetails);
            }
            // Based on details computed, compute total (proposed) amount
            toBeBilledUsage = toBeBilledUsageDetails.getAmount();

            // Compute final amount by subtracting  amount that was already billed.
            if (!billedItems.iterator().hasNext() || billedUsage.compareTo(toBeBilledUsage) < 0) {
                final BigDecimal amountToBill;
                if (areAllBilledItemsWithDetails && usage.getUsageType() == UsageType.CONSUMABLE) {
                    amountToBill = toBeBilledUsage;
                } else {
                    amountToBill = toBeBilledUsage.subtract(billedUsage);
                }

                if (amountToBill.compareTo(BigDecimal.ZERO) > 0) {
                    if (UsageDetailMode.DETAIL == usageDetailMode && usage.getUsageType() == UsageType.CONSUMABLE) {
                        for (UsageConsumableInArrearTierUnitDetail toBeBilledUsageDetail : ((UsageConsumableInArrearDetail) toBeBilledUsageDetails).getTierDetails()) {
                            final InvoiceItem item = new UsageInvoiceItem(invoiceId, accountId, getBundleId(), getSubscriptionId(), getPlanName(),
                                                                          getPhaseName(), usage.getName(), ru.getStart(), ru.getEnd(), toBeBilledUsageDetail.getAmount(), toBeBilledUsageDetail.getTierPrice(), getCurrency(), toBeBilledUsageDetail.getQuantity(), null);
                            result.add(item);
                        }
                    } else {
                        final String itemDetails = areAllBilledItemsWithDetails ? toBeBilledUsageDetails.toJson(objectMapper) : null;
                        final InvoiceItem item = new UsageInvoiceItem(invoiceId, accountId, getBundleId(), getSubscriptionId(), getPlanName(),
                                                                      getPhaseName(), usage.getName(), ru.getStart(), ru.getEnd(), amountToBill, null, getCurrency(), null, itemDetails);
                        result.add(item);
                    }
                }
            }
        }
        final LocalDate nextNotificationdate = computeNextNotificationDate();
        return new UsageInArrearItemsAndNextNotificationDate(result, nextNotificationdate);
    }

    private List<UsageConsumableInArrearTierUnitDetail> getUsageConsumableInArrearDetailForUnitType(final Iterable<InvoiceItem> billedItems, final String unitType) {

        final List<UsageConsumableInArrearTierUnitDetail> result = new ArrayList<UsageConsumableInArrearTierUnitDetail>();
        for (final InvoiceItem bi : billedItems) {
            final List<UsageConsumableInArrearTierUnitDetail> billedUsageItemDetails = fromJson(bi.getItemDetails());
            for (final UsageConsumableInArrearTierUnitDetail curDetail : billedUsageItemDetails) {
                if (curDetail.getTierUnit().equals(unitType)) {
                    result.add(curDetail);
                }
            }
        }
        return result;
    }

    private boolean areAllBilledItemsWithDetails(final Iterable<InvoiceItem> billedItems) {
        boolean atLeastOneItemWithoutDetails = Iterables.any(billedItems, new Predicate<InvoiceItem>() {
            @Override
            public boolean apply(final InvoiceItem input) {
                return input.getItemDetails() == null || input.getItemDetails().isEmpty();
            }
        });
        return !atLeastOneItemWithoutDetails;
    }

    private LocalDate computeNextNotificationDate() {
        LocalDate result = null;
        final Iterator<BillingEvent> eventIt = billingEvents.iterator();
        BillingEvent nextEvent = eventIt.next();
        while (eventIt.hasNext()) {
            final BillingEvent thisEvent = nextEvent;
            nextEvent = eventIt.next();
            final LocalDate startDate = internalTenantContext.toLocalDate(thisEvent.getEffectiveDate());
            final LocalDate endDate = internalTenantContext.toLocalDate(nextEvent.getEffectiveDate());

            final BillingIntervalDetail bid = new BillingIntervalDetail(startDate, endDate, targetDate, thisEvent.getBillCycleDayLocal(), usage.getBillingPeriod(), BillingMode.IN_ARREAR);
            final LocalDate nextBillingCycleDate = bid.getNextBillingCycleDate();
            result = (result == null || result.compareTo(nextBillingCycleDate) < 0) ? nextBillingCycleDate : result;
        }

        final LocalDate startDate = internalTenantContext.toLocalDate(nextEvent.getEffectiveDate());
        final BillingIntervalDetail bid = new BillingIntervalDetail(startDate, null, targetDate, nextEvent.getBillCycleDayLocal(), usage.getBillingPeriod(), BillingMode.IN_ARREAR);
        final LocalDate nextBillingCycleDate = bid.getNextBillingCycleDate();
        result = (result == null || result.compareTo(nextBillingCycleDate) < 0) ? nextBillingCycleDate : result;
        return result;
    }

    @VisibleForTesting
    List<RolledUpUsage> getRolledUpUsage() {
        final Iterator<RawUsage> rawUsageIterator = rawSubscriptionUsage.iterator();
        if (!rawUsageIterator.hasNext()) {
            return ImmutableList.of();
        }

        final List<RolledUpUsage> result = new ArrayList<RolledUpUsage>();

        //
        // Skip all items before our first transition date
        //
        // prevRawUsage keeps track of first unconsumed raw usage element
        RawUsage prevRawUsage = null;
        while (rawUsageIterator.hasNext()) {
            final RawUsage curRawUsage = rawUsageIterator.next();
            if (curRawUsage.getDate().compareTo(transitionTimes.get(0)) >= 0) {
                prevRawUsage = curRawUsage;
                break;
            }
        }

        // Optimize path where all raw usage items are outside or our transitionTimes range
        if (prevRawUsage.getDate().compareTo(transitionTimes.get(transitionTimes.size() - 1)) >= 0) {
            return ImmutableList.of();
        }

        //
        // Loop through each interval [prevDate, curDate) and consume as many rawSubscriptionUsage elements within that range
        // to create one RolledUpUsage per interval. If an interval does not have any rawSubscriptionUsage element, there will be no
        // matching RolledUpUsage for that interval, and we'll detect that in the 'computeMissingItems' logic
        //
        LocalDate prevDate = null;
        for (final LocalDate curDate : transitionTimes) {

            if (prevDate != null) {

                // Allocate new perRangeUnitToAmount for this interval and populate with rawSubscriptionUsage items
                final Map<String, Long> perRangeUnitToAmount = new HashMap<String, Long>();

                // Start consuming prevRawUsage element if it exists and falls into the range
                if (prevRawUsage != null) {
                    if (prevRawUsage.getDate().compareTo(prevDate) >= 0 && prevRawUsage.getDate().compareTo(curDate) < 0) {
                        final Long currentAmount = perRangeUnitToAmount.get(prevRawUsage.getUnitType());
                        final Long updatedAmount = computeUpdatedAmount(currentAmount, prevRawUsage.getAmount());
                        perRangeUnitToAmount.put(prevRawUsage.getUnitType(), updatedAmount);
                        prevRawUsage = null;
                    }
                }

                //
                // If prevRawUsage != null it means that our first and current rawSubscriptionUsage does not fall into that interval; we can't
                // just 'continue' as we need to correctly set next 'prevDate'
                // If prevRawUsage == null, then consume as much as we can for that interval. Note that the stop condition requires consuming
                // one additional element which will become the prevRawUsage for the next interval.
                //
                if (prevRawUsage == null) {
                    while (rawUsageIterator.hasNext()) {
                        final RawUsage curRawUsage = rawUsageIterator.next();
                        if (curRawUsage.getDate().compareTo(curDate) >= 0) {
                            prevRawUsage = curRawUsage;
                            break;
                        }

                        final Long currentAmount = perRangeUnitToAmount.get(curRawUsage.getUnitType());
                        final Long updatedAmount = computeUpdatedAmount(currentAmount, curRawUsage.getAmount());
                        perRangeUnitToAmount.put(curRawUsage.getUnitType(), updatedAmount);
                    }
                }

                // If we did find some usage for that date range, let's populate the result
                if (!perRangeUnitToAmount.isEmpty()) {
                    final List<RolledUpUnit> rolledUpUnits = new ArrayList<RolledUpUnit>(perRangeUnitToAmount.size());
                    for (final String unitType : perRangeUnitToAmount.keySet()) {
                        rolledUpUnits.add(new DefaultRolledUpUnit(unitType, perRangeUnitToAmount.get(unitType)));
                    }
                    result.add(new DefaultRolledUpUsage(getSubscriptionId(), prevDate, curDate, rolledUpUnits));
                }
            }
            prevDate = curDate;
        }
        return result;
    }

    /**
     * Based on usage type compute new amount
     *
     * @param currentAmount
     * @param newAmount
     * @return
     */
    private Long computeUpdatedAmount(@Nullable Long currentAmount, @Nullable Long newAmount) {

        currentAmount = currentAmount == null ? 0L : currentAmount;
        newAmount = newAmount == null ? 0L : newAmount;

        if (usage.getUsageType() == UsageType.CAPACITY) {
            return Math.max(currentAmount, newAmount);
        } else /* UsageType.CONSUMABLE */ {
            return currentAmount + newAmount;
        }
    }

    private Limit getTierLimit(final Tier tier, final String unitType) {
        for (final Limit cur : tier.getLimits()) {
            if (cur.getUnit().getName().equals(unitType)) {
                return cur;
            }
        }
        Preconditions.checkState(false, "Could not find unit type " + unitType + " in usage tier ");
        return null;
    }

    /**
     * @param roUnits the list of rolled up units for the period
     * @return the price amount that should be billed for that period/unitType
     * @throws CatalogApiException
     */
    @VisibleForTesting
    UsageCapacityInArrearDetail computeToBeBilledCapacityInArrear(final List<RolledUpUnit> roUnits) throws CatalogApiException {
        Preconditions.checkState(isBuilt.get());

        final List<Tier> tiers = getCapacityInArrearTier(usage);

        final Set<String> perUnitTypeDetailTierLevel = new HashSet<String>();
        int tierNum = 0;
        final List<UsageInArrearTierUnitDetail> toBeBilledDetails = Lists.newLinkedList();
        for (final Tier cur : tiers) {
            tierNum++;
            boolean complies = true;
            for (final RolledUpUnit ro : roUnits) {
                final Limit tierLimit = getTierLimit(cur, ro.getUnitType());
                // We ignore the min and only look at the max Limit as the tiers should be contiguous.
                // Specifying a -1 value for last max tier will make the validation works
                if (tierLimit.getMax() != (double) -1 && ro.getAmount().doubleValue() > tierLimit.getMax()) {
                    complies = false;
                } else {
                    if (!perUnitTypeDetailTierLevel.contains(ro.getUnitType())) {
                        toBeBilledDetails.add(new UsageConsumableInArrearTierUnitDetail(tierNum, ro.getUnitType(), cur.getRecurringPrice().getPrice(getCurrency()), 1, ro.getAmount().intValue(), BigDecimal.ZERO));
                        perUnitTypeDetailTierLevel.add(ro.getUnitType());
                    }
                }
            }
            if (complies) {
                return new UsageCapacityInArrearDetail(toBeBilledDetails, cur.getRecurringPrice().getPrice(getCurrency()));
            }
        }
        // Probably invalid catalog config
        final Joiner joiner = Joiner.on(", ");
        joiner.join(roUnits);
        Preconditions.checkState(false, "Could not find tier for usage " + usage.getName() + "matching with data = " + joiner.join(roUnits));
        return null;
    }

    /**
     * @param roUnit the rolled up unit for the period
     * @return the price amount that should be billed for that period/unitType
     * @throws CatalogApiException
     */
    @VisibleForTesting
    List<UsageConsumableInArrearTierUnitDetail> computeToBeBilledConsumableInArrear(final RolledUpUnit roUnit, final List<UsageConsumableInArrearTierUnitDetail> previousUsage, final boolean areAllBilledItemsWithDetails) throws CatalogApiException {

        Preconditions.checkState(isBuilt.get());
        final List<TieredBlock> tieredBlocks = getConsumableInArrearTieredBlocks(usage, roUnit.getUnitType());

        switch (usage.getTierBlockPolicy()) {
            case ALL_TIERS:
                return computeToBeBilledConsumableInArrearWith_ALL_TIERS(tieredBlocks, previousUsage, roUnit.getAmount());
            case TOP_TIER:
                return Arrays.asList(computeToBeBilledConsumableInArrearWith_TOP_TIER(tieredBlocks, previousUsage, roUnit.getAmount(), areAllBilledItemsWithDetails));
            default:
                throw new IllegalStateException("Unknown TierBlockPolicy " + usage.getTierBlockPolicy());
        }
    }

    List<UsageConsumableInArrearTierUnitDetail> computeToBeBilledConsumableInArrearWith_ALL_TIERS(final List<TieredBlock> tieredBlocks, final List<UsageConsumableInArrearTierUnitDetail> previousUsage, final Long units) throws CatalogApiException {

        List<UsageConsumableInArrearTierUnitDetail> toBeBilledDetails = Lists.newLinkedList();
        int remainingUnits = units.intValue();
        int tierNum = 0;

        final int lastPreviousUsageTier = previousUsage.size(); // we count tier from 1, 2, ...
        final boolean hasPreviousUsage = lastPreviousUsageTier > 0;

        for (final TieredBlock tieredBlock : tieredBlocks) {

            tierNum++;
            final int blockTierSize = tieredBlock.getSize().intValue();
            final int tmp = remainingUnits / blockTierSize + (remainingUnits % blockTierSize == 0 ? 0 : 1);
            int nbUsedTierBlocks;
            if (tmp > tieredBlock.getMax()) {
                nbUsedTierBlocks = tieredBlock.getMax().intValue();
                remainingUnits -= tieredBlock.getMax() * blockTierSize;
            } else {
                nbUsedTierBlocks = tmp;
                remainingUnits = 0;
            }

            if (nbUsedTierBlocks > 0) {
                if (hasPreviousUsage) {
                    final Integer previousUsageQuantity = tierNum <= lastPreviousUsageTier ? previousUsage.get(tierNum - 1).getQuantity() : 0;
                    if (tierNum < lastPreviousUsageTier) {
                        Preconditions.checkState(nbUsedTierBlocks == previousUsageQuantity, String.format("Expected usage for tier='%d', unit='%s' to be full, instead found units='[%d/%d]'",
                                                                                                          tierNum, tieredBlock.getUnit().getName(), nbUsedTierBlocks, previousUsageQuantity));
                    } else {
                        Preconditions.checkState(nbUsedTierBlocks - previousUsageQuantity >= 0, String.format("Expected usage for tier='%d', unit='%s' to contain at least as mush as current usage, instead found units='[%d/%d]",
                                                                                                              tierNum, tieredBlock.getUnit().getName(), nbUsedTierBlocks, previousUsageQuantityq));
                    }
                    nbUsedTierBlocks = nbUsedTierBlocks - previousUsageQuantity;
                }
                if (nbUsedTierBlocks > 0) {
                    toBeBilledDetails.add(new UsageConsumableInArrearTierUnitDetail(tierNum, tieredBlock.getUnit().getName(), tieredBlock.getPrice().getPrice(getCurrency()), blockTierSize, nbUsedTierBlocks));
                }
            }
        }
        return toBeBilledDetails;
    }

    UsageConsumableInArrearTierUnitDetail computeToBeBilledConsumableInArrearWith_TOP_TIER(final List<TieredBlock> tieredBlocks, final List<UsageConsumableInArrearTierUnitDetail> previousUsage, final Long units, final boolean areAllBilledItemsWithDetails) throws CatalogApiException {

        int remainingUnits = units.intValue();

        // By default last last tierBlock
        TieredBlock targetBlock = tieredBlocks.get(tieredBlocks.size() - 1);
        int targetTierNum = tieredBlocks.size();
        int tierNum = 0;
        // Loop through all tier block
        for (final TieredBlock tieredBlock : tieredBlocks) {

            tierNum++;
            final int blockTierSize = tieredBlock.getSize().intValue();
            final int tmp = remainingUnits / blockTierSize + (remainingUnits % blockTierSize == 0 ? 0 : 1);
            if (tmp > tieredBlock.getMax()) {
                remainingUnits -= tieredBlock.getMax() * blockTierSize;
            } else {
                targetBlock = tieredBlock;
                targetTierNum = tierNum;
                break;
            }
        }
        final int lastBlockTierSize = targetBlock.getSize().intValue();
        final int nbBlocks = units.intValue() / lastBlockTierSize + (units.intValue() % lastBlockTierSize == 0 ? 0 : 1);

        return new UsageConsumableInArrearTierUnitDetail(targetTierNum, targetBlock.getUnit().getName(), targetBlock.getPrice().getPrice(getCurrency()), targetBlock.getSize().intValue(), nbBlocks);
    }

    /**
     * @param filteredUsageForInterval the list of invoiceItem to consider
     * @return the price amount that was already billed for that period and usage section (across unitTypes)
     */
    @VisibleForTesting
    BigDecimal computeBilledUsage(final Iterable<InvoiceItem> filteredUsageForInterval) {

        Preconditions.checkState(isBuilt.get());
        BigDecimal billedAmount = BigDecimal.ZERO;
        for (final InvoiceItem ii : filteredUsageForInterval) {
            billedAmount = billedAmount.add(ii.getAmount());
        }
        // Return the billed $ amount (not the # of units)
        return billedAmount;
    }

    List<InvoiceItem> getBilledItems(final LocalDate startDate, final LocalDate endDate, final List<InvoiceItem> existingUsage) {

        Preconditions.checkState(isBuilt.get());
        final Iterable<InvoiceItem> filteredResult = Iterables.filter(existingUsage, new Predicate<InvoiceItem>() {
            @Override
            public boolean apply(final InvoiceItem input) {
                if (input.getInvoiceItemType() != InvoiceItemType.USAGE) {
                    return false;
                }
                // STEPH what happens if we discover usage period that overlap (one side or both side) the [startDate, endDate] interval
                final UsageInvoiceItem usageInput = (UsageInvoiceItem) input;
                return usageInput.getUsageName().equals(usage.getName()) &&
                       usageInput.getStartDate().compareTo(startDate) >= 0 &&
                       usageInput.getEndDate().compareTo(endDate) <= 0;
            }
        });
        return ImmutableList.copyOf(filteredResult);
    }

    @VisibleForTesting
    List<LocalDate> getTransitionTimes() {
        return transitionTimes;
    }

    public void addBillingEvent(final BillingEvent event) {
        Preconditions.checkState(!isBuilt.get());
        billingEvents.add(event);
    }

    public Usage getUsage() {
        return usage;
    }

    public int getBCD() {
        return billingEvents.get(0).getBillCycleDayLocal();
    }

    public UUID getBundleId() {
        return billingEvents.get(0).getSubscription().getBundleId();
    }

    public UUID getSubscriptionId() {
        return billingEvents.get(0).getSubscription().getId();
    }

    // STEPH_USAGE planName/phaseName,BCD,... might not be correct if we changed plan but Usage section was exactly similar
    public String getPlanName() {
        return billingEvents.get(0).getPlan().getName();
    }

    public String getPhaseName() {
        return billingEvents.get(0).getPlanPhase().getName();
    }

    public Currency getCurrency() {
        return billingEvents.get(0).getCurrency();
    }

    @Override
    public String toString() {
        final StringBuilder sb = new StringBuilder("ContiguousIntervalConsumableInArrear{");
        sb.append("transitionTimes=").append(transitionTimes);
        sb.append(", billingEvents=").append(billingEvents);
        sb.append(", rawSubscriptionUsage=").append(rawSubscriptionUsage);
        sb.append(", rawUsageStartDate=").append(rawUsageStartDate);
        sb.append('}');
        return sb.toString();
    }

    public class UsageInArrearItemsAndNextNotificationDate {

        private final List<InvoiceItem> invoiceItems;
        private final LocalDate nextNotificationDate;

        public UsageInArrearItemsAndNextNotificationDate(final List<InvoiceItem> invoiceItems, final LocalDate nextNotificationDate) {
            this.invoiceItems = invoiceItems;
            this.nextNotificationDate = nextNotificationDate;
        }

        public List<InvoiceItem> getInvoiceItems() {
            return invoiceItems;
        }

        public LocalDate getNextNotificationDate() {
            return nextNotificationDate;
        }
    }

    public BigDecimal toBeBilledForUnit(final List<UsageConsumableInArrearTierUnitDetail> toBeBilledDetails, final UsageType usageType) {
        if (usageType == UsageType.CAPACITY) {
            return toBeBilledDetails.get(0).getAmount();
        } else {
            BigDecimal result = BigDecimal.ZERO;
            for (UsageConsumableInArrearTierUnitDetail toBeBilled : toBeBilledDetails) {
                result = result.add(toBeBilled.getAmount());
            }
            return result;
        }
    }

    public static List<UsageConsumableInArrearTierUnitDetail> fromJson(String itemDetails) {
        List<UsageConsumableInArrearTierUnitDetail> toBeBilledUsageConsumableInArrearTierUnitDetails = null;
        if (itemDetails != null) {
            try {
                toBeBilledUsageConsumableInArrearTierUnitDetails = objectMapper.readValue(itemDetails, new TypeReference<List<UsageConsumableInArrearTierUnitDetail>>() {});
            } catch (IOException e) {
                Preconditions.checkState(false, e.getMessage());
            }
        }

        return toBeBilledUsageConsumableInArrearTierUnitDetails;
    }

}
