1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

-- SQLite schema optimized for edge computing
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
PRAGMA cache_size = 10000;
PRAGMA temp_store = memory;
PRAGMA mmap_size = 268435456; -- 256MB

-- Core tables for edge operations
CREATE TABLE IF NOT EXISTS agents (
    id TEXT PRIMARY KEY,
    name TEXT NOT NULL,
    type TEXT NOT NULL,
    status TEXT DEFAULT 'inactive',
    config TEXT, -- JSON configuration
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    last_sync DATETIME,
    sync_status TEXT DEFAULT 'pending', -- pending, synced, conflict, error
    version INTEGER DEFAULT 1
);

CREATE TABLE IF NOT EXISTS workflows (
    id TEXT PRIMARY KEY,
    name TEXT NOT NULL,
    definition TEXT, -- JSON workflow definition
    status TEXT DEFAULT 'draft',
    agent_id TEXT,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    last_execution DATETIME,
    execution_count INTEGER DEFAULT 0,
    sync_status TEXT DEFAULT 'pending',
    version INTEGER DEFAULT 1,
    FOREIGN KEY (agent_id) REFERENCES agents(id)
);

CREATE TABLE IF NOT EXISTS executions (
    id TEXT PRIMARY KEY,
    workflow_id TEXT NOT NULL,
    status TEXT DEFAULT 'queued',
    input_data TEXT, -- JSON input
    output_data TEXT, -- JSON output
    error_message TEXT,
    started_at DATETIME,
    completed_at DATETIME,
    duration_ms INTEGER,
    resource_usage TEXT, -- JSON metrics
    sync_status TEXT DEFAULT 'pending',
    FOREIGN KEY (workflow_id) REFERENCES workflows(id)
);

CREATE TABLE IF NOT EXISTS sync_queue (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    table_name TEXT NOT NULL,
    record_id TEXT NOT NULL,
    operation TEXT NOT NULL, -- INSERT, UPDATE, DELETE
    data TEXT, -- JSON data
    priority INTEGER DEFAULT 5, -- 1-10, higher is more important
    attempts INTEGER DEFAULT 0,
    max_attempts INTEGER DEFAULT 5,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    last_attempt DATETIME,
    error_message TEXT,
    status TEXT DEFAULT 'pending' -- pending, processing, completed, failed
);

CREATE TABLE IF NOT EXISTS conflict_resolution (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    table_name TEXT NOT NULL,
    record_id TEXT NOT NULL,
    local_version INTEGER,
    remote_version INTEGER,
    local_data TEXT, -- JSON
    remote_data TEXT, -- JSON
    resolution_strategy TEXT, -- auto, manual, local_wins, remote_wins
    resolved_data TEXT, -- JSON
    resolved_at DATETIME,
    resolved_by TEXT, -- system, user_id
    status TEXT DEFAULT 'pending' -- pending, resolved, escalated
);

-- Performance indexes
CREATE INDEX IF NOT EXISTS idx_agents_status ON agents(status);
CREATE INDEX IF NOT EXISTS idx_agents_sync_status ON agents(sync_status);
CREATE INDEX IF NOT EXISTS idx_workflows_agent_id ON workflows(agent_id);
CREATE INDEX IF NOT EXISTS idx_executions_workflow_id ON executions(workflow_id);
CREATE INDEX IF NOT EXISTS idx_executions_status ON executions(status);
CREATE INDEX IF NOT EXISTS idx_sync_queue_status ON sync_queue(status);
CREATE INDEX IF NOT EXISTS idx_sync_queue_priority ON sync_queue(priority DESC);
CREATE INDEX IF NOT EXISTS idx_conflict_status ON conflict_resolution(status);

-- Triggers for automatic timestamp updates
CREATE TRIGGER IF NOT EXISTS update_agents_timestamp 
    AFTER UPDATE ON agents
BEGIN
    UPDATE agents SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id;
END;

CREATE TRIGGER IF NOT EXISTS update_workflows_timestamp 
    AFTER UPDATE ON workflows
BEGIN
    UPDATE workflows SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id;
END;

-- Automatic sync queue population
CREATE TRIGGER IF NOT EXISTS agents_sync_trigger 
    AFTER INSERT OR UPDATE ON agents
    WHEN NEW.sync_status = 'pending'
