From 3fb5d416cc9d5df20c489849655f9295b526594a Mon Sep 17 00:00:00 2001 From: Anirudh Oppiliappan Date: Thu, 6 Aug 2026 15:56:01 +0300 Subject: [PATCH] spindle/db,models: add webhook types, tables and queries Webhook + delivery domain/payload types, the webhooks and webhook_deliveries tables, and their queries. Re-add a nullable repos.name column (migration) so rename payloads can carry the display name. Signed-off-by: dawn --- spindle/db/db.go | 56 +++++ spindle/db/pull_rounds.go | 52 +++++ spindle/db/pull_rounds_test.go | 47 +++++ spindle/db/repos.go | 23 ++- spindle/db/webhooks.go | 332 ++++++++++++++++++++++++++++++ spindle/db/webhooks_smoke_test.go | 94 +++++++++ spindle/models/webhook.go | 153 ++++++++++++++ spindle/wipe.go | 2 + spindle/wipe_test.go | 11 +- 9 files changed, 761 insertions(+), 9 deletions(-) create mode 100644 spindle/db/pull_rounds.go create mode 100644 spindle/db/pull_rounds_test.go create mode 100644 spindle/db/webhooks.go create mode 100644 spindle/db/webhooks_smoke_test.go create mode 100644 spindle/models/webhook.go diff --git a/spindle/db/db.go b/spindle/db/db.go index 3a3a5faf..72ee4a96 100644 --- a/spindle/db/db.go +++ b/spindle/db/db.go @@ -238,6 +238,41 @@ func Make(ctx context.Context, dbPath string) (*DB, error) { mill_records_terminal_metrics integer not null default 0 ); + create table if not exists webhooks ( + id integer primary key autoincrement, + repo_did text not null, + url text not null, + secret text, + active integer not null default 1, + events text not null, -- comma-separated event types + created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), + updated_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) + ); + create index if not exists idx_webhooks_repo_did on webhooks(repo_did); + + create table if not exists webhook_deliveries ( + id integer primary key autoincrement, + webhook_id integer not null references webhooks(id) on delete cascade, + event text not null, + delivery_id text not null, + url text not null, + request_body text, + response_code integer, + response_body text, + success integer not null default 0, + created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) + ); + create index if not exists idx_webhook_deliveries_webhook_id on webhook_deliveries(webhook_id); + create unique index if not exists idx_webhook_deliveries_delivery_id on webhook_deliveries(delivery_id); + + create table if not exists pull_rounds ( + repo_did text not null, + rkey text not null, + rounds integer not null, + updated_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), + primary key (repo_did, rkey) + ); + create table if not exists migrations ( id integer primary key autoincrement, name text unique @@ -845,6 +880,27 @@ func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error return err } + // re-introduce a repo display name, needed to build repository:renamed + // webhook payloads. the earlier repos-to-repo-did migration dropped the + // legacy name column; this adds it back as nullable metadata. + if err := orm.RunMigration(conn, logger, "repos-add-name-column", func(tx *sql.Tx) error { + var hasName int + if err := tx.QueryRow( + `select count(*) from pragma_table_info('repos') where name = 'name'`, + ).Scan(&hasName); err != nil { + return err + } + if hasName == 0 { + if _, err := tx.Exec(`alter table repos add column name text`); err != nil { + return err + } + } + _, err := tx.Exec(`update repos set name = rkey where name is null`) + return err + }); err != nil { + return err + } + return nil } diff --git a/spindle/db/pull_rounds.go b/spindle/db/pull_rounds.go new file mode 100644 index 00000000..654694f5 --- /dev/null +++ b/spindle/db/pull_rounds.go @@ -0,0 +1,52 @@ +package db + +import ( + "database/sql" + "errors" + "fmt" + + "github.com/bluesky-social/indigo/atproto/syntax" +) + +func (d *DB) ObservePullRounds(repoDid syntax.DID, rkey string, rounds int) (bool, error) { + var stored int + err := d.QueryRow( + `select rounds from pull_rounds where repo_did = ? and rkey = ?`, + repoDid.String(), rkey, + ).Scan(&stored) + switch { + case errors.Is(err, sql.ErrNoRows): + if _, err := d.Exec( + `insert into pull_rounds (repo_did, rkey, rounds) values (?, ?, ?)`, + repoDid.String(), rkey, rounds, + ); err != nil { + return false, fmt.Errorf("failed to record pull rounds: %w", err) + } + return false, nil + case err != nil: + return false, fmt.Errorf("failed to read pull rounds: %w", err) + } + + if rounds == stored { + return false, nil + } + if _, err := d.Exec( + `update pull_rounds + set rounds = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') + where repo_did = ? and rkey = ?`, + rounds, repoDid.String(), rkey, + ); err != nil { + return false, fmt.Errorf("failed to update pull rounds: %w", err) + } + return rounds > stored, nil +} + +func (d *DB) ForgetPullRounds(repoDid syntax.DID, rkey string) error { + _, err := d.Exec(`delete from pull_rounds where repo_did = ? and rkey = ?`, repoDid.String(), rkey) + return err +} + +func (d *DB) DeletePullRoundsByRepo(repoDid syntax.DID) error { + _, err := d.Exec(`delete from pull_rounds where repo_did = ?`, repoDid.String()) + return err +} diff --git a/spindle/db/pull_rounds_test.go b/spindle/db/pull_rounds_test.go new file mode 100644 index 00000000..1d8fd563 --- /dev/null +++ b/spindle/db/pull_rounds_test.go @@ -0,0 +1,47 @@ +package db + +import ( + "context" + "path/filepath" + "testing" + + "github.com/bluesky-social/indigo/atproto/syntax" +) + +func TestObservePullRounds(t *testing.T) { + d, err := Make(context.Background(), filepath.Join(t.TempDir(), "test.db")) + if err != nil { + t.Fatalf("make db: %v", err) + } + + repoDid := syntax.DID("did:plc:repo123") + const rkey = "3kabc" + + appended, err := d.ObservePullRounds(repoDid, rkey, 1) + if err != nil || appended { + t.Fatalf("first sighting: appended=%v err=%v", appended, err) + } + + if appended, err := d.ObservePullRounds(repoDid, rkey, 1); err != nil || appended { + t.Fatalf("edit: appended=%v err=%v", appended, err) + } + + if appended, err := d.ObservePullRounds(repoDid, rkey, 2); err != nil || !appended { + t.Fatalf("resubmission: appended=%v err=%v", appended, err) + } + + if appended, err := d.ObservePullRounds(repoDid, "3kdef", 3); err != nil || appended { + t.Fatalf("other pull: appended=%v err=%v", appended, err) + } + + if err := d.ForgetPullRounds(repoDid, rkey); err != nil { + t.Fatalf("forget: %v", err) + } + var n int + if err := d.QueryRow(`select count(*) from pull_rounds where repo_did = ?`, repoDid.String()).Scan(&n); err != nil { + t.Fatalf("count: %v", err) + } + if n != 1 { + t.Fatalf("expected one remaining row, got %d", n) + } +} diff --git a/spindle/db/repos.go b/spindle/db/repos.go index 23d6864f..79179a8f 100644 --- a/spindle/db/repos.go +++ b/spindle/db/repos.go @@ -11,6 +11,7 @@ type Repo struct { Owner syntax.DID Rkey syntax.RecordKey RepoDid syntax.DID + Name string CreatedAt string } @@ -19,14 +20,19 @@ func (d *DB) AddRepo(repo Repo) error { if repo.CreatedAt != "" { createdAt = sql.NullString{String: repo.CreatedAt, Valid: true} } + var name sql.NullString + if repo.Name != "" { + name = sql.NullString{String: repo.Name, Valid: true} + } _, err := d.Exec( - `insert into repos (knot, owner, rkey, repo_did, created_at) - values (?, ?, ?, ?, ?) + `insert into repos (knot, owner, rkey, repo_did, name, created_at) + values (?, ?, ?, ?, ?, ?) on conflict(owner, rkey) do update set knot = excluded.knot, repo_did = excluded.repo_did, + name = coalesce(excluded.name, repos.name), created_at = coalesce(excluded.created_at, repos.created_at)`, - repo.Knot, repo.Owner.String(), repo.Rkey.String(), repo.RepoDid.String(), createdAt, + repo.Knot, repo.Owner.String(), repo.Rkey.String(), repo.RepoDid.String(), name, createdAt, ) return err } @@ -81,8 +87,8 @@ func (d *DB) Knots() ([]string, error) { } func scanRepo(row interface{ Scan(...any) error }) (*Repo, error) { - var knot, owner, rkey, repoDid string - if err := row.Scan(&knot, &owner, &rkey, &repoDid); err != nil { + var knot, owner, rkey, repoDid, name string + if err := row.Scan(&knot, &owner, &rkey, &repoDid, &name); err != nil { return nil, err } return &Repo{ @@ -90,6 +96,7 @@ func scanRepo(row interface{ Scan(...any) error }) (*Repo, error) { Owner: syntax.DID(owner), Rkey: syntax.RecordKey(rkey), RepoDid: syntax.DID(repoDid), + Name: name, }, nil } @@ -122,20 +129,20 @@ func (d *DB) SiblingRkeysForRepoDid(owner, repoDid syntax.DID, excludeRkey synta func (d *DB) GetRepoByDid(repoDid syntax.DID) (*Repo, error) { return scanRepo(d.QueryRow( - `select knot, owner, rkey, coalesce(repo_did, '') from repos where repo_did = ?`, + `select knot, owner, rkey, coalesce(repo_did, ''), coalesce(name, '') from repos where repo_did = ?`, repoDid.String(), )) } func (d *DB) GetRepoByOwnerRkey(owner syntax.DID, rkey syntax.RecordKey) (*Repo, error) { return scanRepo(d.QueryRow( - `select knot, owner, rkey, coalesce(repo_did, '') from repos where owner = ? and rkey = ?`, + `select knot, owner, rkey, coalesce(repo_did, ''), coalesce(name, '') from repos where owner = ? and rkey = ?`, owner.String(), rkey.String(), )) } func (d *DB) AllRepos() ([]Repo, error) { - rows, err := d.Query(`select knot, owner, rkey, coalesce(repo_did, '') from repos`) + rows, err := d.Query(`select knot, owner, rkey, coalesce(repo_did, ''), coalesce(name, '') from repos`) if err != nil { return nil, err } diff --git a/spindle/db/webhooks.go b/spindle/db/webhooks.go new file mode 100644 index 00000000..a59bcf42 --- /dev/null +++ b/spindle/db/webhooks.go @@ -0,0 +1,332 @@ +package db + +import ( + "context" + "database/sql" + "fmt" + "strings" + "time" + + "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/spindle/models" +) + +const webhookColumns = `id, repo_did, url, secret, active, events, created_at, updated_at` + +func scanWebhook(row interface{ Scan(...any) error }) (*models.Webhook, error) { + var wh models.Webhook + var repoDid, createdAt, updatedAt, eventsStr string + var secret sql.NullString + var active int + + if err := row.Scan( + &wh.Id, + &repoDid, + &wh.Url, + &secret, + &active, + &eventsStr, + &createdAt, + &updatedAt, + ); err != nil { + return nil, err + } + + wh.RepoDid = syntax.DID(repoDid) + if secret.Valid { + wh.Secret = secret.String + } + wh.Active = active == 1 + if eventsStr != "" { + wh.Events = strings.Split(eventsStr, ",") + } + if t, err := time.Parse(time.RFC3339, createdAt); err == nil { + wh.CreatedAt = t + } + if t, err := time.Parse(time.RFC3339, updatedAt); err == nil { + wh.UpdatedAt = t + } + return &wh, nil +} + +// GetWebhooksForRepo returns all webhooks configured for a repository, newest first. +func (d *DB) GetWebhooksForRepo(repoDid syntax.DID) ([]models.Webhook, error) { + rows, err := d.Query( + `select `+webhookColumns+` from webhooks where repo_did = ? order by created_at desc`, + repoDid.String(), + ) + if err != nil { + return nil, fmt.Errorf("failed to query webhooks: %w", err) + } + defer rows.Close() + + var webhooks []models.Webhook + for rows.Next() { + wh, err := scanWebhook(rows) + if err != nil { + return nil, fmt.Errorf("failed to scan webhook: %w", err) + } + webhooks = append(webhooks, *wh) + } + return webhooks, rows.Err() +} + +// GetActiveWebhooksForRepo returns only active webhooks for a repository. +func (d *DB) GetActiveWebhooksForRepo(repoDid syntax.DID) ([]models.Webhook, error) { + rows, err := d.Query( + `select `+webhookColumns+` from webhooks where repo_did = ? and active = 1 order by created_at desc`, + repoDid.String(), + ) + if err != nil { + return nil, fmt.Errorf("failed to query active webhooks: %w", err) + } + defer rows.Close() + + var webhooks []models.Webhook + for rows.Next() { + wh, err := scanWebhook(rows) + if err != nil { + return nil, fmt.Errorf("failed to scan webhook: %w", err) + } + webhooks = append(webhooks, *wh) + } + return webhooks, rows.Err() +} + +// GetWebhook returns a single webhook by ID. +func (d *DB) GetWebhook(id int64) (*models.Webhook, error) { + return scanWebhook(d.QueryRow( + `select `+webhookColumns+` from webhooks where id = ?`, id, + )) +} + +// AddWebhook creates a new webhook, setting webhook.Id on success. +func (d *DB) AddWebhook(ctx context.Context, webhook *models.Webhook, repoLimit, ownerLimit int64) (bool, error) { + tx, err := d.BeginTx(ctx, nil) + if err != nil { + return false, err + } + defer tx.Rollback() + + var owner string + if err := tx.QueryRowContext(ctx, + `select owner from repos where repo_did = ?`, webhook.RepoDid.String(), + ).Scan(&owner); err != nil { + return false, fmt.Errorf("failed to look up repo owner: %w", err) + } + + atLimit := func(query string, limit int64, args ...any) (bool, error) { + if limit <= 0 { + return false, nil + } + var n int64 + if err := tx.QueryRowContext(ctx, query, args...).Scan(&n); err != nil { + return false, fmt.Errorf("failed to count webhooks: %w", err) + } + return n >= limit, nil + } + + repoAtLimit, err := atLimit( + `select count(*) from webhooks where repo_did = ?`, repoLimit, webhook.RepoDid.String(), + ) + if err != nil || repoAtLimit { + return false, err + } + ownerAtLimit, err := atLimit( + `select count(*) from webhooks w join repos r on r.repo_did = w.repo_did where r.owner = ?`, + ownerLimit, owner, + ) + if err != nil || ownerAtLimit { + return false, err + } + + active := 0 + if webhook.Active { + active = 1 + } + result, err := tx.ExecContext(ctx, + `insert into webhooks (repo_did, url, secret, active, events) + values (?, ?, ?, ?, ?)`, + webhook.RepoDid.String(), webhook.Url, webhook.Secret, active, strings.Join(webhook.Events, ","), + ) + if err != nil { + return false, fmt.Errorf("failed to insert webhook: %w", err) + } + id, err := result.LastInsertId() + if err != nil { + return false, fmt.Errorf("failed to get webhook id: %w", err) + } + webhook.Id = id + return true, tx.Commit() +} + +// UpdateWebhook updates an existing webhook's mutable fields. +func (d *DB) UpdateWebhook(webhook *models.Webhook) error { + eventsStr := strings.Join(webhook.Events, ",") + active := 0 + if webhook.Active { + active = 1 + } + + _, err := d.Exec( + `update webhooks + set url = ?, secret = ?, active = ?, events = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') + where id = ?`, + webhook.Url, webhook.Secret, active, eventsStr, webhook.Id, + ) + if err != nil { + return fmt.Errorf("failed to update webhook: %w", err) + } + return nil +} + +// DeleteWebhook deletes a webhook and (via cascade) its deliveries. +func (d *DB) DeleteWebhook(id int64) error { + if _, err := d.Exec(`delete from webhooks where id = ?`, id); err != nil { + return fmt.Errorf("failed to delete webhook: %w", err) + } + return nil +} + +// AddWebhookDelivery records a webhook delivery attempt, setting delivery.Id. +func (d *DB) AddWebhookDelivery(delivery *models.WebhookDelivery) error { + success := 0 + if delivery.Success { + success = 1 + } + + result, err := d.Exec( + `insert into webhook_deliveries ( + webhook_id, event, delivery_id, url, request_body, response_code, response_body, success + ) values (?, ?, ?, ?, ?, ?, ?, ?)`, + delivery.WebhookId, + delivery.Event, + delivery.DeliveryId, + delivery.Url, + delivery.RequestBody, + delivery.ResponseCode, + delivery.ResponseBody, + success, + ) + if err != nil { + return fmt.Errorf("failed to insert webhook delivery: %w", err) + } + id, err := result.LastInsertId() + if err != nil { + return fmt.Errorf("failed to get delivery id: %w", err) + } + delivery.Id = id + return nil +} + +const deliveryColumns = `id, webhook_id, event, delivery_id, url, request_body, response_code, response_body, success, created_at` + +func scanDelivery(row interface{ Scan(...any) error }) (*models.WebhookDelivery, error) { + var dv models.WebhookDelivery + var createdAt string + var success int + var responseCode sql.NullInt64 + var responseBody sql.NullString + + if err := row.Scan( + &dv.Id, + &dv.WebhookId, + &dv.Event, + &dv.DeliveryId, + &dv.Url, + &dv.RequestBody, + &responseCode, + &responseBody, + &success, + &createdAt, + ); err != nil { + return nil, err + } + + dv.Success = success == 1 + if responseCode.Valid { + dv.ResponseCode = int(responseCode.Int64) + } + if responseBody.Valid { + dv.ResponseBody = responseBody.String + } + if t, err := time.Parse(time.RFC3339, createdAt); err == nil { + dv.CreatedAt = t + } + return &dv, nil +} + +func (d *DB) GetWebhookDeliveriesForWebhooks(repoDid syntax.DID, webhookIds []int64, perWebhook int) ([]models.WebhookDelivery, error) { + if len(webhookIds) == 0 { + return nil, nil + } + if perWebhook <= 0 { + perWebhook = 20 + } + placeholders := strings.TrimSuffix(strings.Repeat("?,", len(webhookIds)), ",") + args := make([]any, 0, len(webhookIds)+2) + args = append(args, repoDid.String()) + for _, id := range webhookIds { + args = append(args, id) + } + args = append(args, perWebhook) + + rows, err := d.Query( + `select `+deliveryColumns+` from ( + select d.id, d.webhook_id, d.event, d.delivery_id, d.url, d.request_body, + d.response_code, d.response_body, d.success, d.created_at, + row_number() over (partition by d.webhook_id order by d.id desc) as rn + from webhook_deliveries d + join webhooks w on w.id = d.webhook_id + where w.repo_did = ? and w.id in (`+placeholders+`) + ) where rn <= ? + order by webhook_id, id desc`, + args..., + ) + if err != nil { + return nil, fmt.Errorf("failed to query webhook deliveries: %w", err) + } + defer rows.Close() + + var deliveries []models.WebhookDelivery + for rows.Next() { + dv, err := scanDelivery(rows) + if err != nil { + return nil, fmt.Errorf("failed to scan webhook delivery: %w", err) + } + deliveries = append(deliveries, *dv) + } + return deliveries, rows.Err() +} + +// GetWebhookDelivery returns a single delivery by its delivery_id (a uuid). +func (d *DB) GetWebhookDelivery(deliveryId string) (*models.WebhookDelivery, error) { + return scanDelivery(d.QueryRow( + `select `+deliveryColumns+` from webhook_deliveries where delivery_id = ?`, deliveryId, + )) +} + +func (d *DB) DeleteWebhooksByRepo(repoDid syntax.DID) error { + if _, err := d.Exec(`delete from webhooks where repo_did = ?`, repoDid.String()); err != nil { + return fmt.Errorf("failed to delete webhooks: %w", err) + } + return nil +} + +func (d *DB) PruneWebhookDeliveries(webhookId int64, keep int) error { + _, err := d.Exec( + `delete from webhook_deliveries + where webhook_id = ? + and id not in ( + select id from webhook_deliveries + where webhook_id = ? + order by id desc + limit ? + )`, + webhookId, webhookId, keep, + ) + if err != nil { + return fmt.Errorf("failed to prune webhook deliveries: %w", err) + } + return nil +} diff --git a/spindle/db/webhooks_smoke_test.go b/spindle/db/webhooks_smoke_test.go new file mode 100644 index 00000000..c99bd94b --- /dev/null +++ b/spindle/db/webhooks_smoke_test.go @@ -0,0 +1,94 @@ +package db + +import ( + "context" + "fmt" + "path/filepath" + "testing" + + "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/spindle/models" +) + +func TestWebhookSmoke(t *testing.T) { + d, err := Make(context.Background(), filepath.Join(t.TempDir(), "test.db")) + if err != nil { + t.Fatalf("make db: %v", err) + } + + repoDid := syntax.DID("did:plc:repo123") + if err := d.AddRepo(Repo{ + Knot: "knot.example", Owner: "did:plc:owner", Rkey: "r1", + RepoDid: repoDid, Name: "myrepo", CreatedAt: "2024-01-01T00:00:00Z", + }); err != nil { + t.Fatalf("add repo: %v", err) + } + got, err := d.GetRepoByDid(repoDid) + if err != nil || got.Name != "myrepo" { + t.Fatalf("get repo: %+v err=%v", got, err) + } + + wh := &models.Webhook{RepoDid: repoDid, Url: "https://example.com/hook", Active: true, Events: []string{"push", "pull_request:created"}} + created, err := d.AddWebhook(context.Background(), wh, 0, 0) + if err != nil || !created || wh.Id == 0 { + t.Fatalf("add webhook: created=%v id=%d err=%v", created, wh.Id, err) + } + + if created, err := d.AddWebhook(context.Background(), &models.Webhook{ + RepoDid: repoDid, Url: "https://example.com/second", Active: true, Events: []string{"push"}, + }, 1, 0); err != nil || created { + t.Fatalf("expected repo limit to refuse the second webhook: created=%v err=%v", created, err) + } + if created, err := d.AddWebhook(context.Background(), &models.Webhook{ + RepoDid: repoDid, Url: "https://example.com/third", Active: true, Events: []string{"push"}, + }, 0, 1); err != nil || created { + t.Fatalf("expected owner limit to refuse the second webhook: created=%v err=%v", created, err) + } + active, err := d.GetActiveWebhooksForRepo(repoDid) + if err != nil || len(active) != 1 || !active[0].HasEvent(models.WebhookEventPush) { + t.Fatalf("active webhooks: %+v err=%v", active, err) + } + + dv := &models.WebhookDelivery{WebhookId: wh.Id, Event: "push", DeliveryId: "uuid-1", Url: wh.Url, RequestBody: "{}", Success: true, ResponseCode: 200} + if err := d.AddWebhookDelivery(dv); err != nil { + t.Fatalf("add delivery: %v", err) + } + list, err := d.GetWebhookDeliveriesForWebhooks(repoDid, []int64{wh.Id}, 10) + if err != nil || len(list) != 1 { + t.Fatalf("deliveries: %d err=%v", len(list), err) + } + + for i := range 3 { + if err := d.AddWebhookDelivery(&models.WebhookDelivery{ + WebhookId: wh.Id, Event: "push", DeliveryId: fmt.Sprintf("uuid-%d", i+2), + Url: wh.Url, RequestBody: "{}", Success: true, ResponseCode: 200, + }); err != nil { + t.Fatalf("add delivery %d: %v", i, err) + } + } + forHooks, err := d.GetWebhookDeliveriesForWebhooks(repoDid, []int64{wh.Id}, 2) + if err != nil { + t.Fatalf("bulk deliveries: %v", err) + } + if len(forHooks) != 2 { + t.Fatalf("expected 2 newest deliveries, got %d", len(forHooks)) + } + if forHooks[0].DeliveryId != "uuid-4" || forHooks[0].WebhookId != wh.Id { + t.Fatalf("unexpected newest delivery: %+v", forHooks[0]) + } + if got, err := d.GetWebhookDeliveriesForWebhooks(repoDid, []int64{wh.Id + 999}, 2); err != nil || len(got) != 0 { + t.Fatalf("expected no deliveries for an unknown webhook: got=%d err=%v", len(got), err) + } + byId, err := d.GetWebhookDelivery("uuid-1") + if err != nil || !byId.Success || byId.ResponseCode != 200 { + t.Fatalf("delivery by id: %+v err=%v", byId, err) + } + + // cascade: deleting the webhook removes its deliveries + if err := d.DeleteWebhook(wh.Id); err != nil { + t.Fatalf("delete webhook: %v", err) + } + if left, _ := d.GetWebhookDeliveriesForWebhooks(repoDid, []int64{wh.Id}, 10); len(left) != 0 { + t.Fatalf("expected deliveries cascade-deleted, got %d", len(left)) + } +} diff --git a/spindle/models/webhook.go b/spindle/models/webhook.go new file mode 100644 index 00000000..dea0434e --- /dev/null +++ b/spindle/models/webhook.go @@ -0,0 +1,153 @@ +package models + +import ( + "fmt" + "slices" + "strings" + "time" + + "github.com/bluesky-social/indigo/atproto/syntax" +) + +type WebhookEvent string + +const ( + WebhookEventPush WebhookEvent = "push" + WebhookEventRepoRenamed WebhookEvent = "repository:renamed" + WebhookEventPullRequestCreated WebhookEvent = "pull_request:created" + WebhookEventPullRequestResubmitted WebhookEvent = "pull_request:resubmitted" + WebhookEventPullRequestMerged WebhookEvent = "pull_request:merged" + WebhookEventPullRequestClosed WebhookEvent = "pull_request:closed" + WebhookEventPullRequestReopened WebhookEvent = "pull_request:reopened" +) + +type Webhook struct { + Id int64 + RepoDid syntax.DID + Url string + Secret string + Active bool + Events []string // comma-separated event types + CreatedAt time.Time + UpdatedAt time.Time +} + +func ParseWebhookEvents(events []string) ([]WebhookEvent, error) { + known := map[string]WebhookEvent{ + string(WebhookEventPush): WebhookEventPush, + string(WebhookEventRepoRenamed): WebhookEventRepoRenamed, + string(WebhookEventPullRequestCreated): WebhookEventPullRequestCreated, + string(WebhookEventPullRequestResubmitted): WebhookEventPullRequestResubmitted, + string(WebhookEventPullRequestMerged): WebhookEventPullRequestMerged, + string(WebhookEventPullRequestClosed): WebhookEventPullRequestClosed, + string(WebhookEventPullRequestReopened): WebhookEventPullRequestReopened, + } + var parsed []WebhookEvent + for _, raw := range events { + event, ok := known[strings.TrimSpace(raw)] + if !ok { + return nil, fmt.Errorf("unknown webhook event %q", raw) + } + if !slices.Contains(parsed, event) { + parsed = append(parsed, event) + } + } + return parsed, nil +} + +// HasEvent checks if the webhook is subscribed to a specific event +func (w *Webhook) HasEvent(event WebhookEvent) bool { + return slices.Contains(w.Events, string(event)) +} + +type WebhookDelivery struct { + Id int64 + WebhookId int64 + Event string + DeliveryId string // UUID for tracking + Url string + RequestBody string + ResponseCode int + ResponseBody string + Success bool + CreatedAt time.Time +} + +// WebhookPayload represents the webhook payload structure +type WebhookPayload struct { + Ref string `json:"ref"` + Before string `json:"before"` + After string `json:"after"` + Repository WebhookRepository `json:"repository"` + Pusher WebhookUser `json:"pusher"` +} + +// WebhookRepository represents repository information in webhook payload. +// +// Note: spindle stores less repo metadata than the appview did, so +// Description, Website, StarsCount, OpenIssues and Fork are best-effort and +// may be empty/false when spindle has no record of them. +type WebhookRepository struct { + Name string `json:"name"` + FullName string `json:"full_name"` + Description string `json:"description"` + Fork bool `json:"fork"` + HtmlUrl string `json:"html_url"` + CloneUrl string `json:"clone_url"` + SshUrl string `json:"ssh_url"` + Website string `json:"website,omitempty"` + StarsCount int `json:"stars_count,omitempty"` + OpenIssues int `json:"open_issues_count,omitempty"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` + Owner WebhookUser `json:"owner"` +} + +// WebhookUser represents user information in webhook payload +type WebhookUser struct { + Did string `json:"did"` +} + +// WebhookRenamePayload represents the payload for a repository:renamed event +type WebhookRenamePayload struct { + OldName string `json:"old_name"` + NewName string `json:"new_name"` + Repository WebhookRepository `json:"repository"` + Sender WebhookUser `json:"sender"` +} + +// WebhookPullRequestPayload represents the payload for pull_request:* events +type WebhookPullRequestPayload struct { + Action string `json:"action"` + PullRequest WebhookPullRequest `json:"pull_request"` + Repository WebhookRepository `json:"repository"` + Sender WebhookUser `json:"sender"` +} + +// WebhookPullRequest represents pull request information in webhook payload. +// +// Note: the sh.tangled.repo.pull record carries no appview-assigned pull +// number or state, so Number is 0 and State is derived from the lifecycle +// action. HtmlUrl/PatchUrl are omitted because spindle does not know the +// appview base URL. +type WebhookPullRequest struct { + Number int `json:"number"` + Title string `json:"title"` + Body string `json:"body"` + State string `json:"state"` + TargetBranch string `json:"target_branch"` + Source *WebhookPullRequestSource `json:"source,omitempty"` + RoundNumber int `json:"round_number"` + Owner WebhookUser `json:"owner"` + HtmlUrl string `json:"html_url,omitempty"` + PatchUrl string `json:"patch_url,omitempty"` + CreatedAt string `json:"created_at"` +} + +// WebhookPullRequestSource represents the source of a branch- or fork-based +// pull request; absent for patch-based pull requests +type WebhookPullRequestSource struct { + Branch string `json:"branch"` + Repo string `json:"repo,omitempty"` + Sha string `json:"sha,omitempty"` +} diff --git a/spindle/wipe.go b/spindle/wipe.go index f3a29ff1..dacf7abb 100644 --- a/spindle/wipe.go +++ b/spindle/wipe.go @@ -65,6 +65,8 @@ func (s *Spindle) WipeRepo(ctx context.Context, repoDid syntax.DID, reason strin fail("delete queued jobs", s.db.DeleteJobsByRepo(ctx, repoDid.String())) fail("delete quota state", s.db.DeleteQuotaStateForRepo(ctx, repoDid.String())) fail("delete secrets", s.vault.RemoveAllSecrets(ctx, secrets.RepoIdentifier(repoDid.String()))) + fail("delete webhooks", s.db.DeleteWebhooksByRepo(repoDid)) + fail("delete pull rounds", s.db.DeletePullRoundsByRepo(repoDid)) collabs, err := s.db.ListCollaboratorsByRepoDid(repoDid) fail("list collaborators", err) diff --git a/spindle/wipe_test.go b/spindle/wipe_test.go index e5c2d0c7..1c63c96f 100644 --- a/spindle/wipe_test.go +++ b/spindle/wipe_test.go @@ -101,6 +101,15 @@ func seedWipeRepo(t *testing.T, s *Spindle, repoDid, owner syntax.DID) { if _, err := s.db.Exec(`insert into quota_allocations (repo_did, resource, kind, key, amount) values (?, 'compute', 'generic', 'k', 1)`, repoDid.String()); err != nil { t.Fatal(err) } + if _, err := s.db.Exec(`insert into webhooks (repo_did, url, events) values (?, 'https://example.com/hook', 'push')`, repoDid.String()); err != nil { + t.Fatal(err) + } + if _, err := s.db.Exec(`insert into webhook_deliveries (webhook_id, event, delivery_id, url, success) values ((select max(id) from webhooks), 'push', ?, 'https://example.com/hook', 1)`, "d"+sfx); err != nil { + t.Fatal(err) + } + if _, err := s.db.Exec(`insert into pull_rounds (repo_did, rkey, rounds) values (?, 'pull1', 2)`, repoDid.String()); err != nil { + t.Fatal(err) + } if err := s.vault.AddSecret(ctx, secrets.UnlockedSecret{Key: "api_key", Value: "v", Repo: secrets.RepoIdentifier(repoDid.String()), CreatedBy: owner}); err != nil { t.Fatal(err) } @@ -145,7 +154,7 @@ func TestWipeRepoRemovesAllState(t *testing.T) { } var n int - for _, table := range []string{"repos", "jobs", "mill_leases", "mill_artifacts", "quota_allocations", "repo_collaborators", "events"} { + for _, table := range []string{"repos", "jobs", "mill_leases", "mill_artifacts", "quota_allocations", "repo_collaborators", "events", "webhooks", "webhook_deliveries", "pull_rounds"} { if err := s.db.QueryRow(`select count(*) from ` + table).Scan(&n); err != nil { t.Fatal(err) } -- 2.51.2