database

package
v0.0.1-alpha6 Latest Latest
Warning

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

 Go to latest Published: Feb 27, 2026 License: MIT Imports: 12 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type AutomaticHealthCheckRecord 

type AutomaticHealthCheckRecord struct {
	FilePath         string
	LibraryPath      *string
	ReleaseDate      *time.Time
	ScheduledCheckAt *time.Time
	SourceNzbPath    *string
	Status           HealthStatus
}

AutomaticHealthCheckRecord represents a batch insert record

type BackfillRecord 

type BackfillRecord struct {
	ID       int64
	FilePath string
}

BackfillRecord represents a record used for metadata backfilling

type BackfillUpdate 

type BackfillUpdate struct {
	ID               int64
	ReleaseDate      time.Time
	ScheduledCheckAt time.Time
}

BackfillUpdate represents an update for release date backfilling

type BulkDeleteResult 

type BulkDeleteResult struct {
	DeletedCount    int64
	ProcessingCount int64
	RequestedCount  int64
}

BulkDeleteResult contains the result of a bulk delete operation

type BulkOperationResult 

type BulkOperationResult struct {
	DeletedCount    int
	ProcessingCount int
	FailedIDs       []int64
}

BulkOperationResult represents the result of a bulk queue operation

type Config 

type Config struct {
	DatabasePath string
}

Config holds database configuration

type DB 

type DB struct {
	Repository *QueueRepository
	// contains filtered or unexported fields
}

DB wraps the database connection and provides access to operations

func NewDB 

func NewDB(config Config) (*DB, error)

NewDB creates a new database connection and runs migrations

func (*DB) Close 

func (db *DB) Close() error

Close closes the database connection

func (*DB) Connection 

func (db *DB) Connection() *sql.DB

Connection returns the underlying database connection

func (*DB) UpdateConnectionPool 

func (db *DB) UpdateConnectionPool(workerCount int)

UpdateConnectionPool adjusts the database connection pool settings based on worker count This should be called when the number of concurrent workers changes to prevent connection starvation

type DBQuerier 

type DBQuerier interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
}

DBQuerier defines the interface for database query operations Both *sql.DB and *sql.Tx implement this interface

type FileHealth 

type FileHealth struct {
	ID               int64        `db:"id"`
	FilePath         string       `db:"file_path"`
	LibraryPath      *string      `db:"library_path"` // Path to file in library directory (symlink or .strm file)
	Status           HealthStatus `db:"status"`
	LastChecked      *time.Time   `db:"last_checked"`
	LastError        *string      `db:"last_error"`
	RetryCount       int          `db:"retry_count"`        // Health check retry count
	MaxRetries       int          `db:"max_retries"`        // Max health check retries
	RepairRetryCount int          `db:"repair_retry_count"` // Repair retry count
	MaxRepairRetries int          `db:"max_repair_retries"` // Max repair retries
	SourceNzbPath    *string      `db:"source_nzb_path"`
	ErrorDetails     *string      `db:"error_details"` // JSON error details
	CreatedAt        time.Time    `db:"created_at"`
	UpdatedAt        time.Time    `db:"updated_at"`
	// Health check scheduling fields
	ReleaseDate      *time.Time     `db:"release_date"`       // Cached from metadata for scheduling
	ScheduledCheckAt *time.Time     `db:"scheduled_check_at"` // Next check time
	Priority         HealthPriority `db:"priority"`           // Priority level for health checks
}

FileHealth represents the health tracking of files in the filesystem

type HealthPriority 

type HealthPriority int

HealthPriority represents the priority level of a health check 

const (
	HealthPriorityNormal HealthPriority = 0
	HealthPriorityHigh   HealthPriority = 1
	HealthPriorityNext   HealthPriority = 2
)

type HealthRepository 

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

HealthRepository handles file health database operations

func NewHealthRepository 

func NewHealthRepository(db *sql.DB) *HealthRepository

NewHealthRepository creates a new health repository

func (*HealthRepository) AddFileToHealthCheck 

func (r *HealthRepository) AddFileToHealthCheck(ctx context.Context, filePath string, maxRetries int, sourceNzbPath *string, priority HealthPriority) error

AddFileToHealthCheck adds a file to the health database for checking

func (*HealthRepository) AddFileToHealthCheckWithMetadata 

func (r *HealthRepository) AddFileToHealthCheckWithMetadata(ctx context.Context, filePath string, maxRetries int, sourceNzbPath *string, priority HealthPriority, releaseDate *time.Time) error

AddFileToHealthCheckWithMetadata adds a file to the health database for checking with metadata

func (*HealthRepository) AddHealthCheck 

func (r *HealthRepository) AddHealthCheck(
	ctx context.Context,
	filePath string,
	releaseDate time.Time,
	scheduledCheckAt time.Time,
	sourceNzbPath *string,
) error

AddHealthCheck adds or updates a health check record

func (*HealthRepository) BackfillReleaseDates 

func (r *HealthRepository) BackfillReleaseDates(ctx context.Context, updates []BackfillUpdate) error

BackfillReleaseDates updates multiple health records with their release dates and next check times

func (*HealthRepository) BatchAddAutomaticHealthChecks 

func (r *HealthRepository) BatchAddAutomaticHealthChecks(ctx context.Context, records []AutomaticHealthCheckRecord) error

