feat: service driven player transfers, send to hub when removed as a builder, drain map when the owner begins verification

Author: mworzala

api/mapsV3/intnl/server_map.go

@@ -6,12 +6,13 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/prometheus/client_golang/prometheus"
 	"math"
 	"strconv"
 	"strings"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/gohugoio/hashstructure"
 	"github.com/hollow-cube/api-server/internal/mapdb"
 	"github.com/hollow-cube/api-server/internal/object"
@@ -574,6 +575,18 @@ func (s *server) BeginMapVerification(ctx context.Context, request BeginMapVerif
 		return nil, fmt.Errorf("failed to update map: %w", err)
 	}
 
+	// If we are starting to verify, send out a drain to destory any possible editing worlds.
+	// TODO: This doesnt really work. If there are people in the world it wont have time to save
+	//  and the verifying world will be out of date which would allow publishing impossible maps
+	err = s.jetStream.PublishJSONAsync(ctx, model.MapUpdateMessage{
+		Action:      model.MapUpdate_Drain,
+		ID:          request.MapId,
+		DrainReason: new("verification"),
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to publish map update: %w", err)
+	}
+
 	return BeginMapVerification200Response{}, nil
 }
 
@@ -1246,4 +1259,4 @@ func createMapSearchCacheKey(params *mapdb.SearchMapsParams) (string, bool) {
 
 func init() {
 	prometheus.MustRegister(mapUploadDuration)
-}
+}

api/v4Internal/server.go

@@ -4,6 +4,7 @@ import (
 	sessiondb "github.com/hollow-cube/api-server/internal/db"
 	"github.com/hollow-cube/api-server/internal/interaction"
 	"github.com/hollow-cube/api-server/internal/mapdb"
+	"github.com/hollow-cube/api-server/internal/pkg/natsutil"
 	"github.com/hollow-cube/api-server/internal/pkg/notification"
 	"github.com/hollow-cube/api-server/internal/playerdb"
 	"go.uber.org/fx"
@@ -17,6 +18,7 @@ type ServerParams struct {
 	PlayerStore  *playerdb.Store
 	MapStore     *mapdb.Store
 	SessionStore *sessiondb.Queries
+	JetStream    *natsutil.JetStreamWrapper
 
 	Notifications notification.Manager
 	Interactions  *interaction.Handler
@@ -28,6 +30,7 @@ type Server struct {
 	playerStore  *playerdb.Store
 	mapStore     *mapdb.Store
 	sessionStore *sessiondb.Queries
+	jetStream    *natsutil.JetStreamWrapper
 
 	notifications notification.Manager
 	interactions  *interaction.Handler
@@ -39,6 +42,7 @@ func NewServer(p ServerParams) (*Server, error) {
 		playerStore:   p.PlayerStore,
 		mapStore:      p.MapStore,
 		sessionStore:  p.SessionStore,
+		jetStream:     p.JetStream,
 		notifications: p.Notifications,
 		interactions:  p.Interactions,
 	}

api/v4Internal/server_maps.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/hollow-cube/api-server/internal/mapdb"
 	"github.com/hollow-cube/api-server/internal/pkg/notification"
+	"github.com/hollow-cube/api-server/internal/pkg/player"
 	"github.com/hollow-cube/api-server/internal/playerdb"
 	"github.com/hollow-cube/api-server/pkg/ox"
 )
@@ -127,6 +128,15 @@ func (s *Server) RemoveMapBuilder(ctx context.Context, request MapPlayerRequest)
 		return fmt.Errorf("failed to claw back invite notifications: %w", err)
 	}
 
+	// Transfer the player back to the hub in case they are currently in the map.
+	// TODO: we should include a message here with the same message logic extracted from interactions
+	err = s.jetStream.PublishJSONAsync(ctx, player.TransferMessage{
+		PlayerId: request.PlayerID,
+		From:     &m.ID,
+		To:       "hub",
+		State:    "playing",
+	})
+
 	return nil
 }
 

internal/pkg/model/map.go

@@ -149,6 +149,8 @@ func (a MapUpdateAction) String() string {
 type MapUpdateMessage struct {
 	Action MapUpdateAction `json:"action"`
 	ID     string          `json:"id"`
+
+	DrainReason *string `json:"drainReason"`
 }
 
 func (m MapUpdateMessage) Subject() string {

internal/pkg/player/model.go

@@ -0,0 +1,13 @@
+package player
+
+type TransferMessage struct {
+	PlayerId string `json:"playerId"`
+
+	From  *string `json:"from"`  // A map id, 'hub', or nil for anywhere
+	To    string  `json:"to"`    // A map id, or 'hub'
+	State string  `json:"state"` // playing/building/verifying/etc. Must be playing for hub.
+}
+
+func (t TransferMessage) Subject() string {
+	return "player.transfer"
+}

internal/pkg/player/tracker.go

@@ -66,6 +66,19 @@ func NewTracker(p TrackerParams) (*Tracker, error) {
 		return nil, err
 	}
 
+	err = p.JetStream.UpsertStream(context.Background(), jetstream.StreamConfig{
+		Name:       "PLAYER_MANAGEMENT",
+		Subjects:   []string{"player.>"},
+		Retention:  jetstream.LimitsPolicy,
+		Storage:    jetstream.FileStorage,
+		MaxAge:     10 * time.Minute,
+		Duplicates: 60 * time.Second,
+	})
+	if err != nil {
+		reportCancel()
+		return nil, err
+	}
+
 	return &Tracker{
 		queries:           p.Queries,
 		jetStream:         p.JetStream,