Reticulum-Go/pkg/controlapi/link.go

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2024-2026 Quad4.io

package controlapi

import (
	"context"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"time"

	"quad4/reticulum-go/pkg/destination"
	"quad4/reticulum-go/pkg/identity"
	"quad4/reticulum-go/pkg/link"
	"quad4/reticulum-go/pkg/packet"
	"quad4/reticulum-go/pkg/resource"
)

// newOutboundLinkCallbacks builds the established/closed callbacks passed
// to link.NewLink for a link.open command, plus the linkSession they
// populate. destHashHex is only used to label a linkFailedEvent if the
// link never becomes active.
//
// link.Teardown and the establishment watchdog both invoke the closed
// callback while still holding the Link's own internal mutex, so closed
// must never call back into locked Link accessors like GetLinkID: doing so
// would deadlock. idHex is therefore cached on linkSession by the caller
// (see handleLinkOpen) and by established on the initiator path, which the
// link package invokes from a fresh goroutine because ValidateLinkProof holds
// the link mutex. The responder path calls established synchronously after RTT.
func newOutboundLinkCallbacks(sess *session, destHashHex string) (ls *linkSession, established func(*link.Link), closed func(*link.Link)) {
	ls = &linkSession{}

	established = func(l *link.Link) {
		ls.link = l
		idHex := hex.EncodeToString(l.GetLinkID())

		ls.establishedMu.Lock()
		ls.idHex = idHex
		ls.established = true
		ls.establishedMu.Unlock()

		sess.addLink(idHex, ls)
		wireLinkEvents(sess, l, idHex)
		sess.broadcast(linkEstablishedEvent{Type: "link.established", LinkID: idHex, RemoteHash: remoteHashOf(l)})
	}

	closed = func(l *link.Link) {
		ls.establishedMu.Lock()
		idHex := ls.idHex
		wasEstablished := ls.established
		ls.establishedMu.Unlock()

		sess.removeLink(idHex)

		if wasEstablished {
			sess.broadcast(linkClosedEvent{Type: "link.closed", LinkID: idHex})
		} else {
			sess.broadcast(linkFailedEvent{
				Type:            "link.failed",
				LinkID:          idHex,
				DestinationHash: destHashHex,
				Error:           "link establishment failed or timed out",
			})
		}
	}

	return ls, established, closed
}

// wireInboundLinks makes dest accept links and reports every one that
// becomes active as a linkEstablishedEvent, forwarding its data and
// eventual teardown to the session's WebSocket clients.
func wireInboundLinks(sess *session, dest *destination.Destination) {
	dest.AcceptsLinks(true)
	dest.SetLinkEstablishedCallback(func(v any) {
		lnk, ok := v.(*link.Link)
		if !ok {
			return
		}

		idHex := hex.EncodeToString(lnk.GetLinkID())
		sess.addLink(idHex, &linkSession{link: lnk, idHex: idHex, established: true})
		wireLinkEvents(sess, lnk, idHex)

		lnk.SetLinkClosedCallback(func(*link.Link) {
			sess.removeLink(idHex)
			sess.broadcast(linkClosedEvent{Type: "link.closed", LinkID: idHex})
		})

		sess.broadcast(linkEstablishedEvent{Type: "link.established", LinkID: idHex, RemoteHash: remoteHashOf(lnk)})
	})
}

func wireLinkEvents(sess *session, lnk *link.Link, idHex string) {
	lnk.SetPacketCallback(func(data []byte, _ *packet.Packet) {
		sess.broadcast(linkDataEvent{Type: "link.data", LinkID: idHex, Data: base64.StdEncoding.EncodeToString(data)})
	})
	_ = lnk.SetResourceStrategy(link.AcceptAll)
	lnk.SetResourceStartedCallback(func(_ any) {
		sess.broadcast(resourceStartedEvent{Type: "resource.started", LinkID: idHex})
	})
	lnk.SetResourceConcludedCallback(func(v any) {
		evt := resourceConcludedEvent{Type: "resource.concluded", LinkID: idHex, Success: true}
		switch r := v.(type) {
		case link.IncomingResource:
			evt.Data = base64.StdEncoding.EncodeToString(r.Data)
			if name, ok := resourceNameFromMeta(r.Metadata); ok {
				evt.Name = name
			}
			if len(r.Hash) > 0 {
				evt.Hash = hex.EncodeToString(r.Hash)
			}
		case []byte:
			evt.Data = base64.StdEncoding.EncodeToString(r)
		case nil:
			evt.Success = false
			evt.Error = "resource transfer failed"
		default:
			evt.Success = false
			evt.Error = "unknown resource conclude payload"
		}
		sess.broadcast(evt)
	})
	lnk.SetRemoteIdentifiedCallback(func(_ *link.Link, remote *identity.Identity) {
		if remote == nil {
			return
		}
		sess.broadcast(linkRemoteIdentifiedEvent{
			Type:         "link.remote_identified",
			LinkID:       idHex,
			IdentityHash: remote.GetHexHash(),
		})
	})
}

