package service import ( "testing" "time" "browser.local/platform/domain" ) func TestDurableStdoutProjectionCreatesUsersAndSuppressesRapidDuplicates(t *testing.T) { svc := newTestCoreService() plugin, endpoint := createPluginAndRunEndpoint(t, svc) plugin.RuntimeProfiles.ClientManagers = append(plugin.RuntimeProfiles.ClientManagers, domain.RuntimeClientManagerProfile{Key: "scum-client", Health: domain.RuntimeClientManagerHealth{RequiredCapabilities: []string{"game-client.bridge"}}}) plugin.GameClientBridge.LogProjections = []domain.GameClientBridgeLogProjectionDeclaration{{ Key: "player.login", StreamKeys: []string{"stdout"}, CorrelationFields: []string{"playerSlot"}, MaxInterveningLines: 4, Steps: []domain.GameClientBridgeLogProjectionStepDeclaration{ {Pattern: `Player "(?P[^"]+)" reported as player (?P[0-9]+)`}, {Pattern: `Player (?P[0-9]+) SteamID \(assumed\): (?P[0-9]+)`}, }, Target: domain.GameClientBridgeLogProjectionTargetDeclaration{ Collection: "scum_users", UpsertKeys: []string{"steamId"}, CaptureMappings: map[string]string{"steamId": "steamId", "displayName": "displayName", "playerSlot": "playerSlot"}, FixedValues: map[string]string{"online": "true", "source": "supervised-stdout"}, ObservedAtField: "lastLoginAt", }, Presence: &domain.GameClientBridgeLogProjectionPresenceDeclaration{ TimestampField: "lastLoginAt", ActiveWindowSeconds: 600, ActivityTarget: &domain.GameClientBridgeLogProjectionTargetDeclaration{ Collection: "scum_activity_events", UpsertKeys: []string{"steamId", "observedAt"}, CaptureMappings: map[string]string{"steamId": "steamId", "displayName": "displayName"}, FixedValues: map[string]string{"eventType": "login"}, ObservedAtField: "observedAt", }, }, }} if err := svc.store.GamePlugins().Update(plugin); err != nil { t.Fatalf("update plugin projection: %v", err) } instance, err := svc.CreateServerInstance(domain.ServerInstance{ID: "server-log-projection", PluginID: plugin.ID, RunEndpointID: endpoint.ID, Name: "SCUM projection", State: domain.ServerInstanceStateRunning}) if err != nil { t.Fatalf("create server: %v", err) } helloRequest := validRunControlHello() helloRequest.CapabilityReport.Fingerprint = "cap-log-projection" hello, err := svc.RegisterRunHello(helloRequest) if err != nil { t.Fatalf("register Run: %v", err) } stream, err := svc.CreateLogStream(domain.LogStream{ID: "log-projection", ServerInstanceID: instance.ID, Source: domain.LogStreamSourceProcess, StreamKey: "stdout", StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default"}) if err != nil { t.Fatalf("create stdout stream: %v", err) } base := time.Date(2026, 8, 18, 23, 25, 12, 0, time.UTC) ingestProjectionLines(t, svc, hello.SessionToken, endpoint.ID, instance.ID, stream.ID, 1, base, []string{ `LogBattlEye: Display: Player "love_fitting" reported as player 0`, `LogBattlEye: Display: Player #0 love_fitting (redacted) connected`, }) ingestProjectionLines(t, svc, hello.SessionToken, endpoint.ID, instance.ID, stream.ID, 3, base.Add(2*time.Second), []string{ `LogBattlEye: Display: Player 0 SteamID (assumed): 76561199510658111`, }) assertPresenceProjectionCounts(t, svc, plugin.ID, instance.ID, 1, 1, 0) ingestProjectionLines(t, svc, hello.SessionToken, endpoint.ID, instance.ID, stream.ID, 4, base.Add(5*time.Minute), []string{ `LogBattlEye: Display: Player "love_fitting" reported as player 0`, `LogBattlEye: Display: Player 0 SteamID (assumed): 76561199510658111`, }) assertPresenceProjectionCounts(t, svc, plugin.ID, instance.ID, 1, 1, 0) ingestProjectionLines(t, svc, hello.SessionToken, endpoint.ID, instance.ID, stream.ID, 6, base.Add(11*time.Minute), []string{ `LogBattlEye: Display: Player "love_fitting" reported as player 0`, `LogBattlEye: Display: Player 0 SteamID (assumed): 76561199510658111`, }) assertPresenceProjectionCounts(t, svc, plugin.ID, instance.ID, 1, 2, 0) } func ingestProjectionLines(t *testing.T, svc *CoreService, sessionToken, endpointID, serverID, streamID string, firstSeq uint64, observedAt time.Time, lines []string) { t.Helper() entries := make([]domain.LogEntry, len(lines)) for index, line := range lines { entries[index] = domain.LogEntry{Seq: firstSeq + uint64(index), Timestamp: observedAt.Add(time.Duration(index) * time.Second), Level: "display", Line: line} } lastSeq := firstSeq + uint64(len(entries)) - 1 batch := domain.LogBatchIngest{RunEndpointID: endpointID, SessionToken: sessionToken, LogStreamID: streamID, ServerInstanceID: serverID, StreamKey: "stdout", Source: domain.LogStreamSourceProcess, FirstSeq: firstSeq, LastSeq: lastSeq, Compression: "none", Checksum: checksumForEntries(t, entries), Entries: entries} if result, err := svc.IngestLogBatch(batch); err != nil || !result.Accepted { t.Fatalf("ingest projection lines result=%+v err=%v", result, err) } } func assertPresenceProjectionCounts(t *testing.T, svc *CoreService, pluginID, serverID string, users, activities, commands int) { t.Helper() userRows, err := svc.store.PluginDataRecords().List(domain.PluginDataFilter{PluginID: pluginID, ServerInstanceID: serverID, Collection: "scum_users"}) if err != nil || len(userRows) != users { t.Fatalf("projected users=%+v err=%v", userRows, err) } activityRows, err := svc.store.PluginDataRecords().List(domain.PluginDataFilter{PluginID: pluginID, ServerInstanceID: serverID, Collection: "scum_activity_events"}) if err != nil || len(activityRows) != activities { t.Fatalf("projected activities=%+v err=%v", activityRows, err) } queued, err := svc.store.GameClientBridgeCommands().List(domain.GameClientBridgeCommandFilter{ServerInstanceID: serverID, PluginID: pluginID}) if err != nil || len(queued) != commands { t.Fatalf("presence bridge commands=%+v err=%v", queued, err) } }