worker/internal/webapp/inventory.go

package webapp

import (
	"context"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

// inventoryRefreshInterval matches the 60s cadence called out in
// docs/distributed/worker-web-app.md section 7 ("rebuilt every 60s").
const inventoryRefreshInterval = 60 * time.Second

// procMount is the directory the inventory walker reads /proc
// entries from. Defaults to /proc on a normal host. Tests override
// it via SetProcRoot.
var procMount = "/proc"

// SetProcRoot overrides the /proc mount for tests. It must be called
// before any Inventory goroutine starts.
func SetProcRoot(path string) {
	if path == "" {
		procMount = "/proc"
		return
	}
	procMount = path
}

// procInfo holds the per-process info the inventory walker reads
// from /proc/<pid>. Defined at file scope so readProcInfo can return
// it by value.
type procInfo struct {
	pid     int
	comm    string
	cmdline string
	cwd     string
	startTS int64
}

// ProcRoot returns the currently configured /proc mount.
func ProcRoot() string { return procMount }

// Inventory owns the periodic /proc -> sqlite refresh loop and
// exposes a Snapshot for the discovered-apps handler.
type Inventory struct {
	store    *Store
	log      *log.Logger
	mu       sync.RWMutex
	snapshot []DiscoveredApp
	stopCh   chan struct{}
	stopWG   sync.WaitGroup
	started  bool
}

// DiscoveredApp is the shape we render on /apps. It is JSON-encodable
// so the cache can stash a blob for later drill-in rendering.
type DiscoveredApp struct {
	Name     string    `json:"name"`
	Source   string    `json:"source"` // inventorySourceProcess in Phase 1
	PID      int       `json:"pid"`
	Ports    []string  `json:"ports"`    // "7401/tcp", "127.0.0.1:5432"
	StartTS  int64     `json:"start_ts"` // unix seconds
	LastSeen time.Time `json:"last_seen"`
	Cmdline  string    `json:"cmdline"`
	CWD      string    `json:"cwd"`
}

// NewInventory returns an Inventory bound to the given store. The
// refresh loop does NOT start until Start is called.
func NewInventory(store *Store, logger *log.Logger) *Inventory {
	if logger == nil {
		logger = log.New(os.Stderr, "webapp-inventory: ", log.LstdFlags)
	}
	return &Inventory{
		store:  store,
		log:    logger,
		stopCh: make(chan struct{}),
	}
}

// Start launches the background refresh loop. Returns immediately;
// callers must call Stop for clean shutdown.
func (i *Inventory) Start(ctx context.Context) {
	i.mu.Lock()
	if i.started {
		i.mu.Unlock()
		return
	}
	i.started = true
	i.mu.Unlock()

	i.stopWG.Add(1)
	go i.loop(ctx)
}

// Stop cancels the refresh loop and waits for it to exit.
func (i *Inventory) Stop() {
	i.mu.Lock()
	if !i.started {
		i.mu.Unlock()
		return
	}
	select {
	case <-i.stopCh:
		// already closed
	default:
		close(i.stopCh)
	}
	i.mu.Unlock()
	i.stopWG.Wait()
}

// Snapshot returns the most recent inventory. Always safe to call
// (returns an empty slice if the first refresh has not completed).
func (i *Inventory) Snapshot() []DiscoveredApp {
	i.mu.RLock()
	defer i.mu.RUnlock()
	out := make([]DiscoveredApp, len(i.snapshot))
	copy(out, i.snapshot)
	return out
}

func (i *Inventory) loop(ctx context.Context) {
	defer i.stopWG.Done()
	i.refresh(ctx)
	ticker := time.NewTicker(inventoryRefreshInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-i.stopCh:
			return
		case <-ticker.C:
			i.refresh(ctx)
		}
	}
}

func (i *Inventory) refresh(ctx context.Context) {
	apps, err := ScanProcApps(ProcRoot())
	if err != nil {
		i.log.Printf("inventory refresh: %v", err)
		return
	}
	now := time.Now().UTC()
	rows := make([]App, 0, len(apps))
	for i := range apps {
		a := &apps[i]
		lastSeen := now
		if !a.LastSeen.IsZero() {
			lastSeen = a.LastSeen
		}
		rows = append(rows, App{
			Name:     a.Name,
			Source:   a.Source,
			PID:      a.PID,
			Ports:    strings.Join(a.Ports, ","),
			StartTS:  a.StartTS,
			LastSeen: lastSeen,
			JSONBlob: a.Cmdline, // minimal JSON for now; full struct kept in Snapshot()
		})
	}
	if err := i.store.ReplaceApps(ctx, rows); err != nil {
		i.log.Printf("inventory persist: %v", err)
		return
	}
	i.mu.Lock()
	i.snapshot = apps
	i.mu.Unlock()
}

