| 1 |
|
package main |
| 2 |
|
|
| 3 |
|
import ( |
| 4 |
|
"context" |
| 5 |
|
"log/slog" |
| 6 |
|
"time" |
| 7 |
|
|
| 8 |
|
work "git.sr.ht/~sircmpwn/dowork" |
| 9 |
|
sq "github.com/Masterminds/squirrel" |
| 10 |
|
"github.com/google/uuid" |
| 11 |
|
|
| 12 |
|
"sourcecraft.dev/bigbes/sr-ht-core/auth" |
| 13 |
|
"sourcecraft.dev/bigbes/sr-ht-core/webhooks" |
| 14 |
|
|
| 15 |
|
"sourcecraft.dev/bigbes/sr-ht-spec/graph" |
| 16 |
|
"sourcecraft.dev/bigbes/sr-ht-spec/graph/model" |
| 17 |
|
"sourcecraft.dev/bigbes/sr-ht-spec/service" |
| 18 |
|
) |
| 19 |
|
|
| 20 |
|
// webhookEventSink implements service.EventSink by enqueuing a dowork task onto |
| 21 |
|
// the webhook queue. The task runs in the queue's worker context (server + |
| 22 |
|
// database + config, from WithQueues), adds the owner's INTERNAL auth context, |
| 23 |
|
// and calls Schedule — which needs all three. Firing is fire-and-forget: a |
| 24 |
|
// webhook must never block or fail a proposal write. |
| 25 |
|
type webhookEventSink struct { |
| 26 |
|
queue *webhooks.WebhookQueue |
| 27 |
|
ownerUserID int |
| 28 |
|
ownerName string |
| 29 |
|
log *slog.Logger |
| 30 |
|
} |
| 31 |
|
|
| 32 |
0 |
func newWebhookEventSink(q *webhooks.WebhookQueue, ownerUserID int, ownerName string, log *slog.Logger) *webhookEventSink { |
| 33 |
0 |
return &webhookEventSink{queue: q, ownerUserID: ownerUserID, ownerName: ownerName, log: log} |
| 34 |
0 |
} |
| 35 |
|
|
| 36 |
|
// Compile-time assertion that the sink satisfies the service seam. |
| 37 |
|
var _ service.EventSink = (*webhookEventSink)(nil) |
| 38 |
|
|
| 39 |
0 |
func (s *webhookEventSink) ProposalEvent(kind service.ProposalEventKind, p service.Proposal) { |
| 40 |
0 |
u := uuid.New() |
| 41 |
0 |
payload, err := graph.NewProposalEvent(model.WebhookEvent(kind), u.String(), time.Now().UTC(), p) |
| 42 |
0 |
if err != nil { |
| 43 |
0 |
s.log.Error("build webhook payload", "err", err) |
| 44 |
0 |
return |
| 45 |
0 |
} |
| 46 |
0 |
event := string(kind) |
| 47 |
0 |
task := work.NewTask(func(ctx context.Context) error { |
| 48 |
0 |
// The worker context carries server+database+config; add the owner auth |
| 49 |
0 |
// Schedule captures (fetchSubscriptions and the delivery both need it). |
| 50 |
0 |
ctx = auth.Context(ctx, &auth.AuthContext{ |
| 51 |
0 |
AuthMethod: auth.AUTH_INTERNAL, UserID: s.ownerUserID, Username: s.ownerName, |
| 52 |
0 |
}) |
| 53 |
0 |
q := sq.Select().From("gql_user_wh_sub sub").Where("sub.user_id = ?", s.ownerUserID) |
| 54 |
0 |
s.queue.Schedule(ctx, q, "user", event, u, payload) |
| 55 |
0 |
return nil |
| 56 |
0 |
}) |
| 57 |
|
// Enqueue off the write path so a full queue never blocks a merge. |
| 58 |
0 |
go func() { _ = s.queue.Queue.Enqueue(task) }() |
| 59 |
|
} |