proto-socket/go/crosstest/go_kotlin.go

//go:build ignore

package main

import (
	"bufio"
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"io"
	"net"
	"os"
	"os/exec"
	"path/filepath"
	"regexp"
	"runtime"
	"sync"
	"time"

	"google.golang.org/protobuf/proto"
	"nhooyr.io/websocket"

	toki "git.toki-labs.com/toki/proto-socket/go"
	"git.toki-labs.com/toki/proto-socket/go/packets"
)

const (
	host                    = "127.0.0.1"
	goKotlinTCPPort         = 29290
	goKotlinWSPort          = 29292
	goKotlinTLSPort         = 29294
	goKotlinWSSPort         = 29296
	wsPath                  = "/"
	processTimeout          = 20 * time.Second
	serverObservationWindow = 200 * time.Millisecond
)

func parserMap() toki.ParserMap {
	return toki.ParserMap{
		toki.TypeNameOf(&packets.TestData{}): func(b []byte) (proto.Message, error) {
			m := &packets.TestData{}
			return m, proto.Unmarshal(b, m)
		},
	}
}

func main() {
	fmt.Printf("INFO typeName go=%s\n", toki.TypeNameOf(&packets.TestData{}))
	if err := run(); err != nil {
		fmt.Fprintf(os.Stderr, "FAIL crosstest error=%v\n", err)
		os.Exit(1)
	}
	fmt.Println("PASS all go-server/kotlin-client crosstests passed")
}

func run() error {
	if err := runTCPSendPush(); err != nil {
		return err
	}
	time.Sleep(150 * time.Millisecond)
	if err := runTCPRequests(); err != nil {
		return err
	}
	time.Sleep(150 * time.Millisecond)
	if err := runWSSendPush(); err != nil {
		return err
	}
	time.Sleep(150 * time.Millisecond)
	if err := runWSRequests(); err != nil {
		return err
	}
	time.Sleep(150 * time.Millisecond)

	dartDir, err := dartPackageDir()
	if err != nil {
		return fmt.Errorf("cannot find dart package dir: %w", err)
	}
	certFile := filepath.Join(dartDir, "test", "certs", "server.crt")
	keyFile := filepath.Join(dartDir, "test", "certs", "server.key")
	serverTLS, err := loadServerTLS(certFile, keyFile)
	if err != nil {
		return fmt.Errorf("load TLS certs: %w", err)
	}

	if err := runTLSTCPSendPush(serverTLS, certFile); err != nil {
		return err
	}
	time.Sleep(150 * time.Millisecond)
	if err := runTLSTCPRequests(serverTLS, certFile); err != nil {
		return err
	}
	time.Sleep(150 * time.Millisecond)
	if err := runWSSendPushSecure(serverTLS, certFile); err != nil {
		return err
	}
	time.Sleep(150 * time.Millisecond)
	return runWSSRequestsSecure(serverTLS, certFile)
}

func loadServerTLS(certFile, keyFile string) (*tls.Config, error) {
	cert, err := tls.LoadX509KeyPair(certFile, keyFile)
	if err != nil {
		return nil, err
	}
	return &tls.Config{Certificates: []tls.Certificate{cert}}, nil
}

func runTCPSendPush() error {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	received := make(chan bool, 1)
	server := toki.NewTcpServer(host, goKotlinTCPPort, func(conn net.Conn) *toki.TcpClient {
		return toki.NewTcpClient(conn, 0, 0, parserMap())
	})
	server.OnClientConnected = func(client *toki.TcpClient) {
		toki.AddListenerTyped[*packets.TestData](&client.Communicator, func(data *packets.TestData) {
			fmt.Printf("SERVER_RECEIVED index=%d message=%s\n", data.GetIndex(), data.GetMessage())
			valid := data.GetIndex() == 101 && data.GetMessage() == "fire from kotlin client"
			select {
			case received <- valid:
			default:
			}
			if valid {
				_ = client.Send(&packets.TestData{Index: 200, Message: "push from go server"})
			}
		})
	}
	if err := server.Start(ctx); err != nil {
		return err
	}
	defer server.Stop()

	if err := runKotlinClient("tcp", goKotlinTCPPort, "send-push", map[string]bool{"1": true, "2": true}); err != nil {
		return err
	}
	select {
	case ok := <-received:
		if !ok {
			return errors.New("TCP send-push server received unexpected data")
		}
	case <-time.After(serverObservationWindow):
		return errors.New("TCP send-push server did not receive expected data")
	}
	return nil
}

