Compare commits

...

38 Commits

Author SHA1 Message Date
Kelly
0295637ed6 fix: Public API column mappings and OOS detection
- Fix store_products column references (name_raw, brand_name_raw, category_raw)
- Fix v_product_snapshots column references (crawled_at, *_cents pricing)
- Fix dispensaries column references (zipcode, logo_image, remove hours/amenities)
- Add services and license_type to dispensary API response
- Add consecutive_misses OOS tracking to product-resync handler
- Add migration 075 for consecutive_misses column
- Add CRAWL_PIPELINE.md documentation

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 20:44:53 -07:00
Kelly
9c6dd37316 fix(ci): Use YAML list format for docker-buildx build_args
The woodpecker docker-buildx plugin requires build_args as a YAML list,
not a comma-separated string. This fixes the build version/hash not being
passed to the Docker image.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 18:03:50 -07:00
kelly
524d13209a Merge pull request 'fix: Remove legacy imports from task handlers' (#9) from fix/task-handler-typescript-errors into master
Reviewed-on: https://code.cannabrands.app/Creationshop/dispensary-scraper/pulls/9
2025-12-10 00:42:39 +00:00
Kelly
9199db3927 fix: Remove legacy imports from task handlers
- Remove non-existent DutchieClient import from product-resync and entry-point-discovery
- Remove non-existent DiscoveryCrawler import from store-discovery
- Use scrapeStore from scraper-v2 for product resync
- Use discoverState from discovery module for store discovery
- Fix Pool type by using getPool() instead of pool wrapper
- Update FullDiscoveryResult property access to use correct field names

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 17:25:19 -07:00
Kelly
a0652c7c73 fix(types): Fix TypeScript errors in TasksDashboard, Layout, and Users
- Fix TaskCounts type in api.ts to match TasksDashboard interface
- Make VersionInfo.version optional in Layout.tsx
- Fix boolean type in Users.tsx disabled prop

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 17:02:40 -07:00
Kelly
89c262ee20 feat(tasks): Add unified task-based worker architecture
Replace fragmented job systems (job_schedules, dispensary_crawl_jobs, SyncOrchestrator)
with a single unified task queue:

- Add worker_tasks table with atomic task claiming via SELECT FOR UPDATE SKIP LOCKED
- Add TaskService for CRUD, claiming, and capacity metrics
- Add TaskWorker with role-based handlers (resync, discovery, analytics)
- Add /api/tasks endpoints for management and migration from legacy systems
- Add TasksDashboard UI and integrate task counts into main dashboard
- Add comprehensive documentation

Task roles: store_discovery, entry_point_discovery, product_discovery, product_resync, analytics_refresh

Run workers with: WORKER_ROLE=product_resync npx tsx src/tasks/task-worker.ts

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 16:27:03 -07:00
Kelly
7f9cf559cf fix(k8s): Update worker deployment to use v2 hydration worker
The old dutchie-az/services/worker.js no longer exists. Workers now use
the hydration pipeline at dist/scripts/run-hydration.js with --loop mode.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 15:01:18 -07:00
Kelly
bbe039c868 feat(api): Add job queue management endpoints and fix SQL type errors
- Add GET /api/job-queue/available - list dispensaries available for crawling
- Add GET /api/job-queue/history - get recent job history with results
- Add POST /api/job-queue/enqueue-batch - queue multiple dispensaries at once
- Add POST /api/job-queue/enqueue-state - queue all crawl-enabled dispensaries for a state
- Add POST /api/job-queue/clear-pending - clear pending jobs with optional filters
- Fix SQL parameter type errors by adding explicit casts ($2::text, $3::integer)
- Fix route ordering to prevent /:id from matching /available and /history

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 14:10:55 -07:00
Kelly
4e5c09a2a5 chore(dashboard): Remove DeployStatus block
Version info already shown in Layout sidebar header.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 14:10:22 -07:00
Kelly
7f65598332 feat(admin): Show version info at top of sidebar
- Add package.json version to /api/version endpoint
- Move version display from footer to top (next to logo)
- Show format: v1.5.1 (abc1234) - 12/9/2024

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 13:58:36 -07:00
Kelly
75315ed91e fix(ci): Use comma-separated build_args for docker-buildx plugin
The docker-buildx plugin expects build_args as a comma-separated string,
not a YAML list. This should fix the build_sha/build_time being null.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 13:56:37 -07:00
Kelly
7fe7d17b43 fix(consumer): Use relative API URLs for findadispo/findagram
The consumer frontends were hardcoded to use cannaiq.co as the API
URL, but each domain has its own /api path in the ingress that routes
to the shared backend. Using relative URLs allows each site to make
API calls to its own domain.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 13:38:10 -07:00
Kelly
7e517b5801 ci: Use self-hosted base images to avoid Docker Hub rate limits
Cached node:20, node:20-slim, and nginx:alpine to code.cannabrands.app.
No more Docker Hub dependency for builds.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 13:07:21 -07:00
Kelly
38ba9021d1 ci: Retry build (Docker Hub rate limit) 2025-12-09 12:58:36 -07:00
Kelly
ddebad48d3 ci: Remove auto-migrations from deploy step
Database was restored from backup - no migrations needed.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 12:52:04 -07:00
Kelly
1cebf2e296 fix(health): Add build_sha and build_time to health endpoint
Reads APP_GIT_SHA and APP_BUILD_TIME env vars set during Docker build.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 12:22:52 -07:00
Kelly
1d6e67d837 feat(api): Add store metrics endpoints with localhost bypass
New public API v1 endpoints for third-party integrations:
- GET /api/v1/stores/:id/metrics - Store performance metrics
- GET /api/v1/stores/:id/product-metrics - Product-level price changes
- GET /api/v1/stores/:id/competitor-snapshot - Competitive intelligence

Also adds localhost IP bypass for local development testing.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 12:14:13 -07:00
Kelly
cfb4b6e4ce fix(cannaiq): Fix TypeScript error in DeployStatus component
Properly destructure api.get response which returns { data: T }

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 12:08:29 -07:00
Kelly
f418c403d6 feat(auth): Add *.cannabrands.app to trusted origins whitelist
Adds pattern-based origin matching to support wildcard subdomains.
All *.cannabrands.app origins now bypass API key authentication.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 12:06:14 -07:00
Kelly
be4221af46 ci: Retrigger build 2025-12-09 11:54:16 -07:00
Kelly
ca07606b05 feat(k8s): Add Redis deployment for production
- Add k8s/redis.yaml with Redis 7 Alpine deployment
- Add REDIS_HOST and REDIS_PORT to configmap
- Redis configured with 200MB max memory and LRU eviction
- 1GB persistent volume for data persistence

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 11:40:11 -07:00
Kelly
baf1bf2eb7 fix(health): Require Redis in production, optional in local
Redis health check now returns error status when not configured in
production/staging environments, but remains optional in local dev.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 11:38:49 -07:00
Kelly
4ef3a8d72b fix(build): Fix TypeScript errors breaking CI build
- Add missing 'original' property to LocalImageSizes in brand logo download
- Remove test scripts with type errors (test-image-download.ts, test-stealth-with-db.ts)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 11:36:28 -07:00
Kelly
09dd756eff feat(admin): Add deploy status panel to dashboard
Shows running version vs latest git commit, pipeline status with steps,
and how many commits behind if not on latest. Uses Woodpecker and Gitea
APIs to fetch CI/CD information. Auto-refreshes every 30 seconds.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 11:26:41 -07:00
Kelly
ec8ef6210c ci: Run migrations inside K8s cluster after deploy
DB is internal to the cluster, so migrations must run via kubectl exec
into the scraper pod after deployment completes.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 11:16:21 -07:00
Kelly
a9b7a4d7a9 ci: Add proper SQL migration runner with tracking
- Creates run-migrations.ts that reads migrations/*.sql files
- Tracks applied migrations in schema_migrations table by filename
- Handles existing version-based schema by adding filename column
- CI now runs migrations before deploy

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 11:12:50 -07:00
Kelly
5119d5ccf9 ci: Add migration step before deploy
Migrations now run automatically after Docker builds but before K8s deploy.
Requires DATABASE_URL secret to be configured in Woodpecker.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 11:09:49 -07:00
Kelly
91efd1d03d feat(images): Add local image storage with on-demand resizing
- Store product images locally with hierarchy: /images/products/<state>/<store>/<brand>/<product>/
- Add /img/* proxy endpoint for on-demand resizing via Sharp
- Implement per-product image checking to skip existing downloads
- Fix pathToUrl() to correctly generate /images/... URLs
- Add frontend getImageUrl() helper with preset sizes (thumb, medium, large)
- Update all product pages to use optimized image URLs
- Add stealth session support for Dutchie GraphQL crawls
- Include test scripts for crawl and image verification

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 11:04:50 -07:00
Kelly
aa776226b0 fix(consumer): Wire findagram/findadispo to public API
- Update Dockerfiles to use cannaiq.co as API base URL
- Change findagram API client from /api/az to /api/v1 endpoints
- Add trusted origin bypass in public-api middleware for consumer sites
- Consumer sites (findagram.co, findadispo.com) can now access /api/v1
  endpoints without API key authentication

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 11:04:50 -07:00
kelly
e9435150e9 Merge pull request 'feature/wp-plugin-versioning-and-fixes' (#7) from feature/wp-plugin-versioning-and-fixes into master 2025-12-09 17:15:33 +00:00
Kelly
d399b966e6 ci: trigger build 2025-12-09 10:03:29 -07:00
Kelly
f5f0e25384 ci: trigger build 2025-12-09 10:03:06 -07:00
Kelly
04de33e5f7 fix(ci): Use correct container name 'worker' for scraper-worker deployment
Verified via kubectl: container name in cluster is 'worker', not 'scraper-worker'.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 09:43:39 -07:00
Kelly
37dfea25e1 feat: WordPress plugin versioning + heatmap fix + dynamic latest download
- Add VERSION file (1.5.4) for tracking WP plugin version
- Update plugin headers to 1.5.4 (cannaiq-menus.php, crawlsy-menus.php)
- Add dynamic /downloads/cannaiq-menus-latest.zip route that auto-redirects
  to highest version (no manual symlinks needed)
- Update frontend download links to use -latest.zip
- Fix StateHeatmap.tsx to parse API values as numbers (fixes string concat bug)
- Document versioning rules in CLAUDE.md

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 09:43:39 -07:00
Kelly
e2166bc25f fix(cannaiq): Parse heatmap values as numbers in frontend
Ensures values from the API are parsed as numbers before using them
in calculations. Fixes string concatenation bug in stats summary.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 09:43:39 -07:00
kelly
b5e8f039bf Merge pull request 'fix(backend): Parse bigint values in heatmap API response' (#6) from feature/seo-template-library-and-enhancements into master
Reviewed-on: https://code.cannabrands.app/Creationshop/dispensary-scraper/pulls/6
2025-12-09 16:26:19 +00:00
kelly
67bfdf47a5 Merge pull request 'fix: Add missing type field and pass build args to CI' (#5) from feature/seo-template-library-and-enhancements into master
Reviewed-on: https://code.cannabrands.app/Creationshop/dispensary-scraper/pulls/5
2025-12-09 15:41:57 +00:00
kelly
9f898f68db Merge pull request 'feat: SEO template library, discovery pipeline, and orchestrator enhancements' (#4) from feature/seo-template-library-and-enhancements into master
Reviewed-on: https://code.cannabrands.app/Creationshop/dispensary-scraper/pulls/4
2025-12-09 08:13:11 +00:00
70 changed files with 7325 additions and 326 deletions

View File

@@ -6,7 +6,7 @@ steps:
# PR VALIDATION: Parallel type checks (PRs only)
# ===========================================
typecheck-backend:
image: node:20
image: code.cannabrands.app/creationshop/node:20
commands:
- cd backend
- npm ci --prefer-offline
@@ -16,7 +16,7 @@ steps:
event: pull_request
typecheck-cannaiq:
image: node:20
image: code.cannabrands.app/creationshop/node:20
commands:
- cd cannaiq
- npm ci --prefer-offline
@@ -26,7 +26,7 @@ steps:
event: pull_request
typecheck-findadispo:
image: node:20
image: code.cannabrands.app/creationshop/node:20
commands:
- cd findadispo/frontend
- npm ci --prefer-offline
@@ -36,7 +36,7 @@ steps:
event: pull_request
typecheck-findagram:
image: node:20
image: code.cannabrands.app/creationshop/node:20
commands:
- cd findagram/frontend
- npm ci --prefer-offline
@@ -65,7 +65,7 @@ steps:
platforms: linux/amd64
provenance: false
build_args:
- APP_BUILD_VERSION=${CI_COMMIT_SHA:0:8}
- APP_BUILD_VERSION=${CI_COMMIT_SHA}
- APP_GIT_SHA=${CI_COMMIT_SHA}
- APP_BUILD_TIME=${CI_PIPELINE_CREATED}
- CONTAINER_IMAGE_TAG=${CI_COMMIT_SHA:0:8}
@@ -138,7 +138,7 @@ steps:
event: push
# ===========================================
# STAGE 3: Deploy (after all Docker builds)
# STAGE 3: Deploy (after Docker builds)
# ===========================================
deploy:
image: bitnami/kubectl:latest
@@ -150,7 +150,7 @@ steps:
- echo "$KUBECONFIG_CONTENT" | tr -d '[:space:]' | base64 -d > ~/.kube/config
- chmod 600 ~/.kube/config
- kubectl set image deployment/scraper scraper=code.cannabrands.app/creationshop/dispensary-scraper:${CI_COMMIT_SHA:0:8} -n dispensary-scraper
- kubectl set image deployment/scraper-worker scraper-worker=code.cannabrands.app/creationshop/dispensary-scraper:${CI_COMMIT_SHA:0:8} -n dispensary-scraper
- kubectl set image deployment/scraper-worker worker=code.cannabrands.app/creationshop/dispensary-scraper:${CI_COMMIT_SHA:0:8} -n dispensary-scraper
- kubectl set image deployment/cannaiq-frontend cannaiq-frontend=code.cannabrands.app/creationshop/cannaiq-frontend:${CI_COMMIT_SHA:0:8} -n dispensary-scraper
- kubectl set image deployment/findadispo-frontend findadispo-frontend=code.cannabrands.app/creationshop/findadispo-frontend:${CI_COMMIT_SHA:0:8} -n dispensary-scraper
- kubectl set image deployment/findagram-frontend findagram-frontend=code.cannabrands.app/creationshop/findagram-frontend:${CI_COMMIT_SHA:0:8} -n dispensary-scraper

View File

@@ -459,15 +459,66 @@ const result = await pool.query(`
### Local Storage Structure
```
/storage/products/{brand}/{state}/{product_id}/
/storage/images/products/{state}/{store}/{brand}/{product}/
image-{hash}.webp
image-{hash}-medium.webp
image-{hash}-thumb.webp
/storage/brands/{brand}/
/storage/images/brands/{brand}/
logo-{hash}.webp
```
### Image Proxy API (On-Demand Resizing)
Images are stored at full resolution and resized on-demand via the `/img` endpoint.
**Endpoint:** `GET /img/<path>?<params>`
**Parameters:**
| Param | Description | Example |
|-------|-------------|---------|
| `w` | Width in pixels (max 4000) | `?w=200` |
| `h` | Height in pixels (max 4000) | `?h=200` |
| `q` | Quality 1-100 (default 80) | `?q=70` |
| `fit` | Resize mode: cover, contain, fill, inside, outside | `?fit=cover` |
| `blur` | Blur sigma 0.3-1000 | `?blur=5` |
| `gray` | Grayscale (1 = enabled) | `?gray=1` |
| `format` | Output: webp, jpeg, png, avif (default webp) | `?format=jpeg` |
**Examples:**
```bash
# Thumbnail (50px)
GET /img/products/az/store/brand/product/image-abc123.webp?w=50
# Card image (200px, cover fit)
GET /img/products/az/store/brand/product/image-abc123.webp?w=200&h=200&fit=cover
# JPEG at 70% quality
GET /img/products/az/store/brand/product/image-abc123.webp?w=400&format=jpeg&q=70
# Grayscale blur
GET /img/products/az/store/brand/product/image-abc123.webp?w=200&gray=1&blur=3
```
**Frontend Usage:**
```typescript
import { getImageUrl, ImageSizes } from '../lib/images';
// Returns /img/products/.../image.webp?w=50 for local images
// Returns original URL for remote images (CDN, etc.)
const thumbUrl = getImageUrl(product.image_url, ImageSizes.thumb);
const cardUrl = getImageUrl(product.image_url, ImageSizes.medium);
const detailUrl = getImageUrl(product.image_url, ImageSizes.detail);
```
**Size Presets:**
| Preset | Width | Use Case |
|--------|-------|----------|
| `thumb` | 50px | Table thumbnails |
| `small` | 100px | Small cards |
| `medium` | 200px | Grid cards |
| `large` | 400px | Large cards |
| `detail` | 600px | Product detail |
| `full` | - | No resize |
### Storage Adapter
```typescript
@@ -480,8 +531,9 @@ import { saveImage, getImageUrl } from '../utils/storage-adapter';
| File | Purpose |
|------|---------|
| `backend/src/utils/local-storage.ts` | Local filesystem adapter |
| `backend/src/utils/storage-adapter.ts` | Unified storage abstraction |
| `backend/src/utils/image-storage.ts` | Image download and storage |
| `backend/src/routes/image-proxy.ts` | On-demand image resizing endpoint |
| `cannaiq/src/lib/images.ts` | Frontend image URL helper |
| `docker-compose.local.yml` | Local stack without MinIO |
| `start-local.sh` | Convenience startup script |
@@ -1195,3 +1247,32 @@ Every analytics v2 endpoint must:
---
# END Analytics V2 spec extension
---
## WordPress Plugin Versioning
The WordPress plugin version is tracked in `wordpress-plugin/VERSION`.
**Current version:** Check `wordpress-plugin/VERSION` for the latest version.
**Versioning rules:**
- **Minor bumps (x.x.N)**: Bug fixes, small improvements - default for most changes
- **Middle bumps (x.N.0)**: New features, significant improvements
- **Major bumps (N.0.0)**: Breaking changes, major rewrites - only when user explicitly requests
**When making WP plugin changes:**
1. Read `wordpress-plugin/VERSION` to get current version
2. Bump the version number (minor by default)
3. Update both files:
- `wordpress-plugin/VERSION`
- Plugin header `Version:` in `cannaiq-menus.php` and/or `crawlsy-menus.php`
- The `define('..._VERSION', '...')` constant in each plugin file
**Plugin files:**
| File | Brand | API URL |
|------|-------|---------|
| `cannaiq-menus.php` | CannaIQ | `https://cannaiq.co/api/v1` |
| `crawlsy-menus.php` | Crawlsy (legacy) | `https://cannaiq.co/api/v1` |
Both plugins use the same API endpoint. The Crawlsy version exists for backward compatibility with existing installations.

View File

@@ -1,6 +1,6 @@
# Build stage
# Image: code.cannabrands.app/creationshop/dispensary-scraper
FROM node:20-slim AS builder
FROM code.cannabrands.app/creationshop/node:20-slim AS builder
WORKDIR /app
@@ -11,7 +11,7 @@ COPY . .
RUN npm run build
# Production stage
FROM node:20-slim
FROM code.cannabrands.app/creationshop/node:20-slim
# Build arguments for version info
ARG APP_BUILD_VERSION=dev

View File

@@ -0,0 +1,308 @@
# Crawl Pipeline Documentation
## Overview
The crawl pipeline fetches product data from Dutchie dispensary menus and stores it in the canonical database. This document covers the complete flow from task scheduling to data storage.
---
## Pipeline Stages
```
┌─────────────────────┐
│ store_discovery │ Find new dispensaries
└─────────┬───────────┘
┌─────────────────────┐
│ entry_point_discovery│ Resolve slug → platform_dispensary_id
└─────────┬───────────┘
┌─────────────────────┐
│ product_discovery │ Initial product crawl
└─────────┬───────────┘
┌─────────────────────┐
│ product_resync │ Recurring crawl (every 4 hours)
└─────────────────────┘
```
---
## Stage Details
### 1. Store Discovery
**Purpose:** Find new dispensaries to crawl
**Handler:** `src/tasks/handlers/store-discovery.ts`
**Flow:**
1. Query Dutchie `ConsumerDispensaries` GraphQL for cities/states
2. Extract dispensary info (name, address, menu_url)
3. Insert into `dutchie_discovery_locations`
4. Queue `entry_point_discovery` for each new location
---
### 2. Entry Point Discovery
**Purpose:** Resolve menu URL slug to platform_dispensary_id (MongoDB ObjectId)
**Handler:** `src/tasks/handlers/entry-point-discovery.ts`
**Flow:**
1. Load dispensary from database
2. Extract slug from `menu_url`:
- `/embedded-menu/<slug>` or `/dispensary/<slug>`
3. Start stealth session (fingerprint + proxy)
4. Query `resolveDispensaryIdWithDetails(slug)` via GraphQL
5. Update dispensary with `platform_dispensary_id`
6. Queue `product_discovery` task
**Example:**
```
menu_url: https://dutchie.com/embedded-menu/deeply-rooted
slug: deeply-rooted
platform_dispensary_id: 6405ef617056e8014d79101b
```
---
### 3. Product Discovery
**Purpose:** Initial crawl of a new dispensary
**Handler:** `src/tasks/handlers/product-discovery.ts`
Same as product_resync but for first-time crawls.
---
### 4. Product Resync
**Purpose:** Recurring crawl to capture price/stock changes
**Handler:** `src/tasks/handlers/product-resync.ts`
**Flow:**
#### Step 1: Load Dispensary Info
```sql
SELECT id, name, platform_dispensary_id, menu_url, state
FROM dispensaries
WHERE id = $1 AND crawl_enabled = true
```
#### Step 2: Start Stealth Session
- Generate random browser fingerprint
- Set locale/timezone matching state
- Optional proxy rotation
#### Step 3: Fetch Products via GraphQL
**Endpoint:** `https://dutchie.com/api-3/graphql`
**Variables:**
```javascript
{
includeEnterpriseSpecials: false,
productsFilter: {
dispensaryId: "<platform_dispensary_id>",
pricingType: "rec",
Status: "All",
types: [],
useCache: false,
isDefaultSort: true,
sortBy: "popularSortIdx",
sortDirection: 1,
bypassOnlineThresholds: true,
isKioskMenu: false,
removeProductsBelowOptionThresholds: false
},
page: 0,
perPage: 100
}
```
**Key Notes:**
- `Status: "All"` returns all products (Active returns same count)
- `Status: null` returns 0 products (broken)
- `pricingType: "rec"` returns BOTH rec and med prices
- Paginate until `products.length < perPage` or `allProducts.length >= totalCount`
#### Step 4: Normalize Data
Transform raw Dutchie payload to canonical format via `DutchieNormalizer`.
#### Step 5: Upsert Products
Insert/update `store_products` table with normalized data.
#### Step 6: Create Snapshots
Insert point-in-time record to `store_product_snapshots`.
#### Step 7: Track Missing Products (OOS Detection)
```sql
-- Reset consecutive_misses for products IN the feed
UPDATE store_products
SET consecutive_misses = 0, last_seen_at = NOW()
WHERE dispensary_id = $1
AND provider = 'dutchie'
AND provider_product_id = ANY($2)
-- Increment for products NOT in feed
UPDATE store_products
SET consecutive_misses = consecutive_misses + 1
WHERE dispensary_id = $1
AND provider = 'dutchie'
AND provider_product_id NOT IN (...)
AND consecutive_misses < 3
-- Mark OOS at 3 consecutive misses
UPDATE store_products
SET stock_status = 'oos', is_in_stock = false
WHERE dispensary_id = $1
AND consecutive_misses >= 3
AND stock_status != 'oos'
```
#### Step 8: Download Images
For new products, download and store images locally.
#### Step 9: Update Dispensary
```sql
UPDATE dispensaries SET last_crawl_at = NOW() WHERE id = $1
```
---
## GraphQL Payload Structure
### Product Fields (from filteredProducts.products[])
| Field | Type | Description |
|-------|------|-------------|
| `_id` / `id` | string | MongoDB ObjectId (24 hex chars) |
| `Name` | string | Product display name |
| `brandName` | string | Brand name |
| `brand.name` | string | Brand name (nested) |
| `brand.description` | string | Brand description |
| `type` | string | Category (Flower, Edible, Concentrate, etc.) |
| `subcategory` | string | Subcategory |
| `strainType` | string | Hybrid, Indica, Sativa, N/A |
| `Status` | string | Always "Active" in feed |
| `Image` | string | Primary image URL |
| `images[]` | array | All product images |
### Pricing Fields
| Field | Type | Description |
|-------|------|-------------|
| `Prices[]` | number[] | Rec prices per option |
| `recPrices[]` | number[] | Rec prices |
| `medicalPrices[]` | number[] | Medical prices |
| `recSpecialPrices[]` | number[] | Rec sale prices |
| `medicalSpecialPrices[]` | number[] | Medical sale prices |
| `Options[]` | string[] | Size options ("1/8oz", "1g", etc.) |
| `rawOptions[]` | string[] | Raw weight options ("3.5g") |
### Inventory Fields (POSMetaData.children[])
| Field | Type | Description |
|-------|------|-------------|
| `quantity` | number | Total inventory count |
| `quantityAvailable` | number | Available for online orders |
| `kioskQuantityAvailable` | number | Available for kiosk orders |
| `option` | string | Which size option this is for |
### Potency Fields
| Field | Type | Description |
|-------|------|-------------|
| `THCContent.range[]` | number[] | THC percentage |
| `CBDContent.range[]` | number[] | CBD percentage |
| `cannabinoidsV2[]` | array | Detailed cannabinoid breakdown |
### Specials (specialData.bogoSpecials[])
| Field | Type | Description |
|-------|------|-------------|
| `specialName` | string | Deal name |
| `specialType` | string | "bogo", "sale", etc. |
| `itemsForAPrice.value` | string | Bundle price |
| `bogoRewards[].totalQuantity.quantity` | number | Required quantity |
---
## OOS Detection Logic
Products disappear from the Dutchie feed when they go out of stock. We track this via `consecutive_misses`:
| Scenario | Action |
|----------|--------|
| Product in feed | `consecutive_misses = 0` |
| Product missing 1st time | `consecutive_misses = 1` |
| Product missing 2nd time | `consecutive_misses = 2` |
| Product missing 3rd time | `consecutive_misses = 3`, mark `stock_status = 'oos'` |
| Product returns to feed | `consecutive_misses = 0`, update stock_status |
**Why 3 misses?**
- Protects against false positives from crawl failures
- Single bad crawl doesn't trigger mass OOS alerts
- Balances detection speed vs accuracy
---
## Database Tables
### store_products
Current state of each product:
- `provider_product_id` - Dutchie's MongoDB ObjectId
- `name_raw`, `brand_name_raw` - Raw values from feed
- `price_rec`, `price_med` - Current prices
- `is_in_stock`, `stock_status` - Availability
- `consecutive_misses` - OOS detection counter
- `last_seen_at` - Last time product was in feed
### store_product_snapshots
Point-in-time records for historical analysis:
- One row per product per crawl
- Captures price, stock, potency at that moment
- Used for price history, analytics
### dispensaries
Store metadata:
- `platform_dispensary_id` - MongoDB ObjectId for GraphQL
- `menu_url` - Source URL
- `last_crawl_at` - Last successful crawl
- `crawl_enabled` - Whether to crawl
---
## Scheduling
Crawls are scheduled via `worker_tasks` table:
| Role | Frequency | Description |
|------|-----------|-------------|
| `product_resync` | Every 4 hours | Regular product refresh |
| `entry_point_discovery` | On-demand | New store setup |
| `store_discovery` | Daily | Find new stores |
---
## Error Handling
- **GraphQL errors:** Logged, task marked failed, retried later
- **Normalization errors:** Logged as warnings, continue with valid products
- **Image download errors:** Non-fatal, logged, continue
- **Database errors:** Task fails, will be retried
---
## Files
| File | Purpose |
|------|---------|
| `src/tasks/handlers/product-resync.ts` | Main crawl handler |
| `src/tasks/handlers/entry-point-discovery.ts` | Slug → ID resolution |
| `src/platforms/dutchie/index.ts` | GraphQL client, session management |
| `src/hydration/normalizers/dutchie.ts` | Payload normalization |
| `src/hydration/canonical-upsert.ts` | Database upsert logic |
| `migrations/075_consecutive_misses.sql` | OOS tracking column |

View File

@@ -0,0 +1,400 @@
# Worker Task Architecture
This document describes the unified task-based worker system that replaces the legacy fragmented job systems.
## Overview
The task worker architecture provides a single, unified system for managing all background work in CannaiQ:
- **Store discovery** - Find new dispensaries on platforms
- **Entry point discovery** - Resolve platform IDs from menu URLs
- **Product discovery** - Initial product fetch for new stores
- **Product resync** - Regular price/stock updates for existing stores
- **Analytics refresh** - Refresh materialized views and analytics
## Architecture
### Database Tables
**`worker_tasks`** - Central task queue
```sql
CREATE TABLE worker_tasks (
id SERIAL PRIMARY KEY,
role task_role NOT NULL, -- What type of work
dispensary_id INTEGER, -- Which store (if applicable)
platform VARCHAR(50), -- Which platform (dutchie, etc.)
status task_status DEFAULT 'pending',
priority INTEGER DEFAULT 0, -- Higher = process first
scheduled_for TIMESTAMP, -- Don't process before this time
worker_id VARCHAR(100), -- Which worker claimed it
claimed_at TIMESTAMP,
started_at TIMESTAMP,
completed_at TIMESTAMP,
last_heartbeat_at TIMESTAMP, -- For stale detection
result JSONB, -- Output from handler
error_message TEXT,
retry_count INTEGER DEFAULT 0,
max_retries INTEGER DEFAULT 3,
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
```
**Key indexes:**
- `idx_worker_tasks_pending_priority` - For efficient task claiming
- `idx_worker_tasks_active_dispensary` - Prevents concurrent tasks per store (partial unique index)
### Task Roles
| Role | Purpose | Per-Store | Scheduled |
|------|---------|-----------|-----------|
| `store_discovery` | Find new stores on a platform | No | Daily |
| `entry_point_discovery` | Resolve platform IDs | Yes | On-demand |
| `product_discovery` | Initial product fetch | Yes | After entry_point |
| `product_resync` | Price/stock updates | Yes | Every 4 hours |
| `analytics_refresh` | Refresh MVs | No | Daily |
### Task Lifecycle
```
pending → claimed → running → completed
failed
```
1. **pending** - Task is waiting to be picked up
2. **claimed** - Worker has claimed it (atomic via SELECT FOR UPDATE SKIP LOCKED)
3. **running** - Worker is actively processing
4. **completed** - Task finished successfully
5. **failed** - Task encountered an error
6. **stale** - Task lost its worker (recovered automatically)
## Files
### Core Files
| File | Purpose |
|------|---------|
| `src/tasks/task-service.ts` | TaskService - CRUD, claiming, capacity metrics |
| `src/tasks/task-worker.ts` | TaskWorker - Main worker loop |
| `src/tasks/index.ts` | Module exports |
| `src/routes/tasks.ts` | API endpoints |
| `migrations/074_worker_task_queue.sql` | Database schema |
### Task Handlers
| File | Role |
|------|------|
| `src/tasks/handlers/store-discovery.ts` | `store_discovery` |
| `src/tasks/handlers/entry-point-discovery.ts` | `entry_point_discovery` |
| `src/tasks/handlers/product-discovery.ts` | `product_discovery` |
| `src/tasks/handlers/product-resync.ts` | `product_resync` |
| `src/tasks/handlers/analytics-refresh.ts` | `analytics_refresh` |
## Running Workers
### Environment Variables
| Variable | Default | Description |
|----------|---------|-------------|
| `WORKER_ROLE` | (required) | Which task role to process |
| `WORKER_ID` | auto-generated | Custom worker identifier |
| `POLL_INTERVAL_MS` | 5000 | How often to check for tasks |
| `HEARTBEAT_INTERVAL_MS` | 30000 | How often to update heartbeat |
### Starting a Worker
```bash
# Start a product resync worker
WORKER_ROLE=product_resync npx tsx src/tasks/task-worker.ts
# Start with custom ID
WORKER_ROLE=product_resync WORKER_ID=resync-1 npx tsx src/tasks/task-worker.ts
# Start multiple workers for different roles
WORKER_ROLE=store_discovery npx tsx src/tasks/task-worker.ts &
WORKER_ROLE=product_resync npx tsx src/tasks/task-worker.ts &
```
### Kubernetes Deployment
```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: task-worker-resync
spec:
replicas: 3
template:
spec:
containers:
- name: worker
image: code.cannabrands.app/creationshop/dispensary-scraper:latest
command: ["npx", "tsx", "src/tasks/task-worker.ts"]
env:
- name: WORKER_ROLE
value: "product_resync"
```
## API Endpoints
### Task Management
| Endpoint | Method | Description |
|----------|--------|-------------|
| `/api/tasks` | GET | List tasks with filters |
| `/api/tasks` | POST | Create a new task |
| `/api/tasks/:id` | GET | Get task by ID |
| `/api/tasks/counts` | GET | Get counts by status |
| `/api/tasks/capacity` | GET | Get capacity metrics |
| `/api/tasks/capacity/:role` | GET | Get role-specific capacity |
| `/api/tasks/recover-stale` | POST | Recover tasks from dead workers |
### Task Generation
| Endpoint | Method | Description |
|----------|--------|-------------|
| `/api/tasks/generate/resync` | POST | Generate daily resync tasks |
| `/api/tasks/generate/discovery` | POST | Create store discovery task |
### Migration (from legacy systems)
| Endpoint | Method | Description |
|----------|--------|-------------|
| `/api/tasks/migration/status` | GET | Compare old vs new systems |
| `/api/tasks/migration/disable-old-schedules` | POST | Disable job_schedules |
| `/api/tasks/migration/cancel-pending-crawl-jobs` | POST | Cancel old crawl jobs |
| `/api/tasks/migration/create-resync-tasks` | POST | Create tasks for all stores |
| `/api/tasks/migration/full-migrate` | POST | One-click migration |
### Role-Specific Endpoints
| Endpoint | Method | Description |
|----------|--------|-------------|
| `/api/tasks/role/:role/last-completion` | GET | Last completion time |
| `/api/tasks/role/:role/recent` | GET | Recent completions |
| `/api/tasks/store/:id/active` | GET | Check if store has active task |
## Capacity Planning
The `v_worker_capacity` view provides real-time metrics:
```sql
SELECT * FROM v_worker_capacity;
```
Returns:
- `pending_tasks` - Tasks waiting to be claimed
- `ready_tasks` - Tasks ready now (scheduled_for is null or past)
- `claimed_tasks` - Tasks claimed but not started
- `running_tasks` - Tasks actively processing
- `completed_last_hour` - Recent completions
- `failed_last_hour` - Recent failures
- `active_workers` - Workers with recent heartbeats
- `avg_duration_sec` - Average task duration
- `tasks_per_worker_hour` - Throughput estimate
- `estimated_hours_to_drain` - Time to clear queue
### Scaling Recommendations
```javascript
// API: GET /api/tasks/capacity/:role
{
"role": "product_resync",
"pending_tasks": 500,
"active_workers": 3,
"workers_needed": {
"for_1_hour": 10,
"for_4_hours": 3,
"for_8_hours": 2
}
}
```
## Task Chaining
Tasks can automatically create follow-up tasks:
```
store_discovery → entry_point_discovery → product_discovery
(store has platform_dispensary_id)
Daily resync tasks
```
The `chainNextTask()` method handles this automatically.
## Stale Task Recovery
Tasks are considered stale if `last_heartbeat_at` is older than the threshold (default 10 minutes).
```sql
SELECT recover_stale_tasks(10); -- 10 minute threshold
```
Or via API:
```bash
curl -X POST /api/tasks/recover-stale \
-H 'Content-Type: application/json' \
-d '{"threshold_minutes": 10}'
```
## Migration from Legacy Systems
### Legacy Systems Replaced
1. **job_schedules + job_run_logs** - Scheduled job definitions
2. **dispensary_crawl_jobs** - Per-dispensary crawl queue
3. **SyncOrchestrator + HydrationWorker** - Raw payload processing
### Migration Steps
**Option 1: One-Click Migration**
```bash
curl -X POST /api/tasks/migration/full-migrate
```
This will:
1. Disable all job_schedules
2. Cancel pending dispensary_crawl_jobs
3. Generate resync tasks for all stores
4. Create discovery and analytics tasks
**Option 2: Manual Migration**
```bash
# 1. Check current status
curl /api/tasks/migration/status
# 2. Disable old schedules
curl -X POST /api/tasks/migration/disable-old-schedules
# 3. Cancel pending crawl jobs
curl -X POST /api/tasks/migration/cancel-pending-crawl-jobs
# 4. Create resync tasks
curl -X POST /api/tasks/migration/create-resync-tasks \
-H 'Content-Type: application/json' \
-d '{"state_code": "AZ"}'
# 5. Generate daily resync schedule
curl -X POST /api/tasks/generate/resync \
-H 'Content-Type: application/json' \
-d '{"batches_per_day": 6}'
```
## Per-Store Locking
The system prevents concurrent tasks for the same store using a partial unique index:
```sql
CREATE UNIQUE INDEX idx_worker_tasks_active_dispensary
ON worker_tasks (dispensary_id)
WHERE dispensary_id IS NOT NULL
AND status IN ('claimed', 'running');
```
This ensures only one task can be active per store at any time.
## Task Priority
Tasks are claimed in priority order (higher first), then by creation time:
```sql
ORDER BY priority DESC, created_at ASC
```
Default priorities:
- `store_discovery`: 0
- `entry_point_discovery`: 10 (high - new stores)
- `product_discovery`: 10 (high - new stores)
- `product_resync`: 0
- `analytics_refresh`: 0
## Scheduled Tasks
Tasks can be scheduled for future execution:
```javascript
await taskService.createTask({
role: 'product_resync',
dispensary_id: 123,
scheduled_for: new Date('2025-01-10T06:00:00Z'),
});
```
The `generate_resync_tasks()` function creates staggered tasks throughout the day:
```sql
SELECT generate_resync_tasks(6, '2025-01-10'); -- 6 batches = every 4 hours
```
## Dashboard Integration
The admin dashboard shows task queue status in the main overview:
```
Task Queue Summary
------------------
Pending: 45
Running: 3
Completed: 1,234
Failed: 12
```
Full task management is available at `/admin/tasks`.
## Error Handling
Failed tasks include the error message in `error_message` and can be retried:
```sql
-- View failed tasks
SELECT id, role, dispensary_id, error_message, retry_count
FROM worker_tasks
WHERE status = 'failed'
ORDER BY completed_at DESC
LIMIT 20;
-- Retry failed tasks
UPDATE worker_tasks
SET status = 'pending', retry_count = retry_count + 1
WHERE status = 'failed' AND retry_count < max_retries;
```
## Monitoring
### Logs
Workers log to stdout:
```
[TaskWorker] Starting worker worker-product_resync-a1b2c3d4 for role: product_resync
[TaskWorker] Claimed task 123 (product_resync) for dispensary 456
[TaskWorker] Task 123 completed successfully
```
### Health Check
Check if workers are active:
```sql
SELECT worker_id, role, COUNT(*), MAX(last_heartbeat_at)
FROM worker_tasks
WHERE last_heartbeat_at > NOW() - INTERVAL '5 minutes'
GROUP BY worker_id, role;
```
### Metrics
```sql
-- Tasks by status
SELECT status, COUNT(*) FROM worker_tasks GROUP BY status;
-- Tasks by role
SELECT role, status, COUNT(*) FROM worker_tasks GROUP BY role, status;
-- Average duration by role
SELECT role, AVG(EXTRACT(EPOCH FROM (completed_at - started_at))) as avg_seconds
FROM worker_tasks
WHERE status = 'completed' AND completed_at > NOW() - INTERVAL '24 hours'
GROUP BY role;
```

View File

@@ -0,0 +1,12 @@
-- Add timezone column to proxies table for geo-consistent fingerprinting
-- This allows matching Accept-Language and other headers to proxy location
ALTER TABLE proxies
ADD COLUMN IF NOT EXISTS timezone VARCHAR(50);
-- Add timezone to failed_proxies as well
ALTER TABLE failed_proxies
ADD COLUMN IF NOT EXISTS timezone VARCHAR(50);
-- Comment explaining usage
COMMENT ON COLUMN proxies.timezone IS 'IANA timezone (e.g., America/Phoenix) for geo-consistent fingerprinting';

View File

@@ -0,0 +1,322 @@
-- Migration 074: Worker Task Queue System
-- Implements role-based task queue with per-store locking and capacity tracking
-- Task queue table
CREATE TABLE IF NOT EXISTS worker_tasks (
id SERIAL PRIMARY KEY,
-- Task identification
role VARCHAR(50) NOT NULL, -- store_discovery, entry_point_discovery, product_discovery, product_resync, analytics_refresh
dispensary_id INTEGER REFERENCES dispensaries(id) ON DELETE CASCADE,
platform VARCHAR(20), -- dutchie, jane, treez, etc.
-- Task state
status VARCHAR(20) NOT NULL DEFAULT 'pending',
priority INTEGER DEFAULT 0, -- Higher = more urgent
-- Scheduling
scheduled_for TIMESTAMPTZ, -- For batch scheduling (e.g., every 4 hours)
-- Ownership
worker_id VARCHAR(100), -- Pod name or worker ID
claimed_at TIMESTAMPTZ,
started_at TIMESTAMPTZ,
completed_at TIMESTAMPTZ,
last_heartbeat_at TIMESTAMPTZ,
-- Results
result JSONB, -- Task output data
error_message TEXT,
retry_count INTEGER DEFAULT 0,
max_retries INTEGER DEFAULT 3,
-- Metadata
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
-- Constraints
CONSTRAINT valid_status CHECK (status IN ('pending', 'claimed', 'running', 'completed', 'failed', 'stale'))
);
-- Indexes for efficient task claiming
CREATE INDEX IF NOT EXISTS idx_worker_tasks_pending
ON worker_tasks(role, priority DESC, created_at ASC)
WHERE status = 'pending';
CREATE INDEX IF NOT EXISTS idx_worker_tasks_claimed
ON worker_tasks(worker_id, claimed_at)
WHERE status = 'claimed';
CREATE INDEX IF NOT EXISTS idx_worker_tasks_running
ON worker_tasks(worker_id, last_heartbeat_at)
WHERE status = 'running';
CREATE INDEX IF NOT EXISTS idx_worker_tasks_dispensary
ON worker_tasks(dispensary_id)
WHERE dispensary_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_worker_tasks_scheduled
ON worker_tasks(scheduled_for)
WHERE status = 'pending' AND scheduled_for IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_worker_tasks_history
ON worker_tasks(role, completed_at DESC)
WHERE status IN ('completed', 'failed');
-- Partial unique index to prevent duplicate active tasks per store
-- Only one task can be claimed/running for a given dispensary at a time
CREATE UNIQUE INDEX IF NOT EXISTS idx_worker_tasks_unique_active_store
ON worker_tasks(dispensary_id)
WHERE status IN ('claimed', 'running') AND dispensary_id IS NOT NULL;
-- Worker registration table (tracks active workers)
CREATE TABLE IF NOT EXISTS worker_registry (
id SERIAL PRIMARY KEY,
worker_id VARCHAR(100) UNIQUE NOT NULL,
role VARCHAR(50) NOT NULL,
pod_name VARCHAR(100),
hostname VARCHAR(100),
started_at TIMESTAMPTZ DEFAULT NOW(),
last_heartbeat_at TIMESTAMPTZ DEFAULT NOW(),
tasks_completed INTEGER DEFAULT 0,
tasks_failed INTEGER DEFAULT 0,
status VARCHAR(20) DEFAULT 'active',
CONSTRAINT valid_worker_status CHECK (status IN ('active', 'idle', 'offline'))
);
CREATE INDEX IF NOT EXISTS idx_worker_registry_role
ON worker_registry(role, status);
CREATE INDEX IF NOT EXISTS idx_worker_registry_heartbeat
ON worker_registry(last_heartbeat_at)
WHERE status = 'active';
-- Task completion tracking (summarized history)
CREATE TABLE IF NOT EXISTS task_completion_log (
id SERIAL PRIMARY KEY,
role VARCHAR(50) NOT NULL,
date DATE NOT NULL DEFAULT CURRENT_DATE,
hour INTEGER NOT NULL DEFAULT EXTRACT(HOUR FROM NOW()),
tasks_created INTEGER DEFAULT 0,
tasks_completed INTEGER DEFAULT 0,
tasks_failed INTEGER DEFAULT 0,
avg_duration_sec NUMERIC(10,2),
min_duration_sec NUMERIC(10,2),
max_duration_sec NUMERIC(10,2),
updated_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(role, date, hour)
);
-- Capacity planning view
CREATE OR REPLACE VIEW v_worker_capacity AS
SELECT
role,
COUNT(*) FILTER (WHERE status = 'pending') as pending_tasks,
COUNT(*) FILTER (WHERE status = 'pending' AND (scheduled_for IS NULL OR scheduled_for <= NOW())) as ready_tasks,
COUNT(*) FILTER (WHERE status = 'claimed') as claimed_tasks,
COUNT(*) FILTER (WHERE status = 'running') as running_tasks,
COUNT(*) FILTER (WHERE status = 'completed' AND completed_at > NOW() - INTERVAL '1 hour') as completed_last_hour,
COUNT(*) FILTER (WHERE status = 'failed' AND completed_at > NOW() - INTERVAL '1 hour') as failed_last_hour,
COUNT(DISTINCT worker_id) FILTER (WHERE status IN ('claimed', 'running')) as active_workers,
AVG(EXTRACT(EPOCH FROM (completed_at - started_at)))
FILTER (WHERE status = 'completed' AND completed_at > NOW() - INTERVAL '1 hour') as avg_duration_sec,
-- Capacity planning metrics
CASE
WHEN COUNT(*) FILTER (WHERE status = 'completed' AND completed_at > NOW() - INTERVAL '1 hour') > 0
THEN 3600.0 / NULLIF(AVG(EXTRACT(EPOCH FROM (completed_at - started_at)))
FILTER (WHERE status = 'completed' AND completed_at > NOW() - INTERVAL '1 hour'), 0)
ELSE NULL
END as tasks_per_worker_hour,
-- Estimated time to drain queue
CASE
WHEN COUNT(DISTINCT worker_id) FILTER (WHERE status IN ('claimed', 'running')) > 0
AND COUNT(*) FILTER (WHERE status = 'completed' AND completed_at > NOW() - INTERVAL '1 hour') > 0
THEN COUNT(*) FILTER (WHERE status = 'pending') / NULLIF(
COUNT(DISTINCT worker_id) FILTER (WHERE status IN ('claimed', 'running')) *
(3600.0 / NULLIF(AVG(EXTRACT(EPOCH FROM (completed_at - started_at)))
FILTER (WHERE status = 'completed' AND completed_at > NOW() - INTERVAL '1 hour'), 0)),
0
)
ELSE NULL
END as estimated_hours_to_drain
FROM worker_tasks
GROUP BY role;
-- Task history view (for UI)
CREATE OR REPLACE VIEW v_task_history AS
SELECT
t.id,
t.role,
t.dispensary_id,
d.name as dispensary_name,
t.platform,
t.status,
t.priority,
t.worker_id,
t.scheduled_for,
t.claimed_at,
t.started_at,
t.completed_at,
t.error_message,
t.retry_count,
t.created_at,
EXTRACT(EPOCH FROM (t.completed_at - t.started_at)) as duration_sec
FROM worker_tasks t
LEFT JOIN dispensaries d ON d.id = t.dispensary_id
ORDER BY t.created_at DESC;
-- Function to claim a task atomically
CREATE OR REPLACE FUNCTION claim_task(
p_role VARCHAR(50),
p_worker_id VARCHAR(100)
) RETURNS worker_tasks AS $$
DECLARE
claimed_task worker_tasks;
BEGIN
UPDATE worker_tasks
SET
status = 'claimed',
worker_id = p_worker_id,
claimed_at = NOW(),
updated_at = NOW()
WHERE id = (
SELECT id FROM worker_tasks
WHERE role = p_role
AND status = 'pending'
AND (scheduled_for IS NULL OR scheduled_for <= NOW())
-- Exclude stores that already have an active task
AND (dispensary_id IS NULL OR dispensary_id NOT IN (
SELECT dispensary_id FROM worker_tasks
WHERE status IN ('claimed', 'running')
AND dispensary_id IS NOT NULL
))
ORDER BY priority DESC, created_at ASC
LIMIT 1
FOR UPDATE SKIP LOCKED
)
RETURNING * INTO claimed_task;
RETURN claimed_task;
END;
$$ LANGUAGE plpgsql;
-- Function to mark stale tasks (workers that died)
CREATE OR REPLACE FUNCTION recover_stale_tasks(
stale_threshold_minutes INTEGER DEFAULT 10
) RETURNS INTEGER AS $$
DECLARE
recovered_count INTEGER;
BEGIN
WITH stale AS (
UPDATE worker_tasks
SET
status = 'pending',
worker_id = NULL,
claimed_at = NULL,
started_at = NULL,
retry_count = retry_count + 1,
updated_at = NOW()
WHERE status IN ('claimed', 'running')
AND last_heartbeat_at < NOW() - (stale_threshold_minutes || ' minutes')::INTERVAL
AND retry_count < max_retries
RETURNING id
)
SELECT COUNT(*) INTO recovered_count FROM stale;
-- Mark tasks that exceeded retries as failed
UPDATE worker_tasks
SET
status = 'failed',
error_message = 'Exceeded max retries after worker failures',
completed_at = NOW(),
updated_at = NOW()
WHERE status IN ('claimed', 'running')
AND last_heartbeat_at < NOW() - (stale_threshold_minutes || ' minutes')::INTERVAL
AND retry_count >= max_retries;
RETURN recovered_count;
END;
$$ LANGUAGE plpgsql;
-- Function to generate daily resync tasks
CREATE OR REPLACE FUNCTION generate_resync_tasks(
p_batches_per_day INTEGER DEFAULT 6, -- Every 4 hours
p_date DATE DEFAULT CURRENT_DATE
) RETURNS INTEGER AS $$
DECLARE
store_count INTEGER;
stores_per_batch INTEGER;
batch_num INTEGER;
scheduled_time TIMESTAMPTZ;
created_count INTEGER := 0;
BEGIN
-- Count active stores that need resync
SELECT COUNT(*) INTO store_count
FROM dispensaries
WHERE crawl_enabled = true
AND menu_type = 'dutchie'
AND platform_dispensary_id IS NOT NULL;
IF store_count = 0 THEN
RETURN 0;
END IF;
stores_per_batch := CEIL(store_count::NUMERIC / p_batches_per_day);
FOR batch_num IN 0..(p_batches_per_day - 1) LOOP
scheduled_time := p_date + (batch_num * 4 || ' hours')::INTERVAL;
INSERT INTO worker_tasks (role, dispensary_id, platform, scheduled_for, priority)
SELECT
'product_resync',
d.id,
'dutchie',
scheduled_time,
0
FROM (
SELECT id, ROW_NUMBER() OVER (ORDER BY id) as rn
FROM dispensaries
WHERE crawl_enabled = true
AND menu_type = 'dutchie'
AND platform_dispensary_id IS NOT NULL
) d
WHERE d.rn > (batch_num * stores_per_batch)
AND d.rn <= ((batch_num + 1) * stores_per_batch)
ON CONFLICT DO NOTHING;
GET DIAGNOSTICS created_count = created_count + ROW_COUNT;
END LOOP;
RETURN created_count;
END;
$$ LANGUAGE plpgsql;
-- Trigger to update timestamp
CREATE OR REPLACE FUNCTION update_worker_tasks_timestamp()
RETURNS TRIGGER AS $$
BEGIN
NEW.updated_at = NOW();
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
DROP TRIGGER IF EXISTS worker_tasks_updated_at ON worker_tasks;
CREATE TRIGGER worker_tasks_updated_at
BEFORE UPDATE ON worker_tasks
FOR EACH ROW
EXECUTE FUNCTION update_worker_tasks_timestamp();
-- Comments
COMMENT ON TABLE worker_tasks IS 'Central task queue for all worker roles';
COMMENT ON TABLE worker_registry IS 'Registry of active workers and their stats';
COMMENT ON TABLE task_completion_log IS 'Hourly aggregated task completion metrics';
COMMENT ON VIEW v_worker_capacity IS 'Real-time capacity planning metrics per role';
COMMENT ON VIEW v_task_history IS 'Task history with dispensary details for UI';
COMMENT ON FUNCTION claim_task IS 'Atomically claim a task for a worker, respecting per-store locking';
COMMENT ON FUNCTION recover_stale_tasks IS 'Release tasks from dead workers back to pending';
COMMENT ON FUNCTION generate_resync_tasks IS 'Generate daily product resync tasks in batches';

View File

@@ -0,0 +1,13 @@
-- Migration 075: Add consecutive_misses column to store_products
-- Used to track how many consecutive crawls a product has been missing from the feed
-- After 3 consecutive misses, product is marked as OOS
ALTER TABLE store_products
ADD COLUMN IF NOT EXISTS consecutive_misses INTEGER NOT NULL DEFAULT 0;
-- Index for finding products that need OOS check
CREATE INDEX IF NOT EXISTS idx_store_products_consecutive_misses
ON store_products (dispensary_id, consecutive_misses)
WHERE consecutive_misses > 0;
COMMENT ON COLUMN store_products.consecutive_misses IS 'Number of consecutive crawls where product was not in feed. Reset to 0 when seen. At 3, mark OOS.';

Binary file not shown.

View File

@@ -29,6 +29,11 @@ const TRUSTED_ORIGINS = [
'http://localhost:5173',
];
// Pattern-based trusted origins (wildcards)
const TRUSTED_ORIGIN_PATTERNS = [
/^https:\/\/.*\.cannabrands\.app$/, // *.cannabrands.app
];
// Trusted IPs for internal pod-to-pod communication
const TRUSTED_IPS = [
'127.0.0.1',
@@ -42,8 +47,16 @@ const TRUSTED_IPS = [
function isTrustedRequest(req: Request): boolean {
// Check origin header
const origin = req.headers.origin;
if (origin && TRUSTED_ORIGINS.includes(origin)) {
return true;
if (origin) {
if (TRUSTED_ORIGINS.includes(origin)) {
return true;
}
// Check pattern-based origins (wildcards like *.cannabrands.app)
for (const pattern of TRUSTED_ORIGIN_PATTERNS) {
if (pattern.test(origin)) {
return true;
}
}
}
// Check referer header (for same-origin requests without CORS)
@@ -54,6 +67,18 @@ function isTrustedRequest(req: Request): boolean {
return true;
}
}
// Check pattern-based referers
try {
const refererUrl = new URL(referer);
const refererOrigin = refererUrl.origin;
for (const pattern of TRUSTED_ORIGIN_PATTERNS) {
if (pattern.test(refererOrigin)) {
return true;
}
}
} catch {
// Invalid referer URL, skip
}
}
// Check IP for internal requests (pod-to-pod, localhost)

View File

@@ -0,0 +1,200 @@
#!/usr/bin/env npx tsx
/**
* Database Migration Runner
*
* Runs SQL migrations from backend/migrations/*.sql in order.
* Tracks applied migrations in schema_migrations table.
*
* Usage:
* npx tsx src/db/run-migrations.ts
*
* Environment:
* DATABASE_URL or CANNAIQ_DB_* variables
*/
import { Pool } from 'pg';
import * as fs from 'fs/promises';
import * as path from 'path';
import dotenv from 'dotenv';
dotenv.config();
function getConnectionString(): string {
if (process.env.DATABASE_URL) {
return process.env.DATABASE_URL;
}
if (process.env.CANNAIQ_DB_URL) {
return process.env.CANNAIQ_DB_URL;
}
const host = process.env.CANNAIQ_DB_HOST || 'localhost';
const port = process.env.CANNAIQ_DB_PORT || '54320';
const name = process.env.CANNAIQ_DB_NAME || 'dutchie_menus';
const user = process.env.CANNAIQ_DB_USER || 'dutchie';
const pass = process.env.CANNAIQ_DB_PASS || 'dutchie_local_pass';
return `postgresql://${user}:${pass}@${host}:${port}/${name}`;
}
interface MigrationFile {
filename: string;
number: number;
path: string;
}
async function getMigrationFiles(migrationsDir: string): Promise<MigrationFile[]> {
const files = await fs.readdir(migrationsDir);
const migrations: MigrationFile[] = files
.filter(f => f.endsWith('.sql'))
.map(filename => {
// Extract number from filename like "005_api_tokens.sql" or "073_proxy_timezone.sql"
const match = filename.match(/^(\d+)_/);
if (!match) return null;
return {
filename,
number: parseInt(match[1], 10),
path: path.join(migrationsDir, filename),
};
})
.filter((m): m is MigrationFile => m !== null)
.sort((a, b) => a.number - b.number);
return migrations;
}
async function ensureMigrationsTable(pool: Pool): Promise<void> {
// Migrate to filename-based tracking (handles duplicate version numbers)
// Check if old version-based PK exists
const pkCheck = await pool.query(`
SELECT constraint_name FROM information_schema.table_constraints
WHERE table_name = 'schema_migrations' AND constraint_type = 'PRIMARY KEY'
`);
if (pkCheck.rows.length === 0) {
// Table doesn't exist, create with filename as PK
await pool.query(`
CREATE TABLE IF NOT EXISTS schema_migrations (
filename VARCHAR(255) NOT NULL PRIMARY KEY,
version VARCHAR(10),
name VARCHAR(255),
applied_at TIMESTAMPTZ DEFAULT NOW()
)
`);
} else {
// Table exists - add filename column if missing
await pool.query(`
ALTER TABLE schema_migrations ADD COLUMN IF NOT EXISTS filename VARCHAR(255)
`);
// Populate filename from version+name for existing rows
await pool.query(`
UPDATE schema_migrations SET filename = version || '_' || name || '.sql'
WHERE filename IS NULL
`);
}
}
async function getAppliedMigrations(pool: Pool): Promise<Set<string>> {
// Try filename first, fall back to version_name combo
const result = await pool.query(`
SELECT COALESCE(filename, version || '_' || name || '.sql') as filename
FROM schema_migrations
`);
return new Set(result.rows.map(r => r.filename));
}
async function applyMigration(pool: Pool, migration: MigrationFile): Promise<void> {
const sql = await fs.readFile(migration.path, 'utf-8');
// Extract version and name from filename like "005_api_tokens.sql"
const version = String(migration.number).padStart(3, '0');
const name = migration.filename.replace(/^\d+_/, '').replace(/\.sql$/, '');
const client = await pool.connect();
try {
await client.query('BEGIN');
// Run the migration SQL
await client.query(sql);
// Record that it was applied - use INSERT with ON CONFLICT for safety
await client.query(`
INSERT INTO schema_migrations (filename, version, name)
VALUES ($1, $2, $3)
ON CONFLICT DO NOTHING
`, [migration.filename, version, name]);
await client.query('COMMIT');
} catch (error) {
await client.query('ROLLBACK');
throw error;
} finally {
client.release();
}
}
async function main() {
const pool = new Pool({ connectionString: getConnectionString() });
// Migrations directory relative to this file
const migrationsDir = path.resolve(__dirname, '../../migrations');
console.log('╔════════════════════════════════════════════════════════════╗');
console.log('║ DATABASE MIGRATION RUNNER ║');
console.log('╚════════════════════════════════════════════════════════════╝');
console.log(`Migrations dir: ${migrationsDir}`);
console.log('');
try {
// Ensure tracking table exists
await ensureMigrationsTable(pool);
// Get all migration files
const allMigrations = await getMigrationFiles(migrationsDir);
console.log(`Found ${allMigrations.length} migration files`);
// Get already-applied migrations
const applied = await getAppliedMigrations(pool);
console.log(`Already applied: ${applied.size} migrations`);
console.log('');
// Find pending migrations (compare by filename)
const pending = allMigrations.filter(m => !applied.has(m.filename));
if (pending.length === 0) {
console.log('✅ No pending migrations. Database is up to date.');
await pool.end();
return;
}
console.log(`Pending migrations: ${pending.length}`);
console.log('─'.repeat(60));
// Apply each pending migration
for (const migration of pending) {
process.stdout.write(` ${migration.filename}... `);
try {
await applyMigration(pool, migration);
console.log('✅');
} catch (error: any) {
console.log('❌');
console.error(`\nError applying ${migration.filename}:`);
console.error(error.message);
process.exit(1);
}
}
console.log('');
console.log('═'.repeat(60));
console.log(`✅ Applied ${pending.length} migrations successfully`);
} catch (error: any) {
console.error('Migration runner failed:', error.message);
process.exit(1);
} finally {
await pool.end();
}
}
main();

View File

@@ -16,6 +16,12 @@ import {
NormalizedBrand,
NormalizationResult,
} from './types';
import {
downloadProductImage,
ProductImageContext,
isImageStorageReady,
LocalImageSizes,
} from '../utils/image-storage';
const BATCH_SIZE = 100;
@@ -23,10 +29,21 @@ const BATCH_SIZE = 100;
// PRODUCT UPSERTS
// ============================================================
export interface NewProductInfo {
id: number; // store_products.id
externalProductId: string; // provider_product_id
name: string;
brandName: string | null;
primaryImageUrl: string | null;
hasLocalImage?: boolean; // True if local_image_path is already set
}
export interface UpsertProductsResult {
upserted: number;
new: number;
updated: number;
newProducts: NewProductInfo[]; // Details of newly created products
productsNeedingImages: NewProductInfo[]; // Products (new or updated) that need image downloads
}
/**
@@ -41,12 +58,14 @@ export async function upsertStoreProducts(
options: { dryRun?: boolean } = {}
): Promise<UpsertProductsResult> {
if (products.length === 0) {
return { upserted: 0, new: 0, updated: 0 };
return { upserted: 0, new: 0, updated: 0, newProducts: [], productsNeedingImages: [] };
}
const { dryRun = false } = options;
let newCount = 0;
let updatedCount = 0;
const newProducts: NewProductInfo[] = [];
const productsNeedingImages: NewProductInfo[] = [];
// Process in batches
for (let i = 0; i < products.length; i += BATCH_SIZE) {
@@ -104,7 +123,7 @@ export async function upsertStoreProducts(
image_url = EXCLUDED.image_url,
last_seen_at = NOW(),
updated_at = NOW()
RETURNING (xmax = 0) as is_new`,
RETURNING id, (xmax = 0) as is_new, (local_image_path IS NOT NULL) as has_local_image`,
[
product.dispensaryId,
product.platform,
@@ -129,10 +148,30 @@ export async function upsertStoreProducts(
]
);
if (result.rows[0]?.is_new) {
const row = result.rows[0];
const productInfo: NewProductInfo = {
id: row.id,
externalProductId: product.externalProductId,
name: product.name,
brandName: product.brandName,
primaryImageUrl: product.primaryImageUrl,
hasLocalImage: row.has_local_image,
};
if (row.is_new) {
newCount++;
// Track new products
newProducts.push(productInfo);
// New products always need images (if they have a source URL)
if (product.primaryImageUrl && !row.has_local_image) {
productsNeedingImages.push(productInfo);
}
} else {
updatedCount++;
// Updated products need images only if they don't have a local image yet
if (product.primaryImageUrl && !row.has_local_image) {
productsNeedingImages.push(productInfo);
}
}
}
@@ -149,6 +188,8 @@ export async function upsertStoreProducts(
upserted: newCount + updatedCount,
new: newCount,
updated: updatedCount,
newProducts,
productsNeedingImages,
};
}
@@ -564,6 +605,19 @@ export async function upsertBrands(
// FULL HYDRATION
// ============================================================
export interface ImageDownloadResult {
downloaded: number;
skipped: number;
failed: number;
bytesTotal: number;
}
export interface DispensaryContext {
stateCode: string;
storeSlug: string;
hasExistingProducts?: boolean; // True if store already has products with local images
}
export interface HydratePayloadResult {
productsUpserted: number;
productsNew: number;
@@ -574,6 +628,154 @@ export interface HydratePayloadResult {
variantsUpserted: number;
variantsNew: number;
variantSnapshotsCreated: number;
imagesDownloaded: number;
imagesSkipped: number;
imagesFailed: number;
imagesBytesTotal: number;
}
/**
* Helper to create slug from string
*/
function slugify(str: string): string {
return str
.toLowerCase()
.replace(/[^a-z0-9]+/g, '-')
.replace(/^-+|-+$/g, '')
.substring(0, 50) || 'unknown';
}
/**
* Download images for new products and update their local paths
*/
export async function downloadProductImages(
pool: Pool,
newProducts: NewProductInfo[],
dispensaryContext: DispensaryContext,
options: { dryRun?: boolean; concurrency?: number } = {}
): Promise<ImageDownloadResult> {
const { dryRun = false, concurrency = 5 } = options;
// Filter products that have images to download
const productsWithImages = newProducts.filter(p => p.primaryImageUrl);
if (productsWithImages.length === 0) {
return { downloaded: 0, skipped: 0, failed: 0, bytesTotal: 0 };
}
// Check if image storage is ready
if (!isImageStorageReady()) {
console.warn('[ImageDownload] Image storage not initialized, skipping downloads');
return { downloaded: 0, skipped: productsWithImages.length, failed: 0, bytesTotal: 0 };
}
if (dryRun) {
console.log(`[DryRun] Would download ${productsWithImages.length} images`);
return { downloaded: 0, skipped: productsWithImages.length, failed: 0, bytesTotal: 0 };
}
let downloaded = 0;
let skipped = 0;
let failed = 0;
let bytesTotal = 0;
// Process in batches with concurrency limit
for (let i = 0; i < productsWithImages.length; i += concurrency) {
const batch = productsWithImages.slice(i, i + concurrency);
const results = await Promise.allSettled(
batch.map(async (product) => {
const ctx: ProductImageContext = {
stateCode: dispensaryContext.stateCode,
storeSlug: dispensaryContext.storeSlug,
brandSlug: slugify(product.brandName || 'unknown'),
productId: product.externalProductId,
};
const result = await downloadProductImage(product.primaryImageUrl!, ctx, { skipIfExists: true });
if (result.success) {
// Update the database with local image path
const imagesJson = JSON.stringify({
full: result.urls!.full,
medium: result.urls!.medium,
thumb: result.urls!.thumb,
});
await pool.query(
`UPDATE store_products
SET local_image_path = $1, images = $2
WHERE id = $3`,
[result.urls!.full, imagesJson, product.id]
);
}
return result;
})
);
for (const result of results) {
if (result.status === 'fulfilled') {
const downloadResult = result.value;
if (downloadResult.success) {
if (downloadResult.skipped) {
skipped++;
} else {
downloaded++;
bytesTotal += downloadResult.bytesDownloaded || 0;
}
} else {
failed++;
console.warn(`[ImageDownload] Failed: ${downloadResult.error}`);
}
} else {
failed++;
console.error(`[ImageDownload] Error:`, result.reason);
}
}
}
console.log(`[ImageDownload] Downloaded: ${downloaded}, Skipped: ${skipped}, Failed: ${failed}, Bytes: ${bytesTotal}`);
return { downloaded, skipped, failed, bytesTotal };
}
/**
* Get dispensary context for image paths
* Also checks if this dispensary already has products with local images
* to skip unnecessary filesystem checks for existing stores
*/
async function getDispensaryContext(pool: Pool, dispensaryId: number): Promise<DispensaryContext | null> {
try {
const result = await pool.query(
`SELECT
d.state,
d.slug,
d.name,
EXISTS(
SELECT 1 FROM store_products sp
WHERE sp.dispensary_id = d.id
AND sp.local_image_path IS NOT NULL
LIMIT 1
) as has_local_images
FROM dispensaries d
WHERE d.id = $1`,
[dispensaryId]
);
if (result.rows.length === 0) {
return null;
}
const row = result.rows[0];
return {
stateCode: row.state || 'unknown',
storeSlug: row.slug || slugify(row.name || `store-${dispensaryId}`),
hasExistingProducts: row.has_local_images,
};
} catch (error) {
console.error('[getDispensaryContext] Error:', error);
return null;
}
}
/**
@@ -584,9 +786,9 @@ export async function hydrateToCanonical(
dispensaryId: number,
normResult: NormalizationResult,
crawlRunId: number | null,
options: { dryRun?: boolean } = {}
options: { dryRun?: boolean; downloadImages?: boolean } = {}
): Promise<HydratePayloadResult> {
const { dryRun = false } = options;
const { dryRun = false, downloadImages: shouldDownloadImages = true } = options;
// 1. Upsert brands
const brandResult = await upsertBrands(pool, normResult.brands, { dryRun });
@@ -634,6 +836,36 @@ export async function hydrateToCanonical(
{ dryRun }
);
// 6. Download images for products that need them
// This includes:
// - New products (always need images)
// - Updated products that don't have local images yet (backfill)
// This avoids:
// - Filesystem checks for products that already have local images
// - Unnecessary HTTP requests for products with existing images
let imageResult: ImageDownloadResult = { downloaded: 0, skipped: 0, failed: 0, bytesTotal: 0 };
if (shouldDownloadImages && productResult.productsNeedingImages.length > 0) {
const dispensaryContext = await getDispensaryContext(pool, dispensaryId);
if (dispensaryContext) {
const newCount = productResult.productsNeedingImages.filter(p => !p.hasLocalImage).length;
const backfillCount = productResult.productsNeedingImages.length - newCount;
console.log(`[Hydration] Downloading images for ${productResult.productsNeedingImages.length} products (${productResult.new} new, ${backfillCount} backfill)...`);
imageResult = await downloadProductImages(
pool,
productResult.productsNeedingImages,
dispensaryContext,
{ dryRun }
);
} else {
console.warn(`[Hydration] Could not get dispensary context for ID ${dispensaryId}, skipping image downloads`);
}
} else if (productResult.productsNeedingImages.length === 0 && productResult.upserted > 0) {
// All products already have local images
console.log(`[Hydration] All ${productResult.upserted} products already have local images, skipping downloads`);
}
return {
productsUpserted: productResult.upserted,
productsNew: productResult.new,
@@ -644,5 +876,9 @@ export async function hydrateToCanonical(
variantsUpserted: variantResult.upserted,
variantsNew: variantResult.new,
variantSnapshotsCreated: variantResult.snapshotsCreated,
imagesDownloaded: imageResult.downloaded,
imagesSkipped: imageResult.skipped,
imagesFailed: imageResult.failed,
imagesBytesTotal: imageResult.bytesTotal,
};
}

View File

@@ -7,6 +7,7 @@ import { initializeImageStorage } from './utils/image-storage';
import { logger } from './services/logger';
import { cleanupOrphanedJobs } from './services/proxyTestQueue';
import healthRoutes from './routes/health';
import imageProxyRoutes from './routes/image-proxy';
dotenv.config();
@@ -29,9 +30,44 @@ app.use(express.json());
const LOCAL_IMAGES_PATH = process.env.LOCAL_IMAGES_PATH || './public/images';
app.use('/images', express.static(LOCAL_IMAGES_PATH));
// Image proxy with on-demand resizing
// Usage: /img/products/az/store/brand/product/image.webp?w=200&h=200
app.use('/img', imageProxyRoutes);
// Serve static downloads (plugin files, etc.)
// Uses ./public/downloads relative to working directory (works for both Docker and local dev)
const LOCAL_DOWNLOADS_PATH = process.env.LOCAL_DOWNLOADS_PATH || './public/downloads';
// Dynamic "latest" redirect for WordPress plugin - finds highest version automatically
app.get('/downloads/cannaiq-menus-latest.zip', (req, res) => {
const fs = require('fs');
const path = require('path');
try {
const files = fs.readdirSync(LOCAL_DOWNLOADS_PATH);
const pluginFiles = files
.filter((f: string) => f.match(/^cannaiq-menus-\d+\.\d+\.\d+\.zip$/))
.sort((a: string, b: string) => {
const vA = a.match(/(\d+)\.(\d+)\.(\d+)/);
const vB = b.match(/(\d+)\.(\d+)\.(\d+)/);
if (!vA || !vB) return 0;
for (let i = 1; i <= 3; i++) {
const diff = parseInt(vB[i]) - parseInt(vA[i]);
if (diff !== 0) return diff;
}
return 0;
});
if (pluginFiles.length > 0) {
const latestFile = pluginFiles[0];
res.redirect(302, `/downloads/${latestFile}`);
} else {
res.status(404).json({ error: 'No plugin versions found' });
}
} catch (err) {
res.status(500).json({ error: 'Failed to find latest plugin' });
}
});
app.use('/downloads', express.static(LOCAL_DOWNLOADS_PATH));
// Simple health check for load balancers/K8s probes
@@ -71,6 +107,7 @@ import apiPermissionsRoutes from './routes/api-permissions';
import parallelScrapeRoutes from './routes/parallel-scrape';
import crawlerSandboxRoutes from './routes/crawler-sandbox';
import versionRoutes from './routes/version';
import deployStatusRoutes from './routes/deploy-status';
import publicApiRoutes from './routes/public-api';
import usersRoutes from './routes/users';
import staleProcessesRoutes from './routes/stale-processes';
@@ -102,6 +139,7 @@ import eventsRoutes from './routes/events';
import clickAnalyticsRoutes from './routes/click-analytics';
import seoRoutes from './routes/seo';
import priceAnalyticsRoutes from './routes/price-analytics';
import tasksRoutes from './routes/tasks';
// Mark requests from trusted domains (cannaiq.co, findagram.co, findadispo.com)
// These domains can access the API without authentication
@@ -144,6 +182,8 @@ app.use('/api/api-permissions', apiPermissionsRoutes);
app.use('/api/parallel-scrape', parallelScrapeRoutes);
app.use('/api/crawler-sandbox', crawlerSandboxRoutes);
app.use('/api/version', versionRoutes);
app.use('/api/admin/deploy-status', deployStatusRoutes);
console.log('[DeployStatus] Routes registered at /api/admin/deploy-status');
app.use('/api/users', usersRoutes);
app.use('/api/stale-processes', staleProcessesRoutes);
// Admin routes - orchestrator actions
@@ -172,6 +212,10 @@ app.use('/api/monitor', workersRoutes);
app.use('/api/job-queue', jobQueueRoutes);
console.log('[Workers] Routes registered at /api/workers, /api/monitor, and /api/job-queue');
// Task queue management - worker tasks with capacity planning
app.use('/api/tasks', tasksRoutes);
console.log('[Tasks] Routes registered at /api/tasks');
// Phase 3: Analytics V2 - Enhanced analytics with rec/med state segmentation
try {
const analyticsV2Router = createAnalyticsV2Router(getPool());

View File

@@ -213,7 +213,24 @@ const FINGERPRINTS: Fingerprint[] = [
let currentFingerprintIndex = 0;
// Forward declaration for session (actual CrawlSession interface defined later)
let currentSession: {
sessionId: string;
fingerprint: Fingerprint;
proxyUrl: string | null;
stateCode?: string;
timezone?: string;
startedAt: Date;
} | null = null;
/**
* Get current fingerprint - returns session fingerprint if active, otherwise default
*/
export function getFingerprint(): Fingerprint {
// Use session fingerprint if a session is active
if (currentSession) {
return currentSession.fingerprint;
}
return FINGERPRINTS[currentFingerprintIndex];
}
@@ -228,6 +245,103 @@ export function resetFingerprint(): void {
currentFingerprintIndex = 0;
}
/**
* Get a random fingerprint from the pool
*/
export function getRandomFingerprint(): Fingerprint {
const index = Math.floor(Math.random() * FINGERPRINTS.length);
return FINGERPRINTS[index];
}
// ============================================================
// SESSION MANAGEMENT
// Per-session fingerprint rotation for stealth
// ============================================================
export interface CrawlSession {
sessionId: string;
fingerprint: Fingerprint;
proxyUrl: string | null;
stateCode?: string;
timezone?: string;
startedAt: Date;
}
// Note: currentSession variable declared earlier in file for proper scoping
/**
* Timezone to Accept-Language mapping
* US timezones all use en-US but this can be extended for international
*/
const TIMEZONE_TO_LOCALE: Record<string, string> = {
'America/Phoenix': 'en-US,en;q=0.9',
'America/Los_Angeles': 'en-US,en;q=0.9',
'America/Denver': 'en-US,en;q=0.9',
'America/Chicago': 'en-US,en;q=0.9',
'America/New_York': 'en-US,en;q=0.9',
'America/Detroit': 'en-US,en;q=0.9',
'America/Anchorage': 'en-US,en;q=0.9',
'Pacific/Honolulu': 'en-US,en;q=0.9',
};
/**
* Get Accept-Language header for a given timezone
*/
export function getLocaleForTimezone(timezone?: string): string {
if (!timezone) return 'en-US,en;q=0.9';
return TIMEZONE_TO_LOCALE[timezone] || 'en-US,en;q=0.9';
}
/**
* Start a new crawl session with a random fingerprint
* Call this before crawling a store to get a fresh identity
*/
export function startSession(stateCode?: string, timezone?: string): CrawlSession {
const baseFp = getRandomFingerprint();
// Override Accept-Language based on timezone for geographic consistency
const fingerprint: Fingerprint = {
...baseFp,
acceptLanguage: getLocaleForTimezone(timezone),
};
currentSession = {
sessionId: `session_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`,
fingerprint,
proxyUrl: currentProxy,
stateCode,
timezone,
startedAt: new Date(),
};
console.log(`[Dutchie Client] Started session ${currentSession.sessionId}`);
console.log(`[Dutchie Client] Fingerprint: ${fingerprint.userAgent.slice(0, 50)}...`);
console.log(`[Dutchie Client] Accept-Language: ${fingerprint.acceptLanguage}`);
if (timezone) {
console.log(`[Dutchie Client] Timezone: ${timezone}`);
}
return currentSession;
}
/**
* End the current crawl session
*/
export function endSession(): void {
if (currentSession) {
const duration = Math.round((Date.now() - currentSession.startedAt.getTime()) / 1000);
console.log(`[Dutchie Client] Ended session ${currentSession.sessionId} (${duration}s)`);
currentSession = null;
}
}
/**
* Get current active session
*/
export function getCurrentSession(): CrawlSession | null {
return currentSession;
}
// ============================================================
// CURL HTTP CLIENT
// ============================================================

View File

@@ -18,6 +18,13 @@ export {
getFingerprint,
rotateFingerprint,
resetFingerprint,
getRandomFingerprint,
getLocaleForTimezone,
// Session Management (per-store fingerprint rotation)
startSession,
endSession,
getCurrentSession,
// Proxy
setProxy,
@@ -32,6 +39,7 @@ export {
// Types
type CurlResponse,
type Fingerprint,
type CrawlSession,
type ExecuteGraphQLOptions,
type FetchPageOptions,
} from './client';

View File

@@ -0,0 +1,269 @@
import { Router, Request, Response } from 'express';
import axios from 'axios';
const router = Router();
// Woodpecker API config - uses env vars or falls back
const WOODPECKER_SERVER = process.env.WOODPECKER_SERVER || 'https://ci.cannabrands.app';
const WOODPECKER_TOKEN = process.env.WOODPECKER_TOKEN;
const GITEA_SERVER = process.env.GITEA_SERVER || 'https://code.cannabrands.app';
const GITEA_TOKEN = process.env.GITEA_TOKEN;
const REPO_OWNER = 'Creationshop';
const REPO_NAME = 'dispensary-scraper';
interface PipelineStep {
name: string;
state: 'pending' | 'running' | 'success' | 'failure' | 'skipped';
started?: number;
stopped?: number;
}
interface PipelineInfo {
number: number;
status: string;
event: string;
branch: string;
message: string;
commit: string;
author: string;
created: number;
started?: number;
finished?: number;
steps?: PipelineStep[];
}
interface DeployStatusResponse {
running: {
sha: string;
sha_full: string;
build_time: string;
image_tag: string;
};
latest: {
sha: string;
sha_full: string;
message: string;
author: string;
timestamp: string;
} | null;
is_latest: boolean;
commits_behind: number;
pipeline: PipelineInfo | null;
error?: string;
}
/**
* Fetch latest commit from Gitea
*/
async function getLatestCommit(): Promise<{
sha: string;
message: string;
author: string;
timestamp: string;
} | null> {
if (!GITEA_TOKEN) {
console.warn('[DeployStatus] GITEA_TOKEN not set, skipping latest commit fetch');
return null;
}
try {
const response = await axios.get(
`${GITEA_SERVER}/api/v1/repos/${REPO_OWNER}/${REPO_NAME}/commits?limit=1`,
{
headers: { Authorization: `token ${GITEA_TOKEN}` },
timeout: 5000,
}
);
if (response.data && response.data.length > 0) {
const commit = response.data[0];
return {
sha: commit.sha,
message: commit.commit?.message?.split('\n')[0] || '',
author: commit.commit?.author?.name || commit.author?.login || 'unknown',
timestamp: commit.commit?.author?.date || commit.created,
};
}
} catch (error: any) {
console.error('[DeployStatus] Failed to fetch latest commit:', error.message);
}
return null;
}
/**
* Fetch latest pipeline from Woodpecker
*/
async function getLatestPipeline(): Promise<PipelineInfo | null> {
if (!WOODPECKER_TOKEN) {
console.warn('[DeployStatus] WOODPECKER_TOKEN not set, skipping pipeline fetch');
return null;
}
try {
// Get latest pipeline
const listResponse = await axios.get(
`${WOODPECKER_SERVER}/api/repos/${REPO_OWNER}/${REPO_NAME}/pipelines?page=1&per_page=1`,
{
headers: { Authorization: `Bearer ${WOODPECKER_TOKEN}` },
timeout: 5000,
}
);
if (!listResponse.data || listResponse.data.length === 0) {
return null;
}
const pipeline = listResponse.data[0];
// Get pipeline steps
let steps: PipelineStep[] = [];
try {
const stepsResponse = await axios.get(
`${WOODPECKER_SERVER}/api/repos/${REPO_OWNER}/${REPO_NAME}/pipelines/${pipeline.number}`,
{
headers: { Authorization: `Bearer ${WOODPECKER_TOKEN}` },
timeout: 5000,
}
);
if (stepsResponse.data?.workflows) {
for (const workflow of stepsResponse.data.workflows) {
if (workflow.children) {
for (const step of workflow.children) {
steps.push({
name: step.name,
state: step.state,
started: step.start_time,
stopped: step.end_time,
});
}
}
}
}
} catch (stepError) {
// Steps fetch failed, continue without them
}
return {
number: pipeline.number,
status: pipeline.status,
event: pipeline.event,
branch: pipeline.branch,
message: pipeline.message?.split('\n')[0] || '',
commit: pipeline.commit?.slice(0, 8) || '',
author: pipeline.author || 'unknown',
created: pipeline.created_at,
started: pipeline.started_at,
finished: pipeline.finished_at,
steps,
};
} catch (error: any) {
console.error('[DeployStatus] Failed to fetch pipeline:', error.message);
}
return null;
}
/**
* Count commits between two SHAs
*/
async function countCommitsBetween(fromSha: string, toSha: string): Promise<number> {
if (!GITEA_TOKEN || !fromSha || !toSha) return 0;
if (fromSha === toSha) return 0;
try {
const response = await axios.get(
`${GITEA_SERVER}/api/v1/repos/${REPO_OWNER}/${REPO_NAME}/commits?sha=${toSha}&limit=50`,
{
headers: { Authorization: `token ${GITEA_TOKEN}` },
timeout: 5000,
}
);
if (response.data) {
const commits = response.data;
for (let i = 0; i < commits.length; i++) {
if (commits[i].sha.startsWith(fromSha)) {
return i;
}
}
// If not found in first 50, assume more than 50 behind
return commits.length;
}
} catch (error: any) {
console.error('[DeployStatus] Failed to count commits:', error.message);
}
return 0;
}
/**
* GET /api/admin/deploy-status
* Returns deployment status with version comparison and CI info
*/
router.get('/', async (req: Request, res: Response) => {
try {
// Get running version from env vars (set during Docker build)
const runningSha = process.env.APP_GIT_SHA || 'unknown';
const running = {
sha: runningSha.slice(0, 8),
sha_full: runningSha,
build_time: process.env.APP_BUILD_TIME || new Date().toISOString(),
image_tag: process.env.CONTAINER_IMAGE_TAG?.slice(0, 8) || 'local',
};
// Fetch latest commit and pipeline in parallel
const [latestCommit, pipeline] = await Promise.all([
getLatestCommit(),
getLatestPipeline(),
]);
// Build latest info
const latest = latestCommit ? {
sha: latestCommit.sha.slice(0, 8),
sha_full: latestCommit.sha,
message: latestCommit.message,
author: latestCommit.author,
timestamp: latestCommit.timestamp,
} : null;
// Determine if running latest
const isLatest = latest
? runningSha.startsWith(latest.sha_full.slice(0, 8)) ||
latest.sha_full.startsWith(runningSha.slice(0, 8))
: true;
// Count commits behind
const commitsBehind = isLatest
? 0
: await countCommitsBetween(runningSha, latest?.sha_full || '');
const response: DeployStatusResponse = {
running,
latest,
is_latest: isLatest,
commits_behind: commitsBehind,
pipeline,
};
res.json(response);
} catch (error: any) {
console.error('[DeployStatus] Error:', error);
res.status(500).json({
error: error.message,
running: {
sha: process.env.APP_GIT_SHA?.slice(0, 8) || 'unknown',
sha_full: process.env.APP_GIT_SHA || 'unknown',
build_time: process.env.APP_BUILD_TIME || 'unknown',
image_tag: process.env.CONTAINER_IMAGE_TAG?.slice(0, 8) || 'local',
},
latest: null,
is_latest: true,
commits_behind: 0,
pipeline: null,
});
}
});
export default router;

View File

@@ -8,10 +8,12 @@ router.use(authMiddleware);
// Valid menu_type values
const VALID_MENU_TYPES = ['dutchie', 'treez', 'jane', 'weedmaps', 'leafly', 'meadow', 'blaze', 'flowhub', 'dispense', 'cova', 'other', 'unknown'];
// Get all dispensaries
// Get all dispensaries (with pagination)
router.get('/', async (req, res) => {
try {
const { menu_type, city, state, crawl_enabled, dutchie_verified } = req.query;
const { menu_type, city, state, crawl_enabled, dutchie_verified, limit, offset, search } = req.query;
const pageLimit = Math.min(parseInt(limit as string) || 50, 500);
const pageOffset = parseInt(offset as string) || 0;
let query = `
SELECT
@@ -98,15 +100,34 @@ router.get('/', async (req, res) => {
}
}
if (conditions.length > 0) {
query += ` WHERE ${conditions.join(' AND ')}`;
// Search filter (name, dba_name, city, company_name)
if (search) {
conditions.push(`(name ILIKE $${params.length + 1} OR dba_name ILIKE $${params.length + 1} OR city ILIKE $${params.length + 1})`);
params.push(`%${search}%`);
}
// Build WHERE clause
const whereClause = conditions.length > 0 ? ` WHERE ${conditions.join(' AND ')}` : '';
// Get total count first
const countResult = await pool.query(`SELECT COUNT(*) FROM dispensaries${whereClause}`, params);
const total = parseInt(countResult.rows[0].count);
// Add pagination
query += whereClause;
query += ` ORDER BY name`;
query += ` LIMIT $${params.length + 1} OFFSET $${params.length + 2}`;
params.push(pageLimit, pageOffset);
const result = await pool.query(query, params);
res.json({ dispensaries: result.rows, total: result.rowCount });
res.json({
dispensaries: result.rows,
total,
limit: pageLimit,
offset: pageOffset,
hasMore: pageOffset + result.rows.length < total
});
} catch (error) {
console.error('Error fetching dispensaries:', error);
res.status(500).json({ error: 'Failed to fetch dispensaries' });

View File

@@ -45,6 +45,8 @@ interface ApiHealth extends HealthStatus {
uptime: number;
timestamp: string;
version: string;
build_sha: string | null;
build_time: string | null;
}
interface DbHealth extends HealthStatus {
@@ -113,6 +115,8 @@ async function getApiHealth(): Promise<ApiHealth> {
uptime: Math.floor((Date.now() - serverStartTime) / 1000),
timestamp: new Date().toISOString(),
version: packageVersion,
build_sha: process.env.APP_GIT_SHA && process.env.APP_GIT_SHA !== 'unknown' ? process.env.APP_GIT_SHA : null,
build_time: process.env.APP_BUILD_TIME && process.env.APP_BUILD_TIME !== 'unknown' ? process.env.APP_BUILD_TIME : null,
};
}
@@ -138,14 +142,16 @@ async function getDbHealth(): Promise<DbHealth> {
async function getRedisHealth(): Promise<RedisHealth> {
const start = Date.now();
const isLocal = process.env.NODE_ENV === 'development' || process.env.NODE_ENV === 'local' || !process.env.NODE_ENV;
// Check if Redis is configured
if (!process.env.REDIS_URL && !process.env.REDIS_HOST) {
// Redis is optional in local dev, required in prod/staging
return {
status: 'ok', // Redis is optional
status: isLocal ? 'ok' : 'error',
connected: false,
latency_ms: 0,
error: 'Redis not configured',
error: isLocal ? 'Redis not configured (optional in local)' : 'Redis not configured (required in production)',
};
}

View File

@@ -0,0 +1,214 @@
/**
* Image Proxy Route
*
* On-demand image resizing service. Serves images with URL-based transforms.
*
* Usage:
* /img/<path>?w=200&h=200&q=80&fit=cover
*
* Parameters:
* w - width (pixels)
* h - height (pixels)
* q - quality (1-100, default 80)
* fit - resize fit: cover, contain, fill, inside, outside (default: inside)
* blur - blur sigma (0.3-1000)
* gray - grayscale (1 = enabled)
* format - output format: webp, jpeg, png, avif (default: webp)
*
* Examples:
* /img/products/az/store/brand/product/image.webp?w=200
* /img/products/az/store/brand/product/image.webp?w=600&h=400&fit=cover
* /img/products/az/store/brand/product/image.webp?w=100&blur=5&gray=1
*/
import { Router, Request, Response } from 'express';
import * as fs from 'fs/promises';
import * as path from 'path';
// @ts-ignore
const sharp = require('sharp');
const router = Router();
// Base path for images
function getImagesBasePath(): string {
if (process.env.IMAGES_PATH) {
return process.env.IMAGES_PATH;
}
if (process.env.STORAGE_BASE_PATH) {
return path.join(process.env.STORAGE_BASE_PATH, 'images');
}
return './storage/images';
}
const IMAGES_BASE_PATH = getImagesBasePath();
// Allowed fit modes
const ALLOWED_FITS = ['cover', 'contain', 'fill', 'inside', 'outside'] as const;
type FitMode = typeof ALLOWED_FITS[number];
// Allowed formats
const ALLOWED_FORMATS = ['webp', 'jpeg', 'jpg', 'png', 'avif'] as const;
type OutputFormat = typeof ALLOWED_FORMATS[number];
// Cache headers (1 year for immutable content-addressed images)
const CACHE_MAX_AGE = 31536000; // 1 year in seconds
interface TransformParams {
width?: number;
height?: number;
quality: number;
fit: FitMode;
blur?: number;
grayscale: boolean;
format: OutputFormat;
}
function parseTransformParams(query: any): TransformParams {
return {
width: query.w ? Math.min(Math.max(parseInt(query.w, 10), 1), 4000) : undefined,
height: query.h ? Math.min(Math.max(parseInt(query.h, 10), 1), 4000) : undefined,
quality: query.q ? Math.min(Math.max(parseInt(query.q, 10), 1), 100) : 80,
fit: ALLOWED_FITS.includes(query.fit) ? query.fit : 'inside',
blur: query.blur ? Math.min(Math.max(parseFloat(query.blur), 0.3), 1000) : undefined,
grayscale: query.gray === '1' || query.grayscale === '1',
format: ALLOWED_FORMATS.includes(query.format) ? query.format : 'webp',
};
}
function getContentType(format: OutputFormat): string {
switch (format) {
case 'jpeg':
case 'jpg':
return 'image/jpeg';
case 'png':
return 'image/png';
case 'avif':
return 'image/avif';
case 'webp':
default:
return 'image/webp';
}
}
/**
* Image proxy endpoint
* GET /img/*
*/
router.get('/*', async (req: Request, res: Response) => {
try {
// Get the image path from URL (everything after /img/)
const imagePath = req.params[0];
if (!imagePath) {
return res.status(400).json({ error: 'Image path required' });
}
// Security: prevent directory traversal
const normalizedPath = path.normalize(imagePath).replace(/^(\.\.(\/|\\|$))+/, '');
const basePath = path.resolve(IMAGES_BASE_PATH);
const fullPath = path.resolve(path.join(IMAGES_BASE_PATH, normalizedPath));
// Ensure path is within base directory
if (!fullPath.startsWith(basePath)) {
console.error(`[ImageProxy] Path traversal attempt: ${fullPath} not in ${basePath}`);
return res.status(403).json({ error: 'Access denied' });
}
// Check if file exists
try {
await fs.access(fullPath);
} catch {
return res.status(404).json({ error: 'Image not found' });
}
// Parse transform parameters
const params = parseTransformParams(req.query);
// Check if any transforms are requested
const hasTransforms = params.width || params.height || params.blur || params.grayscale;
// Read the original image
const imageBuffer = await fs.readFile(fullPath);
let outputBuffer: Buffer;
if (hasTransforms) {
// Apply transforms
let pipeline = sharp(imageBuffer);
// Resize
if (params.width || params.height) {
pipeline = pipeline.resize(params.width, params.height, {
fit: params.fit,
withoutEnlargement: true,
});
}
// Blur
if (params.blur) {
pipeline = pipeline.blur(params.blur);
}
// Grayscale
if (params.grayscale) {
pipeline = pipeline.grayscale();
}
// Output format
switch (params.format) {
case 'jpeg':
case 'jpg':
pipeline = pipeline.jpeg({ quality: params.quality });
break;
case 'png':
pipeline = pipeline.png({ quality: params.quality });
break;
case 'avif':
pipeline = pipeline.avif({ quality: params.quality });
break;
case 'webp':
default:
pipeline = pipeline.webp({ quality: params.quality });
}
outputBuffer = await pipeline.toBuffer();
} else {
// No transforms - serve original (but maybe convert format)
if (params.format !== 'webp' || params.quality !== 80) {
let pipeline = sharp(imageBuffer);
switch (params.format) {
case 'jpeg':
case 'jpg':
pipeline = pipeline.jpeg({ quality: params.quality });
break;
case 'png':
pipeline = pipeline.png({ quality: params.quality });
break;
case 'avif':
pipeline = pipeline.avif({ quality: params.quality });
break;
case 'webp':
default:
pipeline = pipeline.webp({ quality: params.quality });
}
outputBuffer = await pipeline.toBuffer();
} else {
outputBuffer = imageBuffer;
}
}
// Set headers
res.setHeader('Content-Type', getContentType(params.format));
res.setHeader('Cache-Control', `public, max-age=${CACHE_MAX_AGE}, immutable`);
res.setHeader('X-Image-Size', outputBuffer.length);
// Send image
res.send(outputBuffer);
} catch (error: any) {
console.error('[ImageProxy] Error:', error.message);
res.status(500).json({ error: 'Failed to process image' });
}
});
export default router;

View File

@@ -143,6 +143,152 @@ router.get('/', async (req: Request, res: Response) => {
}
});
/**
* GET /api/job-queue/available - List dispensaries available for crawling
* Query: { state_code?: string, limit?: number }
* NOTE: Must be defined BEFORE /:id route to avoid conflict
*/
router.get('/available', async (req: Request, res: Response) => {
try {
const { state_code, limit = '100' } = req.query;
let query = `
SELECT
d.id,
d.name,
d.city,
s.code as state_code,
d.platform_dispensary_id,
d.crawl_enabled,
(SELECT MAX(created_at) FROM dispensary_crawl_jobs WHERE dispensary_id = d.id AND status = 'completed') as last_crawl,
EXISTS (
SELECT 1 FROM dispensary_crawl_jobs
WHERE dispensary_id = d.id AND status IN ('pending', 'running')
) as has_pending_job
FROM dispensaries d
LEFT JOIN states s ON s.id = d.state_id
WHERE d.crawl_enabled = true
AND d.platform_dispensary_id IS NOT NULL
`;
const params: any[] = [];
let paramIndex = 1;
if (state_code) {
params.push((state_code as string).toUpperCase());
query += ` AND s.code = $${paramIndex++}`;
}
query += ` ORDER BY d.name LIMIT $${paramIndex}`;
params.push(parseInt(limit as string));
const { rows } = await pool.query(query, params);
// Get counts by state
const { rows: stateCounts } = await pool.query(`
SELECT s.code, COUNT(*) as count
FROM dispensaries d
JOIN states s ON s.id = d.state_id
WHERE d.crawl_enabled = true
AND d.platform_dispensary_id IS NOT NULL
GROUP BY s.code
ORDER BY count DESC
`);
res.json({
success: true,
dispensaries: rows,
total: rows.length,
by_state: stateCounts
});
} catch (error: any) {
console.error('[JobQueue] Error listing available:', error);
res.status(500).json({ success: false, error: error.message });
}
});
/**
* GET /api/job-queue/history - Get recent job history with results
* Query: { state_code?: string, status?: string, limit?: number, hours?: number }
* NOTE: Must be defined BEFORE /:id route to avoid conflict
*/
router.get('/history', async (req: Request, res: Response) => {
try {
const {
state_code,
status,
limit = '50',
hours = '24'
} = req.query;
let query = `
SELECT
j.id,
j.dispensary_id,
d.name as dispensary_name,
s.code as state_code,
j.job_type,
j.status,
j.products_found,
j.error_message,
j.started_at,
j.completed_at,
j.duration_ms,
j.created_at
FROM dispensary_crawl_jobs j
LEFT JOIN dispensaries d ON d.id = j.dispensary_id
LEFT JOIN states s ON s.id = d.state_id
WHERE j.created_at > NOW() - INTERVAL '${parseInt(hours as string)} hours'
`;
const params: any[] = [];
let paramIndex = 1;
if (status && status !== 'all') {
params.push(status);
query += ` AND j.status = $${paramIndex++}`;
}
if (state_code) {
params.push((state_code as string).toUpperCase());
query += ` AND s.code = $${paramIndex++}`;
}
query += ` ORDER BY j.created_at DESC LIMIT $${paramIndex}`;
params.push(parseInt(limit as string));
const { rows } = await pool.query(query, params);
// Get summary stats
const { rows: stats } = await pool.query(`
SELECT
COUNT(*) FILTER (WHERE status = 'completed') as completed,
COUNT(*) FILTER (WHERE status = 'failed') as failed,
COUNT(*) FILTER (WHERE status = 'running') as running,
COUNT(*) FILTER (WHERE status = 'pending') as pending,
SUM(products_found) FILTER (WHERE status = 'completed') as total_products,
AVG(duration_ms) FILTER (WHERE status = 'completed') as avg_duration_ms
FROM dispensary_crawl_jobs
WHERE created_at > NOW() - INTERVAL '${parseInt(hours as string)} hours'
`);
res.json({
success: true,
jobs: rows,
summary: {
completed: parseInt(stats[0].completed) || 0,
failed: parseInt(stats[0].failed) || 0,
running: parseInt(stats[0].running) || 0,
pending: parseInt(stats[0].pending) || 0,
total_products: parseInt(stats[0].total_products) || 0,
avg_duration_ms: Math.round(parseFloat(stats[0].avg_duration_ms)) || null
},
hours: parseInt(hours as string)
});
} catch (error: any) {
console.error('[JobQueue] Error getting history:', error);
res.status(500).json({ success: false, error: error.message });
}
});
/**
* GET /api/job-queue/stats - Queue statistics
*/
@@ -463,5 +609,165 @@ router.get('/paused', async (_req: Request, res: Response) => {
res.json({ success: true, queue_paused: queuePaused });
});
/**
* POST /api/job-queue/enqueue-batch - Queue multiple dispensaries at once
* Body: { dispensary_ids: number[], job_type?: string, priority?: number }
*/
router.post('/enqueue-batch', async (req: Request, res: Response) => {
try {
const { dispensary_ids, job_type = 'dutchie_product_crawl', priority = 0 } = req.body;
if (!Array.isArray(dispensary_ids) || dispensary_ids.length === 0) {
return res.status(400).json({ success: false, error: 'dispensary_ids array is required' });
}
if (dispensary_ids.length > 500) {
return res.status(400).json({ success: false, error: 'Maximum 500 dispensaries per batch' });
}
// Insert jobs, skipping duplicates
const { rows } = await pool.query(`
INSERT INTO dispensary_crawl_jobs (dispensary_id, job_type, priority, trigger_type, status, created_at)
SELECT
d.id,
$2::text,
$3::integer,
'api_batch',
'pending',
NOW()
FROM dispensaries d
WHERE d.id = ANY($1::int[])
AND d.crawl_enabled = true
AND d.platform_dispensary_id IS NOT NULL
AND NOT EXISTS (
SELECT 1 FROM dispensary_crawl_jobs cj
WHERE cj.dispensary_id = d.id
AND cj.job_type = $2::text
AND cj.status IN ('pending', 'running')
)
RETURNING id, dispensary_id
`, [dispensary_ids, job_type, priority]);
res.json({
success: true,
queued: rows.length,
requested: dispensary_ids.length,
job_ids: rows.map(r => r.id),
message: `Queued ${rows.length} of ${dispensary_ids.length} dispensaries`
});
} catch (error: any) {
console.error('[JobQueue] Error batch enqueuing:', error);
res.status(500).json({ success: false, error: error.message });
}
});
/**
* POST /api/job-queue/enqueue-state - Queue all crawl-enabled dispensaries for a state
* Body: { state_code: string, job_type?: string, priority?: number, limit?: number }
*/
router.post('/enqueue-state', async (req: Request, res: Response) => {
try {
const { state_code, job_type = 'dutchie_product_crawl', priority = 0, limit = 200 } = req.body;
if (!state_code) {
return res.status(400).json({ success: false, error: 'state_code is required (e.g., "AZ")' });
}
// Get state_id and queue jobs
const { rows } = await pool.query(`
WITH target_state AS (
SELECT id FROM states WHERE code = $1
)
INSERT INTO dispensary_crawl_jobs (dispensary_id, job_type, priority, trigger_type, status, created_at)
SELECT
d.id,
$2::text,
$3::integer,
'api_state',
'pending',
NOW()
FROM dispensaries d, target_state
WHERE d.state_id = target_state.id
AND d.crawl_enabled = true
AND d.platform_dispensary_id IS NOT NULL
AND NOT EXISTS (
SELECT 1 FROM dispensary_crawl_jobs cj
WHERE cj.dispensary_id = d.id
AND cj.job_type = $2::text
AND cj.status IN ('pending', 'running')
)
LIMIT $4::integer
RETURNING id, dispensary_id
`, [state_code.toUpperCase(), job_type, priority, limit]);
// Get total available count
const countResult = await pool.query(`
WITH target_state AS (
SELECT id FROM states WHERE code = $1
)
SELECT COUNT(*) as total
FROM dispensaries d, target_state
WHERE d.state_id = target_state.id
AND d.crawl_enabled = true
AND d.platform_dispensary_id IS NOT NULL
`, [state_code.toUpperCase()]);
res.json({
success: true,
queued: rows.length,
total_available: parseInt(countResult.rows[0].total),
state: state_code.toUpperCase(),
job_type,
message: `Queued ${rows.length} dispensaries for ${state_code.toUpperCase()}`
});
} catch (error: any) {
console.error('[JobQueue] Error enqueuing state:', error);
res.status(500).json({ success: false, error: error.message });
}
});
/**
* POST /api/job-queue/clear-pending - Clear all pending jobs (optionally filtered)
* Body: { state_code?: string, job_type?: string }
*/
router.post('/clear-pending', async (req: Request, res: Response) => {
try {
const { state_code, job_type } = req.body;
let query = `
UPDATE dispensary_crawl_jobs
SET status = 'cancelled', completed_at = NOW(), updated_at = NOW()
WHERE status = 'pending'
`;
const params: any[] = [];
let paramIndex = 1;
if (job_type) {
params.push(job_type);
query += ` AND job_type = $${paramIndex++}`;
}
if (state_code) {
params.push((state_code as string).toUpperCase());
query += ` AND dispensary_id IN (
SELECT d.id FROM dispensaries d
JOIN states s ON s.id = d.state_id
WHERE s.code = $${paramIndex++}
)`;
}
const result = await pool.query(query, params);
res.json({
success: true,
cleared: result.rowCount,
message: `Cancelled ${result.rowCount} pending jobs`
});
} catch (error: any) {
console.error('[JobQueue] Error clearing pending:', error);
res.status(500).json({ success: false, error: error.message });
}
});
export default router;
export { queuePaused };

View File

@@ -120,6 +120,50 @@ function isDomainAllowed(origin: string, allowedDomains: string[]): boolean {
}
}
// Trusted origins for consumer sites (bypass API key auth)
const CONSUMER_TRUSTED_ORIGINS = [
'https://findagram.co',
'https://www.findagram.co',
'https://findadispo.com',
'https://www.findadispo.com',
'http://localhost:3001',
'http://localhost:3002',
];
// Trusted IPs for local development (bypass API key auth)
const TRUSTED_IPS = ['127.0.0.1', '::1', '::ffff:127.0.0.1'];
/**
* Check if request is from localhost
*/
function isLocalhost(req: Request): boolean {
const clientIp = req.ip || req.socket.remoteAddress || '';
return TRUSTED_IPS.includes(clientIp);
}
/**
* Check if request is from a trusted consumer origin
*/
function isConsumerTrustedRequest(req: Request): boolean {
// Localhost always bypasses
if (isLocalhost(req)) {
return true;
}
const origin = req.headers.origin;
if (origin && CONSUMER_TRUSTED_ORIGINS.includes(origin)) {
return true;
}
const referer = req.headers.referer;
if (referer) {
for (const trusted of CONSUMER_TRUSTED_ORIGINS) {
if (referer.startsWith(trusted)) {
return true;
}
}
}
return false;
}
/**
* Middleware to validate API key and build scope
*/
@@ -128,6 +172,19 @@ async function validatePublicApiKey(
res: Response,
next: NextFunction
) {
// Allow trusted consumer origins without API key (read-only access to all dispensaries)
if (isConsumerTrustedRequest(req)) {
// Create a synthetic internal permission for consumer sites
req.scope = {
type: 'internal',
dispensaryIds: 'ALL',
apiKeyId: 0,
apiKeyName: 'consumer-site',
rateLimit: 100,
};
return next();
}
const apiKey = req.headers['x-api-key'] as string;
if (!apiKey) {
@@ -373,14 +430,14 @@ router.get('/products', async (req: PublicApiRequest, res: Response) => {
// Filter by category
if (category) {
whereClause += ` AND LOWER(p.category) = LOWER($${paramIndex})`;
whereClause += ` AND LOWER(p.category_raw) = LOWER($${paramIndex})`;
params.push(category);
paramIndex++;
}
// Filter by brand
if (brand) {
whereClause += ` AND LOWER(p.brand_name) LIKE LOWER($${paramIndex})`;
whereClause += ` AND LOWER(p.brand_name_raw) LIKE LOWER($${paramIndex})`;
params.push(`%${brand}%`);
paramIndex++;
}
@@ -411,7 +468,7 @@ router.get('/products', async (req: PublicApiRequest, res: Response) => {
// Search by name or brand
if (search) {
whereClause += ` AND (LOWER(p.name) LIKE LOWER($${paramIndex}) OR LOWER(p.brand_name) LIKE LOWER($${paramIndex}))`;
whereClause += ` AND (LOWER(p.name_raw) LIKE LOWER($${paramIndex}) OR LOWER(p.brand_name_raw) LIKE LOWER($${paramIndex}))`;
params.push(`%${search}%`);
paramIndex++;
}
@@ -422,10 +479,11 @@ router.get('/products', async (req: PublicApiRequest, res: Response) => {
// Build ORDER BY clause (use pricing_type for price sorting)
const sortDirection = sort_dir === 'desc' ? 'DESC' : 'ASC';
let orderBy = 'p.name ASC';
let orderBy = 'p.name_raw ASC';
switch (sort_by) {
case 'price':
const sortPriceCol = pricing_type === 'med' ? 's.price_med' : 's.price_rec';
// View uses *_cents columns, but we SELECT as price_rec/price_med
const sortPriceCol = pricing_type === 'med' ? 's.med_min_price_cents' : 's.rec_min_price_cents';
orderBy = `${sortPriceCol} ${sortDirection} NULLS LAST`;
break;
case 'thc':
@@ -436,13 +494,14 @@ router.get('/products', async (req: PublicApiRequest, res: Response) => {
break;
case 'name':
default:
orderBy = `p.name ${sortDirection}`;
orderBy = `p.name_raw ${sortDirection}`;
}
params.push(limitNum, offsetNum);
// Determine which price column to use for filtering based on pricing_type
const priceColumn = pricing_type === 'med' ? 's.price_med' : 's.price_rec';
// View uses *_cents columns, divide by 100 for dollar comparison
const priceColumn = pricing_type === 'med' ? 's.med_min_price_cents / 100.0' : 's.rec_min_price_cents / 100.0';
// Query products with latest snapshot data
// Uses store_products + v_product_snapshots (canonical tables with raw_data)
@@ -451,10 +510,10 @@ router.get('/products', async (req: PublicApiRequest, res: Response) => {
p.id,
p.dispensary_id,
p.provider_product_id as dutchie_id,
p.name,
p.brand_name as brand,
p.category,
p.subcategory,
p.name_raw as name,
p.brand_name_raw as brand,
p.category_raw as category,
p.subcategory_raw as subcategory,
p.strain_type,
p.stock_status,
p.thc_percent as thc,
@@ -462,19 +521,19 @@ router.get('/products', async (req: PublicApiRequest, res: Response) => {
p.image_url,
p.created_at,
p.updated_at,
s.price_rec,
s.price_med,
s.price_rec_special,
s.price_med_special,
s.rec_min_price_cents / 100.0 as price_rec,
s.med_min_price_cents / 100.0 as price_med,
s.rec_min_special_price_cents / 100.0 as price_rec_special,
s.med_min_special_price_cents / 100.0 as price_med_special,
s.stock_quantity as total_quantity_available,
s.is_on_special as special,
s.captured_at as snapshot_at,
s.special,
s.crawled_at as snapshot_at,
${include_variants === 'true' || include_variants === '1' ? "s.raw_data->'POSMetaData'->'children' as variants_raw" : 'NULL as variants_raw'}
FROM store_products p
LEFT JOIN LATERAL (
SELECT * FROM v_product_snapshots
WHERE store_product_id = p.id
ORDER BY captured_at DESC
ORDER BY crawled_at DESC
LIMIT 1
) s ON true
${whereClause}
@@ -488,9 +547,9 @@ router.get('/products', async (req: PublicApiRequest, res: Response) => {
const { rows: countRows } = await pool.query(`
SELECT COUNT(*) as total FROM store_products p
LEFT JOIN LATERAL (
SELECT price_rec, price_med, is_on_special FROM v_product_snapshots
SELECT rec_min_price_cents / 100.0 as price_rec, med_min_price_cents / 100.0 as price_med, special as is_on_special FROM v_product_snapshots
WHERE store_product_id = p.id
ORDER BY captured_at DESC
ORDER BY crawled_at DESC
LIMIT 1
) s ON true
${whereClause}
@@ -945,22 +1004,27 @@ router.get('/dispensaries', async (req: PublicApiRequest, res: Response) => {
SELECT
d.id,
d.name,
d.address,
d.address1,
d.address2,
d.city,
d.state,
d.zip,
d.zipcode as zip,
d.phone,
d.email,
d.website,
d.latitude,
d.longitude,
d.menu_type as platform,
d.menu_url,
d.hours,
d.amenities,
d.description,
d.image_url,
d.logo_image as image_url,
d.google_rating,
d.google_review_count,
d.offer_pickup,
d.offer_delivery,
d.offer_curbside_pickup,
d.is_medical,
d.is_recreational,
COALESCE(pc.product_count, 0) as product_count,
COALESCE(pc.in_stock_count, 0) as in_stock_count,
pc.last_updated
@@ -994,11 +1058,13 @@ router.get('/dispensaries', async (req: PublicApiRequest, res: Response) => {
dispensaries: [{
id: d.id,
name: d.name,
address: d.address,
address1: d.address1,
address2: d.address2,
city: d.city,
state: d.state,
zip: d.zip,
phone: d.phone,
email: d.email,
website: d.website,
menu_url: d.menu_url,
location: d.latitude && d.longitude ? {
@@ -1006,10 +1072,17 @@ router.get('/dispensaries', async (req: PublicApiRequest, res: Response) => {
longitude: parseFloat(d.longitude)
} : null,
platform: d.platform,
hours: d.hours || null,
amenities: d.amenities || [],
description: d.description || null,
image_url: d.image_url || null,
services: {
pickup: d.offer_pickup || false,
delivery: d.offer_delivery || false,
curbside: d.offer_curbside_pickup || false
},
license_type: {
medical: d.is_medical || false,
recreational: d.is_recreational || false
},
rating: d.google_rating ? parseFloat(d.google_rating) : null,
review_count: d.google_review_count ? parseInt(d.google_review_count, 10) : null,
product_count: parseInt(d.product_count || '0', 10),
@@ -1052,22 +1125,27 @@ router.get('/dispensaries', async (req: PublicApiRequest, res: Response) => {
SELECT
d.id,
d.name,
d.address,
d.address1,
d.address2,
d.city,
d.state,
d.zip,
d.zipcode as zip,
d.phone,
d.email,
d.website,
d.latitude,
d.longitude,
d.menu_type as platform,
d.menu_url,
d.hours,
d.amenities,
d.description,
d.image_url,
d.logo_image as image_url,
d.google_rating,
d.google_review_count,
d.offer_pickup,
d.offer_delivery,
d.offer_curbside_pickup,
d.is_medical,
d.is_recreational,
COALESCE(pc.product_count, 0) as product_count,
COALESCE(pc.in_stock_count, 0) as in_stock_count,
pc.last_updated
@@ -1101,11 +1179,13 @@ router.get('/dispensaries', async (req: PublicApiRequest, res: Response) => {
const transformedDispensaries = dispensaries.map((d) => ({
id: d.id,
name: d.name,
address: d.address,
address1: d.address1,
address2: d.address2,
city: d.city,
state: d.state,
zip: d.zip,
phone: d.phone,
email: d.email,
website: d.website,
menu_url: d.menu_url,
location: d.latitude && d.longitude ? {
@@ -1113,10 +1193,17 @@ router.get('/dispensaries', async (req: PublicApiRequest, res: Response) => {
longitude: parseFloat(d.longitude)
} : null,
platform: d.platform,
hours: d.hours || null,
amenities: d.amenities || [],
description: d.description || null,
image_url: d.image_url || null,
services: {
pickup: d.offer_pickup || false,
delivery: d.offer_delivery || false,
curbside: d.offer_curbside_pickup || false
},
license_type: {
medical: d.is_medical || false,
recreational: d.is_recreational || false
},
rating: d.google_rating ? parseFloat(d.google_rating) : null,
review_count: d.google_review_count ? parseInt(d.google_review_count, 10) : null,
product_count: parseInt(d.product_count || '0', 10),
@@ -1307,6 +1394,510 @@ router.get('/search', async (req: PublicApiRequest, res: Response) => {
}
});
// ============================================================
// STORE METRICS & INTELLIGENCE ENDPOINTS
// ============================================================
/**
* GET /api/v1/stores/:id/metrics
* Get performance metrics for a specific store
*
* Returns:
* - Product counts (total, in-stock, out-of-stock)
* - Brand counts
* - Category breakdown
* - Price statistics (avg, min, max)
* - Stock health metrics
* - Crawl status
*/
router.get('/stores/:id/metrics', async (req: PublicApiRequest, res: Response) => {
try {
const scope = req.scope!;
const storeId = parseInt(req.params.id, 10);
if (isNaN(storeId)) {
return res.status(400).json({ error: 'Invalid store ID' });
}
// Validate access
if (scope.type === 'wordpress' && scope.dispensaryIds !== 'ALL') {
if (!scope.dispensaryIds.includes(storeId)) {
return res.status(403).json({ error: 'Access denied to this store' });
}
}
// Get store info
const { rows: storeRows } = await pool.query(`
SELECT id, name, city, state, last_crawl_at, product_count, crawl_enabled
FROM dispensaries
WHERE id = $1
`, [storeId]);
if (storeRows.length === 0) {
return res.status(404).json({ error: 'Store not found' });
}
const store = storeRows[0];
// Get product metrics
const { rows: productMetrics } = await pool.query(`
SELECT
COUNT(*) as total_products,
COUNT(*) FILTER (WHERE stock_status = 'in_stock') as in_stock,
COUNT(*) FILTER (WHERE stock_status = 'out_of_stock') as out_of_stock,
COUNT(DISTINCT brand_name_raw) FILTER (WHERE brand_name_raw IS NOT NULL) as unique_brands,
COUNT(DISTINCT category_raw) FILTER (WHERE category_raw IS NOT NULL) as unique_categories
FROM store_products
WHERE dispensary_id = $1
`, [storeId]);
// Get price statistics from latest snapshots
const { rows: priceStats } = await pool.query(`
SELECT
ROUND(AVG(price_rec)::numeric, 2) as avg_price,
MIN(price_rec) as min_price,
MAX(price_rec) as max_price,
COUNT(*) FILTER (WHERE is_on_special = true) as on_special_count
FROM store_product_snapshots sps
INNER JOIN (
SELECT store_product_id, MAX(captured_at) as latest
FROM store_product_snapshots
WHERE dispensary_id = $1
GROUP BY store_product_id
) latest ON sps.store_product_id = latest.store_product_id AND sps.captured_at = latest.latest
WHERE sps.dispensary_id = $1 AND sps.price_rec > 0
`, [storeId]);
// Get category breakdown
const { rows: categoryBreakdown } = await pool.query(`
SELECT
COALESCE(category_raw, 'Uncategorized') as category,
COUNT(*) as count,
COUNT(*) FILTER (WHERE stock_status = 'in_stock') as in_stock
FROM store_products
WHERE dispensary_id = $1
GROUP BY category_raw
ORDER BY count DESC
LIMIT 10
`, [storeId]);
// Calculate stock health
const metrics = productMetrics[0] || {};
const totalProducts = parseInt(metrics.total_products || '0', 10);
const inStock = parseInt(metrics.in_stock || '0', 10);
const stockHealthPercent = totalProducts > 0 ? Math.round((inStock / totalProducts) * 100) : 0;
const prices = priceStats[0] || {};
res.json({
success: true,
store_id: storeId,
store_name: store.name,
location: {
city: store.city,
state: store.state
},
metrics: {
products: {
total: totalProducts,
in_stock: inStock,
out_of_stock: parseInt(metrics.out_of_stock || '0', 10),
stock_health_percent: stockHealthPercent
},
brands: {
unique_count: parseInt(metrics.unique_brands || '0', 10)
},
categories: {
unique_count: parseInt(metrics.unique_categories || '0', 10),
breakdown: categoryBreakdown.map(c => ({
name: c.category,
total: parseInt(c.count, 10),
in_stock: parseInt(c.in_stock, 10)
}))
},
pricing: {
average: prices.avg_price ? parseFloat(prices.avg_price) : null,
min: prices.min_price ? parseFloat(prices.min_price) : null,
max: prices.max_price ? parseFloat(prices.max_price) : null,
on_special_count: parseInt(prices.on_special_count || '0', 10)
},
crawl: {
enabled: store.crawl_enabled,
last_crawl_at: store.last_crawl_at,
product_count_from_crawl: store.product_count
}
},
generated_at: new Date().toISOString()
});
} catch (error: any) {
console.error('Store metrics error:', error);
res.status(500).json({ error: 'Failed to fetch store metrics', message: error.message });
}
});
/**
* GET /api/v1/stores/:id/product-metrics
* Get detailed product-level metrics for a store
*
* Query params:
* - category: Filter by category
* - brand: Filter by brand
* - sort_by: price_change, stock_status, price (default: price_change)
* - limit: Max results (default: 50, max: 200)
*
* Returns per-product:
* - Current price and stock
* - Price change from last crawl
* - Days in stock / out of stock
* - Special/discount status
*/
router.get('/stores/:id/product-metrics', async (req: PublicApiRequest, res: Response) => {
try {
const scope = req.scope!;
const storeId = parseInt(req.params.id, 10);
if (isNaN(storeId)) {
return res.status(400).json({ error: 'Invalid store ID' });
}
// Validate access
if (scope.type === 'wordpress' && scope.dispensaryIds !== 'ALL') {
if (!scope.dispensaryIds.includes(storeId)) {
return res.status(403).json({ error: 'Access denied to this store' });
}
}
const { category, brand, sort_by = 'price_change', limit = '50' } = req.query;
const limitNum = Math.min(parseInt(limit as string, 10) || 50, 200);
let whereClause = 'WHERE sp.dispensary_id = $1';
const params: any[] = [storeId];
let paramIndex = 2;
if (category) {
whereClause += ` AND LOWER(sp.category) = LOWER($${paramIndex})`;
params.push(category);
paramIndex++;
}
if (brand) {
whereClause += ` AND LOWER(sp.brand_name) LIKE LOWER($${paramIndex})`;
params.push(`%${brand}%`);
paramIndex++;
}
params.push(limitNum);
// Get products with their latest and previous snapshots for price comparison
const { rows: products } = await pool.query(`
WITH latest_snapshots AS (
SELECT DISTINCT ON (store_product_id)
store_product_id,
price_rec as current_price,
price_rec_special as current_special_price,
is_on_special,
stock_quantity,
captured_at as last_seen
FROM store_product_snapshots
WHERE dispensary_id = $1
ORDER BY store_product_id, captured_at DESC
),
previous_snapshots AS (
SELECT DISTINCT ON (store_product_id)
store_product_id,
price_rec as previous_price,
captured_at as previous_seen
FROM store_product_snapshots sps
WHERE dispensary_id = $1
AND captured_at < (SELECT MIN(last_seen) FROM latest_snapshots ls WHERE ls.store_product_id = sps.store_product_id)
ORDER BY store_product_id, captured_at DESC
)
SELECT
sp.id,
sp.name_raw as name,
sp.brand_name_raw as brand_name,
sp.category_raw as category,
sp.stock_status,
ls.current_price,
ls.current_special_price,
ls.is_on_special,
ls.stock_quantity,
ls.last_seen,
ps.previous_price,
ps.previous_seen,
CASE
WHEN ls.current_price IS NOT NULL AND ps.previous_price IS NOT NULL
THEN ROUND(((ls.current_price - ps.previous_price) / ps.previous_price * 100)::numeric, 2)
ELSE NULL
END as price_change_percent
FROM store_products sp
LEFT JOIN latest_snapshots ls ON sp.id = ls.store_product_id
LEFT JOIN previous_snapshots ps ON sp.id = ps.store_product_id
${whereClause}
ORDER BY
${sort_by === 'price' ? 'ls.current_price DESC NULLS LAST' :
sort_by === 'stock_status' ? "CASE sp.stock_status WHEN 'out_of_stock' THEN 0 ELSE 1 END, sp.name_raw" :
'ABS(COALESCE(price_change_percent, 0)) DESC'}
LIMIT $${paramIndex}
`, params);
res.json({
success: true,
store_id: storeId,
products: products.map(p => ({
id: p.id,
name: p.name,
brand: p.brand_name,
category: p.category,
stock_status: p.stock_status,
pricing: {
current: p.current_price ? parseFloat(p.current_price) : null,
special: p.current_special_price ? parseFloat(p.current_special_price) : null,
previous: p.previous_price ? parseFloat(p.previous_price) : null,
change_percent: p.price_change_percent ? parseFloat(p.price_change_percent) : null,
is_on_special: p.is_on_special || false
},
inventory: {
quantity: p.stock_quantity || 0,
last_seen: p.last_seen
}
})),
filters: {
category: category || null,
brand: brand || null,
sort_by
},
count: products.length,
generated_at: new Date().toISOString()
});
} catch (error: any) {
console.error('Product metrics error:', error);
res.status(500).json({ error: 'Failed to fetch product metrics', message: error.message });
}
});
/**
* GET /api/v1/stores/:id/competitor-snapshot
* Get competitive intelligence for a store
*
* Returns:
* - Nearby competitor stores (same city/state)
* - Price comparisons by category
* - Brand overlap analysis
* - Market position indicators
*/
router.get('/stores/:id/competitor-snapshot', async (req: PublicApiRequest, res: Response) => {
try {
const scope = req.scope!;
const storeId = parseInt(req.params.id, 10);
if (isNaN(storeId)) {
return res.status(400).json({ error: 'Invalid store ID' });
}
// Validate access
if (scope.type === 'wordpress' && scope.dispensaryIds !== 'ALL') {
if (!scope.dispensaryIds.includes(storeId)) {
return res.status(403).json({ error: 'Access denied to this store' });
}
}
// Get store info
const { rows: storeRows } = await pool.query(`
SELECT id, name, city, state, latitude, longitude
FROM dispensaries
WHERE id = $1
`, [storeId]);
if (storeRows.length === 0) {
return res.status(404).json({ error: 'Store not found' });
}
const store = storeRows[0];
// Get competitor stores in same city (or nearby if coordinates available)
const { rows: competitors } = await pool.query(`
SELECT
d.id,
d.name,
d.city,
d.state,
d.product_count,
d.last_crawl_at,
CASE
WHEN d.latitude IS NOT NULL AND d.longitude IS NOT NULL
AND $2::numeric IS NOT NULL AND $3::numeric IS NOT NULL
THEN ROUND((
6371 * acos(
cos(radians($2::numeric)) * cos(radians(d.latitude::numeric))
* cos(radians(d.longitude::numeric) - radians($3::numeric))
+ sin(radians($2::numeric)) * sin(radians(d.latitude::numeric))
)
)::numeric, 2)
ELSE NULL
END as distance_km
FROM dispensaries d
WHERE d.id != $1
AND d.state = $4
AND d.crawl_enabled = true
AND d.product_count > 0
AND (d.city = $5 OR d.latitude IS NOT NULL)
ORDER BY distance_km NULLS LAST, d.name
LIMIT 10
`, [storeId, store.latitude, store.longitude, store.state, store.city]);
// Get this store's average prices by category
const { rows: storePrices } = await pool.query(`
SELECT
sp.category_raw as category,
ROUND(AVG(sps.price_rec)::numeric, 2) as avg_price,
COUNT(*) as product_count
FROM store_products sp
INNER JOIN (
SELECT DISTINCT ON (store_product_id) store_product_id, price_rec
FROM store_product_snapshots
WHERE dispensary_id = $1
ORDER BY store_product_id, captured_at DESC
) sps ON sp.id = sps.store_product_id
WHERE sp.dispensary_id = $1 AND sp.category_raw IS NOT NULL AND sps.price_rec > 0
GROUP BY sp.category_raw
`, [storeId]);
// Get market average prices by category (all competitors)
const competitorIds = competitors.map(c => c.id);
let marketPrices: any[] = [];
if (competitorIds.length > 0) {
const { rows } = await pool.query(`
SELECT
sp.category_raw as category,
ROUND(AVG(sps.price_rec)::numeric, 2) as market_avg_price,
COUNT(DISTINCT sp.dispensary_id) as store_count
FROM store_products sp
INNER JOIN (
SELECT DISTINCT ON (store_product_id) store_product_id, price_rec
FROM store_product_snapshots
WHERE dispensary_id = ANY($1)
ORDER BY store_product_id, captured_at DESC
) sps ON sp.id = sps.store_product_id
WHERE sp.dispensary_id = ANY($1) AND sp.category_raw IS NOT NULL AND sps.price_rec > 0
GROUP BY sp.category_raw
`, [competitorIds]);
marketPrices = rows;
}
// Get this store's brands
const { rows: storeBrands } = await pool.query(`
SELECT DISTINCT brand_name_raw as brand_name
FROM store_products
WHERE dispensary_id = $1 AND brand_name_raw IS NOT NULL
`, [storeId]);
const storeBrandSet = new Set(storeBrands.map(b => b.brand_name.toLowerCase()));
// Get brand overlap with competitors
let brandOverlap: any[] = [];
if (competitorIds.length > 0) {
const { rows } = await pool.query(`
SELECT
d.id as competitor_id,
d.name as competitor_name,
COUNT(DISTINCT sp.brand_name_raw) as total_brands,
COUNT(DISTINCT sp.brand_name_raw) FILTER (
WHERE LOWER(sp.brand_name_raw) = ANY($2)
) as shared_brands
FROM dispensaries d
INNER JOIN store_products sp ON sp.dispensary_id = d.id
WHERE d.id = ANY($1) AND sp.brand_name_raw IS NOT NULL
GROUP BY d.id, d.name
`, [competitorIds, Array.from(storeBrandSet)]);
brandOverlap = rows;
}
// Build price comparison
const priceComparison = storePrices.map(sp => {
const marketPrice = marketPrices.find(mp => mp.category === sp.category);
const diff = marketPrice
? parseFloat(((parseFloat(sp.avg_price) - parseFloat(marketPrice.market_avg_price)) / parseFloat(marketPrice.market_avg_price) * 100).toFixed(2))
: null;
return {
category: sp.category,
your_avg_price: parseFloat(sp.avg_price),
market_avg_price: marketPrice ? parseFloat(marketPrice.market_avg_price) : null,
diff_percent: diff,
position: diff === null ? 'unknown' : diff < -5 ? 'below_market' : diff > 5 ? 'above_market' : 'at_market'
};
});
res.json({
success: true,
store: {
id: storeId,
name: store.name,
city: store.city,
state: store.state
},
competitors: competitors.map(c => ({
id: c.id,
name: c.name,
city: c.city,
distance_km: c.distance_km ? parseFloat(c.distance_km) : null,
product_count: c.product_count,
last_crawl: c.last_crawl_at
})),
price_comparison: priceComparison,
brand_analysis: {
your_brand_count: storeBrandSet.size,
overlap_with_competitors: brandOverlap.map(bo => ({
competitor_id: bo.competitor_id,
competitor_name: bo.competitor_name,
shared_brands: parseInt(bo.shared_brands, 10),
their_total_brands: parseInt(bo.total_brands, 10),
overlap_percent: Math.round((parseInt(bo.shared_brands, 10) / storeBrandSet.size) * 100)
}))
},
generated_at: new Date().toISOString()
});
} catch (error: any) {
console.error('Competitor snapshot error:', error);
res.status(500).json({ error: 'Failed to fetch competitor snapshot', message: error.message });
}
});
/**
* GET /api/v1/stats
* Get aggregate stats for consumer sites (product count, brand count, dispensary count)
*/
router.get('/stats', async (req: PublicApiRequest, res: Response) => {
try {
// Get aggregate stats across all data
const { rows: stats } = await pool.query(`
SELECT
(SELECT COUNT(*) FROM store_products) as product_count,
(SELECT COUNT(DISTINCT brand_name_raw) FROM store_products WHERE brand_name_raw IS NOT NULL) as brand_count,
(SELECT COUNT(*) FROM dispensaries WHERE crawl_enabled = true AND product_count > 0) as dispensary_count
`);
const s = stats[0] || {};
res.json({
success: true,
stats: {
products: parseInt(s.product_count || '0', 10),
brands: parseInt(s.brand_count || '0', 10),
dispensaries: parseInt(s.dispensary_count || '0', 10)
}
});
} catch (error: any) {
console.error('Public API stats error:', error);
res.status(500).json({
error: 'Failed to fetch stats',
message: error.message
});
}
});
/**
* GET /api/v1/menu
* Get complete menu summary for the authenticated dispensary

565
backend/src/routes/tasks.ts Normal file
View File

@@ -0,0 +1,565 @@
/**
* Task Queue API Routes
*
* Endpoints for managing worker tasks, viewing capacity metrics,
* and generating batch tasks.
*/
import { Router, Request, Response } from 'express';
import {
taskService,
TaskRole,
TaskStatus,
TaskFilter,
} from '../tasks/task-service';
import { pool } from '../db/pool';
const router = Router();
/**
* GET /api/tasks
* List tasks with optional filters
*
* Query params:
* - role: Filter by role
* - status: Filter by status (comma-separated for multiple)
* - dispensary_id: Filter by dispensary
* - worker_id: Filter by worker
* - limit: Max results (default 100)
* - offset: Pagination offset
*/
router.get('/', async (req: Request, res: Response) => {
try {
const filter: TaskFilter = {};
if (req.query.role) {
filter.role = req.query.role as TaskRole;
}
if (req.query.status) {
const statuses = (req.query.status as string).split(',') as TaskStatus[];
filter.status = statuses.length === 1 ? statuses[0] : statuses;
}
if (req.query.dispensary_id) {
filter.dispensary_id = parseInt(req.query.dispensary_id as string, 10);
}
if (req.query.worker_id) {
filter.worker_id = req.query.worker_id as string;
}
if (req.query.limit) {
filter.limit = parseInt(req.query.limit as string, 10);
}
if (req.query.offset) {
filter.offset = parseInt(req.query.offset as string, 10);
}
const tasks = await taskService.listTasks(filter);
res.json({ tasks, count: tasks.length });
} catch (error: unknown) {
console.error('Error listing tasks:', error);
res.status(500).json({ error: 'Failed to list tasks' });
}
});
/**
* GET /api/tasks/counts
* Get task counts by status
*/
router.get('/counts', async (_req: Request, res: Response) => {
try {
const counts = await taskService.getTaskCounts();
res.json(counts);
} catch (error: unknown) {
console.error('Error getting task counts:', error);
res.status(500).json({ error: 'Failed to get task counts' });
}
});
/**
* GET /api/tasks/capacity
* Get capacity metrics for all roles
*/
router.get('/capacity', async (_req: Request, res: Response) => {
try {
const metrics = await taskService.getCapacityMetrics();
res.json({ metrics });
} catch (error: unknown) {
console.error('Error getting capacity metrics:', error);
res.status(500).json({ error: 'Failed to get capacity metrics' });
}
});
/**
* GET /api/tasks/capacity/:role
* Get capacity metrics for a specific role
*/
router.get('/capacity/:role', async (req: Request, res: Response) => {
try {
const role = req.params.role as TaskRole;
const capacity = await taskService.getRoleCapacity(role);
if (!capacity) {
return res.status(404).json({ error: 'Role not found or no data' });
}
// Calculate workers needed for different SLAs
const workersFor1Hour = await taskService.calculateWorkersNeeded(role, 1);
const workersFor4Hours = await taskService.calculateWorkersNeeded(role, 4);
const workersFor8Hours = await taskService.calculateWorkersNeeded(role, 8);
res.json({
...capacity,
workers_needed: {
for_1_hour: workersFor1Hour,
for_4_hours: workersFor4Hours,
for_8_hours: workersFor8Hours,
},
});
} catch (error: unknown) {
console.error('Error getting role capacity:', error);
res.status(500).json({ error: 'Failed to get role capacity' });
}
});
/**
* GET /api/tasks/:id
* Get a specific task by ID
*/
router.get('/:id', async (req: Request, res: Response) => {
try {
const taskId = parseInt(req.params.id, 10);
const task = await taskService.getTask(taskId);
if (!task) {
return res.status(404).json({ error: 'Task not found' });
}
res.json(task);
} catch (error: unknown) {
console.error('Error getting task:', error);
res.status(500).json({ error: 'Failed to get task' });
}
});
/**
* POST /api/tasks
* Create a new task
*
* Body:
* - role: TaskRole (required)
* - dispensary_id: number (optional)
* - platform: string (optional)
* - priority: number (optional, default 0)
* - scheduled_for: ISO date string (optional)
*/
router.post('/', async (req: Request, res: Response) => {
try {
const { role, dispensary_id, platform, priority, scheduled_for } = req.body;
if (!role) {
return res.status(400).json({ error: 'Role is required' });
}
// Check if store already has an active task
if (dispensary_id) {
const hasActive = await taskService.hasActiveTask(dispensary_id);
if (hasActive) {
return res.status(409).json({
error: 'Store already has an active task',
dispensary_id,
});
}
}
const task = await taskService.createTask({
role,
dispensary_id,
platform,
priority,
scheduled_for: scheduled_for ? new Date(scheduled_for) : undefined,
});
res.status(201).json(task);
} catch (error: unknown) {
console.error('Error creating task:', error);
res.status(500).json({ error: 'Failed to create task' });
}
});
/**
* POST /api/tasks/generate/resync
* Generate daily resync tasks for all active stores
*
* Body:
* - batches_per_day: number (optional, default 6 = every 4 hours)
* - date: ISO date string (optional, default today)
*/
router.post('/generate/resync', async (req: Request, res: Response) => {
try {
const { batches_per_day, date } = req.body;
const batchesPerDay = batches_per_day ?? 6;
const targetDate = date ? new Date(date) : new Date();
const createdCount = await taskService.generateDailyResyncTasks(
batchesPerDay,
targetDate
);
res.json({
success: true,
tasks_created: createdCount,
batches_per_day: batchesPerDay,
date: targetDate.toISOString().split('T')[0],
});
} catch (error: unknown) {
console.error('Error generating resync tasks:', error);
res.status(500).json({ error: 'Failed to generate resync tasks' });
}
});
/**
* POST /api/tasks/generate/discovery
* Generate store discovery tasks for a platform
*
* Body:
* - platform: string (required, e.g., 'dutchie')
* - state_code: string (optional, e.g., 'AZ')
* - priority: number (optional)
*/
router.post('/generate/discovery', async (req: Request, res: Response) => {
try {
const { platform, state_code, priority } = req.body;
if (!platform) {
return res.status(400).json({ error: 'Platform is required' });
}
const task = await taskService.createStoreDiscoveryTask(
platform,
state_code,
priority ?? 0
);
res.status(201).json(task);
} catch (error: unknown) {
console.error('Error creating discovery task:', error);
res.status(500).json({ error: 'Failed to create discovery task' });
}
});
/**
* POST /api/tasks/recover-stale
* Recover stale tasks from dead workers
*
* Body:
* - threshold_minutes: number (optional, default 10)
*/
router.post('/recover-stale', async (req: Request, res: Response) => {
try {
const { threshold_minutes } = req.body;
const recovered = await taskService.recoverStaleTasks(threshold_minutes ?? 10);
res.json({
success: true,
tasks_recovered: recovered,
});
} catch (error: unknown) {
console.error('Error recovering stale tasks:', error);
res.status(500).json({ error: 'Failed to recover stale tasks' });
}
});
/**
* GET /api/tasks/role/:role/last-completion
* Get the last completion time for a role
*/
router.get('/role/:role/last-completion', async (req: Request, res: Response) => {
try {
const role = req.params.role as TaskRole;
const lastCompletion = await taskService.getLastCompletion(role);
res.json({
role,
last_completion: lastCompletion?.toISOString() ?? null,
time_since: lastCompletion
? Math.floor((Date.now() - lastCompletion.getTime()) / 1000)
: null,
});
} catch (error: unknown) {
console.error('Error getting last completion:', error);
res.status(500).json({ error: 'Failed to get last completion' });
}
});
/**
* GET /api/tasks/role/:role/recent
* Get recent completions for a role
*/
router.get('/role/:role/recent', async (req: Request, res: Response) => {
try {
const role = req.params.role as TaskRole;
const limit = parseInt(req.query.limit as string, 10) || 10;
const tasks = await taskService.getRecentCompletions(role, limit);
res.json({ tasks });
} catch (error: unknown) {
console.error('Error getting recent completions:', error);
res.status(500).json({ error: 'Failed to get recent completions' });
}
});
/**
* GET /api/tasks/store/:dispensaryId/active
* Check if a store has an active task
*/
router.get('/store/:dispensaryId/active', async (req: Request, res: Response) => {
try {
const dispensaryId = parseInt(req.params.dispensaryId, 10);
const hasActive = await taskService.hasActiveTask(dispensaryId);
res.json({
dispensary_id: dispensaryId,
has_active_task: hasActive,
});
} catch (error: unknown) {
console.error('Error checking active task:', error);
res.status(500).json({ error: 'Failed to check active task' });
}
});
// ============================================================
// MIGRATION ROUTES - Disable old job systems
// ============================================================
/**
* GET /api/tasks/migration/status
* Get status of old job systems vs new task queue
*/
router.get('/migration/status', async (_req: Request, res: Response) => {
try {
// Get old job system counts
const [schedules, crawlJobs, rawPayloads, taskCounts] = await Promise.all([
pool.query(`
SELECT
COUNT(*) as total,
COUNT(*) FILTER (WHERE enabled = true) as enabled
FROM job_schedules
`),
pool.query(`
SELECT
COUNT(*) as total,
COUNT(*) FILTER (WHERE status = 'pending') as pending,
COUNT(*) FILTER (WHERE status = 'running') as running
FROM dispensary_crawl_jobs
`),
pool.query(`
SELECT
COUNT(*) as total,
COUNT(*) FILTER (WHERE processed = false) as unprocessed
FROM raw_payloads
`),
taskService.getTaskCounts(),
]);
res.json({
old_systems: {
job_schedules: {
total: parseInt(schedules.rows[0].total) || 0,
enabled: parseInt(schedules.rows[0].enabled) || 0,
},
dispensary_crawl_jobs: {
total: parseInt(crawlJobs.rows[0].total) || 0,
pending: parseInt(crawlJobs.rows[0].pending) || 0,
running: parseInt(crawlJobs.rows[0].running) || 0,
},
raw_payloads: {
total: parseInt(rawPayloads.rows[0].total) || 0,
unprocessed: parseInt(rawPayloads.rows[0].unprocessed) || 0,
},
},
new_task_queue: taskCounts,
recommendation: schedules.rows[0].enabled > 0
? 'Disable old job schedules before switching to new task queue'
: 'Ready to use new task queue',
});
} catch (error: unknown) {
console.error('Error getting migration status:', error);
res.status(500).json({ error: 'Failed to get migration status' });
}
});
/**
* POST /api/tasks/migration/disable-old-schedules
* Disable all old job schedules to prepare for new task queue
*/
router.post('/migration/disable-old-schedules', async (_req: Request, res: Response) => {
try {
const result = await pool.query(`
UPDATE job_schedules
SET enabled = false,
updated_at = NOW()
WHERE enabled = true
RETURNING id, job_name
`);
res.json({
success: true,
disabled_count: result.rowCount,
disabled_schedules: result.rows.map(r => ({ id: r.id, job_name: r.job_name })),
});
} catch (error: unknown) {
console.error('Error disabling old schedules:', error);
res.status(500).json({ error: 'Failed to disable old schedules' });
}
});
/**
* POST /api/tasks/migration/cancel-pending-crawl-jobs
* Cancel all pending crawl jobs from the old system
*/
router.post('/migration/cancel-pending-crawl-jobs', async (_req: Request, res: Response) => {
try {
const result = await pool.query(`
UPDATE dispensary_crawl_jobs
SET status = 'cancelled',
completed_at = NOW(),
updated_at = NOW()
WHERE status = 'pending'
RETURNING id
`);
res.json({
success: true,
cancelled_count: result.rowCount,
});
} catch (error: unknown) {
console.error('Error cancelling pending crawl jobs:', error);
res.status(500).json({ error: 'Failed to cancel pending crawl jobs' });
}
});
/**
* POST /api/tasks/migration/create-resync-tasks
* Create product_resync tasks for all crawl-enabled dispensaries
*/
router.post('/migration/create-resync-tasks', async (req: Request, res: Response) => {
try {
const { priority = 0, state_code } = req.body;
let query = `
SELECT id, name FROM dispensaries
WHERE crawl_enabled = true
AND platform_dispensary_id IS NOT NULL
`;
const params: any[] = [];
if (state_code) {
query += `
AND state_id = (SELECT id FROM states WHERE code = $1)
`;
params.push(state_code.toUpperCase());
}
query += ` ORDER BY id`;
const dispensaries = await pool.query(query, params);
let created = 0;
for (const disp of dispensaries.rows) {
// Check if already has pending/running task
const hasActive = await taskService.hasActiveTask(disp.id);
if (!hasActive) {
await taskService.createTask({
role: 'product_resync',
dispensary_id: disp.id,
platform: 'dutchie',
priority,
});
created++;
}
}
res.json({
success: true,
tasks_created: created,
dispensaries_checked: dispensaries.rows.length,
state_filter: state_code || 'all',
});
} catch (error: unknown) {
console.error('Error creating resync tasks:', error);
res.status(500).json({ error: 'Failed to create resync tasks' });
}
});
/**
* POST /api/tasks/migration/full-migrate
* One-click migration: disable old systems, create new tasks
*/
router.post('/migration/full-migrate', async (req: Request, res: Response) => {
try {
const results: any = {
success: true,
steps: [],
};
// Step 1: Disable old job schedules
const disableResult = await pool.query(`
UPDATE job_schedules
SET enabled = false, updated_at = NOW()
WHERE enabled = true
RETURNING id
`);
results.steps.push({
step: 'disable_job_schedules',
count: disableResult.rowCount,
});
// Step 2: Cancel pending crawl jobs
const cancelResult = await pool.query(`
UPDATE dispensary_crawl_jobs
SET status = 'cancelled', completed_at = NOW(), updated_at = NOW()
WHERE status = 'pending'
RETURNING id
`);
results.steps.push({
step: 'cancel_pending_crawl_jobs',
count: cancelResult.rowCount,
});
// Step 3: Generate initial resync tasks
const resyncCount = await taskService.generateDailyResyncTasks(6);
results.steps.push({
step: 'generate_resync_tasks',
count: resyncCount,
});
// Step 4: Create store discovery task
const discoveryTask = await taskService.createStoreDiscoveryTask('dutchie', undefined, 0);
results.steps.push({
step: 'create_discovery_task',
task_id: discoveryTask.id,
});
// Step 5: Create analytics refresh task
const analyticsTask = await taskService.createTask({
role: 'analytics_refresh',
priority: 0,
});
results.steps.push({
step: 'create_analytics_task',
task_id: analyticsTask.id,
});
results.message = 'Migration complete. New task workers will pick up tasks.';
res.json(results);
} catch (error: unknown) {
console.error('Error during full migration:', error);
res.status(500).json({ error: 'Failed to complete migration' });
}
});
export default router;

View File

@@ -1,18 +1,32 @@
import { Router, Request, Response } from 'express';
import { readFileSync } from 'fs';
import { join } from 'path';
const router = Router();
// Read package.json version at startup
let packageVersion = 'unknown';
try {
const packageJson = JSON.parse(readFileSync(join(__dirname, '../../package.json'), 'utf-8'));
packageVersion = packageJson.version || 'unknown';
} catch {
// Fallback if package.json not found
}
/**
* GET /api/version
* Returns build version information for display in admin UI
*/
router.get('/', async (req: Request, res: Response) => {
try {
const gitSha = process.env.APP_GIT_SHA || 'unknown';
const versionInfo = {
build_version: process.env.APP_BUILD_VERSION || 'dev',
git_sha: process.env.APP_GIT_SHA || 'local',
build_time: process.env.APP_BUILD_TIME || new Date().toISOString(),
image_tag: process.env.CONTAINER_IMAGE_TAG || 'local',
version: packageVersion,
build_version: process.env.APP_BUILD_VERSION?.slice(0, 8) || 'dev',
git_sha: gitSha.slice(0, 8) || 'unknown',
git_sha_full: gitSha,
build_time: process.env.APP_BUILD_TIME || 'unknown',
image_tag: process.env.CONTAINER_IMAGE_TAG?.slice(0, 8) || 'local',
};
res.json(versionInfo);

View File

@@ -0,0 +1,250 @@
#!/usr/bin/env npx tsx
/**
* Crawl Single Store - Verbose test showing each step
*
* Usage:
* DATABASE_URL="postgresql://dutchie:dutchie_local_pass@localhost:54320/dutchie_menus" \
* npx tsx src/scripts/crawl-single-store.ts <dispensaryId>
*
* Example:
* DATABASE_URL="..." npx tsx src/scripts/crawl-single-store.ts 112
*/
import { Pool } from 'pg';
import dotenv from 'dotenv';
import {
executeGraphQL,
startSession,
endSession,
getFingerprint,
GRAPHQL_HASHES,
DUTCHIE_CONFIG,
} from '../platforms/dutchie';
dotenv.config();
// ============================================================
// DATABASE CONNECTION
// ============================================================
function getConnectionString(): string {
if (process.env.DATABASE_URL) {
return process.env.DATABASE_URL;
}
if (process.env.CANNAIQ_DB_URL) {
return process.env.CANNAIQ_DB_URL;
}
const host = process.env.CANNAIQ_DB_HOST || 'localhost';
const port = process.env.CANNAIQ_DB_PORT || '54320';
const name = process.env.CANNAIQ_DB_NAME || 'dutchie_menus';
const user = process.env.CANNAIQ_DB_USER || 'dutchie';
const pass = process.env.CANNAIQ_DB_PASS || 'dutchie_local_pass';
return `postgresql://${user}:${pass}@${host}:${port}/${name}`;
}
const pool = new Pool({ connectionString: getConnectionString() });
// ============================================================
// MAIN
// ============================================================
async function main() {
const dispensaryId = parseInt(process.argv[2], 10);
if (!dispensaryId) {
console.error('Usage: npx tsx src/scripts/crawl-single-store.ts <dispensaryId>');
console.error('Example: npx tsx src/scripts/crawl-single-store.ts 112');
process.exit(1);
}
console.log('');
console.log('╔════════════════════════════════════════════════════════════╗');
console.log('║ SINGLE STORE CRAWL - VERBOSE OUTPUT ║');
console.log('╚════════════════════════════════════════════════════════════╝');
console.log('');
try {
// ============================================================
// STEP 1: Get dispensary info from database
// ============================================================
console.log('┌─────────────────────────────────────────────────────────────┐');
console.log('│ STEP 1: Load Dispensary Info from Database │');
console.log('└─────────────────────────────────────────────────────────────┘');
const dispResult = await pool.query(`
SELECT
id,
name,
platform_dispensary_id,
menu_url,
menu_type,
city,
state
FROM dispensaries
WHERE id = $1
`, [dispensaryId]);
if (dispResult.rows.length === 0) {
throw new Error(`Dispensary ${dispensaryId} not found`);
}
const disp = dispResult.rows[0];
console.log(` Dispensary ID: ${disp.id}`);
console.log(` Name: ${disp.name}`);
console.log(` City, State: ${disp.city}, ${disp.state}`);
console.log(` Menu Type: ${disp.menu_type}`);
console.log(` Platform ID: ${disp.platform_dispensary_id}`);
console.log(` Menu URL: ${disp.menu_url}`);
if (!disp.platform_dispensary_id) {
throw new Error('Dispensary does not have a platform_dispensary_id - cannot crawl');
}
// Extract cName from menu_url
const cNameMatch = disp.menu_url?.match(/\/(?:embedded-menu|dispensary)\/([^/?]+)/);
const cName = cNameMatch ? cNameMatch[1] : 'dispensary';
console.log(` cName (derived): ${cName}`);
console.log('');
// ============================================================
// STEP 2: Start stealth session
// ============================================================
console.log('┌─────────────────────────────────────────────────────────────┐');
console.log('│ STEP 2: Start Stealth Session │');
console.log('└─────────────────────────────────────────────────────────────┘');
// Use Arizona timezone for this store
const session = startSession(disp.state || 'AZ', 'America/Phoenix');
const fp = getFingerprint();
console.log(` Session ID: ${session.sessionId}`);
console.log(` User-Agent: ${fp.userAgent.slice(0, 60)}...`);
console.log(` Accept-Language: ${fp.acceptLanguage}`);
console.log(` Sec-CH-UA: ${fp.secChUa || '(not set)'}`);
console.log('');
// ============================================================
// STEP 3: Execute GraphQL query
// ============================================================
console.log('┌─────────────────────────────────────────────────────────────┐');
console.log('│ STEP 3: Execute GraphQL Query (FilteredProducts) │');
console.log('└─────────────────────────────────────────────────────────────┘');
const variables = {
includeEnterpriseSpecials: false,
productsFilter: {
dispensaryId: disp.platform_dispensary_id,
pricingType: 'rec',
Status: 'Active',
types: [],
useCache: true,
isDefaultSort: true,
sortBy: 'popularSortIdx',
sortDirection: 1,
bypassOnlineThresholds: true,
isKioskMenu: false,
removeProductsBelowOptionThresholds: false,
},
page: 0,
perPage: 100,
};
console.log(` Endpoint: ${DUTCHIE_CONFIG.graphqlEndpoint}`);
console.log(` Operation: FilteredProducts`);
console.log(` Hash: ${GRAPHQL_HASHES.FilteredProducts.slice(0, 20)}...`);
console.log(` dispensaryId: ${variables.productsFilter.dispensaryId}`);
console.log(` pricingType: ${variables.productsFilter.pricingType}`);
console.log(` Status: ${variables.productsFilter.Status}`);
console.log(` perPage: ${variables.perPage}`);
console.log('');
console.log(' Sending request...');
const startTime = Date.now();
const result = await executeGraphQL(
'FilteredProducts',
variables,
GRAPHQL_HASHES.FilteredProducts,
{ cName, maxRetries: 3 }
);
const elapsed = Date.now() - startTime;
console.log(` Response time: ${elapsed}ms`);
console.log('');
// ============================================================
// STEP 4: Process response
// ============================================================
console.log('┌─────────────────────────────────────────────────────────────┐');
console.log('│ STEP 4: Process Response │');
console.log('└─────────────────────────────────────────────────────────────┘');
const data = result?.data?.filteredProducts;
if (!data) {
console.log(' ERROR: No data returned from GraphQL');
console.log(' Raw result:', JSON.stringify(result, null, 2).slice(0, 500));
endSession();
return;
}
const products = data.products || [];
const totalCount = data.queryInfo?.totalCount || 0;
const totalPages = Math.ceil(totalCount / 100);
console.log(` Total products: ${totalCount}`);
console.log(` Products in page: ${products.length}`);
console.log(` Total pages: ${totalPages}`);
console.log('');
// Show first few products
console.log(' First 5 products:');
console.log(' ─────────────────────────────────────────────────────────');
for (let i = 0; i < Math.min(5, products.length); i++) {
const p = products[i];
const name = (p.name || 'Unknown').slice(0, 40);
const brand = (p.brand?.name || 'Unknown').slice(0, 15);
const price = p.Prices?.[0]?.price || p.medPrice || p.recPrice || 'N/A';
const category = p.type || p.category || 'N/A';
console.log(` ${i + 1}. ${name.padEnd(42)} | ${brand.padEnd(17)} | $${price}`);
}
console.log('');
// ============================================================
// STEP 5: End session
// ============================================================
console.log('┌─────────────────────────────────────────────────────────────┐');
console.log('│ STEP 5: End Session │');
console.log('└─────────────────────────────────────────────────────────────┘');
endSession();
console.log('');
// ============================================================
// SUMMARY
// ============================================================
console.log('╔════════════════════════════════════════════════════════════╗');
console.log('║ SUMMARY ║');
console.log('╠════════════════════════════════════════════════════════════╣');
console.log(`║ Store: ${disp.name.slice(0, 38).padEnd(38)}`);
console.log(`║ Products Found: ${String(totalCount).padEnd(38)}`);
console.log(`║ Response Time: ${(elapsed + 'ms').padEnd(38)}`);
console.log(`║ Status: ${'SUCCESS'.padEnd(38)}`);
console.log('╚════════════════════════════════════════════════════════════╝');
} catch (error: any) {
console.error('');
console.error('╔════════════════════════════════════════════════════════════╗');
console.error('║ ERROR ║');
console.error('╚════════════════════════════════════════════════════════════╝');
console.error(` ${error.message}`);
if (error.stack) {
console.error('');
console.error('Stack trace:');
console.error(error.stack.split('\n').slice(0, 5).join('\n'));
}
process.exit(1);
} finally {
await pool.end();
}
}
main();

View File

@@ -23,6 +23,7 @@ import {
DutchieNormalizer,
hydrateToCanonical,
} from '../hydration';
import { initializeImageStorage } from '../utils/image-storage';
dotenv.config();
@@ -137,6 +138,11 @@ async function main() {
console.log(`Test Crawl to Canonical - Dispensary ${dispensaryId}`);
console.log('============================================================\n');
// Initialize image storage
console.log('[Init] Initializing image storage...');
await initializeImageStorage();
console.log(' Image storage ready\n');
try {
// Step 1: Get dispensary info
console.log('[Step 1] Getting dispensary info...');

View File

@@ -0,0 +1,80 @@
#!/usr/bin/env npx tsx
/**
* Test Image Proxy - Standalone test without backend
*
* Usage:
* npx tsx src/scripts/test-image-proxy.ts
*/
import express from 'express';
import imageProxyRoutes from '../routes/image-proxy';
const app = express();
const PORT = 3099;
// Mount the image proxy
app.use('/img', imageProxyRoutes);
// Start server
app.listen(PORT, async () => {
console.log(`Test image proxy running on http://localhost:${PORT}`);
console.log('');
console.log('Testing image proxy...');
console.log('');
const axios = require('axios');
// Test cases
const tests = [
{
name: 'Original image',
url: '/img/products/az/az-deeply-rooted/clout-king/68b4b20a0f9ef3e90eb51e96/image-268a6e44.webp',
},
{
name: 'Resize to 200px width',
url: '/img/products/az/az-deeply-rooted/clout-king/68b4b20a0f9ef3e90eb51e96/image-268a6e44.webp?w=200',
},
{
name: 'Resize to 100x100 cover',
url: '/img/products/az/az-deeply-rooted/clout-king/68b4b20a0f9ef3e90eb51e96/image-268a6e44.webp?w=100&h=100&fit=cover',
},
{
name: 'Grayscale + blur',
url: '/img/products/az/az-deeply-rooted/clout-king/68b4b20a0f9ef3e90eb51e96/image-268a6e44.webp?w=200&gray=1&blur=2',
},
{
name: 'Convert to JPEG',
url: '/img/products/az/az-deeply-rooted/clout-king/68b4b20a0f9ef3e90eb51e96/image-268a6e44.webp?w=200&format=jpeg&q=70',
},
{
name: 'Non-existent image',
url: '/img/products/az/nonexistent/image.webp',
},
];
for (const test of tests) {
try {
const response = await axios.get(`http://localhost:${PORT}${test.url}`, {
responseType: 'arraybuffer',
validateStatus: () => true,
});
const contentType = response.headers['content-type'];
const size = response.data.length;
const status = response.status;
console.log(`${test.name}:`);
console.log(` URL: ${test.url.slice(0, 80)}${test.url.length > 80 ? '...' : ''}`);
console.log(` Status: ${status}`);
console.log(` Content-Type: ${contentType}`);
console.log(` Size: ${(size / 1024).toFixed(1)} KB`);
console.log('');
} catch (error: any) {
console.log(`${test.name}: ERROR - ${error.message}`);
console.log('');
}
}
console.log('Tests complete!');
process.exit(0);
});

View File

@@ -0,0 +1,117 @@
/**
* Test script for stealth session management
*
* Tests:
* 1. Per-session fingerprint rotation
* 2. Geographic consistency (timezone → Accept-Language)
* 3. Proxy location loading from database
*
* Usage:
* npx tsx src/scripts/test-stealth-session.ts
*/
import {
startSession,
endSession,
getCurrentSession,
getFingerprint,
getRandomFingerprint,
getLocaleForTimezone,
buildHeaders,
} from '../platforms/dutchie';
console.log('='.repeat(60));
console.log('STEALTH SESSION TEST');
console.log('='.repeat(60));
// Test 1: Timezone to Locale mapping
console.log('\n[Test 1] Timezone to Locale Mapping:');
const testTimezones = [
'America/Phoenix',
'America/Los_Angeles',
'America/New_York',
'America/Chicago',
undefined,
'Invalid/Timezone',
];
for (const tz of testTimezones) {
const locale = getLocaleForTimezone(tz);
console.log(` ${tz || '(undefined)'}${locale}`);
}
// Test 2: Random fingerprint selection
console.log('\n[Test 2] Random Fingerprint Selection (5 samples):');
for (let i = 0; i < 5; i++) {
const fp = getRandomFingerprint();
console.log(` ${i + 1}. ${fp.userAgent.slice(0, 60)}...`);
}
// Test 3: Session Management
console.log('\n[Test 3] Session Management:');
// Before session - should use default fingerprint
console.log(' Before session:');
const beforeFp = getFingerprint();
console.log(` getFingerprint(): ${beforeFp.userAgent.slice(0, 50)}...`);
console.log(` getCurrentSession(): ${getCurrentSession()}`);
// Start session with Arizona timezone
console.log('\n Starting session (AZ, America/Phoenix):');
const session1 = startSession('AZ', 'America/Phoenix');
console.log(` Session ID: ${session1.sessionId}`);
console.log(` Fingerprint UA: ${session1.fingerprint.userAgent.slice(0, 50)}...`);
console.log(` Accept-Language: ${session1.fingerprint.acceptLanguage}`);
console.log(` Timezone: ${session1.timezone}`);
// During session - should use session fingerprint
console.log('\n During session:');
const duringFp = getFingerprint();
console.log(` getFingerprint(): ${duringFp.userAgent.slice(0, 50)}...`);
console.log(` Same as session? ${duringFp.userAgent === session1.fingerprint.userAgent}`);
// Test buildHeaders with session
console.log('\n buildHeaders() during session:');
const headers = buildHeaders('/embedded-menu/test-store');
console.log(` User-Agent: ${headers['user-agent'].slice(0, 50)}...`);
console.log(` Accept-Language: ${headers['accept-language']}`);
console.log(` Origin: ${headers['origin']}`);
console.log(` Referer: ${headers['referer']}`);
// End session
console.log('\n Ending session:');
endSession();
console.log(` getCurrentSession(): ${getCurrentSession()}`);
// Test 4: Multiple sessions should have different fingerprints
console.log('\n[Test 4] Multiple Sessions (fingerprint variety):');
const fingerprints: string[] = [];
for (let i = 0; i < 10; i++) {
const session = startSession('CA', 'America/Los_Angeles');
fingerprints.push(session.fingerprint.userAgent);
endSession();
}
const uniqueCount = new Set(fingerprints).size;
console.log(` 10 sessions created, ${uniqueCount} unique fingerprints`);
console.log(` Variety: ${uniqueCount >= 3 ? '✅ Good' : '⚠️ Low - may need more fingerprint options'}`);
// Test 5: Geographic consistency check
console.log('\n[Test 5] Geographic Consistency:');
const geoTests = [
{ state: 'AZ', tz: 'America/Phoenix' },
{ state: 'CA', tz: 'America/Los_Angeles' },
{ state: 'NY', tz: 'America/New_York' },
{ state: 'IL', tz: 'America/Chicago' },
];
for (const { state, tz } of geoTests) {
const session = startSession(state, tz);
const consistent = session.fingerprint.acceptLanguage.includes('en-US');
console.log(` ${state} (${tz}): Accept-Language=${session.fingerprint.acceptLanguage} ${consistent ? '✅' : '❌'}`);
endSession();
}
console.log('\n' + '='.repeat(60));
console.log('TEST COMPLETE');
console.log('='.repeat(60));

View File

@@ -0,0 +1,92 @@
/**
* Analytics Refresh Handler
*
* Refreshes materialized views and pre-computed analytics tables.
* Should run daily or on-demand after major data changes.
*/
import { TaskContext, TaskResult } from '../task-worker';
export async function handleAnalyticsRefresh(ctx: TaskContext): Promise<TaskResult> {
const { pool } = ctx;
console.log(`[AnalyticsRefresh] Starting analytics refresh...`);
const refreshed: string[] = [];
const failed: string[] = [];
// List of materialized views to refresh
const materializedViews = [
'mv_state_metrics',
'mv_brand_metrics',
'mv_category_metrics',
'v_brand_summary',
'v_dashboard_stats',
];
for (const viewName of materializedViews) {
try {
// Heartbeat before each refresh
await ctx.heartbeat();
// Check if view exists
const existsResult = await pool.query(`
SELECT EXISTS (
SELECT 1 FROM pg_matviews WHERE matviewname = $1
UNION
SELECT 1 FROM pg_views WHERE viewname = $1
) as exists
`, [viewName]);
if (!existsResult.rows[0].exists) {
console.log(`[AnalyticsRefresh] View ${viewName} does not exist, skipping`);
continue;
}
// Try to refresh (only works for materialized views)
try {
await pool.query(`REFRESH MATERIALIZED VIEW CONCURRENTLY ${viewName}`);
refreshed.push(viewName);
console.log(`[AnalyticsRefresh] Refreshed ${viewName}`);
} catch (refreshError: any) {
// Try non-concurrent refresh
try {
await pool.query(`REFRESH MATERIALIZED VIEW ${viewName}`);
refreshed.push(viewName);
console.log(`[AnalyticsRefresh] Refreshed ${viewName} (non-concurrent)`);
} catch (nonConcurrentError: any) {
// Not a materialized view or other error
console.log(`[AnalyticsRefresh] ${viewName} is not a materialized view or refresh failed`);
}
}
} catch (error: any) {
console.error(`[AnalyticsRefresh] Error refreshing ${viewName}:`, error.message);
failed.push(viewName);
}
}
// Run analytics capture functions if they exist
const captureFunctions = [
'capture_brand_snapshots',
'capture_category_snapshots',
];
for (const funcName of captureFunctions) {
try {
await pool.query(`SELECT ${funcName}()`);
console.log(`[AnalyticsRefresh] Executed ${funcName}()`);
} catch (error: any) {
// Function might not exist
console.log(`[AnalyticsRefresh] ${funcName}() not available`);
}
}
console.log(`[AnalyticsRefresh] Complete: ${refreshed.length} refreshed, ${failed.length} failed`);
return {
success: failed.length === 0,
refreshed,
failed,
error: failed.length > 0 ? `Failed to refresh: ${failed.join(', ')}` : undefined,
};
}

View File

@@ -0,0 +1,87 @@
/**
* Entry Point Discovery Handler
*
* Detects menu type and resolves platform IDs for a discovered store.
* This is the step between store_discovery and product_discovery.
*
* TODO: Integrate with platform ID resolution when available
*/
import { TaskContext, TaskResult } from '../task-worker';
export async function handleEntryPointDiscovery(ctx: TaskContext): Promise<TaskResult> {
const { pool, task } = ctx;
const dispensaryId = task.dispensary_id;
if (!dispensaryId) {
return { success: false, error: 'No dispensary_id specified for entry_point_discovery task' };
}
try {
// Get dispensary info
const dispResult = await pool.query(`
SELECT id, name, menu_url, platform_dispensary_id, menu_type
FROM dispensaries
WHERE id = $1
`, [dispensaryId]);
if (dispResult.rows.length === 0) {
return { success: false, error: `Dispensary ${dispensaryId} not found` };
}
const dispensary = dispResult.rows[0];
// If already has platform_dispensary_id, we're done
if (dispensary.platform_dispensary_id) {
console.log(`[EntryPointDiscovery] Dispensary ${dispensaryId} already has platform ID`);
return {
success: true,
alreadyResolved: true,
platformId: dispensary.platform_dispensary_id,
};
}
const menuUrl = dispensary.menu_url;
if (!menuUrl) {
return { success: false, error: `Dispensary ${dispensaryId} has no menu_url` };
}
console.log(`[EntryPointDiscovery] Would resolve platform ID for ${dispensary.name} from ${menuUrl}`);
// Extract slug from menu URL
let slug: string | null = null;
const embeddedMatch = menuUrl.match(/\/embedded-menu\/([^/?]+)/);
const dispensaryMatch = menuUrl.match(/\/dispensary\/([^/?]+)/);
if (embeddedMatch) {
slug = embeddedMatch[1];
} else if (dispensaryMatch) {
slug = dispensaryMatch[1];
}
if (!slug) {
return {
success: false,
error: `Could not extract slug from menu_url: ${menuUrl}`,
};
}
// TODO: Integrate with actual platform ID resolution
// For now, mark the task as needing manual resolution
console.log(`[EntryPointDiscovery] Found slug: ${slug} - manual resolution needed`);
return {
success: true,
message: 'Slug extracted, awaiting platform ID resolution',
slug,
};
} catch (error: unknown) {
const errorMessage = error instanceof Error ? error.message : 'Unknown error';
console.error(`[EntryPointDiscovery] Error for dispensary ${dispensaryId}:`, errorMessage);
return {
success: false,
error: errorMessage,
};
}
}

View File

@@ -0,0 +1,11 @@
/**
* Task Handlers Index
*
* Exports all task handlers for the task worker.
*/
export { handleProductResync } from './product-resync';
export { handleProductDiscovery } from './product-discovery';
export { handleStoreDiscovery } from './store-discovery';
export { handleEntryPointDiscovery } from './entry-point-discovery';
export { handleAnalyticsRefresh } from './analytics-refresh';

View File

@@ -0,0 +1,16 @@
/**
* Product Discovery Handler
*
* Initial product fetch for stores that have 0 products.
* Same logic as product_resync, but for initial discovery.
*/
import { TaskContext, TaskResult } from '../task-worker';
import { handleProductResync } from './product-resync';
export async function handleProductDiscovery(ctx: TaskContext): Promise<TaskResult> {
// Product discovery is essentially the same as resync for the first time
// The main difference is in when this task is triggered (new store vs scheduled)
console.log(`[ProductDiscovery] Starting initial product fetch for dispensary ${ctx.task.dispensary_id}`);
return handleProductResync(ctx);
}

View File

@@ -0,0 +1,344 @@
/**
* Product Resync Handler
*
* Re-crawls a store to capture price/stock changes using the GraphQL pipeline.
*
* Flow:
* 1. Load dispensary info from database
* 2. Start stealth session (fingerprint + optional proxy)
* 3. Fetch products via GraphQL (Status: 'All')
* 4. Normalize data via DutchieNormalizer
* 5. Upsert to store_products and store_product_snapshots
* 6. Track missing products (increment consecutive_misses, mark OOS at 3)
* 7. Download new product images
* 8. End session
*/
import { TaskContext, TaskResult } from '../task-worker';
import {
executeGraphQL,
startSession,
endSession,
GRAPHQL_HASHES,
DUTCHIE_CONFIG,
} from '../../platforms/dutchie';
import { DutchieNormalizer } from '../../hydration/normalizers/dutchie';
import {
upsertStoreProducts,
createStoreProductSnapshots,
downloadProductImages,
} from '../../hydration/canonical-upsert';
const normalizer = new DutchieNormalizer();
export async function handleProductResync(ctx: TaskContext): Promise<TaskResult> {
const { pool, task } = ctx;
const dispensaryId = task.dispensary_id;
if (!dispensaryId) {
return { success: false, error: 'No dispensary_id specified for product_resync task' };
}
try {
// ============================================================
// STEP 1: Load dispensary info
// ============================================================
const dispResult = await pool.query(`
SELECT
id, name, platform_dispensary_id, menu_url, menu_type, city, state
FROM dispensaries
WHERE id = $1 AND crawl_enabled = true
`, [dispensaryId]);
if (dispResult.rows.length === 0) {
return { success: false, error: `Dispensary ${dispensaryId} not found or not crawl_enabled` };
}
const dispensary = dispResult.rows[0];
const platformId = dispensary.platform_dispensary_id;
if (!platformId) {
return { success: false, error: `Dispensary ${dispensaryId} has no platform_dispensary_id` };
}
// Extract cName from menu_url
const cNameMatch = dispensary.menu_url?.match(/\/(?:embedded-menu|dispensary)\/([^/?]+)/);
const cName = cNameMatch ? cNameMatch[1] : 'dispensary';
console.log(`[ProductResync] Starting crawl for ${dispensary.name} (ID: ${dispensaryId})`);
console.log(`[ProductResync] Platform ID: ${platformId}, cName: ${cName}`);
// ============================================================
// STEP 2: Start stealth session
// ============================================================
const session = startSession(dispensary.state || 'AZ', 'America/Phoenix');
console.log(`[ProductResync] Session started: ${session.sessionId}`);
await ctx.heartbeat();
// ============================================================
// STEP 3: Fetch products via GraphQL (Status: 'All')
// ============================================================
const allProducts: any[] = [];
let page = 0;
let totalCount = 0;
const perPage = DUTCHIE_CONFIG.perPage;
const maxPages = DUTCHIE_CONFIG.maxPages;
try {
while (page < maxPages) {
const variables = {
includeEnterpriseSpecials: false,
productsFilter: {
dispensaryId: platformId,
pricingType: 'rec',
Status: 'All',
types: [],
useCache: false,
isDefaultSort: true,
sortBy: 'popularSortIdx',
sortDirection: 1,
bypassOnlineThresholds: true,
isKioskMenu: false,
removeProductsBelowOptionThresholds: false,
},
page,
perPage,
};
console.log(`[ProductResync] Fetching page ${page + 1}...`);
const result = await executeGraphQL(
'FilteredProducts',
variables,
GRAPHQL_HASHES.FilteredProducts,
{ cName, maxRetries: 3 }
);
const data = result?.data?.filteredProducts;
if (!data || !data.products) {
if (page === 0) {
throw new Error('No product data returned from GraphQL');
}
break;
}
const products = data.products;
allProducts.push(...products);
if (page === 0) {
totalCount = data.queryInfo?.totalCount || products.length;
console.log(`[ProductResync] Total products reported: ${totalCount}`);
}
if (allProducts.length >= totalCount || products.length < perPage) {
break;
}
page++;
if (page < maxPages) {
await new Promise(r => setTimeout(r, DUTCHIE_CONFIG.pageDelayMs));
}
if (page % 5 === 0) {
await ctx.heartbeat();
}
}
console.log(`[ProductResync] Fetched ${allProducts.length} products in ${page + 1} pages`);
} finally {
endSession();
}
if (allProducts.length === 0) {
return {
success: false,
error: 'No products returned from GraphQL',
productsProcessed: 0,
};
}
await ctx.heartbeat();
// ============================================================
// STEP 4: Normalize data
// ============================================================
console.log(`[ProductResync] Normalizing ${allProducts.length} products...`);
// Build RawPayload for the normalizer
const rawPayload = {
id: `resync-${dispensaryId}-${Date.now()}`,
dispensary_id: dispensaryId,
crawl_run_id: null,
platform: 'dutchie',
payload_version: 1,
raw_json: { data: { filteredProducts: { products: allProducts } } },
product_count: allProducts.length,
pricing_type: 'dual',
crawl_mode: 'dual_mode',
fetched_at: new Date(),
processed: false,
normalized_at: null,
hydration_error: null,
hydration_attempts: 0,
created_at: new Date(),
};
const normalizationResult = normalizer.normalize(rawPayload);
if (normalizationResult.errors.length > 0) {
console.warn(`[ProductResync] Normalization warnings: ${normalizationResult.errors.map(e => e.message).join(', ')}`);
}
if (normalizationResult.products.length === 0) {
return {
success: false,
error: 'Normalization produced no products',
productsProcessed: 0,
};
}
console.log(`[ProductResync] Normalized ${normalizationResult.products.length} products`);
await ctx.heartbeat();
// ============================================================
// STEP 5: Upsert to canonical tables
// ============================================================
console.log(`[ProductResync] Upserting to store_products...`);
const upsertResult = await upsertStoreProducts(
pool,
normalizationResult.products,
normalizationResult.pricing,
normalizationResult.availability
);
console.log(`[ProductResync] Upserted: ${upsertResult.upserted} (${upsertResult.new} new, ${upsertResult.updated} updated)`);
await ctx.heartbeat();
// Create snapshots
console.log(`[ProductResync] Creating snapshots...`);
const snapshotsResult = await createStoreProductSnapshots(
pool,
dispensaryId,
normalizationResult.products,
normalizationResult.pricing,
normalizationResult.availability,
null // No crawl_run_id in new system
);
console.log(`[ProductResync] Created ${snapshotsResult.created} snapshots`);
await ctx.heartbeat();
// ============================================================
// STEP 6: Track missing products (consecutive_misses logic)
// - Products in feed: reset consecutive_misses to 0
// - Products not in feed: increment consecutive_misses
// - At 3 consecutive misses: mark as OOS
// ============================================================
const currentProductIds = allProducts
.map((p: any) => p._id || p.id)
.filter(Boolean);
// Reset consecutive_misses for products that ARE in the feed
if (currentProductIds.length > 0) {
await pool.query(`
UPDATE store_products
SET consecutive_misses = 0, last_seen_at = NOW()
WHERE dispensary_id = $1
AND provider = 'dutchie'
AND provider_product_id = ANY($2)
`, [dispensaryId, currentProductIds]);
}
// Increment consecutive_misses for products NOT in the feed
const incrementResult = await pool.query(`
UPDATE store_products
SET consecutive_misses = consecutive_misses + 1
WHERE dispensary_id = $1
AND provider = 'dutchie'
AND provider_product_id NOT IN (SELECT unnest($2::text[]))
AND consecutive_misses < 3
RETURNING id
`, [dispensaryId, currentProductIds]);
const incrementedCount = incrementResult.rowCount || 0;
if (incrementedCount > 0) {
console.log(`[ProductResync] Incremented consecutive_misses for ${incrementedCount} products`);
}
// Mark as OOS any products that hit 3 consecutive misses
const oosResult = await pool.query(`
UPDATE store_products
SET stock_status = 'oos', is_in_stock = false
WHERE dispensary_id = $1
AND provider = 'dutchie'
AND consecutive_misses >= 3
AND stock_status != 'oos'
RETURNING id
`, [dispensaryId]);
const markedOosCount = oosResult.rowCount || 0;
if (markedOosCount > 0) {
console.log(`[ProductResync] Marked ${markedOosCount} products as OOS (3+ consecutive misses)`);
}
await ctx.heartbeat();
// ============================================================
// STEP 7: Download images for new products
// ============================================================
if (upsertResult.productsNeedingImages.length > 0) {
console.log(`[ProductResync] Downloading images for ${upsertResult.productsNeedingImages.length} products...`);
try {
const dispensaryContext = {
stateCode: dispensary.state || 'AZ',
storeSlug: cName,
};
await downloadProductImages(
pool,
upsertResult.productsNeedingImages,
dispensaryContext
);
} catch (imgError: any) {
// Image download errors shouldn't fail the whole task
console.warn(`[ProductResync] Image download error (non-fatal): ${imgError.message}`);
}
}
// ============================================================
// STEP 8: Update dispensary last_crawl_at
// ============================================================
await pool.query(`
UPDATE dispensaries
SET last_crawl_at = NOW()
WHERE id = $1
`, [dispensaryId]);
console.log(`[ProductResync] Completed ${dispensary.name}`);
return {
success: true,
productsProcessed: normalizationResult.products.length,
snapshotsCreated: snapshotsResult.created,
newProducts: upsertResult.new,
updatedProducts: upsertResult.updated,
markedOos: markedOosCount,
};
} catch (error: unknown) {
const errorMessage = error instanceof Error ? error.message : 'Unknown error';
console.error(`[ProductResync] Error for dispensary ${dispensaryId}:`, errorMessage);
return {
success: false,
error: errorMessage,
};
}
}

View File

@@ -0,0 +1,66 @@
/**
* Store Discovery Handler
*
* Discovers new stores by crawling location APIs and adding them
* to discovery_locations table.
*/
import { TaskContext, TaskResult } from '../task-worker';
import { discoverState } from '../../discovery';
export async function handleStoreDiscovery(ctx: TaskContext): Promise<TaskResult> {
const { pool, task } = ctx;
const platform = task.platform || 'default';
console.log(`[StoreDiscovery] Starting discovery for platform: ${platform}`);
try {
// Get states to discover
const statesResult = await pool.query(`
SELECT code FROM states WHERE active = true ORDER BY code
`);
const stateCodes = statesResult.rows.map(r => r.code);
if (stateCodes.length === 0) {
return { success: true, storesDiscovered: 0, message: 'No active states to discover' };
}
let totalDiscovered = 0;
let totalPromoted = 0;
// Run discovery for each state
for (const stateCode of stateCodes) {
// Heartbeat before each state
await ctx.heartbeat();
console.log(`[StoreDiscovery] Discovering stores in ${stateCode}...`);
try {
const result = await discoverState(pool, stateCode);
totalDiscovered += result.totalLocationsFound || 0;
totalPromoted += result.totalLocationsUpserted || 0;
console.log(`[StoreDiscovery] ${stateCode}: found ${result.totalLocationsFound}, upserted ${result.totalLocationsUpserted}`);
} catch (error: unknown) {
const errorMessage = error instanceof Error ? error.message : 'Unknown error';
console.error(`[StoreDiscovery] Error discovering ${stateCode}:`, errorMessage);
// Continue with other states
}
}
console.log(`[StoreDiscovery] Complete: ${totalDiscovered} discovered, ${totalPromoted} promoted`);
return {
success: true,
storesDiscovered: totalDiscovered,
storesPromoted: totalPromoted,
statesProcessed: stateCodes.length,
};
} catch (error: unknown) {
const errorMessage = error instanceof Error ? error.message : 'Unknown error';
console.error(`[StoreDiscovery] Error:`, errorMessage);
return {
success: false,
error: errorMessage,
};
}
}

View File

@@ -0,0 +1,25 @@
/**
* Task Queue Module
*
* Exports task service, worker, and types for use throughout the application.
*/
export {
taskService,
TaskRole,
TaskStatus,
WorkerTask,
CreateTaskParams,
CapacityMetrics,
TaskFilter,
} from './task-service';
export { TaskWorker, TaskContext, TaskResult } from './task-worker';
export {
handleProductResync,
handleProductDiscovery,
handleStoreDiscovery,
handleEntryPointDiscovery,
handleAnalyticsRefresh,
} from './handlers';

View File

@@ -0,0 +1,474 @@
/**
* Task Service
*
* Central service for managing worker tasks with:
* - Atomic task claiming (per-store locking)
* - Task lifecycle management
* - Auto-chaining of related tasks
* - Capacity planning metrics
*/
import { pool } from '../db/pool';
export type TaskRole =
| 'store_discovery'
| 'entry_point_discovery'
| 'product_discovery'
| 'product_resync'
| 'analytics_refresh';
export type TaskStatus =
| 'pending'
| 'claimed'
| 'running'
| 'completed'
| 'failed'
| 'stale';
export interface WorkerTask {
id: number;
role: TaskRole;
dispensary_id: number | null;
platform: string | null;
status: TaskStatus;
priority: number;
scheduled_for: Date | null;
worker_id: string | null;
claimed_at: Date | null;
started_at: Date | null;
completed_at: Date | null;
last_heartbeat_at: Date | null;
result: Record<string, unknown> | null;
error_message: string | null;
retry_count: number;
max_retries: number;
created_at: Date;
updated_at: Date;
}
export interface CreateTaskParams {
role: TaskRole;
dispensary_id?: number;
platform?: string;
priority?: number;
scheduled_for?: Date;
}
export interface CapacityMetrics {
role: string;
pending_tasks: number;
ready_tasks: number;
claimed_tasks: number;
running_tasks: number;
completed_last_hour: number;
failed_last_hour: number;
active_workers: number;
avg_duration_sec: number | null;
tasks_per_worker_hour: number | null;
estimated_hours_to_drain: number | null;
}
export interface TaskFilter {
role?: TaskRole;
status?: TaskStatus | TaskStatus[];
dispensary_id?: number;
worker_id?: string;
limit?: number;
offset?: number;
}
class TaskService {
/**
* Create a new task
*/
async createTask(params: CreateTaskParams): Promise<WorkerTask> {
const result = await pool.query(
`INSERT INTO worker_tasks (role, dispensary_id, platform, priority, scheduled_for)
VALUES ($1, $2, $3, $4, $5)
RETURNING *`,
[
params.role,
params.dispensary_id ?? null,
params.platform ?? null,
params.priority ?? 0,
params.scheduled_for ?? null,
]
);
return result.rows[0] as WorkerTask;
}
/**
* Create multiple tasks in a batch
*/
async createTasks(tasks: CreateTaskParams[]): Promise<number> {
if (tasks.length === 0) return 0;
const values = tasks.map((t, i) => {
const base = i * 5;
return `($${base + 1}, $${base + 2}, $${base + 3}, $${base + 4}, $${base + 5})`;
});
const params = tasks.flatMap((t) => [
t.role,
t.dispensary_id ?? null,
t.platform ?? null,
t.priority ?? 0,
t.scheduled_for ?? null,
]);
const result = await pool.query(
`INSERT INTO worker_tasks (role, dispensary_id, platform, priority, scheduled_for)
VALUES ${values.join(', ')}
ON CONFLICT DO NOTHING`,
params
);
return result.rowCount ?? 0;
}
/**
* Claim a task atomically for a worker
* Uses the SQL function for proper locking
*/
async claimTask(role: TaskRole, workerId: string): Promise<WorkerTask | null> {
const result = await pool.query(
`SELECT * FROM claim_task($1, $2)`,
[role, workerId]
);
return (result.rows[0] as WorkerTask) || null;
}
/**
* Mark a task as running (worker started processing)
*/
async startTask(taskId: number): Promise<void> {
await pool.query(
`UPDATE worker_tasks
SET status = 'running', started_at = NOW(), last_heartbeat_at = NOW()
WHERE id = $1`,
[taskId]
);
}
/**
* Update heartbeat to prevent stale detection
*/
async heartbeat(taskId: number): Promise<void> {
await pool.query(
`UPDATE worker_tasks
SET last_heartbeat_at = NOW()
WHERE id = $1 AND status = 'running'`,
[taskId]
);
}
/**
* Mark a task as completed
*/
async completeTask(taskId: number, result?: Record<string, unknown>): Promise<void> {
await pool.query(
`UPDATE worker_tasks
SET status = 'completed', completed_at = NOW(), result = $2
WHERE id = $1`,
[taskId, result ? JSON.stringify(result) : null]
);
}
/**
* Mark a task as failed
*/
async failTask(taskId: number, errorMessage: string): Promise<void> {
await pool.query(
`UPDATE worker_tasks
SET status = 'failed', completed_at = NOW(), error_message = $2
WHERE id = $1`,
[taskId, errorMessage]
);
}
/**
* Get a task by ID
*/
async getTask(taskId: number): Promise<WorkerTask | null> {
const result = await pool.query(
`SELECT * FROM worker_tasks WHERE id = $1`,
[taskId]
);
return (result.rows[0] as WorkerTask) || null;
}
/**
* List tasks with filters
*/
async listTasks(filter: TaskFilter = {}): Promise<WorkerTask[]> {
const conditions: string[] = [];
const params: (string | number | string[])[] = [];
let paramIndex = 1;
if (filter.role) {
conditions.push(`role = $${paramIndex++}`);
params.push(filter.role);
}
if (filter.status) {
if (Array.isArray(filter.status)) {
conditions.push(`status = ANY($${paramIndex++})`);
params.push(filter.status);
} else {
conditions.push(`status = $${paramIndex++}`);
params.push(filter.status);
}
}
if (filter.dispensary_id) {
conditions.push(`dispensary_id = $${paramIndex++}`);
params.push(filter.dispensary_id);
}
if (filter.worker_id) {
conditions.push(`worker_id = $${paramIndex++}`);
params.push(filter.worker_id);
}
const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : '';
const limit = filter.limit ?? 100;
const offset = filter.offset ?? 0;
const result = await pool.query(
`SELECT * FROM worker_tasks
${whereClause}
ORDER BY created_at DESC
LIMIT ${limit} OFFSET ${offset}`,
params
);
return result.rows as WorkerTask[];
}
/**
* Get capacity metrics for all roles
*/
async getCapacityMetrics(): Promise<CapacityMetrics[]> {
const result = await pool.query(
`SELECT * FROM v_worker_capacity`
);
return result.rows as CapacityMetrics[];
}
/**
* Get capacity metrics for a specific role
*/
async getRoleCapacity(role: TaskRole): Promise<CapacityMetrics | null> {
const result = await pool.query(
`SELECT * FROM v_worker_capacity WHERE role = $1`,
[role]
);
return (result.rows[0] as CapacityMetrics) || null;
}
/**
* Recover stale tasks from dead workers
*/
async recoverStaleTasks(staleThresholdMinutes = 10): Promise<number> {
const result = await pool.query(
`SELECT recover_stale_tasks($1)`,
[staleThresholdMinutes]
);
return (result.rows[0] as { recover_stale_tasks: number })?.recover_stale_tasks ?? 0;
}
/**
* Generate daily resync tasks for all active stores
*/
async generateDailyResyncTasks(batchesPerDay = 6, date?: Date): Promise<number> {
const result = await pool.query(
`SELECT generate_resync_tasks($1, $2)`,
[batchesPerDay, date ?? new Date()]
);
return (result.rows[0] as { generate_resync_tasks: number })?.generate_resync_tasks ?? 0;
}
/**
* Chain next task after completion
* Called automatically when a task completes successfully
*/
async chainNextTask(completedTask: WorkerTask): Promise<WorkerTask | null> {
if (completedTask.status !== 'completed') {
return null;
}
switch (completedTask.role) {
case 'store_discovery': {
// New stores discovered -> create entry_point_discovery tasks
const newStoreIds = (completedTask.result as { newStoreIds?: number[] })?.newStoreIds;
if (newStoreIds && newStoreIds.length > 0) {
for (const storeId of newStoreIds) {
await this.createTask({
role: 'entry_point_discovery',
dispensary_id: storeId,
platform: completedTask.platform ?? undefined,
priority: 10, // High priority for new stores
});
}
}
break;
}
case 'entry_point_discovery': {
// Entry point resolved -> create product_discovery task
const success = (completedTask.result as { success?: boolean })?.success;
if (success && completedTask.dispensary_id) {
return this.createTask({
role: 'product_discovery',
dispensary_id: completedTask.dispensary_id,
platform: completedTask.platform ?? undefined,
priority: 10,
});
}
break;
}
case 'product_discovery': {
// Product discovery done -> store is now ready for regular resync
// No immediate chaining needed; will be picked up by daily batch generation
break;
}
}
return null;
}
/**
* Create store discovery task for a platform/state
*/
async createStoreDiscoveryTask(
platform: string,
stateCode?: string,
priority = 0
): Promise<WorkerTask> {
return this.createTask({
role: 'store_discovery',
platform,
priority,
});
}
/**
* Create entry point discovery task for a specific store
*/
async createEntryPointTask(
dispensaryId: number,
platform: string,
priority = 10
): Promise<WorkerTask> {
return this.createTask({
role: 'entry_point_discovery',
dispensary_id: dispensaryId,
platform,
priority,
});
}
/**
* Create product discovery task for a specific store
*/
async createProductDiscoveryTask(
dispensaryId: number,
platform: string,
priority = 10
): Promise<WorkerTask> {
return this.createTask({
role: 'product_discovery',
dispensary_id: dispensaryId,
platform,
priority,
});
}
/**
* Get task counts by status for dashboard
*/
async getTaskCounts(): Promise<Record<TaskStatus, number>> {
const result = await pool.query(
`SELECT status, COUNT(*) as count
FROM worker_tasks
GROUP BY status`
);
const counts: Record<TaskStatus, number> = {
pending: 0,
claimed: 0,
running: 0,
completed: 0,
failed: 0,
stale: 0,
};
for (const row of result.rows) {
const typedRow = row as { status: TaskStatus; count: string };
counts[typedRow.status] = parseInt(typedRow.count, 10);
}
return counts;
}
/**
* Get recent task completions for a role
*/
async getRecentCompletions(role: TaskRole, limit = 10): Promise<WorkerTask[]> {
const result = await pool.query(
`SELECT * FROM worker_tasks
WHERE role = $1 AND status = 'completed'
ORDER BY completed_at DESC
LIMIT $2`,
[role, limit]
);
return result.rows as WorkerTask[];
}
/**
* Check if a store has any active tasks
*/
async hasActiveTask(dispensaryId: number): Promise<boolean> {
const result = await pool.query(
`SELECT EXISTS(
SELECT 1 FROM worker_tasks
WHERE dispensary_id = $1
AND status IN ('claimed', 'running')
) as exists`,
[dispensaryId]
);
return (result.rows[0] as { exists: boolean })?.exists ?? false;
}
/**
* Get the last completion time for a role
*/
async getLastCompletion(role: TaskRole): Promise<Date | null> {
const result = await pool.query(
`SELECT MAX(completed_at) as completed_at
FROM worker_tasks
WHERE role = $1 AND status = 'completed'`,
[role]
);
return (result.rows[0] as { completed_at: Date | null })?.completed_at ?? null;
}
/**
* Calculate workers needed to complete tasks within SLA
*/
async calculateWorkersNeeded(role: TaskRole, slaHours: number): Promise<number> {
const capacity = await this.getRoleCapacity(role);
if (!capacity || !capacity.tasks_per_worker_hour) {
return 1; // Default to 1 worker if no data
}
const pendingTasks = capacity.pending_tasks;
const tasksPerWorkerHour = capacity.tasks_per_worker_hour;
const totalTaskCapacityNeeded = pendingTasks / slaHours;
return Math.ceil(totalTaskCapacityNeeded / tasksPerWorkerHour);
}
}
export const taskService = new TaskService();

View File

@@ -0,0 +1,266 @@
/**
* Task Worker
*
* A unified worker that processes tasks from the worker_tasks queue.
* Replaces the fragmented job systems (job_schedules, dispensary_crawl_jobs, etc.)
*
* Usage:
* WORKER_ROLE=product_resync npx tsx src/tasks/task-worker.ts
*
* Environment:
* WORKER_ROLE - Which task role to process (required)
* WORKER_ID - Optional custom worker ID
* POLL_INTERVAL_MS - How often to check for tasks (default: 5000)
* HEARTBEAT_INTERVAL_MS - How often to update heartbeat (default: 30000)
*/
import { Pool } from 'pg';
import { v4 as uuidv4 } from 'uuid';
import { taskService, TaskRole, WorkerTask } from './task-service';
import { getPool } from '../db/pool';
// Task handlers by role
import { handleProductResync } from './handlers/product-resync';
import { handleProductDiscovery } from './handlers/product-discovery';
import { handleStoreDiscovery } from './handlers/store-discovery';
import { handleEntryPointDiscovery } from './handlers/entry-point-discovery';
import { handleAnalyticsRefresh } from './handlers/analytics-refresh';
const POLL_INTERVAL_MS = parseInt(process.env.POLL_INTERVAL_MS || '5000');
const HEARTBEAT_INTERVAL_MS = parseInt(process.env.HEARTBEAT_INTERVAL_MS || '30000');
export interface TaskContext {
pool: Pool;
workerId: string;
task: WorkerTask;
heartbeat: () => Promise<void>;
}
export interface TaskResult {
success: boolean;
productsProcessed?: number;
snapshotsCreated?: number;
storesDiscovered?: number;
error?: string;
[key: string]: unknown;
}
type TaskHandler = (ctx: TaskContext) => Promise<TaskResult>;
const TASK_HANDLERS: Record<TaskRole, TaskHandler> = {
product_resync: handleProductResync,
product_discovery: handleProductDiscovery,
store_discovery: handleStoreDiscovery,
entry_point_discovery: handleEntryPointDiscovery,
analytics_refresh: handleAnalyticsRefresh,
};
export class TaskWorker {
private pool: Pool;
private workerId: string;
private role: TaskRole;
private isRunning: boolean = false;
private heartbeatInterval: NodeJS.Timeout | null = null;
private currentTask: WorkerTask | null = null;
constructor(role: TaskRole, workerId?: string) {
this.pool = getPool();
this.role = role;
this.workerId = workerId || `worker-${role}-${uuidv4().slice(0, 8)}`;
}
/**
* Start the worker loop
*/
async start(): Promise<void> {
this.isRunning = true;
console.log(`[TaskWorker] Starting worker ${this.workerId} for role: ${this.role}`);
while (this.isRunning) {
try {
await this.processNextTask();
} catch (error: any) {
console.error(`[TaskWorker] Loop error:`, error.message);
await this.sleep(POLL_INTERVAL_MS);
}
}
console.log(`[TaskWorker] Worker ${this.workerId} stopped`);
}
/**
* Stop the worker
*/
stop(): void {
this.isRunning = false;
this.stopHeartbeat();
console.log(`[TaskWorker] Stopping worker ${this.workerId}...`);
}
/**
* Process the next available task
*/
private async processNextTask(): Promise<void> {
// Try to claim a task
const task = await taskService.claimTask(this.role, this.workerId);
if (!task) {
// No tasks available, wait and retry
await this.sleep(POLL_INTERVAL_MS);
return;
}
this.currentTask = task;
console.log(`[TaskWorker] Claimed task ${task.id} (${task.role}) for dispensary ${task.dispensary_id || 'N/A'}`);
// Start heartbeat
this.startHeartbeat(task.id);
try {
// Mark as running
await taskService.startTask(task.id);
// Get handler for this role
const handler = TASK_HANDLERS[task.role];
if (!handler) {
throw new Error(`No handler registered for role: ${task.role}`);
}
// Create context
const ctx: TaskContext = {
pool: this.pool,
workerId: this.workerId,
task,
heartbeat: async () => {
await taskService.heartbeat(task.id);
},
};
// Execute the task
const result = await handler(ctx);
if (result.success) {
// Mark as completed
await taskService.completeTask(task.id, result);
console.log(`[TaskWorker] Task ${task.id} completed successfully`);
// Chain next task if applicable
const chainedTask = await taskService.chainNextTask({
...task,
status: 'completed',
result,
});
if (chainedTask) {
console.log(`[TaskWorker] Chained new task ${chainedTask.id} (${chainedTask.role})`);
}
} else {
// Mark as failed
await taskService.failTask(task.id, result.error || 'Unknown error');
console.log(`[TaskWorker] Task ${task.id} failed: ${result.error}`);
}
} catch (error: any) {
// Mark as failed
await taskService.failTask(task.id, error.message);
console.error(`[TaskWorker] Task ${task.id} threw error:`, error.message);
} finally {
this.stopHeartbeat();
this.currentTask = null;
}
}
/**
* Start heartbeat interval
*/
private startHeartbeat(taskId: number): void {
this.heartbeatInterval = setInterval(async () => {
try {
await taskService.heartbeat(taskId);
} catch (error: any) {
console.warn(`[TaskWorker] Heartbeat failed:`, error.message);
}
}, HEARTBEAT_INTERVAL_MS);
}
/**
* Stop heartbeat interval
*/
private stopHeartbeat(): void {
if (this.heartbeatInterval) {
clearInterval(this.heartbeatInterval);
this.heartbeatInterval = null;
}
}
/**
* Sleep helper
*/
private sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
/**
* Get worker info
*/
getInfo(): { workerId: string; role: TaskRole; isRunning: boolean; currentTaskId: number | null } {
return {
workerId: this.workerId,
role: this.role,
isRunning: this.isRunning,
currentTaskId: this.currentTask?.id || null,
};
}
}
// ============================================================
// CLI ENTRY POINT
// ============================================================
async function main(): Promise<void> {
const role = process.env.WORKER_ROLE as TaskRole;
if (!role) {
console.error('Error: WORKER_ROLE environment variable is required');
console.error('Valid roles: store_discovery, entry_point_discovery, product_discovery, product_resync, analytics_refresh');
process.exit(1);
}
const validRoles: TaskRole[] = [
'store_discovery',
'entry_point_discovery',
'product_discovery',
'product_resync',
'analytics_refresh',
];
if (!validRoles.includes(role)) {
console.error(`Error: Invalid WORKER_ROLE: ${role}`);
console.error(`Valid roles: ${validRoles.join(', ')}`);
process.exit(1);
}
const workerId = process.env.WORKER_ID;
const worker = new TaskWorker(role, workerId);
// Handle graceful shutdown
process.on('SIGTERM', () => {
console.log('[TaskWorker] Received SIGTERM, shutting down...');
worker.stop();
});
process.on('SIGINT', () => {
console.log('[TaskWorker] Received SIGINT, shutting down...');
worker.stop();
});
await worker.start();
}
// Run if this is the main module
if (require.main === module) {
main().catch((error) => {
console.error('[TaskWorker] Fatal error:', error);
process.exit(1);
});
}
export { main };

View File

@@ -1,26 +1,29 @@
/**
* Local Image Storage Utility
*
* Downloads and stores product images to local filesystem.
* Replaces MinIO-based storage with simple local file storage.
* Downloads and stores product images to local filesystem with proper hierarchy.
*
* Directory structure:
* /images/products/<dispensary_id>/<product_id>.webp
* /images/products/<dispensary_id>/<product_id>-thumb.webp
* /images/products/<dispensary_id>/<product_id>-medium.webp
* /images/brands/<brand_slug>.webp
* /images/products/<state>/<store_slug>/<brand_slug>/<product_id>/image.webp
* /images/products/<state>/<store_slug>/<brand_slug>/<product_id>/image-medium.webp
* /images/products/<state>/<store_slug>/<brand_slug>/<product_id>/image-thumb.webp
* /images/brands/<brand_slug>/logo.webp
*
* This structure allows:
* - Easy migration to MinIO/S3 (bucket per state)
* - Browsing by state/store/brand
* - Multiple images per product (future: gallery)
*/
import axios from 'axios';
import sharp from 'sharp';
// @ts-ignore - sharp module typing quirk
const sharp = require('sharp');
import * as fs from 'fs/promises';
import * as path from 'path';
import { createHash } from 'crypto';
// Base path for image storage - configurable via env
// Uses project-relative paths by default, NOT /app or other privileged paths
function getImagesBasePath(): string {
// Priority: IMAGES_PATH > STORAGE_BASE_PATH/images > ./storage/images
if (process.env.IMAGES_PATH) {
return process.env.IMAGES_PATH;
}
@@ -35,16 +38,28 @@ const IMAGES_BASE_PATH = getImagesBasePath();
const IMAGES_PUBLIC_URL = process.env.IMAGES_PUBLIC_URL || '/images';
export interface LocalImageSizes {
full: string; // URL path: /images/products/123/456.webp
medium: string; // URL path: /images/products/123/456-medium.webp
thumb: string; // URL path: /images/products/123/456-thumb.webp
original: string; // URL path to original image
// Legacy compatibility - all point to original until we add image proxy
full: string;
medium: string;
thumb: string;
}
export interface DownloadResult {
success: boolean;
urls?: LocalImageSizes;
localPaths?: LocalImageSizes;
error?: string;
bytesDownloaded?: number;
skipped?: boolean; // True if image already exists
}
export interface ProductImageContext {
stateCode: string; // e.g., "AZ", "CA"
storeSlug: string; // e.g., "deeply-rooted"
brandSlug: string; // e.g., "high-west-farms"
productId: string; // External product ID
dispensaryId?: number; // For backwards compat
}
/**
@@ -58,6 +73,17 @@ async function ensureDir(dirPath: string): Promise<void> {
}
}
/**
* Sanitize a string for use in file paths
*/
function slugify(str: string): string {
return str
.toLowerCase()
.replace(/[^a-z0-9]+/g, '-')
.replace(/^-+|-+$/g, '')
.substring(0, 50) || 'unknown';
}
/**
* Generate a short hash from a URL for deduplication
*/
@@ -81,53 +107,30 @@ async function downloadImage(imageUrl: string): Promise<Buffer> {
}
/**
* Process and save image in multiple sizes
* Returns the file paths relative to IMAGES_BASE_PATH
* Process and save original image (convert to webp for consistency)
*
* We store only the original - resizing will be done on-demand via
* an image proxy service (imgproxy, thumbor, or similar) in the future.
*/
async function processAndSaveImage(
buffer: Buffer,
outputDir: string,
baseFilename: string
): Promise<{ full: string; medium: string; thumb: string; totalBytes: number }> {
): Promise<{ original: string; totalBytes: number }> {
await ensureDir(outputDir);
const fullPath = path.join(outputDir, `${baseFilename}.webp`);
const mediumPath = path.join(outputDir, `${baseFilename}-medium.webp`);
const thumbPath = path.join(outputDir, `${baseFilename}-thumb.webp`);
const originalPath = path.join(outputDir, `${baseFilename}.webp`);
// Process images in parallel
const [fullBuffer, mediumBuffer, thumbBuffer] = await Promise.all([
// Full: max 1200x1200, high quality
sharp(buffer)
.resize(1200, 1200, { fit: 'inside', withoutEnlargement: true })
.webp({ quality: 85 })
.toBuffer(),
// Medium: 600x600
sharp(buffer)
.resize(600, 600, { fit: 'inside', withoutEnlargement: true })
.webp({ quality: 80 })
.toBuffer(),
// Thumb: 200x200
sharp(buffer)
.resize(200, 200, { fit: 'inside', withoutEnlargement: true })
.webp({ quality: 75 })
.toBuffer(),
]);
// Convert to webp, preserve original dimensions, high quality
const originalBuffer = await sharp(buffer)
.webp({ quality: 90 })
.toBuffer();
// Save all sizes
await Promise.all([
fs.writeFile(fullPath, fullBuffer),
fs.writeFile(mediumPath, mediumBuffer),
fs.writeFile(thumbPath, thumbBuffer),
]);
const totalBytes = fullBuffer.length + mediumBuffer.length + thumbBuffer.length;
await fs.writeFile(originalPath, originalBuffer);
return {
full: fullPath,
medium: mediumPath,
thumb: thumbPath,
totalBytes,
original: originalPath,
totalBytes: originalBuffer.length,
};
}
@@ -135,47 +138,107 @@ async function processAndSaveImage(
* Convert a file path to a public URL
*/
function pathToUrl(filePath: string): string {
// Find /products/ or /brands/ in the path and extract from there
const productsMatch = filePath.match(/(\/products\/.*)/);
const brandsMatch = filePath.match(/(\/brands\/.*)/);
if (productsMatch) {
return `${IMAGES_PUBLIC_URL}${productsMatch[1]}`;
}
if (brandsMatch) {
return `${IMAGES_PUBLIC_URL}${brandsMatch[1]}`;
}
// Fallback: try to replace base path (works if paths match exactly)
const relativePath = filePath.replace(IMAGES_BASE_PATH, '');
return `${IMAGES_PUBLIC_URL}${relativePath}`;
}
/**
* Download and store a product image locally
* Build the directory path for a product image
* Structure: /images/products/<state>/<store>/<brand>/<product>/
*/
function buildProductImagePath(ctx: ProductImageContext): string {
const state = slugify(ctx.stateCode || 'unknown');
const store = slugify(ctx.storeSlug || 'unknown');
const brand = slugify(ctx.brandSlug || 'unknown');
const product = slugify(ctx.productId || 'unknown');
return path.join(IMAGES_BASE_PATH, 'products', state, store, brand, product);
}
/**
* Download and store a product image with proper hierarchy
*
* @param imageUrl - The third-party image URL to download
* @param dispensaryId - The dispensary ID (for directory organization)
* @param productId - The product ID or external ID (for filename)
* @param ctx - Product context (state, store, brand, product)
* @param options - Download options
* @returns Download result with local URLs
*/
export async function downloadProductImage(
imageUrl: string,
dispensaryId: number,
productId: string | number
ctx: ProductImageContext,
options: { skipIfExists?: boolean } = {}
): Promise<DownloadResult> {
const { skipIfExists = true } = options;
try {
if (!imageUrl) {
return { success: false, error: 'No image URL provided' };
}
const outputDir = buildProductImagePath(ctx);
const urlHash = hashUrl(imageUrl);
const baseFilename = `image-${urlHash}`;
// Check if image already exists
if (skipIfExists) {
const existingPath = path.join(outputDir, `${baseFilename}.webp`);
try {
await fs.access(existingPath);
// Image exists, return existing URL
const url = pathToUrl(existingPath);
return {
success: true,
skipped: true,
urls: {
original: url,
full: url,
medium: url,
thumb: url,
},
localPaths: {
original: existingPath,
full: existingPath,
medium: existingPath,
thumb: existingPath,
},
};
} catch {
// Image doesn't exist, continue to download
}
}
// Download the image
const buffer = await downloadImage(imageUrl);
// Organize by dispensary ID
const outputDir = path.join(IMAGES_BASE_PATH, 'products', String(dispensaryId));
// Use product ID + URL hash for uniqueness
const urlHash = hashUrl(imageUrl);
const baseFilename = `${productId}-${urlHash}`;
// Process and save
// Process and save (original only)
const result = await processAndSaveImage(buffer, outputDir, baseFilename);
const url = pathToUrl(result.original);
return {
success: true,
urls: {
full: pathToUrl(result.full),
medium: pathToUrl(result.medium),
thumb: pathToUrl(result.thumb),
original: url,
full: url,
medium: url,
thumb: url,
},
localPaths: {
original: result.original,
full: result.original,
medium: result.original,
thumb: result.original,
},
bytesDownloaded: result.totalBytes,
};
@@ -188,33 +251,71 @@ export async function downloadProductImage(
}
/**
* Download and store a brand logo locally
* Legacy function - backwards compatible with old signature
* Maps to new hierarchy using dispensary_id as store identifier
*/
export async function downloadProductImageLegacy(
imageUrl: string,
dispensaryId: number,
productId: string | number
): Promise<DownloadResult> {
return downloadProductImage(imageUrl, {
stateCode: 'unknown',
storeSlug: `store-${dispensaryId}`,
brandSlug: 'unknown',
productId: String(productId),
dispensaryId,
});
}
/**
* Download and store a brand logo
*
* @param logoUrl - The brand logo URL
* @param brandId - The brand ID or slug
* @param brandSlug - The brand slug/ID
* @returns Download result with local URL
*/
export async function downloadBrandLogo(
logoUrl: string,
brandId: string
brandSlug: string,
options: { skipIfExists?: boolean } = {}
): Promise<DownloadResult> {
const { skipIfExists = true } = options;
try {
if (!logoUrl) {
return { success: false, error: 'No logo URL provided' };
}
const safeBrandSlug = slugify(brandSlug);
const outputDir = path.join(IMAGES_BASE_PATH, 'brands', safeBrandSlug);
const urlHash = hashUrl(logoUrl);
const baseFilename = `logo-${urlHash}`;
// Check if logo already exists
if (skipIfExists) {
const existingPath = path.join(outputDir, `${baseFilename}.webp`);
try {
await fs.access(existingPath);
return {
success: true,
skipped: true,
urls: {
original: pathToUrl(existingPath),
full: pathToUrl(existingPath),
medium: pathToUrl(existingPath),
thumb: pathToUrl(existingPath),
},
};
} catch {
// Logo doesn't exist, continue
}
}
// Download the image
const buffer = await downloadImage(logoUrl);
// Brand logos go in /images/brands/
const outputDir = path.join(IMAGES_BASE_PATH, 'brands');
// Sanitize brand ID for filename
const safeBrandId = brandId.replace(/[^a-zA-Z0-9-_]/g, '_');
const urlHash = hashUrl(logoUrl);
const baseFilename = `${safeBrandId}-${urlHash}`;
// Process and save (single size for logos)
// Brand logos in their own directory
await ensureDir(outputDir);
const logoPath = path.join(outputDir, `${baseFilename}.webp`);
@@ -228,6 +329,7 @@ export async function downloadBrandLogo(
return {
success: true,
urls: {
original: pathToUrl(logoPath),
full: pathToUrl(logoPath),
medium: pathToUrl(logoPath),
thumb: pathToUrl(logoPath),
@@ -243,20 +345,16 @@ export async function downloadBrandLogo(
}
/**
* Check if a local image already exists
* Check if a product image already exists
*/
export async function imageExists(
dispensaryId: number,
productId: string | number,
export async function productImageExists(
ctx: ProductImageContext,
imageUrl: string
): Promise<boolean> {
const outputDir = buildProductImagePath(ctx);
const urlHash = hashUrl(imageUrl);
const imagePath = path.join(
IMAGES_BASE_PATH,
'products',
String(dispensaryId),
`${productId}-${urlHash}.webp`
);
const imagePath = path.join(outputDir, `image-${urlHash}.webp`);
try {
await fs.access(imagePath);
return true;
@@ -266,24 +364,27 @@ export async function imageExists(
}
/**
* Delete a product's local images
* Get the local image URL for a product (if exists)
*/
export async function deleteProductImages(
dispensaryId: number,
productId: string | number,
imageUrl?: string
): Promise<void> {
const productDir = path.join(IMAGES_BASE_PATH, 'products', String(dispensaryId));
const prefix = imageUrl
? `${productId}-${hashUrl(imageUrl)}`
: String(productId);
export async function getProductImageUrl(
ctx: ProductImageContext,
imageUrl: string
): Promise<LocalImageSizes | null> {
const outputDir = buildProductImagePath(ctx);
const urlHash = hashUrl(imageUrl);
const imagePath = path.join(outputDir, `image-${urlHash}.webp`);
try {
const files = await fs.readdir(productDir);
const toDelete = files.filter(f => f.startsWith(prefix));
await Promise.all(toDelete.map(f => fs.unlink(path.join(productDir, f))));
await fs.access(imagePath);
const url = pathToUrl(imagePath);
return {
original: url,
full: url,
medium: url,
thumb: url,
};
} catch {
// Directory might not exist, that's fine
return null;
}
}
@@ -296,19 +397,17 @@ export function isImageStorageReady(): boolean {
/**
* Initialize the image storage directories
* Does NOT throw on failure - logs warning and continues
*/
export async function initializeImageStorage(): Promise<void> {
try {
await ensureDir(path.join(IMAGES_BASE_PATH, 'products'));
await ensureDir(path.join(IMAGES_BASE_PATH, 'brands'));
console.log(`Image storage initialized at ${IMAGES_BASE_PATH}`);
console.log(`[ImageStorage] Initialized at ${IMAGES_BASE_PATH}`);
imageStorageReady = true;
} catch (error: any) {
console.warn(`⚠️ WARNING: Could not initialize image storage at ${IMAGES_BASE_PATH}: ${error.message}`);
console.warn(' Image upload/processing is disabled. Server will continue without image features.');
console.warn(`[ImageStorage] WARNING: Could not initialize at ${IMAGES_BASE_PATH}: ${error.message}`);
console.warn(' Image features disabled. Server will continue without image downloads.');
imageStorageReady = false;
// Do NOT throw - server should still start
}
}
@@ -316,34 +415,43 @@ export async function initializeImageStorage(): Promise<void> {
* Get storage stats
*/
export async function getStorageStats(): Promise<{
productsDir: string;
brandsDir: string;
basePath: string;
productCount: number;
brandCount: number;
totalSizeBytes: number;
}> {
const productsDir = path.join(IMAGES_BASE_PATH, 'products');
const brandsDir = path.join(IMAGES_BASE_PATH, 'brands');
let productCount = 0;
let brandCount = 0;
let totalSizeBytes = 0;
try {
const productDirs = await fs.readdir(productsDir);
for (const dir of productDirs) {
const files = await fs.readdir(path.join(productsDir, dir));
productCount += files.filter(f => f.endsWith('.webp') && !f.includes('-')).length;
}
} catch { /* ignore */ }
async function countDir(dirPath: string): Promise<{ count: number; size: number }> {
let count = 0;
let size = 0;
try {
const entries = await fs.readdir(dirPath, { withFileTypes: true });
for (const entry of entries) {
const fullPath = path.join(dirPath, entry.name);
if (entry.isDirectory()) {
const sub = await countDir(fullPath);
count += sub.count;
size += sub.size;
} else if (entry.name.endsWith('.webp') && !entry.name.includes('-')) {
count++;
const stat = await fs.stat(fullPath);
size += stat.size;
}
}
} catch { /* ignore */ }
return { count, size };
}
try {
const brandFiles = await fs.readdir(brandsDir);
brandCount = brandFiles.filter(f => f.endsWith('.webp')).length;
} catch { /* ignore */ }
const products = await countDir(path.join(IMAGES_BASE_PATH, 'products'));
const brands = await countDir(path.join(IMAGES_BASE_PATH, 'brands'));
return {
productsDir,
brandsDir,
productCount,
brandCount,
basePath: IMAGES_BASE_PATH,
productCount: products.count,
brandCount: brands.count,
totalSizeBytes: products.size + brands.size,
};
}

View File

@@ -1,5 +1,5 @@
# Build stage
FROM node:20-slim AS builder
FROM code.cannabrands.app/creationshop/node:20-slim AS builder
WORKDIR /app
@@ -20,7 +20,7 @@ COPY . .
RUN npm run build
# Production stage
FROM nginx:alpine
FROM code.cannabrands.app/creationshop/nginx:alpine
# Copy built assets from builder stage
COPY --from=builder /app/dist /usr/share/nginx/html

View File

@@ -7,8 +7,8 @@
<title>CannaIQ - Cannabis Menu Intelligence Platform</title>
<meta name="description" content="CannaIQ provides real-time cannabis dispensary menu data, product tracking, and analytics for dispensaries across Arizona." />
<meta name="keywords" content="cannabis, dispensary, menu, products, analytics, Arizona" />
<script type="module" crossorigin src="/assets/index-DTnhZh6X.js"></script>
<link rel="stylesheet" crossorigin href="/assets/index-9PqXc--D.css">
<script type="module" crossorigin src="/assets/index-BML8-px1.js"></script>
<link rel="stylesheet" crossorigin href="/assets/index-B2gR-58G.css">
</head>
<body>
<div id="root"></div>

View File

@@ -47,6 +47,7 @@ import StateDetail from './pages/StateDetail';
import { Discovery } from './pages/Discovery';
import { WorkersDashboard } from './pages/WorkersDashboard';
import { JobQueue } from './pages/JobQueue';
import TasksDashboard from './pages/TasksDashboard';
import { ScraperOverviewDashboard } from './pages/ScraperOverviewDashboard';
import { SeoOrchestrator } from './pages/admin/seo/SeoOrchestrator';
import { StatePage } from './pages/public/StatePage';
@@ -124,6 +125,8 @@ export default function App() {
<Route path="/workers" element={<PrivateRoute><WorkersDashboard /></PrivateRoute>} />
{/* Job Queue Management */}
<Route path="/job-queue" element={<PrivateRoute><JobQueue /></PrivateRoute>} />
{/* Task Queue Dashboard */}
<Route path="/tasks" element={<PrivateRoute><TasksDashboard /></PrivateRoute>} />
{/* Scraper Overview Dashboard (new primary) */}
<Route path="/scraper/overview" element={<PrivateRoute><ScraperOverviewDashboard /></PrivateRoute>} />
<Route path="*" element={<Navigate to="/dashboard" replace />} />

View File

@@ -0,0 +1,329 @@
import { useEffect, useState } from 'react';
import { api } from '../lib/api';
interface PipelineStep {
name: string;
state: 'pending' | 'running' | 'success' | 'failure' | 'skipped';
}
interface DeployStatusData {
running: {
sha: string;
sha_full: string;
build_time: string;
image_tag: string;
};
latest: {
sha: string;
sha_full: string;
message: string;
author: string;
timestamp: string;
} | null;
is_latest: boolean;
commits_behind: number;
pipeline: {
number: number;
status: string;
event: string;
branch: string;
message: string;
commit: string;
author: string;
created: number;
steps?: PipelineStep[];
} | null;
error?: string;
}
const statusColors: Record<string, string> = {
success: '#10b981',
running: '#f59e0b',
pending: '#6b7280',
failure: '#ef4444',
error: '#ef4444',
skipped: '#9ca3af',
};
const statusIcons: Record<string, string> = {
success: '\u2713',
running: '\u25B6',
pending: '\u25CB',
failure: '\u2717',
error: '\u2717',
skipped: '\u2212',
};
export function DeployStatus() {
const [data, setData] = useState<DeployStatusData | null>(null);
const [loading, setLoading] = useState(true);
const [error, setError] = useState<string | null>(null);
const fetchStatus = async () => {
try {
setLoading(true);
const { data: responseData } = await api.get<DeployStatusData>('/api/admin/deploy-status');
setData(responseData);
setError(null);
} catch (err: any) {
setError(err.message || 'Failed to fetch deploy status');
} finally {
setLoading(false);
}
};
useEffect(() => {
fetchStatus();
// Auto-refresh every 30 seconds
const interval = setInterval(fetchStatus, 30000);
return () => clearInterval(interval);
}, []);
const formatTime = (timestamp: string | number) => {
const date = typeof timestamp === 'number'
? new Date(timestamp * 1000)
: new Date(timestamp);
return date.toLocaleString();
};
const formatTimeAgo = (timestamp: string | number) => {
const date = typeof timestamp === 'number'
? new Date(timestamp * 1000)
: new Date(timestamp);
const now = new Date();
const diffMs = now.getTime() - date.getTime();
const diffMins = Math.floor(diffMs / 60000);
const diffHours = Math.floor(diffMins / 60);
const diffDays = Math.floor(diffHours / 24);
if (diffMins < 1) return 'just now';
if (diffMins < 60) return `${diffMins}m ago`;
if (diffHours < 24) return `${diffHours}h ago`;
return `${diffDays}d ago`;
};
if (loading && !data) {
return (
<div style={{ padding: '20px', background: '#1f2937', borderRadius: '8px', color: '#9ca3af' }}>
Loading deploy status...
</div>
);
}
if (error && !data) {
return (
<div style={{ padding: '20px', background: '#1f2937', borderRadius: '8px', color: '#ef4444' }}>
Error: {error}
</div>
);
}
if (!data) return null;
const pipelineStatus = data.pipeline?.status || 'unknown';
return (
<div style={{
background: '#1f2937',
borderRadius: '8px',
overflow: 'hidden',
border: '1px solid #374151'
}}>
{/* Header */}
<div style={{
padding: '16px 20px',
borderBottom: '1px solid #374151',
display: 'flex',
justifyContent: 'space-between',
alignItems: 'center'
}}>
<div style={{ display: 'flex', alignItems: 'center', gap: '12px' }}>
<span style={{ fontSize: '18px', fontWeight: '600', color: '#f3f4f6' }}>
Deploy Status
</span>
{data.is_latest ? (
<span style={{
background: '#10b981',
color: 'white',
padding: '2px 8px',
borderRadius: '4px',
fontSize: '12px',
fontWeight: '500'
}}>
Up to date
</span>
) : (
<span style={{
background: '#f59e0b',
color: 'white',
padding: '2px 8px',
borderRadius: '4px',
fontSize: '12px',
fontWeight: '500'
}}>
{data.commits_behind} commit{data.commits_behind !== 1 ? 's' : ''} behind
</span>
)}
</div>
<button
onClick={fetchStatus}
disabled={loading}
style={{
background: '#374151',
border: 'none',
padding: '6px 12px',
borderRadius: '4px',
color: '#9ca3af',
cursor: loading ? 'not-allowed' : 'pointer',
fontSize: '12px'
}}
>
{loading ? 'Refreshing...' : 'Refresh'}
</button>
</div>
{/* Version Info */}
<div style={{ padding: '16px 20px', display: 'grid', gridTemplateColumns: '1fr 1fr', gap: '20px' }}>
{/* Running Version */}
<div>
<div style={{ color: '#9ca3af', fontSize: '12px', marginBottom: '8px', textTransform: 'uppercase' }}>
Running Version
</div>
<div style={{ display: 'flex', alignItems: 'center', gap: '8px' }}>
<code style={{
background: '#374151',
padding: '4px 8px',
borderRadius: '4px',
color: '#10b981',
fontSize: '14px',
fontFamily: 'monospace'
}}>
{data.running.sha}
</code>
<span style={{ color: '#6b7280', fontSize: '12px' }}>
{formatTimeAgo(data.running.build_time)}
</span>
</div>
</div>
{/* Latest Commit */}
<div>
<div style={{ color: '#9ca3af', fontSize: '12px', marginBottom: '8px', textTransform: 'uppercase' }}>
Latest Commit
</div>
{data.latest ? (
<div>
<div style={{ display: 'flex', alignItems: 'center', gap: '8px' }}>
<code style={{
background: '#374151',
padding: '4px 8px',
borderRadius: '4px',
color: data.is_latest ? '#10b981' : '#f59e0b',
fontSize: '14px',
fontFamily: 'monospace'
}}>
{data.latest.sha}
</code>
<span style={{ color: '#6b7280', fontSize: '12px' }}>
{formatTimeAgo(data.latest.timestamp)}
</span>
</div>
<div style={{
color: '#9ca3af',
fontSize: '13px',
marginTop: '4px',
overflow: 'hidden',
textOverflow: 'ellipsis',
whiteSpace: 'nowrap',
maxWidth: '300px'
}}>
{data.latest.message}
</div>
</div>
) : (
<span style={{ color: '#6b7280' }}>Unable to fetch</span>
)}
</div>
</div>
{/* Pipeline Status */}
{data.pipeline && (
<div style={{
padding: '16px 20px',
borderTop: '1px solid #374151',
background: '#111827'
}}>
<div style={{
display: 'flex',
justifyContent: 'space-between',
alignItems: 'center',
marginBottom: '12px'
}}>
<div style={{ display: 'flex', alignItems: 'center', gap: '10px' }}>
<span style={{
color: statusColors[pipelineStatus] || '#6b7280',
fontSize: '16px'
}}>
{statusIcons[pipelineStatus] || '?'}
</span>
<span style={{ color: '#f3f4f6', fontWeight: '500' }}>
Pipeline #{data.pipeline.number}
</span>
<span style={{
color: statusColors[pipelineStatus] || '#6b7280',
fontSize: '13px',
textTransform: 'capitalize'
}}>
{pipelineStatus}
</span>
</div>
<span style={{ color: '#6b7280', fontSize: '12px' }}>
{data.pipeline.branch} \u2022 {data.pipeline.commit}
</span>
</div>
{/* Pipeline Steps */}
{data.pipeline.steps && data.pipeline.steps.length > 0 && (
<div style={{
display: 'flex',
gap: '4px',
flexWrap: 'wrap'
}}>
{data.pipeline.steps.map((step, idx) => (
<div
key={idx}
style={{
display: 'flex',
alignItems: 'center',
gap: '4px',
background: '#1f2937',
padding: '4px 8px',
borderRadius: '4px',
fontSize: '12px'
}}
>
<span style={{ color: statusColors[step.state] || '#6b7280' }}>
{statusIcons[step.state] || '?'}
</span>
<span style={{ color: '#9ca3af' }}>{step.name}</span>
</div>
))}
</div>
)}
{/* Commit message */}
<div style={{
color: '#6b7280',
fontSize: '12px',
marginTop: '8px',
overflow: 'hidden',
textOverflow: 'ellipsis',
whiteSpace: 'nowrap'
}}>
{data.pipeline.message}
</div>
</div>
)}
</div>
);
}

View File

@@ -20,9 +20,11 @@ import {
Menu,
X,
Users,
UserCog,
ListOrdered,
Key,
Bot
Bot,
ListChecks
} from 'lucide-react';
interface LayoutProps {
@@ -30,6 +32,7 @@ interface LayoutProps {
}
interface VersionInfo {
version?: string;
build_version: string;
git_sha: string;
build_time: string;
@@ -124,7 +127,14 @@ export function Layout({ children }: LayoutProps) {
<path d="M3.5 6C2 8 1 10.5 1 13C1 18.5 6 22 12 22C18 22 23 18.5 23 13C23 10.5 22 8 20.5 6L12 12L3.5 6Z" opacity="0.7" />
</svg>
</div>
<span className="text-lg font-bold text-gray-900">CannaIQ</span>
<div>
<span className="text-lg font-bold text-gray-900">CannaIQ</span>
{versionInfo && (
<p className="text-xs text-gray-400">
v{versionInfo.version} ({versionInfo.git_sha}) {versionInfo.build_time !== 'unknown' && `- ${new Date(versionInfo.build_time).toLocaleDateString()}`}
</p>
)}
</div>
</div>
<p className="text-xs text-gray-500 mt-2 truncate">{user?.email}</p>
</div>
@@ -152,8 +162,10 @@ export function Layout({ children }: LayoutProps) {
<NavSection title="Admin">
<NavLink to="/admin/orchestrator" icon={<Activity className="w-4 h-4" />} label="Orchestrator" isActive={isActive('/admin/orchestrator')} />
<NavLink to="/users" icon={<UserCog className="w-4 h-4" />} label="Users" isActive={isActive('/users')} />
<NavLink to="/workers" icon={<Users className="w-4 h-4" />} label="Workers" isActive={isActive('/workers')} />
<NavLink to="/job-queue" icon={<ListOrdered className="w-4 h-4" />} label="Job Queue" isActive={isActive('/job-queue')} />
<NavLink to="/tasks" icon={<ListChecks className="w-4 h-4" />} label="Task Queue" isActive={isActive('/tasks')} />
<NavLink to="/admin/seo" icon={<FileText className="w-4 h-4" />} label="SEO Pages" isActive={isActive('/admin/seo')} />
<NavLink to="/proxies" icon={<Shield className="w-4 h-4" />} label="Proxies" isActive={isActive('/proxies')} />
<NavLink to="/api-permissions" icon={<Key className="w-4 h-4" />} label="API Keys" isActive={isActive('/api-permissions')} />
@@ -169,14 +181,6 @@ export function Layout({ children }: LayoutProps) {
<span>Logout</span>
</button>
</div>
{/* Version Footer */}
{versionInfo && (
<div className="px-3 py-2 border-t border-gray-200 bg-gray-50">
<p className="text-xs text-gray-500 text-center">{versionInfo.build_version} ({versionInfo.git_sha.slice(0, 7)})</p>
<p className="text-xs text-gray-400 text-center mt-0.5">{versionInfo.image_tag}</p>
</div>
)}
</>
);

View File

@@ -113,8 +113,16 @@ class ApiClient {
});
}
async getDispensaries() {
return this.request<{ dispensaries: any[] }>('/api/dispensaries');
async getDispensaries(params?: { limit?: number; offset?: number; search?: string; city?: string; state?: string; crawl_enabled?: string }) {
const searchParams = new URLSearchParams();
if (params?.limit) searchParams.append('limit', params.limit.toString());
if (params?.offset) searchParams.append('offset', params.offset.toString());
if (params?.search) searchParams.append('search', params.search);
if (params?.city) searchParams.append('city', params.city);
if (params?.state) searchParams.append('state', params.state);
if (params?.crawl_enabled) searchParams.append('crawl_enabled', params.crawl_enabled);
const queryString = searchParams.toString() ? `?${searchParams.toString()}` : '';
return this.request<{ dispensaries: any[]; total: number; limit: number; offset: number; hasMore: boolean }>(`/api/dispensaries${queryString}`);
}
async getDispensary(slug: string) {
@@ -2769,6 +2777,101 @@ class ApiClient {
sampleValues: Record<string, any>;
}>(`/api/seo/templates/variables/${encodeURIComponent(pageType)}`);
}
// ==========================================
// Task Queue API
// ==========================================
async getTasks(params?: {
role?: string;
status?: string;
dispensary_id?: number;
limit?: number;
offset?: number;
}) {
const query = new URLSearchParams();
if (params?.role) query.set('role', params.role);
if (params?.status) query.set('status', params.status);
if (params?.dispensary_id) query.set('dispensary_id', String(params.dispensary_id));
if (params?.limit) query.set('limit', String(params.limit));
if (params?.offset) query.set('offset', String(params.offset));
const qs = query.toString();
return this.request<{ tasks: any[]; count: number }>(`/api/tasks${qs ? '?' + qs : ''}`);
}
async getTask(id: number) {
return this.request<any>(`/api/tasks/${id}`);
}
async getTaskCounts() {
return this.request<{
pending: number;
claimed: number;
running: number;
completed: number;
failed: number;
stale: number;
}>('/api/tasks/counts');
}
async getTaskCapacity() {
return this.request<{ metrics: any[] }>('/api/tasks/capacity');
}
async getRoleCapacity(role: string) {
return this.request<any>(`/api/tasks/capacity/${role}`);
}
async createTask(params: {
role: string;
dispensary_id?: number;
platform?: string;
priority?: number;
scheduled_for?: string;
}) {
return this.request<any>('/api/tasks', {
method: 'POST',
body: JSON.stringify(params),
});
}
async generateResyncTasks(params?: { batches_per_day?: number; date?: string }) {
return this.request<{ success: boolean; tasks_created: number }>('/api/tasks/generate/resync', {
method: 'POST',
body: JSON.stringify(params ?? {}),
});
}
async generateDiscoveryTask(platform: string, stateCode?: string, priority?: number) {
return this.request<any>('/api/tasks/generate/discovery', {
method: 'POST',
body: JSON.stringify({ platform, state_code: stateCode, priority }),
});
}
async recoverStaleTasks(thresholdMinutes?: number) {
return this.request<{ success: boolean; tasks_recovered: number }>('/api/tasks/recover-stale', {
method: 'POST',
body: JSON.stringify({ threshold_minutes: thresholdMinutes }),
});
}
async getLastRoleCompletion(role: string) {
return this.request<{ role: string; last_completion: string | null; time_since: number | null }>(
`/api/tasks/role/${role}/last-completion`
);
}
async getRecentRoleCompletions(role: string, limit?: number) {
const qs = limit ? `?limit=${limit}` : '';
return this.request<{ tasks: any[] }>(`/api/tasks/role/${role}/recent${qs}`);
}
async checkStoreActiveTask(dispensaryId: number) {
return this.request<{ dispensary_id: number; has_active_task: boolean }>(
`/api/tasks/store/${dispensaryId}/active`
);
}
}
export const api = new ApiClient(API_URL);

119
cannaiq/src/lib/images.ts Normal file
View File

@@ -0,0 +1,119 @@
/**
* Image URL utilities for on-demand resizing
*
* Uses the backend's /img proxy endpoint for local images.
* Falls back to original URL for remote images.
*/
const API_BASE = import.meta.env.VITE_API_URL || '';
interface ImageOptions {
width?: number;
height?: number;
quality?: number;
fit?: 'cover' | 'contain' | 'fill' | 'inside' | 'outside';
}
/**
* Check if URL is a local image path
*/
function isLocalImage(url: string): boolean {
return url.startsWith('/images/') || url.startsWith('/img/');
}
/**
* Build an image URL with optional resize parameters
*
* @param imageUrl - Original image URL (local or remote)
* @param options - Resize options
* @returns Optimized image URL
*
* @example
* // Thumbnail (50px)
* getImageUrl(product.image_url, { width: 50 })
*
* // Card image (200px)
* getImageUrl(product.image_url, { width: 200 })
*
* // Detail view (600px)
* getImageUrl(product.image_url, { width: 600 })
*
* // Square crop
* getImageUrl(product.image_url, { width: 200, height: 200, fit: 'cover' })
*/
export function getImageUrl(
imageUrl: string | null | undefined,
options: ImageOptions = {}
): string | null {
if (!imageUrl) return null;
// For remote images (AWS, Dutchie CDN, etc.), return as-is
// These can't be resized by our proxy
if (imageUrl.startsWith('http://') || imageUrl.startsWith('https://')) {
return imageUrl;
}
// For local images, use the /img proxy with resize params
if (isLocalImage(imageUrl)) {
// Convert /images/ path to /img/ proxy path
let proxyPath = imageUrl;
if (imageUrl.startsWith('/images/')) {
proxyPath = imageUrl.replace('/images/', '/img/');
}
// Build query params
const params = new URLSearchParams();
if (options.width) params.set('w', String(options.width));
if (options.height) params.set('h', String(options.height));
if (options.quality) params.set('q', String(options.quality));
if (options.fit) params.set('fit', options.fit);
const queryString = params.toString();
const url = queryString ? `${proxyPath}?${queryString}` : proxyPath;
// Prepend API base if needed
return API_BASE ? `${API_BASE}${url}` : url;
}
// Unknown format, return as-is
return imageUrl;
}
/**
* Preset sizes for common use cases
*/
export const ImageSizes = {
/** Tiny thumbnail for lists (50px) */
thumb: { width: 50 },
/** Small card (100px) */
small: { width: 100 },
/** Medium card (200px) */
medium: { width: 200 },
/** Large card (400px) */
large: { width: 400 },
/** Detail view (600px) */
detail: { width: 600 },
/** Full size (no resize) */
full: {},
} as const;
/**
* Convenience function for thumbnail
*/
export function getThumbUrl(imageUrl: string | null | undefined): string | null {
return getImageUrl(imageUrl, ImageSizes.thumb);
}
/**
* Convenience function for card images
*/
export function getCardUrl(imageUrl: string | null | undefined): string | null {
return getImageUrl(imageUrl, ImageSizes.medium);
}
/**
* Convenience function for detail images
*/
export function getDetailUrl(imageUrl: string | null | undefined): string | null {
return getImageUrl(imageUrl, ImageSizes.detail);
}

View File

@@ -18,7 +18,11 @@ import {
Globe,
MapPin,
ArrowRight,
BarChart3
BarChart3,
ListChecks,
Play,
CheckCircle2,
XCircle
} from 'lucide-react';
import {
LineChart,
@@ -41,6 +45,7 @@ export function Dashboard() {
const [refreshing, setRefreshing] = useState(false);
const [pendingChangesCount, setPendingChangesCount] = useState(0);
const [showNotification, setShowNotification] = useState(false);
const [taskCounts, setTaskCounts] = useState<Record<string, number> | null>(null);
useEffect(() => {
loadData();
@@ -119,6 +124,15 @@ export function Dashboard() {
// National stats not critical, just skip
setNationalStats(null);
}
// Fetch task queue counts
try {
const counts = await api.getTaskCounts();
setTaskCounts(counts);
} catch {
// Task counts not critical, just skip
setTaskCounts(null);
}
} catch (error) {
console.error('Failed to load dashboard:', error);
} finally {
@@ -471,6 +485,60 @@ export function Dashboard() {
</div>
)}
{/* Task Queue Summary */}
{taskCounts && (
<div className="bg-white rounded-xl border border-gray-200 p-4 sm:p-6">
<div className="flex items-center justify-between mb-4">
<div className="flex items-center gap-3">
<div className="p-2 bg-violet-50 rounded-lg">
<ListChecks className="w-5 h-5 text-violet-600" />
</div>
<div>
<h3 className="text-sm sm:text-base font-semibold text-gray-900">Task Queue</h3>
<p className="text-xs text-gray-500">Worker task processing status</p>
</div>
</div>
<button
onClick={() => navigate('/tasks')}
className="flex items-center gap-1 text-sm text-violet-600 hover:text-violet-700"
>
View Dashboard
<ArrowRight className="w-4 h-4" />
</button>
</div>
<div className="grid grid-cols-2 md:grid-cols-4 gap-4">
<div className="p-3 bg-amber-50 rounded-lg">
<div className="flex items-center gap-2">
<Clock className="w-4 h-4 text-amber-600" />
<span className="text-xs text-gray-500">Pending</span>
</div>
<div className="text-xl font-bold text-amber-600 mt-1">{taskCounts.pending || 0}</div>
</div>
<div className="p-3 bg-blue-50 rounded-lg">
<div className="flex items-center gap-2">
<Play className="w-4 h-4 text-blue-600" />
<span className="text-xs text-gray-500">Running</span>
</div>
<div className="text-xl font-bold text-blue-600 mt-1">{(taskCounts.claimed || 0) + (taskCounts.running || 0)}</div>
</div>
<div className="p-3 bg-emerald-50 rounded-lg">
<div className="flex items-center gap-2">
<CheckCircle2 className="w-4 h-4 text-emerald-600" />
<span className="text-xs text-gray-500">Completed</span>
</div>
<div className="text-xl font-bold text-emerald-600 mt-1">{taskCounts.completed || 0}</div>
</div>
<div className="p-3 bg-red-50 rounded-lg">
<div className="flex items-center gap-2">
<XCircle className="w-4 h-4 text-red-600" />
<span className="text-xs text-gray-500">Failed</span>
</div>
<div className="text-xl font-bold text-red-600 mt-1">{(taskCounts.failed || 0) + (taskCounts.stale || 0)}</div>
</div>
</div>
</div>
)}
{/* Activity Lists */}
<div className="grid grid-cols-1 lg:grid-cols-2 gap-4 sm:gap-6">
{/* Recent Scrapes */}

View File

@@ -1,33 +1,71 @@
import React, { useEffect, useState } from 'react';
import React, { useEffect, useState, useCallback } from 'react';
import { useNavigate } from 'react-router-dom';
import { Layout } from '../components/Layout';
import { api } from '../lib/api';
import { Building2, Phone, Mail, MapPin, ExternalLink, Search, Eye, Pencil, X, Save } from 'lucide-react';
import { Building2, Phone, Mail, MapPin, ExternalLink, Search, Eye, Pencil, X, Save, ChevronLeft, ChevronRight } from 'lucide-react';
const PAGE_SIZE = 50;
export function Dispensaries() {
const navigate = useNavigate();
const [dispensaries, setDispensaries] = useState<any[]>([]);
const [loading, setLoading] = useState(true);
const [searchTerm, setSearchTerm] = useState('');
const [filterCity, setFilterCity] = useState('');
const [debouncedSearch, setDebouncedSearch] = useState('');
const [filterState, setFilterState] = useState('');
const [editingDispensary, setEditingDispensary] = useState<any | null>(null);
const [editForm, setEditForm] = useState<any>({});
const [total, setTotal] = useState(0);
const [offset, setOffset] = useState(0);
const [hasMore, setHasMore] = useState(false);
const [states, setStates] = useState<string[]>([]);
// Debounce search
useEffect(() => {
loadDispensaries();
const timer = setTimeout(() => {
setDebouncedSearch(searchTerm);
setOffset(0); // Reset to first page on search
}, 300);
return () => clearTimeout(timer);
}, [searchTerm]);
// Load states once for filter dropdown
useEffect(() => {
const loadStates = async () => {
try {
const data = await api.getDispensaries({ limit: 500, crawl_enabled: 'all' });
const uniqueStates = Array.from(new Set(data.dispensaries.map((d: any) => d.state).filter(Boolean))).sort() as string[];
setStates(uniqueStates);
} catch (error) {
console.error('Failed to load states:', error);
}
};
loadStates();
}, []);
const loadDispensaries = async () => {
const loadDispensaries = useCallback(async () => {
setLoading(true);
try {
const data = await api.getDispensaries();
const data = await api.getDispensaries({
limit: PAGE_SIZE,
offset,
search: debouncedSearch || undefined,
state: filterState || undefined,
crawl_enabled: 'all'
});
setDispensaries(data.dispensaries);
setTotal(data.total);
setHasMore(data.hasMore);
} catch (error) {
console.error('Failed to load dispensaries:', error);
} finally {
setLoading(false);
}
};
}, [offset, debouncedSearch, filterState]);
useEffect(() => {
loadDispensaries();
}, [loadDispensaries]);
const handleEdit = (dispensary: any) => {
setEditingDispensary(dispensary);
@@ -59,17 +97,18 @@ export function Dispensaries() {
setEditForm({});
};
const filteredDispensaries = dispensaries.filter(disp => {
const searchLower = searchTerm.toLowerCase();
const matchesSearch = !searchTerm ||
disp.name.toLowerCase().includes(searchLower) ||
(disp.company_name && disp.company_name.toLowerCase().includes(searchLower)) ||
(disp.dba_name && disp.dba_name.toLowerCase().includes(searchLower));
const matchesCity = !filterCity || disp.city === filterCity;
return matchesSearch && matchesCity;
});
const currentPage = Math.floor(offset / PAGE_SIZE) + 1;
const totalPages = Math.ceil(total / PAGE_SIZE);
const cities = Array.from(new Set(dispensaries.map(d => d.city).filter(Boolean))).sort();
const goToPage = (page: number) => {
const newOffset = (page - 1) * PAGE_SIZE;
setOffset(newOffset);
};
const handleStateFilter = (state: string) => {
setFilterState(state);
setOffset(0); // Reset to first page
};
return (
<Layout>
@@ -78,7 +117,7 @@ export function Dispensaries() {
<div>
<h1 className="text-2xl font-bold text-gray-900">Dispensaries</h1>
<p className="text-sm text-gray-600 mt-1">
AZDHS official dispensary directory ({dispensaries.length} total)
USA and Canada Dispensary Directory ({total} total)
</p>
</div>
@@ -102,16 +141,16 @@ export function Dispensaries() {
</div>
<div>
<label className="block text-sm font-medium text-gray-700 mb-2">
Filter by City
Filter by State
</label>
<select
value={filterCity}
onChange={(e) => setFilterCity(e.target.value)}
value={filterState}
onChange={(e) => handleStateFilter(e.target.value)}
className="w-full px-3 py-2 border border-gray-300 rounded-lg focus:ring-2 focus:ring-blue-500 focus:border-blue-500"
>
<option value="">All Cities</option>
{cities.map(city => (
<option key={city} value={city}>{city}</option>
<option value="">All States</option>
{states.map(state => (
<option key={state} value={state}>{state}</option>
))}
</select>
</div>
@@ -133,9 +172,6 @@ export function Dispensaries() {
<th className="px-4 py-3 text-left text-xs font-medium text-gray-700 uppercase tracking-wider">
Name
</th>
<th className="px-4 py-3 text-left text-xs font-medium text-gray-700 uppercase tracking-wider">
Company
</th>
<th className="px-4 py-3 text-left text-xs font-medium text-gray-700 uppercase tracking-wider">
Address
</th>
@@ -157,14 +193,14 @@ export function Dispensaries() {
</tr>
</thead>
<tbody className="divide-y divide-gray-200">
{filteredDispensaries.length === 0 ? (
{dispensaries.length === 0 ? (
<tr>
<td colSpan={8} className="px-4 py-8 text-center text-sm text-gray-500">
<td colSpan={7} className="px-4 py-8 text-center text-sm text-gray-500">
No dispensaries found
</td>
</tr>
) : (
filteredDispensaries.map((disp) => (
dispensaries.map((disp) => (
<tr key={disp.id} className="hover:bg-gray-50">
<td className="px-4 py-3">
<div className="flex items-center gap-2">
@@ -181,13 +217,10 @@ export function Dispensaries() {
</div>
</div>
</td>
<td className="px-4 py-3">
<span className="text-sm text-gray-600">{disp.company_name || '-'}</span>
</td>
<td className="px-4 py-3">
<div className="flex items-start gap-1">
<MapPin className="w-3 h-3 text-gray-400 flex-shrink-0 mt-0.5" />
<span className="text-sm text-gray-600">{disp.address || '-'}</span>
<span className="text-sm text-gray-600">{disp.address1 || '-'}</span>
</div>
</td>
<td className="px-4 py-3">
@@ -266,10 +299,33 @@ export function Dispensaries() {
</table>
</div>
{/* Footer */}
{/* Footer with Pagination */}
<div className="bg-gray-50 px-4 py-3 border-t border-gray-200">
<div className="text-sm text-gray-600">
Showing {filteredDispensaries.length} of {dispensaries.length} dispensaries
<div className="flex items-center justify-between">
<div className="text-sm text-gray-600">
Showing {offset + 1}-{Math.min(offset + dispensaries.length, total)} of {total} dispensaries
</div>
<div className="flex items-center gap-2">
<button
onClick={() => goToPage(currentPage - 1)}
disabled={currentPage === 1}
className="inline-flex items-center gap-1 px-3 py-1.5 text-sm font-medium text-gray-700 bg-white border border-gray-300 rounded-lg hover:bg-gray-50 disabled:opacity-50 disabled:cursor-not-allowed"
>
<ChevronLeft className="w-4 h-4" />
Prev
</button>
<span className="text-sm text-gray-600">
Page {currentPage} of {totalPages}
</span>
<button
onClick={() => goToPage(currentPage + 1)}
disabled={!hasMore}
className="inline-flex items-center gap-1 px-3 py-1.5 text-sm font-medium text-gray-700 bg-white border border-gray-300 rounded-lg hover:bg-gray-50 disabled:opacity-50 disabled:cursor-not-allowed"
>
Next
<ChevronRight className="w-4 h-4" />
</button>
</div>
</div>
</div>
</div>

View File

@@ -2,6 +2,7 @@ import { useEffect, useState } from 'react';
import { useParams, useNavigate, Link } from 'react-router-dom';
import { Layout } from '../components/Layout';
import { api } from '../lib/api';
import { getImageUrl, ImageSizes } from '../lib/images';
import {
Building2,
Phone,
@@ -497,7 +498,7 @@ export function DispensaryDetail() {
<td className="whitespace-nowrap">
{product.image_url ? (
<img
src={product.image_url}
src={getImageUrl(product.image_url, ImageSizes.thumb) || product.image_url}
alt={product.name}
className="w-12 h-12 object-cover rounded"
onError={(e) => e.currentTarget.style.display = 'none'}
@@ -686,7 +687,7 @@ export function DispensaryDetail() {
<div className="flex items-start gap-3">
{special.image_url && (
<img
src={special.image_url}
src={getImageUrl(special.image_url, ImageSizes.small) || special.image_url}
alt={special.name}
className="w-16 h-16 object-cover rounded"
onError={(e) => e.currentTarget.style.display = 'none'}

View File

@@ -20,7 +20,7 @@ import {
} from 'lucide-react';
const API_URL = import.meta.env.VITE_API_URL || '';
const PLUGIN_DOWNLOAD_URL = `${API_URL}/downloads/cannaiq-menus-1.5.3.zip`;
const PLUGIN_DOWNLOAD_URL = `${API_URL}/downloads/cannaiq-menus-latest.zip`;
import { api } from '../lib/api';
interface VersionInfo {

View File

@@ -23,7 +23,7 @@ export default function LandingPage() {
<Link to="/login" className="px-8 py-3 bg-white text-emerald-700 font-semibold rounded-lg hover:bg-gray-100 transition-colors shadow-lg">
Sign In
</Link>
<a href="/downloads/cannaiq-menus-1.5.3.zip" className="px-8 py-3 border-2 border-white text-white font-semibold rounded-lg hover:bg-white hover:text-emerald-700 transition-colors">
<a href="/downloads/cannaiq-menus-latest.zip" className="px-8 py-3 border-2 border-white text-white font-semibold rounded-lg hover:bg-white hover:text-emerald-700 transition-colors">
Download WordPress Plugin
</a>
</div>
@@ -84,10 +84,10 @@ export default function LandingPage() {
<div className="text-emerald-400">[cannaiq_product id="123"]</div>
</div>
<a
href="/downloads/cannaiq-menus-1.5.3.zip"
href="/downloads/cannaiq-menus-latest.zip"
className="inline-block px-8 py-3 bg-emerald-600 text-white font-semibold rounded-lg hover:bg-emerald-700 transition-colors shadow-lg"
>
Download CannaIQ Menus v1.5.3
Download CannaIQ Menus Plugin
</a>
</div>
</div>

View File

@@ -23,6 +23,7 @@ import {
ArrowUpCircle,
} from 'lucide-react';
import { StoreOrchestratorPanel } from '../components/StoreOrchestratorPanel';
import { DeployStatus } from '../components/DeployStatus';
interface CrawlHealth {
status: 'ok' | 'degraded' | 'stale' | 'error';
@@ -286,6 +287,9 @@ export function OrchestratorDashboard() {
</div>
</div>
{/* Deploy Status Panel */}
<DeployStatus />
{/* Metrics Cards - Clickable - Responsive: 2→3→4→7 columns */}
{metrics && (
<div className="grid grid-cols-2 sm:grid-cols-3 md:grid-cols-4 xl:grid-cols-7 gap-3 md:gap-4">

View File

@@ -3,6 +3,7 @@ import { Layout } from '../components/Layout';
import { Package, ArrowLeft, TrendingUp, TrendingDown, DollarSign, Search, Filter, ChevronDown, X, LineChart } from 'lucide-react';
import { useNavigate, useSearchParams } from 'react-router-dom';
import { api } from '../lib/api';
import { getImageUrl, ImageSizes } from '../lib/images';
interface Product {
id: number;
@@ -324,7 +325,7 @@ export function OrchestratorProducts() {
<div className="flex items-center gap-3">
{product.image_url ? (
<img
src={product.image_url}
src={getImageUrl(product.image_url, ImageSizes.thumb) || product.image_url}
alt={product.name}
className="w-10 h-10 rounded object-cover"
/>
@@ -395,7 +396,7 @@ export function OrchestratorProducts() {
<div className="flex items-center gap-4">
{selectedProduct.image_url ? (
<img
src={selectedProduct.image_url}
src={getImageUrl(selectedProduct.image_url, ImageSizes.small) || selectedProduct.image_url}
alt={selectedProduct.name}
className="w-16 h-16 rounded object-cover"
/>

View File

@@ -3,6 +3,7 @@ import { Layout } from '../components/Layout';
import { Scale, Search, Package, Store, Trophy, TrendingDown, TrendingUp, MapPin } from 'lucide-react';
import { useNavigate, useSearchParams } from 'react-router-dom';
import { api } from '../lib/api';
import { getImageUrl, ImageSizes } from '../lib/images';
interface CompareResult {
product_id: number;
@@ -311,7 +312,7 @@ export function PriceCompare() {
<div className="flex items-center gap-3">
{item.image_url ? (
<img
src={item.image_url}
src={getImageUrl(item.image_url, ImageSizes.thumb) || item.image_url}
alt={item.product_name}
className="w-10 h-10 rounded object-cover"
/>

View File

@@ -2,6 +2,7 @@ import { useEffect, useState } from 'react';
import { useParams, useNavigate } from 'react-router-dom';
import { Layout } from '../components/Layout';
import { api } from '../lib/api';
import { getImageUrl, ImageSizes } from '../lib/images';
import { ArrowLeft, ExternalLink, Package, Code, Copy, CheckCircle, FileJson, TrendingUp, TrendingDown, Minus, BarChart3 } from 'lucide-react';
export function ProductDetail() {
@@ -114,14 +115,9 @@ export function ProductDetail() {
const metadata = product.metadata || {};
const getImageUrl = () => {
if (product.image_url_full) return product.image_url_full;
if (product.medium_path) return `/api/images/dutchie/${product.medium_path}`;
if (product.thumbnail_path) return `/api/images/dutchie/${product.thumbnail_path}`;
return null;
};
const imageUrl = getImageUrl();
// Use the centralized image URL helper for on-demand resizing
const productImageUrl = product.image_url_full || product.image_url || product.medium_path || product.thumbnail_path;
const imageUrl = getImageUrl(productImageUrl, ImageSizes.detail);
return (
<Layout>

View File

@@ -2,6 +2,7 @@ import { useEffect, useState } from 'react';
import { useSearchParams, useNavigate } from 'react-router-dom';
import { Layout } from '../components/Layout';
import { api } from '../lib/api';
import { getImageUrl, ImageSizes } from '../lib/images';
export function Products() {
const [searchParams, setSearchParams] = useSearchParams();
@@ -417,9 +418,9 @@ function ProductCard({ product, onViewDetails }: { product: any; onViewDetails:
onMouseEnter={(e) => e.currentTarget.style.transform = 'translateY(-4px)'}
onMouseLeave={(e) => e.currentTarget.style.transform = 'translateY(0)'}
>
{product.image_url_full ? (
{(product.image_url_full || product.image_url) ? (
<img
src={product.image_url_full}
src={getImageUrl(product.image_url_full || product.image_url, ImageSizes.medium) || product.image_url_full || product.image_url}
alt={product.name}
style={{
width: '100%',

View File

@@ -3,6 +3,7 @@ import { Layout } from '../components/Layout';
import { Tag, Package, Store, Percent, Search, Filter, ArrowUpDown, ExternalLink } from 'lucide-react';
import { useNavigate, useSearchParams } from 'react-router-dom';
import { api } from '../lib/api';
import { getImageUrl, ImageSizes } from '../lib/images';
interface Special {
variant_id: number;
@@ -284,7 +285,7 @@ export function Specials() {
<div className="relative">
{special.image_url ? (
<img
src={special.image_url}
src={getImageUrl(special.image_url, ImageSizes.medium) || special.image_url}
alt={special.product_name}
className="w-full h-32 object-cover"
/>

View File

@@ -121,12 +121,13 @@ export default function StateHeatmap() {
try {
const response = await api.get(`/api/analytics/national/heatmap?metric=${selectedMetric}`);
// Response structure: { success, data: { metric, heatmap } }
if (response.data?.data?.heatmap) {
setHeatmapData(response.data.data.heatmap);
} else if (response.data?.heatmap) {
// Fallback for direct structure
setHeatmapData(response.data.heatmap);
}
let rawData = response.data?.data?.heatmap || response.data?.heatmap || [];
// Ensure values are numbers (PostgreSQL bigint can come as strings)
const parsedData = rawData.map((d: any) => ({
...d,
value: typeof d.value === 'string' ? parseFloat(d.value) : (d.value || 0),
}));
setHeatmapData(parsedData);
} catch (err: any) {
setError(err.message || 'Failed to load heatmap data');
} finally {

View File

@@ -2,6 +2,7 @@ import { useEffect, useState } from 'react';
import { useParams, useNavigate } from 'react-router-dom';
import { Layout } from '../components/Layout';
import { api } from '../lib/api';
import { getImageUrl as getResizedImageUrl, ImageSizes } from '../lib/images';
import {
Package, Tag, Zap, Clock, ExternalLink, CheckCircle, XCircle,
AlertCircle, Building, MapPin, RefreshCw, Calendar, Activity
@@ -101,9 +102,10 @@ export function StoreDetail() {
};
const getImageUrl = (product: any) => {
if (product.image_url_full) return product.image_url_full;
if (product.medium_path) return `/api/images/dutchie/${product.medium_path}`;
if (product.thumbnail_path) return `/api/images/dutchie/${product.thumbnail_path}`;
const rawUrl = product.image_url_full || product.image_url || product.medium_path || product.thumbnail_path;
if (rawUrl) {
return getResizedImageUrl(rawUrl, ImageSizes.medium) || rawUrl;
}
return 'https://via.placeholder.com/300x300?text=No+Image';
};

View File

@@ -3,6 +3,7 @@ import { useParams, useNavigate } from 'react-router-dom';
import { Layout } from '../components/Layout';
import { api } from '../lib/api';
import { trackProductView } from '../lib/analytics';
import { getImageUrl, ImageSizes } from '../lib/images';
import {
Building2,
Phone,
@@ -470,7 +471,7 @@ export function StoreDetailPage() {
<td className="whitespace-nowrap">
{product.image_url ? (
<img
src={product.image_url}
src={getImageUrl(product.image_url, ImageSizes.thumb) || product.image_url}
alt={product.name}
className="w-12 h-12 object-cover rounded"
onError={(e) => e.currentTarget.style.display = 'none'}

View File

@@ -0,0 +1,525 @@
import { useState, useEffect } from 'react';
import { api } from '../lib/api';
import { Layout } from '../components/Layout';
import {
ListChecks,
Clock,
CheckCircle2,
XCircle,
AlertTriangle,
PlayCircle,
RefreshCw,
Search,
ChevronDown,
ChevronUp,
Gauge,
Users,
Calendar,
Zap,
} from 'lucide-react';
interface Task {
id: number;
role: string;
dispensary_id: number | null;
dispensary_name?: string;
platform: string | null;
status: string;
priority: number;
scheduled_for: string | null;
worker_id: string | null;
claimed_at: string | null;
started_at: string | null;
completed_at: string | null;
error_message: string | null;
retry_count: number;
created_at: string;
duration_sec?: number;
}
interface CapacityMetric {
role: string;
pending_tasks: number;
ready_tasks: number;
claimed_tasks: number;
running_tasks: number;
completed_last_hour: number;
failed_last_hour: number;
active_workers: number;
avg_duration_sec: number | null;
tasks_per_worker_hour: number | null;
estimated_hours_to_drain: number | null;
workers_needed?: {
for_1_hour: number;
for_4_hours: number;
for_8_hours: number;
};
}
interface TaskCounts {
pending: number;
claimed: number;
running: number;
completed: number;
failed: number;
stale: number;
}
const ROLES = [
'store_discovery',
'entry_point_discovery',
'product_discovery',
'product_resync',
'analytics_refresh',
];
const STATUS_COLORS: Record<string, string> = {
pending: 'bg-yellow-100 text-yellow-800',
claimed: 'bg-blue-100 text-blue-800',
running: 'bg-indigo-100 text-indigo-800',
completed: 'bg-green-100 text-green-800',
failed: 'bg-red-100 text-red-800',
stale: 'bg-gray-100 text-gray-800',
};
const STATUS_ICONS: Record<string, React.ReactNode> = {
pending: <Clock className="w-4 h-4" />,
claimed: <PlayCircle className="w-4 h-4" />,
running: <RefreshCw className="w-4 h-4 animate-spin" />,
completed: <CheckCircle2 className="w-4 h-4" />,
failed: <XCircle className="w-4 h-4" />,
stale: <AlertTriangle className="w-4 h-4" />,
};
function formatDuration(seconds: number | null): string {
if (seconds === null) return '-';
if (seconds < 60) return `${Math.round(seconds)}s`;
if (seconds < 3600) return `${Math.floor(seconds / 60)}m ${Math.round(seconds % 60)}s`;
return `${Math.floor(seconds / 3600)}h ${Math.floor((seconds % 3600) / 60)}m`;
}
function formatTimeAgo(dateStr: string | null): string {
if (!dateStr) return '-';
const date = new Date(dateStr);
const now = new Date();
const diff = (now.getTime() - date.getTime()) / 1000;
if (diff < 60) return `${Math.round(diff)}s ago`;
if (diff < 3600) return `${Math.floor(diff / 60)}m ago`;
if (diff < 86400) return `${Math.floor(diff / 3600)}h ago`;
return `${Math.floor(diff / 86400)}d ago`;
}
export default function TasksDashboard() {
const [tasks, setTasks] = useState<Task[]>([]);
const [counts, setCounts] = useState<TaskCounts | null>(null);
const [capacity, setCapacity] = useState<CapacityMetric[]>([]);
const [loading, setLoading] = useState(true);
const [error, setError] = useState<string | null>(null);
// Filters
const [roleFilter, setRoleFilter] = useState<string>('');
const [statusFilter, setStatusFilter] = useState<string>('');
const [searchQuery, setSearchQuery] = useState('');
const [showCapacity, setShowCapacity] = useState(true);
// Actions
const [actionLoading, setActionLoading] = useState(false);
const [actionMessage, setActionMessage] = useState<string | null>(null);
const fetchData = async () => {
try {
const [tasksRes, countsRes, capacityRes] = await Promise.all([
api.getTasks({
role: roleFilter || undefined,
status: statusFilter || undefined,
limit: 100,
}),
api.getTaskCounts(),
api.getTaskCapacity(),
]);
setTasks(tasksRes.tasks || []);
setCounts(countsRes);
setCapacity(capacityRes.metrics || []);
setError(null);
} catch (err: any) {
setError(err.message || 'Failed to load tasks');
} finally {
setLoading(false);
}
};
useEffect(() => {
fetchData();
const interval = setInterval(fetchData, 10000); // Refresh every 10 seconds
return () => clearInterval(interval);
}, [roleFilter, statusFilter]);
const handleGenerateResync = async () => {
setActionLoading(true);
try {
const result = await api.generateResyncTasks();
setActionMessage(`Generated ${result.tasks_created} resync tasks`);
fetchData();
} catch (err: any) {
setActionMessage(`Error: ${err.message}`);
} finally {
setActionLoading(false);
setTimeout(() => setActionMessage(null), 5000);
}
};
const handleRecoverStale = async () => {
setActionLoading(true);
try {
const result = await api.recoverStaleTasks();
setActionMessage(`Recovered ${result.tasks_recovered} stale tasks`);
fetchData();
} catch (err: any) {
setActionMessage(`Error: ${err.message}`);
} finally {
setActionLoading(false);
setTimeout(() => setActionMessage(null), 5000);
}
};
const filteredTasks = tasks.filter((task) => {
if (searchQuery) {
const query = searchQuery.toLowerCase();
return (
task.role.toLowerCase().includes(query) ||
task.dispensary_name?.toLowerCase().includes(query) ||
task.worker_id?.toLowerCase().includes(query) ||
String(task.id).includes(query)
);
}
return true;
});
const totalActive = (counts?.claimed || 0) + (counts?.running || 0);
const totalPending = counts?.pending || 0;
if (loading) {
return (
<Layout>
<div className="flex items-center justify-center h-64">
<RefreshCw className="w-8 h-8 animate-spin text-emerald-600" />
</div>
</Layout>
);
}
return (
<Layout>
<div className="space-y-6">
{/* Header */}
<div className="flex flex-col sm:flex-row sm:items-center sm:justify-between gap-4">
<div>
<h1 className="text-2xl font-bold text-gray-900 flex items-center gap-2">
<ListChecks className="w-7 h-7 text-emerald-600" />
Task Queue
</h1>
<p className="text-gray-500 mt-1">
{totalActive} active, {totalPending} pending tasks
</p>
</div>
<div className="flex gap-2">
<button
onClick={handleGenerateResync}
disabled={actionLoading}
className="flex items-center gap-2 px-4 py-2 bg-emerald-600 text-white rounded-lg hover:bg-emerald-700 disabled:opacity-50"
>
<Calendar className="w-4 h-4" />
Generate Resync
</button>
<button
onClick={handleRecoverStale}
disabled={actionLoading}
className="flex items-center gap-2 px-4 py-2 bg-gray-600 text-white rounded-lg hover:bg-gray-700 disabled:opacity-50"
>
<Zap className="w-4 h-4" />
Recover Stale
</button>
<button
onClick={fetchData}
className="flex items-center gap-2 px-4 py-2 bg-gray-100 text-gray-700 rounded-lg hover:bg-gray-200"
>
<RefreshCw className="w-4 h-4" />
Refresh
</button>
</div>
</div>
{/* Action Message */}
{actionMessage && (
<div
className={`p-4 rounded-lg ${
actionMessage.startsWith('Error')
? 'bg-red-50 text-red-700'
: 'bg-green-50 text-green-700'
}`}
>
{actionMessage}
</div>
)}
{error && (
<div className="p-4 bg-red-50 text-red-700 rounded-lg">{error}</div>
)}
{/* Status Summary Cards */}
<div className="grid grid-cols-2 sm:grid-cols-3 lg:grid-cols-6 gap-4">
{Object.entries(counts || {}).map(([status, count]) => (
<div
key={status}
className={`p-4 rounded-lg border ${
statusFilter === status ? 'ring-2 ring-emerald-500' : ''
} cursor-pointer hover:shadow-md transition-shadow`}
onClick={() => setStatusFilter(statusFilter === status ? '' : status)}
>
<div className="flex items-center gap-2 mb-2">
<span className={`p-1.5 rounded ${STATUS_COLORS[status]}`}>
{STATUS_ICONS[status]}
</span>
<span className="text-sm font-medium text-gray-600 capitalize">{status}</span>
</div>
<div className="text-2xl font-bold text-gray-900">{count}</div>
</div>
))}
</div>
{/* Capacity Planning Section */}
<div className="bg-white rounded-lg border border-gray-200 overflow-hidden">
<button
onClick={() => setShowCapacity(!showCapacity)}
className="w-full flex items-center justify-between p-4 hover:bg-gray-50"
>
<div className="flex items-center gap-2">
<Gauge className="w-5 h-5 text-emerald-600" />
<span className="font-medium text-gray-900">Capacity Planning</span>
</div>
{showCapacity ? (
<ChevronUp className="w-5 h-5 text-gray-400" />
) : (
<ChevronDown className="w-5 h-5 text-gray-400" />
)}
</button>
{showCapacity && (
<div className="p-4 border-t border-gray-200">
{capacity.length === 0 ? (
<p className="text-gray-500 text-center py-4">No capacity data available</p>
) : (
<div className="overflow-x-auto">
<table className="min-w-full divide-y divide-gray-200">
<thead>
<tr>
<th className="px-4 py-3 text-left text-xs font-medium text-gray-500 uppercase">
Role
</th>
<th className="px-4 py-3 text-right text-xs font-medium text-gray-500 uppercase">
Pending
</th>
<th className="px-4 py-3 text-right text-xs font-medium text-gray-500 uppercase">
Running
</th>
<th className="px-4 py-3 text-right text-xs font-medium text-gray-500 uppercase">
Active Workers
</th>
<th className="px-4 py-3 text-right text-xs font-medium text-gray-500 uppercase">
Avg Duration
</th>
<th className="px-4 py-3 text-right text-xs font-medium text-gray-500 uppercase">
Tasks/Worker/Hr
</th>
<th className="px-4 py-3 text-right text-xs font-medium text-gray-500 uppercase">
Est. Drain Time
</th>
<th className="px-4 py-3 text-right text-xs font-medium text-gray-500 uppercase">
Completed/Hr
</th>
<th className="px-4 py-3 text-right text-xs font-medium text-gray-500 uppercase">
Failed/Hr
</th>
</tr>
</thead>
<tbody className="divide-y divide-gray-200">
{capacity.map((metric) => (
<tr key={metric.role} className="hover:bg-gray-50">
<td className="px-4 py-3 text-sm font-medium text-gray-900">
{metric.role.replace(/_/g, ' ')}
</td>
<td className="px-4 py-3 text-sm text-right text-gray-600">
{metric.pending_tasks}
</td>
<td className="px-4 py-3 text-sm text-right text-gray-600">
{metric.running_tasks}
</td>
<td className="px-4 py-3 text-sm text-right">
<span className="inline-flex items-center gap-1">
<Users className="w-4 h-4 text-gray-400" />
{metric.active_workers}
</span>
</td>
<td className="px-4 py-3 text-sm text-right text-gray-600">
{formatDuration(metric.avg_duration_sec)}
</td>
<td className="px-4 py-3 text-sm text-right text-gray-600">
{metric.tasks_per_worker_hour?.toFixed(1) || '-'}
</td>
<td className="px-4 py-3 text-sm text-right">
{metric.estimated_hours_to_drain ? (
<span
className={
metric.estimated_hours_to_drain > 4
? 'text-red-600 font-medium'
: 'text-gray-600'
}
>
{metric.estimated_hours_to_drain.toFixed(1)}h
</span>
) : (
'-'
)}
</td>
<td className="px-4 py-3 text-sm text-right text-green-600">
{metric.completed_last_hour}
</td>
<td className="px-4 py-3 text-sm text-right text-red-600">
{metric.failed_last_hour}
</td>
</tr>
))}
</tbody>
</table>
</div>
)}
</div>
)}
</div>
{/* Filters */}
<div className="flex flex-col sm:flex-row gap-4">
<div className="relative flex-1">
<Search className="absolute left-3 top-1/2 -translate-y-1/2 w-5 h-5 text-gray-400" />
<input
type="text"
placeholder="Search tasks..."
value={searchQuery}
onChange={(e) => setSearchQuery(e.target.value)}
className="w-full pl-10 pr-4 py-2 border border-gray-300 rounded-lg focus:ring-2 focus:ring-emerald-500 focus:border-emerald-500"
/>
</div>
<select
value={roleFilter}
onChange={(e) => setRoleFilter(e.target.value)}
className="px-4 py-2 border border-gray-300 rounded-lg focus:ring-2 focus:ring-emerald-500"
>
<option value="">All Roles</option>
{ROLES.map((role) => (
<option key={role} value={role}>
{role.replace(/_/g, ' ')}
</option>
))}
</select>
<select
value={statusFilter}
onChange={(e) => setStatusFilter(e.target.value)}
className="px-4 py-2 border border-gray-300 rounded-lg focus:ring-2 focus:ring-emerald-500"
>
<option value="">All Statuses</option>
<option value="pending">Pending</option>
<option value="claimed">Claimed</option>
<option value="running">Running</option>
<option value="completed">Completed</option>
<option value="failed">Failed</option>
<option value="stale">Stale</option>
</select>
</div>
{/* Tasks Table */}
<div className="bg-white rounded-lg border border-gray-200 overflow-hidden">
<div className="overflow-x-auto">
<table className="min-w-full divide-y divide-gray-200">
<thead className="bg-gray-50">
<tr>
<th className="px-4 py-3 text-left text-xs font-medium text-gray-500 uppercase">
ID
</th>
<th className="px-4 py-3 text-left text-xs font-medium text-gray-500 uppercase">
Role
</th>
<th className="px-4 py-3 text-left text-xs font-medium text-gray-500 uppercase">
Store
</th>
<th className="px-4 py-3 text-left text-xs font-medium text-gray-500 uppercase">
Status
</th>
<th className="px-4 py-3 text-left text-xs font-medium text-gray-500 uppercase">
Worker
</th>
<th className="px-4 py-3 text-left text-xs font-medium text-gray-500 uppercase">
Duration
</th>
<th className="px-4 py-3 text-left text-xs font-medium text-gray-500 uppercase">
Created
</th>
<th className="px-4 py-3 text-left text-xs font-medium text-gray-500 uppercase">
Error
</th>
</tr>
</thead>
<tbody className="divide-y divide-gray-200">
{filteredTasks.length === 0 ? (
<tr>
<td colSpan={8} className="px-4 py-8 text-center text-gray-500">
No tasks found
</td>
</tr>
) : (
filteredTasks.map((task) => (
<tr key={task.id} className="hover:bg-gray-50">
<td className="px-4 py-3 text-sm font-mono text-gray-600">#{task.id}</td>
<td className="px-4 py-3 text-sm text-gray-900">
{task.role.replace(/_/g, ' ')}
</td>
<td className="px-4 py-3 text-sm text-gray-600">
{task.dispensary_name || task.dispensary_id || '-'}
</td>
<td className="px-4 py-3">
<span
className={`inline-flex items-center gap-1 px-2 py-1 rounded-full text-xs font-medium ${
STATUS_COLORS[task.status]
}`}
>
{STATUS_ICONS[task.status]}
{task.status}
</span>
</td>
<td className="px-4 py-3 text-sm font-mono text-gray-600">
{task.worker_id?.split('-').slice(-1)[0] || '-'}
</td>
<td className="px-4 py-3 text-sm text-gray-600">
{formatDuration(task.duration_sec ?? null)}
</td>
<td className="px-4 py-3 text-sm text-gray-500">
{formatTimeAgo(task.created_at)}
</td>
<td className="px-4 py-3 text-sm text-red-600 max-w-xs truncate">
{task.error_message || '-'}
</td>
</tr>
))
)}
</tbody>
</table>
</div>
</div>
</div>
</Layout>
);
}

View File

@@ -141,13 +141,21 @@ export function Users() {
};
const canModifyUser = (user: User) => {
// Can't modify yourself
if (currentUser?.id === user.id) return false;
// Only superadmin can modify superadmin users
if (user.role === 'superadmin' && currentUser?.role !== 'superadmin') return false;
return true;
};
const canDeleteUser = (user: User) => {
// Can't delete yourself
if (currentUser?.id === user.id) return false;
// Only superadmin can delete superadmin users
if (user.role === 'superadmin' && currentUser?.role !== 'superadmin') return false;
return true;
};
const isEditingSelf = (user: User) => currentUser?.id === user.id;
return (
<Layout>
<div className="space-y-6">
@@ -236,15 +244,17 @@ export function Users() {
{new Date(user.created_at).toLocaleDateString()}
</td>
<td className="px-6 py-4 whitespace-nowrap text-right text-sm font-medium">
{canModifyUser(user) ? (
<div className="flex items-center justify-end gap-2">
<div className="flex items-center justify-end gap-2">
{canModifyUser(user) && (
<button
onClick={() => openEditModal(user)}
className="p-1.5 text-gray-400 hover:text-blue-600 hover:bg-blue-50 rounded transition-colors"
title="Edit user"
title={isEditingSelf(user) ? "Edit your profile" : "Edit user"}
>
<Pencil className="w-4 h-4" />
</button>
)}
{canDeleteUser(user) ? (
<button
onClick={() => handleDelete(user)}
className="p-1.5 text-gray-400 hover:text-red-600 hover:bg-red-50 rounded transition-colors"
@@ -252,10 +262,10 @@ export function Users() {
>
<Trash2 className="w-4 h-4" />
</button>
</div>
) : (
<span className="text-xs text-gray-400"></span>
)}
) : !canModifyUser(user) && (
<span className="text-xs text-gray-400"></span>
)}
</div>
</td>
</tr>
))}
@@ -349,11 +359,15 @@ export function Users() {
<div>
<label className="block text-sm font-medium text-gray-700 mb-1">
Role
{editingUser && currentUser?.id === editingUser.id && (
<span className="ml-2 text-xs text-gray-400 font-normal">(cannot change your own role)</span>
)}
</label>
<select
value={formData.role}
onChange={(e) => setFormData({ ...formData, role: e.target.value })}
className="w-full px-3 py-2 border border-gray-300 rounded-lg focus:ring-2 focus:ring-blue-500 focus:border-blue-500"
disabled={editingUser !== null && currentUser?.id === editingUser.id}
className="w-full px-3 py-2 border border-gray-300 rounded-lg focus:ring-2 focus:ring-blue-500 focus:border-blue-500 disabled:bg-gray-100 disabled:cursor-not-allowed"
>
<option value="viewer">Viewer</option>
<option value="analyst">Analyst</option>

View File

@@ -1,5 +1,5 @@
# Build stage
FROM node:20-slim AS builder
FROM code.cannabrands.app/creationshop/node:20-slim AS builder
WORKDIR /app
@@ -12,14 +12,15 @@ RUN npm install
# Copy source files
COPY . .
# Set build-time environment variable for API URL (CRA uses REACT_APP_ prefix)
ENV REACT_APP_API_URL=https://api.findadispo.com
# Note: REACT_APP_API_URL is intentionally NOT set here
# The frontend uses relative URLs (same domain) in production
# API calls go to /api/* which the ingress routes to the backend
# Build the app (CRA produces /build, not /dist)
RUN npm run build
# Production stage
FROM nginx:alpine
FROM code.cannabrands.app/creationshop/nginx:alpine
# Copy built assets from builder stage (CRA outputs to /build)
COPY --from=builder /app/build /usr/share/nginx/html

View File

@@ -1,5 +1,5 @@
# Build stage
FROM node:20-slim AS builder
FROM code.cannabrands.app/creationshop/node:20-slim AS builder
WORKDIR /app
@@ -12,14 +12,15 @@ RUN npm install
# Copy source files
COPY . .
# Set build-time environment variable for API URL (CRA uses REACT_APP_ prefix)
ENV REACT_APP_API_URL=https://api.findagram.co
# Note: REACT_APP_API_URL is intentionally NOT set here
# The frontend uses relative URLs (same domain) in production
# API calls go to /api/* which the ingress routes to the backend
# Build the app (CRA produces /build, not /dist)
RUN npm run build
# Production stage
FROM nginx:alpine
FROM code.cannabrands.app/creationshop/nginx:alpine
# Copy built assets from builder stage (CRA outputs to /build)
COPY --from=builder /app/build /usr/share/nginx/html

View File

@@ -1,11 +1,11 @@
/**
* Findagram API Client
*
* Connects to the backend /api/az/* endpoints which are publicly accessible.
* Connects to the backend /api/v1/* public endpoints.
* Uses REACT_APP_API_URL environment variable for the base URL.
*
* Local development: http://localhost:3010
* Production: https://findagram.co (proxied to backend via ingress)
* Production: https://cannaiq.co (shared API backend)
*/
const API_BASE_URL = process.env.REACT_APP_API_URL || '';
@@ -70,14 +70,14 @@ export async function getProducts(params = {}) {
offset: params.offset || 0,
});
return request(`/api/az/products${queryString}`);
return request(`/api/v1/products${queryString}`);
}
/**
* Get a single product by ID
*/
export async function getProduct(id) {
return request(`/api/az/products/${id}`);
return request(`/api/v1/products/${id}`);
}
/**
@@ -103,7 +103,7 @@ export async function getProductAvailability(productId, params = {}) {
max_radius_miles: maxRadiusMiles,
});
return request(`/api/az/products/${productId}/availability${queryString}`);
return request(`/api/v1/products/${productId}/availability${queryString}`);
}
/**
@@ -113,7 +113,7 @@ export async function getProductAvailability(productId, params = {}) {
* @returns {Promise<{similarProducts: Array<{productId: number, name: string, brandName: string, imageUrl: string, price: number}>}>}
*/
export async function getSimilarProducts(productId) {
return request(`/api/az/products/${productId}/similar`);
return request(`/api/v1/products/${productId}/similar`);
}
/**
@@ -130,7 +130,7 @@ export async function getStoreProducts(storeId, params = {}) {
offset: params.offset || 0,
});
return request(`/api/az/stores/${storeId}/products${queryString}`);
return request(`/api/v1/stores/${storeId}/products${queryString}`);
}
// ============================================================
@@ -154,42 +154,42 @@ export async function getDispensaries(params = {}) {
offset: params.offset || 0,
});
return request(`/api/az/stores${queryString}`);
return request(`/api/v1/stores${queryString}`);
}
/**
* Get a single dispensary by ID
*/
export async function getDispensary(id) {
return request(`/api/az/stores/${id}`);
return request(`/api/v1/stores/${id}`);
}
/**
* Get dispensary by slug or platform ID
*/
export async function getDispensaryBySlug(slug) {
return request(`/api/az/stores/slug/${slug}`);
return request(`/api/v1/stores/slug/${slug}`);
}
/**
* Get dispensary summary (product counts, categories, brands)
*/
export async function getDispensarySummary(id) {
return request(`/api/az/stores/${id}/summary`);
return request(`/api/v1/stores/${id}/summary`);
}
/**
* Get brands available at a specific dispensary
*/
export async function getDispensaryBrands(id) {
return request(`/api/az/stores/${id}/brands`);
return request(`/api/v1/stores/${id}/brands`);
}
/**
* Get categories available at a specific dispensary
*/
export async function getDispensaryCategories(id) {
return request(`/api/az/stores/${id}/categories`);
return request(`/api/v1/stores/${id}/categories`);
}
// ============================================================
@@ -200,7 +200,7 @@ export async function getDispensaryCategories(id) {
* Get all categories with product counts
*/
export async function getCategories() {
return request('/api/az/categories');
return request('/api/v1/categories');
}
// ============================================================
@@ -220,7 +220,7 @@ export async function getBrands(params = {}) {
offset: params.offset || 0,
});
return request(`/api/az/brands${queryString}`);
return request(`/api/v1/brands${queryString}`);
}
// ============================================================

View File

@@ -7,3 +7,5 @@ data:
NODE_ENV: "production"
PORT: "3010"
LOG_LEVEL: "info"
REDIS_HOST: "redis"
REDIS_PORT: "6379"

66
k8s/redis.yaml Normal file
View File

@@ -0,0 +1,66 @@
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: redis-data
namespace: dispensary-scraper
spec:
accessModes:
- ReadWriteOnce
resources:
requests:
storage: 1Gi
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: redis
namespace: dispensary-scraper
spec:
replicas: 1
selector:
matchLabels:
app: redis
template:
metadata:
labels:
app: redis
spec:
containers:
- name: redis
image: redis:7-alpine
ports:
- containerPort: 6379
resources:
requests:
memory: "64Mi"
cpu: "50m"
limits:
memory: "256Mi"
cpu: "200m"
volumeMounts:
- name: redis-data
mountPath: /data
command:
- redis-server
- --appendonly
- "yes"
- --maxmemory
- "200mb"
- --maxmemory-policy
- allkeys-lru
volumes:
- name: redis-data
persistentVolumeClaim:
claimName: redis-data
---
apiVersion: v1
kind: Service
metadata:
name: redis
namespace: dispensary-scraper
spec:
selector:
app: redis
ports:
- port: 6379
targetPort: 6379

View File

@@ -1,12 +1,12 @@
# Dutchie AZ Worker Deployment
# These workers poll the job queue and process crawl jobs.
# Scale this deployment to increase crawl throughput.
# Hydration Worker Deployment
# These workers process raw_payloads → canonical tables.
# Scale this deployment to increase hydration throughput.
#
# Architecture:
# - The main 'scraper' deployment runs the API server + scheduler (1 replica)
# - This 'scraper-worker' deployment runs workers that poll and claim jobs (5 replicas)
# - Workers use DB-level locking (FOR UPDATE SKIP LOCKED) to prevent double-crawls
# - Each worker sends heartbeats; stale jobs are recovered automatically
# - This 'scraper-worker' deployment runs hydration workers (5 replicas)
# - Workers use DB-level locking to prevent double-processing
# - Each worker processes payloads in batches with configurable limits
apiVersion: apps/v1
kind: Deployment
metadata:
@@ -27,9 +27,9 @@ spec:
containers:
- name: worker
image: code.cannabrands.app/creationshop/dispensary-scraper:latest
# Run the worker process instead of the main server
# Run the hydration worker in loop mode
command: ["node"]
args: ["dist/dutchie-az/services/worker.js"]
args: ["dist/scripts/run-hydration.js", "--mode=payload", "--loop"]
envFrom:
- configMapRef:
name: scraper-config
@@ -57,9 +57,9 @@ spec:
command:
- /bin/sh
- -c
- "pgrep -f 'worker.js' > /dev/null"
- "pgrep -f 'run-hydration' > /dev/null"
initialDelaySeconds: 10
periodSeconds: 30
failureThreshold: 3
# Graceful shutdown - give workers time to complete current job
# Graceful shutdown - give workers time to complete current batch
terminationGracePeriodSeconds: 60

1
wordpress-plugin/VERSION Normal file
View File

@@ -0,0 +1 @@
1.5.4

View File

@@ -36,9 +36,16 @@ zip -r "${OUTPUT_DIR}/${OUTPUT_FILE}" . \
-x "assets/css/crawlsy-menus.css" \
-x "assets/js/crawlsy-menus.js"
# Create/update the "latest" symlink
cd "${OUTPUT_DIR}"
rm -f cannaiq-menus-latest.zip
ln -s "${OUTPUT_FILE}" cannaiq-menus-latest.zip
echo ""
echo "Build complete!"
echo " File: ${OUTPUT_DIR}/${OUTPUT_FILE}"
echo " Size: $(ls -lh "${OUTPUT_DIR}/${OUTPUT_FILE}" | awk '{print $5}')"
echo ""
echo "Download URL: https://cannaiq.co/downloads/cannaiq-menus-${VERSION}.zip"
echo "Download URLs:"
echo " Versioned: https://cannaiq.co/downloads/cannaiq-menus-${VERSION}.zip"
echo " Latest: https://cannaiq.co/downloads/cannaiq-menus-latest.zip"

View File

@@ -3,7 +3,7 @@
* Plugin Name: CannaIQ Menus
* Plugin URI: https://cannaiq.co
* Description: Display cannabis product menus from CannaIQ with Elementor integration. Real-time menu data updated daily.
* Version: 1.5.3
* Version: 1.5.4
* Author: CannaIQ
* Author URI: https://cannaiq.co
* License: GPL v2 or later
@@ -15,7 +15,7 @@ if (!defined('ABSPATH')) {
exit; // Exit if accessed directly
}
define('CANNAIQ_MENUS_VERSION', '1.5.3');
define('CANNAIQ_MENUS_VERSION', '1.5.4');
define('CANNAIQ_MENUS_API_URL', 'https://cannaiq.co/api/v1');
define('CANNAIQ_MENUS_PLUGIN_DIR', plugin_dir_path(__FILE__));
define('CANNAIQ_MENUS_PLUGIN_URL', plugin_dir_url(__FILE__));

View File

@@ -3,7 +3,7 @@
* Plugin Name: Crawlsy Menus
* Plugin URI: https://creationshop.io
* Description: Display cannabis product menus from Crawlsy with Elementor integration
* Version: 1.5.2
* Version: 1.5.4
* Author: Creationshop
* Author URI: https://creationshop.io
* License: GPL v2 or later
@@ -15,7 +15,7 @@ if (!defined('ABSPATH')) {
exit; // Exit if accessed directly
}
define('CRAWLSY_MENUS_VERSION', '1.5.2');
define('CRAWLSY_MENUS_VERSION', '1.5.4');
define('CRAWLSY_MENUS_API_URL', 'https://cannaiq.co/api/v1');
define('CRAWLSY_MENUS_PLUGIN_DIR', plugin_dir_path(__FILE__));
define('CRAWLSY_MENUS_PLUGIN_URL', plugin_dir_url(__FILE__));