package main import ( "context" "encoding/json" "fmt" "log" "net/http" "strconv" "time" "github.com/jackc/pgx/v4" ) var metricsText string func enableCors(w *http.ResponseWriter) { (*w).Header().Set("Access-Control-Allow-Origin", "*") } func runMetrics() { hashtagtotal := runTrendingHashtags() wordstotal := runTrendingWords() totalJson := make(map[string]interface{}) totalJson["hashtags"] = hashtagtotal totalJson["words"] = wordstotal totalJson["datetime"] = time.Now().UTC() data, err := json.Marshal(totalJson) if err != nil { log.Fatalf("error marshaling combined activity: %v\n", err) } metricsText = string(data) } func runTrendingHashtags() map[string]interface{} { sql := `SELECT UNNEST(activities.hashtags) as hashtags, count(actors.id) from activities LEFT JOIN actors ON activities.document->>'attributedTo'=actors.document->>'id' WHERE actors.bot=false AND activities.identifiedat > LOCALTIMESTAMP - INTERVAL '30 MINUTES' GROUP BY hashtags ORDER BY count DESC LIMIT 20;` rows, err := pool.Query(context.Background(), sql) if err != nil { panic(err) } hashtagitems := make([]interface{}, 0); hashcount := 0 for rows.Next() { var hashtag string var count int err = rows.Scan(&hashtag, &count) if err != nil { panic(err) } hashtagitem := make(map[string]interface{}) hashtagitem["hashtag"] = hashtag hashtagitem["count"] = count hashtagitems = append(hashtagitems, hashtagitem) hashcount = hashcount + 1 } rows.Close() hashtagtotal := make(map[string]interface{}); hashtagtotal["count"] = hashcount hashtagtotal["items"] = hashtagitems return hashtagtotal } func runTrendingWords() map[string]interface{} { sql := `WITH popular_words AS ( select word FROM ts_stat( ' SELECT to_tsvector(''simple'', normalized) FROM activities LEFT JOIN actors ON activities.document->>''attributedTo''=actors.document->>''id'' WHERE activities.identifiedat > current_timestamp - interval ''60 minutes'' AND actors.bot=false ' ) WHERE length(word) > 3 AND NOT word in (SELECT word FROM stopwords) ORDER BY ndoc DESC LIMIT 100) SELECT concat_ws(' ', a1.word, a2.word) phrase, count(*) FROM popular_words AS a1 CROSS JOIN popular_words AS a2 CROSS JOIN activities WHERE normalized ilike format('%%%s %s%%', a1.word, a2.word) AND identifiedat > current_timestamp - interval '60 minutes' GROUP BY 1 HAVING count(*) > 1 ORDER BY 2 DESC LIMIT 20;` rows, err := pool.Query(context.Background(), sql) if err != nil { panic(err) } trendingitems := make([]interface{}, 0); trendingcount := 0 for rows.Next() { var trendingword string var count int err = rows.Scan(&trendingword, &count) if err != nil { panic(err) } trendingitem := make(map[string]interface{}) trendingitem["trending"] = trendingword trendingitem["count"] = count trendingitems = append(trendingitems, trendingitem) trendingcount = trendingcount + 1 } rows.Close() trendingwordtotal := make(map[string]interface{}); trendingwordtotal["count"] = trendingcount trendingwordtotal["items"] = trendingitems return trendingwordtotal } // GET handlers func getTrending(w http.ResponseWriter, r *http.Request) { enableCors(&w) fmt.Fprintf(w, "%s", metricsText) } func getSearch(w http.ResponseWriter, r *http.Request) { enableCors(&w) searchkeys, exists_search := r.URL.Query()["s"] offsetkeys, exists_offset := r.URL.Query()["o"] var err error var rows pgx.Rows var searchKey string if exists_search { searchKey = searchkeys[0] } var offsetKey int if exists_offset { offsetKey, _ = strconv.Atoi(offsetkeys[0]) } else { offsetKey = -1 } if exists_search && searchKey != "" { if offsetKey == -1 { rows, err = pool.Query(context.Background(), "SELECT activities.id, activities.document, actors.document FROM activities as activities INNER JOIN actors as actors ON activities.document->>'actor' = actors.document->>'id' WHERE activities.normalized_tsvector @@ plainto_tsquery($1) ORDER BY activities.id DESC LIMIT 10", searchKey) } else { rows, err = pool.Query(context.Background(), "SELECT activities.id, activities.document, actors.document FROM activities as activities INNER JOIN actors as actors ON activities.document->>'actor' = actors.document->>'id' WHERE activities.normalized_tsvector @@ plainto_tsquery($1) AND activities.id < $2 ORDER BY activities.id DESC LIMIT 10", searchKey, offsetKey) } } else { if offsetKey == -1 { rows, err = pool.Query(context.Background(), "SELECT activities.id, activities.document, actors.document FROM activities as activities INNER JOIN actors as actors ON activities.document->>'actor' = actors.document->>'id' ORDER BY activities.id DESC LIMIT 10") } else { rows, err = pool.Query(context.Background(), "SELECT activities.id, activities.document, actors.document FROM activities as activities INNER JOIN actors as actors ON activities.document->>'actor' = actors.document->>'id' AND activities.id < $1 ORDER BY activities.id DESC LIMIT 10", offsetKey) } } if err != nil { panic(err) } defer rows.Close() var earliestid int earliestid = 0 var activitiesJson []map[string]json.RawMessage for rows.Next() { var id int var activityRaw string var actorRaw string var activityJson map[string]json.RawMessage err = rows.Scan(&id, &activityRaw, &actorRaw) if err != nil { panic(err) } err := json.Unmarshal([]byte(activityRaw), &activityJson) if err != nil { fmt.Println(err) } if earliestid == 0 { earliestid = id } else if earliestid > id { earliestid = id } activityJson["actor"] = json.RawMessage(actorRaw) activitiesJson = append(activitiesJson, activityJson) } requestData := make(map[string]int) requestData["earliestid"] = earliestid totalJson := make(map[string]interface{}) totalJson["requestdata"] = requestData totalJson["activities"] = activitiesJson data, err := json.Marshal(totalJson) if err != nil { log.Fatalf("error marshaling combined activity: %v\n", err) } fmt.Fprintf(w, "%s", data) } func main() { pool = getDbPool() metricsText = "[]" go func() { for { runMetrics() time.Sleep(10 * time.Minute) } }() http.HandleFunc("/api/v1/search", getSearch) http.HandleFunc("/api/v1/trending", getTrending) log.Print("Starting HTTP inbox on port http://0.0.0.0:6431") log.Fatal(http.ListenAndServe("0.0.0.0:6431", nil)) }