queue

package
v0.0.29-rc6 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Mar 3, 2026 License: MIT Imports: 10 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	StatusPending  = "pending"
	StatusRunning  = "running"
	StatusComplete = "complete"
	StatusError    = "error"
)

Variables

This section is empty.

Functions

This section is empty.

Types

type CompletedItem 

type CompletedItem struct {
	ID                   string     `json:"id"`
	Path                 string     `json:"path"`
	Size                 int64      `json:"size"`
	Priority             int        `json:"priority"`
	NzbPath              string     `json:"nzbPath"`
	CreatedAt            time.Time  `json:"createdAt"`
	CompletedAt          time.Time  `json:"completedAt"`
	JobData              []byte     `json:"jobData"`
	ScriptRetryCount     int        `json:"scriptRetryCount"`     // Number of script retry attempts
	ScriptFirstFailureAt *time.Time `json:"scriptFirstFailureAt"` // When the first script failure occurred
}

type FileJob 

type FileJob struct {
	Path       string    `json:"path"`
	Size       int64     `json:"size"`
	Priority   int       `json:"priority"`
	CreatedAt  time.Time `json:"createdAt"`
	RetryCount int       `json:"retryCount"`
}

type PaginatedResult added in v0.0.26 

type PaginatedResult struct {
	Items        []QueueItem `json:"items"`
	TotalItems   int         `json:"totalItems"`
	TotalPages   int         `json:"totalPages"`
	CurrentPage  int         `json:"currentPage"`
	ItemsPerPage int         `json:"itemsPerPage"`
	HasNext      bool        `json:"hasNext"`
	HasPrev      bool        `json:"hasPrev"`
}

PaginatedResult contains paginated queue items and metadata

type PaginationParams added in v0.0.26 

type PaginationParams struct {
	Page   int    `json:"page"`   // 1-based page number
	Limit  int    `json:"limit"`  // Items per page
	SortBy string `json:"sortBy"` // Sort field: "created", "priority", "status", "filename", "size"
	Order  string `json:"order"`  // Sort order: "asc", "desc"
	Status string `json:"status"` // Status filter: "pending", "complete", "error", or "" for all
}

PaginationParams defines parameters for paginated queries

type PendingArticleCheck added in v0.0.29 

type PendingArticleCheck struct {
	ID              int64
	CompletedItemID string
	MessageID       string
	Groups          string // JSON-encoded array of group names
	Status          string // pending, verified, failed
	RetryCount      int
	NextRetryAt     time.Time
	FirstFailureAt  time.Time
	LastCheckedAt   *time.Time
	CreatedAt       time.Time
}

PendingArticleCheck represents an article that needs deferred verification

type Queue 

type Queue struct {
	// contains filtered or unexported fields
}

func New 

func New(ctx context.Context, database *database.Database) (*Queue, error)

func (*Queue) AddFile 

func (q *Queue) AddFile(ctx context.Context, path string, size int64) error

AddFile adds a file to the queue for processing

func (*Queue) AddFileWithPriority 

func (q *Queue) AddFileWithPriority(ctx context.Context, path string, size int64, priority int) error

AddFileWithPriority adds a file to the queue with a specific priority

func (*Queue) AddFileWithPriorityWithoutDuplicateCheck 

func (q *Queue) AddFileWithPriorityWithoutDuplicateCheck(ctx context.Context, path string, size int64, priority int) error

AddFileWithPriorityWithoutDuplicateCheck adds a file to the queue with a specific priority without checking for duplicates This is useful when original files are deleted after processing, making duplicate checks unnecessary

func (*Queue) AddFileWithoutDuplicateCheck 

func (q *Queue) AddFileWithoutDuplicateCheck(ctx context.Context, path string, size int64) error

AddFileWithoutDuplicateCheck adds a file to the queue without checking for duplicates This is useful when original files are deleted after processing, making duplicate checks unnecessary

func (*Queue) AddPendingArticleChecks added in v0.0.29 

func (q *Queue) AddPendingArticleChecks(ctx context.Context, completedItemID string, articles []PendingArticleCheck) error

