Explorer-Workbench/src/Explorer.Indexing/UsnChangeApplier.cs

using Explorer.Domain;
using Explorer.Domain.Abstractions;
using Microsoft.Extensions.Logging;

namespace Explorer.Indexing;

public sealed class UsnChangeApplier
{
    private readonly IIndexStore _store;
    private readonly IFileSystemEnumerator _enumerator;
    private readonly ILogger<UsnChangeApplier> _logger;
    private readonly ArchiveContentsIndexer? _archives;

    public UsnChangeApplier(
        IIndexStore store,
        IFileSystemEnumerator enumerator,
        ILogger<UsnChangeApplier> logger,
        ArchiveContentsIndexer? archives = null)
    {
        _store = store;
        _enumerator = enumerator;
        _logger = logger;
        _archives = archives;
    }

    public async Task<UsnReadStatus> ApplyAsync(Source source, IUsnJournal journal, CancellationToken cancellationToken)
    {
        if (source.LastRootPath is null || source.Kind is not (SourceKind.NtfsLocal or SourceKind.Removable))
        {
            return UsnReadStatus.Unavailable;
        }

        if (!journal.TryQuery(source.LastRootPath, out var current, out _))
        {
            return UsnReadStatus.Unavailable;
        }

        var from = new UsnJournalState
        {
            JournalId = source.UsnJournalId ?? 0,
            NextUsn = source.UsnNext ?? 0
        };

        if (from.JournalId != 0 && from.JournalId != current.JournalId)
        {
            await _store.Sources.UpdateStatusAsync(source.Id, SourceStatus.Stale, "Change journal reset", cancellationToken)
                .ConfigureAwait(false);
            return UsnReadStatus.JournalReset;
        }

        if (from.JournalId == 0)
        {
            await _store.Sources.UpdateUsnAsync(source.Id, current.JournalId, current.NextUsn, cancellationToken)
                .ConfigureAwait(false);
            return UsnReadStatus.Ok;
        }

        var records = journal.Read(source.LastRootPath, from, 4000, out var next, out var status);
        if (status is UsnReadStatus.JournalReset)
        {
            await _store.Sources.UpdateStatusAsync(source.Id, SourceStatus.Stale, "Change journal reset", cancellationToken)
                .ConfigureAwait(false);
            return status;
        }

        if (status != UsnReadStatus.Ok)
        {
            return status;
        }

        var coalesced = Coalesce(records);
        var now = DateTimeOffset.UtcNow;
        var expand = new List<IndexEntry>();
        await _store.RunWriteAsync(async s =>
        {
            foreach (var rec in coalesced)
            {
                cancellationToken.ThrowIfCancellationRequested();
                try
                {
                    var updated = await ApplyRecord(s, source, rec, now, cancellationToken).ConfigureAwait(false);
                    if (updated is not null)
                    {
                        expand.Add(updated);
                    }
                }
                catch (Exception ex)
                {
                    _logger.LogDebug(ex, "USN apply failed for {Name}", rec.FileName);
                }
            }

            await s.Sources.UpdateUsnAsync(source.Id, next.JournalId, next.NextUsn, cancellationToken).ConfigureAwait(false);
        }, cancellationToken).ConfigureAwait(false);

        if (_archives is { IsEnabled: true })
        {
            foreach (var archive in expand)
            {
                await _archives.ExpandIfNeededAsync(source, archive, now, source.ScanGeneration, cancellationToken)
                    .ConfigureAwait(false);
            }
        }

        return UsnReadStatus.Ok;
    }

