Building Custom Workers

A custom worker is a queue-backed gRPC service assembled from the shared github.com/mediamoth/go-libs/workerservice components and the runtime pieces used by Video Service, Chapter Inserter, and VHS Service. The legacy mediamoth/pkg/worker processor API is not available.

Runtime contract

A worker service implements the protobuf ServicesInterface surface:

• QueueJob validates and inserts typed River arguments;
• RequeueJob inserts again and cancels a stuck running duplicate when needed;
• CancelJob cancels active River rows for the job ID;
• GetServiceMetadata publishes the service's parameter schema, UI schema, and version.

Job events are received from Kafka and forwarded to the worker's gRPC server through workerservice.JobEventHandler.

Implement a validator

The shared server needs a validator that converts a protobuf request into typed river.JobArgs:

type JobValidator interface {
    ValidateJobRequest(req workerservice.JobRequest) (river.JobArgs, error)
}

Define a typed parameter structure, validate it, and return a River job kind:

type Parameters struct {
    JobID       string               `json:"job_id" river:"unique" validate:"required" jsonschema:"-"`
    AggregateID string               `json:"aggregate_id" validate:"required" jsonschema:"-"`
    InputSource files.FileSourcePath `json:"input_source" validate:"required"`
    OutputSource files.FileSourcePath `json:"output_source" validate:"required"`
}

func (Parameters) Kind() string { return "my_worker_job" }

Use the service implementations in video-service/internal/server, chapter-inserter/internal/server, or vhs-service/internal/*/server as working examples of request decoding and validation.

Construct the shared gRPC server

After creating a River client and validator:

workerServer := workerservice.NewWorkerService(riverClient, validator)

Embed or wrap that server in your protobuf implementation and register it with gRPC. Register the standard gRPC health server too.

Run jobs with River

Register a River worker for the same Kind() returned by your parameters. The worker should:

1. stage input files through util/files;
2. perform the processing operation;
3. write outputs through the selected file provider;
4. publish start, progress, completion, or failure events;
5. honor cancellation from the job context;
6. clean temporary files on every return path.

Shared queue helpers insert jobs with MaxAttempts: 1 and uniqueness by arguments. Retry policy belongs to MediaMoth's explicit requeue flow, not an implicit River retry loop.

Publish schemas through health checks

Current workers build JSON Schema and UI Schema from their parameter types and attach both to service metadata:

parameterSchema, err := jsonschema.BuildParameterSchema[Parameters]()
uiSchema, err := jsonschema.BuildUISchema[Parameters]()

Workflow Service stores schema updates from heartbeats. That lets clients render job forms from the worker's actual parameters and version rather than maintaining a separate form definition.

Required service wiring

A deployable worker normally needs:

• PostgreSQL and a River queue;
• a Kafka consumer for job events;
• a Kafka publisher for job status and health events;
• a gRPC server implementing ServicesInterface and gRPC health;
• a configuration section for database, Kafka, River, health interval, and service address;
• a serve, worker, or combined command;
• Compose, database, topic, deployment, and mise task entries.

There is no single constructor that supplies all of this. Copy the shape of the closest built-in service and replace its domain parameters and River work function.

Validation checklist

• Parameter JSON is strongly typed and schema generation passes.
• File fields use validated file://, rsync://, or s3:// URIs.
• Queue, requeue, and cancellation paths are tested.
• The worker publishes progress and terminal status.
• The same service ID is used in configuration, metadata, pipeline nodes, and topic wiring.
• Server-only, worker-only, and combined startup modes behave as documented.

Related documentation

• API Reference
• File Sources
• Key Concepts