/*
 * 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.tenant.api;

// import java.util.Collection;
// import java.util.List;
// import java.util.concurrent.ScheduledExecutorService;
// import java.util.concurrent.TimeUnit;
// import java.util.concurrent.atomic.AtomicLong;

// import javax.inject.Inject;
// import javax.inject.Named;

// import org.killbill.billing.callcontext.InternalTenantContext;
// import org.killbill.billing.events.BusInternalEvent;
// import org.killbill.billing.tenant.api.TenantInternalApi.CacheInvalidationCallback;
// import org.killbill.billing.tenant.api.TenantKV.TenantKey;
// import org.killbill.billing.tenant.api.user.DefaultTenantConfigChangeInternalEvent;
// import org.killbill.billing.tenant.api.user.DefaultTenantConfigDeletionInternalEvent;
// import org.killbill.billing.tenant.dao.TenantBroadcastDao;
// import org.killbill.billing.tenant.dao.TenantBroadcastModelDao;
// import org.killbill.billing.tenant.dao.TenantDao;
// import org.killbill.billing.tenant.dao.TenantKVModelDao;
// import org.killbill.billing.tenant.glue.DefaultTenantModule;
// import org.killbill.billing.util.config.definition.TenantConfig;
// import org.killbill.bus.api.PersistentBus;
// import org.killbill.bus.api.PersistentBus.EventBusException;
// import org.killbill.commons.concurrent.Executors;
// import org.slf4j.Logger;
// import org.slf4j.LoggerFactory;

// import com.google.common.base.Predicate;
// import com.google.common.collect.HashMultimap;
// import com.google.common.collect.ImmutableList;
// import com.google.common.collect.Iterables;
// import com.google.common.collect.Multimap;

// /**
//  * This class manages the callbacks that have been registered when per tenant objects have been inserted into the
//  * tenant_kvs store; the flow is the following (for e.g catalog):
//  * 1. CatalogUserApi is invoked to retrieve per tenant catalog
//  * 2. If cache is empty, TenantCacheLoader is invoked and uses TenantInternalApi is load the data; at that time, the invalidation callback
//  * is registered
//  * <p/>
//  * When this class initializes, it reads the current entry in the tenant_broadcasts table and from then on, keeps polling for new entries; when new
//  * entries are found, it invokes the callback to invalidate the current caching and force the TenantCacheLoader to be invoked again.
//  */
// public class TenantCacheInvalidation {

//     private final static int TERMINATION_TIMEOUT_SEC = 5;

//     private static final Logger logger = LoggerFactory.getLogger(TenantCacheInvalidation.class);

//     private final Multimap<TenantKey, CacheInvalidationCallback> cache;
//     private final TenantBroadcastDao broadcastDao;
//     private final TenantConfig tenantConfig;
//     private final PersistentBus eventBus;
//     private final TenantDao tenantDao;
//     private AtomicLong latestRecordIdProcessed;
//     private volatile boolean isStopped;

//     private ScheduledExecutorService tenantExecutor;

//     @Inject
//     public TenantCacheInvalidation(@Named(DefaultTenantModule.NO_CACHING_TENANT) final TenantBroadcastDao broadcastDao,
//                                    @Named(DefaultTenantModule.NO_CACHING_TENANT) final TenantDao tenantDao,
//                                    final PersistentBus eventBus,
//                                    final TenantConfig tenantConfig) {
//         this.cache = HashMultimap.<TenantKey, CacheInvalidationCallback>create();
//         this.broadcastDao = broadcastDao;
//         this.tenantConfig = tenantConfig;
//         this.tenantDao = tenantDao;
//         this.eventBus = eventBus;
//         this.isStopped = false;
//     }

//     public void initialize() {
//         final TenantBroadcastModelDao entry = broadcastDao.getLatestEntry();
//         this.latestRecordIdProcessed = entry != null ? new AtomicLong(entry.getRecordId()) : new AtomicLong(0L);
//         this.tenantExecutor = Executors.newSingleThreadScheduledExecutor("TenantExecutor");
//         this.isStopped = false;
//     }

//     public void start() {
//         final TimeUnit pendingRateUnit = tenantConfig.getTenantBroadcastServiceRunningRate().getUnit();
//         final long pendingPeriod = tenantConfig.getTenantBroadcastServiceRunningRate().getPeriod();
//         tenantExecutor.scheduleAtFixedRate(new TenantCacheInvalidationRunnable(this, broadcastDao, tenantDao), pendingPeriod, pendingPeriod, pendingRateUnit);

//     }

//     public void stop() {
//         if (isStopped) {
//             logger.warn("TenantExecutor is already in a stopped state");
//             return;
//         }
//         try {
//             tenantExecutor.shutdown();
//             boolean success = tenantExecutor.awaitTermination(TERMINATION_TIMEOUT_SEC, TimeUnit.SECONDS);
//             if (!success) {
//                 logger.warn("TenantExecutor failed to complete termination within " + TERMINATION_TIMEOUT_SEC + "sec");
//             }
//         } catch (InterruptedException e) {
//             Thread.currentThread().interrupt();
//             logger.warn("TenantExecutor stop sequence got interrupted");
//         } finally {
//             isStopped = true;
//         }
//     }

