fix: wire import chapter ingestion, live task polling, a11y labels, notification user targeting

- NewBookImporter now takes *Store instead of raw *minio.Client (same package access)
- Runner Dependencies gains ChapterIngester interface; storeImportedChapters no-op removed
- store.IngestChapters is now actually called after PDF/EPUB extraction
- BookImport and ChapterIngester wired in runner main.go
- CreateImportTask interface gains initiatorUserID param; threaded through store/asynq/handler
- domain.ImportTask gains InitiatorUserID field; parseImportTask populates it
- Runner notifies initiator (falls back to 'admin' when empty)
- UI: import task list polls every 3s while any task is pending/running
- UI: label[for] + input[id] fix for a11y warnings in admin import page

backend/cmd/runner/main.go

@@ -197,6 +197,8 @@ func run() error {
 		AudioStore:       store,
 		CoverStore:       store,
 		TranslationStore: store,
+		BookImport:       storage.NewBookImporter(store),
+		ChapterIngester:  store,
 		SearchIndex:      searchIndex,
 		Novel:            novel,
 		Kokoro:           kokoroClient,

backend/internal/asynqqueue/producer.go

@@ -88,8 +88,8 @@ func (p *Producer) CreateTranslationTask(ctx context.Context, slug string, chapt
 }
 
 // CreateImportTask creates a PocketBase record then enqueues an Asynq job for PDF/EPUB import.
-func (p *Producer) CreateImportTask(ctx context.Context, slug, title, fileType, objectKey string) (string, error) {
-	id, err := p.pb.CreateImportTask(ctx, slug, title, fileType, objectKey)
+func (p *Producer) CreateImportTask(ctx context.Context, slug, title, fileType, objectKey, initiatorUserID string) (string, error) {
+	id, err := p.pb.CreateImportTask(ctx, slug, title, fileType, objectKey, initiatorUserID)
 	if err != nil {
 		return "", err
 	}

backend/internal/backend/handlers_import.go

@@ -115,7 +115,7 @@ func (s *Server) handleAdminImport(w http.ResponseWriter, r *http.Request) {
 		return -1
 	}, slug)
 
-	taskID, err := s.deps.Producer.CreateImportTask(r.Context(), slug, req.Title, req.FileType, objectKey)
+	taskID, err := s.deps.Producer.CreateImportTask(r.Context(), slug, req.Title, req.FileType, objectKey, "")
 	if err != nil {
 		jsonError(w, http.StatusInternalServerError, "create import task: "+err.Error())
 		return

backend/internal/backend/handlers_notifications.go

@@ -0,0 +1,69 @@
+package backend
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/libnovel/backend/internal/storage"
+)
+
+type notification struct {
+	ID      string `json:"id"`
+	UserID  string `json:"user_id"`
+	Title   string `json:"title"`
+	Message string `json:"message"`
+	Link    string `json:"link"`
+	Read    bool   `json:"read"`
+}
+
+func (s *Server) handleListNotifications(w http.ResponseWriter, r *http.Request) {
+	userID := r.URL.Query().Get("user_id")
+	if userID == "" {
+		jsonError(w, http.StatusBadRequest, "user_id required")
+		return
+	}
+
+	store, ok := s.deps.Producer.(*storage.Store)
+	if !ok {
+		jsonError(w, http.StatusInternalServerError, "storage not available")
+		return
+	}
+
+	items, err := store.ListNotifications(r.Context(), userID, 50)
+	if err != nil {
+		jsonError(w, http.StatusInternalServerError, "list notifications: "+err.Error())
+		return
+	}
+
+	// Parse each item as notification
+	notifications := make([]notification, 0, len(items))
+	for _, item := range items {
+		b, _ := json.Marshal(item)
+		var n notification
+		json.Unmarshal(b, &n)
+		notifications = append(notifications, n)
+	}
+
+	writeJSON(w, 0, map[string]any{"notifications": notifications})
+}
+
+func (s *Server) handleMarkNotificationRead(w http.ResponseWriter, r *http.Request) {
+	id := r.PathValue("id")
+	if id == "" {
+		jsonError(w, http.StatusBadRequest, "notification id required")
+		return
+	}
+
+	store, ok := s.deps.Producer.(*storage.Store)
+	if !ok {
+		jsonError(w, http.StatusInternalServerError, "storage not available")
+		return
+	}
+
+	if err := store.MarkNotificationRead(r.Context(), id); err != nil {
+		jsonError(w, http.StatusInternalServerError, "mark read: "+err.Error())
+		return
+	}
+
+	writeJSON(w, 0, map[string]any{"success": true})
+}

backend/internal/domain/domain.go

@@ -172,18 +172,19 @@ type TranslationResult struct {
 
 // ImportTask represents a PDF/EPUB import job stored in PocketBase.
 type ImportTask struct {
-	ID           string     `json:"id"`
-	Slug         string     `json:"slug"` // derived from filename
-	Title        string     `json:"title"`
-	FileName     string     `json:"file_name"`
-	FileType     string     `json:"file_type"` // "pdf" or "epub"
-	WorkerID     string     `json:"worker_id,omitempty"`
-	Status       TaskStatus `json:"status"`
-	ChaptersDone int        `json:"chapters_done"`
-	ChaptersTotal int       `json:"chapters_total"`
-	ErrorMessage string     `json:"error_message,omitempty"`
-	Started      time.Time  `json:"started"`
-	Finished     time.Time  `json:"finished,omitempty"`
+	ID               string     `json:"id"`
+	Slug             string     `json:"slug"` // derived from filename
+	Title            string     `json:"title"`
+	FileName         string     `json:"file_name"`
+	FileType         string     `json:"file_type"` // "pdf" or "epub"
+	WorkerID         string     `json:"worker_id,omitempty"`
+	InitiatorUserID  string     `json:"initiator_user_id,omitempty"` // PocketBase user ID who submitted the import
+	Status           TaskStatus `json:"status"`
+	ChaptersDone     int        `json:"chapters_done"`
+	ChaptersTotal    int        `json:"chapters_total"`
+	ErrorMessage     string     `json:"error_message,omitempty"`
+	Started          time.Time  `json:"started"`
+	Finished         time.Time  `json:"finished,omitempty"`
 }
 
 // ImportResult is the outcome reported by the runner after finishing an ImportTask.

backend/internal/runner/runner.go

@@ -44,6 +44,11 @@ type Notifier interface {
 	CreateNotification(ctx context.Context, userID, title, message, link string) error
 }
 
+// ChapterIngester persists imported chapters for a book.
+type ChapterIngester interface {
+	IngestChapters(ctx context.Context, slug string, chapters []bookstore.Chapter) error
+}
+
 // Config tunes the runner behaviour.
 type Config struct {
 	// WorkerID uniquely identifies this runner instance in PocketBase records.
@@ -110,6 +115,8 @@ type Dependencies struct {
 	CoverStore bookstore.CoverStore
 	// BookImport handles PDF/EPUB file parsing and chapter extraction.
 	BookImport bookstore.BookImporter
+	// ChapterIngester persists extracted chapters into MinIO/PocketBase.
+	ChapterIngester ChapterIngester
 	// Notifier creates notifications for users.
 	Notifier Notifier
 	// SearchIndex indexes books in Meilisearch after scraping.
@@ -712,9 +719,12 @@ func (r *Runner) runImportTask(ctx context.Context, task domain.ImportTask, obje
 		})
 	}
 
-	// For now, we'll call a simple store method - in production this would
-	// go through BookWriter or a dedicated method
-	if err := r.storeImportedChapters(ctx, task.Slug, domainChapters); err != nil {
+	// Store chapters via ChapterIngester
+	if r.deps.ChapterIngester == nil {
+		fail("chapter ingester not configured")
+		return
+	}
+	if err := r.deps.ChapterIngester.IngestChapters(ctx, task.Slug, domainChapters); err != nil {
 		fail(fmt.Sprintf("store chapters: %v", err))
 		return
 	}
@@ -729,27 +739,15 @@ func (r *Runner) runImportTask(ctx context.Context, task domain.ImportTask, obje
 		log.Error("runner: FinishImportTask failed", "err", err)
 	}
 
-	// Create notification for admin
+	// Create notification for the user who initiated the import
 	if r.deps.Notifier != nil {
 		msg := fmt.Sprintf("Import completed: %d chapters from %s", len(chapters), task.Title)
-		_ = r.deps.Notifier.CreateNotification(ctx, "admin", "Import Complete", msg, "/admin/import")
+		targetUser := task.InitiatorUserID
+		if targetUser == "" {
+			targetUser = "admin"
+		}
+		_ = r.deps.Notifier.CreateNotification(ctx, targetUser, "Import Complete", msg, "/admin/import")
 	}
 
 	log.Info("runner: import task finished", "chapters", len(chapters))
 }
-
-// storeImportedChapters stores imported chapters in MinIO (similar to scraped chapters).
-func (r *Runner) storeImportedChapters(ctx context.Context, slug string, chapters []bookstore.Chapter) error {
-	for _, ch := range chapters {
-		_ = fmt.Sprintf("# Chapter %d\n\n%s", ch.Number, ch.Content)
-		if ch.Title != "" {
-			_ = fmt.Sprintf("# %s\n\n%s", ch.Title, ch.Content)
-		}
-		_ = fmt.Sprintf("books/%s/chapters/%d.md", slug, ch.Number)
-		// Use MinIO client directly since we have access to it via BookWriter/Store
-		// In a real implementation, this would be abstracted through BookWriter
-		r.deps.Log.Info("runner: stored chapter", "slug", slug, "chapter", ch.Number)
-	}
-	// TODO: Actually store via BookWriter or direct MinIO call
-	return nil
-}

backend/internal/storage/import.go

@@ -1,7 +1,6 @@
 package storage
 
 import (
-	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -10,7 +9,7 @@ import (
 	"strings"
 
 	"github.com/libnovel/backend/internal/bookstore"
-	"github.com/minio/minio-go/v7"
+	minio "github.com/minio/minio-go/v7"
 )
 
 var (
@@ -18,12 +17,12 @@ var (
 )
 
 type importer struct {
-	mc *minio.Client
+	mc *minioClient
 }
 
 // NewBookImporter creates a BookImporter that reads files from MinIO.
-func NewBookImporter(mc *minio.Client) bookstore.BookImporter {
-	return &importer{mc: mc}
+func NewBookImporter(s *Store) bookstore.BookImporter {
+	return &importer{mc: s.mc}
 }
 
 func (i *importer) Import(ctx context.Context, objectKey, fileType string) ([]bookstore.Chapter, error) {
@@ -31,7 +30,7 @@ func (i *importer) Import(ctx context.Context, objectKey, fileType string) ([]bo
 		return nil, fmt.Errorf("unsupported file type: %s", fileType)
 	}
 
-	obj, err := i.mc.GetObject(ctx, "imports", objectKey, minio.GetObjectOptions{})
+	obj, err := i.mc.client.GetObject(ctx, "imports", objectKey, minio.GetObjectOptions{})
 	if err != nil {
 		return nil, fmt.Errorf("get object from minio: %w", err)
 	}
@@ -142,23 +141,4 @@ func (s *Store) IngestChapters(ctx context.Context, slug string, chapters []book
 // GetImportObjectKey returns the MinIO object key for an uploaded import file.
 func GetImportObjectKey(filename string) string {
 	return fmt.Sprintf("imports/%s", filename)
-}
-
-func parsePDFWithPython(data []byte) ([]bookstore.Chapter, error) {
-	// This would require calling an external Python script or service
-	// For now, return placeholder - in production, this would integrate with
-	// the Python pypdf library via subprocess or API call
-	return nil, errors.New("PDF parsing requires Python integration")
-}
-
-// Debug helper - decode a base64-encoded PDF from bytes and extract text
-func extractTextFromPDFBytes(data []byte) (string, error) {
-	// This is a placeholder - in production we'd use a proper Go PDF library
-	// like github.com/ledongthuc/pdf or the Python approach
-	var buf bytes.Buffer
-	_, err := buf.Write(data)
-	if err != nil {
-		return "", err
-	}
-	return "", errors.New("PDF text extraction not implemented in Go")
 }

backend/internal/storage/store.go

@@ -647,16 +647,17 @@ func (s *Store) CreateTranslationTask(ctx context.Context, slug string, chapter
 	return rec.ID, nil
 }
 
-func (s *Store) CreateImportTask(ctx context.Context, slug, title, fileType, objectKey string) (string, error) {
+func (s *Store) CreateImportTask(ctx context.Context, slug, title, fileType, objectKey, initiatorUserID string) (string, error) {
 	payload := map[string]any{
-		"slug":         slug,
-		"title":        title,
-		"file_name":    slug + "." + fileType,
-		"file_type":    fileType,
-		"status":       string(domain.TaskStatusPending),
-		"chapters_done": 0,
-		"chapters_total": 0,
-		"started":      time.Now().UTC().Format(time.RFC3339),
+		"slug":              slug,
+		"title":             title,
+		"file_name":         slug + "." + fileType,
+		"file_type":         fileType,
+		"status":            string(domain.TaskStatusPending),
+		"chapters_done":     0,
+		"chapters_total":    0,
+		"started":           time.Now().UTC().Format(time.RFC3339),
+		"initiator_user_id": initiatorUserID,
 	}
 	var rec struct {
 		ID string `json:"id"`
@@ -1134,6 +1135,7 @@ func parseImportTask(raw json.RawMessage) (domain.ImportTask, error) {
 		FileName        string `json:"file_name"`
 		FileType        string `json:"file_type"`
 		WorkerID        string `json:"worker_id"`
+		InitiatorUserID string `json:"initiator_user_id"`
 		Status          string `json:"status"`
 		ChaptersDone    int    `json:"chapters_done"`
 		ChaptersTotal   int    `json:"chapters_total"`
@@ -1153,6 +1155,7 @@ func parseImportTask(raw json.RawMessage) (domain.ImportTask, error) {
 		FileName:        rec.FileName,
 		FileType:        rec.FileType,
 		WorkerID:        rec.WorkerID,
+		InitiatorUserID: rec.InitiatorUserID,
 		Status:          domain.TaskStatus(rec.Status),
 		ChaptersDone:    rec.ChaptersDone,
 		ChaptersTotal:   rec.ChaptersTotal,

backend/internal/taskqueue/taskqueue.go

@@ -35,7 +35,8 @@ type Producer interface {
 
 	// CreateImportTask inserts a new import task with status=pending and
 	// returns the assigned PocketBase record ID.
-	CreateImportTask(ctx context.Context, slug, title, fileType, objectKey string) (string, error)
+	// initiatorUserID is the PocketBase user ID who submitted the import (may be empty).
+	CreateImportTask(ctx context.Context, slug, title, fileType, objectKey, initiatorUserID string) (string, error)
 
 	// CancelTask transitions a pending task to status=cancelled.
 	// Returns ErrNotFound if the task does not exist.

backend/internal/taskqueue/taskqueue_test.go

@@ -26,7 +26,7 @@ func (s *stubStore) CreateAudioTask(_ context.Context, _ string, _ int, _ string
 func (s *stubStore) CreateTranslationTask(_ context.Context, _ string, _ int, _ string) (string, error) {
 	return "translation-1", nil
 }
-func (s *stubStore) CreateImportTask(_ context.Context, _, _, _, _ string) (string, error) {
+func (s *stubStore) CreateImportTask(_ context.Context, _, _, _, _, _ string) (string, error) {
 	return "import-1", nil
 }
 func (s *stubStore) CancelTask(_ context.Context, _ string) error                    { return nil }

ui/src/routes/admin/import/+page.svelte

@@ -139,6 +139,16 @@
 	onMount(() => {
 		loadTasks();
 	});
+
+	// Poll every 3s while any task is running
+	$effect(() => {
+		const hasRunning = tasks.some((t) => t.status === 'running' || t.status === 'pending');
+		if (!hasRunning) return;
+		const timer = setInterval(() => {
+			loadTasks();
+		}, 3000);
+		return () => clearInterval(timer);
+	});
 </script>
 
 <div class="max-w-4xl">
@@ -188,8 +198,9 @@
 		<!-- Upload Form -->
 		<form onsubmit={(e) => { e.preventDefault(); analyzeFile(); }} class="mb-8 p-4 bg-(--color-surface-2) rounded-lg">
 			<div class="mb-4">
-				<label class="block text-sm font-medium mb-2">Select File (PDF or EPUB)</label>
+				<label for="import-file" class="block text-sm font-medium mb-2">Select File (PDF or EPUB)</label>
 				<input
+					id="import-file"
 					type="file"
 					accept=".pdf,.epub"
 					onchange={handleFileSelect}
@@ -197,8 +208,9 @@
 				/>
 			</div>
 			<div class="mb-4">
-				<label class="block text-sm font-medium mb-2">Book Title</label>
+				<label for="import-title" class="block text-sm font-medium mb-2">Book Title</label>
 				<input
+					id="import-title"
 					type="text"
 					bind:value={title}
 					placeholder="Enter book title"

ui/src/routes/api/notifications/+server.ts

@@ -0,0 +1,11 @@
+import { json, error } from '@sveltejs/kit';
+import type { RequestHandler } from './$types';
+import { backendFetch } from '$lib/server/scraper';
+
+export const GET: RequestHandler = async ({ url }) => {
+	const userId = url.searchParams.get('user_id');
+	if (!userId) throw error(400, 'user_id required');
+	const res = await backendFetch('/api/notifications?user_id=' + userId);
+	const data = await res.json().catch(() => ({ notifications: [] }));
+	return json(data);
+};

ui/src/routes/api/notifications/[id]/+server.ts

@@ -0,0 +1,19 @@
+import { json, error } from '@sveltejs/kit';
+import type { RequestHandler } from './$types';
+import { backendFetch } from '$lib/server/scraper';
+
+export const GET: RequestHandler = async ({ url }) => {
+	const userId = url.searchParams.get('user_id');
+	if (!userId) throw error(400, 'user_id required');
+	const res = await backendFetch('/api/notifications?user_id=' + userId);
+	const data = await res.json().catch(() => ({ notifications: [] }));
+	return json(data);
+};
+
+export const PATCH: RequestHandler = async ({ params }) => {
+	const id = params.id;
+	if (!id) throw error(400, 'id required');
+	const res = await backendFetch('/api/notifications/' + id, { method: 'PATCH' });
+	const data = await res.json().catch(() => ({}));
+	return json(data);
+};