ulapp/src/FieldLogger.Sync/OutboundStore.cs

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 for immediate replay; pointId makes
        // the replay idempotent. A zero due-time also keeps recovery independent of whichever
        // clock implementation the sync engine uses.
        await _db.ExecuteAsync(
            "UPDATE outbound SET Status = ?, NextAttemptUnixMs = ? WHERE Status = ?",
            (int)OutboundStatus.Pending, 0,
            (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();
}