//     public void registerCallback(final TenantKey key, final CacheInvalidationCallback value) {
//         cache.put(key, value);

//     }

//     public Collection<CacheInvalidationCallback> getCacheInvalidations(final TenantKey key) {
//         return cache.get(key);
//     }

//     public AtomicLong getLatestRecordIdProcessed() {
//         return latestRecordIdProcessed;
//     }

//     public boolean isStopped() {
//         return isStopped;
//     }

//     public void setLatestRecordIdProcessed(final Long newProcessedRecordId) {
//         this.latestRecordIdProcessed.set(newProcessedRecordId);
//     }

//     public PersistentBus getEventBus() {
//         return eventBus;
//     }

//     public static class TenantCacheInvalidationRunnable implements Runnable {

//         private final TenantCacheInvalidation parent;
//         private final TenantBroadcastDao broadcastDao;
//         private final TenantDao tenantDao;

//         public TenantCacheInvalidationRunnable(final TenantCacheInvalidation parent,
//                                                final TenantBroadcastDao broadcastDao,
//                                                final TenantDao tenantDao) {
//             this.parent = parent;
//             this.broadcastDao = broadcastDao;
//             this.tenantDao = tenantDao;
//         }

//         @Override
//         public void run() {
//             if (parent.isStopped) {
//                 return;
//             }

//             final List<TenantBroadcastModelDao> entries = broadcastDao.getLatestEntriesFrom(parent.getLatestRecordIdProcessed().get());
//             for (TenantBroadcastModelDao cur : entries) {
//                 if (parent.isStopped()) {
//                     return;
//                 }

//                 try {
//                     final TenantKeyAndCookie tenantKeyAndCookie = extractTenantKeyAndCookie(cur.getType());
//                     if (tenantKeyAndCookie != null) {
//                         final Collection<CacheInvalidationCallback> callbacks = parent.getCacheInvalidations(tenantKeyAndCookie.getTenantKey());
//                         if (!callbacks.isEmpty()) {
//                             final InternalTenantContext tenantContext = new InternalTenantContext(cur.getTenantRecordId());
//                             // TODO In case of Redis, we don't want any invalidation, but we still want the events to notify the plugins (ideally, our bus would also support a Topic model)
//                             for (final CacheInvalidationCallback callback : callbacks) {
//                                 callback.invalidateCache(tenantKeyAndCookie.getTenantKey(), tenantKeyAndCookie.getCookie(), tenantContext);
//                             }

//                             final Long tenantKvsTargetRecordId = cur.getTargetRecordId();
//                             final BusInternalEvent event;
//                             if (tenantKvsTargetRecordId != null) {
//                                 final TenantKVModelDao tenantModelDao = tenantDao.getKeyByRecordId(tenantKvsTargetRecordId, tenantContext);
//                                 if (tenantModelDao == null) {
//                                     // Probably inactive entry
//                                     continue;
//                                 }
//                                 event = new DefaultTenantConfigChangeInternalEvent(tenantModelDao.getId(), cur.getType(),
//                                                                                    null, tenantContext.getTenantRecordId(), cur.getUserToken());
//                             } else {
//                                 event = new DefaultTenantConfigDeletionInternalEvent(cur.getType(),
//                                                                                      null, tenantContext.getTenantRecordId(), cur.getUserToken());
//                             }
//                             try {
//                                 parent.getEventBus().post(event);
//                             } catch (final EventBusException e) {
//                                 logger.warn("Failed to post event {}", event, e);
//                             }
//                         }
//                     } else {
//                         logger.warn("Failed to find CacheInvalidationCallback for " + cur.getType());
//                     }
//                 } finally {
//                     parent.setLatestRecordIdProcessed(cur.getRecordId());
//                 }
//             }
//         }

//         private TenantKeyAndCookie extractTenantKeyAndCookie(final String key) {
//             final TenantKey tenantKey = Iterables.tryFind(ImmutableList.copyOf(TenantKey.values()), new Predicate<TenantKey>() {
//                 @Override
//                 public boolean apply(final TenantKey input) {
//                     return key.startsWith(input.toString());
//                 }
//             }).orNull();
//             if (tenantKey == null) {
//                 return null;
//             }

//             final String cookie = !key.equals(tenantKey.toString()) ?
//                                   key.substring(tenantKey.toString().length()) :
//                                   null;
//             return new TenantKeyAndCookie(tenantKey, cookie);
//         }

//     }

//     private static final class TenantKeyAndCookie {

//         private final TenantKey tenantKey;
//         private final Object cookie;

//         public TenantKeyAndCookie(final TenantKey tenantKey, final Object cookie) {
//             this.tenantKey = tenantKey;
//             this.cookie = cookie;
//         }

//         public TenantKey getTenantKey() {
//             return tenantKey;
//         }

//         public Object getCookie() {
//             return cookie;
//         }
//     }
// }
