package com.ai.trackwalker.uavaccess.event;

import com.ai.trackwalker.uavaccess.common.Jsons;
import com.ai.trackwalker.uavaccess.config.UavAccessProperties;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.sql.Timestamp;
import java.time.Instant;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

@Service
public class OutboxEventService {
    private final JdbcTemplate jdbc;
    private final KafkaTemplate<String, Object> kafka;
    private final UavAccessProperties properties;

    public OutboxEventService(
            JdbcTemplate jdbc,
            KafkaTemplate<String, Object> kafka,
            UavAccessProperties properties
    ) {
        this.jdbc = jdbc;
        this.kafka = kafka;
        this.properties = properties;
    }

    @Transactional
    public void enqueue(String eventType, String aggregateId, String eventKey, Map<String, Object> payload) {
        Instant now = Instant.now();
        Map<String, Object> envelope = new LinkedHashMap<>();
        envelope.put("event_id", UUID.randomUUID().toString());
        envelope.put("event_type", eventType);
        envelope.put("schema_version", 1);
        envelope.put("occurred_at", now.toString());
        envelope.put("trace_id", UUID.randomUUID().toString());
        envelope.put("workspace_id", properties.getFlighthub().getProjectUuid());
        envelope.put("vendor_code", "FLIGHTHUB2");
        envelope.put("aggregate_id", aggregateId);
        envelope.put("payload", payload);
        jdbc.update("""
                        insert into access_outbox_events(
                            id,event_type,aggregate_id,event_key,payload,state,occurred_at
                        ) values (?,?,?,?,?::jsonb,'PENDING',?)
                        on conflict (event_key) do nothing
                        """,
                String.valueOf(envelope.get("event_id")),
                eventType,
                aggregateId,
                eventKey,
                Jsons.write(envelope),
                Timestamp.from(now)
        );
    }

    @Scheduled(fixedDelayString = "${uav-access.outbox-relay-delay-ms:5000}")
    public void relay() {
        List<Map<String, Object>> pending = jdbc.queryForList("""
                select id,event_type,aggregate_id,payload::text as payload,attempt_count
                from access_outbox_events
                where state in ('PENDING','FAILED') and attempt_count < 20
                order by occurred_at
                limit 100
                """);
        for (Map<String, Object> row : pending) {
            String id = String.valueOf(row.get("id"));
            try {
                Map<String, Object> envelope = Jsons.map(String.valueOf(row.get("payload")));
                kafka.send(
                        properties.getEventTopic(),
                        String.valueOf(row.get("aggregate_id")),
                        envelope
                ).get(10, TimeUnit.SECONDS);
                jdbc.update("""
                        update access_outbox_events set state='PUBLISHED',attempt_count=attempt_count+1,
                            published_at=?,last_error=null where id=?
                        """, Timestamp.from(Instant.now()), id);
            } catch (Exception exception) {
                jdbc.update("""
                        update access_outbox_events set state='FAILED',attempt_count=attempt_count+1,
                            last_error=? where id=?
                        """, bounded(exception.getClass().getSimpleName() + ": " + exception.getMessage()), id);
            }
        }
    }

    private String bounded(String message) {
        if (message == null) return "unknown";
        return message.substring(0, Math.min(2000, message.length()));
    }
}
