feat: Docker connector, artifact system, and site improvements

examples/docker-data-transform.yaml

@@ -0,0 +1,32 @@
+name: docker-data-transform
+description: >
+  Query a database for recent records, process them through
+  a containerized tool, and post the results to Slack.
+  Demonstrates the docker/run connector with stdin/stdout
+  data passing.
+
+steps:
+  - name: fetch-records
+    action: postgres/query
+    credential: my-db
+    timeout: "15s"
+    params:
+      query: "SELECT id, name, status, updated_at FROM tasks WHERE updated_at > now() - interval '24 hours' ORDER BY updated_at DESC"
+
+  - name: transform-data
+    action: docker/run
+    credential: my-docker
+    timeout: "30s"
+    params:
+      image: giantswarm/tiny-tools
+      cmd: ["jq", "[.[] | {id, name, status}]"]
+      stdin: "{{ jsonEncode(steps['fetch-records'].output.rows) }}"
+      memory: "128m"
+
+  - name: post-results
+    action: slack/send
+    credential: slack-token
+    if: "steps['transform-data'].output.exit_code == 0"
+    params:
+      channel: "#daily-report"
+      text: "Daily task summary:\n{{ steps['transform-data'].output.stdout }}"

examples/docker-volume-backup.yaml

@@ -0,0 +1,56 @@
+name: docker-volume-backup
+description: >
+  Back up a Docker volume to S3 on a daily schedule.
+  Compresses the volume contents, uploads to S3, and
+  sends a Slack alert on success or failure.
+
+triggers:
+  - type: cron
+    schedule: "0 2 * * *"
+
+steps:
+  - name: backup-volume
+    action: docker/run
+    credential: my-docker
+    timeout: "10m"
+    params:
+      image: alpine
+      cmd: ["tar", "-czf", "/mantle/artifacts/backup.tar.gz", "-C", "/data", "."]
+      mounts:
+        - source: "my-app-data"
+          target: "/data"
+          readonly: true
+      memory: "256m"
+      remove: true
+    artifacts:
+      - path: /mantle/artifacts/backup.tar.gz
+        name: backup-archive
+
+  - name: upload-to-s3
+    action: s3/put
+    credential: aws-prod
+    timeout: "5m"
+    params:
+      bucket: my-backups
+      key: "volumes/my-app-data/{{ date }}/backup.tar.gz"
+      content: "{{ artifacts['backup-archive'].url }}"
+      content_type: "application/gzip"
+
+  - name: notify-success
+    action: slack/send
+    credential: slack-token
+    if: "steps['upload-to-s3'].error == null && steps['upload-to-s3'].output.size > 0"
+    params:
+      channel: "#ops-alerts"
+      text: "Volume backup completed — {{ artifacts['backup-archive'].size }} bytes uploaded to s3://my-backups/volumes/my-app-data/"
+
+  # NOTE: notify-failure requires continue_on_error support (see issue #29).
+  # Without it, the engine halts on the first failed step and this step
+  # will not execute. This is included to show the intended pattern.
+  - name: notify-failure
+    action: slack/send
+    credential: slack-token
+    if: "steps['upload-to-s3'].error != null || steps['backup-volume'].error != null"
+    params:
+      channel: "#ops-alerts"
+      text: "Volume backup FAILED"

go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/spf13/cobra v1.10.2
 	github.com/spf13/viper v1.21.0
 	github.com/stretchr/testify v1.11.1
+	github.com/docker/docker v28.5.2+incompatible
 	github.com/testcontainers/testcontainers-go v0.41.0
 	github.com/testcontainers/testcontainers-go/modules/postgres v0.41.0
 	golang.org/x/oauth2 v0.35.0
