main.go

/*
Copyright 2025      Leil Storage OÜ

LeilFS is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, version 3.

This program is is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
package main

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"encoding/binary"
	"fmt"
	"io"
	"net"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/spf13/pflag"

	"github.com/leil-io/leilfs-inotifier/config"
	"github.com/leil-io/leilfs-inotifier/interfaces"
	"github.com/leil-io/leilfs-inotifier/logger"
	natsclient "github.com/leil-io/leilfs-inotifier/nats"
	"github.com/leil-io/leilfs-inotifier/operations"
	"github.com/leil-io/leilfs-inotifier/parser"
	"github.com/leil-io/leilfs-inotifier/protocol"
)

// Processor serves as the central coordination point for the LeilFS inotifier system,
// integrating configuration, network communication, event processing, and external
// systems into a unified processing pipeline.
//
// Pipeline stages:
//   - Ingestion: Network protocol with SFS master server
//   - Parsing: Changelog to structured operations
//   - Enrichment: Path resolution and filesystem context
//   - Processing: Business logic execution
//   - Publication: NATS messaging
//   - Caching: Performance optimization
type Processor struct {
	config       *config.Config                      // Application configuration
	conn         net.Conn                            // SFS master connection
	tlsConfig    *tls.Config                         // TLS configuration
	natsClient   *natsclient.NATSClient              // NATS message bus client
	opHandler    *operations.OperationHandler        // Operation processing
	protocol     *protocol.Protocol                  // NTTOMA protocol state machine
	sfsConnected bool                                // Connection health state
	stopChan     chan struct{}                       // Graceful shutdown signal
	responseChan chan *interfaces.InodeTypePath      // Path resolution responses
	cache        map[uint64]*interfaces.InodeTypePath // Path resolution cache
	pendingReqs  map[uint64]*pendingRequest          // Outstanding requests
	mu           sync.RWMutex                        // Thread safety
}

// pendingRequest represents an outstanding path resolution request within the
// asynchronous protocol communication system.
//
// Fields:
//   - response: Buffered channel (capacity 1) for delivering resolution results
//     from protocol layer back to the requestor. Carries InodeTypePath on success
//     or nil on failure.
//   - createdAt: Timestamp of request initiation for timeout detection and
//     latency monitoring.
type pendingRequest struct {
	response chan *interfaces.InodeTypePath
	createdAt time.Time
}

// NewProcessor creates and initializes a new Processor instance with comprehensive
// subsystem integration and dependency injection.
//
// Initialization sequence:
//   1. Create NATS client for event publication
//   2. Setup operation handler with configuration
//   3. Initialize protocol layer
//   4. Load TLS config (if enabled)
//   5. Assemble processor with all dependencies
//
// Returns fully initialized processor ready for operation, or error on
// NATS connection failure, TLS config loading error, or component initialization issues.
func NewProcessor(cfg *config.Config) (*Processor, error) {
	natsClient, err := natsclient.NewNATSClient(cfg.NATSURL, cfg.NATSSubject)
	if err != nil {
		logger.Errorf("Failed to create NATS client: %v", err)
		return nil, err
	}

	opHandler := operations.NewOperationHandler(cfg)
	opHandler.SetLogger(logger.GetLogger())

	operations.InitOperationHandlers(parser.GetGlobalRegistry(), opHandler)
	pt := protocol.NewProtocol(opHandler)

	var tlsConfig *tls.Config
	if cfg.UseTLS {
		tlsConfig, err = loadTLSConfig(cfg)
		if err != nil {
			logger.Errorf("Failed to load TLS configuration: %v", err)
			return nil, err
		}
	}

	processor := &Processor{
		config: cfg,
		tlsConfig: tlsConfig,
		natsClient: natsClient,
		opHandler: opHandler,
		protocol: pt,
		stopChan: make(chan struct{}),
		responseChan: make(chan *interfaces.InodeTypePath, 10),
		cache: make(map[uint64]*interfaces.InodeTypePath),
		pendingReqs: make(map[uint64]*pendingRequest),
	}

	opHandler.SetPathTranslator(processor)

	logger.Info("Processor created successfully")
	return processor, nil
}

// loadTLSConfig creates and configures a TLS configuration for connecting to the SFS master server.
//
// Configuration steps:
//   1. Resolve master host IP for ServerName
//   2. Load client certificate and key pair
//   3. Load and verify CA certificate for root CAs
//
// Returns TLS config with TLS 1.2 minimum version, or error if host resolution,
// certificate loading, or CA verification fails
func loadTLSConfig(cfg *config.Config) (*tls.Config, error) {
	ips, err := net.LookupIP(cfg.SFSMaster.Host)
	if err != nil || len(ips) == 0 {
		return nil, fmt.Errorf("failed to resolve SFS master host: %w", err)
	}
	tlsConfig := &tls.Config{
		InsecureSkipVerify: cfg.TLSConfig.InsecureSkipVerify,
		ServerName: ips[0].String(),
		MinVersion: tls.VersionTLS12,
	}

	cert, err := tls.LoadX509KeyPair(cfg.TLSConfig.TLSCertFile, cfg.TLSConfig.TLSKeyFile)
	if err != nil {
		return nil, fmt.Errorf("failed to load TLS certificate/key: %w", err)
	}
	tlsConfig.Certificates = []tls.Certificate{cert}

	caCert, err := os.ReadFile(cfg.TLSConfig.TLSCAFile)
	if err != nil {
		return nil, fmt.Errorf("failed to read TLS CA file: %w", err)
	}
	caCertPool := x509.NewCertPool()
	if !caCertPool.AppendCertsFromPEM(caCert) {
		return nil, fmt.Errorf("failed to append CA certificate to pool")
	}
	tlsConfig.RootCAs = caCertPool

	return tlsConfig, nil
}

// Exit performs graceful shutdown and resource cleanup for the Processor instance.
//
// Shutdown sequence:
//   1. Close SFS master network connection
//   2. Signal goroutines to terminate via stopChan
//   3. Close pending request channels
//   4. Clear cache and pending request maps
//
// Thread-safe and idempotent (uses sync.Once). Does not block waiting for goroutines.
func (p *Processor) Exit() {
	var once sync.Once
	once.Do(func() {
		p.mu.Lock()
		defer p.mu.Unlock()

		if p.conn != nil {
			p.conn.Close()
			p.conn = nil
		}

		close(p.stopChan)

		for inode, req := range p.pendingReqs {
			if req != nil && req.response != nil {
				close(req.response)
			}
			delete(p.pendingReqs, inode)
		}

		p.cache = make(map[uint64]*interfaces.InodeTypePath)
	})
}

func (p *Processor) SetLogger(l logger.Logger) {
	// Processor uses the package logger, so we can ignore this
}

// ResolveInode resolves an inode number to its filesystem path and type with
// comprehensive caching, concurrency management, and timeout handling.
//
// Resolution pipeline:
//   1. Cache lookup (double-checked locking)
//   2. Join pending request if one exists (deduplication)
//   3. Send NTTOMA_GET_PATH_TYPE_INODE request to master
//   4. Wait for async response with timeout
//
// Returns:
//   - Cached response on cache hit
//   - Joined response on concurrent request
//   - Fresh response after master query
//   - Error on timeout (1s query, 10s join), shutdown, or protocol failure
//
// Caching: All successful resolutions except Unknown/Deleted types are cached.
func (p *Processor) ResolveInode(inode uint64) (*interfaces.InodeTypePath, error) {
	if cached, exists := p.cache[inode]; exists {
		return cached, nil
	}

	p.mu.Lock()
	if cached, exists := p.cache[inode]; exists {
		p.mu.Unlock()
		return cached, nil
	}

	if pending, exists := p.pendingReqs[inode]; exists && pending != nil {
		p.mu.Unlock()

		select {
		case response := <-pending.response:
			if response != nil {
				p.cache[inode] = response
				return response, nil
			}
			return nil, fmt.Errorf("failed to resolve inode %d", inode)
		case <-time.After(10 * time.Second):
			return nil, fmt.Errorf("timeout waiting for inode %d resolution", inode)
		case <-p.stopChan:
			return nil, fmt.Errorf("processor stopped while resolving inode %d", inode)
		}
	}

	requestData := make([]byte, 8)
	binary.BigEndian.PutUint64(requestData, inode)

	requestPacket := &protocol.Packet{
		Type: protocol.NTTOMA_GET_PATH_TYPE_INODE,
		Size: uint32(len(requestData)),
		Data: requestData,
	}

	respChan := make(chan *interfaces.InodeTypePath, 1)
	p.pendingReqs[inode] = &pendingRequest{
		response: respChan,
		createdAt: time.Now(),
	}
	p.mu.Unlock()

	if err := p.protocol.WritePacket(p.conn, requestPacket); err != nil {
		p.mu.Lock()
		delete(p.pendingReqs, inode)
		p.mu.Unlock()
		return nil, fmt.Errorf("failed to send inode path/type request: %w", err)
	}

	select {
	case response := <-respChan:
		p.mu.Lock()
		delete(p.pendingReqs, inode)
		p.mu.Unlock()
		if response == nil {
			return nil, fmt.Errorf("failed to resolve inode %d", inode)
		}
		if response.Type != interfaces.FileTypeUnknown &&
			response.Type != interfaces.FileTypeDeleted {
			p.cache[inode] = response
		}
		return response, nil
	case <-time.After(1 * time.Second): ///TODO Query Timeout from config
		logger.Warn("***TIMEOUT*****Waiting for inode:", inode)
		p.mu.Lock()
		delete(p.pendingReqs, inode)
		p.mu.Unlock()
		return nil, fmt.Errorf("timeout waiting for inode %d resolution", inode)
	case <-p.stopChan:
		p.mu.Lock()
		delete(p.pendingReqs, inode)
		p.mu.Unlock()
		return nil, fmt.Errorf("processor stopped while resolving inode %d", inode)
	}
}

// HandlePathTypeResponse processes NTTOMA protocol responses for path/type resolution
// requests and delivers results to waiting consumers.
//
// Response processing:
//   1. Extract inode (bytes 0-7, big-endian)
//   2. Decode file type (byte 8)
//   3. Parse null-terminated path (bytes 9+)
//   4. Deliver to pending request via channel or cache for future use
//
// Packet structure:
//   Bytes 0-7:  inode number
//   Byte 8:     file type
//   Bytes 9+:   null-terminated path (optional)
//
// Returns error if packet data length < 8 bytes.
func (p *Processor) HandlePathTypeResponse(packet *protocol.Packet) error {
	if len(packet.Data) < 8 {
		return fmt.Errorf("invalid packet data length: %d", len(packet.Data))
	}

	inode := binary.BigEndian.Uint64(packet.Data[0:8])

	var response *interfaces.InodeTypePath

	if len(packet.Data) < 9 {
		response = &interfaces.InodeTypePath{
			Inode: inode,
			Type: interfaces.FileTypeUnknown,
			Path: "",
		}
	} else {
		fileTypeByte := packet.Data[8]
		ft := p.nodeTypeToFileType(fileTypeByte)
		path := ""
		if len(packet.Data) > 9 {
			for i := 9; i < len(packet.Data); i++ {
				if packet.Data[i] == protocol.LogNullTerminator {
					path = string(packet.Data[9:i])
					break
				}
			}
			if path == "" && len(packet.Data) > 9 {
				path = string(packet.Data[9:])
			}
		}

		response = &interfaces.InodeTypePath{
			Inode: inode,
			Type: ft,
			Path: path,
		}
	}

	p.mu.Lock()
	pending, exists := p.pendingReqs[inode]
	p.mu.Unlock()

	if exists && pending != nil {   // Non-blocking send to avoid deadlocks
		select {
		case pending.response <- response:
			logger.Debugf("Response for inode: %d, type: %s, path: `%s`",
				response.Inode, string(response.Type), response.Path)
		default:
			logger.Warn("Failed to deliver response (channel blocked) for inode:", inode)
			// The ResolveInode goroutine might have timed out or exited
		}
	} else {
		logger.Warn("No pending request found for inode:", inode)
		// Cache the response anyway for future use
		p.cache[inode] = response
	}

	return nil
}

// nodeTypeToFileType converts a protocol-specific node type byte into
// the internal FileType representation.
//
// Type mapping:
//   'd' → FileTypeDir      'l' → FileTypeSymlink
//   'f' → FileTypeFile     't' → FileTypeDeleted
//   *   → FileTypeUnknown
//
// Returns corresponding FileType, or FileTypeUnknown for unrecognized types.
func (p *Processor) nodeTypeToFileType(nodeType byte) interfaces.FileType {
	switch nodeType {
	case 'd':
		return interfaces.FileTypeDir
	case 'f':
		return interfaces.FileTypeFile
	case 'l':
		return interfaces.FileTypeSymlink
	case 't':
		return interfaces.FileTypeDeleted
	default:
		return interfaces.FileTypeUnknown
	}
}

// InvalidateCache removes a specific inode entry from the resolution cache.
//
// This method selectively deletes the cached path/type information for the
// given inode, forcing subsequent ResolveInode calls to fetch fresh data
// from the master server.
//
// Thread-safe via mutex protection.
//
// Parameters:
//   - inode: The inode number whose cache entry should be invalidated
func (p *Processor) InvalidateCache(inode uint64) {
	p.mu.Lock()
	defer p.mu.Unlock()
	delete(p.cache, inode)
}

// ClearCache immediately invalidates and clears the entire inode resolution cache.
//
// Performs atomic cache replacement with a new empty map, forcing subsequent
// ResolveInode calls to query the master server for fresh data.
//
// Use cases: Memory management, forced cache refresh, testing, recovery after
// master server restarts.
func (p *Processor) ClearCache() {
	p.cache = make(map[uint64]*interfaces.InodeTypePath)
}

// ProcessChangelog validates, parses, and processes raw changelog entries from
// the SFS master server, implementing the core changelog ingestion pipeline.
//
// Validation steps:
//   1. Verify minimum packet length
//   2. Handle force-rotate signals (warn, return empty)
//   3. Check expected size match
//   4. Validate header tag and separator format
//
// Returns formatted changelog entry string or error on validation failure.
func (p *Processor) ProcessChangelog(rawLine []byte, length uint32) (string, error) {
	dataLength := len(rawLine)

	if dataLength < protocol.LogMinimumLength {
		logger.Errorf("MATONF_METACHANGES_LOG - wrong size: %d bytes", dataLength)
		return "", fmt.Errorf("invalid changelog packet size: %d bytes", dataLength)
	}
	if dataLength == 1 && rawLine[protocol.LogHeaderTagOffset] == protocol.LogForceRotate {
		logger.Warn("MATONF_FORCE_LOG_ROTATE received - not implemented")
		return "", nil
	}
	if dataLength != int(length) {
		logger.Errorf("Data length mismatch: expected %d bytes, got %d bytes", length, dataLength)
		return "", fmt.Errorf("data length mismatch: expected %d bytes, got %d bytes", length, dataLength)
	}
	if dataLength == 0 {
		logger.Errorf("ProcessChangelog Not Implemented yet...")
		return "", fmt.Errorf("received empty data")
	}
	if rawLine[protocol.LogHeaderTagOffset] != protocol.LogHeaderTag {
		logger.Error("MATONF_METACHANGE_LOG - wrong packet")
		return "", fmt.Errorf("invalid changelog packet tag")
	}

	version := binary.BigEndian.Uint64(rawLine[protocol.LogVersionOffset:protocol.LogTimestampOffset])
	timestamp := string(rawLine[protocol.LogTimestampOffset:protocol.LogSeparatorOffset])
	separator := string(rawLine[protocol.LogSeparatorOffset])
	if separator != protocol.LogSeparatorStr {
		logger.Errorf("Invalid changelog format: missing '|' separator")
		return "", fmt.Errorf("invalid changelog format: missing '|' separator")
	}

	payload := string(rawLine[protocol.LogDataOffset:dataLength])
	logger.Debugf("Received changelog rawline: %d: %s%s%s", version, timestamp, separator, payload)
	return fmt.Sprintf("%d: %s%s%s", version, timestamp, separator, payload), nil
}

// Register establishes and maintains persistent connection to the SFS master server
// with robust retry logic and graceful shutdown handling.
//
// Registration flow:
//   1. TCP dial to SFS master (retries on connection refused)
//   2. Enable TCP no-delay for low latency
//   3. TLS handshake (if configured)
//   4. Send NTTOMA registration packet
//   5. Mark processor as connected
//
// Retry strategy:
//   - Connection refused: Continuous retry with 1-second interval
//   - Other errors: Immediate failure
//   - Context/stopChan aware for graceful shutdown
//
// Returns error on registration failure (network errors, protocol failures, TLS issues),
// nil on successful registration.
func (p *Processor) Register(ctx context.Context) error {
	var err error
	var conn net.Conn
	retryInterval := time.Second
	var tcpAddr string = net.JoinHostPort(p.config.SFSMaster.Host, p.config.SFSMaster.Port)

	for {
		select {
		case <-p.stopChan:
			logger.Info("Stop Processor and exiting service")
			return nil
		case <-ctx.Done():
			return fmt.Errorf("context cancelled while trying to connect to sfsmaster")
		default:
			conn, err = net.Dial(config.NetworkType, tcpAddr)
			if err != nil {
				if isConnectionRefused(err) {
					logger.Debugf("SFSMaster not available at %s, retrying in %s...",
					tcpAddr, retryInterval)
					select {
					case <-time.After(retryInterval):
						continue
					case <-ctx.Done():
						return fmt.Errorf("context cancelled while trying to connect to sfsmaster")
					}
				}
				logger.Errorf("Failed to connect to SFSMaster at %s: %v", tcpAddr, err)
				return fmt.Errorf("failed to connect to sfsmaster: %w", err)
			}
			p.conn = conn
			if tcpConn, ok := p.conn.(*net.TCPConn); ok {
				tcpConn.SetNoDelay(true)
			}

			if p.config.UseTLS { // TLS connection
				tlsAddr := net.JoinHostPort(p.tlsConfig.ServerName, p.config.SFSMaster.Port)
				logger.Debugf("Attempting to connect to SFSMaster at %s (TLS: %v)", tlsAddr, p.config.UseTLS)
				tlsMessage, err := p.protocol.CreateStartTLSPacket()
				if err != nil {
					conn.Close()
					p.conn = nil
					logger.Errorf("Failed to create TLS start packet: %v", err)
					return fmt.Errorf("failed to create TLS start packet: %w", err)
				}
				if err := p.protocol.WritePacket(p.conn, tlsMessage); err != nil {
					conn.Close()
					p.conn = nil
					logger.Errorf("Failed to send TLS start packet to SFSMaster at %s: %v", tcpAddr, err)
					return fmt.Errorf("failed to send TLS start packet to sfsmaster: %w", err)
				}

				tlsConn := tls.Client(conn, p.tlsConfig)

				if err := tlsConn.SetDeadline(time.Now().Add(30 * time.Second)); err != nil {
					conn.Close()
					p.conn = nil
					logger.Errorf("Failed to set TLS handshake deadline for SFSMaster at %s: %v", tcpAddr, err)
					return fmt.Errorf("failed to set TLS handshake deadline: %w", err)
				}
				if err := tlsConn.Handshake(); err != nil {
					conn.Close()
					p.conn = nil
					if nerErr, ok := err.(net.Error); ok && nerErr.Timeout() {
						logger.Errorf("TLS handshake timed out with SFSMaster at %s: %v", tcpAddr, err)
						return fmt.Errorf("tls handshake timed out with sfsmaster: %w", err)
					}
					logger.Errorf("TLS handshake failed with SFSMaster at %s: %v", tcpAddr, err)
					return fmt.Errorf("tls handshake failed with sfsmaster: %w", err)
				}
				if err := tlsConn.SetDeadline(time.Time{}); err != nil {
					logger.Warnf("Failed to clear TLS connection deadline for SFSMaster at %s: %v", tcpAddr, err)
				}

				p.conn = tlsConn
				logger.Debugf("Successfully established TLS connection to SFSMaster at %s", tlsAddr)

				if err := p.verifyTLSConnection(tlsConn); err != nil {
					conn.Close()
					logger.Errorf("TLS verification failed for SFSMaster at %s: %v", tlsAddr, err)
					return fmt.Errorf("tls verification failed for sfsmaster: %w", err)
				}
			}

			message, err := p.protocol.CreateRegisterPacket()
			if err != nil {
				p.conn.Close()
				p.conn = nil
				logger.Errorf("Failed to create register packet: %v", err)
				return fmt.Errorf("failed to create register packet: %w", err)
			}

			if err := p.protocol.WritePacket(p.conn, message); err != nil {
				p.conn.Close()
				p.conn = nil
				logger.Errorf("Failed to send register to SFSMaster at %s: %v", tcpAddr, err)
				return fmt.Errorf("failed to send register to sfsmaster: %w", err)
			}

			logger.Debugf("Registration packet sent, waiting for response...")

			p.sfsConnected = true
			logger.Infof("Registered with SFSMaster at %s", tcpAddr)
			return nil
		}
	}
}

// Close gracefully terminates the connection to the SFS master server
// and updates the processor's connection state.
//
// Termination sequence:
//   1. Close underlying TCP connection
//   2. Clear connection reference and set sfsConnected to false
//   3. Log closure status
//
// Idempotent: Safe to call multiple times (returns nil if already closed).
//
// Returns error if connection closure fails, nil on success.
func (p *Processor) Close() error {
	if p.conn != nil {
		err := p.conn.Close()
		p.conn = nil
		p.sfsConnected = false
		if err != nil {
			logger.Warnf("Failed to close connection to: %v",
				net.JoinHostPort(p.config.SFSMaster.Host, p.config.SFSMaster.Port))
			return err
		}
		logger.Infof("Connection to %v closed",
			net.JoinHostPort(p.config.SFSMaster.Host, p.config.SFSMaster.Port))
	}
	return nil
}

// createOperationMessage routes parsed changelog entries to appropriate operation handlers
// based on their operation tag, implementing the strategy pattern for filesystem operations.
//
// Routing logic:
//   1. Look up specialized handler by operation tag
//   2. Use default handler as fallback for unregistered operations
//
// Returns structured operation object and human-readable description.
func (p *Processor) createOperationMessage(parsed *interfaces.ParsedLine) (interfaces.Operation, string) {
	psr := parsed.Parser

	if handler, exists := parser.GetHandler(psr.OperationTag()); exists {
		return handler.Handle(psr)
	}

	return p.opHandler.HandleDefaultOperation(psr)
}

// ProcessLine executes the complete processing pipeline for a parsed changelog entry,
// transforming it into a structured operation and publishing it to NATS.
//
// Processing pipeline:
//   1. Create operation and message from parsed line
//   2. Skip empty messages (optimization for no-op operations)
//   3. Publish to NATS (either as NATS event or raw message)
//
// Returns error on NATS publication failure, nil on successful processing.
func (p *Processor) ProcessLine(parsedLine *interfaces.ParsedLine) error {

	op, message := p.createOperationMessage(parsedLine)
	if message == "" {
		return nil
	}
	parsedLine.Operation = op

	if eventOp, ok := op.(interface{ GetNatsEvent() *natsclient.NATsEvent }); ok {
		if natsEvent := eventOp.GetNatsEvent(); natsEvent != nil {
			if err := p.natsClient.PublishEvent(natsEvent); err != nil {
				logger.Errorf("Failed to publish event to NATS: %v", err)
				return err
			}
			return nil
		}
	}

	if err := p.natsClient.Publish(message); err != nil {
		logger.Errorf("Failed to publish to NATS: %v", err)
		return err
	}

	return nil
}

// listenToSFSMaster establishes a continuous bidirectional stream processor for
// communicating with the SFS master server using a dual-goroutine architecture.
//
// Architecture:
//   - Reader goroutine: Reads packets, handles path/type responses immediately,
//     routes other packets to a buffered channel (capacity 10)
//   - Processor goroutine: Handles non-path packets asynchronously
//
// Packet routing:
//   - Path/Type responses: Immediate handling for low latency
//   - Other packets: Async processing via channel with backpressure
//
// Returns error on connection failure (EOF, network errors) or nil on graceful shutdown.
func (p *Processor) listenToSFSMaster(ctx context.Context) error {
	if p.conn == nil {
		return fmt.Errorf("not connected to sfsmaster")
	}

	logger.Info("Listening for events from sfsmaster")

	listenerCtx, cancelListener := context.WithCancel(ctx)
	defer cancelListener()

	done := make(chan error, 1)
	packetChan := make(chan *protocol.Packet, 10)

	///< Goroutine 1: Reads packets and handles path responses immediately
	go func() {
		defer close(packetChan)

		for {   // Check if we should stop before attempting to read
			select {
			case <-p.stopChan:
				return
			case <-listenerCtx.Done():
				return
			default:   // Continue to read
			}

			if p.conn == nil {
				return
			}

			packet, err := p.protocol.ReadPacket(p.conn)
			if err != nil {
				select {
				case <-p.stopChan:
					// logger.Debug("Reader stopping due to stopChan after read error")
					return
				case <-listenerCtx.Done():
					logger.Debug("Reader stopping due to context after read error")
					return
				default:
				}
				if err == io.EOF {
					logger.Info("Connection closed by remote sfsmaster (EOF)")
					select {
					case done <- fmt.Errorf("connection closed by remote"):
					default:
					}
					return
				}
				// Check if connection was closed by us
				if strings.Contains(err.Error(), "use of closed network connection") {
					return
				}
				if isConnectionError(err) {
					select {
					case done <- fmt.Errorf("connection error: %w", err):
					default:
					}
					return
				}
				logger.Warnf("Error reading from sfsmaster: %v", err)
				continue
			}

			// Handle path responses immediately in the reader goroutine
			if packet.Type == protocol.MATONT_GET_PATH_TYPE_INODE {
				logger.Debugf("Rx packet for inode: %d", binary.BigEndian.Uint64(packet.Data[0:8]))
				if err := p.HandlePathTypeResponse(packet); err != nil {
					logger.Warnf("Error handling path type response: %v", err)
				}
			} else {   // Other packets go to channel for async processing
				select {
				case packetChan <- packet:
					// logger.Debugf("Forwarded packet type %d to handler", packet.Type)
				case <-p.stopChan:
					return
				case <-listenerCtx.Done():
					return
				}
			}
		}
	}()

	///< Goroutine 2: Handles non-path packets
	go func() {
		defer cancelListener()

		for {
			select {
			case <-p.stopChan:
				return
			case <-listenerCtx.Done():
				return
			case packet, ok := <-packetChan:
				if !ok {
					select {
					case done <- fmt.Errorf("packet reader exited"):
					default:
					}
					return
				}

				// Handle other packets
				shouldContinue, err := p.protocol.HandlePacket(packet, p)
				if err != nil {
					logger.Warnf("Error handling packet: %v", err)
				}
				if !shouldContinue {
					p.sfsConnected = false
					select {
					case done <- nil:
					default:
					}
					return
				}
			}
		}
	}()

	select {
	case <-p.stopChan:
		logger.Info("Stopping listening to sfsmaster due to processor stop")
		cancelListener()
		return nil
	case <-ctx.Done():
		logger.Info("Stopping listening to sfsmaster due to context cancellation")
		if p.conn != nil {
			p.conn.Close()
			p.conn = nil
		}
		return nil
	case err := <-done:
		logger.Debug("Stopped listening to sfsmaster")
		p.sfsConnected = false
		if p.conn != nil {
			p.conn.Close()
			p.conn = nil
		}
		return err
	}
}

// processPacketsFromMaster implements a resilient connection manager that maintains
// persistent communication with the SFS master server through automatic reconnection.
//
// Connection lifecycle:
//   - Establishes registration with master server
//   - Listens for packets and monitors connection health
//   - Automatically reconnects with 1-second interval on failures
//   - Responds to stopChan or context cancellation for graceful shutdown
//
// Reconnection strategy:
//   - Fixed 1-second reconnect interval
//   - Attempt counter for operational visibility
//   - Context-aware retry delays
//
// Returns nil on graceful shutdown via stopChan, or error on context cancellation.
func (p *Processor) processPacketsFromMaster(ctx context.Context) error {
	reconnectInterval := time.Second

	attempt := 0
	for {
		select {
		case <-p.stopChan:
			logger.Info("Stop Processor and exiting service")
			return nil
		case <-ctx.Done():
			return fmt.Errorf("context cancelled while processing packets from sfsmaster")
		default:
			if attempt > 0 {
				select {
				case <-time.After(reconnectInterval):
					logger.Infof("Reconnection attempt %d to sfsmaster", attempt)
				case <-p.stopChan:
					return nil
				case <-ctx.Done():
					return nil
				}
			}
			if !p.sfsConnected {
				if err := p.Register(ctx); err != nil {
					p.sfsConnected = false
					select {
					case <-p.stopChan:
						return nil
					default:
					}
					logger.Warnf("Failed to register with SFSMaster: %v", err)
					attempt++
					continue
				}
				attempt = 0
			}

			err := p.listenToSFSMaster(ctx)
			if err != nil {
				p.sfsConnected = false

				select {
				case <-p.stopChan:
					return nil
				default:
					attempt++
					continue
				}
			}

			select {
			case <-p.stopChan:
				return nil
			default:
				p.sfsConnected = false
				attempt++
			}
		}
	}
}

// Run orchestrates the complete processor lifecycle, initializing all subsystems
// and executing the processing mode based on configuration.
//
// Processing flow:
//   - Master mode: Real-time event streaming from SFS master server
//
// Resource guarantees (via defer):
//   - NATS client closed
//   - SFS connection closed
//   - Context canceled
//
// Returns error if processing fails (connection issues, context cancellation, etc.).
func (p *Processor) Run() error {
	defer p.natsClient.Close()
	defer p.Close()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Start monitoring NATS responses
	go p.natsClient.MonitorResponses(ctx, 30*time.Second)

	// Process inotify events received from SFSMaster
	return p.processPacketsFromMaster(ctx)
}

// verifyTLSConnection performs TLS handshake and validates the connection to the LeilFS master.
//
// Sets a 30-second deadline for the handshake, clears it afterward, and logs connection details
// (TLS version, cipher suite, server name) upon success.
//
// Parameters:
//   - tlsConn: The TLS connection to verify.
//
// Returns:
//   - error: nil on success, or an error if deadline setting fails or handshake times out/fails.
func (p *Processor) verifyTLSConnection(tlsConn *tls.Conn) error {
	if err := tlsConn.SetDeadline(time.Now().Add(30 * time.Second)); err != nil {
		return fmt.Errorf("failed to set TLS connection deadline: %w", err)
	}
	defer tlsConn.SetDeadline(time.Time{}) // Clear deadline after handshake

	if err := tlsConn.Handshake(); err != nil {
		if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
			return fmt.Errorf("TLS handshake timed out: %w", err)
		}
		return fmt.Errorf("TLS handshake failed: %w", err)
	}

	state := tlsConn.ConnectionState()
	serverName := state.ServerName
	if serverName == "" {
		serverName = p.config.SFSMaster.Host
	}

	logger.Infof("TLS connection established. Version: %s, Cipher: %s, ServerName: %s",
		tls.VersionName(state.Version),
		tls.CipherSuiteName(state.CipherSuite),
		serverName)

	return nil
}

// isConnectionError determines if the error indicates a lost or broken network connection.
//
// Returns true for EOF, common connection errors (refused/reset/broken pipe/unreachable/timeout),
// or network read/write operation errors.
func isConnectionError(err error) bool {
	if err == io.EOF {
		return true
	}

	errMsg := strings.ToLower(err.Error())
	connectionErrors := []string{
		"connection refused",
		"connection reset",
		"broken pipe",
		"network is down",
		"network unreachable",
		"io timeout",
		"no route to host",
	}

	for _, msg := range connectionErrors {
		if strings.Contains(errMsg, msg) {
			return true
		}
	}

	if opErr, ok := err.(*net.OpError); ok {
		if opErr.Op == "read" || opErr.Op == "write" {
			return true
		}
	}

	return false
}

// isConnectionRefused checks if the error indicates a "connection refused" network failure.
//
// Returns true if the error is ECONNREFUSED or contains "connection refused" in its message.
func isConnectionRefused(err error) bool {
	if opErr, ok := err.(*net.OpError); ok {
		if syscallErr, ok := opErr.Err.(*os.SyscallError); ok {
			if errno, ok := syscallErr.Err.(syscall.Errno); ok {
				return errno == syscall.ECONNREFUSED
			}
		}
		return strings.Contains(err.Error(), "connection refused")
	}
	return strings.Contains(err.Error(), "connection refused")
}

// main is the application entry point for the LeilFS inotifier system.
//
// Lifecycle phases:
//   - Initialization: Parse CLI, load config, setup logger
//   - Execution: Run main processing loop
//   - Shutdown: Handle signals, graceful termination, cleanup
//
// Configuration priority (highest to lowest):
//   1. Command-line flags  2. Config file  3. Defaults
//
// Key CLI flags:
//   -c, --config           Config file path
//   --log-level            debug/info/warn/error
//   -H, --host             sfsmaster host
//   -P, --port             sfsmaster port
//   --nats-url             NATS server URL
//   --nats-subject         NATS subject
//
// Signal handling:
//   - SIGINT/SIGTERM trigger graceful shutdown
//
// Exit codes:
//   - 0: Success   - 1: Initialization or runtime error
func main() {
	var configFile string
	var logLevel string
	var sfsmasterHost string
	var sfsmasterPort string
	var logFile string
	var filterRegisteredOps bool
	var useTLS bool

	pflag.StringVarP(&configFile, "config", "c", "", "Config file path (default: searches in /etc/leilfs/inotifier, ./, ./config/)")
	pflag.String("nats-url", "nats://localhost:4222", "NATS server URL")
	pflag.String("nats-subject", "leilfs.events", "NATS subject")
	pflag.StringVar(&logLevel, "log-level", "info", "Log level (debug, info, warn, error)")
	pflag.StringVar(&logFile, "log-file", "/var/log/leilfs/inotifier.log", "Log file path (empty for stdout)")

	pflag.StringVarP(&sfsmasterHost, "host", "H", "sfsmaster", "sfsmaster hostname or IP address")
	pflag.StringVarP(&sfsmasterPort, "port", "P", "9418", "sfsmaster port")

	pflag.String("filter-registered-ops", "false", "Only reports to NATs registered operations: true/false, default: 'false'")
	pflag.String("use-tls", "false", "Use TLS for connection to sfsmaster: true/false, default: 'false'")

	pflag.Parse()

	loggerConfig := logger.LoggerConfig{
		LogPath: logFile,
		MaxEntries: 10000, //maxLogEntries,
		MaxBackups: 3, //maxLogBackups,
	}

	if err := logger.InitializeLogger(loggerConfig); err != nil {
		fmt.Fprintf(os.Stderr, "Failed to initialize logger: %v\n", err)
		os.Exit(1)
	}

	logInstance := logger.GetLogger()
	if closer, ok := logInstance.(interface{ Close() error }); ok {
		defer closer.Close()
	}

	level := logger.LevelFromString(logLevel)
	logger.SetLevel(level)

	cfg, err := config.LoadConfig(configFile)
	if err != nil {
		logger.Errorf("Failed to load config: %v", err)
		os.Exit(1)
	}
	if pflag.Lookup("nats-subject").Changed {
		cfg.NATSSubject = pflag.Lookup("nats-subject").Value.String()
	}
	if NATSURL := pflag.Lookup("nats-url").Value.String(); NATSURL != "" {
		cfg.NATSURL = NATSURL
	}
	if sfsmasterHost != "sfsmaster" {
		cfg.SFSMaster.Host = sfsmasterHost
	}
	if sfsmasterPort != "9418" {
		cfg.SFSMaster.Port = sfsmasterPort
	}
	if pflag.Lookup("filter-registered-ops").Changed {
		filterOpsStr := pflag.Lookup("filter-registered-ops").Value.String()
		filterRegisteredOps, err = strconv.ParseBool(filterOpsStr)
		if err != nil {
			logger.Warnf("Invalid value for filter-registered-ops: %v", err)
			filterRegisteredOps = false
		}
		cfg.FilterRegisteredOps = filterRegisteredOps
	}
	if pflag.Lookup("use-tls").Changed {
		useTLSStr := pflag.Lookup("use-tls").Value.String()
		useTLS, err = strconv.ParseBool(useTLSStr)
		if err != nil {
			logger.Warnf("Invalid value for use-tls: %v", err)
			useTLS = false
		}
		cfg.UseTLS = useTLS
	}

	logger.Debug("leilfs-inotifier start successfully")
	logger.Debugf("Using config with version: %s", config.GetVersionStr())

	processor, err := NewProcessor(cfg)
	if err != nil {
		logger.Errorf("Failed to create processor: %v", err)
		os.Exit(1)
	}

	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	done := make(chan error, 1)
	go func() {
		done <- processor.Run()
	}()

	select {
		case err := <-done:
			if err != nil {
				logger.Errorf("Processor exited with error: %v", err)
				os.Exit(1)
			}
			logger.Info("Processor exited successfully")
		case sig := <-sigChan:
			logger.Infof("Received signal: %v, shutting down...", sig)
			processor.Exit()
			if err := <-done; err != nil {
				logger.Errorf("Processor exited with error: %v", err)
			}
		logger.Info("leilfs-inotifier shutdown gracefully")
	}
}