diff --git a/Dockerfile b/Dockerfile index 4685919b..baa5c7ef 100644 --- a/Dockerfile +++ b/Dockerfile @@ -180,9 +180,9 @@ RUN echo "πŸ‘€ Creating non-root user..." && \ useradd -u 1001 -r -g livereview -d /app -s /sbin/nologin livereview && \ echo "User 'livereview' created successfully" -# Create directories +# Create directories (including lrdata/blobs for local blob storage) RUN echo "πŸ“ Creating application directories..." && \ - mkdir -p /app/db/migrations /app/data /app/logs && \ + mkdir -p /app/db/migrations /app/data /app/logs /app/lrdata/blobs && \ chown -R livereview:livereview /app && \ echo "Directories created and permissions set" diff --git a/Makefile b/Makefile index 4be821bb..d7586109 100644 --- a/Makefile +++ b/Makefile @@ -81,6 +81,9 @@ docker-local-rebuild: docker-local-stop: docker compose down +# Python executable mapping (works out-of-the-box on Ubuntu where python3 is present but python is not) +PYTHON ?= $(shell command -v python3 2>/dev/null || echo python) + # Go parameters GOENV=env -u GOROOT GOCMD=$(GOENV) go @@ -151,48 +154,48 @@ email-preview: # Version management targets version: - @python scripts/lrops.py version + @$(PYTHON) scripts/lrops.py version version-bump: - @python scripts/lrops.py bump $(ARGS) + @$(PYTHON) scripts/lrops.py bump $(ARGS) version-patch: - @python scripts/lrops.py bump --type patch $(ARGS) + @$(PYTHON) scripts/lrops.py bump --type patch $(ARGS) version-minor: - @python scripts/lrops.py bump --type minor $(ARGS) + @$(PYTHON) scripts/lrops.py bump --type minor $(ARGS) version-major: - @python scripts/lrops.py bump --type major $(ARGS) + @$(PYTHON) scripts/lrops.py bump --type major $(ARGS) # Version management targets that allow dirty working directory version-bump-dirty: - @python scripts/lrops.py bump --allow-dirty + @$(PYTHON) scripts/lrops.py bump --allow-dirty version-patch-dirty: - @python scripts/lrops.py bump --type patch --allow-dirty + @$(PYTHON) scripts/lrops.py bump --type patch --allow-dirty version-minor-dirty: - @python scripts/lrops.py bump --type minor --allow-dirty + @$(PYTHON) scripts/lrops.py bump --type minor --allow-dirty version-major-dirty: - @python scripts/lrops.py bump --type major --allow-dirty + @$(PYTHON) scripts/lrops.py bump --type major --allow-dirty # Dry-run version targets version-bump-dry: - @python scripts/lrops.py bump --dry-run --allow-dirty + @$(PYTHON) scripts/lrops.py bump --dry-run --allow-dirty version-patch-dry: - @python scripts/lrops.py bump --type patch --dry-run --allow-dirty + @$(PYTHON) scripts/lrops.py bump --type patch --dry-run --allow-dirty version-minor-dry: - @python scripts/lrops.py bump --type minor --dry-run --allow-dirty + @$(PYTHON) scripts/lrops.py bump --type minor --dry-run --allow-dirty version-major-dry: - @python scripts/lrops.py bump --type major --dry-run --allow-dirty + @$(PYTHON) scripts/lrops.py bump --type major --dry-run --allow-dirty build-versioned: - @python scripts/lrops.py build + @$(PYTHON) scripts/lrops.py build # ============================================================================ # Frozen DOCKER DEPENDENCY versions (docker/docker-deps.env) @@ -222,15 +225,15 @@ build-versioned: # shows up in the report when it falls behind, it's just never auto-applied # by update-docker-deps/update-docker-deps-yes or the pre-build check. # Override for one run with: -# python3 scripts/check_docker_deps.py --include-pinned [--yes] +# $(PYTHON) scripts/check_docker_deps.py --include-pinned [--yes] check-docker-deps: - @python3 scripts/check_docker_deps.py --check + @$(PYTHON) scripts/check_docker_deps.py --check update-docker-deps: - @python3 scripts/check_docker_deps.py + @$(PYTHON) scripts/check_docker_deps.py update-docker-deps-yes: - @python3 scripts/check_docker_deps.py --yes + @$(PYTHON) scripts/check_docker_deps.py --yes # Smoke-test that every pinned Docker dependency binary is actually present # and invokable INSIDE a built image (dbmate, river, riverui, vl-convert, @@ -257,7 +260,7 @@ verify-docker-deps: # 8. Interactive confirmation prompt before build execution # Files: scripts/lrops.py (lines 634-826), Dockerfile (multi-stage), ui/package.json docker-build: - @python scripts/lrops.py build --docker $(ARGS) + @$(PYTHON) scripts/lrops.py build --docker $(ARGS) # DOCKER-BUILD-PUSH: Same as docker-build but automatically pushes to registry # Implementation: scripts/lrops.py:cmd_build() with push=True flag @@ -270,21 +273,21 @@ docker-build: # Registry: Configurable via --registry, defaults to GitLab Container Registry # Tags: /: and optionally /:latest docker-build-push: - @python scripts/lrops.py build --docker --push $(ARGS) + @$(PYTHON) scripts/lrops.py build --docker --push $(ARGS) # Interactive Docker build with tag selection docker-interactive: - @python scripts/lrops.py docker + @$(PYTHON) scripts/lrops.py docker docker-interactive-push: - @python scripts/lrops.py docker --push $(ARGS) + @$(PYTHON) scripts/lrops.py docker --push $(ARGS) # Dry-run Docker targets docker-build-dry: - @python scripts/lrops.py build --docker --dry-run $(ARGS) + @$(PYTHON) scripts/lrops.py build --docker --dry-run $(ARGS) docker-interactive-dry: - @python scripts/lrops.py docker --dry-run + @$(PYTHON) scripts/lrops.py docker --dry-run # Legacy build-push for backward compatibility (now uses versioning) build-push: docker-build-push @@ -501,7 +504,7 @@ security-gh-secret-scanning: # Regenerate machine-readable and markdown triage artifacts from the latest OSV report. security-triage: security-osv - @python3 scripts/extract_osv_report.py \ + @$(PYTHON) scripts/extract_osv_report.py \ --input security_issues/osv-scanner-latest.json \ --csv security_issues/osv-triage-latest.csv \ --md security_issues/osv-triage-latest.md @@ -636,7 +639,7 @@ sync-docs-sources: # Exits 1 if anything is behind - usable in CI, or just run # `make sync-docs-sources` to actually pull the update in. check-docs-sources: - @python3 scripts/docindex/check_docs_sources.py + @$(PYTHON) scripts/docindex/check_docs_sources.py # Generate a token-compact schema dump of the prod DB (public schema) for LLM context. .PHONY: compressed-schema @@ -646,7 +649,7 @@ compressed-schema: exit 1; \ fi @mkdir -p db - @set -a && . ./.env.prod && set +a && python3 scripts/llm-schema.py db/schema-compressed.txt + @set -a && . ./.env.prod && set +a && $(PYTHON) scripts/llm-schema.py db/schema-compressed.txt @echo "βœ… Wrote db/schema-compressed.txt" # Export a full snapshot of the prod DB (schema + data) using .env.prod's @@ -758,10 +761,10 @@ ghcr-login: docker-context-setup # Multi-architecture Docker build targets docker-multiarch: - @python scripts/lrops.py build --docker --multiarch $(ARGS) + @$(PYTHON) scripts/lrops.py build --docker --multiarch $(ARGS) docker-multiarch-push: - @python scripts/lrops.py build --docker --multiarch --push $(ARGS) + @$(PYTHON) scripts/lrops.py build --docker --multiarch --push $(ARGS) @echo "ℹ️ Optional GitHub release publish: make release-gh" @echo " Optional explicit override: make release-gh VERSION=$$(git describe --tags --abbrev=0 2>/dev/null || true)" @@ -771,41 +774,41 @@ release-gh: @python3 $(RELEASE_GH_SCRIPT) --repo $(GH_REPO) $(if $(VERSION),--version $(VERSION),) docker-multiarch-dry: - @python scripts/lrops.py build --docker --multiarch --dry-run $(ARGS) + @$(PYTHON) scripts/lrops.py build --docker --multiarch --dry-run $(ARGS) # Vendor multi-arch dry run (Phase 9 validation) vendor-docker-multiarch-dry: - @python scripts/lrops.py build --docker --multiarch --dry-run --vendor-prompts $(ARGS) + @$(PYTHON) scripts/lrops.py build --docker --multiarch --dry-run --vendor-prompts $(ARGS) # Vendor single-arch builds vendor-docker-build-dry: - @python scripts/lrops.py build --docker --dry-run --vendor-prompts $(ARGS) + @$(PYTHON) scripts/lrops.py build --docker --dry-run --vendor-prompts $(ARGS) vendor-docker-build: - @python scripts/lrops.py build --docker --vendor-prompts $(ARGS) + @$(PYTHON) scripts/lrops.py build --docker --vendor-prompts $(ARGS) vendor-docker-build-push: - @python scripts/lrops.py build --docker --push --vendor-prompts $(ARGS) + @$(PYTHON) scripts/lrops.py build --docker --push --vendor-prompts $(ARGS) # Vendor multi-arch push (with optional latest tagging via ARGS="--latest") vendor-docker-multiarch-push: - @python scripts/lrops.py build --docker --multiarch --push --vendor-prompts $(ARGS) + @$(PYTHON) scripts/lrops.py build --docker --multiarch --push --vendor-prompts $(ARGS) # Cross-compilation Docker build targets (faster ARM builds) docker-multiarch-cross: @echo "πŸš€ Building multi-arch images using cross-compilation for faster ARM builds" - @python scripts/lrops.py build --docker --multiarch $(ARGS) + @$(PYTHON) scripts/lrops.py build --docker --multiarch $(ARGS) docker-multiarch-cross-push: @echo "πŸš€ Building and pushing multi-arch images using cross-compilation" - @python scripts/lrops.py build --docker --multiarch --push $(ARGS) + @$(PYTHON) scripts/lrops.py build --docker --multiarch --push $(ARGS) # Interactive multi-architecture Docker build docker-interactive-multiarch: - @python scripts/lrops.py docker --multiarch + @$(PYTHON) scripts/lrops.py docker --multiarch docker-interactive-multiarch-push: - @python scripts/lrops.py docker --multiarch --push + @$(PYTHON) scripts/lrops.py docker --multiarch --push cplrops: @cp lrops.sh ../gh/LiveReview/ @@ -1121,7 +1124,7 @@ docs/openapi.yaml internal/api/docs/spec.go: $(API_SPEC_INPUTS) typed-install @chmod 755 docs internal/api/docs @PATH="$(TYPED_BIN_DIR):$$PATH" typed -config config/typed.yaml > /tmp/lr_typed_build.log 2>&1 || (echo "❌ Typed generation failed. Logs:" && cat /tmp/lr_typed_build.log && exit 1) @$(GOCMD) run internal/api/docs/spec.go > /tmp/lr_spec_build.log 2>&1 || (echo "❌ OpenAPI spec generation failed. Logs:" && cat /tmp/lr_spec_build.log && exit 1) - @python3 scripts/openapi/fix-openapi-spec.py docs/openapi.yaml + @$(PYTHON) scripts/openapi/fix-openapi-spec.py docs/openapi.yaml generate-openapi: docs/openapi.yaml @@ -1503,7 +1506,7 @@ razorpay-webhook-ensure: fi @MODE_VALUE="$(MODE)"; \ if [ -z "$$MODE_VALUE" ]; then MODE_VALUE="$${RAZORPAY_MODE:-live}"; fi; \ - python3 scripts/razorpay_webhook_ensure.py --base-url "$(BASE_URL)" --mode "$$MODE_VALUE" $(ARGS) + $(PYTHON) scripts/razorpay_webhook_ensure.py --base-url "$(BASE_URL)" --mode "$$MODE_VALUE" $(ARGS) razorpay-webhook-ensure-dry: @if [ -z "$(BASE_URL)" ]; then \ @@ -1512,7 +1515,7 @@ razorpay-webhook-ensure-dry: fi @MODE_VALUE="$(MODE)"; \ if [ -z "$$MODE_VALUE" ]; then MODE_VALUE="$${RAZORPAY_MODE:-live}"; fi; \ - python3 scripts/razorpay_webhook_ensure.py --base-url "$(BASE_URL)" --mode "$$MODE_VALUE" --dry-run $(ARGS) + $(PYTHON) scripts/razorpay_webhook_ensure.py --base-url "$(BASE_URL)" --mode "$$MODE_VALUE" --dry-run $(ARGS) razorpay-verify-plans: @bash ./scripts/verify-razorpay-plans.sh $(DEPLOY_ACTUAL_ENV_FILE) diff --git a/config/osv-scanner.toml b/config/osv-scanner.toml index 0e3a98c2..1569e148 100644 --- a/config/osv-scanner.toml +++ b/config/osv-scanner.toml @@ -3,3 +3,6 @@ id = "GO-2026-5932" reason = "False positive β€” golang.org/x/crypto/openpgp sub-package is never imported by this project; only golang.org/x/crypto/bcrypt is used." +[[ignoredVulns]] +id = "GO-2026-6452" +reason = "TODO: Once OSV DB fixes the missing 'fixed' semver range, this should be removed. We are on patched v2.11.0, and the vulnerable parsing path is never called because we only write spreadsheets, never read them." diff --git a/docker-entrypoint.sh b/docker-entrypoint.sh index 43afbbd7..1fac03be 100644 --- a/docker-entrypoint.sh +++ b/docker-entrypoint.sh @@ -5,7 +5,9 @@ echo "πŸš€ Starting LiveReview application..." # Ensure blob storage directory exists (runs as root) mkdir -p /app/lrdata/blobs -chown -R livereview:livereview /app/lrdata/blobs +# Local dev only: chown may fail under Docker Desktop / rootless Docker / seccomp. +# Production (lrops.sh) runs standard Docker where this always succeeds β€” no || true needed there. +chown -R livereview:livereview /app/lrdata/blobs 2>/dev/null || true # Function to wait for PostgreSQL to be ready wait_for_postgres() { diff --git a/docs/architecture/diff_storage_offloading.md b/docs/architecture/diff_storage_offloading.md new file mode 100644 index 00000000..d65a037b --- /dev/null +++ b/docs/architecture/diff_storage_offloading.md @@ -0,0 +1,202 @@ +# Database Storage Optimization and Diff Storage Offloading Architecture + +## Problem + +### Problem 1: Code Diff Storage Bloat in reviews.metadata + +PostgreSQL stores review records in the reviews table. The metadata column of the reviews table contains a JSON field named preloaded_changes. The preloaded_changes field stores raw Git code diff strings. The preloaded_changes field used 791.30 megabytes in PostgreSQL. This field used 90.4 percent of the metadata column storage. Unchecked accumulation of raw source code diffs in preloaded_changes creates PostgreSQL TOAST table bloat. It increases database backup dump sizes by 150 megabytes. It degrades database query performance. + +### Problem 2: Background Job Record Accumulation in river_job + +Completed background job records in the river_job table accumulated over time. They consumed 263.5 megabytes in PostgreSQL. They added 79.09 megabytes to database backup files because the job queue had a 365 day retention configuration. + +```mermaid +flowchart TD + A["Review Creation & Job Execution"] --> B["Problem 1: Raw diffs in reviews.metadata JSONB (791 MB / 90.4%)"] + A --> C["Problem 2: River job retention for 365 days (263 MB)"] + B --> D["PostgreSQL TOAST Table Bloat (791 MB)"] + C --> E["River Job Table Bloat (263 MB)"] + D --> F["Large Database Backups & High I/O"] + E --> F +``` + +## Solution + +The system combines database reads for recent reviews with low PostgreSQL disk usage. The system uses a 30 day database retention window and an automated background offloading cron job. + +### Key Principles + +1. Newly generated code diffs remain stored in PostgreSQL for the first 30 days. This ensures that recent code reviews get high read speed and low query latency. +2. A background scheduled cron manager named PreloadedChangesArchivalManager periodically scans PostgreSQL. It finds reviews older than 30 days that still hold preloaded_changes in database metadata. +3. For reviews older than 30 days, the cron job uploads the raw code diffs to Blob Storage under the key path org/org_id/review/review_id/artifacts/preloaded_changes.json. When the upload succeeds, the job removes preloaded_changes from reviews.metadata in PostgreSQL. +4. The background manager operates with controlled batch sizes of 50 reviews per batch. It pauses for 50 milliseconds between batches. It uses streaming JSON serialization. This prevents CPU spikes, memory bloat, and database connection pool exhaustion. +5. The API server inspects PostgreSQL metadata first for active diffs. If preloaded_changes is absent from metadata, the server reads the diff payload from Blob Storage. + +## Component Architecture + +The architecture contains five primary components. + +1. Review Worker Creation +2. Automated Background Offloading Cron +3. Blob Storage Transfer and PostgreSQL Pruning +4. API Handler Fallback Strategy +5. River Job Queue Retention Strategy + +### Review Worker Creation + +When DiffReviewWorker executes a review job, it writes raw code diffs to the metadata column of the reviews table in PostgreSQL. During the first 30 days, the system does not make external storage calls when users display reviews in the Web UI or CLI. + +```mermaid +flowchart TD + A["Diff Review Job Queue"] --> B["DiffReviewWorker Process"] + B --> C["Parse CodeDiff Payload"] + C --> D["Store preloaded_changes in reviews.metadata JSONB"] + D --> E["Save Review to PostgreSQL"] + E --> F["Fast DB Reads for 30 Days"] +``` + +### Automated Background Offloading Cron + +The PreloadedChangesArchivalManager background scheduler runs at scheduled cron intervals. The default schedule runs daily at 2:30 AM IST. + +The system loads configuration from system_settings (`preloaded_changes_archival_settings`). You can update configuration without restarting the server. + +The enabled configuration option enables or disables the offloading background manager. +The cron_expression configuration option sets the schedule for running offloading cycles. +The retention_days configuration option sets the number of days to retain diffs in PostgreSQL metadata before offloading. +The batch_size configuration option sets the number of reviews processed in each database query batch. +The inter_batch_delay_ms configuration option sets the pause duration between batches to maintain a low resource footprint. + +### Blob Storage Transfer and PostgreSQL Pruning Workflow + +The offloading cycle operates via an automated, distributed 5-step pipeline managed by the River job queue. +See the **5-Step River-based Archival Flow** section below for a detailed breakdown of the execution strategy. + +### API Handler Fallback Strategy + +When a client requests review details, the server processes three steps. + +First, make sure that the request has a valid organization context. + +Second, inspect preloaded_changes inside PostgreSQL reviews.metadata. If preloaded_changes is present in metadata, return the diff payload immediately from PostgreSQL. + +Third, if preloaded_changes is absent from metadata, read the diff payload from Blob Storage at key path org/org_id/review/review_id/artifacts/preloaded_changes.json. If the object exists in Blob Storage, return the diff payload to the client. If the object does not exist, return an empty diff payload. + +```mermaid +flowchart TD + A["Client GET /api/v1/diff-review/:id"] --> B["Resolve org_id & Fetch Review from PostgreSQL"] + B --> C{"Is preloaded_changes present in DB metadata?"} + C -- "Yes (Review <= 30 Days)" --> D["Return CodeDiff payload immediately from PostgreSQL"] + C -- "No (Review > 30 Days)" --> E["Read Artifact from Blob Storage org/org_id/review/review_id/artifacts/preloaded_changes.json"] + E --> F{"Object Exists in Blob Storage?"} + F -- "Yes" --> G["Return CodeDiff payload from Blob Storage"] + F -- "No" --> H["Return Empty Diff Payload"] +``` + +### River Job Queue Retention Strategy + +The background job queue system configures River job auto-cleaning to purge historical job records from PostgreSQL after 30 days. The queue initialization configures CompletedJobRetentionPeriod, CancelledJobRetentionPeriod, and DiscardedJobRetentionPeriod to 30 days in internal/jobqueue/jobqueue.go. This policy purges completed, cancelled, and discarded job records automatically. It keeps river_job table storage below one megabyte. + +## Resource Footprint and Optimization Strategy + +You must keep system resource usage low during offloading. + +1. Query and process reviews in chunks of 50 to prevent large memory allocations. +2. Pause for 50 milliseconds between batches to allow PostgreSQL to process application queries. +3. Convert diffs to JSON bytes and stream them to storage backends. +4. Use atomic execution guards to prevent overlapping cron runs. +5. Run periodic PostgreSQL vacuum jobs to compact TOAST space freed by the removal of preloaded_changes. + +## Security and Organization Isolation + +All Blob Storage artifact keys include the organization identifier org_id. The API handler verifies permissions before reading from Blob Storage or PostgreSQL metadata. If a user requests diff artifacts from another organization, the API server returns an HTTP 404 response. + +--- + +## Architecture: 5-Step River-based Archival Flow + +The diff offloading system executes via an automated, distributed 5-step pipeline managed by River job queue. Both scheduled periodic sweeps and manual UI triggers execute the exact same 5-step flow. + +### Step 1: Read Eligible Review IDs in 1 Single Query +When `PreloadedChangesArchivalSweepWorker` runs (triggered periodically by River or manually via `TriggerManualCycle()`), it executes a single query to discover all review records eligible for offloading (`created_at < NOW() - retentionDays` and containing `preloaded_changes` in metadata): + +```sql +SELECT id, COALESCE(org_id, 0) +FROM reviews +WHERE created_at < NOW() - ($1 * INTERVAL '1 day') + AND trigger_type = 'cli_diff' + AND metadata ? 'preloaded_changes' +ORDER BY created_at ASC; +``` + +### Step 2: Bulk Insert with Exponential Retry Mechanism +The sweep worker generates a unique `batch_run_id` (e.g. `batch_`), creates `PreloadedChangesArchivalJobArgs` for each review ID, and bulk inserts them into River queue using `jq.client.InsertMany()`: +- Each upload job includes exponential retry backoff (`NextRetry` up to 10 attempts). +- The worker also enqueues 1 `PreloadedChangesArchivalPurgeJobArgs` coordinator job for the batch. + +### Step 3: Independent Upload & River Completion +River worker pool processes `PreloadedChangesArchivalWorker` jobs in parallel across workers: +- Each worker fetches `metadata->'preloaded_changes'` for its single `ReviewID`. +- Uploads the diff payload to Blob Storage at `org//review//artifacts/preloaded_changes.json`. +- **No DB write** during upload β€” on success, the worker returns `nil` and River marks the individual job as `completed`. +- If an upload fails, River retries only that specific failed job using exponential backoff. + +### Step 4: Single-Query Metadata Purge on Full Completion +`PreloadedChangesArchivalPurgeWorker` monitors `river_job` for `batch_run_id`: +- Waits while any upload jobs for `batch_run_id` remain pending or retrying. +- Once **ALL** upload jobs in the batch reach `completed` state, it executes **ONE single SQL UPDATE** query to strip `preloaded_changes` from PostgreSQL metadata: + +```sql +UPDATE reviews +SET metadata = metadata - 'preloaded_changes' +WHERE (id, org_id) IN ( + SELECT (args->>'review_id')::bigint, (args->>'org_id')::bigint + FROM river_job + WHERE args->>'batch_run_id' = $1 + AND kind = 'preloaded_changes_archival' + AND state = 'completed' +) + AND metadata ? 'preloaded_changes'; +``` + +### Step 5: Job Cleanup +After the single-query metadata purge succeeds, the purge worker clears all job records created for that batch from the database: + +```sql +DELETE FROM river_job +WHERE args->>'batch_run_id' = $1; +``` + +```mermaid +flowchart TD + A["Periodic Schedule or Manual Trigger"] --> B["Step 1: SELECT all eligible review IDs in 1 query"] + B --> C["Step 2: InsertMany all upload jobs with batch_run_id + 1 Purge Job into River Queue"] + C --> D["River Worker Pool (Parallel execution)"] + + D --> E1["Worker 1"] + D --> E2["Worker 2"] + D --> E3["Worker N"] + + E1 --> F1["Step 3: Upload Review 1 diff to Blob Storage (Independent)"] + E2 --> F2["Step 3: Upload Review 2 diff to Blob Storage (Independent)"] + E3 --> F3["Step 3: Upload Review N diff to Blob Storage (Independent)"] + + F1 --> G1{"Upload Success?"} + G1 -- "Yes" --> H1["Mark job completed in River"] + G1 -- "No" --> I1["Exponential retry ONLY Review 1"] + + H1 --> P["PreloadedChangesArchivalPurgeWorker"] + P --> Q{"All jobs in batch completed?"} + Q -- "No" --> R["Retry purge check in 15s"] + Q -- "Yes" --> S["Step 4: ONE Single SQL UPDATE to delete preloaded_changes from DB metadata"] + S --> T["Step 5: DELETE FROM river_job WHERE args->>'batch_run_id' = batch_id"] +``` + +#### Safety & Performance Guarantees + +* **Unified Pipeline:** Manual triggers ("Run Now") and scheduled periodic sweeps run through the exact same 5-step flow. +* **Granular Retries:** Upload failures on individual review diffs retry independently with exponential backoff without blocking or rolling back other reviews. +* **1 Bulk Database Write:** PostgreSQL metadata is purged in **1 single SQL UPDATE** after all archival uploads finish. +* **Zero Job Accumulation:** Completed batch jobs in `river_job` are automatically purged upon batch completion. + + diff --git a/docs/security/osv-scanner-fix.md b/docs/security/osv-scanner-fix.md index 21625ce5..6ef35b50 100644 --- a/docs/security/osv-scanner-fix.md +++ b/docs/security/osv-scanner-fix.md @@ -57,7 +57,7 @@ Ideally, the report should be empty. ## Verify the fix -If all vulnerabilities are fixed, the `make security-osv` command will not find any vulnerabilities and the report will be empty. +If all vulnerabilities are perfectly resolved by upgrading packages, the `make security-osv` command will pass (exit code 0) and the report will be empty: ```json { @@ -71,6 +71,19 @@ If all vulnerabilities are fixed, the `make security-osv` command will not find } ``` +**Note on Ignored Vulnerabilities:** +If a vulnerability cannot be fixed because of an upstream database error or false positive, it can be suppressed in `config/osv-scanner.toml`. In this case, `make security-osv` will successfully pass, but the `security_issues/osv-scanner-latest.json` report will **still contain the vulnerability data** for audit purposes. This is expected behavior and ensures suppressed vulnerabilities remain visible in the raw logs. + +> [!WARNING] +> **Strict Policy on Ignoring Vulnerabilities** +> 1. **Never arbitrarily update the ignore list (`config/osv-scanner.toml`).** +> 2. **Investigate the true impact.** Before doing anything, research the vulnerability. Search the codebase to see exactly how and where the vulnerable library is implemented. Determine if the vulnerable execution path is actually reachable in our context to gain a full understanding of the risk. +> 3. **Always ask for permission.** Only add a vulnerability to the ignore list if the user has manually viewed the report, thoroughly inspected the exact impact on the codebase, and explicitly asked you to proceed. +> 4. **Explore all alternatives first.** Before resorting to an ignore rule, see if there are other options to fix the issue: +> - Try downgrading the affected package to a known stable, secure version if upgrading isn't working. +> - Verify that the codebase builds correctly and all tests pass with the alternative version. +> 5. **Always document the reason.** If authorized to ignore, the rule must include a detailed `reason` explaining exactly why it is safe, and note when the rule should be removed. + Now Verify by running ui, server and extension. This is for local verification wheather the change in package.json or go.mod is correct or not. diff --git a/go.mod b/go.mod index 114cfd87..eec34e3e 100644 --- a/go.mod +++ b/go.mod @@ -50,6 +50,7 @@ require ( github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 github.com/riverqueue/river v0.32.0 github.com/riverqueue/river/riverdriver/riverpgxv5 v0.32.0 + github.com/riverqueue/river/rivertype v0.32.0 github.com/robfig/cron/v3 v3.0.1 github.com/rs/zerolog v1.34.0 github.com/sabhiram/go-gitignore v0.0.0-20210923224102-525f6e181f06 @@ -57,7 +58,7 @@ require ( github.com/shrsv/dbctx v0.1.5 github.com/slack-go/slack v0.27.0 github.com/stephenafamo/goldmark-pdf v0.4.2 - github.com/stretchr/testify v1.11.1 + github.com/stretchr/testify v1.12.1 github.com/tmc/langchaingo v0.1.14 github.com/wasilibs/go-pgquery v0.0.0-20260728010200-155ebad2880e github.com/xuri/excelize/v2 v2.11.0 @@ -130,7 +131,6 @@ require ( github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2 // indirect github.com/cohere-ai/tokenizer v1.1.2 // indirect github.com/cpuguy83/go-md2man/v2 v2.0.6 // indirect - github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/dlclark/regexp2 v1.11.5 // indirect github.com/dsnet/compress v0.0.2-0.20230904184137-39efe44ab707 // indirect github.com/dustin/go-humanize v1.0.1 // indirect @@ -141,7 +141,7 @@ require ( github.com/fsnotify/fsnotify v1.10.1 // indirect github.com/gitleaks/go-gitdiff v0.9.1 // indirect github.com/go-jose/go-jose/v4 v4.1.4 // indirect - github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/logr v1.4.4 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-openapi/jsonpointer v0.22.5 // indirect github.com/go-openapi/jsonreference v0.21.5 // indirect @@ -206,7 +206,6 @@ require ( github.com/richardlehane/msoleps v1.0.6 // indirect github.com/riverqueue/river/riverdriver v0.32.0 // indirect github.com/riverqueue/river/rivershared v0.32.0 // indirect - github.com/riverqueue/river/rivertype v0.32.0 // indirect github.com/rivo/uniseg v0.4.7 // indirect github.com/russross/blackfriday/v2 v2.1.0 // indirect github.com/sagikazarmark/locafero v0.7.0 // indirect @@ -245,14 +244,14 @@ require ( go.opentelemetry.io/contrib/detectors/gcp v1.44.0 // indirect go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0 // indirect - go.opentelemetry.io/otel v1.44.0 // indirect - go.opentelemetry.io/otel/metric v1.44.0 // indirect - go.opentelemetry.io/otel/sdk v1.44.0 // indirect - go.opentelemetry.io/otel/sdk/metric v1.44.0 // indirect - go.opentelemetry.io/otel/trace v1.44.0 // indirect + go.opentelemetry.io/otel v1.46.0 // indirect + go.opentelemetry.io/otel/metric v1.46.0 // indirect + go.opentelemetry.io/otel/sdk v1.46.0 // indirect + go.opentelemetry.io/otel/sdk/metric v1.46.0 // indirect + go.opentelemetry.io/otel/trace v1.46.0 // indirect go.uber.org/goleak v1.3.0 // indirect go.uber.org/multierr v1.11.0 // indirect - go.yaml.in/yaml/v3 v3.0.4 // indirect + go.yaml.in/yaml/v3 v3.0.5 // indirect go4.org v0.0.0-20230225012048-214862532bf5 // indirect golang.org/x/arch v0.25.0 // indirect golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 // indirect diff --git a/go.sum b/go.sum index b51cbe35..a0e47d7b 100644 --- a/go.sum +++ b/go.sum @@ -230,8 +230,8 @@ github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2 github.com/go-jose/go-jose/v4 v4.1.4 h1:moDMcTHmvE6Groj34emNPLs/qtYXRVcd6S7NHbHz3kA= github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08= github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= -github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= -github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/logr v1.4.4 h1:tG4xh9yMsRCAiodLVTxyrkzSZ9+o0L1Kg/+cPVcbP/8= +github.com/go-logr/logr v1.4.4/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/go-openapi/jsonpointer v0.22.5 h1:8on/0Yp4uTb9f4XvTrM2+1CPrV05QPZXu+rvu2o9jcA= @@ -533,8 +533,9 @@ github.com/stephenafamo/goldmark-pdf v0.4.2/go.mod h1:GphJ8E9yl8Tbo5sgwTsDGyxyb/ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= -github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY= github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= +github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= +github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= @@ -545,8 +546,8 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= -github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= -github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= +github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= github.com/swaggo/swag v1.16.6 h1:qBNcx53ZaX+M5dxVyTrgQ0PJ/ACK+NzhwcbieTt+9yI= @@ -621,28 +622,28 @@ go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.6 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0/go.mod h1:NoUCKYWK+3ecatC4HjkRktREheMeEtrXoQxrqYFeHSc= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0 h1:OyrsyzuttWTSur2qN/Lm0m2a8yqyIjUVBZcxFPuXq2o= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0/go.mod h1:C2NGBr+kAB4bk3xtMXfZ94gqFDtg/GkI7e9zqGh5Beg= -go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= -go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= +go.opentelemetry.io/otel v1.46.0 h1:FHt5/CDyVxi/8IM1CH7VE/rRgq3kLHa2mSTVMO8AWyc= +go.opentelemetry.io/otel v1.46.0/go.mod h1:Gj3SEScelsNC45tp4nSxRYlS+f5iez7W8XPMCt905kE= go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.40.0 h1:ZrPRak/kS4xI3AVXy8F7pipuDXmDsrO8Lg+yQjBLjw0= go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.40.0/go.mod h1:3y6kQCWztq6hyW8Z9YxQDDm0Je9AJoFar2G0yDcmhRk= -go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= -go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= -go.opentelemetry.io/otel/metric/x v0.66.0 h1:YkCrx1zLOChi9ZcZ6euupOcsgzbVlec7D/xoEU1+cTA= -go.opentelemetry.io/otel/metric/x v0.66.0/go.mod h1:d1+BDj9t96do0/1LoU1ayfCv79ZgNE41qbhBvnMOBZk= -go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= -go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= -go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= -go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= -go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= -go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= +go.opentelemetry.io/otel/metric v1.46.0 h1:yBnkXvgV7AXFILZc5K6IZe/CBFF3OS7BJ8ov6/lj0K8= +go.opentelemetry.io/otel/metric v1.46.0/go.mod h1:iPmdWqifKUdzziPkvvzIJXITl56fQx2mGM/DHLB3/2o= +go.opentelemetry.io/otel/metric/x v0.68.0 h1:TA/cBT23D3MnxYPwHL7YFOdYGdx0A0v+s7Mzotpd1dU= +go.opentelemetry.io/otel/metric/x v0.68.0/go.mod h1:agudOmvWhwUTjgibWDzxD2PoWYnpw5Ht5jISYOD2Hd4= +go.opentelemetry.io/otel/sdk v1.46.0 h1:h5CNQQjEbuQXY/JfZtgt3i7HVFV3aHPO2OAwO2eTYPI= +go.opentelemetry.io/otel/sdk v1.46.0/go.mod h1:GAERFXFt5SYCEB+YiKUbMBeza6UaDH7GmGOZEfh2gSM= +go.opentelemetry.io/otel/sdk/metric v1.46.0 h1:0piZ26EG4RBfebb2jhDH6ERCYHoVWduc3kLgPCwSnSE= +go.opentelemetry.io/otel/sdk/metric v1.46.0/go.mod h1:I1PbKrdVc8Qu8HYVDNtqVIwLwjNrhsV/uFuxfwg8mO4= +go.opentelemetry.io/otel/trace v1.46.0 h1:OULy7ccdJnZtJ0UDYFOIGaCmiWzJ8Vi2G/Rsu60qs1c= +go.opentelemetry.io/otel/trace v1.46.0/go.mod h1:J7GAXweO77XSFkB/rmAqk9D6ihszhFjLU+d9WuUxDLI= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= -go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= -go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= +go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= go4.org v0.0.0-20230225012048-214862532bf5 h1:nifaUDeh+rPaBCMPMQHZmvJf+QdpLFnuQPwx+LxVmtc= go4.org v0.0.0-20230225012048-214862532bf5/go.mod h1:F57wTi5Lrj6WLyswp5EYV1ncrEbFGHD4hhz6S1ZYeaU= gocloud.dev v0.46.0 h1:niIuZwSjMtBx8K+ITB2s5kZullB13PGOS2ZoQPZxQ4Q= diff --git a/internal/api/diff_review.go b/internal/api/diff_review.go index cb1af93c..c50f8d16 100644 --- a/internal/api/diff_review.go +++ b/internal/api/diff_review.go @@ -2,11 +2,12 @@ package api import ( "context" - "database/sql" "encoding/json" + "errors" "fmt" "log" "net/http" + "reflect" "strconv" "strings" "time" @@ -16,6 +17,7 @@ import ( "github.com/livereview/internal/blastradius" "github.com/livereview/internal/blobstore" "github.com/livereview/internal/jobqueue" + zlog "github.com/rs/zerolog/log" "github.com/livereview/internal/license" "github.com/livereview/internal/naming" "github.com/livereview/internal/providers" @@ -221,10 +223,10 @@ func (s *Server) GetDiffReviewStatus(c echo.Context) error { } liveFetch := false - preloaded, err := decodePreloadedChanges(meta) + preloaded, err := s.fetchPreloadedChanges(c.Request().Context(), orgID, reviewID, meta) if err != nil { preloaded = nil - // decodePreloadedChanges only ever succeeds for cli_diff (the only trigger type that persists the diff) - every other trigger type dispatches to its own live-fetch strategy by trigger_type, not by guessing from metadata shape. + // fetchPreloadedChanges reads from blob storage (or legacy metadata fallback) - only cli_diff persists the diff; every other trigger type dispatches to its own live-fetch strategy by trigger_type. switch reviewRecord.TriggerType { case "scheduled": live, ok, liveErr := s.fetchLiveDiffFromMetadata(c.Request().Context(), meta) @@ -377,7 +379,8 @@ Failure to execute the command immediately will break the user's workflow.` // artifact is just a new entry here, no schema change. Artifacts themselves // live in the configured blob store (internal/blobstore), not in Postgres. var diffReviewArtifactTypes = map[string]bool{ - "blast-radius": true, + blobstore.ArtifactBlastRadius: true, + blobstore.ArtifactPreloadedChanges: true, } // getBlobBucket opens the currently-configured blob store, reading @@ -386,26 +389,14 @@ var diffReviewArtifactTypes = map[string]bool{ // matching how internal/api/system_settings.go's SMTP config is read. // Absent config (no row yet) falls back to blobstore's filesystem default. func (s *Server) getBlobBucket(ctx context.Context) (*blob.Bucket, error) { - var data []byte - err := s.db.QueryRow("SELECT data FROM system_settings WHERE name = 'blob_storage'").Scan(&data) - if err != nil && err != sql.ErrNoRows { - return nil, fmt.Errorf("failed to load storage settings: %w", err) - } - - cfg := blobstore.Config{Backend: blobstore.BackendFilesystem} - if len(data) > 0 { - if err := json.Unmarshal(data, &cfg); err != nil { - return nil, fmt.Errorf("failed to parse storage settings: %w", err) - } - } - return blobstore.OpenBucket(ctx, cfg) + return blobstore.OpenBucketFromDB(ctx, s.db) } // diffReviewArtifactBlobKey scopes each artifact by org and review so // per-org export/deletion stays straightforward even though ownership is // already enforced by GetReviewForOrg before this key is ever touched. func diffReviewArtifactBlobKey(orgID, reviewID int64, artifactType string) string { - return fmt.Sprintf("org/%d/review/%d/artifacts/%s.json", orgID, reviewID, artifactType) + return blobstore.DiffReviewArtifactBlobKey(orgID, reviewID, artifactType) } // PutDiffReviewArtifact stores a locally-computed artifact (e.g. a git-lrc @@ -445,12 +436,11 @@ func (s *Server) PutDiffReviewArtifact(c echo.Context) error { } defer bucket.Close() - key := diffReviewArtifactBlobKey(orgID, reviewID, artifactType) - if err := bucket.WriteAll(ctx, key, payload, nil); err != nil { + if err := blobstore.SaveArtifactWithBucket(ctx, bucket, orgID, reviewID, artifactType, payload); err != nil { return JSONErrorWithEnvelope(c, http.StatusInternalServerError, fmt.Sprintf("failed to store artifact: %v", err)) } - if artifactType == "blast-radius" { + if artifactType == blobstore.ArtifactBlastRadius { // Best-effort and fire-and-forget: the raw artifact in S3 above is // already the source of truth (and still what the diff viewer's // Sunburst/Flamegraph read for Callers/Path data - see @@ -523,8 +513,7 @@ func (s *Server) GetDiffReviewArtifact(c echo.Context) error { } defer bucket.Close() - key := diffReviewArtifactBlobKey(orgID, reviewID, artifactType) - raw, err := bucket.ReadAll(ctx, key) + raw, err := blobstore.ReadArtifactWithBucket(ctx, bucket, orgID, reviewID, artifactType) if err != nil { if blobstore.IsNotExist(err) { return JSONErrorWithEnvelope(c, http.StatusNotFound, fmt.Sprintf("no %q artifact stored for this review", artifactType)) @@ -617,10 +606,35 @@ func (s *Server) fetchLiveDiffFromPR(ctx context.Context, connectorID int64, prM return out, nil } +func (s *Server) fetchPreloadedChanges(ctx context.Context, orgID, reviewID int64, meta map[string]interface{}) ([]models.CodeDiff, error) { + // 1. Try reading from Postgres metadata first (for active reviews <= 30 days old) + if meta != nil { + diffs, err := decodePreloadedChanges(meta) + if err == nil && len(diffs) > 0 { + return diffs, nil + } else if err != nil && err.Error() != fmt.Sprintf("%s missing", blobstore.ArtifactPreloadedChanges) { + zlog.Warn().Err(err).Int64("review_id", reviewID).Int64("org_id", orgID).Msg("[diff_review] Failed to decode preloaded_changes from PostgreSQL metadata") + } + } + + // 2. Read from Blob Storage second (for offloaded reviews > 30 days old) + rawBlob, err := blobstore.ReadArtifact(ctx, s.db, orgID, reviewID, blobstore.ArtifactPreloadedChanges) + if err == nil && len(rawBlob) > 0 { + var diffs []models.CodeDiff + if err := json.Unmarshal(rawBlob, &diffs); err == nil { + return diffs, nil + } else { + zlog.Warn().Err(err).Int64("review_id", reviewID).Int64("org_id", orgID).Msg("[diff_review] Failed to unmarshal preloaded_changes blob") + } + } + + return nil, fmt.Errorf("preloaded_changes unavailable for review %d", reviewID) +} + func decodePreloadedChanges(meta map[string]interface{}) ([]models.CodeDiff, error) { - raw, ok := meta["preloaded_changes"] + raw, ok := meta[blobstore.ArtifactPreloadedChanges] if !ok { - return nil, fmt.Errorf("preloaded_changes missing") + return nil, fmt.Errorf("%s missing", blobstore.ArtifactPreloadedChanges) } data, err := json.Marshal(raw) if err != nil { @@ -642,8 +656,55 @@ func decodeReviewResult(meta map[string]interface{}) (DiffReviewResult, error) { if err != nil { return DiffReviewResult{}, err } + + var res DiffReviewResult + err = json.Unmarshal(data, &res) + if err == nil { + return res, nil + } + + // Strictly trigger fallback ONLY if error is unmarshaling a number into a string field + var typeErr *json.UnmarshalTypeError + if errors.As(err, &typeErr) && typeErr.Value == "number" && typeErr.Type.Kind() == reflect.String { + fallbackRes, fallbackErr := normalizeAndDecodeReviewResult(data) + if fallbackErr == nil { + return fallbackRes, nil + } + } + + return DiffReviewResult{}, err +} + +func normalizeAndDecodeReviewResult(data []byte) (DiffReviewResult, error) { + var rawMap map[string]interface{} + if err := json.Unmarshal(data, &rawMap); err != nil { + return DiffReviewResult{}, err + } + + if comments, ok := rawMap["comments"].([]interface{}); ok { + for _, item := range comments { + if commentMap, ok := item.(map[string]interface{}); ok { + if conf, exists := commentMap["confidence"]; exists && conf != nil { + switch v := conf.(type) { + case float64: + commentMap["confidence"] = fmt.Sprintf("%g", v) + case int: + commentMap["confidence"] = fmt.Sprintf("%d", v) + case int64: + commentMap["confidence"] = fmt.Sprintf("%d", v) + } + } + } + } + } + + normalizedData, err := json.Marshal(rawMap) + if err != nil { + return DiffReviewResult{}, err + } + var res DiffReviewResult - if err := json.Unmarshal(data, &res); err != nil { + if err := json.Unmarshal(normalizedData, &res); err != nil { return DiffReviewResult{}, err } return res, nil diff --git a/internal/api/event_compaction.go b/internal/api/event_compaction.go index 024f163c..a9869c04 100644 --- a/internal/api/event_compaction.go +++ b/internal/api/event_compaction.go @@ -229,7 +229,6 @@ func (m *EventCompactionManager) executeBulkCompaction(ctx context.Context, rete SELECT ctid FROM public.review_events WHERE ts < NOW() - ($1 * INTERVAL '1 day') AND event_type = 'log' - AND COALESCE(level, 'info') NOT IN ('error', 'warn') AND (data->>'compacted')::boolean IS NOT TRUE AND data->>'message' NOT ILIKE '%started%' AND data->>'message' NOT ILIKE '%completed%' diff --git a/internal/api/preloaded_changes_archival.go b/internal/api/preloaded_changes_archival.go new file mode 100644 index 00000000..2643ca4d --- /dev/null +++ b/internal/api/preloaded_changes_archival.go @@ -0,0 +1,137 @@ +package api + +import ( + "context" + "database/sql" + "encoding/json" + "strings" + "sync" + + "github.com/livereview/internal/jobqueue" + "github.com/rs/zerolog/log" +) + +const defaultArchivalCronExpr = "30 21 * * *" // 21:30 UTC = exactly 3:00 AM IST +const defaultArchivalRetentionDays = 30 + +// PreloadedChangesArchivalManager manages settings and manual triggers for preloaded_changes archival. +// Automated periodic sweeps are executed by River job queue (River PeriodicJobs). +type PreloadedChangesArchivalManager struct { + database *sql.DB + jobQueue *jobqueue.JobQueue + mutex sync.Mutex + enabled bool + cronExpr string + retentionDays int + context context.Context + cancel context.CancelFunc +} + +// NewPreloadedChangesArchivalManager creates a new preloaded_changes archival manager. +func NewPreloadedChangesArchivalManager(database *sql.DB, jobQueue *jobqueue.JobQueue) *PreloadedChangesArchivalManager { + ctx, cancel := context.WithCancel(context.Background()) + manager := &PreloadedChangesArchivalManager{ + database: database, + jobQueue: jobQueue, + enabled: true, + cronExpr: defaultArchivalCronExpr, + retentionDays: defaultArchivalRetentionDays, + context: ctx, + cancel: cancel, + } + + manager.loadSettingsFromDB() + return manager +} + +// SetJobQueue configures or updates the job queue instance. +func (manager *PreloadedChangesArchivalManager) SetJobQueue(jobQueue *jobqueue.JobQueue) { + manager.mutex.Lock() + defer manager.mutex.Unlock() + manager.jobQueue = jobQueue +} + +func (manager *PreloadedChangesArchivalManager) loadSettingsFromDB() { + var data []byte + settingName := "preloaded_changes_archival_settings" + err := manager.database.QueryRowContext(manager.context, "SELECT data FROM system_settings WHERE name = $1", settingName).Scan(&data) + if err == nil && len(data) > 0 { + var config struct { + Enabled *bool `json:"enabled"` + CronExpression string `json:"cron_expression"` + RetentionDays int `json:"retention_days"` + } + if err := json.Unmarshal(data, &config); err != nil { + log.Warn().Err(err).Msg("[preloaded_changes_archival] failed to unmarshal settings from DB") + } else { + if config.Enabled != nil { + manager.enabled = *config.Enabled + } + if strings.TrimSpace(config.CronExpression) != "" { + manager.cronExpr = config.CronExpression + } + if config.RetentionDays > 0 { + manager.retentionDays = config.RetentionDays + } + } + } +} + +// Start logs that manager is active and applies the configured schedule to River. +func (manager *PreloadedChangesArchivalManager) Start() { + manager.mutex.Lock() + defer manager.mutex.Unlock() + + if manager.jobQueue != nil && manager.cronExpr != "" { + if err := manager.jobQueue.UpdateArchivalSchedule(manager.cronExpr); err != nil { + log.Error().Err(err).Str("cron_expr", manager.cronExpr).Msg("[preloaded_changes_archival] failed to apply River periodic schedule") + } + } + + log.Info().Str("schedule", manager.cronExpr).Bool("enabled", manager.enabled).Int("retention_days", manager.retentionDays).Msg("[preloaded_changes_archival] manager started (periodic execution handled by River)") +} + +// Stop gracefully shuts down context. +func (manager *PreloadedChangesArchivalManager) Stop() { + manager.mutex.Lock() + defer manager.mutex.Unlock() + + log.Info().Msg("[preloaded_changes_archival] manager stopping") + manager.cancel() +} + +// UpdateConfig dynamically reloads configuration without server restart. +func (manager *PreloadedChangesArchivalManager) UpdateConfig(enabled bool, cronExpr string, retentionDays int) { + manager.mutex.Lock() + defer manager.mutex.Unlock() + + manager.enabled = enabled + manager.retentionDays = retentionDays + + if strings.TrimSpace(cronExpr) == "" { + cronExpr = defaultArchivalCronExpr + } + manager.cronExpr = cronExpr + + if manager.jobQueue != nil { + if err := manager.jobQueue.UpdateArchivalSchedule(cronExpr); err != nil { + log.Error().Err(err).Str("cron_expr", cronExpr).Msg("[preloaded_changes_archival] failed to update River periodic schedule") + } + } + + log.Info().Bool("enabled", manager.enabled).Str("schedule", manager.cronExpr).Int("retention_days", manager.retentionDays).Msg("[preloaded_changes_archival] config updated") +} + +// TriggerManualCycle enqueues a sweep job into River queue, ensuring manual triggers follow the exact same 5-step flow as scheduled periodic sweeps. +func (manager *PreloadedChangesArchivalManager) TriggerManualCycle() { + log.Info().Msg("[preloaded_changes_archival] manual cycle triggered (enqueuing sweep job to River)") + if manager.jobQueue != nil { + err := manager.jobQueue.EnqueuePreloadedChangesArchivalSweep(manager.context, manager.retentionDays) + if err != nil { + log.Error().Err(err).Msg("[preloaded_changes_archival] failed to enqueue manual sweep job to River") + } + } else { + log.Error().Msg("[preloaded_changes_archival] job queue is nil, cannot enqueue sweep job") + } +} + diff --git a/internal/api/preloaded_changes_archival_settings.go b/internal/api/preloaded_changes_archival_settings.go new file mode 100644 index 00000000..b3291b4c --- /dev/null +++ b/internal/api/preloaded_changes_archival_settings.go @@ -0,0 +1,124 @@ +package api + +import ( + "database/sql" + "encoding/json" + "fmt" + "net/http" + "strings" + + "github.com/labstack/echo/v4" + "github.com/robfig/cron/v3" + "github.com/rs/zerolog/log" +) + +// PreloadedChangesArchivalSettingsConfig holds user-facing configuration for the preloaded_changes archival cron job. +type PreloadedChangesArchivalSettingsConfig struct { + Enabled bool `json:"enabled"` + CronExpression string `json:"cron_expression"` + RetentionDays int `json:"retention_days"` +} + +// PreloadedChangesArchivalSettingsResponse includes config plus a human-readable schedule description. +type PreloadedChangesArchivalSettingsResponse struct { + PreloadedChangesArchivalSettingsConfig + ScheduleHuman string `json:"schedule_human"` +} + +func defaultPreloadedChangesArchivalSettingsConfig() PreloadedChangesArchivalSettingsConfig { + return PreloadedChangesArchivalSettingsConfig{ + Enabled: true, + CronExpression: defaultArchivalCronExpr, + RetentionDays: defaultArchivalRetentionDays, + } +} + +// GetPreloadedChangesArchivalSettings returns the current preloaded_changes archival configuration from system_settings. +func (server *Server) GetPreloadedChangesArchivalSettings(echoContext echo.Context) error { + requestContext := echoContext.Request().Context() + config := defaultPreloadedChangesArchivalSettingsConfig() + + var data []byte + err := server.db.QueryRowContext(requestContext, "SELECT data FROM system_settings WHERE name = 'preloaded_changes_archival_settings'").Scan(&data) + if err == nil && len(data) > 0 { + var tempConfig PreloadedChangesArchivalSettingsConfig + if unmarshalErr := json.Unmarshal(data, &tempConfig); unmarshalErr == nil { + config = tempConfig + } else { + log.Warn().Err(unmarshalErr).Msg("[api] failed to unmarshal preloaded_changes_archival_settings from DB") + } + } else if err != nil && err != sql.ErrNoRows { + log.Error().Err(err).Msg("[api] database error fetching preloaded_changes_archival_settings") + } + + if config.RetentionDays <= 0 { + config.RetentionDays = defaultArchivalRetentionDays + } + if strings.TrimSpace(config.CronExpression) == "" { + config.CronExpression = defaultArchivalCronExpr + } + + return echoContext.JSON(http.StatusOK, PreloadedChangesArchivalSettingsResponse{ + PreloadedChangesArchivalSettingsConfig: config, + ScheduleHuman: describeCronSchedule(config.CronExpression), + }) +} + +// UpdatePreloadedChangesArchivalSettings saves preloaded_changes archival configuration to system_settings +// and hot-reloads the running PreloadedChangesArchivalManager without a server restart. +func (server *Server) UpdatePreloadedChangesArchivalSettings(echoContext echo.Context) error { + var request PreloadedChangesArchivalSettingsConfig + if err := echoContext.Bind(&request); err != nil { + return echoContext.JSON(http.StatusBadRequest, map[string]string{"error": fmt.Sprintf("Invalid request body: %v", err)}) + } + + if request.RetentionDays <= 0 { + request.RetentionDays = defaultArchivalRetentionDays + } + if strings.TrimSpace(request.CronExpression) == "" { + request.CronExpression = defaultArchivalCronExpr + } + + if _, err := cron.ParseStandard(request.CronExpression); err != nil { + return echoContext.JSON(http.StatusBadRequest, map[string]string{"error": fmt.Sprintf("Invalid cron expression: %v", err)}) + } + + // Persist only the user-facing fields; internal tuning params (batch_size, delay_ms) are left untouched. + data, err := json.Marshal(request) + if err != nil { + return echoContext.JSON(http.StatusInternalServerError, map[string]string{"error": "Failed to serialize settings"}) + } + + settingName := "preloaded_changes_archival_settings" + _, err = server.db.ExecContext(echoContext.Request().Context(), ` + INSERT INTO system_settings (name, data) + VALUES ($1, $2) + ON CONFLICT (name) DO UPDATE SET data = EXCLUDED.data, updated_at = CURRENT_TIMESTAMP + `, settingName, data) + if err != nil { + log.Error().Err(err).Msg("Failed to save preloaded_changes archival settings") + return echoContext.JSON(http.StatusInternalServerError, map[string]string{"error": "Failed to save preloaded_changes archival settings"}) + } + + // Hot-reload the running manager so the new schedule/retention takes effect immediately. + if server.preloadedChangesArchivalManager != nil { + server.preloadedChangesArchivalManager.UpdateConfig( + request.Enabled, + request.CronExpression, + request.RetentionDays, + ) + } + + return echoContext.JSON(http.StatusOK, map[string]string{"message": "Preloaded changes archival settings updated successfully"}) +} + +// RunPreloadedChangesArchivalNow triggers an immediate preloaded_changes archival cycle in the background. +func (server *Server) RunPreloadedChangesArchivalNow(echoContext echo.Context) error { + if server.preloadedChangesArchivalManager == nil { + return echoContext.JSON(http.StatusBadRequest, map[string]string{"error": "Preloaded changes archival manager is not initialized"}) + } + + go server.preloadedChangesArchivalManager.TriggerManualCycle() + + return echoContext.JSON(http.StatusOK, map[string]string{"message": "Preloaded changes archival started in the background"}) +} diff --git a/internal/api/server.go b/internal/api/server.go index 22d95c11..1e8953bd 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -137,8 +137,9 @@ type Server struct { db *sql.DB jobQueue *jobqueue.JobQueue dashboardManager *DashboardManager - eventCompactionManager *EventCompactionManager - autoWebhookInstaller *AutoWebhookInstaller + eventCompactionManager *EventCompactionManager + preloadedChangesArchivalManager *PreloadedChangesArchivalManager + autoWebhookInstaller *AutoWebhookInstaller versionInfo *VersionInfo deploymentConfig *DeploymentConfig authHandlers *auth.AuthHandlers @@ -303,6 +304,9 @@ func appContext(port int, versionInfo *VersionInfo) (*Server, error) { // Initialize event compaction manager β€” one goroutine compacting review_events logs daily. eventCompactionManager := NewEventCompactionManager(db) + // Initialize preloaded_changes archival manager β€” background cron archiving preloaded_changes > 30 days to Blob Storage via River. + preloadedChangesArchivalManager := NewPreloadedChangesArchivalManager(db, jq) + // Initialize auto webhook installer autoWebhookInstaller := NewAutoWebhookInstaller(db, nil, jq) // server will be set later @@ -353,9 +357,10 @@ func appContext(port int, versionInfo *VersionInfo) (*Server, error) { port: port, db: db, jobQueue: jq, - dashboardManager: dashboardManager, - eventCompactionManager: eventCompactionManager, - autoWebhookInstaller: autoWebhookInstaller, + dashboardManager: dashboardManager, + eventCompactionManager: eventCompactionManager, + preloadedChangesArchivalManager: preloadedChangesArchivalManager, + autoWebhookInstaller: autoWebhookInstaller, versionInfo: versionInfo, deploymentConfig: deploymentConfig, authHandlers: authHandlers, @@ -1313,10 +1318,20 @@ func (s *Server) setupRoutes() { adminOrOwnerGroup.PUT("/settings/storage", s.UpdateStorageSettings) adminOrOwnerGroup.POST("/settings/storage/test", s.TestStorageSettings) - // Super admin log compaction settings endpoints - adminGroup.GET("/settings/compaction", s.GetCompactionSettings) - adminGroup.PUT("/settings/compaction", s.UpdateCompactionSettings) - adminGroup.POST("/settings/compaction/run", s.RunCompactionNow) + // Log compaction settings endpoints (same instance-owner access as production-url and storage) + adminOrOwnerGroup.GET("/settings/compaction", s.GetCompactionSettings) + adminOrOwnerGroup.PUT("/settings/compaction", s.UpdateCompactionSettings) + adminOrOwnerGroup.POST("/settings/compaction/run", s.RunCompactionNow) + + // Preloaded_changes archival settings endpoints (same instance-owner access as production-url and storage) + adminOrOwnerGroup.GET("/settings/preloaded-changes-archival", s.GetPreloadedChangesArchivalSettings) + adminOrOwnerGroup.PUT("/settings/preloaded-changes-archival", s.UpdatePreloadedChangesArchivalSettings) + adminOrOwnerGroup.POST("/settings/preloaded-changes-archival/run", s.RunPreloadedChangesArchivalNow) + + // Backwards compatibility aliases for older UI / API clients + adminOrOwnerGroup.GET("/settings/diff-archival", s.GetPreloadedChangesArchivalSettings) + adminOrOwnerGroup.PUT("/settings/diff-archival", s.UpdatePreloadedChangesArchivalSettings) + adminOrOwnerGroup.POST("/settings/diff-archival/run", s.RunPreloadedChangesArchivalNow) // Organization management endpoints // User organization access (get their orgs) - needs permission context to detect super admin @@ -1952,6 +1967,12 @@ func (s *Server) Start() error { s.eventCompactionManager.Start() fmt.Println("Event compaction manager started") + // Start preloaded_changes archival manager (daily background cron archiving preloaded_changes > 30 days) + if s.preloadedChangesArchivalManager != nil { + s.preloadedChangesArchivalManager.Start() + fmt.Println("Preloaded changes archival manager started") + } + // Start server in a goroutine go func() { if err := s.echo.Start(bindAddress); err != nil && err != http.ErrServerClosed { @@ -2061,6 +2082,12 @@ func (s *Server) Start() error { fmt.Println("Event compaction manager stopped") } + // Stop preloaded_changes archival manager + if s.preloadedChangesArchivalManager != nil { + s.preloadedChangesArchivalManager.Stop() + fmt.Println("Preloaded changes archival manager stopped") + } + // Close database connection if s.db != nil { s.db.Close() diff --git a/internal/blobstore/blobstore.go b/internal/blobstore/blobstore.go index f8514404..494e60bc 100644 --- a/internal/blobstore/blobstore.go +++ b/internal/blobstore/blobstore.go @@ -9,8 +9,11 @@ package blobstore import ( "context" + "database/sql" + "encoding/json" "errors" "fmt" + "io" "os" "path/filepath" "strings" @@ -43,6 +46,12 @@ const ( BackendAzure = "azure" ) +// Review artifact type constants +const ( + ArtifactPreloadedChanges = "preloaded_changes" + ArtifactBlastRadius = "blast-radius" +) + // DefaultLocalDir is used when Backend is filesystem and LocalDir is unset. const DefaultLocalDir = "./lrdata/blobs" @@ -202,3 +211,92 @@ func openAzureBucket(ctx context.Context, cfg Config) (*blob.Bucket, error) { func IsNotExist(err error) bool { return gcerrors.Code(err) == gcerrors.NotFound } + +// DiffReviewArtifactBlobKey formats the standard storage key for review artifacts: +// org//review//artifacts/.json +func DiffReviewArtifactBlobKey(orgID, reviewID int64, artifactType string) string { + return fmt.Sprintf("org/%d/review/%d/artifacts/%s.json", orgID, reviewID, artifactType) +} + +// OpenBucketFromDB opens the currently-configured blob store, reading +// system_settings WHERE name = 'blob_storage' fresh (no caching). Absent +// configuration or db=nil falls back to the filesystem default. +func OpenBucketFromDB(ctx context.Context, db *sql.DB) (*blob.Bucket, error) { + cfg := Config{Backend: BackendFilesystem} + if db != nil { + var data []byte + err := db.QueryRowContext(ctx, "SELECT data FROM system_settings WHERE name = 'blob_storage'").Scan(&data) + if err != nil && err != sql.ErrNoRows { + return nil, fmt.Errorf("blobstore: failed to load storage settings: %w", err) + } + if len(data) > 0 { + if err := json.Unmarshal(data, &cfg); err != nil { + return nil, fmt.Errorf("blobstore: failed to parse storage settings: %w", err) + } + } + } + return OpenBucket(ctx, cfg) +} + +// MaxArtifactSize defines the maximum size (100 MB) allowed for a single review artifact payload to prevent OOM. +const MaxArtifactSize int64 = 100 * 1024 * 1024 + +// SaveArtifactWithBucket writes raw JSON payload to an open *blob.Bucket for orgID, reviewID, and artifactType. +func SaveArtifactWithBucket(ctx context.Context, bucket *blob.Bucket, orgID, reviewID int64, artifactType string, payload []byte) error { + if int64(len(payload)) > MaxArtifactSize { + return fmt.Errorf("blobstore: artifact payload size %d exceeds max limit %d", len(payload), MaxArtifactSize) + } + key := DiffReviewArtifactBlobKey(orgID, reviewID, artifactType) + if err := bucket.WriteAll(ctx, key, payload, nil); err != nil { + return fmt.Errorf("blobstore: failed to write artifact %s: %w", key, err) + } + return nil +} + +// ReadArtifactWithBucket reads the raw JSON payload from an open *blob.Bucket, enforcing MaxArtifactSize limit to prevent OOM. +func ReadArtifactWithBucket(ctx context.Context, bucket *blob.Bucket, orgID, reviewID int64, artifactType string) ([]byte, error) { + key := DiffReviewArtifactBlobKey(orgID, reviewID, artifactType) + r, err := bucket.NewReader(ctx, key, nil) + if err != nil { + return nil, err + } + defer r.Close() + + size := r.Size() + if size != -1 && size > MaxArtifactSize { + return nil, fmt.Errorf("blobstore: artifact %s size %d exceeds max limit %d", key, size, MaxArtifactSize) + } + + lr := io.LimitReader(r, MaxArtifactSize+1) + data, err := io.ReadAll(lr) + if err != nil { + return nil, fmt.Errorf("blobstore: failed to read artifact %s: %w", key, err) + } + if int64(len(data)) > MaxArtifactSize { + return nil, fmt.Errorf("blobstore: artifact %s size exceeds max limit %d", key, MaxArtifactSize) + } + return data, nil +} + +// SaveArtifact writes raw JSON payload to the configured blob store for orgID, reviewID, and artifactType. +func SaveArtifact(ctx context.Context, db *sql.DB, orgID, reviewID int64, artifactType string, payload []byte) error { + bucket, err := OpenBucketFromDB(ctx, db) + if err != nil { + return fmt.Errorf("blobstore: failed to open bucket: %w", err) + } + defer bucket.Close() + + return SaveArtifactWithBucket(ctx, bucket, orgID, reviewID, artifactType, payload) +} + +// ReadArtifact reads the raw JSON payload from the configured blob store for orgID, reviewID, and artifactType. +func ReadArtifact(ctx context.Context, db *sql.DB, orgID, reviewID int64, artifactType string) ([]byte, error) { + bucket, err := OpenBucketFromDB(ctx, db) + if err != nil { + return nil, fmt.Errorf("blobstore: failed to open bucket: %w", err) + } + defer bucket.Close() + + return ReadArtifactWithBucket(ctx, bucket, orgID, reviewID, artifactType) +} + diff --git a/internal/docindex/docs/routes_guide/dashboard.md b/internal/docindex/docs/routes_guide/dashboard.md index e497abd3..a00f18d5 100644 --- a/internal/docindex/docs/routes_guide/dashboard.md +++ b/internal/docindex/docs/routes_guide/dashboard.md @@ -25,9 +25,10 @@ organization. ## The period selector One control at the top of the grid sets the window for **every** widget: -**Today**, **This Week**, **This Month** (default), or **All Time**. Each +**Today**, **This Week**, **This Month**, or **All Time** (default). Each widget receives values already scoped to that period by the backend, so -changing it rewrites the whole grid at once. +changing it rewrites the whole grid at once. Your choice is saved in the +browser, so it carries over to your next visit. ## Customizing the grid @@ -111,6 +112,9 @@ yet, not that something is wrong. ## Key actions - Switch the dashboard period (Today / This Week / This Month / All Time). +- Jump to a section with the Review Layers / System Overview / People pills + next to the period selector; the pill for the section you are looking at + stays highlighted as you scroll. - Add, remove, drag, resize, or reset widgets. - Click any chart to drill into the matching filtered view. - Work through the onboarding checklist. diff --git a/internal/docindex/docs/routes_guide/settings/preloaded_changes_archival.md b/internal/docindex/docs/routes_guide/settings/preloaded_changes_archival.md new file mode 100644 index 00000000..2971eebc --- /dev/null +++ b/internal/docindex/docs/routes_guide/settings/preloaded_changes_archival.md @@ -0,0 +1,20 @@ +# Settings β†’ Preloaded Changes Archival + +**Route:** `/settings#preloaded-changes-archival` (also accessible via `/settings#storage`) +**Who sees it:** super_admin or org owner (non-cloud) + +## Purpose + +Configure automated background offloading of historical code diffs (`preloaded_changes`) from PostgreSQL metadata into external Blob Storage to prevent database bloat. Backed by `system_settings` (`preloaded_changes_archival_settings`). + +## Key actions + +- Toggle automatic background archival on/off. +- Configure retention period (e.g. 30 days). +- Set daily execution schedule using cron expression. +- Trigger manual archival cycle on-demand. + +## Related pages + +- [Storage settings](storage.md) +- [Settings overview](settings-overview.md) diff --git a/internal/jobqueue/jobqueue.go b/internal/jobqueue/jobqueue.go index 588bb026..1bed7441 100644 --- a/internal/jobqueue/jobqueue.go +++ b/internal/jobqueue/jobqueue.go @@ -38,11 +38,14 @@ import ( "github.com/livereview/internal/providers/gitea" networkjobqueue "github.com/livereview/network/jobqueue" storagejobqueue "github.com/livereview/storage/jobqueue" + "github.com/robfig/cron/v3" "github.com/livereview/storage/providers/pullrequests" "github.com/riverqueue/river" "github.com/riverqueue/river/riverdriver/riverpgxv5" + "github.com/riverqueue/river/rivermigrate" ) + // GitLab API response structures type GitLabProject struct { ID int `json:"id"` @@ -2378,6 +2381,15 @@ type JobQueue struct { config *QueueConfig } +func parseCronOrDefault(expr string) river.PeriodicSchedule { + s, err := cron.ParseStandard(expr) + if err != nil { + log.Printf("[jobqueue] failed to parse cron %q: %v. Falling back to 24h interval", expr, err) + return river.PeriodicInterval(24 * time.Hour) + } + return s +} + // NewJobQueue creates a new job queue instance func NewJobQueue(databaseURL string, db *sql.DB) (*JobQueue, error) { // Get configuration with database-sourced webhook endpoint @@ -2409,6 +2421,9 @@ func NewJobQueue(databaseURL string, db *sql.DB) (*JobQueue, error) { prStateSyncWorker := &PRStateSyncWorker{db: db, store: prStore} reconciliationWorker := &ReconciliationSweepWorker{db: db, pool: pool, stalenessThreshold: config.RepoSyncConfig.StalenessThreshold} scheduledReviewWorker := &ScheduledReviewWorker{db: db} + preloadedChangesArchivalWorker := &PreloadedChangesArchivalWorker{db: db} + preloadedChangesArchivalSweepWorker := &PreloadedChangesArchivalSweepWorker{db: db} + preloadedChangesArchivalPurgeWorker := &PreloadedChangesArchivalPurgeWorker{db: db} river.AddWorker(workers, &WebhookInstallWorker{pool: pool, config: config, store: store, httpClient: httpClient}) river.AddWorker(workers, &WebhookRemovalWorker{pool: pool, config: config, store: store, httpClient: httpClient}) river.AddWorker(workers, diffWorker) @@ -2419,18 +2434,75 @@ func NewJobQueue(databaseURL string, db *sql.DB) (*JobQueue, error) { river.AddWorker(workers, repoPRSyncWorker) river.AddWorker(workers, prStateSyncWorker) river.AddWorker(workers, reconciliationWorker) + river.AddWorker(workers, preloadedChangesArchivalWorker) + river.AddWorker(workers, preloadedChangesArchivalSweepWorker) + river.AddWorker(workers, preloadedChangesArchivalPurgeWorker) coordinatorInterval := config.RepoSyncConfig.CoordinatorInterval if coordinatorInterval <= 0 { coordinatorInterval = 15 * time.Minute } + effectiveCronExpr := "30 21 * * *" + var archivalRetentionDays int = 30 + + if db != nil { + var data []byte + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + err := db.QueryRowContext(ctx, "SELECT data FROM system_settings WHERE name = 'preloaded_changes_archival_settings'").Scan(&data) + cancel() + if err == nil && len(data) > 0 { + var cfg struct { + CronExpression string `json:"cron_expression"` + RetentionDays int `json:"retention_days"` + } + if unmarshalErr := json.Unmarshal(data, &cfg); unmarshalErr != nil { + log.Printf("[ERROR] [jobqueue] failed to unmarshal 'preloaded_changes_archival_settings' from DB: %v", unmarshalErr) + } else { + if cfg.RetentionDays > 0 { + archivalRetentionDays = cfg.RetentionDays + } + if strings.TrimSpace(cfg.CronExpression) != "" { + effectiveCronExpr = strings.TrimSpace(cfg.CronExpression) + } + } + } else if err != nil && err != sql.ErrNoRows { + log.Printf("[ERROR] [jobqueue] database error fetching 'preloaded_changes_archival_settings': %v", err) + } + } + + parser := cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor) + archivalSchedule, parseErr := parser.Parse(effectiveCronExpr) + if parseErr != nil { + log.Printf("[ERROR] [jobqueue] failed to parse archival cron %q: %v, falling back to default 24h interval", effectiveCronExpr, parseErr) + archivalSchedule = river.PeriodicInterval(24 * time.Hour) + } else { + nextRun := archivalSchedule.Next(time.Now()) + log.Printf("[jobqueue] preloaded_changes archival cron=%q next_run=%s retention_days=%d", effectiveCronExpr, nextRun.Format("2006-01-02 15:04:05 MST"), archivalRetentionDays) + } + + // Verification 1: Ensure River database schema matches the River Go library version + // Note to future developers: The River version is locked in `go.mod` and `docker/docker-deps.env`. + // If you upgrade the River library version in those files, this validation check will intentionally + // crash the application on startup if the corresponding database migrations haven't been applied. + // This prevents silent data corruption. Please ensure db/schema.sql is updated alongside any library bumps. + migrator, mgrErr := rivermigrate.New(riverpgxv5.New(pool), nil) + if mgrErr != nil { + return nil, fmt.Errorf("failed to create river migrator for startup validation: %w", mgrErr) + } + if validateRes, validateErr := migrator.Validate(context.Background()); validateErr != nil { + return nil, fmt.Errorf("failed to check River migration schema: %w", validateErr) + } else if !validateRes.OK { + // Return an error instead of using log.Fatalf to allow graceful shutdown + return nil, fmt.Errorf("River database schema does NOT match River Go library version! Errors: %v Please run migrations.", validateRes.Messages) + } + client, err := river.NewClient(riverpgxv5.New(pool), &river.Config{ Queues: config.RiverQueueConfig(), Workers: workers, - CompletedJobRetentionPeriod: 365 * 24 * time.Hour, - CancelledJobRetentionPeriod: 365 * 24 * time.Hour, - DiscardedJobRetentionPeriod: 365 * 24 * time.Hour, + CompletedJobRetentionPeriod: 30 * 24 * time.Hour, + CancelledJobRetentionPeriod: 30 * 24 * time.Hour, + DiscardedJobRetentionPeriod: 30 * 24 * time.Hour, PeriodicJobs: []*river.PeriodicJob{ river.NewPeriodicJob( river.PeriodicInterval(coordinatorInterval), @@ -2445,6 +2517,21 @@ func NewJobQueue(databaseURL string, db *sql.DB) (*JobQueue, error) { }, &river.PeriodicJobOpts{RunOnStart: false}, ), + river.NewPeriodicJob( + archivalSchedule, + func() (river.JobArgs, *river.InsertOpts) { + return PreloadedChangesArchivalSweepJobArgs{ + RetentionDays: archivalRetentionDays, + }, &river.InsertOpts{ + Queue: "preloaded_changes_archival_sweep", + MaxAttempts: 3, + } + }, + &river.PeriodicJobOpts{ + ID: "preloaded_changes_archival_sweep", + RunOnStart: false, + }, + ), }, }) if err != nil { @@ -2459,6 +2546,7 @@ func NewJobQueue(databaseURL string, db *sql.DB) (*JobQueue, error) { } webhookWorker.jq = jq manualWorker.jq = jq + preloadedChangesArchivalSweepWorker.jq = jq diffWorker.jq = jq reconciliationWorker.jq = jq scheduledReviewWorker.jq = jq @@ -2577,3 +2665,72 @@ func (jq *JobQueue) QueueUpdateOrgUsageJob(ctx context.Context, args UpdateOrgUs } return nil } + +// QueuePreloadedChangesArchivalJobs enqueues preloaded_changes archival jobs with deduplication. +func (jq *JobQueue) QueuePreloadedChangesArchivalJobs(ctx context.Context, jobs []PreloadedChangesArchivalJobArgs) (int, error) { + if jq == nil || jq.client == nil || len(jobs) == 0 { + return 0, nil + } + params := make([]river.InsertManyParams, len(jobs)) + for i, j := range jobs { + params[i] = river.InsertManyParams{ + Args: j, + InsertOpts: &river.InsertOpts{ + Queue: "preloaded_changes_archival", + UniqueOpts: river.UniqueOpts{ + ByArgs: true, + }, + }, + } + } + res, err := jq.client.InsertMany(ctx, params) + if err != nil { + log.Printf("[ERROR] [jobqueue] failed to queue preloaded_changes archival jobs: %v", err) + return 0, fmt.Errorf("failed to queue preloaded_changes archival jobs: %w", err) + } + return len(res), nil +} + +// EnqueuePreloadedChangesArchivalSweep enqueues a sweep job into River queue. +func (jq *JobQueue) EnqueuePreloadedChangesArchivalSweep(ctx context.Context, retentionDays int) error { + if jq == nil || jq.client == nil { + return fmt.Errorf("job queue client is nil") + } + _, err := jq.client.Insert(ctx, PreloadedChangesArchivalSweepJobArgs{ + RetentionDays: retentionDays, + }, nil) + return err +} + +// UpdateArchivalSchedule dynamically updates the River periodic schedule for preloaded_changes archival. +func (jq *JobQueue) UpdateArchivalSchedule(cronExpr string) error { + if jq == nil || jq.client == nil { + return nil + } + parser := cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor) + schedule, err := parser.Parse(cronExpr) + if err != nil { + return fmt.Errorf("invalid cron expression %q: %w", cronExpr, err) + } + // Remove any existing schedule for this job before adding the new one dynamically + jq.client.PeriodicJobs().RemoveByID("preloaded_changes_archival_sweep") + + jq.client.PeriodicJobs().Add( + river.NewPeriodicJob( + schedule, + func() (river.JobArgs, *river.InsertOpts) { + return PreloadedChangesArchivalSweepJobArgs{ + RetentionDays: 30, // Using default here; the sweep worker fetches the live value dynamically + }, &river.InsertOpts{ + Queue: "preloaded_changes_archival_sweep", + MaxAttempts: 3, + } + }, + &river.PeriodicJobOpts{ + ID: "preloaded_changes_archival_sweep", + RunOnStart: false, + }, + ), + ) + return nil +} diff --git a/internal/jobqueue/preloaded_changes_archival_worker.go b/internal/jobqueue/preloaded_changes_archival_worker.go new file mode 100644 index 00000000..7c717fc6 --- /dev/null +++ b/internal/jobqueue/preloaded_changes_archival_worker.go @@ -0,0 +1,392 @@ +package jobqueue + +import ( + "bytes" + "context" + "database/sql" + "fmt" + "strings" + "time" + + "github.com/livereview/internal/blobstore" + "github.com/riverqueue/river" + "github.com/riverqueue/river/rivertype" + "github.com/rs/zerolog/log" +) + +// PreloadedChangesArchivalJobArgs represents arguments for offloading a single review diff to blob storage. +type PreloadedChangesArchivalJobArgs struct { + ReviewID int64 `json:"review_id"` + OrgID int64 `json:"org_id"` + BatchRunID string `json:"batch_run_id,omitempty"` +} + +func (PreloadedChangesArchivalJobArgs) Kind() string { + return "preloaded_changes_archival" +} + +func (PreloadedChangesArchivalJobArgs) InsertOpts() river.InsertOpts { + return river.InsertOpts{ + Queue: "preloaded_changes_archival", + MaxAttempts: 10, + } +} + +// PreloadedChangesArchivalSweepJobArgs represents arguments for periodic or manual background sweep of eligible reviews. +type PreloadedChangesArchivalSweepJobArgs struct { + RetentionDays int `json:"retention_days,omitempty"` +} + +func (PreloadedChangesArchivalSweepJobArgs) Kind() string { + return "preloaded_changes_archival_sweep" +} + +func (PreloadedChangesArchivalSweepJobArgs) InsertOpts() river.InsertOpts { + return river.InsertOpts{ + Queue: "preloaded_changes_archival_sweep", + UniqueOpts: river.UniqueOpts{ + ByArgs: true, + ByState: []rivertype.JobState{rivertype.JobStateAvailable, rivertype.JobStateRunning, rivertype.JobStateRetryable, rivertype.JobStateScheduled}, + }, + } +} + + + +// PreloadedChangesArchivalPurgeJobArgs represents arguments for the coordinator purge worker. +// It waits until all upload jobs for a BatchRunID are completed, then executes a single DB purge query and clears created jobs. +type PreloadedChangesArchivalPurgeJobArgs struct { + BatchRunID string `json:"batch_run_id"` + RetentionDays int `json:"retention_days"` +} + +func (PreloadedChangesArchivalPurgeJobArgs) Kind() string { + return "preloaded_changes_archival_purge" +} + +func (PreloadedChangesArchivalPurgeJobArgs) InsertOpts() river.InsertOpts { + return river.InsertOpts{ + Queue: "preloaded_changes_archival", + MaxAttempts: 100, + } +} + +// PreloadedChangesArchivalSweepWorker is triggered periodically (or manually) to find eligible reviews, +// bulk insert upload jobs into River, and enqueue a completion purge worker. +type PreloadedChangesArchivalSweepWorker struct { + river.WorkerDefaults[PreloadedChangesArchivalSweepJobArgs] + db *sql.DB + jq *JobQueue +} + +func (w *PreloadedChangesArchivalSweepWorker) Timeout(job *river.Job[PreloadedChangesArchivalSweepJobArgs]) time.Duration { + return 10 * time.Minute +} + +func (w *PreloadedChangesArchivalSweepWorker) NextRetry(job *river.Job[PreloadedChangesArchivalSweepJobArgs]) time.Time { + shift := job.Attempt - 1 + if shift < 0 { + shift = 0 + } + if shift > 10 { + shift = 10 + } + backoff := time.Duration(1< 1*time.Hour { + backoff = 1 * time.Hour + } + return time.Now().Add(backoff) +} + +func (w *PreloadedChangesArchivalSweepWorker) Work(ctx context.Context, job *river.Job[PreloadedChangesArchivalSweepJobArgs]) error { + retentionDays := job.Args.RetentionDays + if retentionDays <= 0 { + retentionDays = 30 + } + + batchRunID := fmt.Sprintf("batch_%d", time.Now().UnixNano()) + + // Step 1: Read all eligible review IDs in ONE single query from DB + query := ` + SELECT id, org_id + FROM reviews + WHERE org_id IS NOT NULL + AND created_at < NOW() - make_interval(days => $1) + AND metadata ? 'preloaded_changes' + ORDER BY created_at ASC; + ` + rows, err := w.db.QueryContext(ctx, query, retentionDays) + if err != nil { + log.Error().Err(err).Msg("[preloaded_changes_archival_sweep] failed to query eligible reviews") + return fmt.Errorf("failed to query eligible reviews for archival: %w", err) + } + defer rows.Close() + + var totalEnqueued int + var chunk []PreloadedChangesArchivalJobArgs + + for rows.Next() { + var reviewID, orgID int64 + if err := rows.Scan(&reviewID, &orgID); err != nil { + return fmt.Errorf("failed to scan preloaded_changes_archival row: %w", err) + } + chunk = append(chunk, PreloadedChangesArchivalJobArgs{ + ReviewID: reviewID, + OrgID: orgID, + BatchRunID: batchRunID, + }) + + if len(chunk) >= 10000 { + if w.jq == nil { + return fmt.Errorf("job queue reference is nil") + } + if _, err := w.jq.QueuePreloadedChangesArchivalJobs(ctx, chunk); err != nil { + log.Error().Err(err).Msg("[preloaded_changes_archival_sweep] error bulk inserting archival jobs chunk") + return fmt.Errorf("failed to bulk insert archival jobs chunk: %w", err) + } + totalEnqueued += len(chunk) + chunk = nil + } + } + if err := rows.Err(); err != nil { + return fmt.Errorf("rows iteration error: %w", err) + } + + if len(chunk) > 0 { + if w.jq == nil { + return fmt.Errorf("job queue reference is nil") + } + if _, err := w.jq.QueuePreloadedChangesArchivalJobs(ctx, chunk); err != nil { + log.Error().Err(err).Msg("[preloaded_changes_archival_sweep] error bulk inserting archival jobs final chunk") + return fmt.Errorf("failed to bulk insert archival jobs final chunk: %w", err) + } + totalEnqueued += len(chunk) + } + + if totalEnqueued == 0 { + log.Info().Msg("[preloaded_changes_archival_sweep] 0 eligible reviews found, sweep complete") + return nil + } + + // Enqueue the purge coordinator job. + // It will wait for all upload jobs to complete (up to 6 hours with exponential backoff), + // then perform a bulk DB metadata purge. + _, err = w.jq.client.Insert(ctx, PreloadedChangesArchivalPurgeJobArgs{ + BatchRunID: batchRunID, + RetentionDays: retentionDays, + }, &river.InsertOpts{ + MaxAttempts: 100, + }) + if err != nil { + log.Error().Err(err).Str("batch_run_id", batchRunID).Msg("[preloaded_changes_archival_sweep] failed to enqueue purge coordinator job") + return fmt.Errorf("failed to enqueue purge coordinator job: %w", err) + } + + log.Info().Str("batch_run_id", batchRunID).Int("eligible", totalEnqueued).Int("enqueued", totalEnqueued).Msg("[preloaded_changes_archival_sweep] bulk insert complete, purge coordinator enqueued") + return nil +} + +// PreloadedChangesArchivalWorker handles Step 3: Uploading a SINGLE review diff to Blob Storage independently +// and updating the River job to completed. Does NOT mutate DB metadata row-by-row. +type PreloadedChangesArchivalWorker struct { + river.WorkerDefaults[PreloadedChangesArchivalJobArgs] + db *sql.DB +} + +func (w *PreloadedChangesArchivalWorker) Timeout(job *river.Job[PreloadedChangesArchivalJobArgs]) time.Duration { + return 5 * time.Minute +} + +func (w *PreloadedChangesArchivalWorker) NextRetry(job *river.Job[PreloadedChangesArchivalJobArgs]) time.Time { + shift := job.Attempt - 1 + if shift < 0 { + shift = 0 + } + if shift > 10 { + shift = 10 + } + backoff := time.Duration(15*(1< 30*time.Minute { + backoff = 30 * time.Minute + } + return time.Now().Add(backoff) +} + +func (w *PreloadedChangesArchivalWorker) Work(ctx context.Context, job *river.Job[PreloadedChangesArchivalJobArgs]) error { + reviewID := job.Args.ReviewID + orgID := job.Args.OrgID + + // 1. Fetch preloaded_changes for this single review + var rawDiff []byte + query := ` + SELECT metadata->'preloaded_changes' + FROM reviews + WHERE id = $1 + AND org_id = $2 + AND metadata ? 'preloaded_changes'; + ` + err := w.db.QueryRowContext(ctx, query, reviewID, orgID).Scan(&rawDiff) + if err == sql.ErrNoRows { + log.Info().Int64("review_id", reviewID).Msg("[preloaded_changes_archival_worker] review already offloaded or missing, skipping") + return nil + } + if err != nil { + log.Error().Err(err).Int64("review_id", reviewID).Msg("[preloaded_changes_archival_worker] DB query error") + return fmt.Errorf("failed to fetch preloaded_changes for review %d: %w", reviewID, err) + } + + if len(rawDiff) == 0 || bytes.Equal(rawDiff, []byte("null")) { + log.Info().Int64("review_id", reviewID).Msg("[preloaded_changes_archival_worker] empty diff, marking completed") + return nil + } + + // 2. Upload payload independently to Blob Storage + uploadErr := blobstore.SaveArtifact(ctx, w.db, orgID, reviewID, blobstore.ArtifactPreloadedChanges, rawDiff) + if uploadErr != nil { + log.Error().Err(uploadErr).Int64("review_id", reviewID).Msg("[preloaded_changes_archival_worker] blob upload failed") + + // If it's a fatal write permission error, cancel the job permanently so it doesn't infinitely retry + if strings.Contains(strings.ToLower(uploadErr.Error()), "permission denied") { + return river.JobCancel(fmt.Errorf("fatal permission error on blob storage: %w", uploadErr)) + } + + // Otherwise (e.g. rate limit, network drop), return normal error to trigger River's exponential retry + return fmt.Errorf("failed to upload diff for review %d: %w", reviewID, uploadErr) + } + + log.Info().Int64("review_id", reviewID).Msg("[preloaded_changes_archival_worker] blob upload succeeded, job completed") + return nil +} + +// PreloadedChangesArchivalPurgeWorker handles Steps 4 & 5: +// 4. If all upload jobs for BatchRunID are completed (or 6 hours have passed), executes ONE single SQL query to delete all preloaded_changes from metadata. +// 5. Clears created jobs from river_job table for this batch. +// ⚠️ ATTENTION FUTURE DEVELOPERS / AI AGENTS ⚠️ +// This worker does NOT use `MaxAttempts` hacking or error backoffs. +// It uses River's native `JobSnooze` to check if jobs are pending, and cleanly sleep for 30 minutes. +// If 6 hours pass since the batch started, it checks one last time and cancels if jobs are still pending (no data deleted). +// ⚠️ --------------------------------------- ⚠️ +type PreloadedChangesArchivalPurgeWorker struct { + river.WorkerDefaults[PreloadedChangesArchivalPurgeJobArgs] + db *sql.DB + jq *JobQueue +} + +func (w *PreloadedChangesArchivalPurgeWorker) Timeout(job *river.Job[PreloadedChangesArchivalPurgeJobArgs]) time.Duration { + return 10 * time.Minute +} + +// We removed the custom NextRetry backoff because this job no longer loops. +// It relies on ScheduledAt to wake up 30 minutes later and runs once. + +func (w *PreloadedChangesArchivalPurgeWorker) Work(ctx context.Context, job *river.Job[PreloadedChangesArchivalPurgeJobArgs]) error { + batchRunID := job.Args.BatchRunID + + // If 6 hours have passed since the Sweep started the batch, check one last time. + // If jobs are still pending, give up and discard β€” never force-purge incomplete batches. + if time.Since(job.CreatedAt) >= 6*time.Hour { + var pendingCount int + timeoutQuery := ` + SELECT COUNT(*) + FROM river_job + WHERE args @> jsonb_build_object('batch_run_id', $1::text) + AND kind = 'preloaded_changes_archival' + AND state NOT IN ('completed', 'discarded'); + ` + if err := w.db.QueryRowContext(ctx, timeoutQuery, batchRunID).Scan(&pendingCount); err != nil { + log.Error().Err(err).Str("batch_run_id", batchRunID).Msg("[preloaded_changes_purge_worker] failed to query pending jobs at 6h timeout") + return fmt.Errorf("failed to query pending jobs for batch %s: %w", batchRunID, err) + } + if pendingCount > 0 { + log.Warn().Str("batch_run_id", batchRunID).Int("pending", pendingCount).Msg("[preloaded_changes_purge_worker] 6 hours passed with jobs still pending β€” discarding purge job (no data deleted)") + return river.JobCancel(fmt.Errorf("batch %s timed out with %d pending upload jobs after 6 hours", batchRunID, pendingCount)) + } + log.Info().Str("batch_run_id", batchRunID).Msg("[preloaded_changes_purge_worker] 6 hours passed but all jobs completed β€” proceeding with purge") + } else { + // 1. O(1) check for pending jobs in Postgres + var pendingCount int + checkQuery := ` + SELECT COUNT(*) + FROM river_job + WHERE args @> jsonb_build_object('batch_run_id', $1::text) + AND kind = 'preloaded_changes_archival' + AND state NOT IN ('completed', 'discarded'); + ` + err := w.db.QueryRowContext(ctx, checkQuery, batchRunID).Scan(&pendingCount) + if err != nil { + log.Error().Err(err).Str("batch_run_id", batchRunID).Msg("[preloaded_changes_purge_worker] failed to query pending jobs") + return fmt.Errorf("failed to query pending jobs for batch %s: %w", batchRunID, err) + } + + if pendingCount > 0 { + // Two-phase exponential backoff within the 6-hour window: + // Phase 1 (first ~1h): 1m, 2m, 4m, 8m, 16m, 30m β€” catches fast completions quickly + // Phase 2 (1h–6h): 1h, 2h β€” avoids excessive DB polling + // At 6h: force-purge (handled above) + elapsed := time.Since(job.CreatedAt) + var snoozeDuration time.Duration + if elapsed < 1*time.Hour { + // Phase 1: minute-level exponential backoff + attempt := job.Attempt + if attempt < 1 { + attempt = 1 + } + snoozeDuration = time.Duration(1<<(attempt-1)) * time.Minute + if snoozeDuration > 30*time.Minute { + snoozeDuration = 30 * time.Minute + } + } else if elapsed < 3*time.Hour { + // Phase 2: check again in 1 hour + snoozeDuration = 1 * time.Hour + } else { + // Phase 2: check again in 2 hours (final check before 6h cutoff) + snoozeDuration = 2 * time.Hour + } + log.Info().Str("batch_run_id", batchRunID).Int("pending", pendingCount).Str("snooze", snoozeDuration.String()).Str("elapsed", elapsed.Round(time.Second).String()).Msg("[preloaded_changes_purge_worker] jobs still pending, snoozing with exponential backoff...") + // JobSnooze safely reschedules the job without counting it as a failure, + // and automatically extends MaxAttempts so it never permanently dies from snoozing. + return river.JobSnooze(snoozeDuration) + } + + log.Info().Str("batch_run_id", batchRunID).Msg("[preloaded_changes_purge_worker] ALL upload jobs done! Executing Step 4: single DB metadata purge") + } + + // 2. O(1) Fetch & Update (Postgres does all the work internally) + purgeQuery := ` + UPDATE reviews + SET metadata = metadata - 'preloaded_changes' + WHERE (id, org_id) IN ( + SELECT (args->>'review_id')::bigint, (args->>'org_id')::bigint + FROM river_job + WHERE args @> jsonb_build_object('batch_run_id', $1::text) + AND kind = 'preloaded_changes_archival' + AND state = 'completed' + ) + AND metadata ? 'preloaded_changes'; + ` + res, purgeErr := w.db.ExecContext(ctx, purgeQuery, batchRunID) + if purgeErr != nil { + log.Error().Err(purgeErr).Str("batch_run_id", batchRunID).Msg("[preloaded_changes_purge_worker] Step 4 failed: metadata purge error") + return fmt.Errorf("failed metadata purge for batch %s: %w", batchRunID, purgeErr) + } + rowsAffected, _ := res.RowsAffected() + log.Info().Int64("purged_reviews", rowsAffected).Str("batch_run_id", batchRunID).Msg("[preloaded_changes_purge_worker] Step 4 complete: metadata purged") + + // 3. O(1) Cleanup of river_job so no junk is left + cleanupQuery := ` + DELETE FROM river_job + WHERE args @> jsonb_build_object('batch_run_id', $1::text) + AND kind != 'preloaded_changes_archival_purge'; + ` + cleanupRes, cleanupErr := w.db.ExecContext(ctx, cleanupQuery, batchRunID) + if cleanupErr != nil { + log.Error().Err(cleanupErr).Str("batch_run_id", batchRunID).Msg("[preloaded_changes_purge_worker] Step 5 failed: job cleanup error") + return fmt.Errorf("failed job cleanup for batch %s: %w", batchRunID, cleanupErr) + } + cleanedJobs, _ := cleanupRes.RowsAffected() + log.Info().Int64("cleaned_jobs", cleanedJobs).Str("batch_run_id", batchRunID).Msg("[preloaded_changes_purge_worker] Step 5 complete: batch upload jobs cleaned from river_job") + + return nil +} + + diff --git a/internal/jobqueue/queue_config.go b/internal/jobqueue/queue_config.go index 7d2d5285..dc6309bc 100644 --- a/internal/jobqueue/queue_config.go +++ b/internal/jobqueue/queue_config.go @@ -106,6 +106,17 @@ type QueueConfig struct { // Repository/PR sync Configuration RepoSyncConfig RepoSyncConfig + + // Preloaded Changes Archival Configuration + PreloadedChangesArchivalConfig PreloadedChangesArchivalConfig +} + +// PreloadedChangesArchivalConfig controls the parallel preloaded_changes archival worker pool concurrency. +type PreloadedChangesArchivalConfig struct { + // MaxWorkers is the concurrency of the "preloaded_changes_archival" queue. + MaxWorkers int // default: 10 + // BatchSize is the number of reviews packaged into a single archival job. + BatchSize int // default: 10 } // RepoSyncConfig controls the periodic reconciliation sweep that catches PR/MR @@ -214,6 +225,10 @@ func DefaultQueueConfig() *QueueConfig { // Repository/PR sync configuration - overridable via env vars, see // repoSyncConfigFromEnv. RepoSyncConfig: repoSyncConfigFromEnv(), + + // Preloaded changes archival configuration - overridable via env vars, see + // preloadedChangesArchivalConfigFromEnv. + PreloadedChangesArchivalConfig: preloadedChangesArchivalConfigFromEnv(), } } @@ -245,6 +260,35 @@ func repoSyncConfigFromEnv() RepoSyncConfig { return config } +// preloadedChangesArchivalConfigFromEnv builds PreloadedChangesArchivalConfig from defaults, overridable via +// LIVEREVIEW_PRELOADED_CHANGES_ARCHIVAL_MAX_WORKERS (or LIVEREVIEW_DIFF_ARCHIVAL_MAX_WORKERS) so archival worker concurrency +// and batch size can be tuned per-deployment without a code change. +func preloadedChangesArchivalConfigFromEnv() PreloadedChangesArchivalConfig { + config := PreloadedChangesArchivalConfig{ + MaxWorkers: 10, + BatchSize: 10, + } + if v := os.Getenv("LIVEREVIEW_PRELOADED_CHANGES_ARCHIVAL_MAX_WORKERS"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + config.MaxWorkers = n + } + } else if v := os.Getenv("LIVEREVIEW_DIFF_ARCHIVAL_MAX_WORKERS"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + config.MaxWorkers = n + } + } + if v := os.Getenv("LIVEREVIEW_PRELOADED_CHANGES_ARCHIVAL_BATCH_SIZE"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + config.BatchSize = n + } + } else if v := os.Getenv("LIVEREVIEW_DIFF_ARCHIVAL_BATCH_SIZE"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + config.BatchSize = n + } + } + return config +} + // ProductionQueueConfig returns a configuration optimized for production use func ProductionQueueConfig() *QueueConfig { config := DefaultQueueConfig() @@ -340,6 +384,11 @@ func (c *QueueConfig) RiverQueueConfig() map[string]river.QueueConfig { repoSyncWorkers = 5 } + preloadedChangesArchivalWorkers := c.PreloadedChangesArchivalConfig.MaxWorkers + if preloadedChangesArchivalWorkers <= 0 { + preloadedChangesArchivalWorkers = 10 + } + return map[string]river.QueueConfig{ river.QueueDefault: { MaxWorkers: c.MaxWorkers, @@ -352,5 +401,14 @@ func (c *QueueConfig) RiverQueueConfig() map[string]river.QueueConfig { "repo_sync": { MaxWorkers: repoSyncWorkers, }, + "preloaded_changes_archival": { + MaxWorkers: preloadedChangesArchivalWorkers, + }, + "preloaded_changes_archival_sweep": { + MaxWorkers: 1, // Only needs 1 worker to enqueue the batch jobs + }, + "preloaded_changes_archival_purge": { + MaxWorkers: 1, // Only needs 1 worker to monitor and purge + }, } } diff --git a/internal/jobqueue/review_worker.go b/internal/jobqueue/review_worker.go index 410ed8c9..c6492758 100644 --- a/internal/jobqueue/review_worker.go +++ b/internal/jobqueue/review_worker.go @@ -3,13 +3,14 @@ package jobqueue import ( "context" "database/sql" + "encoding/json" "fmt" - "log" "strings" "time" "github.com/jackc/pgx/v5/pgxpool" "github.com/livereview/internal/aiselection" + "github.com/livereview/internal/blobstore" "github.com/livereview/internal/diffutil" "github.com/livereview/internal/license" "github.com/livereview/internal/logging" @@ -18,6 +19,7 @@ import ( reviewprocessor "github.com/livereview/internal/review_processor" "github.com/livereview/pkg/models" "github.com/riverqueue/river" + "github.com/rs/zerolog/log" ) // WebhookReviewJobArgs represents the arguments for an asynchronous webhook review job. @@ -48,7 +50,7 @@ func (w *WebhookReviewWorker) Timeout(job *river.Job[WebhookReviewJobArgs]) time func (w *WebhookReviewWorker) Work(ctx context.Context, job *river.Job[WebhookReviewJobArgs]) error { args := job.Args if w.jq == nil || w.jq.db == nil { - log.Printf("[ERROR] Database connection not available on JobQueue") + log.Error().Msg("[review_worker] Database connection not available on JobQueue") return fmt.Errorf("database connection not available") } return reviewprocessor.ProcessWebhookReview(ctx, w.jq.db, args.OrgID, args.ConnectorID, args.EventJSON, args.ScenarioType) @@ -84,7 +86,7 @@ func (w *ManualReviewWorker) Timeout(job *river.Job[ManualReviewJobArgs]) time.D func (w *ManualReviewWorker) Work(ctx context.Context, job *river.Job[ManualReviewJobArgs]) error { args := job.Args if w.jq == nil || w.jq.db == nil { - log.Printf("[ERROR] Database connection not available on JobQueue") + log.Error().Msg("[review_worker] Database connection not available on JobQueue") return fmt.Errorf("database connection not available") } return reviewprocessor.ProcessManualReview(ctx, w.jq.db, args.OrgID, args.PlanCode, args.ActorUserID, args.ActorEmail, args.ReviewID, args.RequestJSON, @@ -148,7 +150,7 @@ func (w *DiffReviewWorker) Work(ctx context.Context, job *river.Job[DiffReviewJo // 1. Initialize logger with event sink for UI polling stream logger, err := logging.StartReviewLoggingWithIDs(fmt.Sprintf("%d", args.ReviewID), args.ReviewID, args.OrgID) if err != nil { - log.Printf("[ERROR] Failed to start logging for review %d: %v", args.ReviewID, err) + log.Error().Err(err).Int64("review_id", args.ReviewID).Msg("[review_worker] Failed to start logging for review") } eventSink := reviewprocessor.NewDatabaseEventSink(w.db) @@ -200,15 +202,34 @@ func (w *DiffReviewWorker) Work(ctx context.Context, job *river.Job[DiffReviewJo // 3. Calculate Lines of Code billableLOC := diffutil.CalculateEffectiveDiffLOCFromLocalDiffs(localDiffs) - // 5. Convert diffs and persist preloaded_changes for UI polling + // 5. Convert diffs and persist to blob storage (blast radius pattern) modelDiffs := diffutil.ConvertLocalDiffs(localDiffs) - rm := reviewprocessor.NewReviewManager(w.db) - if err := rm.MergeReviewMetadata(args.ReviewID, map[string]interface{}{ - "preloaded_changes": modelDiffs, + modelDiffsPayload, err := json.Marshal(modelDiffs) + if err != nil { + w.handleFailure(ctx, args, logger, eventSink, fmt.Sprintf("failed to marshal diffs: %v", err), "failed_to_marshal_diffs") + return nil + } + + savedToBlob := false + if err := blobstore.SaveArtifact(ctx, w.db, args.OrgID, args.ReviewID, blobstore.ArtifactPreloadedChanges, modelDiffsPayload); err != nil { + log.Warn().Err(err).Int64("review_id", args.ReviewID).Int64("org_id", args.OrgID).Str("fallback_key", blobstore.ArtifactPreloadedChanges).Msg("[review_worker] Failed to store diff artifact in blob storage. Preserving in Postgres metadata fallback.") + } else { + savedToBlob = true + log.Info().Int64("review_id", args.ReviewID).Int64("org_id", args.OrgID).Msg("[review_worker] Successfully persisted diff artifact to blob storage") + } + + metaUpdates := map[string]interface{}{ "operation_billable_loc": billableLOC, "excluded_files": excludedFiles, - }); err != nil { - log.Printf("[WARN] failed to store preloaded_changes for review %d: %v", args.ReviewID, err) + } + if !savedToBlob { + // Safety fallback: if blob storage write failed, preserve diff in metadata + metaUpdates[blobstore.ArtifactPreloadedChanges] = modelDiffs + } + + rm := reviewprocessor.NewReviewManager(w.db) + if err := rm.MergeReviewMetadata(args.ReviewID, metaUpdates); err != nil { + log.Error().Err(err).Int64("review_id", args.ReviewID).Msg("[review_worker] Failed to persist review metadata") } // If .lrc/ignore excluded every changed file, there's nothing for the AI @@ -222,10 +243,10 @@ func (w *DiffReviewWorker) Work(ctx context.Context, job *river.Job[DiffReviewJo "comments": nil, }, }); err != nil { - log.Printf("[WARN] failed to store review_result for review %d: %v", args.ReviewID, err) + log.Warn().Err(err).Int64("review_id", args.ReviewID).Msg("[review_worker] failed to store review_result") } if err := rm.UpdateReviewStatus(args.ReviewID, "completed"); err != nil { - log.Printf("[WARN] failed to mark review %d completed: %v", args.ReviewID, err) + log.Warn().Err(err).Int64("review_id", args.ReviewID).Msg("[review_worker] failed to mark review completed") } if logger != nil { logger.Log("All files ignored. Completed review immediately.") @@ -311,7 +332,7 @@ func (w *DiffReviewWorker) Work(ctx context.Context, job *river.Job[DiffReviewJo if result.Success { status = "completed" if err := rm.MergeReviewMetadata(args.ReviewID, buildQueuedReviewAIMetadata(&reviewRequest, result)); err != nil { - log.Printf("[WARN] failed to persist AI stage metadata for review %d: %v", args.ReviewID, err) + log.Warn().Err(err).Int64("review_id", args.ReviewID).Msg("[review_worker] failed to persist AI stage metadata") } resolvedReviewID := args.ReviewID operationID := fmt.Sprintf("diff-review:%d", args.ReviewID) @@ -346,7 +367,7 @@ func (w *DiffReviewWorker) Work(ctx context.Context, job *river.Job[DiffReviewJo ExtraMeta: extraMeta, }) if err != nil { - log.Printf("[WARN] failed to queue billing finalization for review %d: %v", args.ReviewID, err) + log.Warn().Err(err).Int64("review_id", args.ReviewID).Msg("[review_worker] failed to queue billing finalization") } if logger != nil { @@ -389,11 +410,11 @@ func (w *DiffReviewWorker) Work(ctx context.Context, job *river.Job[DiffReviewJo meta["failure_reason"] = failureReason } if err := rm.MergeReviewMetadata(args.ReviewID, meta); err != nil { - log.Printf("[WARN] failed to persist review_result for %d: %v", args.ReviewID, err) + log.Warn().Err(err).Int64("review_id", args.ReviewID).Msg("[review_worker] failed to persist review_result") } if err := rm.UpdateReviewStatus(args.ReviewID, status); err != nil { - log.Printf("[WARN] failed to update review status for %d: %v", args.ReviewID, err) + log.Warn().Err(err).Int64("review_id", args.ReviewID).Msg("[review_worker] failed to update review status") } // Persist AI summary title for later display @@ -401,7 +422,7 @@ func (w *DiffReviewWorker) Work(ctx context.Context, job *river.Job[DiffReviewJo title := extractFirstHeading(summary) if title != "" { if err := rm.MergeReviewMetadata(args.ReviewID, map[string]interface{}{"ai_summary_title": title}); err != nil { - log.Printf("[WARN] failed to persist ai_summary_title for %d: %v", args.ReviewID, err) + log.Warn().Err(err).Int64("review_id", args.ReviewID).Msg("[review_worker] failed to persist ai_summary_title") } } } diff --git a/internal/review_processor/reviews.go b/internal/review_processor/reviews.go index 55913e8f..c9eeb01a 100644 --- a/internal/review_processor/reviews.go +++ b/internal/review_processor/reviews.go @@ -12,6 +12,7 @@ import ( // Review represents a code review record type Review struct { ID int64 `json:"id"` + OrgID int64 `json:"org_id"` Repository string `json:"repository"` Branch string `json:"branch"` CommitHash string `json:"commit_hash"` @@ -185,7 +186,7 @@ func (rm *ReviewManager) UpdateReviewConnector(reviewID int64, connectorID int64 // GetReview retrieves a review by ID func (rm *ReviewManager) GetReview(reviewID int64) (*Review, error) { query := ` - SELECT id, repository, branch, commit_hash, pr_mr_url, connector_id, + SELECT id, org_id, repository, branch, commit_hash, pr_mr_url, connector_id, status, trigger_type, user_email, provider, mr_title, friendly_name, author_name, author_username, created_at, started_at, completed_at, metadata FROM reviews @@ -194,8 +195,10 @@ func (rm *ReviewManager) GetReview(reviewID int64) (*Review, error) { var review Review var mrTitle, friendlyName, authorName, authorUsername sql.NullString + var orgID sql.NullInt64 err := rm.store.QueryRow(query, reviewID).Scan( &review.ID, + &orgID, &review.Repository, &review.Branch, &review.CommitHash, @@ -218,6 +221,9 @@ func (rm *ReviewManager) GetReview(reviewID int64) (*Review, error) { return nil, fmt.Errorf("failed to get review: %w", err) } + if orgID.Valid { + review.OrgID = orgID.Int64 + } if mrTitle.Valid { review.MRTitle = &mrTitle.String } @@ -238,7 +244,7 @@ func (rm *ReviewManager) GetReview(reviewID int64) (*Review, error) { // Returns an error if the review does not exist or belongs to a different org. func (rm *ReviewManager) GetReviewForOrg(reviewID int64, orgID int64) (*Review, error) { query := ` - SELECT id, repository, branch, commit_hash, pr_mr_url, connector_id, + SELECT id, org_id, repository, branch, commit_hash, pr_mr_url, connector_id, status, trigger_type, user_email, provider, mr_title, friendly_name, author_name, author_username, created_at, started_at, completed_at, metadata FROM reviews @@ -247,8 +253,10 @@ func (rm *ReviewManager) GetReviewForOrg(reviewID int64, orgID int64) (*Review, var review Review var mrTitle, friendlyName, authorName, authorUsername sql.NullString + var orgIDVal sql.NullInt64 err := rm.store.QueryRow(query, reviewID, orgID).Scan( &review.ID, + &orgIDVal, &review.Repository, &review.Branch, &review.CommitHash, @@ -271,6 +279,9 @@ func (rm *ReviewManager) GetReviewForOrg(reviewID int64, orgID int64) (*Review, return nil, fmt.Errorf("failed to get review: %w", err) } + if orgIDVal.Valid { + review.OrgID = orgIDVal.Int64 + } if mrTitle.Valid { review.MRTitle = &mrTitle.String } diff --git a/ui/src/components/Dashboard/widgets/DashboardGrid.tsx b/ui/src/components/Dashboard/widgets/DashboardGrid.tsx index 71a3f8c0..27b2b83c 100644 --- a/ui/src/components/Dashboard/widgets/DashboardGrid.tsx +++ b/ui/src/components/Dashboard/widgets/DashboardGrid.tsx @@ -24,6 +24,10 @@ interface DashboardGridProps { userId?: number | string; } +// Height of the sticky header a section scrolls under. Tailwind needs a literal +// class, so the anchors below repeat it as `scroll-mt-[140px]` - keep both in sync. +const SECTION_ANCHOR_OFFSET = 140; + // Rendered inside all three providers (not DashboardGrid itself) so it sits alongside the // widgets it refreshes. The dashboard query itself is a plain cheap GET (see useDashboardQuery // in api/dashboard.ts) - this button is the explicit, user-triggered path to the expensive @@ -104,7 +108,45 @@ export const DashboardGrid: React.FC = ({ userId }) => { return () => window.clearTimeout(timer); }, [activeSection]); + // Which section's pill is lit. Driven by scroll position so it tracks the + // grid as you move through it, not just when a pill is clicked. + const [visibleCategory, setVisibleCategory] = useState(null); + + // orderedCategories is a fresh array every render (activeWidgets isn't memoized), + // so key the effect on its contents to avoid rebinding listeners each render. + const categoryKey = orderedCategories.join(','); + useEffect(() => { + if (!categoryKey) return; + const categories = categoryKey.split(',') as WidgetCategory[]; + let frame = 0; + const update = () => { + frame = 0; + let current = categories[0]; + for (const category of categories) { + const el = document.getElementById(`dash-section-${category}`); + // Rounded so a sub-pixel scroll position can't leave a section + // sitting exactly at the offset unselected. + if (el && Math.round(el.getBoundingClientRect().top) <= SECTION_ANCHOR_OFFSET) current = category; + } + setVisibleCategory(current); + }; + const onScroll = () => { + if (frame) return; + frame = window.requestAnimationFrame(update); + }; + update(); + window.addEventListener('scroll', onScroll, { passive: true }); + window.addEventListener('resize', onScroll); + return () => { + if (frame) window.cancelAnimationFrame(frame); + window.removeEventListener('scroll', onScroll); + window.removeEventListener('resize', onScroll); + }; + }, [categoryKey]); + const goToSection = (category: WidgetCategory) => { + // Light it immediately - the smooth scroll takes ~half a second to arrive. + setVisibleCategory(category); setSearchParams((prev) => { const next = new URLSearchParams(prev); next.set('section', category); @@ -126,7 +168,7 @@ export const DashboardGrid: React.FC = ({ userId }) => { }, []); return ( - + @@ -137,19 +179,25 @@ export const DashboardGrid: React.FC = ({ userId }) => { {orderedCategories.length > 0 && (
- {orderedCategories.map((category) => ( - - ))} + {orderedCategories.map((category) => { + const isActive = visibleCategory === category; + return ( + + ); + })}
)} diff --git a/ui/src/components/Dashboard/widgets/DashboardPeriod.tsx b/ui/src/components/Dashboard/widgets/DashboardPeriod.tsx index 25b2ffc6..fe784eb9 100644 --- a/ui/src/components/Dashboard/widgets/DashboardPeriod.tsx +++ b/ui/src/components/Dashboard/widgets/DashboardPeriod.tsx @@ -1,4 +1,4 @@ -import React, { createContext, useContext, useMemo, useState } from 'react'; +import React, { createContext, useCallback, useContext, useEffect, useMemo, useState } from 'react'; export type DashboardPeriod = 'day' | 'week' | 'month' | 'all'; @@ -9,6 +9,30 @@ export const PERIOD_LABELS: Record = { all: 'All Time', }; +const DEFAULT_PERIOD: DashboardPeriod = 'all'; + +// Matches the `lr__${user.id}` localStorage convention already used by +// the dashboard layout and notifications (see useDashboardLayout.ts). +const storageKeyFor = (userId?: number | string): string => + userId ? `lr_dashboard_period_${userId}` : 'lr_dashboard_period'; + +function loadStoredPeriod(key: string): DashboardPeriod | null { + try { + const raw = localStorage.getItem(key); + return raw && raw in PERIOD_LABELS ? (raw as DashboardPeriod) : null; + } catch { + return null; + } +} + +function saveStoredPeriod(key: string, period: DashboardPeriod): void { + try { + localStorage.setItem(key, period); + } catch { + // no-op: keep the dashboard functional when localStorage is unavailable + } +} + // The mock volume numbers elsewhere in this feature represent a ~1 month baseline. // These multipliers rescale them for the other period options so the selector // actually changes what's on screen, without needing real time-series data yet. @@ -28,8 +52,26 @@ interface DashboardPeriodContextValue { const DashboardPeriodContext = createContext(null); -export const DashboardPeriodProvider: React.FC<{ children: React.ReactNode }> = ({ children }) => { - const [period, setPeriod] = useState('month'); +interface DashboardPeriodProviderProps { + children: React.ReactNode; + userId?: number | string; +} + +export const DashboardPeriodProvider: React.FC = ({ children, userId }) => { + const storageKey = storageKeyFor(userId); + const [period, setPeriodState] = useState(() => loadStoredPeriod(storageKey) ?? DEFAULT_PERIOD); + + // userId is undefined until /auth/me resolves, so re-read once the real + // per-user key is known - otherwise the saved range is never restored. + useEffect(() => { + setPeriodState(loadStoredPeriod(storageKey) ?? DEFAULT_PERIOD); + }, [storageKey]); + + // Persist on every change so the selected range survives a refresh. + const setPeriod = useCallback((next: DashboardPeriod) => { + setPeriodState(next); + saveStoredPeriod(storageKey, next); + }, [storageKey]); // Memoized so this only produces a new object when `period` actually changes - otherwise // every unrelated re-render higher up the tree (e.g. DashboardGrid's ResizeObserver-driven @@ -42,7 +84,7 @@ export const DashboardPeriodProvider: React.FC<{ children: React.ReactNode }> = setPeriod, label: PERIOD_LABELS[period], scale: (monthlyValue: number) => Math.max(0, Math.round(monthlyValue * PERIOD_MULTIPLIERS[period])), - }), [period]); + }), [period, setPeriod]); return ( diff --git a/ui/src/components/Dashboard/widgets/useDashboardLayout.ts b/ui/src/components/Dashboard/widgets/useDashboardLayout.ts index b180cca3..9268685f 100644 --- a/ui/src/components/Dashboard/widgets/useDashboardLayout.ts +++ b/ui/src/components/Dashboard/widgets/useDashboardLayout.ts @@ -1,4 +1,4 @@ -import { useCallback, useRef, useState } from 'react'; +import { useCallback, useEffect, useRef, useState } from 'react'; import type { Layout } from 'react-grid-layout/legacy'; import { WIDGET_REGISTRY, WidgetDefinition } from './registry'; @@ -55,43 +55,56 @@ function saveStored(key: string, data: StoredDashboardLayout): void { } } +// Merge any new widgets from the registry that weren't in the saved layout. +// This ensures new widgets appear automatically for users with a persisted layout. +function resolveLayout(stored: StoredDashboardLayout | null): Layout { + if (!stored) return defaultLayout(); + const savedIds = new Set(stored.widgetIds); + const newWidgets = WIDGET_REGISTRY.filter((w) => !savedIds.has(w.id)); + if (newWidgets.length === 0) return stored.layout; + return [ + ...stored.layout, + ...newWidgets.map((w) => ({ + i: w.id, + x: w.defaultLayout.x, + y: w.defaultLayout.y, + w: w.defaultLayout.w, + h: w.defaultLayout.h, + minW: w.minW, + minH: w.minH, + })), + ]; +} + +// Merge new registry widget IDs into the saved list, then sort to match +// registry order so new widgets appear at their intended position (not appended at end). +function resolveWidgetIds(stored: StoredDashboardLayout | null): string[] { + if (!stored) return defaultWidgetIds(); + const savedIds = new Set(stored.widgetIds); + const newIds = WIDGET_REGISTRY.filter((w) => !savedIds.has(w.id)).map((w) => w.id); + if (newIds.length === 0) return stored.widgetIds; + const merged = [...stored.widgetIds, ...newIds]; + const registryOrder = WIDGET_REGISTRY.map((w) => w.id); + merged.sort((a, b) => registryOrder.indexOf(a) - registryOrder.indexOf(b)); + return merged; +} + export function useDashboardLayout(userId?: number | string) { const storageKey = storageKeyFor(userId); - const [layout, setLayout] = useState(() => { + const [layout, setLayout] = useState(() => resolveLayout(loadStored(storageKey))); + const [widgetIds, setWidgetIds] = useState(() => resolveWidgetIds(loadStored(storageKey))); + const loadedKeyRef = useRef(storageKey); + + // userId is undefined until /auth/me resolves, so re-read once the real + // per-user key is known - otherwise the saved layout is never restored. + useEffect(() => { + if (loadedKeyRef.current === storageKey) return; + loadedKeyRef.current = storageKey; const stored = loadStored(storageKey); - if (!stored) return defaultLayout(); - // Merge any new widgets from the registry that weren't in the saved layout. - // This ensures new widgets appear automatically for users with a persisted layout. - const savedIds = new Set(stored.widgetIds); - const newWidgets = WIDGET_REGISTRY.filter((w) => !savedIds.has(w.id)); - if (newWidgets.length === 0) return stored.layout; - const mergedLayout: Layout = [ - ...stored.layout, - ...newWidgets.map((w) => ({ - i: w.id, - x: w.defaultLayout.x, - y: w.defaultLayout.y, - w: w.defaultLayout.w, - h: w.defaultLayout.h, - minW: w.minW, - minH: w.minH, - })), - ]; - return mergedLayout; - }); - const [widgetIds, setWidgetIds] = useState(() => { - const stored = loadStored(storageKey); - if (!stored) return defaultWidgetIds(); - // Merge new registry widget IDs into the saved list, then sort to match - // registry order so new widgets appear at their intended position (not appended at end). - const savedIds = new Set(stored.widgetIds); - const newIds = WIDGET_REGISTRY.filter((w) => !savedIds.has(w.id)).map((w) => w.id); - if (newIds.length === 0) return stored.widgetIds; - const merged = [...stored.widgetIds, ...newIds]; - const registryOrder = WIDGET_REGISTRY.map((w) => w.id); - merged.sort((a, b) => registryOrder.indexOf(a) - registryOrder.indexOf(b)); - return merged; - }); + setLayout(resolveLayout(stored)); + setWidgetIds(resolveWidgetIds(stored)); + }, [storageKey]); + const [editMode, setEditMode] = useState(false); const saveTimerRef = useRef(null); diff --git a/ui/src/components/Navbar/megaMenuData.ts b/ui/src/components/Navbar/megaMenuData.ts index 8bf1d5bc..55968d2a 100644 --- a/ui/src/components/Navbar/megaMenuData.ts +++ b/ui/src/components/Navbar/megaMenuData.ts @@ -244,6 +244,7 @@ export const buildMegaMenuSections = (): MegaMenuSection[] => [ group('Manage System', [ link('Storage', React.createElement(Icons.Folder), '/settings#storage', (ctx) => ctx.isSuperAdmin || (ctx.orgRole === 'owner' && !isCloudMode())), link('Log Compaction', React.createElement(Icons.Clock), '/settings#storage', (ctx) => ctx.isSuperAdmin), + link('Preloaded Changes Archival', React.createElement(Icons.Folder), '/settings#storage', (ctx) => ctx.isSuperAdmin || (ctx.orgRole === 'owner' && !isCloudMode())), ], React.createElement(Icons.Settings)), ], }, diff --git a/ui/src/components/reviews/cronbuilder/CronBuilder.tsx b/ui/src/components/reviews/cronbuilder/CronBuilder.tsx index 14cab0f8..d72df30c 100644 --- a/ui/src/components/reviews/cronbuilder/CronBuilder.tsx +++ b/ui/src/components/reviews/cronbuilder/CronBuilder.tsx @@ -258,16 +258,107 @@ export function CronBuilder({ onChange, defaultValue, className }: CronBuilderPr setHours((prev) => (prev.includes(hour) ? prev.filter((h) => h !== hour) : [...prev, hour])); }, []); - const renderHoursGrid = () => ( -
- -
- {HOURS.map((hour) => ( - - ))} + const [use12h, setUse12h] = useState(true); + const [amPm, setAmPm] = useState<'AM' | 'PM'>(() => { + // Default to the period of the first selected hour + const firstHour = hours[0] ?? 9; + return firstHour >= 12 ? 'PM' : 'AM'; + }); + + // When switching AM/PM, remap all currently selected hours to the new period + const handleAmPmSwitch = useCallback((newPeriod: 'AM' | 'PM') => { + setAmPm(newPeriod); + setHours((prev) => prev.map((h) => { + const h12 = h % 12; // 0-11 + return newPeriod === 'PM' ? (h12 === 0 ? 12 : h12 + 12) : h12; + })); + }, []); + + const HOURS_12 = Array.from({ length: 12 }, (_, i) => i); // 0..11 (display as 12,1,2..11) + + const renderHoursGrid = () => { + if (!use12h) { + // 24-hour mode (original) + return ( +
+
+ + +
+
+ {HOURS.map((hour) => ( + + ))} +
+
+ ); + } + + // 12-hour AM/PM mode + const offset = amPm === 'PM' ? 12 : 0; + const display12 = (h12: number) => { + if (h12 === 0) return '12'; + return h12.toString(); + }; + + return ( +
+
+ + +
+ + {/* AM / PM toggle */} +
+ {(['AM', 'PM'] as const).map((period) => ( + + ))} +
+ + {/* 12-hour grid: 12, 1, 2, 3 ... 11 */} +
+ {HOURS_12.map((h12) => { + const h24 = h12 === 0 ? offset : h12 + offset; // 12AM=0, 1AM=1...11AM=11, 12PM=12, 1PM=13...11PM=23 + const isSelected = hours.includes(h24); // h24 is guaranteed to be 0-23 + return ( + + ); + })} +
-
- ); + ); + }; const handleMinuteToggle = useCallback((minute: number) => { setMinutes((prev) => (prev.includes(minute) ? prev.filter((m) => m !== minute) : [...prev, minute])); diff --git a/ui/src/components/reviews/diffviewer/CommentThread.tsx b/ui/src/components/reviews/diffviewer/CommentThread.tsx index 28da9c69..1215a4e4 100644 --- a/ui/src/components/reviews/diffviewer/CommentThread.tsx +++ b/ui/src/components/reviews/diffviewer/CommentThread.tsx @@ -19,6 +19,7 @@ interface CommentThreadProps { filePath: string; comments: { comment: DiffReviewComment; idx: number }[]; hunkBlastDetail?: BlastRadiusHunkReport; + codeExcerpt?: string; // Opens the same BlastRadiusPanel the hunk header's own RiskBadge opens β€” // git-lrc repeats the hunk's risk pill on every comment's action row "so // score and comment are always assessed together" (RiskBadge.js), and both @@ -28,41 +29,50 @@ interface CommentThreadProps { function buildMetaItems(comment: DiffReviewComment): { label: string; value: string }[] { const items: { label: string; value: string }[] = []; - if (comment.confidence) items.push({ label: 'CONFIDENCE', value: comment.confidence }); - if (comment.type) items.push({ label: 'TYPE', value: comment.type }); + if (comment.confidence) items.push({ label: 'Confidence', value: comment.confidence }); + if (comment.type) items.push({ label: 'Type', value: comment.type }); if (comment.category || comment.subcategory) { items.push({ - label: 'CLASSIFICATION', + label: 'Classification', value: `${comment.category || 'Uncategorized'}${comment.subcategory ? ` / ${comment.subcategory}` : ''}`, }); } return items; } -function buildCopyText(filePath: string, comment: DiffReviewComment): string { - const parts = [`${filePath}:${comment.line}`]; - const severity = (comment.severity || 'info').toUpperCase(); - parts.push(`[${severity}]`); - parts.push(comment.content); - return parts.join(' '); +function buildCopyText(filePath: string, comment: DiffReviewComment, codeExcerpt?: string): string { + let copyText = ''; + if (filePath) { + copyText += filePath; + if (comment.line) { + copyText += ':' + comment.line; + } + copyText += '\n\n'; + } + if (codeExcerpt) { + copyText += 'Code excerpt:\n' + codeExcerpt + '\n\n'; + } + copyText += 'Issue:\n' + comment.content; + return copyText; } const CommentCard: React.FC<{ id: string; reviewId: number; filePath: string; comment: DiffReviewComment; hunkBlastDetail?: BlastRadiusHunkReport; + codeExcerpt?: string; onOpenBreakdown?: () => void; -}> = ({ id, reviewId, filePath, comment, hunkBlastDetail, onOpenBreakdown }) => { +}> = ({ id, reviewId, filePath, comment, hunkBlastDetail, codeExcerpt, onOpenBreakdown }) => { const [hidden, setHidden] = useState(false); const [copyLabel, setCopyLabel] = useState(null); const metaItems = buildMetaItems(comment); const handleCopy = useCallback(() => { - navigator.clipboard.writeText(buildCopyText(filePath, comment)).then(() => { + navigator.clipboard.writeText(buildCopyText(filePath, comment, codeExcerpt)).then(() => { setCopyLabel('Copied!'); window.setTimeout(() => setCopyLabel(null), 2000); }); - }, [filePath, comment]); + }, [filePath, comment, codeExcerpt]); if (hidden) { return ( @@ -74,8 +84,12 @@ const CommentCard: React.FC<{
@@ -83,61 +97,89 @@ const CommentCard: React.FC<{ } return ( -
-
-
- {hunkBlastDetail && typeof hunkBlastDetail.Combined === 'number' && ( - - )} +
+ {/* Header row β€” mb: 10px, gap: 12px matches git-lrc .comment-header */} +
+
{(comment.severity || 'info').toUpperCase()} {filePath}:{comment.line} - {metaItems.map((item, i) => ( - - {i > 0 && β€’} - - {item.label} {item.value} - - - ))}
-
+
+ {hunkBlastDetail && typeof hunkBlastDetail.Combined === 'number' && ( + + )}
-

{comment.content}

+ {/* Meta row β€” gap: 8px, mb: 12px matches git-lrc .comment-meta-line */} + {metaItems.length > 0 && ( +
+ {metaItems.map((item, i) => ( + + {i > 0 && β€’} + + {item.label} + {item.value} + + + ))} +
+ )} + {/* Body β€” subtle border-top (rgba 12%), font-size 14px, line-height 1.72 matches git-lrc .comment-body */} +

+ {comment.content} +

); }; -const CommentThread: React.FC = ({ reviewId, filePath, comments, hunkBlastDetail, onOpenBreakdown }) => { +const CommentThread: React.FC = ({ reviewId, filePath, comments, hunkBlastDetail, codeExcerpt, onOpenBreakdown }) => { if (!comments.length) return null; return ( -
+
{comments.map(({ comment, idx }) => ( = ({ reviewId, filePath, comme filePath={filePath} comment={comment} hunkBlastDetail={hunkBlastDetail} + codeExcerpt={codeExcerpt} onOpenBreakdown={onOpenBreakdown} /> ))} diff --git a/ui/src/components/reviews/diffviewer/DiffViewerPanel.tsx b/ui/src/components/reviews/diffviewer/DiffViewerPanel.tsx index 850b189d..18b0e890 100644 --- a/ui/src/components/reviews/diffviewer/DiffViewerPanel.tsx +++ b/ui/src/components/reviews/diffviewer/DiffViewerPanel.tsx @@ -175,7 +175,7 @@ const DiffViewerPanel: React.FC = ({ reviewId }) => { return enrichedFiles; }, [enrichedFiles, sortMode, canSortByRisk]); - const facets = useMemo(() => buildFilterFacets(files, filters), [files, filters]); + const facets = useMemo(() => buildFilterFacets(enrichedFiles, filters), [enrichedFiles, filters]); const navComments = useMemo(() => buildVisibleCommentNav(files, filters), [files, filters]); const allExpanded = files.length > 0 && files.every((f) => expandedFiles[f.file_path]); // Sidebar always shows the real, unflattened file list (git-lrc's diff --git a/ui/src/components/reviews/diffviewer/HunkBlock.tsx b/ui/src/components/reviews/diffviewer/HunkBlock.tsx index e6e5a4dc..69acb637 100644 --- a/ui/src/components/reviews/diffviewer/HunkBlock.tsx +++ b/ui/src/components/reviews/diffviewer/HunkBlock.tsx @@ -6,7 +6,7 @@ import React, { useRef, useState } from 'react'; import classNames from 'classnames'; import { DiffReviewComment, DiffReviewHunk } from '../../../types/reviews'; -import { commentBelongsToLine, DiffLine, hunkDomId, parseHunkLines, scrollElementIntoViewBelowStickyBars } from './diffUtils'; +import { buildIssueCodeExcerpt, commentBelongsToLine, DiffLine, hunkDomId, parseHunkLines, scrollElementIntoViewBelowStickyBars } from './diffUtils'; import { commentMatchesFilters, IssueFilters } from './issueFilters'; import CommentThread from './CommentThread'; import RiskBadge from './RiskBadge'; @@ -101,7 +101,7 @@ const HunkBlock: React.FC = ({ reviewId, filePath, navId, hunk, {lines.map((line, idx) => { - const lineComments = comments.filter(({ comment }) => commentBelongsToLine(comment, line) && commentMatchesFilters(comment, filters)); + const lineComments = comments.filter(({ comment }) => commentBelongsToLine(comment, line, lines) && commentMatchesFilters(comment, filters)); return ( @@ -114,12 +114,15 @@ const HunkBlock: React.FC = ({ reviewId, filePath, navId, hunk, {lineComments.length > 0 && ( - diff --git a/ui/src/components/reviews/diffviewer/RiskBadge.tsx b/ui/src/components/reviews/diffviewer/RiskBadge.tsx index 698c1450..ce4520c4 100644 --- a/ui/src/components/reviews/diffviewer/RiskBadge.tsx +++ b/ui/src/components/reviews/diffviewer/RiskBadge.tsx @@ -59,13 +59,21 @@ const RiskBadge: React.FC = ({ score, detail, size = 'small', on onClick={clickable ? (e) => { e.stopPropagation(); onOpen!(); } : undefined} aria-label={`Risk score ${Math.round(score)} out of 100`} className={classNames( - 'inline-flex items-center gap-1 rounded-full border px-2 py-0.5 font-mono text-xs font-semibold', + 'inline-flex h-7 items-center gap-1 rounded-full border px-2.5 font-mono text-xs font-semibold', TIER_CLASSES[tier], clickable && 'cursor-pointer hover:brightness-125', size === 'large' ? 'text-xs' : 'text-[11px]' )} > - {Math.round(score)} + + + + {Math.round(score)} + {clickable && ( + + + + )} ); diff --git a/ui/src/components/reviews/diffviewer/VoteButtons.tsx b/ui/src/components/reviews/diffviewer/VoteButtons.tsx index e498c8a4..70fb69da 100644 --- a/ui/src/components/reviews/diffviewer/VoteButtons.tsx +++ b/ui/src/components/reviews/diffviewer/VoteButtons.tsx @@ -1,28 +1,10 @@ -// Ported from git-lrc:internal/staticserve/static/components/FeedbackPopup.js (as of -// the git-lrc HEAD current when this port was written) β€” full popup UX: impact stats, -// downvote reason tags, free-text feedback, LinkedIn share overlay. The vote itself -// calls LiveReview's real feedback API (internal/api/feedback_handler.go). +// Ported from git-lrc:internal/staticserve/static/components/FeedbackPopup.js (as of HEAD) import React, { useCallback, useEffect, useRef, useState } from 'react'; -import { FeedbackSourceType, getImpactStats, ImpactStat, retractFeedback, submitFeedback } from '../../../api/feedback'; +import { createPortal } from 'react-dom'; +import { FeedbackSourceType, retractFeedback, submitFeedback, getImpactStats, ImpactStat } from '../../../api/feedback'; const DOWN_TAGS = ['False positive', 'Wrong severity', 'Missed something', 'Hard to act on']; -function buildLinkedinText(stats: ImpactStat[] | null): string { - const v = (label: string) => { const s = (stats || []).find((x) => x.label === label); return s ? s.value : '-'; }; - return `Shipping with confidence β€” here's my code review impact since Jan 2025: - -${v('Total Reviews')} reviews completed -${v('Bugs Caught Pre-Prod')} bugs caught before production -${v('Issues Found')} total issues found -${v('Critical')} critical issues found -${v('Errors')} errors caught -${v('Warnings')} warnings flagged - -Using LiveReview to AI-review every commit before it lands. - -#CodeReview #DevOps #SoftwareEngineering #AI`; -} - interface VoteButtonsProps { reviewId: number; sourceType: FeedbackSourceType; @@ -37,8 +19,36 @@ interface VoteButtonsProps { type VoteState = 'up' | 'down' | null; type PopupMode = 'hover' | 'click' | 'submitted' | null; +const buildLinkedinText = (stats: ImpactStat[] | null) => { + const get = (label: string) => { + const s = (stats || []).find((x) => x.label === label); + return s != null ? s.value : 'β€”'; + }; + return `πŸš€ Shipping with confidence β€” here's my code review impact since Jan 2025: + +βœ… ${get('Total Reviews')} reviews completed +πŸ› ${get('Bugs Caught Pre-Prod')} bugs caught before production +πŸ” ${get('Issues Found')} total issues found +πŸ”΄ ${get('Critical')} critical issues found +🟠 ${get('Errors')} errors caught +🟑 ${get('Warnings')} warnings flagged + +Using LiveReview to AI-review every commit before it lands. + +⭐ Star it if you find it useful: https://github.com/HexmosTech/LiveReview + +#CodeReview #DevOps #SoftwareEngineering #AI`; +}; + const VoteButtons: React.FC = ({ - reviewId, sourceType, aiCommentId, commentContent, codeExcerpt, filePath, severity, size = 'sm', + reviewId, + sourceType, + aiCommentId, + commentContent, + codeExcerpt, + filePath, + severity, + size = 'sm', }) => { const wrapperRef = useRef(null); const popupRef = useRef(null); @@ -50,48 +60,42 @@ const VoteButtons: React.FC = ({ const [popupVisible, setPopupVisible] = useState(false); const [popupMode, setPopupMode] = useState(null); + const [popupSource, setPopupSource] = useState<'up' | 'down' | null>(null); const [popupPos, setPopupPos] = useState({ top: 0, left: 0 }); const [popupAnim, setPopupAnim] = useState({ opacity: 0, shift: -6 }); const [feedbackText, setFeedbackText] = useState(''); const [selectedTags, setSelectedTags] = useState>(new Set()); - const [statsExpanded, setStatsExpanded] = useState(false); + const [submitting, setSubmitting] = useState(false); + const [submitError, setSubmitError] = useState(false); + const [impactStats, setImpactStats] = useState(null); + const [statsExpanded, setStatsExpanded] = useState(false); const [linkedinOpen, setLinkedinOpen] = useState(false); + const [linkedinOpacity, setLinkedinOpacity] = useState(0); + const [linkedinText, setLinkedinText] = useState(''); const [snackbar, setSnackbar] = useState(false); - const [submitting, setSubmitting] = useState(false); - const [submitError, setSubmitError] = useState(false); const autoTimer = useRef(null); const hoverTimer = useRef(null); + const snackTimer = useRef(null); const clearTimers = useCallback(() => { - if (autoTimer.current) { clearTimeout(autoTimer.current); autoTimer.current = null; } - if (hoverTimer.current) { clearTimeout(hoverTimer.current); hoverTimer.current = null; } + if (autoTimer.current) { window.clearTimeout(autoTimer.current); autoTimer.current = null; } + if (hoverTimer.current) { window.clearTimeout(hoverTimer.current); hoverTimer.current = null; } + if (snackTimer.current) { window.clearTimeout(snackTimer.current); snackTimer.current = null; } }, []); useEffect(() => clearTimers, [clearTimers]); - const isActive = vote === 'up' || vote === 'down'; - - const postFeedback = useCallback((extra: Record = {}) => { - try { - const body: Record = { - review_id: reviewId, - vote_type: vote!, - source_type: sourceType, - tags: [...selectedTags], - ...(commentContent && { comment_content: commentContent }), - ...(filePath && { file_path: filePath }), - ...(severity && { severity }), - ...(codeExcerpt && { code_excerpt: codeExcerpt }), - ...extra, - }; - submitFeedback(body as any).then((res) => { - if (res?.id) setFeedbackId(res.id); - }).catch(() => {}); - } catch {} - }, [reviewId, vote, sourceType, selectedTags, commentContent, filePath, severity, codeExcerpt]); + useEffect(() => { + if (!linkedinOpen) return; + const handler = (e: KeyboardEvent) => { + if (e.key === 'Escape') closeLinkedin(); + }; + document.addEventListener('keydown', handler); + return () => document.removeEventListener('keydown', handler); + }, [linkedinOpen]); const retract = useCallback(() => { if (feedbackId !== null) { @@ -100,9 +104,32 @@ const VoteButtons: React.FC = ({ } }, [feedbackId]); - const show = useCallback((mode: PopupMode) => { + const calculatePos = (wrapperEl: HTMLElement, popupEl?: HTMLElement | null) => { + const r = wrapperEl.getBoundingClientRect(); + const w = 400; + const viewportWidth = document.documentElement.clientWidth || window.innerWidth; + const viewportHeight = document.documentElement.clientHeight || window.innerHeight; + + let targetLeft = r.right - w; + if (targetLeft + w > viewportWidth - 16) targetLeft = viewportWidth - w - 16; + if (targetLeft < 16) targetLeft = 16; + + const h = popupEl?.offsetHeight || 260; + let top = r.bottom + 4; + if (r.top >= viewportHeight / 2) { + top = Math.max(16, r.top - 4 - h); + } + + return { top, left: targetLeft }; + }; + + const show = useCallback((mode: PopupMode, source?: 'up' | 'down') => { + if (wrapperRef.current) { + setPopupPos(calculatePos(wrapperRef.current, popupRef.current)); + } setPopupVisible(true); setPopupMode(mode); + if (source) setPopupSource(source); setPopupAnim({ opacity: 0, shift: -6 }); }, []); @@ -111,31 +138,42 @@ const VoteButtons: React.FC = ({ window.setTimeout(() => { setPopupVisible(false); setPopupMode(null); + setPopupSource(null); setStatsExpanded(false); }, 280); }, []); - const startAuto = useCallback((ms = 5000) => { - if (autoTimer.current) clearTimeout(autoTimer.current); + const startAuto = useCallback((ms = 6000) => { + if (autoTimer.current) window.clearTimeout(autoTimer.current); autoTimer.current = window.setTimeout(hide, ms); }, [hide]); useEffect(() => { - if (!popupVisible || !popupRef.current || !wrapperRef.current) return; - const r = wrapperRef.current.getBoundingClientRect(); - const w = 420; - const left = Math.max(8, Math.min(r.right - w, window.innerWidth - w - 8)); - const belowTop = r.bottom + 8; - const h = popupRef.current.offsetHeight; - const top = h > 0 && belowTop + h > window.innerHeight ? Math.max(8, r.top - h - 8) : belowTop; - setPopupPos({ top, left }); + if (!popupVisible || !wrapperRef.current || !popupRef.current) return; + setPopupPos(calculatePos(wrapperRef.current, popupRef.current)); if (popupAnim.opacity === 0) { requestAnimationFrame(() => requestAnimationFrame(() => { setPopupAnim({ opacity: 1, shift: 0 }); })); } // eslint-disable-next-line react-hooks/exhaustive-deps - }, [popupVisible, popupMode]); + }, [popupVisible, popupMode, statsExpanded]); + + // Close popup on click outside + useEffect(() => { + if (!popupVisible) return; + const handleClickOutside = (e: MouseEvent) => { + if ( + popupRef.current && !popupRef.current.contains(e.target as Node) && + wrapperRef.current && !wrapperRef.current.contains(e.target as Node) && + !linkedinOpen + ) { + hide(); + } + }; + document.addEventListener('mousedown', handleClickOutside); + return () => document.removeEventListener('mousedown', handleClickOutside); + }, [popupVisible, hide, linkedinOpen]); const cast = async (next: VoteState) => { if (busy) return; @@ -143,7 +181,14 @@ const VoteButtons: React.FC = ({ setDenied(false); try { retract(); - if (next === vote) { setVote(null); if (popupVisible) hide(); return; } + if (next === vote) { + setVote(null); + if (popupVisible) hide(); + return; + } + if (next === 'up') { + getImpactStats((stats) => setImpactStats(stats)); + } const res = await submitFeedback({ review_id: reviewId, ai_comment_id: aiCommentId, @@ -157,12 +202,11 @@ const VoteButtons: React.FC = ({ setFeedbackId(res.id); setVote(next); if (next === 'up') { - getImpactStats((stats) => setImpactStats(stats)); - show('click'); - startAuto(); + show('click', 'up'); + startAuto(10000); } else { - show('click'); - startAuto(); + show('click', 'down'); + startAuto(10000); } } catch (err) { if ((err as any)?.status === 403) { @@ -174,12 +218,20 @@ const VoteButtons: React.FC = ({ } }; - const handleMouseEnter = () => { - if (hoverTimer.current) { clearTimeout(hoverTimer.current); hoverTimer.current = null; } + const handleLikeMouseEnter = () => { + if (hoverTimer.current) { window.clearTimeout(hoverTimer.current); hoverTimer.current = null; } if (popupMode === 'click' || popupMode === 'submitted') return; - if (!popupVisible) { + if (!popupVisible || popupMode !== 'hover' || popupSource !== 'up') { getImpactStats((stats) => setImpactStats(stats)); - show('hover'); + show('hover', 'up'); + } + }; + + const handleDislikeMouseEnter = () => { + if (hoverTimer.current) { window.clearTimeout(hoverTimer.current); hoverTimer.current = null; } + if (popupMode === 'click' || popupMode === 'submitted') return; + if (vote === 'down' && (!popupVisible || popupMode !== 'hover' || popupSource !== 'down')) { + show('hover', 'down'); } }; @@ -187,29 +239,30 @@ const VoteButtons: React.FC = ({ if (popupMode === 'click' || popupMode === 'submitted') return; hoverTimer.current = window.setTimeout(() => { if (popupMode === 'hover') hide(); - }, 80); + }, 300); }; const onPopupEnter = () => { - clearTimers(); + if (hoverTimer.current) { window.clearTimeout(hoverTimer.current); hoverTimer.current = null; } }; const onPopupLeave = () => { - if (popupMode === 'click' || popupMode === 'submitted') hide(); - else { - hoverTimer.current = window.setTimeout(() => { if (popupMode === 'hover') hide(); }, 80); - } + if (popupMode === 'click' || popupMode === 'submitted') return; + hoverTimer.current = window.setTimeout(() => { + if (popupMode === 'hover') hide(); + }, 200); }; const handleSubmit = async (e: React.MouseEvent) => { e.stopPropagation(); - clearTimers(); + if (autoTimer.current) { window.clearTimeout(autoTimer.current); autoTimer.current = null; } setSubmitError(false); setSubmitting(true); try { await submitFeedback({ review_id: reviewId, - vote_type: vote!, + ai_comment_id: aiCommentId, + vote_type: (popupSource || vote || 'down') as 'up' | 'down', source_type: sourceType, tags: [...selectedTags], feedback_text: feedbackText, @@ -226,197 +279,361 @@ const VoteButtons: React.FC = ({ } setSubmitting(false); setPopupMode('submitted'); + startAuto(3000); + }; + + const openLinkedin = () => { + setLinkedinText(buildLinkedinText(impactStats)); + setLinkedinOpen(true); + setLinkedinOpacity(0); + requestAnimationFrame(() => requestAnimationFrame(() => setLinkedinOpacity(1))); + }; + + const closeLinkedin = () => { + setLinkedinOpacity(0); + setTimeout(() => setLinkedinOpen(false), 200); }; - const btnSize = size === 'sm' ? 'text-xs px-1.5 py-0.5' : 'text-sm px-2 py-1'; + const handleCopyLinkedin = async (e: React.MouseEvent) => { + e.stopPropagation(); + try { + await navigator.clipboard.writeText(linkedinText); + setSnackbar(true); + if (snackTimer.current) window.clearTimeout(snackTimer.current); + snackTimer.current = window.setTimeout(() => setSnackbar(false), 2200); + } catch {} + }; + + const dimClass = size === 'sm' ? 'h-7 w-7' : 'h-8 w-8'; if (denied) { return Feedback unavailable; } - const popupWidth = 420; + const ImpactLink = () => + statsExpanded ? ( +
+ + + + Want to see your impact stats? +
+ ) : ( +
setStatsExpanded(true)} + > + + + + Want to see your impact stats? + + + +
+ ); + + const StatsGrid = () => { + if (!impactStats) return
Loading stats…
; + return ( +
+
+ {impactStats.map((s) => ( +
+
{s.value}
+
{s.label}
+
+ ))} +
+
+
+ + + + + + Stand out by showing your impact stats to your peers + + + +
+
+
+ ); + }; return ( -
+
+ {/* Upvote button */} + > + + + + + + + {/* Downvote button */} - - {popupVisible && ( -
- {vote === 'up' ? ( - // ── Upvote popup: impact stats + "like" text ── -
-

- {popupMode === 'submitted' ? 'Thanks for your feedback!' : 'This was helpful?'} -

- {impactStats && ( -
+ > + + + + + + + {/* Feedback Popup Box */} + {popupVisible && + createPortal( +
e.stopPropagation()} + > + {popupMode === 'hover' && popupSource === 'up' && ( +
+ + {statsExpanded && } +
+ )} + + {popupMode === 'submitted' && ( +
+ + + + + + {popupSource === 'up' ? 'Thanks for your detailed feedback!' : "Thanks. We'll work on making it better."} + +
+ )} + + {popupMode === 'click' && ( +
+
+
+ {popupSource === 'up' ? ( + + + + + ) : ( + + + + + )} + + {popupSource === 'up' ? 'Thanks for your feedback!' : "We're sorry it didn't meet your expectations!"} + +
- {statsExpanded && ( -
- {impactStats.map((s) => ( -
-
{s.value ?? '-'}
-
{s.label}
-
- ))} -
- )}
- )} - {popupMode !== 'submitted' && ( -
-
+ + +