Reports Generate Job

Verified Scenario

This page is generated from an executable spec. An automated suite renders a fresh App from the current GoForj templates, applies every step below in order, and runs every verification command. If any step fails, the page does not ship.

Scenario 5 of 7 in the verified path. Plan on about 25 minutes.

This scenario turns the users.created subscriber into a queue-backed work dispatcher.

The event still announces that a user was created. The subscriber now queues reports:generate, and a queue worker generates a small report artifact. This keeps event fan-out separate from work that can gain persistence, retries, and cross-process delivery through a durable queue driver.

What You Will Build

• QUEUE_* config selects the queue backend used by API and worker processes.
• STORAGE_REPORTS_* defines a named disk for generated report artifacts.
• reports.Service writes a report file to storage.
• reports.GenerateJob owns the queue payload, dispatch shape, and handler.
• notifications.Service dispatches the job from the users.created subscriber.
• Wire binds the job to a small queueing interface used by notifications.

Prerequisites

Complete Users Created Event first.

The generated App should have queues, jobs, events, and storage enabled.

Golden Path State

Before this scenario, users.created is an in-process fact with a subscriber reaction.

After this scenario, the subscriber dispatches named background work, workers process reports:generate, and generated report files land on the named reports storage disk. Event fan-out and job execution are now separate boundaries, while the selected queue driver determines whether jobs survive process failure and cross process boundaries.

Files

This scenario edits or creates:

Configuration

.env

Reports feature

internal/reports/service.go
internal/reports/service_test.go

Jobs

internal/reports/generate_job.go
internal/reports/generate_job_test.go
app/wire/inject_jobs_app.go

Notifications

internal/notifications/service.go

App wiring

app/wire/inject_services_app.go

The queue and storage generators update generated manager and accessor files.

internal/queues/manager_gen.go
internal/storages/accessors_gen.go
internal/storages/manager_gen.go

Do not edit generated queue or storage files by hand.

Step 1: Configure Report Storage

Use a named local storage disk for report artifacts. The default workerpool queue stays inside the App process; a separate worker requires the shared driver described below.

Append to .env:

STORAGE_REPORTS_DRIVER=local
STORAGE_REPORTS_ROOT=storage/app/reports
STORAGE_REPORTS_PREFIX=

Step 2: Enable the Sync Queue for Tests

Compile the synchronous queue driver so the handler test can exercise payload binding deterministically without starting external infrastructure.

Update .env so it includes:

QUEUE_SUPPORTED_DRIVERS=workerpool,redis,sync

Step 3: Refresh Generated Resources

Run the build pipeline so the generated App exposes app.Queue() and app.Storage().Reports().

forj build

Step 4: Add the Report Service

Create internal/reports/service.go.

The service writes through storage.Storage, not a local filesystem or cloud SDK. The selected driver remains configuration.

Create or replace internal/reports/service.go:

// Package reports owns report generation and the queue boundary used to request it.
package reports

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"path"
	"strings"
	"time"

	"github.com/goforj/storage"
)

var (
	// ErrUserIDRequired identifies an invalid report request before storage is touched.
	ErrUserIDRequired = errors.New("user id is required")
	// ErrEmailRequired identifies an invalid report request before storage is touched.
	ErrEmailRequired = errors.New("email is required")
)

// Service writes report artifacts through a configured storage disk rather than a concrete backend.
type Service struct {
	disk storage.Storage
}

// ReportQueue keeps report requesters independent of queue payloads and dispatch policy.
type ReportQueue interface {
	// Queue moves report generation behind the configured worker lifecycle.
	Queue(ctx context.Context, userID string, email string) error
}

// UserReport is the stable artifact stored by the runnable report workflow.
type UserReport struct {
	UserID      string    `json:"user_id"`
	Email       string    `json:"email"`
	GeneratedAt time.Time `json:"generated_at"`
}

// NewService requires the named report disk because successful generation must persist an artifact.
func NewService(disk storage.Storage) *Service {
	return &Service{disk: disk}
}

// GenerateForUser validates path and identity data before writing one deterministic report location.
func (s *Service) GenerateForUser(ctx context.Context, userID string, email string) (string, error) {
	userID = reportPathSegment(userID)
	if userID == "" {
		return "", ErrUserIDRequired
	}

	email = strings.TrimSpace(email)
	if email == "" {
		return "", ErrEmailRequired
	}

	report := UserReport{
		UserID:      userID,
		Email:       email,
		GeneratedAt: time.Now().UTC(),
	}

	body, err := json.MarshalIndent(report, "", "  ")
	if err != nil {
		return "", fmt.Errorf("encode user report: %w", err)
	}

	reportPath := path.Join("users", userID, "profile.json")
	if err := s.disk.WithContext(ctx).Put(reportPath, body); err != nil {
		return "", fmt.Errorf("store user report: %w", err)
	}

	return reportPath, nil
}

