v2

.github/workflows/v2.yml

@@ -0,0 +1,253 @@
+name: Build V2
+
+on:
+  push:
+    branches:
+      - v2
+  pull_request:
+    branches:
+      - v2
+  workflow_dispatch:
+
+
+jobs:
+  pre-check:
+    runs-on: ubuntu-latest
+    outputs:
+      pyMetadata: ${{ steps.filter.outputs.pyMetadata }}
+      commonCode: ${{ steps.filter.outputs.commonCode }}
+      reader: ${{ steps.filter.outputs.reader }}
+      encode: ${{ steps.filter.outputs.encode }}
+      convert: ${{ steps.filter.outputs.convert }}
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v2
+
+      - uses: dorny/paths-filter@v2
+        id: filter
+        with:
+          filters: |
+            apps/pyMetadata:
+              - 'pyMetadata/**'
+            apps/coordinator:
+              - 'coordinator/**'
+            apps/processer:
+              - 'processer/**'
+            apps/converter:
+              - 'converter/**'
+            
+            shared:
+              - 'shared/**'
+      # Step to print the outputs from "pre-check" job
+      - name: Print Outputs from pre-check job
+        run: |
+          echo "Apps"
+          echo "app:pyMetadata: ${{ needs.pre-check.outputs.pyMetadata }}"
+          echo "app:coordinator: ${{ needs.pre-check.outputs.coordinator }}"
+          echo "app:processer: ${{ needs.pre-check.outputs.processer }}"
+          echo "app:converter: ${{ needs.pre-check.outputs.converter }}"
+          
+          echo "Shared"
+          echo "shared: ${{ needs.pre-check.outputs.shared }}"
+          
+  build-shared:
+    runs-on: ubuntu-latest
+    needs: pre-check
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v2
+
+      - name: Cache Shared code Gradle dependencies
+        id: cache-gradle
+        uses: actions/cache@v2
+        with:
+          path: ~/.gradle/caches
+          key: ${{ runner.os }}-gradle-${{ hashFiles('shared/build.gradle.kts') }}
+
+      - name: Build Shared code
+        if: steps.cache-gradle.outputs.cache-hit != 'true' || needs.pre-check.outputs.commonCode == 'true' || github.event_name == 'workflow_dispatch'
+        run: |
+          cd CommonCode 
+          chmod +x ./gradlew
+          ./gradlew build          
+
+
+  build-processer:
+    needs: build-shared
+    if: ${{ needs.pre-check.outputs.processer == 'true' || github.event_name == 'workflow_dispatch' || needs.pre-check.outputs.shared == 'true' }}
+    runs-on: ubuntu-latest
+    #if: ${{ github.event_name == 'push' || github.event_name == 'workflow_dispatch' }}
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v2
+
+      - name: Cache Shared Gradle dependencies
+        id: cache-gradle
+        uses: actions/cache@v2
+        with:
+          path: ~/.gradle/caches
+          key: ${{ runner.os }}-gradle-${{ hashFiles('shared/build.gradle.kts') }}
+
+      - name: Build Processer module
+        id: build-processer
+        run: |
+          cd apps/processer
+          chmod +x ./gradlew
+          ./gradlew build
+          echo "Build completed"
+
+
+      - name: Generate Docker image tag
+        id: docker-tag
+        run: echo "::set-output name=tag::$(date -u +'%Y.%m.%d')-$(uuidgen | cut -c 1-8)"
+
+      - name: Docker login
+        uses: docker/login-action@f054a8b539a109f9f41c372932f1ae047eff08c9
+        with:
+          username: ${{ secrets.DOCKER_HUB_NAME }}
+          password: ${{ secrets.DOCKER_HUB_TOKEN }}
+
+      - name: Build and push Docker image
+        uses: docker/build-push-action@v2
+        with:
+          context: ./apps/processer
+          push: true
+          tags: |
+            bskjon/mediaprocessing-processer:v2-latest
+            bskjon/mediaprocessing-processer:v2-${{ github.sha }}
+            bskjon/mediaprocessing-processer:v2-${{ steps.docker-tag.outputs.tag }}
+
+  build-converter:
+    needs: build-shared
+    if: ${{ needs.pre-check.outputs.converter == 'true' || github.event_name == 'workflow_dispatch' || needs.pre-check.outputs.shared == 'true' }}
+    runs-on: ubuntu-latest
+    #if: ${{ github.event_name == 'push' || github.event_name == 'workflow_dispatch' }}
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v2
+
+      - name: Cache Shared Gradle dependencies
+        id: cache-gradle
+        uses: actions/cache@v2
+        with:
+          path: ~/.gradle/caches
+          key: ${{ runner.os }}-gradle-${{ hashFiles('shared/build.gradle.kts') }}
+
+      - name: Build Converter module
+        id: build-converter
+        run: |
+          cd apps/converter
+          chmod +x ./gradlew
+          ./gradlew build
+          echo "Build completed"
+
+
+      - name: Generate Docker image tag
+        id: docker-tag
+        run: echo "::set-output name=tag::$(date -u +'%Y.%m.%d')-$(uuidgen | cut -c 1-8)"
+
+      - name: Docker login
+        uses: docker/login-action@f054a8b539a109f9f41c372932f1ae047eff08c9
+        with:
+          username: ${{ secrets.DOCKER_HUB_NAME }}
+          password: ${{ secrets.DOCKER_HUB_TOKEN }}
+
+      - name: Build and push Docker image
+        uses: docker/build-push-action@v2
+        with:
+          context: ./apps/converter
+          push: true
+          tags: |
+            bskjon/mediaprocessing-converter:v2-latest
+            bskjon/mediaprocessing-converter:v2-${{ github.sha }}
+            bskjon/mediaprocessing-converter:v2-${{ steps.docker-tag.outputs.tag }}
+            
+  build-coordinator:
+    needs: build-shared
+    if: ${{ needs.pre-check.outputs.coordinator == 'true' || github.event_name == 'workflow_dispatch' || needs.pre-check.outputs.shared == 'true' }}
+    runs-on: ubuntu-latest
+    #if: ${{ github.event_name == 'push' || github.event_name == 'workflow_dispatch' }}
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v2
+
+      - name: Cache Shared Gradle dependencies
+        id: cache-gradle
+        uses: actions/cache@v2
+        with:
+          path: ~/.gradle/caches
+          key: ${{ runner.os }}-gradle-${{ hashFiles('shared/build.gradle.kts') }}
+
+      - name: Build Coordinator module
+        id: build-coordinator
+        run: |
+          cd apps/coordinator
+          chmod +x ./gradlew
+          ./gradlew build
+          echo "Build completed"
+
+
+      - name: Generate Docker image tag
+        id: docker-tag
+        run: echo "::set-output name=tag::$(date -u +'%Y.%m.%d')-$(uuidgen | cut -c 1-8)"
+
+      - name: Docker login
+        uses: docker/login-action@f054a8b539a109f9f41c372932f1ae047eff08c9
+        with:
+          username: ${{ secrets.DOCKER_HUB_NAME }}
+          password: ${{ secrets.DOCKER_HUB_TOKEN }}
+
+      - name: Build and push Docker image
+        uses: docker/build-push-action@v2
+        with:
+          context: ./apps/coordinator
+          push: true
+          tags: |
+            bskjon/mediaprocessing-coordinator:v2-latest
+            bskjon/mediaprocessing-coordinator:v2-${{ github.sha }}
+            bskjon/mediaprocessing-coordinator:v2-${{ steps.docker-tag.outputs.tag }}
+            
+  build-pymetadata:
+    needs: pre-check
+    if:  ${{ needs.pre-check.outputs.pyMetadata == 'true' || github.event_name == 'workflow_dispatch' }}
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v2
+
+      - name: Build pyMetadata module
+        id: build-pymetadata
+        run: |
+          if [[ "${{ steps.check-pymetadata.outputs.changed }}" == "true" || "${{ github.event_name }}" == "push" || "${{ github.event_name }}" == "workflow_dispatch" ]]; then
+            cd apps/pyMetadata
+            # Add the necessary build steps for your Python module here
+            echo "Build completed"
+          else
+            echo "pyMetadata has not changed. Skipping pyMetadata module build."
+            echo "::set-output name=job_skipped::true"
+          fi
+
+      - name: Generate Docker image tag
+        id: docker-tag
+        run: echo "::set-output name=tag::$(date -u +'%Y.%m.%d')-$(uuidgen | cut -c 1-8)"
+
+      - name: Docker login
+        uses: docker/login-action@f054a8b539a109f9f41c372932f1ae047eff08c9
+        with:
+          username: ${{ secrets.DOCKER_HUB_NAME }}
+          password: ${{ secrets.DOCKER_HUB_TOKEN }}
+
+      - name: Build and push Docker image
+        uses: docker/build-push-action@v2
+        with:
+          context: ./apps/pyMetadata
+          push: true
+          tags: |
+            bskjon/mediaprocessing-pymetadata:v2-latest
+            bskjon/mediaprocessing-pymetadata:v2-${{ github.sha }}
+            bskjon/mediaprocessing-pymetadata:v2-${{ steps.docker-tag.outputs.tag }}

.idea/gradle.xml

