animated-sniffle/app/worker.go

package main

import (
	"fmt"
	"github.com/gorilla/websocket"
	jsoniter "github.com/json-iterator/go"
	"net/http"
	"net/url"
	"strings"
	"time"
	"encoding/base64"
)

var uptime = time.Now().Unix()

type AuthResponse struct {
	Status    string `json:"status"`
	LocalData struct {
		DataKey string `json:"dataKey"`
	} `json:"localData"`
	UserData struct {
		Username    string `json:"username"`
		DisplayName string `json:"displayName"`
		Location    string `json:"location"`
		Chathost    string `json:"chathost"`
		IsRu        bool   `json:"isRu"`
	} `json:"userData"`
}

type ServerResponse struct {
	TS   int64               `json:"ts"`
	Type string              `json:"type"`
	Body jsoniter.RawMessage `json:"body"`
}

type DonateResponse struct {
	F struct {
		Username string `json:"username"`
	} `json:"f"`
	A int64 `json:"a"`
}

func mapRooms() {

	data := make(map[string]*Info)

	for {
		select {
		case m := <-rooms.Add:
			data[m.room] = &Info{Server: m.Server, Proxy: m.Proxy, Rid: m.Rid, Start: m.Start, Last: m.Last, Online: m.Online, Income: m.Income, Dons: m.Dons, Tips: m.Tips, ch: m.ch}

		case s := <-rooms.Json:
			j, err := json.Marshal(data)
			if err == nil {
				s = string(j)
			}
			rooms.Json <- s

		case <-rooms.Count:
			rooms.Count <- len(data)

		case key := <-rooms.Del:
			delete(data, key)

		case room := <-rooms.Check:
			if _, ok := data[room]; !ok {
				room = ""
			}
			rooms.Check <- room

		case room := <-rooms.Stop:
			if _, ok := data[room]; ok {
				close(data[room].ch)
			}
		}
	}
}

func announceCount() {
	for {
		time.Sleep(30 * time.Second)
		rooms.Count <- 0
		l := <-rooms.Count
		msg, err := json.Marshal(struct {
			Chanel string `json:"chanel"`
			Count  int    `json:"count"`
		}{
			Chanel: "bongacams",
			Count:  l,
		})
		if err == nil {
			socketServer <- msg
		}
	}
}

func reconnectRoom(workerData Info) {
	time.Sleep(5 * time.Second)
	fmt.Println("reconnect:", workerData.room, workerData.Proxy)
	startRoom(workerData)
}

