//go:build dev package s3 import ( "context" "fmt" "io" "log" "os" "sync" "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" ) var Client Uploader type Uploader interface { Upload(ctx context.Context, bucket, key string, body io.Reader, contentType string) error Download(ctx context.Context, bucket, key string, w io.Writer) error Delete(ctx context.Context, bucket, key string) error GetURL(ctx context.Context, bucket, key string) (string, error) HealthCheck(ctx context.Context) error } type S3Client struct { client *s3.Client bucket string publicURL string } // inMemS3 is an in-memory fallback for when RUSTFS is unavailable. // Stored data is lost on process exit — suitable for test isolation. type inMemS3 struct { mu sync.Mutex objects map[string][]byte } func (m *inMemS3) Upload(_ context.Context, bucket, key string, body io.Reader, _ string) error { m.mu.Lock() defer m.mu.Unlock() if m.objects == nil { m.objects = make(map[string][]byte) } data, err := io.ReadAll(body) if err != nil { return err } m.objects[bucket+"/"+key] = data return nil } func (m *inMemS3) Download(_ context.Context, bucket, key string, w io.Writer) error { m.mu.Lock() defer m.mu.Unlock() data, ok := m.objects[bucket+"/"+key] if !ok { return fmt.Errorf("object %s/%s not found", bucket, key) } _, err := w.Write(data) return err } func (m *inMemS3) Delete(_ context.Context, bucket, key string) error { m.mu.Lock() defer m.mu.Unlock() delete(m.objects, bucket+"/"+key) return nil } func (m *inMemS3) GetURL(_ context.Context, bucket, key string) (string, error) { return fmt.Sprintf("https://cdn.example.com/%s/%s", bucket, key), nil } func (m *inMemS3) HealthCheck(_ context.Context) error { return nil } func Connect() error { // Check for RUSTFS_* vars first (matching compose.yml), fall back to S3_* vars endpoint := os.Getenv("RUSTFS_ENDPOINT") if endpoint == "" { endpoint = os.Getenv("S3_ENDPOINT") } if endpoint == "" { // Default: localhost for bare metal dev, use rustfs:9000 for docker endpoint = "http://localhost:9000" } accessKey := os.Getenv("RUSTFS_ACCESS_KEY") if accessKey == "" { accessKey = os.Getenv("S3_ACCESS_KEY") } if accessKey == "" { accessKey = "minioadmin" } secretKey := os.Getenv("RUSTFS_SECRET_KEY") if secretKey == "" { secretKey = os.Getenv("S3_SECRET_KEY") } if secretKey == "" { secretKey = "minioadmin" } bucket := os.Getenv("RUSTFS_BUCKET") if bucket == "" { bucket = os.Getenv("S3_BUCKET") } if bucket == "" { bucket = "crussell" } profilePicsBucket := os.Getenv("S3_PROFILE_PICS_BUCKET") if profilePicsBucket == "" { profilePicsBucket = "crussell-profile-pics" } region := os.Getenv("AWS_REGION") if region == "" { region = "eu-west-2" } publicURL := os.Getenv("S3_PUBLIC_URL") if publicURL == "" { publicURL = endpoint } awsCfg, err := awsconfig.LoadDefaultConfig(context.Background(), awsconfig.WithRegion(region), awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider( accessKey, secretKey, "", )), ) if err != nil { log.Printf("S3: AWS config failed (%v) — falling back to in-memory S3", err) Client = &inMemS3{} return nil } // Attempt RUSTFS/S3 connection; fall back to in-memory on any failure. ctx := context.Background() s3Raw := s3.NewFromConfig(awsCfg, func(o *s3.Options) { o.BaseEndpoint = aws.String(endpoint) o.UsePathStyle = true }) // Verify connectivity with a HeadBucket call before committing. _, err = s3Raw.HeadBucket(ctx, &s3.HeadBucketInput{Bucket: aws.String(bucket)}) if err != nil { log.Printf("S3: RUSTFS not reachable at %s (%v) — falling back to in-memory S3", endpoint, err) Client = &inMemS3{} return nil } Client = &S3Client{ client: s3Raw, bucket: bucket, publicURL: publicURL, } // Create bucket if it doesn't exist _, err = s3Raw.CreateBucket(ctx, &s3.CreateBucketInput{ Bucket: aws.String(bucket), }) if err != nil { log.Printf("Bucket creation: %v (may already exist)", err) } // Set bucket policy for public read access policy := fmt.Sprintf(`{ "Version": "2012-10-17", "Statement": [{ "Sid": "PublicReadGetObject", "Effect": "Allow", "Principal": "*", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::%s/*" }] }`, bucket) _, err = s3Raw.PutBucketPolicy(ctx, &s3.PutBucketPolicyInput{ Bucket: aws.String(bucket), Policy: aws.String(policy), }) if err != nil { log.Printf("Bucket policy: %v (may already exist)", err) } // Create profile pics bucket if it doesn't exist if profilePicsBucket != bucket { _, err = s3Raw.CreateBucket(ctx, &s3.CreateBucketInput{ Bucket: aws.String(profilePicsBucket), }) if err != nil { log.Printf("Profile pics bucket creation: %v (may already exist)", err) } profilePolicy := fmt.Sprintf(`{ "Version": "2012-10-17", "Statement": [{ "Sid": "PublicReadGetObject", "Effect": "Allow", "Principal": "*", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::%s/*" }] }`, profilePicsBucket) _, err = s3Raw.PutBucketPolicy(ctx, &s3.PutBucketPolicyInput{ Bucket: aws.String(profilePicsBucket), Policy: aws.String(profilePolicy), }) if err != nil { log.Printf("Profile pics bucket policy: %v (may already exist)", err) } } log.Printf("Connected to local S3 (Rustfs): bucket=%s, endpoint=%s", bucket, endpoint) return nil } func (s *S3Client) Upload(ctx context.Context, bucket, key string, body io.Reader, contentType string) error { _, err := s.client.PutObject(ctx, &s3.PutObjectInput{ Bucket: aws.String(bucket), Key: aws.String(key), Body: body, ContentType: aws.String(contentType), }) return err } func (s *S3Client) Download(ctx context.Context, bucket, key string, w io.Writer) error { result, err := s.client.GetObject(ctx, &s3.GetObjectInput{ Bucket: aws.String(bucket), Key: aws.String(key), }) if err != nil { return err } defer result.Body.Close() _, err = io.Copy(w, result.Body) return err } func (s *S3Client) Delete(ctx context.Context, bucket, key string) error { _, err := s.client.DeleteObject(ctx, &s3.DeleteObjectInput{ Bucket: aws.String(bucket), Key: aws.String(key), }) return err } func (s *S3Client) GetURL(ctx context.Context, bucket, key string) (string, error) { return fmt.Sprintf("%s/%s/%s", s.publicURL, bucket, key), nil } func (s *S3Client) HealthCheck(ctx context.Context) error { _, err := s.client.HeadBucket(ctx, &s3.HeadBucketInput{ Bucket: aws.String(s.bucket), }) if err != nil { return fmt.Errorf("S3 bucket %q is not accessible (check S3_PUBLIC_URL / RUSTFS_ENDPOINT config): %w", s.bucket, err) } return nil }