using System.Text.Json; using Dapper; using Npgsql; using Kbx.Shared.Experiments; using Kbx.Shared.Experiments.Generated; namespace Kbx.Shared.Telemetry; public sealed class KbxUxTelemetryRepository(NpgsqlDataSource dataSource) { public async Task AppendAsync(Guid tenantId,Guid userId,IReadOnlyList events,CancellationToken ct) { if(events.Count is 0)return;if(events.Count>100)throw new ArgumentException("Telemetry batch is limited to 100 events."); foreach(var item in events)KbxUxTelemetryGuard.Validate(item); await using var connection=await dataSource.OpenConnectionAsync(ct); var normalizedEvents=await NormalizeExperimentContextAsync(connection,tenantId,userId,events,ct); await using var tx=await connection.BeginTransactionAsync(ct); const string sql=""" insert into kbx.ux_events(id,tenant_id,event_name,screen_id,screen_version,session_id,task_session_id,app_version,experiment_id,experiment_variant,occurred_at,duration_ms,event_count,attributes) values(@Id,@TenantId,@EventName,@ScreenId,@ScreenVersion,@SessionId,@TaskSessionId,@AppVersion,@ExperimentId,@ExperimentVariant,@OccurredAt,@DurationMs,@Count,cast(@Attributes as jsonb)); """; foreach(var item in normalizedEvents) { await connection.ExecuteAsync(new CommandDefinition(sql,new{Id=Guid.NewGuid(),TenantId=tenantId,item.EventName,item.ScreenId,item.ScreenVersion,item.SessionId,item.TaskSessionId,item.AppVersion,item.ExperimentId,item.ExperimentVariant,item.OccurredAt,item.DurationMs,Count=item.Count??1,Attributes=JsonSerializer.Serialize(item.Attributes??new Dictionary())},tx,cancellationToken:ct)); } await tx.CommitAsync(ct); } private static async Task> NormalizeExperimentContextAsync(NpgsqlConnection connection,Guid tenantId,Guid userId,IReadOnlyList events,CancellationToken ct) { var candidates=events.Where(x=>x.EventName!="experiment.exposed"||x.ExperimentId is not null).ToArray(); var ids=candidates.Where(x=>x.ExperimentId is not null).Select(x=>x.ExperimentId!).Distinct(StringComparer.Ordinal).ToArray(); if(ids.Length==0)return candidates; const string sql=""" select a.experiment_id ExperimentId,a.variant Variant,r.state State,r.rollout_percent RolloutPercent,r.kill_switch KillSwitch from kbx.experiment_assignments a join kbx.experiment_runtime r on r.tenant_id=a.tenant_id and r.experiment_id=a.experiment_id where a.tenant_id=@TenantId and a.user_id=@UserId and a.experiment_id=any(@Ids); """; var rows=(await connection.QueryAsync(new CommandDefinition(sql,new{TenantId=tenantId,UserId=userId,Ids=ids},cancellationToken:ct))).ToDictionary(x=>x.ExperimentId,StringComparer.Ordinal); var result=new List(candidates.Length); foreach(var item in candidates) { if(item.ExperimentId is null){result.Add(item);continue;} var valid=KbxExperimentCatalog.Experiments.TryGetValue(item.ExperimentId,out var definition) && definition.Variants.Any(x=>x.Key==item.ExperimentVariant) && rows.TryGetValue(item.ExperimentId,out var row) && row.State=="running"&&!row.KillSwitch&&row.Variant==item.ExperimentVariant && KbxExperimentAssignmentPolicy.IsEnrolled(tenantId,userId,item.ExperimentId,row.RolloutPercent); if(valid)result.Add(item); else if(item.EventName!="experiment.exposed")result.Add(item with { ExperimentId=null,ExperimentVariant=null }); } return result; } private sealed record ExperimentContextRow(string ExperimentId,string Variant,string State,int RolloutPercent,bool KillSwitch); } public sealed class KbxUxBusinessOutcomeRecorder(NpgsqlDataSource dataSource) { public async Task RecordAsync(Guid tenantId,DateOnly businessDate,string? screenId,string workType,int observedCount,int autoProcessedCount,int manualInterventionCount,string? reasonCode,CancellationToken ct) { if(observedCount<0||autoProcessedCount<0||manualInterventionCount<0||autoProcessedCount+manualInterventionCount>observedCount)throw new ArgumentOutOfRangeException(nameof(observedCount)); await using var connection=await dataSource.OpenConnectionAsync(ct); const string sql=""" insert into kbx.ux_business_outcomes(tenant_id,business_date,screen_id,work_type,observed_count,auto_processed_count,manual_intervention_count,reason_code,updated_at) values(@TenantId,@BusinessDate,@ScreenId,@WorkType,@Observed,@Auto,@Manual,@Reason,now()) on conflict(tenant_id,business_date,screen_id,work_type,reason_code) do update set observed_count=excluded.observed_count,auto_processed_count=excluded.auto_processed_count,manual_intervention_count=excluded.manual_intervention_count,updated_at=now(); """; await connection.ExecuteAsync(new CommandDefinition(sql,new{TenantId=tenantId,BusinessDate=businessDate,ScreenId=screenId??string.Empty,WorkType=workType,Observed=observedCount,Auto=autoProcessedCount,Manual=manualInterventionCount,Reason=reasonCode??string.Empty},cancellationToken:ct)); } }