@@ -68,7 +69,6 @@ require (
 	github.com/cpuguy83/dockercfg v0.3.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/distribution/reference v0.6.0 // indirect
-	github.com/docker/docker v28.5.2+incompatible // indirect
 	github.com/docker/go-connections v0.6.0 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
 	github.com/ebitengine/purego v0.10.0 // indirect

internal/artifact/context.go

@@ -0,0 +1,16 @@
+package artifact
+
+import "context"
+
+type artifactsDirKey struct{}
+
+// WithArtifactsDir returns a context carrying the artifacts directory path.
+func WithArtifactsDir(ctx context.Context, dir string) context.Context {
+	return context.WithValue(ctx, artifactsDirKey{}, dir)
+}
+
+// ArtifactsDirFromContext extracts the artifacts directory from context.
+func ArtifactsDirFromContext(ctx context.Context) string {
+	dir, _ := ctx.Value(artifactsDirKey{}).(string)
+	return dir
+}

internal/artifact/reaper.go

@@ -0,0 +1,60 @@
+package artifact
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"time"
+)
+
+// Reaper cleans up expired artifacts from both tmp storage and the database.
+type Reaper struct {
+	Store      *Store
+	TmpStorage TmpStorage
+	Retention  time.Duration
+	Logger     *slog.Logger
+}
+
+// Sweep finds and removes artifacts older than the retention period.
+// Returns the number of artifacts cleaned up.
+func (r *Reaper) Sweep(ctx context.Context) (int, error) {
+	if r.Retention <= 0 {
+		return 0, nil // cleanup disabled
+	}
+
+	logger := r.Logger
+	if logger == nil {
+		logger = slog.Default()
+	}
+
+	expired, err := r.Store.ListExpired(ctx, r.Retention)
+	if err != nil {
+		return 0, fmt.Errorf("listing expired artifacts: %w", err)
+	}
+
+	if len(expired) == 0 {
+		return 0, nil
+	}
+
+	cleaned := 0
+	for _, a := range expired {
+		// Delete file from tmp storage.
+		if delErr := r.TmpStorage.Delete(ctx, a.URL); delErr != nil {
+			logger.Error("failed to delete artifact file",
+				"artifact", a.Name, "url", a.URL, "error", delErr)
+			continue // skip metadata deletion if file delete failed
+		}
+
+		// Delete metadata from database.
+		if err := r.Store.DeleteByID(ctx, a.ID); err != nil {
+			logger.Error("failed to delete artifact metadata",
+				"artifact", a.Name, "id", a.ID, "error", err)
+			continue
+		}
+		cleaned++
+		logger.Info("cleaned expired artifact",
+			"artifact", a.Name, "execution_id", a.ExecutionID)
+	}
+
+	return cleaned, nil
+}

internal/artifact/reaper_test.go

@@ -0,0 +1,90 @@
+package artifact
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestReaper_CleansExpiredArtifacts(t *testing.T) {
+	dir := t.TempDir()
+	fs := &FilesystemTmpStorage{BasePath: dir}
+	db := setupTestDB(t)
+	store := &Store{DB: db}
+	ctx := context.Background()
+	execID := createTestExecution(t, db)
+
+	// Write a file and create metadata
+	srcPath := filepath.Join(dir, "src.bin")
+	if err := os.WriteFile(srcPath, []byte("data"), 0644); err != nil {
+		t.Fatalf("os.WriteFile: %v", err)
+	}
+	key := "wf/" + execID + "/artifact/file.bin"
+	url, err := fs.Put(ctx, key, srcPath)
+	if err != nil {
+		t.Fatalf("fs.Put: %v", err)
+	}
+
+	if err := store.Create(ctx, &Artifact{
+		ExecutionID: execID,
+		StepName:    "step-a",
+		Name:        "test-artifact",
+		URL:         url,
+		Size:        4,
+	}); err != nil {
+		t.Fatalf("store.Create: %v", err)
+	}
+
+	// Backdate the artifact so it appears expired
+	_, err = db.ExecContext(ctx, `
+		UPDATE execution_artifacts SET created_at = NOW() - interval '48 hours'
+		WHERE execution_id = $1
+	`, execID)
+	if err != nil {
+		t.Fatalf("backdate: %v", err)
+	}
+
+	reaper := &Reaper{
+		Store:      store,
+		TmpStorage: fs,
+		Retention:  24 * time.Hour,
+	}
+
+	cleaned, err := reaper.Sweep(ctx)
+	if err != nil {
+		t.Fatalf("Sweep: %v", err)
+	}
+	if cleaned != 1 {
+		t.Errorf("cleaned = %d, want 1", cleaned)
+	}
+
+	// Metadata should be gone
+	arts, listErr := store.ListByExecution(ctx, execID)
+	if listErr != nil {
+		t.Fatalf("ListByExecution: %v", listErr)
+	}
+	if len(arts) != 0 {
+		t.Errorf("expected 0 artifacts after sweep, got %d", len(arts))
+	}
+}
+
+func TestReaper_SkipsWhenRetentionZero(t *testing.T) {
+	dir := t.TempDir()
+	fs := &FilesystemTmpStorage{BasePath: dir}
+
+	reaper := &Reaper{
+		Store:      &Store{}, // safe: Sweep returns early when Retention <= 0, before accessing Store.DB
+		TmpStorage: fs,
+		Retention:  0,
+	}
+
+	cleaned, err := reaper.Sweep(context.Background())
+	if err != nil {
+		t.Fatalf("Sweep: %v", err)
+	}
+	if cleaned != 0 {
+		t.Errorf("cleaned = %d, want 0 when retention is 0", cleaned)
+	}
+}

internal/artifact/store.go

