Lora-Scanner/internal/gateway/scraper.go

package gateway

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net"
	"net/http"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/PuerkitoBio/goquery"
	"github.com/skinnyrad/Lora-Scanner/internal/frames"
	"github.com/skinnyrad/Lora-Scanner/internal/hub"
)

const (
	// A dead gateway fails fast, but a slow one (the log page is large and
	// the gateway's CGI is slow) gets time to finish sending it.
	dialTimeout    = 5 * time.Second
	headerTimeout  = 10 * time.Second
	requestTimeout = 45 * time.Second
	// Offline gateways are retried this often until they answer again.
	RetryInterval = 5 * time.Second
	// A gateway is only marked offline after this many failed polls in a row,
	// so one slow or dropped response doesn't flip it.
	FailThreshold = 2
	tickEvery     = time.Second
)

var (
	rssiRe      = regexp.MustCompile(`"Rssi":(-?\d+)`)
	hiddenStyle = "display: none;"
)

// Row is one packet parsed from the gateway's log-traffic page.
type Row struct {
	DevID string
	Freq  float64
	RSSI  *int
	Text  string
	// Detail is the hidden row's JSON-ish text (address, size, RSSI, SNR, MIC...).
	Detail string
}

// Parse extracts packet rows from a Dragino /cgi-bin/log-traffic.has page.
// Visible rows are summaries; each is followed by a hidden row holding JSON-ish
// details (not valid JSON, hence the regex for RSSI).
func Parse(r io.Reader) ([]Row, error) {
	doc, err := goquery.NewDocumentFromReader(r)
	if err != nil {
		return nil, err
	}
	table := doc.Find("table").First()
	var rows []Row
	table.Find("tr").Each(func(i int, tr *goquery.Selection) {
		if i == 0 || strings.Contains(tr.AttrOr("style", ""), hiddenStyle) {
			return
		}
		var cells []string
		tr.Find("td").Each(func(_ int, td *goquery.Selection) {
			cells = append(cells, strings.TrimSpace(td.Text()))
		})
		if len(cells) < 2 {
			return
		}
		fields := cells[1:] // first cell is the chevron
		if len(fields) < 4 {
			return
		}
		freq, err := strconv.ParseFloat(strings.TrimSpace(fields[3]), 64)
		if err != nil {
			return
		}
		devID := extractDevID(fields[len(fields)-1])
		if devID == "" {
			return
		}
		row := Row{DevID: devID, Freq: freq, Text: strings.Join(fields, " | ")}
		hidden := tr.NextFiltered("tr")
		if hidden.Length() > 0 && strings.Contains(hidden.AttrOr("style", ""), hiddenStyle) {
			row.Detail = strings.Join(strings.Fields(hidden.Text()), " ")
			if m := rssiRe.FindStringSubmatch(hidden.Text()); m != nil {
				if v, err := strconv.Atoi(m[1]); err == nil {
					row.RSSI = &v
				}
			}
		}
		rows = append(rows, row)
	})
	return rows, nil
}

// extractDevID turns "Dev Addr: 27FD3566, Size: 20" into "27FD3566".
func extractDevID(content string) string {
	first := strings.SplitN(content, ",", 2)[0]
	parts := strings.Split(first, ":")
	return strings.TrimSpace(parts[len(parts)-1])
}

// Gateway states reported to the UI.
const (
	StateUnknown = "unknown" // not polled yet
	StateOnline  = "online"  // last poll succeeded
	StateOffline = "offline" // last poll failed
)

// GatewayStatus is what the last poll of a gateway found.
type GatewayStatus struct {
	State       string     `json:"state"`
	Error       string     `json:"error,omitempty"`
	LastSuccess *time.Time `json:"lastSuccess"`
}

// StatusMsg is broadcast on the hub whenever any gateway changes state.
type StatusMsg struct {
	Type     string                   `json:"type"` // "gateways"
	Gateways map[string]GatewayStatus `json:"gateways"`
}

type gwState struct {
	GatewayStatus
	lastFinish time.Time
	running    bool
	fails      int // consecutive failed polls
}

type Scraper struct {
	Config *Config
	Frames *frames.Store
	Hub    *hub.Hub // optional: receives StatusMsg on state changes
	User   string
	Pass   string
	Client *http.Client
	URLFor func(ip string) string

	mu    sync.Mutex // guards seen and states
	seen  map[string]struct{}
	state map[string]*gwState

	retry time.Duration // offline retry cadence (RetryInterval; tests shorten it)
	tick  time.Duration
}

func NewScraper(cfg *Config, f *frames.Store, user, pass string) *Scraper {
	return &Scraper{
		Config: cfg, Frames: f, User: user, Pass: pass,
		Client: &http.Client{
			Timeout: requestTimeout,
			Transport: &http.Transport{
				// A host that silently drops packets must fail fast, not after the full request timeout.
				DialContext:           (&net.Dialer{Timeout: dialTimeout}).DialContext,
				ResponseHeaderTimeout: headerTimeout,
			},
		},
		URLFor: func(ip string) string { return fmt.Sprintf("http://%s:8000/cgi-bin/log-traffic.has", ip) },
		seen:   map[string]struct{}{},
		state:  map[string]*gwState{},
		retry:  RetryInterval,
		tick:   tickEvery,
	}
}

