How to implement a daily job for each user object?

[DavidS-ovm]

In my app, I need to run a cleanup job for each user once a day. Preferably splayed across 24hrs to avoid peak loads. Users get added and removed regularly.

What I came up with was a pair of workers:

• the CleanupWorker cleans up a single user, based on ID.
• the CleanupSchedulingWorker is scheduled every 15 minutes to check for users with an expired last_cleanup timestamp and inserts new CleanupWorker jobs for all that need cleanup.

This works pretty well, but requires some contortions to get the river.Client into the CleanupSchedulingWorker, specifically adding the worker after the client has been initialized:

workers := river.NewWorkers()

riverClient, err = river.NewClient(riverpgxv5.New(dbPool), &river.Config{
	Queues: map[string]river.QueueConfig{
		river.QueueDefault: {MaxWorkers: 100},
	},
	PeriodicJobs: []*river.PeriodicJob{
		river.NewPeriodicJob(
			river.PeriodicInterval(CLEANUP_SCHEDULING_FREQUENCY_MINUTES*time.Minute),
			func() (river.JobArgs, *river.InsertOpts) {
				return CleanupSchedulingArgs{}, &river.InsertOpts{
					UniqueOpts: river.UniqueOpts{
						ByPeriod: CLEANUP_SCHEDULING_FREQUENCY_MINUTES * time.Minute,
					},
				}
			},
			&river.PeriodicJobOpts{RunOnStart: true},
		),
	},
	Workers: workers,
})
if err != nil {
	return nil, fmt.Errorf("error creating river queue client: %w", err)
}

river.AddWorker(workers, &CleanupSchedulingWorker{
	Queries: queries,
	River:   riverClient, // depends on the client already being created
})
river.AddWorker(workers, &CleanupWorker{
	Config:     config,
	Queries:    queries,
})
river.AddWorker(workers, OTHER STUFF)

Since the docs all initialize the workers before creating the client, I was wondering if I'm holding this wrong and if anyone has a better idea how to structure this?

[bgentry]

Hi @DavidS-ovm, as of #88 you can now create a Client with an empty or incomplete Workers bundle and continue to use AddWorker to register additional workers up until the client has been started with Start() (as you've shown in your example). This was a more recent addition and most of the docs were written prior to this.

We've run into this issue of workers that want access to the client a few times and while what you've shown above is the main way I know to do it today, I agree with you that it's not ideal. The ugliness is part of what keeps us from wanting to show more examples of this usage in the docs, though clearly it's something we should try to improve.

Another idea I came up with to try to improve this is summarized in #87 (comment): essentially we could stick the client into a context key in the context that gets passed to Work() and provide an API for accessing it. While this would be a more implicit API, it might feel a lot cleaner and less awkward than what you're doing above.

If you wanted to, I think you could experiment with this approach on your own entirely in userspace. The context you provide to Start(ctx) is used as the base context for all jobs that are worked by that Client, so prior to starting it you could use context.WithValue() to set the *Client to a known key, and later retrieve it with ctx.Value() within your worker. Ideally using the approach outlined in the context blog post to expose a clean API for this instead of a globally available map key. If you do give this a try, we'd be very interested to hear how it goes and think about whether or not we should make this an official feature 😄

[DavidS-ovm]

Hmm. If it were documented like I'm currently doing it, I wouldn't have spent a second thought on it.

I do have other bits and bobs in this code-base that use the context to pass stuff around (primarily auth tokens and oauth2.HttpClients) and that is OK-ish, but doesn't feel well-suited for "main line" things like the current river.Client. An example to highlight this: the oauth2.HttpClient context value is a *http.Client to allow oauth2 to make look-aside requests to refresh auth tokens while a http request (on a different http.Client) is made. I recently had a case where a piece of code worked in one situation, but not another because the broken case was missing a specific initialiser that injected the http client into the context as a side-effect.

[bgentry]

@DavidS-ovm I took a crack at this in #145. I agree that context values can be fraught generally, although in this case it feels pretty safe: as long as you're using the context provided to your Worker (or one derived from it) the Client will always be set on it.

Interested to hear your thoughts on it! I think it's quite common for workers to need to insert additional jobs, so it makes sense to try to streamline this.

[DavidS-ovm]

Yeah, I don't hate it :-D

Code components other than the top-level Work() methods might still run into ambiguity on whether or not a client is available in the context. On the other hand, non-Worker components (at least in my code-base) are already provided the client (and db connection, and API clients, etc) through other means anyways.

The last thing that came to my mind was "what if someone wants to use a different client than the one running the current job, but that seems niche and rather straightforward too: just pass it in through other means 🤷