<commit_before>package ee.cone.c4actor.rdb_impl

import com.squareup.wire.{FieldEncoding, ProtoAdapter, ProtoReader, ProtoWriter}
import java.nio.ByteBuffer
import java.nio.charset.StandardCharsets.UTF_8
import java.time.Instant
import java.util.UUID

import FromExternalDBProtocol.DBOffset
import ToExternalDBProtocol.HasState
import ToExternalDBTypes.{NeedSrcId, PseudoOrigNeedSrcId}
import com.typesafe.scalalogging.LazyLogging
import ee.cone.c4actor.QProtocol.{Firstborn, Update}
import ee.cone.c4actor.Types.SrcId
import ee.cone.c4actor._
import ee.cone.c4actor.rdb_impl.ToExternalDBAssembleTypes.PseudoOrig
import ee.cone.c4assemble.Types.{Each, Values}
import ee.cone.c4assemble._
import ee.cone.c4proto
import ee.cone.c4proto._
import okio.ByteString



class RDBOptionFactoryImpl(toUpdate: ToUpdate) extends RDBOptionFactory {
  def dbProtocol(value: Protocol): ExternalDBOption = new ProtocolDBOption(value)
  def fromDB[P <: Product](cl: Class[P]): ExternalDBOption = new FromDBOption(cl.getName)
  def toDB[P <: Product](cl: Class[P], code: List[String]): ExternalDBOption =
    new ToDBOption(cl.getName, code, new ToExternalDBOrigAssemble(toUpdate,cl))
}

////

@protocol(ExchangeCat) object ToExternalDBProtocol extends c4proto.Protocol {
  @Id(0x0063) case class HasState(
    @Id(0x0061) srcId: String,
    @Id(0x0064) valueTypeId: Long,
    @Id(0x0065) value: okio.ByteString
  )
}

object ToBytes {
  def apply(value: Long): Array[Byte] =
    ByteBuffer.allocate(java.lang.Long.BYTES).putLong(value).array()
}

object ToExternalDBTypes {
  type NeedSrcId = SrcId
  type PseudoOrigNeedSrcId = SrcId
}

object ToExternalDBAssembles {
  def apply(options: List[ExternalDBOption]): List[Assemble] =
    new ToExternalDBTxAssemble ::
      options.collect{ case o: ToDBOption ⇒ o.assemble }
}

object ToExternalDBAssembleTypes {
  type PseudoOrig = SrcId
}

trait  ToExternalDBItemAssembleUtil {
  def toUpdate: ToUpdate

  def itemToHasState[Item <: Product]: Item ⇒ Values[(String,HasState)] = item ⇒
    for(e ← LEvent.update(item)) yield {
      val u = toUpdate.toUpdate(e)
      val key = UUID.nameUUIDFromBytes(ToBytes(u.valueTypeId) ++ u.srcId.getBytes(UTF_8)).toString
      key → HasState(key,u.valueTypeId,u.value)
    }
}

@assemble class ToExternalDBOrigAssemble[Item<:Product](
  val toUpdate: ToUpdate,
  classItem: Class[Item]
)  extends Assemble with ToExternalDBItemAssembleUtil {
  def OrigJoin(
    key: SrcId,
    item: Each[Item]
  ): Values[(NeedSrcId,HasState)] =
    itemToHasState(item)

  def PseudoOrigJoin(
    key: SrcId,
    @by[PseudoOrig] item: Each[Item]
  ): Values[(PseudoOrigNeedSrcId,HasState)] =
    itemToHasState(item)
}

@assemble class ToExternalDBTxAssemble extends Assemble with LazyLogging{
  type TypeHex = String
  def joinTasks(
    key: SrcId,
    @by[PseudoOrigNeedSrcId] pseudoNeedStates: Values[HasState],
    @by[NeedSrcId] needStates: Values[HasState],
    hasStates: Values[HasState]
  ): Values[(TypeHex, ToExternalDBTask)] = {
    val mergedNeedStates =
      if (needStates.isEmpty)
        pseudoNeedStates
      else {
        if (pseudoNeedStates.nonEmpty) logger.warn(s"Orig and PseudoOrig conflict: O-$needStates,PSO-$pseudoNeedStates")
        needStates
      }
    if (hasStates.toList == mergedNeedStates) Nil else {
      val typeHex = Hex(Single((hasStates ++ mergedNeedStates).map(_.valueTypeId).distinct))
      List(typeHex → ToExternalDBTask(key, typeHex, Single.option(hasStates), Single.option(mergedNeedStates)))
    }
  }
  def join(
    key: SrcId,
    @by[TypeHex] tasks: Values[ToExternalDBTask]
  ): Values[(SrcId,TxTransform)] = List(WithPK(ToExternalDBTx(key, tasks.toList)))
}

case class ToExternalDBTask(
  srcId: SrcId,
  typeHex: String,
  from: Option[HasState],
  to: Option[HasState]
)

case object RDBSleepUntilKey extends TransientLens[Map[SrcId,(Instant,Option[HasState])]](Map.empty)