@@ -11,7 +11,6 @@
             <option value="$PROJECT_DIR$/apps" />
             <option value="$PROJECT_DIR$/apps/converter" />
             <option value="$PROJECT_DIR$/apps/coordinator" />
-            <option value="$PROJECT_DIR$/apps/encoder" />
             <option value="$PROJECT_DIR$/apps/processer" />
             <option value="$PROJECT_DIR$/apps/ui" />
             <option value="$PROJECT_DIR$/shared" />

apps/converter/src/main/kotlin/no/iktdev/mediaprocessing/converter/tasks/ConvertService.kt

@@ -16,9 +16,8 @@ import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
 import no.iktdev.mediaprocessing.shared.kafka.dto.SimpleMessageData
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.ConvertWorkPerformed
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.ConvertWorkerRequest
-import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.FfmpegWorkPerformed
 import no.iktdev.mediaprocessing.shared.kafka.dto.isSuccess
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
 import java.util.*
@@ -111,6 +110,8 @@ class ConvertService(@Autowired override var coordinator: ConverterCoordinator)
     }
 
     fun skipConvertEvent(event: PersistentProcessDataMessage, requiresEventId: String) {
+        if (event.event == KafkaEvents.EVENT_WORK_CONVERT_CREATED)
+            return
         val producesPayload = SimpleMessageData(status = Status.COMPLETED, message = "Convert event contains a payload stating that it waits for eventId: $requiresEventId with referenceId: ${event.referenceId}")
         coordinator.producer.sendMessage(
             referenceId = event.referenceId,
@@ -126,7 +127,7 @@ class ConvertService(@Autowired override var coordinator: ConverterCoordinator)
                 status = Status.COMPLETED,
                 producedBy = serviceId,
                 derivedFromEventId = converter.eventId,
-                result = result.map { it.absolutePath }
+                outFiles = result.map { it.absolutePath }
             )
         } catch (e: Converter.FileUnavailableException) {
             e.printStackTrace()
@@ -135,7 +136,7 @@ class ConvertService(@Autowired override var coordinator: ConverterCoordinator)
                 message = e.message,
                 producedBy = serviceId,
                 derivedFromEventId = converter.eventId,
-                result = emptyList()
+                outFiles = emptyList()
             )
         } catch (e : Converter.FileIsNullOrEmpty) {
             e.printStackTrace()
@@ -144,16 +145,8 @@ class ConvertService(@Autowired override var coordinator: ConverterCoordinator)
                 message = e.message,
                 producedBy = serviceId,
                 derivedFromEventId = converter.eventId,
-                result = emptyList()
+                outFiles = emptyList()
             )
         }
     }
-
-
-
-    data class PendingWorkerCache(
-        val referenceId: String,
-        val eventId: String,
-        val requiresEventId: String
-    )
 }

apps/coordinator/build.gradle.kts

