importer

package
v0.4.0 Latest Latest
Warning

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

 Go to latest Published: Nov 3, 2026 License: MIT Imports: 60 Imported by: 0

Documentation

Overview

Package importer provides import queue processing for NZB files.

Index

Constants

View Source 
const (
	ScanStatusIdle      = scanner.ScanStatusIdle
	ScanStatusScanning  = scanner.ScanStatusScanning
	ScanStatusCanceling = scanner.ScanStatusCanceling
	ImportStatusIdle    = scanner.ImportStatusIdle
	ImportStatusRunning = scanner.ImportStatusRunning
)

Re-export scanner status constants for backward compatibility

Variables

View Source 
var (
	// NewNonRetryableError creates a new non-retryable error with a message and optional cause.
	NewNonRetryableError = sharedErrors.NewNonRetryableError

	// WrapNonRetryable wraps an existing error as non-retryable.
	WrapNonRetryable = sharedErrors.WrapNonRetryable

	// IsNonRetryable checks if an error is non-retryable.
	IsNonRetryable = sharedErrors.IsNonRetryable

	// ErrFallbackNotConfigured indicates that SABnzbd fallback is not enabled or configured.
	ErrFallbackNotConfigured = sharedErrors.ErrFallbackNotConfigured

	// ErrArticlesNotFound indicates that one or more NZB segments could not be found on any provider.
	ErrArticlesNotFound = sharedErrors.ErrArticlesNotFound
)
View Source 
var ErrContentProbeInconclusive = errors.New("content verification inconclusive")

ErrContentProbeInconclusive means the media content probe could not reach a verdict because of a transient provider, timeout, or connection error. Nothing is known about the written content, so the release must not be reported to the Arr app or counted towards its blocklist breaker.

View Source 
var ErrDeferredForRepair = errors.New("import deferred pending PAR2 repair")

ErrDeferredForRepair reports that an import was parked pending a PAR2 repair rather than failed. Archive-set volumes with missing articles cannot be zero-filled — a holed volume breaks extraction — but PAR2 can rebuild them byte-exactly, so dropping the release outright throws away a repair that is most likely to succeed right now, while the recovery volumes are still retrievable.

Functions

This section is empty.

Types

type DeferredRepairError 

type DeferredRepairError struct {
	FirstMissingSegmentID string
}

DeferredRepairError carries the deferral out of the fast-fail sweep together with the segment that proves the damage, so the repair plan has a starting point.

func (*DeferredRepairError) Error 

func (e *DeferredRepairError) Error() string

func (*DeferredRepairError) Unwrap 

func (e *DeferredRepairError) Unwrap() error

type DirectoryScanner 

type DirectoryScanner interface {
	// StartManualScan begins scanning a directory for NZB files
	StartManualScan(scanPath string) error
	// GetScanStatus returns the current scan status
	GetScanStatus() ScanInfo
	// CancelScan cancels an in-progress scan
	CancelScan() error
}

DirectoryScanner provides manual directory scanning functionality

type HistoryRecorder 

type HistoryRecorder interface {
	// AddImportHistory records a successful file import
	AddImportHistory(ctx context.Context, history *database.ImportHistory) error
}

HistoryRecorder records successful import events in persistent storage

type ImportInfo 

type ImportInfo = scanner.ImportInfo

Type aliases from scanner package for backward compatibility

type ImportJobStatus 

type ImportJobStatus = scanner.ImportJobStatus

Type aliases from scanner package for backward compatibility

type ImportService 

type ImportService interface {
	QueueManager
	DirectoryScanner
	NzbDavImporter
	QueueOperations

	// Close releases all resources
	Close() error
	// SetRcloneClient sets the rclone client for VFS notifications
	SetRcloneClient(client any)
	// SetArrsService sets the ARRs service for notifications
	SetArrsService(service any)
	// RegisterConfigChangeHandler registers a handler for configuration changes
	RegisterConfigChangeHandler(configManager any)
	// RegenerateMetadata attempts to rebuild metadata for a file by finding its original NZB
	RegenerateMetadata(ctx context.Context, mountRelativePath string) error
}

ImportService is the main interface combining all importer capabilities

type NonRetryableError 

type NonRetryableError = sharedErrors.NonRetryableError

Re-export error types and functions from shared errors package for backward compatibility with existing code.

type NzbDavImporter 

type NzbDavImporter interface {
	// StartNzbdavImport begins importing from an NzbDav database
	StartNzbdavImport(dbPath string, blobsPath string, cleanupFile bool) error
	// GetImportStatus returns the current import status
	GetImportStatus() ImportInfo
	// CancelImport cancels an in-progress import
	CancelImport() error
}

NzbDavImporter handles bulk import from NzbDav databases

type NzbRepairEnqueuer 

