package gw.rating.rtm.query

uses com.google.common.base.Ticker
uses com.google.common.cache.CacheBuilder
uses com.google.common.cache.CacheLoader
uses com.google.common.cache.LoadingCache
uses com.google.common.cache.RemovalListener
uses com.guidewire.pl.system.cache.CacheChangeBatch
uses com.guidewire.pl.system.cache.GlobalCacheListener
uses gw.api.plugin.StartablePluginBase
uses gw.api.rating.cache.IRateTableRowCache
uses gw.api.startable.StartablePluginCallbackHandler
uses gw.api.startable.StartablePluginState
uses gw.api.system.PCLoggerCategory
uses gw.api.system.PLDependenciesGateway
uses gw.pl.persistence.core.Key
uses gw.rating.flow.util.QueryUtils
uses gw.rating.rtm.domain.OrderedPersistenceAdapter

uses java.util.concurrent.TimeUnit
uses java.util.stream.Stream

/**
 * RateTableRow cache is required to be implemented as a global cache to account for cross cluster changes to rate
 * tables.
 * <p>
 * Note though that this is a cache implementation implemented as a plugin. This was done because
 * GlobalCacheListeners are typically implemented as java methods and then initialized in the server init
 * (@See PCInitTab.java), also written in java. However, RateTableRowCache is dependent on a lot of gosu code which made
 * conversion to java more work then we were willing to take on at the moment. Plugins are intialized after Global Cache
 * in the initialization life cycle and so we could achieve the same effect.
 * <p>
 * Plugin must be marked @Distributed to ensure it is not created with an expirable lease. @See StartableServiceImp.startPluginsAtRunlevel()
 */
@ReadOnly
@Distributed
class RateTableRowCache extends StartablePluginBase implements IRateTableRowCache, GlobalCacheListener {


  /**
   * static config
   * <p>
   * There's a tuning issue here: set this too low, and all rate tables will be flushed, say,
   * over a weekend when nothing much happens. Set it too high, and no tables will ever be evicted.
   */
  public static var CACHE_TIMEOUT : long = 36
  public static var CACHE_TIMEOUT_UNIT : TimeUnit = TimeUnit.HOURS
  private static var DEFAULT_CONCURRENCY = 16

  construct() {
    this(DEFAULT_CONCURRENCY, \notification -> {
      PCLoggerCategory.RTM.debug("Removed ${notification}, reason ${notification.Cause}")
    }, null)
  }

  /**
   * Allow some callback extensibility for testing
   *
   * @param removalListener
   * @param ticker
   */
  protected construct(concurrency : Integer, removalListener : RemovalListener, ticker : Ticker) {
    try {
      PCLoggerCategory.RTM.info("Creating RTM row cache")
      PCLoggerCategory.RTM.info("Cache timeout (in nanoseconds): " + CACHE_TIMEOUT_UNIT.toNanos(CACHE_TIMEOUT))

      var builder = CacheBuilder.newBuilder().expireAfterAccess(CACHE_TIMEOUT, CACHE_TIMEOUT_UNIT)
      if (concurrency != null) {
        builder.concurrencyLevel(concurrency)
      }

      if (removalListener != null) {
        builder.removalListener(removalListener)
      }

      if (ticker != null) {
        builder.ticker(ticker)
      }

      _cache = builder.build<Key, CacheEntry>(_cacheLoader)
    } catch (e) {
      PCLoggerCategory.RTM.error("Exception occurred in constructing rate table row cache", e)
      throw e
    }
  }

  /**
   * Cache is implemented using Guava LoadingCache. Cache.get() calls getOrLoad backed by a default _cacheLoader.
   * Cache is lazily loaded when someone requests for a specific table. Changes to rate tables are handled by the
   * GlobalCacheListener method cacheChanged()
   */
  private final var _cache : LoadingCache<Key, CacheEntry>

  /**
   * restricted access to loadingcache to allow for some extensibility
   */
  protected property get LoadingCache() : LoadingCache<Key, CacheEntry> {
    return _cache
  }

  /**
   * lazily called by loadingcache getOrLoad()
   */
  private final var _cacheLoader : CacheLoader<Key, CacheEntry> = new CacheLoader<Key, CacheEntry>() {
    override function load(key : Key) : CacheEntry {
      return new CacheEntry(key)
    }
  }

