feat(persistence): executions + node_executions migrations and repositories

src/VolcanicCore.Api/Program.cs

@@ -74,10 +74,14 @@ builder.Services.AddSingleton<Migrator>(_ => new Migrator(new IMigration[]
     new Migration001Initial(),
     new Migration002Connections(),
     new Migration003Workflows(),
+    new Migration004Executions(),
+    new Migration005NodeExecutions(),
 }));
 builder.Services.AddSingleton<PasswordService>();
 builder.Services.AddSingleton<ConnectionRepository>();
 builder.Services.AddSingleton<WorkflowRepository>();
+builder.Services.AddSingleton<ExecutionRepository>();
+builder.Services.AddSingleton<NodeExecutionRepository>();
 builder.Services.AddSingleton<IConnectionSecretPolicy, SchemaSecretPolicy>();
 
 builder.Services

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();
+    }
+}

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();
+    }
+}

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<ExecutionRow> ListForWorkflow(string workflowId, int limit)
+    {
+        var rows = new List<ExecutionRow>();
+        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));
+}

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<NodeExecutionRow> ListByExecution(string executionId)
+    {
+        var rows = new List<NodeExecutionRow>();
+        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;
+    }
+}

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<string>();
+        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);
+    }
+}

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));
+    }
+}