database

package
v0.3.2 Latest Latest
Warning

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

 Go to latest Published: Jul 18, 2026 License: MIT Imports: 18 Imported by: 0

Documentation

Index

Constants

View Source 
const IndexerUnknown = "Unknown"

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
	MaxRetries       int
	MaxRepairRetries int
}

AutomaticHealthCheckRecord represents a batch insert record

type BackfillRecord 

type BackfillRecord struct {
	ID       int64
	FilePath string
	Metadata *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 BulkOperationResult 

type BulkOperationResult struct {
	DeletedCount    int
	ProcessingCount int
	FailedIDs       []int64
	// DeletedPaths holds the nzb_path values for rows that were actually removed,
	// so callers can clean up the corresponding files on disk.
	DeletedPaths []string
}

BulkOperationResult represents the result of a bulk queue operation

type Config 

type Config struct {
	// Type selects the backend: "sqlite" (default) or "postgres".
	Type         string
	DatabasePath string // SQLite only
	DSN          string // PostgreSQL only
}

Config holds database configuration.

type DB 

type DB struct {

	// Repository is kept for backwards-compat; prefer using Connection() directly.
	Repository    *QueueRepository
	MigrationRepo *ImportMigrationRepository
	StoreRefRepo  *StoreRefRepository
	// 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) Dialect 

func (db *DB) Dialect() Dialect

Dialect returns the dialect helper for this database.

func (*DB) StartCheckpointLoop added in v0.3.0 

func (db *DB) StartCheckpointLoop(ctx context.Context, interval time.Duration)

StartCheckpointLoop starts a background goroutine that periodically forces a WAL checkpoint (SQLite only). Call once after opening the DB; stops when ctx is cancelled.

func (*DB) UpdateConnectionPool 

func (db *DB) UpdateConnectionPool(workerCount int)

UpdateConnectionPool adjusts the database connection pool settings based on worker count.

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 DBSymlinkLookup added in v0.3.0 

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

DBSymlinkLookup adapts ImportMigrationRepository to migration.SymlinkLookup.

func NewDBSymlinkLookup added in v0.3.0 

func NewDBSymlinkLookup(repo *ImportMigrationRepository) *DBSymlinkLookup

NewDBSymlinkLookup creates a new DBSymlinkLookup wrapping the given repository.

func (*DBSymlinkLookup) LookupFinalPath added in v0.3.0 

func (l *DBSymlinkLookup) LookupFinalPath(ctx context.Context, source, externalID string) (string, bool, error)

LookupFinalPath returns the final AltMount path for the given source and externalID. Returns ("", false, nil) when no matching row exists or the row has no final_path.

Season-pack episode rows use a "file:<episodeFilename>" relative_path to signal that final_path stores the season directory and the episode path must be computed by joining them. This keeps MarkImported simple (always stores the directory) while allowing per-episode resolution here.

func (*DBSymlinkLookup) MarkSymlinksMigrated added in v0.3.0 

func (l *DBSymlinkLookup) MarkSymlinksMigrated(ctx context.Context, ids []int64) error

MarkSymlinksMigrated sets status=symlinks_migrated for the given row IDs.

type Dialect 

type Dialect string

Dialect identifies the database backend. 

const (
	DialectSQLite   Dialect = "sqlite"
	DialectPostgres Dialect = "postgres"
)

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
	Metadata         *string      `db:"metadata"`      // JSON metadata
	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
	// Failure masking fields
	StreamingFailureCount int     `db:"streaming_failure_count"`
	IsMasked              bool    `db:"is_masked"`
	Indexer               *string `db:"indexer"`
	DownloadID            *string `db:"download_id"`
}

FileHealth represents the health tracking of files in the filesystem

func (*FileHealth) EffectiveLibraryPath added in v0.3.0 

func (f *FileHealth) EffectiveLibraryPath() (string, bool)

EffectiveLibraryPath returns the real library path and true when the record has been relinked by an ARR; otherwise ("", false).

func (*FileHealth) IsImported added in v0.3.0 

func (f *FileHealth) IsImported() bool

IsImported reports whether library_path points to a real, ARR-relinked library location rather than the import-time placeholder. The placeholder is stored equal to file_path (optionally with a leading slash from path normalization); a real filesystem path can never equal file_path or "/"+file_path, so this is unambiguous on every platform.

type HealthCheckUpsert added in v0.3.0 

type HealthCheckUpsert struct {
	FilePath         string
	LibraryPath      *string
	SourceNzbPath    *string
	Indexer          *string
	Priority         HealthPriority
	MaxRetries       int
	MaxRepairRetries int
	ReleaseDate      *time.Time
	Metadata         *string
	DownloadID       *string
}

HealthCheckUpsert is one record for BatchAddFileToHealthCheck.

type HealthErrorDetails added in v0.3.0 

type HealthErrorDetails struct {
	ErrorType       string        `json:"error_type"`
	Message         string        `json:"message,omitempty"`
	MissingArticles int           `json:"missing_articles,omitempty"`
	TotalArticles   int           `json:"total_articles,omitempty"`
	Sampled         int           `json:"sampled,omitempty"`
	PlaybackImpact  *holes.Impact `json:"playback_impact,omitempty"`
}

HealthErrorDetails is the structured JSON stored in file_health.error_details. Both the health checker and the streaming failure path marshal this envelope; the frontend parses it to render playback-impact information. Legacy rows may contain other ad-hoc JSON shapes or plain strings — parsers must tolerate that.

func (*HealthErrorDetails) Marshal added in v0.3.0 

func (d *HealthErrorDetails) Marshal() *string

Marshal renders the envelope for storage, returning nil on the (practically impossible) marshal error so callers can assign it directly to *string fields.

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, d Dialect) *HealthRepository

