diff --git a/src/lib/server/db/migrations.ts b/src/lib/server/db/migrations.ts index 24acfcd..6fad909 100644 --- a/src/lib/server/db/migrations.ts +++ b/src/lib/server/db/migrations.ts @@ -13,6 +13,7 @@ import { migration as migration008 } from './migrations/008_create_database_inst import { migration as migration009 } from './migrations/009_add_personal_access_token.ts'; import { migration as migration010 } from './migrations/010_add_is_private.ts'; import { migration as migration011 } from './migrations/011_create_upgrade_configs.ts'; +import { migration as migration012 } from './migrations/012_add_upgrade_last_run.ts'; export interface Migration { version: number; @@ -241,7 +242,8 @@ export function loadMigrations(): Migration[] { migration008, migration009, migration010, - migration011 + migration011, + migration012 ]; // Sort by version number diff --git a/src/lib/server/db/migrations/012_add_upgrade_last_run.ts b/src/lib/server/db/migrations/012_add_upgrade_last_run.ts new file mode 100644 index 0000000..3fb2948 --- /dev/null +++ b/src/lib/server/db/migrations/012_add_upgrade_last_run.ts @@ -0,0 +1,46 @@ +import type { Migration } from '../migrations.ts'; + +/** + * Migration 012: Add last_run_at to upgrade_configs + * + * Adds timestamp tracking for when each upgrade config was last executed. + * Used by the upgrade manager job to determine if enough time has passed + * based on the config's schedule. + */ + +export const migration: Migration = { + version: 12, + name: 'Add last_run_at to upgrade_configs', + + up: ` + ALTER TABLE upgrade_configs + ADD COLUMN last_run_at DATETIME; + `, + + down: ` + -- SQLite doesn't support DROP COLUMN easily, so we recreate the table + CREATE TABLE upgrade_configs_backup ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + arr_instance_id INTEGER NOT NULL UNIQUE, + enabled INTEGER NOT NULL DEFAULT 0, + schedule INTEGER NOT NULL DEFAULT 360, + filter_mode TEXT NOT NULL DEFAULT 'round_robin', + filters TEXT NOT NULL DEFAULT '[]', + current_filter_index INTEGER NOT NULL DEFAULT 0, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY (arr_instance_id) REFERENCES arr_instances(id) ON DELETE CASCADE + ); + + INSERT INTO upgrade_configs_backup + SELECT id, arr_instance_id, enabled, schedule, filter_mode, filters, + current_filter_index, created_at, updated_at + FROM upgrade_configs; + + DROP TABLE upgrade_configs; + + ALTER TABLE upgrade_configs_backup RENAME TO upgrade_configs; + + CREATE INDEX idx_upgrade_configs_arr_instance ON upgrade_configs(arr_instance_id); + ` +}; diff --git a/src/lib/server/db/queries/upgradeConfigs.ts b/src/lib/server/db/queries/upgradeConfigs.ts index 082f7ec..2cda551 100644 --- a/src/lib/server/db/queries/upgradeConfigs.ts +++ b/src/lib/server/db/queries/upgradeConfigs.ts @@ -12,6 +12,7 @@ interface UpgradeConfigRow { filter_mode: string; filters: string; current_filter_index: number; + last_run_at: string | null; created_at: string; updated_at: string; } @@ -39,6 +40,7 @@ function rowToConfig(row: UpgradeConfigRow): UpgradeConfig { filterMode: row.filter_mode as FilterMode, filters: JSON.parse(row.filters) as FilterConfig[], currentFilterIndex: row.current_filter_index, + lastRunAt: row.last_run_at, createdAt: row.created_at, updatedAt: row.updated_at }; @@ -196,5 +198,31 @@ export const upgradeConfigsQueries = { 'UPDATE upgrade_configs SET current_filter_index = 0, updated_at = CURRENT_TIMESTAMP WHERE arr_instance_id = ?', arrInstanceId ); + }, + + /** + * Update last_run_at to current timestamp + */ + updateLastRun(arrInstanceId: number): void { + db.execute( + 'UPDATE upgrade_configs SET last_run_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE arr_instance_id = ?', + arrInstanceId + ); + }, + + /** + * Get all enabled configs that are due to run + * A config is due if: last_run_at is null OR (now - last_run_at) >= schedule minutes + */ + getDueConfigs(): UpgradeConfig[] { + const rows = db.query(` + SELECT * FROM upgrade_configs + WHERE enabled = 1 + AND ( + last_run_at IS NULL + OR (julianday('now') - julianday(last_run_at)) * 24 * 60 >= schedule + ) + `); + return rows.map(rowToConfig); } }; diff --git a/src/lib/server/db/schema.sql b/src/lib/server/db/schema.sql index 1577f61..4d71ca7 100644 --- a/src/lib/server/db/schema.sql +++ b/src/lib/server/db/schema.sql @@ -247,6 +247,7 @@ CREATE TABLE upgrade_configs ( -- State tracking current_filter_index INTEGER NOT NULL DEFAULT 0, -- For round-robin mode + last_run_at DATETIME, -- When upgrade job last ran (Migration 012) -- Metadata created_at DATETIME DEFAULT CURRENT_TIMESTAMP, diff --git a/src/lib/server/jobs/definitions/upgradeManager.ts b/src/lib/server/jobs/definitions/upgradeManager.ts new file mode 100644 index 0000000..d7c9826 --- /dev/null +++ b/src/lib/server/jobs/definitions/upgradeManager.ts @@ -0,0 +1,79 @@ +import { logger } from '$logger/logger.ts'; +import { runUpgradeManager } from '../logic/upgradeManager.ts'; +import type { JobDefinition, JobResult } from '../types.ts'; + +/** + * Upgrade manager job + * Checks for upgrade configs that are due to run and processes them + * Each config has its own schedule - this job just checks every 30 minutes + */ +export const upgradeManagerJob: JobDefinition = { + name: 'upgrade_manager', + description: 'Process library upgrades for arr instances', + schedule: '*/30 minutes', + + handler: async (): Promise => { + try { + await logger.info('Starting upgrade manager job', { + source: 'UpgradeManagerJob' + }); + + const result = await runUpgradeManager(); + + // Build output message + if (result.totalProcessed === 0) { + return { + success: true, + output: 'No upgrade configs due to run' + }; + } + + const message = `Processed ${result.totalProcessed} config(s): ${result.successCount} successful, ${result.failureCount} failed, ${result.skippedCount} skipped`; + + // Log individual results + for (const instance of result.instances) { + if (instance.success) { + await logger.info(`Upgrade completed for "${instance.instanceName}"`, { + source: 'UpgradeManagerJob', + meta: { + instanceId: instance.instanceId, + filterName: instance.filterName, + itemsSearched: instance.itemsSearched + } + }); + } else { + await logger.warn(`Upgrade skipped/failed for "${instance.instanceName}": ${instance.error}`, { + source: 'UpgradeManagerJob', + meta: { + instanceId: instance.instanceId, + error: instance.error + } + }); + } + } + + // Consider job failed only if all configs failed + if (result.failureCount > 0 && result.successCount === 0) { + return { + success: false, + error: message + }; + } + + return { + success: true, + output: message + }; + } catch (error) { + await logger.error('Upgrade manager job failed', { + source: 'UpgradeManagerJob', + meta: { error: error instanceof Error ? error.message : String(error) } + }); + + return { + success: false, + error: error instanceof Error ? error.message : String(error) + }; + } + } +}; diff --git a/src/lib/server/jobs/init.ts b/src/lib/server/jobs/init.ts index 1c5006e..b3a59cf 100644 --- a/src/lib/server/jobs/init.ts +++ b/src/lib/server/jobs/init.ts @@ -7,6 +7,7 @@ import { cleanupLogsJob } from './definitions/cleanupLogs.ts'; import { createBackupJob } from './definitions/createBackup.ts'; import { cleanupBackupsJob } from './definitions/cleanupBackups.ts'; import { syncDatabasesJob } from './definitions/syncDatabases.ts'; +import { upgradeManagerJob } from './definitions/upgradeManager.ts'; /** * Register all job definitions @@ -17,6 +18,7 @@ function registerAllJobs(): void { jobRegistry.register(createBackupJob); jobRegistry.register(cleanupBackupsJob); jobRegistry.register(syncDatabasesJob); + jobRegistry.register(upgradeManagerJob); } /** diff --git a/src/lib/server/jobs/logic/upgradeManager.ts b/src/lib/server/jobs/logic/upgradeManager.ts new file mode 100644 index 0000000..b987331 --- /dev/null +++ b/src/lib/server/jobs/logic/upgradeManager.ts @@ -0,0 +1,209 @@ +/** + * Core logic for the upgrade manager job + * Checks for upgrade configs that are due to run and processes them + */ + +import { upgradeConfigsQueries } from '$db/queries/upgradeConfigs.ts'; +import { arrInstancesQueries } from '$db/queries/arrInstances.ts'; +import { logger } from '$logger/logger.ts'; +import type { FilterConfig, UpgradeConfig } from '$lib/shared/filters'; + +export interface UpgradeInstanceStatus { + instanceId: number; + instanceName: string; + success: boolean; + filterName?: string; + itemsSearched?: number; + error?: string; +} + +export interface UpgradeManagerResult { + totalProcessed: number; + successCount: number; + failureCount: number; + skippedCount: number; + instances: UpgradeInstanceStatus[]; +} + +/** + * Get the next filter to run based on the config's mode + */ +function getNextFilter(config: UpgradeConfig): FilterConfig | null { + const enabledFilters = config.filters.filter((f) => f.enabled); + + if (enabledFilters.length === 0) { + return null; + } + + if (config.filterMode === 'random') { + // Random: pick a random filter + const randomIndex = Math.floor(Math.random() * enabledFilters.length); + return enabledFilters[randomIndex]; + } + + // Round robin: use currentFilterIndex + const index = config.currentFilterIndex % enabledFilters.length; + return enabledFilters[index]; +} + +/** + * Process a single upgrade config + */ +async function processUpgradeConfig(config: UpgradeConfig): Promise { + const instance = arrInstancesQueries.getById(config.arrInstanceId); + + if (!instance) { + return { + instanceId: config.arrInstanceId, + instanceName: 'Unknown', + success: false, + error: 'Arr instance not found' + }; + } + + if (!instance.enabled) { + return { + instanceId: config.arrInstanceId, + instanceName: instance.name, + success: false, + error: 'Arr instance is disabled' + }; + } + + // Get the filter to run + const filter = getNextFilter(config); + + if (!filter) { + return { + instanceId: config.arrInstanceId, + instanceName: instance.name, + success: false, + error: 'No enabled filters' + }; + } + + await logger.info(`Processing upgrade for "${instance.name}" with filter "${filter.name}"`, { + source: 'UpgradeManager', + meta: { + instanceId: instance.id, + instanceType: instance.type, + filterName: filter.name, + selector: filter.selector, + count: filter.count + } + }); + + try { + // TODO: Implement actual upgrade logic: + // 1. Fetch library items from arr instance + // 2. Apply filter rules to get matching items + // 3. Apply selector to pick items (random, oldest, etc.) + // 4. Check search cooldown (via arr tags) + // 5. Trigger search for selected items + // 6. Tag items with search timestamp + + await logger.debug('Upgrade config details', { + source: 'UpgradeManager', + meta: { + instanceId: instance.id, + filter: { + id: filter.id, + name: filter.name, + cutoff: filter.cutoff, + searchCooldown: filter.searchCooldown, + selector: filter.selector, + count: filter.count, + rulesCount: filter.group.children.length + } + } + }); + + // Update filter index for round-robin mode + if (config.filterMode === 'round_robin') { + upgradeConfigsQueries.incrementFilterIndex(config.arrInstanceId); + } + + // Update last run timestamp + upgradeConfigsQueries.updateLastRun(config.arrInstanceId); + + return { + instanceId: instance.id, + instanceName: instance.name, + success: true, + filterName: filter.name, + itemsSearched: 0 // TODO: Return actual count when implemented + }; + } catch (error) { + await logger.error(`Failed to process upgrade for "${instance.name}"`, { + source: 'UpgradeManager', + meta: { + instanceId: instance.id, + error: error instanceof Error ? error.message : String(error) + } + }); + + return { + instanceId: instance.id, + instanceName: instance.name, + success: false, + filterName: filter.name, + error: error instanceof Error ? error.message : 'Unknown error' + }; + } +} + +/** + * Run the upgrade manager + * Checks for configs that are due and processes them + */ +export async function runUpgradeManager(): Promise { + const dueConfigs = upgradeConfigsQueries.getDueConfigs(); + + const totalProcessed = dueConfigs.length; + let successCount = 0; + let failureCount = 0; + let skippedCount = 0; + const statuses: UpgradeInstanceStatus[] = []; + + if (dueConfigs.length === 0) { + await logger.debug('No upgrade configs due to run', { + source: 'UpgradeManager' + }); + + return { + totalProcessed: 0, + successCount: 0, + failureCount: 0, + skippedCount: 0, + instances: [] + }; + } + + await logger.info(`Found ${dueConfigs.length} upgrade config(s) to process`, { + source: 'UpgradeManager', + meta: { + configIds: dueConfigs.map((c) => c.arrInstanceId) + } + }); + + for (const config of dueConfigs) { + const status = await processUpgradeConfig(config); + statuses.push(status); + + if (status.success) { + successCount++; + } else if (status.error?.includes('disabled') || status.error?.includes('No enabled')) { + skippedCount++; + } else { + failureCount++; + } + } + + return { + totalProcessed, + successCount, + failureCount, + skippedCount, + instances: statuses + }; +} diff --git a/src/lib/shared/filters.ts b/src/lib/shared/filters.ts index be290b1..f4c9e11 100644 --- a/src/lib/shared/filters.ts +++ b/src/lib/shared/filters.ts @@ -55,6 +55,7 @@ export interface UpgradeConfig { filterMode: FilterMode; filters: FilterConfig[]; currentFilterIndex: number; + lastRunAt?: string | null; createdAt?: string; updatedAt?: string; }