package org.kne.cloud.network.monitor;

import java.util.ArrayList;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.kne.cloud.clock.HighAccuracyClock;
import org.kne.cloud.network.klalb.KLALBUtils;
import org.kne.math.Long128;

public class ArrayListTimestampMonitor<K> implements TimestampMonitor<K> {

	private HighAccuracyClock clock;
	private String name;

	// 使用LongAdder替代AtomicLong，在高并发下性能更好
	private final LongAdder totalDataSize = new LongAdder();
	private final LongAdder totalPackets = new LongAdder();

	// 配置参数
	private final int maxRetry;
	private final long cleanupThreshold;

	// 使用ArrayList存储数据
	private final List<TimestampMonitorValue> entries = new ArrayList<>();

	// 读写锁用于保护ArrayList的并发访问
	private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();

	// 时间戳验证容差（纳秒），考虑到网络延迟和时钟同步
	private static final long TIMESTAMP_TOLERANCE_NS = 10_000_000L; // 10毫秒

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

	public HighAccuracyClock getClock() {
		return clock;
	}

	protected ArrayListTimestampMonitor(HighAccuracyClock clock, String name, int maxRetry, long cleanupThreshold) {
		super();
		this.clock = clock;
		this.name = name;
		this.maxRetry = maxRetry;
		this.cleanupThreshold = cleanupThreshold;
	}

	public ArrayListTimestampMonitor(HighAccuracyClock clock, String name) {
		this(clock, name, 100, 1000000000L);
	}

	private volatile long prev = System.nanoTime();

	/*public boolean recordPacket(K key,int count, int length) {
		return recordPacket(key, clock.getCurrentTimeNanos(),count, length, Long.MAX_VALUE / 1024);
	}*/
	 public boolean recordPacket(K key,int count, int length) {
		 return recordPacket(key,clock.getCurrentTimeNanos(),count,length);
	 }
	 
	public boolean recordPacket(K key,int count, int length, long delay) {
		return recordPacket(key, clock.getCurrentTimeNanos(),count, length, delay);
	}

	public boolean recordPacket(K key, Long128 timestamp,int count, int length) {
		return recordPacket(key, timestamp,count, length, Long.MAX_VALUE / 1024);
	}

	
	/**
	 * 记录数据包 - 使用插入排序保持列表有序，添加时间戳验证
	 */
	public boolean recordPacket(K key, Long128 timestamp,int count, int length, long delay) {
		// 验证时间戳的合理性
		validateTimestamp(timestamp);

		// 快速路径：尝试sequence=0
		TimestampMonitorKey<K> monitorKey = new TimestampMonitorKey<>(key, 0);
		TimestampMonitorValue monitorValue = new TimestampMonitorValue(monitorKey, timestamp,count, length, delay);
		autoCleanup(cleanupThreshold);

		rwLock.writeLock().lock();
		try {
			// 使用二分查找插入排序，保持列表有序
			insertSorted(monitorValue);
			totalDataSize.add(length);
			totalPackets.increment();
		} finally {
			rwLock.writeLock().unlock();
		}
			return true;
	}

	/**
	 * 验证时间戳的合理性
	 */
	private void validateTimestamp(Long128 timestamp) {
		/*Long128 currentTime = clock.getCurrentTimeNanos();
		Long128 maxAllowedTime = currentTime.add(Long128.valueOf(TIMESTAMP_TOLERANCE_NS));

		if (timestamp.compareTo(maxAllowedTime) > 0) {
			throw new IllegalArgumentException(
					String.format("Timestamp %s is in the future (current time: %s, tolerance: %d ns)", timestamp,
							currentTime, TIMESTAMP_TOLERANCE_NS));
		}
	*/
	}

