init
This commit is contained in:
Josh
commit
d69342b2e9
160 changed files with 28681 additions and 0 deletions
308
internal/tenant/webhooks.go
Normal file
308
internal/tenant/webhooks.go
Normal file
|
|
@ -0,0 +1,308 @@
|
|||
package tenant
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
type Webhook struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
URL string `json:"url"`
|
||||
Events []string `json:"events"`
|
||||
Secret string `json:"secret,omitempty"`
|
||||
Enabled bool `json:"enabled"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
LastTriggeredAt *time.Time `json:"last_triggered_at"`
|
||||
LastStatus *string `json:"last_status"`
|
||||
}
|
||||
|
||||
type WebhookDelivery struct {
|
||||
ID int64 `json:"id"`
|
||||
WebhookID string `json:"webhook_id"`
|
||||
Event string `json:"event"`
|
||||
Payload string `json:"payload"`
|
||||
Status string `json:"status"`
|
||||
ResponseCode *int `json:"response_code"`
|
||||
ResponseBody *string `json:"response_body"`
|
||||
Attempts int `json:"attempts"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
type WebhookPayload struct {
|
||||
Event string `json:"event"`
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
Data any `json:"data"`
|
||||
}
|
||||
|
||||
func (q *Queries) CountWebhooks(ctx context.Context) (int, error) {
|
||||
var count int
|
||||
err := q.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM webhooks`).Scan(&count)
|
||||
return count, err
|
||||
}
|
||||
|
||||
func (q *Queries) CreateWebhook(ctx context.Context, name, url string, events []string, secret string) (*Webhook, error) {
|
||||
id := uuid.New().String()
|
||||
eventsJSON, _ := json.Marshal(events)
|
||||
|
||||
_, err := q.db.ExecContext(ctx, `
|
||||
INSERT INTO webhooks (id, name, url, events, secret, enabled, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, 1, CURRENT_TIMESTAMP)
|
||||
`, id, name, url, string(eventsJSON), secret)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return q.GetWebhook(ctx, id)
|
||||
}
|
||||
|
||||
func (q *Queries) GetWebhook(ctx context.Context, id string) (*Webhook, error) {
|
||||
var w Webhook
|
||||
var eventsJSON string
|
||||
var lastTriggeredAt, lastStatus sql.NullString
|
||||
var createdAtStr string
|
||||
|
||||
err := q.db.QueryRowContext(ctx, `
|
||||
SELECT id, name, url, events, secret, enabled, created_at, last_triggered_at, last_status
|
||||
FROM webhooks WHERE id = ?
|
||||
`, id).Scan(&w.ID, &w.Name, &w.URL, &eventsJSON, &w.Secret, &w.Enabled, &createdAtStr, &lastTriggeredAt, &lastStatus)
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
json.Unmarshal([]byte(eventsJSON), &w.Events)
|
||||
w.CreatedAt, _ = time.Parse(time.RFC3339, createdAtStr)
|
||||
if lastTriggeredAt.Valid {
|
||||
t, _ := time.Parse(time.RFC3339, lastTriggeredAt.String)
|
||||
w.LastTriggeredAt = &t
|
||||
}
|
||||
if lastStatus.Valid {
|
||||
w.LastStatus = &lastStatus.String
|
||||
}
|
||||
|
||||
return &w, nil
|
||||
}
|
||||
|
||||
func (q *Queries) ListWebhooks(ctx context.Context) ([]Webhook, error) {
|
||||
rows, err := q.db.QueryContext(ctx, `
|
||||
SELECT id, name, url, events, secret, enabled, created_at, last_triggered_at, last_status
|
||||
FROM webhooks ORDER BY created_at DESC
|
||||
`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var webhooks []Webhook
|
||||
for rows.Next() {
|
||||
var w Webhook
|
||||
var eventsJSON string
|
||||
var lastTriggeredAt, lastStatus sql.NullString
|
||||
var createdAtStr string
|
||||
|
||||
if err := rows.Scan(&w.ID, &w.Name, &w.URL, &eventsJSON, &w.Secret, &w.Enabled, &createdAtStr, &lastTriggeredAt, &lastStatus); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
json.Unmarshal([]byte(eventsJSON), &w.Events)
|
||||
w.CreatedAt, _ = time.Parse(time.RFC3339, createdAtStr)
|
||||
if lastTriggeredAt.Valid {
|
||||
t, _ := time.Parse(time.RFC3339, lastTriggeredAt.String)
|
||||
w.LastTriggeredAt = &t
|
||||
}
|
||||
if lastStatus.Valid {
|
||||
w.LastStatus = &lastStatus.String
|
||||
}
|
||||
|
||||
webhooks = append(webhooks, w)
|
||||
}
|
||||
|
||||
return webhooks, rows.Err()
|
||||
}
|
||||
|
||||
func (q *Queries) UpdateWebhook(ctx context.Context, id, name, url string, events []string, secret string, enabled bool) error {
|
||||
eventsJSON, _ := json.Marshal(events)
|
||||
|
||||
_, err := q.db.ExecContext(ctx, `
|
||||
UPDATE webhooks SET name = ?, url = ?, events = ?, secret = ?, enabled = ?
|
||||
WHERE id = ?
|
||||
`, name, url, string(eventsJSON), secret, enabled, id)
|
||||
return err
|
||||
}
|
||||
|
||||
func (q *Queries) DeleteWebhook(ctx context.Context, id string) error {
|
||||
_, err := q.db.ExecContext(ctx, `DELETE FROM webhooks WHERE id = ?`, id)
|
||||
return err
|
||||
}
|
||||
|
||||
func (q *Queries) ListWebhooksByEvent(ctx context.Context, event string) ([]Webhook, error) {
|
||||
webhooks, err := q.ListWebhooks(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var result []Webhook
|
||||
for _, w := range webhooks {
|
||||
if !w.Enabled {
|
||||
continue
|
||||
}
|
||||
for _, e := range w.Events {
|
||||
if e == event {
|
||||
result = append(result, w)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (q *Queries) TriggerWebhooks(ctx context.Context, event string, data any, baseURL string) {
|
||||
webhooks, err := q.ListWebhooksByEvent(ctx, event)
|
||||
if err != nil || len(webhooks) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
payload := WebhookPayload{
|
||||
Event: event,
|
||||
Timestamp: time.Now().UTC(),
|
||||
Data: data,
|
||||
}
|
||||
payloadJSON, _ := json.Marshal(payload)
|
||||
|
||||
for _, w := range webhooks {
|
||||
go q.deliverWebhook(ctx, w, event, payloadJSON)
|
||||
}
|
||||
}
|
||||
|
||||
func (q *Queries) deliverWebhook(ctx context.Context, w Webhook, event string, payloadJSON []byte) {
|
||||
client := &http.Client{Timeout: 10 * time.Second}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, "POST", w.URL, bytes.NewReader(payloadJSON))
|
||||
if err != nil {
|
||||
q.logDelivery(ctx, w.ID, event, string(payloadJSON), "failed", nil, stringPtr(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("User-Agent", "WriteKit-Webhook/1.0")
|
||||
|
||||
if w.Secret != "" {
|
||||
mac := hmac.New(sha256.New, []byte(w.Secret))
|
||||
mac.Write(payloadJSON)
|
||||
signature := hex.EncodeToString(mac.Sum(nil))
|
||||
req.Header.Set("X-WriteKit-Signature", signature)
|
||||
}
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
q.logDelivery(ctx, w.ID, event, string(payloadJSON), "failed", nil, stringPtr(err.Error()))
|
||||
q.updateWebhookStatus(ctx, w.ID, "failed")
|
||||
return
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
var respBody string
|
||||
buf := make([]byte, 1024)
|
||||
n, _ := resp.Body.Read(buf)
|
||||
respBody = string(buf[:n])
|
||||
|
||||
status := "success"
|
||||
if resp.StatusCode >= 400 {
|
||||
status = "failed"
|
||||
}
|
||||
|
||||
q.logDelivery(ctx, w.ID, event, string(payloadJSON), status, &resp.StatusCode, &respBody)
|
||||
q.updateWebhookStatus(ctx, w.ID, status)
|
||||
}
|
||||
|
||||
func (q *Queries) logDelivery(ctx context.Context, webhookID, event, payload, status string, responseCode *int, responseBody *string) {
|
||||
q.db.ExecContext(ctx, `
|
||||
INSERT INTO webhook_deliveries (webhook_id, event, payload, status, response_code, response_body, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
|
||||
`, webhookID, event, truncate(payload, 1024), status, responseCode, truncate(ptrToString(responseBody), 1024))
|
||||
|
||||
// Cleanup old deliveries - keep last 50 per webhook
|
||||
q.db.ExecContext(ctx, `
|
||||
DELETE FROM webhook_deliveries
|
||||
WHERE webhook_id = ? AND id NOT IN (
|
||||
SELECT id FROM webhook_deliveries WHERE webhook_id = ?
|
||||
ORDER BY created_at DESC LIMIT 50
|
||||
)
|
||||
`, webhookID, webhookID)
|
||||
}
|
||||
|
||||
func (q *Queries) updateWebhookStatus(ctx context.Context, webhookID, status string) {
|
||||
q.db.ExecContext(ctx, `
|
||||
UPDATE webhooks SET last_triggered_at = CURRENT_TIMESTAMP, last_status = ?
|
||||
WHERE id = ?
|
||||
`, status, webhookID)
|
||||
}
|
||||
|
||||
func (q *Queries) ListWebhookDeliveries(ctx context.Context, webhookID string) ([]WebhookDelivery, error) {
|
||||
rows, err := q.db.QueryContext(ctx, `
|
||||
SELECT id, webhook_id, event, payload, status, response_code, response_body, attempts, created_at
|
||||
FROM webhook_deliveries
|
||||
WHERE webhook_id = ?
|
||||
ORDER BY created_at DESC
|
||||
LIMIT 50
|
||||
`, webhookID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var deliveries []WebhookDelivery
|
||||
for rows.Next() {
|
||||
var d WebhookDelivery
|
||||
var createdAtStr string
|
||||
var respCode sql.NullInt64
|
||||
var respBody sql.NullString
|
||||
|
||||
if err := rows.Scan(&d.ID, &d.WebhookID, &d.Event, &d.Payload, &d.Status, &respCode, &respBody, &d.Attempts, &createdAtStr); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
d.CreatedAt, _ = time.Parse(time.RFC3339, createdAtStr)
|
||||
if respCode.Valid {
|
||||
code := int(respCode.Int64)
|
||||
d.ResponseCode = &code
|
||||
}
|
||||
if respBody.Valid {
|
||||
d.ResponseBody = &respBody.String
|
||||
}
|
||||
|
||||
deliveries = append(deliveries, d)
|
||||
}
|
||||
|
||||
return deliveries, rows.Err()
|
||||
}
|
||||
|
||||
func truncate(s string, max int) string {
|
||||
if len(s) <= max {
|
||||
return s
|
||||
}
|
||||
return s[:max]
|
||||
}
|
||||
|
||||
func stringPtr(s string) *string {
|
||||
return &s
|
||||
}
|
||||
|
||||
func ptrToString(p *string) string {
|
||||
if p == nil {
|
||||
return ""
|
||||
}
|
||||
return *p
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue