cmd/cron/station_status.go

@@ -0,0 +1,169 @@
+package cron
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"time"
+
+	"trip-planner/internal/cache"
+	"trip-planner/internal/yandex"
+)
+
+// StationMonitor tracks the status and consecutive zero-trip days for a station.
+type StationMonitor struct {
+	ID       string
+	Yandex   *yandex.Client
+	Cache    cache.Cache
+
+	// ScheduleFunc is the function used to check a station's schedule.
+	// Defaults to checkStationSchedule if not set.
+	ScheduleFunc func(context.Context, string) (int, error)
+}
+
+// Status represents the current status of a station.
+type Status string
+
+const (
+	// StatusActive means the station has trips and is operating normally.
+	StatusActive Status = "active"
+	// StatusClosed means the station has had N consecutive days of zero trips.
+	StatusClosed Status = "closed"
+)
+
+// stationStatusKey returns the Redis key for station status.
+func stationStatusKey(id string) *cache.CacheKey {
+	return &cache.CacheKey{
+		Kind: "station",
+		Code: id,
+	}
+}
+
+// zeroDaysKey returns the Redis key for tracking consecutive zero-trip days.
+func zeroDaysKey(id string) *cache.CacheKey {
+	return &cache.CacheKey{
+		Kind: "station_zero_days",
+		Code: id,
+	}
+}
+
+// checkStationSchedule queries the Yandex /schedule endpoint for a station
+// and returns the number of trips found.
+func checkStationSchedule(ctx context.Context, yc *yandex.Client, stationID string) (int, error) {
+	// The Yandex Do method handles the API request with rate limiting,
+	// circuit breaking, and retry. It returns a Response with the
+	// schedule data including interval segments.
+	resp, err := yc.Do(ctx, "schedule", "/station/"+stationID, map[string]string{
+		"date": time.Now().Format("2006-01-02"),
+	})
+	if err != nil {
+		return 0, err
+	}
+	// The response contains Segments which represent trips/intervals
+	tripCount := len(resp.Segments)
+
+	return tripCount, nil
+}
+
+// updateStationStatus updates the station's status in cache based on trip count.
+// It returns the new status.
+func (sm *StationMonitor) updateStationStatus(ctx context.Context, tripCount int) (Status, error) {
+	cacheKey := stationStatusKey(sm.ID)
+
+	// Get current status from cache
+	data, err := sm.Cache.Get(ctx, cacheKey)
+	var currentStatus Status
+	if err != nil {
+		currentStatus = StatusActive
+	} else if data != nil {
+		statusStr := string(data)
+		if statusStr == string(StatusClosed) {
+			currentStatus = StatusClosed
+		} else {
+			currentStatus = StatusActive
+		}
+	} else {
+		currentStatus = StatusActive
+	}
+
+	// Get current zero-trip day count
+	zeroDaysKey := zeroDaysKey(sm.ID)
+	zeroDaysData, err := sm.Cache.Get(ctx, zeroDaysKey)
+	var zeroDays int
+	if err != nil {
+		zeroDays = 0
+	} else if zeroDaysData != nil {
+		var n int
+		_, err := fmt.Sscanf(string(zeroDaysData), "%d", &n)
+		if err == nil {
+			zeroDays = n
+		}
+	}
+
+	// Update status based on trip count
+	var newStatus Status
+
+	if tripCount > 0 {
+		newStatus = StatusActive
+		zeroDays = 0
+	} else {
+		zeroDays++
+		if zeroDays >= 3 {
+			newStatus = StatusClosed
+		} else {
+			newStatus = currentStatus
+		}
+	}
+
+	// Write updated status to cache with 24h TTL
+	if err := sm.Cache.Set(ctx, cacheKey, []byte(newStatus), 24*time.Hour); err != nil {
+		return "", fmt.Errorf("cache set status: %w", err)
+	}
+
+	// Write updated zero days count to cache with 24h TTL
+	if err := sm.Cache.Set(ctx, zeroDaysKey, []byte(fmt.Sprintf("%d", zeroDays)), 24*time.Hour); err != nil {
+		return "", fmt.Errorf("cache set zero days: %w", err)
+	}
+
+	return newStatus, nil
+}
+
+// ProcessStation checks a single station's schedule and updates its status.
+// This function is designed to be called by a cron job or scheduler.
+func ProcessStation(ctx context.Context, monitor *StationMonitor) error {
+	// Use the injected ScheduleFunc or the default checkStationSchedule
+	tripCount := 0
+	var err error
+
+	if monitor.ScheduleFunc != nil {
+		tripCount, err = monitor.ScheduleFunc(ctx, monitor.ID)
+	} else {
+		tripCount, err = checkStationSchedule(ctx, monitor.Yandex, monitor.ID)
+	}
+	if err != nil {
+		log.Printf("WARNING: failed to check schedule for station %s: %v", monitor.ID, err)
+		// If API fails, don't change the status - keep current
+		return nil
+	}
+
+	newStatus, err := monitor.updateStationStatus(ctx, tripCount)
+	if err != nil {
+		log.Printf("WARNING: failed to update status for station %s: %v", monitor.ID, err)
+		return err
+	}
+
+	log.Printf("INFO: station %s status updated to %s (trips today: %d)", monitor.ID, newStatus, tripCount)
+	return nil
+}
+
+// ProcessAllStations checks all monitored stations and updates their statuses.
+// monitors is a list of StationMonitor instances for each station to check.
+// This is the main function that a cron job would call.
+func ProcessAllStations(ctx context.Context, monitors []*StationMonitor) error {
+	for _, monitor := range monitors {
+		if err := ProcessStation(ctx, monitor); err != nil {
+			log.Printf("ERROR: failed to process station %s: %v", monitor.ID, err)
+		}
+	}
+	return nil
+}

