meeting-assistant/MeetingAssistant/Speakers/SpeakerIdentityMergeService.cs

using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Options;

namespace MeetingAssistant.Speakers;

public interface ISpeakerIdentityMergeService
{
    Task<SpeakerIdentityMergeResult> 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<SpeakerIdentityDbContext> dbContextFactory;
    private readonly ISpeakerIdentityMatcher matcher;
    private readonly SpeakerIdentificationOptions options;
    private readonly ILogger<SpeakerIdentityMergeService> logger;

    public SpeakerIdentityMergeService(
        IDbContextFactory<SpeakerIdentityDbContext> dbContextFactory,
        ISpeakerIdentityMatcher matcher,
        IOptions<MeetingAssistantOptions> options,
        ILogger<SpeakerIdentityMergeService> logger)
    {
        this.dbContextFactory = dbContextFactory;
        this.matcher = matcher;
        this.options = options.Value.SpeakerIdentification;
        this.logger = logger;
    }

    public async Task<SpeakerIdentityMergeResult> 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<SpeakerIdentity> 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();
    }

}