kawa-storage 0.1.1

//! # Kawa Storage
//! 
//! 高性能な Write-Ahead Log (WAL) ベースのストレージエンジン。
//! Kafka互換プロトコルでのイベントソーシングを支援する。
//!
//! ## 機能
//! - セグメント化されたWAL
//! - メモリマップドI/O
//! - CRCチェック付きデータ整合性保証
//! - ゼロコピー読み取り

use derive_more::{Deref, DerefMut, From, Into};
use serde::{Deserialize, Serialize};
use std::path::PathBuf;
use tokio::sync::RwLock;
use std::sync::Arc;

pub mod segment;
pub mod wal;
pub mod event;
pub mod error;
pub mod memory;

// ★ 追加: Ultra Performance Engine
pub mod ultra_performance;

// ★ 追加: GPU Acceleration Engine (Phase 2)
pub mod gpu_acceleration;

// ★ 追加: Security Module (DoS Protection)
pub mod security;

pub use error::{StorageError, StorageResult};
pub use event::{Event, EventId, EventData};
pub use segment::{Segment, SegmentId, SegmentManager};
pub use wal::{WriteAheadLog, WalConfig};
pub use memory::{MemoryPool, PooledBuffer};

// ★ 追加: Ultra Performance Engine のエクスポート
pub use ultra_performance::{
    UltraPerformanceEngine, 
    LockFreeRingBuffer, 
    SIMDBatchProcessor,
    PerformanceMetrics,
    NUMATopology,
};

// ★ 追加: GPU Acceleration Engine のエクスポート (Phase 2)
pub use gpu_acceleration::{
    GPUAcceleratedEngine,
    GPUAccelerationType,
    GPUContext,
    GPUMetrics,
    HybridConfig,
    DistributionStrategy,
};

// ★ 追加: Security Module のエクスポート (DoS Protection)
pub use security::{
    SecurityManager,
    SecurityConfig,
    SecurityMetrics,
    SecurityError,
    RateLimiter,
    ResourceMonitor,
    GPUTimeoutManager,
};

/// ストレージエンジンのメインエントリポイント
/// 
/// # Example
/// ```rust
/// use kawa_storage::{StorageEngine, StorageConfig};
/// 
/// #[tokio::main]
/// async fn main() -> anyhow::Result<()> {
///     let config = StorageConfig::default();
///     let storage = StorageEngine::new(config).await?;
///     
///     // @todo イベント書き込み・読み取りのサンプル実装
///     Ok(())
/// }
/// ```
#[derive(Debug)]
pub struct StorageEngine {
    /// WALインスタンス
    wal: RwLock<WriteAheadLog>,
    /// セグメント管理
    segment_manager: RwLock<SegmentManager>,
    /// 高性能メモリプール
    memory_pool: Arc<MemoryPool>,
    /// 設定
    config: StorageConfig,
}

/// ストレージエンジンの設定
/// 
/// # Fields
/// - `data_dir`: データディレクトリパス
/// - `segment_size`: セグメントサイズ（バイト）
/// - `sync_interval`: 同期間隔（ミリ秒）
/// - `memory_pool_size`: メモリプール初期サイズ
/// - `batch_size`: バッチ処理サイズ
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StorageConfig {
    /// データディレクトリ
    pub data_dir: PathBuf,
    /// セグメントサイズ（デフォルト: 1GB）
    pub segment_size: u64,
    /// 同期間隔（デフォルト: 1000ms）
    pub sync_interval_ms: u64,
    /// 圧縮有効フラグ
    pub enable_compression: bool,
    /// メモリプールサイズ（デフォルト: 128MB）
    pub memory_pool_size: usize,
    /// バッチサイズ（デフォルト: 1000）
    pub batch_size: usize,
    /// 並行ワーカー数（デフォルト: CPU数×2）
    pub worker_count: Option<usize>,
}

impl Default for StorageConfig {
    fn default() -> Self {
        Self {
            data_dir: PathBuf::from("./data"),
            segment_size: 1024 * 1024 * 1024, // 1GB
            sync_interval_ms: 1000,
            enable_compression: false,
            memory_pool_size: 128 * 1024 * 1024, // 128MB
            batch_size: 1000,
            worker_count: None, // 自動設定（CPU数×2）
        }
    }
}

/// オフセット型（Newtypeパターン）
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize, Deref, DerefMut, From, Into)]
pub struct Offset(pub u64);

impl Offset {
    /// 新しいオフセットを作成
    pub fn new(value: u64) -> Self {
        Self(value)
    }
    
    /// 次のオフセットを取得
    pub fn next(self) -> Self {
        Self(self.0 + 1)
    }
}

/// パーティション型（Newtypeパターン）
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Deref, DerefMut, From, Into)]
pub struct Partition(pub u32);

impl Partition {
    /// 新しいパーティションを作成
    pub fn new(value: u32) -> Self {
        Self(value)
    }
}

/// トピック型（Newtypeパターン）
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, Deref, DerefMut, From, Into)]
pub struct Topic(pub String);

impl Topic {
    /// 新しいトピックを作成
    pub fn new(name: impl Into<String>) -> Self {
        Self(name.into())
    }
}