BEGIN
    INSERT INTO sync_queue (table_name, record_id, operation, data, priority)
    VALUES ('agents', NEW.id, 
           CASE WHEN OLD.id IS NULL THEN 'INSERT' ELSE 'UPDATE' END,
           json_object(
               'id', NEW.id,
               'name', NEW.name,
               'type', NEW.type,
               'status', NEW.status,
               'config', NEW.config,
               'version', NEW.version
           ),
           CASE WHEN NEW.status = 'critical' THEN 10 ELSE 5 END
    );
END;

-- Views for common queries
CREATE VIEW IF NOT EXISTS active_agents AS
SELECT * FROM agents 
WHERE status IN ('active', 'running', 'processing')
ORDER BY updated_at DESC;

CREATE VIEW IF NOT EXISTS pending_sync AS
SELECT table_name, record_id, operation, priority, created_at
FROM sync_queue 
WHERE status = 'pending'
ORDER BY priority DESC, created_at ASC;

CREATE VIEW IF NOT EXISTS conflict_summary AS
SELECT 
    table_name,
    COUNT(*) as conflict_count,
    MIN(created_at) as oldest_conflict,
    MAX(created_at) as newest_conflict
FROM conflict_resolution 
WHERE status = 'pending'
GROUP BY table_name;

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364

# Python SQLite optimization and connection management
import sqlite3
import json
import threading
import time
from contextlib import contextmanager
from typing import Dict, List, Any, Optional, Union
from dataclasses import dataclass
from pathlib import Path
import logging

