feat(content): emit content.updated on published metadata changes

docs/contracts/rest-api-v1.md

@@ -83,6 +83,7 @@ This document defines the MVP API contract groups, standards, and core request/r
 ### `PATCH /v1/content/{content_id}`
 - Partially updates content metadata for channel members
 - Mutable fields (MVP): `title`, `description`, `visibility`
+- Emits `content.updated` event when content is in `PUBLISHED` state
 
 ### `POST /v1/content/{content_id}/publish`
 - Publishes content from `DRAFT` to `PUBLISHED`

services/java/content-service/src/main/java/com/cloudmedia/content/application/content/ContentService.java

@@ -6,6 +6,7 @@
 import com.cloudmedia.content.api.content.dto.UpdateContentRequest;
 import com.cloudmedia.content.events.ContentEventPublisher;
 import com.cloudmedia.content.events.ContentPublishedPayload;
+import com.cloudmedia.content.events.ContentUpdatedPayload;
 import com.cloudmedia.content.error.ApiException;
 import com.cloudmedia.content.persistence.entity.ContentState;
 import com.cloudmedia.content.persistence.entity.ChannelEntity;
@@ -83,7 +84,14 @@ public ContentResponse updateMetadata(String contentId, UpdateContentRequest req
 		}
 		content.setUpdatedAt(LocalDateTime.now());
 
-		return toResponse(contentRepository.save(content));
+		ContentEntity savedContent = contentRepository.save(content);
+		if (savedContent.getState() == ContentState.PUBLISHED) {
+			contentEventPublisher.publishContentUpdated(new ContentUpdatedPayload(savedContent.getId(),
+					savedContent.getChannel().getId(), savedContent.getTitle(), savedContent.getContentType().name(),
+					savedContent.getVisibility().name()), null);
+		}
+
+		return toResponse(savedContent);
 	}
 
 	@Transactional

services/java/content-service/src/main/java/com/cloudmedia/content/events/ContentEventPublisher.java

@@ -3,4 +3,6 @@
 public interface ContentEventPublisher {
 
 	void publishContentPublished(ContentPublishedPayload payload, String traceId);
+
+	void publishContentUpdated(ContentUpdatedPayload payload, String traceId);
 }

services/java/content-service/src/main/java/com/cloudmedia/content/events/ContentEventsProperties.java

@@ -25,12 +25,22 @@ public static class Topics {
 
 		private String contentPublished = "cloudmedia.content.published";
 
+		private String contentUpdated = "cloudmedia.content.updated";
+
 		public String getContentPublished() {
 			return contentPublished;
 		}
 
 		public void setContentPublished(String contentPublished) {
 			this.contentPublished = contentPublished;
 		}
+
+		public String getContentUpdated() {
+			return contentUpdated;
+		}
+
+		public void setContentUpdated(String contentUpdated) {
+			this.contentUpdated = contentUpdated;
+		}
 	}
 }

services/java/content-service/src/main/java/com/cloudmedia/content/events/ContentUpdatedPayload.java

@@ -0,0 +1,5 @@
+package com.cloudmedia.content.events;
+
+public record ContentUpdatedPayload(String contentId, String channelId, String title, String contentType,
+		String visibility) {
+}

services/java/content-service/src/main/java/com/cloudmedia/content/events/KafkaContentEventPublisher.java

@@ -25,6 +25,14 @@ public void publishContentPublished(ContentPublishedPayload payload, String trac
 		kafkaTemplate.send(properties.getTopics().getContentPublished(), payload.contentId(), envelope);
 	}
 
