package main

import (
	"errors"
	"fmt"
	"log"
	"net"
	"strconv"
	"sync"
	"time"

	"github.com/flynn/flynn/discoverd/cache"
	"github.com/flynn/flynn/pkg/connutil"
	"github.com/flynn/flynn/router/proxy"
	router "github.com/flynn/flynn/router/types"
	"golang.org/x/net/context"
)

type TCPListener struct {
	Watcher

	IP string

	discoverd DiscoverdClient
	syncer    *Syncer
	wm        *WatchManager
	stopSync  func()

	startPort     int
	endPort       int
	reservedPorts []int
	listeners     map[int]net.Listener

	mtx      sync.RWMutex
	services map[string]*service
	routes   map[string]*tcpRoute
	ports    map[int]*tcpRoute
	closed   bool
}

func (l *TCPListener) Start() error {
	ctx := context.Background() // TODO(benburkert): make this an argument
	ctx, l.stopSync = context.WithCancel(ctx)

	if l.Watcher != nil {
		return errors.New("router: tcp listener already started")
	}
	if l.wm == nil {
		l.wm = NewWatchManager()
	}
	l.Watcher = l.wm

	if l.syncer == nil {
		return errors.New("router: tcp listener missing syncer")
	}

	l.services = make(map[string]*service)
	l.routes = make(map[string]*tcpRoute)
	l.ports = make(map[int]*tcpRoute)
	l.listeners = make(map[int]net.Listener)

	if l.startPort != 0 && l.endPort != 0 {
		for i := l.startPort; i <= l.endPort; i++ {
			addr := fmt.Sprintf("%s:%d", l.IP, i)
			listener, err := listenFunc("tcp4", addr)
			if err != nil {
				l.Close()
				return listenErr{addr, err}
			}
			l.listeners[i] = listener
		}
	}

	// TODO(benburkert): the sync API cannot handle routes deleted while the
	// listen/notify connection is disconnected
	if err := l.startSync(ctx); err != nil {
		l.Close()
		return err
	}

	return nil
}

func (l *TCPListener) startSync(ctx context.Context) error {
	errc := make(chan error)
	startc := l.doSync(ctx, errc)

	select {
	case err := <-errc:
		return err
	case <-startc:
		go l.runSync(ctx, errc)
		return nil
	}
}

func (l *TCPListener) runSync(ctx context.Context, errc chan error) {
	err := <-errc

	for {
		if err == nil {
			return
		}
		log.Printf("router: tcp sync error: %s", err)

		time.Sleep(2 * time.Second)

		l.doSync(ctx, errc)

		err = <-errc
	}
}

func (l *TCPListener) doSync(ctx context.Context, errc chan<- error) <-chan struct{} {
	startc := make(chan struct{})

	go func() { errc <- l.syncer.Sync(ctx, &tcpSyncHandler{l: l}, startc) }()

	return startc
}

func (l *TCPListener) Close() error {
	l.mtx.Lock()
	defer l.mtx.Unlock()
	if l.closed {
		return nil
	}
	l.stopSync()
	for _, s := range l.routes {
		s.Close()
	}
	for _, listener := range l.listeners {
		listener.Close()
	}
	l.closed = true
	return nil
}

type tcpSyncHandler struct {
	l *TCPListener
}

func (h *tcpSyncHandler) Current() map[string]struct{} {
	h.l.mtx.RLock()
	defer h.l.mtx.RUnlock()
	ids := make(map[string]struct{}, len(h.l.routes))
	for id := range h.l.routes {
		ids[id] = struct{}{}
	}
	return ids
}

func (h *tcpSyncHandler) Set(data *router.Route) error {
	route := data.TCPRoute()
	r := &tcpRoute{
		TCPRoute: route,
		addr:     h.l.IP + ":" + strconv.Itoa(route.Port),
		parent:   h.l,
	}

	h.l.mtx.Lock()
	defer h.l.mtx.Unlock()
	if h.l.closed {
		return nil
	}

	service := h.l.services[r.Service]
	if service != nil && service.name != r.Service {
		service.refs--
		if service.refs <= 0 {
			service.Close()
			delete(h.l.services, service.name)
		}
		service = nil
	}
	if service == nil {
		sc, err := cache.New(h.l.discoverd.Service(r.Service))
		if err != nil {
			return err
		}

		service = newService(r.Service, sc, h.l.wm, r.DrainBackends)
		h.l.services[r.Service] = service
	}
	r.service = service
	var bf proxy.BackendListFunc
	if r.Leader {
		bf = backendFunc(r.Service, service.sc.Leader)
	} else {
		bf = backendFunc(r.Service, service.sc.Instances)
	}
	r.rp = proxy.NewReverseProxy(proxy.ReverseProxyConfig{
		BackendListFunc: bf,
		RequestTracker:  service,
		Logger:          logger,
	})
	if listener, ok := h.l.listeners[r.Port]; ok {
		r.l = listener
		delete(h.l.listeners, r.Port)
	}
	started := make(chan error)
	go r.Serve(started)
	if err := <-started; err != nil {
		if r.l != nil {
			h.l.listeners[r.Port] = r.l
		}
		return err
	}
	service.refs++
	h.l.routes[data.ID] = r
	h.l.ports[r.Port] = r

	go h.l.wm.Send(&router.Event{Event: router.EventTypeRouteSet, ID: data.ID, Route: r.ToRoute()})
	return nil
}

func (h *tcpSyncHandler) Remove(id string) error {
	h.l.mtx.Lock()
	defer h.l.mtx.Unlock()
	if h.l.closed {
		return nil
	}
	r, ok := h.l.routes[id]
	if !ok {
		return ErrNotFound
	}
	r.Close()

	r.service.refs--
	if r.service.refs <= 0 {
		r.service.sc.Close()
		delete(h.l.services, r.service.name)
	}

	delete(h.l.routes, id)
	delete(h.l.ports, r.Port)
	go h.l.wm.Send(&router.Event{Event: router.EventTypeRouteRemove, ID: id, Route: r.ToRoute()})
	return nil
}

type tcpRoute struct {
	parent *TCPListener
	*router.TCPRoute
	l       net.Listener
	addr    string
	service *service
	rp      *proxy.ReverseProxy
}

func (r *tcpRoute) Serve(started chan<- error) {
	var err error
	// TODO: close the listener while there are no backends available
	if r.l == nil {
		r.l, err = listenFunc("tcp4", r.addr)
	}
	if err != nil {
		err = listenErr{r.addr, err}
	}
	started <- err
	if err != nil {
		return
	}
	for {
		conn, err := r.l.Accept()
		if err != nil {
			break
		}
		go r.ServeConn(conn)
	}
}

func (r *tcpRoute) Close() {
	if r.Port >= r.parent.startPort && r.Port <= r.parent.endPort {
		// make a copy of the fd and create a new listener with it
		fd, err := r.l.(*net.TCPListener).File()
		if err != nil {
			log.Println("Error getting listener fd", r.l)
			return
		}
		r.parent.listeners[r.Port], err = net.FileListener(fd)
		if err != nil {
			log.Println("Error copying listener", r.l)
			return
		}
		fd.Close()
	}
	r.l.Close()
}

func (r *tcpRoute) ServeConn(conn net.Conn) {
	r.rp.ServeConn(context.Background(), connutil.CloseNotifyConn(conn))
}