func runTCPRequests() error {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	server := toki.NewTcpServer(host, goKotlinTCPPort, func(conn net.Conn) *toki.TcpClient {
		return toki.NewTcpClient(conn, 0, 0, parserMap())
	})
	server.OnClientConnected = func(client *toki.TcpClient) {
		toki.AddRequestListenerTyped[*packets.TestData, *packets.TestData](&client.Communicator, func(req *packets.TestData) (*packets.TestData, error) {
			return &packets.TestData{
				Index:   req.GetIndex() * 2,
				Message: "echo: " + req.GetMessage(),
			}, nil
		})
	}
	if err := server.Start(ctx); err != nil {
		return err
	}
	defer server.Stop()

	return runKotlinClient("tcp", goKotlinTCPPort, "requests", map[string]bool{"3": true, "4": true})
}

func runWSSendPush() error {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	received := make(chan bool, 1)
	server := toki.NewWsServer(host, goKotlinWSPort, wsPath, func(conn *websocket.Conn) *toki.WsClient {
		return toki.NewWsClient(conn, 0, 0, parserMap())
	})
	server.OnClientConnected = func(client *toki.WsClient) {
		toki.AddListenerTyped[*packets.TestData](&client.Communicator, func(data *packets.TestData) {
			fmt.Printf("SERVER_RECEIVED index=%d message=%s\n", data.GetIndex(), data.GetMessage())
			valid := data.GetIndex() == 101 && data.GetMessage() == "fire from kotlin client"
			select {
			case received <- valid:
			default:
			}
			if valid {
				_ = client.Send(&packets.TestData{Index: 200, Message: "push from go server"})
			}
		})
	}
	if err := server.Start(ctx); err != nil {
		return err
	}
	defer server.Stop()

	if err := runKotlinClient("ws", goKotlinWSPort, "send-push", map[string]bool{"1": true, "2": true}); err != nil {
		return err
	}
	select {
	case ok := <-received:
		if !ok {
			return errors.New("WS send-push server received unexpected data")
		}
	case <-time.After(serverObservationWindow):
		return errors.New("WS send-push server did not receive expected data")
	}
	return nil
}

func runWSRequests() error {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	server := toki.NewWsServer(host, goKotlinWSPort, wsPath, func(conn *websocket.Conn) *toki.WsClient {
		return toki.NewWsClient(conn, 0, 0, parserMap())
	})
	server.OnClientConnected = func(client *toki.WsClient) {
		toki.AddRequestListenerTyped[*packets.TestData, *packets.TestData](&client.Communicator, func(req *packets.TestData) (*packets.TestData, error) {
			return &packets.TestData{
				Index:   req.GetIndex() * 2,
				Message: "echo: " + req.GetMessage(),
			}, nil
		})
	}
	if err := server.Start(ctx); err != nil {
		return err
	}
	defer server.Stop()

	return runKotlinClient("ws", goKotlinWSPort, "requests", map[string]bool{"3": true, "4": true})
}

func runTLSTCPSendPush(serverTLS *tls.Config, certFile string) error {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	received := make(chan bool, 1)
	server := toki.NewTcpServerTLS(host, goKotlinTLSPort, serverTLS, func(conn net.Conn) *toki.TcpClient {
		return toki.NewTcpClient(conn, 0, 0, parserMap())
	})
	server.OnClientConnected = func(client *toki.TcpClient) {
		toki.AddListenerTyped[*packets.TestData](&client.Communicator, func(data *packets.TestData) {
			fmt.Printf("SERVER_RECEIVED index=%d message=%s\n", data.GetIndex(), data.GetMessage())
			valid := data.GetIndex() == 101 && data.GetMessage() == "fire from kotlin client"
			select {
			case received <- valid:
			default:
			}
			if valid {
				_ = client.Send(&packets.TestData{Index: 200, Message: "push from go server"})
			}
		})
	}
	if err := server.Start(ctx); err != nil {
		return err
	}
	defer server.Stop()

	if err := runKotlinClient("tls", goKotlinTLSPort, "send-push", map[string]bool{"1": true, "2": true}, certFile); err != nil {
		return err
	}
	select {
	case ok := <-received:
		if !ok {
			return errors.New("TLS TCP send-push server received unexpected data")
		}
	case <-time.After(serverObservationWindow):
		return errors.New("TLS TCP send-push server did not receive expected data")
	}
	return nil
}

