spooled-sdk-go

Spooled Go SDK

Official Go SDK for Spooled Cloud — a modern, scalable job queue and task scheduler.

Live Demo (SpriteForge) • Documentation

Features

• Idiomatic Go API with functional options pattern
• Full API Coverage — Access all Spooled API endpoints
• Type Safety — Strongly typed requests and responses
• Context Support — Full context.Context propagation
• Automatic Retries — Exponential backoff with jitter
• Circuit Breaker — Fault tolerance for unreliable networks
• Worker Runtime — Process jobs with concurrent workers
• Real-time Events — WebSocket and SSE support
• gRPC Support — High-performance streaming client
• Workflow DAGs — Complex job dependencies
• Automatic JWT Refresh — Single-flight token refresh

Installation

go get github.com/spooled-cloud/spooled-sdk-go

Requirements: Go 1.25 or later

Quick Start

Create a Job

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/spooled-cloud/spooled-sdk-go/spooled"
    "github.com/spooled-cloud/spooled-sdk-go/spooled/resources"
)

func main() {
    // Create client
    client, err := spooled.NewClient(
        spooled.WithAPIKey("sp_live_your_api_key"),
        // For self-hosted: spooled.WithBaseURL("https://your-server.com"),
    )
    if err != nil {
        log.Fatal(err)
    }

    ctx := context.Background()

    // Create a job
    priority := 5
    maxRetries := 3
    result, err := client.Jobs().Create(ctx, &resources.CreateJobRequest{
        QueueName: "email-notifications",
        Payload: map[string]any{
            "to":      "user@example.com",
            "subject": "Welcome!",
            "body":    "Thanks for signing up.",
        },
        Priority:   &priority,
        MaxRetries: &maxRetries,
    })
    if err != nil {
        log.Fatal(err)
    }

    fmt.Printf("Created job: %s\n", result.ID)

    // Get job status
    job, err := client.Jobs().Get(ctx, result.ID)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("Status: %s\n", job.Status)
}

Process Jobs with a Worker

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"

    "github.com/spooled-cloud/spooled-sdk-go/spooled"
    "github.com/spooled-cloud/spooled-sdk-go/spooled/worker"
)

func main() {
    client, err := spooled.NewClient(
        spooled.WithAPIKey("sp_live_your_api_key"),
    )
    if err != nil {
        log.Fatal(err)
    }

    // Create worker
    w := worker.NewWorker(client.Jobs(), client.Workers(), worker.Options{
        QueueName:   "email-notifications",
        Concurrency: 10,
    })

    // Register job handler
    w.Process(func(ctx *worker.JobContext) (map[string]any, error) {
        fmt.Printf("Processing job %s\n", ctx.JobID)
        
        // Access payload. Check the assertion: a payload without the key would
        // otherwise panic (the worker recovers it and fails the job with the panic
        // text, but a missing field is better reported as a plain error).
        to, ok := ctx.Payload["to"].(string)
        if !ok {
            return nil, fmt.Errorf("payload has no string \"to\"")
        }
        subject, _ := ctx.Payload["subject"].(string)
        
        // Do your work here
        fmt.Printf("Sending email to %s: %s\n", to, subject)
        
        return map[string]any{"sent": true}, nil
    })

    // Handle events
    w.OnEvent(func(event worker.Event) {
        fmt.Printf("Event: %s\n", event.Type)
    })

    // Start worker
    ctx, cancel := context.WithCancel(context.Background())
    if err := w.Start(ctx); err != nil {
        log.Fatal(err)
    }

    // Graceful shutdown: on SIGTERM stop claiming, let running handlers finish
    // (up to Options.ShutdownTimeout), send their results, then deregister.
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
    <-sigCh

    cancel() // starts the drain; running handlers keep their contexts
    w.Stop() // blocks until the drain has finished
}

Real-world examples (beginner friendly)

