477 lines
19 KiB
TypeScript
477 lines
19 KiB
TypeScript
import { Injectable, Logger } from '@nestjs/common';
|
|
import { randomUUID } from 'crypto';
|
|
import { PrismaService } from '../../shared/prisma/prisma.service';
|
|
import { YouTubeApiClient } from './youtube-api.client';
|
|
import { hashMetadata } from '../../shared/render-engine/hash';
|
|
import { PrivacyStatus } from '@prisma/client';
|
|
|
|
const YT_PRIVACY: Record<string, PrivacyStatus> = {
|
|
public: PrivacyStatus.PUBLIC,
|
|
private: PrivacyStatus.PRIVATE,
|
|
unlisted: PrivacyStatus.UNLISTED,
|
|
};
|
|
|
|
type ActionCtx = { actionId?: string; actionType?: string };
|
|
|
|
@Injectable()
|
|
export class ChannelImportService {
|
|
private readonly logger = new Logger(ChannelImportService.name);
|
|
|
|
constructor(
|
|
private readonly prisma: PrismaService,
|
|
private readonly ytApi: YouTubeApiClient,
|
|
) {}
|
|
|
|
async importChannel(
|
|
channelId: string,
|
|
teamId: string,
|
|
): Promise<{ total: number; created: number; updated: number; deleted: number; deletedTitles: string[] }> {
|
|
this.logger.log(`Channel import started for channel ${channelId}`);
|
|
|
|
const ctx: ActionCtx = { actionId: randomUUID(), actionType: 'channel_import' };
|
|
|
|
const channel = await this.prisma.channel.findFirstOrThrow({
|
|
where: { id: channelId, teamId },
|
|
});
|
|
|
|
const playlistId =
|
|
channel.uploadsPlaylistId ?? (await this.ytApi.getUploadsPlaylistId(channelId, ctx));
|
|
|
|
const { ids: rawIds, pageCount } = await this.ytApi.listPlaylistVideoIds(channelId, playlistId, ctx);
|
|
const playlistIds = new Set(rawIds);
|
|
if (rawIds.length !== playlistIds.size) {
|
|
const seen = new Set<string>();
|
|
const dupes = rawIds.filter((id) => seen.size === seen.add(id).size);
|
|
this.logger.warn(`Playlist has ${rawIds.length - playlistIds.size} duplicate entries: ${[...new Set(dupes)].join(', ')}`);
|
|
}
|
|
|
|
const supplemental = channel.supplementalVideoIds ?? [];
|
|
const newSupplemental = supplemental.filter((id) => !playlistIds.has(id));
|
|
if (newSupplemental.length > 0) {
|
|
this.logger.log(`Adding ${newSupplemental.length} supplemental video IDs not in playlist: ${newSupplemental.join(', ')}`);
|
|
}
|
|
|
|
const ids = [...playlistIds, ...newSupplemental];
|
|
this.logger.log(`Found ${playlistIds.size} unique videos (${rawIds.length} playlist entries) across ${pageCount} pages, plus ${newSupplemental.length} supplemental`);
|
|
|
|
let created = 0;
|
|
let updated = 0;
|
|
|
|
for (let i = 0; i < ids.length; i += 50) {
|
|
const batch = ids.slice(i, i + 50);
|
|
const items = await this.ytApi.getVideosBatch(batch, channelId, ctx);
|
|
|
|
const returnedIds = new Set(items.map((item) => item.id));
|
|
const missing = batch.filter((id) => !returnedIds.has(id));
|
|
if (missing.length > 0) {
|
|
this.logger.warn(`videos.list did not return ${missing.length} IDs: ${missing.join(', ')}`);
|
|
}
|
|
|
|
for (const item of items) {
|
|
const result = await this.upsertVideoFromYouTubeItem(item, channelId);
|
|
if (result === 'created') created++;
|
|
else if (result === 'updated') updated++;
|
|
}
|
|
}
|
|
|
|
await this.syncPlaylistMembership(channelId, ctx);
|
|
|
|
const purgeResult = await this._purgeForChannel(channelId);
|
|
|
|
this.logger.log(`Import complete: ${created} created, ${updated} updated, ${purgeResult.softDeleted} soft-deleted, ${purgeResult.hardDeleted} hard-deleted, ${purgeResult.restored} restored`);
|
|
|
|
return { total: ids.length, created, updated, deleted: purgeResult.softDeleted + purgeResult.hardDeleted, deletedTitles: purgeResult.softDeletedTitles };
|
|
}
|
|
|
|
async importSpecificVideos(
|
|
channelId: string,
|
|
teamId: string,
|
|
videoIds: string[],
|
|
): Promise<{ total: number; created: number; updated: number; notFound: string[] }> {
|
|
this.logger.log(`Importing ${videoIds.length} specific video IDs for channel ${channelId}`);
|
|
|
|
const ctx: ActionCtx = { actionId: randomUUID(), actionType: 'channel_import' };
|
|
|
|
await this.prisma.channel.findFirstOrThrow({ where: { id: channelId, teamId } });
|
|
|
|
const unique = [...new Set(videoIds)];
|
|
let created = 0;
|
|
let updated = 0;
|
|
const notFoundIds: string[] = [];
|
|
|
|
for (let i = 0; i < unique.length; i += 50) {
|
|
const batch = unique.slice(i, i + 50);
|
|
const items = await this.ytApi.getVideosBatch(batch, channelId, ctx);
|
|
|
|
const returnedIds = new Set(items.map((item) => item.id));
|
|
notFoundIds.push(...batch.filter((id) => !returnedIds.has(id)));
|
|
|
|
for (const item of items) {
|
|
const result = await this.upsertVideoFromYouTubeItem(item, channelId);
|
|
if (result === 'created') created++;
|
|
else if (result === 'updated') updated++;
|
|
}
|
|
}
|
|
|
|
if (notFoundIds.length > 0) {
|
|
this.logger.warn(`YouTube did not return ${notFoundIds.length} IDs: ${notFoundIds.join(', ')}`);
|
|
}
|
|
this.logger.log(`Specific import complete: ${created} created, ${updated} updated, ${notFoundIds.length} not found`);
|
|
return { total: unique.length, created, updated, notFound: notFoundIds };
|
|
}
|
|
|
|
async fullRefreshChannel(
|
|
channelId: string,
|
|
teamId: string,
|
|
): Promise<{
|
|
total: number;
|
|
created: number;
|
|
updated: number;
|
|
duplicateUploadsEntries: number;
|
|
notFoundOnYouTube: string[];
|
|
playlistsForceSynced: number;
|
|
orphansImported: number;
|
|
orphansRejected: number;
|
|
deleted: number;
|
|
deletedTitles: string[];
|
|
}> {
|
|
this.logger.log(`Full refresh started for channel ${channelId}`);
|
|
|
|
const ctx: ActionCtx = { actionId: randomUUID(), actionType: 'channel_import' };
|
|
|
|
const channel = await this.prisma.channel.findFirstOrThrow({
|
|
where: { id: channelId, teamId },
|
|
});
|
|
|
|
const playlistId =
|
|
channel.uploadsPlaylistId ?? (await this.ytApi.getUploadsPlaylistId(channelId, ctx));
|
|
|
|
const { ids: rawIds, pageCount } = await this.ytApi.listPlaylistVideoIds(channelId, playlistId, ctx);
|
|
const uploadsIds = new Set(rawIds);
|
|
|
|
const duplicateUploadsEntries = rawIds.length - uploadsIds.size;
|
|
if (duplicateUploadsEntries > 0) {
|
|
const seen = new Set<string>();
|
|
const dupes = rawIds.filter((id) => seen.size === seen.add(id).size);
|
|
this.logger.warn(`Uploads playlist has ${duplicateUploadsEntries} duplicate entries: ${[...new Set(dupes)].join(', ')}`);
|
|
}
|
|
|
|
const supplemental = channel.supplementalVideoIds ?? [];
|
|
const newSupplemental = supplemental.filter((id) => !uploadsIds.has(id));
|
|
const ids = [...uploadsIds, ...newSupplemental];
|
|
|
|
this.logger.log(`Full refresh: ${uploadsIds.size} unique videos (${rawIds.length} entries) across ${pageCount} pages, plus ${newSupplemental.length} supplemental`);
|
|
|
|
let created = 0;
|
|
let updated = 0;
|
|
const notFoundOnYouTube: string[] = [];
|
|
|
|
for (let i = 0; i < ids.length; i += 50) {
|
|
const batch = ids.slice(i, i + 50);
|
|
const items = await this.ytApi.getVideosBatch(batch, channelId, ctx);
|
|
|
|
const returnedIds = new Set(items.map((item) => item.id));
|
|
notFoundOnYouTube.push(...batch.filter((id) => !returnedIds.has(id)));
|
|
|
|
for (const item of items) {
|
|
const result = await this.upsertVideoFromYouTubeItem(item, channelId);
|
|
if (result === 'created') created++;
|
|
else if (result === 'updated') updated++;
|
|
}
|
|
}
|
|
|
|
if (notFoundOnYouTube.length > 0) {
|
|
this.logger.warn(`YouTube did not return ${notFoundOnYouTube.length} IDs during full refresh: ${notFoundOnYouTube.join(', ')}`);
|
|
}
|
|
|
|
// Force-sync all playlists and collect every video ID seen across them
|
|
const { synced: playlistsForceSynced, allVideoIds: playlistVideoIds } =
|
|
await this.syncPlaylistMembership(channelId, ctx, true);
|
|
|
|
// Find IDs that appear in other playlists but not in the uploads playlist or supplemental list
|
|
const knownIds = new Set([...uploadsIds, ...supplemental]);
|
|
const orphanCandidates = [...playlistVideoIds].filter((id) => !knownIds.has(id));
|
|
|
|
let orphansImported = 0;
|
|
let orphansRejected = 0;
|
|
|
|
if (orphanCandidates.length > 0) {
|
|
this.logger.log(`Found ${orphanCandidates.length} playlist video IDs not in uploads playlist — verifying channel ownership`);
|
|
|
|
for (let i = 0; i < orphanCandidates.length; i += 50) {
|
|
const batch = orphanCandidates.slice(i, i + 50);
|
|
const items = await this.ytApi.getVideosBatch(batch, channelId, ctx);
|
|
|
|
for (const item of items) {
|
|
// Only import videos that actually belong to this channel
|
|
if (item.snippet?.channelId !== channel.youtubeChannelId) {
|
|
this.logger.log(`Skipping ${item.id} — belongs to channel ${item.snippet?.channelId}, not ${channel.youtubeChannelId}`);
|
|
orphansRejected++;
|
|
continue;
|
|
}
|
|
const result = await this.upsertVideoFromYouTubeItem(item, channelId);
|
|
if (result === 'created' || result === 'updated') {
|
|
orphansImported++;
|
|
created += result === 'created' ? 1 : 0;
|
|
updated += result === 'updated' ? 1 : 0;
|
|
}
|
|
}
|
|
}
|
|
|
|
if (orphansImported > 0) {
|
|
this.logger.log(`Imported ${orphansImported} orphaned videos found in playlists but not in uploads playlist`);
|
|
}
|
|
if (orphansRejected > 0) {
|
|
this.logger.warn(`Rejected ${orphansRejected} playlist videos — belong to a different channel`);
|
|
}
|
|
}
|
|
|
|
const purgeResult = await this._purgeForChannel(channelId);
|
|
|
|
this.logger.log(`Full refresh complete: ${created} created, ${updated} updated, ${playlistsForceSynced} playlists force-synced, ${orphansImported} orphans imported, ${purgeResult.softDeleted} soft-deleted, ${purgeResult.hardDeleted} hard-deleted, ${purgeResult.restored} restored`);
|
|
|
|
return {
|
|
total: ids.length + orphansImported,
|
|
created,
|
|
updated,
|
|
duplicateUploadsEntries,
|
|
notFoundOnYouTube,
|
|
playlistsForceSynced,
|
|
orphansImported,
|
|
orphansRejected,
|
|
deleted: purgeResult.softDeleted + purgeResult.hardDeleted,
|
|
deletedTitles: purgeResult.softDeletedTitles,
|
|
};
|
|
}
|
|
|
|
async purgeDeletedVideos(channelId: string, teamId: string) {
|
|
await this.prisma.channel.findFirstOrThrow({ where: { id: channelId, teamId } });
|
|
return this._purgeForChannel(channelId);
|
|
}
|
|
|
|
private async _purgeForChannel(channelId: string): Promise<{ checked: number; softDeleted: number; restored: number; hardDeleted: number; softDeletedTitles: string[] }> {
|
|
const localVideos = await this.prisma.video.findMany({
|
|
where: { channelId },
|
|
select: { id: true, youtubeVideoId: true, title: true, youtubeDeletedAt: true },
|
|
});
|
|
|
|
const ctx: ActionCtx = { actionId: randomUUID(), actionType: 'purge_deleted' };
|
|
const softDeleteIds: string[] = [];
|
|
const softDeletedTitles: string[] = [];
|
|
const hardDeleteIds: string[] = [];
|
|
const restoreIds: string[] = [];
|
|
const thirtyDaysAgo = new Date(Date.now() - 30 * 24 * 60 * 60 * 1000);
|
|
|
|
for (let i = 0; i < localVideos.length; i += 50) {
|
|
const batch = localVideos.slice(i, i + 50);
|
|
const items = await this.ytApi.getVideosBatch(batch.map((v) => v.youtubeVideoId), channelId, ctx);
|
|
// Videos returned with uploadStatus !== 'deleted' are live on YouTube
|
|
const liveYtIds = new Set(
|
|
items.filter((item) => item.status?.uploadStatus !== 'deleted').map((item) => item.id),
|
|
);
|
|
for (const v of batch) {
|
|
if (!liveYtIds.has(v.youtubeVideoId)) {
|
|
if (v.youtubeDeletedAt && v.youtubeDeletedAt < thirtyDaysAgo) {
|
|
hardDeleteIds.push(v.id);
|
|
} else if (!v.youtubeDeletedAt) {
|
|
softDeleteIds.push(v.id);
|
|
softDeletedTitles.push(v.title);
|
|
}
|
|
// already soft-deleted and within 30 days — no action
|
|
} else if (v.youtubeDeletedAt) {
|
|
restoreIds.push(v.id);
|
|
}
|
|
}
|
|
}
|
|
|
|
if (softDeleteIds.length > 0) {
|
|
await this.prisma.video.updateMany({ where: { id: { in: softDeleteIds } }, data: { youtubeDeletedAt: new Date() } });
|
|
this.logger.log(`Soft-deleted ${softDeleteIds.length} videos from channel ${channelId}: ${softDeletedTitles.join(', ')}`);
|
|
}
|
|
if (hardDeleteIds.length > 0) {
|
|
await this.prisma.video.deleteMany({ where: { id: { in: hardDeleteIds } } });
|
|
this.logger.log(`Hard-deleted ${hardDeleteIds.length} expired videos from channel ${channelId}`);
|
|
}
|
|
if (restoreIds.length > 0) {
|
|
await this.prisma.video.updateMany({ where: { id: { in: restoreIds } }, data: { youtubeDeletedAt: null } });
|
|
this.logger.log(`Restored ${restoreIds.length} videos for channel ${channelId}`);
|
|
}
|
|
|
|
return {
|
|
checked: localVideos.length,
|
|
softDeleted: softDeleteIds.length,
|
|
restored: restoreIds.length,
|
|
hardDeleted: hardDeleteIds.length,
|
|
softDeletedTitles,
|
|
};
|
|
}
|
|
|
|
private async upsertVideoFromYouTubeItem(
|
|
item: any,
|
|
channelId: string,
|
|
): Promise<'created' | 'updated' | 'skipped'> {
|
|
const ytId = item.id;
|
|
if (!ytId) return 'skipped';
|
|
|
|
const snippet = item.snippet;
|
|
const status = item.status;
|
|
|
|
const title = snippet?.title ?? '(Untitled)';
|
|
const youtubeDescription = snippet?.description ?? undefined;
|
|
const tags = snippet?.tags ?? [];
|
|
const categoryId = snippet?.categoryId ?? undefined;
|
|
const publishedAt = snippet?.publishedAt ? new Date(snippet.publishedAt) : undefined;
|
|
const privacyStatus: PrivacyStatus =
|
|
YT_PRIVACY[status?.privacyStatus ?? ''] ?? PrivacyStatus.PRIVATE;
|
|
const defaultLanguage = snippet?.defaultLanguage ?? undefined;
|
|
const defaultAudioLanguage = snippet?.defaultAudioLanguage ?? undefined;
|
|
const selfDeclaredMadeForKids = (status as any)?.selfDeclaredMadeForKids ?? undefined;
|
|
const embeddable = (status as any)?.embeddable ?? undefined;
|
|
const license = (status as any)?.license ?? undefined;
|
|
const scheduledAt = (status as any)?.publishAt ? new Date((status as any).publishAt) : undefined;
|
|
const thumbnails = snippet?.thumbnails as any;
|
|
const thumbnailUrl =
|
|
thumbnails?.maxres?.url ??
|
|
thumbnails?.standard?.url ??
|
|
thumbnails?.high?.url ??
|
|
thumbnails?.medium?.url ??
|
|
thumbnails?.default?.url ??
|
|
undefined;
|
|
|
|
const ytFields = {
|
|
title, youtubeDescription, tags, categoryId, publishedAt, privacyStatus,
|
|
defaultLanguage, defaultAudioLanguage, selfDeclaredMadeForKids, embeddable, license,
|
|
scheduledAt, thumbnailUrl,
|
|
youtubeDeletedAt: null,
|
|
};
|
|
|
|
const existing = await this.prisma.video.findUnique({ where: { youtubeVideoId: ytId } });
|
|
|
|
// For fields YouTube doesn't reliably return, fall back to existing DB value so
|
|
// the import hash stays consistent with what computePendingChanges will compute.
|
|
// recordingDate is never returned by the API, so preserving it prevents false
|
|
// "push pending" after re-import.
|
|
const resolvedFields = {
|
|
defaultLanguage: defaultLanguage ?? existing?.defaultLanguage ?? null,
|
|
defaultAudioLanguage: defaultAudioLanguage ?? existing?.defaultAudioLanguage ?? null,
|
|
selfDeclaredMadeForKids: selfDeclaredMadeForKids ?? existing?.selfDeclaredMadeForKids ?? false,
|
|
embeddable: embeddable ?? existing?.embeddable ?? true,
|
|
license: license ?? existing?.license ?? 'youtube',
|
|
recordingDate: existing?.recordingDate ?? null,
|
|
};
|
|
|
|
const importHash = hashMetadata({
|
|
title,
|
|
description: youtubeDescription ?? '',
|
|
tags,
|
|
categoryId: categoryId ?? null,
|
|
privacyStatus: privacyStatus ? String(privacyStatus) : null,
|
|
...resolvedFields,
|
|
});
|
|
|
|
const youtubeSnapshot = {
|
|
title,
|
|
tags,
|
|
categoryId: categoryId ?? null,
|
|
privacyStatus: String(privacyStatus),
|
|
defaultLanguage: resolvedFields.defaultLanguage,
|
|
defaultAudioLanguage: resolvedFields.defaultAudioLanguage,
|
|
selfDeclaredMadeForKids: resolvedFields.selfDeclaredMadeForKids,
|
|
embeddable: resolvedFields.embeddable,
|
|
license: resolvedFields.license,
|
|
recordingDate: resolvedFields.recordingDate
|
|
? resolvedFields.recordingDate.toISOString().slice(0, 10)
|
|
: null,
|
|
};
|
|
|
|
if (!existing) {
|
|
await this.prisma.video.create({
|
|
data: { youtubeVideoId: ytId, channelId, ...ytFields, lastSyncedHash: importHash, youtubeSnapshot },
|
|
});
|
|
return 'created';
|
|
} else {
|
|
// Always update channelId in case the video was previously imported under a
|
|
// stale/wrong channel — this re-assigns ownership to the correct channel.
|
|
await this.prisma.video.update({
|
|
where: { youtubeVideoId: ytId },
|
|
data: { channelId, ...ytFields, lastSyncedHash: importHash, youtubeSnapshot },
|
|
});
|
|
return 'updated';
|
|
}
|
|
}
|
|
|
|
private async syncPlaylistMembership(
|
|
channelId: string,
|
|
ctx?: ActionCtx,
|
|
force = false,
|
|
): Promise<{ synced: number; skipped: number; allVideoIds: Set<string> }> {
|
|
const ytPlaylists = await this.ytApi.listChannelPlaylists(channelId, ctx);
|
|
|
|
const existingPlaylists = await this.prisma.playlist.findMany({
|
|
where: { channelId },
|
|
select: { id: true, youtubePlaylistId: true, itemCount: true },
|
|
});
|
|
const existingMap = new Map(existingPlaylists.map((pl) => [pl.youtubePlaylistId, pl]));
|
|
|
|
for (const pl of ytPlaylists) {
|
|
await this.prisma.playlist.upsert({
|
|
where: { youtubePlaylistId: pl.youtubePlaylistId },
|
|
create: { channelId, ...pl },
|
|
update: { title: pl.title, description: pl.description, itemCount: pl.itemCount, privacyStatus: pl.privacyStatus },
|
|
});
|
|
}
|
|
|
|
const dbPlaylists = await this.prisma.playlist.findMany({
|
|
where: { channelId },
|
|
select: { id: true, youtubePlaylistId: true },
|
|
});
|
|
const playlistIdMap = new Map(dbPlaylists.map((pl) => [pl.youtubePlaylistId, pl.id]));
|
|
|
|
let skipped = 0;
|
|
let synced = 0;
|
|
const allVideoIds = new Set<string>();
|
|
|
|
for (const pl of ytPlaylists) {
|
|
const dbPlaylistId = playlistIdMap.get(pl.youtubePlaylistId);
|
|
if (!dbPlaylistId) continue;
|
|
|
|
const existing = existingMap.get(pl.youtubePlaylistId);
|
|
if (!force && existing && existing.itemCount === pl.itemCount) {
|
|
skipped++;
|
|
continue;
|
|
}
|
|
|
|
const { ids: ytVideoIds } = await this.ytApi.listPlaylistVideoIds(channelId, pl.youtubePlaylistId, ctx);
|
|
|
|
for (const id of ytVideoIds) allVideoIds.add(id);
|
|
|
|
const matchingVideos = await this.prisma.video.findMany({
|
|
where: { youtubeVideoId: { in: ytVideoIds }, channelId },
|
|
select: { id: true },
|
|
});
|
|
|
|
await this.prisma.videoPlaylist.deleteMany({ where: { playlistId: dbPlaylistId } });
|
|
if (matchingVideos.length > 0) {
|
|
await this.prisma.videoPlaylist.createMany({
|
|
data: matchingVideos.map((v) => ({ videoId: v.id, playlistId: dbPlaylistId })),
|
|
skipDuplicates: true,
|
|
});
|
|
}
|
|
synced++;
|
|
}
|
|
|
|
this.logger.log(`Playlist membership synced: ${synced} updated, ${skipped} skipped (itemCount unchanged)`);
|
|
return { synced, skipped, allVideoIds };
|
|
}
|
|
|
|
async addSupplementalVideoIds(channelId: string, teamId: string, videoIds: string[]): Promise<string[]> {
|
|
const channel = await this.prisma.channel.findFirstOrThrow({ where: { id: channelId, teamId } });
|
|
const existing = new Set(channel.supplementalVideoIds ?? []);
|
|
for (const id of videoIds) existing.add(id);
|
|
const updated = [...existing];
|
|
await this.prisma.channel.update({ where: { id: channelId }, data: { supplementalVideoIds: updated } });
|
|
this.logger.log(`Channel ${channelId} supplemental IDs updated: ${updated.length} total`);
|
|
return updated;
|
|
}
|
|
}
|