fetch_ml/internal/queue/filesystem/queue.go

// Package filesystem provides a filesystem-based queue implementation
package filesystem

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"

	"github.com/jfraeys/fetch_ml/internal/domain"
)

// Queue implements a filesystem-based task queue
type Queue struct {
	root   string
	ctx    context.Context
	cancel context.CancelFunc
}

type queueIndex struct {
	Version   int              `json:"version"`
	UpdatedAt string           `json:"updated_at"`
	Tasks     []queueIndexTask `json:"tasks"`
}

type queueIndexTask struct {
	ID        string `json:"id"`
	Priority  int64  `json:"priority"`
	CreatedAt string `json:"created_at"`
}

// NewQueue creates a new filesystem queue instance
func NewQueue(root string) (*Queue, error) {
	root = strings.TrimSpace(root)
	if root == "" {
		return nil, fmt.Errorf("filesystem queue root is required")
	}
	root = filepath.Clean(root)
	if err := os.MkdirAll(filepath.Join(root, "pending", "entries"), 0750); err != nil {
		return nil, err
	}
	for _, d := range []string{"running", "finished", "failed"} {
		if err := os.MkdirAll(filepath.Join(root, d), 0750); err != nil {
			return nil, err
		}
	}

	ctx, cancel := context.WithCancel(context.Background())
	q := &Queue{root: root, ctx: ctx, cancel: cancel}
	_ = q.rebuildIndex()
	return q, nil
}

// Close closes the queue
func (q *Queue) Close() error {
	q.cancel()
	return nil
}

// AddTask adds a task to the queue
func (q *Queue) AddTask(task *domain.Task) error {
	if task == nil {
		return errors.New("task is nil")
	}
	if task.ID == "" {
		return errors.New("task ID is required")
	}

	pendingDir := filepath.Join(q.root, "pending", "entries")
	taskFile := filepath.Join(pendingDir, task.ID+".json")

	data, err := json.Marshal(task)
	if err != nil {
		return fmt.Errorf("failed to marshal task: %w", err)
	}

	if err := os.WriteFile(taskFile, data, 0640); err != nil {
		return fmt.Errorf("failed to write task file: %w", err)
	}

	return nil
}

// GetTask retrieves a task by ID
func (q *Queue) GetTask(id string) (*domain.Task, error) {
	if id == "" {
		return nil, errors.New("task ID is required")
	}

	// Search in all directories
	for _, dir := range []string{"pending", "running", "finished", "failed"} {
		taskFile := filepath.Join(q.root, dir, "entries", id+".json")
		data, err := os.ReadFile(taskFile)
		if err == nil {
			var task domain.Task
			if err := json.Unmarshal(data, &task); err != nil {
				return nil, fmt.Errorf("failed to unmarshal task: %w", err)
			}
			return &task, nil
		}
	}

	return nil, fmt.Errorf("task not found: %s", id)
}

// ListTasks lists all tasks in the queue
func (q *Queue) ListTasks() ([]*domain.Task, error) {
	var tasks []*domain.Task

	for _, dir := range []string{"pending", "running", "finished", "failed"} {
		entriesDir := filepath.Join(q.root, dir, "entries")
		entries, err := os.ReadDir(entriesDir)
		if err != nil {
			continue
		}

		for _, entry := range entries {
			if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".json") {
				continue
			}

			data, err := os.ReadFile(filepath.Join(entriesDir, entry.Name()))
			if err != nil {
				continue
			}

			var task domain.Task
			if err := json.Unmarshal(data, &task); err != nil {
				continue
			}
			tasks = append(tasks, &task)
		}
	}

	return tasks, nil
}

// CancelTask cancels a task
func (q *Queue) CancelTask(id string) error {
	// Remove from pending if exists
	pendingFile := filepath.Join(q.root, "pending", "entries", id+".json")
	if _, err := os.Stat(pendingFile); err == nil {
		return os.Remove(pendingFile)
	}
	return nil
}

// UpdateTask updates a task
func (q *Queue) UpdateTask(task *domain.Task) error {
	if task == nil || task.ID == "" {
		return errors.New("task is nil or missing ID")
	}

	// Find current location
	var currentFile string
	for _, dir := range []string{"pending", "running", "finished", "failed"} {
		f := filepath.Join(q.root, dir, "entries", task.ID+".json")
		if _, err := os.Stat(f); err == nil {
			currentFile = f
			break
		}
	}

	if currentFile == "" {
		return fmt.Errorf("task not found: %s", task.ID)
	}

	data, err := json.Marshal(task)
	if err != nil {
		return fmt.Errorf("failed to marshal task: %w", err)
	}

	return os.WriteFile(currentFile, data, 0640)
}

// rebuildIndex rebuilds the queue index
func (q *Queue) rebuildIndex() error {
	// Implementation would rebuild the index file
	return nil
}