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
11 "github.com/bnema/zerowrap"
12 "go.opentelemetry.io/otel"
13 "go.opentelemetry.io/otel/attribute"
14 "go.opentelemetry.io/otel/metric"
15 "go.opentelemetry.io/otel/trace"
16
17 "github.com/bnema/gordon/internal/adapters/out/telemetry"
18 "github.com/bnema/gordon/internal/boundaries/out"
19 "github.com/bnema/gordon/internal/domain"
20 "github.com/bnema/gordon/pkg/validation"
21)
22
23var registryTracer = otel.Tracer("gordon.registry")
24
25// Service implements the RegistryService interface.
26type Service struct {
27 blobStorage out.BlobStorage
28 manifestStorage out.ManifestStorage
29 eventBus out.EventPublisher
30 metrics *telemetry.Metrics
31}
32
33// SetMetrics sets the telemetry metrics for the registry service.
34func (s *Service) SetMetrics(m *telemetry.Metrics) {
35 s.metrics = m
36}
37
38// NewService creates a new registry service.
39func NewService(
40 blobStorage out.BlobStorage,
41 manifestStorage out.ManifestStorage,
42 eventBus out.EventPublisher,
43) *Service {
44 return &Service{
45 blobStorage: blobStorage,
46 manifestStorage: manifestStorage,
47 eventBus: eventBus,
48 }
49}
50
51// GetManifest retrieves a manifest by name and reference.
52func (s *Service) GetManifest(ctx context.Context, name, reference string) (*domain.Manifest, error) {
53 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
54 zerowrap.FieldLayer: "usecase",
55 zerowrap.FieldUseCase: "GetManifest",
56 "name": name,
57 "reference": reference,
58 })
59 log := zerowrap.FromCtx(ctx)
60
61 data, contentType, err := s.manifestStorage.GetManifest(name, reference)
62 if err != nil {
63 return nil, log.WrapErr(err, "failed to get manifest")
64 }
65
66 return &domain.Manifest{
67 Name: name,
68 Reference: reference,
69 ContentType: contentType,
70 Data: data,
71 }, nil
72}
73
74// PutManifest stores a manifest and returns the calculated digest.
75func (s *Service) PutManifest(ctx context.Context, manifest *domain.Manifest) (string, error) {
76 ctx, span := registryTracer.Start(ctx, "registry.put_manifest",
77 trace.WithAttributes(
78 attribute.String("name", manifest.Name),
79 attribute.String("reference", manifest.Reference),
80 attribute.Int("manifest_size", len(manifest.Data)),
81 ))
82 defer span.End()
83
84 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
85 zerowrap.FieldLayer: "usecase",
86 zerowrap.FieldUseCase: "PutManifest",
87 "name": manifest.Name,
88 "reference": manifest.Reference,
89 })
90 log := zerowrap.FromCtx(ctx)
91
92 // Calculate digest
93 digest := fmt.Sprintf("sha256:%x", sha256.Sum256(manifest.Data))
94
95 if err := s.manifestStorage.PutManifest(manifest.Name, manifest.Reference, manifest.ContentType, manifest.Data); err != nil {
96 return "", log.WrapErr(err, "failed to store manifest")
97 }
98
99 // Record push metrics
100 if s.metrics != nil {
101 attrs := metric.WithAttributes(
102 attribute.String("name", manifest.Name),
103 attribute.String("reference", manifest.Reference),
104 )
105 s.metrics.ImagePushTotal.Add(ctx, 1, attrs)
106 s.metrics.ImagePushSize.Add(ctx, int64(len(manifest.Data)), attrs)
107 }
108
109 // Publish image pushed event only for tag references (not digests).
110 // A docker push sends manifests by both digest and tag; firing only on
111 // tag prevents duplicate deploy triggers for the same push.
112 if s.eventBus != nil && !strings.HasPrefix(manifest.Reference, "sha256:") {
113 if err := s.eventBus.Publish(domain.EventImagePushed, domain.ImagePushedPayload{
114 Name: manifest.Name,
115 Reference: manifest.Reference,
116 Manifest: manifest.Data,
117 Annotations: manifest.Annotations,
118 }); err != nil {
119 log.Warn().Err(err).Msg("failed to publish image pushed event")
120 }
121 }
122
123 log.Info().Str("digest", digest).Msg("manifest stored")
124 return digest, nil
125}
126
127// DeleteManifest removes a manifest.
128func (s *Service) DeleteManifest(ctx context.Context, name, reference string) error {
129 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
130 zerowrap.FieldLayer: "usecase",
131 zerowrap.FieldUseCase: "DeleteManifest",
132 "name": name,
133 "reference": reference,
134 })
135 log := zerowrap.FromCtx(ctx)
136
137 if err := s.manifestStorage.DeleteManifest(name, reference); err != nil {
138 return log.WrapErr(err, "failed to delete manifest")
139 }
140
141 log.Info().Msg("manifest deleted")
142 return nil
143}
144
145// GetBlob retrieves a blob by digest.
146func (s *Service) GetBlob(ctx context.Context, digest string) (io.ReadCloser, error) {
147 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
148 zerowrap.FieldLayer: "usecase",
149 zerowrap.FieldUseCase: "GetBlob",
150 "digest": digest,
151 })
152 log := zerowrap.FromCtx(ctx)
153
154 reader, err := s.blobStorage.GetBlob(digest)
155 if err != nil {
156 return nil, log.WrapErr(err, "failed to get blob")
157 }
158
159 return reader, nil
160}
161
162// GetBlobPath returns the filesystem path to a blob for direct serving.
163func (s *Service) GetBlobPath(ctx context.Context, digest string) (string, error) {
164 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
165 zerowrap.FieldLayer: "usecase",
166 zerowrap.FieldUseCase: "GetBlobPath",
167 "digest": digest,
168 })
169 log := zerowrap.FromCtx(ctx)
170
171 path, err := s.blobStorage.GetBlobPath(digest)
172 if err != nil {
173 return "", log.WrapErr(err, "failed to get blob path")
174 }
175
176 return path, nil
177}
178
179// PutBlob stores a blob with the given digest.
180func (s *Service) PutBlob(ctx context.Context, digest string, data io.Reader, size int64) error {
181 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
182 zerowrap.FieldLayer: "usecase",
183 zerowrap.FieldUseCase: "PutBlob",
184 "digest": digest,
185 zerowrap.FieldSize: size,
186 })
187 log := zerowrap.FromCtx(ctx)
188
189 if err := s.blobStorage.PutBlob(digest, data, size); err != nil {
190 return log.WrapErr(err, "failed to store blob")
191 }
192
193 log.Info().Msg("blob stored")
194 return nil
195}
196
197// BlobExists checks if a blob exists.
198func (s *Service) BlobExists(_ context.Context, digest string) bool {
199 return s.blobStorage.BlobExists(digest)
200}
201
202// StartUpload starts a new blob upload.
203func (s *Service) StartUpload(ctx context.Context, name string) (string, error) {
204 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
205 zerowrap.FieldLayer: "usecase",
206 zerowrap.FieldUseCase: "StartUpload",
207 "name": name,
208 })
209 log := zerowrap.FromCtx(ctx)
210
211 uuid, err := s.blobStorage.StartBlobUpload(name)
212 if err != nil {
213 return "", log.WrapErr(err, "failed to start blob upload")
214 }
215
216 log.Info().Str("uuid", uuid).Msg("blob upload started")
217 return uuid, nil
218}
219
220// AppendBlobChunk appends data to an in-progress blob upload.
221func (s *Service) AppendBlobChunk(ctx context.Context, name, uuid string, data io.Reader) (int64, error) {
222 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
223 zerowrap.FieldLayer: "usecase",
224 zerowrap.FieldUseCase: "AppendBlobChunk",
225 "name": name,
226 "uuid": uuid,
227 })
228 log := zerowrap.FromCtx(ctx)
229
230 length, err := s.blobStorage.AppendBlobChunk(name, uuid, data)
231 if err != nil {
232 return 0, log.WrapErr(err, "failed to append blob chunk")
233 }
234
235 return length, nil
236}
237
238// FinishUpload completes a blob upload.
239func (s *Service) FinishUpload(ctx context.Context, uuid, digest string) error {
240 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
241 zerowrap.FieldLayer: "usecase",
242 zerowrap.FieldUseCase: "FinishUpload",
243 "uuid": uuid,
244 "digest": digest,
245 })
246 log := zerowrap.FromCtx(ctx)
247
248 if err := s.blobStorage.FinishBlobUpload(uuid, digest); err != nil {
249 return log.WrapErr(err, "failed to finish blob upload")
250 }
251
252 log.Info().Msg("blob upload finished")
253 return nil
254}
255
256// CancelUpload cancels an in-progress upload.
257func (s *Service) CancelUpload(ctx context.Context, uuid string) error {
258 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
259 zerowrap.FieldLayer: "usecase",
260 zerowrap.FieldUseCase: "CancelUpload",
261 "uuid": uuid,
262 })
263 log := zerowrap.FromCtx(ctx)
264
265 if err := s.blobStorage.CancelBlobUpload(uuid); err != nil {
266 return log.WrapErr(err, "failed to cancel blob upload")
267 }
268
269 log.Info().Msg("blob upload cancelled")
270 return nil
271}
272
273// ListTags returns all tags for a repository.
274func (s *Service) ListTags(ctx context.Context, name string) ([]string, error) {
275 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
276 zerowrap.FieldLayer: "usecase",
277 zerowrap.FieldUseCase: "ListTags",
278 "name": name,
279 })
280 log := zerowrap.FromCtx(ctx)
281
282 tags, err := s.manifestStorage.ListTags(name)
283 if err != nil {
284 return nil, log.WrapErr(err, "failed to list tags")
285 }
286
287 filtered := make([]string, 0, len(tags))
288 for _, tag := range tags {
289 if validation.IsDigest(tag) {
290 continue
291 }
292 filtered = append(filtered, tag)
293 }
294
295 return filtered, nil
296}
297
298// ListRepositories returns all repository names.
299func (s *Service) ListRepositories(ctx context.Context) ([]string, error) {
300 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
301 zerowrap.FieldLayer: "usecase",
302 zerowrap.FieldUseCase: "ListRepositories",
303 })
304 log := zerowrap.FromCtx(ctx)
305
306 repos, err := s.manifestStorage.ListRepositories()
307 if err != nil {
308 return nil, log.WrapErr(err, "failed to list repositories")
309 }
310
311 return repos, nil
312}