/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.hertzbeat.alert.calculate.realtime.window;

import org.apache.hertzbeat.alert.reduce.AlarmCommonReduce;
import org.apache.hertzbeat.common.constants.CommonConstants;
import org.apache.hertzbeat.common.config.VirtualThreadProperties;
import org.apache.hertzbeat.common.entity.alerter.AlertDefine;
import org.apache.hertzbeat.common.entity.alerter.SingleAlert;
import org.apache.hertzbeat.common.entity.log.LogEntry;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;

import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import static org.junit.jupiter.api.Assertions.assertAll;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyMap;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;

/**
 * Test case for {@link AlarmEvaluator}
 */
@ExtendWith(MockitoExtension.class)
class AlarmEvaluatorTest {

    @Mock
    private AlarmCommonReduce alarmCommonReduce;

    @InjectMocks
    private AlarmEvaluator alarmEvaluator;

    private AlertDefine alertDefine;
    private WindowAggregator.WindowData windowData;

    @BeforeEach
    void setUp() {
        // Create test log entry
        LogEntry logEntry = LogEntry.builder()
                .timeUnixNano(System.currentTimeMillis() * 1_000_000L)
                .severityText("ERROR")
                .body("Test error message")
                .attributes(Map.of("service", "test-service", "pod", "test-pod"))
                .resource(Map.of("cluster", "test-cluster"))
                .build();

        // Create test alert define for individual mode
        alertDefine = AlertDefine.builder()
                .id(1L)
                .name("test_log_alert")
                .type("realtime_log")
                .expr("log.severityText == 'ERROR'")
                .times(1)
                .template("Alert: {{log.body}}")
                .labels(Map.of(CommonConstants.ALERT_MODE_LABEL, CommonConstants.ALERT_MODE_INDIVIDUAL, CommonConstants.LABEL_ALERT_SEVERITY, CommonConstants.ALERT_SEVERITY_CRITICAL))
                .annotations(Map.of("summary", "Test alert summary"))
                .enable(true)
                .build();

        // Create matching log events
        MatchingLogEvent matchingEvent = MatchingLogEvent.builder()
                .logEntry(logEntry)
                .alertDefine(alertDefine)
                .eventTimestamp(System.currentTimeMillis())
                .workerTimestamp(System.currentTimeMillis())
                .build();

        // Create window data
        WindowAggregator.WindowKey windowKey = new WindowAggregator.WindowKey(1L, 
            System.currentTimeMillis() - 60000, System.currentTimeMillis());
        windowData = new WindowAggregator.WindowData(windowKey, alertDefine);
        windowData.addMatchingLog(matchingEvent);
    }

    @AfterEach
    void tearDown() {
        if (alarmEvaluator != null) {
            alarmEvaluator.destroy();
        }
    }

    @Test
    void testProcessWindowDataWithIndividualMode() throws InterruptedException {
        // Given - alert define with individual mode
        alertDefine.setLabels(Map.of(CommonConstants.ALERT_MODE_LABEL, CommonConstants.ALERT_MODE_INDIVIDUAL));
        
        // When
        alarmEvaluator.sendAndProcessWindowData(windowData);
        
        // Wait for async processing
        Thread.sleep(2000);

        // Then
        verify(alarmCommonReduce, times(1)).reduceAndSendAlarm(any(SingleAlert.class));
        verify(alarmCommonReduce, never()).reduceAndSendAlarmGroup(anyMap(), anyList());
    }

    @Test
    void testProcessWindowDataWithGroupMode() throws InterruptedException {
        // Given - alert define with group mode
        alertDefine.setLabels(Map.of(CommonConstants.ALERT_MODE_LABEL, CommonConstants.ALERT_MODE_GROUP));
        
        // When
        alarmEvaluator.sendAndProcessWindowData(windowData);
        
        // Wait for async processing
        Thread.sleep(2000);

        // Then
        verify(alarmCommonReduce, never()).reduceAndSendAlarm(any(SingleAlert.class));
        verify(alarmCommonReduce, times(1)).reduceAndSendAlarmGroup(anyMap(), anyList());
    }

    @Test
    void testProcessWindowDataWithEmptyLogs() throws InterruptedException {
        // Given - window data with no matching logs
        WindowAggregator.WindowKey emptyWindowKey = new WindowAggregator.WindowKey(1L, 
            System.currentTimeMillis() - 60000, System.currentTimeMillis());
        WindowAggregator.WindowData emptyWindowData = new WindowAggregator.WindowData(emptyWindowKey, alertDefine);
        
        // When
        alarmEvaluator.sendAndProcessWindowData(emptyWindowData);
        
        // Wait for async processing
        Thread.sleep(2000);

        // Then - should not process any alerts
        verify(alarmCommonReduce, never()).reduceAndSendAlarm(any(SingleAlert.class));
        verify(alarmCommonReduce, never()).reduceAndSendAlarmGroup(anyMap(), anyList());
    }