AddPendingArticleChecks inserts pending article checks for a completed item

func (*Queue) CancelFile added in v0.0.29 

func (q *Queue) CancelFile(ctx context.Context, msgID goqite.ID) error

CancelFile removes a cancelled job from in-progress tracking.

func (*Queue) ClearCompletedItems 

func (q *Queue) ClearCompletedItems() error

ClearCompletedItems removes only completed items and their NZB files

func (*Queue) ClearQueue 

func (q *Queue) ClearQueue() error

ClearQueue removes all completed, errored, and active items from the queue

func (*Queue) Close 

func (q *Queue) Close() error

Close closes the queue context (database is managed externally)

func (*Queue) CompleteFile 

func (q *Queue) CompleteFile(ctx context.Context, msgID goqite.ID, nzbPath string, job *FileJob) error

CompleteFile marks a file job as completed and adds it to completed_items table

func (*Queue) DebugQueueItem 

func (q *Queue) DebugQueueItem(id string) (map[string]any, error)

DebugQueueItem returns debug information about a specific queue item

func (*Queue) ExtendTimeout 

func (q *Queue) ExtendTimeout(ctx context.Context, msgID goqite.ID, duration time.Duration) error

ExtendTimeout extends the processing timeout for a file job

func (*Queue) GetArticlesForCheck added in v0.0.29 

func (q *Queue) GetArticlesForCheck(ctx context.Context, limit int) ([]PendingArticleCheck, error)

GetArticlesForCheck retrieves pending articles that are ready for checking

func (*Queue) GetCompletedItemNzbPath 

func (q *Queue) GetCompletedItemNzbPath(id string) (string, error)

GetCompletedItemNzbPath returns the NZB path for a completed item

func (*Queue) GetItemsForScriptRetry added in v0.0.29 

func (q *Queue) GetItemsForScriptRetry(ctx context.Context, limit int) ([]CompletedItem, error)

GetItemsForScriptRetry retrieves completed items that need script retry

func (*Queue) GetPendingCheckCountForItem added in v0.0.29 

func (q *Queue) GetPendingCheckCountForItem(ctx context.Context, completedItemID string) (total int, pending int, failed int, err error)

GetPendingCheckCountForItem returns the total, pending, and failed count of article checks for a completed item

func (*Queue) GetQueueItems 

func (q *Queue) GetQueueItems(params PaginationParams) (*PaginatedResult, error)

GetQueueItems returns paginated queue items with metadata

func (*Queue) GetQueueStats 

func (q *Queue) GetQueueStats() (map[string]any, error)

GetQueueStats returns statistics about the queue including completed and errored items

func (*Queue) IsPathInQueue added in v0.0.17 

func (q *Queue) IsPathInQueue(path string) (bool, error)

IsPathInQueue checks if a file path already exists in pending queue, completed items, or errored items

func (*Queue) MarkArticleCheckFailed added in v0.0.29 

func (q *Queue) MarkArticleCheckFailed(ctx context.Context, id int64) error

MarkArticleCheckFailed marks a pending article check as permanently failed

func (*Queue) MarkArticleVerified added in v0.0.29 

func (q *Queue) MarkArticleVerified(ctx context.Context, id int64) error

MarkArticleVerified marks a pending article check as verified

func (*Queue) MarkAsError 

func (q *Queue) MarkAsError(ctx context.Context, msgID goqite.ID, job *FileJob, errMsg string) error

MarkAsError marks a file job as errored and adds it to the errored_items table

func (*Queue) MarkScriptCompleted added in v0.0.29 

func (q *Queue) MarkScriptCompleted(ctx context.Context, itemID string) error

MarkScriptCompleted marks the script execution as completed for a completed item

func (*Queue) MarkScriptFailed added in v0.0.29 

func (q *Queue) MarkScriptFailed(ctx context.Context, itemID string, lastError string) error

MarkScriptFailed marks the script execution as permanently failed

func (*Queue) ReaddJob 

func (q *Queue) ReaddJob(ctx context.Context, job *FileJob) error

ReaddJob re-adds a job to the queue (used when processing fails)