@@ -36,6 +36,8 @@ dependencies {
     implementation("org.json:json:20210307")
 
     implementation("no.iktdev:exfl:0.0.13-SNAPSHOT")
+    implementation("no.iktdev.streamit.library:streamit-library-db:0.0.6-alpha22")
+
 
     implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.1")
     implementation("com.github.vishna:watchservice-ktx:master-SNAPSHOT")

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/Coordinator.kt

@@ -17,7 +17,7 @@ import no.iktdev.mediaprocessing.shared.kafka.core.DefaultMessageListener
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEnv
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.*
 import no.iktdev.mediaprocessing.shared.kafka.dto.isSuccess
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
 import java.io.File
@@ -41,7 +41,7 @@ class Coordinator() {
 
     public fun startProcess(file: File, type: ProcessType) {
         val processStartEvent = ProcessStarted(
-            status = Status.STARTED,
+            status = Status.COMPLETED,
             file = file.absolutePath,
             type = type
         )
@@ -184,6 +184,7 @@ class Coordinator() {
             val data = message.data as FfmpegWorkerArgumentsCreated
             data.entries.forEach {
                 FfmpegWorkRequestCreated(
+                    status = Status.COMPLETED,
                     inputFile = data.inputFile,
                     arguments = it.arguments,
                     outFile = it.outputFile
@@ -206,6 +207,7 @@ class Coordinator() {
             val data = message.data as FfmpegWorkerArgumentsCreated
             data.entries.forEach {
                 FfmpegWorkRequestCreated(
+                    status = Status.COMPLETED,
                     inputFile = data.inputFile,
                     arguments = it.arguments,
                     outFile = it.outputFile
@@ -218,6 +220,7 @@ class Coordinator() {
                 }
                 val outFile = File(it.outputFile)
                 ConvertWorkerRequest(
+                    status = Status.COMPLETED,
                     requiresEventId = message.eventId,
                     inputFile = it.outputFile,
                     true,

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/CoordinatorApplication.kt

@@ -1,14 +1,20 @@
 package no.iktdev.mediaprocessing.coordinator
 
 
-import kotlinx.coroutines.launch
 import mu.KotlinLogging
 import no.iktdev.exfl.coroutines.Coroutines
+import no.iktdev.exfl.observable.Observables
 import no.iktdev.mediaprocessing.shared.common.DatabaseConfig
 import no.iktdev.mediaprocessing.shared.common.SharedConfig
 import no.iktdev.mediaprocessing.shared.common.datasource.MySqlDataSource
 import no.iktdev.mediaprocessing.shared.common.persistance.events
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEnv
+import no.iktdev.streamit.library.db.tables.*
+import no.iktdev.streamit.library.db.tables.helper.cast_errors
+import no.iktdev.streamit.library.db.tables.helper.data_audio
+import no.iktdev.streamit.library.db.tables.helper.data_video
+import org.jetbrains.exposed.sql.SchemaUtils
+import org.jetbrains.exposed.sql.transactions.transaction
 import org.springframework.boot.autoconfigure.SpringBootApplication
 import org.springframework.boot.runApplication
 import org.springframework.context.ApplicationContext
@@ -27,13 +33,37 @@ fun getContext(): ApplicationContext? {
 }
 
 fun main(args: Array<String>) {
+    Coroutines.addListener(listener = object: Observables.ObservableValue.ValueListener<Throwable> {
+        override fun onUpdated(value: Throwable) {
+            value.printStackTrace()
+        }
+    })
     val dataSource = MySqlDataSource.fromDatabaseEnv();
-    Coroutines.default().launch {
+    dataSource.createDatabase()
-        dataSource.createDatabase()
+
-        dataSource.createTables(
+    val kafkaTables = listOf(
-            events
+        events, // For kafka
-        )
+    )
+
+    dataSource.createTables(*kafkaTables.toTypedArray())
+
+    val tables = arrayOf(
+        catalog,
+        genre,
+        movie,
+        serie,
+        subtitle,
+        summary,
+        users,
+        progress,
+        data_audio,
+        data_video,
+        cast_errors
+    )
+    transaction {
+        SchemaUtils.createMissingTablesAndColumns(*tables)
     }
+
     context = runApplication<CoordinatorApplication>(*args)
     printSharedConfig()
 }
@@ -47,5 +77,5 @@ fun printSharedConfig() {
 
     log.info { "Database: ${DatabaseConfig.database}@${DatabaseConfig.address}:${DatabaseConfig.port}" }
     log.info { "Username: ${DatabaseConfig.username}" }
-    log.info { "Password: ${ if(DatabaseConfig.password.isNullOrBlank()) "Is not set" else "Is set"}" }
+    log.info { "Password: ${if (DatabaseConfig.password.isNullOrBlank()) "Is not set" else "Is set"}" }
 }

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/mapping/MetadataMapping.kt

@@ -1,15 +1,12 @@
 package no.iktdev.mediaprocessing.coordinator.mapping
 
 import no.iktdev.mediaprocessing.shared.common.persistance.PersistentMessage
+import no.iktdev.mediaprocessing.shared.contract.reader.MetadataCoverDto
 import no.iktdev.mediaprocessing.shared.contract.reader.MetadataDto
-import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
+import no.iktdev.mediaprocessing.shared.contract.reader.SummaryInfo
-import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
+import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.*
-import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.BaseInfoPerformed
-import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.MetadataPerformed
-import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.pyMetadata
 import no.iktdev.mediaprocessing.shared.kafka.dto.isSuccess
-import no.iktdev.streamit.library.kafka.dto.Status
+import java.io.File
-
 
 
 class MetadataMapping(val events: List<PersistentMessage>) {
@@ -17,19 +14,45 @@ class MetadataMapping(val events: List<PersistentMessage>) {
 
     fun map(): MetadataDto? {
         val baseInfo = events.find { it.data is BaseInfoPerformed }?.data as BaseInfoPerformed?
+        val mediaReadOut = events.find { it.data is VideoInfoPerformed }?.data as VideoInfoPerformed?
         val meta = events.find { it.data is MetadataPerformed }?.data as MetadataPerformed?
 
+        val coverDownloadTask = events.find { it.data is CoverInfoPerformed }?.data as CoverInfoPerformed?
+        val cover = events.find { it.data is CoverDownloadWorkPerformed }?.data as CoverDownloadWorkPerformed?
+
         if (!baseInfo.isSuccess()) {
             return null
         }
-        return null
-        /*return MetadataDto(
-            title = meta?.data?.title ?: return null,
-            type = meta?.data?.type ?: return null,
 
+        val videoInfo = mediaReadOut?.toValueObject()
 
+        val mediaCover = if (coverDownloadTask != null || cover != null) {
+            val coverFile = cover?.coverFile?.let { File(it) }
+            MetadataCoverDto(
+                cover = coverFile?.name,
+                coverFile = cover?.coverFile,
+                coverUrl = coverDownloadTask?.url
+            )
+        } else null
 
-        )*/
+        return if (meta != null || videoInfo != null) {
+            MetadataDto(
+                title = meta?.data?.title ?: videoInfo?.fullName ?: return null,
+                collection = baseInfo?.title ?: return null,
+                cover = mediaCover,
+                type = meta?.data?.type ?: videoInfo?.type ?: return null,
+                summary = meta?.data?.summary?.filter {it.summary != null }?.map { SummaryInfo(language = it.language, summary = it.summary!! ) } ?: emptyList(),
+                genres = meta?.data?.genres ?: emptyList(),
+            )
+        } else null
+    }
+
+    fun getCollection(): String? {
+        val baseInfo = events.find { it.data is BaseInfoPerformed }?.data as BaseInfoPerformed?
+        if (!baseInfo.isSuccess()) {
+            return null
+        }
+        return baseInfo?.title
     }
 
 }

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/mapping/OutputFilesMapping.kt

@@ -0,0 +1,33 @@
+package no.iktdev.mediaprocessing.coordinator.mapping
+
+import no.iktdev.mediaprocessing.shared.common.persistance.PersistentMessage
+import no.iktdev.mediaprocessing.shared.contract.reader.OutputFilesDto
+import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.ConvertWorkPerformed
+import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.work.ProcesserEncodeWorkPerformed
+import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.work.ProcesserExtractWorkPerformed
+import no.iktdev.mediaprocessing.shared.kafka.dto.isSuccess
+
+class OutputFilesMapping(val events: List<PersistentMessage>) {
+
+    fun mapTo(): OutputFilesDto {
+
+        val videoResult = events.filter { it.data is ProcesserEncodeWorkPerformed }
+            .map { it.data as ProcesserEncodeWorkPerformed }
+
+        val subtitleResult = events.filter { it.data is ProcesserExtractWorkPerformed && it.data.isSuccess() }.map { it.data as ProcesserExtractWorkPerformed }
+        val convertedSubtitleResult = events.filter { it.data is ConvertWorkPerformed && it.data.isSuccess() }.map { it.data as ConvertWorkPerformed }
+
+
+
+        return OutputFilesDto(
+            video = videoResult.lastOrNull { it.isSuccess() }?.outFile,
+            subtitles = toSubtitleList(subtitleResult, convertedSubtitleResult)
+        )
+    }
+
+    private fun toSubtitleList(extracted: List<ProcesserExtractWorkPerformed>, converted: List<ConvertWorkPerformed>): List<String> {
+        val sub1 = extracted.mapNotNull { it.outFile }
+        val sub2 = converted.flatMap { it.outFiles }
+        return sub1 + sub2
+    }
+}

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/mapping/ProcessMapping.kt

@@ -10,12 +10,15 @@ class ProcessMapping(val events: List<PersistentMessage>) {
     fun map(): MediaProcessedDto? {
         val referenceId = events.firstOrNull()?.referenceId ?: return null
         val processStarted = getProcessStarted()
+        val meta = MetadataMapping(events)
         return MediaProcessedDto(
             referenceId = referenceId,
             process = processStarted?.type,
             inputFile = processStarted?.file,
-            metadata = MetadataMapping(events).map(),
+            collection = meta.getCollection(),
-            outputFiles = null
+            metadata = meta.map(),
+            videoDetails = VideoDetailsMapper(events).mapTo(),
+            outputFiles = OutputFilesMapping(events).mapTo()
         )
     }
 
@@ -23,29 +26,37 @@ class ProcessMapping(val events: List<PersistentMessage>) {
         return events.lastOrNull { it.data is ProcessStarted }?.data as ProcessStarted?
     }
 
+
     fun waitsForEncode(): Boolean {
-        val arguments = events.find { it.event == KafkaEvents.EVENT_MEDIA_ENCODE_PARAMETER_CREATED } != null
+        val arguments = events.filter { it.event == KafkaEvents.EVENT_MEDIA_ENCODE_PARAMETER_CREATED }
-        val performed = events.find { it.event == KafkaEvents.EVENT_WORK_ENCODE_PERFORMED } != null
+        val created = events.filter { it.event == KafkaEvents.EVENT_WORK_ENCODE_CREATED}
-        val isSkipped = events.find { it.event == KafkaEvents.EVENT_WORK_ENCODE_SKIPPED } != null
+
-        return !(isSkipped || (arguments && performed))
+        val performed = events.filter { it.event == KafkaEvents.EVENT_WORK_ENCODE_PERFORMED }
+        val isSkipped = events.filter { it.event == KafkaEvents.EVENT_WORK_ENCODE_SKIPPED }
+
+        return (arguments.isNotEmpty() && created.isEmpty()) || created.size > performed.size + isSkipped.size
     }
 
     fun waitsForExtract(): Boolean {
-        val arguments = events.find { it.event == KafkaEvents.EVENT_MEDIA_EXTRACT_PARAMETER_CREATED } != null
+        val arguments = events.filter { it.event == KafkaEvents.EVENT_MEDIA_EXTRACT_PARAMETER_CREATED }
-        val performed = events.find { it.event == KafkaEvents.EVENT_WORK_EXTRACT_PERFORMED } != null
+        val created = events.filter { it.event == KafkaEvents.EVENT_WORK_EXTRACT_CREATED }
-        val isSkipped = events.find { it.event == KafkaEvents.EVENT_WORK_EXTRACT_SKIPPED } != null
+
-        return !(isSkipped || (arguments && performed))
+        val performed = events.filter { it.event == KafkaEvents.EVENT_WORK_EXTRACT_PERFORMED }
+        val isSkipped = events.filter { it.event == KafkaEvents.EVENT_WORK_EXTRACT_SKIPPED }
+
+        return (arguments.isNotEmpty() && created.isEmpty()) || created.size > performed.size + isSkipped.size
     }
 
     fun waitsForConvert(): Boolean {
-        val arguments = events.find { it.event == KafkaEvents.EVENT_WORK_CONVERT_CREATED } != null
+        val created = events.filter { it.event == KafkaEvents.EVENT_WORK_CONVERT_CREATED }
-        val performed = events.find { it.event == KafkaEvents.EVENT_WORK_CONVERT_PERFORMED } != null
+        val performed = events.filter { it.event == KafkaEvents.EVENT_WORK_CONVERT_PERFORMED }
-        val isSkipped = events.find { it.event == KafkaEvents.EVENT_WORK_CONVERT_SKIPPED } != null
+        val isSkipped = events.filter { it.event == KafkaEvents.EVENT_WORK_CONVERT_SKIPPED }
-        return !(isSkipped || (arguments && performed))
+
+        return created.size > performed.size + isSkipped.size
     }
 
     fun canCollect(): Boolean {
-        return waitsForEncode() && waitsForExtract() && waitsForConvert()
+        return (!waitsForEncode() && !waitsForExtract() && !waitsForConvert())
     }
 
 }

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/mapping/VideoDetailsMapper.kt

@@ -0,0 +1,27 @@
+package no.iktdev.mediaprocessing.coordinator.mapping
+
+import no.iktdev.mediaprocessing.shared.common.persistance.PersistentMessage
+import no.iktdev.mediaprocessing.shared.contract.reader.SerieInfo
+import no.iktdev.mediaprocessing.shared.contract.reader.VideoDetails
+import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.EpisodeInfo
+import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.VideoInfoPerformed
+
+class VideoDetailsMapper(val events: List<PersistentMessage>) {
+
+    fun mapTo(): VideoDetails? {
+        val mediaReadOut = events.lastOrNull { it.data is VideoInfoPerformed }?.data as VideoInfoPerformed?
+        val proper = mediaReadOut?.toValueObject() ?: return null
+
+        val details = VideoDetails(
+            type = proper.type,
+            fullName = proper.fullName,
+            serieInfo = if (proper !is EpisodeInfo) null else SerieInfo(
+                episodeTitle = proper.episodeTitle,
+                episodeNumber = proper.episode,
+                seasonNumber = proper.season,
+                title = proper.title
+            )
+        )
+        return details
+    }
+}

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/event/BaseInfoFromFile.kt

@@ -7,7 +7,7 @@ import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.BaseInfoPerformed
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.ProcessStarted
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 import org.springframework.stereotype.Service
 import java.io.File
 

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/event/CollectAndStoreTask.kt

@@ -0,0 +1,144 @@
+package no.iktdev.mediaprocessing.coordinator.tasks.event
+
+import no.iktdev.mediaprocessing.coordinator.TaskCreator
+import no.iktdev.mediaprocessing.coordinator.mapping.ProcessMapping
+import no.iktdev.mediaprocessing.shared.common.datasource.executeOrException
+import no.iktdev.mediaprocessing.shared.common.datasource.executeWithStatus
+import no.iktdev.mediaprocessing.shared.common.datasource.withTransaction
+import no.iktdev.mediaprocessing.shared.common.persistance.PersistentMessage
+import no.iktdev.mediaprocessing.shared.contract.reader.MetadataDto
+import no.iktdev.mediaprocessing.shared.contract.reader.VideoDetails
+import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
+import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents.*
+import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
+import no.iktdev.mediaprocessing.shared.kafka.dto.SimpleMessageData
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.isSuccess
+import no.iktdev.streamit.library.db.query.*
+import org.jetbrains.exposed.exceptions.ExposedSQLException
+import org.springframework.stereotype.Service
+import java.io.File
+import java.sql.SQLIntegrityConstraintViolationException
+
+@Service
+class CollectAndStoreTask() : TaskCreator() {
+    override val producesEvent: KafkaEvents = KafkaEvents.EVENT_COLLECT_AND_STORE
+
+    override val requiredEvents: List<KafkaEvents> = listOf(
+        EVENT_PROCESS_STARTED,
+        EVENT_PROCESS_COMPLETED
+    )
+    override val listensForEvents: List<KafkaEvents> = KafkaEvents.entries
+
+
+
+    override fun onProcessEvents(event: PersistentMessage, events: List<PersistentMessage>): MessageDataWrapper? {
+        val started = events.find { it.event == EVENT_PROCESS_STARTED } ?: return null
+        val completed = events.find { it.event == EVENT_PROCESS_COMPLETED } ?: return null
+        if (!started.data.isSuccess() || !completed.data.isSuccess() && completed.data.status != Status.SKIPPED) {
+            return null
+        }
+        val mapped = ProcessMapping(events).map() ?: return null
+        val collection = mapped.collection ?: return null
+
+        val subtitlesStored = mapped.outputFiles?.subtitles?.let {
+            storeSubtitles(collection = collection, subtitles = it)
+        } ?: false
+
+        val videoFile = mapped.outputFiles?.video?.let { File(it).name }
+        val videoInfo = mapped.videoDetails
+
+
+        val genres = mapped.metadata?.genres?.let {
+            storeAndGetGenres(it)
+        }
+
+        val catalogId = mapped.metadata?.let { meta ->
+            if (videoInfo == null || videoFile == null)
+                null
+            else
+                storeCatalog(metadata = meta,genres = genres, videoFile = videoFile, videoDetails = videoInfo)
+        } ?: return SimpleMessageData(Status.ERROR, "Unable to store catalog when metadata is null")
+
+        mapped.metadata?.let {
+            storeMetadata(catalogId = catalogId, metadata = it)
+        }
+
+        return SimpleMessageData(Status.COMPLETED)
+    }
+
+    private fun storeSubtitles(collection: String, subtitles: List<String>): Boolean {
+        val result = subtitles.map { subtitle ->
+            val subtitleFile = File(subtitle)
+            val language = subtitleFile.parentFile.name
+            subtitle to executeWithStatus {
+                SubtitleQuery(
+                    collection = collection,
+                    associatedWithVideo = subtitleFile.nameWithoutExtension,
+                    language = language,
+                    format = subtitleFile.extension.uppercase(),
+                    file = subtitleFile.name
+                ).insert()
+            }
+        }
+        return result.none { !it.second }
+    }
+
+    private fun storeMetadata(catalogId: Int, metadata: MetadataDto) {
+        metadata.summary.forEach {
+            withTransaction {
+                SummaryQuery(
+                    cid = catalogId,
+                    language = it.language,
+                    description = it.summary
+                ).insert()
+            }
+        }
+    }
+
+    private fun storeAndGetGenres(genres: List<String>): String? {
+        return withTransaction {
+            val gq = GenreQuery( *genres.toTypedArray() )
+            gq.insertAndGetIds()
+            gq.getIds().joinToString(",")
+        }
+    }
+
+    private fun storeCatalog(metadata: MetadataDto, videoDetails: VideoDetails, videoFile: String, genres: String?): Int? {
+        val precreatedCatalogQuery = CatalogQuery(
+                title = metadata.title,
+                cover = metadata.cover?.cover,
+                type = metadata.type,
+                collection = metadata.collection,
+                genres = genres
+            )
+
+        val result = when (videoDetails.type) {
+            "serie" -> {
+                val serieInfo = videoDetails.serieInfo ?: throw RuntimeException("SerieInfo missing in VideoDetails for Serie! $videoFile")
+                executeOrException {
+                    precreatedCatalogQuery.insertWithSerie(
+                        episodeTitle = serieInfo.episodeTitle ?: "",
+                        videoFile = videoFile,
+                        episode = serieInfo.episodeNumber,
+                        season = serieInfo.seasonNumber
+                    )
+                }
+            }
+            "movie" -> {
+                executeOrException {
+                    precreatedCatalogQuery.insertWithMovie(videoFile)
+                }
+            }
+            else -> throw RuntimeException("${videoDetails.type} is not supported!")
+        }
+        val ignoreException = result?.cause is SQLIntegrityConstraintViolationException && (result as ExposedSQLException).errorCode == 1062
+        return if (result == null || ignoreException ) {
+            return withTransaction {
+                precreatedCatalogQuery.getId()
+            }
+        } else null
+    }
+
+
+}

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/event/CompleteTask.kt

@@ -0,0 +1,56 @@
+package no.iktdev.mediaprocessing.coordinator.tasks.event
+
+import no.iktdev.mediaprocessing.coordinator.TaskCreator
+import no.iktdev.mediaprocessing.coordinator.mapping.ProcessMapping
+import no.iktdev.mediaprocessing.shared.common.persistance.PersistentMessage
+import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
+import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents.*
+import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.ProcessCompleted
+import no.iktdev.mediaprocessing.shared.kafka.dto.isSuccess
+import org.springframework.stereotype.Service
+
+@Service
+class CompleteTask() : TaskCreator() {
+    override val producesEvent: KafkaEvents = KafkaEvents.EVENT_PROCESS_COMPLETED
+
+    override val requiredEvents: List<KafkaEvents> = listOf(
+        EVENT_PROCESS_STARTED,
+        EVENT_MEDIA_READ_BASE_INFO_PERFORMED,
+        EVENT_MEDIA_READ_OUT_NAME_AND_TYPE
+    )
+    override val listensForEvents: List<KafkaEvents> = KafkaEvents.entries
+
+
+
+    override fun onProcessEvents(event: PersistentMessage, events: List<PersistentMessage>): MessageDataWrapper? {
+        val started = events.find { it.event == KafkaEvents.EVENT_PROCESS_STARTED } ?: return null
+        if (!started.data.isSuccess()) {
+            return null
+        }
+
+        val receivedEvents = events.map { it.event }
+
+        val requiresOneOf = listOf(
+            EVENT_MEDIA_EXTRACT_PARAMETER_CREATED,
+            EVENT_MEDIA_ENCODE_PARAMETER_CREATED,
+            EVENT_MEDIA_DOWNLOAD_COVER_PARAMETER_CREATED,
+            EVENT_WORK_CONVERT_CREATED
+        )
+
+        if (!requiresOneOf.any { it in receivedEvents }) {
+            log.info { "Can't complete at this moment. Missing required event" }
+            return null //SimpleMessageData(Status.SKIPPED, "Can't collect at this moment. Missing required event")
+        }
+
+
+
+
+        val mapper = ProcessMapping(events)
+        if (mapper.canCollect()) {
+            return ProcessCompleted(Status.COMPLETED)
+        }
+        return null
+    }
+}

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/event/DownloadAndStoreCoverTask.kt

@@ -0,0 +1,57 @@
+package no.iktdev.mediaprocessing.coordinator.tasks.event
+
+import kotlinx.coroutines.runBlocking
+import no.iktdev.mediaprocessing.coordinator.TaskCreator
+import no.iktdev.mediaprocessing.shared.common.DownloadClient
+import no.iktdev.mediaprocessing.shared.common.getComputername
+import no.iktdev.mediaprocessing.shared.common.persistance.PersistentMessage
+import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
+import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
+import no.iktdev.mediaprocessing.shared.kafka.dto.SimpleMessageData
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.CoverDownloadWorkPerformed
+import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.CoverInfoPerformed
+import org.springframework.stereotype.Service
+import java.io.File
+import java.util.*
+
+@Service
+class DownloadAndStoreCoverTask: TaskCreator() {
+    val serviceId = "${getComputername()}::${this.javaClass.simpleName}::${UUID.randomUUID()}"
+    override val producesEvent: KafkaEvents
+        get() = KafkaEvents.EVENT_WORK_DOWNLOAD_COVER_PERFORMED
+
+    override val requiredEvents: List<KafkaEvents>
+        get() = listOf(
+            KafkaEvents.EVENT_MEDIA_METADATA_SEARCH_PERFORMED,
+            KafkaEvents.EVENT_MEDIA_READ_OUT_COVER,
+            KafkaEvents.EVENT_WORK_ENCODE_PERFORMED
+        )
+    override fun prerequisitesRequired(events: List<PersistentMessage>): List<() -> Boolean> {
+        return super.prerequisitesRequired(events) + listOf {
+            isPrerequisiteDataPresent(events)
+        }
+    }
+
+    override fun onProcessEvents(event: PersistentMessage, events: List<PersistentMessage>): MessageDataWrapper? {
+        val cover = events.find { it.event == KafkaEvents.EVENT_MEDIA_READ_OUT_COVER }
+        if (cover == null || cover.data !is CoverInfoPerformed) {
+            return SimpleMessageData(Status.ERROR, "Wrong type triggered and caused an execution for $serviceId")
+        }
+        val coverData = cover.data as CoverInfoPerformed
+        val outDir = File(coverData.outDir)
+        if (!outDir.exists())
+            return SimpleMessageData(Status.ERROR, "Check for output directory for cover storage failed for $serviceId")
+
+        val client = DownloadClient(coverData.url, File(coverData.outDir), coverData.outFileBaseName)
+        val result = runBlocking {
+            client.download()
+        }
+        return if (result == null) {
+            SimpleMessageData(Status.ERROR, "Could not download cover, check logs")
+        } else {
+            val status = if (result.exists() && result.canRead()) Status.COMPLETED else Status.ERROR
+            CoverDownloadWorkPerformed(status = status, coverFile = result.absolutePath)
+        }
+    }
+}

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/event/MetadataAndBaseInfoToCoverTask.kt

@@ -8,7 +8,7 @@ import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.BaseInfoPerforme
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.CoverInfoPerformed
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.MetadataPerformed
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.VideoInfoPerformed
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 import org.springframework.stereotype.Service
 
 @Service

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/event/MetadataAndBaseInfoToFileOut.kt

@@ -14,7 +14,7 @@ import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.MetadataPerforme
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.VideoInfoPerformed
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.hasValidData
 import no.iktdev.mediaprocessing.shared.kafka.dto.isSuccess
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 import org.springframework.scheduling.annotation.EnableScheduling
 import org.springframework.scheduling.annotation.Scheduled
 import org.springframework.stereotype.Service

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/event/ParseVideoFileStreams.kt

@@ -11,7 +11,7 @@ import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.MediaStreamsParsePerformed
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.ReaderPerformed
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 import org.springframework.stereotype.Service
 
 @Service

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/event/ReadVideoFileStreams.kt

@@ -12,7 +12,7 @@ import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.ProcessStarted
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.ReaderPerformed
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 import org.springframework.stereotype.Service
 import java.io.File
 

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/event/ffmpeg/EncodeArgumentCreatorTask.kt

@@ -9,7 +9,7 @@ import no.iktdev.mediaprocessing.shared.contract.ffmpeg.*
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.*
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 import org.springframework.stereotype.Service
 import java.io.File
 

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/event/ffmpeg/ExtractArgumentCreatorTask.kt

@@ -12,7 +12,7 @@ import no.iktdev.mediaprocessing.shared.contract.ffmpeg.SubtitleStream
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.*
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 import org.springframework.stereotype.Service
 import java.io.File
 

apps/coordinator/src/main/kotlin/no/iktdev/mediaprocessing/coordinator/tasks/input/watcher/InputDirectoryWatcher.kt

@@ -1,6 +1,7 @@
 package no.iktdev.mediaprocessing.coordinator.tasks.input.watcher
 
 import dev.vishna.watchservice.KWatchEvent.Kind.Deleted
+import dev.vishna.watchservice.KWatchEvent.Kind.Initialized
 import dev.vishna.watchservice.asWatchChannel
 import kotlinx.coroutines.channels.consumeEach
 import kotlinx.coroutines.launch
@@ -51,6 +52,7 @@ class InputDirectoryWatcher(@Autowired var coordinator: Coordinator): FileWatche
                 }
                 when (it.kind) {
                     Deleted -> queue.removeFromQueue(it.file, this@InputDirectoryWatcher::onFileRemoved)
+                    Initialized -> { /* Do nothing */ }
                     else -> {
                         if (it.file.isFile && it.file.isSupportedVideoFile()) {
                             queue.addToQueue(it.file, this@InputDirectoryWatcher::onFilePending, this@InputDirectoryWatcher::onFileAvailable)

apps/coordinator/src/test/kotlin/no/iktdev/mediaprocessing/coordinator/reader/BaseInfoFromFileTest.kt

@@ -5,7 +5,7 @@ import no.iktdev.mediaprocessing.shared.kafka.core.CoordinatorProducer
 import no.iktdev.mediaprocessing.shared.contract.ProcessType
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.BaseInfoPerformed
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.ProcessStarted
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Named
 import org.junit.jupiter.api.Test

apps/coordinator/src/test/kotlin/no/iktdev/mediaprocessing/coordinator/reader/ParseVideoFileStreamsTest.kt

@@ -4,7 +4,7 @@ import com.google.gson.Gson
 import com.google.gson.JsonObject
 import no.iktdev.mediaprocessing.shared.kafka.core.CoordinatorProducer
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.ReaderPerformed
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 import org.junit.jupiter.api.Test
 
 import org.junit.jupiter.api.Assertions.*

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/services/EncodeService.kt

@@ -14,10 +14,10 @@ import no.iktdev.mediaprocessing.shared.common.persistance.PersistentDataStore
 import no.iktdev.mediaprocessing.shared.common.persistance.PersistentProcessDataMessage
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.FfmpegWorkPerformed
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.FfmpegWorkRequestCreated
 import no.iktdev.mediaprocessing.processer.ProcesserEnv
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.work.ProcesserEncodeWorkPerformed
 import org.springframework.stereotype.Service
 import java.io.File
 import java.util.*
@@ -136,7 +136,8 @@ class EncodeService: TaskCreator() {
                     readbackIsSuccess = PersistentDataReader().isProcessEventDefinedAsConsumed(runner.referenceId, runner.eventId, encodeServiceId)
                 }
                 producer.sendMessage(referenceId = runner.referenceId, event = producesEvent,
-                    FfmpegWorkPerformed(status = Status.COMPLETED, producedBy = encodeServiceId, derivedFromEventId =  runner.eventId))
+                    ProcesserEncodeWorkPerformed(status = Status.COMPLETED, producedBy = encodeServiceId, derivedFromEventId =  runner.eventId, outFile = runner.info.outFile)
+                )
                 clearWorker()
             }
 
@@ -150,7 +151,8 @@ class EncodeService: TaskCreator() {
             }
             log.info { "Encode failed for ${runner.referenceId}" }
             producer.sendMessage(referenceId = runner.referenceId, event = producesEvent,
-                FfmpegWorkPerformed(status = Status.ERROR, message = errorMessage, producedBy = encodeServiceId, derivedFromEventId =  runner.eventId))
+                ProcesserEncodeWorkPerformed(status = Status.ERROR, message = errorMessage, producedBy = encodeServiceId, derivedFromEventId =  runner.eventId)
+            )
             sendProgress(info = info, ended = true)
             clearWorker()
         }

apps/processer/src/main/kotlin/no/iktdev/mediaprocessing/processer/services/ExtractService.kt

@@ -15,10 +15,10 @@ import no.iktdev.mediaprocessing.shared.common.persistance.PersistentDataStore
 import no.iktdev.mediaprocessing.shared.common.persistance.PersistentProcessDataMessage
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.FfmpegWorkPerformed
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.FfmpegWorkRequestCreated
 import no.iktdev.mediaprocessing.processer.ProcesserEnv
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.work.ProcesserExtractWorkPerformed
 import org.springframework.stereotype.Service
 import java.io.File
 import java.util.*
@@ -141,7 +141,11 @@ class ExtractService: TaskCreator() {
 
 
                 producer.sendMessage(referenceId = runner.referenceId, event = producesEvent,
-                    FfmpegWorkPerformed(status = Status.COMPLETED, producedBy = extractServiceId, derivedFromEventId =  runner.eventId)
+                    ProcesserExtractWorkPerformed(
+                        status = Status.COMPLETED,
+                        producedBy = extractServiceId,
+                        derivedFromEventId =  runner.eventId,
+                        outFile = runner.info.outFile)
                 )
                 log.info { "Extract is releasing worker" }
                 clearWorker()
@@ -156,7 +160,7 @@ class ExtractService: TaskCreator() {
             }
             log.info { "Extract failed for ${runner.referenceId}" }
             producer.sendMessage(referenceId = runner.referenceId, event = producesEvent,
-                FfmpegWorkPerformed(status = Status.ERROR, message = errorMessage, producedBy = extractServiceId, derivedFromEventId =  runner.eventId)
+                ProcesserExtractWorkPerformed(status = Status.ERROR, message = errorMessage, producedBy = extractServiceId, derivedFromEventId =  runner.eventId)
             )
             sendState(info, ended= true)
             clearWorker()

apps/pyMetadata/app.py

@@ -7,6 +7,7 @@ import uuid
 import threading
 import json
 import time
+
 from kafka import KafkaConsumer, KafkaProducer
 from fuzzywuzzy import fuzz
 from sources.result import DataResult, Metadata
@@ -18,8 +19,8 @@ from sources.select import UseSource
 
 # Konfigurer Kafka-forbindelsen
 bootstrap_servers = os.environ.get("KAFKA_BOOTSTRAP_SERVER") or "127.0.0.1:9092"
-consumer_group = os.environ.get("KAFKA_CONSUMER_ID") or f"Metadata-{uuid.uuid4()}"
+consumer_group = os.environ.get("KAFKA_CONSUMER_ID") or f"MetadataConsumer"
-kafka_topic = os.environ.get("KAFKA_TOPIC") or "127.0.0.1:9092"
+kafka_topic = os.environ.get("KAFKA_TOPIC") or "mediaEvents"
 
 # Konfigurer logging
 logging.basicConfig(
@@ -154,7 +155,7 @@ class MessageHandlerThread(threading.Thread):
             logger.info("Not in cache: %s", name)
             logger.info("Searching in sources for information about %s", name)
             result: Optional[DataResult] = UseSource(title=name).select_result()
-            if (result.status == "SUCCESS"):
+            if (result.status == "COMPLETED"):
                 logger.info("Storing response for %s in in-memory cache", name)
                 ResultCache.add(name, result)
         return result

apps/pyMetadata/sources/anii.py

@@ -1,5 +1,5 @@
 from AnilistPython import Anilist
-from .result import Metadata, DataResult
+from .result import Metadata, DataResult, Summary
 
 class metadata():
     name: str = None
@@ -17,19 +17,24 @@ class metadata():
                 title = result.get("name_english", None),
                 altTitle = [result.get("name_romaji", [])],
                 cover = result.get("cover_image", None),
-                summary = result.get("desc", None),
+                summary = [
+                    Summary(
+                        language = "eng",
+                        summary = result.get("desc", None)
+                    )
+                ],
                 type = 'movie' if result.get('airing_format', '').lower() == 'movie' else 'serie',
                 genres = result.get('genres', []),
                 source="anii",
                 usedTitle=self.name
             )
             if (meta.title is None) or (meta.type is None):
-                return DataResult("SUCCESS", None, None)
+                return DataResult("COMPLETED", None, None)
 
-            return DataResult("SUCCESS", None, meta)
+            return DataResult("COMPLETED", None, meta)
 
         except IndexError as ingore:
-            return DataResult(statusType="SUCCESS", message=f"No result for {self.name}")
+            return DataResult(statusType="COMPLETED", message=f"No result for {self.name}")
         except Exception as e:
             return DataResult(statusType="ERROR", message=str(e))
             

apps/pyMetadata/sources/imdb.py

@@ -1,5 +1,5 @@
 import imdb
-from .result import Metadata, DataResult
+from .result import Metadata, DataResult, Summary
 
 class metadata():
     name: str = None
@@ -19,15 +19,20 @@ class metadata():
                 title = result.get("title", None),
                 altTitle = [result.get("localized title", [])],
                 cover = result.get("cover url", None),
-                summary = result.get("plot outline", None),
+                summary = [
+                    Summary(
+                        language = "eng",
+                        summary = result.get("plot outline", None)
+                    )
+                ],
                 type = 'movie' if result.get('kind', '').lower() == 'movie' else 'serie',
                 genres = result.get('genres', []),
                 source="imdb",
                 usedTitle=self.name
             )
             if (meta.title is None) or (meta.type is None):
-                return DataResult("SUCCESS", None, None)
+                return DataResult("COMPLETED", None, None)
             
-            return DataResult("SUCCESS", None, meta)
+            return DataResult("COMPLETED", None, meta)
         except Exception as e:
             return DataResult(status="ERROR", data=None, message=str(e))

apps/pyMetadata/sources/mal.py

@@ -1,5 +1,5 @@
 from mal import *
-from .result import Metadata, DataResult
+from .result import Metadata, DataResult, Summary
 
 class metadata():
     name: str = None
@@ -11,21 +11,26 @@ class metadata():
         try:
             search = AnimeSearch(self.name)
             if (len(search.results) == 0):
-                return DataResult(status="SUCCESS", message="No results")
+                return DataResult(status="SKIPPED", message="No results")
             anime = Anime(search.results[0].mal_id)
             meta = Metadata(
                 title = anime.title,
                 altTitle = [altName for altName in [anime.title_english, *anime.title_synonyms] if altName],
                 cover = anime.image_url,
-                summary = anime.synopsis,
+                summary = [
+                    Summary(
+                        language = "eng",
+                        summary = anime.synopsis
+                    )
+                ],
                 type = 'movie' if anime.type.lower() == 'movie' else 'serie',
                 genres = anime.genres,
                 source="mal",
                 usedTitle=self.name
             )
             if (meta.title is None) or (meta.type is None):
-                return DataResult("SUCCESS", None, None)
+                return DataResult("COMPLETED", None, None)
 
-            return DataResult("SUCCESS", None, meta)
+            return DataResult("COMPLETED", None, meta)
         except Exception as e:
             return DataResult(status="ERROR", message=str(e))

apps/pyMetadata/sources/result.py

@@ -1,13 +1,21 @@
 from typing import List, Optional
 from dataclasses import dataclass, asdict
 
+@dataclass
+class Summary:
+    summary: str
+    language: str
+
+    def to_dict(self):
+        return asdict(self)
+
 @dataclass
 class Metadata:
     title: str
     altTitle: List[str]
     cover: str
     type: str  # Serie/Movie
-    summary: str
+    summary: List[Summary]
     genres: List[str]
     source: str
     usedTitle: str
@@ -15,6 +23,7 @@ class Metadata:
     def to_dict(self):
         return asdict(self)
 
+
 @dataclass
 class DataResult:
     status: str # COMPLETED / ERROR

apps/pyMetadata/sources/select.py

@@ -28,11 +28,11 @@ class UseSource():
         mal = MalMetadata(title).lookup()
 
         result: List[WeightedData] = []
-        if (anii is not None) and (anii.status == "SUCCESS" and anii.data is not None):
+        if (anii is not None) and (anii.status == "COMPLETED" and anii.data is not None):
             result.append(WeightedData(anii, 4))
-        if (imdb is not None) and (imdb.status == "SUCCESS" and imdb.data is not None):
+        if (imdb is not None) and (imdb.status == "COMPLETED" and imdb.data is not None):
             result.append(WeightedData(imdb, 1))
-        if (mal is not None) and (mal.status == "SUCCESS" and mal.data is not None):
+        if (mal is not None) and (mal.status == "COMPLETED" and mal.data is not None):
             result.append(WeightedData(mal, 8))
         return result
     

settings.gradle.kts

@@ -3,20 +3,21 @@ plugins {
 }
 rootProject.name = "MediaProcessing"
 include("apps")
+include("apps:ui")
+include("apps:coordinator")
+include("apps:converter")
+include("apps:processer")
+
 include("shared")
 include("shared:kafka")
-findProject(":shared:kafka")?.name = "kafka"
-include("apps:coordinator")
-findProject(":apps:coordinator")?.name = "coordinator"
-include("apps:ui")
-findProject(":apps:ui")?.name = "ui"
-include("apps:encoder")
-findProject(":apps:encoder")?.name = "encoder"
-include("apps:converter")
-findProject(":apps:converter")?.name = "converter"
 include("shared:contract")
-findProject(":shared:contract")?.name = "contract"
 include("shared:common")
-findProject(":shared:common")?.name = "common"
+
-include("apps:processer")
+findProject(":apps:ui")?.name = "ui"
+findProject(":apps:coordinator")?.name = "coordinator"
+findProject(":apps:converter")?.name = "converter"
 findProject(":apps:processer")?.name = "processer"
+
+findProject(":shared:kafka")?.name = "kafka"
+findProject(":shared:contract")?.name = "contract"
+findProject(":shared:common")?.name = "common"

shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/DownloadClient.kt

@@ -1,5 +1,8 @@
 package no.iktdev.mediaprocessing.shared.common
 
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
+import mu.KotlinLogging
 import no.iktdev.exfl.using
 import java.io.File
 import java.io.FileOutputStream
@@ -7,6 +10,7 @@ import java.net.HttpURLConnection
 import java.net.URL
 
 open class DownloadClient(val url: String, val outDir: File, val baseName: String) {
+    val log = KotlinLogging.logger {}
     protected val http: HttpURLConnection = openConnection()
     private val BUFFER_SIZE = 4096
 
@@ -27,10 +31,17 @@ open class DownloadClient(val url: String, val outDir: File, val baseName: Strin
         return ((read * 100) / total)
     }
 
-    suspend fun download(): File? {
+    suspend fun download(): File? = withContext(Dispatchers.IO) {
         val extension = getExtension()
             ?: throw UnsupportedFormatException("Provided url does not contain a supported file extension")
         val outFile = outDir.using("$baseName.$extension")
+        if (!outDir.exists())
+            return@withContext null
+        if (outFile.exists()) {
+            log.info { "${outFile.name} already exists. Download skipped!" }
+            return@withContext outFile
+        }
+
         val inputStream = http.inputStream
         val fos = FileOutputStream(outFile, false)
 
@@ -51,7 +62,7 @@ open class DownloadClient(val url: String, val outDir: File, val baseName: Strin
         }
         inputStream.close()
         fos.close()
-        return outFile
+        return@withContext outFile
     }
 
     open fun getExtension(): String? {

shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/datasource/TableDefaultOperations.kt

@@ -4,7 +4,7 @@ import org.jetbrains.exposed.sql.Table
 
 import org.jetbrains.exposed.sql.transactions.transaction
 
-open class TableDefaultOperations<T: Table> {
+open class TableDefaultOperations<T : Table> {
 
 }
 
@@ -45,7 +45,7 @@ fun <T> insertWithSuccess(block: () -> T): Boolean {
     }
 }
 
-fun <T> executeOrException(block: () -> T): Exception? {
+fun <T> executeOrException(rollbackOnFailure: Boolean = false, block: () -> T): Exception? {
     return try {
         transaction {
             try {
@@ -54,7 +54,8 @@ fun <T> executeOrException(block: () -> T): Exception? {
                 null
             } catch (e: Exception) {
                 // log the error here or handle the exception as needed
-                rollback()
+                if (rollbackOnFailure)
+                    rollback()
                 e
 
             }
@@ -65,6 +66,25 @@ fun <T> executeOrException(block: () -> T): Exception? {
     }
 }
 
+fun <T> executeWithResult(block: () -> T): Pair<T?, Exception?> {
+    return try {
+        transaction {
+            try {
+                val res = block()
+                commit()
+                res to null
+            } catch (e: Exception) {
+                // log the error here or handle the exception as needed
+                rollback()
+                null to e
+            }
+        }
+    } catch (e: Exception) {
+        e.printStackTrace()
+        return null to e
+    }
+}
+
 fun <T> executeWithStatus(block: () -> T): Boolean {
     return try {
         transaction {

shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/persistance/PersistentDataStore.kt

@@ -5,7 +5,6 @@ import no.iktdev.mediaprocessing.shared.common.datasource.executeWithStatus
 import no.iktdev.mediaprocessing.shared.common.datasource.withTransaction
 import no.iktdev.mediaprocessing.shared.kafka.dto.Message
 import org.jetbrains.exposed.exceptions.ExposedSQLException
-import org.jetbrains.exposed.sql.SqlExpressionBuilder.eq
 import org.jetbrains.exposed.sql.and
 import org.jetbrains.exposed.sql.insert
 import org.jetbrains.exposed.sql.javatime.CurrentDateTime
@@ -24,6 +23,7 @@ open class PersistentDataStore {
         }
         return if (exception == null) true else {
             if (exception.cause is SQLIntegrityConstraintViolationException) {
+                exception.printStackTrace()
                 (exception as ExposedSQLException).errorCode == 1062
             }
             else {

shared/common/src/main/kotlin/no/iktdev/mediaprocessing/shared/common/persistance/events.kt

@@ -11,6 +11,7 @@ object events: IntIdTable() {
     val eventId: Column<String> = varchar("eventId", 50)
     val event: Column<String> = varchar("event",100)
     val data: Column<String> = text("data")
+    //val success: Column<Boolean> = bool("success").default(false)
     val created: Column<LocalDateTime> = datetime("created").defaultExpression(CurrentDateTime)
 
     init {

shared/contract/src/main/kotlin/no/iktdev/mediaprocessing/shared/contract/reader/MediaProcessedDto.kt

@@ -5,7 +5,9 @@ import no.iktdev.mediaprocessing.shared.contract.ProcessType
 data class MediaProcessedDto(
     val referenceId: String,
     val process: ProcessType?,
+    val collection: String?,
     val inputFile: String?,
     val metadata: MetadataDto?,
+    val videoDetails: VideoDetails? = null,
     val outputFiles: OutputFilesDto?
 )

shared/contract/src/main/kotlin/no/iktdev/mediaprocessing/shared/contract/reader/MetadataDto.kt

@@ -2,14 +2,20 @@ package no.iktdev.mediaprocessing.shared.contract.reader
 
 data class MetadataDto(
     val title: String,
+    val collection: String,
     val type: String,
-    val cover: MetadataCoverDto,
+    val cover: MetadataCoverDto?,
+    val summary: List<SummaryInfo> = emptyList(),
+    val genres: List<String>,
+)
+
+data class SummaryInfo(
     val summary: String,
-    val genres: List<String>
+    val language: String = "eng"
 )
 
 data class MetadataCoverDto(
-    val cover: String,
+    val cover: String?, // ex Fancy.jpeg
-    val coverUrl: String,
+    val coverUrl: String?,
-    val coverFile: String
+    val coverFile: String?
 )

shared/contract/src/main/kotlin/no/iktdev/mediaprocessing/shared/contract/reader/OutputFilesDto.kt

@@ -1,7 +1,7 @@
 package no.iktdev.mediaprocessing.shared.contract.reader
 
-class OutputFilesDto(
+
-    val videoFile: String,
+data class OutputFilesDto(
-    val videoArguments: List<String>,
+    val video: String?, // FullName (Path + name)
-    val subtitleFiles: List<String>
+    val subtitles: List<String> = emptyList() // (Path + Name)
-)
+)

shared/contract/src/main/kotlin/no/iktdev/mediaprocessing/shared/contract/reader/VideoDetails.kt

@@ -0,0 +1,14 @@
+package no.iktdev.mediaprocessing.shared.contract.reader
+
+data class VideoDetails(
+    val serieInfo: SerieInfo? = null,
+    val type: String,
+    val fullName: String
+)
+
+data class SerieInfo(
+    val episodeTitle: String? = null,
+    val episodeNumber: Int,
+    val seasonNumber: Int,
+    val title: String
+)

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/core/DeserializingRegistry.kt

@@ -7,6 +7,8 @@ import no.iktdev.mediaprocessing.shared.kafka.dto.Message
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
 import no.iktdev.mediaprocessing.shared.kafka.dto.SimpleMessageData
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.*
+import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.work.ProcesserEncodeWorkPerformed
+import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.work.ProcesserExtractWorkPerformed
 
 class DeserializingRegistry {
     private val log = KotlinLogging.logger {}
@@ -29,14 +31,16 @@ class DeserializingRegistry {
             KafkaEvents.EVENT_WORK_EXTRACT_CREATED to FfmpegWorkRequestCreated::class.java,
             KafkaEvents.EVENT_WORK_CONVERT_CREATED to ConvertWorkerRequest::class.java,
 
-            KafkaEvents.EVENT_WORK_ENCODE_PERFORMED to FfmpegWorkPerformed::class.java,
+            KafkaEvents.EVENT_WORK_ENCODE_PERFORMED to ProcesserEncodeWorkPerformed::class.java,
-            KafkaEvents.EVENT_WORK_EXTRACT_PERFORMED to FfmpegWorkPerformed::class.java,
+            KafkaEvents.EVENT_WORK_EXTRACT_PERFORMED to ProcesserExtractWorkPerformed::class.java,
-            KafkaEvents.EVENT_WORK_CONVERT_PERFORMED to null,
+            KafkaEvents.EVENT_WORK_CONVERT_PERFORMED to ConvertWorkPerformed::class.java,
-            KafkaEvents.EVENT_WORK_DOWNLOAD_COVER_PERFORMED to null,
+            KafkaEvents.EVENT_WORK_DOWNLOAD_COVER_PERFORMED to CoverDownloadWorkPerformed::class.java,
 
             KafkaEvents.EVENT_WORK_ENCODE_SKIPPED to null,
             KafkaEvents.EVENT_WORK_EXTRACT_SKIPPED to null,
             KafkaEvents.EVENT_WORK_CONVERT_SKIPPED to null,
+
+            KafkaEvents.EVENT_PROCESS_COMPLETED to ProcessCompleted::class.java
             )
     }
 
@@ -55,7 +59,7 @@ class DeserializingRegistry {
             }
         }
         // Fallback
-        val type = object : TypeToken<Message<out MessageDataWrapper>>() {}.type
+        val type = object : TypeToken<Message<out SimpleMessageData>>() {}.type
         return gson.fromJson<Message<SimpleMessageData>>(json, type)
     }
 

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/core/KafkaEvents.kt

@@ -35,7 +35,8 @@ enum class KafkaEvents(val event: String) {
     EVENT_STORE_COVER_PERFORMED("event:store-cover:performed"),
     EVENT_STORE_METADATA_PERFORMED("event:store-metadata:performed"),
 
-    EVENT_PROCESS_COMPLETED("event:process:completed");
+    EVENT_PROCESS_COMPLETED("event:process:completed"),
+    EVENT_COLLECT_AND_STORE("event::save");
 
     companion object {
         fun toEvent(event: String): KafkaEvents? {

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/Message.kt

@@ -2,7 +2,6 @@ package no.iktdev.mediaprocessing.shared.kafka.dto
 
 import com.google.gson.Gson
 import com.google.gson.reflect.TypeToken
-import no.iktdev.streamit.library.kafka.dto.Status
 import java.lang.reflect.Type
 import java.util.*
 

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/MessageDataWrapper.kt

@@ -1,7 +1,5 @@
 package no.iktdev.mediaprocessing.shared.kafka.dto
 
-import no.iktdev.streamit.library.kafka.dto.Status
-
 
 open class MessageDataWrapper(
     @Transient open val status: Status = Status.ERROR,

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/Status.kt

@@ -1,7 +1,7 @@
-package no.iktdev.streamit.library.kafka.dto
+package no.iktdev.mediaprocessing.shared.kafka.dto
 
 enum class Status {
-    STARTED,
+    SKIPPED,
     COMPLETED,
     ERROR
 }

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/BaseInfoPerformed.kt

@@ -3,7 +3,7 @@ package no.iktdev.mediaprocessing.shared.kafka.dto.events_result
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 
 @KafkaBelongsToEvent(KafkaEvents.EVENT_MEDIA_READ_BASE_INFO_PERFORMED)
 data class BaseInfoPerformed(

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/ConvertWorkPerformed.kt

@@ -3,7 +3,7 @@ package no.iktdev.mediaprocessing.shared.kafka.dto.events_result
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 
 @KafkaBelongsToEvent(KafkaEvents.EVENT_WORK_CONVERT_PERFORMED)
 data class ConvertWorkPerformed(
@@ -11,5 +11,5 @@ data class ConvertWorkPerformed(
     override val message: String? = null,
     val producedBy: String,
     val derivedFromEventId: String,
-    val result: List<String>
+    val outFiles: List<String>
 ): MessageDataWrapper(status, message)

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/ConvertWorkerRequest.kt

@@ -3,12 +3,14 @@ package no.iktdev.mediaprocessing.shared.kafka.dto.events_result
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 
 @KafkaBelongsToEvent(KafkaEvents.EVENT_WORK_CONVERT_CREATED)
 data class ConvertWorkerRequest(
+    override val status: Status,
     val requiresEventId: String? = null,
     val inputFile: String,
     val allowOverwrite: Boolean,
     val outFileBaseName: String,
     val outDirectory: String
-): MessageDataWrapper()
+): MessageDataWrapper(status)

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/CoverDownloadWorkPerformed.kt

@@ -3,11 +3,11 @@ package no.iktdev.mediaprocessing.shared.kafka.dto.events_result
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 
-@KafkaBelongsToEvent(KafkaEvents.EVENT_WORK_CONVERT_CREATED)
+@KafkaBelongsToEvent(KafkaEvents.EVENT_WORK_DOWNLOAD_COVER_PERFORMED)
-data class MediaConvertInfo(
+data class CoverDownloadWorkPerformed(
     override val status: Status,
-    val arguments: List<List<String>>
+    override val message: String? = null,
-
+    val coverFile: String
-): MessageDataWrapper(status)
+): MessageDataWrapper(status, message)

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/CoverInfoPerformed.kt

@@ -1,8 +1,11 @@
 package no.iktdev.mediaprocessing.shared.kafka.dto.events_result
 
+import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
+import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 
+@KafkaBelongsToEvent(KafkaEvents.EVENT_MEDIA_READ_OUT_COVER)
 data class CoverInfoPerformed(
     override val status: Status,
     val url: String,

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/FfmpegWorkRequestCreated.kt

@@ -3,13 +3,15 @@ package no.iktdev.mediaprocessing.shared.kafka.dto.events_result
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 
 @KafkaBelongsToEvent(
     KafkaEvents.EVENT_WORK_ENCODE_CREATED,
     KafkaEvents.EVENT_WORK_EXTRACT_CREATED
 )
 data class FfmpegWorkRequestCreated(
+    override val status: Status,
     val inputFile: String,
     val arguments: List<String>,
     val outFile: String
-): MessageDataWrapper()
+): MessageDataWrapper(status)

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/FfmpegWorkerArgumentsCreated.kt

@@ -3,7 +3,7 @@ package no.iktdev.mediaprocessing.shared.kafka.dto.events_result
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 
 /**
  * @param status Status type

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/MediaEncodeInfo.kt

@@ -1,13 +0,0 @@
-package no.iktdev.mediaprocessing.shared.kafka.dto.events_result
-
-import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
-import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
-import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.streamit.library.kafka.dto.Status
-
-@KafkaBelongsToEvent(KafkaEvents.EVENT_WORK_ENCODE_CREATED)
-data class MediaEncodeInfo(
-    override val status: Status,
-    val arguments: List<String>
-) :
-    MessageDataWrapper(status)

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/MediaExtractInfo.kt

@@ -1,12 +0,0 @@
-package no.iktdev.mediaprocessing.shared.kafka.dto.events_result
-
-import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
-import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
-import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.streamit.library.kafka.dto.Status
-
-@KafkaBelongsToEvent(KafkaEvents.EVENT_WORK_EXTRACT_CREATED)
-data class MediaExtractInfo(
-    override val status: Status,
-    val arguments: List<List<String>>
-) : MessageDataWrapper(status)

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/MediaStreamsParsePerformed.kt

@@ -4,7 +4,7 @@ import no.iktdev.mediaprocessing.shared.contract.ffmpeg.ParsedMediaStreams
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 
 @KafkaBelongsToEvent(KafkaEvents.EVENT_MEDIA_PARSE_STREAM_PERFORMED)
 data class MediaStreamsParsePerformed(

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/MetadataPerformed.kt

@@ -3,7 +3,7 @@ package no.iktdev.mediaprocessing.shared.kafka.dto.events_result
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 
 @KafkaBelongsToEvent(KafkaEvents.EVENT_MEDIA_METADATA_SEARCH_PERFORMED)
 data class MetadataPerformed(
@@ -17,6 +17,11 @@ data class pyMetadata(
     val altTitle: List<String> = emptyList(),
     val cover: String? = null,
     val type: String,
-    val summary: String? = null,
+    val summary: List<pySummary> = emptyList(),
     val genres: List<String> = emptyList()
+)
+
+data class pySummary(
+    val summary: String?,
+    val language: String = "eng"
 )

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/ProcessCompleted.kt

@@ -3,8 +3,10 @@ package no.iktdev.mediaprocessing.shared.kafka.dto.events_result
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 
 @KafkaBelongsToEvent(KafkaEvents.EVENT_PROCESS_COMPLETED)
-data class ProcessCompleted(override val status: Status) : MessageDataWrapper(status) {
+data class ProcessCompleted(
+    override val status: Status
+) : MessageDataWrapper(status) {
 }

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/ProcessStarted.kt

@@ -4,7 +4,7 @@ import no.iktdev.mediaprocessing.shared.contract.ProcessType
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 
 @KafkaBelongsToEvent(KafkaEvents.EVENT_PROCESS_STARTED)
 data class ProcessStarted(

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/ReaderPerformed.kt

@@ -4,7 +4,7 @@ import com.google.gson.JsonObject
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 
 @KafkaBelongsToEvent(KafkaEvents.EVENT_MEDIA_READ_STREAM_PERFORMED)
 data class ReaderPerformed(

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/VideoInfoPerformed.kt

@@ -3,7 +3,7 @@ package no.iktdev.mediaprocessing.shared.kafka.dto.events_result
 import com.google.gson.Gson
 import com.google.gson.JsonObject
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 
 data class VideoInfoPerformed(
     override val status: Status,

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/work/ProcesserEncodeWorkPerformed.kt

@@ -0,0 +1,18 @@
+package no.iktdev.mediaprocessing.shared.kafka.dto.events_result.work
+
+import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
+import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
+import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
+
+// Derived from ffmpeg work
+@KafkaBelongsToEvent(
+    KafkaEvents.EVENT_WORK_ENCODE_PERFORMED
+)
+data class ProcesserEncodeWorkPerformed(
+    override val status: Status,
+    override val message: String? = null,
+    val producedBy: String,
+    val derivedFromEventId: String,
+    val outFile: String? = null
+): MessageDataWrapper(status, message)

shared/kafka/src/main/kotlin/no/iktdev/mediaprocessing/shared/kafka/dto/events_result/work/ProcesserExtractWorkPerformed.kt

@@ -1,17 +1,18 @@
-package no.iktdev.mediaprocessing.shared.kafka.dto.events_result
+package no.iktdev.mediaprocessing.shared.kafka.dto.events_result.work
 
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaBelongsToEvent
 import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 
+// Derived from ffmpeg work
 @KafkaBelongsToEvent(
-    KafkaEvents.EVENT_WORK_ENCODE_PERFORMED,
     KafkaEvents.EVENT_WORK_EXTRACT_PERFORMED
 )
-data class FfmpegWorkPerformed(
+data class ProcesserExtractWorkPerformed(
     override val status: Status,
     override val message: String? = null,
     val producedBy: String,
-    val derivedFromEventId: String
+    val derivedFromEventId: String,
+    val outFile: String? = null
 ): MessageDataWrapper(status, message)

shared/kafka/src/test/kotlin/no/iktdev/mediaprocessing/shared/kafka/SerializationTest.kt

@@ -7,7 +7,7 @@ import no.iktdev.mediaprocessing.shared.kafka.core.KafkaEvents
 import no.iktdev.mediaprocessing.shared.kafka.dto.Message
 import no.iktdev.mediaprocessing.shared.kafka.dto.MessageDataWrapper
 import no.iktdev.mediaprocessing.shared.kafka.dto.events_result.ProcessStarted
-import no.iktdev.streamit.library.kafka.dto.Status
+import no.iktdev.mediaprocessing.shared.kafka.dto.Status
 import org.junit.jupiter.api.Test
 import org.assertj.core.api.Assertions.assertThat