package jobs import ( "context" "encoding/json" "fmt" "log/slog" "path/filepath" "time" "git.nakama.town/fmartingr/hako/internal/archival/archiver" "git.nakama.town/fmartingr/hako/internal/extractors" "git.nakama.town/fmartingr/hako/internal/model" ) // ArchiveService defines the interface for archive operations type ArchiveService interface { ProcessArchive(ctx context.Context, archive *model.Archive, link *model.Link, archiverKeys []string) error } // ArchiveStore defines the interface for archive operations type ArchiveStore interface { GetByID(ctx context.Context, id string) (*model.Archive, error) UpdateStatus(ctx context.Context, id string, status model.ArchiveStatus, errorMessage string) error } // LinkStore defines the interface for link operations type LinkStore interface { GetByID(ctx context.Context, id string) (*model.Link, error) } // ArchiveFileStore defines the interface for archive file operations type ArchiveFileStore interface { GetByID(ctx context.Context, id string) (*model.ArchiveFile, error) UpdateContent(ctx context.Context, fileID string, content string, contentMimeType string) error Create(ctx context.Context, file *model.ArchiveFile) error } // Worker processes jobs from the queue type Worker struct { queue model.Queue archiverMgr *archiver.Manager extractorMgr *extractors.Manager archiveService ArchiveService archiveStore ArchiveStore archiveFileStore ArchiveFileStore linkStore LinkStore storageBasePath string logger *slog.Logger } // NewWorker creates a new worker func NewWorker(queue model.Queue, archiverMgr *archiver.Manager, extractorMgr *extractors.Manager, archiveService ArchiveService, archiveStore ArchiveStore, archiveFileStore ArchiveFileStore, linkStore LinkStore, storageBasePath string, logger *slog.Logger) *Worker { return &Worker{ queue: queue, archiverMgr: archiverMgr, extractorMgr: extractorMgr, archiveService: archiveService, archiveStore: archiveStore, archiveFileStore: archiveFileStore, linkStore: linkStore, storageBasePath: storageBasePath, logger: logger, } } // Start starts the worker func (w *Worker) Start(ctx context.Context) error { w.logger.Info("Starting job worker") for { select { case <-ctx.Done(): w.logger.Info("Job worker stopping") return nil default: job, err := w.queue.Dequeue(ctx) if err != nil { w.logger.Error("Failed to dequeue job", "error", err) time.Sleep(1 * time.Second) continue } if job == nil { time.Sleep(1 * time.Second) continue } w.processJob(ctx, job) } } } // processJob processes a single job func (w *Worker) processJob(ctx context.Context, job *model.Job) { w.logger.Info("Processing job", "id", job.ID, "type", job.Type) var err error switch job.Type { case model.JobTypeArchiveLink: err = w.processArchiveLink(ctx, job) case model.JobTypeExtractContent: err = w.processExtractContent(ctx, job) default: err = fmt.Errorf("unknown job type: %s", job.Type) } if err != nil { w.logger.Error("Job failed", "id", job.ID, "error", err) if err := w.queue.Fail(ctx, job.ID, err.Error()); err != nil { w.logger.Error("Failed to mark job as failed", "id", job.ID, "error", err) } return } if err := w.queue.Complete(ctx, job.ID); err != nil { w.logger.Error("Failed to mark job as complete", "id", job.ID, "error", err) } w.logger.Info("Job completed", "id", job.ID) } // processArchiveLink processes an archive link job func (w *Worker) processArchiveLink(ctx context.Context, job *model.Job) error { var payload model.ArchiveLinkPayload if err := json.Unmarshal([]byte(job.Payload), &payload); err != nil { return fmt.Errorf("failed to unmarshal payload: %w", err) } // Get the archive storeArchive, err := w.archiveStore.GetByID(ctx, payload.ArchiveID) if err != nil { return fmt.Errorf("failed to get archive: %w", err) } // Get the link storeLink, err := w.linkStore.GetByID(ctx, payload.LinkID) if err != nil { // Update archive status to failed _ = w.archiveStore.UpdateStatus(ctx, payload.ArchiveID, model.ArchiveStatusFailed, fmt.Sprintf("failed to get link: %v", err)) return fmt.Errorf("failed to get link: %w", err) } // Stores now return model types directly archive := storeArchive link := storeLink // Process the archive with archiver keys from payload (if provided) if err := w.archiveService.ProcessArchive(ctx, archive, link, payload.ArchiverKeys); err != nil { // Update archive status to failed _ = w.archiveStore.UpdateStatus(ctx, payload.ArchiveID, model.ArchiveStatusFailed, err.Error()) return fmt.Errorf("failed to process archive: %w", err) } return nil } // ExtractContentPayload is the payload for content extraction jobs type ExtractContentPayload struct { FileID string `json:"file_id"` } // processExtractContent processes a content extraction job func (w *Worker) processExtractContent(ctx context.Context, job *model.Job) error { var payload ExtractContentPayload if err := json.Unmarshal([]byte(job.Payload), &payload); err != nil { return fmt.Errorf("failed to unmarshal payload: %w", err) } // Get the archive file file, err := w.archiveFileStore.GetByID(ctx, payload.FileID) if err != nil { return fmt.Errorf("failed to get archive file: %w", err) } // Skip if no MIME type if file.MimeType == "" { w.logger.Debug("Skipping content extraction - no MIME type", "fileID", payload.FileID) return nil } // Find extractor for this MIME type extractor, found := w.extractorMgr.GetExtractorForMimeType(file.MimeType) if !found { w.logger.Debug("No extractor found for MIME type", "fileID", payload.FileID, "mimeType", file.MimeType) return nil // Not an error - just no extractor available } // Construct full file path fullPath := filepath.Join(w.storageBasePath, file.StoragePath) // Extract content content, err := extractor.Extract(ctx, fullPath) if err != nil { w.logger.Error("Failed to extract content", "fileID", payload.FileID, "error", err) return fmt.Errorf("failed to extract content: %w", err) } // Determine content MIME type based on extractor contentMimeType := "" if extractor.Key() == "readable" { contentMimeType = "text/markdown" } else if extractor.Key() == "pdf" { contentMimeType = "text/plain" } // Update archive file with extracted content and content MIME type if err := w.archiveFileStore.UpdateContent(ctx, payload.FileID, content, contentMimeType); err != nil { return fmt.Errorf("failed to update archive file content: %w", err) } w.logger.Info("Content extracted successfully", "fileID", payload.FileID, "contentLength", len(content), "contentMimeType", contentMimeType) return nil }