Support for sending demos to parser service.

Makefile

@@ -68,7 +68,7 @@ test-go-cover:
 install_deps:
 	go install github.com/daixiang0/[email protected]
 	go install mvdan.cc/[email protected]
-	go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.60.3
+	go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.62.2
 	go install honnef.co/go/tools/cmd/[email protected]
 
 check: lint_golangci static lint_ts typecheck_ts

frontend/src/api/admin.ts

@@ -58,6 +58,7 @@ type Demos = {
     demo_cleanup_min_pct: string;
     demo_cleanup_mount: string;
     demo_count_limit: string;
+    demo_parser_url: string;
 };
 
 type Patreon = {

frontend/src/routes/_admin.admin.settings.tsx

@@ -676,7 +676,8 @@ const DemosSection = ({ tab, settings, mutate }: { tab: tabs; settings: Config;
             demo_cleanup_strategy: settings.demo.demo_cleanup_strategy,
             demo_cleanup_min_pct: settings.demo.demo_cleanup_min_pct,
             demo_cleanup_mount: settings.demo.demo_cleanup_mount,
-            demo_count_limit: settings.demo.demo_count_limit
+            demo_count_limit: settings.demo.demo_count_limit,
+            demo_parser_url: settings.demo.demo_parser_url
         }
     });
 
