corejobs

README

Core Jobs

This package defines job types used by the Openlane API (core) with Riverqueue. Due to cyclical dependency issues, any job that uses the openlane API, needs to be defined in this repo, and just configured in riverboat.

Usage

1. Add new jobs to this directory in a new file, refer to the upstream docs for implementation details. The following is a stem job that could be copied to get you started.

   package jobs
   
   import (
      "context"
   
      "github.com/riverqueue/river"
      "github.com/rs/zerolog/log"
   )
   
   // ExampleArgs for the example worker to process the job
   type ExampleArgs struct {
      // ExampleArg is an example argument
      ExampleArg string `json:"example_arg"`
   }
   
   // Kind satisfies the river.Job interface
   func (ExampleArgs) Kind() string { return "example" }
   
   // ExampleWorker does all sorts of neat stuff
   type ExampleWorker struct {
      river.WorkerDefaults[ExampleArgs]
   
      ExampleConfig
   }
   
   // ExampleConfig contains the configuration for the example worker
   type ExampleConfig struct {
      // DevMode is a flag to enable dev mode so we don't actually send millions of carrier pigeons
      DevMode bool `koanf:"devMode" json:"devMode" jsonschema:"description=enable dev mode" default:"true"`
   }
   
   // Work satisfies the river.Worker interface for the example worker
   func (w *ExampleConfig) Work(ctx context.Context, job *river.Job[ExampleArgs]) error {
      // do some work
   
      return nil
   }
   

2. Add a test for the new job. There are additional helper functions that can be used, see river test helpers for details.

3. Add a test job to test/ directory by creating a new directory with a main.go function that will insert the job into the queue.

4. Workers should be registered in the riverboat

Test Jobs

Included in the test/ directory are test jobs corresponding to the job types in pkg/jobs.

1. Start the riverboat server using task run-dev

2. Run the test main, for example the email:

   go run test/email/main.go
   

3. This should insert the job successfully, it should be processed by river and the email should be added to fixtures/email

Contributing

See the contributing guide for more information.

Documentation

Overview

Package corejobs holds jobs definitions for riverqueue that interact with the core api

Index

• Variables
• type CreateCustomDomainArgs
  • func (CreateCustomDomainArgs) Kind() string
• type CreateCustomDomainWorker
  • func (w *CreateCustomDomainWorker) WithCloudflareClient(cl intcloudflare.Client) *CreateCustomDomainWorker
  • func (w *CreateCustomDomainWorker) WithOpenlaneClient(cl olclient.OpenlaneClient) *CreateCustomDomainWorker
  • func (w *CreateCustomDomainWorker) WithRiverClient(cl riverqueue.JobClient) *CreateCustomDomainWorker
  • func (w *CreateCustomDomainWorker) Work(ctx context.Context, job *river.Job[CreateCustomDomainArgs]) error
• type CustomDomainConfig
• type DeleteCustomDomainArgs
  • func (DeleteCustomDomainArgs) Kind() string
• type DeleteCustomDomainWorker
  • func (w *DeleteCustomDomainWorker) WithCloudflareClient(cl intcloudflare.Client) *DeleteCustomDomainWorker
  • func (w *DeleteCustomDomainWorker) WithOpenlaneClient(cl olclient.OpenlaneClient) *DeleteCustomDomainWorker
  • func (w *DeleteCustomDomainWorker) Work(ctx context.Context, job *river.Job[DeleteCustomDomainArgs]) error
• type MissingRequiredArgError
  • func (e *MissingRequiredArgError) Error() string
• type ValidateCustomDomainArgs
  • func (ValidateCustomDomainArgs) Kind() string
• type ValidateCustomDomainWorker
  • func (w *ValidateCustomDomainWorker) WithCloudflareClient(cl intcloudflare.Client) *ValidateCustomDomainWorker
  • func (w *ValidateCustomDomainWorker) WithOpenlaneClient(cl olclient.OpenlaneClient) *ValidateCustomDomainWorker
  • func (w *ValidateCustomDomainWorker) Work(ctx context.Context, job *river.Job[ValidateCustomDomainArgs]) error

Variables

var (
	// ErrDomainVerificationAlreadyExists is returned when a custom domain already has a verification member
	ErrDomainVerificationAlreadyExists = errors.New("custom domain already has a verification member")
)

Types

type CreateCustomDomainArgs

type CreateCustomDomainArgs struct {
	// ID of the custom domain in our system
	CustomDomainID string `json:"custom_domain_id"`
}

CreateCustomDomainArgs for the worker to process the custom domain

func (CreateCustomDomainArgs) Kind

func (CreateCustomDomainArgs) Kind() string

Kind satisfies the river.Job interface

type CreateCustomDomainWorker

type CreateCustomDomainWorker struct {
	river.WorkerDefaults[CreateCustomDomainArgs]

	Config CustomDomainConfig
	// contains filtered or unexported fields
}

CreateCustomDomainWorker creates a custom hostname in cloudflare, and creates and updates the records in our system

func (*CreateCustomDomainWorker) WithCloudflareClient

func (w *CreateCustomDomainWorker) WithCloudflareClient(cl intcloudflare.Client) *CreateCustomDomainWorker

WithCloudflareClient sets the Cloudflare client for the worker and returns the worker for method chaining

