diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 04856c6..8005330 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -83,6 +83,10 @@ jobs: push: true tags: ${{ steps.meta-ui.outputs.tags }} labels: ${{ steps.meta-ui.outputs.labels }} + # Shown in the UI. Same value as the image's org.opencontainers.image.version + # label: `1.2.3` on a v* tag, `main` or `pr-` otherwise. + build-args: | + APP_VERSION=${{ steps.meta-ui.outputs.version }} cache-from: type=gha,scope=ui cache-to: type=gha,mode=max,scope=ui diff --git a/apps/backend/drizzle/0014_uploads.sql b/apps/backend/drizzle/0014_uploads.sql new file mode 100644 index 0000000..7cf8624 --- /dev/null +++ b/apps/backend/drizzle/0014_uploads.sql @@ -0,0 +1,20 @@ +CREATE TABLE `uploads` ( + `id` integer PRIMARY KEY AUTOINCREMENT NOT NULL, + `peer_name` text NOT NULL, + `item_id` text NOT NULL, + `filename` text NOT NULL, + `total_bytes` integer NOT NULL, + `range_start` integer DEFAULT 0 NOT NULL, + `requested_bytes` integer NOT NULL, + `sent_bytes` integer DEFAULT 0 NOT NULL, + `duration_ms` integer, + `status` text NOT NULL, + `error` text, + `started_at` text NOT NULL, + `updated_at` text NOT NULL, + `finished_at` text, + CONSTRAINT "uploads_status_check" CHECK("uploads"."status" in ('uploading', 'completed', 'aborted', 'unconfirmed')) +); +--> statement-breakpoint +CREATE INDEX `uploads_status_idx` ON `uploads` (`status`);--> statement-breakpoint +CREATE INDEX `uploads_started_at_idx` ON `uploads` (`started_at`); \ No newline at end of file diff --git a/apps/backend/drizzle/0015_dashboard_history.sql b/apps/backend/drizzle/0015_dashboard_history.sql new file mode 100644 index 0000000..bcaaaa6 --- /dev/null +++ b/apps/backend/drizzle/0015_dashboard_history.sql @@ -0,0 +1,10 @@ +CREATE TABLE `catalog_size_history` ( + `day` text NOT NULL, + `source_id` text NOT NULL, + `source_name` text NOT NULL, + `titles` integer NOT NULL, + `bytes` integer NOT NULL, + PRIMARY KEY(`day`, `source_id`) +); +--> statement-breakpoint +ALTER TABLE `downloads` ADD `imported_at` text; \ No newline at end of file diff --git a/apps/backend/drizzle/meta/0014_snapshot.json b/apps/backend/drizzle/meta/0014_snapshot.json new file mode 100644 index 0000000..c098ab0 --- /dev/null +++ b/apps/backend/drizzle/meta/0014_snapshot.json @@ -0,0 +1,1054 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "f6d7a728-eaa3-4692-9836-71fd937cbe60", + "prevId": "7a0a25df-a2e9-4f9a-b477-42d28d9a01fc", + "tables": { + "api_keys": { + "name": "api_keys", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "key_hash": { + "name": "key_hash", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "expires_at": { + "name": "expires_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "api_keys_key_hash_unique": { + "name": "api_keys_key_hash_unique", + "columns": [ + "key_hash" + ], + "isUnique": true + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "downloads": { + "name": "downloads", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "torrent_filename": { + "name": "torrent_filename", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "peer_id": { + "name": "peer_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "peer_name": { + "name": "peer_name", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "item_id": { + "name": "item_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "filename": { + "name": "filename", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "dest_path": { + "name": "dest_path", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "part_path": { + "name": "part_path", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "release_size": { + "name": "release_size", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "release_json": { + "name": "release_json", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "expected_bytes": { + "name": "expected_bytes", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "expected_bytes_source": { + "name": "expected_bytes_source", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "expected_bytes_mismatch": { + "name": "expected_bytes_mismatch", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": false + }, + "downloaded_bytes": { + "name": "downloaded_bytes", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "attempts": { + "name": "attempts", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "started_at": { + "name": "started_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "completed_at": { + "name": "completed_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error_code": { + "name": "error_code", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error_details": { + "name": "error_details", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "last_operation": { + "name": "last_operation", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'transfer'" + }, + "operation_failed": { + "name": "operation_failed", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": false + }, + "qb_category": { + "name": "qb_category", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "qb_source_server": { + "name": "qb_source_server", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "source_server_id": { + "name": "source_server_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "import_mode": { + "name": "import_mode", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "import_target": { + "name": "import_target", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "manual_import_command_id": { + "name": "manual_import_command_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "downloads_status_idx": { + "name": "downloads_status_idx", + "columns": [ + "status" + ], + "isUnique": false + }, + "downloads_updated_at_idx": { + "name": "downloads_updated_at_idx", + "columns": [ + "updated_at" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "downloads_status_check": { + "name": "downloads_status_check", + "value": "\"downloads\".\"status\" in ('downloading', 'import_queued', 'imported', 'failed')" + }, + "downloads_expected_bytes_source_check": { + "name": "downloads_expected_bytes_source_check", + "value": "\"downloads\".\"expected_bytes_source\" is null or \"downloads\".\"expected_bytes_source\" in ('content_length', 'content_range', 'release_size')" + }, + "downloads_last_operation_check": { + "name": "downloads_last_operation_check", + "value": "\"downloads\".\"last_operation\" in ('transfer', 'import')" + } + } + }, + "managed_keys": { + "name": "managed_keys", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "key_hash": { + "name": "key_hash", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "server_id": { + "name": "server_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "managed_keys_key_hash_unique": { + "name": "managed_keys_key_hash_unique", + "columns": [ + "key_hash" + ], + "isUnique": true + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "own_catalog_meta": { + "name": "own_catalog_meta", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "generation": { + "name": "generation", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "own_catalog_releases": { + "name": "own_catalog_releases", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "source_id": { + "name": "source_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "title_key": { + "name": "title_key", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "media_type": { + "name": "media_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "tmdb_id": { + "name": "tmdb_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "imdb_id": { + "name": "imdb_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "tvdb_id": { + "name": "tvdb_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "season": { + "name": "season", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "episode": { + "name": "episode", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "own_catalog_releases_source_id_idx": { + "name": "own_catalog_releases_source_id_idx", + "columns": [ + "source_id" + ], + "isUnique": false + }, + "own_catalog_releases_title_key_idx": { + "name": "own_catalog_releases_title_key_idx", + "columns": [ + "title_key" + ], + "isUnique": false + }, + "own_catalog_releases_tmdb_id_idx": { + "name": "own_catalog_releases_tmdb_id_idx", + "columns": [ + "tmdb_id" + ], + "isUnique": false + }, + "own_catalog_releases_imdb_id_idx": { + "name": "own_catalog_releases_imdb_id_idx", + "columns": [ + "imdb_id" + ], + "isUnique": false + }, + "own_catalog_releases_tvdb_id_idx": { + "name": "own_catalog_releases_tvdb_id_idx", + "columns": [ + "tvdb_id" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "own_catalog_releases_media_type_check": { + "name": "own_catalog_releases_media_type_check", + "value": "\"own_catalog_releases\".\"media_type\" in ('movie', 'tv')" + } + } + }, + "own_catalog_source_reads": { + "name": "own_catalog_source_reads", + "columns": { + "source_id": { + "name": "source_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "read_at": { + "name": "read_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "own_catalog_source_titles": { + "name": "own_catalog_source_titles", + "columns": { + "source_id": { + "name": "source_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "title_key": { + "name": "title_key", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": { + "own_catalog_source_titles_source_id_title_key_pk": { + "columns": [ + "source_id", + "title_key" + ], + "name": "own_catalog_source_titles_source_id_title_key_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "own_catalog_titles": { + "name": "own_catalog_titles", + "columns": { + "title_key": { + "name": "title_key", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "media_type": { + "name": "media_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "tmdb_id": { + "name": "tmdb_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "imdb_id": { + "name": "imdb_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "tvdb_id": { + "name": "tvdb_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "release_count": { + "name": "release_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "total_size": { + "name": "total_size", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "resolutions": { + "name": "resolutions", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "poster_url": { + "name": "poster_url", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "backdrop_url": { + "name": "backdrop_url", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "genres": { + "name": "genres", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "revision": { + "name": "revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "deleted": { + "name": "deleted", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": false + } + }, + "indexes": { + "own_catalog_titles_revision_idx": { + "name": "own_catalog_titles_revision_idx", + "columns": [ + "revision" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "own_catalog_titles_media_type_check": { + "name": "own_catalog_titles_media_type_check", + "value": "\"own_catalog_titles\".\"media_type\" in ('movie', 'tv')" + } + } + }, + "peer_catalog_sync": { + "name": "peer_catalog_sync", + "columns": { + "peer_id": { + "name": "peer_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "revision": { + "name": "revision", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "generation": { + "name": "generation", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "synced_at": { + "name": "synced_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "peer_catalog_titles": { + "name": "peer_catalog_titles", + "columns": { + "peer_id": { + "name": "peer_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "title_key": { + "name": "title_key", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "media_type": { + "name": "media_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "tmdb_id": { + "name": "tmdb_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "imdb_id": { + "name": "imdb_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "tvdb_id": { + "name": "tvdb_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "release_count": { + "name": "release_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "total_size": { + "name": "total_size", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "resolutions": { + "name": "resolutions", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "poster_url": { + "name": "poster_url", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "backdrop_url": { + "name": "backdrop_url", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "genres": { + "name": "genres", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": { + "peer_catalog_titles_peer_id_title_key_pk": { + "columns": [ + "peer_id", + "title_key" + ], + "name": "peer_catalog_titles_peer_id_title_key_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": { + "peer_catalog_titles_media_type_check": { + "name": "peer_catalog_titles_media_type_check", + "value": "\"peer_catalog_titles\".\"media_type\" in ('movie', 'tv')" + } + } + }, + "tmdb_metadata": { + "name": "tmdb_metadata", + "columns": { + "media_type": { + "name": "media_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "tmdb_id": { + "name": "tmdb_id", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "fetched_at": { + "name": "fetched_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": { + "tmdb_metadata_media_type_tmdb_id_pk": { + "columns": [ + "media_type", + "tmdb_id" + ], + "name": "tmdb_metadata_media_type_tmdb_id_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": { + "tmdb_metadata_media_type_check": { + "name": "tmdb_metadata_media_type_check", + "value": "\"tmdb_metadata\".\"media_type\" in ('movie', 'tv')" + } + } + }, + "uploads": { + "name": "uploads", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "peer_name": { + "name": "peer_name", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "item_id": { + "name": "item_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "filename": { + "name": "filename", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "total_bytes": { + "name": "total_bytes", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "range_start": { + "name": "range_start", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "requested_bytes": { + "name": "requested_bytes", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "sent_bytes": { + "name": "sent_bytes", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "duration_ms": { + "name": "duration_ms", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "started_at": { + "name": "started_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "finished_at": { + "name": "finished_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "uploads_status_idx": { + "name": "uploads_status_idx", + "columns": [ + "status" + ], + "isUnique": false + }, + "uploads_started_at_idx": { + "name": "uploads_started_at_idx", + "columns": [ + "started_at" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "uploads_status_check": { + "name": "uploads_status_check", + "value": "\"uploads\".\"status\" in ('uploading', 'completed', 'aborted', 'unconfirmed')" + } + } + } + }, + "views": {}, + "enums": {}, + "_meta": { + "schemas": {}, + "tables": {}, + "columns": {} + }, + "internal": { + "indexes": {} + } +} \ No newline at end of file diff --git a/apps/backend/drizzle/meta/0015_snapshot.json b/apps/backend/drizzle/meta/0015_snapshot.json new file mode 100644 index 0000000..764cd5e --- /dev/null +++ b/apps/backend/drizzle/meta/0015_snapshot.json @@ -0,0 +1,1114 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "1a854ff0-e165-4eef-91e0-eb3b7a620ed8", + "prevId": "f6d7a728-eaa3-4692-9836-71fd937cbe60", + "tables": { + "api_keys": { + "name": "api_keys", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "key_hash": { + "name": "key_hash", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "expires_at": { + "name": "expires_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "api_keys_key_hash_unique": { + "name": "api_keys_key_hash_unique", + "columns": [ + "key_hash" + ], + "isUnique": true + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "catalog_size_history": { + "name": "catalog_size_history", + "columns": { + "day": { + "name": "day", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "source_id": { + "name": "source_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "source_name": { + "name": "source_name", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "titles": { + "name": "titles", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "bytes": { + "name": "bytes", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": { + "catalog_size_history_day_source_id_pk": { + "columns": [ + "day", + "source_id" + ], + "name": "catalog_size_history_day_source_id_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "downloads": { + "name": "downloads", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "torrent_filename": { + "name": "torrent_filename", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "peer_id": { + "name": "peer_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "peer_name": { + "name": "peer_name", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "item_id": { + "name": "item_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "filename": { + "name": "filename", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "dest_path": { + "name": "dest_path", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "part_path": { + "name": "part_path", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "release_size": { + "name": "release_size", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "release_json": { + "name": "release_json", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "expected_bytes": { + "name": "expected_bytes", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "expected_bytes_source": { + "name": "expected_bytes_source", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "expected_bytes_mismatch": { + "name": "expected_bytes_mismatch", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": false + }, + "downloaded_bytes": { + "name": "downloaded_bytes", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "attempts": { + "name": "attempts", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "started_at": { + "name": "started_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "completed_at": { + "name": "completed_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "imported_at": { + "name": "imported_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error_code": { + "name": "error_code", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error_details": { + "name": "error_details", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "last_operation": { + "name": "last_operation", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'transfer'" + }, + "operation_failed": { + "name": "operation_failed", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": false + }, + "qb_category": { + "name": "qb_category", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "qb_source_server": { + "name": "qb_source_server", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "source_server_id": { + "name": "source_server_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "import_mode": { + "name": "import_mode", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "import_target": { + "name": "import_target", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "manual_import_command_id": { + "name": "manual_import_command_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "downloads_status_idx": { + "name": "downloads_status_idx", + "columns": [ + "status" + ], + "isUnique": false + }, + "downloads_updated_at_idx": { + "name": "downloads_updated_at_idx", + "columns": [ + "updated_at" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "downloads_status_check": { + "name": "downloads_status_check", + "value": "\"downloads\".\"status\" in ('downloading', 'import_queued', 'imported', 'failed')" + }, + "downloads_expected_bytes_source_check": { + "name": "downloads_expected_bytes_source_check", + "value": "\"downloads\".\"expected_bytes_source\" is null or \"downloads\".\"expected_bytes_source\" in ('content_length', 'content_range', 'release_size')" + }, + "downloads_last_operation_check": { + "name": "downloads_last_operation_check", + "value": "\"downloads\".\"last_operation\" in ('transfer', 'import')" + } + } + }, + "managed_keys": { + "name": "managed_keys", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "key_hash": { + "name": "key_hash", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "server_id": { + "name": "server_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "managed_keys_key_hash_unique": { + "name": "managed_keys_key_hash_unique", + "columns": [ + "key_hash" + ], + "isUnique": true + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "own_catalog_meta": { + "name": "own_catalog_meta", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "generation": { + "name": "generation", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "own_catalog_releases": { + "name": "own_catalog_releases", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "source_id": { + "name": "source_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "title_key": { + "name": "title_key", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "media_type": { + "name": "media_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "tmdb_id": { + "name": "tmdb_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "imdb_id": { + "name": "imdb_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "tvdb_id": { + "name": "tvdb_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "season": { + "name": "season", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "episode": { + "name": "episode", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "own_catalog_releases_source_id_idx": { + "name": "own_catalog_releases_source_id_idx", + "columns": [ + "source_id" + ], + "isUnique": false + }, + "own_catalog_releases_title_key_idx": { + "name": "own_catalog_releases_title_key_idx", + "columns": [ + "title_key" + ], + "isUnique": false + }, + "own_catalog_releases_tmdb_id_idx": { + "name": "own_catalog_releases_tmdb_id_idx", + "columns": [ + "tmdb_id" + ], + "isUnique": false + }, + "own_catalog_releases_imdb_id_idx": { + "name": "own_catalog_releases_imdb_id_idx", + "columns": [ + "imdb_id" + ], + "isUnique": false + }, + "own_catalog_releases_tvdb_id_idx": { + "name": "own_catalog_releases_tvdb_id_idx", + "columns": [ + "tvdb_id" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "own_catalog_releases_media_type_check": { + "name": "own_catalog_releases_media_type_check", + "value": "\"own_catalog_releases\".\"media_type\" in ('movie', 'tv')" + } + } + }, + "own_catalog_source_reads": { + "name": "own_catalog_source_reads", + "columns": { + "source_id": { + "name": "source_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "read_at": { + "name": "read_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "own_catalog_source_titles": { + "name": "own_catalog_source_titles", + "columns": { + "source_id": { + "name": "source_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "title_key": { + "name": "title_key", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": { + "own_catalog_source_titles_source_id_title_key_pk": { + "columns": [ + "source_id", + "title_key" + ], + "name": "own_catalog_source_titles_source_id_title_key_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "own_catalog_titles": { + "name": "own_catalog_titles", + "columns": { + "title_key": { + "name": "title_key", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "media_type": { + "name": "media_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "tmdb_id": { + "name": "tmdb_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "imdb_id": { + "name": "imdb_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "tvdb_id": { + "name": "tvdb_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "release_count": { + "name": "release_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "total_size": { + "name": "total_size", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "resolutions": { + "name": "resolutions", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "poster_url": { + "name": "poster_url", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "backdrop_url": { + "name": "backdrop_url", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "genres": { + "name": "genres", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "revision": { + "name": "revision", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "deleted": { + "name": "deleted", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": false + } + }, + "indexes": { + "own_catalog_titles_revision_idx": { + "name": "own_catalog_titles_revision_idx", + "columns": [ + "revision" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "own_catalog_titles_media_type_check": { + "name": "own_catalog_titles_media_type_check", + "value": "\"own_catalog_titles\".\"media_type\" in ('movie', 'tv')" + } + } + }, + "peer_catalog_sync": { + "name": "peer_catalog_sync", + "columns": { + "peer_id": { + "name": "peer_id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "revision": { + "name": "revision", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "generation": { + "name": "generation", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "synced_at": { + "name": "synced_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "peer_catalog_titles": { + "name": "peer_catalog_titles", + "columns": { + "peer_id": { + "name": "peer_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "title_key": { + "name": "title_key", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "media_type": { + "name": "media_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "tmdb_id": { + "name": "tmdb_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "imdb_id": { + "name": "imdb_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "tvdb_id": { + "name": "tvdb_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "release_count": { + "name": "release_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "total_size": { + "name": "total_size", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "resolutions": { + "name": "resolutions", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "poster_url": { + "name": "poster_url", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "backdrop_url": { + "name": "backdrop_url", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "genres": { + "name": "genres", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": { + "peer_catalog_titles_peer_id_title_key_pk": { + "columns": [ + "peer_id", + "title_key" + ], + "name": "peer_catalog_titles_peer_id_title_key_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": { + "peer_catalog_titles_media_type_check": { + "name": "peer_catalog_titles_media_type_check", + "value": "\"peer_catalog_titles\".\"media_type\" in ('movie', 'tv')" + } + } + }, + "tmdb_metadata": { + "name": "tmdb_metadata", + "columns": { + "media_type": { + "name": "media_type", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "tmdb_id": { + "name": "tmdb_id", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "data": { + "name": "data", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "fetched_at": { + "name": "fetched_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": { + "tmdb_metadata_media_type_tmdb_id_pk": { + "columns": [ + "media_type", + "tmdb_id" + ], + "name": "tmdb_metadata_media_type_tmdb_id_pk" + } + }, + "uniqueConstraints": {}, + "checkConstraints": { + "tmdb_metadata_media_type_check": { + "name": "tmdb_metadata_media_type_check", + "value": "\"tmdb_metadata\".\"media_type\" in ('movie', 'tv')" + } + } + }, + "uploads": { + "name": "uploads", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "peer_name": { + "name": "peer_name", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "item_id": { + "name": "item_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "filename": { + "name": "filename", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "total_bytes": { + "name": "total_bytes", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "range_start": { + "name": "range_start", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "requested_bytes": { + "name": "requested_bytes", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "sent_bytes": { + "name": "sent_bytes", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "duration_ms": { + "name": "duration_ms", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "started_at": { + "name": "started_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "updated_at": { + "name": "updated_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "finished_at": { + "name": "finished_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "uploads_status_idx": { + "name": "uploads_status_idx", + "columns": [ + "status" + ], + "isUnique": false + }, + "uploads_started_at_idx": { + "name": "uploads_started_at_idx", + "columns": [ + "started_at" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": { + "uploads_status_check": { + "name": "uploads_status_check", + "value": "\"uploads\".\"status\" in ('uploading', 'completed', 'aborted', 'unconfirmed')" + } + } + } + }, + "views": {}, + "enums": {}, + "_meta": { + "schemas": {}, + "tables": {}, + "columns": {} + }, + "internal": { + "indexes": {} + } +} \ No newline at end of file diff --git a/apps/backend/drizzle/meta/_journal.json b/apps/backend/drizzle/meta/_journal.json index 1739174..a8bc335 100644 --- a/apps/backend/drizzle/meta/_journal.json +++ b/apps/backend/drizzle/meta/_journal.json @@ -99,6 +99,20 @@ "when": 1793642343855, "tag": "0013_milky_zzzax", "breakpoints": true + }, + { + "idx": 14, + "version": "6", + "when": 1794177717284, + "tag": "0014_uploads", + "breakpoints": true + }, + { + "idx": 15, + "version": "6", + "when": 1794177973906, + "tag": "0015_dashboard_history", + "breakpoints": true } ] } \ No newline at end of file diff --git a/apps/backend/src/__tests__/downloads-service.test.ts b/apps/backend/src/__tests__/downloads-service.test.ts index 6edadb4..dc016fe 100644 --- a/apps/backend/src/__tests__/downloads-service.test.ts +++ b/apps/backend/src/__tests__/downloads-service.test.ts @@ -46,6 +46,9 @@ function downloadsConfig(overrides: Partial> = {}) { manualImportBackoffBaseMs: 60_000, manualImportBackoffMaxMs: 1_800_000, unlinkImportedFiles: false, + removeFailedDownloads: false, + failedDownloadRetries: 3, + failedRetryBaseDelayMs: 60_000, ...overrides, } } diff --git a/apps/backend/src/__tests__/failed-download-reaper.test.ts b/apps/backend/src/__tests__/failed-download-reaper.test.ts new file mode 100644 index 0000000..da39432 --- /dev/null +++ b/apps/backend/src/__tests__/failed-download-reaper.test.ts @@ -0,0 +1,115 @@ +import type { DownloadRecord } from '../modules/downloads/downloads.repository' +import { expect, test } from 'bun:test' +import { FailedDownloadReaper } from '../modules/downloads/failed-download-reaper' + +function failedRow(overrides: Partial = {}): DownloadRecord { + return { + id: 1, + filename: 'Movie.mkv', + status: 'failed', + errorCode: 'PEER_UNREACHABLE', + operationFailed: true, + lastOperation: 'transfer', + importMode: null, + importTarget: null, + ...overrides, + } as DownloadRecord +} + +function setup(rows: DownloadRecord[], options: { enabled?: boolean, retryFails?: boolean } = {}) { + const clock = { now: 0 } + const calls: string[] = [] + const reaper = new FailedDownloadReaper({ + enabled: () => options.enabled ?? true, + list: () => rows, + retry: async (id) => { + calls.push(`retry:${id}`) + if (options.retryFails) + throw new Error('nope') + }, + remove: async (id) => { + calls.push(`remove:${id}`) + rows.splice(rows.findIndex(row => row.id === id), 1) + }, + maxRetries: 3, + baseDelayMs: 100, + maxDelayMs: 250, + now: () => clock.now, + }) + return { reaper, clock, calls } +} + +test('retries with backoff, then removes the download once retries are spent', async () => { + const rows = [ + failedRow(), + ] + const { reaper, clock, calls } = setup(rows) + + // Delays: 100, 200, then 400 capped to 250. Every retry fails again (row stays failed). + for (const at of [ + 0, + 99, + 100, + 100, + 299, + 300, + 300, + 549, + 550, + 550, + ]) + await (clock.now = at, reaper.tick()) + + expect(calls).toEqual([ + 'retry:1', + 'retry:1', + 'retry:1', + 'remove:1', + ]) + expect(rows).toHaveLength(0) +}) + +test('a retry that throws still counts toward the limit', async () => { + const rows = [ + failedRow(), + ] + const { reaper, clock, calls } = setup(rows, { retryFails: true }) + for (let at = 0; at <= 2000; at += 50) + await (clock.now = at, reaper.tick()) + expect(calls).toEqual([ + 'retry:1', + 'retry:1', + 'retry:1', + 'remove:1', + ]) +}) + +test('removes a failure that cannot be retried right away', async () => { + const rows = [ + failedRow({ lastOperation: 'import', importMode: null }), + ] + const { reaper, calls } = setup(rows) + await reaper.tick() + expect(calls).toEqual(['remove:1']) +}) + +test('leaves user cancels and other statuses alone', async () => { + const rows = [ + failedRow({ id: 1, errorCode: 'CANCELLED' }), + failedRow({ id: 2, status: 'downloading' }), + failedRow({ id: 3, status: 'imported' }), + ] + const { reaper, clock, calls } = setup(rows) + for (const at of [0, 1000, 5000]) + await (clock.now = at, reaper.tick()) + expect(calls).toEqual([]) +}) + +test('does nothing while disabled', async () => { + const rows = [ + failedRow({ lastOperation: 'import' }), + ] + const { reaper, calls } = setup(rows, { enabled: false }) + await reaper.tick() + expect(calls).toEqual([]) +}) diff --git a/apps/backend/src/__tests__/live-progress.test.ts b/apps/backend/src/__tests__/live-progress.test.ts new file mode 100644 index 0000000..b3e1cd1 --- /dev/null +++ b/apps/backend/src/__tests__/live-progress.test.ts @@ -0,0 +1,48 @@ +import { expect, test } from 'bun:test' +import { LiveProgressTracker } from '../modules/downloads/live-progress' +import { enrichDownload } from '../modules/status/status.controller' + +test('speed is averaged over the window and resets when bytes go backwards', () => { + const clock = { now: 0 } + const live = new LiveProgressTracker(() => clock.now) + + live.record(1, 0) + expect(live.get(1)).toEqual({ downloadedBytes: 0, speedBps: null }) + + clock.now = 2000 + live.record(1, 4000) + expect(live.get(1)).toEqual({ downloadedBytes: 4000, speedBps: 2000 }) + + // A restart drops the window instead of reporting a negative speed. + clock.now = 3000 + live.record(1, 100) + expect(live.get(1)).toEqual({ downloadedBytes: 100, speedBps: null }) + + live.clear(1) + expect(live.get(1)).toBeNull() +}) + +test('enrichDownload overlays live bytes, speed and ETA on a running transfer', () => { + const clock = { now: 0 } + const live = new LiveProgressTracker(() => clock.now) + live.record(7, 1000) + clock.now = 1000 + live.record(7, 3000) + + const row = { + id: 7, + status: 'downloading', + downloadedBytes: 500, + expectedBytes: 10_000, + releaseSize: 10_000, + } as Parameters[0] + expect(enrichDownload(row, live)).toMatchObject({ + downloadedBytes: 3000, + progress: 0.3, + speedBps: 2000, + etaSeconds: 4, + }) + + // Finished rows ignore whatever the tracker still holds. + expect(enrichDownload({ ...row, status: 'import_queued' }, live)).toMatchObject({ downloadedBytes: 500, speedBps: null, etaSeconds: null }) +}) diff --git a/apps/backend/src/__tests__/peer-download.test.ts b/apps/backend/src/__tests__/peer-download.test.ts index 740d967..eaee098 100644 --- a/apps/backend/src/__tests__/peer-download.test.ts +++ b/apps/backend/src/__tests__/peer-download.test.ts @@ -151,6 +151,45 @@ describe('PeerConnector.downloadFile', () => { } }) + test('asks for a receipt and posts one after the file arrives', async () => { + const receipts: unknown[] = [] + const seen = { receiptHeader: null as string | null } + server.use( + http.get(`${PEER_JACK_URL}/peer/items/:itemId/file`, ({ request }) => { + seen.receiptHeader = request.headers.get('X-Jack-Receipt') + return new Response(streamOf([1, 2, 3, 4]), { + headers: { 'Content-Length': '4', 'X-Jack-Upload-Id': '42' }, + }) + }), + http.post(`${PEER_JACK_URL}/peer/uploads/:uploadId/receipt`, async ({ request, params }) => { + receipts.push({ + uploadId: params.uploadId, + apiKey: request.headers.get('X-Api-Key'), + body: await request.json(), + }) + return new HttpResponse(null, { status: 204 }) + }), + ) + + const peer = markInitialized(new PeerConnector({ url: PEER_JACK_URL, apiKey: 'peer-api-key', name: 'Friend Jack' })) + const dir = await mkdtemp(join(tmpdir(), 'jack-peer-receipt-')) + try { + await peer.downloadFile('remote1:movie:99', join(dir, 'Movie.mkv'), { releaseSize: 4 }) + // Fire and forget: the download resolves without waiting for it. + await waitFor(async () => receipts.length === 1) + expect(seen.receiptHeader).toBe('1') + expect(receipts[0]).toMatchObject({ + uploadId: '42', + apiKey: 'peer-api-key', + body: { bytes: 4 }, + }) + expect((receipts[0] as { body: { durationMs: number } }).body.durationMs).toBeGreaterThanOrEqual(0) + } + finally { + await rm(dir, { recursive: true, force: true }) + } + }) + test('falls back to releaseSize for expected bytes when Content-Length is missing or invalid', async () => { for (const contentLength of [null, 'not-a-number']) { server.resetHandlers() diff --git a/apps/backend/src/__tests__/qbittorrent-mapper.test.ts b/apps/backend/src/__tests__/qbittorrent-mapper.test.ts index 7a2dccd..a3e7829 100644 --- a/apps/backend/src/__tests__/qbittorrent-mapper.test.ts +++ b/apps/backend/src/__tests__/qbittorrent-mapper.test.ts @@ -29,6 +29,7 @@ function baseRecord(overrides: Partial = {}): DownloadRecord { startedAt: '2026-06-06T00:00:00.000Z', updatedAt: '2026-06-06T00:00:00.000Z', completedAt: null, + importedAt: null, error: null, errorCode: null, errorDetails: null, diff --git a/apps/backend/src/__tests__/stats.test.ts b/apps/backend/src/__tests__/stats.test.ts new file mode 100644 index 0000000..fef64f1 --- /dev/null +++ b/apps/backend/src/__tests__/stats.test.ts @@ -0,0 +1,170 @@ +import type { DownloadRecord } from '../modules/downloads/downloads.repository' +import type { UploadRecord } from '../modules/uploads/uploads.repository' +import { Database } from 'bun:sqlite' +import { expect, test } from 'bun:test' +import { drizzle } from 'drizzle-orm/bun-sqlite' +import { runMigrations } from '../database/connection' +import * as schema from '../database/schema' +import { catalogSizeHistory, ownCatalogTitles, peerCatalogTitles } from '../database/schema' +import { buildStats } from '../modules/stats/stats' +import { StatsRepository } from '../modules/stats/stats.repository' + +const NOW = Date.parse('2026-11-08T12:00:00Z') + +function download(overrides: Partial): DownloadRecord { + return { + peerName: 'Ranieri', + status: 'imported', + downloadedBytes: 0, + errorCode: null, + startedAt: '2026-11-08T10:00:00Z', + completedAt: null, + importedAt: null, + ...overrides, + } as DownloadRecord +} + +function upload(overrides: Partial): UploadRecord { + return { + peerName: 'Roz', + filename: 'Movie.mkv', + status: 'completed', + sentBytes: 0, + startedAt: '2026-11-08T10:00:00Z', + finishedAt: null, + ...overrides, + } as UploadRecord +} + +test('buckets transfers, speed and import latency per day', () => { + const stats = buildStats([ + download({ + downloadedBytes: 1000, + startedAt: '2026-11-08T10:00:00Z', + completedAt: '2026-11-08T10:00:10Z', + importedAt: '2026-11-08T10:01:10Z', + }), + download({ + peerName: 'Roz', + status: 'failed', + errorCode: 'PEER_UNREACHABLE', + downloadedBytes: 500, + }), + // Outside the window: counts toward lifetime peers, not the daily series. + download({ downloadedBytes: 7, completedAt: '2026-10-01T00:00:00Z' }), + ], [ + // Confirmed by a receipt: 400 bytes in 4 s. + upload({ sentBytes: 400, durationMs: 4000, finishedAt: '2026-11-07T09:00:04Z' }), + // No receipt, so nothing is known to have arrived. + upload({ sentBytes: 0, status: 'aborted', finishedAt: '2026-11-07T09:30:00Z' }), + upload({ + filename: 'Show.mkv', + peerName: 'Ranieri', + status: 'uploading', + finishedAt: null, + }), + ], [], 3, NOW) + + expect(stats.days).toEqual(['2026-11-06', '2026-11-07', '2026-11-08']) + expect(stats.transfers[2]).toEqual({ + day: '2026-11-08', + downloadedBytes: 1000, + uploadedBytes: 0, + downloadBps: 100, + uploadBps: null, + }) + // Upload speed is the receipt's bytes over its duration. + expect(stats.transfers[1]).toMatchObject({ uploadedBytes: 400, uploadBps: 100 }) + expect(stats.importLatency[2]).toEqual({ day: '2026-11-08', avgSeconds: 60, count: 1 }) + expect(stats.peers.gaveToMe).toEqual([ + { name: 'Ranieri', bytes: 1007 }, + { name: 'Roz', bytes: 500 }, + ]) + expect(stats.peers.tookFromMe).toEqual([ + { name: 'Roz', bytes: 400 }, + ]) + expect(stats.failures).toEqual([ + { code: 'PEER_UNREACHABLE', count: 1 }, + ]) + expect(stats.topTitles[0]).toEqual({ + filename: 'Movie.mkv', + bytes: 400, + requests: 2, + peers: 1, + }) +}) + +test('records one catalog size row per source per day', () => { + const database = new Database(':memory:') + const db = drizzle({ client: database, schema }) + runMigrations(db) + const repo = new StatsRepository(db) + const title = { + mediaType: 'movie' as const, + title: 'A', + releaseCount: 1, + resolutions: [1080], + updatedAt: 'x', + } + db.insert(ownCatalogTitles).values([ + { + ...title, + titleKey: 'a', + totalSize: 10, + revision: 1, + }, + { + ...title, + titleKey: 'b', + totalSize: 99, + revision: 2, + deleted: true, + }, + ]).run() + db.insert(peerCatalogTitles).values([ + { + ...title, + peerId: 'p1', + titleKey: 'a', + totalSize: 5, + }, + { + ...title, + peerId: 'p1', + titleKey: 'b', + totalSize: 6, + }, + { + ...title, + peerId: 'gone', + titleKey: 'a', + totalSize: 1, + }, + ]).run() + + repo.recordCatalogSizes(new Map([ + ['p1', 'Friend'], + ]), '2026-11-08') + repo.recordCatalogSizes(new Map([ + ['p1', 'Friend renamed'], + ]), '2026-11-08') + + expect(db.select().from(catalogSizeHistory).all()).toEqual([ + { + day: '2026-11-08', + sourceId: 'self', + sourceName: 'This instance', + titles: 1, + bytes: 10, + }, + { + day: '2026-11-08', + sourceId: 'p1', + sourceName: 'Friend renamed', + titles: 2, + bytes: 11, + }, + ]) + expect(repo.catalogSizesSince('2026-11-09')).toEqual([]) + database.close() +}) diff --git a/apps/backend/src/__tests__/upload-tracking.test.ts b/apps/backend/src/__tests__/upload-tracking.test.ts new file mode 100644 index 0000000..1e8602b --- /dev/null +++ b/apps/backend/src/__tests__/upload-tracking.test.ts @@ -0,0 +1,122 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, test } from 'bun:test' +import { Hono } from 'hono' +import { openMemoryDatabase } from '../database/connection' +import { PeerController } from '../modules/peer/peer.controller' +import { getPeerRouter } from '../modules/peer/peer.router' +import { UploadsRepository } from '../modules/uploads/uploads.repository' +import { UploadsService } from '../modules/uploads/uploads.service' + +const input = { + peerName: 'alice', + itemId: 'radarr1:movie:1', + filename: 'Movie.mkv', + totalBytes: 12, + rangeStart: 0, + requestedBytes: 12, +} + +describe('UploadsService', () => { + test('a disconnect aborts the row; a finished response leaves it for the receipt', () => { + const repository = new UploadsRepository(openMemoryDatabase()) + const service = new UploadsService(repository) + const controller = new AbortController() + const dropped = service.start(input, true, controller.signal) + const finished = service.start(input, true, new AbortController().signal) + + controller.abort() + + expect(repository.get(dropped)).toMatchObject({ status: 'aborted', error: 'peer disconnected' }) + expect(repository.get(finished)?.status).toBe('uploading') + expect(service.receipt(finished, 'alice', 12, 4000)).toBe(true) + expect(repository.get(finished)).toMatchObject({ status: 'completed', sentBytes: 12, durationMs: 4000 }) + }) + + test('a peer that sends no receipt starts unconfirmed', () => { + const repository = new UploadsRepository(openMemoryDatabase()) + const id = new UploadsService(repository).start(input, false) + expect(repository.get(id)?.status).toBe('unconfirmed') + }) +}) + +describe('peer file route and receipt', () => { + let tempDir: string + let filePath: string + + beforeEach(async () => { + tempDir = await mkdtemp(join(tmpdir(), 'jack-upload-')) + filePath = join(tempDir, 'Movie.mkv') + await Bun.write(filePath, new Uint8Array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])) + }) + + afterEach(async () => { + await rm(tempDir, { recursive: true, force: true }) + }) + + function setup() { + const source = { + id: 'remote1', + name: 'My Radarr', + type: 'radarr', + canSource: true, + getFilePath: async () => filePath, + } + const repository = new UploadsRepository(openMemoryDatabase()) + const app = new Hono() + app.route('/peer', getPeerRouter(new PeerController(() => [source as any]), new UploadsService(repository))) + return { app, repository } + } + + test('serves a range request, then the receipt completes its row', async () => { + const { app, repository } = setup() + const res = await app.request('/peer/items/remote1:movie:1/file', { + headers: { 'Range': 'bytes=6-', 'X-Jack-Receipt': '1' }, + }) + expect([ + ...new Uint8Array(await res.arrayBuffer()), + ]).toEqual([6, 7, 8, 9]) + const uploadId = res.headers.get('X-Jack-Upload-Id') + expect(repository.list()).toEqual([ + expect.objectContaining({ + peerName: 'master', + totalBytes: 10, + rangeStart: 6, + requestedBytes: 4, + status: 'uploading', + }), + ]) + + const receipt = await app.request(`/peer/uploads/${uploadId}/receipt`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ bytes: 4, durationMs: 2 }), + }) + expect(receipt.status).toBe(204) + expect(repository.list()[0]).toMatchObject({ status: 'completed', sentBytes: 4, durationMs: 2 }) + }) + + test('an unknown receipt id is ignored, a malformed body rejected', async () => { + const { app } = setup() + const unknown = await app.request('/peer/uploads/999/receipt', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ bytes: 4, durationMs: 2 }), + }) + expect(unknown.status).toBe(204) + const malformed = await app.request('/peer/uploads/1/receipt', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ bytes: -1 }), + }) + expect(malformed.status).toBe(400) + }) + + test('does not record a HEAD request', async () => { + const { app, repository } = setup() + const res = await app.request('/peer/items/remote1:movie:1/file', { method: 'HEAD' }) + expect(res.headers.get('X-Jack-Upload-Id')).toBeNull() + expect(repository.list()).toEqual([]) + }) +}) diff --git a/apps/backend/src/__tests__/uploads-repository.test.ts b/apps/backend/src/__tests__/uploads-repository.test.ts new file mode 100644 index 0000000..ee2ba89 --- /dev/null +++ b/apps/backend/src/__tests__/uploads-repository.test.ts @@ -0,0 +1,94 @@ +import { describe, expect, test } from 'bun:test' +import { openMemoryDatabase } from '../database/connection' +import { UploadsRepository } from '../modules/uploads/uploads.repository' + +function makeRepo() { + return new UploadsRepository(openMemoryDatabase()) +} + +const input = { + peerName: 'alice', + itemId: 'radarr1:movie:1', + filename: 'Movie.mkv', + totalBytes: 100, + rangeStart: 40, + requestedBytes: 60, +} + +describe('UploadsRepository', () => { + test('creates a row with nothing sent yet', () => { + const row = makeRepo().create(input, 'uploading') + expect(row).toMatchObject({ + ...input, + sentBytes: 0, + durationMs: null, + status: 'uploading', + error: null, + finishedAt: null, + }) + }) + + test('a receipt from the same peer completes it with bytes and duration', () => { + const repo = makeRepo() + const { id } = repo.create(input, 'uploading') + + expect(repo.complete(id, 'mallory', 60, 1000)).toBe(false) + expect(repo.complete(id, 'alice', 60, 2000)).toBe(true) + expect(repo.get(id)).toMatchObject({ status: 'completed', sentBytes: 60, durationMs: 2000 }) + expect(repo.get(id)?.finishedAt).not.toBeNull() + // Closed rows stay closed. + expect(repo.complete(id, 'alice', 1, 1)).toBe(false) + }) + + test('an older peer\'s row can still be completed or aborted', () => { + const repo = makeRepo() + const a = repo.create(input, 'unconfirmed') + const b = repo.create(input, 'unconfirmed') + repo.markAborted(a.id, 'peer disconnected') + expect(repo.complete(b.id, 'alice', 60, 10)).toBe(true) + expect(repo.get(a.id)?.status).toBe('aborted') + }) + + test('a disconnect after completion changes nothing', () => { + const repo = makeRepo() + const { id } = repo.create(input, 'uploading') + repo.complete(id, 'alice', 60, 10) + repo.markAborted(id, 'peer disconnected') + expect(repo.get(id)?.status).toBe('completed') + }) + + test('expires rows that waited too long for a receipt', () => { + const repo = makeRepo() + const { id } = repo.create(input, 'uploading') + expect(repo.expireUnconfirmed('2000-01-01T00:00:00Z')).toBe(0) + expect(repo.expireUnconfirmed('9999-01-01T00:00:00Z')).toBe(1) + expect(repo.get(id)?.status).toBe('unconfirmed') + }) + + test('lists newest first', () => { + const repo = makeRepo() + const first = repo.create(input, 'uploading') + const second = repo.create(input, 'uploading') + expect(repo.list().map(r => r.id)).toEqual([second.id, first.id]) + }) + + test('marks rows left uploading by a previous run as aborted', () => { + const repo = makeRepo() + const running = repo.create(input, 'uploading') + const done = repo.create(input, 'uploading') + repo.complete(done.id, 'alice', 60, 10) + + expect(repo.reconcileInterrupted()).toBe(1) + expect(repo.get(running.id)).toMatchObject({ status: 'aborted', error: 'interrupted by a jack restart' }) + expect(repo.get(done.id)?.status).toBe('completed') + }) + + test('notifies subscribers on every change', () => { + const repo = makeRepo() + let calls = 0 + repo.subscribe(() => calls++) + const { id } = repo.create(input, 'uploading') + repo.complete(id, 'alice', 60, 10) + expect(calls).toBe(2) + }) +}) diff --git a/apps/backend/src/app.ts b/apps/backend/src/app.ts index 7f21166..14bab53 100644 --- a/apps/backend/src/app.ts +++ b/apps/backend/src/app.ts @@ -6,6 +6,7 @@ import type { OwnCatalogRepository } from './modules/catalog/own-catalog.reposit import type { DownloadsRepository } from './modules/downloads/downloads.repository' import type { DownloadsService } from './modules/downloads/downloads.service' import type { ManagedKeysRepository } from './modules/managed-keys/managed-keys.repository' +import type { UploadsService } from './modules/uploads/uploads.service' import { httpInstrumentationMiddleware } from '@hono/otel' import { Hono } from 'hono' import { describeRoute, openAPIRouteHandler, resolver } from 'hono-openapi' @@ -35,6 +36,9 @@ interface AppServices { downloadsService?: DownloadsService // Serves `/peer/catalog`; without it the catalog comes back empty. ownCatalogRepository?: OwnCatalogRepository + // Records files peers pull from `/peer/items/:id/file`; without it they are + // served untracked. + uploadsService?: UploadsService } // This is the external surface, gated per audience rather than by one global key: @@ -134,7 +138,7 @@ export function getApp(envs: Envs, config: AppConfig, connManager: { servers: Co // Peer API — other Jacks talk to us to search and download. Serves empty results // when there's no local source to read from. app.use('/peer/*', peerAuth) - app.route('/peer', getPeerRouter(peerController)) + app.route('/peer', getPeerRouter(peerController, services.uploadsService)) // Torznab API — the operator's Radarr/Sonarr search through us and grab via the // download router. Returns empty results when there are no peers to fan out to. diff --git a/apps/backend/src/database/schema.ts b/apps/backend/src/database/schema.ts index 9f2aabe..4b91dd5 100644 --- a/apps/backend/src/database/schema.ts +++ b/apps/backend/src/database/schema.ts @@ -37,6 +37,8 @@ export const downloads = sqliteTable('downloads', { startedAt: text('started_at').notNull(), updatedAt: text('updated_at').notNull(), completedAt: text('completed_at'), + // When *arr confirmed the import; with completedAt, how long the import took. + importedAt: text('imported_at'), error: text('error'), // A DownloadErrorCode (download-error-codes.ts). No check constraint, so new // codes don't need a migration. @@ -81,6 +83,67 @@ export const apiKeys = sqliteTable('api_keys', { updatedAt: text('updated_at').notNull(), }) +// Files peers pull from this instance through `/peer/items/:id/file`. One row +// per HTTP request: a resumed transfer sends a Range header and gets its own +// row, starting at `rangeStart`. +// +// Files go out zero-copy, so jack never counts the bytes itself. It learns an +// upload ended early from the disconnect (`aborted`), and that it finished from +// the receipt the downloading jack posts back (`completed`, with its duration). +// No receipt (an older peer, or a lost POST) leaves it `unconfirmed`. +export const UPLOAD_STATUSES = [ + 'uploading', + 'completed', + 'aborted', + 'unconfirmed', +] as const +export type UploadStatus = typeof UPLOAD_STATUSES[number] + +export const uploads = sqliteTable('uploads', { + id: integer('id').primaryKey({ autoIncrement: true }), + // Name of the api key the peer authenticated with ('master' for the master key). + peerName: text('peer_name').notNull(), + itemId: text('item_id').notNull(), + filename: text('filename').notNull(), + // Size of the whole file on disk. + totalBytes: integer('total_bytes').notNull(), + // First byte this request serves (0 without a Range header). + rangeStart: integer('range_start').notNull().default(0), + // Bytes this request was asked to send (the response Content-Length). + requestedBytes: integer('requested_bytes').notNull(), + // Bytes the peer says it received; 0 until its receipt arrives. + sentBytes: integer('sent_bytes').notNull().default(0), + // Transfer time the peer reported; null without a receipt, so speed is unknown. + durationMs: integer('duration_ms'), + status: text('status').$type().notNull(), + error: text('error'), + startedAt: text('started_at').notNull(), + updatedAt: text('updated_at').notNull(), + finishedAt: text('finished_at'), +}, t => [ + check('uploads_status_check', sql`${t.status} in ('uploading', 'completed', 'aborted', 'unconfirmed')`), + index('uploads_status_idx').on(t.status), + index('uploads_started_at_idx').on(t.startedAt), +]) + +export type UploadRow = typeof uploads.$inferSelect +export type NewUploadRow = typeof uploads.$inferInsert + +// Catalog size per day: this instance's own catalog (sourceId `self`) and each +// peer's, written after every catalog sync. One row per source per UTC day; a +// later sync the same day overwrites it, so a day holds its last reading. +export const catalogSizeHistory = sqliteTable('catalog_size_history', { + day: text('day').notNull(), + sourceId: text('source_id').notNull(), + sourceName: text('source_name').notNull(), + titles: integer('titles').notNull(), + bytes: integer('bytes').notNull(), +}, t => [ + primaryKey({ + columns: [t.day, t.sourceId], + }), +]) + export type ApiKeyRow = typeof apiKeys.$inferSelect export type NewApiKeyRow = typeof apiKeys.$inferInsert diff --git a/apps/backend/src/index.ts b/apps/backend/src/index.ts index 7fabe33..e39a220 100644 --- a/apps/backend/src/index.ts +++ b/apps/backend/src/index.ts @@ -16,12 +16,17 @@ import { CatalogSyncService } from './modules/catalog/catalog-sync.service' import { OwnCatalogRepository } from './modules/catalog/own-catalog.repository' import { TmdbMetadataRepository } from './modules/catalog/tmdb-metadata.repository' import { ConfigService } from './modules/config/config.service' +import { DownloadsManagementController } from './modules/downloads/downloads.controller' import { DownloadsRepository } from './modules/downloads/downloads.repository' import { DownloadsService } from './modules/downloads/downloads.service' +import { FailedDownloadReaper } from './modules/downloads/failed-download-reaper' import { ImportWatcher } from './modules/downloads/import-watcher' import { ManagedKeysRepository } from './modules/managed-keys/managed-keys.repository' import { ManagedApiKeys } from './modules/managed-keys/managed-keys.service' import { qbCategoryForServer } from './modules/qbittorrent/qbittorrent.mapper' +import { StatsRepository } from './modules/stats/stats.repository' +import { UploadsRepository } from './modules/uploads/uploads.repository' +import { UploadsService } from './modules/uploads/uploads.service' function logRegistrationFailure(what: string, destName: string | undefined, err: unknown) { if (err instanceof FetchError) { @@ -49,6 +54,15 @@ const apiKeysRepository = new ApiKeysRepository(database.db) const managedKeysRepository = new ManagedKeysRepository(database.db) const catalogCacheRepository = new CatalogCacheRepository(database.db) const tmdbMetadataRepository = new TmdbMetadataRepository(database.db) +const uploadsService = new UploadsService(new UploadsRepository(database.db)) +const statsRepository = new StatsRepository(database.db) + +// Peers connected to the previous run lost those transfers with it. +const interruptedUploads = uploadsService.repository.reconcileInterrupted() +if (interruptedUploads > 0) + logger.warn({ uploads: interruptedUploads }, 'Marked uploads interrupted by the previous jack run as aborted') +// A receipt that never came (a lost POST) shouldn't leave an upload running forever. +setInterval(() => uploadsService.expireUnconfirmed(), 60 * 60 * 1000).unref() // Seed the management service from the shared raw object returned by getAppConfig // so the service's persisted state can never diverge from the loaded runtime config. @@ -74,6 +88,15 @@ const catalogSync = new CatalogSyncService({ peerCache: catalogCacheRepository, metadata: tmdbMetadataRepository, schedule: () => configService?.getCatalog().refreshSchedule ?? config.catalog.refreshSchedule, + // Record today's catalog sizes for the dashboard's history chart. + onSynced: () => { + try { + statsRepository.recordCatalogSizes(new Map(connectorManager.peers.map(p => [p.id, p.name ?? p.url]))) + } + catch (err) { + logger.warn({ err }, 'Could not record catalog sizes') + } + }, }) catalogSync.start() // A schedule edited in the UI applies without a restart. @@ -85,6 +108,7 @@ const app = getApp(envs, config, connectorManager, { apiKeysRepository, managedKeysRepository, ownCatalogRepository, + uploadsService, }) const server = Bun.serve({ fetch: app.fetch, @@ -120,6 +144,23 @@ const importWatcher = downloadsConfig }) : undefined +// Retries failed downloads with backoff, then deletes the ones that keep failing. +// Routed through the management controller so transfers and imports each retry the +// way the ***Retry*** button does. +const downloadsController = new DownloadsManagementController(downloadsRepository, downloadsService, importWatcher) +const failedDownloadReaper = downloadsConfig + ? new FailedDownloadReaper({ + // Read per tick, like unlinkImportedFiles, so the UI toggle applies live. + enabled: () => configService?.getRawDownloads()?.removeFailedDownloads ?? downloadsConfig.removeFailedDownloads, + list: () => downloadsRepository.list(), + retry: id => downloadsController.retry(id), + remove: id => downloadsController.delete(id), + maxRetries: downloadsConfig.failedDownloadRetries, + baseDelayMs: downloadsConfig.failedRetryBaseDelayMs, + maxDelayMs: downloadsConfig.retryMaxDelayMs, + }) + : undefined + function startManagementServer() { if (!envs.MANAGEMENT_KEY) return undefined @@ -137,6 +178,8 @@ function startManagementServer() { downloadsRepository, downloadsService, importWatcher, + uploadsService, + statsRepository, apiKeysRepository, catalogCacheRepository, tmdbMetadataRepository, @@ -185,6 +228,8 @@ for (const dest of registrable) { managedApiKeys.prune(registrable.map(d => d.id)) importWatcher?.start() +if (downloadsConfig) + failedDownloadReaper?.start(downloadsConfig.importPollIntervalMs) // Re-drive interrupted downloads from a prior run. if (config.downloads && downloadsService) { @@ -204,6 +249,7 @@ else { process.on('SIGINT', async () => { logger.info('SIGINT received, exiting') importWatcher?.stop() + failedDownloadReaper?.stop() catalogSync.stop() database.close() server.stop() @@ -215,6 +261,7 @@ process.on('SIGINT', async () => { process.on('SIGTERM', async () => { logger.info('SIGTERM received, exiting') importWatcher?.stop() + failedDownloadReaper?.stop() catalogSync.stop() database.close() server.stop() diff --git a/apps/backend/src/lib/config.ts b/apps/backend/src/lib/config.ts index 176a83f..7c3cbfc 100644 --- a/apps/backend/src/lib/config.ts +++ b/apps/backend/src/lib/config.ts @@ -276,6 +276,15 @@ export const DownloadsConfig = z.object({ // library's link) or copied it. Enabling this unlinks jack's directory entry — // never a recursive delete — right after a row flips to `imported`. unlinkImportedFiles: z.boolean().default(false), + // When enabled, a download that ends up `failed` (a transfer or import whose own + // retries above are spent, or a permanent error) is retried as a whole up to + // `failedDownloadRetries` times, backing off exponentially from + // `failedRetryBaseDelayMs` (capped at `retryMaxDelayMs`). If it still fails, its + // row and any partial or completed file are deleted. User cancels are left alone. + removeFailedDownloads: z.boolean().default(false), + // At least 3: network errors are often only reported once per attempt. + failedDownloadRetries: z.number().int().min(3).default(3), + failedRetryBaseDelayMs: z.number().int().min(0).default(60_000), }) export type DownloadsConfig = z.infer @@ -300,6 +309,9 @@ export const RawDownloadsConfig = z.object({ manualImportBackoffBaseMs: z.number().int().min(0).nullish(), manualImportBackoffMaxMs: z.number().int().min(0).nullish(), unlinkImportedFiles: z.boolean().nullish(), + removeFailedDownloads: z.boolean().nullish(), + failedDownloadRetries: z.number().int().min(3).nullish(), + failedRetryBaseDelayMs: z.number().int().min(0).nullish(), }) export type RawDownloadsConfig = z.infer diff --git a/apps/backend/src/lib/openapi.ts b/apps/backend/src/lib/openapi.ts index 5e8824a..be7d3ea 100644 --- a/apps/backend/src/lib/openapi.ts +++ b/apps/backend/src/lib/openapi.ts @@ -69,6 +69,7 @@ export const managementDocumentation: GenerateSpecOptions['documentation'] = { { name: 'System', description: 'Health probe and the connector overview.' }, { name: 'Config', description: 'Read and mutate the persisted jack/servers/peers configuration.' }, { name: 'Downloads', description: 'List and stream download records, and manage individual downloads by their numeric id.' }, + { name: 'Uploads', description: 'List and stream the files peers pull from this instance.' }, { name: 'Logs', description: 'Backfill and live-tail the process\'s own logs.' }, { name: 'Catalog', description: 'Aggregated peer catalog and TMDB-backed metadata for the UI.' }, { name: 'API keys', description: 'Issue, list, and revoke the peer API keys this instance hands out.' }, diff --git a/apps/backend/src/lib/servers/peer.ts b/apps/backend/src/lib/servers/peer.ts index 818b28e..a0dba21 100644 --- a/apps/backend/src/lib/servers/peer.ts +++ b/apps/backend/src/lib/servers/peer.ts @@ -22,11 +22,14 @@ const PeerSearchResponse = z.object({ const MAX_DOWNLOAD_BYTES = 100 * 1024 * 1024 * 1024 // 100GB const DOWNLOAD_PROGRESS_INTERVAL_MS = 10_000 const DOWNLOAD_PROGRESS_BYTES = 64 * 1024 * 1024 +const DOWNLOAD_LIVE_INTERVAL_MS = 1000 const CONTENT_RANGE_PATTERN = /^bytes (\d+)-(\d+)\/(\d+)$/ export type PeerDownloadProgressEvent = | { type: 'headers', expectedBytes: number | null, expectedBytesSource: 'content_length' | 'content_range' | 'release_size' | null, expectedBytesMismatch: boolean } | { type: 'progress', downloadedBytes: number, expectedBytes: number | null } + // About once a second, for live progress only; `progress` is the durable one. + | { type: 'bytes', downloadedBytes: number } | { type: 'restart', reason: 'range_ignored' | 'content_range_mismatch' | 'range_not_satisfiable' | 'part_oversize', discardedBytes: number } | { type: 'completed', downloadedBytes: number, expectedBytes: number | null } @@ -255,6 +258,40 @@ export class PeerConnector extends ServerConnector { } @requiresInitialization + /** + * Tell the serving jack this file arrived and how long it took. It sends files + * zero-copy, so it can see an early disconnect but not the end of a transfer. + * Fire and forget: a lost receipt only leaves that upload's speed unknown, and + * a peer too old to issue upload ids gets none. + */ + private sendUploadReceipt(uploadId: string | null, bytes: number, durationMs: number): void { + if (!uploadId || !/^\d+$/.test(uploadId)) + return + const url = new URL(`/peer/uploads/${uploadId}/receipt`, this.url) + fetch(url, { + method: 'POST', + headers: { ...this.headers, 'X-Api-Key': this.apiKey, 'Content-Type': 'application/json' }, + body: JSON.stringify({ + bytes, + durationMs: Math.round(durationMs), + }), + signal: AbortSignal.timeout(10_000), + }) + .then((response) => { + if (!response.ok) + logger.debug({ peer: this.name, uploadId, status: response.status }, 'Peer did not accept the upload receipt') + }) + .catch((err) => { + logger.debug({ + peer: this.name, + uploadId, + error: err instanceof Error + ? err.message + : String(err), + }, 'Could not send the upload receipt') + }) + } + async downloadFile(id: string, destPath: string, options: PeerDownloadOptions = {}): Promise { return withSpan('peer.download_file', { 'peer.name': this.name, @@ -266,7 +303,8 @@ export class PeerConnector extends ServerConnector { const torrentFilename = options.torrentFilename const url = new URL(`/peer/items/${encodeURIComponent(id)}/file`, this.url) const partPath = options.partPath ?? `${destPath}.part` - const baseHeaders = { ...this.headers, 'X-Api-Key': this.apiKey } + // X-Jack-Receipt: we'll report back when the file arrived (see sendUploadReceipt). + const baseHeaders = { ...this.headers, 'X-Api-Key': this.apiKey, 'X-Jack-Receipt': '1' } setSpanAttributes(span, { 'http.request.idle_timeout_ms': idleTimeoutMs, 'url.path': url.pathname }) // Idle (inactivity) timeout, armed ONLY around network waits (fetch + each @@ -496,8 +534,12 @@ export class PeerConnector extends ServerConnector { throw err } let downloadedBytes = existingBytes + // What this request carries, for the receipt: a resume only counts its range. + const requestStartBytes = existingBytes + const requestStartedAt = performance.now() let lastLoggedAt = Date.now() let lastLoggedBytes = downloadedBytes + let lastLiveAt = 0 let handleClosed = false const closeHandle = async () => { @@ -565,6 +607,11 @@ export class PeerConnector extends ServerConnector { lastLoggedAt = now lastLoggedBytes = downloadedBytes } + // Cheap in-memory tick for live progress and speed: no fsync, no DB write. + if (now - lastLiveAt >= DOWNLOAD_LIVE_INTERVAL_MS) { + await options.onProgress?.({ type: 'bytes', downloadedBytes }) + lastLiveAt = now + } } await syncPart(`flushing the .part to disk after the last chunk (${downloadedBytes} bytes written)`) @@ -582,6 +629,7 @@ export class PeerConnector extends ServerConnector { signal.throwIfAborted() await withFsContext('move_completed', 'moving the completed .part into place', [partPath, destPath], () => rename(partPath, destPath)) setSpanAttribute(span, 'download.downloaded_bytes', downloadedBytes) + this.sendUploadReceipt(response.headers.get('X-Jack-Upload-Id'), downloadedBytes - requestStartBytes, performance.now() - requestStartedAt) try { await options.onProgress?.({ type: 'completed', downloadedBytes, expectedBytes }) } diff --git a/apps/backend/src/management-app.ts b/apps/backend/src/management-app.ts index 981d28b..fece721 100644 --- a/apps/backend/src/management-app.ts +++ b/apps/backend/src/management-app.ts @@ -7,6 +7,8 @@ import type { ConfigService } from './modules/config/config.service' import type { DownloadsRepository } from './modules/downloads/downloads.repository' import type { DownloadsService } from './modules/downloads/downloads.service' import type { ImportWatcher } from './modules/downloads/import-watcher' +import type { StatsRepository } from './modules/stats/stats.repository' +import type { UploadsService } from './modules/uploads/uploads.service' import { Hono } from 'hono' import { describeRoute, openAPIRouteHandler, resolver } from 'hono-openapi' import { secureHeaders } from 'hono/secure-headers' @@ -40,6 +42,8 @@ export function getManagementApp(params: { downloadsRepository?: DownloadsRepository downloadsService?: DownloadsService importWatcher?: ImportWatcher + uploadsService?: UploadsService + statsRepository?: StatsRepository apiKeysRepository?: ApiKeysRepository catalogCacheRepository?: CatalogCacheRepository tmdbMetadataRepository?: TmdbMetadataRepository @@ -84,7 +88,7 @@ export function getManagementApp(params: { const configController = new ConfigController(params.connectors, params.configService) app.route('/config', getConfigRouter(configController)) - const statusController = new StatusController(params.connectors, params.downloadsRepository) + const statusController = new StatusController(params.connectors, params.downloadsRepository, params.uploadsService, params.statsRepository) app.route('/', getStatusRouter(statusController)) app.route('/downloads', getDownloadsManagementRouter(new DownloadsManagementController( diff --git a/apps/backend/src/modules/catalog/catalog-sync.service.ts b/apps/backend/src/modules/catalog/catalog-sync.service.ts index 88e9509..bf9a815 100644 --- a/apps/backend/src/modules/catalog/catalog-sync.service.ts +++ b/apps/backend/src/modules/catalog/catalog-sync.service.ts @@ -50,6 +50,8 @@ export interface CatalogSyncDeps { metadata?: Pick // The current cron expression; read again on every reschedule(). schedule: () => string + // Called after each successful full sync, e.g. to record catalog sizes. Must not throw. + onSynced?: () => void } /** @@ -207,6 +209,7 @@ export class CatalogSyncService { ]) }) this.syncedAt = new Date().toISOString() + this.deps.onSynced?.() } catch (err) { logger.error({ err }, 'Catalog sync failed') diff --git a/apps/backend/src/modules/config/config.router.ts b/apps/backend/src/modules/config/config.router.ts index 2eb18c0..ed7abe0 100644 --- a/apps/backend/src/modules/config/config.router.ts +++ b/apps/backend/src/modules/config/config.router.ts @@ -96,8 +96,9 @@ export function getConfigRouter(controller: ConfigController) { }) // Partial patch: the body is merged onto the stored downloads block. Every knob - // but `unlinkImportedFiles` is captured at boot, so those land on next restart. - app.patch('/downloads', configDoc('Update the downloads block', 'Merges the given fields into the stored downloads config. `unlinkImportedFiles` applies immediately; the other values take effect on next boot.'), zValidator('json', RawDownloadsConfig), async (c) => { + // but `unlinkImportedFiles` and `removeFailedDownloads` is captured at boot, so + // those take effect on next restart. + app.patch('/downloads', configDoc('Update the downloads block', 'Merges the given fields into the stored downloads config. `unlinkImportedFiles` and `removeFailedDownloads` apply immediately; the other values take effect on next boot.'), zValidator('json', RawDownloadsConfig), async (c) => { return c.json(await controller.updateDownloads(c.req.valid('json'))) }) diff --git a/apps/backend/src/modules/config/config.service.ts b/apps/backend/src/modules/config/config.service.ts index 70609b3..3d14dc4 100644 --- a/apps/backend/src/modules/config/config.service.ts +++ b/apps/backend/src/modules/config/config.service.ts @@ -376,8 +376,9 @@ export class ConfigService { * this is what the UI's empty field means. `undefined` (absent) leaves the stored * value alone. * - * Only `unlinkImportedFiles` takes effect immediately (the import watcher reads it - * per import); the rest are captured at boot and land on the next restart. + * Only `unlinkImportedFiles` and `removeFailedDownloads` take effect immediately + * (the import watcher and the failed-download reaper read them as they run); the + * rest are captured at boot and apply on the next restart. */ async updateDownloads(input: unknown): Promise { const patch = RawDownloadsConfig.parse(input) diff --git a/apps/backend/src/modules/downloads/downloads.repository.ts b/apps/backend/src/modules/downloads/downloads.repository.ts index 284728b..fff2b52 100644 --- a/apps/backend/src/modules/downloads/downloads.repository.ts +++ b/apps/backend/src/modules/downloads/downloads.repository.ts @@ -27,6 +27,7 @@ export interface DownloadRecord { startedAt: string updatedAt: string completedAt: string | null + importedAt: string | null error: string | null errorCode: DownloadErrorCode | null errorDetails: FileSystemErrorDetails | null @@ -83,6 +84,7 @@ function toRecord(row: DownloadRow): DownloadRecord { startedAt: row.startedAt, updatedAt: row.updatedAt, completedAt: row.completedAt, + importedAt: row.importedAt, error: row.error, errorCode: (row.errorCode as DownloadErrorCode | null) ?? null, errorDetails: row.errorDetails ?? null, @@ -217,15 +219,17 @@ export class DownloadsRepository { // *arr finished importing the file into the library — terminal success. The row // is kept (not deleted) so the downloads list doubles as a history. markImported(id: number): void { + const timestamp = nowIso() this.db.update(downloads) .set({ status: 'imported', + importedAt: timestamp, lastOperation: 'import', operationFailed: false, error: null, errorCode: null, errorDetails: null, - updatedAt: nowIso(), + updatedAt: timestamp, }) .where(eq(downloads.id, id)) .run() diff --git a/apps/backend/src/modules/downloads/downloads.service.ts b/apps/backend/src/modules/downloads/downloads.service.ts index ed33cc4..048e60d 100644 --- a/apps/backend/src/modules/downloads/downloads.service.ts +++ b/apps/backend/src/modules/downloads/downloads.service.ts @@ -14,6 +14,7 @@ import { logger } from '../../logger' import { unlinkDownloadArtifact } from './artifact-cleanup' import { transferErrorCode, transferErrorLogFields } from './download-error-codes' import { coordinatorFor } from './download-operation-coordinator' +import { liveProgressFor } from './live-progress' import { downloadRetryAfterMs, isTransientDownloadError } from './retry-policy' type DownloadsServiceConfig = NonNullable @@ -149,6 +150,7 @@ export class DownloadsService { startedAt: '', updatedAt: '', completedAt: null, + importedAt: null, error: null, errorCode: null, errorDetails: null, @@ -374,11 +376,14 @@ export class DownloadsService { } finally { this.active.delete(record.destPath) + if (this.downloadsRepository) + liveProgressFor(this.downloadsRepository).clear(record.id) } } private async downloadWithRetry(record: DownloadRecord, signal: AbortSignal): Promise { const repo = this.downloadsRepository + const live = repo && liveProgressFor(repo) const peer = this.peers.find(p => p.id === record.peerId) if (!peer) { repo?.markFailed(record.id, 'PEER_NOT_FOUND', `Peer ${record.peerId} not found`) @@ -391,12 +396,18 @@ export class DownloadsService { repo?.setExpectedBytes(record.id, event.expectedBytes, event.expectedBytesSource, event.expectedBytesMismatch) return } + if (event.type === 'bytes') { + live?.record(record.id, event.downloadedBytes) + return + } if (event.type === 'progress') { repo?.updateProgress(record.id, event.downloadedBytes) + live?.record(record.id, event.downloadedBytes) return } if (event.type === 'restart') { repo?.markResumeReset(record.id) + live?.clear(record.id) return } // event.type === 'completed': the transfer finished. Record the final byte diff --git a/apps/backend/src/modules/downloads/failed-download-reaper.ts b/apps/backend/src/modules/downloads/failed-download-reaper.ts new file mode 100644 index 0000000..a39977d --- /dev/null +++ b/apps/backend/src/modules/downloads/failed-download-reaper.ts @@ -0,0 +1,153 @@ +import type { DownloadRecord } from './downloads.repository' +import { logger } from '../../logger' + +export interface FailedDownloadReaperOptions { + /** Read every tick, so toggling `downloads.removeFailedDownloads` applies live. */ + enabled: () => boolean + list: () => DownloadRecord[] + /** Repeats the row's last failed operation (transfer or import). */ + retry: (id: number) => Promise + /** Deletes the row and its unshared partial or completed files. */ + remove: (id: number) => Promise + maxRetries: number + baseDelayMs: number + maxDelayMs: number + /** Injectable for tests. */ + now?: () => number +} + +/** Same rule the UI uses to offer ***Retry***: only these failures can be repeated. */ +function canRetry(row: DownloadRecord): boolean { + if (!row.operationFailed) + return false + if (row.lastOperation === 'transfer') + return true + return row.lastOperation === 'import' && row.importMode === 'jack_manual' && row.importTarget != null +} + +/** + * Retries failed downloads with exponential backoff and deletes the ones that keep + * failing. Each tick, a failed row either waits out its backoff, gets retried, or + * (retries spent, or not retryable) is removed with its files. + * + * Retry counts live in memory: a restart starts every row's count over. + */ +export class FailedDownloadReaper { + // `nextAt` is unset right after a retry fires; the next tick that sees the row + // failed again schedules the following retry from that moment. + private readonly state = new Map() + private timer?: ReturnType + private tickInFlight?: Promise + + constructor(private readonly options: FailedDownloadReaperOptions) {} + + private delayFor(retries: number): number { + return Math.min(this.options.maxDelayMs, this.options.baseDelayMs * 2 ** retries) + } + + start(intervalMs: number): void { + if (this.timer) + return + this.timer = setInterval(() => { + void this.tick().catch((err) => { + logger.warn({ + error: err instanceof Error + ? err.message + : String(err), + }, 'Failed download reaper tick failed') + }) + }, intervalMs) + } + + stop(): void { + clearInterval(this.timer) + this.timer = undefined + } + + tick(): Promise { + // One pass at a time: a slow delete must not overlap the next interval. + this.tickInFlight ??= this.run().finally(() => { + this.tickInFlight = undefined + }) + return this.tickInFlight + } + + private async run(): Promise { + if (!this.options.enabled()) + return + const now = (this.options.now ?? Date.now)() + const rows = this.options.list() + + // Forget rows that are gone or finished, so their ids don't keep stale counts. + const live = new Map(rows.map(row => [row.id, row])) + for (const id of this.state.keys()) { + const status = live.get(id)?.status + if (status === undefined || status === 'imported') + this.state.delete(id) + } + + for (const row of rows) { + // A user cancel keeps its .part on purpose; leave it for them. + if (row.status !== 'failed' || row.errorCode === 'CANCELLED') + continue + + const entry = this.state.get(row.id) ?? { retries: 0 } + this.state.set(row.id, entry) + + if (entry.nextAt === undefined) { + if (entry.retries >= this.options.maxRetries || !canRetry(row)) { + await this.removeRow(row, entry.retries) + continue + } + entry.nextAt = now + this.delayFor(entry.retries) + continue + } + if (now < entry.nextAt) + continue + + entry.retries++ + entry.nextAt = undefined + try { + await this.options.retry(row.id) + logger.info({ + id: row.id, + filename: row.filename, + retry: entry.retries, + maxRetries: this.options.maxRetries, + }, 'Retrying failed download') + } + catch (err) { + // Counts as a spent retry; the next tick schedules the following one. + logger.warn({ + id: row.id, + retry: entry.retries, + error: err instanceof Error + ? err.message + : String(err), + }, 'Could not retry failed download') + } + } + } + + private async removeRow(row: DownloadRecord, retries: number): Promise { + try { + await this.options.remove(row.id) + this.state.delete(row.id) + logger.info({ + id: row.id, + filename: row.filename, + retries, + error_code: row.errorCode, + }, 'Removed failed download') + } + catch (err) { + // State is kept, so the next tick tries the removal again. + logger.warn({ + id: row.id, + error: err instanceof Error + ? err.message + : String(err), + }, 'Could not remove failed download') + } + } +} diff --git a/apps/backend/src/modules/downloads/live-progress.ts b/apps/backend/src/modules/downloads/live-progress.ts new file mode 100644 index 0000000..ef4179f --- /dev/null +++ b/apps/backend/src/modules/downloads/live-progress.ts @@ -0,0 +1,92 @@ +import type { DownloadsRepository } from './downloads.repository' + +// Speed is averaged over this much recent history. +const SPEED_WINDOW_MS = 10_000 +// Live updates fire at most this often, so the SSE stream (a full snapshot per +// notify) stays at about one frame per second however many transfers run. +const NOTIFY_INTERVAL_MS = 1000 + +export interface LiveProgress { + downloadedBytes: number + /** Bytes per second over the recent window, or null until two samples exist. */ + speedBps: number | null +} + +/** + * In-memory byte counts for running transfers, fed about once a second by the + * peer download loop. The database only stores progress every 10 s / 64 MiB + * (each write also fsyncs the .part), so this is what lets the UI move the bar + * and show a speed in between. Nothing here is persisted. + */ +export class LiveProgressTracker { + private readonly samples = new Map() + private readonly subscribers = new Set<() => void>() + private notifyTimer?: ReturnType + + constructor(private readonly now: () => number = Date.now) {} + + record(id: number, bytes: number): void { + const t = this.now() + const kept = (this.samples.get(id) ?? []).filter(s => t - s.t <= SPEED_WINDOW_MS) + // Fewer bytes than before means the transfer started over (restart or retry). + const window = kept.length > 0 && bytes < kept.at(-1)!.bytes + ? [] + : kept + window.push({ t, bytes }) + this.samples.set(id, window) + this.scheduleNotify() + } + + clear(id: number): void { + if (this.samples.delete(id)) + this.scheduleNotify() + } + + get(id: number): LiveProgress | null { + const window = this.samples.get(id) + if (!window?.length) + return null + const first = window[0]! + const last = window.at(-1)! + const elapsedMs = last.t - first.t + return { + downloadedBytes: last.bytes, + speedBps: elapsedMs > 0 + ? ((last.bytes - first.bytes) * 1000) / elapsedMs + : null, + } + } + + subscribe(subscriber: () => void): () => void { + this.subscribers.add(subscriber) + return () => this.subscribers.delete(subscriber) + } + + private scheduleNotify(): void { + if (this.notifyTimer) + return + this.notifyTimer = setTimeout(() => { + this.notifyTimer = undefined + for (const subscriber of this.subscribers) { + try { + subscriber() + } + catch { + // A broken observer must not stop the transfer that fed it. + } + } + }, NOTIFY_INTERVAL_MS) + } +} + +const trackers = new WeakMap() + +/** One tracker per repository, shared by the service (writer) and status API (reader). */ +export function liveProgressFor(repository: DownloadsRepository): LiveProgressTracker { + let tracker = trackers.get(repository) + if (!tracker) { + tracker = new LiveProgressTracker() + trackers.set(repository, tracker) + } + return tracker +} diff --git a/apps/backend/src/modules/peer/peer.controller.ts b/apps/backend/src/modules/peer/peer.controller.ts index 267d34e..89853ec 100644 --- a/apps/backend/src/modules/peer/peer.controller.ts +++ b/apps/backend/src/modules/peer/peer.controller.ts @@ -39,7 +39,9 @@ export function parseRangeHeader(value: string | undefined | null): { start?: nu // Handing the BunFile straight to `new Response` lets Bun.serve stream it with native // backpressure (sendfile): if the consumer stalls or aborts mid-download, we stop // reading from disk instead of buffering the whole file into RAM. Pumping `.stream()` -// ourselves does NOT backpressure and lets one stalled peer OOM the process. +// ourselves does NOT backpressure and lets one stalled peer OOM the process. It is +// also why upload tracking never touches the body: the downloading jack reports +// the bytes and duration back instead (see UploadsService). export type StreamFileResult = | { type: 'full', body: Blob, size: number, filename: string } | { type: 'partial', body: Blob, size: number, totalSize: number, start: number, end: number, filename: string } diff --git a/apps/backend/src/modules/peer/peer.router.ts b/apps/backend/src/modules/peer/peer.router.ts index af780b8..6080e2b 100644 --- a/apps/backend/src/modules/peer/peer.router.ts +++ b/apps/backend/src/modules/peer/peer.router.ts @@ -1,3 +1,5 @@ +import type { AuthVariables } from '../../middleware/require-auth' +import type { UploadsService } from '../uploads/uploads.service' import type { PeerController } from './peer.controller' import { Hono } from 'hono' import { describeRoute, resolver, validator as zValidator } from 'hono-openapi' @@ -12,8 +14,13 @@ function safeFilename(name: string): string { return name.replace(UNSAFE_FILENAME_CHARS, '') } -export function getPeerRouter(controller: PeerController) { - const app = new Hono() +const UploadReceipt = z.object({ + bytes: z.number().int().min(0), + durationMs: z.number().int().min(0), +}) + +export function getPeerRouter(controller: PeerController, uploads?: UploadsService) { + const app = new Hono<{ Variables: AuthVariables }>() app.get( '/search', @@ -132,6 +139,29 @@ export function getPeerRouter(controller: PeerController) { }) } + // Record the upload; the body still goes out zero-copy. A HEAD request sends + // no body, so it is not an upload. The master key sets no name. + const uploadId = uploads && c.req.method !== 'HEAD' + ? uploads.start({ + peerName: c.get('apiKeyName') ?? 'master', + itemId, + filename: result.filename, + totalBytes: result.type === 'partial' + ? result.totalSize + : result.size, + rangeStart: result.type === 'partial' + ? result.start + : 0, + requestedBytes: result.size, + }, c.req.header('X-Jack-Receipt') === '1', c.req.raw.signal) + : undefined + // Where the downloading jack posts its receipt (see /uploads/:id/receipt). + const uploadHeader: Record = uploadId === undefined + ? {} + : { + 'X-Jack-Upload-Id': String(uploadId), + } + if (result.type === 'partial') { return new Response(result.body, { status: 206, @@ -141,6 +171,7 @@ export function getPeerRouter(controller: PeerController) { 'Content-Range': `bytes ${result.start}-${result.end}/${result.totalSize}`, 'Accept-Ranges': 'bytes', 'Content-Disposition': `attachment; filename="${safeFilename(result.filename)}"`, + ...uploadHeader, }, }) } @@ -151,9 +182,29 @@ export function getPeerRouter(controller: PeerController) { 'Content-Length': String(result.size), 'Accept-Ranges': 'bytes', 'Content-Disposition': `attachment; filename="${result.filename}"`, + ...uploadHeader, }, }) }) + app.post('/uploads/:uploadId/receipt', describeRoute({ + tags: ['Peer'], + summary: 'Report a finished download', + description: 'Sent by the downloading jack after a file from `/peer/items/:itemId/file` arrived, with the `X-Jack-Upload-Id` that response carried. Records how many bytes the request delivered and how long it took. Always answers 204, so a stale or unknown id is not an error worth retrying.', + security: [ + { 'X-Api-Key': [] }, + ], + responses: { + 204: { description: 'Receipt taken (or ignored)' }, + 400: { description: 'Malformed receipt' }, + }, + }), zValidator('json', UploadReceipt), (c) => { + const uploadId = Number(c.req.param('uploadId')) + const { bytes, durationMs } = c.req.valid('json') + if (uploads && Number.isSafeInteger(uploadId)) + uploads.receipt(uploadId, c.get('apiKeyName') ?? 'master', bytes, durationMs) + return c.body(null, 204) + }) + return app } diff --git a/apps/backend/src/modules/stats/stats.repository.ts b/apps/backend/src/modules/stats/stats.repository.ts new file mode 100644 index 0000000..6b6dceb --- /dev/null +++ b/apps/backend/src/modules/stats/stats.repository.ts @@ -0,0 +1,73 @@ +import type { AppDatabase } from '../../database/connection' +import { count, eq, gte, sql, sum } from 'drizzle-orm' +import { catalogSizeHistory, ownCatalogTitles, peerCatalogTitles } from '../../database/schema' + +export const OWN_CATALOG_SOURCE_ID = 'self' + +export type CatalogSizePoint = typeof catalogSizeHistory.$inferSelect + +/** History the dashboard can't rebuild from the download and upload rows. */ +export class StatsRepository { + constructor(private readonly db: AppDatabase) {} + + /** + * Write today's catalog size for this instance and every peer. Re-running the + * same day overwrites that day's row. `peerNames` maps peer ids to the name to + * show; peers missing from it (no longer configured) are skipped. + */ + recordCatalogSizes(peerNames: Map, day = new Date().toISOString().slice(0, 10)): void { + const own = this.db.select({ + titles: count(), + bytes: sum(ownCatalogTitles.totalSize), + }) + .from(ownCatalogTitles) + .where(eq(ownCatalogTitles.deleted, false)) + .get() + const peers = this.db.select({ + peerId: peerCatalogTitles.peerId, + titles: count(), + bytes: sum(peerCatalogTitles.totalSize), + }) + .from(peerCatalogTitles) + .groupBy(peerCatalogTitles.peerId) + .all() + + const rows = [ + { + day, + sourceId: OWN_CATALOG_SOURCE_ID, + sourceName: 'This instance', + titles: own?.titles ?? 0, + bytes: Number(own?.bytes ?? 0), + }, + ] + for (const peer of peers) { + const name = peerNames.get(peer.peerId) + if (name) { + rows.push({ + day, + sourceId: peer.peerId, + sourceName: name, + titles: peer.titles, + bytes: Number(peer.bytes ?? 0), + }) + } + } + + this.db.insert(catalogSizeHistory) + .values(rows) + .onConflictDoUpdate({ + target: [catalogSizeHistory.day, catalogSizeHistory.sourceId], + set: { + sourceName: sql`excluded.source_name`, + titles: sql`excluded.titles`, + bytes: sql`excluded.bytes`, + }, + }) + .run() + } + + catalogSizesSince(day: string): CatalogSizePoint[] { + return this.db.select().from(catalogSizeHistory).where(gte(catalogSizeHistory.day, day)).orderBy(catalogSizeHistory.day).all() + } +} diff --git a/apps/backend/src/modules/stats/stats.ts b/apps/backend/src/modules/stats/stats.ts new file mode 100644 index 0000000..6dc7865 --- /dev/null +++ b/apps/backend/src/modules/stats/stats.ts @@ -0,0 +1,188 @@ +import type { DownloadRecord } from '../downloads/downloads.repository' +import type { UploadRecord } from '../uploads/uploads.repository' +import type { CatalogSizePoint } from './stats.repository' + +const DAY_MS = 86_400_000 +const TOP_TITLES = 10 + +export interface DailyTransfer { + day: string + downloadedBytes: number + uploadedBytes: number + /** Average over that day's finished transfers: their bytes over their wall time. */ + downloadBps: number | null + uploadBps: number | null +} + +export interface DashboardStats { + /** UTC days, oldest first; every series below has one entry per day. */ + days: string[] + transfers: DailyTransfer[] + importLatency: { day: string, avgSeconds: number | null, count: number }[] + /** Lifetime bytes per peer, largest first. */ + peers: { + gaveToMe: { name: string, bytes: number }[] + tookFromMe: { name: string, bytes: number }[] + } + /** Downloads that are failed right now, by error code, most common first. */ + failures: { code: string, count: number }[] + /** Files peers pulled most from this instance, by bytes sent. */ + topTitles: { filename: string, bytes: number, requests: number, peers: number }[] + catalog: { + sources: { id: string, name: string }[] + points: { day: string, sourceId: string, titles: number, bytes: number }[] + } +} + +function dayOf(iso: string): string { + return iso.slice(0, 10) +} + +function durationMs(from: string, to: string): number { + return Math.max(0, Date.parse(to) - Date.parse(from)) +} + +function ranked(totals: Map): { name: string, bytes: number }[] { + return [...totals].map(([name, bytes]) => ({ name, bytes })).filter(p => p.bytes > 0).sort((a, b) => b.bytes - a.bytes) +} + +function add(map: Map, key: string, value: number): void { + map.set(key, (map.get(key) ?? 0) + value) +} + +/** Everything the dashboard charts need, computed from the stored rows. */ +export function buildStats( + downloads: DownloadRecord[], + uploads: UploadRecord[], + catalogPoints: CatalogSizePoint[], + dayCount: number, + now = Date.now(), +): DashboardStats { + const days = Array.from({ length: dayCount }, (_, i) => dayOf(new Date(now - (dayCount - 1 - i) * DAY_MS).toISOString())) + const inRange = new Set(days) + const buckets = new Map(days.map(day => [ + day, + { + down: 0, + up: 0, + downTimed: 0, + downMs: 0, + upTimed: 0, + upMs: 0, + latencyMs: 0, + latencyCount: 0, + }, + ])) + + const gaveToMe = new Map() + const failures = new Map() + for (const d of downloads) { + add(gaveToMe, d.peerName, d.downloadedBytes) + if (d.status === 'failed') + add(failures, d.errorCode ?? 'UNKNOWN_ERROR', 1) + if (!d.completedAt) + continue + const bucket = buckets.get(dayOf(d.completedAt)) + if (bucket) { + bucket.down += d.downloadedBytes + bucket.downTimed += d.downloadedBytes + bucket.downMs += durationMs(d.startedAt, d.completedAt) + } + if (!d.importedAt) + continue + const imported = buckets.get(dayOf(d.importedAt)) + if (imported) { + imported.latencyMs += durationMs(d.completedAt, d.importedAt) + imported.latencyCount++ + } + } + + const tookFromMe = new Map() + const titles = new Map }>() + for (const u of uploads) { + add(tookFromMe, u.peerName, u.sentBytes) + const title = titles.get(u.filename) ?? { + bytes: 0, + requests: 0, + peers: new Set(), + } + title.bytes += u.sentBytes + title.requests++ + title.peers.add(u.peerName) + titles.set(u.filename, title) + if (!u.finishedAt) + continue + const bucket = buckets.get(dayOf(u.finishedAt)) + if (!bucket) + continue + // Bytes and time come from the downloading jack's receipt, so only confirmed + // uploads count; files go out zero-copy and jack never measures them itself. + bucket.up += u.sentBytes + if (u.status === 'completed' && u.durationMs) { + bucket.upTimed += u.sentBytes + bucket.upMs += u.durationMs + } + } + + const sources = new Map() + const points = [] + for (const p of catalogPoints) { + if (!inRange.has(p.day)) + continue + // The newest name wins, so a renamed peer shows its current name. + sources.set(p.sourceId, p.sourceName) + points.push({ + day: p.day, + sourceId: p.sourceId, + titles: p.titles, + bytes: p.bytes, + }) + } + + return { + days, + transfers: days.map((day) => { + const b = buckets.get(day)! + return { + day, + downloadedBytes: b.down, + uploadedBytes: b.up, + downloadBps: b.downMs > 0 + ? (b.downTimed * 1000) / b.downMs + : null, + uploadBps: b.upMs > 0 + ? (b.upTimed * 1000) / b.upMs + : null, + } + }), + importLatency: days.map((day) => { + const b = buckets.get(day)! + return { + day, + avgSeconds: b.latencyCount > 0 + ? Math.round(b.latencyMs / b.latencyCount / 1000) + : null, + count: b.latencyCount, + } + }), + peers: { + gaveToMe: ranked(gaveToMe), + tookFromMe: ranked(tookFromMe), + }, + failures: [...failures].map(([code, count]) => ({ code, count })).sort((a, b) => b.count - a.count), + topTitles: [...titles] + .map(([filename, t]) => ({ + filename, + bytes: t.bytes, + requests: t.requests, + peers: t.peers.size, + })) + .filter(t => t.bytes > 0) + .sort((a, b) => b.bytes - a.bytes) + .slice(0, TOP_TITLES), + catalog: { + sources: [...sources].map(([id, name]) => ({ id, name })), + points, + }, + } +} diff --git a/apps/backend/src/modules/status/status.controller.ts b/apps/backend/src/modules/status/status.controller.ts index 2c3cdbe..4c3ac79 100644 --- a/apps/backend/src/modules/status/status.controller.ts +++ b/apps/backend/src/modules/status/status.controller.ts @@ -1,9 +1,15 @@ -import type { DownloadStatus } from '../../database/schema' +import type { DownloadStatus, UploadStatus } from '../../database/schema' import type { ArrServerConnector } from '../../lib/servers/arr/base' import type { ServerConnector } from '../../lib/servers/base' import type { PeerConnector } from '../../lib/servers/peer' import type { DownloadRecord, DownloadsRepository } from '../downloads/downloads.repository' -import { DOWNLOAD_STATUSES } from '../../database/schema' +import type { LiveProgressTracker } from '../downloads/live-progress' +import type { StatsRepository } from '../stats/stats.repository' +import type { UploadRecord } from '../uploads/uploads.repository' +import type { UploadsService } from '../uploads/uploads.service' +import { DOWNLOAD_STATUSES, UPLOAD_STATUSES } from '../../database/schema' +import { liveProgressFor } from '../downloads/live-progress' +import { buildStats } from '../stats/stats' // The management surface can expose richer detail than the public API: every // connector here carries its URL-derived `id` (the management UI keys edit/delete @@ -25,27 +31,69 @@ function baseConnector(c: ServerConnector) { } } -export function enrichDownload(d: DownloadRecord) { +export function enrichDownload(d: DownloadRecord, live?: LiveProgressTracker) { // Prefer the negotiated expected size; fall back to the release-advertised size. const totalBytes = d.expectedBytes ?? (d.releaseSize > 0 ? d.releaseSize : null) + // A running transfer reports bytes about once a second in memory; the row is only + // written every 10 s / 64 MiB, so prefer the live count while it's downloading. + const current = d.status === 'downloading' + ? live?.get(d.id) + : null + const downloadedBytes = Math.max(d.downloadedBytes, current?.downloadedBytes ?? 0) const progress = totalBytes && totalBytes > 0 - ? Math.min(1, d.downloadedBytes / totalBytes) + ? Math.min(1, downloadedBytes / totalBytes) + : null + const speedBps = current?.speedBps ?? null + const etaSeconds = speedBps && speedBps > 0 && totalBytes + ? Math.max(0, Math.round((totalBytes - downloadedBytes) / speedBps)) : null - return { ...d, totalBytes, progress } + return { + ...d, + downloadedBytes, + totalBytes, + progress, + speedBps, + etaSeconds, + } +} + +export function enrichUpload(u: UploadRecord) { + // Files go out zero-copy, so only a receipt tells us how far it got. Progress is + // through the whole file, so a resumed request starts where the range does. + const progress = u.status === 'completed' && u.totalBytes > 0 + ? Math.min(1, (u.rangeStart + u.sentBytes) / u.totalBytes) + : null + // Average over the whole transfer, from the receipt; unknown without one. + const speedBps = u.durationMs && u.durationMs > 0 + ? (u.sentBytes * 1000) / u.durationMs + : null + return { ...u, progress, speedBps } } export class StatusController { constructor( private readonly connectors: { servers: ArrServerConnector[], peers: PeerConnector[], subscribe?: (subscriber: () => void) => () => void }, private readonly downloads?: DownloadsRepository, + private readonly uploads?: UploadsService, + private readonly stats?: StatsRepository, ) {} + private enrichUpload = (u: UploadRecord) => enrichUpload(u) + + private get live(): LiveProgressTracker | undefined { + return this.downloads && liveProgressFor(this.downloads) + } + + private enrich = (d: DownloadRecord) => enrichDownload(d, this.live) + subscribeOverview(subscriber: () => void): () => void { const unsubscribers = [ this.connectors.subscribe?.(subscriber), this.downloads?.subscribe(subscriber), + this.live?.subscribe(subscriber), + this.uploads?.subscribe(subscriber), ].filter((unsubscribe): unsubscribe is () => void => Boolean(unsubscribe)) return () => { for (const unsubscribe of unsubscribers) @@ -53,8 +101,19 @@ export class StatusController { } } + subscribeUploads(subscriber: () => void): () => void { + return this.uploads?.subscribe(subscriber) ?? (() => {}) + } + subscribeDownloads(subscriber: () => void): () => void { - return this.downloads?.subscribe(subscriber) ?? (() => {}) + const unsubscribers = [ + this.downloads?.subscribe(subscriber), + this.live?.subscribe(subscriber), + ].filter((unsubscribe): unsubscribe is () => void => Boolean(unsubscribe)) + return () => { + for (const unsubscribe of unsubscribers) + unsubscribe() + } } /** Whether download-backed endpoints are available (a repository was injected). */ @@ -107,17 +166,52 @@ export class StatusController { bytesMoved, mismatched, // `records` is sorted by updatedAt desc, so each slice is the most recent. - active: records.filter(r => r.status === 'downloading').map(enrichDownload), - importQueued: records.filter(r => r.status === 'import_queued').slice(0, ATTENTION_LIMIT).map(enrichDownload), - failed: records.filter(r => r.status === 'failed').slice(0, ATTENTION_LIMIT).map(enrichDownload), + active: records.filter(r => r.status === 'downloading').map(this.enrich), + importQueued: records.filter(r => r.status === 'import_queued').slice(0, ATTENTION_LIMIT).map(this.enrich), + failed: records.filter(r => r.status === 'failed').slice(0, ATTENTION_LIMIT).map(this.enrich), }, + uploads: this.getUploadsSummary(), + } + } + + private getUploadsSummary() { + const records = this.uploads?.list() ?? [] + const byStatus = Object.fromEntries(UPLOAD_STATUSES.map(s => [s, 0])) as Record + let bytesSent = 0 + for (const record of records) { + byStatus[record.status]++ + bytesSent += record.sentBytes + } + return { + total: records.length, + byStatus, + bytesSent, + active: records.filter(r => r.status === 'uploading').map(this.enrichUpload), + } + } + + /** Chart data for the dashboard: the last `days` UTC days plus lifetime rankings. */ + getStats(days: number) { + const since = new Date(Date.now() - (days - 1) * 86_400_000).toISOString().slice(0, 10) + return buildStats( + this.downloads?.list() ?? [], + this.uploads?.list() ?? [], + this.stats?.catalogSizesSince(since) ?? [], + days, + ) + } + + listUploads() { + const records = this.uploads?.list() ?? [] + return { + uploads: records.map(this.enrichUpload), } } listDownloads() { const records = this.downloads?.list() ?? [] return { - downloads: records.map(enrichDownload), + downloads: records.map(this.enrich), } } } diff --git a/apps/backend/src/modules/status/status.router.ts b/apps/backend/src/modules/status/status.router.ts index 6c9e5d9..fd5309d 100644 --- a/apps/backend/src/modules/status/status.router.ts +++ b/apps/backend/src/modules/status/status.router.ts @@ -62,6 +62,55 @@ export function getStatusRouter(controller: StatusController) { }, }, }), c => streamSnapshots(c, () => controller.listDownloads(), subscriber => controller.subscribeDownloads(subscriber))) + app.get('/uploads', describeRoute({ + tags: ['Uploads'], + summary: 'List uploads', + description: 'Files peers pulled or are pulling from this instance, newest first. One record per file request, so a resumed transfer adds a record that starts at its `rangeStart`.', + security: [ + { 'X-Management-Key': [] }, + ], + responses: { + 200: { + description: 'Upload records', + content: { 'application/json': {} }, + }, + }, + }), c => c.json(controller.listUploads())) + app.get('/uploads/stream', describeRoute({ + tags: ['Uploads'], + summary: 'Stream uploads', + description: 'Server-Sent Events stream with an initial upload snapshot and an update after every upload change, at most about once a second while files stream.', + security: [ + { 'X-Management-Key': [] }, + ], + responses: { + 200: { + description: 'Live upload snapshots', + content: { 'text/event-stream': {} }, + }, + }, + }), c => streamSnapshots(c, () => controller.listUploads(), subscriber => controller.subscribeUploads(subscriber))) + + app.get('/stats', describeRoute({ + tags: ['Status'], + summary: 'Dashboard statistics', + description: 'Per-day transfer volume, average speed and import latency for the last `days` UTC days (default 30, at most 365), lifetime bytes per peer, current failures by error code, the files peers pulled most, and the catalog size history.', + security: [ + { 'X-Management-Key': [] }, + ], + responses: { + 200: { + description: 'Dashboard statistics', + content: { 'application/json': {} }, + }, + }, + }), (c) => { + const requested = Number(c.req.query('days') ?? 30) + const days = Number.isInteger(requested) + ? Math.min(365, Math.max(1, requested)) + : 30 + return c.json(controller.getStats(days)) + }) return app } diff --git a/apps/backend/src/modules/uploads/uploads.repository.ts b/apps/backend/src/modules/uploads/uploads.repository.ts new file mode 100644 index 0000000..93df261 --- /dev/null +++ b/apps/backend/src/modules/uploads/uploads.repository.ts @@ -0,0 +1,146 @@ +import type { AppDatabase } from '../../database/connection' +import type { NewUploadRow, UploadRow } from '../../database/schema' +import { and, desc, eq, inArray, lt } from 'drizzle-orm' +import { uploads } from '../../database/schema' + +export type UploadRecord = UploadRow + +export interface CreateUploadInput { + peerName: string + itemId: string + filename: string + totalBytes: number + rangeStart: number + requestedBytes: number +} + +// Statuses an upload can still leave: a receipt or a disconnect may yet arrive. +const OPEN = ['uploading', 'unconfirmed'] as const + +function nowIso() { + return new Date().toISOString() +} + +/** + * Files peers pull from this instance, one row per request. Rows are kept after + * they finish so the list doubles as a history. + */ +export class UploadsRepository { + private readonly subscribers = new Set<() => void>() + + constructor(private readonly db: AppDatabase) {} + + subscribe(subscriber: () => void): () => void { + this.subscribers.add(subscriber) + return () => this.subscribers.delete(subscriber) + } + + private notifyChanged(): void { + for (const subscriber of this.subscribers) { + try { + subscriber() + } + catch { + // Observers must never make a committed database mutation look failed. + } + } + } + + /** `uploading` when the peer promised a receipt, `unconfirmed` when it can't send one. */ + create(input: CreateUploadInput, status: 'uploading' | 'unconfirmed'): UploadRecord { + const timestamp = nowIso() + const values: NewUploadRow = { + ...input, + sentBytes: 0, + status, + startedAt: timestamp, + updatedAt: timestamp, + } + const row = this.db.insert(uploads).values(values).returning().get() + this.notifyChanged() + return row + } + + get(id: number): UploadRecord | null { + return this.db.select().from(uploads).where(eq(uploads.id, id)).get() ?? null + } + + /** Newest first; the id breaks ties between rows started in the same millisecond. */ + list(): UploadRecord[] { + return this.db.select().from(uploads).orderBy(desc(uploads.startedAt), desc(uploads.id)).all() + } + + /** The peer disconnected before the body was sent. No-op once the row has ended. */ + markAborted(id: number, error: string): void { + const timestamp = nowIso() + const changed = this.db.update(uploads) + .set({ + status: 'aborted', + error, + updatedAt: timestamp, + finishedAt: timestamp, + }) + .where(and(eq(uploads.id, id), inArray(uploads.status, [...OPEN]))) + .returning({ id: uploads.id }) + .all() + if (changed.length > 0) + this.notifyChanged() + } + + /** + * The downloading peer's receipt: it got `sentBytes` in `durationMs`. Only the + * peer the row belongs to can close it, and only while it is still open. + * Returns whether the row was updated. + */ + complete(id: number, peerName: string, sentBytes: number, durationMs: number): boolean { + const timestamp = nowIso() + const changed = this.db.update(uploads) + .set({ + status: 'completed', + sentBytes, + durationMs, + error: null, + updatedAt: timestamp, + finishedAt: timestamp, + }) + .where(and(eq(uploads.id, id), eq(uploads.peerName, peerName), inArray(uploads.status, [...OPEN]))) + .returning({ id: uploads.id }) + .all() + if (changed.length > 0) + this.notifyChanged() + return changed.length > 0 + } + + /** Rows still waiting for a receipt that started before `before`: it isn't coming. */ + expireUnconfirmed(before: string): number { + const rows = this.db.update(uploads) + .set({ + status: 'unconfirmed', + updatedAt: nowIso(), + }) + .where(and(eq(uploads.status, 'uploading'), lt(uploads.startedAt, before))) + .returning({ id: uploads.id }) + .all() + if (rows.length > 0) + this.notifyChanged() + return rows.length + } + + /** Rows still `uploading` from a previous run lost their connection with it. */ + reconcileInterrupted(): number { + const timestamp = nowIso() + const rows = this.db.update(uploads) + .set({ + status: 'aborted', + error: 'interrupted by a jack restart', + updatedAt: timestamp, + finishedAt: timestamp, + }) + .where(eq(uploads.status, 'uploading')) + .returning({ id: uploads.id }) + .all() + if (rows.length > 0) + this.notifyChanged() + return rows.length + } +} diff --git a/apps/backend/src/modules/uploads/uploads.service.ts b/apps/backend/src/modules/uploads/uploads.service.ts new file mode 100644 index 0000000..8d27caf --- /dev/null +++ b/apps/backend/src/modules/uploads/uploads.service.ts @@ -0,0 +1,42 @@ +import type { CreateUploadInput, UploadRecord, UploadsRepository } from './uploads.repository' + +// A receipt that hasn't come after this long won't: the row goes `unconfirmed`. +const RECEIPT_WAIT_MS = 48 * 60 * 60 * 1000 + +/** + * Records files peers pull from `/peer/items/:id/file`. The file itself is sent + * zero-copy, untouched; this only keeps the row. A disconnect marks it aborted, + * and the downloading jack's receipt (`/peer/uploads/:id/receipt`) completes it + * with the bytes and time the transfer took. + */ +export class UploadsService { + constructor(readonly repository: UploadsRepository) {} + + /** + * Starts a row for this request and returns its id, for the `X-Jack-Upload-Id` + * response header. `expectsReceipt` is whether the peer said it will report back. + */ + start(input: CreateUploadInput, expectsReceipt: boolean, signal?: AbortSignal): number { + const upload = this.repository.create(input, expectsReceipt ? 'uploading' : 'unconfirmed') + // Bun aborts the request signal only when the peer disconnects before the + // whole body went out; a finished response never fires it. + signal?.addEventListener('abort', () => this.repository.markAborted(upload.id, 'peer disconnected'), { once: true }) + return upload.id + } + + receipt(id: number, peerName: string, sentBytes: number, durationMs: number): boolean { + return this.repository.complete(id, peerName, sentBytes, durationMs) + } + + expireUnconfirmed(now = Date.now()): number { + return this.repository.expireUnconfirmed(new Date(now - RECEIPT_WAIT_MS).toISOString()) + } + + list(): UploadRecord[] { + return this.repository.list() + } + + subscribe(subscriber: () => void): () => void { + return this.repository.subscribe(subscriber) + } +} diff --git a/apps/ui/Dockerfile b/apps/ui/Dockerfile index 98dbea2..254bf88 100644 --- a/apps/ui/Dockerfile +++ b/apps/ui/Dockerfile @@ -28,6 +28,9 @@ CMD ["bun", "run", "--cwd", "apps/ui", "dev", "--host", "0.0.0.0", "--port", "30 # ---- build: the Nuxt/Nitro server ---- FROM development AS build +# The version shown in the UI (CI passes the image version). Unset falls back to +# `dev`, since the build context has no .git to read tags from. +ARG APP_VERSION RUN bun run --cwd apps/ui build # ---- runtime: the self-contained Nitro output ---- diff --git a/apps/ui/app/assets/css/main.css b/apps/ui/app/assets/css/main.css index 7286ec8..1873e90 100644 --- a/apps/ui/app/assets/css/main.css +++ b/apps/ui/app/assets/css/main.css @@ -7,3 +7,28 @@ with a system-mono fallback so an offline build still renders cleanly. */ --font-mono: 'JetBrains Mono', ui-monospace, 'SFMono-Regular', 'Menlo', monospace; } + +/* Dashboard chart series, in fixed order: a series keeps its slot, never its rank. + The order is CVD-checked for adjacent pairs; dark mode has its own steps. + Downloads are slot 1 and uploads slot 2 everywhere. */ +:root { + --viz-1: #2a78d6; + --viz-2: #eb6834; + --viz-3: #1baf7a; + --viz-4: #eda100; + --viz-5: #e87ba4; + --viz-6: #008300; + --viz-7: #6250d6; + --viz-8: #e34948; +} + +.dark { + --viz-1: #3987e5; + --viz-2: #d95926; + --viz-3: #199e70; + --viz-4: #c98500; + --viz-5: #d55181; + --viz-6: #008300; + --viz-7: #9085e9; + --viz-8: #e66767; +} diff --git a/apps/ui/app/components/ColumnChart.vue b/apps/ui/app/components/ColumnChart.vue new file mode 100644 index 0000000..0ab9883 --- /dev/null +++ b/apps/ui/app/components/ColumnChart.vue @@ -0,0 +1,89 @@ + + + diff --git a/apps/ui/app/components/DashSection.vue b/apps/ui/app/components/DashSection.vue new file mode 100644 index 0000000..4b13cd2 --- /dev/null +++ b/apps/ui/app/components/DashSection.vue @@ -0,0 +1,26 @@ + + + diff --git a/apps/ui/app/components/DownloadsSection.vue b/apps/ui/app/components/DownloadsSection.vue index 7c178c7..140e574 100644 --- a/apps/ui/app/components/DownloadsSection.vue +++ b/apps/ui/app/components/DownloadsSection.vue @@ -27,6 +27,9 @@ function blankState() { manualImportBackoffBaseMs: null as number | null, manualImportBackoffMaxMs: null as number | null, unlinkImportedFiles: false, + removeFailedDownloads: false, + failedDownloadRetries: null as number | null, + failedRetryBaseDelayMs: null as number | null, } } @@ -99,6 +102,9 @@ async function save() { manualImportBackoffMaxMs: state.manualImportBackoffMaxMs, // Off is the default, so write it as an absent key rather than an explicit false. unlinkImportedFiles: state.unlinkImportedFiles ? true : null, + removeFailedDownloads: state.removeFailedDownloads ? true : null, + failedDownloadRetries: state.failedDownloadRetries, + failedRetryBaseDelayMs: state.failedRetryBaseDelayMs, } const creating = !downloads.value try { @@ -133,6 +139,8 @@ const transferKeys = [ 'idleTimeoutMs', 'retryBaseDelayMs', 'retryMaxDelayMs', + 'failedDownloadRetries', + 'failedRetryBaseDelayMs', ] as const const importKeys = [ 'importPollIntervalMs', @@ -151,6 +159,8 @@ const DEFAULTS: Record = { maxManualImportAttempts: 6, manualImportBackoffBaseMs: 60_000, manualImportBackoffMaxMs: 1_800_000, + failedDownloadRetries: 3, + failedRetryBaseDelayMs: 60_000, } function customCount(keys: readonly string[]) { @@ -218,6 +228,14 @@ const importOpen = ref(false) > + + + + +
+ {{ selected.length }} selected + + +
+