oc-peer/routers/router.go

// @APIVersion 1.0.0
// @Title oc-peer
// @Description Manage OpenCloud peers
// @Contact admin@o-cloud.io
// @TermsOfServiceUrl http://cloud.o-forge.io/
// @License AGPL
// @LicenseUrl https://www.gnu.org/licenses/agpl-3.0.html
package routers

import (
	"encoding/json"
	"fmt"
	"net/http"
	"oc-peer/controllers"
	"oc-peer/infrastructure"
	"strings"
	"time"

	oclib "cloud.o-forge.io/core/oc-lib"
	"cloud.o-forge.io/core/oc-lib/tools"
	beego "github.com/beego/beego/v2/server/web"
	"github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{
	CheckOrigin: func(r *http.Request) bool {
		return true
	},
}

func init() {
	ns := beego.NewNamespace("/oc",
		beego.NSNamespace("/status",
			beego.NSInclude(
				&controllers.StatusController{},
			),
		),
		beego.NSInclude(
			&controllers.PeerController{},
		),
		beego.NSNamespace("/version",
			beego.NSInclude(
				&controllers.VersionController{},
			),
		),
	)
	beego.AddNamespace(ns)

	// WebSocket — peer search (returns found peers + online/offline updates for them)
	beego.Handler("/oc/decentralized/search/:search", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		conn, err := upgrader.Upgrade(w, r, nil)
		if err != nil {
			return
		}
		defer conn.Close()
		parts := strings.Split(strings.TrimSuffix(r.URL.Path, "/"), "/")
		search := parts[len(parts)-1]

		user, _, groups := oclib.ExtractTokenInfoWs(*r)
		b, _ := json.Marshal(map[string]string{"search": search})
		infrastructure.EmitNATS(user, groups, tools.PropalgationMessage{
			Action:   tools.PB_SEARCH,
			DataType: tools.PEER.EnumIndex(),
			Payload:  b,
		})
		fmt.Println("SEARCH", search)
		controllers.Websocket(r.Context(), user, conn)
	}))

	// WebSocket — dedicated online/offline feed, user-scoped.
	//
	// The frontend sends lists of ShallowPeer to watch; oc-peer forwards them
	// to oc-discovery via NATS (with user identity). On any disconnect or error
	// oc-peer emits NATS close for all peers watched in this session.
	beego.Handler("/oc/decentralized/observe", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		conn, err := upgrader.Upgrade(w, r, nil)
		if err != nil {
			return
		}

		user, _, _ := oclib.ExtractTokenInfoWs(*r)
		connID := fmt.Sprintf("%p", conn)
		ch := make(chan infrastructure.WSMessage, 64)
		fmt.Println("SUBSCRIBED TO", connID)
		infrastructure.RegisterObserveSession(connID, user, ch)
		done := make(chan struct{})
		defer func() {
			infrastructure.CloseObserveSession(connID)
			close(done)
			conn.Close()
		}()

		const pongWait = 60 * time.Second
		const pingPeriod = 50 * time.Second
		const writeWait = 10 * time.Second

		conn.SetReadDeadline(time.Now().Add(pongWait))
		conn.SetPongHandler(func(string) error {
			return conn.SetReadDeadline(time.Now().Add(pongWait))
		})

		// Write loop: forward state-change messages and send keepalive pings.
		go func() {
			ticker := time.NewTicker(pingPeriod)
			defer ticker.Stop()
			for {
				select {
				case msg := <-ch:
					fmt.Println("ONLINE", msg)
					conn.SetWriteDeadline(time.Now().Add(writeWait))
					if conn.WriteJSON(msg) != nil {
						infrastructure.CloseObserveSession(connID)
						conn.Close()
						return
					}
				case <-ticker.C:
					conn.SetWriteDeadline(time.Now().Add(writeWait))
					if conn.WriteMessage(websocket.PingMessage, nil) != nil {
						infrastructure.CloseObserveSession(connID)
						conn.Close()
						return
					}
				case <-done:
					return
				}
			}
		}()

		// Read loop: receive peer lists from the frontend.
		var req struct {
			Peers []infrastructure.ShallowPeer `json:"peers"`
		}
		for {
			if err := conn.ReadJSON(&req); err != nil {
				return
			}
			// AddObservedPeers deduplicates, emits NATS observe, returns snapshot.
			for _, msg := range infrastructure.AddObservedPeers(connID, req.Peers) {
				select {
				case ch <- msg:
				default:
				}
			}
		}
	}))
}