git init
This commit is contained in:
Bastian Wagner
218
api/src/strava/strava-sync.service.ts
Normal file
218
api/src/strava/strava-sync.service.ts
Normal file
@@ -0,0 +1,218 @@
|
||||
import { Injectable, NotFoundException } from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { Repository } from 'typeorm';
|
||||
import { QueryDeepPartialEntity } from 'typeorm/query-builder/QueryPartialEntity';
|
||||
import {
|
||||
StravaActivityEntity,
|
||||
StravaActivityStreamPointEntity,
|
||||
StravaAthleteEntity,
|
||||
StravaSyncJobEntity,
|
||||
StravaSyncJobItemEntity,
|
||||
} from '../database/entities';
|
||||
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 { StravaTokenService } from './strava-token.service';
|
||||
import { StravaActivityPayload } from './strava.types';
|
||||
|
||||
@Injectable()
|
||||
export class StravaSyncService {
|
||||
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,
|
||||
) {}
|
||||
|
||||
async startSync(): Promise<StravaSyncJobEntity> {
|
||||
const athlete = await this.resolveAthlete();
|
||||
const job = await this.jobRepository.save(
|
||||
this.jobRepository.create({
|
||||
stravaAthleteId: athlete.id,
|
||||
status: 'queued',
|
||||
}),
|
||||
);
|
||||
|
||||
void this.runJob(job.id);
|
||||
|
||||
return job;
|
||||
}
|
||||
|
||||
async getJob(jobId: string): Promise<StravaSyncJobEntity> {
|
||||
const job = await this.jobRepository.findOne({
|
||||
where: { id: jobId },
|
||||
relations: { items: true },
|
||||
});
|
||||
if (!job) {
|
||||
throw new NotFoundException('Sync job not found');
|
||||
}
|
||||
|
||||
return job;
|
||||
}
|
||||
|
||||
async runJob(jobId: string): Promise<void> {
|
||||
const job = await this.getJob(jobId);
|
||||
job.status = 'running';
|
||||
job.startedAt = new Date();
|
||||
await this.jobRepository.save(job);
|
||||
|
||||
try {
|
||||
const accessToken = await this.stravaTokenService.getValidAccessToken(
|
||||
job.stravaAthleteId,
|
||||
);
|
||||
|
||||
let page = 1;
|
||||
const perPage = 100;
|
||||
let summaries: StravaActivityPayload[] = [];
|
||||
|
||||
do {
|
||||
summaries = await this.stravaClientService.listActivities(
|
||||
accessToken,
|
||||
page,
|
||||
perPage,
|
||||
);
|
||||
|
||||
for (const summary of summaries) {
|
||||
await this.importActivity(job, accessToken, summary);
|
||||
}
|
||||
|
||||
page += 1;
|
||||
} while (summaries.length === perPage);
|
||||
|
||||
job.status = 'completed';
|
||||
job.finishedAt = new Date();
|
||||
await this.jobRepository.save(job);
|
||||
} catch (error) {
|
||||
await this.failJob(job, error);
|
||||
}
|
||||
}
|
||||
|
||||
private async importActivity(
|
||||
job: StravaSyncJobEntity,
|
||||
accessToken: string,
|
||||
summary: StravaActivityPayload,
|
||||
): Promise<void> {
|
||||
const stravaActivityId = String(summary.id);
|
||||
const item = await this.getOrCreateJobItem(job.id, stravaActivityId);
|
||||
|
||||
try {
|
||||
const existingSummary = await this.activityRepository.findOne({
|
||||
where: {
|
||||
stravaAthleteId: job.stravaAthleteId,
|
||||
stravaActivityId,
|
||||
},
|
||||
});
|
||||
let activity = await this.activityRepository.save(
|
||||
mapStravaActivity(job.stravaAthleteId, summary, existingSummary),
|
||||
);
|
||||
job.activityCount += 1;
|
||||
|
||||
const detail = await this.stravaClientService.getActivity(
|
||||
accessToken,
|
||||
stravaActivityId,
|
||||
);
|
||||
activity = await this.activityRepository.save(
|
||||
mapStravaActivity(job.stravaAthleteId, detail, activity),
|
||||
);
|
||||
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;
|
||||
|
||||
item.status = 'completed';
|
||||
item.errorMessage = null;
|
||||
await this.jobItemRepository.save(item);
|
||||
await this.jobRepository.save(job);
|
||||
} catch (error) {
|
||||
if (error instanceof StravaRateLimitError) {
|
||||
throw error;
|
||||
}
|
||||
|
||||
item.status = 'failed';
|
||||
item.errorMessage = this.errorMessage(error);
|
||||
await this.jobItemRepository.save(item);
|
||||
await this.jobRepository.save(job);
|
||||
}
|
||||
}
|
||||
|
||||
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,
|
||||
): Promise<StravaSyncJobItemEntity> {
|
||||
const existing = await this.jobItemRepository.findOne({
|
||||
where: { jobId, stravaActivityId },
|
||||
});
|
||||
if (existing) {
|
||||
existing.status = 'pending';
|
||||
existing.errorMessage = null;
|
||||
return this.jobItemRepository.save(existing);
|
||||
}
|
||||
|
||||
return this.jobItemRepository.save(
|
||||
this.jobItemRepository.create({
|
||||
jobId,
|
||||
stravaActivityId,
|
||||
status: 'pending',
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
private async resolveAthlete(): Promise<StravaAthleteEntity> {
|
||||
const athlete = await this.athleteRepository.findOne({
|
||||
where: { accountKey: 'primary' },
|
||||
});
|
||||
|
||||
if (!athlete) {
|
||||
throw new NotFoundException('No Strava athlete connected');
|
||||
}
|
||||
|
||||
return athlete;
|
||||
}
|
||||
|
||||
private async failJob(
|
||||
job: StravaSyncJobEntity,
|
||||
error: unknown,
|
||||
): Promise<void> {
|
||||
job.status =
|
||||
error instanceof StravaRateLimitError ? 'rate_limited' : 'failed';
|
||||
job.errorMessage = this.errorMessage(error);
|
||||
job.retryAfter =
|
||||
error instanceof StravaRateLimitError ? error.retryAfter : null;
|
||||
job.finishedAt = new Date();
|
||||
await this.jobRepository.save(job);
|
||||
}
|
||||
|
||||
private errorMessage(error: unknown): string {
|
||||
return error instanceof Error ? error.message : 'Unknown sync error';
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user