Background jobs & workflows, code-first

What you get

Quickstart

Install the core plus the adapters you need. While Klassd.Workflows is in beta the packages are prerelease:

dotnet add package Klassd.Workflows.Core --prerelease
dotnet add package Klassd.Workflows.Storage.Postgres --prerelease   # durable store (or .Storage.MongoDb / .Storage.Sqlite)
dotnet add package Klassd.Workflows.Kubernetes --prerelease         # K8s executor (omit for local only)
dotnet add package Klassd.Workflows.Artifacts.S3 --prerelease       # artifact store (or .Artifacts.Gcs)
dotnet add package Klassd.Workflows.Dashboard --prerelease          # the live UI (Razor Class Library)
dotnet add package Klassd.Workflows.Auth --prerelease               # optional: dashboard users/SSO (+ .Auth.OpenIdConnect)

1. Define a job

Any class implementing IJob is a unit of work. Use the IJobContext to log, report progress, and read arguments:

public sealed class MyJob : IJob
{
    public async Task RunAsync(IJobContext ctx)
    {
        ctx.Log("starting");
        ctx.ReportProgress(50, "halfway");
        await Task.Delay(1000, ctx.CancellationToken);
        ctx.Log("done");
    }
}

2. Wire it up in Program.cs

AddKlassdWorkflowsCore() returns a builder you use to pick a durable store; pick an executor separately:

var workflows = builder.Services.AddKlassdWorkflowsCore();
workflows.UsePostgres("Host=…;Database=…;Username=…;Password=…");  // or .UseMongo(...) / in-memory

builder.Services.AddKubernetesExecutor(builder.Configuration);     // or AddLocalExecutor(workerDll)

3. Mount the dashboard (optional)

The dashboard ships as a Razor Class Library — add the package and two calls to get the live UI (jobs catalog, run history, per-job console with progress bars, DAG views) in your own ASP.NET Core host:

builder.Services.AddHttpContextAccessor();        // dashboard reads a theme cookie during SSR
builder.Services.AddKlassdWorkflowsDashboard();   // the Blazor Interactive Server UI

var app = builder.Build();
app.UseAntiforgery();
app.MapKlassdWorkflowsDashboard();                // static assets + component endpoints
app.Run();

If your host has no .razor of its own, set <RequiresAspNetWebAssets>true</RequiresAspNetWebAssets> in its csproj (otherwise _framework/blazor.web.js 404s). The samples/Klassd.Workflows.DashboardHost project is a complete, runnable example.

4. Run

Enqueue jobs from code, or open the dashboard to start/stop them and watch live console output.

Scheduling

Fire a job now, or register a recurring job with a cron expression (parsed by Cronos):

scheduler.AddOrUpdateRecurring<MyJob>("nightly", "0 2 * * *");   // cron
await scheduler.EnqueueAsync<MyJob>();                            // fire now

// Standalone jobs can carry init containers too (e.g. a migration before the worker runs):
await scheduler.EnqueueAsync<MyJob>(initContainers: new[]
{
    new InitContainerSpec { Name = "migrate", Image = "myorg/migrate:1" }
});

Recurring workflows are registered the same way with AddOrUpdateRecurringWorkflow(id, name, cron). Pod resources (CPU/memory requests & limits) are set per job with a [JobResources] attribute and can be retuned from config without a recompile.

Multi-tenant

One job image can serve many tenants, each loading its own configuration — no per-tenant code. Pass a tenant at enqueue (every enqueue, recurring and workflow overload takes an optional tenant); it rides through to the worker as KLASSD_TENANT, which layers tenant-specific configuration over the shared base before building the job. A job reads the tenant from its (already tenant-scoped) IConfiguration in Configure, or from IJobContext.Tenant at runtime.

// Pass a tenant on any enqueue / recurring / workflow overload:
await scheduler.EnqueueAsync<ReportJob>(tenant: "acme");
await scheduler.EnqueueWorkflowAsync("nightly-sync", tenant: "globex");
scheduler.AddOrUpdateRecurring<ReportJob>("acme-nightly", "0 2 * * *", tenant: "acme");

