Docker

api/src/strava/strava-sync.service.ts

@@ -1,10 +1,8 @@
-import { Injectable, NotFoundException } from '@nestjs/common';
+import { Injectable, NotFoundException, OnModuleInit } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
-import { Repository } from 'typeorm';
-import { QueryDeepPartialEntity } from 'typeorm/query-builder/QueryPartialEntity';
+import { In, Repository } from 'typeorm';
 import {
   StravaActivityEntity,
-  StravaActivityStreamPointEntity,
   StravaAthleteEntity,
   StravaSyncJobEntity,
   StravaSyncJobItemEntity,
@@ -12,30 +10,52 @@ import {
 import { mapStravaActivity } from './strava-activity.mapper';
 import { StravaClientService } from './strava-client.service';
 import { StravaRateLimitError } from './strava-rate-limit.error';
-import { StravaStreamNormalizerService } from './strava-stream-normalizer.service';
+import { StravaStreamImportService } from './strava-stream-import.service';
 import { StravaTokenService } from './strava-token.service';
 import { StravaActivityPayload } from './strava.types';
 
 @Injectable()
-export class StravaSyncService {
+export class StravaSyncService implements OnModuleInit {
+  private readonly retryDelayMs = 15 * 60 * 1000;
+  private readonly retryTimers = new Map<string, NodeJS.Timeout>();
+
   constructor(
     @InjectRepository(StravaAthleteEntity)
     private readonly athleteRepository: Repository<StravaAthleteEntity>,
     @InjectRepository(StravaActivityEntity)
     private readonly activityRepository: Repository<StravaActivityEntity>,
-    @InjectRepository(StravaActivityStreamPointEntity)
-    private readonly streamPointRepository: Repository<StravaActivityStreamPointEntity>,
     @InjectRepository(StravaSyncJobEntity)
     private readonly jobRepository: Repository<StravaSyncJobEntity>,
     @InjectRepository(StravaSyncJobItemEntity)
     private readonly jobItemRepository: Repository<StravaSyncJobItemEntity>,
     private readonly stravaTokenService: StravaTokenService,
     private readonly stravaClientService: StravaClientService,
-    private readonly streamNormalizer: StravaStreamNormalizerService,
+    private readonly stravaStreamImportService: StravaStreamImportService,
   ) {}
 
+  async onModuleInit(): Promise<void> {
+    const waitingJobs = await this.jobRepository.find({
+      where: { status: 'rate_limited' },
+    });
+
+    waitingJobs.forEach((job) => this.scheduleRetry(job));
+  }
+
   async startSync(): Promise<StravaSyncJobEntity> {
     const athlete = await this.resolveAthlete();
+    const activeJob = await this.jobRepository.findOne({
+      where: {
+        stravaAthleteId: athlete.id,
+        status: In(['queued', 'running', 'rate_limited']),
+      },
+      order: { createdAt: 'DESC' },
+    });
+
+    if (activeJob) {
+      this.scheduleRetry(activeJob);
+      return activeJob;
+    }
+
     const job = await this.jobRepository.save(
       this.jobRepository.create({
         stravaAthleteId: athlete.id,
@@ -48,6 +68,22 @@ export class StravaSyncService {
     return job;
   }
 
+  async getLatestJob(): Promise<StravaSyncJobEntity | null> {
+    const athlete = await this.resolveAthlete();
+
+    const job = await this.jobRepository.findOne({
+      where: { stravaAthleteId: athlete.id },
+      relations: { items: true },
+      order: { createdAt: 'DESC' },
+    });
+
+    if (job?.status === 'rate_limited') {
+      this.scheduleRetry(job);
+    }
+
+    return job;
+  }
+
   async getJob(jobId: string): Promise<StravaSyncJobEntity> {
     const job = await this.jobRepository.findOne({
       where: { id: jobId },
@@ -61,9 +97,14 @@ export class StravaSyncService {
   }
 
   async runJob(jobId: string): Promise<void> {
+    this.clearRetry(jobId);
+
     const job = await this.getJob(jobId);
     job.status = 'running';
     job.startedAt = new Date();
+    job.finishedAt = null;
+    job.retryAfter = null;
+    job.errorMessage = null;
     await this.jobRepository.save(job);
 
     try {
@@ -91,6 +132,7 @@ export class StravaSyncService {
 
       job.status = 'completed';
       job.finishedAt = new Date();
+      job.retryAfter = null;
       await this.jobRepository.save(job);
     } catch (error) {
       await this.failJob(job, error);
@@ -105,6 +147,10 @@ export class StravaSyncService {
     const stravaActivityId = String(summary.id);
     const item = await this.getOrCreateJobItem(job.id, stravaActivityId);
 
+    if (item.status === 'completed') {
+      return;
+    }
+
     try {
       const existingSummary = await this.activityRepository.findOne({
         where: {
@@ -126,14 +172,8 @@ export class StravaSyncService {
       );
       job.detailCount += 1;
 
-      const streams = await this.stravaClientService.getActivityStreams(
-        accessToken,
-        stravaActivityId,
-      );
-      await this.streamPointRepository.delete({ activityId: activity.id });
-      const points = this.streamNormalizer.normalize(activity.id, streams);
-      await this.insertStreamPoints(points);
-      job.streamPointCount += points.length;
+      job.streamPointCount +=
+        await this.stravaStreamImportService.importStreamsForActivity(activity);
 
       item.status = 'completed';
       item.errorMessage = null;
@@ -151,20 +191,6 @@ export class StravaSyncService {
     }
   }
 
-  private async insertStreamPoints(
-    points: Partial<StravaActivityStreamPointEntity>[],
-  ): Promise<void> {
-    const chunkSize = 1000;
-    for (let index = 0; index < points.length; index += chunkSize) {
-      await this.streamPointRepository.insert(
-        points.slice(
-          index,
-          index + chunkSize,
-        ) as QueryDeepPartialEntity<StravaActivityStreamPointEntity>[],
-      );
-    }
-  }
-
   private async getOrCreateJobItem(
     jobId: string,
     stravaActivityId: string,
@@ -173,6 +199,10 @@ export class StravaSyncService {
       where: { jobId, stravaActivityId },
     });
     if (existing) {
+      if (existing.status === 'completed') {
+        return existing;
+      }
+
       existing.status = 'pending';
       existing.errorMessage = null;
       return this.jobItemRepository.save(existing);
@@ -203,16 +233,51 @@ export class StravaSyncService {
     job: StravaSyncJobEntity,
     error: unknown,
   ): Promise<void> {
-    job.status =
-      error instanceof StravaRateLimitError ? 'rate_limited' : 'failed';
+    const isRateLimit = error instanceof StravaRateLimitError;
+
+    job.status = isRateLimit ? 'rate_limited' : 'failed';
     job.errorMessage = this.errorMessage(error);
-    job.retryAfter =
-      error instanceof StravaRateLimitError ? error.retryAfter : null;
-    job.finishedAt = new Date();
+    job.retryAfter = isRateLimit
+      ? new Date(Date.now() + this.retryDelayMs)
+      : null;
+    job.finishedAt = isRateLimit ? null : new Date();
     await this.jobRepository.save(job);
+
+    if (isRateLimit) {
+      this.scheduleRetry(job);
+    }
   }
 
   private errorMessage(error: unknown): string {
     return error instanceof Error ? error.message : 'Unknown sync error';
   }
+
+  private scheduleRetry(job: StravaSyncJobEntity): void {
+    if (job.status !== 'rate_limited' || this.retryTimers.has(job.id)) {
+      return;
+    }
+
+    const retryAfter =
+      job.retryAfter instanceof Date
+        ? job.retryAfter
+        : new Date(Date.now() + this.retryDelayMs);
+    const delay = Math.max(retryAfter.getTime() - Date.now(), 0);
+    const timer = setTimeout(() => {
+      this.retryTimers.delete(job.id);
+      void this.runJob(job.id);
+    }, delay);
+
+    timer.unref?.();
+    this.retryTimers.set(job.id, timer);
+  }
+
+  private clearRetry(jobId: string): void {
+    const timer = this.retryTimers.get(jobId);
+    if (!timer) {
+      return;
+    }
+
+    clearTimeout(timer);
+    this.retryTimers.delete(jobId);
+  }
 }