/*
 * Copyright 2014-2020 Groupon, Inc
 * Copyright 2014-2020 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.payment;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

import javax.inject.Inject;

import org.awaitility.Awaitility;
import org.joda.time.LocalDate;
import org.killbill.billing.account.api.Account;
import org.killbill.billing.api.FlakyRetryAnalyzer;
import org.killbill.billing.api.TestApiListener;
import org.killbill.billing.api.TestApiListener.NextEvent;
import org.killbill.billing.callcontext.InternalCallContext;
import org.killbill.billing.catalog.api.Currency;
import org.killbill.billing.invoice.api.Invoice;
import org.killbill.billing.invoice.api.InvoiceApiException;
import org.killbill.billing.invoice.api.InvoiceItem;
import org.killbill.billing.payment.api.Payment;
import org.killbill.billing.payment.api.PaymentApiException;
import org.killbill.billing.payment.api.PaymentOptions;
import org.killbill.billing.payment.api.PaymentTransaction;
import org.killbill.billing.payment.api.PluginProperty;
import org.killbill.billing.payment.api.TransactionStatus;
import org.killbill.billing.payment.api.TransactionType;
import org.killbill.billing.payment.bus.PaymentBusEventHandler;
import org.killbill.billing.payment.core.janitor.Janitor;
import org.killbill.billing.payment.dao.PaymentAttemptModelDao;
import org.killbill.billing.payment.dao.PaymentTransactionModelDao;
import org.killbill.billing.payment.invoice.InvoicePaymentControlPluginApi;
import org.killbill.billing.payment.plugin.api.PaymentPluginStatus;
import org.killbill.billing.payment.plugin.api.PaymentTransactionInfoPlugin;
import org.killbill.billing.payment.provider.DefaultNoOpPaymentInfoPlugin;
import org.killbill.billing.payment.provider.MockPaymentProviderPlugin;
import org.killbill.billing.platform.api.KillbillConfigSource;
import org.killbill.billing.platform.api.KillbillService.KILLBILL_SERVICES;
import org.killbill.billing.util.api.AuditLevel;
import org.killbill.billing.util.audit.AuditLogWithHistory;
import org.killbill.billing.util.callcontext.InternalCallContextFactory;
import org.killbill.commons.utils.collect.Iterables;
import org.killbill.billing.util.entity.dao.DBRouterUntyped;
import org.killbill.billing.util.entity.dao.DBRouterUntyped.THREAD_STATE;
import org.killbill.bus.api.PersistentBus.EventBusException;
import org.killbill.commons.profiling.Profiling.WithProfilingCallback;
import org.killbill.notificationq.api.NotificationEvent;
import org.killbill.notificationq.api.NotificationEventWithMetadata;
import org.killbill.notificationq.api.NotificationQueueService;
import org.killbill.notificationq.api.NotificationQueueService.NoSuchNotificationQueue;
import org.skife.config.TimeSpan;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

import static java.util.concurrent.TimeUnit.SECONDS;
import static org.awaitility.Awaitility.await;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.fail;

public class TestJanitor extends PaymentTestSuiteWithEmbeddedDB {

    private static final int TIMEOUT = 10;

    final PaymentOptions INVOICE_PAYMENT = new PaymentOptions() {
        @Override
        public boolean isExternalPayment() {
            return false;
        }

        @Override
        public List<String> getPaymentControlPluginNames() {
            return List.of(InvoicePaymentControlPluginApi.PLUGIN_NAME);
        }
    };

    @Inject
    protected TestApiListener testListener;
    @Inject
    protected InternalCallContextFactory internalCallContextFactory;
    @Inject
    protected NotificationQueueService notificationQueueService;
    @Inject
    private PaymentBusEventHandler handler;
    private MockPaymentProviderPlugin mockPaymentProviderPlugin;

    private Account account;

    @Override
    protected KillbillConfigSource getConfigSource(final Map<String, String> extraProperties) {
        final Map<String, String> allExtraProperties = new HashMap<String, String>(extraProperties);
        allExtraProperties.put("org.killbill.payment.provider.default", MockPaymentProviderPlugin.PLUGIN_NAME);
        allExtraProperties.put("killbill.payment.engine.events.off", "false");
        allExtraProperties.put("org.killbill.payment.janitor.rate", "500ms");
        return getConfigSource(null, allExtraProperties);
    }

    @Override
    protected void assertListenerStatus() {
        testListener.assertListenerStatus();
    }

    @BeforeClass(groups = "slow")
    protected void beforeClass() throws Exception {
        if (hasFailed()) {
            return;
        }
        super.beforeClass();
        mockPaymentProviderPlugin = (MockPaymentProviderPlugin) registry.getServiceForName(MockPaymentProviderPlugin.PLUGIN_NAME);
    }

    @BeforeMethod(groups = "slow")
    public void beforeMethod() throws Exception {
        if (hasFailed()) {
            return;
        }

        super.beforeMethod();

        eventBus.register(handler);
        testListener.reset();
        eventBus.register(testListener);
        mockPaymentProviderPlugin.clear();
        account = testHelper.createTestAccount("bobo@gmail.com", true);
    }

    @AfterMethod(groups = "slow")
    public void afterMethod() throws Exception {
        if (hasFailed()) {
            return;
        }

        eventBus.unregister(handler);
        eventBus.unregister(testListener);
        super.afterMethod();
    }

    @Test(groups = "slow")
    public void testCreateSuccessPurchaseWithPaymentControl() throws PaymentApiException, InvoiceApiException, EventBusException {

        final BigDecimal requestedAmount = BigDecimal.TEN;
        final UUID subscriptionId = UUID.randomUUID();
        final UUID bundleId = UUID.randomUUID();
        final LocalDate now = clock.getUTCToday();

        testListener.pushExpectedEvent(NextEvent.INVOICE);
        final Invoice invoice = testHelper.createTestInvoice(account, now, Currency.USD);
        testListener.assertListenerStatus();

        final String paymentExternalKey = invoice.getId().toString();
        final String transactionExternalKey = "wouf wouf";

        invoice.addInvoiceItem(new MockRecurringInvoiceItem(invoice.getId(), account.getId(),
                                                            subscriptionId,
                                                            bundleId,
                                                            "test plan",
                                                            "test phase", null,
                                                            now,
                                                            now.plusMonths(1),
                                                            requestedAmount,
                                                            new BigDecimal("1.0"),
                                                            Currency.USD));

        testListener.pushExpectedEvent(NextEvent.PAYMENT);
        invoicePaymentApi.createPurchaseForInvoicePayment(account,
                                                          invoice.getId(),
                                                          account.getPaymentMethodId(),
                                                          null,
                                                          requestedAmount,
                                                          Currency.USD,
                                                          null,
                                                          paymentExternalKey,
                                                          transactionExternalKey,
                                                          Collections.emptyList(),
                                                          INVOICE_PAYMENT,
                                                          callContext);
        final Payment payment = paymentApi.getPaymentByExternalKey(paymentExternalKey, false, false, Collections.emptyList(), callContext);
        testListener.assertListenerStatus();
        assertEquals(payment.getTransactions().size(), 1);
        assertEquals(payment.getTransactions().get(0).getTransactionStatus(), TransactionStatus.SUCCESS);
        assertEquals(payment.getTransactions().get(0).getTransactionType(), TransactionType.PURCHASE);

        final List<PaymentAttemptModelDao> attempts = paymentDao.getPaymentAttempts(paymentExternalKey, internalCallContext);
        assertEquals(attempts.size(), 1);

        final PaymentAttemptModelDao attempt = attempts.get(0);
        assertEquals(attempt.getStateName(), "SUCCESS");
        assertEquals(attempt.getAmount().compareTo(requestedAmount), 0);
        assertEquals(attempt.getCurrency(), Currency.USD);

        // Ok now the fun part starts... we modify the attempt state to be 'INIT' and wait the the Janitor to do its job.
        paymentDao.updatePaymentAttemptWithProperties(attempt.getId(),
                                                      attempt.getPaymentMethodId(),
                                                      attempt.getTransactionId(),
                                                      "INIT",
                                                      null,
                                                      null,
                                                      null,
                                                      internalCallContext);
        final PaymentAttemptModelDao attempt2 = paymentDao.getPaymentAttempt(attempt.getId(), internalCallContext);
        assertEquals(attempt2.getStateName(), "INIT");
        assertNull(attempt2.getAmount());
        assertNull(attempt2.getCurrency());

        clock.addDays(1);
        Awaitility.await()
                  .atMost(5, TimeUnit.SECONDS)
                  .until(new Callable<Boolean>() {
                      @Override
                      public Boolean call() throws Exception {
                          final PaymentAttemptModelDao attempt3 = paymentDao.getPaymentAttempt(attempt.getId(), internalCallContext);
                          return "SUCCESS".equals(attempt3.getStateName());
                      }
                  });

        final PaymentAttemptModelDao attempt3 = paymentDao.getPaymentAttempt(attempt.getId(), internalCallContext);
        assertEquals(attempt3.getStateName(), "SUCCESS");
        assertEquals(attempt3.getAmount().compareTo(requestedAmount), 0);
        assertEquals(attempt3.getCurrency(), Currency.USD);
    }

    @Test(groups = "slow")
    public void testCreateSuccessRefundPaymentControlWithItemAdjustments() throws Exception {

        final BigDecimal requestedAmount = BigDecimal.TEN;
        final UUID subscriptionId = UUID.randomUUID();
        final UUID bundleId = UUID.randomUUID();
        final LocalDate now = clock.getUTCToday();

        testListener.pushExpectedEvent(NextEvent.INVOICE);
        final Invoice invoice = testHelper.createTestInvoice(account, now, Currency.USD);
        testListener.assertListenerStatus();

        final String paymentExternalKey = invoice.getId().toString();
        final String transactionExternalKey = "craboom";
        final String transactionExternalKey2 = "qwerty";

        final InvoiceItem invoiceItem = new MockRecurringInvoiceItem(invoice.getId(), account.getId(),
                                                                     subscriptionId,
                                                                     bundleId,
                                                                     "test plan", "test phase", null,
                                                                     now,
                                                                     now.plusMonths(1),
                                                                     requestedAmount,
                                                                     new BigDecimal("1.0"),
                                                                     Currency.USD);
        invoice.addInvoiceItem(invoiceItem);

        testListener.pushExpectedEvent(NextEvent.PAYMENT);
        invoicePaymentApi.createPurchaseForInvoicePayment(account,
                                                          invoice.getId(),
                                                          account.getPaymentMethodId(),
                                                          null,
                                                          requestedAmount,
                                                          Currency.USD,
                                                          null,
                                                          paymentExternalKey,
                                                          transactionExternalKey,
                                                          Collections.emptyList(),
                                                          INVOICE_PAYMENT,
                                                          callContext);
        final Payment payment = paymentApi.getPaymentByExternalKey(paymentExternalKey, false, false, Collections.emptyList(), callContext);
        testListener.assertListenerStatus();

        final List<PluginProperty> refundProperties = new ArrayList<PluginProperty>();
        final HashMap<UUID, BigDecimal> uuidBigDecimalHashMap = new HashMap<UUID, BigDecimal>();
        uuidBigDecimalHashMap.put(invoiceItem.getId(), new BigDecimal("1.0"));

        testListener.pushExpectedEvent(NextEvent.PAYMENT);
        invoicePaymentApi.createRefundForInvoicePayment(false, uuidBigDecimalHashMap, account, payment.getId(), null, Currency.USD, null, transactionExternalKey2,
                                                        refundProperties, INVOICE_PAYMENT, callContext);
        final Payment payment2 = paymentApi.getPayment(payment.getId(), false, false, refundProperties, callContext);
        testListener.assertListenerStatus();

        assertEquals(payment2.getTransactions().size(), 2);
        PaymentTransaction refundTransaction = payment2.getTransactions().get(1);
        assertEquals(refundTransaction.getTransactionType(), TransactionType.REFUND);

        final List<PaymentAttemptModelDao> attempts = paymentDao.getPaymentAttempts(paymentExternalKey, internalCallContext);
        assertEquals(attempts.size(), 2);

        final PaymentAttemptModelDao refundAttempt = attempts.get(1);
        assertEquals(refundAttempt.getTransactionType(), TransactionType.REFUND);
        assertEquals(refundAttempt.getAmount().compareTo(BigDecimal.ONE), 0);
        assertEquals(refundAttempt.getCurrency(), Currency.USD);

        // Ok now the fun part starts... we modify the attempt state to be 'INIT' and wait the the Janitor to do its job.
        paymentDao.updatePaymentAttemptWithProperties(refundAttempt.getId(),
                                                      refundAttempt.getPaymentMethodId(),
                                                      refundAttempt.getTransactionId(),
                                                      "INIT",
                                                      null,
                                                      null,
                                                      null,
                                                      internalCallContext);
        final PaymentAttemptModelDao attempt2 = paymentDao.getPaymentAttempt(refundAttempt.getId(), internalCallContext);
        assertEquals(attempt2.getStateName(), "INIT");
        assertNull(attempt2.getAmount());
        assertNull(attempt2.getCurrency());

        clock.addDays(1);

        await().atMost(TIMEOUT, TimeUnit.SECONDS).until(new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                final PaymentAttemptModelDao attempt3 = paymentDao.getPaymentAttempt(refundAttempt.getId(), internalCallContext);
                return "SUCCESS".equals(attempt3.getStateName());
            }
        });

        final PaymentAttemptModelDao refundAttempt2 = paymentDao.getPaymentAttempt(refundAttempt.getId(), internalCallContext);
        assertEquals(refundAttempt2.getStateName(), "SUCCESS");
        assertEquals(refundAttempt2.getAmount().compareTo(BigDecimal.ONE), 0);
        assertEquals(refundAttempt2.getCurrency(), Currency.USD);
    }

    @Test(groups = "slow")
    public void testUnknownEntries() throws PaymentApiException, InvoiceApiException, EventBusException {

        final BigDecimal requestedAmount = BigDecimal.TEN;
        final String paymentExternalKey = "qwru";
        final String transactionExternalKey = "lkjdsf";

        testListener.pushExpectedEvent(NextEvent.PAYMENT);
        final Payment payment = paymentApi.createAuthorization(account, account.getPaymentMethodId(), null, requestedAmount, account.getCurrency(), null, paymentExternalKey,
                                                               transactionExternalKey, Collections.emptyList(), callContext);
        testListener.assertListenerStatus();

        // Artificially move the transaction status to UNKNOWN
        final String paymentStateName = paymentSMHelper.getErroredStateForTransaction(TransactionType.AUTHORIZE).toString();
        testListener.pushExpectedEvent(NextEvent.PAYMENT_PLUGIN_ERROR);
        paymentDao.updatePaymentAndTransactionOnCompletion(account.getId(), null, payment.getId(), TransactionType.AUTHORIZE, paymentStateName, paymentStateName,
                                                           payment.getTransactions().get(0).getId(), TransactionStatus.UNKNOWN, requestedAmount, account.getCurrency(),
                                                           "foo", "bar", true, internalCallContext);
        testListener.assertListenerStatus();

        // Move clock for notification to be processed
        testListener.pushExpectedEvent(NextEvent.PAYMENT);
        clock.addDeltaFromReality(5 * 60 * 1000);
        assertNotificationsCompleted(internalCallContext, 5);
        testListener.assertListenerStatus();

        final Payment updatedPayment = paymentApi.getPayment(payment.getId(), false, false, Collections.emptyList(), callContext);
        assertEquals(updatedPayment.getTransactions().get(0).getTransactionStatus(), TransactionStatus.SUCCESS);
    }

    // Flaky, see https://github.com/killbill/killbill/issues/860
    @Test(groups = "slow", retryAnalyzer = FlakyRetryAnalyzer.class)
    public void testUnknownEntriesWithFailures() throws PaymentApiException, EventBusException {

        final BigDecimal requestedAmount = BigDecimal.TEN;
        final String paymentExternalKey = "minus";
        final String transactionExternalKey = "plus";

        // Make sure the state as seen by the plugin will be in PaymentPluginStatus.ERROR, which will be returned later to Janitor
        mockPaymentProviderPlugin.makeNextPaymentFailWithError();
        testListener.pushExpectedEvent(NextEvent.PAYMENT_ERROR);
        final Payment payment = paymentApi.createAuthorization(account, account.getPaymentMethodId(), null, requestedAmount, account.getCurrency(), null, paymentExternalKey,
                                                               transactionExternalKey, Collections.emptyList(), callContext);
        testListener.assertListenerStatus();

        // Artificially move the transaction status to UNKNOWN
        final String paymentStateName = paymentSMHelper.getErroredStateForTransaction(TransactionType.AUTHORIZE).toString();
        testListener.pushExpectedEvent(NextEvent.PAYMENT_PLUGIN_ERROR);
        paymentDao.updatePaymentAndTransactionOnCompletion(account.getId(), null, payment.getId(), TransactionType.AUTHORIZE, paymentStateName, paymentStateName,
                                                           payment.getTransactions().get(0).getId(), TransactionStatus.UNKNOWN, requestedAmount, account.getCurrency(),
                                                           "foo", "bar", true, internalCallContext);
        testListener.assertListenerStatus();

        final List<AuditLogWithHistory> paymentTransactionHistoryBeforeJanitor = paymentDao.getPaymentTransactionAuditLogsWithHistoryForId(payment.getTransactions().get(0).getId(), AuditLevel.FULL, internalCallContext);
        Assert.assertEquals(paymentTransactionHistoryBeforeJanitor.size(), 3);

        // Move clock for notification to be processed
        testListener.pushExpectedEvent(NextEvent.PAYMENT_ERROR);
        clock.addDeltaFromReality(5 * 60 * 1000);
        assertNotificationsCompleted(internalCallContext, 5);
        testListener.assertListenerStatus();

        // Proves the Janitor ran (and updated the transaction)
        final List<AuditLogWithHistory> paymentTransactionHistoryAfterJanitor = paymentDao.getPaymentTransactionAuditLogsWithHistoryForId(payment.getTransactions().get(0).getId(), AuditLevel.FULL, internalCallContext);
        Assert.assertEquals(paymentTransactionHistoryAfterJanitor.size(), 4);

        PaymentTransactionModelDao history3 = (PaymentTransactionModelDao) paymentTransactionHistoryAfterJanitor.get(3).getEntity();
        Assert.assertEquals(history3.getTransactionStatus(), TransactionStatus.PAYMENT_FAILURE);

        final Payment updatedPayment = paymentApi.getPayment(payment.getId(), false, false, Collections.emptyList(), callContext);
        // Janitor should have moved us to PAYMENT_FAILURE
        assertEquals(updatedPayment.getTransactions().get(0).getTransactionStatus(), TransactionStatus.PAYMENT_FAILURE);
    }

    @Test(groups = "slow")
    public void testUnknownEntriesWithExceptions() throws PaymentApiException, EventBusException {

        final BigDecimal requestedAmount = BigDecimal.TEN;
        final String paymentExternalKey = "minus";
        final String transactionExternalKey = "plus";

        // Make sure the state as seen by the plugin will be in PaymentPluginStatus.ERROR, which will be returned later to Janitor
        mockPaymentProviderPlugin.makeNextPaymentFailWithException();
        try {
            testListener.pushExpectedEvent(NextEvent.PAYMENT_PLUGIN_ERROR);
            paymentApi.createAuthorization(account, account.getPaymentMethodId(), null, requestedAmount, account.getCurrency(), null, paymentExternalKey,
                                           transactionExternalKey, Collections.emptyList(), callContext);
        } catch (PaymentApiException ignore) {
            testListener.assertListenerStatus();
        }
        final Payment payment = paymentApi.getPaymentByExternalKey(paymentExternalKey, false, false, Collections.emptyList(), callContext);

        // Artificially move the transaction status to UNKNOWN
        final String paymentStateName = paymentSMHelper.getErroredStateForTransaction(TransactionType.AUTHORIZE).toString();
        testListener.pushExpectedEvent(NextEvent.PAYMENT_PLUGIN_ERROR);
        paymentDao.updatePaymentAndTransactionOnCompletion(account.getId(), null, payment.getId(), TransactionType.AUTHORIZE, paymentStateName, paymentStateName,
                                                           payment.getTransactions().get(0).getId(), TransactionStatus.UNKNOWN, requestedAmount, account.getCurrency(),
                                                           "foo", "bar", true, internalCallContext);
        testListener.assertListenerStatus();

        // Move clock for notification to be processed
        clock.addDeltaFromReality(5 * 60 * 1000);
        // NO because we will keep retrying as we can't fix it...
        //assertNotificationsCompleted(internalCallContext, 5);

        final List<AuditLogWithHistory> paymentTransactionHistoryBeforeJanitor = paymentDao.getPaymentTransactionAuditLogsWithHistoryForId(payment.getTransactions().get(0).getId(), AuditLevel.FULL, internalCallContext);
        Assert.assertEquals(paymentTransactionHistoryBeforeJanitor.size(), 3);

        // Nothing new happened
        final List<AuditLogWithHistory> paymentTransactionHistoryAfterJanitor = paymentDao.getPaymentTransactionAuditLogsWithHistoryForId(payment.getTransactions().get(0).getId(), AuditLevel.FULL, internalCallContext);
        Assert.assertEquals(paymentTransactionHistoryAfterJanitor.size(), 3);
    }

    @Test(groups = "slow")
    public void testPendingEntries() throws PaymentApiException, EventBusException, NoSuchNotificationQueue {

        final BigDecimal requestedAmount = BigDecimal.TEN;
        final String paymentExternalKey = "jhj44";
        final String transactionExternalKey = "4jhjj2";

        testListener.pushExpectedEvent(NextEvent.PAYMENT);
        final Payment payment = paymentApi.createAuthorization(account, account.getPaymentMethodId(), null, requestedAmount, account.getCurrency(), null, paymentExternalKey,
                                                               transactionExternalKey, Collections.emptyList(), callContext);
        testListener.assertListenerStatus();

        // Artificially move the transaction status to PENDING
        final String paymentStateName = paymentSMHelper.getPendingStateForTransaction(TransactionType.AUTHORIZE).toString();
        testListener.pushExpectedEvent(NextEvent.PAYMENT);
        paymentDao.updatePaymentAndTransactionOnCompletion(account.getId(), null, payment.getId(), TransactionType.AUTHORIZE, paymentStateName, paymentStateName,
                                                           payment.getTransactions().get(0).getId(), TransactionStatus.PENDING, requestedAmount, account.getCurrency(),
                                                           "loup", "chat", true, internalCallContext);
        testListener.assertListenerStatus();

        // Move clock for notification to be processed ((default config is set for one hour)
        testListener.pushExpectedEvent(NextEvent.PAYMENT);
        clock.addDeltaFromReality(1000 * (3600 + 1));

        assertNotificationsCompleted(internalCallContext, 5);
        testListener.assertListenerStatus();
        final Payment updatedPayment = paymentApi.getPayment(payment.getId(), false, false, Collections.emptyList(), callContext);
        Assert.assertEquals(updatedPayment.getTransactions().get(0).getTransactionStatus(), TransactionStatus.SUCCESS);
    }

    // The test will check that when a PENDING entry stays PENDING, we go through all our retries and eventually give up (no infinite loop of retries)
    // Flaky, see https://github.com/killbill/killbill/issues/860
    @Test(groups = "slow", retryAnalyzer = FlakyRetryAnalyzer.class)
    public void testPendingEntriesThatDontMove() throws Exception {

        final BigDecimal requestedAmount = BigDecimal.TEN;
        final String paymentExternalKey = "haha";
        final String transactionExternalKey = "hoho!";

        testListener.pushExpectedEvent(NextEvent.PAYMENT);
        final Payment payment = paymentApi.createAuthorization(account, account.getPaymentMethodId(), null, requestedAmount, account.getCurrency(), null, paymentExternalKey,
                                                               transactionExternalKey, Collections.emptyList(), callContext);
        testListener.assertListenerStatus();

        // Artificially move the transaction status to PENDING AND update state on the plugin as well
        final List<PaymentTransactionInfoPlugin> paymentTransactions = mockPaymentProviderPlugin.getPaymentInfo(account.getId(), payment.getId(), Collections.emptyList(), callContext);
        final PaymentTransactionInfoPlugin oTx = paymentTransactions.remove(0);
        final PaymentTransactionInfoPlugin updatePaymentTransaction = new DefaultNoOpPaymentInfoPlugin(oTx.getKbPaymentId(), oTx.getKbTransactionPaymentId(), oTx.getTransactionType(), oTx.getAmount(), oTx.getCurrency(), oTx.getCreatedDate(), oTx.getCreatedDate(), PaymentPluginStatus.PENDING, null, null);
        paymentTransactions.add(updatePaymentTransaction);
        mockPaymentProviderPlugin.updatePaymentTransactions(payment.getId(), paymentTransactions);

        final String paymentStateName = paymentSMHelper.getPendingStateForTransaction(TransactionType.AUTHORIZE).toString();

        testListener.pushExpectedEvent(NextEvent.PAYMENT);
        paymentDao.updatePaymentAndTransactionOnCompletion(account.getId(), null, payment.getId(), TransactionType.AUTHORIZE, paymentStateName, paymentStateName,
                                                           payment.getTransactions().get(0).getId(), TransactionStatus.PENDING, requestedAmount, account.getCurrency(),
                                                           "loup", "chat", true, internalCallContext);
        testListener.assertListenerStatus();

        // 1h, 1d
        for (final TimeSpan cur : paymentConfig.getPendingTransactionsRetries(internalCallContext)) {
            // Verify there is a notification to retry updating the value
            assertEquals(getPendingNotificationCnt(internalCallContext), 1);

            clock.addDeltaFromReality(cur.getMillis() + 1000);

            assertNotificationsCompleted(internalCallContext, 5);
            // We add a sleep here to make sure the notification gets processed. Note that calling assertNotificationsCompleted alone would not work
            // because there is a point in time where the notification queue is empty (showing notification was processed), but the processing of the notification
            // will itself enter a new notification, and so the synchronization is difficult without writing *too much code*.
            Thread.sleep(1500);
            assertNotificationsCompleted(internalCallContext, 5);

            final Payment updatedPayment = paymentApi.getPayment(payment.getId(), false, false, Collections.emptyList(), callContext);
            Assert.assertEquals(updatedPayment.getTransactions().get(0).getTransactionStatus(), TransactionStatus.PENDING);
        }

        await().atMost(TIMEOUT, TimeUnit.SECONDS).until(new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                return getPendingNotificationCnt(internalCallContext) == 0;
            }
        });
    }

    @Test(groups = "slow")
    public void testDBRouterThreadState() throws Throwable {
        final Payment payment = (Payment) DBRouterUntyped.withRODBIAllowed(true,
                                                                           new WithProfilingCallback<Object, Throwable>() {
                                                                               @Override
                                                                               public Payment execute() throws Throwable {
                                                                                   // Shouldn't happen in practice, but it's just to verify the behavior
                                                                                   assertEquals(DBRouterUntyped.getCurrentState(), THREAD_STATE.RO_ALLOWED);

                                                                                   final BigDecimal requestedAmount = BigDecimal.TEN;
                                                                                   testListener.pushExpectedEvent(NextEvent.PAYMENT);
                                                                                   final Payment payment = paymentApi.createAuthorization(account, account.getPaymentMethodId(), null, requestedAmount, account.getCurrency(), null, UUID.randomUUID().toString(),
                                                                                                                                          UUID.randomUUID().toString(), Collections.emptyList(), callContext);
                                                                                   testListener.assertListenerStatus();

                                                                                   // Thread switch, RW by default
                                                                                   assertEquals(mockPaymentProviderPlugin.getLastThreadState(), THREAD_STATE.RW_ONLY);
                                                                                   // Switched to RW, because of RW DAO call
                                                                                   assertEquals(DBRouterUntyped.getCurrentState(), THREAD_STATE.RW_ONLY);
                                                                                   return payment;
                                                                               }
                                                                           });

        DBRouterUntyped.withRODBIAllowed(true,
                                         new WithProfilingCallback<Object, Throwable>() {
                                             @Override
                                             public Object execute() throws Throwable {
                                                 assertEquals(DBRouterUntyped.getCurrentState(), THREAD_STATE.RO_ALLOWED);

                                                 final Payment retrievedPayment2 = paymentApi.getPayment(payment.getId(), true, false, Collections.emptyList(), callContext);
                                                 Assert.assertEquals(retrievedPayment2.getTransactions().get(0).getTransactionStatus(), TransactionStatus.SUCCESS);

                                                 // No thread switch, RO as well
                                                 assertEquals(mockPaymentProviderPlugin.getLastThreadState(), THREAD_STATE.RO_ALLOWED);
                                                 assertEquals(DBRouterUntyped.getCurrentState(), THREAD_STATE.RO_ALLOWED);
                                                 return null;
                                             }
                                         });
    }

    private void assertNotificationsCompleted(final InternalCallContext internalCallContext, final long timeoutSec) {
        try {
            await().atMost(timeoutSec, SECONDS).until(new Callable<Boolean>() {
                @Override
                public Boolean call() throws Exception {
                    boolean completed = true;
                    final Iterator<NotificationEventWithMetadata<NotificationEvent>> iterator = notificationQueueService.getNotificationQueue(KILLBILL_SERVICES.PAYMENT_SERVICE.getServiceName(), Janitor.QUEUE_NAME).getFutureOrInProcessingNotificationForSearchKeys(internalCallContext.getAccountRecordId(), internalCallContext.getTenantRecordId()).iterator();
                    try {
                        while (iterator.hasNext()) {
                            final NotificationEventWithMetadata<NotificationEvent> notificationEvent = iterator.next();
                            if (!notificationEvent.getEffectiveDate().isAfter(clock.getUTCNow())) {
                                completed = false;
                            }
                        }
                    } finally {
                        // Go through all results to close the connection
                        while (iterator.hasNext()) {
                            iterator.next();
                        }
                    }
                    return completed;
                }
            });
        } catch (final Exception e) {
            fail("Test failed ", e);
        }
    }

    private int getPendingNotificationCnt(final InternalCallContext internalCallContext) {
        try {
            return Iterables.size(notificationQueueService.getNotificationQueue(KILLBILL_SERVICES.PAYMENT_SERVICE.getServiceName(), Janitor.QUEUE_NAME).getFutureOrInProcessingNotificationForSearchKeys(internalCallContext.getAccountRecordId(), internalCallContext.getTenantRecordId()));
        } catch (final Exception e) {
            fail("Test failed ", e);
        }
        // Not reached..
        return -1;
    }
}