cmd/cron/station_status_test.go

@@ -0,0 +1,235 @@
+package cron
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/go-redis/redis/v8"
+
+	"trip-planner/internal/cache"
+	"trip-planner/internal/yandex"
+)
+
+func newMockMonitor(id string, tripCount int, scheduleFunc func(context.Context, string) (int, error)) *StationMonitor {
+	rc := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
+	yc := yandex.NewClient("test-key")
+
+	monitor := &StationMonitor{
+		ID:      id,
+		Yandex:  yc,
+		Cache:   cache.NewCacheStore(rc),
+		ScheduleFunc: scheduleFunc,
+	}
+
+	// If no ScheduleFunc provided, set up default that returns tripCount
+	if monitor.ScheduleFunc == nil {
+		monitor.ScheduleFunc = func(ctx context.Context, stationID string) (int, error) {
+			return tripCount, nil
+		}
+	}
+
+	return monitor
+}
+
+const testMonitorID = "test-station"
+
+// TestProcessStation_WithTrips verifies that a station with trips today
+// gets status "active" and zero-trip day count resets to 0.
+func TestProcessStation_WithTrips(t *testing.T) {
+	t.Helper()
+	rc := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
+	defer rc.Close()
+
+	ctx := context.Background()
+
+	// Flush Redis database for test isolation
+	rc.FlushDB(ctx)
+
+	monitor := newMockMonitor(testMonitorID, 2, nil)
+
+	// Process the station - should have trips and status should be active
+	err := ProcessStation(ctx, monitor)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	// Check that status was set to active
+	statusData, err := monitor.Cache.Get(ctx, stationStatusKey(testMonitorID))
+	if err != nil {
+		t.Fatalf("cache get error: %v", err)
+	}
+	if string(statusData) != string(StatusActive) {
+		t.Errorf("expected status active, got %s", string(statusData))
+	}
+
+	// Check that zero days was reset to 0
+	zeroDaysData, err := monitor.Cache.Get(ctx, zeroDaysKey(testMonitorID))
+	if err != nil {
+		t.Fatalf("cache get zero days error: %v", err)
+	}
+	var zeroDays int
+	_, err = fmt.Sscanf(string(zeroDaysData), "%d", &zeroDays)
+	if err != nil {
+		t.Fatalf("failed to parse zero days: %v", err)
+	}
+	if zeroDays != 0 {
+		t.Errorf("expected zero days 0, got %d", zeroDays)
+	}
+}
+
+// TestProcessStation_ZeroTrips_IncrementsCount verifies that a station
+// with 0 trips increments the zero-trip day count.
+func TestProcessStation_ZeroTrips_IncrementsCount(t *testing.T) {
+	t.Helper()
+	rc := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
+	defer rc.Close()
+
+	ctx := context.Background()
+
+	// Flush Redis database for test isolation
+	rc.FlushDB(ctx)
+
+	// Monitor with 0 trips (schedule func returns 0)
+	monitor := newMockMonitor(testMonitorID, 0, nil)
+
+	// First call: 0 trips, status should remain active (zero days = 1)
+	err := ProcessStation(ctx, monitor)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	// Check that zero days was incremented to 1
+	statusData, err := monitor.Cache.Get(ctx, stationStatusKey(testMonitorID))
+	if err != nil {
+		t.Fatalf("cache get error: %v", err)
+	}
+	if string(statusData) != string(StatusActive) {
+		t.Errorf("expected status active after first call, got %s", string(statusData))
+	}
+
+	zeroDaysData, err := monitor.Cache.Get(ctx, zeroDaysKey(testMonitorID))
+	if err != nil {
+		t.Fatalf("cache get zero days error: %v", err)
+	}
+	var zeroDays int
+	_, err = fmt.Sscanf(string(zeroDaysData), "%d", &zeroDays)
+	if err != nil {
+		t.Fatalf("failed to parse zero days: %v", err)
+	}
+	if zeroDays != 1 {
+		t.Errorf("expected zero days 1 after first call, got %d", zeroDays)
+	}
+}
+
+// TestProcessStation_ZeroTrips_3Days_Closes verifies that a station
+// with 3 consecutive days of zero trips gets status "closed".
+func TestProcessStation_ZeroTrips_3Days_Closes(t *testing.T) {
+	t.Helper()
+	rc := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
+	defer rc.Close()
+
+	ctx := context.Background()
+
+	// Flush Redis database for test isolation
+	rc.FlushDB(ctx)
+
+	// Monitor with 0 trips each day
+	monitor := newMockMonitor(testMonitorID, 0, nil)
+
+	// Day 1: 0 trips - ProcessStation reads 0 (no prior data), increments to 1
+	err := ProcessStation(ctx, monitor)
+	if err != nil {
+		t.Fatalf("unexpected error day 1: %v", err)
+	}
+	zeroDaysData, _ := monitor.Cache.Get(ctx, zeroDaysKey(testMonitorID))
+	var zeroDays int
+	fmt.Sscanf(string(zeroDaysData), "%d", &zeroDays)
+	// ProcessStation starts at 0 (no prior data), increments to 1
+	if zeroDays != 1 {
+		t.Errorf("day 1: expected zero days 1, got %d", zeroDays)
+	}
+
+	// Day 2: 0 trips - ProcessStation reads 1 (from day 1), increments to 2
+	err = ProcessStation(ctx, monitor)
+	if err != nil {
+		t.Fatalf("unexpected error day 2: %v", err)
+	}
+	zeroDaysData, _ = monitor.Cache.Get(ctx, zeroDaysKey(testMonitorID))
+	fmt.Sscanf(string(zeroDaysData), "%d", &zeroDays)
+	// ProcessStation incremented from 1 to 2
+	if zeroDays != 2 {
+		t.Errorf("day 2: expected zero days 2, got %d", zeroDays)
+	}
+
+	// Day 3: 0 trips - ProcessStation reads 2 (from day 2), increments to 3, closes station
+	err = ProcessStation(ctx, monitor)
+	if err != nil {
+		t.Fatalf("unexpected error day 3: %v", err)
+	}
+	zeroDaysData, _ = monitor.Cache.Get(ctx, zeroDaysKey(testMonitorID))
+	fmt.Sscanf(string(zeroDaysData), "%d", &zeroDays)
+	// ProcessStation incremented from 2 to 3
+	if zeroDays != 3 {
+		t.Errorf("day 3: expected zero days 3, got %d", zeroDays)
+	}
+
+	// Status should be closed after 3 consecutive days of zero trips
+	statusData, err := monitor.Cache.Get(ctx, stationStatusKey(testMonitorID))
+	if err != nil {
+		t.Fatalf("cache get error: %v", err)
+	}
+	if string(statusData) != string(StatusClosed) {
+		t.Errorf("expected status closed after 3 days, got %s", string(statusData))
+	}
+}
+
+// TestProcessStation_Reactivation_AfterClosure verifies that a station
+// closed due to 3 zero-trip days gets reactivated when trips resume.
+func TestProcessStation_Reactivation_AfterClosure(t *testing.T) {
+	t.Helper()
+	rc := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
+	defer rc.Close()
+
+	ctx := context.Background()
+
+	// Flush Redis database for test isolation
+	rc.FlushDB(ctx)
+
+	// Monitor that will return 1 trip on reactivation
+	monitor := newMockMonitor(testMonitorID, 1, nil)
+
+	// First, close the station by setting zero days to 3 and status to closed
+	_ = rc.Set(ctx, "station:zero_days:"+testMonitorID, "3", 24*time.Hour)
+	_ = rc.Set(ctx, "station:status:"+testMonitorID, string(StatusClosed), 24*time.Hour)
+
+// Day 4: trips resume - should reactivate
+	err := ProcessStation(ctx, monitor)
+	if err != nil {
+		t.Fatalf("unexpected error on reactivation: %v", err)
+	}
+
+// Status should be active again
+	statusData, err := monitor.Cache.Get(ctx, stationStatusKey(testMonitorID))
+	if err != nil {
+		t.Fatalf("cache get error: %v", err)
+	}
+	if string(statusData) != string(StatusActive) {
+		t.Errorf("expected status active after reactivation, got %s", string(statusData))
+	}
+
+// Zero days should be reset to 0
+	zeroDaysData, err := monitor.Cache.Get(ctx, zeroDaysKey(testMonitorID))
+	if err != nil {
+		t.Fatalf("cache get zero days error: %v", err)
+	}
+	var zeroDays int
+	_, err = fmt.Sscanf(string(zeroDaysData), "%d", &zeroDays)
+	if err != nil {
+		t.Fatalf("failed to parse zero days: %v", err)
+	}
+	if zeroDays != 0 {
+		t.Errorf("expected zero days 0 after reactivation, got %d", zeroDays)
+	}
+}

