package main import ( "encoding/json" "io/ioutil" "net/http" "time" "git.farhan.codes/farhan/fedilogue/shared" ) type ImageData struct { Type string `"json:type"` Url string `"json:url"` } type PublicKeyData struct { Id string `"json:id"` Owner string `"json:owner"` PublicKeyPem string `"json:publicKeyPem"` } type UserInfo struct { Id string `"json:id"` Type string `"json:type"` Following string `"json:following"` Followers string `"json:followers"` Inbox string `"json:inbox"` Outbox string `"json:outbox"` Featured string `"json:featured"` PreferredUsername string `"json:preferredUsername"` PublicKey PublicKeyData `"json:publicKeyPem"` Name string `"json:name"` Summary string `"json:summary"` Url string `"json:Url"` Icon ImageData `"json:icon"` Image ImageData `"json:image"` } type PostInfo struct { Id string `"json:id"` Type string `"json:type"` Published string `"json:published"` Url string `"json:Url"` Content string `"json:content"` } func PollMastodonPleroma(endpoint string, o *shared.RunningInstance) { newactivities := make([]shared.ReportActivity, 0) min_id := "" parsing_error := 0 use_auth := false var last_refresh int64 var client_id string var client_secret string var oauthData OAuth for { ri_mutex.Lock() m := runninginstances[endpoint] ri_mutex.Unlock() api_timeline := "https://" + endpoint + "/api/v1/timelines/public?limit=40&since_id=" + min_id req, err := http.NewRequest("GET", api_timeline, nil) req.Header.Set("User-Agent", "Tusky") if err != nil { logFatal("Unable to create new request for "+endpoint+": ", err) return } if use_auth == true { if time.Now().Unix() > last_refresh+oauthData.Expires_in { oauthData, err = oauth_refresh(endpoint, client_id, client_secret, oauthData.Refresh_token) if err != nil { logWarn("Unable to refresh oauth token for "+endpoint+": ", err) return } last_refresh = time.Now().Unix() } req.Header.Add("Authorization", oauthData.Access_token) } m.LastRun = time.Now().Format(time.RFC3339) resp, err := DoTries(o, req) if err != nil { m.Status = shared.CLIENT_ISSUE ri_mutex.Lock() runninginstances[endpoint] = m ri_mutex.Unlock() logWarn("Giving up on "+endpoint+": ", err.Error()) return } if resp.StatusCode == shared.TOOMANYREQUESTS { // Short Delay, 30 seconds logWarn("Delaying "+endpoint+", gave status ", resp.StatusCode, ", 1 hour delay") _, _ = ioutil.ReadAll(resp.Body) resp.Body.Close() // Release as soon as done m.Status = resp.StatusCode ri_mutex.Lock() runninginstances[endpoint] = m ri_mutex.Unlock() time.Sleep(time.Second * 30) continue } else if resp.StatusCode == shared.INTERNAL_ERROR { // Longer delay, 1 hour logWarn("Suspending "+endpoint+", gave status ", resp.StatusCode, ", 1 hour delay") _, _ = ioutil.ReadAll(resp.Body) resp.Body.Close() // Release as soon as done m.Status = 765 ri_mutex.Lock() runninginstances[endpoint] = m ri_mutex.Unlock() time.Sleep(time.Second * 3600) continue } else if resp.StatusCode != 200 { // Crash logErr("Terminating "+endpoint+", gave status ", resp.StatusCode) _, _ = ioutil.ReadAll(resp.Body) resp.Body.Close() // Release as soon as done m.Status = resp.StatusCode ri_mutex.Lock() runninginstances[endpoint] = m ri_mutex.Unlock() return } err = json.NewDecoder(resp.Body).Decode(&newactivities) resp.Body.Close() // Release as soon as done if err != nil { if parsing_error > 5 { m.Status = shared.BAD_RESPONSE ri_mutex.Lock() runninginstances[endpoint] = m ri_mutex.Unlock() logErr("Giving up on " + endpoint + " after 5 unmarshal errors.") return } parsing_error = parsing_error + 1 time.Sleep(time.Second * 30) } m.Status = shared.RUNNING ri_mutex.Lock() runninginstances[endpoint] = m ri_mutex.Unlock() for _, newactivity := range newactivities { go check_activity(newactivity.Uri) matchset := re.FindStringSubmatch(newactivity.Uri) if matchset != nil { newinstance := matchset[1] // Check min_id if newactivity.Id > min_id { min_id = newactivity.Id } go CheckInstance(newinstance, endpoint) } } time.Sleep(time.Second * 10) } }