NewHealthRepository creates a new health repository

func (*HealthRepository) AddFileToHealthCheck 

func (r *HealthRepository) AddFileToHealthCheck(ctx context.Context, filePath string, libraryPath *string, maxRetries int, maxRepairRetries 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, libraryPath *string, maxRetries int, maxRepairRetries int, sourceNzbPath *string, priority HealthPriority, releaseDate *time.Time, metadata *string, indexer *string, downloadID *string) error

AddFileToHealthCheckWithMetadata adds a file to the health database for checking with metadata. On conflict (re-import over an existing record) the record is reset to pending for re-validation, but repair_retry_count is intentionally preserved: it is the per-title repair budget, so a re-download of a broken release cannot reset its own escalation counter. A successful health check resets it to 0.

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) BatchAddFileToHealthCheck added in v0.3.0 

func (r *HealthRepository) BatchAddFileToHealthCheck(ctx context.Context, records []HealthCheckUpsert) error

BatchAddFileToHealthCheck upserts many health records in a few multi-row statements instead of one transaction per file. It has the SAME conflict semantics as AddFileToHealthCheckWithMetadata (reset to pending for re-validation, repair_retry_count preserved as the per-title budget). Used by the import post-processor, where a single archive/season-pack import can expand to hundreds of per-file checks.

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) DeleteHealthRecordByLibraryPath 

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

DeleteHealthRecordByLibraryPath deletes the health record matching the given absolute library path. Returns the file_path of the deleted record so the caller can use it for metadata cleanup.

func (*HealthRepository) DeleteHealthRecordsBulk 

func (r *HealthRepository) DeleteHealthRecordsBulk(ctx context.Context, filePaths []string) (int64, 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) DeleteHealthRecordsByLibraryPathPrefix 

func (r *HealthRepository) DeleteHealthRecordsByLibraryPathPrefix(ctx context.Context, libraryPathPrefix string) ([]string, int64, error)

DeleteHealthRecordsByLibraryPathPrefix deletes health records where library_path matches the given prefix. Returns the file_paths of deleted records for metadata cleanup, plus the count.

func (*HealthRepository) DeleteHealthRecordsByPrefix 

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

DeleteHealthRecordsByPrefix removes ALL health records at or under the given virtual path prefix. Used by the webhook directory-delete handler, where the whole subtree is genuinely gone and every record (including healthy/relinked ones) must be removed. For failed-import rollback use DeleteUnvalidatedHealthRecordsByPrefix instead.

func (*HealthRepository) DeleteUnvalidatedHealthRecordsByPrefix added in v0.3.0 

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

DeleteUnvalidatedHealthRecordsByPrefix removes only the still-unvalidated placeholder records at or under the prefix — those an ARR webhook has not yet relinked to a real library path (library_path NULL or still equal to the virtual file_path) and that are not in a terminal/repair state. This is the failed-import rollback path: the nzbFolder is deterministic per release (not unique per queue item), so a failed re-import of a release that previously imported successfully shares the subtree. Scoping to unvalidated records protects the prior successful import's healthy/relinked/repair_triggered/corrupted records (and the repair budget they carry) from being wiped by an unrelated failed attempt.

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)

func (*HealthRepository) GetFileHealthByLibraryPath added in v0.3.0 

func (r *HealthRepository) GetFileHealthByLibraryPath(ctx context.Context, libraryPath string) (*FileHealth, error)

GetFileHealthByLibraryPath retrieves health record for a specific file by its library path