BatchAddAutomaticHealthChecks inserts multiple automatic health checks efficiently

func (*HealthRepository) CleanupHealthRecords 

func (r *HealthRepository) CleanupHealthRecords(ctx context.Context, existingFiles []string) error

CleanupHealthRecords removes health records for files that no longer exist

func (*HealthRepository) CountHealthItems 

func (r *HealthRepository) CountHealthItems(ctx context.Context, statusFilter *HealthStatus, sinceFilter *time.Time, search string) (int, error)

CountHealthItems returns the total count of health records with optional filtering

func (*HealthRepository) DeleteHealthRecord 

func (r *HealthRepository) DeleteHealthRecord(ctx context.Context, filePath string) error

DeleteHealthRecord removes a specific health record from the database

func (*HealthRepository) DeleteHealthRecordByID 

func (r *HealthRepository) DeleteHealthRecordByID(ctx context.Context, id int64) error

DeleteHealthRecordByID removes a specific health record from the database by ID

func (*HealthRepository) DeleteHealthRecordsBulk 

func (r *HealthRepository) DeleteHealthRecordsBulk(ctx context.Context, filePaths []string) error

DeleteHealthRecordsBulk removes multiple health records from the database

func (*HealthRepository) DeleteHealthRecordsByDate 

func (r *HealthRepository) DeleteHealthRecordsByDate(ctx context.Context, olderThan time.Time, statusFilter *HealthStatus) (int, error)

DeleteHealthRecordsByDate deletes health records older than the specified date with optional status filter

func (*HealthRepository) DeleteHealthRecordsByPrefix 

func (r *HealthRepository) DeleteHealthRecordsByPrefix(ctx context.Context, prefix string) (int64, error)

DeleteHealthRecordsByPrefix removes health records that start with the given prefix

func (*HealthRepository) GetAllHealthCheckPaths 

func (r *HealthRepository) GetAllHealthCheckPaths(ctx context.Context) ([]string, error)

GetAllHealthCheckPaths returns all health check file paths (memory optimized)

func (*HealthRepository) GetAllHealthCheckRecords 

func (r *HealthRepository) GetAllHealthCheckRecords(ctx context.Context) ([]AutomaticHealthCheckRecord, error)

GetAllHealthCheckRecords returns all health check records tracked in health system

func (*HealthRepository) GetFileHealth 

func (r *HealthRepository) GetFileHealth(ctx context.Context, filePath string) (*FileHealth, error)

GetFileHealth retrieves health record for a specific file

func (*HealthRepository) GetFileHealthByID 

func (r *HealthRepository) GetFileHealthByID(ctx context.Context, id int64) (*FileHealth, error)

GetFileHealthByID retrieves health record for a specific file by ID

func (*HealthRepository) GetFilesForRepairNotification 

func (r *HealthRepository) GetFilesForRepairNotification(ctx context.Context, limit int) ([]*FileHealth, error)

GetFilesForRepairNotification returns files that need repair notification (repair_triggered status)

func (*HealthRepository) GetFilesMissingReleaseDate 

func (r *HealthRepository) GetFilesMissingReleaseDate(ctx context.Context, limit int) ([]BackfillRecord, error)

GetFilesMissingReleaseDate returns a list of files that don't have a release date cached

func (*HealthRepository) GetFilesWithoutLibraryPath 

func (r *HealthRepository) GetFilesWithoutLibraryPath(ctx context.Context) ([]*FileHealth, error)

GetFilesWithoutLibraryPath returns all health records where library_path is NULL

func (*HealthRepository) GetHealthStats 

func (r *HealthRepository) GetHealthStats(ctx context.Context) (map[HealthStatus]int, error)

GetHealthStats returns statistics about file health

func (*HealthRepository) GetSystemState 

func (r *HealthRepository) GetSystemState(ctx context.Context, key string) (string, error)

GetSystemState retrieves a persistent state value

func (*HealthRepository) GetUnhealthyFiles 

func (r *HealthRepository) GetUnhealthyFiles(ctx context.Context, limit int) ([]*FileHealth, error)

GetUnhealthyFiles returns files that need health checks (excluding repair_triggered files)

func (*HealthRepository) IncrementRepairRetryCount 

func (r *HealthRepository) IncrementRepairRetryCount(ctx context.Context, filePath string, errorMessage *string, errorDetails *string) error

IncrementRepairRetryCount increments the repair retry count

func (*HealthRepository) IncrementRetryCount 

func (r *HealthRepository) IncrementRetryCount(ctx context.Context, filePath string, errorMessage *string, errorDetails *string, nextCheck time.Time) error

IncrementRetryCount increments the retry count and schedules next check

func (*HealthRepository) ListHealthItems 

func (r *HealthRepository) ListHealthItems(ctx context.Context, statusFilter *HealthStatus, limit, offset int, sinceFilter *time.Time, search string, sortBy string, sortOrder string) ([]*FileHealth, error)

ListHealthItems returns all health records with optional filtering, sorting and pagination

func (*HealthRepository) MarkAsCorrupted 

func (r *HealthRepository) MarkAsCorrupted(ctx context.Context, filePath string, finalError *string, errorDetails *string) error

MarkAsCorrupted permanently marks a file as corrupted after all retries are exhausted

func (*HealthRepository) MarkAsHealthy 

func (r *HealthRepository) MarkAsHealthy(ctx context.Context, filePath string, nextCheckTime time.Time) error

