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 = 200) =>
_queue.Find(Query.All(nameof(SyncEvent.SequenceNr))).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);
}
// ── 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; }
}