package depot import ( "context" "errors" "fmt" "log/slog" "time" "cloud.google.com/go/storage" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgxpool" ) const ( defaultUploadURLExpiry = 15 * time.Minute defaultDownloadURLExpiry = 24 * time.Hour ) type Service interface { PrepareUpload(ctx context.Context, input PrepareUploadInput) (*PrepareUploadResult, error) ConfirmUpload(ctx context.Context, objectID string) (*Object, error) GetByID(ctx context.Context, objectID string) (*Object, error) GetDownloadURL(ctx context.Context, objectID string) (string, error) Delete(ctx context.Context, objectID string) error Exists(ctx context.Context, objectID string) (bool, error) } type serviceImpl struct { repo repository storageClient *storage.Client bucketName string uploadURLExpiry time.Duration downloadURLExpiry time.Duration googleServiceAccountEmail string } func NewService(pool *pgxpool.Pool, storageClient *storage.Client, config Config) Service { uploadExpiry := config.UploadURLExpiry if uploadExpiry == 0 { uploadExpiry = defaultUploadURLExpiry } downloadExpiry := config.DownloadURLExpiry if downloadExpiry == 0 { downloadExpiry = defaultDownloadURLExpiry } if config.GoogleServiceAccountEmail == "" { slog.Error("GoogleServiceAccountEmail is not set in config. Signed URLs may not work if the storage client is not properly authenticated with a service account.") panic("GoogleServiceAccountEmail is required for signed URL generation") } return &serviceImpl{ repo: newRepository(pool), storageClient: storageClient, bucketName: config.BucketName, uploadURLExpiry: uploadExpiry, downloadURLExpiry: downloadExpiry, googleServiceAccountEmail: config.GoogleServiceAccountEmail, } } func (s *serviceImpl) PrepareUpload(ctx context.Context, input PrepareUploadInput) (*PrepareUploadResult, error) { if input.Name == "" { return nil, errors.Join(ErrInvalidInput, errors.New("name is required")) } if input.ContentType == "" { return nil, errors.Join(ErrInvalidInput, errors.New("content_type is required")) } if input.ContentLength <= 0 { return nil, errors.Join(ErrInvalidInput, errors.New("content_length must be positive")) } // Generate object key: {prefix}/{uuid}/{filename} objectKey := fmt.Sprintf("%s/%s/%s", input.Prefix, uuid.New().String(), input.Name) // Create the database record (contains_content = false initially) obj := &Object{ Name: input.Name, ContentType: input.ContentType, ContentLength: input.ContentLength, BucketName: s.bucketName, ObjectKey: objectKey, ContainsContent: false, } created, err := s.repo.create(ctx, obj) if err != nil { return nil, err } // Generate a signed URL for uploading with Content-Length enforcement // The Headers field specifies headers that MUST be included in the upload request contentLengthHeader := fmt.Sprintf("Content-Length:%d", input.ContentLength) uploadURL, err := s.storageClient.Bucket(s.bucketName).SignedURL(objectKey, &storage.SignedURLOptions{ GoogleAccessID: s.googleServiceAccountEmail, Method: "PUT", Expires: time.Now().Add(s.uploadURLExpiry), ContentType: input.ContentType, Headers: []string{contentLengthHeader}, }) if err != nil { slog.Error("failed to generate signed upload URL", "error", err, "bucket", s.bucketName, "object_key", objectKey) // Clean up the database record if we can't generate the URL if delErr := s.repo.delete(ctx, created.ID); delErr != nil { slog.Warn("failed to cleanup db record after signed URL failure", "error", delErr, "object_id", created.ID) } return nil, err } return &PrepareUploadResult{ ObjectID: created.ID, UploadURL: uploadURL, UploadHeaders: map[string]string{ "Content-Type": input.ContentType, "Content-Length": fmt.Sprintf("%d", input.ContentLength), }, }, nil } func (s *serviceImpl) ConfirmUpload(ctx context.Context, objectID string) (*Object, error) { obj, err := s.repo.getByID(ctx, objectID) if err != nil { if errors.Is(err, errNotFound) { return nil, ErrNotFound } return nil, err } // Verify the object exists in GCS and check its size matches expected attrs, err := s.storageClient.Bucket(obj.BucketName).Object(obj.ObjectKey).Attrs(ctx) if err != nil { if errors.Is(err, storage.ErrObjectNotExist) { return nil, errors.Join(ErrNotFound, errors.New("object not found in storage")) } slog.Error("failed to get GCS object attrs", "error", err, "bucket", obj.BucketName, "object_key", obj.ObjectKey) return nil, err } // Verify content length matches what was declared if attrs.Size != obj.ContentLength { return nil, errors.Join(ErrInvalidInput, fmt.Errorf("content length mismatch: expected %d, got %d", obj.ContentLength, attrs.Size)) } // Mark as containing content if err := s.repo.setContainsContent(ctx, objectID, true); err != nil { if errors.Is(err, errNotFound) { return nil, ErrNotFound } return nil, err } // Fetch and return the updated object return s.repo.getByID(ctx, objectID) } func (s *serviceImpl) GetByID(ctx context.Context, objectID string) (*Object, error) { obj, err := s.repo.getByID(ctx, objectID) if err != nil { if errors.Is(err, errNotFound) { return nil, ErrNotFound } return nil, err } return obj, nil } func (s *serviceImpl) GetDownloadURL(ctx context.Context, objectID string) (string, error) { obj, err := s.repo.getByID(ctx, objectID) if err != nil { if errors.Is(err, errNotFound) { return "", ErrNotFound } return "", err } // Generate a signed URL for downloading downloadURL, err := s.storageClient.Bucket(obj.BucketName).SignedURL(obj.ObjectKey, &storage.SignedURLOptions{ GoogleAccessID: s.googleServiceAccountEmail, Method: "GET", Expires: time.Now().Add(s.downloadURLExpiry), }) if err != nil { slog.Error("failed to generate signed download URL", "error", err, "bucket", obj.BucketName, "object_key", obj.ObjectKey) return "", err } return downloadURL, nil } func (s *serviceImpl) Delete(ctx context.Context, objectID string) error { obj, err := s.repo.getByID(ctx, objectID) if err != nil { if errors.Is(err, errNotFound) { return ErrNotFound } return err } // Delete from GCS (ignore not found errors) gcsErr := s.storageClient.Bucket(obj.BucketName).Object(obj.ObjectKey).Delete(ctx) if gcsErr != nil && !errors.Is(gcsErr, storage.ErrObjectNotExist) { slog.Error("failed to delete object from GCS", "error", gcsErr, "bucket", obj.BucketName, "object_key", obj.ObjectKey) return gcsErr } // Delete from database if err := s.repo.delete(ctx, objectID); err != nil { if errors.Is(err, errNotFound) { return ErrNotFound } return err } return nil } func (s *serviceImpl) Exists(ctx context.Context, objectID string) (bool, error) { return s.repo.exists(ctx, objectID) }