/**
 * Copyright 2009 Google Inc.
 * 
 * Licensed 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 net.appjet.oui;

import java.util.Date;

import scala.collection.mutable.{HashMap, HashSet, Set, Map, ArrayBuffer};
import scala.util.Sorting;

trait BucketMap extends scala.collection.mutable.Map[int, BucketedLastHits] {
  def t = 1000*60;
  override def apply(s: int) = synchronized { getOrElseUpdate(s, new BucketedLastHits(t)) };
  def counts = { val p = this; new scala.collection.Map.Projection[int, int] {
    def size = p.size;
    def get(s: int) = p.get(s).map(_.count);
    def elements = p.elements.map(o => (o._1, o._2.count));
  }};
}

abstract class BucketKeeper[A, B](val size: Long, val numbuckets: int, val noUpdate: Boolean) {
  def this(size: Long, noUpdate: Boolean) = 
    this(size, Math.max(100, if (noUpdate) 1 else (size/60000).toInt), noUpdate)
  def this(size: Long) = this(size, false);

  val buckets = new Array[A](numbuckets);

  val millisPerBucket = size/numbuckets;
  var lastSwitch = System.currentTimeMillis();
  var currentBucket = 0;
  
  def withSyncUpdate[E](block: E): E = synchronized {
    updateBuckets();
    block;
  }
  
  protected def bucketAtTime(d: Date) = {
    val msAgo = lastSwitch - d.getTime();
    val bucketsAgo = Math.floor(msAgo/millisPerBucket).asInstanceOf[Int];
    if (bucketsAgo < numbuckets) {
      val bucket = (currentBucket - bucketsAgo + numbuckets) % numbuckets
      // println("Applying to old bucket: "+bucket+" / current: "+currentBucket+", old count: "+count);
      Some(bucket);
    } else {
      // println("No bucket found for: "+d);
      None;
    }
  }

  protected def updateBuckets(): Unit = {
    if (! noUpdate) {
      val now = System.currentTimeMillis();
      while (now > lastSwitch + millisPerBucket) {
        lastSwitch += millisPerBucket;
        currentBucket = (currentBucket + 1) % numbuckets;
        bucketClear(currentBucket);
      }      
    }
  }
  
  protected def bucketClear(index: Int);
  protected def bucketsInOrder: Seq[A] = 
    buckets.slice((currentBucket+1)%numbuckets, numbuckets) ++ 
    buckets.slice(0, currentBucket)
  
  def mergeBuckets(b: Seq[A]): B;
  
  def history(bucketsPerSample: Int, numSamples: Int): Array[B] = withSyncUpdate {
    val bseq = bucketsInOrder.reverse.take(bucketsPerSample*numSamples);
    val sampleCount = Math.min(numSamples, bseq.length);
    val samples =
      for (i <- 0 until sampleCount) yield {
        mergeBuckets(bseq.slice(i*bucketsPerSample, (i+1)*bucketsPerSample));
      }
    samples.reverse.toArray;
  }
  def latest(bucketsPerSample: Int): B = history(bucketsPerSample, 1)(0);
  def count: B = withSyncUpdate { mergeBuckets(buckets); }
    
  for (i <- 0 until numbuckets) {
    bucketClear(i);
  }
}

class BucketedUniques(size: Long, noUpdate: Boolean) 
extends BucketKeeper[Set[Any], Int](size, noUpdate) {
  def this(size: Long) = this(size, false);
      
  override protected def bucketClear(index: Int): Unit = {
    buckets(index) = new HashSet[Any];
  }
  
  override def mergeBuckets(b: Seq[Set[Any]]) = {
    b.foldLeft(scala.collection.immutable.Set[Any]())(_ ++ _).size;
  }
  
  def hit(d: Date, value: Any): Unit = withSyncUpdate {
    for (bucket <- bucketAtTime(d)) {
      buckets(bucket) += value;      
    }
  }
}

class BucketedValueCounts(size: Long, noUpdate: Boolean) 
extends BucketKeeper[HashMap[String, Int], (Int, Map[String, Int])](size, noUpdate) {
  def this(size: Long) = this(size, false);
  
  override protected def bucketClear(index: Int): Unit = {
    buckets(index) = new HashMap[String, Int];
  }

  override def mergeBuckets(b: Seq[HashMap[String, Int]]) = {
    val out = new HashMap[String, Int];
    var total = 0;
    for (m <- b) {
      for ((k, v) <- m) {
        out(k) = out.getOrElse(k, 0) + v;
        total += v;
      }
    }
    (total, out);    
  }
    
  def hit(d: Date, value: String, increment: Int): Unit = withSyncUpdate {
    for (bucket <- bucketAtTime(d)) {
      buckets(bucket)(value) = 
        buckets(bucket).getOrElse(value, 0)+increment;      
    }
  }
  
  def hit(d: Date, value: String): Unit = hit(d, value, 1);
}
    

/**
 * Keeps track of how many "hits" in the last size milliseconds.
 * Has granularity speicified by numbuckets.
 */
