package multilangclient

import (
	"bufio"
	"encoding/json"
	"fmt"
	"io"
	"os"
)

// A client handles communication between a Consumer and the KCL
// MultiLangDaemon.
type client struct {
	in  *bufio.Reader
	out io.Writer

	consumer Consumer
}

func newClient(c Consumer) *client {
	return &client{
		in:       bufio.NewReader(os.Stdin),
		out:      os.Stdout,
		consumer: c,
	}
}

// run is the main loop of the client, retrieving messages and
// handling them. It will block as long as it doesn't hit errors. If
// it does hit an error, it will stop and return it. The recommended
// thing to do in the face of an error is call os.Exit() and let the
// KCL MultiLangDaemon restart the process.
func (c *client) run() error {
	for {
		msg, err := c.recv()
		if err != nil {
			return err
		}

		switch msg.Action {
		case initialize:
			err = c.handleInitialize(msg)
		case processRecords:
			err = c.handleProcessRecords(msg)
		case shutdown:
			// immediately acknowledge the shutdown because
			// `c.handleShutdown` has a high chance of causing the
			// process to exit, which would confuse the daemon because
			// its shutdown request never gets acked. If we fail to
			// even ack, then we're *really* hosed - we should still
			// try to shut down, though.
			_ = c.send(ack(msg))
			return c.handleShutdown(msg)
		default:
			err = fmt.Errorf("received unexpected message %s", msg.Action)
		}

		if err != nil {
			return err
		}

		err = c.send(ack(msg))
		if err != nil {
			return err
		}
	}
}

// recv retrieves a message from the client's input. Blocks until a
// message is available.
func (c *client) recv() (*message, error) {
	data, err := c.readLine()
	if err != nil {
		return nil, err
	}
	msg, err := c.parse(data)

	return msg, err
}

// readLine receives a single line from the client's input
func (c *client) readLine() ([]byte, error) {
	line, err := c.in.ReadBytes('\n')
	if err != nil {
		return nil, err
	}

	// line includes the delimiting \n rune - trim it off
	return line[:len(line)-1], nil
}

// parse interperets a byte slice as a KCL message.
func (c *client) parse(data []byte) (*message, error) {
	msg := &message{}
	err := json.Unmarshal(data, msg)
	if err != nil {
		return nil, err
	}
	return msg, nil
}

// send sends a KCL message to the client's output.
func (c *client) send(msg *message) error {
	bytes, err := json.Marshal(msg)
	if err != nil {
		return err
	}
	// messages always end with a newline
	bytes = append(bytes, '\n')

	_, err = c.out.Write(bytes)
	if err != nil {
		return err
	}

	return nil
}

// handlers for each inbound message type below.

func (c *client) handleInitialize(msg *message) error {
	logger.Printf("recv initialize shard=%s", msg.ShardID)
	return c.consumer.Initialize(msg.ShardID)
}

func (c *client) handleProcessRecords(msg *message) error {
	logger.Printf("recv records count=%d", len(msg.Records))

	var lastSuccess string
	defer func() {
		if lastSuccess != "" {
			c.checkpoint(lastSuccess)
		}
	}()

	for _, rec := range msg.Records {
		data, err := rec.decode()
		if err != nil {
			return err
		}
		err = c.consumer.Process(data)
		if err != nil {
			return err
		}
		lastSuccess = rec.SequenceNumber
	}
	return nil
}

func (c *client) handleShutdown(msg *message) error {
	logger.Printf("recv shutdown reason=%q", msg.Reason)
	return c.consumer.Shutdown()
}

// Mark a record as processed.
func (c *client) checkpoint(sequenceNumber string) error {
	msg := &message{Action: checkpoint, Checkpoint: sequenceNumber}
	if err := c.send(msg); err != nil {
		return err
	}

	resp, err := c.recv()
	if err != nil {
		return err
	}

	if resp.Error != "" {
		return fmt.Errorf("KCL MultiLangDaemon responded to checkpoint %s with err=%q", sequenceNumber, resp.Error)
	}

	return nil
}