// reportPathSegment prevents an external identifier from escaping the named report disk prefix.
func reportPathSegment(value string) string {
	value = strings.TrimSpace(value)
	value = path.Base(strings.ReplaceAll(value, "\\", "/"))
	value = strings.Trim(value, ".")
	return value
}

Step 5: Generate the Job

Use the generated App's make command to create the job file and add its constructor to job wiring.

forj make:job reports:generate --output-dir ./internal/jobs

Step 6: Replace the Generated Job

Replace internal/reports/generate_job.go.

The grouped generator keeps the job beside the report service. The job owns the queue payload and dispatch options, while its handler delegates report behavior to Service.

Create or replace internal/reports/generate_job.go:

// Package reports owns report generation and the queue boundary used to request it.
package reports

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/goforj/queue"

	"your/module/internal/queues"
)

// GenerateJobTypeName is the stable queue identity shared by dispatchers and workers.
const GenerateJobTypeName = "reports:generate"

// GeneratePayload keeps queued data small so report artifacts remain in storage rather than the queue.
type GeneratePayload struct {
	UserID string `json:"user_id"`
	Email  string `json:"email"`
}

// GenerateJob owns report dispatch policy and translates queue messages into service calls.
type GenerateJob struct {
	queues  *queues.Manager
	service *Service
}

// NewGenerateJob requires both queue and report collaborators because it serves producer and worker paths.
func NewGenerateJob(queues *queues.Manager, service *Service) *GenerateJob {
	return &GenerateJob{
		queues:  queues,
		service: service,
	}
}

// Queue serializes the stable payload and applies retry and timeout policy at the job boundary.
func (j *GenerateJob) Queue(ctx context.Context, userID string, email string) error {
	payload, err := json.Marshal(GeneratePayload{
		UserID: userID,
		Email:  email,
	})
	if err != nil {
		return fmt.Errorf("encode generate report payload: %w", err)
	}

	_, err = j.queues.WithContext(ctx).Dispatch(
		queue.NewJob(GenerateJobTypeName).
			Payload(payload).
			OnQueue("default").
			Retry(3).
			Timeout(2 * time.Minute),
	)
	if err != nil {
		return fmt.Errorf("dispatch %s job: %w", GenerateJobTypeName, err)
	}
	return nil
}

// HandleTask binds only the queue contract before delegating report generation to the service.
func (j *GenerateJob) HandleTask(ctx context.Context, msg queue.Message) error {
	var payload GeneratePayload
	if err := msg.Bind(&payload); err != nil {
		return fmt.Errorf("bind generate report payload: %w", err)
	}

	if _, err := j.service.GenerateForUser(ctx, payload.UserID, payload.Email); err != nil {
		return fmt.Errorf("generate user report: %w", err)
	}
	return nil
}

Step 7: Share the Job with App Services

Handler registration remains in app/wire/inject_jobs_app.go, but the App service set owns construction so Wire can bind GenerateJob to the smaller ReportQueue interface used by notifications.

Remove from app/wire/inject_jobs_app.go:

reports.NewGenerateJob,

Step 8: Dispatch the Job from Notifications

Replace internal/notifications/service.go.

The application-facing method remains stable while its implementation moves work behind the queue boundary.

Create or replace internal/notifications/service.go:

// Package notifications owns reactions to application facts without coupling publishers to their effects.
package notifications

import (
	"context"

	"your/module/internal/reports"
)

// Service keeps event reactions application-facing while report execution moves behind a queue.
type Service struct {
	generateReport reports.ReportQueue
}

// NewService requires the report queue because user-created reactions now dispatch background work.
func NewService(generateReport reports.ReportQueue) *Service {
	return &Service{generateReport: generateReport}
}

// HandleUserCreated dispatches report work without making the event subscriber understand queue details.
func (s *Service) HandleUserCreated(ctx context.Context, userID string, email string) error {
	return s.generateReport.Queue(ctx, userID, email)
}

Step 9: Keep Lifecycle Subscriber Registration

Update app/lifecycle.go.

The lifecycle still owns subscriber registration. The subscriber dispatches queue-backed work through the notification service.

Create or replace app/lifecycle.go:

// Package app owns application composition and lifecycle hooks.
package app

import (
	"context"

	"your/module/internal/events"
	"your/module/internal/notifications"
	"your/module/internal/runtime"
)

