/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF 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.apache.geode;

import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.geode.cache.ExpirationAction.DESTROY;
import static org.apache.geode.cache.ExpirationAction.INVALIDATE;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;

import java.util.Properties;
import java.util.function.Consumer;

import junitparams.Parameters;
import junitparams.naming.TestCaseName;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;

import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.CacheTransactionManager;
import org.apache.geode.cache.CommitConflictException;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.ExpirationAttributes;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionEvent;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.internal.cache.ExpiryTask;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalRegion;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.TXManagerImpl;
import org.apache.geode.internal.cache.TXStateProxy;
import org.apache.geode.test.junit.runners.GeodeParamsRunner;

/**
 * Tests transaction expiration functionality
 *
 * @since GemFire 4.0
 */
@RunWith(GeodeParamsRunner.class)
public class TXExpirationIntegrationTest {

  private static final String REGION_NAME =
      TXExpirationIntegrationTest.class.getSimpleName() + "_region";
  private static final String KEY = "key";

  private InternalCache cache;
  private CacheTransactionManager transactionManager;

  @Before
  public void setUp() throws Exception {
    cache = (InternalCache) new CacheFactory(getConfig()).create();
    transactionManager = cache.getCacheTransactionManager();
  }

  protected Properties getConfig() {
    Properties config = new Properties();
    config.setProperty(LOCATORS, "");
    config.setProperty(MCAST_PORT, "0");
    return config;
  }

  @After
  public void tearDown() throws Exception {
    try {
      transactionManager.rollback();
    } catch (IllegalStateException ignore) {
    }
    cache.close();

    ExpiryTask.permitExpiration();
  }

  @Test
  @Parameters({"ENTRY_IDLE_DESTROY", "ENTRY_IDLE_INVALIDATE", "ENTRY_TTL_DESTROY",
      "ENTRY_TTL_INVALIDATE"})
  @TestCaseName("{method}({params})")
  public void entryExpirationDoesNotCauseConflict(ExpirationOperation operation) throws Exception {
    InternalRegion region = createRegion(REGION_NAME);
    AttributesMutator<String, String> mutator = region.getAttributesMutator();

    KeyCacheListener keyCacheListener = spy(new KeyCacheListener());
    mutator.addCacheListener(keyCacheListener);

    ExpiryTask.suspendExpiration();
    operation.mutate(mutator);

    region.put(KEY, "value1");

    transactionManager.begin();
    region.put(KEY, "value2");
    awaitEntryExpiration(region, KEY); // enables expiration
    assertThat(region.getEntry(KEY).getValue()).isEqualTo("value2");

    ExpiryTask.suspendExpiration();
    transactionManager.commit();

    assertThat(region.getEntry(KEY).getValue()).isEqualTo("value2");
    awaitEntryExpiration(region, KEY); // enables expiration
    operation.verifyInvoked(keyCacheListener);

    operation.verifyState(region);
  }

  @Test
  @Parameters({"ENTRY_IDLE_DESTROY", "ENTRY_IDLE_INVALIDATE", "ENTRY_TTL_DESTROY",
      "ENTRY_TTL_INVALIDATE"})
  @TestCaseName("{method}({params})")
  public void entryExpirationContinuesAfterCommitConflict(ExpirationOperation operation)
      throws Exception {
    Region<String, String> region = createRegion(REGION_NAME);

    AttributesMutator<String, String> mutator = region.getAttributesMutator();

    KeyCacheListener keyCacheListener = spy(new KeyCacheListener());
    mutator.addCacheListener(keyCacheListener);

    ExpiryTask.suspendExpiration();
    operation.mutate(mutator);

    ExpiryTask.suspendExpiration();
    region.put(KEY, "value1");

    transactionManager.begin();
    region.put(KEY, "value2");
    awaitEntryExpiration(region, KEY); // enables expiration
    assertThat(region.getEntry(KEY).getValue()).isEqualTo("value2");

    ExpiryTask.suspendExpiration();

    TXManagerImpl txManagerImpl = (TXManagerImpl) transactionManager;
    TXStateProxy txStateProxy = txManagerImpl.pauseTransaction();
    region.put(KEY, "conflict");
    txManagerImpl.unpauseTransaction(txStateProxy);

    assertThatThrownBy(() -> transactionManager.commit())
        .isInstanceOf(CommitConflictException.class);

    awaitEntryExpiration(region, KEY); // enables expiration
    operation.verifyInvoked(keyCacheListener);
    operation.verifyState(region);
  }

