Fix logs for the schema validation job (#5157)

* Fix logs for the schema validation job --------- Co-authored-by: Simon Dumas <[email protected]>
BlueBrain · Sep 30, 2024 · d241d12 · d241d12
1 parent 1531189
commit d241d12
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 17 deletions.
diff --git a/...ain/scala/ch/epfl/bluebrain/nexus/delta/sdk/schemas/job/SchemaValidationCoordinator.scala b/...ain/scala/ch/epfl/bluebrain/nexus/delta/sdk/schemas/job/SchemaValidationCoordinator.scala
@@ -19,7 +19,7 @@ object SchemaValidationCoordinator {
   private val logger = Logger[SchemaValidationCoordinator]
 
   def projectionMetadata(project: ProjectRef): ProjectionMetadata =
-    ProjectionMetadata("schema", s"schema-$project-validate-resources", Some(project), None)
+    ProjectionMetadata("schema", s"schema-validate-resources-$project", Some(project), None)
 
   def apply(supervisor: Supervisor, schemaValidationStream: SchemaValidationStream): SchemaValidationCoordinator =
     new SchemaValidationCoordinator {

diff --git a/...src/main/scala/ch/epfl/bluebrain/nexus/delta/sdk/schemas/job/SchemaValidationStream.scala b/...src/main/scala/ch/epfl/bluebrain/nexus/delta/sdk/schemas/job/SchemaValidationStream.scala
@@ -43,21 +43,18 @@ object SchemaValidationStream {
                   }
       } yield (Some(()))
 
-    private def log(message: String) = Stream.eval(logger.info(message))
-
-    override def apply(project: ProjectRef, offset: Offset): ElemStream[Unit] = {
-      for {
-        _      <- log(s"Starting validation of resources for project '$project'")
-        stream <- resourceStream(project, offset).evalMap {
-                    _.evalMapFilter {
-                      case r if r.deprecated                      => IO.none
-                      case r if r.schema.iri == schemas.resources => IO.none
-                      case r                                      => validateSingle(r)
-                    }
-                  }
-        _      <- log(s"Validation of resources for project '$project' has been completed.")
-      } yield stream
-
-    }
+    override def apply(project: ProjectRef, offset: Offset): ElemStream[Unit] =
+      Stream.eval(logger.info(s"Starting validation of resources for project '$project'")) >>
+        resourceStream(project, offset)
+          .evalMap {
+            _.evalMapFilter {
+              case r if r.deprecated                      => IO.none
+              case r if r.schema.iri == schemas.resources => IO.none
+              case r                                      => validateSingle(r)
+            }
+          }
+          .onFinalize {
+            logger.info(s"Validation of resources for project '$project' has been completed.")
+          }
   }
 }