From 0bcb1538ce1d8a98cdf24df3bc7f54d56f33e2a5 Mon Sep 17 00:00:00 2001 From: SayaAndy Date: Thu, 16 Apr 2026 02:06:56 +0700 Subject: feat: s3 input & output clients --- CLAUDE.md | 42 ++++++ config/config-s3.sample.json | 53 +++++++ config/config.go | 26 +++- go.mod | 19 +++ go.sum | 38 +++++ internal/client/input/input_client_interface.go | 1 + internal/client/input/s3.go | 156 ++++++++++++++++++++ internal/client/output/output_client_interface.go | 1 + internal/client/output/s3.go | 169 ++++++++++++++++++++++ 9 files changed, 503 insertions(+), 2 deletions(-) create mode 100644 CLAUDE.md create mode 100644 config/config-s3.sample.json create mode 100644 internal/client/input/s3.go create mode 100644 internal/client/output/s3.go diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..94be9b2 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,42 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Build & Run + +```bash +go build -o saya-today-thumbnail-generator +./saya-today-thumbnail-generator -c config/config-local-unix.sample.json +``` + +No Makefile. No test files exist yet — `go test ./...` would be the convention. + +## What This Is + +Concurrent batch thumbnail generator. Reads images from a source (Backblaze B2, S3/S3-compatible, or local filesystem), converts to WebP/JPEG with configurable quality/size, writes to output storage. Maintains a CSV-based file cache to skip already-processed images. + +## Architecture + +Three pluggable abstractions, each using a factory/registry pattern keyed by string type: + +- **InputClient** (`internal/client/input/`) — scans source files, provides readers. Implementations: `b2`, `s3`, `local-unix` +- **OutputClient** (`internal/client/output/`) — writes converted files. Implementations: `b2`, `s3`, `local-unix` +- **Converter** (`internal/converter/`) — decodes image, resizes (Catmull-Rom), encodes to target format. Implementations: `webp`, `jpeg` + +Each has a `New*Map` factory function returning a map of type-string → constructor. + +## Config System (`config/config.go`) + +JSON config with discriminated unions — `"Type"` field selects which struct to unmarshal into. Supports `${ENV_VAR}` expansion in string values. Validated with go-playground/validator. + +Key config knobs: `MaxProcessThreads` (conversion concurrency), `MaxPreProcessThreads` (I/O concurrency), `RewriteOn` strategy per converter (`Never`/`UnequalHashInCache`/`Always`). + +Sample configs in `config/config-*.sample.json`. Actual configs are gitignored. + +## Concurrency Model + +Two-tier semaphore system in `main.go`: pre-process (file scanning/reading) and process (image conversion) run with separate thread pool limits. Graceful shutdown on SIGTERM/SIGINT via early-termination flag. Cache map protected by `sync.RWMutex`. + +## Environment + +Uses direnv (`.envrc`) for credentials (`B2_KEY_ID`, `B2_APPLICATION_KEY`, `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`). S3 client supports custom endpoints and path-style addressing for S3-compatible stores (MinIO, etc.). diff --git a/config/config-s3.sample.json b/config/config-s3.sample.json new file mode 100644 index 0000000..5b067a5 --- /dev/null +++ b/config/config-s3.sample.json @@ -0,0 +1,53 @@ +{ + "MaxProcessThreads": 4, + "MaxPreProcessThreads": 12, + "LogLevel": "info", + "Input": { + "Storage": { + "Type": "s3", + "Config": { + "BucketName": "my-photos", + "Region": "", + "Prefix": "full/", + "Endpoint": "https://minio.local", + "UsePathStyle": true, + "AccessKeyID": "${AWS_ACCESS_KEY_ID}", + "SecretAccessKey": "${AWS_SECRET_ACCESS_KEY}" + } + }, + "KnownExtensions": [ + "jpg", + "jpeg", + "png" + ], + "CacheProcessed": true, + "CacheProcessedCsvPath": "cache.csv" + }, + "Converters": [ + { + "Type": "webp", + "Config": { + "Quality": 80, + "Size": { + "MaxWidth": 320, + "MaxHeight": 0 + } + }, + "Output": { + "RewriteOn": "UnequalHashInCache", + "Storage": { + "Type": "s3", + "Config": { + "BucketName": "my-photos", + "Region": "", + "Prefix": "webp-320p/", + "Endpoint": "https://minio.local", + "UsePathStyle": true, + "AccessKeyID": "${AWS_ACCESS_KEY_ID}", + "SecretAccessKey": "${AWS_SECRET_ACCESS_KEY}" + } + } + } + } + ] +} diff --git a/config/config.go b/config/config.go index e55d4b7..06fa94f 100644 --- a/config/config.go +++ b/config/config.go @@ -25,7 +25,7 @@ type InputConfig struct { } type InputStorageConfig struct { - Type string `json:"Type" validate:"required,oneof=b2 local-unix"` + Type string `json:"Type" validate:"required,oneof=b2 s3 local-unix"` Config any `json:"Config" validate:"required"` } @@ -48,6 +48,12 @@ func (sc *InputStorageConfig) UnmarshalJSON(data []byte) error { return fmt.Errorf("unmarshal B2Config: %w", err) } sc.Config = &b2Config + case "s3": + var s3Config S3Config + if err := json.Unmarshal(tmp.Config, &s3Config); err != nil { + return fmt.Errorf("unmarshal S3Config: %w", err) + } + sc.Config = &s3Config case "local-unix": var localUnixConfig InputLocalUnixConfig if err := json.Unmarshal(tmp.Config, &localUnixConfig); err != nil { @@ -118,7 +124,7 @@ type SizeConfig struct { } type OutputStorageConfig struct { - Type string `json:"Type" validate:"required,oneof=b2 local-unix"` + Type string `json:"Type" validate:"required,oneof=b2 s3 local-unix"` Config any `json:"Config" validate:"required"` } @@ -141,6 +147,12 @@ func (sc *OutputStorageConfig) UnmarshalJSON(data []byte) error { return fmt.Errorf("unmarshal B2Config: %w", err) } sc.Config = &b2Config + case "s3": + var s3Config S3Config + if err := json.Unmarshal(tmp.Config, &s3Config); err != nil { + return fmt.Errorf("unmarshal S3Config: %w", err) + } + sc.Config = &s3Config case "local-unix": var localUnixConfig OutputLocalUnixConfig if err := json.Unmarshal(tmp.Config, &localUnixConfig); err != nil { @@ -162,6 +174,16 @@ type B2Config struct { ApplicationKey string `json:"ApplicationKey"` } +type S3Config struct { + BucketName string `json:"BucketName" validate:"required,min=1"` + Region string `json:"Region" validate:"required,min=1"` + Prefix string `json:"Prefix"` + Endpoint string `json:"Endpoint"` + UsePathStyle bool `json:"UsePathStyle"` + AccessKeyID string `json:"AccessKeyID"` + SecretAccessKey string `json:"SecretAccessKey"` +} + type InputLocalUnixConfig struct { MaxDepth int `json:"MaxDepth" validate:"required,min=0"` Path string `json:"Path" validate:"required,min=1"` diff --git a/go.mod b/go.mod index 3833a84..a68df13 100644 --- a/go.mod +++ b/go.mod @@ -11,6 +11,25 @@ require ( ) require ( + github.com/aws/aws-sdk-go-v2 v1.41.5 // indirect + github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8 // indirect + github.com/aws/aws-sdk-go-v2/config v1.32.14 // indirect + github.com/aws/aws-sdk-go-v2/credentials v1.19.14 // indirect + github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.21 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.21 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.21 // indirect + github.com/aws/aws-sdk-go-v2/internal/ini v1.8.6 // indirect + github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.22 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.7 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.13 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.21 // indirect + github.com/aws/aws-sdk-go-v2/service/s3 v1.99.0 // indirect + github.com/aws/aws-sdk-go-v2/service/signin v1.0.9 // indirect + github.com/aws/aws-sdk-go-v2/service/sso v1.30.15 // indirect + github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.19 // indirect + github.com/aws/aws-sdk-go-v2/service/sts v1.41.10 // indirect + github.com/aws/smithy-go v1.24.2 // indirect github.com/gabriel-vasile/mimetype v1.4.13 // indirect github.com/go-playground/locales v0.14.1 // indirect github.com/go-playground/universal-translator v0.18.1 // indirect diff --git a/go.sum b/go.sum index d8ea4fd..1a4b043 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,43 @@ github.com/Backblaze/blazer v0.7.2 h1:UWNHMLB+Nf+UmbO2qkVvgriODLEMz4kIyr2Hm+DVXQM= github.com/Backblaze/blazer v0.7.2/go.mod h1:T4y3EYa9IQ5J0PKc/C/J8/CEnSd3qa/lgNw938wZg10= +github.com/aws/aws-sdk-go-v2 v1.41.5 h1:dj5kopbwUsVUVFgO4Fi5BIT3t4WyqIDjGKCangnV/yY= +github.com/aws/aws-sdk-go-v2 v1.41.5/go.mod h1:mwsPRE8ceUUpiTgF7QmQIJ7lgsKUPQOUl3o72QBrE1o= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8 h1:eBMB84YGghSocM7PsjmmPffTa+1FBUeNvGvFou6V/4o= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8/go.mod h1:lyw7GFp3qENLh7kwzf7iMzAxDn+NzjXEAGjKS2UOKqI= +github.com/aws/aws-sdk-go-v2/config v1.32.14 h1:opVIRo/ZbbI8OIqSOKmpFaY7IwfFUOCCXBsUpJOwDdI= +github.com/aws/aws-sdk-go-v2/config v1.32.14/go.mod h1:U4/V0uKxh0Tl5sxmCBZ3AecYny4UNlVmObYjKuuaiOo= +github.com/aws/aws-sdk-go-v2/credentials v1.19.14 h1:n+UcGWAIZHkXzYt87uMFBv/l8THYELoX6gVcUvgl6fI= +github.com/aws/aws-sdk-go-v2/credentials v1.19.14/go.mod h1:cJKuyWB59Mqi0jM3nFYQRmnHVQIcgoxjEMAbLkpr62w= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.21 h1:NUS3K4BTDArQqNu2ih7yeDLaS3bmHD0YndtA6UP884g= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.21/go.mod h1:YWNWJQNjKigKY1RHVJCuupeWDrrHjRqHm0N9rdrWzYI= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.21 h1:Rgg6wvjjtX8bNHcvi9OnXWwcE0a2vGpbwmtICOsvcf4= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.21/go.mod h1:A/kJFst/nm//cyqonihbdpQZwiUhhzpqTsdbhDdRF9c= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.21 h1:PEgGVtPoB6NTpPrBgqSE5hE/o47Ij9qk/SEZFbUOe9A= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.21/go.mod h1:p+hz+PRAYlY3zcpJhPwXlLC4C+kqn70WIHwnzAfs6ps= +github.com/aws/aws-sdk-go-v2/internal/ini v1.8.6 h1:qYQ4pzQ2Oz6WpQ8T3HvGHnZydA72MnLuFK9tJwmrbHw= +github.com/aws/aws-sdk-go-v2/internal/ini v1.8.6/go.mod h1:O3h0IK87yXci+kg6flUKzJnWeziQUKciKrLjcatSNcY= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.22 h1:rWyie/PxDRIdhNf4DzRk0lvjVOqFJuNnO8WwaIRVxzQ= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.22/go.mod h1:zd/JsJ4P7oGfUhXn1VyLqaRZwPmZwg44Jf2dS84Dm3Y= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.7 h1:5EniKhLZe4xzL7a+fU3C2tfUN4nWIqlLesfrjkuPFTY= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.7/go.mod h1:x0nZssQ3qZSnIcePWLvcoFisRXJzcTVvYpAAdYX8+GI= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.13 h1:JRaIgADQS/U6uXDqlPiefP32yXTda7Kqfx+LgspooZM= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.13/go.mod h1:CEuVn5WqOMilYl+tbccq8+N2ieCy0gVn3OtRb0vBNNM= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21 h1:c31//R3xgIJMSC8S6hEVq+38DcvUlgFY0FM6mSI5oto= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21/go.mod h1:r6+pf23ouCB718FUxaqzZdbpYFyDtehyZcmP5KL9FkA= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.21 h1:ZlvrNcHSFFWURB8avufQq9gFsheUgjVD9536obIknfM= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.21/go.mod h1:cv3TNhVrssKR0O/xxLJVRfd2oazSnZnkUeTf6ctUwfQ= +github.com/aws/aws-sdk-go-v2/service/s3 v1.99.0 h1:hlSuz394kV0vhv9drL5lhuEFbEOEP1VyQpy15qWh1Pk= +github.com/aws/aws-sdk-go-v2/service/s3 v1.99.0/go.mod h1:uoA43SdFwacedBfSgfFSjjCvYe8aYBS7EnU5GZ/YKMM= +github.com/aws/aws-sdk-go-v2/service/signin v1.0.9 h1:QKZH0S178gCmFEgst8hN0mCX1KxLgHBKKY/CLqwP8lg= +github.com/aws/aws-sdk-go-v2/service/signin v1.0.9/go.mod h1:7yuQJoT+OoH8aqIxw9vwF+8KpvLZ8AWmvmUWHsGQZvI= +github.com/aws/aws-sdk-go-v2/service/sso v1.30.15 h1:lFd1+ZSEYJZYvv9d6kXzhkZu07si3f+GQ1AaYwa2LUM= +github.com/aws/aws-sdk-go-v2/service/sso v1.30.15/go.mod h1:WSvS1NLr7JaPunCXqpJnWk1Bjo7IxzZXrZi1QQCkuqM= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.19 h1:dzztQ1YmfPrxdrOiuZRMF6fuOwWlWpD2StNLTceKpys= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.19/go.mod h1:YO8TrYtFdl5w/4vmjL8zaBSsiNp3w0L1FfKVKenZT7w= +github.com/aws/aws-sdk-go-v2/service/sts v1.41.10 h1:p8ogvvLugcR/zLBXTXrTkj0RYBUdErbMnAFFp12Lm/U= +github.com/aws/aws-sdk-go-v2/service/sts v1.41.10/go.mod h1:60dv0eZJfeVXfbT1tFJinbHrDfSJ2GZl4Q//OSSNAVw= +github.com/aws/smithy-go v1.24.2 h1:FzA3bu/nt/vDvmnkg+R8Xl46gmzEDam6mZ1hzmwXFng= +github.com/aws/smithy-go v1.24.2/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= diff --git a/internal/client/input/input_client_interface.go b/internal/client/input/input_client_interface.go index a2cf84b..dca64db 100644 --- a/internal/client/input/input_client_interface.go +++ b/internal/client/input/input_client_interface.go @@ -27,5 +27,6 @@ type MetadataStruct struct { var NewInputClientMap = map[string]func(cfg *config.InputConfig) (InputClient, error){ "b2": NewB2InputClient, + "s3": NewS3InputClient, "local-unix": NewLocalUnixInputClient, } diff --git a/internal/client/input/s3.go b/internal/client/input/s3.go new file mode 100644 index 0000000..4f84788 --- /dev/null +++ b/internal/client/input/s3.go @@ -0,0 +1,156 @@ +package input + +import ( + "context" + "fmt" + "io" + "slices" + "strings" + + "github.com/aws/aws-sdk-go-v2/aws" + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/s3" + + "github.com/SayaAndy/saya-today-thumbnail-generator/config" +) + +var _ InputClient = (*S3InputClient)(nil) + +type S3InputClient struct { + prefix string + bucketName string + s3cl *s3.Client + knownExtensions []string +} + +func NewS3InputClient(cfg *config.InputConfig) (InputClient, error) { + if cfg.Storage.Type != "s3" { + return nil, fmt.Errorf("invalid storage type for S3InputClient") + } + s3cfg := cfg.Storage.Config.(*config.S3Config) + + s3cl, err := newS3Client(s3cfg) + if err != nil { + return nil, fmt.Errorf("create S3 client: %w", err) + } + + return &S3InputClient{ + s3cl: s3cl, + bucketName: s3cfg.BucketName, + prefix: s3cfg.Prefix, + knownExtensions: cfg.KnownExtensions, + }, nil +} + +func (c *S3InputClient) Scan() ([]string, error) { + var filePaths []string + + paginator := s3.NewListObjectsV2Paginator(c.s3cl, &s3.ListObjectsV2Input{ + Bucket: aws.String(c.bucketName), + Prefix: aws.String(c.prefix), + }) + + for paginator.HasMorePages() { + output, err := paginator.NextPage(context.Background()) + if err != nil { + return nil, fmt.Errorf("list S3 objects: %w", err) + } + + for _, obj := range output.Contents { + name := aws.ToString(obj.Key) + + if len(c.knownExtensions) != 0 { + nameParts := strings.Split(name, ".") + if len(nameParts) < 2 { + continue + } + ext := strings.ToLower(nameParts[len(nameParts)-1]) + if !slices.Contains(c.knownExtensions, ext) { + continue + } + } + + filePaths = append(filePaths, strings.TrimPrefix(name, c.prefix)) + } + } + + return filePaths, nil +} + +func (c *S3InputClient) ReadMetadata(path string) (*MetadataStruct, error) { + key := c.prefix + path + + head, err := c.s3cl.HeadObject(context.Background(), &s3.HeadObjectInput{ + Bucket: aws.String(c.bucketName), + Key: aws.String(key), + }) + if err != nil { + return nil, fmt.Errorf("head S3 object %s: %w", key, err) + } + + metadata := MetadataStruct{ + Name: key, + StorageType: "s3", + Hash: strings.Trim(aws.ToString(head.ETag), "\""), + ContentType: aws.ToString(head.ContentType), + Misc: head.Metadata, + } + + if head.ContentLength != nil { + metadata.Size = *head.ContentLength + } + if head.LastModified != nil { + metadata.LastModified = *head.LastModified + metadata.FirstCreated = *head.LastModified + } + + return &metadata, nil +} + +func (c *S3InputClient) ID(path string) string { + return fmt.Sprintf("s3://%s/%s%s", c.bucketName, c.prefix, path) +} + +func (c *S3InputClient) GetReader(path string) (io.ReadCloser, error) { + key := c.prefix + path + + output, err := c.s3cl.GetObject(context.Background(), &s3.GetObjectInput{ + Bucket: aws.String(c.bucketName), + Key: aws.String(key), + }) + if err != nil { + return nil, fmt.Errorf("get S3 object %s: %w", key, err) + } + + return output.Body, nil +} + +func newS3Client(cfg *config.S3Config) (*s3.Client, error) { + opts := []func(*awsconfig.LoadOptions) error{ + awsconfig.WithRegion(cfg.Region), + } + + if cfg.AccessKeyID != "" && cfg.SecretAccessKey != "" { + opts = append(opts, awsconfig.WithCredentialsProvider( + credentials.NewStaticCredentialsProvider(cfg.AccessKeyID, cfg.SecretAccessKey, ""), + )) + } + + awsCfg, err := awsconfig.LoadDefaultConfig(context.Background(), opts...) + if err != nil { + return nil, fmt.Errorf("load AWS config: %w", err) + } + + var s3Opts []func(*s3.Options) + if cfg.Endpoint != "" { + s3Opts = append(s3Opts, func(o *s3.Options) { + o.BaseEndpoint = aws.String(cfg.Endpoint) + }) + } + s3Opts = append(s3Opts, func(o *s3.Options) { + o.UsePathStyle = cfg.UsePathStyle + }) + + return s3.NewFromConfig(awsCfg, s3Opts...), nil +} diff --git a/internal/client/output/output_client_interface.go b/internal/client/output/output_client_interface.go index f9e421b..4415aa4 100644 --- a/internal/client/output/output_client_interface.go +++ b/internal/client/output/output_client_interface.go @@ -28,5 +28,6 @@ type MetadataStruct struct { var NewOutputClientMap = map[string]func(cfg *config.OutputConfig) (OutputClient, error){ "b2": NewB2OutputClient, + "s3": NewS3OutputClient, "local-unix": NewLocalUnixOutputClient, } diff --git a/internal/client/output/s3.go b/internal/client/output/s3.go new file mode 100644 index 0000000..9d54a5e --- /dev/null +++ b/internal/client/output/s3.go @@ -0,0 +1,169 @@ +package output + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "strings" + + "github.com/aws/aws-sdk-go-v2/aws" + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/s3" + + "github.com/SayaAndy/saya-today-thumbnail-generator/config" + "github.com/SayaAndy/saya-today-thumbnail-generator/internal/client/input" +) + +var _ OutputClient = (*S3OutputClient)(nil) + +type S3OutputClient struct { + prefix string + bucketName string + s3cl *s3.Client +} + +func NewS3OutputClient(cfg *config.OutputConfig) (OutputClient, error) { + if cfg.Storage.Type != "s3" { + return nil, fmt.Errorf("invalid storage type for S3OutputClient") + } + s3cfg := cfg.Storage.Config.(*config.S3Config) + + s3cl, err := newS3Client(s3cfg) + if err != nil { + return nil, fmt.Errorf("create S3 client: %w", err) + } + + return &S3OutputClient{ + s3cl: s3cl, + bucketName: s3cfg.BucketName, + prefix: s3cfg.Prefix, + }, nil +} + +func (c *S3OutputClient) GetWriter(path string, inputMetadata *input.MetadataStruct, outputContentType string) (io.WriteCloser, error) { + key := c.prefix + path + + return &s3WriteCloser{ + key: key, + bucketName: c.bucketName, + s3cl: c.s3cl, + contentType: outputContentType, + hashOriginal: inputMetadata.Hash, + buf: &bytes.Buffer{}, + }, nil +} + +func (c *S3OutputClient) ReadMetadata(path string) (*MetadataStruct, error) { + key := c.prefix + path + + head, err := c.s3cl.HeadObject(context.Background(), &s3.HeadObjectInput{ + Bucket: aws.String(c.bucketName), + Key: aws.String(key), + }) + if err != nil { + return nil, fmt.Errorf("head S3 object %s: %w", key, err) + } + + metadata := MetadataStruct{ + Name: key, + StorageType: "s3", + Hash: strings.Trim(aws.ToString(head.ETag), "\""), + HashOriginal: head.Metadata["sha1-original"], + ContentType: aws.ToString(head.ContentType), + Misc: head.Metadata, + } + + if head.ContentLength != nil { + metadata.Size = *head.ContentLength + } + if head.LastModified != nil { + metadata.LastModified = *head.LastModified + metadata.FirstCreated = *head.LastModified + } + + return &metadata, nil +} + +func (c *S3OutputClient) IsMissing(path string) bool { + key := c.prefix + path + + head, err := c.s3cl.HeadObject(context.Background(), &s3.HeadObjectInput{ + Bucket: aws.String(c.bucketName), + Key: aws.String(key), + }) + if err != nil { + return true + } + + headJson, _ := json.Marshal(head) + slog.Debug("got object attrs", slog.String("path", path), slog.String("attrs", string(headJson))) + + if head.ContentLength == nil || *head.ContentLength == 0 { + return true + } + + return false +} + +// s3WriteCloser buffers writes and uploads to S3 on Close. +type s3WriteCloser struct { + key string + bucketName string + s3cl *s3.Client + contentType string + hashOriginal string + buf *bytes.Buffer +} + +func (w *s3WriteCloser) Write(p []byte) (int, error) { + return w.buf.Write(p) +} + +func (w *s3WriteCloser) Close() error { + _, err := w.s3cl.PutObject(context.Background(), &s3.PutObjectInput{ + Bucket: aws.String(w.bucketName), + Key: aws.String(w.key), + Body: bytes.NewReader(w.buf.Bytes()), + ContentType: aws.String(w.contentType), + Metadata: map[string]string{ + "sha1-original": w.hashOriginal, + }, + }) + if err != nil { + return fmt.Errorf("put S3 object %s: %w", w.key, err) + } + return nil +} + +func newS3Client(cfg *config.S3Config) (*s3.Client, error) { + opts := []func(*awsconfig.LoadOptions) error{ + awsconfig.WithRegion(cfg.Region), + } + + if cfg.AccessKeyID != "" && cfg.SecretAccessKey != "" { + opts = append(opts, awsconfig.WithCredentialsProvider( + credentials.NewStaticCredentialsProvider(cfg.AccessKeyID, cfg.SecretAccessKey, ""), + )) + } + + awsCfg, err := awsconfig.LoadDefaultConfig(context.Background(), opts...) + if err != nil { + return nil, fmt.Errorf("load AWS config: %w", err) + } + + var s3Opts []func(*s3.Options) + if cfg.Endpoint != "" { + s3Opts = append(s3Opts, func(o *s3.Options) { + o.BaseEndpoint = aws.String(cfg.Endpoint) + }) + } + s3Opts = append(s3Opts, func(o *s3.Options) { + o.UsePathStyle = cfg.UsePathStyle + }) + + return s3.NewFromConfig(awsCfg, s3Opts...), nil +} -- cgit v1.3.1+13