From adf23c067e72964cbdb2720e239e4be3d8812718 Mon Sep 17 00:00:00 2001 From: Simao Mata Date: Thu, 3 Sep 2026 14:49:28 +0100 Subject: [PATCH] Use device uuid for partition assignment when possible Formatting --- .../libats/messaging_datatype/Messages.scala | 226 ++++++++++++------ 1 file changed, 153 insertions(+), 73 deletions(-) diff --git a/libats-messaging-datatype/src/main/scala/com/advancedtelematic/libats/messaging_datatype/Messages.scala b/libats-messaging-datatype/src/main/scala/com/advancedtelematic/libats/messaging_datatype/Messages.scala index c1a38c52..37f5be31 100644 --- a/libats-messaging-datatype/src/main/scala/com/advancedtelematic/libats/messaging_datatype/Messages.scala +++ b/libats-messaging-datatype/src/main/scala/com/advancedtelematic/libats/messaging_datatype/Messages.scala @@ -1,13 +1,32 @@ package com.advancedtelematic.libats.messaging_datatype -import cats.syntax.show._ +import cats.syntax.show.* import eu.timepit.refined.api.* import eu.timepit.refined.predicates.all.* import com.advancedtelematic.libats.codecs.CirceRefined.* -import com.advancedtelematic.libats.data.DataType.{CorrelationId, Namespace, ResultCode, ResultDescription} -import com.advancedtelematic.libats.messaging_datatype.DataType.UpdateType.UpdateType -import com.advancedtelematic.libats.messaging_datatype.DataType._ -import com.advancedtelematic.libats.messaging_datatype.Messages.{CampaignLaunched, DeviceEventMessage, DeviceSystemInfoChanged, DeviceUpdateAssigned, DeviceUpdateCanceled, DeviceUpdateCompleted, DeviceUpdateEvent, EcuAndHardwareId, EcuReplaced, EcuReplacement, EcuReplacementFailed, SystemInfo, UserCreated} +import com.advancedtelematic.libats.data.DataType.{ + CorrelationId, + Namespace, + ResultCode, + ResultDescription +} +import DataType.UpdateType.UpdateType +import DataType.* +import Messages.{ + CampaignLaunched, + DeviceEventMessage, + DeviceSystemInfoChanged, + DeviceUpdateAssigned, + DeviceUpdateCanceled, + DeviceUpdateCompleted, + DeviceUpdateEvent, + EcuAndHardwareId, + EcuReplaced, + EcuReplacement, + EcuReplacementFailed, + SystemInfo, + UserCreated +} import io.circe._ import io.circe.generic.semiauto._ import io.circe.syntax._ @@ -22,68 +41,93 @@ object MessageCodecs { implicit val eventTypeCodec: Codec[EventType] = deriveCodec implicit val eventCodec: Codec[Event] = deriveCodec - implicit val deviceEventEncoder: Encoder[DeviceEventMessage] = Encoder.instance { x => - eventCodec(x.event).mapObject(_.add("namespace", x.namespace.get.asJson)) - } + implicit val deviceEventEncoder: Encoder[DeviceEventMessage] = + Encoder.instance { x => + eventCodec(x.event).mapObject(_.add("namespace", x.namespace.get.asJson)) + } - implicit val deviceEventDecoder: Decoder[DeviceEventMessage] = Decoder.instance { c => - for { - event <- c.as[Event] - ns <- c.get[String]("namespace").map(Namespace.apply) - } yield DeviceEventMessage(ns, event) - } + implicit val deviceEventDecoder: Decoder[DeviceEventMessage] = + Decoder.instance { c => + for { + event <- c.as[Event] + ns <- c.get[String]("namespace").map(Namespace.apply) + } yield DeviceEventMessage(ns, event) + } implicit val deviceUpdateEventCodec: Codec[DeviceUpdateEvent] = deriveCodec - implicit val deviceUpdateAvailableCodec: Codec[DeviceUpdateAssigned] = deriveCodec - implicit val deviceUpdateCanceledCodec: Codec[DeviceUpdateCanceled] = deriveCodec - implicit val deviceUpdateCompletedCodec: Codec[DeviceUpdateCompleted] = deriveCodec + implicit val deviceUpdateAvailableCodec: Codec[DeviceUpdateAssigned] = + deriveCodec + implicit val deviceUpdateCanceledCodec: Codec[DeviceUpdateCanceled] = + deriveCodec + implicit val deviceUpdateCompletedCodec: Codec[DeviceUpdateCompleted] = + deriveCodec implicit val userCreatedCodec: Codec[UserCreated] = deriveCodec implicit val campaignLaunchedCodec: Codec[CampaignLaunched] = deriveCodec implicit val packageIdCodec: Codec[PackageId] = deriveCodec - implicit val resultCodeEncoder: Encoder[ResultCode] = Encoder.encodeString.contramap(_.value) - implicit val resultCodeDecoder: Decoder[ResultCode] = Decoder.decodeString.map(ResultCode.apply) - implicit val resultDescriptionEncoder: Encoder[ResultDescription] = Encoder.encodeString.contramap(_.value) - implicit val resultDescriptionDecoder: Decoder[ResultDescription] = Decoder.decodeString.map(ResultDescription.apply) + implicit val resultCodeEncoder: Encoder[ResultCode] = + Encoder.encodeString.contramap(_.value) + implicit val resultCodeDecoder: Decoder[ResultCode] = + Decoder.decodeString.map(ResultCode.apply) + implicit val resultDescriptionEncoder: Encoder[ResultDescription] = + Encoder.encodeString.contramap(_.value) + implicit val resultDescriptionDecoder: Decoder[ResultDescription] = + Decoder.decodeString.map(ResultDescription.apply) implicit val installationResultCodec: Codec[InstallationResult] = deriveCodec - implicit val ecuInstallationReportCodec: Codec[EcuInstallationReport] = deriveCodec - implicit val updateTypeCodec: Codec[UpdateType] = Codec.codecForEnumeration(UpdateType) + implicit val ecuInstallationReportCodec: Codec[EcuInstallationReport] = + deriveCodec + implicit val updateTypeCodec: Codec[UpdateType] = + Codec.codecForEnumeration(UpdateType) implicit val systemInfoCodec: Codec[SystemInfo] = deriveCodec - implicit val deviceSystemInfoChangedCodec: Codec[DeviceSystemInfoChanged] = deriveCodec + implicit val deviceSystemInfoChangedCodec: Codec[DeviceSystemInfoChanged] = + deriveCodec implicit val ecuAndHardwareIdCodec: Codec[EcuAndHardwareId] = deriveCodec implicit val ecuReplacementCodec: Codec[EcuReplacement] = Codec.from( Decoder.instance { c => c.get[Boolean]("success").flatMap { - case true => deriveDecoder[EcuReplaced].tryDecode(c) + case true => deriveDecoder[EcuReplaced].tryDecode(c) case false => deriveDecoder[EcuReplacementFailed].tryDecode(c) } }, Encoder.instance { - case e: EcuReplaced => deriveEncoder[EcuReplaced].apply(e).mapObject(("success", Json.fromBoolean(true)) +: _) - case e: EcuReplacementFailed => deriveEncoder[EcuReplacementFailed].apply(e).mapObject(("success", Json.fromBoolean(false)) +: _) + case e: EcuReplaced => + deriveEncoder[EcuReplaced] + .apply(e) + .mapObject(("success", Json.fromBoolean(true)) +: _) + case e: EcuReplacementFailed => + deriveEncoder[EcuReplacementFailed] + .apply(e) + .mapObject(("success", Json.fromBoolean(false)) +: _) } ) } object Messages { - import MessageCodecs._ - import com.advancedtelematic.libats.codecs.CirceCodecs._ + import MessageCodecs.* + import com.advancedtelematic.libats.codecs.CirceCodecs.* final case class UserCreated(id: String) - final case class DeviceSeen(namespace: Namespace, uuid: DeviceId, lastSeen: Instant = Instant.now) - - final case class CampaignLaunched(namespace: String, updateId: UUID, - devices: Set[UUID], pkgUri: URI, - pkg: PackageId, pkgSize: Long, pkgChecksum: String) + final case class DeviceSeen(namespace: Namespace, + uuid: DeviceId, + lastSeen: Instant = Instant.now) final case class DeviceEventMessage(namespace: Namespace, event: Event) - final case class DeviceMetricsObservation(namespace: Namespace, uuid: DeviceId, payload: Json, receivedAt: Instant) + final case class DeviceMetricsObservation(namespace: Namespace, + uuid: DeviceId, + payload: Json, + receivedAt: Instant) - case class BandwidthUsage(id: UUID, namespace: Namespace, timestamp: Instant, byteCount: Long, - updateType: UpdateType, updateId: String) + case class BandwidthUsage(id: UUID, + namespace: Namespace, + timestamp: Instant, + byteCount: Long, + updateType: UpdateType, + updateId: String) - case class ImageStorageUsage(namespace: Namespace, timestamp: Instant, byteCount: Long) + case class ImageStorageUsage(namespace: Namespace, + timestamp: Instant, + byteCount: Long) sealed trait DeviceUpdateEvent { def namespace: Namespace @@ -140,48 +184,84 @@ object Messages { final case class SystemInfo(product: Option[String]) - final case class DeviceSystemInfoChanged(namespace: Namespace, uuid: DeviceId, newSystemInfo: Option[SystemInfo]) + final case class DeviceSystemInfoChanged(namespace: Namespace, + uuid: DeviceId, + newSystemInfo: Option[SystemInfo]) - final case class CommitManifestUpdated(namespace: Namespace, commit: Commit, releaseBranch: String, - metaUpdaterVersion: String, receivedAt: Instant = Instant.now()) + final case class CommitManifestUpdated(namespace: Namespace, + commit: Commit, + releaseBranch: String, + metaUpdaterVersion: String, + receivedAt: Instant = Instant.now()) - final case class AktualizrConfigChanged(namespace: Namespace, uuid: DeviceId, pollingSec: Int, - secondaryPreinstallWaitSec: Option[Int], forceInstallCompletion: Boolean, - installerType: String, receivedAt: Instant) + final case class AktualizrConfigChanged( + namespace: Namespace, + uuid: DeviceId, + pollingSec: Int, + secondaryPreinstallWaitSec: Option[Int], + forceInstallCompletion: Boolean, + installerType: String, + receivedAt: Instant) - final case class DeleteDeviceRequest(namespace: Namespace, uuid: DeviceId, timestamp: Instant = Instant.now()) + final case class DeleteDeviceRequest(namespace: Namespace, + uuid: DeviceId, + timestamp: Instant = Instant.now()) sealed trait EcuReplacement { val deviceUuid: DeviceId val eventTime: Instant } case class EcuAndHardwareId(ecuId: EcuIdentifier, hardwareId: String) - final case class EcuReplaced(deviceUuid: DeviceId, former: EcuAndHardwareId, current: EcuAndHardwareId, eventTime: Instant = Instant.now) extends EcuReplacement - final case class EcuReplacementFailed(deviceUuid: DeviceId, eventTime: Instant = Instant.now) extends EcuReplacement - - implicit val deviceSystemInfoChangedMessageLike: com.advancedtelematic.libats.messaging_datatype.MessageLike[com.advancedtelematic.libats.messaging_datatype.Messages.DeviceSystemInfoChanged] = MessageLike.derive[DeviceSystemInfoChanged](_.uuid.toString) - - implicit val commitManifestUpdatedMessageLike: com.advancedtelematic.libats.messaging_datatype.MessageLike[com.advancedtelematic.libats.messaging_datatype.Messages.CommitManifestUpdated] = MessageLike.derive[CommitManifestUpdated](_.commit.value) - - implicit val aktualizrConfigChangedMessageLike: com.advancedtelematic.libats.messaging_datatype.MessageLike[com.advancedtelematic.libats.messaging_datatype.Messages.AktualizrConfigChanged] = MessageLike.derive[AktualizrConfigChanged](_.uuid.toString) - - implicit val userCreatedMessageLike: com.advancedtelematic.libats.messaging_datatype.MessageLike[com.advancedtelematic.libats.messaging_datatype.Messages.UserCreated] = MessageLike[UserCreated](_.id) - - implicit val deviceSeenMessageLike: com.advancedtelematic.libats.messaging_datatype.MessageLike[com.advancedtelematic.libats.messaging_datatype.Messages.DeviceSeen] = MessageLike.derive[DeviceSeen](_.uuid.toString) - - implicit val campaignLaunchedMessageLike: com.advancedtelematic.libats.messaging_datatype.MessageLike[com.advancedtelematic.libats.messaging_datatype.Messages.CampaignLaunched] = MessageLike[CampaignLaunched](_.updateId.toString) - - implicit val bandwidthUsageMessageLike: com.advancedtelematic.libats.messaging_datatype.MessageLike[com.advancedtelematic.libats.messaging_datatype.Messages.BandwidthUsage] = MessageLike.derive[BandwidthUsage](_.id.toString) - - implicit val imageStorageMessageLike: com.advancedtelematic.libats.messaging_datatype.MessageLike[com.advancedtelematic.libats.messaging_datatype.Messages.ImageStorageUsage] = MessageLike.derive[ImageStorageUsage](_.namespace.get) - - implicit val deviceEventMessageType: com.advancedtelematic.libats.messaging_datatype.MessageLike[com.advancedtelematic.libats.messaging_datatype.Messages.DeviceEventMessage] = MessageLike[DeviceEventMessage](_.namespace.get) - - implicit val deviceUpdateEventMessageLike: com.advancedtelematic.libats.messaging_datatype.MessageLike[com.advancedtelematic.libats.messaging_datatype.Messages.DeviceUpdateEvent] = MessageLike[DeviceUpdateEvent](_.namespace.get) - - implicit val deleteDeviceRequestMessageLike: com.advancedtelematic.libats.messaging_datatype.MessageLike[com.advancedtelematic.libats.messaging_datatype.Messages.DeleteDeviceRequest] = MessageLike.derive[DeleteDeviceRequest](_.uuid.show) - - implicit val ecuReplacementMsgLike: com.advancedtelematic.libats.messaging_datatype.MessageLike[com.advancedtelematic.libats.messaging_datatype.Messages.EcuReplacement] = MessageLike[EcuReplacement](_.deviceUuid.show) - - implicit val deviceMetricsObservationMessageLike: com.advancedtelematic.libats.messaging_datatype.MessageLike[com.advancedtelematic.libats.messaging_datatype.Messages.DeviceMetricsObservation] = MessageLike.derive[DeviceMetricsObservation](_.uuid.uuid.toString) + final case class EcuReplaced(deviceUuid: DeviceId, + former: EcuAndHardwareId, + current: EcuAndHardwareId, + eventTime: Instant = Instant.now) + extends EcuReplacement + final case class EcuReplacementFailed(deviceUuid: DeviceId, + eventTime: Instant = Instant.now) + extends EcuReplacement + + implicit val deviceSystemInfoChangedMessageLike + : MessageLike[Messages.DeviceSystemInfoChanged] = + MessageLike.derive[DeviceSystemInfoChanged](_.uuid.toString) + + implicit val commitManifestUpdatedMessageLike + : MessageLike[Messages.CommitManifestUpdated] = + MessageLike.derive[CommitManifestUpdated](_.commit.value) + + implicit val aktualizrConfigChangedMessageLike + : MessageLike[Messages.AktualizrConfigChanged] = + MessageLike.derive[AktualizrConfigChanged](_.uuid.toString) + + implicit val userCreatedMessageLike: MessageLike[Messages.UserCreated] = + MessageLike[UserCreated](_.id) + + implicit val deviceSeenMessageLike: MessageLike[Messages.DeviceSeen] = + MessageLike.derive[DeviceSeen](_.uuid.toString) + + implicit val bandwidthUsageMessageLike: MessageLike[Messages.BandwidthUsage] = + MessageLike.derive[BandwidthUsage](_.id.toString) + + implicit val imageStorageMessageLike + : MessageLike[Messages.ImageStorageUsage] = + MessageLike.derive[ImageStorageUsage](_.namespace.get) + + implicit val deviceEventMessageType + : MessageLike[Messages.DeviceEventMessage] = + MessageLike[DeviceEventMessage](_.namespace.get) + + implicit val deviceUpdateEventMessageLike + : MessageLike[Messages.DeviceUpdateEvent] = + MessageLike[DeviceUpdateEvent](_.deviceUuid.uuid.toString) + + implicit val deleteDeviceRequestMessageLike + : MessageLike[Messages.DeleteDeviceRequest] = + MessageLike.derive[DeleteDeviceRequest](_.uuid.show) + + implicit val ecuReplacementMsgLike: MessageLike[Messages.EcuReplacement] = + MessageLike[EcuReplacement](_.deviceUuid.show) + + implicit val deviceMetricsObservationMessageLike + : MessageLike[Messages.DeviceMetricsObservation] = + MessageLike.derive[DeviceMetricsObservation](_.uuid.uuid.toString) }