// LifecycleRegistry keeps subscription ownership aligned with App startup and shutdown ordering.
type LifecycleRegistry struct {
	eventManager             *events.Manager
	notificationSubscribers  *notifications.Subscribers
	notificationSubscription events.Subscription
}

// NewLifecycleRegistry requires the generated event manager and the App's subscriber collection.
func NewLifecycleRegistry(
	eventManager *events.Manager,
	notificationSubscribers *notifications.Subscribers,
) *LifecycleRegistry {
	return &LifecycleRegistry{
		eventManager:            eventManager,
		notificationSubscribers: notificationSubscribers,
	}
}

// Register starts subscriptions after event buses and closes them before those buses shut down.
func (r *LifecycleRegistry) Register(lifecycle *runtime.Lifecycle) {
	lifecycle.On(runtime.Startup, r.Startup)
	lifecycle.On(runtime.Shutdown, r.Shutdown)
}

// Startup retains the subscription handle so shutdown can release the exact registered consumer.
func (r *LifecycleRegistry) Startup(ctx context.Context) error {
	subscription, err := r.notificationSubscribers.Register(ctx, r.eventManager.Default())
	if err != nil {
		return err
	}
	r.notificationSubscription = subscription
	return nil
}

// Shutdown closes the subscriber before the generated lifecycle stops its event bus.
func (r *LifecycleRegistry) Shutdown(_ context.Context) error {
	return r.notificationSubscription.Close()
}

Step 10: Add Report Imports

Add the report service import. The generated storage manager was already introduced by the upload scenario.

Update app/wire/inject_services_app.go so it includes:

"your/module/internal/notifications"
"your/module/internal/reports"

Step 11: Add Report Providers

The report service receives the named reports storage disk without adding another unqualified storage.Storage provider to Wire.

Update app/wire/inject_services_app.go so it includes:

provideReportService,
reports.NewGenerateJob,
wire.Bind(new(reports.ReportQueue), new(*reports.GenerateJob)),
provideEventBus,

Step 12: Add the Report Service Provider

provideReportService selects the generated named disk at the composition root.

Update app/wire/inject_services_app.go so it includes:

// provideReportService selects the named disk where dependencies are composed instead of inside report behavior.
func provideReportService(manager *storages.Manager) *reports.Service {
        return reports.NewService(manager.Reports())
}

// provideEventBus exposes the default generated bus without coupling the publisher to its manager.
func provideEventBus(manager *events.Manager) events.Bus {

Step 13: Add a Report Service Test

Create internal/reports/service_test.go.

This test verifies the report behavior without starting the queue worker.

Create or replace internal/reports/service_test.go:

// Package reports owns report generation and the queue boundary used to request it.
package reports

import (
	"context"
	"encoding/json"
	"errors"
	"testing"

	"github.com/goforj/storage"
	"github.com/goforj/storage/driver/memorystorage"
)

// newTestDisk keeps test setup focused on report behavior while failing immediately on invalid storage wiring.
func newTestDisk(t *testing.T) storage.Storage {
	t.Helper()

	disk, err := storage.Build(memorystorage.Config{})
	if err != nil {
		t.Fatalf("build storage: %v", err)
	}
	return disk
}

// readTestReport keeps artifact decoding consistent across service and handler tests.
func readTestReport(t *testing.T, disk storage.Storage, reportPath string) UserReport {
	t.Helper()

	body, err := disk.WithContext(context.Background()).Get(reportPath)
	if err != nil {
		t.Fatalf("read report: %v", err)
	}

	var report UserReport
	if err := json.Unmarshal(body, &report); err != nil {
		t.Fatalf("decode report: %v", err)
	}
	return report
}

// TestServiceGeneratesUserReport verifies both the stable storage path and the artifact contract.
func TestServiceGeneratesUserReport(t *testing.T) {
	ctx := context.Background()
	disk := newTestDisk(t)
	service := NewService(disk)
	reportPath, err := service.GenerateForUser(ctx, "42", "ada@example.test")
	if err != nil {
		t.Fatalf("generate report: %v", err)
	}
	if reportPath != "users/42/profile.json" {
		t.Fatalf("report path = %q", reportPath)
	}

	report := readTestReport(t, disk, reportPath)
	if report.UserID != "42" {
		t.Fatalf("report user id = %q, want %q", report.UserID, "42")
	}
	if report.Email != "ada@example.test" {
		t.Fatalf("report email = %q, want %q", report.Email, "ada@example.test")
	}
	if report.GeneratedAt.IsZero() {
		t.Fatal("expected report generation time")
	}
}

// TestServiceRejectsInvalidReports keeps malformed identity data from reaching storage.
func TestServiceRejectsInvalidReports(t *testing.T) {
	ctx := context.Background()
	service := NewService(newTestDisk(t))
	tests := []struct {
		name    string
		userID  string
		email   string
		wantErr error
	}{
		{
			name:    "missing user id",
			email:   "ada@example.test",
			wantErr: ErrUserIDRequired,
		},
		{
			name:    "missing email",
			userID:  "42",
			wantErr: ErrEmailRequired,
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			_, err := service.GenerateForUser(ctx, test.userID, test.email)
			if !errors.Is(err, test.wantErr) {
				t.Fatalf("GenerateForUser() error = %v, want %v", err, test.wantErr)
			}
		})
	}
}