+	@Override
+	public void publishContentUpdated(ContentUpdatedPayload payload, String traceId) {
+		ContentEventEnvelope envelope = new ContentEventEnvelope(UUID.randomUUID().toString(), "content.updated",
+				EVENT_VERSION, Instant.now(), "content-service", "content", payload.contentId(),
+				resolveTraceId(traceId), payload);
+		kafkaTemplate.send(properties.getTopics().getContentUpdated(), payload.contentId(), envelope);
+	}
+
 	private String resolveTraceId(String traceId) {
 		if (traceId != null && !traceId.isBlank()) {
 			return traceId;

services/java/content-service/src/main/java/com/cloudmedia/content/events/NoopContentEventPublisher.java

@@ -6,4 +6,9 @@ public class NoopContentEventPublisher implements ContentEventPublisher {
 	public void publishContentPublished(ContentPublishedPayload payload, String traceId) {
 		// intentionally no-op when content event publishing is disabled
 	}
+
+	@Override
+	public void publishContentUpdated(ContentUpdatedPayload payload, String traceId) {
+		// intentionally no-op when content event publishing is disabled
+	}
 }

services/java/content-service/src/test/java/com/cloudmedia/content/application/content/ContentServiceIntegrationTest.java

@@ -4,6 +4,7 @@
 import com.cloudmedia.content.api.content.dto.UpdateContentRequest;
 import com.cloudmedia.content.events.ContentEventPublisher;
 import com.cloudmedia.content.events.ContentPublishedPayload;
+import com.cloudmedia.content.events.ContentUpdatedPayload;
 import com.cloudmedia.content.error.ApiException;
 import com.cloudmedia.content.persistence.entity.ChannelEntity;
 import com.cloudmedia.content.persistence.entity.ChannelMemberEntity;
@@ -213,6 +214,21 @@ void unpublishRejectsWhenStateIsNotPublished() {
 		assertEquals("CONTENT_STATE_INVALID", exception.getCode());
 	}
 
+	@Test
+	void updateMetadataEmitsContentUpdatedWhenPublished() {
+		ChannelEntity channel = saveChannel("channel-content-11", "content-channel-eleven");
+		saveMembership(channel, "updater-1", ChannelMemberRole.ADMIN);
+		ContentEntity content = saveContent(channel, "Original", "desc", ContentVisibility.PUBLIC,
+				ContentState.PUBLISHED, true);
+
+		contentService.updateMetadata(content.getId(),
+				new UpdateContentRequest("updater-1", "Updated Title", null, ContentVisibility.UNLISTED));
+
+		verify(contentEventPublisher).publishContentUpdated(eq(new ContentUpdatedPayload(content.getId(),
+				channel.getId(), "Updated Title", ContentType.VIDEO.name(), ContentVisibility.UNLISTED.name())),
+				isNull());
+	}
+
 	private ChannelEntity saveChannel(String id, String slug) {
 		ChannelEntity channel = new ChannelEntity();
 		channel.setId(id);

services/java/content-service/src/test/java/com/cloudmedia/content/events/KafkaContentEventPublisherTest.java

@@ -57,4 +57,53 @@ public Map<String, Object> getConfigurationProperties() {
 		assertNotNull(envelope.eventId());
 		assertNotNull(envelope.occurredAt());
 	}
+
+	@Test
+	void publishContentUpdatedSendsEnvelopeToConfiguredTopic() {
+		MockProducer<String, Object> mockProducer = new MockProducer<>(true, new StringSerializer(),
+				new JsonSerializer<>());
+		ProducerFactory<String, Object> producerFactory = new ProducerFactory<>() {
+			@Override
+			public org.apache.kafka.clients.producer.Producer<String, Object> createProducer() {
+				return mockProducer;
+			}
+
+			@Override
+			public boolean transactionCapable() {
+				return false;
+			}
+
+			@Override
+			public Map<String, Object> getConfigurationProperties() {
+				return new HashMap<>();
+			}
+		};
+		KafkaTemplate<String, Object> kafkaTemplate = new KafkaTemplate<>(producerFactory);
+		ContentEventsProperties properties = new ContentEventsProperties();
+		properties.getTopics().setContentUpdated("cloudmedia.content.updated");
+		KafkaContentEventPublisher publisher = new KafkaContentEventPublisher(kafkaTemplate, properties);
+
+		publisher.publishContentUpdated(
+				new ContentUpdatedPayload("cnt_2", "chn_2", "Updated Title", "VIDEO", "UNLISTED"), "req_456");
+
+		assertEquals(1, mockProducer.history().size());
+		ProducerRecord<String, Object> record = mockProducer.history().getFirst();
+		assertEquals("cloudmedia.content.updated", record.topic());
+		assertEquals("cnt_2", record.key());
+		ContentEventEnvelope envelope = (ContentEventEnvelope) record.value();
+		assertEquals("content.updated", envelope.eventType());
+		assertEquals(1, envelope.eventVersion());
+		assertEquals("content-service", envelope.producer());
+		assertEquals("content", envelope.entityType());
+		assertEquals("cnt_2", envelope.entityId());
+		assertEquals("req_456", envelope.traceId());
+		assertNotNull(envelope.eventId());
+		assertNotNull(envelope.occurredAt());
+		ContentUpdatedPayload payload = (ContentUpdatedPayload) envelope.payload();
+		assertEquals("cnt_2", payload.contentId());
+		assertEquals("chn_2", payload.channelId());
+		assertEquals("Updated Title", payload.title());
+		assertEquals("VIDEO", payload.contentType());
+		assertEquals("UNLISTED", payload.visibility());
+	}
 }