groupWithin?

Author: swalker2m

modules/service/src/main/scala/lucuma/odb/graphql/mapping/SubscriptionMapping.scala

@@ -6,6 +6,7 @@ package lucuma.odb.graphql
 package mapping
 
 import cats.data.Nested
+import cats.effect.kernel.Temporal
 import cats.syntax.all.*
 import fs2.Stream
 import grackle.Env
@@ -33,10 +34,11 @@ import lucuma.odb.graphql.mapping.ResultMapping.mapSomeFields
 import lucuma.odb.graphql.predicate.Predicates
 import lucuma.odb.instances.given
 import org.tpolecat.typename.TypeName
+import scala.concurrent.duration.*
 
 import scala.reflect.ClassTag
 
-trait SubscriptionMapping[F[_]] extends Predicates[F] {
+trait SubscriptionMapping[F[_]: Temporal] extends Predicates[F] {
 
   def topics: Topics[F]
   def user: User
@@ -90,6 +92,8 @@ trait SubscriptionMapping[F[_]] extends Predicates[F] {
       topics
         .executionEvent
         .subscribe(1024)
+        .groupWithin(10, 100.millis)
+        .flatMap(Stream.chunk(_).changes)
         .filter: e =>
           e.canRead(user)                                              &&
           input.flatMap(_.programId).forall(_ === e.programId)         &&
@@ -118,6 +122,8 @@ trait SubscriptionMapping[F[_]] extends Predicates[F] {
       topics
         .observation
         .subscribe(1024)
+        .groupWithin(10, 100.millis)
+        .flatMap(Stream.chunk(_).changes)
         .filter { e =>
           e.canRead(user) && ((
             input.flatMap(_.programId).forall(_ === e.programId) &&

modules/service/src/main/scala/lucuma/odb/graphql/topic/ExecutionEventAddedTopic.scala

@@ -3,6 +3,7 @@
 
 package lucuma.odb.graphql.topic
 
+import cats.derived.*
 import cats.effect.Concurrent
 import cats.effect.std.Supervisor
 import cats.syntax.apply.*
@@ -27,7 +28,7 @@ object ExecutionEventAddedTopic:
     visitId:         Visit.Id,
     eventType:       ExecutionEventType,
     users:           List[User.Id]
-  ) extends TopicElement
+  ) extends TopicElement derives cats.Eq
 
   private val topic =
     OdbTopic.define[(ExecutionEvent.Id, Program.Id, Observation.Id, Visit.Id, ExecutionEventType), Element](

modules/service/src/main/scala/lucuma/odb/graphql/topic/ObservationTopic.scala

@@ -3,6 +3,8 @@
 
 package lucuma.odb.graphql.topic
 
+import cats.Eq
+import cats.derived.*
 import cats.effect.*
 import cats.effect.std.Supervisor
 import cats.implicits.*
@@ -29,7 +31,7 @@ object ObservationTopic:
     programId:     Program.Id,
     editType:      EditType,
     users:         List[User.Id]
-  ) extends TopicElement
+  ) extends TopicElement derives Eq
 
   private val topic =
     OdbTopic.define[(Observation.Id, Program.Id, EditType), Element](

modules/service/src/main/scala/lucuma/odb/graphql/topic/OdbTopic.scala

@@ -37,7 +37,7 @@ object OdbTopic:
   ): F[List[User.Id]] =
     s.execute(
       sql"""
-        select c_user_id from t_program_user where c_program_id = '#${pid.toString}' and c_user_id notnull
+        select c_user_id from t_program_user where c_program_id = '#${pid.toString}' and c_user_id notnull order by c_user_id
       """.query(user_id)
     )
 

modules/service/src/test/scala/lucuma/odb/graphql/subscription/observationEditSn.scala

@@ -14,6 +14,8 @@ import lucuma.odb.graphql.query.ExecutionTestSupport
 import lucuma.odb.graphql.query.ObservingModeSetupOperations
 
 class observationEditSn extends ExecutionTestSupport with ObservingModeSetupOperations with SubscriptionUtils:
+  val AcquisitionTotal: Long = 219362500l
+  val ScienceTotal: Long     = 784200000l
 
   def subscriptionQuery(pid: Program.Id) =
     s"""
@@ -23,7 +25,26 @@ class observationEditSn extends ExecutionTestSupport with ObservingModeSetupOper
           editType
           value {
             id
+            itc {
+              science {
+                selected {
+                  exposureCount
+                }
+              }
+            }
             execution {
+              digest {
+                acquisition {
+                  timeEstimate {
+                    total { microseconds }
+                  }
+                }
+                science {
+                  timeEstimate {
+                    total { microseconds }
+                  }
+                }
+              }
               config {
                 gmosNorth {
                   science {
@@ -38,6 +59,8 @@ class observationEditSn extends ExecutionTestSupport with ObservingModeSetupOper
         }
       }
     """
+/*
+*/
 
   def subscriptionResponse(oid: Observation.Id): Json =
       Json.obj(
@@ -46,7 +69,30 @@ class observationEditSn extends ExecutionTestSupport with ObservingModeSetupOper
           "editType"      -> Json.fromString(EditType.Updated.tag.toUpperCase),
           "value" -> Json.obj(
             "id"          -> oid.asJson,
+            "itc" -> Json.obj(
+              "science" -> Json.obj(
+                "selected" -> Json.obj(
+                  "exposureCount" -> 6.asJson
+                )
+              )
+            ),
             "execution" -> Json.obj(
+              "digest" -> Json.obj(
+                "acquisition" -> Json.obj(
+                  "timeEstimate" -> Json.obj(
+                    "total" -> Json.obj(
+                      "microseconds" -> AcquisitionTotal.asJson
+                    )
+                  )
+                ),
+                "science" -> Json.obj(
+                  "timeEstimate" -> Json.obj(
+                    "total" -> Json.obj(
+                      "microseconds" -> ScienceTotal.asJson
+                    )
+                  )
+                )
+              ),
               "config" -> Json.obj(
                 "gmosNorth" -> Json.obj(
                   "science" -> Json.obj(
@@ -72,12 +118,18 @@ class observationEditSn extends ExecutionTestSupport with ObservingModeSetupOper
             SET: {
               scienceRequirements: {
                 spectroscopy: {
+                  wavelength: { micrometers: 0.700000 }
+                  resolution: 1000
                   exposureTimeMode: {
                     signalToNoise: {
                       value: 99
                       at: { nanometers: 500 }
                     }
                   }
+                  wavelengthCoverage: { micrometers: 0.400000 }
+                  focalPlane: null
+                  focalPlaneAngle: null
+                  capability: null
                 }
               }
             },
@@ -107,13 +159,11 @@ class observationEditSn extends ExecutionTestSupport with ObservingModeSetupOper
       tid <- createTargetWithProfileAs(pi, pid)
       oid <- createGmosNorthLongSlitObservationAs(pi, pid, List(tid))
       _   <- generateOrFail(pid, oid)
-      // expect two responses, one from editing the S/N, one because our query
-      // requests the sequence which requires a cache update
       _   <- subscriptionExpect(
         user      = pi,
         query     = subscriptionQuery(pid),
         mutations = Right(sleep >> updateSn(oid)),
-        expected  = List(subscriptionResponse(oid), subscriptionResponse(oid))
+        expected  = List(subscriptionResponse(oid))
       )
     yield ()