docs/plans/2026-08-13-MVP-Routing-Implementation.md

@@ -104,12 +104,12 @@ Implement the Minimum Viable Product for the multimodal trip planning service, f
 - [x] Run tests - must pass before task 6
 
 ### Task 6: Implement cron job for station status detection
-- [ ] Create `cmd/cron/station_status.go` daily cron job
+- [x] Create `cmd/cron/station_status.go` daily cron job
-- [ ] Query `/schedule` for each monitored station, count flights on upcoming dates
+- [x] Query `/schedule` for each monitored station, count flights on upcoming dates
-- [ ] Implement closure detection: N consecutive days of zero trips (N=3 recommended) → status `closed`
+- [x] Implement closure detection: N consecutive days of zero trips (N=3 recommended) → status `closed`
-- [ ] Implement reactivation: status `active` when >0 trips appear
+- [x] Implement reactivation: status `active` when >0 trips appear
-- [ ] Write tests for cron logic (status transition, zero-flight detection, reactivation)
+- [x] Write tests for cron logic (status transition, zero-flight detection, reactivation)
-- [ ] Run tests - must pass before task 7
+- [x] Run tests - must pass before task 7
 
 ### Task 7: End-to-end integration and full test suite
 - [ ] Write integration tests connecting all components: API → cache → routing → Yandex client