func (*HealthRepository) GetFilesByPaths 

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

GetFilesByPaths returns health records for the specified file paths

func (*HealthRepository) GetFilesForLibrarySync 

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

GetFilesForLibrarySync returns all health records to verify their physical presence in the library

func (*HealthRepository) GetFilesForRepairNotification 

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

GetFilesForRepairNotification returns files that need repair notification (repair_triggered status). Records whose repair_retry_count has reached max_repair_retries are returned too: the worker finalizes them as corrupted (prepareRepairNotificationUpdate). Filtering them out here would leave them permanently stuck in repair_triggered — no other query ever selects that 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) 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, strategy string, libraryDir string, maxRetries int) ([]*FileHealth, error)

GetUnhealthyFiles returns files that need health checks

func (*HealthRepository) HasImportHistoryForPath 

func (r *HealthRepository) HasImportHistoryForPath(ctx context.Context, virtualPath string) (bool, error)

HasImportHistoryForPath checks if any import history record exists for the given virtual path. Used to protect symlinks from deletion when an import has been recorded by AltMount, regardless of current metadata state.

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) IncrementStreamingFailureCount 

func (r *HealthRepository) IncrementStreamingFailureCount(ctx context.Context, filePath string, threshold int) (bool, bool, error)

IncrementStreamingFailureCount increments the streaming failure count and returns whether masking/repair threshold was reached

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) LogIndexerImport added in v0.3.0 

func (r *HealthRepository) LogIndexerImport(ctx context.Context, indexer string, status string, errMsg string, downloadID string) error

LogIndexerImport records a success or failure for an indexer persistently.

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, metadataStr *string, revalidate bool) (bool, error)

RelinkFileByFilename updates the file_path and library_path for a record that matches by filename. This is typically called by webhooks during renames or downloads to provide a definitive library path.

revalidate controls what happens to records in repair_triggered/corrupted state:

  • true (Download events — a re-downloaded copy was just imported): reset the record to pending with an immediate check so the fresh copy is validated instead of being destroyed by the next repair re-trigger. retry_count restarts for the new copy, but repair_retry_count is preserved as the per-title repair budget so repeatedly broken re-downloads still escalate to corrupted instead of looping forever.
  • false (Rename events — no new content): preserve repair/corrupted state so a library reorganization cannot wipe repair progress.

func (*HealthRepository) RelinkFileByMetadata added in v0.3.0 

func (r *HealthRepository) RelinkFileByMetadata(ctx context.Context, webMeta *model.WebhookMetadata, filePath, libraryPath string, metadataStr *string, revalidate bool) (bool, error)

RelinkFileByMetadata matches repair_triggered or corrupted records by metadata and updates them to pending.

revalidate controls what happens to records in repair_triggered/corrupted state:

  • true (Download events — a re-downloaded copy was just imported): reset the record to pending with an immediate check so the fresh copy is validated instead of being destroyed by the next repair re-trigger. retry_count restarts for the new copy, but repair_retry_count is preserved as the per-title repair budget so repeatedly broken re-downloads still escalate to corrupted instead of looping forever.
  • false (Rename events — no new content): preserve repair/corrupted state so a library reorganization cannot wipe repair progress.

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) 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) SetFilesCheckingBulk added in v0.3.0 

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

SetFilesCheckingBulk marks many files 'checking' in as few writes as possible. The health cycle calls this once for the whole batch instead of issuing one UPDATE per file, which under SQLite's single writer would serialize N transactions against each other and the final bulk status write. Crash recovery is unchanged: ResetFileAllChecking at worker startup re-arms any record stranded in 'checking'.

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) UnmaskFile 

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

UnmaskFile removes the mask from a file and resets the failure count

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) UpdateFileHealthScheduled 

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

UpdateFileHealthScheduled is like UpdateFileHealth but uses an explicit scheduledAt time instead of datetime('now') for the scheduled_check_at column.

func (*HealthRepository) UpdateFileMetadata added in v0.3.0 

func (r *HealthRepository) UpdateFileMetadata(ctx context.Context, id int64, metadata []byte) error

UpdateFileMetadata updates the metadata column for a 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
	HealthStatusDegraded        HealthStatus = "degraded"         // Missing segments only hit media payload: still playable, no repair
)

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)
	// ExpectedStatus, when non-nil, makes the write conditional on the record still being
	// in that status (the status the worker based its decision on). It closes the TOCTOU
	// window where a concurrent webhook relink, re-import upsert or manual recheck lands
	// between the cycle's read and its write: if the status changed underneath us the
	// guarded UPDATE matches no rows and the concurrent actor's decision wins instead of
	// being silently clobbered (last-writer-wins re-entering the repair loop).
	ExpectedStatus *HealthStatus
}

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"`
	DownloadID  *string   `db:"download_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"`
	Metadata    *string   `db:"metadata"`
	Indexer     *string   `db:"indexer"`
	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 ImportMigration added in v0.3.0 

