158 lines
8.8 KiB
C#
158 lines
8.8 KiB
C#
using Microsoft.Data.Sqlite;
|
|
|
|
namespace WxAgent.Service;
|
|
|
|
public sealed record OperationRecord(string Id, string PrincipalId, string AccountId, string Capability,
|
|
string State, string Stage, string CorrelationId, DateTimeOffset CreatedAt, DateTimeOffset ExpiresAt,
|
|
string? ErrorCode, bool HasSideEffects, string? Details = null);
|
|
|
|
public sealed record OperationSummary(string Id, string AccountId, string Capability,
|
|
string State, string Stage, string CorrelationId, DateTimeOffset CreatedAt, DateTimeOffset ExpiresAt,
|
|
string? ErrorCode, bool HasSideEffects)
|
|
{
|
|
public static OperationSummary From(OperationRecord operation) => new(operation.Id, operation.AccountId,
|
|
operation.Capability, operation.State, operation.Stage, operation.CorrelationId, operation.CreatedAt,
|
|
operation.ExpiresAt, operation.ErrorCode, operation.HasSideEffects);
|
|
}
|
|
|
|
public sealed class OperationStore : IDisposable
|
|
{
|
|
private readonly SqliteConnection database;
|
|
private readonly object gate = new();
|
|
|
|
public OperationStore(ServiceOptions options)
|
|
{
|
|
Directory.CreateDirectory(options.DataDirectory);
|
|
database = new SqliteConnection(new SqliteConnectionStringBuilder
|
|
{ DataSource = Path.Combine(options.DataDirectory, "operations.sqlite"), Mode = SqliteOpenMode.ReadWriteCreate }.ToString());
|
|
database.Open();
|
|
Execute("""
|
|
PRAGMA journal_mode=WAL;
|
|
CREATE TABLE IF NOT EXISTS operations (
|
|
id TEXT PRIMARY KEY, principal TEXT NOT NULL, account TEXT NOT NULL, capability TEXT NOT NULL,
|
|
state TEXT NOT NULL, stage TEXT NOT NULL, correlation TEXT NOT NULL, created TEXT NOT NULL,
|
|
expires TEXT NOT NULL, error TEXT, side_effects INTEGER NOT NULL, idempotency TEXT, digest TEXT NOT NULL,
|
|
UNIQUE(principal, account, capability, idempotency));
|
|
UPDATE operations SET state=CASE WHEN side_effects=1 THEN 'Unconfirmed' ELSE 'Failed' END,
|
|
stage='restart', error='AgentRestarted' WHERE state='Running';
|
|
UPDATE operations SET state='Cancelled', stage='restart', error='AgentRestarted' WHERE state='Queued';
|
|
""");
|
|
try { Execute("ALTER TABLE operations ADD COLUMN details TEXT"); }
|
|
catch (SqliteException exception) when (exception.SqliteErrorCode == 1) { }
|
|
}
|
|
|
|
public (OperationRecord Record, bool Created) Enqueue(string principal, string account, string capability,
|
|
string? idempotency, string digest, bool sideEffects, TimeSpan budget, int capacity, string? details = null)
|
|
{
|
|
lock (gate)
|
|
{
|
|
using var transaction = database.BeginTransaction();
|
|
using var lookup = database.CreateCommand();
|
|
lookup.Transaction = transaction;
|
|
lookup.CommandText = "SELECT *, digest FROM operations WHERE principal=$p AND account=$a AND capability=$c AND idempotency=$k";
|
|
lookup.Parameters.AddWithValue("$p", principal);
|
|
lookup.Parameters.AddWithValue("$a", account);
|
|
lookup.Parameters.AddWithValue("$c", capability);
|
|
lookup.Parameters.AddWithValue("$k", (object?)idempotency ?? DBNull.Value);
|
|
using (var reader = lookup.ExecuteReader())
|
|
{
|
|
if (reader.Read())
|
|
{
|
|
if (reader.GetString(12) != digest) throw new ServiceException("IdempotencyConflict", 409, "Key already used for different parameters.");
|
|
return (Read(reader), false);
|
|
}
|
|
}
|
|
using var count = database.CreateCommand();
|
|
count.Transaction = transaction;
|
|
count.CommandText = "SELECT COUNT(*) FROM operations WHERE state IN ('Queued','Running')";
|
|
if (Convert.ToInt64(count.ExecuteScalar()) >= capacity)
|
|
throw new ServiceException("QueueFull", 429, "Agent queue is full.");
|
|
var now = DateTimeOffset.UtcNow;
|
|
var operation = new OperationRecord(Guid.NewGuid().ToString("N"), principal, account, capability,
|
|
"Queued", "queued", Guid.NewGuid().ToString("N"), now, now.Add(budget), null, sideEffects, details);
|
|
using var insert = database.CreateCommand();
|
|
insert.Transaction = transaction;
|
|
insert.CommandText = "INSERT INTO operations (id,principal,account,capability,state,stage,correlation,created,expires,error,side_effects,idempotency,digest,details) VALUES ($id,$p,$a,$c,'Queued','queued',$correlation,$created,$expires,NULL,$effects,$key,$digest,$details)";
|
|
insert.Parameters.AddWithValue("$id", operation.Id);
|
|
insert.Parameters.AddWithValue("$p", principal);
|
|
insert.Parameters.AddWithValue("$a", account);
|
|
insert.Parameters.AddWithValue("$c", capability);
|
|
insert.Parameters.AddWithValue("$correlation", operation.CorrelationId);
|
|
insert.Parameters.AddWithValue("$created", now.ToString("O"));
|
|
insert.Parameters.AddWithValue("$expires", operation.ExpiresAt.ToString("O"));
|
|
insert.Parameters.AddWithValue("$effects", sideEffects ? 1 : 0);
|
|
insert.Parameters.AddWithValue("$key", (object?)idempotency ?? DBNull.Value);
|
|
insert.Parameters.AddWithValue("$digest", digest);
|
|
insert.Parameters.AddWithValue("$details", (object?)details ?? DBNull.Value);
|
|
insert.ExecuteNonQuery();
|
|
transaction.Commit();
|
|
return (operation, true);
|
|
}
|
|
}
|
|
|
|
public OperationRecord Get(string id, string principal)
|
|
{
|
|
lock (gate)
|
|
{
|
|
using var command = database.CreateCommand();
|
|
command.CommandText = "SELECT * FROM operations WHERE id=$id AND principal=$principal";
|
|
command.Parameters.AddWithValue("$id", id);
|
|
command.Parameters.AddWithValue("$principal", principal);
|
|
using var reader = command.ExecuteReader();
|
|
if (!reader.Read()) throw new ServiceException("NotFound", 404, "Operation not found.");
|
|
return Read(reader);
|
|
}
|
|
}
|
|
|
|
public Page<OperationRecord> List(string principal, string? accountId, int limit, int offset)
|
|
{
|
|
lock (gate)
|
|
{
|
|
using var command = database.CreateCommand();
|
|
command.CommandText = string.IsNullOrWhiteSpace(accountId)
|
|
? "SELECT * FROM operations WHERE principal=$principal ORDER BY created DESC, id DESC LIMIT $limit OFFSET $offset"
|
|
: "SELECT * FROM operations WHERE principal=$principal AND account=$account ORDER BY created DESC, id DESC LIMIT $limit OFFSET $offset";
|
|
command.Parameters.AddWithValue("$principal", principal);
|
|
if (!string.IsNullOrWhiteSpace(accountId)) command.Parameters.AddWithValue("$account", accountId);
|
|
command.Parameters.AddWithValue("$limit", limit + 1);
|
|
command.Parameters.AddWithValue("$offset", offset);
|
|
using var reader = command.ExecuteReader();
|
|
var items = new List<OperationRecord>();
|
|
while (reader.Read()) items.Add(Read(reader));
|
|
var hasMore = items.Count > limit;
|
|
if (hasMore) items.RemoveAt(items.Count - 1);
|
|
return new Page<OperationRecord>(items, limit, offset, hasMore, hasMore ? offset + limit : null);
|
|
}
|
|
}
|
|
|
|
public void Transition(string id, string state, string stage, string? error = null, string? details = null)
|
|
{
|
|
lock (gate)
|
|
{
|
|
using var command = database.CreateCommand();
|
|
command.CommandText = "UPDATE operations SET state=$state,stage=$stage,error=$error,details=COALESCE($details,details) WHERE id=$id AND state IN ('Queued','Running')";
|
|
command.Parameters.AddWithValue("$id", id);
|
|
command.Parameters.AddWithValue("$state", state);
|
|
command.Parameters.AddWithValue("$stage", stage);
|
|
command.Parameters.AddWithValue("$error", (object?)error ?? DBNull.Value);
|
|
command.Parameters.AddWithValue("$details", (object?)details ?? DBNull.Value);
|
|
command.ExecuteNonQuery();
|
|
}
|
|
}
|
|
|
|
private static OperationRecord Read(SqliteDataReader reader) => new(reader.GetString(0), reader.GetString(1), reader.GetString(2),
|
|
reader.GetString(3), reader.GetString(4), reader.GetString(5), reader.GetString(6),
|
|
DateTimeOffset.Parse(reader.GetString(7), System.Globalization.CultureInfo.InvariantCulture),
|
|
DateTimeOffset.Parse(reader.GetString(8), System.Globalization.CultureInfo.InvariantCulture), reader.IsDBNull(9) ? null : reader.GetString(9), reader.GetInt64(10) != 0,
|
|
reader.IsDBNull(13) ? null : reader.GetString(13));
|
|
|
|
private void Execute(string sql)
|
|
{
|
|
using var command = database.CreateCommand();
|
|
command.CommandText = sql;
|
|
command.ExecuteNonQuery();
|
|
}
|
|
|
|
public void Dispose() => database.Dispose();
|
|
}
|