Source Code Reading: OntologyRuntimeService — Core Entity CRUD

“

Series: S9 Source Code Reading · Article 1 | Level: Advanced | Reading Time: 25 min

#TL;DR

OntologyRuntimeService is the most critical service in coomia-dip's Data Layer (Data Layer), managing the full lifecycle of ontology instances. Built on Quarkus 3.x with gRPC protocol, it supports single and batch CRUD operations with enterprise features like soft delete, version history, and change event emission. This article dissects its layered architecture, WorldContext resolution strategy, Proto-Domain bidirectional conversion, batch operation error aggregation patterns, and change trigger dispatching.

#Table of Contents

Overall Architecture and Layered Design
gRPC Service Layer: OntologyRuntimeGrpcService
WorldContext Resolution: Dual-Source Priority Strategy
Bidirectional Proto-Domain Conversion: ProtoConverter
Single Entity CRUD Implementation
Batch Operations and Partial Failure Semantics
Soft Delete and Version History
Relation Operations: CreateRelation / DeleteRelation / GetRelations
Change Event Trigger: OntologyChangeTriggerDispatcher
Exception Handling: GrpcExceptionHandler Unified Mapping
Key Takeaways

#1. Overall Architecture and Layered Design

The OntologyRuntimeService code is distributed across three packages, following classic hexagonal architecture:

Code

data-Layer/src/main/java/com/onto/data/
├── api/grpc/                          # Inbound adapter (gRPC)
│   ├── OntologyRuntimeGrpcService.java  # gRPC service implementation
│   ├── converter/ProtoConverter.java    # Proto <-> Domain conversion
│   ├── exception/GrpcExceptionHandler.java
│   └── interceptor/WorldContextInterceptor.java
├── service/                           # Application service layer
│   ├── OntologyRuntimeService.java      # Interface definition
│   ├── DefaultOntologyRuntimeService.java
│   └── dto/                            # Data transfer objects
│       ├── BatchCreateResponse.java
│       ├── BatchUpdateResponse.java
│       ├── UpdateMode.java
│       └── UpdateOptions.java
├── domain/instance/                   # Domain model
│   └── OntologyInstance.java
├── repository/                        # Outbound adapter (storage)
│   └── exception/InstanceNotFoundException.java
└── event/                             # Event mechanism
    └── OntologyChangeTriggerDispatcher.java

Design decision: The gRPC service layer contains no business logic — it handles exactly three concerns: (1) extracting WorldContext, (2) Proto to Domain conversion, (3) exception mapping. All business rules live in the OntologyRuntimeService interface and its implementation.

This separation means adding a Flight SQL or REST entry point requires only a new adapter, with zero changes to the core service layer.

#2. gRPC Service Layer: OntologyRuntimeGrpcService

Java

@GrpcService
@Blocking
public class OntologyRuntimeGrpcService
        extends OntologyRuntimeServiceGrpc.OntologyRuntimeServiceImplBase {

    private static final Logger LOG = LoggerFactory.getLogger(OntologyRuntimeGrpcService.class);

    private final OntologyRuntimeService service;
    private final ProtoConverter converter;
    private final GrpcExceptionHandler exceptionHandler;
    private final OntologyChangeTriggerDispatcher triggerDispatcher;

    @Inject
    public OntologyRuntimeGrpcService(
            OntologyRuntimeService service,
            ProtoConverter converter,
            GrpcExceptionHandler exceptionHandler,
            OntologyChangeTriggerDispatcher triggerDispatcher) {
        this.service = service;
        this.converter = converter;
        this.exceptionHandler = exceptionHandler;
        this.triggerDispatcher = triggerDispatcher;
    }
}

Key annotation breakdown:

@GrpcService: Quarkus gRPC extension annotation that auto-registers this class as a gRPC endpoint
@Blocking: Declares all methods execute on the blocking thread pool (not the Vert.x event loop), since the underlying layer involves synchronous JDBC/Doris calls
@Inject: CDI constructor injection with four dependencies, each with a single responsibility

Why @Blocking? Quarkus processes gRPC requests on Vert.x IO threads by default, but OntologyRuntimeService relies on the Doris JDBC connection pool (synchronous blocking). Running on the IO thread would cause BlockingNotAllowedException. @Blocking dispatches requests to the worker thread pool at the cost of context switching, but guarantees thread safety.

#3. WorldContext Resolution: Dual-Source Priority Strategy