type ImportMigration struct {
	ID           int64                 `db:"id"`
	Source       string                `db:"source"`        // e.g. "nzbdav"
	ExternalID   string                `db:"external_id"`   // source-specific ID (nzbdav GUID)
	QueueItemID  *int64                `db:"queue_item_id"` // FK → import_queue.id (nullable)
	RelativePath string                `db:"relative_path"` // virtual path when enqueued
	FinalPath    *string               `db:"final_path"`    // storage_path after import
	Status       ImportMigrationStatus `db:"status"`
	Error        *string               `db:"error"`
	CreatedAt    time.Time             `db:"created_at"`
	UpdatedAt    time.Time             `db:"updated_at"`
}

ImportMigration tracks progress of two-phase migrations (e.g. nzbdav → altmount)

type ImportMigrationRepository added in v0.3.0 

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

ImportMigrationRepository handles database operations for import_migrations.

func NewImportMigrationRepository added in v0.3.0 

func NewImportMigrationRepository(db *sql.DB, d Dialect) *ImportMigrationRepository

NewImportMigrationRepository creates a new ImportMigrationRepository.

func (*ImportMigrationRepository) BackfillFromImportQueue added in v0.3.0 

func (r *ImportMigrationRepository) BackfillFromImportQueue(ctx context.Context) (int, error)

BackfillFromImportQueue reads completed import_queue rows that contain a nzbdav_id in their metadata JSON and inserts them as status=imported rows into import_migrations (idempotent via ON CONFLICT IGNORE / INSERT OR IGNORE). Returns the number of rows inserted.

func (*ImportMigrationRepository) DeleteAllBySource added in v0.3.0 

func (r *ImportMigrationRepository) DeleteAllBySource(ctx context.Context, source string) (int64, error)

DeleteAllBySource removes every migration row for a source regardless of status. Returns the number of rows deleted. Use to force a full re-import after the imported files have been deleted from AltMount.

func (*ImportMigrationRepository) DeletePendingBySource added in v0.3.0 

func (r *ImportMigrationRepository) DeletePendingBySource(ctx context.Context, source string) (int64, error)

DeletePendingBySource removes all migration rows for a source that have status='pending'. Returns the number of rows deleted. Use this to clear orphaned rows from a previous import attempt so a fresh import starts clean (imported/symlinks_migrated rows are preserved).

func (*ImportMigrationRepository) ExistsForSource added in v0.3.0 

func (r *ImportMigrationRepository) ExistsForSource(ctx context.Context, source string) (bool, error)

ExistsForSource returns true if any rows exist for the given source.

func (*ImportMigrationRepository) LinkQueueItemID added in v0.3.0 

func (r *ImportMigrationRepository) LinkQueueItemID(ctx context.Context, source string, externalIDs []string, queueItemID int64) error

LinkQueueItemID sets queue_item_id for all migration rows matching (source, externalIDs). Unconditionally overwrites any existing queue_item_id so that re-imports after a cancelled/failed first attempt can re-link to the new queue item. This is safe because IsMigrationCompleted already short-circuits rows with status=imported before we get here.

func (*ImportMigrationRepository) ListByStatus added in v0.3.0 

func (r *ImportMigrationRepository) ListByStatus(ctx context.Context, source string, status ImportMigrationStatus, limit, offset int) ([]*ImportMigration, error)

ListByStatus returns paginated rows for source with the given status.

func (*ImportMigrationRepository) LookupByExternalID added in v0.3.0 

func (r *ImportMigrationRepository) LookupByExternalID(ctx context.Context, source, externalID string) (*ImportMigration, error)

LookupByExternalID returns the migration row for (source, external_id), or nil if not found.

func (*ImportMigrationRepository) MarkFailed added in v0.3.0 

func (r *ImportMigrationRepository) MarkFailed(ctx context.Context, queueItemID int64, errMsg string) error

MarkFailed sets status=failed and error for all rows matching queue_item_id.

func (*ImportMigrationRepository) MarkImported added in v0.3.0 

func (r *ImportMigrationRepository) MarkImported(ctx context.Context, queueItemID int64, finalPath string) error

MarkImported sets status=imported and final_path for all rows matching queue_item_id.

func (*ImportMigrationRepository) MarkSymlinksMigrated added in v0.3.0 

