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
}