  internal static class CacheEntry {
    // map from NumberOfActiveMatchOps to applicable rows
    final private var _rows : Map<Integer, List<OrderedPersistenceAdapter>>as readonly Rows
    final private var _date : java.util.Date as readonly Date
    final private var _rowsNormal : Boolean as readonly Normalized

    construct(tableID : Key) {
      var table = QueryUtils.loadReadOnlyRateTableByKey(tableID)
      PCLoggerCategory.RTM.info("Rate table cache miss: owning table ${table.TableOwningFactors.DisplayName}, ID: ${table.ID}")
      // NOTE: Could extend here to allow for various persistence mechanisms
      var loadedRows = new DatabaseRowLoader().loadRows(table)
      var cachedPartitionedRowsMap : Map<Integer, List<OrderedPersistenceAdapter>>
      var rowsAreNormal : boolean
      var ops = AbstractStatelessQuery.getMatchOperators(table.Definition)

      try {
        // Only normalize if it's not already normalized and we're willing to spend memory for perf
        // (This also means we will not attempt to normalize if we know the table is TC_INTRACTABLE
        var normalizeRows = table.RowUniformityStatus == TC_NONUNIFORM and !table.ReduceMemoryUsage

        cachedPartitionedRowsMap = new StatelessRowLoadUtil()
            .getPartitionedOrderedRowsMap(ops, loadedRows, normalizeRows)

        rowsAreNormal = normalizeRows or table.RowUniformityStatus == TC_UNIFORM
      } catch (pae : StatelessRowLoadUtil.PartitionAbortedException) {
        cachedPartitionedRowsMap = new StatelessRowLoadUtil().getPartitionedOrderedRowsMap(ops, loadedRows, false)
        rowsAreNormal = false
      }

      // make sure that get(0) returns something
      if (not cachedPartitionedRowsMap.containsKey(0)) {
        cachedPartitionedRowsMap.put(0, {})
      }

      _rows = Collections.unmodifiableMap(cachedPartitionedRowsMap)
      _date = table.TableOwningFactors.UpdateTime
      _rowsNormal = rowsAreNormal
      PCLoggerCategory.RTM.info("Rate table cache populated: owning table ${table.TableOwningFactors.DisplayName}, ID: ${table.ID}")
    }
  }

  /**
   * @throws IllegalStateException if the requested table was updated without triggering the global cache listener. If used, all queries should go through this cache
   */
  override function getCachedRows(table : RateTable) : CacheEntry {
    var entry = _cache.get(table.TableOwningFactors.ID)

    if (entry._date.before(table.TableOwningFactors.UpdateTime)) {
      throw new java.lang.IllegalStateException("Cache entry for ${table.DisplayName} in book ${table.RateBook.DisplayName} is out of date and should have been evicted.")
    }
    return entry
  }

  /**
   * Global cache listener methods
   */

  override function cacheChanged(batch : CacheChangeBatch) {
    // Search for global changes that are present in the rate table cache and evict them
    if (batch.containsChangesFor(RateTable)) {
      Stream.of(batch.getAllChangesFor(RateTable))
          .filter(\elt -> _cache.asMap().containsKey(elt.VersionKey.Id))
          .forEach(\elt -> {
            PCLoggerCategory.RTM.debug("GlobalCache.cacheChanged() invoked. Evicting cache entry for ${elt.VersionKey.Id}")
            _cache.invalidate(elt.VersionKey.Id);
          })
    }
  }

  override function flushCaches() {
    invalidateAll()
  }

  override function invalidateAll() {
    PCLoggerCategory.RTM.debug("RateTableRowCache.invalidateAll() invoked. Evicting cache entry for all rate tables")
    _cache.invalidateAll()
  }

  override function resetStatistics() {
    // statistics are not tracked for this cache
  }

  /**
   * IStartablePlugin methods
   */

  override function start(pluginCallbackHandler : StartablePluginCallbackHandler, serverStarting : boolean) {
    super.start(pluginCallbackHandler, serverStarting)
    PLDependenciesGateway.GlobalCache.addListener(this)
  }

  override function stop(serverShuttingDown : boolean) {
    super.stop(serverShuttingDown)
    PLDependenciesGateway.GlobalCache.removeListener(this)
    invalidateAll()
  }

  override property get State() : StartablePluginState {
    return PLDependenciesGateway.GlobalCache.isListener(this) ? StartablePluginState.Started : Stopped
  }
}