w4c-workflows-api/Data/WorkflowsSchema.cs

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