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	// ValidateURL checks a resolved mirror URL; default domain.ValidateMirrorURL. Tests of
33	// local transports inject a permissive check.
34	ValidateURL func(string) error
35}
36
37// mirrorGen counts the changes of a repository (dirty) and the count the last successful
38// sync saw (clean, meaningful once cleanKnown).
39type mirrorGen struct {
40	dirty, clean uint64
41	cleanKnown   bool
42}
43
44// Mirrors manages outbound repository mirrors.
45type Mirrors struct {
46	cfg      MirrorsConfig
47	configMu sync.RWMutex // configuration changes; held for reads only around store reads and writes, never during a push
48	mu       sync.Mutex
49	gens     map[domain.RepoName]mirrorGen
50	flights  map[domain.RepoName]*mirrorFlight
51	flightWG sync.WaitGroup // one per flight, for tests that wait for the queue to drain
52	ctx      context.Context
53	cancel   context.CancelFunc
54	wg       sync.WaitGroup
55	start    sync.Once
56	closed   bool
57	jobs     chan domain.RepoName
58
59	scheduleMu sync.Mutex
60	lastMinute time.Time
61	deferred   map[domain.RepoName]struct{} // due, but the queue was full
62
63	keyMu      sync.Mutex
64	key        ed25519.PrivateKey
65	keyCreated time.Time
66}
67
68// NewMirrors builds the mirror use case.
69func NewMirrors(cfg MirrorsConfig) *Mirrors {
70	if cfg.ScheduleTick <= 0 {
71		cfg.ScheduleTick = time.Second
72	}
73	if cfg.Workers <= 0 {
74		cfg.Workers = 2
75	}
76	if cfg.QueueSize <= 0 {
77		cfg.QueueSize = mirrorQueueSize
78	}
79	if cfg.ValidateURL == nil {
80		cfg.ValidateURL = domain.ValidateMirrorURL
81	}
82	ctx, cancel := context.WithCancel(context.Background())
83	return &Mirrors{
84		cfg: cfg, ctx: ctx, cancel: cancel,
85		gens:       make(map[domain.RepoName]mirrorGen),
86		flights:    make(map[domain.RepoName]*mirrorFlight),
87		deferred:   make(map[domain.RepoName]struct{}),
88		jobs:       make(chan domain.RepoName, cfg.QueueSize),
89		lastMinute: cfg.Clock.Now().UTC().Truncate(time.Minute),
90	}
91}