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: "{{ steps['fetch-records'].output.json }}"
+      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,53 @@
+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/"
+
+  - name: notify-failure
+    action: slack/send
+    credential: slack-token
+    if: "steps['upload-to-s3'].error != null"
+    params:
+      channel: "#ops-alerts"
+      text: "Volume backup FAILED — {{ steps['upload-to-s3'].error }}"

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,74 @@
+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
+	}
+
+	// Group by execution for batch deletion.
+	byExecution := make(map[string][]Artifact)
+	for _, a := range expired {
+		byExecution[a.ExecutionID] = append(byExecution[a.ExecutionID], a)
+	}
+
+	cleaned := 0
+	for execID, arts := range byExecution {
+		// Delete files from tmp storage.
+		fileDeleteFailed := false
+		for _, a := range arts {
+			if delErr := r.TmpStorage.DeleteByPrefix(ctx, a.URL); delErr != nil {
+				logger.Error("failed to delete artifact file",
+					"artifact", a.Name, "url", a.URL, "error", delErr)
+				fileDeleteFailed = true
+			}
+		}
+
+		// Delete metadata from database.
+		if err := r.Store.DeleteByExecution(ctx, execID); err != nil {
+			logger.Error("failed to delete artifact metadata",
+				"execution_id", execID, "error", err)
+			continue
+		}
+
+		if fileDeleteFailed {
+			logger.Warn("some artifact files could not be deleted",
+				"execution_id", execID)
+		}
+		cleaned += len(arts)
+		logger.Info("cleaned expired artifacts",
+			"execution_id", execID, "count", len(arts))
+	}
+
+	return cleaned, nil
+}

internal/artifact/reaper_test.go

@@ -0,0 +1,87 @@
+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, _ := store.ListByExecution(ctx, execID)
+	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{}, // won't be called
+		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,113 @@
+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.
+func (s *Store) Create(ctx context.Context, a *Artifact) error {
+	_, err := s.DB.ExecContext(ctx, `
+		INSERT INTO execution_artifacts (execution_id, step_name, name, url, size)
+		VALUES ($1, $2, $3, $4, $5)
+	`, a.ExecutionID, a.StepName, a.Name, a.URL, a.Size)
+	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
+}
+
+// ListExpired returns artifacts older than the given duration.
+func (s *Store) ListExpired(ctx context.Context, olderThan time.Duration) ([]Artifact, error) {
+	cutoff := time.Now().Add(-olderThan)
+	rows, err := s.DB.QueryContext(ctx, `
+		SELECT id, execution_id, step_name, name, url, size, created_at
+		FROM execution_artifacts
+		WHERE created_at < $1
+		ORDER BY created_at ASC
+	`, cutoff)
+	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()
+}