If you want 5 copy/paste “real life” setups (Stripe → jobs, GitHub Actions → jobs, cron schedules, CSV import, website signup), see:

• https://github.com/spooled-cloud/spooled-backend/blob/main/docs/guides/real-world-examples.md

Documentation

Guide	Description
Getting Started	Installation, setup, and first job
Configuration	All configuration options
Workers	Job processing with Worker
Workflows	Building DAGs with job dependencies
gRPC	High-performance streaming
Real-time Events	WebSocket and SSE
Error Handling	Typed errors and retries

Examples

See the examples/ directory for runnable code:

Example	Description
quick-start/	Basic SDK usage
worker/	Processing jobs with Worker
workflow/	Complex workflows with dependencies
grpc/	High-performance gRPC streaming
realtime/	Real-time event streaming
schedules/	Cron schedules
webhooks/	Webhook configuration
api-keys/	API key management
bulk-operations/	Bulk job operations
error-handling/	Error handling patterns

Run any example with:

API_KEY=sp_live_... go run examples/quick-start/main.go

Configuration

Client Options

client, err := spooled.NewClient(
    // Authentication (required - one of these)
    spooled.WithAPIKey("sp_live_..."),
    spooled.WithAccessToken("eyJ..."),
    
    // Endpoints (optional)
    spooled.WithBaseURL("https://api.spooled.cloud"),
    spooled.WithWSURL("wss://api.spooled.cloud/api/v1/ws"),
    spooled.WithGRPCAddress("grpc.spooled.cloud:443"),
    
    // Timeouts (optional)
    spooled.WithTimeout(30 * time.Second),
    
    // Retries (optional)
    spooled.WithRetryConfig(spooled.RetryConfig{
        MaxRetries: 3,
        BaseDelay:  100 * time.Millisecond,
        MaxDelay:   5 * time.Second,
    }),
    
    // Circuit breaker (optional)
    spooled.WithCircuitBreaker(spooled.CircuitBreakerConfig{
        Enabled:          true,
        FailureThreshold: 5,
        Timeout:          30 * time.Second,
    }),
    
    // Debug logging (optional)
    spooled.WithDebug(true),
)

Environment Variables

SPOOLED_API_KEY=sp_live_...
SPOOLED_BASE_URL=https://api.spooled.cloud
SPOOLED_GRPC_ADDRESS=grpc.spooled.cloud:443

Core Concepts

Jobs

Jobs are units of work with payloads, priorities, and retry policies:

priority := 5
maxRetries := 3
timeout := 300
scheduledAt := time.Now().Add(time.Hour)

result, err := client.Jobs().Create(ctx, &resources.CreateJobRequest{
    QueueName:      "my-queue",
    Payload:        map[string]any{"data": "value"},
    Priority:       &priority,        // -100 to 100
    MaxRetries:     &maxRetries,
    TimeoutSeconds: &timeout,
    ScheduledAt:    &scheduledAt,
    IdempotencyKey: ptr("unique-key"),
})

// List jobs
jobs, err := client.Jobs().List(ctx, &resources.ListJobsParams{
    QueueName: ptr("my-queue"),
    Status:    ptr(resources.JobStatusPending),
    Tag:       ptr("billing"), // Optional: filter by a single tag
    Limit:     ptr(10),
})

// Cancel a job
err = client.Jobs().Cancel(ctx, jobID)

Workers

Process jobs with the built-in worker runtime:

w := worker.NewWorker(client.Jobs(), client.Workers(), worker.Options{
    QueueName:         "my-queue",
    WorkerID:          os.Getenv("HOSTNAME"), // Stable identity, see below
    Concurrency:       10,              // Max concurrent jobs
    PollInterval:      time.Second,     // Polling frequency
    LeaseDuration:     30,              // Lease duration in seconds
    HeartbeatFraction: 0.5,             // Heartbeat at 50% of lease
    ShutdownTimeout:   30 * time.Second,
})

