package registry

import (
	"context"
	"database/sql"
	"encoding/gob"
	"errors"
	"fmt"
	"io"
	"os"
	"sync"
	"time"
)

var errRemoteMethod = errors.New("registry: unsupported remote operation")

var remoteErrors = []error{os.ErrNotExist, os.ErrPermission, sql.ErrNoRows, ErrClaimOwnership, ErrTimerChanged, ErrTimerKind, ErrPathsChanged, ErrFleetBoxActive, context.Canceled, context.DeadlineExceeded, ErrGoodBuildChanged}

func responseError(out remoteResponse) error {
	if !out.Failed {
		return nil
	}
	if out.ErrorKind > 0 && out.ErrorKind <= len(remoteErrors) {
		sentinel := remoteErrors[out.ErrorKind-1]
		if out.Error == sentinel.Error() {
			return sentinel
		}
		return fmt.Errorf("%s: %w", out.Error, sentinel)
	}
	return errors.New(out.Error)
}

func setResponseError(out *remoteResponse, err error) {
	if err == nil {
		return
	}
	out.Failed = true
	out.Error = err.Error()
	for i, sentinel := range remoteErrors {
		if errors.Is(err, sentinel) {
			out.ErrorKind = i + 1
			break
		}
	}
}

type remoteClient struct {
	mu      sync.Mutex
	enc     *gob.Encoder
	dec     *gob.Decoder
	input   io.WriteCloser
	output  io.ReadCloser
	wait    func() error
	closed  bool
	connect func() (io.WriteCloser, io.ReadCloser, func() error, error)
}

// OpenRemote connects to an already privilege-dropped registry worker.
func OpenRemote(path string, input io.WriteCloser, output io.ReadCloser, wait func() error) (*Registry, error) {
	c := &remoteClient{}
	if err := c.attach(input, output, wait); err != nil {
		return nil, err
	}
	return &Registry{path: path, remote: c}, nil
}

// OpenRemoteReconnect reads service-controlled answers, not an authority boundary.
// Failed or cancelled mutations are never retried; the next call reconnects.
// Canceled does not mean not applied: cancellation can race a committed mutation.
func OpenRemoteReconnect(path string, connect func() (io.WriteCloser, io.ReadCloser, func() error, error)) (*Registry, error) {
	c := &remoteClient{connect: connect}
	if err := c.reconnect(context.Background()); err != nil {
		return nil, err
	}
	return &Registry{path: path, remote: c}, nil
}

func (c *remoteClient) reconnect(ctx context.Context) (err error) {
	input, output, wait, err := c.connect()
	if err != nil {
		return err
	}
	finished := make(chan struct{})
	stop := context.AfterFunc(ctx, func() { input.Close(); output.Close(); close(finished) })
	defer func() {
		if !stop() {
			<-finished
		}
		if ctx.Err() != nil {
			c.disconnect()
			err = ctx.Err()
		}
	}()
	return c.attach(input, output, wait)
}

func (c *remoteClient) attach(input io.WriteCloser, output io.ReadCloser, wait func() error) error {
	c.input, c.output, c.wait = input, output, wait
	c.enc, c.dec = gob.NewEncoder(input), gob.NewDecoder(output)
	var ready remoteResponse
	err := c.dec.Decode(&ready)
	if err == nil {
		err = responseError(ready)
	}
	if err != nil {
		c.disconnect()
	}
	return err
}

func (c *remoteClient) disconnect() error {
	if c.input == nil {
		return nil
	}
	c.input.Close()
	c.output.Close()
	c.input, c.output = nil, nil
	if c.wait != nil {
		return c.wait()
	}
	return nil
}

func (c *remoteClient) stop() error {
	if c.closed {
		return nil
	}
	c.closed = true
	return c.disconnect()
}

func (c *remoteClient) close() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.stop()
}