    private async Task<IndexEntry?> ApplyRecord(IIndexStore store, Source source, UsnRecord rec, DateTimeOffset now, CancellationToken cancellationToken)
    {
        var existing = rec.FileReferenceNumber != 0
            ? await store.Entries.GetByFileIdAsync(source.Id, rec.FileReferenceNumber, cancellationToken).ConfigureAwait(false)
            : null;

        if (rec.IsDelete)
        {
            if (existing is not null)
            {
                await store.Entries.TombstoneAsync(existing.Id, now, cancellationToken).ConfigureAwait(false);
                if (!existing.IsDirectory && ArchiveFormats.IsArchive(existing.Name))
                {
                    await store.Entries.TombstoneByPathPrefixAsync(source.Id, existing.PathRel, now, cancellationToken)
                        .ConfigureAwait(false);
                }
            }

            return null;
        }

        var parent = rec.ParentFileReferenceNumber != 0
            ? await store.Entries.GetByFileIdAsync(source.Id, rec.ParentFileReferenceNumber, cancellationToken).ConfigureAwait(false)
            : await store.Entries.GetRootAsync(source.Id, cancellationToken).ConfigureAwait(false);

        var parentRel = parent?.PathRel ?? "";
        var rel = string.IsNullOrEmpty(parentRel) ? rec.FileName : parentRel + "\\" + rec.FileName;
        var full = PathRules.Combine(source.LastRootPath!, rel);
        var live = _enumerator.GetItem(full);

        if (live is null)
        {
            if (existing is not null)
            {
                await store.Entries.TombstoneAsync(existing.Id, now, cancellationToken).ConfigureAwait(false);
                if (!existing.IsDirectory && ArchiveFormats.IsArchive(existing.Name))
                {
                    await store.Entries.TombstoneByPathPrefixAsync(source.Id, existing.PathRel, now, cancellationToken)
                        .ConfigureAwait(false);
                }
            }

            return null;
        }

        var oldSize = existing is { IsDirectory: false } ? existing.SizeBytes : 0;
        var entry = new IndexEntry
        {
            SourceId = source.Id,
            ParentId = parent?.Id,
            Name = live.Name,
            NameNorm = NameNormalizer.Normalize(live.Name),
            Extension = live.IsDirectory ? null : NameNormalizer.Extension(live.Name),
            IsDirectory = live.IsDirectory,
            SizeBytes = live.IsDirectory ? 0 : live.SizeBytes,
            AggregateSize = existing?.AggregateSize ?? (live.IsDirectory ? 0 : live.SizeBytes),
            CreatedUtc = live.CreatedUtc,
            ModifiedUtc = live.ModifiedUtc,
            LastSeenUtc = now,
            LastIndexedUtc = now,
            Attributes = live.Attributes,
            FileId = rec.FileReferenceNumber,
            ParentFileId = rec.ParentFileReferenceNumber,
            ReparseTag = live.ReparseTag,
            Status = EntryStatus.Present,
            PathRel = rel,
            ScanGeneration = source.ScanGeneration
        };

        if (existing is not null && !string.Equals(existing.PathRel, rel, StringComparison.OrdinalIgnoreCase))
        {
            await store.Entries.RenameSubtreePathAsync(source.Id, existing.PathRel, rel, cancellationToken)
                .ConfigureAwait(false);
        }

        entry.Id = await store.Entries.UpsertAsync(entry, cancellationToken).ConfigureAwait(false);
        if (!live.IsDirectory)
        {
            var delta = live.SizeBytes - oldSize;
            if (existing is null)
            {
                await store.Entries.ApplySizeDeltaToAncestorsAsync(parent?.Id, live.SizeBytes, 1, 0, cancellationToken)
                    .ConfigureAwait(false);
            }
            else if (delta != 0)
            {
                await store.Entries.ApplySizeDeltaToAncestorsAsync(parent?.Id, delta, 0, 0, cancellationToken)
                    .ConfigureAwait(false);
            }

            if (ArchiveFormats.IsArchive(live.Name))
            {
                return entry;
            }
        }

        return null;
    }

    private static List<UsnRecord> Coalesce(IReadOnlyList<UsnRecord> records)
    {
        var map = new Dictionary<long, UsnRecord>();
        foreach (var rec in records)
        {
            map[rec.FileReferenceNumber] = rec;
        }

        return map.Values.ToList();
    }
}