<commit_before>/**
 * Copyright (C) 2010-2012 LShift Ltd.
 *
 * 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.lshift.diffa.kernel.differencing

import net.lshift.diffa.kernel.events.VersionID
import reflect.BeanProperty
import org.hibernate.SessionFactory
import net.lshift.diffa.kernel.util.db.{Cursor, HibernateQueryUtils, DatabaseFacade}
import net.lshift.diffa.kernel.util.db.SessionHelper._
import org.hibernate.Session
import net.sf.ehcache.CacheManager
import scala.collection.JavaConversions._
import org.joda.time.{DateTimeZone, DateTime, Interval}
import org.jadira.usertype.dateandtime.joda.columnmapper.TimestampColumnDateTimeMapper
import net.lshift.diffa.kernel.config.DomainScopedKey._
import net.lshift.diffa.kernel.config.Domain._
import net.lshift.diffa.kernel.hooks.HookManager
import net.lshift.diffa.kernel.config.{DomainScopedKey, Domain, DiffaPairRef, DiffaPair}
import org.hibernate.dialect.{Oracle10gDialect, Dialect}
import org.hibernate.criterion.{Projections, Restrictions}
import net.lshift.diffa.kernel.util.cache.CacheProvider
import java.io.Serializable

/**
 * Hibernate backed Domain Cache provider.
 */