class SQLiteManager:
    def __init__(self, db_path: Path, pool_size: int = 5):
        self.db_path = db_path
        self.pool_size = pool_size
        self._connection_pool = []
        self._pool_lock = threading.Lock()
        self._local = threading.local()
        
        # Create database and tables
        self._initialize_database()
        
        # Start background optimization
        self._start_optimization_thread()
    
    def _initialize_database(self):
        """Initialize database with optimized settings"""
        with self._get_connection() as conn:
            # Apply performance optimizations
            conn.execute("PRAGMA journal_mode = WAL")
            conn.execute("PRAGMA synchronous = NORMAL") 
            conn.execute("PRAGMA cache_size = 10000")
            conn.execute("PRAGMA temp_store = MEMORY")
            conn.execute("PRAGMA mmap_size = 268435456")  # 256MB
            conn.execute("PRAGMA page_size = 4096")
            conn.execute("PRAGMA auto_vacuum = INCREMENTAL")
            
            # Enable foreign keys
            conn.execute("PRAGMA foreign_keys = ON")
            
            # Load schema from file or create tables
            self._create_tables(conn)
    
    @contextmanager
    def _get_connection(self):
        """Get connection from pool or create new one"""
        conn = None
        try:
            with self._pool_lock:
                if self._connection_pool:
                    conn = self._connection_pool.pop()
                else:
                    conn = sqlite3.connect(
                        str(self.db_path),
                        timeout=30.0,
                        check_same_thread=False
                    )
                    conn.row_factory = sqlite3.Row
            
            yield conn
            
        finally:
            if conn:
                with self._pool_lock:
                    if len(self._connection_pool) < self.pool_size:
                        self._connection_pool.append(conn)
                    else:
                        conn.close()
    
    def insert_agent(self, agent_data: Dict[str, Any]) -> str:
        """Insert new agent with automatic sync queue entry"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            
            try:
                cursor.execute("""
                    INSERT INTO agents (id, name, type, status, config, version)
                    VALUES (?, ?, ?, ?, ?, ?)
                """, (
                    agent_data['id'],
                    agent_data['name'],
                    agent_data['type'],
                    agent_data.get('status', 'inactive'),
                    json.dumps(agent_data.get('config', {})),
                    agent_data.get('version', 1)
                ))
                
                conn.commit()
                return agent_data['id']
                
            except sqlite3.IntegrityError as e:
                conn.rollback()
                raise Exception(f"Agent with ID {agent_data['id']} already exists")
    
    def update_agent(self, agent_id: str, updates: Dict[str, Any]) -> bool:
        """Update agent with version control and conflict detection"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            
            try:
                # Get current version
                cursor.execute("SELECT version FROM agents WHERE id = ?", (agent_id,))
                result = cursor.fetchone()
                
                if not result:
                    return False
                
                current_version = result['version']
                expected_version = updates.get('expected_version', current_version)
                
                if current_version != expected_version:
                    # Version conflict detected
                    self._create_conflict_record(
                        conn, 'agents', agent_id, 
                        expected_version, current_version,
                        updates
                    )
                    return False
                
                # Build update query
                set_clauses = []
                values = []
                
                for key, value in updates.items():
                    if key in ['id', 'expected_version']:
                        continue
                    
                    if key == 'config' and isinstance(value, dict):
                        value = json.dumps(value)
                    
                    set_clauses.append(f"{key} = ?")
                    values.append(value)
                
                set_clauses.append("version = version + 1")
                set_clauses.append("sync_status = 'pending'")
                
                values.append(agent_id)
                
                cursor.execute(f"""
                    UPDATE agents 
                    SET {', '.join(set_clauses)}
                    WHERE id = ?
                """, values)
                
                conn.commit()
                return cursor.rowcount > 0
                
            except Exception as e:
                conn.rollback()
                raise
    
    def get_agents(self, 
                  status: Optional[str] = None, 
                  limit: Optional[int] = None,
                  offset: int = 0) -> List[Dict[str, Any]]:
        """Get agents with filtering and pagination"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            
            query = "SELECT * FROM agents"
            params = []
            
            if status:
                query += " WHERE status = ?"
                params.append(status)
            
            query += " ORDER BY updated_at DESC"
            
            if limit:
                query += " LIMIT ? OFFSET ?"
                params.extend([limit, offset])
            
            cursor.execute(query, params)
            
            agents = []
            for row in cursor.fetchall():
                agent = dict(row)
                if agent['config']:
                    agent['config'] = json.loads(agent['config'])
                agents.append(agent)
            
            return agents
    
    def get_sync_queue(self, status: str = 'pending', limit: int = 100) -> List[Dict[str, Any]]:
        """Get items from sync queue ordered by priority"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            
            cursor.execute("""
                SELECT * FROM sync_queue 
                WHERE status = ?
                ORDER BY priority DESC, created_at ASC
                LIMIT ?
            """, (status, limit))
            
            return [dict(row) for row in cursor.fetchall()]
    
    def mark_sync_completed(self, sync_id: int) -> bool:
        """Mark sync queue item as completed"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            
            cursor.execute("""
                UPDATE sync_queue 
                SET status = 'completed', last_attempt = CURRENT_TIMESTAMP
                WHERE id = ?
            """, (sync_id,))
            
            conn.commit()
            return cursor.rowcount > 0
    
    def mark_sync_failed(self, sync_id: int, error_message: str) -> bool:
        """Mark sync queue item as failed and increment attempts"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            
            cursor.execute("""
                UPDATE sync_queue 
                SET attempts = attempts + 1,
                    last_attempt = CURRENT_TIMESTAMP,
                    error_message = ?,
                    status = CASE 
                        WHEN attempts + 1 >= max_attempts THEN 'failed'
                        ELSE 'pending'
                    END
                WHERE id = ?
            """, (error_message, sync_id))
            
            conn.commit()
            return cursor.rowcount > 0
    
    def _create_conflict_record(self, conn, table_name: str, record_id: str,
                              local_version: int, remote_version: int, 
                              local_data: Dict[str, Any]):
        """Create conflict resolution record"""
        cursor = conn.cursor()
        
        cursor.execute("""
            INSERT INTO conflict_resolution 
            (table_name, record_id, local_version, remote_version, local_data, status)
            VALUES (?, ?, ?, ?, ?, 'pending')
        """, (table_name, record_id, local_version, remote_version, json.dumps(local_data)))
    
    def get_conflicts(self, status: str = 'pending') -> List[Dict[str, Any]]:
        """Get unresolved conflicts"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            
            cursor.execute("""
                SELECT * FROM conflict_resolution 
                WHERE status = ?
                ORDER BY created_at ASC
            """, (status,))
            
            conflicts = []
            for row in cursor.fetchall():
                conflict = dict(row)
                if conflict['local_data']:
                    conflict['local_data'] = json.loads(conflict['local_data'])
                if conflict['remote_data']:
                    conflict['remote_data'] = json.loads(conflict['remote_data'])
                if conflict['resolved_data']:
                    conflict['resolved_data'] = json.loads(conflict['resolved_data'])
                conflicts.append(conflict)
            
            return conflicts
    
    def resolve_conflict(self, conflict_id: int, resolution_data: Dict[str, Any],
                        strategy: str = 'manual') -> bool:
        """Resolve a conflict with specified resolution"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            
            try:
                cursor.execute("""
                    UPDATE conflict_resolution 
                    SET resolved_data = ?,
                        resolution_strategy = ?,
                        resolved_at = CURRENT_TIMESTAMP,
                        status = 'resolved'
                    WHERE id = ?
                """, (json.dumps(resolution_data), strategy, conflict_id))
                
                conn.commit()
                return cursor.rowcount > 0
                
            except Exception as e:
                conn.rollback()
                raise
    
    def _start_optimization_thread(self):
        """Start background optimization thread"""
        def optimization_worker():
            while True:
                try:
                    self._run_maintenance()
                    time.sleep(300)  # Run every 5 minutes
                except Exception as e:
                    logging.error(f"SQLite optimization error: {e}")
                    time.sleep(60)
        
        thread = threading.Thread(target=optimization_worker, daemon=True)
        thread.start()
    
    def _run_maintenance(self):
        """Run database maintenance operations"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            
            # Analyze query performance
            cursor.execute("ANALYZE")
            
            # Clean up old completed sync queue entries
            cursor.execute("""
                DELETE FROM sync_queue 
                WHERE status = 'completed' 
                AND created_at < datetime('now', '-7 days')
            """)
            
            # Clean up old resolved conflicts
            cursor.execute("""
                DELETE FROM conflict_resolution 
                WHERE status = 'resolved' 
                AND resolved_at < datetime('now', '-30 days')
            """)
            
            # Incremental vacuum
            cursor.execute("PRAGMA incremental_vacuum(1000)")
            
            conn.commit()
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get database statistics and health information"""
        with self._get_connection() as conn:
            cursor = conn.cursor()
            
            stats = {}
            
            # Table counts
            cursor.execute("SELECT COUNT(*) FROM agents")
            stats['total_agents'] = cursor.fetchone()[0]
            
            cursor.execute("SELECT COUNT(*) FROM workflows")
            stats['total_workflows'] = cursor.fetchone()[0]
            
            cursor.execute("SELECT COUNT(*) FROM executions")
            stats['total_executions'] = cursor.fetchone()[0]
            
            # Sync queue statistics
            cursor.execute("SELECT status, COUNT(*) FROM sync_queue GROUP BY status")
            stats['sync_queue'] = dict(cursor.fetchall())
            
            # Conflict statistics
            cursor.execute("SELECT status, COUNT(*) FROM conflict_resolution GROUP BY status")
            stats['conflicts'] = dict(cursor.fetchall())
            
            # Database size information
            cursor.execute("PRAGMA page_count")
            page_count = cursor.fetchone()[0]
            cursor.execute("PRAGMA page_size")
            page_size = cursor.fetchone()[0]
            stats['database_size_mb'] = (page_count * page_size) / (1024 * 1024)
            
            return stats

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342

