using Jiaowu.Api.Infrastructure.Persistence;
using Microsoft.EntityFrameworkCore;
namespace Jiaowu.Api.Infrastructure.Analytics;
///
/// Projects MySQL facts to ClickHouse. The projection is deliberately
/// best-effort: business writes never depend on an analytics database.
/// ReplacingMergeTree plus FINAL reads make repeated lookback batches safe.
///
public sealed class ClickHouseAnalyticsProjectionWorker(
IServiceScopeFactory scopeFactory,
ClickHouseAnalyticsClient client,
ClickHouseAnalyticsOptions options,
ILogger logger) : BackgroundService
{
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
if (!options.Enabled) return;
try
{
if (options.CreateSchemaOnStartup)
await client.EnsureSchemaAsync(stoppingToken);
}
catch (Exception exception) when (!stoppingToken.IsCancellationRequested)
{
logger.LogError(exception, "ClickHouse 分析表初始化失败,将在下一轮重试。");
}
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(options.SyncIntervalSeconds));
do
{
try
{
await ProjectAsync(stoppingToken);
}
catch (Exception exception) when (!stoppingToken.IsCancellationRequested)
{
logger.LogError(exception, "ClickHouse 分析投影失败,将在下一轮重试。");
}
} while (await timer.WaitForNextTickAsync(stoppingToken));
}
private async Task ProjectAsync(CancellationToken cancellationToken)
{
var projectedAt = DateTime.UtcNow;
var from = projectedAt.AddDays(-options.SourceLookbackDays);
await using var scope = scopeFactory.CreateAsyncScope();
var db = scope.ServiceProvider.GetRequiredService();
var auditCount = await ProjectAuditsAsync(db, from, projectedAt, cancellationToken);
var attendanceCount = await ProjectAttendanceAsync(db, from.Date, projectedAt, cancellationToken);
var gradeCount = await ProjectGradesAsync(db, projectedAt, cancellationToken);
logger.LogInformation(
"ClickHouse 分析投影完成:审计 {AuditCount},考勤 {AttendanceCount},成绩 {GradeCount}。",
auditCount, attendanceCount, gradeCount);
}
private async Task ProjectAuditsAsync(AppDbContext db, DateTime from, DateTime projectedAt, CancellationToken cancellationToken)
{
var cursorAt = from;
var cursorId = Guid.Empty;
var count = 0;
while (true)
{
var rows = await db.AuditLogs.AsNoTracking()
.Where(x => x.CreatedAt > cursorAt || x.CreatedAt == cursorAt && x.Id.CompareTo(cursorId) > 0)
.OrderBy(x => x.CreatedAt).ThenBy(x => x.Id).Take(options.BatchSize)
.Select(x => new { x.Id, OccurredAt = x.CreatedAt, x.UserId, x.Method, x.Path, x.StatusCode, x.IpAddress, ProjectedAt = projectedAt })
.ToListAsync(cancellationToken);
if (rows.Count == 0) return count;
await client.InsertAsync("auditEvents", rows, cancellationToken);
count += rows.Count;
cursorAt = rows[^1].OccurredAt;
cursorId = rows[^1].Id;
}
}
private async Task ProjectAttendanceAsync(AppDbContext db, DateTime from, DateTime projectedAt, CancellationToken cancellationToken)
{
var cursorDate = from;
var cursorSheetId = Guid.Empty;
var cursorStudentId = Guid.Empty;
var count = 0;
while (true)
{
var rows = await db.AttendanceRecords.AsNoTracking()
.Where(x => x.AttendanceSheet!.AttendanceDate > cursorDate ||
x.AttendanceSheet.AttendanceDate == cursorDate &&
(x.AttendanceSheetId.CompareTo(cursorSheetId) > 0 ||
x.AttendanceSheetId == cursorSheetId && x.StudentId.CompareTo(cursorStudentId) > 0))
.OrderBy(x => x.AttendanceSheet!.AttendanceDate).ThenBy(x => x.AttendanceSheetId).ThenBy(x => x.StudentId).Take(options.BatchSize)
.Select(x => new { x.AttendanceSheetId, x.StudentId, AttendanceDate = x.AttendanceSheet!.AttendanceDate, TeachingTaskId = x.AttendanceSheet.TeachingTaskId, AcademicTermId = x.AttendanceSheet.TeachingTask!.AcademicTermId, CollegeId = x.AttendanceSheet.TeachingTask.Course!.CollegeId, Status = (byte)x.Status, x.CheckInAt, CheckedInMethod = x.CheckedInMethod == null ? null : (byte?)x.CheckedInMethod, AppealStatus = (byte)x.AppealStatus, ProjectedAt = projectedAt })
.ToListAsync(cancellationToken);
if (rows.Count == 0) return count;
await client.InsertAsync("attendanceRecords", rows, cancellationToken);
count += rows.Count;
cursorDate = rows[^1].AttendanceDate;
cursorSheetId = rows[^1].AttendanceSheetId;
cursorStudentId = rows[^1].StudentId;
}
}
private async Task ProjectGradesAsync(AppDbContext db, DateTime projectedAt, CancellationToken cancellationToken)
{
var cursorAt = DateTime.MinValue;
var cursorSheetId = Guid.Empty;
var count = 0;
while (true)
{
var rows = await db.TeachingTaskGradeStatistics.AsNoTracking()
.Where(x => x.CalculatedAt > cursorAt || x.CalculatedAt == cursorAt && x.GradeSheetId.CompareTo(cursorSheetId) > 0)
.OrderBy(x => x.CalculatedAt).ThenBy(x => x.GradeSheetId).Take(options.BatchSize)
.Select(x => new { x.GradeSheetId, x.TeachingTaskId, x.CourseId, x.AcademicTermId, CollegeId = x.TeachingTask!.Course!.CollegeId, AcademicTermName = x.TeachingTask.AcademicTerm!.Name, x.StudentCount, x.PassedCount, x.ExcellentCount, x.AverageScore, x.PassRate, x.ExcellentRate, x.CalculatedAt, ProjectedAt = projectedAt })
.ToListAsync(cancellationToken);
if (rows.Count == 0) return count;
await client.InsertAsync("gradeStatistics", rows, cancellationToken);
count += rows.Count;
cursorAt = rows[^1].CalculatedAt;
cursorSheetId = rows[^1].GradeSheetId;
}
}
}