MarkAsHealthy marks a file as healthy and clears all retry/error state

func (*HealthRepository) RegisterCorruptedFile 

func (r *HealthRepository) RegisterCorruptedFile(ctx context.Context, filePath string, libraryPath *string, errorMessage string) error

RegisterCorruptedFile adds or updates a file as corrupted and schedules it for immediate check/repair

func (*HealthRepository) RelinkFileByFilename 

func (r *HealthRepository) RelinkFileByFilename(ctx context.Context, filename, filePath, libraryPath string) error

RelinkFileByFilename updates the file_path and library_path for a corrupted/triggered record that matches by filename

func (*HealthRepository) RenameHealthRecord 

func (r *HealthRepository) RenameHealthRecord(ctx context.Context, oldPath, newPath string) error

RenameHealthRecord updates the file_path of a health record or records under a directory after a MOVE operation

func (*HealthRepository) ResetAllHealthChecks 

func (r *HealthRepository) ResetAllHealthChecks(ctx context.Context) (int, error)

ResetAllHealthChecks resets all health records to pending status

func (*HealthRepository) ResetFileAllChecking 

func (r *HealthRepository) ResetFileAllChecking(ctx context.Context) error

func (*HealthRepository) ResetHealthChecksBulk 

func (r *HealthRepository) ResetHealthChecksBulk(ctx context.Context, filePaths []string) (int, error)

ResetHealthChecksBulk resets multiple health records to pending status

func (*HealthRepository) ResetStalePendingFiles 

func (r *HealthRepository) ResetStalePendingFiles(ctx context.Context) error

ResetStalePendingFiles resets pending files that have exhausted retries back to retry_count=0 so they can be re-checked in the next health cycle. Called during worker startup.

func (*HealthRepository) ResolvePendingRepairsInDirectory 

func (r *HealthRepository) ResolvePendingRepairsInDirectory(ctx context.Context, dirPath string) (int64, error)

ResolvePendingRepairsInDirectory removes health records with repair_triggered or corrupted status that exist in the specified directory. This is used when a new file is imported into a directory, implying it is a replacement for the broken file.

func (*HealthRepository) SetCorrupted 

func (r *HealthRepository) SetCorrupted(ctx context.Context, filePath string, errorMessage *string, errorDetails *string) error

SetCorrupted sets a file's status to corrupted

func (*HealthRepository) SetFileChecking 

func (r *HealthRepository) SetFileChecking(ctx context.Context, filePath string) error

SetFileChecking sets a file's status to 'checking'

func (*HealthRepository) SetFileCheckingByID 

func (r *HealthRepository) SetFileCheckingByID(ctx context.Context, id int64) error

SetFileCheckingByID sets a file's status to 'checking' by ID

func (*HealthRepository) SetPriority 

func (r *HealthRepository) SetPriority(ctx context.Context, id int64, priority HealthPriority) error

SetPriority sets the priority for a file health record

func (*HealthRepository) SetRepairTriggered 

func (r *HealthRepository) SetRepairTriggered(ctx context.Context, filePath string, errorMessage *string, errorDetails *string) error

SetRepairTriggered sets a file's status to repair_triggered

func (*HealthRepository) SetRepairTriggeredByID 

func (r *HealthRepository) SetRepairTriggeredByID(ctx context.Context, id int64, errorMessage *string, errorDetails *string) error

SetRepairTriggeredByID sets a file's status to repair_triggered by ID

func (*HealthRepository) UpdateFileHealth 

func (r *HealthRepository) UpdateFileHealth(ctx context.Context, filePath string, status HealthStatus, errorMessage *string, sourceNzbPath *string, errorDetails *string, noRetry bool) error

UpdateFileHealth updates or inserts a file health record

func (*HealthRepository) UpdateHealthStatusBulk 

func (r *HealthRepository) UpdateHealthStatusBulk(ctx context.Context, updates []HealthStatusUpdate) error

UpdateHealthStatusBulk updates multiple health records in a single transaction

func (*HealthRepository) UpdateLibraryPath 

func (r *HealthRepository) UpdateLibraryPath(ctx context.Context, filePath string, libraryPath string) error

UpdateLibraryPath updates the library_path for a specific file

func (*HealthRepository) UpdateScheduledCheckTime 

func (r *HealthRepository) UpdateScheduledCheckTime(ctx context.Context, filePath string, nextCheckTime time.Time) error

UpdateScheduledCheckTime updates the scheduled check time for a file

func (*HealthRepository) UpdateSystemState 

func (r *HealthRepository) UpdateSystemState(ctx context.Context, key string, value string) error

UpdateSystemState updates or inserts a persistent state value

type HealthStatus 

type HealthStatus string

HealthStatus represents the health status of a file 

const (
	HealthStatusPending         HealthStatus = "pending"          // File has not been checked yet
	HealthStatusChecking        HealthStatus = "checking"         // File is currently being checked
	HealthStatusHealthy         HealthStatus = "healthy"          // File passed health check
	HealthStatusRepairTriggered HealthStatus = "repair_triggered" // File repair has been triggered in Arrs
	HealthStatusCorrupted       HealthStatus = "corrupted"        // File has missing segments or is corrupted
)

type HealthStatusUpdate 

