Use merge instead of zip study pulling stream

With zip, each element of each stream need to wait for the other
stream to emit, which may take a long time.

modules/ingestor/src/main/scala/mongo.chapter.scala

@@ -111,5 +111,5 @@ object ChapterRepo:
       coll
         .aggregateWithCodec[StudyData](Query.aggregate(ids))
         .all
-        .flatTap(docs => Logger[IO].debug(s"Received $docs chapters"))
+        // .flatTap(docs => Logger[IO].debug(s"Received $docs chapters"))
         .map(_.map(x => x._id -> x).toMap)

modules/ingestor/src/main/scala/mongo.study.scala

@@ -42,11 +42,10 @@ object StudyRepo:
 
     def fetch(since: Instant, until: Instant): fs2.Stream[IO, Result[StudySource]] =
       fs2.Stream.eval(info"Fetching studies from $since to $until") *>
-        pullAndIndex(since, until)
-          .zip(pullAndDelete(since, until))
-          .map((toIndex, toDelete) => Result(toIndex, toDelete, until.some))
+        pullForIndex(since, until)
+          .merge(pullForDelete(since, until))
 
-    def pullAndIndex(since: Instant, until: Instant) =
+    def pullForIndex(since: Instant, until: Instant): fs2.Stream[IO, Result[StudySource]] =
       val filter = range(F.createdAt)(since, until.some)
         .or(range(F.updatedAt)(since, until.some))
       studies
@@ -55,10 +54,11 @@ object StudyRepo:
         .boundedStream(config.batchSize)
         .chunkN(config.batchSize)
         .map(_.toList)
-        .evalTap(_.traverse_(x => debug"received $x"))
+        // .evalTap(_.traverse_(x => debug"received $x"))
         .evalMap(_.toSources)
+        .map(Result(_, Nil, until.some))
 
-    def pullAndDelete(since: Instant, until: Instant) =
+    def pullForDelete(since: Instant, until: Instant): fs2.Stream[IO, Result[StudySource]] =
       val filter =
         Filter
           .gte("ts", since.asBsonTimestamp)
@@ -72,6 +72,7 @@ object StudyRepo:
         .chunkN(config.batchSize)
         .map(_.toList.flatMap(extractId))
         .evalTap(xs => info"Deleting $xs")
+        .map(Result(Nil, _, until.some))
 
     def extractId(doc: Document): Option[Id] =
       doc.getNestedAs[String](F.oplogId).map(Id.apply)
@@ -91,8 +92,7 @@ object StudyRepo:
         chapters
           .byStudyIds(studyIds)
           .flatMap: chapters =>
-            docs
-              .traverseFilter(_.toSource(chapters))
+            docs.traverseFilter(_.toSource(chapters))
 
     type StudySourceWithId = (String, StudySource)
     extension (doc: Document)