func (r *ImportMigrationRepository) MarkSymlinksMigrated(ctx context.Context, ids []int64) error

MarkSymlinksMigrated sets status=symlinks_migrated for the given row IDs.

func (*ImportMigrationRepository) Stats added in v0.3.0 

func (r *ImportMigrationRepository) Stats(ctx context.Context, source string) (*ImportMigrationStats, error)

Stats returns aggregate counts for a source.

func (*ImportMigrationRepository) Upsert added in v0.3.0 

func (r *ImportMigrationRepository) Upsert(ctx context.Context, row *ImportMigration) (int64, error)

Upsert inserts or updates a migration row keyed by (source, external_id). Returns the row ID.

type ImportMigrationStats added in v0.3.0 

type ImportMigrationStats struct {
	Pending          int
	Imported         int
	Failed           int
	SymlinksMigrated int
	Total            int
}

ImportMigrationStats holds aggregate counts for a source

type ImportMigrationStatus added in v0.3.0 

type ImportMigrationStatus string

ImportMigrationStatus represents the status of a migration item 

const (
	ImportMigrationStatusPending          ImportMigrationStatus = "pending"
	ImportMigrationStatusImported         ImportMigrationStatus = "imported"
	ImportMigrationStatusFailed           ImportMigrationStatus = "failed"
	ImportMigrationStatusSymlinksMigrated ImportMigrationStatus = "symlinks_migrated"
)

type ImportQueueItem 

type ImportQueueItem struct {
	ID                  int64         `db:"id"`
	DownloadID          *string       `db:"download_id"` // GUID/String ID for external tracking (e.g. Sonarr/Radarr)
	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
	TargetPath          *string       `db:"target_path"` // Optional forced symlink destination path
	SkipArrNotification bool          `db:"skip_arr_notification"`
	SkipPostImportLinks bool          `db:"skip_post_import_links"`
	Indexer             *string       `db:"indexer"`
}

ImportQueueItem represents a queued NZB file waiting for import

type IndexerAggregatedHealth added in v0.3.0 

type IndexerAggregatedHealth struct {
	Indexer      string    `json:"indexer"`
	TotalImports int       `json:"total_imports"`
	SuccessCount int       `json:"success_count"`
	FailedCount  int       `json:"failed_count"`
	Last24hCount int       `json:"last_24h_count"`
	SuccessRate  float64   `json:"success_rate"`
	LastSeenAt   time.Time `json:"last_seen_at"`
}

type ProviderHistoricalStat added in v0.3.0 

type ProviderHistoricalStat struct {
	Timestamp       time.Time `db:"timestamp"`
	ProviderID      string    `db:"provider_id"`
	BytesDownloaded int64     `db:"bytes_downloaded"`
}

ProviderHistoricalStat represents aggregated data usage for a provider over a time period

type ProviderSpeedTestStat added in v0.3.0 

type ProviderSpeedTestStat struct {
	ID         int64     `db:"id"`
	ProviderID string    `db:"provider_id"`
	SpeedMbps  float64   `db:"speed_mbps"`
	CreatedAt  time.Time `db:"created_at"`
}

ProviderSpeedTestStat represents a historical speed test result for a provider

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, d Dialect) *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, import_hourly_stats, and provider_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) ClearProviderHourlyStats added in v0.2.0 

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

ClearProviderHourlyStats deletes all records from the provider_hourly_stats table

func (*QueueRepository) DeleteFailedItemsOlderThan 

func (r *QueueRepository) DeleteFailedItemsOlderThan(ctx context.Context, olderThan time.Time) ([]*ImportQueueItem, error)

DeleteFailedItemsOlderThan deletes failed queue items older than the given time. Returns the deleted items so the caller can clean up associated NZB files.

func (*QueueRepository) DeleteImportHistoryOlderThan added in v0.3.0 

func (r *QueueRepository) DeleteImportHistoryOlderThan(ctx context.Context, olderThan time.Time) error

DeleteImportHistoryOlderThan deletes import_history records completed before olderThan.

func (*QueueRepository) DeleteIndexerStats added in v0.3.0 

func (r *QueueRepository) DeleteIndexerStats(ctx context.Context, indexer string) (int64, error)

DeleteIndexerStats deletes all records for a specific indexer and returns the number of rows affected.

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) GetIndexerHealthStats added in v0.3.0 

func (r *QueueRepository) GetIndexerHealthStats(ctx context.Context) ([]*IndexerAggregatedHealth, error)

GetIndexerHealthStats aggregates all historical records to calculate success/failure rates.

func (*QueueRepository) GetPendingQueueItemsByPathPrefix added in v0.3.0 

