package filemonitor

import (
	"bufio"
	"io"
	"net"
	"sync"
	"time"

	slog "github.com/burke/zeus/go/shinylog"
)

type fileListener struct {
	gatheringMonitor
	netListener net.Listener
	connections map[net.Conn]chan string
	stop        chan struct{}
	sync.Mutex
	wg sync.WaitGroup
}

func NewFileListener(fileChangeDelay time.Duration, ln net.Listener) FileMonitor {
	fl := fileListener{
		netListener: ln,
		connections: make(map[net.Conn]chan string),
		stop:        make(chan struct{}),
	}
	fl.fileChangeDelay = fileChangeDelay
	fl.changes = make(chan string)

	go fl.serveListeners()
	go fl.serve()

	return &fl
}

func (f *fileListener) Add(file string) error {
	f.Lock()
	defer f.Unlock()

	for _, ch := range f.connections {
		ch <- file
	}

	return nil
}

func (f *fileListener) Close() error {
	f.Lock()

	select {
	case <-f.stop:
		f.Unlock()
		return nil // Already stopped
	default:
		close(f.stop)
	}

	var firstErr error
	if firstErr = f.netListener.Close(); firstErr != nil {
		slog.Trace("Error closing file listener: %v", firstErr)
	}

	for conn := range f.connections {
		if err := conn.Close(); err != nil {
			if firstErr == nil {
				firstErr = err
			}
			slog.Trace("Error closing connection: %v", err)
		}
	}

	f.Unlock()
	f.wg.Wait()
	close(f.changes)

	return firstErr
}

func (f *fileListener) serve() {
	var tempDelay time.Duration // how long to sleep on accept failure

	for {
		conn, err := f.netListener.Accept()
		if err != nil {
			if ne, ok := err.(net.Error); ok && ne.Temporary() {
				if tempDelay == 0 {
					tempDelay = 5 * time.Millisecond
				} else {
					tempDelay *= 2
				}
				if max := 1 * time.Second; tempDelay > max {
					tempDelay = max
				}
				slog.Trace("filelistener: Accept error: %v; retrying in %v", err, tempDelay)
				time.Sleep(tempDelay)
				continue
			}

			select {
			case <-f.stop:
				return
			default:
				panic(err)
			}
		}

		ch := make(chan string)
		f.Lock()
		f.connections[conn] = ch
		f.wg.Add(1)
		f.Unlock()

		go f.handleConnection(conn, ch)
	}
}

func (f *fileListener) handleConnection(conn net.Conn, ch chan string) {
	// Handle writes
	stop := make(chan struct{})
	go func() {
		for {
			select {
			case s := <-ch:
				conn.SetWriteDeadline(time.Now().Add(1 * time.Second))
				if _, err := conn.Write([]byte(s + "\n")); err == io.EOF {
					return
				} else if err != nil {
					slog.Trace("Error writing to connection: %v", err)
				}
			case <-stop:
				return
			}
		}
	}()

	// Handle reads
	scanner := bufio.NewScanner(conn)
	for {
		if scanner.Scan() {
			f.changes <- scanner.Text()
		} else {
			if err := scanner.Err(); err != nil {
				select {
				case <-f.stop:
					break
				default:
					slog.Trace("Error reading from connection: %v", err)
				}
			}
			break
		}
	}

	f.Lock()
	defer f.Unlock()

	close(stop)
	delete(f.connections, conn)
	f.wg.Done()
}
