Storage Module (recerdo-storage)¶

// MediaFile エンティティ
type MediaFile struct {
    ID               string    // ULID
    OrgID            string
    UploaderID       string
    OriginalFilename string
    MimeType         string
    FileSizeBytes    int64
    StorageKey       string    // オブジェクトストレージキー for original
    Status           string    // UPLOADING / PROCESSING / READY / FAILED
    AccessPolicy     string    // PRIVATE / ORG_MEMBERS / EVENT_MEMBERS
    CreatedAt        time.Time
    UpdatedAt        time.Time
    DeletedAt        *time.Time
}

func NewMediaFile(orgID, uploaderID, filename string, mimeType MimeType, sizeBytes int64) (*MediaFile, error) {
    if err := mimeType.Validate(); err != nil {
        return nil, ErrInvalidMimeType
    }
    if sizeBytes == 0 {
        return nil, ErrEmptyFile
    }
    if sizeBytes > mimeType.MaxBytes() {
        return nil, ErrFileTooLarge
    }

    return &MediaFile{
        ID:               ulid.Make().String(),
        OrgID:            orgID,
        UploaderID:       uploaderID,
        OriginalFilename: filename,
        MimeType:         mimeType.String(),
        FileSizeBytes:    sizeBytes,
        Status:           "UPLOADING",
        AccessPolicy:     "PRIVATE",
        CreatedAt:        time.Now(),
        UpdatedAt:        time.Now(),
    }, nil
}

func (m *MediaFile) TransitionTo(newStatus string) error {
    validTransitions := map[string][]string{
        "UPLOADING":  {"PROCESSING", "FAILED"},
        "PROCESSING": {"READY", "FAILED"},
        "READY":      {},
        "FAILED":     {},
    }

    allowed := validTransitions[m.Status]
    if !contains(allowed, newStatus) {
        return ErrInvalidStatusTransition
    }

    m.Status = newStatus
    m.UpdatedAt = time.Now()
    return nil
}

func (m *MediaFile) CanBeAccessedBy(userID string, orgMembers []string, eventMembers []string) bool {
    if m.Status == "FAILED" {
        return false
    }

    switch m.AccessPolicy {
    case "PRIVATE":
        return m.UploaderID == userID
    case "ORG_MEMBERS":
        return contains(orgMembers, userID)
    case "EVENT_MEMBERS":
        return contains(eventMembers, userID)
    default:
        return false
    }
}

func (m *MediaFile) CanBeDeletedBy(userID string) bool {
    // アップロード者またはOrg Adminのみ削除可能
    return m.UploaderID == userID
}

// MediaChunk エンティティ
type MediaChunk struct {
    ID         string
    MediaID    string
    ChunkIndex int
    StorageKey string    // オブジェクトストレージ（Garage / OCI Object Storage）のキー
    SizeBytes  int64
    UploadedAt time.Time
    ExpiresAt  time.Time
}

func NewMediaChunk(mediaID string, chunkIndex int, sizeBytes int64) *MediaChunk {
    return &MediaChunk{
        ID:         ulid.Make().String(),
        MediaID:    mediaID,
        ChunkIndex: chunkIndex,
        SizeBytes:  sizeBytes,
        UploadedAt: time.Now(),
        ExpiresAt:  time.Now().Add(24 * time.Hour),
    }
}

func (c *MediaChunk) IsExpired() bool {
    return time.Now().After(c.ExpiresAt)
}

// ProcessingJob エンティティ
type ProcessingJob struct {
    ID                 string
    MediaID            string
    JobType            string    // HLS_TRANSCODE / HEIC_CONVERT / THUMBNAIL_GEN / LIVE_PHOTO_PAIRING / HIGHLIGHT_CONCAT
    Status             string    // PENDING / RUNNING / DONE / FAILED
    StartedAt          *time.Time
    CompletedAt        *time.Time
    ErrorMsg           *string
    ResultStorageKey   *string
    CreatedAt          time.Time
}

