Add notify package and durable notification outbox worker (Feature 19)
Emit(ctx, Event) is the single decision point for who is told what: it reads the event type's notification_prefs row and stores one pending notification_outbox row per enabled channel, keyed by (event_key, channel) with ON CONFLICT DO NOTHING, so a duplicate key never adds or resets a row. The worker, started and stopped by cmd/server with the job runner, leases due rows with a conditional update, delivers them through the Sender registered for their channel, and marks them sent, or retries with a backoff of 1 minute doubling to 1 hour until 24 hours after creation, then marks them failed with last_error. A channel with no registered sender fails its rows with an explicit error. Maestro task addeb256-a4d4-4acf-8acd-6b6778d967ee. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
+15
-5
@@ -24,6 +24,7 @@ import (
|
||||
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/gcal"
|
||||
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/household"
|
||||
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
|
||||
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/notify"
|
||||
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/plan"
|
||||
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/push"
|
||||
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/recipes"
|
||||
@@ -144,6 +145,11 @@ func run(cfg config.Config, logger *slog.Logger) error {
|
||||
return err
|
||||
}
|
||||
|
||||
notifier, err := notify.NewService(db, logger)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
sessions, err := auth.NewSessionManager(db, cfg.SessionIdle, cfg.SessionMax)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -166,14 +172,18 @@ func run(cfg config.Config, logger *slog.Logger) error {
|
||||
}
|
||||
|
||||
// Start fails the jobs a previous process left running before any
|
||||
// worker runs. The workers stop, and are waited for, before the
|
||||
// database closes.
|
||||
jobsCtx, stopJobs := context.WithCancel(ctx)
|
||||
// worker runs. The job workers and the notification worker stop, and
|
||||
// are waited for, before the database closes.
|
||||
workersCtx, stopWorkers := context.WithCancel(ctx)
|
||||
defer func() {
|
||||
stopJobs()
|
||||
stopWorkers()
|
||||
runner.Wait()
|
||||
notifier.Wait()
|
||||
}()
|
||||
if err := runner.Start(jobsCtx); err != nil {
|
||||
if err := runner.Start(workersCtx); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := notifier.Start(workersCtx); err != nil {
|
||||
return err
|
||||
}
|
||||
// With those jobs failed, a dinner still pending with no grounding job
|
||||
|
||||
Reference in New Issue
Block a user