feat(import): move PDF parsing to backend; fix heartbeat/reap for import_tasks

- parsePDF function restored in import.go (body was orphaned outside function)
- ParseImportFile() called at upload time with 3-min timeout; chapters stored as JSON in MinIO
- runner.go: prefer ChaptersKey path (read pre-parsed JSON) over BookImport.Import()
- ImportChapterStore interface added; store wired in runner/main.go
- HeartbeatTask and ReapStaleTasks now include import_tasks collection
- parseImportTask now returns ChaptersKey in domain.ImportTask
- asynq_runner.go handleImportTask passes ChaptersKey
- pb-init-v3.sh: chapters_key field added to import_tasks schema

backend/internal/storage/import.go

@@ -109,6 +109,40 @@ func decryptPDF(data []byte) ([]byte, error) {
 	return out.Bytes(), nil
 }
 
+// ParseImportFile parses a PDF or EPUB and returns chapters.
+// Unlike AnalyzeFile it respects ctx cancellation so callers can apply a timeout.
+// For PDFs it first attempts to strip encryption with an empty password.
+func ParseImportFile(ctx context.Context, data []byte, fileType string) ([]bookstore.Chapter, error) {
+	type result struct {
+		chapters []bookstore.Chapter
+		err      error
+	}
+	ch := make(chan result, 1)
+	go func() {
+		var chapters []bookstore.Chapter
+		var err error
+		switch fileType {
+		case "pdf":
+			chapters, err = parsePDF(data)
+		case "epub":
+			chapters, err = parseEPUB(data)
+		default:
+			err = fmt.Errorf("unsupported file type: %s", fileType)
+		}
+		ch <- result{chapters, err}
+	}()
+	select {
+	case <-ctx.Done():
+		return nil, fmt.Errorf("parse timed out: %w", ctx.Err())
+	case r := <-ch:
+		return r.chapters, r.err
+	}
+}
+
+// parsePDF extracts chapters from PDF bytes using dslipak/pdf.
+// It first attempts to decrypt the PDF with an empty password in case the file
+// uses owner-only encryption (copy/print restrictions), which is common for
+// publisher PDFs that open normally in readers but confuse raw parsers.
 func parsePDF(data []byte) ([]bookstore.Chapter, error) {
 	// If the PDF is encrypted, try to decrypt it with an empty password.
 	// Many publisher PDFs use owner-only encryption (copy/print restrictions)

backend/internal/storage/store.go

@@ -654,13 +654,14 @@ func (s *Store) CreateImportTask(ctx context.Context, task domain.ImportTask) (s
 		"file_name":         task.Slug + "." + task.FileType,
 		"file_type":         task.FileType,
 		"object_key":        task.ObjectKey,
+		"chapters_key":      task.ChaptersKey,
 		"author":            task.Author,
 		"cover_url":         task.CoverURL,
 		"summary":           task.Summary,
 		"book_status":       task.BookStatus,
 		"status":            string(domain.TaskStatusPending),
 		"chapters_done":     0,
-		"chapters_total":    0,
+		"chapters_total":    task.ChaptersTotal,
 		"started":           time.Now().UTC().Format(time.RFC3339),
 		"initiator_user_id": task.InitiatorUserID,
 	}
@@ -914,7 +915,7 @@ func (s *Store) FailTask(ctx context.Context, id, errMsg string) error {
 }
 
 // HeartbeatTask updates the heartbeat_at field on a running task.
-// Tries scraping_tasks first, then audio_jobs, then translation_jobs.
+// Tries scraping_tasks, audio_jobs, translation_jobs, then import_tasks.
 func (s *Store) HeartbeatTask(ctx context.Context, id string) error {
 	payload := map[string]any{
 		"heartbeat_at": time.Now().UTC().Format(time.RFC3339),
@@ -925,7 +926,10 @@ func (s *Store) HeartbeatTask(ctx context.Context, id string) error {
 	if err := s.pb.patch(ctx, fmt.Sprintf("/api/collections/audio_jobs/records/%s", id), payload); err == nil {
 		return nil
 	}
-	return s.pb.patch(ctx, fmt.Sprintf("/api/collections/translation_jobs/records/%s", id), payload)
+	if err := s.pb.patch(ctx, fmt.Sprintf("/api/collections/translation_jobs/records/%s", id), payload); err == nil {
+		return nil
+	}
+	return s.pb.patch(ctx, fmt.Sprintf("/api/collections/import_tasks/records/%s", id), payload)
 }
 
 // ReapStaleTasks finds all running tasks whose heartbeat_at is either missing
@@ -943,7 +947,7 @@ func (s *Store) ReapStaleTasks(ctx context.Context, staleAfter time.Duration) (i
 	}
 
 	total := 0
-	for _, collection := range []string{"scraping_tasks", "audio_jobs", "translation_jobs"} {
+	for _, collection := range []string{"scraping_tasks", "audio_jobs", "translation_jobs", "import_tasks"} {
 		items, err := s.pb.listAll(ctx, collection, filter, "")
 		if err != nil {
 			return total, fmt.Errorf("ReapStaleTasks list %s: %w", collection, err)
@@ -1185,6 +1189,7 @@ func parseImportTask(raw json.RawMessage) (domain.ImportTask, error) {
 		FileName        string `json:"file_name"`
 		FileType        string `json:"file_type"`
 		ObjectKey       string `json:"object_key"`
+		ChaptersKey     string `json:"chapters_key"`
 		Author          string `json:"author"`
 		CoverURL        string `json:"cover_url"`
 		Genres          string `json:"genres"` // stored as comma-separated
@@ -1219,6 +1224,7 @@ func parseImportTask(raw json.RawMessage) (domain.ImportTask, error) {
 		FileName:        rec.FileName,
 		FileType:        rec.FileType,
 		ObjectKey:       rec.ObjectKey,
+		ChaptersKey:     rec.ChaptersKey,
 		Author:          rec.Author,
 		CoverURL:        rec.CoverURL,
 		Genres:          genres,
@@ -1266,6 +1272,20 @@ func (s *Store) PutImportFile(ctx context.Context, key string, data []byte) erro
 	return s.mc.putObject(ctx, "imports", key, "application/octet-stream", data)
 }
 
+// PutImportChapters stores a pre-parsed chapters JSON blob in MinIO.
+func (s *Store) PutImportChapters(ctx context.Context, key string, data []byte) error {
+	return s.mc.putObject(ctx, "imports", key, "application/json", data)
+}
+
+// GetImportChapters retrieves the pre-parsed chapters JSON from MinIO.
+func (s *Store) GetImportChapters(ctx context.Context, key string) ([]byte, error) {
+	data, err := s.mc.getObject(ctx, "imports", key)
+	if err != nil {
+		return nil, fmt.Errorf("get chapters object: %w", err)
+	}
+	return data, nil
+}
+
 func (s *Store) CoverExists(ctx context.Context, slug string) bool {
 	return s.mc.coverExists(ctx, CoverObjectKey(slug))
 }