using System; using System.Collections.Generic; using System.Data.Common; using System.Text.Json; using System.Threading; using System.Threading.Tasks; using Dapper; using Junction.Domain; using Junction.Domain.Models; using Junction.Domain.Persistence; namespace Junction.Persistence { /// /// Dapper + SQLite adapter for . /// /// /// Provider-swap seam: all SQLite/Dapper/ADO.NET types stay inside this class. Only /// types and BCL primitives cross the boundary. /// /// Column mapping (see DDL): /// /// <-> TEXT via ToString("D") (canonical 8-4-4-4-12). /// <-> INTEGER ticks (). /// <-> TEXT JSON of a Dictionary<string,string> (System.Text.Json). Null/blank JSON rehydrates to an empty dictionary. /// <-> TEXT ISO 8601 round-trip ("O"). /// <-> INTEGER (enum cast). /// /// /// Not-found policy: and return /// Result.Fail carrying an with code "not_found" — never throw, /// never a success with a null value. /// /// Errors: expected failures are reported via /. Store /// exceptions are caught and mapped to Result.Fail; cancellation is mapped to Result.Cancelled. /// public sealed class SqliteMachineRepository : IMachineRepository { private const string Source = "SqliteMachineRepository"; private const string NotFoundCode = "not_found"; private static readonly JsonSerializerOptions JsonOptions = new JsonSerializerOptions(); private readonly IConnectionFactory _connectionFactory; public SqliteMachineRepository(IConnectionFactory connectionFactory) { _connectionFactory = connectionFactory ?? throw new ArgumentNullException(nameof(connectionFactory)); } public async Task>> GetAllAsync(CancellationToken cancellationToken) { if (cancellationToken.IsCancellationRequested) { return Result>.Cancelled(); } try { using (var connection = OpenConnection()) { var command = new CommandDefinition( "SELECT Id, Name, ProtocolId, PollIntervalTicks, ConnectionConfigJson FROM machines;", cancellationToken: cancellationToken); var rows = await connection.QueryAsync(command).ConfigureAwait(false); var machines = new List(); foreach (var row in rows) { machines.Add(MapMachine(row)); } return Result>.Ok(machines); } } catch (OperationCanceledException) { return Result>.Cancelled(); } catch (Exception ex) { return Result>.Fail( OperationError.Of(Source, "GetAll failed: " + ex.Message)); } } public async Task> GetByIdAsync(Guid id, CancellationToken cancellationToken) { if (cancellationToken.IsCancellationRequested) { return Result.Cancelled(); } try { using (var connection = OpenConnection()) { var command = new CommandDefinition( "SELECT Id, Name, ProtocolId, PollIntervalTicks, ConnectionConfigJson " + "FROM machines WHERE Id = @Id;", new { Id = GuidText(id) }, cancellationToken: cancellationToken); var row = await connection.QuerySingleOrDefaultAsync(command).ConfigureAwait(false); if (row == null) { return Result.Fail(new OperationError( NotFoundCode, Source, "Machine not found: " + GuidText(id))); } return Result.Ok(MapMachine(row)); } } catch (OperationCanceledException) { return Result.Cancelled(); } catch (Exception ex) { return Result.Fail( OperationError.Of(Source, "GetById failed: " + ex.Message)); } } public async Task UpsertAsync(Machine machine, CancellationToken cancellationToken) { if (machine == null) { return Result.Fail(OperationError.Of(Source, "Machine is null.")); } if (cancellationToken.IsCancellationRequested) { return Result.Cancelled(); } try { using (var connection = OpenConnection()) { var command = new CommandDefinition( "INSERT INTO machines (Id, Name, ProtocolId, PollIntervalTicks, ConnectionConfigJson) " + "VALUES (@Id, @Name, @ProtocolId, @PollIntervalTicks, @ConnectionConfigJson) " + "ON CONFLICT(Id) DO UPDATE SET " + "Name = excluded.Name, " + "ProtocolId = excluded.ProtocolId, " + "PollIntervalTicks = excluded.PollIntervalTicks, " + "ConnectionConfigJson = excluded.ConnectionConfigJson;", new { Id = GuidText(machine.Id), machine.Name, machine.ProtocolId, PollIntervalTicks = machine.PollInterval.Ticks, ConnectionConfigJson = SerializeConfig(machine.ConnectionConfig) }, cancellationToken: cancellationToken); await connection.ExecuteAsync(command).ConfigureAwait(false); return Result.Ok(); } } catch (OperationCanceledException) { return Result.Cancelled(); } catch (Exception ex) { return Result.Fail(OperationError.Of(Source, "Upsert failed: " + ex.Message)); } } public async Task DeleteAsync(Guid id, CancellationToken cancellationToken) { if (cancellationToken.IsCancellationRequested) { return Result.Cancelled(); } try { using (var connection = OpenConnection()) using (var transaction = connection.BeginTransaction()) { var idParam = new { Id = GuidText(id) }; await connection.ExecuteAsync(new CommandDefinition( "DELETE FROM snapshot_items WHERE MachineId = @Id;", idParam, transaction, cancellationToken: cancellationToken)).ConfigureAwait(false); await connection.ExecuteAsync(new CommandDefinition( "DELETE FROM latest_snapshots WHERE MachineId = @Id;", idParam, transaction, cancellationToken: cancellationToken)).ConfigureAwait(false); await connection.ExecuteAsync(new CommandDefinition( "DELETE FROM machines WHERE Id = @Id;", idParam, transaction, cancellationToken: cancellationToken)).ConfigureAwait(false); transaction.Commit(); return Result.Ok(); } } catch (OperationCanceledException) { return Result.Cancelled(); } catch (Exception ex) { return Result.Fail(OperationError.Of(Source, "Delete failed: " + ex.Message)); } } public async Task SaveSnapshotAsync(MachineSnapshot snapshot, CancellationToken cancellationToken) { if (snapshot == null) { return Result.Fail(OperationError.Of(Source, "Snapshot is null.")); } if (cancellationToken.IsCancellationRequested) { return Result.Cancelled(); } try { using (var connection = OpenConnection()) using (var transaction = connection.BeginTransaction()) { var machineIdText = GuidText(snapshot.MachineId); // Upsert the single latest-snapshot header row. await connection.ExecuteAsync(new CommandDefinition( "INSERT INTO latest_snapshots (MachineId, CapturedAt, ConnectionState) " + "VALUES (@MachineId, @CapturedAt, @ConnectionState) " + "ON CONFLICT(MachineId) DO UPDATE SET " + "CapturedAt = excluded.CapturedAt, " + "ConnectionState = excluded.ConnectionState;", new { MachineId = machineIdText, CapturedAt = IsoText(snapshot.CapturedAt), ConnectionState = (int)snapshot.ConnectionState }, transaction, cancellationToken: cancellationToken)).ConfigureAwait(false); // Only-latest policy: drop previous items, insert current set. await connection.ExecuteAsync(new CommandDefinition( "DELETE FROM snapshot_items WHERE MachineId = @MachineId;", new { MachineId = machineIdText }, transaction, cancellationToken: cancellationToken)).ConfigureAwait(false); if (snapshot.Items.Count > 0) { var itemParams = new List(snapshot.Items.Count); foreach (var item in snapshot.Items) { itemParams.Add(new { MachineId = machineIdText, ItemId = item.Id, item.Name, item.Value, item.Category, Timestamp = IsoText(item.Timestamp) }); } await connection.ExecuteAsync(new CommandDefinition( "INSERT INTO snapshot_items (MachineId, ItemId, Name, Value, Category, Timestamp) " + "VALUES (@MachineId, @ItemId, @Name, @Value, @Category, @Timestamp);", itemParams, transaction, cancellationToken: cancellationToken)).ConfigureAwait(false); } transaction.Commit(); return Result.Ok(); } } catch (OperationCanceledException) { return Result.Cancelled(); } catch (Exception ex) { return Result.Fail(OperationError.Of(Source, "SaveSnapshot failed: " + ex.Message)); } } public async Task> GetLatestSnapshotAsync(Guid machineId, CancellationToken cancellationToken) { if (cancellationToken.IsCancellationRequested) { return Result.Cancelled(); } try { using (var connection = OpenConnection()) { var machineIdText = GuidText(machineId); var header = await connection.QuerySingleOrDefaultAsync(new CommandDefinition( "SELECT MachineId, CapturedAt, ConnectionState FROM latest_snapshots WHERE MachineId = @MachineId;", new { MachineId = machineIdText }, cancellationToken: cancellationToken)).ConfigureAwait(false); if (header == null) { return Result.Fail(new OperationError( NotFoundCode, Source, "Snapshot not found for machine: " + machineIdText)); } var itemRows = await connection.QueryAsync(new CommandDefinition( "SELECT ItemId, Name, Value, Category, Timestamp FROM snapshot_items WHERE MachineId = @MachineId;", new { MachineId = machineIdText }, cancellationToken: cancellationToken)).ConfigureAwait(false); var items = new List(); foreach (var itemRow in itemRows) { items.Add(new DataItem( itemRow.ItemId, itemRow.Name, itemRow.Value, itemRow.Category, ParseIso(itemRow.Timestamp))); } var snapshot = new MachineSnapshot( machineId, ParseIso(header.CapturedAt), (ConnectionState)header.ConnectionState, items); return Result.Ok(snapshot); } } catch (OperationCanceledException) { return Result.Cancelled(); } catch (Exception ex) { return Result.Fail( OperationError.Of(Source, "GetLatestSnapshot failed: " + ex.Message)); } } // -- helpers ------------------------------------------------------------------------- private DbConnection OpenConnection() { // SqliteConnectionFactory yields a SqliteConnection (a DbConnection); Dapper's *Async // needs the DbConnection surface. The concrete Sqlite type never escapes this class. var connection = _connectionFactory.CreateOpenConnection(); if (connection is DbConnection dbConnection) { return dbConnection; } connection?.Dispose(); throw new InvalidOperationException( "IConnectionFactory must return a DbConnection for async Dapper operations."); } private static Machine MapMachine(MachineRow row) { return new Machine( Guid.Parse(row.Id), row.Name, row.ProtocolId, DeserializeConfig(row.ConnectionConfigJson), TimeSpan.FromTicks(row.PollIntervalTicks)); } private static string GuidText(Guid id) => id.ToString("D"); private static string IsoText(DateTimeOffset value) => value.ToString("O"); private static DateTimeOffset ParseIso(string value) => DateTimeOffset.Parse(value, null, System.Globalization.DateTimeStyles.RoundtripKind); private static string SerializeConfig(IReadOnlyDictionary config) { Dictionary dict; if (config is Dictionary concrete) { dict = concrete; } else { dict = new Dictionary(config.Count); foreach (var pair in config) { dict[pair.Key] = pair.Value; } } return JsonSerializer.Serialize(dict, JsonOptions); } private static IReadOnlyDictionary DeserializeConfig(string? json) { if (string.IsNullOrWhiteSpace(json)) { return new Dictionary(0); } var dict = JsonSerializer.Deserialize>(json!, JsonOptions); return dict ?? new Dictionary(0); } // -- row DTOs (private; never cross the boundary) ----------------------------------- private sealed class MachineRow { public string Id { get; set; } = ""; public string Name { get; set; } = ""; public string ProtocolId { get; set; } = ""; public long PollIntervalTicks { get; set; } public string? ConnectionConfigJson { get; set; } } private sealed class SnapshotRow { public string MachineId { get; set; } = ""; public string CapturedAt { get; set; } = ""; public int ConnectionState { get; set; } } private sealed class SnapshotItemRow { public string ItemId { get; set; } = ""; public string Name { get; set; } = ""; public string Value { get; set; } = ""; public string Category { get; set; } = ""; public string Timestamp { get; set; } = ""; } } }