// Package relay wires the relay's domains into the fixed spawn tree. The Tree
// value is the orchestrator: it creates channels and breakers, spawns the
// domains, waits for their startup, and tears the tree down. It holds no
// protocol, storage or socket state of its own.
package relay

import (
	"errors"
	"fmt"
	"sync"
	"time"

	"git.smesh.lol/torly/pkg/config"
	"git.smesh.lol/torly/pkg/domain/blob"
	"git.smesh.lol/torly/pkg/domain/db"
	iodomain "git.smesh.lol/torly/pkg/domain/io"
	netdomain "git.smesh.lol/torly/pkg/domain/net"
	"git.smesh.lol/torly/pkg/domain/verify"
	tinyspawn "git.smesh.lol/torly/pkg/tinyspawn"
	"git.smesh.lol/torly/pkg/wire"
)

// Tree is a running relay.
type Tree struct {
	cfg  config.Config
	addr string
	stop tinyspawn.Breaker
	done []<-chan struct{}
	once sync.Once
}

// Start opens the listening socket, spawns the domain tree and waits for the
// domains that own durable state to finish opening it.
func Start(cfg config.Config) (*Tree, error) {
	sk, err := loadRelayKey(&cfg)
	if err != nil {
		return nil, fmt.Errorf("relay identity: %w", err)
	}
	t := &Tree{cfg: cfg, addr: cfg.Addr(), stop: tinyspawn.NewBreaker()}

	dbReady := make(chan wire.Ready, 1)
	blobReady := make(chan wire.Ready, 1)

	dbReqNet, dbReqDB := tinyspawn.NewBufferedDuplex[wire.DBRequest](64)
	dbRespNet, dbRespDB := tinyspawn.NewBufferedDuplex[wire.DBResponse](64)
	blobReqNet, blobReqBlob := tinyspawn.NewBufferedDuplex[wire.BlobRequest](32)
	blobRespNet, blobRespBlob := tinyspawn.NewBufferedDuplex[wire.BlobResponse](32)

	t.add("db", tinyspawn.Spawn(db.Run, db.Data{
		Dir:        cfg.DataDir + "/db",
		MaxResults: cfg.QueryLimit,
		Expiry:     time.Duration(cfg.ExpiryInterval) * time.Second,
	}, db.Chans{Req: dbReqDB, Resp: dbRespDB, Ready: dbReady, Stop: t.stop}))

	t.add("blob", tinyspawn.Spawn(blob.Run, blob.Data{
		Dir:     cfg.DataDir,
		MaxBlob: cfg.BlossomMaxBlob,
	}, blob.Chans{Req: blobReqBlob, Resp: blobRespBlob, Ready: blobReady, Stop: t.stop}))

	if err := waitReadyFor(dbReady, 2*time.Minute); err != nil {
		t.Close()
		return nil, fmt.Errorf("database: %w", err)
	}
	if cfg.BlossomEnabled {
		if err := waitReady(blobReady); err != nil {
			t.Close()
			return nil, fmt.Errorf("blossom: %w", err)
		}
	} else {
		// The blob domain still opened an index; it is harmless and keeps the
		// fixed tree shape.
		_ = waitReady(blobReady)
	}

	netCh := netdomain.Chans{
		DBReq:    dbReqNet,
		DBResp:   dbRespNet,
		BlobReq:  blobReqNet,
		BlobResp: blobRespNet,
		Ready:    make(chan wire.Ready, 1),
		Stop:     t.stop,
	}

	for i := 0; i < cfg.VerifyWorkers; i++ {
		reqNet, reqWorker := tinyspawn.NewBufferedDuplex[wire.VerifyRequest](8)
		respNet, respWorker := tinyspawn.NewBufferedDuplex[wire.VerifyResponse](8)
		netCh.Verifiers = append(netCh.Verifiers, reqNet)
		netCh.VerifyResp = append(netCh.VerifyResp, respNet)
		t.add("verify", tinyspawn.Spawn(verify.Run, struct{}{}, verify.Chans{
			Req:   reqWorker,
			Resp:  respWorker,
			Ready: make(chan wire.Ready, 1),
			Stop:  t.stop,
		}))
	}

	workers := cfg.ConnWorkers
	if cfg.MaxConnections > 0 && cfg.MaxConnections < workers {
		workers = cfg.MaxConnections
	}
	acceptorCh := iodomain.AcceptorChans{Stop: t.stop}
	for i := 0; i < workers; i++ {
		assignAcceptor, assignReader := tinyspawn.NewBufferedDuplex[wire.Assign](1)
		inReader, inNet := tinyspawn.NewBufferedDuplex[wire.Inbound](64)
		outNet, outWriter := tinyspawn.NewBufferedDuplex[wire.Outbound](64)
		ctlReader, ctlWriter := tinyspawn.NewBufferedDuplex[wire.WriterCtl](32)
		ackReader, ackWriter := tinyspawn.NewBufferedDuplex[uint8](2)

		netCh.Workers = append(netCh.Workers, netdomain.WorkerLink{In: inNet, Out: outNet})
		acceptorCh.Workers = append(acceptorCh.Workers, assignAcceptor)

		t.add("reader", tinyspawn.Spawn(iodomain.Reader, iodomain.ReaderData{Index: i, Cfg: cfg}, iodomain.ReaderChans{
			Assign: assignReader,
			In:     inReader,
			Ctl:    ctlReader,
			Ack:    ackReader,
			Stop:   t.stop,
		}))
		t.add("writer", tinyspawn.Spawn(iodomain.Writer, iodomain.WriterData{Index: i}, iodomain.WriterChans{
			Ctl:  ctlWriter,
			Out:  outWriter,
			Ack:  ackWriter,
			Stop: t.stop,
		}))
	}

	t.add("net", tinyspawn.Spawn(netdomain.Run, netdomain.Data{Cfg: cfg, RelaySk: sk}, netCh))

	acceptorReady := make(chan wire.Ready, 1)
	acceptorCh.Ready = acceptorReady
	t.add("acceptor", tinyspawn.Spawn(iodomain.Acceptor, iodomain.AcceptorData{Addr: cfg.Addr(), Cfg: cfg}, acceptorCh))
	select {
	case r := <-acceptorReady:
		if !r.OK {
			t.Close()
			return nil, errors.New("acceptor: " + r.Err)
		}
		if r.Addr != "" {
			t.addr = r.Addr
		}
	case <-time.After(10 * time.Second):
		t.Close()
		return nil, errors.New("acceptor: startup timed out")
	}
	return t, nil
}

