feat: implement real-time stock updates via Socket.IO

Add real-time stock synchronization across POS terminals when invoices
are submitted. This ensures all terminals see updated stock quantities
immediately without manual refresh.

Backend:
- Add realtime_events.py with stock update event emitter
- Use optimized bulk SQL query for performance
- Register doc_events hooks for Sales Invoice on_submit/on_cancel
- Broadcast pos_stock_update events via Socket.IO

Frontend:
- Add useRealtimeStock composable for event management
- Implement warehouse-based filtering to prevent data corruption
- Update reactive arrays in-place for immediate UI refresh
- Add worker support for IndexedDB cache updates
- Integrate stock update listener in POSSale.vue

Fixes:
- Prevent multi-warehouse data corruption with warehouse filtering
- Ensure UI refreshes by mutating reactive arrays directly
- Clean and efficient with minimal logging

Author: Ahmed Osama

POS/src/composables/useRealtimeStock.js

@@ -0,0 +1,192 @@
+/**
+ * Real-time Stock Updates Composable
+ *
+ * Listens to Socket.IO events for stock changes and notifies registered handlers.
+ * Provides intelligent event management with deduplication and batching.
+ * Each handler is responsible for filtering by warehouse and updating its cache.
+ */
+
+import { ref, onUnmounted } from 'vue'
+
+// Shared state across all instances
+const isListening = ref(false)
+const eventHandlers = new Set()
+const pendingUpdates = new Map()
+let batchTimeout = null
+
+/**
+ * Batch update configuration
+ */
+const BATCH_DELAY_MS = 500 // Wait 500ms before applying batched updates
+const MAX_BATCH_SIZE = 100 // Maximum items to batch before forcing update
+
+/**
+ * Process pending stock updates in batch
+ */
+async function processBatchedUpdates() {
+	if (pendingUpdates.size === 0) {
+		return
+	}
+
+	const updates = Array.from(pendingUpdates.values())
+	pendingUpdates.clear()
+
+	try {
+		// Notify all registered handlers
+		// Each handler can filter by warehouse before applying updates
+		eventHandlers.forEach(handler => {
+			try {
+				handler(updates)
+			} catch (error) {
+				console.error('[Realtime Stock] Handler error:', error)
+			}
+		})
+	} catch (error) {
+		console.error('[Realtime Stock] Failed to process batch updates:', error)
+	}
+}
+
+/**
+ * Schedule batch processing
+ */
+function scheduleBatchUpdate() {
+	if (batchTimeout) {
+		clearTimeout(batchTimeout)
+	}
+
+	// Force update if batch is getting too large
+	if (pendingUpdates.size >= MAX_BATCH_SIZE) {
+		processBatchedUpdates()
+		return
+	}
+
+	batchTimeout = setTimeout(() => {
+		processBatchedUpdates()
+		batchTimeout = null
+	}, BATCH_DELAY_MS)
+}
+
+/**
+ * Handle incoming stock update event
+ */
+function handleStockUpdate(data) {
+	if (!data || !data.stock_updates) {
+		return
+	}
+
+	// Add updates to pending batch (deduplicate by item_code + warehouse)
+	data.stock_updates.forEach(update => {
+		const key = `${update.item_code}|${update.warehouse}`
+		pendingUpdates.set(key, update)
+	})
+
+	scheduleBatchUpdate()
+}
+
+/**
+ * Handle invoice created event (optional, for future use)
+ */
+function handleInvoiceCreated(data) {
+	// Can be used to update sales dashboards, notifications, etc.
+}
+
+/**
+ * Start listening to real-time events
+ */
+function startListening() {
+	if (isListening.value) {
+		return
+	}
+
+	if (!window.frappe?.realtime) {
+		console.warn('[Realtime Stock] Socket.IO not available')
+		return
+	}
+
+	// Subscribe to stock update events
+	window.frappe.realtime.on('pos_stock_update', handleStockUpdate)
+	window.frappe.realtime.on('pos_invoice_created', handleInvoiceCreated)
+
+	isListening.value = true
+}
+
+/**
+ * Stop listening to real-time events
+ */
+function stopListening() {
+	if (!isListening.value) {
+		return
+	}
+
+	if (window.frappe?.realtime) {
+		window.frappe.realtime.off('pos_stock_update', handleStockUpdate)
+		window.frappe.realtime.off('pos_invoice_created', handleInvoiceCreated)
+	}
+
+	// Clear pending updates
+	if (batchTimeout) {
+		clearTimeout(batchTimeout)
+		batchTimeout = null
+	}
+	pendingUpdates.clear()
+
+	isListening.value = false
+}
+
+/**
+ * Flush pending updates immediately
+ */
+async function flushUpdates() {
+	if (batchTimeout) {
+		clearTimeout(batchTimeout)
+		batchTimeout = null
+	}
+	await processBatchedUpdates()
+}
+
+/**
+ * Main composable
+ */
+export function useRealtimeStock() {
+	/**
+	 * Register a callback to be notified of stock updates
+	 * @param {Function} handler - Called with array of stock updates
+	 * @returns {Function} Cleanup function to unregister handler
+	 */
+	function onStockUpdate(handler) {
+		if (typeof handler !== 'function') {
+			throw new Error('Handler must be a function')
+		}
+
+		eventHandlers.add(handler)
+
+		// Start listening when first handler is registered
+		if (eventHandlers.size === 1) {
+			startListening()
+		}
+
+		// Return cleanup function
+		return () => {
+			eventHandlers.delete(handler)
+
+			// Stop listening when last handler is removed
+			if (eventHandlers.size === 0) {
+				stopListening()
+			}
+		}
+	}
+
+	// Cleanup on component unmount
+	onUnmounted(() => {
+		// Note: We don't stop listening here because other components
+		// might still be using it. Each handler cleans up individually.
+	})
+
+	return {
+		isListening,
+		onStockUpdate,
+		flushUpdates,
+		startListening,
+		stopListening
+	}
+}