type HealthStatusUpdate struct {
	Type             UpdateType
	FilePath         string
	Status           HealthStatus
	ErrorMessage     *string
	ErrorDetails     *string
	ScheduledCheckAt time.Time
	Skip             bool // if true, skip this record in the bulk update (e.g. record already deleted)
}

HealthStatusUpdate represents a single update request for batch processing

type ImportDailyStat 

type ImportDailyStat struct {
	Day             time.Time `db:"day"`
	CompletedCount  int       `db:"completed_count"`
	FailedCount     int       `db:"failed_count"`
	BytesDownloaded int64     `db:"bytes_downloaded"`
	UpdatedAt       time.Time `db:"updated_at"`
}

ImportDailyStat represents historical import statistics for a specific day

type ImportHistory 

type ImportHistory struct {
	ID          int64     `db:"id"`
	NzbID       *int64    `db:"nzb_id"` // Nullable if queue item deleted
	NzbName     string    `db:"nzb_name"`
	FileName    string    `db:"file_name"`
	FileSize    int64     `db:"file_size"`
	VirtualPath string    `db:"virtual_path"`
	LibraryPath *string   `db:"library_path"` // Added to show final location from file_health
	Category    *string   `db:"category"`
	CompletedAt time.Time `db:"completed_at"`
}

ImportHistory represents a persistent record of a single imported file

type ImportHourlyStat 

type ImportHourlyStat struct {
	Hour            time.Time `db:"hour"`
	CompletedCount  int       `db:"completed_count"`
	FailedCount     int       `db:"failed_count"`
	BytesDownloaded int64     `db:"bytes_downloaded"`
	UpdatedAt       time.Time `db:"updated_at"`
}

ImportHourlyStat represents historical import statistics for a specific hour

type ImportQueueItem 

type ImportQueueItem struct {
	ID           int64         `db:"id"`
	NzbPath      string        `db:"nzb_path"`
	RelativePath *string       `db:"relative_path"`
	StoragePath  *string       `db:"storage_path"`
	Category     *string       `db:"category"` // SABnzbd-compatible category
	Priority     QueuePriority `db:"priority"`
	Status       QueueStatus   `db:"status"`
	CreatedAt    time.Time     `db:"created_at"`
	UpdatedAt    time.Time     `db:"updated_at"`
	StartedAt    *time.Time    `db:"started_at"`
	CompletedAt  *time.Time    `db:"completed_at"`
	RetryCount   int           `db:"retry_count"`
	MaxRetries   int           `db:"max_retries"`
	ErrorMessage *string       `db:"error_message"`
	BatchID      *string       `db:"batch_id"`
	Metadata     *string       `db:"metadata"`  // JSON metadata
	FileSize     *int64        `db:"file_size"` // Total size in bytes calculated from segments
}

ImportQueueItem represents a queued NZB file waiting for import

type MediaFile 

type MediaFile struct {
	ID           int64     `db:"id"`
	InstanceName string    `db:"instance_name"` // Name from configuration
	InstanceType string    `db:"instance_type"` // "radarr" or "sonarr"
	ExternalID   int64     `db:"external_id"`   // Movie ID or Episode ID from API
	FileID       int64     `db:"file_id"`       // Movie File ID or Episode File ID from API (nullable)
	FilePath     string    `db:"file_path"`     // Full file path
	FileSize     *int64    `db:"file_size"`     // File size in bytes (nullable)
	CreatedAt    time.Time `db:"created_at"`    // When record was created
	UpdatedAt    time.Time `db:"updated_at"`    // When record was last updated
}

MediaFile represents a media file tracked by scrapers

type MediaFileInput 

type MediaFileInput struct {
	InstanceName string
	InstanceType string
	ExternalID   int64  // Movie ID or Episode ID
	FileID       *int64 // Movie File ID or Episode File ID (nullable)
	FilePath     string
	FileSize     *int64
}

MediaFileInput represents input data for media file operations

type MediaRepository 

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

MediaRepository handles operations for media files table

func NewMediaRepository 

func NewMediaRepository(db *sql.DB) *MediaRepository

NewMediaRepository creates a new media repository

func (*MediaRepository) CleanupInstanceData 

func (r *MediaRepository) CleanupInstanceData(ctx context.Context, instanceName, instanceType string) error

CleanupInstanceData removes all media files for a specific instance This can be called when an instance is removed from configuration

func (*MediaRepository) GetMediaFilesByInstance 

func (r *MediaRepository) GetMediaFilesByInstance(ctx context.Context, instanceName, instanceType string) ([]MediaFile, error)

GetMediaFilesByInstance returns all media files for a specific instance

func (*MediaRepository) GetMediaFilesByPath 

func (r *MediaRepository) GetMediaFilesByPath(ctx context.Context, filePath string) ([]MediaFile, error)

GetMediaFilesByPath returns media files matching a file path This can be used for health correlation

func (*MediaRepository) GetMediaFilesCount 

func (r *MediaRepository) GetMediaFilesCount(ctx context.Context) (int64, error)

GetMediaFilesCount returns the total count of media files

func (*MediaRepository) GetMediaFilesCountByInstance 

func (r *MediaRepository) GetMediaFilesCountByInstance(ctx context.Context, instanceName, instanceType string) (int64, error)

GetMediaFilesCountByInstance returns the count of media files for a specific instance

func (*MediaRepository) SyncMediaFiles 

func (r *MediaRepository) SyncMediaFiles(ctx context.Context, instanceName, instanceType string, files []MediaFileInput) (*SyncResult, error)

