story/internal/media/processor.go

package media

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"data-terminal-cloud/internal/audit"
	"data-terminal-cloud/internal/database"
	"data-terminal-cloud/internal/objectstore"
	"data-terminal-cloud/internal/outbox"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

type verificationStore interface {
	VerifyDigest(context.Context, string, string, int64) error
	ReadPrefix(context.Context, string, int64) ([]byte, error)
}

type Processor struct {
	pool             *pgxpool.Pool
	store            verificationStore
	audit            *audit.Repository
	localDeleteGrace time.Duration
}

func NewProcessor(pool *pgxpool.Pool, store verificationStore, auditRepository *audit.Repository, localDeleteGrace ...time.Duration) (*Processor, error) {
	if pool == nil || store == nil || auditRepository == nil {
		return nil, errors.New("media processor requires database, object store and audit repository")
	}
	grace := time.Hour
	if len(localDeleteGrace) > 0 {
		grace = localDeleteGrace[0]
	}
	if grace <= 0 {
		return nil, errors.New("media local delete grace must be positive")
	}
	return &Processor{pool: pool, store: store, audit: auditRepository, localDeleteGrace: grace}, nil
}

func (p *Processor) Handle(ctx context.Context, event outbox.Event) error {
	var payload struct {
		BatchID string `json:"batch_id"`
	}
	if err := json.Unmarshal(event.Payload, &payload); err != nil || payload.BatchID == "" {
		return errors.New("media verification event has no batch_id")
	}
	var batchID, tenantID, status string
	err := p.pool.QueryRow(ctx, `SELECT id::text,tenant_id::text,status FROM upload_batch WHERE public_id=$1::uuid`, payload.BatchID).Scan(&batchID, &tenantID, &status)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil
	}
	if err != nil {
		return err
	}
	if status == "COMPLETED" || status == "FAILED" || status == "EXPIRED" {
		return nil
	}
	if status != "VERIFYING" {
		return fmt.Errorf("media batch %s is in unexpected status %s", payload.BatchID, status)
	}

	rows, err := p.pool.Query(ctx, `
		SELECT ms.id::text,ms.object_ref,ms.expected_digest,ms.expected_size,ms.codec,ms.duration_ms
		FROM media_segment ms JOIN upload_batch_segment ubs ON ubs.media_segment_id=ms.id
		WHERE ubs.upload_batch_id=$1::uuid AND ms.status='VERIFYING'
		ORDER BY ms.track,ms.sequence_no`, batchID)
	if err != nil {
		return err
	}
	type pendingSegment struct {
		id, objectRef, digest, codec string
		size                         int64
		durationMS                   int
	}
	var pending []pendingSegment
	for rows.Next() {
		var item pendingSegment
		if err := rows.Scan(&item.id, &item.objectRef, &item.digest, &item.size, &item.codec, &item.durationMS); err != nil {
			rows.Close()
			return err
		}
		pending = append(pending, item)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return err
	}

	for _, segment := range pending {
		err := p.store.VerifyDigest(ctx, segment.objectRef, segment.digest, segment.size)
		if err == nil {
			prefix, readErr := p.store.ReadPrefix(ctx, segment.objectRef, 64*1024)
			if readErr != nil {
				_, _ = p.pool.Exec(ctx, `UPDATE upload_batch SET verification_attempts=verification_attempts+1,last_error_code='OBJECT_NOT_VISIBLE',updated_at=now() WHERE id=$1::uuid`, batchID)
				return outbox.Retryable(fmt.Errorf("read media segment header: %w", readErr))
			}
			validationCode := validateAudio(segment.codec, segment.durationMS, prefix)
			if validationCode != "" {
				if _, updateErr := p.pool.Exec(ctx, `UPDATE media_segment SET status='REJECTED',verification_attempts=verification_attempts+1,last_error_code=$2,updated_at=now() WHERE id=$1::uuid AND status='VERIFYING'`, segment.id, validationCode); updateErr != nil {
					return updateErr
				}
				continue
			}
			if _, updateErr := p.pool.Exec(ctx, `UPDATE media_segment SET status='VERIFIED',verification_attempts=verification_attempts+1,verified_at=now(),local_delete_after=now()+($2::bigint*interval '1 microsecond'),last_error_code=NULL,updated_at=now() WHERE id=$1::uuid AND status='VERIFYING'`, segment.id, p.localDeleteGrace.Microseconds()); updateErr != nil {
				return updateErr
			}
			continue
		}
		if errors.Is(err, objectstore.ErrObjectSizeMismatch) || errors.Is(err, objectstore.ErrObjectDigestMismatch) {
			code := "DIGEST_MISMATCH"
			if errors.Is(err, objectstore.ErrObjectSizeMismatch) {
				code = "SIZE_MISMATCH"
			}
			if _, updateErr := p.pool.Exec(ctx, `UPDATE media_segment SET status='REJECTED',verification_attempts=verification_attempts+1,last_error_code=$2,updated_at=now() WHERE id=$1::uuid AND status='VERIFYING'`, segment.id, code); updateErr != nil {
				return updateErr
			}
			continue
		}
		_, _ = p.pool.Exec(ctx, `UPDATE upload_batch SET verification_attempts=verification_attempts+1,last_error_code='OBJECT_NOT_VISIBLE',updated_at=now() WHERE id=$1::uuid`, batchID)
		return outbox.Retryable(fmt.Errorf("verify media segment object: %w", err))
	}

	return database.Transact(ctx, p.pool, func(tx pgx.Tx) error {
		var verified, rejected, remaining int
		if err := tx.QueryRow(ctx, `
			SELECT count(*) FILTER (WHERE ms.status='VERIFIED'),count(*) FILTER (WHERE ms.status='REJECTED'),count(*) FILTER (WHERE ms.status NOT IN ('VERIFIED','REJECTED','DELETED'))
			FROM media_segment ms JOIN upload_batch_segment ubs ON ubs.media_segment_id=ms.id WHERE ubs.upload_batch_id=$1::uuid`, batchID,
		).Scan(&verified, &rejected, &remaining); err != nil {
			return err
		}
		if remaining > 0 {
			return fmt.Errorf("media batch still has %d non-terminal segments", remaining)
		}
		batchStatus := "COMPLETED"
		lastError := any(nil)
		if rejected > 0 {
			batchStatus = "FAILED"
			lastError = "SEGMENT_REJECTED"
		}
		if _, err := tx.Exec(ctx, `UPDATE upload_batch SET status=$2,verified_count=$3,rejected_count=$4,verification_attempts=verification_attempts+1,last_error_code=$5,completed_at=now(),updated_at=now() WHERE id=$1::uuid`, batchID, batchStatus, verified, rejected, lastError); err != nil {
			return err
		}
		auditResult := "success"
		if rejected > 0 {
			auditResult = "failure"
		}
		return p.audit.Record(ctx, tx, audit.Event{TenantID: &tenantID, ActorType: "worker", Action: "media.upload_batch.verify", ResourceType: "upload_batch", ResourceID: &payload.BatchID, Result: auditResult, Detail: map[string]any{"status": batchStatus, "verified": verified, "rejected": rejected}})
	})
}