diff --git a/backend/cmd/runner/main.go b/backend/cmd/runner/main.go index 5559acd..070c9fc 100644 --- a/backend/cmd/runner/main.go +++ b/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, diff --git a/backend/internal/asynqqueue/producer.go b/backend/internal/asynqqueue/producer.go index 94604b7..b4edc47 100644 --- a/backend/internal/asynqqueue/producer.go +++ b/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 } diff --git a/backend/internal/backend/handlers_import.go b/backend/internal/backend/handlers_import.go index 3fef60c..2c4e276 100644 --- a/backend/internal/backend/handlers_import.go +++ b/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 diff --git a/backend/internal/backend/handlers_notifications.go b/backend/internal/backend/handlers_notifications.go new file mode 100644 index 0000000..0c071e9 --- /dev/null +++ b/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}) +} \ No newline at end of file diff --git a/backend/internal/domain/domain.go b/backend/internal/domain/domain.go index 3947cbc..8b8b1d6 100644 --- a/backend/internal/domain/domain.go +++ b/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. diff --git a/backend/internal/runner/runner.go b/backend/internal/runner/runner.go index 3592d4b..1f1560a 100644 --- a/backend/internal/runner/runner.go +++ b/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 -} diff --git a/backend/internal/storage/import.go b/backend/internal/storage/import.go index 5f62788..8cbda43 100644 --- a/backend/internal/storage/import.go +++ b/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") } \ No newline at end of file diff --git a/backend/internal/storage/store.go b/backend/internal/storage/store.go index dd6944d..db570dc 100644 --- a/backend/internal/storage/store.go +++ b/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, diff --git a/backend/internal/taskqueue/taskqueue.go b/backend/internal/taskqueue/taskqueue.go index 067413e..78fa2d8 100644 --- a/backend/internal/taskqueue/taskqueue.go +++ b/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. diff --git a/backend/internal/taskqueue/taskqueue_test.go b/backend/internal/taskqueue/taskqueue_test.go index de87bd7..4ccab18 100644 --- a/backend/internal/taskqueue/taskqueue_test.go +++ b/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 } diff --git a/ui/src/routes/admin/import/+page.svelte b/ui/src/routes/admin/import/+page.svelte index 2893fbc..da072cf 100644 --- a/ui/src/routes/admin/import/+page.svelte +++ b/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); + });
@@ -188,8 +198,9 @@
{ e.preventDefault(); analyzeFile(); }} class="mb-8 p-4 bg-(--color-surface-2) rounded-lg">
- +
- + { + 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); +}; \ No newline at end of file diff --git a/ui/src/routes/api/notifications/[id]/+server.ts b/ui/src/routes/api/notifications/[id]/+server.ts new file mode 100644 index 0000000..428f1d8 --- /dev/null +++ b/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); +}; \ No newline at end of file