SyncMediaFiles performs a complete sync operation for an instance This replaces all files for the instance with the provided list

type QueuePriority 

type QueuePriority int

QueuePriority represents the priority level of a queued import 

const (
	QueuePriorityHigh   QueuePriority = 1
	QueuePriorityNormal QueuePriority = 2
	QueuePriorityLow    QueuePriority = 3
)

type QueueRepository 

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

QueueRepository handles queue-specific database operations

func NewQueueRepository 

func NewQueueRepository(db *sql.DB) *QueueRepository

NewQueueRepository creates a new queue repository

func (*QueueRepository) AddBatchToQueue 

func (r *QueueRepository) AddBatchToQueue(ctx context.Context, items []*ImportQueueItem) error

AddBatchToQueue adds multiple items to the queue in a single transaction

func (*QueueRepository) AddImportHistory 

func (r *QueueRepository) AddImportHistory(ctx context.Context, history *ImportHistory) error

AddImportHistory records a successful file import in the persistent history table

func (*QueueRepository) AddStoragePath 

func (r *QueueRepository) AddStoragePath(ctx context.Context, itemID int64, storagePath string) error

func (*QueueRepository) AddToQueue 

func (r *QueueRepository) AddToQueue(ctx context.Context, item *ImportQueueItem) error

AddToQueue adds a new NZB file to the import queue

func (*QueueRepository) ClaimNextQueueItem 

func (r *QueueRepository) ClaimNextQueueItem(ctx context.Context) (*ImportQueueItem, error)

ClaimNextQueueItem atomically claims and returns the next available queue item

func (*QueueRepository) ClearDailyStats 

func (r *QueueRepository) ClearDailyStats(ctx context.Context) error

ClearDailyStats deletes all records from the import_daily_stats and import_hourly_stats tables

func (*QueueRepository) ClearHourlyStats 

func (r *QueueRepository) ClearHourlyStats(ctx context.Context) error

ClearHourlyStats deletes all records from the import_hourly_stats table

func (*QueueRepository) ClearImportHistory 

func (r *QueueRepository) ClearImportHistory(ctx context.Context) error

ClearImportHistory deletes all records from the import_history and import_daily_stats tables

func (*QueueRepository) ClearImportHistorySince 

func (r *QueueRepository) ClearImportHistorySince(ctx context.Context, since time.Time) error

ClearImportHistorySince deletes records from the import_history and import_queue tables since the specified time, and adjusts the import_daily_stats and import_hourly_stats accordingly.

func (*QueueRepository) FilterExistingNzbdavIds 

func (r *QueueRepository) FilterExistingNzbdavIds(ctx context.Context, ids []string) ([]string, error)

FilterExistingNzbdavIds checks a list of nzbdav IDs and returns those that already exist in the queue

func (*QueueRepository) GetImportDailyStats 

func (r *QueueRepository) GetImportDailyStats(ctx context.Context, days int) ([]*ImportDailyStat, error)

GetImportDailyStats retrieves historical import statistics for the last N days

func (*QueueRepository) GetImportHistory 

func (r *QueueRepository) GetImportHistory(ctx context.Context, days int) ([]*ImportDailyStat, error)

GetImportHistory retrieves historical import statistics for the last N days (Alias for GetImportDailyStats)

func (*QueueRepository) GetImportHourlyStats 

func (r *QueueRepository) GetImportHourlyStats(ctx context.Context, hours int) ([]*ImportHourlyStat, error)

GetImportHourlyStats retrieves import statistics for the specified number of hours

func (*QueueRepository) GetQueueItem 

func (r *QueueRepository) GetQueueItem(ctx context.Context, id int64) (*ImportQueueItem, error)

GetQueueItem retrieves a specific queue item by ID

func (*QueueRepository) GetQueueStats 

func (r *QueueRepository) GetQueueStats(ctx context.Context) (*QueueStats, error)

GetQueueStats returns current queue statistics

func (*QueueRepository) IncrementDailyStat 

func (r *QueueRepository) IncrementDailyStat(ctx context.Context, statType string) error

IncrementDailyStat increments the completed or failed count for the current day

func (*QueueRepository) IncrementHourlyStat 

func (r *QueueRepository) IncrementHourlyStat(ctx context.Context, statType string) error

IncrementHourlyStat increments the completed or failed count for the current hour

func (*QueueRepository) IncrementRetryCountAndResetStatus 

func (r *QueueRepository) IncrementRetryCountAndResetStatus(ctx context.Context, id int64, errorMessage *string) (bool, error)

IncrementRetryCountAndResetStatus increments the retry count and resets the status to pending

func (*QueueRepository) IsFileInQueue 

func (r *QueueRepository) IsFileInQueue(ctx context.Context, filePath string) (bool, error)

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

func (*QueueRepository) ListImportHistory 

func (r *QueueRepository) ListImportHistory(ctx context.Context, limit int) ([]*ImportHistory, error)

ListImportHistory retrieves the last N successful imports from the persistent history

func (*QueueRepository) RemoveFromQueue 

func (r *QueueRepository) RemoveFromQueue(ctx context.Context, id int64) error

RemoveFromQueue removes an item from the queue

func (*QueueRepository) RemoveFromQueueBulk 

func (r *QueueRepository) RemoveFromQueueBulk(ctx context.Context, ids []int64) (*BulkOperationResult, error)

