Joseph Doherty
596 lines
29 KiB
C#
596 lines
29 KiB
C#
using Microsoft.Data.Sqlite;
|
|
using Microsoft.Extensions.Logging;
|
|
using ZB.MOM.WW.LocalDb;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.StoreAndForward;
|
|
|
|
/// <summary>
|
|
/// SQLite persistence layer for store-and-forward messages.
|
|
/// Uses direct Microsoft.Data.Sqlite (not EF Core) for lightweight site-side storage.
|
|
/// No max buffer size per design decision.
|
|
/// </summary>
|
|
public class StoreAndForwardStorage
|
|
{
|
|
private readonly ILocalDb _localDb;
|
|
private readonly ILogger<StoreAndForwardStorage> _logger;
|
|
|
|
/// <summary>
|
|
/// Initializes the store over the consolidated site database and applies the schema.
|
|
/// </summary>
|
|
/// <param name="localDb">
|
|
/// The consolidated site database. Every connection it hands out is already open and carries
|
|
/// the per-connection pragmas (including <c>busy_timeout</c>) plus the <c>zb_hlc_next()</c>
|
|
/// UDF that <c>sf_messages</c>' capture triggers call — which is exactly why the store no
|
|
/// longer opens its own <see cref="SqliteConnection"/> from a connection string. A raw
|
|
/// connection would lack the UDF and every write to the replicated table would fail closed.
|
|
/// </param>
|
|
/// <param name="logger">Logger for diagnostics.</param>
|
|
public StoreAndForwardStorage(ILocalDb localDb, ILogger<StoreAndForwardStorage> logger)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(localDb);
|
|
ArgumentNullException.ThrowIfNull(logger);
|
|
_localDb = localDb;
|
|
_logger = logger;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Creates the sf_messages table if it does not exist.
|
|
/// </summary>
|
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
public Task InitializeAsync()
|
|
{
|
|
// No directory creation and no journal-mode pragma here any more: LocalDb owns
|
|
// the file (it creates the directory) and sets WAL plus the per-connection
|
|
// pragmas on every connection it hands out.
|
|
using var connection = OpenConnection();
|
|
|
|
// The DDL itself lives in StoreAndForwardSchema so the Host can apply it to a
|
|
// LocalDb-managed connection before RegisterReplicated installs the capture
|
|
// triggers. The store still calls it, so a directly-constructed store (tests,
|
|
// tooling) remains self-sufficient.
|
|
StoreAndForwardSchema.Apply(connection);
|
|
|
|
_logger.LogInformation("Store-and-forward SQLite storage initialized");
|
|
return Task.CompletedTask;
|
|
}
|
|
|
|
/// <summary>
|
|
/// CreateConnection returns an ALREADY-OPEN, pragma-configured connection with the
|
|
/// <c>zb_hlc_next()</c> UDF registered. Calling <c>OpenAsync</c> on it throws, and a raw
|
|
/// <see cref="SqliteConnection"/> would lack the UDF, making every capture trigger fail
|
|
/// closed.
|
|
/// </summary>
|
|
private SqliteConnection OpenConnection() => _localDb.CreateConnection();
|
|
|
|
/// <summary>
|
|
/// INSERT statement for a full message row. Shared by <see cref="EnqueueAsync"/>
|
|
/// and <see cref="UpsertMessageAsync"/>; bind with <see cref="BindMessageParameters"/>
|
|
/// so the column list and the parameters never drift apart.
|
|
/// </summary>
|
|
private const string InsertMessageSql = @"
|
|
INSERT INTO sf_messages (id, category, target, payload_json, retry_count, max_retries,
|
|
retry_interval_ms, created_at, last_attempt_at, last_attempt_at_ms, status, last_error,
|
|
origin_instance, execution_id, source_script, parent_execution_id)
|
|
VALUES (@id, @category, @target, @payload, @retryCount, @maxRetries,
|
|
@retryIntervalMs, @createdAt, @lastAttempt, @lastAttemptMs, @status, @lastError,
|
|
@origin, @executionId, @sourceScript, @parentExecutionId)";
|
|
|
|
/// <summary>
|
|
/// Binds every column parameter for a full message row (used by
|
|
/// <see cref="InsertMessageSql"/>). GUID ids are stored in canonical "D" string
|
|
/// form and the epoch-ms sibling of last_attempt_at is derived here so a single
|
|
/// place owns the mapping.
|
|
/// </summary>
|
|
private static void BindMessageParameters(SqliteCommand cmd, StoreAndForwardMessage message)
|
|
{
|
|
cmd.Parameters.AddWithValue("@id", message.Id);
|
|
cmd.Parameters.AddWithValue("@category", (int)message.Category);
|
|
cmd.Parameters.AddWithValue("@target", message.Target);
|
|
cmd.Parameters.AddWithValue("@payload", message.PayloadJson);
|
|
cmd.Parameters.AddWithValue("@retryCount", message.RetryCount);
|
|
cmd.Parameters.AddWithValue("@maxRetries", message.MaxRetries);
|
|
cmd.Parameters.AddWithValue("@retryIntervalMs", message.RetryIntervalMs);
|
|
cmd.Parameters.AddWithValue("@createdAt", message.CreatedAt.ToString("O"));
|
|
cmd.Parameters.AddWithValue("@lastAttempt", message.LastAttemptAt.HasValue
|
|
? message.LastAttemptAt.Value.ToString("O") : DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@lastAttemptMs",
|
|
(object?)message.LastAttemptAt?.ToUnixTimeMilliseconds() ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@status", (int)message.Status);
|
|
cmd.Parameters.AddWithValue("@lastError", (object?)message.LastError ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@origin", (object?)message.OriginInstanceName ?? DBNull.Value);
|
|
// GUID ids stored as canonical "D" strings; null when not threaded.
|
|
cmd.Parameters.AddWithValue("@executionId",
|
|
message.ExecutionId.HasValue ? message.ExecutionId.Value.ToString("D") : DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@sourceScript", (object?)message.SourceScript ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@parentExecutionId",
|
|
message.ParentExecutionId.HasValue ? message.ParentExecutionId.Value.ToString("D") : DBNull.Value);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Enqueues a new message with Pending status.
|
|
/// </summary>
|
|
/// <param name="message">The message to enqueue.</param>
|
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
public async Task EnqueueAsync(StoreAndForwardMessage message)
|
|
{
|
|
await using var connection = OpenConnection();
|
|
|
|
await using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = InsertMessageSql;
|
|
BindMessageParameters(cmd, message);
|
|
|
|
await cmd.ExecuteNonQueryAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Inserts a message, or updates every mutable column in place if a row with the
|
|
/// same id already exists (<c>ON CONFLICT(id) DO UPDATE</c>). Used by the standby
|
|
/// node when applying a replicated Add/Park/Requeue: a Park/Requeue whose original
|
|
/// Add was lost self-heals (the full message rides in the operation), and a
|
|
/// duplicate Add (e.g. after an anti-entropy resync) applies newest-wins instead of
|
|
/// violating the primary key. The original <c>created_at</c> is preserved — it
|
|
/// orders the retry sweep and must never be overwritten on update.
|
|
/// </summary>
|
|
/// <param name="message">The message to insert or update.</param>
|
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
public async Task UpsertMessageAsync(StoreAndForwardMessage message)
|
|
{
|
|
await using var connection = OpenConnection();
|
|
|
|
await using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = @"
|
|
INSERT INTO sf_messages (id, category, target, payload_json, retry_count, max_retries,
|
|
retry_interval_ms, created_at, last_attempt_at, last_attempt_at_ms, status, last_error,
|
|
origin_instance, execution_id, source_script, parent_execution_id)
|
|
VALUES (@id, @category, @target, @payload, @retryCount, @maxRetries,
|
|
@retryIntervalMs, @createdAt, @lastAttempt, @lastAttemptMs, @status, @lastError,
|
|
@origin, @executionId, @sourceScript, @parentExecutionId)
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
category = excluded.category,
|
|
target = excluded.target,
|
|
payload_json = excluded.payload_json,
|
|
retry_count = excluded.retry_count,
|
|
max_retries = excluded.max_retries,
|
|
retry_interval_ms = excluded.retry_interval_ms,
|
|
last_attempt_at = excluded.last_attempt_at,
|
|
last_attempt_at_ms = excluded.last_attempt_at_ms,
|
|
status = excluded.status,
|
|
last_error = excluded.last_error,
|
|
origin_instance = excluded.origin_instance,
|
|
execution_id = excluded.execution_id,
|
|
source_script = excluded.source_script,
|
|
parent_execution_id = excluded.parent_execution_id";
|
|
|
|
cmd.Parameters.AddWithValue("@id", message.Id);
|
|
cmd.Parameters.AddWithValue("@category", (int)message.Category);
|
|
cmd.Parameters.AddWithValue("@target", message.Target);
|
|
cmd.Parameters.AddWithValue("@payload", message.PayloadJson);
|
|
cmd.Parameters.AddWithValue("@retryCount", message.RetryCount);
|
|
cmd.Parameters.AddWithValue("@maxRetries", message.MaxRetries);
|
|
cmd.Parameters.AddWithValue("@retryIntervalMs", message.RetryIntervalMs);
|
|
cmd.Parameters.AddWithValue("@createdAt", message.CreatedAt.ToString("O"));
|
|
cmd.Parameters.AddWithValue("@lastAttempt", message.LastAttemptAt.HasValue
|
|
? message.LastAttemptAt.Value.ToString("O") : DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@lastAttemptMs",
|
|
(object?)message.LastAttemptAt?.ToUnixTimeMilliseconds() ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@status", (int)message.Status);
|
|
cmd.Parameters.AddWithValue("@lastError", (object?)message.LastError ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@origin", (object?)message.OriginInstanceName ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@executionId",
|
|
message.ExecutionId.HasValue ? message.ExecutionId.Value.ToString("D") : DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@sourceScript", (object?)message.SourceScript ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@parentExecutionId",
|
|
message.ParentExecutionId.HasValue ? message.ParentExecutionId.Value.ToString("D") : DBNull.Value);
|
|
|
|
await cmd.ExecuteNonQueryAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Gets all messages that are due for retry (Pending status, last attempt older than retry interval).
|
|
/// </summary>
|
|
/// <param name="limit">Maximum number of messages to return, or 0 for no limit.</param>
|
|
/// <returns>A task that resolves to the list of messages due for retry, ordered by creation time ascending.</returns>
|
|
public async Task<List<StoreAndForwardMessage>> GetMessagesForRetryAsync(int limit = 0)
|
|
{
|
|
await using var connection = OpenConnection();
|
|
|
|
await using var cmd = connection.CreateCommand();
|
|
const string cols = @"id, category, target, payload_json, retry_count, max_retries,
|
|
retry_interval_ms, created_at, last_attempt_at, status, last_error, origin_instance,
|
|
execution_id, source_script, parent_execution_id";
|
|
const string due = @"status = @pending
|
|
AND (last_attempt_at_ms IS NULL
|
|
OR retry_interval_ms = 0
|
|
OR (@nowMs - last_attempt_at_ms) >= retry_interval_ms)";
|
|
|
|
if (limit > 0)
|
|
{
|
|
// Page FAIRLY across (category, target) lanes: rank each due row within its
|
|
// lane (oldest-first) and take the page by rank, so one dead target holding
|
|
// hundreds of due rows contributes one row per "round" like every other
|
|
// lane instead of owning the whole page. Within a lane rank order == created_at
|
|
// order, so per-target FIFO is preserved for the service's GroupBy.
|
|
cmd.CommandText = $@"
|
|
SELECT {cols}
|
|
FROM (
|
|
SELECT {cols},
|
|
ROW_NUMBER() OVER (PARTITION BY category, target ORDER BY created_at ASC) AS lane_rank
|
|
FROM sf_messages
|
|
WHERE {due}
|
|
)
|
|
ORDER BY lane_rank ASC, created_at ASC
|
|
LIMIT @limit";
|
|
cmd.Parameters.AddWithValue("@limit", limit);
|
|
}
|
|
else
|
|
{
|
|
cmd.CommandText = $@"
|
|
SELECT {cols}
|
|
FROM sf_messages
|
|
WHERE {due}
|
|
ORDER BY created_at ASC";
|
|
}
|
|
|
|
cmd.Parameters.AddWithValue("@pending", (int)StoreAndForwardMessageStatus.Pending);
|
|
cmd.Parameters.AddWithValue("@nowMs", DateTimeOffset.UtcNow.ToUnixTimeMilliseconds());
|
|
|
|
return await ReadMessagesAsync(cmd);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Updates a message after a delivery attempt.
|
|
/// </summary>
|
|
/// <param name="message">The message with updated retry count, status, and last error.</param>
|
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
public async Task UpdateMessageAsync(StoreAndForwardMessage message)
|
|
{
|
|
await using var connection = OpenConnection();
|
|
|
|
await using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = @"
|
|
UPDATE sf_messages
|
|
SET retry_count = @retryCount,
|
|
last_attempt_at = @lastAttempt,
|
|
last_attempt_at_ms = @lastAttemptMs,
|
|
status = @status,
|
|
last_error = @lastError
|
|
WHERE id = @id";
|
|
|
|
cmd.Parameters.AddWithValue("@id", message.Id);
|
|
cmd.Parameters.AddWithValue("@retryCount", message.RetryCount);
|
|
cmd.Parameters.AddWithValue("@lastAttempt", message.LastAttemptAt.HasValue
|
|
? message.LastAttemptAt.Value.ToString("O") : DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@lastAttemptMs",
|
|
(object?)message.LastAttemptAt?.ToUnixTimeMilliseconds() ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@status", (int)message.Status);
|
|
cmd.Parameters.AddWithValue("@lastError", (object?)message.LastError ?? DBNull.Value);
|
|
|
|
await cmd.ExecuteNonQueryAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Updates a message after a delivery attempt, but only if the row is still
|
|
/// in the expected status. Returns true if the row was updated, false if it had
|
|
/// already been changed (e.g. an operator retried or discarded the message) and so
|
|
/// was skipped.
|
|
/// </summary>
|
|
/// <param name="message">The message with the updated values to persist.</param>
|
|
/// <param name="expectedStatus">The status the row must currently have for the update to proceed.</param>
|
|
/// <returns>A task that resolves to <c>true</c> if the row was updated; <c>false</c> if its status had already changed.</returns>
|
|
public async Task<bool> UpdateMessageIfStatusAsync(
|
|
StoreAndForwardMessage message,
|
|
StoreAndForwardMessageStatus expectedStatus)
|
|
{
|
|
await using var connection = OpenConnection();
|
|
|
|
await using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = @"
|
|
UPDATE sf_messages
|
|
SET retry_count = @retryCount,
|
|
last_attempt_at = @lastAttempt,
|
|
last_attempt_at_ms = @lastAttemptMs,
|
|
status = @status,
|
|
last_error = @lastError
|
|
WHERE id = @id AND status = @expectedStatus";
|
|
|
|
cmd.Parameters.AddWithValue("@id", message.Id);
|
|
cmd.Parameters.AddWithValue("@retryCount", message.RetryCount);
|
|
cmd.Parameters.AddWithValue("@lastAttempt", message.LastAttemptAt.HasValue
|
|
? message.LastAttemptAt.Value.ToString("O") : DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@lastAttemptMs",
|
|
(object?)message.LastAttemptAt?.ToUnixTimeMilliseconds() ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@status", (int)message.Status);
|
|
cmd.Parameters.AddWithValue("@lastError", (object?)message.LastError ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("@expectedStatus", (int)expectedStatus);
|
|
|
|
var rows = await cmd.ExecuteNonQueryAsync();
|
|
return rows > 0;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Removes a successfully delivered message.
|
|
/// </summary>
|
|
/// <param name="messageId">The id of the message to remove.</param>
|
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
public async Task RemoveMessageAsync(string messageId)
|
|
{
|
|
await using var connection = OpenConnection();
|
|
|
|
await using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = "DELETE FROM sf_messages WHERE id = @id";
|
|
cmd.Parameters.AddWithValue("@id", messageId);
|
|
|
|
await cmd.ExecuteNonQueryAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Gets all parked messages, optionally filtered by category, with pagination.
|
|
/// </summary>
|
|
/// <param name="category">Optional category filter; null returns parked messages from all categories.</param>
|
|
/// <param name="pageNumber">1-based page number.</param>
|
|
/// <param name="pageSize">Maximum number of messages to return per page.</param>
|
|
/// <returns>A task that resolves to the page of parked messages and the total count of matching rows.</returns>
|
|
public async Task<(List<StoreAndForwardMessage> Messages, int TotalCount)> GetParkedMessagesAsync(
|
|
StoreAndForwardCategory? category = null,
|
|
int pageNumber = 1,
|
|
int pageSize = 50)
|
|
{
|
|
await using var connection = OpenConnection();
|
|
|
|
await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync();
|
|
|
|
// Count
|
|
await using var countCmd = connection.CreateCommand();
|
|
countCmd.Transaction = transaction;
|
|
countCmd.CommandText = category.HasValue
|
|
? "SELECT COUNT(*) FROM sf_messages WHERE status = @parked AND category = @category"
|
|
: "SELECT COUNT(*) FROM sf_messages WHERE status = @parked";
|
|
countCmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);
|
|
if (category.HasValue) countCmd.Parameters.AddWithValue("@category", (int)category.Value);
|
|
var totalCount = Convert.ToInt32(await countCmd.ExecuteScalarAsync());
|
|
|
|
// Page
|
|
await using var pageCmd = connection.CreateCommand();
|
|
pageCmd.Transaction = transaction;
|
|
var categoryFilter = category.HasValue ? " AND category = @category" : "";
|
|
pageCmd.CommandText = $@"
|
|
SELECT id, category, target, payload_json, retry_count, max_retries,
|
|
retry_interval_ms, created_at, last_attempt_at, status, last_error, origin_instance,
|
|
execution_id, source_script, parent_execution_id
|
|
FROM sf_messages
|
|
WHERE status = @parked{categoryFilter}
|
|
ORDER BY created_at ASC
|
|
LIMIT @limit OFFSET @offset";
|
|
|
|
pageCmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);
|
|
if (category.HasValue) pageCmd.Parameters.AddWithValue("@category", (int)category.Value);
|
|
pageCmd.Parameters.AddWithValue("@limit", pageSize);
|
|
pageCmd.Parameters.AddWithValue("@offset", (pageNumber - 1) * pageSize);
|
|
|
|
var messages = await ReadMessagesAsync(pageCmd);
|
|
|
|
await transaction.CommitAsync();
|
|
return (messages, totalCount);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Moves a parked message back to pending for retry.
|
|
/// </summary>
|
|
/// <param name="messageId">The id of the parked message to move back to Pending.</param>
|
|
/// <returns>A task that resolves to <c>true</c> if the message was found and reset to Pending; <c>false</c> if not found or not in Parked status.</returns>
|
|
public async Task<bool> RetryParkedMessageAsync(string messageId)
|
|
{
|
|
await using var connection = OpenConnection();
|
|
|
|
await using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = @"
|
|
UPDATE sf_messages
|
|
SET status = @pending, retry_count = 0, last_error = NULL,
|
|
last_attempt_at = NULL, last_attempt_at_ms = NULL
|
|
WHERE id = @id AND status = @parked";
|
|
|
|
cmd.Parameters.AddWithValue("@id", messageId);
|
|
cmd.Parameters.AddWithValue("@pending", (int)StoreAndForwardMessageStatus.Pending);
|
|
cmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);
|
|
|
|
var rows = await cmd.ExecuteNonQueryAsync();
|
|
return rows > 0;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Permanently discards a parked message.
|
|
/// </summary>
|
|
/// <param name="messageId">The id of the parked message to discard.</param>
|
|
/// <returns>A task that resolves to <c>true</c> if the message was found and deleted; <c>false</c> if not found or not in Parked status.</returns>
|
|
public async Task<bool> DiscardParkedMessageAsync(string messageId)
|
|
{
|
|
await using var connection = OpenConnection();
|
|
|
|
await using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = "DELETE FROM sf_messages WHERE id = @id AND status = @parked";
|
|
cmd.Parameters.AddWithValue("@id", messageId);
|
|
cmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);
|
|
|
|
var rows = await cmd.ExecuteNonQueryAsync();
|
|
return rows > 0;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Gets buffer depth by category (count of pending messages per category).
|
|
/// </summary>
|
|
/// <returns>A task that resolves to a dictionary mapping each category to its pending message count.</returns>
|
|
public async Task<Dictionary<StoreAndForwardCategory, int>> GetBufferDepthByCategoryAsync()
|
|
{
|
|
await using var connection = OpenConnection();
|
|
|
|
await using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = @"
|
|
SELECT category, COUNT(*) as cnt
|
|
FROM sf_messages
|
|
WHERE status = @pending
|
|
GROUP BY category";
|
|
cmd.Parameters.AddWithValue("@pending", (int)StoreAndForwardMessageStatus.Pending);
|
|
|
|
var result = new Dictionary<StoreAndForwardCategory, int>();
|
|
await using var reader = await cmd.ExecuteReaderAsync();
|
|
while (await reader.ReadAsync())
|
|
{
|
|
var category = (StoreAndForwardCategory)reader.GetInt32(0);
|
|
var count = reader.GetInt32(1);
|
|
result[category] = count;
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies messages are NOT deleted when an instance is deleted.
|
|
/// Returns the count of messages for a given origin instance.
|
|
/// </summary>
|
|
/// <param name="instanceName">The origin instance name to count messages for.</param>
|
|
/// <returns>A task that resolves to the number of messages whose origin instance matches <paramref name="instanceName"/>.</returns>
|
|
public async Task<int> GetMessageCountByOriginInstanceAsync(string instanceName)
|
|
{
|
|
await using var connection = OpenConnection();
|
|
|
|
await using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = @"
|
|
SELECT COUNT(*)
|
|
FROM sf_messages
|
|
WHERE origin_instance = @origin";
|
|
cmd.Parameters.AddWithValue("@origin", instanceName);
|
|
|
|
return Convert.ToInt32(await cmd.ExecuteScalarAsync());
|
|
}
|
|
|
|
/// <summary>
|
|
/// Gets a message by ID.
|
|
/// </summary>
|
|
/// <param name="messageId">The id of the message to retrieve.</param>
|
|
/// <returns>A task that resolves to the matching message, or <c>null</c> if not found.</returns>
|
|
public async Task<StoreAndForwardMessage?> GetMessageByIdAsync(string messageId)
|
|
{
|
|
await using var connection = OpenConnection();
|
|
|
|
await using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = @"
|
|
SELECT id, category, target, payload_json, retry_count, max_retries,
|
|
retry_interval_ms, created_at, last_attempt_at, status, last_error, origin_instance,
|
|
execution_id, source_script, parent_execution_id
|
|
FROM sf_messages
|
|
WHERE id = @id";
|
|
cmd.Parameters.AddWithValue("@id", messageId);
|
|
|
|
var messages = await ReadMessagesAsync(cmd);
|
|
return messages.FirstOrDefault();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Gets the count of parked messages (for health reporting).
|
|
/// </summary>
|
|
/// <returns>A task that resolves to the number of messages currently in Parked status.</returns>
|
|
public async Task<int> GetParkedMessageCountAsync()
|
|
{
|
|
await using var conn = OpenConnection();
|
|
await using var cmd = conn.CreateCommand();
|
|
cmd.CommandText = "SELECT COUNT(*) FROM sf_messages WHERE status = @parked";
|
|
cmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);
|
|
var result = await cmd.ExecuteScalarAsync();
|
|
return Convert.ToInt32(result);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Gets the <c>created_at</c> of the oldest parked message, or <c>null</c> when
|
|
/// no rows are parked. Backs the parked-retention aging signal on the site health
|
|
/// report: parked rows persist until an operator acts, so their <i>age</i> — not
|
|
/// just their count — is what tells an operator a forgotten site's backlog is
|
|
/// growing stale.
|
|
/// </summary>
|
|
/// <returns>A task that resolves to the oldest parked row's creation time, or <c>null</c> if none are parked.</returns>
|
|
public async Task<DateTimeOffset?> GetOldestParkedCreatedAtAsync()
|
|
{
|
|
await using var conn = OpenConnection();
|
|
await using var cmd = conn.CreateCommand();
|
|
cmd.CommandText = "SELECT MIN(created_at) FROM sf_messages WHERE status = @parked";
|
|
cmd.Parameters.AddWithValue("@parked", (int)StoreAndForwardMessageStatus.Parked);
|
|
var result = await cmd.ExecuteScalarAsync();
|
|
return result is null or DBNull
|
|
? null
|
|
: DateTimeOffset.Parse((string)result);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Gets total message count by status.
|
|
/// </summary>
|
|
/// <param name="status">The status to filter by.</param>
|
|
/// <returns>A task that resolves to the count of messages with the specified status.</returns>
|
|
public async Task<int> GetMessageCountByStatusAsync(StoreAndForwardMessageStatus status)
|
|
{
|
|
await using var connection = OpenConnection();
|
|
|
|
await using var cmd = connection.CreateCommand();
|
|
cmd.CommandText = "SELECT COUNT(*) FROM sf_messages WHERE status = @status";
|
|
cmd.Parameters.AddWithValue("@status", (int)status);
|
|
|
|
return Convert.ToInt32(await cmd.ExecuteScalarAsync());
|
|
}
|
|
|
|
private static async Task<List<StoreAndForwardMessage>> ReadMessagesAsync(SqliteCommand cmd)
|
|
{
|
|
var results = new List<StoreAndForwardMessage>();
|
|
await using var reader = await cmd.ExecuteReaderAsync();
|
|
while (await reader.ReadAsync())
|
|
{
|
|
results.Add(new StoreAndForwardMessage
|
|
{
|
|
Id = reader.GetString(0),
|
|
Category = (StoreAndForwardCategory)reader.GetInt32(1),
|
|
Target = reader.GetString(2),
|
|
PayloadJson = reader.GetString(3),
|
|
RetryCount = reader.GetInt32(4),
|
|
MaxRetries = reader.GetInt32(5),
|
|
RetryIntervalMs = reader.GetInt64(6),
|
|
CreatedAt = DateTimeOffset.Parse(reader.GetString(7)),
|
|
LastAttemptAt = reader.IsDBNull(8) ? null : DateTimeOffset.Parse(reader.GetString(8)),
|
|
Status = (StoreAndForwardMessageStatus)reader.GetInt32(9),
|
|
LastError = reader.IsDBNull(10) ? null : reader.GetString(10),
|
|
OriginInstanceName = reader.IsDBNull(11) ? null : reader.GetString(11),
|
|
// Rows persisted before the
|
|
// additive migration have no execution_id / source_script value;
|
|
// IsDBNull guards keep those reading back as null (back-compat).
|
|
// Guid.TryParse (not Parse) guards the retry sweep: a corrupt
|
|
// non-null execution_id is treated as "no execution id" rather
|
|
// than throwing FormatException and aborting the whole sweep.
|
|
ExecutionId = ParseGuidColumn(reader, 12),
|
|
SourceScript = reader.IsDBNull(13) ? null : reader.GetString(13),
|
|
// Rows persisted
|
|
// before the additive migration have no parent_execution_id
|
|
// value; the IsDBNull guard inside ParseGuidColumn keeps those
|
|
// reading back as null (back-compat). Guid.TryParse (not Parse)
|
|
// guards the retry sweep against a corrupt non-null value.
|
|
ParentExecutionId = ParseGuidColumn(reader, 14)
|
|
});
|
|
}
|
|
return results;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Defensively reads a nullable GUID column (<c>execution_id</c> or
|
|
/// <c>parent_execution_id</c>). A <c>null</c> value (legacy pre-migration
|
|
/// rows) and a malformed non-null value both yield <c>null</c> — a corrupt
|
|
/// id must not throw and abort the retry sweep, which reads many rows.
|
|
/// </summary>
|
|
private static Guid? ParseGuidColumn(System.Data.Common.DbDataReader reader, int ordinal)
|
|
{
|
|
if (reader.IsDBNull(ordinal))
|
|
{
|
|
return null;
|
|
}
|
|
|
|
return Guid.TryParse(reader.GetString(ordinal), out var value)
|
|
? value
|
|
: null;
|
|
}
|
|
}
|