alice.pds.demo.boltless.dev
This repository has no description
1package webhook
2
3import (
4 "bytes"
5 "context"
6 "crypto/hmac"
7 "crypto/sha256"
8 "encoding/hex"
9 "encoding/json"
10 "fmt"
11 "io"
12 "log/slog"
13 "net/http"
14 "time"
15
16 "github.com/avast/retry-go/v4"
17 "github.com/bluesky-social/indigo/atproto/syntax"
18 "github.com/google/uuid"
19 "tangled.org/core/appview/db"
20 "tangled.org/core/appview/models"
21 "tangled.org/core/appview/notify"
22 "tangled.org/core/log"
23 "tangled.org/core/orm"
24)
25
26type Notifier struct {
27 notify.BaseNotifier
28 db *db.DB
29 baseUrl string
30 logger *slog.Logger
31 client *http.Client
32}
33
34func NewNotifier(database *db.DB, baseUrl string) *Notifier {
35 return &Notifier{
36 db: database,
37 baseUrl: baseUrl,
38 logger: log.New("webhook-notifier"),
39 client: &http.Client{
40 Timeout: 30 * time.Second,
41 },
42 }
43}
44
45var _ notify.Notifier = &Notifier{}
46
47func (w *Notifier) Push(ctx context.Context, repo *models.Repo, ref, oldSha, newSha, committerDid string) {
48 webhooks, err := w.activeWebhooksForEvent(repo.RepoDid, models.WebhookEventPush)
49 if err != nil {
50 w.logger.Error("failed to get webhooks for repo", "repo_did", repo.RepoDid, "err", err)
51 return
52 }
53 if len(webhooks) == 0 {
54 return
55 }
56
57 payload := w.buildPushPayload(repo, ref, oldSha, newSha, committerDid)
58 payloadBytes, err := json.Marshal(payload)
59 if err != nil {
60 w.logger.Error("failed to marshal push payload", "repo_did", repo.RepoDid, "err", err)
61 return
62 }
63
64 userAgent := "Tangled-Hook/" + newSha[:7]
65 for _, webhook := range webhooks {
66 go w.sendWebhook(ctx, webhook, string(models.WebhookEventPush), payload.Repository.FullName, userAgent, payloadBytes)
67 }
68}
69
70func (w *Notifier) RenameRepo(ctx context.Context, actor syntax.DID, oldRepo, newRepo *models.Repo) {
71 webhooks, err := w.activeWebhooksForEvent(newRepo.RepoDid, models.WebhookEventRepoRenamed)
72 if err != nil {
73 w.logger.Error("failed to get webhooks for repo", "repo_did", newRepo.RepoDid, "err", err)
74 return
75 }
76 if len(webhooks) == 0 {
77 return
78 }
79
80 payload := &models.WebhookRenamePayload{
81 OldName: oldRepo.Name,
82 NewName: newRepo.Name,
83 Repository: buildWebhookRepository(newRepo),
84 Sender: models.WebhookUser{Did: actor.String()},
85 }
86 payloadBytes, err := json.Marshal(payload)
87 if err != nil {
88 w.logger.Error("failed to marshal rename payload", "repo_did", newRepo.RepoDid, "err", err)
89 return
90 }
91
92 userAgent := "Tangled-Hook/rename"
93 for _, webhook := range webhooks {
94 go w.sendWebhook(ctx, webhook, string(models.WebhookEventRepoRenamed), payload.Repository.FullName, userAgent, payloadBytes)
95 }
96}
97
98func (w *Notifier) NewPull(ctx context.Context, pull *models.Pull) {
99 w.pullRequestEvent(ctx, models.WebhookEventPullRequestCreated, "created", pull.OwnerDid, pull)
100}
101
102func (w *Notifier) ResubmitPull(ctx context.Context, pull *models.Pull) {
103 w.pullRequestEvent(ctx, models.WebhookEventPullRequestResubmitted, "resubmitted", pull.OwnerDid, pull)
104}
105
106func (w *Notifier) NewPullState(ctx context.Context, actor syntax.DID, pull *models.Pull) {
107 event, action, ok := pullStateEvent(pull.State)
108 if !ok {
109 return
110 }
111 w.pullRequestEvent(ctx, event, action, actor.String(), pull)
112}
113
114// pullStateEvent maps a pull's state to the webhook event announcing the
115// transition into that state
116func pullStateEvent(state models.PullState) (models.WebhookEvent, string, bool) {
117 switch state {
118 case models.PullMerged:
119 return models.WebhookEventPullRequestMerged, "merged", true
120 case models.PullClosed:
121 return models.WebhookEventPullRequestClosed, "closed", true
122 case models.PullOpen:
123 return models.WebhookEventPullRequestReopened, "reopened", true
124 default:
125 return "", "", false
126 }
127}
128
129func (w *Notifier) pullRequestEvent(ctx context.Context, event models.WebhookEvent, action, sender string, pull *models.Pull) {
130 // pull request events originate from http handlers, whose context is
131 // canceled as soon as the handler returns; detach so in-flight
132 // deliveries are not cut short
133 ctx = context.WithoutCancel(ctx)
134
135 webhooks, err := w.activeWebhooksForEvent(string(pull.RepoDid), event)
136 if err != nil {
137 w.logger.Error("failed to get webhooks for repo", "repo_did", pull.RepoDid, "err", err)
138 return
139 }
140 if len(webhooks) == 0 {
141 return
142 }
143
144 repo, err := db.GetRepo(w.db, orm.FilterEq("repo_did", string(pull.RepoDid)))
145 if err != nil {
146 w.logger.Error("failed to get repo", "repo_did", pull.RepoDid, "err", err)
147 return
148 }
149
150 payload := buildPullRequestPayload(action, repo, pull, sender, w.baseUrl)
151 payloadBytes, err := json.Marshal(payload)
152 if err != nil {
153 w.logger.Error("failed to marshal pull request payload", "repo_did", pull.RepoDid, "err", err)
154 return
155 }
156
157 userAgent := "Tangled-Hook/pull_request"
158 for _, webhook := range webhooks {
159 go w.sendWebhook(ctx, webhook, string(event), payload.Repository.FullName, userAgent, payloadBytes)
160 }
161}
162
163func buildPullRequestPayload(action string, repo *models.Repo, pull *models.Pull, sender, baseUrl string) *models.WebhookPullRequestPayload {
164 htmlUrl := fmt.Sprintf("%s/%s/%s/pulls/%d", baseUrl, repo.Did, repo.Slug(), pull.PullId)
165
166 pullRequest := models.WebhookPullRequest{
167 Number: pull.PullId,
168 Title: pull.Title,
169 Body: pull.Body,
170 State: pull.State.String(),
171 TargetBranch: pull.TargetBranch,
172 Owner: models.WebhookUser{Did: pull.OwnerDid},
173 HtmlUrl: htmlUrl,
174 CreatedAt: pull.Created.Format(time.RFC3339),
175 }
176 if len(pull.Submissions) > 0 {
177 pullRequest.RoundNumber = pull.LastRoundNumber()
178 pullRequest.PatchUrl = fmt.Sprintf("%s/round/%d.patch", htmlUrl, pull.LastRoundNumber())
179 }
180 if pull.PullSource != nil {
181 source := &models.WebhookPullRequestSource{
182 Branch: pull.PullSource.Branch,
183 }
184 if len(pull.Submissions) > 0 {
185 source.Sha = pull.LatestSha()
186 }
187 if pull.IsForkBased() {
188 source.Repo = pull.PullSource.RepoDid.String()
189 }
190 pullRequest.Source = source
191 }
192
193 return &models.WebhookPullRequestPayload{
194 Action: action,
195 PullRequest: pullRequest,
196 Repository: buildWebhookRepository(repo),
197 Sender: models.WebhookUser{Did: sender},
198 }
199}
200
201func (w *Notifier) activeWebhooksForEvent(repoDid string, event models.WebhookEvent) ([]models.Webhook, error) {
202 webhooks, err := db.GetActiveWebhooksForRepo(w.db, repoDid)
203 if err != nil {
204 return nil, err
205 }
206 var matching []models.Webhook
207 for _, webhook := range webhooks {
208 if webhook.HasEvent(event) {
209 matching = append(matching, webhook)
210 }
211 }
212 return matching, nil
213}
214
215func buildWebhookRepository(repo *models.Repo) models.WebhookRepository {
216 repository := models.WebhookRepository{
217 Name: repo.Name,
218 FullName: fmt.Sprintf("%s/%s", repo.Did, repo.Rkey),
219 Description: repo.Description,
220 Fork: repo.Source != "",
221 HtmlUrl: fmt.Sprintf("https://%s/%s/%s", repo.Knot, repo.Did, repo.Rkey),
222 CloneUrl: fmt.Sprintf("https://%s/%s/%s", repo.Knot, repo.Did, repo.Rkey),
223 SshUrl: fmt.Sprintf("ssh://git@%s/%s/%s", repo.Knot, repo.Did, repo.Rkey),
224 CreatedAt: repo.Created.Format(time.RFC3339),
225 UpdatedAt: repo.Created.Format(time.RFC3339),
226 Owner: models.WebhookUser{
227 Did: repo.Did,
228 },
229 }
230 if repo.Website != "" {
231 repository.Website = repo.Website
232 }
233 if repo.RepoStats != nil {
234 repository.StarsCount = repo.RepoStats.StarCount
235 repository.OpenIssues = repo.RepoStats.IssueCount.Open
236 }
237 return repository
238}
239
240func (w *Notifier) buildPushPayload(repo *models.Repo, ref, oldSha, newSha, committerDid string) *models.WebhookPayload {
241 pusher := committerDid
242 if committerDid == "" {
243 pusher = repo.Did
244 }
245 return &models.WebhookPayload{
246 Ref: ref,
247 Before: oldSha,
248 After: newSha,
249 Repository: buildWebhookRepository(repo),
250 Pusher: models.WebhookUser{
251 Did: pusher,
252 },
253 }
254}
255
256func (w *Notifier) sendWebhook(ctx context.Context, webhook models.Webhook, event, repoFullName, userAgent string, payloadBytes []byte) {
257 deliveryId := uuid.New().String()
258
259 req, err := http.NewRequestWithContext(ctx, "POST", webhook.Url, bytes.NewReader(payloadBytes))
260 if err != nil {
261 w.logger.Error("failed to create webhook request", "webhook_id", webhook.Id, "err", err)
262 return
263 }
264
265 req.Header.Set("Content-Type", "application/json")
266 req.Header.Set("User-Agent", userAgent)
267 req.Header.Set("X-Tangled-Event", event)
268 req.Header.Set("X-Tangled-Hook-ID", fmt.Sprintf("%d", webhook.Id))
269 req.Header.Set("X-Tangled-Delivery", deliveryId)
270 req.Header.Set("X-Tangled-Repo", repoFullName)
271
272 if webhook.Secret != "" {
273 signature := w.computeSignature(payloadBytes, webhook.Secret)
274 req.Header.Set("X-Tangled-Signature-256", "sha256="+signature)
275 }
276
277 delivery := &models.WebhookDelivery{
278 WebhookId: webhook.Id,
279 Event: event,
280 DeliveryId: deliveryId,
281 Url: webhook.Url,
282 RequestBody: string(payloadBytes),
283 }
284
285 retryOpts := []retry.Option{
286 retry.Attempts(3),
287 retry.Delay(1 * time.Second),
288 retry.MaxDelay(10 * time.Second),
289 retry.DelayType(retry.BackOffDelay),
290 retry.LastErrorOnly(true),
291 retry.OnRetry(func(n uint, err error) {
292 w.logger.Info("retrying webhook delivery",
293 "webhook_id", webhook.Id,
294 "attempt", n+1,
295 "err", err)
296 }),
297 retry.Context(ctx),
298 retry.RetryIf(func(err error) bool {
299 return err != nil
300 }),
301 }
302
303 var resp *http.Response
304 err = retry.Do(func() error {
305 var err error
306 resp, err = w.client.Do(req)
307 if err != nil {
308 return err
309 }
310 if resp.StatusCode >= 500 {
311 defer resp.Body.Close()
312 return fmt.Errorf("server error: %d", resp.StatusCode)
313 }
314 return nil
315 }, retryOpts...)
316
317 if err != nil {
318 w.logger.Error("webhook request failed after retries", "webhook_id", webhook.Id, "err", err)
319 delivery.Success = false
320 delivery.ResponseBody = err.Error()
321 } else {
322 defer resp.Body.Close()
323
324 delivery.ResponseCode = resp.StatusCode
325 delivery.Success = resp.StatusCode >= 200 && resp.StatusCode < 300
326
327 bodyBytes, err := io.ReadAll(io.LimitReader(resp.Body, 10*1024))
328 if err != nil {
329 w.logger.Warn("failed to read webhook response body", "webhook_id", webhook.Id, "err", err)
330 } else {
331 delivery.ResponseBody = string(bodyBytes)
332 }
333
334 if !delivery.Success {
335 w.logger.Warn("webhook delivery failed",
336 "webhook_id", webhook.Id,
337 "status", resp.StatusCode,
338 "url", webhook.Url)
339 } else {
340 w.logger.Info("webhook delivered successfully",
341 "webhook_id", webhook.Id,
342 "url", webhook.Url,
343 "delivery_id", deliveryId)
344 }
345 }
346
347 if err := db.AddWebhookDelivery(w.db, delivery); err != nil {
348 w.logger.Error("failed to record webhook delivery", "webhook_id", webhook.Id, "err", err)
349 }
350}
351
352func (w *Notifier) computeSignature(payload []byte, secret string) string {
353 mac := hmac.New(sha256.New, []byte(secret))
354 mac.Write(payload)
355 return hex.EncodeToString(mac.Sum(nil))
356}