87 lines
2.8 KiB
Go
87 lines
2.8 KiB
Go
package webhook
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"time"
|
|
|
|
"github.com/noderunners/nip05api/internal/db"
|
|
)
|
|
|
|
type Repo struct{ db *db.DB }
|
|
|
|
func NewRepo(d *db.DB) *Repo { return &Repo{db: d} }
|
|
|
|
func (r *Repo) Insert(ctx context.Context, eventType EventType, payload string) error {
|
|
_, err := r.db.ExecContext(ctx, `INSERT INTO webhook_outbox
|
|
(event_type, payload, next_attempt_at) VALUES (?, ?, ?)`,
|
|
string(eventType), payload, time.Now().UTC().Format(time.RFC3339))
|
|
return err
|
|
}
|
|
|
|
func (r *Repo) Claim(ctx context.Context, limit int) ([]*OutboxItem, error) {
|
|
rows, err := r.db.QueryContext(ctx, `SELECT id, event_type, payload, attempts,
|
|
last_attempt_at, next_attempt_at, status, COALESCE(last_error, ''), created_at
|
|
FROM webhook_outbox
|
|
WHERE status = 'pending' AND next_attempt_at <= ?
|
|
ORDER BY next_attempt_at ASC LIMIT ?`,
|
|
time.Now().UTC().Format(time.RFC3339), limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
out := []*OutboxItem{}
|
|
for rows.Next() {
|
|
var it OutboxItem
|
|
var status, eventType string
|
|
var lastAttempt, nextAttempt, created sql.NullString
|
|
if err := rows.Scan(&it.ID, &eventType, &it.Payload, &it.Attempts,
|
|
&lastAttempt, &nextAttempt, &status, &it.LastError, &created); err != nil {
|
|
return nil, err
|
|
}
|
|
it.EventType = EventType(eventType)
|
|
it.Status = Status(status)
|
|
if lastAttempt.Valid {
|
|
if t, err := time.Parse(time.RFC3339, lastAttempt.String); err == nil {
|
|
it.LastAttemptAt = &t
|
|
}
|
|
}
|
|
if nextAttempt.Valid {
|
|
if t, err := time.Parse(time.RFC3339, nextAttempt.String); err == nil {
|
|
it.NextAttemptAt = t
|
|
}
|
|
}
|
|
if created.Valid {
|
|
if t, err := time.Parse(time.RFC3339, created.String); err == nil {
|
|
it.CreatedAt = t
|
|
} else if t, err := time.Parse("2006-01-02 15:04:05", created.String); err == nil {
|
|
it.CreatedAt = t
|
|
}
|
|
}
|
|
out = append(out, &it)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (r *Repo) MarkDelivered(ctx context.Context, id int64) error {
|
|
_, err := r.db.ExecContext(ctx, `UPDATE webhook_outbox SET status = 'delivered',
|
|
last_attempt_at = ?, last_error = '' WHERE id = ?`,
|
|
time.Now().UTC().Format(time.RFC3339), id)
|
|
return err
|
|
}
|
|
|
|
func (r *Repo) MarkRetry(ctx context.Context, id int64, attempts int, nextAt time.Time, errMsg string) error {
|
|
_, err := r.db.ExecContext(ctx, `UPDATE webhook_outbox SET attempts = ?,
|
|
last_attempt_at = ?, next_attempt_at = ?, last_error = ? WHERE id = ?`,
|
|
attempts, time.Now().UTC().Format(time.RFC3339),
|
|
nextAt.UTC().Format(time.RFC3339), errMsg, id)
|
|
return err
|
|
}
|
|
|
|
func (r *Repo) MarkDead(ctx context.Context, id int64, errMsg string) error {
|
|
_, err := r.db.ExecContext(ctx, `UPDATE webhook_outbox SET status = 'dead',
|
|
last_attempt_at = ?, last_error = ? WHERE id = ?`,
|
|
time.Now().UTC().Format(time.RFC3339), errMsg, id)
|
|
return err
|
|
}
|