hako/internal/jobs/worker.go

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
}