func NewProcessingJob(mediaID, jobType string) *ProcessingJob {
    return &ProcessingJob{
        ID:        ulid.Make().String(),
        MediaID:   mediaID,
        JobType:   jobType,
        Status:    "PENDING",
        CreatedAt: time.Now(),
    }
}

func (p *ProcessingJob) StartProcessing() {
    now := time.Now()
    p.StartedAt = &now
    p.Status = "RUNNING"
}

func (p *ProcessingJob) Complete(resultStorageKey string) {
    now := time.Now()
    p.CompletedAt = &now
    p.Status = "DONE"
    p.ResultStorageKey = &resultStorageKey
}

func (p *ProcessingJob) Failed(errMsg string) {
    now := time.Now()
    p.CompletedAt = &now
    p.Status = "FAILED"
    p.ErrorMsg = &errMsg
}

// Repository Ports
type MediaRepository interface {
    Save(ctx context.Context, media *MediaFile) error
    GetByID(ctx context.Context, orgID, mediaID string) (*MediaFile, error)
    GetByIDAndOrgID(ctx context.Context, orgID, mediaID string) (*MediaFile, error)
    Update(ctx context.Context, media *MediaFile) error
    Delete(ctx context.Context, mediaID string) error
    ListByOrg(ctx context.Context, orgID string, limit, offset int) ([]*MediaFile, int, error)
}

type MediaChunkRepository interface {
    Save(ctx context.Context, chunk *MediaChunk) error
    GetByMediaID(ctx context.Context, mediaID string) ([]*MediaChunk, error)
    GetByMediaIDAndIndex(ctx context.Context, mediaID string, index int) (*MediaChunk, error)
    DeleteByMediaID(ctx context.Context, mediaID string) error
    DeleteExpired(ctx context.Context) (int, error)
}

type ProcessingJobRepository interface {
    Save(ctx context.Context, job *ProcessingJob) error
    GetByID(ctx context.Context, jobID string) (*ProcessingJob, error)
    GetByMediaID(ctx context.Context, mediaID string) ([]*ProcessingJob, error)
    Update(ctx context.Context, job *ProcessingJob) error
    ListPending(ctx context.Context, limit int) ([]*ProcessingJob, error)
}

// Service Ports
type StoragePort interface {
    // オブジェクトストレージ操作（Garage / OCI Object Storage 両対応、S3 互換 API）
    UploadObject(ctx context.Context, key string, content []byte, contentType string) error
    MergeChunks(ctx context.Context, targetKey string, chunkKeys []string) error  // CompleteMultipartUpload
    GeneratePresignedURL(ctx context.Context, key string, ttl time.Duration) (string, error)
    DeleteObject(ctx context.Context, key string) error
    ObjectExists(ctx context.Context, key string) (bool, error)
}

// MediaTranscoderPort - FFmpeg HLS / libheif HEIC 変換・Live Photo ペアリングの抽象化
type MediaTranscoderPort interface {
    TranscodeToHLS(ctx context.Context, sourceKey string, targetPrefix string, profiles []HLSProfile) (manifestKey string, err error)
    ConvertHEIC(ctx context.Context, sourceKey string, outputFormat HEICTargetFormat) (optimizedKey string, err error)
    ExtractLivePhotoIdentifier(ctx context.Context, sourceKey string) (identifier string, err error)
    GenerateThumbnail(ctx context.Context, sourceKey string, maxEdgePx int) (thumbKey string, err error)
    ConcatenateVideos(ctx context.Context, sourceKeys []string, outputKey string) error  // ハイライト動画用
}

type QueuePort interface {
    // キュー抽象化（Beta: Redis+BullMQ/asynq、本番: OCI Queue）
    Publish(ctx context.Context, topic string, message []byte) error
    EnqueueJob(ctx context.Context, queueName string, job *ProcessingJob) error
    GetJobStatus(ctx context.Context, jobID string) (string, error)
}