w.Process(func(ctx *worker.JobContext) (map[string]any, error) {
    // ctx.Context - Go context with cancellation
    // ctx.JobID - Unique job ID
    // ctx.QueueName - Queue name
    // ctx.Payload - Job payload when it is a JSON object (nil otherwise)
    // ctx.RawPayload - Job payload of any JSON type (array, string, number...)
    // ctx.RetryCount - Current retry attempt
    // ctx.MaxRetries - Maximum retries
    // ctx.Log(level, message, meta) - Log messages
    // (ctx.Progress is deprecated: the API has no progress route, it returns
    // an error that matches spooled.ErrUnsupported)
    
    return map[string]any{"result": "success"}, nil
})

// Event handlers
w.OnEvent(func(event worker.Event) {
    switch event.Type {
    case worker.EventJobCompleted:
        data := event.Data.(worker.JobCompletedData)
        fmt.Printf("Job %s completed in %v\n", data.JobID, data.Duration)
    case worker.EventJobFailed:
        data := event.Data.(worker.JobFailedData)
        fmt.Printf("Job %s failed: %v (will retry: %t)\n", data.JobID, data.Error, data.WillRetry)
    case worker.EventJobLeaseLost:
        data := event.Data.(worker.JobLeaseLostData)
        fmt.Printf("Job %s lost lease %s during %s: %v\n", data.JobID, data.LeaseID, data.Operation, data.Error)
    }
})

// The worker automatically echoes the claim's LeaseID on heartbeat and settlement.
// A stale lease cancels that handler context and emits EventJobLeaseLost.
ctx, cancel := context.WithCancel(context.Background())
w.Start(ctx)

// Graceful shutdown (SIGTERM handler): cancelling ctx and calling Stop do the
// same thing, and Stop blocks until it is finished.
cancel()
w.Stop()

How the worker behaves:

• Shutdown drains. Stop (or cancelling the context passed to Start) stops claiming new jobs and gives running handlers up to ShutdownTimeout to finish with their contexts still live. A handler that returns in that window has its result sent (complete, or fail if it returned an error) before the worker deregisters. Only handlers still running when the timeout expires are canceled, and their jobs failed with Worker shutdown timeout. A handler that ignores its context keeps running (Go cannot kill a goroutine), but the worker stops waiting for it. Do not call Stop from inside a handler or event callback.
• Panics are job failures. A panic in a handler is recovered and the job is failed with the panic text (panic: ...); the worker and its other jobs keep running.
• Job timeouts. A job with timeout_seconds set is canceled and failed with Job timeout exceeded when it runs longer (worker.EventJobTimeout); its lease stops being renewed.
• Acknowledgements are retried. complete, fail, job heartbeats and worker heartbeats are retried on network errors, timeouts, 408, 429 and 5xx with the client's retry settings, and they ignore an open circuit breaker.
• Failure text is always accepted. An empty error becomes Job failed and anything longer than 2048 characters is truncated (the API rejects both).
• Numbers keep their precision. Payload and result numbers are float64, except an integer too large for a float64 (above 2^53, such as a Discord ID), which arrives as a json.Number: n.Int64().

WorkerID is optional but worth setting. It must be 1-128 characters from [A-Za-z0-9._-], and keeping it the same across restarts makes registration an upsert, so the process reclaims one worker row. Leave it empty and the server mints a fresh UUID on every Start, which leaves the previous row occupying your plan's worker cap until the stale-worker reaper clears it (roughly two minutes) - long enough for a crash-looping worker on a tight plan to hit 429 QUOTA_EXCEEDED on its own registrations. Re-registering an ID your organization already owns costs nothing against the cap; an ID owned by another organization is rejected with 409.

Workflows (DAGs)

Orchestrate multiple jobs with dependencies:

