package io

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

	"git.smesh.lol/torly/pkg/unix"

	"git.smesh.lol/torly/pkg/wire"
)

// Acceptor owns the listening socket. It polls the socket with a bounded
// timeout so the shut-down breaker is observed promptly, and hands accepted
// sockets to reader workers in round-robin order.
func Acceptor(data AcceptorData, ch AcceptorChans) {
	fd, bound, err := listenTCP(data.Addr)
	if err != nil {
		ch.Ready <- wire.Ready{Err: err.Error()}
		return
	}
	ch.Ready <- wire.Ready{OK: true, Addr: bound}
	defer unix.Close(fd)

	if len(ch.Workers) == 0 {
		return
	}
	next := 0
	var id wire.ConnID
	for {
		if ch.Stop.Closed() {
			return
		}
		pfd := []unix.PollFd{{Fd: int32(fd), Events: int16(unix.POLLIN)}}
		n, err := unix.Poll(pfd, 200)
		if err != nil {
			if errors.Is(err, unix.EINTR) {
				continue
			}
			// A transient poll error must not kill the only acceptor.
			time.Sleep(10 * time.Millisecond)
			continue
		}
		if n == 0 {
			continue
		}
		nfd, sa, err := unix.Accept4(fd, unix.SOCK_NONBLOCK|unix.SOCK_CLOEXEC)
		if err != nil {
			if errors.Is(err, unix.EAGAIN) || errors.Is(err, unix.EINTR) {
				continue
			}
			// ECONNABORTED, EMFILE and similar errors are transient under
			// connection churn. Exiting here left the backlog permanently full
			// and made the relay look hung.
			time.Sleep(10 * time.Millisecond)
			continue
		}
		id++
		msg := wire.Assign{ID: id, Socket: &wire.Socket{Conn: newFDConn(nfd, sockaddrString(sa))}}
		sent := false
		for i := 0; i < len(ch.Workers); i++ {
			idx := (next + i) % len(ch.Workers)
			if ch.Workers[idx].TrySend(msg) {
				next = idx + 1
				sent = true
				break
			}
		}
		if !sent {
			idx := next % len(ch.Workers)
			select {
			case ch.Workers[idx].Tx <- msg:
				next = idx + 1
			case <-ch.Stop.Done():
				msg.Socket.Conn.Close()
				return
			}
		}
	}
}

func listenTCP(addr string) (int, string, error) {
	host, portStr, err := net.SplitHostPort(addr)
	if err != nil {
		return -1, "", err
	}
	port, err := strconv.Atoi(portStr)
	if err != nil || port < 0 || port > 65535 {
		return -1, "", fmt.Errorf("bad listen port %q", portStr)
	}
	var ip [4]byte
	if host != "" && host != "0.0.0.0" {
		parsed := net.ParseIP(host)
		if parsed == nil || parsed.To4() == nil {
			return -1, "", fmt.Errorf("unsupported listen address %q", host)
		}
		copy(ip[:], parsed.To4())
	}
	fd, err := unix.Socket(unix.AF_INET, unix.SOCK_STREAM|unix.SOCK_NONBLOCK, 0)
	if err != nil {
		return -1, "", err
	}
	if err := unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_REUSEADDR, 1); err != nil {
		unix.Close(fd)
		return -1, "", err
	}
	sa := &unix.SockaddrInet4{Port: port, Addr: ip}
	if err := unix.Bind(fd, sa); err != nil {
		unix.Close(fd)
		return -1, "", err
	}
	if err := unix.Listen(fd, 1024); err != nil {
		unix.Close(fd)
		return -1, "", err
	}
	name, err := unix.Getsockname(fd)
	if err != nil {
		unix.Close(fd)
		return -1, "", err
	}
	boundPort := port
	if got, ok := name.(*unix.SockaddrInet4); ok {
		boundPort = got.Port
	}
	return fd, fmt.Sprintf("%s:%d", hostOrAny(host), boundPort), nil
}

func hostOrAny(host string) string {
	if host == "" || host == "0.0.0.0" {
		return "0.0.0.0"
	}
	return host
}

func sockaddrString(sa unix.Sockaddr) string {
	switch a := sa.(type) {
	case *unix.SockaddrInet4:
		ip := net.IP(a.Addr[:]).String()
		return ip + ":" + strconv.Itoa(a.Port)
	default:
		return "unknown"
	}
}