Remove queue

This commit is contained in:
Michał Zieliński
2025-06-08 11:08:48 +02:00
parent 99d8593c49
commit 4fd0b1cd50
5 changed files with 2 additions and 777 deletions

View File

@@ -21,7 +21,6 @@ public class LayersController : Controller
private readonly IConfiguration _configuration;
private readonly PluginManager _pluginManager;
private readonly ILogger<LayersController> _logger;
private readonly IJobQueueService _queueService;
public LayersController(
AppDbContext db,
@@ -29,8 +28,7 @@ public class LayersController : Controller
GoogleDriveHelper googleDriveHelper,
IConfiguration configuration,
PluginManager pluginManager,
ILogger<LayersController> logger,
IJobQueueService queueService
ILogger<LayersController> logger
)
{
_db = db;
@@ -39,7 +37,6 @@ public class LayersController : Controller
_configuration = configuration;
_pluginManager = pluginManager;
_logger = logger;
_queueService = queueService;
}
[HttpGet]
@@ -241,51 +238,6 @@ public class LayersController : Controller
}
}
[HttpGet]
[Route("AutoImportWithQueue/{apiKey}")]
[AllowAnonymous]
public IActionResult AutoImportWithQueue(string apiKey)
{
if (Request.Host.Value != _configuration["apiLocalUrl"] || apiKey != _configuration["apiKey"])
{
_logger.LogWarning("AutoImportQueue: Unauthorized request with apiKey {ApiKey}", apiKey);
return Unauthorized();
}
var importWorkerLayers = _db.Layers
.Include(x => x.Records)
.Where(x =>
x.Records!.Any(y => y.Code == "Type" && y.Desc1 == "ImportWorker") &&
x.Records!.Any(y => y.Code == "IsEnabled" && y.Desc1 == "True")
)
.OrderBy(x => x.CreatedAt)
.AsNoTracking()
.ToList();
if (importWorkerLayers.Count == 0)
{
_logger.LogInformation("AutoImportQueue: No layers to import");
return Ok();
}
_logger.LogInformation("AutoImportQueue: Found {LayerCount} layers to queue", importWorkerLayers.Count);
foreach (var importWorker in importWorkerLayers)
{
try
{
// Queue job implementation would go here
_logger.LogDebug("AutoImportQueue: Queued layer {LayerName} ({LayerId})", importWorker.Name, importWorker.Id);
}
catch (Exception e)
{
_logger.LogError(e, "AutoImportQueue: Error while adding job for layer {LayerName} ({LayerId})",
importWorker.Name, importWorker.Id);
}
}
return Ok();
}
[HttpGet]
[Route("ProcessQueue/{apiKey}")]
[AllowAnonymous]
@@ -679,42 +631,6 @@ public class LayersController : Controller
[HttpGet]
[Route("CheckProcessors")]
[AllowAnonymous]
public IActionResult CheckProcessors()
{
// get
/*
var importWorkerLayers = _db.Layers
.Include(x => x.Records)
.Where(x =>
x.Records!.Any(y => y.Code == "Type" && y.Desc1 == "ImportWorker") &&
x.Records!.Any(y => y.Code == "ImportType" && y.Desc1 == "Import-D3")
)
.OrderByDescending(x => x.CreatedAt)
.AsNoTracking()
.ToList();
foreach (var importWorker in importWorkerLayers)
{
var record = new Record
{
Id = Guid.NewGuid(),
LayerId = importWorker.Id,
Code = "Plugin",
Desc1 = "Morska.Import.D3",
CreatedAt = DateTime.UtcNow,
ModifiedAt = DateTime.UtcNow,
CreatedById = Guid.Parse("F392209E-123E-4651-A5A4-0B1D6CF9FF9D"),
ModifiedById = Guid.Parse("F392209E-123E-4651-A5A4-0B1D6CF9FF9D")
};
//_db.Records.Add(record);
}
//_db.SaveChanges();
*/
return Ok();
}
private static void WriteToConsole(params string[] messages)
{
@@ -809,278 +725,4 @@ public class LayersController : Controller
throw;
}
}
[HttpGet]
[Route("EnqueueImportWorkers/{apiKey}")]
[AllowAnonymous]
public async Task<IActionResult> EnqueueImportWorkers(string apiKey, [FromQuery] Guid? layerId = null)
{
if (Request.Host.Value != _configuration["apiLocalUrl"] || apiKey != _configuration["apiKey"])
{
_logger.LogWarning("EnqueueImportWorkers: Unauthorized request with apiKey {ApiKey}", apiKey);
return Unauthorized();
}
try
{
var query = _db.Layers
.Include(x => x.Records)
.Where(x =>
x.Records!.Any(y => y.Code == "Type" && y.Desc1 == "ImportWorker") &&
x.Records!.Any(y => y.Code == "IsEnabled" && y.Desc1 == "True") &&
!x.IsDeleted && !x.IsCancelled
).Take(5);
// If specific layerId is provided, filter to that layer only
if (layerId.HasValue)
{
query = query.Where(x => x.Id == layerId.Value);
}
var importWorkerLayers = await query
.OrderBy(x => x.CreatedAt)
.AsNoTracking()
.ToListAsync();
_logger.LogInformation("EnqueueImportWorkers: Found {LayerCount} import worker layers to queue{LayerFilter}",
importWorkerLayers.Count, layerId.HasValue ? $" (filtered by LayerId: {layerId})" : "");
if (importWorkerLayers.Count == 0)
{
return Ok(new
{
Message = "No import workers found to queue",
QueuedJobs = 0,
TotalQueueSize = await _queueService.GetQueueCountAsync(),
SkippedLayers = 0
});
}
int queuedCount = 0;
foreach (var importWorker in importWorkerLayers)
{
var pluginName = importWorker.Records!.FirstOrDefault(x => x.Code == "Plugin")?.Desc1;
if (string.IsNullOrEmpty(pluginName))
{
_logger.LogWarning("EnqueueImportWorkers: No plugin name found for layer {LayerName} ({LayerId}), skipping",
importWorker.Name, importWorker.Id);
continue;
}
// Check if plugin exists
var importer = _pluginManager.GetImporter(pluginName);
if (importer == null)
{
_logger.LogWarning("EnqueueImportWorkers: Importer {PluginName} not found for layer {LayerName} ({LayerId}), skipping",
pluginName, importWorker.Name, importWorker.Id);
continue;
}
var job = new QueueJob
{
LayerId = importWorker.Id,
LayerName = importWorker.Name ?? "Unknown",
PluginName = pluginName,
JobType = JobType.Import,
Priority = 0, // All imports have same priority
MaxRetries = 5,
CreatedById = Guid.Parse("F392209E-123E-4651-A5A4-0B1D6CF9FF9D"),
ModifiedById = Guid.Parse("F392209E-123E-4651-A5A4-0B1D6CF9FF9D")
};
await _queueService.EnqueueJobAsync(job);
queuedCount++;
_logger.LogDebug("EnqueueImportWorkers: Queued import job for layer {LayerName} ({LayerId}) with plugin {PluginName}",
importWorker.Name, importWorker.Id, pluginName);
}
var totalQueueSize = await _queueService.GetQueueCountAsync();
_logger.LogInformation("EnqueueImportWorkers: Successfully queued {QueuedCount} import jobs. Total queue size: {QueueSize}",
queuedCount, totalQueueSize);
return Ok(new
{
Message = $"Queued {queuedCount} import jobs",
QueuedJobs = queuedCount,
TotalQueueSize = totalQueueSize,
SkippedLayers = importWorkerLayers.Count - queuedCount
});
}
catch (Exception e)
{
_logger.LogError(e, "EnqueueImportWorkers: Error queuing import workers");
return BadRequest(e.ToString());
}
}
[HttpGet]
[Route("EnqueueProcessWorkers/{apiKey}")]
[AllowAnonymous]
public async Task<IActionResult> EnqueueProcessWorkers(string apiKey, [FromQuery] Guid? layerId = null)
{
if (Request.Host.Value != _configuration["apiLocalUrl"] || apiKey != _configuration["apiKey"])
{
_logger.LogWarning("EnqueueProcessWorkers: Unauthorized request with apiKey {ApiKey}", apiKey);
return Unauthorized();
}
try
{
var query = _db.Layers
.Include(x => x.Records)
.Where(x =>
x.Records!.Any(y => y.Code == "Type" && y.Desc1 == "ProcessWorker") &&
x.Records!.Any(y => y.Code == "IsEnabled" && y.Desc1 == "True") &&
!x.IsDeleted && !x.IsCancelled
).Take(5);
// If specific layerId is provided, filter to that layer only
if (layerId.HasValue)
{
query = query.Where(x => x.Id == layerId.Value);
}
var processWorkerLayers = await query
.OrderBy(x => x.CreatedAt)
.AsNoTracking()
.ToListAsync();
_logger.LogInformation("EnqueueProcessWorkers: Found {LayerCount} process worker layers to queue{LayerFilter}",
processWorkerLayers.Count, layerId.HasValue ? $" (filtered by LayerId: {layerId})" : "");
if (processWorkerLayers.Count == 0)
{
return Ok(new
{
Message = "No process workers found to queue",
QueuedJobs = 0,
TotalQueueSize = await _queueService.GetQueueCountAsync(),
SkippedLayers = 0
});
}
int queuedCount = 0;
foreach (var processWorker in processWorkerLayers)
{
// POPRAWIONE: Używaj Plugin zamiast ProcessorType
var pluginName = processWorker.Records!.FirstOrDefault(x => x.Code == "Plugin")?.Desc1;
if (string.IsNullOrEmpty(pluginName))
{
_logger.LogWarning("EnqueueProcessWorkers: No plugin name found for layer {LayerName} ({LayerId}), skipping",
processWorker.Name, processWorker.Id);
continue;
}
// POPRAWIONE: Sprawdź czy processor istnieje przez pluginName
var processor = _pluginManager.GetProcessor(pluginName);
if (processor == null)
{
_logger.LogWarning("EnqueueProcessWorkers: Processor {PluginName} not found for layer {LayerName} ({LayerId}), skipping",
pluginName, processWorker.Name, processWorker.Id);
continue;
}
// Get priority from ProcessWorker record, default to 10 if not found
var priorityStr = processWorker.Records!.FirstOrDefault(x => x.Code == "Priority")?.Desc1;
var priority = 10; // Default priority
if (!string.IsNullOrEmpty(priorityStr) && int.TryParse(priorityStr, out var parsedPriority))
{
priority = parsedPriority;
}
var job = new QueueJob
{
LayerId = processWorker.Id,
LayerName = processWorker.Name ?? "Unknown",
PluginName = pluginName, // POPRAWIONE: Używaj pluginName bezpośrednio
JobType = JobType.Process,
Priority = priority,
MaxRetries = 3,
CreatedById = Guid.Parse("F392209E-123E-4651-A5A4-0B1D6CF9FF9D"),
ModifiedById = Guid.Parse("F392209E-123E-4651-A5A4-0B1D6CF9FF9D")
};
await _queueService.EnqueueJobAsync(job);
queuedCount++;
_logger.LogDebug("EnqueueProcessWorkers: Queued process job for layer {LayerName} ({LayerId}) with plugin {PluginName}, priority {Priority}",
processWorker.Name, processWorker.Id, pluginName, priority);
}
var totalQueueSize = await _queueService.GetQueueCountAsync();
_logger.LogInformation("EnqueueProcessWorkers: Successfully queued {QueuedCount} process jobs. Total queue size: {QueueSize}",
queuedCount, totalQueueSize);
return Ok(new
{
Message = $"Queued {queuedCount} process jobs",
QueuedJobs = queuedCount,
TotalQueueSize = totalQueueSize,
SkippedLayers = processWorkerLayers.Count - queuedCount
});
}
catch (Exception e)
{
_logger.LogError(e, "EnqueueProcessWorkers: Error queuing process workers");
return BadRequest(e.ToString());
}
}
[HttpGet]
[Route("RunQueueJobs/{apiKey}")]
[AllowAnonymous]
public async Task<IActionResult> RunQueueJobs(string apiKey)
{
if (Request.Host.Value != _configuration["apiLocalUrl"] || apiKey != _configuration["apiKey"])
{
_logger.LogWarning("RunQueueJobs: Unauthorized request with apiKey {ApiKey}", apiKey);
return Unauthorized();
}
try
{
var queueSize = await _queueService.GetQueueCountAsync();
if (queueSize == 0)
{
return Ok(new
{
Message = "Queue is empty",
QueueSize = 0,
Status = "No jobs to process"
});
}
_logger.LogInformation("RunQueueJobs: Triggering queue processing for {QueueSize} jobs", queueSize);
// ZMIEŃ NA DOSTĘP PRZEZ IHostedService:
var hostedServices = HttpContext.RequestServices.GetServices<IHostedService>();
var queueProcessor = hostedServices.OfType<JobQueueProcessor>().FirstOrDefault();
if (queueProcessor == null)
{
_logger.LogError("RunQueueJobs: JobQueueProcessor not found");
return BadRequest("JobQueueProcessor not found");
}
queueProcessor.TriggerProcessing();
return Ok(new
{
Message = $"Queue processing triggered for {queueSize} jobs",
QueueSize = queueSize,
Status = "Processing started in background"
});
}
catch (Exception e)
{
_logger.LogError(e, "RunQueueJobs: Error triggering queue processing");
return BadRequest(e.ToString());
}
}
}