TEAMModelOS/TEAMModelOS.Extension/IES.Exam/IES.ExamServer/DI/SubjectPushService.cs

using IES.ExamServer.Helpers;
using IES.ExamServer.Models;
using LiteDB;
using System;
using System.Net.Http;
using System.Text;
using System.Threading.Channels;

namespace IES.ExamServer.DI
{
    public class SubjectPushService: BackgroundService
    {
        private readonly DataQueue _dataQueue;
        private readonly IHttpClientFactory _httpClientFactory;
        private readonly LiteDBFactory _liteDBFactory;
        private readonly CenterServiceConnectionService _connectionService;
        private readonly IConfiguration _configuration;
        public SubjectPushService(DataQueue dataQueue, IHttpClientFactory httpClientFactory, LiteDBFactory liteDBFactory, CenterServiceConnectionService connectionService, IConfiguration configuration)
        {
            
            _dataQueue = dataQueue;
            _httpClientFactory = httpClientFactory;
            _liteDBFactory = liteDBFactory;
            _connectionService = connectionService;
            _configuration= configuration;

        }
        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            ///数据中心链接了，开始推送数据
            if (_connectionService.centerIsConnected) 
            {
                // 启动时加载未推送的数据
                //await LoadUnpushedDataAsync(stoppingToken);
            }
            while (!stoppingToken.IsCancellationRequested)
            {
                // 从 Channel 中读取数据
                if (await _dataQueue.Reader.WaitToReadAsync(stoppingToken))
                {
                    while (_dataQueue.Reader.TryRead(out var data))
                    {

                        await PushSubjectResultDataToCloudAsync(data, stoppingToken);
                       
                    }
                }
            }
        }
        
