From af859a85f9b9b99c8d600356a41b2162288844d5 Mon Sep 17 00:00:00 2001 From: Kelly Date: Sun, 14 Dec 2025 15:53:04 -0700 Subject: [PATCH] feat: Add Real-Time Inventory Tracking infrastructure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements per-store high-frequency crawl scheduling and inventory snapshot tracking for sales velocity estimation (Hoodie Analytics parity). Database migrations: - 117: Per-store crawl_interval_minutes and next_crawl_at columns - 118: inventory_snapshots table (30-day retention) - 119: product_visibility_events table for OOS/brand alerts (90-day) Backend changes: - inventory-snapshots.ts: Shared utility normalizing Dutchie/Jane/Treez - visibility-events.ts: Detects OOS, price changes, brand drops - task-scheduler.ts: checkHighFrequencyStores() runs every 60s - Handler updates: 2-line additions to save snapshots/events API endpoints: - GET /api/tasks/schedules/high-frequency - PUT /api/tasks/schedules/high-frequency/:id - DELETE /api/tasks/schedules/high-frequency/:id Frontend: - TasksDashboard: Per-Store Schedules section with stats Features: - Per-store intervals (15/30/60 min configurable) - Jitter (0-20%) to avoid detection patterns - Cross-platform support (Dutchie, Jane, Treez) - No crawler core changes - scheduling/post-crawl only 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- backend/Dockerfile | 8 +- .../117_per_store_crawl_interval.sql | 32 ++ .../migrations/118_inventory_snapshots.sql | 48 +++ .../119_product_visibility_events.sql | 53 +++ backend/src/routes/tasks.ts | 73 ++++ backend/src/services/inventory-snapshots.ts | 252 ++++++++++++ backend/src/services/task-scheduler.ts | 189 +++++++++ backend/src/services/visibility-events.ts | 374 ++++++++++++++++++ .../handlers/product-discovery-dutchie.ts | 9 + .../tasks/handlers/product-discovery-jane.ts | 8 + .../tasks/handlers/product-discovery-treez.ts | 13 +- backend/src/tasks/task-service.ts | 2 + cannaiq/src/lib/api.ts | 52 +++ cannaiq/src/pages/TasksDashboard.tsx | 170 +++++++- findagram/frontend/Dockerfile | 9 +- 15 files changed, 1284 insertions(+), 8 deletions(-) create mode 100644 backend/migrations/117_per_store_crawl_interval.sql create mode 100644 backend/migrations/118_inventory_snapshots.sql create mode 100644 backend/migrations/119_product_visibility_events.sql create mode 100644 backend/src/services/inventory-snapshots.ts create mode 100644 backend/src/services/visibility-events.ts diff --git a/backend/Dockerfile b/backend/Dockerfile index ed7451dc..67eaf5f6 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -12,7 +12,13 @@ RUN apt-get update && apt-get install -y \ WORKDIR /app COPY package*.json ./ -RUN npm install + +# Install dependencies with retry and fallback registry +RUN npm config set fetch-retries 3 && \ + npm config set fetch-retry-mintimeout 20000 && \ + npm config set fetch-retry-maxtimeout 120000 && \ + npm install || \ + (npm config set registry https://registry.npmmirror.com && npm install) COPY . . RUN npm run build diff --git a/backend/migrations/117_per_store_crawl_interval.sql b/backend/migrations/117_per_store_crawl_interval.sql new file mode 100644 index 00000000..2a69a755 --- /dev/null +++ b/backend/migrations/117_per_store_crawl_interval.sql @@ -0,0 +1,32 @@ +-- Migration 117: Per-store crawl interval scheduling +-- Adds columns for configurable per-store crawl intervals +-- Part of Real-Time Inventory Tracking feature + +-- Per-store crawl interval (NULL = use state schedule default 4h) +ALTER TABLE dispensaries ADD COLUMN IF NOT EXISTS crawl_interval_minutes INT DEFAULT NULL; + +-- When this store should next be crawled (used by high-frequency scheduler) +ALTER TABLE dispensaries ADD COLUMN IF NOT EXISTS next_crawl_at TIMESTAMPTZ DEFAULT NULL; + +-- Track last request time to enforce minimum spacing +ALTER TABLE dispensaries ADD COLUMN IF NOT EXISTS last_crawl_started_at TIMESTAMPTZ DEFAULT NULL; + +-- Change tracking for optimization +ALTER TABLE dispensaries ADD COLUMN IF NOT EXISTS last_inventory_hash TEXT DEFAULT NULL; +ALTER TABLE dispensaries ADD COLUMN IF NOT EXISTS last_price_hash TEXT DEFAULT NULL; +ALTER TABLE dispensaries ADD COLUMN IF NOT EXISTS inventory_changes_24h INT DEFAULT 0; +ALTER TABLE dispensaries ADD COLUMN IF NOT EXISTS price_changes_24h INT DEFAULT 0; + +-- Index for scheduler query: find stores due for high-frequency crawl +CREATE INDEX IF NOT EXISTS idx_dispensaries_next_crawl +ON dispensaries(next_crawl_at) +WHERE crawl_interval_minutes IS NOT NULL AND crawl_enabled = TRUE; + +-- Comment for documentation +COMMENT ON COLUMN dispensaries.crawl_interval_minutes IS 'Custom crawl interval in minutes. NULL = use state schedule (4h default). Set to 15/30/60 for high-frequency tracking.'; +COMMENT ON COLUMN dispensaries.next_crawl_at IS 'When this store should next be crawled. Updated after each crawl with interval + jitter.'; +COMMENT ON COLUMN dispensaries.last_crawl_started_at IS 'When the last crawl task was created. Used to enforce minimum spacing.'; +COMMENT ON COLUMN dispensaries.last_inventory_hash IS 'Hash of inventory state from last crawl. Used to detect changes and skip unchanged payloads.'; +COMMENT ON COLUMN dispensaries.last_price_hash IS 'Hash of price state from last crawl. Used to detect price changes.'; +COMMENT ON COLUMN dispensaries.inventory_changes_24h IS 'Number of inventory changes detected in last 24h. Indicates store volatility.'; +COMMENT ON COLUMN dispensaries.price_changes_24h IS 'Number of price changes detected in last 24h.'; diff --git a/backend/migrations/118_inventory_snapshots.sql b/backend/migrations/118_inventory_snapshots.sql new file mode 100644 index 00000000..78a67a88 --- /dev/null +++ b/backend/migrations/118_inventory_snapshots.sql @@ -0,0 +1,48 @@ +-- Migration 118: Inventory snapshots table +-- Lightweight per-product tracking for sales velocity estimation +-- Part of Real-Time Inventory Tracking feature + +CREATE TABLE IF NOT EXISTS inventory_snapshots ( + id BIGSERIAL PRIMARY KEY, + dispensary_id INT NOT NULL REFERENCES dispensaries(id) ON DELETE CASCADE, + product_id TEXT NOT NULL, -- provider_product_id (normalized across platforms) + captured_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + + -- Platform (for debugging/filtering) + platform TEXT NOT NULL, -- 'dutchie' | 'jane' | 'treez' + + -- Inventory fields (normalized from all platforms) + quantity_available INT, -- Dutchie: quantityAvailable, Jane: quantity, Treez: quantityAvailable + is_below_threshold BOOLEAN, -- Dutchie: isBelowThreshold, Jane: computed, Treez: lowInventory + status TEXT, -- Active/Inactive/available + + -- Price fields (normalized) + price_rec NUMERIC(10,2), -- recreational price + price_med NUMERIC(10,2), -- medical price (if different) + + -- Denormalized for fast queries + brand_name TEXT, + category TEXT, + product_name TEXT +); + +-- Primary query: get snapshots for a store over time +CREATE INDEX idx_inv_snap_store_time ON inventory_snapshots(dispensary_id, captured_at DESC); + +-- Delta calculation: get consecutive snapshots for a product +CREATE INDEX idx_inv_snap_product_time ON inventory_snapshots(dispensary_id, product_id, captured_at DESC); + +-- Brand-level analytics +CREATE INDEX idx_inv_snap_brand_time ON inventory_snapshots(brand_name, captured_at DESC) WHERE brand_name IS NOT NULL; + +-- Platform filtering +CREATE INDEX idx_inv_snap_platform ON inventory_snapshots(platform, captured_at DESC); + +-- Retention cleanup (30 days) +CREATE INDEX idx_inv_snap_cleanup ON inventory_snapshots(captured_at) WHERE captured_at < NOW() - INTERVAL '30 days'; + +-- Comments +COMMENT ON TABLE inventory_snapshots IS 'Lightweight inventory snapshots for sales velocity tracking. Retained 30 days.'; +COMMENT ON COLUMN inventory_snapshots.product_id IS 'Provider product ID, normalized across platforms'; +COMMENT ON COLUMN inventory_snapshots.platform IS 'Menu platform: dutchie, jane, or treez'; +COMMENT ON COLUMN inventory_snapshots.quantity_available IS 'Current quantity in stock (Dutchie: quantityAvailable, Jane: quantity)'; diff --git a/backend/migrations/119_product_visibility_events.sql b/backend/migrations/119_product_visibility_events.sql new file mode 100644 index 00000000..1d8d8fd3 --- /dev/null +++ b/backend/migrations/119_product_visibility_events.sql @@ -0,0 +1,53 @@ +-- Migration 119: Product visibility events table +-- Tracks OOS, brand drops, and other notable events for alerts +-- Part of Real-Time Inventory Tracking feature + +CREATE TABLE IF NOT EXISTS product_visibility_events ( + id SERIAL PRIMARY KEY, + dispensary_id INT NOT NULL REFERENCES dispensaries(id) ON DELETE CASCADE, + + -- Product identification (null for brand-level events) + product_id TEXT, -- provider_product_id + product_name TEXT, -- For display in alerts + + -- Brand (always populated) + brand_name TEXT, + + -- Event details + event_type TEXT NOT NULL, -- 'oos', 'back_in_stock', 'brand_dropped', 'brand_added', 'price_change' + detected_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + + -- Context + previous_quantity INT, -- For OOS events: what quantity was before + previous_price NUMERIC(10,2), -- For price change events + new_price NUMERIC(10,2), -- For price change events + price_change_pct NUMERIC(5,2), -- Percentage change (e.g., -15.5 for 15.5% decrease) + + -- Platform + platform TEXT, -- 'dutchie' | 'jane' | 'treez' + + -- Alert status + notified BOOLEAN DEFAULT FALSE, -- Has external system been notified? + acknowledged_at TIMESTAMPTZ, -- When user acknowledged the alert + acknowledged_by TEXT -- User who acknowledged +); + +-- Primary query: recent events by store +CREATE INDEX idx_vis_events_store_time ON product_visibility_events(dispensary_id, detected_at DESC); + +-- Alert queries: unnotified events +CREATE INDEX idx_vis_events_unnotified ON product_visibility_events(notified, detected_at DESC) WHERE notified = FALSE; + +-- Event type filtering +CREATE INDEX idx_vis_events_type ON product_visibility_events(event_type, detected_at DESC); + +-- Brand-level queries +CREATE INDEX idx_vis_events_brand ON product_visibility_events(brand_name, event_type, detected_at DESC) WHERE brand_name IS NOT NULL; + +-- Cleanup (90 days retention) +CREATE INDEX idx_vis_events_cleanup ON product_visibility_events(detected_at) WHERE detected_at < NOW() - INTERVAL '90 days'; + +-- Comments +COMMENT ON TABLE product_visibility_events IS 'Notable inventory events for alerting. OOS, brand drops, significant price changes. Retained 90 days.'; +COMMENT ON COLUMN product_visibility_events.event_type IS 'Event type: oos (out of stock), back_in_stock, brand_dropped, brand_added, price_change'; +COMMENT ON COLUMN product_visibility_events.notified IS 'Whether external systems (other apps) have been notified of this event'; diff --git a/backend/src/routes/tasks.ts b/backend/src/routes/tasks.ts index c93e7eed..1b85948e 100644 --- a/backend/src/routes/tasks.ts +++ b/backend/src/routes/tasks.ts @@ -56,6 +56,7 @@ import { openTaskPool, getTaskPoolStatus, } from '../tasks/task-pool-state'; +import { taskScheduler } from '../services/task-scheduler'; const router = Router(); @@ -642,6 +643,78 @@ router.post('/schedules/:id/toggle', async (req: Request, res: Response) => { } }); +// ============================================================ +// HIGH-FREQUENCY (PER-STORE) SCHEDULE ROUTES +// Part of Real-Time Inventory Tracking feature +// ============================================================ + +/** + * GET /api/tasks/schedules/high-frequency + * List all stores with high-frequency crawl intervals + */ +router.get('/schedules/high-frequency', async (req: Request, res: Response) => { + try { + const stores = await taskScheduler.getHighFrequencyStores(); + const stats = await taskScheduler.getHighFrequencyStats(); + res.json({ stores, stats }); + } catch (error: unknown) { + console.error('Error listing high-frequency schedules:', error); + res.status(500).json({ error: 'Failed to list high-frequency schedules' }); + } +}); + +/** + * PUT /api/tasks/schedules/high-frequency/:dispensaryId + * Set crawl interval for a specific store + * + * Body: + * - interval_minutes: number (15, 30, 60, etc.) or null to remove + */ +router.put('/schedules/high-frequency/:dispensaryId', async (req: Request, res: Response) => { + try { + const dispensaryId = parseInt(req.params.dispensaryId, 10); + const { interval_minutes } = req.body; + + if (interval_minutes !== null && (typeof interval_minutes !== 'number' || interval_minutes < 15)) { + return res.status(400).json({ error: 'Interval must be at least 15 minutes' }); + } + + await taskScheduler.setStoreInterval(dispensaryId, interval_minutes); + + res.json({ + success: true, + dispensary_id: dispensaryId, + interval_minutes, + message: interval_minutes + ? `Store ${dispensaryId} set to crawl every ${interval_minutes} minutes` + : `High-frequency crawling disabled for store ${dispensaryId}`, + }); + } catch (error: unknown) { + console.error('Error setting store interval:', error); + res.status(500).json({ error: 'Failed to set store interval' }); + } +}); + +/** + * DELETE /api/tasks/schedules/high-frequency/:dispensaryId + * Remove high-frequency crawl interval for a store + */ +router.delete('/schedules/high-frequency/:dispensaryId', async (req: Request, res: Response) => { + try { + const dispensaryId = parseInt(req.params.dispensaryId, 10); + await taskScheduler.setStoreInterval(dispensaryId, null); + + res.json({ + success: true, + dispensary_id: dispensaryId, + message: `High-frequency crawling disabled for store ${dispensaryId}`, + }); + } catch (error: unknown) { + console.error('Error removing store interval:', error); + res.status(500).json({ error: 'Failed to remove store interval' }); + } +}); + // ============================================================ // TASK-SPECIFIC ROUTES (with :id parameter) // ============================================================ diff --git a/backend/src/services/inventory-snapshots.ts b/backend/src/services/inventory-snapshots.ts new file mode 100644 index 00000000..6ec6cdc5 --- /dev/null +++ b/backend/src/services/inventory-snapshots.ts @@ -0,0 +1,252 @@ +/** + * Inventory Snapshots Service + * + * Shared utility for saving lightweight inventory snapshots after each crawl. + * Normalizes fields across all platforms (Dutchie, Jane, Treez) into a + * common format for sales velocity tracking and analytics. + * + * Part of Real-Time Inventory Tracking feature. + * + * Field mappings: + * | Field | Dutchie | Jane | Treez | + * |-----------|------------------------|--------------------|------------------| + * | ID | id | product_id | id | + * | Quantity | children.quantityAvailable | max_cart_quantity | availableUnits | + * | Low stock | isBelowThreshold | false | !isAboveThreshold| + * | Price rec | recPrices[0] | bucket_price | customMinPrice | + * | Brand | brand.name | brand | brand | + * | Category | category | kind | category | + * | Name | Name | name | name | + * | Status | Status | (presence=active) | status | + */ + +import { Pool } from 'pg'; + +export type Platform = 'dutchie' | 'jane' | 'treez'; + +interface SnapshotRow { + product_id: string; + quantity_available: number | null; + is_below_threshold: boolean; + status: string | null; + price_rec: number | null; + price_med: number | null; + brand_name: string | null; + category: string | null; + product_name: string | null; +} + +/** + * Extract a normalized snapshot row from a raw product based on platform. + */ +function normalizeProduct(product: any, platform: Platform): SnapshotRow | null { + let productId: string | null = null; + let quantityAvailable: number | null = null; + let isBelowThreshold = false; + let status: string | null = null; + let priceRec: number | null = null; + let priceMed: number | null = null; + let brandName: string | null = null; + let category: string | null = null; + let productName: string | null = null; + + switch (platform) { + case 'dutchie': { + productId = product.id; + productName = product.Name || product.name; + brandName = product.brand?.name || null; + category = product.category || null; + status = product.Status || 'Active'; + isBelowThreshold = product.isBelowThreshold === true; + + // Quantity: sum from children if available + if (product.children && Array.isArray(product.children)) { + quantityAvailable = product.children.reduce( + (sum: number, child: any) => sum + (child.quantityAvailable || child.quantity || 0), + 0 + ); + } else if (product.quantityAvailable != null) { + quantityAvailable = product.quantityAvailable; + } + + // Prices: prefer recPrices, fall back to Prices + const recPrices = product.recPrices || product.Prices || []; + priceRec = recPrices.length > 0 ? parseFloat(recPrices[0]) : null; + + const medPrices = product.medicalPrices || product.medPrices || []; + priceMed = medPrices.length > 0 ? parseFloat(medPrices[0]) : null; + break; + } + + case 'jane': { + productId = String(product.product_id); + productName = product.name; + brandName = product.brand || null; + category = product.kind || null; + status = 'Active'; // Jane products present = active + isBelowThreshold = false; // Jane doesn't expose this + + // Quantity: max_cart_quantity + quantityAvailable = product.max_cart_quantity ?? null; + + // Price: bucket_price or first available weight-based price + priceRec = + product.bucket_price || + product.price_gram || + product.price_eighth_ounce || + product.price_each || + null; + priceMed = null; // Jane doesn't separate med prices clearly + break; + } + + case 'treez': { + productId = product.id; + productName = product.name || product.menuTitle; + brandName = product.brand || null; + category = product.category || null; + status = product.status || (product.isActive ? 'ACTIVE' : 'INACTIVE'); + + // Quantity: availableUnits + quantityAvailable = product.availableUnits ?? null; + + // Low stock: inverse of isAboveThreshold + isBelowThreshold = product.isAboveThreshold === false; + + // Price: customMinPrice + priceRec = product.customMinPrice ?? null; + priceMed = null; // Treez doesn't distinguish med pricing + break; + } + } + + if (!productId) { + return null; + } + + return { + product_id: productId, + quantity_available: quantityAvailable, + is_below_threshold: isBelowThreshold, + status, + price_rec: priceRec, + price_med: priceMed, + brand_name: brandName, + category, + product_name: productName, + }; +} + +/** + * Save inventory snapshots for all products in a crawl result. + * + * Call this after fetching products in any platform handler. + * Uses bulk insert for efficiency. + * + * @param pool - Database connection pool + * @param dispensaryId - The dispensary ID + * @param products - Array of raw products from the platform + * @param platform - The platform type + * @returns Number of snapshots saved + */ +export async function saveInventorySnapshots( + pool: Pool, + dispensaryId: number, + products: any[], + platform: Platform +): Promise { + if (!products || products.length === 0) { + return 0; + } + + const snapshots: SnapshotRow[] = []; + for (const product of products) { + const row = normalizeProduct(product, platform); + if (row) { + snapshots.push(row); + } + } + + if (snapshots.length === 0) { + return 0; + } + + // Bulk insert using VALUES list + // Build parameterized query + const values: any[] = []; + const placeholders: string[] = []; + + let paramIndex = 1; + for (const s of snapshots) { + placeholders.push( + `($${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++})` + ); + values.push( + dispensaryId, + s.product_id, + platform, + s.quantity_available, + s.is_below_threshold, + s.status, + s.price_rec, + s.price_med, + s.brand_name, + s.category, + s.product_name + ); + } + + const query = ` + INSERT INTO inventory_snapshots ( + dispensary_id, + product_id, + platform, + quantity_available, + is_below_threshold, + status, + price_rec, + price_med, + brand_name, + category, + product_name + ) VALUES ${placeholders.join(', ')} + `; + + await pool.query(query, values); + + return snapshots.length; +} + +/** + * Get the previous snapshot for a dispensary (for delta calculation). + * Returns a map of product_id -> snapshot data. + */ +export async function getPreviousSnapshots( + pool: Pool, + dispensaryId: number +): Promise> { + const result = await pool.query( + ` + SELECT DISTINCT ON (product_id) + product_id, + quantity_available, + is_below_threshold, + status, + price_rec, + price_med, + brand_name, + category, + product_name + FROM inventory_snapshots + WHERE dispensary_id = $1 + ORDER BY product_id, captured_at DESC + `, + [dispensaryId] + ); + + const map = new Map(); + for (const row of result.rows) { + map.set(row.product_id, row); + } + return map; +} diff --git a/backend/src/services/task-scheduler.ts b/backend/src/services/task-scheduler.ts index 08db2e0e..b9e463b0 100644 --- a/backend/src/services/task-scheduler.ts +++ b/backend/src/services/task-scheduler.ts @@ -67,10 +67,12 @@ class TaskScheduler { // Per TASK_WORKFLOW_2024-12-10.md: Check immediately on startup await this.checkAndRunDueSchedules(); + await this.checkHighFrequencyStores(); // Per TASK_WORKFLOW_2024-12-10.md: Then poll every 60 seconds this.pollTimer = setInterval(async () => { await this.checkAndRunDueSchedules(); + await this.checkHighFrequencyStores(); }, POLL_INTERVAL_MS); console.log('[TaskScheduler] Started - polling every 60s'); @@ -199,6 +201,90 @@ class TaskScheduler { } } + /** + * Check for stores with custom high-frequency crawl intervals + * These are stores with crawl_interval_minutes set (15min, 30min, 1h, etc.) + * + * Part of Real-Time Inventory Tracking feature. + */ + private async checkHighFrequencyStores(): Promise { + try { + // Find stores that are due for high-frequency crawl + // Uses crawl_interval_minutes (from migration 117) + const result = await pool.query(` + SELECT + d.id, + d.name, + d.menu_type, + d.crawl_interval_minutes, + d.next_crawl_at, + d.last_crawl_started_at + FROM dispensaries d + WHERE d.crawl_enabled = true + AND d.platform_dispensary_id IS NOT NULL + AND d.crawl_interval_minutes IS NOT NULL + AND (d.next_crawl_at IS NULL OR d.next_crawl_at <= NOW()) + -- Enforce minimum spacing (10 min) to prevent hammering + AND ( + d.last_crawl_started_at IS NULL + OR d.last_crawl_started_at < NOW() - INTERVAL '10 minutes' + ) + -- No pending/running crawl task already + AND NOT EXISTS ( + SELECT 1 FROM worker_tasks t + WHERE t.dispensary_id = d.id + AND t.role = 'product_discovery' + AND t.status IN ('pending', 'claimed', 'running') + ) + ORDER BY d.next_crawl_at NULLS FIRST + LIMIT 50 -- Process up to 50 stores per poll cycle + `); + + if (result.rows.length === 0) { + return; + } + + console.log(`[TaskScheduler] ${result.rows.length} high-frequency stores due for crawl`); + + for (const store of result.rows) { + try { + // Determine platform from menu_type + let platform = store.menu_type || 'dutchie'; + if (platform === 'embedded') platform = 'dutchie'; + + // Create product_discovery task + await taskService.createTask({ + role: 'product_discovery', + dispensary_id: store.id, + platform, + method: 'http', + priority: 5, // Higher priority for high-frequency stores + source: 'high_frequency_schedule', + }); + + // Add jitter: interval + random(0, 20% of interval) + const jitterMinutes = Math.floor(Math.random() * (store.crawl_interval_minutes * 0.2)); + const nextIntervalMinutes = store.crawl_interval_minutes + jitterMinutes; + + // Update next_crawl_at and last_crawl_started_at + await pool.query(` + UPDATE dispensaries + SET + next_crawl_at = NOW() + ($1 || ' minutes')::interval, + last_crawl_started_at = NOW() + WHERE id = $2 + `, [nextIntervalMinutes, store.id]); + + console.log(`[TaskScheduler] Created high-freq task for ${store.name} (${platform}), next in ${nextIntervalMinutes}min`); + } catch (err: any) { + console.error(`[TaskScheduler] Failed to create task for store ${store.id}:`, err.message); + } + } + } catch (err: any) { + console.error('[TaskScheduler] Failed to check high-frequency stores:', err.message); + } + } + /** * Execute a schedule and create tasks * Per TASK_WORKFLOW_2024-12-10.md: Different logic per role @@ -589,6 +675,109 @@ class TaskScheduler { return { total: 0, enabled: 0, byRole: {}, byState: {} }; } } + + /** + * Get stores with high-frequency crawl intervals configured + * For Task Scheduler Dashboard display + */ + async getHighFrequencyStores(): Promise<{ + id: number; + name: string; + menu_type: string; + crawl_interval_minutes: number; + next_crawl_at: Date | null; + last_crawl_started_at: Date | null; + last_fetch_at: Date | null; + inventory_changes_24h: number; + price_changes_24h: number; + }[]> { + try { + const result = await pool.query(` + SELECT + id, + name, + COALESCE(menu_type, 'dutchie') as menu_type, + crawl_interval_minutes, + next_crawl_at, + last_crawl_started_at, + last_fetch_at, + COALESCE(inventory_changes_24h, 0) as inventory_changes_24h, + COALESCE(price_changes_24h, 0) as price_changes_24h + FROM dispensaries + WHERE crawl_interval_minutes IS NOT NULL + AND crawl_enabled = true + ORDER BY crawl_interval_minutes, name + `); + return result.rows; + } catch { + return []; + } + } + + /** + * Set high-frequency crawl interval for a store + * @param dispensaryId - Store ID + * @param intervalMinutes - Crawl interval in minutes (15, 30, 60, etc.) or null to remove + */ + async setStoreInterval(dispensaryId: number, intervalMinutes: number | null): Promise { + if (intervalMinutes !== null && intervalMinutes < 15) { + throw new Error('Minimum crawl interval is 15 minutes'); + } + + await pool.query(` + UPDATE dispensaries + SET + crawl_interval_minutes = $1, + next_crawl_at = CASE + WHEN $1 IS NOT NULL THEN NOW() + ELSE NULL + END, + updated_at = NOW() + WHERE id = $2 + `, [intervalMinutes, dispensaryId]); + + console.log(`[TaskScheduler] Set interval for dispensary ${dispensaryId}: ${intervalMinutes}min`); + } + + /** + * Get high-frequency scheduling stats + */ + async getHighFrequencyStats(): Promise<{ + totalStores: number; + byInterval: Record; + byPlatform: Record; + nextDueCount: number; + }> { + try { + const result = await pool.query(` + SELECT + crawl_interval_minutes as interval, + COALESCE(menu_type, 'dutchie') as platform, + COUNT(*)::int as count, + SUM(CASE WHEN next_crawl_at <= NOW() THEN 1 ELSE 0 END)::int as due_count + FROM dispensaries + WHERE crawl_interval_minutes IS NOT NULL + AND crawl_enabled = true + GROUP BY crawl_interval_minutes, menu_type + `); + + let totalStores = 0; + let nextDueCount = 0; + const byInterval: Record = {}; + const byPlatform: Record = {}; + + for (const row of result.rows) { + totalStores += row.count; + nextDueCount += row.due_count; + byInterval[row.interval] = (byInterval[row.interval] || 0) + row.count; + byPlatform[row.platform] = (byPlatform[row.platform] || 0) + row.count; + } + + return { totalStores, byInterval, byPlatform, nextDueCount }; + } catch { + return { totalStores: 0, byInterval: {}, byPlatform: {}, nextDueCount: 0 }; + } + } } // Per TASK_WORKFLOW_2024-12-10.md: Singleton instance diff --git a/backend/src/services/visibility-events.ts b/backend/src/services/visibility-events.ts new file mode 100644 index 00000000..75130df4 --- /dev/null +++ b/backend/src/services/visibility-events.ts @@ -0,0 +1,374 @@ +/** + * Visibility Events Service + * + * Shared utility for detecting notable inventory events: + * - OOS (out of stock) - product disappeared from menu + * - Back in stock - product returned to menu + * - Brand dropped - brand no longer at store + * - Brand added - new brand at store + * - Price change - significant price change (>5%) + * + * Part of Real-Time Inventory Tracking feature. + */ + +import { Pool } from 'pg'; +import { Platform } from './inventory-snapshots'; + +export type EventType = 'oos' | 'back_in_stock' | 'brand_dropped' | 'brand_added' | 'price_change'; + +interface VisibilityEvent { + dispensary_id: number; + product_id: string | null; + product_name: string | null; + brand_name: string | null; + event_type: EventType; + previous_quantity: number | null; + previous_price: number | null; + new_price: number | null; + price_change_pct: number | null; + platform: Platform; +} + +interface ProductInfo { + id: string; + name: string | null; + brand: string | null; + price: number | null; + quantity: number | null; +} + +/** + * Extract product info from raw product based on platform. + */ +function extractProductInfo(product: any, platform: Platform): ProductInfo | null { + let id: string | null = null; + let name: string | null = null; + let brand: string | null = null; + let price: number | null = null; + let quantity: number | null = null; + + switch (platform) { + case 'dutchie': + id = product.id; + name = product.Name || product.name; + brand = product.brand?.name || null; + price = product.recPrices?.[0] || product.Prices?.[0] || null; + if (product.children && Array.isArray(product.children)) { + quantity = product.children.reduce( + (sum: number, child: any) => sum + (child.quantityAvailable || child.quantity || 0), + 0 + ); + } + break; + + case 'jane': + id = String(product.product_id); + name = product.name; + brand = product.brand || null; + price = product.bucket_price || product.price_gram || null; + quantity = product.max_cart_quantity ?? null; + break; + + case 'treez': + id = product.id; + name = product.name || product.menuTitle; + brand = product.brand || null; + price = product.customMinPrice ?? null; + quantity = product.availableUnits ?? null; + break; + } + + if (!id) return null; + + return { id, name, brand, price, quantity }; +} + +/** + * Detect visibility events by comparing current products to previous state. + * + * Call this after fetching products in any platform handler. + * Compares current products to stored state and creates events for changes. + * + * @param pool - Database connection pool + * @param dispensaryId - The dispensary ID + * @param products - Array of raw products from the platform + * @param platform - The platform type + * @returns Number of events created + */ +export async function detectVisibilityEvents( + pool: Pool, + dispensaryId: number, + products: any[], + platform: Platform +): Promise { + if (!products || products.length === 0) { + return 0; + } + + // Get previous product state from store_products + const { rows: previousProducts } = await pool.query( + ` + SELECT + provider_product_id as id, + name, + brand, + price_rec as price + FROM store_products + WHERE dispensary_id = $1 + `, + [dispensaryId] + ); + + // Build maps for comparison + const previousMap = new Map(); + const previousBrands = new Set(); + + for (const p of previousProducts) { + previousMap.set(p.id, { name: p.name, brand: p.brand, price: p.price }); + if (p.brand) previousBrands.add(p.brand); + } + + const currentMap = new Map(); + const currentBrands = new Set(); + + for (const product of products) { + const info = extractProductInfo(product, platform); + if (info) { + currentMap.set(info.id, info); + if (info.brand) currentBrands.add(info.brand); + } + } + + const events: VisibilityEvent[] = []; + + // Detect OOS events (products that disappeared) + Array.from(previousMap.entries()).forEach(([id, prev]) => { + if (!currentMap.has(id)) { + events.push({ + dispensary_id: dispensaryId, + product_id: id, + product_name: prev.name, + brand_name: prev.brand, + event_type: 'oos', + previous_quantity: null, + previous_price: prev.price, + new_price: null, + price_change_pct: null, + platform, + }); + } + }); + + // Detect back_in_stock events (products that appeared) + Array.from(currentMap.entries()).forEach(([id, curr]) => { + if (!previousMap.has(id)) { + events.push({ + dispensary_id: dispensaryId, + product_id: id, + product_name: curr.name, + brand_name: curr.brand, + event_type: 'back_in_stock', + previous_quantity: null, + previous_price: null, + new_price: curr.price, + price_change_pct: null, + platform, + }); + } + }); + + // Detect price changes (>5% change) + Array.from(currentMap.entries()).forEach(([id, curr]) => { + const prev = previousMap.get(id); + if (prev && prev.price != null && curr.price != null) { + const priceDiff = curr.price - prev.price; + const pctChange = (priceDiff / prev.price) * 100; + + if (Math.abs(pctChange) >= 5) { + events.push({ + dispensary_id: dispensaryId, + product_id: id, + product_name: curr.name, + brand_name: curr.brand, + event_type: 'price_change', + previous_quantity: null, + previous_price: prev.price, + new_price: curr.price, + price_change_pct: Math.round(pctChange * 100) / 100, + platform, + }); + } + } + }); + + // Detect brand drops (brands that disappeared from store) + Array.from(previousBrands).forEach((brand) => { + if (!currentBrands.has(brand)) { + events.push({ + dispensary_id: dispensaryId, + product_id: null, + product_name: null, + brand_name: brand, + event_type: 'brand_dropped', + previous_quantity: null, + previous_price: null, + new_price: null, + price_change_pct: null, + platform, + }); + } + }); + + // Detect brand additions (new brands at store) + Array.from(currentBrands).forEach((brand) => { + if (!previousBrands.has(brand)) { + events.push({ + dispensary_id: dispensaryId, + product_id: null, + product_name: null, + brand_name: brand, + event_type: 'brand_added', + previous_quantity: null, + previous_price: null, + new_price: null, + price_change_pct: null, + platform, + }); + } + }); + + if (events.length === 0) { + return 0; + } + + // Bulk insert events + const values: any[] = []; + const placeholders: string[] = []; + + let paramIndex = 1; + for (const e of events) { + placeholders.push( + `($${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++}, $${paramIndex++})` + ); + values.push( + e.dispensary_id, + e.product_id, + e.product_name, + e.brand_name, + e.event_type, + e.previous_quantity, + e.previous_price, + e.new_price, + e.price_change_pct, + e.platform + ); + } + + const query = ` + INSERT INTO product_visibility_events ( + dispensary_id, + product_id, + product_name, + brand_name, + event_type, + previous_quantity, + previous_price, + new_price, + price_change_pct, + platform + ) VALUES ${placeholders.join(', ')} + `; + + await pool.query(query, values); + + console.log(`[VisibilityEvents] Created ${events.length} events for dispensary ${dispensaryId}`); + + return events.length; +} + +/** + * Get recent visibility events for a dispensary. + */ +export async function getRecentEvents( + pool: Pool, + dispensaryId: number, + limit = 100 +): Promise { + const { rows } = await pool.query( + ` + SELECT + id, + product_id, + product_name, + brand_name, + event_type, + detected_at, + previous_price, + new_price, + price_change_pct, + platform, + notified, + acknowledged_at + FROM product_visibility_events + WHERE dispensary_id = $1 + ORDER BY detected_at DESC + LIMIT $2 + `, + [dispensaryId, limit] + ); + + return rows; +} + +/** + * Get unnotified events for external system integration. + */ +export async function getUnnotifiedEvents( + pool: Pool, + limit = 100 +): Promise { + const { rows } = await pool.query( + ` + SELECT + e.id, + e.dispensary_id, + d.name as dispensary_name, + e.product_id, + e.product_name, + e.brand_name, + e.event_type, + e.detected_at, + e.previous_price, + e.new_price, + e.price_change_pct, + e.platform + FROM product_visibility_events e + JOIN dispensaries d ON d.id = e.dispensary_id + WHERE e.notified = FALSE + ORDER BY e.detected_at DESC + LIMIT $1 + `, + [limit] + ); + + return rows; +} + +/** + * Mark events as notified. + */ +export async function markEventsNotified( + pool: Pool, + eventIds: number[] +): Promise { + if (eventIds.length === 0) return; + + await pool.query( + ` + UPDATE product_visibility_events + SET notified = TRUE + WHERE id = ANY($1) + `, + [eventIds] + ); +} diff --git a/backend/src/tasks/handlers/product-discovery-dutchie.ts b/backend/src/tasks/handlers/product-discovery-dutchie.ts index 71dc1436..e345ad34 100644 --- a/backend/src/tasks/handlers/product-discovery-dutchie.ts +++ b/backend/src/tasks/handlers/product-discovery-dutchie.ts @@ -24,6 +24,8 @@ import { TaskContext, TaskResult } from '../task-worker'; import { saveRawPayload } from '../../utils/payload-storage'; import { taskService } from '../task-service'; +import { saveInventorySnapshots } from '../../services/inventory-snapshots'; +import { detectVisibilityEvents } from '../../services/visibility-events'; // GraphQL hash for FilteredProducts query - MUST match CLAUDE.md const FILTERED_PRODUCTS_HASH = 'ee29c060826dc41c527e470e9ae502c9b2c169720faa0a9f5d25e1b9a530a4a0'; @@ -389,6 +391,13 @@ export async function handleProductDiscoveryDutchie(ctx: TaskContext): Promise { const { pool, task, crawlRotator } = ctx; @@ -110,6 +112,12 @@ export async function handleProductDiscoveryJane(ctx: TaskContext): Promise p.raw); + const snapshotCount = await saveInventorySnapshots(pool, dispensaryId, rawProducts, 'jane'); + const eventCount = await detectVisibilityEvents(pool, dispensaryId, rawProducts, 'jane'); + console.log(`[JaneProductDiscovery] Saved ${snapshotCount} inventory snapshots, detected ${eventCount} visibility events`); + // Update dispensary stage and timestamps await pool.query( `UPDATE dispensaries diff --git a/backend/src/tasks/handlers/product-discovery-treez.ts b/backend/src/tasks/handlers/product-discovery-treez.ts index dd5e02ed..dca9e052 100644 --- a/backend/src/tasks/handlers/product-discovery-treez.ts +++ b/backend/src/tasks/handlers/product-discovery-treez.ts @@ -31,6 +31,8 @@ import { } from '../../platforms/treez'; import { saveRawPayload } from '../../utils/payload-storage'; import { taskService } from '../task-service'; +import { saveInventorySnapshots } from '../../services/inventory-snapshots'; +import { detectVisibilityEvents } from '../../services/visibility-events'; export async function handleProductDiscoveryTreez(ctx: TaskContext): Promise { const { pool, task, crawlRotator } = ctx; @@ -127,6 +129,11 @@ export async function handleProductDiscoveryTreez(ctx: TaskContext): Promise product_discovery) diff --git a/cannaiq/src/lib/api.ts b/cannaiq/src/lib/api.ts index 3fba5261..84edc1b6 100755 --- a/cannaiq/src/lib/api.ts +++ b/cannaiq/src/lib/api.ts @@ -3129,6 +3129,45 @@ class ApiClient { ); } + // ============================================================ + // HIGH-FREQUENCY (PER-STORE) SCHEDULES API + // Part of Real-Time Inventory Tracking feature + // ============================================================ + + async getHighFrequencySchedules() { + return this.request<{ + stores: HighFrequencyStore[]; + stats: { + totalStores: number; + byInterval: Record; + byPlatform: Record; + nextDueCount: number; + }; + }>('/api/tasks/schedules/high-frequency'); + } + + async setHighFrequencyInterval(dispensaryId: number, intervalMinutes: number | null) { + return this.request<{ + success: boolean; + dispensary_id: number; + interval_minutes: number | null; + message: string; + }>(`/api/tasks/schedules/high-frequency/${dispensaryId}`, { + method: 'PUT', + body: JSON.stringify({ interval_minutes: intervalMinutes }), + }); + } + + async removeHighFrequencyInterval(dispensaryId: number) { + return this.request<{ + success: boolean; + dispensary_id: number; + message: string; + }>(`/api/tasks/schedules/high-frequency/${dispensaryId}`, { + method: 'DELETE', + }); + } + // ============================================================ // PAYLOADS API // ============================================================ @@ -3232,4 +3271,17 @@ export interface PayloadMetadata { dispensary_name?: string; } +// Type for high-frequency (per-store) schedules +export interface HighFrequencyStore { + id: number; + name: string; + menu_type: string; + crawl_interval_minutes: number; + next_crawl_at: string | null; + last_crawl_started_at: string | null; + last_fetch_at: string | null; + inventory_changes_24h: number; + price_changes_24h: number; +} + export const api = new ApiClient(API_URL); diff --git a/cannaiq/src/pages/TasksDashboard.tsx b/cannaiq/src/pages/TasksDashboard.tsx index 27783731..01d62dd2 100644 --- a/cannaiq/src/pages/TasksDashboard.tsx +++ b/cannaiq/src/pages/TasksDashboard.tsx @@ -1,5 +1,5 @@ import { useState, useEffect } from 'react'; -import { api, TaskSchedule } from '../lib/api'; +import { api, TaskSchedule, HighFrequencyStore } from '../lib/api'; import { Layout } from '../components/Layout'; import { ListChecks, @@ -961,6 +961,16 @@ export default function TasksDashboard() { const [editingSchedule, setEditingSchedule] = useState(null); const [showScheduleModal, setShowScheduleModal] = useState(false); + // High-frequency (per-store) schedules state + const [highFreqStores, setHighFreqStores] = useState([]); + const [highFreqStats, setHighFreqStats] = useState<{ + totalStores: number; + byInterval: Record; + byPlatform: Record; + nextDueCount: number; + }>({ totalStores: 0, byInterval: {}, byPlatform: {}, nextDueCount: 0 }); + const [showHighFreq, setShowHighFreq] = useState(true); + // Pagination const [page, setPage] = useState(0); const tasksPerPage = 25; @@ -982,7 +992,7 @@ export default function TasksDashboard() { const fetchData = async () => { try { - const [tasksRes, countsRes, stateCountsRes, capacityRes, poolStatus, schedulesRes, workersRes, poolsRes] = await Promise.all([ + const [tasksRes, countsRes, stateCountsRes, capacityRes, poolStatus, schedulesRes, workersRes, poolsRes, highFreqRes] = await Promise.all([ api.getTasks({ role: roleFilter || undefined, status: statusFilter || undefined, @@ -996,6 +1006,7 @@ export default function TasksDashboard() { api.getTaskSchedules(), api.getWorkerRegistry().catch(() => ({ workers: [] })), api.get('/api/tasks/pools').catch(() => ({ data: { pools: [] } })), + api.getHighFrequencySchedules().catch(() => ({ stores: [], stats: { totalStores: 0, byInterval: {}, byPlatform: {}, nextDueCount: 0 } })), ]); setTasks(tasksRes.tasks || []); @@ -1006,6 +1017,8 @@ export default function TasksDashboard() { setSchedules(schedulesRes.schedules || []); setWorkers(workersRes.workers || []); setPools(poolsRes.data?.pools || []); + setHighFreqStores(highFreqRes.stores || []); + setHighFreqStats(highFreqRes.stats || { totalStores: 0, byInterval: {}, byPlatform: {}, nextDueCount: 0 }); setError(null); } catch (err: any) { setError(err.message || 'Failed to load tasks'); @@ -1765,6 +1778,159 @@ export default function TasksDashboard() { )} + {/* High-Frequency (Per-Store) Schedules Section */} +
+ + + {showHighFreq && ( +
+ {/* Stats Summary */} + {highFreqStats.totalStores > 0 && ( +
+
+ By Interval:{' '} + {Object.entries(highFreqStats.byInterval).map(([interval, count]) => ( + + {interval}m: {count} + + ))} +
+
+ By Platform:{' '} + {Object.entries(highFreqStats.byPlatform).map(([platform, count]) => ( + + {platform}: {count} + + ))} +
+
+ )} + + {highFreqStores.length === 0 ? ( +
+ No stores configured for high-frequency crawling. +
+ Set a crawl interval on individual dispensaries to enable. +
+ ) : ( +
+ + + + + + + + + + + + + + {highFreqStores.map((store) => ( + + + + + + + + + + ))} + +
+ Store + + Platform + + Interval + + Next Crawl + + Last Fetch + + Changes (24h) + + Actions +
+
{store.name}
+
ID: {store.id}
+ 		+ + {store.menu_type} + + + {store.crawl_interval_minutes}m + + {store.next_crawl_at ? formatNextRun(store.next_crawl_at) : '-'} + + {store.last_fetch_at ? formatTimeAgo(store.last_fetch_at) : '-'} + + {(store.inventory_changes_24h > 0 || store.price_changes_24h > 0) ? ( +
+ {store.inventory_changes_24h > 0 && ( + {store.inventory_changes_24h} inv + )} + {store.price_changes_24h > 0 && ( + {store.price_changes_24h} price + )} +
+ ) : ( + - + )} + 		+ +
+
+ )} +
+ )} +
+ {/* Filters */}
diff --git a/findagram/frontend/Dockerfile b/findagram/frontend/Dockerfile index e751f6ae..3aec0312 100644 --- a/findagram/frontend/Dockerfile +++ b/findagram/frontend/Dockerfile @@ -6,8 +6,13 @@ WORKDIR /app # Copy package files COPY package*.json ./ -# Install dependencies (using npm install since package-lock.json may not exist) -RUN npm install +# Install dependencies with retry and fallback registry +# Primary: npmjs.org, Fallback: npmmirror.com (China mirror, globally accessible) +RUN npm config set fetch-retries 3 && \ + npm config set fetch-retry-mintimeout 20000 && \ + npm config set fetch-retry-maxtimeout 120000 && \ + npm install || \ + (npm config set registry https://registry.npmmirror.com && npm install) # Copy source files COPY . .