**feat(metadata): Add PDS fleet management, heartbeat tracking, and submission provenance**

- Introduced schema for `pds_node`, `pds_heartbeat_log`, and `pds_fleet_config` tables to enable fleet management, status tracking, and configuration.
- Added `pds_submission` table for tracking submission provenance, including haplogroup calls and variant proposals.
- Implemented `PdsFleetService` for fleet operations like node registration, heartbeat processing, and fleet summaries.
- Created models, repositories, and services for managing PDS metadata entities and interactions.
- Developed unit tests to validate service logic, repository operations, and heartbeat handling.

Author: JamesKane

app/models/dal/MetadataSchema.scala

@@ -2,12 +2,18 @@ package models.dal
 
 import models.PDSRegistration
 import models.dal.MyPostgresProfile.api.*
+import models.dal.domain.pds.*
 import slick.lifted.ProvenShape
 
 import java.time.ZonedDateTime
 
 object MetadataSchema {
 
+  val pdsNodes = TableQuery[PdsNodeTable]
+  val pdsHeartbeatLogs = TableQuery[PdsHeartbeatLogTable]
+  val pdsFleetConfigs = TableQuery[PdsFleetConfigTable]
+  val pdsSubmissions = TableQuery[PdsSubmissionTable]
+
   class PDSRegistrationsTable(tag: Tag) extends Table[PDSRegistration](tag, "pds_registrations") {
     def did = column[String]("did", O.PrimaryKey)
 

app/models/dal/domain/pds/PdsFleetConfigTable.scala

@@ -0,0 +1,19 @@
+package models.dal.domain.pds
+
+import models.dal.MyPostgresProfile.api.*
+import models.domain.pds.PdsFleetConfig
+
+import java.time.LocalDateTime
+
+class PdsFleetConfigTable(tag: Tag) extends Table[PdsFleetConfig](tag, "pds_fleet_config") {
+  def id = column[Int]("id", O.PrimaryKey, O.AutoInc)
+  def configKey = column[String]("config_key")
+  def configValue = column[String]("config_value")
+  def description = column[Option[String]]("description")
+  def updatedBy = column[Option[String]]("updated_by")
+  def updatedAt = column[LocalDateTime]("updated_at")
+
+  def * = (id.?, configKey, configValue, description, updatedBy, updatedAt).mapTo[PdsFleetConfig]
+
+  def uniqueKey = index("idx_pds_fleet_config_key_unique", configKey, unique = true)
+}

app/models/dal/domain/pds/PdsHeartbeatLogTable.scala

@@ -0,0 +1,24 @@
+package models.dal.domain.pds
+
+import models.dal.MyPostgresProfile.api.*
+import models.domain.pds.PdsHeartbeatLog
+import play.api.libs.json.JsValue
+
+import java.time.LocalDateTime
+
+class PdsHeartbeatLogTable(tag: Tag) extends Table[PdsHeartbeatLog](tag, "pds_heartbeat_log") {
+  def id = column[Int]("id", O.PrimaryKey, O.AutoInc)
+  def pdsNodeId = column[Int]("pds_node_id")
+  def status = column[String]("status")
+  def softwareVersion = column[Option[String]]("software_version")
+  def loadMetrics = column[Option[JsValue]]("load_metrics")
+  def processingQueueSize = column[Option[Int]]("processing_queue_size")
+  def errorMessage = column[Option[String]]("error_message")
+  def recordedAt = column[LocalDateTime]("recorded_at")
+
+  def * = (
+    id.?, pdsNodeId, status, softwareVersion, loadMetrics, processingQueueSize, errorMessage, recordedAt
+  ).mapTo[PdsHeartbeatLog]
+
+  def nodeFk = foreignKey("pds_heartbeat_log_node_fk", pdsNodeId, TableQuery[PdsNodeTable])(_.id)
+}

app/models/dal/domain/pds/PdsNodeTable.scala

@@ -0,0 +1,32 @@
+package models.dal.domain.pds
+
+import models.dal.MyPostgresProfile.api.*
+import models.domain.pds.PdsNode
+import play.api.libs.json.JsValue
+
+import java.time.LocalDateTime
+
+class PdsNodeTable(tag: Tag) extends Table[PdsNode](tag, "pds_node") {
+  def id = column[Int]("id", O.PrimaryKey, O.AutoInc)
+  def did = column[String]("did")
+  def pdsUrl = column[String]("pds_url")
+  def handle = column[Option[String]]("handle")
+  def nodeName = column[Option[String]]("node_name")
+  def softwareVersion = column[Option[String]]("software_version")
+  def status = column[String]("status")
+  def capabilities = column[JsValue]("capabilities")
+  def lastHeartbeat = column[Option[LocalDateTime]]("last_heartbeat")
+  def lastCommitCid = column[Option[String]]("last_commit_cid")
+  def lastCommitRev = column[Option[String]]("last_commit_rev")
+  def ipAddress = column[Option[String]]("ip_address")
+  def osInfo = column[Option[String]]("os_info")
+  def createdAt = column[LocalDateTime]("created_at")
+  def updatedAt = column[LocalDateTime]("updated_at")
+
+  def * = (
+    id.?, did, pdsUrl, handle, nodeName, softwareVersion, status, capabilities,
+    lastHeartbeat, lastCommitCid, lastCommitRev, ipAddress, osInfo, createdAt, updatedAt
+  ).mapTo[PdsNode]
+
+  def uniqueDid = index("idx_pds_node_did_unique", did, unique = true)
+}

app/models/dal/domain/pds/PdsSubmissionTable.scala

@@ -0,0 +1,36 @@
+package models.dal.domain.pds
+
+import models.dal.MyPostgresProfile.api.*
+import models.domain.pds.PdsSubmission
+import play.api.libs.json.JsValue
+
+import java.time.LocalDateTime
+import java.util.UUID
+
+class PdsSubmissionTable(tag: Tag) extends Table[PdsSubmission](tag, "pds_submission") {
+  def id = column[Int]("id", O.PrimaryKey, O.AutoInc)
+  def pdsNodeId = column[Int]("pds_node_id")
+  def submissionType = column[String]("submission_type")
+  def biosampleId = column[Option[Int]]("biosample_id")
+  def biosampleGuid = column[Option[UUID]]("biosample_guid")
+  def proposedValue = column[String]("proposed_value")
+  def confidenceScore = column[Option[Double]]("confidence_score")
+  def algorithmVersion = column[Option[String]]("algorithm_version")
+  def softwareVersion = column[Option[String]]("software_version")
+  def payload = column[Option[JsValue]]("payload")
+  def status = column[String]("status")
+  def reviewedBy = column[Option[String]]("reviewed_by")
+  def reviewedAt = column[Option[LocalDateTime]]("reviewed_at")
+  def reviewNotes = column[Option[String]]("review_notes")
+  def atUri = column[Option[String]]("at_uri")
+  def atCid = column[Option[String]]("at_cid")
+  def createdAt = column[LocalDateTime]("created_at")
+
+  def * = (
+    id.?, pdsNodeId, submissionType, biosampleId, biosampleGuid, proposedValue,
+    confidenceScore, algorithmVersion, softwareVersion, payload, status,
+    reviewedBy, reviewedAt, reviewNotes, atUri, atCid, createdAt
+  ).mapTo[PdsSubmission]
+
+  def nodeFk = foreignKey("pds_submission_node_fk", pdsNodeId, TableQuery[PdsNodeTable])(_.id)
+}

app/models/domain/pds/PdsNode.scala

@@ -0,0 +1,73 @@
+package models.domain.pds
+
+import play.api.libs.json.{Json, JsValue, OFormat}
+
+import java.time.LocalDateTime
+
+case class PdsNode(
+                    id: Option[Int] = None,
+                    did: String,
+                    pdsUrl: String,
+                    handle: Option[String] = None,
+                    nodeName: Option[String] = None,
+                    softwareVersion: Option[String] = None,
+                    status: String = "UNKNOWN",
+                    capabilities: JsValue = Json.obj(),
+                    lastHeartbeat: Option[LocalDateTime] = None,
+                    lastCommitCid: Option[String] = None,
+                    lastCommitRev: Option[String] = None,
+                    ipAddress: Option[String] = None,
+                    osInfo: Option[String] = None,
+                    createdAt: LocalDateTime = LocalDateTime.now(),
+                    updatedAt: LocalDateTime = LocalDateTime.now()
+                  )
+
+object PdsNode {
+  implicit val format: OFormat[PdsNode] = Json.format[PdsNode]
+
+  val ValidStatuses: Set[String] = Set("ONLINE", "OFFLINE", "BUSY", "ERROR", "UNKNOWN")
+}
+
+case class PdsHeartbeatLog(
+                            id: Option[Int] = None,
+                            pdsNodeId: Int,
+                            status: String,
+                            softwareVersion: Option[String] = None,
+                            loadMetrics: Option[JsValue] = None,
+                            processingQueueSize: Option[Int] = Some(0),
+                            errorMessage: Option[String] = None,
+                            recordedAt: LocalDateTime = LocalDateTime.now()
+                          )
+
+object PdsHeartbeatLog {
+  implicit val format: OFormat[PdsHeartbeatLog] = Json.format[PdsHeartbeatLog]
+}
+
+case class PdsFleetConfig(
+                           id: Option[Int] = None,
+                           configKey: String,
+                           configValue: String,
+                           description: Option[String] = None,
+                           updatedBy: Option[String] = None,
+                           updatedAt: LocalDateTime = LocalDateTime.now()
+                         )
+
+object PdsFleetConfig {
+  implicit val format: OFormat[PdsFleetConfig] = Json.format[PdsFleetConfig]
+}
+
+case class PdsFleetSummary(
+                            totalNodes: Int,
+                            onlineNodes: Int,
+                            offlineNodes: Int,
+                            busyNodes: Int,
+                            errorNodes: Int,
+                            unknownNodes: Int,
+                            targetVersion: Option[String],
+                            nodesOnTargetVersion: Int,
+                            nodesOutdated: Int
+                          )
+
+object PdsFleetSummary {
+  implicit val format: OFormat[PdsFleetSummary] = Json.format[PdsFleetSummary]
+}

app/models/domain/pds/PdsSubmission.scala

@@ -0,0 +1,47 @@
+package models.domain.pds
+
+import play.api.libs.json.{JsValue, Json, OFormat}
+
+import java.time.LocalDateTime
+import java.util.UUID
+
+case class PdsSubmission(
+                          id: Option[Int] = None,
+                          pdsNodeId: Int,
+                          submissionType: String,
+                          biosampleId: Option[Int] = None,
+                          biosampleGuid: Option[UUID] = None,
+                          proposedValue: String,
+                          confidenceScore: Option[Double] = None,
+                          algorithmVersion: Option[String] = None,
+                          softwareVersion: Option[String] = None,
+                          payload: Option[JsValue] = None,
+                          status: String = "PENDING",
+                          reviewedBy: Option[String] = None,
+                          reviewedAt: Option[LocalDateTime] = None,
+                          reviewNotes: Option[String] = None,
+                          atUri: Option[String] = None,
+                          atCid: Option[String] = None,
+                          createdAt: LocalDateTime = LocalDateTime.now()
+                        )
+
+object PdsSubmission {
+  implicit val format: OFormat[PdsSubmission] = Json.format[PdsSubmission]
+
+  val ValidTypes: Set[String] = Set("HAPLOGROUP_CALL", "VARIANT_CALL", "BRANCH_PROPOSAL", "PRIVATE_VARIANT", "STR_PROFILE")
+  val ValidStatuses: Set[String] = Set("PENDING", "ACCEPTED", "REJECTED", "SUPERSEDED")
+}
+
+case class SubmissionSummary(
+                              pdsNodeId: Int,
+                              did: String,
+                              totalSubmissions: Int,
+                              pendingCount: Int,
+                              acceptedCount: Int,
+                              rejectedCount: Int,
+                              acceptanceRate: Double
+                            )
+
+object SubmissionSummary {
+  implicit val format: OFormat[SubmissionSummary] = Json.format[SubmissionSummary]
+}

app/modules/BaseModule.scala

@@ -178,5 +178,22 @@ class BaseModule extends AbstractModule {
     bind(classOf[GroupProjectMemberRepository])
       .to(classOf[GroupProjectMemberRepositoryImpl])
       .asEagerSingleton()
+
+    // PDS Fleet Management
+    bind(classOf[PdsNodeRepository])
+      .to(classOf[PdsNodeRepositoryImpl])
+      .asEagerSingleton()
+
+    bind(classOf[PdsHeartbeatLogRepository])
+      .to(classOf[PdsHeartbeatLogRepositoryImpl])
+      .asEagerSingleton()
+
+    bind(classOf[PdsFleetConfigRepository])
+      .to(classOf[PdsFleetConfigRepositoryImpl])
+      .asEagerSingleton()
+
+    bind(classOf[PdsSubmissionRepository])
+      .to(classOf[PdsSubmissionRepositoryImpl])
+      .asEagerSingleton()
   }
 }