|
|
@@ -3,18 +3,55 @@ import { Injectable, Logger } from '@nestjs/common';
|
|
|
import { HttpService } from '@nestjs/axios';
|
|
|
import { MongoPrismaService } from '@box/db/prisma/mongo-prisma.service';
|
|
|
import { firstValueFrom } from 'rxjs';
|
|
|
+import { EntityType } from '@prisma/mongo/client';
|
|
|
|
|
|
export interface ProviderVideoSyncOptions {
|
|
|
providerCode?: string;
|
|
|
- page?: number;
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Optional override. In normal usage, we resume from SyncState cursor:
|
|
|
+ * - fullSync: pageNum resumes
|
|
|
+ * - incremental: always pageNum=1
|
|
|
+ */
|
|
|
+ pageNum?: number;
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Default 200, hard-capped to 500.
|
|
|
+ */
|
|
|
pageSize?: number;
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Provider search param.
|
|
|
+ * - status: required in your business rule ("Completed")
|
|
|
+ * - updatedAt: ISO string filter "updated after"
|
|
|
+ */
|
|
|
+ param?: {
|
|
|
+ status?: string;
|
|
|
+ updatedAt?: string;
|
|
|
+ [k: string]: any;
|
|
|
+ };
|
|
|
+
|
|
|
+ /**
|
|
|
+ * fullSync:
|
|
|
+ * - true: no param.updatedAt; resume using stored pageNum
|
|
|
+ * - false: use param.updatedAt (cursor); pageNum forced to 1
|
|
|
+ */
|
|
|
+ fullSync?: boolean;
|
|
|
+
|
|
|
+ /**
|
|
|
+ * If true: ignore stored cursor and start fresh.
|
|
|
+ * - fullSync: pageNum from options.pageNum or 1
|
|
|
+ * - incremental: updatedAtCursor from options.param.updatedAt (if provided)
|
|
|
+ */
|
|
|
+ resetState?: boolean;
|
|
|
+
|
|
|
[key: string]: any;
|
|
|
}
|
|
|
|
|
|
export interface ProviderVideoSyncResult {
|
|
|
imported: number;
|
|
|
- created: number;
|
|
|
- updated: number;
|
|
|
+ created: number; // Option B: always 0
|
|
|
+ updated: number; // Option B: equals successful upserts
|
|
|
skipped: number;
|
|
|
errors?: Array<{ id?: string; error: string }>;
|
|
|
}
|
|
|
@@ -71,6 +108,23 @@ interface RawProviderVideo {
|
|
|
infoTsName?: string;
|
|
|
}
|
|
|
|
|
|
+type SyncCursor = {
|
|
|
+ pageNum: number;
|
|
|
+ pageSize: number;
|
|
|
+ updatedAtCursor?: string;
|
|
|
+};
|
|
|
+
|
|
|
+type UpsertOutcome =
|
|
|
+ | { ok: true }
|
|
|
+ | { ok: false; error: { id?: string; error: string } };
|
|
|
+
|
|
|
+type UpsertTagsResult = {
|
|
|
+ unique: number;
|
|
|
+ upserted: number;
|
|
|
+ skipped: number;
|
|
|
+ errors: Array<{ name: string; error: string }>;
|
|
|
+};
|
|
|
+
|
|
|
@Injectable()
|
|
|
export class ProviderVideoSyncService {
|
|
|
private readonly logger = new Logger(ProviderVideoSyncService.name);
|
|
|
@@ -80,218 +134,287 @@ export class ProviderVideoSyncService {
|
|
|
private readonly PROVIDER_API_URL =
|
|
|
'https://vm.rvakc.xyz/api/web/mediafile/search';
|
|
|
|
|
|
+ private readonly DEFAULT_PROVIDER_CODE = 'RVAKC';
|
|
|
+ private readonly MAX_PAGE_SIZE = 500;
|
|
|
+ private readonly DEFAULT_PAGE_SIZE = 200;
|
|
|
+ private readonly BATCH_SIZE = 100;
|
|
|
+
|
|
|
constructor(
|
|
|
private readonly mongo: MongoPrismaService,
|
|
|
private readonly httpService: HttpService,
|
|
|
) {}
|
|
|
|
|
|
- /**
|
|
|
- * Sync video media from provider(s).
|
|
|
- * Fetches data from external provider API, normalizes it, and upserts to MongoDB.
|
|
|
- *
|
|
|
- * @param options Configuration options (providerCode, page, pageSize, etc.)
|
|
|
- * @returns Sync result with counts and errors
|
|
|
- */
|
|
|
async syncFromProvider(
|
|
|
options: ProviderVideoSyncOptions = {},
|
|
|
): Promise<ProviderVideoSyncResult> {
|
|
|
- this.logger.log('[syncFromProvider] Starting provider video sync');
|
|
|
- this.logger.debug('[syncFromProvider] Options:', options);
|
|
|
+ const providerCode = options.providerCode ?? this.DEFAULT_PROVIDER_CODE;
|
|
|
|
|
|
- const { providerCode = 'RVAKC', page = 1, pageSize = 100 } = options;
|
|
|
+ const requestedPageSize = options.pageSize ?? this.DEFAULT_PAGE_SIZE;
|
|
|
+ const pageSize = this.clampInt(requestedPageSize, 1, this.MAX_PAGE_SIZE);
|
|
|
|
|
|
- try {
|
|
|
- // Fetch data from provider API
|
|
|
- this.logger.log(
|
|
|
- `[syncFromProvider] Fetching from provider API: ${this.PROVIDER_API_URL}`,
|
|
|
- );
|
|
|
+ const fullSync = options.fullSync ?? false;
|
|
|
+ const resetState = options.resetState ?? false;
|
|
|
|
|
|
- let rawList: RawProviderVideo[] = [];
|
|
|
+ const paramStatus = options.param?.status ?? 'Completed';
|
|
|
+ const optionUpdatedAt = options.param?.updatedAt;
|
|
|
|
|
|
- try {
|
|
|
- const response = await firstValueFrom(
|
|
|
- this.httpService.get(this.PROVIDER_API_URL, {
|
|
|
- params: {
|
|
|
- page,
|
|
|
- pageSize,
|
|
|
- providerCode,
|
|
|
- },
|
|
|
- timeout: 30000,
|
|
|
- }),
|
|
|
- );
|
|
|
+ // Only one entity exists in your enum now
|
|
|
+ const entity = EntityType.VIDEO;
|
|
|
|
|
|
- rawList = this.extractList(response.data);
|
|
|
- this.logger.log(
|
|
|
- `[syncFromProvider] Extracted ${rawList.length} items from provider API`,
|
|
|
- );
|
|
|
- } catch (error: any) {
|
|
|
- this.logger.error(
|
|
|
- `[syncFromProvider] Provider API call failed: ${error.message}`,
|
|
|
- );
|
|
|
- const result: ProviderVideoSyncResult = {
|
|
|
- imported: 0,
|
|
|
- created: 0,
|
|
|
- updated: 0,
|
|
|
- skipped: 0,
|
|
|
- errors: [{ error: `Provider API error: ${error.message}` }],
|
|
|
- };
|
|
|
- this.lastSyncSummary = result;
|
|
|
- return result;
|
|
|
- }
|
|
|
+ this.logger.log(
|
|
|
+ `[syncFromProvider] Start provider=${providerCode} entity=${entity} fullSync=${fullSync} pageSize=${pageSize} resetState=${resetState}`,
|
|
|
+ );
|
|
|
|
|
|
- if (!rawList.length) {
|
|
|
- this.logger.log('[syncFromProvider] No videos from provider to sync');
|
|
|
- const result: ProviderVideoSyncResult = {
|
|
|
- imported: 0,
|
|
|
- created: 0,
|
|
|
- updated: 0,
|
|
|
- skipped: 0,
|
|
|
- };
|
|
|
- this.lastSyncSummary = result;
|
|
|
- return result;
|
|
|
- }
|
|
|
+ // Load cursor from SyncState (or fresh if resetState)
|
|
|
+ const { cursor: initialCursor } = await this.loadCursor({
|
|
|
+ entity,
|
|
|
+ pageSize,
|
|
|
+ resetState,
|
|
|
+ overridePageNum: options.pageNum,
|
|
|
+ optionUpdatedAt,
|
|
|
+ fullSync,
|
|
|
+ });
|
|
|
+
|
|
|
+ // Counters (Option B: created always 0, updated counts successful upserts)
|
|
|
+ let imported = 0;
|
|
|
+ let updated = 0;
|
|
|
+ let skipped = 0;
|
|
|
+ const created = 0;
|
|
|
+ const errors: Array<{ id?: string; error: string }> = [];
|
|
|
+
|
|
|
+ // Track max updatedAt seen (for incremental cursor advancement)
|
|
|
+ let maxUpdatedAtSeen: Date | null = null;
|
|
|
+
|
|
|
+ // Full sync resumes with pageNum; incremental always starts at 1
|
|
|
+ let pageNum = fullSync ? initialCursor.pageNum : 1;
|
|
|
+
|
|
|
+ // Keep a working cursor that we will persist as we go
|
|
|
+ const cursor: SyncCursor = {
|
|
|
+ pageNum,
|
|
|
+ pageSize: initialCursor.pageSize,
|
|
|
+ updatedAtCursor: initialCursor.updatedAtCursor,
|
|
|
+ };
|
|
|
|
|
|
- // Normalize items
|
|
|
- let normalized: any[] = [];
|
|
|
+ try {
|
|
|
+ while (true) {
|
|
|
+ const body = this.buildProviderBody({
|
|
|
+ pageNum,
|
|
|
+ pageSize: cursor.pageSize,
|
|
|
+ status: paramStatus,
|
|
|
+ // fullSync: no updatedAt filter
|
|
|
+ updatedAt: fullSync
|
|
|
+ ? undefined
|
|
|
+ : (cursor.updatedAtCursor ?? optionUpdatedAt),
|
|
|
+ extraParam: options.param,
|
|
|
+ });
|
|
|
|
|
|
- try {
|
|
|
- normalized = rawList.map((item) => this.normalizeItem(item));
|
|
|
this.logger.log(
|
|
|
- `[syncFromProvider] Ready to import ${normalized.length} records`,
|
|
|
+ `[syncFromProvider] POST pageNum=${pageNum} pageSize=${cursor.pageSize} status=${paramStatus} updatedAt=${fullSync ? '(none)' : (body.param.updatedAt ?? '(none)')}`,
|
|
|
);
|
|
|
- this.logger.debug(
|
|
|
- '[syncFromProvider] First record sample:',
|
|
|
- normalized[0],
|
|
|
- );
|
|
|
- } catch (error: any) {
|
|
|
- this.logger.error(
|
|
|
- `[syncFromProvider] Normalization failed: ${error.message}`,
|
|
|
- );
|
|
|
- const result: ProviderVideoSyncResult = {
|
|
|
- imported: rawList.length,
|
|
|
- created: 0,
|
|
|
- updated: 0,
|
|
|
- skipped: rawList.length,
|
|
|
- errors: [{ error: `Normalization error: ${error.message}` }],
|
|
|
- };
|
|
|
- this.lastSyncSummary = result;
|
|
|
- return result;
|
|
|
- }
|
|
|
|
|
|
- // Batch processing - try to create each record individually and catch duplicate errors
|
|
|
- const BATCH_SIZE = 100;
|
|
|
- let created = 0;
|
|
|
- let updated = 0;
|
|
|
- let skipped = 0;
|
|
|
- const errors: Array<{ id?: string; error: string }> = [];
|
|
|
-
|
|
|
- for (let i = 0; i < normalized.length; i += BATCH_SIZE) {
|
|
|
- const batch = normalized.slice(i, i + BATCH_SIZE);
|
|
|
- this.logger.debug(
|
|
|
- `[syncFromProvider] Processing batch ${i / BATCH_SIZE + 1}, size: ${batch.length}`,
|
|
|
+ const rawList = await this.fetchPage(body);
|
|
|
+ if (!rawList.length) {
|
|
|
+ this.logger.log(
|
|
|
+ `[syncFromProvider] No more records (pageNum=${pageNum}). Stop.`,
|
|
|
+ );
|
|
|
+
|
|
|
+ // On completion:
|
|
|
+ // - fullSync: reset pageNum to 1 and set lastFullSyncAt
|
|
|
+ // - incremental: advance updatedAtCursor to maxUpdatedAtSeen, keep pageNum=1
|
|
|
+ const fullSyncCompleted = fullSync;
|
|
|
+
|
|
|
+ if (fullSync) {
|
|
|
+ cursor.pageNum = 1;
|
|
|
+ // Optional visibility: store last seen updatedAt as updatedAtCursor too
|
|
|
+ if (maxUpdatedAtSeen)
|
|
|
+ cursor.updatedAtCursor = maxUpdatedAtSeen.toISOString();
|
|
|
+ } else {
|
|
|
+ cursor.pageNum = 1;
|
|
|
+ if (maxUpdatedAtSeen)
|
|
|
+ cursor.updatedAtCursor = maxUpdatedAtSeen.toISOString();
|
|
|
+ }
|
|
|
+
|
|
|
+ await this.saveCursor({
|
|
|
+ entity,
|
|
|
+ cursor,
|
|
|
+ fullSyncCompleted,
|
|
|
+ });
|
|
|
+
|
|
|
+ const result: ProviderVideoSyncResult = {
|
|
|
+ imported,
|
|
|
+ created,
|
|
|
+ updated,
|
|
|
+ skipped,
|
|
|
+ errors: errors.length ? errors.slice(0, 10) : undefined,
|
|
|
+ };
|
|
|
+ this.lastSyncSummary = result;
|
|
|
+ return result;
|
|
|
+ }
|
|
|
+
|
|
|
+ imported += rawList.length;
|
|
|
+
|
|
|
+ const normalized = rawList.map((item) => this.normalizeItem(item));
|
|
|
+
|
|
|
+ const hasSecondTags = normalized.some(
|
|
|
+ (v) => Array.isArray(v.secondTags) && v.secondTags.length > 0,
|
|
|
);
|
|
|
|
|
|
- // eslint-disable-next-line no-await-in-loop
|
|
|
- await Promise.all(
|
|
|
- batch.map(async (record) => {
|
|
|
- try {
|
|
|
- // Try to create the record
|
|
|
- await this.mongo.videoMedia.create({ data: record });
|
|
|
- // eslint-disable-next-line no-plusplus
|
|
|
- created++;
|
|
|
- this.logger.debug(
|
|
|
- `[syncFromProvider] Created record: ${record.id}`,
|
|
|
- );
|
|
|
- } catch (error: any) {
|
|
|
- this.logger.debug(
|
|
|
- `[syncFromProvider] Create failed for ${record.id}: ${error.code} ${error.message?.substring(0, 100)}`,
|
|
|
- );
|
|
|
- // If duplicate key error (code 11000), try to update
|
|
|
- if (
|
|
|
- error.code === 11000 ||
|
|
|
- error.message?.includes('duplicate')
|
|
|
- ) {
|
|
|
- try {
|
|
|
- const { id, ...updateData } = record;
|
|
|
- await this.mongo.videoMedia.update({
|
|
|
- where: { id },
|
|
|
- data: updateData,
|
|
|
- });
|
|
|
- // eslint-disable-next-line no-plusplus
|
|
|
- updated++;
|
|
|
- this.logger.debug(`[syncFromProvider] Updated record: ${id}`);
|
|
|
- } catch (updateError: any) {
|
|
|
- this.logger.error(
|
|
|
- `[syncFromProvider] Update failed for ${record.id}: ${updateError.message}`,
|
|
|
- );
|
|
|
- // eslint-disable-next-line no-plusplus
|
|
|
- skipped++;
|
|
|
- errors.push({ id: record.id, error: updateError.message });
|
|
|
- }
|
|
|
- } else {
|
|
|
- this.logger.error(
|
|
|
- `[syncFromProvider] Skipped ${record.id}: ${error.message}`,
|
|
|
- );
|
|
|
- // eslint-disable-next-line no-plusplus
|
|
|
- skipped++;
|
|
|
- errors.push({ id: record.id, error: error.message });
|
|
|
- }
|
|
|
- }
|
|
|
- }),
|
|
|
+ if (hasSecondTags) {
|
|
|
+ await this.upsertSecondTagsFromVideos_NoUniqueName(normalized);
|
|
|
+ }
|
|
|
+
|
|
|
+ // update maxUpdatedAtSeen for cursor advance (incremental correctness)
|
|
|
+ for (const n of normalized) {
|
|
|
+ const d = n.updatedAt as Date;
|
|
|
+ if (!maxUpdatedAtSeen || d.getTime() > maxUpdatedAtSeen.getTime()) {
|
|
|
+ maxUpdatedAtSeen = d;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // Upsert in batches (Option B)
|
|
|
+ for (let i = 0; i < normalized.length; i += this.BATCH_SIZE) {
|
|
|
+ const batch = normalized.slice(i, i + this.BATCH_SIZE);
|
|
|
+
|
|
|
+ // eslint-disable-next-line no-await-in-loop
|
|
|
+ const outcomes = await Promise.all(
|
|
|
+ batch.map((r) => this.upsertOne(r)),
|
|
|
+ );
|
|
|
+
|
|
|
+ const okCount = outcomes.filter((o) => o.ok).length;
|
|
|
+ const fail = outcomes.filter((o) => !o.ok) as Array<
|
|
|
+ Extract<UpsertOutcome, { ok: false }>
|
|
|
+ >;
|
|
|
+
|
|
|
+ updated += okCount;
|
|
|
+ skipped += fail.length;
|
|
|
+ for (const f of fail) errors.push(f.error);
|
|
|
+ }
|
|
|
+
|
|
|
+ // Persist progress so we can resume on crash
|
|
|
+ if (fullSync) {
|
|
|
+ cursor.pageNum = pageNum + 1;
|
|
|
+ // Optional: keep moving max updatedAt for visibility
|
|
|
+ if (maxUpdatedAtSeen)
|
|
|
+ cursor.updatedAtCursor = maxUpdatedAtSeen.toISOString();
|
|
|
+ } else {
|
|
|
+ // incremental resumes by updatedAtCursor, so keep pageNum=1
|
|
|
+ cursor.pageNum = 1;
|
|
|
+ if (maxUpdatedAtSeen)
|
|
|
+ cursor.updatedAtCursor = maxUpdatedAtSeen.toISOString();
|
|
|
+ }
|
|
|
+
|
|
|
+ await this.saveCursor({
|
|
|
+ entity,
|
|
|
+ cursor,
|
|
|
+ fullSyncCompleted: false,
|
|
|
+ });
|
|
|
+
|
|
|
+ pageNum += 1;
|
|
|
+ }
|
|
|
+ } catch (e: any) {
|
|
|
+ this.logger.error(
|
|
|
+ `[syncFromProvider] Unexpected error: ${e?.message ?? e}`,
|
|
|
+ );
|
|
|
+
|
|
|
+ // Best-effort cursor persistence
|
|
|
+ try {
|
|
|
+ if (!fullSync && maxUpdatedAtSeen) {
|
|
|
+ cursor.pageNum = 1;
|
|
|
+ cursor.updatedAtCursor = maxUpdatedAtSeen.toISOString();
|
|
|
+ }
|
|
|
+ await this.saveCursor({
|
|
|
+ entity,
|
|
|
+ cursor,
|
|
|
+ fullSyncCompleted: false,
|
|
|
+ });
|
|
|
+ } catch (saveErr: any) {
|
|
|
+ this.logger.error(
|
|
|
+ `[syncFromProvider] Failed to persist cursor after error: ${saveErr?.message ?? saveErr}`,
|
|
|
);
|
|
|
}
|
|
|
|
|
|
const result: ProviderVideoSyncResult = {
|
|
|
- imported: normalized.length,
|
|
|
+ imported,
|
|
|
created,
|
|
|
updated,
|
|
|
skipped,
|
|
|
- errors: errors.length > 0 ? errors.slice(0, 10) : undefined,
|
|
|
- };
|
|
|
-
|
|
|
- this.logger.log(
|
|
|
- `[syncFromProvider] Sync complete: ${created} created, ${updated} updated, ${skipped} skipped`,
|
|
|
- );
|
|
|
- if (errors.length > 0) {
|
|
|
- this.logger.log('[syncFromProvider] Errors:', errors.slice(0, 5));
|
|
|
- }
|
|
|
-
|
|
|
- this.lastSyncSummary = result;
|
|
|
- return result;
|
|
|
- } catch (error: any) {
|
|
|
- this.logger.error(
|
|
|
- `[syncFromProvider] Unexpected error: ${error.message}`,
|
|
|
- );
|
|
|
- const result: ProviderVideoSyncResult = {
|
|
|
- imported: 0,
|
|
|
- created: 0,
|
|
|
- updated: 0,
|
|
|
- skipped: 0,
|
|
|
- errors: [{ error: error.message || 'Unexpected error' }],
|
|
|
+ errors: [
|
|
|
+ ...(errors.length ? errors.slice(0, 9) : []),
|
|
|
+ { error: e?.message ?? 'Unexpected error' },
|
|
|
+ ],
|
|
|
};
|
|
|
this.lastSyncSummary = result;
|
|
|
return result;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- /**
|
|
|
- * Get the last sync summary.
|
|
|
- * Returns null if no sync has been performed yet.
|
|
|
- */
|
|
|
getLastSyncSummary(): ProviderVideoSyncResult | null {
|
|
|
return this.lastSyncSummary;
|
|
|
}
|
|
|
|
|
|
- /**
|
|
|
- * Extracts the list of items from different possible API response shapes.
|
|
|
- * Supports: { list: [...] }, { data: { list: [...] } }, or direct array
|
|
|
- */
|
|
|
+ private buildProviderBody(args: {
|
|
|
+ pageNum: number;
|
|
|
+ pageSize: number;
|
|
|
+ status: string;
|
|
|
+ updatedAt?: string;
|
|
|
+ extraParam?: ProviderVideoSyncOptions['param'];
|
|
|
+ }) {
|
|
|
+ // Provider contract:
|
|
|
+ // {
|
|
|
+ // pageNum: 1,
|
|
|
+ // pageSize: 200,
|
|
|
+ // param: { status: "Completed", updatedAt: "ISO" }
|
|
|
+ // }
|
|
|
+ const param: Record<string, any> = {
|
|
|
+ status: args.status,
|
|
|
+ };
|
|
|
+
|
|
|
+ // Keep only if present (incremental)
|
|
|
+ if (args.updatedAt) param.updatedAt = args.updatedAt;
|
|
|
+
|
|
|
+ // Merge any extraParam fields, but status/updatedAt above remain authoritative
|
|
|
+ if (args.extraParam && typeof args.extraParam === 'object') {
|
|
|
+ for (const [k, v] of Object.entries(args.extraParam)) {
|
|
|
+ if (k === 'status' || k === 'updatedAt') continue;
|
|
|
+ param[k] = v;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ return {
|
|
|
+ pageNum: args.pageNum,
|
|
|
+ pageSize: args.pageSize,
|
|
|
+ param,
|
|
|
+ };
|
|
|
+ }
|
|
|
+
|
|
|
+ private async fetchPage(body: {
|
|
|
+ pageNum: number;
|
|
|
+ pageSize: number;
|
|
|
+ param: Record<string, any>;
|
|
|
+ }): Promise<RawProviderVideo[]> {
|
|
|
+ try {
|
|
|
+ const response = await firstValueFrom(
|
|
|
+ this.httpService.post(this.PROVIDER_API_URL, body, {
|
|
|
+ headers: { 'Content-Type': 'application/json' },
|
|
|
+ timeout: 30000,
|
|
|
+ }),
|
|
|
+ );
|
|
|
+
|
|
|
+ const list = this.extractList(response.data);
|
|
|
+ this.logger.log(`[fetchPage] Received ${list.length} items`);
|
|
|
+ return list;
|
|
|
+ } catch (error: any) {
|
|
|
+ this.logger.error(
|
|
|
+ `[fetchPage] Provider API call failed: ${error?.message ?? error}`,
|
|
|
+ );
|
|
|
+ throw new Error(`Provider API error: ${error?.message ?? 'unknown'}`);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
private extractList(apiResponse: unknown): RawProviderVideo[] {
|
|
|
const data = apiResponse as any;
|
|
|
|
|
|
- if (Array.isArray(data)) {
|
|
|
- return data as RawProviderVideo[];
|
|
|
- }
|
|
|
+ if (Array.isArray(data)) return data as RawProviderVideo[];
|
|
|
|
|
|
if (data?.data?.list && Array.isArray(data.data.list)) {
|
|
|
return data.data.list as RawProviderVideo[];
|
|
|
@@ -301,21 +424,18 @@ export class ProviderVideoSyncService {
|
|
|
return data.list as RawProviderVideo[];
|
|
|
}
|
|
|
|
|
|
+ if (data?.data?.records && Array.isArray(data.data.records)) {
|
|
|
+ return data.data.records as RawProviderVideo[];
|
|
|
+ }
|
|
|
+
|
|
|
this.logger.warn(
|
|
|
'[extractList] Unexpected API response structure, defaulting to empty list',
|
|
|
);
|
|
|
return [];
|
|
|
}
|
|
|
|
|
|
- /**
|
|
|
- * Maps RawProviderVideo to Prisma videoMedia create/update input.
|
|
|
- * Applies defaults and type conversions to match the Prisma model.
|
|
|
- */
|
|
|
private normalizeItem(item: RawProviderVideo) {
|
|
|
- // Basic validation
|
|
|
- if (!item.id) {
|
|
|
- throw new Error('Each item must have an id');
|
|
|
- }
|
|
|
+ if (!item.id) throw new Error('Each item must have an id');
|
|
|
if (!item.addedTime || !item.createdAt || !item.updatedAt) {
|
|
|
throw new Error(`Item ${item.id} is missing required datetime fields`);
|
|
|
}
|
|
|
@@ -333,7 +453,7 @@ export class ProviderVideoSyncService {
|
|
|
}
|
|
|
|
|
|
return {
|
|
|
- id: item.id, // String mapped to Mongo ObjectId via @db.ObjectId
|
|
|
+ id: item.id, // confirmed Mongo ObjectId string
|
|
|
|
|
|
srcId: item.srcId ?? 0,
|
|
|
title: item.title ?? '',
|
|
|
@@ -353,30 +473,21 @@ export class ProviderVideoSyncService {
|
|
|
country: item.country ?? '',
|
|
|
firstTag: item.firstTag ?? '',
|
|
|
|
|
|
- // null → []
|
|
|
secondTags: item.secondTags ?? [],
|
|
|
-
|
|
|
- // null → ""
|
|
|
mediaSet: item.mediaSet ?? '',
|
|
|
-
|
|
|
preFileName: item.preFileName ?? '',
|
|
|
|
|
|
status: item.status ?? '',
|
|
|
desc: item.desc ?? '',
|
|
|
|
|
|
- // number → BigInt
|
|
|
size: BigInt(item.size ?? 0),
|
|
|
|
|
|
bango: item.bango ?? '',
|
|
|
-
|
|
|
- // null → []
|
|
|
actors: item.actors ?? [],
|
|
|
-
|
|
|
studio: item.studio ?? '',
|
|
|
|
|
|
addedTime,
|
|
|
appids: item.appids ?? [],
|
|
|
-
|
|
|
japanNames: item.japanNames ?? [],
|
|
|
|
|
|
filename: item.filename ?? '',
|
|
|
@@ -410,4 +521,330 @@ export class ProviderVideoSyncService {
|
|
|
updatedAt,
|
|
|
};
|
|
|
}
|
|
|
+
|
|
|
+ private async upsertOne(record: any): Promise<UpsertOutcome> {
|
|
|
+ const id = record?.id as string | undefined;
|
|
|
+ if (!id) return { ok: false, error: { error: 'Missing id' } };
|
|
|
+
|
|
|
+ try {
|
|
|
+ const { id: _, ...updateData } = record;
|
|
|
+
|
|
|
+ await this.mongo.videoMedia.upsert({
|
|
|
+ where: { id },
|
|
|
+ create: record,
|
|
|
+ update: updateData,
|
|
|
+ });
|
|
|
+
|
|
|
+ return { ok: true };
|
|
|
+ } catch (e: any) {
|
|
|
+ return {
|
|
|
+ ok: false,
|
|
|
+ error: { id, error: e?.message ?? 'Upsert failed' },
|
|
|
+ };
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private async loadCursor(args: {
|
|
|
+ entity: EntityType;
|
|
|
+ pageSize: number;
|
|
|
+ resetState: boolean;
|
|
|
+ overridePageNum?: number;
|
|
|
+ optionUpdatedAt?: string;
|
|
|
+ fullSync: boolean;
|
|
|
+ }): Promise<{ cursor: SyncCursor; hasState: boolean }> {
|
|
|
+ const nowSec = Math.floor(Date.now() / 1000);
|
|
|
+
|
|
|
+ if (args.resetState) {
|
|
|
+ return {
|
|
|
+ cursor: {
|
|
|
+ pageNum: args.overridePageNum ?? 1,
|
|
|
+ pageSize: args.pageSize,
|
|
|
+ updatedAtCursor: args.fullSync ? undefined : args.optionUpdatedAt,
|
|
|
+ },
|
|
|
+ hasState: false,
|
|
|
+ };
|
|
|
+ }
|
|
|
+
|
|
|
+ const state = await this.mongo.syncState.upsert({
|
|
|
+ where: { entity: args.entity },
|
|
|
+ update: {
|
|
|
+ updatedAt: nowSec,
|
|
|
+ },
|
|
|
+ create: {
|
|
|
+ entity: args.entity,
|
|
|
+ referId: null,
|
|
|
+ lastRunAt: null,
|
|
|
+ lastFullSyncAt: null,
|
|
|
+ createdAt: nowSec,
|
|
|
+ updatedAt: nowSec,
|
|
|
+ },
|
|
|
+ });
|
|
|
+
|
|
|
+ const parsed = this.safeParseCursor(state.referId);
|
|
|
+
|
|
|
+ const cursor: SyncCursor = {
|
|
|
+ pageNum: args.overridePageNum ?? parsed?.pageNum ?? 1,
|
|
|
+ pageSize: args.pageSize,
|
|
|
+ updatedAtCursor: args.fullSync
|
|
|
+ ? undefined
|
|
|
+ : (parsed?.updatedAtCursor ?? args.optionUpdatedAt),
|
|
|
+ };
|
|
|
+
|
|
|
+ return { cursor, hasState: Boolean(state.referId) };
|
|
|
+ }
|
|
|
+
|
|
|
+ private async saveCursor(args: {
|
|
|
+ entity: EntityType;
|
|
|
+ cursor: SyncCursor;
|
|
|
+ fullSyncCompleted: boolean;
|
|
|
+ }) {
|
|
|
+ const now = new Date();
|
|
|
+ const nowSec = Math.floor(Date.now() / 1000);
|
|
|
+
|
|
|
+ await this.mongo.syncState.update({
|
|
|
+ where: { entity: args.entity },
|
|
|
+ data: {
|
|
|
+ referId: JSON.stringify(args.cursor),
|
|
|
+ lastRunAt: now,
|
|
|
+ lastFullSyncAt: args.fullSyncCompleted ? now : undefined,
|
|
|
+ updatedAt: nowSec,
|
|
|
+ },
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ private async saveCheckpoint(args: {
|
|
|
+ entity: EntityType;
|
|
|
+ nextUpdatedAtCursor?: string; // ONLY when batch completed
|
|
|
+ fullSyncCompleted: boolean;
|
|
|
+ }) {
|
|
|
+ const now = new Date();
|
|
|
+ const nowSec = Math.floor(Date.now() / 1000);
|
|
|
+
|
|
|
+ // Build referId safely (do not overwrite blindly)
|
|
|
+ const state = await this.mongo.syncState.findUnique({
|
|
|
+ where: { entity: args.entity },
|
|
|
+ select: { referId: true },
|
|
|
+ });
|
|
|
+
|
|
|
+ let persisted: { updatedAtCursor?: string } = {};
|
|
|
+ if (state?.referId) {
|
|
|
+ try {
|
|
|
+ persisted = JSON.parse(state.referId);
|
|
|
+ } catch {
|
|
|
+ persisted = {};
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // Only commit updatedAtCursor when explicitly provided
|
|
|
+ if (args.nextUpdatedAtCursor) {
|
|
|
+ persisted.updatedAtCursor = args.nextUpdatedAtCursor;
|
|
|
+ }
|
|
|
+
|
|
|
+ await this.mongo.syncState.update({
|
|
|
+ where: { entity: args.entity },
|
|
|
+ data: {
|
|
|
+ referId: JSON.stringify(persisted),
|
|
|
+ lastRunAt: now,
|
|
|
+ lastFullSyncAt: args.fullSyncCompleted ? now : undefined,
|
|
|
+ updatedAt: nowSec,
|
|
|
+ },
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ private safeParseCursor(
|
|
|
+ raw: string | null | undefined,
|
|
|
+ ): Partial<SyncCursor> | null {
|
|
|
+ if (!raw) return null;
|
|
|
+ try {
|
|
|
+ const parsed = JSON.parse(raw) as any;
|
|
|
+ if (!parsed || typeof parsed !== 'object') return null;
|
|
|
+ return parsed as Partial<SyncCursor>;
|
|
|
+ } catch {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private clampInt(n: number, min: number, max: number): number {
|
|
|
+ const x = Number.isFinite(n) ? Math.trunc(n) : min;
|
|
|
+ return Math.max(min, Math.min(max, x));
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Extract secondTags (string[]) from normalized video records and upsert into Tag collection.
|
|
|
+ * - Dedup in-memory per call for performance
|
|
|
+ * - Trims whitespace, filters empty
|
|
|
+ * - Option B performance-first: upsert without pre-check
|
|
|
+ */
|
|
|
+ // private async upsertSecondTagsFromVideos(
|
|
|
+ // normalizedVideos: Array<{ secondTags?: string[] }>,
|
|
|
+ // ): Promise<UpsertTagsResult> {
|
|
|
+ // // 1) Collect + normalize
|
|
|
+ // const set = new Set<string>();
|
|
|
+
|
|
|
+ // for (const v of normalizedVideos) {
|
|
|
+ // const tags = v.secondTags ?? [];
|
|
|
+ // for (const t of tags) {
|
|
|
+ // if (typeof t !== 'string') continue;
|
|
|
+ // const name = t.trim();
|
|
|
+ // if (!name) continue;
|
|
|
+ // set.add(name);
|
|
|
+ // }
|
|
|
+ // }
|
|
|
+
|
|
|
+ // const names = Array.from(set);
|
|
|
+ // if (!names.length) {
|
|
|
+ // return { unique: 0, upserted: 0, skipped: 0, errors: [] };
|
|
|
+ // }
|
|
|
+
|
|
|
+ // // 2) Upsert in chunks (avoid massive Promise.all)
|
|
|
+ // const CHUNK = 200;
|
|
|
+ // let upserted = 0;
|
|
|
+ // let skipped = 0;
|
|
|
+ // const errors: Array<{ name: string; error: string }> = [];
|
|
|
+
|
|
|
+ // for (let i = 0; i < names.length; i += CHUNK) {
|
|
|
+ // const batch = names.slice(i, i + CHUNK);
|
|
|
+
|
|
|
+ // // eslint-disable-next-line no-await-in-loop
|
|
|
+ // const outcomes = await Promise.all(
|
|
|
+ // batch.map(async (name) => {
|
|
|
+ // try {
|
|
|
+ // // 🔁 Adjust `where/create/update` if your Tag schema differs
|
|
|
+ // await this.mongo.tag.upsert({
|
|
|
+ // where: { name },
|
|
|
+ // create: {
|
|
|
+ // name,
|
|
|
+ // // If Tag requires createdAt/updatedAt ints (seconds), uncomment:
|
|
|
+ // // createdAt: Math.floor(Date.now() / 1000),
|
|
|
+ // // updatedAt: Math.floor(Date.now() / 1000),
|
|
|
+ // },
|
|
|
+ // update: {
|
|
|
+ // // keep it minimal; optionally touch updatedAt
|
|
|
+ // // updatedAt: Math.floor(Date.now() / 1000),
|
|
|
+ // },
|
|
|
+ // });
|
|
|
+ // return { ok: true as const };
|
|
|
+ // } catch (e: any) {
|
|
|
+ // return {
|
|
|
+ // ok: false as const,
|
|
|
+ // error: e?.message ?? 'Tag upsert failed',
|
|
|
+ // };
|
|
|
+ // }
|
|
|
+ // }),
|
|
|
+ // );
|
|
|
+
|
|
|
+ // for (let j = 0; j < outcomes.length; j += 1) {
|
|
|
+ // const o = outcomes[j];
|
|
|
+ // if (o.ok) upserted += 1;
|
|
|
+ // else {
|
|
|
+ // skipped += 1;
|
|
|
+ // errors.push({ name: batch[j], error: o.error });
|
|
|
+ // }
|
|
|
+ // }
|
|
|
+ // }
|
|
|
+
|
|
|
+ // if (errors.length) {
|
|
|
+ // this.logger.warn(
|
|
|
+ // `[upsertSecondTagsFromVideos] tag upsert errors=${errors.length}, sample=${JSON.stringify(
|
|
|
+ // errors.slice(0, 3),
|
|
|
+ // )}`,
|
|
|
+ // );
|
|
|
+ // } else {
|
|
|
+ // this.logger.log(
|
|
|
+ // `[upsertSecondTagsFromVideos] Upserted tags=${upserted} (unique=${names.length})`,
|
|
|
+ // );
|
|
|
+ // }
|
|
|
+
|
|
|
+ // return {
|
|
|
+ // unique: names.length,
|
|
|
+ // upserted,
|
|
|
+ // skipped,
|
|
|
+ // errors,
|
|
|
+ // };
|
|
|
+ // }
|
|
|
+
|
|
|
+ private async upsertSecondTagsFromVideos_NoUniqueName(
|
|
|
+ normalizedVideos: Array<{ secondTags?: string[] }>,
|
|
|
+ ): Promise<UpsertTagsResult> {
|
|
|
+ const set = new Set<string>();
|
|
|
+
|
|
|
+ for (const v of normalizedVideos) {
|
|
|
+ const tags = v.secondTags ?? [];
|
|
|
+ for (const t of tags) {
|
|
|
+ if (typeof t !== 'string') continue;
|
|
|
+ const name = t.trim();
|
|
|
+ if (!name) continue;
|
|
|
+ set.add(name);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ const names = Array.from(set);
|
|
|
+ if (!names.length)
|
|
|
+ return { unique: 0, upserted: 0, skipped: 0, errors: [] };
|
|
|
+
|
|
|
+ // Concurrency limit to reduce race collisions and DB pressure
|
|
|
+ const CONCURRENCY = 20;
|
|
|
+ let idx = 0;
|
|
|
+
|
|
|
+ let upserted = 0;
|
|
|
+ let skipped = 0;
|
|
|
+ const errors: Array<{ name: string; error: string }> = [];
|
|
|
+
|
|
|
+ const worker = async () => {
|
|
|
+ while (true) {
|
|
|
+ const current = idx;
|
|
|
+ idx += 1;
|
|
|
+ if (current >= names.length) return;
|
|
|
+
|
|
|
+ const name = names[current];
|
|
|
+
|
|
|
+ try {
|
|
|
+ // 1) check existence by name (NOT unique)
|
|
|
+ const exists = await this.mongo.tag.findFirst({
|
|
|
+ where: { name },
|
|
|
+ select: { id: true },
|
|
|
+ });
|
|
|
+
|
|
|
+ if (exists?.id) {
|
|
|
+ // already exists
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 2) create if not exists
|
|
|
+ await this.mongo.tag.create({
|
|
|
+ data: {
|
|
|
+ name,
|
|
|
+ // If your Tag schema requires seconds fields:
|
|
|
+ // createdAt: Math.floor(Date.now() / 1000),
|
|
|
+ // updatedAt: Math.floor(Date.now() / 1000),
|
|
|
+ },
|
|
|
+ });
|
|
|
+
|
|
|
+ upserted += 1;
|
|
|
+ } catch (e: any) {
|
|
|
+ // If another worker created it after our check, create may fail (duplicate on some index)
|
|
|
+ // We treat that as skipped (safe).
|
|
|
+ const msg = e?.message ?? 'Tag create failed';
|
|
|
+ skipped += 1;
|
|
|
+ errors.push({ name, error: msg });
|
|
|
+ }
|
|
|
+ }
|
|
|
+ };
|
|
|
+
|
|
|
+ await Promise.all(Array.from({ length: CONCURRENCY }, () => worker()));
|
|
|
+
|
|
|
+ if (errors.length) {
|
|
|
+ this.logger.warn(
|
|
|
+ `[upsertSecondTagsFromVideos] errors=${errors.length}, sample=${JSON.stringify(
|
|
|
+ errors.slice(0, 3),
|
|
|
+ )}`,
|
|
|
+ );
|
|
|
+ } else {
|
|
|
+ this.logger.log(
|
|
|
+ `[upsertSecondTagsFromVideos] unique=${names.length} created=${upserted}`,
|
|
|
+ );
|
|
|
+ }
|
|
|
+
|
|
|
+ return { unique: names.length, upserted, skipped, errors };
|
|
|
+ }
|
|
|
}
|
FC_DAN\c9837