From d089b960f9c0485fcb7b255fd8a6b167e7f5fcc5 Mon Sep 17 00:00:00 2001 From: FoxxMD Date: Tue, 18 Aug 2026 23:17:34 +0000 Subject: [PATCH] feat: Refactor activity timeline elements into proper event structure * Implement new database playEvents table/entity * Refactor usage of lifecycle,scrobbler,etc... with write-only play events * Refactor/simplify activity timeline to use new event structure --- .../common/database/drizzle/drizzleTypes.ts | 5 +- .../common/database/drizzle/entityUtils.ts | 18 +- .../migration.sql | 11 + .../snapshot.json | 1206 +++++++++++++++++ .../repositories/PlayEventsRepository.ts | 8 + .../drizzle/repositories/PlayRepository.ts | 42 +- .../common/database/drizzle/schema/schema.ts | 28 +- .../scrobblers/AbstractScrobbleClient.ts | 252 ++-- src/backend/sources/AbstractSource.ts | 57 +- src/backend/sources/DeezerInternalSource.ts | 47 +- src/backend/sources/MemorySource.ts | 6 +- src/client/components/ActivityTimeline.tsx | 322 ++--- src/client/components/TransformSteps.tsx | 1 + src/core/Api.ts | 2 + src/core/PlayEvent.ts | 50 + 15 files changed, 1683 insertions(+), 372 deletions(-) create mode 100644 src/backend/common/database/drizzle/migrations/20260818175856_daffy_kate_bishop/migration.sql create mode 100644 src/backend/common/database/drizzle/migrations/20260818175856_daffy_kate_bishop/snapshot.json create mode 100644 src/backend/common/database/drizzle/repositories/PlayEventsRepository.ts create mode 100644 src/core/PlayEvent.ts diff --git a/src/backend/common/database/drizzle/drizzleTypes.ts b/src/backend/common/database/drizzle/drizzleTypes.ts index 3e8bed84..44927cc7 100644 --- a/src/backend/common/database/drizzle/drizzleTypes.ts +++ b/src/backend/common/database/drizzle/drizzleTypes.ts @@ -1,5 +1,5 @@ import type {DBQueryConfig, DBQueryConfigWith, KnownKeysOnly, RelationFieldsFilterInternals, BuildQueryResult, RelationsFilter} from "drizzle-orm"; -import type { components, componentMigrations, playInputs, plays, queueStates, relations, playsHistorical } from "./schema/schema.ts"; +import type { components, componentMigrations, playInputs, plays, queueStates, relations, playsHistorical, playEvents } from "./schema/schema.ts"; import type {TSchema, TableName} from "./schema/schema.ts"; @@ -24,6 +24,9 @@ export type PlayNew = typeof plays.$inferInsert; export type PlayHistoricalSelect = typeof playsHistorical.$inferSelect; export type PlayHistoricalNew = typeof playsHistorical.$inferInsert; +export type PlayEventSelect = typeof playEvents.$inferSelect; +export type PlayEventNew = typeof playEvents.$inferInsert; + // useful references for building types // https://github.com/drizzle-team/drizzle-orm/discussions/2596 // https://github.com/drizzle-team/drizzle-orm/discussions/1539 diff --git a/src/backend/common/database/drizzle/entityUtils.ts b/src/backend/common/database/drizzle/entityUtils.ts index 524c295f..d7cec546 100644 --- a/src/backend/common/database/drizzle/entityUtils.ts +++ b/src/backend/common/database/drizzle/entityUtils.ts @@ -1,5 +1,5 @@ import assert from "node:assert"; -import type {PlayHistoricalNew, PlayHistoricalSelect, PlayNew, PlaySelect, PlaySelectWithQueueStates} from "./drizzleTypes.ts"; +import type {PlayHistoricalNew, PlayHistoricalSelect, PlayNew, PlaySelect, PlaySelectWithQueueStates, QueueStateSelect} from "./drizzleTypes.ts"; import type {PlayInputNew} from "./drizzleTypes.ts"; import type {QueueStateNew} from "./drizzleTypes.ts"; import type {ComponentNew} from "./drizzleTypes.ts"; @@ -9,6 +9,7 @@ import dayjs from "dayjs"; import { playContentBasicInvariantTransform, playMbidIdentifier } from "../../../utils/PlayComparisonUtils.ts"; import { hashObject } from "../../../utils/StringUtils.ts"; import { serializeError } from "serialize-error"; +import type { PlayEventQueueStateChange, PlayEventQueueStateChangeData } from "../../../../core/PlayEvent.ts"; export const generateComponentEntity = (data: MarkOptional): ComponentNew => { assert(data.name !== undefined, 'Must provide name'); @@ -93,4 +94,19 @@ export const generateInputEntity = (data: PlayInputNew): PlayInputNew => { export const generateQueueStateEntity = (data: QueueStateNew): QueueStateNew => { return data; +} + +export const queueStateToEventData = (qs: QueueStateSelect): PlayEventQueueStateChangeData => { + const { + queueName, + queueStatus, + error, + retries + } = qs; + return { + queueName, + queueStatus, + error, + retries + } } \ No newline at end of file diff --git a/src/backend/common/database/drizzle/migrations/20260818175856_daffy_kate_bishop/migration.sql b/src/backend/common/database/drizzle/migrations/20260818175856_daffy_kate_bishop/migration.sql new file mode 100644 index 00000000..03b34a5c --- /dev/null +++ b/src/backend/common/database/drizzle/migrations/20260818175856_daffy_kate_bishop/migration.sql @@ -0,0 +1,11 @@ +CREATE TABLE `play_events` ( + `id` integer PRIMARY KEY, + `playId` integer NOT NULL, + `eventName` text(50) NOT NULL, + `data` text, + `error` text, + `createdAt` number NOT NULL, + CONSTRAINT `fk_play_events_playId_plays_id_fk` FOREIGN KEY (`playId`) REFERENCES `plays`(`id`) ON UPDATE CASCADE ON DELETE CASCADE +); +--> statement-breakpoint +CREATE INDEX `play_event_id_idx` ON `play_events` (`playId`); \ No newline at end of file diff --git a/src/backend/common/database/drizzle/migrations/20260818175856_daffy_kate_bishop/snapshot.json b/src/backend/common/database/drizzle/migrations/20260818175856_daffy_kate_bishop/snapshot.json new file mode 100644 index 00000000..32f7627b --- /dev/null +++ b/src/backend/common/database/drizzle/migrations/20260818175856_daffy_kate_bishop/snapshot.json @@ -0,0 +1,1206 @@ +{ + "version": "7", + "dialect": "sqlite", + "id": "6bdac6ab-f4d9-4f6e-aac2-341c0709497b", + "prevIds": [ + "cd13b640-3e52-4386-bf41-72071daa3d4d" + ], + "ddl": [ + { + "name": "component_migrations", + "entityType": "tables" + }, + { + "name": "components", + "entityType": "tables" + }, + { + "name": "jobs", + "entityType": "tables" + }, + { + "name": "play_events", + "entityType": "tables" + }, + { + "name": "play_inputs", + "entityType": "tables" + }, + { + "name": "plays", + "entityType": "tables" + }, + { + "name": "plays_historical", + "entityType": "tables" + }, + { + "name": "play_queue_states", + "entityType": "tables" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "component_migrations" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "componentId", + "entityType": "columns", + "table": "component_migrations" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "name", + "entityType": "columns", + "table": "component_migrations" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "success", + "entityType": "columns", + "table": "component_migrations" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "error", + "entityType": "columns", + "table": "component_migrations" + }, + { + "type": "number", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "attemptedAt", + "entityType": "columns", + "table": "component_migrations" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "components" + }, + { + "type": "text(200)", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "uid", + "entityType": "columns", + "table": "components" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "mode", + "entityType": "columns", + "table": "components" + }, + { + "type": "text(50)", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "type", + "entityType": "columns", + "table": "components" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "name", + "entityType": "columns", + "table": "components" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "0", + "generated": null, + "name": "countLive", + "entityType": "columns", + "table": "components" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "0", + "generated": null, + "name": "countNonLive", + "entityType": "columns", + "table": "components" + }, + { + "type": "number", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "createdAt", + "entityType": "columns", + "table": "components" + }, + { + "type": "number", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "lastReadyAt", + "entityType": "columns", + "table": "components" + }, + { + "type": "number", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "lastActiveAt", + "entityType": "columns", + "table": "components" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "componentFromId", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "componentToId", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "text(50)", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "name", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": "'idle'", + "generated": null, + "name": "status", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "0", + "generated": null, + "name": "retries", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "error", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "transformOptions", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "initialParameters", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "cursor", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "total", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "0", + "generated": null, + "name": "imported", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "0", + "generated": null, + "name": "scrobbled", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "number", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "createdAt", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "number", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "updatedAt", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "number", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "completedAt", + "entityType": "columns", + "table": "jobs" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "play_events" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "playId", + "entityType": "columns", + "table": "play_events" + }, + { + "type": "text(50)", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "eventName", + "entityType": "columns", + "table": "play_events" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "data", + "entityType": "columns", + "table": "play_events" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "error", + "entityType": "columns", + "table": "play_events" + }, + { + "type": "number", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "createdAt", + "entityType": "columns", + "table": "play_events" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "play_inputs" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "playId", + "entityType": "columns", + "table": "play_inputs" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "data", + "entityType": "columns", + "table": "play_inputs" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "play", + "entityType": "columns", + "table": "play_inputs" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "playHash", + "entityType": "columns", + "table": "play_inputs" + }, + { + "type": "number", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "createdAt", + "entityType": "columns", + "table": "play_inputs" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "plays" + }, + { + "type": "text(30)", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "uid", + "entityType": "columns", + "table": "plays" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "componentId", + "entityType": "columns", + "table": "plays" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "error", + "entityType": "columns", + "table": "plays" + }, + { + "type": "number", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "playedAt", + "entityType": "columns", + "table": "plays" + }, + { + "type": "number", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "seenAt", + "entityType": "columns", + "table": "plays" + }, + { + "type": "number", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "updatedAt", + "entityType": "columns", + "table": "plays" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "play", + "entityType": "columns", + "table": "plays" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "state", + "entityType": "columns", + "table": "plays" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "parentId", + "entityType": "columns", + "table": "plays" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "jobId", + "entityType": "columns", + "table": "plays" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "playHash", + "entityType": "columns", + "table": "plays" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "mbidIdentifier", + "entityType": "columns", + "table": "plays" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "compacted", + "entityType": "columns", + "table": "plays" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "plays_historical" + }, + { + "type": "text(200)", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "uid", + "entityType": "columns", + "table": "plays_historical" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "componentId", + "entityType": "columns", + "table": "plays_historical" + }, + { + "type": "number", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "playedAt", + "entityType": "columns", + "table": "plays_historical" + }, + { + "type": "number", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "seenAt", + "entityType": "columns", + "table": "plays_historical" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "play", + "entityType": "columns", + "table": "plays_historical" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "playHash", + "entityType": "columns", + "table": "plays_historical" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "mbidIdentifier", + "entityType": "columns", + "table": "plays_historical" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "play_queue_states" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "playId", + "entityType": "columns", + "table": "play_queue_states" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "componentId", + "entityType": "columns", + "table": "play_queue_states" + }, + { + "type": "text(50)", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "queueName", + "entityType": "columns", + "table": "play_queue_states" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": "'queued'", + "generated": null, + "name": "queueStatus", + "entityType": "columns", + "table": "play_queue_states" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "0", + "generated": null, + "name": "retries", + "entityType": "columns", + "table": "play_queue_states" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "error", + "entityType": "columns", + "table": "play_queue_states" + }, + { + "type": "number", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "createdAt", + "entityType": "columns", + "table": "play_queue_states" + }, + { + "type": "number", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "updatedAt", + "entityType": "columns", + "table": "play_queue_states" + }, + { + "columns": [ + "componentId" + ], + "tableTo": "components", + "columnsTo": [ + "id" + ], + "onUpdate": "CASCADE", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_component_migrations_componentId_components_id_fk", + "entityType": "fks", + "table": "component_migrations" + }, + { + "columns": [ + "componentFromId" + ], + "tableTo": "components", + "columnsTo": [ + "id" + ], + "onUpdate": "CASCADE", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_jobs_componentFromId_components_id_fk", + "entityType": "fks", + "table": "jobs" + }, + { + "columns": [ + "componentToId" + ], + "tableTo": "components", + "columnsTo": [ + "id" + ], + "onUpdate": "CASCADE", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_jobs_componentToId_components_id_fk", + "entityType": "fks", + "table": "jobs" + }, + { + "columns": [ + "playId" + ], + "tableTo": "plays", + "columnsTo": [ + "id" + ], + "onUpdate": "CASCADE", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_play_events_playId_plays_id_fk", + "entityType": "fks", + "table": "play_events" + }, + { + "columns": [ + "playId" + ], + "tableTo": "plays", + "columnsTo": [ + "id" + ], + "onUpdate": "CASCADE", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_play_inputs_playId_plays_id_fk", + "entityType": "fks", + "table": "play_inputs" + }, + { + "columns": [ + "componentId" + ], + "tableTo": "components", + "columnsTo": [ + "id" + ], + "onUpdate": "CASCADE", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_plays_componentId_components_id_fk", + "entityType": "fks", + "table": "plays" + }, + { + "columns": [ + "parentId" + ], + "tableTo": "plays", + "columnsTo": [ + "id" + ], + "onUpdate": "CASCADE", + "onDelete": "SET NULL", + "nameExplicit": false, + "name": "fk_plays_parentId_plays_id_fk", + "entityType": "fks", + "table": "plays" + }, + { + "columns": [ + "jobId" + ], + "tableTo": "jobs", + "columnsTo": [ + "id" + ], + "onUpdate": "CASCADE", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_plays_jobId_jobs_id_fk", + "entityType": "fks", + "table": "plays" + }, + { + "columns": [ + "componentId" + ], + "tableTo": "components", + "columnsTo": [ + "id" + ], + "onUpdate": "CASCADE", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_plays_historical_componentId_components_id_fk", + "entityType": "fks", + "table": "plays_historical" + }, + { + "columns": [ + "playId" + ], + "tableTo": "plays", + "columnsTo": [ + "id" + ], + "onUpdate": "CASCADE", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_play_queue_states_playId_plays_id_fk", + "entityType": "fks", + "table": "play_queue_states" + }, + { + "columns": [ + "componentId" + ], + "tableTo": "components", + "columnsTo": [ + "id" + ], + "onUpdate": "CASCADE", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_play_queue_states_componentId_components_id_fk", + "entityType": "fks", + "table": "play_queue_states" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "component_migrations_pk", + "table": "component_migrations", + "entityType": "pks" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "components_pk", + "table": "components", + "entityType": "pks" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "jobs_pk", + "table": "jobs", + "entityType": "pks" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "play_events_pk", + "table": "play_events", + "entityType": "pks" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "play_inputs_pk", + "table": "play_inputs", + "entityType": "pks" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "plays_pk", + "table": "plays", + "entityType": "pks" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "plays_historical_pk", + "table": "plays_historical", + "entityType": "pks" + }, + { + "columns": [ + "id" + ], + "nameExplicit": false, + "name": "play_queue_states_pk", + "table": "play_queue_states", + "entityType": "pks" + }, + { + "columns": [ + { + "value": "uid", + "isExpression": false + }, + { + "value": "mode", + "isExpression": false + }, + { + "value": "type", + "isExpression": false + } + ], + "isUnique": true, + "where": null, + "origin": "manual", + "name": "uid_mode_type_idx", + "entityType": "indexes", + "table": "components" + }, + { + "columns": [ + { + "value": "playId", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "play_event_id_idx", + "entityType": "indexes", + "table": "play_events" + }, + { + "columns": [ + { + "value": "playId", + "isExpression": false + } + ], + "isUnique": true, + "where": null, + "origin": "manual", + "name": "play_input_id_idx", + "entityType": "indexes", + "table": "play_inputs" + }, + { + "columns": [ + { + "value": "parentId", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "play_parent_id_idx", + "entityType": "indexes", + "table": "plays" + }, + { + "columns": [ + { + "value": "componentId", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "play_component_id_idx", + "entityType": "indexes", + "table": "plays" + }, + { + "columns": [ + { + "value": "uid", + "isExpression": false + } + ], + "isUnique": true, + "where": null, + "origin": "manual", + "name": "play_uid_idx", + "entityType": "indexes", + "table": "plays" + }, + { + "columns": [ + { + "value": "playedAt", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "play_playedAt_idx", + "entityType": "indexes", + "table": "plays" + }, + { + "columns": [ + { + "value": "seenAt", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "play_seenAt_idx", + "entityType": "indexes", + "table": "plays" + }, + { + "columns": [ + { + "value": "componentId", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "play_historical_component_id_idx", + "entityType": "indexes", + "table": "plays_historical" + }, + { + "columns": [ + { + "value": "uid", + "isExpression": false + } + ], + "isUnique": true, + "where": null, + "origin": "manual", + "name": "play_historical_uid_idx", + "entityType": "indexes", + "table": "plays_historical" + }, + { + "columns": [ + { + "value": "playedAt", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "play_historical_playedAt_idx", + "entityType": "indexes", + "table": "plays_historical" + }, + { + "columns": [ + { + "value": "playId", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "play_queue_state_id_idx", + "entityType": "indexes", + "table": "play_queue_states" + } + ], + "renames": [] +} \ No newline at end of file diff --git a/src/backend/common/database/drizzle/repositories/PlayEventsRepository.ts b/src/backend/common/database/drizzle/repositories/PlayEventsRepository.ts new file mode 100644 index 00000000..e40f7501 --- /dev/null +++ b/src/backend/common/database/drizzle/repositories/PlayEventsRepository.ts @@ -0,0 +1,8 @@ +import { DrizzleBaseRepository, type DrizzleRepositoryOpts } from "./BaseRepository.ts"; +import type {DbConcrete} from "../drizzleUtils.ts"; +export class DrizzlePlayEventsRepository extends DrizzleBaseRepository<'playEvents'> { + + constructor(db: DbConcrete, opts: DrizzleRepositoryOpts = {}) { + super(db, 'playEvents', 'Play Events', opts); + } +} \ No newline at end of file diff --git a/src/backend/common/database/drizzle/repositories/PlayRepository.ts b/src/backend/common/database/drizzle/repositories/PlayRepository.ts index 65181ef8..f312cb11 100644 --- a/src/backend/common/database/drizzle/repositories/PlayRepository.ts +++ b/src/backend/common/database/drizzle/repositories/PlayRepository.ts @@ -14,11 +14,11 @@ import type {SourceType} from "../../../../../core/Atomic.ts"; import type {FindMany, FindWhere, FindWith, PlayInputNew, PlayNew, PlaySelect, PlaySelectWithQueueStates, PlayWith, QueueStateSelect, WhereClause} from "../drizzleTypes.ts"; import { type DbConcrete, runTransaction } from "../drizzleUtils.ts"; import { generateInputEntity, generatePlayEntity, hydratePlaySelect, type PlayEntityOpts, type PlayHydateOptions } from "../entityUtils.ts"; -import { playInputs, plays, relations } from "../schema/schema.ts"; +import { playEvents, playInputs, plays, relations } from "../schema/schema.ts"; import { buildDateCompare, type CompareDateOp, type ComponentConstrainedRepoOpts, DrizzleBaseRepository, type DrizzleRepositoryOpts } from "./BaseRepository.ts"; import type {PaginatedResponse} from "../../../../../core/Api.ts"; import type {PaginatedQueryResponse} from "../../../../../core/Api.ts"; -; +import type { PlayEventTransform } from "../../../../../core/PlayEvent.ts"; // https://github.com/drizzle-team/drizzle-orm/issues/695 may be useful for typing models with relations? @@ -38,7 +38,7 @@ export interface PlayWhereOpts { text?: string[] } -export type WithPlayRelation = 'input' | 'parent' | 'parent-input' | 'queues'; +export type WithPlayRelation = 'input' | 'parent' | 'parent-input' | 'queues' | 'events'; export interface QueryPlaysOpts extends PlayWhereOpts { sort?: 'seenAt' | 'playedAt' order?: 'asc' | 'desc' @@ -103,11 +103,14 @@ export class DrizzlePlayRepository extends DrizzleBaseRepository<'plays'> { const entitiesData = entitiesOpts.map((data) => { const { - play, + play: { + lifecycle, + ...playRest + }, input, ...rest } = data; - return generatePlayEntity(play, { componentId: this.componentId, ...rest }); + return generatePlayEntity(playRest, { componentId: this.componentId, ...rest }); }); const nakedPlays = await this.db.insert(plays).values(entitiesData).returning(); @@ -126,6 +129,29 @@ export class DrizzlePlayRepository extends DrizzleBaseRepository<'plays'> { }); const inputRow = await this.db.insert(playInputs).values(inputDatas); + + const eventData = nakedPlays.map((x, index) => { + const { + play: { + lifecycle = [], + } = {} + } = entitiesOpts[index]; + + if(lifecycle.length > 0) { + const transformEvent: PlayEventTransform = { + playId: x.id, + eventName: 'transform', + createdAt: dayjs(lifecycle[0].createdAt), + data: lifecycle + } + return transformEvent; + } + return undefined; + }).filter(x => x !== undefined); + if(eventData.length > 0) { + await this.db.insert(playEvents).values(eventData); + } + playRows = nakedPlays.map((x, index) => ({...x, play: hydratePlaySelect(x, hydrate), input: inputRow[index]})); }); @@ -182,6 +208,9 @@ export class DrizzlePlayRepository extends DrizzleBaseRepository<'plays'> { case 'queues': query.with.queueStates = true; break; + case 'events': + query.with.events = true; + break; default: throw new Error(`Unknown relation ${w}`); } @@ -813,6 +842,9 @@ export const buildPlayWith = (args: WithPlayRelation[] | undefined): FindWith<'p case 'queues': qWith.queueStates = true; break; + case 'events': + qWith.events = true; + break; default: throw new Error(`Unknown relation ${w}`); } diff --git a/src/backend/common/database/drizzle/schema/schema.ts b/src/backend/common/database/drizzle/schema/schema.ts index 9aaca4b0..cc488579 100644 --- a/src/backend/common/database/drizzle/schema/schema.ts +++ b/src/backend/common/database/drizzle/schema/schema.ts @@ -212,7 +212,18 @@ export const jobs = sqliteTable("jobs", { completedAt: DayjsTimestamp('completedAt') }); -const playRelations = defineRelations({ plays, queueStates, playInputs, components, jobs, componentMigrations,playsHistorical }, (r) => ({ +export const playEvents = sqliteTable("play_events", { + id: integer({ mode: 'number' }).primaryKey(), + playId: integer().notNull().references(() => plays.id, {onDelete: 'cascade', onUpdate: 'cascade'}), + eventName: text({length: 50}).notNull(), + data: text({ mode: 'json' }), + error: ErrorLikeJson('error'), + createdAt: DayjsTimestamp('createdAt').notNull().$defaultFn(() => dayjs()), +}, (table) => [ + index('play_event_id_idx').on(table.playId) +]); + +const playRelations = defineRelations({ plays, queueStates, playEvents, playInputs, components, jobs, componentMigrations,playsHistorical }, (r) => ({ plays: { queueStates: r.many.queueStates(), input: r.one.playInputs({ @@ -234,7 +245,8 @@ const playRelations = defineRelations({ plays, queueStates, playInputs, componen from: r.plays.jobId, to: r.jobs.id, optional: true - }) + }), + events: r.many.playEvents(), }, playsHistorical: { component: r.one.components({ @@ -266,7 +278,13 @@ const playRelations = defineRelations({ plays, queueStates, playInputs, componen }, jobs: { plays: r.many.plays() - } + }, + playEvents: { + play: r.one.plays({ + from: r.playEvents.playId, + to: r.plays.id + }), + }, })); export const relations = playRelations; @@ -281,6 +299,8 @@ export const getConfigByTableName = (name: T) => { return components; case 'playInputs': return playInputs; + case 'playEvents': + return playEvents; case 'queueStates': return queueStates; case 'componentMigrations': @@ -290,7 +310,7 @@ export const getConfigByTableName = (name: T) => { } } -export const schema = {playInputs, plays, components, componentMigrations, queueStates, jobs}; +export const schema = {playInputs, plays, playEvents, components, componentMigrations, queueStates, jobs}; export type TSchema = typeof relations; export type Schema = typeof schema; diff --git a/src/backend/scrobblers/AbstractScrobbleClient.ts b/src/backend/scrobblers/AbstractScrobbleClient.ts index 0a330688..78c0bb62 100644 --- a/src/backend/scrobblers/AbstractScrobbleClient.ts +++ b/src/backend/scrobblers/AbstractScrobbleClient.ts @@ -32,7 +32,7 @@ import { type TimeRangeListensFetcher, } from "../common/infrastructure/Atomic.ts"; import { CALCULATED_PLAYER_STATUSES } from '../../core/Atomic.ts'; -import type {ReportedPlayerStatus} from '../../core/Atomic.ts'; +import type {ReportedPlayerStatus, ScrobbleResult} from '../../core/Atomic.ts'; import type {ClientType} from "../../core/Atomic.ts"; import type {CommonClientConfig, NowPlayingOptions, UpstreamRefreshOptions} from "../common/infrastructure/config/client/index.ts"; import { TRANSFORM_HOOK } from "../../core/Transform.ts"; @@ -73,6 +73,9 @@ import { GenericRepository } from "../common/database/drizzle/repositories/BaseR import assert from "node:assert"; import { COMPONENT_STATE, type ComponentClientApiJson, type PlayApiCommonDetailed } from "../../core/Api.ts"; import type {ComponentState} from "react"; +import { DrizzlePlayEventsRepository } from "../common/database/drizzle/repositories/PlayEventsRepository.ts"; +import { PLAY_EVENT_TYPE, type PlayEvent } from "../../core/PlayEvent.ts"; +import { queueStateToEventData } from "../common/database/drizzle/entityUtils.ts"; type SourceMappedPlayer = {player: SourcePlayerObj, source: SourceIdentifier}; type PlatformMappedPlays = Map; @@ -157,6 +160,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i protected playRepo!: DrizzlePlayRepository; protected queueRepo!: DrizzleQueueRepository; + protected playEventsRepo!: DrizzlePlayEventsRepository; protected migrationRepo!: GenericRepository<'componentMigrations'>; constructor(type: any, name: any, config: CommonClientConfig, emitter: EventEmitter, logger: Logger) { @@ -389,6 +393,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i protected async postDatabase(): Promise { this.playRepo = new DrizzlePlayRepository(this.db, {logger: this.logger}); this.queueRepo = new DrizzleQueueRepository(this.db, {logger: this.logger}); + this.playEventsRepo = new DrizzlePlayEventsRepository(this.db, {logger: this.logger}); this.migrationRepo = new GenericRepository<'componentMigrations'>(this.db, 'componentMigrations', 'Component Migrations', {logger: this.logger}); this.playRepo.componentId = this.dbComponent.id; this.queueRepo.componentId = this.dbComponent.id; @@ -1232,6 +1237,8 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i let successState: PlaySelect['state']; let deadQueueEntity: QueueStateSelect; + const events: Omit[] = []; + try { // want to fail scrobbles that are being *ingested* by component @@ -1262,35 +1269,43 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i } if (historicalError === undefined) { const { summary, ...matchResult } = await this.existingScrobble(currQueuedPlay.play, historicalPlays); - currQueuedPlay.play.scrobble = { - ...(currQueuedPlay.play.scrobble ?? {}), - match: matchResult, - createdAt: dayjs() - } + events.push({eventName: PLAY_EVENT_TYPE.dupeCheck, data: {summary, ...matchResult}, createdAt: dayjs()}); + // currQueuedPlay.play.scrobble = { + // ...(currQueuedPlay.play.scrobble ?? {}), + // match: matchResult, + // createdAt: dayjs() + // } signal.throwIfAborted(); if (!matchResult.match) { const transformedScrobble = await this.transformPlay(currQueuedPlay.play, TRANSFORM_HOOK.postCompare); + const { lifecycle = [] } = transformedScrobble; + const psLifecycle = lifecycle.filter(x => x.hook === TRANSFORM_HOOK.postCompare); + if(psLifecycle.length > 0) { + events.push({eventName: PLAY_EVENT_TYPE.transform, createdAt: dayjs(psLifecycle[0].createdAt), data: psLifecycle}); + } signal.throwIfAborted(); try { const scrobbledPlay = await this.scrobble(transformedScrobble, {signal}); - currQueuedPlay.play = scrobbledPlay; + const {scrobble} = scrobbledPlay; + events.push({eventName: PLAY_EVENT_TYPE.scrobbleResult, createdAt: scrobble.createdAt, data: scrobble}); + //currQueuedPlay.play = scrobbledPlay; await this.addScrobbledTrack(scrobbledPlay); //handledShiftedPlay = true; } catch (e) { - currQueuedPlay.play.scrobble = { + const scrobbleRes: ScrobbleResult = { createdAt: dayjs() - }; + } const submitError = findCauseByReference(e, ScrobbleSubmitError); if (submitError !== undefined) { - currQueuedPlay.play.scrobble.payload = submitError.payload; - currQueuedPlay.play.scrobble.response = submitError.responseBody; - currQueuedPlay.play.scrobble.error = serializeError(submitError); + scrobbleRes.payload = submitError.payload; + scrobbleRes.response = submitError.responseBody; + scrobbleRes.error = serializeError(submitError); } else { - currQueuedPlay.play.scrobble.payload = this.playToClientPayload(transformedScrobble); - currQueuedPlay.play.scrobble.error = serializeError(e); + scrobbleRes.payload = this.playToClientPayload(transformedScrobble); + scrobbleRes.error = serializeError(e); } - + events.push({eventName: PLAY_EVENT_TYPE.scrobbleResult, createdAt: scrobbleRes.createdAt, data: scrobbleRes}); queueError = e; deadQueueEntity = await this.addDeadLetterScrobble(currQueuedPlay, e); //handledShiftedPlay = true; @@ -1335,14 +1350,19 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i queueState.queueStatus = 'failed'; queueState.error = queueError; await this.playRepo.updateById(currQueuedPlay.id, {state: 'failed', error: queueError, play: currQueuedPlay.play}); + events.push({eventName: PLAY_EVENT_TYPE.playStateChange, data: {state: 'failed'}, createdAt: dayjs()}); + events.push({eventName: PLAY_EVENT_TYPE.queueStateChange, data: queueStateToEventData(queueState), createdAt: dayjs()}); currQueuedPlay.state = 'failed'; //currQueuedPlay.error = queueError; } else { await this.queueRepo.updateById(queueState.id, {queueStatus: 'completed'}); await this.playRepo.updateById(currQueuedPlay.id, {state: successState ?? 'scrobbled', play: currQueuedPlay.play}); + events.push({eventName: PLAY_EVENT_TYPE.playStateChange, data: {state: successState ?? 'scrobbled'}, createdAt: dayjs()}); currQueuedPlay.state = successState ?? 'scrobbled'; queueState.queueStatus = 'completed'; + events.push({eventName: PLAY_EVENT_TYPE.queueStateChange, data: queueStateToEventData(queueState), createdAt: dayjs()}); } + this.playEventsRepo.createMany(events.map(x => ({...x, playId: currQueuedPlay.id}))); this.emitPlayUpdate({...currQueuedPlay, queueStates: [queueState]} as unknown as PlayApiCommonDetailed); this.emitEvent('scrobbleDequeued', { queuedScrobble: currQueuedPlay }) this.queuedGauge.labels(this.getPrometheusLabels()).dec(); @@ -1443,97 +1463,124 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i if(deadScrobble === undefined) { throw new Error(`Play ${uid} does not exist for ${this.name}`); } - if(deadScrobble.state === 'scrobbled') { - throw new Error(`Play ${uid} is already scrobbled.`); - } - const deadQueueState: QueueStateSelect = deadScrobble.queueStates.find(x => x.queueName === DEAD_QUEUE); - if(deadQueueState === undefined) { - throw new Error(`Play ${uid} is not currently queued in dead letter.`); - } - this.setStatus(`Processing Dead Play ${uid}`); - //const deadScrobble = await this.playRepo.getQueueNext(this.dbComponent.id, CLIENT_INGRESS_QUEUE); - const deadLabel = {labels: deadScrobble.uid}; - //const deadScrobble = this.deadLetterScrobbles[deadScrobbleIndex]; - this.deadLogger.trace(deadLabel, `Processing dead scrobble => ${buildTrackString(deadScrobble.play)}`); - - await this.handleQueuedScrobbleRanges(); - signal?.throwIfAborted(); + const events: Omit[] = []; + let deadQueueState: QueueStateSelect; + try { + if (deadScrobble.state === 'scrobbled') { + throw new Error(`Play ${uid} is already scrobbled.`); + } + deadQueueState = deadScrobble.queueStates.find(x => x.queueName === DEAD_QUEUE); + if (deadQueueState === undefined) { + throw new Error(`Play ${uid} is not currently queued in dead letter.`); + } + this.setStatus(`Processing Dead Play ${uid}`); + //const deadScrobble = await this.playRepo.getQueueNext(this.dbComponent.id, CLIENT_INGRESS_QUEUE); + const deadLabel = { labels: deadScrobble.uid }; + //const deadScrobble = this.deadLetterScrobbles[deadScrobbleIndex]; + this.deadLogger.trace(deadLabel, `Processing dead scrobble => ${buildTrackString(deadScrobble.play)}`); - if (!(await this.isReady())) { - this.deadLogger.warn(deadLabel, 'Cannot process dead letter scrobble because client is not ready.'); - return [false, deadScrobble]; - } - let historicalPlays: PlayObject[] = []; - if(this.upstreamRefresh.refreshEnabled) { - try { - historicalPlays = await this.getSOTScrobblesForPlay(deadScrobble.play); - } catch (e) { - if(e.message === 'Cannot get historical plays due to cached error') { - this.deadLogger.warn(deadLabel, `Previous error while getting historical scrobbles means this scrobble cannot be compared`); - this.deadLogger.trace(e); - } else { - this.deadLogger.warn(new SimpleError(`${deadScrobble.uid} - ${buildTrackString(deadScrobble.play)} from Source '${deadScrobble.play.meta.source}' => cannot get historical scrobbles`, {cause: e, shortStack: true})); - } + await this.handleQueuedScrobbleRanges(); + signal?.throwIfAborted(); - this.queueRepo.updateById(deadQueueState.id, {retries: deadQueueState.retries + 1, error: e, updatedAt: dayjs(), queueStatus: 'failed'}); - //this.playRepo.updateById(deadScrobble.id, {error: e}); - // deadScrobble.retries++; - // deadScrobble.error = messageWithCauses(e); - // deadScrobble.lastRetry = dayjs(); - // this.deadLetterScrobbles[deadScrobbleIndex] = deadScrobble; - this.emitEvent('updateDeadLetter', {dead: deadScrobble}); + if (!(await this.isReady())) { + this.deadLogger.warn(deadLabel, 'Cannot process dead letter scrobble because client is not ready.'); return [false, deadScrobble]; } - } - signal?.throwIfAborted(); - const {summary, ...matchResult} = await this.existingScrobble(deadScrobble.play, historicalPlays); - deadScrobble.play.scrobble = { - ...(deadScrobble.play.scrobble ?? {}), - match: matchResult, - createdAt: dayjs() - } - if(!matchResult.match) { - const transformedScrobble = await this.transformPlay(deadScrobble.play, TRANSFORM_HOOK.postCompare); + let historicalPlays: PlayObject[] = []; + if (this.upstreamRefresh.refreshEnabled) { + try { + historicalPlays = await this.getSOTScrobblesForPlay(deadScrobble.play); + } catch (e) { + if (e.message === 'Cannot get historical plays due to cached error') { + this.deadLogger.warn(deadLabel, `Previous error while getting historical scrobbles means this scrobble cannot be compared`); + this.deadLogger.trace(e); + } else { + this.deadLogger.warn(new SimpleError(`${deadScrobble.uid} - ${buildTrackString(deadScrobble.play)} from Source '${deadScrobble.play.meta.source}' => cannot get historical scrobbles`, { cause: e, shortStack: true })); + } + + events.push({eventName: PLAY_EVENT_TYPE.queueStateChange, data: queueStateToEventData({...deadQueueState, queueStatus: 'failed', error: e}), createdAt: dayjs()}); + this.queueRepo.updateById(deadQueueState.id, { retries: deadQueueState.retries + 1, error: e, updatedAt: dayjs(), queueStatus: 'failed' }); + //this.playRepo.updateById(deadScrobble.id, {error: e}); + // deadScrobble.retries++; + // deadScrobble.error = messageWithCauses(e); + // deadScrobble.lastRetry = dayjs(); + // this.deadLetterScrobbles[deadScrobbleIndex] = deadScrobble; + this.emitEvent('updateDeadLetter', { dead: deadScrobble }); + return [false, deadScrobble]; + } + } signal?.throwIfAborted(); - try { - const scrobbledPlay = await this.scrobble(transformedScrobble); - deadScrobble.play = scrobbledPlay; - await this.addScrobbledTrack(scrobbledPlay); - this.playRepo.updateById(deadScrobble.id, {play: deadScrobble.play}); - this.queueRepo.updateById(deadQueueState.id, {error: null, updatedAt: dayjs(), queueStatus: QUEUE_STATUS_COMPLETED}); - this.removeDeadLetterScrobble(deadScrobble, 'scrobbled', true); - } catch (e) { - deadScrobble.play.scrobble = { - ...(deadScrobble.play.scrobble ?? {}), - //createdAt: dayjs() + const { summary, ...matchResult } = await this.existingScrobble(deadScrobble.play, historicalPlays); + events.push({eventName: PLAY_EVENT_TYPE.dupeCheck, data: {summary, ...matchResult}, createdAt: dayjs()}); + // deadScrobble.play.scrobble = { + // ...(deadScrobble.play.scrobble ?? {}), + // match: matchResult, + // createdAt: dayjs() + // } + if (!matchResult.match) { + const transformedScrobble = await this.transformPlay(deadScrobble.play, TRANSFORM_HOOK.postCompare); + const { lifecycle = [] } = transformedScrobble; + const psLifecycle = lifecycle.filter(x => x.hook === TRANSFORM_HOOK.postCompare); + if(psLifecycle.length > 0) { + events.push({eventName: PLAY_EVENT_TYPE.transform, createdAt: dayjs(psLifecycle[0].createdAt), data: psLifecycle}); } - const submitError = findCauseByReference(e, ScrobbleSubmitError); - if(submitError !== undefined) { - deadScrobble.play.scrobble.payload = submitError.payload; - deadScrobble.play.scrobble.response = submitError.responseBody; - deadScrobble.play.scrobble.error = serializeError(submitError); - } else { - deadScrobble.play.scrobble.payload = this.playToClientPayload(transformedScrobble); - deadScrobble.play.scrobble.error = serializeError(e); + signal?.throwIfAborted(); + try { + const scrobbledPlay = await this.scrobble(transformedScrobble); + const {scrobble} = scrobbledPlay; + events.push({eventName: PLAY_EVENT_TYPE.scrobbleResult, createdAt: scrobble.createdAt, data: scrobble}); + deadScrobble.play = scrobbledPlay; + await this.addScrobbledTrack(scrobbledPlay); + this.playRepo.updateById(deadScrobble.id, { play: deadScrobble.play, state: 'scrobbled' }); + this.queueRepo.updateById(deadQueueState.id, { error: null, updatedAt: dayjs(), queueStatus: QUEUE_STATUS_COMPLETED }); + events.push({eventName: PLAY_EVENT_TYPE.queueStateChange, data: {queueName: DEAD_QUEUE, queueStatus: QUEUE_STATUS_COMPLETED}, createdAt: dayjs()}); + events.push({eventName: PLAY_EVENT_TYPE.playStateChange, data: {state: 'scrobbled'}, createdAt: dayjs()}); + this.removeDeadLetterScrobble(deadScrobble, 'scrobbled', true); + } catch (e) { + const scrobbleRes: ScrobbleResult = { + createdAt: dayjs() + } + // deadScrobble.play.scrobble = { + // ...(deadScrobble.play.scrobble ?? {}), + // //createdAt: dayjs() + // } + const submitError = findCauseByReference(e, ScrobbleSubmitError); + if (submitError !== undefined) { + scrobbleRes.payload = submitError.payload; + scrobbleRes.response = submitError.responseBody; + scrobbleRes.error = serializeError(submitError); + } else { + scrobbleRes.payload = this.playToClientPayload(transformedScrobble); + scrobbleRes.error = serializeError(e); + } + events.push({eventName: PLAY_EVENT_TYPE.scrobbleResult, createdAt: scrobbleRes.createdAt, data: scrobbleRes}); + this.queueRepo.updateById(deadQueueState.id, { retries: deadQueueState.retries + 1, error: e, updatedAt: dayjs(), queueStatus: 'failed' }); + events.push({eventName: PLAY_EVENT_TYPE.queueStateChange, data: queueStateToEventData({...deadQueueState, queueStatus: 'failed', error: e}), createdAt: dayjs()}); + //this.playRepo.updateById(deadScrobble.id, { play: deadScrobble.play }); + // deadScrobble.retries++; + // deadScrobble.error = messageWithCauses(e); + // deadScrobble.lastRetry = dayjs(); + this.deadLogger.error(new Error(`${deadScrobble.uid} - Could not scrobble ${buildTrackString(transformedScrobble)} from Source '${deadScrobble.play.meta.source}' due to error`, { cause: e })); + //this.deadLetterScrobbles[deadScrobbleIndex] = deadScrobble; + this.emitEvent('updateDeadLetter', { dead: deadScrobble }); + return [false, deadScrobble]; } + } else { + this.playRepo.updateById(deadScrobble.id, { play: deadScrobble.play }); + this.deadLogger.verbose(`Looks like ${buildTrackString(deadScrobble.play)} was already scrobbled!\n${summary}`); + this.removeDeadLetterScrobble(deadScrobble, 'duped', true); + events.push({eventName: PLAY_EVENT_TYPE.queueStateChange, data: {queueName: DEAD_QUEUE, queueStatus: QUEUE_STATUS_COMPLETED}, createdAt: dayjs()}); + events.push({eventName: PLAY_EVENT_TYPE.playStateChange, data: {state: 'duped', reason: 'Looks like it was already scrobbled downstream'}, createdAt: dayjs()}); + } - this.queueRepo.updateById(deadQueueState.id, {retries: deadQueueState.retries + 1, error: e, updatedAt: dayjs(), queueStatus: 'failed'}); - this.playRepo.updateById(deadScrobble.id, {play: deadScrobble.play}); - // deadScrobble.retries++; - // deadScrobble.error = messageWithCauses(e); - // deadScrobble.lastRetry = dayjs(); - this.deadLogger.error(new Error(`${deadScrobble.uid} - Could not scrobble ${buildTrackString(transformedScrobble)} from Source '${deadScrobble.play.meta.source}' due to error`, {cause: e})); - //this.deadLetterScrobbles[deadScrobbleIndex] = deadScrobble; - this.emitEvent('updateDeadLetter', {dead: deadScrobble}); - return [false, deadScrobble]; + return [true, deadScrobble]; + } catch (e) { + if(deadQueueState !== undefined) { + events.push({eventName: PLAY_EVENT_TYPE.queueStateChange, data: queueStateToEventData({...deadQueueState, queueStatus: 'failed', error: e}), createdAt: dayjs()}); } - } else { - this.playRepo.updateById(deadScrobble.id, {play: deadScrobble.play}); - this.deadLogger.verbose(`Looks like ${buildTrackString(deadScrobble.play)} was already scrobbled!\n${summary}`); - this.removeDeadLetterScrobble(deadScrobble, 'duped', true); + } finally { + this.playEventsRepo.createMany(events.map(x => ({...x, playId: deadScrobble.id}))); } - - return [true, deadScrobble]; } removeDeadLetterScrobble = async (dead: (PlaySelect & {queueStates: QueueStateSelect[]}) | string, state: PlaySelect['state'], success: boolean) => { @@ -1569,6 +1616,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i queueUpdate.error = null; } await this.queueRepo.updateById(deadQueueState.id, queueUpdate); + await this.playRepo.updateById(deadScrobble.id, removeUndefinedKeys({state, error: success ? null : undefined})); this.deadLogger.info({labels: deadScrobble.uid}, `Scrobble ${buildTrackString(deadScrobble.play)} marked as completed`); this.deadLetterLength -= 1; @@ -1602,6 +1650,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i const createdQueuedPlays: PlaySelect[] = []; for await(const play of pMapIterable(playDatas, this.staggerMappers.preCompare(async x => transformFunc !== undefined ? await transformFunc(x) : await this.transformPlay(x, TRANSFORM_HOOK.preCompare)), {concurrency: 3})) { + const events: Omit[] = []; try { // cheap check, looks for play data (non-meta) hash, playdate, and optionally mbid recording const cheapExisting = await this.playRepo.checkExisting(play, { queueName: INGRESS_QUEUE }); @@ -1650,7 +1699,11 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i }); const playRow = await this.playRepo.createPlays([createPlayData]); - const queueState = await this.queueRepo.create({componentId: this.dbComponent.id, playId: playRow[0].id, queueName: INGRESS_QUEUE}); + const queueState = await this.queueRepo.create({componentId: this.dbComponent.id, playId: playRow[0].id, queueName: INGRESS_QUEUE}) as QueueStateSelect; + await this.playEventsRepo.createMany([ + {playId: playRow[0].id, eventName: PLAY_EVENT_TYPE.playStateChange, data: {state: 'queued'}, createdAt: playRow[0].seenAt.add(1,'ms')}, + {playId: playRow[0].id, eventName: PLAY_EVENT_TYPE.queueStateChange, data: queueStateToEventData(queueState), createdAt: queueState.createdAt} + ]); createdQueuedPlays.push(playRow[0]); this.logger.debug(`Added ${buildTrackString(play)} to the queue`); this.setStatus(`Added Play from parent ${play.uid} to queue`); @@ -1684,6 +1737,9 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i playId: data.id, queueName: DEAD_QUEUE }) as QueueStateSelect; + await this.playEventsRepo.createMany([ + {playId: data.id, eventName: PLAY_EVENT_TYPE.queueStateChange, data: queueStateToEventData(newQueue), createdAt: newQueue.createdAt} + ]); const deadData = {id: nanoid(), retries: 0, error: e, play: data.play}; //this.deadLetterScrobbles.push(deadData); //this.deadLetterScrobbles.sort((a, b) => sortByOldestPlayDate(a.play, b.play)); @@ -1939,7 +1995,7 @@ export default abstract class AbstractScrobbleClient extends AbstractComponent i public async getPlayApiResponse(uid: string, opts: {with?: WithPlayRelation[]} = {}): Promise { const { - with: withQuery = ['input','parent-input','queues'], + with: withQuery = ['input','parent-input','queues','events'], } = opts; return await this.playRepo.findByUid(uid, { with: withQuery as WithPlayRelation[] }) as unknown as PlayApiCommonDetailed; } diff --git a/src/backend/sources/AbstractSource.ts b/src/backend/sources/AbstractSource.ts index c1ff4508..5e2b4a28 100644 --- a/src/backend/sources/AbstractSource.ts +++ b/src/backend/sources/AbstractSource.ts @@ -47,8 +47,11 @@ import { asPlay } from '../../core/PlayMarshalUtils.ts'; import { AsyncTask, SimpleIntervalJob, ToadScheduler } from 'toad-scheduler'; import { COMPONENT_STATE, type ComponentSourceApiJson, type ComponentState, type PlayApiCommonDetailed } from '../../core/Api.ts'; import type {PaginatedResponse} from "../../core/Api.ts"; -import type { PlaySelect, PlaySelectWithQueueStates, QueueStateNew } from '../common/database/drizzle/drizzleTypes.ts'; +import type { PlaySelect, PlaySelectWithQueueStates, QueueStateNew, QueueStateSelect } from '../common/database/drizzle/drizzleTypes.ts'; import { DrizzleQueueRepository } from '../common/database/drizzle/repositories/QueueRepository.ts'; +import { DrizzlePlayEventsRepository } from '../common/database/drizzle/repositories/PlayEventsRepository.ts'; +import { PLAY_EVENT_TYPE, type PlayEvent } from '../../core/PlayEvent.ts'; +import { queueStateToEventData } from '../common/database/drizzle/entityUtils.ts'; export interface RecentlyPlayedOptions { limit?: number @@ -109,6 +112,7 @@ export default abstract class AbstractSource extends AbstractComponent implement protected playRepo!: DrizzlePlayRepository; protected queueRepo!: DrizzleQueueRepository; + protected playEventsRepo!: DrizzlePlayEventsRepository; protected queuedGauge: Gauge; @@ -283,6 +287,7 @@ export default abstract class AbstractSource extends AbstractComponent implement protected async postDatabase(): Promise { this.playRepo = new DrizzlePlayRepository(this.db, {logger: this.logger}); this.queueRepo = new DrizzleQueueRepository(this.db, {logger: this.logger}); + this.playEventsRepo = new DrizzlePlayEventsRepository(this.db, {logger: this.logger}); this.playRepo.componentId = this.dbComponent.id; this.queueRepo.componentId = this.dbComponent.id; const counts = await this.playRepo.getComponentPlayCountByState(); @@ -420,7 +425,11 @@ export default abstract class AbstractSource extends AbstractComponent implement }); const playRow = await this.playRepo.createPlays([createPlayData]); - const queueState = await this.queueRepo.create({componentId: this.dbComponent.id, playId: playRow[0].id, queueName: INGRESS_QUEUE}); + const queueState = await this.queueRepo.create({componentId: this.dbComponent.id, playId: playRow[0].id, queueName: INGRESS_QUEUE}) as QueueStateSelect; + await this.playEventsRepo.createMany([ + {playId: playRow[0].id, eventName: PLAY_EVENT_TYPE.playStateChange, data: {state: 'queued'}, createdAt: playRow[0].seenAt.add(1,'ms')}, + {playId: playRow[0].id, eventName: PLAY_EVENT_TYPE.queueStateChange, data: queueStateToEventData(queueState), createdAt: queueState.createdAt} + ]); createdQueuedPlays.push(playRow[0]); this.logger.debug(`Added ${buildTrackString(queueablePlay)} to the queue`); this.emitPlayInsert({...playRow[0], queueStates: [queueState]} as unknown as PlayApiCommonDetailed); @@ -487,21 +496,30 @@ export default abstract class AbstractSource extends AbstractComponent implement return list; } - async existingDiscovered(play: PlayObject): Promise { + async existingDiscovered(play: PlayObject): Promise { const list: PlayObject[] = await this.getRecentPlays(true); - const matchResults = await this.existingDiscoveredPlay(play, list); - if(matchResults.match) { - return matchResults.closestMatchedPlay; - } - return undefined; + return await this.existingDiscoveredPlay(play, list); + // if(matchResults.match) { + // return matchResults.closestMatchedPlay; + // } + // return undefined; } protected scrobble = async (newDiscoveredPlays: PlayObject[], options: { forceRefresh?: boolean, [key: string]: any, discoverLocation?: 'backlog' | [key: string] } = {}) => { if(newDiscoveredPlays.length > 0) { newDiscoveredPlays.sort(sortByOldestPlayDate); + const postCompareMapped = await pMap(newDiscoveredPlays, async (x) => await this.transformPlay(x, TRANSFORM_HOOK.postCompare), {concurrency: 3}); + const events: PlayEvent[] = []; + for(const p of postCompareMapped) { + const {lifecycle = []} = p; + const psLifecycle = lifecycle.filter(x => x.hook === TRANSFORM_HOOK.postCompare); + if(psLifecycle.length > 0) { + events.push({playId: p.id, eventName: PLAY_EVENT_TYPE.transform, createdAt: dayjs(psLifecycle[0].createdAt), data: psLifecycle}); + } + } this.emitEvent('discoveredToScrobble', { - data: await pMap(newDiscoveredPlays, async (x) => await this.transformPlay(x, TRANSFORM_HOOK.postCompare), {concurrency: 3}), + data: postCompareMapped, options: { ...options, checkTime: newDiscoveredPlays[newDiscoveredPlays.length-1].data.playDate.add(2, 'second'), @@ -960,17 +978,24 @@ export default abstract class AbstractSource extends AbstractComponent implement const queueState = currQueuedPlay.queueStates.find(x => x.queueName === INGRESS_QUEUE); const updatedQueueState: Partial = {}; let state: PlayState; + const events: Omit[] = []; try { - const preCompared = await this.transformPlay(currQueuedPlay.play, TRANSFORM_HOOK.preCompare); + const {lifecycle = [], ...preCompared} = await this.transformPlay(currQueuedPlay.play, TRANSFORM_HOOK.preCompare); + if(lifecycle.length > 0) { + events.push({eventName: PLAY_EVENT_TYPE.transform, createdAt: dayjs(lifecycle[0].createdAt), data: lifecycle}); + } let existing: PlayObject; // cheap check for existing const cheapExisting = await this.playRepo.checkExisting(preCompared, { notId: currQueuedPlay.id }); if(cheapExisting !== undefined) { + events.push({eventName: PLAY_EVENT_TYPE.dupeCheck, data: {match: true, closestMatchedPlay: cheapExisting.play, score: 1, breakdowns: [], createdAt: dayjs().toISOString(), reason: `Matched hash on existing Play ${cheapExisting.uid} with close temporality`}, createdAt: dayjs()}); updatedQueueState.error = {message: `Matched hash on existing Play ${cheapExisting.uid} with close temporality`}; existing = {...cheapExisting.play, id: cheapExisting.id, uid: cheapExisting.uid}; } else { - existing = await this.existingDiscovered(preCompared); - if(existing !== undefined) { + const matchRes = await this.existingDiscovered(preCompared); + events.push({eventName: PLAY_EVENT_TYPE.dupeCheck, data: matchRes, createdAt: dayjs()}); + if(matchRes.match) { + existing = matchRes.closestMatchedPlay; updatedQueueState.error = {message: `Matched with Play ${existing.uid ?? existing.id}`}; } } @@ -981,8 +1006,10 @@ export default abstract class AbstractSource extends AbstractComponent implement this.logger.debug(`Not adding ${buildTrackString(preCompared)} as discovered because monitoring was disabled when Play was created.`); state = 'discarded'; updatedQueueState.error = {message: 'Play was not added as discovered because monitoring was disabled when Play was created.'} + events.push({eventName: PLAY_EVENT_TYPE.playStateChange, data: {state, reason: 'Not added as discovered because monitoring was disabled when Play was created'}, createdAt: dayjs()}); } else { state = 'discovered'; + events.push({eventName: PLAY_EVENT_TYPE.playStateChange, data: {state}, createdAt: dayjs()}); this.tracksDiscovered++; this.tracksDiscoveredTotal++ this.discoveredCounter.labels(this.getPrometheusLabels()).inc(); @@ -991,6 +1018,7 @@ export default abstract class AbstractSource extends AbstractComponent implement } else { this.playRepo.updateById(existing.id, {updatedAt: dayjs()}); state = 'duped'; + events.push({eventName: PLAY_EVENT_TYPE.playStateChange, data: {state}, createdAt: dayjs()}); currQueuedPlay.parentId = existing.id; } this.playRepo.updateById(currQueuedPlay.id, {play: preCompared, state}); @@ -1011,14 +1039,17 @@ export default abstract class AbstractSource extends AbstractComponent implement } } updatedQueueState.queueStatus = 'completed'; + events.push({eventName: PLAY_EVENT_TYPE.queueStateChange, data: queueStateToEventData({...queueState, ...updatedQueueState}), createdAt: dayjs()}); this.logger.info(`${capitalize(state)} => ${buildTrackString(preCompared)}`); } catch (e) { const err = new Error(`Error ocurred while trying to discover Play ${currQueuedPlay.uid}`, {cause: e}); updatedQueueState.error = err; updatedQueueState.queueStatus = 'failed'; + events.push({eventName: PLAY_EVENT_TYPE.queueStateChange, data: queueStateToEventData({...queueState, ...updatedQueueState}), createdAt: dayjs()}); } finally { this.queueRepo.updateById(queueState.id, updatedQueueState); + this.playEventsRepo.createMany(events.map(x => ({...x, playId: currQueuedPlay.id}))); } if(state === 'discovered') { @@ -1075,7 +1106,7 @@ export default abstract class AbstractSource extends AbstractComponent implement public async getPlayApiResponse(uid: string, opts: {with?: WithPlayRelation[]} = {}): Promise { const { - with: withQuery = ['input','parent-input','queues'], + with: withQuery = ['input','parent-input','queues','events'], } = opts; return await this.playRepo.findByUid(uid, { with: withQuery as WithPlayRelation[] }) as unknown as PlayApiCommonDetailed; } diff --git a/src/backend/sources/DeezerInternalSource.ts b/src/backend/sources/DeezerInternalSource.ts index 3f3ff099..4edd3ee9 100644 --- a/src/backend/sources/DeezerInternalSource.ts +++ b/src/backend/sources/DeezerInternalSource.ts @@ -2,7 +2,7 @@ import dayjs from "dayjs"; import type EventEmitter from "events"; import type { Request } from 'superagent'; import request from 'superagent'; -import { COMPONENT_AUTH_TYPE, type ComponentAuthType, type PlayObject, type PlayObjectMinimal, SOURCE_SOT, TA_CLOSE, TA_DURING, TA_EXACT, TA_FUZZY, type TemporalAccuracy } from "../../core/Atomic.ts"; +import { COMPONENT_AUTH_TYPE, type ComponentAuthType, type PlayMatchResult, type PlayObject, type PlayObjectMinimal, SOURCE_SOT, TA_CLOSE, TA_DURING, TA_EXACT, TA_FUZZY, type TemporalAccuracy } from "../../core/Atomic.ts"; import { DEFAULT_RETRY_MULTIPLIER, type FormatPlayObjectOptions, type InternalConfig } from "../common/infrastructure/Atomic.ts"; import type {DeezerInternalSourceConfig, DeezerInternalTrackData} from "../common/infrastructure/config/source/deezer.ts"; import { TRANSFORM_HOOK } from "../../core/Transform.ts"; @@ -333,7 +333,7 @@ export default class DeezerInternalSource extends MemorySource { protected getBackloggedPlays = async (options: RecentlyPlayedOptions = {}) => await this.getRecentlyPlayed({formatted: true, ...options}) - async existingDiscovered(play: PlayObject): Promise { + async existingDiscovered(play: PlayObject): Promise { const list: PlayObject[] = await this.getRecentlyDiscoveredPlays(); const candidate = await this.transformPlay(play, TRANSFORM_HOOK.candidate); const existing = await findAsync(list, async x => { @@ -341,7 +341,14 @@ export default class DeezerInternalSource extends MemorySource { return genericSourcePlayMatch(e, candidate); }); if(existing) { - return existing; + return { + match: true, + score: 1, + breakdowns: [], + reason: 'Has matching data with very close timestamps', + closestMatchedPlay: existing, + createdAt: dayjs().toISOString() + } } if(this.config.options?.fuzzyDiscoveryIgnore === true || this.config.options?.fuzzyDiscoveryIgnore === 'aggressive') { const fuzzyIndex = await findIndexAsync(list, async x => { @@ -362,19 +369,45 @@ export default class DeezerInternalSource extends MemorySource { if(this.config.options?.fuzzyDiscoveryIgnore === 'aggressive') { // always return fuzzy match as existing // likely will make MS miss scrobbles for repeated plays - return list[fuzzyIndex]; + return { + match: true, + score: 1, + breakdowns: [], + reason: 'Has matching data and timestamp is during the duration of a previous play', + closestMatchedPlay: list[fuzzyIndex], + createdAt: dayjs().toISOString() + } } if(fuzzyIndex + 1 === list.length || playObjDataMatch(list[fuzzyIndex], list[fuzzyIndex + 1])) { // last discovered play was this one, or next played play was also this one // so we'll assume this means the play is on repeat, don't count as existing - return undefined; + return { + match: false, + score: 0.5, + breakdowns: [], + reason: 'Has matching data for previous play but assuming its on repeat', + closestMatchedPlay: existing, + createdAt: dayjs().toISOString() + } } // next played play was *not* this one (Deezer reports play between candidate TS and fuzzy match) // so this is likely a duplicate deezer should not have reported - return list[fuzzyIndex]; + return { + match: true, + score: 1, + breakdowns: [], + reason: 'Has matching data and looks like a misreported play', + closestMatchedPlay: existing, + createdAt: dayjs().toISOString() + } } } - return undefined; + return { + match: false, + score: 0, + breakdowns: [], + createdAt: dayjs().toISOString() + } } } diff --git a/src/backend/sources/MemorySource.ts b/src/backend/sources/MemorySource.ts index bbcdf075..a453640b 100644 --- a/src/backend/sources/MemorySource.ts +++ b/src/backend/sources/MemorySource.ts @@ -366,11 +366,11 @@ export default class MemorySource extends AbstractSource { if (thresholdResults.passes) { const matchingRecent = await this.existingDiscovered(candidate); //sRecentlyPlayed.find(x => playObjDataMatch(x, candidate)); - if (matchingRecent === undefined) { + if (matchingRecent.match === false) { return [true,`${stPrefix} added after ${thresholdResultSummary(thresholdResults)} and not matching any prior plays`]; } else { const {data: {playDate, duration}} = candidate; - const {data: {playDate: rplayDate}} = matchingRecent; + const {closestMatchedPlay: {data: {playDate: rplayDate}} = {}} = matchingRecent; if (!playDate.isSame(rplayDate)) { if (duration !== undefined) { if (playDate.isAfter(rplayDate.add(duration, 's'))) { @@ -386,7 +386,7 @@ export default class MemorySource extends AbstractSource { return [false, `${stPrefix} not added because it matched the last discovered play and could not determine time frame of play`]; } } else { - return [false, `${stPrefix} ${EXPECTED_NON_DISCOVERED_REASON}`] + return [false, `${stPrefix} ${EXPECTED_NON_DISCOVERED_REASON}`]; } } } diff --git a/src/client/components/ActivityTimeline.tsx b/src/client/components/ActivityTimeline.tsx index 80e01d46..a3b9ee6f 100644 --- a/src/client/components/ActivityTimeline.tsx +++ b/src/client/components/ActivityTimeline.tsx @@ -1,14 +1,13 @@ import type { Collapsible } from '@chakra-ui/react'; import { Card, Icon, SkeletonCircle, SkeletonText, Span, Tabs, Timeline} from '@chakra-ui/react'; -import type { Dayjs } from "dayjs"; import dayjs from "dayjs"; import React from "react"; import { BiWrench } from "react-icons/bi"; import { HiMiniMagnifyingGlass } from "react-icons/hi2"; import { IoMdCodeDownload } from "react-icons/io"; import { TbDatabaseEdit } from "react-icons/tb"; -import type {PlayApiCommonDetailed, QueueStateApi} from "../../core/Api"; -import { DEAD_QUEUE, INGRESS_QUEUE, QUEUE_STATUS_COMPLETED, QUEUE_STATUS_FAILED, QUEUE_STATUS_QUEUED, type ComponentType, type JsonPlayObject, type LifecycleStep, type PlayMatchResult, type ScrobbleResult } from "../../core/Atomic"; +import type {PlayApiCommonDetailed} from "../../core/Api"; +import { DEAD_QUEUE, QUEUE_STATUS_COMPLETED, QUEUE_STATUS_FAILED, QUEUE_STATUS_QUEUED, type ComponentType, type JsonPlayObject, type LifecycleStep, type PlayMatchResult, type ScrobbleResult } from "../../core/Atomic"; import { sortByNewestDate } from "../../core/PlayUtils"; import { capitalizeWords } from "../../core/StringUtils"; import { shortTodayAwareFormat } from "../../core/TimeUtils"; @@ -24,6 +23,7 @@ import { ScrobbleMatchResult } from "./ScrobbleMatchResult"; import { TimelineErrorIcon } from "./timeline/TimelineIcon"; import { diffElements, TransformSteps } from "./TransformSteps"; import { Muted } from "./Typography"; +import type { PlayEventQueueStateChange } from '../../core/PlayEvent'; interface ActivityTimelineProps { @@ -56,22 +56,6 @@ const TimelineLoading = () => ( ) -const QueuedCreatedItem = (props: { dead?: boolean, datetime: string }) => ( - - - - - - - - - - {props.dead ? 'Dead ' : ''}Queued at {shortTodayAwareFormat(dayjs(props.datetime))} - - - -) - const NewItem = (props: Pick) => { const { activity: { @@ -166,6 +150,7 @@ const TransformsItem = (props: Pick{transformVerb} using configured Rules for {steps[0].hook} {transformResult}} + unmountOnExit defaultOpen={collapsibleOpen} timeline> @@ -221,6 +206,7 @@ const ScrobbleMatchItem = (props: Pick indicator={Found {match.match ? a duplicate Scrobble : 'no duplicate Scrobbles'}} defaultOpen={collapsibleOpen} disableUntil="md" + unmountOnExit timeline> @@ -280,6 +266,7 @@ const ScrobbleResponseItem = (props: Pick @@ -293,84 +280,70 @@ const ScrobbleResponseItem = (props: Pick { +const QueueTimelineItem = (props: {queueState: PlayEventQueueStateChange, collapsibleOpen: boolean}) => { const { - queueState, + queueState: { + data: { + queueStatus, + queueName, + error, + retries + }, + createdAt, + } = {}, collapsibleOpen, } = props; - if(queueState.queueStatus === QUEUE_STATUS_QUEUED) { - return ( - - - - - - - - - - {queueState.queueName === DEAD_QUEUE ? 'Dead ' : ''}Queued at {shortTodayAwareFormat(dayjs(queueState.updatedAt))} - - - - ); - } - if(queueState.queueStatus === QUEUE_STATUS_COMPLETED) { - return ( - - - - - - - - - - {queueState.queueName === DEAD_QUEUE ? 'Dead ' : ''}Queue finished processing at {shortTodayAwareFormat(dayjs(queueState.updatedAt))} - - - - ); + let indicator: React.JSX.Element, + text: React.JSX.Element, + title: React.JSX.Element; + + switch(queueStatus) { + case QUEUE_STATUS_QUEUED: + indicator = ; + text = {queueName === DEAD_QUEUE ? 'Dead ' : ''}Queued at {shortTodayAwareFormat(dayjs(createdAt))}; + break; + case QUEUE_STATUS_COMPLETED: + indicator = ; + text = {queueName === DEAD_QUEUE ? 'Dead ' : ''}Queue finished processing at {shortTodayAwareFormat(dayjs(createdAt))}; + break; + case QUEUE_STATUS_FAILED: + indicator = ; + text = {queueName === DEAD_QUEUE ? 'Dead ' : ''}Queue failed at {shortTodayAwareFormat(dayjs(createdAt))}; } - if(queueState.queueStatus === QUEUE_STATUS_FAILED) { - let titleContent: React.JSX.Element; - if(queueState.error === undefined) { - titleContent = {queueState.queueName === DEAD_QUEUE ? 'Dead ' : ''}Queue failed at {shortTodayAwareFormat(dayjs(queueState.updatedAt))}; - } else { - titleContent = ( - {queueState.queueName === DEAD_QUEUE ? 'Dead ' : ''}Queue failed at {shortTodayAwareFormat(dayjs(queueState.updatedAt))}} - defaultOpen={collapsibleOpen} - disableUntil="md" - timeline> - - - ) - } - return ( - - - - - - - - - - {titleContent} - - - - ); + indicator={text} + defaultOpen={collapsibleOpen} + disableUntil="md" + timeline + unmountOnExit> + + + ) + } else { + title = text; } -} -type TransformStepsTimelineData = {id: 'transform-steps', dt: Dayjs, steps: LifecycleStep[], original?: JsonPlayObject}; -type TimelineDataTypes = 'new' | 'queue-created-ingress' | 'queue-created-dead' | 'queue-updated-ingress' | 'queue-updated-dead' | 'scrobble-match' | 'scrobble-response' | 'transform-steps'; -type TimelineData = {id: TimelineDataTypes, dt: Dayjs}; + return ( + + + + + {indicator} + + + + + {title} + + + + ); +} export const ActivityTimeline = (props: ActivityTimelineProps) => { @@ -380,173 +353,42 @@ export const ActivityTimeline = (props: ActivityTimelineProps) => { const { activity:{ - play, input, - seenAt, - queueStates, + events = [], } = {}, collapsibleOpen, componentType, componentName } = props; - const { - lifecycle: steps = [], - scrobble: { - match, - payload, - createdAt: scrobbleResultCreatedAt - } = {}, - scrobble, - } = play; const { play: original, } = input || {}; - const timelineItems: (TimelineData|TransformStepsTimelineData)[] = [ - {id: 'new', dt: dayjs(seenAt)}, - ]; + events.sort((a, b) => sortByNewestDate(b.createdAt, a.createdAt)); - // group transforms by hook - const transformGroups: Record = steps.length === 0 ? {} : steps.reduce((acc, curr) => { - if(acc[curr.hook] === undefined) { - acc[curr.hook] = []; - } - return {...acc, [curr.hook]: [...acc[curr.hook], curr]}; - }, {}); + const timelineElements: React.JSX.Element[] = [ + + ]; let lastTransformedPlay = original; - for(const [_,v] of Object.entries(transformGroups)) { - const d: TransformStepsTimelineData = {id: 'transform-steps', dt: dayjs(v[0].createdAt), steps: v, original: lastTransformedPlay}; - const [__, finalPlay] = diffElements(lastTransformedPlay, v); - lastTransformedPlay = finalPlay; - timelineItems.push(d); - } - - const ingressQueue = queueStates.find(x => x.queueName === INGRESS_QUEUE); - if(ingressQueue !== undefined) { - if(ingressQueue.updatedAt === ingressQueue.createdAt) { - // if queue was never updated but contains extra context then only show updated - if(ingressQueue.error !== undefined || ingressQueue.queueStatus === QUEUE_STATUS_FAILED) { - timelineItems.push({id: 'queue-updated-ingress', dt: dayjs(ingressQueue.updatedAt)}); - } else { - timelineItems.push({id: 'queue-created-ingress', dt: dayjs(ingressQueue.createdAt)}); - } - } else { - timelineItems.push({id: 'queue-created-ingress', dt: dayjs(ingressQueue.createdAt)}); - timelineItems.push({id: 'queue-updated-ingress', dt: dayjs(ingressQueue.updatedAt)}); + for(const event of events) { + switch(event.eventName) { + case 'transform': { + //const d: TransformStepsTimelineData = {id: 'transform-steps', dt: dayjs(event.data[0].createdAt), steps: event.data, original: lastTransformedPlay}; + timelineElements.push(); + const [__, finalPlay] = diffElements(lastTransformedPlay, event.data); + lastTransformedPlay = finalPlay; + } break; + case 'queueStateChange': { + timelineElements.push(); + } break; + case 'dupeCheck': { + timelineElements.push(); + } break; + case 'scrobbleResult': + timelineElements.push(); } } - const deadqueue = queueStates.find(x => x.queueName === DEAD_QUEUE); - if(deadqueue !== undefined) { - if(deadqueue.updatedAt === deadqueue.createdAt) { - // if queue was never updated but contains extra context then only show updated - if(deadqueue.error !== undefined || deadqueue.queueStatus === QUEUE_STATUS_FAILED) { - timelineItems.push({id: 'queue-updated-dead', dt: dayjs(deadqueue.updatedAt)}); - } else { - timelineItems.push({id: 'queue-created-dead', dt: dayjs(deadqueue.createdAt)}); - } - } else { - timelineItems.push({id: 'queue-created-dead', dt: dayjs(deadqueue.createdAt)}); - timelineItems.push({id: 'queue-updated-dead', dt: dayjs(deadqueue.updatedAt)}); - }; - } - - if(match !== undefined) { - timelineItems.push({id: 'scrobble-match', dt: dayjs(match.createdAt)}); - } - - // since scrobbleResultCreatedAt has just been implemented older play data will not have it - // and if match was never run, due to error earlier in lifecycle, we need to fallback to oldest event + 1s - timelineItems.sort((a, b) => sortByNewestDate(b.dt, a.dt)); - if(payload !== undefined) { - timelineItems.push({id: 'scrobble-response', dt: dayjs(scrobbleResultCreatedAt ?? match?.createdAt ?? timelineItems[timelineItems.length - 1].dt)}); - } - - // now we sort by date as well as logical order - timelineItems.sort((a, b) => { - // new is always sorted to first in order regardless of timestamp - if(b.id === 'new') { - return 1; - } - if(a.id === 'new') { - return -1; - } - - if(!a.dt.isSame(b.dt)) { - return a.dt.isBefore(b.dt) ? -1 : 1; - } - // if they are the same timestamp then we need to determine the likely logical order - - // queue created always occurs before other actions as the play is queued first, then processed - if(b.id.includes('queue-created')) { - return 1; - } - if(a.id.includes('queue-created')) { - return -1; - } - - // transform steps always occur before scrobble actions - if(a.id.includes('scrobble') && b.id === 'transform-steps') { - return 1; - } - if(b.id.includes('scrobble') && a.id === 'transform-steps') { - return -1; - } - - // dupe matching always occurs before scrobbling - if(b.id === 'scrobble-match' && a.id === 'scrobble-response') { - return 1; - } - if(a.id === 'scrobble-match' && b.id === 'scrobble-response') { - return -1; - } - - // queue updated (finished) always occurs last - if(a.id.includes('queue-updated')) { - return 1; - } - if(b.id.includes('queue-updated')) { - return -1; - } - - // nothing else matched, keep order - return 0; - }); - - const f = timelineItems; - console.log(f); - - - const timelineElements: React.JSX.Element[] = timelineItems.flatMap((x) => { - const timelineKey = `${x.id}-${x.dt.unix()}`; - switch(x.id) { - case 'new': - { - const newElm = ; - if(steps.length === 0) { - return [newElm, ]; - } - return newElm; - } - case 'queue-created-ingress': - case 'queue-created-dead': - return ; - case 'queue-updated-ingress': - return ; - case 'queue-updated-dead': - return ; - case 'scrobble-match': - return ; - case 'scrobble-response': - return ; - case 'transform-steps': - { - const val = x as TransformStepsTimelineData; - return - } - } - return undefined; - }); return ( diff --git a/src/client/components/TransformSteps.tsx b/src/client/components/TransformSteps.tsx index 377a0aca..36302eff 100644 --- a/src/client/components/TransformSteps.tsx +++ b/src/client/components/TransformSteps.tsx @@ -148,6 +148,7 @@ export const TransformSteps = (props: LifeycleStepsTimelineProps) => { indicator={Stage {stageType}-{stageName} in Hook {hook} from {source} {summary}} defaultOpen={collapsibleOpen} disableUntil="md" + unmountOnExit timeline> {error !== undefined && error !== null ? : null} diff --git a/src/core/Api.ts b/src/core/Api.ts index 2a32fa3c..5a7dd90f 100644 --- a/src/core/Api.ts +++ b/src/core/Api.ts @@ -5,6 +5,7 @@ import type { SourceType } from "./Atomic.ts" import type { ComponentType, DateLike, ErrorLike, JsonPlayObject, PlayState, QueueName, Replace, SOURCE_SOT_TYPES, SourcePlayerJson } from "./Atomic.ts" import type { Dayjs } from "dayjs" import type { ErrorIsh } from "./ErrorUtils.ts" +import type { PlayEvent } from "./PlayEvent.ts" export interface PlayApiCommon { uid: string @@ -40,6 +41,7 @@ export interface PlayApiCommonDetailed extends PlayApiCommon { error?: ErrorIsh input?: PlayInputApi queueStates: QueueStateApi[] + events: PlayEvent[] } export type ComponentState = 1 | 2 | 3 | 4 | 5 | 6 | 7; diff --git a/src/core/PlayEvent.ts b/src/core/PlayEvent.ts new file mode 100644 index 00000000..c4ec05e7 --- /dev/null +++ b/src/core/PlayEvent.ts @@ -0,0 +1,50 @@ +import type { Dayjs } from "dayjs"; +import type { DateLike, ErrorLike, LifecycleStep, PlayMatchResult, PlayState, QueueStatus, ScrobbleResult } from "./Atomic.ts"; + +export type PlayEventType = 'transform' | 'queueStateChange' | 'playStateChange' | 'dupeCheck' | 'scrobbleResult'; +export const PLAY_EVENT_TYPE = { + transform: 'transform', + queueStateChange: 'queueStateChange', + playStateChange: 'playStateChange', + dupeCheck: 'dupeCheck', + scrobbleResult: 'scrobbleResult' +} as const satisfies Record; + +export interface BasePlayEvent { + id?: number + playId: number + eventName: K + error?: ErrorLike + createdAt?: D + data: T +} + +export type PlayEventTranformData = LifecycleStep[]; +export type PlayEventTransform = BasePlayEvent<'transform', PlayEventTranformData, D>; + +export interface PlayEventQueueStateChangeData { + queueName: string + queueStatus: QueueStatus + retries?: number + error?: ErrorLike +} +export type PlayEventQueueStateChange = BasePlayEvent<'queueStateChange', PlayEventQueueStateChangeData, D>; + +export interface PlayEventPlayStateChangeData { + state: PlayState + error?: ErrorLike + reason?: string +} +export type PlayEventPlayStateChange = BasePlayEvent<'playStateChange', PlayEventPlayStateChangeData, D>; + +export type PlayEventDupeCheckData = PlayMatchResult; +export type PlayEventDupeCheck = BasePlayEvent<'dupeCheck', PlayEventDupeCheckData, D>; + +export type PlayEventScrobbleResultData = ScrobbleResult; +export type PlayEventScrobbleResult = BasePlayEvent<'scrobbleResult', PlayEventScrobbleResultData, D>; + +export type PlayEvent = PlayEventTransform + | PlayEventQueueStateChange + | PlayEventPlayStateChange + | PlayEventDupeCheck + | PlayEventScrobbleResult; \ No newline at end of file -- 2.51.2