From 2ef3e1fc8dc177f77b6ce7034aabd05fdc09c276 Mon Sep 17 00:00:00 2001 From: Vasiliy Tolstov Date: Fri, 12 Aug 2022 01:12:24 +0300 Subject: [PATCH] initial work Signed-off-by: Vasiliy Tolstov --- .gitignore | 4 + LICENSE | 21 ++++ app/cmd.go | 118 +++++++++++++++++++++ app/conf.go | 34 ++++++ app/echo.go | 49 +++++++++ app/main.go | 132 +++++++++++++++++++++++ app/save.go | 193 ++++++++++++++++++++++++++++++++++ app/worker.go | 276 +++++++++++++++++++++++++++++++++++++++++++++++++ go.mod | 17 +++ go.sum | 37 +++++++ test/main.go | 77 ++++++++++++++ test/worker.go | 142 +++++++++++++++++++++++++ 12 files changed, 1100 insertions(+) create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 app/cmd.go create mode 100644 app/conf.go create mode 100644 app/echo.go create mode 100644 app/main.go create mode 100644 app/save.go create mode 100644 app/worker.go create mode 100644 go.mod create mode 100644 go.sum create mode 100644 test/main.go create mode 100644 test/worker.go diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ad0bc14 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +.DS_Store +/app/chaturbate/chaturbate +/app/bongacams/bongacams +test/test 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..94d4822 --- /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 + ".bcccdn.com", Path: "/websocket"}) +} diff --git a/app/conf.go b/app/conf.go new file mode 100644 index 0000000..2ac77b3 --- /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/bongaStart.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..53cb10e --- /dev/null +++ b/app/main.go @@ -0,0 +1,132 @@ +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("/bongacams/ws/", wsHandler) + http.HandleFunc("/bongacams/cmd/", cmdHandler) + http.HandleFunc("/bongacams/list/", listHandler) + http.HandleFunc("/bongacams/debug/", debugHandler) + + go fastStart() + + const SOCK = "/tmp/bongacams.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 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(2 * time.Second) + } +} 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..e2f8a93 --- /dev/null +++ b/app/worker.go @@ -0,0 +1,276 @@ +package main + +import ( + "fmt" + "github.com/gorilla/websocket" + jsoniter "github.com/json-iterator/go" + "net/http" + "net/url" + "strings" + "time" +) + +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, 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 getAMF(room string) (bool, *AuthResponse) { + + v := &AuthResponse{} + + req, err := http.NewRequest(http.MethodPost, "https://rt.bongocams.com/tools/amf.php?res=771840&t=1654437233142", strings.NewReader(`method=getRoomData&args[]=`+room)) + if err != nil { + fmt.Println(err.Error()) + return false, v + } + req.Header.Add("Content-Type", "application/x-www-form-urlencoded; charset=UTF-8") + req.Header.Add("X-Requested-With", "XMLHttpRequest") + req.Header.Add("Accept", "application/json") + req.Header.Add("Referrer", "https://bongacams.com") + req.Header.Add("User-agent", "curl/7.79.1") + + rsp, err := http.DefaultClient.Do(req) + if err != nil { + fmt.Println(err.Error()) + return false, v + } + defer rsp.Body.Close() + + if err = json.NewDecoder(rsp.Body).Decode(v); err != nil { + fmt.Println(err.Error()) + return false, v + } + + return true, v +} + +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 + }() + + ok, v := getAMF(workerData.room) + if !ok { + fmt.Println("exit: no amf parms") + 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.UserData.Chathost, v.UserData.Username, v.UserData.DisplayName, v.UserData.Location, v.UserData.Chathost, v.UserData.IsRu, v.LocalData.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{}) + + 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()) + return + } + + now := time.Now().Unix() + + slog <- saveLog{workerData.Rid, now, string(message)} + + m := &ServerResponse{} + + if err = json.Unmarshal(message, m); err != nil { + fmt.Println(err.Error(), workerData.room) + continue + } + + workerData.Last = now + rooms.Add <- workerData + + 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 { + //fmt.Println(d.F.Username, "send", d.A, "tokens") + + workerData.Tips++ + if _, ok := dons[d.F.Username]; !ok { + dons[d.F.Username] = struct{}{} + workerData.Dons++ + } + + save <- saveData{workerData.room, d.F.Username, workerData.Rid, d.A, now} + + workerData.Income += d.A + rooms.Add <- workerData + } + } + } +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..eb3b3f3 --- /dev/null +++ b/go.mod @@ -0,0 +1,17 @@ +module stripchat + +go 1.19 + +require ( + github.com/ClickHouse/clickhouse-go v1.5.4 + github.com/go-sql-driver/mysql v1.6.0 + github.com/gorilla/websocket v1.5.0 + github.com/jmoiron/sqlx v1.3.5 + github.com/json-iterator/go v1.1.12 +) + +require ( + github.com/cloudflare/golz4 v0.0.0-20150217214814-ef862a3cdc58 // indirect + github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 // indirect + github.com/modern-go/reflect2 v1.0.2 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..20e681b --- /dev/null +++ b/go.sum @@ -0,0 +1,37 @@ +github.com/ClickHouse/clickhouse-go v1.5.4 h1:cKjXeYLNWVJIx2J1K6H2CqyRmfwVJVY1OV1coaaFcI0= +github.com/ClickHouse/clickhouse-go v1.5.4/go.mod h1:EaI/sW7Azgz9UATzd5ZdZHRUhHgv5+JMS9NSr2smCJI= +github.com/bkaradzic/go-lz4 v1.0.0 h1:RXc4wYsyz985CkXXeX04y4VnZFGG8Rd43pRaHsOXAKk= +github.com/bkaradzic/go-lz4 v1.0.0/go.mod h1:0YdlkowM3VswSROI7qDxhRvJ3sLhlFrRRwjwegp5jy4= +github.com/cloudflare/golz4 v0.0.0-20150217214814-ef862a3cdc58 h1:F1EaeKL/ta07PY/k9Os/UFtwERei2/XzGemhpGnBKNg= +github.com/cloudflare/golz4 v0.0.0-20150217214814-ef862a3cdc58/go.mod h1:EOBUe0h4xcZ5GoxqC5SDxFQ8gwyZPKQoEzownBlhI80= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w= +github.com/go-sql-driver/mysql v1.6.0 h1:BCTh4TKNUYmOmMUcQ3IipzF5prigylS7XXjEkfCHuOE= +github.com/go-sql-driver/mysql v1.6.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc= +github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/jmoiron/sqlx v1.2.0/go.mod h1:1FEQNm3xlJgrMD+FBdI9+xvCksHtbpVBBw5dYhBSsks= +github.com/jmoiron/sqlx v1.3.5 h1:vFFPA71p1o5gAeqtEAwLU4dnX2napprKtHr7PYIcN3g= +github.com/jmoiron/sqlx v1.3.5/go.mod h1:nRVWtLre0KfCLJvgxzCsLVMogSvQ1zNJtpYr2Ccp0mQ= +github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/lib/pq v1.0.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo= +github.com/lib/pq v1.2.0 h1:LXpIM/LZ5xGFhOpXAQUIMM1HdyqzVYM13zNdjCEEcA0= +github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo= +github.com/mattn/go-sqlite3 v1.9.0/go.mod h1:FPy6KqzDD04eiIsT53CuJW3U88zkxoIYsOqkbpncsNc= +github.com/mattn/go-sqlite3 v1.14.6 h1:dNPt6NO46WmLVt2DLNpwczCmdV5boIZ6g/tlDrlRUbg= +github.com/mattn/go-sqlite3 v1.14.6/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 h1:ZqeYNhU3OHLH3mGKHDcjJRFFRrJa6eAM5H+CtDdOsPc= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/pierrec/lz4 v2.0.5+incompatible h1:2xWsjqPFWcplujydGg4WmhC/6fZqK42wMM8aXeqhl0I= +github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= diff --git a/test/main.go b/test/main.go new file mode 100644 index 0000000..c1b1655 --- /dev/null +++ b/test/main.go @@ -0,0 +1,77 @@ +package main + +import ( + "bytes" + "context" + "fmt" + "io/ioutil" + "net/http" + "net/url" + "os" + "regexp" + "strings" +) + +func main() { + if len(os.Args) < 4 { + fmt.Println("./test room server proxy") + return + } + + room := os.Args[1] + server := os.Args[2] + proxy := os.Args[3] + + u, token, id := getToken(room) + + startRoom(room, server, proxy, u, token, id) +} + +func getToken(room string) (*url.URL, string, string) { + req, err := http.NewRequestWithContext(context.TODO(), http.MethodGet, "https://ru.stripchat.com/"+room, nil) + if err != nil { + panic(err) + } + req.Header.Add("User-Agent", "Mozilla/5.0 (X11; Linux x86_64; rv:33.0) Gecko/20100101 Firefox/33.0") + req.Header.Add("Accept", "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8") + req.Header.Add("Accept-Language", "en-US,en;q=0.5") + req.Header.Add("Connection", "keep-alive") + req.Header.Add("Referer", "https://ru.stripchat.com") + rsp, err := http.DefaultClient.Do(req) + if err != nil { + panic(err) + } else if rsp.StatusCode != http.StatusOK { + panic("failed to get rsp") + } + + defer rsp.Body.Close() + + buf, err := ioutil.ReadAll(rsp.Body) + if err != nil { + panic(err) + } + re := regexp.MustCompile(`"websocketUrl":"*(.*?)\s*"`) + m := re.FindSubmatch(buf) + if len(m) != 2 { + panic("not match") + } + r := bytes.ReplaceAll(m[1], []byte(`\u002F`), []byte(`/`)) + u, err := url.Parse(string(r)) + if err != nil { + panic(err) + } + + re = regexp.MustCompile(`"token":"*(.*?)\s*"`) + m = re.FindSubmatch(buf) + if len(m) != 2 { + panic("not match") + } + + re = regexp.MustCompile(`img.strpst.com/thumbs/*(.*?)\s*"`) + id := re.FindSubmatch(buf) + if len(id) != 2 { + panic("not match") + } + xid := string(id[1]) + return u, string(m[1]), xid[strings.Index(xid, "/")+1:] +} diff --git a/test/worker.go b/test/worker.go new file mode 100644 index 0000000..d21f575 --- /dev/null +++ b/test/worker.go @@ -0,0 +1,142 @@ +package main + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" + "net/url" + "strings" + "time" + + "github.com/gorilla/websocket" +) + +var messages [][]byte + +func init() { + messages = append(messages, []byte(`{"id":"1660248194970-sub-lotteryChanged","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/lotteryChanged"}`)) + messages = append(messages, []byte(`{"id":"1660248194970-sub-userBanned:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/userBanned:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194970-sub-goalChanged:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/goalChanged:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194970-sub-modelStatusChanged:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/modelStatusChanged:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194971-sub-broadcastSettingsChanged:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/broadcastSettingsChanged:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194971-sub-tipMenuUpdated:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/tipMenuUpdated:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194971-sub-topicChanged:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/topicChanged:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194971-sub-userUpdated:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/userUpdated:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194971-sub-interactiveToyStatusChanged:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/interactiveToyStatusChanged:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194971-sub-groupShow:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/groupShow:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194971-sub-deleteChatMessages:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/deleteChatMessages:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194971-sub-tipLeaderboardSettingsUpdated:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/tipLeaderboardSettingsUpdated:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194971-sub-modelAppUpdated:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/modelAppUpdated:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194972-sub-newKing:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/newKing:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194972-sub-privateMessageSettingsChanged:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/privateMessageSettingsChanged:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194972-sub-newChatMessage:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/newChatMessage:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194972-sub-fanClubUpdated:XID","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/fanClubUpdated:XID"}`)) + messages = append(messages, []byte(`{"id":"1660248194972-sub-viewServerChanged:hls-07","method":"PUT","url":"/front/clients/CLIENT_ID/subscriptions/viewServerChanged:hls-07"}`)) +} + +type ServerResponse struct { + SubscriptionKey string `json:"subscriptionKey,omitempty"` + Params struct { + ClientId string `json:"clientId,omitempty"` + Message struct { + Type string `json:"type,omitempty"` + Userdata struct { + Username string `json:"username,omitempty"` + } `json:"userdata,omitempty"` + Details struct { + Amount float64 `json:"amount,omitempty"` + LovenseDetails struct { + Type string `json:"type,omitempty"` + Detail struct { + Name string `json:"name,omitempty"` + Amount float64 `json:"amount,omitempty"` + } `json:"detail,omitempty"` + } `json:"lovenseDetails"` + } `json:"details,omitempty"` + } `json:"message,omitempty"` + } `json:"params,omitempty"` +} + +func startRoom(room, server, proxy string, u *url.URL, token string, id string) { + // curl -vvv -X POST -H "Content-Type: application/x-www-form-urlencoded; charset=UTF-8" -H "X-Requested-With: XMLHttpRequest" -d "method=getRoomData" -d "args[]=Icehotangel" "https://rt.bongocams.com/tools/amf.php?res=771840&t=1654437233142" + fmt.Printf("token %s id %s\n", token, id) + fmt.Println("Start", room, "server", server, "proxy", proxy) + + Dialer := *websocket.DefaultDialer + + proxyMap := make(map[string]string) + proxyMap["us"] = "aaa:port" + proxyMap["fi"] = "bbb:port" + + if _, ok := proxyMap[proxy]; ok { + Dialer = websocket.Dialer{ + Proxy: http.ProxyURL(&url.URL{ + Scheme: "http", // or "https" depending on your proxy + Host: proxyMap[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()) + return + } + + defer c.Close() + + ws, _, err := Dialer.Dial("wss://websocket.stripchat.com/connection/websocket", nil) + if err != nil { + panic(err) + } + defer ws.Close() + + for { + _, message, err := c.ReadMessage() + if err != nil { + fmt.Println("return " + err.Error()) + return + } + + fmt.Println(string(message)) + + m := &ServerResponse{} + + if err = json.Unmarshal(message, m); err != nil { + fmt.Println(err.Error()) + continue + } + + if m.SubscriptionKey == "connected" { + fmt.Println("connected") + for _, msg := range messages { + b := bytes.ReplaceAll(msg, []byte(`CLIENT_ID`), []byte(m.Params.ClientId)) + b = bytes.ReplaceAll(b, []byte(`XID`), []byte(id)) + if err = c.WriteMessage(websocket.TextMessage, b); err != nil { + fmt.Println("return " + err.Error()) + return + } + } + if err = ws.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"params":{"token":"%s","name":"js"},"id":1}`, token))); err != nil { + panic(err) + } + } + + if !strings.Contains(m.SubscriptionKey, "newChatMessage") { + continue + } + + fmt.Println("proper msg") + + if m.Params.Message.Type == "lovense" { + fmt.Println("donate") + fmt.Println(m.Params.Message.Details.LovenseDetails.Detail.Name, " send ", m.Params.Message.Details.LovenseDetails.Detail.Amount, "tokens") + } else if m.Params.Message.Type == "tip" { + fmt.Println("donate") + fmt.Println(m.Params.Message.Userdata.Username, " send ", m.Params.Message.Details.Amount, "tokens") + } + } +}