Files
cannaiq/backend/src/index.ts
Kelly 8206dce821 feat(admin): Worker scaling controls via k8s API
- Add /api/k8s/workers endpoint to get deployment status
- Add /api/k8s/workers/scale endpoint to scale replicas (0-50)
- Add worker scaling UI to Tasks Dashboard (+/- 5 workers)
- Shows ready/desired replica count
- Uses in-cluster config in k8s, kubeconfig locally

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-11 08:24:32 -07:00

365 lines
15 KiB
TypeScript
Executable File

import express from 'express';
import cors from 'cors';
import path from 'path';
import dotenv from 'dotenv';
import { initializeMinio, isMinioEnabled } from './utils/minio';
import { initializeImageStorage } from './utils/image-storage';
import { logger } from './services/logger';
import { cleanupOrphanedJobs } from './services/proxyTestQueue';
// Per TASK_WORKFLOW_2024-12-10.md: Database-driven task scheduler
import { taskScheduler } from './services/task-scheduler';
import { runAutoMigrations } from './db/auto-migrate';
import { getPool } from './db/pool';
import healthRoutes from './routes/health';
import imageProxyRoutes from './routes/image-proxy';
dotenv.config();
const app = express();
const PORT = process.env.PORT || 3010;
// CORS configuration - allow requests from any origin with API key auth
// WordPress plugins need to make requests from their own domains
app.use(cors({
origin: true, // Reflect the request origin
credentials: true,
methods: ['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS'],
allowedHeaders: ['Content-Type', 'Authorization', 'x-api-key', 'X-API-Key'],
exposedHeaders: ['Content-Length', 'X-Request-Id'],
}));
app.use(express.json());
// Serve static images when MinIO is not configured
// Uses ./public/images relative to working directory (works for both Docker and local dev)
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
app.get('/health', (req, res) => {
res.json({ status: 'ok', timestamp: new Date().toISOString() });
});
// Comprehensive health endpoints for monitoring (no auth required)
app.use('/api/health', healthRoutes);
// Endpoint to check server's outbound IP (for proxy whitelist setup)
app.get('/outbound-ip', async (req, res) => {
try {
const axios = require('axios');
const response = await axios.get('https://api.ipify.org?format=json', { timeout: 10000 });
res.json({ outbound_ip: response.data.ip });
} catch (error: any) {
res.status(500).json({ error: error.message });
}
});
import authRoutes from './routes/auth';
import dashboardRoutes from './routes/dashboard';
import storesRoutes from './routes/stores';
import dispensariesRoutes from './routes/dispensaries';
import changesRoutes from './routes/changes';
import categoriesRoutes from './routes/categories';
import productsRoutes from './routes/products';
import campaignsRoutes from './routes/campaigns';
import analyticsRoutes from './routes/analytics';
import settingsRoutes from './routes/settings';
import proxiesRoutes from './routes/proxies';
import logsRoutes from './routes/logs';
import scraperMonitorRoutes from './routes/scraper-monitor';
import apiTokensRoutes from './routes/api-tokens';
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';
import orchestratorAdminRoutes from './routes/orchestrator-admin';
import adminDebugRoutes from './routes/admin-debug';
import intelligenceRoutes from './routes/intelligence';
import marketsRoutes from './routes/markets';
import workersRoutes from './routes/workers';
import jobQueueRoutes from './routes/job-queue';
import { createMultiStateRoutes } from './multi-state';
import { trackApiUsage, checkRateLimit } from './middleware/apiTokenTracker';
import { validateWordPressPermissions } from './middleware/wordpressPermissions';
import { markTrustedDomains } from './middleware/trustedDomains';
import { createSystemRouter, createPrometheusRouter } from './system/routes';
import { createPortalRoutes } from './portals';
import { createStatesRouter } from './routes/states';
import { createAnalyticsV2Router } from './routes/analytics-v2';
import { createDiscoveryRoutes } from './discovery';
import pipelineRoutes from './routes/pipeline';
// Consumer API routes (findadispo.com, findagram.co)
import consumerAuthRoutes from './routes/consumer-auth';
import consumerFavoritesRoutes from './routes/consumer-favorites';
import consumerAlertsRoutes from './routes/consumer-alerts';
import consumerSavedSearchesRoutes from './routes/consumer-saved-searches';
import consumerDealsRoutes from './routes/consumer-deals';
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';
import workerRegistryRoutes from './routes/worker-registry';
// Per TASK_WORKFLOW_2024-12-10.md: Raw payload access API
import payloadsRoutes from './routes/payloads';
import k8sRoutes from './routes/k8s';
// Mark requests from trusted domains (cannaiq.co, findagram.co, findadispo.com)
// These domains can access the API without authentication
app.use(markTrustedDomains);
// Apply WordPress permissions validation (sets req.apiToken for API key requests)
app.use(validateWordPressPermissions);
// Apply API tracking middleware globally
app.use(trackApiUsage);
app.use(checkRateLimit);
app.use('/api/auth', authRoutes);
app.use('/api/dashboard', dashboardRoutes);
app.use('/api/stores', storesRoutes);
app.use('/api/dispensaries', dispensariesRoutes);
app.use('/api/changes', changesRoutes);
app.use('/api/categories', categoriesRoutes);
app.use('/api/products', productsRoutes);
app.use('/api/campaigns', campaignsRoutes);
// Multi-state API routes - national analytics and cross-state comparisons (NO AUTH)
// IMPORTANT: Must be mounted BEFORE /api/analytics to avoid auth middleware blocking these routes
try {
const multiStateRoutes = createMultiStateRoutes(getPool());
app.use('/api', multiStateRoutes);
console.log('[MultiState] Routes registered at /api (analytics/national/*, states/*, etc.)');
} catch (error) {
console.warn('[MultiState] Failed to register routes (DB may not be configured):', error);
}
// Legacy click analytics routes (requires auth)
app.use('/api/analytics', analyticsRoutes);
app.use('/api/settings', settingsRoutes);
app.use('/api/proxies', proxiesRoutes);
app.use('/api/logs', logsRoutes);
app.use('/api/scraper-monitor', scraperMonitorRoutes);
app.use('/api/api-tokens', apiTokensRoutes);
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
app.use('/api/admin/orchestrator', orchestratorAdminRoutes);
// Admin routes - debug endpoints (snapshot inspection)
app.use('/api/admin/debug', adminDebugRoutes);
console.log('[AdminDebug] Routes registered at /api/admin/debug');
// Admin routes - intelligence (brands, pricing analytics)
app.use('/api/admin/intelligence', intelligenceRoutes);
console.log('[Intelligence] Routes registered at /api/admin/intelligence');
// Markets routes - store and product data for admin dashboard
app.use('/api/markets', marketsRoutes);
console.log('[Markets] Routes registered at /api/markets');
// SEO orchestrator routes
app.use('/api/seo', seoRoutes);
// Provider-agnostic worker management routes
app.use('/api/workers', workersRoutes);
// Monitor routes - aliased from workers for convenience
app.use('/api/monitor', workersRoutes);
// Job queue management
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');
// Worker registry - dynamic worker registration, heartbeats, and name management
app.use('/api/worker-registry', workerRegistryRoutes);
console.log('[WorkerRegistry] Routes registered at /api/worker-registry');
// Per TASK_WORKFLOW_2024-12-10.md: Raw payload access API
app.use('/api/payloads', payloadsRoutes);
console.log('[Payloads] Routes registered at /api/payloads');
// K8s control routes - worker scaling from admin UI
app.use('/api/k8s', k8sRoutes);
console.log('[K8s] Routes registered at /api/k8s');
// Phase 3: Analytics V2 - Enhanced analytics with rec/med state segmentation
try {
const analyticsV2Router = createAnalyticsV2Router(getPool());
app.use('/api/analytics/v2', analyticsV2Router);
console.log('[AnalyticsV2] Routes registered at /api/analytics/v2');
} catch (error) {
console.warn('[AnalyticsV2] Failed to register routes:', error);
}
// Public API v1 - External consumer endpoints (WordPress, etc.)
// Uses dutchie_az data pipeline with per-dispensary API key auth
app.use('/api/v1', publicApiRoutes);
// Consumer API - findadispo.com and findagram.co user features
// Auth routes don't require authentication
app.use('/api/consumer/auth', consumerAuthRoutes);
// Protected consumer routes (favorites, alerts, saved searches)
app.use('/api/consumer/favorites', consumerFavoritesRoutes);
app.use('/api/consumer/alerts', consumerAlertsRoutes);
app.use('/api/consumer/saved-searches', consumerSavedSearchesRoutes);
// Deals endpoint - public, no auth required
app.use('/api/v1/deals', consumerDealsRoutes);
console.log('[Consumer] Routes registered at /api/consumer/*');
// Events API - product click tracking for analytics and campaigns
app.use('/api/events', eventsRoutes);
console.log('[Events] Routes registered at /api/events');
// Click Analytics API - brand and campaign engagement aggregations
app.use('/api/analytics/clicks', clickAnalyticsRoutes);
console.log('[ClickAnalytics] Routes registered at /api/analytics/clicks');
// Price Analytics API - price history, specials, and market comparisons
app.use('/api/analytics/price', priceAnalyticsRoutes);
console.log('[PriceAnalytics] Routes registered at /api/analytics/price');
// States API routes - cannabis legalization status and targeting
try {
const statesRouter = createStatesRouter(getPool());
app.use('/api/states', statesRouter);
console.log('[States] Routes registered at /api/states');
} catch (error) {
console.warn('[States] Failed to register routes:', error);
}
// Phase 5: Production Sync + Monitoring
// System orchestrator, DLQ, integrity checks, auto-fix routines, alerts
try {
const systemRouter = createSystemRouter(getPool());
const prometheusRouter = createPrometheusRouter(getPool());
app.use('/api/system', systemRouter);
app.use('/metrics', prometheusRouter);
console.log('[System] Routes registered at /api/system and /metrics');
} catch (error) {
console.warn('[System] Failed to register routes:', error);
}
// Phase 6 & 7: Portals (Brand, Buyer), Intelligence, Orders, Inventory, Pricing
try {
const portalRoutes = createPortalRoutes(getPool());
app.use('/api/portal', portalRoutes);
console.log('[Portals] Routes registered at /api/portal');
} catch (error) {
console.warn('[Portals] Failed to register routes:', error);
}
// Discovery Pipeline - Store discovery from Dutchie with verification workflow
try {
const discoveryRoutes = createDiscoveryRoutes(getPool());
app.use('/api/discovery', discoveryRoutes);
console.log('[Discovery] Routes registered at /api/discovery');
} catch (error) {
console.warn('[Discovery] Failed to register routes:', error);
}
// Pipeline Stage Transitions - Explicit API for moving stores through 6-stage pipeline
app.use('/api/pipeline', pipelineRoutes);
console.log('[Pipeline] Routes registered at /api/pipeline');
// Platform-specific Discovery Routes
// TODO: Rebuild with /platforms/dutchie/ module
async function startServer() {
try {
logger.info('system', 'Starting server...');
// Run auto-migrations before anything else
const pool = getPool();
const migrationsApplied = await runAutoMigrations(pool);
if (migrationsApplied > 0) {
logger.info('system', `Applied ${migrationsApplied} database migrations`);
} else if (migrationsApplied === 0) {
logger.info('system', 'Database schema up to date');
} else {
logger.warn('system', 'Some migrations failed - check logs');
}
await initializeMinio();
await initializeImageStorage();
logger.info('system', isMinioEnabled() ? 'MinIO storage initialized' : 'Local filesystem storage initialized');
// Clean up any orphaned proxy test jobs from previous server runs
await cleanupOrphanedJobs();
// Per TASK_WORKFLOW_2024-12-10.md: Start database-driven task scheduler
// This replaces node-cron - schedules are stored in DB and survive restarts
// Uses SELECT FOR UPDATE SKIP LOCKED for multi-replica safety
try {
await taskScheduler.start();
logger.info('system', 'Task scheduler started');
} catch (err: any) {
// Non-fatal - scheduler can recover on next poll
logger.warn('system', `Task scheduler startup warning: ${err.message}`);
}
app.listen(PORT, () => {
logger.info('system', `Server running on port ${PORT}`);
console.log(`🚀 Server running on port ${PORT}`);
});
} catch (error) {
logger.error('system', `Failed to start server: ${error}`);
console.error('Failed to start server:', error);
process.exit(1);
}
}
startServer();