type PermissionPort interface {
    // Permission Service gRPC 委譲
    CheckOrgMembership(ctx context.Context, userID, orgID string) (bool, error)
    CheckEventMembership(ctx context.Context, userID, eventID string) (bool, error)
}

type CachePort interface {
    // Redis キャッシュ操作
    Get(ctx context.Context, key string) (string, error)
    Set(ctx context.Context, key, value string, ttl time.Duration) error
    Delete(ctx context.Context, key string) error
}

type EventPublisherPort interface {
    // QueuePort 経由のドメインイベント発行（Beta: Redis+BullMQ、本番: OCI Queue）
    Publish(ctx context.Context, event DomainEvent) error
}

// Use Case Interfaces
type UploadMediaSingleUseCase interface {
    Execute(ctx context.Context, input UploadMediaSingleInput) (*UploadMediaSingleOutput, error)
}

type DeliverMediaUseCase interface {
    Execute(ctx context.Context, input DeliverMediaInput) (*DeliverMediaOutput, error)
}

type GetMediaMetadataUseCase interface {
    Execute(ctx context.Context, input GetMediaMetadataInput) (*GetMediaMetadataOutput, error)
}

type DeleteMediaUseCase interface {
    Execute(ctx context.Context, input DeleteMediaInput) (*DeleteMediaOutput, error)
}

fx.Provide(
    // Repositories
    NewMySQLMediaRepository,            // → MediaRepository
    NewMySQLMediaChunkRepository,       // → MediaChunkRepository
    NewMySQLProcessingJobRepository,    // → ProcessingJobRepository
    NewRedisCacheAdapter,                    // → CachePort

    // External Service Adapters（Feature Flag で Beta / 本番切替）
    NewGarageStorageAdapter,                 // → StoragePort (Beta: Garage)
    NewOCIObjectStorageAdapter,              // → StoragePort (本番: OCI Object Storage)
    NewFFmpegHLSAdapter,                     // → MediaTranscoderPort（HLS/concat）
    NewLibheifAdapter,                       // → MediaTranscoderPort（HEIC 変換・Live Photo 識別子）
    NewRedisBullMQAdapter,                   // → QueuePort (Beta)
    NewOCIQueueAdapter,                      // → QueuePort (本番)
    NewPermissionServiceGRPCAdapter,         // → PermissionPort
    NewQueueEventPublisher,                  // → EventPublisherPort (QueuePort ラッパ)

    // Use Cases
    NewUploadMediaSingleUseCase,
    NewInitiateChunkedUploadUseCase,
    NewUploadMediaChunkUseCase,
    NewMergeChunkedUploadUseCase,
    NewDeliverMediaUseCase,
    NewGetMediaMetadataUseCase,
    NewDeleteMediaUseCase,
    NewRetryProcessingUseCase,
    NewListOrgMediaUseCase,
    NewCleanupExpiredChunksUseCase,

    // Controllers / Handlers
    NewHTTPMediaHandler,
    NewHTTPChunkedUploadHandler,
    NewHealthHandler,
    NewMetricsHandler,
    NewAsyncJobWorker,
    NewBatchCleanupWorker,
)

// Entity Test
func TestMediaFile_NewMediaFile_Success(t *testing.T) {
    mimeType := MimeTypeJPEG
    media, err := NewMediaFile("org-1", "user-1", "photo.jpg", mimeType, 1000000)

    require.NoError(t, err)
    assert.Equal(t, "org-1", media.OrgID)
    assert.Equal(t, "user-1", media.UploaderID)
    assert.Equal(t, "photo.jpg", media.OriginalFilename)
    assert.Equal(t, "UPLOADING", media.Status)
    assert.Equal(t, "PRIVATE", media.AccessPolicy)
}

func TestMediaFile_NewMediaFile_InvalidMimeType(t *testing.T) {
    _, err := NewMediaFile("org-1", "user-1", "file.exe", MimeType("application/octet-stream"), 1000000)
    assert.ErrorIs(t, err, ErrInvalidMimeType)
}

