feat: image variants cache panel

internal/infrastructure/usage/variantUsageTracker.go

@@ -0,0 +1,150 @@
+package usage
+
+import (
+	"RenewCMS/internal/domain/image"
+	"context"
+	"sync"
+	"time"
+)
+
+const FlushKeyLimit = 500
+
+func FlushInterval() time.Duration {
+	return 30 * time.Second
+}
+
+type hitKey struct {
+	variantName string
+	sourceName  string
+}
+
+type Tracker struct {
+	repository image.VariantUsageRepository
+	now        func() time.Time
+	interval   time.Duration
+	keyLimit   int
+	flushing   sync.Mutex
+	pending    sync.Mutex
+	hits       map[hitKey]time.Time
+	rejections map[time.Time]int
+	wake       chan struct{}
+}
+
+var _ image.VariantUsageTracker = &Tracker{}
+
+func NewTracker(repository image.VariantUsageRepository, now func() time.Time) *Tracker {
+	return NewTrackerWith(repository, now, FlushInterval(), FlushKeyLimit)
+}
+
+func NewTrackerWith(repository image.VariantUsageRepository, now func() time.Time, interval time.Duration, keyLimit int) *Tracker {
+	return &Tracker{
+		repository: repository,
+		now:        now,
+		interval:   interval,
+		keyLimit:   keyLimit,
+		hits:       map[hitKey]time.Time{},
+		rejections: map[time.Time]int{},
+		wake:       make(chan struct{}, 1),
+	}
+}
+
+func (t *Tracker) RecordHit(variantName string, sourceName string) {
+	at := t.now().UTC()
+	t.pending.Lock()
+	defer t.pending.Unlock()
+	keepLatest(t.hits, hitKey{variantName, sourceName}, at)
+	if len(t.hits) >= t.keyLimit {
+		t.wakeLoop()
+	}
+}
+
+func (t *Tracker) RecordRejection() {
+	day := image.RejectionDay(t.now())
+	t.pending.Lock()
+	defer t.pending.Unlock()
+	t.rejections[day]++
+}
+
+func (t *Tracker) wakeLoop() {
+	select {
+	case t.wake <- struct{}{}:
+	default:
+	}
+}
+
+func keepLatest(hits map[hitKey]time.Time, key hitKey, at time.Time) {
+	if current, found := hits[key]; !found || at.After(current) {
+		hits[key] = at
+	}
+}
+
+func (t *Tracker) Flush() error {
+	t.flushing.Lock()
+	defer t.flushing.Unlock()
+
+	hits, rejections := t.take()
+	if len(hits) == 0 && len(rejections) == 0 {
+		return nil
+	}
+	batch := image.VariantUsageBatch{
+		Hits:        hitsOf(hits),
+		Rejections:  rejectionsOf(rejections),
+		PruneBefore: image.RejectionPruneBefore(t.now()),
+	}
+	if err := t.repository.Record(batch); err != nil {
+		t.restore(hits, rejections)
+		return err
+	}
+	return nil
+}
+
+func (t *Tracker) take() (map[hitKey]time.Time, map[time.Time]int) {
+	t.pending.Lock()
+	defer t.pending.Unlock()
+	hits, rejections := t.hits, t.rejections
+	t.hits, t.rejections = map[hitKey]time.Time{}, map[time.Time]int{}
+	return hits, rejections
+}
+
+func (t *Tracker) restore(hits map[hitKey]time.Time, rejections map[time.Time]int) {
+	t.pending.Lock()
+	defer t.pending.Unlock()
+	for key, at := range hits {
+		keepLatest(t.hits, key, at)
+	}
+	for day, count := range rejections {
+		t.rejections[day] += count
+	}
+}
+
+func hitsOf(hits map[hitKey]time.Time) []image.VariantHit {
+	listed := make([]image.VariantHit, 0, len(hits))
+	for key, at := range hits {
+		listed = append(listed, image.VariantHit{VariantName: key.variantName, SourceName: key.sourceName, At: at})
+	}
+	return listed
+}
+
+func rejectionsOf(rejections map[time.Time]int) []image.RejectionCount {
+	listed := make([]image.RejectionCount, 0, len(rejections))
+	for day, count := range rejections {
+		listed = append(listed, image.RejectionCount{Day: day, Count: count})
+	}
+	return listed
+}
+
+func (t *Tracker) Run(ctx context.Context) {
+	ticker := time.NewTicker(t.interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			_ = t.Flush()
+			return
+		case <-ticker.C:
+			_ = t.Flush()
+		case <-t.wake:
+			_ = t.Flush()
+		}
+	}
+}