@@ -797,6 +798,22 @@ const DemosSection = ({ tab, settings, mutate }: { tab: tabs; settings: Config;
                         </SubHeading>
                     </Grid>
 
+                    <Grid xs={12}>
+                        <Field
+                            name={'demo_parser_url'}
+                            validators={{
+                                onChange: z.string()
+                            }}
+                            children={(props) => {
+                                return <TextFieldSimple {...props} label={'URL for demo parsing submissions'} />;
+                            }}
+                        />
+                        <SubHeading>
+                            This url should point to an instance of https://github.com/leighmacdonald/tf2_demostats.
+                            This is used to pull stats & player steamids out of demos that are fetched.
+                        </SubHeading>
+                    </Grid>
+
                     <Grid xs={12}>
                         <Subscribe
                             selector={(state) => [state.canSubmit, state.isSubmitting]}

gbans_example.yml

@@ -27,6 +27,9 @@ external_url: "https://example.com"
 #
 # client_timeout: 10
 
+# UDP Port to receive srcds log messages (logaddress_add)
+udp_port: 27715
+
 # Encryption key for JWT. https://numbergenerator.org/random-64-digit-hex-codes-generator is a good option.
 #
 # Minimum length of 10 characters

internal/cmd/serve.go

@@ -100,7 +100,7 @@ func firstTimeSetup(ctx context.Context, persons domain.PersonUsecase, news doma
 func createQueueWorkers(people domain.PersonUsecase, notifications domain.NotificationUsecase,
 	discordUC domain.DiscordUsecase, authRepo domain.AuthRepository, memberships *steamgroup.Memberships,
 	patreonUC domain.PatreonUsecase, bansSteam domain.BanSteamUsecase, bansNet domain.BanNetUsecase, bansASN domain.BanASNUsecase,
-	configUC domain.ConfigUsecase, fetcher *demo.Fetcher, demos domain.DemoUsecase, reports domain.ReportUsecase,
+	configUC domain.ConfigUsecase, demos domain.DemoUsecase, reports domain.ReportUsecase,
 	blocklists domain.BlocklistUsecase, discordOAuth domain.DiscordOAuthUsecase,
 ) *river.Workers {
 	workers := river.NewWorkers()
@@ -110,7 +110,6 @@ func createQueueWorkers(people domain.PersonUsecase, notifications domain.Notifi
 	river.AddWorker[steamgroup.MembershipArgs](workers, steamgroup.NewMembershipWorker(memberships))
 	river.AddWorker[patreon.AuthUpdateArgs](workers, patreon.NewSyncWorker(patreonUC))
 	river.AddWorker[ban.ExpirationArgs](workers, ban.NewExpirationWorker(bansSteam, bansNet, bansASN, people, notifications, configUC))
-	river.AddWorker[demo.FetcherArgs](workers, demo.NewFetcherWorker(fetcher, configUC))
 	river.AddWorker[demo.CleanupArgs](workers, demo.NewCleanupWorker(demos))
 	river.AddWorker[report.MetaInfoArgs](workers, report.NewMetaInfoWorker(reports))
 	river.AddWorker[blocklist.ListUpdaterArgs](workers, blocklist.NewListUpdaterWorker(blocklists))
@@ -150,13 +149,6 @@ func createPeriodicJobs() []*river.PeriodicJob {
 			},
 			&river.PeriodicJobOpts{RunOnStart: true}),
 
-		river.NewPeriodicJob(
-			river.PeriodicInterval(time.Minute*10),
-			func() (river.JobArgs, *river.InsertOpts) {
-				return demo.FetcherArgs{}, nil
-			},
-			&river.PeriodicJobOpts{RunOnStart: true}),
-
 		river.NewPeriodicJob(
 			river.PeriodicInterval(time.Hour*24),
 			func() (river.JobArgs, *river.InsertOpts) {
@@ -305,14 +297,14 @@ func serveCmd() *cobra.Command { //nolint:maintidx
 				return err
 			}
 
-			assetUsecase := asset.NewAssetUsecase(assetRepository)
-			serversUsecase := servers.NewServersUsecase(servers.NewServersRepository(dbConn))
-			demoUsecase := demo.NewDemoUsecase(domain.BucketDemo, demo.NewDemoRepository(dbConn), assetUsecase, configUsecase, serversUsecase)
+			assets := asset.NewAssetUsecase(assetRepository)
+			serversUC := servers.NewServersUsecase(servers.NewServersRepository(dbConn))
+			demos := demo.NewDemoUsecase(domain.BucketDemo, demo.NewDemoRepository(dbConn), assets, configUsecase, serversUC)
 
-			reportUsecase := report.NewReportUsecase(report.NewReportRepository(dbConn), notificationUsecase, configUsecase, personUsecase, demoUsecase)
+			reportUsecase := report.NewReportUsecase(report.NewReportRepository(dbConn), notificationUsecase, configUsecase, personUsecase, demos)
 
 			stateUsecase := state.NewStateUsecase(eventBroadcaster,
-				state.NewStateRepository(state.NewCollector(serversUsecase)), configUsecase, serversUsecase)
+				state.NewStateRepository(state.NewCollector(serversUC)), configUsecase, serversUC)
 
 			banUsecase := ban.NewBanSteamUsecase(ban.NewBanSteamRepository(dbConn, personUsecase, networkUsecase), personUsecase, configUsecase, notificationUsecase, reportUsecase, stateUsecase)
 
@@ -334,10 +326,10 @@ func serveCmd() *cobra.Command { //nolint:maintidx
 
 			appeals := appeal.NewAppealUsecase(appeal.NewAppealRepository(dbConn), banUsecase, personUsecase, notificationUsecase, configUsecase)
 
-			matchRepo := match.NewMatchRepository(eventBroadcaster, dbConn, personUsecase, serversUsecase, notificationUsecase, stateUsecase, weaponsMap)
+			matchRepo := match.NewMatchRepository(eventBroadcaster, dbConn, personUsecase, serversUC, notificationUsecase, stateUsecase, weaponsMap)
 			go matchRepo.Start(ctx)
 
-			matchUsecase := match.NewMatchUsecase(matchRepo, stateUsecase, serversUsecase, notificationUsecase)
+			matchUsecase := match.NewMatchUsecase(matchRepo, stateUsecase, serversUC, notificationUsecase)
 
 			if errWeapons := matchUsecase.LoadWeapons(ctx, weaponsMap); errWeapons != nil {
 				slog.Error("Failed to import weapons", log.ErrAttr(errWeapons))
@@ -360,12 +352,12 @@ func serveCmd() *cobra.Command { //nolint:maintidx
 
 			patreonUsecase := patreon.NewPatreonUsecase(patreon.NewPatreonRepository(dbConn), configUsecase)
 
-			srcdsUsecase := srcds.NewSrcdsUsecase(srcds.NewRepository(dbConn), configUsecase, serversUsecase, personUsecase, reportUsecase, notificationUsecase, banUsecase)
+			srcdsUsecase := srcds.NewSrcdsUsecase(srcds.NewRepository(dbConn), configUsecase, serversUC, personUsecase, reportUsecase, notificationUsecase, banUsecase)
 
 			wikiUsecase := wiki.NewWikiUsecase(wiki.NewWikiRepository(dbConn))
 
 			authRepo := auth.NewAuthRepository(dbConn)
-			authUsecase := auth.NewAuthUsecase(authRepo, configUsecase, personUsecase, banUsecase, serversUsecase)
+			authUsecase := auth.NewAuthUsecase(authRepo, configUsecase, personUsecase, banUsecase, serversUC)
 
 			voteUsecase := votes.NewVoteUsecase(votes.NewVoteRepository(dbConn), personUsecase, matchUsecase, notificationUsecase, configUsecase, eventBroadcaster)
 			go voteUsecase.Start(ctx)
@@ -393,7 +385,7 @@ func serveCmd() *cobra.Command { //nolint:maintidx
 			}
 
 			discordHandler := discord.NewDiscordHandler(discordUsecase, personUsecase, banUsecase,
-				stateUsecase, serversUsecase, configUsecase, networkUsecase, wordFilterUsecase, matchUsecase, banNetUsecase, banASNUsecase)
+				stateUsecase, serversUC, configUsecase, networkUsecase, wordFilterUsecase, matchUsecase, banNetUsecase, banASNUsecase)
 			discordHandler.Start(ctx)
 
 			appeal.NewAppealHandler(router, appeals, authUsecase)
@@ -406,21 +398,21 @@ func serveCmd() *cobra.Command { //nolint:maintidx
 			steamgroup.NewSteamgroupHandler(router, banGroupUsecase, authUsecase)
 			blocklist.NewBlocklistHandler(router, blocklistUsecase, networkUsecase, authUsecase)
 			chat.NewChatHandler(router, chatUsecase, authUsecase)
-			contest.NewContestHandler(router, contestUsecase, configUsecase, assetUsecase, authUsecase)
-			demo.NewDemoHandler(router, demoUsecase)
+			contest.NewContestHandler(router, contestUsecase, configUsecase, assets, authUsecase)
+			demo.NewDemoHandler(router, demos)
 			forum.NewForumHandler(router, forumUsecase, authUsecase)
-			match.NewMatchHandler(ctx, router, matchUsecase, serversUsecase, authUsecase, configUsecase)
-			asset.NewAssetHandler(router, configUsecase, assetUsecase, authUsecase)
+			match.NewMatchHandler(ctx, router, matchUsecase, serversUC, authUsecase, configUsecase)
+			asset.NewAssetHandler(router, configUsecase, assets, authUsecase)
 			metrics.NewMetricsHandler(router)
 			network.NewNetworkHandler(router, networkUsecase, authUsecase)
 			news.NewNewsHandler(router, newsUsecase, notificationUsecase, authUsecase)
 			notification.NewNotificationHandler(router, notificationUsecase, authUsecase)
 			patreon.NewPatreonHandler(router, patreonUsecase, authUsecase, configUsecase)
 			person.NewPersonHandler(router, configUsecase, personUsecase, authUsecase)
 			report.NewReportHandler(router, reportUsecase, authUsecase, notificationUsecase)
-			servers.NewServerHandler(router, serversUsecase, stateUsecase, authUsecase, personUsecase)
-			srcds.NewSRCDSHandler(router, srcdsUsecase, serversUsecase, personUsecase, assetUsecase,
-				reportUsecase, banUsecase, networkUsecase, banGroupUsecase, demoUsecase, authUsecase, banASNUsecase, banNetUsecase,
+			servers.NewServerHandler(router, serversUC, stateUsecase, authUsecase, personUsecase)
+			srcds.NewSRCDSHandler(router, srcdsUsecase, serversUC, personUsecase, assets,
+				reportUsecase, banUsecase, networkUsecase, banGroupUsecase, demos, authUsecase, banASNUsecase, banNetUsecase,
 				configUsecase, notificationUsecase, stateUsecase, blocklistUsecase)
 			votes.NewVoteHandler(router, voteUsecase, authUsecase)
 			wiki.NewWIkiHandler(router, wikiUsecase, authUsecase)
@@ -430,8 +422,6 @@ func serveCmd() *cobra.Command { //nolint:maintidx
 				go stateUsecase.LogAddressAdd(ctx, conf.Debug.AddRCONLogAddress)
 			}
 
-			demoFetcher := demo.NewFetcher(dbConn, configUsecase, serversUsecase, assetUsecase, demoUsecase)
-
 			// River Queue
 			workers := createQueueWorkers(
 				personUsecase,
@@ -444,8 +434,7 @@ func serveCmd() *cobra.Command { //nolint:maintidx
 				banNetUsecase,
 				banASNUsecase,
 				configUsecase,
-				demoFetcher,
-				demoUsecase,
+				demos,
 				reportUsecase,
 				blocklistUsecase,
 				discordOAuthUsecase)
@@ -468,6 +457,9 @@ func serveCmd() *cobra.Command { //nolint:maintidx
 
 			httpServer := httphelper.NewHTTPServer(conf.Addr(), router)
 
+			demoDownloader := demo.NewDownloader(configUsecase, dbConn, serversUC, assets, demos)
+			go demoDownloader.Start(ctx)
+
 			go func() {
 				<-ctx.Done()
 

internal/config/config_repository.go

@@ -35,7 +35,7 @@ func (c *configRepository) Read(ctx context.Context) (domain.Config, error) {
 
 		       filters_enabled, filters_dry, filters_ping_discord, filters_max_weight, filters_warning_timeout, filters_check_timeout, filters_match_timeout,
 
-		       demo_cleanup_enabled, demo_cleanup_strategy, demo_cleanup_min_pct, demo_cleanup_mount, demo_count_limit,
+		       demo_cleanup_enabled, demo_cleanup_strategy, demo_cleanup_min_pct, demo_cleanup_mount, demo_count_limit, demo_parser_url,
 
 		       patreon_enabled, patreon_client_id, patreon_client_secret, patreon_creator_access_token, patreon_creator_refresh_token, patreon_integrations_enabled,
 
@@ -69,7 +69,7 @@ func (c *configRepository) Read(ctx context.Context) (domain.Config, error) {
 			&cfg.General.DefaultRoute, &cfg.General.NewsEnabled, &cfg.General.ForumsEnabled, &cfg.General.ContestsEnabled, &cfg.General.WikiEnabled,
 			&cfg.General.StatsEnabled, &cfg.General.ServersEnabled, &cfg.General.ReportsEnabled, &cfg.General.ChatlogsEnabled, &cfg.General.DemosEnabled,
 			&cfg.Filters.Enabled, &cfg.Filters.Dry, &cfg.Filters.PingDiscord, &cfg.Filters.MaxWeight, &cfg.Filters.WarningTimeout, &cfg.Filters.CheckTimeout, &cfg.Filters.MatchTimeout,
-			&cfg.Demo.DemoCleanupEnabled, &cfg.Demo.DemoCleanupStrategy, &cfg.Demo.DemoCleanupMinPct, &cfg.Demo.DemoCleanupMount, &cfg.Demo.DemoCountLimit,
+			&cfg.Demo.DemoCleanupEnabled, &cfg.Demo.DemoCleanupStrategy, &cfg.Demo.DemoCleanupMinPct, &cfg.Demo.DemoCleanupMount, &cfg.Demo.DemoCountLimit, &cfg.Demo.DemoParserURL,
 			&cfg.Patreon.Enabled, &cfg.Patreon.ClientID, &cfg.Patreon.ClientSecret, &cfg.Patreon.CreatorAccessToken, &cfg.Patreon.CreatorRefreshToken, &cfg.Patreon.IntegrationsEnabled,
 			&cfg.Discord.Enabled, &cfg.Discord.AppID, &cfg.Discord.AppSecret, &cfg.Discord.LinkID, &cfg.Discord.Token, &cfg.Discord.GuildID, &cfg.Discord.LogChannelID,
 			&cfg.Discord.PublicLogChannelEnable, &cfg.Discord.PublicLogChannelID, &cfg.Discord.PublicMatchLogChannelID, &cfg.Discord.ModPingRoleID,
@@ -144,6 +144,7 @@ func (c *configRepository) Write(ctx context.Context, config domain.Config) erro
 			"demo_cleanup_min_pct":                config.Demo.DemoCleanupMinPct,
 			"demo_cleanup_mount":                  config.Demo.DemoCleanupMount,
 			"demo_count_limit":                    config.Demo.DemoCountLimit,
+			"demo_parser_url":                     config.Demo.DemoParserURL,
 			"patreon_enabled":                     config.Patreon.Enabled,
 			"patreon_integrations_enabled":        config.Patreon.IntegrationsEnabled,
 			"patreon_client_id":                   config.Patreon.ClientID,

internal/database/migrations/000102_add_demo_url.down.sql

@@ -0,0 +1,6 @@
+BEGIN;
+
+ALTER TABLE config
+    DROP COLUMN IF EXISTS demo_parser_url;
+
+COMMIT;

internal/database/migrations/000102_add_demo_url.up.sql

@@ -0,0 +1,5 @@
+BEGIN;
+
+ALTER TABLE config ADD COLUMN IF NOT EXISTS demo_parser_url text default 'http://localhost:8811/';
+
+COMMIT;

internal/demo/demo_usecase.go

@@ -1,18 +1,23 @@
 package demo
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"log/slog"
+	"mime/multipart"
+	"net/http"
+	"os"
 	"strings"
 	"time"
 
 	"github.com/dustin/go-humanize"
 	"github.com/gin-gonic/gin"
 	"github.com/leighmacdonald/gbans/internal/domain"
 	"github.com/leighmacdonald/gbans/internal/queue"
-	"github.com/leighmacdonald/gbans/pkg/demoparser"
 	"github.com/leighmacdonald/gbans/pkg/fs"
 	"github.com/leighmacdonald/gbans/pkg/log"
 	"github.com/ricochet2200/go-disk-usage/du"
@@ -202,6 +207,69 @@ func (d demoUsecase) GetDemos(ctx context.Context) ([]domain.DemoFile, error) {
 	return d.repository.GetDemos(ctx)
 }
 
+func (d demoUsecase) SendAndParseDemo(ctx context.Context, path string) (*domain.DemoDetails, error) {
+	fileHandle, errDF := os.Open(path)
+	if errDF != nil {
+		return nil, errors.Join(errDF, domain.ErrDemoLoad)
+	}
+
+	content, errContent := io.ReadAll(fileHandle)
+	if errContent != nil {
+		return nil, errors.Join(errDF, domain.ErrDemoLoad)
+	}
+
+	info, errInfo := fileHandle.Stat()
+	if errInfo != nil {
+		return nil, errors.Join(errInfo, domain.ErrDemoLoad)
+	}
+
+	log.Closer(fileHandle)
+
+	body := new(bytes.Buffer)
+	writer := multipart.NewWriter(body)
+
+	part, errCreate := writer.CreateFormFile("file", info.Name())
+	if errCreate != nil {
+		return nil, errors.Join(errCreate, domain.ErrDemoLoad)
+	}
+
+	if _, err := part.Write(content); err != nil {
+		return nil, errors.Join(errCreate, domain.ErrDemoLoad)
+	}
+
+	if errClose := writer.Close(); errClose != nil {
+		return nil, errors.Join(errClose, domain.ErrDemoLoad)
+	}
+
+	req, errReq := http.NewRequestWithContext(ctx, http.MethodPost, d.config.Config().Demo.DemoParserURL, body)
+	if errReq != nil {
+		return nil, errors.Join(errReq, domain.ErrDemoLoad)
+	}
+	req.Header.Set("Content-Type", writer.FormDataContentType())
+
+	client := &http.Client{}
+	resp, errSend := client.Do(req)
+	if errSend != nil {
+		return nil, errors.Join(errSend, domain.ErrDemoLoad)
+	}
+
+	defer resp.Body.Close()
+
+	var demo domain.DemoDetails
+
+	// TODO remove this extra copy once this feature doesnt have much need for debugging/inspection.
+	rawBody, errRead := io.ReadAll(resp.Body)
+	if errRead != nil {
+		return nil, errors.Join(errRead, domain.ErrDemoLoad)
+	}
+
+	if errDecode := json.NewDecoder(bytes.NewReader(rawBody)).Decode(&demo); errDecode != nil {
+		return nil, errors.Join(errDecode, domain.ErrDemoLoad)
+	}
+
+	return &demo, nil
+}
+
 func (d demoUsecase) CreateFromAsset(ctx context.Context, asset domain.Asset, serverID int) (*domain.DemoFile, error) {
 	_, errGetServer := d.servers.Server(ctx, serverID)
 	if errGetServer != nil {
@@ -221,20 +289,16 @@ func (d demoUsecase) CreateFromAsset(ctx context.Context, asset domain.Asset, se
 		mapName = nameParts[0]
 	}
 
+	// TODO change this data shape as we have not needed this in a long time. Only keys the are used.
 	intStats := map[string]gin.H{}
 
-	// temp thing until proper demo parsing is implemented
-	if d.config.Config().General.Mode != domain.TestMode {
-		var demoInfo demoparser.DemoInfo
-		if errParse := demoparser.Parse(ctx, asset.LocalPath, &demoInfo); errParse != nil {
-			return nil, errParse
-		}
+	demoDetail, errDetail := d.SendAndParseDemo(ctx, asset.LocalPath)
+	if errDetail != nil {
+		return nil, errDetail
+	}
 
-		for _, steamID := range demoInfo.SteamIDs() {
-			intStats[steamID.String()] = gin.H{}
-		}
-	} else {
-		intStats[d.config.Config().Owner] = gin.H{}
+	for key := range demoDetail.State.Users {
+		intStats[key] = gin.H{}
 	}
 
 	timeStr := fmt.Sprintf("%s-%s", namePartsAll[0], namePartsAll[1])