This commit is contained in:
poiuty 2022-07-27 11:34:26 +03:00
commit b080cea2df
10 changed files with 936 additions and 0 deletions

118
app/cmd.go Normal file
View file

@ -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"})
}

34
app/conf.go Normal file
View file

@ -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,
}
}

49
app/echo.go Normal file
View file

@ -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
}

142
app/main.go Normal file
View file

@ -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)
}
}

193
app/save.go Normal file
View file

@ -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)
}
}
}
}
}

261
app/worker.go Normal file
View file

@ -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)
}
}
}
}