107 lines
2.9 KiB
Go
107 lines
2.9 KiB
Go
package worker
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"time"
|
|
|
|
"github.com/prabhupadasays/spsays-bff/internal/repository"
|
|
"github.com/prabhupadasays/spsays-bff/pkg/shorturl"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
// ShortURLWorker drains the Redis create-queue + click-queue into Postgres.
|
|
type ShortURLWorker struct {
|
|
repo *repository.ShortURLRepo
|
|
queue *shorturl.Queue
|
|
logger *zap.Logger
|
|
}
|
|
|
|
func NewShortURLWorker(repo *repository.ShortURLRepo, queue *shorturl.Queue, logger *zap.Logger) *ShortURLWorker {
|
|
return &ShortURLWorker{repo: repo, queue: queue, logger: logger}
|
|
}
|
|
|
|
// Run starts the create-worker and click-worker goroutines. Both exit cleanly
|
|
// when ctx is cancelled. Call from server startup.
|
|
func (w *ShortURLWorker) Run(ctx context.Context) {
|
|
go w.runCreateLoop(ctx)
|
|
go w.runClickLoop(ctx)
|
|
}
|
|
|
|
func (w *ShortURLWorker) runCreateLoop(ctx context.Context) {
|
|
w.logger.Info("shorturl create-worker started")
|
|
defer w.logger.Info("shorturl create-worker stopped")
|
|
|
|
for {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
code, err := w.queue.DequeueCreate(ctx, 5*time.Second)
|
|
if err != nil {
|
|
if errors.Is(err, context.Canceled) {
|
|
return
|
|
}
|
|
w.logger.Warn("create-worker dequeue", zap.Error(err))
|
|
time.Sleep(time.Second)
|
|
continue
|
|
}
|
|
if code == "" {
|
|
continue
|
|
}
|
|
|
|
payload, ok := w.queue.LookupPending(ctx, code)
|
|
if !ok {
|
|
// Placeholder without a real payload — likely the producer crashed
|
|
// between MarkPending and EnqueueCreate. Skip.
|
|
w.logger.Warn("create-worker: missing pending payload", zap.String("code", code))
|
|
continue
|
|
}
|
|
|
|
parsed := shorturl.ParsePath(payload.URL)
|
|
if _, err := w.repo.Insert(ctx, code, payload, parsed); err != nil {
|
|
if repository.IsUniqueViolation(err) {
|
|
// Already persisted by a previous attempt; clean up the queue entry.
|
|
_ = w.queue.UnmarkPending(ctx, code)
|
|
continue
|
|
}
|
|
w.logger.Error("create-worker insert", zap.String("code", code), zap.Error(err))
|
|
_ = w.queue.PushDeadLetter(ctx, code, err.Error())
|
|
continue
|
|
}
|
|
|
|
if err := w.queue.MarkCached(ctx, code, payload.URL); err != nil {
|
|
w.logger.Warn("create-worker mark cached", zap.String("code", code), zap.Error(err))
|
|
}
|
|
if err := w.queue.UnmarkPending(ctx, code); err != nil {
|
|
w.logger.Warn("create-worker unmark pending", zap.String("code", code), zap.Error(err))
|
|
}
|
|
}
|
|
}
|
|
|
|
func (w *ShortURLWorker) runClickLoop(ctx context.Context) {
|
|
w.logger.Info("shorturl click-worker started")
|
|
defer w.logger.Info("shorturl click-worker stopped")
|
|
|
|
for {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
ev, ok, err := w.queue.DequeueClick(ctx, 5*time.Second)
|
|
if err != nil {
|
|
if errors.Is(err, context.Canceled) {
|
|
return
|
|
}
|
|
w.logger.Warn("click-worker dequeue", zap.Error(err))
|
|
time.Sleep(time.Second)
|
|
continue
|
|
}
|
|
if !ok {
|
|
continue
|
|
}
|
|
|
|
if err := w.repo.RecordClick(ctx, ev); err != nil {
|
|
w.logger.Warn("click-worker record", zap.String("code", ev.Code), zap.Error(err))
|
|
}
|
|
}
|
|
}
|