Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
7.1 kB · 250 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251package webhook
import ( "bytes" "context" "crypto/hmac" "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "io" "log/slog" "net/http" "time"
"github.com/avast/retry-go/v4" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/google/uuid" "tangled.org/core/appview/db" "tangled.org/core/appview/models" "tangled.org/core/appview/notify" "tangled.org/core/log")
type Notifier struct { notify.BaseNotifier db *db.DB logger *slog.Logger client *http.Client}
func NewNotifier(database *db.DB) *Notifier { return &Notifier{ db: database, logger: log.New("webhook-notifier"), client: &http.Client{ Timeout: 30 * time.Second, }, }}
var _ notify.Notifier = &Notifier{}
func (w *Notifier) Push(ctx context.Context, repo *models.Repo, ref, oldSha, newSha, committerDid string) { webhooks, err := w.activeWebhooksForEvent(repo.RepoDid, models.WebhookEventPush) if err != nil { w.logger.Error("failed to get webhooks for repo", "repo_did", repo.RepoDid, "err", err) return } if len(webhooks) == 0 { return }
payload := w.buildPushPayload(repo, ref, oldSha, newSha, committerDid) payloadBytes, err := json.Marshal(payload) if err != nil { w.logger.Error("failed to marshal push payload", "repo_did", repo.RepoDid, "err", err) return }
userAgent := "Tangled-Hook/" + newSha[:7] for _, webhook := range webhooks { go w.sendWebhook(ctx, webhook, string(models.WebhookEventPush), payload.Repository.FullName, userAgent, payloadBytes) }}
func (w *Notifier) RenameRepo(ctx context.Context, actor syntax.DID, oldRepo, newRepo *models.Repo) { webhooks, err := w.activeWebhooksForEvent(newRepo.RepoDid, models.WebhookEventRepoRenamed) if err != nil { w.logger.Error("failed to get webhooks for repo", "repo_did", newRepo.RepoDid, "err", err) return } if len(webhooks) == 0 { return }
payload := &models.WebhookRenamePayload{ OldName: oldRepo.Name, NewName: newRepo.Name, Repository: buildWebhookRepository(newRepo), Sender: models.WebhookUser{Did: actor.String()}, } payloadBytes, err := json.Marshal(payload) if err != nil { w.logger.Error("failed to marshal rename payload", "repo_did", newRepo.RepoDid, "err", err) return }
userAgent := "Tangled-Hook/rename" for _, webhook := range webhooks { go w.sendWebhook(ctx, webhook, string(models.WebhookEventRepoRenamed), payload.Repository.FullName, userAgent, payloadBytes) }}
func (w *Notifier) activeWebhooksForEvent(repoDid string, event models.WebhookEvent) ([]models.Webhook, error) { webhooks, err := db.GetActiveWebhooksForRepo(w.db, repoDid) if err != nil { return nil, err } var matching []models.Webhook for _, webhook := range webhooks { if webhook.HasEvent(event) { matching = append(matching, webhook) } } return matching, nil}
func buildWebhookRepository(repo *models.Repo) models.WebhookRepository { repository := models.WebhookRepository{ Name: repo.Name, FullName: fmt.Sprintf("%s/%s", repo.Did, repo.Rkey), Description: repo.Description, Fork: repo.Source != "", HtmlUrl: fmt.Sprintf("https://%s/%s/%s", repo.Knot, repo.Did, repo.Rkey), CloneUrl: fmt.Sprintf("https://%s/%s/%s", repo.Knot, repo.Did, repo.Rkey), SshUrl: fmt.Sprintf("ssh://git@%s/%s/%s", repo.Knot, repo.Did, repo.Rkey), CreatedAt: repo.Created.Format(time.RFC3339), UpdatedAt: repo.Created.Format(time.RFC3339), Owner: models.WebhookUser{ Did: repo.Did, }, } if repo.Website != "" { repository.Website = repo.Website } if repo.RepoStats != nil { repository.StarsCount = repo.RepoStats.StarCount repository.OpenIssues = repo.RepoStats.IssueCount.Open } return repository}
func (w *Notifier) buildPushPayload(repo *models.Repo, ref, oldSha, newSha, committerDid string) *models.WebhookPayload { pusher := committerDid if committerDid == "" { pusher = repo.Did } return &models.WebhookPayload{ Ref: ref, Before: oldSha, After: newSha, Repository: buildWebhookRepository(repo), Pusher: models.WebhookUser{ Did: pusher, }, }}
func (w *Notifier) sendWebhook(ctx context.Context, webhook models.Webhook, event, repoFullName, userAgent string, payloadBytes []byte) { deliveryId := uuid.New().String()
req, err := http.NewRequestWithContext(ctx, "POST", webhook.Url, bytes.NewReader(payloadBytes)) if err != nil { w.logger.Error("failed to create webhook request", "webhook_id", webhook.Id, "err", err) return }
req.Header.Set("Content-Type", "application/json") req.Header.Set("User-Agent", userAgent) req.Header.Set("X-Tangled-Event", event) req.Header.Set("X-Tangled-Hook-ID", fmt.Sprintf("%d", webhook.Id)) req.Header.Set("X-Tangled-Delivery", deliveryId) req.Header.Set("X-Tangled-Repo", repoFullName)
if webhook.Secret != "" { signature := w.computeSignature(payloadBytes, webhook.Secret) req.Header.Set("X-Tangled-Signature-256", "sha256="+signature) }
delivery := &models.WebhookDelivery{ WebhookId: webhook.Id, Event: event, DeliveryId: deliveryId, Url: webhook.Url, RequestBody: string(payloadBytes), }
retryOpts := []retry.Option{ retry.Attempts(3), retry.Delay(1 * time.Second), retry.MaxDelay(10 * time.Second), retry.DelayType(retry.BackOffDelay), retry.LastErrorOnly(true), retry.OnRetry(func(n uint, err error) { w.logger.Info("retrying webhook delivery", "webhook_id", webhook.Id, "attempt", n+1, "err", err) }), retry.Context(ctx), retry.RetryIf(func(err error) bool { return err != nil }), }
var resp *http.Response err = retry.Do(func() error { var err error resp, err = w.client.Do(req) if err != nil { return err } if resp.StatusCode >= 500 { defer resp.Body.Close() return fmt.Errorf("server error: %d", resp.StatusCode) } return nil }, retryOpts...)
if err != nil { w.logger.Error("webhook request failed after retries", "webhook_id", webhook.Id, "err", err) delivery.Success = false delivery.ResponseBody = err.Error() } else { defer resp.Body.Close()
delivery.ResponseCode = resp.StatusCode delivery.Success = resp.StatusCode >= 200 && resp.StatusCode < 300
bodyBytes, err := io.ReadAll(io.LimitReader(resp.Body, 10*1024)) if err != nil { w.logger.Warn("failed to read webhook response body", "webhook_id", webhook.Id, "err", err) } else { delivery.ResponseBody = string(bodyBytes) }
if !delivery.Success { w.logger.Warn("webhook delivery failed", "webhook_id", webhook.Id, "status", resp.StatusCode, "url", webhook.Url) } else { w.logger.Info("webhook delivered successfully", "webhook_id", webhook.Id, "url", webhook.Url, "delivery_id", deliveryId) } }
if err := db.AddWebhookDelivery(w.db, delivery); err != nil { w.logger.Error("failed to record webhook delivery", "webhook_id", webhook.Id, "err", err) }}
func (w *Notifier) computeSignature(payload []byte, secret string) string { mac := hmac.New(sha256.New, []byte(secret)) mac.Write(payload) return hex.EncodeToString(mac.Sum(nil))}