package main import ( "encoding/base64" "fmt" "github.com/gorilla/websocket" "net/http" "net/url" "strconv" "strings" "time" ) var uptime = time.Now().Unix() func mapRooms() { data := make(map[string]*Info) for { select { case m := <-rooms.Add: data[m.room] = &Info{Rid: m.Rid, Key: m.Key, Proxy: m.Proxy, 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: "camsoda", 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 getMessageID(s string) int { i := strings.IndexByte(s, ',') if i != -1 { if v, err := strconv.Atoi(s[1:i]); err == nil { return v } } return 0 } func getMessageData(s string) string { a := strings.IndexByte(s, '{') b := strings.LastIndexByte(s, '}') if a != -1 && b != -1 { return s[a : b+1] } return "" } func getWS(workerData Info, key []byte) string { 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 } } u := url.URL{Scheme: "wss", Host: "node2-ord.livemediahost.com:3000"} c, _, err := Dialer.Dial(u.String(), nil) if err != nil { fmt.Println(err.Error(), u.String(), workerData.room) return "" } defer c.Close() if err = c.WriteMessage(websocket.TextMessage, []byte(`["v3.authorize",{"token":"`+string(key)+`"}]`)); err != nil { fmt.Println(err.Error()) return "" } for { c.SetReadDeadline(time.Now().Add(45 * time.Second)) _, message, err := c.ReadMessage() if err != nil { fmt.Println(err.Error()) return "" } //fmt.Println(string(message)) messageID := getMessageID(string(message)) if messageID == 2 { if err = c.WriteMessage(websocket.TextMessage, []byte(`[5,{"room": "`+workerData.room+`"}]`)); err != nil { fmt.Println(err.Error()) return "" } //fmt.Println("send message!") continue } if messageID == 6 { input := struct { Url string `json:"url"` Room string `json:"room"` }{} if err := json.Unmarshal([]byte(getMessageData(string(message))), &input); err != nil { fmt.Println(err) } return input.Url } } return "" } func xWorker(workerData Info) { fmt.Println("Start", workerData.room, "proxy", workerData.Proxy) rooms.Add <- workerData defer func() { rooms.Del <- workerData.room }() if len(workerData.Key) < 50 { fmt.Println(workerData.Key, workerData.room) return } key, err := base64.StdEncoding.DecodeString(workerData.Key) if err != nil { fmt.Println(err, workerData.room) return } u, err := url.Parse(getWS(workerData, key)) if err != nil { fmt.Println(err, workerData.room) 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(), u.String(), workerData.room) return } defer c.Close() if err = c.WriteMessage(websocket.TextMessage, []byte(`["v3.authorize",{"token":"`+string(key)+`"}]`)); err != nil { fmt.Println(err.Error(), workerData.room) return } dons := make(map[string]struct{}) ticker := time.NewTicker(60 * 60 * 8 * time.Second) defer ticker.Stop() var income int64 income = 0 for { select { case <-ticker.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 { fmt.Println(err.Error(), workerData.room) 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 } messageID := getMessageID(string(message)) if messageID == 28 && strings.Contains(string(message), "offline") { fmt.Println("offline:", workerData.room) return } if messageID == 27 { input := struct { Value int64 `json:"value"` Username string `json:"subject_username"` Tip bool `json:"is_tip"` Stealth bool `json:"is_stealth"` }{} //[27,{"recent_tips":88,"value":1,"subject_username":"max4you","is_tip":true,"is_stealth":false},1,["max4you tipped 1 tokens"]] if err := json.Unmarshal([]byte(getMessageData(string(message))), &input); err != nil { //fmt.Println(string(message)) //fmt.Println(err) continue } if !input.Tip || input.Value < 1 { //fmt.Println(input.Tip, input.Value, "continue") continue } if input.Stealth || len(input.Username) < 3 { input.Username = "anon_tips" } if _, ok := dons[input.Username]; !ok { dons[input.Username] = struct{}{} workerData.Dons++ } save <- saveData{workerData.room, strings.ToLower(input.Username), workerData.Rid, input.Value, now} income += input.Value workerData.Tips++ workerData.Last = now workerData.Income += input.Value rooms.Add <- workerData fmt.Println(input.Username, "send", input.Value, "tokens to", workerData.room) } } }