meeting-assistant/MeetingAssistant/Recording/OfflineTranscriptionBacklogProcessor.cs

using MeetingAssistant.LaunchProfiles;
using MeetingAssistant.MeetingNotes;
using MeetingAssistant.Summary;
using MeetingAssistant.Transcription;
using MeetingAssistant.Workflow;
using Microsoft.Extensions.Options;

namespace MeetingAssistant.Recording;

public sealed class OfflineTranscriptionBacklogProcessor
{
    private readonly IOfflineTranscriptionBacklog backlog;
    private readonly ISpeechRecognitionPipelineFactory pipelineFactory;
    private readonly ITranscriptStore transcriptStore;
    private readonly IMeetingNoteStore meetingNoteStore;
    private readonly IMeetingArtifactStore meetingArtifactStore;
    private readonly IMeetingSummaryPipeline summaryPipeline;
    private readonly IMeetingWorkflowEngine workflowEngine;
    private readonly IRecordingDictationWordProvider dictationWordProvider;
    private readonly MeetingAssistantOptions options;
    private readonly ILogger<OfflineTranscriptionBacklogProcessor> logger;
    private readonly ILaunchProfileOptionsProvider? launchProfiles;

    public OfflineTranscriptionBacklogProcessor(
        IOfflineTranscriptionBacklog backlog,
        ISpeechRecognitionPipelineFactory pipelineFactory,
        ITranscriptStore transcriptStore,
        IMeetingNoteStore meetingNoteStore,
        IMeetingArtifactStore meetingArtifactStore,
        IMeetingSummaryPipeline summaryPipeline,
        IMeetingWorkflowEngine workflowEngine,
        IRecordingDictationWordProvider dictationWordProvider,
        IOptions<MeetingAssistantOptions> options,
        ILogger<OfflineTranscriptionBacklogProcessor> logger,
        ILaunchProfileOptionsProvider? launchProfiles = null)
    {
        this.backlog = backlog;
        this.pipelineFactory = pipelineFactory;
        this.transcriptStore = transcriptStore;
        this.meetingNoteStore = meetingNoteStore;
        this.meetingArtifactStore = meetingArtifactStore;
        this.summaryPipeline = summaryPipeline;
        this.workflowEngine = workflowEngine;
        this.dictationWordProvider = dictationWordProvider;
        this.options = options.Value;
        this.logger = logger;
        this.launchProfiles = launchProfiles;
    }

    public async Task<int> ProcessPendingAsync(CancellationToken cancellationToken)
    {
        var processed = 0;
        foreach (var item in await backlog.ListAsync(cancellationToken))
        {
            cancellationToken.ThrowIfCancellationRequested();
            if (await TryProcessAsync(item, cancellationToken))
            {
                processed++;
            }
        }

        return processed;
    }

    private async Task<bool> TryProcessAsync(
        OfflineTranscriptionBacklogItem item,
        CancellationToken cancellationToken)
    {
        try
        {
            await ProcessAsync(item, cancellationToken);
            await backlog.CompleteAsync(item, cancellationToken);
            logger.LogInformation(
                "Completed offline transcription backlog item {BacklogItemId} for recording {RecordingPath}",
                item.Id,
                item.AudioPath);
            return true;
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            throw;
        }
        catch (Exception exception)
        {
            logger.LogWarning(
                exception,
                "Could not process offline transcription backlog item {BacklogItemId}; it will be retried later",
                item.Id);
            return false;
        }
    }

