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}";
                SpeakerIdentityMerger.MergeInto(
                    target,
                    source,
                    options.MaxSnippetsPerSpeaker);
                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);
    }

}