func (r *QueueRepository) GetPendingQueueItemsByPathPrefix(ctx context.Context, prefix string) ([]*ImportQueueItem, error)

GetPendingQueueItemsByPathPrefix returns pending queue items whose nzb_path starts with prefix.

func (*QueueRepository) GetQueueItem 

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

GetQueueItem retrieves a specific queue item by ID

func (*QueueRepository) GetQueueItemByDownloadID added in v0.3.0 

func (r *QueueRepository) GetQueueItemByDownloadID(ctx context.Context, downloadID string) (*ImportQueueItem, error)

GetQueueItemByDownloadID retrieves a queue item by its DownloadID

func (*QueueRepository) GetQueueItemByNzbPath added in v0.3.0 

func (r *QueueRepository) GetQueueItemByNzbPath(ctx context.Context, nzbPath string) (*ImportQueueItem, error)

GetQueueItemByNzbPath returns the queue item with the given NZB path, or nil if not found.

func (*QueueRepository) GetQueueStats 

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

GetQueueStats returns current queue statistics

func (*QueueRepository) GetUnknownIndexerStatsDownloadIDs added in v0.3.0 

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

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, processing, or paused). It matches by exact nzb_path first, and also by filename suffix to handle the case where ensurePersistentNzb has already moved the file and updated nzb_path from the temp path to the persistent storage path (e.g. /tmp/altmount-uploads/x.nzb → /.nzbs/stremio/42-x.nzb). The LIKE pattern uses "%-filename" (dash wildcard) rather than "%/filename" so it also matches the "<id>-<base>.nzb" naming that ensurePersistentNzb applies on collision.

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) LogIndexerImport added in v0.3.0 

func (r *QueueRepository) LogIndexerImport(ctx context.Context, indexer string, status string, errMsg string, downloadID string) error

LogIndexerImport records a success or failure for an indexer persistently.

func (*QueueRepository) PruneIndexerStats added in v0.3.0 

func (r *QueueRepository) PruneIndexerStats(ctx context.Context, hours int) (int64, error)

PruneIndexerStats deletes records that were created within the last N hours.

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) UpdateImportHistoryIndexerByDownloadID added in v0.3.0 

func (r *QueueRepository) UpdateImportHistoryIndexerByDownloadID(ctx context.Context, downloadID string, indexer string) error

func (*QueueRepository) UpdateIndexerStatsByDownloadID added in v0.3.0 

func (r *QueueRepository) UpdateIndexerStatsByDownloadID(ctx context.Context, downloadID string, indexer string) error

func (*QueueRepository) UpdateQueueItemCategory added in v0.3.0 

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

UpdateQueueItemCategory updates the category and priority of a queue item.

func (*QueueRepository) UpdateQueueItemIndexer added in v0.3.0 

func (r *QueueRepository) UpdateQueueItemIndexer(ctx context.Context, id int64, indexer string) error

UpdateQueueItemIndexer updates the indexer for a queue item by its ID

func (*QueueRepository) UpdateQueueItemIndexerByDownloadID added in v0.3.0 

func (r *QueueRepository) UpdateQueueItemIndexerByDownloadID(ctx context.Context, downloadID string, indexer string) error

UpdateQueueItemIndexerByDownloadID updates the indexer for a queue item by its DownloadID

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, d Dialect) *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) AddProviderBytesToHourlyStat added in v0.2.0 

func (r *Repository) AddProviderBytesToHourlyStat(ctx context.Context, providerID string, bytes int64) error

AddProviderBytesToHourlyStat adds bytes downloaded to the current hour's stat for a specific provider

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) ([]string, int, error)

ClearCompletedQueueItems removes completed items from the queue and returns the paths of deleted rows.

func (*Repository) ClearDailyStats 

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

ClearDailyStats deletes all records from the import_daily_stats, import_hourly_stats, and provider_hourly_stats tables

func (*Repository) ClearDailyStatsSince 

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

ClearDailyStatsSince deletes records from the import_daily_stats table since the specified time

func (*Repository) ClearFailedQueueItems 

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

ClearFailedQueueItems removes failed items from the queue and returns the paths of deleted rows.

func (*Repository) ClearHourlyStats 

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

ClearHourlyStats deletes all records from the import_hourly_stats table

func (*Repository) ClearHourlyStatsSince 

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

ClearHourlyStatsSince deletes records from the import_hourly_stats table since the specified time

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) ([]string, int, error)

ClearPendingQueueItems removes pending items from the queue and returns the paths of deleted rows.

func (*Repository) ClearProviderHourlyStats added in v0.2.0 

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