    private async Task ProcessAsync(
        OfflineTranscriptionBacklogItem item,
        CancellationToken cancellationToken)
    {
        var runOptions = ResolveOptions(item.LaunchProfileName);
        var artifacts = new MeetingSessionArtifacts(
            item.MeetingNotePath,
            item.TranscriptPath,
            item.AssistantContextPath,
            item.SummaryPath);
        var session = new TranscriptSession(item.TranscriptPath);
        var meetingNote = await meetingNoteStore.ReadAsync(item.MeetingNotePath, cancellationToken);
        var pipelineOptions = await RecordingSpeechRecognitionPipelineOptions.BuildAsync(
            dictationWordProvider,
            runOptions,
            meetingNote,
            cancellationToken);

        await using var pipeline = pipelineFactory.Create(item.LaunchProfileName);
        await pipeline.InitializeAsync(pipelineOptions, cancellationToken);
        await pipeline.WaitUntilReadyAsync(cancellationToken);
        await WriteRecordingToPipelineAsync(item.AudioPath, pipeline, cancellationToken);
        var finishedSegments = await pipeline.ReadFinishedTranscriptAsync(
            item.AudioPath,
            pipelineOptions,
            cancellationToken);

        var lines = await TransformTranscriptLinesAsync(artifacts, runOptions, finishedSegments, cancellationToken);
        await transcriptStore.ReplaceLinesAsync(session, lines, cancellationToken);

        meetingNote.Frontmatter.EndTime = item.InferredEndTime ?? DateTimeOffset.Now;
        var completedMeetingNote = await meetingNoteStore.SaveAsync(meetingNote, runOptions, cancellationToken);
        await transcriptStore.UpdateMetadataAsync(session, artifacts, completedMeetingNote, cancellationToken);
        await meetingArtifactStore.UpdateAssistantContextMeetingAsync(artifacts, completedMeetingNote, cancellationToken);

        await TransitionMeetingAsync(
            artifacts,
            runOptions,
            AssistantContextState.Transcribing,
            AssistantContextState.Summarizing,
            cancellationToken);
        var summaryResult = await summaryPipeline.RunAsync(artifacts, runOptions, cancellationToken);
        await TransitionMeetingAsync(
            artifacts,
            runOptions,
            AssistantContextState.Summarizing,
            summaryResult.Succeeded ? AssistantContextState.Finished : AssistantContextState.Error,
            cancellationToken);
    }

    private MeetingAssistantOptions ResolveOptions(string? launchProfileName)
    {
        if (launchProfiles is not null)
        {
            return launchProfiles.GetRequiredProfile(launchProfileName).Options;
        }

        return options;
    }

    private async Task<List<string>> TransformTranscriptLinesAsync(
        MeetingSessionArtifacts artifacts,
        MeetingAssistantOptions runOptions,
        IReadOnlyList<TranscriptionSegment> segments,
        CancellationToken cancellationToken)
    {
        var lines = new List<string>(segments.Count);
        foreach (var segment in segments)
        {
            var formatted = TranscriptLineFormatter.Format(segment);
            lines.Add(await TransformTranscriptLineAsync(artifacts, runOptions, formatted, cancellationToken));
        }

        return lines;
    }

    private async Task<string> TransformTranscriptLineAsync(
        MeetingSessionArtifacts artifacts,
        MeetingAssistantOptions runOptions,
        FormattedTranscriptLine formatted,
        CancellationToken cancellationToken)
    {
        try
        {
            return await workflowEngine.TransformTranscriptLineAsync(
                MeetingWorkflowEvent.TranscriptLine(artifacts, formatted.Line, formatted.Speaker),
                runOptions,
                cancellationToken);
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            throw;
        }
        catch (Exception exception)
        {
            logger.LogError(
                exception,
                "Offline transcript line workflow failed for speaker {Speaker}; keeping original transcript line",
                formatted.Speaker);
            return formatted.Line;
        }
    }

    private async Task TransitionMeetingAsync(
        MeetingSessionArtifacts artifacts,
        MeetingAssistantOptions runOptions,
        AssistantContextState from,
        AssistantContextState to,
        CancellationToken cancellationToken)
    {
        await meetingArtifactStore.UpdateAssistantContextStateAsync(artifacts, to, cancellationToken);
        await workflowEngine.RunAsync(
            MeetingWorkflowEvent.StateTransition(artifacts, from, to),
            runOptions,
            cancellationToken);
    }

    private static async Task WriteRecordingToPipelineAsync(
        string audioPath,
        ISpeechRecognitionPipeline pipeline,
        CancellationToken cancellationToken)
    {
        await foreach (var chunk in PcmWavAudioChunkReader.ReadChunksAsync(audioPath, cancellationToken: cancellationToken))
        {
            await pipeline.WriteAsync(chunk, cancellationToken);
        }
    }
}