WorldContext is the fundamental isolation unit in coomia-dip. Every request must carry a WorldContext; otherwise, the service rejects it.

Java

private WorldContext resolveWorldContext(RequestContext requestContext) {
    // Priority 1: Context Key injected by gRPC Interceptor
    WorldContext fromInterceptor = WorldContextInterceptor.WORLD_CONTEXT_KEY.get();
    if (fromInterceptor != null) {
        return fromInterceptor;
    }

    // Priority 2: RequestContext.world field from the request body
    if (requestContext != null && requestContext.hasWorld()) {
        return WorldContext.of(
            requestContext.getWorld().getWorldId(),
            requestContext.getWorld().getBranch()
        );
    }

    throw new IllegalStateException(
        "WorldContext not found in gRPC context or request"
    );
}

Dual-source strategy rationale:

Source	Scenario	Priority
gRPC Interceptor	SDK clients auto-inject metadata	High
RequestContext field	Manual request construction, testing, cross-gateway calls	Low

When both sources are present, the Interceptor value takes precedence because it executes before the request reaches the service method and has already been authenticated and validated.

#4. Bidirectional Proto-Domain Conversion: ProtoConverter

ProtoConverter is a stateless converter handling bidirectional mapping between Protobuf messages and domain objects:

Java

@ApplicationScoped
public class ProtoConverter {

    public OntologyInstance toDomain(CreateInstanceInput input, WorldContext ctx) {
        OntologyInstance instance = new OntologyInstance();
        instance.setEntityType(input.getEntityType());
        instance.setPrimaryKey(input.hasPrimaryKey()
            ? input.getPrimaryKey()
            : UUID.randomUUID().toString());
        instance.setWorldId(ctx.getWorldId());
        instance.setBranch(ctx.getBranch());

        // Attribute mapping: Proto Struct -> Java Map
        Map<String, Object> attributes = new HashMap<>();
        for (Map.Entry<String, Value> entry :
                input.getAttributesMap().entrySet()) {
            attributes.put(entry.getKey(),
                AttributeValueConverter.fromProto(entry.getValue()));
        }
        instance.setAttributes(attributes);
        return instance;
    }

    public InstanceProto toProto(OntologyInstance instance) {
        InstanceProto.Builder builder = InstanceProto.newBuilder()
            .setInstanceId(instance.getId())
            .setEntityType(instance.getEntityType())
            .setPrimaryKey(instance.getPrimaryKey())
            .setVersion(instance.getVersion())
            .setCreatedAt(toTimestamp(instance.getCreatedAt()))
            .setUpdatedAt(toTimestamp(instance.getUpdatedAt()));

        if (instance.isDeleted()) {
            builder.setDeletedAt(toTimestamp(instance.getDeletedAt()));
        }

        // Attribute mapping: Java Map -> Proto Struct
        for (Map.Entry<String, Object> entry :
                instance.getAttributes().entrySet()) {
            builder.putAttributes(entry.getKey(),
                AttributeValueConverter.toProto(entry.getValue()));
        }
        return builder.build();
    }
}

Design highlights:

Primary key strategy: If the client does not provide a primaryKey, a UUID is auto-generated — supporting both business keys (e.g., employee number) and system keys
AttributeValueConverter: Handles recursive type mapping between Proto Value and Java Object, supporting null, string, number, bool, list, and map types
Version number propagation: toProto always outputs the version field, supporting optimistic locking scenarios

#5. Single Entity CRUD Implementation

#5.1 Create Instance

Java

@Override
public void createInstance(CreateInstanceRequest request,
        StreamObserver<CreateInstanceResponse> responseObserver) {
    try {
        WorldContext ctx = resolveWorldContext(request.getContext());
        OntologyInstance domain = converter.toDomain(
            request.getInstance(), ctx);

        OntologyInstance created = service.createInstance(domain, ctx);

        // Emit change event
        triggerDispatcher.dispatch(
            ChangeType.CREATE, created.getEntityType(),
            created.getId(), ctx);

        CreateInstanceResponse response = CreateInstanceResponse.newBuilder()
            .setInstance(converter.toProto(created))
            .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    } catch (Exception e) {
        exceptionHandler.handle(e, responseObserver);
    }
}

Execution flow:

Code

Client -> gRPC -> resolveWorldContext -> converter.toDomain
  -> service.createInstance (business logic + storage)
  -> triggerDispatcher.dispatch (async event)
  -> converter.toProto -> responseObserver.onNext