	/**
	 * 使用二分查找进行插入排序，保持列表有序
	 */
	private void insertSorted(TimestampMonitorValue newValue) {
		if(newValue.getDelay()<Long.MAX_VALUE/1024) {
			if(recent==null||recent.getTimestamp().compareTo( newValue.getTimestamp())<0) {
			recent=newValue;
			}
		}
		
		if (entries.isEmpty()) {
			entries.add(newValue);
			return;
		}

		// 优化：检查是否可以快速添加到末尾（大部分数据包是按时间顺序到达的）
		TimestampMonitorValue lastValue = entries.get(entries.size() - 1);
		if (newValue.getTimestamp().compareTo(lastValue.getTimestamp()) >= 0) {
			entries.add(newValue);
			return;
		}

		// 使用二分查找找到插入位置
		int insertIndex = findInsertIndex(newValue.getTimestamp());
		entries.add(insertIndex, newValue);
		
	
	}

	/**
	 * 使用二分查找找到插入位置
	 */
	private int findInsertIndex(Long128 timestamp) {
		int low = 0;
		int high = entries.size() - 1;

		while (low <= high) {
			int mid = (low + high) >>> 1;
			Long128 midTimestamp = entries.get(mid).getTimestamp();
			int cmp = timestamp.compareTo(midTimestamp);

			if (cmp < 0) {
				high = mid - 1;
			} else if (cmp > 0) {
				low = mid + 1;
			} else {
				// 时间戳相等，插入到相同时间戳的后面
				return findLastEqualIndex(mid, timestamp) + 1;
			}
		}

		return low;
	}

	/**
	 * 找到相同时间戳的最后一个元素的索引
	 */
	private int findLastEqualIndex(int startIndex, Long128 timestamp) {
		int index = startIndex;
		while (index < entries.size() - 1 && entries.get(index + 1).getTimestamp().equals(timestamp)) {
			index++;
		}
		return index;
	}

	private void autoCleanup(long th) {
		long curr = System.nanoTime();
		if (curr - prev > th) {
			prev = curr;
			cleanup();
		}
	}

	public void cleanup() {
		cleanup(clock.getCurrentTimeNanos(), cleanupThreshold);
	}

	/**
	 * 清理旧记录 - 由于列表已排序，直接使用二分查找
	 */
	private Long128 prevTime;

	public void cleanup(Long128 currentTime, long timeWindowNs) {
		Long128 startTime = currentTime.subtract(Long128.valueOf(timeWindowNs));
		Long128 maxAllowedTime = currentTime.add(Long128.valueOf(TIMESTAMP_TOLERANCE_NS));
		rwLock.writeLock().lock();
		try {
			if (prevTime != null && prevTime.compareTo(maxAllowedTime) > 0) {

				//System.out.println("Time change detected!");
				entries.clear();
			} else {
				// 使用二分查找找到第一个不过期的元素位置
				int firstValidIndex = findFirstValidIndex(startTime);

				if (firstValidIndex > 0) {
					// 删除所有过期的元素
					entries.subList(0, firstValidIndex).clear();
				}
			}
			prevTime = currentTime;
		} finally {
			rwLock.writeLock().unlock();
		}
		prev = System.nanoTime();
	}

	/**
	 * 使用二分查找找到第一个时间戳 >= startTime 的元素索引
	 */
	private int findFirstValidIndex(Long128 startTime) {
		int low = 0;
		int high = entries.size();

		while (low < high) {
			int mid = (low + high) >>> 1;
			Long128 midTimestamp = entries.get(mid).getTimestamp();

			if (midTimestamp.compareTo(startTime) < 0) {
				low = mid + 1;
			} else {
				high = mid;
			}
		}

		return low;
	}

	public long calculateDataVolume(long timeWindowNs) {
		return calculateDataVolume(clock.getCurrentTimeNanos(), timeWindowNs);
	}

	/**
	 * 高性能数据量计算 - 优化：由于数据包不会来自未来，只需要查找startTime
	 */
	public long calculateDataVolume(Long128 currentTime, long timeWindowNs) {
		Long128 startTime = currentTime.subtract(Long128.valueOf(timeWindowNs));

		rwLock.readLock().lock();
		try {
			return calculateDataVolumeFromSortedList(entries, startTime);
		} finally {
			rwLock.readLock().unlock();
		}
	}

