133728e barerepo 1mo
1
package store
133728e barerepo 1mo
3
import (
133728e barerepo 1mo
4
"context"
133728e barerepo 1mo
5
"database/sql"
133728e barerepo 1mo
6
"errors"
133728e barerepo 1mo
7
"time"
133728e barerepo 1mo
10
// HookFailureLimit is chapter 23.4's twenty, after which the hook stops and the page says so.
133728e barerepo 1mo
11
const HookFailureLimit = 20
133728e barerepo 1mo
13
// HookState is what the server remembers about one webhook, keyed by the url in .barerepo/config.
133728e barerepo 1mo
14
type HookState struct {
133728e barerepo 1mo
15
URL string
133728e barerepo 1mo
16
Failures int
133728e barerepo 1mo
17
Disabled bool
133728e barerepo 1mo
18
LastError string
133728e barerepo 1mo
19
LastAt time.Time
133728e barerepo 1mo
22
// HooksOf returns the delivery state of every hook this repository has ever had.
133728e barerepo 1mo
23
func (db *DB) HooksOf(ctx context.Context, repo string) (map[string]HookState, error) {
133728e barerepo 1mo
24
rows, err := db.QueryContext(ctx,
133728e barerepo 1mo
25
`SELECT url, failures, disabled_at, last_error, last_at FROM webhooks WHERE repo = ?`, repo)
133728e barerepo 1mo
26
if err != nil {
133728e barerepo 1mo
27
return nil, err
133728e barerepo 1mo
29
defer rows.Close()
133728e barerepo 1mo
30
out := map[string]HookState{}
133728e barerepo 1mo
31
for rows.Next() {
133728e barerepo 1mo
32
var h HookState
133728e barerepo 1mo
33
var disabled, last sql.NullInt64
133728e barerepo 1mo
34
if err := rows.Scan(&h.URL, &h.Failures, &disabled, &h.LastError, &last); err != nil {
133728e barerepo 1mo
35
return nil, err
133728e barerepo 1mo
37
h.Disabled = disabled.Valid
133728e barerepo 1mo
38
if last.Valid {
133728e barerepo 1mo
39
h.LastAt = time.Unix(last.Int64, 0)
133728e barerepo 1mo
41
out[h.URL] = h
133728e barerepo 1mo
43
return out, rows.Err()
133728e barerepo 1mo
46
// HookDelivered clears the count, because twenty consecutive failures means consecutive.
133728e barerepo 1mo
47
func (db *DB) HookDelivered(ctx context.Context, repo, url string) error {
133728e barerepo 1mo
48
return db.upsertHook(ctx, repo, url, false, "")
133728e barerepo 1mo
51
// HookFailed counts one failure and disables the hook at the limit.
133728e barerepo 1mo
52
func (db *DB) HookFailed(ctx context.Context, repo, url, reason string) error {
133728e barerepo 1mo
53
return db.upsertHook(ctx, repo, url, true, reason)
133728e barerepo 1mo
56
func (db *DB) upsertHook(ctx context.Context, repo, url string, failed bool, reason string) error {
133728e barerepo 1mo
57
at := now().Unix()
133728e barerepo 1mo
58
if !failed {
133728e barerepo 1mo
59
res, err := db.ExecContext(ctx,
133728e barerepo 1mo
60
`UPDATE webhooks SET failures = 0, disabled_at = NULL, last_error = '', last_at = ?
133728e barerepo 1mo
61
WHERE repo = ? AND url = ?`, at, repo, url)
133728e barerepo 1mo
62
if err != nil {
133728e barerepo 1mo
63
return err
133728e barerepo 1mo
65
if n, _ := res.RowsAffected(); n > 0 {
133728e barerepo 1mo
66
return nil
133728e barerepo 1mo
68
_, err = db.ExecContext(ctx,
133728e barerepo 1mo
69
`INSERT INTO webhooks (repo, url, failures, last_error, last_at) VALUES (?, ?, 0, '', ?)`,
133728e barerepo 1mo
70
repo, url, at)
133728e barerepo 1mo
71
return err
133728e barerepo 1mo
73
res, err := db.ExecContext(ctx,
133728e barerepo 1mo
74
`UPDATE webhooks SET failures = failures + 1, last_error = ?, last_at = ?,
133728e barerepo 1mo
75
disabled_at = CASE WHEN failures + 1 >= ? THEN ? ELSE disabled_at END
133728e barerepo 1mo
76
WHERE repo = ? AND url = ?`, reason, at, HookFailureLimit, at, repo, url)
133728e barerepo 1mo
77
if err != nil {
133728e barerepo 1mo
78
return err
133728e barerepo 1mo
80
if n, _ := res.RowsAffected(); n > 0 {
133728e barerepo 1mo
81
return nil
133728e barerepo 1mo
83
_, err = db.ExecContext(ctx,
133728e barerepo 1mo
84
`INSERT INTO webhooks (repo, url, failures, last_error, last_at) VALUES (?, ?, 1, ?, ?)`,
133728e barerepo 1mo
85
repo, url, reason, at)
133728e barerepo 1mo
86
return err
133728e barerepo 1mo
89
// ForgetHook drops the state, so an operator who fixed the receiver can start it again.
133728e barerepo 1mo
90
func (db *DB) ForgetHook(ctx context.Context, repo, url string) error {
133728e barerepo 1mo
91
_, err := db.ExecContext(ctx, `DELETE FROM webhooks WHERE repo = ? AND url = ?`, repo, url)
133728e barerepo 1mo
92
return err
133728e barerepo 1mo
95
// WebhookCursor is the last event dispatched, so a restart neither repeats nor skips.
133728e barerepo 1mo
96
func (db *DB) WebhookCursor(ctx context.Context) (int64, error) {
133728e barerepo 1mo
97
var id int64
133728e barerepo 1mo
98
err := db.QueryRowContext(ctx, `SELECT event_id FROM webhook_cursor WHERE id = 1`).Scan(&id)
133728e barerepo 1mo
99
if errors.Is(err, sql.ErrNoRows) {
133728e barerepo 1mo
100
return 0, nil
133728e barerepo 1mo
101
}
133728e barerepo 1mo
102
return id, err
133728e barerepo 1mo
105
func (db *DB) SetWebhookCursor(ctx context.Context, id int64) error {
133728e barerepo 1mo
106
res, err := db.ExecContext(ctx, `UPDATE webhook_cursor SET event_id = ? WHERE id = 1`, id)
133728e barerepo 1mo
107
if err != nil {
133728e barerepo 1mo
108
return err
133728e barerepo 1mo
109
}
133728e barerepo 1mo
110
if n, _ := res.RowsAffected(); n > 0 {
133728e barerepo 1mo
111
return nil
133728e barerepo 1mo
112
}
133728e barerepo 1mo
113
_, err = db.ExecContext(ctx, `INSERT INTO webhook_cursor (id, event_id) VALUES (1, ?)`, id)
133728e barerepo 1mo
114
return err
133728e barerepo 1mo
117
// StartWebhooksHere puts the cursor at the newest event, so a first start sends no backlog.
133728e barerepo 1mo
118
func (db *DB) StartWebhooksHere(ctx context.Context) error {
133728e barerepo 1mo
119
if _, err := db.WebhookCursor(ctx); err != nil {
133728e barerepo 1mo
120
return err
133728e barerepo 1mo
121
}
133728e barerepo 1mo
122
var id sql.NullInt64
133728e barerepo 1mo
123
if err := db.QueryRowContext(ctx, `SELECT MAX(id) FROM events`).Scan(&id); err != nil {
133728e barerepo 1mo
124
return err
133728e barerepo 1mo
125
}
133728e barerepo 1mo
126
var have int
133728e barerepo 1mo
127
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM webhook_cursor WHERE id = 1`).Scan(&have); err != nil {
133728e barerepo 1mo
128
return err
133728e barerepo 1mo
129
}
133728e barerepo 1mo
130
if have > 0 {
133728e barerepo 1mo
131
return nil
133728e barerepo 1mo
132
}
133728e barerepo 1mo
133
return db.SetWebhookCursor(ctx, id.Int64)
133728e barerepo 1mo
136
// EventsAfter returns the events a hook has not seen, oldest first, which is the order they happened.
133728e barerepo 1mo
137
func (db *DB) EventsAfter(ctx context.Context, after int64, limit int) ([]Event, error) {
133728e barerepo 1mo
138
rows, err := db.QueryContext(ctx,
133728e barerepo 1mo
139
`SELECT id, kind, actor, repo, ref, number, title, detail, created_at FROM events
133728e barerepo 1mo
140
WHERE id > ? ORDER BY id ASC LIMIT ?`, after, limit)
133728e barerepo 1mo
141
if err != nil {
133728e barerepo 1mo
142
return nil, err
133728e barerepo 1mo
143
}
133728e barerepo 1mo
144
defer rows.Close()
133728e barerepo 1mo
145
return scanEvents(rows)