Note that triggerDispatcher.dispatch is called after successful creation but before the response. This means event emission failure will not roll back the creation, but will log an error. This is a classic "eventual consistency" choice.

#5.2 Get Instance

Java

@Override
public void getInstance(GetInstanceRequest request,
        StreamObserver<GetInstanceResponse> responseObserver) {
    try {
        WorldContext ctx = resolveWorldContext(request.getContext());

        Optional<OntologyInstance> found = service.getInstance(
            request.getEntityType(),
            request.getInstanceId(),
            ctx
        );

        if (found.isEmpty()) {
            throw new InstanceNotFoundException(
                request.getEntityType(),
                request.getInstanceId()
            );
        }

        GetInstanceResponse response = GetInstanceResponse.newBuilder()
            .setInstance(converter.toProto(found.get()))
            .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    } catch (Exception e) {
        exceptionHandler.handle(e, responseObserver);
    }
}

Note: The code uses explicit isEmpty() check rather than Optional.orElseThrow(). This is because InstanceNotFoundException requires both entityType and instanceId parameters for error message construction, making the lambda version more verbose.

#5.3 Update Instance

Java

@Override
public void updateInstance(UpdateInstanceRequest request,
        StreamObserver<UpdateInstanceResponse> responseObserver) {
    try {
        WorldContext ctx = resolveWorldContext(request.getContext());
        UpdateInstanceInput input = request.getInstance();

        UpdateOptions options = UpdateOptions.builder()
            .mode(input.hasUpdateMode()
                ? UpdateMode.valueOf(input.getUpdateMode().name())
                : UpdateMode.MERGE)
            .expectedVersion(input.hasExpectedVersion()
                ? input.getExpectedVersion()
                : null)
            .build();

        OntologyInstance updated = service.updateInstance(
            input.getEntityType(),
            input.getInstanceId(),
            converter.toAttributeMap(input.getAttributesMap()),
            options,
            ctx
        );

        triggerDispatcher.dispatch(
            ChangeType.UPDATE, updated.getEntityType(),
            updated.getId(), ctx);

        UpdateInstanceResponse response = UpdateInstanceResponse.newBuilder()
            .setInstance(converter.toProto(updated))
            .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    } catch (Exception e) {
        exceptionHandler.handle(e, responseObserver);
    }
}

Two update modes:

Mode	Behavior	Use Case
`MERGE`	Updates only provided attributes, preserves others	Incremental update
`REPLACE`	Completely replaces existing attributes with provided ones	Full overwrite

Optimistic locking: expectedVersion is optional. When provided, the service layer compares the current version number, throwing OptimisticLockException on mismatch, which GrpcExceptionHandler maps to Status.ABORTED.

#6. Batch Operations and Partial Failure Semantics

Batch operations are a critical feature of coomia-dip. Unlike single-entity operations with "all-or-nothing" semantics, batch operations use partial success semantics:

Java

@Override
public void batchCreateInstances(BatchCreateInstancesRequest request,
        StreamObserver<BatchCreateInstancesResponse> responseObserver) {
    try {
        WorldContext ctx = resolveWorldContext(request.getContext());

        List<OntologyInstance> domainList = request.getInstancesList()
            .stream()
            .map(input -> converter.toDomain(input, ctx))
            .collect(Collectors.toList());

        BatchCreateResponse batchResult =
            service.batchCreateInstances(domainList, ctx);

        BatchCreateInstancesResponse.Builder responseBuilder =
            BatchCreateInstancesResponse.newBuilder()
                .setTotalRequested(request.getInstancesCount())
                .setSuccessCount(batchResult.getSuccessful().size())
                .setFailureCount(batchResult.getErrors().size());

        // Successful instances
        for (OntologyInstance created : batchResult.getSuccessful()) {
            responseBuilder.addInstances(converter.toProto(created));
            triggerDispatcher.dispatch(
                ChangeType.CREATE, created.getEntityType(),
                created.getId(), ctx);
        }

        // Failed records
        for (BatchOperationError error : batchResult.getErrors()) {
            responseBuilder.addErrors(
                BatchError.newBuilder()
                    .setIndex(error.getIndex())
                    .setMessage(error.getMessage())
                    .setCode(error.getCode())
                    .build()
            );
        }

        responseObserver.onNext(responseBuilder.build());
        responseObserver.onCompleted();
    } catch (Exception e) {
        exceptionHandler.handle(e, responseObserver);
    }
}

Partial failure response structure:

PROTOBUF