	/**
	 * 从已排序的列表中计算数据量 - 优化版本，只需要startTime
	 */
	private long calculateDataVolumeFromSortedList(List<TimestampMonitorValue> sortedList, Long128 startTime) {
		if (sortedList.isEmpty()) {
			return 0;
		}

		// 使用二分查找找到第一个 >= startTime 的元素位置
		int startIndex = findFirstValidIndexFromSorted(sortedList, startTime);

		if (startIndex >= sortedList.size()) {
			return 0;
		}

		// 计算从startIndex到末尾的所有数据量（因为数据包不会来自未来）
		long sum = 0;
		for (int i = startIndex; i < sortedList.size(); i++) {
			sum += sortedList.get(i).getLength();
		}

		return sum;
	}

	/**
	 * 在已排序列表中找到第一个时间戳 >= startTime 的索引
	 */
	private int findFirstValidIndexFromSorted(List<TimestampMonitorValue> sortedList, Long128 startTime) {
		int low = 0;
		int high = sortedList.size();
		//int x=0;
		while (low < high) {
			int mid = (low + high) >>> 1;
			Long128 midTimestamp = sortedList.get(mid).getTimestamp();

			if (midTimestamp.compareTo(startTime) < 0) {
				low = mid + 1;
			} else {
				high = mid;
			}
			//x++;
		}
		//System.out.println(x+" "+sortedList.size());
		return low;
	}

	public long calculatePacketCount(long timeWindowNs) {
		return calculatePacketCount(clock.getCurrentTimeNanos(), timeWindowNs);
	}

	/**
	 * 高性能包数计算 - 优化：由于数据包不会来自未来，只需要查找startTime
	 */
	public long calculatePacketCount(Long128 currentTime, long timeWindowNs) {
		Long128 startTime = currentTime.subtract(Long128.valueOf(timeWindowNs));

		rwLock.readLock().lock();
		try {
			return calculatePacketCountFromSortedList(entries, startTime);
		} finally {
			rwLock.readLock().unlock();
		}
	}

	/**
	 * 从已排序的列表中计算包数 - 优化版本，只需要startTime
	 */
	/*private long calculatePacketCountFromSortedList(List<TimestampMonitorValue> sortedList, Long128 startTime) {
		if (sortedList.isEmpty()) {
			return 0;
		}

		int startIndex = findFirstValidIndexFromSorted(sortedList, startTime);

		if (startIndex >= sortedList.size()) {
			return 0;
		}

		// 计算从startIndex到末尾的所有包数（因为数据包不会来自未来）
		return sortedList.size() - startIndex;
	}*/
	
	private long calculatePacketCountFromSortedList(List<TimestampMonitorValue> sortedList, Long128 startTime) {
	
	if (sortedList.isEmpty()) {
		return 0;
	}

	// 使用二分查找找到第一个 >= startTime 的元素位置
	int startIndex = findFirstValidIndexFromSorted(sortedList, startTime);

	if (startIndex >= sortedList.size()) {
		return 0;
	}

	// 计算从startIndex到末尾的所有数据量（因为数据包不会来自未来）
	long sum = 0;
	for (int i = startIndex; i < sortedList.size(); i++) {
		sum += sortedList.get(i).getCount();
	}

	return sum;
	}
	
	public long calculateBandwidth(long timeWindowNs) {
		return calculateBandwidth(clock.getCurrentTimeNanos(), timeWindowNs);
	}

	/**
	 * 计算带宽
	 */
	public long calculateBandwidth(Long128 currentTime, long timeWindowNs) {
		if (timeWindowNs <= 0) {
			return 0;
		}

		long dataVolume = calculateDataVolume(currentTime, timeWindowNs);
		return (long) (dataVolume * 1000000000.0 / timeWindowNs);
	}

	public TrafficStats calculateTrafficStats(long timeWindowNs) {
		return calculateTrafficStats(clock.getCurrentTimeNanos(), timeWindowNs);
	}

	@Override
	public long calculatePacketRate(long timeWindowNs) {
		return calculatePacketRate(clock.getCurrentTimeNanos(), timeWindowNs);
	}

	@Override
	public long calculatePacketRate(Long128 currentTime, long timeWindowNs) {
		if (timeWindowNs <= 0) {
			return 0;
		}

		long dataVolume = calculatePacketCount(currentTime, timeWindowNs);
		return (long) (dataVolume * 1000000000.0 / timeWindowNs);
	}

