Run AI classification in background with SignalR notifications
Move the classify-uncategorized endpoint from synchronous to async. The controller now queues a ClassificationJob to an unbounded channel and returns 202 Accepted. A new BackgroundService processes jobs and pushes results (or errors) to the user via SignalR. The frontend listens for the AiClassificationComplete event and surfaces it in the notification system. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,7 +1,7 @@
|
|||||||
import { ref, computed } from 'vue'
|
import { ref, computed } from 'vue'
|
||||||
import { HubConnectionBuilder, HubConnection, LogLevel } from '@microsoft/signalr'
|
import { HubConnectionBuilder, HubConnection, LogLevel } from '@microsoft/signalr'
|
||||||
import { useAuthStore } from '@/stores/auth'
|
import { useAuthStore } from '@/stores/auth'
|
||||||
import type { AppNotification, FileImportNotification, BankSyncNotification } from '@/types'
|
import type { AppNotification, FileImportNotification, BankSyncNotification, AiClassificationNotification } from '@/types'
|
||||||
|
|
||||||
const notifications = ref<AppNotification[]>([])
|
const notifications = ref<AppNotification[]>([])
|
||||||
const isConnected = ref(false)
|
const isConnected = ref(false)
|
||||||
@@ -42,6 +42,16 @@ function connect() {
|
|||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
|
||||||
|
connection.on('AiClassificationComplete', (data: AiClassificationNotification) => {
|
||||||
|
notifications.value.unshift({
|
||||||
|
id: crypto.randomUUID(),
|
||||||
|
type: 'AiClassificationComplete',
|
||||||
|
data,
|
||||||
|
timestamp: new Date().toISOString(),
|
||||||
|
read: false
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|
||||||
connection.onclose(() => {
|
connection.onclose(() => {
|
||||||
isConnected.value = false
|
isConnected.value = false
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -414,10 +414,17 @@ export interface BankSyncNotification {
|
|||||||
status: string
|
status: string
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export interface AiClassificationNotification {
|
||||||
|
totalUncategorized: number
|
||||||
|
classified: number
|
||||||
|
skipped: number
|
||||||
|
error?: string
|
||||||
|
}
|
||||||
|
|
||||||
export interface AppNotification {
|
export interface AppNotification {
|
||||||
id: string
|
id: string
|
||||||
type: 'FileImportReady' | 'BankSyncComplete'
|
type: 'FileImportReady' | 'BankSyncComplete' | 'AiClassificationComplete'
|
||||||
data: FileImportNotification | BankSyncNotification
|
data: FileImportNotification | BankSyncNotification | AiClassificationNotification
|
||||||
timestamp: string
|
timestamp: string
|
||||||
read: boolean
|
read: boolean
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,5 +1,7 @@
|
|||||||
|
using System.Threading.Channels;
|
||||||
using Microsoft.AspNetCore.Authorization;
|
using Microsoft.AspNetCore.Authorization;
|
||||||
using Microsoft.AspNetCore.Mvc;
|
using Microsoft.AspNetCore.Mvc;
|
||||||
|
using Purrse.Api.Services;
|
||||||
using Purrse.Core.DTOs;
|
using Purrse.Core.DTOs;
|
||||||
using Purrse.Core.Interfaces.Services;
|
using Purrse.Core.Interfaces.Services;
|
||||||
using System.Security.Claims;
|
using System.Security.Claims;
|
||||||
@@ -12,9 +14,14 @@ namespace Purrse.Api.Controllers;
|
|||||||
public class AiCategorizationController : ControllerBase
|
public class AiCategorizationController : ControllerBase
|
||||||
{
|
{
|
||||||
private readonly IAiCategorizationService _aiService;
|
private readonly IAiCategorizationService _aiService;
|
||||||
|
private readonly Channel<ClassificationJob> _classificationChannel;
|
||||||
private Guid UserId => Guid.Parse(User.FindFirstValue(ClaimTypes.NameIdentifier)!);
|
private Guid UserId => Guid.Parse(User.FindFirstValue(ClaimTypes.NameIdentifier)!);
|
||||||
|
|
||||||
public AiCategorizationController(IAiCategorizationService aiService) => _aiService = aiService;
|
public AiCategorizationController(IAiCategorizationService aiService, Channel<ClassificationJob> classificationChannel)
|
||||||
|
{
|
||||||
|
_aiService = aiService;
|
||||||
|
_classificationChannel = classificationChannel;
|
||||||
|
}
|
||||||
|
|
||||||
[HttpGet("settings")]
|
[HttpGet("settings")]
|
||||||
public async Task<ActionResult<AiCategorizationSettingsResponse>> GetSettings()
|
public async Task<ActionResult<AiCategorizationSettingsResponse>> GetSettings()
|
||||||
@@ -29,6 +36,10 @@ public class AiCategorizationController : ControllerBase
|
|||||||
=> Ok(await _aiService.TestConnectionAsync(UserId));
|
=> Ok(await _aiService.TestConnectionAsync(UserId));
|
||||||
|
|
||||||
[HttpPost("classify")]
|
[HttpPost("classify")]
|
||||||
public async Task<ActionResult<ClassifyUncategorizedResponse>> ClassifyUncategorized(ClassifyUncategorizedRequest request)
|
public async Task<IActionResult> ClassifyUncategorized(ClassifyUncategorizedRequest request)
|
||||||
=> Ok(await _aiService.ClassifyUncategorizedAsync(UserId, request));
|
{
|
||||||
|
var job = new ClassificationJob(UserId, request.AccountId, request.StartDate, request.EndDate, request.SearchText, request.Status);
|
||||||
|
await _classificationChannel.Writer.WriteAsync(job);
|
||||||
|
return Accepted(new { message = "Classification job queued" });
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
using System.Text;
|
using System.Text;
|
||||||
|
using System.Threading.Channels;
|
||||||
using Microsoft.AspNetCore.Authentication.JwtBearer;
|
using Microsoft.AspNetCore.Authentication.JwtBearer;
|
||||||
using Microsoft.EntityFrameworkCore;
|
using Microsoft.EntityFrameworkCore;
|
||||||
using Microsoft.IdentityModel.Tokens;
|
using Microsoft.IdentityModel.Tokens;
|
||||||
@@ -99,6 +100,8 @@ builder.Services.AddSingleton<PluginService>();
|
|||||||
builder.Services.AddHostedService<ScheduledTransactionProcessor>();
|
builder.Services.AddHostedService<ScheduledTransactionProcessor>();
|
||||||
builder.Services.AddHostedService<FileWatcherService>();
|
builder.Services.AddHostedService<FileWatcherService>();
|
||||||
builder.Services.AddHostedService<BankSyncBackgroundService>();
|
builder.Services.AddHostedService<BankSyncBackgroundService>();
|
||||||
|
builder.Services.AddSingleton(Channel.CreateUnbounded<ClassificationJob>());
|
||||||
|
builder.Services.AddHostedService<AiClassificationBackgroundService>();
|
||||||
|
|
||||||
// SignalR
|
// SignalR
|
||||||
builder.Services.AddSignalR();
|
builder.Services.AddSignalR();
|
||||||
|
|||||||
@@ -0,0 +1,93 @@
|
|||||||
|
using System.Threading.Channels;
|
||||||
|
using Microsoft.AspNetCore.SignalR;
|
||||||
|
using Purrse.Api.Hubs;
|
||||||
|
using Purrse.Core.DTOs;
|
||||||
|
using Purrse.Core.Enums;
|
||||||
|
using Purrse.Core.Interfaces.Services;
|
||||||
|
|
||||||
|
namespace Purrse.Api.Services;
|
||||||
|
|
||||||
|
public record ClassificationJob(
|
||||||
|
Guid UserId,
|
||||||
|
Guid? AccountId,
|
||||||
|
DateTime? StartDate,
|
||||||
|
DateTime? EndDate,
|
||||||
|
string? SearchText,
|
||||||
|
TransactionStatus? Status)
|
||||||
|
{
|
||||||
|
public ClassifyUncategorizedRequest ToRequest() =>
|
||||||
|
new(AccountId, StartDate, EndDate, SearchText, Status);
|
||||||
|
}
|
||||||
|
|
||||||
|
public class AiClassificationBackgroundService : BackgroundService
|
||||||
|
{
|
||||||
|
private readonly Channel<ClassificationJob> _channel;
|
||||||
|
private readonly IServiceScopeFactory _scopeFactory;
|
||||||
|
private readonly IHubContext<NotificationHub> _hubContext;
|
||||||
|
private readonly ILogger<AiClassificationBackgroundService> _logger;
|
||||||
|
|
||||||
|
public AiClassificationBackgroundService(
|
||||||
|
Channel<ClassificationJob> channel,
|
||||||
|
IServiceScopeFactory scopeFactory,
|
||||||
|
IHubContext<NotificationHub> hubContext,
|
||||||
|
ILogger<AiClassificationBackgroundService> logger)
|
||||||
|
{
|
||||||
|
_channel = channel;
|
||||||
|
_scopeFactory = scopeFactory;
|
||||||
|
_hubContext = hubContext;
|
||||||
|
_logger = logger;
|
||||||
|
}
|
||||||
|
|
||||||
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||||
|
{
|
||||||
|
_logger.LogInformation("AI classification background service started");
|
||||||
|
|
||||||
|
await foreach (var job in _channel.Reader.ReadAllAsync(stoppingToken))
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
_logger.LogInformation("Processing AI classification job for user {UserId}", job.UserId);
|
||||||
|
|
||||||
|
using var scope = _scopeFactory.CreateScope();
|
||||||
|
var aiService = scope.ServiceProvider.GetRequiredService<IAiCategorizationService>();
|
||||||
|
var result = await aiService.ClassifyUncategorizedAsync(job.UserId, job.ToRequest());
|
||||||
|
|
||||||
|
await _hubContext.Clients.Group(job.UserId.ToString())
|
||||||
|
.SendAsync("AiClassificationComplete", new
|
||||||
|
{
|
||||||
|
totalUncategorized = result.TotalUncategorized,
|
||||||
|
classified = result.Classified,
|
||||||
|
skipped = result.Skipped
|
||||||
|
}, stoppingToken);
|
||||||
|
|
||||||
|
_logger.LogInformation(
|
||||||
|
"AI classification complete for user {UserId}: {Classified}/{Total} classified",
|
||||||
|
job.UserId, result.Classified, result.TotalUncategorized);
|
||||||
|
}
|
||||||
|
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
catch (Exception ex)
|
||||||
|
{
|
||||||
|
_logger.LogError(ex, "AI classification job failed for user {UserId}", job.UserId);
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await _hubContext.Clients.Group(job.UserId.ToString())
|
||||||
|
.SendAsync("AiClassificationComplete", new
|
||||||
|
{
|
||||||
|
totalUncategorized = 0,
|
||||||
|
classified = 0,
|
||||||
|
skipped = 0,
|
||||||
|
error = ex.Message
|
||||||
|
}, stoppingToken);
|
||||||
|
}
|
||||||
|
catch (Exception notifyEx)
|
||||||
|
{
|
||||||
|
_logger.LogWarning(notifyEx, "Failed to send error notification for user {UserId}", job.UserId);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user