Arandu Async

Background work, events and notifications

When to use

Anything that runs after the response, on a schedule, or because something happened: a job, a scheduled task, an event and its listeners, a notification, a mail, a console command. The service that a job or a listener calls is arandu-module; an external system it reaches is arandu-integrations.

Before you start

• Read the example's chain: NoteService.Publish stores note.published; app/Listeners/NotifyNoteAuthor.go reacts and sends app/Notifications/NotePublished.go; AppServiceProvider.Schedule enqueues app/Jobs/SendNotesDigest.go every night, and the worker runs it.
• Know where each runs. The relay and the scheduler run inside aru serve; jobs run in aru queue:work, the same binary with another argument.

Contracts and imports

Procedure

1. A fact that happened is an event. aru make:event NotePublished --aggregate=note --event-name=note.published --fields "title:string". The service stores it in the transaction of the write it describes, so an event is never stored for a write that rolled back, and a write is never committed without its event.
2. A reaction to it is a listener. aru make:listener NotifyNoteAuthor --event=note.published, then add it to listeners.Each in bootstrap/app.go. The relay calls it after the commit, at least once: what it does has to be safe to do twice -- an upsert keyed by e.ID, a call with an idempotency key -- and an error means "try again later", never "skip".
3. Telling a person is a notification. aru make:notification NotePublished --channels=database writes the notification and its test; it is sent with the Notifier bootstrap builds (app.Notifier), from the listener or the service that decided to tell somebody.
4. Work worth retrying is a job. aru make:job SendNotesDigest --event-name=notes.digest --fields "since:timestamp". The caller that holds the queue dispatches it: a listener, after the write that stored its event has committed, or a task in AppServiceProvider.Schedule, as the digest is. A service never does -- the job's package imports app/Services once its handler takes a service, and a service importing it back is an import cycle. The handler calls a service; j.UUID is stable across retries and is the key to deduplicate on.
5. Recurring work is a task that enqueues a job. Declare it in AppServiceProvider.Schedule; the task decides what work exists and the job does it, with the retries a task does not get. A task that reads a tenant's rows is PerTenant, and runs for each tenant the scheduler's Tenants returns.
6. An operator's action is a command. aru make:command and a line in routes/console.go.

Commands

• aru make:job <Name> --event-name=<name> --fields "..." [--services=<Entity>,...]
• aru make:event <Name> --aggregate=<entity> --event-name=<entity.verb> --fields "..."
• aru make:listener <Name> --event=<entity.verb>
• aru make:notification <Name> --channels=database
• aru make:mail <Name> --subject "..." --fields "...", aru make:command <Name> --signature=notes:prune --description="..."
• aru schedule:list, aru schedule:run <id> --tenant=<id>
• aru queue:work, aru queue:failed --tenant=<id>, aru queue:retry --tenant=<id>

Example

A listener that reacts to the example's event, and a task that enqueues the example's job every Monday:

package example

import (
	"context"
	"time"

	"github.com/arandu-io/hesape/auth"
	"github.com/arandu-io/hesape/events"
	hfoundation "github.com/arandu-io/hesape/foundation"
	"github.com/arandu-io/hesape/log"
	"github.com/arandu-io/hesape/queue"

	appevents "<module>/app/Events"
	appjobs "<module>/app/Jobs"
	policies "<module>/app/Policies"
)

// LogPublication writes a line for every published note, after the write
// committed. A repeat writes the line again, which is safe.
type LogPublication struct{}

var _ events.Publisher = LogPublication{}

// Publish answers note.published and nothing else.
func (LogPublication) Publish(ctx context.Context, e events.Stored) error {
	if e.Name != appevents.NotePublishedName {
		return nil
	}
	var published struct {
		Title string `json:"title"`
	}
	if err := e.Decode(&published); err != nil {
		return err
	}
	log.For(ctx).Info("note published", "note", e.AggregateID, "title", published.Title)
	return nil
}

// WeeklyDigest decides that a digest is due, per tenant, and enqueues it.
func WeeklyDigest(q queue.Queue) hfoundation.Task {
	return hfoundation.Task{
		ID:        "notes.weekly-digest",
		Spec:      "0 8 * * 1",
		Scope:     hfoundation.PerTenant,
		Singleton: true,
		Timeout:   time.Minute,
		Action:    policies.NoteList,
		Run: func(ctx context.Context, g auth.Grant) error {
			return appjobs.DispatchSendNotesDigest(ctx, q, g, appjobs.SendNotesDigest{
				Since: time.Now().AddDate(0, 0, -7),
			})
		},
	}
}