clickhouse

2026-08-10 17:30:52 +08:00
parent e4f1d88ca1
commit d11d038770
13 changed files with 591 additions and 0 deletions
@@ -0,0 +1,122 @@
+using System.Net.Http.Headers;
+using System.Text;
+using System.Text.Json;
+
+namespace Jiaowu.Api.Infrastructure.Analytics;
+
+public sealed class ClickHouseAnalyticsClient(
+    HttpClient httpClient,
+    ClickHouseAnalyticsOptions options)
+{
+    private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web);
+
+    public bool IsEnabled => options.Enabled;
+
+    public async Task EnsureSchemaAsync(CancellationToken cancellationToken)
+    {
+        await ExecuteAsync($"CREATE DATABASE IF NOT EXISTS {options.Database}", cancellationToken);
+
+        await ExecuteAsync($"""
+            CREATE TABLE IF NOT EXISTS {options.Database}.auditEvents
+            (
+                id UUID,
+                occurredAt DateTime64(3, 'UTC'),
+                userId Nullable(UUID),
+                method LowCardinality(String),
+                path String,
+                statusCode UInt16,
+                ipAddress Nullable(String),
+                projectedAt DateTime64(3, 'UTC')
+            ) ENGINE = ReplacingMergeTree(projectedAt)
+            PARTITION BY toYYYYMM(occurredAt)
+            ORDER BY (id)
+            """, cancellationToken);
+
+        await ExecuteAsync($"""
+            CREATE TABLE IF NOT EXISTS {options.Database}.attendanceRecords
+            (
+                attendanceSheetId UUID,
+                studentId UUID,
+                attendanceDate Date,
+                teachingTaskId UUID,
+                academicTermId UUID,
+                collegeId UUID,
+                status UInt8,
+                checkInAt Nullable(DateTime64(3, 'UTC')),
+                checkedInMethod Nullable(UInt8),
+                appealStatus UInt8,
+                projectedAt DateTime64(3, 'UTC')
+            ) ENGINE = ReplacingMergeTree(projectedAt)
+            PARTITION BY toYYYYMM(attendanceDate)
+            ORDER BY (attendanceSheetId, studentId)
+            """, cancellationToken);
+
+        await ExecuteAsync($"""
+            CREATE TABLE IF NOT EXISTS {options.Database}.gradeStatistics
+            (
+                gradeSheetId UUID,
+                teachingTaskId UUID,
+                courseId UUID,
+                academicTermId UUID,
+                collegeId UUID,
+                academicTermName LowCardinality(String),
+                studentCount UInt32,
+                passedCount UInt32,
+                excellentCount UInt32,
+                averageScore Decimal(8, 2),
+                passRate Decimal(8, 4),
+                excellentRate Decimal(8, 4),
+                calculatedAt DateTime64(3, 'UTC'),
+                projectedAt DateTime64(3, 'UTC')
+            ) ENGINE = ReplacingMergeTree(projectedAt)
+            PARTITION BY toYYYYMM(calculatedAt)
+            ORDER BY (gradeSheetId)
+            """, cancellationToken);
+    }
+
+    public async Task InsertAsync<T>(string table, IReadOnlyCollection<T> rows, CancellationToken cancellationToken)
+    {
+        if (rows.Count == 0) return;
+        var payload = JsonSerializer.Serialize(rows, JsonOptions);
+        await ExecuteAsync(
+            $"INSERT INTO {options.Database}.{table} FORMAT JSONEachRow\n{ToJsonLines(payload)}",
+            cancellationToken);
+    }
+
+    public async Task<JsonElement[]> QueryAsync(string sql, CancellationToken cancellationToken)
+    {
+        using var response = await SendAsync(sql + " FORMAT JSON", cancellationToken);
+        var content = await response.Content.ReadAsStringAsync(cancellationToken);
+        if (!response.IsSuccessStatusCode)
+            throw new HttpRequestException($"ClickHouse 查询失败 ({(int)response.StatusCode})：{content}");
+        using var document = JsonDocument.Parse(content);
+        return document.RootElement.GetProperty("data")
+            .EnumerateArray().Select(x => x.Clone()).ToArray();
+    }
+
+    private async Task ExecuteAsync(string sql, CancellationToken cancellationToken)
+    {
+        using var response = await SendAsync(sql, cancellationToken);
+        if (response.IsSuccessStatusCode) return;
+        var content = await response.Content.ReadAsStringAsync(cancellationToken);
+        throw new HttpRequestException($"ClickHouse 写入失败 ({(int)response.StatusCode})：{content}");
+    }
+
+    private Task<HttpResponseMessage> SendAsync(string sql, CancellationToken cancellationToken)
+    {
+        var request = new HttpRequestMessage(HttpMethod.Post, "")
+        {
+            Content = new StringContent(sql, Encoding.UTF8, "text/plain")
+        };
+        var credentials = Convert.ToBase64String(Encoding.UTF8.GetBytes(
+            $"{options.UserName}:{options.Password}"));
+        request.Headers.Authorization = new AuthenticationHeaderValue("Basic", credentials);
+        return httpClient.SendAsync(request, cancellationToken);
+    }
+
+    private static string ToJsonLines(string json)
+    {
+        using var document = JsonDocument.Parse(json);
+        return string.Join('\n', document.RootElement.EnumerateArray().Select(x => x.GetRawText()));
+    }
+}
@@ -0,0 +1,24 @@
+using System.Text.RegularExpressions;
+
+namespace Jiaowu.Api.Infrastructure.Analytics;
+
+public sealed partial class ClickHouseAnalyticsOptions
+{
+    public const string SectionName = "ClickHouseAnalytics";
+
+    public bool Enabled { get; set; }
+    public string Endpoint { get; set; } = "http://localhost:8123";
+    public string Database { get; set; } = "jiaowu_analytics";
+    public string UserName { get; set; } = "jiaowu_analytics";
+    public string Password { get; set; } = "";
+    public bool CreateSchemaOnStartup { get; set; } = true;
+    public int SyncIntervalSeconds { get; set; } = 60;
+    public int SourceLookbackDays { get; set; } = 90;
+    public int BatchSize { get; set; } = 1000;
+
+    [GeneratedRegex("^[a-zA-Z_][a-zA-Z0-9_]{0,62}$")]
+    private static partial Regex IdentifierPattern();
+
+    public bool HasValidIdentifiers() =>
+        IdentifierPattern().IsMatch(Database);
+}
@@ -0,0 +1,123 @@
+using Jiaowu.Api.Infrastructure.Persistence;
+using Microsoft.EntityFrameworkCore;
+
+namespace Jiaowu.Api.Infrastructure.Analytics;
+
+/// <summary>
+/// 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.
+/// </summary>
+public sealed class ClickHouseAnalyticsProjectionWorker(
+    IServiceScopeFactory scopeFactory,
+    ClickHouseAnalyticsClient client,
+    ClickHouseAnalyticsOptions options,
+    ILogger<ClickHouseAnalyticsProjectionWorker> 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<AppDbContext>();
+
+        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<int> 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<int> 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<int> 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;
+        }
+    }
+}