ClearProviderHourlyStats deletes all records from the provider_hourly_stats table

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) DeleteIndexerStats added in v0.3.0 

func (r *Repository) DeleteIndexerStats(ctx context.Context, indexer string) (int64, error)

DeleteIndexerStats deletes all records for a specific indexer and returns the number of rows affected.

func (*Repository) DeleteQueueItemsByPath added in v0.3.0 

func (r *Repository) DeleteQueueItemsByPath(ctx context.Context, path string) error

DeleteQueueItemsByPath removes items from the queue matching the given path

func (*Repository) GetExpiredStremioQueueItems 

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

GetExpiredStremioQueueItems returns completed Stremio queue items whose completed_at is older than ttlHours. Items are identified as Stremio-originated by their download_id having the "stremio:" prefix set when the addon enqueues an 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) GetImportHistoryByDownloadID added in v0.2.0 

func (r *Repository) GetImportHistoryByDownloadID(ctx context.Context, downloadID string) (*ImportHistory, error)

GetImportHistoryByDownloadID retrieves an import history item by its DownloadID

func (*Repository) GetImportHistoryByNzbID added in v0.3.0 

func (r *Repository) GetImportHistoryByNzbID(ctx context.Context, nzbID int64) (*ImportHistory, error)

GetImportHistoryByNzbID retrieves an import history item by its original NZB ID (the integer ID of the queue row that produced this history entry). Returns (nil, nil) when no matching row exists.

func (*Repository) GetImportHistoryByPath 

func (r *Repository) GetImportHistoryByPath(ctx context.Context, virtualPath string) (*ImportHistory, error)

GetImportHistoryByPath retrieves an import history item by its virtual path

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) GetIndexerHealthStats added in v0.3.0 

func (r *Repository) GetIndexerHealthStats(ctx context.Context) ([]*IndexerAggregatedHealth, error)

GetIndexerHealthStats aggregates all historical records to calculate success/failure rates.

func (*Repository) GetOldestProviderStatDates added in v0.3.0 

func (r *Repository) GetOldestProviderStatDates(ctx context.Context) (map[string]time.Time, error)

GetOldestProviderStatDates returns the date of the oldest record per provider in provider_hourly_stats

func (*Repository) GetOldestStatDate added in v0.3.0 

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

GetOldestStatDate returns the date of the oldest record in import_daily_stats or import_history

func (*Repository) GetProviderHistoricalStats added in v0.3.0 

func (r *Repository) GetProviderHistoricalStats(ctx context.Context, days int, interval string) ([]*ProviderHistoricalStat, error)

GetProviderHistoricalStats retrieves aggregated data usage per provider over the given number of days. The data is grouped by the specified interval.

func (*Repository) GetProviderHourlyStats added in v0.2.0 

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

GetProviderHourlyStats retrieves provider download statistics for the last N hours

func (*Repository) GetProviderSpeedTestHistory added in v0.3.0 

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

GetProviderSpeedTestHistory retrieves the speed test history over the given number of days

func (*Repository) GetQueueItem 

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

GetQueueItem retrieves a specific queue item by ID

func (*Repository) GetQueueItemByDownloadID added in v0.2.0 

func (r *Repository) GetQueueItemByDownloadID(ctx context.Context, downloadID string) (*ImportQueueItem, error)

GetQueueItemByDownloadID retrieves a queue item by its DownloadID

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) GetUnknownIndexerStatsDownloadIDs added in v0.3.0 

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

GetUnknownIndexerStatsDownloadIDs gets all unique download_ids that have indexer = 'Unknown' in the stats table.

func (*Repository) HasActiveOrRecentQueueItemForStoragePath added in v0.3.0 

func (r *Repository) HasActiveOrRecentQueueItemForStoragePath(
	ctx context.Context,
	relPath string,
	completedGrace time.Duration,
) (bool, error)

HasActiveOrRecentQueueItemForStoragePath reports whether any import_queue row references a storage_path equal to (or nested under) the given relative path and is either still active (pending/processing/paused/retrying) or was completed within the supplied grace window. It is used to guard against arr webhook directory deletions racing a sibling import that just wrote into the same release folder.

relPath is matched as a path prefix; both "<relPath>" and "/<relPath>" are considered (storage_path is typically absolute, e.g. "/moviesHQ/release/...", while webhook-normalized paths are relative).

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) ListRecentImportHistory 

func (r *Repository) ListRecentImportHistory(ctx context.Context, minutes int, category string) ([]*ImportHistory, error)

ListRecentImportHistory retrieves import history items completed within the last N minutes.

func (*Repository) LogIndexerImport added in v0.3.0 

