1// Package registry implements the container registry use case.
2package registry
3
4import (
5 "context"
6 "crypto/sha256"
7 "fmt"
8 "io"
9 "strings"
10 "sync"
11 "time"
12
13 "github.com/bnema/zerowrap"
14 "go.opentelemetry.io/otel"
15 "go.opentelemetry.io/otel/attribute"
16 "go.opentelemetry.io/otel/metric"
17 "go.opentelemetry.io/otel/trace"
18
19 "github.com/bnema/gordon/internal/adapters/out/telemetry"
20 "github.com/bnema/gordon/internal/boundaries/out"
21 "github.com/bnema/gordon/internal/domain"
22 "github.com/bnema/gordon/pkg/validation"
23)
24
25var registryTracer = otel.Tracer("gordon.registry")
26
27// Service implements the RegistryService interface.
28type Service struct {
29 blobStorage out.BlobStorage
30 manifestStorage out.ManifestStorage
31 eventBus out.EventPublisher
32 metrics *telemetry.Metrics
33 suppressedImages sync.Map // imageName -> *time.Timer
34}
35
36// SetMetrics sets the telemetry metrics for the registry service.
37func (s *Service) SetMetrics(m *telemetry.Metrics) {
38 s.metrics = m
39}
40
41// NewService creates a new registry service.
42func NewService(
43 blobStorage out.BlobStorage,
44 manifestStorage out.ManifestStorage,
45 eventBus out.EventPublisher,
46) *Service {
47 return &Service{
48 blobStorage: blobStorage,
49 manifestStorage: manifestStorage,
50 eventBus: eventBus,
51 }
52}
53
54// SuppressDeployEvent marks an image name to skip image.pushed events.
55// The suppression auto-expires after 2 minutes to prevent leaks.
56func (s *Service) SuppressDeployEvent(imageName string) {
57 imageName = ExtractImageName(strings.TrimSpace(imageName))
58 if imageName == "" {
59 return
60 }
61
62 var timer *time.Timer
63 timer = time.AfterFunc(2*time.Minute, func() {
64 // Only delete if this timer is still the current one, preventing an
65 // old timer's callback from removing a newer suppression entry.
66 if v, ok := s.suppressedImages.Load(imageName); ok && v == timer {
67 s.suppressedImages.Delete(imageName)
68 }
69 })
70 if existing, loaded := s.suppressedImages.LoadOrStore(imageName, timer); loaded {
71 existing.(*time.Timer).Stop()
72 s.suppressedImages.Store(imageName, timer)
73 }
74}
75
76// ClearDeployEventSuppression removes event suppression for an image.
77func (s *Service) ClearDeployEventSuppression(imageName string) {
78 imageName = ExtractImageName(strings.TrimSpace(imageName))
79 if imageName == "" {
80 return
81 }
82
83 if v, loaded := s.suppressedImages.LoadAndDelete(imageName); loaded {
84 v.(*time.Timer).Stop()
85 }
86}
87
88// ExtractImageName returns just the repository path of a container image
89// reference, stripping any registry host prefix, tag, and digest.
90// Examples:
91//
92// "reg.example.com/team/my-app:latest" -> "team/my-app"
93// "reg.example.com/my-app@sha256:abc" -> "my-app"
94// "my-app:v1.2" -> "my-app"
95func ExtractImageName(imageRef string) string {
96 name := imageRef
97 // Strip digest
98 if idx := strings.Index(name, "@"); idx != -1 {
99 name = name[:idx]
100 }
101 // Strip tag
102 // Find the last colon, but only strip it if it comes after any slash
103 // (to avoid treating a port number in the host as a tag).
104 if idx := strings.LastIndex(name, ":"); idx != -1 {
105 slashIdx := strings.LastIndex(name, "/")
106 if idx > slashIdx {
107 name = name[:idx]
108 }
109 }
110 // Strip registry host: if the first segment contains a dot or colon it is
111 // a registry hostname; remove it.
112 parts := strings.SplitN(name, "/", 2)
113 if len(parts) == 2 {
114 host := parts[0]
115 if strings.ContainsAny(host, ".:") || host == "localhost" {
116 name = parts[1]
117 }
118 }
119 return name
120}
121
122// IsDeployEventSuppressed checks if deploy events are suppressed for an image.
123func (s *Service) IsDeployEventSuppressed(imageName string) bool {
124 imageName = ExtractImageName(strings.TrimSpace(imageName))
125 if imageName == "" {
126 return false
127 }
128
129 _, exists := s.suppressedImages.Load(imageName)
130 return exists
131}
132
133// GetManifest retrieves a manifest by name and reference.
134func (s *Service) GetManifest(ctx context.Context, name, reference string) (*domain.Manifest, error) {
135 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
136 zerowrap.FieldLayer: "usecase",
137 zerowrap.FieldUseCase: "GetManifest",
138 "name": name,
139 "reference": reference,
140 })
141 log := zerowrap.FromCtx(ctx)
142
143 data, contentType, err := s.manifestStorage.GetManifest(name, reference)
144 if err != nil {
145 return nil, log.WrapErr(err, "failed to get manifest")
146 }
147
148 return &domain.Manifest{
149 Name: name,
150 Reference: reference,
151 ContentType: contentType,
152 Data: data,
153 }, nil
154}
155
156// PutManifest stores a manifest and returns the calculated digest.
157func (s *Service) PutManifest(ctx context.Context, manifest *domain.Manifest) (string, error) {
158 ctx, span := registryTracer.Start(ctx, "registry.put_manifest",
159 trace.WithAttributes(
160 attribute.String("name", manifest.Name),
161 attribute.String("reference", manifest.Reference),
162 attribute.Int("manifest_size", len(manifest.Data)),
163 ))
164 defer span.End()
165
166 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
167 zerowrap.FieldLayer: "usecase",
168 zerowrap.FieldUseCase: "PutManifest",
169 "name": manifest.Name,
170 "reference": manifest.Reference,
171 })
172 log := zerowrap.FromCtx(ctx)
173
174 // Calculate digest
175 digest := fmt.Sprintf("sha256:%x", sha256.Sum256(manifest.Data))
176
177 if err := s.manifestStorage.PutManifest(manifest.Name, manifest.Reference, manifest.ContentType, manifest.Data); err != nil {
178 return "", log.WrapErr(err, "failed to store manifest")
179 }
180
181 // Record push metrics
182 if s.metrics != nil {
183 attrs := metric.WithAttributes(
184 attribute.String("name", manifest.Name),
185 attribute.String("reference", manifest.Reference),
186 )
187 s.metrics.ImagePushTotal.Add(ctx, 1, attrs)
188 s.metrics.ImagePushSize.Add(ctx, int64(len(manifest.Data)), attrs)
189 }
190
191 // Publish image pushed event only for tag references (not digests).
192 // A docker push sends manifests by both digest and tag; firing only on
193 // tag prevents duplicate deploy triggers for the same push.
194 if s.eventBus != nil && !strings.HasPrefix(manifest.Reference, "sha256:") {
195 if s.IsDeployEventSuppressed(manifest.Name) {
196 log.Info().Str("image", manifest.Name).Msg("skipping image.pushed event: CLI deploy intent active")
197 } else {
198 if err := s.eventBus.Publish(domain.EventImagePushed, domain.ImagePushedPayload{
199 Name: manifest.Name,
200 Reference: manifest.Reference,
201 Manifest: manifest.Data,
202 Annotations: manifest.Annotations,
203 }); err != nil {
204 log.Warn().Err(err).Msg("failed to publish image pushed event")
205 }
206 }
207 }
208
209 log.Info().Str("digest", digest).Msg("manifest stored")
210 return digest, nil
211}
212
213// DeleteManifest removes a manifest.
214func (s *Service) DeleteManifest(ctx context.Context, name, reference string) error {
215 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
216 zerowrap.FieldLayer: "usecase",
217 zerowrap.FieldUseCase: "DeleteManifest",
218 "name": name,
219 "reference": reference,
220 })
221 log := zerowrap.FromCtx(ctx)
222
223 if err := s.manifestStorage.DeleteManifest(name, reference); err != nil {
224 return log.WrapErr(err, "failed to delete manifest")
225 }
226
227 log.Info().Msg("manifest deleted")
228 return nil
229}
230
231// GetBlob retrieves a blob by digest.
232func (s *Service) GetBlob(ctx context.Context, digest string) (io.ReadCloser, error) {
233 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
234 zerowrap.FieldLayer: "usecase",
235 zerowrap.FieldUseCase: "GetBlob",
236 "digest": digest,
237 })
238 log := zerowrap.FromCtx(ctx)
239
240 reader, err := s.blobStorage.GetBlob(digest)
241 if err != nil {
242 return nil, log.WrapErr(err, "failed to get blob")
243 }
244
245 return reader, nil
246}
247
248// GetBlobPath returns the filesystem path to a blob for direct serving.
249func (s *Service) GetBlobPath(ctx context.Context, digest string) (string, error) {
250 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
251 zerowrap.FieldLayer: "usecase",
252 zerowrap.FieldUseCase: "GetBlobPath",
253 "digest": digest,
254 })
255 log := zerowrap.FromCtx(ctx)
256
257 path, err := s.blobStorage.GetBlobPath(digest)
258 if err != nil {
259 return "", log.WrapErr(err, "failed to get blob path")
260 }
261
262 return path, nil
263}
264
265// PutBlob stores a blob with the given digest.
266func (s *Service) PutBlob(ctx context.Context, digest string, data io.Reader, size int64) error {
267 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
268 zerowrap.FieldLayer: "usecase",
269 zerowrap.FieldUseCase: "PutBlob",
270 "digest": digest,
271 zerowrap.FieldSize: size,
272 })
273 log := zerowrap.FromCtx(ctx)
274
275 if err := s.blobStorage.PutBlob(digest, data, size); err != nil {
276 return log.WrapErr(err, "failed to store blob")
277 }
278
279 log.Info().Msg("blob stored")
280 return nil
281}
282
283// BlobExists checks if a blob exists.
284func (s *Service) BlobExists(_ context.Context, digest string) bool {
285 return s.blobStorage.BlobExists(digest)
286}
287
288// StartUpload starts a new blob upload.
289func (s *Service) StartUpload(ctx context.Context, name string) (string, error) {
290 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
291 zerowrap.FieldLayer: "usecase",
292 zerowrap.FieldUseCase: "StartUpload",
293 "name": name,
294 })
295 log := zerowrap.FromCtx(ctx)
296
297 uuid, err := s.blobStorage.StartBlobUpload(name)
298 if err != nil {
299 return "", log.WrapErr(err, "failed to start blob upload")
300 }
301
302 log.Info().Str("uuid", uuid).Msg("blob upload started")
303 return uuid, nil
304}
305
306// AppendBlobChunk appends data to an in-progress blob upload.
307func (s *Service) AppendBlobChunk(ctx context.Context, name, uuid string, data io.Reader, contentLength, maxBlobSize int64) (int64, error) {
308 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
309 zerowrap.FieldLayer: "usecase",
310 zerowrap.FieldUseCase: "AppendBlobChunk",
311 "name": name,
312 "uuid": uuid,
313 })
314 log := zerowrap.FromCtx(ctx)
315
316 length, err := s.blobStorage.AppendBlobChunk(name, uuid, data, contentLength, maxBlobSize)
317 if err != nil {
318 return 0, log.WrapErr(err, "failed to append blob chunk")
319 }
320
321 return length, nil
322}
323
324// FinishUpload completes a blob upload.
325func (s *Service) FinishUpload(ctx context.Context, uuid, digest string) error {
326 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
327 zerowrap.FieldLayer: "usecase",
328 zerowrap.FieldUseCase: "FinishUpload",
329 "uuid": uuid,
330 "digest": digest,
331 })
332 log := zerowrap.FromCtx(ctx)
333
334 if err := s.blobStorage.FinishBlobUpload(uuid, digest); err != nil {
335 return log.WrapErr(err, "failed to finish blob upload")
336 }
337
338 log.Info().Msg("blob upload finished")
339 return nil
340}
341
342// CancelUpload cancels an in-progress upload.
343func (s *Service) CancelUpload(ctx context.Context, uuid string) error {
344 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
345 zerowrap.FieldLayer: "usecase",
346 zerowrap.FieldUseCase: "CancelUpload",
347 "uuid": uuid,
348 })
349 log := zerowrap.FromCtx(ctx)
350
351 if err := s.blobStorage.CancelBlobUpload(uuid); err != nil {
352 return log.WrapErr(err, "failed to cancel blob upload")
353 }
354
355 log.Info().Msg("blob upload cancelled")
356 return nil
357}
358
359// ListTags returns all tags for a repository.
360func (s *Service) ListTags(ctx context.Context, name string) ([]string, error) {
361 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
362 zerowrap.FieldLayer: "usecase",
363 zerowrap.FieldUseCase: "ListTags",
364 "name": name,
365 })
366 log := zerowrap.FromCtx(ctx)
367
368 tags, err := s.manifestStorage.ListTags(name)
369 if err != nil {
370 return nil, log.WrapErr(err, "failed to list tags")
371 }
372
373 filtered := make([]string, 0, len(tags))
374 for _, tag := range tags {
375 if validation.IsDigest(tag) {
376 continue
377 }
378 filtered = append(filtered, tag)
379 }
380
381 return filtered, nil
382}
383
384// ListRepositories returns all repository names.
385func (s *Service) ListRepositories(ctx context.Context) ([]string, error) {
386 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
387 zerowrap.FieldLayer: "usecase",
388 zerowrap.FieldUseCase: "ListRepositories",
389 })
390 log := zerowrap.FromCtx(ctx)
391
392 repos, err := s.manifestStorage.ListRepositories()
393 if err != nil {
394 return nil, log.WrapErr(err, "failed to list repositories")
395 }
396
397 return repos, nil
398}