diff --git a/app/cmd.go b/app/cmd.go
new file mode 100644
index 0000000..8c9be9f
--- /dev/null
+++ b/app/cmd.go
@@ -0,0 +1,113 @@
+package main
+
+import (
+ "fmt"
+ "net/http"
+ "os"
+ "runtime"
+ "strings"
+ "time"
+)
+
+var memInfo runtime.MemStats
+
+type Info struct {
+ ch chan struct{}
+ room string
+ Key string `json:"key"`
+ 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) {
+ runtime.ReadMemStats(&memInfo)
+ j, err := json.Marshal(struct {
+ Goroutines int
+ Uptime int64
+ Alloc uint64
+ HeapSys uint64
+ }{
+ Goroutines: runtime.NumGoroutine(),
+ Alloc: memInfo.Alloc,
+ HeapSys: memInfo.HeapSys,
+ Uptime: uptime,
+ })
+ 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["key"]) > 0 && len(params["proxy"]) > 0 {
+ now := time.Now().Unix()
+ workerData := Info{
+ room: params["room"][0],
+ Key: params["key"][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)
+}
diff --git a/app/conf.go b/app/conf.go
new file mode 100644
index 0000000..9cbcf6a
--- /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/sodaStart.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/main.go b/app/main.go
new file mode 100644
index 0000000..3544654
--- /dev/null
+++ b/app/main.go
@@ -0,0 +1,157 @@
+package main
+
+import (
+ "fmt"
+ "log"
+ "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
+
+ socketServer = make(chan []byte, 100)
+
+ 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() {
+ startConfig()
+
+ initMysql()
+ initClickhouse()
+
+ go mapRooms()
+ go announceCount()
+ go saveDB()
+ go saveLogs()
+ go socketHandler()
+
+ http.HandleFunc("/camsoda/cmd/", cmdHandler)
+ http.HandleFunc("/camsoda/list/", listHandler)
+ http.HandleFunc("/camsoda/debug/", debugHandler)
+
+ go fastStart()
+
+ const SOCK = "/tmp/camsoda.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 socketHandler() {
+
+ var (
+ err error
+ conn net.Conn
+ )
+
+ for {
+ select {
+ case b := <-socketServer:
+
+ if conn == nil {
+ conn, err = net.Dial("unix", "/tmp/echo.sock")
+ if err != nil {
+ fmt.Println(err.Error())
+ continue
+ }
+ }
+
+ if conn != nil {
+ if _, err = conn.Write(b); err != nil {
+ fmt.Println(err.Error())
+ conn.Close()
+ conn = nil
+ }
+ }
+
+ }
+ }
+}
+
+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.Key, v.Proxy)
+ workerData := Info{
+ room: k,
+ Key: v.Key,
+ 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..996840a
--- /dev/null
+++ b/app/save.go
@@ -0,0 +1,212 @@
+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 := []saveData{}
+ update := []struct {
+ Rid int64
+ Now int64
+ }{}
+
+ data := make(map[string]*DonatorCache)
+ index := make(map[string]int64)
+
+ ticker := time.NewTicker(10 * time.Second)
+ defer ticker.Stop()
+
+ cleanCache := time.NewTicker(12 * time.Hour)
+ defer cleanCache.Stop()
+
+ index = map[string]int64{"hours": int64(hours), "tokens": getSumTokens(), "last": time.Now().Unix()}
+
+ for {
+ select {
+ case <-cleanCache.C:
+ l := len(data)
+ now := time.Now().Unix()
+ for k, v := range data {
+ if now > v.Last+60*60*48 {
+ delete(data, k)
+ }
+ }
+ fmt.Println("Clean map:", l, "=>", len(data))
+ case <-ticker.C:
+ if len(bulk) > 0 {
+ 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 _, v := range update {
+ st.Exec(v.Now, v.Rid)
+ }
+ 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 = nil
+ update = nil
+ }
+ 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}
+ }
+
+ bulk = append(bulk, m)
+ update = append(update, struct {
+ Rid int64
+ Now int64
+ }{
+ Rid: m.Rid,
+ Now: m.Now,
+ })
+
+ if m.Amount > 49 {
+ msg, err := json.Marshal(struct {
+ Chanel string `json:"chanel"`
+ Room string `json:"room"`
+ Donator string `json:"donator"`
+ Amount int64 `json:"amount"`
+ }{
+ Chanel: "camsoda",
+ Room: m.Room,
+ Donator: m.From,
+ Amount: m.Amount,
+ })
+ if err == nil {
+ socketServer <- 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 {
+ Chanel string `json:"chanel"`
+ Index float64 `json:"index"`
+ }{
+ Chanel: "camsoda",
+ Index: float64(index["tokens"]) / float64(seconds) * 3600 * 0.05 / 1000,
+ })
+ if err == nil {
+ socketServer <- msg
+ }
+ index["last"] = now
+ }
+ }
+ }
+}
+
+func saveLogs() {
+ bulk := []saveLog{}
+ ticker := time.NewTicker(10 * time.Second)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ticker.C:
+ if len(bulk) > 0 {
+ 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 = nil
+ }
+ case m := <-slog:
+ if len(m.Mes) > 0 {
+ bulk = append(bulk, m)
+ }
+ }
+ }
+}
diff --git a/app/worker.go b/app/worker.go
new file mode 100644
index 0000000..c493b0d
--- /dev/null
+++ b/app/worker.go
@@ -0,0 +1,290 @@
+package main
+
+import (
+ "fmt"
+ "github.com/gorilla/websocket"
+ "net/http"
+ "net/url"
+ "strconv"
+ "strings"
+ "time"
+ "encoding/base64"
+)
+
+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 == 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)
+ }
+ }
+}