workflow, err := client.Workflows().Create(ctx, &resources.CreateWorkflowRequest{
    Name: "ETL Pipeline",
    Jobs: []resources.WorkflowJob{
        {
            Key:       "extract",
            QueueName: "etl",
            Payload:   map[string]any{"step": "extract"},
        },
        {
            Key:       "transform",
            QueueName: "etl",
            Payload:   map[string]any{"step": "transform"},
            DependsOn: []string{"extract"},
        },
        {
            Key:       "load",
            QueueName: "etl",
            Payload:   map[string]any{"step": "load"},
            DependsOn: []string{"transform"},
        },
    },
})

// Get workflow status
wf, err := client.Workflows().Get(ctx, workflow.WorkflowID)
fmt.Printf("Status: %s\n", wf.Status)

Schedules

Run jobs on a cron schedule:

schedule, err := client.Schedules().Create(ctx, &resources.CreateScheduleRequest{
    Name:           "Daily Report",
    CronExpression: "0 9 * * *",       // 9 AM daily (5-field cron)
    Timezone:       ptr("America/New_York"),
    QueueName:      "reports",
    PayloadTemplate: map[string]any{"type": "daily"},
})

fmt.Printf("Next run: %v\n", schedule.NextRunAt)

// Pause/resume
client.Schedules().Pause(ctx, schedule.ID)
client.Schedules().Resume(ctx, schedule.ID)

// Trigger immediately
client.Schedules().Trigger(ctx, schedule.ID)

Queues

Manage queues and view statistics:

// List queues
queues, err := client.Queues().List(ctx)

// Get queue stats
stats, err := client.Queues().GetStats(ctx, "my-queue")
fmt.Printf("Pending: %d, Processing: %d\n", stats.Pending, stats.Processing)

// Pause/resume queue
client.Queues().Pause(ctx, "my-queue", nil)
client.Queues().Resume(ctx, "my-queue")

// Purge all pending jobs
client.Queues().Purge(ctx, "my-queue")

Real-time Events

Subscribe to real-time job events via WebSocket or SSE:

// WebSocket client
ws := realtime.NewWebSocketClient(realtime.ConnectionOptions{
    WSURL:  "wss://api.spooled.cloud/api/v1/ws",
    Token:  "your-jwt-token",
    APIKey: "sp_live_...",
})

ws.OnJobEvent(realtime.EventJobCompleted, func(event *realtime.JobEvent) {
    fmt.Printf("Job completed: %s\n", event.JobID)
})

ws.OnStateChange(func(state realtime.ConnectionState) {
    fmt.Printf("Connection state: %s\n", state)
})

ws.Connect()
ws.Subscribe(realtime.SubscriptionFilter{QueueName: "my-queue"})

// Clients reconnect with backoff by default (the server closes every stream
// after 30 minutes). Set DisableAutoReconnect: true to turn that off. Events up
// to 4 MiB are read; raise or lower it with ConnectionOptions.MaxMessageBytes.

// SSE client (unidirectional)
sse := realtime.NewSSEClient(realtime.ConnectionOptions{
    BaseURL: "https://api.spooled.cloud",
    APIKey:  "sp_live_...",
})

sse.OnEvent(func(event *realtime.Event) {
    fmt.Printf("Event: %s\n", event.Type)
})

sse.ConnectWithFilter(&realtime.SubscriptionFilter{
    QueueName: "my-queue",
})
// The API key is sent in the Authorization header, never in the URL.
// A job.status_changed event carries OldStatus/NewStatus (and Status = NewStatus).

gRPC (High Performance)

Use gRPC for high-throughput scenarios. TLS is on for port 443 and off for any other port (so localhost:50051 is plaintext); force it either way with grpc.ClientOptions.UseTLS or, through the high-level client, spooled.WithGRPCTLS(true) / spooled.WithGRPCOptions(...) (custom tls.Config, dial options). Payloads and results may hold any JSON-encodable Go value; one that cannot be encoded returns an error instead of being dropped. Fail retries the job (up to its max_retries) unless NoRetry: true.

