From 89091ca9d5a2edfe56a2b0f323ac57ca442a96d5 Mon Sep 17 00:00:00 2001 From: Ben Senescu <44480372+bensenescu@users.noreply.github.com> Date: Tue, 5 May 2026 22:18:35 -0400 Subject: [PATCH] refactor: simplify rank tracking backend (#125) --- drizzle/0011_colorful_dark_beast.sql | 12 + drizzle/meta/0011_snapshot.json | 1992 +++++++++++++++++ drizzle/meta/_journal.json | 7 + src/db/app.schema.ts | 24 +- .../repositories/RankTrackingRepository.ts | 67 +- .../services/RankTrackingService.ts | 4 +- .../services/rankCheckRunGuards.ts | 265 +-- src/server/workflows/RankCheckWorkflow.ts | 46 +- 8 files changed, 2182 insertions(+), 235 deletions(-) create mode 100644 drizzle/0011_colorful_dark_beast.sql create mode 100644 drizzle/meta/0011_snapshot.json diff --git a/drizzle/0011_colorful_dark_beast.sql b/drizzle/0011_colorful_dark_beast.sql new file mode 100644 index 0000000..3b3a709 --- /dev/null +++ b/drizzle/0011_colorful_dark_beast.sql @@ -0,0 +1,12 @@ +-- Reconcile any stranded pending/running runs before creating the partial +-- unique index that replaces the rank_check_locks table. With the old +-- lock-table model, at most one active run per config could exist, so this +-- mostly protects against orphaned rows that outlived their locks. +UPDATE `rank_check_runs` +SET + `status` = 'failed', + `error_message` = COALESCE(`error_message`, 'Reconciled during lock-table migration'), + `completed_at` = COALESCE(`completed_at`, CURRENT_TIMESTAMP) +WHERE `status` IN ('pending', 'running');--> statement-breakpoint +DROP TABLE `rank_check_locks`;--> statement-breakpoint +CREATE UNIQUE INDEX `rank_check_runs_one_active_per_config_idx` ON `rank_check_runs` (`config_id`) WHERE "rank_check_runs"."status" IN ('pending', 'running'); \ No newline at end of file diff --git a/drizzle/meta/0011_snapshot.json b/drizzle/meta/0011_snapshot.json new file mode 100644 index 0000000..510b13e --- /dev/null +++ b/drizzle/meta/0011_snapshot.json @@ -0,0 +1,1992 @@ +{ + "version": "6", + "dialect": "sqlite", + "id": "23436739-8687-4515-96f9-26c79db71c35", + "prevId": "c5ffffad-fce7-44dd-9009-c9880c7cb8c0", + "tables": { + "audit_lighthouse_results": { + "name": "audit_lighthouse_results", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "audit_id": { + "name": "audit_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "page_id": { + "name": "page_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "strategy": { + "name": "strategy", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "performance_score": { + "name": "performance_score", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "accessibility_score": { + "name": "accessibility_score", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "best_practices_score": { + "name": "best_practices_score", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "seo_score": { + "name": "seo_score", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "lcp_ms": { + "name": "lcp_ms", + "type": "real", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "cls": { + "name": "cls", + "type": "real", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "inp_ms": { + "name": "inp_ms", + "type": "real", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "ttfb_ms": { + "name": "ttfb_ms", + "type": "real", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "error_message": { + "name": "error_message", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "r2_key": { + "name": "r2_key", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "payload_size_bytes": { + "name": "payload_size_bytes", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "audit_lighthouse_results_audit_id_idx": { + "name": "audit_lighthouse_results_audit_id_idx", + "columns": [ + "audit_id" + ], + "isUnique": false + } + }, + "foreignKeys": { + "audit_lighthouse_results_audit_id_audits_id_fk": { + "name": "audit_lighthouse_results_audit_id_audits_id_fk", + "tableFrom": "audit_lighthouse_results", + "tableTo": "audits", + "columnsFrom": [ + "audit_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "audit_lighthouse_results_page_id_audit_pages_id_fk": { + "name": "audit_lighthouse_results_page_id_audit_pages_id_fk", + "tableFrom": "audit_lighthouse_results", + "tableTo": "audit_pages", + "columnsFrom": [ + "page_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "audit_pages": { + "name": "audit_pages", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "audit_id": { + "name": "audit_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "url": { + "name": "url", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status_code": { + "name": "status_code", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "redirect_url": { + "name": "redirect_url", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "meta_description": { + "name": "meta_description", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "canonical_url": { + "name": "canonical_url", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "robots_meta": { + "name": "robots_meta", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "og_title": { + "name": "og_title", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "og_description": { + "name": "og_description", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "og_image": { + "name": "og_image", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "h1_count": { + "name": "h1_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "h2_count": { + "name": "h2_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "h3_count": { + "name": "h3_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "h4_count": { + "name": "h4_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "h5_count": { + "name": "h5_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "h6_count": { + "name": "h6_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "heading_order_json": { + "name": "heading_order_json", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "word_count": { + "name": "word_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "images_total": { + "name": "images_total", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "images_missing_alt": { + "name": "images_missing_alt", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "images_json": { + "name": "images_json", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "internal_link_count": { + "name": "internal_link_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "external_link_count": { + "name": "external_link_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "has_structured_data": { + "name": "has_structured_data", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": false + }, + "hreflang_tags_json": { + "name": "hreflang_tags_json", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "is_indexable": { + "name": "is_indexable", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": true + }, + "response_time_ms": { + "name": "response_time_ms", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "audit_pages_audit_id_idx": { + "name": "audit_pages_audit_id_idx", + "columns": [ + "audit_id" + ], + "isUnique": false + } + }, + "foreignKeys": { + "audit_pages_audit_id_audits_id_fk": { + "name": "audit_pages_audit_id_audits_id_fk", + "tableFrom": "audit_pages", + "tableTo": "audits", + "columnsFrom": [ + "audit_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "audits": { + "name": "audits", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "project_id": { + "name": "project_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "started_by_user_id": { + "name": "started_by_user_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "start_url": { + "name": "start_url", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'running'" + }, + "workflow_instance_id": { + "name": "workflow_instance_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "config": { + "name": "config", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'{}'" + }, + "pages_crawled": { + "name": "pages_crawled", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "pages_total": { + "name": "pages_total", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "lighthouse_total": { + "name": "lighthouse_total", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "lighthouse_completed": { + "name": "lighthouse_completed", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "lighthouse_failed": { + "name": "lighthouse_failed", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "current_phase": { + "name": "current_phase", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false, + "default": "'discovery'" + }, + "started_at": { + "name": "started_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(current_timestamp)" + }, + "completed_at": { + "name": "completed_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "audits_project_id_idx": { + "name": "audits_project_id_idx", + "columns": [ + "project_id" + ], + "isUnique": false + }, + "audits_started_by_user_id_idx": { + "name": "audits_started_by_user_id_idx", + "columns": [ + "started_by_user_id" + ], + "isUnique": false + } + }, + "foreignKeys": { + "audits_project_id_projects_id_fk": { + "name": "audits_project_id_projects_id_fk", + "tableFrom": "audits", + "tableTo": "projects", + "columnsFrom": [ + "project_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "delegated_users": { + "name": "delegated_users", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "email": { + "name": "email", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(current_timestamp)" + } + }, + "indexes": { + "delegated_users_email_unique": { + "name": "delegated_users_email_unique", + "columns": [ + "email" + ], + "isUnique": true + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "keyword_metrics": { + "name": "keyword_metrics", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "project_id": { + "name": "project_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "keyword": { + "name": "keyword", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "location_code": { + "name": "location_code", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "language_code": { + "name": "language_code", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'en'" + }, + "search_volume": { + "name": "search_volume", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "cpc": { + "name": "cpc", + "type": "real", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "competition": { + "name": "competition", + "type": "real", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "keyword_difficulty": { + "name": "keyword_difficulty", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "intent": { + "name": "intent", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "monthly_searches": { + "name": "monthly_searches", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "fetched_at": { + "name": "fetched_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(current_timestamp)" + } + }, + "indexes": { + "keyword_metrics_unique_project_keyword_location_language": { + "name": "keyword_metrics_unique_project_keyword_location_language", + "columns": [ + "project_id", + "keyword", + "location_code", + "language_code" + ], + "isUnique": true + }, + "keyword_metrics_lookup_idx": { + "name": "keyword_metrics_lookup_idx", + "columns": [ + "project_id", + "keyword", + "location_code", + "language_code", + "fetched_at" + ], + "isUnique": false + } + }, + "foreignKeys": { + "keyword_metrics_project_id_projects_id_fk": { + "name": "keyword_metrics_project_id_projects_id_fk", + "tableFrom": "keyword_metrics", + "tableTo": "projects", + "columnsFrom": [ + "project_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "projects": { + "name": "projects", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "organization_id": { + "name": "organization_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "domain": { + "name": "domain", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(current_timestamp)" + } + }, + "indexes": {}, + "foreignKeys": { + "projects_organization_id_organization_id_fk": { + "name": "projects_organization_id_organization_id_fk", + "tableFrom": "projects", + "tableTo": "organization", + "columnsFrom": [ + "organization_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "rank_check_runs": { + "name": "rank_check_runs", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "config_id": { + "name": "config_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "project_id": { + "name": "project_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'pending'" + }, + "keywords_total": { + "name": "keywords_total", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "keywords_checked": { + "name": "keywords_checked", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 0 + }, + "is_subset_run": { + "name": "is_subset_run", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": false + }, + "error_message": { + "name": "error_message", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "started_at": { + "name": "started_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(current_timestamp)" + }, + "completed_at": { + "name": "completed_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "rank_check_runs_config_idx": { + "name": "rank_check_runs_config_idx", + "columns": [ + "config_id", + "started_at" + ], + "isUnique": false + }, + "rank_check_runs_project_idx": { + "name": "rank_check_runs_project_idx", + "columns": [ + "project_id", + "started_at" + ], + "isUnique": false + }, + "rank_check_runs_one_active_per_config_idx": { + "name": "rank_check_runs_one_active_per_config_idx", + "columns": [ + "config_id" + ], + "isUnique": true, + "where": "\"rank_check_runs\".\"status\" IN ('pending', 'running')" + } + }, + "foreignKeys": { + "rank_check_runs_config_id_rank_tracking_configs_id_fk": { + "name": "rank_check_runs_config_id_rank_tracking_configs_id_fk", + "tableFrom": "rank_check_runs", + "tableTo": "rank_tracking_configs", + "columnsFrom": [ + "config_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "rank_check_runs_project_id_projects_id_fk": { + "name": "rank_check_runs_project_id_projects_id_fk", + "tableFrom": "rank_check_runs", + "tableTo": "projects", + "columnsFrom": [ + "project_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "rank_snapshots": { + "name": "rank_snapshots", + "columns": { + "id": { + "name": "id", + "type": "integer", + "primaryKey": true, + "notNull": true, + "autoincrement": true + }, + "run_id": { + "name": "run_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "tracking_keyword_id": { + "name": "tracking_keyword_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "keyword": { + "name": "keyword", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "device": { + "name": "device", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "position": { + "name": "position", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "url": { + "name": "url", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "serp_features": { + "name": "serp_features", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "checked_at": { + "name": "checked_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(current_timestamp)" + } + }, + "indexes": { + "rank_snapshots_run_idx": { + "name": "rank_snapshots_run_idx", + "columns": [ + "run_id" + ], + "isUnique": false + }, + "rank_snapshots_keyword_device_idx": { + "name": "rank_snapshots_keyword_device_idx", + "columns": [ + "tracking_keyword_id", + "device", + "checked_at" + ], + "isUnique": false + }, + "rank_snapshots_run_keyword_device_idx": { + "name": "rank_snapshots_run_keyword_device_idx", + "columns": [ + "run_id", + "tracking_keyword_id", + "device" + ], + "isUnique": true + } + }, + "foreignKeys": { + "rank_snapshots_run_id_rank_check_runs_id_fk": { + "name": "rank_snapshots_run_id_rank_check_runs_id_fk", + "tableFrom": "rank_snapshots", + "tableTo": "rank_check_runs", + "columnsFrom": [ + "run_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "rank_tracking_configs": { + "name": "rank_tracking_configs", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "project_id": { + "name": "project_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "domain": { + "name": "domain", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "location_code": { + "name": "location_code", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 2840 + }, + "language_code": { + "name": "language_code", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'en'" + }, + "devices": { + "name": "devices", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'both'" + }, + "serp_depth": { + "name": "serp_depth", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "schedule_interval": { + "name": "schedule_interval", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'weekly'" + }, + "is_active": { + "name": "is_active", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": true + }, + "last_checked_at": { + "name": "last_checked_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "next_check_at": { + "name": "next_check_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "last_skip_reason": { + "name": "last_skip_reason", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(current_timestamp)" + } + }, + "indexes": { + "rank_tracking_configs_project_domain_location_idx": { + "name": "rank_tracking_configs_project_domain_location_idx", + "columns": [ + "project_id", + "domain", + "location_code" + ], + "isUnique": true + } + }, + "foreignKeys": { + "rank_tracking_configs_project_id_projects_id_fk": { + "name": "rank_tracking_configs_project_id_projects_id_fk", + "tableFrom": "rank_tracking_configs", + "tableTo": "projects", + "columnsFrom": [ + "project_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "rank_tracking_keywords": { + "name": "rank_tracking_keywords", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "config_id": { + "name": "config_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "keyword": { + "name": "keyword", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "search_volume": { + "name": "search_volume", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "keyword_difficulty": { + "name": "keyword_difficulty", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "cpc": { + "name": "cpc", + "type": "real", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "metrics_fetched_at": { + "name": "metrics_fetched_at", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(current_timestamp)" + } + }, + "indexes": { + "rank_tracking_keywords_config_keyword_idx": { + "name": "rank_tracking_keywords_config_keyword_idx", + "columns": [ + "config_id", + "keyword" + ], + "isUnique": true + } + }, + "foreignKeys": { + "rank_tracking_keywords_config_id_rank_tracking_configs_id_fk": { + "name": "rank_tracking_keywords_config_id_rank_tracking_configs_id_fk", + "tableFrom": "rank_tracking_keywords", + "tableTo": "rank_tracking_configs", + "columnsFrom": [ + "config_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "saved_keywords": { + "name": "saved_keywords", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "project_id": { + "name": "project_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "keyword": { + "name": "keyword", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "location_code": { + "name": "location_code", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": 2840 + }, + "language_code": { + "name": "language_code", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'en'" + }, + "created_at": { + "name": "created_at", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(current_timestamp)" + } + }, + "indexes": { + "saved_keywords_unique_project_keyword_location_language": { + "name": "saved_keywords_unique_project_keyword_location_language", + "columns": [ + "project_id", + "keyword", + "location_code", + "language_code" + ], + "isUnique": true + }, + "saved_keywords_project_created_idx": { + "name": "saved_keywords_project_created_idx", + "columns": [ + "project_id", + "created_at" + ], + "isUnique": false + } + }, + "foreignKeys": { + "saved_keywords_project_id_projects_id_fk": { + "name": "saved_keywords_project_id_projects_id_fk", + "tableFrom": "saved_keywords", + "tableTo": "projects", + "columnsFrom": [ + "project_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "account": { + "name": "account", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "account_id": { + "name": "account_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "provider_id": { + "name": "provider_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "access_token": { + "name": "access_token", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "refresh_token": { + "name": "refresh_token", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "id_token": { + "name": "id_token", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "access_token_expires_at": { + "name": "access_token_expires_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "refresh_token_expires_at": { + "name": "refresh_token_expires_at", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "scope": { + "name": "scope", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "password": { + "name": "password", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(cast(unixepoch('subsecond') * 1000 as integer))" + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "account_userId_idx": { + "name": "account_userId_idx", + "columns": [ + "user_id" + ], + "isUnique": false + } + }, + "foreignKeys": { + "account_user_id_user_id_fk": { + "name": "account_user_id_user_id_fk", + "tableFrom": "account", + "tableTo": "user", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "invitation": { + "name": "invitation", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "organization_id": { + "name": "organization_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "email": { + "name": "email", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "role": { + "name": "role", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'pending'" + }, + "expires_at": { + "name": "expires_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(cast(unixepoch('subsecond') * 1000 as integer))" + }, + "inviter_id": { + "name": "inviter_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "invitation_organizationId_idx": { + "name": "invitation_organizationId_idx", + "columns": [ + "organization_id" + ], + "isUnique": false + }, + "invitation_email_idx": { + "name": "invitation_email_idx", + "columns": [ + "email" + ], + "isUnique": false + } + }, + "foreignKeys": { + "invitation_organization_id_organization_id_fk": { + "name": "invitation_organization_id_organization_id_fk", + "tableFrom": "invitation", + "tableTo": "organization", + "columnsFrom": [ + "organization_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "invitation_inviter_id_user_id_fk": { + "name": "invitation_inviter_id_user_id_fk", + "tableFrom": "invitation", + "tableTo": "user", + "columnsFrom": [ + "inviter_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "member": { + "name": "member", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "organization_id": { + "name": "organization_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "role": { + "name": "role", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "'member'" + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + } + }, + "indexes": { + "member_organizationId_idx": { + "name": "member_organizationId_idx", + "columns": [ + "organization_id" + ], + "isUnique": false + }, + "member_userId_idx": { + "name": "member_userId_idx", + "columns": [ + "user_id" + ], + "isUnique": false + } + }, + "foreignKeys": { + "member_organization_id_organization_id_fk": { + "name": "member_organization_id_organization_id_fk", + "tableFrom": "member", + "tableTo": "organization", + "columnsFrom": [ + "organization_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "member_user_id_user_id_fk": { + "name": "member_user_id_user_id_fk", + "tableFrom": "member", + "tableTo": "user", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "organization": { + "name": "organization", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "slug": { + "name": "slug", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "logo": { + "name": "logo", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "metadata": { + "name": "metadata", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "organization_slug_unique": { + "name": "organization_slug_unique", + "columns": [ + "slug" + ], + "isUnique": true + }, + "organization_slug_uidx": { + "name": "organization_slug_uidx", + "columns": [ + "slug" + ], + "isUnique": true + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "session": { + "name": "session", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "expires_at": { + "name": "expires_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "token": { + "name": "token", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(cast(unixepoch('subsecond') * 1000 as integer))" + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "ip_address": { + "name": "ip_address", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "user_agent": { + "name": "user_agent", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "active_organization_id": { + "name": "active_organization_id", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "session_token_unique": { + "name": "session_token_unique", + "columns": [ + "token" + ], + "isUnique": true + }, + "session_userId_idx": { + "name": "session_userId_idx", + "columns": [ + "user_id" + ], + "isUnique": false + } + }, + "foreignKeys": { + "session_user_id_user_id_fk": { + "name": "session_user_id_user_id_fk", + "tableFrom": "session", + "tableTo": "user", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "user": { + "name": "user", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "email": { + "name": "email", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "email_verified": { + "name": "email_verified", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": false + }, + "image": { + "name": "image", + "type": "text", + "primaryKey": false, + "notNull": false, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(cast(unixepoch('subsecond') * 1000 as integer))" + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(cast(unixepoch('subsecond') * 1000 as integer))" + }, + "analytics_opted_out": { + "name": "analytics_opted_out", + "type": "integer", + "primaryKey": false, + "notNull": false, + "autoincrement": false + } + }, + "indexes": { + "user_email_unique": { + "name": "user_email_unique", + "columns": [ + "email" + ], + "isUnique": true + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + }, + "verification": { + "name": "verification", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true, + "autoincrement": false + }, + "identifier": { + "name": "identifier", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "value": { + "name": "value", + "type": "text", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "expires_at": { + "name": "expires_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false + }, + "created_at": { + "name": "created_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(cast(unixepoch('subsecond') * 1000 as integer))" + }, + "updated_at": { + "name": "updated_at", + "type": "integer", + "primaryKey": false, + "notNull": true, + "autoincrement": false, + "default": "(cast(unixepoch('subsecond') * 1000 as integer))" + } + }, + "indexes": { + "verification_identifier_idx": { + "name": "verification_identifier_idx", + "columns": [ + "identifier" + ], + "isUnique": false + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "checkConstraints": {} + } + }, + "views": {}, + "enums": {}, + "_meta": { + "schemas": {}, + "tables": {}, + "columns": {} + }, + "internal": { + "indexes": {} + } +} \ No newline at end of file diff --git a/drizzle/meta/_journal.json b/drizzle/meta/_journal.json index 366a4e4..6c442ba 100644 --- a/drizzle/meta/_journal.json +++ b/drizzle/meta/_journal.json @@ -78,6 +78,13 @@ "when": 1777422740449, "tag": "0010_high_liz_osborn", "breakpoints": true + }, + { + "idx": 11, + "version": "6", + "when": 1778031161783, + "tag": "0011_colorful_dark_beast", + "breakpoints": true } ] } \ No newline at end of file diff --git a/src/db/app.schema.ts b/src/db/app.schema.ts index 048c649..e395011 100644 --- a/src/db/app.schema.ts +++ b/src/db/app.schema.ts @@ -168,7 +168,11 @@ export const rankTrackingKeywords = sqliteTable( ], ); -// One row per check execution (manual or scheduled) +// One row per check execution (manual or scheduled). +// A partial unique index on `config_id WHERE status IN ('pending','running')` +// enforces at most one in-flight run per config at the DB level, which is how +// duplicate-trigger protection is implemented — INSERT of a second pending run +// for the same config fails with a unique-constraint violation. export const rankCheckRuns = sqliteTable( "rank_check_runs", { @@ -198,24 +202,12 @@ export const rankCheckRuns = sqliteTable( (table) => [ index("rank_check_runs_config_idx").on(table.configId, table.startedAt), index("rank_check_runs_project_idx").on(table.projectId, table.startedAt), + uniqueIndex("rank_check_runs_one_active_per_config_idx") + .on(table.configId) + .where(sql`${table.status} IN ('pending', 'running')`), ], ); -// One active lock per rank tracking config to prevent overlapping runs -export const rankCheckLocks = sqliteTable( - "rank_check_locks", - { - configId: text("config_id") - .primaryKey() - .references(() => rankTrackingConfigs.id, { onDelete: "cascade" }), - runId: text("run_id").notNull(), - acquiredAt: text("acquired_at") - .notNull() - .default(sql`(current_timestamp)`), - }, - (table) => [index("rank_check_locks_run_idx").on(table.runId)], -); - // One row per keyword per device per check run export const rankSnapshots = sqliteTable( "rank_snapshots", diff --git a/src/server/features/rank-tracking/repositories/RankTrackingRepository.ts b/src/server/features/rank-tracking/repositories/RankTrackingRepository.ts index fe7ed60..5367790 100644 --- a/src/server/features/rank-tracking/repositories/RankTrackingRepository.ts +++ b/src/server/features/rank-tracking/repositories/RankTrackingRepository.ts @@ -4,7 +4,6 @@ import { db } from "@/db"; import { rankTrackingConfigs, rankCheckRuns, - rankCheckLocks, rankSnapshots, rankTrackingKeywords, projects, @@ -137,17 +136,27 @@ async function getDueConfigsWithOrganization(nowIso: string) { // Run CRUD // --------------------------------------------------------------------------- -async function createRun(data: { +/** + * Try to insert a new pending run. Returns true if inserted, false if blocked + * by the partial unique index on (config_id) WHERE status IN ('pending', + * 'running') — i.e. another active run exists for this config. + * + * This is how duplicate-trigger protection is enforced: the DB rejects the + * second insert rather than a separate lock table. + */ +async function tryCreateRun(data: { id: string; configId: string; projectId: string; keywordsTotal: number; isSubsetRun?: boolean; -}) { - await db.insert(rankCheckRuns).values({ - ...data, - status: "pending", - }); +}): Promise { + const inserted = await db + .insert(rankCheckRuns) + .values({ ...data, status: "pending" }) + .onConflictDoNothing() + .returning({ id: rankCheckRuns.id }); + return inserted.length > 0; } async function updateRun( @@ -176,38 +185,24 @@ async function getLatestRunForConfig(configId: string) { return rows[0] ?? null; } -async function tryCreateRunLock(configId: string, runId: string) { - const inserted = await db - .insert(rankCheckLocks) - .values({ configId, runId }) - .onConflictDoNothing({ target: rankCheckLocks.configId }) - .returning({ runId: rankCheckLocks.runId }); - - return inserted.length > 0; -} - -async function getRunLock(configId: string) { +/** + * Returns the currently active (pending or running) run for a config, if any. + * At most one such row exists, enforced by the partial unique index. + */ +async function getActiveRunForConfig(configId: string) { const rows = await db .select() - .from(rankCheckLocks) - .where(eq(rankCheckLocks.configId, configId)) + .from(rankCheckRuns) + .where( + and( + eq(rankCheckRuns.configId, configId), + inArray(rankCheckRuns.status, ["pending", "running"]), + ), + ) .limit(1); return rows[0] ?? null; } -async function deleteRunLock(configId: string, runId?: string) { - await db - .delete(rankCheckLocks) - .where( - runId - ? and( - eq(rankCheckLocks.configId, configId), - eq(rankCheckLocks.runId, runId), - ) - : eq(rankCheckLocks.configId, configId), - ); -} - // --------------------------------------------------------------------------- // Snapshots // --------------------------------------------------------------------------- @@ -369,13 +364,11 @@ export const RankTrackingRepository = { createConfig, updateConfig, getDueConfigsWithOrganization, - createRun, + tryCreateRun, updateRun, getRunById, getLatestRunForConfig, - tryCreateRunLock, - getRunLock, - deleteRunLock, + getActiveRunForConfig, insertSnapshots, getSnapshotsForRun, getKeywordsForConfig, diff --git a/src/server/features/rank-tracking/services/RankTrackingService.ts b/src/server/features/rank-tracking/services/RankTrackingService.ts index 7c05b9a..a30fc9e 100644 --- a/src/server/features/rank-tracking/services/RankTrackingService.ts +++ b/src/server/features/rank-tracking/services/RankTrackingService.ts @@ -218,8 +218,8 @@ async function getLatestRun(configId: string, projectId: string) { if (!run) return null; // If the DB says the run is still active, check the workflow instance. - // We only report staleness here — the cron handler cleans up stale locks - // when it next tries to acquire one (via cleanupStaleLock). Mutating from + // We only report staleness here — the next call to beginRankCheckRun will + // mark a stale blocker as failed before retrying its insert. Mutating from // this read path caused a race where the original workflow kept running // while a replacement was started. const reconciliation = await reconcileActiveRankCheckRun(run); diff --git a/src/server/features/rank-tracking/services/rankCheckRunGuards.ts b/src/server/features/rank-tracking/services/rankCheckRunGuards.ts index a2c795a..a265b3f 100644 --- a/src/server/features/rank-tracking/services/rankCheckRunGuards.ts +++ b/src/server/features/rank-tracking/services/rankCheckRunGuards.ts @@ -7,16 +7,16 @@ import type { } from "@/types/schemas/rank-tracking"; type RunRow = Awaited>; -type RunLockRow = Awaited>; -// Coordination invariants for rank checks: -// - `workflow id === run id`, so the workflow instance is the authoritative -// runtime identity for a stored run. -// - `rank_check_locks` enforces at most one active runner per config. -// - Only the lock owner is allowed to spend credits, write snapshots, or -// finalize the run. -// - Missing/unknown workflow state is tolerated briefly during startup before -// we treat the run as stale and repair it. +// Coordination model: +// - workflow id === run id (workflow instance is the authoritative runtime). +// - A partial unique index on rank_check_runs(config_id) WHERE status IN +// ('pending','running') enforces at most one active run per config at the +// DB level. A failed INSERT *is* the "already running" signal — no +// separate lock table is needed. +// - Flipping status to 'completed'/'failed' is what frees the slot. +// - Missing/unknown workflow state is tolerated briefly during startup +// before we treat a run as stale and mark it failed. type RankCheckWorkflowStatus = { status: @@ -77,85 +77,11 @@ function getStaleReason( return workflowStatus.error?.message ?? `Workflow ${workflowStatus.status}`; } if (workflowStatus.status === "complete") { - return "Workflow completed without releasing the run lock"; + return "Workflow completed without finalizing the run"; } return `Workflow is no longer active (${workflowStatus.status})`; } -export async function beginRankCheckRun(input: { - workflow: Env["RANK_CHECK_WORKFLOW"]; - config: RankCheckConfigForStart; - projectId: string; - billingCustomer: BillingCustomerContext; - keywordsTotal: number; - keywordIds?: string[]; - trigger: "manual" | "scheduled"; - workflowStartErrorMessage: string; -}): Promise { - const runId = crypto.randomUUID(); - const lockResult = await acquireRankCheckRunLock(input.config.id, runId); - if (!lockResult.acquired) { - return { - ok: false, - reason: "already_running", - blockingRunId: lockResult.blockingRunId, - }; - } - - try { - await RankTrackingRepository.createRun({ - id: runId, - configId: input.config.id, - projectId: input.projectId, - keywordsTotal: input.keywordsTotal, - isSubsetRun: (input.keywordIds?.length ?? 0) > 0, - }); - - await input.workflow.create({ - id: runId, - params: { - runId, - configId: input.config.id, - billingCustomer: { - userId: input.billingCustomer.userId, - userEmail: input.billingCustomer.userEmail, - organizationId: input.billingCustomer.organizationId, - projectId: input.billingCustomer.projectId, - }, - projectId: input.projectId, - domain: input.config.domain, - locationCode: input.config.locationCode, - languageCode: input.config.languageCode, - devices: input.config.devices, - serpDepth: input.config.serpDepth, - trigger: input.trigger, - keywordIds: input.keywordIds, - }, - }); - } catch (error) { - try { - await failRunAndReleaseRankCheckLock( - input.config.id, - runId, - input.workflowStartErrorMessage, - ); - } catch { - await releaseRankCheckRunLock(input.config.id, runId); - } - - try { - const instance = await input.workflow.get(runId); - await instance.terminate(); - } catch { - // Workflow may not have been created - } - - throw error; - } - - return { ok: true, runId }; -} - async function getStaleRankCheckRunReason(input: { run: RunRow; runId: string; @@ -181,8 +107,23 @@ async function getStaleRankCheckRunReason(input: { return getStaleReason(workflowStatus, input.run); } -async function failRunIfNeeded(runId: string, reason: string, run: RunRow) { - if (!run || run.status === "completed" || run.status === "failed") return; +/** + * Mark a run as failed if it's still in an active state. Idempotent — safe to + * call on runs that are already completed/failed. + */ +export async function failRunIfActive( + runId: string, + reason: string, + run?: RunRow, +) { + const current = run ?? (await RankTrackingRepository.getRunById(runId)); + if ( + !current || + current.status === "completed" || + current.status === "failed" + ) { + return; + } await RankTrackingRepository.updateRun(runId, { status: "failed", errorMessage: reason, @@ -190,23 +131,95 @@ async function failRunIfNeeded(runId: string, reason: string, run: RunRow) { }); } -async function cleanupStaleLock(lock: NonNullable) { - const run = await RankTrackingRepository.getRunById(lock.runId); - if (run?.status === "completed" || run?.status === "failed") { - await RankTrackingRepository.deleteRunLock(lock.configId, lock.runId); - return true; +export async function beginRankCheckRun(input: { + workflow: Env["RANK_CHECK_WORKFLOW"]; + config: RankCheckConfigForStart; + projectId: string; + billingCustomer: BillingCustomerContext; + keywordsTotal: number; + keywordIds?: string[]; + trigger: "manual" | "scheduled"; + workflowStartErrorMessage: string; +}): Promise { + // At most two attempts: once normally, once after clearing a stale blocker. + for (let attempt = 0; attempt < 2; attempt++) { + const runId = crypto.randomUUID(); + const inserted = await RankTrackingRepository.tryCreateRun({ + id: runId, + configId: input.config.id, + projectId: input.projectId, + keywordsTotal: input.keywordsTotal, + isSubsetRun: (input.keywordIds?.length ?? 0) > 0, + }); + + if (inserted) { + try { + await input.workflow.create({ + id: runId, + params: { + runId, + configId: input.config.id, + billingCustomer: input.billingCustomer, + projectId: input.projectId, + domain: input.config.domain, + locationCode: input.config.locationCode, + languageCode: input.config.languageCode, + devices: input.config.devices, + serpDepth: input.config.serpDepth, + trigger: input.trigger, + keywordIds: input.keywordIds, + }, + }); + } catch (error) { + // Workflow couldn't start — flip the run to failed so the + // partial-index slot is released. Best-effort cleanup of any + // zombie instance. + await failRunIfActive(runId, input.workflowStartErrorMessage); + try { + const instance = await input.workflow.get(runId); + await instance.terminate(); + } catch { + // Workflow may not have been created. + } + throw error; + } + return { ok: true, runId }; + } + + // INSERT was blocked by the partial unique index — another active run + // exists. Inspect it to decide whether to retry or return already_running. + const blocker = await RankTrackingRepository.getActiveRunForConfig( + input.config.id, + ); + if (!blocker) { + // Raced: blocker's status flipped between insert and select. Loop. + continue; + } + + if (attempt === 0) { + const staleReason = await getStaleRankCheckRunReason({ + run: blocker, + runId: blocker.id, + ageMs: Date.now() - new Date(blocker.startedAt).getTime(), + }); + if (staleReason) { + await failRunIfActive(blocker.id, staleReason, blocker); + continue; // slot is free now — retry insert + } + } + + return { ok: false, reason: "already_running", blockingRunId: blocker.id }; } - const staleReason = await getStaleRankCheckRunReason({ - runId: lock.runId, - run, - ageMs: Date.now() - new Date(lock.acquiredAt).getTime(), - }); - if (!staleReason) return false; - - await failRunIfNeeded(lock.runId, staleReason, run); - await RankTrackingRepository.deleteRunLock(lock.configId, lock.runId); - return true; + // Exhausted attempts (rapid churn). Report whatever's blocking now. + const final = await RankTrackingRepository.getActiveRunForConfig( + input.config.id, + ); + return { + ok: false, + reason: "already_running", + blockingRunId: final?.id ?? null, + }; } export async function reconcileActiveRankCheckRun(run: NonNullable) { @@ -226,53 +239,3 @@ export async function reconcileActiveRankCheckRun(run: NonNullable) { completedAt: new Date().toISOString(), }; } - -async function acquireRankCheckRunLock(configId: string, runId: string) { - // Optimistic: try to grab the lock immediately - if (await RankTrackingRepository.tryCreateRunLock(configId, runId)) { - return { acquired: true as const }; - } - - // Lock exists — check if it's stale and can be cleaned up - const existingLock = await RankTrackingRepository.getRunLock(configId); - if (!existingLock) { - // Lock was released between our insert and select — retry once - if (await RankTrackingRepository.tryCreateRunLock(configId, runId)) { - return { acquired: true as const }; - } - const blocker = await RankTrackingRepository.getRunLock(configId); - return { acquired: false as const, blockingRunId: blocker?.runId ?? null }; - } - - const cleaned = await cleanupStaleLock(existingLock); - if (!cleaned) { - return { acquired: false as const, blockingRunId: existingLock.runId }; - } - - // Stale lock cleaned — retry once - if (await RankTrackingRepository.tryCreateRunLock(configId, runId)) { - return { acquired: true as const }; - } - - const blocker = await RankTrackingRepository.getRunLock(configId); - return { acquired: false as const, blockingRunId: blocker?.runId ?? null }; -} - -export async function runOwnsRankCheckLock(configId: string, runId: string) { - const lock = await RankTrackingRepository.getRunLock(configId); - return lock?.runId === runId; -} - -export async function releaseRankCheckRunLock(configId: string, runId: string) { - await RankTrackingRepository.deleteRunLock(configId, runId); -} - -export async function failRunAndReleaseRankCheckLock( - configId: string, - runId: string, - errorMessage: string, -) { - const run = await RankTrackingRepository.getRunById(runId); - await failRunIfNeeded(runId, errorMessage, run); - await RankTrackingRepository.deleteRunLock(configId, runId); -} diff --git a/src/server/workflows/RankCheckWorkflow.ts b/src/server/workflows/RankCheckWorkflow.ts index 3a1092a..7520e8b 100644 --- a/src/server/workflows/RankCheckWorkflow.ts +++ b/src/server/workflows/RankCheckWorkflow.ts @@ -6,11 +6,7 @@ import { import { NonRetryableError } from "cloudflare:workflows"; import type { BillingCustomerContext } from "@/server/billing/subscription"; import { RankTrackingRepository } from "@/server/features/rank-tracking/repositories/RankTrackingRepository"; -import { - failRunAndReleaseRankCheckLock, - releaseRankCheckRunLock, - runOwnsRankCheckLock, -} from "@/server/features/rank-tracking/services/rankCheckRunGuards"; +import { failRunIfActive } from "@/server/features/rank-tracking/services/rankCheckRunGuards"; import { runLiveCheck } from "@/server/workflows/rankCheckPaths"; import { createDataforseoClient } from "@/server/lib/dataforseoClient"; import { captureServerEvent } from "@/server/lib/posthog"; @@ -50,10 +46,12 @@ async function prepareRankCheckKeywords(input: { serpDepth: number; keywordIds?: string[]; }) { - const ownsLock = await runOwnsRankCheckLock(input.configId, input.runId); - if (!ownsLock) { + // If stale-cleanup marked our run failed before we got here, bail out + // rather than resurrecting a superseded run. + const run = await RankTrackingRepository.getRunById(input.runId); + if (!run || run.status === "failed" || run.status === "completed") { throw new NonRetryableError( - `Rank check lock is not held by run ${input.runId}`, + `Run ${input.runId} is no longer active (status=${run?.status ?? "missing"})`, ); } @@ -122,13 +120,13 @@ async function finalizeRankCheckRun(input: { trigger: RankCheckParams["trigger"]; batchError: string | null; }) { - // Re-check lock ownership before finalizing. If the lock was stolen - // (stale cleanup raced with a slow workflow), bail out to avoid - // overwriting the replacement run's state. - const ownsLock = await runOwnsRankCheckLock(input.configId, input.runId); - if (!ownsLock) { + // If stale-cleanup already marked our run failed, don't overwrite that + // decision with a completed status — a replacement run may already be + // underway. + const run = await RankTrackingRepository.getRunById(input.runId); + if (!run || run.status === "failed" || run.status === "completed") { console.warn( - `[rank-check] ${input.runId} lost lock ownership, skipping finalization`, + `[rank-check] ${input.runId} no longer active (status=${run?.status ?? "missing"}), skipping finalization`, ); return; } @@ -143,9 +141,7 @@ async function finalizeRankCheckRun(input: { const keywordsChecked = new Set(snapshots.map((s) => s.trackingKeywordId)) .size; - // Derive incompleteCount from the run's keywordsTotal (set in prepare step) - const run = await RankTrackingRepository.getRunById(input.runId); - const keywordsTotal = run?.keywordsTotal ?? keywordsChecked; + const keywordsTotal = run.keywordsTotal || keywordsChecked; const incompleteCount = keywordsTotal - keywordsChecked; let errorMessage: string | undefined; @@ -155,6 +151,8 @@ async function finalizeRankCheckRun(input: { errorMessage = `${incompleteCount} keyword(s) could not be checked`; } + // Flipping status away from 'pending'/'running' is what releases the + // partial-index slot for the next run. await RankTrackingRepository.updateRun(input.runId, { status: "completed", keywordsChecked, @@ -170,8 +168,6 @@ async function finalizeRankCheckRun(input: { lastSkipReason: null, }); - await releaseRankCheckRunLock(input.configId, input.runId); - await captureServerEvent({ distinctId: input.billingCustomer.userId, event: "rank_tracking:check_complete", @@ -194,11 +190,7 @@ async function markRankCheckRunFailed(input: { }) { const errorMessage = input.error instanceof Error ? input.error.message : "Unknown error"; - await failRunAndReleaseRankCheckLock( - input.configId, - input.runId, - errorMessage, - ); + await failRunIfActive(input.runId, errorMessage); // Flag the config so the UI can show why the scheduled check was skipped const isInsufficientCredits = @@ -256,11 +248,7 @@ export class RankCheckWorkflow extends WorkflowEntrypoint< }, ); if (!configCheck.isActive) { - await failRunAndReleaseRankCheckLock( - configId, - runId, - "Config has been archived", - ); + await failRunIfActive(runId, "Config has been archived"); return; }