<commit_before>package jobs

import javax.inject.{Inject, Singleton}

import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global
import akka.actor.ActorSystem
import common.Config
import controllers.Croissants
import models.Croissant
import modules.mail.Mail
import org.joda.time.DateTime
import play.api.libs.json.{Json, Reads}
import play.api.libs.ws.WSClient
import play.modules.reactivemongo.ReactiveMongoApi

import scala.concurrent.Future
import scala.util.{Failure, Success}

case class AccessToken(
  value: String,
  refreshAt: DateTime
)

@Singleton
class GmailJob @Inject()(
  val ws: WSClient,
  val system: ActorSystem
)(implicit
  val config: Config,
  val reactiveMongoApi: ReactiveMongoApi,
  val mailer: Mail) {

  val logger = play.api.Logger("GmailJob")

  val refreshtoken = config.Gmail.refreshtoken
  val clientId = config.Gmail.clientId
  val clientSecret = config.Gmail.clientSecret

  val fromRegex = "(.*)<(.*)>".r

  def schedule(accessToken: Option[AccessToken]): Unit = {
    system.scheduler.scheduleOnce(5 seconds) {
      val now = DateTime.now

      val accessTokenFtr = if (accessToken.isEmpty || now.isAfter(accessToken.get.refreshAt)) {
        refreshAccessToken()
      } else Future.successful(accessToken.get)

      val result = for {
        accessToken <- accessTokenFtr
        messages <- listUnreadGmailMessages(accessToken)
      } yield (accessToken, messages)

      result.onComplete {
        case Success((accessToken, messages)) =>
          messages.foreach { message =>
            for {
              from <- message.payload.headersMap.get("From")
              (name, email) <- from match {
                case fromRegex(name, email) => Some((name.trim, email.trim))
              }
              dkim <- message.payload.headersMap.get("X-Google-DKIM-Signature")
            } yield {
              val subject = message.payload.headersMap.get("Subject").map(_.trim)
              Croissant.addCroissant(email, subject)
            }
          }
          schedule(Some(accessToken))
        case Failure(e) =>
          logger.error("Error during gmail job: ", e)
          schedule(accessToken)
      }
    }
    ()
  }

  case class GmailMessage(
    id: String,
    payload: Payload
  )
  case class Payload(
    headers: Seq[Header]
  ) {
    lazy val headersMap = headers.map { header =>
      (header.name -> header.value)
    }.toMap
  }
  case class Header(
    name: String,
    value: String
  )
  implicit val headerReads: Reads[Header] = Json.reads[Header]
  implicit val payloadReads: Reads[Payload] = Json.reads[Payload]
  implicit val gmailMessagesReads: Reads[GmailMessage] = Json.reads[GmailMessage]
  private def listUnreadGmailMessages(accessToken: AccessToken): Future[Seq[GmailMessage]] = {
    val gmailMessagesBaseUrl = "https://www.googleapis.com/gmail/v1/users/me/messages"
    ws.url(gmailMessagesBaseUrl)
      .withQueryString("q" -> "is:unread")
      .withHeaders(
        "Authorization" -> s"Bearer ${accessToken.value}"
      )
      .get()
      .map {
        case response if response.status == 200 =>
          (response.json \ "messages" \\ "id").map(_.as[String])
        case response => throw new Exception(s"Could not get new messages ids, error: ${response.status} ${response.statusText}")<commit_msg>add header check<commit_after>package jobs

import javax.inject.{Inject, Singleton}

import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global
import akka.actor.ActorSystem
import common.Config
import controllers.Croissants
import models.Croissant
import modules.mail.Mail
import org.joda.time.DateTime
import play.api.libs.json.{Json, Reads}
import play.api.libs.ws.WSClient
import play.modules.reactivemongo.ReactiveMongoApi

import scala.concurrent.Future
import scala.util.{Failure, Success}

case class AccessToken(
  value: String,
  refreshAt: DateTime
)

@Singleton
class GmailJob @Inject()(
  val ws: WSClient,
  val system: ActorSystem
)(implicit
  val config: Config,
  val reactiveMongoApi: ReactiveMongoApi,
  val mailer: Mail) {

  val logger = play.api.Logger("GmailJob")

  val refreshtoken = config.Gmail.refreshtoken
  val clientId = config.Gmail.clientId
  val clientSecret = config.Gmail.clientSecret

  val fromRegex = "(.*)<(.*)>".r
  val receivedFromRegex = "from +.*.google.com +\\(".r

  def schedule(accessToken: Option[AccessToken]): Unit = {
    system.scheduler.scheduleOnce(5 seconds) {
      val now = DateTime.now

      val accessTokenFtr = if (accessToken.isEmpty || now.isAfter(accessToken.get.refreshAt)) {
        refreshAccessToken()
      } else Future.successful(accessToken.get)

      val result = for {
        accessToken <- accessTokenFtr
        messages <- listUnreadGmailMessages(accessToken)
      } yield (accessToken, messages)

      result.onComplete {
        case Success((accessToken, messages)) =>
          messages.foreach { message =>
            for {
              from <- message.payload.headers.collect { case Header("From", value) => value }.headOption
              (name, email) <- from match {
                case fromRegex(name, email) => Some((name.trim, email.trim))
              }
            } yield {
              val receivedFromHeaders = message.payload.headers.filter(header => header.name == "Received" && header.value.trim.startsWith("from"))
              val headerValid = receivedFromHeaders.nonEmpty && receivedFromHeaders.forall { header =>
                receivedFromRegex.findFirstIn(header.value).isDefined
              }

              if (headerValid) {
                val subject = message.payload.headers.collect { case Header("Subject", value) => value }.headOption
                Croissant.addCroissant(email, subject)
              } else {
                logger.warn(s"Received invalid email with headers: ${message.payload.headers}")
              }
            }
          }
          schedule(Some(accessToken))
        case Failure(e) =>
          logger.error("Error during gmail job: ", e)
          schedule(accessToken)
      }
    }
    ()
  }

  case class GmailMessage(
    id: String,
    payload: Payload
  )
  case class Payload(
    headers: Seq[Header]
  )
  case class Header(
    name: String,
    value: String
  )
  implicit val headerReads: Reads[Header] = Json.reads[Header]
  implicit val payloadReads: Reads[Payload] = Json.reads[Payload]
  implicit val gmailMessagesReads: Reads[GmailMessage] = Json.reads[GmailMessage]
  private def listUnreadGmailMessages(accessToken: AccessToken): Future[Seq[GmailMessage]] = {
    val gmailMessagesBaseUrl = "https://www.googleapis.com/gmail/v1/users/me/messages"
    ws.url(gmailMessagesBaseUrl)
      .withQueryString("q" -> "is:unread")
      .withHeaders(
        "Authorization" -> s"Bearer ${accessToken.value}"
      )
      .get()
      .map {
        case response if response.status == 200 =>
          (response.json \ "messages" \\ "id").map(_.as[String])
        case response => throw new Exception(s"Could not get new messages ids, error: ${response.status} ${response.statusText}")
{'fraction_non_alphanumeric': 0.11364910313901345, 'fraction_numerical': 0.0014013452914798206, 'mean_word_length': 2.9, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 27, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4475999', 'n_tokens_mistral': 2129, 'n_tokens_neox': 2077, 'n_words': 461}