Phase 3 integration: enqueue via RabbitMQ and execute jobs with downloader-backed workers

[Copilot]

This PR advances the current Phase 3 gap by replacing in-process job submission/processing shortcuts with the intended async flow. API-created jobs now enter RabbitMQ, and workers execute media items through the downloader abstraction with persisted job/item progress updates.

• API → Queue boundary (source of truth for async dispatch)

  • POST /jobs now publishes a queue.Message to RabbitMQ instead of submitting directly to the in-process worker channel.
  • Route wiring was updated so API handlers depend on queue.Service for dispatch.

• Worker execution path (remove simulated processing)

  • Worker pool now accepts a downloader.Downloader dependency and runs real item execution via Download(...).
  • Processing creates a media item from source_url when none exists, updates per-item status/size, and drives job lifecycle states (PROCESSING → COMPLETED / FAILED).
  • Failure handling is centralized to persist job failure state and emit failure events.

• Jobs service internal accessors for background workers

  • Added internal (non-user-scoped) methods for worker-side reads (GetJobInternal, GetMediaItemsInternal) to avoid auth-bound APIs in background execution.
  • Added UpdateMediaItemSize passthrough for worker result persistence.

• Schema/repository alignment

  • Repository SQL was aligned with the existing migration schema by using media_items.size (instead of size_bytes) in select/return/update queries.

// API: enqueue instead of direct in-process submit
if err := queueSvc.Publish(ctx, queue.Message{
    JobID:  resp.ID.String(),
    Action: "PROCESS",
    Payload: map[string]interface{}{
        "source_url": req.SourceURL,
    },
    CreatedAt: time.Now().Format(time.RFC3339),
}); err != nil {
    // handle queue publish failure
}

[amirmalekian]

Review Fixes Before Merge — Async Download Execution Flow

I reviewed the current PR changes.

The overall direction is correct:

• API publishing jobs to RabbitMQ ✅
• Worker using downloader abstraction ✅
• Removing simulated processing ✅

Before merging, apply the following fixes.

IMPORTANT:
Keep this PR scoped to:
"Connect async jobs with downloader execution pipeline"

Do NOT start:

• playlist support
• progress reporting feature
• unrelated refactoring

---

1. Verify and complete RabbitMQ consumer flow

The intended architecture must be:

API:

HTTP Request
|
|
Create Job
|
|
Publish RabbitMQ Message

Worker:

RabbitMQ Consumer
|
|
Worker Pool Submit
|
|
Downloader Execution
|
|
Persist Result

Verify that cmd/worker actually consumes RabbitMQ messages.

If consumer wiring is incomplete:

• add the missing consumer connection
• make consumer dispatch messages into worker pool

Do not leave worker pool running without queue consumption.

---

2. Move media item creation out of worker execution

Current behavior:

Worker creates media items when none exist.

Example:

Worker receives job
|
|
No media items found
|
|
Create media items

This mixes business creation logic with execution logic.

Change responsibility:

Job/API service should handle:

Create Job
Create Media Items
Publish Queue Message

Worker should only:

Consume Message
Load Existing Items
Download Items
Update Status
Persist Result

Remove automatic media item creation from worker processTask flow.

If existing flow requires temporary fallback behavior, add a TODO explaining migration path instead of silently creating entities.

---

3. Validate worker lifecycle states

Ensure job lifecycle remains consistent:

Expected flow:

PENDING
|
v
PROCESSING
|
+------------+
| |
v v
COMPLETED FAILED

Verify:

• Job is marked PROCESSING before execution
• Failed downloads update job state
• Completed downloads update job state
• Partial failures are handled correctly

Do not add new progress features yet.

---

4. Review queue publish failure handling

Scenario:

1. Database job creation succeeds
2. RabbitMQ publish fails

Current state must not become inconsistent.

Handle this case:

• Log publish failure
• Update job state appropriately OR return clear failure response

Do not leave jobs permanently stuck in PENDING without a queue message.

---

5. Verify Redis nil safety

Review all worker paths using Redis.

Ensure no panic occurs when Redis is unavailable.

Check patterns like:

if redis != nil {
    ...
}

Apply consistently where required.

6. Keep repository/schema changes focused

The media_items size column alignment is correct.

Keep:

size_bytes -> size

changes only where required.

Do not perform unrelated database cleanup.

7. Testing

After changes run:

go test ./...
golangci-lint run ./...
go build ./...

Add/update tests for:

queue dispatch flow
worker processing with downloader mock
failure handling

Avoid tests requiring:

external services
YouTube network access
real RabbitMQ unless marked integration.

Before finishing provide:

Files changed
Architecture flow after changes
Tests executed
Any remaining technical debt

Suggested commit:

fix(worker): improve async download execution flow

[Copilot]

Review Fixes Before Merge — Async Download Execution Flow

I reviewed the current PR changes.
...

Implemented ✅

• Verified worker consumer wiring remains active in /home/runner/work/streamforge/streamforge/cmd/worker/main.go (queueSvc.Consume(ctx, workerPool.JobChan())).
• Moved media-item creation to API flow before publish in /home/runner/work/streamforge/streamforge/internal/api/routes.go.
• Removed worker auto-creation fallback; worker now fails fast when no media items exist.
• Added queue publish failure handling to mark job FAILED and return clear 500.
• Mark jobs QUEUED after successful enqueue.
• Hardened Redis nil checks in touched job-service paths.
• Kept size_bytes -> size scope limited to repository SQL usage only.

Tests run:

• go test ./... ✅
• go build ./... ✅
• golangci-lint run ./... ⚠️ unavailable in this runtime (command not found)

Added focused tests:

• /home/runner/work/streamforge/streamforge/internal/api/routes_createjob_test.go
• /home/runner/work/streamforge/streamforge/internal/worker/pool_test.go

Remaining technical debt:

• Worker currently treats any item failure as terminal job failure (no partial-success terminal state yet).