// The worker layers tenant config over the shared base, so one image serves every tenant:
//   appsettings.json → appsettings.{ENV}.json → appsettings.{tenant}.json
//     → /secrets/*.json → /secrets/{tenant}/*.json → environment variables
public sealed class ReportJob : IJob
{
    // The tenant is on the (already tenant-scoped) configuration — branch your DI on it:
    public static void Configure(IServiceCollection services, IConfiguration cfg)
    {
        var tenant = cfg[WorkerProtocol.ConfigTenantKey];        // null when not multi-tenant
        services.AddSingleton<IReportSink>(tenant == "acme"
            ? new S3Sink(cfg["acme:bucket"]!) : new LocalSink());
    }

    public async Task RunAsync(IJobContext ctx) =>
        ctx.Log(quot;running for {ctx.Tenant ?? "(no tenant)"}");   // also on the context at runtime
}

Configuration precedence (last wins): appsettings.json → appsettings.{ENV}.json → appsettings.{tenant}.json → /secrets/*.json → /secrets/{tenant}/*.json → environment variables. Workflow nodes inherit the run's tenant; recurring definitions keep theirs across fires.

Workflows (DAGs)

Jobs compose into a directed acyclic graph that fans out, waits on dependencies and passes data between nodes. The orchestrator runs in the scheduler; each node runs as a normal worker pod, so every node has its own live console.

registry.Register(new WorkflowBuilder("catalog-integration")
    .Add<MarketFinderJob>("markets")                       // root: emits "market_ids"
    .Add<DataProxyJob>("data-proxy")                       // parallel root: writes an artifact
    .Add<IntegrationJob>("integration", n => n
        .DependsOn("markets", "data-proxy")
        .FanOutOver("markets", "market_ids", itemArgument: "market"))   // one pod per market
    .Add<PublishJob>("publish", n => n.DependsOn("integration").WithRetries(2))
    .Add<FinalizerJob>("finalizer", n => n
        .DependsOn("publish", "data-proxy")
        .BindInput("dataset_ref", "data-proxy", "dataset_ref"))          // reads the artifact
    .Build());

• Dependencies — a node starts once all its dependencies are satisfied; a failed dependency skips dependents.
• Fan-out — read an upstream output as a JSON array and start one execution per element. maxParallelism: n caps how many run at once (0 = unlimited) so a big list doesn't spawn n pods at once.
• Inputs — bind an argument to an upstream output; for a service/container node's address use BindServiceAddress / BindServiceIp (like Argo's tasks.x.ip, without the magic key).
• File outputs — WithFileOutput(name, path, default) publishes a node output from a file the step writes (or the default if absent) — Argo's valueFrom.path, for both IJob and container nodes.
• Conditions — run a node only when a predicate over upstream outputs holds (otherwise it's benignly omitted).
• Retries — re-run a failed execution up to n times, per fan-out item.
• Artifacts — large payloads pass through an IArtifactStore; a node saves an artifact and publishes the small reference downstream.

Container jobs & service nodes

A node doesn't have to be a C# IJob — it can be any container image. That lets you run existing tools (a legacy Go binary, a vendor CLI) as first-class jobs and DAG nodes without porting them. Mark one .AsService() to keep it running as a sidecar and forward its address to dependents — the pattern for a cloud-sql-proxy:

registry.Register(new WorkflowBuilder("cloud-sql-integration")
    .AddContainer("sql-proxy", "gcr.io/cloud-sql-connectors/cloud-sql-proxy:2.11.0", c => c
        .WithArgs("--address=0.0.0.0", "--port=5432", "my-project:region:instance")
        .ServicePort(5432).ReadyOnTcp(5432)
        .AsService())                                   // long-running; torn down at the end
    .Add<IntegrationJob>("integration", n => n
        .BindServiceAddress("db_host", "sql-proxy"))    // {podIP}:5432 forwarded; also adds the dependency
    .Build());

• Address forwarding — the engine reads the pod IP and publishes ip / address outputs; dependents bind them with BindInput.
• Readiness — a service node satisfies dependents once its pod is ready (optionally gated on a ReadyOnTcp port), not when it exits.
• Teardown — services are stopped automatically when the rest of the run finishes; a Kubernetes activeDeadlineSeconds backstop reaps an orphan if the scheduler dies.

You can also run a container as a standalone job (no workflow) — enqueue or schedule it like any other job:

// Run an existing image as a standalone job — no IJob port needed.
containerJobs.Register(new ContainerJobDefinition
{
    Name = "legacy-importer",
    Container = new ContainerSpec { Image = "ghcr.io/acme/go-importer:1.4", Args = ["--full"] },
});
await scheduler.EnqueueContainerAsync("legacy-importer",
    new ContainerSpec { Image = "ghcr.io/acme/go-importer:1.4" });

Under the Kubernetes executor a container node runs as its own pod; under the local executor it runs via docker run, so the same DAG works in dev without a cluster.

Executors: local & Kubernetes

The same worker runs locally and in the cluster — only the executor that launches it differs. Communication is a line protocol on stdout, so Kubernetes pod logs are the transport for free.

• Local — AddLocalExecutor(...) launches the worker as a child process per job. No cluster required; ideal for dev.
• Kubernetes — AddKubernetesExecutor(...) creates a batch/v1 Job (one pod, restartPolicy: Never) per execution, tails its logs, and cleans up via ttlSecondsAfterFinished. Stopping a job deletes the Job; SIGTERM cancels the worker's token.

The worker ships as a package: reference Klassd.Workflows.Worker from a thin exe, register the jobs it can run with WorkerHost.CreateBuilder(args).RegisterJobs(…), and publish it as your own worker image. Each job declares its own dependencies on the static IJob.Configure (config from appsettings + /secrets + env) — see Dependency injection.

Dependency injection

Register the jobs a worker can run with RegisterJobs. Each job declares its own dependencies on a static IJob.Configure(IServiceCollection, IConfiguration) — run only when that job is dispatched (a worker pod runs one job), so a worker image hosting dozens of jobs never registers services the invoked job doesn't use. The job is then built by a source-generated new T(sp.GetRequiredService<…>()) factory — no reflection, no ActivatorUtilities, trim/AOT-friendly (the generator ships as an analyzer inside Klassd.Workflows.Abstractions). Reserve the worker-wide ConfigureServices for genuinely cross-cutting services. Configuration is composed from appsettings[.{ENV}].json → every /secrets/*.json (the Vault-agent drop dir) → environment variables (last wins), and is itself injectable.

// Your worker exe (Program.cs): register the jobs it can run, then run the single job
// the scheduler dispatched. Each job declares its own dependencies (see below), so the
// only thing on ConfigureServices is genuinely cross-cutting (shared by every job).
return await WorkerHost.CreateBuilder(args)
    .ConfigureServices((services, config) => services.AddHttpClient())  // cross-cutting only
    .RegisterJobs(MyJobs.Register)                          // the shared registration, below
    .AddArtifactProvider(new GcsArtifactStoreProvider())    // backends picked by name; "file" is built in
    .RunAsync();

Registering jobs

Each job maps to a dispatch key (the value the scheduler sends). Put the registration in one shared method that both the worker exe (RegisterJobs) and the dashboard host (AddJobs) call, so both sides agree on the keys:

// One shared registration, referenced by BOTH the worker exe and the dashboard host,
// so the two agree on the dispatch keys. The default key is the job's full type name —
// matching what EnqueueAsync<T>(), recurring jobs and workflow nodes already emit.
public static class MyJobs
{
    public static void Register(JobRegistrationBuilder j) => j
        .Add<SyncCatalogJob>()                              // key = full type name
        .Add<ConfiguredGreetingJob>("greeting")             // explicit key
        .Add("report", sp => new ReportJob(sp.GetRequiredService<IFoo>())); // explicit factory
}

// A job declares its OWN dependencies on the static Configure — run only when this job is
// dispatched (a worker pod runs one job), so jobs that aren't invoked cost nothing. The job
// is then constructed by a source-generated factory (no reflection, no ActivatorUtilities).
public sealed class SyncCatalogJob(IHttpClientFactory http, CatalogOptions opts) : IJob
{
    public static void Configure(IServiceCollection services, IConfiguration config) =>
        services.AddSingleton(new CatalogOptions { BaseUrl = config["Catalog:BaseUrl"]! });

    public async Task RunAsync(IJobContext ctx) { /* use http, opts */ }
}

