usoscal/internal/syncer/syncer.go

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])
}