lingniu-vehicle-ingest/vehicle-data-platform/apps/api/internal/dailygeo/worker.go

package dailygeo

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"math"
	"regexp"
	"sync"
	"time"
)

var Shanghai = time.FixedZone("Asia/Shanghai", 8*3600)
var identifier = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`)

type Point struct {
	VIN, Protocol       string
	Time                time.Time
	Longitude, Latitude float64
}
type Worker struct {
	RequestInterval time.Duration
	DB, TD          *sql.DB
	Database        string
	Resolver        Resolver
	Concurrency     int
	requestMu       sync.Mutex
	nextRequest     time.Time
}

func New(db, td *sql.DB, database string, r Resolver) *Worker {
	return &Worker{DB: db, TD: td, Database: database, Resolver: r, Concurrency: 3}
}
func validPoint(p Point) bool {
	return p.VIN != "" && !(p.Longitude == -0.999999 && p.Latitude == -0.999999) && !p.Time.IsZero() && p.Longitude != 0 && p.Latitude != 0 && !math.IsNaN(p.Longitude) && !math.IsNaN(p.Latitude) && math.Abs(p.Longitude) <= 180 && math.Abs(p.Latitude) <= 90
}
func newer(a, b Point) bool {
	return a.Time.After(b.Time) || (a.Time.Equal(b.Time) && a.Protocol < b.Protocol)
}

// Last() is applied only to non-null valid coordinates, partitioned by VIN and
// protocol. All selected columns therefore refer to the same final source row.
func (w *Worker) RefreshDay(ctx context.Context, date string) (int, error) {
	day, e := time.ParseInLocation("2006-01-02", date, Shanghai)
	if e != nil {
		return 0, e
	}
	if w.TD == nil || !identifier.MatchString(w.Database) {
		return 0, fmt.Errorf("historical location store unavailable")
	}
	q := fmt.Sprintf(`SELECT vin,protocol,LAST(ts),LAST(longitude),LAST(latitude) FROM %s.vehicle_locations WHERE ts >= '%s' AND ts < '%s' AND NOT (longitude = -0.999999 AND latitude = -0.999999) AND longitude <> 0 AND longitude BETWEEN -180 AND 180 AND latitude <> 0 AND latitude BETWEEN -90 AND 90 PARTITION BY vin,protocol`, w.Database, day.Format(time.RFC3339), day.AddDate(0, 0, 1).Format(time.RFC3339))
	rows, e := w.TD.QueryContext(ctx, q)
	if e != nil {
		return 0, e
	}
	points := map[string]Point{}
	for rows.Next() {
		var p Point
		if e = rows.Scan(&p.VIN, &p.Protocol, &p.Time, &p.Longitude, &p.Latitude); e != nil {
			rows.Close()
			return 0, e
		}
		if !validPoint(p) || p.Time.In(Shanghai).Format("2006-01-02") != date {
			continue
		}
		if old, ok := points[p.VIN]; !ok || newer(p, old) {
			points[p.VIN] = p
		}
	}
	e = rows.Err()
	rows.Close()
	if e != nil {
		return 0, e
	}
	// Repair the device's invalid-location sentinel before applying real points.
	// Clearing its timestamp allows an earlier valid point from this day to win.
	_, e = w.DB.ExecContext(ctx, `UPDATE vehicle_daily_geography SET longitude=NULL,latitude=NULL,location_time=NULL,province='',city='',region='',adcode='',status='NO_LOCATION',attempts=0,last_error='',next_attempt_at=NULL,resolved_at=NULL WHERE stat_date=? AND longitude=-0.999999 AND latitude=-0.999999`, date)
	if e != nil {
		return 0, e
	}
	for _, p := range points {
		if e = w.SavePoint(ctx, p); e != nil {
			return 0, e
		}
	}
	// A successful historical lookup with no point is distinct from an unavailable
	// history service. Never record NO_LOCATION when the lookup itself failed.
	_, e = w.DB.ExecContext(ctx, `INSERT IGNORE INTO vehicle_daily_geography(vin,stat_date,status) SELECT vin,stat_date,'NO_LOCATION' FROM vehicle_open_daily_energy WHERE energy_type='HYDROGEN' AND stat_date=?`, date)
	return len(points), e
}

const savePointSQL = `INSERT INTO vehicle_daily_geography(vin,stat_date,longitude,latitude,location_time,source_protocol,status) VALUES(?,?,?,?,?,?,'PENDING') ON DUPLICATE KEY UPDATE
 province=IF((location_time IS NULL OR VALUES(location_time)>location_time) AND (longitude IS NULL OR longitude<>VALUES(longitude) OR latitude<>VALUES(latitude)),'',province),
 city=IF((location_time IS NULL OR VALUES(location_time)>location_time) AND (longitude IS NULL OR longitude<>VALUES(longitude) OR latitude<>VALUES(latitude)),'',city),
 region=IF((location_time IS NULL OR VALUES(location_time)>location_time) AND (longitude IS NULL OR longitude<>VALUES(longitude) OR latitude<>VALUES(latitude)),'',region),
 adcode=IF((location_time IS NULL OR VALUES(location_time)>location_time) AND (longitude IS NULL OR longitude<>VALUES(longitude) OR latitude<>VALUES(latitude)),'',adcode),
 status=IF((location_time IS NULL OR VALUES(location_time)>location_time) AND (longitude IS NULL OR longitude<>VALUES(longitude) OR latitude<>VALUES(latitude)),'PENDING',status),
 attempts=IF((location_time IS NULL OR VALUES(location_time)>location_time) AND (longitude IS NULL OR longitude<>VALUES(longitude) OR latitude<>VALUES(latitude)),0,attempts),
 next_attempt_at=IF((location_time IS NULL OR VALUES(location_time)>location_time) AND (longitude IS NULL OR longitude<>VALUES(longitude) OR latitude<>VALUES(latitude)),NULL,next_attempt_at),
 last_error=IF((location_time IS NULL OR VALUES(location_time)>location_time) AND (longitude IS NULL OR longitude<>VALUES(longitude) OR latitude<>VALUES(latitude)),'',last_error),
 resolved_at=IF((location_time IS NULL OR VALUES(location_time)>location_time) AND (longitude IS NULL OR longitude<>VALUES(longitude) OR latitude<>VALUES(latitude)),NULL,resolved_at),
 longitude=IF(location_time IS NULL OR VALUES(location_time)>location_time,VALUES(longitude),longitude),
 latitude=IF(location_time IS NULL OR VALUES(location_time)>location_time,VALUES(latitude),latitude),
 source_protocol=IF(location_time IS NULL OR VALUES(location_time)>location_time,VALUES(source_protocol),source_protocol),
 location_time=IF(location_time IS NULL OR VALUES(location_time)>location_time,VALUES(location_time),location_time)`

func (w *Worker) SavePoint(ctx context.Context, p Point) error {
	if !validPoint(p) {
		return fmt.Errorf("invalid historical location")
	}
	_, e := w.DB.ExecContext(ctx, savePointSQL, p.VIN, p.Time.In(Shanghai).Format("2006-01-02"), math.Round(p.Longitude*1e6)/1e6, math.Round(p.Latitude*1e6)/1e6, p.Time.In(Shanghai), p.Protocol)
	return e
}

type pending struct {
	VIN, Date           string
	Longitude, Latitude float64
}

func (w *Worker) ResolvePending(ctx context.Context, from, to string, limit int) (int, int, error) {
	if limit <= 0 || limit > 5000 {
		limit = 500
	}
	rows, e := w.DB.QueryContext(ctx, `SELECT vin,DATE_FORMAT(stat_date,'%Y-%m-%d'),longitude,latitude FROM vehicle_daily_geography WHERE stat_date BETWEEN ? AND ? AND status IN ('PENDING','ERROR') AND (next_attempt_at IS NULL OR next_attempt_at<=NOW(3)) ORDER BY stat_date DESC,vin LIMIT ?`, from, to, limit)
	if e != nil {
		return 0, 0, e
	}
	items := []pending{}
	for rows.Next() {
		var p pending
		if e = rows.Scan(&p.VIN, &p.Date, &p.Longitude, &p.Latitude); e != nil {
			rows.Close()
			return 0, 0, e
		}
		items = append(items, p)
	}
	e = rows.Err()
	rows.Close()
	if e != nil {
		return 0, 0, e
	}
	jobs := make(chan pending, len(items))
	for _, p := range items {
		jobs <- p
	}
	close(jobs)
	count := w.Concurrency
	if count < 1 {
		count = 1
	}
	if count > 8 {
		count = 8
	}
	var mu sync.Mutex
	var wg sync.WaitGroup
	ok, failed := 0, 0
	var first error
	for i := 0; i < count; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for p := range jobs {
				if ctx.Err() != nil {
					return
				}
				e := w.resolveOne(ctx, p)
				mu.Lock()
				if e == nil {
					ok++
				} else {
					failed++
					if first == nil {
						first = e
					}
				}
				mu.Unlock()
			}
		}()
	}
	wg.Wait()
	if ctx.Err() != nil {
		return ok, failed, ctx.Err()
	}
	return ok, failed, first
}
func (w *Worker) resolveOne(ctx context.Context, p pending) error {
	var a Address
	e := w.DB.QueryRowContext(ctx, `SELECT province,city,region,adcode FROM vehicle_geography_cache WHERE longitude=? AND latitude=? AND resolved_at>=DATE_SUB(NOW(),INTERVAL 180 DAY)`, p.Longitude, p.Latitude).Scan(&a.Province, &a.City, &a.Region, &a.Adcode)
	if e != nil && e != sql.ErrNoRows {
		return e
	}
	if e == sql.ErrNoRows {
		// At most three external calls in flight by default; quota/network failures
		// are persisted and retried with backoff, never exported as a guessed city.
		w.requestMu.Lock()
		delay := time.Until(w.nextRequest)
		if delay > 0 {
			timer := time.NewTimer(delay)
			select {
			case <-ctx.Done():
				timer.Stop()
				w.requestMu.Unlock()
				return ctx.Err()
			case <-timer.C:
			}
		}
		interval := w.RequestInterval
		if interval <= 0 {
			interval = 20 * time.Millisecond
		}
		w.nextRequest = time.Now().Add(interval)
		w.requestMu.Unlock()
		rctx, cancel := context.WithTimeout(ctx, 12*time.Second)
		a, e = w.Resolver.Resolve(rctx, p.Longitude, p.Latitude)
		cancel()
		if e != nil {
			_, saveErr := w.DB.ExecContext(ctx, `UPDATE vehicle_daily_geography SET status='ERROR',attempts=attempts+1,last_error=?,next_attempt_at=DATE_ADD(NOW(3),INTERVAL LEAST(3600,60*POW(2,LEAST(attempts,6))) SECOND) WHERE vin=? AND stat_date=? AND longitude=? AND latitude=?`, e.Error(), p.VIN, p.Date, p.Longitude, p.Latitude)
			if saveErr != nil {
				return saveErr
			}
			return e
		}
		_, e = w.DB.ExecContext(ctx, `INSERT INTO vehicle_geography_cache(longitude,latitude,province,city,region,adcode) VALUES(?,?,?,?,?,?) ON DUPLICATE KEY UPDATE province=VALUES(province),city=VALUES(city),region=VALUES(region),adcode=VALUES(adcode),resolved_at=NOW(3)`, p.Longitude, p.Latitude, a.Province, a.City, a.Region, a.Adcode)
		if e != nil {
			return e
		}
	}
	_, e = w.DB.ExecContext(ctx, `UPDATE vehicle_daily_geography SET province=?,city=?,region=?,adcode=?,status='RESOLVED',last_error='',attempts=0,next_attempt_at=NULL,resolved_at=NOW(3) WHERE vin=? AND stat_date=? AND longitude=? AND latitude=?`, a.Province, a.City, a.Region, a.Adcode, p.VIN, p.Date, p.Longitude, p.Latitude)
	return e
}

// An advisory lock is tied to this dedicated connection and released even on
// cancellation/crash, preventing multiple API replicas from running the loop.
func (w *Worker) WithLock(ctx context.Context, name string, fn func() error) error {
	conn, e := w.DB.Conn(ctx)
	if e != nil {
		return e
	}
	defer conn.Close()
	var acquired int
	if e = conn.QueryRowContext(ctx, `SELECT GET_LOCK(?,0)`, name).Scan(&acquired); e != nil {
		return e
	}
	if acquired != 1 {
		return fmt.Errorf("daily geography worker already running")
	}
	defer conn.ExecContext(context.Background(), `SELECT RELEASE_LOCK(?)`, name)
	return fn()
}
func (w *Worker) Run(ctx context.Context) {
	ticker := time.NewTicker(time.Minute)
	defer ticker.Stop()
	round := 0
	for {
		e := w.WithLock(ctx, "vehicle_daily_geography_live", func() error {
			today := time.Now().In(Shanghai)
			date := today.Format("2006-01-02")
			rctx, c := context.WithTimeout(ctx, 50*time.Second)
			defer c()
			if _, e := w.RefreshDay(rctx, date); e != nil {
				return e
			}
			if round%60 == 0 {
				for d := 1; d <= 2; d++ {
					if _, e := w.RefreshDay(rctx, today.AddDate(0, 0, -d).Format("2006-01-02")); e != nil {
						return e
					}
				}
			}
			n, failed, e := w.ResolvePending(rctx, today.AddDate(0, 0, -2).Format("2006-01-02"), date, 500)
			log.Printf("daily_geography resolved=%d failed=%d", n, failed)
			if e != nil {
				return e
			}
			// Historical retry is independent of the live queue. The same lock used
			// by the CLI prevents duplicate external calls during a bulk backfill.
			_ = w.WithLock(rctx, "vehicle_daily_geography_backfill", func() error {
				n, failed, e := w.ResolvePending(rctx, "2000-01-01", today.AddDate(0, 0, -3).Format("2006-01-02"), 200)
				if n+failed > 0 {
					log.Printf("daily_geography_history_retry resolved=%d failed=%d error=%v", n, failed, e)
				}
				return e
			})
			return nil
		})
		if e != nil {
			log.Printf("daily_geography: %v", e)
		}
		round++
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}