using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Options; namespace MeetingAssistant.Speakers; public interface ISpeakerIdentityMergeService { Task MergeRecentIdentitiesAsync( TimeSpan? recentIdentityAge, CancellationToken cancellationToken); } public sealed record SpeakerIdentityMergeResult( int RecentIdentityCount, int CandidateIdentityCount, int MatchAttempts, int MergedPairs); public sealed class SpeakerIdentityMergeService : ISpeakerIdentityMergeService { private readonly IDbContextFactory dbContextFactory; private readonly ISpeakerIdentityMatcher matcher; private readonly SpeakerIdentificationOptions options; private readonly ILogger logger; public SpeakerIdentityMergeService( IDbContextFactory dbContextFactory, ISpeakerIdentityMatcher matcher, IOptions options, ILogger logger) { this.dbContextFactory = dbContextFactory; this.matcher = matcher; this.options = options.Value.SpeakerIdentification; this.logger = logger; } public async Task MergeRecentIdentitiesAsync( TimeSpan? recentIdentityAge, CancellationToken cancellationToken) { await using var context = await dbContextFactory.CreateDbContextAsync(cancellationToken); await SpeakerIdentitySchema.EnsureCreatedOrUpdatedAsync(context, cancellationToken); var cutoff = DateTimeOffset.UtcNow - (recentIdentityAge ?? options.MergeRecentIdentityAge); var identities = await context.SpeakerIdentities .Include(identity => identity.Aliases) .Include(identity => identity.CandidateNames) .Include(identity => identity.Snippets) .Include(identity => identity.References) .OrderByDescending(identity => identity.References.Count) .ThenBy(identity => identity.Id) .ToListAsync(cancellationToken); var recentIds = identities .Where(identity => identity.CreatedAt >= cutoff) .Select(identity => identity.Id) .ToHashSet(); var mergedPairs = 0; var attempts = 0; foreach (var sourceId in recentIds.ToList()) { var source = identities.SingleOrDefault(identity => identity.Id == sourceId); if (source is null || source.Snippets.Count == 0) { continue; } var targets = identities .Where(identity => identity.Id != source.Id && identity.Snippets.Count > 0) .ToList(); foreach (var batch in targets.Chunk(Math.Max(1, options.MatchBatchSize))) { var firstSnippet = SelectSnippet(source, excludedSnippet: null); if (firstSnippet is null) { break; } attempts++; var firstMatch = await matcher.MatchAsync( CreateRequest(source, firstSnippet, batch), cancellationToken); if (firstMatch is null || firstMatch.IdentityId == source.Id) { continue; } var target = batch.SingleOrDefault(identity => identity.Id == firstMatch.IdentityId); if (target is null) { continue; } var secondSnippet = SelectSnippet(source, excludedSnippet: firstSnippet); if (secondSnippet is null) { continue; } attempts++; var secondMatch = await matcher.MatchAsync( CreateRequest(source, secondSnippet, batch), cancellationToken); if (secondMatch?.IdentityId != target.Id) { continue; } var targetName = target.GetDisplayName() ?? $"identity-{target.Id}"; var sourceName = source.GetDisplayName() ?? $"identity-{source.Id}"; MergeIdentities(target, source); await SpeakerIdentityTranscriptAudit.AppendMergedAsync( target.References, targetName, sourceName, cancellationToken); context.SpeakerIdentities.Remove(source); identities.Remove(source); mergedPairs++; break; } } await context.SaveChangesAsync(cancellationToken); logger.LogInformation( "Speaker identity merge diagnostics completed: {MergedPairs} merge(s), {Attempts} match attempt(s)", mergedPairs, attempts); return new SpeakerIdentityMergeResult( recentIds.Count, identities.Count, attempts, mergedPairs); } private static SpeakerIdentityMatchRequest CreateRequest( SpeakerIdentity source, SpeakerSnippet unknownSnippet, IReadOnlyList candidates) { return new SpeakerIdentityMatchRequest( source.GetDisplayName() ?? $"identity-{source.Id}", unknownSnippet.WavBytes, candidates.Select(identity => new SpeakerIdentityMatchCandidate( identity.Id, identity.GetDisplayName(), identity.ReferenceCount, identity.Snippets .OrderBy(snippet => snippet.CreatedAt) .Select(snippet => snippet.WavBytes) .ToList())) .ToList()); } private static SpeakerSnippet? SelectSnippet(SpeakerIdentity identity, SpeakerSnippet? excludedSnippet) { return identity.Snippets .OrderBy(snippet => snippet.CreatedAt) .FirstOrDefault(snippet => excludedSnippet is null || snippet.Id != excludedSnippet.Id); } private void MergeIdentities(SpeakerIdentity target, SpeakerIdentity source) { AddAlias(target, source.CanonicalName); foreach (var alias in source.Aliases) { AddAlias(target, alias.Name); } foreach (var candidate in source.CandidateNames) { AddAlias(target, candidate.Name); } target.UpdatedAt = DateTimeOffset.UtcNow; foreach (var reference in source.References) { SpeakerIdentityReferences.AddIfMissing(target, reference); } var retainedSnippets = target.Snippets .Concat(source.Snippets) .OrderBy(snippet => StableSnippetKey(snippet.WavBytes)) .Take(Math.Max(1, options.MaxSnippetsPerSpeaker)) .Select(snippet => new SpeakerSnippet { WavBytes = snippet.WavBytes, CreatedAt = snippet.CreatedAt }) .ToList(); target.Snippets.Clear(); target.Snippets.AddRange(retainedSnippets); } private static void AddAlias(SpeakerIdentity identity, string? alias) { if (string.IsNullOrWhiteSpace(alias) || string.Equals(identity.CanonicalName, alias, StringComparison.OrdinalIgnoreCase) || identity.Aliases.Any(existing => string.Equals(existing.Name, alias, StringComparison.OrdinalIgnoreCase))) { return; } identity.Aliases.Add(new SpeakerAlias { Name = alias.Trim() }); } private static int StableSnippetKey(byte[] snippet) { var hash = new HashCode(); foreach (var value in snippet) { hash.Add(value); } return hash.ToHashCode(); } }