diff --git a/.hoplite/settings.json b/.hoplite/settings.json new file mode 100644 index 0000000..9e5af58 --- /dev/null +++ b/.hoplite/settings.json @@ -0,0 +1,14 @@ +{ + "version": 1, + "ports": {"preview": 3000, "additional": {}}, + "scripts": { + "setup": { + "enabled": true, + "command": "command -v git-lfs >/dev/null 2>&1 || (apt-get update -qq && apt-get install -y -qq git-lfs); mise use -g go@1.25.0 >/dev/null 2>&1; export PATH=\"$HOME/.local/share/mise/shims:$PATH\"; python3 -m venv .venv && .venv/bin/pip install -q --upgrade pip && .venv/bin/pip install -q 'psycopg[binary]' psycopg-pool redis Pillow google-cloud-storage boto3 python-magic pydantic structlog attrs pytest && go mod download" + }, + "run": {"enabled": true, "command": null}, + "archive": {"enabled": true, "command": null}, + "check": {"enabled": true, "command": null} + }, + "mcpServers": [] +} diff --git a/go.mod b/go.mod index 97919fb..50cc503 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.25.0 require ( cloud.google.com/go/storage v1.58.0 + github.com/DATA-DOG/go-sqlmock v1.5.2 github.com/aws/aws-sdk-go-v2 v1.42.0 github.com/aws/aws-sdk-go-v2/config v1.32.25 github.com/aws/aws-sdk-go-v2/credentials v1.19.24 diff --git a/go.sum b/go.sum index b24f082..88b24a4 100644 --- a/go.sum +++ b/go.sum @@ -24,6 +24,8 @@ filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA= filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4= github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 h1:L/gRVlceqvL25UVaW/CKtUDjefjrs0SPonmDGUVOYP0= github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= +github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU= +github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU= github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.30.0 h1:sBEjpZlNHzK1voKq9695PJSX2o5NEXl7/OL3coiIY0c= github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.30.0/go.mod h1:P4WPRUkOhJC13W//jWpyfJNDAIpvRbAUIYLX/4jtlE0= github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.54.0 h1:lhhYARPUu3LmHysQ/igznQphfzynnqI3D75oUyw1HXk= @@ -145,6 +147,7 @@ github.com/jmoiron/sqlx v1.4.0 h1:1PLqN7S1UYp5t4SrVVnt4nUVNemrDAtxlulVe+Qgm3o= github.com/jmoiron/sqlx v1.4.0/go.mod h1:ZrZ7UsYB/weZdl2Bxg6jCRO9c3YHl8r3ahlKmRT4JLY= github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= +github.com/kisielk/sqlstruct v0.0.0-20201105191214-5f3e10d3ab46/go.mod h1:yyMNCyc/Ib3bDTKd379tNMpB/7/H5TjM2Y9QJ5THLbE= github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= github.com/mattn/go-sqlite3 v1.14.22 h1:2gZY6PC6kBnID23Tichd1K+Z0oS6nE/XwU+Vz/5o4kU= diff --git a/internal/handler/asset_handler_test.go b/internal/handler/asset_handler_test.go new file mode 100644 index 0000000..3cdcca3 --- /dev/null +++ b/internal/handler/asset_handler_test.go @@ -0,0 +1,218 @@ +package handler + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/go-chi/chi/v5" + "github.com/google/uuid" + "github.com/rndmcodeguy20/mpiper/internal/config" + "github.com/rndmcodeguy20/mpiper/internal/models" + "github.com/rndmcodeguy20/mpiper/internal/service" + "go.uber.org/zap" +) + +const testMaxAssetSize = 1024 * 1024 // 1 MiB + +type stubService struct { + createFn func(ctx context.Context, req models.UploadAssetRequest) (*models.UploadAssetResponse, error) + markFn func(ctx context.Context, id uuid.UUID) error +} + +func (s *stubService) CreateAsset(ctx context.Context, req models.UploadAssetRequest) (*models.UploadAssetResponse, error) { + return s.createFn(ctx, req) +} + +func (s *stubService) MarkAssetUploaded(ctx context.Context, id uuid.UUID) error { + return s.markFn(ctx, id) +} + +func initHandlerConfig(t *testing.T) { + t.Helper() + config.Init(config.EnvConfig{MaxAssetSizeBytes: testMaxAssetSize}) +} + +func newTestHandler(svc service.AssetService) *AssetHandler { + return NewAssetHandler(svc, zap.NewNop(), nil) +} + +func doRequest(t *testing.T, h *AssetHandler, method, path, body string) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(method, path, strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + h.CreateAsset(w, req) + return w +} + +func TestCreateAssetRejectsMissingContentType(t *testing.T) { + initHandlerConfig(t) + h := newTestHandler(&stubService{}) + + w := doRequest(t, h, http.MethodPost, "/api/v1/storage/presign", `{"fileName":"a.jpg","size":10}`) + + if w.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400", w.Code) + } +} + +func TestCreateAssetRejectsUnsupportedContentType(t *testing.T) { + initHandlerConfig(t) + h := newTestHandler(&stubService{}) + + w := doRequest(t, h, http.MethodPost, "/api/v1/storage/presign", `{"fileName":"a.gif","contentType":"image/gif","size":10}`) + + if w.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400", w.Code) + } +} + +func TestCreateAssetRejectsOversizedAsset(t *testing.T) { + initHandlerConfig(t) + var called bool + h := newTestHandler(&stubService{createFn: func(ctx context.Context, req models.UploadAssetRequest) (*models.UploadAssetResponse, error) { + called = true + return nil, nil + }}) + + body := `{"fileName":"a.jpg","contentType":"image/jpeg","size":` + "2097152" + `}` + w := doRequest(t, h, http.MethodPost, "/api/v1/storage/presign", body) + + if w.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400", w.Code) + } + if called { + t.Error("service was called for an oversized asset; validation must short-circuit") + } +} + +func TestCreateAssetRejectsMalformedJSON(t *testing.T) { + initHandlerConfig(t) + h := newTestHandler(&stubService{}) + + w := doRequest(t, h, http.MethodPost, "/api/v1/storage/presign", `{"fileName":`) + + if w.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400", w.Code) + } +} + +func TestCreateAssetServiceErrorReturns500(t *testing.T) { + initHandlerConfig(t) + h := newTestHandler(&stubService{createFn: func(ctx context.Context, req models.UploadAssetRequest) (*models.UploadAssetResponse, error) { + return nil, context.DeadlineExceeded + }}) + + w := doRequest(t, h, http.MethodPost, "/api/v1/storage/presign", `{"fileName":"a.jpg","contentType":"image/jpeg","size":10}`) + + if w.Code != http.StatusInternalServerError { + t.Fatalf("status = %d, want 500", w.Code) + } +} + +func TestCreateAssetSuccess(t *testing.T) { + initHandlerConfig(t) + var got models.UploadAssetRequest + h := newTestHandler(&stubService{createFn: func(ctx context.Context, req models.UploadAssetRequest) (*models.UploadAssetResponse, error) { + got = req + return &models.UploadAssetResponse{ + UploadUrl: "https://presigned.example/put", + AssetID: "11111111-1111-1111-1111-111111111111", + Method: "PUT", + Headers: map[string]string{"Content-Type": "image/jpeg"}, + ObjectPath: "a.jpg", + PublicUrl: "https://cdn.example/a.jpg", + ExpiresAt: 300, + }, nil + }}) + + w := doRequest(t, h, http.MethodPost, "/api/v1/storage/presign", `{"fileName":"a.jpg","contentType":"image/jpeg","size":1234}`) + + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", w.Code, w.Body.String()) + } + if got.ContentType != "image/jpeg" || got.FileName != "a.jpg" || got.Size != 1234 { + t.Errorf("service received %+v, want the parsed request", got) + } + var resp struct { + Status string `json:"status"` + Data struct { + UploadURL string `json:"uploadUrl"` + ExpiresAt int64 `json:"expiresAt"` + ObjectPath string `json:"objectPath"` + } `json:"data"` + } + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("response is not JSON: %v", err) + } + if resp.Status != "success" || resp.Data.UploadURL == "" || resp.Data.ExpiresAt != 300 || resp.Data.ObjectPath != "a.jpg" { + t.Errorf("unexpected success payload: %+v", resp) + } +} + +func markUploadedRequest(t *testing.T, h *AssetHandler, assetID string) *httptest.ResponseRecorder { + t.Helper() + r := chi.NewRouter() + r.Get("/api/v1/assets/{assetID}/complete", h.MarkAssetUploaded) + req := httptest.NewRequest(http.MethodGet, "/api/v1/assets/"+assetID+"/complete", nil) + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + return w +} + +func TestMarkAssetUploadedRejectsMissingID(t *testing.T) { + initHandlerConfig(t) + h := newTestHandler(&stubService{}) + + w := markUploadedRequest(t, h, "") + + if w.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400", w.Code) + } +} + +func TestMarkAssetUploadedRejectsInvalidUUID(t *testing.T) { + initHandlerConfig(t) + h := newTestHandler(&stubService{}) + + w := markUploadedRequest(t, h, "not-a-uuid") + + if w.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want 400", w.Code) + } +} + +func TestMarkAssetUploadedServiceErrorReturns500(t *testing.T) { + initHandlerConfig(t) + h := newTestHandler(&stubService{markFn: func(ctx context.Context, id uuid.UUID) error { + return context.DeadlineExceeded + }}) + + w := markUploadedRequest(t, h, "11111111-1111-1111-1111-111111111111") + + if w.Code != http.StatusInternalServerError { + t.Fatalf("status = %d, want 500", w.Code) + } +} + +func TestMarkAssetUploadedSuccess(t *testing.T) { + initHandlerConfig(t) + var gotID uuid.UUID + h := newTestHandler(&stubService{markFn: func(ctx context.Context, id uuid.UUID) error { + gotID = id + return nil + }}) + + w := markUploadedRequest(t, h, "11111111-1111-1111-1111-111111111111") + + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", w.Code, w.Body.String()) + } + if gotID.String() != "11111111-1111-1111-1111-111111111111" { + t.Errorf("service received %s, want the parsed uuid", gotID) + } +} diff --git a/internal/queue/queue_test.go b/internal/queue/queue_test.go new file mode 100644 index 0000000..e26e2c5 --- /dev/null +++ b/internal/queue/queue_test.go @@ -0,0 +1,199 @@ +package queue + +import ( + "context" + "errors" + "strings" + "sync" + "testing" + "time" + + "github.com/redis/go-redis/v9" + appErrors "github.com/rndmcodeguy20/mpiper/pkg/errors" + "go.uber.org/zap" +) + +// failForHook intercepts every command and makes the first n of them fail, +// answering the rest successfully without contacting a real Redis server. +type failForHook struct { + mu sync.Mutex + failures int + lastErr error +} + +func (h *failForHook) DialHook(next redis.DialHook) redis.DialHook { return next } + +func (h *failForHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook { + return func(ctx context.Context, cmd redis.Cmder) error { + h.mu.Lock() + fail := h.failures > 0 + if fail { + h.failures-- + } + h.mu.Unlock() + if fail { + return h.lastErr + } + // Fake a successful XADD by filling in the reply value directly. + if sc, ok := cmd.(*redis.StringCmd); ok { + sc.SetVal("1-0") + } + return nil + } +} + +func (h *failForHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook { + return next +} + +func newScriptedQueue(t *testing.T, opts RedisQueueOptions) (*RedisQueue, *failForHook, *[]string) { + t.Helper() + hook := &failForHook{} + var names []string + var mu sync.Mutex + // Wrap hook process to also record commands and inject a ctx error on demand. + client := redis.NewClient(&redis.Options{Addr: "localhost:0", DialTimeout: 200 * time.Millisecond}) + t.Cleanup(func() { _ = client.Close() }) + client.AddHook(&recordingHook{hook: hook, names: &names, mu: &mu}) + + opts.QueueName = "media:jobs" + opts.MaxRetries = 2 + opts.RetryInterval = time.Millisecond + opts.ConnectionTimeOut = 500 * time.Millisecond + if opts.MaxStreamLength == 0 { + opts.MaxStreamLength = 10_000 + } + rq := NewRedisQueue(context.Background(), &RedisClient{client: client}, opts, zap.NewNop(), nil) + return rq, hook, &names +} + +// recordingHook decorates failForHook and records every processed command name. +type recordingHook struct { + hook *failForHook + mu *sync.Mutex + names *[]string +} + +func (h *recordingHook) DialHook(next redis.DialHook) redis.DialHook { return next } + +func (h *recordingHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook { + return func(ctx context.Context, cmd redis.Cmder) error { + h.mu.Lock() + *h.names = append(*h.names, cmd.Name()) + h.mu.Unlock() + return h.hook.ProcessHook(next)(ctx, cmd) + } +} + +func (h *recordingHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook { + return next +} + +func TestEnqueueNilRedisClientReturnsTypedError(t *testing.T) { + rq := NewRedisQueue(context.Background(), nil, RedisQueueOptions{QueueName: "media:jobs"}, zap.NewNop(), nil) + _, err := rq.Enqueue(context.Background(), map[string]interface{}{"job_id": int64(1)}) + var apiErr *appErrors.InternalServerErrorError + if !errors.As(err, &apiErr) { + t.Fatalf("Enqueue error = %v, want *InternalServerErrorError", err) + } + if apiErr.Code != "INTERNAL_SERVER_ERROR" { + t.Errorf("error code = %q, want INTERNAL_SERVER_ERROR", apiErr.Code) + } +} + +func TestEnqueueRetriesThenSucceeds(t *testing.T) { + rq, hook, names := newScriptedQueue(t, RedisQueueOptions{}) + hook.failures = 1 + hook.lastErr = errors.New("NOAUTH Authentication required.") + + id, err := rq.Enqueue(context.Background(), map[string]interface{}{"job_id": int64(7)}) + if err != nil { + t.Fatalf("Enqueue returned error: %v", err) + } + if id != "1-0" { + t.Errorf("id = %q, want 1-0", id) + } + count := 0 + for _, n := range *names { + if n == "xadd" { + count++ + } + } + if count != 2 { + t.Errorf("xadd attempts = %d, want 2 (one failure + one success)", count) + } +} + +func TestEnqueueExhaustsRetries(t *testing.T) { + rq, hook, _ := newScriptedQueue(t, RedisQueueOptions{}) + hook.failures = 99 + hook.lastErr = errors.New("NOPERM this user has no permissions to run the 'xadd' command") + + _, err := rq.Enqueue(context.Background(), map[string]interface{}{"job_id": int64(7)}) + if err == nil { + t.Fatal("expected error after exhausting retries") + } + var apiErr *appErrors.InternalServerErrorError + if !errors.As(err, &apiErr) { + t.Fatalf("error = %v, want *InternalServerErrorError", err) + } + if !strings.Contains(err.Error(), "after retries") { + t.Errorf("error message %q should mention retries exhaustion", err.Error()) + } +} + +func TestEnqueueStopsOnContextDeadline(t *testing.T) { + rq, hook, names := newScriptedQueue(t, RedisQueueOptions{}) + hook.failures = 99 + hook.lastErr = context.DeadlineExceeded + + ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond) + defer cancel() + time.Sleep(2 * time.Millisecond) + + _, err := rq.Enqueue(ctx, map[string]interface{}{"job_id": int64(7)}) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("error = %v, want context.DeadlineExceeded", err) + } + count := 0 + for _, n := range *names { + if n == "xadd" { + count++ + } + } + if count != 1 { + t.Errorf("xadd attempts = %d, want 1 (deadline must short-circuit retries)", count) + } +} + +func TestNewRedisQueueDefaults(t *testing.T) { + rq := NewRedisQueue(context.Background(), &RedisClient{}, RedisQueueOptions{}, zap.NewNop(), nil) + if rq.options.QueueName != "media:jobs" { + t.Errorf("QueueName = %q, want media:jobs", rq.options.QueueName) + } + if rq.options.RetryInterval != 2*time.Second { + t.Errorf("RetryInterval = %v, want 2s", rq.options.RetryInterval) + } + if rq.options.ConnectionTimeOut != 2*time.Second { + t.Errorf("ConnectionTimeOut = %v, want 2s", rq.options.ConnectionTimeOut) + } + if rq.options.MaxStreamLength != 10_000 { + t.Errorf("MaxStreamLength = %d, want 10000", rq.options.MaxStreamLength) + } + if rq.options.PoolSize != 10 { + t.Errorf("PoolSize = %d, want 10", rq.options.PoolSize) + } +} + +func TestNewRedisQueueNormalizesNegativeMaxRetries(t *testing.T) { + // A negative MaxRetries is treated as "unset" and normalized to 3, while + // an explicit 0 disables retrying. + rq := NewRedisQueue(context.Background(), &RedisClient{}, RedisQueueOptions{MaxRetries: -5}, zap.NewNop(), nil) + if rq.options.MaxRetries != 3 { + t.Errorf("MaxRetries = %d, want 3 after normalization", rq.options.MaxRetries) + } + rq = NewRedisQueue(context.Background(), &RedisClient{}, RedisQueueOptions{MaxRetries: 0}, zap.NewNop(), nil) + if rq.options.MaxRetries != 0 { + t.Errorf("MaxRetries = %d, want 0 (explicit zero preserves the value)", rq.options.MaxRetries) + } +} diff --git a/internal/repository/retry_test.go b/internal/repository/retry_test.go new file mode 100644 index 0000000..e3896e5 --- /dev/null +++ b/internal/repository/retry_test.go @@ -0,0 +1,105 @@ +package repository + +import ( + "context" + "errors" + "fmt" + "testing" +) + +func TestShouldRetryClassifiesTransientFailures(t *testing.T) { + transient := []error{ + errors.New(`ERROR: could not serialize access due to concurrent update (SQLSTATE 40001)`), + errors.New(`pq: deadlock detected`), + errors.New(`ERROR: deadlock detected (SQLSTATE 40P01)`), + errors.New("connection: lock wait timeout exceeded; try restarting transaction"), + errors.New("MySQL error 1213: Deadlock found when trying to get lock"), + errors.New("serialization error"), + errors.New("write conflict on table assets"), + errors.New("customer led us to believe it was 40001"), + fmt.Errorf("wrapped: %w", errors.New("could not serialize")), + } + for _, err := range transient { + if !shouldRetry(err) { + t.Errorf("shouldRetry(%q) = false, want true", err) + } + } +} + +func TestShouldRetryRejectsPermanentFailures(t *testing.T) { + permanent := []error{ + nil, + errors.New("permission denied for table assets"), + errors.New("connection refused"), + errors.New("unique_violation"), + errors.New("syntax error at or near \"INSERT\""), + errors.New(""), + } + for _, err := range permanent { + if shouldRetry(err) { + t.Errorf("shouldRetry(%q) = true, want false", err) + } + } + + // Cancellation and deadline errors must never be retried, even when the + // message happens to contain a transient marker. + neverRetry := []error{ + context.Canceled, + context.DeadlineExceeded, + fmt.Errorf("deadlock: %w", context.Canceled), + fmt.Errorf("serialization: %w", context.DeadlineExceeded), + } + for _, err := range neverRetry { + if shouldRetry(err) { + t.Errorf("shouldRetry(%q) = true, want false", err) + } + } +} + +func TestToAssetTypeFromMimeType(t *testing.T) { + cases := []struct { + mime string + want AssetType + }{ + {"image/jpeg", ImageAsset}, + {"image/png", ImageAsset}, + {"image/gif", ImageAsset}, // broad classifier, even though the upload gate rejects it + {"video/mp4", VideoAsset}, + {"video/quicktime", VideoAsset}, + {"audio/mpeg", AudioAsset}, + {"application/pdf", DocumentAsset}, + {"application/msword", DocumentAsset}, + {"application/vnd.openxmlformats-officedocument.wordprocessingml.document", DocumentAsset}, + {"application/octet-stream", OtherAsset}, + {"text/plain", OtherAsset}, + {"video", VideoAsset}, + {"image", ImageAsset}, + // Prefix-slicing guard: shorter than 5 chars must not panic. + {"", OtherAsset}, + {"tex", OtherAsset}, + } + for _, tc := range cases { + if got := ToAssetTypeFromMimeType(tc.mime); got != tc.want { + t.Errorf("ToAssetTypeFromMimeType(%q) = %v, want %v", tc.mime, got, tc.want) + } + } +} + +func TestToAssetType(t *testing.T) { + cases := []struct { + in string + want AssetType + }{ + {"image", ImageAsset}, + {"video", VideoAsset}, + {"audio", AudioAsset}, + {"document", DocumentAsset}, + {"", OtherAsset}, + {"Image", OtherAsset}, + } + for _, tc := range cases { + if got := ToAssetType(tc.in); got != tc.want { + t.Errorf("ToAssetType(%q) = %v, want %v", tc.in, got, tc.want) + } + } +} diff --git a/internal/service/asset.go b/internal/service/asset.go index 8e1d48a..90b07b4 100644 --- a/internal/service/asset.go +++ b/internal/service/asset.go @@ -27,12 +27,20 @@ type AssetService interface { MarkAssetUploaded(ctx context.Context, assetID uuid.UUID) error } +// objectStorage is the slice of storagex.StorageX the service needs; kept as a +// local interface so the upload flow can be tested against a fake. +type objectStorage interface { + GeneratePresignedURL(ctx context.Context, bucket, key string, options *storagex.PresignedURLOptions) (string, error) + PublicURL(ctx context.Context, bucket, key string) (string, error) + GetObjectAttrs(ctx context.Context, bucket, key string) (*storagex.ObjectAttrs, error) +} + type assetService struct { assetRepo repository.AssetRepository logger *zap.Logger - storageClient storagex.StorageX + storageClient objectStorage bucket string - queue *queue.RedisQueue + queue queue.Queue m *metrics.Metrics } diff --git a/internal/service/asset_test.go b/internal/service/asset_test.go new file mode 100644 index 0000000..9fae22e --- /dev/null +++ b/internal/service/asset_test.go @@ -0,0 +1,283 @@ +package service + +import ( + "context" + "database/sql" + "errors" + "testing" + "time" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/google/uuid" + "github.com/jmoiron/sqlx" + "github.com/rndmcodeguy20/mpiper/internal/models" + "github.com/rndmcodeguy20/mpiper/internal/repository" + "github.com/rndmcodeguy20/mpiper/pkg/utils/storagex" + "go.uber.org/zap" +) + +type fakeStorage struct { + presignedURL string + presignErr error + publicURL string + publicErr error + attrsErr error + + genBucket, genKey string + genOptions *storagex.PresignedURLOptions + pubBucket, pubKey string + attrsBucket, attrsKey string +} + +func (f *fakeStorage) GeneratePresignedURL(ctx context.Context, bucket, key string, options *storagex.PresignedURLOptions) (string, error) { + f.genBucket, f.genKey, f.genOptions = bucket, key, options + return f.presignedURL, f.presignErr +} + +func (f *fakeStorage) PublicURL(ctx context.Context, bucket, key string) (string, error) { + f.pubBucket, f.pubKey = bucket, key + return f.publicURL, f.publicErr +} + +func (f *fakeStorage) GetObjectAttrs(ctx context.Context, bucket, key string) (*storagex.ObjectAttrs, error) { + f.attrsBucket, f.attrsKey = bucket, key + if f.attrsErr != nil { + return nil, f.attrsErr + } + return &storagex.ObjectAttrs{Size: 42, ContentType: "", ETag: ""}, nil +} + +type fakeRepo struct { + db *sqlx.DB + + createdID uuid.UUID + createdURL string + createdSize int64 + createdType repository.AssetType + createErr error + + markChanged bool + markErr error + markID uuid.UUID + + jobID *int64 + jobErr error + jobAsset uuid.UUID +} + +func (f *fakeRepo) CreateAsset(ctx context.Context, id uuid.UUID, url string, size int64, fileType repository.AssetType, mimeType string) error { + f.createdID, f.createdURL, f.createdSize, f.createdType = id, url, size, fileType + return f.createErr +} + +func (f *fakeRepo) CreateAssetTx(ctx context.Context, tx *sql.Tx, id uuid.UUID, url string, size int64, fileType repository.AssetType, mimeType string) error { + return nil +} + +func (f *fakeRepo) MarkAssetUploadedTx(ctx context.Context, tx *sql.Tx, id uuid.UUID) (bool, error) { + f.markID = id + return f.markChanged, f.markErr +} + +func (f *fakeRepo) InsertProcessAssetJobTx(ctx context.Context, tx *sql.Tx, assetID uuid.UUID) (*int64, error) { + f.jobAsset = assetID + return f.jobID, f.jobErr +} + +func (f *fakeRepo) GetDB() *sqlx.DB { + return f.db +} + +type fakeQueue struct { + payloads []map[string]interface{} + err error +} + +func (f *fakeQueue) Enqueue(ctx context.Context, payload map[string]interface{}) (string, error) { + f.payloads = append(f.payloads, payload) + return "1-0", f.err +} + +func newTestService(storage *fakeStorage, repo *fakeRepo, q *fakeQueue) *assetService { + return &assetService{ + assetRepo: repo, + logger: zap.NewNop(), + storageClient: storage, + bucket: "mpiper", + queue: q, + m: nil, + } +} + +func TestCreateAssetSuccess(t *testing.T) { + storage := &fakeStorage{presignedURL: "https://signed.example/put", publicURL: "https://cdn.example/o"} + repo := &fakeRepo{} + svc := newTestService(storage, repo, &fakeQueue{}) + + res, err := svc.CreateAsset(context.Background(), models.UploadAssetRequest{ + FileName: "photo.jpg", + ContentType: "image/jpeg", + Size: 1234, + }) + if err != nil { + t.Fatalf("CreateAsset returned error: %v", err) + } + + if storage.genBucket != "mpiper" || storage.genKey != "media/raw/"+res.AssetID { + t.Errorf("presign called with bucket=%q key=%q, want mpiper/media/raw/", storage.genBucket, storage.genKey) + } + if storage.genOptions == nil || storage.genOptions.Method != "PUT" || + storage.genOptions.ExpiresInSeconds != 300 || storage.genOptions.ContentType != "image/jpeg" { + t.Errorf("presign options = %+v, want PUT / 300s / image/jpeg", storage.genOptions) + } + if storage.pubBucket != "mpiper" || storage.pubKey != "media/raw/"+res.AssetID { + t.Errorf("publicURL called with bucket=%q key=%q", storage.pubBucket, storage.pubKey) + } + if got := uuid.MustParse(res.AssetID); got == uuid.Nil { + t.Error("response must carry a fresh asset id") + } + if repo.createdSize != 1234 || repo.createdType != repository.ImageAsset || repo.createdURL != "https://cdn.example/o" || repo.createdID.String() != res.AssetID { + t.Errorf("repo insert = id:%s url:%q size:%d type:%s", repo.createdID, repo.createdURL, repo.createdSize, repo.createdType) + } + if res.Method != "PUT" || res.UploadUrl != "https://signed.example/put" || res.ExpiresAt != 300 || res.ObjectPath != "photo.jpg" { + t.Errorf("response = %+v, want PUT/300s and echo of fileName", res) + } +} + +func TestCreateAssetPresignErrorPropagates(t *testing.T) { + storage := &fakeStorage{presignErr: errors.New("no creds")} + repo := &fakeRepo{} + svc := newTestService(storage, repo, &fakeQueue{}) + + _, err := svc.CreateAsset(context.Background(), models.UploadAssetRequest{ContentType: "image/png", Size: 10}) + if err == nil || err.Error() != "no creds" { + t.Fatalf("error = %v, want the storage error", err) + } + if repo.createdSize != 0 { + t.Error("repo must not be reached when presigning fails") + } +} + +func TestCreateAssetPublicURLErrorPropagates(t *testing.T) { + storage := &fakeStorage{presignedURL: "https://signed", publicErr: errors.New("no public url")} + repo := &fakeRepo{} + svc := newTestService(storage, repo, &fakeQueue{}) + + _, err := svc.CreateAsset(context.Background(), models.UploadAssetRequest{ContentType: "image/png", Size: 10}) + if err == nil || err.Error() != "no public url" { + t.Fatalf("error = %v, want the public url error", err) + } + if repo.createdSize != 0 { + t.Error("repo must not be reached when public URL fails") + } +} + +func TestMarkAssetUploadedVerifiesObjectFirst(t *testing.T) { + storage := &fakeStorage{attrsErr: errors.New("object not found")} + repo := &fakeRepo{} + svc := newTestService(storage, repo, &fakeQueue{}) + + id := uuid.New() + err := svc.MarkAssetUploaded(context.Background(), id) + if err == nil || err.Error() != "object not found" { + t.Fatalf("error = %v, want the storage error", err) + } + if repo.markID != uuid.Nil { + t.Error("db must not be touched when the object is missing") + } +} + +func TestMarkAssetUploadedSuccessEnqueuesJob(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock: %v", err) + } + defer db.Close() + mock.ExpectBegin() + mock.ExpectCommit() + + jobID := int64(57) + repo := &fakeRepo{db: sqlx.NewDb(db, "sqlmock"), markChanged: true, jobID: &jobID} + q := &fakeQueue{} + svc := newTestService(&fakeStorage{}, repo, q) + + id := uuid.New() + if err := svc.MarkAssetUploaded(context.Background(), id); err != nil { + t.Fatalf("MarkAssetUploaded returned error: %v", err) + } + + if repo.markID != id || repo.jobAsset != id { + t.Errorf("repo ops used asset %s (mark=%s job=%s), want %s", id, repo.markID, repo.jobAsset, id) + } + if len(q.payloads) != 1 { + t.Fatalf("queue payloads = %d, want 1", len(q.payloads)) + } + p := q.payloads[0] + if p["event"] != "asset_uploaded" || p["asset_id"] != id.String() { + t.Errorf("payload = %+v, want event=asset_uploaded asset_id=%s", p, id) + } + if jid, ok := p["job_id"].(int64); !ok || jid != 57 { + t.Errorf("payload job_id = %v (%T), want int64(57)", p["job_id"], p["job_id"]) + } + if ts, ok := p["timestamp"].(string); !ok { + t.Errorf("payload timestamp = %v, want RFC3339 string", p["timestamp"]) + } else if _, err := time.Parse(time.RFC3339, ts); err != nil { + t.Errorf("timestamp %q is not RFC3339: %v", ts, err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Errorf("unmet sql expectations: %v", err) + } +} + +func TestMarkAssetUploadedAlreadyUploadedSkipsEnqueue(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock: %v", err) + } + defer db.Close() + mock.ExpectBegin() + mock.ExpectRollback() + + repo := &fakeRepo{db: sqlx.NewDb(db, "sqlmock"), markChanged: false} + q := &fakeQueue{} + svc := newTestService(&fakeStorage{}, repo, q) + + id := uuid.New() + if err := svc.MarkAssetUploaded(context.Background(), id); err != nil { + t.Fatalf("MarkAssetUploaded returned error: %v", err) + } + if len(q.payloads) != 0 { + t.Errorf("queue payloads = %d, want 0 for an already-uploaded asset", len(q.payloads)) + } + if repo.jobAsset != uuid.Nil { + t.Error("no job must be created for an already-uploaded asset") + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Errorf("unmet sql expectations: %v", err) + } +} + +func TestMarkAssetUploadedDBFailureRollsBack(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock: %v", err) + } + defer db.Close() + mock.ExpectBegin() + mock.ExpectRollback() + + repo := &fakeRepo{db: sqlx.NewDb(db, "sqlmock"), markErr: errors.New("update failed")} + q := &fakeQueue{} + svc := newTestService(&fakeStorage{}, repo, q) + + err = svc.MarkAssetUploaded(context.Background(), uuid.New()) + if err == nil { + t.Fatal("expected the db error to propagate") + } + if len(q.payloads) != 0 { + t.Errorf("queue payloads = %d, want 0 when the tx fails", len(q.payloads)) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Errorf("unmet sql expectations: %v", err) + } +} diff --git a/worker/tests/test_consumer_lifecycle.py b/worker/tests/test_consumer_lifecycle.py new file mode 100644 index 0000000..c4ce9b2 --- /dev/null +++ b/worker/tests/test_consumer_lifecycle.py @@ -0,0 +1,75 @@ +import unittest +from unittest.mock import MagicMock, patch + +from worker.consumer.consumer import Consumer +from worker.processing.processor import process_asset_dispatch + + +def _make_consumer(job_row): + """Consumer whose PG cursor returns `job_row` for the claim SELECT. + + The single mocked cursor is reused for both database blocks (claim + + finalize), which is enough to assert on the SQL text of each transition. + """ + cfg = MagicMock() + cfg.stream_name = "media:jobs" + cfg.consumer_group = "media-workers" + cfg.redis.max_retries = 3 + with patch("worker.consumer.consumer.redis.Redis.from_url") as from_url: + client = MagicMock() + from_url.return_value = client + consumer = Consumer( + pg_pool=MagicMock(), redis_url="redis://x", storage=MagicMock(), cfg=cfg + ) + cursor = MagicMock() + cursor.fetchone.side_effect = [job_row] + conn = MagicMock() + conn.cursor.return_value = cursor + consumer.pg.get_pg_conn.return_value.__enter__.return_value = conn + return consumer, client, cursor + + +def _executed_sql(cursor): + return " | ".join(c.args[0] for c in cursor.execute.call_args_list if c.args) + + +class TestJobLifecycle(unittest.TestCase): + """Success, idempotency and unknown-job handling for _handle_job (DEV-52).""" + + @patch("worker.consumer.consumer.process_asset_dispatch") + def test_success_marks_job_done_and_asset_ready_and_acks(self, mock_dispatch): + mock_dispatch.return_value = None + consumer, client, cursor = _make_consumer((42, "asset-1", "pending", 0)) + + consumer._handle_job(42, "1-0") + + sql = _executed_sql(cursor) + self.assertIn("UPDATE jobs SET status = 'in_progress'", sql) + self.assertIn("UPDATE jobs SET status = 'done'", sql) + self.assertIn("UPDATE assets SET status = 'ready'", sql) + client.xack.assert_called_once_with("media:jobs", "media-workers", "1-0") + mock_dispatch.assert_called_once() + + @patch("worker.consumer.consumer.process_asset_dispatch") + def test_already_done_job_is_acked_without_processing(self, mock_dispatch): + consumer, client, cursor = _make_consumer((42, "asset-1", "done", 2)) + + consumer._handle_job(42, "1-0") + + client.xack.assert_called_once_with("media:jobs", "media-workers", "1-0") + mock_dispatch.assert_not_called() + # No state transitions at all beyond the claim SELECT. + self.assertEqual(len(cursor.execute.call_args_list), 1) + + @patch("worker.consumer.consumer.process_asset_dispatch") + def test_unknown_job_is_acked_without_processing(self, mock_dispatch): + consumer, client, _ = _make_consumer(None) + + consumer._handle_job(999, "1-0") + + client.xack.assert_called_once_with("media:jobs", "media-workers", "1-0") + mock_dispatch.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/worker/tests/test_consumer_message.py b/worker/tests/test_consumer_message.py new file mode 100644 index 0000000..5afa84c --- /dev/null +++ b/worker/tests/test_consumer_message.py @@ -0,0 +1,77 @@ +import json +import unittest +from unittest.mock import MagicMock, patch + +from worker.consumer.consumer import Consumer + + +def _make_consumer(message_fields): + """Consumer with redis mocked; returns the redis client mock too.""" + cfg = MagicMock() + cfg.stream_name = "media:jobs" + cfg.consumer_group = "media-workers" + cfg.redis.max_retries = 3 + with patch("worker.consumer.consumer.redis.Redis.from_url") as from_url: + client = MagicMock() + from_url.return_value = client + consumer = Consumer( + pg_pool=MagicMock(), redis_url="redis://x", storage=MagicMock(), cfg=cfg + ) + # Message routing is the subject here; isolate the recovery cadence. + consumer._maybe_recover = MagicMock() + if message_fields is not None: + client.xreadgroup.return_value = [("media:jobs", [("1-0", message_fields)])] + else: + client.xreadgroup.return_value = [] + return consumer, client + + +class TestConsumeRouting(unittest.TestCase): + """consume() must unwrap the producer's JSON body and route correctly.""" + + def test_unwraps_body_json_and_routes_by_job_id(self): + body = json.dumps( + {"job_id": 42, "asset_id": "asset-1", "event": "asset_uploaded"} + ) + consumer, client = _make_consumer({"body": body}) + consumer._handle_job = MagicMock() + consumer._handle_asset_message = MagicMock() + + result = consumer.consume("worker-1") + + self.assertTrue(result) + # The Go API publishes the payload under a JSON `body` field; the + # consumer must route on the unwrapped job_id. + consumer._handle_job.assert_called_once_with(42, "1-0") + consumer._handle_asset_message.assert_not_called() + + def test_routes_asset_id_message(self): + consumer, client = _make_consumer({"asset_id": "asset-1"}) + consumer._handle_asset_message = MagicMock() + consumer._handle_job = MagicMock() + + consumer.consume("worker-1") + + consumer._handle_asset_message.assert_called_once_with("asset-1", "1-0") + consumer._handle_job.assert_not_called() + + def test_acks_malformed_message_without_handlers(self): + consumer, client = _make_consumer({"unknown": "field"}) + consumer._handle_job = MagicMock() + consumer._handle_asset_message = MagicMock() + + result = consumer.consume("worker-1") + + self.assertTrue(result) + client.xack.assert_called_once_with("media:jobs", "media-workers", "1-0") + consumer._handle_job.assert_not_called() + consumer._handle_asset_message.assert_not_called() + + def test_returns_false_when_no_message(self): + consumer, client = _make_consumer(None) + + self.assertFalse(consumer.consume("worker-1")) + + +if __name__ == "__main__": + unittest.main() diff --git a/worker/tests/test_dedup.py b/worker/tests/test_dedup.py new file mode 100644 index 0000000..302ed11 --- /dev/null +++ b/worker/tests/test_dedup.py @@ -0,0 +1,81 @@ +import unittest +from unittest.mock import MagicMock, patch + +from worker.processing.processor import ( + DedupResult, + check_for_duplicate, +) + + +class TestCheckForDuplicate(unittest.TestCase): + """Branch coverage of the dedup gate: four distinct outcomes.""" + + def _pg_pool(self, canonical_row): + cursor = MagicMock() + cursor.fetchone.return_value = canonical_row + conn = MagicMock() + conn.cursor.return_value = cursor + pg_pool = MagicMock() + pg_pool.get_pg_conn.return_value.__enter__.return_value = conn + return pg_pool, cursor, conn + + def test_no_existing_asset_returns_no_duplicate(self): + pg_pool, cursor, conn = self._pg_pool(None) + + result = check_for_duplicate("hash-x", "new-asset", pg_pool) + + self.assertEqual(result, DedupResult.NO_DUPLICATE) + cursor.execute.assert_called_once() + + def test_pending_canonical_returns_duplicate_pending(self): + pg_pool, cursor, conn = self._pg_pool(("canonical-1", "image", "processing")) + + result = check_for_duplicate("hash-x", "new-asset", pg_pool) + + self.assertEqual(result, DedupResult.DUPLICATE_PENDING) + # Only the SELECT ran; no mutation, no clone, no commit. + self.assertEqual(len(cursor.execute.call_args_list), 1) + conn.commit.assert_not_called() + + @patch("worker.processing.processor.clone_image_variants", return_value=3) + def test_ready_canonical_clones_variants_and_marks_new_asset_ready( + self, mock_clone + ): + pg_pool, cursor, conn = self._pg_pool(("canonical-1", "image", "ready")) + + result = check_for_duplicate("hash-x", "new-asset", pg_pool) + + self.assertEqual(result, DedupResult.DUPLICATE_READY) + mock_clone.assert_called_once_with(cursor, "canonical-1", "new-asset") + conn.commit.assert_called_once() + + updates = [c.args[0] for c in cursor.execute.call_args_list] + ready_update = next(u for u in updates if "canonical_asset_id" in u) + self.assertIn("status", ready_update) + # The new asset is marked ready and linked to the canonical asset. + params = cursor.execute.call_args.args[1] + self.assertEqual(params, ("ready", "canonical-1", "new-asset")) + + @patch("worker.processing.processor.clone_image_variants", return_value=0) + def test_canonical_without_variants_fails_new_asset(self, mock_clone): + pg_pool, cursor, conn = self._pg_pool(("canonical-1", "image", "ready")) + + result = check_for_duplicate("hash-x", "new-asset", pg_pool) + + self.assertEqual(result, DedupResult.DUPLICATE_READY) + conn.commit.assert_called_once() + updates = [c.args[0] for c in cursor.execute.call_args_list] + failed_update = next(u for u in updates if "error_reason" in u) + self.assertIn("status", failed_update) + params = cursor.execute.call_args.args[1] + self.assertEqual(params[0], "failed") + + def test_unknown_canonical_type_raises(self): + pg_pool, _, _ = self._pg_pool(("canonical-1", "archive", "ready")) + + with self.assertRaises(ValueError): + check_for_duplicate("hash-x", "new-asset", pg_pool) + + +if __name__ == "__main__": + unittest.main() diff --git a/worker/tests/test_image_pipeline.py b/worker/tests/test_image_pipeline.py index 78edad8..9adfe3e 100644 --- a/worker/tests/test_image_pipeline.py +++ b/worker/tests/test_image_pipeline.py @@ -10,6 +10,9 @@ class StorageMock: def __init__(self): self.calls = [] + def public_url(self, key): + return f"https://mock/{key}" + def upload_bytes(self, key, data, content_type=None): self.calls.append((key, len(data), content_type)) # return fake URL @@ -64,6 +67,7 @@ def test_process_image_file(self, mock_image_open): process_image_file( asset_id="test-123", local_raw_path="dummy.jpg", + content_hash="testhash", pg_pool=mock_pg_pool, storage=storage, cfg=cfg, @@ -79,19 +83,24 @@ def test_process_image_file(self, mock_image_open): # 2. DB UPDATE for metadata was called self.assertTrue(mock_cursor.execute.called) - # 3. Storage upload was called for all 3 variants + # 3. Storage upload was called for all 3 variants. Keys are now + # dedup-scoped by content hash: media/processed//.. self.assertEqual(len(storage.calls), 3) for key, size, content_type in storage.calls: - self.assertIn("media/processed/test-123/img/", key) + self.assertTrue(key.startswith("media/processed/testhash/"), key) + self.assertTrue(key.endswith(".webp"), key) self.assertTrue(size > 0) self.assertIn("image/", content_type) - # 4. Correct roles - uploaded_roles = [ - call[0].split("/")[-1].split(".")[0] for call in storage.calls + # 4. Correct roles, sourced from the variants.image INSERT params since + # roles no longer appear in the storage key. + roles = [ + call.args[1][2] + for call in mock_cursor.execute.call_args_list + if "INSERT INTO variants.image" in call.args[0] ] self.assertCountEqual( - uploaded_roles, ["thumbnail", "display_small", "display_large"] + roles, ["thumbnail", "display_small", "display_large"] )