func (*CreateCustomDomainWorker) WithOpenlaneClient

func (w *CreateCustomDomainWorker) WithOpenlaneClient(cl olclient.OpenlaneClient) *CreateCustomDomainWorker

WithOpenlaneClient sets the Openlane client for the worker and returns the worker for method chaining

func (*CreateCustomDomainWorker) WithRiverClient

func (w *CreateCustomDomainWorker) WithRiverClient(cl riverqueue.JobClient) *CreateCustomDomainWorker

WithRiverClient sets the River client for the worker and returns the worker for method chaining

func (*CreateCustomDomainWorker) Work

func (w *CreateCustomDomainWorker) Work(ctx context.Context, job *river.Job[CreateCustomDomainArgs]) error

Work satisfies the river.Worker interface for the create custom domain worker it creates a custom domain for an organization

type CustomDomainConfig

type CustomDomainConfig struct {
	CloudflareAPIKey string `koanf:"cloudflareApiKey" json:"cloudflareApiKey" jsonschema:"required description=the cloudflare api key"`

	OpenlaneAPIHost  string `koanf:"openlaneAPIHost" json:"openlaneAPIHost" jsonschema:"required description=the openlane api host"`
	OpenlaneAPIToken string `koanf:"openlaneAPIToken" json:"openlaneAPIToken" jsonschema:"required description=the openlane api token"`

	DatabaseHost string `koanf:"databaseHost" json:"databaseHost" jsonschema:"required description=the database host"`
}

CustomDomainConfig contains the configuration for the custom domain workers

type DeleteCustomDomainArgs

type DeleteCustomDomainArgs struct {
	// ID of the custom domain in our system
	CustomDomainID             string `json:"custom_domain_id"`
	DNSVerificationID          string `json:"dns_verification_id"`
	CloudflareCustomHostnameID string `json:"cloudflare_custom_hostname_id"`
	CloudflareZoneID           string `json:"cloudflare_zone_id"`
}

DeleteCustomDomainArgs for the worker to process the custom domain

func (DeleteCustomDomainArgs) Kind

func (DeleteCustomDomainArgs) Kind() string

Kind satisfies the river.Job interface

type DeleteCustomDomainWorker

type DeleteCustomDomainWorker struct {
	river.WorkerDefaults[DeleteCustomDomainArgs]

	Config CustomDomainConfig
	// contains filtered or unexported fields
}

DeleteCustomDomainWorker delete the custom hostname from cloudflare and updates the records in our system

func (*DeleteCustomDomainWorker) WithCloudflareClient

func (w *DeleteCustomDomainWorker) WithCloudflareClient(cl intcloudflare.Client) *DeleteCustomDomainWorker

WithCloudflareClient sets the Cloudflare client for the worker and returns the worker for method chaining

func (*DeleteCustomDomainWorker) WithOpenlaneClient

func (w *DeleteCustomDomainWorker) WithOpenlaneClient(cl olclient.OpenlaneClient) *DeleteCustomDomainWorker

WithOpenlaneClient sets the Openlane client for the worker and returns the worker for method chaining

func (*DeleteCustomDomainWorker) Work

func (w *DeleteCustomDomainWorker) Work(ctx context.Context, job *river.Job[DeleteCustomDomainArgs]) error

Work satisfies the river.Worker interface for the delete custom domain worker it deletes a custom domain for an organization

type MissingRequiredArgError

type MissingRequiredArgError struct {
	// Arg is the required argument that was not provided
	Arg string `json:"arg"`
	// Job is the job that requires the argument
	Job string `json:"job"`
}

MissingRequiredArgError is returned when a required argument was not provided to a job

func (*MissingRequiredArgError) Error

func (e *MissingRequiredArgError) Error() string

Error returns the MissingRequiredArgError in string format

type ValidateCustomDomainArgs

type ValidateCustomDomainArgs struct {
	CustomDomainID string `json:"custom_domain_id"`
}

ValidateCustomDomainArgs for the worker to process the custom domain

func (ValidateCustomDomainArgs) Kind

func (ValidateCustomDomainArgs) Kind() string

Kind satisfies the river.Job interface

type ValidateCustomDomainWorker

type ValidateCustomDomainWorker struct {
	river.WorkerDefaults[ValidateCustomDomainArgs]

	Config CustomDomainConfig
	// contains filtered or unexported fields
}

ValidateCustomDomainWorker checks cloudflare custom domain(s), and updates the status in our system

func (*ValidateCustomDomainWorker) WithCloudflareClient

func (w *ValidateCustomDomainWorker) WithCloudflareClient(cl intcloudflare.Client) *ValidateCustomDomainWorker

func (*ValidateCustomDomainWorker) WithOpenlaneClient

func (w *ValidateCustomDomainWorker) WithOpenlaneClient(cl olclient.OpenlaneClient) *ValidateCustomDomainWorker

func (*ValidateCustomDomainWorker) Work

func (w *ValidateCustomDomainWorker) Work(ctx context.Context, job *river.Job[ValidateCustomDomainArgs]) error

ValidateCustomDomainConfig contains the configuration for the worker Work satisfies the river.Worker interface for the validate custom domain worker. It validates custom domains by checking their status in Cloudflare and updating our system with the current verification status.