RemoveFromQueueBulk removes multiple items from the queue in bulk

func (*QueueRepository) ResetStaleItems 

func (r *QueueRepository) ResetStaleItems(ctx context.Context) error

ResetStaleItems resets processing items back to pending on service startup

func (*QueueRepository) RestartQueueItemsBulk 

func (r *QueueRepository) RestartQueueItemsBulk(ctx context.Context, ids []int64) error

RestartQueueItemsBulk resets multiple queue items back to pending status

func (*QueueRepository) UpdateQueueItemNzbPath 

func (r *QueueRepository) UpdateQueueItemNzbPath(ctx context.Context, id int64, nzbPath string) error

UpdateQueueItemNzbPath updates the NZB path of a queue item

func (*QueueRepository) UpdateQueueItemPriority 

func (r *QueueRepository) UpdateQueueItemPriority(ctx context.Context, id int64, priority QueuePriority) error

UpdateQueueItemPriority updates the priority of a queue item

func (*QueueRepository) UpdateQueueItemStatus 

func (r *QueueRepository) UpdateQueueItemStatus(ctx context.Context, id int64, status QueueStatus, errorMessage *string) error

UpdateQueueItemStatus updates the status of a queue item

type QueueStats 

type QueueStats struct {
	ID                  int64     `db:"id"`
	TotalQueued         int       `db:"total_queued"`
	TotalProcessing     int       `db:"total_processing"`
	TotalCompleted      int       `db:"total_completed"`
	TotalFailed         int       `db:"total_failed"`
	AvgProcessingTimeMs *int      `db:"avg_processing_time_ms"`
	LastUpdated         time.Time `db:"last_updated"`
}

QueueStats represents statistics about the import queue

type QueueStatus 

type QueueStatus string

QueueStatus represents the status of a queued import 

const (
	QueueStatusPending    QueueStatus = "pending"
	QueueStatusProcessing QueueStatus = "processing"
	QueueStatusCompleted  QueueStatus = "completed"
	QueueStatusFailed     QueueStatus = "failed"
	QueueStatusPaused     QueueStatus = "paused"
	QueueStatusFallback   QueueStatus = "fallback" // Sent to external SABnzbd as fallback
)

type Repository 

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

Repository provides database operations for NZB and file management

func NewRepository 

func NewRepository(db *sql.DB) *Repository

NewRepository creates a new repository instance

func (*Repository) AddBatchToQueue 

func (r *Repository) AddBatchToQueue(ctx context.Context, items []*ImportQueueItem) error

AddBatchToQueue adds multiple items to the queue in a single transaction for better performance

func (*Repository) AddBytesDownloadedToDailyStat 

func (r *Repository) AddBytesDownloadedToDailyStat(ctx context.Context, bytes int64) error

AddBytesDownloadedToDailyStat increments the bytes_downloaded counter for the current day

func (*Repository) AddBytesDownloadedToHourlyStat 

func (r *Repository) AddBytesDownloadedToHourlyStat(ctx context.Context, bytes int64) error

AddBytesDownloadedToHourlyStat increments the bytes_downloaded counter for the current hour

func (*Repository) AddImportHistory 

func (r *Repository) AddImportHistory(ctx context.Context, history *ImportHistory) error

AddImportHistory records a successful file import in the persistent history table

func (*Repository) AddToQueue 

func (r *Repository) AddToQueue(ctx context.Context, item *ImportQueueItem) error

AddToQueue adds an NZB file to the import queue with optimized concurrency

func (*Repository) BatchUpdateSystemStats 

func (r *Repository) BatchUpdateSystemStats(ctx context.Context, stats map[string]int64) error

BatchUpdateSystemStats updates multiple system statistics in a single transaction

func (*Repository) ClaimNextQueueItem 

func (r *Repository) ClaimNextQueueItem(ctx context.Context) (*ImportQueueItem, error)

ClaimNextQueueItem atomically claims and returns the next available queue item This prevents multiple workers from processing the same item Uses a single atomic UPDATE...RETURNING query to eliminate race conditions

func (*Repository) ClearCompletedQueueItems 

func (r *Repository) ClearCompletedQueueItems(ctx context.Context) (int, error)

ClearCompletedQueueItems removes completed and failed items from the queue

func (*Repository) ClearDailyStats 

func (r *Repository) ClearDailyStats(ctx context.Context) error

ClearDailyStats deletes all records from the import_daily_stats and import_hourly_stats tables

func (*Repository) ClearFailedQueueItems 

func (r *Repository) ClearFailedQueueItems(ctx context.Context) (int, error)

ClearFailedQueueItems removes failed items from the queue

func (*Repository) ClearHourlyStats 

func (r *Repository) ClearHourlyStats(ctx context.Context) error

ClearHourlyStats deletes all records from the import_hourly_stats table

func (*Repository) ClearImportHistory 

func (r *Repository) ClearImportHistory(ctx context.Context) error

ClearImportHistory deletes all records from the import_history and import_daily_stats tables

func (*Repository) ClearImportHistorySince 

func (r *Repository) ClearImportHistorySince(ctx context.Context, since time.Time) error

ClearImportHistorySince deletes records from the import_history and import_queue tables since the specified time, and adjusts the import_daily_stats accordingly.

func (*Repository) ClearPendingQueueItems 

