초·중·고 분산 이벤트 스트리밍 교육과 국방 전술 센서 네트워크 복합 이벤트 처리(CEP): ZeliDesk의 온디바이스 순수 C# 실시간 스트림 파이프라인, 0-Alloc 슬라이딩 윈도우(Sliding Window) 집계, 및 1,000만 이벤트/초 초저지연 시계열 CEP 엔진 아키텍처

using System;
using System.Runtime.CompilerServices;
using System.Runtime.InteropServices;
using System.Threading;

namespace ZeliDesk.Core.Streaming.CEP
{
    [StructLayout(LayoutKind.Sequential, Pack = 8)]
    public struct SensorEvent
    {
        public long TimestampNs; // 나노초 타임스탬프
        public int SensorId;     // 센서 고유 식별자
        public int EventType;    // 1: 음향, 2: 지진동, 3: 열선, 4: RF
        public float Value;      // 진폭 또는 신호 강도
    }

    public sealed class LockFreeRingBuffer
    {
        private readonly int _bufferSize;
        private readonly int _mask;
        private readonly SensorEvent[] _entries;
        private long _cursor = -1;

        public LockFreeRingBuffer(int powerOfTwoSize = 1048576) // 1M 슬롯
        {
            _bufferSize = powerOfTwoSize;
            _mask = powerOfTwoSize - 1;
            _entries = new SensorEvent[powerOfTwoSize];
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public long Next() => Interlocked.Increment(ref _cursor);

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public void Publish(long sequence, ref SensorEvent evt)
        {
            _entries[sequence & _mask] = evt;
        }

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public ref readonly SensorEvent Get(long sequence) => ref _entries[sequence & _mask];
    }

    public sealed class SlidingWindowAggregator
    {
        private readonly float[] _windowValues;
        private readonly int _capacity;
        private int _head = 0;
        private int _count = 0;
        private float _runningSum = 0f;

        public SlidingWindowAggregator(int windowCapacity)
        {
            _capacity = windowCapacity;
            _windowValues = new float[windowCapacity];
        }

        [MethodImpl(MethodImplOptions.AggressiveOptimization)]
        public float PushAndGetMean(float newVal)
        {
            if (_count == _capacity)
            {
                _runningSum -= _windowValues[_head]; // 이전 값 제거 (O(1))
            }
            else
            {
                _count++;
            }

            _windowValues[_head] = newVal;
            _runningSum += newVal; // 신규 값 가산 (O(1))

            _head = (_head + 1) % _capacity;
            return _runningSum / _count;
        }
    }

    public sealed class TacticalPatternMatcher
    {
        // 패턴: 음향 감지(Type 1) 후 500ms 이내에 열선 감지(Type 3) 시 침투 경보
        private long _lastAcousticTime = 0;
        private const long MaxDeltaNs = 500_000_000; // 500 ms

        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        public bool ProcessEvent(in SensorEvent evt)
        {
            if (evt.EventType == 1 && evt.Value > 70.0f) // 음향 임계치 초과
            {
                _lastAcousticTime = evt.TimestampNs;
                return false;
            }

            if (evt.EventType == 3 && evt.Value > 50.0f) // 열선 임계치 초과
            {
                if (_lastAcousticTime > 0 && (evt.TimestampNs - _lastAcousticTime) <= MaxDeltaNs)
                {
                    _lastAcousticTime = 0; // 매칭 성공 후 리셋
                    return true; // 침투 복합 이벤트 발생!
                }
            }

            return false;
        }
    }
}