// Rust implementation for high-performance file system operations
use std::path::{Path, PathBuf};
use std::collections::HashMap;
use std::fs::{self, File, OpenOptions};
use std::io::{self, Read, Write, BufReader, BufWriter};
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::fs as tokio_fs;
use serde::{Serialize, Deserialize};
use blake3;
use lz4_flex;
use walkdir::WalkDir;

#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct FileMetadata {
    pub path: PathBuf,
    pub size: u64,
    pub modified: u64,
    pub hash: String,
    pub compressed: bool,
    pub encrypted: bool,
    pub sync_status: SyncStatus,
    pub priority: u8,
}

#[derive(Serialize, Deserialize, Debug, Clone)]
pub enum SyncStatus {
    Synced,
    Pending,
    Conflict,
    Error(String),
}

pub struct FileSystemManager {
    root_path: PathBuf,
    metadata_cache: HashMap<PathBuf, FileMetadata>,
    compression_enabled: bool,
    encryption_enabled: bool,
    max_file_size: u64,
}

impl FileSystemManager {
    pub fn new(root_path: PathBuf) -> io::Result<Self> {
        if !root_path.exists() {
            fs::create_dir_all(&root_path)?;
        }
        
        let mut manager = Self {
            root_path,
            metadata_cache: HashMap::new(),
            compression_enabled: true,
            encryption_enabled: false,
            max_file_size: 100 * 1024 * 1024, // 100MB
        };
        
        manager.load_metadata_cache()?;
        Ok(manager)
    }
    
