Added mailboxes as local data sources (#1022)

app/MindWork AI Studio/Tools/Services/DataSourceEmbeddingService.Watchers.cs

@@ -1,17 +1,19 @@
-using System.Collections.Concurrent;
-using AIStudio.Settings;
-using AIStudio.Settings.DataModel;
-
 namespace AIStudio.Tools.Services;
 
+/// <remarks>
+/// A watcher here is whatever an indexer uses to notice that one of its data sources changed, a file
+/// system watcher for local files, an interval for mailboxes. The indexers own them; this service
+/// only decides when they run.
+/// </remarks>
 public sealed partial class DataSourceEmbeddingService
 {
-    private const int WATCHER_DEBOUNCE_SECONDS = 2;
-
-    private readonly ConcurrentDictionary<string, DataSourceWatcherRegistration> watchers = new(StringComparer.OrdinalIgnoreCase);
-    private readonly Dictionary<string, CancellationTokenSource> watcherDebounceTokens = new(StringComparer.OrdinalIgnoreCase);
-    private readonly object watcherDebounceLock = new();
-    
+    /// <summary>
+    /// Lets every indexer track changes to its data sources, or stops all tracking.
+    /// </summary>
+    /// <remarks>
+    /// Tracking runs only while local data sources refresh on their own, and not before the startup
+    /// hash check is done.
+    /// </remarks>
     private void RefreshWatchers()
     {
         if (!settingsManager.ConfigurationData.App.DataSourceIndexing.AutomaticRefresh)
@@ -27,263 +29,56 @@ public sealed partial class DataSourceEmbeddingService
             return;
         }
 
-        var supportedSources = settingsManager.ConfigurationData.DataSources
-            .Where(this.IsSupportedInternalDataSource)
-            .ToDictionary(source => source.Id, StringComparer.OrdinalIgnoreCase);
-
-        foreach (var existingWatcherId in this.watchers.Keys.Except(supportedSources.Keys, StringComparer.OrdinalIgnoreCase).ToList())
-            this.RemoveWatcher(existingWatcherId);
-
-        foreach (var dataSource in supportedSources.Values)
-            this.EnsureWatcher(dataSource);
-    }
-
-    private void EnsureWatcher(IDataSource dataSource)
-    {
-        if (!settingsManager.ConfigurationData.App.DataSourceIndexing.AutomaticRefresh)
-            return;
-
-        var configuration = GetWatchConfiguration(dataSource);
-        if (configuration is null)
-            return;
-
-        if (this.watchers.TryGetValue(dataSource.Id, out var existingRegistration))
-        {
-            if (IsSameWatchConfiguration(existingRegistration.Configuration, configuration))
-                return;
-
-            this.RemoveWatcher(dataSource.Id);
-        }
-
-        var watcher = this.CreateWatcher(dataSource.Id, configuration);
-        if (watcher is null)
-            return;
-
-        if (!this.watchers.TryAdd(dataSource.Id, new DataSourceWatcherRegistration(watcher, configuration)))
-            watcher.Dispose();
-    }
-
-    private FileSystemWatcher? CreateWatcher(string dataSourceId, DataSourceWatcherConfiguration configuration)
-    {
-        try
-        {
-            var watcher = new FileSystemWatcher(configuration.RootPath)
-            {
-                Filter = configuration.Filter,
-                IncludeSubdirectories = configuration.IncludeSubdirectories,
-                NotifyFilter = NotifyFilters.FileName | NotifyFilters.DirectoryName | NotifyFilters.LastWrite | NotifyFilters.CreationTime | NotifyFilters.Size,
-            };
-
-            watcher.Changed += (_, args) => this.OnWatchedDataSourceChanged(dataSourceId, configuration, args);
-            watcher.Deleted += (_, args) => this.OnWatchedDataSourceChanged(dataSourceId, configuration, args);
-            watcher.Created += (_, args) => this.OnWatchedDataSourceChanged(dataSourceId, configuration, args);
-            watcher.Renamed += (_, args) => this.OnWatchedDataSourceChanged(dataSourceId, configuration, args);
-            watcher.Error += (_, args) =>
-            {
-                logger.LogWarning(args.GetException(), "The file watcher for data source '{DataSourceId}' failed. Recreating it.", dataSourceId);
-                this.RemoveWatcher(dataSourceId);
-                this.EnsureWatcher(dataSourceId);
-                this.ScheduleWatchedDataSourceRefresh(dataSourceId);
-            };
-            watcher.EnableRaisingEvents = true;
-            return watcher;
-        }
-        catch (Exception exception)
-        {
-            logger.LogWarning(exception, "Failed to create file watcher for data source '{DataSourceId}' at '{RootPath}'.", dataSourceId, configuration.RootPath);
-            return null;
-        }
+        var supportedSources = this.GetConfiguredIndexedSources();
+        foreach (var indexer in this.indexers)
+            indexer.TrackChanges(supportedSources.Where(indexer.Supports).ToList(), this.RequestRunAsync);
     }
 
     private void RemoveWatcher(string dataSourceId)
     {
-        this.CancelPendingWatcherRefresh(dataSourceId);
-
-        if (this.watchers.TryRemove(dataSourceId, out var registration))
-            registration.Watcher.Dispose();
+        foreach (var indexer in this.indexers)
+            indexer.StopTracking(dataSourceId);
     }
-    
+
     private void RemoveAllWatchers()
     {
-        foreach (var watcherId in this.watchers.Keys.ToList())
-            this.RemoveWatcher(watcherId);
+        foreach (var indexer in this.indexers)
+            indexer.StopTrackingAll();
     }
 
     private void DisposeWatchers()
     {
-        this.CancelAllPendingWatcherRefreshes();
-
-        foreach (var registration in this.watchers.Values)
-            registration.Watcher.Dispose();
-
-        this.watchers.Clear();
+        foreach (var indexer in this.indexers)
+            indexer.Dispose();
     }
 
-    private void OnWatchedDataSourceChanged(string dataSourceId, DataSourceWatcherConfiguration configuration, FileSystemEventArgs args)
+    /// <summary>
+    /// Queues the run an indexer asked for after it noticed a change, or because its interval came round.
+    /// </summary>
+    /// <param name="dataSourceId">The id of the data source which changed.</param>
+    /// <param name="refreshMode">Why the indexer asks.</param>
+    private async Task RequestRunAsync(string dataSourceId, DataSourceEmbeddingRefreshMode refreshMode)
     {
-        if (!this.IsRelevantWatcherEvent(configuration, args))
+        if (!this.TryGetConfiguredIndexedSource(dataSourceId, out var dataSource))
+            return;
+
+        if (this.statuses.TryGetValue(dataSourceId, out var status) && IsWaitingForSignIn(status))
         {
-            logger.LogDebug(
-                "Ignoring file system change for data source '{DataSourceId}' at '{Path}' (event={ChangeType}) because the path is not part of the RAG index.",
-                dataSourceId,
-                args.FullPath,
-                args.ChangeType);
+            logger.LogDebug("Skipped the requested run of data source '{DataSourceId}' because its sign-in failed and waits for the user. RefreshMode={RefreshMode}.", dataSourceId, refreshMode);
             return;
         }
 
-        logger.LogDebug(
-            "Detected relevant file system change for data source '{DataSourceId}' at '{Path}' (event={ChangeType}). Scheduling a debounced embedding run.",
-            dataSourceId,
-            args.FullPath,
-            args.ChangeType);
-
-        this.ScheduleWatchedDataSourceRefresh(dataSourceId);
+        await this.QueueDataSourceAsync(dataSource, true, refreshMode);
     }
 
-    private void ScheduleWatchedDataSourceRefresh(string dataSourceId)
-    {
-        if (!settingsManager.ConfigurationData.App.DataSourceIndexing.AutomaticRefresh)
-            return;
-
-        var debounceToken = new CancellationTokenSource();
-
-        lock (this.watcherDebounceLock)
-        {
-            if (this.watcherDebounceTokens.Remove(dataSourceId, out var existingToken))
-                existingToken.Cancel();
-
-            this.watcherDebounceTokens[dataSourceId] = debounceToken;
-        }
-
-        _ = Task.Run(async () =>
-        {
-            try
-            {
-                await Task.Delay(TimeSpan.FromSeconds(WATCHER_DEBOUNCE_SECONDS), debounceToken.Token);
-                if (!this.TryCompletePendingWatcherRefresh(dataSourceId, debounceToken))
-                    return;
-
-                var dataSource = settingsManager.ConfigurationData.DataSources
-                    .FirstOrDefault(source => source.Id.Equals(dataSourceId, StringComparison.OrdinalIgnoreCase));
-
-                if (dataSource is not null)
-                {
-                    logger.LogInformation("Queueing data source '{DataSourceName}' ({DataSourceId}) after file system changes settled. The hash pipeline will reindex only changed files.", dataSource.Name, dataSource.Id);
-                    await this.QueueDataSourceAsync(dataSource, true, DataSourceEmbeddingRefreshMode.WATCHER_HASH_CHECK);
-                }
-            }
-            catch (OperationCanceledException)
-            {
-            }
-            catch (Exception exception)
-            {
-                logger.LogWarning(exception, "Failed to queue watched data source '{DataSourceId}' after a file system change.", dataSourceId);
-            }
-            finally
-            {
-                debounceToken.Dispose();
-            }
-        });
-    }
-
-    private void EnsureWatcher(string dataSourceId)
-    {
-        var dataSource = settingsManager.ConfigurationData.DataSources
-            .FirstOrDefault(source => source.Id.Equals(dataSourceId, StringComparison.OrdinalIgnoreCase));
-
-        if (dataSource is not null)
-            this.EnsureWatcher(dataSource);
-    }
-
-    private void CancelPendingWatcherRefresh(string dataSourceId)
-    {
-        lock (this.watcherDebounceLock)
-        {
-            if (this.watcherDebounceTokens.Remove(dataSourceId, out var token))
-                token.Cancel();
-        }
-    }
-
-    private void CancelAllPendingWatcherRefreshes()
-    {
-        lock (this.watcherDebounceLock)
-        {
-            foreach (var token in this.watcherDebounceTokens.Values)
-                token.Cancel();
-
-            this.watcherDebounceTokens.Clear();
-        }
-    }
-
-    private bool TryCompletePendingWatcherRefresh(string dataSourceId, CancellationTokenSource debounceToken)
-    {
-        lock (this.watcherDebounceLock)
-        {
-            if (!this.watcherDebounceTokens.TryGetValue(dataSourceId, out var currentToken) || !ReferenceEquals(currentToken, debounceToken))
-                return false;
-
-            this.watcherDebounceTokens.Remove(dataSourceId);
-            return true;
-        }
-    }
-
-    private bool IsRelevantWatcherEvent(DataSourceWatcherConfiguration configuration, FileSystemEventArgs args)
-    {
-        if (args is RenamedEventArgs renamedArgs)
-        {
-            return this.IsRelevantWatcherPath(configuration, renamedArgs.FullPath, args.ChangeType)
-                   || this.IsRelevantWatcherPath(configuration, renamedArgs.OldFullPath, args.ChangeType);
-        }
-
-        return this.IsRelevantWatcherPath(configuration, args.FullPath, args.ChangeType);
-    }
-
-    private bool IsRelevantWatcherPath(DataSourceWatcherConfiguration configuration, string path, WatcherChangeTypes changeType)
-    {
-        if (string.IsNullOrWhiteSpace(path))
-            return false;
-
-        var fileName = Path.GetFileName(path);
-        if (string.IsNullOrWhiteSpace(fileName))
-            return true;
-
-        if (!configuration.IncludeSubdirectories && !string.Equals(fileName, configuration.Filter, StringComparison.OrdinalIgnoreCase))
-            return false;
-
-        if (Directory.Exists(path))
-            return true;
-
-        if (IsSkippedRagFileName(fileName))
-            return false;
-
-        if (this.IsSupportedRagFilePath(path))
-            return true;
-
-        return changeType is WatcherChangeTypes.Deleted or WatcherChangeTypes.Renamed
-               && string.IsNullOrWhiteSpace(Path.GetExtension(path));
-    }
-
-    private static DataSourceWatcherConfiguration? GetWatchConfiguration(IDataSource dataSource) => dataSource switch
-    {
-        DataSourceLocalDirectory localDirectory when Directory.Exists(localDirectory.Path) => new DataSourceWatcherConfiguration(
-            localDirectory.Path,
-            "*.*",
-            true),
-        DataSourceLocalFile localFile when File.Exists(localFile.FilePath) && !string.IsNullOrWhiteSpace(Path.GetDirectoryName(localFile.FilePath)) => new DataSourceWatcherConfiguration(
-            Path.GetDirectoryName(localFile.FilePath)!,
-            Path.GetFileName(localFile.FilePath),
-            false),
-        _ => null,
-    };
-
-    private static bool IsSameWatchConfiguration(DataSourceWatcherConfiguration left, DataSourceWatcherConfiguration right)
-    {
-        return left.IncludeSubdirectories == right.IncludeSubdirectories
-               && string.Equals(left.RootPath, right.RootPath, StringComparison.OrdinalIgnoreCase)
-               && string.Equals(left.Filter, right.Filter, StringComparison.OrdinalIgnoreCase);
-    }
-
-    private sealed record DataSourceWatcherConfiguration(string RootPath, string Filter, bool IncludeSubdirectories);
-
-    private sealed record DataSourceWatcherRegistration(FileSystemWatcher Watcher, DataSourceWatcherConfiguration Configuration);
+    /// <summary>
+    /// Whether a data source waits for the user since its sign-in failed.
+    /// </summary>
+    /// <remarks>
+    /// Then only the user starts its next run, by saving a new password or with a retry. The indexer
+    /// would not sign in on any other run either, but every one of them would make the row of the
+    /// data source change from failed to queued and back, every interval anew.
+    /// </remarks>
+    /// <param name="status">The current status of the data source.</param>
+    internal static bool IsWaitingForSignIn(DataSourceEmbeddingStatus status) => status.Attention is DataSourceAttention.AUTH_FAILED;
 }