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..7887b20 --- /dev/null +++ b/apps/backend/drizzle/0014_uploads.sql @@ -0,0 +1,21 @@ +CREATE TABLE `uploads` ( + `id` integer PRIMARY KEY AUTOINCREMENT NOT NULL, + `peer_name` text NOT NULL, + `peer_key_id` integer, + `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_upload_transfers.sql b/apps/backend/drizzle/0015_upload_transfers.sql new file mode 100644 index 0000000..1393099 --- /dev/null +++ b/apps/backend/drizzle/0015_upload_transfers.sql @@ -0,0 +1,29 @@ +PRAGMA foreign_keys=OFF;--> statement-breakpoint +CREATE TABLE `__new_uploads` ( + `id` integer PRIMARY KEY AUTOINCREMENT NOT NULL, + `transfer_id` text NOT NULL, + `idempotency_key` text NOT NULL, + `peer_name` text NOT NULL, + `peer_key_id` integer, + `item_id` text NOT NULL, + `filename` text NOT NULL, + `total_bytes` integer NOT NULL, + `range_start` integer, + `sent_bytes` integer DEFAULT 0 NOT NULL, + `duration_ms` integer, + `status` text NOT NULL, + `started_at` text NOT NULL, + `updated_at` text NOT NULL, + `finished_at` text, + CONSTRAINT "uploads_status_check" CHECK("__new_uploads"."status" in ('uploading', 'completed', 'unconfirmed')) +); +--> statement-breakpoint +INSERT INTO `__new_uploads`("id", "transfer_id", "idempotency_key", "peer_name", "peer_key_id", "item_id", "filename", "total_bytes", "range_start", "sent_bytes", "duration_ms", "status", "started_at", "updated_at", "finished_at") SELECT "id", lower(hex(randomblob(16))), 'legacy:' || "id", "peer_name", "peer_key_id", "item_id", "filename", "total_bytes", "range_start", "sent_bytes", "duration_ms", CASE WHEN "status" = 'aborted' THEN 'unconfirmed' ELSE "status" END, "started_at", "updated_at", "finished_at" FROM `uploads`;--> statement-breakpoint +DROP TABLE `uploads`;--> statement-breakpoint +ALTER TABLE `__new_uploads` RENAME TO `uploads`;--> statement-breakpoint +PRAGMA foreign_keys=ON;--> statement-breakpoint +CREATE UNIQUE INDEX `uploads_transfer_id_unique` ON `uploads` (`transfer_id`);--> statement-breakpoint +CREATE UNIQUE INDEX `uploads_idempotency_key_unique` ON `uploads` (`idempotency_key`);--> statement-breakpoint +CREATE INDEX `uploads_status_idx` ON `uploads` (`status`);--> statement-breakpoint +CREATE INDEX `uploads_started_at_idx` ON `uploads` (`started_at`);--> statement-breakpoint +ALTER TABLE `downloads` ADD `speed_bps` integer; \ 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..5ca2f7c --- /dev/null +++ b/apps/backend/drizzle/meta/0014_snapshot.json @@ -0,0 +1,1061 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "c9f970ae-d2a3-41bd-b206-35a480a193bf", + "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 + }, + "peer_key_id": { + "name": "peer_key_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "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..8f4f4b3 --- /dev/null +++ b/apps/backend/drizzle/meta/0015_snapshot.json @@ -0,0 +1,1081 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "14e99ced-c742-4437-9940-60a0a5a6863a", + "prevId": "c9f970ae-d2a3-41bd-b206-35a480a193bf", + "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 + }, + "speed_bps": { + "name": "speed_bps", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "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 + }, + "transfer_id": { + "name": "transfer_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "idempotency_key": { + "name": "idempotency_key", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "peer_name": { + "name": "peer_name", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "peer_key_id": { + "name": "peer_key_id", + "type": "integer", + "primaryKey": false, + "notNull": false, + "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": false, + "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 + }, + "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_transfer_id_unique": { + "name": "uploads_transfer_id_unique", + "columns": [ + "transfer_id" + ], + "isUnique": true + }, + "uploads_idempotency_key_unique": { + "name": "uploads_idempotency_key_unique", + "columns": [ + "idempotency_key" + ], + "isUnique": true + }, + "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', '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..a891d91 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": 1794180331966, + "tag": "0014_uploads", + "breakpoints": true + }, + { + "idx": 15, + "version": "6", + "when": 1794270647120, + "tag": "0015_upload_transfers", + "breakpoints": true } ] } \ No newline at end of file diff --git a/apps/backend/src/__tests__/downloads-repository.test.ts b/apps/backend/src/__tests__/downloads-repository.test.ts index cc7afe5..1045b6d 100644 --- a/apps/backend/src/__tests__/downloads-repository.test.ts +++ b/apps/backend/src/__tests__/downloads-repository.test.ts @@ -39,6 +39,28 @@ function seed(repo: DownloadsRepository, peerId: string, filename: string) { }) } +describe('DownloadsRepository.summary', () => { + test('counts by status and sums bytes in SQL', () => { + const repo = makeRepo() + const a = seed(repo, 'p1', 'A.mkv') + seed(repo, 'p1', 'B.mkv') + repo.updateProgress(a.id, 40) + repo.markFailed(a.id, 'PEER_UNREACHABLE', 'nope') + expect(repo.summary()).toMatchObject({ + byStatus: { + downloading: 1, + failed: 1, + import_queued: 0, + imported: 0, + }, + total: 2, + bytesMoved: 40, + mismatched: 0, + }) + expect(repo.listByStatus('failed', 1).map(r => r.id)).toEqual([a.id]) + }) +}) + describe('DownloadsRepository.reassignPeerId', () => { test('moves only the matching rows', () => { const repo = makeRepo() 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..9a8c169 --- /dev/null +++ b/apps/backend/src/__tests__/failed-download-reaper.test.ts @@ -0,0 +1,165 @@ +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], + get: id => rows.find(row => row.id === id) ?? null, + retry: async (id) => { + calls.push(`retry:${id}`) + if (options.retryFails) + throw new Error('nope') + }, + remove: async (id, onlyIf) => { + const index = rows.findIndex(row => row.id === id) + if (!onlyIf(rows[index]!)) + return + calls.push(`remove:${id}`) + rows.splice(index, 1) + }, + maxRetries: 3, + baseDelayMs: 100, + maxDelayMs: 250, + now: () => clock.now, + }) + return { reaper, clock, calls } +} + +test('stops mid-pass when removal is switched off', async () => { + const rows = [ + failedRow({ id: 1, lastOperation: 'import' }), + failedRow({ id: 2, lastOperation: 'import' }), + ] + const switches = { on: true } + const calls: string[] = [] + const reaper = new FailedDownloadReaper({ + enabled: () => switches.on, + list: () => [...rows], + get: id => rows.find(row => row.id === id) ?? null, + retry: async () => {}, + remove: async (id, onlyIf) => { + if (!onlyIf(rows.find(row => row.id === id)!)) + return + calls.push(`remove:${id}`) + // The user turns removal off while this deletion runs. + switches.on = false + }, + maxRetries: 3, + baseDelayMs: 100, + maxDelayMs: 250, + }) + await reaper.tick() + expect(calls).toEqual(['remove:1']) +}) + +test('re-reads each row, so one the user cancelled mid-pass is left alone', async () => { + const second = failedRow({ id: 2, lastOperation: 'import' }) + const rows = [ + failedRow({ id: 1, lastOperation: 'import' }), + second, + ] + const { reaper, calls } = setup(rows) + // The pass lists both rows, then the user cancels row 2 before the reaper gets to it. + const original = rows[0]! + Object.defineProperty(original, 'status', { + get() { + second.errorCode = 'CANCELLED' + return 'failed' + }, + }) + await reaper.tick() + expect(calls).toEqual(['remove:1']) +}) + +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__/peer-download.test.ts b/apps/backend/src/__tests__/peer-download.test.ts index 740d967..dc21956 100644 --- a/apps/backend/src/__tests__/peer-download.test.ts +++ b/apps/backend/src/__tests__/peer-download.test.ts @@ -151,6 +151,84 @@ describe('PeerConnector.downloadFile', () => { } }) + test('opens a transfer on a 0.3.0 peer, streams it, and posts the receipt', async () => { + const seen: unknown[] = [] + server.use( + http.post(`${PEER_JACK_URL}/peer/items/:itemId/downloads`, async ({ request, params }) => { + seen.push({ create: params.itemId, body: await request.json() }) + return HttpResponse.json({ + downloadId: 'abc', + filename: 'Movie.mkv', + totalBytes: 4, + // Ignored: the file URL is built from the downloadId on the peer's own host. + url: 'http://internal.test/admin', + }, { status: 201 }) + }), + http.get(`${PEER_JACK_URL}/peer/downloads/:downloadId/file`, ({ params }) => { + seen.push({ stream: params.downloadId }) + return new Response(streamOf([1, 2, 3, 4]), { + headers: { 'Content-Length': '4' }, + }) + }), + http.post(`${PEER_JACK_URL}/peer/downloads/:downloadId/receipt`, ({ request, params }) => { + seen.push({ + receipt: params.downloadId, + apiKey: request.headers.get('X-Api-Key'), + }) + return new HttpResponse(null, { status: 204 }) + }), + ) + + const peer = markInitialized(new PeerConnector({ url: PEER_JACK_URL, apiKey: 'peer-api-key', name: 'Friend Jack' })) + ;(peer as any)._peerVersion = '0.3.0' + const dir = await mkdtemp(join(tmpdir(), 'jack-peer-receipt-')) + try { + await peer.downloadFile('remote1:movie:99', join(dir, 'Movie.mkv'), { releaseSize: 4, transferId: 'download-1-0' }) + // Fire and forget: the download resolves without waiting for it. + await waitFor(async () => seen.length === 3) + expect(seen).toEqual([ + { + create: 'remote1:movie:99', + body: { clientTransferId: 'download-1-0' }, + }, + { stream: 'abc' }, + { receipt: 'abc', apiKey: 'peer-api-key' }, + ]) + } + finally { + await rm(dir, { recursive: true, force: true }) + } + }) + + test('uses the untracked item route for an older peer or a failed create', async () => { + const creates: unknown[] = [] + server.use( + http.post(`${PEER_JACK_URL}/peer/items/:itemId/downloads`, () => { + creates.push(1) + return HttpResponse.json({ error: 'database or disk is full' }, { status: 500 }) + }), + http.get(`${PEER_JACK_URL}/peer/items/:itemId/file`, () => new Response(streamOf([1, 2, 3, 4]), { + headers: { 'Content-Length': '4' }, + })), + ) + + const dir = await mkdtemp(join(tmpdir(), 'jack-peer-legacy-')) + try { + for (const version of ['0.2.0', '0.3.0']) { + const peer = markInitialized(new PeerConnector({ url: PEER_JACK_URL, apiKey: 'peer-api-key', name: 'Friend Jack' })) + ;(peer as any)._peerVersion = version + const destPath = join(dir, `Movie-${version}.mkv`) + await peer.downloadFile('remote1:movie:99', destPath, { releaseSize: 4, transferId: 'download-1-0' }) + expect(new Uint8Array(await Bun.file(destPath).arrayBuffer())).toEqual(new Uint8Array([1, 2, 3, 4])) + } + // Only the 0.3.0 peer was asked for a transfer. + expect(creates).toHaveLength(1) + } + 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..b8bd474 100644 --- a/apps/backend/src/__tests__/qbittorrent-mapper.test.ts +++ b/apps/backend/src/__tests__/qbittorrent-mapper.test.ts @@ -24,6 +24,7 @@ function baseRecord(overrides: Partial = {}): DownloadRecord { expectedBytesSource: null, expectedBytesMismatch: false, downloadedBytes: 0, + speedBps: null, attempts: 0, status: 'downloading', startedAt: '2026-06-06T00:00:00.000Z', diff --git a/apps/backend/src/__tests__/stream-coalesce.test.ts b/apps/backend/src/__tests__/stream-coalesce.test.ts new file mode 100644 index 0000000..f185cbc --- /dev/null +++ b/apps/backend/src/__tests__/stream-coalesce.test.ts @@ -0,0 +1,62 @@ +import { expect, test } from 'bun:test' +import { coalesce, StatusController } from '../modules/status/status.controller' + +test('the overview stream subscribes to connectors through their object', () => { + // Like ConnectorManager: subscribe() relies on `this`. + class Connectors { + servers = [] + peers = [] + private readonly subscribers = new Set<() => void>() + subscribe(subscriber: () => void) { + this.subscribers.add(subscriber) + return () => this.subscribers.delete(subscriber) + } + + emit() { + for (const s of this.subscribers) s() + } + } + const connectors = new Connectors() + const fired: number[] = [] + const unsubscribe = new StatusController(connectors).subscribeOverview(() => fired.push(1)) + connectors.emit() + expect(fired).toHaveLength(1) + unsubscribe() +}) + +function source() { + const subscribers = new Set<() => void>() + return { + emit: () => { + for (const s of subscribers) s() + }, + subscribe: (s: () => void) => { + subscribers.add(s) + return () => subscribers.delete(s) + }, + } +} + +test('folds bursts from every source into one snapshot per interval', async () => { + const a = source() + const b = source() + const fired: number[] = [] + const unsubscribe = coalesce(() => fired.push(Date.now()), [a.subscribe, undefined, b.subscribe], 50) + + a.emit() + expect(fired).toHaveLength(1) + // Within the interval: folded into one trailing snapshot. + b.emit() + a.emit() + b.emit() + expect(fired).toHaveLength(1) + await Bun.sleep(80) + expect(fired).toHaveLength(2) + + // A pending snapshot is dropped on unsubscribe. + a.emit() + unsubscribe() + await Bun.sleep(80) + a.emit() + expect(fired.length).toBeLessThanOrEqual(3) +}) 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..3d392fc --- /dev/null +++ b/apps/backend/src/__tests__/upload-tracking.test.ts @@ -0,0 +1,157 @@ +import type { AuthVariables } from '../middleware/require-auth' +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 { enrichDownload } from '../modules/status/status.controller' +import { UploadsRepository } from '../modules/uploads/uploads.repository' +import { UploadsService } from '../modules/uploads/uploads.service' + +describe('peer transfers', () => { + 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<{ Variables: AuthVariables }>() + // Stands in for require-auth: the `X-Key` header picks the api key row. + app.use(async (c, next) => { + const key = c.req.header('X-Key') + if (key) + c.set('apiKeyId', Number(key)) + await next() + }) + app.route('/peer', getPeerRouter(new PeerController(() => [source as any]), new UploadsService(repository))) + const create = (clientTransferId: string, key = '7') => app.request('/peer/items/remote1:movie:1/downloads', { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Key': key }, + body: JSON.stringify({ clientTransferId }), + }) + return { app, repository, create } + } + + test('one transfer through a dropped request, a resume and the receipt', async () => { + const { app, repository, create } = setup() + const res = await create('download-1') + expect(res.status).toBe(201) + const transfer = await res.json() as { downloadId: string, url: string, totalBytes: number } + expect(transfer).toMatchObject({ filename: 'Movie.mkv', totalBytes: 10, url: `/peer/downloads/${transfer.downloadId}/file` }) + + // A lost 201: creating it again returns the same transfer. + const again = await create('download-1') + expect(again.status).toBe(200) + expect((await again.json() as { downloadId: string }).downloadId).toBe(transfer.downloadId) + + // The first request drops after a few bytes; the transfer stays open. + const first = await app.request(transfer.url, { + headers: { 'X-Key': '7' }, + }) + await first.body?.cancel() + const resumed = await app.request(transfer.url, { + headers: { 'Range': 'bytes=6-', 'X-Key': '7' }, + }) + expect(resumed.status).toBe(206) + expect([ + ...new Uint8Array(await resumed.arrayBuffer()), + ]).toEqual([6, 7, 8, 9]) + expect(repository.list()).toEqual([ + expect.objectContaining({ status: 'uploading', rangeStart: 0, totalBytes: 10 }), + ]) + + const receipt = await app.request(`/peer/downloads/${transfer.downloadId}/receipt`, { + method: 'POST', + headers: { 'X-Key': '7' }, + }) + expect(receipt.status).toBe(204) + expect(repository.list()).toEqual([ + expect.objectContaining({ status: 'completed', sentBytes: 10 }), + ]) + }) + + test('another key can neither stream nor complete a transfer', async () => { + const { app, repository, create } = setup() + const { downloadId, url } = await (await create('download-1')).json() as { downloadId: string, url: string } + expect((await app.request(url, { + headers: { 'X-Key': '8' }, + })).status).toBe(404) + await app.request(url, { + headers: { 'X-Key': '7' }, + }) + expect((await app.request(`/peer/downloads/${downloadId}/receipt`, { + method: 'POST', + headers: { 'X-Key': '8' }, + })).status).toBe(204) + expect(repository.list()[0]?.status).toBe('uploading') + }) + + test('a client transfer id reused for another item is a conflict', async () => { + const { app, create } = setup() + await create('download-1') + const res = await app.request('/peer/items/remote1:movie:2/downloads', { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Key': '7' }, + body: JSON.stringify({ clientTransferId: 'download-1' }), + }) + expect(res.status).toBe(409) + }) + + test('the item file route serves without recording, and a HEAD serves nothing', async () => { + const { app, repository, create } = setup() + const legacy = await app.request('/peer/items/remote1:movie:1/file', { + headers: { Range: 'bytes=6-' }, + }) + expect([ + ...new Uint8Array(await legacy.arrayBuffer()), + ]).toEqual([6, 7, 8, 9]) + expect(repository.list()).toEqual([]) + + const { url } = await (await create('download-1')).json() as { url: string } + await app.request(url, { + method: 'HEAD', + headers: { 'X-Key': '7' }, + }) + expect(repository.list()[0]?.rangeStart).toBeNull() + }) + + test('a malformed create is rejected', async () => { + const { create } = setup() + expect((await create('')).status).toBe(400) + }) +}) + +test('download speed and ETA come from the row, only while bytes still arrive', () => { + const now = Date.parse('2026-11-08T12:00:00Z') + const row = { + status: 'downloading', + expectedBytes: 10_000, + releaseSize: 10_000, + downloadedBytes: 2000, + speedBps: 2000, + updatedAt: '2026-11-08T11:59:50Z', + } as Parameters[0] + expect(enrichDownload(row, now)).toMatchObject({ progress: 0.2, speedBps: 2000, etaSeconds: 4 }) + expect(enrichDownload({ ...row, status: 'import_queued' }, now)).toMatchObject({ speedBps: null, etaSeconds: null }) + // Nothing written for a minute: stalled, so no speed. + expect(enrichDownload({ ...row, updatedAt: '2026-11-08T11:59:00Z' }, now)).toMatchObject({ speedBps: null, etaSeconds: null }) +}) 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..054b152 --- /dev/null +++ b/apps/backend/src/__tests__/uploads-repository.test.ts @@ -0,0 +1,148 @@ +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', + peerKeyId: 7, + itemId: 'radarr1:movie:1', + filename: 'Movie.mkv', + totalBytes: 100, + clientTransferId: 'download-1', +} +// The file as served, unchanged since the transfer started. +const file = { filename: 'Movie.mkv', totalBytes: 100 } + +describe('UploadsRepository', () => { + test('creates an open transfer with nothing served yet', () => { + const { upload, created } = makeRepo().create(input) + expect(created).toBe(true) + expect(upload).toMatchObject({ + peerName: 'alice', + peerKeyId: 7, + totalBytes: 100, + rangeStart: null, + sentBytes: 0, + durationMs: null, + status: 'uploading', + finishedAt: null, + }) + expect(upload.transferId).toMatch(/^[0-9a-f-]{36}$/) + }) + + test('creating the same transfer again returns it; ids are scoped per key', () => { + const repo = makeRepo() + const first = repo.create(input).upload + expect(repo.create(input)).toEqual({ upload: first, created: false }) + expect(repo.create({ ...input, peerKeyId: 8 }).created).toBe(true) + expect(repo.create({ ...input, peerKeyId: null }).created).toBe(true) + expect(repo.create({ ...input, peerKeyId: null }).created).toBe(false) + expect(repo.list()).toHaveLength(3) + }) + + test('only the owning key finds a transfer', () => { + const repo = makeRepo() + const { transferId } = repo.create(input).upload + expect(repo.findOwned(transferId, 7)?.transferId).toBe(transferId) + expect(repo.findOwned(transferId, 8)).toBeNull() + expect(repo.findOwned(transferId, null)).toBeNull() + }) + + test('the receipt completes it with the bytes served since the lowest range, from creation', () => { + const repo = makeRepo() + const { id, transferId, startedAt } = repo.create(input).upload + const now = Date.parse(startedAt) + 5000 + + // Never requested: nothing was sent, so there is nothing to complete. + expect(repo.complete(transferId, 7, now)).toBe(false) + repo.served(id, 60, file) + repo.served(id, 40, file) + repo.served(id, 70, file) + // Another key, even one with the same name, can't close it; nor can the master key. + expect(repo.complete(transferId, 8, now)).toBe(false) + expect(repo.complete(transferId, null, now)).toBe(false) + expect(repo.complete(transferId, 7, now)).toBe(true) + expect(repo.get(id)).toMatchObject({ + status: 'completed', + rangeStart: 40, + sentBytes: 60, + durationMs: 5000, + }) + // Closed rows stay closed, and serving them again changes nothing. + expect(repo.complete(transferId, 7, now)).toBe(false) + repo.served(id, 0, file) + expect(repo.get(id)?.rangeStart).toBe(40) + }) + + test('a file replaced mid-transfer counts from the request that served the new one', () => { + const repo = makeRepo() + const { id, transferId } = repo.create(input).upload + repo.served(id, 40, file) + // *arr swapped in a 250-byte file; the peer's resume fails and it starts over. + repo.served(id, 0, { filename: 'Movie.2160p.mkv', totalBytes: 250 }) + repo.served(id, 100, { filename: 'Movie.2160p.mkv', totalBytes: 250 }) + expect(repo.complete(transferId, 7)).toBe(true) + expect(repo.get(id)).toMatchObject({ + filename: 'Movie.2160p.mkv', + totalBytes: 250, + rangeStart: 0, + sentBytes: 250, + }) + }) + + test('expires transfers that waited too long; a late receipt still completes one', () => { + const repo = makeRepo() + const { id, transferId } = repo.create(input).upload + repo.served(id, 0, file) + 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') + expect(repo.complete(transferId, 7)).toBe(true) + expect(repo.get(id)).toMatchObject({ status: 'completed', sentBytes: 100 }) + }) + + test('summarizes in SQL and lists only open transfers as active', () => { + const repo = makeRepo() + repo.create({ ...input, clientTransferId: 'a' }) + const done = repo.create({ ...input, clientTransferId: 'b' }).upload + repo.served(done.id, 40, file) + repo.complete(done.transferId, 7) + repo.create({ ...input, clientTransferId: 'c' }) + // Expires a and c; d is the only one still open. + repo.expireUnconfirmed('9999-01-01T00:00:00Z') + const open = repo.create({ ...input, clientTransferId: 'd' }).upload + + expect(repo.summary()).toEqual({ + byStatus: { + uploading: 1, + completed: 1, + unconfirmed: 2, + }, + total: 4, + bytesSent: 60, + }) + expect(repo.listActive().map(r => r.id)).toEqual([open.id]) + }) + + test('lists newest first', () => { + const repo = makeRepo() + const first = repo.create({ ...input, clientTransferId: 'a' }).upload + const second = repo.create({ ...input, clientTransferId: 'b' }).upload + expect(repo.list().map(r => r.id)).toEqual([second.id, first.id]) + }) + + test('notifies subscribers on every change', () => { + const repo = makeRepo() + let calls = 0 + repo.subscribe(() => calls++) + const { id, transferId } = repo.create(input).upload + repo.create(input) + repo.served(id, 0, file) + repo.complete(transferId, 7) + expect(calls).toBe(3) + }) +}) 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/connection.ts b/apps/backend/src/database/connection.ts index fab5908..13472cf 100644 --- a/apps/backend/src/database/connection.ts +++ b/apps/backend/src/database/connection.ts @@ -47,6 +47,9 @@ export async function openDatabase({ appConfigPath }: { appConfigPath: string }) const sqlite = new Database(path) sqlite.exec('pragma journal_mode = WAL') + // Wait for another connection's write lock (a second process on the same + // file, a backup tool) instead of failing with "database is locked". + sqlite.exec('pragma busy_timeout = 5000') sqlite.exec('pragma foreign_keys = ON') const db = drizzle({ client: sqlite, schema }) diff --git a/apps/backend/src/database/schema.ts b/apps/backend/src/database/schema.ts index 9f2aabe..e8b0353 100644 --- a/apps/backend/src/database/schema.ts +++ b/apps/backend/src/database/schema.ts @@ -32,6 +32,9 @@ export const downloads = sqliteTable('downloads', { expectedBytesSource: text('expected_bytes_source').$type(), expectedBytesMismatch: integer('expected_bytes_mismatch', { mode: 'boolean' }).notNull().default(false), downloadedBytes: integer('downloaded_bytes').notNull().default(0), + // Bytes per second between the last two progress writes; null until there are + // two in the current attempt. + speedBps: integer('speed_bps'), attempts: integer('attempts').notNull().default(0), status: text('status').$type().notNull(), startedAt: text('started_at').notNull(), @@ -81,6 +84,58 @@ export const apiKeys = sqliteTable('api_keys', { updatedAt: text('updated_at').notNull(), }) +// Files peers pull from this instance. One row per transfer, which a peer +// creates with `POST /peer/items/:itemId/downloads` and then streams, resumes +// and retries as often as it needs through `/peer/downloads/:downloadId/file`. +// +// Files go out zero-copy, so jack never counts the bytes itself. A dropped +// request leaves the transfer open to resume; the receipt the downloading jack +// posts once the file arrived completes it. No receipt within 48 hours leaves +// it `unconfirmed`. +export const UPLOAD_STATUSES = [ + 'uploading', + 'completed', + 'unconfirmed', +] as const +export type UploadStatus = typeof UPLOAD_STATUSES[number] + +export const uploads = sqliteTable('uploads', { + id: integer('id').primaryKey({ autoIncrement: true }), + // The opaque `downloadId` peers address the transfer by. + transferId: text('transfer_id').notNull().unique(), + // `:`: creating the same transfer again + // (a retry after a lost response) returns this row instead of a new one. + idempotencyKey: text('idempotency_key').notNull().unique(), + // Name of the api key the peer authenticated with ('master' for the master key). + // Display only: names are optional, editable and not unique. + peerName: text('peer_name').notNull(), + // The api key's row id, which owns the upload: only that key's receipt can + // close it. Null for the master key. + peerKeyId: integer('peer_key_id'), + itemId: text('item_id').notNull(), + filename: text('filename').notNull(), + // Size of the whole file on disk. + totalBytes: integer('total_bytes').notNull(), + // Lowest byte any request for this transfer started at (a resume of a .part + // the peer already had starts past 0). Null until the file is first requested. + rangeStart: integer('range_start'), + // Bytes this transfer delivered, `totalBytes - rangeStart`; 0 until the receipt. + sentBytes: integer('sent_bytes').notNull().default(0), + // Creation to receipt, retries and resumes included; null without a receipt. + durationMs: integer('duration_ms'), + status: text('status').$type().notNull(), + 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', '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 + 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..c7657dd 100644 --- a/apps/backend/src/index.ts +++ b/apps/backend/src/index.ts @@ -16,12 +16,16 @@ 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 { 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 +53,10 @@ 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)) + +// 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. @@ -85,6 +93,7 @@ const app = getApp(envs, config, connectorManager, { apiKeysRepository, managedKeysRepository, ownCatalogRepository, + uploadsService, }) const server = Bun.serve({ fetch: app.fetch, @@ -120,6 +129,28 @@ 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 so the UI toggle applies live. With the management API, the + // file is the truth: the UI turns it off by dropping the key, which must read + // as off rather than falling back to the value jack started with. + enabled: () => configService + ? configService.getRawDownloads()?.removeFailedDownloads ?? false + : downloadsConfig.removeFailedDownloads, + list: () => downloadsRepository.list(), + get: id => downloadsRepository.get(id), + retry: id => downloadsController.retry(id), + remove: (id, onlyIf) => downloadsService!.delete(id, onlyIf), + maxRetries: downloadsConfig.failedDownloadRetries, + baseDelayMs: downloadsConfig.failedRetryBaseDelayMs, + maxDelayMs: downloadsConfig.retryMaxDelayMs, + }) + : undefined + function startManagementServer() { if (!envs.MANAGEMENT_KEY) return undefined @@ -137,6 +168,7 @@ function startManagementServer() { downloadsRepository, downloadsService, importWatcher, + uploadsService, apiKeysRepository, catalogCacheRepository, tmdbMetadataRepository, @@ -185,6 +217,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 +238,7 @@ else { process.on('SIGINT', async () => { logger.info('SIGINT received, exiting') importWatcher?.stop() + failedDownloadReaper?.stop() catalogSync.stop() database.close() server.stop() @@ -215,6 +250,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..7d5d4af 100644 --- a/apps/backend/src/lib/servers/peer.ts +++ b/apps/backend/src/lib/servers/peer.ts @@ -13,7 +13,7 @@ import { PeerCatalogResponse } from '../peer-catalog' import { normalizeImdbId, Release } from '../release' import { setSpanAttribute, setSpanAttributes } from '../span-attributes' import { withSpan } from '../tracing' -import { compareVersions, isPeerVersionCompatible, MIN_PEER_PROTOCOL_VERSION, PEER_CATALOG_SUMMARY_VERSION } from '../version' +import { compareVersions, isPeerVersionCompatible, MIN_PEER_PROTOCOL_VERSION, PEER_CATALOG_SUMMARY_VERSION, PEER_TRANSFERS_VERSION } from '../version' import { ServerConnector } from './base' const PeerSearchResponse = z.object({ @@ -24,6 +24,10 @@ const DOWNLOAD_PROGRESS_INTERVAL_MS = 10_000 const DOWNLOAD_PROGRESS_BYTES = 64 * 1024 * 1024 const CONTENT_RANGE_PATTERN = /^bytes (\d+)-(\d+)\/(\d+)$/ +const PeerTransfer = z.object({ + downloadId: z.string().min(1), +}) + 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 } @@ -36,6 +40,9 @@ export interface PeerDownloadOptions { torrentFilename?: string partPath?: string releaseSize?: number + // Stable for every attempt at one download, so the peer keeps a single + // transfer for it however often it is retried or resumed. + transferId?: string onProgress?: (event: PeerDownloadProgressEvent) => void | Promise } @@ -254,6 +261,80 @@ export class PeerConnector extends ServerConnector { return this.fetch(`/peer/items/${encodeURIComponent(id)}`, { method: 'GET', schema: Release }) } + /** + * Where to stream item `id` from. A peer on protocol 0.3.0+ gets a transfer it + * records as an upload; `transferId` makes creating it idempotent, so every + * attempt at one download reuses the same transfer. Older peers, or a create + * that fails (say the peer's database is full), fall back to the untracked + * file route: bookkeeping never blocks a download. + */ + private async openTransfer(id: string, transferId: string | undefined, signal: AbortSignal): Promise<{ url: URL, downloadId: string | null }> { + const legacy = { + url: new URL(`/peer/items/${encodeURIComponent(id)}/file`, this.url), + downloadId: null, + } + if (!transferId || !this._peerVersion || compareVersions(this._peerVersion, PEER_TRANSFERS_VERSION) < 0) + return legacy + try { + const transfer = await this.fetch(`/peer/items/${encodeURIComponent(id)}/downloads`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ clientTransferId: transferId }), + schema: PeerTransfer, + signal: AbortSignal.any([ + signal, + AbortSignal.timeout(30_000), + ]), + }) + // Built here rather than taken from the response, so a peer can't point the + // request (and the api key it carries) at another host. + return { + url: new URL(`/peer/downloads/${encodeURIComponent(transfer.downloadId)}/file`, this.url), + downloadId: transfer.downloadId, + } + } + catch (err) { + signal.throwIfAborted() + logger.debug({ + peer: this.name, + id, + error: err instanceof Error + ? err.message + : String(err), + }, 'Could not start a transfer; downloading untracked') + return legacy + } + } + + /** + * Tell the serving jack the whole file arrived. It sends files zero-copy, so it + * can't see a transfer end. Fire and forget: a lost receipt only leaves that + * upload unconfirmed. + */ + private sendUploadReceipt(downloadId: string | null): void { + if (!downloadId) + return + const url = new URL(`/peer/downloads/${encodeURIComponent(downloadId)}/receipt`, this.url) + fetch(url, { + method: 'POST', + headers: { ...this.headers, 'X-Api-Key': this.apiKey }, + signal: AbortSignal.timeout(10_000), + }) + .then((response) => { + if (!response.ok) + logger.debug({ peer: this.name, downloadId, status: response.status }, 'Peer did not accept the upload receipt') + }) + .catch((err) => { + logger.debug({ + peer: this.name, + downloadId, + error: err instanceof Error + ? err.message + : String(err), + }, 'Could not send the upload receipt') + }) + } + @requiresInitialization async downloadFile(id: string, destPath: string, options: PeerDownloadOptions = {}): Promise { return withSpan('peer.download_file', { @@ -264,10 +345,15 @@ export class PeerConnector extends ServerConnector { }, async (span) => { const idleTimeoutMs = options.idleTimeoutMs ?? 60_000 const torrentFilename = options.torrentFilename - const url = new URL(`/peer/items/${encodeURIComponent(id)}/file`, this.url) + // The untracked file route until the transfer is opened, right before the + // first request (a .part that is already complete needs none). + let transfer: { url: URL, downloadId: string | null } = { + url: new URL(`/peer/items/${encodeURIComponent(id)}/file`, this.url), + downloadId: null, + } const partPath = options.partPath ?? `${destPath}.part` const baseHeaders = { ...this.headers, 'X-Api-Key': this.apiKey } - setSpanAttributes(span, { 'http.request.idle_timeout_ms': idleTimeoutMs, 'url.path': url.pathname }) + setSpanAttribute(span, 'http.request.idle_timeout_ms', idleTimeoutMs) // Idle (inactivity) timeout, armed ONLY around network waits (fetch + each // read) and cleared before local file/progress work, so slow disk I/O never @@ -302,7 +388,7 @@ export class PeerConnector extends ServerConnector { signal.throwIfAborted() armIdle() try { - return await fetch(url, { + return await fetch(transfer.url, { headers: withRange ? { ...baseHeaders, Range: `bytes=${existingBytes}-` } : baseHeaders, @@ -381,6 +467,8 @@ export class PeerConnector extends ServerConnector { } } + transfer = await this.openTransfer(id, options.transferId, signal) + setSpanAttribute(span, 'url.path', transfer.url.pathname) let response = await doFetch(existingBytes > 0) setSpanAttribute(span, 'http.response.status_code', response.status) @@ -582,6 +670,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(transfer.downloadId) try { await options.onProgress?.({ type: 'completed', downloadedBytes, expectedBytes }) } diff --git a/apps/backend/src/lib/version.ts b/apps/backend/src/lib/version.ts index 9044a24..7dd2252 100644 --- a/apps/backend/src/lib/version.ts +++ b/apps/backend/src/lib/version.ts @@ -1,6 +1,6 @@ // jack's own version, reported to peers over /handshake. This doubles as the // peer-protocol version: a bump here signals a potential protocol change. -export const PROTOCOL_VERSION = '0.2.0' +export const PROTOCOL_VERSION = '0.3.0' // Oldest peer version we can still talk to. Peers below this — or peers too old // to expose a version at all — are rejected at init time as incompatible. @@ -10,6 +10,10 @@ export const MIN_PEER_PROTOCOL_VERSION = '0.1.0' // only have `/peer/search`, so their catalog is summarized from full releases. export const PEER_CATALOG_SUMMARY_VERSION = '0.2.0' +// First version with transfers (`POST /peer/items/:itemId/downloads`), which it +// records as uploads. Older peers only serve the untracked `/items/:itemId/file`. +export const PEER_TRANSFERS_VERSION = '0.3.0' + const VERSION_PATTERN = /^(\d+)\.(\d+)\.(\d+)$/ function parseVersion(version: string): [number, number, number] | null { diff --git a/apps/backend/src/management-app.ts b/apps/backend/src/management-app.ts index 981d28b..12411e4 100644 --- a/apps/backend/src/management-app.ts +++ b/apps/backend/src/management-app.ts @@ -7,6 +7,7 @@ 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 { 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 +41,7 @@ export function getManagementApp(params: { downloadsRepository?: DownloadsRepository downloadsService?: DownloadsService importWatcher?: ImportWatcher + uploadsService?: UploadsService apiKeysRepository?: ApiKeysRepository catalogCacheRepository?: CatalogCacheRepository tmdbMetadataRepository?: TmdbMetadataRepository @@ -84,7 +86,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) app.route('/', getStatusRouter(statusController)) app.route('/downloads', getDownloadsManagementRouter(new DownloadsManagementController( diff --git a/apps/backend/src/middleware/require-auth.ts b/apps/backend/src/middleware/require-auth.ts index 8e9c1bb..aef7371 100644 --- a/apps/backend/src/middleware/require-auth.ts +++ b/apps/backend/src/middleware/require-auth.ts @@ -5,7 +5,11 @@ import { constantTimeEqual, hashKey, isGeneratedKey, isManagedKey } from '../lib import { UnauthorizedError } from '../lib/errors/UnauthorizedError' export interface AuthVariables { + // For display: key names are optional, editable and not unique. apiKeyName?: string + // Stable identity of the peer key (its row id), for ownership checks. Unset for + // the master key and managed keys. + apiKeyId?: number } // Each protected surface accepts exactly one class of key, so the scope is mounted @@ -63,6 +67,7 @@ export function requireApiKey(masterKey: string, scope: ApiKeyScope) { } ctx.set('apiKeyName', resolution.row.name ?? 'unnamed') + ctx.set('apiKeyId', resolution.row.id) return next() }) } 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..f0e9dbf 100644 --- a/apps/backend/src/modules/downloads/downloads.repository.ts +++ b/apps/backend/src/modules/downloads/downloads.repository.ts @@ -4,8 +4,8 @@ import type { FileSystemErrorDetails } from '../../lib/errors/FileSystemError' import type { Release } from '../../lib/release' import type { ManualImportTarget } from '../../lib/servers/arr/base' import type { DownloadErrorCode } from './download-error-codes' -import { desc, eq, sql } from 'drizzle-orm' -import { downloads } from '../../database/schema' +import { count, desc, eq, sql, sum } from 'drizzle-orm' +import { DOWNLOAD_STATUSES, downloads } from '../../database/schema' export interface DownloadRecord { id: number @@ -22,6 +22,8 @@ export interface DownloadRecord { expectedBytesSource: ExpectedBytesSource | null expectedBytesMismatch: boolean downloadedBytes: number + // Bytes per second between the last two progress writes of this attempt. + speedBps: number | null attempts: number status: DownloadStatus startedAt: string @@ -78,6 +80,7 @@ function toRecord(row: DownloadRow): DownloadRecord { expectedBytesSource: row.expectedBytesSource ?? null, expectedBytesMismatch: row.expectedBytesMismatch, downloadedBytes: row.downloadedBytes, + speedBps: row.speedBps, attempts: row.attempts, status: row.status, startedAt: row.startedAt, @@ -160,6 +163,40 @@ export class DownloadsRepository { : null } + /** + * Overview totals, counted in SQLite so the history (and its release JSON) is + * never loaded. `bytesMoved` counts partial transfers too; `mismatched` is the + * in-flight or queued rows flagged with a size mismatch. + */ + summary(): { byStatus: Record, total: number, bytesMoved: number, mismatched: number } { + const rows = this.db.select({ + status: downloads.status, + count: count(), + bytes: sum(downloads.downloadedBytes), + mismatched: sql`sum(case when ${downloads.expectedBytesMismatch} then 1 else 0 end)`, + }) + .from(downloads) + .groupBy(downloads.status) + .all() + const byStatus = Object.fromEntries(DOWNLOAD_STATUSES.map(s => [s, 0])) as Record + let total = 0 + let bytesMoved = 0 + let mismatched = 0 + for (const row of rows) { + byStatus[row.status] = row.count + total += row.count + bytesMoved += Number(row.bytes ?? 0) + if (row.status === 'downloading' || row.status === 'import_queued') + mismatched += Number(row.mismatched ?? 0) + } + return { + byStatus, + total, + bytesMoved, + mismatched, + } + } + list(): DownloadRecord[] { // Secondary sort by id breaks ties when two rows share an updatedAt // (ISO timestamps can collide within the same millisecond). @@ -179,10 +216,11 @@ export class DownloadsRepository { this.notifyChanged() } - updateProgress(id: number, downloadedBytes: number): void { + updateProgress(id: number, downloadedBytes: number, speedBps: number | null = null): void { this.db.update(downloads) .set({ downloadedBytes, + speedBps, updatedAt: nowIso(), }) .where(eq(downloads.id, id)) @@ -190,6 +228,11 @@ export class DownloadsRepository { this.notifyChanged() } + clearSpeed(id: number): void { + this.db.update(downloads).set({ speedBps: null }).where(eq(downloads.id, id)).run() + this.notifyChanged() + } + // Download finished and the file is in completedPath, handed to *arr to import. // `completedAt` records when the transfer finished (what the qB API reports as // completion_on); the row stays here until *arr's import is detected. @@ -280,6 +323,8 @@ export class DownloadsRepository { const row = this.db.update(downloads) .set({ attempts: sql`${downloads.attempts} + 1`, + // A new attempt has no samples yet. + speedBps: null, updatedAt: nowIso(), }) .where(eq(downloads.id, id)) @@ -309,6 +354,7 @@ export class DownloadsRepository { this.db.update(downloads) .set({ downloadedBytes: 0, + speedBps: null, error: 'resume validation failed; restarted from byte 0', updatedAt: nowIso(), }) @@ -345,9 +391,12 @@ export class DownloadsRepository { this.notifyChanged() } - /** Rows in a given status (uses the status index). */ - listByStatus(status: DownloadStatus): DownloadRecord[] { - return this.db.select().from(downloads).where(eq(downloads.status, status)).all().map(toRecord) + /** Rows in a given status (uses the status index), most recently updated first, optionally capped. */ + listByStatus(status: DownloadStatus, limit?: number): DownloadRecord[] { + const query = this.db.select().from(downloads).where(eq(downloads.status, status)).orderBy(desc(downloads.updatedAt), desc(downloads.id)) + return (limit === undefined + ? query.all() + : query.limit(limit).all()).map(toRecord) } /** Stale `downloading` rows from a prior run, returned for active re-drive (no mutation). */ diff --git a/apps/backend/src/modules/downloads/downloads.service.ts b/apps/backend/src/modules/downloads/downloads.service.ts index ed33cc4..2a2a7df 100644 --- a/apps/backend/src/modules/downloads/downloads.service.ts +++ b/apps/backend/src/modules/downloads/downloads.service.ts @@ -144,6 +144,7 @@ export class DownloadsService { expectedBytesSource: null, expectedBytesMismatch: false, downloadedBytes: 0, + speedBps: null, attempts: 0, status: 'downloading', startedAt: '', @@ -330,7 +331,11 @@ export class DownloadsService { return repo.get(id) ?? record } - async delete(id: number): Promise { + /** + * `onlyIf`, when given, is checked against the row inside the per-row lock: if + * it no longer holds (say, the user retried it meanwhile), nothing is deleted. + */ + async delete(id: number, onlyIf?: (row: DownloadRecord) => boolean): Promise { const repo = this.requireRepository() const record = repo.get(id) if (!record) @@ -343,6 +348,8 @@ export class DownloadsService { const current = repo.get(id) if (!current) throw new NotFoundError(`Download ${id} not found`) + if (onlyIf && !onlyIf(current)) + return if (current.status === 'downloading') await this.cancel(id) @@ -379,6 +386,8 @@ export class DownloadsService { private async downloadWithRetry(record: DownloadRecord, signal: AbortSignal): Promise { const repo = this.downloadsRepository + // The previous progress sample in this attempt, to work out the speed from. + let last: { at: number, bytes: number } | undefined const peer = this.peers.find(p => p.id === record.peerId) if (!peer) { repo?.markFailed(record.id, 'PEER_NOT_FOUND', `Peer ${record.peerId} not found`) @@ -392,26 +401,35 @@ export class DownloadsService { return } if (event.type === 'progress') { - repo?.updateProgress(record.id, event.downloadedBytes) + const at = Date.now() + const speedBps = last && at > last.at + ? Math.round(((event.downloadedBytes - last.bytes) * 1000) / (at - last.at)) + : null + last = { at, bytes: event.downloadedBytes } + repo?.updateProgress(record.id, event.downloadedBytes, speedBps) return } if (event.type === 'restart') { + last = undefined repo?.markResumeReset(record.id) return } // event.type === 'completed': the transfer finished. Record the final byte // count; the status flips to import_queued once downloadFile resolves below. - repo?.updateProgress(record.id, event.downloadedBytes) + repo?.updateProgress(record.id, event.downloadedBytes, null) } try { await retry(async () => { signal.throwIfAborted() + last = undefined repo?.incrementAttempts(record.id) await peer.downloadFile(record.itemId, record.destPath, { torrentFilename: record.torrentFilename, partPath: record.partPath, releaseSize: record.releaseSize, + // Stable for this download across retries and restarts. + transferId: `download-${record.id}-${Date.parse(record.startedAt)}`, idleTimeoutMs: this.config.idleTimeoutMs, onProgress, signal, @@ -429,6 +447,8 @@ export class DownloadsService { delayMs, ...transferErrorLogFields(error), }, 'Retrying peer download after transient failure') + // Nothing arrives while it waits to retry. + repo?.clearSpeed(record.id) }, signal, }) 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..8fe53d5 --- /dev/null +++ b/apps/backend/src/modules/downloads/failed-download-reaper.ts @@ -0,0 +1,168 @@ +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[] + /** The row as it is now; each one is re-read right before acting on it. */ + get: (id: number) => DownloadRecord | null + /** Repeats the row's last failed operation (transfer or import). */ + retry: (id: number) => Promise + /** + * Deletes the row and its unshared partial or completed files, but only if + * `isReapable` still holds once the per-row delete lock is taken. + */ + remove: (id: number, onlyIf: (row: DownloadRecord) => boolean) => Promise + maxRetries: number + baseDelayMs: number + maxDelayMs: number + /** Injectable for tests. */ + now?: () => number +} + +/** A failed row the reaper may touch. A user cancel keeps its .part on purpose. */ +export function isReapable(row: DownloadRecord): boolean { + return row.status === 'failed' && row.errorCode !== 'CANCELLED' +} + +/** 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 { id } of rows) { + // Earlier iterations await, so the user may have switched removal off, or + // retried or cancelled this row, meanwhile: check both again. + if (!this.options.enabled()) + return + const row = this.options.get(id) + if (!row || !isReapable(row)) + 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 { + // Checked again inside the per-row delete lock, setting included. + await this.options.remove(row.id, current => this.options.enabled() && isReapable(current)) + 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/peer/peer.controller.ts b/apps/backend/src/modules/peer/peer.controller.ts index 267d34e..6e02000 100644 --- a/apps/backend/src/modules/peer/peer.controller.ts +++ b/apps/backend/src/modules/peer/peer.controller.ts @@ -39,7 +39,10 @@ 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: jack knows where each request +// starts, and the downloading jack's receipt says when it all arrived (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..dcf3f3e 100644 --- a/apps/backend/src/modules/peer/peer.router.ts +++ b/apps/backend/src/modules/peer/peer.router.ts @@ -1,8 +1,11 @@ -import type { PeerController } from './peer.controller' +import type { AuthVariables } from '../../middleware/require-auth' +import type { UploadsService } from '../uploads/uploads.service' +import type { PeerController, StreamFileResult } from './peer.controller' import { Hono } from 'hono' import { describeRoute, resolver, validator as zValidator } from 'hono-openapi' import z from 'zod' import { PeerCatalogResponse } from '../../lib/peer-catalog' +import { logger } from '../../logger' // Strip characters that could break out of the quoted filename or inject extra // response headers (CR/LF/quote). The name comes from a trusted *arr basename, so @@ -12,8 +15,51 @@ function safeFilename(name: string): string { return name.replace(UNSAFE_FILENAME_CHARS, '') } -export function getPeerRouter(controller: PeerController) { - const app = new Hono() +const CreateTransfer = z.object({ + // Stable across the downloading jack's retries of one download, so a create + // whose response got lost returns the same transfer. + clientTransferId: z.string().min(1).max(200), +}) + +const FILE_RESPONSES = { + 200: { + description: 'Full file stream', + content: { 'application/octet-stream': {} }, + }, + 206: { + description: 'Partial content for a satisfiable Range request', + content: { 'application/octet-stream': {} }, + }, + 416: { description: 'Unsatisfiable Range request' }, +} + +function fileResponse(result: StreamFileResult): Response { + if (result.type === 'unsatisfiable') { + return new Response(null, { + status: 416, + headers: { + 'Content-Range': `bytes */${result.totalSize}`, + 'Accept-Ranges': 'bytes', + }, + }) + } + const headers = { + 'Content-Type': 'application/octet-stream', + 'Content-Length': String(result.size), + 'Accept-Ranges': 'bytes', + 'Content-Disposition': `attachment; filename="${safeFilename(result.filename)}"`, + } + if (result.type === 'partial') { + return new Response(result.body, { + status: 206, + headers: { ...headers, 'Content-Range': `bytes ${result.start}-${result.end}/${result.totalSize}` }, + }) + } + return new Response(result.body, { headers }) +} + +export function getPeerRouter(controller: PeerController, uploads?: UploadsService) { + const app = new Hono<{ Variables: AuthVariables }>() app.get( '/search', @@ -97,62 +143,112 @@ export function getPeerRouter(controller: PeerController) { app.get('/items/:itemId/file', describeRoute({ tags: ['Peer'], - summary: 'Stream the release file', - description: 'Streams the file straight from disk at the path the local *arr reports. Supports HTTP Range requests for resumable transfers.', + summary: 'Stream the release file (untracked)', + description: 'Streams the file straight from disk at the path the local *arr reports, with HTTP Range support. Not recorded as an upload: peers on protocol 0.3.0 and later create a transfer with `POST /peer/items/:itemId/downloads` instead. Kept for older peers.', security: [ { 'X-Api-Key': [] }, ], responses: { - 200: { - description: 'Full file stream', - content: { 'application/octet-stream': {} }, - }, - 206: { - description: 'Partial content for a satisfiable Range request', - content: { 'application/octet-stream': {} }, - }, + ...FILE_RESPONSES, 404: { description: 'No item with this id, or its file is missing on disk' }, - 416: { description: 'Unsatisfiable Range request' }, }, }), async (c) => { - const { itemId } = c.req.param() - const result = await controller.streamFile(itemId, c.req.header('Range')) - - if (!result) { + const result = await controller.streamFile(c.req.param('itemId'), c.req.header('Range')) + if (!result) return c.json({ error: 'File not found' }, 404) - } + return fileResponse(result) + }) - if (result.type === 'unsatisfiable') { - return new Response(null, { - status: 416, - headers: { - 'Content-Range': `bytes */${result.totalSize}`, - 'Accept-Ranges': 'bytes', - }, - }) - } + app.post('/items/:itemId/downloads', describeRoute({ + tags: ['Peer'], + summary: 'Start a transfer', + description: 'Creates the transfer a peer then streams from `url` and completes with a receipt. Idempotent per api key: the same `clientTransferId` returns the transfer it created before (200) rather than a new one (201). Added in protocol 0.3.0.', + security: [ + { 'X-Api-Key': [] }, + ], + responses: { + 200: { description: 'The transfer this `clientTransferId` already created' }, + 201: { description: 'The new transfer: `downloadId`, `filename`, `totalBytes` and `url`' }, + 404: { description: 'No item with this id, or its file is missing on disk' }, + 409: { description: 'This `clientTransferId` already belongs to a transfer of another item' }, + }, + }), zValidator('json', CreateTransfer), async (c) => { + const itemId = c.req.param('itemId') + const result = await controller.streamFile(itemId) + if (!result || result.type !== 'full' || !uploads) + return c.json({ error: 'File not found' }, 404) + // The body is a lazy file handle; leaving it unread opens nothing. + const { upload, created } = uploads.repository.create({ + peerName: c.get('apiKeyName') ?? 'master', + peerKeyId: c.get('apiKeyId') ?? null, + itemId, + filename: result.filename, + totalBytes: result.size, + clientTransferId: c.req.valid('json').clientTransferId, + }) + if (upload.itemId !== itemId) + return c.json({ error: 'clientTransferId is in use for another item' }, 409) + return c.json({ + downloadId: upload.transferId, + filename: upload.filename, + totalBytes: upload.totalBytes, + url: `/peer/downloads/${upload.transferId}/file`, + }, created + ? 201 + : 200) + }) - if (result.type === 'partial') { - return new Response(result.body, { - status: 206, - headers: { - 'Content-Type': 'application/octet-stream', - 'Content-Length': String(result.size), - 'Content-Range': `bytes ${result.start}-${result.end}/${result.totalSize}`, - 'Accept-Ranges': 'bytes', - 'Content-Disposition': `attachment; filename="${safeFilename(result.filename)}"`, - }, - }) + app.get('/downloads/:downloadId/file', describeRoute({ + tags: ['Peer'], + summary: 'Stream a transfer', + description: 'Streams the transfer\'s file straight from disk, with HTTP Range support for resuming. Request it as often as needed: every request belongs to the same transfer, and a dropped one leaves it open. Added in protocol 0.3.0.', + security: [ + { 'X-Api-Key': [] }, + ], + responses: { + ...FILE_RESPONSES, + 404: { description: 'No transfer with this id for this api key, or its file is missing on disk' }, + }, + }), async (c) => { + const upload = uploads?.repository.findOwned(c.req.param('downloadId'), c.get('apiKeyId') ?? null) + if (!upload) + return c.json({ error: 'Transfer not found' }, 404) + const result = await controller.streamFile(upload.itemId, c.req.header('Range')) + if (!result) + return c.json({ error: 'File not found' }, 404) + // A HEAD request sends no body, so it serves nothing. Bookkeeping never fails + // a transfer: a database error is logged and the file still goes out. + if (result.type !== 'unsatisfiable' && c.req.method !== 'HEAD') { + try { + uploads!.repository.served(upload.id, result.type === 'partial' + ? result.start + : 0, { + filename: result.filename, + totalBytes: result.type === 'partial' + ? result.totalSize + : result.size, + }) + } + catch (err) { + logger.warn({ uploadId: upload.id, err }, 'Could not record a transfer request') + } } + return fileResponse(result) + }) - return new Response(result.body, { - headers: { - 'Content-Type': 'application/octet-stream', - 'Content-Length': String(result.size), - 'Accept-Ranges': 'bytes', - 'Content-Disposition': `attachment; filename="${result.filename}"`, - }, - }) + app.post('/downloads/:downloadId/receipt', describeRoute({ + tags: ['Peer'], + summary: 'Complete a transfer', + description: 'Sent by the downloading jack once the whole file arrived. jack records the bytes it served and the time since the transfer was created. Always answers 204, so a stale or unknown id is not an error worth retrying. Added in protocol 0.3.0.', + security: [ + { 'X-Api-Key': [] }, + ], + responses: { + 204: { description: 'Receipt taken (or ignored)' }, + }, + }), (c) => { + uploads?.repository.complete(c.req.param('downloadId'), c.get('apiKeyId') ?? null) + return c.body(null, 204) }) return app diff --git a/apps/backend/src/modules/status/status.controller.ts b/apps/backend/src/modules/status/status.controller.ts index 2c3cdbe..7dcad61 100644 --- a/apps/backend/src/modules/status/status.controller.ts +++ b/apps/backend/src/modules/status/status.controller.ts @@ -1,9 +1,11 @@ -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 { UploadRecord } from '../uploads/uploads.repository' +import type { UploadsService } from '../uploads/uploads.service' +import { DOWNLOAD_STATUSES, UPLOAD_STATUSES } from '../../database/schema' // 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,7 +27,11 @@ function baseConnector(c: ServerConnector) { } } -export function enrichDownload(d: DownloadRecord) { +// Progress is written at least every 10 s while bytes arrive; a row quiet for +// longer than this is stalled, so its last speed no longer holds. +const SPEED_STALE_MS = 30_000 + +export function enrichDownload(d: DownloadRecord, now = Date.now()) { // Prefer the negotiated expected size; fall back to the release-advertised size. const totalBytes = d.expectedBytes ?? (d.releaseSize > 0 ? d.releaseSize @@ -33,28 +39,102 @@ export function enrichDownload(d: DownloadRecord) { const progress = totalBytes && totalBytes > 0 ? Math.min(1, d.downloadedBytes / totalBytes) : null - return { ...d, totalBytes, progress } + // Measured between progress writes (every 10 s / 64 MiB); only while bytes + // still arrive. A failed attempt clears it before its retry wait. + const speedBps = d.status === 'downloading' && now - Date.parse(d.updatedAt) <= SPEED_STALE_MS + ? d.speedBps + : null + const etaSeconds = speedBps && speedBps > 0 && totalBytes + ? Math.max(0, Math.round((totalBytes - d.downloadedBytes) / speedBps)) + : null + return { + ...d, + totalBytes, + progress, + speedBps, + etaSeconds, + } +} + +export function enrichUpload(u: UploadRecord) { + // Files go out zero-copy, so only a receipt says the file arrived. + const progress = u.status === 'completed' + ? 1 + : 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 } +} + +type ChangeSource = ((subscriber: () => void) => () => void) | undefined + +// Each stream sends a full snapshot per notification, so all of its sources are +// folded together: the first change goes out at once, and any further changes +// within the interval become a single trailing snapshot. +export const STREAM_INTERVAL_MS = 1000 + +export function coalesce(subscriber: () => void, sources: ChangeSource[], intervalMs = STREAM_INTERVAL_MS): () => void { + let lastFired = 0 + let timer: ReturnType | undefined + const fire = () => { + timer = undefined + lastFired = Date.now() + subscriber() + } + const notify = () => { + if (timer) + return + const wait = lastFired + intervalMs - Date.now() + if (wait <= 0) { + fire() + return + } + timer = setTimeout(fire, wait) + } + const unsubscribers = sources.flatMap(source => source + ? [ + source(notify), + ] + : []) + return () => { + clearTimeout(timer) + for (const unsubscribe of unsubscribers) + unsubscribe() + } } export class StatusController { constructor( private readonly connectors: { servers: ArrServerConnector[], peers: PeerConnector[], subscribe?: (subscriber: () => void) => () => void }, private readonly downloads?: DownloadsRepository, + private readonly uploads?: UploadsService, ) {} + private enrichUpload = (u: UploadRecord) => enrichUpload(u) + + private enrich = (d: DownloadRecord) => enrichDownload(d) + subscribeOverview(subscriber: () => void): () => void { - const unsubscribers = [ - this.connectors.subscribe?.(subscriber), - this.downloads?.subscribe(subscriber), - ].filter((unsubscribe): unsubscribe is () => void => Boolean(unsubscribe)) - return () => { - for (const unsubscribe of unsubscribers) - unsubscribe() - } + return coalesce(subscriber, [ + // Called through the object: ConnectorManager.subscribe reads `this`. + this.connectors.subscribe && (s => this.connectors.subscribe!(s)), + this.downloads && (s => this.downloads!.subscribe(s)), + this.uploads && (s => this.uploads!.subscribe(s)), + ]) + } + + subscribeUploads(subscriber: () => void): () => void { + return coalesce(subscriber, [ + this.uploads && (s => this.uploads!.subscribe(s)), + ]) } subscribeDownloads(subscriber: () => void): () => void { - return this.downloads?.subscribe(subscriber) ?? (() => {}) + return coalesce(subscriber, [ + this.downloads && (s => this.downloads!.subscribe(s)), + ]) } /** Whether download-backed endpoints are available (a repository was injected). */ @@ -65,21 +145,16 @@ export class StatusController { getOverview() { const peers = this.connectors.peers const servers = this.connectors.servers - const records = this.downloads?.list() ?? [] - - const byStatus = Object.fromEntries(DOWNLOAD_STATUSES.map(s => [s, 0])) as Record - // `bytesMoved` is the lifetime total of bytes actually pulled over the wire - // (partial transfers count what they got), powering the dashboard's vanity stat. - let bytesMoved = 0 - // True count of in-flight/queued transfers flagged with a size mismatch, so the - // dashboard reports an accurate total even though the detail arrays are capped. - let mismatched = 0 - for (const record of records) { - byStatus[record.status]++ - bytesMoved += record.downloadedBytes - if (record.expectedBytesMismatch && (record.status === 'downloading' || record.status === 'import_queued')) - mismatched++ + // This runs on every overview snapshot (about once a second while downloads + // move), so totals come from SQLite and only the rows shown are loaded. The + // detail lists are capped; `mismatched` still counts every flagged row. + const totals = this.downloads?.summary() ?? { + byStatus: Object.fromEntries(DOWNLOAD_STATUSES.map(s => [s, 0])) as Record, + total: 0, + bytesMoved: 0, + mismatched: 0, } + const inStatus = (status: DownloadStatus, limit?: number) => (this.downloads?.listByStatus(status, limit) ?? []).map(this.enrich) return { peers: { @@ -102,22 +177,45 @@ export class StatusController { })), }, downloads: { - total: records.length, - byStatus, - 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), + ...totals, + // Most recently updated first. + active: inStatus('downloading'), + importQueued: inStatus('import_queued', ATTENTION_LIMIT), + failed: inStatus('failed', ATTENTION_LIMIT), }, + uploads: this.getUploadsSummary(), + } + } + + // Runs on every overview snapshot (about once a second while downloads move), so + // totals come from SQLite and only running uploads are loaded: the history + // grows forever. + private getUploadsSummary() { + if (!this.uploads) { + return { + total: 0, + byStatus: Object.fromEntries(UPLOAD_STATUSES.map(s => [s, 0])) as Record, + bytesSent: 0, + active: [], + } + } + return { + ...this.uploads.repository.summary(), + active: this.uploads.repository.listActive().map(this.enrichUpload), + } + } + + 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..62002b6 100644 --- a/apps/backend/src/modules/status/status.router.ts +++ b/apps/backend/src/modules/status/status.router.ts @@ -62,6 +62,34 @@ 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 transfer, however often the peer resumed or retried it.', + 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))) 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..9af3d16 --- /dev/null +++ b/apps/backend/src/modules/uploads/uploads.repository.ts @@ -0,0 +1,179 @@ +import type { AppDatabase } from '../../database/connection' +import type { UploadRow, UploadStatus } from '../../database/schema' +import { and, count, desc, eq, inArray, isNotNull, isNull, lt, sql, sum } from 'drizzle-orm' +import { UPLOAD_STATUSES, uploads } from '../../database/schema' + +export type UploadRecord = UploadRow + +// A transfer can still be resumed and completed; a late receipt revives an expired one. +const OPEN = ['uploading', 'unconfirmed'] as const + +export interface CreateTransferInput { + peerName: string + peerKeyId: number | null + itemId: string + filename: string + totalBytes: number + clientTransferId: string +} + +function nowIso() { + return new Date().toISOString() +} + +/** Rows a key owns: only its own api key (null for the master key) may use a transfer. */ +function ownedBy(peerKeyId: number | null) { + return peerKeyId === null + ? isNull(uploads.peerKeyId) + : eq(uploads.peerKeyId, peerKeyId) +} + +/** + * Files peers pull from this instance, one row per transfer. 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. + } + } + } + + /** + * Creates the transfer, or returns the one this key already created with the + * same client transfer id. `created` is false for the latter. + */ + create(input: CreateTransferInput): { upload: UploadRecord, created: boolean } { + const { clientTransferId, ...row } = input + const idempotencyKey = `${input.peerKeyId ?? 'master'}:${clientTransferId}` + const timestamp = nowIso() + const inserted = this.db.insert(uploads) + .values({ + ...row, + transferId: crypto.randomUUID(), + idempotencyKey, + status: 'uploading', + startedAt: timestamp, + updatedAt: timestamp, + }) + .onConflictDoNothing({ target: uploads.idempotencyKey }) + .returning() + .get() + if (inserted) { + this.notifyChanged() + return { upload: inserted, created: true } + } + const existing = this.db.select().from(uploads).where(eq(uploads.idempotencyKey, idempotencyKey)).get()! + return { upload: existing, created: false } + } + + get(id: number): UploadRecord | null { + return this.db.select().from(uploads).where(eq(uploads.id, id)).get() ?? null + } + + /** The transfer behind a `downloadId`, if this key owns it. */ + findOwned(transferId: string, peerKeyId: number | null): UploadRecord | null { + return this.db.select().from(uploads).where(and(eq(uploads.transferId, transferId), ownedBy(peerKeyId))).get() ?? null + } + + /** Totals for the dashboard, counted in SQLite so the history is never loaded. */ + summary(): { byStatus: Record, total: number, bytesSent: number } { + const rows = this.db.select({ + status: uploads.status, + count: count(), + bytes: sum(uploads.sentBytes), + }) + .from(uploads) + .groupBy(uploads.status) + .all() + const byStatus = Object.fromEntries(UPLOAD_STATUSES.map(s => [s, 0])) as Record + let total = 0 + let bytesSent = 0 + for (const row of rows) { + byStatus[row.status] = row.count + total += row.count + bytesSent += Number(row.bytes ?? 0) + } + return { byStatus, total, bytesSent } + } + + /** Transfers still open, newest first. */ + listActive(): UploadRecord[] { + return this.db.select().from(uploads).where(eq(uploads.status, 'uploading')).orderBy(desc(uploads.startedAt), desc(uploads.id)).all() + } + + /** 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() + } + + /** + * A request for the file as it is now on disk, starting at byte `start`. Keeps + * the lowest start seen; a file *arr replaced since (another size) starts over + * from this request, since the peer has to download the new file anyway. + */ + served(id: number, start: number, file: { filename: string, totalBytes: number }): void { + this.db.update(uploads) + .set({ + rangeStart: sql`case when ${uploads.totalBytes} = ${file.totalBytes} then min(coalesce(${uploads.rangeStart}, ${start}), ${start}) else ${start} end`, + totalBytes: file.totalBytes, + filename: file.filename, + updatedAt: nowIso(), + }) + .where(and(eq(uploads.id, id), inArray(uploads.status, [...OPEN]))) + .run() + this.notifyChanged() + } + + /** + * The downloading peer's receipt: the whole file arrived. Only the owning key + * can close a transfer, only while it is open, and only once the file was + * requested. The bytes and time come from what jack served, so a peer can't + * skew the totals. Returns whether the row was updated. + */ + complete(transferId: string, peerKeyId: number | null, now = Date.now()): boolean { + const changed = this.db.update(uploads) + .set({ + status: 'completed', + sentBytes: sql`${uploads.totalBytes} - ${uploads.rangeStart}`, + durationMs: sql`max(0, ${now} - cast(round(unixepoch(${uploads.startedAt}, 'subsec') * 1000) as integer))`, + updatedAt: new Date(now).toISOString(), + finishedAt: new Date(now).toISOString(), + }) + .where(and(eq(uploads.transferId, transferId), ownedBy(peerKeyId), inArray(uploads.status, [...OPEN]), isNotNull(uploads.rangeStart))) + .returning({ id: uploads.id }) + .all() + if (changed.length > 0) + this.notifyChanged() + return changed.length > 0 + } + + /** Transfers still open that started before `before`: their receipt 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 + } +} 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..0aefbca --- /dev/null +++ b/apps/backend/src/modules/uploads/uploads.service.ts @@ -0,0 +1,35 @@ +import type { UploadRecord, UploadsRepository } from './uploads.repository' +import { logger } from '../../logger' + +// A receipt that hasn't come after this long won't: the row goes `unconfirmed`. +const RECEIPT_WAIT_MS = 48 * 60 * 60 * 1000 + +/** + * Transfers peers pull from this instance. A peer creates one with + * `POST /peer/items/:itemId/downloads`, streams it from + * `/peer/downloads/:downloadId/file` as many times as it needs to resume, and + * completes it with `/peer/downloads/:downloadId/receipt`. The file itself is + * sent zero-copy, untouched; this only keeps the row. + */ +export class UploadsService { + constructor(readonly repository: UploadsRepository) {} + + /** Called from a timer, so it never throws: a failure is logged and retried next time. */ + expireUnconfirmed(now = Date.now()): number { + try { + return this.repository.expireUnconfirmed(new Date(now - RECEIPT_WAIT_MS).toISOString()) + } + catch (err) { + logger.warn({ err }, 'Could not expire uploads waiting for a receipt') + return 0 + } + } + + 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/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 + + +
+