atc_hooks.go

package netpipe

import (
	"fmt"
	"time"
)

// ---------------------------------------------------------------------------
// Server ATC integration
// ---------------------------------------------------------------------------

// startATC initializes the ATC addon for a Server. Called from Listen() if enabled.
func (s *Server) startATC() {
	if !s.config.EnableATC {
		return
	}

	port := s.config.ATCPort
	if port == 0 {
		port = s.config.Port + 1 // default: data port + 1
	}

	s.atcBus = newATCBus(true)

	snapshotFunc := func() ATCSnapshot {
		clients := s.GetClients()
		nodes := make([]ATCNode, 0, len(clients))
		for _, id := range clients {
			nodes = append(nodes, ATCNode{
				ID:    id,
				Label: id[:8],
			})
		}
		return ATCSnapshot{
			Type:       ATCEventSnapshot,
			Timestamp:  time.Now().UnixMilli(),
			Mode:       "server",
			ServerPort: s.config.Port,
			Protocol:   s.config.Protocol,
			Nodes:      nodes,
			Encrypted:  s.config.EncryptionKey != "",
		}
	}

	srv := newATCServer(s.atcBus, port, snapshotFunc)
	srv.start()

	s.atcBus.emit(ATCEvent{
		Type:    ATCEventServerUp,
		Mode:    "server",
		Message: fmt.Sprintf("server listening on %s :%d", s.config.Protocol, s.config.Port),
	})
}

// atcEmitConnect emits a connect event to ATC.
func (s *Server) atcEmitConnect(clientID string) {
	if s.atcBus == nil {
		return
	}
	s.atcBus.emit(ATCEvent{
		Type:    ATCEventConnect,
		NodeID:  clientID,
		Mode:    "server",
		Message: fmt.Sprintf("client %s connected", clientID[:8]),
	})
}

// atcEmitDisconnect emits a disconnect event to ATC.
func (s *Server) atcEmitDisconnect(clientID string) {
	if s.atcBus == nil {
		return
	}
	s.atcBus.emit(ATCEvent{
		Type:    ATCEventDisconnect,
		NodeID:  clientID,
		Mode:    "server",
		Message: fmt.Sprintf("client %s disconnected", clientID[:8]),
	})
}

// atcEmitDataRecv emits a data-received event to ATC.
func (s *Server) atcEmitDataRecv(clientID string, size int) {
	if s.atcBus == nil {
		return
	}
	s.atcBus.emit(ATCEvent{
		Type:    ATCEventDataRecv,
		NodeID:  clientID,
		Bytes:   size,
		Mode:    "server",
		Message: fmt.Sprintf("← %d bytes from %s", size, clientID[:8]),
	})
}

// atcEmitDataSend emits a data-sent event to ATC.
func (s *Server) atcEmitDataSend(targetID string, size int) {
	if s.atcBus == nil {
		return
	}
	s.atcBus.emit(ATCEvent{
		Type:     ATCEventDataSend,
		TargetID: targetID,
		Bytes:    size,
		Mode:     "server",
		Message:  fmt.Sprintf("→ %d bytes to %s", size, targetID[:8]),
	})
}

// ---------------------------------------------------------------------------
// Peer ATC integration
// ---------------------------------------------------------------------------

// startATC initializes the ATC addon for a Peer. Called from Listen() if enabled.
func (p *Peer) startATC() {
	if !p.config.EnableATC {
		return
	}

	port := p.config.ATCPort
	if port == 0 {
		port = p.config.Port + 1
	}

	p.atcBus = newATCBus(true)

	snapshotFunc := func() ATCSnapshot {
		peers := p.GetPeers()
		nodes := make([]ATCNode, 0, len(peers))
		for _, id := range peers {
			nodes = append(nodes, ATCNode{
				ID:    id,
				Label: id[:8],
			})
		}
		return ATCSnapshot{
			Type:       ATCEventSnapshot,
			Timestamp:  time.Now().UnixMilli(),
			Mode:       "p2p",
			ServerPort: p.config.Port,
			Protocol:   p.config.Protocol,
			Nodes:      nodes,
			Encrypted:  !p.config.NoEncrypt,
			SelfID:     p.peerID,
		}
	}

	srv := newATCServer(p.atcBus, port, snapshotFunc)
	srv.start()

	p.atcBus.emit(ATCEvent{
		Type:    ATCEventServerUp,
		Mode:    "p2p",
		NodeID:  p.peerID,
		Message: fmt.Sprintf("peer %s listening on :%d", p.peerID[:8], p.config.Port),
	})
}

// atcEmitConnect emits a connect event for P2P.
func (p *Peer) atcEmitConnect(peerID string) {
	if p.atcBus == nil {
		return
	}
	p.atcBus.emit(ATCEvent{
		Type:    ATCEventConnect,
		NodeID:  peerID,
		Mode:    "p2p",
		Message: fmt.Sprintf("peer %s joined mesh", peerID[:8]),
	})
}

// atcEmitDisconnect emits a disconnect event for P2P.
func (p *Peer) atcEmitDisconnect(peerID string) {
	if p.atcBus == nil {
		return
	}
	p.atcBus.emit(ATCEvent{
		Type:    ATCEventDisconnect,
		NodeID:  peerID,
		Mode:    "p2p",
		Message: fmt.Sprintf("peer %s left mesh", peerID[:8]),
	})
}

// atcEmitDataSend emits a data-sent event for P2P.
func (p *Peer) atcEmitDataSend(peerID string, size int) {
	if p.atcBus == nil {
		return
	}
	p.atcBus.emit(ATCEvent{
		Type:     ATCEventDataSend,
		NodeID:   p.peerID,
		TargetID: peerID,
		Bytes:    size,
		Mode:     "p2p",
		Message:  fmt.Sprintf("→ %d bytes to %s", size, peerID[:8]),
	})
}

// atcEmitDataRecv emits a data-received event for P2P.
func (p *Peer) atcEmitDataRecv(peerID string, size int) {
	if p.atcBus == nil {
		return
	}
	p.atcBus.emit(ATCEvent{
		Type:    ATCEventDataRecv,
		NodeID:  peerID,
		Bytes:   size,
		Mode:    "p2p",
		Message: fmt.Sprintf("← %d bytes from %s", size, peerID[:8]),
	})
}