Update app.js

app.js

@@ -3,7 +3,7 @@ import cors from 'cors';
 import { createClient } from '@supabase/supabase-js';
 
 // --- CONFIG ---
-const PORT = 7860;
+const PORT = 7861; 
 const SUPABASE_URL = process.env.SUPABASE_URL;
 const SUPABASE_KEY = process.env.SUPABASE_SERVICE_KEY;
 const CRON_SECRET = process.env.CRON_SECRET || "default_secret";
@@ -13,126 +13,132 @@ if (!SUPABASE_URL || !SUPABASE_KEY) {
     process.exit(1);
 }
 
-// --- STATE ---
 const app = express();
 const supabase = createClient(SUPABASE_URL, SUPABASE_KEY);
-// In-Memory Map: JobID -> Timeout/Interval Object
+
 const activeJobs = new Map();
 
 app.use(express.json());
 app.use(cors());
 
-// --- HYDRATION ---
-async function hydrateJobs() {
-    console.log("💧 Hydrating Cron Jobs from DB...");
-    const { data, error } = await supabase.from('system_jobs').select('*');
+// --- BULLETPROOF DELAY MATH (UPDATED TO 5 AM) ---
+function getMsUntilNextFiveAM(offset = 0) {
+    const now = new Date();
+    const target = new Date(now);
     
-    if (error) {
-        console.error("Hydration Failed:", error);
-        return;
-    }
+    // Calculate target UTC minutes for local 5 AM
+    const targetUtcMinutes = (5 * 60) - (offset * 60); 
 
-    let count = 0;
-    for (const job of data) {
-        // Calculate remaining time if it was a one-off, or restart interval
-        // For simplicity in this architecture, we restart the cycle based on stored config
-        startJobInternal(job.id, job.interval_ms, job.webhook_url, job.payload);
-        count++;
+    target.setUTCHours(0, targetUtcMinutes, 0, 0); 
+    
+    if (target <= now) {
+        target.setDate(target.getDate() + 1);
     }
-    console.log(`✅ Hydrated ${count} jobs.`);
+
+    return target.getTime() - now.getTime();
 }
 
-// --- LOGIC ---
-function startJobInternal(jobId, intervalMs, url, payload, initialDelay = 0) {
+function startJobInternal(jobId, intervalMs, url, payload, initialDelay) {
     if (activeJobs.has(jobId)) {
-        clearTimeout(activeJobs.get(jobId));
-        clearInterval(activeJobs.get(jobId));
+        const existing = activeJobs.get(jobId);
+        if (existing.timeout) clearTimeout(existing.timeout);
+        if (existing.interval) clearInterval(existing.interval);
+        activeJobs.delete(jobId);
     }
 
     const runTask = async () => {
-        console.log(`⏰ Executing Job: ${jobId}`);
+        console.log(`⏰ [${new Date().toLocaleTimeString()}] Executing: ${jobId}`);
         try {
-            await fetch(url, {
+            const res = await fetch(url, {
                 method: 'POST',
                 headers: { 'Content-Type': 'application/json' },
                 body: JSON.stringify(payload)
             });
+            console.log(`   └─ Response: ${res.status}`);
         } catch (e) {
-            console.error(`❌ Job ${jobId} Failed:`, e.message);
+            console.error(`❌ Job ${jobId} HTTP Failed:`, e.message);
         }
     };
 
+    const jobHandles = { timeout: null, interval: null };
+
     if (initialDelay > 0) {
-        // Schedule the first run with delay, then set interval
-        console.log(`⏳ Job ${jobId} delayed by ${(initialDelay/1000/60).toFixed(1)} mins`);
-        const timeout = setTimeout(() => {
-            runTask(); // Run once
-            // Then start interval
-            const interval = setInterval(runTask, intervalMs);
-            activeJobs.set(jobId, interval);
+        console.log(`⏳ Job ${jobId} scheduled. First 5AM run in ${(initialDelay/1000/60/60).toFixed(2)} hours.`);
+        
+        jobHandles.timeout = setTimeout(async () => {
+            await runTask(); 
+            jobHandles.interval = setInterval(runTask, intervalMs);
+            console.log(`🔄 Job ${jobId} entering 24h interval cycle.`);
         }, initialDelay);
-        activeJobs.set(jobId, timeout);
     } else {
-        // Immediate interval start
-        const interval = setInterval(runTask, intervalMs);
-        activeJobs.set(jobId, interval);
+        console.log(`🚀 Job ${jobId} starting immediate interval.`);
+        jobHandles.interval = setInterval(runTask, intervalMs);
     }
+
+    activeJobs.set(jobId, jobHandles);
 }
 
-// --- ENDPOINTS ---
+async function hydrateJobs() {
+    console.log("💧 Hydrating Cron Jobs from DB...");
+    const { data, error } = await supabase.from('system_jobs').select('*');
+    
+    if (error) {
+        console.error("❌ Hydration Failed:", error.message);
+        return;
+    }
+
+    let count = 0;
+    for (const job of data) {
+        const offset = job.payload?.timezoneOffset ?? 0;
+        const newDelay = getMsUntilNextFiveAM(offset);
+
+        startJobInternal(job.id, job.interval_ms, job.webhook_url, job.payload, newDelay);
+        count++;
+    }
+    console.log(`✅ Successfully hydrated and scheduled ${count} jobs.`);
+}
 
-// app.get('/', (req, res) => res.send('Cron Registry Active'));
+app.get('/', (req, res) => res.json({ status: "Cron Registry Active", active_jobs: activeJobs.size }));
 
 app.post('/register', async (req, res) => {
-const { secret, jobId, intervalMs, webhookUrl, payload, initialDelay, leadId } = req.body;
-
-if (secret !== CRON_SECRET) return res.status(403).json({ error: "Unauthorized" });
-
-// Save to DB (Now including lead_id)
-const { error } = await supabase.from('system_jobs').upsert({
-    id: jobId,
-    lead_id: leadId, // <--- Save the UUID here
-    interval_ms: intervalMs,
-    webhook_url: webhookUrl,
-    payload: payload,
-    updated_at: new Date()
-});
+    const { secret, jobId, intervalMs, webhookUrl, payload, initialDelay, leadId } = req.body;
+
+    if (secret !== CRON_SECRET) return res.status(403).json({ error: "Unauthorized" });
+
+    const { error } = await supabase.from('system_jobs').upsert({
+        id: jobId,
+        lead_id: leadId,
+        interval_ms: intervalMs,
+        webhook_url: webhookUrl,
+        payload: payload,
+        updated_at: new Date()
+    });
   
     if (error) return res.status(500).json({ error: error.message });
 
-    // 2. Start in Memory
-    startJobInternal(jobId, intervalMs, webhookUrl, payload, initialDelay || 0);
+    const delay = (initialDelay !== undefined) ? initialDelay : getMsUntilNextFiveAM(payload?.timezoneOffset || 0);
+    
+    startJobInternal(jobId, intervalMs, webhookUrl, payload, delay);
 
-    console.log(`➕ Registered Job: ${jobId}`);
+    console.log(`➕ Registered & Scheduled Job (5AM Anchor): ${jobId}`);
     res.json({ success: true, jobId });
 });
 
 app.post('/deregister', async (req, res) => {
     const { secret, jobId } = req.body;
-
     if (secret !== CRON_SECRET) return res.status(403).json({ error: "Unauthorized" });
-
-    // 1. Remove from DB
     await supabase.from('system_jobs').delete().eq('id', jobId);
-
-    // 2. Stop in Memory
     if (activeJobs.has(jobId)) {
-        const handle = activeJobs.get(jobId);
-        clearTimeout(handle); // safe to call even if it's an interval
-        clearInterval(handle);
+        const handles = activeJobs.get(jobId);
+        if (handles.timeout) clearTimeout(handles.timeout);
+        if (handles.interval) clearInterval(handles.interval);
         activeJobs.delete(jobId);
     }
-
     console.log(`➖ Deregistered Job: ${jobId}`);
     res.json({ success: true });
 });
 
-app.get('/', async (req, res) => {
-   res.status(200).json({ status: "Alive" });
-});
-
-// --- BOOT ---
 app.listen(PORT, async () => {
-    console.log(`🚀 Cron Registry running on port ${PORT}`);
-    await hydrateJobs();
+    console.log(`🚀 Cron Registry live on port ${PORT}`);
+    setTimeout(hydrateJobs, 2000);
 });