// The dashboard host registers the same jobs so the catalog + workflow validation match:
//   builder.Services.AddKlassdWorkflowsCore().AddJobs(MyJobs.Register);

• The default Add<T>() key is the job's full type name, matching what EnqueueAsync<T>(), recurring jobs and workflow nodes emit — so registering by type just works.
• Pass an explicit key (Add<T>("my-key")) for a stable key decoupled from the type, or a factory (Add("key", sp => new MyJob(...))) for bespoke construction.
• Set option values from any config source — e.g. the env var Greeting__Salutation=Hej — and the job picks it up with no recompile.

More behaviour-oriented recipes (inputs, progress, output & artifact passing, fan-out, conditions, retries, service nodes, cancellation) live in docs/recipes.md.

Shaping the pod

On the Kubernetes executor you control the pod each job runs in, at three scopes that combine: executor-wide (options), per DAG node, and per container/job.

new WorkflowBuilder("nightly")
    .AddContainer("sql-proxy", proxyImage, c => c.AsService().ServicePort(5432).ReadyOnTcp(5432))
    .Add<CleanupJob>("cleanup", n => n
        .BindServiceAddress("db_host", "sql-proxy")     // also adds the dependency
        .WithInitContainer("migrate", "myorg/migrate:1", "--db", "$(db_host)")  // runs first
        .WithEmptyDir("scratch").WithVolumeMount("scratch", "/scratch")          // shared volume
        .WithEnvFromSecret("db-creds")                                           // ConfigMap/Secret -> env
        .WithNodeSelector("pool", "batch")                                       // + tolerations / affinity
        .WithSecurityContext(new() { RunAsNonRoot = true, ReadOnlyRootFilesystem = true }))
    .Build();

