1898 lines
51 KiB
Plaintext
1898 lines
51 KiB
Plaintext
/**
|
||
* ApsGo Railway Worker
|
||
* Background service untuk automation scheduling 24/7
|
||
* Features:
|
||
* - Waktu Mode: Scheduled watering by time
|
||
* - Sensor Mode: Automatic watering by soil moisture threshold
|
||
* - Redis Queue: Prevent race conditions & concurrent task management
|
||
* - Firebase Realtime DB: Sync dengan Flutter app dan ESP32
|
||
* - History Monitoring dan History Penyiraman dengan penanda jenis_histori
|
||
*/
|
||
|
||
// ==================== SUPPRESS FIREBASE WARNINGS ====================
|
||
|
||
process.env.FIREBASE_DATABASE_EMULATOR_HOST = undefined;
|
||
process.env.FIRESTORE_EMULATOR_HOST = undefined;
|
||
|
||
const originalWarn = console.warn;
|
||
console.warn = function (...args) {
|
||
const message = args.join(' ');
|
||
|
||
if (
|
||
message.includes('FIREBASE WARNING') ||
|
||
message.includes('@firebase/database') ||
|
||
message.includes('firebase/database')
|
||
) {
|
||
return;
|
||
}
|
||
|
||
originalWarn.apply(console, args);
|
||
};
|
||
|
||
// ==================== IMPORTS ====================
|
||
|
||
require('dotenv').config();
|
||
|
||
const admin = require('firebase-admin');
|
||
const { Queue, Worker } = require('bullmq');
|
||
const Redis = require('ioredis');
|
||
const cron = require('cron');
|
||
|
||
// ==================== CONFIGURATION ====================
|
||
|
||
process.env.TZ = process.env.TZ || 'Asia/Jakarta';
|
||
|
||
const FIREBASE_PATHS = {
|
||
kontrol: 'kontrol_1',
|
||
aktuator: 'aktuator',
|
||
data: 'data',
|
||
history: 'history',
|
||
};
|
||
|
||
const NOTIFICATION_TOPIC = 'apsgo_notifications';
|
||
const NOTIFICATION_CHANNEL_ID = 'apsgo_watering_channel';
|
||
|
||
const config = {
|
||
redis: {
|
||
host: process.env.REDIS_HOST || 'localhost',
|
||
port: parseInt(process.env.REDIS_PORT) || 6379,
|
||
password: process.env.REDIS_PASSWORD || undefined,
|
||
maxRetriesPerRequest: null,
|
||
},
|
||
|
||
firebase: {
|
||
projectId: process.env.FIREBASE_PROJECT_ID,
|
||
clientEmail: process.env.FIREBASE_CLIENT_EMAIL,
|
||
privateKey: process.env.FIREBASE_PRIVATE_KEY?.replace(/\\n/g, '\n'),
|
||
databaseURL: process.env.FIREBASE_DATABASE_URL,
|
||
},
|
||
|
||
worker: {
|
||
concurrency: 1,
|
||
checkInterval: 60000,
|
||
sensorDebounce: 120000,
|
||
scheduleGraceMs: parseInt(process.env.SCHEDULE_GRACE_MS || '15000', 10),
|
||
scheduleMaxCatchupMs: parseInt(
|
||
process.env.SCHEDULE_MAX_CATCHUP_MS || '300000',
|
||
10
|
||
),
|
||
breakBetweenPotsMs: 30000,
|
||
},
|
||
};
|
||
|
||
console.log('🚀 Starting ApsGo Railway Worker...');
|
||
console.log(`📡 Firebase Project: ${config.firebase.projectId}`);
|
||
console.log(`🔥 Firebase DB URL: ${config.firebase.databaseURL}`);
|
||
console.log(`📦 Redis: ${config.redis.host}:${config.redis.port}`);
|
||
console.log(
|
||
`⏰ Timezone: ${process.env.TZ} | Current: ${new Date().toLocaleString(
|
||
'id-ID',
|
||
{ timeZone: 'Asia/Jakarta' }
|
||
)}`
|
||
);
|
||
console.log(`📍 Kontrol Path: /${FIREBASE_PATHS.kontrol}`);
|
||
|
||
// ==================== ENVIRONMENT VALIDATION ====================
|
||
|
||
const requiredEnvs = [
|
||
'FIREBASE_PROJECT_ID',
|
||
'FIREBASE_CLIENT_EMAIL',
|
||
'FIREBASE_PRIVATE_KEY',
|
||
'FIREBASE_DATABASE_URL',
|
||
];
|
||
|
||
const missingEnvs = requiredEnvs.filter((env) => !process.env[env]);
|
||
|
||
if (missingEnvs.length > 0) {
|
||
console.error('❌ Missing required environment variables:');
|
||
|
||
missingEnvs.forEach((env) => {
|
||
console.error(` - ${env}`);
|
||
});
|
||
|
||
console.error('\n📋 To fix this:');
|
||
console.error('1. Go to Railway Dashboard');
|
||
console.error('2. Select your worker service');
|
||
console.error('3. Go to Variables tab');
|
||
console.error('4. Add the missing variables');
|
||
console.error('5. Redeploy');
|
||
|
||
process.exit(1);
|
||
}
|
||
|
||
console.log('✅ All required environment variables are set');
|
||
|
||
// ==================== FIREBASE INITIALIZATION ====================
|
||
|
||
try {
|
||
admin.initializeApp({
|
||
credential: admin.credential.cert({
|
||
projectId: config.firebase.projectId,
|
||
clientEmail: config.firebase.clientEmail,
|
||
privateKey: config.firebase.privateKey,
|
||
}),
|
||
databaseURL: config.firebase.databaseURL,
|
||
});
|
||
|
||
console.log('✅ Firebase Admin initialized');
|
||
} catch (error) {
|
||
console.error('❌ Firebase initialization failed:', error.message);
|
||
process.exit(1);
|
||
}
|
||
|
||
const db = admin.database();
|
||
|
||
// ==================== NOTIFICATION FUNCTION ====================
|
||
|
||
async function sendAutomationNotification({ title, body, type, data = {} }) {
|
||
try {
|
||
if (!admin.messaging) {
|
||
console.log(
|
||
'📝 [INFO] Cloud Messaging not available. App will use local notifications via Firebase listener.'
|
||
);
|
||
return;
|
||
}
|
||
|
||
const messaging = admin.messaging();
|
||
|
||
if (!messaging) {
|
||
console.log(
|
||
'📝 [INFO] Cloud Messaging SDK not initialized. App will use local notifications via Firebase listener.'
|
||
);
|
||
return;
|
||
}
|
||
|
||
const result = await messaging.send({
|
||
topic: NOTIFICATION_TOPIC,
|
||
|
||
notification: {
|
||
title,
|
||
body,
|
||
},
|
||
|
||
data: {
|
||
type: String(type || 'automation'),
|
||
title: String(title || ''),
|
||
body: String(body || ''),
|
||
...Object.fromEntries(
|
||
Object.entries(data).map(([key, value]) => [
|
||
key,
|
||
String(value),
|
||
])
|
||
),
|
||
},
|
||
|
||
android: {
|
||
priority: 'high',
|
||
notification: {
|
||
channelId: NOTIFICATION_CHANNEL_ID,
|
||
icon: 'ic_launcher',
|
||
color: '#2E7D32',
|
||
},
|
||
},
|
||
});
|
||
|
||
console.log(`🔔 FCM Notification sent: ${type} | ID: ${result}`);
|
||
} catch (error) {
|
||
console.log(
|
||
`📝 [INFO] FCM not available (${error.message}). App will use local notifications via Firebase listener.`
|
||
);
|
||
}
|
||
}
|
||
|
||
// ==================== FIREBASE CONNECTION STATUS ====================
|
||
|
||
db.ref('.info/connected').on('value', (snap) => {
|
||
if (snap.val() === true) {
|
||
console.log('🔌 Firebase realtime connection active');
|
||
} else {
|
||
console.log('⚠️ Firebase realtime connection inactive');
|
||
}
|
||
});
|
||
|
||
// ==================== REDIS & QUEUE SETUP ====================
|
||
|
||
const redis = new Redis(config.redis);
|
||
|
||
const wateringQueue = new Queue('watering', {
|
||
connection: redis,
|
||
});
|
||
|
||
redis.on('connect', () => {
|
||
console.log('✅ Redis connected');
|
||
});
|
||
|
||
redis.on('error', (err) => {
|
||
console.error('❌ Redis error:', err.message);
|
||
});
|
||
|
||
// Track cooldown threshold
|
||
const lastThresholdTime = {};
|
||
|
||
// ==================== FIREBASE SMART HELPER ====================
|
||
|
||
let consecutiveFirebaseErrors = 0;
|
||
let sdkSuccessCount = 0;
|
||
let restFallbackCount = 0;
|
||
|
||
const SKIP_SDK_THRESHOLD = 3;
|
||
const RESET_THRESHOLD_AFTER = 50;
|
||
|
||
async function fetchWithTimeout(ref, timeoutMs = 5000) {
|
||
return Promise.race([
|
||
ref.once('value'),
|
||
new Promise((_, reject) =>
|
||
setTimeout(() => reject(new Error('Firebase fetch timeout')), timeoutMs)
|
||
),
|
||
]);
|
||
}
|
||
|
||
async function fetchWithTimeout2(url, options = {}, timeoutMs = 8000) {
|
||
const controller = new AbortController();
|
||
const timeout = setTimeout(() => controller.abort(), timeoutMs);
|
||
|
||
try {
|
||
const response = await fetch(url, {
|
||
...options,
|
||
signal: controller.signal,
|
||
});
|
||
|
||
clearTimeout(timeout);
|
||
return response;
|
||
} catch (error) {
|
||
clearTimeout(timeout);
|
||
|
||
if (error.name === 'AbortError') {
|
||
throw new Error('REST API timeout');
|
||
}
|
||
|
||
throw error;
|
||
}
|
||
}
|
||
|
||
async function fetchKontrolViaREST() {
|
||
const url = `${config.firebase.databaseURL}/${FIREBASE_PATHS.kontrol}.json`;
|
||
|
||
console.log(` [DEBUG] Trying REST API: ${url}`);
|
||
|
||
const response = await fetchWithTimeout2(
|
||
url,
|
||
{
|
||
method: 'GET',
|
||
headers: {
|
||
'Content-Type': 'application/json',
|
||
},
|
||
},
|
||
8000
|
||
);
|
||
|
||
if (!response.ok) {
|
||
throw new Error(`REST API failed: ${response.status} ${response.statusText}`);
|
||
}
|
||
|
||
const data = await response.json();
|
||
|
||
console.log(' [DEBUG] REST API successful!');
|
||
return data;
|
||
}
|
||
|
||
async function fetchKontrolSmart() {
|
||
const shouldSkipSDK = consecutiveFirebaseErrors >= SKIP_SDK_THRESHOLD;
|
||
|
||
if (shouldSkipSDK) {
|
||
console.log(
|
||
' [SMART] Skipping SDK because 3+ consecutive failures. Using REST API directly...'
|
||
);
|
||
|
||
try {
|
||
const data = await fetchKontrolViaREST();
|
||
|
||
restFallbackCount++;
|
||
|
||
if (restFallbackCount >= RESET_THRESHOLD_AFTER) {
|
||
console.log(' [SMART] Resetting SDK retry counter...');
|
||
consecutiveFirebaseErrors = 0;
|
||
restFallbackCount = 0;
|
||
}
|
||
|
||
return data;
|
||
} catch (restError) {
|
||
console.error(' ❌ REST API failed:', restError.message);
|
||
throw new Error('REST API failed');
|
||
}
|
||
}
|
||
|
||
try {
|
||
console.log(' [DEBUG] Attempting SDK fetch...');
|
||
|
||
const snapshot = await fetchWithTimeout(
|
||
db.ref(FIREBASE_PATHS.kontrol),
|
||
5000
|
||
);
|
||
|
||
consecutiveFirebaseErrors = 0;
|
||
restFallbackCount = 0;
|
||
sdkSuccessCount++;
|
||
|
||
return snapshot.val();
|
||
} catch (sdkError) {
|
||
console.warn(' ⚠️ SDK fetch failed, trying REST API...');
|
||
|
||
consecutiveFirebaseErrors++;
|
||
|
||
try {
|
||
const data = await fetchKontrolViaREST();
|
||
|
||
restFallbackCount++;
|
||
|
||
if (consecutiveFirebaseErrors === SKIP_SDK_THRESHOLD) {
|
||
console.warn(
|
||
` 🚨 SDK failed ${SKIP_SDK_THRESHOLD}x consecutively. REST API will be used directly.`
|
||
);
|
||
}
|
||
|
||
return data;
|
||
} catch (restError) {
|
||
console.error(' ❌ REST API also failed:', restError.message);
|
||
throw new Error('Both SDK and REST API failed');
|
||
}
|
||
}
|
||
}
|
||
|
||
async function updateFirebaseViaREST(path, updates) {
|
||
const url = `${config.firebase.databaseURL}/${path}.json`;
|
||
|
||
console.log(` [DEBUG] REST API PATCH: ${url}`);
|
||
|
||
const response = await fetchWithTimeout2(
|
||
url,
|
||
{
|
||
method: 'PATCH',
|
||
headers: {
|
||
'Content-Type': 'application/json',
|
||
},
|
||
body: JSON.stringify(updates),
|
||
},
|
||
8000
|
||
);
|
||
|
||
if (!response.ok) {
|
||
throw new Error(`REST PATCH failed: ${response.status}`);
|
||
}
|
||
|
||
const result = await response.json();
|
||
|
||
console.log(' [DEBUG] REST API update successful!');
|
||
return result;
|
||
}
|
||
|
||
async function updateWithTimeout(ref, updates, timeoutMs = 5000) {
|
||
return Promise.race([
|
||
ref.update(updates),
|
||
new Promise((_, reject) =>
|
||
setTimeout(() => reject(new Error('Firebase update timeout')), timeoutMs)
|
||
),
|
||
]);
|
||
}
|
||
|
||
async function updateFirebaseSmart(path, updates, maxAttempts = 3) {
|
||
const updateStr = JSON.stringify(updates);
|
||
|
||
console.log(` [UPDATE START] Path: /${path}, Data: ${updateStr}`);
|
||
console.log(` [UPDATE] Max attempts: ${maxAttempts}`);
|
||
|
||
let lastError = null;
|
||
|
||
for (let attempt = 1; attempt <= maxAttempts; attempt++) {
|
||
try {
|
||
const shouldSkipSDK = consecutiveFirebaseErrors >= SKIP_SDK_THRESHOLD;
|
||
|
||
if (shouldSkipSDK) {
|
||
console.log(
|
||
` [UPDATE Attempt ${attempt}/${maxAttempts}] Using REST API directly`
|
||
);
|
||
|
||
await updateFirebaseViaREST(path, updates);
|
||
|
||
console.log(` ✅ [UPDATE] REST API successful on attempt ${attempt}`);
|
||
return true;
|
||
}
|
||
|
||
console.log(
|
||
` [UPDATE Attempt ${attempt}/${maxAttempts}] Attempting SDK update...`
|
||
);
|
||
|
||
await updateWithTimeout(db.ref(path), updates, 5000);
|
||
|
||
console.log(` ✅ [UPDATE] SDK update successful on attempt ${attempt}`);
|
||
return true;
|
||
} catch (sdkError) {
|
||
lastError = sdkError;
|
||
|
||
console.warn(
|
||
` ⚠️ [UPDATE Attempt ${attempt}/${maxAttempts}] SDK failed: ${sdkError.message}`
|
||
);
|
||
|
||
try {
|
||
console.log(
|
||
` [UPDATE Attempt ${attempt}/${maxAttempts}] Fallback to REST API...`
|
||
);
|
||
|
||
await updateFirebaseViaREST(path, updates);
|
||
|
||
console.log(` ✅ [UPDATE] REST API successful on attempt ${attempt}`);
|
||
return true;
|
||
} catch (restError) {
|
||
lastError = restError;
|
||
|
||
console.error(
|
||
` ❌ [UPDATE Attempt ${attempt}/${maxAttempts}] REST failed: ${restError.message}`
|
||
);
|
||
|
||
if (attempt < maxAttempts) {
|
||
const delayMs = attempt * 1000;
|
||
|
||
console.log(` ⏳ [UPDATE] Retrying after ${delayMs}ms...`);
|
||
|
||
await new Promise((resolve) => setTimeout(resolve, delayMs));
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
console.error(` ❌ [UPDATE] ALL ${maxAttempts} ATTEMPTS FAILED!`);
|
||
console.error(` ❌ [UPDATE] Last error: ${lastError.message}`);
|
||
|
||
throw new Error(
|
||
`Firebase update failed after ${maxAttempts} attempts: ${lastError.message}`
|
||
);
|
||
}
|
||
|
||
async function setFirebaseViaREST(path, data) {
|
||
const url = `${config.firebase.databaseURL}/${path}.json`;
|
||
|
||
const response = await fetchWithTimeout2(
|
||
url,
|
||
{
|
||
method: 'PUT',
|
||
headers: {
|
||
'Content-Type': 'application/json',
|
||
},
|
||
body: JSON.stringify(data),
|
||
},
|
||
8000
|
||
);
|
||
|
||
if (!response.ok) {
|
||
throw new Error(`REST PUT failed: ${response.status}`);
|
||
}
|
||
|
||
return await response.json();
|
||
}
|
||
|
||
async function setWithTimeout(ref, data, timeoutMs = 5000) {
|
||
return Promise.race([
|
||
ref.set(data),
|
||
new Promise((_, reject) =>
|
||
setTimeout(() => reject(new Error('Firebase set timeout')), timeoutMs)
|
||
),
|
||
]);
|
||
}
|
||
|
||
async function setFirebaseSmart(path, data) {
|
||
const dataStr = JSON.stringify(data);
|
||
|
||
console.log(` [SET START] Path: ${path}, Data: ${dataStr.substring(0, 100)}...`);
|
||
|
||
const shouldSkipSDK = consecutiveFirebaseErrors >= SKIP_SDK_THRESHOLD;
|
||
|
||
if (shouldSkipSDK) {
|
||
console.log(' [SET] Using REST API directly');
|
||
|
||
try {
|
||
await setFirebaseViaREST(path, data);
|
||
|
||
console.log(' ✅ [SET] REST API successful!');
|
||
return true;
|
||
} catch (restError) {
|
||
console.error(` ❌ [SET] REST failed: ${restError.message}`);
|
||
throw new Error('REST set failed');
|
||
}
|
||
}
|
||
|
||
try {
|
||
console.log(' [SET] Attempting SDK set...');
|
||
|
||
await setWithTimeout(db.ref(path), data, 5000);
|
||
|
||
console.log(' ✅ [SET] SDK set successful!');
|
||
return true;
|
||
} catch (sdkError) {
|
||
console.warn(
|
||
` ⚠️ [SET] SDK failed (${sdkError.message}), trying REST API...`
|
||
);
|
||
|
||
try {
|
||
await setFirebaseViaREST(path, data);
|
||
|
||
console.log(' ✅ [SET] REST API successful!');
|
||
return true;
|
||
} catch (restError) {
|
||
console.error(' ❌ [SET] REST failed. Both methods failed.');
|
||
throw new Error('Both SDK and REST set failed');
|
||
}
|
||
}
|
||
}
|
||
|
||
async function readFirebaseSmart(path) {
|
||
const shouldSkipSDK = consecutiveFirebaseErrors >= SKIP_SDK_THRESHOLD;
|
||
|
||
if (shouldSkipSDK) {
|
||
console.log(' [READ] Using REST API directly');
|
||
|
||
const url = `${config.firebase.databaseURL}/${path}.json`;
|
||
|
||
const response = await fetchWithTimeout2(
|
||
url,
|
||
{
|
||
method: 'GET',
|
||
headers: {
|
||
'Content-Type': 'application/json',
|
||
},
|
||
},
|
||
8000
|
||
);
|
||
|
||
if (!response.ok) {
|
||
throw new Error(`REST GET failed: ${response.status}`);
|
||
}
|
||
|
||
return await response.json();
|
||
}
|
||
|
||
try {
|
||
const snapshot = await fetchWithTimeout(db.ref(path), 5000);
|
||
|
||
return snapshot.val();
|
||
} catch (sdkError) {
|
||
const url = `${config.firebase.databaseURL}/${path}.json`;
|
||
|
||
const response = await fetchWithTimeout2(
|
||
url,
|
||
{
|
||
method: 'GET',
|
||
headers: {
|
||
'Content-Type': 'application/json',
|
||
},
|
||
},
|
||
8000
|
||
);
|
||
|
||
if (!response.ok) {
|
||
throw new Error(`REST GET failed: ${response.status}`);
|
||
}
|
||
|
||
return await response.json();
|
||
}
|
||
}
|
||
|
||
// ==================== DATE & HISTORY HELPERS ====================
|
||
|
||
function getTanggalIndonesia(now) {
|
||
return `${now.getDate()}/${now.getMonth() + 1}/${now.getFullYear()}`;
|
||
}
|
||
|
||
function getWaktuIndonesia(now) {
|
||
return `${now.getHours().toString().padStart(2, '0')}:${now
|
||
.getMinutes()
|
||
.toString()
|
||
.padStart(2, '0')}`;
|
||
}
|
||
|
||
function getDateKey(now) {
|
||
return `${now.getFullYear()}-${(now.getMonth() + 1)
|
||
.toString()
|
||
.padStart(2, '0')}-${now.getDate().toString().padStart(2, '0')}`;
|
||
}
|
||
|
||
function getModeFromType(type) {
|
||
const t = String(type || '').toLowerCase();
|
||
|
||
if (t.includes('waktu')) return 'waktu';
|
||
if (t.includes('sensor')) return 'sensor';
|
||
if (t.includes('threshold')) return 'sensor';
|
||
if (t.includes('manual')) return 'manual';
|
||
|
||
return 'otomatis';
|
||
}
|
||
|
||
// ==================== HISTORY LOGGING ====================
|
||
|
||
async function logHistory(type, potNumbers, duration) {
|
||
try {
|
||
const now = new Date();
|
||
const sensorData = await readFirebaseSmart(FIREBASE_PATHS.data) || {};
|
||
const mode = getModeFromType(type);
|
||
|
||
console.log(' 📝 Saving penyiraman history per pot...');
|
||
|
||
for (const pot of potNumbers) {
|
||
const historyData = {
|
||
timestamp: now.getTime(),
|
||
tanggal: getTanggalIndonesia(now),
|
||
waktu: getWaktuIndonesia(now),
|
||
|
||
jenis_histori: 'penyiraman',
|
||
source: 'railway_worker',
|
||
mode: mode,
|
||
type: type,
|
||
|
||
pot: pot,
|
||
nama_pot: `Pot ${pot}`,
|
||
|
||
suhu: Number(sensorData.suhu || 0),
|
||
kelembapan: Number(sensorData.kelembapan || 0),
|
||
soil: Number(sensorData[`soil_${pot}`] || 0),
|
||
|
||
durasi: Number(duration || 0),
|
||
};
|
||
|
||
await db.ref(FIREBASE_PATHS.history).push().set(historyData);
|
||
|
||
console.log(
|
||
` ✅ Histori penyiraman tersimpan: Pot ${pot}, Mode ${mode}, Soil ${historyData.soil}%`
|
||
);
|
||
}
|
||
|
||
console.log(` 📊 History penyiraman logged: ${potNumbers.length} pot`);
|
||
} catch (error) {
|
||
console.error(' ⚠️ Failed to log penyiraman history:', error.message);
|
||
}
|
||
}
|
||
|
||
async function logMonitoringHistory() {
|
||
try {
|
||
const sensorData = await readFirebaseSmart(FIREBASE_PATHS.data);
|
||
|
||
if (!sensorData) {
|
||
console.log('⚠️ Auto-log skipped: sensor data kosong');
|
||
return;
|
||
}
|
||
|
||
const now = new Date();
|
||
|
||
console.log('📊 Saving monitoring history every 30 minutes...');
|
||
|
||
for (let pot = 1; pot <= 5; pot++) {
|
||
const historyData = {
|
||
timestamp: now.getTime(),
|
||
tanggal: getTanggalIndonesia(now),
|
||
waktu: getWaktuIndonesia(now),
|
||
|
||
jenis_histori: 'monitoring',
|
||
source: 'railway_worker',
|
||
mode: '-',
|
||
type: 'auto_log',
|
||
|
||
pot: pot,
|
||
nama_pot: `Pot ${pot}`,
|
||
|
||
suhu: Number(sensorData.suhu || 0),
|
||
kelembapan: Number(sensorData.kelembapan || 0),
|
||
soil: Number(sensorData[`soil_${pot}`] || 0),
|
||
|
||
durasi: 0,
|
||
};
|
||
|
||
await db.ref(FIREBASE_PATHS.history).push().set(historyData);
|
||
|
||
console.log(
|
||
` ✅ Histori monitoring tersimpan: Pot ${pot}, Soil ${historyData.soil}%`
|
||
);
|
||
}
|
||
|
||
console.log('📊 Auto-logged monitoring data selesai');
|
||
} catch (error) {
|
||
console.error('❌ Auto-log monitoring failed:', error.message);
|
||
}
|
||
}
|
||
|
||
// ==================== SCHEDULE HELPER ====================
|
||
|
||
function parseScheduleTimeToday(scheduleTime, now) {
|
||
if (typeof scheduleTime !== 'string' || !/^\d{2}:\d{2}$/.test(scheduleTime)) {
|
||
return null;
|
||
}
|
||
|
||
const [hourStr, minuteStr] = scheduleTime.split(':');
|
||
const hour = Number(hourStr);
|
||
const minute = Number(minuteStr);
|
||
|
||
if (
|
||
Number.isNaN(hour) ||
|
||
Number.isNaN(minute) ||
|
||
hour < 0 ||
|
||
hour > 23 ||
|
||
minute < 0 ||
|
||
minute > 59
|
||
) {
|
||
return null;
|
||
}
|
||
|
||
const scheduledAt = new Date(now);
|
||
scheduledAt.setHours(hour, minute, 0, 0);
|
||
|
||
return scheduledAt;
|
||
}
|
||
|
||
function isScheduleInTriggerWindow(scheduleTime, now, windowStartMs, windowEndMs) {
|
||
const scheduleDate = parseScheduleTimeToday(scheduleTime, now);
|
||
|
||
if (!scheduleDate) {
|
||
return {
|
||
match: false,
|
||
scheduleMs: null,
|
||
};
|
||
}
|
||
|
||
const scheduleMs = scheduleDate.getTime();
|
||
const match = scheduleMs >= windowStartMs && scheduleMs <= windowEndMs;
|
||
|
||
return {
|
||
match,
|
||
scheduleMs,
|
||
};
|
||
}
|
||
|
||
// ==================== WATERING WORKER ====================
|
||
|
||
const wateringWorker = new Worker(
|
||
'watering',
|
||
async (job) => {
|
||
const {
|
||
type,
|
||
potNumbers,
|
||
pompaAir,
|
||
pompaPupuk,
|
||
duration,
|
||
scheduleId,
|
||
thresholdId,
|
||
smartMode,
|
||
sensorData,
|
||
} = job.data;
|
||
|
||
console.log(`\n💧 Processing Job: ${job.id}`);
|
||
console.log(` Type: ${type}`);
|
||
console.log(` Pots: [${potNumbers.join(', ')}]`);
|
||
console.log(` Mode: ${smartMode ? 'SMART' : 'FIXED'}`);
|
||
console.log(` Duration: ${duration}s`);
|
||
console.log(` Break antar pot: ${config.worker.breakBetweenPotsMs / 1000} detik`);
|
||
|
||
if (sensorData) {
|
||
console.log(` Target: ${sensorData.batasBawah}% → ${sensorData.batasAtas}%`);
|
||
}
|
||
|
||
try {
|
||
const queueCounts = await wateringQueue.getJobCounts();
|
||
|
||
console.log('\n 📊 QUEUE STATUS AT JOB START:');
|
||
console.log(` Active jobs: ${queueCounts.active}`);
|
||
console.log(` Waiting jobs: ${queueCounts.waiting}`);
|
||
} catch (queueError) {
|
||
console.warn(` ⚠️ Could not get queue status: ${queueError.message}`);
|
||
}
|
||
|
||
try {
|
||
const potCount = potNumbers.length;
|
||
let totalWateringTime = 0;
|
||
|
||
console.log('\n🛡️ SAFETY: Ensuring all pot valves are OFF at start...');
|
||
|
||
await updateFirebaseSmart(
|
||
FIREBASE_PATHS.aktuator,
|
||
{
|
||
mosvet_3: false,
|
||
mosvet_4: false,
|
||
mosvet_5: false,
|
||
mosvet_6: false,
|
||
mosvet_7: false,
|
||
},
|
||
2
|
||
);
|
||
|
||
console.log('✅ All pot valves confirmed OFF');
|
||
await sleep(2000);
|
||
|
||
if (String(type || '').startsWith('waktu_')) {
|
||
const scheduleTime = job.data.scheduleTime || 'jadwal';
|
||
const potText =
|
||
potCount > 1 ? `pot ${potNumbers.join(', ')}` : `pot ${potNumbers[0]}`;
|
||
|
||
await sendAutomationNotification({
|
||
title: 'ApsGo - Jadwal Penyiraman',
|
||
body: `Penyiraman bergiliran untuk ${potText} dimulai dari jam ${scheduleTime}.`,
|
||
type: 'schedule_triggered',
|
||
data: {
|
||
scheduleId: scheduleId || '',
|
||
scheduleTime,
|
||
pots: potNumbers.join(','),
|
||
duration,
|
||
sequence: 'serial',
|
||
},
|
||
});
|
||
}
|
||
|
||
console.log(
|
||
`\n🔄 STARTING SEQUENTIAL WATERING: ${potCount} pot akan disiram bergiliran`
|
||
);
|
||
|
||
for (let potIndex = 0; potIndex < potCount; potIndex++) {
|
||
const pot = potNumbers[potIndex];
|
||
const potSequence = potIndex + 1;
|
||
|
||
console.log('\n ╔════════════════════════════════════════╗');
|
||
console.log(` ║ POT ${pot} [${potSequence}/${potCount}]`.padEnd(40) + ' ║');
|
||
console.log(' ╚════════════════════════════════════════╝');
|
||
|
||
const offOthersUpdates = {};
|
||
|
||
for (const otherPot of potNumbers) {
|
||
if (otherPot !== pot && otherPot >= 1 && otherPot <= 5) {
|
||
offOthersUpdates[`mosvet_${otherPot + 2}`] = false;
|
||
}
|
||
}
|
||
|
||
if (Object.keys(offOthersUpdates).length > 0) {
|
||
console.log(
|
||
` 🔴 Turn OFF other pots: ${Object.keys(offOthersUpdates).join(', ')}`
|
||
);
|
||
|
||
await updateFirebaseSmart(FIREBASE_PATHS.aktuator, offOthersUpdates, 2);
|
||
await sleep(1000);
|
||
}
|
||
|
||
const onUpdates = {};
|
||
|
||
if (pompaAir) onUpdates.mosvet_1 = true;
|
||
if (pompaPupuk) onUpdates.mosvet_2 = true;
|
||
|
||
if (pot >= 1 && pot <= 5) {
|
||
onUpdates[`mosvet_${pot + 2}`] = true;
|
||
}
|
||
|
||
console.log(` 🔛 Turn ON: ${Object.keys(onUpdates).join(', ')}`);
|
||
|
||
await updateFirebaseSmart(FIREBASE_PATHS.aktuator, onUpdates, 3);
|
||
|
||
let verified = false;
|
||
|
||
for (let i = 0; i < 3; i++) {
|
||
try {
|
||
const currentState = await readFirebaseSmart(FIREBASE_PATHS.aktuator);
|
||
|
||
const allSet = Object.keys(onUpdates).every(
|
||
(key) => currentState[key] === onUpdates[key]
|
||
);
|
||
|
||
if (allSet) {
|
||
console.log(' ✅ Verified: Valve ON successfully');
|
||
verified = true;
|
||
break;
|
||
}
|
||
|
||
console.warn(` ⚠️ Verification attempt ${i + 1}/3 failed`);
|
||
|
||
if (i < 2) {
|
||
await sleep(500);
|
||
}
|
||
} catch (verifyError) {
|
||
console.warn(
|
||
` ⚠️ Verification read failed attempt ${i + 1}/3: ${verifyError.message}`
|
||
);
|
||
}
|
||
}
|
||
|
||
if (!verified) {
|
||
console.warn(' ⚠️ Valve verification not fully confirmed, continuing...');
|
||
}
|
||
|
||
if (smartMode && sensorData && sensorData.batasAtas) {
|
||
const targetSoil = sensorData.batasAtas;
|
||
const maxDurationMs = duration * 1000;
|
||
const startTime = Date.now();
|
||
|
||
let potCompleted = false;
|
||
|
||
console.log(` 🎯 SMART MODE: Monitoring POT ${pot}, target ${targetSoil}%`);
|
||
|
||
while (Date.now() - startTime < maxDurationMs && !potCompleted) {
|
||
await sleep(2000);
|
||
|
||
try {
|
||
const currentSensorData = await readFirebaseSmart(FIREBASE_PATHS.data);
|
||
const soilKey = `soil_${pot}`;
|
||
const currentValue = parseInt(currentSensorData[soilKey]) || 0;
|
||
const elapsed = Math.floor((Date.now() - startTime) / 1000);
|
||
|
||
console.log(
|
||
` ⏳ [${elapsed}s] POT ${pot}: ${currentValue}% target ${targetSoil}%`
|
||
);
|
||
|
||
if (currentValue >= targetSoil) {
|
||
console.log(
|
||
` ✅ POT ${pot}: TARGET REACHED ${currentValue}% >= ${targetSoil}%`
|
||
);
|
||
|
||
potCompleted = true;
|
||
totalWateringTime += elapsed;
|
||
}
|
||
} catch (sensorError) {
|
||
console.warn(` ⚠️ Failed to read sensor: ${sensorError.message}`);
|
||
}
|
||
}
|
||
|
||
if (!potCompleted) {
|
||
console.log(` ⏱️ Max duration ${duration}s reached for POT ${pot}`);
|
||
totalWateringTime += duration;
|
||
}
|
||
} else {
|
||
console.log(` ⏱️ FIXED MODE: Watering POT ${pot} for ${duration}s`);
|
||
|
||
const startTime = Date.now();
|
||
const endTime = startTime + duration * 1000;
|
||
|
||
while (Date.now() < endTime) {
|
||
const remaining = Math.ceil((endTime - Date.now()) / 1000);
|
||
|
||
if (remaining % 10 === 0 || remaining <= 5) {
|
||
console.log(` ⏳ POT ${pot}: ${remaining}s remaining...`);
|
||
}
|
||
|
||
await sleep(1000);
|
||
}
|
||
|
||
totalWateringTime += duration;
|
||
}
|
||
|
||
const offUpdates = {};
|
||
|
||
if (pot >= 1 && pot <= 5) {
|
||
offUpdates[`mosvet_${pot + 2}`] = false;
|
||
}
|
||
|
||
if (pompaAir) offUpdates.mosvet_1 = false;
|
||
if (pompaPupuk) offUpdates.mosvet_2 = false;
|
||
|
||
console.log(
|
||
` 🔴 Turn OFF POT + PUMPS: ${Object.keys(offUpdates).join(', ')}`
|
||
);
|
||
|
||
await updateFirebaseSmart(FIREBASE_PATHS.aktuator, offUpdates, 2);
|
||
|
||
console.log(' ✅ Pot and pumps OFF confirmed');
|
||
|
||
if (potIndex < potCount - 1) {
|
||
console.log(' ⏳ Waiting 2 seconds to stabilize...');
|
||
await sleep(2000);
|
||
|
||
const nextPot = potNumbers[potIndex + 1];
|
||
const breakSeconds = config.worker.breakBetweenPotsMs / 1000;
|
||
|
||
console.log(`\n ⏸️ BREAK ${breakSeconds} DETIK sebelum POT ${nextPot}...`);
|
||
|
||
for (let breakTime = breakSeconds; breakTime > 0; breakTime--) {
|
||
if (breakTime % 10 === 0 || breakTime <= 5) {
|
||
console.log(` ⏳ Break: ${breakTime}s remaining...`);
|
||
}
|
||
|
||
await sleep(1000);
|
||
}
|
||
}
|
||
}
|
||
|
||
console.log('\n ✅ Semua pot selesai disiram!');
|
||
|
||
const pumpStop = {};
|
||
|
||
if (pompaAir) pumpStop.mosvet_1 = false;
|
||
if (pompaPupuk) pumpStop.mosvet_2 = false;
|
||
|
||
if (Object.keys(pumpStop).length > 0) {
|
||
console.log(` 🔴 Turning OFF pumps: ${Object.keys(pumpStop).join(', ')}`);
|
||
|
||
await updateFirebaseSmart(FIREBASE_PATHS.aktuator, pumpStop, 2);
|
||
|
||
console.log(' ✅ Pumps stopped');
|
||
}
|
||
|
||
console.log(
|
||
` 📝 Logging penyiraman history: ${potCount} pot, ${totalWateringTime}s total`
|
||
);
|
||
|
||
await logHistory(type, potNumbers, totalWateringTime);
|
||
|
||
console.log(' ✅ History penyiraman logged successfully');
|
||
|
||
if (thresholdId) {
|
||
lastThresholdTime[thresholdId] = Date.now();
|
||
|
||
console.log(` ⏰ Cooldown set for ${thresholdId} | 2 minutes`);
|
||
}
|
||
|
||
console.log(`\n ✅ Job completed successfully. Total: ${totalWateringTime}s`);
|
||
|
||
return {
|
||
success: true,
|
||
duration: totalWateringTime,
|
||
pots: potNumbers,
|
||
mode: 'sequential',
|
||
};
|
||
} catch (error) {
|
||
console.error(' ❌ Job failed:', error.message);
|
||
console.error(' [ERROR DETAILS] Stack:', error.stack);
|
||
|
||
const safetyUpdates = {
|
||
mosvet_1: false,
|
||
mosvet_2: false,
|
||
mosvet_3: false,
|
||
mosvet_4: false,
|
||
mosvet_5: false,
|
||
mosvet_6: false,
|
||
mosvet_7: false,
|
||
mosvet_8: false,
|
||
};
|
||
|
||
try {
|
||
console.log(' 🛡️ Safety: turning OFF all aktuators...');
|
||
|
||
await updateFirebaseSmart(FIREBASE_PATHS.aktuator, safetyUpdates, 2);
|
||
|
||
console.log(' 🛡️ Safety: all aktuators turned OFF successfully');
|
||
} catch (safetyError) {
|
||
console.error(' ❌ CRITICAL: Safety OFF failed:', safetyError.message);
|
||
console.error(' ❌ Penyiraman mungkin terjebak ON.');
|
||
}
|
||
|
||
throw error;
|
||
}
|
||
},
|
||
{
|
||
connection: redis,
|
||
concurrency: config.worker.concurrency,
|
||
lockDuration: 900000,
|
||
removeOnComplete: {
|
||
count: 100,
|
||
},
|
||
removeOnFail: {
|
||
count: 50,
|
||
},
|
||
}
|
||
);
|
||
|
||
wateringWorker.on('completed', (job) => {
|
||
console.log(`✅ Worker completed job ${job.id}`);
|
||
});
|
||
|
||
wateringWorker.on('failed', (job, err) => {
|
||
console.error(`❌ Worker failed job ${job?.id}:`, err.message);
|
||
});
|
||
|
||
console.log('\n🔧 WATERING WORKER CONFIGURATION:');
|
||
console.log(` Concurrency: ${config.worker.concurrency}`);
|
||
console.log(' Lock Duration: 900000ms');
|
||
console.log(' Remove on Complete: Keep 100 completed jobs');
|
||
console.log(' Remove on Fail: Keep 50 failed jobs');
|
||
|
||
if (config.worker.concurrency !== 1) {
|
||
console.error('❌ CRITICAL: Worker concurrency is NOT 1!');
|
||
}
|
||
|
||
console.log('\n💧 SEQUENTIAL WATERING MODE ENABLED');
|
||
console.log(' - Pots watered one by one');
|
||
console.log(` - Break between pots: ${config.worker.breakBetweenPotsMs / 1000}s`);
|
||
console.log(' - Applies to Waktu Mode and Sensor Mode');
|
||
console.log(' - Only 1 job runs at a time');
|
||
|
||
// ==================== WAKTU MODE ====================
|
||
|
||
let lastScheduleCheck = {};
|
||
let lastSchedulerTickAt = null;
|
||
let checkCounter = 0;
|
||
|
||
async function checkScheduledWatering() {
|
||
checkCounter++;
|
||
|
||
console.log(`\n🔎 checkScheduledWatering() called | Counter: ${checkCounter}`);
|
||
|
||
try {
|
||
const kontrolConfig = await fetchKontrolSmart();
|
||
|
||
if (!kontrolConfig) {
|
||
console.warn(' ⚠️ Kontrol config NULL. Schedule check aborted.');
|
||
return;
|
||
}
|
||
|
||
const now = new Date();
|
||
const checkEndMs = now.getTime();
|
||
|
||
const currentTime = `${now.getHours().toString().padStart(2, '0')}:${now
|
||
.getMinutes()
|
||
.toString()
|
||
.padStart(2, '0')}`;
|
||
|
||
const currentSeconds = now.getSeconds();
|
||
const dateKey = getDateKey(now);
|
||
|
||
const graceMs = config.worker.scheduleGraceMs;
|
||
const maxCatchupMs = config.worker.scheduleMaxCatchupMs;
|
||
const fallbackStartMs = checkEndMs - config.worker.checkInterval;
|
||
const baselineStartMs = lastSchedulerTickAt || fallbackStartMs;
|
||
const boundedStartMs = Math.max(baselineStartMs, checkEndMs - maxCatchupMs);
|
||
|
||
const triggerWindowStartMs = boundedStartMs - graceMs;
|
||
const triggerWindowEndMs = checkEndMs;
|
||
|
||
console.log(
|
||
`\n⏱️ CHECK #${checkCounter}: ${currentTime}:${currentSeconds
|
||
.toString()
|
||
.padStart(2, '0')} | Mode Waktu: ${kontrolConfig?.waktu ? 'ON' : 'OFF'}`
|
||
);
|
||
|
||
const allSchedules = Object.keys(kontrolConfig).filter((key) =>
|
||
key.startsWith('jadwal_')
|
||
);
|
||
|
||
if (allSchedules.length === 0) {
|
||
console.warn(' ⚠️ No jadwal found in kontrol config.');
|
||
}
|
||
|
||
if (checkCounter % 3 === 0 || now.getMinutes() % 5 === 0) {
|
||
console.log(` 📅 Date: ${dateKey}`);
|
||
console.log(` 🕐 Current: ${currentTime}`);
|
||
console.log(
|
||
` 🪟 Trigger window: ${new Date(
|
||
triggerWindowStartMs
|
||
).toLocaleTimeString('id-ID')} - ${new Date(
|
||
triggerWindowEndMs
|
||
).toLocaleTimeString('id-ID')}`
|
||
);
|
||
console.log(` 📊 API Stats: SDK=${sdkSuccessCount}, REST=${restFallbackCount}`);
|
||
console.log(` 📋 Total Jadwal: ${allSchedules.length}`);
|
||
|
||
if (kontrolConfig?.waktu && allSchedules.length > 0) {
|
||
allSchedules.forEach((scheduleKey) => {
|
||
const schedule = kontrolConfig[scheduleKey];
|
||
|
||
if (schedule && typeof schedule === 'object') {
|
||
const isActive = schedule.aktif !== false;
|
||
const waktu = schedule.waktu || 'not set';
|
||
const potAktif = schedule.pot_aktif || [];
|
||
|
||
console.log(
|
||
` ${isActive ? '✅' : '❌'} ${scheduleKey}: ${waktu} → Pot [${potAktif.join(
|
||
', '
|
||
)}]`
|
||
);
|
||
}
|
||
});
|
||
}
|
||
}
|
||
|
||
if (!kontrolConfig.waktu) {
|
||
console.log(' ⏭️ Mode waktu disabled. Skipping.');
|
||
return;
|
||
}
|
||
|
||
for (const scheduleKey of allSchedules) {
|
||
const schedule = kontrolConfig[scheduleKey];
|
||
|
||
if (!schedule || typeof schedule !== 'object') {
|
||
console.log(` ⚠️ ${scheduleKey}: invalid structure. Skipping.`);
|
||
continue;
|
||
}
|
||
|
||
const isActive = schedule.aktif !== false;
|
||
|
||
if (!isActive) {
|
||
continue;
|
||
}
|
||
|
||
const scheduleWaktu = schedule.waktu;
|
||
|
||
const { match: isInWindow, scheduleMs } = isScheduleInTriggerWindow(
|
||
scheduleWaktu,
|
||
now,
|
||
triggerWindowStartMs,
|
||
triggerWindowEndMs
|
||
);
|
||
|
||
if (!isInWindow) {
|
||
continue;
|
||
}
|
||
|
||
if (scheduleMs !== null) {
|
||
const delayedSec = Math.max(
|
||
0,
|
||
Math.floor((triggerWindowEndMs - scheduleMs) / 1000)
|
||
);
|
||
|
||
if (delayedSec > 0) {
|
||
console.log(
|
||
` ⏱️ ${scheduleKey}: Triggered with ${delayedSec}s delay`
|
||
);
|
||
}
|
||
}
|
||
|
||
const potAktif = schedule.pot_aktif || [];
|
||
const durasi = schedule.durasi || 60;
|
||
const pompaAir = schedule.pompa_air !== false;
|
||
const pompaPupuk = schedule.pompa_pupuk || false;
|
||
|
||
if (!Array.isArray(potAktif) || potAktif.length === 0) {
|
||
console.log(` ⚠️ ${scheduleKey}: no active pots. Skipping.`);
|
||
continue;
|
||
}
|
||
|
||
const normalizedScheduleTime = scheduleWaktu.replace(':', '_');
|
||
const jobKey = `${scheduleKey}_${dateKey}_${normalizedScheduleTime}`;
|
||
|
||
if (lastScheduleCheck[jobKey]) {
|
||
console.log(` ⏭️ ${scheduleKey} already triggered: ${jobKey}`);
|
||
continue;
|
||
}
|
||
|
||
console.log(`\n🕐 ${scheduleKey.toUpperCase()} TRIGGERED`);
|
||
console.log(` 🎯 Pot aktif: [${potAktif.join(', ')}]`);
|
||
console.log(` ⏱️ Durasi: ${durasi}s`);
|
||
console.log(` 💧 Pompa Air: ${pompaAir ? 'ON' : 'OFF'}`);
|
||
console.log(` 🌿 Pompa Pupuk: ${pompaPupuk ? 'ON' : 'OFF'}`);
|
||
|
||
try {
|
||
await wateringQueue.add(
|
||
scheduleKey,
|
||
{
|
||
type: `waktu_${scheduleKey}`,
|
||
potNumbers: potAktif,
|
||
pompaAir: pompaAir,
|
||
pompaPupuk: pompaPupuk,
|
||
duration: durasi,
|
||
scheduleId: jobKey,
|
||
scheduleTime: scheduleWaktu,
|
||
},
|
||
{
|
||
jobId: jobKey,
|
||
removeOnComplete: true,
|
||
}
|
||
);
|
||
|
||
lastScheduleCheck[jobKey] = true;
|
||
|
||
console.log(` ✅ Successfully added to queue: ${jobKey}`);
|
||
|
||
const queueStatus = await wateringQueue.getJobCounts();
|
||
|
||
console.log(
|
||
` 📊 Queue status: ${queueStatus.active} active, ${queueStatus.waiting} waiting`
|
||
);
|
||
} catch (queueError) {
|
||
console.error(` ❌ Failed to add ${scheduleKey}:`, queueError.message);
|
||
}
|
||
}
|
||
|
||
// Legacy support waktu_1 dan waktu_2
|
||
const legacySchedules = [
|
||
{
|
||
key: 'legacy_jadwal_1',
|
||
waktu: kontrolConfig.waktu_1,
|
||
durasi: kontrolConfig.durasi_1 || 60,
|
||
},
|
||
{
|
||
key: 'legacy_jadwal_2',
|
||
waktu: kontrolConfig.waktu_2,
|
||
durasi: kontrolConfig.durasi_2 || 60,
|
||
},
|
||
];
|
||
|
||
for (const legacy of legacySchedules) {
|
||
const legacyWindow = isScheduleInTriggerWindow(
|
||
legacy.waktu,
|
||
now,
|
||
triggerWindowStartMs,
|
||
triggerWindowEndMs
|
||
);
|
||
|
||
if (!legacyWindow.match) {
|
||
continue;
|
||
}
|
||
|
||
const legacyTimeKey = legacy.waktu.replace(':', '_');
|
||
const scheduleKey = `${legacy.key}_${dateKey}_${legacyTimeKey}`;
|
||
|
||
if (lastScheduleCheck[scheduleKey]) {
|
||
continue;
|
||
}
|
||
|
||
console.log(`\n🕐 [LEGACY] ${legacy.key.toUpperCase()} TRIGGERED`);
|
||
|
||
try {
|
||
await wateringQueue.add(
|
||
legacy.key,
|
||
{
|
||
type: 'waktu_jadwal_legacy',
|
||
potNumbers: [1, 2, 3, 4, 5],
|
||
pompaAir: true,
|
||
pompaPupuk: true,
|
||
duration: legacy.durasi,
|
||
scheduleId: scheduleKey,
|
||
scheduleTime: legacy.waktu,
|
||
},
|
||
{
|
||
jobId: scheduleKey,
|
||
removeOnComplete: true,
|
||
}
|
||
);
|
||
|
||
lastScheduleCheck[scheduleKey] = true;
|
||
|
||
console.log(` ✅ Successfully added legacy queue: ${scheduleKey}`);
|
||
} catch (queueError) {
|
||
console.error(' ❌ Failed legacy queue:', queueError.message);
|
||
}
|
||
}
|
||
|
||
for (const key in lastScheduleCheck) {
|
||
if (key.includes(dateKey)) {
|
||
continue;
|
||
}
|
||
|
||
delete lastScheduleCheck[key];
|
||
}
|
||
} catch (error) {
|
||
console.error('❌ Error checking scheduled watering:', error.message);
|
||
console.error('[DEBUG] Stack:', error.stack);
|
||
} finally {
|
||
lastSchedulerTickAt = Date.now();
|
||
}
|
||
}
|
||
// ==================== START WAKTU MODE SCHEDULER ====================
|
||
|
||
setInterval(async () => {
|
||
try {
|
||
await checkScheduledWatering();
|
||
} catch (error) {
|
||
console.error('❌ Error in scheduled check interval:', error.message);
|
||
}
|
||
}, config.worker.checkInterval);
|
||
|
||
console.log(
|
||
`✅ Waktu Mode scheduler started every ${config.worker.checkInterval / 1000}s`
|
||
);
|
||
|
||
setTimeout(async () => {
|
||
try {
|
||
console.log('\n🚀 Running first schedule check...');
|
||
await checkScheduledWatering();
|
||
console.log('✅ First schedule check completed');
|
||
} catch (error) {
|
||
console.error('❌ First schedule check failed:', error.message);
|
||
}
|
||
}, 8000);
|
||
|
||
// ==================== SENSOR MODE ====================
|
||
|
||
let sensorCheckCounter = 0;
|
||
|
||
async function checkSensorThresholds() {
|
||
sensorCheckCounter++;
|
||
|
||
try {
|
||
const sensorData = await readFirebaseSmart(FIREBASE_PATHS.data);
|
||
|
||
if (!sensorData) {
|
||
console.log('⚠️ Sensor data kosong. ESP32 mungkin belum mengirim data.');
|
||
return;
|
||
}
|
||
|
||
const kontrolConfig = await fetchKontrolSmart();
|
||
|
||
if (!kontrolConfig) {
|
||
console.log('⚠️ Kontrol config kosong.');
|
||
return;
|
||
}
|
||
|
||
if (!kontrolConfig.otomatis) {
|
||
if (sensorCheckCounter % 10 === 0) {
|
||
console.log('⚠️ Sensor mode disabled. Skipping threshold check.');
|
||
}
|
||
|
||
return;
|
||
}
|
||
|
||
const allThresholds = Object.keys(kontrolConfig).filter((key) =>
|
||
key.startsWith('threshold_')
|
||
);
|
||
|
||
console.log(
|
||
`\n🌡️ SENSOR CHECK #${sensorCheckCounter} | Total Thresholds: ${allThresholds.length}`
|
||
);
|
||
|
||
if (allThresholds.length === 0) {
|
||
console.log(' ⚠️ No thresholds configured.');
|
||
return;
|
||
}
|
||
|
||
for (const thresholdKey of allThresholds) {
|
||
const threshold = kontrolConfig[thresholdKey];
|
||
|
||
console.log(`\n 🔍 Checking ${thresholdKey}`);
|
||
|
||
if (!threshold || !threshold.aktif) {
|
||
console.log(
|
||
` ❌ Skipped: ${!threshold ? 'not found' : 'aktif=false'}`
|
||
);
|
||
continue;
|
||
}
|
||
|
||
const lastTime = lastThresholdTime[thresholdKey];
|
||
|
||
if (lastTime && Date.now() - lastTime < config.worker.sensorDebounce) {
|
||
const remainingSeconds = Math.ceil(
|
||
(config.worker.sensorDebounce - (Date.now() - lastTime)) / 1000
|
||
);
|
||
|
||
console.log(
|
||
` ⏳ ${thresholdKey}: cooldown aktif ${remainingSeconds}s`
|
||
);
|
||
|
||
continue;
|
||
}
|
||
|
||
const batasBawah = threshold.batas_bawah || 30;
|
||
const batasAtas = threshold.batas_atas || 70;
|
||
const durasi = threshold.durasi || 600;
|
||
const smartMode = threshold.smart_mode === true;
|
||
const potAktif = threshold.pot_aktif || [];
|
||
const pompaAir = threshold.pompa_air === true;
|
||
const pompaPupuk = threshold.pompa_pupuk === true;
|
||
|
||
const potsNeedWatering = [];
|
||
const potDetails = [];
|
||
|
||
for (const potNumber of potAktif) {
|
||
if (potNumber < 1 || potNumber > 5) {
|
||
console.log(` ⚠️ POT ${potNumber}: invalid`);
|
||
continue;
|
||
}
|
||
|
||
const soilKey = `soil_${potNumber}`;
|
||
const soilValue = parseInt(sensorData[soilKey]) || 0;
|
||
|
||
console.log(
|
||
` 🌱 POT ${potNumber}: ${soilValue}% | Threshold ${batasBawah}-${batasAtas}%`
|
||
);
|
||
|
||
if (soilValue >= batasAtas) {
|
||
console.log(
|
||
` ✅ POT ${potNumber}: skip, sudah basah ${soilValue}%`
|
||
);
|
||
continue;
|
||
}
|
||
|
||
if (soilValue < batasBawah) {
|
||
console.log(
|
||
` 🚨 POT ${potNumber} kering: ${soilValue}% < ${batasBawah}%`
|
||
);
|
||
|
||
potsNeedWatering.push(potNumber);
|
||
potDetails.push({
|
||
pot: potNumber,
|
||
value: soilValue,
|
||
});
|
||
} else {
|
||
console.log(` ✅ POT ${potNumber}: OK`);
|
||
}
|
||
}
|
||
|
||
if (potsNeedWatering.length > 0) {
|
||
console.log(`\n🌡️ THRESHOLD TRIGGERED: ${thresholdKey}`);
|
||
console.log(` Pots needing water: [${potsNeedWatering.join(', ')}]`);
|
||
console.log(` Mode: ${smartMode ? 'Smart' : 'Fixed'}`);
|
||
console.log(` Durasi: ${durasi}s`);
|
||
console.log(` Pompa Air: ${pompaAir}`);
|
||
console.log(` Pompa Pupuk: ${pompaPupuk}`);
|
||
|
||
await sendAutomationNotification({
|
||
title: 'ApsGo - Penyiraman Otomatis',
|
||
body: `Penyiraman dilakukan pada pot ${potsNeedWatering.join(
|
||
', '
|
||
)} karena kelembapan di bawah ${batasBawah}%.`,
|
||
type: 'sensor_triggered',
|
||
data: {
|
||
thresholdId: thresholdKey,
|
||
pots: potsNeedWatering.join(','),
|
||
batasBawah,
|
||
batasAtas,
|
||
durasi,
|
||
mode: smartMode ? 'smart' : 'fixed',
|
||
},
|
||
});
|
||
|
||
const jobId = `${thresholdKey}-${Date.now()}`;
|
||
|
||
await wateringQueue.add(
|
||
thresholdKey,
|
||
{
|
||
type: 'sensor_threshold',
|
||
potNumbers: potsNeedWatering,
|
||
pompaAir: pompaAir,
|
||
pompaPupuk: pompaPupuk,
|
||
duration: durasi,
|
||
scheduleId: jobId,
|
||
thresholdId: thresholdKey,
|
||
smartMode: smartMode,
|
||
sensorData: {
|
||
batasBawah,
|
||
batasAtas,
|
||
mode: smartMode ? 'smart' : 'fixed',
|
||
potValues: potDetails,
|
||
},
|
||
},
|
||
{
|
||
jobId,
|
||
removeOnComplete: true,
|
||
priority: 1,
|
||
}
|
||
);
|
||
|
||
console.log(` 📌 Added to queue: ${jobId}`);
|
||
}
|
||
}
|
||
} catch (error) {
|
||
console.error('❌ Error in sensor threshold check:', error.message);
|
||
}
|
||
}
|
||
|
||
async function setupSensorMonitoring() {
|
||
console.log('\n🌡️ SENSOR MODE ENABLED');
|
||
console.log('✅ Polling every 30 seconds');
|
||
console.log(`📍 Config path: /${FIREBASE_PATHS.kontrol}`);
|
||
|
||
setInterval(async () => {
|
||
try {
|
||
await checkSensorThresholds();
|
||
} catch (error) {
|
||
console.error('❌ Polling sensor check failed:', error.message);
|
||
}
|
||
}, 30000);
|
||
|
||
setTimeout(async () => {
|
||
try {
|
||
console.log('🚀 Running first sensor check...');
|
||
await checkSensorThresholds();
|
||
console.log('✅ First sensor check completed');
|
||
} catch (error) {
|
||
console.error('❌ First sensor check failed:', error.message);
|
||
}
|
||
}, 10000);
|
||
|
||
try {
|
||
db.ref(FIREBASE_PATHS.data).on(
|
||
'value',
|
||
async () => {
|
||
console.log('🔔 Firebase data listener triggered');
|
||
await checkSensorThresholds();
|
||
},
|
||
(error) => {
|
||
console.error('❌ Firebase listener error:', error.message);
|
||
}
|
||
);
|
||
|
||
console.log('✅ Firebase data listener attached');
|
||
} catch (error) {
|
||
console.log('⚠️ Listener failed, polling tetap berjalan.');
|
||
}
|
||
}
|
||
|
||
setupSensorMonitoring();
|
||
|
||
// ==================== PERIODIC MONITORING HISTORY ====================
|
||
|
||
const autoLogJob = new cron.CronJob('*/30 * * * *', async () => {
|
||
await logMonitoringHistory();
|
||
});
|
||
|
||
autoLogJob.start();
|
||
|
||
console.log('✅ Auto monitoring history started every 30 minutes');
|
||
|
||
// ==================== CLEANUP OLD HISTORY ====================
|
||
|
||
const cleanupJob = new cron.CronJob('0 2 * * *', async () => {
|
||
try {
|
||
console.log('\n🧹 Running history cleanup...');
|
||
|
||
const daysToKeep = 10;
|
||
const cutoffDate = new Date();
|
||
|
||
cutoffDate.setDate(cutoffDate.getDate() - daysToKeep);
|
||
|
||
const historyData = await readFirebaseSmart(FIREBASE_PATHS.history);
|
||
|
||
if (!historyData) {
|
||
console.log('🧹 History empty. Cleanup skipped.');
|
||
return;
|
||
}
|
||
|
||
let deletedCount = 0;
|
||
|
||
for (const historyId in historyData) {
|
||
const item = historyData[historyId];
|
||
|
||
if (!item || !item.timestamp) {
|
||
continue;
|
||
}
|
||
|
||
const itemDate = new Date(item.timestamp);
|
||
|
||
if (itemDate < cutoffDate) {
|
||
const url = `${config.firebase.databaseURL}/${FIREBASE_PATHS.history}/${historyId}.json`;
|
||
|
||
await fetchWithTimeout2(
|
||
url,
|
||
{
|
||
method: 'DELETE',
|
||
},
|
||
10000
|
||
);
|
||
|
||
deletedCount++;
|
||
|
||
console.log(` 🗑️ Deleted history item: ${historyId}`);
|
||
}
|
||
}
|
||
|
||
console.log(`✅ Cleanup completed: ${deletedCount} item removed`);
|
||
} catch (error) {
|
||
console.error('❌ Cleanup failed:', error.message);
|
||
}
|
||
});
|
||
|
||
cleanupJob.start();
|
||
|
||
console.log('✅ History cleanup scheduled daily at 2 AM');
|
||
|
||
// ==================== UTILITIES ====================
|
||
|
||
function sleep(ms) {
|
||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||
}
|
||
|
||
// ==================== DIAGNOSTIC FUNCTIONS ====================
|
||
|
||
async function checkAktuatorNode() {
|
||
try {
|
||
console.log('\n🔍 Checking aktuator node...');
|
||
|
||
const aktuatorData = await readFirebaseSmart(FIREBASE_PATHS.aktuator);
|
||
|
||
if (!aktuatorData) {
|
||
console.log('❌ Aktuator node not found. Creating default...');
|
||
|
||
await setFirebaseSmart(FIREBASE_PATHS.aktuator, {
|
||
mosvet_1: false,
|
||
mosvet_2: false,
|
||
mosvet_3: false,
|
||
mosvet_4: false,
|
||
mosvet_5: false,
|
||
mosvet_6: false,
|
||
mosvet_7: false,
|
||
mosvet_8: false,
|
||
});
|
||
|
||
console.log('✅ Aktuator node created');
|
||
return;
|
||
}
|
||
|
||
const required = [
|
||
'mosvet_1',
|
||
'mosvet_2',
|
||
'mosvet_3',
|
||
'mosvet_4',
|
||
'mosvet_5',
|
||
'mosvet_6',
|
||
'mosvet_7',
|
||
'mosvet_8',
|
||
];
|
||
|
||
const missing = required.filter((key) => !(key in aktuatorData));
|
||
|
||
if (missing.length > 0) {
|
||
const updates = {};
|
||
|
||
missing.forEach((key) => {
|
||
updates[key] = false;
|
||
});
|
||
|
||
await updateFirebaseSmart(FIREBASE_PATHS.aktuator, updates);
|
||
|
||
console.log(`✅ Missing mosvets added: ${missing.join(', ')}`);
|
||
} else {
|
||
console.log('✅ Aktuator node complete');
|
||
}
|
||
} catch (error) {
|
||
console.error('❌ Aktuator check failed:', error.message);
|
||
}
|
||
}
|
||
|
||
async function showCurrentTime() {
|
||
try {
|
||
const now = new Date();
|
||
|
||
console.log('\n🕐 CURRENT TIME ANALYSIS');
|
||
console.log(` Server Local: ${now.toString()}`);
|
||
console.log(
|
||
` Asia/Jakarta: ${now.toLocaleString('id-ID', {
|
||
timeZone: 'Asia/Jakarta',
|
||
})}`
|
||
);
|
||
console.log(` ISO: ${now.toISOString()}`);
|
||
console.log(` Unix: ${now.getTime()}`);
|
||
console.log(` TZ Env: ${process.env.TZ}`);
|
||
console.log(` HH:MM: ${getWaktuIndonesia(now)}`);
|
||
} catch (error) {
|
||
console.error('❌ Time check failed:', error.message);
|
||
}
|
||
}
|
||
|
||
async function healthCheck() {
|
||
try {
|
||
const firebaseOk = !!config.firebase.databaseURL;
|
||
|
||
await redis.ping();
|
||
|
||
const queueStatus = await wateringQueue.getJobCounts();
|
||
|
||
console.log('\n💚 HEALTH CHECK');
|
||
console.log(` Firebase: ${firebaseOk ? '✅' : '❌'}`);
|
||
console.log(' Redis: ✅');
|
||
console.log(
|
||
` Queue: ${queueStatus.active} active, ${queueStatus.waiting} waiting`
|
||
);
|
||
} catch (error) {
|
||
console.error('❤️🩹 HEALTH CHECK FAILED:', error.message);
|
||
}
|
||
}
|
||
|
||
// ==================== HEALTH CHECK ====================
|
||
|
||
setInterval(healthCheck, 300000);
|
||
|
||
setTimeout(healthCheck, 5000);
|
||
|
||
// ==================== GRACEFUL SHUTDOWN ====================
|
||
|
||
async function shutdown() {
|
||
console.log('\n🛑 Shutting down gracefully...');
|
||
|
||
try {
|
||
await wateringWorker.close();
|
||
console.log('✅ Worker closed');
|
||
|
||
await wateringQueue.close();
|
||
console.log('✅ Queue closed');
|
||
|
||
await redis.quit();
|
||
console.log('✅ Redis disconnected');
|
||
|
||
await admin.app().delete();
|
||
console.log('✅ Firebase disconnected');
|
||
|
||
process.exit(0);
|
||
} catch (error) {
|
||
console.error('❌ Shutdown error:', error.message);
|
||
process.exit(1);
|
||
}
|
||
}
|
||
|
||
process.on('SIGTERM', shutdown);
|
||
process.on('SIGINT', shutdown);
|
||
|
||
// ==================== PREVENT CRASHES ====================
|
||
|
||
process.on('uncaughtException', (error) => {
|
||
console.error('❌ Uncaught Exception:', error.message);
|
||
console.error(error.stack);
|
||
console.log('⚠️ Worker continuing despite error...');
|
||
});
|
||
|
||
process.on('unhandledRejection', (reason, promise) => {
|
||
console.error('❌ Unhandled Rejection at:', promise);
|
||
console.error('Reason:', reason);
|
||
console.log('⚠️ Worker continuing despite rejection...');
|
||
});
|
||
|
||
// ==================== STARTUP COMPLETE ====================
|
||
|
||
console.log('\n✨ ApsGo Railway Worker is running!');
|
||
console.log('📊 Features enabled:');
|
||
console.log(' • Waktu Mode');
|
||
console.log(' • Sensor Mode');
|
||
console.log(' • History Penyiraman dengan jenis_histori');
|
||
console.log(' • History Monitoring setiap 30 menit');
|
||
console.log(' • Cleanup History');
|
||
console.log(' • Health Check');
|
||
console.log('\n🎯 Worker is ready to process jobs...\n');
|
||
|
||
// ==================== STARTUP DIAGNOSTICS ====================
|
||
|
||
setTimeout(async () => {
|
||
try {
|
||
console.log('🔍 Verifying Firebase connection...');
|
||
|
||
const snapshot = await fetchWithTimeout(
|
||
db.ref(FIREBASE_PATHS.kontrol),
|
||
10000
|
||
);
|
||
|
||
const data = snapshot.val();
|
||
|
||
if (data) {
|
||
console.log(`✅ Firebase /${FIREBASE_PATHS.kontrol} readable`);
|
||
console.log(` Mode waktu: ${data.waktu ? 'ENABLED' : 'DISABLED'}`);
|
||
console.log(` Mode sensor: ${data.otomatis ? 'ENABLED' : 'DISABLED'}`);
|
||
} else {
|
||
console.log(`⚠️ Firebase /${FIREBASE_PATHS.kontrol} empty`);
|
||
}
|
||
} catch (error) {
|
||
console.error('❌ Firebase verification failed:', error.message);
|
||
}
|
||
}, 3000);
|
||
|
||
setTimeout(async () => {
|
||
try {
|
||
console.log('\n🔧 RUNNING DIAGNOSTIC CHECKS...');
|
||
await showCurrentTime();
|
||
await checkAktuatorNode();
|
||
console.log('✅ Diagnostic checks completed');
|
||
} catch (error) {
|
||
console.error('❌ Diagnostic checks failed:', error.message);
|
||
}
|
||
}, 5000);
|
||
|
||
// ==================== KEEP ALIVE ====================
|
||
|
||
setInterval(() => {
|
||
const uptime = Math.floor(process.uptime());
|
||
const hours = Math.floor(uptime / 3600);
|
||
const minutes = Math.floor((uptime % 3600) / 60);
|
||
|
||
console.log(`💓 Heartbeat: Worker alive for ${hours}h ${minutes}m`);
|
||
}, 30000);
|
||
|
||
setInterval(() => {
|
||
const now = new Date();
|
||
|
||
if (now.getMinutes() % 10 === 0 && now.getSeconds() < 30) {
|
||
showCurrentTime();
|
||
}
|
||
}, 30000); |