1package session
 2
 3import (
 4	"context"
 5	"sync"
 6	"sync/atomic"
 7	"time"
 8
 9	"tug/internal/storage"
10)
11
12const (
13	SeenInterval   = 5 * time.Minute
14	SeenFlushEvery = 5 * time.Second
15	SeenPendingMax = 1024
16)
17
18type seenGate struct {
19	interval time.Duration
20	mu       sync.Mutex
21	pending  map[string]time.Time
22	attempts atomic.Int64
23	dropped  atomic.Int64
24}
25
26func (s *Service) touch(row storage.ForwardSession, now time.Time) {
27	if row.LastSeenAt != nil && now.Sub(*row.LastSeenAt) < s.seen.interval {
28		return
29	}
30	s.seen.mu.Lock()
31	defer s.seen.mu.Unlock()
32	if _, ok := s.seen.pending[row.ID]; ok {
33		return
34	}
35	if len(s.seen.pending) >= SeenPendingMax {
36		s.seen.dropped.Add(1)
37		return
38	}
39	if s.seen.pending == nil {
40		s.seen.pending = make(map[string]time.Time)
41	}
42	s.seen.pending[row.ID] = now
43}
44
45func (s *Service) flushSeen(ctx context.Context) {
46	s.seen.mu.Lock()
47	batch := s.seen.pending
48	s.seen.pending = nil
49	s.seen.mu.Unlock()
50	if len(batch) == 0 {
51		return
52	}
53	seen := make([]storage.SessionSeen, 0, len(batch))
54	for id, at := range batch {
55		seen = append(seen, storage.SessionSeen{ID: id, At: at, StaleBefore: at.Add(-s.seen.interval)})
56	}
57	s.seen.attempts.Add(1)
58	s.store.TouchForwardSessions(ctx, seen)
59}