llink/go/internal/depot/service.go

package depot

import (
	"context"
	"errors"
	"fmt"
	"io"
	"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)
	CreateFromReader(ctx context.Context, input CreateFromReaderInput, body io.Reader) (*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)
}

// CreateFromReader streams bytes directly to GCS using the storage client and
// records the depot_objects row in one shot. Unlike PrepareUpload, there is no
// signed URL or client round-trip — the caller already has the bytes. Intended
// for worker-side flows (e.g. transcoded media variants).
func (s *serviceImpl) CreateFromReader(ctx context.Context, input CreateFromReaderInput, body io.Reader) (*Object, 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"))
	}

	objectKey := fmt.Sprintf("%s/%s/%s", input.Prefix, uuid.New().String(), input.Name)

	w := s.storageClient.Bucket(s.bucketName).Object(objectKey).NewWriter(ctx)
	w.ContentType = input.ContentType
	if _, err := io.Copy(w, body); err != nil {
		// Close to release resources, then surface the original copy error.
		if cerr := w.Close(); cerr != nil {
			slog.Warn("failed to close GCS writer after copy failure", "error", cerr, "object_key", objectKey)
		}
		slog.Error("failed to stream object to GCS", "error", err, "bucket", s.bucketName, "object_key", objectKey)
		return nil, err
	}
	if err := w.Close(); err != nil {
		slog.Error("failed to close GCS writer", "error", err, "bucket", s.bucketName, "object_key", objectKey)
		return nil, err
	}

	obj := &Object{
		Name:            input.Name,
		ContentType:     input.ContentType,
		ContentLength:   w.Attrs().Size,
		BucketName:      s.bucketName,
		ObjectKey:       objectKey,
		ContainsContent: true,
	}

	created, err := s.repo.create(ctx, obj)
	if err != nil {
		// Best-effort: clean up the GCS object since we can't track it in the DB.
		if delErr := s.storageClient.Bucket(s.bucketName).Object(objectKey).Delete(ctx); delErr != nil {
			slog.Warn("failed to clean up GCS object after db create failure", "error", delErr, "object_key", objectKey)
		}
		return nil, err
	}

	return created, nil
}

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)
}