class BucketedLastHits(size: Long, noUpdate: Boolean) 
extends BucketKeeper[Int, Int](size, noUpdate) {
  def this(size: Long) = this(size, false);
      
  override protected def bucketClear(index: int): Unit = {
    buckets(index) = 0;
  }

  override def mergeBuckets(b: Seq[Int]) = {
    b.foldRight(0)(_+_);
  }

  def hit(d: Date): Unit = hit(d, 1);
  def hit(d: Date, n: Int): Unit = withSyncUpdate {
    for (bucket <- bucketAtTime(d)) {
      buckets(bucket) = buckets(bucket) + n;
    }
  }
}

class BucketedLastHitsHistogram(size: Long, noUpdate: Boolean)
extends BucketKeeper[ArrayBuffer[Int], Function1[Float, Int]](size, noUpdate) {
  def this(size: Long) = this(size, false);
  
  override protected def bucketClear(index: Int): Unit = {
    buckets(index) = new ArrayBuffer[Int];
  }

  // elements will end up sorted.
  protected def histogramFunction(elements: Array[Int]): Function1[Float, Int] = {
    Sorting.quickSort(elements);
    (percentile: Float) => {
      if (elements.length == 0) {
        0
      } else {
        elements(
          Math.round(percentile/100.0f*(elements.length-1)));
      }
    }    
  }
  
  override def mergeBuckets(b: Seq[ArrayBuffer[Int]]) = {
    val elements = new Array[Int](b.foldRight(0)(_.size + _));
    var currentIndex = 0;
    for (bucket <- b if bucket.length > 0) {
      // copyToArray is broken through scala 2.7.5, fixed in trunk.
      // bucket.copyToArray(allElements, currentIndex);
      val bucketArray = bucket.toArray;
      System.arraycopy(bucketArray, 0, elements, currentIndex, bucketArray.length);
      currentIndex += bucket.size
    }
    histogramFunction(elements);
  }
    
  def hit(d: Date): Unit = hit(d, 1);
  def hit(d: Date, n: Int): Unit = withSyncUpdate {
    for (bucket <- bucketAtTime(d)) {
      buckets(bucket) += n;
    }
  }
}

object appstats {
  val minutelyStatus = new HashMap[int, BucketedLastHits] with BucketMap;
  val hourlyStatus = new HashMap[int, BucketedLastHits] with BucketMap { override val t = 1000*60*60 };
  val dailyStatus = new HashMap[int, BucketedLastHits] with BucketMap { override val t = 1000*60*60*24 };
  val weeklyStatus = new HashMap[int, BucketedLastHits] with BucketMap { override val t = 1000*60*60*24*7 };
  val stati = Array(minutelyStatus, hourlyStatus, dailyStatus, weeklyStatus);
}