mc-god/cmd/mcgod/main.go

package main

import (
	"context"
	"fmt"
	"log"
	"log/slog"
	"os"
	"os/signal"
	"regexp"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/gogo/protobuf/proto"
	"github.com/ollama/ollama/api"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"tipsy.codes/charles/mc-god/v2/internal/pkg/logs"
	"tipsy.codes/charles/mc-god/v2/internal/pkg/rcon"
	"tipsy.codes/charles/mc-god/v2/internal/pkg/tools"
	"tipsy.codes/charles/mc-god/v2/internal/pkg/tools/weather"
	"tipsy.codes/charles/mc-god/v2/internal/pkg/tools/zombie"
)

type chatContext struct {
	chatRequest *api.ChatRequest
	totalSize   int
	maxSize     int
	mu          sync.Mutex
}

func (c *chatContext) AddLog(msg string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.chatRequest.Messages = append(c.chatRequest.Messages, api.Message{
		Role:    "user",
		Content: msg,
	})
	c.totalSize += len(msg)
	c.truncate()
}

func (c *chatContext) AddSelf(msg api.Message) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.chatRequest.Messages = append(c.chatRequest.Messages, msg)
	c.totalSize += len(msg.Content)
	slog.Info("adding message", "msg", msg, "content", msg.Content)
	c.truncate()
}

func (c *chatContext) AddTool(msg string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.chatRequest.Messages = append(c.chatRequest.Messages, api.Message{
		Role:    "tool",
		Content: msg,
	})
	c.totalSize += len(msg)
	c.truncate()
}

func (c *chatContext) truncate() {
	for c.maxSize != 0 && c.totalSize > c.maxSize && len(c.chatRequest.Messages) > 1 {
		t := c.chatRequest.Messages[1]
		c.chatRequest.Messages = c.chatRequest.Messages[2:]
		c.totalSize -= len(t.Content)
	}
}

func main() {
	// Create a context that will be cancelled on interrupt signals
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	_ = ctx

	// Set up signal handling for graceful shutdown
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
	go func() {
		<-sigChan
		slog.Info("received interrupt signal, shutting down...")
		cancel()
	}()

	// Create RCON client from environment variables
	log.Println("Connecting to Minecraft server via RCON...")
	client, err := rcon.NewFromEnv()
	if err != nil {
		slog.Error("failed to create RCON client", "error", err)
		return
	}
	defer func() {
		if err := client.Close(); err != nil {
			slog.Warn("error closing RCON connection", "error", err)
		}
	}()

	// Perform a health check
	log.Println("Performing healthcheck...")
	if err := client.HealthCheck(); err != nil {
		slog.Error("Health check failed", "error", err)
		return
	}
	log.Println("Connected successfully!")

	// Create Kubernetes client
	kClient, err := createKubernetesClient()
	if err != nil {
		slog.Error("failed to create kubernetes client", "error", err)
		return
	}
	slog.Info("got kubernetes config")

	tailer, done := logs.LoggerFromEnv().Start(ctx, kClient)
	defer func() {
		if err := done(); err != nil {
			slog.Error("problem with tailer", "error", err)
		}
	}()
	slog.Info("logger started")

	ollamaClient, err := api.ClientFromEnvironment()
	if err != nil {
		slog.Error("error getting ollama client", "error", err)
	}

	rClient, err := rcon.NewFromEnv()
	if err != nil {
		slog.Error("failed to get rcon client", "error", err)
		return
	}

	tools := tools.New(
		weather.Get(),
		zombie.Get(),
	)

	// Start goroutines to do the things
	chatRequest := &api.ChatRequest{
		Model:     "qwen3-coder",
		Stream:    proto.Bool(false),
		KeepAlive: &api.Duration{Duration: time.Hour},
		Tools:     tools.AsAPI(),
		Think:     &api.ThinkValue{Value: false},
		Shift:     proto.Bool(true),
		Messages: []api.Message{
			api.Message{
				Role: "system",
				Content: `
		You are Minecraft server admin with a god complex. You are a benevolent god.

		We are having fun with the players, but not trying to kill them.
		Spawn zombies very sparingly, and only in response to direct challenge.

		When a user talks, you will see this in the logs:

			[18:45:10] [Server thread/INFO]: <SomePlayer> hello world.

		The user here is SomePlayer, who said "hello world."

		A log message like:

			[18:45:10] [Server thread/INFO]: SomePlayer joined the game

		Indicates that SomePlayer has joined the game.

		Logs like:

			[00:40:10] [Server thread/INFO]: SomePlayer lost connection: Disconnected
			[00:40:10] [Server thread/INFO]: SomePlayer left the game

		Indicate the player SomePlayer has left the game.

		Some messages indicate a player died; it varies depending on how they died
		and we can't know all variations up front. Here is an example where SomePlayer
		was killed by a zombie.

			[05:21:51] [Server thread/INFO]: OrangeYouSad was slain by Zombie

		If a player dies, mock them.

		If a player talks, talk back.

		When a player joins the game, greet them. Include their name.

		Responses should be short; one sentence. Only write messages
		in response to the situations described above.
		If there is nothing interesting to say, say "SKIP".
		`,
			},
		},
	}
	chat := &chatContext{
		chatRequest: chatRequest,
		maxSize:     10000000,
	}

	events := make(chan bool, 1000)

	doneWg := sync.WaitGroup{}
	doneWg.Go(handleOllama(ctx, ollamaClient, chat, rClient, tools, events))

	rconRegex := regexp.MustCompile(`^\[\d\d:\d\d:\d\d\] \[Server thread\/INFO\]: (\[Not Secure\] \[Rcon\]|\[Rcon: ) .*`)
	//allowedMessages := regexp.MustCompile(`^\[\d\d:\d\d:\d\d\] \[Server thread/INFO\]: (<.*>|.* has lost connection|.*left the game|.*joined the game)`)
	for line := range tailer.NextLine() {
		if rconRegex.Match([]byte(line)) {
			slog.Info("Skipping line; RCON")
			continue
		}
		//if allowedMessages.Match([]byte(line)) {
		slog.Info("mc log", "msg", line)
		chat.AddLog(line)
		events <- true
		//}
	}

	doneWg.Wait()
}