func (r *Repository) LogIndexerImport(ctx context.Context, indexer string, status string, errMsg string, downloadID string) error

LogIndexerImport records a success or failure for an indexer persistently.

func (*Repository) PruneIndexerStats added in v0.3.0 

func (r *Repository) PruneIndexerStats(ctx context.Context, hours int) (int64, error)

PruneIndexerStats deletes records that were created within the last N hours.

func (*Repository) RecordProviderSpeedTest added in v0.3.0 

func (r *Repository) RecordProviderSpeedTest(ctx context.Context, providerID string, speedMbps float64) error

RecordProviderSpeedTest saves a speed test result for a provider

func (*Repository) RemoveFromHistory 

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

RemoveFromHistory removes a record from import_history by its own ID

func (*Repository) RemoveFromHistoryByDownloadID added in v0.2.0 

func (r *Repository) RemoveFromHistoryByDownloadID(ctx context.Context, downloadID string) (int64, error)

RemoveFromHistoryByDownloadID removes a record from import_history by its DownloadID

func (*Repository) RemoveFromHistoryByNzbID 

func (r *Repository) RemoveFromHistoryByNzbID(ctx context.Context, nzbID int64) (int64, error)

RemoveFromHistoryByNzbID removes a record from import_history by its original NZB ID

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) (*BulkOperationResult, error)

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

func (*Repository) RemoveFromQueueByDownloadID added in v0.2.0 

func (r *Repository) RemoveFromQueueByDownloadID(ctx context.Context, downloadID string) error

RemoveFromQueueByDownloadID removes an item from the queue by its DownloadID

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) UpdateImportHistoryIndexerByDownloadID added in v0.3.0 

func (r *Repository) UpdateImportHistoryIndexerByDownloadID(ctx context.Context, downloadID string, indexer string) error

UpdateImportHistoryIndexerByDownloadID updates the indexer for an import history item by its DownloadID

func (*Repository) UpdateIndexerStatsByDownloadID added in v0.3.0 

func (r *Repository) UpdateIndexerStatsByDownloadID(ctx context.Context, downloadID string, indexer string) error

UpdateIndexerStatsByDownloadID updates the indexer for a success or failure record in indexer_import_stats by its DownloadID

func (*Repository) UpdateQueueItemIndexer added in v0.3.0 

func (r *Repository) UpdateQueueItemIndexer(ctx context.Context, id int64, indexer string) error

UpdateQueueItemIndexer updates the indexer for a queue item by its ID

func (*Repository) UpdateQueueItemIndexerByDownloadID added in v0.3.0 

func (r *Repository) UpdateQueueItemIndexerByDownloadID(ctx context.Context, downloadID string, indexer string) error

UpdateQueueItemIndexerByDownloadID updates the indexer for a queue item by its DownloadID

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) UpdateQueueItemsPriorityBulk added in v0.3.0 

func (r *Repository) UpdateQueueItemsPriorityBulk(ctx context.Context, ids []int64, priority QueuePriority) (updated int, skipped int, err error)

UpdateQueueItemsPriorityBulk updates the priority of multiple queue items, skipping any that are currently processing. Returns the number of items updated and the number skipped.

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 StoreRefRepository added in v0.3.0 

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

StoreRefRepository tracks reference counts on .nzbz store files. When all .meta files that reference a given store are deleted, the refcount hits 0 and the caller can delete the store file.

func NewStoreRefRepository added in v0.3.0 

func NewStoreRefRepository(db *sql.DB, d Dialect) *StoreRefRepository

NewStoreRefRepository creates a new StoreRefRepository.

func (*StoreRefRepository) DecStoreRef added in v0.3.0 

func (r *StoreRefRepository) DecStoreRef(ctx context.Context, storePath string) (int64, error)

DecStoreRef decrements the ref_count for storePath by 1. If the resulting count is ≤ 0, the row is deleted and 0 is returned. Returns the new count (0 if the row was deleted or did not exist).

func (*StoreRefRepository) GetStoreRefCount added in v0.3.0 

func (r *StoreRefRepository) GetStoreRefCount(ctx context.Context, storePath string) (int64, error)

GetStoreRefCount returns the current ref_count for storePath. Returns 0 (and no error) if the row does not exist.

func (*StoreRefRepository) IncStoreRef added in v0.3.0 

func (r *StoreRefRepository) IncStoreRef(ctx context.Context, storePath string) error

IncStoreRef increments the ref_count for storePath by 1. If no row exists yet, it inserts one with ref_count = 1.

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
	UpdateTypeDegraded      UpdateType = 6 // playable with glitches; no repair, periodic re-check
)

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, d Dialect) *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.