Setup reading and persistence

.gitignore

@@ -1,3 +1,4 @@
 robin/bin/*
 robin/obj/*
 robin/.env
+.vs/*

robin/Configuration/BuilderExtensions.cs

@@ -10,7 +10,9 @@ public static class BuilderExtensions
         this IHostApplicationBuilder builder
     )
     {
-        DotNetEnv.Env.Load();
+#if DEBUG
+        DotNetEnv.Env.Load(Path.Combine(AppContext.BaseDirectory, ".env"));
+#endif
         builder.ThrowIfNull().Configuration.AddEnvironmentVariables();
         return builder;
     }
@@ -20,7 +22,7 @@ public static class BuilderExtensions
         builder
             .ThrowIfNull()
             .Configuration.AddJsonFile(
-                path: "appsettings.json",
+                path: Path.Combine(AppContext.BaseDirectory, "appsettings.json"),
                 optional: false,
                 reloadOnChange: true
             );

robin/Options/BuilderExtensions.cs

@@ -9,7 +9,15 @@ public static class BuilderExtensions
     {
         builder
             .Services.AddOptions<MonitorOptions>()
-            .Bind(builder.Configuration.GetSection("Monitor"));
+            .Bind(builder.Configuration.GetSection(MonitorOptions.SectionName));
+
+        builder
+            .Services.AddOptions<ElasticOptions>()
+            .Bind(builder.Configuration.GetSection(ElasticOptions.SectionName));
+
+        builder
+            .Services.AddOptions<PositionPersistenceOptions>()
+            .Bind(builder.Configuration.GetSection(PositionPersistenceOptions.SectionName));
 
         return builder;
     }

robin/Options/ElasticOptions.cs

@@ -0,0 +1,11 @@
+namespace Robin;
+
+public sealed class ElasticOptions
+{
+    public const string SectionName = "Elastic";
+
+    public Uri? Uri { get; set; }
+    public string? Index { get; set; }
+    public string? Username { get; set; }
+    public string? Password { get; set; }
+}

robin/Options/MonitorOptions.cs

@@ -1,6 +1,8 @@
 namespace Robin.Options;
 
-public sealed class MonitorOptions()
+public sealed class MonitorOptions
 {
-    public required List<string> Paths { get; init; } = [];
+    public const string SectionName = "Monitor";
+
+    public List<string> Paths { get; set; } = [];
 }

robin/Options/PositionPersistenceOptions.cs

@@ -0,0 +1,8 @@
+namespace Robin.Options;
+
+public sealed class PositionPersistenceOptions
+{
+    public const string SectionName = "PositionPersistence";
+
+    public string? FilePath { get; set; }
+}

robin/Robin.csproj

@@ -15,6 +15,7 @@
     <PackageReference Include="Microsoft.Extensions.FileSystemGlobbing" Version="10.0.0" />
     <PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.0" />
     <PackageReference Include="Microsoft.Extensions.Options.DataAnnotations" Version="10.0.0" />
+    <PackageReference Include="NEST" Version="7.17.5" />
     <PackageReference Include="Serilog.Enrichers.Process" Version="3.0.0" />
     <PackageReference Include="Serilog.Extensions.Logging" Version="10.0.0" />
     <PackageReference Include="Serilog.Formatting.Elasticsearch" Version="10.0.0" />

robin/Serialization/SerializationContext.cs

@@ -0,0 +1,7 @@
+using System.Collections.Concurrent;
+using System.Text.Json.Serialization;
+
+namespace Robin.Serialization;
+
+[JsonSerializable(typeof(ConcurrentDictionary<string, long>))]
+public partial class SerializationContext : JsonSerializerContext { }

robin/Services/BuilderExtensions.cs

@@ -1,6 +1,11 @@
 using System.Core.Extensions;
+using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
+using Nest;
+using Robin.Services.Monitoring;
+using Robin.Services.Persistence;
+using Robin.Services.Publishing;
 
 namespace Robin.Services;
 
@@ -8,9 +13,31 @@ public static class BuilderExtensions
 {
     public static IHostApplicationBuilder AddRobinServices(this IHostApplicationBuilder builder)
     {
+        var elasticClientOptions = builder
+            .Configuration.GetSection(ElasticOptions.SectionName)
+            .Get<ElasticOptions>();
+
+        if (elasticClientOptions is null)
+        {
+            throw new InvalidOperationException(
+                "ElasticOptions section is missing in configuration."
+            );
+        }
+
         builder.ThrowIfNull();
         builder.Services.AddHostedService<MonitoringService>();
-
+        builder.Services.AddSingleton<IElasticClient>(
+            new ElasticClient(
+                new ConnectionSettings(elasticClientOptions.Uri)
+                    .BasicAuthentication(
+                        elasticClientOptions.Username,
+                        elasticClientOptions.Password
+                    )
+                    .DefaultIndex(elasticClientOptions.Index)
+            )
+        );
+        builder.Services.AddSingleton<IPublishingService, ElasticPublishingService>();
+        builder.Services.AddSingleton<PositionPersistenceService>();
         return builder;
     }
 }

robin/Services/Monitoring/MonitoringService.cs

@@ -1,19 +1,24 @@
+using System.Extensions;
 using System.Extensions.Core;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using Robin.Options;
+using Robin.Services.Persistence;
 
-namespace Robin.Services;
+namespace Robin.Services.Monitoring;
 
 public sealed class MonitoringService(
+    PositionPersistenceService persistenceService,
     IOptions<MonitorOptions> options,
     ILogger<MonitoringService> logger
 ) : IHostedLifecycleService
 {
+    private readonly PositionPersistenceService persistenceService = persistenceService;
     private readonly MonitorOptions options = options.Value;
+    private readonly CancellationTokenSource cancellationTokenSource = new();
     private readonly ILogger<MonitoringService> logger = logger;
-    private readonly List<FileSystemWatcher> watchers = [];
+    private readonly List<Task> watchers = [];
 
     public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
 
@@ -22,12 +27,18 @@ public sealed class MonitoringService(
 
     public Task StartingAsync(CancellationToken cancellationToken) => Task.CompletedTask;
 
-    public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+    public Task StopAsync(CancellationToken cancellationToken) => this.CancelMonitors();
 
     public Task StoppedAsync(CancellationToken cancellationToken) => Task.CompletedTask;
 
     public Task StoppingAsync(CancellationToken cancellationToken) => Task.CompletedTask;
 
+    public Task CancelMonitors()
+    {
+        this.cancellationTokenSource.Cancel();
+        return Task.WhenAll(this.watchers);
+    }
+
     public async Task MonitorFiles(CancellationToken cancellationToken)
     {
         var scopedLogger = this.logger.CreateScopedLogger();
@@ -41,65 +52,56 @@ public sealed class MonitoringService(
             string.Join(", ", expandedPaths)
         );
 
+        var cts = CancellationTokenSource.CreateLinkedTokenSource(
+            cancellationToken,
+            this.cancellationTokenSource.Token
+        );
         foreach (var filePath in expandedPaths)
         {
-            var watcher = CreateWatcher(filePath);
-            if (watcher is null)
-            {
-                scopedLogger.LogWarning("Failed to watch file: {file}", filePath);
-                continue;
-            }
-
-            this.watchers.Add(watcher);
-            scopedLogger.LogDebug("Watching file: {file}", filePath);
+            this.watchers.Add(Task.Run(() => this.MonitorFile(filePath, cts.Token), cts.Token));
         }
     }
 
-    private FileSystemWatcher? CreateWatcher(string filePath)
-    {
-        var scpoedLogger = this.logger.CreateScopedLogger();
-        var directory = Path.GetDirectoryName(filePath);
-        var fileName = Path.GetFileName(filePath);
-        if (directory is null || fileName is null)
-        {
-            scpoedLogger.LogWarning("Invalid file path: {filePath}", filePath);
-            return default;
-        }
-
-        var watcher = new FileSystemWatcher(directory, fileName)
-        {
-            NotifyFilter = NotifyFilters.LastWrite | NotifyFilters.Size,
-            EnableRaisingEvents = true,
-        };
-
-        watcher.Changed += this.OnFileChanged;
-        watcher.Error += this.OnWatcherError;
-        return watcher;
-    }
-
-    private void OnFileChanged(object sender, FileSystemEventArgs e)
+    private async Task MonitorFile(string filePath, CancellationToken cancellationToken)
     {
         var scopedLogger = this.logger.CreateScopedLogger();
-        scopedLogger.LogDebug("File changed: {file}", e.FullPath);
-        // TODO:: Read changes and push to Elastic
-    }
-
-    private void OnWatcherError(object sender, ErrorEventArgs e)
-    {
-        var scopedLogger = this.logger.CreateScopedLogger();
-        scopedLogger.LogError(e.GetException(), "Watcher error");
-    }
-
-    public void Dispose()
-    {
-        foreach (var watcher in this.watchers)
+        if (!File.Exists(filePath))
         {
-            watcher.Changed -= this.OnFileChanged;
-            watcher.Error -= this.OnWatcherError;
-            watcher.Dispose();
+            scopedLogger.LogWarning("File does not exist: {filePath}", filePath);
+            return;
         }
 
-        this.watchers.Clear();
+        scopedLogger.LogInformation("Starting to monitor file: {filePath}", filePath);
+        var previousPosition = this.persistenceService.GetPosition(filePath);
+        using var fs = new FileStream(
+            filePath,
+            FileMode.Open,
+            FileAccess.Read,
+            FileShare.ReadWrite | FileShare.Delete
+        );
+
+        if (fs.Length < previousPosition)
+        {
+            previousPosition = 0;
+        }
+
+        if (previousPosition > 0)
+        {
+            fs.Seek(previousPosition, SeekOrigin.Begin);
+        }
+
+        using var reader = new StreamReader(fs);
+        var line = string.Empty;
+        while ((line = await reader.ReadLineAsync(cancellationToken))?.IsNullOrEmpty() is false)
+        {
+            scopedLogger.LogDebug("Read line: {line}", line);
+            //TODO: Publish to IPublishingService
+            await this.persistenceService.SavePositionAsync(
+                filePath,
+                fs.Position,
+                cancellationToken
+            );
+        }
     }
 
     private static List<string> ExpandPaths(List<string>? patterns)

robin/Services/Persistence/PositionPersistenceService.cs

@@ -0,0 +1,64 @@
+using System.Collections.Concurrent;
+using System.Core.Extensions;
+using System.Text.Json;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using Robin.Options;
+using Robin.Serialization;
+
+namespace Robin.Services.Persistence;
+
+public sealed class PositionPersistenceService
+{
+    private readonly PositionPersistenceOptions options;
+    private readonly ConcurrentDictionary<string, long> positions = new();
+
+    public PositionPersistenceService(
+        IOptions<PositionPersistenceOptions> options,
+        ILogger<PositionPersistenceService> logger
+    )
+    {
+        this.options = options.Value;
+        this.options.FilePath.ThrowIfNull();
+
+        try
+        {
+            if (File.Exists(this.options.FilePath))
+            {
+                this.positions =
+                    JsonSerializer.Deserialize(
+                        File.ReadAllText(this.options.FilePath),
+                        SerializationContext.Default.ConcurrentDictionaryStringInt64
+                    ) ?? new ConcurrentDictionary<string, long>();
+            }
+        }
+        catch (Exception ex)
+        {
+            logger.LogError(ex, "Failed to load positions from file.");
+        }
+
+        this.positions ??= [];
+    }
+
+    public async ValueTask SavePositionAsync(
+        string source,
+        long position,
+        CancellationToken cancellationToken
+    )
+    {
+        this.positions[source] = position;
+        await File.WriteAllTextAsync(
+            this.options.FilePath.ThrowIfNull(),
+            JsonSerializer.Serialize(
+                this.positions,
+                SerializationContext.Default.ConcurrentDictionaryStringInt64
+            ),
+            cancellationToken
+        );
+    }
+
+    public long GetPosition(string source)
+    {
+        return this.positions.GetOrAdd(source, 0);
+    }
+}

robin/Services/Publishing/ElasticPublishingService.cs

@@ -0,0 +1,35 @@
+using System.Extensions;
+using System.Extensions.Core;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using Nest;
+
+namespace Robin.Services.Publishing;
+
+public sealed class ElasticPublishingService(
+    IElasticClient elasticClient,
+    IOptions<ElasticOptions> options,
+    ILogger<ElasticPublishingService> logger
+) : IPublishingService
+{
+    private readonly IElasticClient elasticClient = elasticClient;
+    private readonly ElasticOptions options = options.Value;
+    private readonly ILogger<ElasticPublishingService> logger = logger;
+
+    public async Task<bool> PublishAsync<T>(T item, CancellationToken cancellationToken)
+        where T : class
+    {
+        var response = await this.elasticClient.IndexDocumentAsync(item, cancellationToken);
+        if (response.IsValid)
+        {
+            return true;
+        }
+
+        var scopedLogger = this.logger.CreateScopedLogger();
+        scopedLogger.LogError(
+            "Failed to publish document to Elasticsearch. Error: {Error}",
+            response.DebugInformation
+        );
+        return false;
+    }
+}

robin/Services/Publishing/IPublishingService.cs

@@ -0,0 +1,7 @@
+namespace Robin.Services.Publishing;
+
+public interface IPublishingService
+{
+    Task<bool> PublishAsync<T>(T item, CancellationToken cancellationToken)
+        where T : class;
+}

robin/appsettings.json

@@ -8,5 +8,8 @@
         "Robin.Services": "Debug"
       }
     }
+  },
+  "PositionPersistence": {
+    "FilePath": "positions.json"
   }
 }