POS/src/pages/POSSale.vue

@@ -620,6 +620,8 @@ import CreateCustomerDialog from "@/components/sale/CreateCustomerDialog.vue"
 import ItemSelectionDialog from "@/components/sale/ItemSelectionDialog.vue"
 import PromotionManagement from "@/components/sale/PromotionManagement.vue"
 import { printInvoiceByName } from "@/utils/printInvoice"
+import { useRealtimeStock } from "@/composables/useRealtimeStock"
+import { offlineWorker } from "@/utils/offline/workerClient"
 
 // Pinia Stores
 import { usePOSCartStore } from "@/stores/posCart"
@@ -637,6 +639,9 @@ const offlineStore = usePOSSyncStore()
 const draftsStore = usePOSDraftsStore()
 const itemStore = useItemSearchStore()
 
+// Real-time stock updates
+const { onStockUpdate } = useRealtimeStock()
+
 // Component refs
 const itemsSelectorRef = ref(null)
 const offersDialogRef = ref(null)
@@ -716,6 +721,29 @@ onMounted(async () => {
 	}
 	window.addEventListener('resize', handleResize, { passive: true })
 
+	// Set up real-time stock update listener
+	const cleanup = onStockUpdate(async (stockUpdates) => {
+		// Filter updates to only include items from our warehouse(s)
+		const profileWarehouses = shiftStore.profileWarehouse
+			? [shiftStore.profileWarehouse]
+			: warehousesList.value.map(w => w.warehouse_name || w.name)
+
+		const relevantUpdates = stockUpdates.filter(update =>
+			profileWarehouses.includes(update.warehouse)
+		)
+
+		if (relevantUpdates.length > 0) {
+			// Update IndexedDB cache with filtered updates (warehouse-specific)
+			await offlineWorker.updateStockQuantities(relevantUpdates)
+
+			// Apply stock updates directly to reactive arrays for immediate UI refresh
+			itemStore.applyStockUpdates(relevantUpdates)
+		}
+	})
+
+	// Store cleanup function for unmount
+	onUnmounted(cleanup)
+
 	try {
 		// Start timers for current time and shift duration
 		shiftStore.startTimers()

POS/src/stores/itemSearch.js

@@ -539,6 +539,47 @@ export const useItemSearchStore = defineStore('itemSearch', () => {
 		posProfile.value = profile
 	}
 
+	function invalidateCache() {
+		// Clear result cache to force UI refresh with updated stock
+		resultCache.value.clear()
+	}
+
+	function applyStockUpdates(stockUpdates) {
+		// Apply stock updates directly to reactive arrays for immediate UI refresh
+		if (!stockUpdates || stockUpdates.length === 0) {
+			return
+		}
+
+		// Build lookup map for O(1) access
+		const updateMap = new Map()
+		stockUpdates.forEach(update => {
+			updateMap.set(update.item_code, update)
+		})
+
+		// Update allItems array in-place
+		allItems.value.forEach(item => {
+			const update = updateMap.get(item.item_code)
+			if (update) {
+				item.actual_qty = update.actual_qty
+				item.stock_qty = update.stock_qty
+				item.warehouse = update.warehouse
+			}
+		})
+
+		// Update searchResults array in-place
+		searchResults.value.forEach(item => {
+			const update = updateMap.get(item.item_code)
+			if (update) {
+				item.actual_qty = update.actual_qty
+				item.stock_qty = update.stock_qty
+				item.warehouse = update.warehouse
+			}
+		})
+
+		// Clear result cache to force recomputation
+		resultCache.value.clear()
+	}
+
 	return {
 		// State
 		allItems,
@@ -576,6 +617,8 @@ export const useItemSearchStore = defineStore('itemSearch', () => {
 		startBackgroundCacheSync,
 		stopBackgroundCacheSync,
 		cleanup,
+		invalidateCache,
+		applyStockUpdates,
 
 		// Resources
 		itemGroupsResource,

POS/src/utils/offline/workerClient.js

@@ -139,6 +139,10 @@ class OfflineWorkerClient {
 		return this.sendMessage('SET_MANUAL_OFFLINE', { value })
 	}
 
+	async updateStockQuantities(stockUpdates) {
+		return this.sendMessage('UPDATE_STOCK_QUANTITIES', { stockUpdates })
+	}
+
 	terminate() {
 		if (this.worker) {
 			this.worker.terminate()

POS/src/workers/offline.worker.js

@@ -342,6 +342,49 @@ async function deleteOfflineInvoice(id) {
 	}
 }
 
+// Update stock quantities in cached items
+async function updateStockQuantities(stockUpdates) {
+	try {
+		const db = await initDB()
+
+		if (!stockUpdates || stockUpdates.length === 0) {
+			return { success: true, updated: 0 }
+		}
+
+		let updatedCount = 0
+
+		// Process each stock update
+		for (const update of stockUpdates) {
+			const { item_code, warehouse, actual_qty, stock_qty } = update
+
+			if (!item_code) {
+				continue
+			}
+
+			// Get the cached item
+			const item = await db.items.get(item_code)
+
+			if (!item) {
+				continue
+			}
+
+			// Update stock quantities for this warehouse
+			item.actual_qty = actual_qty !== undefined ? actual_qty : stock_qty
+			item.stock_qty = stock_qty !== undefined ? stock_qty : actual_qty
+			item.warehouse = warehouse || item.warehouse
+
+			// Save updated item back to cache
+			await db.items.put(item)
+			updatedCount++
+		}
+
+		return { success: true, updated: updatedCount }
+	} catch (error) {
+		console.error('Worker: Error updating stock quantities:', error)
+		throw error
+	}
+}
+
 // Message handler
 self.onmessage = async (event) => {
 	const { type, payload, id } = event.data
@@ -403,6 +446,10 @@ self.onmessage = async (event) => {
 				result = { success: true, manualOffline }
 				break
 
+			case 'UPDATE_STOCK_QUANTITIES':
+				result = await updateStockQuantities(payload.stockUpdates)
+				break
+
 			default:
 				throw new Error(`Unknown message type: ${type}`)
 		}

pos_next/hooks.py

@@ -179,6 +179,11 @@
 doc_events = {
 	"Item": {
 		"validate": "pos_next.validations.validate_item"
+	},
+	"Sales Invoice": {
+		"on_submit": "pos_next.realtime_events.emit_stock_update_event",
+		"on_cancel": "pos_next.realtime_events.emit_stock_update_event",
+		"after_insert": "pos_next.realtime_events.emit_invoice_created_event"
 	}
 }