    @Test
    void testProcessWindowDataWithInsufficientTimes() throws InterruptedException {
        // Given - alert define requiring 3 times but only 1 matching log
        alertDefine.setTimes(3);
        
        // When
        alarmEvaluator.sendAndProcessWindowData(windowData);
        
        // Wait for async processing
        Thread.sleep(2000);

        // Then - should not process any alerts due to insufficient count
        verify(alarmCommonReduce, never()).reduceAndSendAlarm(any(SingleAlert.class));
        verify(alarmCommonReduce, never()).reduceAndSendAlarmGroup(anyMap(), anyList());
    }

    @Test
    void testIndividualAlertGeneration() throws InterruptedException {
        // Given
        alertDefine.setLabels(Map.of(CommonConstants.ALERT_MODE_LABEL, CommonConstants.ALERT_MODE_INDIVIDUAL, CommonConstants.LABEL_ALERT_SEVERITY, CommonConstants.ALERT_SEVERITY_CRITICAL));
        alertDefine.setAnnotations(Map.of("summary", "Test summary", "description", "Test description"));
        
        // When
        alarmEvaluator.sendAndProcessWindowData(windowData);
        
        // Wait for async processing
        Thread.sleep(2000);

        // Then
        ArgumentCaptor<SingleAlert> alertCaptor = ArgumentCaptor.forClass(SingleAlert.class);
        verify(alarmCommonReduce).reduceAndSendAlarm(alertCaptor.capture());
        
        SingleAlert capturedAlert = alertCaptor.getValue();
        assertAll(
            () -> assertNotNull(capturedAlert),
            () -> assertEquals(CommonConstants.ALERT_STATUS_FIRING, capturedAlert.getStatus()),
            () -> assertEquals(1, capturedAlert.getTriggerTimes()),
            () -> assertNotNull(capturedAlert.getContent()),
            () -> assertNotNull(capturedAlert.getLabels()),
            () -> assertNotNull(capturedAlert.getAnnotations()),
            () -> assertTrue(capturedAlert.getStartAt() > 0),
            () -> assertTrue(capturedAlert.getActiveAt() > 0)
        );
        
        // Verify labels contain alert define info and log entry data
        Map<String, String> labels = capturedAlert.getLabels();
        assertTrue(labels.containsKey(CommonConstants.LABEL_ALERT_SEVERITY));
        assertTrue(labels.containsKey("severityText"));
        assertTrue(labels.containsKey("body"));
    }

    @Test
    void testGroupAlertGeneration() throws InterruptedException {
        // Given
        alertDefine.setLabels(Map.of(CommonConstants.ALERT_MODE_LABEL, CommonConstants.ALERT_MODE_GROUP, CommonConstants.LABEL_ALERT_SEVERITY, CommonConstants.ALERT_SEVERITY_WARNING));
        
        // Add more matching logs to test group functionality
        MatchingLogEvent secondEvent = MatchingLogEvent.builder()
                .logEntry(LogEntry.builder()
                    .timeUnixNano((System.currentTimeMillis() + 1000) * 1_000_000L)
                    .severityText("ERROR") 
                    .body("Second error message")
                    .build())
                .alertDefine(alertDefine)
                .eventTimestamp(System.currentTimeMillis() + 1000)
                .workerTimestamp(System.currentTimeMillis() + 1000)
                .build();
        
        windowData.addMatchingLog(secondEvent);
        
        // When
        alarmEvaluator.sendAndProcessWindowData(windowData);
        
        // Wait for async processing
        Thread.sleep(2000);

        // Then
        @SuppressWarnings("unchecked")
        ArgumentCaptor<Map<String, String>> labelsCaptor = ArgumentCaptor.forClass(Map.class);
        @SuppressWarnings("unchecked")
        ArgumentCaptor<List<SingleAlert>> alertsCaptor = ArgumentCaptor.forClass(List.class);
        verify(alarmCommonReduce).reduceAndSendAlarmGroup(labelsCaptor.capture(), alertsCaptor.capture());
        
        Map<String, String> groupLabels = labelsCaptor.getValue();
        List<SingleAlert> alerts = alertsCaptor.getValue();
        
        assertAll(
            () -> assertNotNull(groupLabels),
            () -> assertNotNull(alerts),
            () -> assertEquals(2, alerts.size()),
            () -> assertTrue(groupLabels.containsKey("window_start_time")),
            () -> assertTrue(groupLabels.containsKey("window_end_time")),
            () -> assertTrue(groupLabels.containsKey("matching_logs_count")),
            () -> assertEquals(CommonConstants.ALERT_MODE_GROUP, groupLabels.get(CommonConstants.ALERT_MODE_LABEL)),
            () -> assertEquals("2", groupLabels.get("matching_logs_count"))
        );
        
        // Verify each alert in the group
        for (SingleAlert alert : alerts) {
            assertEquals(CommonConstants.ALERT_STATUS_FIRING, alert.getStatus());
            assertEquals(2, alert.getTriggerTimes()); // Should equal the count of matching logs
        }
    }

