package syncer import ( "context" "crypto/sha256" "encoding/hex" "errors" "log" "slices" "strings" "time" "usoscal/internal/seal" "usoscal/internal/store" "usoscal/internal/usos" ) const ( week = 7 * 24 * time.Hour windowDays = 7 fallbackHorizon = 26 * week maxHorizon = 40 * week termLookahead = 60 * 24 * time.Hour longestSemester = 250 * 24 * time.Hour backfillLimit = 6 * 52 * week backfillQuietRun = 26 queueSize = 256 retryAfter = time.Hour TokenPurpose = "/usos_token" SecretPurpose = "/usos_secret" revokedMessage = "USOS no longer accepts this authorization. Sign in again to resume updates." failedMessage = "The last update from USOS failed. It will be retried." ) type Syncer struct { Store *store.Store USOS *usos.Client Box *seal.Box Interval time.Duration Pause time.Duration Now func() time.Time queue chan string } func New(st *store.Store, client *usos.Client, box *seal.Box, interval time.Duration) *Syncer { return &Syncer{ Store: st, USOS: client, Box: box, Interval: interval, Pause: 200 * time.Millisecond, Now: time.Now, queue: make(chan string, queueSize), } } func Today(now time.Time) time.Time { local := now.In(usos.Location) return time.Date(local.Year(), local.Month(), local.Day(), 0, 0, 0, 0, usos.Location) } func minTime(a, b time.Time) time.Time { if a.Before(b) { return a } return b } func (s *Syncer) Kick(userID string) { select { case s.queue <- userID: default: } } func (s *Syncer) Run(ctx context.Context) { ticker := time.NewTicker(15 * time.Minute) defer ticker.Stop() s.enqueueDue(ctx) for { select { case <-ctx.Done(): return case <-ticker.C: s.enqueueDue(ctx) case userID := <-s.queue: if err := s.Sync(ctx, userID); err != nil { log.Printf("sync of user %s failed: %v", userID, err) } } } } func (s *Syncer) enqueueDue(ctx context.Context) { ids, err := s.Store.DueUsers(ctx, s.Now().Add(-s.Interval)) if err != nil { log.Printf("listing users due for sync failed: %v", err) return } for _, id := range ids { s.Kick(id) } } func (s *Syncer) Sync(ctx context.Context, userID string) error { user, err := s.Store.User(ctx, userID) if errors.Is(err, store.ErrNotFound) || (err == nil && !user.Connected()) { return nil } if err != nil { return err } token, err := s.token(user) if err != nil { return s.Store.DropTokens(ctx, userID, revokedMessage) } err = s.pull(ctx, user, token) switch { case errors.Is(err, usos.ErrUnauthorized): return s.Store.DropTokens(ctx, userID, revokedMessage) case errors.Is(err, context.Canceled): return err case err != nil: retryAt := s.Now().Add(min(retryAfter, s.Interval) - s.Interval) return errors.Join(err, s.Store.SetSynced(ctx, userID, retryAt, failedMessage)) } return s.Store.SetSynced(ctx, userID, s.Now(), "") } func (s *Syncer) token(user store.User) (usos.Token, error) { key, err := s.Box.Open(user.USOSToken, user.ID+TokenPurpose) if err != nil { return usos.Token{}, err } secret, err := s.Box.Open(user.USOSSecret, user.ID+SecretPurpose) return usos.Token{Key: key, Secret: secret}, err } func (s *Syncer) pull(ctx context.Context, user store.User, token usos.Token) error { today := Today(s.Now()) horizon := s.horizon(ctx, today) for from := today; from.Before(horizon); from = from.AddDate(0, 0, windowDays) { if _, err := s.window(ctx, user.ID, token, from); err != nil { return err } } if !user.Archive { return s.Store.DeleteEventsBefore(ctx, user.ID, today) } if user.Backfilled { return nil } quiet := 0 for from := today.AddDate(0, 0, -windowDays); quiet < backfillQuietRun && today.Sub(from) <= backfillLimit; from = from.AddDate(0, 0, -windowDays) { found, err := s.window(ctx, user.ID, token, from) if err != nil { return err } if found == 0 { quiet++ } else { quiet = 0 } } return s.Store.SetBackfilled(ctx, user.ID) } func (s *Syncer) horizon(ctx context.Context, today time.Time) time.Time { terms, err := s.USOS.Terms(ctx, today.Add(termLookahead), today) if err != nil { log.Printf("term dates unavailable, using the default horizon: %v", err) } var finish time.Time for _, term := range terms { if term.Finish.Sub(term.Start) <= longestSemester && term.Finish.After(finish) { finish = term.Finish } } if finish.IsZero() { return today.Add(fallbackHorizon) } return minTime(finish.AddDate(0, 0, 1), today.Add(maxHorizon)) } func (s *Syncer) window(ctx context.Context, userID string, token usos.Token, from time.Time) (int, error) { select { case <-ctx.Done(): return 0, ctx.Err() case <-time.After(s.Pause): } activities, err := s.USOS.Timetable(ctx, token, from, windowDays) if err != nil { return 0, err } to := from.AddDate(0, 0, windowDays) events := make([]store.Event, 0, len(activities)) seen := map[string]bool{} lecturers := s.lecturers(ctx, activities) for _, activity := range activities { if activity.Start.Before(from) || !activity.Start.Before(to) { continue } event := s.event(ctx, activity, lecturers) if seen[event.UID] { continue } seen[event.UID] = true events = append(events, event) } return len(events), s.Store.ReplaceWindow(ctx, userID, from, to, events) } func (s *Syncer) lecturers(ctx context.Context, activities []usos.Activity) map[string]string { var ids []string for _, activity := range activities { for _, id := range activity.LecturerIDs { if !slices.Contains(ids, id) { ids = append(ids, id) } } } names, err := s.Store.LecturerNames(ctx, ids) if err != nil { log.Printf("lecturer cache read failed: %v", err) names = map[string]string{} } missing := slices.DeleteFunc(slices.Clone(ids), func(id string) bool { _, known := names[id]; return known }) if len(missing) == 0 { return names } fetched, err := s.USOS.Lecturers(ctx, missing) if err != nil { log.Printf("lecturer names unavailable: %v", err) } if err := s.Store.SaveLecturers(ctx, fetched); err != nil { log.Printf("lecturer cache write failed: %v", err) } for id, name := range fetched { names[id] = name } return names } func (s *Syncer) building(ctx context.Context, id, fallbackName string) store.Building { if id == "" { return store.Building{Name: fallbackName} } if cached, ok := s.Store.Building(ctx, id); ok { return cached } fetched, err := s.USOS.Building(ctx, id) if err != nil { log.Printf("building %s unavailable: %v", id, err) return store.Building{Name: fallbackName} } building := store.Building{Name: fetched.Name.String(), Address: fetched.Address} if building.Name == "" { building.Name = fallbackName } if err := s.Store.SaveBuilding(ctx, id, building); err != nil { log.Printf("building cache write failed: %v", err) } return building } func (s *Syncer) event(ctx context.Context, activity usos.Activity, lecturers map[string]string) store.Event { title, kind := activity.CourseName, activity.ClasstypeName if title == "" { title = activity.Name } if kind == "" && activity.Name != title { kind = activity.Name } names := make([]string, 0, len(activity.LecturerIDs)) for _, id := range activity.LecturerIDs { if name := lecturers[id]; name != "" { names = append(names, name) } } building := s.building(ctx, activity.BuildingID, activity.BuildingName) return store.Event{ UID: uid(activity), Start: activity.Start, End: activity.End, Title: title, Kind: kind, Group: activity.GroupNumber, Room: activity.RoomNumber, Building: building.Name, Address: building.Address, Lecturers: strings.Join(names, ", "), URL: activity.URL, } } func uid(activity usos.Activity) string { if activity.MeetingID != "" { return "sm-" + activity.MeetingID } sum := sha256.Sum256([]byte(strings.Join([]string{ activity.Type, activity.CourseID, activity.UnitID, activity.GroupNumber, activity.Name, activity.Start.UTC().Format(time.RFC3339), }, "\x00"))) return "a-" + hex.EncodeToString(sum[:16]) }