macker/server/main.go

package main

import (
	"context"
	_ "embed"
	"encoding/gob"
	"encoding/json"
	"flag"
	"io"
	"log"
	"math"
	"net"
	"net/http"
	"os"
	"sync"
	"time"

	"macker/datapoint"
)

//go:embed index.html
var page []byte

type location struct {
	Point      datapoint.DataPoint `json:"point"`
	UpdatedAt  time.Time           `json:"updated_at"`
	ReceivedAt time.Time           `json:"received_at"`
	Nearby     addressResult       `json:"nearby"`
}

type tracker struct {
	mu        sync.RWMutex
	latest    *location
	geocoder  *geocoder
	addresses *addressIndex
}

func (t *tracker) accept(points []datapoint.DataPoint, now time.Time) {
	t.mu.Lock()
	defer t.mu.Unlock()
	for _, p := range points {
		if math.IsNaN(p.Lat) || math.IsInf(p.Lat, 0) || math.IsNaN(p.Long) || math.IsInf(p.Long, 0) || p.Lat < -90 || p.Lat > 90 || p.Long < -180 || p.Long > 180 {
			continue
		}
		if p.Speed != nil && (math.IsNaN(*p.Speed) || math.IsInf(*p.Speed, 0) || *p.Speed < 0) {
			p.Speed = nil
		}
		updated := now
		if p.Timestamp != 0 {
			if p.Timestamp > uint64(now.Add(time.Minute).UnixMilli()) {
				continue
			}
			updated = time.UnixMilli(int64(p.Timestamp))
		}
		if t.latest != nil && !updated.After(t.latest.UpdatedAt) {
			continue
		}
		t.latest = &location{Point: p, UpdatedAt: updated, ReceivedAt: now}
	}
}

func (t *tracker) receive(c net.Conn) {
	defer c.Close()
	_ = c.SetReadDeadline(time.Now().Add(10 * time.Second))
	var raw []byte
	if err := gob.NewDecoder(io.LimitReader(c, datapoint.MaxBatchBytes+1024)).Decode(&raw); err != nil {
		log.Printf("decode batch: %v", err)
		return
	}
	points, err := datapoint.UnzipBatch(raw)
	if err != nil {
		log.Printf("decode points: %v", err)
		return
	}
	t.accept(points, time.Now().UTC())
}

func (t *tracker) handler() http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("GET /api/location", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		w.Header().Set("Cache-Control", "no-store")
		t.mu.RLock()
		if t.latest == nil {
			t.mu.RUnlock()
			_ = json.NewEncoder(w).Encode(nil)
			return
		}
		snapshot := *t.latest
		t.mu.RUnlock()
		if t.addresses != nil {
			snapshot.Nearby = t.addresses.view(snapshot.Point.Lat, snapshot.Point.Long)
		} else {
			snapshot.Nearby = t.geocoder.view(snapshot.Point.Lat, snapshot.Point.Long, time.Now())
		}
		_ = json.NewEncoder(w).Encode(snapshot)
	})
	mux.HandleFunc("GET /{$}", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/html; charset=utf-8")
		_, _ = w.Write(page)
	})
	return mux
}

func main() {
	ingest := flag.String("ingest", ":2948", "TCP address for car GPS batches")
	web := flag.String("http", ":8080", "HTTP address for the map")
	geocodeURL := flag.String("geocoder", os.Getenv("MACKER_GEOCODER_URL"), "Nominatim-compatible reverse geocoding endpoint (empty disables address lookups)")
	addressFile := flag.String("addresses", "data/mississippi-addresses.json", "local OSM address index (takes precedence over geocoder)")
	importOSM := flag.String("import-osm", "", "import address nodes/ways from OSM XML into -addresses, then exit")
	flag.Parse()
	if *importOSM != "" {
		count, err := buildAddresses(*importOSM, *addressFile)
		if err != nil {
			log.Fatal(err)
		}
		log.Printf("Saved %d local addresses to %s", count, *addressFile)
		return
	}
	var local *addressIndex
	if *addressFile != "" {
		var err error
		local, err = loadAddresses(*addressFile)
		if err != nil && !os.IsNotExist(err) {
			log.Fatal(err)
		}
		if os.IsNotExist(err) {
			log.Printf("Local address data missing: run scripts/prepare-addresses.sh, then restart")
		}
		if local != nil {
			log.Printf("Loaded %d local addresses", local.count)
		}
	}
	var gc *geocoder
	var err error
	if local == nil {
		gc, err = newGeocoder(*geocodeURL, os.Getenv("MACKER_GEOCODER_KEY"))
	}
	if err != nil {
		log.Fatal(err)
	}
	ln, err := net.Listen("tcp", *ingest)
	if err != nil {
		log.Fatal(err)
	}
	defer ln.Close()
	t := &tracker{geocoder: gc, addresses: local}
	if gc != nil {
		go t.runGeocoder(context.Background())
	}
	go func() {
		// Bound concurrent decoders and their memory usage.
		slots := make(chan struct{}, 32)
		for {
			c, err := ln.Accept()
			if err != nil {
				return
			}
			select {
			case slots <- struct{}{}:
				go func() { defer func() { <-slots }(); t.receive(c) }()
			default:
				c.Close()
			}
		}
	}()
	log.Printf("GPS receiver on %s; map on %s", *ingest, *web)
	srv := &http.Server{Addr: *web, Handler: t.handler(), ReadHeaderTimeout: 5 * time.Second, WriteTimeout: 15 * time.Second, IdleTimeout: 60 * time.Second}
	log.Fatal(srv.ListenAndServe())
}