func (c *remoteClient) call(ctx context.Context, q remoteRequest, callback func(remoteResponse) error) (out remoteResponse, err error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.closed {
		return out, os.ErrClosed
	}
	if ctx != nil {
		if err = ctx.Err(); err != nil {
			return
		}
		if deadline, ok := ctx.Deadline(); ok {
			q.Deadline = deadline.UnixNano()
		}
	}
	if c.input == nil {
		if c.connect == nil {
			return out, os.ErrClosed
		}
		reconnectContext := ctx
		if reconnectContext == nil {
			reconnectContext = context.Background()
		}
		if err = c.reconnect(reconnectContext); err != nil {
			return
		}
	}
	// Closing the pipe cancels the worker context, including an in-flight SQL call.
	if ctx != nil {
		input, output := c.input, c.output
		stopped := make(chan struct{})
		stop := context.AfterFunc(ctx, func() { input.Close(); output.Close(); close(stopped) })
		defer func() {
			if !stop() {
				<-stopped
			}
			if ctx.Err() != nil {
				err = ctx.Err()
				c.disconnect()
			}
		}()
	}
	// A callback panic or a broken pipe must release any child transaction.
	complete := false
	defer func() {
		if !complete {
			c.disconnect()
		}
	}()
	if err = c.enc.Encode(q); err != nil {
		return
	}
	if err = c.dec.Decode(&out); err != nil {
		return
	}
	var callbackErr error
	if out.Callback != "" {
		if callback == nil {
			return out, errRemoteMethod
		}
		callbackErr = callback(out)
		if ctx != nil && ctx.Err() != nil {
			return out, ctx.Err()
		}
		ack := remoteRequest{Method: "callback"}
		if callbackErr != nil {
			ack.CallbackFailed = true
			ack.CallbackError = callbackErr.Error()
		}
		if err = c.enc.Encode(ack); err != nil {
			return
		}
		out = remoteResponse{}
		if err = c.dec.Decode(&out); err != nil {
			return
		}
		if out.Callback != "" {
			return out, errRemoteMethod
		}
	}
	complete = true
	if callbackErr != nil {
		return out, callbackErr
	}
	return out, responseError(out)
}

// ServeRemote owns SQLite in the worker. Callers must drop credentials first.
func ServeRemote(path string, input io.Reader, output io.Writer) error {
	return serveRemote(path, input, output, false)
}

func ServeRemoteReadOnly(path string, input io.Reader, output io.Writer) error {
	return serveRemote(path, input, output, true)
}

func serveRemote(path string, input io.Reader, output io.Writer, readOnly bool) error {
	if os.Geteuid() == 0 {
		return errors.New("registry: remote worker refuses root")
	}
	enc, dec := gob.NewEncoder(output), gob.NewDecoder(input)
	var r *Registry
	var err error
	if readOnly {
		r, err = OpenReadOnly(path)
	} else {
		r, err = Open(path)
	}
	ready := remoteResponse{}
	setResponseError(&ready, err)
	if writeErr := enc.Encode(ready); writeErr != nil {
		if r != nil {
			r.Close()
		}
		return writeErr
	}
	if err != nil {
		return err
	}
	defer r.Close()
	workerCtx, stopWorker := context.WithCancel(context.Background())
	defer stopWorker()
	requests := make(chan remoteRequest)
	readErrors := make(chan error, 1)
	go func() {
		defer close(requests)
		defer stopWorker()
		for {
			var q remoteRequest
			if err := dec.Decode(&q); err != nil {
				readErrors <- err
				return
			}
			select {
			case requests <- q:
			case <-workerCtx.Done():
				return
			}
		}
	}()
	for q := range requests {
		if readOnly && q.Method != "Get" && q.Method != "TimersFor" && q.Method != "ArtifactHosts" {
			out := remoteResponse{}
			setResponseError(&out, errRemoteMethod)
			if err := enc.Encode(out); err != nil {
				return err
			}
			continue
		}
		ctx := workerCtx
		cancel := func() {}
		if q.Deadline != 0 {
			ctx, cancel = context.WithDeadline(ctx, time.Unix(0, q.Deadline))
		}
		out, callErr := r.dispatchRemote(ctx, q, func(event remoteResponse) error {
			if err := enc.Encode(event); err != nil {
				return err
			}
			ack, ok := <-requests
			if !ok {
				return workerCtx.Err()
			}
			if ack.Method != "callback" {
				return errRemoteMethod
			}
			if ack.CallbackFailed {
				return errors.New(ack.CallbackError)
			}
			return nil
		})
		cancel()
		setResponseError(&out, callErr)
		if err = enc.Encode(out); err != nil {
			return err
		}
	}
	select {
	case err := <-readErrors:
		if errors.Is(err, io.EOF) {
			return nil
		}
		return err
	default:
		return workerCtx.Err()
	}

}