start working on supporting rtcv requests

Author: mjarkk

README.md

@@ -1,6 +1,6 @@
 # RT-CV scraper client
+A helper program that aims to ease the communication between a scraper and [RT-CV](https://bitbucket.org/teamscript/rt-cv)
 
-A helper program that aims to ease the communication between a scraper and [RT-CV](https://github.com/script-development/RT-CV)
 
 ## How does this work?
 
@@ -144,6 +144,42 @@ Check if a reference number is in the cache
 - Body: The reference number
 - Resp: **true** / **false**
 
+### `$SCRAPER_ADDRESS/server_request`
+
+This route only response when once rt-cv has a request for the scraper.
+
+This url should be called continously by the scraper and should have no request timeout as this request might take hours to before RT-CV sends a request.
+
+- Resp: a request for something by RT-CV
+
+The request and respones are defined in [bitbucket.org/teamscript/rt-cv > /controller/scraperWebsocket/README.md](https://bitbucket.org/teamscript/rt-cv/src/main/controller/scraperWebsocket/README.md)
+
+```jsonc
+{
+  "type": "message type",
+  "id": "message id",
+  "data": {} // Change this
+}
+```
+
+### `$SCRAPER_ADDRESS/server_response`
+
+You should send a response to `/server_request` to this url
+
+- Body: Almost equal to the response of `/server_request` but with the data changed to what RT-CV expected
+
+The request and respones are defined in [bitbucket.org/teamscript/rt-cv > /controller/scraperWebsocket/README.md](https://bitbucket.org/teamscript/rt-cv/src/main/controller/scraperWebsocket/README.md)
+
+```jsonc
+{
+  "type": "message type",
+  "id": "message id",
+  "data": {} // Change this
+}
+```
+
+- Resp: **true**
+
 ## `env.json` is in another dir or has another name?
 
 You can change the credentials file location using this shell variable

api.go

@@ -10,7 +10,11 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"strings"
+	"sync/atomic"
 	"time"
+
+	"github.com/gorilla/websocket"
 )
 
 type serverConn struct {
@@ -26,12 +30,20 @@ type API struct {
 	MockMode  bool
 	mockCache map[string]time.Time
 
+	CancelPreviouseCommunicationChan chan struct{}
+	WebsocketReq                     chan []byte
+	WebsocketResp                    chan []byte
+
 	Cache map[string]time.Time
 }
 
 // NewAPI creates a new instance of the API
 func NewAPI() *API {
 	return &API{
+		CancelPreviouseCommunicationChan: make(chan struct{}),
+		WebsocketReq:                     make(chan []byte),
+		WebsocketResp:                    make(chan []byte),
+
 		Cache: map[string]time.Time{},
 	}
 }
@@ -64,6 +76,9 @@ func (a *API) SetCredentials(credentialsList []SetCredentialsArg) error {
 		if credentials.ServerLocation == "" {
 			return errors.New("server_location cannot be empty")
 		}
+		if !strings.HasPrefix(credentials.ServerLocation, "http://") && !strings.HasPrefix(credentials.ServerLocation, "https://") {
+			return errors.New("server_location must start with a supported protocol like: http:// or https://")
+		}
 		conn.serverLocation = credentials.ServerLocation
 
 		if credentials.APIKeyID == "" {
@@ -166,6 +181,41 @@ func (c *serverConn) DoRequest(method, path string, body, unmarshalResInto inter
 	}
 }
 
+func (c *serverConn) tryConnectToWS() *websocket.Conn {
+	url := c.serverLocation
+	url = strings.Replace(url, "http://", "ws://", 1)
+	url = strings.Replace(url, "https://", "wss://", 1)
+	url += "/api/v1/scraper/ws"
+
+	attempt := 0
+	for {
+		conn, _, err := websocket.DefaultDialer.Dial(url, http.Header{"Authorization": []string{c.authHeaderValue}})
+		if err == nil {
+			if attempt > 0 {
+				fmt.Println("connected to web socket")
+			}
+			return conn
+		}
+
+		attempt++
+		retryInSeconds := time.Second
+		if attempt == 1 {
+			// retry in 1 second
+		} else if attempt <= 2 {
+			retryInSeconds *= 2
+		} else if attempt <= 4 {
+			retryInSeconds *= 4
+		} else if attempt <= 6 {
+			retryInSeconds *= 10
+		} else {
+			retryInSeconds *= 15
+		}
+
+		fmt.Printf("unable to connect to web socket, error: %s, retrying in %s\n", err, retryInSeconds)
+		time.Sleep(retryInSeconds)
+	}
+}
+
 // NoCredentials returns true if the SetCredentials method was not yet called and we aren't in mock mode
 func (a *API) NoCredentials() bool {
 	return len(a.connections) == 0 && !a.MockMode
@@ -193,3 +243,115 @@ func (a *API) CacheEntryExists(referenceNr string) bool {
 	}
 	return !expired
 }
+
+// WSMsg is a message recived and send to the websocket
+type WSMsg[T any] struct {
+	Type string `json:"type"`
+	ID   string `json:"id,omitempty"`
+	Data T      `json:"data"`
+}
+
+// ConnectToWS connects to the rtcv websocket
+func (a *API) ConnectToWS() {
+	if a.MockMode {
+		go func() {
+			for {
+				resp := <-a.WebsocketResp
+				fmt.Println("got websocket response but we are in mock mode", string(resp))
+			}
+		}()
+		return
+	}
+
+	server := a.connections[0]
+
+	url := server.serverLocation
+	url = strings.Replace(url, "http://", "ws://", 1)
+	url = strings.Replace(url, "https://", "wss://", 1)
+	url += "/api/v1/scraper/ws"
+
+	var c *websocket.Conn
+	defer func() {
+		if c != nil {
+			c.Close()
+		}
+	}()
+
+	go func() {
+		for {
+			// TODO: if the response fails to send data might get lost.
+			//   It would be nice if the response is retried when WriteMessage fails
+			resp := <-a.WebsocketResp
+			err := c.WriteMessage(1, resp)
+			if err != nil {
+				fmt.Println("unable to write ws response:", err)
+			}
+		}
+	}()
+
+	firstMessage := true
+	var aMessageWasHandled atomic.Bool
+	for {
+		c = a.connections[0].tryConnectToWS()
+
+		for {
+			msgType, msgBytes, err := c.ReadMessage()
+			if err != nil {
+				fmt.Println("error reading from web socket:", err)
+				break
+			}
+
+			switch msgType {
+			case 1, 2:
+				// 1 - text message
+				// 2 - binary message
+				// Ok continue
+			default:
+				// Ignore other message types
+				continue
+			}
+
+			msg := WSMsg[json.RawMessage]{}
+			err = json.Unmarshal(msgBytes, &msg)
+			if err != nil {
+				fmt.Println("error unmarshaling web socket message:", err)
+				continue
+			}
+
+			timeout := time.Second
+			if aMessageWasHandled.Load() || firstMessage {
+				// It might be this scraper does not listen to the /server_request url thus we will try to send something over a channel that will never read
+				// That's a lot of waisted time
+				timeout = time.Second * 30
+			}
+			firstMessage = false
+
+			go func(msgBytes []byte, timeout time.Duration) {
+				select {
+				case a.WebsocketReq <- msgBytes:
+					// Ok message was send
+					aMessageWasHandled.Store(true)
+				case <-time.After(timeout):
+					errMsg := "Unable to handle request by RT-CV server"
+					if !aMessageWasHandled.Load() {
+						errMsg += ", probably becuase there is no one waiting for a response"
+					}
+					fmt.Println(errMsg)
+				}
+			}(msgBytes, timeout)
+		}
+
+		c.Close()
+	}
+
+}
+
+// CancelPreviouseCommunication cancels the previous communication if it's still running
+func (a *API) CancelPreviouseCommunication() {
+	select {
+	case a.CancelPreviouseCommunicationChan <- struct{}{}:
+		// A previous communication was canceled
+	default:
+		// There are no more previous communications to cancel
+	}
+}

env.go

@@ -0,0 +1,89 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+)
+
+// Env contains the structure of an env.json file
+type Env struct {
+	PrivateKey         string      `json:"private_key"`
+	PublicKey          string      `json:"public_key"`
+	PrimaryServer      EnvServer   `json:"primary_server"`
+	AlternativeServers []EnvServer `json:"alternative_servers"`
+	MockMode           bool        `json:"mock_mode"`
+	MockUsers          []EnvUser   `json:"mock_users"`
+}
+
+func (e *Env) validate() error {
+	if e.MockMode {
+		if len(e.MockUsers) == 0 {
+			fmt.Println(`"mock_users" is empty in env.json, most scrapers require at least one user to login.`)
+			fmt.Println(`For documentation about mocking see https://github.com/script-development/rtcv_scraper_client`)
+			if e.MockUsers == nil {
+				e.MockUsers = []EnvUser{}
+			}
+		}
+		return nil
+	}
+
+	err := e.PrimaryServer.validate()
+	if err != nil {
+		return fmt.Errorf("primary_server.%s", err.Error())
+	}
+
+	for idx, server := range e.AlternativeServers {
+		err := server.validate()
+		if err != nil {
+			return fmt.Errorf("%s[%d].%s", "alternative_servers", idx, err.Error())
+		}
+	}
+
+	keyPairHelpMsg := `, use the go program inside the "gen_key" folder to generate a key pair`
+	if e.PrivateKey == "" && e.PublicKey == "" {
+		return errors.New(`"public_key" and "private_key" are required` + keyPairHelpMsg)
+	} else if e.PrivateKey == "" {
+		return errors.New(`"private_key" required` + keyPairHelpMsg)
+	} else if e.PublicKey == "" {
+		return errors.New(`"public_key" required` + keyPairHelpMsg)
+	}
+
+	return nil
+}
+
+// EnvServer contains the structure of the primary_server and alternative_servers inside the .env file
+type EnvServer struct {
+	ServerLocation string `json:"server_location"`
+	APIKeyID       string `json:"api_key_id"`
+	APIKey         string `json:"api_key"`
+}
+
+func (e *EnvServer) validate() error {
+	if e.ServerLocation == "" {
+		return errors.New("server_location is required")
+	}
+	if e.APIKeyID == "" {
+		return errors.New("api_key_id is required")
+	}
+	if e.APIKey == "" {
+		return errors.New("api_key is required")
+	}
+
+	return nil
+}
+
+func (e *EnvServer) toCredArg(isPrimary bool) SetCredentialsArg {
+	return SetCredentialsArg{
+		ServerLocation: e.ServerLocation,
+		APIKeyID:       e.APIKeyID,
+		APIKey:         e.APIKey,
+		Primary:        isPrimary,
+	}
+}
+
+// EnvUser contains the structure of the login_users inside the .env file
+type EnvUser struct {
+	Username          string `json:"username"`
+	Password          string `json:"password"`
+	EncryptedPassword string `json:"encryptedPassword,omitempty"`
+}

go.mod

@@ -4,6 +4,8 @@ go 1.18
 
 require github.com/valyala/fasthttp v1.40.0
 
+require github.com/gorilla/websocket v1.5.0 // indirect
+
 require (
 	github.com/andybalholm/brotli v1.0.4 // indirect
 	github.com/klauspost/compress v1.15.9 // indirect

go.sum

@@ -1,5 +1,7 @@
 github.com/andybalholm/brotli v1.0.4 h1:V7DdXeJtZscaqfNuAdSRuRFzuiKlHSC/Zh3zl9qY3JY=
 github.com/andybalholm/brotli v1.0.4/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig=
+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/klauspost/compress v1.15.0/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/klauspost/compress v1.15.9 h1:wKRjX6JRtDdrE9qwa4b/Cip7ACOshUI4smpCQanqjSY=
 github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=