@@ -0,0 +1,124 @@
+package artifact
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"time"
+)
+
+// Artifact represents metadata for a persisted artifact.
+type Artifact struct {
+	ID          string
+	ExecutionID string
+	StepName    string
+	Name        string
+	URL         string
+	Size        int64
+	CreatedAt   time.Time
+}
+
+// Store manages artifact metadata in Postgres.
+type Store struct {
+	DB *sql.DB
+}
+
+// Create inserts artifact metadata into the database and populates a.ID and a.CreatedAt.
+func (s *Store) Create(ctx context.Context, a *Artifact) error {
+	err := s.DB.QueryRowContext(ctx, `
+		INSERT INTO execution_artifacts (execution_id, step_name, name, url, size)
+		VALUES ($1, $2, $3, $4, $5)
+		RETURNING id, created_at
+	`, a.ExecutionID, a.StepName, a.Name, a.URL, a.Size).Scan(&a.ID, &a.CreatedAt)
+	if err != nil {
+		return fmt.Errorf("artifact create: %w", err)
+	}
+	return nil
+}
+
+// GetByName retrieves a single artifact by execution ID and artifact name.
+func (s *Store) GetByName(ctx context.Context, executionID, name string) (*Artifact, error) {
+	var a Artifact
+	err := s.DB.QueryRowContext(ctx, `
+		SELECT id, execution_id, step_name, name, url, size, created_at
+		FROM execution_artifacts
+		WHERE execution_id = $1 AND name = $2
+	`, executionID, name).Scan(&a.ID, &a.ExecutionID, &a.StepName, &a.Name, &a.URL, &a.Size, &a.CreatedAt)
+	if err == sql.ErrNoRows {
+		return nil, fmt.Errorf("artifact %q not found for execution %s", name, executionID)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("artifact get: %w", err)
+	}
+	return &a, nil
+}
+
+// ListByExecution returns all artifacts associated with the given execution, ordered by creation time.
+func (s *Store) ListByExecution(ctx context.Context, executionID string) ([]Artifact, error) {
+	rows, err := s.DB.QueryContext(ctx, `
+		SELECT id, execution_id, step_name, name, url, size, created_at
+		FROM execution_artifacts
+		WHERE execution_id = $1
+		ORDER BY created_at ASC
+	`, executionID)
+	if err != nil {
+		return nil, fmt.Errorf("artifact list: %w", err)
+	}
+	defer rows.Close()
+
+	var artifacts []Artifact
+	for rows.Next() {
+		var a Artifact
+		if err := rows.Scan(&a.ID, &a.ExecutionID, &a.StepName, &a.Name, &a.URL, &a.Size, &a.CreatedAt); err != nil {
+			return nil, fmt.Errorf("artifact scan: %w", err)
+		}
+		artifacts = append(artifacts, a)
+	}
+	return artifacts, rows.Err()
+}
+
+// DeleteByExecution removes all artifact metadata for an execution.
+func (s *Store) DeleteByExecution(ctx context.Context, executionID string) error {
+	_, err := s.DB.ExecContext(ctx, `
+		DELETE FROM execution_artifacts WHERE execution_id = $1
+	`, executionID)
+	if err != nil {
+		return fmt.Errorf("artifact delete: %w", err)
+	}
+	return nil
+}
+
+// DeleteByID removes a single artifact metadata row.
+func (s *Store) DeleteByID(ctx context.Context, id string) error {
+	_, err := s.DB.ExecContext(ctx, `
+		DELETE FROM execution_artifacts WHERE id = $1
+	`, id)
+	if err != nil {
+		return fmt.Errorf("artifact delete by id: %w", err)
+	}
+	return nil
+}
+
+// ListExpired returns artifacts older than the given duration, using the DB clock for comparison.
+func (s *Store) ListExpired(ctx context.Context, olderThan time.Duration) ([]Artifact, error) {
+	rows, err := s.DB.QueryContext(ctx, `
+		SELECT id, execution_id, step_name, name, url, size, created_at
+		FROM execution_artifacts
+		WHERE created_at < NOW() - $1::interval
+		ORDER BY created_at ASC
+	`, olderThan.String())
+	if err != nil {
+		return nil, fmt.Errorf("artifact list expired: %w", err)
+	}
+	defer rows.Close()
+
+	var artifacts []Artifact
+	for rows.Next() {
+		var a Artifact
+		if err := rows.Scan(&a.ID, &a.ExecutionID, &a.StepName, &a.Name, &a.URL, &a.Size, &a.CreatedAt); err != nil {
+			return nil, fmt.Errorf("artifact scan: %w", err)
+		}
+		artifacts = append(artifacts, a)
+	}
+	return artifacts, rows.Err()
+}