import "github.com/spooled-cloud/spooled-sdk-go/spooled/grpc"

client, err := grpc.NewClient(grpc.ClientOptions{
    Address: "grpc.spooled.cloud:443",
    APIKey:  "sp_live_...",
})
defer client.Close()

// Enqueue a job
resp, err := client.Enqueue(ctx, &grpc.EnqueueRequest{
    QueueName: "high-throughput",
    Payload:   map[string]any{"data": "value"},
    Priority:  5,
})
fmt.Printf("Job ID: %s\n", resp.JobID)

// Dequeue jobs
jobs, err := client.Dequeue(ctx, &grpc.DequeueRequest{
    QueueName: "high-throughput",
    WorkerID:  "worker-1",
    BatchSize: 10,
})

// Echo each dequeued job's LeaseID on settlement so stale executions are fenced.
for _, job := range jobs.Jobs {
    // Process job...
    err = client.Complete(ctx, &grpc.CompleteRequest{
        JobID: job.ID, WorkerID: "worker-1", LeaseID: job.LeaseID,
    })
}

// Stream jobs (server-side streaming)
stream, err := client.StreamJobs(ctx, "high-throughput", "worker-1")
for {
    job, err := stream.Recv()
    if err == io.EOF {
        break
    }
    // Process job...
}

Webhooks

Configure outgoing webhooks for job events:

// Create webhook. The secret signs every delivery with an X-Spooled-Signature header.
webhook, err := client.Webhooks().Create(ctx, &resources.CreateOutgoingWebhookRequest{
    Name: "Job Notifications",
    URL:  "https://your-app.com/webhooks/spooled",
    Events: []resources.WebhookEvent{
        resources.WebhookEventJobCompleted,
        resources.WebhookEventJobFailed,
    },
    Secret: ptr("whsec_..."),
})

// Test webhook
client.Webhooks().Test(ctx, webhook.ID)

// List deliveries
deliveries, err := client.Webhooks().Deliveries(ctx, webhook.ID, nil)

Webhooks disable themselves. After 20 consecutive failed deliveries the server sets Enabled to false and LastStatus to "auto_disabled", and the webhook stops receiving events. LastStatus is one of "success", "failed" or "auto_disabled". Bring it back with an update:

if wh.LastStatus != nil && *wh.LastStatus == "auto_disabled" {
    // Re-enabling counts against the plan webhook cap, so it can fail with
    // 429 QUOTA_EXCEEDED.
    _, err = client.Webhooks().Update(ctx, wh.ID, &resources.UpdateOutgoingWebhookRequest{
        Enabled: ptr(true),
    })
}

FailureCount is the number of consecutive failed deliveries, counted once per delivery rather than once per retry attempt, so it is roughly five times smaller than a per-attempt count for the same real-world failures - recheck any alerting threshold you built on it. Any successful delivery resets it to 0, including a successful manual RetryDelivery.

Clearing the signing secret is destructive. The secret is three-state on update: leave Secret nil to keep it, set Secret to replace it, or set ClearSecret to remove it. Removing it means deliveries go out unsigned, with no X-Spooled-Signature header for your receiver to verify, and there is no way to recover the old value - set a new Secret to sign again.

// Keeps the current secret: Secret is nil, so the field is left out entirely.
_, err = client.Webhooks().Update(ctx, webhook.ID, &resources.UpdateOutgoingWebhookRequest{
    Events: &newEvents,
})

// Replaces the secret.
_, err = client.Webhooks().Update(ctx, webhook.ID, &resources.UpdateOutgoingWebhookRequest{
    Secret: ptr("whsec_rotated"),
})

// Removes the secret. Deliveries are unsigned from here on.
_, err = client.Webhooks().Update(ctx, webhook.ID, &resources.UpdateOutgoingWebhookRequest{
    ClearSecret: true,
})