func resourceNameFromMeta(meta map[string]any) (string, bool) {
	if meta == nil {
		return "", false
	}
	raw, ok := meta["name"]
	if !ok {
		return "", false
	}
	switch v := raw.(type) {
	case string:
		return v, true
	case []byte:
		return string(v), true
	default:
		return "", false
	}
}

func remoteHashOf(lnk *link.Link) string {
	if remote := lnk.GetRemoteIdentity(); remote != nil {
		return remote.GetHexHash()
	}
	return ""
}

// handleLinkOpen processes a link.open command: it resolves the peer's
// identity from a previously-seen announce, opens an outbound link, and
// reports the outcome as linkEstablishedEvent or linkFailedEvent.
func (c *wsClient) handleLinkOpen(raw []byte) {
	var cmd linkOpenCommand
	if err := json.Unmarshal(raw, &cmd); err != nil {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.open", Error: "invalid command json"})
		return
	}

	destHash, err := hex.DecodeString(cmd.DestinationHash)
	if err != nil || len(destHash) != 16 {
		c.send(linkFailedEvent{Type: "link.failed", DestinationHash: cmd.DestinationHash, Error: "destination_hash must be 16 hex-encoded bytes"})
		return
	}

	remoteIdentity, err := identity.Recall(destHash)
	if err != nil {
		c.send(linkFailedEvent{Type: "link.failed", DestinationHash: cmd.DestinationHash, Error: "unknown destination identity, wait for an announce first"})
		return
	}

	destOut, err := destination.FromHash(destHash, remoteIdentity, destination.Single, c.server.transport)
	if err != nil {
		c.send(linkFailedEvent{Type: "link.failed", DestinationHash: cmd.DestinationHash, Error: err.Error()})
		return
	}

	ls, established, closed := newOutboundLinkCallbacks(c.session, cmd.DestinationHash)
	lnk := link.NewLink(destOut, c.server.transport, nil, established, closed)
	go func() {
		if err := c.server.transport.AwaitPath(context.Background(), destHash); err != nil {
			c.send(linkFailedEvent{Type: "link.failed", DestinationHash: cmd.DestinationHash, Error: err.Error()})
			return
		}
		if err := lnk.Establish(); err != nil {
			c.send(linkFailedEvent{Type: "link.failed", DestinationHash: cmd.DestinationHash, Error: err.Error()})
			return
		}
		ls.establishedMu.Lock()
		if ls.idHex == "" {
			ls.idHex = hex.EncodeToString(lnk.GetLinkID())
		}
		ls.establishedMu.Unlock()
		lnk.Start()
	}()
}

// handleLinkSend processes a link.send command, forwarding data over an
// already-established link owned by the session.
func (c *wsClient) handleLinkSend(raw []byte) {
	var cmd linkSendCommand
	if err := json.Unmarshal(raw, &cmd); err != nil {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.send", Error: "invalid command json"})
		return
	}
	ls, ok := c.session.getLink(cmd.LinkID)
	if !ok {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.send", Error: "unknown link_id"})
		return
	}
	data, err := base64.StdEncoding.DecodeString(cmd.Data)
	if err != nil {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.send", Error: "data must be base64"})
		return
	}
	if err := ls.link.SendPacket(data); err != nil {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.send", Error: err.Error()})
	}
}