func (r *Repository) ClearPendingQueueItems(ctx context.Context) (int, error)

ClearPendingQueueItems removes pending items from the queue

func (*Repository) CountActiveQueueItems 

func (r *Repository) CountActiveQueueItems(ctx context.Context, search string, category string) (int, error)

CountActiveQueueItems counts the total number of pending and processing queue items

func (*Repository) CountQueueItems 

func (r *Repository) CountQueueItems(ctx context.Context, status *QueueStatus, search string, category string) (int, error)

CountQueueItems counts the total number of queue items matching the given filters

func (*Repository) FilterExistingNzbdavIds 

func (r *Repository) FilterExistingNzbdavIds(ctx context.Context, ids []string) ([]string, error)

FilterExistingNzbdavIds checks a list of nzbdav IDs and returns those that already exist in the queue This is used for deduplication during import

func (*Repository) GetImportDailyStats 

func (r *Repository) GetImportDailyStats(ctx context.Context, days int) ([]*ImportDailyStat, error)

GetImportDailyStats retrieves import statistics for the specified number of days

func (*Repository) GetImportHistory 

func (r *Repository) GetImportHistory(ctx context.Context, days int) ([]*ImportDailyStat, error)

GetImportHistory retrieves historical import statistics for the last N days (Alias for GetImportDailyStats)

func (*Repository) GetImportHistoryItem added in v0.2.0 

func (r *Repository) GetImportHistoryItem(ctx context.Context, id int64) (*ImportHistory, error)

GetImportHistoryItem retrieves a specific import history item by ID

func (*Repository) GetImportHourlyStats 

func (r *Repository) GetImportHourlyStats(ctx context.Context, hours int) ([]*ImportHourlyStat, error)

GetImportHourlyStats retrieves import statistics for the specified number of hours

func (*Repository) GetNextQueueItems 

func (r *Repository) GetNextQueueItems(ctx context.Context, limit int) ([]*ImportQueueItem, error)

GetNextQueueItems retrieves the next batch of items to process from the queue Uses optimized query with row-level locking for better concurrency

func (*Repository) GetQueueItem 

func (r *Repository) GetQueueItem(ctx context.Context, id int64) (*ImportQueueItem, error)

GetQueueItem retrieves a specific queue item by ID

func (*Repository) GetQueueItemByPath 

func (r *Repository) GetQueueItemByPath(ctx context.Context, nzbPath string) (*ImportQueueItem, error)

GetQueueItemByPath retrieves a queue item by NZB path

func (*Repository) GetQueueStats 

func (r *Repository) GetQueueStats(ctx context.Context) (*QueueStats, error)

GetQueueStats retrieves current queue statistics

func (*Repository) GetSystemState 

func (r *Repository) GetSystemState(ctx context.Context, key string) (string, error)

GetSystemState retrieves a system state string by key

func (*Repository) GetSystemStats 

func (r *Repository) GetSystemStats(ctx context.Context) (map[string]int64, error)

GetSystemStats retrieves all system statistics as a map

func (*Repository) IncrementDailyStat 

func (r *Repository) IncrementDailyStat(ctx context.Context, statType string) error

IncrementDailyStat increments the completed or failed count for the current day

func (*Repository) IncrementHourlyStat 

func (r *Repository) IncrementHourlyStat(ctx context.Context, statType string) error

IncrementHourlyStat increments the completed or failed count for the current hour

func (*Repository) IsFileInQueue 

func (r *Repository) IsFileInQueue(ctx context.Context, filePath string) (bool, error)

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

func (*Repository) ListActiveQueueItems 

func (r *Repository) ListActiveQueueItems(ctx context.Context, search string, category string, limit, offset int, sortBy, sortOrder string) ([]*ImportQueueItem, error)

ListActiveQueueItems retrieves pending and processing queue items

func (*Repository) ListImportHistory 

func (r *Repository) ListImportHistory(ctx context.Context, limit, offset int, search string, category string) ([]*ImportHistory, error)

ListImportHistory retrieves import history items with optional filtering and pagination

func (*Repository) ListQueueItems 

func (r *Repository) ListQueueItems(ctx context.Context, status *QueueStatus, search string, category string, limit, offset int, sortBy, sortOrder string) ([]*ImportQueueItem, error)

ListQueueItems retrieves queue items with optional filtering

func (*Repository) RemoveFromQueue 

func (r *Repository) RemoveFromQueue(ctx context.Context, id int64) error

RemoveFromQueue removes an item from the queue

func (*Repository) RemoveFromQueueBulk 

func (r *Repository) RemoveFromQueueBulk(ctx context.Context, ids []int64) (*BulkDeleteResult, error)

RemoveFromQueueBulk removes multiple items from the queue, excluding those currently being processed

func (*Repository) RestartQueueItemsBulk 

func (r *Repository) RestartQueueItemsBulk(ctx context.Context, ids []int64) error

RestartQueueItemsBulk resets multiple queue items to pending status for reprocessing

func (*Repository) UpdateQueueItemPriority 

func (r *Repository) UpdateQueueItemPriority(ctx context.Context, id int64, priority QueuePriority) error

UpdateQueueItemPriority updates the priority of a queue item

func (*Repository) UpdateQueueItemStatus 

func (r *Repository) UpdateQueueItemStatus(ctx context.Context, id int64, status QueueStatus, errorMessage *string) error