func runTLSTCPRequests(serverTLS *tls.Config, certFile string) error {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	server := toki.NewTcpServerTLS(host, goKotlinTLSPort, serverTLS, func(conn net.Conn) *toki.TcpClient {
		return toki.NewTcpClient(conn, 0, 0, parserMap())
	})
	server.OnClientConnected = func(client *toki.TcpClient) {
		toki.AddRequestListenerTyped[*packets.TestData, *packets.TestData](&client.Communicator, func(req *packets.TestData) (*packets.TestData, error) {
			return &packets.TestData{
				Index:   req.GetIndex() * 2,
				Message: "echo: " + req.GetMessage(),
			}, nil
		})
	}
	if err := server.Start(ctx); err != nil {
		return err
	}
	defer server.Stop()

	return runKotlinClient("tls", goKotlinTLSPort, "requests", map[string]bool{"3": true, "4": true}, certFile)
}

func runWSSendPushSecure(serverTLS *tls.Config, certFile string) error {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	received := make(chan bool, 1)
	server := toki.NewWsServerTLS(host, goKotlinWSSPort, wsPath, serverTLS, func(conn *websocket.Conn) *toki.WsClient {
		return toki.NewWsClient(conn, 0, 0, parserMap())
	})
	server.OnClientConnected = func(client *toki.WsClient) {
		toki.AddListenerTyped[*packets.TestData](&client.Communicator, func(data *packets.TestData) {
			fmt.Printf("SERVER_RECEIVED index=%d message=%s\n", data.GetIndex(), data.GetMessage())
			valid := data.GetIndex() == 101 && data.GetMessage() == "fire from kotlin client"
			select {
			case received <- valid:
			default:
			}
			if valid {
				_ = client.Send(&packets.TestData{Index: 200, Message: "push from go server"})
			}
		})
	}
	if err := server.Start(ctx); err != nil {
		return err
	}
	defer server.Stop()

	if err := runKotlinClient("wss", goKotlinWSSPort, "send-push", map[string]bool{"1": true, "2": true}, certFile); err != nil {
		return err
	}
	select {
	case ok := <-received:
		if !ok {
			return errors.New("WSS send-push server received unexpected data")
		}
	case <-time.After(serverObservationWindow):
		return errors.New("WSS send-push server did not receive expected data")
	}
	return nil
}

func runWSSRequestsSecure(serverTLS *tls.Config, certFile string) error {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	server := toki.NewWsServerTLS(host, goKotlinWSSPort, wsPath, serverTLS, func(conn *websocket.Conn) *toki.WsClient {
		return toki.NewWsClient(conn, 0, 0, parserMap())
	})
	server.OnClientConnected = func(client *toki.WsClient) {
		toki.AddRequestListenerTyped[*packets.TestData, *packets.TestData](&client.Communicator, func(req *packets.TestData) (*packets.TestData, error) {
			return &packets.TestData{
				Index:   req.GetIndex() * 2,
				Message: "echo: " + req.GetMessage(),
			}, nil
		})
	}
	if err := server.Start(ctx); err != nil {
		return err
	}
	defer server.Stop()

	return runKotlinClient("wss", goKotlinWSSPort, "requests", map[string]bool{"3": true, "4": true}, certFile)
}

func runKotlinClient(mode string, port int, phase string, expected map[string]bool, certFile ...string) error {
	ctx, cancel := context.WithTimeout(context.Background(), processTimeout)
	defer cancel()

	kotlinDir, err := kotlinPackageDir()
	if err != nil {
		return err
	}

	clientArgs := "--mode=" + mode + " --port=" + fmt.Sprint(port) + " --phase=" + phase
	if len(certFile) > 0 && certFile[0] != "" {
		clientArgs += " --cert=" + certFile[0]
	}
	cmd := exec.CommandContext(ctx, "./gradlew", "run", "--args="+clientArgs)
	cmd.Dir = kotlinDir

	stdoutPipe, err := cmd.StdoutPipe()
	if err != nil {
		return err
	}
	stderrPipe, err := cmd.StderrPipe()
	if err != nil {
		return err
	}
	if err := cmd.Start(); err != nil {
		return err
	}

	var mu sync.Mutex
	var resultLines []string
	var wg sync.WaitGroup
	wg.Add(2)
	go scanLines(&wg, stdoutPipe, os.Stdout, &mu, &resultLines)
	go scanLines(&wg, stderrPipe, os.Stderr, nil, nil)

	waitErr := cmd.Wait()
	wg.Wait()
	if ctx.Err() == context.DeadlineExceeded {
		return fmt.Errorf("kotlin client %s/%s timed out", mode, phase)
	}
	return validateResultLines("kotlin-client "+mode+"/"+phase, waitErr, resultLines, expected)
}