    @Test
    void testLogEntryDataMapping() throws InterruptedException {
        // Given - log entry with comprehensive data
        LogEntry detailedLogEntry = LogEntry.builder()
                .timeUnixNano(System.currentTimeMillis() * 1_000_000L)
                .severityNumber(17) // ERROR level
                .severityText("ERROR")
                .body("Detailed error message")
                .attributes(Map.of("service", "api-service", "version", "1.0.0"))
                .resource(Map.of("cluster", "prod-cluster", "namespace", "default"))
                .droppedAttributesCount(0)
                .traceId("trace123")
                .spanId("span456")
                .traceFlags(1)
                .build();

        MatchingLogEvent detailedEvent = MatchingLogEvent.builder()
                .logEntry(detailedLogEntry)
                .alertDefine(alertDefine)
                .eventTimestamp(System.currentTimeMillis())
                .workerTimestamp(System.currentTimeMillis())
                .build();

        WindowAggregator.WindowData detailedWindowData = new WindowAggregator.WindowData(
            new WindowAggregator.WindowKey(1L, System.currentTimeMillis() - 60000, System.currentTimeMillis()),
            alertDefine);
        detailedWindowData.addMatchingLog(detailedEvent);

        alertDefine.setLabels(Map.of(CommonConstants.ALERT_MODE_LABEL, CommonConstants.ALERT_MODE_INDIVIDUAL));
        
        // When
        alarmEvaluator.sendAndProcessWindowData(detailedWindowData);
        Thread.sleep(2000);

        // Then
        ArgumentCaptor<SingleAlert> alertCaptor = ArgumentCaptor.forClass(SingleAlert.class);
        verify(alarmCommonReduce).reduceAndSendAlarm(alertCaptor.capture());
        
        SingleAlert alert = alertCaptor.getValue();
        Map<String, String> labels = alert.getLabels();
        
        // Verify log entry fields are mapped to labels
        assertAll(
            () -> assertEquals("17", labels.get("severityNumber")),
            () -> assertEquals("ERROR", labels.get("severityText")),
            () -> assertEquals("Detailed error message", labels.get("body")),
            () -> assertEquals("0", labels.get("droppedAttributesCount")),
            () -> assertEquals("trace123", labels.get("traceId")),
            () -> assertEquals("span456", labels.get("spanId")),
            () -> assertEquals("1", labels.get("traceFlags")),
            () -> assertEquals("api-service", labels.get("attr_service")),
            () -> assertEquals("1.0.0", labels.get("attr_version")),
            () -> assertEquals("prod-cluster", labels.get("resource_cluster")),
            () -> assertEquals("default", labels.get("resource_namespace"))
        );
    }

    @Test
    void testMultipleMatchingLogsInWindow() throws InterruptedException {
        // Given - multiple matching logs in the same window
        alertDefine.setTimes(2); // Require at least 2 occurrences
        alertDefine.setLabels(Map.of(CommonConstants.ALERT_MODE_LABEL, CommonConstants.ALERT_MODE_GROUP));
        
        // Add second matching log
        MatchingLogEvent secondEvent = MatchingLogEvent.builder()
                .logEntry(LogEntry.builder()
                    .severityText("ERROR")
                    .body("Another error")
                    .build())
                .alertDefine(alertDefine)
                .eventTimestamp(System.currentTimeMillis() + 5000)
                .workerTimestamp(System.currentTimeMillis() + 5000)
                .build();
        
        windowData.addMatchingLog(secondEvent);
        
        // When
        alarmEvaluator.sendAndProcessWindowData(windowData);
        Thread.sleep(2000);

        // Then - should trigger group alert since threshold is met
        verify(alarmCommonReduce).reduceAndSendAlarmGroup(anyMap(), anyList());
        
        @SuppressWarnings("unchecked")
        ArgumentCaptor<List<SingleAlert>> alertsCaptor = ArgumentCaptor.forClass(List.class);
        verify(alarmCommonReduce).reduceAndSendAlarmGroup(anyMap(), alertsCaptor.capture());
        
        List<SingleAlert> alerts = alertsCaptor.getValue();
        assertEquals(2, alerts.size());
        assertEquals(2, alerts.get(0).getTriggerTimes()); // Each alert should have trigger times = total count
    }