UpdateQueueItemStatus updates the status of a queue item

func (*Repository) UpdateQueueStats 

func (r *Repository) UpdateQueueStats(ctx context.Context) error

UpdateQueueStats updates queue statistics based on current queue state

func (*Repository) UpdateSystemStat 

func (r *Repository) UpdateSystemStat(ctx context.Context, key string, value int64) error

UpdateSystemStat updates or inserts a single system statistic

func (*Repository) UpdateSystemState 

func (r *Repository) UpdateSystemState(ctx context.Context, key string, value string) error

UpdateSystemState updates a system state string (JSON) by key

func (*Repository) WithImmediateTransaction 

func (r *Repository) WithImmediateTransaction(ctx context.Context, fn func(*Repository) error) error

WithImmediateTransaction executes a function within an immediate database transaction This reduces lock contention for queue operations by acquiring write locks immediately Uses SQLite's IMMEDIATE transaction mode via BeginTx with Serializable isolation

func (*Repository) WithTransaction 

func (r *Repository) WithTransaction(ctx context.Context, fn func(*Repository) error) error

WithTransaction executes a function within a database transaction

type SyncResult 

type SyncResult struct {
	Added   int
	Updated int
	Removed int
}

SyncResult represents the result of a sync operation

type SystemStat 

type SystemStat struct {
	Key       string    `db:"key"`
	Value     int64     `db:"value"`
	UpdatedAt time.Time `db:"updated_at"`
}

SystemStat represents a persistent system statistic

type UpdateType 

type UpdateType int

UpdateType represents the type of health update 

const (
	UpdateTypeHealthy       UpdateType = 1
	UpdateTypeRetry         UpdateType = 2
	UpdateTypeRepairRetry   UpdateType = 3 // re-check of an already-triggered repair; increments repair_retry_count
	UpdateTypeCorrupted     UpdateType = 4
	UpdateTypeRepairTrigger UpdateType = 5 // first-time trigger; does not increment repair_retry_count
)

type User 

type User struct {
	ID           int64      `db:"id"`
	UserID       string     `db:"user_id"`       // Unique identifier from auth provider
	Email        *string    `db:"email"`         // User email address (nullable)
	Name         *string    `db:"name"`          // User display name (nullable)
	AvatarURL    *string    `db:"avatar_url"`    // User avatar image URL (nullable)
	Provider     string     `db:"provider"`      // Auth provider (direct, github, google, dev, etc.)
	ProviderID   *string    `db:"provider_id"`   // Provider-specific user ID (nullable)
	PasswordHash *string    `db:"password_hash"` // Bcrypt password hash for direct auth (nullable)
	APIKey       *string    `db:"api_key"`       // API key for user authentication (nullable)
	IsAdmin      bool       `db:"is_admin"`      // Admin privileges flag
	CreatedAt    time.Time  `db:"created_at"`    // Account creation timestamp
	UpdatedAt    time.Time  `db:"updated_at"`    // Last profile update timestamp
	LastLogin    *time.Time `db:"last_login"`    // Last login timestamp (nullable)
}

User represents a user account in the system

type UserRepository 

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

UserRepository handles user database operations

func NewUserRepository 

func NewUserRepository(db *sql.DB) *UserRepository

NewUserRepository creates a new user repository

func (*UserRepository) CreateUser 

func (r *UserRepository) CreateUser(ctx context.Context, user *User) error

CreateUser creates a new user account

func (*UserRepository) GetAllUsers 

func (r *UserRepository) GetAllUsers(ctx context.Context) ([]*User, error)

GetAllUsers retrieves all users with API keys for authentication purposes

func (*UserRepository) GetUserByAPIKey 

func (r *UserRepository) GetUserByAPIKey(ctx context.Context, apiKey string) (*User, error)

GetUserByAPIKey retrieves a user by their API key

func (*UserRepository) GetUserByEmail 

func (r *UserRepository) GetUserByEmail(ctx context.Context, email string) (*User, error)

GetUserByEmail retrieves a user by their email address for direct authentication

func (*UserRepository) GetUserByID 

func (r *UserRepository) GetUserByID(ctx context.Context, userID string) (*User, error)

GetUserByID retrieves a user by their unique user ID

func (*UserRepository) GetUserByProvider 

func (r *UserRepository) GetUserByProvider(ctx context.Context, provider, providerID string) (*User, error)

GetUserByProvider retrieves a user by provider and provider ID

func (*UserRepository) GetUserByUsername 

func (r *UserRepository) GetUserByUsername(ctx context.Context, username string) (*User, error)

GetUserByUsername retrieves a user by their username (user_id) for direct authentication

func (*UserRepository) GetUserCount 

func (r *UserRepository) GetUserCount(ctx context.Context) (int, error)

GetUserCount returns the total number of users

func (*UserRepository) RegenerateAPIKey 

func (r *UserRepository) RegenerateAPIKey(ctx context.Context, userID string) (string, error)

RegenerateAPIKey generates and updates a new API key for the user

func (*UserRepository) UpdateLastLogin 

func (r *UserRepository) UpdateLastLogin(ctx context.Context, userID string) error

UpdateLastLogin updates the user's last login timestamp

func (*UserRepository) UpdatePassword 

func (r *UserRepository) UpdatePassword(ctx context.Context, userID string, passwordHash string) error

UpdatePassword updates a user's password hash

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.