using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text.Json.Serialization;
using System.Threading.Tasks;
using FinlyticBackend.Util;
using FinlyticCore.Dtos;
using FinlyticCore.Dtos.Settings;
using FinlyticCore.Util;
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Cors;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Logging;
namespace FinlyticBackend.Controllers;
///
/// Represents a single configuration item for a microservice.
///
public record ServiceConfigItem(
string Key,
string Value,
string DataType,
string Description,
DateTime UpdatedAt
);
///
/// DTO for returning service configuration items (AOT-compliant).
///
public record ServiceConfigItemResponseDto(
[property: JsonPropertyName("key")] string Key,
[property: JsonPropertyName("value")] string Value,
[property: JsonPropertyName("dataType")] string DataType,
[property: JsonPropertyName("description")] string Description,
[property: JsonPropertyName("updatedAt")] DateTime UpdatedAt
);
///
/// DTO representing the operational health status of a microservice (AOT-compliant).
///
public record ServiceHealthStatusDto(
[property: JsonPropertyName("name")] string Name,
[property: JsonPropertyName("type")] string Type,
[property: JsonPropertyName("status")] string Status,
[property: JsonPropertyName("port")] string Port,
[property: JsonPropertyName("communication")] string Communication,
[property: JsonPropertyName("db")] string Db,
[property: JsonPropertyName("lastPing")] DateTime LastPing
);
[ApiController]
[Route("api/v1/admin/settings")]
[Authorize(Roles = "Admin")]
[EnableCors("AllowAll")]
public class AdminSettingsController : ControllerBase
{
private readonly WebMqttClient _mqttClient;
private readonly ILogger _logger;
private static readonly Dictionary ServiceRpcPrefixes = new(StringComparer.OrdinalIgnoreCase)
{
["FinlyticFundamentals"] = "fundamentals",
["FinlyticNews"] = "news",
["FinlyticTechnicalAnalysis"] = "ta",
["FinlyticSentiment"] = "sentiment",
["FinlyticAnalyzer"] = "analyzer",
["FinlyticTrades"] = "trades",
["FinlyticAssets"] = "assets"
};
private static readonly ConcurrentDictionary> _inMemorySettings = new(StringComparer.OrdinalIgnoreCase);
public AdminSettingsController(
WebMqttClient mqttClient,
ILogger logger)
{
_mqttClient = mqttClient;
_logger = logger;
}
///
/// Retrieves recent buffered in-memory logs for a specific service.
///
[HttpGet("logs/{serviceName}")]
public IActionResult GetServiceLogs(string serviceName)
{
if (BackendMqttBridge.ServiceLogsRingBuffer.TryGetValue(serviceName, out var queue))
{
return Ok(queue.ToList());
}
return Ok(new List());
}
///
/// Retrieves all service configurations grouped by service name via live MQTT RPC queries.
///
[HttpGet]
public async Task GetAllSettings()
{
if (_mqttClient.IsConnected)
{
var fetchTasks = ServiceRpcPrefixes.Select(async kvp =>
{
var serviceName = kvp.Key;
var prefix = kvp.Value;
try
{
var liveSettings = await _mqttClient.SendRpcRequestAsync, string>(
$"{prefix}_settings_GetAll",
"",
TimeSpan.FromSeconds(2));
if (liveSettings != null && liveSettings.Count > 0)
{
_inMemorySettings[serviceName] = liveSettings.Select(d => new ServiceConfigItem(
Key: d.Key,
Value: d.Value?.ToString() ?? "",
DataType: d.Type,
Description: d.Description,
UpdatedAt: d.UpdatedAt ?? DateTime.UtcNow
)).ToList();
}
}
catch (Exception ex)
{
_logger.LogWarning(ex, "[AdminSettings] Could not fetch live settings from {ServiceName} via MQTT.", serviceName);
}
});
await Task.WhenAll(fetchTasks);
}
var grouped = _inMemorySettings.ToDictionary(
g => g.Key,
g => g.Value.Select(item => new ServiceConfigItemResponseDto(
Key: item.Key,
Value: item.Value,
DataType: item.DataType,
Description: item.Description,
UpdatedAt: item.UpdatedAt
)).ToList()
);
return Ok(grouped);
}
///
/// Performs live MQTT RPC health pings across all microservices and returns their real operational status.
///
[HttpGet("health")]
public async Task GetServicesHealth()
{
var servicesToCheck = new (string Name, string Channel, string Type, string Db)[]
{
("FinlyticAssets", "health_Ping/FinlyticAssets", "Asset Catalog & Scraper", "PostgreSQL assets"),
("FinlyticNews", "health_Ping/FinlyticNews", "News RSS Scraper & AI", "PostgreSQL news"),
("FinlyticTechnicalAnalysis", "health_Ping/FinlyticTechnicalAnalysis", "Technical Indicators (EMA/RSI)", "PostgreSQL ta"),
("FinlyticSentiment", "health_Ping/FinlyticSentiment", "NLP Sentiment Engine", "PostgreSQL sentiment"),
("FinlyticAnalyzer", "health_Ping/FinlyticAnalyzer", "Multi-Layer Signal Engine", "PostgreSQL analyzer"),
("FinlyticTrades", "health_Ping/FinlyticTrades", "Trade Lifecycle Manager", "PostgreSQL trades"),
("FinlyticFundamentals", "health_Ping/FinlyticFundamentals", "Financial Statements & Estimates", "PostgreSQL fundamentals"),
};
var results = new List
{
new(
Name: "FinlyticBackend",
Type: "REST API & SignalR Gateway",
Status: "Online",
Port: "5000",
Communication: "Kestrel HTTP / WebSocket",
Db: "PostgreSQL backend",
LastPing: DateTime.UtcNow
)
};
var tasks = servicesToCheck.Select(async s =>
{
try
{
if (_mqttClient.IsConnected)
{
var resp = await _mqttClient.SendRpcRequestAsync(
s.Channel,
new EmptyRequest(),
TimeSpan.FromMilliseconds(1200)
);
if (resp != null)
{
return new ServiceHealthStatusDto(
Name: s.Name,
Type: s.Type,
Status: "Online",
Port: "MQTT Only (No HTTP Port)",
Communication: "MQTT RPC & Pub/Sub",
Db: s.Db,
LastPing: resp.Timestamp
);
}
}
}
catch (Exception ex)
{
_logger.LogWarning(ex, "[AdminSettings] Health ping timed out or failed for service {ServiceName}", s.Name);
}
return new ServiceHealthStatusDto(
Name: s.Name,
Type: s.Type,
Status: "Offline",
Port: "MQTT Only (No HTTP Port)",
Communication: "MQTT (No Response / Timeout)",
Db: s.Db,
LastPing: DateTime.UtcNow
);
});
var pingResults = await Task.WhenAll(tasks);
results.AddRange(pingResults);
return Ok(results);
}
///
/// Retrieves settings for a specific service via live MQTT RPC.
///
[HttpGet("{serviceName}")]
public async Task GetServiceSettings(string serviceName)
{
if (ServiceRpcPrefixes.TryGetValue(serviceName, out var prefix) && _mqttClient.IsConnected)
{
try
{
var liveSettings = await _mqttClient.SendRpcRequestAsync, string>(
$"{prefix}_settings_GetAll",
"",
TimeSpan.FromSeconds(2));
if (liveSettings != null && liveSettings.Count > 0)
{
_inMemorySettings[serviceName] = liveSettings.Select(d => new ServiceConfigItem(
Key: d.Key,
Value: d.Value?.ToString() ?? "",
DataType: d.Type,
Description: d.Description,
UpdatedAt: d.UpdatedAt ?? DateTime.UtcNow
)).ToList();
}
}
catch (Exception ex)
{
_logger.LogWarning(ex, "[AdminSettings] Could not fetch live settings for {ServiceName} via MQTT.", serviceName);
}
}
if (_inMemorySettings.TryGetValue(serviceName, out var list))
{
var dtos = list.Select(item => new ServiceConfigItemResponseDto(
Key: item.Key,
Value: item.Value,
DataType: item.DataType,
Description: item.Description,
UpdatedAt: item.UpdatedAt
)).ToList();
return Ok(dtos);
}
return Ok(new List());
}
///
/// Broadcasts configuration settings to the targeted microservice via MQTT RPC and updates in-memory cache.
///
[HttpPut("{serviceName}")]
public async Task UpdateServiceSettings(string serviceName, [FromBody] Dictionary updatedValues)
{
if (updatedValues == null || !updatedValues.Any())
{
return BadRequest(new { message = "No settings provided for update." });
}
_logger.LogInformation("[AdminSettings] Transmitting {Count} config settings to microservice '{ServiceName}' via MQTT", updatedValues.Count, serviceName);
bool mqttPublished = false;
try
{
if (_mqttClient.IsConnected && ServiceRpcPrefixes.TryGetValue(serviceName, out var prefix))
{
var updated = await _mqttClient.SendRpcRequestAsync, Dictionary>(
$"{prefix}_settings_Update",
updatedValues,
TimeSpan.FromSeconds(3));
if (updated != null && updated.Count > 0)
{
_inMemorySettings[serviceName] = updated.Select(d => new ServiceConfigItem(
Key: d.Key,
Value: d.Value?.ToString() ?? "",
DataType: d.Type,
Description: d.Description,
UpdatedAt: d.UpdatedAt ?? DateTime.UtcNow
)).ToList();
}
string topic = $"services/config/updated/{serviceName}";
var payload = new ServiceConfigUpdatePayload(
ServiceName: serviceName,
Timestamp: DateTime.UtcNow,
Settings: updatedValues.ToDictionary(kv => kv.Key, kv => kv.Value?.ToString() ?? "")
);
await _mqttClient.PublishAsync(topic, payload);
mqttPublished = true;
_logger.LogInformation("[AdminSettings] Broadcasted config update event to MQTT topic '{Topic}' for microservice persistence.", topic);
}
}
catch (Exception ex)
{
_logger.LogError(ex, "[AdminSettings] Failed to publish MQTT config update event for service '{ServiceName}'", serviceName);
}
return Ok(new
{
message = $"Settings successfully transmitted to microservice {serviceName}.",
mqttDispatched = mqttPublished
});
}
}