first commit

internal/dm/worker.go

@@ -0,0 +1,109 @@
+package dm
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/noderunners/nip05api/internal/nostr"
+)
+
+var retrySchedule = []time.Duration{
+	1 * time.Minute,
+	5 * time.Minute,
+	30 * time.Minute,
+	2 * time.Hour,
+	12 * time.Hour,
+}
+
+type Worker struct {
+	repo       *Repo
+	pool       *nostr.Pool
+	senderSk   string
+	kind       int
+	maxRetries int
+}
+
+func NewWorker(repo *Repo, pool *nostr.Pool, nsec string, kind int) (*Worker, error) {
+	sk, err := nostr.NsecToHex(nsec)
+	if err != nil {
+		return nil, err
+	}
+	return &Worker{
+		repo:       repo,
+		pool:       pool,
+		senderSk:   sk,
+		kind:       kind,
+		maxRetries: len(retrySchedule),
+	}, nil
+}
+
+func (w *Worker) Run(ctx context.Context) {
+	t := time.NewTicker(2 * time.Second)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			w.tick(ctx)
+		}
+	}
+}
+
+func (w *Worker) tick(ctx context.Context) {
+	items, err := w.repo.Claim(ctx, 5)
+	if err != nil {
+		slog.Error("dm claim", "err", err)
+		return
+	}
+	if len(items) == 0 {
+		return
+	}
+	var wg sync.WaitGroup
+	for _, it := range items {
+		wg.Add(1)
+		go func(it *OutboxItem) {
+			defer wg.Done()
+			w.deliver(ctx, it)
+		}(it)
+	}
+	wg.Wait()
+}
+
+func (w *Worker) deliver(ctx context.Context, it *OutboxItem) {
+	pubCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+
+	events, err := BuildEvent(w.kind, w.senderSk, it.Pubkey, it.Content)
+	if err != nil {
+		w.handleErr(ctx, it, err)
+		return
+	}
+	for _, ev := range events {
+		ev := ev
+		if err := nostr.Publish(pubCtx, w.pool, &ev); err != nil {
+			w.handleErr(ctx, it, err)
+			return
+		}
+	}
+	_ = w.repo.MarkDelivered(ctx, it.ID)
+	slog.Info("dm delivered", "event", it.EventType, "pubkey", it.Pubkey, "id", it.ID)
+}
+
+func (w *Worker) handleErr(ctx context.Context, it *OutboxItem, err error) {
+	attempts := it.Attempts + 1
+	if attempts >= w.maxRetries {
+		_ = w.repo.MarkDead(ctx, it.ID, err.Error())
+		slog.Error("dm dead", "id", it.ID, "pubkey", it.Pubkey, "err", err)
+		return
+	}
+	idx := attempts - 1
+	if idx >= len(retrySchedule) {
+		idx = len(retrySchedule) - 1
+	}
+	next := time.Now().UTC().Add(retrySchedule[idx])
+	_ = w.repo.MarkRetry(ctx, it.ID, attempts, next, err.Error())
+	slog.Warn("dm retry", "id", it.ID, "attempts", attempts, "err", err)
+}