type NzbRepairEnqueuer interface {
	EnqueueNzb(ctx context.Context, nzbPath string, failingSegmentID string)
}

NzbRepairEnqueuer queues a repair planned from an NZB, for releases deferred before they were ever imported.

type Processor 

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

Processor handles the processing and storage of parsed NZB files using metadata storage

func NewProcessor 

func NewProcessor(metadataService *metadata.MetadataService, poolManager pool.Manager, broadcaster *progress.ProgressBroadcaster, configGetter config.ConfigGetter, recorder HistoryRecorder) *Processor

NewProcessor creates a new NZB processor using metadata storage

func (*Processor) ProcessNzbFile 

func (proc *Processor) ProcessNzbFile(ctx context.Context, filePath, relativePath string, queueID int, allowedExtensionsOverride *[]string, virtualDirOverride *string, extractedFiles []parser.ExtractedFileInfo, category *string, metadata *string, downloadID *string) (string, []string, error)

ProcessNzbFile processes an NZB or STRM file maintaining the folder structure relative to relative path. Returns (resultPath, writtenMetadataPaths, error). writtenMetadataPaths contains all virtual paths of metadata files written to disk; it is populated even on partial failure so callers can clean up. Paths prefixed with "DIR:" indicate a metadata directory that should be removed entirely.

func (*Processor) SetPatchIndex 

func (proc *Processor) SetPatchIndex(idx validation.PatchIndex)

SetPatchIndex wires the PAR2 patch store so articles repaired locally count as available during the fast-fail availability sweep.

func (*Processor) SetRecorder 

func (proc *Processor) SetRecorder(recorder HistoryRecorder)

func (*Processor) SetRepairEnqueuer 

func (proc *Processor) SetRepairEnqueuer(re RepairEnqueuer)

SetRepairEnqueuer wires the PAR2 repair queue. Call during boot, before imports run.

func (*Processor) SetSegmentStore 

func (proc *Processor) SetSegmentStore(resolve func() usenet.SegmentStore)

SetSegmentStore publishes first articles fetched at import to the streaming segment store, so the cold open right after an import is a cache hit.

type QueueManager 

type QueueManager interface {
	// Start begins processing queue items with the configured number of workers
	Start(ctx context.Context) error
	// Stop gracefully stops all workers and waits for completion
	Stop(ctx context.Context) error
	// Pause temporarily stops processing new items (workers remain active)
	Pause()
	// Resume continues processing after a pause
	Resume()
	// IsPaused returns whether the queue is currently paused
	IsPaused() bool
	// IsRunning returns whether the queue manager is active
	IsRunning() bool
	// CancelProcessing cancels processing of a specific item
	CancelProcessing(itemID int64) error
	// ProcessItemInBackground starts processing a specific item in the background
	ProcessItemInBackground(ctx context.Context, itemID int64)
}

QueueManager manages the import queue worker lifecycle

type QueueOperations 

type QueueOperations interface {
	// AddToQueue adds an item to the import queue
	AddToQueue(ctx context.Context, filePath string, relativePath *string, category *string, priority *database.QueuePriority, metadata *string, downloadID *string, indexer *string) (*database.ImportQueueItem, error)
	// GetQueueStats returns queue statistics
	GetQueueStats(ctx context.Context) (*database.QueueStats, error)
}

QueueOperations provides queue manipulation operations

type RSSPoller 

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

RSSPoller imports unseen cart entries through the normal import queue. Run and Poll serialize so a slow feed cannot overlap with its next poll.

func NewRSSPoller 

func NewRSSPoller(service *Service, getter config.ConfigGetter) *RSSPoller

func (*RSSPoller) Poll 

func (p *RSSPoller) Poll(ctx context.Context) error

Poll retrieves one feed snapshot. Failed entries remain eligible on the next poll; one bad entry does not prevent the remaining entries from importing.

func (*RSSPoller) Run 

func (p *RSSPoller) Run(ctx context.Context)

Run checks for settings changes every second. Polls happen immediately on startup/enabling/changing the feed, then at the configured interval.

type RepairEnqueuer 

type RepairEnqueuer interface {
	Enqueue(ctx context.Context, filePath string, failingSegmentID string)
}

RepairEnqueuer queues a file for background PAR2 repair (implemented by par2repair.Service). Implementations must be non-blocking.

type ScanInfo 

type ScanInfo = scanner.ScanInfo

Type aliases from scanner package for backward compatibility

type ScanStatus 

type ScanStatus = scanner.ScanStatus

Type aliases from scanner package for backward compatibility

type Service 

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

Service provides NZB import functionality with manual directory scanning and queue-based processing

func NewService 

func NewService(config ServiceConfig, metadataService *metadata.MetadataService, database *database.DB, poolManager pool.Manager, rcloneClient rclonecli.RcloneRcClient, configGetter config.ConfigGetter, healthRepo *database.HealthRepository, broadcaster *progress.ProgressBroadcaster, userRepo *database.UserRepository) (*Service, error)

