using LiteDB; using LehrerApp.Sync.Models; namespace LehrerApp.Sync; /// /// Lokale Event-Queue in LiteDB. Puffert Events bis sie /// erfolgreich zum Server gepusht wurden. /// public class EventQueue : IDisposable { private readonly LiteDatabase _db; private readonly ILiteCollection _queue; private readonly ILiteCollection _meta; private readonly ILiteCollection _conflicts; private readonly ILiteCollection _attachmentUploads; private readonly ILiteCollection _entityVersions; private long _currentSeq; public EventQueue(string path) { _db = new LiteDatabase(path); _queue = _db.GetCollection("queue"); _meta = _db.GetCollection("meta"); _conflicts = _db.GetCollection("conflicts"); _attachmentUploads = _db.GetCollection("attachment_uploads"); _entityVersions = _db.GetCollection("entity_versions"); _attachmentUploads.EnsureIndex(x => x.StorageId, unique: true); _queue.EnsureIndex(x => x.SequenceNr); _currentSeq = _meta.FindById("seq")?.Value ?? 0; } public SyncEvent Enqueue(string deviceId, DeviceType deviceType, string entityType, string entityId, string operation, string payload) { var evt = new SyncEvent { DeviceId = deviceId, DeviceType = deviceType, Timestamp = DateTime.UtcNow, SequenceNr = ++_currentSeq, EntityType = entityType, EntityId = entityId, Operation = operation, Payload = payload, }; _queue.Insert(evt); _meta.Upsert(new SyncMeta { Id = "seq", Value = _currentSeq }); return evt; } public List GetPending(int max = SyncProtocol.PushBatchSize, IReadOnlySet? excludedEventIds = null) { var pending = _queue.Find(Query.All(nameof(SyncEvent.SequenceNr))).AsEnumerable(); if (excludedEventIds is not null) pending = pending.Where(e => !excludedEventIds.Contains(e.EventId)); return pending.Take(max).ToList(); } public int PendingCount() => _queue.Count(); public void Acknowledge(IEnumerable ids) { foreach (var id in ids) _queue.Delete(id); } public long GetLastServerSeq() => _meta.FindById("serverSeq")?.Value ?? 0; public void SetLastServerSeq(long nr) => _meta.Upsert(new SyncMeta { Id = "serverSeq", Value = nr }); public DateTime? GetLastSyncAt() => _meta.FindById("lastSync")?.Timestamp; public void SetLastSyncAt(DateTime dt) => _meta.Upsert(new SyncMeta { Id = "lastSync", Value = 0, Timestamp = dt }); public void AddConflict(ConflictEntry c) => _conflicts.Insert(c); public List GetUnreviewed() => _conflicts.Find(c => !c.Reviewed).ToList(); public int ConflictCount() => _conflicts.Count(c => !c.Reviewed); public void MarkReviewed(Guid id) { var conflict = _conflicts.FindById(id); if (conflict is null) return; conflict.Reviewed = true; _conflicts.Update(conflict); } public void ClearConflicts() => _conflicts.DeleteAll(); // ── Lokale Versionsverfolgung je Entität (optimistische Nebenläufigkeitskontrolle) ────── // Merkt sich pro Entität die zuletzt bekannte ServerSeq — Grundlage für SyncEvent. // BasedOnServerSeq beim Push (siehe SyncEngine.PushAsync) und dafür, wie ein abgelehnter // Push nach dem Nachladen des aktuellen Server-Stands aufgelöst wird. public long? GetKnownServerSeq(string entityType, string entityId) => _entityVersions.FindById(EntityVersionKey(entityType, entityId))?.ServerSeq; public void SetKnownServerSeq(string entityType, string entityId, long serverSeq) => _entityVersions.Upsert(new EntityVersion { Key = EntityVersionKey(entityType, entityId), ServerSeq = serverSeq }); /// Löscht den bekannten Stand EINER Entität — z.B. wenn der Server auf eine /// BasedOnServerSeq-Ablehnung hin meldet, die Entität gar nicht zu kennen (404 bei /// GetLatestForEntity): der lokale Cache war dann stale, siehe SyncEngine.HandleRejectedAsync /// und TODO 10.3.5. public void ClearKnownServerSeq(string entityType, string entityId) => _entityVersions.Delete(EntityVersionKey(entityType, entityId)); /// Verwirft die GESAMTE lokale Versionsverfolgung — nötig nach einem Kontowechsel /// (siehe SettingsViewModel.SyncLogin/SyncForceFullResync), da ServerSeq-Werte ausschließlich /// innerhalb des Event-Logs EINES Server-Kontos bedeutungsvoll sind (TODO 10.3.5). Sicher: der /// nächste Push behandelt jede Entität dann als "erstmals für dieses Konto", der Server nimmt /// sie an, solange er sie unter der aktuellen userId selbst noch nicht kennt. public void ResetKnownServerSeqs() => _entityVersions.DeleteAll(); private static string EntityVersionKey(string entityType, string entityId) => $"{entityType}:{entityId}"; // ── Anhang-Warteliste (getrennt von der JSON-Ereignis-Outbox, siehe AttachmentSyncer) ──── public void QueueAttachmentUpload(string storageId) { if (!_attachmentUploads.Exists(a => a.StorageId == storageId)) _attachmentUploads.Insert(new PendingAttachmentUpload { StorageId = storageId }); } public List GetPendingAttachmentUploads() => _attachmentUploads.FindAll().Select(a => a.StorageId).ToList(); public void MarkAttachmentUploaded(string storageId) => _attachmentUploads.DeleteMany(a => a.StorageId == storageId); public void Dispose() => _db.Dispose(); } public class ConflictEntry { public Guid Id { get; init; } = Guid.NewGuid(); public DateTime DetectedAt { get; init; } = DateTime.UtcNow; public SyncEvent LocalEvent { get; init; } = null!; public SyncEvent RemoteEvent { get; init; } = null!; public string Resolution { get; init; } = ""; public bool Reviewed { get; set; } } internal class SyncMeta { public string Id { get; set; } = ""; public long Value { get; set; } public DateTime? Timestamp { get; set; } } internal class PendingAttachmentUpload { public ObjectId Id { get; set; } = ObjectId.NewObjectId(); public string StorageId { get; set; } = ""; } internal class EntityVersion { [BsonId] public string Key { get; set; } = ""; public long ServerSeq { get; set; } }