run.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"net/url"
	"time"

	crdbpgx "github.com/cockroachdb/cockroach-go/v2/crdb/crdbpgxv5"
	"github.com/gofrs/uuid/v5"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
	"golang.org/x/sync/errgroup"
	"golang.org/x/time/rate"
)

const (
	validCheckTimeout  = time.Hour * 24 * 7
	validCheckInterval = time.Hour * 24
	nameCheckInterval  = validCheckInterval
)

type token struct {
	AccountId      uuid.UUID
	Gw2AccountId   uuid.UUID
	Gw2AccountName string
	Gw2ApiToken    string
}

type apiRequestError struct {
	statusCode int
	error
}

func (err apiRequestError) penalize() bool {
	return err.statusCode >= 400 && err.statusCode < 500
}

type TokenChecker struct {
	pool       *pgxpool.Pool
	httpClient *http.Client
	tracer     trace.Tracer
	limiter    *rate.Limiter
}

func NewTokenChecker(pool *pgxpool.Pool, httpClient *http.Client) *TokenChecker {
	tracer := otel.Tracer("github.com/gw2auth/gw2auth.com-background-jobs::TokenChecker", trace.WithInstrumentationVersion("v0.0.1"))
	limiter := rate.NewLimiter(rate.Every(time.Minute)*250, 300)

	return &TokenChecker{
		pool:       pool,
		httpClient: httpClient,
		tracer:     tracer,
		limiter:    limiter,
	}
}

func (tc *TokenChecker) Run(ctx context.Context) error {
	for {
		tokensToCheck, err := tc.loadBatch(ctx, 20)
		if err != nil {
			return fmt.Errorf("failed to load token batch: %w", err)
		}

		if len(tokensToCheck) < 1 {
			break
		}

		if err := tc.checkBatch(ctx, tokensToCheck); err != nil {
			return fmt.Errorf("failed to check token batch: %w", err)
		}
	}

	return nil
}

func (tc *TokenChecker) loadBatch(ctx context.Context, batchSize int) ([]token, error) {
	now := time.Now()
	minLastValidTime := now.Add(-validCheckTimeout)
	maxValidCheckTime := now.Add(-validCheckInterval)
	maxNameCheckTime := now.Add(-nameCheckInterval)

	tokensToCheck := make([]token, 0, batchSize)
	err := crdbpgx.ExecuteTx(ctx, tc.pool, pgx.TxOptions{AccessMode: pgx.ReadOnly}, func(tx pgx.Tx) error {
		rows, err := tx.Query(
			ctx,
			`
SELECT tk.account_id, tk.gw2_account_id, acc.gw2_account_name, tk.gw2_api_token
FROM gw2_account_api_tokens tk
INNER JOIN gw2_accounts acc
ON tk.account_id = acc.account_id AND tk.gw2_account_id = acc.gw2_account_id
WHERE tk.last_valid_time >= $1
AND (tk.last_valid_check_time <= $2 OR acc.last_name_check_time <= $3)
ORDER BY LEAST(tk.last_valid_check_time, acc.last_name_check_time) ASC
LIMIT $4
`,
			minLastValidTime,
			maxValidCheckTime,
			maxNameCheckTime,
			batchSize,
		)
		if err != nil {
			return err
		}
		defer rows.Close()

		for rows.Next() {
			var t token
			if err := rows.Scan(&t.AccountId, &t.Gw2AccountId, &t.Gw2AccountName, &t.Gw2ApiToken); err != nil {
				return err
			}

			tokensToCheck = append(tokensToCheck, t)
		}

		return rows.Err()
	})
	if err != nil {
		return nil, err
	}

	return tokensToCheck, nil
}

func (tc *TokenChecker) checkBatch(ctx context.Context, tokensToCheck []token) error {
	slog.InfoContext(ctx, "checking batch", slog.Int("batch.size", len(tokensToCheck)))

	g, ctx := errgroup.WithContext(ctx)
	for _, t := range tokensToCheck {
		g.Go(func() error {
			return tc.checkSingle(ctx, t)
		})
	}

	if err := g.Wait(); err != nil {
		return fmt.Errorf("token check failed: %w", err)
	}

	return nil
}

func (tc *TokenChecker) checkSingle(ctx context.Context, tk token) error {
	ctx, span := tc.tracer.Start(
		ctx,
		"CheckSingle",
		trace.WithSpanKind(trace.SpanKindServer),
		trace.WithAttributes(
			attribute.String("account.id", tk.AccountId.String()),
			attribute.String("gw2account.id", tk.Gw2AccountId.String()),
		),
	)
	defer span.End()

	if err := tc.checkSingleInternal(ctx, tk); err != nil {
		span.RecordError(err)
		return err
	}

	return nil
}

