commit b080cea2dfd292342acc6d70feda9c6994f6dffb
Author: poiuty <poiuty@protonmail.com>
Date:   Wed Jul 27 11:34:26 2022 +0300

    update

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..445e913
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,3 @@
+.DS_Store
+/app/chaturbate/chaturbate
+/app/bongacams/bongacams
diff --git a/LICENSE b/LICENSE
new file mode 100644
index 0000000..8531c11
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2020 poiuty
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
diff --git a/app/cmd.go b/app/cmd.go
new file mode 100644
index 0000000..0f2e895
--- /dev/null
+++ b/app/cmd.go
@@ -0,0 +1,118 @@
+package main
+
+import (
+	"fmt"
+	"net/http"
+	"net/url"
+	"os"
+	"runtime"
+	"strings"
+	"time"
+)
+
+var memInfo runtime.MemStats
+
+type Info struct {
+	ch     chan struct{}
+	room   string
+	Server string `json:"server"`
+	Proxy  string `json:"proxy"`
+	Online string `json:"online"`
+	Rid    int64  `json:"rid"`
+	Start  int64  `json:"start"`
+	Last   int64  `json:"last"`
+	Income int64  `json:"income"`
+	Dons   int64  `json:"dons"`
+	Tips   int64  `json:"tips"`
+}
+
+func updateFileRooms() string {
+	for {
+		rooms.Json <- ""
+		s := <-rooms.Json
+		err := os.WriteFile(conf.Conn["start"], []byte(s), 0644)
+		if err != nil {
+			fmt.Println(err)
+		}
+		time.Sleep(10 * time.Second)
+	}
+}
+
+func listHandler(w http.ResponseWriter, _ *http.Request) {
+	dat, err := os.ReadFile(conf.Conn["start"])
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+	fmt.Fprint(w, string(dat))
+}
+
+func debugHandler(w http.ResponseWriter, _ *http.Request) {
+	ws.Count <- 0
+	l := <-ws.Count
+	runtime.ReadMemStats(&memInfo)
+	j, err := json.Marshal(struct {
+		Goroutines int
+		WebSocket  int
+		Uptime     int64
+		Alloc      uint64
+		HeapSys    uint64
+	}{
+		Goroutines: runtime.NumGoroutine(),
+		Alloc:      memInfo.Alloc,
+		HeapSys:    memInfo.HeapSys,
+		Uptime:     uptime,
+		WebSocket:  l,
+	})
+	if err == nil {
+		fmt.Fprint(w, string(j))
+	}
+}
+
+func cmdHandler(w http.ResponseWriter, r *http.Request) {
+	if !conf.List[r.Header.Get("X-REAL-IP")] {
+		fmt.Fprint(w, "403")
+		return
+	}
+	params := r.URL.Query()
+	if len(params["room"]) > 0 && len(params["server"]) > 0 && len(params["proxy"]) > 0 {
+		now := time.Now().Unix()
+		workerData := Info{
+			room:   params["room"][0],
+			Server: params["server"][0],
+			Proxy:  params["proxy"][0],
+			Online: "0",
+			Start:  now,
+			Last:   now,
+			Rid:    0,
+			Income: 0,
+			Dons:   0,
+			Tips:   0,
+		}
+		startRoom(workerData)
+	}
+	if len(params["exit"]) > 0 {
+		rooms.Stop <- strings.Join(params["exit"], "")
+	}
+	fmt.Fprint(w, string("ok"))
+}
+
+func startRoom(workerData Info) {
+	rooms.Check <- workerData.room
+	testRoom := <-rooms.Check
+	if testRoom == workerData.room {
+		fmt.Println("Already track:", workerData.room)
+		return
+	}
+
+	rid, ok := getRoomInfo(workerData.room)
+	if !ok {
+		fmt.Println("No room in MySQL:", workerData.room)
+		return
+	}
+
+	workerData.Rid = rid
+	workerData.ch = make(chan struct{})
+
+	go xWorker(workerData, url.URL{Scheme: "wss", Host: workerData.Server + ".stream.highwebmedia.com", Path: "/ws/555/kmdqiune/websocket"})
+}
diff --git a/app/conf.go b/app/conf.go
new file mode 100644
index 0000000..b6e1ceb
--- /dev/null
+++ b/app/conf.go
@@ -0,0 +1,34 @@
+package main
+
+type Conf struct {
+	Conn  map[string]string
+	Proxy map[string]string
+	List  map[string]bool
+}
+
+var conf = &Conf{
+	Conn:  make(map[string]string),
+	Proxy: make(map[string]string),
+	List:  make(map[string]bool),
+}
+
+func startConfig() {
+	// DB
+	conf.Conn = map[string]string{
+		"mysql": "user:passwd@unix(/var/run/mysqld/mysqld.sock)/base?interpolateParams=true",
+		"click": "tcp://127.0.0.1:9000/base?compress=true&debug=false",
+		"start": "/tmp/fastStart.txt",
+	}
+
+	// 3proxy
+	conf.Proxy = map[string]string{
+		"us": "ip:port",
+		"fi": "ip:port",
+	}
+
+	// allow ips
+	conf.List = map[string]bool{
+		"::1":       true,
+		"127.0.0.1": true,
+	}
+}
diff --git a/app/echo.go b/app/echo.go
new file mode 100644
index 0000000..3bfddf1
--- /dev/null
+++ b/app/echo.go
@@ -0,0 +1,49 @@
+package main
+
+import (
+	"net/http"
+
+	"github.com/gorilla/websocket"
+)
+
+var (
+	wsClients = make(map[*websocket.Conn]struct{})
+
+	ws = struct {
+		Count chan int
+		Send  chan []byte
+		Add   chan *websocket.Conn
+	}{
+		Count: make(chan int, 100),
+		Send:  make(chan []byte, 100),
+		Add:   make(chan *websocket.Conn, 100),
+	}
+)
+
+func broadcast() {
+	for {
+		select {
+		case conn := <-ws.Add:
+			wsClients[conn] = struct{}{}
+
+		case <-ws.Count:
+			ws.Count <- len(wsClients)
+
+		case message := <-ws.Send:
+			for conn := range wsClients {
+				if err := conn.WriteMessage(1, message); err != nil {
+					conn.Close()
+					delete(wsClients, conn)
+				}
+			}
+		}
+	}
+}
+
+func wsHandler(w http.ResponseWriter, r *http.Request) {
+	conn, err := websocket.Upgrade(w, r, w.Header(), 1024, 1024)
+	if err != nil {
+		return
+	}
+	ws.Add <- conn
+}
diff --git a/app/main.go b/app/main.go
new file mode 100644
index 0000000..81e6d05
--- /dev/null
+++ b/app/main.go
@@ -0,0 +1,142 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"math/rand"
+	"net"
+	"net/http"
+	"os"
+	"time"
+
+	_ "github.com/ClickHouse/clickhouse-go"
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/jmoiron/sqlx"
+	jsoniter "github.com/json-iterator/go"
+)
+
+type Rooms struct {
+	Count chan int
+	Json  chan string
+	Check chan string
+	Stop  chan string
+	Del   chan string
+	Add   chan Info
+}
+
+var (
+	Mysql, Clickhouse *sqlx.DB
+
+	json = jsoniter.ConfigCompatibleWithStandardLibrary
+
+	save = make(chan saveData, 100)
+	slog = make(chan saveLog, 100)
+
+	rooms = &Rooms{
+		Count: make(chan int),
+		Json:  make(chan string),
+		Check: make(chan string),
+		Stop:  make(chan string),
+		Del:   make(chan string),
+		Add:   make(chan Info),
+	}
+)
+
+func main() {
+	rand.Seed(time.Now().UnixNano())
+
+	startConfig()
+
+	initMysql()
+	initClickhouse()
+
+	go mapRooms()
+	go announceCount()
+	go saveDB()
+	go saveLogs()
+	go broadcast()
+
+	http.HandleFunc("/ws/", wsHandler)
+	http.HandleFunc("/cmd/", cmdHandler)
+	http.HandleFunc("/list/", listHandler)
+	http.HandleFunc("/debug/", debugHandler)
+
+	go fastStart()
+
+	const SOCK = "/tmp/statbate.sock"
+	os.Remove(SOCK)
+	unixListener, err := net.Listen("unix", SOCK)
+	if err != nil {
+		log.Fatal("Listen (UNIX socket): ", err)
+	}
+	defer unixListener.Close()
+	os.Chmod(SOCK, 0777)
+	log.Fatal(http.Serve(unixListener, nil))
+}
+
+func initMysql() {
+	db, err := sqlx.Connect("mysql", conf.Conn["mysql"])
+	if err != nil {
+		panic(err)
+	}
+	Mysql = db
+}
+
+func initClickhouse() {
+	db, err := sqlx.Connect("clickhouse", conf.Conn["click"])
+	if err != nil {
+		panic(err)
+	}
+	Clickhouse = db
+}
+
+func randInt(min int, max int) int {
+	return min + rand.Intn(max-min)
+}
+
+func randString(n int) string {
+	const alphanum = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
+	var bytes = make([]byte, n)
+	rand.Read(bytes)
+	for i, b := range bytes {
+		bytes[i] = alphanum[b%byte(len(alphanum))]
+	}
+	return string(bytes)
+}
+
+func fastStart() {
+	defer func() {
+		go updateFileRooms()
+	}()
+	dat, err := os.ReadFile(conf.Conn["start"])
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+	list := make(map[string]Info)
+	if err := json.Unmarshal(dat, &list); err != nil {
+		fmt.Println(err.Error())
+		return
+	}
+	now := time.Now().Unix()
+	for k, v := range list {
+		if now > v.Last+60*20 {
+			continue
+		}
+		fmt.Println("fastStart:", k, v.Server, v.Proxy)
+		workerData := Info{
+			room:   k,
+			Server: v.Server,
+			Proxy:  v.Proxy,
+			Online: v.Online,
+			Start:  v.Start,
+			Last:   now,
+			Rid:    v.Rid,
+			Income: v.Income,
+			Dons:   v.Dons,
+			Tips:   v.Tips,
+		}
+		startRoom(workerData)
+		time.Sleep(100 * time.Millisecond)
+	}
+}
diff --git a/app/save.go b/app/save.go
new file mode 100644
index 0000000..7a7779c
--- /dev/null
+++ b/app/save.go
@@ -0,0 +1,193 @@
+package main
+
+import (
+	"fmt"
+	"time"
+)
+
+type saveData struct {
+	Room   string
+	From   string
+	Rid    int64
+	Amount int64
+	Now    int64
+}
+
+type saveLog struct {
+	Rid int64
+	Now int64
+	Mes string
+}
+
+type DonatorCache struct {
+	Id   int64
+	Last int64
+}
+
+func getDonId(name string) int64 {
+	var id int64
+	err := Mysql.Get(&id, "SELECT id FROM donator WHERE name=?", name)
+	if err != nil {
+		res, _ := Mysql.Exec("INSERT INTO donator (`name`) VALUES (?)", name)
+		id, _ = res.LastInsertId()
+	}
+	return id
+}
+
+func getRoomInfo(name string) (int64, bool) {
+	var id int64
+	result := true
+	err := Mysql.Get(&id, "SELECT id FROM room WHERE name=?", name)
+	if err != nil {
+		result = false
+	}
+	return id, result
+}
+
+func getSumTokens() int64 {
+	r := struct {
+		Date string
+		Sum  int64
+	}{}
+	err := Clickhouse.Get(&r, "SELECT toStartOfHour(toDateTime(`unix`)) as date, SUM(`token`) as sum FROM `stat` WHERE time = today() GROUP BY date ORDER BY date DESC LIMIT 1")
+	if err == nil && r.Sum > 0 {
+		return r.Sum
+	}
+	return 0
+}
+
+func saveDB() {
+	hours, _, _ := time.Now().Clock()
+
+	bulk := make(map[int]saveData)
+	update := make(map[int64]int64)
+	data := make(map[string]*DonatorCache)
+	index := make(map[string]int64)
+
+	index = map[string]int64{"hours": int64(hours), "tokens": getSumTokens(), "last": time.Now().Unix()}
+
+	for {
+		select {
+		case m := <-save:
+			//fmt.Println("Save channel:", len(save), cap(save))
+
+			now := time.Now().Unix()
+
+			if _, ok := data[m.From]; ok {
+				data[m.From].Last = now
+			} else {
+				data[m.From] = &DonatorCache{Id: getDonId(m.From), Last: now}
+			}
+
+			num := len(bulk)
+
+			bulk[num] = m
+
+			update[m.Rid] = m.Now
+
+			if num > 512 {
+
+				tx, err := Mysql.Begin()
+				if err == nil {
+					st, _ := tx.Prepare("INSERT INTO `stat` (`did`, `rid`, `token`, `time`) VALUES (?, ?, ?, ?)")
+					for _, v := range bulk {
+						st.Exec(data[v.From].Id, v.Rid, v.Amount, v.Now)
+					}
+					tx.Commit()
+					st.Close()
+				}
+
+				tx, err = Mysql.Begin()
+				if err == nil {
+					st, _ := tx.Prepare("UPDATE `room` SET `last` = ? WHERE `id` = ?")
+					for k, v := range update {
+						st.Exec(v, k)
+					}
+					tx.Commit()
+					st.Close()
+				}
+
+				tx, err = Clickhouse.Begin()
+				if err == nil {
+					st, _ := tx.Prepare("INSERT INTO stat VALUES (?, ?, ?, ?, ?)")
+					for _, v := range bulk {
+						st.Exec(uint32(data[v.From].Id), uint32(v.Rid), uint32(v.Amount), time.Unix(v.Now, 0), uint32(v.Now))
+					}
+					tx.Commit()
+					st.Close()
+				}
+
+				bulk = make(map[int]saveData)
+				update = make(map[int64]int64)
+			}
+
+			if m.Amount > 99 {
+				msg, err := json.Marshal(struct {
+					Room    string `json:"room"`
+					Donator string `json:"donator"`
+					Amount  int64  `json:"amount"`
+				}{
+					Room:    m.Room,
+					Donator: m.From,
+					Amount:  m.Amount,
+				})
+				if err == nil {
+					ws.Send <- msg
+				}
+			}
+
+			hours, minutes, seconds := time.Now().Clock()
+			if int64(hours) == index["hours"] {
+				index["tokens"] += m.Amount
+			} else {
+				index = map[string]int64{"hours": int64(hours), "tokens": 0, "last": 0}
+			}
+
+			if minutes >= 5 && now > index["last"]+30 {
+				seconds += minutes * 60
+				msg, err := json.Marshal(struct {
+					Index int64 `json:"index"`
+				}{Index: index["tokens"] / int64(seconds) * 3600 / 1000 * 5 / 100})
+				if err == nil {
+					ws.Send <- msg
+				}
+				index["last"] = now
+			}
+
+			if randInt(0, 10000) == 777 { // 0.001%
+				l := len(data)
+				for k, v := range data {
+					if now > v.Last+60*60*48 {
+						delete(data, k)
+					}
+				}
+				fmt.Println("Clean map:", l, "=>", len(data))
+			}
+		}
+	}
+}
+
+func saveLogs() {
+	bulk := make(map[int]saveLog)
+	for {
+		select {
+		case m := <-slog:
+			if len(m.Mes) > 0 {
+				num := len(bulk)
+				bulk[num] = m
+				if num > 2048 {
+					tx, err := Mysql.Begin()
+					if err == nil {
+						st, _ := tx.Prepare("INSERT INTO `logs` (`rid`, `time`, `mes`) VALUES (?, ?, ?)")
+						for _, v := range bulk {
+							st.Exec(v.Rid, v.Now, v.Mes)
+						}
+						tx.Commit()
+						st.Close()
+					}
+					bulk = make(map[int]saveLog)
+				}
+			}
+		}
+	}
+}
diff --git a/app/worker.go b/app/worker.go
new file mode 100644
index 0000000..e24551f
--- /dev/null
+++ b/app/worker.go
@@ -0,0 +1,261 @@
+package main
+
+import (
+	"fmt"
+	"net/http"
+	"net/url"
+	"strconv"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+var uptime = time.Now().Unix()
+
+func mapRooms() {
+
+	data := make(map[string]*Info)
+
+	for {
+		select {
+		case m := <-rooms.Add:
+			data[m.room] = &Info{Server: m.Server, 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 {
+			Count int `json:"count"`
+		}{Count: l})
+		if err == nil {
+			ws.Send <- msg
+		}
+	}
+}
+
+func reconnectRoom(workerData Info) {
+	n := randInt(10, 30)
+	fmt.Printf("Sleeping %d seconds...\n", n)
+	time.Sleep(time.Duration(n) * time.Second)
+	fmt.Println("reconnect:", workerData.room, workerData.Server, workerData.Proxy)
+	workerData.Last = time.Now().Unix()
+	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
+	}()
+
+	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()
+
+	leave := false
+	var timeout int64
+
+	dons := make(map[string]struct{})
+
+	for {
+
+		select {
+		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 workerData.Income > 1 && !leave {
+				go reconnectRoom(workerData)
+			}
+			return
+		}
+
+		now := time.Now().Unix()
+
+		m := string(message)
+		slog <- saveLog{Rid: workerData.Rid, Now: now, Mes: m}
+
+		if leave && now > timeout {
+			fmt.Println("room_leave exit:", workerData.room)
+			return
+		}
+
+		if now > workerData.Last+60*20 {
+			fmt.Println("no_mes exit:", workerData.room)
+			return
+		}
+
+		if m == "o" {
+			anon := "__anonymous__" + randString(9)
+			if err = c.WriteMessage(websocket.TextMessage, []byte(`["{\"method\":\"connect\",\"data\":{\"user\":\"`+anon+`\",\"password\":\"anonymous\",\"room\":\"`+workerData.room+`\",\"room_password\":\"12345\"}}"]`)); err != nil {
+				fmt.Println(err.Error(), workerData.room)
+				return
+			}
+			continue
+		}
+
+		if m == "h" {
+			if err = c.WriteMessage(websocket.TextMessage, []byte(`["{\"method\":\"updateRoomCount\",\"data\":{\"model_name\":\"`+workerData.room+`\",\"private_room\":\"false\"}}"]`)); err != nil {
+				fmt.Println(err.Error(), workerData.room)
+				return
+			}
+			continue
+		}
+
+		// remove a[...]
+		if len(m) > 3 && m[0:2] == "a[" {
+			m, _ = strconv.Unquote(m[2 : len(m)-1])
+		}
+
+		input := struct {
+			Method string   `json:"method"`
+			Args   []string `json:"args"`
+		}{}
+
+		if err := json.Unmarshal([]byte(m), &input); err != nil {
+			fmt.Println(err.Error(), workerData.room)
+			continue
+		}
+
+		if input.Method == "onAuthResponse" {
+			if err = c.WriteMessage(websocket.TextMessage, []byte(`["{\"method\":\"joinRoom\",\"data\":{\"room\":\"`+workerData.room+`\"}}"]`)); err != nil {
+				fmt.Println(err.Error(), workerData.room)
+				return
+			}
+			continue
+		}
+
+		if input.Method == "onRoomMsg" {
+			workerData.Last = now
+			rooms.Add <- workerData
+			continue
+		}
+
+		if input.Method == "onRoomCountUpdate" {
+			online, err := strconv.Atoi(input.Args[0])
+			if err == nil {
+				if online < 10 {
+					fmt.Println("few viewers room:", workerData.room)
+					return
+				}
+			}
+			workerData.Online = input.Args[0]
+			rooms.Add <- workerData
+			continue
+		}
+
+		if input.Method == "onPersonallyKicked" {
+			fmt.Println("onPersonallyKicked room:", workerData.room)
+			go reconnectRoom(workerData)
+			return
+		}
+
+		if input.Method == "onNotify" {
+			workerData.Last = now
+			rooms.Add <- workerData
+
+			arg := struct {
+				Type   string `json:"type"`
+				Name   string `json:"username"`
+				From   string `json:"from_username"`
+				Amount int64  `json:"amount"`
+			}{}
+
+			if err := json.Unmarshal([]byte(input.Args[0]), &arg); err != nil {
+				fmt.Println(err.Error(), workerData.room)
+				continue
+			}
+
+			if arg.Type == "clear_app" {
+				leave = true
+				timeout = now + 60*10
+				continue
+			}
+
+			if arg.Type == "room_leave" && workerData.room == arg.Name {
+				leave = true
+				timeout = now + 60*10
+				//fmt.Println("room_leave:", workerData.room)
+				continue
+			}
+
+			if arg.Type == "room_entry" && workerData.room == arg.Name {
+				leave = false
+				//fmt.Println("room_entry:", workerData.room)
+				continue
+			}
+
+			if arg.Type == "tip_alert" && len(arg.From) > 3 && arg.Amount > 0 {
+				workerData.Tips++
+				if _, ok := dons[arg.From]; !ok {
+					dons[arg.From] = struct{}{}
+					workerData.Dons++
+				}
+				save <- saveData{Room: workerData.room, From: arg.From, Rid: workerData.Rid, Amount: arg.Amount, Now: now}
+				workerData.Income += arg.Amount
+				rooms.Add <- workerData
+				if leave {
+					timeout = now + 60*20
+				}
+
+				// fmt.Println(donate.From)
+				// fmt.Println(donate.Amount)
+			}
+		}
+	}
+}
diff --git a/test/main.go b/test/main.go
new file mode 100644
index 0000000..7221c30
--- /dev/null
+++ b/test/main.go
@@ -0,0 +1,27 @@
+package main
+
+import (
+	"os"
+	"fmt"
+	"net/url"
+	"math/rand"
+)
+
+func randInt(min int, max int) int {
+    return min + rand.Intn(max-min)
+}
+
+func main() {
+	
+	if len(os.Args) < 3 {
+		fmt.Println("./test room server")
+		return
+	}
+	
+	room := os.Args[1]
+	server := os.Args[2]
+	
+	u := url.URL{Scheme: "wss", Host: server + ".stream.highwebmedia.com", Path: "/ws/555/kmdqiune/websocket"}
+	
+	statRoom(room, server, u)
+}
diff --git a/test/worker.go b/test/worker.go
new file mode 100644
index 0000000..34a25cf
--- /dev/null
+++ b/test/worker.go
@@ -0,0 +1,88 @@
+package main
+
+import (
+	"fmt"
+	"time"
+	"net/url"
+	"strconv"
+	"encoding/json"
+	"github.com/gorilla/websocket"
+)
+
+type Input struct {
+	Args   []string `json:"args"`
+	Method string   `json:"method"`
+}
+
+type Donate struct {
+	From   string `json:"from_username"`
+	Amount int64   `json:"amount"`
+}
+
+func statRoom(room, server string, u url.URL) {
+	c, _, err := websocket.DefaultDialer.Dial(u.String(), nil); if err != nil {
+		fmt.Println(err.Error())
+		return
+	}
+	timeout := time.Now().Unix() + 60*60
+	for {
+		
+		_, message, err := c.ReadMessage(); if err != nil {
+			fmt.Println(err.Error())
+			break 
+		}
+		
+		if time.Now().Unix() > timeout { 
+			fmt.Println("Timeout room:", room)
+			break 
+		}
+		
+		m := string(message)
+		
+		if m == "o"{
+			c.WriteMessage(websocket.TextMessage, []byte(`["{\"method\":\"connect\",\"data\":{\"user\":\"__anonymous__777\",\"password\":\"anonymous\",\"room\":\"` + room + `\",\"room_password\":\"12345\"}}"]`))
+			continue
+		}
+		
+		if m == "h"{
+			c.WriteMessage(websocket.TextMessage, []byte(`["{\"method\":\"updateRoomCount\",\"data\":{\"model_name\":\"` + room + `\",\"private_room\":\"false\"}}"]`))
+			continue
+		}
+
+		// remove a[...]
+		if len(m) > 3 && m[0:2] == "a[" {
+			m, _ = strconv.Unquote(m[2 : len(m)-1])
+		}
+		
+		input := Input{}
+		if err := json.Unmarshal([]byte(m), &input); err != nil {
+			fmt.Println(err.Error())
+			continue;
+		}
+
+		if(input.Method == "onAuthResponse"){
+			c.WriteMessage(websocket.TextMessage, []byte(`["{\"method\":\"joinRoom\",\"data\":{\"room\":\"` + room + `\"}}"]`))
+			continue
+		}
+		
+		if(input.Method == "onRoomCountUpdate"){
+			fmt.Println(input.Args[0], "online")
+			continue;
+		}
+
+		donate := Donate{}
+		if(input.Method == "onNotify"){
+			
+			timeout = time.Now().Unix() + 60*60
+			
+			if err := json.Unmarshal([]byte(input.Args[0]), &donate); err != nil {
+				fmt.Println(err.Error())
+				continue;
+			}
+			if(len(donate.From) > 3){
+				fmt.Println(donate.From, " send ", donate.Amount, "tokens")
+			}
+		}
+	}
+	c.Close()
+}