message BatchCreateInstancesResponse {
    int32 total_requested = 1;
    int32 success_count = 2;
    int32 failure_count = 3;
    repeated InstanceProto instances = 4;     // Successful ones
    repeated BatchError errors = 5;            // Failed ones with index
}

Design decision: Each BatchError includes the index from the original request, allowing clients to pinpoint which records failed. This is far more suitable for bulk import scenarios than full rollback — succeeding on 9999 records and failing on 1 does not require retrying everything.

#7. Soft Delete and Version History

#7.1 Soft Delete

Java

@Override
public void deleteInstance(DeleteInstanceRequest request,
        StreamObserver<DeleteInstanceResponse> responseObserver) {
    try {
        WorldContext ctx = resolveWorldContext(request.getContext());

        boolean deleted = service.deleteInstance(
            request.getEntityType(),
            request.getInstanceId(),
            request.getHardDelete(),  // Physical delete flag
            ctx
        );

        triggerDispatcher.dispatch(
            ChangeType.DELETE,
            request.getEntityType(),
            request.getInstanceId(),
            ctx);

        DeleteInstanceResponse response = DeleteInstanceResponse.newBuilder()
            .setDeleted(deleted)
            .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    } catch (Exception e) {
        exceptionHandler.handle(e, responseObserver);
    }
}

By default, deleteInstance performs a soft delete: setting the deleted_at timestamp. The instance becomes invisible in normal queries but remains accessible in audit queries and history traversal. Only explicitly passing hardDelete=true triggers physical deletion.

#7.2 Version History Query

Java

@Override
public void getInstanceHistory(GetInstanceHistoryRequest request,
        StreamObserver<GetInstanceHistoryResponse> responseObserver) {
    try {
        WorldContext ctx = resolveWorldContext(request.getContext());

        List<OntologyInstance> history = service.getInstanceHistory(
            request.getEntityType(),
            request.getInstanceId(),
            request.getMaxVersions(),
            ctx
        );

        GetInstanceHistoryResponse.Builder builder =
            GetInstanceHistoryResponse.newBuilder();
        for (OntologyInstance version : history) {
            builder.addVersions(converter.toProto(version));
        }

        responseObserver.onNext(builder.build());
        responseObserver.onCompleted();
    } catch (Exception e) {
        exceptionHandler.handle(e, responseObserver);
    }
}

Version history queries leverage Nessie's Git-like version management, where each update creates a new commit. The maxVersions parameter limits the returned version count, preventing memory overflow from full history of large objects.

There is also a method for exact version retrieval:

Java

@Override
public void getInstanceAtVersion(GetInstanceAtVersionRequest request,
        StreamObserver<GetInstanceResponse> responseObserver) {
    try {
        WorldContext ctx = resolveWorldContext(request.getContext());

        Optional<OntologyInstance> found = service.getInstanceAtVersion(
            request.getEntityType(),
            request.getInstanceId(),
            request.getVersion(),
            ctx
        );
        // ...
    } catch (Exception e) {
        exceptionHandler.handle(e, responseObserver);
    }
}

This supports "time travel" queries — viewing an entity's state at any historical point.

#8. Relation Operations: CreateRelation / DeleteRelation / GetRelations

Relations are a core concept in the ontology model. Entities connect through relations to form a knowledge graph.

Java

@Override
public void createRelation(CreateRelationRequest request,
        StreamObserver<CreateRelationResponse> responseObserver) {
    try {
        WorldContext ctx = resolveWorldContext(request.getContext());

        Relation relation = converter.toRelationDomain(
            request.getRelation(), ctx);
        Relation created = service.createRelation(relation, ctx);

        triggerDispatcher.dispatch(
            ChangeType.RELATION_CREATE,
            relation.getRelationType(),
            created.getId(), ctx);

        responseObserver.onNext(
            CreateRelationResponse.newBuilder()
                .setRelation(converter.toRelationProto(created))
                .build()
        );
        responseObserver.onCompleted();
    } catch (Exception e) {
        exceptionHandler.handle(e, responseObserver);
    }
}

@Override
public void getRelations(GetRelationsRequest request,
        StreamObserver<GetRelationsResponse> responseObserver) {
    try {
        WorldContext ctx = resolveWorldContext(request.getContext());

        List<Relation> relations = service.getRelations(
            request.getEntityType(),
            request.getInstanceId(),
            request.hasRelationType()
                ? request.getRelationType() : null,
            request.hasDirection()
                ? Direction.valueOf(request.getDirection().name())
                : Direction.BOTH,
            ctx
        );

        GetRelationsResponse.Builder builder =
            GetRelationsResponse.newBuilder();
        for (Relation rel : relations) {
            builder.addRelations(converter.toRelationProto(rel));
        }

        responseObserver.onNext(builder.build());
        responseObserver.onCompleted();
    } catch (Exception e) {
        exceptionHandler.handle(e, responseObserver);
    }
}