  @Test
  @Parameters({"ENTRY_IDLE_DESTROY", "ENTRY_IDLE_INVALIDATE", "ENTRY_TTL_DESTROY",
      "ENTRY_TTL_INVALIDATE"})
  @TestCaseName("{method}({params})")
  public void entryExpirationContinuesAfterRollback(ExpirationOperation operation)
      throws Exception {
    Region<String, String> region = createRegion(REGION_NAME);

    AttributesMutator<String, String> mutator = region.getAttributesMutator();

    KeyCacheListener keyCacheListener = spy(new KeyCacheListener());
    mutator.addCacheListener(keyCacheListener);

    ExpiryTask.suspendExpiration();
    operation.mutate(mutator);

    region.put(KEY, "value1");

    transactionManager.begin();
    region.put(KEY, "value2");
    awaitEntryExpiration(region, KEY); // enables expiration
    assertThat(region.getEntry(KEY).getValue()).isEqualTo("value2");

    ExpiryTask.suspendExpiration();

    transactionManager.rollback();

    awaitEntryExpiration(region, KEY); // enables expiration
    operation.verifyInvoked(keyCacheListener);
    operation.verifyState(region);
  }

  @Test
  @Parameters({"REGION_IDLE_DESTROY", "REGION_IDLE_INVALIDATE", "REGION_TTL_DESTROY",
      "REGION_TTL_INVALIDATE"})
  @TestCaseName("{method}({params})")
  public void regionExpirationDoesNotCauseConflict(ExpirationOperation operation) throws Exception {
    Region<String, String> region = createRegion(REGION_NAME);

    KeyCacheListener keyCacheListener = spy(new KeyCacheListener());

    AttributesMutator<String, String> mutator = region.getAttributesMutator();
    mutator.addCacheListener(keyCacheListener);

    ExpiryTask.suspendExpiration();
    operation.mutate(mutator);

    transactionManager.begin();
    region.put(KEY, "value1");

    awaitRegionExpiration(region); // enables expiration
    assertThat(region.getEntry(KEY).getValue()).isEqualTo("value1");

    ExpiryTask.suspendExpiration();
    transactionManager.commit();

    assertThat(region.getEntry(KEY).getValue()).isEqualTo("value1");

    awaitRegionExpiration(region); // enables expiration
    operation.verifyInvoked(keyCacheListener);
    operation.verifyState(region);
  }

  private enum ExpirationOperation {
    ENTRY_IDLE_DESTROY(m -> m.setEntryIdleTimeout(new ExpirationAttributes(1, DESTROY)),
        s -> verify(s, atLeast(1)).afterDestroyKey(eq(KEY)),
        r -> assertThat(r.containsKey(KEY)).isFalse()),
    ENTRY_IDLE_INVALIDATE(m -> m.setEntryIdleTimeout(new ExpirationAttributes(1, INVALIDATE)),
        s -> verify(s, atLeast(1)).afterInvalidateKey(eq(KEY)),
        r -> assertThat(r.get(KEY)).isNull()),
    ENTRY_TTL_DESTROY(m -> m.setEntryTimeToLive(new ExpirationAttributes(1, DESTROY)),
        s -> verify(s, atLeast(1)).afterDestroyKey(eq(KEY)),
        r -> assertThat(r.containsKey(KEY)).isFalse()),
    ENTRY_TTL_INVALIDATE(m -> m.setEntryTimeToLive(new ExpirationAttributes(1, INVALIDATE)),
        s -> verify(s, atLeast(1)).afterInvalidateKey(eq(KEY)),
        r -> assertThat(r.get(KEY)).isNull()),
    REGION_IDLE_DESTROY(m -> m.setRegionIdleTimeout(new ExpirationAttributes(1, DESTROY)),
        s -> verify(s, atLeast(1)).afterRegionDestroyName(eq(REGION_NAME)),
        r -> await().until(r::isDestroyed)),
    REGION_IDLE_INVALIDATE(m -> m.setRegionIdleTimeout(new ExpirationAttributes(1, INVALIDATE)),
        s -> verify(s, atLeast(1)).afterRegionInvalidateName(eq(REGION_NAME)),
        r -> await().until(() -> r.get(KEY) == null)),
    REGION_TTL_DESTROY(m -> m.setRegionTimeToLive(new ExpirationAttributes(1, DESTROY)),
        s -> verify(s, atLeast(1)).afterRegionDestroyName(eq(REGION_NAME)),
        r -> await().until(r::isDestroyed)),
    REGION_TTL_INVALIDATE(m -> m.setRegionTimeToLive(new ExpirationAttributes(1, INVALIDATE)),
        s -> verify(s, atLeast(1)).afterRegionInvalidateName(eq(REGION_NAME)),
        r -> await().until(() -> r.get(KEY) == null));