func (*Queue) ReceiveFile 

func (q *Queue) ReceiveFile(ctx context.Context) (*goqite.Message, *FileJob, error)

ReceiveFile gets the next file job from the queue and removes it immediately. The item is tracked in the in_progress_items table until CompleteFile or MarkAsError is called. On startup, RecoverInProgressItems restores any items that were lost due to a crash.

func (*Queue) RemoveCompletedItem 

func (q *Queue) RemoveCompletedItem(id string) error

RemoveCompletedItem removes a completed item and its associated NZB file

func (*Queue) RemoveErroredItem 

func (q *Queue) RemoveErroredItem(id string) error

RemoveErroredItem removes an errored item from the database

func (*Queue) RemoveFromQueue 

func (q *Queue) RemoveFromQueue(id string) error

RemoveFromQueue removes an item from the queue by ID (handles active, completed, and errored items)

func (*Queue) RetryErroredJob 

func (q *Queue) RetryErroredJob(ctx context.Context, id string) error

RetryErroredJob retries an errored job by moving it back to the main queue

func (*Queue) SetQueueItemPriority 

func (q *Queue) SetQueueItemPriority(id string, priority int) error

SetQueueItemPriority updates the priority of a pending queue item by id

func (*Queue) SetQueueItemPriorityWithReorder added in v0.0.13 

func (q *Queue) SetQueueItemPriorityWithReorder(ctx context.Context, id string, newPriority int) error

SetQueueItemPriorityWithReorder updates the priority of a pending queue item using goqite's native priority Higher priority numbers (0, 1, 2, ...) are processed first

func (*Queue) UpdateArticleCheckRetry added in v0.0.29 

func (q *Queue) UpdateArticleCheckRetry(ctx context.Context, id int64, retryCount int, nextRetryAt time.Time) error

UpdateArticleCheckRetry updates a pending article check for the next retry

func (*Queue) UpdateCompletedItemVerificationStatus added in v0.0.29 

func (q *Queue) UpdateCompletedItemVerificationStatus(ctx context.Context, completedItemID string, status string) error

UpdateCompletedItemVerificationStatus updates the verification status of a completed item

func (*Queue) UpdateScriptStatus added in v0.0.29 

func (q *Queue) UpdateScriptStatus(ctx context.Context, itemID string, status string, retryCount int, lastError string, nextRetryAt *time.Time, firstFailureAt *time.Time) error

UpdateScriptStatus updates the script execution status for a completed item. If firstFailureAt is provided and it's the first failure (retryCount == 1), it will be recorded.

type QueueInterface added in v0.0.13 

type QueueInterface interface {
	AddFile(ctx context.Context, path string, size int64) error
	GetQueueItems(params PaginationParams) (*PaginatedResult, error)
	RemoveFromQueue(id string) error
	ClearQueue() error
	GetQueueStats() (map[string]any, error)
	SetQueueItemPriorityWithReorder(ctx context.Context, id string, newPriority int) error
	IsPathInQueue(path string) (bool, error)
}

QueueInterface defines the interface for queue operations

type QueueItem 

type QueueItem struct {
	ID           string     `json:"id"`
	Path         string     `json:"path"`
	FileName     string     `json:"fileName"`
	Size         int64      `json:"size"`
	Status       string     `json:"status"`
	RetryCount   int        `json:"retryCount"`
	Priority     int        `json:"priority"`
	ErrorMessage *string    `json:"errorMessage"`
	CreatedAt    time.Time  `json:"createdAt"`
	UpdatedAt    time.Time  `json:"updatedAt"`
	CompletedAt  *time.Time `json:"completedAt"`
	NzbPath      *string    `json:"nzbPath"`
	// Script execution tracking fields
	ScriptStatus      *string    `json:"scriptStatus"` // null, completed, pending_retry, failed_permanent
	ScriptRetryCount  int        `json:"scriptRetryCount"`
	ScriptLastError   *string    `json:"scriptLastError"`
	ScriptNextRetryAt *time.Time `json:"scriptNextRetryAt"`
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.