Step 14: Add a Generate Job Handler Test

Create internal/reports/generate_job_test.go.

The synchronous queue supplies the real queue.Message contract, so the test proves payload binding and delegation without testing private fields or starting external infrastructure.

Create or replace internal/reports/generate_job_test.go:

// Package reports owns report generation and the queue boundary used to request it.
package reports

import (
	"context"
	"testing"

	"github.com/goforj/queue"

	"your/module/internal/queues"
)

// TestGenerateJobHandlesPayload proves a queue message is bound and delegated to report generation.
func TestGenerateJobHandlesPayload(t *testing.T) {
	ctx := context.Background()
	t.Setenv("QUEUE_DRIVER", "sync")

	queueManager, err := queues.NewManager()
	if err != nil {
		t.Fatalf("build queue manager: %v", err)
	}
	runtimeQueue := queueManager.Default()
	t.Cleanup(func() {
		if err := runtimeQueue.Shutdown(context.Background()); err != nil {
			t.Errorf("shutdown queue: %v", err)
		}
	})

	disk := newTestDisk(t)
	job := NewGenerateJob(queueManager, NewService(disk))
	queueManager.Register(GenerateJobTypeName, job.HandleTask)
	if err := runtimeQueue.StartWorkers(ctx); err != nil {
		t.Fatalf("start queue workers: %v", err)
	}

	_, err = queueManager.WithContext(ctx).Dispatch(
		queue.NewJob(GenerateJobTypeName).
			Payload([]byte(`{"user_id":"42","email":"ada@example.test"}`)).
			OnQueue("default"),
	)
	if err != nil {
		t.Fatalf("dispatch generate report job: %v", err)
	}

	report := readTestReport(t, disk, "users/42/profile.json")
	if report.UserID != "42" {
		t.Fatalf("report user id = %q, want %q", report.UserID, "42")
	}
	if report.Email != "ada@example.test" {
		t.Fatalf("report email = %q, want %q", report.Email, "ada@example.test")
	}
}

Build and Verify

forj build

go test ./...

forj route:list

Expected output includes:

• /api/v1/users

Try the Route

The default workerpool driver is process-local, so start the combined App to keep the API and Jobs runtime together:

forj app

Create a user:

curl -X POST http://localhost:3000/api/v1/users \
  -H 'Content-Type: application/json' \
  -d '{"name":"Grace Hopper","email":"grace@example.test"}'

The API publishes users.created. The subscriber dispatches reports:generate. The in-process worker consumes the job and writes storage/app/reports/users/43/profile.json.

To run API and worker as separate processes, first select a shared queue driver such as Redis using the configuration below. Then start the API and worker in separate terminals:

forj api

forj worker

Operations

Operational notes:

• Queued jobs leave the HTTP request path and can use retry, delay, timeout, and worker lifecycle policies.
• Durability and cross-process delivery come from the selected queue driver; workerpool remains process-local.
• Use this boundary for work that sends email, generates reports, calls external APIs, or may need operational recovery.
• The job can appear in queue metrics, inspect records, Lighthouse queue views, worker logs, and driver backend state.
• Keep job payloads stable and small. Store large artifacts in storage, not inside queue payloads.

Swap the Driver

To use Redis in production, compile Redis queue support and select it. Keep sync compiled because the handler test selects it explicitly:

QUEUE_SUPPORTED_DRIVERS=workerpool,redis,sync
QUEUE_DRIVER=redis
QUEUE_ADDR=redis:6379
QUEUE_NAME=default
QUEUE_WORKERS=30

Then run:

forj build

Business code does not change. GenerateJob still dispatches reports:generate, and workers still run with forj worker or ./bin/app worker.

Common Mistakes

Common mistakes

• Do not use an in-process queue when API and worker run as separate processes.
• Do not put report generation logic directly inside the event subscriber.
• Do not make HTTP controllers build raw queue jobs.
• Do not register job handlers after workers have started.
• Do not put full report contents into the queue payload.
• Do not assume retries are safe unless the handler is idempotent.

Next Steps

• Next, schedule recurring report dispatch with Reports Daily Schedule.