Class ModuleRepository

java.lang.Object
ai.pipestream.registration.repository.ModuleRepository
@ApplicationScoped public class ModuleRepository extends Object
Repository for managing service module registrations in PostgreSQL. This is the primary data store (system of record).

Blocking Hibernate ORM (Panache active-record) over JDBC. Write methods are @Transactional; reads run without a transaction. Callers invoke these from virtual threads, so the blocking JDBC work never pins the event loop.

  • Constructor Details

    • ModuleRepository

      public ModuleRepository()
      Creates the repository bean. The Apicurio client used for schema sync is supplied by CDI field injection.

  • Method Details

    • registerModule

      public ServiceModule registerModule(String serviceName, String host, int port, String version, Map<String,Object> metadata, String jsonSchema)
      Register a new service module with optional schema. Updates existing module if it already exists.
      Parameters:
      serviceName - the service name
      host - the advertised host
      port - the advertised port
      version - the module version
      metadata - free-form registration metadata
      jsonSchema - the config JSON schema to persist, or null/blank to skip schema handling
      Returns:
      the created or updated module entity

    • saveSchema

      public ConfigSchema saveSchema(String serviceName, String version, String jsonSchema)
      Save a configuration schema (dual storage: PostgreSQL + Apicurio).
      Parameters:
      serviceName - the service name
      version - the schema version
      jsonSchema - the JSON schema content
      Returns:
      the persisted schema entity; its sync status reflects whether the Apicurio sync succeeded

    • updateHeartbeat

      public ServiceModule updateHeartbeat(String serviceId)
      Update heartbeat for a service.
      Parameters:
      serviceId - the service ID to refresh
      Returns:
      the updated module, or null when no module with that ID exists

    • markUnhealthy

      public ServiceModule markUnhealthy(String serviceId)
      Mark service as unhealthy.
      Parameters:
      serviceId - the service ID to mark
      Returns:
      the updated module, or null when no module with that ID exists

    • unregisterModule

      public boolean unregisterModule(String serviceId)
      Unregister a service.
      Parameters:
      serviceId - the service ID to remove
      Returns:
      true if a module was found and deleted, false otherwise

    • getActiveServices

      public List<ServiceModule> getActiveServices()
      Get all active services.
      Returns:
      all modules whose status is ServiceStatus.ACTIVE

    • getAllServices

      public List<ServiceModule> getAllServices()
      Get all services (for admin dashboard).
      Returns:
      every registered module regardless of status

    • findStaleServices

      public List<ServiceModule> findStaleServices()
      Find stale services (no heartbeat for > 30 seconds).
      Returns:
      active modules whose last heartbeat is older than 30 seconds

    • findById

      public ServiceModule findById(String serviceId)
      Get service by ID.
      Parameters:
      serviceId - the service ID to look up
      Returns:
      the module, or null when none exists with that ID

    • findSchemaById

      public ConfigSchema findSchemaById(String schemaId)
      Get schema by ID.
      Parameters:
      schemaId - the schema ID to look up
      Returns:
      the schema, or null when none exists with that ID

    • upsertNamedArtifact

      public ConfigSchema upsertNamedArtifact(String artifactName, String version, String json)
      Upsert a NAMED schema artifact (recipe schema, uischema, ...) as a ConfigSchema row keyed by the artifact name — same update-on-change semantics as module config schemas, so SNAPSHOT re-registrations refresh content and mark PENDING for Apicurio re-sync.
      Parameters:
      artifactName - the artifact name used as the schema key
      version - the artifact version
      json - the artifact JSON content
      Returns:
      the existing (possibly updated) or newly created schema row

    • findLatestSchemaByServiceName

      public ConfigSchema findLatestSchemaByServiceName(String serviceName)
      Get the latest schema for a service (most recently created).
      Parameters:
      serviceName - the service name
      Returns:
      the most recently created schema for the service, or null when none exist

    • findCurrentSchemaByServiceName

      public ConfigSchema findCurrentSchemaByServiceName(String serviceName)
      Resolve the schema of the module's CURRENT registration: the instance with the freshest heartbeat wins. "Latest by createdAt" is wrong once historical version rows exist (e.g. a branch-suffixed snapshot version registered after the canonical one shadows it forever).
      Parameters:
      serviceName - the service name
      Returns:
      the config schema of the freshest-heartbeat module instance, or null when none resolves

    • findSchemaVersionsByServiceName

      public List<ConfigSchema> findSchemaVersionsByServiceName(String serviceName)
      Get all schema versions for a service name, ordered by version descending.
      Parameters:
      serviceName - the service name
      Returns:
      the service's schemas ordered by version descending

    • findSchemasNeedingSync

      public List<ConfigSchema> findSchemasNeedingSync()
      Get all schemas needing sync to Apicurio.
      Returns:
      schemas whose sync status is PENDING, FAILED, or OUT_OF_SYNC

    • countServicesByStatus

      public Map<ServiceStatus, Long> countServicesByStatus()
      Count registered services by status.
      Returns:
      a map of each ServiceStatus to the number of modules in that status