MeterVault/src/Infrastructure/Ingestion/IngestionService.cs

using MeterVault.Core.Domain;
using MeterVault.Infrastructure.Persistence;
using Microsoft.EntityFrameworkCore;

namespace MeterVault.Infrastructure.Ingestion;

/// <summary>The outcome of ingesting one reading.</summary>
public enum IngestionOutcome
{
    Written,
    Updated,
    RejectedDecrease,
    UnknownSource,
}

/// <summary>
/// Persists a single incoming reading (SDD §6.1, FR-4): applies the source's scale/offset, is
/// idempotent on (meter_id, time), and guards monotonic registers against spurious decreases
/// unless an active reset/swap event explains them. Updates the source's last-seen status.
/// Consumption normalization is recomputed separately (batch/scheduled), not per message.
/// </summary>
public sealed class IngestionService(
    MeterVaultDbContext db, Normalization.NormalizationService normalization)
{
    private static readonly HashSet<MeterMode> MonotonicModes =
        [MeterMode.CumulativeCounter, MeterMode.GenerationCounter, MeterMode.RuntimeCounter];

    private readonly MeterVaultDbContext _db = db;
    private readonly Normalization.NormalizationService _normalization = normalization;

    /// <summary>Ingests through a configured source (MQTT/HA workers): applies scale/offset and updates source status.</summary>
    public async Task<IngestionOutcome> IngestAsync(
        int sourceId, DateTimeOffset time, double rawValue, CancellationToken cancellationToken = default)
    {
        var source = await _db.MeterSources
            .FirstOrDefaultAsync(s => s.Id == sourceId, cancellationToken).ConfigureAwait(false);
        if (source is null)
        {
            return IngestionOutcome.UnknownSource;
        }

        var meter = await _db.Meters
            .FirstOrDefaultAsync(m => m.Id == source.MeterId, cancellationToken).ConfigureAwait(false);
        if (meter is null)
        {
            return IngestionOutcome.UnknownSource;
        }

        var value = (rawValue * source.Scale) + source.Offset;
        var utc = time.ToUniversalTime();

        if (MonotonicModes.Contains(meter.Mode)
            && await IsSpuriousDecreaseAsync(meter.Id, utc, value, cancellationToken).ConfigureAwait(false))
        {
            await UpdateSourceStatusAsync(source, utc, value, "rejected: decrease", cancellationToken).ConfigureAwait(false);
            return IngestionOutcome.RejectedDecrease;
        }

        var outcome = await UpsertAsync(meter, utc, value, source.Id, quality: null, ReadingFlags.None, cancellationToken).ConfigureAwait(false);
        await UpdateSourceStatusAsync(source, utc, value, "ok", cancellationToken).ConfigureAwait(false);
        await RenormalizeAsync(meter.Id, outcome, cancellationToken).ConfigureAwait(false);
        return outcome;
    }

    /// <summary>Ingests directly against a meter (REST push, or a hand-entered reading from the UI).</summary>
    /// <param name="renormalize">
    /// False to skip deriving consumption, for callers ingesting a batch into one meter: recomputing
    /// rewrites the meter's entire series, so doing it per reading is quadratic in batch size. Such a
    /// caller must recompute the affected meters itself once the batch is in.
    /// </param>
    /// <param name="quality">
    /// Provenance to stamp on the row. Null keeps the default for a new row and leaves an existing
    /// row's quality alone — a source re-reporting a timestamp must not silently relabel a reading
    /// somebody entered by hand or that came from an import.
    /// </param>
    /// <param name="flags">
    /// Annotations to add to the row, e.g. <see cref="ReadingFlags.MeterSwap"/> on the new register's
    /// start value written together with a swap. Added to an existing row's flags, never cleared.
    /// </param>
    public async Task<IngestionOutcome> IngestByMeterAsync(
        int meterId, DateTimeOffset time, double value, bool renormalize = true,
        ReadingQuality? quality = null, ReadingFlags flags = ReadingFlags.None,
        CancellationToken cancellationToken = default)
    {
        var meter = await _db.Meters
            .FirstOrDefaultAsync(m => m.Id == meterId, cancellationToken).ConfigureAwait(false);
        if (meter is null)
        {
            return IngestionOutcome.UnknownSource;
        }

        var utc = time.ToUniversalTime();
        if (MonotonicModes.Contains(meter.Mode)
            && await IsSpuriousDecreaseAsync(meter.Id, utc, value, cancellationToken).ConfigureAwait(false))
        {
            return IngestionOutcome.RejectedDecrease;
        }

        var outcome = await UpsertAsync(meter, utc, value, sourceId: null, quality, flags, cancellationToken).ConfigureAwait(false);
        await _db.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
        if (renormalize)
        {
            await RenormalizeAsync(meter.Id, outcome, cancellationToken).ConfigureAwait(false);
        }

        return outcome;
    }

    /// <summary>
    /// Derives consumption for one meter after a batch of readings has been written. The public
    /// counterpart to skipping <c>renormalize</c> on each individual ingest.
    /// </summary>
    public Task RenormalizeMeterAsync(int meterId, CancellationToken cancellationToken = default) =>
        RecomputeAtomicallyAsync(meterId, cancellationToken);

    /// <summary>
    /// Derives consumption from the reading just written. Without this a live-ingested reading sits
    /// in <c>reading</c> forever and every derived figure — consumption, generation, cost — stays
    /// frozen at the last import, because nothing else recomputes that meter.
    /// </summary>
    /// <remarks>
    /// Recomputes inline rather than on a debounce. <see cref="Normalization.NormalizationService"/>
    /// rewrites a meter's whole consumption series, which is cheap at metering cadence — HA polls
    /// hourly — but would be wasteful under a chatty MQTT source publishing every few seconds. If
    /// such a source is ever added, batch this behind a dirty-set worker rather than making the
    /// normalizer incremental: consumption being a pure function of readings + events is what makes
    /// it reproducible.
    /// </remarks>
    private async Task RenormalizeAsync(
        int meterId, IngestionOutcome outcome, CancellationToken cancellationToken)
    {
        // A rejected decrease changed nothing, so the existing series is still correct.
        if (outcome is not (IngestionOutcome.Written or IngestionOutcome.Updated))
        {
            return;
        }

        // The reading must already be persisted: RecomputeMeterAsync re-reads the meter's readings
        // from the database, so anything still pending in the change tracker would be missed.
        await RecomputeAtomicallyAsync(meterId, cancellationToken).ConfigureAwait(false);
    }

    /// <summary>
    /// Rebuilds a meter's consumption series as one atomic unit.
    /// </summary>
    /// <remarks>
    /// <see cref="Normalization.NormalizationService.RecomputeMeterAsync"/> clears the series with
    /// <c>ExecuteDelete</c>, which commits on its own when no transaction is ambient, and only then
    /// adds the rebuilt rows. Without a transaction around both halves the meter has *no*
    /// consumption in between: a dashboard read in that window reports zero, and a crash or a
    /// cancelled request makes the loss permanent — for data the SDD treats as the long-term source
    /// of truth (§5.5). Import and the events API already wrap their recomputes this way; live
    /// ingestion was the path that did not.
    ///
    /// Respects an ambient transaction rather than nesting, so callers that already opened one keep
    /// a single unit of work.
    /// </remarks>
    private async Task RecomputeAtomicallyAsync(int meterId, CancellationToken cancellationToken)
    {
        if (_db.Database.CurrentTransaction is not null)
        {
            await _normalization.RecomputeMeterAsync(meterId, batchId: null, cancellationToken).ConfigureAwait(false);
            await _db.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
            return;
        }

        await using var tx = await _db.Database.BeginTransactionAsync(cancellationToken).ConfigureAwait(false);
        await _normalization.RecomputeMeterAsync(meterId, batchId: null, cancellationToken).ConfigureAwait(false);
        await _db.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
        await tx.CommitAsync(cancellationToken).ConfigureAwait(false);
    }

    private async Task<IngestionOutcome> UpsertAsync(
        Meter meter, DateTimeOffset utc, double value, int? sourceId, ReadingQuality? quality,
        ReadingFlags flags, CancellationToken cancellationToken)
    {
        var existing = _db.Readings.Local.FirstOrDefault(r => r.MeterId == meter.Id && r.Time == utc)
            ?? await _db.Readings.FirstOrDefaultAsync(r => r.MeterId == meter.Id && r.Time == utc, cancellationToken).ConfigureAwait(false);

        if (existing is null)
        {
            _db.Readings.Add(new Reading
            {
                MeterId = meter.Id,
                Time = utc,
                Value = value,
                SourceId = sourceId,
                Quality = quality ?? ReadingQuality.Measured,
                Flags = flags,
            });
            return IngestionOutcome.Written;
        }

        existing.Value = value;
        existing.SourceId = sourceId ?? existing.SourceId;
        existing.Flags |= flags;
        if (quality is { } stamp)
        {
            existing.Quality = stamp;
        }

        // An imported month row stamped at this instant describes the end of its month. Correcting it by
        // hand or by re-import keeps that meaning; a live value, or the start reading of a swap, is what
        // the register showed at this very instant, so the row stops being a month row.
        if (existing.Flags.HasFlag(ReadingFlags.MonthLabel)
            && (quality is not (ReadingQuality.Manual or ReadingQuality.Imported)
                || (flags & (ReadingFlags.MeterSwap | ReadingFlags.CounterReset)) != 0))
        {
            existing.Flags &= ~ReadingFlags.MonthLabel;
            existing.Quality = quality ?? ReadingQuality.Measured;
        }

        return IngestionOutcome.Updated;
    }

    private async Task<bool> IsSpuriousDecreaseAsync(
        int meterId, DateTimeOffset time, double value, CancellationToken cancellationToken)
    {
        // The previous reading on the normalizer's timeline, not by stamp: a month row stamped on the 1st
        // describes the end of its month and must not reject a lower reading taken during that month.
        var neighbours = await RegisterNeighbours.FindAsync(_db, meterId, time, _normalization.TimeZone, cancellationToken)
            .ConfigureAwait(false);

        if (neighbours.Previous is not { } previous || value >= previous.Reading.Value)
        {
            return false;
        }

        // Only a reset/swap in the window (previousReading, thisReading] explains the decrease —
        // an old historical reset must not permanently disable the guard.
        return !neighbours.BoundaryAfterPreviousUpTo(time);
    }

    private async Task UpdateSourceStatusAsync(
        MeterSource source, DateTimeOffset time, double value, string status, CancellationToken cancellationToken)
    {
        source.LastSeenAt = time;
        source.LastValue = value;
        source.LastStatus = status;
        await _db.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
    }
}