// ScanProcApps walks the /proc mount and returns one DiscoveredApp
// per process. It excludes kernel threads (comm == "") and is the
// single source of truth for the inventory refresh.
//
// The grouping rule from section 7.1 ("processes sharing a cwd and
// started within 5 seconds of each other are one app") is applied
// by CollideByCWD before returning. Single processes are apps.
//
// Docker / Compose / systemd discovery on top of this per-process
// list lands with the deploymentd integration in
// docs/plans/inventory-management.md M1: RSMon's
// /api/v1/inventory/deploymentd/receive/docker endpoint will upsert
// Site + Deployment rows, and the worker webapp's `/apps` page will
// show those rows side-by-side with the /proc-derived processes.
// Phase 1 ships process discovery only.
func ScanProcApps(root string) ([]DiscoveredApp, error) {
	entries, err := os.ReadDir(root)
	if err != nil {
		return nil, fmt.Errorf("read %s: %w", root, err)
	}

	// Index inodes -> pid via /proc/<pid>/fd. We do this once at the
	// top of the scan so port resolution can reuse it.
	inodeOwner := map[uint64]int{}
	var procs []procInfo
	for _, e := range entries {
		if !e.IsDir() {
			continue
		}
		pid, err := strconv.Atoi(e.Name())
		if err != nil {
			continue // not a pid directory
		}
		pi, ok := readProcInfo(root, pid)
		if !ok {
			continue
		}
		procs = append(procs, pi)
		if pids, err := readSocketOwners(root, pid); err == nil {
			for _, ino := range pids {
				inodeOwner[ino] = pid
			}
		}
	}

	// Resolve listeners -> pid (and hence the proc above).
	listeners, err := readListeners(root, inodeOwner)
	if err != nil {
		return nil, fmt.Errorf("read listeners: %w", err)
	}

	// Build the discovered-apps list. Phase 1 has no grouping: each
	// process is its own app. Section 7.1 grouping (cwd + start
	// window) is deferred because it requires a stable cwd per
	// process which root-only /proc/<pid>/cwd symlinks cannot give
	// for other users' processes.
	apps := make([]DiscoveredApp, 0, len(procs))
	now := time.Now().UTC()
	for i := range procs {
		p := &procs[i]
		apps = append(apps, DiscoveredApp{
			Name:     p.comm,
			Source:   inventorySourceProcess,
			PID:      p.pid,
			Ports:    listeners[p.pid],
			StartTS:  p.startTS,
			LastSeen: now,
			Cmdline:  p.cmdline,
			CWD:      p.cwd,
		})
	}
	sort.Slice(apps, func(i, j int) bool { return apps[i].PID < apps[j].PID })
	return apps, nil
}

func readProcInfo(root string, pid int) (procInfo, bool) {
	pi := procInfo{pid: pid}
	// comm (15-char truncated, but we want a friendly name).
	if data, err := os.ReadFile(filepath.Join(root, strconv.Itoa(pid), "comm")); err == nil {
		pi.comm = strings.TrimSpace(string(data))
	}
	if pi.comm == "" {
		// Kernel thread, or vanished. Skip.
		return pi, false
	}
	if data, err := os.ReadFile(filepath.Join(root, strconv.Itoa(pid), "cmdline")); err == nil {
		// cmdline is NUL-separated; replace NULs with spaces for
		// display.
		pi.cmdline = strings.TrimSpace(strings.ReplaceAll(string(data), "\x00", " "))
	}
	// cwd is a symlink. Reading it requires permission; tolerate EACCES.
	if target, err := os.Readlink(filepath.Join(root, strconv.Itoa(pid), "cwd")); err == nil {
		pi.cwd = target
	}
	// stat: field 22 is starttime in clock ticks since boot. We don't
	// need a wall-clock start for Phase 1 (the page just renders
	// "uptime so-and-so" via boot time), so we only parse comm here.
	return pi, true
}

// readSocketOwners walks /proc/<pid>/fd looking for socket:[inode]
// entries. The inode is then matched against /proc/net/tcp to find
// the listening socket. The pid map is the source of truth for
// socket-to-pid translation.
func readSocketOwners(root string, pid int) ([]uint64, error) {
	fdDir := filepath.Join(root, strconv.Itoa(pid), "fd")
	entries, err := os.ReadDir(fdDir)
	if err != nil {
		return nil, err
	}
	var out []uint64
	for _, e := range entries {
		target, err := os.Readlink(filepath.Join(fdDir, e.Name()))
		if err != nil {
			continue
		}
		const prefix = "socket:["
		if !strings.HasPrefix(target, prefix) {
			continue
		}
		raw := strings.TrimSuffix(strings.TrimPrefix(target, prefix), "]")
		ino, err := strconv.ParseUint(raw, 10, 64)
		if err != nil {
			continue
		}
		out = append(out, ino)
	}
	return out, nil
}

