diff --git a/apps/admin/src/app/(authenticated)/media/[id]/MediaDetailClient.test.tsx b/apps/admin/src/app/(authenticated)/media/[id]/MediaDetailClient.test.tsx index 4554a2cf..4ca78506 100644 --- a/apps/admin/src/app/(authenticated)/media/[id]/MediaDetailClient.test.tsx +++ b/apps/admin/src/app/(authenticated)/media/[id]/MediaDetailClient.test.tsx @@ -145,4 +145,50 @@ describe('MediaDetailClient', () => { render(); expect(screen.getByTestId('media-detail-storage')).toMatchSnapshot(); }); + + it('hides the derivatives panel when no derivatives are present', () => { + render(); + expect(screen.queryByTestId('media-detail-derivatives')).not.toBeInTheDocument(); + }); + + it('renders the HLS link when hls_url is set (issue #52)', () => { + const video: MediaAsset = { + ...asset, + id: 'video-1', + mime_type: 'video/mp4', + filename: 'clip.mp4', + hls_url: 'https://cdn.example/hls/video-1/index.m3u8', + }; + render(); + const link = screen.getByTestId('hls-link'); + expect(link).toHaveAttribute('href', 'https://cdn.example/hls/video-1/index.m3u8'); + }); + + it('renders the extracted-text link when has_extracted_text is true (issue #60)', () => { + const pdf: MediaAsset = { + ...asset, + id: 'pdf-1', + mime_type: 'application/pdf', + filename: 'doc.pdf', + has_extracted_text: true, + }; + render(); + const link = screen.getByTestId('extracted-text-link'); + expect(link).toHaveAttribute('href', '/media/pdf-1/text'); + }); + + it('renders the proxy source link for proxied assets (issue #187)', () => { + const proxied: MediaAsset = { + ...asset, + id: 'proxy-1', + is_proxied: true, + source_url: 'https://oldsite.example/uploads/2024/03/photo.jpg', + }; + render(); + const link = screen.getByTestId('proxy-source-link'); + expect(link).toHaveAttribute( + 'href', + 'https://oldsite.example/uploads/2024/03/photo.jpg', + ); + }); }); diff --git a/apps/admin/src/app/(authenticated)/media/[id]/MediaDetailClient.tsx b/apps/admin/src/app/(authenticated)/media/[id]/MediaDetailClient.tsx index 0157469a..e75f0baa 100644 --- a/apps/admin/src/app/(authenticated)/media/[id]/MediaDetailClient.tsx +++ b/apps/admin/src/app/(authenticated)/media/[id]/MediaDetailClient.tsx @@ -282,6 +282,8 @@ export function MediaDetailClient(props: MediaDetailClientProps): ReactElement { onCopy={onCopyUrl} /> + +
@@ -300,6 +302,86 @@ export function MediaDetailClient(props: MediaDetailClientProps): ReactElement { ); } +/** + * Derivatives panel — surfaces the asynchronously-produced artifacts + * the heavy-media worker writes back to the row. Currently: + * + * * HLS playlist URL (issue #52) — shown when `hls_url` is set; + * used by the public player to stream video over HLS instead of + * pulling the entire mp4. + * * Extracted text link (issue #60) — shown when + * `has_extracted_text` is true; opens the read-only extracted + * text view in a new tab so the operator can verify what was + * indexed. + * * Proxied / source URL (issue #187) — shown for migration rows + * in proxy mode; surfaces the origin URL with an "open" link so + * an operator can trace the row back to the source site. + * + * The panel renders nothing when none of the fields are set — most + * assets (images uploaded directly) have no derivatives and the + * panel stays out of the layout. + */ +function DerivativesPanel({ asset }: { asset: MediaAsset }): ReactElement | null { + const hasHls = Boolean(asset.hls_url); + const hasText = Boolean(asset.has_extracted_text); + const hasProxy = Boolean(asset.is_proxied && asset.source_url); + if (!hasHls && !hasText && !hasProxy) { + return null; + } + return ( +
+ + Derivatives + +
+ {hasHls && ( + + + )} + {hasText && ( + +
+
+ ); +} + /** * Storage URL panel — the operator's primary "give me the link" * surface. Shows the URL in Geist Mono inside a paper-3 sunken pill diff --git a/apps/admin/src/app/(authenticated)/media/types.ts b/apps/admin/src/app/(authenticated)/media/types.ts index 9fc207bf..fc7c2a2e 100644 --- a/apps/admin/src/app/(authenticated)/media/types.ts +++ b/apps/admin/src/app/(authenticated)/media/types.ts @@ -88,6 +88,32 @@ export interface BulkResult { op: string; succeeded: number; failed?: Record; + + /** + * HLS playlist URL for video assets — populated by the + * media.video.transcode worker (#52). The video player picks HLS + * over the raw mp4 source when this is set. + */ + hls_url?: string; + + /** + * True when the media_text row exists for this asset. The detail + * page surfaces a "View extracted text" link based on this flag. + * Issue #60. + */ + has_extracted_text?: boolean; + + /** + * True for assets registered in proxy mode by the migration + * importer (#187). The grid shows a "proxied" badge so operators + * can tell at a glance which assets live remotely. + */ + is_proxied?: boolean; + + /** + * Origin URL for proxied assets. Empty for locally-stored assets. + */ + source_url?: string; } /** diff --git a/apps/api/internal/admin/media/handler.go b/apps/api/internal/admin/media/handler.go index 97e2c4fb..3cbbcf7d 100644 --- a/apps/api/internal/admin/media/handler.go +++ b/apps/api/internal/admin/media/handler.go @@ -41,6 +41,17 @@ type Deps struct { // passes the taskspec-backed adapter. Processor ProcessEnqueuer + // VideoProcessor enqueues media.video.transcode for video/* mime + // types. Optional — nil means no transcoding is fired (the upload + // row is still committed and queryable; the player will fall back + // to the raw video src). Issue #52. + VideoProcessor ProcessEnqueuer + + // PDFProcessor enqueues media.pdf.process for application/pdf + // mime types. Optional — nil means no thumbnail/text extraction + // is fired. Issue #60. + PDFProcessor ProcessEnqueuer + // Logger receives structured log lines. nil falls back to // slog.Default — useful for tests; production wiring should always // pass a service logger. @@ -72,13 +83,15 @@ func (d Deps) validate() error { } type handlers struct { - store Store - putter ObjectPutter - policy policy.Policy - processor ProcessEnqueuer - logger *slog.Logger - now func() time.Time - maxBytes int64 + store Store + putter ObjectPutter + policy policy.Policy + processor ProcessEnqueuer + videoProcessor ProcessEnqueuer + pdfProcessor ProcessEnqueuer + logger *slog.Logger + now func() time.Time + maxBytes int64 } // Mount wires the media routes onto mux under base (typically @@ -106,13 +119,15 @@ func Mount(mux *http.ServeMux, base string, deps Deps) error { } h := &handlers{ - store: deps.Store, - putter: deps.Putter, - policy: deps.Policy, - processor: deps.Processor, - logger: deps.Logger, - now: deps.Now, - maxBytes: maxBytes, + store: deps.Store, + putter: deps.Putter, + policy: deps.Policy, + processor: deps.Processor, + videoProcessor: deps.VideoProcessor, + pdfProcessor: deps.PDFProcessor, + logger: deps.Logger, + now: deps.Now, + maxBytes: maxBytes, } base = strings.TrimRight(base, "/") mux.Handle("POST "+base, h.gate(policy.CapMediaUpload, h.upload)) @@ -295,6 +310,30 @@ func (h *handlers) upload(w http.ResponseWriter, r *http.Request, pr policy.Prin } } + // MIME-routed pipelines. Video uploads go through the HLS + // transcoder (#52); PDFs go through pdftoppm + pdftotext (#60). + // Both follow the same "log on enqueue failure, never fail the + // upload" policy as the image pipeline — the row is the user- + // visible artifact, derivatives are a follow-up. + if h.videoProcessor != nil && strings.HasPrefix(strings.ToLower(asset.MimeType), "video/") { + if err := h.videoProcessor.Enqueue(r.Context(), asset.ID, asset.StorageKey, asset.MimeType); err != nil { + h.logger.WarnContext(r.Context(), "admin/media: enqueue video transcode failed", + slog.String("asset_id", asset.ID), + slog.String("storage_key", asset.StorageKey), + slog.Any("err", err), + ) + } + } + if h.pdfProcessor != nil && strings.EqualFold(strings.TrimSpace(asset.MimeType), "application/pdf") { + if err := h.pdfProcessor.Enqueue(r.Context(), asset.ID, asset.StorageKey, asset.MimeType); err != nil { + h.logger.WarnContext(r.Context(), "admin/media: enqueue pdf process failed", + slog.String("asset_id", asset.ID), + slog.String("storage_key", asset.StorageKey), + slog.Any("err", err), + ) + } + } + router.WriteJSON(w, http.StatusCreated, asset) } diff --git a/apps/api/internal/admin/media/model.go b/apps/api/internal/admin/media/model.go index c5e885c1..99404c45 100644 --- a/apps/api/internal/admin/media/model.go +++ b/apps/api/internal/admin/media/model.go @@ -77,6 +77,31 @@ type Asset struct { // not yet completed; clients should treat absence as "fall back // to the original via PublicURL". Variants []Variant `json:"variants,omitempty"` + + // HLSURL is the public URL of the HLS playlist produced by the + // media.video.transcode task (#52). Empty for non-video assets + // and for video assets whose transcode hasn't completed yet; the + // public player should fall back to PublicURL when this is + // empty. + HLSURL string `json:"hls_url,omitempty"` + + // HasExtractedText is true when the media_text table has a row + // for this asset (#60). The detail page surfaces a "View + // extracted text" link based on this flag; the full payload + // lives behind a separate endpoint to keep the list response + // from ballooning on long documents. + HasExtractedText bool `json:"has_extracted_text,omitempty"` + + // IsProxied is true when the row represents a remotely-hosted + // asset registered in proxy mode by the migration importer + // (#187). The image proxy serves the bytes via SourceURL; the + // admin grid surfaces a "proxied" badge so an operator can tell + // at a glance which assets are local vs remote. + IsProxied bool `json:"is_proxied,omitempty"` + + // SourceURL is the origin URL for proxied assets. Empty for + // locally-stored assets. + SourceURL string `json:"source_url,omitempty"` } // Variant is one rendition produced by packages/go/media/imageproc. diff --git a/apps/api/internal/admin/media/processing_test.go b/apps/api/internal/admin/media/processing_test.go index f3b5891f..b32d7331 100644 --- a/apps/api/internal/admin/media/processing_test.go +++ b/apps/api/internal/admin/media/processing_test.go @@ -258,3 +258,160 @@ func TestStore_SetVariantsOnDeletedRow(t *testing.T) { t.Errorf("SetVariants on deleted row: err = %v, want ErrNotFound", err) } } + +// newMuxWithAllProcessors mirrors newMuxWithProcessor but wires three +// independent enqueuers: the image processor (always fired), the +// video processor (fired only on video/*), and the PDF processor +// (fired only on application/pdf). Used to assert the MIME routing +// of the upload handler. +func newMuxWithAllProcessors(t *testing.T, image, video, pdf ProcessEnqueuer) (*http.ServeMux, *MemoryStore, *MemoryPutter) { + t.Helper() + var idSeq int + idGen := func() string { + idSeq++ + return "asset-" + strings.Repeat("0", 4-len(itoa(idSeq))) + itoa(idSeq) + } + base := time.Date(2026, 1, 15, 12, 0, 0, 0, time.UTC) + store := NewMemoryStore(func() time.Time { return base }, idGen) + putter := NewMemoryPutter() + mux := http.NewServeMux() + if err := Mount(mux, "/api/v1/admin/media", Deps{ + Store: store, + Putter: putter, + Policy: policy.NewBasicPolicy(policy.DefaultRoleCapabilities()), + Processor: image, + VideoProcessor: video, + PDFProcessor: pdf, + Now: func() time.Time { return base }, + MaxBytes: 1024 * 1024, + }); err != nil { + t.Fatalf("Mount: %v", err) + } + return mux, store, putter +} + +// mp4Bytes returns bytes that http.DetectContentType sniffs as a +// video file. The 'ftyp' box at offset 4 with brand 'isom' is the +// canonical MP4/ISO BMFF signature. +func mp4Bytes() []byte { + // Box size (8 bytes), 'ftyp' (4 bytes), brand 'isom' (4 bytes), + // minor version (4 bytes), compatible brands. Total 20 bytes is + // enough for http.DetectContentType to recognise this as video. + return []byte{ + 0x00, 0x00, 0x00, 0x18, 'f', 't', 'y', 'p', + 'i', 's', 'o', 'm', 0x00, 0x00, 0x00, 0x01, + 'i', 's', 'o', 'm', 'm', 'p', '4', '1', + } +} + +// pdfBytes returns bytes that http.DetectContentType sniffs as a PDF +// document. +func pdfBytes() []byte { + return []byte("%PDF-1.4\n%\xE2\xE3\xCF\xD3\n1 0 obj\n<<>>\nendobj\n%%EOF\n") +} + +// TestUpload_RoutesVideoToVideoProcessor confirms a video upload +// fires the video processor and NOT the PDF processor. +func TestUpload_RoutesVideoToVideoProcessor(t *testing.T) { + image := &recordingEnqueuer{} + video := &recordingEnqueuer{} + pdf := &recordingEnqueuer{} + mux, _, _ := newMuxWithAllProcessors(t, image, video, pdf) + + body, ct := buildMultipart(t, "clip.mp4", mp4Bytes()) + req := withAuth(httptest.NewRequest(http.MethodPost, "/api/v1/admin/media", body), authedPrincipal()) + req.Header.Set("Content-Type", ct) + w := httptest.NewRecorder() + mux.ServeHTTP(w, req) + if w.Code != http.StatusCreated { + t.Fatalf("upload status = %d, body = %s", w.Code, w.Body.String()) + } + + if len(video.Calls()) != 1 { + t.Errorf("video processor calls = %d, want 1", len(video.Calls())) + } + if len(pdf.Calls()) != 0 { + t.Errorf("pdf processor should not fire for video upload; got %d calls", len(pdf.Calls())) + } + // Image processor (the existing media.process path) still fires + // regardless of MIME — the image pipeline decides internally + // whether to skip non-images. + if len(image.Calls()) != 1 { + t.Errorf("image processor calls = %d, want 1", len(image.Calls())) + } +} + +// TestUpload_RoutesPDFToPDFProcessor confirms a PDF upload fires the +// PDF processor and NOT the video processor. +func TestUpload_RoutesPDFToPDFProcessor(t *testing.T) { + image := &recordingEnqueuer{} + video := &recordingEnqueuer{} + pdf := &recordingEnqueuer{} + mux, _, _ := newMuxWithAllProcessors(t, image, video, pdf) + + body, ct := buildMultipart(t, "doc.pdf", pdfBytes()) + req := withAuth(httptest.NewRequest(http.MethodPost, "/api/v1/admin/media", body), authedPrincipal()) + req.Header.Set("Content-Type", ct) + w := httptest.NewRecorder() + mux.ServeHTTP(w, req) + if w.Code != http.StatusCreated { + t.Fatalf("upload status = %d, body = %s", w.Code, w.Body.String()) + } + + if len(pdf.Calls()) != 1 { + t.Errorf("pdf processor calls = %d, want 1", len(pdf.Calls())) + } + if len(video.Calls()) != 0 { + t.Errorf("video processor should not fire for PDF upload; got %d calls", len(video.Calls())) + } +} + +// TestUpload_ImageDoesNotFireVideoOrPDF confirms an image upload only +// hits the image processor. +func TestUpload_ImageDoesNotFireVideoOrPDF(t *testing.T) { + image := &recordingEnqueuer{} + video := &recordingEnqueuer{} + pdf := &recordingEnqueuer{} + mux, _, _ := newMuxWithAllProcessors(t, image, video, pdf) + + body, ct := buildMultipart(t, "logo.png", pngBytes()) + req := withAuth(httptest.NewRequest(http.MethodPost, "/api/v1/admin/media", body), authedPrincipal()) + req.Header.Set("Content-Type", ct) + w := httptest.NewRecorder() + mux.ServeHTTP(w, req) + if w.Code != http.StatusCreated { + t.Fatalf("upload status = %d", w.Code) + } + + if len(image.Calls()) != 1 { + t.Errorf("image processor calls = %d, want 1", len(image.Calls())) + } + if len(video.Calls()) != 0 { + t.Errorf("video processor should not fire for image; got %d calls", len(video.Calls())) + } + if len(pdf.Calls()) != 0 { + t.Errorf("pdf processor should not fire for image; got %d calls", len(pdf.Calls())) + } +} + +// TestUpload_VideoEnqueueErrorDoesNotFailUpload pins the same contract +// as TestUpload_EnqueueErrorDoesNotFailUpload but for the video path: +// a worker outage at video-enqueue time must not lose the upload. +func TestUpload_VideoEnqueueErrorDoesNotFailUpload(t *testing.T) { + image := &recordingEnqueuer{} + video := &recordingEnqueuer{err: errors.New("queue down")} + mux, store, _ := newMuxWithAllProcessors(t, image, video, nil) + + body, ct := buildMultipart(t, "clip.mp4", mp4Bytes()) + req := withAuth(httptest.NewRequest(http.MethodPost, "/api/v1/admin/media", body), authedPrincipal()) + req.Header.Set("Content-Type", ct) + w := httptest.NewRecorder() + mux.ServeHTTP(w, req) + if w.Code != http.StatusCreated { + t.Fatalf("upload should succeed even with video enqueue failure: status = %d", w.Code) + } + page, _ := store.List(context.Background(), ListFilter{}) + if len(page.Data) != 1 { + t.Errorf("row not committed after video enqueue failure: rows = %d", len(page.Data)) + } +} diff --git a/apps/worker/cmd/worker/main.go b/apps/worker/cmd/worker/main.go index 534e0a29..635a6cde 100644 --- a/apps/worker/cmd/worker/main.go +++ b/apps/worker/cmd/worker/main.go @@ -24,6 +24,7 @@ import ( "github.com/hibiken/asynq" + workermedia "github.com/Singleton-Solution/GoNext/apps/worker/internal/media" "github.com/Singleton-Solution/GoNext/packages/go/buildinfo" "github.com/Singleton-Solution/GoNext/packages/go/config" jobsasynq "github.com/Singleton-Solution/GoNext/packages/go/jobs/asynq" @@ -109,7 +110,7 @@ func run(ctx context.Context) error { return fmt.Errorf("parse REDIS_URL: %w", err) } - srv, _, err := jobsasynq.New(redisOpt, jobsasynq.Config{ + srv, mux, err := jobsasynq.New(redisOpt, jobsasynq.Config{ Logger: logger, Metrics: mreg.Prometheus(), }) @@ -117,6 +118,20 @@ func run(ctx context.Context) error { return fmt.Errorf("jobs/asynq: %w", err) } + // Heavy-media tasks. Registered in stub mode for the boot-time + // skeleton — the package consults the PATH and the wired storage + // handles to decide between the real handler and the stub. + // Production wiring (when the worker grows S3 access) replaces + // the zero-value Deps with real Source/Sink implementations. + // + // See apps/worker/internal/media for the dispatch contract. + mediaTaskRegistry := taskspec.NewRegistry() + if _, err := workermedia.Register(mux, mediaTaskRegistry, workermedia.Deps{ + Logger: logger, + }); err != nil { + return fmt.Errorf("worker/media: register: %w", err) + } + // Registration order (locked in by issue #112): // // 1. db.pool (registered first → drains last) — future diff --git a/apps/worker/internal/media/wire.go b/apps/worker/internal/media/wire.go new file mode 100644 index 00000000..bda044b7 --- /dev/null +++ b/apps/worker/internal/media/wire.go @@ -0,0 +1,223 @@ +// Package media wires the heavy-media task handlers (HLS video +// transcoding and PDF thumbnail/text extraction) onto the worker's +// asynq mux. +// +// The package lives inside apps/worker (not in packages/go) because +// the wiring is binary-specific: it depends on the storage layer the +// worker chooses, the logger the worker spins up, and the registry +// the worker keeps. The handlers themselves live in packages/go/media/ +// (videoproc, pdfproc) — only the boot-time wiring lives here. +// +// # Skip-graceful binary checks +// +// Both pipelines need on-PATH binaries (ffmpeg for video; pdftoppm +// /pdftotext for PDF). At boot the package probes the PATH and picks +// between the production handler and a stub. The stub still +// registers on the mux so Enqueue calls from the API don't error with +// "unknown task"; the stub just logs and returns nil for every +// payload. That keeps a deployment without the binaries running +// healthy — uploads succeed, rows commit, the derivative pipeline +// is a no-op. +// +// # Wiring contract +// +// Register accepts a Deps bag, registers the appropriate specs onto +// the worker's taskspec registry, and dispatches them onto the +// asynq mux. It returns a Report describing what was wired ("real" +// vs "stub") so the worker can log the decision and operators have a +// visible record of why a deployment isn't transcoding. +package media + +import ( + "context" + "fmt" + "log/slog" + + "github.com/hibiken/asynq" + + "github.com/Singleton-Solution/GoNext/packages/go/jobs/taskspec" + "github.com/Singleton-Solution/GoNext/packages/go/media/pdfproc" + "github.com/Singleton-Solution/GoNext/packages/go/media/videoproc" +) + +// Deps is the dependency bag for Register. Storage handles are +// optional in the boot-time skeleton: when nil, Register installs the +// stub handlers (matching the case where ffmpeg/pdftoppm are missing +// on PATH). Operators wiring a real worker pass real Source/Sink +// implementations backed by their S3 layer. +type Deps struct { + // VideoSource pulls original bytes for the transcoder. Optional; + // nil triggers stub registration for media.video.transcode. + VideoSource videoproc.Source + + // VideoSink writes the HLS playlist + segments. Optional. + VideoSink videoproc.Sink + + // VideoHLSWriter updates the media row's hls_url column. + // Optional. + VideoHLSWriter videoproc.HLSWriter + + // VideoWorkDir is the scratch directory for the video pipeline. + // Empty falls back to os.TempDir() inside the handler. + VideoWorkDir string + + // PDFSource pulls original PDF bytes. Optional. + PDFSource pdfproc.Source + + // PDFSink writes the page-1 thumbnail. Optional. + PDFSink pdfproc.Sink + + // PDFTextWriter persists the extracted text to media_text. + // Optional. + PDFTextWriter pdfproc.TextWriter + + // PDFThumbnailWriter records the thumbnail storage key. Optional. + PDFThumbnailWriter pdfproc.ThumbnailWriter + + // PDFWorkDir scratch directory. Empty falls back to os.TempDir(). + PDFWorkDir string + + // Logger receives structured log lines from the wiring layer. + // nil falls back to slog.Default. + Logger *slog.Logger +} + +// Report describes what Register actually wired. +type Report struct { + // VideoMode is "real" when ffmpeg was found on PATH and all + // dependencies were wired, "stub" otherwise. + VideoMode string + + // PDFMode is "real" or "stub" — same semantics as VideoMode. + PDFMode string + + // FFmpegPath is the absolute path to ffmpeg when found, "". + FFmpegPath string + + // PDFAvailability reports which PDF binaries were found on PATH. + PDFAvailability pdfproc.Availability +} + +// Register wires both heavy-media task handlers onto mux, falling +// back to stubs when the host's binaries or the caller's storage +// handles are missing. +// +// Logs at info for each task ("registered media.video.transcode in +// real mode" / "...in stub mode"). The log line is the operator's +// signal that the deployment supports HLS — a missing or stub line +// in the boot log is the first place to look when a video upload +// doesn't transcode. +func Register(mux *asynq.ServeMux, reg *taskspec.Registry, deps Deps) (Report, error) { + if mux == nil { + return Report{}, fmt.Errorf("worker/media: nil asynq mux") + } + if reg == nil { + return Report{}, fmt.Errorf("worker/media: nil taskspec registry") + } + logger := deps.Logger + if logger == nil { + logger = slog.Default() + } + + report := Report{} + + // ────────────────────────────────────────────────────────────── + // Video transcoding (media.video.transcode) + // ────────────────────────────────────────────────────────────── + ffmpegPath, ffmpegOK := videoproc.IsAvailable() + report.FFmpegPath = ffmpegPath + + videoSpec, vmode, err := buildVideoSpec(deps, ffmpegOK, logger) + if err != nil { + return report, fmt.Errorf("worker/media: build video spec: %w", err) + } + if err := reg.Register(videoSpec); err != nil { + return report, fmt.Errorf("worker/media: register video spec: %w", err) + } + mux.HandleFunc(videoSpec.Name, asynqAdapter(videoSpec.Handler)) + report.VideoMode = vmode + logger.Info("worker/media: video task registered", + slog.String("task", videoSpec.Name), + slog.String("mode", vmode), + slog.String("ffmpeg_path", ffmpegPath), + ) + + // ────────────────────────────────────────────────────────────── + // PDF processing (media.pdf.process) + // ────────────────────────────────────────────────────────────── + pdfAvail := pdfproc.Probe() + report.PDFAvailability = pdfAvail + + pdfSpec, pmode, err := buildPDFSpec(deps, pdfAvail, logger) + if err != nil { + return report, fmt.Errorf("worker/media: build pdf spec: %w", err) + } + if err := reg.Register(pdfSpec); err != nil { + return report, fmt.Errorf("worker/media: register pdf spec: %w", err) + } + mux.HandleFunc(pdfSpec.Name, asynqAdapter(pdfSpec.Handler)) + report.PDFMode = pmode + logger.Info("worker/media: pdf task registered", + slog.String("task", pdfSpec.Name), + slog.String("mode", pmode), + slog.String("pdftoppm_path", pdfAvail.PDFToPPMPath), + slog.String("pdftotext_path", pdfAvail.PDFToTextPath), + slog.String("pdfcpu_path", pdfAvail.PDFCPUPath), + ) + return report, nil +} + +// buildVideoSpec picks between the real and stub video spec based on +// ffmpeg availability and dependency wiring. +// +// The stub fires when EITHER ffmpeg is missing on PATH OR the storage +// handles haven't been wired. +func buildVideoSpec(deps Deps, ffmpegOK bool, logger *slog.Logger) (taskspec.TaskSpec, string, error) { + canRunReal := ffmpegOK && deps.VideoSource != nil && deps.VideoSink != nil + if !canRunReal { + spec, err := videoproc.NewStubSpec(logger) + return spec, "stub", err + } + spec, err := videoproc.NewSpec(videoproc.HandlerDeps{ + Source: deps.VideoSource, + Sink: deps.VideoSink, + HLSWriter: deps.VideoHLSWriter, + Runner: videoproc.ExecRunner{}, + WorkDir: deps.VideoWorkDir, + Logger: logger, + }) + return spec, "real", err +} + +// buildPDFSpec picks between the real and stub PDF spec. +func buildPDFSpec(deps Deps, avail pdfproc.Availability, logger *slog.Logger) (taskspec.TaskSpec, string, error) { + canRunReal := (avail.CanRender() || avail.CanExtractText()) && deps.PDFSource != nil && deps.PDFSink != nil + if !canRunReal { + spec, err := pdfproc.NewStubSpec(logger) + return spec, "stub", err + } + spec, err := pdfproc.NewSpec(pdfproc.HandlerDeps{ + Source: deps.PDFSource, + Sink: deps.PDFSink, + TextWriter: deps.PDFTextWriter, + ThumbnailWriter: deps.PDFThumbnailWriter, + Runner: pdfproc.ExecRunner{}, + Availability: avail, + WorkDir: deps.PDFWorkDir, + Logger: logger, + }) + return spec, "real", err +} + +// asynqAdapter wraps a (ctx, []byte) taskspec.Handler in the asynq +// handler signature mux.HandleFunc expects. +// +// We duplicate the adapter that lives in taskspec.Dispatch here +// because the wiring layer wants per-task control: a future task +// may need bespoke registration (priority hints, middleware) that +// the all-or-nothing Dispatch helper doesn't offer. +func asynqAdapter(h func(context.Context, []byte) error) func(context.Context, *asynq.Task) error { + return func(ctx context.Context, t *asynq.Task) error { + return h(ctx, t.Payload()) + } +} diff --git a/apps/worker/internal/media/wire_test.go b/apps/worker/internal/media/wire_test.go new file mode 100644 index 00000000..b5665680 --- /dev/null +++ b/apps/worker/internal/media/wire_test.go @@ -0,0 +1,129 @@ +package media + +import ( + "context" + "testing" + + "github.com/hibiken/asynq" + + "github.com/Singleton-Solution/GoNext/packages/go/jobs/taskspec" + "github.com/Singleton-Solution/GoNext/packages/go/media/pdfproc" + "github.com/Singleton-Solution/GoNext/packages/go/media/videoproc" +) + +// fakeVideoSource implements videoproc.Source. The map is intentionally +// empty — these tests only assert registration succeeds; they don't +// exercise the handler body. +type fakeVideoSource struct{} + +func (fakeVideoSource) GetObject(ctx context.Context, key string) ([]byte, error) { + return nil, nil +} + +type fakeVideoSink struct{} + +func (fakeVideoSink) PutObject(ctx context.Context, key string, body []byte, mime string) error { + return nil +} +func (fakeVideoSink) PublicURL(key string) string { return "https://x/" + key } + +type fakePDFSource struct{} + +func (fakePDFSource) GetObject(ctx context.Context, key string) ([]byte, error) { + return nil, nil +} + +type fakePDFSink struct{} + +func (fakePDFSink) PutObject(ctx context.Context, key string, body []byte, mime string) error { + return nil +} +func (fakePDFSink) PublicURL(key string) string { return "https://x/" + key } + +// TestRegister_StubMode confirms that the boot path with no +// dependencies wires both tasks as stubs without erroring. This is +// the "fresh skeleton" case the worker boots into until the storage +// wiring lands. +func TestRegister_StubMode(t *testing.T) { + reg := taskspec.NewRegistry() + mux := asynq.NewServeMux() + rep, err := Register(mux, reg, Deps{}) + if err != nil { + t.Fatalf("Register: %v", err) + } + if rep.VideoMode != "stub" { + t.Errorf("VideoMode = %q, want stub", rep.VideoMode) + } + if rep.PDFMode != "stub" { + t.Errorf("PDFMode = %q, want stub", rep.PDFMode) + } + // Both names should be on the registry now. + if _, ok := reg.Get(videoproc.TaskName); !ok { + t.Errorf("video task not registered") + } + if _, ok := reg.Get(pdfproc.TaskName); !ok { + t.Errorf("pdf task not registered") + } +} + +// TestRegister_WithStorageMode confirms that supplying storage +// handles selects the "real" mode — independent of whether ffmpeg +// is actually on PATH (the test machine may not have it; the report +// flips back to stub in that case which is also fine, but we assert +// that providing storage does NOT itself produce an error). +func TestRegister_WithStorageWiring(t *testing.T) { + reg := taskspec.NewRegistry() + mux := asynq.NewServeMux() + _, err := Register(mux, reg, Deps{ + VideoSource: fakeVideoSource{}, + VideoSink: fakeVideoSink{}, + PDFSource: fakePDFSource{}, + PDFSink: fakePDFSink{}, + }) + if err != nil { + t.Fatalf("Register: %v", err) + } +} + +// TestRegister_NilMux rejects nil dependencies loudly. +func TestRegister_NilMux(t *testing.T) { + reg := taskspec.NewRegistry() + if _, err := Register(nil, reg, Deps{}); err == nil { + t.Fatal("Register(nil mux): expected error") + } + mux := asynq.NewServeMux() + if _, err := Register(mux, nil, Deps{}); err == nil { + t.Fatal("Register(nil registry): expected error") + } +} + +// TestRegister_Idempotency confirms a second call with the same +// registry returns the "already registered" error. +func TestRegister_Idempotency(t *testing.T) { + reg := taskspec.NewRegistry() + mux := asynq.NewServeMux() + if _, err := Register(mux, reg, Deps{}); err != nil { + t.Fatalf("first Register: %v", err) + } + // Use a fresh mux for the second call so asynq doesn't panic + // on the duplicate pattern; the registry should still reject. + mux2 := asynq.NewServeMux() + if _, err := Register(mux2, reg, Deps{}); err == nil { + t.Fatal("second Register: expected error from duplicate spec") + } +} + +// TestRegister_PDFAvailabilityReported just exercises the field on +// the report — the contents are environment-dependent. +func TestRegister_PDFAvailabilityReported(t *testing.T) { + reg := taskspec.NewRegistry() + mux := asynq.NewServeMux() + rep, err := Register(mux, reg, Deps{}) + if err != nil { + t.Fatalf("Register: %v", err) + } + // Just confirm the struct is populated (zero-value when nothing + // is on PATH is also a valid Availability state). + var avail pdfproc.Availability = rep.PDFAvailability + _ = avail +} diff --git a/cli/gonext/cmd/migrate/wp.go b/cli/gonext/cmd/migrate/wp.go index ec15cbaa..69c4a0d0 100644 --- a/cli/gonext/cmd/migrate/wp.go +++ b/cli/gonext/cmd/migrate/wp.go @@ -28,6 +28,7 @@ Flags: --on-conflict string Conflict policy: skip | update | fail. Default "skip". --batch-size int Number of post rows per transaction. Default 100. --skip-comments Do not import comment threads. + --media-mode string Media migration mode: '' (off, default), 'copy', or 'proxy'. See issue #187. Environment: DATABASE_URL Required (unless --dry-run). Postgres DSN. @@ -54,12 +55,14 @@ func runWP(args []string, stdout, stderr io.Writer) int { conflictFlag string batchSizeFlag int skipCommentsFlag bool + mediaModeFlag string ) fs.StringVar(&fileFlag, "file", "", "Path to WXR XML export (required)") fs.BoolVar(&dryFlag, "dry-run", false, "Walk the WXR but write no rows") fs.StringVar(&conflictFlag, "on-conflict", "skip", "Conflict policy: skip | update | fail") fs.IntVar(&batchSizeFlag, "batch-size", 100, "Posts per transaction") fs.BoolVar(&skipCommentsFlag, "skip-comments", false, "Skip comment import") + fs.StringVar(&mediaModeFlag, "media-mode", "", "Media migration mode: '' (off), 'copy' (download bytes), 'proxy' (proxy via image proxy). See issue #187.") fs.Usage = func() { fmt.Fprintln(stderr, wpUsage) } if err := fs.Parse(args); err != nil { @@ -81,6 +84,20 @@ func runWP(args []string, stdout, stderr io.Writer) int { return ExitUsage } + // Media migration mode is opt-in: an empty flag value disables + // the MediaMigrator entirely (post bodies retain their source + // URLs; the imported site hot-links to the source). Operators + // pick 'copy' or 'proxy' explicitly when they want migration. + // The CLI does not (yet) wire the MediaMigrator's storage + // backend — that requires DB + bucket configuration the wizard + // UI in #234 will supply. + if mediaModeFlag != "" { + if _, perr := importer.ParseMediaMode(mediaModeFlag); perr != nil { + fmt.Fprintf(stderr, "gonext migrate wp: %v\n", perr) + return ExitUsage + } + } + opts := importer.Options{ Dryrun: dryFlag, OnConflict: policy, @@ -163,6 +180,14 @@ func printReport(w io.Writer, r *importer.Report, dryrun bool) { fmt.Fprintf(w, " posts: %d\n", r.Posts) fmt.Fprintf(w, " attachments: %d\n", r.Attachments) fmt.Fprintf(w, " comments: %d\n", r.Comments) + if r.MediaCopied > 0 || r.MediaProxied > 0 || r.MediaSkipped > 0 { + fmt.Fprintf(w, " media copied: %d\n", r.MediaCopied) + fmt.Fprintf(w, " media proxied: %d\n", r.MediaProxied) + fmt.Fprintf(w, " media skipped: %d\n", r.MediaSkipped) + if r.MediaBytesFetched > 0 { + fmt.Fprintf(w, " media bytes: %d\n", r.MediaBytesFetched) + } + } fmt.Fprintf(w, " errors: %d\n", len(r.Errors)) fmt.Fprintf(w, " took: %s\n", r.Took) } diff --git a/migrations/000036_media_hls.down.sql b/migrations/000036_media_hls.down.sql new file mode 100644 index 00000000..473b70bc --- /dev/null +++ b/migrations/000036_media_hls.down.sql @@ -0,0 +1,8 @@ +-- 000036_media_hls.down.sql +-- +-- Reverse of 000036_media_hls.up.sql. Drops the HLS playlist column +-- from the media table. The bytes on disk (m3u8 + segments) are NOT +-- swept by this migration — that's a separate purge cron concern; +-- this only releases the row-side reference. + +ALTER TABLE media DROP COLUMN IF EXISTS hls_url; diff --git a/migrations/000036_media_hls.up.sql b/migrations/000036_media_hls.up.sql new file mode 100644 index 00000000..bb79d043 --- /dev/null +++ b/migrations/000036_media_hls.up.sql @@ -0,0 +1,37 @@ +-- 000036_media_hls.up.sql +-- +-- Adds the HLS playlist URL column to the media table — backs the +-- video transcoding pipeline (issue #52). When the worker's +-- media.video.transcode task lands, it writes the resulting +-- index.m3u8 URL here so the public player can pick HLS over the +-- raw mp4 source. +-- +-- Design notes +-- +-- * NULLABLE on purpose. Every existing video row predates the +-- pipeline and the playlist won't exist for them until an +-- operator triggers a reprocess; nullable means "not yet +-- transcoded" without a sentinel value. New uploads also flow +-- through nullable: the row is committed at upload-time, the +-- HLS URL fills in asynchronously once the worker finishes. +-- +-- * TEXT (not bytea / json) because the URL is exactly what the +--