diff --git a/src/backend/common/AbstractComponent.ts b/src/backend/common/AbstractComponent.ts index da33cf20..4486fbb9 100644 --- a/src/backend/common/AbstractComponent.ts +++ b/src/backend/common/AbstractComponent.ts @@ -30,7 +30,12 @@ import { objectsEqual } from "../utils/DataUtils.js"; import { RetentionOptions } from "./infrastructure/config/database.js"; import { getRetentionCompactAfterFromEnv, getRetentionDeleteAfterFromEnv, isCompactableProperty, parseRetentionOptions, parseRetentionOptionsDurations } from "./database/Database.js"; import { DbConcrete } from "./database/drizzle/drizzleUtils.js"; -import { ComponentSelect } from "./database/drizzle/drizzleTypes.js"; +import { ComponentSelect, FindWhere, PlaySelect } from "./database/drizzle/drizzleTypes.js"; +import { DrizzlePlayRepository } from "./database/drizzle/repositories/PlayRepository.js"; +import { ClientType } from "./infrastructure/config/client/clients.js"; +import { SourceType } from "./infrastructure/config/source/sources.js"; +import { generateComponentEntity } from "./database/drizzle/entityUtils.js"; +import { components } from "./database/drizzle/schema/schema.js"; export type AbstractComponentConfig = (CommonClientConfig | CommonSourceConfig) & { transformManager?: TransformerManager }; @@ -46,6 +51,9 @@ export default abstract class AbstractComponent extends AbstractInitializable { protected dbComponent: ComponentSelect; protected retentionOpts: RetentionOptions; + protected componentType: 'source' | 'client'; + type: ClientType | SourceType; + protected constructor(config: AbstractComponentConfig) { super(config); this.transformManager = config.transformManager ?? getRoot().items.transformerManager; @@ -73,7 +81,25 @@ export default abstract class AbstractComponent extends AbstractInitializable { protected async doBuildDatabase(): Promise { super.doBuildDatabase(); - return; + let where: FindWhere<'components'> = { + mode: this.componentType, + type: this.type, + uid: this.config.id ?? this.config.name + }; + const component = await this.db.query.components.findFirst({ + where + }); + if(component !== undefined) { + this.dbComponent = component; + return; + } + + this.dbComponent = (await this.db.insert(components).values(generateComponentEntity({ + uid: this.config.id ?? this.config.name, + mode: 'source', + type: this.type, + name: this.config.name + })).returning())[0]; } public buildTransformRules() { @@ -176,6 +202,15 @@ export default abstract class AbstractComponent extends AbstractInitializable { } } + public retentionCleanup = async () => { + try { + const repo = new DrizzlePlayRepository(this.db, {logger: this.logger}); + await repo.retentionCleanup(this.dbComponent.id, this.componentType, this.retentionOpts); + } catch (e) { + this.logger.warn(new Error('Failed to do retention cleanup', {cause: e})); + } + } + protected transformPartToStrong(data: any) { if(data === undefined) { return undefined; diff --git a/src/backend/common/database/drizzle/migrations/20260424124953_flippant_lifeguard/migration.sql b/src/backend/common/database/drizzle/migrations/20260425013931_aspiring_jack_power/migration.sql similarity index 99% rename from src/backend/common/database/drizzle/migrations/20260424124953_flippant_lifeguard/migration.sql rename to src/backend/common/database/drizzle/migrations/20260425013931_aspiring_jack_power/migration.sql index eacab3bf..3ba2650e 100644 --- a/src/backend/common/database/drizzle/migrations/20260424124953_flippant_lifeguard/migration.sql +++ b/src/backend/common/database/drizzle/migrations/20260425013931_aspiring_jack_power/migration.sql @@ -28,6 +28,7 @@ CREATE TABLE `plays` ( `play` text NOT NULL, `state` text NOT NULL, `parentId` integer, + `compacted` text, CONSTRAINT `fk_plays_componentId_components_id_fk` FOREIGN KEY (`componentId`) REFERENCES `components`(`id`) ON UPDATE CASCADE ON DELETE CASCADE, CONSTRAINT `fk_plays_parentId_plays_id_fk` FOREIGN KEY (`parentId`) REFERENCES `plays`(`id`) ON UPDATE CASCADE ON DELETE SET NULL ); diff --git a/src/backend/common/database/drizzle/migrations/20260424124953_flippant_lifeguard/snapshot.json b/src/backend/common/database/drizzle/migrations/20260425013931_aspiring_jack_power/snapshot.json similarity index 97% rename from src/backend/common/database/drizzle/migrations/20260424124953_flippant_lifeguard/snapshot.json rename to src/backend/common/database/drizzle/migrations/20260425013931_aspiring_jack_power/snapshot.json index 3d9c50b4..de13e82a 100644 --- a/src/backend/common/database/drizzle/migrations/20260424124953_flippant_lifeguard/snapshot.json +++ b/src/backend/common/database/drizzle/migrations/20260425013931_aspiring_jack_power/snapshot.json @@ -1,7 +1,7 @@ { "version": "7", "dialect": "sqlite", - "id": "c2744e30-39a9-4d83-8b7c-229ab78fe6e5", + "id": "7a00c1dc-5940-4e2a-8d13-a15faadbe71c", "prevIds": [ "00000000-0000-0000-0000-000000000000" ], @@ -242,6 +242,16 @@ "entityType": "columns", "table": "plays" }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "compacted", + "entityType": "columns", + "table": "plays" + }, { "type": "integer", "notNull": false, diff --git a/src/backend/common/database/drizzle/repositories/PlayRepository.ts b/src/backend/common/database/drizzle/repositories/PlayRepository.ts index f0a726ad..67bb0165 100644 --- a/src/backend/common/database/drizzle/repositories/PlayRepository.ts +++ b/src/backend/common/database/drizzle/repositories/PlayRepository.ts @@ -1,4 +1,4 @@ -import { Logger, LoggerAppExtras } from "@foxxmd/logging"; +import { childLogger, Logger, LoggerAppExtras } from "@foxxmd/logging"; import { DbConcrete, getDb, runTransaction } from "../drizzleUtils.js"; import { loggerNoop } from "../../../MaybeLogger.js"; import { PlayObject } from "../../../../../core/Atomic.js"; @@ -9,6 +9,8 @@ import { MarkOptional, MarkRequired, PathValue } from "ts-essentials"; import { removeUndefinedKeys } from "../../../../utils.js"; import dayjs, { Dayjs } from "dayjs"; import { RelationsFieldFilter, eq, inArray } from "drizzle-orm"; +import { CompactableProperty, RetentionOptions, retentionPlayTypes } from "../../../infrastructure/config/database.js"; +import { shortTodayAwareFormat } from "../../../../../core/TimeUtils.js"; // https://github.com/drizzle-team/drizzle-orm/issues/695 may be useful for typing models with relations? @@ -119,11 +121,11 @@ export class DrizzlePlayRepository { await this.db.delete(plays).where(inArray(plays.id, ids)); } - findPurgablePlayIds = async (componentId: number, olderThanDate: Dayjs, opts: { countOnly?: boolean, states?: PlaySelect['state'][] } = {}) => { + findPurgablePlayIds = async (componentId: number, olderThanDate: Dayjs, opts: { states?: PlaySelect['state'][], compacted?: string } = {}): Promise => { const { - countOnly = false, - states + states, + compacted } = opts; let where: FindWhere<'plays'> = { @@ -144,6 +146,21 @@ export class DrizzlePlayRepository { } } + if(compacted !== undefined) { + where.compacted = { + OR: [ + { + isNull: true + }, + { + NOT: { + eq: compacted + } + } + ] + } + } + const rows = await this.db.query.plays.findMany({ columns: { id: true @@ -154,11 +171,121 @@ export class DrizzlePlayRepository { } }); - if (countOnly) { - return rows.length; + return rows.map(x => x.id); + } + + public retentionCleanup = async (componentId: number, componentType: string, retentionOpts: RetentionOptions) => { + + const loggerDel = childLogger(this.logger, ['Retention', 'Delete']); + const loggerCom = childLogger(this.logger, ['Retention', 'Compact']); + let summaryDelStates: string[] = []; + let summaryCompactStates: string[] = []; + + loggerDel.debug('Starting cleanup...'); + for(const retentionType of retentionPlayTypes) { + try { + const date = dayjs().subtract(retentionOpts.deleteAfter[retentionType].asMilliseconds()); + let state: PlaySelect['state']; + if(retentionType === 'completed') { + state = componentType === 'source' ? 'discovered' : 'scrobbled'; + } else { + state = retentionType; + } + loggerDel.trace(`Finding '${retentionType}' plays older than ${shortTodayAwareFormat(date)}...`); + const ids = await this.findPurgablePlayIds(componentId, date, {states: [state]}); + loggerDel.trace(`Found ${ids.length} '${retentionType}' plays`); + if(ids.length === 0) { + summaryDelStates.push(`No '${retentionType}' Plays older than ${shortTodayAwareFormat(date)}`); + } else { + loggerDel.trace(`Deleting ${ids.length} '${retentionType}' plays`); + await this.deletePlays(ids); + loggerDel.trace(`'${retentionType}' plays deleted!`); + summaryDelStates.push(`${ids.length} '${retentionType}' Plays older than ${shortTodayAwareFormat(date)}`) + } + } catch (e) { + loggerDel.warn(new Error(`Failed to perform retention cleanup on '${retentionType}' type`, {cause: e})); + } } + loggerDel.verbose(`Cleanup done! Summary:\n${summaryDelStates.join(' | ')}`); - return rows.map(x => x.id); + if(retentionOpts.compact.length === 0) { + loggerCom.debug('Compacting is disabled, skipping cleanup.'); + return; + } + + const compactTypes = retentionOpts.compact; + let compactedFlags: CompactableProperty[] = []; + if(compactTypes.includes('input')) { + compactedFlags.push('input'); + } + if(compactTypes.includes('transform')) { + compactedFlags.push('transform'); + } + + loggerCom.debug('Starting cleanup...'); + for(const retentionType of retentionPlayTypes) { + if(retentionOpts.compactAfter[retentionType] === false) { + summaryCompactStates.push(`Skipped ${retentionType}`); + continue; + } + try { + const date = dayjs().subtract(retentionOpts.compactAfter[retentionType].asMilliseconds()); + let state: PlaySelect['state']; + if(retentionType === 'completed') { + state = componentType === 'source' ? 'discovered' : 'scrobbled'; + } else { + state = retentionType; + } + loggerCom.trace(`Finding '${retentionType}' plays older than ${shortTodayAwareFormat(date)}...`); + const ids = await this.findPurgablePlayIds(componentId, date, {compacted: compactedFlags.join('-'), states: [state]}); + loggerCom.trace(`Found ${ids.length} '${retentionType}' plays`); + if(ids.length === 0) { + summaryDelStates.push(`No '${retentionType}' Plays older than ${shortTodayAwareFormat(date)}`); + } else { + for(const id of ids) { + let compactedPlay: PlayObject; + if(compactTypes.includes('input')) { + this.db.update(playInputs).set({ + data: {removedReason: 'Removed by compaction'} + }).where(eq(playInputs.playId, id)); + } + if(compactTypes.includes('transform')) { + const playRow = await this.db.query.plays.findFirst({where: {id: id}}); + if(playRow === undefined) { + // uhh shouldn't be + loggerCom.warn(`No Play found with ID ${id}, but it should have been...`); + continue; + } + + const compactedPlay: PlayObject = playRow.play; + compactedPlay.meta.lifecycle.steps = compactedPlay.meta.lifecycle.steps.map(x => { + if(x.inputs == undefined) { + return x; + } + return {...x, inputs: x.inputs.map(y => ({type: y.type, input: 'Removed by compaction'}))}; + }); + } + + const updater = this.db.update(plays); + const vals: Parameters[0] = { + compacted: compactedFlags.join('-') + }; + if(compactedPlay !== undefined) { + vals.play = compactedPlay; + } + this.db.update(plays).set(vals) + } + loggerCom.trace(`Compacted ${ids.length} '${retentionType}' plays`); + await this.deletePlays(ids); + loggerCom.trace(`'${retentionType}' plays deleted!`); + summaryCompactStates.push(`${ids.length} '${retentionType}' Plays older than ${shortTodayAwareFormat(date)}`) + } + } catch (e) { + loggerCom.warn(new Error(`Failed to perform retention cleanup on '${retentionType}' type`, {cause: e})); + } + } + + loggerCom.verbose(`Cleanup done! Summary:\n${summaryDelStates.join(' | ')}`); } } diff --git a/src/backend/common/database/drizzle/schema/schema.ts b/src/backend/common/database/drizzle/schema/schema.ts index 57833e81..feb973af 100644 --- a/src/backend/common/database/drizzle/schema/schema.ts +++ b/src/backend/common/database/drizzle/schema/schema.ts @@ -31,7 +31,8 @@ export const plays = sqliteTable("plays", { play: text({ mode: 'json' }).notNull().$type(), state: text({enum: ['queued','discovered','scrobbled','failed','duped']}).notNull(), // https://orm.drizzle.team/docs/indexes-constraints#foreign-key - parentId: integer().references((): AnySQLiteColumn => plays.id, {onDelete: 'set null', onUpdate: 'cascade'}) + parentId: integer().references((): AnySQLiteColumn => plays.id, {onDelete: 'set null', onUpdate: 'cascade'}), + compacted: text() }, (table) => [ index("play_parent_id_idx").on(table.parentId), index("play_component_id_idx").on(table.componentId), diff --git a/src/backend/common/infrastructure/config/database.ts b/src/backend/common/infrastructure/config/database.ts index aa05d1a9..92085e1e 100644 --- a/src/backend/common/infrastructure/config/database.ts +++ b/src/backend/common/infrastructure/config/database.ts @@ -1,6 +1,9 @@ import { Duration } from "dayjs/plugin/duration.js"; import { DurationValue } from "../Atomic.js"; +export type RetentionPlayType = 'failed' | 'completed' | 'duped'; +export const retentionPlayTypes: RetentionPlayType[] = ['failed','completed','duped']; + export type RetentionValueUnparsed = DurationValue | Duration | false; export type RetentionValue = Duration | false; export interface RententionGranular { diff --git a/src/backend/scrobblers/AbstractScrobbleClient.ts b/src/backend/scrobblers/AbstractScrobbleClient.ts index 6062a9c8..9c105b46 100644 --- a/src/backend/scrobblers/AbstractScrobbleClient.ts +++ b/src/backend/scrobblers/AbstractScrobbleClient.ts @@ -69,9 +69,6 @@ import {serializeError} from 'serialize-error'; import { DEFAULT_NEW_PADDING, groupPlaysToTimeRanges } from "../utils/ListenFetchUtils.js"; import { spawn, catchAbortError, isAbortError, rethrowAbortError, delay, forever, AbortError, throwIfAborted } from 'abort-controller-x'; import { Queue, MemoryStorage } from '@platformatic/job-queue' -import { FindOne, FindWhere } from "../common/database/drizzle/drizzleTypes.js"; -import { components } from "../common/database/drizzle/schema/schema.js"; -import { generateComponentEntity } from "../common/database/drizzle/entityUtils.js"; type PlatformMappedPlays = Map; type NowPlayingQueue = Map; @@ -82,7 +79,7 @@ const platformTruncate = truncateStringToLength(10); export default abstract class AbstractScrobbleClient extends AbstractComponent implements Authenticatable { name: string; - type: ClientType; + declare type: ClientType; scheduler: ToadScheduler = new ToadScheduler(); @@ -141,8 +138,11 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i existing: staggerMapper({concurrency: 2}) } + declare protected componentType: 'client'; + constructor(type: any, name: any, config: CommonClientConfig, notifier: Notifiers, emitter: EventEmitter, logger: Logger) { super(config); + this.componentType = 'client'; this.type = type; this.name = name; this.logger = childLogger(logger, this.getIdentifier()); @@ -435,29 +435,6 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i return `Scrobbles from Cache: ${cachedQLength} Queue | ${cachedDLength} Dead Letter`; } - protected async doBuildDatabase(): Promise { - super.doBuildDatabase(); - let where: FindWhere<'components'> = { - mode: 'client', - type: this.type, - uid: this.config.id ?? this.config.name - }; - const component = await this.db.query.components.findFirst({ - where - }); - if(component !== undefined) { - this.dbComponent = component; - return; - } - - this.dbComponent = (await this.db.insert(components).values(generateComponentEntity({ - uid: this.config.id ?? this.config.name, - mode: 'client', - type: this.type, - name: this.config.name - })).returning())[0]; - } - protected async postInitialize(): Promise { const { options: { diff --git a/src/backend/sources/AbstractSource.ts b/src/backend/sources/AbstractSource.ts index 3dc13a22..474fad13 100644 --- a/src/backend/sources/AbstractSource.ts +++ b/src/backend/sources/AbstractSource.ts @@ -46,9 +46,6 @@ import prom, { Counter, Gauge } from 'prom-client'; import { normalizeStr } from '../utils/StringUtils.js'; import { spawn, catchAbortError, isAbortError, rethrowAbortError, delay, forever, AbortError, throwIfAborted } from 'abort-controller-x'; import { AbortedError, generateLoggableAbortReason } from '../common/errors/MSErrors.js'; -import { FindWhere } from '../common/database/drizzle/drizzleTypes.js'; -import { components } from '../common/database/drizzle/schema/schema.js'; -import { generateComponentEntity } from '../common/database/drizzle/entityUtils.js'; export interface RecentlyPlayedOptions { limit?: number @@ -60,7 +57,7 @@ export interface RecentlyPlayedOptions { export default abstract class AbstractSource extends AbstractComponent implements Authenticatable { name: string; - type: SourceType; + declare type: SourceType; declare config: SourceConfig; clients: string[]; @@ -106,8 +103,11 @@ export default abstract class AbstractSource extends AbstractComponent implement postCompare: staggerMapper({concurrency: 2}) } + declare protected componentType: 'source'; + constructor(type: SourceType, name: string, config: SourceConfig, internal: InternalConfig, emitter: EventEmitter) { super(config); + this.componentType = 'source'; const {clients = [] } = config; this.type = type; this.name = name; @@ -130,29 +130,6 @@ export default abstract class AbstractSource extends AbstractComponent implement } } - protected async doBuildDatabase(): Promise { - super.doBuildDatabase(); - let where: FindWhere<'components'> = { - mode: 'source', - type: this.type, - uid: this.config.id ?? this.config.name - }; - const component = await this.db.query.components.findFirst({ - where - }); - if(component !== undefined) { - this.dbComponent = component; - return; - } - - this.dbComponent = (await this.db.insert(components).values(generateComponentEntity({ - uid: this.config.id ?? this.config.name, - mode: 'source', - type: this.type, - name: this.config.name - })).returning())[0]; - } - protected async postCache(): Promise { await super.postCache(); this.generateStaggerMappers();