wangjia
126 lines
4.0 KiB
Go
126 lines
4.0 KiB
Go
package store
|
|
|
|
import (
|
|
"fmt"
|
|
"time"
|
|
"unicode/utf8"
|
|
|
|
"gorm.io/gorm"
|
|
"gorm.io/gorm/clause"
|
|
|
|
"github.com/wangjia/pay/internal/model"
|
|
)
|
|
|
|
// WebhookDeliveryRow is the row type returned by WebhookStore (alias to the model).
|
|
type WebhookDeliveryRow = model.WebhookDelivery
|
|
|
|
type WebhookStore struct{ db *gorm.DB }
|
|
|
|
func NewWebhookStore(db *gorm.DB) *WebhookStore { return &WebhookStore{db: db} }
|
|
|
|
// EnqueueDelivery inserts an outbox row; a duplicate (out_trade_no,event_type,refund_id)
|
|
// is a no-op (idempotent enqueue) via ON CONFLICT DO NOTHING. refundID is the
|
|
// refund's idempotency dimension (payment events pass "").
|
|
func (s *WebhookStore) EnqueueDelivery(outTradeNo, bizSystem, eventType, refundID, payload string) error {
|
|
row := model.WebhookDelivery{
|
|
OutTradeNo: outTradeNo, BizSystem: bizSystem, EventType: eventType,
|
|
RefundID: refundID, Payload: payload,
|
|
}
|
|
err := s.db.Clauses(clause.OnConflict{
|
|
Columns: []clause.Column{{Name: "out_trade_no"}, {Name: "event_type"}, {Name: "refund_id"}},
|
|
DoNothing: true,
|
|
}).Create(&row).Error
|
|
if err != nil {
|
|
return fmt.Errorf("store.EnqueueDelivery: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *WebhookStore) ListUndelivered(limit int) ([]WebhookDeliveryRow, error) {
|
|
if limit <= 0 || limit > 200 {
|
|
limit = 50
|
|
}
|
|
var out []WebhookDeliveryRow
|
|
if err := s.db.Where("delivered = ?", false).Order("id ASC").Limit(limit).Find(&out).Error; err != nil {
|
|
return nil, fmt.Errorf("store.ListUndelivered: %w", err)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (s *WebhookStore) MarkDelivered(id uint64) error {
|
|
if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id).
|
|
Update("delivered", true).Error; err != nil {
|
|
return fmt.Errorf("store.MarkDelivered: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// truncateUTF8 safely truncates a string to n bytes without splitting UTF-8 runes.
|
|
func truncateUTF8(s string, n int) string {
|
|
if len(s) <= n {
|
|
return s
|
|
}
|
|
s = s[:n]
|
|
for len(s) > 0 && !utf8.ValidString(s) {
|
|
s = s[:len(s)-1]
|
|
}
|
|
return s
|
|
}
|
|
|
|
// MarkFailed increments attempts and records the last error, leaving the row
|
|
// undelivered for the next retry sweep. Superseded in production by
|
|
// ScheduleRetry/MarkDead (退避感知);kept for existing callers/tests.
|
|
func (s *WebhookStore) MarkFailed(id uint64, errMsg string) error {
|
|
errMsg = truncateUTF8(errMsg, 255)
|
|
if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id).
|
|
Updates(map[string]any{
|
|
"attempts": gorm.Expr("attempts + 1"),
|
|
"last_error": errMsg,
|
|
}).Error; err != nil {
|
|
return fmt.Errorf("store.MarkFailed: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ListDeliverable 取「未投递、未死信、且退避到点(next_attempt_at NULL 或 <= now)」的行。
|
|
func (s *WebhookStore) ListDeliverable(now time.Time, limit int) ([]WebhookDeliveryRow, error) {
|
|
if limit <= 0 || limit > 200 {
|
|
limit = 50
|
|
}
|
|
var out []WebhookDeliveryRow
|
|
if err := s.db.
|
|
Where("delivered = ? AND dead = ? AND (next_attempt_at IS NULL OR next_attempt_at <= ?)", false, false, now).
|
|
Order("id ASC").Limit(limit).Find(&out).Error; err != nil {
|
|
return nil, fmt.Errorf("store.ListDeliverable: %w", err)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// ScheduleRetry 记一次失败并排下次重试:attempts+1、last_error、next_attempt_at=nextAt。
|
|
func (s *WebhookStore) ScheduleRetry(id uint64, errMsg string, nextAt time.Time) error {
|
|
errMsg = truncateUTF8(errMsg, 255)
|
|
if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id).
|
|
Updates(map[string]any{
|
|
"attempts": gorm.Expr("attempts + 1"),
|
|
"last_error": errMsg,
|
|
"next_attempt_at": nextAt,
|
|
}).Error; err != nil {
|
|
return fmt.Errorf("store.ScheduleRetry: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// MarkDead 达最大次数后放弃:attempts+1、dead=true、last_error。行留库供人工/对账排查。
|
|
func (s *WebhookStore) MarkDead(id uint64, errMsg string) error {
|
|
errMsg = truncateUTF8(errMsg, 255)
|
|
if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id).
|
|
Updates(map[string]any{
|
|
"attempts": gorm.Expr("attempts + 1"),
|
|
"dead": true,
|
|
"last_error": errMsg,
|
|
}).Error; err != nil {
|
|
return fmt.Errorf("store.MarkDead: %w", err)
|
|
}
|
|
return nil
|
|
}
|