case class ToExternalDBTx(typeHex: SrcId, tasks: List[ToExternalDBTask]) extends TxTransform with LazyLogging {
  def transform(local: Context): Context = {
    val now = Instant.now()
    tasks.find{ task ⇒
      val skip = RDBSleepUntilKey.of(local)
      val(until,wasTo) = skip.getOrElse(task.srcId, (Instant.MIN,None))
      until.isBefore(now) || task.to != wasTo
    }.map{ task ⇒ WithJDBCKey.of(local) { conn ⇒
      import task.{from,to}<commit_msg>toList<commit_after>package ee.cone.c4actor.rdb_impl

import com.squareup.wire.{FieldEncoding, ProtoAdapter, ProtoReader, ProtoWriter}
import java.nio.ByteBuffer
import java.nio.charset.StandardCharsets.UTF_8
import java.time.Instant
import java.util.UUID

import FromExternalDBProtocol.DBOffset
import ToExternalDBProtocol.HasState
import ToExternalDBTypes.{NeedSrcId, PseudoOrigNeedSrcId}
import com.typesafe.scalalogging.LazyLogging
import ee.cone.c4actor.QProtocol.{Firstborn, Update}
import ee.cone.c4actor.Types.SrcId
import ee.cone.c4actor._
import ee.cone.c4actor.rdb_impl.ToExternalDBAssembleTypes.PseudoOrig
import ee.cone.c4assemble.Types.{Each, Values}
import ee.cone.c4assemble._
import ee.cone.c4proto
import ee.cone.c4proto._
import okio.ByteString



class RDBOptionFactoryImpl(toUpdate: ToUpdate) extends RDBOptionFactory {
  def dbProtocol(value: Protocol): ExternalDBOption = new ProtocolDBOption(value)
  def fromDB[P <: Product](cl: Class[P]): ExternalDBOption = new FromDBOption(cl.getName)
  def toDB[P <: Product](cl: Class[P], code: List[String]): ExternalDBOption =
    new ToDBOption(cl.getName, code, new ToExternalDBOrigAssemble(toUpdate,cl))
}

////

@protocol(ExchangeCat) object ToExternalDBProtocol extends c4proto.Protocol {
  @Id(0x0063) case class HasState(
    @Id(0x0061) srcId: String,
    @Id(0x0064) valueTypeId: Long,
    @Id(0x0065) value: okio.ByteString
  )
}

object ToBytes {
  def apply(value: Long): Array[Byte] =
    ByteBuffer.allocate(java.lang.Long.BYTES).putLong(value).array()
}

object ToExternalDBTypes {
  type NeedSrcId = SrcId
  type PseudoOrigNeedSrcId = SrcId
}

object ToExternalDBAssembles {
  def apply(options: List[ExternalDBOption]): List[Assemble] =
    new ToExternalDBTxAssemble ::
      options.collect{ case o: ToDBOption ⇒ o.assemble }
}

object ToExternalDBAssembleTypes {
  type PseudoOrig = SrcId
}

trait  ToExternalDBItemAssembleUtil {
  def toUpdate: ToUpdate

  def itemToHasState[Item <: Product]: Item ⇒ Values[(String,HasState)] = item ⇒
    for(e ← LEvent.update(item)) yield {
      val u = toUpdate.toUpdate(e)
      val key = UUID.nameUUIDFromBytes(ToBytes(u.valueTypeId) ++ u.srcId.getBytes(UTF_8)).toString
      key → HasState(key,u.valueTypeId,u.value)
    }
}

@assemble class ToExternalDBOrigAssemble[Item<:Product](
  val toUpdate: ToUpdate,
  classItem: Class[Item]
)  extends Assemble with ToExternalDBItemAssembleUtil {
  def OrigJoin(
    key: SrcId,
    item: Each[Item]
  ): Values[(NeedSrcId,HasState)] =
    itemToHasState(item)

  def PseudoOrigJoin(
    key: SrcId,
    @by[PseudoOrig] item: Each[Item]
  ): Values[(PseudoOrigNeedSrcId,HasState)] =
    itemToHasState(item)
}

@assemble class ToExternalDBTxAssemble extends Assemble with LazyLogging{
  type TypeHex = String
  def joinTasks(
    key: SrcId,
    @by[PseudoOrigNeedSrcId] pseudoNeedStates: Values[HasState],
    @by[NeedSrcId] needStates: Values[HasState],
    hasStates: Values[HasState]
  ): Values[(TypeHex, ToExternalDBTask)] = {
    val mergedNeedStates =
      if (needStates.isEmpty)
        pseudoNeedStates.toList
      else {
        if (pseudoNeedStates.nonEmpty) logger.warn(s"Orig and PseudoOrig conflict: O-$needStates,PSO-$pseudoNeedStates")
        needStates.toList
      }
    if (hasStates.toList == mergedNeedStates) Nil else {
      val typeHex = Hex(Single((hasStates ++ mergedNeedStates).map(_.valueTypeId).distinct))
      List(typeHex → ToExternalDBTask(key, typeHex, Single.option(hasStates), Single.option(mergedNeedStates)))
    }
  }
  def join(
    key: SrcId,
    @by[TypeHex] tasks: Values[ToExternalDBTask]
  ): Values[(SrcId,TxTransform)] = List(WithPK(ToExternalDBTx(key, tasks.toList)))
}

case class ToExternalDBTask(
  srcId: SrcId,
  typeHex: String,
  from: Option[HasState],
  to: Option[HasState]
)

case object RDBSleepUntilKey extends TransientLens[Map[SrcId,(Instant,Option[HasState])]](Map.empty)

case class ToExternalDBTx(typeHex: SrcId, tasks: List[ToExternalDBTask]) extends TxTransform with LazyLogging {
  def transform(local: Context): Context = {
    val now = Instant.now()
    tasks.find{ task ⇒
      val skip = RDBSleepUntilKey.of(local)
      val(until,wasTo) = skip.getOrElse(task.srcId, (Instant.MIN,None))
      until.isBefore(now) || task.to != wasTo
    }.map{ task ⇒ WithJDBCKey.of(local) { conn ⇒
      import task.{from,to}
{'fraction_non_alphanumeric': 0.1129346952829106, 'fraction_numerical': 0.0073453460346608514, 'mean_word_length': 4.797737857618097, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '2276139', 'n_tokens_mistral': 3126, 'n_tokens_neox': 3030, 'n_words': 595}