{
"name": "Intelligent_Shopify_Sync_Scheduler",
"nodes": [
{
"name": "Dynamic_Schedule_Trigger",
"type": "n8n-nodes-base.cron",
"parameters": {
"rule": {
"minute": "*/5"
}
}
},
{
"name": "Calculate_Sync_Strategy",
"type": "n8n-nodes-base.function",
"parameters": {
"functionCode": "// Tính toán strategy sync thông minh\nconst currentHour = new Date().getHours();\nconst currentMinute = new Date().getMinutes();\nconst dayOfWeek = new Date().getDay();\n\n// Business hours detection\nconst isBusinessHours = (currentHour >= 9 && currentHour <= 17) && (dayOfWeek >= 1 && dayOfWeek <= 5);\n\n// Sync strategies\nconst strategies = {\n // High frequency trong business hours\n high_frequency: {\n enabled: isBusinessHours,\n interval_minutes: 5,\n max_records: 1000,\n priority_filter: [1, 2]\n },\n \n // Medium frequency ngoài business hours\n medium_frequency: {\n enabled: !isBusinessHours && currentHour >= 8 && currentHour <= 22,\n interval_minutes: 15,\n max_records: 500,\n priority_filter: [1]\n },\n \n // Low frequency ban đêm\n low_frequency: {\n enabled: currentHour < 8 || currentHour > 22,\n interval_minutes: 60,\n max_records: 100,\n priority_filter: [1]\n },\n \n // Deep sync vào cuối tuần\n deep_sync: {\n enabled: dayOfWeek === 0 && currentHour === 2, // Chủ nhật 2AM\n interval_minutes: 0,\n max_records: 10000,\n priority_filter: [1, 2, 3],\n full_sync: true\n }\n};\n\n// Chọn strategy phù hợp\nconst activeStrategy = Object.keys(strategies).find(key => strategies[key].enabled);\n\nif (!activeStrategy) {\n return [{ skip: true, reason: 'No active strategy' }];\n}\n\nconst strategy = strategies[activeStrategy];\n\nreturn [{\n strategy_name: activeStrategy,\n strategy: strategy,\n should_run: true,\n current_time: new Date().toISOString(),\n business_hours: isBusinessHours\n}];"
}
},
{
"name": "Should_Run_Check",
"type": "n8n-nodes-base.if",
"parameters": {
"conditions": {
"boolean": [
{
"value1": "{{ $json.should_run }}",
"operation": "equal",
"value2": true
}
]
}
}
},
{
"name": "Get_Entities_To_Sync",
"type": "n8n-nodes-base.postgres",
"parameters": {
"operation": "select",
"query": "SELECT \n entity_type,\n entity_id,\n parent_entity_type,\n parent_entity_id,\n last_updated_at,\n sync_priority,\n data_checksum\nFROM entity_sync_tracker \nWHERE needs_sync = TRUE \n AND sync_priority = ANY($1::int[])\nORDER BY sync_priority ASC, last_updated_at DESC\nLIMIT $2",
"values": [
"{{ $json.strategy.priority_filter }}",
"{{ $json.strategy.max_records }}"
]
}
},
{
"name": "Group_By_Entity_Type",
"type": "n8n-nodes-base.function",
"parameters": {
"functionCode": "// Nhóm entities theo type và tạo batch\nconst entityGroups = {\n orders: [],\n customers: [],\n products: [],\n refunds: [],\n fulfillments: [],\n returns: []\n};\n\n// Group entities\nfor (const entity of $input.all()) {\n if (entityGroups[entity.entity_type]) {\n entityGroups[entity.entity_type].push(entity);\n }\n}\n\n// Tạo execution plan\nconst executionPlan = [];\n\n// Orders first (vì các entity khác depend vào orders)\nif (entityGroups.orders.length > 0) {\n executionPlan.push({\n entity_type: 'orders',\n entities: entityGroups.orders,\n batch_size: 50 // Shopify orders API limit\n });\n}\n\n// Sau đó sync các entity con\n['refunds', 'fulfillments', 'returns'].forEach(type => {\n if (entityGroups[type].length > 0) {\n executionPlan.push({\n entity_type: type,\n entities: entityGroups[type],\n batch_size: 100\n });\n }\n});\n\n// Cuối cùng sync customers và products\n['customers', 'products'].forEach(type => {\n if (entityGroups[type].length > 0) {\n executionPlan.push({\n entity_type: type,\n entities: entityGroups[type],\n batch_size: 100\n });\n }\n});\n\nreturn executionPlan;"
}
},
{
"name": "Execute_Sync_Batches",
"type": "n8n-nodes-base.function",
"parameters": {
"functionCode": "// Execute sync cho từng batch\nconst results = [];\n\nfor (const batch of $input.all()) {\n const batchResult = await this.executeBatch(batch);\n results.push(batchResult);\n \n // Rate limiting between batches\n await new Promise(resolve => setTimeout(resolve, 500));\n}\n\nreturn results;\n\n// Helper function\nthis.executeBatch = async function(batch) {\n const { entity_type, entities, batch_size } = batch;\n const results = [];\n \n // Chia nhỏ thành các sub-batches\n for (let i = 0; i < entities.length; i += batch_size) {\n const subBatch = entities.slice(i, i + batch_size);\n \n switch (entity_type) {\n case 'orders':\n const orderResults = await this.syncOrders(subBatch);\n results.push(...orderResults);\n break;\n \n case 'refunds':\n const refundResults = await this.syncRefunds(subBatch);\n results.push(...refundResults);\n break;\n \n case 'fulfillments':\n const fulfillmentResults = await this.syncFulfillments(subBatch);\n results.push(...fulfillmentResults);\n break;\n \n case 'returns':\n const returnResults = await this.syncReturns(subBatch);\n results.push(...returnResults);\n break;\n }\n \n // Rate limiting between sub-batches\n await new Promise(resolve => setTimeout(resolve, 200));\n }\n \n return {\n entity_type,\n total_processed: results.length,\n success_count: results.filter(r => r.success).length,\n error_count: results.filter(r => !r.success).length,\n results\n };\n};\n\n// Sync functions\nthis.syncOrders = async function(orders) {\n const results = [];\n \n for (const orderTracker of orders) {\n try {\n // Fetch full order từ Shopify\n const order = await this.fetchShopifyOrder(orderTracker.entity_id);\n \n // Fetch child entities\n const [refunds, fulfillments, returns] = await Promise.all([\n this.fetchOrderRefunds(orderTracker.entity_id),\n this.fetchOrderFulfillments(orderTracker.entity_id),\n this.fetchOrderReturns(orderTracker.entity_id)\n ]);\n \n // Tạo comprehensive checksum\n const newChecksum = this.createComprehensiveChecksum(order, {\n refunds, fulfillments, returns\n });\n \n // Chỉ update nếu có thay đổi\n if (newChecksum !== orderTracker.data_checksum) {\n await this.updateOrderInDatabase(order, refunds, fulfillments, returns, newChecksum);\n \n results.push({\n entity_id: orderTracker.entity_id,\n success: true,\n updated: true,\n changes_detected: true\n });\n } else {\n results.push({\n entity_id: orderTracker.entity_id,\n success: true,\n updated: false,\n changes_detected: false\n });\n }\n \n } catch (error) {\n results.push({\n entity_id: orderTracker.entity_id,\n success: false,\n error: error.message\n });\n }\n }\n \n return results;\n};\n\nthis.fetchShopifyOrder = async function(orderId) {\n const response = await fetch(`https://${process.env.SHOPIFY_SHOP}.myshopify.com/admin/api/2024-01/orders/${orderId}.json`, {\n method: 'GET',\n headers: {\n 'X-Shopify-Access-Token': process.env.SHOPIFY_ACCESS_TOKEN\n }\n });\n \n return response.json().order;\n};\n\nthis.createComprehensiveChecksum = function(order, childEntities) {\n const crypto = require('crypto');\n \n const checksumData = {\n order: {\n status: order.status,\n financial_status: order.financial_status,\n fulfillment_status: order.fulfillment_status,\n total_price: order.total_price,\n updated_at: order.updated_at\n },\n refunds: childEntities.refunds?.map(r => ({\n id: r.id,\n amount: r.amount,\n processed_at: r.processed_at\n })) || [],\n fulfillments: childEntities.fulfillments?.map(f => ({\n id: f.id,\n status: f.status,\n tracking_number: f.tracking_number,\n updated_at: f.updated_at\n })) || [],\n returns: childEntities.returns?.map(r => ({\n id: r.id,\n status: r.status\n })) || []\n };\n \n return crypto.createHash('md5').update(JSON.stringify(checksumData)).digest('hex');\n};"
}
},
{
"name": "Update_Sync_Status",
"type": "n8n-nodes-base.postgres",
"parameters": {
"operation": "update",
"query": "UPDATE entity_sync_tracker \nSET needs_sync = FALSE,\n last_synced_at = NOW(),\n data_checksum = $2\nWHERE entity_type = $3 AND entity_id = $4",
"values": [
"{{ $json.success }}",
"{{ $json.new_checksum }}",
"{{ $json.entity_type }}",
"{{ $json.entity_id }}"
]
}
},
{
"name": "Log_Sync_Results",
"type": "n8n-nodes-base.function",
"parameters": {
"functionCode": "// Log kết quả sync\nconst summary = {\n sync_time: new Date().toISOString(),\n strategy_used: $node['Calculate_Sync_Strategy'].json[0].strategy_name,\n total_entities: $input.all().length,\n success_count: $input.all().filter(r => r.success).length,\n error_count: $input.all().filter(r => !r.success).length,\n entities_with_changes: $input.all().filter(r => r.changes_detected).length\n};\n\nconsole.log('Sync Summary:', JSON.stringify(summary, null, 2));\n\n// Log errors nếu có\nconst errors = $input.all().filter(r => !r.success);\nif (errors.length > 0) {\n console.log('Sync Errors:', JSON.stringify(errors, null, 2));\n}\n\nreturn [summary];"
}
},
{
"name": "Schedule_Next_Sync",
"type": "n8n-nodes-base.function",
"parameters": {
"functionCode": "// Tự động schedule next sync dựa trên kết quả\nconst results = $input.all()[0];\nconst strategy = $node['Calculate_Sync_Strategy'].json[0].strategy;\n\n// Điều chỉnh frequency dựa trên số lượng changes\nlet nextSyncMinutes = strategy.interval_minutes;\n\nif (results.entities_with_changes > 100) {\n // Nhiều changes -> tăng frequency\n nextSyncMinutes = Math.max(nextSyncMinutes / 2, 2);\n} else if (results.entities_with_changes === 0) {\n // Không có changes -> giảm frequency\n nextSyncMinutes = Math.min(nextSyncMinutes * 1.5, 60);\n}\n\n// Update next sync time trong database\nconst nextSyncTime = new Date(Date.now() + nextSyncMinutes * 60000);\n\nreturn [{\n next_sync_time: nextSyncTime.toISOString(),\n adjusted_interval: nextSyncMinutes,\n reason: results.entities_with_changes > 100 ? 'high_activity' : \n results.entities_with_changes === 0 ? 'low_activity' : 'normal'\n}];"
}
}
],
"connections": {
"Dynamic_Schedule_Trigger": {
"main": [["Calculate_Sync_Strategy"]]
},
"Calculate_Sync_Strategy": {
"main": [["Should_Run_Check"]]
},
"Should_Run_Check": {
"main": [
["Get_Entities_To_Sync"],
[]
]
},
"Get_Entities_To_Sync": {
"main": [["Group_By_Entity_Type"]]
},
"Group_By_Entity_Type": {
"main": [["Execute_Sync_Batches"]]
},
"Execute_Sync_Batches": {
"main": [["Update_Sync_Status"]]
},
"Update_Sync_Status": {
"main": [["Log_Sync_Results"]]
},
"Log_Sync_Results": {
"main": [["Schedule_Next_