func TestMediaFile_NewMediaFile_FileTooLarge(t *testing.T) {
    // 100MBを超える画像
    _, err := NewMediaFile("org-1", "user-1", "huge.jpg", MimeTypeJPEG, 101*1024*1024)
    assert.ErrorIs(t, err, ErrFileTooLarge)
}

func TestMediaFile_TransitionTo_ValidTransition(t *testing.T) {
    media := &MediaFile{Status: "UPLOADING"}
    err := media.TransitionTo("PROCESSING")

    require.NoError(t, err)
    assert.Equal(t, "PROCESSING", media.Status)
}

func TestMediaFile_TransitionTo_InvalidTransition(t *testing.T) {
    media := &MediaFile{Status: "READY"}
    err := media.TransitionTo("UPLOADING")

    assert.ErrorIs(t, err, ErrInvalidStatusTransition)
}

func TestMediaFile_CanBeAccessedBy_PRIVATE(t *testing.T) {
    media := &MediaFile{UploaderID: "user-1", Status: "READY", AccessPolicy: "PRIVATE"}

    assert.True(t, media.CanBeAccessedBy("user-1", []string{}, []string{}))
    assert.False(t, media.CanBeAccessedBy("user-2", []string{}, []string{}))
}

func TestMediaFile_CanBeAccessedBy_ORG_MEMBERS(t *testing.T) {
    media := &MediaFile{Status: "READY", AccessPolicy: "ORG_MEMBERS"}

    assert.True(t, media.CanBeAccessedBy("user-1", []string{"user-1", "user-2"}, []string{}))
    assert.False(t, media.CanBeAccessedBy("user-3", []string{"user-1", "user-2"}, []string{}))
}

func TestMediaFile_CanBeAccessedBy_FAILED_Status(t *testing.T) {
    media := &MediaFile{Status: "FAILED", AccessPolicy: "PRIVATE", UploaderID: "user-1"}

    assert.False(t, media.CanBeAccessedBy("user-1", []string{}, []string{}))
}

func TestMediaFile_CanBeDeletedBy_UploaderOnly(t *testing.T) {
    media := &MediaFile{UploaderID: "user-1"}

    assert.True(t, media.CanBeDeletedBy("user-1"))
    assert.False(t, media.CanBeDeletedBy("user-2"))
}

func TestMediaChunk_IsExpired(t *testing.T) {
    chunk := &MediaChunk{ExpiresAt: time.Now().Add(-1 * time.Hour)}
    assert.True(t, chunk.IsExpired())

    chunk2 := &MediaChunk{ExpiresAt: time.Now().Add(1 * time.Hour)}
    assert.False(t, chunk2.IsExpired())
}

// UseCase Test
func TestUploadMediaSingleUseCase_ValidFile_Success(t *testing.T) {
    mockMediaRepo := new(MockMediaRepository)
    mockStoragePort := new(MockStoragePort)
    mockPermissionPort := new(MockPermissionPort)
    mockQueue := new(MockProcessingQueuePort)
    mockPublisher := new(MockEventPublisherPort)

    // Setup expectations
    mockPermissionPort.On("CheckOrgMembership", mock.Anything, "user-1", "org-1").Return(true, nil)
    mockStoragePort.On("UploadObject", mock.Anything, mock.MatchedBy(func(key string) bool {
        return strings.Contains(key, "/original")
    }), mock.Anything, "image/jpeg").Return(nil)
    mockMediaRepo.On("Save", mock.Anything, mock.AnythingOfType("*domain.MediaFile")).Return(nil)
    mockQueue.On("EnqueueJob", mock.Anything, mock.AnythingOfType("*domain.ProcessingJob")).Return(nil)
    mockPublisher.On("Publish", mock.Anything, mock.AnythingOfType("domain.DomainEvent")).Return(nil)

    uc := NewUploadMediaSingleUseCase(mockMediaRepo, mockStoragePort, mockPermissionPort, mockQueue, mockPublisher)
    input := UploadMediaSingleInput{
        OrgID:      "org-1",
        UploaderID: "user-1",
        Filename:   "photo.jpg",
        MimeType:   "image/jpeg",
        FileBytes:  []byte("fake jpeg content"),
        AccessPolicy: "PRIVATE",
    }

    output, err := uc.Execute(context.Background(), input)

    require.NoError(t, err)
    assert.NotEmpty(t, output.MediaID)
    assert.Equal(t, "PROCESSING", output.Status)
    mockMediaRepo.AssertExpectations(t)
    mockStoragePort.AssertExpectations(t)
    mockPublisher.AssertExpectations(t)
}