	/**
	 * 计算流量统计信息 - 优化：由于数据包不会来自未来，只需要查找startTime
	 */
	public TrafficStats calculateTrafficStats(Long128 currentTime, long timeWindowNs) {
		Long128 startTime = currentTime.subtract(Long128.valueOf(timeWindowNs));

		rwLock.readLock().lock();
		try {
			return calculateTrafficStatsFromSortedList(entries, startTime, currentTime, timeWindowNs);
		} finally {
			rwLock.readLock().unlock();
		}
	}

	/**
	 * 从已排序的列表中计算流量统计 - 优化版本
	 */
	private TrafficStats calculateTrafficStatsFromSortedList(List<TimestampMonitorValue> sortedList, Long128 startTime,
			Long128 currentTime, long timeWindowNs) {
		TrafficStats stats = new TrafficStats();
		stats.timeWindowNs = timeWindowNs;
		stats.currentTime = currentTime;

		if (sortedList.isEmpty()) {
			return stats;
		}

		int startIndex = findFirstValidIndexFromSorted(sortedList, startTime);

		if (startIndex >= sortedList.size()) {
			return stats;
		}

		// 计算从startIndex到末尾的所有数据（因为数据包不会来自未来）
		for (int i = startIndex; i < sortedList.size(); i++) {
			TimestampMonitorValue value = sortedList.get(i);
			stats.packetCount+=value.getCount();
			stats.totalBytes += value.getLength();

			if (stats.earliestTimestamp.equals(Long128.ZERO)
					|| value.getTimestamp().compareTo(stats.earliestTimestamp) < 0) {
				stats.earliestTimestamp = value.getTimestamp();
			}
			if (value.getTimestamp().compareTo(stats.latestTimestamp) > 0) {
				stats.latestTimestamp = value.getTimestamp();
			}
		}

		return stats;
	}

	@Override
	public long getRecentDelay() {
		TimestampMonitorValue val= getRecentDelayEntry();
		if(val!=null) {
			return val.getDelay();
		}else {
			return Long.MAX_VALUE / 1024;
		}
	}
	
	private volatile TimestampMonitorValue recent=null;
	@Override
	public TimestampMonitorValue getRecentDelayEntry() {
		return recent;
	}

	// Getters
	public long getTotalDataSize() {
		return totalDataSize.sum();
	}

	public long getTotalPackets() {
		return totalPackets.sum();
	}

	public int getEntryCount() {
		rwLock.readLock().lock();
		try {
			return entries.size();
		} finally {
			rwLock.readLock().unlock();
		}
	}

	public void reset() {
		rwLock.writeLock().lock();
		try {
			totalDataSize.reset();
			totalPackets.reset();
			entries.clear();
		} finally {
			rwLock.writeLock().unlock();
		}
	}

	@Override
	public String toString() {
		StringBuilder sb = new StringBuilder();
		sb.append(name);
		sb.append('\n');
		sb.append(KLALBUtils.convertIBUint(totalDataSize.sum())).append("\t")
				.append(KLALBUtils.convertIBUint(calculateBandwidth(Math.min(cleanupThreshold, 1000000000L))))
				.append("/s\t");
		return sb.toString();
	}

	public static void main(String[] args) throws InterruptedException {
		AtomicLong along = new AtomicLong(0);
		ArrayListTimestampMonitor<Long> monitor = new ArrayListTimestampMonitor<>(new HighAccuracyClock(), "Monitor",
				100, 200000000L);
		Thread t = new Thread(() -> {
			while (true) {
				monitor.recordPacket(along.getAndIncrement(),1, 1000);
			}
		});
		t.start();
		Timer tx = new Timer();
		tx.scheduleAtFixedRate(new TimerTask() {

			@Override
			public void run() {
				// 计算带宽
				long testEndTime = System.nanoTime();
				monitor.cleanup();
				// 计算统计信息
				// TrafficStats stats = monitor.calculateTrafficStats(1000000000L);
				System.out.println(monitor);
				System.out.println(KLALBUtils.convertUintNanoDefalut(System.nanoTime() - testEndTime));
			}
		}, 100, 100);

	}

}