class HibernateDomainDifferenceStore(val sessionFactory:SessionFactory,
                                     db:DatabaseFacade,
                                     cacheProvider:CacheProvider,
                                     val cacheManager:CacheManager,
                                     val dialect:Dialect,
                                     val hookManager:HookManager)
    extends DomainDifferenceStore
    with HibernateQueryUtils {

  val dateTimeMapper = new TimestampColumnDateTimeMapper

  val aggregationCache = new DifferenceAggregationCache(this, cacheManager)
  val hook = hookManager.createDifferencePartitioningHook(sessionFactory)

  val columnMapper = new TimestampColumnDateTimeMapper()

  val pendingEvents = cacheProvider.getCachedMap[VersionID, PendingDifferenceEvent]("pending.difference.events")

  /**
   * This is a heuristic that allows the cache to get prefilled if the agent is booted and
   * there were persistent pending diffs. The motivation is to reduce cache misses in subsequent calls.
   */
  val prefetchLimit = 1000 // TODO This should be a tuning parameter
  prefetchPendingEvents(prefetchLimit)


  def reset {
    pendingEvents.evictAll()
    aggregationCache.clear()
  }

  def removeDomain(domain:String) = {
    // If difference partitioning is enabled, ask the hook to clean up each pair. Note that we'll end up running a
    // delete over all pair differences later anyway, so we won't record the result of the removal operation.
    if (hook.isDifferencePartitioningEnabled) {
      listPairsInDomain(domain).foreach(p => hook.removeAllPairDifferences(domain, p.key))
    }

    removeDomainDifferences(domain)
    preenPendingEventsCache("objId.pair.domain", domain)
  }

  def removePair(pair: DiffaPairRef) = {
    val hookHelped = hook.removeAllPairDifferences(pair.domain, pair.key)

    sessionFactory.withSession { s =>
      if (!hookHelped) {
        executeUpdate(s, "removeDiffsByPairAndDomain", Map("pairKey" -> pair.key, "domain" -> pair.domain))
      }
      executeUpdate(s, "removePendingDiffsByPairAndDomain", Map("pairKey" -> pair.key, "domain" -> pair.domain))
      removeLatestRecordedVersion(pair)
    }

    preenPendingEventsCache("objId.pair.key", pair.key)
  }
  
  def currentSequenceId(domain:String) =
    db.singleQueryMaybe[java.lang.Integer]("maxSeqIdByDomain", Map("domain" -> domain)).getOrElse(0).toString

  def maxSequenceId(pair: DiffaPairRef, start:DateTime, end:DateTime) = sessionFactory.withSession(s => {
    val c = s.createCriteria(classOf[ReportedDifferenceEvent])
    c.add(Restrictions.eq("objId.pair.domain", pair.domain))
    c.add(Restrictions.eq("objId.pair.key", pair.key))
    if (start != null) c.add(Restrictions.ge("detectedAt", start))
    if (end != null) c.add(Restrictions.lt("detectedAt", end))
    c.setProjection(Projections.max("seqId"))

    val count:Option[java.lang.Integer] = Option(c.uniqueResult().asInstanceOf[java.lang.Integer])
    count.getOrElse(new java.lang.Integer(0)).intValue
  })

  def addPendingUnmatchedEvent(id: VersionID, lastUpdate: DateTime, upstreamVsn: String, downstreamVsn: String, seen: DateTime) = {

    val pending = getPendingEvent(id)

    if (pending.exists()) {
      updatePendingEvent(pending, upstreamVsn, downstreamVsn, seen)
    }
    else {
      val pendingUnmatched = PendingDifferenceEvent(null, id, lastUpdate, upstreamVsn, downstreamVsn, seen)
      createPendingEvent(pendingUnmatched)
    }
  }

  def addReportableUnmatchedEvent(id: VersionID, lastUpdate: DateTime, upstreamVsn: String, downstreamVsn: String, seen: DateTime) =
    addReportableMismatch(ReportedDifferenceEvent(null, id, lastUpdate, false, upstreamVsn, downstreamVsn, seen))


  def upgradePendingUnmatchedEvent(id: VersionID) = {

    val pending = getPendingEvent(id)

    if (pending.exists()) {
      // Remove the pending and report a mismatch
      removePendingEvent(pending)
      addReportableMismatch(pending.convertToUnmatched)
    }
    else {
      // No pending difference, nothing to do
      null
    }

  }

  def cancelPendingUnmatchedEvent(id: VersionID, vsn: String) = {
    val pending = getPendingEvent(id)

    if (pending.exists()) {
      if (pending.upstreamVsn == vsn || pending.downstreamVsn == vsn) {
        removePendingEvent(pending)
        true
      } else {
        false
      }
    }
    else {
      false
    }

  }

  def addMatchedEvent(id: VersionID, vsn: String) = {
    sessionFactory.withSession(s => {

      // Remove any pending events with the given id
      val pending = getPendingEvent(id)

      if (pending.exists()) {
        removePendingEvent(pending)
      }

      // Find any existing events we've got for this ID
      getEventById(id) match {
          // No unmatched event. Nothing to do.
        case None           => null
        case Some(existing) =>
          existing.state match {
            case MatchState.MATCHED => // Ignore. We've already got an event for what we want.
              existing.asDifferenceEvent
            case MatchState.UNMATCHED | MatchState.IGNORED =>
              // A difference has gone away. Remove the difference, and add in a match
              deletePreviouslyReportedEvent(existing)

              val previousDetectionTime = existing.detectedAt
              saveAndConvertEvent(ReportedDifferenceEvent(null, id, new DateTime, true, vsn, vsn, new DateTime), previousDetectionTime)
          }
      }
    })
  }

  /**
   * Convert all unmatched DifferenceEvents with a lastSeen time earlier
   * than the cut-off to matched DifferenceEvents (matched: false -> true).
   */
  def matchEventsOlderThan(pair:DiffaPairRef, cutoff: DateTime) = {

    /**
     * Convert an unmatched difference event to a matched difference event
     * and set its lastSeen and detectedAt values to the current time.
     * Converting from unmatched to matched means setting its matched value to 'true'.
     */
    def convertOldEvent(s:Session, old:ReportedDifferenceEvent) {
      s.delete(old)
      val lastSeen = new DateTime
      val detectedAt = lastSeen
      val matched = true // 'convert' to matched
      saveAndConvertEvent(ReportedDifferenceEvent(null, old.objId, detectedAt, matched, old.upstreamVsn, old.upstreamVsn, lastSeen) )
    }

     val params = Map("domain" -> pair.domain,
                      "pair"   -> pair.key,
                      "cutoff" -> cutoff)<commit_msg>Didn't actually delete the session reference in last commit<commit_after>/**
 * Copyright (C) 2010-2012 LShift Ltd.
 *
 * 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.lshift.diffa.kernel.differencing

import net.lshift.diffa.kernel.events.VersionID
import reflect.BeanProperty
import org.hibernate.SessionFactory
import net.lshift.diffa.kernel.util.db.{Cursor, HibernateQueryUtils, DatabaseFacade}
import net.lshift.diffa.kernel.util.db.SessionHelper._
import org.hibernate.Session
import net.sf.ehcache.CacheManager
import scala.collection.JavaConversions._
import org.joda.time.{DateTimeZone, DateTime, Interval}
import org.jadira.usertype.dateandtime.joda.columnmapper.TimestampColumnDateTimeMapper
import net.lshift.diffa.kernel.config.DomainScopedKey._
import net.lshift.diffa.kernel.config.Domain._
import net.lshift.diffa.kernel.hooks.HookManager
import net.lshift.diffa.kernel.config.{DomainScopedKey, Domain, DiffaPairRef, DiffaPair}
import org.hibernate.dialect.{Oracle10gDialect, Dialect}
import org.hibernate.criterion.{Projections, Restrictions}
import net.lshift.diffa.kernel.util.cache.CacheProvider
import java.io.Serializable

/**
 * Hibernate backed Domain Cache provider.
 */
class HibernateDomainDifferenceStore(val sessionFactory:SessionFactory,
                                     db:DatabaseFacade,
                                     cacheProvider:CacheProvider,
                                     val cacheManager:CacheManager,
                                     val dialect:Dialect,
                                     val hookManager:HookManager)
    extends DomainDifferenceStore
    with HibernateQueryUtils {

  val dateTimeMapper = new TimestampColumnDateTimeMapper

  val aggregationCache = new DifferenceAggregationCache(this, cacheManager)
  val hook = hookManager.createDifferencePartitioningHook(sessionFactory)

  val columnMapper = new TimestampColumnDateTimeMapper()

  val pendingEvents = cacheProvider.getCachedMap[VersionID, PendingDifferenceEvent]("pending.difference.events")

  /**
   * This is a heuristic that allows the cache to get prefilled if the agent is booted and
   * there were persistent pending diffs. The motivation is to reduce cache misses in subsequent calls.
   */
  val prefetchLimit = 1000 // TODO This should be a tuning parameter
  prefetchPendingEvents(prefetchLimit)


  def reset {
    pendingEvents.evictAll()
    aggregationCache.clear()
  }

  def removeDomain(domain:String) = {
    // If difference partitioning is enabled, ask the hook to clean up each pair. Note that we'll end up running a
    // delete over all pair differences later anyway, so we won't record the result of the removal operation.
    if (hook.isDifferencePartitioningEnabled) {
      listPairsInDomain(domain).foreach(p => hook.removeAllPairDifferences(domain, p.key))
    }

    removeDomainDifferences(domain)
    preenPendingEventsCache("objId.pair.domain", domain)
  }

  def removePair(pair: DiffaPairRef) = {
    val hookHelped = hook.removeAllPairDifferences(pair.domain, pair.key)

    sessionFactory.withSession { s =>
      if (!hookHelped) {
        executeUpdate(s, "removeDiffsByPairAndDomain", Map("pairKey" -> pair.key, "domain" -> pair.domain))
      }
      executeUpdate(s, "removePendingDiffsByPairAndDomain", Map("pairKey" -> pair.key, "domain" -> pair.domain))
      removeLatestRecordedVersion(pair)
    }

    preenPendingEventsCache("objId.pair.key", pair.key)
  }
  
  def currentSequenceId(domain:String) =
    db.singleQueryMaybe[java.lang.Integer]("maxSeqIdByDomain", Map("domain" -> domain)).getOrElse(0).toString

  def maxSequenceId(pair: DiffaPairRef, start:DateTime, end:DateTime) = sessionFactory.withSession(s => {
    val c = s.createCriteria(classOf[ReportedDifferenceEvent])
    c.add(Restrictions.eq("objId.pair.domain", pair.domain))
    c.add(Restrictions.eq("objId.pair.key", pair.key))
    if (start != null) c.add(Restrictions.ge("detectedAt", start))
    if (end != null) c.add(Restrictions.lt("detectedAt", end))
    c.setProjection(Projections.max("seqId"))

    val count:Option[java.lang.Integer] = Option(c.uniqueResult().asInstanceOf[java.lang.Integer])
    count.getOrElse(new java.lang.Integer(0)).intValue
  })

  def addPendingUnmatchedEvent(id: VersionID, lastUpdate: DateTime, upstreamVsn: String, downstreamVsn: String, seen: DateTime) = {

    val pending = getPendingEvent(id)

    if (pending.exists()) {
      updatePendingEvent(pending, upstreamVsn, downstreamVsn, seen)
    }
    else {
      val pendingUnmatched = PendingDifferenceEvent(null, id, lastUpdate, upstreamVsn, downstreamVsn, seen)
      createPendingEvent(pendingUnmatched)
    }
  }

  def addReportableUnmatchedEvent(id: VersionID, lastUpdate: DateTime, upstreamVsn: String, downstreamVsn: String, seen: DateTime) =
    addReportableMismatch(ReportedDifferenceEvent(null, id, lastUpdate, false, upstreamVsn, downstreamVsn, seen))


  def upgradePendingUnmatchedEvent(id: VersionID) = {

    val pending = getPendingEvent(id)

    if (pending.exists()) {
      // Remove the pending and report a mismatch
      removePendingEvent(pending)
      addReportableMismatch(pending.convertToUnmatched)
    }
    else {
      // No pending difference, nothing to do
      null
    }

  }

  def cancelPendingUnmatchedEvent(id: VersionID, vsn: String) = {
    val pending = getPendingEvent(id)

    if (pending.exists()) {
      if (pending.upstreamVsn == vsn || pending.downstreamVsn == vsn) {
        removePendingEvent(pending)
        true
      } else {
        false
      }
    }
    else {
      false
    }

  }

  def addMatchedEvent(id: VersionID, vsn: String) = {

    // Remove any pending events with the given id
    val pending = getPendingEvent(id)

    if (pending.exists()) {
      removePendingEvent(pending)
    }

    // Find any existing events we've got for this ID
    getEventById(id) match {
        // No unmatched event. Nothing to do.
      case None           => null
      case Some(existing) =>
        existing.state match {
          case MatchState.MATCHED => // Ignore. We've already got an event for what we want.
            existing.asDifferenceEvent
          case MatchState.UNMATCHED | MatchState.IGNORED =>
            // A difference has gone away. Remove the difference, and add in a match
            deletePreviouslyReportedEvent(existing)

            val previousDetectionTime = existing.detectedAt
            saveAndConvertEvent(ReportedDifferenceEvent(null, id, new DateTime, true, vsn, vsn, new DateTime), previousDetectionTime)
        }
    }

  }

  /**
   * Convert all unmatched DifferenceEvents with a lastSeen time earlier
   * than the cut-off to matched DifferenceEvents (matched: false -> true).
   */
  def matchEventsOlderThan(pair:DiffaPairRef, cutoff: DateTime) = {

    /**
     * Convert an unmatched difference event to a matched difference event
     * and set its lastSeen and detectedAt values to the current time.
     * Converting from unmatched to matched means setting its matched value to 'true'.
     */
    def convertOldEvent(s:Session, old:ReportedDifferenceEvent) {
      s.delete(old)
      val lastSeen = new DateTime
      val detectedAt = lastSeen
      val matched = true // 'convert' to matched
      saveAndConvertEvent(ReportedDifferenceEvent(null, old.objId, detectedAt, matched, old.upstreamVsn, old.upstreamVsn, lastSeen) )
    }

     val params = Map("domain" -> pair.domain,
                      "pair"   -> pair.key,
                      "cutoff" -> cutoff)
{'fraction_non_alphanumeric': 0.09158415841584158, 'fraction_numerical': 0.0024752475247524753, 'mean_word_length': 3.317659631311782, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 36, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3146066', 'n_tokens_mistral': 4697, 'n_tokens_neox': 4522, 'n_words': 1192}