<commit_before>package io.flow.stream
import io.apibuilder.spec.v0.models.{Field, Model, Service, Union}
import io.apibuilder.validation.{ApiBuilderService, ApibuilderType, MultiService}
import io.flow.build.{Application, BuildType, Downloader}
import io.flow.util.{FlowEnvironment, StreamNames, VersionParser}

import scala.annotation.tailrec
import scala.concurrent.ExecutionContext

case class Controller() extends io.flow.build.Controller {
  override val name = "Stream"
  override val command = "stream"


  override def run(buildType: BuildType, downloader: Downloader, services: Seq[Service])(implicit ec: ExecutionContext): Unit = {

    @tailrec
    def loadImports(services: Seq[Service], cached: Map[String, Service]): Seq[Service] = {
      val imports = services.flatMap(_.imports).groupBy(_.uri).values.toList.map(_.sortBy(_.version).last)
      val filteredImports = imports.filterNot(imp => services.exists(svc => svc.organization.key == imp.organization.key && svc.application.key == imp.application.key))
      if (filteredImports.size == 0) {
        services
      } else {
        val importedServices = filteredImports.flatMap { imp =>
          cached.get(imp.organization.key + imp.application.key + imp.version).orElse {
            println("Loading imports...")
            downloader.service(Application(imp.organization.key, imp.application.key, imp.version)) match {
              case Right(svc) => Some(svc)
              case Left(_) =>
                println(s"Error fetching import $imp")
                None
            }
          }
        }
        loadImports(services ++ importedServices, cached)
      }
    }

    buildType match {
      case BuildType.ApiEvent | BuildType.ApiInternalEvent | BuildType.ApiMiscEvent =>
      case class Aggregator(streams: Seq[KinesisStream] = Nil, cache: Map[String, Service] = Map.empty)
        val agg = services.foldLeft(Aggregator()) { case (agg, service) =>
          val allServices = loadImports(Seq(service), agg.cache)
          val ms = MultiService(allServices.map(ApiBuilderService.apply))
          val streams = processService(ms, service)
          Aggregator(agg.streams ++ streams, agg.cache ++ allServices.map(s => s.organization.key + s.application.key + s.version -> s))
        }
        saveDescriptor(buildType, StreamDescriptor(agg.streams))
      case BuildType.Api | BuildType.ApiInternal | BuildType.ApiMisc | BuildType.ApiPartner => // do nothing
    }
  }

  private def processService(multiService: MultiService, service: Service): Seq[KinesisStream] = {
    service.unions.filter(u => u.name.endsWith("_event") && u.discriminator.isDefined).flatMap { union =>
      multiService.findType(union.name) match {
        case None =>
          println(s"Unable to find union ${union.name}")
          None
        case Some(typ) =>
          val className = s"${ApiBuilderUtils.toPackageName(typ.service.namespace, false)}.${ApiBuilderUtils.toClassName(union.name, false)}"
          StreamNames(FlowEnvironment.Production).json(className) match {
            case None =>
              println(s"Unable to generate stream name for union ${union.name} [$className]")
              None
            case Some(streamName) =>
              val candidates = processUnion(multiService, union, streamName).toList
              val upserted = candidates.collect { case u: EventType.Upserted => u }
              val deleted  = candidates.collect { case d: EventType.Deleted => d }
              val pairs = pairUpEvents(upserted, deleted)
              if (pairs.isEmpty) {
                None
              } else {
                val serviceMajorVersion = VersionParser.parse(service.version).major.getOrElse(0)
                val shortName = s"${union.name}_v${serviceMajorVersion}"
                val allModels = multiService.services.flatMap(_.service.models)
                val allUnions = multiService.services.flatMap(_.service.unions)
                val allEnums = multiService.services.flatMap(_.service.enums)
                Some(KinesisStream(streamName, shortName, pairs, allModels, allUnions, allEnums))
              }
          }
      }
    }
  }

  private val UnionMemberRx = "(.*)_(upserted|deleted)_?(.*)".r

  private def processUnion(multiService: MultiService, union: Union, streamName: String): Seq[EventType] = {
    union.types.flatMap { member =>
      multiService.findType(member.`type`) match {
        case None =>
          println(s"Unable to find model for union ${union.name} member ${member.`type`}")
          None
        case Some(ApibuilderType.Enum(ns, enum)) =>
          println(s"Don't know what to do with an union ${union.name} member $enum of type enum")
          None
        case Some(ApibuilderType.Model(ns, model)) =>
          val discriminator = member.discriminatorValue.getOrElse(member.`type`)
          model.name match {
            case UnionMemberRx(typeName, eventType, version) =>
              if (eventType == "upserted") {<commit_msg>Better short name for internal streams<commit_after>package io.flow.stream
import io.apibuilder.spec.v0.models.{Field, Model, Service, Union}
import io.apibuilder.validation.{ApiBuilderService, ApibuilderType, MultiService}
import io.flow.build.{Application, BuildType, Downloader}
import io.flow.util.{FlowEnvironment, StreamNames, VersionParser}

import scala.annotation.tailrec
import scala.concurrent.ExecutionContext

case class Controller() extends io.flow.build.Controller {
  override val name = "Stream"
  override val command = "stream"


  override def run(buildType: BuildType, downloader: Downloader, services: Seq[Service])(implicit ec: ExecutionContext): Unit = {

    @tailrec
    def loadImports(services: Seq[Service], cached: Map[String, Service]): Seq[Service] = {
      val imports = services.flatMap(_.imports).groupBy(_.uri).values.toList.map(_.sortBy(_.version).last)
      val filteredImports = imports.filterNot(imp => services.exists(svc => svc.organization.key == imp.organization.key && svc.application.key == imp.application.key))
      if (filteredImports.size == 0) {
        services
      } else {
        val importedServices = filteredImports.flatMap { imp =>
          cached.get(imp.organization.key + imp.application.key + imp.version).orElse {
            println("Loading imports...")
            downloader.service(Application(imp.organization.key, imp.application.key, imp.version)) match {
              case Right(svc) => Some(svc)
              case Left(_) =>
                println(s"Error fetching import $imp")
                None
            }
          }
        }
        loadImports(services ++ importedServices, cached)
      }
    }

    buildType match {
      case BuildType.ApiEvent | BuildType.ApiInternalEvent | BuildType.ApiMiscEvent =>
      case class Aggregator(streams: Seq[KinesisStream] = Nil, cache: Map[String, Service] = Map.empty)
        val agg = services.foldLeft(Aggregator()) { case (agg, service) =>
          val allServices = loadImports(Seq(service), agg.cache)
          val ms = MultiService(allServices.map(ApiBuilderService.apply))
          val streams = processService(ms, service)
          Aggregator(agg.streams ++ streams, agg.cache ++ allServices.map(s => s.organization.key + s.application.key + s.version -> s))
        }
        saveDescriptor(buildType, StreamDescriptor(agg.streams))
      case BuildType.Api | BuildType.ApiInternal | BuildType.ApiMisc | BuildType.ApiPartner => // do nothing
    }
  }

  private def processService(multiService: MultiService, service: Service): Seq[KinesisStream] = {
    service.unions.filter(u => u.name.endsWith("_event") && u.discriminator.isDefined).flatMap { union =>
      multiService.findType(union.name) match {
        case None =>
          println(s"Unable to find union ${union.name}")
          None
        case Some(typ) =>
          val className = s"${ApiBuilderUtils.toPackageName(typ.service.namespace, false)}.${ApiBuilderUtils.toClassName(union.name, false)}"
          StreamNames(FlowEnvironment.Production).json(className) match {
            case None =>
              println(s"Unable to generate stream name for union ${union.name} [$className]")
              None
            case Some(streamName) =>
              val candidates = processUnion(multiService, union, streamName).toList
              val upserted = candidates.collect { case u: EventType.Upserted => u }
              val deleted  = candidates.collect { case d: EventType.Deleted => d }
              val pairs = pairUpEvents(upserted, deleted)
              if (pairs.isEmpty) {
                None
              } else {
                val serviceMajorVersion = VersionParser.parse(service.version).major.getOrElse(0)
                val internal = if (service.name.contains("-internal-") && !union.name.contains("_internal_")) "internal_" else ""
                val shortName = s"${union.name}_${internal}_v${serviceMajorVersion}"
                val allModels = multiService.services.flatMap(_.service.models)
                val allUnions = multiService.services.flatMap(_.service.unions)
                val allEnums = multiService.services.flatMap(_.service.enums)
                Some(KinesisStream(streamName, shortName, pairs, allModels, allUnions, allEnums))
              }
          }
      }
    }
  }

  private val UnionMemberRx = "(.*)_(upserted|deleted)_?(.*)".r

  private def processUnion(multiService: MultiService, union: Union, streamName: String): Seq[EventType] = {
    union.types.flatMap { member =>
      multiService.findType(member.`type`) match {
        case None =>
          println(s"Unable to find model for union ${union.name} member ${member.`type`}")
          None
        case Some(ApibuilderType.Enum(ns, enum)) =>
          println(s"Don't know what to do with an union ${union.name} member $enum of type enum")
          None
        case Some(ApibuilderType.Model(ns, model)) =>
          val discriminator = member.discriminatorValue.getOrElse(member.`type`)
          model.name match {
            case UnionMemberRx(typeName, eventType, version) =>
              if (eventType == "upserted") {
{'fraction_non_alphanumeric': 0.11578120396739665, 'fraction_numerical': 0.0005892173229892959, 'mean_word_length': 2.7886904761904763, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 49, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3098917', 'n_tokens_mistral': 2877, 'n_tokens_neox': 2814, 'n_words': 612}