Relation query directionality: The Direction enum supports three values: OUTGOING (from this entity), INCOMING (toward this entity), and BOTH (bidirectional). The underlying storage in the graph database (TuGraph) uses directed edges, but queries can traverse in reverse.

#9. Change Event Trigger: OntologyChangeTriggerDispatcher

Java

@ApplicationScoped
public class OntologyChangeTriggerDispatcher {

    @Inject
    Event<OntologyChangeEvent> changeEventBus;

    @Inject
    SubscriptionEventRouter subscriptionRouter;

    public void dispatch(ChangeType type, String entityType,
            String instanceId, WorldContext ctx) {
        OntologyChangeEvent event = OntologyChangeEvent.builder()
            .changeType(type)
            .entityType(entityType)
            .instanceId(instanceId)
            .worldId(ctx.getWorldId())
            .branch(ctx.getBranch())
            .timestamp(Instant.now())
            .build();

        // 1. CDI Event (in-process synchronous)
        changeEventBus.fire(event);

        // 2. Subscription Router (Kafka asynchronous)
        subscriptionRouter.route(event);

        LOG.debugf("Dispatched %s event for %s/%s",
            type, entityType, instanceId);
    }
}

Events flow through two channels:

CDI Event: In-process synchronous broadcast for triggering derived property recomputation, cache invalidation, and other local operations
Subscription Router: Sends events to Kafka, consumed by SubscriptionService for delivery to external subscribers

This dual-channel design ensures both immediacy for local side effects and reliability for cross-process notifications.

#10. Exception Handling: GrpcExceptionHandler Unified Mapping

Java

@ApplicationScoped
public class GrpcExceptionHandler {

    private static final Map<Class<? extends Exception>, Status.Code> EXCEPTION_MAP =
        Map.of(
            InstanceNotFoundException.class, Status.Code.NOT_FOUND,
            DuplicateKeyException.class, Status.Code.ALREADY_EXISTS,
            OptimisticLockException.class, Status.Code.ABORTED,
            SchemaValidationException.class, Status.Code.INVALID_ARGUMENT,
            WorldContextMissingException.class, Status.Code.UNAUTHENTICATED,
            QuotaExceededException.class, Status.Code.RESOURCE_EXHAUSTED
        );

    public <T> void handle(Exception e,
            StreamObserver<T> responseObserver) {
        Status.Code code = EXCEPTION_MAP.getOrDefault(
            e.getClass(), Status.Code.INTERNAL);

        LOG.errorf(e, "gRPC error [%s]: %s", code, e.getMessage());

        responseObserver.onError(
            Status.fromCode(code)
                .withDescription(e.getMessage())
                .withCause(e)
                .asRuntimeException()
        );
    }
}

Mapping strategy: Uses a static Map instead of an if-else chain — adding new exception types requires only one line. Unmatched exceptions default to INTERNAL to avoid leaking sensitive information.

#11. Key Takeaways

Hexagonal architecture layering: The gRPC layer handles only protocol adaptation (WorldContext resolution + Proto conversion + exception mapping); business logic lives in the Service layer
Dual-source WorldContext resolution: Interceptor takes priority over Request fields, balancing security and flexibility
Batch partial success: Returns success list + error list (with index), more suitable for data import than full rollback
Soft delete by default: Default soft delete preserves audit capabilities; hardDelete parameter controls physical deletion
Dual-channel change events: CDI Event ensures local immediacy; Kafka ensures cross-process reliable delivery
Exception mapping table: Declarative exception mapping via static Map provides extensibility and readability
@Blocking annotation: Required because the underlying Doris JDBC is synchronous; dispatches gRPC handling to the blocking thread pool

#Next Article

S9-02: SchemaRegistryService — Ontology Registration State Machine. We will dive into the Control Layer's Schema Registry to examine how it manages the DRAFT -> ACTIVE -> DEPRECATED lifecycle state machine and implements forward/backward compatibility checks.

Tags: #coomia-dip #source-code-reading #data-Layer #quarkus #grpc #entity-crud #soft-delete #batch-operations #world-context