    pub async fn store_file(&mut self, 
                           relative_path: &Path, 
                           data: &[u8], 
                           priority: u8) -> io::Result<FileMetadata> {
        
        let full_path = self.root_path.join(relative_path);
        
        // Create parent directories
        if let Some(parent) = full_path.parent() {
            tokio_fs::create_dir_all(parent).await?;
        }
        
        // Check file size limits
        if data.len() as u64 > self.max_file_size {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                format!("File too large: {} bytes", data.len())
            ));
        }
        
        // Calculate hash
        let hash = blake3::hash(data).to_string();
        
        // Prepare data for storage
        let storage_data = if self.compression_enabled && data.len() > 1024 {
            // Compress data
            let compressed = lz4_flex::compress_prepend_size(data);
            if compressed.len() < data.len() {
                (compressed, true)
            } else {
                (data.to_vec(), false)
            }
        } else {
            (data.to_vec(), false)
        };
        
        // Write to file
        tokio_fs::write(&full_path, &storage_data.0).await?;
        
        // Create metadata
        let metadata = FileMetadata {
            path: relative_path.to_path_buf(),
            size: data.len() as u64,
            modified: SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_secs(),
            hash,
            compressed: storage_data.1,
            encrypted: false, // TODO: implement encryption
            sync_status: SyncStatus::Pending,
            priority,
        };
        
        // Update cache
        self.metadata_cache.insert(relative_path.to_path_buf(), metadata.clone());
        self.save_metadata_cache()?;
        
        Ok(metadata)
    }
    
    pub async fn load_file(&self, relative_path: &Path) -> io::Result<Vec<u8>> {
        let full_path = self.root_path.join(relative_path);
        
        if !full_path.exists() {
            return Err(io::Error::new(
                io::ErrorKind::NotFound,
                "File not found"
            ));
        }
        
        let data = tokio_fs::read(&full_path).await?;
        
        // Check if file is compressed
        if let Some(metadata) = self.metadata_cache.get(relative_path) {
            if metadata.compressed {
                // Decompress data
                match lz4_flex::decompress_size_prepended(&data) {
                    Ok(decompressed) => return Ok(decompressed),
                    Err(e) => {
                        eprintln!("Decompression failed: {}", e);
                        // Fall through to return raw data
                    }
                }
            }
        }
        
        Ok(data)
    }
    
    pub fn get_pending_sync_files(&self, max_count: usize) -> Vec<FileMetadata> {
        let mut pending_files: Vec<_> = self.metadata_cache
            .values()
            .filter(|meta| matches!(meta.sync_status, SyncStatus::Pending))
            .cloned()
            .collect();
        
        // Sort by priority (highest first), then by modification time
        pending_files.sort_by(|a, b| {
            b.priority.cmp(&a.priority)
                .then(b.modified.cmp(&a.modified))
        });
        
        pending_files.into_iter().take(max_count).collect()
    }
    
    pub fn mark_synced(&mut self, relative_path: &Path) -> io::Result<()> {
        if let Some(metadata) = self.metadata_cache.get_mut(relative_path) {
            metadata.sync_status = SyncStatus::Synced;
            self.save_metadata_cache()?;
        }
        Ok(())
    }
    
    pub fn mark_conflict(&mut self, relative_path: &Path, error: String) -> io::Result<()> {
        if let Some(metadata) = self.metadata_cache.get_mut(relative_path) {
            metadata.sync_status = SyncStatus::Conflict;
            self.save_metadata_cache()?;
        }
        Ok(())
    }
    
    pub async fn cleanup_old_files(&mut self, max_age_days: u64) -> io::Result<u64> {
        let cutoff_time = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs() - (max_age_days * 24 * 3600);
        
        let mut cleaned_size = 0u64;
        let mut to_remove = Vec::new();
        
        for (path, metadata) in &self.metadata_cache {
            if metadata.modified < cutoff_time && 
               matches!(metadata.sync_status, SyncStatus::Synced) {
                
                let full_path = self.root_path.join(path);
                
                if full_path.exists() {
                    let file_size = tokio_fs::metadata(&full_path).await?.len();
                    tokio_fs::remove_file(&full_path).await?;
                    cleaned_size += file_size;
                    to_remove.push(path.clone());
                }
            }
        }
        
        // Remove from cache
        for path in to_remove {
            self.metadata_cache.remove(&path);
        }
        
        self.save_metadata_cache()?;
        Ok(cleaned_size)
    }
    
    pub fn get_storage_statistics(&self) -> io::Result<StorageStatistics> {
        let mut stats = StorageStatistics::default();
        
        // Walk through all files in root directory
        for entry in WalkDir::new(&self.root_path) {
            let entry = entry?;
            
            if entry.file_type().is_file() {
                let metadata = entry.metadata()?;
                stats.total_files += 1;
                stats.total_size += metadata.len();
                
                if let Some(relative_path) = entry.path().strip_prefix(&self.root_path).ok() {
                    if let Some(file_meta) = self.metadata_cache.get(relative_path) {
                        match file_meta.sync_status {
                            SyncStatus::Synced => stats.synced_files += 1,
                            SyncStatus::Pending => stats.pending_files += 1,
                            SyncStatus::Conflict => stats.conflict_files += 1,
                            SyncStatus::Error(_) => stats.error_files += 1,
                        }
                        
                        if file_meta.compressed {
                            stats.compressed_files += 1;
                        }
                    }
                }
            }
        }
        
        Ok(stats)
    }
    
    fn load_metadata_cache(&mut self) -> io::Result<()> {
        let cache_path = self.root_path.join(".metadata_cache.json");
        
        if cache_path.exists() {
            let data = fs::read_to_string(&cache_path)?;
            
            match serde_json::from_str::<HashMap<PathBuf, FileMetadata>>(&data) {
                Ok(cache) => {
                    self.metadata_cache = cache;
                }
                Err(e) => {
                    eprintln!("Failed to load metadata cache: {}", e);
                    // Continue with empty cache
                }
            }
        }
        
        Ok(())
    }
    
    fn save_metadata_cache(&self) -> io::Result<()> {
        let cache_path = self.root_path.join(".metadata_cache.json");
        let data = serde_json::to_string_pretty(&self.metadata_cache)?;
        fs::write(&cache_path, data)?;
        Ok(())
    }
}

