MoShiro/backend/src/MoShiro.Backend.Application/Ingest/IngestService.cs

using Fleet.Core;
using Microsoft.Extensions.Logging;
using MoShiro.Backend.Analysis;
using MoShiro.Protocol;

namespace MoShiro.Backend.Application;

/// <summary>
/// <see cref="Stored"/> 和 <see cref="Duplicate"/> 都会让节点把消息从发件箱里移除。
/// <see cref="Discarded"/> 表示这条消息永远不可能被接受（损坏、冒名、未知 Kind、时间离谱），
/// 同样确认掉，否则节点会无限重发——原因记入日志和事件。
/// </summary>
public enum IngestOutcome
{
    Stored,
    Duplicate,
    Discarded,
}

/// <summary>
/// 原始块入库：校验 → 追加到段文件并写索引 → 流式路径（1s 热层 + 10s 层）→ 登记分钟窗口等待持久路径计算。
/// 持久化成功之后才返回，Fleet 随后才会 ACK。
/// </summary>
public sealed partial class IngestService
{
    private const long MaxFutureSkewMs = 3_600_000;
    private const int MaxDurationMs = 600_000;

    private readonly IRawStore _raw;
    private readonly IMetricStore _metrics;
    private readonly HotLayer _hot;
    private readonly LiveHub _live;
    private readonly BackendCounters _counters;
    private readonly IEventStore _events;
    private readonly BackendOptions _options;
    private readonly TimeProvider _time;
    private readonly ILogger<IngestService> _logger;

    public IngestService(
        IRawStore raw,
        IMetricStore metrics,
        HotLayer hot,
        LiveHub live,
        BackendCounters counters,
        IEventStore events,
        BackendOptions options,
        TimeProvider time,
        ILogger<IngestService> logger)
    {
        _raw = raw;
        _metrics = metrics;
        _hot = hot;
        _live = live;
        _counters = counters;
        _events = events;
        _options = options;
        _time = time;
        _logger = logger;
    }

    public async ValueTask<IngestOutcome> IngestAsync(NodeId agent, Guid msgId, ReadOnlyMemory<byte> bytes, CancellationToken ct)
    {
        Result<RawChunk> decoded = RawChunkCodec.Decode(bytes.Span);
        if (!decoded.TryGet(out RawChunk? chunk, out FleetError? error))
        {
            return await DiscardAsync(agent, "undecodable chunk: " + error.Message, ct);
        }

        RawChunkHeader header = chunk.Header;
        long now = _time.GetUtcNow().ToUnixTimeMilliseconds();
        string? problem =
            !IpVersions.IsValid(header.IpVersion) ? "unknown ip_version"
            : header.AgentId != agent ? "agent_id does not match the authenticated node"
            : header.ChunkId.Value != msgId ? "chunk_id does not match the message id"
            : Kinds.Find(header.Kind) is null ? $"unknown kind {header.Kind}"
            : header.DurationMs is <= 0 or > MaxDurationMs ? "duration out of range"
            : header.StartUtcMs > now + MaxFutureSkewMs ? "chunk starts in the future"
            : header.StartUtcMs < now - (long)TimeSpan.FromDays(_options.TestRawRetentionDays).TotalMilliseconds ? "chunk is older than the retention"
            : null;
        if (problem is not null)
        {
            return await DiscardAsync(agent, problem, ct);
        }

        string series = header.Series;
        var meta = new RawChunkMeta(
            header.ChunkId, agent, series, header.Kind, header.TaskId, header.TestId, header.Phase,
            header.StartUtcMs, header.DurationMs, header.Partial, "", 0, bytes.Length);
        bool stored = await _raw.AppendAsync(new RawChunkAppend(meta, bytes, DateOnly.FromDateTime(DateTimeOffset.FromUnixTimeMilliseconds(header.StartUtcMs).UtcDateTime).ToString("yyyy-MM-dd")), ct);
        if (!stored)
        {
            _counters.CountDuplicate();
            return IngestOutcome.Duplicate;
        }

        _counters.CountIngested(bytes.Length);
        if (header.TestId is { } testId)
        {
            // 测试数据不进入持续线路的指标
            _live.Publish(new TestSignal(testId));
            return IngestOutcome.Stored;
        }

        await FeedAnalysisAsync(agent, series, chunk, now, ct);
        _live.Publish(new ChunkSignal(agent, series));
        return IngestOutcome.Stored;
    }

    private async ValueTask FeedAnalysisAsync(NodeId agent, string series, RawChunk chunk, long now, CancellationToken ct)
    {
        _hot.Merge(agent, series, chunk);

        var tenSeconds = new SortedDictionary<long, MetricSet>();
        WindowEvaluator.Default.Evaluate(chunk, MetricLevel.S10.WidthMs(), tenSeconds);
        var rows = new List<MetricRow>(tenSeconds.Count);
        foreach ((long start, MetricSet set) in tenSeconds)
        {
            rows.Add(new MetricRow(agent, series, start, chunk.Header.Partial, AccumulatorCodec.Encode(set)));
        }

        await _metrics.MergeAsync(MetricLevel.S10, rows, ct);

        // 持久路径：块覆盖到的每个分钟窗口，等窗口结束再加宽限时间之后计算；迟到的块立即重算
        long minute = MetricLevel.M1.WidthMs();
        long first = Math.DivRem(chunk.Header.StartUtcMs, minute, out _) * minute;
        long last = Math.DivRem(chunk.Header.StartUtcMs + chunk.Header.DurationMs, minute, out _) * minute;
        var queue = new List<QueueItem>();
        for (long window = first; window <= last; window += minute)
        {
            long ready = window + minute + (long)_options.WindowGrace.TotalMilliseconds;
            queue.Add(new QueueItem(agent, series, window, Math.Max(now, ready), ready <= now ? "late_chunk" : "new"));
        }

        await _metrics.EnqueueAsync(queue, ct);
    }

    private async ValueTask<IngestOutcome> DiscardAsync(NodeId agent, string reason, CancellationToken ct)
    {
        _counters.CountDiscarded();
        LogDiscarded(agent.Value, reason);
        await _events.AddAsync(agent, "chunk_rejected", _time.GetUtcNow().ToUnixTimeMilliseconds(), reason, ct);
        return IngestOutcome.Discarded;
    }

    [LoggerMessage(Level = LogLevel.Warning, Message = "chunk discarded agent={Agent} err={Err}")]
    private partial void LogDiscarded(string agent, string err);
}