diff --git a/src/VolcanicCore.Api/Program.cs b/src/VolcanicCore.Api/Program.cs index 61ff2ac..67a57a1 100644 --- a/src/VolcanicCore.Api/Program.cs +++ b/src/VolcanicCore.Api/Program.cs @@ -74,10 +74,14 @@ builder.Services.AddSingleton(_ => new Migrator(new IMigration[] new Migration001Initial(), new Migration002Connections(), new Migration003Workflows(), + new Migration004Executions(), + new Migration005NodeExecutions(), })); builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); +builder.Services.AddSingleton(); +builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services diff --git a/src/VolcanicCore.Persistence/Migrations/Migration004Executions.cs b/src/VolcanicCore.Persistence/Migrations/Migration004Executions.cs new file mode 100644 index 0000000..a8b97ba --- /dev/null +++ b/src/VolcanicCore.Persistence/Migrations/Migration004Executions.cs @@ -0,0 +1,27 @@ +using Microsoft.Data.Sqlite; + +namespace VolcanicCore.Persistence.Migrations; + +public sealed class Migration004Executions : IMigration +{ + public int Version => 4; + public void Apply(SqliteConnection conn) + { + using var cmd = conn.CreateCommand(); + cmd.CommandText = @" + CREATE TABLE executions ( + id TEXT PRIMARY KEY, + workflow_id TEXT NOT NULL, + trigger TEXT NOT NULL, -- 'nats.hook' | 'http.webhook' | 'manual' + trigger_data_json TEXT NOT NULL, + status TEXT NOT NULL, -- 'running' | 'completed' | 'partial' | 'failed' | 'aborted' + started_at TEXT NOT NULL, + ended_at TEXT, + error TEXT + ); + CREATE INDEX ix_executions_workflow_started ON executions (workflow_id, started_at DESC); + CREATE INDEX ix_executions_ended_at ON executions (ended_at); + "; + cmd.ExecuteNonQuery(); + } +} diff --git a/src/VolcanicCore.Persistence/Migrations/Migration005NodeExecutions.cs b/src/VolcanicCore.Persistence/Migrations/Migration005NodeExecutions.cs new file mode 100644 index 0000000..aed664e --- /dev/null +++ b/src/VolcanicCore.Persistence/Migrations/Migration005NodeExecutions.cs @@ -0,0 +1,27 @@ +using Microsoft.Data.Sqlite; + +namespace VolcanicCore.Persistence.Migrations; + +public sealed class Migration005NodeExecutions : IMigration +{ + public int Version => 5; + public void Apply(SqliteConnection conn) + { + using var cmd = conn.CreateCommand(); + cmd.CommandText = @" + CREATE TABLE node_executions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + execution_id TEXT NOT NULL REFERENCES executions(id) ON DELETE CASCADE, + node_id TEXT NOT NULL, + inputs_json TEXT NOT NULL, -- [[items on port 0], [items on port 1], ...] or [] + outputs_json TEXT NOT NULL, -- [[items on out 0], ...] or [] + status TEXT NOT NULL, -- 'completed' | 'failed' | 'skipped_due_to_failure' + started_at TEXT NOT NULL, + ended_at TEXT NOT NULL, + error TEXT + ); + CREATE INDEX ix_node_executions_by_execution ON node_executions (execution_id); + "; + cmd.ExecuteNonQuery(); + } +} diff --git a/src/VolcanicCore.Persistence/Repositories/ExecutionRepository.cs b/src/VolcanicCore.Persistence/Repositories/ExecutionRepository.cs new file mode 100644 index 0000000..a070e2e --- /dev/null +++ b/src/VolcanicCore.Persistence/Repositories/ExecutionRepository.cs @@ -0,0 +1,121 @@ +using Microsoft.Data.Sqlite; + +namespace VolcanicCore.Persistence.Repositories; + +public sealed record ExecutionRow( + string Id, string WorkflowId, string Trigger, string TriggerDataJson, + string Status, DateTimeOffset StartedAt, DateTimeOffset? EndedAt, string? Error); + +public sealed class ExecutionRepository +{ + private readonly SqliteConnectionFactory _factory; + public ExecutionRepository(SqliteConnectionFactory f) => _factory = f; + + public string Create(string workflowId, string trigger, string triggerDataJson) + { + var id = Guid.NewGuid().ToString("N"); + var now = DateTimeOffset.UtcNow.ToString("o"); + using var conn = _factory.Open(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = @" + INSERT INTO executions (id, workflow_id, trigger, trigger_data_json, status, started_at) + VALUES ($id, $w, $t, $td, 'running', $now); + "; + cmd.Parameters.AddWithValue("$id", id); + cmd.Parameters.AddWithValue("$w", workflowId); + cmd.Parameters.AddWithValue("$t", trigger); + cmd.Parameters.AddWithValue("$td", triggerDataJson); + cmd.Parameters.AddWithValue("$now", now); + cmd.ExecuteNonQuery(); + return id; + } + + public void SetStatus(string id, string status, string? error, DateTimeOffset? endedAt) + { + using var conn = _factory.Open(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = @"UPDATE executions + SET status = $s, error = $e, ended_at = $end + WHERE id = $id;"; + cmd.Parameters.AddWithValue("$id", id); + cmd.Parameters.AddWithValue("$s", status); + cmd.Parameters.AddWithValue("$e", (object?)error ?? DBNull.Value); + cmd.Parameters.AddWithValue("$end", endedAt is null ? DBNull.Value : (object)endedAt.Value.ToString("o")); + cmd.ExecuteNonQuery(); + } + + public ExecutionRow? Get(string id) + { + using var conn = _factory.Open(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = @"SELECT id, workflow_id, trigger, trigger_data_json, status, + started_at, ended_at, error + FROM executions WHERE id = $id;"; + cmd.Parameters.AddWithValue("$id", id); + using var r = cmd.ExecuteReader(); + return r.Read() ? Materialize(r) : null; + } + + public IReadOnlyList ListForWorkflow(string workflowId, int limit) + { + var rows = new List(); + using var conn = _factory.Open(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = @"SELECT id, workflow_id, trigger, trigger_data_json, status, + started_at, ended_at, error + FROM executions + WHERE workflow_id = $w + ORDER BY started_at DESC + LIMIT $lim;"; + cmd.Parameters.AddWithValue("$w", workflowId); + cmd.Parameters.AddWithValue("$lim", limit); + using var r = cmd.ExecuteReader(); + while (r.Read()) rows.Add(Materialize(r)); + return rows; + } + + public int AbortAllRunning() + { + using var conn = _factory.Open(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = @"UPDATE executions + SET status = 'aborted', ended_at = $now + WHERE status = 'running';"; + cmd.Parameters.AddWithValue("$now", DateTimeOffset.UtcNow.ToString("o")); + return cmd.ExecuteNonQuery(); + } + + public int PruneOlderThan(DateTimeOffset cutoff) + { + using var conn = _factory.Open(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = @"DELETE FROM executions + WHERE ended_at IS NOT NULL AND ended_at < $c;"; + cmd.Parameters.AddWithValue("$c", cutoff.ToString("o")); + return cmd.ExecuteNonQuery(); + } + + public int PruneOldestPerWorkflow(int keep) + { + // For each workflow, delete rows that are not in the newest `keep` by started_at. + using var conn = _factory.Open(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = @" + DELETE FROM executions WHERE id IN ( + SELECT id FROM ( + SELECT id, ROW_NUMBER() OVER ( + PARTITION BY workflow_id ORDER BY started_at DESC + ) AS rn + FROM executions + ) WHERE rn > $keep + );"; + cmd.Parameters.AddWithValue("$keep", keep); + return cmd.ExecuteNonQuery(); + } + + private static ExecutionRow Materialize(SqliteDataReader r) => new( + r.GetString(0), r.GetString(1), r.GetString(2), r.GetString(3), r.GetString(4), + DateTimeOffset.Parse(r.GetString(5)), + r.IsDBNull(6) ? null : DateTimeOffset.Parse(r.GetString(6)), + r.IsDBNull(7) ? null : r.GetString(7)); +} diff --git a/src/VolcanicCore.Persistence/Repositories/NodeExecutionRepository.cs b/src/VolcanicCore.Persistence/Repositories/NodeExecutionRepository.cs new file mode 100644 index 0000000..6d8eede --- /dev/null +++ b/src/VolcanicCore.Persistence/Repositories/NodeExecutionRepository.cs @@ -0,0 +1,58 @@ +using Microsoft.Data.Sqlite; + +namespace VolcanicCore.Persistence.Repositories; + +public sealed record NodeExecutionRow( + long Id, string ExecutionId, string NodeId, + string InputsJson, string OutputsJson, + string Status, DateTimeOffset StartedAt, DateTimeOffset EndedAt, string? Error); + +public sealed class NodeExecutionRepository +{ + private readonly SqliteConnectionFactory _factory; + public NodeExecutionRepository(SqliteConnectionFactory f) => _factory = f; + + public void Create(string executionId, string nodeId, + string inputsJson, string outputsJson, + string status, DateTimeOffset startedAt, DateTimeOffset endedAt, string? error) + { + using var conn = _factory.Open(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = @" + INSERT INTO node_executions + (execution_id, node_id, inputs_json, outputs_json, status, started_at, ended_at, error) + VALUES ($eid, $nid, $inp, $out, $s, $sa, $ea, $err); + "; + cmd.Parameters.AddWithValue("$eid", executionId); + cmd.Parameters.AddWithValue("$nid", nodeId); + cmd.Parameters.AddWithValue("$inp", inputsJson); + cmd.Parameters.AddWithValue("$out", outputsJson); + cmd.Parameters.AddWithValue("$s", status); + cmd.Parameters.AddWithValue("$sa", startedAt.ToString("o")); + cmd.Parameters.AddWithValue("$ea", endedAt.ToString("o")); + cmd.Parameters.AddWithValue("$err", (object?)error ?? DBNull.Value); + cmd.ExecuteNonQuery(); + } + + public IReadOnlyList ListByExecution(string executionId) + { + var rows = new List(); + using var conn = _factory.Open(); + using var cmd = conn.CreateCommand(); + cmd.CommandText = @"SELECT id, execution_id, node_id, inputs_json, outputs_json, + status, started_at, ended_at, error + FROM node_executions + WHERE execution_id = $eid + ORDER BY started_at ASC, id ASC;"; + cmd.Parameters.AddWithValue("$eid", executionId); + using var r = cmd.ExecuteReader(); + while (r.Read()) + rows.Add(new NodeExecutionRow( + r.GetInt64(0), r.GetString(1), r.GetString(2), r.GetString(3), r.GetString(4), + r.GetString(5), + DateTimeOffset.Parse(r.GetString(6)), + DateTimeOffset.Parse(r.GetString(7)), + r.IsDBNull(8) ? null : r.GetString(8))); + return rows; + } +} diff --git a/tests/VolcanicCore.Tests/ExecutionRepositoryTests.cs b/tests/VolcanicCore.Tests/ExecutionRepositoryTests.cs new file mode 100644 index 0000000..654ecc5 --- /dev/null +++ b/tests/VolcanicCore.Tests/ExecutionRepositoryTests.cs @@ -0,0 +1,102 @@ +using System.Security.Cryptography; +using VolcanicCore.Persistence; +using VolcanicCore.Persistence.Encryption; +using VolcanicCore.Persistence.Migrations; +using VolcanicCore.Persistence.Repositories; + +namespace VolcanicCore.Tests; + +public class ExecutionRepositoryTests +{ + private static ExecutionRepository Setup() + { + var path = Path.Combine(Path.GetTempPath(), $"vc-{Guid.NewGuid():N}.db"); + var f = new SqliteConnectionFactory(path); + using (var conn = f.Open()) + new Migrator(new IMigration[] + { + new Migration001Initial(), + new Migration002Connections(), + new Migration003Workflows(), + new Migration004Executions(), + }).Apply(conn); + return new ExecutionRepository(f); + } + + [Fact] + public void Create_and_get_round_trip() + { + var repo = Setup(); + var id = repo.Create("wf1", "nats", "{\"subject\":\"foo\"}"); + var row = repo.Get(id); + Assert.NotNull(row); + Assert.Equal("wf1", row!.WorkflowId); + Assert.Equal("nats", row.Trigger); + Assert.Equal("running", row.Status); + Assert.Null(row.EndedAt); + } + + [Fact] + public void SetStatus_updates_row() + { + var repo = Setup(); + var id = repo.Create("wf1", "manual", "{}"); + repo.SetStatus(id, "completed", error: null, endedAt: DateTimeOffset.UtcNow); + var row = repo.Get(id)!; + Assert.Equal("completed", row.Status); + Assert.NotNull(row.EndedAt); + } + + [Fact] + public void AbortAllRunning_flips_running_rows_to_aborted() + { + var repo = Setup(); + var a = repo.Create("wf1", "nats", "{}"); + var b = repo.Create("wf1", "manual", "{}"); + repo.SetStatus(b, "completed", null, DateTimeOffset.UtcNow); + var n = repo.AbortAllRunning(); + Assert.Equal(1, n); + Assert.Equal("aborted", repo.Get(a)!.Status); + Assert.Equal("completed", repo.Get(b)!.Status); + } + + [Fact] + public void ListForWorkflow_returns_desc_by_started_at_capped_by_limit() + { + var repo = Setup(); + var ids = new List(); + for (var i = 0; i < 5; i++) { ids.Add(repo.Create("wf1", "manual", "{}")); Thread.Sleep(5); } + var got = repo.ListForWorkflow("wf1", limit: 3); + Assert.Equal(3, got.Count); + Assert.Equal(ids[^1], got[0].Id); // newest first + } + + [Fact] + public void PruneOlderThan_deletes_by_ended_at_cutoff() + { + var repo = Setup(); + var old = repo.Create("wf1", "manual", "{}"); + repo.SetStatus(old, "completed", null, DateTimeOffset.UtcNow.AddDays(-30)); + var young = repo.Create("wf1", "manual", "{}"); + repo.SetStatus(young, "completed", null, DateTimeOffset.UtcNow); + + var deleted = repo.PruneOlderThan(DateTimeOffset.UtcNow.AddDays(-7)); + Assert.Equal(1, deleted); + Assert.Null(repo.Get(old)); + Assert.NotNull(repo.Get(young)); + } + + [Fact] + public void PruneOldestPerWorkflow_keeps_last_N_per_workflow() + { + var repo = Setup(); + for (var i = 0; i < 10; i++) + { + var id = repo.Create("wf1", "manual", "{}"); + repo.SetStatus(id, "completed", null, DateTimeOffset.UtcNow.AddSeconds(i)); + } + var deleted = repo.PruneOldestPerWorkflow(keep: 3); + Assert.Equal(7, deleted); + Assert.Equal(3, repo.ListForWorkflow("wf1", limit: 100).Count); + } +} diff --git a/tests/VolcanicCore.Tests/NodeExecutionRepositoryTests.cs b/tests/VolcanicCore.Tests/NodeExecutionRepositoryTests.cs new file mode 100644 index 0000000..1575cb0 --- /dev/null +++ b/tests/VolcanicCore.Tests/NodeExecutionRepositoryTests.cs @@ -0,0 +1,50 @@ +using VolcanicCore.Persistence; +using VolcanicCore.Persistence.Migrations; +using VolcanicCore.Persistence.Repositories; + +namespace VolcanicCore.Tests; + +public class NodeExecutionRepositoryTests +{ + private static (ExecutionRepository ex, NodeExecutionRepository ne) Setup() + { + var path = Path.Combine(Path.GetTempPath(), $"vc-{Guid.NewGuid():N}.db"); + var f = new SqliteConnectionFactory(path); + using (var conn = f.Open()) + new Migrator(new IMigration[] + { + new Migration001Initial(), + new Migration002Connections(), + new Migration003Workflows(), + new Migration004Executions(), + new Migration005NodeExecutions(), + }).Apply(conn); + return (new ExecutionRepository(f), new NodeExecutionRepository(f)); + } + + [Fact] + public void Create_and_list_by_execution() + { + var (ex, ne) = Setup(); + var eid = ex.Create("wf1", "manual", "{}"); + var start = DateTimeOffset.UtcNow; + ne.Create(eid, "n1", "[]", "[[]]", "completed", start, start.AddMilliseconds(3), null); + ne.Create(eid, "n2", "[[]]", "[]", "failed", start, start.AddMilliseconds(5), "boom"); + + var list = ne.ListByExecution(eid); + Assert.Equal(2, list.Count); + Assert.Contains(list, x => x.NodeId == "n1" && x.Status == "completed"); + Assert.Contains(list, x => x.NodeId == "n2" && x.Status == "failed" && x.Error == "boom"); + } + + [Fact] + public void Cascade_delete_removes_rows_when_execution_deleted() + { + var (ex, ne) = Setup(); + var eid = ex.Create("wf1", "manual", "{}"); + ne.Create(eid, "n1", "[]", "[[]]", "completed", DateTimeOffset.UtcNow, DateTimeOffset.UtcNow, null); + ex.SetStatus(eid, "completed", null, DateTimeOffset.UtcNow); + Assert.Equal(1, ex.PruneOldestPerWorkflow(keep: 0)); + Assert.Empty(ne.ListByExecution(eid)); + } +}