diff --git a/webhooks/queue.go b/webhooks/queue.go index 2a4042871c5f3e988ac5404e0e962247cad00451..ac8107f85e10f0f50b74d82bd2839473da17688e 100644 --- a/webhooks/queue.go +++ b/webhooks/queue.go @@ -12,7 +12,7 @@ "net/http" "strings" "time" - "git.sr.ht/~sircmpwn/dowork" + work "git.sr.ht/~sircmpwn/dowork" "github.com/99designs/gqlgen/graphql" sq "github.com/Masterminds/squirrel" "github.com/google/uuid" @@ -63,56 +63,61 @@ // context which contains the necessary state for your application to process // the webhook resolvers. func (queue *WebhookQueue) Schedule(ctx context.Context, q sq.SelectBuilder, name, event string, payloadUUID uuid.UUID, payload interface{}) { - user := auth.ForContext(ctx) + err := queue.schedule(ctx, q, name, event, payloadUUID, payload) + if err != nil { + log.Printf("Failed to enqueue webhook deliveries: %v", err) + } +} + +func (queue *WebhookQueue) schedule(ctx context.Context, q sq.SelectBuilder, + name, event string, payloadUUID uuid.UUID, payload interface{}) error { // The following tasks are done during this process: // // 1. Fetch subscription details from the database // 2. Prepare deliveries and create delivery records // 3. Deliver the webhooks // - // The first two steps are done in this task, then N tasks are created for + // The first two steps are done synchronously, then N tasks are created for // step 3 where N = number of subscriptions. - task := work.NewTask(func(ctx context.Context) error { - ctx = Context(ctx, payload) - subs, err := queue.fetchSubscriptions(ctx, q, event) - if err != nil { - return err - } - if len(subs) == 0 { - return nil - } + user := auth.ForContext(ctx) + ctx = Context(ctx, payload) + subs, err := queue.fetchSubscriptions(ctx, q, event) + if err != nil { + return err + } + if len(subs) == 0 { + return nil + } - tasks := make([]*work.Task, len(subs)) - if err := database.WithTx(ctx, nil, func(tx *sql.Tx) error { - var err error - for i, sub := range subs { - webhook := WebhookContext{ - Name: name, - Event: event, - User: user, - Payload: payload, - PayloadUUID: payloadUUID, - Subscription: sub, - } - tasks[i], err = queue.queueStage2(ctx, tx, &webhook) - if err != nil { - return err - } + tasks := make([]*work.Task, len(subs)) + if err := database.WithTx(ctx, nil, func(tx *sql.Tx) error { + var err error + for i, sub := range subs { + webhook := WebhookContext{ + Name: name, + Event: event, + User: user, + Payload: payload, + PayloadUUID: payloadUUID, + Subscription: sub, + } + tasks[i], err = queue.queueStage2(ctx, tx, &webhook) + if err != nil { + return err } - return nil - }); err != nil { - log.Printf("Failed to enqueue %s/%s webhooks: %v", name, event, err) - return err } + return nil + }); err != nil { + log.Printf("Failed to enqueue %s/%s webhooks: %v", name, event, err) + return err + } - for _, task := range tasks { - queue.Queue.Enqueue(task) - } - log.Printf("Enqueued %s/%s webhook delivery for %d subscriptions", - name, event, len(subs)) - return nil - }) - queue.Queue.Enqueue(task) + for _, task := range tasks { + queue.Queue.Enqueue(task) + } + log.Printf("Enqueued %s/%s webhook delivery for %d subscriptions", + name, event, len(subs)) + return nil } func (queue *WebhookQueue) fetchSubscriptions(ctx context.Context,