NewService creates a new NZB import service with manual scanning and queue processing capabilities

func (*Service) AddImportHistory 

func (s *Service) AddImportHistory(ctx context.Context, history *database.ImportHistory) error

AddImportHistory records a successful file import in persistent history

func (*Service) AddToQueue 

func (s *Service) AddToQueue(ctx context.Context, filePath string, relativePath *string, category *string, priority *database.QueuePriority, metadata *string, downloadID *string, indexer *string) (*database.ImportQueueItem, error)

AddToQueue adds a new NZB file to the import queue.

func (*Service) CalculateFileSizeOnly 

func (s *Service) CalculateFileSizeOnly(filePath string) (int64, error)

CalculateFileSizeOnly calculates the total file size from NZB/STRM segments This is a lightweight parser that only extracts size information without full processing

func (*Service) CancelImport 

func (s *Service) CancelImport() error

CancelImport cancels the current import operation

func (*Service) CancelProcessing 

func (s *Service) CancelProcessing(itemID int64) error

CancelProcessing cancels a processing queue item by cancelling its context

func (*Service) CancelScan 

func (s *Service) CancelScan() error

CancelScan cancels the current scan operation

func (*Service) Close 

func (s *Service) Close() error

Close closes the NZB import service and releases all resources

func (*Service) ExecuteItem 

func (s *Service) ExecuteItem(ctx context.Context, itemID int64) error

ExecuteItem manually triggers processing for a specific queue item, bypassing concurrency limits.

func (*Service) FailWaitingRepair 

func (s *Service) FailWaitingRepair(ctx context.Context, nzbPath string, reason string) error

FailWaitingRepair fails a parked import whose repair proved impossible, which is the outcome it would have had before deferral existed.

func (*Service) FindAndUpdatePendingUpload 

func (s *Service) FindAndUpdatePendingUpload(ctx context.Context, filename string, category *string, priority *database.QueuePriority) (*database.ImportQueueItem, error)

FindAndUpdatePendingUpload de-duplicates manual NZB re-uploads. UNIQUE(nzb_path) can't catch them because ensurePersistentNzb rewrites the path after insert, so this matches a still-pending item by its category-independent base filename and updates its category/priority in place. Returns nil if none.

func (*Service) GetFailedNzbFolder 

func (s *Service) GetFailedNzbFolder() string

GetFailedNzbFolder returns the path to the directory for failed NZB files

func (*Service) GetGrabbedIndexer 

func (s *Service) GetGrabbedIndexer(downloadID string, releaseTitle string) (string, bool)

GetGrabbedIndexer retrieves a grabbed indexer by download ID or release title It only returns a match if it was stored within the last 24 hours.

func (*Service) GetImportStatus 

func (s *Service) GetImportStatus() ImportInfo

GetImportStatus returns the current import status

func (*Service) GetNzbFolder 

func (s *Service) GetNzbFolder() string

GetNzbFolder returns the path to the persistent NZB storage directory

func (*Service) GetPostProcessor 

func (s *Service) GetPostProcessor() *postprocessor.Coordinator

GetPostProcessor returns the post-processor coordinator

func (*Service) GetQueueStats 

func (s *Service) GetQueueStats(ctx context.Context) (*database.QueueStats, error)

GetQueueStats returns current queue statistics from database

func (*Service) GetScanStatus 

func (s *Service) GetScanStatus() ScanInfo

GetScanStatus returns the current scan status

func (*Service) HandleFailure 

func (s *Service) HandleFailure(ctx context.Context, item *database.ImportQueueItem, err error)

HandleFailure implements queue.ItemProcessor - handles failed processing

func (*Service) HandleSuccess 

func (s *Service) HandleSuccess(ctx context.Context, item *database.ImportQueueItem, resultingPath string) error

HandleSuccess implements queue.ItemProcessor - handles successful processing

func (*Service) IsFileInQueue 

func (s *Service) IsFileInQueue(ctx context.Context, filePath string) (bool, error)

IsFileInQueue checks if a file is already in the queue (pending or processing)

func (*Service) IsPaused 

func (s *Service) IsPaused() bool

IsPaused returns whether the service is paused

func (*Service) IsRunning 

func (s *Service) IsRunning() bool

IsRunning returns whether the service is running

func (*Service) MoveToFailedFolder 

func (s *Service) MoveToFailedFolder(ctx context.Context, item *database.ImportQueueItem) error

MoveToFailedFolder moves a failed NZB file to the failed directory

func (*Service) OnItemClaimed 

func (s *Service) OnItemClaimed(ctx context.Context, item *database.ImportQueueItem)