// Run polls until ctx ends. Each gateway has its own schedule: every
// Config.PollInterval() while online, every RetryInterval while offline or
// confirming a failed poll. A gateway whose previous scrape is still running is skipped.
func (s *Scraper) Run(ctx context.Context) {
	t := time.NewTicker(s.tick)
	defer t.Stop()
	for {
		s.launch(ctx, false)
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
	}
}

// ScrapeAll scrapes every idle gateway now, regardless of schedule, and waits.
func (s *Scraper) ScrapeAll(ctx context.Context) { s.launch(ctx, true)() }

// launch starts a scrape for each gateway that is due (or all idle ones when
// force is set) and returns a func that waits for them.
func (s *Scraper) launch(ctx context.Context, force bool) (wait func()) {
	ips := s.Config.List()
	interval := s.Config.PollInterval()
	now := time.Now()

	s.mu.Lock()
	configured := make(map[string]bool, len(ips))
	for _, ip := range ips {
		configured[ip] = true
	}
	for ip, st := range s.state {
		if !configured[ip] && !st.running {
			delete(s.state, ip) // removed from the list
		}
	}
	var start []string
	for _, ip := range ips {
		st := s.state[ip]
		if st == nil {
			st = &gwState{GatewayStatus: GatewayStatus{State: StateUnknown}}
			s.state[ip] = st
		}
		if st.running {
			continue
		}
		wait := interval
		if st.State != StateOnline || st.fails > 0 { // offline, or confirming a failure
			wait = s.retry
		}
		if force || st.lastFinish.IsZero() || now.Sub(st.lastFinish) >= wait {
			st.running = true
			start = append(start, ip)
		}
	}
	s.mu.Unlock()

	var wg sync.WaitGroup
	for _, ip := range start {
		wg.Add(1)
		go func(ip string) {
			defer wg.Done()
			began := time.Now()
			s.finish(ip, s.scrapeOne(ctx, ip), time.Since(began))
		}(ip)
	}
	return wg.Wait
}

// finish records a poll result and announces state changes.
func (s *Scraper) finish(ip string, err error, took time.Duration) {
	s.mu.Lock()
	st := s.state[ip]
	if st == nil { // removed while the scrape was running
		s.mu.Unlock()
		return
	}
	prev := st.State
	st.running = false
	st.lastFinish = time.Now()
	if err != nil {
		st.fails++
		// One failure is only a warning: the state flips after FailThreshold in a row.
		if st.fails >= FailThreshold {
			st.State, st.Error = StateOffline, err.Error()
		}
	} else {
		now := time.Now()
		st.fails = 0
		st.State, st.Error, st.LastSuccess = StateOnline, "", &now
	}
	cur := st.State
	changed := prev != cur
	s.mu.Unlock()

	if !changed {
		return
	}
	if cur == StateOffline {
		fmt.Printf("gateway %s offline after %d failed polls: %v (last took %s); retrying every %s\n",
			ip, FailThreshold, err, took.Round(time.Millisecond), s.retry)
	} else if prev == StateOffline {
		fmt.Printf("gateway %s back online\n", ip)
	}
	if s.Hub != nil {
		s.Hub.Broadcast(s.StatusMsg())
	}
}

// Statuses reports the last poll result for every configured gateway.
func (s *Scraper) Statuses() map[string]GatewayStatus {
	ips := s.Config.List()
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make(map[string]GatewayStatus, len(ips))
	for _, ip := range ips {
		if st := s.state[ip]; st != nil {
			out[ip] = st.GatewayStatus
		} else {
			out[ip] = GatewayStatus{State: StateUnknown}
		}
	}
	return out
}

func (s *Scraper) StatusMsg() StatusMsg { return StatusMsg{Type: "gateways", Gateways: s.Statuses()} }

func (s *Scraper) request(ctx context.Context, method, ip string) (*http.Response, error) {
	req, err := http.NewRequestWithContext(ctx, method, s.URLFor(ip), nil)
	if err != nil {
		return nil, err
	}
	req.SetBasicAuth(s.User, s.Pass)
	return s.Client.Do(req)
}

func (s *Scraper) scrapeOne(ctx context.Context, ip string) error {
	resp, err := s.request(ctx, http.MethodGet, ip)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("status %s", resp.Status)
	}
	// Read the whole page first so a stalled download is reported as such and
	// nothing is half-parsed; the next poll sees the same rows again (deduped).
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return fmt.Errorf("reading page (got %d bytes): %w", len(body), err)
	}
	rows, err := Parse(bytes.NewReader(body))
	if err != nil {
		return err
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	// Oldest first so survey order is chronological (the page lists newest first).
	for i := len(rows) - 1; i >= 0; i-- {
		r := rows[i]
		key := fmt.Sprintf("%s_%v_%s", r.DevID, r.Freq, r.Text)
		if _, dup := s.seen[key]; dup {
			continue
		}
		s.seen[key] = struct{}{}
		raw := r.Text
		if r.Detail != "" {
			raw += "\n" + r.Detail
		}
		s.Frames.Add(frames.Frame{
			Kind: frames.KindLoRaWAN, Source: ip, Identifier: r.DevID,
			Freq: r.Freq, RSSI: r.RSSI, Info: r.Text, Raw: raw,
		})
	}
	return nil
}