func TestUploadMediaSingleUseCase_PermissionDenied(t *testing.T) {
    mockMediaRepo := new(MockMediaRepository)
    mockPermissionPort := new(MockPermissionPort)

    mockPermissionPort.On("CheckOrgMembership", mock.Anything, "user-1", "org-1").Return(false, nil)

    uc := NewUploadMediaSingleUseCase(mockMediaRepo, nil, mockPermissionPort, nil, nil)
    input := UploadMediaSingleInput{
        OrgID:      "org-1",
        UploaderID: "user-1",
        Filename:   "photo.jpg",
        MimeType:   "image/jpeg",
        FileBytes:  []byte("content"),
    }

    _, err := uc.Execute(context.Background(), input)
    assert.ErrorIs(t, err, ErrPermissionDenied)
}

func TestDeliverMediaUseCase_CacheHit(t *testing.T) {
    mockMediaRepo := new(MockMediaRepository)
    mockCache := new(MockCachePort)
    mockPermissionPort := new(MockPermissionPort)

    media := &MediaFile{
        ID:           "media-1",
        UploaderID:   "user-1",
        Status:       "READY",
        AccessPolicy: "PRIVATE",
    }

    mockMediaRepo.On("GetByID", mock.Anything, "org-1", "media-1").Return(media, nil)
    mockPermissionPort.On("CheckOrgMembership", mock.Anything, "user-1", "org-1").Return(true, nil)
    mockCache.On("Get", mock.Anything, "presigned:org-1:media-1:original").Return("https://s3.amazonaws.com/presigned-url", nil)

    uc := NewDeliverMediaUseCase(mockMediaRepo, mockCache, mockPermissionPort, nil)
    input := DeliverMediaInput{
        OrgID:            "org-1",
        MediaID:          "media-1",
        DeliveryType:     "original",
        RequestingUserID: "user-1",
    }

    output, err := uc.Execute(context.Background(), input)

    require.NoError(t, err)
    assert.Equal(t, "https://s3.amazonaws.com/presigned-url", output.PresignedURL)
}

func TestDeliverMediaUseCase_StatusFailed(t *testing.T) {
    mockMediaRepo := new(MockMediaRepository)

    media := &MediaFile{
        ID:     "media-1",
        Status: "FAILED",
    }

    mockMediaRepo.On("GetByID", mock.Anything, "org-1", "media-1").Return(media, nil)

    uc := NewDeliverMediaUseCase(mockMediaRepo, nil, nil, nil)
    input := DeliverMediaInput{
        OrgID:        "org-1",
        MediaID:      "media-1",
        DeliveryType: "original",
    }

    _, err := uc.Execute(context.Background(), input)
    assert.ErrorIs(t, err, ErrMediaNotFound)
}

