diff --git a/samples/EntglDb.Test.Maui/MauiProgram.cs b/samples/EntglDb.Test.Maui/MauiProgram.cs index 62d881a..b3e0e2c 100644 --- a/samples/EntglDb.Test.Maui/MauiProgram.cs +++ b/samples/EntglDb.Test.Maui/MauiProgram.cs @@ -74,7 +74,9 @@ public static MauiApp CreateMauiApp() { NodeId = $"CHANGEME-{nodeId}", TcpPort = 5001, - AuthToken = "Test-Cluster-Key" + AuthToken = "Test-Cluster-Key", + OplogRetentionHours = 2, + MaintenanceIntervalMinutes = 5 }); builder.Services.AddSingleton(peerNodeConfigurationProvider); diff --git a/src/EntglDb.Core/Network/PeerNodeConfiguration.cs b/src/EntglDb.Core/Network/PeerNodeConfiguration.cs index f401670..c1522f4 100644 --- a/src/EntglDb.Core/Network/PeerNodeConfiguration.cs +++ b/src/EntglDb.Core/Network/PeerNodeConfiguration.cs @@ -45,6 +45,16 @@ public class PeerNodeConfiguration /// public int RetryDelayMs { get; set; } = 1000; + /// + /// Interval between periodic maintenance operations (Oplog pruning) in minutes. Default: 60 minutes. + /// + public int MaintenanceIntervalMinutes { get; set; } = 60; + + /// + /// Oplog retention period in hours. Entries older than this will be pruned. Default: 24 hours. + /// + public int OplogRetentionHours { get; set; } = 24; + /// /// Gets the default configuration settings for a peer node. /// diff --git a/src/EntglDb.Core/Storage/IPeerStore.cs b/src/EntglDb.Core/Storage/IPeerStore.cs index 2d19355..664beee 100644 --- a/src/EntglDb.Core/Storage/IPeerStore.cs +++ b/src/EntglDb.Core/Storage/IPeerStore.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.IO; using System.Threading; using System.Threading.Tasks; using EntglDb.Core; // Added for ChangesAppliedEventArgs @@ -107,4 +108,40 @@ public interface IPeerStore /// The unique identifier of the peer to remove. /// Cancellation token. Task RemoveRemotePeerAsync(string nodeId, CancellationToken cancellationToken = default); + + // Snapshot & Maintenance routines + + /// + /// Prunes the oplog, removing entries older than the specified timestamp. + /// Preserves the latest state in SnapshotMetadata to maintain chain continuity. + /// + Task PruneOplogAsync(HlcTimestamp cutoff, CancellationToken cancellationToken = default); + + /// + /// Creates a full snapshot of the underlying database and writes it to the destination stream. + /// + Task CreateSnapshotAsync(Stream destination, CancellationToken cancellationToken = default); + + /// + /// Replaces the underlying database with the provided stream. + /// This is used for full sync/snapshot recovery. + /// WARNING: This will overwrite all local data. + /// + Task ReplaceDatabaseAsync(Stream databaseStream, CancellationToken cancellationToken = default); + + /// + /// Merges a remote snapshot into the local database without overwriting existing data. + /// Used for Split-Brain resolution. + /// + Task MergeSnapshotAsync(Stream snapshotStream, CancellationToken cancellationToken = default); + + /// + /// Clears all data from the store, resetting it to an empty state. + /// + Task ClearAllDataAsync(CancellationToken cancellationToken = default); +} + +public class CorruptDatabaseException : Exception +{ + public CorruptDatabaseException(string message, Exception innerException) : base(message, innerException) { } } diff --git a/src/EntglDb.Network/SyncOrchestrator.cs b/src/EntglDb.Network/SyncOrchestrator.cs index 4081e56..3827799 100644 --- a/src/EntglDb.Network/SyncOrchestrator.cs +++ b/src/EntglDb.Network/SyncOrchestrator.cs @@ -7,6 +7,7 @@ using System; using System.Collections.Concurrent; using System.Collections.Generic; +using System.IO; using System.Linq; using System.Net.Sockets; using System.Threading; @@ -31,9 +32,17 @@ public class SyncOrchestrator : ISyncOrchestrator // Persistent clients pool private readonly ConcurrentDictionary _clients = new(); + private readonly ConcurrentDictionary _peerStates = new(); private readonly IPeerHandshakeService? _handshakeService; - private readonly EntglDb.Network.Telemetry.INetworkTelemetryService? _telemetry; + private readonly INetworkTelemetryService? _telemetry; + private class PeerStatus + { + public int FailureCount { get; set; } + public DateTime NextRetryTime { get; set; } + } + + private DateTime _lastMaintenanceTime = DateTime.MinValue; public SyncOrchestrator( IDiscoveryService discovery, @@ -41,7 +50,7 @@ public SyncOrchestrator( IPeerNodeConfigurationProvider peerNodeConfigurationProvider, ILoggerFactory loggerFactory, IPeerHandshakeService? handshakeService = null, - EntglDb.Network.Telemetry.INetworkTelemetryService? telemetry = null) + INetworkTelemetryService? telemetry = null) { _discovery = discovery; _store = store; @@ -137,14 +146,44 @@ private async Task SyncLoopAsync(CancellationToken token) var config = await _peerNodeConfigurationProvider.GetConfiguration(); try { - var peers = _discovery.GetActivePeers().Where(p => p.NodeId != config.NodeId).ToList(); + var allPeers = _discovery.GetActivePeers().Where(p => p.NodeId != config.NodeId).ToList(); + + // Filter peers based on backoff + var now = DateTime.UtcNow; + var eligiblePeers = allPeers.Where(p => + { + if (_peerStates.TryGetValue(p.NodeId, out var status)) + { + return status.NextRetryTime <= now; + } + return true; + }).ToList(); - // Gossip Fanout: Pick 3 random peers - var targets = peers.OrderBy(x => _random.Next()).Take(3).ToList(); + // Gossip Fanout: Pick 3 random peers from eligible set + var targets = eligiblePeers.OrderBy(x => _random.Next()).Take(3).ToList(); // NetStandard 2.0 fallback: Use Task.WhenAll var tasks = targets.Select(peer => TrySyncWithPeer(peer, token)); await Task.WhenAll(tasks); + + // Periodic Maintenance: Prune Oplog based on configuration + var maintenanceInterval = TimeSpan.FromMinutes(config.MaintenanceIntervalMinutes); + if ((now - _lastMaintenanceTime) >= maintenanceInterval) + { + _logger.LogInformation("Running periodic maintenance (Oplog pruning)..."); + try + { + var retentionHours = config.OplogRetentionHours; + var cutoff = new HlcTimestamp(DateTimeOffset.UtcNow.AddHours(-retentionHours).ToUnixTimeMilliseconds(), 0, config.NodeId); + await _store.PruneOplogAsync(cutoff, token); + _lastMaintenanceTime = now; + _logger.LogInformation("Maintenance completed successfully (Retention: {RetentionHours}h).", retentionHours); + } + catch (Exception maintenanceEx) + { + _logger.LogError(maintenanceEx, "Maintenance failed."); + } + } } catch (OperationCanceledException) { @@ -176,6 +215,7 @@ private async Task TrySyncWithPeer(PeerNode peer, CancellationToken token) { TcpPeerClient? client = null; bool shouldRemoveClient = false; + bool syncSuccessful = false; try { @@ -199,7 +239,7 @@ private async Task TrySyncWithPeer(PeerNode peer, CancellationToken token) { _logger.LogWarning("Handshake rejected by {NodeId}", peer.NodeId); shouldRemoveClient = true; - return; + throw new Exception("Handshake rejected"); } // 1. Exchange Vector Clocks @@ -229,7 +269,13 @@ private async Task TrySyncWithPeer(PeerNode peer, CancellationToken token) var changes = await client.PullChangesFromNodeAsync(nodeId, localTs, token); if (changes != null && changes.Count > 0) { - await ProcessInboundBatchAsync(client, peer.NodeId, changes, token); + var result = await ProcessInboundBatchAsync(client, peer.NodeId, changes, token); + if (result != SyncBatchResult.Success) + { + _logger.LogWarning("Inbound batch processing failed with status {Status}. Aborting sync for this session.", result); + RecordFailure(peer.NodeId); + return; + } } } } @@ -264,21 +310,77 @@ private async Task TrySyncWithPeer(PeerNode peer, CancellationToken token) { _logger.LogDebug("Vector clocks are concurrent with {PeerNodeId}, but no divergence detected.", peer.NodeId); } + + syncSuccessful = true; + RecordSuccess(peer.NodeId); + } + + catch (SnapshotRequiredException) + { + _logger.LogWarning("Snapshot required for peer {NodeId}. Initiating merge sync.", peer.NodeId); + if (client != null && client.IsConnected) + { + try + { + await PerformSnapshotSyncAsync(client, true, token); + syncSuccessful = true; + RecordSuccess(peer.NodeId); + } + catch + { + RecordFailure(peer.NodeId); + shouldRemoveClient = true; + } + } + else + { + RecordFailure(peer.NodeId); + shouldRemoveClient = true; + } + } + catch (CorruptDatabaseException cex) + { + _logger.LogCritical(cex, "Local database corruption detected during sync with {NodeId}. Initiating EMERGENCY SNAPSHOT RECOVERY.", peer.NodeId); + if (client != null && client.IsConnected) + { + try + { + // EMERGENCY RECOVERY: Replace local DB with remote snapshot (mergeOnly: false) + await PerformSnapshotSyncAsync(client, false, token); + syncSuccessful = true; + RecordSuccess(peer.NodeId); + _logger.LogInformation("Emergency recovery successful. Local database replaced."); + } + catch (Exception recoveryEx) + { + _logger.LogCritical(recoveryEx, "Emergency recovery failed. App state is critical."); + RecordFailure(peer.NodeId); + shouldRemoveClient = true; + } + } + else + { + RecordFailure(peer.NodeId); + shouldRemoveClient = true; + } } catch (TimeoutException tex) { _logger.LogWarning("Sync with {NodeId} timed out: {Message}. Will retry later.", peer.NodeId, tex.Message); shouldRemoveClient = true; + RecordFailure(peer.NodeId); } catch (SocketException sex) { _logger.LogWarning("Network error syncing with {NodeId}: {Message}. Will retry later.", peer.NodeId, sex.Message); shouldRemoveClient = true; + RecordFailure(peer.NodeId); } catch (Exception ex) { _logger.LogWarning("Sync failed with {NodeId}: {Message}. Resetting connection.", peer.NodeId, ex.Message); shouldRemoveClient = true; + RecordFailure(peer.NodeId); } finally { @@ -289,14 +391,53 @@ private async Task TrySyncWithPeer(PeerNode peer, CancellationToken token) try { removedClient.Dispose(); } catch { /* Ignore disposal errors */ } } } + + // Log successful sync outcome (failures are already logged in catch blocks) + if (syncSuccessful) + { + _logger.LogInformation("Sync with {NodeId} completed successfully.", peer.NodeId); + } } } + + private void RecordSuccess(string nodeId) + { + _peerStates.AddOrUpdate(nodeId, + new PeerStatus { FailureCount = 0, NextRetryTime = DateTime.MinValue }, + (k, v) => { v.FailureCount = 0; v.NextRetryTime = DateTime.MinValue; return v; }); + } + + private void RecordFailure(string nodeId) + { + _peerStates.AddOrUpdate(nodeId, + new PeerStatus { FailureCount = 1, NextRetryTime = DateTime.UtcNow.AddSeconds(1) }, + (k, v) => + { + v.FailureCount++; + // Exponential backoff: 1s, 2s, 4s... max 60s + var delaySeconds = Math.Min(Math.Pow(2, v.FailureCount), 60); + v.NextRetryTime = DateTime.UtcNow.AddSeconds(delaySeconds); + return v; + }); + } /// /// Validates an inbound batch of changes, checks for gaps, performs recovery if needed, and applies to store. /// Extracted to enforce Single Responsibility Principle. /// - private async Task ProcessInboundBatchAsync(TcpPeerClient client, string peerNodeId, IList changes, CancellationToken token) + private enum SyncBatchResult + { + Success, + GapDetected, + IntegrityError, + ChainBroken + } + + /// + /// Validates an inbound batch of changes, checks for gaps, performs recovery if needed, and applies to store. + /// Extracted to enforce Single Responsibility Principle. + /// + private async Task ProcessInboundBatchAsync(TcpPeerClient client, string peerNodeId, IList changes, CancellationToken token) { _logger.LogInformation("Received {Count} changes from {NodeId}", changes.Count, peerNodeId); @@ -328,7 +469,8 @@ private async Task ProcessInboundBatchAsync(TcpPeerClient client, string peerNod { if (authorChain[i].PreviousHash != authorChain[i - 1].Hash) { - throw new InvalidOperationException($"Chain Broken in Batch for Node {authorNodeId}"); + _logger.LogError("Chain Broken in Batch for Node {AuthorId}", authorNodeId); + return SyncBatchResult.ChainBroken; } } @@ -345,22 +487,37 @@ private async Task ProcessInboundBatchAsync(TcpPeerClient client, string peerNod _logger.LogWarning("Gap Detected for Node {AuthorId}. Local Head: {Local}, Remote Prev: {Prev}. Initiating Recovery.", authorNodeId, localHeadHash, firstEntry.PreviousHash); // Gap Recovery (Range Sync) - var missingChain = await client.GetChainRangeAsync(localHeadHash, firstEntry.PreviousHash, token); + List? missingChain = null; + try + { + missingChain = await client.GetChainRangeAsync(localHeadHash, firstEntry.PreviousHash, token); + } + catch (SnapshotRequiredException) + { + throw; // Propagate up to trigger full sync + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Gap Recovery failed."); + /* Fallthrough to decision logic */ + } if (missingChain != null && missingChain.Any()) { _logger.LogInformation("Gap Recovery: Retrieved {Count} missing entries.", missingChain.Count); // Validate Recovery Chain Linkage - if (missingChain[0].PreviousHash != localHeadHash) - throw new InvalidOperationException("Recovery Chain does not link to Local Head"); - + bool linkValid = true; + if (missingChain[0].PreviousHash != localHeadHash) linkValid = false; for (int i = 1; i < missingChain.Count; i++) - if (missingChain[i].PreviousHash != missingChain[i - 1].Hash) - throw new InvalidOperationException("Recovery Chain has internal breaks"); + if (missingChain[i].PreviousHash != missingChain[i - 1].Hash) linkValid = false; + if (missingChain.Last().Hash != firstEntry.PreviousHash) linkValid = false; - if (missingChain.Last().Hash != firstEntry.PreviousHash) - throw new InvalidOperationException("Recovery Chain does not link to Batch Start"); + if (!linkValid) + { + _logger.LogError("Recovery Chain Invalid Linkage. Aborting Gap Recovery."); + return SyncBatchResult.GapDetected; + } // Apply Missing Chain First await _store.ApplyBatchAsync(Enumerable.Empty(), missingChain, token); @@ -391,5 +548,57 @@ private async Task ProcessInboundBatchAsync(TcpPeerClient client, string peerNod // Apply original batch (grouped by node for clarity, but store usually handles bulk) await _store.ApplyBatchAsync(Enumerable.Empty(), authorChain, token); } + + return SyncBatchResult.Success; + } + + private async Task PerformSnapshotSyncAsync(TcpPeerClient client, bool mergeOnly, CancellationToken token) + { + _logger.LogInformation(mergeOnly ? "Starting Snapshot Merge..." : "Starting Full Database Replacement..."); + + var tempFile = Path.GetTempFileName(); + try + { + _logger.LogInformation("Downloading snapshot to {TempFile}...", tempFile); + using (var fs = File.Create(tempFile)) + { + await client.GetSnapshotAsync(fs, token); + } + + _logger.LogInformation("Snapshot Downloaded. applying to store..."); + + using (var fs = File.OpenRead(tempFile)) + { + if (mergeOnly) + { + await _store.MergeSnapshotAsync(fs, token); + } + else + { + await _store.ReplaceDatabaseAsync(fs, token); + } + } + + _logger.LogInformation("Snapshot applied successfully."); + } + catch (Exception ex) + { + _logger.LogError(ex, "Failed to perform snapshot sync"); + throw; + } + finally + { + if (File.Exists(tempFile)) + { + try + { + File.Delete(tempFile); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to delete temporary snapshot file {TempFile}", tempFile); + } + } + } } } \ No newline at end of file diff --git a/src/EntglDb.Network/TcpPeerClient.cs b/src/EntglDb.Network/TcpPeerClient.cs index 16e6037..6bdaac8 100644 --- a/src/EntglDb.Network/TcpPeerClient.cs +++ b/src/EntglDb.Network/TcpPeerClient.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.IO; using System.Linq; using System.Net.Sockets; using System.Threading; @@ -276,6 +277,8 @@ public async Task> GetChainRangeAsync(string startHash, string var res = ChainRangeResponse.Parser.ParseFrom(payload); + if (res.SnapshotRequired) throw new SnapshotRequiredException(); + return res.Entries.Select(e => new OplogEntry( e.Collection, e.Key, @@ -316,12 +319,35 @@ public async Task PushChangesAsync(IEnumerable entries, Cancellation var (type, payload) = await _protocol.ReadMessageAsync(_stream!, _cipherState, token); if (type != MessageType.AckRes) throw new Exception("Push failed"); + + var res = AckResponse.Parser.ParseFrom(payload); + if (res.SnapshotRequired) throw new SnapshotRequiredException(); + if (!res.Success) throw new Exception("Push failed"); } private bool _useCompression = false; // Negotiated after handshake private OperationType ParseOp(string op) => Enum.TryParse(op, out var val) ? val : OperationType.Put; + public async Task GetSnapshotAsync(Stream destination, CancellationToken token) + { + await _protocol.SendMessageAsync(_stream!, MessageType.GetSnapshotReq, new GetSnapshotRequest(), _useCompression, _cipherState, token); + + while (true) + { + var (type, payload) = await _protocol.ReadMessageAsync(_stream!, _cipherState, token); + if (type != MessageType.SnapshotChunkMsg) throw new Exception($"Unexpected message type during snapshot: {type}"); + + var chunk = SnapshotChunk.Parser.ParseFrom(payload); + if (chunk.Data.Length > 0) + { + await destination.WriteAsync(chunk.Data.ToByteArray(), 0, chunk.Data.Length, token); + } + + if (chunk.IsLast) break; + } + } + public void Dispose() { lock (_connectionLock) @@ -351,3 +377,8 @@ public void Dispose() _logger.LogDebug("Disposed connection to peer: {Address}", _peerAddress); } } + +public class SnapshotRequiredException : Exception +{ + public SnapshotRequiredException() : base("Peer requires a full snapshot sync.") { } +} diff --git a/src/EntglDb.Network/TcpSyncServer.cs b/src/EntglDb.Network/TcpSyncServer.cs index 1657f96..deee549 100644 --- a/src/EntglDb.Network/TcpSyncServer.cs +++ b/src/EntglDb.Network/TcpSyncServer.cs @@ -361,24 +361,68 @@ private async Task HandleClientAsync(TcpClient client, CancellationToken token) var rangeReq = GetChainRangeRequest.Parser.ParseFrom(payload); var rangeEntries = await _store.GetChainRangeAsync(rangeReq.StartHash, rangeReq.EndHash, token); var rangeRes = new ChainRangeResponse(); - foreach (var e in rangeEntries) + + if (!rangeEntries.Any() && rangeReq.StartHash != rangeReq.EndHash) { - rangeRes.Entries.Add(new ProtoOplogEntry + // Gap cannot be filled (likely pruned or unknown branch) + rangeRes.SnapshotRequired = true; + } + else + { + foreach (var e in rangeEntries) { - Collection = e.Collection, - Key = e.Key, - Operation = e.Operation.ToString(), - JsonData = e.Payload?.GetRawText() ?? "", - HlcWall = e.Timestamp.PhysicalTime, - HlcLogic = e.Timestamp.LogicalCounter, - HlcNode = e.Timestamp.NodeId, - Hash = e.Hash, - PreviousHash = e.PreviousHash - }); + rangeRes.Entries.Add(new ProtoOplogEntry + { + Collection = e.Collection, + Key = e.Key, + Operation = e.Operation.ToString(), + JsonData = e.Payload?.GetRawText() ?? "", + HlcWall = e.Timestamp.PhysicalTime, + HlcLogic = e.Timestamp.LogicalCounter, + HlcNode = e.Timestamp.NodeId, + Hash = e.Hash, + PreviousHash = e.PreviousHash + }); + } } response = rangeRes; resType = MessageType.ChainRangeRes; break; + + case MessageType.GetSnapshotReq: + _logger.LogInformation("Processing GetSnapshotReq from {Endpoint}", remoteEp); + var tempFile = Path.GetTempFileName(); + try + { + // Create backup + using (var fs = File.Create(tempFile)) + { + await _store.CreateSnapshotAsync(fs, token); + } + + using (var fs = File.OpenRead(tempFile)) + { + byte[] buffer = new byte[80 * 1024]; // 80KB chunks + int bytesRead; + while ((bytesRead = await fs.ReadAsync(buffer, 0, buffer.Length, token)) > 0) + { + var chunk = new SnapshotChunk + { + Data = ByteString.CopyFrom(buffer, 0, bytesRead), + IsLast = false + }; + await protocol.SendMessageAsync(stream, MessageType.SnapshotChunkMsg, chunk, false, cipherState, token); + } + + // Send End of Snapshot + await protocol.SendMessageAsync(stream, MessageType.SnapshotChunkMsg, new SnapshotChunk { IsLast = true }, false, cipherState, token); + } + } + finally + { + if (File.Exists(tempFile)) File.Delete(tempFile); + } + break; } if (response != null) diff --git a/src/EntglDb.Network/sync.proto b/src/EntglDb.Network/sync.proto index 770ec2c..a6ff023 100644 --- a/src/EntglDb.Network/sync.proto +++ b/src/EntglDb.Network/sync.proto @@ -16,26 +16,26 @@ message HandshakeResponse { string selected_compression = 3; // v4 } -message GetClockRequest { -} - -message ClockResponse { - int64 hlc_wall = 1; - int32 hlc_logic = 2; - string hlc_node = 3; -} - -message GetVectorClockRequest { -} - -message VectorClockResponse { - repeated VectorClockEntry entries = 1; -} - -message VectorClockEntry { - string node_id = 1; - int64 hlc_wall = 2; - int32 hlc_logic = 3; +message GetClockRequest { +} + +message ClockResponse { + int64 hlc_wall = 1; + int32 hlc_logic = 2; + string hlc_node = 3; +} + +message GetVectorClockRequest { +} + +message VectorClockResponse { + repeated VectorClockEntry entries = 1; +} + +message VectorClockEntry { + string node_id = 1; + int64 hlc_wall = 2; + int32 hlc_logic = 3; } message PullChangesRequest { @@ -59,10 +59,12 @@ message GetChainRangeRequest { message ChainRangeResponse { repeated ProtoOplogEntry entries = 1; + bool snapshot_required = 2; } message AckResponse { bool success = 1; + bool snapshot_required = 2; } message ProtoOplogEntry { @@ -77,6 +79,14 @@ message ProtoOplogEntry { string previous_hash = 9; } +message GetSnapshotRequest { +} + +message SnapshotChunk { + bytes data = 1; + bool is_last = 2; +} + // Enum for wire framing (1 byte) enum MessageType { Unknown = 0; @@ -93,6 +103,8 @@ enum MessageType { ChainRangeRes = 11; GetVectorClockReq = 12; VectorClockRes = 13; + GetSnapshotReq = 14; + SnapshotChunkMsg = 15; } message SecureEnvelope { diff --git a/src/EntglDb.Persistence.EntityFramework/EfCorePeerStore.cs b/src/EntglDb.Persistence.EntityFramework/EfCorePeerStore.cs index 463306d..87443e6 100644 --- a/src/EntglDb.Persistence.EntityFramework/EfCorePeerStore.cs +++ b/src/EntglDb.Persistence.EntityFramework/EfCorePeerStore.cs @@ -588,4 +588,353 @@ public async Task RemoveRemotePeerAsync(string nodeId, CancellationToken cancell _logger.LogWarning("Attempted to remove non-existent remote peer: {NodeId}", nodeId); } } + + public async Task PruneOplogAsync(HlcTimestamp cutoff, CancellationToken cancellationToken = default) + { + _logger.LogInformation("Pruning oplog entries older than {Cutoff}...", cutoff); + + using var transaction = await _context.Database.BeginTransactionAsync(cancellationToken); + try + { + // Find entries to delete + var entriesToDelete = await _context.Oplog + .Where(o => o.TimestampPhysicalTime < cutoff.PhysicalTime || + (o.TimestampPhysicalTime == cutoff.PhysicalTime && o.TimestampLogicalCounter < cutoff.LogicalCounter)) + .ToListAsync(cancellationToken); + + if (!entriesToDelete.Any()) + { + _logger.LogInformation("No oplog entries to prune."); + return; + } + + // Update SnapshotMetadata with boundary entries (latest before cutoff per node) + var boundaryEntries = entriesToDelete + .GroupBy(o => o.TimestampNodeId) + .Select(g => g.OrderByDescending(o => o.TimestampPhysicalTime) + .ThenByDescending(o => o.TimestampLogicalCounter) + .First()) + .ToList(); + + foreach (var entry in boundaryEntries) + { + var existingMeta = await _context.SnapshotMetadata + .FirstOrDefaultAsync(s => s.NodeId == entry.TimestampNodeId, cancellationToken); + + if (existingMeta == null) + { + _context.SnapshotMetadata.Add(new SnapshotMetadataEntity + { + NodeId = entry.TimestampNodeId, + TimestampPhysicalTime = entry.TimestampPhysicalTime, + TimestampLogicalCounter = entry.TimestampLogicalCounter, + Hash = entry.Hash ?? "" + }); + } + else + { + existingMeta.TimestampPhysicalTime = entry.TimestampPhysicalTime; + existingMeta.TimestampLogicalCounter = entry.TimestampLogicalCounter; + existingMeta.Hash = entry.Hash ?? ""; + } + } + + // Delete old entries + _context.Oplog.RemoveRange(entriesToDelete); + await _context.SaveChangesAsync(cancellationToken); + await transaction.CommitAsync(cancellationToken); + + _logger.LogInformation("Pruned {Count} oplog entries.", entriesToDelete.Count); + } + catch + { + await transaction.RollbackAsync(cancellationToken); + throw; + } + } + + public async Task CreateSnapshotAsync(Stream destination, CancellationToken cancellationToken = default) + { + _logger.LogInformation("Creating EF Core snapshot (JSON format)..."); + + // Load all entities + var documents = await _context.Documents.ToListAsync(cancellationToken); + var oplog = await _context.Oplog.ToListAsync(cancellationToken); + var snapshotMetadata = await _context.SnapshotMetadata.ToListAsync(cancellationToken); + var remotePeers = await _context.RemotePeers.ToListAsync(cancellationToken); + + // Build snapshot DTO + var snapshot = new Snapshot.SnapshotDto + { + Version = "1.0", + CreatedAt = DateTime.UtcNow.ToString("O"), + NodeId = "", // Will be set by caller if needed + Documents = documents.Select(d => new Snapshot.DocumentDto + { + Collection = d.Collection, + Key = d.Key, + JsonData = d.ContentJson, + IsDeleted = d.IsDeleted, + HlcWall = d.UpdatedAtPhysicalTime, + HlcLogic = d.UpdatedAtLogicalCounter, + HlcNode = d.UpdatedAtNodeId + }).ToList(), + Oplog = oplog.Select(o => new Snapshot.OplogDto + { + Collection = o.Collection, + Key = o.Key, + Operation = o.Operation, + JsonData = o.PayloadJson, + HlcWall = o.TimestampPhysicalTime, + HlcLogic = o.TimestampLogicalCounter, + HlcNode = o.TimestampNodeId, + Hash = o.Hash ?? "", + PreviousHash = o.PreviousHash + }).ToList(), + SnapshotMetadata = snapshotMetadata.Select(s => new Snapshot.SnapshotMetadataDto + { + NodeId = s.NodeId, + HlcWall = s.TimestampPhysicalTime, + HlcLogic = s.TimestampLogicalCounter, + Hash = s.Hash ?? "" + }).ToList(), + RemotePeers = remotePeers.Select(p => new Snapshot.RemotePeerDto + { + NodeId = p.NodeId, + Address = p.Address, + Type = p.Type, + OAuth2Json = p.OAuth2Json, + IsEnabled = p.IsEnabled + }).ToList() + }; + + // Serialize to JSON + await JsonSerializer.SerializeAsync(destination, snapshot, cancellationToken: cancellationToken); + _logger.LogInformation("Snapshot created: {DocCount} documents, {OplogCount} oplog entries", documents.Count, oplog.Count); + } + + public async Task ReplaceDatabaseAsync(Stream databaseStream, CancellationToken cancellationToken = default) + { + _logger.LogWarning("Replacing EF Core database from snapshot stream..."); + + await ClearAllDataAsync(cancellationToken); + + var snapshot = await JsonSerializer.DeserializeAsync(databaseStream, cancellationToken: cancellationToken); + if (snapshot == null) throw new InvalidOperationException("Failed to deserialize snapshot"); + + await BulkInsertSnapshotAsync(snapshot, cancellationToken); + + _logger.LogInformation("Database replaced successfully."); + } + + public async Task MergeSnapshotAsync(Stream snapshotStream, CancellationToken cancellationToken = default) + { + _logger.LogInformation("Merging remote snapshot into local database..."); + + var snapshot = await JsonSerializer.DeserializeAsync(snapshotStream, cancellationToken: cancellationToken); + if (snapshot == null) throw new InvalidOperationException("Failed to deserialize snapshot"); + + using var transaction = await _context.Database.BeginTransactionAsync(cancellationToken); + try + { + var existingHashes = (await _context.Oplog.Select(o => o.Hash).ToListAsync(cancellationToken)).ToHashSet(); + var newOplogEntries = snapshot.Oplog + .Where(o => !string.IsNullOrEmpty(o.Hash) && !existingHashes.Contains(o.Hash)) + .Select(o => new OplogEntity + { + Collection = o.Collection, + Key = o.Key, + Operation = o.Operation, + PayloadJson = o.JsonData, + TimestampPhysicalTime = o.HlcWall, + TimestampLogicalCounter = o.HlcLogic, + TimestampNodeId = o.HlcNode, + Hash = o.Hash, + PreviousHash = o.PreviousHash + }).ToList(); + + if (newOplogEntries.Any()) + { + await _context.Oplog.AddRangeAsync(newOplogEntries, cancellationToken); + } + + foreach (var remoteMeta in snapshot.SnapshotMetadata) + { + var localMeta = await _context.SnapshotMetadata + .FirstOrDefaultAsync(s => s.NodeId == remoteMeta.NodeId, cancellationToken); + + if (localMeta == null) + { + _context.SnapshotMetadata.Add(new SnapshotMetadataEntity + { + NodeId = remoteMeta.NodeId, + TimestampPhysicalTime = remoteMeta.HlcWall, + TimestampLogicalCounter = remoteMeta.HlcLogic, + Hash = remoteMeta.Hash + }); + } + else + { + if (remoteMeta.HlcWall > localMeta.TimestampPhysicalTime || + (remoteMeta.HlcWall == localMeta.TimestampPhysicalTime && remoteMeta.HlcLogic > localMeta.TimestampLogicalCounter)) + { + localMeta.TimestampPhysicalTime = remoteMeta.HlcWall; + localMeta.TimestampLogicalCounter = remoteMeta.HlcLogic; + localMeta.Hash = remoteMeta.Hash; + } + } + } + + foreach (var remoteDoc in snapshot.Documents) + { + var localDoc = await _context.Documents + .FirstOrDefaultAsync(d => d.Collection == remoteDoc.Collection && d.Key == remoteDoc.Key, cancellationToken); + + if (localDoc == null) + { + _context.Documents.Add(new DocumentEntity + { + Collection = remoteDoc.Collection, + Key = remoteDoc.Key, + ContentJson = remoteDoc.JsonData ?? "{}", + IsDeleted = remoteDoc.IsDeleted, + UpdatedAtPhysicalTime = remoteDoc.HlcWall, + UpdatedAtLogicalCounter = remoteDoc.HlcLogic, + UpdatedAtNodeId = remoteDoc.HlcNode + }); + } + else + { + if (remoteDoc.HlcWall > localDoc.UpdatedAtPhysicalTime || + (remoteDoc.HlcWall == localDoc.UpdatedAtPhysicalTime && remoteDoc.HlcLogic > localDoc.UpdatedAtLogicalCounter)) + { + localDoc.ContentJson = remoteDoc.JsonData ?? "{}"; + localDoc.IsDeleted = remoteDoc.IsDeleted; + localDoc.UpdatedAtPhysicalTime = remoteDoc.HlcWall; + localDoc.UpdatedAtLogicalCounter = remoteDoc.HlcLogic; + localDoc.UpdatedAtNodeId = remoteDoc.HlcNode; + } + } + } + + foreach (var remotePeer in snapshot.RemotePeers) + { + var localPeer = await _context.RemotePeers + .FirstOrDefaultAsync(p => p.NodeId == remotePeer.NodeId, cancellationToken); + + if (localPeer == null) + { + _context.RemotePeers.Add(new RemotePeerEntity + { + NodeId = remotePeer.NodeId, + Address = remotePeer.Address, + Type = remotePeer.Type, + OAuth2Json = remotePeer.OAuth2Json, + IsEnabled = remotePeer.IsEnabled + }); + } + else + { + localPeer.Address = remotePeer.Address; + localPeer.Type = remotePeer.Type; + localPeer.OAuth2Json = remotePeer.OAuth2Json; + localPeer.IsEnabled = remotePeer.IsEnabled; + } + } + + await _context.SaveChangesAsync(cancellationToken); + await transaction.CommitAsync(cancellationToken); + + lock (_cacheLock) + { + _cacheInitialized = false; + } + + _logger.LogInformation("Database merge completed successfully."); + } + catch + { + await transaction.RollbackAsync(cancellationToken); + throw; + } + } + + public async Task ClearAllDataAsync(CancellationToken cancellationToken = default) + { + _logger.LogWarning("CLEARING ALL DATA FROM STORE!"); + + _context.Documents.RemoveRange(_context.Documents); + _context.Oplog.RemoveRange(_context.Oplog); + _context.SnapshotMetadata.RemoveRange(_context.SnapshotMetadata); + _context.RemotePeers.RemoveRange(_context.RemotePeers); + + await _context.SaveChangesAsync(cancellationToken); + + lock (_cacheLock) + { + _cacheInitialized = false; + _nodeCache.Clear(); + } + + _logger.LogInformation("Store cleared successfully."); + } + + private async Task BulkInsertSnapshotAsync(Snapshot.SnapshotDto snapshot, CancellationToken cancellationToken) + { + var documents = snapshot.Documents.Select(d => new DocumentEntity + { + Collection = d.Collection, + Key = d.Key, + ContentJson = d.JsonData ?? "{}", + IsDeleted = d.IsDeleted, + UpdatedAtPhysicalTime = d.HlcWall, + UpdatedAtLogicalCounter = d.HlcLogic, + UpdatedAtNodeId = d.HlcNode + }).ToList(); + + if (documents.Any()) await _context.Documents.AddRangeAsync(documents, cancellationToken); + + var oplogEntries = snapshot.Oplog.Select(o => new OplogEntity + { + Collection = o.Collection, + Key = o.Key, + Operation = o.Operation, + PayloadJson = o.JsonData, + TimestampPhysicalTime = o.HlcWall, + TimestampLogicalCounter = o.HlcLogic, + TimestampNodeId = o.HlcNode, + Hash = o.Hash, + PreviousHash = o.PreviousHash + }).ToList(); + + if (oplogEntries.Any()) await _context.Oplog.AddRangeAsync(oplogEntries, cancellationToken); + + var metadata = snapshot.SnapshotMetadata.Select(s => new SnapshotMetadataEntity + { + NodeId = s.NodeId, + TimestampPhysicalTime = s.HlcWall, + TimestampLogicalCounter = s.HlcLogic, + Hash = s.Hash + }).ToList(); + + if (metadata.Any()) await _context.SnapshotMetadata.AddRangeAsync(metadata, cancellationToken); + + var peers = snapshot.RemotePeers.Select(p => new RemotePeerEntity + { + NodeId = p.NodeId, + Address = p.Address, + Type = p.Type, + OAuth2Json = p.OAuth2Json, + IsEnabled = p.IsEnabled + }).ToList(); + + if (peers.Any()) await _context.RemotePeers.AddRangeAsync(peers, cancellationToken); + + await _context.SaveChangesAsync(cancellationToken); + + lock (_cacheLock) + { + _cacheInitialized = false; + } + } } diff --git a/src/EntglDb.Persistence.EntityFramework/EntglDbContext.cs b/src/EntglDb.Persistence.EntityFramework/EntglDbContext.cs index df88ac5..4109b09 100644 --- a/src/EntglDb.Persistence.EntityFramework/EntglDbContext.cs +++ b/src/EntglDb.Persistence.EntityFramework/EntglDbContext.cs @@ -24,6 +24,11 @@ public class EntglDbContext : DbContext /// public DbSet RemotePeers { get; set; } = null!; + /// + /// Gets or sets the SnapshotMetadata DbSet. + /// + public DbSet SnapshotMetadata { get; set; } = null!; + /// /// Initializes a new instance of the class. /// @@ -64,5 +69,12 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) entity.HasKey(e => e.NodeId); entity.HasIndex(e => e.IsEnabled); }); + + // Configure SnapshotMetadataEntity + modelBuilder.Entity(entity => + { + entity.HasKey(e => e.NodeId); + entity.HasIndex(e => new { e.TimestampPhysicalTime, e.TimestampLogicalCounter }); + }); } } diff --git a/src/EntglDb.Persistence.EntityFramework/Entities/SnapshotMetadataEntity.cs b/src/EntglDb.Persistence.EntityFramework/Entities/SnapshotMetadataEntity.cs new file mode 100644 index 0000000..8523427 --- /dev/null +++ b/src/EntglDb.Persistence.EntityFramework/Entities/SnapshotMetadataEntity.cs @@ -0,0 +1,21 @@ +using System.ComponentModel.DataAnnotations; +using System.ComponentModel.DataAnnotations.Schema; + +namespace EntglDb.Persistence.EntityFramework.Entities; + +/// +/// Entity representing snapshot metadata (oplog pruning checkpoint). +/// +[Table("SnapshotMetadata")] +public class SnapshotMetadataEntity +{ + [Key] + [MaxLength(256)] + public string NodeId { get; set; } = ""; + + public long TimestampPhysicalTime { get; set; } + public int TimestampLogicalCounter { get; set; } + + [MaxLength(128)] + public string Hash { get; set; } = ""; +} diff --git a/src/EntglDb.Persistence.EntityFramework/Snapshot/SnapshotDto.cs b/src/EntglDb.Persistence.EntityFramework/Snapshot/SnapshotDto.cs new file mode 100644 index 0000000..51db3d8 --- /dev/null +++ b/src/EntglDb.Persistence.EntityFramework/Snapshot/SnapshotDto.cs @@ -0,0 +1,58 @@ +using System.Collections.Generic; + +namespace EntglDb.Persistence.EntityFramework.Snapshot; + +/// +/// Root DTO for EF Core snapshots (JSON format). +/// +public class SnapshotDto +{ + public string Version { get; set; } = "1.0"; + public string CreatedAt { get; set; } = ""; + public string NodeId { get; set; } = ""; + public List Documents { get; set; } = new(); + public List Oplog { get; set; } = new(); + public List SnapshotMetadata { get; set; } = new(); + public List RemotePeers { get; set; } = new(); +} + +public class DocumentDto +{ + public string Collection { get; set; } = ""; + public string Key { get; set; } = ""; + public string? JsonData { get; set; } + public bool IsDeleted { get; set; } + public long HlcWall { get; set; } + public int HlcLogic { get; set; } + public string HlcNode { get; set; } = ""; +} + +public class OplogDto +{ + public string? Collection { get; set; } + public string? Key { get; set; } + public int Operation { get; set; } + public string? JsonData { get; set; } + public long HlcWall { get; set; } + public int HlcLogic { get; set; } + public string HlcNode { get; set; } = ""; + public string Hash { get; set; } = ""; + public string? PreviousHash { get; set; } +} + +public class SnapshotMetadataDto +{ + public string NodeId { get; set; } = ""; + public long HlcWall { get; set; } + public int HlcLogic { get; set; } + public string Hash { get; set; } = ""; +} + +public class RemotePeerDto +{ + public string NodeId { get; set; } = ""; + public string Address { get; set; } = ""; + public int Type { get; set; } + public string? OAuth2Json { get; set; } + public bool IsEnabled { get; set; } +} diff --git a/src/EntglDb.Persistence.Sqlite/SqlitePeerStore.cs b/src/EntglDb.Persistence.Sqlite/SqlitePeerStore.cs index d16f2f2..ad354ab 100644 --- a/src/EntglDb.Persistence.Sqlite/SqlitePeerStore.cs +++ b/src/EntglDb.Persistence.Sqlite/SqlitePeerStore.cs @@ -31,10 +31,10 @@ public class SqlitePeerStore : IPeerStore private readonly HashSet _createdTables = new HashSet(); private readonly object _tableLock = new object(); private readonly object _cacheLock = new object(); - + // Per-node cache: tracks latest timestamp and hash for each node private readonly Dictionary _nodeCache = new Dictionary(StringComparer.Ordinal); - + private class NodeCacheEntry { public HlcTimestamp Timestamp { get; set; } @@ -59,9 +59,9 @@ public SqlitePeerStore(string connectionString, ILogger? logger /// New constructor with dynamic database path and per-collection table support. /// public SqlitePeerStore( - IPeerNodeConfigurationProvider configProvider, + IPeerNodeConfigurationProvider configProvider, SqlitePersistenceOptions options, - ILogger? logger = null, + ILogger? logger = null, IConflictResolver? conflictResolver = null) { _options = options ?? throw new ArgumentNullException(nameof(options)); @@ -75,17 +75,17 @@ private async Task BuildConnectionString(IPeerNodeConfigurationProvider { var config = await configProvider.GetConfiguration(); var basePath = options.BasePath ?? SqlitePersistenceOptions.DefaultBasePath; - + var filename = options.DatabaseFilenameTemplate.Replace("{NodeId}", config.NodeId); var dbPath = Path.Combine(basePath, filename); - + var directory = Path.GetDirectoryName(dbPath); if (!string.IsNullOrEmpty(directory) && !Directory.Exists(directory)) { Directory.CreateDirectory(directory); _logger.LogInformation("Created database directory: {Directory}", directory); } - + _logger.LogInformation("Database path: {DbPath}", dbPath); return $"Data Source={dbPath}"; } @@ -130,6 +130,13 @@ IsEnabled INTEGER NOT NULL CREATE INDEX IF NOT EXISTS IDX_Oplog_HlcWall ON Oplog(HlcWall); CREATE INDEX IF NOT EXISTS IDX_Oplog_Hash ON Oplog(Hash); + + CREATE TABLE IF NOT EXISTS SnapshotMetadata ( + NodeId TEXT PRIMARY KEY, + HlcWall INTEGER NOT NULL, + HlcLogic INTEGER NOT NULL, + Hash TEXT + ); "); // Documents Table (Legacy Mode Only) @@ -153,7 +160,7 @@ PRIMARY KEY (Collection, Key) { _logger.LogInformation("Initialized with per-collection table mode"); } - + // Initialize node cache from database InitializeNodeCache(connection); _logger.LogInformation("Node cache initialized with {Count} nodes", _nodeCache.Count); @@ -177,38 +184,56 @@ private void EnsureWalMode(SqliteConnection connection) private void InitializeNodeCache(SqliteConnection connection) { - // Query latest entry per node - var rows = connection.Query<(string NodeId, long HlcWall, int HlcLogic, string Hash)>(@" - SELECT HlcNode as NodeId, HlcWall, HlcLogic, Hash - FROM Oplog o1 - WHERE (HlcWall, HlcLogic) = ( - SELECT MAX(HlcWall), MAX(HlcLogic) - FROM Oplog o2 - WHERE o2.HlcNode = o1.HlcNode - ) - GROUP BY HlcNode"); - lock (_cacheLock) { _nodeCache.Clear(); - foreach (var row in rows) + + // 1. Load from SnapshotMetadata (Base State) + var snapshots = connection.Query<(string NodeId, long HlcWall, int HlcLogic, string Hash)>("SELECT NodeId, HlcWall, HlcLogic, Hash FROM SnapshotMetadata"); + foreach (var s in snapshots) { - _nodeCache[row.NodeId] = new NodeCacheEntry + _nodeCache[s.NodeId] = new NodeCacheEntry { - Timestamp = new HlcTimestamp(row.HlcWall, row.HlcLogic, row.NodeId), - Hash = row.Hash ?? "" + Timestamp = new HlcTimestamp(s.HlcWall, s.HlcLogic, s.NodeId), + Hash = s.Hash ?? "" }; } + + // 2. Load from Oplog (Latest State - Overrides Snapshot if newer) + var rows = connection.Query<(string NodeId, long HlcWall, int HlcLogic, string Hash)>(@" + SELECT HlcNode as NodeId, HlcWall, HlcLogic, Hash + FROM Oplog o1 + WHERE (HlcWall, HlcLogic) = ( + SELECT MAX(HlcWall), MAX(HlcLogic) + FROM Oplog o2 + WHERE o2.HlcNode = o1.HlcNode + ) + GROUP BY HlcNode"); + + foreach (var row in rows) + { + var timestamp = new HlcTimestamp(row.HlcWall, row.HlcLogic, row.NodeId); + + // Only update if newer (though Oplog usually contains newer data than snapshot) + if (!_nodeCache.TryGetValue(row.NodeId, out var existing) || timestamp.CompareTo(existing.Timestamp) > 0) + { + _nodeCache[row.NodeId] = new NodeCacheEntry + { + Timestamp = timestamp, + Hash = row.Hash ?? "" + }; + } + } } } private string GetDocumentTableName(string collection) => - _options?.UsePerCollectionTables == true + _options?.UsePerCollectionTables == true ? $"Documents_{SanitizeCollectionName(collection)}" : "Documents"; private string GetOplogTableName(string collection) => - _options?.UsePerCollectionTables == true + _options?.UsePerCollectionTables == true ? $"Oplog_{SanitizeCollectionName(collection)}" : "Oplog"; @@ -218,16 +243,16 @@ private string SanitizeCollectionName(string collection) => private async Task EnsureCollectionTablesAsync(SqliteConnection connection, string collection) { if (_options?.UsePerCollectionTables != true) return; - + // Check if already created (thread-safe) lock (_tableLock) { if (_createdTables.Contains(collection)) return; } - + var docTable = GetDocumentTableName(collection); var oplogTable = GetOplogTableName(collection); - + await connection.ExecuteAsync($@" CREATE TABLE IF NOT EXISTS {docTable} ( Key TEXT PRIMARY KEY, @@ -250,12 +275,12 @@ HlcNode TEXT NOT NULL CREATE INDEX IF NOT EXISTS IDX_{oplogTable}_HlcWall ON {oplogTable}(HlcWall); "); - + lock (_tableLock) { _createdTables.Add(collection); } - + _logger.LogDebug("Created tables for collection: {Collection}", collection); } @@ -268,16 +293,16 @@ public async Task CheckIntegrityAsync(CancellationToken cancellationToken { using var connection = new SqliteConnection(_connectionString); await connection.OpenAsync(cancellationToken); - + var result = await connection.QuerySingleAsync("PRAGMA integrity_check"); var isHealthy = result == "ok"; - + if (!isHealthy) { _logger.LogError("Database corruption detected: {Result}", result); throw new DatabaseCorruptionException($"Database integrity check failed: {result}"); } - + _logger.LogDebug("Database integrity check passed"); return true; } @@ -296,7 +321,7 @@ public async Task BackupAsync(string backupPath, CancellationToken cancellationT try { _logger.LogInformation("Creating database backup at {Path}", backupPath); - + var directory = Path.GetDirectoryName(backupPath); if (!string.IsNullOrEmpty(directory) && !Directory.Exists(directory)) { @@ -305,12 +330,12 @@ public async Task BackupAsync(string backupPath, CancellationToken cancellationT using var source = new SqliteConnection(_connectionString); using var destination = new SqliteConnection($"Data Source={backupPath}"); - + await source.OpenAsync(cancellationToken); await destination.OpenAsync(cancellationToken); - + source.BackupDatabase(destination); - + _logger.LogInformation("Database backup completed successfully"); } catch (Exception ex) @@ -331,7 +356,7 @@ public async Task SaveDocumentAsync(Document document, CancellationToken cancell { using var connection = new SqliteConnection(_connectionString); await connection.OpenAsync(cancellationToken); - + await EnsureCollectionTablesAsync(connection, document.Collection); var tableName = GetDocumentTableName(document.Collection); @@ -375,7 +400,7 @@ INSERT OR REPLACE INTO Documents (Collection, Key, JsonData, IsDeleted, HlcWall, VALUES (@Collection, @Key, @JsonData, @IsDeleted, @HlcWall, @HlcLogic, @HlcNode)", new { - document.Collection, + document.Collection, document.Key, JsonData = document.Content.ValueKind == JsonValueKind.Undefined ? null : document.Content.GetRawText(), document.IsDeleted, @@ -407,7 +432,7 @@ INSERT OR REPLACE INTO Documents (Collection, Key, JsonData, IsDeleted, HlcWall, { using var connection = new SqliteConnection(_connectionString); await connection.OpenAsync(cancellationToken); - + await EnsureCollectionTablesAsync(connection, collection); var tableName = GetDocumentTableName(collection); @@ -436,18 +461,18 @@ FROM Documents if (row == null) return null; var hlc = new HlcTimestamp(row.HlcWall, row.HlcLogic, row.HlcNode); - var content = row.JsonData != null - ? JsonSerializer.Deserialize(row.JsonData) + var content = row.JsonData != null + ? JsonSerializer.Deserialize(row.JsonData) : default; - return new Document(collection, row.Key, content, hlc, row.IsDeleted); + return new Document(collection, row.Key, content, hlc, row.IsDeleted != 0); } public async Task AppendOplogEntryAsync(OplogEntry entry, CancellationToken cancellationToken = default) { using var connection = new SqliteConnection(_connectionString); await connection.OpenAsync(cancellationToken); - + // Unified Oplog Table: Always use single Oplog table regardless of per-collection setting await connection.ExecuteAsync(@" INSERT INTO Oplog (Collection, Key, Operation, JsonData, HlcWall, HlcLogic, HlcNode, Hash, PreviousHash) @@ -464,7 +489,7 @@ INSERT INTO Oplog (Collection, Key, Operation, JsonData, HlcWall, HlcLogic, HlcN entry.Hash, entry.PreviousHash }); - + // Update node cache with both timestamp and hash lock (_cacheLock) { @@ -522,10 +547,10 @@ FROM Oplog } } - // Cache miss - query database + // Cache miss - query database (Oplog first, then SnapshotMetadata) using var connection = new SqliteConnection(_connectionString); await connection.OpenAsync(cancellationToken); - + var hash = await connection.QuerySingleOrDefaultAsync(@" SELECT Hash FROM Oplog @@ -533,21 +558,43 @@ FROM Oplog ORDER BY HlcWall DESC, HlcLogic DESC LIMIT 1", new { NodeId = nodeId }); + if (hash == null) + { + // Fallback to snapshot + hash = await connection.QuerySingleOrDefaultAsync(@" + SELECT Hash + FROM SnapshotMetadata + WHERE NodeId = @NodeId", new { NodeId = nodeId }); + } + // Update cache if found if (hash != null) { - var row = await connection.QuerySingleOrDefaultAsync<(long Wall, int Logic)>(@" + // Try to get timestamp from Oplog + var row = await connection.QuerySingleOrDefaultAsync<(long Wall, int Logic)?>(@" SELECT HlcWall as Wall, HlcLogic as Logic FROM Oplog WHERE Hash = @Hash", new { Hash = hash }); - lock (_cacheLock) + if (row == null) { - _nodeCache[nodeId] = new NodeCacheEntry + // Try to get timestamp from Snapshot + row = await connection.QuerySingleOrDefaultAsync<(long Wall, int Logic)?>(@" + SELECT HlcWall as Wall, HlcLogic as Logic + FROM SnapshotMetadata + WHERE Hash = @Hash", new { Hash = hash }); + } + + if (row.HasValue) + { + lock (_cacheLock) { - Timestamp = new HlcTimestamp(row.Wall, row.Logic, nodeId), - Hash = hash - }; + _nodeCache[nodeId] = new NodeCacheEntry + { + Timestamp = new HlcTimestamp(row.Value.Wall, row.Value.Logic, nodeId), + Hash = hash + }; + } } } @@ -558,14 +605,14 @@ FROM Oplog { using var connection = new SqliteConnection(_connectionString); await connection.OpenAsync(cancellationToken); - + var row = await connection.QuerySingleOrDefaultAsync(@" SELECT Collection, Key, Operation, JsonData, HlcWall, HlcLogic, HlcNode, PreviousHash, Hash FROM Oplog WHERE Hash = @Hash", new { Hash = hash }); - + if (row == null) return null; - + return new OplogEntry( row.Collection ?? "unknown", row.Key ?? "unknown", @@ -580,16 +627,16 @@ public async Task> GetChainRangeAsync(string startHash, { using var connection = new SqliteConnection(_connectionString); await connection.OpenAsync(cancellationToken); - + // 1. Fetch range bounds var startRow = await connection.QuerySingleOrDefaultAsync( "SELECT HlcWall, HlcLogic, HlcNode FROM Oplog WHERE Hash = @Hash", new { Hash = startHash }); var endRow = await connection.QuerySingleOrDefaultAsync( "SELECT HlcWall, HlcLogic, HlcNode FROM Oplog WHERE Hash = @Hash", new { Hash = endHash }); - + if (startRow == null || endRow == null) return Enumerable.Empty(); if (startRow.HlcNode != endRow.HlcNode) return Enumerable.Empty(); // Must be same chain - + // 2. Fetch range (Start < Entry <= End) var rows = await connection.QueryAsync(@" SELECT Collection, Key, Operation, JsonData, HlcWall, HlcLogic, HlcNode, PreviousHash, Hash @@ -598,12 +645,15 @@ FROM Oplog AND ( (HlcWall > @StartWall) OR (HlcWall = @StartWall AND HlcLogic > @StartLogic) ) AND ( (HlcWall < @EndWall) OR (HlcWall = @EndWall AND HlcLogic <= @EndLogic) ) ORDER BY HlcWall ASC, HlcLogic ASC", - new { + new + { NodeId = startRow.HlcNode, - StartWall = startRow.HlcWall, StartLogic = startRow.HlcLogic, - EndWall = endRow.HlcWall, EndLogic = endRow.HlcLogic + StartWall = startRow.HlcWall, + StartLogic = startRow.HlcLogic, + EndWall = endRow.HlcWall, + EndLogic = endRow.HlcLogic }); - + return rows.Select(r => new OplogEntry( r.Collection ?? "unknown", r.Key ?? "unknown", @@ -623,7 +673,7 @@ private List GetKnownCollections(SqliteConnection connection) { var tables = connection.Query( "SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'Oplog_%'"); - + return tables.Select(t => t.Substring(6)).ToList(); // Remove "Oplog_" prefix } @@ -713,16 +763,16 @@ public async Task ApplyBatchAsync(IEnumerable documents, IEnumerable($@" @@ -734,10 +784,10 @@ public async Task ApplyBatchAsync(IEnumerable documents, IEnumerable(local.JsonData) + var content = local.JsonData != null + ? JsonSerializer.Deserialize(local.JsonData) : default; - localDoc = new Document(entry.Collection, local.Key, content, localHlc, local.IsDeleted); + localDoc = new Document(entry.Collection, local.Key, content, localHlc, local.IsDeleted != 0); } } else @@ -751,10 +801,10 @@ FROM Documents if (local != null) { var localHlc = new HlcTimestamp(local.HlcWall, local.HlcLogic, local.HlcNode); - var content = local.JsonData != null - ? JsonSerializer.Deserialize(local.JsonData) + var content = local.JsonData != null + ? JsonSerializer.Deserialize(local.JsonData) : default; - localDoc = new Document(entry.Collection, local.Key, content, localHlc, local.IsDeleted); + localDoc = new Document(entry.Collection, local.Key, content, localHlc, local.IsDeleted != 0); } } @@ -762,41 +812,41 @@ FROM Documents if (resolution.ShouldApply && resolution.MergedDocument != null) { - var doc = resolution.MergedDocument; - - if (usePerCollection) - { - await connection.ExecuteAsync($@" + var doc = resolution.MergedDocument; + + if (usePerCollection) + { + await connection.ExecuteAsync($@" INSERT OR REPLACE INTO {docTableName} (Key, JsonData, IsDeleted, HlcWall, HlcLogic, HlcNode) VALUES (@Key, @JsonData, @IsDeleted, @HlcWall, @HlcLogic, @HlcNode)", - new - { - doc.Key, - JsonData = doc.Content.ValueKind == JsonValueKind.Undefined ? null : doc.Content.GetRawText(), - IsDeleted = doc.IsDeleted ? 1 : 0, - HlcWall = doc.UpdatedAt.PhysicalTime, - HlcLogic = doc.UpdatedAt.LogicalCounter, - HlcNode = doc.UpdatedAt.NodeId - }, transaction); - } - else - { - await connection.ExecuteAsync(@" + new + { + doc.Key, + JsonData = doc.Content.ValueKind == JsonValueKind.Undefined ? null : doc.Content.GetRawText(), + IsDeleted = doc.IsDeleted ? 1 : 0, + HlcWall = doc.UpdatedAt.PhysicalTime, + HlcLogic = doc.UpdatedAt.LogicalCounter, + HlcNode = doc.UpdatedAt.NodeId + }, transaction); + } + else + { + await connection.ExecuteAsync(@" INSERT OR REPLACE INTO Documents (Collection, Key, JsonData, IsDeleted, HlcWall, HlcLogic, HlcNode) VALUES (@Collection, @Key, @JsonData, @IsDeleted, @HlcWall, @HlcLogic, @HlcNode)", - new - { - doc.Collection, - doc.Key, - JsonData = doc.Content.ValueKind == JsonValueKind.Undefined ? null : doc.Content.GetRawText(), - IsDeleted = doc.IsDeleted ? 1 : 0, - HlcWall = doc.UpdatedAt.PhysicalTime, - HlcLogic = doc.UpdatedAt.LogicalCounter, - HlcNode = doc.UpdatedAt.NodeId - }, transaction); - } + new + { + doc.Collection, + doc.Key, + JsonData = doc.Content.ValueKind == JsonValueKind.Undefined ? null : doc.Content.GetRawText(), + IsDeleted = doc.IsDeleted ? 1 : 0, + HlcWall = doc.UpdatedAt.PhysicalTime, + HlcLogic = doc.UpdatedAt.LogicalCounter, + HlcNode = doc.UpdatedAt.NodeId + }, transaction); + } } - + // Unified Oplog Table: Always use single Oplog table regardless of per-collection setting await connection.ExecuteAsync(@" INSERT INTO Oplog (Collection, Key, Operation, JsonData, HlcWall, HlcLogic, HlcNode, Hash, PreviousHash) @@ -806,7 +856,7 @@ INSERT INTO Oplog (Collection, Key, Operation, JsonData, HlcWall, HlcLogic, HlcN entry.Collection, entry.Key, Operation = (int)entry.Operation, - JsonData = entry.Payload != null && entry.Payload!.Value.ValueKind != JsonValueKind.Undefined ? entry.Payload.Value.GetRawText() : null, + JsonData = entry.Payload.HasValue && entry.Payload.Value.ValueKind != JsonValueKind.Undefined ? entry.Payload.Value.GetRawText() : null, HlcWall = entry.Timestamp.PhysicalTime, HlcLogic = entry.Timestamp.LogicalCounter, HlcNode = entry.Timestamp.NodeId, @@ -814,40 +864,141 @@ INSERT INTO Oplog (Collection, Key, Operation, JsonData, HlcWall, HlcLogic, HlcN entry.PreviousHash }, transaction); } - - transaction.Commit(); - - try - { - ChangesApplied?.Invoke(this, new ChangesAppliedEventArgs(oplogEntries)); - // Update node cache for all entries - lock (_cacheLock) - { - foreach (var entry in oplogEntries) - { - var nodeId = entry.Timestamp.NodeId; - if (!_nodeCache.TryGetValue(nodeId, out var existing) || entry.Timestamp.CompareTo(existing.Timestamp) > 0) - { - _nodeCache[nodeId] = new NodeCacheEntry - { - Timestamp = entry.Timestamp, - Hash = entry.Hash ?? "" - }; - } - } - } - } - catch (Exception ex) - { - _logger.LogError(ex, "Error handling ChangesApplied event or updating cache"); - } + transaction.Commit(); + } + catch (SqliteException ex) when (ex.SqliteErrorCode == 11 || ex.SqliteErrorCode == 26) // SQLITE_CORRUPT or SQLITE_NOTADB + { + _logger.LogCritical(ex, "Database corruption detected during ApplyBatchAsync!"); + try { transaction.Rollback(); } catch { } + throw new CorruptDatabaseException("SQLite database is corrupt", ex); } - catch + catch (Exception ex) { - transaction.Rollback(); + _logger.LogError(ex, "Failed to apply batch"); + try { transaction.Rollback(); } catch { } throw; } + + // Invalidate node cache so vector clocks and last-entry hashes will be recomputed + _nodeCache?.Clear(); + // Notify changes + ChangesApplied?.Invoke(this, new ChangesAppliedEventArgs(oplogEntries)); + } + + public async Task RemoveRemotePeerAsync(string nodeId, CancellationToken cancellationToken = default) + { + using var connection = new SqliteConnection(_connectionString); + await connection.OpenAsync(cancellationToken); + await connection.ExecuteAsync("DELETE FROM RemotePeers WHERE NodeId = @NodeId", new { NodeId = nodeId }); + } + // Remote Peer Management + public async Task SaveRemotePeerAsync(RemotePeerConfiguration peer, CancellationToken cancellationToken = default) + { + using var connection = new SqliteConnection(_connectionString); + await connection.OpenAsync(cancellationToken); + + var sql = @" + INSERT OR REPLACE INTO RemotePeers (NodeId, Address, Type, OAuth2Json, IsEnabled) + VALUES (@NodeId, @Address, @Type, @OAuth2Json, @IsEnabled)"; + + await connection.ExecuteAsync(sql, new + { + peer.NodeId, + peer.Address, + Type = (int)peer.Type, + peer.OAuth2Json, + IsEnabled = peer.IsEnabled ? 1 : 0 + }); + + _logger.LogInformation("Saved remote peer configuration: {NodeId} ({Type})", peer.NodeId, peer.Type); + } + + public async Task> GetRemotePeersAsync(CancellationToken cancellationToken = default) + { + using var connection = new SqliteConnection(_connectionString); + await connection.OpenAsync(cancellationToken); + + var sql = "SELECT NodeId, Address, Type, OAuth2Json, IsEnabled FROM RemotePeers"; + var rows = await connection.QueryAsync(sql); + + return rows.Select(row => new RemotePeerConfiguration + { + NodeId = row.NodeId, + Address = row.Address, + Type = (PeerType)row.Type, + OAuth2Json = row.OAuth2Json, + IsEnabled = row.IsEnabled == 1 + }); + } + + public async Task CountDocumentsAsync(string collection, QueryNode? queryExpression, CancellationToken cancellationToken = default) + { + // Delegate to QueryDocumentsAsync to ensure identical filtering semantics, + // then count the resulting documents. + var documents = await QueryDocumentsAsync(collection, queryExpression, null, null, null, true, cancellationToken); + + if (documents is ICollection collectionDocuments) + { + return collectionDocuments.Count; + } + + return documents.Count(); + } + + public async Task EnsureIndexAsync(string collection, string propertyPath, CancellationToken cancellationToken = default) + { + using var connection = new SqliteConnection(_connectionString); + await connection.OpenAsync(cancellationToken); + + await EnsureCollectionTablesAsync(connection, collection); + + var tableName = GetDocumentTableName(collection); + var usePerCollection = _options?.UsePerCollectionTables == true; + + // Sanitize names to prevent injection + var safeColl = new string(collection.Where(char.IsLetterOrDigit).ToArray()); + var safeProp = new string(propertyPath.Where(c => char.IsLetterOrDigit(c) || c == '_' || c == '.').ToArray()); + var indexName = $"IDX_{safeColl}_{safeProp.Replace(".", "_")}"; + + string sql; + if (usePerCollection) + { + // Per-collection mode: simple index without collection filter + sql = $@"CREATE INDEX IF NOT EXISTS {indexName} + ON {tableName}(json_extract(JsonData, '$.{safeProp}'))"; + } + else + { + // Legacy mode: index with collection filter + sql = $@"CREATE INDEX IF NOT EXISTS {indexName} + ON Documents(json_extract(JsonData, '$.{safeProp}')) + WHERE Collection = '{safeColl}'"; + } + + await connection.ExecuteAsync(sql); + + _logger.LogInformation("Ensured index {IndexName} on {Collection}.{Property}", indexName, collection, propertyPath); + } + + public async Task> GetCollectionsAsync(CancellationToken cancellationToken = default) + { + using var connection = new SqliteConnection(_connectionString); + await connection.OpenAsync(cancellationToken).ConfigureAwait(false); + + if (_options?.UsePerCollectionTables == true) + { + // In per-collection mode, rely on known collection tables. + return GetKnownCollections(connection); + } + + // In legacy single-table mode, enumerate distinct collection names from the Documents table + const string sql = "SELECT DISTINCT Collection FROM Documents"; + var collections = await connection.QueryAsync( + new CommandDefinition(sql, cancellationToken: cancellationToken) + ).ConfigureAwait(false); + + return collections; } public async Task> QueryDocumentsAsync(string collection, QueryNode? queryExpression, int? skip = null, int? take = null, string? orderBy = null, bool ascending = true, CancellationToken cancellationToken = default) @@ -916,204 +1067,314 @@ public async Task> QueryDocumentsAsync(string collection, var rows = await connection.QueryAsync(sqlBuilder.ToString(), parameters); return rows.Select(r => { - var hlc = new HlcTimestamp(r.HlcWall, r.HlcLogic, r.HlcNode); - var content = r.JsonData != null + var hlc = new HlcTimestamp(r.HlcWall, r.HlcLogic, r.HlcNode); + var content = r.JsonData != null ? JsonSerializer.Deserialize(r.JsonData) : default; - return new Document(collection, r.Key, content, hlc, r.IsDeleted); + return new Document(collection, r.Key, content, hlc, r.IsDeleted != 0); }); } - public async Task CountDocumentsAsync(string collection, QueryNode? queryExpression, CancellationToken cancellationToken = default) + // --- Snapshotting Implementation --- + + public async Task PruneOplogAsync(HlcTimestamp cutoff, CancellationToken cancellationToken = default) { using var connection = new SqliteConnection(_connectionString); await connection.OpenAsync(cancellationToken); - - await EnsureCollectionTablesAsync(connection, collection); + using var transaction = connection.BeginTransaction(); - var tableName = GetDocumentTableName(collection); - var usePerCollection = _options?.UsePerCollectionTables == true; + try + { + // 1. Identify entries that will become the "boundary" (Max <= Cutoff) + var boundaries = await connection.QueryAsync<(string NodeId, long HlcWall, int HlcLogic, string Hash)>(@" + SELECT HlcNode as NodeId, HlcWall, HlcLogic, Hash + FROM Oplog o1 + WHERE (HlcWall, HlcLogic) = ( + SELECT MAX(HlcWall), MAX(HlcLogic) + FROM Oplog o2 + WHERE o2.HlcNode = o1.HlcNode + AND (o2.HlcWall < @Wall OR (o2.HlcWall = @Wall AND o2.HlcLogic <= @Logic)) + ) + GROUP BY HlcNode", + new { Wall = cutoff.PhysicalTime, Logic = cutoff.LogicalCounter }, transaction); + + // 2. Upsert SnapshotMetadata + foreach (var b in boundaries) + { + await connection.ExecuteAsync(@" + INSERT OR REPLACE INTO SnapshotMetadata (NodeId, HlcWall, HlcLogic, Hash) + VALUES (@NodeId, @HlcWall, @HlcLogic, @Hash)", + b, transaction); + } + + // 3. Delete old entries + await connection.ExecuteAsync(@" + DELETE FROM Oplog + WHERE HlcWall < @Wall OR (HlcWall = @Wall AND HlcLogic <= @Logic)", + new { Wall = cutoff.PhysicalTime, Logic = cutoff.LogicalCounter }, transaction); - var sqlBuilder = new StringBuilder(); - sqlBuilder.Append($"SELECT COUNT(*) FROM {tableName} WHERE IsDeleted = 0"); + transaction.Commit(); - var dynamicParams = new DynamicParameters(); - - if (!usePerCollection) + _logger.LogInformation("Pruned oplog entries older than {Cutoff}. Updated metadata for {Count} nodes.", cutoff, boundaries.Count()); + } + catch (SqliteException ex) when (ex.SqliteErrorCode == 11 || ex.SqliteErrorCode == 26) // SQLITE_CORRUPT or SQLITE_NOTADB { - sqlBuilder.Append(" AND Collection = @Collection"); - dynamicParams.Add("@Collection", collection); + _logger.LogCritical(ex, "Database corruption detected during oplog pruning (PruneOplogAsync)."); + try + { + transaction.Rollback(); + } + catch (Exception rollbackEx) + { + _logger.LogError(rollbackEx, "Failed to rollback transaction after database corruption."); + } + throw new CorruptDatabaseException("SQLite database is corrupt", ex); } - - if (queryExpression != null) + catch (Exception ex) { - var translator = new SqlQueryTranslator(); - var (whereClause, queryParams) = translator.Translate(queryExpression); - if (!string.IsNullOrEmpty(whereClause)) - { - sqlBuilder.Append($" AND ({whereClause})"); - dynamicParams.AddDynamicParams(queryParams); - } + _logger.LogError(ex, "Failed to prune oplog"); + throw; } - - return await connection.ExecuteScalarAsync(sqlBuilder.ToString(), dynamicParams); } - public async Task> GetCollectionsAsync(CancellationToken cancellationToken = default) + public async Task CreateSnapshotAsync(Stream destination, CancellationToken cancellationToken = default) { - using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(cancellationToken); - - var usePerCollection = _options?.UsePerCollectionTables == true; - - if (!usePerCollection) + // 1. Force a checkpoint to ensure WAL is merged (basic consistency) + using (var connection = new SqliteConnection(_connectionString)) { - // Legacy mode: query Documents table - return await connection.QueryAsync(@" - SELECT DISTINCT Collection - FROM Documents - ORDER BY Collection"); + await connection.OpenAsync(cancellationToken); + await connection.ExecuteAsync("PRAGMA wal_checkpoint(FULL);"); } - else + + // 2. Safely copy the DB file + var dbPath = new SqliteConnectionStringBuilder(_connectionString).DataSource; + + // We use a shared read lock approach or just copy. + // For strict consistency, we might need SQLite Online Backup API, but simple copy often suffices if WAL is checkpointed. + using (var sourceStream = new FileStream(dbPath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite)) { - // Per-collection mode: return known collections from cache - return GetKnownCollections(connection); + await sourceStream.CopyToAsync(destination, 81920, cancellationToken); } } - public async Task EnsureIndexAsync(string collection, string propertyPath, CancellationToken cancellationToken = default) + public async Task ReplaceDatabaseAsync(Stream databaseStream, CancellationToken cancellationToken = default) { - using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(cancellationToken); - - await EnsureCollectionTablesAsync(connection, collection); + _logger.LogWarning("Replacing database file from snapshot stream..."); - var tableName = GetDocumentTableName(collection); - var usePerCollection = _options?.UsePerCollectionTables == true; + // Ensure connections are cleared + SqliteConnection.ClearAllPools(); - // Sanitize names to prevent injection - var safeColl = new string(collection.Where(char.IsLetterOrDigit).ToArray()); - var safeProp = new string(propertyPath.Where(c => char.IsLetterOrDigit(c) || c == '_' || c == '.').ToArray()); - var indexName = $"IDX_{safeColl}_{safeProp.Replace(".", "_")}"; + var dbPath = new SqliteConnectionStringBuilder(_connectionString).DataSource; + var backupPath = dbPath + ".bak"; - string sql; - if (usePerCollection) - { - // Per-collection mode: simple index without collection filter - sql = $@"CREATE INDEX IF NOT EXISTS {indexName} - ON {tableName}(json_extract(JsonData, '$.{safeProp}'))"; - } - else + try { - // Legacy mode: index with collection filter - sql = $@"CREATE INDEX IF NOT EXISTS {indexName} - ON Documents(json_extract(JsonData, '$.{safeProp}')) - WHERE Collection = '{collection}'"; - } + // Backup current DB just in case + if (File.Exists(dbPath)) + { + File.Move(dbPath, backupPath); + } - await connection.ExecuteAsync(sql); - - _logger.LogInformation("Ensured index {IndexName} on {Collection}.{Property}", indexName, collection, propertyPath); - } + // Write new DB + using (var fileStream = File.Create(dbPath)) + { + databaseStream.Seek(0, SeekOrigin.Begin); // Ensure stream is at start + await databaseStream.CopyToAsync(fileStream, 81920, cancellationToken); + } - // Remote Peer Management - public async Task SaveRemotePeerAsync(RemotePeerConfiguration peer, CancellationToken cancellationToken = default) - { - using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(cancellationToken); + // Cleanup WAL/SHM to prevent corruption with new DB + var walPath = dbPath + "-wal"; + var shmPath = dbPath + "-shm"; + if (File.Exists(walPath)) File.Delete(walPath); + if (File.Exists(shmPath)) File.Delete(shmPath); - var sql = @" - INSERT OR REPLACE INTO RemotePeers (NodeId, Address, Type, OAuth2Json, IsEnabled) - VALUES (@NodeId, @Address, @Type, @OAuth2Json, @IsEnabled)"; + // Re-initialize (force strict check) + Initialize(); - await connection.ExecuteAsync(sql, new + // Cleanup backup if successful + if (File.Exists(backupPath)) File.Delete(backupPath); + + _logger.LogInformation("Database replaced successfully."); + } + catch (Exception ex) { - peer.NodeId, - peer.Address, - Type = (int)peer.Type, - peer.OAuth2Json, - IsEnabled = peer.IsEnabled ? 1 : 0 - }); + _logger.LogError(ex, "Failed to replace database. Attempting restore from backup..."); - _logger.LogInformation("Saved remote peer configuration: {NodeId} ({Type})", peer.NodeId, peer.Type); + // Restore backup + if (File.Exists(backupPath)) + { + if (File.Exists(dbPath)) File.Delete(dbPath); + File.Move(backupPath, dbPath); + } + + throw; + } } - public async Task> GetRemotePeersAsync(CancellationToken cancellationToken = default) + public async Task MergeSnapshotAsync(Stream snapshotStream, CancellationToken cancellationToken = default) { - using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(cancellationToken); + _logger.LogInformation("Merging remote snapshot into local database..."); - var sql = "SELECT NodeId, Address, Type, OAuth2Json, IsEnabled FROM RemotePeers"; - var rows = await connection.QueryAsync(sql); - - return rows.Select(row => new RemotePeerConfiguration + // 1. Save stream to temp file + var tempDbPath = Path.GetTempFileName(); + try { - NodeId = row.NodeId, - Address = row.Address, - Type = (PeerType)row.Type, - OAuth2Json = row.OAuth2Json, - IsEnabled = row.IsEnabled == 1 - }); - } + using (var fileStream = File.Create(tempDbPath)) + { + snapshotStream.Seek(0, SeekOrigin.Begin); + await snapshotStream.CopyToAsync(fileStream, 81920, cancellationToken); + } - public async Task RemoveRemotePeerAsync(string nodeId, CancellationToken cancellationToken = default) - { - using var connection = new SqliteConnection(_connectionString); - await connection.OpenAsync(cancellationToken); + // 2. Attach and Merge + using var connection = new SqliteConnection(_connectionString); + await connection.OpenAsync(cancellationToken); + + await connection.ExecuteAsync($"ATTACH DATABASE '{tempDbPath}' AS remote_snapshot"); + + using var transaction = connection.BeginTransaction(); + try + { + // Merge Oplog (Insert new, Ignore existing) + await connection.ExecuteAsync(@" + INSERT OR IGNORE INTO main.Oplog (Collection, Key, Operation, JsonData, HlcWall, HlcLogic, HlcNode, Hash, PreviousHash) + SELECT Collection, Key, Operation, JsonData, HlcWall, HlcLogic, HlcNode, Hash, PreviousHash + FROM remote_snapshot.Oplog"); - var sql = "DELETE FROM RemotePeers WHERE NodeId = @NodeId"; - var affected = await connection.ExecuteAsync(sql, new { NodeId = nodeId }); + // Merge SnapshotMetadata + await connection.ExecuteAsync(@" + INSERT OR REPLACE INTO main.SnapshotMetadata (NodeId, HlcWall, HlcLogic, Hash) + SELECT NodeId, HlcWall, HlcLogic, Hash + FROM remote_snapshot.SnapshotMetadata + WHERE 1=1 + ON CONFLICT(NodeId) DO UPDATE SET + HlcWall = MAX(HlcWall, excluded.HlcWall), + HlcLogic = MAX(HlcLogic, excluded.HlcLogic), + Hash = CASE WHEN excluded.HlcWall > HlcWall OR (excluded.HlcWall = HlcWall AND excluded.HlcLogic > HlcLogic) THEN excluded.Hash ELSE Hash END"); + + if (_options?.UsePerCollectionTables != true) + { + await connection.ExecuteAsync(@" + INSERT OR REPLACE INTO main.Documents (Collection, Key, JsonData, IsDeleted, HlcWall, HlcLogic, HlcNode) + SELECT r.Collection, r.Key, r.JsonData, r.IsDeleted, r.HlcWall, r.HlcLogic, r.HlcNode + FROM remote_snapshot.Documents r + LEFT JOIN main.Documents l ON l.Collection = r.Collection AND l.Key = r.Key + WHERE l.Key IS NULL + OR (r.HlcWall > l.HlcWall) + OR (r.HlcWall = l.HlcWall AND r.HlcLogic > l.HlcLogic)"); + } + else + { + var tables = await connection.QueryAsync("SELECT name FROM remote_snapshot.sqlite_master WHERE type='table' AND name LIKE 'Documents_%'"); + foreach(var table in tables) + { + var collectionName = table.Substring(10); + await EnsureCollectionTablesAsync(connection, collectionName); + + await connection.ExecuteAsync($@" + INSERT OR REPLACE INTO main.{table} (Key, JsonData, IsDeleted, HlcWall, HlcLogic, HlcNode) + SELECT r.Key, r.JsonData, r.IsDeleted, r.HlcWall, r.HlcLogic, r.HlcNode + FROM remote_snapshot.{table} r + LEFT JOIN main.{table} l ON l.Key = r.Key + WHERE l.Key IS NULL + OR (r.HlcWall > l.HlcWall) + OR (r.HlcWall = l.HlcWall AND r.HlcLogic > l.HlcLogic)"); + } + } + + transaction.Commit(); + } + finally + { + await connection.ExecuteAsync("DETACH DATABASE remote_snapshot"); + } - if (affected > 0) + InitializeNodeCache(connection); + _logger.LogInformation("Database merge completed successfully."); + } + catch (Exception ex) { - _logger.LogInformation("Removed remote peer configuration: {NodeId}", nodeId); + _logger.LogError(ex, "Failed to merge snapshot."); + throw; } - else + finally { - _logger.LogWarning("Attempted to remove non-existent remote peer: {NodeId}", nodeId); + if (File.Exists(tempDbPath)) File.Delete(tempDbPath); } } - // Inner classes for Dapper mapping - private class RemotePeerRow + public async Task ClearAllDataAsync(CancellationToken cancellationToken = default) { - public string NodeId { get; set; } = ""; - public string Address { get; set; } = ""; - public int Type { get; set; } - public string? OAuth2Json { get; set; } - public int IsEnabled { get; set; } + _logger.LogWarning("CLEARING ALL DATA FROM STORE!"); + using var connection = new SqliteConnection(_connectionString); + await connection.OpenAsync(cancellationToken); + + await connection.ExecuteAsync("DELETE FROM Oplog"); + await connection.ExecuteAsync("DELETE FROM SnapshotMetadata"); + + if (_options?.UsePerCollectionTables != true) + { + await connection.ExecuteAsync("DELETE FROM Documents"); + } + else + { + var tables = await connection.QueryAsync("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'Documents_%'"); + foreach(var table in tables) + { + await connection.ExecuteAsync($"DELETE FROM {table}"); + } + } + + await connection.ExecuteAsync("VACUUM"); + + lock (_cacheLock) + { + _nodeCache.Clear(); + } + + _logger.LogInformation("Store cleared successfully."); } +} - private class DocumentRow - { - public string Key { get; set; } = ""; - public string? JsonData { get; set; } - public bool IsDeleted { get; set; } - public long HlcWall { get; set; } - public int HlcLogic { get; set; } - public string HlcNode { get; set; } = ""; - } +// Inner classes for Dapper mapping +internal class RemotePeerRow +{ + public string NodeId { get; set; } = ""; + public string Address { get; set; } = ""; + public int Type { get; set; } + public string? OAuth2Json { get; set; } + public int IsEnabled { get; set; } +} - private class OplogRow - { - public string? Collection { get; set; } - public string? Key { get; set; } - public int Operation { get; set; } - public string? JsonData { get; set; } - public long HlcWall { get; set; } - public int HlcLogic { get; set; } - public string? HlcNode { get; set; } - public string? Hash { get; set; } - public string? PreviousHash { get; set; } - } +internal class DocumentRow +{ + public string Key { get; set; } = ""; + public string? JsonData { get; set; } + public int IsDeleted { get; set; } + public long HlcWall { get; set; } + public int HlcLogic { get; set; } + public string HlcNode { get; set; } = ""; +} - private class OplogRowPerCollection - { - public string? Key { get; set; } - public int Operation { get; set; } - public string? JsonData { get; set; } - public long HlcWall { get; set; } - public int HlcLogic { get; set; } - public string? HlcNode { get; set; } - } +internal class OplogRow +{ + public string? Collection { get; set; } + public string? Key { get; set; } + public int Operation { get; set; } + public string? JsonData { get; set; } + public long HlcWall { get; set; } + public int HlcLogic { get; set; } + public string? HlcNode { get; set; } + public string? Hash { get; set; } + public string? PreviousHash { get; set; } +} + +internal class OplogRowPerCollection +{ + public string? Key { get; set; } + public int Operation { get; set; } + public string? JsonData { get; set; } + public long HlcWall { get; set; } + public int HlcLogic { get; set; } + public string? HlcNode { get; set; } } + diff --git a/tests/EntglDb.Core.Tests/PeerCollectionTests.cs b/tests/EntglDb.Core.Tests/PeerCollectionTests.cs index abc790c..c6fd96a 100644 --- a/tests/EntglDb.Core.Tests/PeerCollectionTests.cs +++ b/tests/EntglDb.Core.Tests/PeerCollectionTests.cs @@ -211,6 +211,32 @@ public Task RemoveRemotePeerAsync(string nodeId, CancellationToken cancellationT _remotePeers.Remove(peer); return Task.CompletedTask; } + + public Task PruneOplogAsync(HlcTimestamp cutoff, CancellationToken cancellationToken = default) + { + _oplog.RemoveAll(e => e.Timestamp.CompareTo(cutoff) < 0); + return Task.CompletedTask; + } + + public Task ReplaceDatabaseAsync(Stream databaseStream, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task CreateSnapshotAsync(Stream destination, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task MergeSnapshotAsync(Stream snapshotStream, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task ClearAllDataAsync(CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } } [Fact] diff --git a/tests/EntglDb.Network.Tests/ConnectionTests.cs b/tests/EntglDb.Network.Tests/ConnectionTests.cs index b0b6a61..44e0bfe 100644 --- a/tests/EntglDb.Network.Tests/ConnectionTests.cs +++ b/tests/EntglDb.Network.Tests/ConnectionTests.cs @@ -109,6 +109,31 @@ private class StubStore : IPeerStore public Task GetLastEntryHashAsync(string nodeId, CancellationToken cancellationToken = default) => Task.FromResult(null); public Task GetEntryByHashAsync(string hash, CancellationToken cancellationToken = default) => Task.FromResult(null); public Task> GetChainRangeAsync(string startHash, string endHash, CancellationToken cancellationToken = default) => Task.FromResult>(new List()); + + public Task PruneOplogAsync(HlcTimestamp cutoff, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task CreateSnapshotAsync(Stream destination, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task ReplaceDatabaseAsync(Stream databaseStream, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task MergeSnapshotAsync(Stream snapshotStream, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task ClearAllDataAsync(CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } } private class StubAuthenticator : IAuthenticator diff --git a/tests/EntglDb.Network.Tests/HandshakeRegressionTests.cs b/tests/EntglDb.Network.Tests/HandshakeRegressionTests.cs index eb1d347..573f42d 100644 --- a/tests/EntglDb.Network.Tests/HandshakeRegressionTests.cs +++ b/tests/EntglDb.Network.Tests/HandshakeRegressionTests.cs @@ -41,6 +41,31 @@ class StubStore : IPeerStore public Task GetLastEntryHashAsync(string nodeId, CancellationToken cancellationToken = default) => Task.FromResult(null); public Task GetEntryByHashAsync(string hash, CancellationToken cancellationToken = default) => Task.FromResult(null); public Task> GetChainRangeAsync(string startHash, string endHash, CancellationToken cancellationToken = default) => Task.FromResult>(new List()); + + public Task PruneOplogAsync(HlcTimestamp cutoff, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task CreateSnapshotAsync(Stream destination, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task ReplaceDatabaseAsync(Stream databaseStream, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task MergeSnapshotAsync(Stream snapshotStream, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task ClearAllDataAsync(CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } } class StubConfigProvider : IPeerNodeConfigurationProvider diff --git a/tests/EntglDb.Network.Tests/VectorClockSyncTests.cs b/tests/EntglDb.Network.Tests/VectorClockSyncTests.cs index 5666f22..9085868 100644 --- a/tests/EntglDb.Network.Tests/VectorClockSyncTests.cs +++ b/tests/EntglDb.Network.Tests/VectorClockSyncTests.cs @@ -259,5 +259,30 @@ public Task> GetOplogForNodeAfterAsync(string nodeId, Hl public Task RemoveRemotePeerAsync(string nodeId, CancellationToken cancellationToken = default) => Task.CompletedTask; public Task SaveDocumentAsync(Document document, CancellationToken cancellationToken = default) => Task.CompletedTask; public Task SaveRemotePeerAsync(RemotePeerConfiguration peer, CancellationToken cancellationToken = default) => Task.CompletedTask; + + public Task PruneOplogAsync(HlcTimestamp cutoff, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task CreateSnapshotAsync(Stream destination, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task ReplaceDatabaseAsync(Stream databaseStream, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task MergeSnapshotAsync(Stream snapshotStream, CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + + public Task ClearAllDataAsync(CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } } }