odc-analyzer/app/services/VulnerabilityNotificationService.scala

package services

import _root_.org.joda.time.DateTime
import com.google.inject.Inject
import com.mohiva.play.silhouette.api.LoginInfo
import com.ysoft.odc.SetDiff
import controllers.{ProjectsWithReports, ReportInfo}
import models._
import play.api.db.slick.{DatabaseConfigProvider, HasDatabaseConfigProvider}
import slick.jdbc.TransactionIsolation

import scala.concurrent.{ExecutionContext, Future}

final class SingleFutureExecutionThrottler() (implicit executionContext: ExecutionContext){
  private var nextFuture: Future[_] = Future.successful(null)

  def throttle[T](f: => Future[T]): Future[T] = synchronized{
    val newFuture = nextFuture.recover{ case _ => null}.flatMap(_ => f)
    nextFuture = newFuture
    newFuture
  }

}

final class NoThrottler() (implicit executionContext: ExecutionContext){
  def throttle[T](f: => Future[T]): Future[T] = f
}

class VulnerabilityNotificationService @Inject() (protected val dbConfigProvider: DatabaseConfigProvider)(implicit executionContext: ExecutionContext) extends HasDatabaseConfigProvider[models.profile.type]{
  import dbConfig.driver.api._
  import models.tables._

  def watchedProjectsByUser(identity: LoginInfo) = db.run(vulnerabilitySubscriptions.filter(_.user === identity).result)
  def subscribe(user: LoginInfo, project: String) = db.run(vulnerabilitySubscriptions += VulnerabilitySubscription(user = user, project = project))

  def unsubscribe(user: LoginInfo, project: String) = db.run(vulnerabilitySubscriptions.filter(vs => vs.user === user && vs.project === project).delete)

  def getRecipientsForProjects(projects: Set[ReportInfo]) = {
    val bareProjects = projects.map(_.bare)
    val expandedProjects = projects ++ bareProjects
    val relevantFullIds = expandedProjects.map(_.fullId)
    db.run(vulnerabilitySubscriptions.filter(_.project inSet relevantFullIds).map(_.user()).result)
  }

  class ExportPlatform[T, U] private[VulnerabilityNotificationService] (ept: ExportPlatformTables[T, U]) {
    def changeProjects(ticketId: Int, diff: SetDiff[String], projects: ProjectsWithReports) = db.run(
      DBIO.seq(
        ept.projects.filter(_.exportedVulnerabilityId === ticketId).delete,
        ept.projects ++= diff.newSet.map(fullId => ExportedVulnerabilityProject(ticketId, fullId)).toSet
      ).transactionally
    )

    def projectsForTickets(ticketsIds: Set[Int]): Future[Map[Int, Set[String]]] = db.run(
      ept.projects.filter(_.exportedVulnerabilityId inSet ticketsIds).result
    ).map{_.groupBy(_.exportedVulnerabilityId).mapValues(_.map(_.projectFullId).toSet).map(identity).withDefaultValue(Set())}

    def ticketsForVulnerabilities(vulnerabilities: Traversable[String]) = db.run(
      ept.tickets.filter(_.vulnerabilityName inSet vulnerabilities).result
    ).map(_.map{ rec =>
        rec._2.vulnerabilityName -> rec
    }.toMap)

    def ticketForVulnerability(vulnerabilityName: String) = db.run(
      ept.tickets.filter(_.vulnerabilityName === vulnerabilityName).map(_.base).result
    ).map(_.headOption)


    def addTicket(vulnerabilityTicket: ExportedVulnerability[T], projects: Set[ReportInfo]): Future[Any] = db.run(
      (
        ept.tickets.map(_.base).returning(ept.tickets.map(_.id)) += vulnerabilityTicket
      ).flatMap( id =>
        ept.projects ++= projects.map(ri => ExportedVulnerabilityProject(id, ri.fullId)).toSet
      ).transactionally
    )

  }

  /**
    * The changelogThrottler is a temporary hack than prevents some congestion that seems to occur in HikariCP or maybe in Slick.
    * It prevents exceptions like “java.sql.SQLException: Timeout of 1001ms encountered waiting for connection”.
    * It is probably prevented at a wrong level, but it works :)
    */
  private val changelogThrottler = new SingleFutureExecutionThrottler()
  // private val changelogThrottler = new NoThrottler()

  def changeAffectedProjects(vulnerabilityName: String, affectedProjectsDiff: SetDiff[String]): Future[Unit] = {
    val time = DateTime.now()
    def createRecord(projectName: String, direction: Change.Direction) = Change(time, vulnerabilityName, projectName, direction)
    val recordsToAdd = affectedProjectsDiff.added.map(projectName => createRecord(projectName, Change.Direction.Added)) ++
      affectedProjectsDiff.removed.map(projectName => createRecord(projectName, Change.Direction.Removed))
    /*
    Transaction:
    It is essential to ensure that records in changelog appear in order. Low level of isolation might be even worse than running outside of transaction.
    In longer term, it should be wrapped with transaction *with a proper isolation level* together with the export status modification.
    */
    changelogThrottler.throttle(db.run(
      (changelog.map(_.base) ++= recordsToAdd).withTransactionIsolation(TransactionIsolation.Serializable)
    ).map(_ => ()))
  }

  val issueTrackerExport = new ExportPlatform(tables.issueTrackerExportTables)

  val mailExport = new ExportPlatform(tables.mailExportTables)

  val diffDbExport = new ExportPlatform(tables.diffDbExportTables)

}