    @Test
    void testSendAndProcessWindowDataRunsOnVirtualThread() throws Exception {
        CountDownLatch latch = new CountDownLatch(1);
        AtomicBoolean virtualThread = new AtomicBoolean(false);
        doAnswer(invocation -> {
            virtualThread.set(Thread.currentThread().isVirtual());
            latch.countDown();
            return null;
        }).when(alarmCommonReduce).reduceAndSendAlarm(any(SingleAlert.class));

        alertDefine.setLabels(Map.of(CommonConstants.ALERT_MODE_LABEL, CommonConstants.ALERT_MODE_INDIVIDUAL));
        alarmEvaluator.sendAndProcessWindowData(windowData);

        assertTrue(latch.await(5, TimeUnit.SECONDS));
        assertTrue(virtualThread.get());
    }

    @Test
    void testSendAndProcessWindowDataQueuesWhenConcurrencyLimitReached() throws Exception {
        VirtualThreadProperties properties = new VirtualThreadProperties(
                true,
                VirtualThreadProperties.PoolProperties.collectorDefaults(),
                VirtualThreadProperties.PoolProperties.commonDefaults(),
                VirtualThreadProperties.PoolProperties.managerDefaults(),
                new VirtualThreadProperties.AlerterProperties(
                        VirtualThreadProperties.PoolProperties.alerterNotifyDefaults(),
                        10,
                        VirtualThreadProperties.QueueProperties.logWorkerDefaults(),
                        VirtualThreadProperties.QueueProperties.reduceDefaults(),
                        new VirtualThreadProperties.QueueProperties(1, 0),
                        4),
                VirtualThreadProperties.PoolProperties.warehouseDefaults(),
                VirtualThreadProperties.AsyncProperties.defaults());
        alarmEvaluator.destroy();
        alarmEvaluator = new AlarmEvaluator(alarmCommonReduce, properties);

        CountDownLatch firstStarted = new CountDownLatch(1);
        CountDownLatch releaseFirst = new CountDownLatch(1);
        CountDownLatch secondStarted = new CountDownLatch(1);
        AtomicInteger invocationOrder = new AtomicInteger();
        doAnswer(invocation -> {
            int order = invocationOrder.incrementAndGet();
            if (order == 1) {
                firstStarted.countDown();
                try {
                    releaseFirst.await(5, TimeUnit.SECONDS);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            } else if (order == 2) {
                secondStarted.countDown();
            }
            return null;
        }).when(alarmCommonReduce).reduceAndSendAlarm(any(SingleAlert.class));

        WindowAggregator.WindowData firstWindow = cloneWindowDataWithBody("first");
        WindowAggregator.WindowData secondWindow = cloneWindowDataWithBody("second");
        firstWindow.getAlertDefine().setLabels(Map.of(CommonConstants.ALERT_MODE_LABEL, CommonConstants.ALERT_MODE_INDIVIDUAL));
        secondWindow.getAlertDefine().setLabels(Map.of(CommonConstants.ALERT_MODE_LABEL, CommonConstants.ALERT_MODE_INDIVIDUAL));

        alarmEvaluator.sendAndProcessWindowData(firstWindow);
        assertTrue(firstStarted.await(5, TimeUnit.SECONDS));

        alarmEvaluator.sendAndProcessWindowData(secondWindow);
        assertFalse(secondStarted.await(200, TimeUnit.MILLISECONDS));

        releaseFirst.countDown();
        assertTrue(secondStarted.await(5, TimeUnit.SECONDS));
    }

    private WindowAggregator.WindowData cloneWindowDataWithBody(String body) {
        LogEntry logEntry = LogEntry.builder()
                .timeUnixNano(System.currentTimeMillis() * 1_000_000L)
                .severityText("ERROR")
                .body(body)
                .build();
        AlertDefine define = AlertDefine.builder()
                .id(alertDefine.getId())
                .name(alertDefine.getName())
                .type(alertDefine.getType())
                .expr(alertDefine.getExpr())
                .times(alertDefine.getTimes())
                .template(alertDefine.getTemplate())
                .labels(alertDefine.getLabels())
                .annotations(alertDefine.getAnnotations())
                .enable(alertDefine.isEnable())
                .build();
        MatchingLogEvent event = MatchingLogEvent.builder()
                .logEntry(logEntry)
                .alertDefine(define)
                .eventTimestamp(System.currentTimeMillis())
                .workerTimestamp(System.currentTimeMillis())
                .build();
        WindowAggregator.WindowData clonedWindowData = new WindowAggregator.WindowData(
                new WindowAggregator.WindowKey(define.getId(), System.currentTimeMillis() - 60000, System.currentTimeMillis()),
                define);
        clonedWindowData.addMatchingLog(event);
        return clonedWindowData;
    }
}