    private final Consumer<AttributesMutator> mutatorConsumer;
    private final Consumer<KeyCacheListener> listenerConsumer;
    private final Consumer<Region> regionConsumer;

    ExpirationOperation(Consumer<AttributesMutator> mutatorConsumer,
        Consumer<KeyCacheListener> listenerConsumer, Consumer<Region> regionConsumer) {
      this.mutatorConsumer = mutatorConsumer;
      this.listenerConsumer = listenerConsumer;
      this.regionConsumer = regionConsumer;
    }

    void mutate(AttributesMutator mutator) {
      mutatorConsumer.accept(mutator);
    }

    void verifyInvoked(KeyCacheListener keyCacheListener) {
      listenerConsumer.accept(keyCacheListener);
    }

    public void verifyState(Region region) {
      regionConsumer.accept(region);
    }
  }

  private InternalRegion createRegion(String name) {
    RegionFactory<String, String> rf = cache.createRegionFactory(RegionShortcut.REPLICATE);
    rf.setStatisticsEnabled(true);

    System.setProperty(LocalRegion.EXPIRY_MS_PROPERTY, "true");
    try {
      return (InternalRegion) rf.create(name);
    } finally {
      System.clearProperty(LocalRegion.EXPIRY_MS_PROPERTY);
    }
  }

  private void awaitEntryExpired(Region region, String key) {
    await().until(() -> region.getEntry(key) == null || region.get(key) == null);
  }

  private void awaitEntryExpiration(Region region, String key) {
    InternalRegion internalRegion = (InternalRegion) region;
    try {
      ExpirationDetector detector;
      do {
        detector = new ExpirationDetector(internalRegion.getEntryExpiryTask(key));
        ExpiryTask.expiryTaskListener = detector;
        ExpiryTask.permitExpiration();
        detector.awaitExecuted(30, SECONDS);
      } while (!detector.hasExpired() && detector.wasRescheduled());
    } finally {
      ExpiryTask.expiryTaskListener = null;
    }
  }

  private void awaitRegionExpiration(Region region) {
    InternalRegion internalRegion = (InternalRegion) region;
    try {
      ExpirationDetector detector;
      do {
        detector = new ExpirationDetector(internalRegion.getRegionTTLExpiryTask() != null
            ? internalRegion.getRegionTTLExpiryTask() : internalRegion.getRegionIdleExpiryTask());
        ExpiryTask.expiryTaskListener = detector;
        ExpiryTask.permitExpiration();
        detector.awaitExecuted(30, SECONDS);
      } while (!detector.hasExpired() && detector.wasRescheduled());
    } finally {
      ExpiryTask.expiryTaskListener = null;
    }
  }

  private static class KeyCacheListener extends CacheListenerAdapter<String, String> {

    @Override
    public void afterCreate(EntryEvent<String, String> event) {
      afterCreateKey(event.getKey());
    }

    @Override
    public void afterUpdate(EntryEvent<String, String> event) {
      afterUpdateKey(event.getKey());
    }

    @Override
    public void afterInvalidate(EntryEvent<String, String> event) {
      afterInvalidateKey(event.getKey());
    }

    @Override
    public void afterDestroy(EntryEvent<String, String> event) {
      afterDestroyKey(event.getKey());
    }

    @Override
    public void afterRegionInvalidate(RegionEvent<String, String> event) {
      afterRegionInvalidateName(event.getRegion().getName());
    }

    @Override
    public void afterRegionDestroy(RegionEvent<String, String> event) {
      afterRegionDestroyName(event.getRegion().getName());
    }

    public void afterCreateKey(Object key) {
      // nothing
    }

    public void afterUpdateKey(Object key) {
      // nothing
    }

    public void afterInvalidateKey(Object key) {
      // nothing
    }

    public void afterDestroyKey(Object key) {
      // nothing
    }

    public void afterRegionInvalidateName(Object key) {
      // nothing
    }

    public void afterRegionDestroyName(Object key) {
      // nothing
    }
  }
}
