package com.ankurm.kafkabasics;

import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.junit.jupiter.api.Test;
import org.springframework.kafka.support.serializer.JacksonJsonDeserializer;
import org.springframework.kafka.support.serializer.JacksonJsonSerializer;
import org.springframework.kafka.support.serializer.JsonSerializer;

import java.nio.charset.StandardCharsets;
import java.util.Map;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;

/**
 * The two JSON serializer families Spring Kafka 4.1 ships, side by side. No broker needed: a
 * Serializer is a function from object to bytes and can be called directly, which is the
 * cheapest possible way to settle a serialisation question.
 *
 * @see <a href="../../../../../docs/03-serialisation.md">docs/03-serialisation.md</a>
 */
class SerialisationTest {

	private static final OrderEvent EVENT = OrderEvent.of("o-1", "c-1", "10.00");

	@Test
	void theJackson2SerializerCannotWriteAnInstant() {
		// JsonSerializer is the Jackson 2 one. Its default ObjectMapper has no JSR-310 module,
		// and Jackson 2.21 refuses java.time types rather than guessing at a representation.
		// This is what you get by following any pre-Boot-4 tutorial.
		try (JsonSerializer<OrderEvent> serializer = new JsonSerializer<>()) {
			assertThatExceptionOfType(SerializationException.class)
					.isThrownBy(() -> serializer.serialize("orders", EVENT))
					.withMessageContaining("Can't serialize data")
					.withStackTraceContaining("Java 8 date/time type `java.time.Instant` not supported by default");
		}
	}

	@Test
	void theJackson3SerializerWritesItWithNoConfiguration() {
		try (JacksonJsonSerializer<OrderEvent> serializer = new JacksonJsonSerializer<>()) {
			String json = new String(serializer.serialize("orders", EVENT), StandardCharsets.UTF_8);
			System.out.println("JacksonJsonSerializer -> " + json);
			assertThat(json).contains("\"orderId\":\"o-1\"").contains("\"placedAt\":");
			// BigDecimal survives as a number, not a string, and keeps its scale.
			assertThat(json).contains("\"amount\":10.00");
		}
	}

	@Test
	void theDeserializerRefusesAnUntrustedClassNamedInTheTypeHeader() {
		Headers headers = new RecordHeaders();
		byte[] bytes;
		try (JacksonJsonSerializer<OrderEvent> serializer = new JacksonJsonSerializer<>()) {
			// The three-argument overload is the one that writes __TypeId__. That header is how the
			// consumer learns which class to build, and it is also why trusted packages exist:
			// instantiating a class named by an inbound message is a deserialization gadget.
			bytes = serializer.serialize("orders", headers, EVENT);
		}
		assertThat(headers.lastHeader("__TypeId__")).isNotNull();
		assertThat(new String(headers.lastHeader("__TypeId__").value(), StandardCharsets.UTF_8))
				.isEqualTo(OrderEvent.class.getName());

		try (JacksonJsonDeserializer<OrderEvent> deserializer = new JacksonJsonDeserializer<>()) {
			deserializer.configure(Map.of(), false);
			assertThatExceptionOfType(Exception.class)
					.isThrownBy(() -> deserializer.deserialize("orders", headers, bytes))
					.withMessageContaining("not in the trusted packages");
		}
	}

	@Test
	void aRoundTripThroughTheJackson3PairIsLossless() {
		byte[] bytes;
		try (JacksonJsonSerializer<OrderEvent> serializer = new JacksonJsonSerializer<>()) {
			bytes = serializer.serialize("orders", EVENT);
		}
		try (JacksonJsonDeserializer<OrderEvent> deserializer = new JacksonJsonDeserializer<>()) {
			deserializer.configure(Map.of(JacksonJsonDeserializer.VALUE_DEFAULT_TYPE,
					OrderEvent.class.getName(), JacksonJsonDeserializer.TRUSTED_PACKAGES,
					"com.ankurm.kafkabasics"), false);
			OrderEvent back = deserializer.deserialize("orders", bytes);
			assertThat(back).isEqualTo(EVENT);
			// BigDecimal scale survives the round trip. It would not if amount were a double.
			assertThat(back.amount().scale()).isEqualTo(2);
		}
	}

}