fedilogue/poll/main.go

package main

import (
	"github.com/jackc/pgx/pgxpool"
	_ "net/http/pprof"
	"net/http"
	"context"
	"time"
	"fmt"
	"net"
	"log"
	"os"
)

func AppendIfMissing(hay []string, needle string) []string {
	for _, ele := range hay {
		if ele == needle {
			return hay
		}
	}
	return append(hay, needle)
}

func writePost(pool *pgxpool.Pool, reportpost ReportPost) {
	conn, err := pool.Acquire(context.Background())
	if err != nil {
		log.Fatal("Error connecting to database:", err)
		os.Exit(1)
	}
	defer conn.Release()

	// Insert new account if new
	var accountid int
	err = conn.QueryRow(context.Background(), "INSERT INTO accounts (acct, avatar, bot, created_at, display_name, url) VALUES($1, $2, $3, $4, $5, $6) ON CONFLICT(acct) DO UPDATE SET acct=EXCLUDED.acct RETURNING id", reportpost.Account.Acct, reportpost.Account.Avatar, reportpost.Account.Bot, reportpost.Account.Created_at, reportpost.Account.Display_name, reportpost.Account.Url).Scan(&accountid)
	if err != nil {
		fmt.Println("First ", err)
		fmt.Println("--------------------------")
		fmt.Println("Reported error: ", err)
		fmt.Println("Account Acct: ", reportpost.Account.Acct)
		fmt.Println("Account Avatar: ", reportpost.Account.Avatar)
		fmt.Println("Account Avatar len: ", len(reportpost.Account.Avatar))
		fmt.Println("Account Bot: ", reportpost.Account.Bot)
		fmt.Println("Account Created_at: ", reportpost.Account.Created_at)
		fmt.Println("Account Display: ", reportpost.Account.Display_name)
		fmt.Println("Account URL: ", reportpost.Account.Url)
		fmt.Println(reportpost)
		fmt.Println("--------------------------")
		os.Exit(1) // For now I want this to die and learn why it failed
		return
	}

	// Insert new post if new
	_, err = conn.Exec(context.Background(), "INSERT INTO posts (url, content, created_at, normalized, account_id, posthash) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (posthash) DO NOTHING", reportpost.Url, reportpost.Content, reportpost.Created_at, reportpost.normalized, accountid, reportpost.posthash)
	if err != nil { // For now I want to know why this failed.
		fmt.Println("Second ", err)
		fmt.Println("--------------------------")
		fmt.Println("Reported error: ", err)
		fmt.Println("Url: ", reportpost.Url)
		fmt.Println("Content: ", reportpost.Content)
		fmt.Println("Created_at: ", reportpost.Created_at)
		fmt.Println("normalized: ", reportpost.normalized)
		fmt.Println("account_id", accountid)
		fmt.Println("posthash: ", reportpost.posthash)
		fmt.Println("--------------------------")
		os.Exit(1) // For now I want this to die and learn why it failed
		return
	}
}

func engine() {

	// Current instances
	runninginstances := make([]RunningInstance, 0)

	// Initial Setup
	reportPostChan := make(chan ReportPost, 2000)
	instanceReportChan := make(chan InstanceReport, 20)

	// Setup Database
	pool, err := pgxpool.Connect(context.Background(), "postgres://postgres@127.0.0.1/fedilogue")
	if err != nil {
		fmt.Fprintln(os.Stderr, "Unable to connect to database:", err)
		os.Exit(1)
	}

	l, err := net.Listen("tcp", "127.0.0.1:5555")
	if err != nil {
		fmt.Println(err)
		return
	}
	defer l.Close()

	x := 0

	commandClient := make(chan net.Conn)

	go func(l net.Listener) {
		for {
			c, err := l.Accept()
			if err != nil {
				fmt.Println("Error on accept")
				commandClient <- nil
				return
			}
			commandClient <- c
		}
	}(l)

	for {
		select {
		case c := <-commandClient:		// New client connection
			go handleClient(c, &runninginstances, instanceReportChan)
		case v := <-reportPostChan:		// New Post
			go writePost(pool, v)
		case w := <-instanceReportChan:		// Start or suspend instance
			if w.status == NEW_INSTANCE {
				fmt.Println("The instance: ", x, ", ", len(runninginstances))
				x = x + 1
				NewInstance(w.endpoint, &runninginstances, instanceReportChan, reportPostChan)
			} else if w.status == RUNNING || w.status == TOOMANYREQUESTS {
				for i, runninginstance := range runninginstances {
					if runninginstance.Endpoint == w.endpoint {
						runninginstances[i].Min_id = w.min_id
						runninginstances[i].Status = w.status
						runninginstances[i].LastRun = time.Now().Format("2006.01.02-15:04:05")
					}
				}
				go DeferPollRun(w, &runninginstances, instanceReportChan, reportPostChan)
			} else {
				SuspendInstance(w, &runninginstances)
			}
		}
	}
}

func main() {
	go engine()
	log.Println("serving on port 8080")
	log.Fatal(http.ListenAndServe(":8080", nil))
}