impl StorageEngine {
    /// 新しいストレージエンジンを作成
    /// 
    /// # Arguments
    /// * `config` - ストレージ設定
    /// 
    /// # Returns
    /// * `Result<Self>` - ストレージエンジンインスタンス
    /// 
    /// # @todo
    /// - [ ] データディレクトリの自動作成
    /// - [ ] 既存データの復旧処理
    /// - [ ] メトリクス収集の初期化
    pub async fn new(config: StorageConfig) -> StorageResult<Self> {
        // データディレクトリを作成
        tokio::fs::create_dir_all(&config.data_dir).await?;
        
        // 高性能メモリプールを初期化
        let memory_pool = Arc::new(MemoryPool::new(
            config.memory_pool_size,
            config.batch_size,
        )?);
        
        let wal_config = WalConfig {
            data_dir: config.data_dir.clone(),
            segment_size: config.segment_size,
            sync_interval_ms: config.sync_interval_ms,
        };
        
        let wal = WriteAheadLog::new(wal_config).await?;
        let segment_manager = SegmentManager::new(config.data_dir.clone()).await?;
        
        Ok(Self {
            wal: RwLock::new(wal),
            segment_manager: RwLock::new(segment_manager),
            memory_pool,
            config,
        })
    }
    
    /// イベントを永続化
    /// 
    /// # Arguments
    /// * `topic` - トピック名
    /// * `partition` - パーティション番号
    /// * `event_data` - イベントデータ
    /// 
    /// # Returns
    /// * `Result<(EventId, Offset)>` - イベントIDとオフセット
    /// 
    /// # @todo
    /// - [ ] バッチ書き込みのサポート
    /// - [ ] 圧縮機能の実装
    /// - [ ] レプリケーション対応
    pub async fn append_event(
        &self,
        topic: &Topic,
        partition: Partition,
        event_data: EventData,
    ) -> StorageResult<(EventId, Offset)> {
        let event_id = EventId::new();
        let event = Event::new(event_id, topic.clone(), partition, event_data);
        
        let mut wal = self.wal.write().await;
        let offset = wal.append(&event).await?;
        
        Ok((event_id, offset))
    }
    
    /// 指定したオフセット範囲のイベントを読み取り
    /// 
    /// # Arguments
    /// * `topic` - トピック名
    /// * `partition` - パーティション番号
    /// * `start_offset` - 開始オフセット
    /// * `max_events` - 最大イベント数
    /// 
    /// # Returns
    /// * `Result<Vec<Event>>` - イベントリスト
    /// 
    /// # @todo
    /// - [ ] ストリーミング読み取り
    /// - [ ] インデックスベース高速検索
    /// - [ ] 非同期イテレータの実装
    pub async fn read_events(
        &self,
        topic: &Topic,
        partition: Partition,
        start_offset: Offset,
        max_events: usize,
    ) -> StorageResult<Vec<Event>> {
        let wal = self.wal.read().await;
        wal.read_events(topic, partition, start_offset, max_events).await
    }
    
    /// 最新のオフセットを取得
    /// 
    /// # Arguments
    /// * `topic` - トピック名
    /// * `partition` - パーティション番号
    /// 
    /// # Returns
    /// * `Result<Option<Offset>>` - 最新オフセット（データがない場合はNone）
    pub async fn get_latest_offset(
        &self,
        topic: &Topic,
        partition: Partition,
    ) -> StorageResult<Option<Offset>> {
        let wal = self.wal.read().await;
        wal.get_latest_offset(topic, partition).await
    }
    
    /// バッチでイベントを永続化（高性能版）
    /// 
    /// # Arguments
    /// * `events` - イベントのバッチ
    /// 
    /// # Returns
    /// * `Result<Vec<(EventId, Offset)>>` - イベントIDとオフセットのリスト
    /// 
    /// # Performance
    /// 1M+ events/secを達成するための最適化バッチ処理
    pub async fn append_events_batch(
        &self,
        events: Vec<(Topic, Partition, EventData)>,
    ) -> StorageResult<Vec<(EventId, Offset)>> {
        if events.is_empty() {
            return Ok(Vec::new());
        }
        
        // メモリプールからバッファを取得
        let mut pooled_buffer = self.memory_pool.get_buffer().await?;
        
        // バッチ用のイベントリストを準備
        let mut batch_events = Vec::with_capacity(events.len());
        let mut result_ids = Vec::with_capacity(events.len());
        
        for (topic, partition, event_data) in events {
            let event_id = EventId::new();
            let event = Event::new(event_id, topic, partition, event_data);
            
            batch_events.push(event);
            result_ids.push(event_id);
        }
        
        // WALにバッチ書き込み
        let mut wal = self.wal.write().await;
        let offsets = wal.append_batch(&batch_events, &mut pooled_buffer).await?;
        
        // 結果を構築
        let results: Vec<(EventId, Offset)> = result_ids.into_iter()
            .zip(offsets.into_iter())
            .collect();
        
        Ok(results)
    }
    
    /// メモリプール統計を取得
    pub fn memory_pool_stats(&self) -> memory::PoolStats {
        self.memory_pool.stats()
    }
    
    /// ストレージエンジンを優雅にシャットダウン
    /// 
    /// # @todo
    /// - [ ] 未書き込みデータのフラッシュ
    /// - [ ] メトリクスの最終出力
    /// - [ ] リソースクリーンアップ
    pub async fn shutdown(&self) -> StorageResult<()> {
        let wal = self.wal.write().await;
        wal.sync().await?;
        
        tracing::info!("Storage engine shutdown completed");
        Ok(())
    }
}