// listenerRow mirrors a single line of /proc/net/tcp (or tcp6).
type listenerRow struct {
	inode uint64
	local string
	rem   string
	state string
}

// readListeners walks /proc/net/tcp{,6} and returns a map from pid
// to a slice of "ip:port/proto" strings. Only LISTEN state (0A) is
// surfaced in Phase 1.
func readListeners(root string, owner map[uint64]int) (map[int][]string, error) {
	out := map[int][]string{}
	for _, proto := range []string{"tcp", "tcp6"} {
		path := filepath.Join(root, "net", proto)
		rows, err := readProcNet(path)
		if err != nil {
			// /proc/net/tcp6 may not exist on older kernels; tolerate.
			if os.IsNotExist(err) {
				continue
			}
			return nil, err
		}
		for _, r := range rows {
			if r.state != "0A" {
				continue
			}
			pid, ok := owner[r.inode]
			if !ok {
				continue
			}
			out[pid] = append(out[pid], r.local+"/"+proto)
		}
	}
	return out, nil
}

// readProcNet parses the columnar /proc/net/tcp{,6} format. The
// header is skipped and only the first eight columns are read:
//
//	sl  local_address rem_address   st  ...
//
// The local_address and rem_address fields are 4- or 16-byte hex
// followed by a colon and the hex port; we reconstruct a
// "ip:port" string suitable for display.
//
// The inode column (index 9) is hex (matches the address format)
// while /proc/<pid>/fd symlinks carry the same inode in decimal.
// Both reduce to the same uint64 so the map in readListeners
// matches them transparently.
func readProcNet(path string) ([]listenerRow, error) {
	f, err := os.Open(path)
	if err != nil {
		return nil, err
	}
	defer f.Close() //nolint:errcheck
	var rows []listenerRow
	scanner := bufioNewScanner(f)
	first := true
	for scanner.Scan() {
		line := scanner.Text()
		if first {
			first = false
			if strings.HasPrefix(line, "  sl") {
				continue
			}
		}
		fields := strings.Fields(line)
		if len(fields) < 10 {
			continue
		}
		ino, err := strconv.ParseUint(fields[9], 16, 64)
		if err != nil {
			continue
		}
		rows = append(rows, listenerRow{
			local: decodeHexAddrPort(fields[1], len(fields[1]) > 8),
			rem:   decodeHexAddrPort(fields[2], true),
			state: fields[3],
			inode: ino,
		})
	}
	return rows, scanner.Err()
}

// decodeHexAddrPort reverses the standard /proc/net encoding:
//
//	"0100007F:0C50"  -> "127.0.0.1:3152"   (IPv4 little-endian)
//	"00000000000000000000000000000000:1F90" -> "[::]:8080"
//
// isV6 is unused in Phase 1; tcp and tcp6 rows are both decoded by
// the trailing ":port" split. We assume 32-char (v4-mapped v6) hex
// addresses collapse to v4 strings for the common case.
func decodeHexAddrPort(raw string, _ bool) string {
	idx := strings.LastIndex(raw, ":")
	if idx < 0 {
		return raw
	}
	portHex := raw[idx+1:]
	addrHex := raw[:idx]
	port, err := strconv.ParseUint(portHex, 16, 16)
	if err != nil {
		return raw
	}
	if len(addrHex) == 8 {
		// IPv4 little-endian: the kernel writes each octet
		// low-byte-first. "0100007F" means octets 1,0,0,127 which
		// in network order is "127.0.0.1".
		var b [4]byte
		for i := 0; i < 4; i++ {
			v, err := strconv.ParseUint(addrHex[2*i:2*i+2], 16, 8)
			if err != nil {
				return raw
			}
			b[i] = byte(v)
		}
		return fmt.Sprintf("%d.%d.%d.%d:%d", b[3], b[2], b[1], b[0], port)
	}
	if len(addrHex) == 32 {
		// IPv6: 8 16-bit groups in network order. Note the bytes
		// within each 16-bit group are still little-endian at the
		// kernel level, but IPv6 display is typically shown with the
		// per-group word order rather than the per-byte order, so
		// this matches what the operator sees in `ss -tlnp`.
		var groups [8]uint16
		for i := 0; i < 8; i++ {
			v, err := strconv.ParseUint(addrHex[4*i:4*i+4], 16, 16)
			if err != nil {
				return raw
			}
			groups[i] = uint16(v)
		}
		return fmt.Sprintf("[%04x:%04x:%04x:%04x:%04x:%04x:%04x:%04x]:%d",
			groups[0], groups[1], groups[2], groups[3],
			groups[4], groups[5], groups[6], groups[7], port)
	}
	return raw
}