func kotlinPackageDir() (string, error) {
	candidates := make([]string, 0, 3)

	_, filename, _, ok := runtime.Caller(0)
	if ok {
		repoRoot := filepath.Dir(filepath.Dir(filepath.Dir(filename)))
		candidates = append(candidates, filepath.Join(repoRoot, "kotlin"))
	}
	if wd, err := os.Getwd(); err == nil {
		candidates = append(candidates, findKotlinPackageCandidates(wd)...)
	}
	if executable, err := os.Executable(); err == nil {
		candidates = append(candidates, findKotlinPackageCandidates(filepath.Dir(executable))...)
	}

	for _, candidate := range candidates {
		if isKotlinPackageDir(candidate) {
			return candidate, nil
		}
	}
	return "", fmt.Errorf("cannot resolve kotlin package directory from candidates %v", candidates)
}

func findKotlinPackageCandidates(start string) []string {
	candidates := make([]string, 0)
	for dir := start; ; dir = filepath.Dir(dir) {
		candidates = append(candidates, filepath.Join(dir, "kotlin"))
		if filepath.Base(dir) == "kotlin" {
			candidates = append(candidates, dir)
		}
		parent := filepath.Dir(dir)
		if parent == dir {
			return candidates
		}
	}
}

func isKotlinPackageDir(dir string) bool {
	info, err := os.Stat(filepath.Join(dir, "build.gradle.kts"))
	return err == nil && !info.IsDir()
}

func dartPackageDir() (string, error) {
	candidates := make([]string, 0, 3)

	_, filename, _, ok := runtime.Caller(0)
	if ok {
		repoRoot := filepath.Dir(filepath.Dir(filepath.Dir(filename)))
		candidates = append(candidates, filepath.Join(repoRoot, "dart"))
	}
	if wd, err := os.Getwd(); err == nil {
		candidates = append(candidates, findDartPackageCandidates(wd)...)
	}
	if executable, err := os.Executable(); err == nil {
		candidates = append(candidates, findDartPackageCandidates(filepath.Dir(executable))...)
	}

	for _, candidate := range candidates {
		if isDartPackageDir(candidate) {
			return candidate, nil
		}
	}
	return "", fmt.Errorf("cannot resolve dart package directory from candidates %v", candidates)
}

func findDartPackageCandidates(start string) []string {
	candidates := make([]string, 0)
	for dir := start; ; dir = filepath.Dir(dir) {
		candidates = append(candidates, filepath.Join(dir, "dart"))
		if filepath.Base(dir) == "dart" {
			candidates = append(candidates, dir)
		}
		parent := filepath.Dir(dir)
		if parent == dir {
			return candidates
		}
	}
}

func isDartPackageDir(dir string) bool {
	info, err := os.Stat(filepath.Join(dir, "pubspec.yaml"))
	return err == nil && !info.IsDir()
}

func scanLines(wg *sync.WaitGroup, reader io.Reader, writer *os.File, mu *sync.Mutex, resultLines *[]string) {
	defer wg.Done()
	scanner := bufio.NewScanner(reader)
	for scanner.Scan() {
		line := scanner.Text()
		fmt.Fprintln(writer, line)
		if mu != nil && startsWithResult(line) {
			mu.Lock()
			*resultLines = append(*resultLines, line)
			mu.Unlock()
		}
	}
}

func validateResultLines(label string, waitErr error, lines []string, expected map[string]bool) error {
	failed := make([]string, 0)
	passed := make(map[string]bool)
	re := regexp.MustCompile(`scenario=([^ ]+)`)
	for _, line := range lines {
		if len(line) >= 5 && line[:5] == "FAIL " {
			failed = append(failed, line)
			continue
		}
		if len(line) >= 5 && line[:5] == "PASS " {
			match := re.FindStringSubmatch(line)
			if len(match) == 2 {
				passed[match[1]] = true
			}
		}
	}

	missing := make([]string, 0)
	for scenario := range expected {
		if !passed[scenario] {
			missing = append(missing, scenario)
		}
	}
	if waitErr != nil || len(failed) > 0 || len(missing) > 0 {
		return fmt.Errorf("%s failed waitErr=%v failed=%v missing=%v", label, waitErr, failed, missing)
	}
	return nil
}

func startsWithResult(line string) bool {
	return (len(line) >= 5 && line[:5] == "PASS ") || (len(line) >= 5 && line[:5] == "FAIL ")
}