MoShiro/backend/src/MoShiro.Backend.Api/Admin/DataEndpoints.cs

using System.Text;
using Fleet.Core;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Routing;
using MoShiro.Backend.Application;
using MoShiro.Protocol;

namespace MoShiro.Backend.Api;

/// <summary>原始数据检索与下载、事件、指标重算进度。</summary>
public static class DataEndpoints
{
    private const string Tag = "admin-data";

    public static void Map(RouteGroupBuilder group)
    {
        group.MapGet("/raw/chunks", ListChunksAsync)
            .Api("raw_chunks_list", "Raw chunk metadata", Tag).Returns<PageDto<RawChunkMetaDto>>().ProducesProblems(400, 401, 403)
            .Requires(Capability.DataRead);
        group.MapGet("/raw/chunks/{chunk_id}", DownloadChunkAsync)
            .Api("raw_chunks_get", "Download a raw chunk (application/vnd.moshiro.rawchunk)", Tag)
            .Produces(StatusCodes.Status200OK, contentType: "application/vnd.moshiro.rawchunk").ProducesProblems(401, 403, 404, 410)
            .Requires(Capability.DataRead);
        group.MapGet("/events", ListEventsAsync)
            .Api("events_list", "Agent events", Tag).Returns<PageDto<EventDto>>().ProducesProblems(400, 401, 403)
            .Requires(Capability.EventsRead);
        group.MapGet("/analysis/recompute", RecomputeAsync)
            .Api("analysis_recompute_get", "Progress of the metric backfill", Tag).Returns<RecomputeDto>().ProducesProblems(401, 403)
            .Requires(Capability.DataWrite);
    }

    private static async Task<IResult> ListChunksAsync(
        HttpContext http, RawBrowserService raw, string? agent_id, string? kind, string? test_id, DateTimeOffset? from, DateTimeOffset? to, int? limit, string? cursor)
    {
        int size = Math.Clamp(limit ?? 50, 1, 500);
        long? after = null;
        if (cursor is not null)
        {
            if (!long.TryParse(cursor, out long parsed))
            {
                return Problems.From(http, ValidationError.Single("cursor", "is not a valid cursor"));
            }

            after = parsed;
        }

        var query = new RawQuery(
            agent_id is null ? null : new NodeId(agent_id), null, kind, test_id is null ? null : new TestId(test_id),
            from?.ToUnixTimeMilliseconds(), to?.ToUnixTimeMilliseconds(), after, size + 1);
        IReadOnlyList<RawChunkMeta> metas = await raw.ListAsync(query, http.RequestAborted);
        bool more = metas.Count > size;
        IReadOnlyList<RawChunkMeta> page = [.. metas.Take(size)];

        // 游标按 start_ts 前进：下一页从最后一项的 start_ts + 1 开始（同一毫秒的多个块在同一页内给出）
        string? next = more ? (page[^1].StartTs + 1).ToString(System.Globalization.CultureInfo.InvariantCulture) : null;
        return TypedResults.Json(new PageDto<RawChunkMetaDto>([.. page.Select(Mappers.ToDto)], next), ApiJsonContext.Default.PageDtoRawChunkMetaDto);
    }

    private static async Task<IResult> DownloadChunkAsync(HttpContext http, RawBrowserService raw, string chunk_id)
    {
        if (!Guid.TryParse(chunk_id, out Guid id))
        {
            return Problems.Create(http, MoShiroErrors.ChunkNotFound, "chunk not found");
        }

        Result<(RawChunkMeta Meta, byte[] Bytes)> result = await raw.GetAsync(new ChunkId(id), http.RequestAborted);
        if (!result.TryGet(out (RawChunkMeta Meta, byte[] Bytes) found, out FleetError? error))
        {
            return Problems.From(http, error);
        }

        http.Response.Headers.CacheControl = "private, max-age=3600";
        return TypedResults.File(found.Bytes, "application/vnd.moshiro.rawchunk", chunk_id + ".msrc");
    }

    private static async Task<IResult> ListEventsAsync(
        HttpContext http, EventService events, TimeProvider time, string? agent_id, string? kind, DateTimeOffset? from, DateTimeOffset? to, int? limit, string? cursor)
    {
        long now = time.GetUtcNow().ToUnixTimeMilliseconds();
        long toMs = to?.ToUnixTimeMilliseconds() ?? now + 1;
        long fromMs = from?.ToUnixTimeMilliseconds() ?? toMs - Steps.Day;
        long? before = null;
        if (cursor is not null)
        {
            if (!long.TryParse(cursor, out long parsed))
            {
                return Problems.From(http, ValidationError.Single("cursor", "is not a valid cursor"));
            }

            before = parsed;
        }

        int size = Math.Clamp(limit ?? 50, 1, 500);
        IReadOnlyList<EventRecord> rows = await events.ListAsync(agent_id is null ? null : new NodeId(agent_id), kind, fromMs, toMs, before, size + 1, http.RequestAborted);
        bool more = rows.Count > size;
        IReadOnlyList<EventRecord> page = [.. rows.Take(size)];
        string? next = more ? page[^1].Id.ToString(System.Globalization.CultureInfo.InvariantCulture) : null;
        return TypedResults.Json(new PageDto<EventDto>([.. page.Select(Mappers.ToDto)], next), ApiJsonContext.Default.PageDtoEventDto);
    }

    private static async Task<IResult> RecomputeAsync(OpsService ops, CancellationToken ct)
    {
        RecomputeProgress p = ops.RecomputeProgress;
        return TypedResults.Json(
            new RecomputeDto(p.State, p.Metrics, p.WindowsDone, p.WindowsTotal, Mappers.Time(p.FromMs), Mappers.Time(p.ToMs), await ops.QueueLengthAsync(ct)),
            ApiJsonContext.Default.RecomputeDto);
    }
}