func xWorker(workerData Info, u url.URL) {
	fmt.Println("Start", workerData.room, "server", workerData.Server, "proxy", workerData.Proxy)

	rooms.Add <- workerData

	defer func() {
		rooms.Del <- workerData.room
	}()

	b64, err := base64.StdEncoding.DecodeString(workerData.Params)
	if err != nil {
		fmt.Println(err, workerData.room)
		return
	}

	v := struct {
		Chathost    string `json:"chathost"`
		Username    string `json:"username"`
		DisplayName string `json:"displayName"`
		Location    string `json:"location"`
		IsRu        bool   `json:"isRu"`
		DataKey     string `json:"dataKey"`
	}{}
	
	if err := json.Unmarshal(b64, &v); err != nil {
		fmt.Println("exit: no amf parms", workerData.room, err)
		return
	}

	Dialer := *websocket.DefaultDialer

	if _, ok := conf.Proxy[workerData.Proxy]; ok {
		Dialer = websocket.Dialer{
			Proxy: http.ProxyURL(&url.URL{
				Scheme: "http", // or "https" depending on your proxy
				Host:   conf.Proxy[workerData.Proxy],
				Path:   "/",
			}),
			HandshakeTimeout: 45 * time.Second, // https://pkg.go.dev/github.com/gorilla/websocket
		}
	}

	c, _, err := Dialer.Dial(u.String(), nil)
	if err != nil {
		fmt.Println(err.Error(), workerData.room)
		return
	}

	defer c.Close()

	c.SetReadDeadline(time.Now().Add(60 * time.Second))

	if err = c.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"id":%d,"name":"joinRoom","args":["%s",{"username":"%s","displayName":"%s","location":"%s","chathost":"%s","isRu":%t,"isPerformer":false,"hasStream":false,"isLogged":false,"isPayable":false,"showType":"public"},"%s"]}`, 1, v.Chathost, v.Username, v.DisplayName, v.Location, v.Chathost, v.IsRu, v.DataKey))); err != nil {
		fmt.Println(err.Error())
		return
	}

	_, message, err := c.ReadMessage()
	if err != nil {
		fmt.Println(err.Error(), workerData.room)
		return
	}

	slog <- saveLog{workerData.Rid, time.Now().Unix(), string(message)}

	if string(message) == `{"id":1,"result":{"audioAvailable":false,"freeShow":false},"error":null}` {
		fmt.Println("room offline, exit", workerData.room)
		return
	}

	if err = c.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"id":%d,"name":"ChatModule.connect","args":["public-chat"]}`, 2))); err != nil {
		fmt.Println(err.Error(), workerData.room)
		return
	}

	_, message, err = c.ReadMessage()
	if err != nil {
		fmt.Println(err.Error(), workerData.room)
		return
	}

	slog <- saveLog{workerData.Rid, time.Now().Unix(), string(message)}

	quit := make(chan struct{})
	pid := 3

	defer func() {
		close(quit)
	}()

	go func() {
		ticker := time.NewTicker(30 * time.Second)
		defer ticker.Stop()
		for {
			select {
			case <-quit:
				return
			case <-ticker.C:
				if err = c.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"id":%d,"name":"ping"}`, pid))); err != nil {
					fmt.Println(err.Error(), workerData.room)
					rooms.Stop <- workerData.room
					return
				}
				pid++
				break
			}
		}
	}()

	dons := make(map[string]struct{})
	
	timeout := time.NewTicker(60 * 60 * 8 * time.Second)
	defer timeout.Stop()
	
	var income int64
	income = 0

	for {
		select {
		case <-timeout.C:
			fmt.Println("too_long exit:", workerData.room)
			return
		case <-workerData.ch:
			fmt.Println("Exit room:", workerData.room)
			return
		default:
		}

		c.SetReadDeadline(time.Now().Add(30 * time.Minute))
		_, message, err := c.ReadMessage()
		if err != nil {
			if income > 1 && websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
				go reconnectRoom(workerData)
			}
			return
		}

		now := time.Now().Unix()
		slog <- saveLog{workerData.Rid, now, string(message)}
		
		if now > workerData.Last+60*60 {
			fmt.Println("no_tips exit:", workerData.room)
			return
		}

		m := &ServerResponse{}
		if err = json.Unmarshal(message, m); err != nil {
			fmt.Println(err.Error(), workerData.room)
			continue
		}

		if m.Type == "ServerMessageEvent:PERFORMER_STATUS_CHANGE" && string(m.Body) == `"offline"` {
			fmt.Println(m.Type, workerData.room)
			return
		}

		if m.Type == "ServerMessageEvent:ROOM_CLOSE" {
			fmt.Println(m.Type, workerData.room)
			return
		}

		if m.Type == "ServerMessageEvent:INCOMING_TIP" {
			d := &DonateResponse{}
			if err = json.Unmarshal(m.Body, d); err == nil {

				workerData.Tips++
				if _, ok := dons[d.F.Username]; !ok {
					dons[d.F.Username] = struct{}{}
					workerData.Dons++
				}

				save <- saveData{workerData.room, strings.ToLower(d.F.Username), workerData.Rid, d.A, now}

				income += d.A
				
				workerData.Last = now
				workerData.Income += d.A
				rooms.Add <- workerData
				
				fmt.Println(d.F.Username, "send", d.A, "tokens to", workerData.room)
			}
		}
	}
}