        private async Task PushSubjectResultDataToCloudAsync((EvaluationStudentResult studentResult, EvaluationSubjectResult subjectResult, string resultId)data, CancellationToken cancellationToken)
        {
            if (_connectionService.centerIsConnected) 
            {
                try
                {
                    //  var json = JsonSerializer.Serialize(data);
                    //var content = new StringContent(json, Encoding.UTF8, "application/json");
                    var httpClient = _httpClientFactory.CreateClient();

                    string url = $"{_connectionService.centerUrl}/common/exam/upsert-new-record";
                    if (httpClient.DefaultRequestHeaders.Contains("X-Auth-AuthToken")) {
                        httpClient.DefaultRequestHeaders.Remove("X-Auth-AuthToken");
                    }
                    httpClient.DefaultRequestHeaders.Add("X-Auth-AuthToken", _connectionService.loginToken);
                    var response = await httpClient.PostAsJsonAsync(url, new { 
                        id =data.subjectResult.examId,
                        answer=data.subjectResult.answers,
                        subjectId=data.subjectResult.subjectId,
                        classId=data.studentResult.classId,
                        ownerId=data.studentResult.ownerId,
                        paperId=data.subjectResult.paperId,
                        studentId=data.studentResult.studentId,
                        studentName=data.studentResult.studentName,
                    }, cancellationToken);

                    if (response.IsSuccessStatusCode)
                    {
                        MarkDataAsPushed(data);
                        Console.WriteLine($"Data {data.resultId} pushed successfully.");
                    }
                    else
                    {
                        Console.WriteLine($"Failed to push data {data.resultId}. Retrying...");
                        // 推送失败，重新加入队列
                        await _dataQueue.Writer.WriteAsync(data, cancellationToken);
                    }
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"Error pushing data {data.resultId}: {ex.Message}");
                    // 推送失败，重新加入队列
                    await _dataQueue.Writer.WriteAsync(data, cancellationToken);
                }
            }
        }
        private void MarkDataAsPushed((EvaluationStudentResult studentResult, EvaluationSubjectResult subjectResult, string resultId) data)
        {
            var collection = _liteDBFactory.GetLiteDatabase().GetCollection<EvaluationStudentResult>();
            EvaluationSubjectResult? subjectResult = data.studentResult.subjectResults.Where(x => x.id!.Equals(data.resultId)).FirstOrDefault();
            if (subjectResult!=null) 
            {
                subjectResult.pushed=1;
                collection.Upsert(data.studentResult);
                var collectionSub = _liteDBFactory.GetLiteDatabase().GetCollection<EvaluationSubjectResult>();
                var unpushedData = collectionSub.FindOne(x => data.resultId.Equals(x.id));
                if (unpushedData!=null) 
                {
                    unpushedData.pushed=1;
                    _dataQueue.MarkAsProcessed((unpushedData.id!, unpushedData.examId!, unpushedData.evaluationId!, unpushedData.subjectId!,unpushedData, data.studentResult));
                }
            }
            
        }
        private async Task LoadUnpushedDataAsync(CancellationToken cancellationToken)
        {
            var collection = _liteDBFactory.GetLiteDatabase().GetCollection<EvaluationSubjectResult>();
            var unpushedData = collection.Find(x => x.pushed!=1);
            foreach (var data in unpushedData)
            {
                var studentResults  = _liteDBFactory.GetLiteDatabase().GetCollection<EvaluationStudentResult>().Find(x => x.subjectResults.Where(x => x.id!.Equals(data.id)).IsNotEmpty()).Distinct();
                // 将未推送的数据写入 Channel
                foreach (var studentResult in studentResults)
                {
                    await _dataQueue.WriteAsync((studentResult!,data, data.id!), cancellationToken); 
                }
            }
        }
    }

    public class DataQueue
    {
        private readonly Channel<(EvaluationStudentResult studentResult, EvaluationSubjectResult subjectResult, string resultId)> _channel;
        private readonly HashSet<(string resultId,string examId,string evaluationId,string subjectId, EvaluationSubjectResult subjectResult, EvaluationStudentResult studentResult)> _uniqueIds; // 用于维护唯一性

        public async Task<bool> TryAddAsync((string resultId, string examId, string evaluationId, string subjectId, EvaluationSubjectResult subjectResult, EvaluationStudentResult studentResult) data, CancellationToken cancellationToken = default)
        {
            var key = (data.resultId, data.examId,data.evaluationId,data.subjectId,data.subjectResult,data.studentResult); // 复合键
            lock (_uniqueIds)
            {
                if (_uniqueIds.Contains(key))
                {
                    return false;
                }
                _uniqueIds.Add(key);
            }

            await _channel.Writer.WriteAsync((data.studentResult,data.subjectResult, data.resultId), cancellationToken);
            return true;
        }
        public DataQueue() {
            // 创建一个无界 Channel
            _channel = Channel.CreateUnbounded<(EvaluationStudentResult studentResult, EvaluationSubjectResult subjectResult, string resultId)>();
            _uniqueIds = new HashSet<(string resultId, string examId, string evaluationId, string subjectId, EvaluationSubjectResult subjectResult, EvaluationStudentResult studentResult)>();
        }
        public ChannelWriter<(EvaluationStudentResult studentResult, EvaluationSubjectResult subjectResult, string resultId)> Writer => _channel.Writer;
        public ChannelReader<(EvaluationStudentResult studentResult, EvaluationSubjectResult subjectResult, string resultId)> Reader => _channel.Reader;

        // 批量写入数据
        public async Task WriteAsync(( EvaluationStudentResult studentResult, EvaluationSubjectResult subjectResult, string resultId) data, CancellationToken cancellationToken = default)
        {
            await _channel.Writer.WriteAsync((data.studentResult,data.subjectResult, data.resultId), cancellationToken);
        }
        // 获取当前队列中的未上传数据数量
        public int GetPendingCount()
        {
            return _channel.Reader.Count;
        }
        // 从队列中移除数据时，同时从 HashSet 中移除 ID
        public void MarkAsProcessed((string resultId, string examId, string evaluationId, string subjectId, EvaluationSubjectResult subjectResult, EvaluationStudentResult studentResult) key)
        {
            lock (_uniqueIds) // 加锁确保线程安全
            {
                _uniqueIds.Remove(key);
            }
        }
    }
}