001// --------------------------------------------------------------------------------
002// Copyright 2002-2026 Echo Three, LLC
003//
004// Licensed under the Apache License, Version 2.0 (the "License");
005// you may not use this file except in compliance with the License.
006// You may obtain a copy of the License at
007//
008//     http://www.apache.org/licenses/LICENSE-2.0
009//
010// Unless required by applicable law or agreed to in writing, software
011// distributed under the License is distributed on an "AS IS" BASIS,
012// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013// See the License for the specific language governing permissions and
014// limitations under the License.
015// --------------------------------------------------------------------------------
016
017package com.echothree.model.control.core.server.kafka;
018
019import com.echothree.model.control.core.server.control.EntityInstanceControl;
020import com.echothree.model.data.core.server.entity.Event;
021import com.echothree.util.server.persistence.Session;
022import com.echothree.util.server.string.EntityInstanceUtils;
023import com.fasterxml.jackson.databind.ObjectMapper;
024import com.fasterxml.uuid.Generators;
025import com.fasterxml.uuid.impl.NameBasedGenerator;
026import com.google.common.net.MediaType;
027import fish.payara.cloud.connectors.kafka.api.KafkaConnectionFactory;
028import io.cloudevents.CloudEvent;
029import io.cloudevents.core.builder.CloudEventBuilder;
030import io.cloudevents.jackson.JsonFormat;
031import io.cloudevents.kafka.KafkaMessageFactory;
032import java.io.IOException;
033import java.net.URI;
034import java.time.Instant;
035import java.time.ZoneOffset;
036import java.util.ArrayList;
037import java.util.UUID;
038import java.util.concurrent.Future;
039import javax.annotation.Resource;
040import javax.enterprise.context.ApplicationScoped;
041import org.apache.kafka.clients.producer.ProducerRecord;
042import org.apache.kafka.clients.producer.RecordMetadata;
043
044@ApplicationScoped
045public class EventTopic {
046
047    @Resource(name = "java:/KafkaConnectionFactory")
048    KafkaConnectionFactory kafkaConnectionFactory;
049
050    // Feature flags
051    private static final boolean SEND_JSON_EVENTS = true;
052    private static final boolean SEND_AVRO_EVENTS = false;
053
054    // Kafka
055    private static final String TOPIC_JSON = "echothree-events-json";
056    private static final String TOPIC_AVRO = "echothree-events-avro";
057
058    // CloudEvents
059    private static final URI EVENT_SOURCE = URI.create("urn:echothree:events");
060
061    // Keep the namespace stable so the same eventId always produces the same UUIDv5.
062    private static final UUID NAMESPACE = Generators.nameBasedGenerator(NameBasedGenerator.NAMESPACE_URL).generate(EVENT_SOURCE.toString());
063    private static final NameBasedGenerator EVENT_ID_GENERATOR = Generators.nameBasedGenerator(NAMESPACE);
064
065    // Jackson
066    // Initialized only when JSON publishing is used.
067    private static class JsonSerialization {
068        private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
069        private static final JsonFormat CLOUD_EVENT_JSON_FORMAT = new JsonFormat();
070    }
071
072    protected EventTopic() {}
073
074    public void sendEvent(Event event) {
075        if((SEND_JSON_EVENTS || SEND_AVRO_EVENTS) && kafkaConnectionFactory != null) {
076            try {
077                try(var kafkaConnection = kafkaConnectionFactory.createConnection()) {
078                    var entityInstanceControl = Session.getModelController(EntityInstanceControl.class);
079                    var eventId = event.getPrimaryKey().getEntityId();
080                    var eventTime = event.getEventTime();
081                    var eventTimeSequence = event.getEventTimeSequence();
082                    var entityInstance = entityInstanceControl.ensureUuidForEntityInstance(event.getEntityInstance(), false);
083                    var entityRef = EntityInstanceUtils.getEntityRefByEntityInstance(entityInstance);
084                    var id = entityInstance.getUuid();
085                    var eventTypeName = event.getEventType().getEventTypeName();
086                    var relatedEntityRef = EntityInstanceUtils.getEntityRefByEntityInstance(event.getRelatedEntityInstance());
087                    var relatedEventType = event.getRelatedEventType();
088                    var relatedEventTypeName = relatedEventType == null ? null : relatedEventType.getEventTypeName();
089                    var createdByEntityRef = EntityInstanceUtils.getEntityRefByEntityInstance(event.getCreatedBy());
090
091                    var eventData = new com.echothree.model.control.core.server.kafka.Event(eventId, eventTime,
092                            eventTimeSequence, entityRef, id, eventTypeName, relatedEntityRef, relatedEventTypeName,
093                            createdByEntityRef);
094                    var cloudEvent = CloudEventBuilder.v1()
095                            .withId(EVENT_ID_GENERATOR.generate(eventId.toString()).toString())
096                            .withSource(EVENT_SOURCE)
097                            .withType("com.echothree.model.control.core.common.EventTypes." + eventTypeName)
098                            .withSubject(entityRef)
099                            .withTime(Instant.ofEpochMilli(eventTime).atOffset(ZoneOffset.UTC))
100                            .build();
101                    var futures = new ArrayList<Future<RecordMetadata>>(2);
102
103                    if(SEND_JSON_EVENTS) {
104                        futures.add(kafkaConnection.send(createJsonRecord(eventData, cloudEvent)));
105                    }
106
107                    if(SEND_AVRO_EVENTS) {
108                        futures.add(kafkaConnection.send(createAvroRecord(eventData, cloudEvent)));
109                    }
110
111                    for(var future : futures) {
112                        future.get();
113                    }
114                }
115            } catch(Exception e) {
116                throw new RuntimeException(e);
117            }
118        }
119    }
120
121    private ProducerRecord<String, byte[]> createJsonRecord(com.echothree.model.control.core.server.kafka.Event eventData,
122            CloudEvent cloudEvent) throws IOException {
123        var jsonCloudEvent = CloudEventBuilder.from(cloudEvent)
124                .withData(MediaType.JSON_UTF_8.toString(), JsonSerialization.OBJECT_MAPPER.writeValueAsBytes(eventData))
125                .build();
126
127        return KafkaMessageFactory.createWriter(TOPIC_JSON, null, eventData.eventTime(), eventData.entityRef())
128                .writeStructured(jsonCloudEvent, JsonSerialization.CLOUD_EVENT_JSON_FORMAT);
129    }
130
131    private ProducerRecord<String, byte[]> createAvroRecord(com.echothree.model.control.core.server.kafka.Event eventData,
132            CloudEvent cloudEvent) throws IOException {
133        var eventAvro = com.echothree.model.avro.core.common.Event.newBuilder()
134                .setEventId(eventData.eventId())
135                .setEventTime(eventData.eventTime())
136                .setEventTimeSequence(eventData.eventTimeSequence())
137                .setEntityRef(eventData.entityRef())
138                .setId(eventData.id())
139                .setEventTypeName(eventData.eventTypeName())
140                .setRelatedEntityRef(eventData.relatedEntityRef())
141                .setRelatedEventTypeName(eventData.relatedEventTypeName())
142                .setCreatedByEntityRef(eventData.createdByEntityRef())
143                .build();
144        // Single-object encoding includes the schema fingerprint for the generated decoder.
145        var avroBuffer = eventAvro.toByteBuffer();
146        var avroBytes = new byte[avroBuffer.remaining()];
147
148        avroBuffer.get(avroBytes);
149
150        var avroCloudEvent = CloudEventBuilder.from(cloudEvent)
151                .withData("avro/binary", avroBytes)
152                .build();
153
154        return KafkaMessageFactory.createWriter(TOPIC_AVRO, null, eventData.eventTime(), eventData.entityRef())
155                .writeBinary(avroCloudEvent);
156    }
157
158}