// Worker crawls the steam API to get data from steam for a given user
// e.g account details and details of a user's friend
Worker crawls the steam API to get data from steam for a given user
e.g account details and details of a user's friend
// Worker crawls the steam API to get data from steam for a given user // e . g account details and details of a user ' s friend
func Worker(cntr controller.CntrInterface, job datastructures.Job) {
	startTime := time.Now().UnixNano() / int64(time.Millisecond)
	userWasFoundInDB, friendsList, err := GetFriends(cntr, job.CurrentTargetSteamID)
	if err != nil {
		configuration.Logger.Sugar().Panicf("error getting friends initially in worker for %s: %+v", job.CurrentTargetSteamID, err)
	}
	if userWasFoundInDB {
		crawlingStatus := common.CrawlingStatus{
			OriginalCrawlTarget: job.OriginalTargetSteamID,
			MaxLevel:            job.MaxLevel,
			CrawlID:             job.CrawlID,
			TotalUsersToCrawl:   len(friendsList),
		}
		success, err := cntr.SaveCrawlingStatsToDataStore(job.CurrentLevel, crawlingStatus)
		if err != nil {
			configuration.Logger.Sugar().Panicf("error saving crawling stats in worker : %+v", err)
		}
		if !success {
			configuration.Logger.Sugar().Panicf("failed to save crawling stats in worker: %+v", err)
		}
		friendsShoudlBeCrawled := util.JobIsNotLevelOneAndNotMax(job)
		if friendsShoudlBeCrawled {
			err = putFriendsIntoQueue(cntr, job, friendsList)
			if err != nil {
				configuration.Logger.Sugar().Panicf("error publishing friends from steamID: %s to queue: %+v", job.CurrentTargetSteamID, err)
			}
		}
		writeAPI := configuration.InfluxDBClient.WriteAPI(os.Getenv("ORG"), "crawlerMetrics")
		point := influxdb2.NewPointWithMeasurement("crawlerMetrics").
			AddTag("service", "crawler").
			AddTag("fromdatastore", "yes").
			AddField("totalTime", commonUtil.GetCurrentTimeInMs()-startTime).
			AddField("totalfriends", len(friendsList)).
			SetTime(time.Now())
		writeAPI.WritePoint(point)
		defer writeAPI.Close()
		return
	}
	playerSummaryForCurrentUser := common.Player{}
	fiftyOrFewerGamesOwnedForCurrentUser := []common.GameOwnedDocument{}
	friendPlayerSummaries := []common.Player{}
	var waitG sync.WaitGroup
	durationForGetSummaryForMainUser := int64(0)
	durationForGetFiftyOrFewerGamesOwned := int64(0)
	durationForGetSummariesForFriends := int64(0)
	waitG.Add(1)
	go getSummaryForMainUserFunc(
		cntr,
		job.CurrentTargetSteamID,
		&playerSummaryForCurrentUser,
		&durationForGetSummaryForMainUser,
		&waitG)
	waitG.Add(1)
	go getFiftyOrFewerGamesOwnedFunc(
		cntr,
		job.CurrentTargetSteamID,
		&fiftyOrFewerGamesOwnedForCurrentUser,
		&durationForGetFiftyOrFewerGamesOwned,
		&waitG)
	waitG.Add(1)
	go getSummariesForFriendsFunc(
		cntr,
		friendsList,
		&friendPlayerSummaries,
		&durationForGetSummariesForFriends,
		&waitG)
	waitG.Wait()
	emptyPlayer := common.Player{}
	if playerSummaryForCurrentUser == emptyPlayer {
		configuration.Logger.Sugar().Infof("caught ultra secure private user, ignoring this job: %+v", job)
		return
	}
	privateFriendCount := len(friendsList) - len(friendPlayerSummaries)
	publicFriendCount := len(friendsList) - privateFriendCount
	logMsg := fmt.Sprintf("Got data for [%s][%s][%s][%d public %d private friends][%d games]",
		playerSummaryForCurrentUser.Steamid, playerSummaryForCurrentUser.Personaname, playerSummaryForCurrentUser.Loccountrycode,
		publicFriendCount, privateFriendCount, len(fiftyOrFewerGamesOwnedForCurrentUser))
	configuration.Logger.Info(logMsg)
	friendPlayerSummarySteamIDs := getSteamIDsFromPlayers(friendPlayerSummaries)
	friendsShoudlBeCrawled := util.JobIsNotLevelOneAndNotMax(job)
	publishFriendsToQueueDuration := int64(0)
	if friendsShoudlBeCrawled {
		waitG.Add(1)
		go publishFriendsToQueueFunc(cntr, job, friendPlayerSummarySteamIDs, &publishFriendsToQueueDuration, &waitG)
	}
	saveUser := dtos.SaveUserDTO{
		OriginalCrawlTarget: job.OriginalTargetSteamID,
		CurrentLevel:        job.CurrentLevel,
		MaxLevel:            job.MaxLevel,
		CrawlID:             job.CrawlID,
		User: common.UserDocument{
			AccDetails: common.AccDetailsDocument{
				SteamID:        playerSummaryForCurrentUser.Steamid,
				Personaname:    playerSummaryForCurrentUser.Personaname,
				Profileurl:     playerSummaryForCurrentUser.Profileurl,
				Avatar:         playerSummaryForCurrentUser.Avatar,
				Timecreated:    playerSummaryForCurrentUser.Timecreated,
				Loccountrycode: playerSummaryForCurrentUser.Loccountrycode,
			},
			FriendIDs:  friendPlayerSummarySteamIDs,
			GamesOwned: fiftyOrFewerGamesOwnedForCurrentUser,
		},
	}
	waitG.Add(1)
	saveUserDuration := int64(0)
	go saveUserFunc(cntr, saveUser, &saveUserDuration, &waitG)
	waitG.Wait()
	writeAPI := configuration.InfluxDBClient.WriteAPI(os.Getenv("ORG"), "crawlerMetrics")
	point := influxdb2.NewPointWithMeasurement("crawlerMetrics").
		AddTag("fromdatastore", "no").
		AddField("totalfriends", publicFriendCount).
		AddField("totalTime", commonUtil.GetCurrentTimeInMs()-startTime).
		AddField("getplayersummaryduration", durationForGetSummaryForMainUser).
		AddField("getgamesownedduration", durationForGetFiftyOrFewerGamesOwned).
		AddField("getfriendsplayersummariesduration", durationForGetSummariesForFriends).
		AddField("publishfriendstoqueueduration", publishFriendsToQueueDuration).
		AddField("saveuserduration", saveUserDuration).
		AddField("gamesowned", len(fiftyOrFewerGamesOwnedForCurrentUser)).
		SetTime(time.Now())
	writeAPI.WritePoint(point)
	defer writeAPI.Close()
}
func Worker ( cntr controller . CntrInterface , job datastructures . Job ) { startTime := time . Now ( ) . UnixNano ( ) / int64 ( time . Millisecond ) userWasFoundInDB , friendsList , err := GetFriends ( cntr , job . CurrentTargetSteamID ) if err != nil { configuration . Logger . Sugar ( ) . Panicf ( " " , job . CurrentTargetSteamID , err ) } if userWasFoundInDB { crawlingStatus := common . CrawlingStatus { OriginalCrawlTarget : job . OriginalTargetSteamID , MaxLevel : job . MaxLevel , CrawlID : job . CrawlID , TotalUsersToCrawl : len ( friendsList ) , } success , err := cntr . SaveCrawlingStatsToDataStore ( job . CurrentLevel , crawlingStatus ) if err != nil { configuration . Logger . Sugar ( ) . Panicf ( " " , err ) } if ! success { configuration . Logger . Sugar ( ) . Panicf ( " " , err ) } friendsShoudlBeCrawled := util . JobIsNotLevelOneAndNotMax ( job ) if friendsShoudlBeCrawled { err = putFriendsIntoQueue ( cntr , job , friendsList ) if err != nil { configuration . Logger . Sugar ( ) . Panicf ( " " , job . CurrentTargetSteamID , err ) } } writeAPI := configuration . InfluxDBClient . WriteAPI ( os . Getenv ( " " ) , " " ) point := influxdb2 . NewPointWithMeasurement ( " " ) . AddTag ( " " , " " ) . AddTag ( " " , " " ) . AddField ( " " , commonUtil . GetCurrentTimeInMs ( ) - startTime ) . AddField ( " " , len ( friendsList ) ) . SetTime ( time . Now ( ) ) writeAPI . WritePoint ( point ) defer writeAPI . Close ( ) return } playerSummaryForCurrentUser := common . Player { } fiftyOrFewerGamesOwnedForCurrentUser := [ ] common . GameOwnedDocument { } friendPlayerSummaries := [ ] common . Player { } var waitG sync . WaitGroup durationForGetSummaryForMainUser := int64 ( 0 ) durationForGetFiftyOrFewerGamesOwned := int64 ( 0 ) durationForGetSummariesForFriends := int64 ( 0 ) waitG . Add ( 1 ) go getSummaryForMainUserFunc ( cntr , job . CurrentTargetSteamID , & playerSummaryForCurrentUser , & durationForGetSummaryForMainUser , & waitG ) waitG . Add ( 1 ) go getFiftyOrFewerGamesOwnedFunc ( cntr , job . CurrentTargetSteamID , & fiftyOrFewerGamesOwnedForCurrentUser , & durationForGetFiftyOrFewerGamesOwned , & waitG ) waitG . Add ( 1 ) go getSummariesForFriendsFunc ( cntr , friendsList , & friendPlayerSummaries , & durationForGetSummariesForFriends , & waitG ) waitG . Wait ( ) emptyPlayer := common . Player { } if playerSummaryForCurrentUser == emptyPlayer { configuration . Logger . Sugar ( ) . Infof ( " " , job ) return } privateFriendCount := len ( friendsList ) - len ( friendPlayerSummaries ) publicFriendCount := len ( friendsList ) - privateFriendCount logMsg := fmt . Sprintf ( " " , playerSummaryForCurrentUser . Steamid , playerSummaryForCurrentUser . Personaname , playerSummaryForCurrentUser . Loccountrycode , publicFriendCount , privateFriendCount , len ( fiftyOrFewerGamesOwnedForCurrentUser ) ) configuration . Logger . Info ( logMsg ) friendPlayerSummarySteamIDs := getSteamIDsFromPlayers ( friendPlayerSummaries ) friendsShoudlBeCrawled := util . JobIsNotLevelOneAndNotMax ( job ) publishFriendsToQueueDuration := int64 ( 0 ) if friendsShoudlBeCrawled { waitG . Add ( 1 ) go publishFriendsToQueueFunc ( cntr , job , friendPlayerSummarySteamIDs , & publishFriendsToQueueDuration , & waitG ) } saveUser := dtos . SaveUserDTO { OriginalCrawlTarget : job . OriginalTargetSteamID , CurrentLevel : job . CurrentLevel , MaxLevel : job . MaxLevel , CrawlID : job . CrawlID , User : common . UserDocument { AccDetails : common . AccDetailsDocument { SteamID : playerSummaryForCurrentUser . Steamid , Personaname : playerSummaryForCurrentUser . Personaname , Profileurl : playerSummaryForCurrentUser . Profileurl , Avatar : playerSummaryForCurrentUser . Avatar , Timecreated : playerSummaryForCurrentUser . Timecreated , Loccountrycode : playerSummaryForCurrentUser . Loccountrycode , } , FriendIDs : friendPlayerSummarySteamIDs , GamesOwned : fiftyOrFewerGamesOwnedForCurrentUser , } , } waitG . Add ( 1 ) saveUserDuration := int64 ( 0 ) go saveUserFunc ( cntr , saveUser , & saveUserDuration , & waitG ) waitG . Wait ( ) writeAPI := configuration . InfluxDBClient . WriteAPI ( os . Getenv ( " " ) , " " ) point := influxdb2 . NewPointWithMeasurement ( " " ) . AddTag ( " " , " " ) . AddField ( " " , publicFriendCount ) . AddField ( " " , commonUtil . GetCurrentTimeInMs ( ) - startTime ) . AddField ( " " , durationForGetSummaryForMainUser ) . AddField ( " " , durationForGetFiftyOrFewerGamesOwned ) . AddField ( " " , durationForGetSummariesForFriends ) . AddField ( " " , publishFriendsToQueueDuration ) . AddField ( " " , saveUserDuration ) . AddField ( " " , len ( fiftyOrFewerGamesOwnedForCurrentUser ) ) . SetTime ( time . Now ( ) ) writeAPI . WritePoint ( point ) defer writeAPI . Close ( ) }
Worker crawls the steam API to get data from steam for a given user
e.g account details and details of a user's friend
Worker crawls the steam API to get data from steam for a given user e . g account details and details of a user ' s friend
// If the job is not at max level or has a max level of one, add // friends to the queue for crawling // ASYNC BLOCK TWO // PUT FRIENDS INTO QUEUE // // Save user to DB