// handleLinkClose processes a link.close command. The link's own closed
// callback emits the resulting linkClosedEvent.
func (c *wsClient) handleLinkClose(raw []byte) {
	var cmd linkCloseCommand
	if err := json.Unmarshal(raw, &cmd); err != nil {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.close", Error: "invalid command json"})
		return
	}
	ls, ok := c.session.getLink(cmd.LinkID)
	if !ok {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.close", Error: "unknown link_id"})
		return
	}
	ls.link.Teardown()
}

func (c *wsClient) handleLinkRequest(raw []byte) {
	var cmd linkRequestCommand
	if err := json.Unmarshal(raw, &cmd); err != nil {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.request", Error: "invalid command json"})
		return
	}
	if cmd.Path == "" {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.request", Error: "path is required"})
		return
	}
	ls, ok := c.session.getLink(cmd.LinkID)
	if !ok {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.request", Error: "unknown link_id"})
		return
	}
	var payload any
	if cmd.Data != "" {
		data, err := base64.StdEncoding.DecodeString(cmd.Data)
		if err != nil {
			c.send(commandErrorEvent{Type: "command.error", Command: "link.request", Error: "data must be base64"})
			return
		}
		payload = data
	}
	timeout := time.Duration(cmd.TimeoutMs) * time.Millisecond
	if cmd.TimeoutMs <= 0 {
		timeout = 0
	}
	receipt, err := ls.link.Request(cmd.Path, payload, timeout)
	if err != nil {
		c.send(requestFailedEvent{
			Type:   "request.failed",
			LinkID: cmd.LinkID,
			Path:   cmd.Path,
			Error:  err.Error(),
		})
		return
	}
	linkID := cmd.LinkID
	path := cmd.Path
	receipt.SetResponseCallback(func(r *link.RequestReceipt) {
		c.session.broadcast(requestResponseEvent{
			Type:      "request.response",
			LinkID:    linkID,
			RequestID: hex.EncodeToString(r.GetRequestID()),
			Path:      path,
			Data:      base64.StdEncoding.EncodeToString(r.GetResponse()),
		})
	})
	receipt.SetFailedCallback(func(r *link.RequestReceipt) {
		c.session.broadcast(requestFailedEvent{
			Type:      "request.failed",
			LinkID:    linkID,
			RequestID: hex.EncodeToString(r.GetRequestID()),
			Path:      path,
			Error:     "request failed or timed out",
		})
	})
}

func (c *wsClient) handleLinkSendResource(raw []byte) {
	var cmd linkSendResourceCommand
	if err := json.Unmarshal(raw, &cmd); err != nil {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.send_resource", Error: "invalid command json"})
		return
	}
	ls, ok := c.session.getLink(cmd.LinkID)
	if !ok {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.send_resource", Error: "unknown link_id"})
		return
	}
	data, err := base64.StdEncoding.DecodeString(cmd.Data)
	if err != nil {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.send_resource", Error: "data must be base64"})
		return
	}
	res, err := resource.New(append([]byte(nil), data...), false)
	if err != nil {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.send_resource", Error: err.Error()})
		return
	}
	if cmd.Name != "" {
		if err := res.SetMetadata(map[string]any{"name": []byte(cmd.Name)}); err != nil {
			c.send(commandErrorEvent{Type: "command.error", Command: "link.send_resource", Error: err.Error()})
			return
		}
	}
	linkID := cmd.LinkID
	go func() {
		if err := ls.link.SendResource(res); err != nil {
			c.session.broadcast(resourceConcludedEvent{
				Type:    "resource.concluded",
				LinkID:  linkID,
				Name:    cmd.Name,
				Success: false,
				Error:   err.Error(),
			})
		}
	}()
}

func (c *wsClient) handleLinkIdentify(raw []byte) {
	var cmd linkIdentifyCommand
	if err := json.Unmarshal(raw, &cmd); err != nil {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.identify", Error: "invalid command json"})
		return
	}
	ls, ok := c.session.getLink(cmd.LinkID)
	if !ok {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.identify", Error: "unknown link_id"})
		return
	}
	if err := ls.link.Identify(c.session.identity); err != nil {
		c.send(commandErrorEvent{Type: "command.error", Command: "link.identify", Error: err.Error()})
	}
}