func (tc *TokenChecker) checkSingleInternal(ctx context.Context, tk token) error {
	var newGw2AccountName string
	var isValid bool
	{
		var err error
		newGw2AccountName, err = tc.getGw2AccountName(ctx, tk.Gw2ApiToken)
		if err != nil {
			slog.InfoContext(
				ctx,
				"could not get gw2 account name",
				slog.String("account.id", tk.AccountId.String()),
				slog.String("gw2account.id", tk.Gw2AccountId.String()),
				slog.String("gw2account.name.old", tk.Gw2AccountName),
				slog.String("error", err.Error()),
			)
		} else {
			slog.InfoContext(
				ctx,
				"token check succeeded",
				slog.String("account.id", tk.AccountId.String()),
				slog.String("gw2account.id", tk.Gw2AccountId.String()),
				slog.String("gw2account.name.new", newGw2AccountName),
				slog.String("gw2account.name.old", tk.Gw2AccountName),
				slog.Bool("gw2account.name.changed", newGw2AccountName != tk.Gw2AccountName),
			)
		}

		if err != nil {
			var rErr apiRequestError
			if !errors.As(err, &rErr) || rErr.statusCode == http.StatusTooManyRequests {
				return err
			}

			if rErr.penalize() {
				isValid = false
			} else {
				isValid = true
				newGw2AccountName = tk.Gw2AccountName
			}
		} else {
			isValid = true
		}
	}

	now := time.Now()
	return crdbpgx.ExecuteTx(ctx, tc.pool, pgx.TxOptions{}, func(tx pgx.Tx) error {
		var err error
		if isValid {
			_, err1 := tx.Exec(
				ctx,
				`
UPDATE gw2_account_api_tokens
SET last_valid_check_time = $1, last_valid_time = $1
WHERE account_id = $2
AND gw2_account_id = $3
`,
				now,
				tk.AccountId,
				tk.Gw2AccountId,
			)

			_, err2 := tx.Exec(
				ctx,
				`
UPDATE gw2_accounts
SET
	gw2_account_name = $1,
	last_name_check_time = $2,
	display_name = IF(display_name = gw2_account_name, $1, display_name)
WHERE account_id = $3
AND gw2_account_id = $4
`,
				newGw2AccountName,
				now,
				tk.AccountId,
				tk.Gw2AccountId,
			)

			err = errors.Join(err1, err2)
		} else {
			_, err1 := tx.Exec(
				ctx,
				`
UPDATE gw2_account_api_tokens
SET last_valid_check_time = $1
WHERE account_id = $2
AND gw2_account_id = $3
`,
				now,
				tk.AccountId,
				tk.Gw2AccountId,
			)

			_, err2 := tx.Exec(
				ctx,
				`
UPDATE gw2_accounts
SET last_name_check_time = $1
WHERE account_id = $2
AND gw2_account_id = $3
`,
				now,
				tk.AccountId,
				tk.Gw2AccountId,
			)

			err = errors.Join(err1, err2)
		}

		return err
	})
}

func (tc *TokenChecker) getGw2AccountName(ctx context.Context, gw2ApiToken string) (string, error) {
	q := make(url.Values)
	q.Set("v", "2025-08-29T00:00:00.000Z")
	q.Set("access_token", gw2ApiToken)

	resp, err := tc.getFromApi(ctx, "https://api.guildwars2.com/v2/account", q)
	if err != nil {
		return "", apiRequestError{statusCode: 0, error: err}
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		content, _ := io.ReadAll(resp.Body)
		err := fmt.Errorf("unexpected status code: %d (body=%q)", resp.StatusCode, string(content))

		return "", apiRequestError{statusCode: resp.StatusCode, error: err}
	}

	var body struct {
		Name string `json:"name"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
		return "", apiRequestError{statusCode: resp.StatusCode, error: err}
	}

	return body.Name, nil
}

func (tc *TokenChecker) getFromApi(ctx context.Context, url string, q url.Values) (*http.Response, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
	if err != nil {
		return nil, err
	}

	baseQuery := req.URL.Query()
	for k, v := range q {
		for _, vv := range v {
			baseQuery.Add(k, vv)
		}
	}

	req.URL.RawQuery = baseQuery.Encode()

	const maxAttempts = 5
	for i := 0; i < maxAttempts; i++ {
		if err := tc.limiter.Wait(ctx); err != nil {
			return nil, err
		}

		resp, err := tc.httpClient.Do(req)
		if err != nil {
			return nil, err
		}

		switch resp.StatusCode {
		case http.StatusOK:
			return resp, nil

		case http.StatusNotFound, http.StatusGatewayTimeout:
			// retry
			_ = resp.Body.Close()

		default:
			// abort
			return resp, nil
		}
	}

	return nil, errors.New("max attempts exceeded")
}