fix(backend): resolve TR json serialization, news concurrency and asset index locks
This commit is contained in:
@@ -1,4 +1,4 @@
|
|||||||
using Microsoft.EntityFrameworkCore.Migrations;
|
using Microsoft.EntityFrameworkCore.Migrations;
|
||||||
|
|
||||||
#nullable disable
|
#nullable disable
|
||||||
|
|
||||||
@@ -10,6 +10,11 @@ namespace FinlyticAssets.Migrations
|
|||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
protected override void Up(MigrationBuilder migrationBuilder)
|
protected override void Up(MigrationBuilder migrationBuilder)
|
||||||
{
|
{
|
||||||
|
migrationBuilder.Sql(@"
|
||||||
|
DELETE FROM ""DynamicSettings"" a USING ""DynamicSettings"" b
|
||||||
|
WHERE a.""Key"" = b.""Key"" AND a.""Id"" < b.""Id"";
|
||||||
|
");
|
||||||
|
|
||||||
migrationBuilder.DropIndex(
|
migrationBuilder.DropIndex(
|
||||||
name: "IX_DynamicSettings_Key",
|
name: "IX_DynamicSettings_Key",
|
||||||
table: "DynamicSettings");
|
table: "DynamicSettings");
|
||||||
|
|||||||
@@ -36,6 +36,8 @@ public class AssetsIndexService : IAssetsIndexService
|
|||||||
private readonly IAssetsDbService _assetsDbService;
|
private readonly IAssetsDbService _assetsDbService;
|
||||||
private static readonly HttpClient _httpClient = new();
|
private static readonly HttpClient _httpClient = new();
|
||||||
|
|
||||||
|
private static readonly SemaphoreSlim _fileLock = new(1, 1);
|
||||||
|
|
||||||
public AssetsIndexService(IFinlyticLogger<AssetsIndexService> finlyticLogger, IAssetsDbService assetsDbService)
|
public AssetsIndexService(IFinlyticLogger<AssetsIndexService> finlyticLogger, IAssetsDbService assetsDbService)
|
||||||
{
|
{
|
||||||
_finlyticLogger = finlyticLogger;
|
_finlyticLogger = finlyticLogger;
|
||||||
@@ -70,10 +72,19 @@ public class AssetsIndexService : IAssetsIndexService
|
|||||||
Directory.CreateDirectory(directoryPath);
|
Directory.CreateDirectory(directoryPath);
|
||||||
}
|
}
|
||||||
|
|
||||||
using (var fileStream = new FileStream(filePath, FileMode.Create, FileAccess.Write, FileShare.None, 4096,
|
await _fileLock.WaitAsync(cancellationToken);
|
||||||
useAsync: true))
|
try
|
||||||
{
|
{
|
||||||
await JsonSerializer.SerializeAsync(fileStream, indexAssets, cancellationToken: cancellationToken);
|
var tempFilePath = Path.Combine(directoryPath, $"index_{Guid.NewGuid():N}.tmp");
|
||||||
|
using (var fileStream = new FileStream(tempFilePath, FileMode.Create, FileAccess.Write, FileShare.ReadWrite, 4096, useAsync: true))
|
||||||
|
{
|
||||||
|
await JsonSerializer.SerializeAsync(fileStream, indexAssets, cancellationToken: cancellationToken);
|
||||||
|
}
|
||||||
|
File.Move(tempFilePath, filePath, overwrite: true);
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
_fileLock.Release();
|
||||||
}
|
}
|
||||||
|
|
||||||
await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetsIndexService] Successfully recreated asset index file with {Count} entries pointing to local logos at {Path}",
|
await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetsIndexService] Successfully recreated asset index file with {Count} entries pointing to local logos at {Path}",
|
||||||
@@ -81,18 +92,15 @@ public class AssetsIndexService : IAssetsIndexService
|
|||||||
}
|
}
|
||||||
catch (IOException ex)
|
catch (IOException ex)
|
||||||
{
|
{
|
||||||
await _finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[AssetsIndexService] Disk I/O error occurred while writing the asset index file.");
|
await _finlyticLogger.LogWarningAsync(SettingKeys.AssetsChannel, ex, "[AssetsIndexService] Concurrent file access while writing asset index file. Skipping cycle.");
|
||||||
throw;
|
|
||||||
}
|
}
|
||||||
catch (JsonException ex)
|
catch (JsonException ex)
|
||||||
{
|
{
|
||||||
await _finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[AssetsIndexService] Failed to serialize the asset index data to JSON.");
|
await _finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[AssetsIndexService] Failed to serialize the asset index data to JSON.");
|
||||||
throw;
|
|
||||||
}
|
}
|
||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
await _finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[AssetsIndexService] An unexpected error occurred while recreating the asset index file.");
|
await _finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[AssetsIndexService] An unexpected error occurred while recreating the asset index file.");
|
||||||
throw;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -130,4 +130,4 @@ public class TradeRepublicAssetConverter : JsonConverter<TradeRepublicAsset>
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
file record TradeRepublicAssetFallback : TradeRepublicAsset;
|
public record TradeRepublicAssetFallback : TradeRepublicAsset;
|
||||||
|
|||||||
@@ -104,6 +104,20 @@ namespace FinlyticCore.Util;
|
|||||||
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicConnectRequest))]
|
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicConnectRequest))]
|
||||||
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicSearchRequest))]
|
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicSearchRequest))]
|
||||||
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicAssetResponse))]
|
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicAssetResponse))]
|
||||||
|
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicAsset))]
|
||||||
|
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicStock))]
|
||||||
|
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicCrypto))]
|
||||||
|
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicEtf))]
|
||||||
|
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicSynthetic))]
|
||||||
|
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicBond))]
|
||||||
|
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicDerivative))]
|
||||||
|
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicAssetFallback))]
|
||||||
|
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicTag))]
|
||||||
|
[JsonSerializable(typeof(List<FinlyticCore.Dtos.TradeRepublic.TradeRepublicTag>))]
|
||||||
|
[JsonSerializable(typeof(IReadOnlyList<FinlyticCore.Dtos.TradeRepublic.TradeRepublicTag>))]
|
||||||
|
[JsonSerializable(typeof(IReadOnlyList<string>))]
|
||||||
|
[JsonSerializable(typeof(IList<FinlyticCore.Dtos.TradeRepublic.TradeRepublicAsset>))]
|
||||||
|
[JsonSerializable(typeof(List<FinlyticCore.Dtos.TradeRepublic.TradeRepublicAsset>))]
|
||||||
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicDerivativesRequest))]
|
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicDerivativesRequest))]
|
||||||
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicDerivativesResponse))]
|
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicDerivativesResponse))]
|
||||||
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicStockDetailsRequest))]
|
[JsonSerializable(typeof(FinlyticCore.Dtos.TradeRepublic.TradeRepublicStockDetailsRequest))]
|
||||||
|
|||||||
@@ -211,52 +211,65 @@ public class NewsDbService : INewsDbService
|
|||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
public async Task<NewsArticleEntity?> SaveArticleClassificationAsync(Guid id, N8nResponsePayload payload, List<MatchedAssetEntity> matchedAssets)
|
public async Task<NewsArticleEntity?> SaveArticleClassificationAsync(Guid id, N8nResponsePayload payload, List<MatchedAssetEntity> matchedAssets)
|
||||||
{
|
{
|
||||||
var article = await _context.NewsArticles
|
DateTime? finalPublishedAt = null;
|
||||||
.FirstOrDefaultAsync(a => a.Id == id);
|
if (DateTime.TryParse(payload.PublishedAt, out var publishedDate))
|
||||||
|
{
|
||||||
|
finalPublishedAt = publishedDate.Kind == DateTimeKind.Unspecified
|
||||||
|
? DateTime.SpecifyKind(publishedDate, DateTimeKind.Utc)
|
||||||
|
: publishedDate.ToUniversalTime();
|
||||||
|
}
|
||||||
|
|
||||||
if (article == null)
|
DateTime? finalScrapedAt = null;
|
||||||
|
if (DateTime.TryParse(payload.ScrapedAt, out var scrapedDate))
|
||||||
|
{
|
||||||
|
finalScrapedAt = scrapedDate.ToUniversalTime();
|
||||||
|
}
|
||||||
|
|
||||||
|
var rowsAffected = await _context.NewsArticles
|
||||||
|
.Where(a => a.Id == id)
|
||||||
|
.ExecuteUpdateAsync(s => s
|
||||||
|
.SetProperty(a => a.Title, payload.Title)
|
||||||
|
.SetProperty(a => a.Author, payload.Author)
|
||||||
|
.SetProperty(a => a.Summary, payload.Summary)
|
||||||
|
.SetProperty(a => a.ContentRaw, payload.ContentRaw)
|
||||||
|
.SetProperty(a => a.Language, payload.Language)
|
||||||
|
.SetProperty(a => a.Status, "Completed")
|
||||||
|
.SetProperty(a => a.PublishedAt, a => finalPublishedAt ?? a.PublishedAt)
|
||||||
|
.SetProperty(a => a.ScrapedAt, a => finalScrapedAt ?? a.ScrapedAt));
|
||||||
|
|
||||||
|
if (rowsAffected == 0)
|
||||||
{
|
{
|
||||||
await _finlyticLogger.LogWarningAsync(SettingKeys.NewsChannel, "[NewsChannel] Article with ID {Id} not found for classification update.", id);
|
await _finlyticLogger.LogWarningAsync(SettingKeys.NewsChannel, "[NewsChannel] Article with ID {Id} not found for classification update.", id);
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
article.Title = payload.Title;
|
// Cleanly delete existing matched assets and insert new ones
|
||||||
article.Author = payload.Author;
|
|
||||||
article.Summary = payload.Summary;
|
|
||||||
article.ContentRaw = payload.ContentRaw;
|
|
||||||
article.Language = payload.Language;
|
|
||||||
article.Status = "Completed";
|
|
||||||
|
|
||||||
if (DateTime.TryParse(payload.PublishedAt, out var publishedDate))
|
|
||||||
{
|
|
||||||
article.PublishedAt = publishedDate.Kind == DateTimeKind.Unspecified
|
|
||||||
? DateTime.SpecifyKind(publishedDate, DateTimeKind.Utc)
|
|
||||||
: publishedDate.ToUniversalTime();
|
|
||||||
}
|
|
||||||
|
|
||||||
if (DateTime.TryParse(payload.ScrapedAt, out var scrapedDate))
|
|
||||||
{
|
|
||||||
article.ScrapedAt = scrapedDate.ToUniversalTime();
|
|
||||||
}
|
|
||||||
|
|
||||||
await _context.MatchedAssets.Where(m => m.NewsArticleId == id).ExecuteDeleteAsync();
|
await _context.MatchedAssets.Where(m => m.NewsArticleId == id).ExecuteDeleteAsync();
|
||||||
|
|
||||||
article.MatchedAssets = new List<MatchedAssetEntity>();
|
if (matchedAssets != null && matchedAssets.Count > 0)
|
||||||
foreach (var asset in matchedAssets)
|
|
||||||
{
|
{
|
||||||
if (asset.Id == Guid.Empty)
|
foreach (var asset in matchedAssets)
|
||||||
{
|
{
|
||||||
asset.Id = Guid.NewGuid();
|
if (asset.Id == Guid.Empty)
|
||||||
|
{
|
||||||
|
asset.Id = Guid.NewGuid();
|
||||||
|
}
|
||||||
|
asset.NewsArticleId = id;
|
||||||
}
|
}
|
||||||
|
await _context.MatchedAssets.AddRangeAsync(matchedAssets);
|
||||||
_context.MatchedAssets.Add(asset);
|
await _context.SaveChangesAsync();
|
||||||
article.MatchedAssets.Add(asset);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
await _context.SaveChangesAsync();
|
_context.ChangeTracker.Clear();
|
||||||
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[Lifecycle] Article {Id} successfully classified and marked 'Completed'. Title: '{Title}'", article.Id, article.Title);
|
|
||||||
|
|
||||||
return article;
|
var completedArticle = await _context.NewsArticles
|
||||||
|
.Include(a => a.MatchedAssets)
|
||||||
|
.AsNoTracking()
|
||||||
|
.FirstOrDefaultAsync(a => a.Id == id);
|
||||||
|
|
||||||
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[Lifecycle] Article {Id} successfully classified and marked 'Completed'. Title: '{Title}'", id, payload.Title);
|
||||||
|
|
||||||
|
return completedArticle;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
|
|||||||
Reference in New Issue
Block a user