alice.pds.demo.boltless.dev
This repository has no description
6.1 kB
240 lines
1package artifactstore
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "io"
8 "os"
9 "path/filepath"
10 "strings"
11
12 "github.com/aws/aws-sdk-go-v2/aws"
13 "github.com/aws/aws-sdk-go-v2/config"
14 "github.com/aws/aws-sdk-go-v2/service/s3"
15 spindleconfig "tangled.org/core/spindle/config"
16)
17
18type Writer interface {
19 Put(ctx context.Context, ref string, r io.Reader) error
20}
21
22type Reader interface {
23 Open(ctx context.Context, ref string) (io.ReadCloser, error)
24}
25
26type Store interface {
27 Writer
28 Reader
29}
30
31type DiskStore struct {
32 root string
33}
34
35func NewDiskStore(root string) (*DiskStore, error) {
36 if root == "" {
37 return nil, fmt.Errorf("artifact disk directory is required")
38 }
39 return &DiskStore{root: filepath.Clean(root)}, nil
40}
41
42func (s *DiskStore) Put(_ context.Context, ref string, r io.Reader) error {
43 path, err := s.resolve(ref)
44 if err != nil {
45 return err
46 }
47 if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
48 return fmt.Errorf("mkdir for artifact %q: %w", path, err)
49 }
50 tmpFile, err := os.CreateTemp(filepath.Dir(path), ".tmp-artifact-*")
51 if err != nil {
52 return fmt.Errorf("create temp artifact: %w", err)
53 }
54 tmpPath := tmpFile.Name()
55 defer func() {
56 _ = tmpFile.Close()
57 _ = os.Remove(tmpPath)
58 }()
59 if _, err := io.Copy(tmpFile, r); err != nil {
60 return fmt.Errorf("write artifact content: %w", err)
61 }
62 if err := tmpFile.Sync(); err != nil {
63 return fmt.Errorf("sync artifact file: %w", err)
64 }
65 if err := tmpFile.Close(); err != nil {
66 return fmt.Errorf("close artifact file: %w", err)
67 }
68 if err := os.Rename(tmpPath, path); err != nil {
69 return fmt.Errorf("rename artifact file to target: %w", err)
70 }
71 return nil
72}
73
74func (s *DiskStore) Open(_ context.Context, ref string) (io.ReadCloser, error) {
75 path, err := s.resolve(ref)
76 if err != nil {
77 return nil, err
78 }
79 f, err := os.Open(path)
80 if err != nil {
81 return nil, fmt.Errorf("open disk artifact %q: %w", path, err)
82 }
83 return f, nil
84}
85
86func (s *DiskStore) resolve(ref string) (string, error) {
87 if ref == "" || filepath.IsAbs(ref) {
88 return "", fmt.Errorf("invalid disk artifact ref %q", ref)
89 }
90 path := filepath.Join(s.root, filepath.Clean(ref))
91 rel, err := filepath.Rel(s.root, path)
92 if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
93 return "", fmt.Errorf("artifact ref %q escapes disk root %q", ref, s.root)
94 }
95 return path, nil
96}
97
98type s3API interface {
99 PutObject(ctx context.Context, params *s3.PutObjectInput, optFns ...func(*s3.Options)) (*s3.PutObjectOutput, error)
100 GetObject(ctx context.Context, params *s3.GetObjectInput, optFns ...func(*s3.Options)) (*s3.GetObjectOutput, error)
101}
102
103type S3Store struct {
104 client s3API
105 bucket string
106}
107
108func NewS3Store(client s3API, bucket string) (*S3Store, error) {
109 if client == nil {
110 return nil, fmt.Errorf("s3 client is required")
111 }
112 if bucket == "" {
113 return nil, fmt.Errorf("artifact S3 bucket is required")
114 }
115 return &S3Store{client: client, bucket: bucket}, nil
116}
117
118func (s *S3Store) Put(ctx context.Context, ref string, r io.Reader) error {
119 if err := validateObjectRef(ref); err != nil {
120 return err
121 }
122 _, err := s.client.PutObject(ctx, &s3.PutObjectInput{
123 Bucket: aws.String(s.bucket),
124 Key: aws.String(ref),
125 Body: r,
126 })
127 if err != nil {
128 return fmt.Errorf("s3 put object: %w", err)
129 }
130 return nil
131}
132
133func (s *S3Store) Open(ctx context.Context, ref string) (io.ReadCloser, error) {
134 if err := validateObjectRef(ref); err != nil {
135 return nil, err
136 }
137 res, err := s.client.GetObject(ctx, &s3.GetObjectInput{
138 Bucket: aws.String(s.bucket),
139 Key: aws.String(ref),
140 })
141 if err != nil {
142 return nil, fmt.Errorf("s3 get object: %w", err)
143 }
144 return res.Body, nil
145}
146
147func validateObjectRef(ref string) error {
148 if ref == "" || strings.HasPrefix(ref, "/") || strings.Contains(ref, "://") {
149 return fmt.Errorf("invalid artifact ref %q", ref)
150 }
151 return nil
152}
153
154type Stores struct {
155 order []string
156 stores map[string]Store
157}
158
159func NewStores(cfg spindleconfig.ArtifactStores, diskFallback, legacyS3Bucket string) (*Stores, error) {
160 stores := &Stores{stores: make(map[string]Store)}
161 diskDir := cfg.Disk.Dir
162 if diskDir == "" {
163 diskDir = diskFallback
164 }
165 if diskDir != "" {
166 disk, err := NewDiskStore(diskDir)
167 if err != nil {
168 return nil, err
169 }
170 stores.order = append(stores.order, "disk")
171 stores.stores["disk"] = disk
172 }
173
174 bucket := cfg.S3.Bucket
175 if bucket == "" {
176 bucket = legacyS3Bucket
177 }
178 if bucket != "" {
179 awsCfg, err := config.LoadDefaultConfig(context.Background(), config.WithRegion(cfg.S3.Region))
180 if err != nil {
181 return nil, fmt.Errorf("load aws config: %w", err)
182 }
183 s3Store, err := NewS3Store(s3.NewFromConfig(awsCfg), bucket)
184 if err != nil {
185 return nil, err
186 }
187 stores.order = append(stores.order, "s3")
188 stores.stores["s3"] = s3Store
189 }
190 return stores, nil
191}
192
193func (s *Stores) Names() []string {
194 return append([]string(nil), s.order...)
195}
196
197func (s *Stores) Store(name string) (Store, bool) {
198 store, ok := s.stores[name]
199 return store, ok
200}
201
202func (s *Stores) Open(ctx context.Context, ref string) (io.ReadCloser, error) {
203 var errs []error
204 for _, name := range s.order {
205 rc, err := s.stores[name].Open(ctx, ref)
206 if err == nil {
207 return rc, nil
208 }
209 errs = append(errs, fmt.Errorf("%s: %w", name, err))
210 }
211 return nil, fmt.Errorf("open artifact %q: %w", ref, errors.Join(errs...))
212}
213
214func (s *Stores) PutFile(ctx context.Context, ref, sourcePath string) []error {
215 var errs []error
216 for _, name := range s.order {
217 store := s.stores[name]
218 if disk, ok := store.(*DiskStore); ok {
219 target, err := disk.resolve(ref)
220 if err == nil {
221 source, sourceErr := filepath.Abs(sourcePath)
222 targetAbs, targetErr := filepath.Abs(target)
223 if sourceErr == nil && targetErr == nil && source == targetAbs {
224 continue
225 }
226 }
227 }
228 file, err := os.Open(sourcePath)
229 if err != nil {
230 errs = append(errs, fmt.Errorf("%s: open source: %w", name, err))
231 continue
232 }
233 err = store.Put(ctx, ref, file)
234 _ = file.Close()
235 if err != nil {
236 errs = append(errs, fmt.Errorf("%s: %w", name, err))
237 }
238 }
239 return errs
240}