// Integration Test
func TestHTTPMediaHandler_UploadMediaSingle_End2End(t *testing.T) {
    // testcontainers で MySQL 8.0/MariaDB 10.11・Garage（S3 互換 OSS）・Redis を起動
    db := setupTestMySQL(t)
    s3 := setupTestS3(t)
    redis := setupTestRedis(t)

    // リポジトリ・ポートを実装で初期化
    mediaRepo := MySQL.NewMediaRepository(db)
    storagePort := s3.NewStorageAdapter(s3.Client())
    cachePort := redis.NewCacheAdapter(redis.Client())

    // HTTPハンドラを起動
    handler := NewHTTPMediaHandler(mediaRepo, storagePort, cachePort)
    router := mux.NewRouter()
    router.HandleFunc("/api/media/{org_id}/single", handler.UploadSingle).Methods("POST")

    // テストリクエスト作成
    buf := new(bytes.Buffer)
    writer := multipart.NewWriter(buf)
    part, _ := writer.CreateFormFile("file", "test.jpg")
    part.Write([]byte("fake jpeg data"))
    writer.WriteField("access_policy", "PRIVATE")
    writer.Close()

    req := httptest.NewRequest("POST", "/api/media/org-1/single", buf)
    req.Header.Set("Content-Type", writer.FormDataContentType())
    req.Header.Set("X-User-Id", "user-1")

    // リクエスト実行
    w := httptest.NewRecorder()
    router.ServeHTTP(w, req)

    // アサーション
    assert.Equal(t, http.StatusOK, w.Code)
    var response map[string]interface{}
    json.Unmarshal(w.Body.Bytes(), &response)
    assert.NotEmpty(t, response["media_id"])
    assert.Equal(t, "PROCESSING", response["status"])
}

// E2E Test
func TestUploadDeliverDeleteFlow_E2E(t *testing.T) {
    // Docker Compose で完全なマイクロサービス環境を起動
    env := setupDockerCompose(t)
    defer env.Down()

    client := &http.Client{}

    // 1. アップロード
    uploadReq := setupUploadRequest("org-1", "user-1", testImagePath)
    uploadResp, _ := client.Do(uploadReq)
    assert.Equal(t, http.StatusOK, uploadResp.StatusCode)
    var uploadBody map[string]interface{}
    json.NewDecoder(uploadResp.Body).Decode(&uploadBody)
    mediaID := uploadBody["media_id"].(string)

    // 2. ポーリング: PROCESSING → READY
    time.Sleep(2 * time.Second) // 処理時間待機
    metadataReq := httptest.NewRequest("GET", "/api/media/org-1/"+mediaID+"/metadata", nil)
    metadataReq.Header.Set("X-User-Id", "user-1")
    metadataResp, _ := client.Do(metadataReq)
    var metadataBody map[string]interface{}
    json.NewDecoder(metadataResp.Body).Decode(&metadataBody)
    assert.Equal(t, "READY", metadataBody["status"])

    // 3. 配信（Presigned URL取得）
    deliveryReq := httptest.NewRequest("GET", "/api/media/org-1/"+mediaID+"?type=thumb", nil)
    deliveryReq.Header.Set("X-User-Id", "user-1")
    deliveryResp, _ := client.Do(deliveryReq)
    assert.Equal(t, http.StatusOK, deliveryResp.StatusCode)
    var deliveryBody map[string]interface{}
    json.NewDecoder(deliveryResp.Body).Decode(&deliveryBody)
    presignedURL := deliveryBody["presigned_url"].(string)
    assert.NotEmpty(t, presignedURL)

    // 4. Presigned URL経由でS3から実ファイル取得可能か確認
    fileResp, _ := client.Get(presignedURL)
    assert.Equal(t, http.StatusOK, fileResp.StatusCode)

    // 5. 削除
    deleteReq := httptest.NewRequest("DELETE", "/api/media/org-1/"+mediaID, nil)
    deleteReq.Header.Set("X-User-Id", "user-1")
    deleteResp, _ := client.Do(deleteReq)
    assert.Equal(t, http.StatusOK, deleteResp.StatusCode)

    // 6. 削除後はアクセス不可
    metadataReq2 := httptest.NewRequest("GET", "/api/media/org-1/"+mediaID+"/metadata", nil)
    metadataReq2.Header.Set("X-User-Id", "user-1")
    metadataResp2, _ := client.Do(metadataReq2)
    assert.Equal(t, http.StatusNotFound, metadataResp2.StatusCode)
}