fedilogue/restapi/restapi.go

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))
}