feat(S2-b): connect durable MQTT sync to capture workflow

src/FieldLogger.Sync/OutboundStore.cs

@@ -0,0 +1,159 @@
+using SQLite;
+
+namespace FieldLogger.Sync;
+
+/// <summary>Lifecycle of a queued outbound record.</summary>
+public enum OutboundStatus
+{
+    /// <summary>Ready to publish.</summary>
+    Pending = 0,
+    /// <summary>Published and waiting for an application acknowledgement.</summary>
+    InFlight = 1,
+    /// <summary>Terminally rejected by the cloud (schema/validation/authz) — not retried; surfaced for LOG-7.</summary>
+    RejectedTerminal = 2,
+}
+
+/// <summary>
+/// A durably-queued outbound point. Survives app restart (persisted in SQLite) and is
+/// released only when the cloud application-level ack references its <see cref="PointId"/>
+/// (SRS-SYN-2/7). Broker PUBACK alone does NOT release it.
+/// </summary>
+[Table("outbound")]
+public sealed class OutboundMessage
+{
+    [PrimaryKey, AutoIncrement] public int Id { get; set; }
+
+    /// <summary>UUIDv7 idempotency key. Unique — a re-enqueue of the same point is a no-op.</summary>
+    [Indexed(Name = "ux_pointid", Order = 1, Unique = true)]
+    public string PointId { get; set; } = "";
+
+    public string Topic { get; set; } = "";
+    public string PayloadJson { get; set; } = "";
+    public string SchemaVersion { get; set; } = "1";
+
+    public long EnqueuedUnixMs { get; set; }
+    public int AttemptCount { get; set; }
+    public long NextAttemptUnixMs { get; set; }
+    public int Status { get; set; } = (int)OutboundStatus.Pending;
+    public string? LastReason { get; set; }
+
+    public static OutboundMessage FromPoint(
+        PointRecord point,
+        string topic,
+        DateTimeOffset enqueuedAt,
+        string? jobId = null,
+        string? ticket = null)
+    {
+        var payload = point.ToAppLogPayloadUtf8(jobId, ticket);
+        return new OutboundMessage
+        {
+            PointId = point.PointId,
+            Topic = topic,
+            PayloadJson = System.Text.Encoding.UTF8.GetString(payload),
+            SchemaVersion = "1",
+            EnqueuedUnixMs = enqueuedAt.ToUnixTimeMilliseconds(),
+            NextAttemptUnixMs = enqueuedAt.ToUnixTimeMilliseconds(),
+            Status = (int)OutboundStatus.Pending,
+        };
+    }
+}
+
+/// <summary>Durable outbound-queue store. Implementations must survive process restart.</summary>
+public interface IOutboundStore
+{
+    Task InitAsync();
+    /// <summary>Enqueue idempotently; returns false if a row with this pointId already exists.</summary>
+    Task<bool> EnqueueAsync(OutboundMessage message);
+    /// <summary>Pending rows whose next-attempt time has arrived, oldest first.</summary>
+    Task<IReadOnlyList<OutboundMessage>> DequeueReadyAsync(DateTimeOffset now, int max = 50);
+    /// <summary>Release a record — accepted by the cloud. Removes it from the queue.</summary>
+    Task ReleaseAckedAsync(string pointId);
+    /// <summary>Record a retryable failure: bump attempt count, schedule the next attempt.</summary>
+    Task RescheduleAsync(string pointId, DateTimeOffset nextAttempt, string reason);
+    /// <summary>Mark a record published-and-awaiting-ack: reschedule a republish (idempotent via
+    /// UUID) after the ack window WITHOUT counting it as a failed attempt.</summary>
+    Task TouchAwaitingAckAsync(string pointId, DateTimeOffset nextAttempt);
+    /// <summary>Terminal rejection (LOG-7): keep the row for surfacing, mark it not-retryable.</summary>
+    Task MarkRejectedAsync(string pointId, string reason);
+    Task<int> PendingCountAsync();
+    Task<IReadOnlyList<OutboundMessage>> GetRejectedAsync();
+}
+
+/// <summary>
+/// SQLite-backed <see cref="IOutboundStore"/>. The DB path is injected (not taken from a
+/// platform API) so it is testable headless with a temp file; reopening the same path is a
+/// process restart. Idempotent enqueue relies on the unique index on pointId.
+/// </summary>
+public sealed class SqliteOutboundStore : IOutboundStore
+{
+    private readonly SQLiteAsyncConnection _db;
+
+    public SqliteOutboundStore(string dbPath)
+    {
+        _db = new SQLiteAsyncConnection(dbPath,
+            SQLiteOpenFlags.ReadWrite | SQLiteOpenFlags.Create | SQLiteOpenFlags.SharedCache);
+    }
+
+    public async Task InitAsync()
+    {
+        await _db.CreateTableAsync<OutboundMessage>();
+        // A process crash can leave rows in-flight after the broker accepted them but before
+        // the application ack was applied. Requeue them; pointId makes the replay idempotent.
+        await _db.ExecuteAsync(
+            "UPDATE outbound SET Status = ?, NextAttemptUnixMs = ? WHERE Status = ?",
+            (int)OutboundStatus.Pending, DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
+            (int)OutboundStatus.InFlight);
+    }
+
+    public async Task<bool> EnqueueAsync(OutboundMessage message)
+    {
+        try
+        {
+            await _db.InsertAsync(message);
+            return true;
+        }
+        catch (SQLiteException)
+        {
+            // Unique-index violation on pointId → already queued/known. Idempotent no-op.
+            return false;
+        }
+    }
+
+    public async Task<IReadOnlyList<OutboundMessage>> DequeueReadyAsync(DateTimeOffset now, int max = 50)
+    {
+        long nowMs = now.ToUnixTimeMilliseconds();
+        return await _db.Table<OutboundMessage>()
+            .Where(m => (m.Status == (int)OutboundStatus.Pending || m.Status == (int)OutboundStatus.InFlight)
+                && m.NextAttemptUnixMs <= nowMs)
+            .OrderBy(m => m.EnqueuedUnixMs)
+            .Take(max)
+            .ToListAsync();
+    }
+
+    public Task ReleaseAckedAsync(string pointId) =>
+        _db.ExecuteAsync("DELETE FROM outbound WHERE PointId = ?", pointId);
+
+    public Task RescheduleAsync(string pointId, DateTimeOffset nextAttempt, string reason) =>
+        _db.ExecuteAsync(
+            "UPDATE outbound SET AttemptCount = AttemptCount + 1, NextAttemptUnixMs = ?, LastReason = ? WHERE PointId = ? AND Status = ?",
+            nextAttempt.ToUnixTimeMilliseconds(), reason, pointId, (int)OutboundStatus.Pending);
+
+    public Task TouchAwaitingAckAsync(string pointId, DateTimeOffset nextAttempt) =>
+        _db.ExecuteAsync(
+            "UPDATE outbound SET Status = ?, AttemptCount = AttemptCount + 1, NextAttemptUnixMs = ? WHERE PointId = ? AND Status != ?",
+            (int)OutboundStatus.InFlight, nextAttempt.ToUnixTimeMilliseconds(), pointId,
+            (int)OutboundStatus.RejectedTerminal);
+
+    public Task MarkRejectedAsync(string pointId, string reason) =>
+        _db.ExecuteAsync(
+            "UPDATE outbound SET Status = ?, LastReason = ? WHERE PointId = ?",
+            (int)OutboundStatus.RejectedTerminal, reason, pointId);
+
+    public async Task<int> PendingCountAsync() =>
+        await _db.Table<OutboundMessage>()
+            .Where(m => m.Status == (int)OutboundStatus.Pending || m.Status == (int)OutboundStatus.InFlight)
+            .CountAsync();
+
+    public async Task<IReadOnlyList<OutboundMessage>> GetRejectedAsync() =>
+        await _db.Table<OutboundMessage>().Where(m => m.Status == (int)OutboundStatus.RejectedTerminal).ToListAsync();
+}