265 lines
14 KiB
C#
265 lines
14 KiB
C#
using System.Data;
|
|
using System.Globalization;
|
|
using Microsoft.EntityFrameworkCore;
|
|
|
|
namespace w4c_workflows.Data;
|
|
|
|
/// <summary>
|
|
/// Applies the workflow database schema at startup: the recorded EF Core
|
|
/// migrations followed by the additive columns/tables that intentionally have no
|
|
/// migration (project house rule — added ones ship as idempotent raw SQL).
|
|
///
|
|
/// Two providers are supported:
|
|
/// * <b>PostgreSQL</b> (full mode) — schema-qualified <c>IF NOT EXISTS</c>
|
|
/// statements, safe to re-run against long-lived databases.
|
|
/// * <b>SQLite</b> (self-hosted Lite mode) — no schemas and no
|
|
/// <c>ALTER TABLE … ADD COLUMN IF NOT EXISTS</c>, so columns are added only
|
|
/// after a <c>pragma_table_info</c> existence probe, and tables/indexes use
|
|
/// <c>IF NOT EXISTS</c>.
|
|
///
|
|
/// Keeping the statements here (instead of inline in <c>Program.cs</c>) lets the
|
|
/// test fixtures apply exactly the schema the production host does, so a green
|
|
/// fixture genuinely proves the host schema.
|
|
/// </summary>
|
|
public static class WorkflowsSchema
|
|
{
|
|
/// <summary>
|
|
/// Ensures the schema exists: applies EF migrations, then the provider-specific
|
|
/// additive statements. Idempotent on every call.
|
|
/// </summary>
|
|
public static async Task ApplyAsync(
|
|
WorkflowsDbContext db, bool liteMode, CancellationToken ct = default)
|
|
{
|
|
await db.Database.MigrateAsync(ct);
|
|
|
|
if (liteMode)
|
|
await ApplyLiteAsync(db, ct);
|
|
else
|
|
await ApplyFullAsync(db, ct);
|
|
}
|
|
|
|
// ================================================================== PostgreSQL
|
|
|
|
/// <summary>
|
|
/// Additive Postgres columns/tables, applied after migrations. Public so the
|
|
/// Postgres test fixture can reuse them verbatim.
|
|
/// </summary>
|
|
public static IReadOnlyList<string> FullStatements { get; } = new[]
|
|
{
|
|
// Task columns added after the original migration (W7 archive, S8 server).
|
|
"ALTER TABLE workflows.\"Tasks\" ADD COLUMN IF NOT EXISTS \"ArchivedAt\" timestamptz NULL;",
|
|
"ALTER TABLE workflows.\"Tasks\" ADD COLUMN IF NOT EXISTS \"Server\" text NULL;",
|
|
// Node-mode task columns + the edge table (node graph kernel).
|
|
"ALTER TABLE workflows.\"Tasks\" ADD COLUMN IF NOT EXISTS \"NodeType\" text NULL;",
|
|
"ALTER TABLE workflows.\"Tasks\" ADD COLUMN IF NOT EXISTS \"NodeVersion\" double precision NULL;",
|
|
"ALTER TABLE workflows.\"Tasks\" ADD COLUMN IF NOT EXISTS \"ParametersJson\" jsonb NULL;",
|
|
"ALTER TABLE workflows.\"Tasks\" ADD COLUMN IF NOT EXISTS \"CredentialsJson\" jsonb NULL;",
|
|
"ALTER TABLE workflows.\"Tasks\" ADD COLUMN IF NOT EXISTS \"RetryJson\" jsonb NULL;",
|
|
"ALTER TABLE workflows.\"Tasks\" ADD COLUMN IF NOT EXISTS \"ContinueOnFail\" boolean NOT NULL DEFAULT FALSE;",
|
|
"ALTER TABLE workflows.\"Tasks\" ADD COLUMN IF NOT EXISTS \"RunMode\" text NULL;",
|
|
"CREATE TABLE IF NOT EXISTS workflows.\"WorkflowTaskEdges\" (" +
|
|
"\"Id\" uuid NOT NULL, " +
|
|
"\"WorkflowId\" uuid NOT NULL, " +
|
|
"\"FromTaskId\" uuid NOT NULL, " +
|
|
"\"FromOutput\" integer NOT NULL, " +
|
|
"\"ToTaskId\" uuid NOT NULL, " +
|
|
"\"ToInput\" integer NOT NULL, " +
|
|
"\"IsLoopBack\" boolean NOT NULL DEFAULT FALSE, " +
|
|
"CONSTRAINT \"PK_WorkflowTaskEdges\" PRIMARY KEY (\"Id\"), " +
|
|
"CONSTRAINT \"FK_WorkflowTaskEdges_Workflows_WorkflowId\" FOREIGN KEY (\"WorkflowId\") " +
|
|
"REFERENCES workflows.\"Workflows\" (\"Id\") ON DELETE CASCADE);",
|
|
"ALTER TABLE workflows.\"WorkflowTaskEdges\" ADD COLUMN IF NOT EXISTS \"IsLoopBack\" boolean NOT NULL DEFAULT FALSE;",
|
|
"CREATE INDEX IF NOT EXISTS \"IX_WorkflowTaskEdges_WorkflowId\" ON workflows.\"WorkflowTaskEdges\" (\"WorkflowId\");",
|
|
// Encrypted credential vault (Phase 3).
|
|
"CREATE TABLE IF NOT EXISTS workflows.\"Credentials\" (" +
|
|
"\"Id\" uuid NOT NULL, " +
|
|
"\"TenantId\" character varying(120) NOT NULL, " +
|
|
"\"Name\" character varying(200) NOT NULL, " +
|
|
"\"Type\" character varying(100) NOT NULL, " +
|
|
"\"EncryptedData\" text NOT NULL, " +
|
|
"\"CreatedAt\" timestamp with time zone NOT NULL, " +
|
|
"\"UpdatedAt\" timestamp with time zone NOT NULL, " +
|
|
"CONSTRAINT \"PK_Credentials\" PRIMARY KEY (\"Id\"));",
|
|
"CREATE UNIQUE INDEX IF NOT EXISTS \"IX_Credentials_TenantId_Name\" ON workflows.\"Credentials\" (\"TenantId\", \"Name\");",
|
|
"CREATE INDEX IF NOT EXISTS \"IX_Credentials_TenantId\" ON workflows.\"Credentials\" (\"TenantId\");",
|
|
// Sub-workflow plumbing on runs (Phase 7): soft references (no FK).
|
|
"ALTER TABLE workflows.\"WorkflowRuns\" ADD COLUMN IF NOT EXISTS \"ParentRunId\" uuid NULL;",
|
|
"ALTER TABLE workflows.\"WorkflowRuns\" ADD COLUMN IF NOT EXISTS \"ParentTaskId\" uuid NULL;",
|
|
"ALTER TABLE workflows.\"WorkflowRuns\" ADD COLUMN IF NOT EXISTS \"Depth\" integer NOT NULL DEFAULT 0;",
|
|
"CREATE INDEX IF NOT EXISTS \"IX_WorkflowRuns_ParentRunId\" ON workflows.\"WorkflowRuns\" (\"ParentRunId\");",
|
|
// Per-tenant workflow repo setting + which repo a compiled workflow came from.
|
|
"ALTER TABLE workflows.\"Workflows\" ADD COLUMN IF NOT EXISTS \"Repo\" text NOT NULL DEFAULT 'workflows';",
|
|
// Denormalized webhook path for the public receiver's indexed lookup.
|
|
"ALTER TABLE workflows.\"Workflows\" ADD COLUMN IF NOT EXISTS \"WebhookPath\" text NULL;",
|
|
"CREATE INDEX IF NOT EXISTS \"IX_Workflows_TenantId_WebhookPath\" ON workflows.\"Workflows\" (\"TenantId\", \"WebhookPath\") WHERE \"WebhookPath\" IS NOT NULL;",
|
|
"CREATE TABLE IF NOT EXISTS workflows.\"WorkflowRepos\" (" +
|
|
"\"TenantId\" text NOT NULL, \"RepoName\" text NOT NULL, \"UpdatedAt\" timestamptz NULL, " +
|
|
"CONSTRAINT \"PK_WorkflowRepos\" PRIMARY KEY (\"TenantId\"));",
|
|
// Workflow runtime registry (platform row seeded per tenant; self-hosted
|
|
// runtimes added by tenants).
|
|
"CREATE TABLE IF NOT EXISTS workflows.\"WorkflowRuntimes\" (" +
|
|
"\"Id\" uuid NOT NULL, " +
|
|
"\"TenantId\" text NOT NULL, " +
|
|
"\"Label\" text NOT NULL, " +
|
|
"\"Kind\" text NOT NULL, " +
|
|
"\"Endpoint\" text NULL, " +
|
|
"\"ApiKeyHash\" text NULL, " +
|
|
"\"SecretHash\" text NULL, " +
|
|
"\"Status\" text NOT NULL DEFAULT 'unknown', " +
|
|
"\"LastSeenAt\" timestamptz NULL, " +
|
|
"\"LastError\" text NULL, " +
|
|
"\"InfoJson\" jsonb NULL, " +
|
|
"\"IsDefault\" boolean NOT NULL DEFAULT FALSE, " +
|
|
"\"CreatedAt\" timestamptz NOT NULL, " +
|
|
"\"UpdatedAt\" timestamptz NOT NULL, " +
|
|
"CONSTRAINT \"PK_WorkflowRuntimes\" PRIMARY KEY (\"Id\"));",
|
|
"CREATE INDEX IF NOT EXISTS \"IX_WorkflowRuntimes_TenantId\" ON workflows.\"WorkflowRuntimes\" (\"TenantId\");",
|
|
// Per-tenant execution quota counter (one row per tenant per calendar month).
|
|
"CREATE TABLE IF NOT EXISTS workflows.\"WorkflowUsages\" (" +
|
|
"\"TenantId\" text NOT NULL, " +
|
|
"\"PeriodStart\" timestamptz NOT NULL, " +
|
|
"\"RunsUsed\" bigint NOT NULL DEFAULT 0, " +
|
|
"\"UpdatedAt\" timestamptz NOT NULL, " +
|
|
"CONSTRAINT \"PK_WorkflowUsages\" PRIMARY KEY (\"TenantId\", \"PeriodStart\"));",
|
|
};
|
|
|
|
private static async Task ApplyFullAsync(WorkflowsDbContext db, CancellationToken ct)
|
|
{
|
|
foreach (var statement in FullStatements)
|
|
await db.Database.ExecuteSqlRawAsync(statement, ct);
|
|
}
|
|
|
|
// ====================================================================== SQLite
|
|
|
|
private static readonly (string Table, string Column, string Sql)[] LiteColumns =
|
|
{
|
|
("Tasks", "ArchivedAt", "ALTER TABLE \"Tasks\" ADD COLUMN \"ArchivedAt\" TEXT NULL;"),
|
|
("Tasks", "Server", "ALTER TABLE \"Tasks\" ADD COLUMN \"Server\" TEXT NULL;"),
|
|
("Tasks", "NodeType", "ALTER TABLE \"Tasks\" ADD COLUMN \"NodeType\" TEXT NULL;"),
|
|
("Tasks", "NodeVersion", "ALTER TABLE \"Tasks\" ADD COLUMN \"NodeVersion\" REAL NULL;"),
|
|
("Tasks", "ParametersJson", "ALTER TABLE \"Tasks\" ADD COLUMN \"ParametersJson\" TEXT NULL;"),
|
|
("Tasks", "CredentialsJson", "ALTER TABLE \"Tasks\" ADD COLUMN \"CredentialsJson\" TEXT NULL;"),
|
|
("Tasks", "RetryJson", "ALTER TABLE \"Tasks\" ADD COLUMN \"RetryJson\" TEXT NULL;"),
|
|
("Tasks", "ContinueOnFail", "ALTER TABLE \"Tasks\" ADD COLUMN \"ContinueOnFail\" INTEGER NOT NULL DEFAULT 0;"),
|
|
("Tasks", "RunMode", "ALTER TABLE \"Tasks\" ADD COLUMN \"RunMode\" TEXT NULL;"),
|
|
("WorkflowRuns", "ParentRunId", "ALTER TABLE \"WorkflowRuns\" ADD COLUMN \"ParentRunId\" TEXT NULL;"),
|
|
("WorkflowRuns", "ParentTaskId", "ALTER TABLE \"WorkflowRuns\" ADD COLUMN \"ParentTaskId\" TEXT NULL;"),
|
|
("WorkflowRuns", "Depth", "ALTER TABLE \"WorkflowRuns\" ADD COLUMN \"Depth\" INTEGER NOT NULL DEFAULT 0;"),
|
|
("Workflows", "Repo", "ALTER TABLE \"Workflows\" ADD COLUMN \"Repo\" TEXT NOT NULL DEFAULT 'workflows';"),
|
|
("Workflows", "WebhookPath", "ALTER TABLE \"Workflows\" ADD COLUMN \"WebhookPath\" TEXT NULL;"),
|
|
};
|
|
|
|
private static readonly string[] LiteTables =
|
|
{
|
|
"CREATE TABLE IF NOT EXISTS \"WorkflowTaskEdges\" (" +
|
|
"\"Id\" TEXT NOT NULL, " +
|
|
"\"WorkflowId\" TEXT NOT NULL, " +
|
|
"\"FromTaskId\" TEXT NOT NULL, " +
|
|
"\"FromOutput\" INTEGER NOT NULL, " +
|
|
"\"ToTaskId\" TEXT NOT NULL, " +
|
|
"\"ToInput\" INTEGER NOT NULL, " +
|
|
"\"IsLoopBack\" INTEGER NOT NULL DEFAULT 0, " +
|
|
"CONSTRAINT \"PK_WorkflowTaskEdges\" PRIMARY KEY (\"Id\"), " +
|
|
"CONSTRAINT \"FK_WorkflowTaskEdges_Workflows_WorkflowId\" FOREIGN KEY (\"WorkflowId\") " +
|
|
"REFERENCES \"Workflows\" (\"Id\") ON DELETE CASCADE);",
|
|
"CREATE TABLE IF NOT EXISTS \"Credentials\" (" +
|
|
"\"Id\" TEXT NOT NULL, " +
|
|
"\"TenantId\" TEXT NOT NULL, " +
|
|
"\"Name\" TEXT NOT NULL, " +
|
|
"\"Type\" TEXT NOT NULL, " +
|
|
"\"EncryptedData\" TEXT NOT NULL, " +
|
|
"\"CreatedAt\" TEXT NOT NULL, " +
|
|
"\"UpdatedAt\" TEXT NOT NULL, " +
|
|
"CONSTRAINT \"PK_Credentials\" PRIMARY KEY (\"Id\"));",
|
|
"CREATE TABLE IF NOT EXISTS \"WorkflowRepos\" (" +
|
|
"\"TenantId\" TEXT NOT NULL, \"RepoName\" TEXT NOT NULL, \"UpdatedAt\" TEXT NULL, " +
|
|
"CONSTRAINT \"PK_WorkflowRepos\" PRIMARY KEY (\"TenantId\"));",
|
|
// The platform runtime row is irrelevant in Lite mode (a self-hosted runtime
|
|
// talks to the platform over the outbound channel), but the table must exist
|
|
// so the registry service can be used uniformly.
|
|
"CREATE TABLE IF NOT EXISTS \"WorkflowRuntimes\" (" +
|
|
"\"Id\" TEXT NOT NULL, " +
|
|
"\"TenantId\" TEXT NOT NULL, " +
|
|
"\"Label\" TEXT NOT NULL, " +
|
|
"\"Kind\" TEXT NOT NULL, " +
|
|
"\"Endpoint\" TEXT NULL, " +
|
|
"\"ApiKeyHash\" TEXT NULL, " +
|
|
"\"SecretHash\" TEXT NULL, " +
|
|
"\"Status\" TEXT NOT NULL DEFAULT 'unknown', " +
|
|
"\"LastSeenAt\" TEXT NULL, " +
|
|
"\"LastError\" TEXT NULL, " +
|
|
"\"InfoJson\" TEXT NULL, " +
|
|
"\"IsDefault\" INTEGER NOT NULL DEFAULT 0, " +
|
|
"\"CreatedAt\" TEXT NOT NULL, " +
|
|
"\"UpdatedAt\" TEXT NOT NULL, " +
|
|
"CONSTRAINT \"PK_WorkflowRuntimes\" PRIMARY KEY (\"Id\"));",
|
|
"CREATE TABLE IF NOT EXISTS \"WorkflowUsages\" (" +
|
|
"\"TenantId\" TEXT NOT NULL, " +
|
|
"\"PeriodStart\" TEXT NOT NULL, " +
|
|
"\"RunsUsed\" INTEGER NOT NULL DEFAULT 0, " +
|
|
"\"UpdatedAt\" TEXT NOT NULL, " +
|
|
"CONSTRAINT \"PK_WorkflowUsages\" PRIMARY KEY (\"TenantId\", \"PeriodStart\"));",
|
|
};
|
|
|
|
private static readonly string[] LiteIndexes =
|
|
{
|
|
"CREATE INDEX IF NOT EXISTS \"IX_WorkflowTaskEdges_WorkflowId\" ON \"WorkflowTaskEdges\" (\"WorkflowId\");",
|
|
"CREATE UNIQUE INDEX IF NOT EXISTS \"IX_Credentials_TenantId_Name\" ON \"Credentials\" (\"TenantId\", \"Name\");",
|
|
"CREATE INDEX IF NOT EXISTS \"IX_Credentials_TenantId\" ON \"Credentials\" (\"TenantId\");",
|
|
"CREATE INDEX IF NOT EXISTS \"IX_WorkflowRuns_ParentRunId\" ON \"WorkflowRuns\" (\"ParentRunId\");",
|
|
"CREATE INDEX IF NOT EXISTS \"IX_WorkflowRuntimes_TenantId\" ON \"WorkflowRuntimes\" (\"TenantId\");",
|
|
"CREATE INDEX IF NOT EXISTS \"IX_Workflows_TenantId_WebhookPath\" ON \"Workflows\" (\"TenantId\", \"WebhookPath\");",
|
|
};
|
|
|
|
private static async Task ApplyLiteAsync(WorkflowsDbContext db, CancellationToken ct)
|
|
{
|
|
foreach (var (table, column, sql) in LiteColumns)
|
|
{
|
|
if (!await LiteColumnExistsAsync(db, table, column, ct))
|
|
await db.Database.ExecuteSqlRawAsync(sql, ct);
|
|
}
|
|
|
|
foreach (var sql in LiteTables)
|
|
await db.Database.ExecuteSqlRawAsync(sql, ct);
|
|
|
|
foreach (var sql in LiteIndexes)
|
|
await db.Database.ExecuteSqlRawAsync(sql, ct);
|
|
}
|
|
|
|
/// <summary>
|
|
/// SQLite has no <c>ADD COLUMN IF NOT EXISTS</c>, so probe <c>pragma_table_info</c>
|
|
/// first. Names are bound as parameters; only the existence test is dynamic.
|
|
/// </summary>
|
|
private static async Task<bool> LiteColumnExistsAsync(
|
|
WorkflowsDbContext db, string table, string column, CancellationToken ct)
|
|
{
|
|
var connection = db.Database.GetDbConnection();
|
|
var closeAfter = connection.State != ConnectionState.Open;
|
|
if (closeAfter)
|
|
await connection.OpenAsync(ct);
|
|
|
|
try
|
|
{
|
|
await using var command = connection.CreateCommand();
|
|
command.CommandText = "SELECT COUNT(*) FROM pragma_table_info(@table) WHERE name = @column";
|
|
AddParameter(command, "@table", table);
|
|
AddParameter(command, "@column", column);
|
|
var result = await command.ExecuteScalarAsync(ct);
|
|
return Convert.ToInt64(result, CultureInfo.InvariantCulture) > 0;
|
|
}
|
|
finally
|
|
{
|
|
if (closeAfter)
|
|
await connection.CloseAsync();
|
|
}
|
|
}
|
|
|
|
private static void AddParameter(System.Data.Common.DbCommand command, string name, object value)
|
|
{
|
|
var parameter = command.CreateParameter();
|
|
parameter.ParameterName = name;
|
|
parameter.Value = value;
|
|
command.Parameters.Add(parameter);
|
|
}
|
|
}
|