Deliveries is not a durable audit log: only the newest 100 deliveries per webhook are readable, and rows are removed once they pass your plan's history retention window (1 day on free, 7 on starter, 30 on pro, 90 on enterprise). Copy anything you need to keep longer.

Organizations

Manage your organization and track usage:

// List organizations
orgs, err := client.Organizations().List(ctx)

// Get organization details
org, err := client.Organizations().Get(ctx, orgID)

// Get usage information
usage, err := client.Organizations().Usage(ctx, orgID)
fmt.Printf("Jobs today: %d/%d\n", usage.Usage.JobsToday.Current, *usage.Limits.MaxJobsPerDay)

API Keys

Manage API keys:

// List API keys
keys, err := client.APIKeys().List(ctx)

// Create new API key
key, err := client.APIKeys().Create(ctx, &resources.CreateAPIKeyRequest{
    Name: "Production Key",
})
fmt.Printf("Key: %s (save this, it won't be shown again)\n", key.Key)

// Revoke API key
client.APIKeys().Revoke(ctx, keyID)

Error Handling

The SDK provides typed errors for different failure scenarios:

import "github.com/spooled-cloud/spooled-sdk-go/spooled"

job, err := client.Jobs().Get(ctx, "invalid-id")
if err != nil {
    switch e := err.(type) {
    case *spooled.NotFoundError:
        fmt.Printf("Job not found: %s\n", e.Message)
    case *spooled.AuthenticationError:
        fmt.Printf("Invalid API key: %s\n", e.Message)
    case *spooled.RateLimitError:
        // 429 covers both per-second rate limiting and plan-quota errors.
        // A plan quota (spooled.IsQuotaExceeded) is never retried by the SDK and
        // never trips the circuit breaker; its details are in e.Quota.
        if e.IsQuotaExceeded() {
            fmt.Printf("Plan quota exceeded: %s (%d/%d on the %s plan)\n", e.Message, e.Quota.Current, e.Quota.Limit, e.Quota.Plan)
        } else {
            fmt.Printf("Rate limited. Retry after: %v\n", e.Reset)
        }
    case *spooled.ValidationError:
        fmt.Printf("Invalid request: %v\n", e.Errors)
    default:
        fmt.Printf("Error: %v\n", err)
    }
}

// Methods whose operation the API does not offer (Ingest().GitHub/Stripe,
// Jobs().UpdateProgress, Organizations().ClearWebhookToken, JobContext.Progress)
// fail without a network call:
if errors.Is(err, spooled.ErrUnsupported) {
    // not supported by the Spooled API
}

// Check error types
if spooled.IsRetryableError(err) {
    // Safe to retry
}
if spooled.IsNetworkError(err) {
    // Network issue
}
if spooled.IsTimeoutError(err) {
    // Request timed out
}

Development

Using Make

# Build, lint, and test
make

# Run specific targets
make build          # Build the SDK
make test           # Run unit tests
make test-coverage  # Tests with coverage report
make lint           # Run linter
make fmt            # Format code
make examples       # Build all examples

# Integration tests (requires API_KEY)
API_KEY=sp_live_... make integration-test

# See all commands
make help

Testing

# Unit tests
go test ./...

# With race detection
go test -race ./...

# With coverage
go test -cover ./...

# Integration tests (requires API key)
API_KEY=sp_live_... go run scripts/test-local/main.go

# Full production test suite (all features)
API_KEY=sp_live_... \
ADMIN_KEY=your_admin_key \
BASE_URL=https://api.spooled.cloud \
GRPC_ADDRESS=grpc.spooled.cloud:443 \
SKIP_GRPC=0 \
SKIP_STRESS=0 \
go run scripts/test-local/main.go

Contributing

Contributions are welcome! Please see this repository's CONTRIBUTING.md.

License

Apache 2.0 - see LICENSE for details.

Support

• 📖 Documentation
• 🐛 Issue Tracker
• 📧 Email Support