#[derive(Default, Debug)]
pub struct StorageStatistics {
    pub total_files: u64,
    pub total_size: u64,
    pub synced_files: u64,
    pub pending_files: u64,
    pub conflict_files: u64,
    pub error_files: u64,
    pub compressed_files: u64,
}

// Async file operations with retry logic
pub struct AsyncFileOperations;

impl AsyncFileOperations {
    pub async fn copy_with_retry(src: &Path, dst: &Path, max_retries: u32) -> io::Result<()> {
        let mut attempts = 0;
        
        loop {
            match tokio_fs::copy(src, dst).await {
                Ok(_) => return Ok(()),
                Err(e) => {
                    attempts += 1;
                    
                    if attempts >= max_retries {
                        return Err(e);
                    }
                    
                    // Exponential backoff
                    let delay = std::time::Duration::from_millis(100 * (1 << attempts));
                    tokio::time::sleep(delay).await;
                }
            }
        }
    }
    
    pub async fn atomic_write(path: &Path, data: &[u8]) -> io::Result<()> {
        let temp_path = path.with_extension("tmp");
        
        // Write to temporary file first
        tokio_fs::write(&temp_path, data).await?;
        
        // Atomically rename to target
        tokio_fs::rename(&temp_path, path).await?;
        
        Ok(())
    }
    
    pub async fn safe_delete(path: &Path) -> io::Result<()> {
        if path.exists() {
            // Move to trash directory first
            let trash_dir = path.parent().unwrap().join(".trash");
            tokio_fs::create_dir_all(&trash_dir).await?;
            
            let timestamp = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_secs();
            
            let trash_path = trash_dir.join(format!("{}_{}", 
                path.file_name().unwrap().to_string_lossy(), 
                timestamp));
            
            tokio_fs::rename(path, &trash_path).await?;
        }
        
        Ok(())
    }
}