Add SCUM trade catalog projections

2026-08-27 16:14:59 +08:00
parent 316efbe780
commit a0e7ae362b
16 changed files with 267 additions and 23 deletions
@@ -71,6 +71,46 @@ func TestDurableStdoutProjectionCreatesUsersAndSuppressesRapidDuplicates(t *test
 	assertPresenceProjectionCounts(t, svc, plugin.ID, instance.ID, 1, 2, 0)
 }

+func TestTradeLogProjectionsCreateCatalogAndTradeEvents(t *testing.T) {
+	svc := newTestCoreService()
+	plugin, endpoint := createPluginAndRunEndpoint(t, svc)
+	pattern := `^\d{4}\.\d{2}\.\d{2}-\d{2}\.\d{2}\.\d{2}: \[Trade\] Tradeable \((?P<itemCode>[A-Za-z0-9_.-]{1,128}) \(x(?P<quantity>\d{1,9})\)\) (?P<tradeVerb>purchased|sold) by .*?\((?P<steamId>\d{1,50})\) for (?P<price>-?\d{1,12})$`
+	plugin.GameClientBridge.LogProjections = []domain.GameClientBridgeLogProjectionDeclaration{{
+		Key: "scum.trade.catalog", StreamKeys: []string{"scum.trade"}, Steps: []domain.GameClientBridgeLogProjectionStepDeclaration{{Pattern: pattern}}, CorrelationFields: []string{"itemCode"}, MaxInterveningLines: 0,
+		Target: domain.GameClientBridgeLogProjectionTargetDeclaration{Collection: "scum_trade_goods", UpsertKeys: []string{"code"}, CaptureMappings: map[string]string{"code": "itemCode"}, FixedValues: map[string]string{"catalogType": "item", "source": "scum.trade"}, ObservedAtField: "lastSeenAt"},
+	}, {
+		Key: "scum.trade.events", StreamKeys: []string{"scum.trade"}, Steps: []domain.GameClientBridgeLogProjectionStepDeclaration{{Pattern: pattern}}, CorrelationFields: []string{"steamId", "itemCode", "tradeVerb"}, MaxInterveningLines: 0,
+		Target: domain.GameClientBridgeLogProjectionTargetDeclaration{Collection: "scum_trade_events", UpsertKeys: []string{"steamId", "itemCode", "tradeVerb", "quantity", "price", "observedAt"}, CaptureMappings: map[string]string{"steamId": "steamId", "itemCode": "itemCode", "tradeVerb": "tradeVerb", "quantity": "quantity", "price": "price"}, FixedValues: map[string]string{"eventType": "trade", "source": "scum.trade"}, ObservedAtField: "observedAt"},
+	}}
+	if err := svc.store.GamePlugins().Update(plugin); err != nil {
+		t.Fatalf("update trade projections: %v", err)
+	}
+	instance, err := svc.CreateServerInstance(domain.ServerInstance{ID: "server-trade-projection", PluginID: plugin.ID, RunEndpointID: endpoint.ID, Name: "SCUM trade projection", State: domain.ServerInstanceStateRunning})
+	if err != nil {
+		t.Fatalf("create server: %v", err)
+	}
+	helloRequest := validRunControlHello()
+	helloRequest.CapabilityReport.Fingerprint = "cap-trade-log-projection"
+	hello, err := svc.RegisterRunHello(helloRequest)
+	if err != nil {
+		t.Fatalf("register Run: %v", err)
+	}
+	stream, err := svc.CreateLogStream(domain.LogStream{ID: "trade-log-projection", ServerInstanceID: instance.ID, Source: domain.LogStreamSourceFile, StreamKey: "scum.trade", StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default"})
+	if err != nil {
+		t.Fatalf("create trade stream: %v", err)
+	}
+	base := time.Date(2026, 8, 27, 12, 34, 56, 0, time.UTC)
+	ingestTradeProjectionLines(t, svc, hello.SessionToken, endpoint.ID, instance.ID, stream.ID, 1, base, []string{`2026.08.27-12.34.56: [Trade] Tradeable (BPC_Apple (x2)) purchased by Mira(76561198000000001) for 120`})
+	goods, err := svc.store.PluginDataRecords().List(domain.PluginDataFilter{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: "scum_trade_goods"})
+	if err != nil || len(goods) != 1 || goods[0].Key != "BPC_Apple" || goods[0].Value["catalogType"] != "item" || goods[0].Value["lastSeenAt"] == nil {
+		t.Fatalf("trade item catalog was not projected: %+v err=%v", goods, err)
+	}
+	trades, err := svc.store.PluginDataRecords().List(domain.PluginDataFilter{PluginID: plugin.ID, ServerInstanceID: instance.ID, Collection: "scum_trade_events"})
+	if err != nil || len(trades) != 1 || trades[0].Value["itemCode"] != "BPC_Apple" || trades[0].Value["quantity"] != "2" || trades[0].Value["tradeVerb"] != "purchased" || trades[0].Value["steamId"] != "76561198000000001" || trades[0].Value["price"] != "120" {
+		t.Fatalf("trade event was not projected: %+v err=%v", trades, err)
+	}
+}
+
 func TestLifecycleProjectionMarksOnlineUsersOffline(t *testing.T) {
 	svc := newTestCoreService()
 	plugin, endpoint := createPluginAndRunEndpoint(t, svc)
@@ -167,6 +207,19 @@ func ingestProjectionLines(t *testing.T, svc *CoreService, sessionToken, endpoin
 	}
 }

+func ingestTradeProjectionLines(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: "info", Line: line}
+	}
+	lastSeq := firstSeq + uint64(len(entries)) - 1
+	batch := domain.LogBatchIngest{RunEndpointID: endpointID, SessionToken: sessionToken, LogStreamID: streamID, ServerInstanceID: serverID, StreamKey: "scum.trade", Source: domain.LogStreamSourceFile, 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 trade 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"})