• Init containers — run to completion before the main container (migrations, pre-flight, seeding a shared volume).
• Volumes & mounts — emptyDir / secret / configMap / PVC / hostPath, mounted into the main and init containers.
• Security contexts — pod-level (runAsUser, fsGroup, seccomp…) and per-container (readOnlyRootFilesystem, drop capabilities…).
• Resources — CPU/memory requests & limits on container jobs and init containers, overlaid on the executor default.
• envFrom — import a ConfigMap or Secret as environment variables.
• Scheduling — nodeSelector, tolerations, and affinity (node + pod (anti-)affinity).
• Annotations & labels — stamped on every pod; the seam for sidecar injectors such as the Vault agent.

Storage & artifacts

Two pluggable seams, both with built-in adapters and open for your own:

• Job store (IJobStore) — holds executions, recurring entries and workflow runs. In-memory by default; UsePostgres(...), UseMongo(...) or UseSqlite(...) for durability. Each also provides the dashboard user store.
• Artifact store (IArtifactStore) — holds large payloads between nodes. The worker selects a provider by name at runtime (file, s3, gcs), so the choice is per-deployment, not compiled in.

Add your own by implementing the interface (plus an IArtifactStoreProvider for artifacts) — exactly how the Postgres/Mongo and S3/GCS packages do it.

Dashboard auth (optional)

The dashboard is unauthenticated by default. Add Klassd.Workflows.Auth for an email/password Users admin and cookie sign-in — the same model as the Klassd CMS — and Klassd.Workflows.Auth.OpenIdConnect for single sign-on (Entra ID, Okta, Auth0, Google, …). An SSO identity is linked to an existing user by email, or auto-provisioned.

builder.Services.AddKlassdWorkflowsAuth(o =>
{
    o.SeedAdminEmail = config["Auth:SeedAdmin:Email"];        // first admin on a fresh deploy
    o.SeedAdminPassword = config["Auth:SeedAdmin:Password"];
});
builder.Services.AddKlassdWorkflowsOpenIdConnect("Company SSO", config.GetSection("Oidc"));  // optional

var app = builder.Build();
app.UseKlassdWorkflowsAuth();        // loopback (local dev / kubectl port-forward) is bypassed
app.MapKlassdWorkflowsDashboard();

• SSO and a password on one account — an SSO sign-in whose provider-verified email matches a password account attaches to it (one user, either method). Staff self-serve from the Your account page: add a password, link/unlink a provider.
• Loopback bypass — requests from 127.0.0.1/::1 skip auth, so local dev and kubectl port-forward need no login. Ingress traffic is always authenticated.
• Seed admin — a first user is created from config on a fresh deployment, so you're never locked out.
• Durable users — stored alongside jobs in your Postgres / MongoDB / SQLite store.
• Embed in an existing app — set OwnsHost = false with the dashboard's BasePath to mount it inside a host that already has its own authentication; dashboard sign-in is scoped to its routes and never touches the host's auth.

Packages

Install the core plus the adapters you need — each keeps its SDK isolated, so you only pull in what you wire up. While in beta, add --prerelease.

View on GitHubAll NuGet packages