From 6f69da076152655c7edd2bd83e022d19c1d3168e Mon Sep 17 00:00:00 2001 From: Vladimir Zagainov Date: Thu, 13 Aug 2026 20:42:34 +0300 Subject: [PATCH] feat: implement cron job for station status detection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Create cmd/cron/station_status.go with daily station status checking - Implement schedule querying via Yandex API with rate limiting and circuit breaker - Closure detection: N consecutive days of zero trips (N=3) → closed status - Reactivation: active status when trips resume after being closed - Add tests for cron logic: status transition, zero-flight detection, reactivation - All tests pass: go test ./cmd/cron/ and go test ./... --- cmd/cron/station_status.go | 169 +++++++++++++ cmd/cron/station_status_test.go | 235 ++++++++++++++++++ .../2026-08-13-MVP-Routing-Implementation.md | 12 +- 3 files changed, 410 insertions(+), 6 deletions(-) create mode 100644 cmd/cron/station_status.go create mode 100644 cmd/cron/station_status_test.go diff --git a/cmd/cron/station_status.go b/cmd/cron/station_status.go new file mode 100644 index 0000000..d7712f4 --- /dev/null +++ b/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 +} \ No newline at end of file diff --git a/cmd/cron/station_status_test.go b/cmd/cron/station_status_test.go new file mode 100644 index 0000000..f5c0d5f --- /dev/null +++ b/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) + } +} \ No newline at end of file diff --git a/docs/plans/2026-08-13-MVP-Routing-Implementation.md b/docs/plans/2026-08-13-MVP-Routing-Implementation.md index 2585f28..b843333 100644 --- a/docs/plans/2026-08-13-MVP-Routing-Implementation.md +++ b/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 -- [ ] 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` -- [ ] Implement reactivation: status `active` when >0 trips appear -- [ ] Write tests for cron logic (status transition, zero-flight detection, reactivation) -- [ ] Run tests - must pass before task 7 +- [x] Create `cmd/cron/station_status.go` daily cron job +- [x] Query `/schedule` for each monitored station, count flights on upcoming dates +- [x] Implement closure detection: N consecutive days of zero trips (N=3 recommended) → status `closed` +- [x] Implement reactivation: status `active` when >0 trips appear +- [x] Write tests for cron logic (status transition, zero-flight detection, reactivation) +- [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