mirrors.go RawHistoryFolder

1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6	"context"
7	"crypto/ed25519"
8	"sync"
9	"time"
10
11	"bnema.dev/ermite/internal/domain"
12	"bnema.dev/ermite/internal/ports"
13)
14
15const mirrorQueueSize = 64
16
17// MirrorsConfig supplies the mirror use case's dependencies.
18type MirrorsConfig struct {
19	Store        ports.MirrorStore
20	Keys         ports.MirrorKeyStore
21	Sealer       ports.SecretSealer
22	Pusher       ports.MirrorPusher
23	Meta         ports.RepoMetaStore
24	Git          ports.GitRepos
25	Reader       ports.RepoReader
26	Bus          ports.EventBus
27	Clock        ports.Clock
28	PublicURL    string
29	ScheduleTick time.Duration
30	Workers      int
31	QueueSize    int // jobs waiting for a worker; default 64
32	// SyncWait bounds how long a manual sync (Sync, SyncAll) holds its caller; past it the
33	// result is Queued and the push continues. Default DefaultMirrorSyncWait.
34	SyncWait time.Duration
35	// ValidateURL checks a resolved mirror URL; default domain.ValidateMirrorURL. Tests of
36	// local transports inject a permissive check.
37	ValidateURL func(string) error
38}
39
40// mirrorGen counts the changes of a repository (dirty) and the count the last successful
41// sync saw (clean, meaningful once cleanKnown).
42type mirrorGen struct {
43	dirty, clean uint64
44	cleanKnown   bool
45}
46
47// Mirrors manages outbound repository mirrors.
48type Mirrors struct {
49	cfg      MirrorsConfig
50	configMu sync.RWMutex // configuration changes; held for reads only around store reads and writes, never during a push
51	mu       sync.Mutex
52	gens     map[domain.RepoName]mirrorGen
53	flights  map[domain.RepoName]*mirrorFlight
54	flightWG sync.WaitGroup // one per flight, for tests that wait for the queue to drain
55	ctx      context.Context
56	cancel   context.CancelFunc
57	wg       sync.WaitGroup
58	start    sync.Once
59	closed   bool
60	jobs     chan domain.RepoName
61
62	scheduleMu sync.Mutex
63	lastMinute time.Time
64	deferred   map[domain.RepoName]struct{} // due, but the queue was full
65
66	keyMu      sync.Mutex
67	key        ed25519.PrivateKey
68	keyCreated time.Time
69}
70
71// NewMirrors builds the mirror use case.
72func NewMirrors(cfg MirrorsConfig) *Mirrors {
73	if cfg.ScheduleTick <= 0 {
74		cfg.ScheduleTick = time.Second
75	}
76	if cfg.Workers <= 0 {
77		cfg.Workers = 2
78	}
79	if cfg.QueueSize <= 0 {
80		cfg.QueueSize = mirrorQueueSize
81	}
82	if cfg.SyncWait <= 0 {
83		cfg.SyncWait = DefaultMirrorSyncWait
84	}
85	if cfg.ValidateURL == nil {
86		cfg.ValidateURL = domain.ValidateMirrorURL
87	}
88	ctx, cancel := context.WithCancel(context.Background())
89	return &Mirrors{
90		cfg: cfg, ctx: ctx, cancel: cancel,
91		gens:       make(map[domain.RepoName]mirrorGen),
92		flights:    make(map[domain.RepoName]*mirrorFlight),
93		deferred:   make(map[domain.RepoName]struct{}),
94		jobs:       make(chan domain.RepoName, cfg.QueueSize),
95		lastMinute: cfg.Clock.Now().UTC().Truncate(time.Minute),
96	}
97}