func (t *Tree) add(_ string, done <-chan struct{}) { t.done = append(t.done, done) }

func waitReady(ch <-chan wire.Ready) error {
	return waitReadyFor(ch, 10*time.Second)
}

func waitReadyFor(ch <-chan wire.Ready, limit time.Duration) error {
	select {
	case r := <-ch:
		if !r.OK {
			return errors.New(r.Err)
		}
		return nil
	case <-time.After(limit):
		return errors.New("startup timed out")
	}
}

// Addr returns the listening address.
func (t *Tree) Addr() string { return t.addr }

// Config returns the immutable configuration used to start the tree.
func (t *Tree) Config() config.Config { return t.cfg }

// Exited reports whether any domain has returned.
func (t *Tree) Exited() bool {
	for _, d := range t.done {
		select {
		case <-d:
			return true
		default:
		}
	}
	return false
}

// Close cancels the tree, closes the listening socket and waits for every
// domain to return.
func (t *Tree) Close() {
	t.once.Do(func() { t.stop.Close() })
	deadline := time.Now().Add(5 * time.Second)
	stopped := 0
	for _, d := range t.done {
		left := time.Until(deadline)
		if left <= 0 {
			break
		}
		select {
		case <-d:
			stopped++
		case <-time.After(left):
		}
	}

}