OnItemClaimed implements queue.QueueEventListener. It broadcasts a queue-changed notification whenever a worker claims a pending item (pending → processing transition).

func (*Service) Pause 

func (s *Service) Pause()

Pause pauses the queue processing

func (*Service) ProcessItem 

func (s *Service) ProcessItem(ctx context.Context, item *database.ImportQueueItem) (string, error)

ProcessItem implements queue.ItemProcessor - processes a single queue item

func (*Service) ProcessItemInBackground 

func (s *Service) ProcessItemInBackground(ctx context.Context, itemID int64)

ProcessItemInBackground processes a specific queue item in the background. NOTE: This intentionally runs outside the worker pool — it is used for manual retries of specific items and should not compete with the normal import queue workers.

func (*Service) RegenerateMetadata 

func (s *Service) RegenerateMetadata(ctx context.Context, mountRelativePath string) error

RegenerateMetadata attempts to find the original NZB for a file and re-process it to fix corrupted metadata.

func (*Service) RegisterConfigChangeHandler 

func (s *Service) RegisterConfigChangeHandler(configManager any)

func (*Service) ResetNzbdavImportStatus 

func (s *Service) ResetNzbdavImportStatus()

ResetNzbdavImportStatus resets the import status to Idle

func (*Service) Resume 

func (s *Service) Resume()

Resume resumes the queue processing

func (*Service) ResumeWaitingRepair 

func (s *Service) ResumeWaitingRepair(ctx context.Context, nzbPath string) error

ResumeWaitingRepair returns an import parked pending a PAR2 repair to the pending queue, so the normal worker retries it — this time with the repaired articles available from the local patch store.

func (*Service) SetArrsService 

func (s *Service) SetArrsService(service any)

SetArrsService sets or updates the ARRs service

func (*Service) SetContentVerifyFilesystem 

func (s *Service) SetContentVerifyFilesystem(fs contentverify.Opener)

SetContentVerifyFilesystem wires the real serving-stack filesystem used to probe imported media files' content signatures. Called once after the NzbFilesystem singleton is constructed (see cmd/altmount/cmd/serve.go), mirroring SetArrsService's late-binding to avoid an import-time dependency cycle between importer and nzbfilesystem.

func (*Service) SetPatchIndex 

func (s *Service) SetPatchIndex(idx validation.PatchIndex)

SetPatchIndex wires the PAR2 patch store into the import availability sweep.

func (*Service) SetRcloneClient 

func (s *Service) SetRcloneClient(client any)

SetRcloneClient sets or updates the RClone client for VFS notifications

func (*Service) SetRepairEnqueuer 

func (s *Service) SetRepairEnqueuer(re RepairEnqueuer)

SetRepairEnqueuer wires the PAR2 repair queue into the import processor so degraded imports can queue a repair. Call during boot.

func (*Service) SetSegmentStore 

func (s *Service) SetSegmentStore(resolve func() usenet.SegmentStore)

SetSegmentStore wires the streaming segment store into the importer so articles fetched at import are already cached when playback starts.

func (*Service) Start 

func (s *Service) Start(ctx context.Context) error

Start starts the NZB import service (queue workers only, manual scanning available via API)

func (*Service) StartManualScan 

func (s *Service) StartManualScan(scanPath string) error

StartManualScan starts a manual scan of the specified directory

func (*Service) StartNzbdavImport 

func (s *Service) StartNzbdavImport(dbPath string, blobsPath string, cleanupFile bool) error

StartNzbdavImport starts an asynchronous import from an NZBDav database

func (*Service) Stop 

func (s *Service) Stop(ctx context.Context) error

Stop stops the NZB import service and all queue workers

func (*Service) StoreGrabbedIndexer 

func (s *Service) StoreGrabbedIndexer(downloadID string, releaseTitle string, indexer string)

StoreGrabbedIndexer stores a downloadID to indexer mapping in-memory

type ServiceConfig 

type ServiceConfig struct {
	Workers int // Number of parallel queue workers (default: 2)
}

ServiceConfig holds configuration for the NZB import service

Source Files

View all Source files

Directories

Path Synopsis
iso
rar
sevenzip
fileinfo
par2
Package postprocessor handles all post-import processing steps including symlink creation, STRM file generation, VFS notifications, health check scheduling, and ARR notifications.
 Package postprocessor handles all post-import processing steps including symlink creation, STRM file generation, VFS notifications, health check scheduling, and ARR notifications.
Package queue provides queue management for the NZB import service.
 Package queue provides queue management for the NZB import service.
Package rarname holds the pure RAR / split-volume filename parsing used across the importer.
 Package rarname holds the pure RAR / split-volume filename parsing used across the importer.
Package scanner provides directory scanning and NZBDav import functionality.
 Package scanner provides directory scanning and NZBDav import functionality.
nzbtrim

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.