Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333package 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}