package main

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"os"
	"path/filepath"
	"sync"
	"time"
)

const (
	// agentDirName is the directory of the root agent in the data directory.
	agentDirName = "agent"

	// socketFileName is the Unix socket of the daemon in the data directory.
	socketFileName = "onechat.sock"

	// tokenLockFileName is the lock file of the token file in the data
	// directory.
	tokenLockFileName = "auth.lock"

	// requestBytes is the most bytes of one request line of a client.
	requestBytes = 16 << 20

	// clientEvents is the number of events that can wait for a slow client
	// before the daemon drops it.
	clientEvents = 1024

	// socketPathBytes is the longest path of a Unix socket on Linux.
	socketPathBytes = 107

	// terminalSource is the source of the user messages from the clients of
	// the socket.
	terminalSource = "terminal"
)

// errTokensInUse tells that another onechat process uses the token file.
var errTokensInUse = errors.New("another onechat process uses the token file, such as the daemon")

// clientRequest is one line from a client: "send" with the text of a user
// message, or "interrupt".
type clientRequest struct {
	Type string `json:"type"`
	Text string `json:"text,omitempty"`
}

// daemon runs "onechat daemon": it owns the root agent and its files and serves
// the clients on the Unix socket in the data directory until ctx ends. When
// telegram.json exists, it also runs the Telegram bot. It holds the lock of the
// token file, so it is the only process that refreshes the tokens.
func daemon(ctx context.Context, out io.Writer) (err error) {
	dir, err := dataDir()
	if err != nil {
		return err
	}

	// Check the socket path before the daemon writes a file, so that a data
	// directory with a long path stays empty.
	socket := filepath.Join(dir, socketFileName)
	err = checkSocketPath(socket)
	if err != nil {
		return err
	}

	config, err := loadConfig(dir)
	if err != nil {
		return err
	}

	botSettings, hasBot, err := loadBotConfig(dir)
	if err != nil {
		return err
	}

	tokens, err := lockTokens(dir)
	if err != nil {
		return err
	}

	defer func() {
		err = errors.Join(err, tokens.Close())
	}()

	address, auth, err := modelEndpoint(config, dir)
	if err != nil {
		return err
	}

	agent, err := openAgent(filepath.Join(dir, agentDirName))
	if err != nil {
		return err
	}

	client := newClient(address, auth, agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl"))
	memory, err := openMemory(ctx, agent.dir, time.Now, newCompactor(client, agent.Prompt, config.Compactor))
	if err != nil {
		return err
	}

	defer func() {
		err = errors.Join(err, memory.Close())
	}()

	listener, err := listenSocket(socket)
	if err != nil {
		return err
	}

	_, err = fmt.Fprintf(out, "onechat: %s on %s, %s at %s effort, in %s\n", agent.config.Name, listener.Addr(), agent.config.Model, agent.config.Effort, agent.config.Workdir)
	if err != nil {
		return errors.Join(err, listener.Close())
	}

	var telegram *telegramBot
	if hasBot {
		telegram = newTelegramBot(newBotAPI(telegramAPI, botSettings.Token), botSettings.Owner, filepath.Join(agent.dir, offsetFileName), log.New(out, "telegram: ", log.LstdFlags))
		_, err = fmt.Fprintln(out, "onechat: the Telegram bot waits for messages of its owner")
		if err != nil {
			return errors.Join(err, listener.Close())
		}
	}

	return serveAgent(ctx, listener, telegram, func(notify func(Event)) (*loop, error) {
		return newLoop(agent, memory, client, time.Now, notify)
	})
}

// serveAgent runs the loop that start makes and serves the clients of
// listener until ctx ends or the loop fails. It closes listener. When bot is
// not nil, it runs the bot too, and a failure of the bot stops the agent.
func serveAgent(ctx context.Context, listener net.Listener, bot *telegramBot, start func(notify func(Event)) (*loop, error)) error {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	hub := &hub{
		clients: make(map[*hubClient]bool),
	}

	notify := hub.Broadcast
	if bot != nil {
		notify = func(event Event) {
			hub.Broadcast(event)
			bot.Notify(event)
		}
	}

	l, err := start(notify)
	if err != nil {
		return errors.Join(err, listener.Close())
	}

	var workers sync.WaitGroup
	workers.Add(1)
	go func() {
		defer workers.Done()

		hub.Accept(ctx, listener, l)
	}()

	var botErr error
	if bot != nil {
		workers.Add(1)
		go func() {
			defer workers.Done()

			botErr = bot.Run(ctx, l)
			cancel()
		}()
	}

	err = l.Run(ctx)
	if errors.Is(err, context.Canceled) {
		err = nil
	}

	cancel()
	err = errors.Join(err, listener.Close())
	hub.CloseAll()
	workers.Wait()

	return errors.Join(err, botErr)
}

// hub holds the connected clients and sends each event to all of them.
type hub struct {
	mutex   sync.Mutex
	clients map[*hubClient]bool
	workers sync.WaitGroup
}

// hubClient is one connected client. Its events wait in a channel for the
// writer, so a slow client does not stop the agent.
type hubClient struct {
	conn   net.Conn
	events chan Event
	once   sync.Once
}

// Accept serves each client of listener until the listener closes. A new
// client gets a ready event with the name of the agent, and a busy event when a
// turn runs.
func (h *hub) Accept(ctx context.Context, listener net.Listener, l *loop) {
	for {
		conn, err := listener.Accept()
		if err != nil {
			return
		}

		client := &hubClient{
			conn:   conn,
			events: make(chan Event, clientEvents),
		}

		h.mutex.Lock()
		h.clients[client] = true
		h.mutex.Unlock()

		h.workers.Add(2)
		go h.write(client)
		go h.read(ctx, client, l)

		h.Send(client, Event{
			Kind: "ready",
			Text: l.agent.config.Name,
		})

		if l.Busy() {
			h.Send(client, Event{
				Kind: "busy",
			})
		}
	}
}

// write sends the events of client as JSON lines until its channel closes.
func (h *hub) write(client *hubClient) {
	defer h.workers.Done()

	encoder := json.NewEncoder(client.conn)
	for event := range client.events {
		err := encoder.Encode(event)
		if err != nil {
			h.drop(client)
		}
	}
}

// read runs the requests of client until it closes its connection.
func (h *hub) read(ctx context.Context, client *hubClient, l *loop) {
	defer h.workers.Done()
	defer h.drop(client)

	scanner := bufio.NewScanner(client.conn)
	scanner.Buffer(make([]byte, 64*1024), requestBytes)
	for ctx.Err() == nil && scanner.Scan() {
		var request clientRequest
		err := json.Unmarshal(scanner.Bytes(), &request)
		if err == nil {
			err = runRequest(l, request)
		}

		if err != nil {
			h.Send(client, Event{
				Kind: "error",
				Text: err.Error(),
			})
		}
	}
}

// Send gives event to one client, or drops the client when its events do not
// fit.
func (h *hub) Send(client *hubClient, event Event) {
	h.mutex.Lock()
	defer h.mutex.Unlock()

	h.send(client, event)
}

// send gives event to client. The caller holds the mutex.
func (h *hub) send(client *hubClient, event Event) {
	if !h.clients[client] {
		return
	}

	select {
	case client.events <- event:
	default:
		h.remove(client)
	}
}

// Broadcast gives event to every client.
func (h *hub) Broadcast(event Event) {
	h.mutex.Lock()
	defer h.mutex.Unlock()

	for client := range h.clients {
		h.send(client, event)
	}
}

// drop removes client and closes its connection.
func (h *hub) drop(client *hubClient) {
	h.mutex.Lock()
	defer h.mutex.Unlock()

	h.remove(client)
}

// remove removes client, closes its connection, and ends its writer. The
// caller holds the mutex.
func (h *hub) remove(client *hubClient) {
	delete(h.clients, client)
	client.once.Do(func() {
		close(client.events)
		_ = client.conn.Close()
	})
}

// CloseAll drops every client and waits for their readers and writers.
func (h *hub) CloseAll() {
	h.mutex.Lock()
	for client := range h.clients {
		h.remove(client)
	}
	h.mutex.Unlock()

	h.workers.Wait()
}

// runRequest runs one request of a client.
func runRequest(l *loop, request clientRequest) error {
	switch request.Type {
	case "send":
		if request.Text == "" {
			return errors.New("empty message")
		}

		_, err := l.Submit(terminalSource, request.Text)

		return err
	case "interrupt":
		if !l.Interrupt() {
			return errors.New("no turn runs")
		}

		return nil
	default:
		return fmt.Errorf("unknown request %q", request.Type)
	}
}

// listenSocket listens on the Unix socket at path, with mode 0600. The caller
// owns the chat, so a socket file at path is left from a daemon that ended,
// and listenSocket removes it.
func listenSocket(path string) (net.Listener, error) {
	err := checkSocketPath(path)
	if err != nil {
		return nil, err
	}

	err = os.Remove(path)
	if err != nil && !errors.Is(err, os.ErrNotExist) {
		return nil, fmt.Errorf("remove old socket: %w", err)
	}

	listener, err := net.Listen("unix", path)
	if err != nil {
		return nil, fmt.Errorf("listen on socket: %w", err)
	}

	err = os.Chmod(path, 0o600)
	if err != nil {
		return nil, errors.Join(fmt.Errorf("set socket mode: %w", err), listener.Close())
	}

	return listener, nil
}

// checkSocketPath makes sure that path fits in the address of a Unix socket.
func checkSocketPath(path string) error {
	if len(path) > socketPathBytes {
		return fmt.Errorf("socket path %s is longer than %d bytes: set ONECHAT_DIR to a shorter path", path, socketPathBytes)
	}

	return nil
}

// lockTokens takes the lock of the token file in dir. Each process that reads
// or refreshes the tokens holds it, because a refresh in one process rotates
// the refresh token of the other and logs it out.
func lockTokens(dir string) (*os.File, error) {
	lock, err := lockFile(filepath.Join(dir, tokenLockFileName))
	if errors.Is(err, errLocked) {
		return nil, errTokensInUse
	}

	return lock, err
}