func handleOllama(ctx context.Context, client *api.Client, chat *chatContext, rClient *rcon.Client, tools tools.Tools, events chan bool) func() {
	slog.Info("got chat request", "object", fmt.Sprintf("%+v", chat.chatRequest))
	return func() {
		var chatResponse api.ChatResponse
		for {
			chat.mu.Lock()
			slog.Info("Chatting...")
			// slog.Info("sending chat request", "object", fmt.Sprintf("%#v", chat.chatRequest))
			err := client.Chat(ctx, chat.chatRequest, func(cr api.ChatResponse) error {
				chatResponse = cr
				return nil
			})
			chat.mu.Unlock()
			slog.Info("Done chatting!")
			if err != nil {
				slog.Error("error calling ollama", "error", err)
				return
			}
			chat.AddSelf(chatResponse.Message)
			for _, toolCall := range chatResponse.Message.ToolCalls {
				if err := tools.Do(ctx, toolCall, rClient); err != nil {
					slog.Warn("failed to run tool", "error", err)
					//chat.AddTool(fmt.Sprintf("failed to call tool %s: %s", toolCall.ID, err))
					continue
				}
			}
			if len(chatResponse.Message.ToolCalls) == 0 {
				if strings.TrimSpace(chatResponse.Message.Content) == "SKIP" {
					slog.Info("nothing to do; napping")
				} else {
					msg := chatResponse.Message.Content
					msg = strings.ReplaceAll(msg, "\n", " ")
					if err := rClient.Say(msg); err != nil {
						slog.Error("error talking", "error", err)
					}
				}
				select {
				case <-events:
					var done bool
					for !done {
						select {
						case <-events:
							continue
						case <-time.Tick(time.Millisecond * 50):
							done = true
						}
					}
				case <-ctx.Done():
					return
				}
				continue
			}
		}
	}
}

func createKubernetesClient() (*kubernetes.Clientset, error) {
	// Try to load in-cluster config first
	config, err := rest.InClusterConfig()
	if err != nil {
		// If in-cluster config fails, try kubeconfig
		loadingRules := clientcmd.NewDefaultClientConfigLoadingRules()
		configOverrides := &clientcmd.ConfigOverrides{}
		kubeConfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, configOverrides)
		config, err = kubeConfig.ClientConfig()
		if err != nil {
			return nil, fmt.Errorf("failed to create kubernetes client: %w", err)
		}
	}

	client, err := kubernetes.NewForConfig(config)
	if err != nil {
		return nil, fmt.Errorf("failed to create kubernetes client: %w", err)
	}

	return client, nil
}